import type { FastifyInstance } from "fastify";
import type { WebSocket } from "ws";
import Redis from "ioredis";
import { eventGroupOf, FEED_CHANNELS } from "@websensor/core";
import { db, sql } from "@websensor/db";
import { config } from "./config";

/**
 * WebSocket gateway `/api/v1/live`. One Redis subscriber fans out to every client; clients
 * pick channels: events:global · events:breaking · events:silent · events:first-party ·
 * events:<ai|cyber|finance|health|government|science|products|infrastructure|news> · group:<security|…> ·
 * country:<CA> · type:<event_type> · entity:<id> · source:<id> · watchlist:<id>.
 * Protocol (JSON): client → {"subscribe":[…]} | {"unsubscribe":[…]} | {"ping":1} | {"since":"<sid>"}
 *                  server → {"type":"hello"} | {"type":"event", "sid":"…", "channels":[…], "event":{…}} | {"type":"replay_done"} | {"type":"pong"} | {"type":"heartbeat"}
 * Every event frame carries the Redis stream id `sid`; after a reconnection the client sends
 * {"since": lastSid} and missed events (up to 500) are replayed from the durable stream (spec §60).
 */
interface Client {
  ws: WebSocket;
  channels: Set<string>;
  watchlists: Map<string, WatchRules>;
}
interface WatchRules {
  entities: Set<string>;
  sources: Set<string>;
  keywords: string[];
  categories: Set<string>;
  types: Set<string>;
  countries: Set<string>;
  urls: string[];
}

const clients = new Set<Client>();
let sub: Redis | null = null;
let cmd: Redis | null = null;
let published = 0;

export function liveStats(): { clients: number; published: number } {
  return { clients: clients.size, published };
}

/** Engine heartbeat written by the scheduler (`ws:engine:status`). */
export async function engineStatus(): Promise<Record<string, unknown> | null> {
  try {
    const raw = await getCmd().get("ws:engine:status");
    return raw ? (JSON.parse(raw) as Record<string, unknown>) : null;
  } catch {
    return null;
  }
}

export async function factoryStatus(): Promise<Record<string, unknown> | null> {
  try {
    const raw = await getCmd().get("ws:factory:status");
    return raw ? (JSON.parse(raw) as Record<string, unknown>) : null;
  } catch {
    return null;
  }
}

function getCmd(): Redis {
  if (!cmd) cmd = new Redis(config.redisUrl, { maxRetriesPerRequest: 2, lazyConnect: false });
  return cmd;
}

export async function registerLive(app: FastifyInstance): Promise<void> {
  sub = new Redis(config.redisUrl, { maxRetriesPerRequest: 3 });
  sub.on("error", (e) => app.log.warn({ err: e.message }, "redis sub error"));
  getCmd().on("error", (e) => app.log.warn({ err: e.message }, "redis cmd error"));
  await sub.subscribe("ws:live");
  sub.on("message", (_ch, msg) => {
    let ev: Record<string, unknown>;
    try {
      ev = JSON.parse(msg) as Record<string, unknown>;
    } catch {
      return;
    }
    published++;
    deliver(ev, String(ev.sid ?? ""));
  });

  app.get("/api/v1/live", { websocket: true }, (socket) => {
    const client: Client = { ws: socket, channels: new Set(["events:global"]), watchlists: new Map() };
    clients.add(client);
    socket.send(JSON.stringify({ type: "hello", channels: [...client.channels], serverTime: new Date().toISOString(), protocol: 2 }));
    socket.on("message", async (raw: Buffer | string) => {
      let msg: { subscribe?: string[]; unsubscribe?: string[]; ping?: number; since?: string };
      try {
        msg = JSON.parse(raw.toString()) as typeof msg;
      } catch {
        return;
      }
      if (msg.ping) socket.send(JSON.stringify({ type: "pong", t: Date.now() }));
      for (const ch of msg.subscribe ?? []) {
        if (typeof ch !== "string" || ch.length > 120 || client.channels.size > 64) continue;
        if (ch.startsWith("watchlist:")) await loadWatchlist(client, ch.slice(10));
        else client.channels.add(ch);
      }
      for (const ch of msg.unsubscribe ?? []) {
        client.channels.delete(ch);
        if (ch.startsWith("watchlist:")) client.watchlists.delete(ch.slice(10));
      }
      if (msg.subscribe || msg.unsubscribe) socket.send(JSON.stringify({ type: "subscribed", channels: [...client.channels, ...[...client.watchlists.keys()].map((w) => `watchlist:${w}`)] }));
      if (typeof msg.since === "string" && /^\d{10,16}-\d{1,6}$/.test(msg.since)) await replay(client, msg.since);
    });
    const hb = setInterval(() => {
      if (socket.readyState === socket.OPEN) socket.send(JSON.stringify({ type: "heartbeat", t: Date.now() }));
    }, 25_000);
    socket.on("close", () => {
      clearInterval(hb);
      clients.delete(client);
    });
    socket.on("error", () => {
      clearInterval(hb);
      clients.delete(client);
    });
  });
}

