/**
 * CancerIndex worker (CLAUDE.md §90, §290): pg-boss scheduler + job handlers.
 *
 *   connector.run        {id, mode?, maxMinutes?, maxRecords?, resetCursor?}  singleton per connector
 *   maintenance.counters {thenRank?}                                          refreshCounters
 *   maintenance.rank     {}                                                   computeAllRankings
 *   health.probe         {}                                                   healthCheck per connector → connector_cursors.health
 *
 * Schedules are (re)registered from connector manifests on every start; stale schedules are removed.
 * pg-boss gotcha: it refuses `undefined` option values — always omit keys instead of passing undefined.
 */
import { CI_MAX_RUN_MINUTES, DATABASE_URL, JOBS, PGBOSS_SCHEMA, SCHEDULES, WORKER_CONCURRENCY, type JobName } from './lib/env.js';
import { PgBoss, type Job } from 'pg-boss';
import { sql } from 'drizzle-orm';
import { logger } from '@cancerindex/shared';
import { getDb, closeDb } from '@cancerindex/database';
import { CONNECTORS, getConnector, runConnector, RunContext, type RunMode } from '@cancerindex/connectors';
import { refreshCounters, computeAllRankings } from '@cancerindex/ranking';

const log = logger.child({ component: 'worker' });
const db = getDb();

interface ConnectorRunData {
  id: string;
  mode?: RunMode;
  maxMinutes?: number;
  maxRecords?: number;
  resetCursor?: boolean;
  requestedBy?: string;
}

const boss = new PgBoss({
  connectionString: DATABASE_URL,
  schema: PGBOSS_SCHEMA,
  application_name: 'cancerindex-worker',
  max: Math.max(4, WORKER_CONCURRENCY + 2),
  // supervise + schedule are on: this process owns maintenance and cron.
});
boss.on('error', (err) => log.error({ err }, 'pg-boss error'));
boss.on('warning', (w) => log.warn({ w }, 'pg-boss warning'));

/**
 * Queue policy `stately`: one job per singletonKey in created/retry/active — a connector is never
 * queued twice nor run concurrently. Policy cannot be altered in place, so a queue created with
 * another policy is dropped (with its pending jobs) and recreated.
 */
const QUEUE_POLICY = 'stately';

async function ensureQueue(name: JobName, expireInSeconds: number): Promise<void> {
  const existing = await boss.getQueue(name);
  if (existing && existing.policy === QUEUE_POLICY) {
    await boss.updateQueue(name, { expireInSeconds, retryLimit: 0 });
    return;
  }
  if (existing) {
    log.warn({ queue: name, from: existing.policy, to: QUEUE_POLICY }, 'recreating queue with new policy');
    await boss.deleteQueue(name);
  }
  await boss.createQueue(name, { policy: QUEUE_POLICY, retryLimit: 0, expireInSeconds });
}

/** Register cron schedules from manifests; one schedule key per connector so schedules can be diffed. */
async function registerSchedules(): Promise<void> {
  const active = CONNECTORS.filter((c) => c.manifest.status === 'active' && c.manifest.schedule);
  const wanted = new Set(active.map((c) => c.manifest.id));
  for (const s of await boss.getSchedules(JOBS.connectorRun)) {
    if (!wanted.has(s.key)) {
      await boss.unschedule(JOBS.connectorRun, s.key);
      log.info({ connector: s.key }, 'removed stale schedule');
    }
  }
  for (const c of active) {
    await boss.schedule(JOBS.connectorRun, c.manifest.schedule!, { id: c.manifest.id, requestedBy: 'schedule' }, { key: c.manifest.id, singletonKey: c.manifest.id, tz: 'UTC' });
    log.info({ connector: c.manifest.id, cron: c.manifest.schedule }, 'schedule registered');
  }
  const skipped = CONNECTORS.filter((c) => !wanted.has(c.manifest.id)).map((c) => `${c.manifest.id}(${c.manifest.status}${c.manifest.schedule ? '' : ',no-schedule'})`);
  if (skipped.length) log.info({ skipped }, 'connectors without schedule');
  await boss.schedule(JOBS.counters, SCHEDULES.counters, { thenRank: false, requestedBy: 'schedule' }, { key: 'daily', singletonKey: 'counters', tz: 'UTC' });
  await boss.schedule(JOBS.rank, SCHEDULES.rank, { requestedBy: 'schedule' }, { key: 'daily', singletonKey: 'rank', tz: 'UTC' });
  await boss.schedule(JOBS.healthProbe, SCHEDULES.healthProbe, { requestedBy: 'schedule' }, { key: 'hourly', singletonKey: 'health', tz: 'UTC' });
  log.info({ counters: SCHEDULES.counters, rank: SCHEDULES.rank, health: SCHEDULES.healthProbe }, 'maintenance schedules registered (UTC)');
}

