import { and, desc, eq, gt, inArray, ne, notInArray, sql } from 'drizzle-orm';
import { createHash } from 'node:crypto';
import { formatRunId, logger, type Logger, type ProvenanceInput, envInt } from '@cancerindex/shared';
import { type Database, ingestRuns, connectorCursors, sourceRecords, provenance, connectorFieldStats, sources, unresolvedLabels, changeEvents, raiseAlertSafe, resolveAlerts } from '@cancerindex/database';
import type { ConnectorManifest } from './manifest.js';
import { HttpClient } from './http.js';
import { RawLake } from './lake.js';

export type RunMode = 'full' | 'incremental' | 'backfill' | 'dry_run' | 'probe';

export interface RunLimits {
  /** Stop fetching new pages after this many minutes; cursor is saved so the next run resumes. */
  maxMinutes: number;
  /** Optional cap on records for smoke runs (CLAUDE.md §227: prove on 10/100/1000 first). */
  maxRecords?: number;
}

export interface UpsertResult {
  id: number;
  status: 'created' | 'updated' | 'unchanged';
}

export interface UpsertOptions {
  sourceUpdatedAt?: Date | null;
  canonicalType?: string;
  canonicalId?: string;
}

export interface BatchUpsertItem extends UpsertOptions {
  sourceRecordId: string;
  payload: unknown;
}

export interface FieldStatsCollector {
  observe(entity: string, obj: unknown): void;
}

/** Thrown by `ctx.guardCount` (CLAUDE.md §171). `runConnector` records it in `ingest_runs.anomaly`. */
export class AnomalyError extends Error {
  constructor(
    message: string,
    readonly entity: string,
    readonly fetched: number,
    readonly previous: number,
    readonly minRatio: number,
  ) {
    super(message);
    this.name = 'AnomalyError';
  }
}

/** Kinds of `system_alerts` rows raised by the SDK / worker (CLAUDE.md §170). */
export const ALERT_KINDS = {
  connectorFailure: 'connector_failure',
  connectorAborted: 'connector_aborted',
  anomaly: 'anomaly',
  schemaDrift: 'schema_drift',
  sourceFailing: 'source_failing',
  sourceStale: 'source_stale',
} as const;

/** Alerts a successful run clears for its connector. */
const RESOLVED_ON_SUCCESS = [ALERT_KINDS.connectorFailure, ALERT_KINDS.connectorAborted, ALERT_KINDS.anomaly, ALERT_KINDS.sourceFailing, ALERT_KINDS.sourceStale];

const CHECKPOINT_INTERVAL_MS = 60_000;

/**
 * Ingest run context: identifiers, counters, raw lake, HTTP client, provenance/record helpers.
 * Every connector receives one per execution (CLAUDE.md §90, §175-176).
 *
 * Restartability: `cursor` is persisted (a) explicitly with `saveCursor()`, (b) automatically every
 * `manifest.checkpointEvery` upserted records or 60 s when it changed, (c) on SIGTERM/SIGINT
 * (`requestAbort`), (d) at the end of the run whatever its status. A connector must only mutate
 * `ctx.cursor` at a point where everything before it has been persisted (end of a page / batch).
 */
export class RunContext {
  readonly runId: string;
  readonly startedAt = new Date();
  readonly http: HttpClient;
  readonly lake: RawLake;
  readonly log: Logger;
  readonly counters = { fetched: 0, created: 0, updated: 0, unchanged: 0, rejected: 0, validationFailures: 0 };
  readonly drift: Array<{ entity: string; field: string; kind: 'new_field' | 'type_change' | 'nullability'; detail: string }> = [];
  private logs: Array<{ t: string; level: string; msg: string }> = [];
  private fieldStats = new Map<string, { types: Set<string>; seen: number; nulls: number }>();
  private knownFields = new Map<string, Set<string>>();
  private knownTypes = new Map<string, Set<string>>();
  private deadline: number;
  private timeBudgetHit = false;
  private lastSavedCursor: string | null = null;
  private recordsSinceCheckpoint = 0;
  private lastCheckpointAt = Date.now();
  private savingCursor: Promise<void> | null = null;
  private abortPromise: Promise<void> | null = null;
  /** Reason of the abort request (signal name…), null while running normally. */
  abortReason: string | null = null;
  /** Set by `guardCount` (or by runConnector for `anomaly:` errors); copied to `ingest_runs.anomaly`. */
  anomaly: string | null = null;
  /** Number of cursor checkpoints written during this run (diagnostics / tests). */
  checkpoints = 0;
  cursor: Record<string, unknown> = {};
  datasetVersion: string | undefined;

