/**
 * Restartability / alerting tests against a real database (the SDK bookkeeping is SQL).
 * Gated: they run only when CI_TEST_DATABASE_URL (or DATABASE_URL) is set, so `pnpm -r test`
 * without Postgres still passes. Point it at a development database that has the schema pushed
 * (system_alerts included) — never at production:
 *
 *   CI_TEST_DATABASE_URL=postgres://localhost:5432/cancerindex_a pnpm --filter @cancerindex/connectors test
 */
import { mkdtempSync, rmSync } from 'node:fs';
import os from 'node:os';
import path from 'node:path';
import { sql } from 'drizzle-orm';
import { afterAll, beforeAll, describe, expect, it } from 'vitest';
import { defineManifest } from './manifest.js';
import { AnomalyError, Connector, RunContext, runConnector, type ConnectorHealth } from './run.js';

const TEST_URL = process.env.CI_TEST_DATABASE_URL ?? process.env.DATABASE_URL;
const ID = 'zz-fake-paged';
const SOURCE_ID = 'CI-SOURCE-99999901';
const tmp = mkdtempSync(path.join(os.tmpdir(), 'ci-run-'));

type Db = import('@cancerindex/database').Database;
let db: Db | null = null;
let closeDb: (() => Promise<void>) | null = null;

const manifest = defineManifest({
  id: ID,
  name: 'Fake paged source (tests)',
  organization: 'CancerIndex tests',
  category: 'terminology',
  tier: 11,
  description: 'In-memory paged source used to test cursor checkpoints, abort and anomaly guard.',
  homepage: 'https://example.invalid/',
  access: { type: 'api', auth: 'none' },
  license: 'CC0',
  licenseStatus: 'approved',
  commercialUse: 'allowed',
  redistribution: 'allowed',
  updateFrequency: 'test',
  supportsIncrementalSync: true,
  entities: ['record'],
  rateLimits: { requestsPerSecond: 1000 },
  documentationVerifiedAt: '2026-09-08',
  status: 'active',
  checkpointEvery: 5,
  anomalyGuard: { minRatioOfPrevious: 0.5 },
});

interface Hooks {
  pages: number;
  pageSize: number;
  /** Called after each page is persisted and the cursor advanced (before the next page). */
  afterPage?: (ctx: RunContext, page: number) => Promise<void> | void;
  /** Declared total handed to ctx.guardCount before the first page (undefined = no guard). */
  declaredTotal?: number;
}

/** ClinVar/ClinicalTrials-like connector: one cursor advance per page, checks shouldStop() between pages. */
class FakePagedConnector extends Connector {
  readonly manifest = manifest;
  constructor(private readonly hooks: Hooks) {
    super();
  }
  async healthCheck(): Promise<ConnectorHealth> {
    return { status: 'healthy' };
  }
  async sync(ctx: RunContext): Promise<void> {
    const cursor = ctx.cursor as { page?: number; done?: boolean };
    if (this.hooks.declaredTotal !== undefined) await ctx.guardCount('record', this.hooks.declaredTotal);
    let page = cursor.page ?? 0;
    while (page < this.hooks.pages) {
      if (ctx.shouldStop()) return;
      for (let i = 0; i < this.hooks.pageSize; i++) {
        const n = page * this.hooks.pageSize + i;
        await ctx.upsertSourceRecord('record', `REC-${n}`, { n, page, v: 1 });
      }
      page++;
      cursor.page = page; // safe point: everything before is persisted
      await this.hooks.afterPage?.(ctx, page);
    }
    cursor.done = true;
  }
}

async function cursorInDb(): Promise<Record<string, unknown> | null> {
  const [row] = await db!.execute<{ cursor: Record<string, unknown> }>(sql`SELECT cursor FROM connector_cursors WHERE connector_id = ${ID}`);
  return row?.cursor ?? null;
}
async function run(id: string) {
  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}`);
  return row!;
}
async function alerts(kind?: string) {
  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`);
}
async function cleanup() {
  if (!db) return;
  await db.execute(sql`DELETE FROM ingest_runs WHERE connector_id = ${ID}`);
  await db.execute(sql`DELETE FROM connector_cursors WHERE connector_id = ${ID}`);
  await db.execute(sql`DELETE FROM connector_field_stats WHERE connector_id = ${ID}`);
  await db.execute(sql`DELETE FROM system_alerts WHERE connector_id = ${ID}`);
  await db.execute(sql`DELETE FROM source_records WHERE source_id = ${SOURCE_ID}`);
  await db.execute(sql`DELETE FROM provenance WHERE source_id = ${SOURCE_ID}`);
}

