import { EVENT_GROUPS, parseSearch } from "@websensor/core";
import { db, sql, textArray } from "@websensor/db";

/** Read-model helpers shared by REST routes. All return plain JSON-ready objects. */

export interface EventFilters {
  after?: string;
  before?: string;
  category?: string;
  entity?: string;
  source?: string;
  domain?: string;
  sensor?: string;
  cluster?: string;
  importance_min?: number;
  confidence_min?: number;
  signal_min?: number;
  event_type?: string;
  group?: string;
  silent_change?: boolean;
  first_party?: boolean;
  confirmed?: boolean;
  country?: string;
  language?: string;
  change_class?: string;
  q?: string;
  limit: number;
  cursor?: string;
  order?: "recent" | "importance" | "signal";
  /** include events from custom (owner) sources — only set by owner-scoped routes */
  includeCustom?: boolean;
}

export const EVENT_SELECT = sql`
  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,
  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,
  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,
  e.importance_components, e.processing_version,
  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,
  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,
  json_build_object('id', sen.id, 'name', sen.name, 'type', sen.type, 'connector', sen.connector, 'tier', sen.tier) as sensor,
  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)
            from event_entities ee join entities en on en.id = ee.entity_id where ee.event_id = e.id), '[]'::json) as entities,
  (select event_count from event_clusters c where c.id = e.cluster_id) as cluster_size,
  (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`;

/** Build WHERE conditions from filters (shared by list, RSS, watchlist and desk queries). */
export function eventConditions(f: EventFilters): ReturnType<typeof sql>[] {
  const conds = [sql`true`];
  let free = f.q?.trim() ?? "";
  // Advanced syntax inside q (entity:… type:… after:…) is merged into the filters.
  if (free) {
    const p = parseSearch(free);
    free = p.text;
    const pf = p.filters;
    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 };
  }
  if (!f.includeCustom) conds.push(sql`s.kind = 'registry'`);
  if (f.after) conds.push(sql`e.detected_at > ${new Date(f.after)}`);
  if (f.before) conds.push(sql`e.detected_at < ${new Date(f.before)}`);
  if (f.category) conds.push(sql`${f.category} = any(e.categories)`);
  if (f.source) conds.push(sql`e.source_id = ${f.source}`);
  if (f.sensor) conds.push(sql`e.sensor_id = ${f.sensor}`);
  if (f.cluster) conds.push(sql`e.cluster_id = ${f.cluster}`);
  if (f.domain) conds.push(sql`s.domain = ${f.domain}`);
  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})`);
  if (f.importance_min !== undefined) conds.push(sql`e.importance >= ${f.importance_min}`);
  if (f.confidence_min !== undefined) conds.push(sql`e.confidence >= ${f.confidence_min}`);
  if (f.signal_min !== undefined) conds.push(sql`coalesce(e.signal_score, e.importance) >= ${f.signal_min}`);
  if (f.event_type) conds.push(sql`e.event_type = any(${textArray(f.event_type.split(",").map((t) => t.trim()).filter(Boolean))})`);
  if (f.group && EVENT_GROUPS[f.group]) conds.push(sql`e.event_type = any(${textArray(EVENT_GROUPS[f.group]!.types)})`);
  if (f.silent_change !== undefined) conds.push(sql`e.silent_change = ${f.silent_change}`);
  if (f.first_party !== undefined) conds.push(sql`e.first_party = ${f.first_party}`);
  if (f.confirmed) conds.push(sql`e.evidence_label = 'CONFIRMED'`);
  if (f.country) conds.push(sql`e.country = ${f.country.toUpperCase()}`);
  if (f.language) conds.push(sql`e.language = ${f.language.toLowerCase()}`);
  if (f.change_class) conds.push(sql`e.change_class = ${f.change_class}`);
  if (free) conds.push(sql`(e.search @@ websearch_to_tsquery('english', ${free}) or e.title ilike ${"%" + free + "%"})`);
  return conds;
}

export async function listEvents(f: EventFilters): Promise<{ items: Record<string, unknown>[]; nextCursor: string | null }> {
  const conds = eventConditions(f);
  if (f.cursor) {
    const [ts, id] = decodeCursor(f.cursor);
    if (f.order === "importance") conds.push(sql`(e.importance, e.id) < (${Number(ts)}, ${id})`);
    else if (f.order === "signal") conds.push(sql`(coalesce(e.signal_score, e.importance), e.id) < (${Number(ts)}, ${id})`);
    else conds.push(sql`(e.detected_at, e.id) < (${new Date(Number(ts))}, ${id})`);
  }
  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`;
  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}`);
  const items = rows.rows.slice(0, f.limit);
  const last = items[items.length - 1];
  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;
  return { items, nextCursor };
}

export async function countEvents(f: EventFilters): Promise<number> {
  const conds = eventConditions(f);
  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 `)}`);
  return Number(r.rows[0]?.n ?? 0);
}

export async function getEvent(idOrSlug: string): Promise<Record<string, unknown> | null> {
  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`);
  return rows.rows[0] ?? null;
}

