import type { FastifyInstance } from "fastify";
import { z } from "zod";
import { COUNTRIES, countryBySlug, diffText, EVENT_GROUPS, EVENT_TYPES, FEED_CHANNELS, parseSearch } from "@websensor/core";
import { db, sql, textArray } from "@websensor/db";
import { getBlobStore } from "@websensor/store";
import { cached } from "./cache";
import { config } from "./config";
import { coverageSector, coverageSummary } from "./coverage";
import { breakingDesk, categoryDesk, clusterDetail, countryDesk, countryList, entityInsights, entityRankings, pulse, radar, sourceSensors } from "./intel";
import { engineStatus, liveStats } from "./live";
import { countEvents, getEvent, listEvents, relatedEvents, sensorHistory, sourceActivity, sourceQuality, stats, trending } from "./queries";

const bool = z
  .enum(["true", "false", "1", "0"])
  .transform((v) => v === "true" || v === "1")
  .optional();

export const eventsQuery = z.object({
  after: z.string().optional(),
  before: z.string().optional(),
  category: z.string().optional(),
  entity: z.string().optional(),
  source: z.string().optional(),
  domain: z.string().optional(),
  sensor: z.string().optional(),
  cluster: z.string().optional(),
  importance_min: z.coerce.number().min(0).max(100).optional(),
  confidence_min: z.coerce.number().min(0).max(100).optional(),
  signal_min: z.coerce.number().min(0).max(100).optional(),
  event_type: z.string().optional(),
  group: z.string().optional(),
  silent_change: bool,
  first_party: bool,
  confirmed: bool,
  country: z.string().max(3).optional(),
  language: z.string().max(2).optional(),
  change_class: z.string().optional(),
  q: z.string().max(300).optional(),
  limit: z.coerce.number().int().min(1).max(200).default(50),
  cursor: z.string().optional(),
  order: z.enum(["recent", "importance", "signal"]).default("recent"),
});

