import { coverageKey, newId, slugify, sourceSchema, type SourceSeed, type Tier } from "@websensor/core";
import { discoverOrganization, type CandidateKind, type DeepDiscoveryHints } from "@websensor/connectors";
import { db, factorySeeds, sensors, sql, type FactorySeed } from "@websensor/db";
import { factoryConfig, log } from "../config";
import { priorityFor, upsertSourceRecord } from "../registry";
import { scoreCandidate, selectForShadow } from "./score";

/**
 * Source Factory — one organization at a time:
 *   seed → deep discovery → (create or attach the source) → score candidates → promotion policy →
 *   shadow sensors (`status = SHADOW`: observed, evidence stored, nothing published) → later `evaluateShadows()`.
 */
export interface ProcessOutcome {
  seedId: string;
  sourceId: string | null;
  status: "discovered" | "blocked" | "error";
  candidates: number;
  shadow: number;
  requests: number;
  durationMs: number;
  notes: string[];
}

interface SourceRow extends Record<string, unknown> {
  id: string;
  domain: string;
  categories: string[];
  tier: string;
  enabled: boolean;
}

let sourceIndex: { at: number; byKey: Map<string, SourceRow>; ids: Set<string> } | null = null;
async function sourcesByDomain(force = false): Promise<{ byKey: Map<string, SourceRow>; ids: Set<string> }> {
  if (!force && sourceIndex && Date.now() - sourceIndex.at < 5 * 60_000) return sourceIndex;
  const rows = await db.execute<SourceRow>(sql`select id, domain, categories, tier, enabled from sources where kind = 'registry'`);
  const byKey = new Map<string, SourceRow>();
  const ids = new Set<string>();
  for (const r of rows.rows) {
    ids.add(r.id);
    const k = coverageKey(r.domain);
    if (!byKey.has(k)) byKey.set(k, r);
  }
  sourceIndex = { at: Date.now(), byKey, ids };
  return sourceIndex;
}