function deliver(ev: Record<string, unknown>, sid: string, only?: Client): void {
  const chans = channelsFor(ev);
  for (const c of only ? [only] : clients) {
    const hit = [...chans].filter((ch) => c.channels.has(ch));
    for (const [wid, w] of c.watchlists) if (matchesWatchlist(ev, w)) hit.push(`watchlist:${wid}`);
    if (!hit.length) continue;
    if (c.ws.readyState === c.ws.OPEN) c.ws.send(JSON.stringify({ type: "event", sid, channels: hit, event: ev }));
  }
}

/** Replay missed events from the durable stream (exclusive of `since`). */
async function replay(client: Client, since: string): Promise<void> {
  try {
    const [ms, seq] = since.split("-");
    const start = `${ms}-${Number(seq) + 1}`;
    const rows = (await getCmd().xrange("ws:events", start, "+", "COUNT", 500)) as [string, string[]][];
    let n = 0;
    for (const [sid, fields] of rows) {
      const i = fields.indexOf("event");
      if (i < 0) continue;
      try {
        const ev = JSON.parse(fields[i + 1]!) as Record<string, unknown>;
        deliver({ ...ev, sid, replayed: true }, sid, client);
        n++;
      } catch {
        // skip malformed
      }
    }
    if (client.ws.readyState === client.ws.OPEN) client.ws.send(JSON.stringify({ type: "replay_done", since, count: n, truncated: rows.length >= 500 }));
  } catch {
    if (client.ws.readyState === client.ws.OPEN) client.ws.send(JSON.stringify({ type: "replay_done", since, count: 0, error: "replay_unavailable" }));
  }
}

export function channelsFor(ev: Record<string, unknown>): Set<string> {
  const out = new Set<string>(["events:global"]);
  const importance = Number(ev.importance ?? 0);
  const signal = Number(ev.signal ?? importance);
  if (signal >= 80 || importance >= 80) out.add("events:breaking");
  if (ev.silent) out.add("events:silent");
  if (ev.firstParty !== false) out.add("events:first-party");
  const cats = (ev.categories as string[] | undefined) ?? [];
  for (const [ch, list] of Object.entries(FEED_CHANNELS)) if (list.some((c) => cats.includes(c)) || cats.includes(ch)) out.add(`events:${ch}`);
  const src = ev.source as { id?: string } | undefined;
  if (src?.id) out.add(`source:${src.id}`);
  for (const e of (ev.entities as { id: string }[] | undefined) ?? []) out.add(`entity:${e.id}`);
  out.add(`type:${String(ev.type)}`);
  out.add(`group:${String(ev.group ?? eventGroupOf(String(ev.type)))}`);
  if (ev.country) out.add(`country:${String(ev.country).toUpperCase()}`);
  if (ev.clusterState === "breaking" || ev.clusterState === "developing") out.add(`state:${String(ev.clusterState)}`);
  return out;
}

async function loadWatchlist(client: Client, id: string): Promise<void> {
  const rows = await db.execute<{ kind: string; value: string }>(sql`select kind, value from watchlist_items where watchlist_id = ${id}`);
  const w: WatchRules = { entities: new Set(), sources: new Set(), keywords: [], categories: new Set(), types: new Set(), countries: new Set(), urls: [] };
  for (const r of rows.rows) {
    if (r.kind === "entity") w.entities.add(r.value);
    else if (r.kind === "source") w.sources.add(r.value);
    else if (r.kind === "keyword") w.keywords.push(r.value.toLowerCase());
    else if (r.kind === "category") w.categories.add(r.value);
    else if (r.kind === "event_type") w.types.add(r.value);
    else if (r.kind === "country") w.countries.add(r.value.toUpperCase());
    else if (r.kind === "url") w.urls.push(r.value.replace(/\/$/, ""));
  }
  client.watchlists.set(id, w);
}

function matchesWatchlist(ev: Record<string, unknown>, w: WatchRules): boolean {
  const src = ev.source as { id?: string } | undefined;
  if (src?.id && w.sources.has(src.id)) return true;
  for (const e of (ev.entities as { id: string }[] | undefined) ?? []) if (w.entities.has(e.id)) return true;
  for (const c of (ev.categories as string[] | undefined) ?? []) if (w.categories.has(c)) return true;
  if (w.types.has(String(ev.type))) return true;
  if (ev.country && w.countries.has(String(ev.country).toUpperCase())) return true;
  const url = String(ev.url ?? "");
  if (w.urls.some((u) => url === u || url.startsWith(u + "/"))) return true;
  if (w.keywords.length) {
    const hay = `${String(ev.title)} ${String(ev.summary)}`.toLowerCase();
    if (w.keywords.some((k) => hay.includes(k))) return true;
  }
  return false;
}

export async function closeLive(): Promise<void> {
  for (const c of clients) c.ws.close(1001, "server shutdown");
  clients.clear();
  if (sub) await sub.quit().catch(() => undefined);
  if (cmd) await cmd.quit().catch(() => undefined);
}