TypeScript 55.4%
Python 43.2%
SQL 1.2%
1import { breakingState, jaccard, newId, shingles, slugify, velocityScore } from "@websensor/core";2import { db, eventClusters, events, gte, sql, textArray, type ClusterTimelineStep } from "@websensor/db";34/**5 * Novelty + clustering over a rolling in-memory window of recent events (loaded from6 * Postgres at startup). Similarity = Jaccard over word 3-shingles of title+summary.7 *8 * 2026-09-11: clusters now track propagation (spec §24, §35, §36): first-party vs external9 * signals, a timeline of every signal, velocity, lead time (first-party detection → first10 * external report) and a breaking state (spec §34).11 */12interface RecentEvent {13 id: string;14 clusterId: string | null;15 sourceId: string;16 sensorId: string;17 eventType: string;18 entityIds: string[];19 detectedAt: number;20 sh: Set<string>;21 importance: number;22 firstParty: boolean;23}2425const WINDOW_MS = 72 * 3600e3;26const CLUSTER_WINDOW_MS = 6 * 3600e3;27let recent: RecentEvent[] = [];28let loaded = false;2930export async function loadRecent(): Promise<void> {31 const since = new Date(Date.now() - WINDOW_MS);32 const rows = await db.execute<{ id: string; cluster_id: string | null; source_id: string; sensor_id: string; event_type: string; title: string; summary: string; detected_at: Date; importance: number; first_party: boolean | null; entity_ids: string[] | null }>(sql`33 select e.id, e.cluster_id, e.source_id, e.sensor_id, e.event_type, e.title, e.summary, e.detected_at, e.importance, e.first_party,34 (select array_agg(entity_id) from event_entities ee where ee.event_id = e.id) as entity_ids35 from events e where e.detected_at >= ${since} order by e.detected_at desc limit 4000`);36 recent = rows.rows.map((r) => ({ id: r.id, clusterId: r.cluster_id, sourceId: r.source_id, sensorId: r.sensor_id, eventType: r.event_type, entityIds: r.entity_ids ?? [], detectedAt: new Date(r.detected_at).getTime(), sh: shingles(`${r.title}\n${r.summary}`), importance: r.importance, firstParty: r.first_party ?? true }));37 loaded = true;38}3940function prune(): void {41 const cutoff = Date.now() - WINDOW_MS;42 recent = recent.filter((r) => r.detectedAt >= cutoff);43}4445export interface NoveltyResult {46 novelty: number;47 nearest: { id: string; similarity: number } | null;48 /** number of distinct OTHER sources reporting near-identical content */49 confirmations: number;50 /** among those, how many are first-party channels */51 firstPartyConfirmations: number;52}5354export async function assessNovelty(text: string, sourceId: string): Promise<NoveltyResult> {55 if (!loaded) await loadRecent();56 prune();57 const sh = shingles(text);58 let best = 0;59 let nearest: RecentEvent | null = null;60 const confirming = new Map<string, boolean>();61 for (const r of recent) {62 const s = jaccard(sh, r.sh);63 if (s > best) {64 best = s;65 nearest = r;66 }67 if (s >= 0.45 && r.sourceId !== sourceId) confirming.set(r.sourceId, r.firstParty);68 }69 return { novelty: Math.round((1 - best) * 100), nearest: nearest ? { id: nearest.id, similarity: Math.round(best * 100) / 100 } : null, confirmations: confirming.size, firstPartyConfirmations: [...confirming.values()].filter(Boolean).length };70}7172export interface ClusterDecision {73 clusterId: string;74 slug: string;75 created: boolean;76 /** stats after this event was attached */77 eventCount: number;78 sourceCount: number;79 firstPartyCount: number;80 externalCount: number;81 velocity: number;82 state: string;83 leadTimeMs: number | null;84}8586/**87 * Attach to an existing open cluster when the event shares an entity (or the same source)88 * with a recent event and is textually related, or when it is the same event type on the89 * same source within 30 minutes (e.g. one launch touching six pages). Otherwise open one.90 */91export async function clusterEvent(ev: { id: string; sourceId: string; sourceName: string; sensorId: string; sensorType: string; eventType: string; entityIds: string[]; detectedAt: Date; title: string; summary: string; importance: number; signal: number; categories: string[]; firstParty: boolean; sourceTier?: string }): Promise<ClusterDecision> {92 if (!loaded) await loadRecent();93 const sh = shingles(`${ev.title}\n${ev.summary}`);94 const now = ev.detectedAt.getTime();95 let bestCluster: string | null = null;96 let bestScore = 0;97 for (const r of recent) {98 if (!r.clusterId || now - r.detectedAt > CLUSTER_WINDOW_MS) continue;99 const sharedEntity = r.entityIds.some((e) => ev.entityIds.includes(e));100 const sameSource = r.sourceId === ev.sourceId;101 if (!sharedEntity && !sameSource) continue;102 const sim = jaccard(sh, r.sh);103 const closeInTime = now - r.detectedAt < 30 * 60e3;104 let score = 0;105 if (sim >= 0.22) score = sim + (sharedEntity ? 0.2 : 0);106 else if (sameSource && r.eventType === ev.eventType && closeInTime && r.sensorId !== ev.sensorId) score = 0.3;107 else if (sameSource && closeInTime && sim >= 0.12) score = 0.25;108 // cross-source, shared entity, moderately similar and same broad type → the same story reported elsewhere109 else if (!sameSource && sharedEntity && sim >= 0.15 && now - r.detectedAt < 2 * 3600e3) score = 0.2 + sim;110 if (score > bestScore) {111 bestScore = score;112 bestCluster = r.clusterId;113 }114 }115 const step: ClusterTimelineStep = { at: ev.detectedAt.toISOString(), eventId: ev.id, sourceId: ev.sourceId, sourceName: ev.sourceName, sensorType: ev.sensorType, firstParty: ev.firstParty, eventType: ev.eventType, importance: ev.importance };116 let clusterId: string;117 let slug: string;118 let created = false;119 let row: { event_count: number; source_count: number; first_party_count: number; external_count: number; first_at: Date; last_at: Date; first_party_at: Date | null; first_external_at: Date | null; lead_time_ms: number | null; slug: string | null; max_importance: number };120 if (bestCluster) {121 clusterId = bestCluster;122 const r = await db.execute<typeof row>(sql`123 update event_clusters set124 event_count = event_count + 1,125 last_at = greatest(last_at, ${ev.detectedAt}),126 max_importance = greatest(max_importance, ${ev.importance}),127 entity_ids = (select array(select distinct unnest(entity_ids || ${textArray(ev.entityIds)}))),128 categories = (select array(select distinct unnest(categories || ${textArray(ev.categories)}))),129 title = case when ${ev.importance} > max_importance then ${ev.title} else title end,130 summary = case when ${ev.importance} > max_importance then ${ev.summary} else summary end,131 primary_event_id = case when ${ev.importance} > max_importance then ${ev.id} else primary_event_id end,132 source_count = (select count(distinct source_id) from events where cluster_id = ${clusterId}) + (case when exists (select 1 from events where cluster_id = ${clusterId} and source_id = ${ev.sourceId}) then 0 else 1 end),133 first_party_count = first_party_count + ${ev.firstParty ? 1 : 0},134 external_count = external_count + ${ev.firstParty ? 0 : 1},135 first_party_at = case when ${ev.firstParty} then least(coalesce(first_party_at, ${ev.detectedAt}), ${ev.detectedAt}) else first_party_at end,136 first_external_at = case when ${!ev.firstParty} then least(coalesce(first_external_at, ${ev.detectedAt}), ${ev.detectedAt}) else first_external_at end,137 timeline = (case when jsonb_array_length(timeline) < 200 then timeline || ${JSON.stringify([step])}::jsonb else timeline end)138 where id = ${clusterId}139 returning event_count, source_count, first_party_count, external_count, first_at, last_at, first_party_at, first_external_at, lead_time_ms, slug, max_importance`);140 row = r.rows[0]!;141 slug = row.slug ?? clusterId;142 } else {143 clusterId = newId("clu");144 slug = `${slugify(ev.title).slice(0, 60)}-${clusterId.slice(-6)}`;145 created = true;146 await db.insert(eventClusters).values({ id: clusterId, slug, title: ev.title, summary: ev.summary, primaryEventId: ev.id, entityIds: ev.entityIds, categories: ev.categories, eventCount: 1, maxImportance: ev.importance, firstAt: ev.detectedAt, lastAt: ev.detectedAt, sourceCount: 1, firstPartyCount: ev.firstParty ? 1 : 0, externalCount: ev.firstParty ? 0 : 1, firstPartyAt: ev.firstParty ? ev.detectedAt : null, firstExternalAt: ev.firstParty ? null : ev.detectedAt, timeline: [step], state: "watching" });147 row = { event_count: 1, source_count: 1, first_party_count: ev.firstParty ? 1 : 0, external_count: ev.firstParty ? 0 : 1, first_at: ev.detectedAt, last_at: ev.detectedAt, first_party_at: ev.firstParty ? ev.detectedAt : null, first_external_at: ev.firstParty ? null : ev.detectedAt, lead_time_ms: null, slug, max_importance: ev.importance };148 }149 // Derived: velocity, lead time, state.150 const windowHours = Math.max(0.25, (new Date(row.last_at).getTime() - new Date(row.first_at).getTime()) / 3600e3);151 const velocity = velocityScore({ signals: row.event_count, windowHours: Math.min(windowHours, 6), uniqueSources: row.source_count, firstPartySignals: row.first_party_count });152 const leadTimeMs = row.first_party_at && row.first_external_at ? new Date(row.first_external_at).getTime() - new Date(row.first_party_at).getTime() : null;153 const confirmations = Math.max(0, row.source_count - 1);154 const state = breakingState({ signal: ev.signal, importance: Math.max(ev.importance, row.max_importance), velocity, confirmations, firstPartyCount: row.first_party_count, ageMinutes: (Date.now() - new Date(row.first_at).getTime()) / 60e3, sourceTier: ev.sourceTier });155 await db.execute(sql`update event_clusters set velocity = ${velocity}, state = ${state}, lead_time_ms = ${leadTimeMs} where id = ${clusterId}`);156 recent.unshift({ id: ev.id, clusterId, sourceId: ev.sourceId, sensorId: ev.sensorId, eventType: ev.eventType, entityIds: ev.entityIds, detectedAt: now, sh, importance: ev.importance, firstParty: ev.firstParty });157 if (recent.length > 6000) recent.length = 6000;158 return { clusterId, slug, created, eventCount: row.event_count, sourceCount: row.source_count, firstPartyCount: row.first_party_count, externalCount: row.external_count, velocity, state, leadTimeMs };159}160161/** Recent events of the same source published through announcement-type sensors (for silent-change detection). */162export function recentAnnouncementSimilarity(sourceId: string, text: string, sinceMs: number): number {163 const sh = shingles(text);164 let best = 0;165 const cutoff = Date.now() - sinceMs;166 for (const r of recent) {167 if (r.sourceId !== sourceId || r.detectedAt < cutoff) continue;168 if (!/announcement|product_launch|model_release|software_release|repository_release|incident|outage|maintenance|security_advisory|service_launch|patch_release/.test(r.eventType)) continue;169 best = Math.max(best, jaccard(sh, r.sh));170 }171 return best;172}173174export async function recentClusterCount(sinceMs: number): Promise<number> {175 const rows = await db.select({ n: sql<number>`count(*)` }).from(eventClusters).where(gte(eventClusters.lastAt, new Date(Date.now() - sinceMs)));176 return Number(rows[0]?.n ?? 0);177}178179/** Periodic: age out states (breaking → developing/confirmed → closed) for clusters that stopped receiving signals. */180export async function refreshClusterStates(): Promise<void> {181 await db.execute(sql`182 update event_clusters set state = case183 when last_at < now() - interval '72 hours' then 'closed'184 when state = 'breaking' and first_at < now() - interval '6 hours' then (case when source_count >= 3 then 'confirmed' else 'developing' end)185 when state = 'developing' and first_at < now() - interval '12 hours' then (case when source_count >= 3 then 'confirmed' else 'watching' end)186 when state = 'confirmed' and last_at < now() - interval '24 hours' then 'watching'187 else state end188 where state <> 'closed' and last_at >= now() - interval '10 days'`);189}190191export { events };192