spb/cancerindex
Public
TypeScript 97.2%
SQL 1.5%
CSS 0.6%
JavaScript 0.5%
1/**2 * Restartability / alerting tests against a real database (the SDK bookkeeping is SQL).3 * Gated: they run only when CI_TEST_DATABASE_URL (or DATABASE_URL) is set, so `pnpm -r test`4 * without Postgres still passes. Point it at a development database that has the schema pushed5 * (system_alerts included) — never at production:6 *7 * CI_TEST_DATABASE_URL=postgres://localhost:5432/cancerindex_a pnpm --filter @cancerindex/connectors test8 */9import { mkdtempSync, rmSync } from 'node:fs';10import os from 'node:os';11import path from 'node:path';12import { sql } from 'drizzle-orm';13import { afterAll, beforeAll, describe, expect, it } from 'vitest';14import { defineManifest } from './manifest.js';15import { AnomalyError, Connector, RunContext, runConnector, type ConnectorHealth } from './run.js';1617const TEST_URL = process.env.CI_TEST_DATABASE_URL ?? process.env.DATABASE_URL;18const ID = 'zz-fake-paged';19const SOURCE_ID = 'CI-SOURCE-99999901';20const tmp = mkdtempSync(path.join(os.tmpdir(), 'ci-run-'));2122type Db = import('@cancerindex/database').Database;23let db: Db | null = null;24let closeDb: (() => Promise<void>) | null = null;2526const manifest = defineManifest({27 id: ID,28 name: 'Fake paged source (tests)',29 organization: 'CancerIndex tests',30 category: 'terminology',31 tier: 11,32 description: 'In-memory paged source used to test cursor checkpoints, abort and anomaly guard.',33 homepage: 'https://example.invalid/',34 access: { type: 'api', auth: 'none' },35 license: 'CC0',36 licenseStatus: 'approved',37 commercialUse: 'allowed',38 redistribution: 'allowed',39 updateFrequency: 'test',40 supportsIncrementalSync: true,41 entities: ['record'],42 rateLimits: { requestsPerSecond: 1000 },43 documentationVerifiedAt: '2026-09-08',44 status: 'active',45 checkpointEvery: 5,46 anomalyGuard: { minRatioOfPrevious: 0.5 },47});4849interface Hooks {50 pages: number;51 pageSize: number;52 /** Called after each page is persisted and the cursor advanced (before the next page). */53 afterPage?: (ctx: RunContext, page: number) => Promise<void> | void;54 /** Declared total handed to ctx.guardCount before the first page (undefined = no guard). */55 declaredTotal?: number;56}5758/** ClinVar/ClinicalTrials-like connector: one cursor advance per page, checks shouldStop() between pages. */59class FakePagedConnector extends Connector {60 readonly manifest = manifest;61 constructor(private readonly hooks: Hooks) {62 super();63 }64 async healthCheck(): Promise<ConnectorHealth> {65 return { status: 'healthy' };66 }67 async sync(ctx: RunContext): Promise<void> {68 const cursor = ctx.cursor as { page?: number; done?: boolean };69 if (this.hooks.declaredTotal !== undefined) await ctx.guardCount('record', this.hooks.declaredTotal);70 let page = cursor.page ?? 0;71 while (page < this.hooks.pages) {72 if (ctx.shouldStop()) return;73 for (let i = 0; i < this.hooks.pageSize; i++) {74 const n = page * this.hooks.pageSize + i;75 await ctx.upsertSourceRecord('record', `REC-${n}`, { n, page, v: 1 });76 }77 page++;78 cursor.page = page; // safe point: everything before is persisted79 await this.hooks.afterPage?.(ctx, page);80 }81 cursor.done = true;82 }83}8485async function cursorInDb(): Promise<Record<string, unknown> | null> {86 const [row] = await db!.execute<{ cursor: Record<string, unknown> }>(sql`SELECT cursor FROM connector_cursors WHERE connector_id = ${ID}`);87 return row?.cursor ?? null;88}89async function run(id: string) {90 const [row] = await db!.execute<{ status: string; anomaly: string | null; error: string | null; cursor_before: unknown; cursor_after: unknown; records_fetched: number }>(sql`SELECT status, anomaly, error, cursor_before, cursor_after, records_fetched FROM ingest_runs WHERE id = ${id}`);91 return row!;92}93async function alerts(kind?: string) {94 return db!.execute<{ kind: string; severity: string; status: string; count: number; message: string }>(sql`SELECT kind, severity, status, count, message FROM system_alerts WHERE connector_id = ${ID} ${kind ? sql`AND kind = ${kind}` : sql``} ORDER BY id`);95}96async function cleanup() {97 if (!db) return;98 await db.execute(sql`DELETE FROM ingest_runs WHERE connector_id = ${ID}`);99 await db.execute(sql`DELETE FROM connector_cursors WHERE connector_id = ${ID}`);100 await db.execute(sql`DELETE FROM connector_field_stats WHERE connector_id = ${ID}`);101 await db.execute(sql`DELETE FROM system_alerts WHERE connector_id = ${ID}`);102 await db.execute(sql`DELETE FROM source_records WHERE source_id = ${SOURCE_ID}`);103 await db.execute(sql`DELETE FROM provenance WHERE source_id = ${SOURCE_ID}`);104}105106/** Emit a real signal event on `process`, delivered only to listeners added since `baseline` (i.e. the run's handler). */107function emitSignal(signal: NodeJS.Signals, baseline: Set<unknown>) {108 const others = process.rawListeners(signal).filter((l) => baseline.has(l));109 for (const l of others) process.off(signal, l as never);110 try {111 process.emit(signal, signal);112 } finally {113 for (const l of others) process.on(signal, l as never);114 }115}116117beforeAll(async () => {118 if (!TEST_URL) return;119 process.env.CI_DATA_DIR = tmp;120 const mod = await import('@cancerindex/database');121 db = mod.getDb({ url: TEST_URL, max: 4 });122 closeDb = mod.closeDb;123 await db.execute(sql`SELECT 1`);124 await db.execute(sql`SELECT 1 FROM system_alerts LIMIT 1`); // ops schema must be pushed125 await db.execute(sql`INSERT INTO sources (id, slug, name, category, access_type, access_auth, license_status, status, manifest)126 VALUES (${SOURCE_ID}, ${ID}, 'Fake paged source (tests)', 'terminology', 'api', 'none', 'approved', 'active', '{}'::jsonb)127 ON CONFLICT (id) DO UPDATE SET slug = EXCLUDED.slug`);128 await cleanup();129}, 30_000);130131afterAll(async () => {132 await cleanup();133 if (db) await db.execute(sql`DELETE FROM sources WHERE id = ${SOURCE_ID}`);134 await closeDb?.();135 rmSync(tmp, { recursive: true, force: true });136});137138describe.skipIf(!TEST_URL)('runConnector restartability (database-backed)', () => {139 it('checkpoints the cursor mid-run every checkpointEvery records when it changed', async () => {140 const seen: Array<{ page: number; saved: unknown }> = [];141 const c = new FakePagedConnector({142 pages: 4,143 pageSize: 5, // = checkpointEvery → after each page's 5th record the cursor of the *previous* page is saved144 afterPage: async (_ctx, page) => {145 seen.push({ page, saved: (await cursorInDb())?.page ?? null });146 },147 });148 const r = await runConnector(db!, c, { maxMinutes: 5, handleSignals: false });149 expect(r.status).toBe('succeeded');150 expect(r.counters.fetched).toBe(20);151 // Page k+1's checkpoint (fires on its 5th record) persisted cursor {page: k}; page 1 has nothing changed yet.152 expect(seen.map((s) => s.saved)).toEqual([null, 1, 2, 3]);153 expect(await cursorInDb()).toMatchObject({ page: 4, done: true });154 const row = await run(r.runId);155 expect(row.status).toBe('succeeded');156 expect(row.cursor_after).toMatchObject({ page: 4, done: true });157 expect((await alerts()).filter((a) => a.status !== 'resolved')).toEqual([]);158 });159160 it('saveCursor() persists immediately and dry runs never touch the cursor', async () => {161 let inside: unknown = null;162 const c = new FakePagedConnector({163 pages: 2,164 pageSize: 2,165 afterPage: async (ctx, page) => {166 if (page === 1) {167 await ctx.saveCursor();168 inside = (await cursorInDb())?.page;169 }170 },171 });172 const before = await cursorInDb();173 expect(before).toMatchObject({ page: 4 }); // left by the previous test174 const dry = await runConnector(db!, c, { maxMinutes: 5, handleSignals: false, mode: 'dry_run', resetCursor: true });175 expect(dry.status).toBe('succeeded');176 expect(inside).toBe(4); // dry_run: saveCursor is a no-op, the stored cursor is still the previous run's177 expect(await cursorInDb()).toEqual(before);178 const real = await runConnector(db!, c, { maxMinutes: 5, handleSignals: false, resetCursor: true });179 expect(real.status).toBe('succeeded');180 expect(inside).toBe(1);181 });182183 it('SIGTERM mid-run: run marked aborted, cursor kept at the last completed page, next run resumes there', async () => {184 let listenersDuring = 0;185 const baselineListeners = new Set(process.rawListeners('SIGTERM'));186 const baseline = baselineListeners.size;187 const c = new FakePagedConnector({188 pages: 6,189 pageSize: 3,190 afterPage: (ctx, page) => {191 if (page === 3) {192 listenersDuring = process.listenerCount('SIGTERM');193 emitSignal('SIGTERM', baselineListeners); // deliver only to the run's handler194 expect(ctx.abortReason).toBe('SIGTERM received');195 expect(ctx.shouldStop()).toBe(true);196 }197 },198 });199 const r = await runConnector(db!, c, { maxMinutes: 5, resetCursor: true });200 expect(r.status).toBe('aborted');201 expect(r.counters.fetched).toBe(9); // 3 pages × 3 records, nothing after the signal202 expect(listenersDuring).toBe(baseline + 1);203 expect(process.listenerCount('SIGTERM')).toBe(baseline); // handlers removed at the end of the run204 const row = await run(r.runId);205 expect(row.status).toBe('aborted');206 expect(row.error).toContain('SIGTERM');207 expect(row.cursor_after).toMatchObject({ page: 3 });208 expect(await cursorInDb()).toMatchObject({ page: 3 });209 const aborted = await alerts('connector_aborted');210 expect(aborted).toHaveLength(1);211 expect(aborted[0]).toMatchObject({ severity: 'info', status: 'open' });212213 // Resume: starts from page 3, finishes the remaining 3 pages, success resolves the alert.214 const c2 = new FakePagedConnector({ pages: 6, pageSize: 3 });215 const r2 = await runConnector(db!, c2, { maxMinutes: 5, handleSignals: false });216 expect(r2.status).toBe('succeeded');217 expect(r2.counters.fetched).toBe(9);218 const row2 = await run(r2.runId);219 expect(row2.cursor_before).toMatchObject({ page: 3 });220 expect(row2.cursor_after).toMatchObject({ page: 6, done: true });221 expect((await alerts('connector_aborted'))[0]!.status).toBe('resolved');222 // Idempotent hand-over: the 9 records of the aborted run were not re-fetched, the 9 of the resume are new to it, no duplicates.223 const touched = await db!.execute<{ run: string; n: string; distinct_ids: string }>(sql`SELECT last_seen_run AS run, count(*)::text AS n, count(DISTINCT source_record_id)::text AS distinct_ids FROM source_records WHERE source_id = ${SOURCE_ID} AND last_seen_run IN (${r.runId}, ${r2.runId}) GROUP BY 1 ORDER BY 1`);224 expect(touched.map((t) => [t.run, Number(t.n), Number(t.distinct_ids)])).toEqual([225 [r.runId, 9, 9],226 [r2.runId, 9, 9],227 ]);228 });229230 it('guardCount() refuses a shrunken total: run failed with anomaly, critical alert, previous data untouched', async () => {231 // Baseline: a successful run that fetched 20 records.232 const ok = await runConnector(db!, new FakePagedConnector({ pages: 4, pageSize: 5 }), { maxMinutes: 5, handleSignals: false, resetCursor: true });233 expect(ok.status).toBe('succeeded');234 expect((await run(ok.runId)).records_fetched).toBe(20);235236 const bad = await runConnector(db!, new FakePagedConnector({ pages: 1, pageSize: 5, declaredTotal: 5 }), { maxMinutes: 5, handleSignals: false, resetCursor: true });237 expect(bad.status).toBe('failed');238 expect(bad.anomaly).toMatch(/^anomaly: record count 5 is 25\.0% of the previous successful run/);239 expect(bad.counters.fetched).toBe(0); // refused before fetching anything240 const row = await run(bad.runId);241 expect(row.status).toBe('failed');242 expect(row.anomaly).toContain('CLAUDE.md §171');243 const crit = await alerts('anomaly');244 expect(crit).toHaveLength(1);245 expect(crit[0]).toMatchObject({ severity: 'critical', status: 'open', count: 1 });246 const active = await db!.execute<{ n: string }>(sql`SELECT count(*)::text AS n FROM source_records WHERE source_id = ${SOURCE_ID} AND status = 'active'`);247 expect(Number(active[0]!.n)).toBe(20);248249 // Same anomaly again → deduplicated (count 2); an acceptable total passes and resolves it.250 const again = await runConnector(db!, new FakePagedConnector({ pages: 1, pageSize: 5, declaredTotal: 5 }), { maxMinutes: 5, handleSignals: false, resetCursor: true });251 expect(again.status).toBe('failed');252 expect((await alerts('anomaly'))[0]!.count).toBe(2);253 const fine = await runConnector(db!, new FakePagedConnector({ pages: 4, pageSize: 5, declaredTotal: 18 }), { maxMinutes: 5, handleSignals: false, resetCursor: true });254 expect(fine.status).toBe('succeeded');255 expect((await alerts('anomaly'))[0]!.status).toBe('resolved');256 });257258 it('guardCount() throws AnomalyError directly when called on a bare context', async () => {259 const ctx = new RunContext(db!, manifest, SOURCE_ID, 'incremental', { maxMinutes: 1 }, 99);260 await expect(ctx.guardCount('record', 1)).rejects.toBeInstanceOf(AnomalyError);261 expect(ctx.anomaly).toContain('anomaly: record count 1');262 const okRes = await ctx.guardCount('record', 15, { minRatio: 0.7 });263 expect(okRes.previous).toBe(20);264 expect(okRes.ratio).toBeCloseTo(0.75);265 });266267 it('cix doctor report lists the connectors and reports no hard failure on a healthy dev database', async () => {268 const { runDoctor, formatDoctorReport } = await import('../ops/doctor.js');269 const report = await runDoctor(db!, { skipDisk: true });270 expect(report.checks.find((c) => c.name === 'reachable')?.level).toBe('ok');271 expect(report.checks.find((c) => c.name === 'ops schema')?.level).toBe('ok');272 expect(report.connectors.length).toBeGreaterThan(5);273 expect(report.tables.find((t) => t.table === 'system_alerts')).toBeTruthy();274 const text = formatDoctorReport(report);275 expect(text).toContain('## connectors');276 expect(text).toMatch(/RESULT: (ready|\d+ hard failure)/);277 });278});279