  constructor(
    readonly db: Database,
    readonly manifest: ConnectorManifest,
    readonly sourceId: string,
    readonly mode: RunMode,
    readonly limits: RunLimits,
    seq: number,
  ) {
    this.runId = formatRunId(manifest.id, this.startedAt, seq);
    this.http = new HttpClient(manifest);
    this.lake = new RawLake(manifest.id, this.runId, this.startedAt);
    this.log = logger.child({ run: this.runId, connector: manifest.id });
    this.deadline = this.startedAt.getTime() + limits.maxMinutes * 60_000;
  }

  /** True when the time budget, the record cap or an abort request is reached; connectors check it between pages. */
  shouldStop(): boolean {
    if (this.abortReason) return true;
    if (Date.now() > this.deadline) {
      if (!this.timeBudgetHit) this.info('time budget reached — cursor will be saved for the next run');
      this.timeBudgetHit = true;
      return true;
    }
    if (this.limits.maxRecords !== undefined && this.counters.fetched >= this.limits.maxRecords) return true;
    return false;
  }
  get stoppedEarly(): boolean {
    return this.timeBudgetHit || (this.limits.maxRecords !== undefined && this.counters.fetched >= this.limits.maxRecords);
  }
  /** Cursor writes are skipped in these modes (a dry run / probe must leave the real cursor untouched). */
  get persistsCursor(): boolean {
    return this.mode !== 'dry_run' && this.mode !== 'probe';
  }

  info(msg: string, extra?: Record<string, unknown>) {
    this.log.info(extra ?? {}, msg);
    this.push('info', msg);
  }
  warn(msg: string, extra?: Record<string, unknown>) {
    this.log.warn(extra ?? {}, msg);
    this.push('warn', msg);
  }
  error(msg: string, extra?: Record<string, unknown>) {
    this.log.error(extra ?? {}, msg);
    this.push('error', msg);
  }
  private push(level: string, msg: string) {
    if (this.logs.length < 500) this.logs.push({ t: new Date().toISOString(), level, msg });
  }

  /* ------------------------------------------------------------------------------------------ */
  /* Cursor persistence (CLAUDE.md §90)                                                          */
  /* ------------------------------------------------------------------------------------------ */

  /**
   * Persist `cursor` to `connector_cursors` (and `ingest_runs.cursor_after`) right now. Idempotent
   * and serialised: concurrent calls share one write. No-op in dry_run / probe modes.
   */
  async saveCursor(): Promise<void> {
    if (!this.persistsCursor) return;
    if (this.savingCursor) await this.savingCursor;
    const json = JSON.stringify(this.cursor ?? {});
    if (json === this.lastSavedCursor) return;
    this.savingCursor = (async () => {
      const now = new Date();
      await this.db
        .insert(connectorCursors)
        .values({ connectorId: this.manifest.id, cursor: this.cursor, lastAttemptAt: now, updatedAt: now })
        .onConflictDoUpdate({ target: connectorCursors.connectorId, set: { cursor: this.cursor, updatedAt: now } });
      await this.db.update(ingestRuns).set({ cursorAfter: this.cursor }).where(eq(ingestRuns.id, this.runId));
      this.lastSavedCursor = json;
      this.checkpoints++;
    })();
    try {
      await this.savingCursor;
    } finally {
      this.savingCursor = null;
    }
    this.recordsSinceCheckpoint = 0;
    this.lastCheckpointAt = Date.now();
  }

  /**
   * Count `records` persisted records towards the automatic checkpoint and save the cursor when the
   * threshold (`manifest.checkpointEvery` records or 60 s) is reached and the cursor changed.
   * Called by `upsertSourceRecord` / `upsertSourceRecordsBatch`; connectors with their own batch
   * writers call it after each batch.
   */
  async checkpoint(records = 1): Promise<void> {
    if (!this.persistsCursor) return;
    this.recordsSinceCheckpoint += records;
    const due = this.recordsSinceCheckpoint >= this.manifest.checkpointEvery || Date.now() - this.lastCheckpointAt >= CHECKPOINT_INTERVAL_MS;
    if (!due) return;
    this.recordsSinceCheckpoint = 0;
    this.lastCheckpointAt = Date.now();
    await this.saveCursor();
  }

