server.ts

import { Kafka, type Producer } from "kafkajs";
import * as dgram from "node:dgram";
import { LAN_IFACE, OPN_RANGES, PAGE_QUERIES, WAN_IFACE } from "./queries";
import { log, logThrottled } from "./logger";

// ─── Process-level safety net (register once, before anything else runs) ───
// unhandledRejection: log + CONTINUE (a single dropped promise must not take the
// KPI publisher / NetFlow collector down). uncaughtException: log + exit(1) so
// k8s/systemd restarts into a known-good state.
process.on("unhandledRejection", (reason) => {
  log.error("unhandledRejection", {
    err: String((reason as Error)?.message ?? reason),
    stack: (reason as Error)?.stack,
  });
});
process.on("uncaughtException", (err) => {
  log.error("uncaughtException", { err: err.message, stack: err.stack });
  process.exit(1);
});

// Standard 500 responder: log the failure (route + err + stack) BEFORE replying.
// The browser read path must never fail silently again.
function respond500(route: string, e: unknown, req: Request): Response {
  log.error("request handler failed", {
    route,
    err: String((e as Error)?.message ?? e),
    stack: (e as Error)?.stack,
  });
  return Response.json({ error: (e as Error).message }, { status: 500, headers: corsHeaders(req.headers.get("origin")) });
}

const CLICKHOUSE_URL = process.env.CLICKHOUSE_URL || "http://localhost:8123";
const PORT = parseInt(process.env.PORT || "3005");
// Your firewall's API base, e.g. https://203.0.113.1/api — REQUIRED for the
// OPNsense-backed routes and collectors; everything else degrades gracefully.
const OPN_URL = process.env.OPN_URL || "";
// Secrets come from the environment (k8s Secret `glm-firewall-secrets` in-cluster,
// or /etc/glm-firewall-api.env via the systemd EnvironmentFile on a bare host).
const OPN_KEY = process.env.OPN_KEY || "";
const OPN_SECRET = process.env.OPN_SECRET || "";
const OPN_AUTH = "Basic " + btoa(`${OPN_KEY}:${OPN_SECRET}`);

const KAFKA_BROKERS = (process.env.KAFKA_BROKERS || "localhost:9092").split(",");
const KAFKA_TOPIC = process.env.KAFKA_TOPIC || "opnsense-stats";
// Optional SASL/SCRAM-SHA-512 + TLS for the authenticated Kafka listeners
// (KAFKA-SECURITY-ROADMAP.md Tier 2). Unset env = plaintext dial, same
// dual-mode convention as the Go clients.
const KAFKA_SASL_USERNAME = process.env.KAFKA_SASL_USERNAME || "";
const KAFKA_SASL_PASSWORD = process.env.KAFKA_SASL_PASSWORD || "";
const KAFKA_CA_FILE = process.env.KAFKA_CA_FILE || "";
const KPI_INTERVAL = 10_000;

const CORS_ORIGINS = (process.env.CORS_ORIGINS || "http://localhost:5173,https://monitor.example.com").split(",");

// ─── CrowdSec LAPI (direct HTTP, no SSH) ───
// e.g. http://203.0.113.1:8080 — unset disables the CrowdSec checks.
const CS_LAPI_URL = process.env.CS_LAPI_URL || "";
const CS_BOUNCER_KEY = process.env.CS_BOUNCER_KEY || "";

async function csLapiGet(path: string) {
  const resp = await fetch(`${CS_LAPI_URL}${path}`, {
    headers: { "X-Api-Key": CS_BOUNCER_KEY },
  });
  return resp.json();
}

// ─── CrowdSec CAPI ban count cache (CAPI updates every ~2h, no need to poll faster) ───
// Split by origin: CAPI/lists = community blocklist (defense strength, not a threat
// signal); anything else (crowdsec/cscli) = a LOCAL detection triggered by our own
// logs — those are the actionable ones.
const CS_COMMUNITY_ORIGINS = new Set(["CAPI", "lists"]);
let csCache = {
  total: 0,
  community: 0,
  local: 0,
  localDecisions: [] as Array<Record<string, unknown>>,
  sample: [] as Array<Record<string, unknown>>,
  updatedAt: 0,
};
const CS_CACHE_TTL = 5 * 60 * 1000; // 5 minutes

async function getCrowdsecSplit(): Promise<typeof csCache> {
  if (!CS_LAPI_URL) return csCache;
  if (Date.now() - csCache.updatedAt < CS_CACHE_TTL) return csCache;
  try {
    const decisions = await csLapiGet("/v1/decisions") as Array<Record<string, unknown>> | null;
    const all = Array.isArray(decisions) ? decisions : [];
    const local = all.filter(d => !CS_COMMUNITY_ORIGINS.has(String(d.origin)));
    csCache = {
      total: all.length,
      community: all.length - local.length,
      local: local.length,
      localDecisions: local.slice(0, 50),
      sample: all.slice(0, 10),
      updatedAt: Date.now(),
    };
    return csCache;
  } catch (e) {
    logThrottled("crowdsec-split", 60_000, "warn", "CrowdSec LAPI fetch failed, serving cached decisions", {
      err: String((e as Error)?.message ?? e),
    });
    return csCache;
  }
}

async function getCrowdsecBanCount(): Promise<number> {
  return (await getCrowdsecSplit()).total;
}

// ─── Benign-context rules (verdict suppression + scorer exclusions) ───
// Site-specific: source prefixes whose P2P/VPN churn dominates NetFlow and
// matches low-confidence blocklists constantly (the nature of public peer
// swarms, not compromise). Comma-separated, e.g. "10.0.50.,10.193." for a
// torrent VLAN and its WireGuard tunnel. Empty = no suppression.
const VPN_SRC_PREFIXES = (process.env.VPN_SRC_PREFIXES || "").split(",").map(s => s.trim()).filter(Boolean);
const TORRENT_PORTS = new Set([51413, 51414]);
const isTorrentPort = (p: number) => (p >= 6881 && p <= 6999) || TORRENT_PORTS.has(p);
const isVpnSource = (ip: string) => VPN_SRC_PREFIXES.some(pre => ip.startsWith(pre));
// Push-notification keepalives (Google FCM 5228-5230) beacon by design.
const PUSH_KEEPALIVE_PORTS = new Set([5228, 5229, 5230]);
// ng_netflow chops long-lived flows at its active timeout (~30 min), so continuous
// connections re-export on a metronome and score as "perfectly regular beacons".
const isFlowTimerArtifact = (intervalSec: number) => intervalSec >= 1400 && intervalSec <= 2200;
// Domains whose subdomains are random-looking BY DESIGN (CDN hashes, per-device
// hostnames) plus your own zones (add them via DNS_BENIGN_EXTRA, comma-separated).
// Suffix match on the registered domain.
const DNS_BENIGN_SUFFIXES = [
  "plex.direct", "amazonaws.com", "steamserver.net", "mozgcp.net", "akadns.net",
  "googlevideo.com", "gvt1.com", "spotify.com", "cloudfront.net", "akamaiedge.net",
  "example.com", "in-addr.arpa", "ip6.arpa",
  "svc", "cluster.local", // k8s service DNS spilling to Unbound
  ...(process.env.DNS_BENIGN_EXTRA || "").split(",").map(s => s.trim().toLowerCase()).filter(Boolean),
];
const isBenignDomain = (domain: string) => {
  const d = (domain || "").toLowerCase();
  return DNS_BENIGN_SUFFIXES.some(suf => d === suf || d.endsWith("." + suf));
};
// Known-benign egress IDS patterns: signature substring + destination CIDR.
// Heavy git-over-SSH to GitHub trips the "SSH Scan OUTBOUND" rate heuristic daily.
const ipToUint = (ip: string): number => {
  const p = ip.split(".").map(Number);
  return p.length === 4 && p.every(n => n >= 0 && n <= 255)
    ? ((p[0] << 24) | (p[1] << 16) | (p[2] << 8) | p[3]) >>> 0 : -1;
};
const inCidr = (ip: string, cidr: string): boolean => {
  const [base, bitsStr] = cidr.split("/");
  const bits = Number(bitsStr);
  const ipN = ipToUint(ip), baseN = ipToUint(base);
  if (ipN < 0 || baseN < 0) return false;
  const mask = bits === 0 ? 0 : (~0 << (32 - bits)) >>> 0;
  return (ipN & mask) === (baseN & mask);
};
const BENIGN_EGRESS_IDS: Array<{ sig: string; cidr: string; label: string }> = [
  { sig: "SSH Scan OUTBOUND", cidr: "140.82.112.0/20", label: "git-over-SSH to GitHub" },
];
// Guard for `INTERVAL ${range}` interpolation on the new endpoint.
const parseRange = (raw: string | null, fallback: string) => {
  const r = (raw || "").trim().toUpperCase();
  return /^\d{1,4} (MINUTE|HOUR|DAY)S?$/.test(r) ? r : fallback;
};

// ─── CORS helper ───
function corsHeaders(origin?: string | null): Record<string, string> {
  const allowed = origin && CORS_ORIGINS.includes(origin) ? origin : CORS_ORIGINS[0];
  return {
    "Access-Control-Allow-Origin": allowed,
    "Access-Control-Allow-Methods": "GET, POST, OPTIONS",
    "Access-Control-Allow-Headers": "Content-Type",
  };
}

// ─── GeoIP cache ───
const geoCache = new Map<string, Record<string, unknown>>();
const GEO_TTL = 24 * 60 * 60 * 1000;

// ─── OPNsense API helpers ───
async function opnGet(path: string) {
  if (!OPN_URL) throw new Error("OPN_URL not configured");
  const resp = await fetch(`${OPN_URL}${path}`, {
    headers: { Authorization: OPN_AUTH },
    // Bun-specific TLS option for self-signed certs
    tls: { rejectUnauthorized: false },
  });
  return resp.json();
}

async function opnPost(path: string, body?: Record<string, unknown>) {
  if (!OPN_URL) throw new Error("OPN_URL not configured");
  const resp = await fetch(`${OPN_URL}${path}`, {
    method: "POST",
    headers: { Authorization: OPN_AUTH, "Content-Type": "application/json" },
    body: body ? JSON.stringify(body) : undefined,
    // Bun-specific TLS option for self-signed certs
    tls: { rejectUnauthorized: false },
  });
  return resp.json();
}

// ─── ClickHouse query helpers ───
// chQuery is the writable path — ONLY for internal jobs (refreshThreatFeeds
// needs CREATE/TRUNCATE/EXCHANGE/DROP, KPI polling is trusted). Everything a
// browser request can trigger goes through chQueryRO: readonly=1 makes
// ClickHouse itself reject INSERT/ALTER/DDL regardless of what the SQL says.
async function chRun(sql: string, readonly: boolean) {
  // Throttle key derived from the query shape (chRun has no name); one line per
  // distinct failing query per minute so a broken read path is visible but not spammy.
  const norm = sql.replace(/\s+/g, " ").trim();
  const throttleKey = "chRun:" + norm.slice(0, 100);
  let resp: Response;
  try {
    resp = await fetch(`${CLICKHOUSE_URL}/?default_format=JSONEachRow${readonly ? "&readonly=1" : ""}`, {
      method: "POST",
      body: sql,
    });
  } catch (e) {
    logThrottled(throttleKey, 60_000, "error", "ClickHouse request failed (unreachable)", {
      err: String((e as Error)?.message ?? e), query: norm.slice(0, 200),
    });
    return { error: String((e as Error)?.message ?? e), data: [] };
  }
  const text = await resp.text();
  if (!resp.ok) {
    logThrottled(throttleKey, 60_000, "error", "ClickHouse query failed", {
      status: resp.status, err: text.slice(0, 500), query: norm.slice(0, 200),
    });
    return { error: text, data: [] };
  }
  const rows = text.trim()
    ? text.trim().split("\n").map((l) => JSON.parse(l))
    : [];
  return { data: rows };
}
const chQuery = (sql: string) => chRun(sql, false);
const chQueryRO = (sql: string) => chRun(sql, true);