export async function processSeed(seed: FactorySeed): Promise<ProcessOutcome> {
  const started = Date.now();
  const notes: string[] = [];
  const hints = (seed.hints ?? {}) as DeepDiscoveryHints & { posture?: boolean };
  const idx = await sourcesByDomain();
  const key = coverageKey(seed.domain);
  let source = idx.byKey.get(key) ?? null;

  const disc = await discoverOrganization(seed.domain, {
    hints,
    budget: factoryConfig.budget,
    deadlineMs: factoryConfig.deadlineMs,
    concurrency: 4,
    sensorIdForLogs: `factory_${seed.id}`,
    probePages: true,
    probeSubdomains: true,
    probePosture: seed.importance >= 3 || hints.posture === true,
    probeOpenApi: true,
  });
  notes.push(...disc.notes);
  // The homepage may redirect to another registrable domain that we already monitor.
  if (!source && disc.canonicalDomain !== key) source = idx.byKey.get(disc.canonicalDomain) ?? null;

  if (!disc.candidates.length) {
    const status = disc.blocked ? "blocked" : "discovered";
    await db.execute(sql`update factory_seeds set status = ${status}, candidates = 0, shadow = 0, source_id = ${source?.id ?? null}, discovered_at = now(), last_error = ${disc.blocked ? `homepage ${disc.homeStatus} (anti-bot)` : null}, updated_at = now() where id = ${seed.id}`);
    await bumpFactoryDaily({ seeds_processed: 1, requests: disc.requests, blocked: disc.blocked ? 1 : 0 });
    return { seedId: seed.id, sourceId: source?.id ?? null, status, candidates: 0, shadow: 0, requests: disc.requests, durationMs: Date.now() - started, notes };
  }

  // ---- source: attach or create --------------------------------------------------------------------------
  let sourceId: string;
  if (source) {
    sourceId = source.id;
    if (!source.enabled) notes.push("source disabled in registry — candidates recorded, no shadow sensors");
  } else {
    sourceId = idx.ids.has(seed.id) ? `${seed.id}-${slugify(seed.domain.split(".")[0] ?? "x")}`.slice(0, 70) : seed.id;
    const seedDoc: SourceSeed = sourceSchema.parse({
      id: sourceId,
      name: seed.name,
      domain: seed.domain,
      homepage: seed.homepage ?? `https://${seed.domain}`,
      categories: seed.categories,
      tier: (seed.tier as Tier) ?? "B",
      weight: seed.weight,
      aliases: seed.aliases,
      country: seed.country ?? undefined,
      language: seed.language ?? undefined,
      first_party: seed.firstParty,
      notes: `Source Factory · ${seed.sector ?? "-"}/${seed.universe ?? "-"} · discovered ${new Date().toISOString().slice(0, 10)}`,
      llm: true,
    });
    await upsertSourceRecord(seedDoc, { origin: "factory", sector: seed.sector });
    idx.ids.add(sourceId);
    idx.byKey.set(key, { id: sourceId, domain: seed.domain, categories: seed.categories, tier: seed.tier, enabled: true });
    source = idx.byKey.get(key)!;
  }

  // ---- existing sensors (this source + global URL uniqueness) ----------------------------------------------
  const existing = await db.execute<{ id: string; url: string; type: string; connector: string; config: Record<string, unknown>; status: string }>(sql`select id, url, type, connector, config, status from sensors where source_id = ${sourceId}`);
  const existingUrls = new Set(existing.rows.map((r) => r.url.replace(/\/$/, "")));
  const existingKinds = new Map<CandidateKind, number>();
  let existingTotal = 0;
  for (const r of existing.rows) {
    if (r.status === "DISABLED") continue;
    existingTotal++;
    const k = kindOfExisting(r);
    existingKinds.set(k, (existingKinds.get(k) ?? 0) + 1);
  }
  const urls = disc.candidates.map((c) => c.url);
  const global = urls.length ? await db.execute<{ url: string; source_id: string }>(sql`select url, source_id from sensors where url = any(${sql.raw("array[" + urls.map((u) => "'" + u.replace(/'/g, "''") + "'").join(",") + "]::text[]")}) and source_id <> ${sourceId}`) : { rows: [] as { url: string; source_id: string }[] };
  const ownedElsewhere = new Map(global.rows.map((r) => [r.url.replace(/\/$/, ""), r.source_id]));

  // ---- score + select ---------------------------------------------------------------------------------------
  const scored = disc.candidates.map((c) => scoreCandidate(c, { importance: seed.importance, weight: seed.weight }));
  const fresh = scored.filter((s) => !existingUrls.has(s.cand.url.replace(/\/$/, "")) && !ownedElsewhere.has(s.cand.url.replace(/\/$/, "")));
  const { selected, skipped } = source.enabled ? selectForShadow(fresh, { kinds: existingKinds, total: existingTotal }) : { selected: [], skipped: fresh.map((s) => ({ s, reason: "source disabled" })) };

  // ---- persist candidates ------------------------------------------------------------------------------------
  const day = new Date().toISOString().slice(0, 10);
  let shadowCreated = 0;
  for (const s of scored) {
    const k = s.cand.url.replace(/\/$/, "");
    let status = "candidate";
    let reason: string | null = null;
    if (existingUrls.has(k)) {
      status = "duplicate";
      reason = "already a sensor of this source";
    } else if (ownedElsewhere.has(k)) {
      status = "duplicate";
      reason = `already a sensor of ${ownedElsewhere.get(k)}`;
    } else if (selected.includes(s)) status = "shadow";
    else {
      const sk = skipped.find((x) => x.s === s);
      status = sk && /score/.test(sk.reason) ? "rejected" : "candidate";
      reason = sk?.reason ?? null;
    }
    const candId = newId("cand");
    let shadowSensorId: string | null = null;
    if (status === "shadow") {
      shadowSensorId = await createShadowSensor(sourceId, source, s.cand, candId, seed);
      if (shadowSensorId) shadowCreated++;
      else {
        status = "duplicate";
        reason = "sensor id/url collision";
      }
    }
    await db.execute(sql`
      insert into discovery_candidates (id, source_id, url, kind, evidence, score, status, seed_id, connector, type, name, kind_class, config, tier, score_value, reason, shadow_sensor_id, found_at, updated_at)
      values (${candId}, ${sourceId}, ${s.cand.url}, ${s.cand.type}, ${s.cand.evidence}, ${JSON.stringify({ value: s.cand.value, score: s.score, reasons: s.reasons, itemCount: s.cand.itemCount ?? null, itemsPerDay: s.cand.itemsPerDay ?? null, firstParty: s.cand.firstPartyConfidence, cost: s.cand.fetchCost, title: s.cand.title ?? null, fingerprint: s.cand.fingerprint ?? null })}::jsonb, ${status}, ${seed.id}, ${s.cand.connector}, ${s.cand.type}, ${s.cand.name}, ${s.cand.kind}, ${JSON.stringify(s.cand.config)}::jsonb, ${s.cand.suggestedTier}, ${s.score}, ${reason}, ${shadowSensorId}, now(), now())
      on conflict (source_id, url) do update set evidence = excluded.evidence, score = excluded.score, score_value = excluded.score_value, seed_id = excluded.seed_id, connector = excluded.connector, type = excluded.type, name = excluded.name, kind_class = excluded.kind_class, config = excluded.config, tier = excluded.tier, updated_at = now(),
        status = case when discovery_candidates.status in ('accepted','rejected','shadow') then discovery_candidates.status else excluded.status end,
        reason = case when discovery_candidates.status in ('accepted','rejected','shadow') then discovery_candidates.reason else excluded.reason end,
        shadow_sensor_id = coalesce(discovery_candidates.shadow_sensor_id, excluded.shadow_sensor_id)`);
  }
  for (const r of disc.rejected.slice(0, 40)) {
    await db.execute(sql`insert into discovery_candidates (id, source_id, url, kind, evidence, status, seed_id, reason, kind_class) values (${newId("cand")}, ${sourceId}, ${r.url}, 'HTML', ${r.reason}, 'rejected', ${seed.id}, ${r.reason}, 'other') on conflict (source_id, url) do nothing`);
  }

  await db.execute(sql`update factory_seeds set status = 'discovered', source_id = ${sourceId}, candidates = ${scored.length}, shadow = shadow + ${shadowCreated}, discovered_at = now(), last_error = null, updated_at = now() where id = ${seed.id}`);
  await db.execute(sql`update sources set robots_checked_at = now(), updated_at = now() where id = ${sourceId}`);
  await bumpFactoryDaily({ seeds_processed: 1, requests: disc.requests, candidates: scored.length, shadow_created: shadowCreated, blocked: disc.blocked ? 1 : 0 });
  log.info({ seed: seed.id, source: sourceId, candidates: scored.length, shadow: shadowCreated, requests: disc.requests, ms: Date.now() - started, day }, "factory: organization discovered");
  return { seedId: seed.id, sourceId, status: "discovered", candidates: scored.length, shadow: shadowCreated, requests: disc.requests, durationMs: Date.now() - started, notes };
}