/** Emit a real signal event on `process`, delivered only to listeners added since `baseline` (i.e. the run's handler). */
function emitSignal(signal: NodeJS.Signals, baseline: Set<unknown>) {
  const others = process.rawListeners(signal).filter((l) => baseline.has(l));
  for (const l of others) process.off(signal, l as never);
  try {
    process.emit(signal, signal);
  } finally {
    for (const l of others) process.on(signal, l as never);
  }
}

beforeAll(async () => {
  if (!TEST_URL) return;
  process.env.CI_DATA_DIR = tmp;
  const mod = await import('@cancerindex/database');
  db = mod.getDb({ url: TEST_URL, max: 4 });
  closeDb = mod.closeDb;
  await db.execute(sql`SELECT 1`);
  await db.execute(sql`SELECT 1 FROM system_alerts LIMIT 1`); // ops schema must be pushed
  await db.execute(sql`INSERT INTO sources (id, slug, name, category, access_type, access_auth, license_status, status, manifest)
    VALUES (${SOURCE_ID}, ${ID}, 'Fake paged source (tests)', 'terminology', 'api', 'none', 'approved', 'active', '{}'::jsonb)
    ON CONFLICT (id) DO UPDATE SET slug = EXCLUDED.slug`);
  await cleanup();
}, 30_000);

afterAll(async () => {
  await cleanup();
  if (db) await db.execute(sql`DELETE FROM sources WHERE id = ${SOURCE_ID}`);
  await closeDb?.();
  rmSync(tmp, { recursive: true, force: true });
});