// ─── GeoIP enrichment handler ───
async function handleGeoEnrich(req: Request): Promise<Response> {
  try {
    const { ips = [] } = (await req.json()) as { ips: string[] };
    const results: Record<string, unknown>[] = [];
    const toFetch: string[] = [];

    for (const ip of ips) {
      const cached = geoCache.get(ip);
      if (cached && Date.now() - (cached._cachedAt as number) < GEO_TTL) {
        results.push(cached);
      } else {
        toFetch.push(ip);
      }
    }

    if (toFetch.length > 0) {
      for (let i = 0; i < toFetch.length; i += 100) {
        const chunk = toFetch.slice(i, i + 100);
        const resp = await fetch("http://ip-api.com/batch", {
          method: "POST",
          headers: { "Content-Type": "application/json" },
          body: JSON.stringify(chunk),
        });
        const data = (await resp.json()) as Array<Record<string, unknown>>;
        for (const d of data) {
          if (d.status !== "success") continue;
          const entry = {
            ip: d.query, country: d.country, cc: d.countryCode,
            city: d.city, region: d.regionName, lat: d.lat, lon: d.lon,
            isp: d.isp, org: d.org, as: d.as, _cachedAt: Date.now(),
          };
          geoCache.set(d.query as string, entry);
          results.push(entry);
        }
      }
    }

    return Response.json(results, { headers: corsHeaders(req.headers.get("origin")) });
  } catch (e: unknown) {
    return respond500("/api/geo/enrich", e, req);
  }
}

// ─── Kafka producer ───
let kafkaProducer: Producer | null = null;

async function initKafka() {
  try {
    const kafka = new Kafka({
      clientId: "glm-firewall-api",
      brokers: KAFKA_BROKERS,
      ...(KAFKA_CA_FILE
        ? { ssl: { ca: [await Bun.file(KAFKA_CA_FILE).text()] } }
        : {}),
      ...(KAFKA_SASL_USERNAME
        ? {
            sasl: {
              mechanism: "scram-sha-512" as const,
              username: KAFKA_SASL_USERNAME,
              password: KAFKA_SASL_PASSWORD,
            },
          }
        : {}),
    });
    kafkaProducer = kafka.producer();
    await kafkaProducer.connect();
    log.info("Kafka connected", {
      brokers: KAFKA_BROKERS.join(","),
      topic: KAFKA_TOPIC,
      sasl: KAFKA_SASL_USERNAME || undefined,
      tls: KAFKA_CA_FILE ? true : undefined,
    });
  } catch (e) {
    log.error("Kafka init failed (KPI publishing disabled until restart)", {
      err: String((e as Error)?.message ?? e), stack: (e as Error)?.stack,
    });
    kafkaProducer = null;
  }
}

// ─── KPI polling queries (subset for Kafka publishing) ───
const PUB_FILTER = `AND NOT (src_ip LIKE '192.168.%' OR src_ip LIKE '10.%' OR (src_ip >= '172.16.0.0' AND src_ip <= '172.31.255.255') OR src_ip LIKE '224.%' OR src_ip LIKE '169.254.%' OR src_ip LIKE '0.%' OR src_ip LIKE '255.%' OR src_ip LIKE '127.%')`;

function kpiQueries(): string[] {
  return [
    // 0: CPU (last 5 min avg)
    `SELECT 100-avg(fields['usage_idle']) AS v FROM opnsense.telegraf_metrics WHERE measurement='cpu' AND tags['cpu']='cpu-total' AND timestamp>now()-INTERVAL 5 MINUTE`,
    // 1: Memory
    `SELECT avg(fields['used_percent']) AS v FROM opnsense.telegraf_metrics WHERE measurement='mem' AND timestamp>now()-INTERVAL 5 MINUTE`,
    // 2: Disk
    `SELECT avg(fields['used_percent']) AS v FROM opnsense.telegraf_metrics WHERE measurement='disk' AND timestamp>now()-INTERVAL 5 MINUTE`,
    // 3: Temperatures
    `SELECT tags['sensor'] AS sensor,argMax(fields['degrees'],timestamp) AS deg FROM opnsense.telegraf_metrics WHERE measurement='temperature' AND timestamp>now()-INTERVAL 5 MINUTE GROUP BY sensor`,
    // 4: WAN bandwidth (last 2 points for delta)
    `SELECT timestamp AS t,fields['bytes_recv'] AS rx,fields['bytes_sent'] AS tx FROM opnsense.telegraf_metrics WHERE measurement='net' AND tags['interface']='${WAN_IFACE}' AND timestamp>now()-INTERVAL 1 MINUTE ORDER BY timestamp DESC LIMIT 2`,
    // 5: LAN bandwidth (last 2 points for delta)
    `SELECT timestamp AS t,fields['bytes_recv'] AS rx,fields['bytes_sent'] AS tx FROM opnsense.telegraf_metrics WHERE measurement='net' AND tags['interface']='${LAN_IFACE}' AND timestamp>now()-INTERVAL 1 MINUTE ORDER BY timestamp DESC LIMIT 2`,
    // 6: Blocked count (last hour)
    `SELECT count() AS cnt FROM opnsense.firewall_logs WHERE action='block' ${PUB_FILTER} AND timestamp>now()-INTERVAL 1 HOUR`,
    // 7: Total events (last hour)
    `SELECT count() AS cnt FROM opnsense.firewall_logs WHERE timestamp>now()-INTERVAL 1 HOUR`,
    // 8: PF states
    `SELECT argMax(fields['entries'],timestamp) AS current, argMax(fields['state-limit'],timestamp) AS lim FROM opnsense.telegraf_metrics WHERE measurement='pf' AND timestamp>now()-INTERVAL 5 MINUTE`,
    // 9: Gateways
    `SELECT tags['gateway_name'] AS gw,avg(fields['delay']) AS delay,avg(fields['loss']) AS loss,argMax(fields['status'],timestamp) AS status FROM opnsense.telegraf_metrics WHERE measurement='gateways' AND timestamp>now()-INTERVAL 5 MINUTE GROUP BY gw`,
    // 10: Uptime
    `SELECT fields['uptime'] AS v FROM opnsense.telegraf_metrics WHERE measurement='system' AND timestamp>now()-INTERVAL 5 MINUTE ORDER BY timestamp DESC LIMIT 1`,
    // 11: DNS stats from ClickHouse (last hour)
    `SELECT sum(total) AS total, sum(blocked) AS blocked, max(blocklist_size) AS blocklist_size FROM opnsense.dns_stats WHERE timestamp>now()-INTERVAL 1 HOUR`,
  ];
}

async function publishKpis() {
  if (!kafkaProducer) {
    logThrottled("kpi-no-producer", 60_000, "warn", "Skipping KPI publish: Kafka producer not connected", {});
    return;
  }

  try {
    const results = await Promise.all(kpiQueries().map(chQuery));

    // Extract KPIs
    const cpuUsage = results[0]?.data?.[0]?.v ?? 0;
    const memoryUsage = results[1]?.data?.[0]?.v ?? 0;
    const diskUsage = results[2]?.data?.[0]?.v ?? 0;

    const temps = (results[3]?.data || []) as Array<{ sensor: string; deg: number }>;
    // OPNsense uses tz0 (thermal zone) — take max of all sensors
    const allDegs = temps.map(t => Number(t.deg)).filter(d => !isNaN(d));
    const cpuTemp = allDegs.length > 0 ? Math.max(...allDegs) : 0;
    const acpiTemp = 0; // OPNsense only has tz0, no separate ACPI sensor

    // Bandwidth: delta between 2 most recent points (bytes/sec * 8 = bps)
    const wanPts = (results[4]?.data || []) as Array<{ t: string; rx: number; tx: number }>;
    let wanDownBps = 0, wanUpBps = 0;
    if (wanPts.length >= 2) {
      const dt = (new Date(wanPts[0].t).getTime() - new Date(wanPts[1].t).getTime()) / 1000;
      if (dt > 0) {
        wanDownBps = Math.max(0, ((wanPts[0].rx - wanPts[1].rx) * 8) / dt);
        wanUpBps = Math.max(0, ((wanPts[0].tx - wanPts[1].tx) * 8) / dt);
      }
    }

    const lanPts = (results[5]?.data || []) as Array<{ t: string; rx: number; tx: number }>;
    let lanDownBps = 0, lanUpBps = 0;
    if (lanPts.length >= 2) {
      const dt = (new Date(lanPts[0].t).getTime() - new Date(lanPts[1].t).getTime()) / 1000;
      if (dt > 0) {
        lanDownBps = Math.max(0, ((lanPts[0].rx - lanPts[1].rx) * 8) / dt);
        lanUpBps = Math.max(0, ((lanPts[0].tx - lanPts[1].tx) * 8) / dt);
      }
    }

    const blockedCount = Number(results[6]?.data?.[0]?.cnt ?? 0);
    const totalEvents = Number(results[7]?.data?.[0]?.cnt ?? 0);
    const blockRate = totalEvents > 0 ? (blockedCount / totalEvents * 100) : 0;

    const pfCurrent = Number(results[8]?.data?.[0]?.current ?? 0);
    const pfLimit = Number(results[8]?.data?.[0]?.lim ?? 0);

    const gws = (results[9]?.data || []) as Array<{ gw: string; delay: number; loss: number; status: number }>;
    const wanGw = gws.find(g => g.gw === "WAN_GW");

    const uptime = Number(results[10]?.data?.[0]?.v ?? 0);

    const dnsData = results[11]?.data?.[0] as { total?: number; blocked?: number; blocklist_size?: number } | undefined;

    // Also fetch live DNS totals from OPNsense API for current numbers
    let dnsTotals = { total: 0, blocked: 0, blocklistSize: 0 };
    try {
      const dnsResp = await opnGet("/unbound/overview/totals/60") as { total?: number; blocked?: { total?: number }; blocklist_size?: number };
      dnsTotals = {
        total: dnsResp.total || Number(dnsData?.total ?? 0),
        blocked: dnsResp.blocked?.total || Number(dnsData?.blocked ?? 0),
        blocklistSize: dnsResp.blocklist_size || Number(dnsData?.blocklist_size ?? 0),
      };
    } catch (e) {
      logThrottled("kpi-dns-totals", 60_000, "warn", "OPNsense DNS totals fetch failed, using ClickHouse fallback", {
        err: String((e as Error)?.message ?? e),
      });
      dnsTotals = {
        total: Number(dnsData?.total ?? 0),
        blocked: Number(dnsData?.blocked ?? 0),
        blocklistSize: Number(dnsData?.blocklist_size ?? 0),
      };
    }

    // CrowdSec CAPI ban count (cached, refreshes every 5 min via LAPI)
    const crowdsecBans = await getCrowdsecBanCount();

    const kpiMessage = {
      timestamp: new Date().toISOString(),
      system: {
        cpuUsage: Number(Number(cpuUsage).toFixed(1)),
        memoryUsage: Number(Number(memoryUsage).toFixed(1)),
        diskUsage: Number(Number(diskUsage).toFixed(1)),
        cpuTemp: Number(Number(cpuTemp).toFixed(1)),
        acpiTemp: Number(Number(acpiTemp).toFixed(1)),
        uptime: Math.round(uptime),
      },
      firewall: {
        blockedCount,
        blockRate: Number(blockRate.toFixed(1)),
        pfStates: pfCurrent,
        pfStateLimit: pfLimit,
      },
      bandwidth: {
        wanDownBps: Math.round(wanDownBps),
        wanUpBps: Math.round(wanUpBps),
        lanDownBps: Math.round(lanDownBps),
        lanUpBps: Math.round(lanUpBps),
      },
      dns: {
        totalQueries: dnsTotals.total,
        blockedQueries: dnsTotals.blocked,
        blockRate: dnsTotals.total > 0 ? Number(((dnsTotals.blocked / dnsTotals.total) * 100).toFixed(1)) : 0,
        blocklistSize: dnsTotals.blocklistSize,
      },
      crowdsec: { activeBans: crowdsecBans },
      gateway: {
        delay: wanGw ? Number(Number(wanGw.delay).toFixed(1)) : 0,
        loss: wanGw ? Number(Number(wanGw.loss).toFixed(1)) : 0,
        status: wanGw ? (wanGw.status === 1 ? "online" : "offline") : "unknown",
      },
    };

    await kafkaProducer.send({
      topic: KAFKA_TOPIC,
      messages: [{ value: JSON.stringify(kpiMessage) }],
    });
  } catch (e) {
    logThrottled("kpi-publish", 60_000, "error", "KPI publish failed", {
      err: String((e as Error)?.message ?? e), stack: (e as Error)?.stack,
    });
  }
}