export async function registerRoutes(app: FastifyInstance): Promise<void> {
  // ---- Health ---------------------------------------------------------------------------
  app.get("/api/health", async () => ({ status: "ok", service: "api", version: config.version, time: new Date().toISOString() }));
  app.get("/api/ready", async (_req, reply) => {
    const checks: Record<string, boolean | number | string | null> = {};
    const t0 = Date.now();
    try {
      await db.execute(sql`select 1`);
      checks.database = true;
      checks.database_latency_ms = Date.now() - t0;
    } catch {
      checks.database = false;
    }
    const engine = await db.execute<{ last: Date | null }>(sql`select max(started_at) as last from sensor_runs where started_at >= now() - interval '15 minutes'`).catch(() => ({ rows: [{ last: null }] }));
    checks.engine_recent = Boolean(engine.rows[0]?.last);
    const es = await engineStatus();
    checks.engine_heartbeat = es ? String(es.at) : null;
    checks.ws_clients = liveStats().clients;
    return reply.status(checks.database ? 200 : 503).send({ status: checks.database ? "ready" : "degraded", checks });
  });

  // ---- Events ---------------------------------------------------------------------------
  app.get("/api/v1/events", async (req, reply) => {
    const q = eventsQuery.parse(req.query);
    const page = await listEvents(q);
    reply.header("x-websensor-order", q.order);
    return { ...page, meta: { limit: q.limit, order: q.order, filters: stripEmpty({ ...q, limit: undefined, cursor: undefined, order: undefined }) } };
  });
  app.get("/api/v1/events/count", async (req) => {
    const q = eventsQuery.parse(req.query);
    return { count: await countEvents(q) };
  });
  app.get<{ Params: { id: string } }>("/api/v1/events/:id", async (req, reply) => {
    const ev = await getEvent(req.params.id);
    if (!ev) return reply.status(404).send({ error: "not_found" });
    const [related, cluster, change, interp, history] = await Promise.all([
      relatedEvents(ev),
      ev.cluster_id ? db.execute<Record<string, unknown>>(sql`select * from event_clusters where id = ${String(ev.cluster_id)}`).then((r) => r.rows[0] ?? null) : null,
      ev.change_id ? db.execute<Record<string, unknown>>(sql`select id, kind, diff, signal, noise_ratio, magnitude, heuristic, detected_at, old_snapshot_id, new_snapshot_id, change_class, field_changes from changes where id = ${String(ev.change_id)}`).then((r) => r.rows[0] ?? null) : null,
      db.execute<Record<string, unknown>>(sql`select version, model, created_at from interpretations where event_id = ${String(ev.id)} order by version`).then((r) => r.rows),
      sensorHistory(String(ev.sensor_id), String(ev.id)),
    ]);
    const snaps = await db.execute<Record<string, unknown>>(sql`select id, url, captured_at, http_status, content_type, content_length, content_hash, canonical_hash, etag, last_modified, title, mode, fetch_duration_ms, extraction_confidence, storage_key is not null as has_raw from snapshots where id in (${String(ev.old_snapshot_id ?? "")}, ${String(ev.new_snapshot_id ?? "")})`);
    const sourceRel = await db.execute<Record<string, unknown>>(sql`select health, success_rate, avg_latency_ms, total_runs, raw_changes, meaningful_changes, last_check_at from (select s.health, ch.success_rate, s.avg_latency_ms, s.total_runs, s.raw_changes, s.meaningful_changes, s.last_check_at from sensors s left join connector_health ch on ch.connector = s.connector where s.id = ${String(ev.sensor_id)}) x`);
    return { event: ev, related, cluster, change, interpretations: interp, snapshots: snaps.rows, sensor_reliability: sourceRel.rows[0] ?? null, history };
  });

  // ---- Changes / snapshots / diffs -------------------------------------------------------
  app.get<{ Params: { id: string } }>("/api/v1/changes/:id", async (req, reply) => {
    const r = await db.execute<Record<string, unknown>>(sql`select c.*, s.url as sensor_url, s.name as sensor_name, s.source_id from changes c join sensors s on s.id = c.sensor_id where c.id = ${req.params.id}`);
    const c = r.rows[0];
    if (!c) return reply.status(404).send({ error: "not_found" });
    let unified: string | null = null;
    if (c.diff_storage_key) unified = await getBlobStore().getText(String(c.diff_storage_key)).catch(() => null);
    return { change: c, unified };
  });
  app.get<{ Params: { id: string }; Querystring: { raw?: string } }>("/api/v1/snapshots/:id", async (req, reply) => {
    const r = await db.execute<Record<string, unknown>>(sql`select * from snapshots where id = ${req.params.id}`);
    const s = r.rows[0];
    if (!s) return reply.status(404).send({ error: "not_found" });
    const store = getBlobStore();
    if (req.query.raw === "1") {
      if (!s.storage_key) return reply.status(410).send({ error: "raw_body_pruned", detail: "The raw body of this snapshot was pruned by the retention policy; the canonical representation and hashes are preserved." });
      const buf = await store.get(String(s.storage_key));
      reply.header("content-type", String(s.content_type ?? "text/plain") + (String(s.content_type ?? "").includes("charset") ? "" : "; charset=utf-8"));
      reply.header("x-content-type-options", "nosniff");
      reply.header("content-security-policy", "default-src 'none'; style-src 'unsafe-inline'; img-src data:");
      return reply.send(buf);
    }
    const canonical = s.canonical_storage_key ? await store.getText(String(s.canonical_storage_key)).catch(() => null) : null;
    return { snapshot: s, canonical: canonical ? JSON.parse(canonical) : null };
  });
  app.get<{ Querystring: { a: string; b: string } }>("/api/v1/snapshots/compare", async (req, reply) => {
    const { a, b } = req.query;
    if (!a || !b) return reply.status(400).send({ error: "a and b required" });
    const rows = await db.execute<Record<string, unknown>>(sql`select id, captured_at, canonical_storage_key, mode, url, sensor_id from snapshots where id in (${a}, ${b})`);
    const sa = rows.rows.find((r) => r.id === a);
    const sb = rows.rows.find((r) => r.id === b);
    if (!sa || !sb) return reply.status(404).send({ error: "not_found" });
    const store = getBlobStore();
    const [ca, cb] = await Promise.all([store.getText(String(sa.canonical_storage_key)), store.getText(String(sb.canonical_storage_key))]);
    const ta = renderCanonical(JSON.parse(ca));
    const tb = renderCanonical(JSON.parse(cb));
    const d = diffText(ta, tb, `${a}@${String(sa.captured_at)}`, `${b}@${String(sb.captured_at)}`);
    return { a: sa, b: sb, before: ta, after: tb, diff: { unified: d.unified, stats: d.stats, added: d.added, removed: d.removed, modified: d.modified } };
  });
  /** Historical memory (spec §79): how a sensor's page looked at any point — list of snapshots with day grouping. */
  app.get<{ Params: { id: string }; Querystring: { limit?: string; before?: string } }>("/api/v1/sensors/:id/snapshots", async (req) => {
    const lim = Math.min(500, Number(req.query.limit ?? 200));
    const rows = await db.execute<Record<string, unknown>>(sql`select id, captured_at, http_status, content_length, canonical_hash, title, mode, storage_key is not null as has_raw, (select count(*) from changes c where c.new_snapshot_id = snapshots.id) > 0 as has_change, (select e.slug from events e where e.new_snapshot_id = snapshots.id limit 1) as event_slug from snapshots where sensor_id = ${req.params.id} ${req.query.before ? sql`and captured_at < ${new Date(req.query.before)}` : sql``} order by captured_at desc limit ${lim}`);
    return { items: rows.rows };
  });

  // ---- Sources & sensors ------------------------------------------------------------------
  app.get<{ Querystring: { category?: string; q?: string; limit?: string; country?: string; tier?: string; first_party?: string } }>("/api/v1/sources", async (req) => {
    const cat = req.query.category;
    const q = req.query.q;
    const rows = await db.execute<Record<string, unknown>>(sql`
      select s.id, s.name, s.domain, s.homepage, s.description, s.categories, s.tier, s.importance_weight, s.enabled, s.notes, s.first_party, s.country, s.language, s.robots_checked_at,
        (select count(*) from sensors x where x.source_id = s.id and x.enabled and x.status <> 'SHADOW')::int as sensor_count,
        (select count(*) from sensors x where x.source_id = s.id and x.enabled and x.status = 'SHADOW')::int as shadow_count, s.origin, s.sector,
        (select count(*) from events e where e.source_id = s.id)::int as event_count,
        (select count(*) from events e where e.source_id = s.id and e.detected_at >= now() - interval '24 hours')::int as events_24h,
        (select max(detected_at) from events e where e.source_id = s.id) as last_event_at,
        (select max(last_check_at) from sensors x where x.source_id = s.id) as last_check_at,
        (select count(*) from sensors x where x.source_id = s.id and x.enabled and x.health <> 'UP')::int as sensors_degraded
      from sources s where s.enabled and s.kind = 'registry' ${cat ? sql`and ${cat} = any(s.categories)` : sql``} ${q ? sql`and (s.name ilike ${"%" + q + "%"} or s.domain ilike ${"%" + q + "%"})` : sql``} ${req.query.country ? sql`and s.country = ${req.query.country.toUpperCase()}` : sql``} ${req.query.tier ? sql`and s.tier = ${req.query.tier.toUpperCase()}` : sql``} ${req.query.first_party ? sql`and s.first_party = ${req.query.first_party === "true"}` : sql``}
      order by s.tier asc, events_24h desc, s.name asc limit ${Math.min(5000, Number(req.query.limit ?? 300))}`);
    return { items: rows.rows };
  });
  app.get<{ Params: { id: string } }>("/api/v1/sources/:id", async (req, reply) => {
    const s = (await db.execute<Record<string, unknown>>(sql`select * from sources where (id = ${req.params.id} or domain = ${req.params.id}) and kind = 'registry' limit 1`)).rows[0];
    if (!s) return reply.status(404).send({ error: "not_found" });
    const { owner_token: _o, ...pub } = s;
    const [sensors, ents, activity, quality, candidates, byType, series] = await Promise.all([
      sourceSensors(String(s.id)),
      db.execute<Record<string, unknown>>(sql`select en.id, en.name, en.type, en.importance, en.event_count from source_entities se join entities en on en.id = se.entity_id where se.source_id = ${String(s.id)} order by en.type, en.name`).then((r) => r.rows),
      sourceActivity(String(s.id)),
      sourceQuality(String(s.id)),
      db.execute<Record<string, unknown>>(sql`select url, kind, evidence, score, status, found_at from discovery_candidates where source_id = ${String(s.id)} order by (score->>'value')::float desc nulls last limit 50`).then((r) => r.rows),
      db.execute<Record<string, unknown>>(sql`select event_type, count(*)::int as n from events where source_id = ${String(s.id)} group by 1 order by 2 desc limit 16`).then((r) => r.rows),
      db.execute<Record<string, unknown>>(sql`select day::text as day, checks, not_modified, errors, raw_changes, events from source_daily where source_id = ${String(s.id)} and day >= (now() at time zone 'UTC')::date - 30 order by day`).then((r) => r.rows),
    ]);
    return { source: pub, sensors, entities: ents, activity, quality, discovery: candidates, by_type: byType, daily: series };
  });
  app.get<{ Params: { id: string } }>("/api/v1/sensors/:id", async (req, reply) => {
    const s = (await db.execute<Record<string, unknown>>(sql`select s.*, so.name as source_name, so.domain, so.kind as source_kind from sensors s join sources so on so.id = s.source_id where s.id = ${req.params.id}`)).rows[0];
    if (!s || s.source_kind === "custom") return reply.status(404).send({ error: "not_found" });
    const [runs, snaps, changes, events] = await Promise.all([
      db.execute<Record<string, unknown>>(sql`select id, started_at, finished_at, http_status, outcome, error, duration_ms, bytes, fetch_method, snapshot_id from sensor_runs where sensor_id = ${String(s.id)} order by started_at desc limit 50`).then((r) => r.rows),
      db.execute<Record<string, unknown>>(sql`select id, captured_at, http_status, content_type, content_length, canonical_hash, title, mode, extraction_confidence, storage_key is not null as has_raw from snapshots where sensor_id = ${String(s.id)} order by captured_at desc limit 50`).then((r) => r.rows),
      db.execute<Record<string, unknown>>(sql`select id, detected_at, kind, signal, noise_ratio, magnitude, meaningful, event_id, old_snapshot_id, new_snapshot_id, change_class, field_changes, heuristic->>'eventType' as heuristic_type from changes where sensor_id = ${String(s.id)} order by detected_at desc limit 50`).then((r) => r.rows),
      listEvents({ sensor: String(s.id), limit: 30 }).then((r) => r.items),
    ]);
    const { state: _st, source_kind: _k, ...pub } = s as Record<string, unknown>;
    const runs24 = await db.execute<Record<string, unknown>>(sql`select count(*)::int as checks_24h, count(*) filter (where outcome = 'not_modified')::int as not_modified_24h, count(*) filter (where outcome in ('error','parse_error','rate_limited'))::int as errors_24h, avg(duration_ms)::int as avg_ms_24h from sensor_runs where sensor_id = ${String(s.id)} and started_at >= now() - interval '24 hours'`);
    return { sensor: { ...pub, ...runs24.rows[0] }, runs, snapshots: snaps, changes, events };
  });

  // ---- Entities ----------------------------------------------------------------------------
  app.get<{ Querystring: { type?: string; q?: string; limit?: string; category?: string } }>("/api/v1/entities", async (req) => {
    const rows = await db.execute<Record<string, unknown>>(sql`
      select en.*, (select count(*) from events e join event_entities ee on ee.event_id = e.id where ee.entity_id = en.id and e.detected_at >= now() - interval '24 hours')::int as events_24h
      from entities en where true ${req.query.type ? sql`and en.type = ${req.query.type}` : sql``} ${req.query.category ? sql`and ${req.query.category} = any(en.categories)` : sql``} ${req.query.q ? sql`and (en.search @@ plainto_tsquery('simple', ${req.query.q}) or en.name ilike ${"%" + req.query.q + "%"})` : sql``}
      order by en.event_count desc, en.importance desc, en.name limit ${Math.min(1000, Number(req.query.limit ?? 200))}`);
    return { items: rows.rows };
  });
  app.get<{ Querystring: { limit?: string } }>("/api/v1/entities/rank", async (req) => ({ items: await entityRankings(Math.min(500, Number(req.query.limit ?? 100))) }));
  app.get<{ Params: { id: string } }>("/api/v1/entities/:id", async (req, reply) => {
    const e = (await db.execute<Record<string, unknown>>(sql`select * from entities where id = ${req.params.id} or id = ${"org_" + req.params.id} limit 1`)).rows[0];
    if (!e) return reply.status(404).send({ error: "not_found" });
    const id = String(e.id);
    const [children, relations, sources, aliases, recent, byType, insights, silent, parent] = await Promise.all([
      db.execute<Record<string, unknown>>(sql`select id, name, type, importance, event_count, last_event_at from entities where parent_id = ${id} order by event_count desc, name`).then((r) => r.rows),
      db.execute<Record<string, unknown>>(sql`select r.relation, r.from_id, r.to_id, f.name as from_name, t.name as to_name, t.type as to_type from entity_relations r join entities f on f.id = r.from_id join entities t on t.id = r.to_id where r.from_id = ${id} or r.to_id = ${id} limit 100`).then((r) => r.rows),
      db.execute<Record<string, unknown>>(sql`select s.id, s.name, s.domain, s.tier, s.first_party, s.country, (select count(*) from sensors x where x.source_id = s.id and x.enabled)::int as sensor_count, (select count(*) from events ev where ev.source_id = s.id and ev.detected_at >= now() - interval '24 hours')::int as events_24h from source_entities se join sources s on s.id = se.source_id where se.entity_id = ${id} and s.kind = 'registry'`).then((r) => r.rows),
      db.execute<{ alias: string }>(sql`select alias from entity_aliases where entity_id = ${id} order by alias`).then((r) => r.rows.map((x) => x.alias)),
      listEvents({ entity: id, limit: 30 }),
      db.execute<Record<string, unknown>>(sql`select e.event_type, count(*)::int as n from events e join event_entities ee on ee.event_id = e.id where ee.entity_id = ${id} group by 1 order by 2 desc`).then((r) => r.rows),
      entityInsights(id),
      listEvents({ entity: id, silent_change: true, limit: 10 }).then((r) => r.items),
      e.parent_id ? db.execute<Record<string, unknown>>(sql`select id, name, type from entities where id = ${String(e.parent_id)}`).then((r) => r.rows[0] ?? null) : null,
    ]);
    // Related entities: co-occurring in events over 30 days
    const related = await db.execute<Record<string, unknown>>(sql`
      select en.id, en.name, en.type, count(*)::int as shared_events from event_entities a join event_entities b on a.event_id = b.event_id and b.entity_id <> a.entity_id join events e on e.id = a.event_id join entities en on en.id = b.entity_id
      where a.entity_id = ${id} and e.detected_at >= now() - interval '30 days' group by en.id, en.name, en.type order by shared_events desc limit 12`).then((r) => r.rows);
    return { entity: e, parent, children, relations, sources, aliases, recent: recent.items, nextCursor: recent.nextCursor, by_type: byType, insights, silent, related };
  });
  app.get<{ Params: { id: string }; Querystring: { limit?: string; cursor?: string; after?: string; before?: string; event_type?: string; silent_change?: string } }>("/api/v1/entities/:id/timeline", async (req) => {
    const id = req.params.id.startsWith("org_") || req.params.id.startsWith("prd_") ? req.params.id : `org_${req.params.id}`;
    return listEvents({ entity: id, limit: Math.min(200, Number(req.query.limit ?? 100)), cursor: req.query.cursor, after: req.query.after, before: req.query.before, event_type: req.query.event_type, silent_change: req.query.silent_change === "true" ? true : undefined });
  });

  // ---- Clusters ------------------------------------------------------------------------------
  app.get<{ Querystring: { limit?: string; since?: string; state?: string; min_events?: string } }>("/api/v1/clusters", async (req) => {
    const rows = await db.execute<Record<string, unknown>>(sql`
      select c.*, (select json_agg(json_build_object('id', e.id, 'slug', e.slug, 'title', e.title, 'importance', e.importance, 'event_type', e.event_type, 'detected_at', e.detected_at, 'source_id', e.source_id, 'url', e.url, 'first_party', e.first_party) order by e.importance desc) from events e where e.cluster_id = c.id) as events,
        (select json_build_object('id', s.id, 'name', s.name, 'domain', s.domain) from events e join sources s on s.id = e.source_id where e.id = c.primary_event_id) as source
      from event_clusters c where c.last_at >= now() - make_interval(hours => ${Math.min(720, Number(req.query.since ?? 72))}) ${req.query.state ? sql`and c.state = ${req.query.state}` : sql``} ${req.query.min_events ? sql`and c.event_count >= ${Number(req.query.min_events)}` : sql``} order by c.last_at desc limit ${Math.min(200, Number(req.query.limit ?? 50))}`);
    return { items: rows.rows };
  });
  app.get<{ Params: { id: string } }>("/api/v1/clusters/:id", async (req, reply) => {
    const d = await clusterDetail(req.params.id);
    if (!d) return reply.status(404).send({ error: "not_found" });
    return d;
  });

  // ---- Domains / URLs ----------------------------------------------------------------------
  app.get<{ Params: { domain: string }; Querystring: { limit?: string; cursor?: string } }>("/api/v1/domains/:domain/timeline", async (req) => {
    const d = req.params.domain.toLowerCase();
    const src = (await db.execute<{ id: string }>(sql`select id from sources where (domain = ${d} or domain = ${"www." + d} or ${d} = 'www.' || domain) and kind = 'registry' limit 1`)).rows[0];
    const urls = await db.execute<Record<string, unknown>>(sql`select url, status, first_seen_at, last_seen_at, snapshot_count, change_count, sensor_id from urls where domain = ${d} or domain like ${"%." + d} order by change_count desc, last_seen_at desc limit 200`);
    const ev = src ? await listEvents({ source: src.id, limit: Math.min(200, Number(req.query.limit ?? 100)), cursor: req.query.cursor }) : { items: [], nextCursor: null };
    return { domain: d, source_id: src?.id ?? null, urls: urls.rows, events: ev.items, nextCursor: ev.nextCursor };
  });
  app.get<{ Querystring: { url: string } }>("/api/v1/urls/history", async (req, reply) => {
    if (!req.query.url) return reply.status(400).send({ error: "url required" });
    const u = (await db.execute<Record<string, unknown>>(sql`select * from urls where url = ${req.query.url}`)).rows[0];
    const history = await db.execute<Record<string, unknown>>(sql`
      select h.id, h.at, h.kind, h.snapshot_id, h.change_id, h.event_id, h.note, e.title as event_title, e.importance, e.event_type, e.slug as event_slug, c.kind as change_kind, c.signal, c.change_class
      from url_history h left join events e on e.id = h.event_id left join changes c on c.id = h.change_id where h.url = ${req.query.url} order by h.at desc limit 300`);
    const snaps = await db.execute<Record<string, unknown>>(sql`select id, captured_at, http_status, content_length, canonical_hash, title, storage_key is not null as has_raw from snapshots where url = ${req.query.url} or sensor_id = (select sensor_id from urls where url = ${req.query.url}) order by captured_at desc limit 200`);
    return { url: u ?? { url: req.query.url }, history: history.rows, snapshots: snaps.rows };
  });

  // ---- Intelligence desks ----------------------------------------------------------------------
  app.get("/api/v1/stats", async () => cached("stats", 5_000, stats));
  // ---- Global Observation Coverage Score ------------------------------------------------------
  app.get("/api/v1/coverage", async () => coverageSummary());
  app.get<{ Params: { sector: string } }>("/api/v1/coverage/:sector", async (req, reply) => {
    const r = await coverageSector(req.params.sector.toLowerCase());
    if (!r) return reply.status(404).send({ error: "not_found" });
    return r;
  });
  app.get<{ Querystring: { hours?: string; limit?: string } }>("/api/v1/trending", async (req) => {
    const hours = Math.min(168, Number(req.query.hours ?? 24));
    const limit = Math.min(50, Number(req.query.limit ?? 12));
    return { items: await cached(`trending:${hours}:${limit}`, 15_000, () => trending(hours, limit)) };
  });
  app.get("/api/v1/breaking", async () => breakingDesk());
  app.get("/api/v1/pulse", async () => pulse());
  app.get("/api/v1/radar", async () => radar());
  app.get("/api/v1/countries", async () => ({ items: await countryList(), known: COUNTRIES }));
  app.get<{ Params: { code: string } }>("/api/v1/countries/:code", async (req, reply) => {
    const c = countryBySlug(req.params.code) ?? (COUNTRIES[req.params.code.toUpperCase()] ? { code: req.params.code.toUpperCase() } : null);
    if (!c) return reply.status(404).send({ error: "not_found" });
    return cached(`country:${c.code}`, 20_000, () => countryDesk(c.code));
  });
  app.get<{ Params: { channel: string } }>("/api/v1/categories/:channel", async (req, reply) => {
    const ch = req.params.channel.toLowerCase();
    if (!FEED_CHANNELS[ch] && !/^[a-z-]{2,30}$/.test(ch)) return reply.status(404).send({ error: "not_found" });
    return cached(`category:${ch}`, 15_000, () => categoryDesk(ch));
  });
  app.get("/api/v1/explore", async () =>
    cached("explore", 20_000, async () => {
      const since48 = new Date(Date.now() - 48 * 3600e3).toISOString();
      const [mostActive, biggest, silent, clusters, unusual, byType, byCategory, newly, firstParty] = await Promise.all([
        db.execute<Record<string, unknown>>(sql`select s.id, s.name, s.domain, s.first_party, count(*)::int as events_24h, max(e.importance)::float as max_importance from events e join sources s on s.id = e.source_id where e.detected_at >= now() - interval '24 hours' and s.kind = 'registry' group by s.id, s.name, s.domain, s.first_party order by events_24h desc limit 12`).then((r) => r.rows),
        listEvents({ limit: 12, order: "signal", after: since48 }).then((r) => r.items),
        listEvents({ limit: 12, silent_change: true, order: "signal", after: since48 }).then((r) => r.items),
        db.execute<Record<string, unknown>>(sql`select c.*, (select json_build_object('id', s.id, 'name', s.name, 'domain', s.domain) from events e join sources s on s.id = e.source_id where e.id = c.primary_event_id) as source, (select e.slug from events e where e.id = c.primary_event_id) as primary_slug from event_clusters c where c.event_count >= 2 and c.last_at >= now() - interval '48 hours' order by c.max_importance desc, c.event_count desc limit 12`).then((r) => r.rows),
        db.execute<Record<string, unknown>>(sql`
          with cur as (select s.source_id, count(*) as n from changes c join sensors s on s.id = c.sensor_id where c.detected_at >= now() - interval '2 hours' group by s.source_id),
               base as (select s.source_id, count(*)::float / (14*24) as per_hour from changes c join sensors s on s.id = c.sensor_id where c.detected_at >= now() - interval '14 days' group by s.source_id)
          select so.id, so.name, so.domain, cur.n::int as changes_2h, round(coalesce(base.per_hour,0)::numeric*24,1)::float as baseline_per_day,
            round((case when coalesce(base.per_hour,0) = 0 then (case when cur.n/2.0 > 2 then 70 else 40 end) else least(100, case when cur.n/2.0/base.per_hour <= 1 then cur.n/2.0/base.per_hour*30 else 30 + 25*(ln(cur.n/2.0/base.per_hour)/ln(2)) end) end)::numeric, 1)::float as activity_score,
            (case when coalesce(base.per_hour,0) > 0 then round(((cur.n/2.0/base.per_hour - 1) * 100)::numeric) else null end)::int as pct_vs_baseline
          from cur join sources so on so.id = cur.source_id left join base on base.source_id = cur.source_id where so.kind = 'registry' order by activity_score desc limit 12`).then((r) => r.rows),
        db.execute<Record<string, unknown>>(sql`select event_type, count(*)::int as n from events where detected_at >= now() - interval '7 days' group by 1 order by 2 desc`).then((r) => r.rows),
        db.execute<Record<string, unknown>>(sql`select c as category, count(*)::int as n from events e, unnest(e.categories) c where e.detected_at >= now() - interval '7 days' group by 1 order by 2 desc`).then((r) => r.rows),
        listEvents({ limit: 12, order: "recent", confidence_min: 60 }).then((r) => r.items),
        listEvents({ limit: 12, order: "signal", first_party: true, confirmed: true, after: since48 }).then((r) => r.items),
      ]);
      return { most_active_sources: mostActive, biggest_changes: biggest, silent_changes: silent, clusters, unusual_activity: unusual, newly_detected: newly, confirmed_first_party: firstParty, by_type: byType, by_category: byCategory, channels: FEED_CHANNELS, groups: Object.fromEntries(Object.entries(EVENT_GROUPS).map(([k, v]) => [k, v.label])), event_types: Object.fromEntries(Object.entries(EVENT_TYPES).map(([k, v]) => [k, v.label])) };
    }),
  );

  // ---- Search ----------------------------------------------------------------------------
  app.get<{ Querystring: { q?: string; limit?: string } }>("/api/v1/search", async (req) => {
    const q = (req.query.q ?? "").trim();
    const parsed = parseSearch(q);
    if (q.length < 2) return { query: q, parsed, events: [], entities: [], sources: [], urls: [], clusters: [] };
    const lim = Math.min(50, Number(req.query.limit ?? 20));
    const free = parsed.text;
    const [ev, ents, srcs, urls, clusters] = await Promise.all([
      listEvents({ q, limit: lim, order: Object.keys(parsed.filters).length && !free ? "recent" : "recent" }).then((r) => r.items),
      free.length >= 2 ? db.execute<Record<string, unknown>>(sql`select id, name, type, domain, importance, event_count from entities where search @@ plainto_tsquery('simple', ${free}) or name ilike ${"%" + free + "%"} order by event_count desc, importance desc limit ${lim}`).then((r) => r.rows) : [],
      free.length >= 2 ? db.execute<Record<string, unknown>>(sql`select id, name, domain, tier, categories, first_party, country from sources where kind = 'registry' and (name ilike ${"%" + free + "%"} or domain ilike ${"%" + free + "%"}) limit ${lim}`).then((r) => r.rows) : [],
      free.length >= 4 ? db.execute<Record<string, unknown>>(sql`select url, domain, status, change_count, last_seen_at from urls where url ilike ${"%" + free + "%"} order by change_count desc limit ${lim}`).then((r) => r.rows) : [],
      free.length >= 2 ? db.execute<Record<string, unknown>>(sql`select id, slug, title, state, event_count, source_count, max_importance, last_at from event_clusters where title ilike ${"%" + free + "%"} and last_at >= now() - interval '30 days' order by max_importance desc limit 10`).then((r) => r.rows) : [],
    ]);
    return { query: q, parsed, events: ev, entities: ents, sources: srcs, urls, clusters };
  });

  // ---- Connector health / system --------------------------------------------------------------
  app.get("/api/v1/health/connectors", async () =>
    cached("health", 10_000, async () => {
      const [connectors, sensorsByHealth, sensorsByStatus, worst, noisy, daily, byConnectorSensors, throughput, es, topFailingDomains, slowest] = await Promise.all([
        db.execute<Record<string, unknown>>(sql`select * from connector_health order by connector`).then((r) => r.rows),
        db.execute<Record<string, unknown>>(sql`select health, count(*)::int as n from sensors where enabled group by health`).then((r) => r.rows),
        db.execute<Record<string, unknown>>(sql`select status, count(*)::int as n from sensors group by status`).then((r) => r.rows),
        db.execute<Record<string, unknown>>(sql`select s.id, s.name, s.url, s.source_id, s.connector, s.health, s.consecutive_errors, s.last_error, s.last_status, s.last_check_at from sensors s where s.enabled and s.health <> 'UP' order by s.consecutive_errors desc, s.last_check_at desc limit 50`).then((r) => r.rows),
        db.execute<Record<string, unknown>>(sql`select s.id, s.name, s.source_id, s.url, s.raw_changes, s.meaningful_changes, case when s.raw_changes > 0 then round(1 - s.meaningful_changes::numeric / s.raw_changes, 3) else null end as noise_ratio from sensors s where s.raw_changes >= 5 order by noise_ratio desc nulls last, raw_changes desc limit 25`).then((r) => r.rows),
        db.execute<Record<string, unknown>>(sql`select * from metrics_daily order by day desc limit 30`).then((r) => r.rows),
        db.execute<Record<string, unknown>>(sql`select connector, count(*)::int as sensors, count(*) filter (where health = 'UP')::int as up, avg(avg_latency_ms)::int as avg_latency_ms from sensors where enabled group by connector order by sensors desc`).then((r) => r.rows),
        db.execute<Record<string, unknown>>(sql`select (select count(*) from sensor_runs where started_at >= now() - interval '5 minutes')::int as checks_5m, (select count(*) from sensor_runs where started_at >= now() - interval '5 minutes' and outcome = 'not_modified')::int as not_modified_5m, (select count(*) from events where detected_at >= now() - interval '5 minutes')::int as events_5m, (select count(*) from changes where detected_at >= now() - interval '5 minutes')::int as changes_5m, (select count(*) from sensors where enabled and next_check_at <= now())::int as queue_due, (select avg(duration_ms)::int from sensor_runs where started_at >= now() - interval '5 minutes')::int as avg_latency_5m_ms, (select count(*) from changes where detected_at >= now() - interval '24 hours' and change_class in ('cosmetic','navigation','timestamp','advertisement','boilerplate'))::int as noise_filtered_24h`).then((r) => r.rows[0]),
        engineStatus(),
        db.execute<Record<string, unknown>>(sql`select split_part(split_part(s.url, '/', 3), ':', 1) as host, count(*)::int as failures, max(r.error) as last_error from sensor_runs r join sensors s on s.id = r.sensor_id where r.started_at >= now() - interval '24 hours' and r.outcome in ('error','parse_error','rate_limited') group by 1 order by failures desc limit 15`).then((r) => r.rows),
        db.execute<Record<string, unknown>>(sql`select s.id, s.name, s.source_id, s.avg_latency_ms, s.connector from sensors s where s.enabled and s.avg_latency_ms is not null order by s.avg_latency_ms desc limit 15`).then((r) => r.rows),
      ]);
      const t = throughput ?? {};
      return { connectors, sensors_by_health: sensorsByHealth, sensors_by_status: sensorsByStatus, degraded_sensors: worst, noisy_sensors: noisy, daily, sensors_by_connector: byConnectorSensors, throughput: { ...t, checks_per_min: Math.round(Number(t.checks_5m ?? 0) / 5), events_per_min: Math.round((Number(t.events_5m ?? 0) / 5) * 10) / 10, not_modified_ratio: Number(t.checks_5m) ? Math.round((Number(t.not_modified_5m) / Number(t.checks_5m)) * 100) / 100 : null }, engine: es, top_failing_domains: topFailingDomains, slowest_sensors: slowest, live: liveStats() };
    }),
  );

  // ---- Machine-readable feeds ------------------------------------------------------------------
  app.get("/api/v1/feed.rss", async (req, reply) => {
    const q = eventsQuery.parse({ ...(req.query as Record<string, unknown>), limit: 50 });
    const { items } = await listEvents(q);
    const esc = (s: unknown): string => String(s ?? "").replace(/&/g, "&amp;").replace(/</g, "&lt;").replace(/>/g, "&gt;").replace(/"/g, "&quot;");
    const base = config.publicBaseUrl;
    const xml = `<?xml version="1.0" encoding="UTF-8"?>\n<rss version="2.0" xmlns:atom="http://www.w3.org/2005/Atom"><channel><title>WebSensor — ${esc(q.category ? q.category + " events" : q.silent_change ? "silent changes" : "live events")}</title><link>${base}</link><description>Meaningful changes detected on the public Web by WebSensor.</description><atom:link href="${base}/api/v1/feed.rss" rel="self" type="application/rss+xml"/>${items
      .map((e) => `<item><title>${esc(e.title)}</title><link>${base}/event/${esc(e.slug)}</link><guid isPermaLink="false">${esc(e.id)}</guid><pubDate>${new Date(String(e.detected_at)).toUTCString()}</pubDate><category>${esc(e.event_type)}</category><description>${esc(e.summary)} (signal ${esc(e.signal_score ?? e.importance)}, importance ${esc(e.importance)}, confidence ${esc(e.confidence)}${e.silent_change ? ", silent change" : ""}${e.first_party ? ", first-party" : ""}) — source: ${esc(e.url)}</description></item>`)
      .join("")}</channel></rss>`;
    reply.header("content-type", "application/rss+xml; charset=utf-8");
    return reply.send(xml);
  });

  app.get("/api/v1", async () => ({
    name: "WebSensor API",
    version: "v1",
    release: config.version,
    docs: `${config.publicBaseUrl}/api`,
    rate_limit: "600 requests / minute / IP (headers x-ratelimit-limit, x-ratelimit-remaining, x-ratelimit-reset)",
    endpoints: [
      "/api/v1/events",
      "/api/v1/events/count",
      "/api/v1/events/{id|slug}",
      "/api/v1/changes/{id}",
      "/api/v1/snapshots/{id}",
      "/api/v1/snapshots/compare?a=&b=",
      "/api/v1/sensors/{id}",
      "/api/v1/sensors/{id}/snapshots",
      "/api/v1/sources",
      "/api/v1/sources/{id}",
      "/api/v1/entities",
      "/api/v1/entities/rank",
      "/api/v1/entities/{id}",
      "/api/v1/entities/{id}/timeline",
      "/api/v1/clusters",
      "/api/v1/clusters/{id|slug}",
      "/api/v1/domains/{domain}/timeline",
      "/api/v1/urls/history?url=",
      "/api/v1/search?q=",
      "/api/v1/stats",
      "/api/v1/trending",
      "/api/v1/breaking",
      "/api/v1/pulse",
      "/api/v1/radar",
      "/api/v1/explore",
      "/api/v1/countries",
      "/api/v1/countries/{code|slug}",
      "/api/v1/categories/{channel}",
      "/api/v1/health/connectors",
      "/api/v1/watchlists",
      "/api/v1/alerts",
      "/api/v1/notifications",
      "/api/v1/bookmarks",
      "/api/v1/views",
      "/api/v1/monitors",
      "/api/v1/feed.rss",
      "wss://…/api/v1/live",
    ],
  }));
}

function stripEmpty(o: Record<string, unknown>): Record<string, unknown> {
  return Object.fromEntries(Object.entries(o).filter(([, v]) => v !== undefined && v !== null && v !== ""));
}

function renderCanonical(c: { mode: string; text?: string; items?: Record<string, unknown>[]; json?: unknown }): string {
  if (c.mode === "text") return c.text ?? "";
  if (c.mode === "list") return (c.items ?? []).map((i) => [i.title ?? i.url ?? i.key, i.url && i.title ? i.url : null, i.summary ? String(i.summary).slice(0, 300) : null].filter(Boolean).join(" — ")).join("\n");
  return JSON.stringify(c.json ?? null, null, 2);
}

export { textArray };