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  healthProbe: 'health.probe',12} as const;13export type JobName = (typeof JOBS)[keyof typeof JOBS];1415export const PGBOSS_SCHEMA = 'pgboss';1617let boss: PgBoss | null = null;18let starting: Promise<PgBoss> | null = null;1920/** Lazy pg-boss client for enqueueing only (no supervision, no cron — the worker owns those). */21export async function getBoss(): Promise<PgBoss> {22  if (boss) return boss;23  if (!starting) {24    starting = (async () => {25      const b = new PgBoss({26        connectionString: process.env.DATABASE_URL ?? 'postgres://localhost:5432/cancerindex',27        schema: PGBOSS_SCHEMA,28        application_name: 'cancerindex-api',29        supervise: false,30        schedule: false,31        max: 2,32      });33      b.on('error', () => {34        /* logged by caller on send failure */35      });36      await b.start();37      boss = b;38      return b;39    })();40  }41  return starting;42}4344/**45 * Queue policy `stately`: at most one job per singletonKey in created/retry/active state, i.e. a46 * connector is never queued twice nor run concurrently (`send` returns null when deduplicated).47 * The worker owns queue creation; this only covers the case where the API enqueues first.48 */49export const QUEUE_POLICY = 'stately' as const;5051export async function ensureQueue(b: PgBoss, name: JobName): Promise<void> {52  const existing = await b.getQueue(name);53  if (existing) return;54  // pg-boss refuses `undefined` option values — pass only defined keys.55  await b.createQueue(name, { policy: QUEUE_POLICY, retryLimit: 0, expireInSeconds: 3 * 3600 });56}5758export async function enqueue(name: JobName, data: Record<string, unknown>, singletonKey: string): Promise<string | null> {59  const b = await getBoss();60  await ensureQueue(b, name);61  return b.send(name, data, { singletonKey });62}6364export async function closeBoss(): Promise<void> {65  if (boss) {66    await boss.stop({ graceful: false, close: true, timeout: 5000 });67    boss = null;68    starting = null;69  }70}71