TypeScript 55.4%
Python 43.2%
SQL 1.2%
1import { COUNTRIES, dailyAnomaly, entityRankScore, EVENT_GROUPS, FEED_CHANNELS } from "@websensor/core";2import { db, sql, textArray } from "@websensor/db";3import { cached } from "./cache";4import { EVENT_SELECT, listEvents } from "./queries";56/**7 * Intelligence read-models (spec §34–39, §101–105): breaking desk, pulse, radar, entity insights,8 * rankings, cluster propagation, country and category desks. Aggregates are cached briefly.9 */1011const CLUSTER_SELECT = sql`c.id, c.slug, c.title, c.summary, c.primary_event_id, c.entity_ids, c.categories, c.event_count, c.max_importance, c.first_at, c.last_at, c.source_count, c.first_party_count, c.external_count, c.velocity, c.state, c.lead_time_ms, c.first_party_at, c.first_external_at`;1213async function clusterPrimaryEvents(where: ReturnType<typeof sql>, order: ReturnType<typeof sql>, limit: number): Promise<Record<string, unknown>[]> {14 const rows = await db.execute<Record<string, unknown>>(sql`15 select ${CLUSTER_SELECT},16 (select row_to_json(x) from (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 e.id = c.primary_event_id) x) as event,17 (select json_agg(json_build_object('id', s2.id, 'name', s2.name, 'domain', s2.domain, 'first_party', s2.first_party) order by s2.name) from (select distinct s3.id, s3.name, s3.domain, s3.first_party from events e3 join sources s3 on s3.id = e3.source_id where e3.cluster_id = c.id limit 12) s2) as sources18 from event_clusters c where ${where} order by ${order} limit ${limit}`);19 return rows.rows;20}2122export async function breakingDesk(): Promise<Record<string, unknown>> {23 return cached("breaking", 8_000, async () => {24 const [breaking, developing, confirmed, watching] = await Promise.all([25 clusterPrimaryEvents(sql`c.state = 'breaking' and c.last_at >= now() - interval '24 hours'`, sql`c.max_importance desc, c.velocity desc, c.last_at desc`, 20),26 clusterPrimaryEvents(sql`c.state = 'developing' and c.last_at >= now() - interval '24 hours'`, sql`c.velocity desc, c.max_importance desc, c.last_at desc`, 20),27 clusterPrimaryEvents(sql`c.state = 'confirmed' and c.last_at >= now() - interval '48 hours'`, sql`c.last_at desc`, 20),28 listEvents({ limit: 25, order: "signal", signal_min: 60, after: new Date(Date.now() - 12 * 3600e3).toISOString() }).then((r) => r.items.filter((e) => !["breaking", "developing", "confirmed"].includes(String((e.cluster as { state?: string } | null)?.state ?? "")))),29 ]);30 return { breaking_now: breaking, developing, recently_confirmed: confirmed, watching, generated_at: new Date().toISOString() };31 });32}3334export async function pulse(): Promise<Record<string, unknown>> {35 return cached("pulse", 10_000, async () => {36 const desks = ["ai", "cyber", "finance", "government", "infrastructure", "health", "science", "products"];37 const [activity, byDesk, rising, anomalies, silent, incidents, breaking, groups, totals] = await Promise.all([38 db.execute<Record<string, unknown>>(sql`39 with b as (select generate_series(date_trunc('hour', now() at time zone 'UTC') - interval '6 hours', date_trunc('minute', now() at time zone 'UTC'), interval '15 minutes') as bucket),40 ev as (select to_timestamp(floor(extract(epoch from detected_at) / 900) * 900) at time zone 'UTC' as bk, count(*) as n from events where detected_at >= now() - interval '7 hours' group by 1),41 ch as (select to_timestamp(floor(extract(epoch from detected_at) / 900) * 900) at time zone 'UTC' as bk, count(*) as n from changes where detected_at >= now() - interval '7 hours' group by 1)42 select to_char(b.bucket, 'YYYY-MM-DD"T"HH24:MI:00"Z"') as t, coalesce(ev.n, 0)::int as events, coalesce(ch.n, 0)::int as changes43 from b left join ev on ev.bk = b.bucket left join ch on ch.bk = b.bucket order by b.bucket`).then((r) => r.rows).catch(() => []),44 Promise.all(desks.map(async (d) => ({ desk: d, items: (await listEvents({ limit: 5, order: "signal", category: d, after: new Date(Date.now() - 12 * 3600e3).toISOString() })).items }))),45 db.execute<Record<string, unknown>>(sql`46 with cur as (select ee.entity_id, count(*) n from events e join event_entities ee on ee.event_id = e.id where e.detected_at >= now() - interval '3 hours' group by 1),47 prev as (select ee.entity_id, count(*) n from events e join event_entities ee on ee.event_id = e.id where e.detected_at >= now() - interval '27 hours' and e.detected_at < now() - interval '3 hours' group by 1)48 select en.id, en.name, en.type, cur.n::int as events_3h, coalesce(prev.n,0)::int as events_prev_24h, round((cur.n::float / greatest(0.125, coalesce(prev.n,0)/8.0))::numeric, 1)::float as acceleration49 from cur join entities en on en.id = cur.entity_id left join prev on prev.entity_id = cur.entity_id where cur.n >= 2 order by acceleration desc, cur.n desc limit 10`).then((r) => r.rows),50 db.execute<Record<string, unknown>>(sql`51 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),52 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)53 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,54 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,55 (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_baseline56 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 8`).then((r) => r.rows),57 listEvents({ limit: 8, silent_change: true, order: "signal", after: new Date(Date.now() - 24 * 3600e3).toISOString() }).then((r) => r.items),58 listEvents({ limit: 8, group: "reliability", order: "recent", after: new Date(Date.now() - 6 * 3600e3).toISOString() }).then((r) => r.items),59 clusterPrimaryEvents(sql`c.state in ('breaking','developing') and c.last_at >= now() - interval '24 hours'`, sql`(c.state = 'breaking') desc, c.max_importance desc, c.velocity desc`, 8),60 db.execute<Record<string, unknown>>(sql`select e.event_type, count(*)::int as n from events e where e.detected_at >= now() - interval '24 hours' group by 1`).then((r) => {61 const byGroup: Record<string, number> = {};62 for (const row of r.rows) {63 const g = Object.entries(EVENT_GROUPS).find(([, spec]) => spec.types.includes(String(row.event_type)))?.[0] ?? "web";64 byGroup[g] = (byGroup[g] ?? 0) + Number(row.n);65 }66 return byGroup;67 }),68 db.execute<Record<string, unknown>>(sql`select (select count(*) from events where detected_at >= now() - interval '1 hour')::int as events_1h, (select count(*) from changes where detected_at >= now() - interval '1 hour')::int as changes_1h, (select count(*) from sensor_runs where started_at >= now() - interval '5 minutes')::int as checks_5m, (select count(distinct source_id) from events where detected_at >= now() - interval '24 hours')::int as active_sources_24h, (select count(distinct country) from events where detected_at >= now() - interval '24 hours' and country is not null)::int as active_countries_24h`).then((r) => r.rows[0]),69 ]);70 return { activity, desks: byDesk, rising_entities: rising, anomalies, silent_changes: silent, infrastructure: incidents, breaking, by_group_24h: groups, totals, generated_at: new Date().toISOString() };71 });72}7374/** Weak signals (spec §101): things that are NOT breaking yet but could become important. Clearly labelled as indicators. */75export async function radar(): Promise<Record<string, unknown>> {76 return cached("radar", 30_000, async () => {77 const [quietBursts, silentClusters, docBursts, repoBursts, statusChanges, developing, newSensorsFiring] = await Promise.all([78 // Sources far above their baseline but without any high-importance event79 db.execute<Record<string, unknown>>(sql`80 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 '3 hours' group by s.source_id),81 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),82 hot as (select distinct source_id from events where detected_at >= now() - interval '6 hours' and coalesce(signal_score, importance) >= 75)83 select so.id, so.name, so.domain, so.categories, cur.n::int as changes_3h, round(coalesce(base.per_hour,0)::numeric*24,1)::float as baseline_per_day,84 round((cur.n/3.0 / greatest(0.02, coalesce(base.per_hour, 0)))::numeric, 1)::float as ratio85 from cur join sources so on so.id = cur.source_id left join base on base.source_id = cur.source_id86 where so.kind = 'registry' and cur.n >= 4 and cur.source_id not in (select source_id from hot) and (coalesce(base.per_hour,0) = 0 or cur.n/3.0 / base.per_hour >= 4)87 order by ratio desc limit 12`).then((r) => r.rows),88 // Entities with several silent changes in 24 h89 db.execute<Record<string, unknown>>(sql`90 select en.id, en.name, en.type, count(*)::int as silent_24h, array_agg(distinct e.event_type) as types, max(e.detected_at) as last_at91 from events e join event_entities ee on ee.event_id = e.id join entities en on en.id = ee.entity_id92 where e.silent_change and e.detected_at >= now() - interval '24 hours' group by en.id, en.name, en.type having count(*) >= 2 order by silent_24h desc, last_at desc limit 12`).then((r) => r.rows),93 // Documentation / API modification bursts94 db.execute<Record<string, unknown>>(sql`95 select s.id, s.name, s.domain, count(*)::int as doc_changes_6h, array_agg(distinct e.event_type) as types, max(e.detected_at) as last_at96 from events e join sources s on s.id = e.source_id where e.detected_at >= now() - interval '6 hours' and e.event_type in ('documentation_change','API_change','api_change','availability_change')97 group by s.id, s.name, s.domain having count(*) >= 3 order by doc_changes_6h desc limit 10`).then((r) => r.rows),98 // Repository activity bursts (commits/tags/releases)99 db.execute<Record<string, unknown>>(sql`100 select s.id, s.name, s.domain, count(*)::int as repo_events_6h, max(e.detected_at) as last_at101 from events e join sources s on s.id = e.source_id join sensors sen on sen.id = e.sensor_id where e.detected_at >= now() - interval '6 hours' and sen.connector = 'github'102 group by s.id, s.name, s.domain having count(*) >= 3 order by repo_events_6h desc limit 10`).then((r) => r.rows),103 // Fresh status-page changes below the breaking bar104 listEvents({ limit: 10, group: "reliability", after: new Date(Date.now() - 3 * 3600e3).toISOString() }).then((r) => r.items.filter((e) => Number(e.importance) < 80)),105 clusterPrimaryEvents(sql`c.state = 'developing' and c.last_at >= now() - interval '12 hours'`, sql`c.velocity desc, c.last_at desc`, 8),106 // Sensors that produced their first-ever event in the last 24 h (new coverage lighting up)107 db.execute<Record<string, unknown>>(sql`108 select sen.id, sen.name, sen.source_id, s.name as source_name, min(e.detected_at) as first_event_at, count(*)::int as events109 from events e join sensors sen on sen.id = e.sensor_id join sources s on s.id = sen.source_id110 where e.detected_at >= now() - interval '24 hours' and not exists (select 1 from events e2 where e2.sensor_id = sen.id and e2.detected_at < now() - interval '24 hours')111 group by sen.id, sen.name, sen.source_id, s.name order by events desc limit 10`).then((r) => r.rows),112 ]);113 return { unusual_source_activity: quietBursts, silent_clusters: silentClusters, documentation_bursts: docBursts, repository_bursts: repoBursts, status_changes: statusChanges, developing, new_coverage: newSensorsFiring, generated_at: new Date().toISOString(), disclaimer: "Indicators, not facts: each item is a pattern in raw observations that has not (yet) produced a high-importance event." };114 });115}116117/** Entity insights (spec §26, §38, §102): heatmap, baseline, anomaly, velocity, rank, most active sensors. */118export async function entityInsights(entityId: string): Promise<Record<string, unknown>> {119 const [heat, today, last24, prev24, sensorsTop, silent24, breaking24, sources24, rank] = await Promise.all([120 db.execute<Record<string, unknown>>(sql`select day::text as day, events, silent, breaking, max_importance from entity_daily where entity_id = ${entityId} and day >= (now() at time zone 'UTC')::date - 34 order by day`).then((r) => r.rows),121 db.execute<{ n: string }>(sql`select count(*)::text as n from events e join event_entities ee on ee.event_id = e.id where ee.entity_id = ${entityId} and e.detected_at >= date_trunc('day', now() at time zone 'UTC')`).then((r) => Number(r.rows[0]?.n ?? 0)),122 db.execute<{ n: string }>(sql`select count(*)::text as n from events e join event_entities ee on ee.event_id = e.id where ee.entity_id = ${entityId} and e.detected_at >= now() - interval '24 hours'`).then((r) => Number(r.rows[0]?.n ?? 0)),123 db.execute<{ n: string }>(sql`select count(*)::text as n from events e join event_entities ee on ee.event_id = e.id where ee.entity_id = ${entityId} and e.detected_at >= now() - interval '48 hours' and e.detected_at < now() - interval '24 hours'`).then((r) => Number(r.rows[0]?.n ?? 0)),124 db.execute<Record<string, unknown>>(sql`select sen.id, sen.name, sen.type, sen.connector, sen.source_id, count(*)::int as events_7d, max(e.detected_at) as last_event_at from events e join event_entities ee on ee.event_id = e.id join sensors sen on sen.id = e.sensor_id where ee.entity_id = ${entityId} and e.detected_at >= now() - interval '7 days' group by sen.id, sen.name, sen.type, sen.connector, sen.source_id order by events_7d desc limit 8`).then((r) => r.rows),125 db.execute<{ n: string }>(sql`select count(*)::text as n from events e join event_entities ee on ee.event_id = e.id where ee.entity_id = ${entityId} and e.silent_change and e.detected_at >= now() - interval '24 hours'`).then((r) => Number(r.rows[0]?.n ?? 0)),126 db.execute<{ n: string }>(sql`select count(*)::text as n from events e join event_entities ee on ee.event_id = e.id where ee.entity_id = ${entityId} and coalesce(e.signal_score, e.importance) >= 80 and e.detected_at >= now() - interval '24 hours'`).then((r) => Number(r.rows[0]?.n ?? 0)),127 db.execute<{ n: string }>(sql`select count(distinct e.source_id)::text as n from events e join event_entities ee on ee.event_id = e.id where ee.entity_id = ${entityId} and e.detected_at >= now() - interval '24 hours'`).then((r) => Number(r.rows[0]?.n ?? 0)),128 entityRank(entityId),129 ]);130 const baselineDays = heat.filter((d) => String(d.day) < new Date().toISOString().slice(0, 10));131 const baselinePerDay = baselineDays.length ? baselineDays.reduce((n, d) => n + Number(d.events), 0) / Math.max(baselineDays.length, 7) : 0;132 // Rolling 24 h vs the 30-day daily baseline (a UTC "today" window is misleading in the first hours of the day).133 const anomaly = dailyAnomaly(last24, baselinePerDay, 24);134 const velocity = last24 && prev24 ? Math.round((last24 / prev24) * 100) / 100 : last24 ? 2 : 0;135 // Fill the 35-day heatmap with zero days136 const map = new Map(heat.map((d) => [String(d.day), d]));137 const days: Record<string, unknown>[] = [];138 for (let i = 34; i >= 0; i--) {139 const d = new Date(Date.now() - i * 86400e3).toISOString().slice(0, 10);140 days.push(map.get(d) ?? { day: d, events: 0, silent: 0, breaking: 0, max_importance: 0 });141 }142 return { heatmap: days, baseline_per_day: Math.round(baselinePerDay * 10) / 10, today, events_24h: last24, events_prev_24h: prev24, velocity_ratio: velocity, anomaly, silent_24h: silent24, breaking_24h: breaking24, sources_24h: sources24, most_active_sensors: sensorsTop, rank };143}144145/** WebSensor entity ranking (spec §105): computed over entities active in the last 7 days, cached 2 minutes. */146export async function entityRankings(limit = 100): Promise<Record<string, unknown>[]> {147 return cached(`rank:${limit}`, 120_000, async () => {148 const rows = await db.execute<Record<string, unknown>>(sql`149 with agg as (150 select ee.entity_id,151 count(*) filter (where e.detected_at >= now() - interval '24 hours') as e24,152 count(*) as e7,153 avg(coalesce(e.signal_score, e.importance)) as avg_signal,154 avg(case when e.evidence_label = 'CONFIRMED' then 1 else 0 end) as confirmed_ratio,155 count(distinct e.source_id) as sources,156 count(*) filter (where e.silent_change and e.detected_at >= now() - interval '24 hours') as silent24,157 count(*) filter (where coalesce(e.signal_score, e.importance) >= 80 and e.detected_at >= now() - interval '24 hours') as breaking24,158 max(e.detected_at) as last_at159 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() - interval '7 days' group by ee.entity_id),160 base as (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)161 select en.id, en.name, en.type, en.domain, en.importance, agg.e24::int as events_24h, agg.e7::int as events_7d, round(agg.avg_signal::numeric,1)::float as avg_signal, round(agg.confirmed_ratio::numeric,2)::float as confirmed_ratio, agg.sources::int as sources, agg.silent24::int as silent_24h, agg.breaking24::int as breaking_24h, agg.last_at, coalesce(base.per_day,0)::float as baseline_per_day162 from agg join entities en on en.id = agg.entity_id left join base on base.entity_id = agg.entity_id where en.type <> 'person'`);163 const scored = rows.rows.map((r) => ({ ...r, rank_score: entityRankScore({ importance: Number(r.importance), events24h: Number(r.events_24h), events7d: Number(r.events_7d), avgSignal: Number(r.avg_signal), confirmedRatio: Number(r.confirmed_ratio), uniqueSources: Number(r.sources), baselinePerDay: Number(r.baseline_per_day) }) }));164 scored.sort((a, b) => b.rank_score - a.rank_score);165 return scored.slice(0, limit).map((r, i) => ({ ...r, rank: i + 1 }));166 });167}168169export async function entityRank(entityId: string): Promise<{ rank: number | null; total: number; score: number | null }> {170 const all = await entityRankings(2000);171 const i = all.findIndex((r) => r.id === entityId);172 return { rank: i >= 0 ? i + 1 : null, total: all.length, score: i >= 0 ? Number(all[i]!.rank_score) : null };173}174175export async function clusterDetail(idOrSlug: string): Promise<Record<string, unknown> | null> {176 const c = (await db.execute<Record<string, unknown>>(sql`select ${CLUSTER_SELECT}, c.timeline from event_clusters c where c.id = ${idOrSlug} or c.slug = ${idOrSlug} limit 1`)).rows[0];177 if (!c) return null;178 const events = 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 e.cluster_id = ${String(c.id)} order by e.detected_at asc limit 200`);179 const ents = (c.entity_ids as string[]).length ? await db.execute<Record<string, unknown>>(sql`select id, name, type, importance from entities where id = any(${textArray(c.entity_ids as string[])}) order by importance desc`) : { rows: [] };180 // Propagation timeline (spec §35): offsets from the first signal.181 const first = new Date(String(c.first_at)).getTime();182 const propagation = events.rows.map((e) => ({ id: e.id, slug: e.slug, at: e.detected_at, offset_ms: new Date(String(e.detected_at)).getTime() - first, source: e.source, sensor: e.sensor, first_party: e.first_party, event_type: e.event_type, importance: e.importance, title: e.title }));183 const firstExternal = events.rows.find((e) => e.first_party === false);184 const firstParty = events.rows.find((e) => e.first_party !== false);185 const leadTime = firstParty && firstExternal ? new Date(String(firstExternal.detected_at)).getTime() - new Date(String(firstParty.detected_at)).getTime() : null;186 return { cluster: c, events: events.rows, entities: ents.rows, propagation, lead_time_ms: leadTime ?? c.lead_time_ms ?? null, first_party_signals: events.rows.filter((e) => e.first_party !== false).length, external_signals: events.rows.filter((e) => e.first_party === false).length };187}188189export async function countryList(): Promise<Record<string, unknown>[]> {190 return cached("countries", 60_000, async () => {191 const rows = await db.execute<Record<string, unknown>>(sql`192 select s.country, count(distinct s.id)::int as sources, (select count(*) from events e where e.country = s.country and e.detected_at >= now() - interval '24 hours')::int as events_24h,193 (select count(*) from events e where e.country = s.country and e.detected_at >= now() - interval '24 hours' and coalesce(e.signal_score, e.importance) >= 80)::int as breaking_24h194 from sources s where s.country is not null and s.enabled and s.kind = 'registry' group by s.country order by events_24h desc, sources desc`);195 return rows.rows.map((r) => ({ ...r, name: COUNTRIES[String(r.country)]?.name ?? String(r.country), slug: COUNTRIES[String(r.country)]?.slug ?? String(r.country).toLowerCase(), flag: COUNTRIES[String(r.country)]?.flag ?? "" }));196 });197}198199export async function countryDesk(code: string): Promise<Record<string, unknown>> {200 const c = code.toUpperCase();201 const cats = ["government", "finance", "infrastructure", "health", "news", "cyber", "ai", "science"];202 const [breaking, byCategory, sources, byType, silent, recent] = await Promise.all([203 listEvents({ limit: 10, country: c, order: "signal", after: new Date(Date.now() - 48 * 3600e3).toISOString() }).then((r) => r.items),204 Promise.all(cats.map(async (cat) => ({ category: cat, items: (await listEvents({ limit: 6, country: c, category: cat })).items }))).then((x) => x.filter((d) => d.items.length)),205 db.execute<Record<string, unknown>>(sql`select s.id, s.name, s.domain, s.tier, s.categories, s.first_party, (select count(*) from sensors x where x.source_id = s.id and x.enabled)::int as sensor_count, (select count(*) from events e where e.source_id = s.id and e.detected_at >= now() - interval '24 hours')::int as events_24h from sources s where s.country = ${c} and s.enabled and s.kind = 'registry' order by events_24h desc, s.tier, s.name limit 200`).then((r) => r.rows),206 db.execute<Record<string, unknown>>(sql`select event_type, count(*)::int as n from events where country = ${c} and detected_at >= now() - interval '7 days' group by 1 order by 2 desc limit 20`).then((r) => r.rows),207 listEvents({ limit: 8, country: c, silent_change: true }).then((r) => r.items),208 listEvents({ limit: 40, country: c }),209 ]);210 return { country: { code: c, name: COUNTRIES[c]?.name ?? c, flag: COUNTRIES[c]?.flag ?? "" }, breaking, by_category: byCategory, sources, by_type: byType, silent, recent: recent.items, nextCursor: recent.nextCursor };211}212213export async function categoryDesk(channel: string): Promise<Record<string, unknown>> {214 const cats = FEED_CHANNELS[channel] ?? [channel];215 const [breaking, silent, activeSources, byType, trendingEntities, recent, series] = await Promise.all([216 listEvents({ limit: 8, category: channel, order: "signal", after: new Date(Date.now() - 24 * 3600e3).toISOString() }).then((r) => r.items),217 listEvents({ limit: 8, category: channel, silent_change: true }).then((r) => r.items),218 db.execute<Record<string, unknown>>(sql`select s.id, s.name, s.domain, s.tier, 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 ${channel} = any(e.categories) and e.detected_at >= now() - interval '24 hours' and s.kind = 'registry' group by s.id, s.name, s.domain, s.tier, s.first_party order by events_24h desc limit 12`).then((r) => r.rows),219 db.execute<Record<string, unknown>>(sql`select event_type, count(*)::int as n from events where ${channel} = any(categories) and detected_at >= now() - interval '7 days' group by 1 order by 2 desc limit 16`).then((r) => r.rows),220 db.execute<Record<string, unknown>>(sql`select en.id, en.name, en.type, count(*)::int as events_24h, count(distinct e.source_id)::int as sources from events e join event_entities ee on ee.event_id = e.id join entities en on en.id = ee.entity_id where ${channel} = any(e.categories) and e.detected_at >= now() - interval '24 hours' group by en.id, en.name, en.type order by events_24h desc limit 10`).then((r) => r.rows),221 listEvents({ limit: 60, category: channel }),222 db.execute<Record<string, unknown>>(sql`select to_char(date_trunc('hour', detected_at at time zone 'UTC'), 'YYYY-MM-DD"T"HH24:00:00"Z"') as t, count(*)::int as n from events where ${channel} = any(categories) and detected_at >= now() - interval '48 hours' group by 1 order by 1`).then((r) => r.rows),223 ]);224 return { channel, categories: cats, breaking, silent, active_sources: activeSources, by_type: byType, trending_entities: trendingEntities, recent: recent.items, nextCursor: recent.nextCursor, series };225}226227/** Sensors table for a source, with polling metadata the source page shows (spec §27). */228export async function sourceSensors(sourceId: string): Promise<Record<string, unknown>[]> {229 const r = await db.execute<Record<string, unknown>>(sql`230 select id, name, url, type, connector, tier, health, status, priority, enabled, next_check_at, last_check_at, last_change_at, last_event_at, last_status, last_error, consecutive_errors, total_runs, total_not_modified, raw_changes, meaningful_changes, avg_latency_ms, base_interval_seconds, validated_at,231 etag is not null as has_etag, last_modified is not null as has_last_modified, etag, last_modified,232 (select count(*) from sensor_runs r where r.sensor_id = sensors.id and r.started_at >= now() - interval '24 hours')::int as checks_24h,233 (select count(*) from changes c where c.sensor_id = sensors.id and c.detected_at >= now() - interval '24 hours')::int as changes_24h,234 (select count(*) from snapshots sn where sn.sensor_id = sensors.id)::int as snapshot_count,235 case when last_check_at is not null and next_check_at is not null then extract(epoch from (next_check_at - last_check_at))::int else null end as current_interval_seconds236 from sensors where source_id = ${sourceId} order by priority, tier, name`);237 return r.rows;238}239