spb/wp-spboucher-ai

1import { sql } from "@/db";2export type JobKind = "process_pdf" | "embed_paper" | "relations" | "daily_stats" | "backup" | "ingest_scan" | "ingest_candidate";3export async function enqueue(kind: JobKind, payload: Record<string, unknown> = {}, opts: { runAfter?: Date; dedupe?: boolean } = {}) {4  if (opts.dedupe) {5    const [dup] = await sql`SELECT id FROM jobs WHERE kind = ${kind} AND payload = ${sql.json(payload as never)} AND status IN ('queued', 'running') LIMIT 1`;6    if (dup) return Number(dup.id);7  }8  const [r] = await sql`INSERT INTO jobs (kind, payload, run_after) VALUES (${kind}, ${sql.json(payload as never)}, ${opts.runAfter ?? new Date()}) RETURNING id`;9  return Number(r.id);10}11export async function claimJob(): Promise<{ id: number; kind: JobKind; payload: Record<string, unknown>; attempts: number } | null> {12  const rows = await sql`UPDATE jobs SET status = 'running', locked_at = now(), attempts = attempts + 113    WHERE id = (SELECT id FROM jobs WHERE (status = 'queued' AND run_after <= now()) OR (status = 'running' AND locked_at < now() - interval '30 minutes') ORDER BY run_after LIMIT 1 FOR UPDATE SKIP LOCKED)14    RETURNING id, kind, payload, attempts`;15  if (!rows.length) return null; const r = rows[0];16  return { id: Number(r.id), kind: r.kind, payload: r.payload, attempts: r.attempts };17}18export async function finishJob(id: number, ok: boolean, error?: string, retryable = true) {19  if (ok) { await sql`UPDATE jobs SET status = 'done', finished_at = now(), last_error = NULL WHERE id = ${id}`; return; }20  const [j] = await sql`SELECT attempts, max_attempts FROM jobs WHERE id = ${id}`;21  if (retryable && j && j.attempts < j.max_attempts) await sql`UPDATE jobs SET status = 'queued', run_after = now() + (${j.attempts} * interval '2 minutes'), last_error = ${error ?? null} WHERE id = ${id}`;22  else await sql`UPDATE jobs SET status = 'failed', finished_at = now(), last_error = ${error ?? null} WHERE id = ${id}`;23}24