// ─── Security verdict ───
// Replaces the client-side 0-100 "threat score" (permanently pinned ~15/CRITICAL
// because every input measured internet background noise or the defenses doing
// their job). The verdict escalates only on actionable evidence; benign patterns
// are suppressed WITH visible counts so the UI can show what was filtered and why.

interface Finding {
  severity: "action" | "review";
  check: string;
  title: string;
  detail: string;
  count: number;
  lastSeen?: string;
}

interface CheckStatus {
  id: string;
  label: string;
  status: "ok" | "findings" | "stale";
  detail: string;
  suppressed: number;
}

// Post-NAT egress logs on the WAN interface carry this src (the firewall's own
// WAN address behind an upstream router, if any). Empty = the WAN-src special
// case never applies.
const WAN_SRC_IP = process.env.WAN_SRC_IP || "";

async function computeVerdict(range: string) {
  const [
    egress, beacons, dnsAnoms, idsSev1, idsEgress, idsTotal,
    gateways, pf, freshness, blocked, fwTotal, nfTotal, dnsCounts, feeds,
  ] = await Promise.all([
    chQueryRO(`
      SELECT fl.src_ip AS src_ip, fl.dst_ip AS dst_ip, fl.dst_port AS dst_port,
             ti.source AS source, ti.confidence AS confidence, count() AS cnt,
             formatDateTime(max(fl.timestamp),'%Y-%m-%d %H:%i:%S') AS last_seen
      FROM opnsense.firewall_logs fl
      INNER JOIN opnsense.threat_intel ti ON fl.dst_ip = ti.ip
      WHERE fl.action='pass' AND fl.direction='out' AND fl.timestamp > now() - INTERVAL ${range}
      GROUP BY src_ip, dst_ip, dst_port, source, confidence
      ORDER BY confidence DESC, cnt DESC LIMIT 200`),
    chQueryRO(`
      SELECT src_ip, dst_ip, dst_port, beacon_score, connection_count,
             interval_mode, total_bytes,
             formatDateTime(last_seen,'%Y-%m-%d %H:%i:%S') AS last_seen_fmt
      FROM opnsense.beacon_scores FINAL
      WHERE beacon_score >= 0.7 AND last_seen > now() - INTERVAL ${range}
      ORDER BY beacon_score DESC LIMIT 500`),
    chQueryRO(`
      SELECT anomaly_type, client_ip, domain,
             max(query_count) AS query_count, max(avg_entropy) AS entropy,
             max(avg_subdomain_len) AS sub_len, count() AS occurrences,
             formatDateTime(max(detected_at),'%Y-%m-%d %H:%i:%S') AS last_seen
      FROM opnsense.dns_anomalies
      WHERE detected_at > now() - INTERVAL ${range}
      GROUP BY anomaly_type, client_ip, domain
      ORDER BY occurrences DESC LIMIT 200`),
    chQueryRO(`
      SELECT alert_signature, alert_category, src_ip, dst_ip, count() AS cnt,
             formatDateTime(max(timestamp),'%Y-%m-%d %H:%i:%S') AS last_seen
      FROM opnsense.suricata_alerts
      WHERE alert_severity = 1 AND timestamp > now() - INTERVAL ${range}
      GROUP BY alert_signature, alert_category, src_ip, dst_ip
      ORDER BY cnt DESC LIMIT 100`),
    chQueryRO(`
      SELECT alert_signature, alert_category, src_ip, dst_ip, count() AS cnt,
             formatDateTime(max(timestamp),'%Y-%m-%d %H:%i:%S') AS last_seen
      FROM opnsense.suricata_alerts
      WHERE alert_severity > 1 AND timestamp > now() - INTERVAL ${range}
        AND positionCaseInsensitive(alert_signature, 'OUTBOUND') > 0
      GROUP BY alert_signature, alert_category, src_ip, dst_ip
      ORDER BY cnt DESC LIMIT 100`),
    chQueryRO(`SELECT count() AS cnt, uniqExact(src_ip) AS srcs FROM opnsense.suricata_alerts WHERE timestamp > now() - INTERVAL ${range}`),
    chQueryRO(`SELECT tags['gateway_name'] AS gw, avg(fields['delay']) AS delay, avg(fields['loss']) AS loss, argMax(fields['status'],timestamp) AS status FROM opnsense.telegraf_metrics WHERE measurement='gateways' AND timestamp>now()-INTERVAL 5 MINUTE GROUP BY gw`),
    chQueryRO(`SELECT argMax(fields['entries'],timestamp) AS current, argMax(fields['state-limit'],timestamp) AS lim FROM opnsense.telegraf_metrics WHERE measurement='pf' AND timestamp>now()-INTERVAL 5 MINUTE`),
    chQueryRO(`
      SELECT 'firewall_logs' AS src, toUnixTimestamp(toDateTime(max(timestamp))) AS last FROM opnsense.firewall_logs
      UNION ALL SELECT 'netflow', toUnixTimestamp(toDateTime(max(timestamp))) FROM opnsense.netflow
      UNION ALL SELECT 'telegraf_metrics', toUnixTimestamp(toDateTime(max(timestamp))) FROM opnsense.telegraf_metrics
      UNION ALL SELECT 'dns_queries', toUnixTimestamp(toDateTime(max(timestamp))) FROM opnsense.dns_queries
      UNION ALL SELECT 'suricata_alerts', toUnixTimestamp(toDateTime(max(timestamp))) FROM opnsense.suricata_alerts
      UNION ALL SELECT 'threat_intel', toUnixTimestamp(toDateTime(max(inserted_at))) FROM opnsense.threat_intel`),
    chQueryRO(`SELECT count() AS cnt, uniqExact(src_ip) AS srcs FROM opnsense.firewall_logs WHERE action='block' ${PUB_FILTER} AND timestamp > now() - INTERVAL ${range}`),
    chQueryRO(`SELECT count() AS cnt FROM opnsense.firewall_logs WHERE timestamp > now() - INTERVAL ${range}`),
    chQueryRO(`SELECT count() AS cnt FROM opnsense.netflow WHERE timestamp > now() - INTERVAL ${range}`),
    chQueryRO(`SELECT count() AS total, countIf(is_blocked=1) AS blocked FROM opnsense.dns_queries WHERE timestamp > now() - INTERVAL ${range}`),
    chQueryRO(`SELECT source, count() AS ips FROM opnsense.threat_intel GROUP BY source`),
  ]);

  const findings: Finding[] = [];
  const checks: CheckStatus[] = [];
  const fmtMB = (b: number) => b >= 1048576 ? `${(b / 1048576).toFixed(1)} MB` : `${(b / 1024).toFixed(1)} KB`;

  // A failed query must not read as a quiet, healthy check.
  const queryErrors = [
    ["egress-intel", egress], ["beacons", beacons], ["dns", dnsAnoms],
    ["ids", idsSev1], ["ids", idsEgress], ["ids", idsTotal],
    ["infra", gateways], ["infra", pf], ["telemetry", freshness],
  ].filter(([, r]) => (r as { error?: string }).error);
  if (queryErrors.length > 0) {
    const failedChecks = [...new Set(queryErrors.map(([id]) => id as string))];
    findings.push({
      severity: "review", check: "telemetry",
      title: `Verdict degraded: ${queryErrors.length} check quer${queryErrors.length === 1 ? "y" : "ies"} failed`,
      detail: `affected: ${failedChecks.join(", ")} — treat those checks as blind, not clean`,
      count: queryErrors.length,
    });
    log.error("Verdict computed on degraded data: check queries failed", {
      failed: failedChecks.join(","),
      errors: queryErrors.map(([id, r]) => `${id}: ${(r as { error?: string }).error?.slice(0, 200)}`).join(" | "),
    });
  }

  // 1. Egress vs threat intel — the highest-value signal.
  let egSuppressed = 0;
  for (const m of (egress.data as Array<Record<string, unknown>>)) {
    const conf = Number(m.confidence), port = Number(m.dst_port), cnt = Number(m.cnt);
    if (conf >= 5) {
      findings.push({
        severity: "action", check: "egress-intel",
        title: `${m.src_ip} connected out to known C2 ${m.dst_ip}:${port}`,
        detail: `${cnt} allowed outbound connection(s) to an IP on the ${m.source} botnet C2 list (confidence ${conf}/5)`,
        count: cnt, lastSeen: String(m.last_seen),
      });
    } else if ((isVpnSource(String(m.src_ip)) || m.src_ip === WAN_SRC_IP) && (port >= 1024 || port === 0)) {
      // Low-confidence feed hit on an ephemeral/high port (or port 0 = ICMP/GRE)
      // from the VPN path or post-NAT WAN src — the signature of torrent peers
      // that also attacked someone's SSH once. Not attributable, not actionable.
      egSuppressed++;
    } else {
      findings.push({
        severity: "review", check: "egress-intel",
        title: `${m.src_ip} → ${m.dst_ip}:${port} is on the ${m.source} blocklist`,
        detail: `${cnt} allowed outbound connection(s) to a listed IP on a well-known service port`,
        count: cnt, lastSeen: String(m.last_seen),
      });
    }
  }
  const egFound = findings.filter(f => f.check === "egress-intel").length;
  checks.push({
    id: "egress-intel", label: "Egress intel",
    status: egFound ? "findings" : "ok",
    detail: egFound ? `${egFound} match(es) need eyes` : `no outbound traffic to high-confidence bad IPs`,
    suppressed: egSuppressed,
  });

  // 2. Beacons — only non-VPN, non-torrent, non-flow-timer, high-score survive.
  let bcVpn = 0, bcTimer = 0, bcMedium = 0;
  for (const b of (beacons.data as Array<Record<string, unknown>>)) {
    const port = Number(b.dst_port), ivl = Number(b.interval_mode), score = Number(b.beacon_score);
    if (isVpnSource(String(b.src_ip)) || isTorrentPort(port) || PUSH_KEEPALIVE_PORTS.has(port)) { bcVpn++; continue; }
    if (isFlowTimerArtifact(ivl)) { bcTimer++; continue; }
    if (score >= 0.85) {
      findings.push({
        severity: "review", check: "beacons",
        title: `${b.src_ip} calls ${b.dst_ip}:${port} every ~${Math.max(1, Math.round(ivl / 60))} min`,
        detail: `beacon score ${score.toFixed(2)}, ${b.connection_count} connections, ${fmtMB(Number(b.total_bytes))} total — regular enough to look like automated check-in`,
        count: Number(b.connection_count), lastSeen: String(b.last_seen_fmt),
      });
    } else {
      bcMedium++;
    }
  }
  const bcFound = findings.filter(f => f.check === "beacons").length;
  checks.push({
    id: "beacons", label: "Beacons",
    status: bcFound ? "findings" : "ok",
    detail: bcFound ? `${bcFound} regular call-out pattern(s)` : bcMedium ? `${bcMedium} medium-score pattern(s), none high` : "no suspicious call-out rhythms",
    suppressed: bcVpn + bcTimer,
  });

  // 3. DNS anomalies — distinct (type, client, domain); benign suffixes suppressed.
  let dnsSuppressed = 0;
  for (const a of (dnsAnoms.data as Array<Record<string, unknown>>)) {
    const type = String(a.anomaly_type);
    // Benign-by-design domains, plus rows that don't meet the current detector
    // bar (older rows predate the query-count floor; real exfil needs volume).
    if (type === "tunneling" && (
      isBenignDomain(String(a.domain)) ||
      Number(a.query_count) < 10 ||
      (Number(a.entropy) <= 4.2 && Number(a.sub_len) <= 45)
    )) { dnsSuppressed++; continue; }
    const label = type === "tunneling"
      ? `DNS tunneling pattern: ${a.client_ip} → ${a.domain}`
      : type === "dga"
        ? `DGA-like lookups from ${a.client_ip}`
        : `DNS volume spike from ${a.client_ip}`;
    findings.push({
      severity: "review", check: "dns",
      title: label,
      detail: type === "tunneling"
        ? `entropy ${Number(a.entropy).toFixed(1)}, avg subdomain ${Math.round(Number(a.sub_len))} chars, ${a.query_count} queries (${a.occurrences} detection(s))`
        : type === "dga"
          ? `many unique domains with >50% NXDOMAIN (${a.occurrences} detection(s))`
          : `${a.query_count} queries in 5 min (${a.occurrences} detection(s))`,
      count: Number(a.query_count), lastSeen: String(a.last_seen),
    });
  }
  const dnsFound = findings.filter(f => f.check === "dns").length;
  checks.push({
    id: "dns", label: "DNS",
    status: dnsFound ? "findings" : "ok",
    detail: dnsFound ? `${dnsFound} unusual pattern(s)` : "query patterns look normal",
    suppressed: dnsSuppressed,
  });

  // 4. IDS — severity-1 and egress-direction alerts only; inbound scan/DROP-list
  //    noise is defense activity, not a finding.
  for (const s of (idsSev1.data as Array<Record<string, unknown>>)) {
    const cat = String(s.alert_category).toLowerCase();
    const hot = cat.includes("malware") || cat.includes("trojan") || cat.includes("command");
    findings.push({
      severity: hot ? "action" : "review", check: "ids",
      title: `IDS priority-1: ${s.alert_signature}`,
      detail: `${s.cnt}× ${s.src_ip} → ${s.dst_ip} (${s.alert_category})`,
      count: Number(s.cnt), lastSeen: String(s.last_seen),
    });
  }
  let idsEgressSuppressed = 0;
  for (const s of (idsEgress.data as Array<Record<string, unknown>>)) {
    const sig = String(s.alert_signature);
    if (BENIGN_EGRESS_IDS.some(b => sig.includes(b.sig) && inCidr(String(s.dst_ip), b.cidr))) {
      idsEgressSuppressed++;
      continue;
    }
    findings.push({
      severity: "review", check: "ids",
      title: `Outbound IDS alert: ${s.alert_signature}`,
      detail: `${s.cnt}× ${s.src_ip} → ${s.dst_ip} — traffic leaving the network tripped a signature`,
      count: Number(s.cnt), lastSeen: String(s.last_seen),
    });
  }
  const idsFound = findings.filter(f => f.check === "ids").length;
  const idsNoise = Number((idsTotal.data?.[0] as Record<string, unknown>)?.cnt ?? 0);
  checks.push({
    id: "ids", label: "IDS",
    status: idsFound ? "findings" : "ok",
    detail: idsFound
      ? `${idsFound} alert group(s) beyond perimeter noise`
      : idsEgressSuppressed
        ? `only perimeter noise + ${idsEgressSuppressed} known-benign egress pattern(s)`
        : "only inbound scan/blocklist noise",
    suppressed: Math.max(0, idsNoise - findings.filter(f => f.check === "ids").reduce((s, f) => s + f.count, 0)),
  });

  // 5. CrowdSec — local detections (our logs tripped a scenario) are findings;
  //    the community blocklist size is defense strength.
  const cs = await getCrowdsecSplit();
  for (const d of cs.localDecisions.slice(0, 20)) {
    findings.push({
      severity: "review", check: "crowdsec",
      title: `CrowdSec local detection: ${d.value} (${d.scenario})`,
      detail: `${d.type} for ${d.duration}, origin ${d.origin} — our own logs triggered this, not the community feed`,
      count: 1,
    });
  }
  checks.push({
    id: "crowdsec", label: "CrowdSec",
    status: cs.local > 0 ? "findings" : "ok",
    detail: cs.local > 0 ? `${cs.local} local decision(s)` : `no local detections; ${cs.community.toLocaleString()} community IPs pre-banned`,
    suppressed: 0,
  });

  // 6. Infrastructure health.
  const pfRow = pf.data?.[0] as Record<string, unknown> | undefined;
  const pfCur = Number(pfRow?.current ?? 0), pfLim = Number(pfRow?.lim ?? 0);
  for (const g of (gateways.data as Array<Record<string, unknown>>)) {
    const loss = Number(g.loss), delay = Number(g.delay), status = Number(g.status);
    if (status !== 1 || loss > 2 || delay > 100) {
      findings.push({
        severity: status !== 1 ? "action" : "review", check: "infra",
        title: status !== 1 ? `Gateway ${g.gw} is offline` : `Gateway ${g.gw} degraded`,
        detail: `delay ${delay.toFixed(1)} ms, loss ${loss.toFixed(1)}%`,
        count: 1,
      });
    }
  }
  if (pfLim > 0 && pfCur / pfLim > 0.8) {
    findings.push({
      severity: "review", check: "infra",
      title: "pf state table above 80%",
      detail: `${pfCur.toLocaleString()} of ${pfLim.toLocaleString()} states in use`,
      count: pfCur,
    });
  }
  const infraFound = findings.filter(f => f.check === "infra").length;
  checks.push({
    id: "infra", label: "Infra",
    status: infraFound ? "findings" : "ok",
    detail: infraFound ? `${infraFound} issue(s)` : "gateways healthy, state table normal",
    suppressed: 0,
  });

  // 7. Telemetry freshness — a quiet detector must not read as a safe network.
  const nowSec = Math.floor(Date.now() / 1000);
  const staleLimits: Record<string, number> = {
    firewall_logs: 900, netflow: 900, telegraf_metrics: 900, dns_queries: 900,
    suricata_alerts: 28 * 3600, threat_intel: 26 * 3600,
  };
  const staleSources: string[] = [];
  for (const f of (freshness.data as Array<{ src: string; last: number }>)) {
    const age = nowSec - Number(f.last);
    if (age > (staleLimits[f.src] ?? 900)) {
      staleSources.push(`${f.src} (${Math.round(age / 3600)}h old)`);
    }
  }
  if (staleSources.length > 0) {
    findings.push({
      severity: "review", check: "telemetry",
      title: `Monitoring blind spot: ${staleSources.length} source(s) stale`,
      detail: staleSources.join(", ") + " — the verdict can't see what isn't flowing",
      count: staleSources.length,
    });
  }
  checks.push({
    id: "telemetry", label: "Telemetry",
    status: staleSources.length ? "stale" : "ok",
    detail: staleSources.length ? staleSources.join(", ") : "all six sources current",
    suppressed: 0,
  });

  findings.sort((a, b) => (a.severity === b.severity ? b.count - a.count : a.severity === "action" ? -1 : 1));
  const status = findings.some(f => f.severity === "action") ? "action"
    : findings.length > 0 ? "review" : "clear";
  const headline = status === "action"
    ? "Possible compromise indicator — act on the findings below."
    : status === "review"
      ? `${findings.length} item${findings.length === 1 ? "" : "s"} worth a look. No confirmed compromise signals.`
      : "No signs of compromise. Perimeter noise is normal and being blocked.";

  const blockedRow = blocked.data?.[0] as Record<string, unknown> | undefined;
  const dnsRow = dnsCounts.data?.[0] as Record<string, unknown> | undefined;
  const eventsAnalyzed =
    Number((fwTotal.data?.[0] as Record<string, unknown>)?.cnt ?? 0) +
    Number((nfTotal.data?.[0] as Record<string, unknown>)?.cnt ?? 0) +
    Number(dnsRow?.total ?? 0) + idsNoise;

  return {
    status, headline, findings, checks,
    defense: {
      blockedProbes: Number(blockedRow?.cnt ?? 0),
      uniqueScanSources: Number(blockedRow?.srcs ?? 0),
      idsAlerts: idsNoise,
      idsSources: Number((idsTotal.data?.[0] as Record<string, unknown>)?.srcs ?? 0),
      crowdsecCommunity: cs.community,
      crowdsecLocal: cs.local,
      dnsQueries: Number(dnsRow?.total ?? 0),
      dnsBlocked: Number(dnsRow?.blocked ?? 0),
      intelFeeds: feeds.data,
    },
    eventsAnalyzed,
    range,
    generatedAt: new Date().toISOString(),
  };
}

