import { COUNTRIES, dailyAnomaly, entityRankScore, EVENT_GROUPS, FEED_CHANNELS } from "@websensor/core";
import { db, sql, textArray } from "@websensor/db";
import { cached } from "./cache";
import { EVENT_SELECT, listEvents } from "./queries";

/**
 * Intelligence read-models (spec §34–39, §101–105): breaking desk, pulse, radar, entity insights,
 * rankings, cluster propagation, country and category desks. Aggregates are cached briefly.
 */

const 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`;

async function clusterPrimaryEvents(where: ReturnType<typeof sql>, order: ReturnType<typeof sql>, limit: number): Promise<Record<string, unknown>[]> {
  const rows = await db.execute<Record<string, unknown>>(sql`
    select ${CLUSTER_SELECT},
      (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,
      (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 sources
    from event_clusters c where ${where} order by ${order} limit ${limit}`);
  return rows.rows;
}

export async function breakingDesk(): Promise<Record<string, unknown>> {
  return cached("breaking", 8_000, async () => {
    const [breaking, developing, confirmed, watching] = await Promise.all([
      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),
      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),
      clusterPrimaryEvents(sql`c.state = 'confirmed' and c.last_at >= now() - interval '48 hours'`, sql`c.last_at desc`, 20),
      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 ?? "")))),
    ]);
    return { breaking_now: breaking, developing, recently_confirmed: confirmed, watching, generated_at: new Date().toISOString() };
  });
}

export async function pulse(): Promise<Record<string, unknown>> {
  return cached("pulse", 10_000, async () => {
    const desks = ["ai", "cyber", "finance", "government", "infrastructure", "health", "science", "products"];
    const [activity, byDesk, rising, anomalies, silent, incidents, breaking, groups, totals] = await Promise.all([
      db.execute<Record<string, unknown>>(sql`
        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),
             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),
             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)
        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 changes
        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(() => []),
      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 }))),
      db.execute<Record<string, unknown>>(sql`
        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),
             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)
        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 acceleration
        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),
      db.execute<Record<string, unknown>>(sql`
        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),
             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)
        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,
          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,
          (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_baseline
        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),
      listEvents({ limit: 8, silent_change: true, order: "signal", after: new Date(Date.now() - 24 * 3600e3).toISOString() }).then((r) => r.items),
      listEvents({ limit: 8, group: "reliability", order: "recent", after: new Date(Date.now() - 6 * 3600e3).toISOString() }).then((r) => r.items),
      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),
      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) => {
        const byGroup: Record<string, number> = {};
        for (const row of r.rows) {
          const g = Object.entries(EVENT_GROUPS).find(([, spec]) => spec.types.includes(String(row.event_type)))?.[0] ?? "web";
          byGroup[g] = (byGroup[g] ?? 0) + Number(row.n);
        }
        return byGroup;
      }),
      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]),
    ]);
    return { activity, desks: byDesk, rising_entities: rising, anomalies, silent_changes: silent, infrastructure: incidents, breaking, by_group_24h: groups, totals, generated_at: new Date().toISOString() };
  });
}

