import { mkdirSync, writeFileSync } from 'node:fs';
import path from 'node:path';
import { and, eq, sql } from 'drizzle-orm';
import { connectorRuns, connectors as connectorsTable, rawRecords } from '@rareindex/database';
import { createCrawlContext, loadConnector, type CrawlOptions, type RawRecordInput } from '@rareindex/connectors';
import { env, logger, newId, sha256 } 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';
import { duplicateExplosion } from '@rareindex/connectors';

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

const SNAPSHOT_THRESHOLD = 8 * 1024;

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

/**
 * 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.
 */
export async function runCrawl(connectorId: string, options: Partial<CrawlOptions> & { trigger?: string; 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);
  if (state && state.status !== 'active' && options.trigger !== 'manual') {
    log.info({ status: state.status }, 'connector not active; skipping');
    return { runId: '', connectorId, status: 'failed', recordsRaw: 0, recordsDuplicate: 0, pagesAttempted: 0, pagesSuccess: 0, error: `connector ${state.status}`, durationMs: 0 };
  }
  const runId = newId('job');
  const mode = options.mode ?? 'incremental';
  const savedCursor = (state?.config as { cursor?: Record<string, unknown> } | undefined)?.cursor;
  const crawlOptions: CrawlOptions = { mode, limit: options.limit, categories: options.categories, seeds: options.seeds, cursor: mode === 'backfill' ? undefined : savedCursor };
  await db().insert(connectorRuns).values({ id: runId, connectorId, trigger: options.trigger ?? 'schedule', startedAt: new Date(), status: 'running', cursor: savedCursor ?? {} });

  const router = getRouter(connectorId);
  const anomalies: string[] = [];
  let cursor: Record<string, unknown> | 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));
    },
    onAnomaly: (kind, detail) => anomalies.push(detail ? `${kind}: ${detail}` : kind),
  });

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

  try {
    for await (const rec of connector.crawl(ctx)) {
      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 dupAnomaly = duplicateExplosion(raw + dupes, dupes);
  if (dupAnomaly && mode !== 'backfill') anomalies.push(dupAnomaly);
  if (attempted > 0 && success === 0) anomalies.push('all_pages_failed');
  const status: CrawlRunResult['status'] = error ? (raw > 0 ? 'partial' : 'failed') : 'success';
  const finishedAt = new Date();
  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: cursor ?? savedCursor ?? {} })
    .where(eq(connectorRuns.id, runId));
  const refresh = state?.refreshFrequencyMinutes ?? meta.refreshFrequencyMinutes;
  const nextRunAt = 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,
      config: cursor ? 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, raw, dupes, attempted, success, error });
  log.info({ runId, status, raw, dupes, attempted, success, credits, ms: Date.now() - started }, 'crawl finished');
  return { runId, connectorId, status, recordsRaw: raw, recordsDuplicate: dupes, pagesAttempted: attempted, pagesSuccess: success, error, durationMs: Date.now() - started };
}