describe.skipIf(!TEST_URL)('runConnector restartability (database-backed)', () => {
  it('checkpoints the cursor mid-run every checkpointEvery records when it changed', async () => {
    const seen: Array<{ page: number; saved: unknown }> = [];
    const c = new FakePagedConnector({
      pages: 4,
      pageSize: 5, // = checkpointEvery → after each page's 5th record the cursor of the *previous* page is saved
      afterPage: async (_ctx, page) => {
        seen.push({ page, saved: (await cursorInDb())?.page ?? null });
      },
    });
    const r = await runConnector(db!, c, { maxMinutes: 5, handleSignals: false });
    expect(r.status).toBe('succeeded');
    expect(r.counters.fetched).toBe(20);
    // Page k+1's checkpoint (fires on its 5th record) persisted cursor {page: k}; page 1 has nothing changed yet.
    expect(seen.map((s) => s.saved)).toEqual([null, 1, 2, 3]);
    expect(await cursorInDb()).toMatchObject({ page: 4, done: true });
    const row = await run(r.runId);
    expect(row.status).toBe('succeeded');
    expect(row.cursor_after).toMatchObject({ page: 4, done: true });
    expect((await alerts()).filter((a) => a.status !== 'resolved')).toEqual([]);
  });

  it('saveCursor() persists immediately and dry runs never touch the cursor', async () => {
    let inside: unknown = null;
    const c = new FakePagedConnector({
      pages: 2,
      pageSize: 2,
      afterPage: async (ctx, page) => {
        if (page === 1) {
          await ctx.saveCursor();
          inside = (await cursorInDb())?.page;
        }
      },
    });
    const before = await cursorInDb();
    expect(before).toMatchObject({ page: 4 }); // left by the previous test
    const dry = await runConnector(db!, c, { maxMinutes: 5, handleSignals: false, mode: 'dry_run', resetCursor: true });
    expect(dry.status).toBe('succeeded');
    expect(inside).toBe(4); // dry_run: saveCursor is a no-op, the stored cursor is still the previous run's
    expect(await cursorInDb()).toEqual(before);
    const real = await runConnector(db!, c, { maxMinutes: 5, handleSignals: false, resetCursor: true });
    expect(real.status).toBe('succeeded');
    expect(inside).toBe(1);
  });

  it('SIGTERM mid-run: run marked aborted, cursor kept at the last completed page, next run resumes there', async () => {
    let listenersDuring = 0;
    const baselineListeners = new Set(process.rawListeners('SIGTERM'));
    const baseline = baselineListeners.size;
    const c = new FakePagedConnector({
      pages: 6,
      pageSize: 3,
      afterPage: (ctx, page) => {
        if (page === 3) {
          listenersDuring = process.listenerCount('SIGTERM');
          emitSignal('SIGTERM', baselineListeners); // deliver only to the run's handler
          expect(ctx.abortReason).toBe('SIGTERM received');
          expect(ctx.shouldStop()).toBe(true);
        }
      },
    });
    const r = await runConnector(db!, c, { maxMinutes: 5, resetCursor: true });
    expect(r.status).toBe('aborted');
    expect(r.counters.fetched).toBe(9); // 3 pages × 3 records, nothing after the signal
    expect(listenersDuring).toBe(baseline + 1);
    expect(process.listenerCount('SIGTERM')).toBe(baseline); // handlers removed at the end of the run
    const row = await run(r.runId);
    expect(row.status).toBe('aborted');
    expect(row.error).toContain('SIGTERM');
    expect(row.cursor_after).toMatchObject({ page: 3 });
    expect(await cursorInDb()).toMatchObject({ page: 3 });
    const aborted = await alerts('connector_aborted');
    expect(aborted).toHaveLength(1);
    expect(aborted[0]).toMatchObject({ severity: 'info', status: 'open' });

    // Resume: starts from page 3, finishes the remaining 3 pages, success resolves the alert.
    const c2 = new FakePagedConnector({ pages: 6, pageSize: 3 });
    const r2 = await runConnector(db!, c2, { maxMinutes: 5, handleSignals: false });
    expect(r2.status).toBe('succeeded');
    expect(r2.counters.fetched).toBe(9);
    const row2 = await run(r2.runId);
    expect(row2.cursor_before).toMatchObject({ page: 3 });
    expect(row2.cursor_after).toMatchObject({ page: 6, done: true });
    expect((await alerts('connector_aborted'))[0]!.status).toBe('resolved');
    // 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.
    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`);
    expect(touched.map((t) => [t.run, Number(t.n), Number(t.distinct_ids)])).toEqual([
      [r.runId, 9, 9],
      [r2.runId, 9, 9],
    ]);
  });

  it('guardCount() refuses a shrunken total: run failed with anomaly, critical alert, previous data untouched', async () => {
    // Baseline: a successful run that fetched 20 records.
    const ok = await runConnector(db!, new FakePagedConnector({ pages: 4, pageSize: 5 }), { maxMinutes: 5, handleSignals: false, resetCursor: true });
    expect(ok.status).toBe('succeeded');
    expect((await run(ok.runId)).records_fetched).toBe(20);

    const bad = await runConnector(db!, new FakePagedConnector({ pages: 1, pageSize: 5, declaredTotal: 5 }), { maxMinutes: 5, handleSignals: false, resetCursor: true });
    expect(bad.status).toBe('failed');
    expect(bad.anomaly).toMatch(/^anomaly: record count 5 is 25\.0% of the previous successful run/);
    expect(bad.counters.fetched).toBe(0); // refused before fetching anything
    const row = await run(bad.runId);
    expect(row.status).toBe('failed');
    expect(row.anomaly).toContain('CLAUDE.md §171');
    const crit = await alerts('anomaly');
    expect(crit).toHaveLength(1);
    expect(crit[0]).toMatchObject({ severity: 'critical', status: 'open', count: 1 });
    const active = await db!.execute<{ n: string }>(sql`SELECT count(*)::text AS n FROM source_records WHERE source_id = ${SOURCE_ID} AND status = 'active'`);
    expect(Number(active[0]!.n)).toBe(20);

    // Same anomaly again → deduplicated (count 2); an acceptable total passes and resolves it.
    const again = await runConnector(db!, new FakePagedConnector({ pages: 1, pageSize: 5, declaredTotal: 5 }), { maxMinutes: 5, handleSignals: false, resetCursor: true });
    expect(again.status).toBe('failed');
    expect((await alerts('anomaly'))[0]!.count).toBe(2);
    const fine = await runConnector(db!, new FakePagedConnector({ pages: 4, pageSize: 5, declaredTotal: 18 }), { maxMinutes: 5, handleSignals: false, resetCursor: true });
    expect(fine.status).toBe('succeeded');
    expect((await alerts('anomaly'))[0]!.status).toBe('resolved');
  });

  it('guardCount() throws AnomalyError directly when called on a bare context', async () => {
    const ctx = new RunContext(db!, manifest, SOURCE_ID, 'incremental', { maxMinutes: 1 }, 99);
    await expect(ctx.guardCount('record', 1)).rejects.toBeInstanceOf(AnomalyError);
    expect(ctx.anomaly).toContain('anomaly: record count 1');
    const okRes = await ctx.guardCount('record', 15, { minRatio: 0.7 });
    expect(okRes.previous).toBe(20);
    expect(okRes.ratio).toBeCloseTo(0.75);
  });

  it('cix doctor report lists the connectors and reports no hard failure on a healthy dev database', async () => {
    const { runDoctor, formatDoctorReport } = await import('../ops/doctor.js');
    const report = await runDoctor(db!, { skipDisk: true });
    expect(report.checks.find((c) => c.name === 'reachable')?.level).toBe('ok');
    expect(report.checks.find((c) => c.name === 'ops schema')?.level).toBe('ok');
    expect(report.connectors.length).toBeGreaterThan(5);
    expect(report.tables.find((t) => t.table === 'system_alerts')).toBeTruthy();
    const text = formatDoctorReport(report);
    expect(text).toContain('## connectors');
    expect(text).toMatch(/RESULT: (ready|\d+ hard failure)/);
  });
});