spb/cancerindex

1import { PgBoss } from 'pg-boss';23/**4 * Job names shared with workers/main.ts. Queues use the `singleton` policy so a connector (or the5 * counters/rank maintenance) never runs twice concurrently — the `singletonKey` is the connector id.6 */7export const JOBS = {8  connectorRun: 'connector.run',9  counters: 'maintenance.counters',10  rank: 'maintenance.rank',11  intel: 'maintenance.intel',12  healthProbe: 'health.probe',13} as const;14export type JobName = (typeof JOBS)[keyof typeof JOBS];1516export const PGBOSS_SCHEMA = 'pgboss';1718let boss: PgBoss | null = null;19let starting: Promise<PgBoss> | null = null;2021/** Lazy pg-boss client for enqueueing only (no supervision, no cron — the worker owns those). */22export async function getBoss(): Promise<PgBoss> {23  if (boss) return boss;24  if (!starting) {25    starting = (async () => {26      const b = new PgBoss({27        connectionString: process.env.DATABASE_URL ?? 'postgres://localhost:5432/cancerindex',28        schema: PGBOSS_SCHEMA,29        application_name: 'cancerindex-api',30        supervise: false,31        schedule: false,32        max: 2,33      });34      b.on('error', () => {35        /* logged by caller on send failure */36      });37      await b.start();38      boss = b;39      return b;40    })();41  }42  return starting;43}4445/**46 * Queue policy `stately`: at most one job per singletonKey in created/retry/active state, i.e. a47 * connector is never queued twice nor run concurrently (`send` returns null when deduplicated).48 * The worker owns queue creation; this only covers the case where the API enqueues first.49 */50export const QUEUE_POLICY = 'stately' as const;5152export async function ensureQueue(b: PgBoss, name: JobName): Promise<void> {53  const existing = await b.getQueue(name);54  if (existing) return;55  // pg-boss refuses `undefined` option values — pass only defined keys.56  await b.createQueue(name, { policy: QUEUE_POLICY, retryLimit: 0, expireInSeconds: 3 * 3600 });57}5859export async function enqueue(name: JobName, data: Record<string, unknown>, singletonKey: string): Promise<string | null> {60  const b = await getBoss();61  await ensureQueue(b, name);62  return b.send(name, data, { singletonKey });63}6465export async function closeBoss(): Promise<void> {66  if (boss) {67    await boss.stop({ graceful: false, close: true, timeout: 5000 });68    boss = null;69    starting = null;70  }71}72