Spaces:
Sleeping
Sleeping
| import { createServerSupabase } from "./supabase"; | |
| export const FRESH_HEARTBEAT_SECONDS = 90; | |
| export const SWEEP_HEARTBEAT_SECONDS = 120; | |
| export type TabularRunStatus = | |
| | "queued" | |
| | "running" | |
| | "completed" | |
| | "failed" | |
| | "canceled"; | |
| export type TabularRunRecord = { | |
| id: string; | |
| review_id: string; | |
| status: TabularRunStatus; | |
| document_ids: string[]; | |
| total_cells: number; | |
| completed_cells: number; | |
| failed_cells: number; | |
| current_document_id: string | null; | |
| error: string | null; | |
| langsmith_run_id: string | null; | |
| created_by: string; | |
| started_at: string; | |
| heartbeat_at: string; | |
| finished_at: string | null; | |
| }; | |
| export const runAbortControllers = new Map<string, AbortController>(); | |
| const lastRunDbUpdate = new Map<string, number>(); | |
| export function nowIso(): string { | |
| return new Date().toISOString(); | |
| } | |
| export function cutoffIso(secondsAgo: number): string { | |
| return new Date(Date.now() - secondsAgo * 1000).toISOString(); | |
| } | |
| export function isTerminalRunStatus(status: string): boolean { | |
| return status === "completed" || status === "failed" || status === "canceled"; | |
| } | |
| export function isStaleHeartbeat( | |
| heartbeatAt: string, | |
| seconds = FRESH_HEARTBEAT_SECONDS, | |
| ): boolean { | |
| return new Date(heartbeatAt).getTime() < Date.now() - seconds * 1000; | |
| } | |
| type Db = ReturnType<typeof createServerSupabase>; | |
| export async function sweepStaleRuns( | |
| db: Db, | |
| reviewId?: string, | |
| ): Promise<string[]> { | |
| const sweepCutoff = cutoffIso(SWEEP_HEARTBEAT_SECONDS); | |
| let query = db | |
| .from("tabular_review_runs") | |
| .select("id, review_id") | |
| .eq("status", "running") | |
| .lt("heartbeat_at", sweepCutoff); | |
| if (reviewId) query = query.eq("review_id", reviewId); | |
| const { data: staleRuns } = await query; | |
| const runIds = (staleRuns ?? []).map((r) => r.id as string); | |
| if (runIds.length === 0) return []; | |
| const finishedAt = nowIso(); | |
| for (const runId of runIds) { | |
| await db | |
| .from("tabular_review_runs") | |
| .update({ | |
| status: "failed", | |
| error: "stale_heartbeat", | |
| finished_at: finishedAt, | |
| }) | |
| .eq("id", runId); | |
| runAbortControllers.get(runId)?.abort(); | |
| runAbortControllers.delete(runId); | |
| } | |
| for (const runId of runIds) { | |
| await db | |
| .from("tabular_cells") | |
| .update({ | |
| status: "pending", | |
| content: null, | |
| error_reason: null, | |
| updated_at: finishedAt, | |
| }) | |
| .eq("run_id", runId) | |
| .eq("status", "generating"); | |
| } | |
| return runIds; | |
| } | |
| export async function markStaleRunningRunsFailed( | |
| db: Db, | |
| reviewId: string, | |
| ): Promise<void> { | |
| const freshCutoff = cutoffIso(FRESH_HEARTBEAT_SECONDS); | |
| const { data: staleRuns } = await db | |
| .from("tabular_review_runs") | |
| .select("id") | |
| .eq("review_id", reviewId) | |
| .eq("status", "running") | |
| .lt("heartbeat_at", freshCutoff); | |
| const runIds = (staleRuns ?? []).map((r) => r.id as string); | |
| if (runIds.length === 0) return; | |
| const finishedAt = nowIso(); | |
| for (const runId of runIds) { | |
| await db | |
| .from("tabular_review_runs") | |
| .update({ | |
| status: "failed", | |
| error: "stale_heartbeat", | |
| finished_at: finishedAt, | |
| }) | |
| .eq("id", runId); | |
| runAbortControllers.get(runId)?.abort(); | |
| runAbortControllers.delete(runId); | |
| } | |
| } | |
| export async function resetOrphanedGeneratingCells( | |
| db: Db, | |
| reviewId: string, | |
| ): Promise<void> { | |
| const staleCutoff = cutoffIso(FRESH_HEARTBEAT_SECONDS); | |
| const ts = nowIso(); | |
| const { data: terminalRuns } = await db | |
| .from("tabular_review_runs") | |
| .select("id") | |
| .eq("review_id", reviewId) | |
| .in("status", ["completed", "failed", "canceled"]); | |
| const terminalRunIds = (terminalRuns ?? []).map((r) => r.id as string); | |
| const { data: staleRunningRuns } = await db | |
| .from("tabular_review_runs") | |
| .select("id") | |
| .eq("review_id", reviewId) | |
| .eq("status", "running") | |
| .lt("heartbeat_at", staleCutoff); | |
| const staleRunIds = (staleRunningRuns ?? []).map((r) => r.id as string); | |
| const resetRunIds = [...new Set([...terminalRunIds, ...staleRunIds])]; | |
| if (resetRunIds.length > 0) { | |
| await db | |
| .from("tabular_cells") | |
| .update({ | |
| status: "pending", | |
| content: null, | |
| error_reason: null, | |
| updated_at: ts, | |
| }) | |
| .eq("review_id", reviewId) | |
| .eq("status", "generating") | |
| .in("run_id", resetRunIds); | |
| } | |
| const { data: legacyStale } = await db | |
| .from("tabular_cells") | |
| .select("id") | |
| .eq("review_id", reviewId) | |
| .eq("status", "generating") | |
| .is("run_id", null) | |
| .lt("updated_at", staleCutoff); | |
| for (const cell of legacyStale ?? []) { | |
| await db | |
| .from("tabular_cells") | |
| .update({ | |
| status: "pending", | |
| content: null, | |
| error_reason: null, | |
| updated_at: ts, | |
| }) | |
| .eq("id", cell.id); | |
| } | |
| } | |
| export async function findFreshActiveRun( | |
| db: Db, | |
| reviewId: string, | |
| ): Promise<TabularRunRecord | null> { | |
| const freshCutoff = cutoffIso(FRESH_HEARTBEAT_SECONDS); | |
| const { data: runs } = await db | |
| .from("tabular_review_runs") | |
| .select("*") | |
| .eq("review_id", reviewId) | |
| .eq("status", "running") | |
| .gte("heartbeat_at", freshCutoff) | |
| .order("started_at", { ascending: false }) | |
| .limit(1); | |
| const run = runs?.[0]; | |
| if (!run) return null; | |
| return normalizeRunRecord(run); | |
| } | |
| export async function findLatestNonTerminalRun( | |
| db: Db, | |
| reviewId: string, | |
| ): Promise<(TabularRunRecord & { is_stale: boolean }) | null> { | |
| const { data: runs } = await db | |
| .from("tabular_review_runs") | |
| .select("*") | |
| .eq("review_id", reviewId) | |
| .in("status", ["queued", "running"]) | |
| .order("started_at", { ascending: false }) | |
| .limit(1); | |
| const run = runs?.[0]; | |
| if (!run) return null; | |
| const normalized = normalizeRunRecord(run); | |
| return { | |
| ...normalized, | |
| is_stale: isStaleHeartbeat(normalized.heartbeat_at), | |
| }; | |
| } | |
| export function normalizeRunRecord( | |
| row: Record<string, unknown>, | |
| ): TabularRunRecord { | |
| const docIds = row.document_ids; | |
| return { | |
| id: String(row.id), | |
| review_id: String(row.review_id), | |
| status: row.status as TabularRunStatus, | |
| document_ids: Array.isArray(docIds) | |
| ? docIds.filter((id): id is string => typeof id === "string") | |
| : [], | |
| total_cells: Number(row.total_cells ?? 0), | |
| completed_cells: Number(row.completed_cells ?? 0), | |
| failed_cells: Number(row.failed_cells ?? 0), | |
| current_document_id: | |
| row.current_document_id != null | |
| ? String(row.current_document_id) | |
| : null, | |
| error: row.error != null ? String(row.error) : null, | |
| langsmith_run_id: | |
| row.langsmith_run_id != null | |
| ? String(row.langsmith_run_id) | |
| : null, | |
| created_by: String(row.created_by), | |
| started_at: String(row.started_at), | |
| heartbeat_at: String(row.heartbeat_at), | |
| finished_at: | |
| row.finished_at != null ? String(row.finished_at) : null, | |
| }; | |
| } | |
| export async function updateRunRow( | |
| db: Db, | |
| runId: string, | |
| patch: Record<string, unknown>, | |
| opts?: { force?: boolean }, | |
| ): Promise<void> { | |
| const now = Date.now(); | |
| const last = lastRunDbUpdate.get(runId) ?? 0; | |
| const hasImportantField = | |
| "current_document_id" in patch || | |
| "status" in patch || | |
| "completed_cells" in patch || | |
| "failed_cells" in patch; | |
| if (!opts?.force && !hasImportantField && now - last < 2000) return; | |
| lastRunDbUpdate.set(runId, now); | |
| await db | |
| .from("tabular_review_runs") | |
| .update({ ...patch, heartbeat_at: nowIso() }) | |
| .eq("id", runId); | |
| } | |
| export async function getRunCellRollup( | |
| db: Db, | |
| runId: string, | |
| ): Promise<Record<string, number>> { | |
| const { data: cells } = await db | |
| .from("tabular_cells") | |
| .select("status") | |
| .eq("run_id", runId); | |
| const rollup: Record<string, number> = {}; | |
| for (const cell of cells ?? []) { | |
| const status = String(cell.status ?? "unknown"); | |
| rollup[status] = (rollup[status] ?? 0) + 1; | |
| } | |
| return rollup; | |
| } | |
| export function registerRunAbortController(runId: string): AbortController { | |
| const existing = runAbortControllers.get(runId); | |
| if (existing) existing.abort(); | |
| const controller = new AbortController(); | |
| runAbortControllers.set(runId, controller); | |
| return controller; | |
| } | |
| export function clearRunAbortController(runId: string): void { | |
| runAbortControllers.delete(runId); | |
| } | |