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.intel    {thenRank?}                                          computeIntelligence (trial intelligence, drug pipeline, research gap)7 *   maintenance.rank     {}                                                   computeAllRankings8 *   health.probe         {}                                                   healthCheck per connector → connector_cursors.health9 *10 * Schedules are (re)registered from connector manifests on every start; stale schedules are removed.11 * pg-boss gotcha: it refuses `undefined` option values — always omit keys instead of passing undefined.12 */13import { CI_MAX_RUN_MINUTES, DATABASE_URL, JOBS, PGBOSS_SCHEMA, SCHEDULES, WORKER_CONCURRENCY, type JobName } from './lib/env.js';14import { PgBoss, type Job } from 'pg-boss';15import { sql } from 'drizzle-orm';16import { logger } from '@cancerindex/shared';17import { getDb, closeDb, raiseAlertSafe, resolveAlerts } from '@cancerindex/database';18import { ALERT_KINDS, CONNECTORS, getConnector, humanDuration, isStale, runConnector, RunContext, type RunMode } from '@cancerindex/connectors';19import { refreshCounters, computeAllRankings, computeIntelligence } from '@cancerindex/ranking';2021const log = logger.child({ component: 'worker' });22const db = getDb();2324interface ConnectorRunData {25  id: string;26  mode?: RunMode;27  maxMinutes?: number;28  maxRecords?: number;29  resetCursor?: boolean;30  requestedBy?: string;31}3233const boss = new PgBoss({34  connectionString: DATABASE_URL,35  schema: PGBOSS_SCHEMA,36  application_name: 'cancerindex-worker',37  max: Math.max(4, WORKER_CONCURRENCY + 2),38  // supervise + schedule are on: this process owns maintenance and cron.39});40boss.on('error', (err) => log.error({ err }, 'pg-boss error'));41boss.on('warning', (w) => log.warn({ w }, 'pg-boss warning'));4243/**44 * Queue policy `stately`: one job per singletonKey in created/retry/active — a connector is never45 * queued twice nor run concurrently. Policy cannot be altered in place, so a queue created with46 * another policy is dropped (with its pending jobs) and recreated.47 */48const QUEUE_POLICY = 'stately';4950async function ensureQueue(name: JobName, expireInSeconds: number): Promise<void> {51  const existing = await boss.getQueue(name);52  if (existing && existing.policy === QUEUE_POLICY) {53    await boss.updateQueue(name, { expireInSeconds, retryLimit: 0 });54    return;55  }56  if (existing) {57    log.warn({ queue: name, from: existing.policy, to: QUEUE_POLICY }, 'recreating queue with new policy');58    await boss.deleteQueue(name);59  }60  await boss.createQueue(name, { policy: QUEUE_POLICY, retryLimit: 0, expireInSeconds });61}6263/** Register cron schedules from manifests; one schedule key per connector so schedules can be diffed. */64async function registerSchedules(): Promise<void> {65  const active = CONNECTORS.filter((c) => c.manifest.status === 'active' && c.manifest.schedule);66  const wanted = new Set(active.map((c) => c.manifest.id));67  for (const s of await boss.getSchedules(JOBS.connectorRun)) {68    if (!wanted.has(s.key)) {69      await boss.unschedule(JOBS.connectorRun, s.key);70      log.info({ connector: s.key }, 'removed stale schedule');71    }72  }73  for (const c of active) {74    await boss.schedule(JOBS.connectorRun, c.manifest.schedule!, { id: c.manifest.id, requestedBy: 'schedule' }, { key: c.manifest.id, singletonKey: c.manifest.id, tz: 'UTC' });75    log.info({ connector: c.manifest.id, cron: c.manifest.schedule }, 'schedule registered');76  }77  const skipped = CONNECTORS.filter((c) => !wanted.has(c.manifest.id)).map((c) => `${c.manifest.id}(${c.manifest.status}${c.manifest.schedule ? '' : ',no-schedule'})`);78  if (skipped.length) log.info({ skipped }, 'connectors without schedule');79  await boss.schedule(JOBS.counters, SCHEDULES.counters, { thenRank: false, requestedBy: 'schedule' }, { key: 'daily', singletonKey: 'counters', tz: 'UTC' });80  await boss.schedule(JOBS.intel, SCHEDULES.intel, { thenRank: false, requestedBy: 'schedule' }, { key: 'daily', singletonKey: 'intel', tz: 'UTC' });81  await boss.schedule(JOBS.rank, SCHEDULES.rank, { requestedBy: 'schedule' }, { key: 'daily', singletonKey: 'rank', tz: 'UTC' });82  await boss.schedule(JOBS.healthProbe, SCHEDULES.healthProbe, { requestedBy: 'schedule' }, { key: 'hourly', singletonKey: 'health', tz: 'UTC' });83  log.info({ counters: SCHEDULES.counters, intel: SCHEDULES.intel, rank: SCHEDULES.rank, health: SCHEDULES.healthProbe }, 'maintenance schedules registered (UTC)');84}8586async function handleConnectorRun([job]: Job<ConnectorRunData>[]): Promise<void> {87  const data = job!.data;88  const connector = getConnector(data.id);89  if (!connector) {90    log.warn({ connector: data.id }, 'unknown connector — job dropped');91    return;92  }93  const [cur] = await db.execute<{ paused: boolean }>(sql`SELECT paused FROM connector_cursors WHERE connector_id = ${data.id}`);94  if (cur?.paused && data.requestedBy !== 'admin-api') {95    log.info({ connector: data.id }, 'connector paused — skipping scheduled run');96    return;97  }98  const opts: Parameters<typeof runConnector>[2] = { maxMinutes: data.maxMinutes ?? CI_MAX_RUN_MINUTES };99  if (data.mode) opts.mode = data.mode;100  if (data.maxRecords) opts.maxRecords = data.maxRecords;101  if (data.resetCursor) opts.resetCursor = true;102  log.info({ connector: data.id, jobId: job!.id, opts }, 'connector run starting');103  const result = await runConnector(db, connector, opts);104  log.info({ connector: data.id, runId: result.runId, status: result.status, ...result.counters }, 'connector run finished');105  const changed = result.counters.created + result.counters.updated;106  if (changed > 0 && data.mode !== 'dry_run') {107    // Derived layers follow canonical changes: counters, then rankings (chained by the counters handler).108    await boss.send(JOBS.counters, { thenRank: true, requestedBy: `connector:${data.id}` }, { singletonKey: 'counters' });109  }110}111112async function handleCounters([job]: Job<{ thenRank?: boolean; requestedBy?: string }>[]): Promise<void> {113  const t0 = Date.now();114  const n = await refreshCounters(db);115  log.info({ entities: n, ms: Date.now() - t0, requestedBy: job!.data.requestedBy }, 'counters refreshed');116  // Derived intelligence follows counters; rankings follow intelligence (chained by handleIntel).117  await boss.send(JOBS.intel, { thenRank: !!job!.data.thenRank, requestedBy: 'counters' }, { singletonKey: 'intel' });118}119120async function handleIntel([job]: Job<{ thenRank?: boolean; requestedBy?: string }>[]): Promise<void> {121  try {122    const r = await computeIntelligence(db, (msg, extra) => log.info({ ...extra }, msg));123    log.info({ ...r, requestedBy: job!.data.requestedBy }, 'intelligence recomputed');124  } catch (err) {125    log.error({ err }, 'intelligence failed (partial results kept)');126    await raiseAlertSafe(db, { kind: 'intelligence_failure', severity: 'warn', message: `intelligence recompute failed: ${(err as Error).message.slice(0, 300)}` });127  } finally {128    if (job!.data.thenRank) await boss.send(JOBS.rank, { requestedBy: 'intel' }, { singletonKey: 'rank' });129  }130}131132async function handleRank([job]: Job<{ requestedBy?: string }>[]): Promise<void> {133  const t0 = Date.now();134  const res = await computeAllRankings(db);135  log.info({ snapshots: res.length, ms: Date.now() - t0, requestedBy: job!.data.requestedBy }, 'rankings recomputed');136}137138/**139 * Hourly liveness probe → connector_cursors.health, plus alerts (CLAUDE.md §170): `source_failing`140 * when the probe fails, `source_stale` when an active scheduled connector has no success within 2×141 * its schedule interval; both are resolved here as soon as the condition clears (runConnector also142 * resolves them on the next successful run).143 */144async function handleHealthProbe(): Promise<void> {145  for (const c of CONNECTORS) {146    const id = c.manifest.id;147    let failing: string | null = null;148    let cur: { health: string; paused: boolean; last_success_at: Date | null } | undefined;149    try {150      [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}`);151      // Never overwrite the credentials gate set by runConnector/credentialsMissing (CLAUDE.md §142).152      if (cur?.health === 'awaiting_credentials' || c.credentialsMissing()) continue;153      const [src] = await db.execute<{ id: string }>(sql`SELECT id FROM sources WHERE slug = ${id}`);154      const ctx = new RunContext(db, c.manifest, src?.id ?? 'CI-SOURCE-00000000', 'probe', { maxMinutes: 2 }, 0);155      const h = await c.healthCheck(ctx);156      await db.execute(sql`157        INSERT INTO connector_cursors (connector_id, health, health_detail, updated_at) VALUES (${id}, ${h.status}, ${h.detail ?? null}, now())158        ON CONFLICT (connector_id) DO UPDATE SET health = ${h.status}, health_detail = ${h.detail ?? null}, updated_at = now()`);159      log.info({ connector: id, health: h.status, detail: h.detail }, 'health probe');160      if (h.status === 'failing') failing = h.detail ?? 'health check failing';161    } catch (err) {162      log.warn({ connector: id, err }, 'health probe failed');163      failing = (err as Error).message.slice(0, 500);164      await db.execute(sql`INSERT INTO connector_cursors (connector_id, health, health_detail, updated_at) VALUES (${id}, 'failing', ${failing}, now())165        ON CONFLICT (connector_id) DO UPDATE SET health = 'failing', health_detail = EXCLUDED.health_detail, updated_at = now()`);166    }167    try {168      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)}` });169      else await resolveAlerts(db, ALERT_KINDS.sourceFailing, id);170      const active = c.manifest.status === 'active' && !cur?.paused && !!c.manifest.schedule;171      const st = isStale(cur?.last_success_at ?? null, c.manifest.schedule ?? null);172      if (active && st.stale) {173        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 } });174      } else await resolveAlerts(db, ALERT_KINDS.sourceStale, id);175    } catch (err) {176      log.warn({ connector: id, err }, 'alert bookkeeping failed');177    }178  }179}180181async function main(): Promise<void> {182  await boss.start();183  await ensureQueue(JOBS.connectorRun, (CI_MAX_RUN_MINUTES + 20) * 60);184  await ensureQueue(JOBS.counters, 30 * 60);185  await ensureQueue(JOBS.intel, 60 * 60);186  await ensureQueue(JOBS.rank, 60 * 60);187  await ensureQueue(JOBS.healthProbe, 10 * 60);188  await registerSchedules();189190  await boss.work<ConnectorRunData>(JOBS.connectorRun, { batchSize: 1, localConcurrency: WORKER_CONCURRENCY, pollingIntervalSeconds: 5 }, handleConnectorRun);191  await boss.work(JOBS.counters, { batchSize: 1, pollingIntervalSeconds: 5 }, handleCounters);192  await boss.work(JOBS.intel, { batchSize: 1, pollingIntervalSeconds: 5 }, handleIntel);193  await boss.work(JOBS.rank, { batchSize: 1, pollingIntervalSeconds: 5 }, handleRank);194  await boss.work(JOBS.healthProbe, { batchSize: 1, pollingIntervalSeconds: 30 }, handleHealthProbe);195  log.info({ concurrency: WORKER_CONCURRENCY, maxRunMinutes: CI_MAX_RUN_MINUTES, connectors: CONNECTORS.map((c) => c.manifest.id) }, 'worker ready');196}197198let stopping = false;199async function shutdown(signal: string): Promise<void> {200  if (stopping) return;201  stopping = true;202  log.info({ signal }, 'worker shutting down (graceful, up to 60 s)');203  try {204    await boss.stop({ graceful: true, close: true, timeout: 60_000 });205    await closeDb();206  } catch (err) {207    log.error({ err }, 'shutdown error');208  } finally {209    process.exit(0);210  }211}212process.on('SIGTERM', () => void shutdown('SIGTERM'));213process.on('SIGINT', () => void shutdown('SIGINT'));214215main().catch((err) => {216  log.error({ err }, 'worker failed to start');217  process.exit(1);218});219