TypeScript 55.4%
Python 43.2%
SQL 1.2%
1import type { FastifyInstance } from "fastify";2import type { WebSocket } from "ws";3import Redis from "ioredis";4import { eventGroupOf, FEED_CHANNELS } from "@websensor/core";5import { db, sql } from "@websensor/db";6import { config } from "./config";78/**9 * WebSocket gateway `/api/v1/live`. One Redis subscriber fans out to every client; clients10 * pick channels: events:global · events:breaking · events:silent · events:first-party ·11 * events:<ai|cyber|finance|health|government|science|products|infrastructure|news> · group:<security|…> ·12 * country:<CA> · type:<event_type> · entity:<id> · source:<id> · watchlist:<id>.13 * Protocol (JSON): client → {"subscribe":[…]} | {"unsubscribe":[…]} | {"ping":1} | {"since":"<sid>"}14 * server → {"type":"hello"} | {"type":"event", "sid":"…", "channels":[…], "event":{…}} | {"type":"replay_done"} | {"type":"pong"} | {"type":"heartbeat"}15 * Every event frame carries the Redis stream id `sid`; after a reconnection the client sends16 * {"since": lastSid} and missed events (up to 500) are replayed from the durable stream (spec §60).17 */18interface Client {19 ws: WebSocket;20 channels: Set<string>;21 watchlists: Map<string, WatchRules>;22}23interface WatchRules {24 entities: Set<string>;25 sources: Set<string>;26 keywords: string[];27 categories: Set<string>;28 types: Set<string>;29 countries: Set<string>;30 urls: string[];31}3233const clients = new Set<Client>();34let sub: Redis | null = null;35let cmd: Redis | null = null;36let published = 0;3738export function liveStats(): { clients: number; published: number } {39 return { clients: clients.size, published };40}4142/** Engine heartbeat written by the scheduler (`ws:engine:status`). */43export async function engineStatus(): Promise<Record<string, unknown> | null> {44 try {45 const raw = await getCmd().get("ws:engine:status");46 return raw ? (JSON.parse(raw) as Record<string, unknown>) : null;47 } catch {48 return null;49 }50}5152export async function factoryStatus(): Promise<Record<string, unknown> | null> {53 try {54 const raw = await getCmd().get("ws:factory:status");55 return raw ? (JSON.parse(raw) as Record<string, unknown>) : null;56 } catch {57 return null;58 }59}6061function getCmd(): Redis {62 if (!cmd) cmd = new Redis(config.redisUrl, { maxRetriesPerRequest: 2, lazyConnect: false });63 return cmd;64}6566export async function registerLive(app: FastifyInstance): Promise<void> {67 sub = new Redis(config.redisUrl, { maxRetriesPerRequest: 3 });68 sub.on("error", (e) => app.log.warn({ err: e.message }, "redis sub error"));69 getCmd().on("error", (e) => app.log.warn({ err: e.message }, "redis cmd error"));70 await sub.subscribe("ws:live");71 sub.on("message", (_ch, msg) => {72 let ev: Record<string, unknown>;73 try {74 ev = JSON.parse(msg) as Record<string, unknown>;75 } catch {76 return;77 }78 published++;79 deliver(ev, String(ev.sid ?? ""));80 });8182 app.get("/api/v1/live", { websocket: true }, (socket) => {83 const client: Client = { ws: socket, channels: new Set(["events:global"]), watchlists: new Map() };84 clients.add(client);85 socket.send(JSON.stringify({ type: "hello", channels: [...client.channels], serverTime: new Date().toISOString(), protocol: 2 }));86 socket.on("message", async (raw: Buffer | string) => {87 let msg: { subscribe?: string[]; unsubscribe?: string[]; ping?: number; since?: string };88 try {89 msg = JSON.parse(raw.toString()) as typeof msg;90 } catch {91 return;92 }93 if (msg.ping) socket.send(JSON.stringify({ type: "pong", t: Date.now() }));94 for (const ch of msg.subscribe ?? []) {95 if (typeof ch !== "string" || ch.length > 120 || client.channels.size > 64) continue;96 if (ch.startsWith("watchlist:")) await loadWatchlist(client, ch.slice(10));97 else client.channels.add(ch);98 }99 for (const ch of msg.unsubscribe ?? []) {100 client.channels.delete(ch);101 if (ch.startsWith("watchlist:")) client.watchlists.delete(ch.slice(10));102 }103 if (msg.subscribe || msg.unsubscribe) socket.send(JSON.stringify({ type: "subscribed", channels: [...client.channels, ...[...client.watchlists.keys()].map((w) => `watchlist:${w}`)] }));104 if (typeof msg.since === "string" && /^\d{10,16}-\d{1,6}$/.test(msg.since)) await replay(client, msg.since);105 });106 const hb = setInterval(() => {107 if (socket.readyState === socket.OPEN) socket.send(JSON.stringify({ type: "heartbeat", t: Date.now() }));108 }, 25_000);109 socket.on("close", () => {110 clearInterval(hb);111 clients.delete(client);112 });113 socket.on("error", () => {114 clearInterval(hb);115 clients.delete(client);116 });117 });118}119120function deliver(ev: Record<string, unknown>, sid: string, only?: Client): void {121 const chans = channelsFor(ev);122 for (const c of only ? [only] : clients) {123 const hit = [...chans].filter((ch) => c.channels.has(ch));124 for (const [wid, w] of c.watchlists) if (matchesWatchlist(ev, w)) hit.push(`watchlist:${wid}`);125 if (!hit.length) continue;126 if (c.ws.readyState === c.ws.OPEN) c.ws.send(JSON.stringify({ type: "event", sid, channels: hit, event: ev }));127 }128}129130/** Replay missed events from the durable stream (exclusive of `since`). */131async function replay(client: Client, since: string): Promise<void> {132 try {133 const [ms, seq] = since.split("-");134 const start = `${ms}-${Number(seq) + 1}`;135 const rows = (await getCmd().xrange("ws:events", start, "+", "COUNT", 500)) as [string, string[]][];136 let n = 0;137 for (const [sid, fields] of rows) {138 const i = fields.indexOf("event");139 if (i < 0) continue;140 try {141 const ev = JSON.parse(fields[i + 1]!) as Record<string, unknown>;142 deliver({ ...ev, sid, replayed: true }, sid, client);143 n++;144 } catch {145 // skip malformed146 }147 }148 if (client.ws.readyState === client.ws.OPEN) client.ws.send(JSON.stringify({ type: "replay_done", since, count: n, truncated: rows.length >= 500 }));149 } catch {150 if (client.ws.readyState === client.ws.OPEN) client.ws.send(JSON.stringify({ type: "replay_done", since, count: 0, error: "replay_unavailable" }));151 }152}153154export function channelsFor(ev: Record<string, unknown>): Set<string> {155 const out = new Set<string>(["events:global"]);156 const importance = Number(ev.importance ?? 0);157 const signal = Number(ev.signal ?? importance);158 if (signal >= 80 || importance >= 80) out.add("events:breaking");159 if (ev.silent) out.add("events:silent");160 if (ev.firstParty !== false) out.add("events:first-party");161 const cats = (ev.categories as string[] | undefined) ?? [];162 for (const [ch, list] of Object.entries(FEED_CHANNELS)) if (list.some((c) => cats.includes(c)) || cats.includes(ch)) out.add(`events:${ch}`);163 const src = ev.source as { id?: string } | undefined;164 if (src?.id) out.add(`source:${src.id}`);165 for (const e of (ev.entities as { id: string }[] | undefined) ?? []) out.add(`entity:${e.id}`);166 out.add(`type:${String(ev.type)}`);167 out.add(`group:${String(ev.group ?? eventGroupOf(String(ev.type)))}`);168 if (ev.country) out.add(`country:${String(ev.country).toUpperCase()}`);169 if (ev.clusterState === "breaking" || ev.clusterState === "developing") out.add(`state:${String(ev.clusterState)}`);170 return out;171}172173async function loadWatchlist(client: Client, id: string): Promise<void> {174 const rows = await db.execute<{ kind: string; value: string }>(sql`select kind, value from watchlist_items where watchlist_id = ${id}`);175 const w: WatchRules = { entities: new Set(), sources: new Set(), keywords: [], categories: new Set(), types: new Set(), countries: new Set(), urls: [] };176 for (const r of rows.rows) {177 if (r.kind === "entity") w.entities.add(r.value);178 else if (r.kind === "source") w.sources.add(r.value);179 else if (r.kind === "keyword") w.keywords.push(r.value.toLowerCase());180 else if (r.kind === "category") w.categories.add(r.value);181 else if (r.kind === "event_type") w.types.add(r.value);182 else if (r.kind === "country") w.countries.add(r.value.toUpperCase());183 else if (r.kind === "url") w.urls.push(r.value.replace(/\/$/, ""));184 }185 client.watchlists.set(id, w);186}187188function matchesWatchlist(ev: Record<string, unknown>, w: WatchRules): boolean {189 const src = ev.source as { id?: string } | undefined;190 if (src?.id && w.sources.has(src.id)) return true;191 for (const e of (ev.entities as { id: string }[] | undefined) ?? []) if (w.entities.has(e.id)) return true;192 for (const c of (ev.categories as string[] | undefined) ?? []) if (w.categories.has(c)) return true;193 if (w.types.has(String(ev.type))) return true;194 if (ev.country && w.countries.has(String(ev.country).toUpperCase())) return true;195 const url = String(ev.url ?? "");196 if (w.urls.some((u) => url === u || url.startsWith(u + "/"))) return true;197 if (w.keywords.length) {198 const hay = `${String(ev.title)} ${String(ev.summary)}`.toLowerCase();199 if (w.keywords.some((k) => hay.includes(k))) return true;200 }201 return false;202}203204export async function closeLive(): Promise<void> {205 for (const c of clients) c.ws.close(1001, "server shutdown");206 clients.clear();207 if (sub) await sub.quit().catch(() => undefined);208 if (cmd) await cmd.quit().catch(() => undefined);209}210