TypeScript 55.4%
Python 43.2%
SQL 1.2%
1import { EVENT_GROUPS, parseSearch } from "@websensor/core";2import { db, sql, textArray } from "@websensor/db";34/** Read-model helpers shared by REST routes. All return plain JSON-ready objects. */56export interface EventFilters {7 after?: string;8 before?: string;9 category?: string;10 entity?: string;11 source?: string;12 domain?: string;13 sensor?: string;14 cluster?: string;15 importance_min?: number;16 confidence_min?: number;17 signal_min?: number;18 event_type?: string;19 group?: string;20 silent_change?: boolean;21 first_party?: boolean;22 confirmed?: boolean;23 country?: string;24 language?: string;25 change_class?: string;26 q?: string;27 limit: number;28 cursor?: string;29 order?: "recent" | "importance" | "signal";30 /** include events from custom (owner) sources — only set by owner-scoped routes */31 includeCustom?: boolean;32}3334export const EVENT_SELECT = sql`35 e.id, e.slug, e.event_type, e.title, e.summary, e.why_it_matters, e.importance, e.confidence, e.novelty, e.categories, e.keywords,36 e.silent_change, e.evidence_label, e.url, e.canonical_url, e.detected_at, e.published_at, e.observed_from, e.processed_at, e.published_to_feed_at,37 e.detection_latency_ms, e.processing_latency_ms, e.cluster_id, e.sensor_id, e.source_id, e.change_id, e.old_snapshot_id, e.new_snapshot_id,38 e.importance_components, e.processing_version,39 e.signal_score, e.velocity_score, e.impact_score, e.anomaly_score, e.change_class, e.first_party, e.country, e.language, e.field_changes, e.score_reasons,40 json_build_object('id', s.id, 'name', s.name, 'domain', s.domain, 'tier', s.tier, 'categories', s.categories, 'first_party', s.first_party, 'country', s.country) as source,41 json_build_object('id', sen.id, 'name', sen.name, 'type', sen.type, 'connector', sen.connector, 'tier', sen.tier) as sensor,42 coalesce((select json_agg(json_build_object('id', en.id, 'name', en.name, 'type', en.type, 'role', ee.role) order by ee.role, en.name)43 from event_entities ee join entities en on en.id = ee.entity_id where ee.event_id = e.id), '[]'::json) as entities,44 (select event_count from event_clusters c where c.id = e.cluster_id) as cluster_size,45 (select json_build_object('id', c.id, 'slug', c.slug, 'state', c.state, 'event_count', c.event_count, 'source_count', c.source_count, 'first_party_count', c.first_party_count, 'external_count', c.external_count, 'velocity', c.velocity, 'lead_time_ms', c.lead_time_ms) from event_clusters c where c.id = e.cluster_id) as cluster`;4647/** Build WHERE conditions from filters (shared by list, RSS, watchlist and desk queries). */48export function eventConditions(f: EventFilters): ReturnType<typeof sql>[] {49 const conds = [sql`true`];50 let free = f.q?.trim() ?? "";51 // Advanced syntax inside q (entity:… type:… after:…) is merged into the filters.52 if (free) {53 const p = parseSearch(free);54 free = p.text;55 const pf = p.filters;56 f = { ...f, entity: f.entity ?? pf.entity, source: f.source ?? pf.source, domain: f.domain ?? pf.domain, event_type: f.event_type ?? pf.event_type, group: f.group ?? pf.group, category: f.category ?? pf.category, country: f.country ?? pf.country, language: f.language ?? pf.language, after: f.after ?? pf.after, before: f.before ?? pf.before, silent_change: f.silent_change ?? pf.silent_change, first_party: f.first_party ?? pf.first_party, confirmed: f.confirmed ?? pf.confirmed, importance_min: f.importance_min ?? pf.importance_min, confidence_min: f.confidence_min ?? pf.confidence_min, signal_min: f.signal_min ?? pf.signal_min, change_class: f.change_class ?? pf.change_class, cluster: f.cluster ?? pf.cluster, sensor: f.sensor ?? pf.sensor };57 }58 if (!f.includeCustom) conds.push(sql`s.kind = 'registry'`);59 if (f.after) conds.push(sql`e.detected_at > ${new Date(f.after)}`);60 if (f.before) conds.push(sql`e.detected_at < ${new Date(f.before)}`);61 if (f.category) conds.push(sql`${f.category} = any(e.categories)`);62 if (f.source) conds.push(sql`e.source_id = ${f.source}`);63 if (f.sensor) conds.push(sql`e.sensor_id = ${f.sensor}`);64 if (f.cluster) conds.push(sql`e.cluster_id = ${f.cluster}`);65 if (f.domain) conds.push(sql`s.domain = ${f.domain}`);66 if (f.entity) conds.push(sql`exists (select 1 from event_entities x where x.event_id = e.id and x.entity_id = ${f.entity})`);67 if (f.importance_min !== undefined) conds.push(sql`e.importance >= ${f.importance_min}`);68 if (f.confidence_min !== undefined) conds.push(sql`e.confidence >= ${f.confidence_min}`);69 if (f.signal_min !== undefined) conds.push(sql`coalesce(e.signal_score, e.importance) >= ${f.signal_min}`);70 if (f.event_type) conds.push(sql`e.event_type = any(${textArray(f.event_type.split(",").map((t) => t.trim()).filter(Boolean))})`);71 if (f.group && EVENT_GROUPS[f.group]) conds.push(sql`e.event_type = any(${textArray(EVENT_GROUPS[f.group]!.types)})`);72 if (f.silent_change !== undefined) conds.push(sql`e.silent_change = ${f.silent_change}`);73 if (f.first_party !== undefined) conds.push(sql`e.first_party = ${f.first_party}`);74 if (f.confirmed) conds.push(sql`e.evidence_label = 'CONFIRMED'`);75 if (f.country) conds.push(sql`e.country = ${f.country.toUpperCase()}`);76 if (f.language) conds.push(sql`e.language = ${f.language.toLowerCase()}`);77 if (f.change_class) conds.push(sql`e.change_class = ${f.change_class}`);78 if (free) conds.push(sql`(e.search @@ websearch_to_tsquery('english', ${free}) or e.title ilike ${"%" + free + "%"})`);79 return conds;80}8182export async function listEvents(f: EventFilters): Promise<{ items: Record<string, unknown>[]; nextCursor: string | null }> {83 const conds = eventConditions(f);84 if (f.cursor) {85 const [ts, id] = decodeCursor(f.cursor);86 if (f.order === "importance") conds.push(sql`(e.importance, e.id) < (${Number(ts)}, ${id})`);87 else if (f.order === "signal") conds.push(sql`(coalesce(e.signal_score, e.importance), e.id) < (${Number(ts)}, ${id})`);88 else conds.push(sql`(e.detected_at, e.id) < (${new Date(Number(ts))}, ${id})`);89 }90 const order = f.order === "importance" ? sql`e.importance desc, e.id desc` : f.order === "signal" ? sql`coalesce(e.signal_score, e.importance) desc, e.id desc` : sql`e.detected_at desc, e.id desc`;91 const rows = await db.execute<Record<string, unknown>>(sql`select ${EVENT_SELECT} from events e join sources s on s.id = e.source_id join sensors sen on sen.id = e.sensor_id where ${sql.join(conds, sql` and `)} order by ${order} limit ${f.limit + 1}`);92 const items = rows.rows.slice(0, f.limit);93 const last = items[items.length - 1];94 const nextCursor = rows.rows.length > f.limit && last ? encodeCursor(f.order === "importance" ? String(last.importance) : f.order === "signal" ? String(last.signal_score ?? last.importance) : String(new Date(last.detected_at as string).getTime()), String(last.id)) : null;95 return { items, nextCursor };96}9798export async function countEvents(f: EventFilters): Promise<number> {99 const conds = eventConditions(f);100 const r = await db.execute<{ n: string }>(sql`select count(*)::text as n from events e join sources s on s.id = e.source_id where ${sql.join(conds, sql` and `)}`);101 return Number(r.rows[0]?.n ?? 0);102}103104export async function getEvent(idOrSlug: string): Promise<Record<string, unknown> | null> {105 const rows = await db.execute<Record<string, unknown>>(sql`select ${EVENT_SELECT}, e.interpretation from events e join sources s on s.id = e.source_id join sensors sen on sen.id = e.sensor_id where e.id = ${idOrSlug} or e.slug = ${idOrSlug} limit 1`);106 return rows.rows[0] ?? null;107}108109export async function relatedEvents(ev: Record<string, unknown>, limit = 8): Promise<Record<string, unknown>[]> {110 const rows = await db.execute<Record<string, unknown>>(sql`111 select ${EVENT_SELECT} from events e join sources s on s.id = e.source_id join sensors sen on sen.id = e.sensor_id112 where e.id <> ${String(ev.id)} and s.kind = 'registry' and (e.cluster_id = ${String(ev.cluster_id ?? "")} or e.source_id = ${String(ev.source_id)} or exists (select 1 from event_entities a join event_entities b on a.entity_id = b.entity_id where a.event_id = e.id and b.event_id = ${String(ev.id)}))113 order by (e.cluster_id = ${String(ev.cluster_id ?? "")}) desc, e.detected_at desc limit ${limit}`);114 return rows.rows;115}116117/** Previous meaningful events on the same sensor (historical context for "what changed" — spec §79–80). */118export async function sensorHistory(sensorId: string, beforeEventId: string, limit = 6): Promise<Record<string, unknown>[]> {119 const rows = await db.execute<Record<string, unknown>>(sql`120 select e.id, e.slug, e.title, e.event_type, e.importance, e.silent_change, e.detected_at, e.field_changes from events e121 where e.sensor_id = ${sensorId} and e.id <> ${beforeEventId} order by e.detected_at desc limit ${limit}`);122 return rows.rows;123}124125export async function stats(): Promise<Record<string, unknown>> {126 const [r] = (127 await db.execute<Record<string, unknown>>(sql`128 select129 (select count(*) from sources where enabled and kind = 'registry') as sources,130 (select count(*) from sensors where enabled and status <> 'SHADOW') as sensors,131 (select count(*) from sensors where enabled and status = 'SHADOW') as sensors_shadow,132 (select count(*) from sources where enabled and kind = 'registry' and origin = 'factory') as sources_factory,133 (select count(*) from entities) as entities,134 (select count(*) from snapshots) as snapshots,135 (select count(*) from events) as events_total,136 (select count(*) from events where detected_at >= now() - interval '24 hours') as events_24h,137 (select count(*) from events where detected_at >= now() - interval '1 hour') as events_1h,138 (select count(*) from events where detected_at >= now() - interval '24 hours' and silent_change) as silent_24h,139 (select count(*) from events where detected_at >= now() - interval '24 hours' and coalesce(signal_score, importance) >= 80) as breaking_24h,140 (select count(*) from event_clusters where state = 'breaking') as breaking_now,141 (select count(*) from event_clusters where state = 'developing') as developing_now,142 (select count(*) from changes where detected_at >= now() - interval '24 hours') as changes_24h,143 (select coalesce(sum(checks),0) from metrics_daily where day = (now() at time zone 'UTC')::date) as checks_today,144 (select coalesce(sum(not_modified),0) from metrics_daily where day = (now() at time zone 'UTC')::date) as not_modified_today,145 (select coalesce(sum(bytes),0) from metrics_daily where day = (now() at time zone 'UTC')::date) as bytes_today,146 (select count(*) from sensor_runs where started_at >= now() - interval '1 hour') as checks_last_hour,147 (select count(*) from sensor_runs where started_at >= now() - interval '5 minutes') as checks_last_5m,148 (select count(*) from sensor_runs where started_at >= now() - interval '5 minutes' and outcome = 'not_modified') as not_modified_last_5m,149 (select count(*) from sensors where health = 'UP' and enabled and status <> 'SHADOW') as sensors_up,150 (select count(*) from sensors where health in ('DEGRADED','ERROR','RATE_LIMITED') and enabled and status <> 'SHADOW') as sensors_degraded,151 (select count(*) from sources where enabled and first_party and kind = 'registry') as sources_first_party,152 (select count(distinct country) from sources where country is not null) as countries,153 (select max(started_at) from sensor_runs) as last_check_at,154 (select max(detected_at) from events) as last_event_at,155 (select percentile_cont(0.5) within group (order by processing_latency_ms) from events where detected_at >= now() - interval '24 hours') as p50_processing_ms,156 (select percentile_cont(0.5) within group (order by detection_latency_ms) from events where detected_at >= now() - interval '24 hours' and detection_latency_ms is not null and detection_latency_ms < 86400000) as p50_detection_ms`)157 ).rows;158 const out = Object.fromEntries(Object.entries(r ?? {}).map(([k, v]) => [k, typeof v === "string" && /^\d+(\.\d+)?$/.test(v) ? Number(v) : v]));159 out.checks_per_min = Math.round(Number(out.checks_last_5m ?? 0) / 5);160 out.events_per_min = Math.round((Number(out.events_1h ?? 0) / 60) * 10) / 10;161 out.not_modified_ratio_5m = Number(out.checks_last_5m) ? Math.round((Number(out.not_modified_last_5m) / Number(out.checks_last_5m)) * 100) / 100 : null;162 return out;163}164165export async function trending(hours = 24, limit = 12): Promise<Record<string, unknown>[]> {166 const rows = await db.execute<Record<string, unknown>>(sql`167 with cur as (168 select ee.entity_id, count(*) as n, sum(e.importance) as imp, count(distinct e.source_id) as sources, sum(case when e.silent_change then 1 else 0 end) as silent, max(e.importance) as max_imp,169 sum(case when e.first_party then 1 else 0 end) as first_party, sum(case when e.evidence_label = 'CONFIRMED' then 1 else 0 end) as confirmed, avg(coalesce(e.signal_score, e.importance)) as avg_signal,170 count(*) filter (where e.detected_at >= now() - make_interval(hours => ${hours}) / 4) as recent_quarter171 from events e join event_entities ee on ee.event_id = e.id join sources s on s.id = e.source_id where s.kind = 'registry' and e.detected_at >= now() - make_interval(hours => ${hours}) group by ee.entity_id),172 prev as (173 select ee.entity_id, count(*) as n from events e join event_entities ee on ee.event_id = e.id174 where e.detected_at >= now() - make_interval(hours => ${hours * 2}) and e.detected_at < now() - make_interval(hours => ${hours}) group by ee.entity_id),175 base as (176 select entity_id, avg(events)::float as per_day from entity_daily where day >= (now() at time zone 'UTC')::date - 30 and day < (now() at time zone 'UTC')::date group by entity_id)177 select en.id, en.name, en.type, en.domain, en.importance as entity_importance, cur.n::int as events, cur.imp::float as importance_sum, cur.sources::int as sources, cur.silent::int as silent, cur.max_imp::float as max_importance,178 cur.first_party::int as first_party, cur.confirmed::int as confirmed, round(cur.avg_signal::numeric, 1)::float as avg_signal, coalesce(prev.n,0)::int as prev_events, coalesce(base.per_day, 0)::float as baseline_per_day,179 cur.recent_quarter::int as recent_quarter,180 case when cur.recent_quarter::float / greatest(1, cur.n) > 0.5 then 'up' when cur.recent_quarter = 0 and cur.n >= 3 then 'down' else 'flat' end as direction,181 round((100 * (1 - exp(-(182 18*(ln(1+cur.n)/ln(2)) + 0.35*(cur.imp/greatest(1,cur.n)) + 10*(ln(1+cur.sources)/ln(2)) + 12*least(2, case when coalesce(prev.n,0)=0 then 2 else cur.n::float/prev.n end) + 5*least(3,cur.silent)183 + 6*least(3, cur.first_party) + 4*least(3, cur.confirmed) + (case when coalesce(base.per_day,0) > 0 then 8*least(3, cur.n / (base.per_day * ${hours} / 24.0)) else 0 end)184 ) / 110.0)))::numeric, 1)::float as score185 from cur join entities en on en.id = cur.entity_id left join prev on prev.entity_id = cur.entity_id left join base on base.entity_id = cur.entity_id186 order by score desc limit ${limit}`);187 return rows.rows;188}189190export async function sourceActivity(sourceId: string): Promise<Record<string, unknown>> {191 const [r] = (192 await db.execute<Record<string, unknown>>(sql`193 select194 (select count(*) from changes c join sensors s on s.id = c.sensor_id where s.source_id = ${sourceId} and c.detected_at >= now() - interval '2 hours')::int as changes_2h,195 (select count(*) from changes c join sensors s on s.id = c.sensor_id where s.source_id = ${sourceId} and c.detected_at >= now() - interval '14 days')::int as changes_14d,196 (select count(*) from events where source_id = ${sourceId} and detected_at >= now() - interval '24 hours')::int as events_24h,197 (select count(*) from events where source_id = ${sourceId} and detected_at >= now() - interval '14 days')::int as events_14d,198 (select count(*) from events where source_id = ${sourceId} and detected_at >= now() - interval '24 hours' and silent_change)::int as silent_24h,199 (select count(*) from events where source_id = ${sourceId} and detected_at >= now() - interval '24 hours' and coalesce(signal_score, importance) >= 80)::int as breaking_24h`)200 ).rows;201 const c2h = Number(r?.changes_2h ?? 0);202 const baselinePerHour = Number(r?.changes_14d ?? 0) / (14 * 24);203 const currentPerHour = c2h / 2;204 let anomaly = 0;205 if (baselinePerHour <= 0) anomaly = currentPerHour > 2 ? 70 : currentPerHour > 0 ? 40 : 0;206 else {207 const ratio = currentPerHour / baselinePerHour;208 anomaly = ratio <= 1 ? ratio * 30 : Math.min(100, 30 + 25 * Math.log2(ratio));209 }210 return { ...r, baseline_changes_per_day: Math.round(baselinePerHour * 24 * 10) / 10, activity_score: Math.round(anomaly * 10) / 10 };211}212213/** Source quality score (spec §48): success rate, latency, structured share, usefulness (events/raw), consistency. Distinct from importance. */214export async function sourceQuality(sourceId: string): Promise<Record<string, unknown>> {215 const [r] = (216 await db.execute<Record<string, unknown>>(sql`217 select218 (select count(*) from sensors where source_id = ${sourceId} and enabled)::int as sensors,219 (select count(*) from sensors where source_id = ${sourceId} and enabled and health = 'UP')::int as sensors_up,220 (select avg(avg_latency_ms) from sensors where source_id = ${sourceId} and enabled)::int as avg_latency_ms,221 (select coalesce(sum(raw_changes),0) from sensors where source_id = ${sourceId})::int as raw_changes,222 (select coalesce(sum(meaningful_changes),0) from sensors where source_id = ${sourceId})::int as meaningful_changes,223 (select coalesce(sum(total_runs),0) from sensors where source_id = ${sourceId})::int as total_runs,224 (select coalesce(sum(total_not_modified),0) from sensors where source_id = ${sourceId})::int as total_not_modified,225 (select count(*) from sensors where source_id = ${sourceId} and enabled and connector <> 'http')::int as structured_sensors,226 (select coalesce(sum(checks),0) from source_daily where source_id = ${sourceId} and day >= (now() at time zone 'UTC')::date - 7)::int as checks_7d,227 (select coalesce(sum(errors),0) from source_daily where source_id = ${sourceId} and day >= (now() at time zone 'UTC')::date - 7)::int as errors_7d,228 (select avg(confidence) from events where source_id = ${sourceId} and detected_at >= now() - interval '30 days')::float as avg_confidence`)229 ).rows;230 const sensors = Number(r?.sensors ?? 0);231 const success = Number(r?.checks_7d) ? 1 - Number(r?.errors_7d) / Number(r?.checks_7d) : sensors ? Number(r?.sensors_up) / sensors : 1;232 const latency = Number(r?.avg_latency_ms ?? 800);233 const latencyScore = latency <= 400 ? 1 : latency <= 1500 ? 0.8 : latency <= 4000 ? 0.6 : 0.4;234 const structured = sensors ? Number(r?.structured_sensors) / sensors : 0;235 const raw = Number(r?.raw_changes ?? 0);236 const usefulness = raw >= 5 ? Math.min(1, (Number(r?.meaningful_changes) / raw) * 2) : 0.6;237 const conf = Number(r?.avg_confidence ?? 70) / 100;238 const score = Math.round(100 * (0.35 * success + 0.15 * latencyScore + 0.15 * structured + 0.2 * usefulness + 0.15 * conf));239 return { ...r, success_rate: Math.round(success * 1000) / 1000, structured_share: Math.round(structured * 100) / 100, usefulness: Math.round(usefulness * 100) / 100, quality_score: score };240}241242export function encodeCursor(a: string, b: string): string {243 return Buffer.from(`${a}|${b}`).toString("base64url");244}245export function decodeCursor(c: string): [string, string] {246 const s = Buffer.from(c, "base64url").toString("utf8");247 const i = s.indexOf("|");248 return [s.slice(0, i), s.slice(i + 1)];249}250