spb/wp-spboucher-ai

1import { config } from "dotenv";2import { existsSync } from "node:fs";3import { join } from "node:path";4for (const f of [".env.local", ".env"]) if (existsSync(join(process.cwd(), f))) config({ path: join(process.cwd(), f), quiet: true } as never);5const { sql } = await import("@/db");6const { claimJob, finishJob, enqueue } = await import("@/lib/jobs/queue");7const { handlers } = await import("@/lib/jobs/handlers");89const POLL = Number(process.env.WORKER_POLL_MS ?? 5000);10const log = (level: string, msg: string, extra: Record<string, unknown> = {}) => console.log(JSON.stringify({ t: new Date().toISOString(), level, component: "worker", msg, ...extra }));11let stopping = false;12process.on("SIGTERM", () => { stopping = true; log("info", "SIGTERM received, finishing current job"); });13process.on("SIGINT", () => { stopping = true; });1415async function schedule() {16  // Local time of the node (America/Toronto on the cluster). Runs once per day each.17  const now = new Date(); const hhmm = now.getHours() * 100 + now.getMinutes(); const today = now.toISOString().slice(0, 10);18  const mark = async (key: string) => { await sql`INSERT INTO settings (key, value) VALUES (${key}, ${sql.json({ day: today } as never)}) ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, updated_at = now()`; };19  const done = async (key: string) => { const [r] = await sql`SELECT value FROM settings WHERE key = ${key}`; return r?.value?.day === today; };20  if (hhmm >= 10 && !(await done("private.worker.daily_stats"))) { await enqueue("daily_stats", {}, { dedupe: true }); await mark("private.worker.daily_stats"); }21  if (hhmm >= 440 && !(await done("private.worker.backup"))) { await enqueue("backup", {}, { dedupe: true }); await mark("private.worker.backup"); }22  if (hhmm >= 300 && !(await done("private.worker.relations"))) { await enqueue("relations", {}, { dedupe: true }); await mark("private.worker.relations"); }23}2425log("info", "worker started", { poll_ms: POLL });26while (!stopping) {27  try {28    await schedule();29    const job = await claimJob();30    if (!job) { await new Promise((r) => setTimeout(r, POLL)); continue; }31    const t0 = Date.now(); log("info", "job start", { id: job.id, kind: job.kind, attempt: job.attempts });32    try { const result = await handlers[job.kind]?.(job.payload); await finishJob(job.id, true); log("info", "job done", { id: job.id, kind: job.kind, ms: Date.now() - t0, result }); }33    catch (e) { const msg = (e as Error).message ?? String(e); await finishJob(job.id, false, msg); log("error", "job failed", { id: job.id, kind: job.kind, error: msg.slice(0, 500) }); }34  } catch (e) { log("error", "loop error", { error: (e as Error).message }); await new Promise((r) => setTimeout(r, POLL)); }35}36await sql.end(); log("info", "worker stopped");37