// ─── HTTP server ───
Bun.serve({
  port: PORT,
  hostname: "0.0.0.0",
  idleTimeout: 30,

  routes: {
    // Named-query batch — the ONLY ClickHouse read path for the browser.
    // Replaces the raw-SQL /api/query[/batch] routes this service once had:
    // the client names a registered query + range key; the server owns the SQL.
    "/api/data/batch": {
      async POST(req: Request) {
        try {
          const { queries } = (await req.json()) as { queries: Array<{ name: string; range?: string }> };
          if (!Array.isArray(queries) || queries.length === 0 || queries.length > 50) {
            logThrottled("batch-bad-request", 60_000, "warn", "Rejected /api/data/batch request", {
              route: "/api/data/batch", reason: "queries must be a non-empty array (max 50)",
              count: Array.isArray(queries) ? queries.length : "not-array",
            });
            return Response.json({ error: "queries must be a non-empty array (max 50)" }, { status: 400, headers: corsHeaders(req.headers.get("origin")) });
          }
          const results: Record<string, { data: Record<string, unknown>[]; error?: string }> = {};
          await Promise.all(queries.map(async ({ name, range }) => {
            const build = PAGE_QUERIES[name];
            const rangeCfg = OPN_RANGES[range || "1h"];
            if (!build || !rangeCfg) {
              results[name] = { data: [], error: !build ? `unknown query: ${name}` : `unknown range: ${range}` };
              return;
            }
            results[name] = await chQueryRO(build(rangeCfg));
          }));
          return Response.json({ results }, { headers: corsHeaders(req.headers.get("origin")) });
        } catch (e: unknown) {
          return respond500("/api/data/batch", e, req);
        }
      },
    },

    "/api/geo/enrich": {
      POST: handleGeoEnrich,
    },

    "/api/ids/alerts": {
      async GET(req: Request) {
        try {
          const data = await opnPost("/ids/service/query_alerts", { current: 1, rowCount: 50 });
          return Response.json(data, { headers: corsHeaders(req.headers.get("origin")) });
        } catch (e: unknown) {
          return respond500("/api/ids/alerts", e, req);
        }
      },
    },

    "/api/dns/totals": {
      async GET(req: Request) {
        try {
          return Response.json(await opnGet("/unbound/overview/totals/60"), { headers: corsHeaders(req.headers.get("origin")) });
        } catch (e: unknown) {
          return respond500("/api/dns/totals", e, req);
        }
      },
    },

    "/api/dns/rolling": {
      async GET(req: Request) {
        try {
          return Response.json(await opnGet("/unbound/overview/_rolling/10"), { headers: corsHeaders(req.headers.get("origin")) });
        } catch (e: unknown) {
          return respond500("/api/dns/rolling", e, req);
        }
      },
    },

    "/api/dns/overrides": {
      async GET(req: Request) {
        try {
          const data = await opnPost("/unbound/settings/searchHostOverride", {});
          return Response.json(data, { headers: corsHeaders(req.headers.get("origin")) });
        } catch (e: unknown) {
          log.warn("OPNsense host-override lookup failed, returning empty rows", {
            route: "/api/dns/overrides", err: String((e as Error)?.message ?? e),
          });
          return Response.json({ rows: [] }, { status: 500, headers: corsHeaders(req.headers.get("origin")) });
        }
      },
    },

    "/api/crowdsec/stats": {
      async GET(req: Request) {
        try {
          const cs = await getCrowdsecSplit();
          return Response.json({
            decisions: cs.sample,
            totalCount: cs.total,
            communityCount: cs.community,
            localCount: cs.local,
            localDecisions: cs.localDecisions,
          }, { headers: corsHeaders(req.headers.get("origin")) });
        } catch (e: unknown) {
          log.warn("CrowdSec stats request failed, returning empty payload", {
            route: "/api/crowdsec/stats", err: String((e as Error)?.message ?? e),
          });
          return Response.json({ decisions: [], totalCount: 0, communityCount: 0, localCount: 0, localDecisions: [] }, { headers: corsHeaders(req.headers.get("origin")) });
        }
      },
    },

    "/api/suricata/alerts": {
      async GET(req: Request) {
        try {
          const url = new URL(req.url);
          const range = parseRange(url.searchParams.get("range"), "24 HOUR");

          const [alerts, bySeverity, topSigs, total] = await Promise.all([
            chQueryRO(`
              SELECT formatDateTime(timestamp,'%Y-%m-%d %H:%i:%S') AS ts,
                     src_ip, dst_ip, src_port, dst_port, protocol,
                     alert_action, alert_signature, alert_category,
                     alert_severity, alert_sid
              FROM opnsense.suricata_alerts
              WHERE timestamp > now() - INTERVAL ${range}
              ORDER BY timestamp DESC LIMIT 100
            `),
            chQueryRO(`
              SELECT alert_severity, count() AS cnt
              FROM opnsense.suricata_alerts
              WHERE timestamp > now() - INTERVAL ${range}
              GROUP BY alert_severity ORDER BY alert_severity
            `),
            chQueryRO(`
              SELECT alert_signature, alert_category, alert_severity,
                     count() AS cnt
              FROM opnsense.suricata_alerts
              WHERE timestamp > now() - INTERVAL ${range}
              GROUP BY alert_signature, alert_category, alert_severity
              ORDER BY cnt DESC LIMIT 10
            `),
            chQueryRO(`
              SELECT count() AS total
              FROM opnsense.suricata_alerts
              WHERE timestamp > now() - INTERVAL ${range}
            `),
          ]);

          return Response.json({
            alerts: alerts.data,
            by_severity: bySeverity.data,
            top_signatures: topSigs.data,
            total: Number(total.data[0]?.total ?? 0),
          }, { headers: corsHeaders(req.headers.get("origin")) });
        } catch (e: unknown) {
          return respond500("/api/suricata/alerts", e, req);
        }
      },
    },

    "/api/suricata/report": {
      async GET(req: Request) {
        try {
          const url = new URL(req.url);
          const range = parseRange(url.searchParams.get("range"), "24 HOUR");
          const result = await chQueryRO(`
            SELECT formatDateTime(timestamp,'%Y-%m-%d %H:%i:%S') AS ts,
                   src_ip, dst_ip, src_port, dst_port, protocol,
                   alert_action, alert_signature, alert_category,
                   alert_severity, alert_sid
            FROM opnsense.suricata_alerts
            WHERE timestamp > now() - INTERVAL ${range}
            ORDER BY timestamp DESC
          `);
          const csv = ["timestamp,src_ip,dst_ip,src_port,dst_port,protocol,action,signature,category,severity,sid"]
            .concat(result.data.map((r: Record<string, unknown>) =>
              `${r.ts},${r.src_ip},${r.dst_ip},${r.src_port},${r.dst_port},${r.protocol},${r.alert_action},"${r.alert_signature}","${r.alert_category}",${r.alert_severity},${r.alert_sid}`
            )).join("\n");
          return new Response(csv, {
            headers: {
              ...corsHeaders(req.headers.get("origin")),
              "Content-Type": "text/csv",
              "Content-Disposition": `attachment; filename="suricata-alerts-${new Date().toISOString().slice(0,10)}.csv"`,
            },
          });
        } catch (e: unknown) {
          return respond500("/api/suricata/report", e, req);
        }
      },
    },

    "/api/threat-intel/matches": {
      async GET(req: Request) {
        try {
          const url = new URL(req.url);
          const range = parseRange(url.searchParams.get("range"), "24 HOUR");
          // Match individual IPs (ipsum, feodo) and CIDR ranges (spamhaus)
          const result = await chQueryRO(`
            SELECT formatDateTime(fl.timestamp,'%Y-%m-%d %H:%i:%S') AS ts,
                   fl.src_ip, fl.dst_ip, fl.dst_port, fl.protocol_name AS proto,
                   fl.interface, ti.source, ti.confidence
            FROM opnsense.firewall_logs fl
            INNER JOIN opnsense.threat_intel ti
              ON fl.dst_ip = ti.ip
            WHERE fl.action='pass' AND fl.direction='out'
              AND fl.timestamp > now() - INTERVAL ${range}
            ORDER BY fl.timestamp DESC LIMIT 200
          `);
          const summary = await chQueryRO(`
            SELECT count() AS total_matches,
                   uniqExact(fl.dst_ip) AS unique_bad_ips,
                   uniqExact(fl.src_ip) AS unique_internal_devices
            FROM opnsense.firewall_logs fl
            INNER JOIN opnsense.threat_intel ti
              ON fl.dst_ip = ti.ip
            WHERE fl.action='pass' AND fl.direction='out'
              AND fl.timestamp > now() - INTERVAL ${range}
          `);
          const feedStatus = await chQueryRO(`
            SELECT source, count() AS ips, max(inserted_at) AS last_updated
            FROM opnsense.threat_intel GROUP BY source
          `);
          return Response.json({
            matches: result.data,
            summary: summary.data[0] || { total_matches: 0, unique_bad_ips: 0, unique_internal_devices: 0 },
            feeds: feedStatus.data,
          }, { headers: corsHeaders(req.headers.get("origin")) });
        } catch (e: unknown) {
          return respond500("/api/threat-intel/matches", e, req);
        }
      },
    },

    "/api/threat-intel/report": {
      async GET(req: Request) {
        try {
          const url = new URL(req.url);
          const range = parseRange(url.searchParams.get("range"), "24 HOUR");
          const result = await chQueryRO(`
            SELECT formatDateTime(fl.timestamp,'%Y-%m-%d %H:%i:%S') AS ts,
                   fl.src_ip, fl.dst_ip, fl.dst_port, fl.protocol_name AS proto,
                   fl.interface, ti.source, ti.confidence
            FROM opnsense.firewall_logs fl
            INNER JOIN opnsense.threat_intel ti
              ON fl.dst_ip = ti.ip
            WHERE fl.action='pass' AND fl.direction='out'
              AND fl.timestamp > now() - INTERVAL ${range}
            ORDER BY fl.timestamp DESC
          `);
          const csv = ["timestamp,src_ip,dst_ip,dst_port,protocol,interface,threat_source,confidence"]
            .concat(result.data.map((r: Record<string, unknown>) =>
              `${r.ts},${r.src_ip},${r.dst_ip},${r.dst_port},${r.proto},${r.interface},${r.source},${r.confidence}`
            )).join("\n");
          return new Response(csv, {
            headers: {
              ...corsHeaders(req.headers.get("origin")),
              "Content-Type": "text/csv",
              "Content-Disposition": `attachment; filename="threat-intel-report-${new Date().toISOString().slice(0,10)}.csv"`,
            },
          });
        } catch (e: unknown) {
          return respond500("/api/threat-intel/report", e, req);
        }
      },
    },

    "/api/threat-intel/refresh": {
      async POST(req: Request) {
        try {
          const result = await refreshThreatFeeds();
          return Response.json(result, { headers: corsHeaders(req.headers.get("origin")) });
        } catch (e: unknown) {
          return respond500("/api/threat-intel/refresh", e, req);
        }
      },
    },

    "/api/threat-intel/beacons": {
      async GET(req: Request) {
        try {
          const url = new URL(req.url);
          const minScoreRaw = parseFloat(url.searchParams.get("min_score") || "0.3");
          const minScore = Number.isFinite(minScoreRaw) ? Math.min(Math.max(minScoreRaw, 0), 1) : 0.3;
          const limitRaw = parseInt(url.searchParams.get("limit") || "20", 10);
          const limit = Number.isFinite(limitRaw) ? Math.min(Math.max(limitRaw, 1), 500) : 20;
          const range = parseRange(url.searchParams.get("range"), "");
          const rangeFilter = range ? `AND last_seen > now() - INTERVAL ${range}` : "";
          const result = await chQueryRO(`
            SELECT src_ip, dst_ip, dst_port, connection_count,
                   beacon_score, interval_mode, interval_range, interval_cv,
                   size_cv, total_bytes, avg_bytes,
                   formatDateTime(first_seen, '%Y-%m-%d %H:%i:%S') AS first_seen,
                   formatDateTime(last_seen, '%Y-%m-%d %H:%i:%S') AS last_seen,
                   formatDateTime(scored_at, '%Y-%m-%d %H:%i:%S') AS scored_at
            FROM (
              SELECT * FROM opnsense.beacon_scores FINAL
              WHERE beacon_score >= ${minScore} ${rangeFilter}
              ORDER BY beacon_score DESC
              LIMIT ${limit}
            )
          `);
          const meta = await chQueryRO(`
            SELECT count() AS total_pairs,
                   countIf(beacon_score >= 0.85) AS high,
                   countIf(beacon_score >= 0.7 AND beacon_score < 0.85) AS medium,
                   countIf(beacon_score >= 0.3 AND beacon_score < 0.7) AS low,
                   max(scored_at) AS last_scored
            FROM opnsense.beacon_scores FINAL
            WHERE 1=1 ${rangeFilter}
          `);
          return Response.json({
            beacons: result.data,
            summary: meta.data[0] || { total_pairs: 0, high: 0, medium: 0, low: 0, last_scored: null },
          }, { headers: corsHeaders(req.headers.get("origin")) });
        } catch (e: unknown) {
          return respond500("/api/threat-intel/beacons", e, req);
        }
      },
    },

    "/api/threat-intel/beacons/report": {
      async GET(req: Request) {
        try {
          const result = await chQueryRO(`
            SELECT src_ip, dst_ip, dst_port, connection_count,
                   beacon_score, interval_mode, interval_range, interval_cv,
                   size_cv, total_bytes, avg_bytes,
                   formatDateTime(first_seen, '%Y-%m-%d %H:%i:%S') AS first_seen,
                   formatDateTime(last_seen, '%Y-%m-%d %H:%i:%S') AS last_seen
            FROM opnsense.beacon_scores FINAL
            WHERE beacon_score >= 0.3
            ORDER BY beacon_score DESC
          `);
          const csv = ["src_ip,dst_ip,dst_port,connections,beacon_score,interval_avg_s,interval_range_s,interval_cv,size_cv,total_bytes,avg_bytes,first_seen,last_seen"]
            .concat(result.data.map((r: Record<string, unknown>) =>
              `${r.src_ip},${r.dst_ip},${r.dst_port},${r.connection_count},${r.beacon_score},${r.interval_mode},${r.interval_range},${r.interval_cv},${r.size_cv},${r.total_bytes},${r.avg_bytes},${r.first_seen},${r.last_seen}`
            )).join("\n");
          return new Response(csv, {
            headers: {
              ...corsHeaders(req.headers.get("origin")),
              "Content-Type": "text/csv",
              "Content-Disposition": `attachment; filename="beacon-scores-${new Date().toISOString().slice(0,10)}.csv"`,
            },
          });
        } catch (e: unknown) {
          return respond500("/api/threat-intel/beacons/report", e, req);
        }
      },
    },

    "/api/threat-intel/dns": {
      async GET(req: Request) {
        try {
          const url = new URL(req.url);
          const range = parseRange(url.searchParams.get("range"), "1 HOUR");

          const [anomalies, queryStats, topDomains, topQueried, topBlocked] = await Promise.all([
            chQueryRO(`
              SELECT formatDateTime(detected_at, '%Y-%m-%d %H:%i:%S') AS ts,
                     anomaly_type, client_ip, domain, query_count,
                     avg_subdomain_len, avg_entropy, nxdomain_ratio, details
              FROM opnsense.dns_anomalies
              WHERE detected_at > now() - INTERVAL ${range}
              ORDER BY detected_at DESC LIMIT 50
            `),
            chQueryRO(`
              SELECT count() AS total_queries,
                     countIf(is_blocked = 1) AS blocked,
                     uniqExact(domain) AS unique_domains,
                     countIf(rcode = 'NXDOMAIN') AS nxdomain,
                     avg(entropy) AS avg_entropy
              FROM opnsense.dns_queries
              WHERE timestamp > now() - INTERVAL ${range}
            `),
            chQueryRO(`
              SELECT domain, count() AS cnt, avg(entropy) AS avg_entropy,
                     max(subdomain_len) AS max_sub_len
              FROM opnsense.dns_queries
              WHERE timestamp > now() - INTERVAL ${range}
                AND subdomain != ''
              GROUP BY domain
              ORDER BY avg_entropy DESC
              LIMIT 10
            `),
            chQueryRO(`
              SELECT domain, count() AS cnt
              FROM opnsense.dns_queries
              WHERE timestamp > now() - INTERVAL ${range}
              GROUP BY domain
              ORDER BY cnt DESC
              LIMIT 10
            `),
            chQueryRO(`
              SELECT domain, count() AS cnt
              FROM opnsense.dns_queries
              WHERE timestamp > now() - INTERVAL ${range}
                AND is_blocked = 1
              GROUP BY domain
              ORDER BY cnt DESC
              LIMIT 10
            `),
          ]);

          return Response.json({
            anomalies: anomalies.data,
            stats: queryStats.data[0] || { total_queries: 0, blocked: 0, unique_domains: 0, nxdomain: 0, avg_entropy: 0 },
            high_entropy_domains: topDomains.data,
            top_queried: topQueried.data,
            top_blocked: topBlocked.data,
          }, { headers: corsHeaders(req.headers.get("origin")) });
        } catch (e: unknown) {
          return respond500("/api/threat-intel/dns", e, req);
        }
      },
    },

    "/api/threat-intel/dns/report": {
      async GET(req: Request) {
        try {
          const url = new URL(req.url);
          const range = parseRange(url.searchParams.get("range"), "1 HOUR");
          const result = await chQueryRO(`
            SELECT formatDateTime(detected_at, '%Y-%m-%d %H:%i:%S') AS ts,
                   anomaly_type, client_ip, domain, query_count,
                   avg_subdomain_len, avg_entropy, nxdomain_ratio, details
            FROM opnsense.dns_anomalies
            WHERE detected_at > now() - INTERVAL ${range}
            ORDER BY detected_at DESC
          `);
          const csv = ["timestamp,type,client_ip,domain,query_count,avg_subdomain_len,avg_entropy,nxdomain_ratio,details"]
            .concat(result.data.map((r: Record<string, unknown>) =>
              `${r.ts},${r.anomaly_type},${r.client_ip},"${r.domain}",${r.query_count},${r.avg_subdomain_len},${r.avg_entropy},${r.nxdomain_ratio},"${r.details}"`
            )).join("\n");
          return new Response(csv, {
            headers: {
              ...corsHeaders(req.headers.get("origin")),
              "Content-Type": "text/csv",
              "Content-Disposition": `attachment; filename="dns-anomalies-${new Date().toISOString().slice(0,10)}.csv"`,
            },
          });
        } catch (e: unknown) {
          return respond500("/api/threat-intel/dns/report", e, req);
        }
      },
    },

    "/api/verdict": {
      async GET(req: Request) {
        try {
          const url = new URL(req.url);
          const range = parseRange(url.searchParams.get("range"), "24 HOUR");
          const verdict = await computeVerdict(range);
          return Response.json(verdict, { headers: corsHeaders(req.headers.get("origin")) });
        } catch (e: unknown) {
          return respond500("/api/verdict", e, req);
        }
      },
    },

    "/api/health": {
      GET(req: Request) {
        return Response.json({ status: "ok", kafka: kafkaProducer !== null }, { headers: corsHeaders(req.headers.get("origin")) });
      },
    },

    "/api/*": Response.json({ error: "Not found" }, { status: 404 }),
  },

  // Handle CORS preflight for all routes
  async fetch(req) {
    if (req.method === "OPTIONS") {
      return new Response(null, { status: 204, headers: corsHeaders(req.headers.get("origin")) });
    }
    return new Response("Not found", { status: 404, headers: corsHeaders(req.headers.get("origin")) });
  },
});

