TypeScript 55.4%
Python 43.2%
SQL 1.2%
1import type { FastifyInstance } from "fastify";2import { z } from "zod";3import { COUNTRIES, countryBySlug, diffText, EVENT_GROUPS, EVENT_TYPES, FEED_CHANNELS, parseSearch } from "@websensor/core";4import { db, sql, textArray } from "@websensor/db";5import { getBlobStore } from "@websensor/store";6import { cached } from "./cache";7import { config } from "./config";8import { coverageSector, coverageSummary } from "./coverage";9import { breakingDesk, categoryDesk, clusterDetail, countryDesk, countryList, entityInsights, entityRankings, pulse, radar, sourceSensors } from "./intel";10import { engineStatus, liveStats } from "./live";11import { countEvents, getEvent, listEvents, relatedEvents, sensorHistory, sourceActivity, sourceQuality, stats, trending } from "./queries";1213const bool = z14 .enum(["true", "false", "1", "0"])15 .transform((v) => v === "true" || v === "1")16 .optional();1718export const eventsQuery = z.object({19 after: z.string().optional(),20 before: z.string().optional(),21 category: z.string().optional(),22 entity: z.string().optional(),23 source: z.string().optional(),24 domain: z.string().optional(),25 sensor: z.string().optional(),26 cluster: z.string().optional(),27 importance_min: z.coerce.number().min(0).max(100).optional(),28 confidence_min: z.coerce.number().min(0).max(100).optional(),29 signal_min: z.coerce.number().min(0).max(100).optional(),30 event_type: z.string().optional(),31 group: z.string().optional(),32 silent_change: bool,33 first_party: bool,34 confirmed: bool,35 country: z.string().max(3).optional(),36 language: z.string().max(2).optional(),37 change_class: z.string().optional(),38 q: z.string().max(300).optional(),39 limit: z.coerce.number().int().min(1).max(200).default(50),40 cursor: z.string().optional(),41 order: z.enum(["recent", "importance", "signal"]).default("recent"),42});4344export async function registerRoutes(app: FastifyInstance): Promise<void> {45 // ---- Health ---------------------------------------------------------------------------46 app.get("/api/health", async () => ({ status: "ok", service: "api", version: config.version, time: new Date().toISOString() }));47 app.get("/api/ready", async (_req, reply) => {48 const checks: Record<string, boolean | number | string | null> = {};49 const t0 = Date.now();50 try {51 await db.execute(sql`select 1`);52 checks.database = true;53 checks.database_latency_ms = Date.now() - t0;54 } catch {55 checks.database = false;56 }57 const engine = await db.execute<{ last: Date | null }>(sql`select max(started_at) as last from sensor_runs where started_at >= now() - interval '15 minutes'`).catch(() => ({ rows: [{ last: null }] }));58 checks.engine_recent = Boolean(engine.rows[0]?.last);59 const es = await engineStatus();60 checks.engine_heartbeat = es ? String(es.at) : null;61 checks.ws_clients = liveStats().clients;62 return reply.status(checks.database ? 200 : 503).send({ status: checks.database ? "ready" : "degraded", checks });63 });6465 // ---- Events ---------------------------------------------------------------------------66 app.get("/api/v1/events", async (req, reply) => {67 const q = eventsQuery.parse(req.query);68 const page = await listEvents(q);69 reply.header("x-websensor-order", q.order);70 return { ...page, meta: { limit: q.limit, order: q.order, filters: stripEmpty({ ...q, limit: undefined, cursor: undefined, order: undefined }) } };71 });72 app.get("/api/v1/events/count", async (req) => {73 const q = eventsQuery.parse(req.query);74 return { count: await countEvents(q) };75 });76 app.get<{ Params: { id: string } }>("/api/v1/events/:id", async (req, reply) => {77 const ev = await getEvent(req.params.id);78 if (!ev) return reply.status(404).send({ error: "not_found" });79 const [related, cluster, change, interp, history] = await Promise.all([80 relatedEvents(ev),81 ev.cluster_id ? db.execute<Record<string, unknown>>(sql`select * from event_clusters where id = ${String(ev.cluster_id)}`).then((r) => r.rows[0] ?? null) : null,82 ev.change_id ? db.execute<Record<string, unknown>>(sql`select id, kind, diff, signal, noise_ratio, magnitude, heuristic, detected_at, old_snapshot_id, new_snapshot_id, change_class, field_changes from changes where id = ${String(ev.change_id)}`).then((r) => r.rows[0] ?? null) : null,83 db.execute<Record<string, unknown>>(sql`select version, model, created_at from interpretations where event_id = ${String(ev.id)} order by version`).then((r) => r.rows),84 sensorHistory(String(ev.sensor_id), String(ev.id)),85 ]);86 const snaps = await db.execute<Record<string, unknown>>(sql`select id, url, captured_at, http_status, content_type, content_length, content_hash, canonical_hash, etag, last_modified, title, mode, fetch_duration_ms, extraction_confidence, storage_key is not null as has_raw from snapshots where id in (${String(ev.old_snapshot_id ?? "")}, ${String(ev.new_snapshot_id ?? "")})`);87 const sourceRel = await db.execute<Record<string, unknown>>(sql`select health, success_rate, avg_latency_ms, total_runs, raw_changes, meaningful_changes, last_check_at from (select s.health, ch.success_rate, s.avg_latency_ms, s.total_runs, s.raw_changes, s.meaningful_changes, s.last_check_at from sensors s left join connector_health ch on ch.connector = s.connector where s.id = ${String(ev.sensor_id)}) x`);88 return { event: ev, related, cluster, change, interpretations: interp, snapshots: snaps.rows, sensor_reliability: sourceRel.rows[0] ?? null, history };89 });9091 // ---- Changes / snapshots / diffs -------------------------------------------------------92 app.get<{ Params: { id: string } }>("/api/v1/changes/:id", async (req, reply) => {93 const r = await db.execute<Record<string, unknown>>(sql`select c.*, s.url as sensor_url, s.name as sensor_name, s.source_id from changes c join sensors s on s.id = c.sensor_id where c.id = ${req.params.id}`);94 const c = r.rows[0];95 if (!c) return reply.status(404).send({ error: "not_found" });96 let unified: string | null = null;97 if (c.diff_storage_key) unified = await getBlobStore().getText(String(c.diff_storage_key)).catch(() => null);98 return { change: c, unified };99 });100 app.get<{ Params: { id: string }; Querystring: { raw?: string } }>("/api/v1/snapshots/:id", async (req, reply) => {101 const r = await db.execute<Record<string, unknown>>(sql`select * from snapshots where id = ${req.params.id}`);102 const s = r.rows[0];103 if (!s) return reply.status(404).send({ error: "not_found" });104 const store = getBlobStore();105 if (req.query.raw === "1") {106 if (!s.storage_key) return reply.status(410).send({ error: "raw_body_pruned", detail: "The raw body of this snapshot was pruned by the retention policy; the canonical representation and hashes are preserved." });107 const buf = await store.get(String(s.storage_key));108 reply.header("content-type", String(s.content_type ?? "text/plain") + (String(s.content_type ?? "").includes("charset") ? "" : "; charset=utf-8"));109 reply.header("x-content-type-options", "nosniff");110 reply.header("content-security-policy", "default-src 'none'; style-src 'unsafe-inline'; img-src data:");111 return reply.send(buf);112 }113 const canonical = s.canonical_storage_key ? await store.getText(String(s.canonical_storage_key)).catch(() => null) : null;114 return { snapshot: s, canonical: canonical ? JSON.parse(canonical) : null };115 });116 app.get<{ Querystring: { a: string; b: string } }>("/api/v1/snapshots/compare", async (req, reply) => {117 const { a, b } = req.query;118 if (!a || !b) return reply.status(400).send({ error: "a and b required" });119 const rows = await db.execute<Record<string, unknown>>(sql`select id, captured_at, canonical_storage_key, mode, url, sensor_id from snapshots where id in (${a}, ${b})`);120 const sa = rows.rows.find((r) => r.id === a);121 const sb = rows.rows.find((r) => r.id === b);122 if (!sa || !sb) return reply.status(404).send({ error: "not_found" });123 const store = getBlobStore();124 const [ca, cb] = await Promise.all([store.getText(String(sa.canonical_storage_key)), store.getText(String(sb.canonical_storage_key))]);125 const ta = renderCanonical(JSON.parse(ca));126 const tb = renderCanonical(JSON.parse(cb));127 const d = diffText(ta, tb, `${a}@${String(sa.captured_at)}`, `${b}@${String(sb.captured_at)}`);128 return { a: sa, b: sb, before: ta, after: tb, diff: { unified: d.unified, stats: d.stats, added: d.added, removed: d.removed, modified: d.modified } };129 });130 /** Historical memory (spec §79): how a sensor's page looked at any point — list of snapshots with day grouping. */131 app.get<{ Params: { id: string }; Querystring: { limit?: string; before?: string } }>("/api/v1/sensors/:id/snapshots", async (req) => {132 const lim = Math.min(500, Number(req.query.limit ?? 200));133 const rows = await db.execute<Record<string, unknown>>(sql`select id, captured_at, http_status, content_length, canonical_hash, title, mode, storage_key is not null as has_raw, (select count(*) from changes c where c.new_snapshot_id = snapshots.id) > 0 as has_change, (select e.slug from events e where e.new_snapshot_id = snapshots.id limit 1) as event_slug from snapshots where sensor_id = ${req.params.id} ${req.query.before ? sql`and captured_at < ${new Date(req.query.before)}` : sql``} order by captured_at desc limit ${lim}`);134 return { items: rows.rows };135 });136137 // ---- Sources & sensors ------------------------------------------------------------------138 app.get<{ Querystring: { category?: string; q?: string; limit?: string; country?: string; tier?: string; first_party?: string } }>("/api/v1/sources", async (req) => {139 const cat = req.query.category;140 const q = req.query.q;141 const rows = await db.execute<Record<string, unknown>>(sql`142 select s.id, s.name, s.domain, s.homepage, s.description, s.categories, s.tier, s.importance_weight, s.enabled, s.notes, s.first_party, s.country, s.language, s.robots_checked_at,143 (select count(*) from sensors x where x.source_id = s.id and x.enabled and x.status <> 'SHADOW')::int as sensor_count,144 (select count(*) from sensors x where x.source_id = s.id and x.enabled and x.status = 'SHADOW')::int as shadow_count, s.origin, s.sector,145 (select count(*) from events e where e.source_id = s.id)::int as event_count,146 (select count(*) from events e where e.source_id = s.id and e.detected_at >= now() - interval '24 hours')::int as events_24h,147 (select max(detected_at) from events e where e.source_id = s.id) as last_event_at,148 (select max(last_check_at) from sensors x where x.source_id = s.id) as last_check_at,149 (select count(*) from sensors x where x.source_id = s.id and x.enabled and x.health <> 'UP')::int as sensors_degraded150 from sources s where s.enabled and s.kind = 'registry' ${cat ? sql`and ${cat} = any(s.categories)` : sql``} ${q ? sql`and (s.name ilike ${"%" + q + "%"} or s.domain ilike ${"%" + q + "%"})` : sql``} ${req.query.country ? sql`and s.country = ${req.query.country.toUpperCase()}` : sql``} ${req.query.tier ? sql`and s.tier = ${req.query.tier.toUpperCase()}` : sql``} ${req.query.first_party ? sql`and s.first_party = ${req.query.first_party === "true"}` : sql``}151 order by s.tier asc, events_24h desc, s.name asc limit ${Math.min(5000, Number(req.query.limit ?? 300))}`);152 return { items: rows.rows };153 });154 app.get<{ Params: { id: string } }>("/api/v1/sources/:id", async (req, reply) => {155 const s = (await db.execute<Record<string, unknown>>(sql`select * from sources where (id = ${req.params.id} or domain = ${req.params.id}) and kind = 'registry' limit 1`)).rows[0];156 if (!s) return reply.status(404).send({ error: "not_found" });157 const { owner_token: _o, ...pub } = s;158 const [sensors, ents, activity, quality, candidates, byType, series] = await Promise.all([159 sourceSensors(String(s.id)),160 db.execute<Record<string, unknown>>(sql`select en.id, en.name, en.type, en.importance, en.event_count from source_entities se join entities en on en.id = se.entity_id where se.source_id = ${String(s.id)} order by en.type, en.name`).then((r) => r.rows),161 sourceActivity(String(s.id)),162 sourceQuality(String(s.id)),163 db.execute<Record<string, unknown>>(sql`select url, kind, evidence, score, status, found_at from discovery_candidates where source_id = ${String(s.id)} order by (score->>'value')::float desc nulls last limit 50`).then((r) => r.rows),164 db.execute<Record<string, unknown>>(sql`select event_type, count(*)::int as n from events where source_id = ${String(s.id)} group by 1 order by 2 desc limit 16`).then((r) => r.rows),165 db.execute<Record<string, unknown>>(sql`select day::text as day, checks, not_modified, errors, raw_changes, events from source_daily where source_id = ${String(s.id)} and day >= (now() at time zone 'UTC')::date - 30 order by day`).then((r) => r.rows),166 ]);167 return { source: pub, sensors, entities: ents, activity, quality, discovery: candidates, by_type: byType, daily: series };168 });169 app.get<{ Params: { id: string } }>("/api/v1/sensors/:id", async (req, reply) => {170 const s = (await db.execute<Record<string, unknown>>(sql`select s.*, so.name as source_name, so.domain, so.kind as source_kind from sensors s join sources so on so.id = s.source_id where s.id = ${req.params.id}`)).rows[0];171 if (!s || s.source_kind === "custom") return reply.status(404).send({ error: "not_found" });172 const [runs, snaps, changes, events] = await Promise.all([173 db.execute<Record<string, unknown>>(sql`select id, started_at, finished_at, http_status, outcome, error, duration_ms, bytes, fetch_method, snapshot_id from sensor_runs where sensor_id = ${String(s.id)} order by started_at desc limit 50`).then((r) => r.rows),174 db.execute<Record<string, unknown>>(sql`select id, captured_at, http_status, content_type, content_length, canonical_hash, title, mode, extraction_confidence, storage_key is not null as has_raw from snapshots where sensor_id = ${String(s.id)} order by captured_at desc limit 50`).then((r) => r.rows),175 db.execute<Record<string, unknown>>(sql`select id, detected_at, kind, signal, noise_ratio, magnitude, meaningful, event_id, old_snapshot_id, new_snapshot_id, change_class, field_changes, heuristic->>'eventType' as heuristic_type from changes where sensor_id = ${String(s.id)} order by detected_at desc limit 50`).then((r) => r.rows),176 listEvents({ sensor: String(s.id), limit: 30 }).then((r) => r.items),177 ]);178 const { state: _st, source_kind: _k, ...pub } = s as Record<string, unknown>;179 const runs24 = await db.execute<Record<string, unknown>>(sql`select count(*)::int as checks_24h, count(*) filter (where outcome = 'not_modified')::int as not_modified_24h, count(*) filter (where outcome in ('error','parse_error','rate_limited'))::int as errors_24h, avg(duration_ms)::int as avg_ms_24h from sensor_runs where sensor_id = ${String(s.id)} and started_at >= now() - interval '24 hours'`);180 return { sensor: { ...pub, ...runs24.rows[0] }, runs, snapshots: snaps, changes, events };181 });182183 // ---- Entities ----------------------------------------------------------------------------184 app.get<{ Querystring: { type?: string; q?: string; limit?: string; category?: string } }>("/api/v1/entities", async (req) => {185 const rows = await db.execute<Record<string, unknown>>(sql`186 select en.*, (select count(*) from events e join event_entities ee on ee.event_id = e.id where ee.entity_id = en.id and e.detected_at >= now() - interval '24 hours')::int as events_24h187 from entities en where true ${req.query.type ? sql`and en.type = ${req.query.type}` : sql``} ${req.query.category ? sql`and ${req.query.category} = any(en.categories)` : sql``} ${req.query.q ? sql`and (en.search @@ plainto_tsquery('simple', ${req.query.q}) or en.name ilike ${"%" + req.query.q + "%"})` : sql``}188 order by en.event_count desc, en.importance desc, en.name limit ${Math.min(1000, Number(req.query.limit ?? 200))}`);189 return { items: rows.rows };190 });191 app.get<{ Querystring: { limit?: string } }>("/api/v1/entities/rank", async (req) => ({ items: await entityRankings(Math.min(500, Number(req.query.limit ?? 100))) }));192 app.get<{ Params: { id: string } }>("/api/v1/entities/:id", async (req, reply) => {193 const e = (await db.execute<Record<string, unknown>>(sql`select * from entities where id = ${req.params.id} or id = ${"org_" + req.params.id} limit 1`)).rows[0];194 if (!e) return reply.status(404).send({ error: "not_found" });195 const id = String(e.id);196 const [children, relations, sources, aliases, recent, byType, insights, silent, parent] = await Promise.all([197 db.execute<Record<string, unknown>>(sql`select id, name, type, importance, event_count, last_event_at from entities where parent_id = ${id} order by event_count desc, name`).then((r) => r.rows),198 db.execute<Record<string, unknown>>(sql`select r.relation, r.from_id, r.to_id, f.name as from_name, t.name as to_name, t.type as to_type from entity_relations r join entities f on f.id = r.from_id join entities t on t.id = r.to_id where r.from_id = ${id} or r.to_id = ${id} limit 100`).then((r) => r.rows),199 db.execute<Record<string, unknown>>(sql`select s.id, s.name, s.domain, s.tier, s.first_party, s.country, (select count(*) from sensors x where x.source_id = s.id and x.enabled)::int as sensor_count, (select count(*) from events ev where ev.source_id = s.id and ev.detected_at >= now() - interval '24 hours')::int as events_24h from source_entities se join sources s on s.id = se.source_id where se.entity_id = ${id} and s.kind = 'registry'`).then((r) => r.rows),200 db.execute<{ alias: string }>(sql`select alias from entity_aliases where entity_id = ${id} order by alias`).then((r) => r.rows.map((x) => x.alias)),201 listEvents({ entity: id, limit: 30 }),202 db.execute<Record<string, unknown>>(sql`select e.event_type, count(*)::int as n from events e join event_entities ee on ee.event_id = e.id where ee.entity_id = ${id} group by 1 order by 2 desc`).then((r) => r.rows),203 entityInsights(id),204 listEvents({ entity: id, silent_change: true, limit: 10 }).then((r) => r.items),205 e.parent_id ? db.execute<Record<string, unknown>>(sql`select id, name, type from entities where id = ${String(e.parent_id)}`).then((r) => r.rows[0] ?? null) : null,206 ]);207 // Related entities: co-occurring in events over 30 days208 const related = await db.execute<Record<string, unknown>>(sql`209 select en.id, en.name, en.type, count(*)::int as shared_events from event_entities a join event_entities b on a.event_id = b.event_id and b.entity_id <> a.entity_id join events e on e.id = a.event_id join entities en on en.id = b.entity_id210 where a.entity_id = ${id} and e.detected_at >= now() - interval '30 days' group by en.id, en.name, en.type order by shared_events desc limit 12`).then((r) => r.rows);211 return { entity: e, parent, children, relations, sources, aliases, recent: recent.items, nextCursor: recent.nextCursor, by_type: byType, insights, silent, related };212 });213 app.get<{ Params: { id: string }; Querystring: { limit?: string; cursor?: string; after?: string; before?: string; event_type?: string; silent_change?: string } }>("/api/v1/entities/:id/timeline", async (req) => {214 const id = req.params.id.startsWith("org_") || req.params.id.startsWith("prd_") ? req.params.id : `org_${req.params.id}`;215 return listEvents({ entity: id, limit: Math.min(200, Number(req.query.limit ?? 100)), cursor: req.query.cursor, after: req.query.after, before: req.query.before, event_type: req.query.event_type, silent_change: req.query.silent_change === "true" ? true : undefined });216 });217218 // ---- Clusters ------------------------------------------------------------------------------219 app.get<{ Querystring: { limit?: string; since?: string; state?: string; min_events?: string } }>("/api/v1/clusters", async (req) => {220 const rows = await db.execute<Record<string, unknown>>(sql`221 select c.*, (select json_agg(json_build_object('id', e.id, 'slug', e.slug, 'title', e.title, 'importance', e.importance, 'event_type', e.event_type, 'detected_at', e.detected_at, 'source_id', e.source_id, 'url', e.url, 'first_party', e.first_party) order by e.importance desc) from events e where e.cluster_id = c.id) as events,222 (select json_build_object('id', s.id, 'name', s.name, 'domain', s.domain) from events e join sources s on s.id = e.source_id where e.id = c.primary_event_id) as source223 from event_clusters c where c.last_at >= now() - make_interval(hours => ${Math.min(720, Number(req.query.since ?? 72))}) ${req.query.state ? sql`and c.state = ${req.query.state}` : sql``} ${req.query.min_events ? sql`and c.event_count >= ${Number(req.query.min_events)}` : sql``} order by c.last_at desc limit ${Math.min(200, Number(req.query.limit ?? 50))}`);224 return { items: rows.rows };225 });226 app.get<{ Params: { id: string } }>("/api/v1/clusters/:id", async (req, reply) => {227 const d = await clusterDetail(req.params.id);228 if (!d) return reply.status(404).send({ error: "not_found" });229 return d;230 });231232 // ---- Domains / URLs ----------------------------------------------------------------------233 app.get<{ Params: { domain: string }; Querystring: { limit?: string; cursor?: string } }>("/api/v1/domains/:domain/timeline", async (req) => {234 const d = req.params.domain.toLowerCase();235 const src = (await db.execute<{ id: string }>(sql`select id from sources where (domain = ${d} or domain = ${"www." + d} or ${d} = 'www.' || domain) and kind = 'registry' limit 1`)).rows[0];236 const urls = await db.execute<Record<string, unknown>>(sql`select url, status, first_seen_at, last_seen_at, snapshot_count, change_count, sensor_id from urls where domain = ${d} or domain like ${"%." + d} order by change_count desc, last_seen_at desc limit 200`);237 const ev = src ? await listEvents({ source: src.id, limit: Math.min(200, Number(req.query.limit ?? 100)), cursor: req.query.cursor }) : { items: [], nextCursor: null };238 return { domain: d, source_id: src?.id ?? null, urls: urls.rows, events: ev.items, nextCursor: ev.nextCursor };239 });240 app.get<{ Querystring: { url: string } }>("/api/v1/urls/history", async (req, reply) => {241 if (!req.query.url) return reply.status(400).send({ error: "url required" });242 const u = (await db.execute<Record<string, unknown>>(sql`select * from urls where url = ${req.query.url}`)).rows[0];243 const history = await db.execute<Record<string, unknown>>(sql`244 select h.id, h.at, h.kind, h.snapshot_id, h.change_id, h.event_id, h.note, e.title as event_title, e.importance, e.event_type, e.slug as event_slug, c.kind as change_kind, c.signal, c.change_class245 from url_history h left join events e on e.id = h.event_id left join changes c on c.id = h.change_id where h.url = ${req.query.url} order by h.at desc limit 300`);246 const snaps = await db.execute<Record<string, unknown>>(sql`select id, captured_at, http_status, content_length, canonical_hash, title, storage_key is not null as has_raw from snapshots where url = ${req.query.url} or sensor_id = (select sensor_id from urls where url = ${req.query.url}) order by captured_at desc limit 200`);247 return { url: u ?? { url: req.query.url }, history: history.rows, snapshots: snaps.rows };248 });249250 // ---- Intelligence desks ----------------------------------------------------------------------251 app.get("/api/v1/stats", async () => cached("stats", 5_000, stats));252 // ---- Global Observation Coverage Score ------------------------------------------------------253 app.get("/api/v1/coverage", async () => coverageSummary());254 app.get<{ Params: { sector: string } }>("/api/v1/coverage/:sector", async (req, reply) => {255 const r = await coverageSector(req.params.sector.toLowerCase());256 if (!r) return reply.status(404).send({ error: "not_found" });257 return r;258 });259 app.get<{ Querystring: { hours?: string; limit?: string } }>("/api/v1/trending", async (req) => {260 const hours = Math.min(168, Number(req.query.hours ?? 24));261 const limit = Math.min(50, Number(req.query.limit ?? 12));262 return { items: await cached(`trending:${hours}:${limit}`, 15_000, () => trending(hours, limit)) };263 });264 app.get("/api/v1/breaking", async () => breakingDesk());265 app.get("/api/v1/pulse", async () => pulse());266 app.get("/api/v1/radar", async () => radar());267 app.get("/api/v1/countries", async () => ({ items: await countryList(), known: COUNTRIES }));268 app.get<{ Params: { code: string } }>("/api/v1/countries/:code", async (req, reply) => {269 const c = countryBySlug(req.params.code) ?? (COUNTRIES[req.params.code.toUpperCase()] ? { code: req.params.code.toUpperCase() } : null);270 if (!c) return reply.status(404).send({ error: "not_found" });271 return cached(`country:${c.code}`, 20_000, () => countryDesk(c.code));272 });273 app.get<{ Params: { channel: string } }>("/api/v1/categories/:channel", async (req, reply) => {274 const ch = req.params.channel.toLowerCase();275 if (!FEED_CHANNELS[ch] && !/^[a-z-]{2,30}$/.test(ch)) return reply.status(404).send({ error: "not_found" });276 return cached(`category:${ch}`, 15_000, () => categoryDesk(ch));277 });278 app.get("/api/v1/explore", async () =>279 cached("explore", 20_000, async () => {280 const since48 = new Date(Date.now() - 48 * 3600e3).toISOString();281 const [mostActive, biggest, silent, clusters, unusual, byType, byCategory, newly, firstParty] = await Promise.all([282 db.execute<Record<string, unknown>>(sql`select s.id, s.name, s.domain, s.first_party, count(*)::int as events_24h, max(e.importance)::float as max_importance from events e join sources s on s.id = e.source_id where e.detected_at >= now() - interval '24 hours' and s.kind = 'registry' group by s.id, s.name, s.domain, s.first_party order by events_24h desc limit 12`).then((r) => r.rows),283 listEvents({ limit: 12, order: "signal", after: since48 }).then((r) => r.items),284 listEvents({ limit: 12, silent_change: true, order: "signal", after: since48 }).then((r) => r.items),285 db.execute<Record<string, unknown>>(sql`select c.*, (select json_build_object('id', s.id, 'name', s.name, 'domain', s.domain) from events e join sources s on s.id = e.source_id where e.id = c.primary_event_id) as source, (select e.slug from events e where e.id = c.primary_event_id) as primary_slug from event_clusters c where c.event_count >= 2 and c.last_at >= now() - interval '48 hours' order by c.max_importance desc, c.event_count desc limit 12`).then((r) => r.rows),286 db.execute<Record<string, unknown>>(sql`287 with cur as (select s.source_id, count(*) as n from changes c join sensors s on s.id = c.sensor_id where c.detected_at >= now() - interval '2 hours' group by s.source_id),288 base as (select s.source_id, count(*)::float / (14*24) as per_hour from changes c join sensors s on s.id = c.sensor_id where c.detected_at >= now() - interval '14 days' group by s.source_id)289 select so.id, so.name, so.domain, cur.n::int as changes_2h, round(coalesce(base.per_hour,0)::numeric*24,1)::float as baseline_per_day,290 round((case when coalesce(base.per_hour,0) = 0 then (case when cur.n/2.0 > 2 then 70 else 40 end) else least(100, case when cur.n/2.0/base.per_hour <= 1 then cur.n/2.0/base.per_hour*30 else 30 + 25*(ln(cur.n/2.0/base.per_hour)/ln(2)) end) end)::numeric, 1)::float as activity_score,291 (case when coalesce(base.per_hour,0) > 0 then round(((cur.n/2.0/base.per_hour - 1) * 100)::numeric) else null end)::int as pct_vs_baseline292 from cur join sources so on so.id = cur.source_id left join base on base.source_id = cur.source_id where so.kind = 'registry' order by activity_score desc limit 12`).then((r) => r.rows),293 db.execute<Record<string, unknown>>(sql`select event_type, count(*)::int as n from events where detected_at >= now() - interval '7 days' group by 1 order by 2 desc`).then((r) => r.rows),294 db.execute<Record<string, unknown>>(sql`select c as category, count(*)::int as n from events e, unnest(e.categories) c where e.detected_at >= now() - interval '7 days' group by 1 order by 2 desc`).then((r) => r.rows),295 listEvents({ limit: 12, order: "recent", confidence_min: 60 }).then((r) => r.items),296 listEvents({ limit: 12, order: "signal", first_party: true, confirmed: true, after: since48 }).then((r) => r.items),297 ]);298 return { most_active_sources: mostActive, biggest_changes: biggest, silent_changes: silent, clusters, unusual_activity: unusual, newly_detected: newly, confirmed_first_party: firstParty, by_type: byType, by_category: byCategory, channels: FEED_CHANNELS, groups: Object.fromEntries(Object.entries(EVENT_GROUPS).map(([k, v]) => [k, v.label])), event_types: Object.fromEntries(Object.entries(EVENT_TYPES).map(([k, v]) => [k, v.label])) };299 }),300 );301302 // ---- Search ----------------------------------------------------------------------------303 app.get<{ Querystring: { q?: string; limit?: string } }>("/api/v1/search", async (req) => {304 const q = (req.query.q ?? "").trim();305 const parsed = parseSearch(q);306 if (q.length < 2) return { query: q, parsed, events: [], entities: [], sources: [], urls: [], clusters: [] };307 const lim = Math.min(50, Number(req.query.limit ?? 20));308 const free = parsed.text;309 const [ev, ents, srcs, urls, clusters] = await Promise.all([310 listEvents({ q, limit: lim, order: Object.keys(parsed.filters).length && !free ? "recent" : "recent" }).then((r) => r.items),311 free.length >= 2 ? db.execute<Record<string, unknown>>(sql`select id, name, type, domain, importance, event_count from entities where search @@ plainto_tsquery('simple', ${free}) or name ilike ${"%" + free + "%"} order by event_count desc, importance desc limit ${lim}`).then((r) => r.rows) : [],312 free.length >= 2 ? db.execute<Record<string, unknown>>(sql`select id, name, domain, tier, categories, first_party, country from sources where kind = 'registry' and (name ilike ${"%" + free + "%"} or domain ilike ${"%" + free + "%"}) limit ${lim}`).then((r) => r.rows) : [],313 free.length >= 4 ? db.execute<Record<string, unknown>>(sql`select url, domain, status, change_count, last_seen_at from urls where url ilike ${"%" + free + "%"} order by change_count desc limit ${lim}`).then((r) => r.rows) : [],314 free.length >= 2 ? db.execute<Record<string, unknown>>(sql`select id, slug, title, state, event_count, source_count, max_importance, last_at from event_clusters where title ilike ${"%" + free + "%"} and last_at >= now() - interval '30 days' order by max_importance desc limit 10`).then((r) => r.rows) : [],315 ]);316 return { query: q, parsed, events: ev, entities: ents, sources: srcs, urls, clusters };317 });318319 // ---- Connector health / system --------------------------------------------------------------320 app.get("/api/v1/health/connectors", async () =>321 cached("health", 10_000, async () => {322 const [connectors, sensorsByHealth, sensorsByStatus, worst, noisy, daily, byConnectorSensors, throughput, es, topFailingDomains, slowest] = await Promise.all([323 db.execute<Record<string, unknown>>(sql`select * from connector_health order by connector`).then((r) => r.rows),324 db.execute<Record<string, unknown>>(sql`select health, count(*)::int as n from sensors where enabled group by health`).then((r) => r.rows),325 db.execute<Record<string, unknown>>(sql`select status, count(*)::int as n from sensors group by status`).then((r) => r.rows),326 db.execute<Record<string, unknown>>(sql`select s.id, s.name, s.url, s.source_id, s.connector, s.health, s.consecutive_errors, s.last_error, s.last_status, s.last_check_at from sensors s where s.enabled and s.health <> 'UP' order by s.consecutive_errors desc, s.last_check_at desc limit 50`).then((r) => r.rows),327 db.execute<Record<string, unknown>>(sql`select s.id, s.name, s.source_id, s.url, s.raw_changes, s.meaningful_changes, case when s.raw_changes > 0 then round(1 - s.meaningful_changes::numeric / s.raw_changes, 3) else null end as noise_ratio from sensors s where s.raw_changes >= 5 order by noise_ratio desc nulls last, raw_changes desc limit 25`).then((r) => r.rows),328 db.execute<Record<string, unknown>>(sql`select * from metrics_daily order by day desc limit 30`).then((r) => r.rows),329 db.execute<Record<string, unknown>>(sql`select connector, count(*)::int as sensors, count(*) filter (where health = 'UP')::int as up, avg(avg_latency_ms)::int as avg_latency_ms from sensors where enabled group by connector order by sensors desc`).then((r) => r.rows),330 db.execute<Record<string, unknown>>(sql`select (select count(*) from sensor_runs where started_at >= now() - interval '5 minutes')::int as checks_5m, (select count(*) from sensor_runs where started_at >= now() - interval '5 minutes' and outcome = 'not_modified')::int as not_modified_5m, (select count(*) from events where detected_at >= now() - interval '5 minutes')::int as events_5m, (select count(*) from changes where detected_at >= now() - interval '5 minutes')::int as changes_5m, (select count(*) from sensors where enabled and next_check_at <= now())::int as queue_due, (select avg(duration_ms)::int from sensor_runs where started_at >= now() - interval '5 minutes')::int as avg_latency_5m_ms, (select count(*) from changes where detected_at >= now() - interval '24 hours' and change_class in ('cosmetic','navigation','timestamp','advertisement','boilerplate'))::int as noise_filtered_24h`).then((r) => r.rows[0]),331 engineStatus(),332 db.execute<Record<string, unknown>>(sql`select split_part(split_part(s.url, '/', 3), ':', 1) as host, count(*)::int as failures, max(r.error) as last_error from sensor_runs r join sensors s on s.id = r.sensor_id where r.started_at >= now() - interval '24 hours' and r.outcome in ('error','parse_error','rate_limited') group by 1 order by failures desc limit 15`).then((r) => r.rows),333 db.execute<Record<string, unknown>>(sql`select s.id, s.name, s.source_id, s.avg_latency_ms, s.connector from sensors s where s.enabled and s.avg_latency_ms is not null order by s.avg_latency_ms desc limit 15`).then((r) => r.rows),334 ]);335 const t = throughput ?? {};336 return { connectors, sensors_by_health: sensorsByHealth, sensors_by_status: sensorsByStatus, degraded_sensors: worst, noisy_sensors: noisy, daily, sensors_by_connector: byConnectorSensors, throughput: { ...t, checks_per_min: Math.round(Number(t.checks_5m ?? 0) / 5), events_per_min: Math.round((Number(t.events_5m ?? 0) / 5) * 10) / 10, not_modified_ratio: Number(t.checks_5m) ? Math.round((Number(t.not_modified_5m) / Number(t.checks_5m)) * 100) / 100 : null }, engine: es, top_failing_domains: topFailingDomains, slowest_sensors: slowest, live: liveStats() };337 }),338 );339340 // ---- Machine-readable feeds ------------------------------------------------------------------341 app.get("/api/v1/feed.rss", async (req, reply) => {342 const q = eventsQuery.parse({ ...(req.query as Record<string, unknown>), limit: 50 });343 const { items } = await listEvents(q);344 const esc = (s: unknown): string => String(s ?? "").replace(/&/g, "&").replace(/</g, "<").replace(/>/g, ">").replace(/"/g, """);345 const base = config.publicBaseUrl;346 const xml = `<?xml version="1.0" encoding="UTF-8"?>\n<rss version="2.0" xmlns:atom="http://www.w3.org/2005/Atom"><channel><title>WebSensor — ${esc(q.category ? q.category + " events" : q.silent_change ? "silent changes" : "live events")}</title><link>${base}</link><description>Meaningful changes detected on the public Web by WebSensor.</description><atom:link href="${base}/api/v1/feed.rss" rel="self" type="application/rss+xml"/>${items347 .map((e) => `<item><title>${esc(e.title)}</title><link>${base}/event/${esc(e.slug)}</link><guid isPermaLink="false">${esc(e.id)}</guid><pubDate>${new Date(String(e.detected_at)).toUTCString()}</pubDate><category>${esc(e.event_type)}</category><description>${esc(e.summary)} (signal ${esc(e.signal_score ?? e.importance)}, importance ${esc(e.importance)}, confidence ${esc(e.confidence)}${e.silent_change ? ", silent change" : ""}${e.first_party ? ", first-party" : ""}) — source: ${esc(e.url)}</description></item>`)348 .join("")}</channel></rss>`;349 reply.header("content-type", "application/rss+xml; charset=utf-8");350 return reply.send(xml);351 });352353 app.get("/api/v1", async () => ({354 name: "WebSensor API",355 version: "v1",356 release: config.version,357 docs: `${config.publicBaseUrl}/api`,358 rate_limit: "600 requests / minute / IP (headers x-ratelimit-limit, x-ratelimit-remaining, x-ratelimit-reset)",359 endpoints: [360 "/api/v1/events",361 "/api/v1/events/count",362 "/api/v1/events/{id|slug}",363 "/api/v1/changes/{id}",364 "/api/v1/snapshots/{id}",365 "/api/v1/snapshots/compare?a=&b=",366 "/api/v1/sensors/{id}",367 "/api/v1/sensors/{id}/snapshots",368 "/api/v1/sources",369 "/api/v1/sources/{id}",370 "/api/v1/entities",371 "/api/v1/entities/rank",372 "/api/v1/entities/{id}",373 "/api/v1/entities/{id}/timeline",374 "/api/v1/clusters",375 "/api/v1/clusters/{id|slug}",376 "/api/v1/domains/{domain}/timeline",377 "/api/v1/urls/history?url=",378 "/api/v1/search?q=",379 "/api/v1/stats",380 "/api/v1/trending",381 "/api/v1/breaking",382 "/api/v1/pulse",383 "/api/v1/radar",384 "/api/v1/explore",385 "/api/v1/countries",386 "/api/v1/countries/{code|slug}",387 "/api/v1/categories/{channel}",388 "/api/v1/health/connectors",389 "/api/v1/watchlists",390 "/api/v1/alerts",391 "/api/v1/notifications",392 "/api/v1/bookmarks",393 "/api/v1/views",394 "/api/v1/monitors",395 "/api/v1/feed.rss",396 "wss://…/api/v1/live",397 ],398 }));399}400401function stripEmpty(o: Record<string, unknown>): Record<string, unknown> {402 return Object.fromEntries(Object.entries(o).filter(([, v]) => v !== undefined && v !== null && v !== ""));403}404405function renderCanonical(c: { mode: string; text?: string; items?: Record<string, unknown>[]; json?: unknown }): string {406 if (c.mode === "text") return c.text ?? "";407 if (c.mode === "list") return (c.items ?? []).map((i) => [i.title ?? i.url ?? i.key, i.url && i.title ? i.url : null, i.summary ? String(i.summary).slice(0, 300) : null].filter(Boolean).join(" — ")).join("\n");408 return JSON.stringify(c.json ?? null, null, 2);409}410411export { textArray };412