/** Weak signals (spec §101): things that are NOT breaking yet but could become important. Clearly labelled as indicators. */
export async function radar(): Promise<Record<string, unknown>> {
  return cached("radar", 30_000, async () => {
    const [quietBursts, silentClusters, docBursts, repoBursts, statusChanges, developing, newSensorsFiring] = await Promise.all([
      // Sources far above their baseline but without any high-importance event
      db.execute<Record<string, unknown>>(sql`
        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),
             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),
             hot as (select distinct source_id from events where detected_at >= now() - interval '6 hours' and coalesce(signal_score, importance) >= 75)
        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,
          round((cur.n/3.0 / greatest(0.02, coalesce(base.per_hour, 0)))::numeric, 1)::float as ratio
        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' 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)
        order by ratio desc limit 12`).then((r) => r.rows),
      // Entities with several silent changes in 24 h
      db.execute<Record<string, unknown>>(sql`
        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_at
        from events e join event_entities ee on ee.event_id = e.id join entities en on en.id = ee.entity_id
        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),
      // Documentation / API modification bursts
      db.execute<Record<string, unknown>>(sql`
        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_at
        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')
        group by s.id, s.name, s.domain having count(*) >= 3 order by doc_changes_6h desc limit 10`).then((r) => r.rows),
      // Repository activity bursts (commits/tags/releases)
      db.execute<Record<string, unknown>>(sql`
        select s.id, s.name, s.domain, count(*)::int as repo_events_6h, max(e.detected_at) as last_at
        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'
        group by s.id, s.name, s.domain having count(*) >= 3 order by repo_events_6h desc limit 10`).then((r) => r.rows),
      // Fresh status-page changes below the breaking bar
      listEvents({ limit: 10, group: "reliability", after: new Date(Date.now() - 3 * 3600e3).toISOString() }).then((r) => r.items.filter((e) => Number(e.importance) < 80)),
      clusterPrimaryEvents(sql`c.state = 'developing' and c.last_at >= now() - interval '12 hours'`, sql`c.velocity desc, c.last_at desc`, 8),
      // Sensors that produced their first-ever event in the last 24 h (new coverage lighting up)
      db.execute<Record<string, unknown>>(sql`
        select sen.id, sen.name, sen.source_id, s.name as source_name, min(e.detected_at) as first_event_at, count(*)::int as events
        from events e join sensors sen on sen.id = e.sensor_id join sources s on s.id = sen.source_id
        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')
        group by sen.id, sen.name, sen.source_id, s.name order by events desc limit 10`).then((r) => r.rows),
    ]);
    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." };
  });
}

/** Entity insights (spec §26, §38, §102): heatmap, baseline, anomaly, velocity, rank, most active sensors. */
export async function entityInsights(entityId: string): Promise<Record<string, unknown>> {
  const [heat, today, last24, prev24, sensorsTop, silent24, breaking24, sources24, rank] = await Promise.all([
    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),
    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)),
    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)),
    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)),
    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),
    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)),
    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)),
    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)),
    entityRank(entityId),
  ]);
  const baselineDays = heat.filter((d) => String(d.day) < new Date().toISOString().slice(0, 10));
  const baselinePerDay = baselineDays.length ? baselineDays.reduce((n, d) => n + Number(d.events), 0) / Math.max(baselineDays.length, 7) : 0;
  // Rolling 24 h vs the 30-day daily baseline (a UTC "today" window is misleading in the first hours of the day).
  const anomaly = dailyAnomaly(last24, baselinePerDay, 24);
  const velocity = last24 && prev24 ? Math.round((last24 / prev24) * 100) / 100 : last24 ? 2 : 0;
  // Fill the 35-day heatmap with zero days
  const map = new Map(heat.map((d) => [String(d.day), d]));
  const days: Record<string, unknown>[] = [];
  for (let i = 34; i >= 0; i--) {
    const d = new Date(Date.now() - i * 86400e3).toISOString().slice(0, 10);
    days.push(map.get(d) ?? { day: d, events: 0, silent: 0, breaking: 0, max_importance: 0 });
  }
  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 };
}

/** WebSensor entity ranking (spec §105): computed over entities active in the last 7 days, cached 2 minutes. */
export async function entityRankings(limit = 100): Promise<Record<string, unknown>[]> {
  return cached(`rank:${limit}`, 120_000, async () => {
    const rows = await db.execute<Record<string, unknown>>(sql`
      with agg as (
        select ee.entity_id,
          count(*) filter (where e.detected_at >= now() - interval '24 hours') as e24,
          count(*) as e7,
          avg(coalesce(e.signal_score, e.importance)) as avg_signal,
          avg(case when e.evidence_label = 'CONFIRMED' then 1 else 0 end) as confirmed_ratio,
          count(distinct e.source_id) as sources,
          count(*) filter (where e.silent_change and e.detected_at >= now() - interval '24 hours') as silent24,
          count(*) filter (where coalesce(e.signal_score, e.importance) >= 80 and e.detected_at >= now() - interval '24 hours') as breaking24,
          max(e.detected_at) as last_at
        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),
      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)
      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_day
      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'`);
    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) }) }));
    scored.sort((a, b) => b.rank_score - a.rank_score);
    return scored.slice(0, limit).map((r, i) => ({ ...r, rank: i + 1 }));
  });
}

