import { readdir, stat } from "node:fs/promises";
import { join, resolve } from "node:path";
import type { FastifyInstance } from "fastify";
import YAML from "yaml";
import { z } from "zod";
import { extendSchema, importDocumentSchema, sourceSchema, type SensorEndpoint, type Tier } from "@websensor/core";
import { getConnector, listConnectors, NormalizeError } from "@websensor/connectors";
import { db, sql, textArray } from "@websensor/db";
import { cacheStats, invalidate } from "./cache";
import { config } from "./config";
import { engineStatus, factoryStatus, liveStats } from "./live";

/**
 * Internal operations (spec §70–72, §85). Enabled only when WS_ADMIN_TOKEN is set; every request
 * must carry `X-WebSensor-Admin: <token>`. Never linked from public pages; `/ops` in the web app
 * asks for the token and stores it in sessionStorage.
 */
export async function registerAdminRoutes(app: FastifyInstance): Promise<void> {
  app.addHook("onRequest", async (req, reply) => {
    if (!req.url.startsWith("/api/v1/admin")) return;
    if (!config.adminToken) return reply.status(404).send({ error: "not_found" });
    const t = req.headers["x-websensor-admin"];
    if (typeof t !== "string" || t !== config.adminToken) return reply.status(401).send({ error: "admin_token_required" });
  });

  app.get("/api/v1/admin/ops", async () => {
    const t0 = Date.now();
    await db.execute(sql`select 1`);
    const dbLatency = Date.now() - t0;
    const [queue, failing, slow, throughput, storage, llm, jobs, es, byStatus, recentErrors] = await Promise.all([
      db.execute<Record<string, unknown>>(sql`select count(*) filter (where next_check_at <= now())::int as due, count(*) filter (where next_check_at <= now() - interval '10 minutes')::int as overdue_10m, count(*) filter (where priority = 0)::int as p0, count(*) filter (where priority = 0 and next_check_at <= now())::int as p0_due, min(next_check_at) as oldest_due from sensors where enabled`).then((r) => r.rows[0]),
      db.execute<Record<string, unknown>>(sql`select split_part(split_part(s.url, '/', 3), ':', 1) as host, count(*)::int as failures, count(distinct s.id)::int as sensors, max(r.error) as last_error, max(r.started_at) as last_at from sensor_runs r join sensors s on s.id = r.sensor_id where r.started_at >= now() - interval '6 hours' and r.outcome in ('error','parse_error','rate_limited') group by 1 order by failures desc limit 20`).then((r) => r.rows),
      db.execute<Record<string, unknown>>(sql`select s.id, s.name, s.source_id, s.url, s.connector, s.avg_latency_ms, s.total_runs from sensors s where s.enabled and s.avg_latency_ms is not null order by s.avg_latency_ms desc limit 20`).then((r) => r.rows),
      db.execute<Record<string, unknown>>(sql`
        select to_char(date_trunc('hour', started_at at time zone 'UTC'), 'YYYY-MM-DD"T"HH24:00:00"Z"') as t, count(*)::int as checks, count(*) filter (where outcome = 'not_modified')::int as not_modified, count(*) filter (where outcome in ('error','parse_error','rate_limited'))::int as errors, count(*) filter (where outcome = 'event')::int as events, avg(duration_ms)::int as avg_ms
        from sensor_runs where started_at >= now() - interval '24 hours' group by 1 order by 1`).then((r) => r.rows),
      Promise.all([
        db.execute<Record<string, unknown>>(sql`select pg_size_pretty(pg_database_size(current_database())) as db_size, pg_database_size(current_database())::bigint as db_bytes, (select count(*) from snapshots)::int as snapshots, (select count(*) from snapshots where storage_key is not null)::int as snapshots_with_raw, (select coalesce(sum(content_length),0) from snapshots where storage_key is not null)::bigint as raw_bytes_uncompressed, (select count(*) from changes)::int as changes, (select count(*) from events)::int as events, (select count(*) from sensor_runs)::int as runs`).then((r) => r.rows[0]),
        blobDirSize(),
      ]).then(([d, blobs]) => ({ ...d, blobs })),
      db.execute<Record<string, unknown>>(sql`select model, count(*)::int as calls, sum(input_tokens)::bigint as input_tokens, sum(output_tokens)::bigint as output_tokens, count(*) filter (where not ok)::int as failures from llm_usage where at >= now() - interval '24 hours' group by model`).then((r) => r.rows),
      db.execute<Record<string, unknown>>(sql`select outcome, count(*)::int as n from sensor_runs where started_at >= now() - interval '1 hour' group by outcome order by n desc`).then((r) => r.rows),
      engineStatus(),
      db.execute<Record<string, unknown>>(sql`select status, health, count(*)::int as n from sensors group by status, health order by n desc`).then((r) => r.rows),
      db.execute<Record<string, unknown>>(sql`select r.started_at, s.id as sensor_id, s.source_id, s.connector, r.http_status, r.outcome, r.error from sensor_runs r join sensors s on s.id = r.sensor_id where r.outcome in ('error','parse_error') and r.started_at >= now() - interval '1 hour' order by r.started_at desc limit 30`).then((r) => r.rows),
    ]);
    return { db: { latency_ms: dbLatency }, queue, engine: es, workers: es ? { inflight: es.inflight, concurrency: es.concurrency, busy_hosts: es.busyHosts, circuit_open: es.circuitOpen } : null, outcomes_1h: jobs, top_failing_domains: failing, slowest_sensors: slow, throughput_24h: throughput, storage, llm_24h: llm, sensors_by_status: byStatus, recent_errors: recentErrors, cache: cacheStats(), live: liveStats(), connectors: listConnectors().map((c) => c.metadata()), generated_at: new Date().toISOString() };
  });

  // ---- Sensor actions ------------------------------------------------------------------------------
  app.post<{ Params: { id: string } }>("/api/v1/admin/sensors/:id/run-now", async (req, reply) => {
    const r = await db.execute(sql`update sensors set next_check_at = now() - interval '1 second', enabled = true where id = ${req.params.id}`);
    if (!r.rowCount) return reply.status(404).send({ error: "not_found" });
    return { ok: true, scheduled: "now" };
  });
  app.post<{ Params: { id: string } }>("/api/v1/admin/sensors/:id/enable", async (req, reply) => {
    const r = await db.execute(sql`update sensors set enabled = true, status = 'ACTIVE', consecutive_errors = 0, next_check_at = now(), updated_at = now() where id = ${req.params.id}`);
    if (!r.rowCount) return reply.status(404).send({ error: "not_found" });
    return { ok: true };
  });
  app.post<{ Params: { id: string } }>("/api/v1/admin/sensors/:id/disable", async (req, reply) => {
    const r = await db.execute(sql`update sensors set enabled = false, status = 'DISABLED', updated_at = now() where id = ${req.params.id}`);
    if (!r.rowCount) return reply.status(404).send({ error: "not_found" });
    return { ok: true };
  });
  app.patch<{ Params: { id: string } }>("/api/v1/admin/sensors/:id", async (req, reply) => {
    const body = z.object({ tier: z.enum(["S", "A", "B", "C", "D"]).optional(), priority: z.number().int().min(0).max(3).optional(), base_interval_seconds: z.number().int().min(15).max(7 * 86400).nullable().optional(), config: z.record(z.string(), z.unknown()).optional(), name: z.string().min(1).max(120).optional() }).parse(req.body ?? {});
    const sets: ReturnType<typeof sql>[] = [sql`updated_at = now()`];
    if (body.tier) sets.push(sql`tier = ${body.tier}`);
    if (body.priority !== undefined) sets.push(sql`priority = ${body.priority}`);
    if (body.base_interval_seconds !== undefined) sets.push(sql`base_interval_seconds = ${body.base_interval_seconds}`);
    if (body.config) sets.push(sql`config = config || ${JSON.stringify(body.config)}::jsonb`);
    if (body.name) sets.push(sql`name = ${body.name}`);
    const r = await db.execute(sql`update sensors set ${sql.join(sets, sql`, `)} where id = ${req.params.id}`);
    if (!r.rowCount) return reply.status(404).send({ error: "not_found" });
    return { ok: true };
  });
  /** Dry-run a connector against a URL (or an existing sensor) without persisting anything: inspect response + normalized output. */
  app.post("/api/v1/admin/sensors/test", async (req, reply) => {
    const body = z.object({ sensor_id: z.string().optional(), url: z.string().url().optional(), connector: z.string().default("http"), type: z.string().default("HTML"), config: z.record(z.string(), z.unknown()).default({}) }).parse(req.body ?? {});
    let endpoint: SensorEndpoint;
    if (body.sensor_id) {
      const s = (await db.execute<Record<string, unknown>>(sql`select * from sensors where id = ${body.sensor_id}`)).rows[0];
      if (!s) return reply.status(404).send({ error: "not_found" });
      endpoint = { id: String(s.id), sourceId: String(s.source_id), name: String(s.name), url: String(s.url), type: String(s.type) as SensorEndpoint["type"], tier: String(s.tier) as Tier, connector: String(s.connector), config: (s.config as Record<string, unknown>) ?? {}, etag: null, lastModified: null, state: null };
    } else if (body.url) {
      endpoint = { id: "admin_test", sourceId: "admin", name: "test", url: body.url, type: body.type as SensorEndpoint["type"], tier: "C", connector: body.connector, config: body.config, etag: null, lastModified: null, state: null };
    } else return reply.status(400).send({ error: "sensor_id_or_url_required" });
    let connector;
    try {
      connector = getConnector(endpoint.connector);
    } catch (e) {
      return reply.status(400).send({ error: "unknown_connector", detail: (e as Error).message });
    }
    const t0 = Date.now();
    const obs = await connector.fetch(endpoint);
    const fetchMs = Date.now() - t0;
    if (obs.error) return { ok: false, stage: "fetch", error: obs.error, meta: obs.meta, fetch_ms: fetchMs };
    if (obs.meta.status >= 400) return { ok: false, stage: "fetch", http_status: obs.meta.status, meta: obs.meta, fetch_ms: fetchMs };
    try {
      const norm = await connector.normalize(endpoint, obs);
      return { ok: true, fetch_ms: fetchMs, meta: obs.meta, normalized: { mode: norm.mode, title: norm.title ?? null, extractionConfidence: norm.extractionConfidence, items: norm.items?.length ?? null, sample_items: norm.items?.slice(0, 5) ?? null, text_preview: norm.text?.slice(0, 1500) ?? null, json_preview: norm.json !== undefined ? JSON.stringify(norm.json).slice(0, 1500) : null, canonicalHash: norm.canonicalHash, publishedAt: norm.publishedAt ?? null } };
    } catch (e) {
      return { ok: false, stage: "normalize", error: e instanceof NormalizeError ? { code: e.code, message: e.message } : { code: "error", message: (e as Error).message }, meta: obs.meta, fetch_ms: fetchMs, body_preview: obs.body?.toString("utf8").slice(0, 800) ?? null };
    }
  });

  // ---- Source actions ------------------------------------------------------------------------------
  app.post<{ Params: { id: string } }>("/api/v1/admin/sources/:id/enable", async (req) => {
    await db.execute(sql`update sources set enabled = true, updated_at = now() where id = ${req.params.id}`);
    invalidate("");
    return { ok: true };
  });
  app.post<{ Params: { id: string } }>("/api/v1/admin/sources/:id/disable", async (req) => {
    await db.execute(sql`update sources set enabled = false, updated_at = now() where id = ${req.params.id}`);
    invalidate("");
    return { ok: true };
  });
  app.post<{ Params: { id: string } }>("/api/v1/admin/sources/:id/run-now", async (req) => {
    const r = await db.execute(sql`update sensors set next_check_at = now() - interval '1 second' where source_id = ${req.params.id} and enabled`);
    return { ok: true, sensors: r.rowCount ?? 0 };
  });
  app.post("/api/v1/admin/sensors/bulk", async (req) => {
    const body = z.object({ ids: z.array(z.string()).min(1).max(2000), action: z.enum(["enable", "disable", "run-now"]) }).parse(req.body ?? {});
    const arr = sql.raw("array[" + body.ids.map((i) => "'" + i.replace(/'/g, "''") + "'").join(",") + "]::text[]");
    let r;
    if (body.action === "enable") r = await db.execute(sql`update sensors set enabled = true, status = 'ACTIVE', consecutive_errors = 0, next_check_at = now() where id = any(${arr})`);
    else if (body.action === "disable") r = await db.execute(sql`update sensors set enabled = false, status = 'DISABLED' where id = any(${arr})`);
    else r = await db.execute(sql`update sensors set next_check_at = now() - interval '1 second' where id = any(${arr}) and enabled`);
    return { ok: true, affected: r.rowCount ?? 0 };
  });

  /**
   * Bulk import (spec §71): JSON body `{ sources: [...] }` or raw YAML (content-type text/yaml). Validates with the
   * registry schema first; `dry_run=1` returns the validation report without writing. Imported sources are stored
   * with `kind = 'registry'` and `config.seed = false` so the file-based sync never disables them.
   */
  app.post<{ Querystring: { dry_run?: string } }>("/api/v1/admin/sources/import", async (req, reply) => {
    let doc: unknown = req.body;
    if (typeof req.body === "string") {
      try {
        doc = YAML.parse(req.body);
      } catch (e) {
        return reply.status(400).send({ error: "yaml_parse_error", detail: (e as Error).message });
      }
    }
    const parsed = importDocumentSchema.safeParse(doc);
    if (!parsed.success) return reply.status(400).send({ error: "invalid_document", issues: parsed.error.issues });
    const report: { id: string | undefined; ok: boolean; extend: boolean; issues?: string[]; sensors?: number }[] = [];
    const existing = new Set((await db.execute<{ id: string }>(sql`select id from sources`)).rows.map((r) => r.id));
    const valid: { extend: boolean; data: z.infer<typeof sourceSchema> | z.infer<typeof extendSchema> }[] = [];
    for (const s of parsed.data.sources) {
      const isExtend = (s as { extend?: boolean })?.extend === true;
      const res = isExtend ? extendSchema.safeParse(s) : sourceSchema.safeParse(s);
      const id = (s as { id?: string })?.id;
      if (!res.success) {
        report.push({ id, ok: false, extend: isExtend, issues: res.error.issues.map((i) => `${i.path.join(".")}: ${i.message}`) });
        continue;
      }
      if (isExtend && !existing.has(res.data.id)) {
        report.push({ id, ok: false, extend: true, issues: ["extend: true but source does not exist"] });
        continue;
      }
      if (!isExtend && existing.has(res.data.id)) {
        report.push({ id, ok: false, extend: false, issues: ["source id already exists (use extend: true)"] });
        continue;
      }
      valid.push({ extend: isExtend, data: res.data });
      report.push({ id, ok: true, extend: isExtend, sensors: res.data.sensors.length });
    }
    if (req.query.dry_run === "1") return { dry_run: true, valid: valid.length, invalid: report.filter((r) => !r.ok).length, report };
    let sensors = 0;
    for (const v of valid) {
      if (!v.extend) {
        const s = v.data as z.infer<typeof sourceSchema>;
        await db.execute(sql`insert into sources (id, name, domain, homepage, description, categories, tier, importance_weight, discover, fallback, notes, enabled, llm_enabled, first_party, country, language, kind)
          values (${s.id}, ${s.name}, ${s.domain}, ${s.homepage ?? `https://${s.domain}`}, ${s.description ?? null}, ${sql.raw("'{" + s.categories.map((c) => '"' + c.replace(/"/g, "") + '"').join(",") + "}'::text[]")}, ${s.tier}, ${s.weight}, ${JSON.stringify(s.discover)}::jsonb, ${JSON.stringify(s.fallback)}::jsonb, ${s.notes ?? null}, ${s.enabled}, ${s.llm}, ${s.first_party ?? true}, ${s.country ?? null}, ${s.language ?? null}, 'registry')`);
        const entId = `org_${s.id}`;
        await db.execute(sql`insert into entities (id, name, type, domain, homepage, description, importance, categories) values (${entId}, ${s.name}, ${s.entity_type}, ${s.domain}, ${s.homepage ?? `https://${s.domain}`}, ${s.description ?? null}, 60, ${sql.raw("'{" + s.categories.map((c) => '"' + c.replace(/"/g, "") + '"').join(",") + "}'::text[]")}) on conflict (id) do nothing`);
        await db.execute(sql`insert into source_entities (source_id, entity_id) values (${s.id}, ${entId}) on conflict do nothing`);
        for (const a of new Set([s.name, s.domain, ...s.aliases])) await db.execute(sql`insert into entity_aliases (alias, entity_id) values (${a.toLowerCase()}, ${entId}) on conflict do nothing`);
      }
      const base = v.data;
      for (const sen of base.sensors) {
        const id = sen.id ?? `${base.id}_${sen.name.toLowerCase().replace(/[^a-z0-9]+/g, "-").replace(/^-|-$/g, "")}`;
        await db.execute(sql`insert into sensors (id, source_id, name, url, type, connector, tier, importance_weight, config, base_interval_seconds, status, priority)
          values (${id}, ${base.id}, ${sen.name}, ${sen.url}, ${sen.type}, ${sen.connector}, ${sen.tier ?? (v.extend ? "B" : (v.data as z.infer<typeof sourceSchema>).tier)}, ${sen.weight ?? 1}, ${JSON.stringify({ ...sen.config, seed: false, imported_at: new Date().toISOString() })}::jsonb, ${sen.interval ?? null}, 'PENDING', 2)
          on conflict (id) do update set name = excluded.name, url = excluded.url, type = excluded.type, connector = excluded.connector, config = excluded.config, enabled = true, updated_at = now()`);
        sensors++;
      }
    }
    invalidate("");
    return { dry_run: false, imported_sources: valid.filter((v) => !v.extend).length, extended_sources: valid.filter((v) => v.extend).length, sensors, invalid: report.filter((r) => !r.ok).length, report };
  });

  // ---- Source Factory ---------------------------------------------------------------------------------
  app.get("/api/v1/admin/factory", async () => {
    const [seeds, cands, shadows, daily, sectors, recent, hb, ready] = await Promise.all([
      db.execute<Record<string, unknown>>(sql`select status, count(*)::int as n from factory_seeds group by status`).then((r) => Object.fromEntries(r.rows.map((x) => [String(x.status), Number(x.n)]))),
      db.execute<Record<string, unknown>>(sql`select status, count(*)::int as n from discovery_candidates group by status`).then((r) => Object.fromEntries(r.rows.map((x) => [String(x.status), Number(x.n)]))),
      db.execute<Record<string, unknown>>(sql`select count(*)::int as total, count(*) filter (where health = 'UP')::int as up, count(*) filter (where health <> 'UP')::int as unhealthy, count(*) filter (where total_runs >= 5)::int as ready, coalesce(sum(raw_changes),0)::int as raw_changes from sensors where status = 'SHADOW' and enabled`).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 = 'discovering')::int as discovering, count(*) filter (where status in ('discovered','done'))::int as discovered, count(*) filter (where status = 'blocked')::int as blocked, count(*) filter (where status = 'error')::int as errors, coalesce(sum(shadow),0)::int as shadow, coalesce(sum(accepted),0)::int as accepted, coalesce(sum(rejected),0)::int as rejected from factory_seeds group by 1 order by seeds desc`).then((r) => r.rows),
      db.execute<Record<string, unknown>>(sql`select id, name, domain, sector, status, candidates, shadow, accepted, rejected, last_error, discovered_at, source_id from factory_seeds where status <> 'queued' order by updated_at desc limit 40`).then((r) => r.rows),
      factoryStatus(),
      db.execute<Record<string, unknown>>(sql`select kind_class, count(*)::int as n from discovery_candidates where status = 'shadow' group by 1 order by n desc`).then((r) => r.rows),
    ]);
    return { seeds, candidates: cands, shadow: shadows, shadow_by_kind: ready, daily, sectors, recent_seeds: recent, heartbeat: hb, generated_at: new Date().toISOString() };
  });
  app.get<{ Querystring: { status?: string; sector?: string; q?: string; limit?: string } }>("/api/v1/admin/factory/seeds", async (req) => {
    const lim = Math.min(1000, Number(req.query.limit ?? 200));
    const rows = await db.execute<Record<string, unknown>>(sql`select * from factory_seeds where 1=1 ${req.query.status ? sql`and status = ${req.query.status}` : sql``} ${req.query.sector ? sql`and sector = ${req.query.sector}` : sql``} ${req.query.q ? sql`and (name ilike ${"%" + req.query.q + "%"} or domain ilike ${"%" + req.query.q + "%"})` : sql``} order by importance desc, updated_at desc limit ${lim}`);
    return { items: rows.rows };
  });
  /** Bulk seeds `{ seeds: [{ name, domain, categories?, country?, tier?, importance?, aliases?, sector?, universe?, hints? }] }` — queued for the factory process. */
  app.post("/api/v1/admin/factory/seeds", async (req) => {
    const seedSchema = z.object({ id: z.string().regex(/^[a-z0-9][a-z0-9-]*$/).optional(), name: z.string().min(1), domain: z.string().min(3), categories: z.array(z.string()).default([]), country: z.string().regex(/^[A-Z]{2,3}$/).optional(), language: z.string().optional(), tier: z.enum(["S", "A", "B", "C", "D"]).default("B"), importance: z.number().int().min(1).max(3).default(2), aliases: z.array(z.string()).default([]), first_party: z.boolean().default(true), sector: z.string().optional(), universe: z.string().optional(), hints: z.record(z.string(), z.unknown()).default({}) });
    const body = z.object({ seeds: z.array(seedSchema).min(1).max(2000) }).parse(req.body ?? {});
    let n = 0;
    for (const s of body.seeds) {
      const domain = s.domain.toLowerCase().replace(/^https?:\/\//, "").replace(/^www\./, "").replace(/\/.*$/, "");
      const id = s.id ?? domain.replace(/[^a-z0-9]+/g, "-").replace(/^-|-$/g, "").slice(0, 60);
      await db.execute(sql`insert into factory_seeds (id, name, domain, categories, country, language, tier, importance, aliases, first_party, sector, universe, hints, status)
        values (${id}, ${s.name}, ${domain}, ${textArray(s.categories)}, ${s.country ?? null}, ${s.language ?? null}, ${s.tier}, ${s.importance}, ${textArray(s.aliases)}, ${s.first_party}, ${s.sector ?? null}, ${s.universe ?? null}, ${JSON.stringify(s.hints)}::jsonb, 'queued')
        on conflict (id) do update set name = excluded.name, hints = factory_seeds.hints || excluded.hints, status = 'queued', updated_at = now()`);
      n++;
    }
    invalidate("coverage");
    return { ok: true, queued: n };
  });
  app.post<{ Params: { id: string } }>("/api/v1/admin/factory/seeds/:id/requeue", async (req, reply) => {
    const r = await db.execute(sql`update factory_seeds set status = 'queued', updated_at = now() where id = ${req.params.id}`);
    if (!r.rowCount) return reply.status(404).send({ error: "not_found" });
    return { ok: true };
  });
  app.get<{ Querystring: { status?: string; seed?: string; source?: string; kind?: string; limit?: string } }>("/api/v1/admin/factory/candidates", async (req) => {
    const lim = Math.min(1000, Number(req.query.limit ?? 200));
    const rows = await db.execute<Record<string, unknown>>(sql`select c.*, s.health as shadow_health, s.total_runs as shadow_runs, s.raw_changes as shadow_changes from discovery_candidates c left join sensors s on s.id = c.shadow_sensor_id where 1=1 ${req.query.status ? sql`and c.status = ${req.query.status}` : sql``} ${req.query.seed ? sql`and c.seed_id = ${req.query.seed}` : sql``} ${req.query.source ? sql`and c.source_id = ${req.query.source}` : sql``} ${req.query.kind ? sql`and c.kind_class = ${req.query.kind}` : sql``} order by c.updated_at desc limit ${lim}`);
    return { items: rows.rows };
  });
  /** Manual decision on a shadow sensor: accept now (→ ACTIVE) or reject (→ DISABLED). */
  app.post<{ Params: { id: string; action: string } }>("/api/v1/admin/factory/shadow/:id/:action", async (req, reply) => {
    const s = (await db.execute<Record<string, unknown>>(sql`select id, config, status from sensors where id = ${req.params.id}`)).rows[0];
    if (!s) return reply.status(404).send({ error: "not_found" });
    const reason = `manual ${req.params.action} via admin API`;
    if (req.params.action === "accept") {
      const tp = Number(((s.config as Record<string, unknown>).targetPriority as number | undefined) ?? 2);
      await db.execute(sql`update sensors set status = 'ACTIVE', enabled = true, priority = ${tp}, config = (config - 'shadow') || jsonb_build_object('acceptedAt', now()::text, 'shadowReport', ${reason}::text), updated_at = now() where id = ${req.params.id}`);
      await db.execute(sql`update discovery_candidates set status = 'accepted', reason = ${reason}, decided_at = now(), updated_at = now() where shadow_sensor_id = ${req.params.id}`);
    } else if (req.params.action === "reject") {
      await db.execute(sql`update sensors set status = 'DISABLED', enabled = false, config = config || jsonb_build_object('rejectedAt', now()::text, 'shadowReport', ${reason}::text), updated_at = now() where id = ${req.params.id}`);
      await db.execute(sql`update discovery_candidates set status = 'rejected', reason = ${reason}, decided_at = now(), updated_at = now() where shadow_sensor_id = ${req.params.id}`);
    } else return reply.status(400).send({ error: "action must be accept or reject" });
    invalidate("");
    return { ok: true };
  });
  /** Promote a plain candidate (never shadowed) straight into shadow monitoring. */
  app.post<{ Params: { id: string } }>("/api/v1/admin/factory/candidates/:id/shadow", async (req, reply) => {
    const c = (await db.execute<Record<string, unknown>>(sql`select * from discovery_candidates where id = ${req.params.id}`)).rows[0];
    if (!c) return reply.status(404).send({ error: "not_found" });
    if (!c.connector || !c.type) return reply.status(400).send({ error: "candidate has no connector/type (legacy discovery row)" });
    const src = (await db.execute<Record<string, unknown>>(sql`select categories from sources where id = ${String(c.source_id)}`)).rows[0];
    const id = `${String(c.source_id)}_${String(c.name ?? c.kind_class ?? "page").toLowerCase().replace(/[^a-z0-9]+/g, "-").replace(/^-|-$/g, "")}-${String(c.id).slice(-5)}`.slice(0, 80);
    await db.execute(sql`insert into sensors (id, source_id, name, url, type, connector, tier, config, status, priority, validated_at)
      values (${id}, ${String(c.source_id)}, ${String(c.name ?? c.kind_class ?? "page")}, ${String(c.url)}, ${String(c.type)}, ${String(c.connector)}, ${String(c.tier ?? "C")}, ${JSON.stringify({ ...((c.config as Record<string, unknown>) ?? {}), factory: true, shadow: true, seed: false, kind: c.kind_class, candidate: c.id, seedId: c.seed_id, shadowSince: new Date().toISOString(), targetPriority: 2, categories: src?.categories ?? [] })}::jsonb, 'SHADOW', 3, now())
      on conflict (id) do nothing`);
    await db.execute(sql`update discovery_candidates set status = 'shadow', shadow_sensor_id = ${id}, updated_at = now() where id = ${req.params.id}`);
    return { ok: true, sensor_id: id };
  });

  app.get("/api/v1/admin/failures", async () => {
    const rows = await db.execute<Record<string, unknown>>(sql`select r.started_at, r.outcome, r.http_status, r.error, r.duration_ms, s.id as sensor_id, s.name, s.source_id, s.url, s.connector from sensor_runs r join sensors s on s.id = r.sensor_id where r.outcome in ('error','parse_error','rate_limited') and r.started_at >= now() - interval '24 hours' order by r.started_at desc limit 200`);
    return { items: rows.rows };
  });
}

async function blobDirSize(): Promise<{ files: number; bytes: number } | null> {
  const root = resolve(process.env.BLOB_STORE_DIR ?? "./data/blobs");
  let files = 0;
  let bytes = 0;
  const started = Date.now();
  const walk = async (dir: string): Promise<void> => {
    if (Date.now() - started > 4000) return; // bounded
    let entries: string[];
    try {
      entries = await readdir(dir);
    } catch {
      return;
    }
    for (const e of entries) {
      const p = join(dir, e);
      const st = await stat(p).catch(() => null);
      if (!st) continue;
      if (st.isDirectory()) await walk(p);
      else {
        files++;
        bytes += st.size;
      }
    }
  };
  await walk(root);
  return { files, bytes };
}