spb/cancerindex

1/**2 * CancerIndex worker (CLAUDE.md §90, §290): pg-boss scheduler + job handlers.3 *4 *   connector.run        {id, mode?, maxMinutes?, maxRecords?, resetCursor?}  singleton per connector5 *   maintenance.counters {thenRank?}                                          refreshCounters6 *   maintenance.rank     {}                                                   computeAllRankings7 *   health.probe         {}                                                   healthCheck per connector → connector_cursors.health8 *9 * Schedules are (re)registered from connector manifests on every start; stale schedules are removed.10 * pg-boss gotcha: it refuses `undefined` option values — always omit keys instead of passing undefined.11 */12import { CI_MAX_RUN_MINUTES, DATABASE_URL, JOBS, PGBOSS_SCHEMA, SCHEDULES, WORKER_CONCURRENCY, type JobName } from './lib/env.js';13import { PgBoss, type Job } from 'pg-boss';14import { sql } from 'drizzle-orm';15import { logger } from '@cancerindex/shared';16import { getDb, closeDb } from '@cancerindex/database';17import { CONNECTORS, getConnector, runConnector, RunContext, type RunMode } from '@cancerindex/connectors';18import { refreshCounters, computeAllRankings } from '@cancerindex/ranking';1920const log = logger.child({ component: 'worker' });21const db = getDb();2223interface ConnectorRunData {24  id: string;25  mode?: RunMode;26  maxMinutes?: number;27  maxRecords?: number;28  resetCursor?: boolean;29  requestedBy?: string;30}3132const boss = new PgBoss({33  connectionString: DATABASE_URL,34  schema: PGBOSS_SCHEMA,35  application_name: 'cancerindex-worker',36  max: Math.max(4, WORKER_CONCURRENCY + 2),37  // supervise + schedule are on: this process owns maintenance and cron.38});39boss.on('error', (err) => log.error({ err }, 'pg-boss error'));40boss.on('warning', (w) => log.warn({ w }, 'pg-boss warning'));4142/**43 * Queue policy `stately`: one job per singletonKey in created/retry/active — a connector is never44 * queued twice nor run concurrently. Policy cannot be altered in place, so a queue created with45 * another policy is dropped (with its pending jobs) and recreated.46 */47const QUEUE_POLICY = 'stately';4849async function ensureQueue(name: JobName, expireInSeconds: number): Promise<void> {50  const existing = await boss.getQueue(name);51  if (existing && existing.policy === QUEUE_POLICY) {52    await boss.updateQueue(name, { expireInSeconds, retryLimit: 0 });53    return;54  }55  if (existing) {56    log.warn({ queue: name, from: existing.policy, to: QUEUE_POLICY }, 'recreating queue with new policy');57    await boss.deleteQueue(name);58  }59  await boss.createQueue(name, { policy: QUEUE_POLICY, retryLimit: 0, expireInSeconds });60}6162/** Register cron schedules from manifests; one schedule key per connector so schedules can be diffed. */63async function registerSchedules(): Promise<void> {64  const active = CONNECTORS.filter((c) => c.manifest.status === 'active' && c.manifest.schedule);65  const wanted = new Set(active.map((c) => c.manifest.id));66  for (const s of await boss.getSchedules(JOBS.connectorRun)) {67    if (!wanted.has(s.key)) {68      await boss.unschedule(JOBS.connectorRun, s.key);69      log.info({ connector: s.key }, 'removed stale schedule');70    }71  }72  for (const c of active) {73    await boss.schedule(JOBS.connectorRun, c.manifest.schedule!, { id: c.manifest.id, requestedBy: 'schedule' }, { key: c.manifest.id, singletonKey: c.manifest.id, tz: 'UTC' });74    log.info({ connector: c.manifest.id, cron: c.manifest.schedule }, 'schedule registered');75  }76  const skipped = CONNECTORS.filter((c) => !wanted.has(c.manifest.id)).map((c) => `${c.manifest.id}(${c.manifest.status}${c.manifest.schedule ? '' : ',no-schedule'})`);77  if (skipped.length) log.info({ skipped }, 'connectors without schedule');78  await boss.schedule(JOBS.counters, SCHEDULES.counters, { thenRank: false, requestedBy: 'schedule' }, { key: 'daily', singletonKey: 'counters', tz: 'UTC' });79  await boss.schedule(JOBS.rank, SCHEDULES.rank, { requestedBy: 'schedule' }, { key: 'daily', singletonKey: 'rank', tz: 'UTC' });80  await boss.schedule(JOBS.healthProbe, SCHEDULES.healthProbe, { requestedBy: 'schedule' }, { key: 'hourly', singletonKey: 'health', tz: 'UTC' });81  log.info({ counters: SCHEDULES.counters, rank: SCHEDULES.rank, health: SCHEDULES.healthProbe }, 'maintenance schedules registered (UTC)');82}8384async function handleConnectorRun([job]: Job<ConnectorRunData>[]): Promise<void> {85  const data = job!.data;86  const connector = getConnector(data.id);87  if (!connector) {88    log.warn({ connector: data.id }, 'unknown connector — job dropped');89    return;90  }91  const [cur] = await db.execute<{ paused: boolean }>(sql`SELECT paused FROM connector_cursors WHERE connector_id = ${data.id}`);92  if (cur?.paused && data.requestedBy !== 'admin-api') {93    log.info({ connector: data.id }, 'connector paused — skipping scheduled run');94    return;95  }96  const opts: Parameters<typeof runConnector>[2] = { maxMinutes: data.maxMinutes ?? CI_MAX_RUN_MINUTES };97  if (data.mode) opts.mode = data.mode;98  if (data.maxRecords) opts.maxRecords = data.maxRecords;99  if (data.resetCursor) opts.resetCursor = true;100  log.info({ connector: data.id, jobId: job!.id, opts }, 'connector run starting');101  const result = await runConnector(db, connector, opts);102  log.info({ connector: data.id, runId: result.runId, status: result.status, ...result.counters }, 'connector run finished');103  const changed = result.counters.created + result.counters.updated;104  if (changed > 0 && data.mode !== 'dry_run') {105    // Derived layers follow canonical changes: counters, then rankings (chained by the counters handler).106    await boss.send(JOBS.counters, { thenRank: true, requestedBy: `connector:${data.id}` }, { singletonKey: 'counters' });107  }108}109110async function handleCounters([job]: Job<{ thenRank?: boolean; requestedBy?: string }>[]): Promise<void> {111  const t0 = Date.now();112  const n = await refreshCounters(db);113  log.info({ entities: n, ms: Date.now() - t0, requestedBy: job!.data.requestedBy }, 'counters refreshed');114  if (job!.data.thenRank) await boss.send(JOBS.rank, { requestedBy: 'counters' }, { singletonKey: 'rank' });115}116117async function handleRank([job]: Job<{ requestedBy?: string }>[]): Promise<void> {118  const t0 = Date.now();119  const res = await computeAllRankings(db);120  log.info({ snapshots: res.length, ms: Date.now() - t0, requestedBy: job!.data.requestedBy }, 'rankings recomputed');121}122123async function handleHealthProbe(): Promise<void> {124  for (const c of CONNECTORS) {125    const id = c.manifest.id;126    try {127      const [cur] = await db.execute<{ health: string }>(sql`SELECT health FROM connector_cursors WHERE connector_id = ${id}`);128      // Never overwrite the credentials gate set by runConnector/credentialsMissing (CLAUDE.md §142).129      if (cur?.health === 'awaiting_credentials' || c.credentialsMissing()) continue;130      const [src] = await db.execute<{ id: string }>(sql`SELECT id FROM sources WHERE slug = ${id}`);131      const ctx = new RunContext(db, c.manifest, src?.id ?? 'CI-SOURCE-00000000', 'probe', { maxMinutes: 2 }, 0);132      const h = await c.healthCheck(ctx);133      await db.execute(sql`134        INSERT INTO connector_cursors (connector_id, health, health_detail, updated_at) VALUES (${id}, ${h.status}, ${h.detail ?? null}, now())135        ON CONFLICT (connector_id) DO UPDATE SET health = ${h.status}, health_detail = ${h.detail ?? null}, updated_at = now()`);136      log.info({ connector: id, health: h.status, detail: h.detail }, 'health probe');137    } catch (err) {138      log.warn({ connector: id, err }, 'health probe failed');139      await db.execute(sql`INSERT INTO connector_cursors (connector_id, health, health_detail, updated_at) VALUES (${id}, 'failing', ${(err as Error).message.slice(0, 500)}, now())140        ON CONFLICT (connector_id) DO UPDATE SET health = 'failing', health_detail = EXCLUDED.health_detail, updated_at = now()`);141    }142  }143}144145async function main(): Promise<void> {146  await boss.start();147  await ensureQueue(JOBS.connectorRun, (CI_MAX_RUN_MINUTES + 20) * 60);148  await ensureQueue(JOBS.counters, 30 * 60);149  await ensureQueue(JOBS.rank, 60 * 60);150  await ensureQueue(JOBS.healthProbe, 10 * 60);151  await registerSchedules();152153  await boss.work<ConnectorRunData>(JOBS.connectorRun, { batchSize: 1, localConcurrency: WORKER_CONCURRENCY, pollingIntervalSeconds: 5 }, handleConnectorRun);154  await boss.work(JOBS.counters, { batchSize: 1, pollingIntervalSeconds: 5 }, handleCounters);155  await boss.work(JOBS.rank, { batchSize: 1, pollingIntervalSeconds: 5 }, handleRank);156  await boss.work(JOBS.healthProbe, { batchSize: 1, pollingIntervalSeconds: 30 }, handleHealthProbe);157  log.info({ concurrency: WORKER_CONCURRENCY, maxRunMinutes: CI_MAX_RUN_MINUTES, connectors: CONNECTORS.map((c) => c.manifest.id) }, 'worker ready');158}159160let stopping = false;161async function shutdown(signal: string): Promise<void> {162  if (stopping) return;163  stopping = true;164  log.info({ signal }, 'worker shutting down (graceful, up to 60 s)');165  try {166    await boss.stop({ graceful: true, close: true, timeout: 60_000 });167    await closeDb();168  } catch (err) {169    log.error({ err }, 'shutdown error');170  } finally {171    process.exit(0);172  }173}174process.on('SIGTERM', () => void shutdown('SIGTERM'));175process.on('SIGINT', () => void shutdown('SIGINT'));176177main().catch((err) => {178  log.error({ err }, 'worker failed to start');179  process.exit(1);180});181