export async function relatedEvents(ev: Record<string, unknown>, limit = 8): Promise<Record<string, unknown>[]> {
  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 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)}))
    order by (e.cluster_id = ${String(ev.cluster_id ?? "")}) desc, e.detected_at desc limit ${limit}`);
  return rows.rows;
}

/** Previous meaningful events on the same sensor (historical context for "what changed" — spec §79–80). */
export async function sensorHistory(sensorId: string, beforeEventId: string, limit = 6): Promise<Record<string, unknown>[]> {
  const rows = await db.execute<Record<string, unknown>>(sql`
    select e.id, e.slug, e.title, e.event_type, e.importance, e.silent_change, e.detected_at, e.field_changes from events e
    where e.sensor_id = ${sensorId} and e.id <> ${beforeEventId} order by e.detected_at desc limit ${limit}`);
  return rows.rows;
}

export async function stats(): Promise<Record<string, unknown>> {
  const [r] = (
    await db.execute<Record<string, unknown>>(sql`
    select
      (select count(*) from sources where enabled and kind = 'registry') as sources,
      (select count(*) from sensors where enabled and status <> 'SHADOW') as sensors,
      (select count(*) from sensors where enabled and status = 'SHADOW') as sensors_shadow,
      (select count(*) from sources where enabled and kind = 'registry' and origin = 'factory') as sources_factory,
      (select count(*) from entities) as entities,
      (select count(*) from snapshots) as snapshots,
      (select count(*) from events) as events_total,
      (select count(*) from events where detected_at >= now() - interval '24 hours') as events_24h,
      (select count(*) from events where detected_at >= now() - interval '1 hour') as events_1h,
      (select count(*) from events where detected_at >= now() - interval '24 hours' and silent_change) as silent_24h,
      (select count(*) from events where detected_at >= now() - interval '24 hours' and coalesce(signal_score, importance) >= 80) as breaking_24h,
      (select count(*) from event_clusters where state = 'breaking') as breaking_now,
      (select count(*) from event_clusters where state = 'developing') as developing_now,
      (select count(*) from changes where detected_at >= now() - interval '24 hours') as changes_24h,
      (select coalesce(sum(checks),0) from metrics_daily where day = (now() at time zone 'UTC')::date) as checks_today,
      (select coalesce(sum(not_modified),0) from metrics_daily where day = (now() at time zone 'UTC')::date) as not_modified_today,
      (select coalesce(sum(bytes),0) from metrics_daily where day = (now() at time zone 'UTC')::date) as bytes_today,
      (select count(*) from sensor_runs where started_at >= now() - interval '1 hour') as checks_last_hour,
      (select count(*) from sensor_runs where started_at >= now() - interval '5 minutes') as checks_last_5m,
      (select count(*) from sensor_runs where started_at >= now() - interval '5 minutes' and outcome = 'not_modified') as not_modified_last_5m,
      (select count(*) from sensors where health = 'UP' and enabled and status <> 'SHADOW') as sensors_up,
      (select count(*) from sensors where health in ('DEGRADED','ERROR','RATE_LIMITED') and enabled and status <> 'SHADOW') as sensors_degraded,
      (select count(*) from sources where enabled and first_party and kind = 'registry') as sources_first_party,
      (select count(distinct country) from sources where country is not null) as countries,
      (select max(started_at) from sensor_runs) as last_check_at,
      (select max(detected_at) from events) as last_event_at,
      (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,
      (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`)
  ).rows;
  const out = Object.fromEntries(Object.entries(r ?? {}).map(([k, v]) => [k, typeof v === "string" && /^\d+(\.\d+)?$/.test(v) ? Number(v) : v]));
  out.checks_per_min = Math.round(Number(out.checks_last_5m ?? 0) / 5);
  out.events_per_min = Math.round((Number(out.events_1h ?? 0) / 60) * 10) / 10;
  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;
  return out;
}

export async function trending(hours = 24, limit = 12): Promise<Record<string, unknown>[]> {
  const rows = await db.execute<Record<string, unknown>>(sql`
    with cur as (
      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,
             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,
             count(*) filter (where e.detected_at >= now() - make_interval(hours => ${hours}) / 4) as recent_quarter
      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),
    prev as (
      select ee.entity_id, count(*) as n from events e join event_entities ee on ee.event_id = e.id
      where e.detected_at >= now() - make_interval(hours => ${hours * 2}) and e.detected_at < now() - make_interval(hours => ${hours}) 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 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,
      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,
      cur.recent_quarter::int as recent_quarter,
      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,
      round((100 * (1 - exp(-(
          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)
          + 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)
        ) / 110.0)))::numeric, 1)::float as score
    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_id
    order by score desc limit ${limit}`);
  return rows.rows;
}

export async function sourceActivity(sourceId: string): Promise<Record<string, unknown>> {
  const [r] = (
    await db.execute<Record<string, unknown>>(sql`
    select
      (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,
      (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,
      (select count(*) from events where source_id = ${sourceId} and detected_at >= now() - interval '24 hours')::int as events_24h,
      (select count(*) from events where source_id = ${sourceId} and detected_at >= now() - interval '14 days')::int as events_14d,
      (select count(*) from events where source_id = ${sourceId} and detected_at >= now() - interval '24 hours' and silent_change)::int as silent_24h,
      (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`)
  ).rows;
  const c2h = Number(r?.changes_2h ?? 0);
  const baselinePerHour = Number(r?.changes_14d ?? 0) / (14 * 24);
  const currentPerHour = c2h / 2;
  let anomaly = 0;
  if (baselinePerHour <= 0) anomaly = currentPerHour > 2 ? 70 : currentPerHour > 0 ? 40 : 0;
  else {
    const ratio = currentPerHour / baselinePerHour;
    anomaly = ratio <= 1 ? ratio * 30 : Math.min(100, 30 + 25 * Math.log2(ratio));
  }
  return { ...r, baseline_changes_per_day: Math.round(baselinePerHour * 24 * 10) / 10, activity_score: Math.round(anomaly * 10) / 10 };
}

/** Source quality score (spec §48): success rate, latency, structured share, usefulness (events/raw), consistency. Distinct from importance. */
export async function sourceQuality(sourceId: string): Promise<Record<string, unknown>> {
  const [r] = (
    await db.execute<Record<string, unknown>>(sql`
    select
      (select count(*) from sensors where source_id = ${sourceId} and enabled)::int as sensors,
      (select count(*) from sensors where source_id = ${sourceId} and enabled and health = 'UP')::int as sensors_up,
      (select avg(avg_latency_ms) from sensors where source_id = ${sourceId} and enabled)::int as avg_latency_ms,
      (select coalesce(sum(raw_changes),0) from sensors where source_id = ${sourceId})::int as raw_changes,
      (select coalesce(sum(meaningful_changes),0) from sensors where source_id = ${sourceId})::int as meaningful_changes,
      (select coalesce(sum(total_runs),0) from sensors where source_id = ${sourceId})::int as total_runs,
      (select coalesce(sum(total_not_modified),0) from sensors where source_id = ${sourceId})::int as total_not_modified,
      (select count(*) from sensors where source_id = ${sourceId} and enabled and connector <> 'http')::int as structured_sensors,
      (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,
      (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,
      (select avg(confidence) from events where source_id = ${sourceId} and detected_at >= now() - interval '30 days')::float as avg_confidence`)
  ).rows;
  const sensors = Number(r?.sensors ?? 0);
  const success = Number(r?.checks_7d) ? 1 - Number(r?.errors_7d) / Number(r?.checks_7d) : sensors ? Number(r?.sensors_up) / sensors : 1;
  const latency = Number(r?.avg_latency_ms ?? 800);
  const latencyScore = latency <= 400 ? 1 : latency <= 1500 ? 0.8 : latency <= 4000 ? 0.6 : 0.4;
  const structured = sensors ? Number(r?.structured_sensors) / sensors : 0;
  const raw = Number(r?.raw_changes ?? 0);
  const usefulness = raw >= 5 ? Math.min(1, (Number(r?.meaningful_changes) / raw) * 2) : 0.6;
  const conf = Number(r?.avg_confidence ?? 70) / 100;
  const score = Math.round(100 * (0.35 * success + 0.15 * latencyScore + 0.15 * structured + 0.2 * usefulness + 0.15 * conf));
  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 };
}

export function encodeCursor(a: string, b: string): string {
  return Buffer.from(`${a}|${b}`).toString("base64url");
}
export function decodeCursor(c: string): [string, string] {
  const s = Buffer.from(c, "base64url").toString("utf8");
  const i = s.indexOf("|");
  return [s.slice(0, i), s.slice(i + 1)];
}