// ─── Threat Intelligence Feed Loader ───
const THREAT_FEED_INTERVAL = 6 * 60 * 60 * 1000; // 6 hours

async function refreshThreatFeeds(): Promise<{ ipsum: number; feodo: number; blocklistde: number }> {
  let ipsumCount = 0, feodoCount = 0, spamhausCount = 0;

  // Insert into staging table, then swap — avoids empty window if all feeds fail
  await chQuery("CREATE TABLE IF NOT EXISTS opnsense.threat_intel_staging AS opnsense.threat_intel");
  await chQuery("TRUNCATE TABLE opnsense.threat_intel_staging");

  // 1. IPsum Level 3+ (IPs on 3+ blacklists)
  try {
    const resp = await fetch("https://raw.githubusercontent.com/stamparm/ipsum/master/levels/3.txt");
    const text = await resp.text();
    const ips = text.split("\n")
      .filter(l => l && !l.startsWith("#"))
      .map(l => l.trim().split(/\s/)[0])
      .filter(ip => ip && /^\d+\.\d+\.\d+\.\d+$/.test(ip));
    if (ips.length > 0) {
      const rows = ips.map(ip => JSON.stringify({ ip, source: "ipsum-l3", confidence: 3 })).join("\n");
      await fetch(`${CLICKHOUSE_URL}/?query=${encodeURIComponent("INSERT INTO opnsense.threat_intel_staging FORMAT JSONEachRow")}`, {
        method: "POST", body: rows,
      });
      ipsumCount = ips.length;
    }
    log.info("Threat Intel: IPsum L3+ loaded", { ips: ipsumCount });
  } catch (e) { log.warn("Threat Intel: IPsum fetch failed (staging keeps prior data)", { err: String((e as Error)?.message ?? e) }); }

  // 2. Feodo Tracker (botnet C2)
  try {
    const resp = await fetch("https://feodotracker.abuse.ch/downloads/ipblocklist_recommended.txt");
    const text = await resp.text();
    const ips = text.split("\n")
      .filter(l => l && !l.startsWith("#"))
      .map(l => l.trim())
      .filter(ip => ip && /^\d+\.\d+\.\d+\.\d+$/.test(ip));
    if (ips.length > 0) {
      const rows = ips.map(ip => JSON.stringify({ ip, source: "feodo-c2", confidence: 5 })).join("\n");
      await fetch(`${CLICKHOUSE_URL}/?query=${encodeURIComponent("INSERT INTO opnsense.threat_intel_staging FORMAT JSONEachRow")}`, {
        method: "POST", body: rows,
      });
      feodoCount = ips.length;
    }
    log.info("Threat Intel: Feodo Tracker loaded", { ips: feodoCount });
  } catch (e) { log.warn("Threat Intel: Feodo fetch failed (staging keeps prior data)", { err: String((e as Error)?.message ?? e) }); }

  // 3. Blocklist.de (active attackers, individual IPs)
  try {
    const resp = await fetch("https://lists.blocklist.de/lists/all.txt");
    const text = await resp.text();
    const ips = text.split("\n")
      .map(l => l.trim())
      .filter(ip => ip && /^\d+\.\d+\.\d+\.\d+$/.test(ip));
    if (ips.length > 0) {
      // Batch insert in chunks to avoid massive request body
      for (let i = 0; i < ips.length; i += 5000) {
        const chunk = ips.slice(i, i + 5000);
        const rows = chunk.map(ip => JSON.stringify({ ip, source: "blocklist-de", confidence: 3 })).join("\n");
        await fetch(`${CLICKHOUSE_URL}/?query=${encodeURIComponent("INSERT INTO opnsense.threat_intel_staging FORMAT JSONEachRow")}`, {
          method: "POST", body: rows,
        });
      }
      spamhausCount = ips.length;
    }
    log.info("Threat Intel: Blocklist.de loaded", { ips: spamhausCount });
  } catch (e) { log.warn("Threat Intel: Blocklist.de fetch failed (staging keeps prior data)", { err: String((e as Error)?.message ?? e) }); }

  // Only swap if at least one feed loaded — avoids leaving table empty on total failure
  const totalLoaded = ipsumCount + feodoCount + spamhausCount;
  if (totalLoaded > 0) {
    await chQuery("EXCHANGE TABLES opnsense.threat_intel AND opnsense.threat_intel_staging");
    await chQuery("DROP TABLE IF EXISTS opnsense.threat_intel_staging");
    log.info("Threat Intel: swapped feeds into live table", {
      total: totalLoaded, ipsum: ipsumCount, feodo: feodoCount, blocklistde: spamhausCount,
    });
  } else {
    await chQuery("DROP TABLE IF EXISTS opnsense.threat_intel_staging");
    log.warn("Threat Intel: all feeds failed, keeping existing data");
  }

  return { ipsum: ipsumCount, feodo: feodoCount, blocklistde: spamhausCount };
}

