From 9411f292a8df96ef9c70482d8aaad700457521e1 Mon Sep 17 00:00:00 2001 From: robbond Date: Thu, 5 Mar 2026 18:02:57 +0000 Subject: [PATCH] feat(worker): add health check runner --- lib/plesk/health.ts | 345 ++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 345 insertions(+) create mode 100644 lib/plesk/health.ts diff --git a/lib/plesk/health.ts b/lib/plesk/health.ts new file mode 100644 index 0000000..ae86cb8 --- /dev/null +++ b/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; + latency_ms: number; + error_message: string | null; +}; + +type RunSingleHealthCheckResult = { + checked: boolean; + status?: Exclude; + 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, +) { + 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( + promise: Promise, + timeoutMs: number, +): Promise { + let timeout: ReturnType | null = null; + + const timeoutPromise = new Promise((_, 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, +): Promise { + 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 { + 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 = { + 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 { + 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, + }; +}