async function handleConnectorRun([job]: Job<ConnectorRunData>[]): Promise<void> {
  const data = job!.data;
  const connector = getConnector(data.id);
  if (!connector) {
    log.warn({ connector: data.id }, 'unknown connector — job dropped');
    return;
  }
  const [cur] = await db.execute<{ paused: boolean }>(sql`SELECT paused FROM connector_cursors WHERE connector_id = ${data.id}`);
  if (cur?.paused && data.requestedBy !== 'admin-api') {
    log.info({ connector: data.id }, 'connector paused — skipping scheduled run');
    return;
  }
  const opts: Parameters<typeof runConnector>[2] = { maxMinutes: data.maxMinutes ?? CI_MAX_RUN_MINUTES };
  if (data.mode) opts.mode = data.mode;
  if (data.maxRecords) opts.maxRecords = data.maxRecords;
  if (data.resetCursor) opts.resetCursor = true;
  log.info({ connector: data.id, jobId: job!.id, opts }, 'connector run starting');
  const result = await runConnector(db, connector, opts);
  log.info({ connector: data.id, runId: result.runId, status: result.status, ...result.counters }, 'connector run finished');
  const changed = result.counters.created + result.counters.updated;
  if (changed > 0 && data.mode !== 'dry_run') {
    // Derived layers follow canonical changes: counters, then rankings (chained by the counters handler).
    await boss.send(JOBS.counters, { thenRank: true, requestedBy: `connector:${data.id}` }, { singletonKey: 'counters' });
  }
}

async function handleCounters([job]: Job<{ thenRank?: boolean; requestedBy?: string }>[]): Promise<void> {
  const t0 = Date.now();
  const n = await refreshCounters(db);
  log.info({ entities: n, ms: Date.now() - t0, requestedBy: job!.data.requestedBy }, 'counters refreshed');
  if (job!.data.thenRank) await boss.send(JOBS.rank, { requestedBy: 'counters' }, { singletonKey: 'rank' });
}

async function handleRank([job]: Job<{ requestedBy?: string }>[]): Promise<void> {
  const t0 = Date.now();
  const res = await computeAllRankings(db);
  log.info({ snapshots: res.length, ms: Date.now() - t0, requestedBy: job!.data.requestedBy }, 'rankings recomputed');
}

async function handleHealthProbe(): Promise<void> {
  for (const c of CONNECTORS) {
    const id = c.manifest.id;
    try {
      const [cur] = await db.execute<{ health: string }>(sql`SELECT health FROM connector_cursors WHERE connector_id = ${id}`);
      // Never overwrite the credentials gate set by runConnector/credentialsMissing (CLAUDE.md §142).
      if (cur?.health === 'awaiting_credentials' || c.credentialsMissing()) continue;
      const [src] = await db.execute<{ id: string }>(sql`SELECT id FROM sources WHERE slug = ${id}`);
      const ctx = new RunContext(db, c.manifest, src?.id ?? 'CI-SOURCE-00000000', 'probe', { maxMinutes: 2 }, 0);
      const h = await c.healthCheck(ctx);
      await db.execute(sql`
        INSERT INTO connector_cursors (connector_id, health, health_detail, updated_at) VALUES (${id}, ${h.status}, ${h.detail ?? null}, now())
        ON CONFLICT (connector_id) DO UPDATE SET health = ${h.status}, health_detail = ${h.detail ?? null}, updated_at = now()`);
      log.info({ connector: id, health: h.status, detail: h.detail }, 'health probe');
    } catch (err) {
      log.warn({ connector: id, err }, 'health probe failed');
      await db.execute(sql`INSERT INTO connector_cursors (connector_id, health, health_detail, updated_at) VALUES (${id}, 'failing', ${(err as Error).message.slice(0, 500)}, now())
        ON CONFLICT (connector_id) DO UPDATE SET health = 'failing', health_detail = EXCLUDED.health_detail, updated_at = now()`);
    }
  }
}

async function main(): Promise<void> {
  await boss.start();
  await ensureQueue(JOBS.connectorRun, (CI_MAX_RUN_MINUTES + 20) * 60);
  await ensureQueue(JOBS.counters, 30 * 60);
  await ensureQueue(JOBS.rank, 60 * 60);
  await ensureQueue(JOBS.healthProbe, 10 * 60);
  await registerSchedules();

  await boss.work<ConnectorRunData>(JOBS.connectorRun, { batchSize: 1, localConcurrency: WORKER_CONCURRENCY, pollingIntervalSeconds: 5 }, handleConnectorRun);
  await boss.work(JOBS.counters, { batchSize: 1, pollingIntervalSeconds: 5 }, handleCounters);
  await boss.work(JOBS.rank, { batchSize: 1, pollingIntervalSeconds: 5 }, handleRank);
  await boss.work(JOBS.healthProbe, { batchSize: 1, pollingIntervalSeconds: 30 }, handleHealthProbe);
  log.info({ concurrency: WORKER_CONCURRENCY, maxRunMinutes: CI_MAX_RUN_MINUTES, connectors: CONNECTORS.map((c) => c.manifest.id) }, 'worker ready');
}

let stopping = false;
async function shutdown(signal: string): Promise<void> {
  if (stopping) return;
  stopping = true;
  log.info({ signal }, 'worker shutting down (graceful, up to 60 s)');
  try {
    await boss.stop({ graceful: true, close: true, timeout: 60_000 });
    await closeDb();
  } catch (err) {
    log.error({ err }, 'shutdown error');
  } finally {
    process.exit(0);
  }
}
process.on('SIGTERM', () => void shutdown('SIGTERM'));
process.on('SIGINT', () => void shutdown('SIGINT'));

main().catch((err) => {
  log.error({ err }, 'worker failed to start');
  process.exit(1);
});