import { mkdirSync, writeFileSync } from 'node:fs';
import path from 'node:path';
import { and, desc, eq, inArray, sql } from 'drizzle-orm';
import { connectorBackfills, connectorRuns, connectors as connectorsTable, rawRecords } from '@rareindex/database';
import { createCrawlContext, duplicateExplosion, loadConnector, missingRequirements, type BackfillProgress, type CrawlOptions, type RawRecordInput } from '@rareindex/connectors';
import { env, logger, newId, sha256, toDateOnly } from '@rareindex/shared';
import { db } from '../lib/db.ts';
import { createBudgetStore } from '../lib/budget.ts';
import { getRouter } from '../lib/router.ts';
import { flushCosts } from '../lib/costs.ts';
import { emit } from '../lib/events.ts';

export interface CrawlRunResult {
  runId: string;
  connectorId: string;
  status: 'success' | 'partial' | 'failed' | 'skipped';
  recordsRaw: number;
  recordsDuplicate: number;
  pagesAttempted: number;
  pagesSuccess: number;
  error: string | null;
  durationMs: number;
  backfillId?: string | null;
}

const SNAPSHOT_THRESHOLD = 8 * 1024;
const BACKFILL_MAX_ERRORS = 8;

function snapshotPath(connectorId: string, hash: string): string {
  const dir = path.resolve(env().RI_DATA_DIR, 'raw', connectorId);
  mkdirSync(dir, { recursive: true });
  return path.join(dir, `${hash}.html`);
}

/** Active (running/paused) backfill campaign for a connector, if any. */
export async function activeBackfill(connectorId: string) {
  const [row] = await db().select().from(connectorBackfills).where(and(eq(connectorBackfills.connectorId, connectorId), inArray(connectorBackfills.status, ['running', 'paused']))).orderBy(desc(connectorBackfills.startedAt)).limit(1);
  return row ?? null;
}

/** Start (or reset) a backfill campaign. Existing running campaigns are resumed unless `reset` is set. */
export async function startBackfill(connectorId: string, opts: { reset?: boolean; startDate?: string | null; endDate?: string | null } = {}) {
  const existing = await activeBackfill(connectorId);
  if (existing && !opts.reset) {
    if (existing.status === 'paused') await db().update(connectorBackfills).set({ status: 'running', updatedAt: new Date() }).where(eq(connectorBackfills.id, existing.id));
    return existing;
  }
  if (existing) await db().update(connectorBackfills).set({ status: 'failed', lastError: 'reset by operator', finishedAt: new Date(), updatedAt: new Date() }).where(eq(connectorBackfills.id, existing.id));
  const id = newId('backfill');
  const [row] = await db().insert(connectorBackfills).values({ id, connectorId, status: 'running', startedAt: new Date(), backfillStartDate: opts.startDate ?? null, backfillEndDate: opts.endDate ?? null }).returning();
  return row!;
}

export async function pauseBackfill(connectorId: string): Promise<boolean> {
  const b = await activeBackfill(connectorId);
  if (!b) return false;
  await db().update(connectorBackfills).set({ status: 'paused', updatedAt: new Date() }).where(eq(connectorBackfills.id, b.id));
  return true;
}

/**
 * Run one connector crawl (§108 raw layer): bookkeeping row in connector_runs, routed fetches with
 * budget + cost tracking, immutable raw_records with content-hash dedupe, cursor persistence.
 * Backfill mode is resumable (SPEC §9): the cursor comes from the active connector_backfills row and
 * progress is written back after every checkpoint, so an interrupted campaign continues where it stopped.
 */