export async function entityRank(entityId: string): Promise<{ rank: number | null; total: number; score: number | null }> {
  const all = await entityRankings(2000);
  const i = all.findIndex((r) => r.id === entityId);
  return { rank: i >= 0 ? i + 1 : null, total: all.length, score: i >= 0 ? Number(all[i]!.rank_score) : null };
}

export async function clusterDetail(idOrSlug: string): Promise<Record<string, unknown> | null> {
  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];
  if (!c) return null;
  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`);
  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: [] };
  // Propagation timeline (spec §35): offsets from the first signal.
  const first = new Date(String(c.first_at)).getTime();
  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 }));
  const firstExternal = events.rows.find((e) => e.first_party === false);
  const firstParty = events.rows.find((e) => e.first_party !== false);
  const leadTime = firstParty && firstExternal ? new Date(String(firstExternal.detected_at)).getTime() - new Date(String(firstParty.detected_at)).getTime() : null;
  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 };
}

export async function countryList(): Promise<Record<string, unknown>[]> {
  return cached("countries", 60_000, async () => {
    const rows = await db.execute<Record<string, unknown>>(sql`
      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,
        (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_24h
      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`);
    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 ?? "" }));
  });
}

export async function countryDesk(code: string): Promise<Record<string, unknown>> {
  const c = code.toUpperCase();
  const cats = ["government", "finance", "infrastructure", "health", "news", "cyber", "ai", "science"];
  const [breaking, byCategory, sources, byType, silent, recent] = await Promise.all([
    listEvents({ limit: 10, country: c, order: "signal", after: new Date(Date.now() - 48 * 3600e3).toISOString() }).then((r) => r.items),
    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)),
    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),
    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),
    listEvents({ limit: 8, country: c, silent_change: true }).then((r) => r.items),
    listEvents({ limit: 40, country: c }),
  ]);
  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 };
}

export async function categoryDesk(channel: string): Promise<Record<string, unknown>> {
  const cats = FEED_CHANNELS[channel] ?? [channel];
  const [breaking, silent, activeSources, byType, trendingEntities, recent, series] = await Promise.all([
    listEvents({ limit: 8, category: channel, order: "signal", after: new Date(Date.now() - 24 * 3600e3).toISOString() }).then((r) => r.items),
    listEvents({ limit: 8, category: channel, silent_change: true }).then((r) => r.items),
    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),
    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),
    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),
    listEvents({ limit: 60, category: channel }),
    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),
  ]);
  return { channel, categories: cats, breaking, silent, active_sources: activeSources, by_type: byType, trending_entities: trendingEntities, recent: recent.items, nextCursor: recent.nextCursor, series };
}

/** Sensors table for a source, with polling metadata the source page shows (spec §27). */
export async function sourceSensors(sourceId: string): Promise<Record<string, unknown>[]> {
  const r = await db.execute<Record<string, unknown>>(sql`
    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,
      etag is not null as has_etag, last_modified is not null as has_last_modified, etag, last_modified,
      (select count(*) from sensor_runs r where r.sensor_id = sensors.id and r.started_at >= now() - interval '24 hours')::int as checks_24h,
      (select count(*) from changes c where c.sensor_id = sensors.id and c.detected_at >= now() - interval '24 hours')::int as changes_24h,
      (select count(*) from snapshots sn where sn.sensor_id = sensors.id)::int as snapshot_count,
      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_seconds
    from sensors where source_id = ${sourceId} order by priority, tier, name`);
  return r.rows;
}