spb/cancerindex
Public
TypeScript 97.2%
SQL 1.5%
CSS 0.6%
JavaScript 0.5%
1import { and, desc, eq, gt, inArray, ne, notInArray, sql } from 'drizzle-orm';2import { createHash } from 'node:crypto';3import { formatRunId, logger, type Logger, type ProvenanceInput, envInt } from '@cancerindex/shared';4import { type Database, ingestRuns, connectorCursors, sourceRecords, provenance, connectorFieldStats, sources, unresolvedLabels, changeEvents, raiseAlertSafe, resolveAlerts } from '@cancerindex/database';5import type { ConnectorManifest } from './manifest.js';6import { HttpClient } from './http.js';7import { RawLake } from './lake.js';89export type RunMode = 'full' | 'incremental' | 'backfill' | 'dry_run' | 'probe';1011export interface RunLimits {12 /** Stop fetching new pages after this many minutes; cursor is saved so the next run resumes. */13 maxMinutes: number;14 /** Optional cap on records for smoke runs (CLAUDE.md §227: prove on 10/100/1000 first). */15 maxRecords?: number;16}1718export interface UpsertResult {19 id: number;20 status: 'created' | 'updated' | 'unchanged';21}2223export interface UpsertOptions {24 sourceUpdatedAt?: Date | null;25 canonicalType?: string;26 canonicalId?: string;27}2829export interface BatchUpsertItem extends UpsertOptions {30 sourceRecordId: string;31 payload: unknown;32}3334export interface FieldStatsCollector {35 observe(entity: string, obj: unknown): void;36}3738/** Thrown by `ctx.guardCount` (CLAUDE.md §171). `runConnector` records it in `ingest_runs.anomaly`. */39export class AnomalyError extends Error {40 constructor(41 message: string,42 readonly entity: string,43 readonly fetched: number,44 readonly previous: number,45 readonly minRatio: number,46 ) {47 super(message);48 this.name = 'AnomalyError';49 }50}5152/** Kinds of `system_alerts` rows raised by the SDK / worker (CLAUDE.md §170). */53export const ALERT_KINDS = {54 connectorFailure: 'connector_failure',55 connectorAborted: 'connector_aborted',56 anomaly: 'anomaly',57 schemaDrift: 'schema_drift',58 sourceFailing: 'source_failing',59 sourceStale: 'source_stale',60} as const;6162/** Alerts a successful run clears for its connector. */63const RESOLVED_ON_SUCCESS = [ALERT_KINDS.connectorFailure, ALERT_KINDS.connectorAborted, ALERT_KINDS.anomaly, ALERT_KINDS.sourceFailing, ALERT_KINDS.sourceStale];6465const CHECKPOINT_INTERVAL_MS = 60_000;6667/**68 * Ingest run context: identifiers, counters, raw lake, HTTP client, provenance/record helpers.69 * Every connector receives one per execution (CLAUDE.md §90, §175-176).70 *71 * Restartability: `cursor` is persisted (a) explicitly with `saveCursor()`, (b) automatically every72 * `manifest.checkpointEvery` upserted records or 60 s when it changed, (c) on SIGTERM/SIGINT73 * (`requestAbort`), (d) at the end of the run whatever its status. A connector must only mutate74 * `ctx.cursor` at a point where everything before it has been persisted (end of a page / batch).75 */76export class RunContext {77 readonly runId: string;78 readonly startedAt = new Date();79 readonly http: HttpClient;80 readonly lake: RawLake;81 readonly log: Logger;82 readonly counters = { fetched: 0, created: 0, updated: 0, unchanged: 0, rejected: 0, validationFailures: 0 };83 readonly drift: Array<{ entity: string; field: string; kind: 'new_field' | 'type_change' | 'nullability'; detail: string }> = [];84 private logs: Array<{ t: string; level: string; msg: string }> = [];85 private fieldStats = new Map<string, { types: Set<string>; seen: number; nulls: number }>();86 private knownFields = new Map<string, Set<string>>();87 private knownTypes = new Map<string, Set<string>>();88 private deadline: number;89 private timeBudgetHit = false;90 private lastSavedCursor: string | null = null;91 private recordsSinceCheckpoint = 0;92 private lastCheckpointAt = Date.now();93 private savingCursor: Promise<void> | null = null;94 private abortPromise: Promise<void> | null = null;95 /** Reason of the abort request (signal name…), null while running normally. */96 abortReason: string | null = null;97 /** Set by `guardCount` (or by runConnector for `anomaly:` errors); copied to `ingest_runs.anomaly`. */98 anomaly: string | null = null;99 /** Number of cursor checkpoints written during this run (diagnostics / tests). */100 checkpoints = 0;101 cursor: Record<string, unknown> = {};102 datasetVersion: string | undefined;103104 constructor(105 readonly db: Database,106 readonly manifest: ConnectorManifest,107 readonly sourceId: string,108 readonly mode: RunMode,109 readonly limits: RunLimits,110 seq: number,111 ) {112 this.runId = formatRunId(manifest.id, this.startedAt, seq);113 this.http = new HttpClient(manifest);114 this.lake = new RawLake(manifest.id, this.runId, this.startedAt);115 this.log = logger.child({ run: this.runId, connector: manifest.id });116 this.deadline = this.startedAt.getTime() + limits.maxMinutes * 60_000;117 }118119 /** True when the time budget, the record cap or an abort request is reached; connectors check it between pages. */120 shouldStop(): boolean {121 if (this.abortReason) return true;122 if (Date.now() > this.deadline) {123 if (!this.timeBudgetHit) this.info('time budget reached — cursor will be saved for the next run');124 this.timeBudgetHit = true;125 return true;126 }127 if (this.limits.maxRecords !== undefined && this.counters.fetched >= this.limits.maxRecords) return true;128 return false;129 }130 get stoppedEarly(): boolean {131 return this.timeBudgetHit || (this.limits.maxRecords !== undefined && this.counters.fetched >= this.limits.maxRecords);132 }133 /** Cursor writes are skipped in these modes (a dry run / probe must leave the real cursor untouched). */134 get persistsCursor(): boolean {135 return this.mode !== 'dry_run' && this.mode !== 'probe';136 }137138 info(msg: string, extra?: Record<string, unknown>) {139 this.log.info(extra ?? {}, msg);140 this.push('info', msg);141 }142 warn(msg: string, extra?: Record<string, unknown>) {143 this.log.warn(extra ?? {}, msg);144 this.push('warn', msg);145 }146 error(msg: string, extra?: Record<string, unknown>) {147 this.log.error(extra ?? {}, msg);148 this.push('error', msg);149 }150 private push(level: string, msg: string) {151 if (this.logs.length < 500) this.logs.push({ t: new Date().toISOString(), level, msg });152 }153154 /* ------------------------------------------------------------------------------------------ */155 /* Cursor persistence (CLAUDE.md §90) */156 /* ------------------------------------------------------------------------------------------ */157158 /**159 * Persist `cursor` to `connector_cursors` (and `ingest_runs.cursor_after`) right now. Idempotent160 * and serialised: concurrent calls share one write. No-op in dry_run / probe modes.161 */162 async saveCursor(): Promise<void> {163 if (!this.persistsCursor) return;164 if (this.savingCursor) await this.savingCursor;165 const json = JSON.stringify(this.cursor ?? {});166 if (json === this.lastSavedCursor) return;167 this.savingCursor = (async () => {168 const now = new Date();169 await this.db170 .insert(connectorCursors)171 .values({ connectorId: this.manifest.id, cursor: this.cursor, lastAttemptAt: now, updatedAt: now })172 .onConflictDoUpdate({ target: connectorCursors.connectorId, set: { cursor: this.cursor, updatedAt: now } });173 await this.db.update(ingestRuns).set({ cursorAfter: this.cursor }).where(eq(ingestRuns.id, this.runId));174 this.lastSavedCursor = json;175 this.checkpoints++;176 })();177 try {178 await this.savingCursor;179 } finally {180 this.savingCursor = null;181 }182 this.recordsSinceCheckpoint = 0;183 this.lastCheckpointAt = Date.now();184 }185186 /**187 * Count `records` persisted records towards the automatic checkpoint and save the cursor when the188 * threshold (`manifest.checkpointEvery` records or 60 s) is reached and the cursor changed.189 * Called by `upsertSourceRecord` / `upsertSourceRecordsBatch`; connectors with their own batch190 * writers call it after each batch.191 */192 async checkpoint(records = 1): Promise<void> {193 if (!this.persistsCursor) return;194 this.recordsSinceCheckpoint += records;195 const due = this.recordsSinceCheckpoint >= this.manifest.checkpointEvery || Date.now() - this.lastCheckpointAt >= CHECKPOINT_INTERVAL_MS;196 if (!due) return;197 this.recordsSinceCheckpoint = 0;198 this.lastCheckpointAt = Date.now();199 await this.saveCursor();200 }201202 /**203 * Abort request (SIGTERM/SIGINT, operator): makes `shouldStop()` true, persists the cursor and204 * provisionally marks the run `aborted` so the state is right even if the process is killed before205 * the connector returns. `runConnector` finalises the run when `sync()` comes back.206 */207 requestAbort(reason: string): Promise<void> {208 if (this.abortPromise) return this.abortPromise;209 this.abortReason = reason;210 this.warn(`abort requested (${reason}) — saving cursor, run marked aborted; the connector stops at its next shouldStop() check`);211 this.abortPromise = (async () => {212 try {213 await this.lake.flush();214 await this.saveCursor();215 if (this.persistsCursor || this.mode === 'dry_run') {216 await this.db217 .update(ingestRuns)218 .set({ status: 'aborted', error: reason, cursorAfter: this.cursor, finishedAt: new Date(), recordsFetched: this.counters.fetched, recordsCreated: this.counters.created, recordsUpdated: this.counters.updated, recordsUnchanged: this.counters.unchanged, log: this.logs })219 .where(eq(ingestRuns.id, this.runId));220 }221 } catch (e) {222 this.log.error({ err: e }, 'abort bookkeeping failed');223 }224 })();225 return this.abortPromise;226 }227228 /* ------------------------------------------------------------------------------------------ */229 /* Anomaly guard (CLAUDE.md §171) */230 /* ------------------------------------------------------------------------------------------ */231232 /**233 * Compare a total the source declares (or the number of records parsed) with the previous234 * successful run's `records_fetched`. Throws `AnomalyError` — and marks the run — when235 * `fetchedCount < previous × manifest.anomalyGuard.minRatioOfPrevious`. Never destructive: it only236 * refuses to continue, the previous data stays published. Dry runs and probes are excluded from237 * the baseline; the first run of a connector is accepted as the baseline.238 */239 async guardCount(entity: string, fetchedCount: number, opts: { minRatio?: number } = {}): Promise<{ previous: number | null; ratio: number | null }> {240 const minRatio = opts.minRatio ?? this.manifest.anomalyGuard.minRatioOfPrevious;241 const [prev] = await this.db242 .select({ id: ingestRuns.id, n: ingestRuns.recordsFetched })243 .from(ingestRuns)244 .where(and(eq(ingestRuns.connectorId, this.manifest.id), eq(ingestRuns.status, 'succeeded'), notInArray(ingestRuns.mode, ['dry_run', 'probe']), ne(ingestRuns.id, this.runId), gt(ingestRuns.recordsFetched, 0)))245 .orderBy(desc(ingestRuns.startedAt))246 .limit(1);247 if (!prev) {248 this.info(`anomaly guard: no previous successful run — ${entity} count ${fetchedCount} accepted as baseline`);249 return { previous: null, ratio: null };250 }251 const ratio = fetchedCount / prev.n;252 if (ratio < minRatio) {253 const msg = `anomaly: ${entity} count ${fetchedCount} is ${(ratio * 100).toFixed(1)}% of the previous successful run ${prev.id} (${prev.n} records fetched) — below anomalyGuard.minRatioOfPrevious=${minRatio}; refusing to continue, previous data stays published (CLAUDE.md §171)`;254 this.anomaly = msg;255 this.error(msg);256 throw new AnomalyError(msg, entity, fetchedCount, prev.n, minRatio);257 }258 this.info(`anomaly guard: ${entity} count ${fetchedCount} = ${(ratio * 100).toFixed(1)}% of previous run ${prev.id} (${prev.n}) ≥ ${minRatio}`);259 return { previous: prev.n, ratio };260 }261262 /* ------------------------------------------------------------------------------------------ */263 /* Schema drift */264 /* ------------------------------------------------------------------------------------------ */265266 /** Record observed fields for schema-drift detection (CLAUDE.md §25). */267 observe(entity: string, obj: unknown): void {268 if (!obj || typeof obj !== 'object') return;269 const known = this.knownFields.get(entity);270 for (const [k, v] of Object.entries(obj as Record<string, unknown>)) {271 const key = `${entity} ${k}`;272 const type = v === null || v === undefined ? 'null' : Array.isArray(v) ? 'array' : typeof v;273 let fs = this.fieldStats.get(key);274 if (!fs) {275 fs = { types: new Set(), seen: 0, nulls: 0 };276 this.fieldStats.set(key, fs);277 if (known && !known.has(k)) this.drift.push({ entity, field: k, kind: 'new_field', detail: `type=${type}` });278 }279 fs.seen++;280 if (type === 'null') fs.nulls++;281 else fs.types.add(type);282 }283 }284285 async loadKnownFields(): Promise<void> {286 const rows = await this.db.select({ entity: connectorFieldStats.entity, field: connectorFieldStats.field, types: connectorFieldStats.types }).from(connectorFieldStats).where(eq(connectorFieldStats.connectorId, this.manifest.id));287 for (const r of rows) {288 if (!this.knownFields.has(r.entity)) this.knownFields.set(r.entity, new Set());289 this.knownFields.get(r.entity)!.add(r.field);290 }291 this.knownTypes = new Map(rows.map((r) => [`${r.entity} ${r.field}`, new Set(r.types)]));292 }293294 private async flushFieldStats(): Promise<void> {295 for (const [key, fs] of this.fieldStats) {296 const [entity, field] = key.split(' ') as [string, string];297 const prev = this.knownTypes.get(key);298 if (prev && prev.size > 0) {299 for (const t of fs.types) if (!prev.has(t)) this.drift.push({ entity, field, kind: 'type_change', detail: `${[...prev].join('|')} → ${t}` });300 }301 await this.db302 .insert(connectorFieldStats)303 .values({ connectorId: this.manifest.id, entity, field, types: [...fs.types], seenCount: fs.seen, nullCount: fs.nulls, firstSeenRun: this.runId, lastSeenRun: this.runId })304 .onConflictDoUpdate({305 target: [connectorFieldStats.connectorId, connectorFieldStats.entity, connectorFieldStats.field],306 set: {307 types: sql`COALESCE((SELECT array_agg(DISTINCT x) FROM unnest(${connectorFieldStats.types} || ${sql.raw(`ARRAY[${[...fs.types].map((t) => `'${t.replace(/'/g, '')}'`).join(',')}]::text[]`)}) AS x), '{}'::text[])`,308 seenCount: sql`${connectorFieldStats.seenCount} + ${fs.seen}`,309 nullCount: sql`${connectorFieldStats.nullCount} + ${fs.nulls}`,310 lastSeenRun: this.runId,311 updatedAt: new Date(),312 },313 });314 }315 }316317 /* ------------------------------------------------------------------------------------------ */318 /* Records, provenance, curation queue */319 /* ------------------------------------------------------------------------------------------ */320321 /** Upsert a source-native record (idempotency, CLAUDE.md §91). Writes the raw payload to the lake. */322 async upsertSourceRecord(entityKind: string, sourceRecordId: string, payload: unknown, opts: UpsertOptions = {}): Promise<UpsertResult> {323 const json = JSON.stringify(payload);324 const hash = createHash('sha256').update(json).digest('hex');325 this.counters.fetched++;326 this.observe(entityKind, payload);327 const existing = await this.db328 .select({ id: sourceRecords.id, payloadHash: sourceRecords.payloadHash })329 .from(sourceRecords)330 .where(and(eq(sourceRecords.sourceId, this.sourceId), eq(sourceRecords.entityKind, entityKind), eq(sourceRecords.sourceRecordId, sourceRecordId)))331 .limit(1);332 const row = existing[0];333 let result: UpsertResult;334 if (row && row.payloadHash === hash) {335 this.counters.unchanged++;336 await this.db.update(sourceRecords).set({ lastSeenRun: this.runId, status: 'active', ...(opts.canonicalId ? { canonicalType: opts.canonicalType, canonicalId: opts.canonicalId } : {}) }).where(eq(sourceRecords.id, row.id));337 result = { id: row.id, status: 'unchanged' };338 } else {339 const rawPath = this.manifest.rawRetention === 'none' || this.mode === 'dry_run' ? null : await this.lake.put(entityKind, payload);340 if (row) {341 this.counters.updated++;342 await this.db343 .update(sourceRecords)344 .set({ payloadHash: hash, rawPath, lastSeenRun: this.runId, retrievedAt: new Date(), sourceUpdatedAt: opts.sourceUpdatedAt ?? null, status: 'active', canonicalType: opts.canonicalType, canonicalId: opts.canonicalId, updatedAt: new Date() })345 .where(eq(sourceRecords.id, row.id));346 result = { id: row.id, status: 'updated' };347 } else {348 this.counters.created++;349 const inserted = await this.db350 .insert(sourceRecords)351 .values({ sourceId: this.sourceId, entityKind, sourceRecordId, payloadHash: hash, rawPath, firstSeenRun: this.runId, lastSeenRun: this.runId, sourceUpdatedAt: opts.sourceUpdatedAt ?? null, canonicalType: opts.canonicalType, canonicalId: opts.canonicalId })352 .returning({ id: sourceRecords.id });353 result = { id: inserted[0]!.id, status: 'created' };354 }355 }356 await this.checkpoint(1);357 return result;358 }359360 /**361 * Batched `upsertSourceRecord` for bulk connectors (one SELECT + one INSERT per batch). Same362 * idempotency, raw-lake and checkpoint semantics; returns the result per `sourceRecordId`.363 */364 async upsertSourceRecordsBatch(entityKind: string, items: BatchUpsertItem[]): Promise<Map<string, UpsertResult>> {365 const out = new Map<string, UpsertResult>();366 if (!items.length) return out;367 const hashes = new Map<string, string>();368 for (const it of items) {369 hashes.set(it.sourceRecordId, createHash('sha256').update(JSON.stringify(it.payload)).digest('hex'));370 this.observe(entityKind, it.payload);371 }372 this.counters.fetched += items.length;373 const existing = await this.db374 .select({ id: sourceRecords.id, sourceRecordId: sourceRecords.sourceRecordId, payloadHash: sourceRecords.payloadHash })375 .from(sourceRecords)376 .where(and(eq(sourceRecords.sourceId, this.sourceId), eq(sourceRecords.entityKind, entityKind), inArray(sourceRecords.sourceRecordId, items.map((i) => i.sourceRecordId))));377 const byId = new Map(existing.map((e) => [e.sourceRecordId, e]));378 const unchangedIds: number[] = [];379 const toInsert: Array<{ item: BatchUpsertItem; values: typeof sourceRecords.$inferInsert }> = [];380 const now = new Date();381 const writeRaw = this.manifest.rawRetention !== 'none' && this.mode !== 'dry_run';382 for (const it of items) {383 const hash = hashes.get(it.sourceRecordId)!;384 const ex = byId.get(it.sourceRecordId);385 if (ex && ex.payloadHash === hash) {386 unchangedIds.push(ex.id);387 this.counters.unchanged++;388 out.set(it.sourceRecordId, { id: ex.id, status: 'unchanged' });389 continue;390 }391 const rawPath = writeRaw ? await this.lake.put(entityKind, it.payload) : null;392 if (ex) {393 this.counters.updated++;394 await this.db395 .update(sourceRecords)396 .set({ payloadHash: hash, rawPath, lastSeenRun: this.runId, retrievedAt: now, sourceUpdatedAt: it.sourceUpdatedAt ?? null, status: 'active', canonicalType: it.canonicalType, canonicalId: it.canonicalId, updatedAt: now })397 .where(eq(sourceRecords.id, ex.id));398 out.set(it.sourceRecordId, { id: ex.id, status: 'updated' });399 } else {400 this.counters.created++;401 toInsert.push({ item: it, values: { sourceId: this.sourceId, entityKind, sourceRecordId: it.sourceRecordId, payloadHash: hash, rawPath, firstSeenRun: this.runId, lastSeenRun: this.runId, sourceUpdatedAt: it.sourceUpdatedAt ?? null, canonicalType: it.canonicalType, canonicalId: it.canonicalId } });402 }403 }404 if (unchangedIds.length) await this.db.update(sourceRecords).set({ lastSeenRun: this.runId, status: 'active' }).where(inArray(sourceRecords.id, unchangedIds));405 if (toInsert.length) {406 const inserted = await this.db407 .insert(sourceRecords)408 .values(toInsert.map((t) => t.values))409 .onConflictDoNothing()410 .returning({ id: sourceRecords.id, sourceRecordId: sourceRecords.sourceRecordId });411 const idBySrc = new Map(inserted.map((r) => [r.sourceRecordId, r.id]));412 for (const t of toInsert) out.set(t.item.sourceRecordId, { id: idBySrc.get(t.item.sourceRecordId) ?? -1, status: 'created' });413 }414 await this.checkpoint(items.length);415 return out;416 }417418 /** Insert a provenance row (CLAUDE.md §2) and return its id. */419 async addProvenance(p: Omit<ProvenanceInput, 'sourceId' | 'retrievedAt' | 'ingestRunId'> & { retrievedAt?: string }): Promise<number> {420 const rows = await this.db421 .insert(provenance)422 .values({423 sourceId: this.sourceId,424 sourceRecordId: p.sourceRecordId,425 sourceUrl: p.sourceUrl,426 dataset: p.dataset,427 datasetVersion: p.datasetVersion ?? this.datasetVersion,428 publicationId: p.publicationId,429 pmid: p.pmid,430 doi: p.doi,431 retrievedAt: p.retrievedAt ? new Date(p.retrievedAt) : new Date(),432 publishedAt: p.publishedAt,433 updatedAtSource: p.updatedAt,434 geography: p.geography,435 population: p.population,436 cohortSize: p.cohortSize,437 methodology: p.methodology,438 evidenceType: p.evidenceType,439 accessLevel: p.accessLevel,440 confidence: p.confidence,441 license: p.license ?? this.manifest.license,442 ingestRunId: this.runId,443 })444 .returning({ id: provenance.id });445 return rows[0]!.id;446 }447448 /** Queue an unmapped label for curation (CLAUDE.md §222). */449 async recordUnresolved(entityKind: string, sourceText: string, normalized: string, context: Record<string, unknown> = {}, suggestion?: { id: string; matchType: string; score: number }): Promise<void> {450 if (!normalized) return;451 await this.db452 .insert(unresolvedLabels)453 .values({ sourceId: this.sourceId, entityKind, sourceText, normalized, context, suggestedId: suggestion?.id, suggestedMatchType: suggestion?.matchType, suggestedScore: suggestion?.score })454 .onConflictDoUpdate({455 target: [unresolvedLabels.sourceId, unresolvedLabels.entityKind, unresolvedLabels.normalized],456 set: { count: sql`${unresolvedLabels.count} + 1`, updatedAt: new Date(), ...(suggestion ? { suggestedId: suggestion.id, suggestedMatchType: suggestion.matchType, suggestedScore: suggestion.score } : {}) },457 });458 }459460 async recordChange(entityType: string, entityId: string, kind: string, summary: string, before?: unknown, after?: unknown): Promise<void> {461 await this.db.insert(changeEvents).values({ entityType, entityId, kind, summary, before: before ?? null, after: after ?? null, ingestRunId: this.runId });462 }463464 /** @internal */465 async _finish(status: 'succeeded' | 'failed' | 'partial' | 'aborted', error?: string, anomaly?: string): Promise<void> {466 if (this.abortPromise) await this.abortPromise;467 await this.lake.close();468 try {469 await this.flushFieldStats();470 } catch (e) {471 this.warn(`field stats flush failed: ${(e as Error).message}`);472 }473 const finishedAt = new Date();474 await this.db475 .update(ingestRuns)476 .set({477 status,478 finishedAt,479 durationMs: finishedAt.getTime() - this.startedAt.getTime(),480 recordsFetched: this.counters.fetched,481 recordsCreated: this.counters.created,482 recordsUpdated: this.counters.updated,483 recordsUnchanged: this.counters.unchanged,484 recordsRejected: this.counters.rejected,485 httpRequests: this.http.stats.requests,486 httpFailures: this.http.stats.failures,487 rateLimitEvents: this.http.stats.rateLimitEvents,488 validationFailures: this.counters.validationFailures,489 schemaDrift: this.drift,490 cursorAfter: this.cursor,491 error: error ?? null,492 log: this.logs,493 datasetVersion: this.datasetVersion ?? null,494 anomaly: anomaly ?? this.anomaly ?? null,495 })496 .where(eq(ingestRuns.id, this.runId));497 }498}499500export interface ConnectorHealth {501 status: 'healthy' | 'degraded' | 'failing' | 'review' | 'awaiting_credentials' | 'unknown';502 detail?: string;503 lastSuccessAt?: Date | null;504}505506/** Connector contract (CLAUDE.md §8). `sync` implements discover→fetch→normalize→reconcile→validate→persist. */507export abstract class Connector {508 abstract readonly manifest: ConnectorManifest;509 /** Cheap liveness probe against the source (no ingestion). */510 abstract healthCheck(ctx: RunContext): Promise<ConnectorHealth>;511 /** Main ingestion. Must be idempotent and restartable via ctx.cursor. */512 abstract sync(ctx: RunContext): Promise<void>;513 /** Optional: connector-specific credentials check. Return a reason when credentials are missing. */514 credentialsMissing(): string | null {515 return null;516 }517}518519export interface RunOptions {520 mode?: RunMode;521 maxMinutes?: number;522 maxRecords?: number;523 resetCursor?: boolean;524 /**525 * Install SIGTERM/SIGINT handlers for the duration of the run (default true): the first signal526 * saves the cursor and marks the run aborted, the connector stops at its next `shouldStop()`;527 * a second signal exits the process immediately (the run is already marked aborted).528 */529 handleSignals?: boolean;530}531532export interface RunResult {533 runId: string;534 status: 'succeeded' | 'partial' | 'failed' | 'aborted';535 counters: RunContext['counters'];536 anomaly: string | null;537}538539const SIGNALS: NodeJS.Signals[] = ['SIGTERM', 'SIGINT'];540541/** Execute one connector run end-to-end with bookkeeping in ingest_runs/connector_cursors/system_alerts. */542export async function runConnector(db: Database, connector: Connector, opts: RunOptions = {}): Promise<RunResult> {543 const m = connector.manifest;544 const [src] = await db.select({ id: sources.id, licenseStatus: sources.licenseStatus }).from(sources).where(eq(sources.slug, m.id)).limit(1);545 if (!src) throw new Error(`source ${m.id} not seeded — run db:seed`);546 const mode: RunMode = opts.mode ?? (m.supportsIncrementalSync ? 'incremental' : 'full');547 const missing = connector.credentialsMissing();548 const [{ n }] = (await db.execute<{ n: string }>(sql`SELECT count(*)::text AS n FROM ingest_runs WHERE connector_id = ${m.id} AND started_at::date = now()::date`)) as unknown as [{ n: string }];549 const ctx = new RunContext(db, m, src.id, mode, { maxMinutes: opts.maxMinutes ?? envInt('CI_MAX_RUN_MINUTES', 45), maxRecords: opts.maxRecords }, Number(n) + 1);550 const [cur] = await db.select().from(connectorCursors).where(eq(connectorCursors.connectorId, m.id)).limit(1);551 ctx.cursor = opts.resetCursor || mode === 'full' ? {} : { ...(cur?.cursor ?? {}) };552 await db.insert(ingestRuns).values({ id: ctx.runId, connectorId: m.id, sourceId: src.id, mode, cursorBefore: ctx.cursor });553 await db554 .insert(connectorCursors)555 .values({ connectorId: m.id, cursor: ctx.cursor, lastAttemptAt: new Date() })556 .onConflictDoUpdate({ target: connectorCursors.connectorId, set: { lastAttemptAt: new Date() } });557558 const alert = (kind: string, severity: 'info' | 'warn' | 'critical', message: string, detail: Record<string, unknown> = {}) => raiseAlertSafe(db, { kind, severity, connectorId: m.id, message: message.slice(0, 300), detail: { runId: ctx.runId, mode, ...detail } });559560 if (missing) {561 ctx.warn(`credentials missing: ${missing}`);562 await ctx._finish('aborted', missing);563 await db.update(connectorCursors).set({ health: 'awaiting_credentials', healthDetail: missing, updatedAt: new Date() }).where(eq(connectorCursors.connectorId, m.id));564 await alert(ALERT_KINDS.connectorAborted, 'warn', `credentials missing: ${missing}`);565 return { runId: ctx.runId, status: 'aborted', counters: ctx.counters, anomaly: null };566 }567 if (m.licenseStatus === 'blocked') {568 const msg = 'license status blocked — connector will not ingest (CLAUDE.md §142)';569 ctx.warn(msg);570 await ctx._finish('aborted', msg);571 return { runId: ctx.runId, status: 'aborted', counters: ctx.counters, anomaly: null };572 }573574 // Signal handling: first signal → graceful abort with cursor saved; second → exit now.575 const onSignal = (signal: NodeJS.Signals) => {576 if (ctx.abortReason) {577 ctx.log.error({ signal }, 'second signal — exiting immediately (run already marked aborted, cursor saved)');578 process.exit(130);579 }580 void ctx.requestAbort(`${signal} received`);581 };582 const handleSignals = opts.handleSignals ?? true;583 if (handleSignals) for (const s of SIGNALS) process.on(s, onSignal);584585 try {586 await ctx.loadKnownFields();587 await connector.sync(ctx);588 if (ctx.abortReason) {589 await ctx._finish('aborted', ctx.abortReason);590 await alert(ALERT_KINDS.connectorAborted, 'info', `run aborted: ${ctx.abortReason}`, { cursor: ctx.cursor, fetched: ctx.counters.fetched });591 return { runId: ctx.runId, status: 'aborted', counters: ctx.counters, anomaly: null };592 }593 const status = ctx.stoppedEarly ? 'partial' : 'succeeded';594 await ctx._finish(status);595 if (mode !== 'dry_run') {596 await db597 .update(connectorCursors)598 .set({ cursor: ctx.cursor, lastSuccessAt: new Date(), health: ctx.drift.length ? 'degraded' : 'healthy', healthDetail: ctx.drift.length ? `${ctx.drift.length} schema drift signal(s)` : null, updatedAt: new Date() })599 .where(eq(connectorCursors.connectorId, m.id));600 await db.update(sources).set({ status: 'active', updatedAt: new Date() }).where(eq(sources.id, src.id));601 // Alerts (CLAUDE.md §170): a success clears failure/stale alerts; drift is raised or cleared.602 try {603 await resolveAlerts(db, RESOLVED_ON_SUCCESS, m.id);604 if (ctx.drift.length) await alert(ALERT_KINDS.schemaDrift, 'warn', 'schema drift detected on upstream records', { signals: ctx.drift.length, sample: ctx.drift.slice(0, 20) });605 else await resolveAlerts(db, ALERT_KINDS.schemaDrift, m.id);606 } catch (e) {607 ctx.warn(`alert bookkeeping failed: ${(e as Error).message}`);608 }609 }610 return { runId: ctx.runId, status, counters: ctx.counters, anomaly: null };611 } catch (err) {612 const e = err as Error;613 const msg = err instanceof Error ? `${err.message}\n${err.stack ?? ''}`.slice(0, 4000) : String(err);614 if (ctx.abortReason) {615 // The connector threw while stopping (e.g. closed socket) — the abort is the real status.616 ctx.warn(`connector threw during abort: ${e.message}`);617 await ctx._finish('aborted', `${ctx.abortReason}; ${e.message}`.slice(0, 4000));618 await alert(ALERT_KINDS.connectorAborted, 'info', `run aborted: ${ctx.abortReason}`, { cursor: ctx.cursor, fetched: ctx.counters.fetched });619 return { runId: ctx.runId, status: 'aborted', counters: ctx.counters, anomaly: null };620 }621 const anomaly = ctx.anomaly ?? (err instanceof AnomalyError || /^anomaly\b/i.test(e.message ?? '') ? e.message.slice(0, 2000) : null);622 ctx.error(`run failed: ${e.message}`);623 await ctx._finish('failed', msg, anomaly ?? undefined);624 // Preserve partial cursor progress so restarts resume (CLAUDE.md §90).625 await db.update(connectorCursors).set({ cursor: ctx.cursor, health: 'failing', healthDetail: e.message?.slice(0, 500), updatedAt: new Date() }).where(eq(connectorCursors.connectorId, m.id));626 if (anomaly) await alert(ALERT_KINDS.anomaly, 'critical', anomaly.split('\n')[0]!, { fetched: ctx.counters.fetched });627 else await alert(ALERT_KINDS.connectorFailure, 'warn', (e.message ?? 'unknown error').split('\n')[0]!, { fetched: ctx.counters.fetched });628 return { runId: ctx.runId, status: 'failed', counters: ctx.counters, anomaly };629 } finally {630 if (handleSignals) for (const s of SIGNALS) process.off(s, onSignal);631 }632}633