feat(worker): add health check runner

lib/plesk/health.ts

@@ -0,0 +1,345 @@
+import { randomUUID } from "crypto";
+
+import { testPleskConnection } from "@/lib/plesk/client";
+
+type SupabaseLike = any;
+
+type HealthStatus = "ok" | "degraded" | "down" | "unknown";
+
+type HealthInstance = {
+  id: string;
+  agency_id: string;
+  base_url: string;
+  auth_type: "api_key" | "basic";
+  encrypted_secret: string;
+  health_enabled: boolean;
+  health_check_frequency_minutes: number | null;
+  health_next_check_at: string | null;
+  health_lock_until: string | null;
+};
+
+type HealthCheckResult = {
+  status: Exclude<HealthStatus, "unknown">;
+  latency_ms: number;
+  error_message: string | null;
+};
+
+type RunSingleHealthCheckResult = {
+  checked: boolean;
+  status?: Exclude<HealthStatus, "unknown">;
+  latency_ms?: number;
+  error_message?: string | null;
+};
+
+export type HealthTickSummary = {
+  checked: number;
+  ok: number;
+  degraded: number;
+  failed: number;
+  duration_ms: number;
+};
+
+const MAX_INSTANCES_PER_TICK = 10;
+const MAX_CONCURRENCY = 3;
+const LOCK_LEASE_SECONDS = 5 * 60;
+const HEALTH_TIMEOUT_MS = 15_000;
+const DEGRADED_LATENCY_THRESHOLD_MS = 2_500;
+
+function isDue(instance: HealthInstance, now: number) {
+  if (!instance.health_next_check_at) return true;
+  const dueAt = new Date(instance.health_next_check_at).getTime();
+  return Number.isFinite(dueAt) && dueAt <= now;
+}
+
+function isUnlocked(instance: HealthInstance, now: number) {
+  if (!instance.health_lock_until) return true;
+  const lockUntil = new Date(instance.health_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("[health] actions_log insert failed", error.message);
+    }
+  } catch (error) {
+    console.error("[health] 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("Health check timed out"));
+    }, timeoutMs);
+  });
+
+  try {
+    return await Promise.race([promise, timeoutPromise]);
+  } finally {
+    if (timeout) clearTimeout(timeout);
+  }
+}
+
+async function checkPleskInstanceHealth(
+  instance: Pick<HealthInstance, "base_url" | "auth_type" | "encrypted_secret">,
+): Promise<HealthCheckResult> {
+  const startedAt = Date.now();
+
+  try {
+    await withTimeout(
+      testPleskConnection({
+        baseUrl: instance.base_url,
+        authType: instance.auth_type,
+        encryptedSecret: instance.encrypted_secret,
+      }),
+      HEALTH_TIMEOUT_MS,
+    );
+
+    const latencyMs = Date.now() - startedAt;
+    const status: "ok" | "degraded" =
+      latencyMs > DEGRADED_LATENCY_THRESHOLD_MS ? "degraded" : "ok";
+
+    return {
+      status,
+      latency_ms: latencyMs,
+      error_message: null,
+    };
+  } catch (error) {
+    return {
+      status: "down",
+      latency_ms: Date.now() - startedAt,
+      error_message:
+        error instanceof Error
+          ? error.message.slice(0, 1000)
+          : "health_check_failed",
+    };
+  }
+}
+
+export async function runInstanceHealthCheck(
+  supabase: SupabaseLike,
+  instance: HealthInstance,
+  input?: {
+    actorUserId?: string | null;
+    force?: boolean;
+    source?: "cron" | "manual";
+  },
+): Promise<RunSingleHealthCheckResult> {
+  const lockToken = randomUUID();
+  const actorUserId = input?.actorUserId ?? null;
+  const source = input?.source ?? "cron";
+
+  const { data: lockAcquired, error: lockError } = await supabase.rpc(
+    "acquire_instance_health_lock",
+    {
+      p_instance_id: instance.id,
+      p_lock_token: lockToken,
+      p_ttl_seconds: LOCK_LEASE_SECONDS,
+      p_force: Boolean(input?.force),
+    },
+  );
+
+  if (lockError || !lockAcquired) {
+    return { checked: false };
+  }
+
+  const startedAt = Date.now();
+  const nowIso = new Date().toISOString();
+  const frequencyMinutes = Math.max(
+    instance.health_check_frequency_minutes ?? 30,
+    5,
+  );
+  const nextCheckAtIso = new Date(
+    Date.now() + frequencyMinutes * 60 * 1000,
+  ).toISOString();
+
+  await insertActionLogBestEffort(supabase, {
+    agency_id: instance.agency_id,
+    actor_user_id: actorUserId,
+    target_type: "plesk_instance",
+    target_id: instance.id,
+    action: "health_check_started",
+    status: "pending",
+    metadata: {
+      instance_id: instance.id,
+      lock_token: lockToken,
+      source,
+    },
+  });
+
+  try {
+    const result = await checkPleskInstanceHealth(instance);
+
+    const updates: Record<string, unknown> = {
+      health_status: result.status,
+      health_last_checked_at: nowIso,
+      health_last_latency_ms: result.latency_ms,
+      health_last_error: result.error_message,
+      health_next_check_at: nextCheckAtIso,
+      health_lock_until: null,
+      health_lock_token: null,
+    };
+
+    if (result.status === "ok") {
+      updates.health_last_ok_at = nowIso;
+    }
+
+    const { error: updateError } = await supabase
+      .from("plesk_instances")
+      .update(updates)
+      .eq("id", instance.id)
+      .eq("health_lock_token", lockToken);
+
+    if (updateError) {
+      throw new Error(updateError.message);
+    }
+
+    const action =
+      result.status === "ok"
+        ? "health_check_ok"
+        : result.status === "degraded"
+          ? "health_check_degraded"
+          : "health_check_failed";
+
+    await insertActionLogBestEffort(supabase, {
+      agency_id: instance.agency_id,
+      actor_user_id: actorUserId,
+      target_type: "plesk_instance",
+      target_id: instance.id,
+      action,
+      status: result.status === "down" ? "failed" : "success",
+      error_message: result.error_message,
+      metadata: {
+        instance_id: instance.id,
+        lock_token: lockToken,
+        source,
+        latency_ms: result.latency_ms,
+        duration_ms: Date.now() - startedAt,
+      },
+    });
+
+    return {
+      checked: true,
+      status: result.status,
+      latency_ms: result.latency_ms,
+      error_message: result.error_message,
+    };
+  } catch (error) {
+    const errorMessage =
+      error instanceof Error
+        ? error.message.slice(0, 1000)
+        : "health_check_failed";
+
+    await supabase
+      .from("plesk_instances")
+      .update({
+        health_status: "down",
+        health_last_checked_at: nowIso,
+        health_last_error: errorMessage,
+        health_next_check_at: nextCheckAtIso,
+      })
+      .eq("id", instance.id);
+
+    await insertActionLogBestEffort(supabase, {
+      agency_id: instance.agency_id,
+      actor_user_id: actorUserId,
+      target_type: "plesk_instance",
+      target_id: instance.id,
+      action: "health_check_failed",
+      status: "failed",
+      error_message: errorMessage,
+      metadata: {
+        instance_id: instance.id,
+        lock_token: lockToken,
+        source,
+        duration_ms: Date.now() - startedAt,
+      },
+    });
+
+    return {
+      checked: true,
+      status: "down",
+      latency_ms: Date.now() - startedAt,
+      error_message: errorMessage,
+    };
+  } finally {
+    await supabase.rpc("release_instance_health_lock", {
+      p_instance_id: instance.id,
+      p_lock_token: lockToken,
+    });
+  }
+}
+
+export async function runHealthCheckTick(
+  supabase: SupabaseLike,
+): Promise<HealthTickSummary> {
+  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, health_enabled, health_check_frequency_minutes, health_next_check_at, health_lock_until",
+    )
+    .eq("health_enabled", true)
+    .order("health_next_check_at", { ascending: true, nullsFirst: true })
+    .limit(50);
+
+  if (error) {
+    throw new Error(error.message);
+  }
+
+  const eligible = ((rows ?? []) as HealthInstance[])
+    .filter((instance) => isDue(instance, now) && isUnlocked(instance, now))
+    .slice(0, MAX_INSTANCES_PER_TICK);
+
+  let checked = 0;
+  let ok = 0;
+  let degraded = 0;
+  let failed = 0;
+
+  const queue = [...eligible];
+
+  const workers = Array.from(
+    { length: Math.min(MAX_CONCURRENCY, queue.length) },
+    async () => {
+      while (queue.length > 0) {
+        const instance = queue.shift();
+        if (!instance) break;
+
+        const result = await runInstanceHealthCheck(supabase, instance, {
+          actorUserId: null,
+          force: false,
+          source: "cron",
+        });
+
+        if (!result.checked || !result.status) continue;
+
+        checked += 1;
+        if (result.status === "ok") ok += 1;
+        else if (result.status === "degraded") degraded += 1;
+        else failed += 1;
+      }
+    },
+  );
+
+  await Promise.all(workers);
+
+  return {
+    checked,
+    ok,
+    degraded,
+    failed,
+    duration_ms: Date.now() - startedAt,
+  };
+}