STCourses/packages/core/src/jobs.ts

import type { JobHandler, JobQueue } from "@stc/contracts";
import { jobsRepo, type Db } from "@stc/db";

export function createJobQueue(db: Db, handlers: Record<string, JobHandler>, opts: { pollMs?: number } = {}): JobQueue & { start(): void; stop(): Promise<void> } {
  const repo = jobsRepo(db);
  const pollMs = opts.pollMs ?? 500;
  const kinds = Object.keys(handlers);
  let running = false;
  let loop: Promise<void> | null = null;
  let wake: (() => void) | null = null;
  let abort = new AbortController();

  const runOne = async (): Promise<boolean> => {
    const job = await repo.claimNext(kinds);
    if (!job) return false;
    const jc = {
      signal: abort.signal,
      reportProgress: (fraction: number, message?: string) =>
        repo.update(job.id, message === undefined ? { progress: fraction } : { progress: fraction, message }),
    };
    try {
      const result = await handlers[job.kind]!(job.payload, jc);
      await repo.update(job.id, { status: "done", progress: 1, result: result ?? null });
    } catch (e) {
      await repo.update(job.id, { status: "failed", error: e instanceof Error ? e.message : String(e) });
    }
    return true;
  };

  const sleep = () => new Promise<void>((res) => {
    const t = setTimeout(res, pollMs);
    wake = () => { clearTimeout(t); res(); };
  });

  const run = async () => {
    while (running) {
      let did = false;
      try { did = await runOne(); } catch (e) { console.error("[stc] job worker error", e); }
      if (!did && running) await sleep();
    }
  };

  return {
    async enqueue(kind, payload) {
      const id = await repo.insert(kind, payload);
      wake?.();
      return id;
    },
    get: (id) => repo.get(id),
    list: (kind) => repo.list(kind),
    start() {
      if (running) return;
      running = true;
      abort = new AbortController();
      loop = (async () => {
        // jobs left running by a previous process go back to the queue
        await db.sql`UPDATE jobs SET status = 'queued', updated_at = ${new Date().toISOString()} WHERE status = 'running'`;
        await run();
      })();
    },
    async stop() {
      running = false;
      abort.abort();
      wake?.();
      await loop;
      loop = null;
    },
  };
}