export async function runCrawl(connectorId: string, options: Partial<CrawlOptions> & { trigger?: string; resetBackfill?: boolean; onRaw?: (ids: string[]) => Promise<void> } = {}): Promise<CrawlRunResult> {
  const started = Date.now();
  const log = logger.child({ component: 'crawler', connector: connectorId });
  const connector = await loadConnector(connectorId);
  const meta = connector.meta;
  const [state] = await db().select().from(connectorsTable).where(eq(connectorsTable.id, connectorId)).limit(1);
  const skip = (error: string): CrawlRunResult => ({ runId: '', connectorId, status: 'skipped', recordsRaw: 0, recordsDuplicate: 0, pagesAttempted: 0, pagesSuccess: 0, error, durationMs: 0 });
  if (state && state.status !== 'active' && options.trigger !== 'manual') {
    log.info({ status: state.status }, 'connector not active; skipping');
    return skip(`connector ${state.status}`);
  }
  const missing = missingRequirements(meta);
  if (missing.length) {
    log.warn({ missing }, 'connector requirements missing; skipping (gated source)');
    await db().update(connectorsTable).set({ nextRunAt: new Date(Date.now() + 24 * 3600_000), updatedAt: new Date() }).where(eq(connectorsTable.id, connectorId));
    return skip(`missing env: ${missing.join(', ')}`);
  }
  const runId = newId('job');
  const mode = options.mode ?? 'incremental';
  const savedCursor = (state?.config as { cursor?: Record<string, unknown> } | undefined)?.cursor;

  // Backfill campaign bookkeeping (resumable)
  let backfill = mode === 'backfill' ? await startBackfill(connectorId, { reset: options.resetBackfill }) : null;
  if (backfill && backfill.status === 'paused' && options.trigger !== 'manual') return skip('backfill paused');
  const backfillCursor = backfill && Object.keys(backfill.lastCursor ?? {}).length ? (backfill.lastCursor as Record<string, unknown>) : undefined;
  if (backfillCursor && (backfillCursor as { done?: boolean }).done) {
    await db().update(connectorBackfills).set({ status: 'completed', finishedAt: new Date(), updatedAt: new Date(), percent: 100 }).where(eq(connectorBackfills.id, backfill!.id));
    return skip('backfill already completed');
  }
  const crawlOptions: CrawlOptions = { mode, limit: options.limit, categories: options.categories, seeds: options.seeds, cursor: mode === 'backfill' ? backfillCursor : savedCursor };
  await db().insert(connectorRuns).values({ id: runId, connectorId, trigger: options.trigger ?? 'schedule', startedAt: new Date(), status: 'running', cursor: crawlOptions.cursor ?? {} });
  if (backfill) await db().update(connectorBackfills).set({ runs: sql`${connectorBackfills.runs} + 1`, status: 'running', updatedAt: new Date() }).where(eq(connectorBackfills.id, backfill.id));

  const router = getRouter(connectorId);
  const anomalies: string[] = [];
  let cursor: Record<string, unknown> | null = null;
  let lastProgress: BackfillProgress | null = null;
  const ctx = createCrawlContext({
    router,
    meta,
    options: crawlOptions,
    log,
    budget: createBudgetStore({ connectorId }),
    onCursor: async (c) => {
      cursor = c;
      await db().update(connectorRuns).set({ cursor: c }).where(eq(connectorRuns.id, runId));
      if (backfill) await db().update(connectorBackfills).set({ lastCursor: c, updatedAt: new Date() }).where(eq(connectorBackfills.id, backfill.id));
    },
    onAnomaly: (kind, detail) => anomalies.push(detail ? `${kind}: ${detail}` : kind),
    onProgress: async (p) => {
      if (!backfill) return;
      lastProgress = p;
      const percent = p.totalPages && p.page ? Math.min(100, Math.round((p.page / p.totalPages) * 1000) / 10) : null;
      await db()
        .update(connectorBackfills)
        .set({
          pagesProcessed: p.page !== undefined ? sql`greatest(${connectorBackfills.pagesProcessed}, ${p.page})` : sql`${connectorBackfills.pagesProcessed} + 1`,
          lastSuccessfulPage: p.page ?? sql`${connectorBackfills.lastSuccessfulPage}`,
          totalPages: p.totalPages ?? sql`${connectorBackfills.totalPages}`,
          itemsProcessed: p.itemsProcessed !== undefined ? sql`greatest(${connectorBackfills.itemsProcessed}, ${p.itemsProcessed})` : sql`${connectorBackfills.itemsProcessed}`,
          reachedDate: p.reachedDate ? toDateOnly(p.reachedDate) : sql`${connectorBackfills.reachedDate}`,
          percent: percent ?? sql`${connectorBackfills.percent}`,
          updatedAt: new Date(),
        })
        .where(eq(connectorBackfills.id, backfill.id));
    },
  });

  let raw = 0;
  let dupes = 0;
  let error: string | null = null;
  let batch: Array<typeof rawRecords.$inferInsert> = [];
  const newIds: string[] = [];

  async function flush() {
    if (batch.length === 0) return;
    const rows = batch;
    batch = [];
    const inserted = await db()
      .insert(rawRecords)
      .values(rows)
      .onConflictDoNothing({ target: [rawRecords.connectorId, rawRecords.contentHash] })
      .returning({ id: rawRecords.id });
    raw += inserted.length;
    dupes += rows.length - inserted.length;
    for (const r of inserted) newIds.push(r.id);
    if (options.onRaw && inserted.length) await options.onRaw(inserted.map((r) => r.id));
  }

  // Time budget per run: very large sources (100k+ pages) would otherwise monopolise crawl slots for hours.
  // The connector's cursor is persisted at its checkpoints, so a time-boxed run simply resumes on the next tick.
  const maxRunMs = Number(process.env.RI_MAX_RUN_MINUTES ?? 40) * 60_000;
  let timeBoxed = false;
  try {
    for await (const rec of connector.crawl(ctx)) {
      if (mode !== 'probe' && Date.now() - started > maxRunMs) {
        timeBoxed = true;
        break;
      }
      const input: RawRecordInput = rec;
      const payloadText = typeof input.payload === 'string' ? input.payload : JSON.stringify(input.payload ?? null);
      const contentHash = sha256(`${input.kind}|${input.externalId ?? input.url}|${payloadText}`);
      let snapshotRef: string | null = null;
      if (input.snapshot && input.snapshot.length > SNAPSHOT_THRESHOLD) {
        const p = snapshotPath(connectorId, contentHash);
        writeFileSync(p, input.snapshot);
        snapshotRef = p;
      }
      batch.push({
        id: newId('raw'),
        connectorId,
        sourceId: meta.sourceId,
        runId,
        engine: input.engine,
        url: input.url,
        externalId: input.externalId ?? null,
        kind: input.kind,
        fetchedAt: input.fetchedAt ?? new Date(),
        contentHash,
        httpStatus: input.httpStatus ?? null,
        payload: input.snapshot && !snapshotRef && typeof input.payload === 'object' && input.payload ? { ...(input.payload as object), snapshot: input.snapshot } : (input.payload as object),
        snapshotRef,
        parserVersion: connector.parserVersion,
        connectorVersion: connector.version,
      });
      if (batch.length >= 200) await flush();
      if (ctx.signal?.aborted) break;
    }
    await flush();
  } catch (err) {
    error = err instanceof Error ? `${err.message}` : String(err);
    log.error({ err }, 'crawl failed');
    try {
      await flush();
    } catch (e2) {
      log.error({ err: e2 }, 'flush after failure failed');
    }
  }

  const stats = ctx.engineStats;
  const attempted = Object.values(stats).reduce((a, s) => a + s.attempts, 0);
  const success = Object.values(stats).reduce((a, s) => a + s.success, 0);
  const credits = Object.values(stats).reduce((a, s) => a + s.credits, 0);
  const blocked = Object.values(stats).reduce((a, s) => a + (s.blocked ?? 0), 0);
  const refused = Object.values(stats).reduce((a, s) => a + (s.circuitOpen ?? 0), 0);
  const dupAnomaly = duplicateExplosion(raw + dupes, dupes);
  if (dupAnomaly && mode !== 'backfill') anomalies.push(dupAnomaly);
  if (attempted > 0 && success === 0) anomalies.push('all_pages_failed');
  if (blocked > 0) anomalies.push(`challenge_events: ${blocked}`);
  if (refused > 0) anomalies.push(`circuit_open: ${refused} requests refused`);
  if (timeBoxed) anomalies.push('time_budget_reached');
  const status: CrawlRunResult['status'] = error ? (raw > 0 ? 'partial' : 'failed') : timeBoxed ? 'partial' : 'success';
  const finishedAt = new Date();
  const finalCursor: Record<string, unknown> | undefined = cursor ?? (crawlOptions.cursor as Record<string, unknown> | undefined) ?? undefined;
  await db()
    .update(connectorRuns)
    .set({ finishedAt, status, pagesAttempted: attempted, pagesSuccess: success, recordsRaw: raw, recordsDuplicate: dupes, engineStats: stats, anomalies: [...anomalies, ...ctx.anomalies.filter((a) => !anomalies.includes(a))], error, costCredits: credits, cursor: finalCursor ?? {} })
    .where(eq(connectorRuns.id, runId));

  // Backfill campaign state (SPEC §9): completed when the connector signalled `done`, or when a run
  // finished naturally (not time-boxed, no error); otherwise it stays running and resumes next tick.
  if (backfill) {
    const done = Boolean((finalCursor as { done?: boolean } | undefined)?.done) || (status === 'success' && !timeBoxed);
    const failedHard = status === 'failed';
    const [bf] = await db().select({ errors: connectorBackfills.errors }).from(connectorBackfills).where(eq(connectorBackfills.id, backfill.id)).limit(1);
    const errors = (bf?.errors ?? 0) + (error ? 1 : 0);
    await db()
      .update(connectorBackfills)
      .set({
        status: done ? 'completed' : failedHard && errors >= BACKFILL_MAX_ERRORS ? 'failed' : 'running',
        finishedAt: done || (failedHard && errors >= BACKFILL_MAX_ERRORS) ? finishedAt : null,
        errors,
        retryCount: error ? sql`${connectorBackfills.retryCount} + 1` : sql`${connectorBackfills.retryCount}`,
        lastError: error,
        lastCursor: finalCursor ?? {},
        itemsProcessed: sql`${connectorBackfills.itemsProcessed} + ${raw}`,
        percent: done ? 100 : sql`${connectorBackfills.percent}`,
        updatedAt: finishedAt,
      })
      .where(eq(connectorBackfills.id, backfill.id));
    if (done) log.info({ backfillId: backfill.id }, 'backfill completed');
  }

  const refresh = state?.refreshFrequencyMinutes ?? meta.refreshFrequencyMinutes;
  // A time-boxed run resumes quickly (cursor kept); failures back off ×2.
  const nextRunAt = timeBoxed ? new Date(finishedAt.getTime() + 3 * 60_000) : new Date(finishedAt.getTime() + refresh * 60_000 * (status === 'failed' ? 2 : 1));
  await db()
    .update(connectorsTable)
    .set({
      lastRunAt: finishedAt,
      lastSuccessAt: status === 'failed' ? sql`${connectorsTable.lastSuccessAt}` : finishedAt,
      nextRunAt,
      // incremental cursors live in connectors.config; backfill cursors live on the campaign row
      config: cursor && mode !== 'backfill' ? sql`${connectorsTable.config} || ${JSON.stringify({ cursor })}::jsonb` : sql`${connectorsTable.config}`,
      updatedAt: finishedAt,
    })
    .where(and(eq(connectorsTable.id, connectorId)));
  await flushCosts();
  await emit('connector_run_finished', { type: 'connector_run', id: runId }, { connectorId, status, mode, raw, dupes, attempted, success, error, backfillId: backfill?.id ?? null, progress: lastProgress });
  log.info({ runId, status, mode, raw, dupes, attempted, success, credits, blocked, ms: Date.now() - started }, 'crawl finished');
  return { runId, connectorId, status, recordsRaw: raw, recordsDuplicate: dupes, pagesAttempted: attempted, pagesSuccess: success, error, durationMs: Date.now() - started, backfillId: backfill?.id ?? null };
}