spb/tools-spboucher-ai

1/** Durable Postgres-backed job queue: enqueue (deduped), claim with SKIP LOCKED, retry with exponential backoff, dead-letter after max attempts. */2import { sql } from "@/db";3export type JobKind = "link_check" | "link_sweep" | "source_rescan" | "embed_tool" | "embed_sweep" | "freshness_sweep" | "backup" | "quality_recompute" | "import_batch";4export type Job = { id: number; kind: JobKind; payload: Record<string, unknown>; attempts: number; maxAttempts: number };56export async function enqueue(kind: JobKind, payload: Record<string, unknown> = {}, opts: { dedupeKey?: string; runAt?: Date; priority?: number; maxAttempts?: number } = {}) {7  const dedupe = opts.dedupeKey ?? null;8  const rows = await sql<{ id: number }[]>`9    INSERT INTO jobs (kind, payload, dedupe_key, run_at, priority, max_attempts) VALUES (${kind}, ${sql.json(payload as any)}, ${dedupe}, ${opts.runAt ?? new Date()}, ${opts.priority ?? 5}, ${opts.maxAttempts ?? 5})10    ON CONFLICT (dedupe_key) DO UPDATE SET run_at = LEAST(jobs.run_at, EXCLUDED.run_at), updated_at = now() WHERE jobs.status IN ('queued') RETURNING id`;11  return rows[0]?.id ?? null;12}13export async function claim(worker: string, kinds?: JobKind[]): Promise<Job | null> {14  const rows = await sql<Job[]>`15    WITH c AS (SELECT id FROM jobs WHERE status = 'queued' AND run_at <= now() ${kinds?.length ? sql`AND kind = ANY(${kinds})` : sql``} ORDER BY priority, run_at LIMIT 1 FOR UPDATE SKIP LOCKED)16    UPDATE jobs j SET status = 'running', locked_at = now(), locked_by = ${worker}, started_at = now(), attempts = attempts + 1, updated_at = now() FROM c WHERE j.id = c.id17    RETURNING j.id, j.kind, j.payload, j.attempts, j.max_attempts AS "maxAttempts"`;18  return rows[0] ?? null;19}20export async function succeed(id: number, result: unknown = null) { await sql`UPDATE jobs SET status = 'succeeded', finished_at = now(), result = ${result === null ? null : sql.json(result as any)}, last_error = NULL, locked_at = NULL, locked_by = NULL, updated_at = now() WHERE id = ${id}`; }21export async function fail(job: Job, err: string) {22  const dead = job.attempts >= job.maxAttempts; const backoffSec = Math.min(6 * 3600, 30 * 2 ** job.attempts);23  await sql`UPDATE jobs SET status = ${dead ? "dead" : "queued"}, run_at = ${dead ? sql`run_at` : sql`now() + make_interval(secs => ${backoffSec})`}, finished_at = ${dead ? sql`now()` : null}, last_error = ${err.slice(0, 2000)}, locked_at = NULL, locked_by = NULL, updated_at = now() WHERE id = ${job.id}`;24}25/** Jobs stuck in `running` beyond the lease are requeued (worker crash). */26export async function reapStale(leaseMinutes = 20) { const r = await sql`UPDATE jobs SET status = 'queued', locked_at = NULL, locked_by = NULL, last_error = coalesce(last_error,'') || ' [lease expired]', updated_at = now() WHERE status = 'running' AND locked_at < now() - make_interval(mins => ${leaseMinutes}) RETURNING id`; return r.length; }27export async function stats() { const rows = await sql<{ status: string; kind: string; n: number }[]>`SELECT status, kind, count(*)::int AS n FROM jobs GROUP BY 1,2 ORDER BY 1,2`; return rows; }28export async function retry(id: number) { await sql`UPDATE jobs SET status='queued', run_at=now(), attempts=0, last_error=NULL, updated_at=now() WHERE id=${id} AND status IN ('dead','failed','cancelled')`; }29export async function cancel(id: number) { await sql`UPDATE jobs SET status='cancelled', updated_at=now() WHERE id=${id} AND status='queued'`; }30export async function prune(days = 30) { await sql`DELETE FROM jobs WHERE status IN ('succeeded','cancelled') AND finished_at < now() - make_interval(days => ${days})`; }31