  /**
   * Abort request (SIGTERM/SIGINT, operator): makes `shouldStop()` true, persists the cursor and
   * provisionally marks the run `aborted` so the state is right even if the process is killed before
   * the connector returns. `runConnector` finalises the run when `sync()` comes back.
   */
  requestAbort(reason: string): Promise<void> {
    if (this.abortPromise) return this.abortPromise;
    this.abortReason = reason;
    this.warn(`abort requested (${reason}) — saving cursor, run marked aborted; the connector stops at its next shouldStop() check`);
    this.abortPromise = (async () => {
      try {
        await this.lake.flush();
        await this.saveCursor();
        if (this.persistsCursor || this.mode === 'dry_run') {
          await this.db
            .update(ingestRuns)
            .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 })
            .where(eq(ingestRuns.id, this.runId));
        }
      } catch (e) {
        this.log.error({ err: e }, 'abort bookkeeping failed');
      }
    })();
    return this.abortPromise;
  }

  /* ------------------------------------------------------------------------------------------ */
  /* Anomaly guard (CLAUDE.md §171)                                                              */
  /* ------------------------------------------------------------------------------------------ */

  /**
   * Compare a total the source declares (or the number of records parsed) with the previous
   * successful run's `records_fetched`. Throws `AnomalyError` — and marks the run — when
   * `fetchedCount < previous × manifest.anomalyGuard.minRatioOfPrevious`. Never destructive: it only
   * refuses to continue, the previous data stays published. Dry runs and probes are excluded from
   * the baseline; the first run of a connector is accepted as the baseline.
   */
  async guardCount(entity: string, fetchedCount: number, opts: { minRatio?: number } = {}): Promise<{ previous: number | null; ratio: number | null }> {
    const minRatio = opts.minRatio ?? this.manifest.anomalyGuard.minRatioOfPrevious;
    const [prev] = await this.db
      .select({ id: ingestRuns.id, n: ingestRuns.recordsFetched })
      .from(ingestRuns)
      .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)))
      .orderBy(desc(ingestRuns.startedAt))
      .limit(1);
    if (!prev) {
      this.info(`anomaly guard: no previous successful run — ${entity} count ${fetchedCount} accepted as baseline`);
      return { previous: null, ratio: null };
    }
    const ratio = fetchedCount / prev.n;
    if (ratio < minRatio) {
      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)`;
      this.anomaly = msg;
      this.error(msg);
      throw new AnomalyError(msg, entity, fetchedCount, prev.n, minRatio);
    }
    this.info(`anomaly guard: ${entity} count ${fetchedCount} = ${(ratio * 100).toFixed(1)}% of previous run ${prev.id} (${prev.n}) ≥ ${minRatio}`);
    return { previous: prev.n, ratio };
  }

  /* ------------------------------------------------------------------------------------------ */
  /* Schema drift                                                                                */
  /* ------------------------------------------------------------------------------------------ */

  /** Record observed fields for schema-drift detection (CLAUDE.md §25). */
  observe(entity: string, obj: unknown): void {
    if (!obj || typeof obj !== 'object') return;
    const known = this.knownFields.get(entity);
    for (const [k, v] of Object.entries(obj as Record<string, unknown>)) {
      const key = `${entity} ${k}`;
      const type = v === null || v === undefined ? 'null' : Array.isArray(v) ? 'array' : typeof v;
      let fs = this.fieldStats.get(key);
      if (!fs) {
        fs = { types: new Set(), seen: 0, nulls: 0 };
        this.fieldStats.set(key, fs);
        if (known && !known.has(k)) this.drift.push({ entity, field: k, kind: 'new_field', detail: `type=${type}` });
      }
      fs.seen++;
      if (type === 'null') fs.nulls++;
      else fs.types.add(type);
    }
  }

  async loadKnownFields(): Promise<void> {
    const rows = await this.db.select({ entity: connectorFieldStats.entity, field: connectorFieldStats.field, types: connectorFieldStats.types }).from(connectorFieldStats).where(eq(connectorFieldStats.connectorId, this.manifest.id));
    for (const r of rows) {
      if (!this.knownFields.has(r.entity)) this.knownFields.set(r.entity, new Set());
      this.knownFields.get(r.entity)!.add(r.field);
    }
    this.knownTypes = new Map(rows.map((r) => [`${r.entity} ${r.field}`, new Set(r.types)]));
  }

  private async flushFieldStats(): Promise<void> {
    for (const [key, fs] of this.fieldStats) {
      const [entity, field] = key.split(' ') as [string, string];
      const prev = this.knownTypes.get(key);
      if (prev && prev.size > 0) {
        for (const t of fs.types) if (!prev.has(t)) this.drift.push({ entity, field, kind: 'type_change', detail: `${[...prev].join('|')} → ${t}` });
      }
      await this.db
        .insert(connectorFieldStats)
        .values({ connectorId: this.manifest.id, entity, field, types: [...fs.types], seenCount: fs.seen, nullCount: fs.nulls, firstSeenRun: this.runId, lastSeenRun: this.runId })
        .onConflictDoUpdate({
          target: [connectorFieldStats.connectorId, connectorFieldStats.entity, connectorFieldStats.field],
          set: {
            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[])`,
            seenCount: sql`${connectorFieldStats.seenCount} + ${fs.seen}`,
            nullCount: sql`${connectorFieldStats.nullCount} + ${fs.nulls}`,
            lastSeenRun: this.runId,
            updatedAt: new Date(),
          },
        });
    }
  }

  /* ------------------------------------------------------------------------------------------ */
  /* Records, provenance, curation queue                                                         */
  /* ------------------------------------------------------------------------------------------ */

  /** Upsert a source-native record (idempotency, CLAUDE.md §91). Writes the raw payload to the lake. */
  async upsertSourceRecord(entityKind: string, sourceRecordId: string, payload: unknown, opts: UpsertOptions = {}): Promise<UpsertResult> {
    const json = JSON.stringify(payload);
    const hash = createHash('sha256').update(json).digest('hex');
    this.counters.fetched++;
    this.observe(entityKind, payload);
    const existing = await this.db
      .select({ id: sourceRecords.id, payloadHash: sourceRecords.payloadHash })
      .from(sourceRecords)
      .where(and(eq(sourceRecords.sourceId, this.sourceId), eq(sourceRecords.entityKind, entityKind), eq(sourceRecords.sourceRecordId, sourceRecordId)))
      .limit(1);
    const row = existing[0];
    let result: UpsertResult;
    if (row && row.payloadHash === hash) {
      this.counters.unchanged++;
      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));
      result = { id: row.id, status: 'unchanged' };
    } else {
      const rawPath = this.manifest.rawRetention === 'none' || this.mode === 'dry_run' ? null : await this.lake.put(entityKind, payload);
      if (row) {
        this.counters.updated++;
        await this.db
          .update(sourceRecords)
          .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() })
          .where(eq(sourceRecords.id, row.id));
        result = { id: row.id, status: 'updated' };
      } else {
        this.counters.created++;
        const inserted = await this.db
          .insert(sourceRecords)
          .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 })
          .returning({ id: sourceRecords.id });
        result = { id: inserted[0]!.id, status: 'created' };
      }
    }
    await this.checkpoint(1);
    return result;
  }

  /**
   * Batched `upsertSourceRecord` for bulk connectors (one SELECT + one INSERT per batch). Same
   * idempotency, raw-lake and checkpoint semantics; returns the result per `sourceRecordId`.
   */
  async upsertSourceRecordsBatch(entityKind: string, items: BatchUpsertItem[]): Promise<Map<string, UpsertResult>> {
    const out = new Map<string, UpsertResult>();
    if (!items.length) return out;
    const hashes = new Map<string, string>();
    for (const it of items) {
      hashes.set(it.sourceRecordId, createHash('sha256').update(JSON.stringify(it.payload)).digest('hex'));
      this.observe(entityKind, it.payload);
    }
    this.counters.fetched += items.length;
    const existing = await this.db
      .select({ id: sourceRecords.id, sourceRecordId: sourceRecords.sourceRecordId, payloadHash: sourceRecords.payloadHash })
      .from(sourceRecords)
      .where(and(eq(sourceRecords.sourceId, this.sourceId), eq(sourceRecords.entityKind, entityKind), inArray(sourceRecords.sourceRecordId, items.map((i) => i.sourceRecordId))));
    const byId = new Map(existing.map((e) => [e.sourceRecordId, e]));
    const unchangedIds: number[] = [];
    const toInsert: Array<{ item: BatchUpsertItem; values: typeof sourceRecords.$inferInsert }> = [];
    const now = new Date();
    const writeRaw = this.manifest.rawRetention !== 'none' && this.mode !== 'dry_run';
    for (const it of items) {
      const hash = hashes.get(it.sourceRecordId)!;
      const ex = byId.get(it.sourceRecordId);
      if (ex && ex.payloadHash === hash) {
        unchangedIds.push(ex.id);
        this.counters.unchanged++;
        out.set(it.sourceRecordId, { id: ex.id, status: 'unchanged' });
        continue;
      }
      const rawPath = writeRaw ? await this.lake.put(entityKind, it.payload) : null;
      if (ex) {
        this.counters.updated++;
        await this.db
          .update(sourceRecords)
          .set({ payloadHash: hash, rawPath, lastSeenRun: this.runId, retrievedAt: now, sourceUpdatedAt: it.sourceUpdatedAt ?? null, status: 'active', canonicalType: it.canonicalType, canonicalId: it.canonicalId, updatedAt: now })
          .where(eq(sourceRecords.id, ex.id));
        out.set(it.sourceRecordId, { id: ex.id, status: 'updated' });
      } else {
        this.counters.created++;
        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 } });
      }
    }
    if (unchangedIds.length) await this.db.update(sourceRecords).set({ lastSeenRun: this.runId, status: 'active' }).where(inArray(sourceRecords.id, unchangedIds));
    if (toInsert.length) {
      const inserted = await this.db
        .insert(sourceRecords)
        .values(toInsert.map((t) => t.values))
        .onConflictDoNothing()
        .returning({ id: sourceRecords.id, sourceRecordId: sourceRecords.sourceRecordId });
      const idBySrc = new Map(inserted.map((r) => [r.sourceRecordId, r.id]));
      for (const t of toInsert) out.set(t.item.sourceRecordId, { id: idBySrc.get(t.item.sourceRecordId) ?? -1, status: 'created' });
    }
    await this.checkpoint(items.length);
    return out;
  }

  /** Insert a provenance row (CLAUDE.md §2) and return its id. */
  async addProvenance(p: Omit<ProvenanceInput, 'sourceId' | 'retrievedAt' | 'ingestRunId'> & { retrievedAt?: string }): Promise<number> {
    const rows = await this.db
      .insert(provenance)
      .values({
        sourceId: this.sourceId,
        sourceRecordId: p.sourceRecordId,
        sourceUrl: p.sourceUrl,
        dataset: p.dataset,
        datasetVersion: p.datasetVersion ?? this.datasetVersion,
        publicationId: p.publicationId,
        pmid: p.pmid,
        doi: p.doi,
        retrievedAt: p.retrievedAt ? new Date(p.retrievedAt) : new Date(),
        publishedAt: p.publishedAt,
        updatedAtSource: p.updatedAt,
        geography: p.geography,
        population: p.population,
        cohortSize: p.cohortSize,
        methodology: p.methodology,
        evidenceType: p.evidenceType,
        accessLevel: p.accessLevel,
        confidence: p.confidence,
        license: p.license ?? this.manifest.license,
        ingestRunId: this.runId,
      })
      .returning({ id: provenance.id });
    return rows[0]!.id;
  }

  /** Queue an unmapped label for curation (CLAUDE.md §222). */
  async recordUnresolved(entityKind: string, sourceText: string, normalized: string, context: Record<string, unknown> = {}, suggestion?: { id: string; matchType: string; score: number }): Promise<void> {
    if (!normalized) return;
    await this.db
      .insert(unresolvedLabels)
      .values({ sourceId: this.sourceId, entityKind, sourceText, normalized, context, suggestedId: suggestion?.id, suggestedMatchType: suggestion?.matchType, suggestedScore: suggestion?.score })
      .onConflictDoUpdate({
        target: [unresolvedLabels.sourceId, unresolvedLabels.entityKind, unresolvedLabels.normalized],
        set: { count: sql`${unresolvedLabels.count} + 1`, updatedAt: new Date(), ...(suggestion ? { suggestedId: suggestion.id, suggestedMatchType: suggestion.matchType, suggestedScore: suggestion.score } : {}) },
      });
  }

  async recordChange(entityType: string, entityId: string, kind: string, summary: string, before?: unknown, after?: unknown): Promise<void> {
    await this.db.insert(changeEvents).values({ entityType, entityId, kind, summary, before: before ?? null, after: after ?? null, ingestRunId: this.runId });
  }

  /** @internal */
  async _finish(status: 'succeeded' | 'failed' | 'partial' | 'aborted', error?: string, anomaly?: string): Promise<void> {
    if (this.abortPromise) await this.abortPromise;
    await this.lake.close();
    try {
      await this.flushFieldStats();
    } catch (e) {
      this.warn(`field stats flush failed: ${(e as Error).message}`);
    }
    const finishedAt = new Date();
    await this.db
      .update(ingestRuns)
      .set({
        status,
        finishedAt,
        durationMs: finishedAt.getTime() - this.startedAt.getTime(),
        recordsFetched: this.counters.fetched,
        recordsCreated: this.counters.created,
        recordsUpdated: this.counters.updated,
        recordsUnchanged: this.counters.unchanged,
        recordsRejected: this.counters.rejected,
        httpRequests: this.http.stats.requests,
        httpFailures: this.http.stats.failures,
        rateLimitEvents: this.http.stats.rateLimitEvents,
        validationFailures: this.counters.validationFailures,
        schemaDrift: this.drift,
        cursorAfter: this.cursor,
        error: error ?? null,
        log: this.logs,
        datasetVersion: this.datasetVersion ?? null,
        anomaly: anomaly ?? this.anomaly ?? null,
      })
      .where(eq(ingestRuns.id, this.runId));
  }
}