function kindOfExisting(r: { url: string; type: string; connector: string; config: Record<string, unknown> }): CandidateKind {
  const k = (r.config as { kind?: string }).kind as CandidateKind | undefined;
  if (k) return k;
  if (r.connector === "statuspage" || r.connector === "statusjson") return "status";
  if (r.connector === "sitemap") return "sitemap";
  if (r.connector === "github") return "releases";
  if (r.connector === "edgar") return "filings";
  if (r.connector === "dns" || r.connector === "tls" || r.connector === "headers" || r.connector === "rdap") return "posture";
  if (r.connector === "openapi") return "api";
  if (r.connector === "rss") return /press/i.test(r.url) ? "press" : /blog/i.test(r.url) ? "blog" : /security|advisor/i.test(r.url) ? "security" : /changelog|release/i.test(r.url) ? "changelog" : "news";
  if (/pricing|plans/i.test(r.url)) return "pricing";
  if (/terms|privacy|legal/i.test(r.url)) return "legal";
  if (/career|jobs/i.test(r.url)) return "careers";
  if (/changelog|release-notes|whats-new/i.test(r.url)) return "changelog";
  if (/security|advisor/i.test(r.url)) return "security";
  return "other";
}

async function createShadowSensor(sourceId: string, source: SourceRow, c: { url: string; type: string; connector: string; name: string; config: Record<string, unknown>; suggestedTier: Tier; kind: CandidateKind }, candId: string, seed: FactorySeed): Promise<string | null> {
  let id = `${sourceId}_${slugify(c.name)}`.slice(0, 80);
  const clash = (await db.execute<{ url: string }>(sql`select url from sensors where id = ${id}`)).rows[0];
  if (clash && clash.url.replace(/\/$/, "") !== c.url.replace(/\/$/, "")) id = `${id}-${newId("x").slice(-5)}`;
  const tier = c.suggestedTier;
  const priority = 3; // shadows never compete with production sensors
  const jitter = Math.floor(Math.random() * 30 * 60e3);
  try {
    await db
      .insert(sensors)
      .values({ id, sourceId, name: c.name, url: c.url, type: c.type, connector: c.connector, tier, importanceWeight: 1, config: { ...c.config, factory: true, shadow: true, seed: false, kind: c.kind, candidate: candId, seedId: seed.id, shadowSince: new Date().toISOString(), targetPriority: priorityFor(tier, source.categories) }, status: "SHADOW", priority, validatedAt: new Date(), nextCheckAt: new Date(Date.now() + jitter) })
      .onConflictDoNothing();
  } catch (e) {
    log.warn({ sensor: id, err: (e as Error).message }, "factory: shadow sensor insert failed");
    return null;
  }
  return id;
}

