spb/note-spboucher-ai

1/** Background worker (PM2 "note-worker"): pg-boss consumers for indexing, imports, backups, agents, link suggestions. */2import "dotenv/config";3import { PgBoss } from "pg-boss";4import { QUEUES } from "../lib/jobs/queues";5import { handleIndex, handleReindexAll } from "./handlers/index-source";6import { handleImport } from "./handlers/import";7import { handleBackup } from "./handlers/backup";8import { handleAgentRun } from "./handlers/agent";9import { handleSuggestLinks } from "./handlers/suggest-links";10import { handleAiProperty } from "./handlers/ai-property";1112const boss = new PgBoss({ connectionString: process.env.DATABASE_URL!, schema: "pgboss", max: 4 });13boss.on("error", (e: Error) => console.error("[worker] pg-boss error", e));14await boss.start();15for (const q of Object.values(QUEUES)) await boss.createQueue(q).catch(() => {});16const log = (q: string, id: string, msg: string) => console.log(`[worker] ${new Date().toISOString()} ${q} ${id.slice(0, 8)} ${msg}`);17const wrap = (q: string, fn: (data: any) => Promise<unknown>) => async (jobs: { id: string; data: any }[]) => { for (const j of jobs) { const t0 = Date.now(); try { const r = await fn(j.data); log(q, j.id, `ok ${Date.now() - t0}ms ${r ? JSON.stringify(r).slice(0, 120) : ""}`); } catch (e: any) { log(q, j.id, `FAIL ${e?.message}`); throw e; } } };18await boss.work(QUEUES.index, { batchSize: 1, pollingIntervalSeconds: 2 }, wrap(QUEUES.index, handleIndex));19await boss.work(QUEUES.reindexAll, { batchSize: 1 }, wrap(QUEUES.reindexAll, handleReindexAll));20await boss.work(QUEUES.importFile, { batchSize: 1 }, wrap(QUEUES.importFile, handleImport));21await boss.work(QUEUES.backup, { batchSize: 1 }, wrap(QUEUES.backup, handleBackup));22await boss.work(QUEUES.agent, { batchSize: 1 }, wrap(QUEUES.agent, handleAgentRun));23await boss.work(QUEUES.suggestLinks, { batchSize: 1 }, wrap(QUEUES.suggestLinks, handleSuggestLinks));24await boss.work(QUEUES.aiProperty, { batchSize: 1 }, wrap(QUEUES.aiProperty, handleAiProperty));25// nightly encrypted backup at 03:30 (node local time)26await boss.schedule(QUEUES.backup, "30 3 * * *", { kind: "full" }, { tz: "America/Toronto" });27console.log(`[worker] started, queues: ${Object.values(QUEUES).join(", ")}`);28process.on("SIGINT", async () => { await boss.stop({ graceful: true, timeout: 10_000 }); process.exit(0); });29process.on("SIGTERM", async () => { await boss.stop({ graceful: true, timeout: 10_000 }); process.exit(0); });30