export interface ConnectorHealth {
  status: 'healthy' | 'degraded' | 'failing' | 'review' | 'awaiting_credentials' | 'unknown';
  detail?: string;
  lastSuccessAt?: Date | null;
}

/** Connector contract (CLAUDE.md §8). `sync` implements discover→fetch→normalize→reconcile→validate→persist. */
export abstract class Connector {
  abstract readonly manifest: ConnectorManifest;
  /** Cheap liveness probe against the source (no ingestion). */
  abstract healthCheck(ctx: RunContext): Promise<ConnectorHealth>;
  /** Main ingestion. Must be idempotent and restartable via ctx.cursor. */
  abstract sync(ctx: RunContext): Promise<void>;
  /** Optional: connector-specific credentials check. Return a reason when credentials are missing. */
  credentialsMissing(): string | null {
    return null;
  }
}

export interface RunOptions {
  mode?: RunMode;
  maxMinutes?: number;
  maxRecords?: number;
  resetCursor?: boolean;
  /**
   * Install SIGTERM/SIGINT handlers for the duration of the run (default true): the first signal
   * saves the cursor and marks the run aborted, the connector stops at its next `shouldStop()`;
   * a second signal exits the process immediately (the run is already marked aborted).
   */
  handleSignals?: boolean;
}

