spb/sp-drive

1import { type Db, sql } from './db.ts';23export interface Job { id: string; kind: string; payload: Record<string, unknown>; attempts: number; max_attempts: number }4export async function enqueue(t: Db, kind: string, payload: Record<string, unknown>, o: { dedupeKey?: string; delayS?: number; maxAttempts?: number } = {}): Promise<void> {5  await t`insert into jobs (kind, payload, dedupe_key, run_at, max_attempts)6    values (${kind}, ${t.json(payload as never)}, ${o.dedupeKey ?? null}, now() + ${`${o.delayS ?? 0} seconds`}::interval, ${o.maxAttempts ?? 3})7    on conflict (dedupe_key) where status in ('queued', 'running') and dedupe_key is not null do nothing`;8}9async function claim(): Promise<Job | undefined> {10  const [j] = await sql<Job[]>`update jobs set status = 'running', locked_at = now(), attempts = attempts + 111    where id = (select id from jobs where status = 'queued' and run_at <= now() order by run_at, id for update skip locked limit 1)12    returning id::text, kind, payload, attempts, max_attempts`;13  return j;14}15export type Handler = (payload: Record<string, unknown>, job: Job) => Promise<void>;1617/** Run queued jobs until none is ready. Returns how many ran. */18export async function drain(handlers: Record<string, Handler>): Promise<number> {19  let n = 0;20  for (let j = await claim(); j; j = await claim()) {21    n++;22    try {23      const h = handlers[j.kind];24      if (!h) throw new Error(`no handler for job kind ${j.kind}`);25      await h(j.payload, j);26      await sql`update jobs set status = 'done', finished_at = now(), last_error = null where id = ${j.id}`;27    } catch (e) {28      const msg = String((e as Error)?.message ?? e).slice(0, 500);29      const retry = j.attempts < j.max_attempts;30      await sql`update jobs set status = ${retry ? 'queued' : 'failed'}, last_error = ${msg}, locked_at = null,31        run_at = now() + ${`${Math.min(600, 15 * 2 ** j.attempts)} seconds`}::interval, finished_at = ${retry ? null : new Date()} where id = ${j.id}`;32      console.error(JSON.stringify({ level: 'error', msg: 'job failed', job: j.id, kind: j.kind, attempt: j.attempts, error: msg }));33    }34  }35  return n;36}37/** Jobs whose worker died mid-run go back to the queue. */38export const requeueStuck = () => sql`update jobs set status = case when attempts >= max_attempts then 'failed' else 'queued' end, locked_at = null,39  last_error = coalesce(last_error, 'worker stopped while running this job') where status = 'running' and locked_at < now() - interval '20 minutes'`;4041export function startWorkerLoop(handlers: Record<string, Handler>, everyMs = 1500): () => Promise<void> {42  let stop = false, busy: Promise<unknown> = Promise.resolve();43  const tick = async () => { if (stop) return; busy = drain(handlers).catch((e) => console.error('worker loop', e)); await busy; if (!stop) setTimeout(tick, everyMs); };44  void tick();45  return async () => { stop = true; await busy; };46}47