spb/tools-spboucher-ai

1/** Background worker: claims jobs, runs handlers, schedules recurring sweeps. Single process, PM2-supervised. */2import "./env";3import { claim, succeed, fail, reapStale, enqueue, prune } from "@/lib/jobs/queue";4import { runJob } from "@/lib/jobs/handlers";5import { closeDb } from "@/db";6import { hostname } from "node:os";78const WORKER = `${hostname()}:${process.pid}`; const POLL = Number(process.env.WORKER_POLL_MS ?? 5000);9let stopping = false;10process.on("SIGTERM", () => { stopping = true; }); process.on("SIGINT", () => { stopping = true; });1112/** Recurring schedule (America/Toronto wall clock): link sweep daily 03:10, freshness 03:40, embed sweep 04:00, backup 04:20, prune weekly. */13function schedule() {14  const now = new Date(); const tz = new Intl.DateTimeFormat("en-CA", { timeZone: "America/Toronto", hour: "2-digit", minute: "2-digit", hour12: false, weekday: "short" }).formatToParts(now);15  const hh = tz.find((p) => p.type === "hour")!.value, mm = tz.find((p) => p.type === "minute")!.value, wd = tz.find((p) => p.type === "weekday")!.value;16  const day = now.toISOString().slice(0, 10); const t = `${hh}:${mm}`;17  const at = (hm: string, kind: Parameters<typeof enqueue>[0], extra = "") => { if (t === hm) enqueue(kind, {}, { dedupeKey: `${kind}:${day}${extra}`, priority: 3 }).catch(() => {}); };18  at("03:10", "link_sweep"); at("03:40", "freshness_sweep"); at("04:00", "embed_sweep"); at("04:20", "backup");19  if (wd === "Sun" && t === "05:00") prune().catch(() => {});20}2122async function loop() {23  console.log(`[worker] ${WORKER} started, poll ${POLL} ms`);24  let tick = 0;25  while (!stopping) {26    try {27      if (tick % 12 === 0) { const n = await reapStale(); if (n) console.log(`[worker] requeued ${n} stale job(s)`); schedule(); }28      const job = await claim(WORKER);29      if (!job) { await new Promise((r) => setTimeout(r, POLL)); tick++; continue; }30      const t0 = Date.now();31      try { const res = await runJob(job); await succeed(job.id, res ?? null); console.log(`[worker] #${job.id} ${job.kind} ok ${Date.now() - t0} ms`); }32      catch (e) { const msg = (e as Error).stack ?? String(e); await fail(job, msg); console.error(`[worker] #${job.id} ${job.kind} failed (attempt ${job.attempts}/${job.maxAttempts}): ${(e as Error).message}`); }33    } catch (e) { console.error("[worker] loop error", e); await new Promise((r) => setTimeout(r, POLL)); }34  }35  await closeDb(); console.log("[worker] stopped");36}37loop();38