feat(worker): add autosync scheduler, endpoint, locking and UI controls

lib/plesk/auto-sync.ts

@@ -0,0 +1,232 @@
+import { randomUUID } from "crypto";
+
+import { syncPleskInstance } from "@/lib/plesk/sync";
+
+type SupabaseLike = any;
+
+type AutoSyncInstance = {
+  id: string;
+  agency_id: string;
+  base_url: string;
+  auth_type: "api_key" | "basic";
+  encrypted_secret: string;
+  last_connected_at: string | null;
+  auto_sync_enabled: boolean;
+  auto_sync_frequency_minutes: number | null;
+  next_auto_sync_at: string | null;
+  auto_sync_lock_until: string | null;
+};
+
+type AutoSyncSummary = {
+  instances_considered: number;
+  locks_acquired: number;
+  succeeded: number;
+  failed: number;
+  duration_ms: number;
+};
+
+const MAX_INSTANCES_PER_TICK = 5;
+const MAX_CONCURRENCY = 2;
+const LOCK_LEASE_SECONDS = 15 * 60;
+const INSTANCE_TIMEOUT_MS = 14 * 60 * 1000;
+
+function isDue(instance: AutoSyncInstance, now: number) {
+  if (!instance.next_auto_sync_at) return true;
+  const dueAt = new Date(instance.next_auto_sync_at).getTime();
+  return Number.isFinite(dueAt) && dueAt <= now;
+}
+
+function isUnlocked(instance: AutoSyncInstance, now: number) {
+  if (!instance.auto_sync_lock_until) return true;
+  const lockUntil = new Date(instance.auto_sync_lock_until).getTime();
+  return !Number.isFinite(lockUntil) || lockUntil <= now;
+}
+
+async function insertActionLogBestEffort(
+  supabase: SupabaseLike,
+  payload: Record<string, unknown>,
+) {
+  try {
+    const { error } = await supabase.from("actions_log").insert(payload);
+    if (error) {
+      console.error("[autosync] actions_log insert failed", error.message);
+    }
+  } catch (error) {
+    console.error("[autosync] actions_log insert threw", error);
+  }
+}
+
+async function withTimeout<T>(
+  promise: Promise<T>,
+  timeoutMs: number,
+): Promise<T> {
+  let timeout: ReturnType<typeof setTimeout> | null = null;
+
+  const timeoutPromise = new Promise<never>((_, reject) => {
+    timeout = setTimeout(() => {
+      reject(new Error("Auto-sync timed out"));
+    }, timeoutMs);
+  });
+
+  try {
+    return await Promise.race([promise, timeoutPromise]);
+  } finally {
+    if (timeout) clearTimeout(timeout);
+  }
+}
+
+export async function runAutoSyncTick(
+  supabase: SupabaseLike,
+): Promise<AutoSyncSummary> {
+  const startedAt = Date.now();
+  const now = Date.now();
+
+  const { data: rows, error } = await supabase
+    .from("plesk_instances")
+    .select(
+      "id, agency_id, base_url, auth_type, encrypted_secret, last_connected_at, auto_sync_enabled, auto_sync_frequency_minutes, next_auto_sync_at, auto_sync_lock_until",
+    )
+    .eq("status", "connected")
+    .eq("auto_sync_enabled", true)
+    .order("next_auto_sync_at", { ascending: true, nullsFirst: true })
+    .limit(50);
+
+  if (error) {
+    throw new Error(error.message);
+  }
+
+  const eligible = ((rows ?? []) as AutoSyncInstance[])
+    .filter((instance) => isDue(instance, now) && isUnlocked(instance, now))
+    .slice(0, MAX_INSTANCES_PER_TICK);
+
+  let locksAcquired = 0;
+  let succeeded = 0;
+  let failed = 0;
+
+  const queue = [...eligible];
+
+  async function processInstance(instance: AutoSyncInstance) {
+    const lockToken = randomUUID();
+    const { data: lockAcquired, error: lockError } = await supabase.rpc(
+      "acquire_instance_auto_sync_lock",
+      {
+        p_instance_id: instance.id,
+        p_lock_token: lockToken,
+        p_ttl_seconds: LOCK_LEASE_SECONDS,
+      },
+    );
+
+    if (lockError || !lockAcquired) return;
+    locksAcquired += 1;
+
+    const perInstanceStartedAt = Date.now();
+    const frequencyMinutes = Math.max(
+      instance.auto_sync_frequency_minutes ?? 60,
+      5,
+    );
+
+    await insertActionLogBestEffort(supabase, {
+      agency_id: instance.agency_id,
+      actor_user_id: null,
+      target_type: "plesk_instance",
+      target_id: instance.id,
+      action: "autosync_started",
+      status: "pending",
+      metadata: {
+        instance_id: instance.id,
+        lock_token: lockToken,
+        frequency_minutes: frequencyMinutes,
+      },
+    });
+
+    try {
+      await withTimeout(
+        syncPleskInstance(supabase, instance, null, "autosync"),
+        INSTANCE_TIMEOUT_MS,
+      );
+
+      succeeded += 1;
+
+      await insertActionLogBestEffort(supabase, {
+        agency_id: instance.agency_id,
+        actor_user_id: null,
+        target_type: "plesk_instance",
+        target_id: instance.id,
+        action: "autosync_completed",
+        status: "success",
+        metadata: {
+          instance_id: instance.id,
+          lock_token: lockToken,
+          duration_ms: Date.now() - perInstanceStartedAt,
+        },
+      });
+    } catch (syncError) {
+      failed += 1;
+
+      const errorMessage =
+        syncError instanceof Error
+          ? syncError.message.slice(0, 500)
+          : "autosync_failed";
+
+      await insertActionLogBestEffort(supabase, {
+        agency_id: instance.agency_id,
+        actor_user_id: null,
+        target_type: "plesk_instance",
+        target_id: instance.id,
+        action: "autosync_failed",
+        status: "failed",
+        error_message: errorMessage,
+        metadata: {
+          instance_id: instance.id,
+          lock_token: lockToken,
+          duration_ms: Date.now() - perInstanceStartedAt,
+          error_summary: errorMessage,
+        },
+      });
+    } finally {
+      const nowIso = new Date().toISOString();
+      const nextAutoSyncAtIso = new Date(
+        Date.now() + frequencyMinutes * 60 * 1000,
+      ).toISOString();
+
+      const { error: scheduleError } = await supabase
+        .from("plesk_instances")
+        .update({
+          last_auto_sync_at: nowIso,
+          next_auto_sync_at: nextAutoSyncAtIso,
+          auto_sync_lock_until: null,
+          auto_sync_lock_token: null,
+        })
+        .eq("id", instance.id)
+        .eq("auto_sync_lock_token", lockToken);
+
+      if (scheduleError) {
+        console.error(
+          "[autosync] schedule update failed",
+          scheduleError.message,
+        );
+      }
+    }
+  }
+
+  const workers = Array.from(
+    { length: Math.min(MAX_CONCURRENCY, queue.length) },
+    async () => {
+      while (queue.length > 0) {
+        const instance = queue.shift();
+        if (!instance) break;
+        await processInstance(instance);
+      }
+    },
+  );
+
+  await Promise.all(workers);
+
+  return {
+    instances_considered: eligible.length,
+    locks_acquired: locksAcquired,
+    succeeded,
+    failed,
+    duration_ms: Date.now() - startedAt,
+  };
+}