export interface RunResult {
  runId: string;
  status: 'succeeded' | 'partial' | 'failed' | 'aborted';
  counters: RunContext['counters'];
  anomaly: string | null;
}

const SIGNALS: NodeJS.Signals[] = ['SIGTERM', 'SIGINT'];

/** Execute one connector run end-to-end with bookkeeping in ingest_runs/connector_cursors/system_alerts. */
export async function runConnector(db: Database, connector: Connector, opts: RunOptions = {}): Promise<RunResult> {
  const m = connector.manifest;
  const [src] = await db.select({ id: sources.id, licenseStatus: sources.licenseStatus }).from(sources).where(eq(sources.slug, m.id)).limit(1);
  if (!src) throw new Error(`source ${m.id} not seeded — run db:seed`);
  const mode: RunMode = opts.mode ?? (m.supportsIncrementalSync ? 'incremental' : 'full');
  const missing = connector.credentialsMissing();
  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 }];
  const ctx = new RunContext(db, m, src.id, mode, { maxMinutes: opts.maxMinutes ?? envInt('CI_MAX_RUN_MINUTES', 45), maxRecords: opts.maxRecords }, Number(n) + 1);
  const [cur] = await db.select().from(connectorCursors).where(eq(connectorCursors.connectorId, m.id)).limit(1);
  ctx.cursor = opts.resetCursor || mode === 'full' ? {} : { ...(cur?.cursor ?? {}) };
  await db.insert(ingestRuns).values({ id: ctx.runId, connectorId: m.id, sourceId: src.id, mode, cursorBefore: ctx.cursor });
  await db
    .insert(connectorCursors)
    .values({ connectorId: m.id, cursor: ctx.cursor, lastAttemptAt: new Date() })
    .onConflictDoUpdate({ target: connectorCursors.connectorId, set: { lastAttemptAt: new Date() } });

  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 } });

  if (missing) {
    ctx.warn(`credentials missing: ${missing}`);
    await ctx._finish('aborted', missing);
    await db.update(connectorCursors).set({ health: 'awaiting_credentials', healthDetail: missing, updatedAt: new Date() }).where(eq(connectorCursors.connectorId, m.id));
    await alert(ALERT_KINDS.connectorAborted, 'warn', `credentials missing: ${missing}`);
    return { runId: ctx.runId, status: 'aborted', counters: ctx.counters, anomaly: null };
  }
  if (m.licenseStatus === 'blocked') {
    const msg = 'license status blocked — connector will not ingest (CLAUDE.md §142)';
    ctx.warn(msg);
    await ctx._finish('aborted', msg);
    return { runId: ctx.runId, status: 'aborted', counters: ctx.counters, anomaly: null };
  }

  // Signal handling: first signal → graceful abort with cursor saved; second → exit now.
  const onSignal = (signal: NodeJS.Signals) => {
    if (ctx.abortReason) {
      ctx.log.error({ signal }, 'second signal — exiting immediately (run already marked aborted, cursor saved)');
      process.exit(130);
    }
    void ctx.requestAbort(`${signal} received`);
  };
  const handleSignals = opts.handleSignals ?? true;
  if (handleSignals) for (const s of SIGNALS) process.on(s, onSignal);

  try {
    await ctx.loadKnownFields();
    await connector.sync(ctx);
    if (ctx.abortReason) {
      await ctx._finish('aborted', ctx.abortReason);
      await alert(ALERT_KINDS.connectorAborted, 'info', `run aborted: ${ctx.abortReason}`, { cursor: ctx.cursor, fetched: ctx.counters.fetched });
      return { runId: ctx.runId, status: 'aborted', counters: ctx.counters, anomaly: null };
    }
    const status = ctx.stoppedEarly ? 'partial' : 'succeeded';
    await ctx._finish(status);
    if (mode !== 'dry_run') {
      await db
        .update(connectorCursors)
        .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() })
        .where(eq(connectorCursors.connectorId, m.id));
      await db.update(sources).set({ status: 'active', updatedAt: new Date() }).where(eq(sources.id, src.id));
      // Alerts (CLAUDE.md §170): a success clears failure/stale alerts; drift is raised or cleared.
      try {
        await resolveAlerts(db, RESOLVED_ON_SUCCESS, m.id);
        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) });
        else await resolveAlerts(db, ALERT_KINDS.schemaDrift, m.id);
      } catch (e) {
        ctx.warn(`alert bookkeeping failed: ${(e as Error).message}`);
      }
    }
    return { runId: ctx.runId, status, counters: ctx.counters, anomaly: null };
  } catch (err) {
    const e = err as Error;
    const msg = err instanceof Error ? `${err.message}\n${err.stack ?? ''}`.slice(0, 4000) : String(err);
    if (ctx.abortReason) {
      // The connector threw while stopping (e.g. closed socket) — the abort is the real status.
      ctx.warn(`connector threw during abort: ${e.message}`);
      await ctx._finish('aborted', `${ctx.abortReason}; ${e.message}`.slice(0, 4000));
      await alert(ALERT_KINDS.connectorAborted, 'info', `run aborted: ${ctx.abortReason}`, { cursor: ctx.cursor, fetched: ctx.counters.fetched });
      return { runId: ctx.runId, status: 'aborted', counters: ctx.counters, anomaly: null };
    }
    const anomaly = ctx.anomaly ?? (err instanceof AnomalyError || /^anomaly\b/i.test(e.message ?? '') ? e.message.slice(0, 2000) : null);
    ctx.error(`run failed: ${e.message}`);
    await ctx._finish('failed', msg, anomaly ?? undefined);
    // Preserve partial cursor progress so restarts resume (CLAUDE.md §90).
    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));
    if (anomaly) await alert(ALERT_KINDS.anomaly, 'critical', anomaly.split('\n')[0]!, { fetched: ctx.counters.fetched });
    else await alert(ALERT_KINDS.connectorFailure, 'warn', (e.message ?? 'unknown error').split('\n')[0]!, { fetched: ctx.counters.fetched });
    return { runId: ctx.runId, status: 'failed', counters: ctx.counters, anomaly };
  } finally {
    if (handleSignals) for (const s of SIGNALS) process.off(s, onSignal);
  }
}