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, raiseAlertSafe, resolveAlerts } from '@cancerindex/database';17import { ALERT_KINDS, CONNECTORS, getConnector, humanDuration, isStale, 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}122123/**124 * Hourly liveness probe → connector_cursors.health, plus alerts (CLAUDE.md §170): `source_failing`125 * when the probe fails, `source_stale` when an active scheduled connector has no success within 2×126 * its schedule interval; both are resolved here as soon as the condition clears (runConnector also127 * resolves them on the next successful run).128 */129async function handleHealthProbe(): Promise<void> {130  for (const c of CONNECTORS) {131    const id = c.manifest.id;132    let failing: string | null = null;133    let cur: { health: string; paused: boolean; last_success_at: Date | null } | undefined;134    try {135      [cur] = await db.execute<{ health: string; paused: boolean; last_success_at: Date | null }>(sql`SELECT health, paused, last_success_at FROM connector_cursors WHERE connector_id = ${id}`);136      // Never overwrite the credentials gate set by runConnector/credentialsMissing (CLAUDE.md §142).137      if (cur?.health === 'awaiting_credentials' || c.credentialsMissing()) continue;138      const [src] = await db.execute<{ id: string }>(sql`SELECT id FROM sources WHERE slug = ${id}`);139      const ctx = new RunContext(db, c.manifest, src?.id ?? 'CI-SOURCE-00000000', 'probe', { maxMinutes: 2 }, 0);140      const h = await c.healthCheck(ctx);141      await db.execute(sql`142        INSERT INTO connector_cursors (connector_id, health, health_detail, updated_at) VALUES (${id}, ${h.status}, ${h.detail ?? null}, now())143        ON CONFLICT (connector_id) DO UPDATE SET health = ${h.status}, health_detail = ${h.detail ?? null}, updated_at = now()`);144      log.info({ connector: id, health: h.status, detail: h.detail }, 'health probe');145      if (h.status === 'failing') failing = h.detail ?? 'health check failing';146    } catch (err) {147      log.warn({ connector: id, err }, 'health probe failed');148      failing = (err as Error).message.slice(0, 500);149      await db.execute(sql`INSERT INTO connector_cursors (connector_id, health, health_detail, updated_at) VALUES (${id}, 'failing', ${failing}, now())150        ON CONFLICT (connector_id) DO UPDATE SET health = 'failing', health_detail = EXCLUDED.health_detail, updated_at = now()`);151    }152    try {153      if (failing) await raiseAlertSafe(db, { kind: ALERT_KINDS.sourceFailing, severity: 'warn', connectorId: id, message: `source health check failing: ${failing.split('\n')[0]!.slice(0, 200)}` });154      else await resolveAlerts(db, ALERT_KINDS.sourceFailing, id);155      const active = c.manifest.status === 'active' && !cur?.paused && !!c.manifest.schedule;156      const st = isStale(cur?.last_success_at ?? null, c.manifest.schedule ?? null);157      if (active && st.stale) {158        await raiseAlertSafe(db, { kind: ALERT_KINDS.sourceStale, severity: 'warn', connectorId: id, message: cur?.last_success_at ? 'no successful run within 2× the schedule interval' : 'scheduled connector has never succeeded', detail: { lastSuccessAt: cur?.last_success_at ?? null, age: humanDuration(st.ageMs), limit: humanDuration(st.limitMs), schedule: c.manifest.schedule } });159      } else await resolveAlerts(db, ALERT_KINDS.sourceStale, id);160    } catch (err) {161      log.warn({ connector: id, err }, 'alert bookkeeping failed');162    }163  }164}165166async function main(): Promise<void> {167  await boss.start();168  await ensureQueue(JOBS.connectorRun, (CI_MAX_RUN_MINUTES + 20) * 60);169  await ensureQueue(JOBS.counters, 30 * 60);170  await ensureQueue(JOBS.rank, 60 * 60);171  await ensureQueue(JOBS.healthProbe, 10 * 60);172  await registerSchedules();173174  await boss.work<ConnectorRunData>(JOBS.connectorRun, { batchSize: 1, localConcurrency: WORKER_CONCURRENCY, pollingIntervalSeconds: 5 }, handleConnectorRun);175  await boss.work(JOBS.counters, { batchSize: 1, pollingIntervalSeconds: 5 }, handleCounters);176  await boss.work(JOBS.rank, { batchSize: 1, pollingIntervalSeconds: 5 }, handleRank);177  await boss.work(JOBS.healthProbe, { batchSize: 1, pollingIntervalSeconds: 30 }, handleHealthProbe);178  log.info({ concurrency: WORKER_CONCURRENCY, maxRunMinutes: CI_MAX_RUN_MINUTES, connectors: CONNECTORS.map((c) => c.manifest.id) }, 'worker ready');179}180181let stopping = false;182async function shutdown(signal: string): Promise<void> {183  if (stopping) return;184  stopping = true;185  log.info({ signal }, 'worker shutting down (graceful, up to 60 s)');186  try {187    await boss.stop({ graceful: true, close: true, timeout: 60_000 });188    await closeDb();189  } catch (err) {190    log.error({ err }, 'shutdown error');191  } finally {192    process.exit(0);193  }194}195process.on('SIGTERM', () => void shutdown('SIGTERM'));196process.on('SIGINT', () => void shutdown('SIGINT'));197198main().catch((err) => {199  log.error({ err }, 'worker failed to start');200  process.exit(1);201});202