spb/cancerindex

1/**2 * Enqueue worker jobs by hand (the worker must be running to execute them).3 *4 *   tsx workers/cli.ts run <connector> [--mode full|incremental|dry_run] [--max-minutes N] [--max-records N] [--reset-cursor]5 *   tsx workers/cli.ts counters [--then-rank]6 *   tsx workers/cli.ts rank7 *   tsx workers/cli.ts health8 *   tsx workers/cli.ts schedules            list registered schedules9 *   tsx workers/cli.ts queues               queue depths10 */11import { DATABASE_URL, JOBS, PGBOSS_SCHEMA } from './lib/env.js';12import { PgBoss } from 'pg-boss';1314const [, , cmd, ...rest] = process.argv;15const flags: Record<string, string | boolean> = {};16const positional: string[] = [];17for (let i = 0; i < rest.length; i++) {18  const a = rest[i]!;19  if (a.startsWith('--')) {20    const next = rest[i + 1];21    if (next && !next.startsWith('--')) {22      flags[a.slice(2)] = next;23      i++;24    } else flags[a.slice(2)] = true;25  } else positional.push(a);26}2728const boss = new PgBoss({ connectionString: DATABASE_URL, schema: PGBOSS_SCHEMA, application_name: 'cancerindex-worker-cli', supervise: false, schedule: false, max: 2 });29boss.on('error', (e) => console.error(e));30await boss.start();3132async function ensure(name: string) {33  if (!(await boss.getQueue(name))) await boss.createQueue(name, { policy: 'stately', retryLimit: 0, expireInSeconds: 3 * 3600 });34}3536try {37  switch (cmd) {38    case 'run': {39      const id = positional[0];40      if (!id) throw new Error('usage: run <connector>');41      // Omit undefined keys — pg-boss rejects undefined option values.42      const data: Record<string, unknown> = { id, requestedBy: 'cli' };43      if (flags.mode) data.mode = flags.mode;44      if (flags['max-minutes']) data.maxMinutes = Number(flags['max-minutes']);45      if (flags['max-records']) data.maxRecords = Number(flags['max-records']);46      if (flags['reset-cursor']) data.resetCursor = true;47      await ensure(JOBS.connectorRun);48      const jobId = await boss.send(JOBS.connectorRun, data, { singletonKey: id });49      console.log(jobId ? `queued ${JOBS.connectorRun} ${id} → job ${jobId}` : `not queued: a run for ${id} is already queued/active (singleton)`);50      break;51    }52    case 'counters': {53      await ensure(JOBS.counters);54      const jobId = await boss.send(JOBS.counters, { thenRank: !!flags['then-rank'], requestedBy: 'cli' }, { singletonKey: 'counters' });55      console.log(jobId ? `queued ${JOBS.counters} → ${jobId}` : 'not queued (already pending)');56      break;57    }58    case 'rank': {59      await ensure(JOBS.rank);60      const jobId = await boss.send(JOBS.rank, { requestedBy: 'cli' }, { singletonKey: 'rank' });61      console.log(jobId ? `queued ${JOBS.rank} → ${jobId}` : 'not queued (already pending)');62      break;63    }64    case 'health': {65      await ensure(JOBS.healthProbe);66      const jobId = await boss.send(JOBS.healthProbe, { requestedBy: 'cli' }, { singletonKey: 'health' });67      console.log(jobId ? `queued ${JOBS.healthProbe} → ${jobId}` : 'not queued (already pending)');68      break;69    }70    case 'schedules': {71      for (const s of await boss.getSchedules()) console.log(`${s.name.padEnd(22)} key=${s.key.padEnd(16)} cron=${s.cron.padEnd(14)} tz=${s.timezone} data=${JSON.stringify(s.data ?? {})}`);72      break;73    }74    case 'queues': {75      for (const q of await boss.getQueues()) console.log(`${q.name.padEnd(22)} policy=${q.policy} queued=${q.queuedCount} active=${q.activeCount} failed=${q.failedCount} total=${q.totalCount}`);76      break;77    }78    default:79      console.log('usage: tsx workers/cli.ts <run <connector>|counters|rank|health|schedules|queues>');80  }81} finally {82  await boss.stop({ graceful: false, close: true, timeout: 5000 });83}84