// ─── NetFlow v9 Collector ───
const NETFLOW_PORT = parseInt(process.env.NETFLOW_PORT || "2055");
const NETFLOW_FLUSH_INTERVAL = 10_000;
let netflowBuffer: Array<Record<string, unknown>> = [];
let netflowFlowCount = 0;

interface NetFlowRecord {
  in_bytes?: number;
  in_pkts?: number;
  protocol?: number;
  tcp_flags?: number;
  l4_src_port?: number;
  l4_dst_port?: number;
  ipv4_src_addr?: string;
  ipv4_dst_addr?: string;
  ipv6_src_addr?: string;
  ipv6_dst_addr?: string;
  input_snmp?: number;
  output_snmp?: number;
}

function onNetFlowData(data: { flows: NetFlowRecord[] }) {
  const now = new Date().toISOString().replace("T", " ").replace("Z", "");
  for (const flow of data.flows) {
    const srcIp = flow.ipv4_src_addr || flow.ipv6_src_addr || "";
    const dstIp = flow.ipv4_dst_addr || flow.ipv6_dst_addr || "";
    if (!srcIp || !dstIp) continue;

    netflowBuffer.push({
      timestamp: now,
      src_ip: srcIp,
      dst_ip: dstIp,
      src_port: flow.l4_src_port || 0,
      dst_port: flow.l4_dst_port || 0,
      protocol: flow.protocol || 0,
      in_bytes: flow.in_bytes || 0,
      in_packets: flow.in_pkts || 0,
      tcp_flags: flow.tcp_flags || 0,
      in_interface: flow.input_snmp || 0,
      out_interface: flow.output_snmp || 0,
    });
  }
  netflowFlowCount += data.flows.length;
}