export async function bumpFactoryDaily(delta: Partial<Record<"seeds_processed" | "requests" | "candidates" | "shadow_created" | "accepted" | "rejected" | "blocked", number>>): Promise<void> {
  const day = new Date().toISOString().slice(0, 10);
  const d = { seeds_processed: 0, requests: 0, candidates: 0, shadow_created: 0, accepted: 0, rejected: 0, blocked: 0, ...delta };
  await db.execute(sql`insert into factory_daily (day, seeds_processed, requests, candidates, shadow_created, accepted, rejected, blocked) values (${day}, ${d.seeds_processed}, ${d.requests}, ${d.candidates}, ${d.shadow_created}, ${d.accepted}, ${d.rejected}, ${d.blocked})
    on conflict (day) do update set seeds_processed = factory_daily.seeds_processed + ${d.seeds_processed}, requests = factory_daily.requests + ${d.requests}, candidates = factory_daily.candidates + ${d.candidates}, shadow_created = factory_daily.shadow_created + ${d.shadow_created}, accepted = factory_daily.accepted + ${d.accepted}, rejected = factory_daily.rejected + ${d.rejected}, blocked = factory_daily.blocked + ${d.blocked}`).catch(() => undefined);
}

/** Claim up to `n` queued seeds (systemic organizations first). */
export async function claimSeeds(n: number, opts: { seedIds?: string[] } = {}): Promise<FactorySeed[]> {
  const claimed = await db.execute<Record<string, unknown>>(sql`
    update factory_seeds set status = 'discovering', attempts = attempts + 1, updated_at = now()
    where id in (
      select id from factory_seeds
      where ${opts.seedIds ? sql`id = any(${sql.raw("array[" + opts.seedIds.map((i) => "'" + i.replace(/'/g, "''") + "'").join(",") + "]::text[]")})` : sql`(status = 'queued' or (status = 'error' and attempts < 3 and updated_at < now() - interval '6 hours') or (status = 'discovering' and updated_at < now() - interval '15 minutes'))`}
      order by importance desc, created_at asc limit ${n} for update skip locked)
    returning *`);
  return claimed.rows.map(normalizeSeed);
}

/** Run a seed with error capture (status → error, retried up to 3 times by the claim query). */
export async function processSeedSafe(seed: FactorySeed): Promise<ProcessOutcome> {
  try {
    return await processSeed(seed);
  } catch (e) {
    log.error({ seed: seed.id, err: (e as Error).stack ?? (e as Error).message }, "factory: seed failed");
    await db.execute(sql`update factory_seeds set status = 'error', last_error = ${(e as Error).message.slice(0, 500)}, updated_at = now() where id = ${seed.id}`).catch(() => undefined);
    return { seedId: seed.id, sourceId: null, status: "error", candidates: 0, shadow: 0, requests: 0, durationMs: 0, notes: [(e as Error).message] };
  }
}

/** One batch: claim `n` seeds and process them with bounded concurrency (CLI / tests; the process uses a continuous pool). */
export async function runFactoryBatch(n = factoryConfig.concurrency, opts: { seedIds?: string[] } = {}): Promise<ProcessOutcome[]> {
  const seeds = await claimSeeds(n, opts);
  const out: ProcessOutcome[] = [];
  let i = 0;
  await Promise.all(
    Array.from({ length: Math.min(factoryConfig.concurrency, seeds.length) }, async () => {
      while (i < seeds.length) {
        const seed = seeds[i++]!;
        out.push(await processSeedSafe(seed));
      }
    }),
  );
  return out;
}

function normalizeSeed(r: Record<string, unknown>): FactorySeed {
  const d = (v: unknown): Date | null => (v ? new Date(v as string) : null);
  return {
    id: r.id as string,
    name: r.name as string,
    domain: r.domain as string,
    homepage: (r.homepage as string | null) ?? null,
    categories: (r.categories as string[]) ?? [],
    country: (r.country as string | null) ?? null,
    language: (r.language as string | null) ?? null,
    tier: (r.tier as string) ?? "B",
    weight: Number(r.weight ?? 1),
    importance: Number(r.importance ?? 2),
    aliases: (r.aliases as string[]) ?? [],
    firstParty: Boolean(r.first_party ?? true),
    sector: (r.sector as string | null) ?? null,
    universe: (r.universe as string | null) ?? null,
    hints: (r.hints as Record<string, unknown>) ?? {},
    status: r.status as string,
    attempts: Number(r.attempts ?? 0),
    sourceId: (r.source_id as string | null) ?? null,
    candidates: Number(r.candidates ?? 0),
    shadow: Number(r.shadow ?? 0),
    accepted: Number(r.accepted ?? 0),
    rejected: Number(r.rejected ?? 0),
    lastError: (r.last_error as string | null) ?? null,
    discoveredAt: d(r.discovered_at),
    createdAt: d(r.created_at) ?? new Date(),
    updatedAt: d(r.updated_at) ?? new Date(),
  };
}

export async function factoryStats(): Promise<Record<string, unknown>> {
  const [seeds, cands, shadows, daily, sectors] = await Promise.all([
    db.execute<Record<string, unknown>>(sql`select status, count(*)::int as n from factory_seeds group by status`).then((r) => r.rows),
    db.execute<Record<string, unknown>>(sql`select status, count(*)::int as n from discovery_candidates group by status`).then((r) => r.rows),
    db.execute<Record<string, unknown>>(sql`select count(*)::int as shadow, count(*) filter (where total_runs >= ${factoryConfig.shadowMinChecks})::int as ready, count(*) filter (where health <> 'UP')::int as unhealthy from sensors where status = 'SHADOW'`).then((r) => r.rows[0]),
    db.execute<Record<string, unknown>>(sql`select day::text as day, seeds_processed, requests, candidates, shadow_created, accepted, rejected, blocked from factory_daily order by day desc limit 30`).then((r) => r.rows),
    db.execute<Record<string, unknown>>(sql`select coalesce(sector, '-') as sector, count(*)::int as seeds, count(*) filter (where status = 'queued')::int as queued, count(*) filter (where status in ('discovered','done'))::int as discovered, count(*) filter (where status = 'blocked')::int as blocked, sum(shadow)::int as shadow, sum(accepted)::int as accepted, sum(rejected)::int as rejected from factory_seeds group by 1 order by seeds desc`).then((r) => r.rows),
  ]);
  return { seeds: Object.fromEntries(seeds.map((r) => [r.status, r.n])), candidates: Object.fromEntries(cands.map((r) => [r.status, r.n])), shadow: shadows, daily, sectors, config: { concurrency: factoryConfig.concurrency, budget: factoryConfig.budget, minScore: factoryConfig.minScore, shadowMinChecks: factoryConfig.shadowMinChecks, shadowMinHours: factoryConfig.shadowMinHours, shadowMaxHours: factoryConfig.shadowMaxHours, caps: factoryConfig.caps } };
}

export { factorySeeds };