async function flushNetflow() {
  if (netflowBuffer.length === 0) return;
  const batch = netflowBuffer;
  netflowBuffer = [];
  netflowFlowCount = 0;

  try {
    const body = batch.map((r) => JSON.stringify(r)).join("\n");
    const resp = await fetch(
      `${CLICKHOUSE_URL}/?query=${encodeURIComponent("INSERT INTO opnsense.netflow FORMAT JSONEachRow")}`,
      { method: "POST", body }
    );
    if (!resp.ok) {
      const err = await resp.text();
      log.error("NetFlow ClickHouse insert failed (flows dropped)", { status: resp.status, err: err.slice(0, 500), flows: batch.length });
    }
  } catch (e) {
    log.error("NetFlow flush failed (flows dropped)", { err: String((e as Error)?.message ?? e), stack: (e as Error)?.stack, flows: batch.length });
  }
}

// ── Inline NetFlow v9 decoder ──
// Templates: map of templateId → array of { id, length }
const nfTemplates = new Map<number, Array<{ id: number; len: number }>>();
function decodeIpv4(buf: Uint8Array, off: number): string {
  return `${buf[off]}.${buf[off+1]}.${buf[off+2]}.${buf[off+3]}`;
}

function readUint(buf: DataView, off: number, len: number): number {
  if (len === 1) return buf.getUint8(off);
  if (len === 2) return buf.getUint16(off);
  if (len === 4) return buf.getUint32(off);
  if (len === 8) return Number(buf.getBigUint64(off));
  return 0;
}

function decodeNetFlowPacket(data: Uint8Array): NetFlowRecord[] {
  if (data.length < 20) return [];
  const dv = new DataView(data.buffer, data.byteOffset, data.byteLength);
  const version = dv.getUint16(0);
  if (version !== 9) return [];

  let pos = 20; // skip header (version, count, sysUptime, unixSecs, seqNum, sourceId)
  const flows: NetFlowRecord[] = [];

  while (pos + 4 <= data.length) {
    const fsId = dv.getUint16(pos);
    const fsLen = dv.getUint16(pos + 2);
    if (fsLen < 4 || pos + fsLen > data.length) break;

    if (fsId === 0) {
      // Template flowset
      let tPos = pos + 4;
      while (tPos + 4 <= pos + fsLen) {
        const tId = dv.getUint16(tPos);
        const fieldCount = dv.getUint16(tPos + 2);
        tPos += 4;
        const fields: Array<{ id: number; len: number }> = [];
        for (let i = 0; i < fieldCount && tPos + 4 <= pos + fsLen; i++) {
          fields.push({ id: dv.getUint16(tPos), len: dv.getUint16(tPos + 2) });
          tPos += 4;
        }
        nfTemplates.set(tId, fields);
        log.debug("NetFlow: template registered", { templateId: tId, fields: fields.length });
      }
    } else if (fsId > 255) {
      // Data flowset - decode using stored template
      const tmpl = nfTemplates.get(fsId);
      if (!tmpl) { pos += fsLen; continue; }
      const recLen = tmpl.reduce((s, f) => s + f.len, 0);
      if (recLen === 0) { pos += fsLen; continue; }

      let rPos = pos + 4;
      while (rPos + recLen <= pos + fsLen) {
        const rec: NetFlowRecord = {};
        let fOff = rPos;
        for (const field of tmpl) {
          switch (field.id) {
            case 1: rec.in_bytes = readUint(dv, fOff, field.len); break;
            case 2: rec.in_pkts = readUint(dv, fOff, field.len); break;
            case 4: rec.protocol = readUint(dv, fOff, field.len); break;
            case 6: rec.tcp_flags = readUint(dv, fOff, field.len); break;
            case 7: rec.l4_src_port = readUint(dv, fOff, field.len); break;
            case 8: if (field.len === 4) rec.ipv4_src_addr = decodeIpv4(data, fOff); break;
            case 10: rec.input_snmp = readUint(dv, fOff, field.len); break;
            case 11: rec.l4_dst_port = readUint(dv, fOff, field.len); break;
            case 12: if (field.len === 4) rec.ipv4_dst_addr = decodeIpv4(data, fOff); break;
            case 14: rec.output_snmp = readUint(dv, fOff, field.len); break;
            case 27: // ipv6_src_addr (16 bytes)
              if (field.len === 16) rec.ipv6_src_addr = Array.from(data.slice(fOff, fOff+16)).map((b,i) => (i%2===0 && i>0 ? ':' : '') + b.toString(16).padStart(2,'0')).join('');
              break;
            case 28: // ipv6_dst_addr
              if (field.len === 16) rec.ipv6_dst_addr = Array.from(data.slice(fOff, fOff+16)).map((b,i) => (i%2===0 && i>0 ? ':' : '') + b.toString(16).padStart(2,'0')).join('');
              break;
          }
          fOff += field.len;
        }
        flows.push(rec);
        rPos += recLen;
      }
    }
    // fsId === 1: options template - skip
    pos += fsLen;
  }
  return flows;
}

// Templates live in memory, so every restart is blind until OPNsense re-sends
// them (naturally minutes). Kicking netflow/reconfigure on boot makes ng_netflow
// re-emit templates within seconds.
async function kickNetflowTemplates(attempt = 1): Promise<void> {
  try {
    // OPNsense reconfigure endpoints 400 without a JSON body — send {}.
    const res = await opnPost("/diagnostics/netflow/reconfigure", {}) as { status?: string };
    log.info("NetFlow: template kick (netflow/reconfigure)", { status: res?.status ?? JSON.stringify(res) });
  } catch (e) {
    log.warn("NetFlow: template kick failed", { attempt, err: String((e as Error)?.message ?? e) });
    if (attempt < 2) setTimeout(() => kickNetflowTemplates(attempt + 1), 15_000);
  }
}

let nfSocket: dgram.Socket | null = null;
try {
  nfSocket = dgram.createSocket({ type: "udp4", reuseAddr: true });
  nfSocket.on("error", (err) => {
    log.error("NetFlow socket error", { err: err.message, stack: err.stack });
  });
  nfSocket.on("message", (msg: Buffer) => {
    try {
      const decoded = decodeNetFlowPacket(new Uint8Array(msg));
      if (decoded.length > 0) onNetFlowData({ flows: decoded });
    } catch (e) {
      logThrottled("netflow-decode", 60_000, "warn", "NetFlow packet decode failed", { err: String((e as Error)?.message ?? e) });
    }
  });
  nfSocket.bind(NETFLOW_PORT, () => {
    log.info("NetFlow collector listening", { port: NETFLOW_PORT, flushSec: NETFLOW_FLUSH_INTERVAL / 1000 });
    kickNetflowTemplates();
  });
  setInterval(flushNetflow, NETFLOW_FLUSH_INTERVAL);
} catch (e) {
  log.error("NetFlow collector failed to start", { err: String((e as Error)?.message ?? e), stack: (e as Error)?.stack });
}

// ─── Start Kafka + KPI loop ───
await initKafka();
setInterval(publishKpis, KPI_INTERVAL);
// First publish immediately
setTimeout(publishKpis, 2000);

log.info("Firewall API started", {
  port: PORT,
  brokers: KAFKA_BROKERS.join(","),
  topic: KAFKA_TOPIC,
  kpiIntervalSec: KPI_INTERVAL / 1000,
  logFormat: (process.env.LOG_FORMAT || "logfmt"),
  logLevel: (process.env.LOG_LEVEL || "info"),
});

// ─── Threat Intel feed loader ───
setTimeout(() => refreshThreatFeeds().catch(e => log.error("Initial threat feed load failed", { err: String((e as Error)?.message ?? e), stack: (e as Error)?.stack })), 5000);
setInterval(() => refreshThreatFeeds().catch(e => log.error("Threat feed refresh failed", { err: String((e as Error)?.message ?? e), stack: (e as Error)?.stack })), THREAT_FEED_INTERVAL);

// ─── Beacon scoring (RITA-style) ───
const BEACON_SCORE_INTERVAL = 60 * 60 * 1000; // 1 hour

// Site-specific: known-benign destinations that beacon by design (cloud
// telemetry endpoints, vendor keepalives). Comma-separated IPs, empty = none.
const BEACON_DST_EXCLUDE = (process.env.BEACON_DST_EXCLUDE || "").split(",").map(s => s.trim()).filter(Boolean);

const BEACON_SCORING_SQL = `
INSERT INTO opnsense.beacon_scores
SELECT
  now() AS scored_at, src_ip, dst_ip, dst_port, conn_count AS connection_count,
  interval_mode, interval_range, interval_skew, interval_cv,
  size_mode, size_range, size_cv,
  round(greatest(0, least(1,
    (1 - least(interval_cv, 1.0)) * 0.4 +
    (1 - least(size_cv, 1.0)) * 0.2 +
    least(conn_count / 200.0, 1.0) * 0.2 +
    (1 - least(interval_range / 3600.0, 1.0)) * 0.2
  )), 4) AS beacon_score,
  total_bytes, avg_bytes, first_seen, last_seen
FROM (
  SELECT
    src_ip, dst_ip, dst_port, conn_count,
    total_bytes, avg_bytes, first_seen, last_seen,
    round(arrayAvg(diffs_ms) / 1000, 2) AS interval_mode,
    round((arrayMax(diffs_ms) - arrayMin(diffs_ms)) / 1000, 2) AS interval_range,
    if(arrayAvg(diffs_ms) > 0,
       round(arrayReduce('stddevPop', arrayMap(x -> toFloat64(x), diffs_ms)) / arrayAvg(diffs_ms), 4), 99) AS interval_cv,
    0 AS interval_skew,
    round(arrayAvg(sizes), 2) AS size_mode,
    toFloat64(arrayMax(sizes) - arrayMin(sizes)) AS size_range,
    if(arrayAvg(sizes) > 0,
       round(arrayReduce('stddevPop', sizes) / arrayAvg(sizes), 4), 99) AS size_cv
  FROM (
    SELECT
      src_ip, dst_ip, dst_port,
      count() AS conn_count,
      arraySlice(arrayDifference(arraySort(groupArray(toUnixTimestamp64Milli(timestamp)))), 2) AS diffs_ms,
      groupArray(toFloat64(in_bytes)) AS sizes,
      sum(in_bytes) AS total_bytes,
      avg(in_bytes) AS avg_bytes,
      min(timestamp) AS first_seen,
      max(timestamp) AS last_seen
    FROM opnsense.netflow
    WHERE timestamp > now() - INTERVAL 24 HOUR
      AND dst_port NOT IN (53, 123, 2055, 5353, 5228, 5229, 5230)
      AND NOT (dst_port BETWEEN 6881 AND 6999)
      AND dst_port NOT IN (51413, 51414)
      ${VPN_SRC_PREFIXES.map(p => `AND NOT (src_ip LIKE '${p}%')`).join("\n      ")}
      AND NOT (src_ip LIKE '10.%' AND dst_ip LIKE '10.%')
      AND NOT (src_ip LIKE '192.168.%' AND dst_ip LIKE '192.168.%')
      AND NOT (dst_ip LIKE '10.%')
      AND NOT (dst_ip LIKE '192.168.%')
      AND NOT (dst_ip LIKE '224.%')
      AND NOT (dst_ip LIKE '255.%')
      ${WAN_SRC_IP ? `AND NOT (src_ip = '${WAN_SRC_IP}')` : ""}
      ${BEACON_DST_EXCLUDE.length ? `AND dst_ip NOT IN (${BEACON_DST_EXCLUDE.map(ip => `'${ip}'`).join(", ")})` : ""}
    GROUP BY src_ip, dst_ip, dst_port
    HAVING conn_count >= 20
  )
)
`;

async function scoreBeacons() {
  try {
    const resp = await fetch(`${CLICKHOUSE_URL}/`, { method: "POST", body: BEACON_SCORING_SQL });
    if (!resp.ok) {
      log.error("Beacon scoring insert failed", { status: resp.status, err: (await resp.text()).slice(0, 500) });
      return;
    }
    const countResult = await chQuery("SELECT count() AS cnt FROM opnsense.beacon_scores FINAL");
    log.info("Beacon scoring complete", { pairs: countResult.data[0]?.cnt ?? 0 });
  } catch (e) {
    log.error("Beacon scoring failed", { err: String((e as Error)?.message ?? e), stack: (e as Error)?.stack });
  }
}

setTimeout(() => scoreBeacons(), 30_000);
setInterval(() => scoreBeacons(), BEACON_SCORE_INTERVAL);

// ─── DNS Intelligence (query collector + anomaly detection) ───
const DNS_POLL_INTERVAL = 60_000; // 1 minute
const DNS_ANOMALY_INTERVAL = 60 * 60 * 1000; // 1 hour
const DNS_STATS_INTERVAL = 5 * 60 * 1000; // 5 minutes
let dnsLastSeenTime = 0; // track last ingested timestamp to avoid duplicates

function shannonEntropy(s: string): number {
  if (s.length === 0) return 0;
  const freq: Record<string, number> = {};
  for (const c of s) freq[c] = (freq[c] || 0) + 1;
  let entropy = 0;
  for (const f of Object.values(freq)) {
    const p = f / s.length;
    if (p > 0) entropy -= p * Math.log2(p);
  }
  return Math.round(entropy * 1000) / 1000;
}

function extractDomain(fqdn: string): { domain: string; subdomain: string } {
  const name = fqdn.replace(/\.$/, "").toLowerCase();
  const parts = name.split(".");
  if (parts.length <= 2) return { domain: name, subdomain: "" };
  // Handle ccTLDs like co.uk, com.au
  const tld = parts[parts.length - 1];
  const sld = parts[parts.length - 2];
  const ccTlds = ["co", "com", "org", "net", "gov", "edu", "ac"];
  if (tld.length === 2 && ccTlds.includes(sld) && parts.length >= 3) {
    const domain = parts.slice(-3).join(".");
    const subdomain = parts.slice(0, -3).join(".");
    return { domain, subdomain };
  }
  const domain = parts.slice(-2).join(".");
  const subdomain = parts.slice(0, -2).join(".");
  return { domain, subdomain };
}

async function collectDnsQueries() {
  try {
    const data = await opnGet("/unbound/overview/search_queries") as {
      rows: Array<{ time: number; client: string; domain: string; type: string; action: string; rcode: string; source: string }>;
    };
    if (!data.rows || data.rows.length === 0) return;

    // Filter to only new queries (after last seen timestamp)
    const newRows = dnsLastSeenTime > 0
      ? data.rows.filter(r => r.time > dnsLastSeenTime)
      : data.rows;

    if (newRows.length === 0) return;

    // Update watermark
    dnsLastSeenTime = Math.max(...newRows.map(r => r.time));

    // Transform and insert
    const chRows = newRows.map(r => {
      const { domain, subdomain } = extractDomain(r.domain);
      return JSON.stringify({
        timestamp: new Date(r.time * 1000).toISOString().replace("T", " ").replace("Z", ""),
        client_ip: r.client,
        query_name: r.domain.replace(/\.$/, ""),
        query_type: r.type,
        domain,
        subdomain,
        subdomain_len: subdomain.length,
        entropy: shannonEntropy(subdomain),
        action: r.action,
        rcode: r.rcode || "",
        source: r.source || "",
        is_blocked: r.action === "Block" ? 1 : 0,
      });
    }).join("\n");

    const resp = await fetch(
      `${CLICKHOUSE_URL}/?query=${encodeURIComponent("INSERT INTO opnsense.dns_queries FORMAT JSONEachRow")}`,
      { method: "POST", body: chRows }
    );
    if (!resp.ok) logThrottled("dns-query-insert", 60_000, "error", "DNS query insert failed (queries dropped)", { status: resp.status, err: (await resp.text()).slice(0, 500) });
  } catch (e) {
    logThrottled("dns-collect", 60_000, "warn", "DNS query collection failed (upstream OPNsense)", { err: String((e as Error)?.message ?? e) });
  }
}

async function detectDnsAnomalies() {
  try {
    // Tunneling detection: high entropy AND long subdomains (conjunctive — either
    // alone false-positives on CDN hash hostnames), or extreme length/volume.
    // Benign-by-design domains (plex.direct etc.) are excluded via DNS_BENIGN_SUFFIXES.
    const benignFilter = DNS_BENIGN_SUFFIXES
      .map(s => `domain != '${s}' AND NOT endsWith(domain, '.${s}')`)
      .join(" AND ");
    const tunneling = await chQuery(`
      SELECT client_ip, domain,
             count() AS query_count,
             avg(subdomain_len) AS avg_sub_len,
             max(subdomain_len) AS max_sub_len,
             avg(entropy) AS avg_entropy
      FROM opnsense.dns_queries
      WHERE timestamp > now() - INTERVAL 1 HOUR
        AND subdomain != ''
        AND domain NOT IN ('grafana.org', 'grafana.com')
        AND ${benignFilter}
      GROUP BY client_ip, domain
      HAVING (avg_entropy > 4.2 AND avg_sub_len > 20 AND query_count >= 25)
          OR (avg_sub_len > 45 AND query_count >= 10)
      ORDER BY avg_entropy DESC
    `);

    // DGA detection: many unique domains + high NXDOMAIN ratio
    const dga = await chQuery(`
      SELECT client_ip,
             count() AS total_queries,
             uniqExact(domain) AS unique_domains,
             countIf(rcode = 'NXDOMAIN') AS nxdomain_count
      FROM opnsense.dns_queries
      WHERE timestamp > now() - INTERVAL 1 HOUR
      GROUP BY client_ip
      HAVING unique_domains > 50 AND nxdomain_count > total_queries * 0.5
    `);

    // Volume spike: >500 queries in 5 minutes from one device
    const volume = await chQuery(`
      SELECT client_ip,
             toStartOfInterval(timestamp, INTERVAL 5 MINUTE) AS bucket,
             count() AS queries
      FROM opnsense.dns_queries
      WHERE timestamp > now() - INTERVAL 1 HOUR
      GROUP BY client_ip, bucket
      HAVING queries > 500
      ORDER BY queries DESC
    `);

    // Insert anomalies
    const anomalies: string[] = [];
    for (const t of (tunneling.data || [])) {
      anomalies.push(JSON.stringify({
        anomaly_type: "tunneling", client_ip: t.client_ip, domain: t.domain,
        query_count: Number(t.query_count), avg_subdomain_len: Number(t.avg_sub_len),
        avg_entropy: Number(t.avg_entropy), details: JSON.stringify({ max_sub_len: t.max_sub_len }),
      }));
    }
    for (const d of (dga.data || [])) {
      anomalies.push(JSON.stringify({
        anomaly_type: "dga", client_ip: d.client_ip, query_count: Number(d.total_queries),
        nxdomain_ratio: Number(d.total_queries) > 0 ? Number(d.nxdomain_count) / Number(d.total_queries) : 0,
        details: JSON.stringify({ unique_domains: d.unique_domains, nxdomain_count: d.nxdomain_count }),
      }));
    }
    for (const v of (volume.data || [])) {
      anomalies.push(JSON.stringify({
        anomaly_type: "volume_spike", client_ip: v.client_ip,
        query_count: Number(v.queries), details: JSON.stringify({ bucket: v.bucket }),
      }));
    }

    if (anomalies.length > 0) {
      await fetch(
        `${CLICKHOUSE_URL}/?query=${encodeURIComponent("INSERT INTO opnsense.dns_anomalies FORMAT JSONEachRow")}`,
        { method: "POST", body: anomalies.join("\n") }
      );
    }

    const qryCount = await chQuery("SELECT count() AS cnt FROM opnsense.dns_queries WHERE timestamp > now() - INTERVAL 1 HOUR");
    log.info("DNS anomaly scan complete", {
      queries: qryCount.data[0]?.cnt ?? 0, anomalies: anomalies.length,
      tunneling: tunneling.data?.length || 0, dga: dga.data?.length || 0, volume: volume.data?.length || 0,
    });
  } catch (e) {
    log.error("DNS anomaly detection failed", { err: String((e as Error)?.message ?? e), stack: (e as Error)?.stack });
  }
}

// DNS totals snapshot → opnsense.dns_stats. Feeds the firewall page's DNS QUERIES
// and DNS BLOCKED KPIs (dashboard sums this table over the selected range). This
// collector lives here (not in a host cron) on purpose: the service already holds
// both ends — ClickHouse via CLICKHOUSE_URL and OPNsense via opnGet — so the
// snapshot can't silently go stale when hosts get rebuilt.
async function collectDnsStats() {
  try {
    const d = await opnGet("/unbound/overview/totals/60") as {
      total?: number; passed?: number; blocklist_size?: number; blocked?: { total?: number };
    };
    const row = JSON.stringify({
      total: Math.trunc(Number(d.total ?? 0)),
      blocked: Math.trunc(Number(d.blocked?.total ?? 0)),
      passed: Math.trunc(Number(d.passed ?? 0)),
      blocklist_size: Math.trunc(Number(d.blocklist_size ?? 0)),
    });
    const resp = await fetch(
      `${CLICKHOUSE_URL}/?query=${encodeURIComponent("INSERT INTO opnsense.dns_stats FORMAT JSONEachRow")}`,
      { method: "POST", body: row }
    );
    if (!resp.ok) logThrottled("dns-stats-insert", 60_000, "error", "DNS stats insert failed", { status: resp.status, err: (await resp.text()).slice(0, 500) });
  } catch (e) {
    logThrottled("dns-stats-collect", 60_000, "warn", "DNS stats collection failed (upstream OPNsense)", { err: String((e as Error)?.message ?? e) });
  }
}

setTimeout(() => collectDnsQueries(), 10_000);
setInterval(() => collectDnsQueries(), DNS_POLL_INTERVAL);
setTimeout(() => detectDnsAnomalies(), 60_000);
setInterval(() => detectDnsAnomalies(), DNS_ANOMALY_INTERVAL);
setTimeout(() => collectDnsStats(), 15_000);
setInterval(() => collectDnsStats(), DNS_STATS_INTERVAL);

// ─── Graceful shutdown ───
async function shutdown(signal: string) {
  log.info("Shutting down", { signal });
  try {
    await flushNetflow();
    log.info("NetFlow buffer flushed");
  } catch (e) {
    log.error("NetFlow flush on shutdown failed", { err: String((e as Error)?.message ?? e), stack: (e as Error)?.stack });
  }
  try {
    if (nfSocket) {
      nfSocket.close();
      log.info("NetFlow socket closed");
    }
  } catch (e) {
    log.error("NetFlow socket close failed", { err: String((e as Error)?.message ?? e), stack: (e as Error)?.stack });
  }
  try {
    if (kafkaProducer) {
      await kafkaProducer.disconnect();
      log.info("Kafka disconnected");
    }
  } catch (e) {
    log.error("Kafka disconnect on shutdown failed", { err: String((e as Error)?.message ?? e), stack: (e as Error)?.stack });
  }
  process.exit(0);
}

process.on("SIGTERM", () => shutdown("SIGTERM"));
process.on("SIGINT", () => shutdown("SIGINT"));