/**
 * Containment-aware aggregates shared by operator / metro / country payloads, /compare and the dashboard:
 * PipelineBreakdown (facility lifecycle buckets + project stages), ExpansionVelocity, market concentration (HHI),
 * momentum components and the CoverageRow. Every MW figure is a sum of published site-scoped figures — no estimate.
 */
import type { CoverageRow, ExpansionVelocity, MarketConcentration, MarketMomentum, PipelineBreakdown } from "@dci/core";
import { pg, facilityView, knownMwAgg, pipelineMwAgg, countedAgg, hasMwAgg, projectLive, OPERATIONAL_SET, CONSTRUCTION_SET, PLANNED_SET, GRID_EVENT_TYPES, type Fragment, type Sql } from "./sql.js";
import { int, num, reqStr, str, type Row } from "./rows.js";
import { asStatus, round2, share } from "./dto.js";

/** Scope of an aggregate: a condition on the facility view row `f` and on the project row `p`. */
export interface Scope { facility: Fragment; project: Fragment; event?: Fragment; cloud?: Fragment }

export function scopeFor(sql: Sql, s: { operatorId?: string; metroId?: string; countryIso2?: string }): Scope {
  if (s.operatorId) return { facility: sql`(f.operator_id = ${s.operatorId} or f.owner_id = ${s.operatorId})`, project: sql`p.operator_id = ${s.operatorId}`, event: sql`e.operator_id = ${s.operatorId}`, cloud: sql`r.provider_id = ${s.operatorId}` };
  if (s.metroId) return { facility: sql`f.metro_id = ${s.metroId}`, project: sql`p.metro_id = ${s.metroId}`, event: sql`(e.metro_id = ${s.metroId} or (e.entity_type = 'facility' and e.entity_id in (select id from facilities where metro_id = ${s.metroId})))`, cloud: sql`r.metro_id = ${s.metroId}` };
  if (s.countryIso2) return { facility: sql`f.country_iso2 = ${s.countryIso2}`, project: sql`p.country_iso2 = ${s.countryIso2}`, event: sql`e.country_iso2 = ${s.countryIso2}`, cloud: sql`r.country_iso2 = ${s.countryIso2}` };
  return { facility: sql`true`, project: sql`true`, event: sql`true`, cloud: sql`true` };
}

export async function pipelineBreakdown(scope: Scope): Promise<PipelineBreakdown> {
  const sql = pg();
  const known = knownMwAgg(sql), pipe = pipelineMwAgg(sql), counted = countedAgg(sql), hasMw = hasMwAgg(sql);
  const [fr, pr] = await Promise.all([
    sql<Row[]>`select
        count(*) filter (where ${counted})::int as total,
        count(*) filter (where ${counted} and ${hasMw})::int as with_mw,
        count(*) filter (where ${counted} and f.status = any(${OPERATIONAL_SET}))::int as op_n, sum(${known}) filter (where f.status = any(${OPERATIONAL_SET}))::float as op_mw,
        count(*) filter (where ${counted} and f.status = any(${CONSTRUCTION_SET}))::int as con_n, sum(${pipe}) filter (where f.status = any(${CONSTRUCTION_SET}))::float as con_mw,
        count(*) filter (where ${counted} and f.status in ('approved', 'permitting'))::int as app_n, sum(${pipe}) filter (where f.status in ('approved', 'permitting'))::float as app_mw,
        count(*) filter (where ${counted} and f.status in ('rumored', 'proposed', 'announced', 'delayed'))::int as ann_n, sum(${pipe}) filter (where f.status in ('rumored', 'proposed', 'announced', 'delayed'))::float as ann_mw
      from ${facilityView(sql)} f where ${scope.facility}`,
    sql<Row[]>`select p.status, count(*)::int as n, sum(p.planned_mw)::float as mw from projects p where ${projectLive(sql)} and ${scope.project} group by p.status order by n desc`,
  ]);
  const r = fr[0] ?? {};
  const total = int(r.total);
  return {
    operational: { count: int(r.op_n), mw: round2(num(r.op_mw)) },
    construction: { count: int(r.con_n), mw: round2(num(r.con_mw)) },
    approved: { count: int(r.app_n), mw: round2(num(r.app_mw)) },
    announced: { count: int(r.ann_n), mw: round2(num(r.ann_mw)) },
    projects: pr.map((x) => ({ status: asStatus(x.status), count: int(x.n), mw: round2(num(x.mw)) })),
    mwCoverage: total ? share(int(r.with_mw), total) : 0,
  };
}

const WINDOWS: Array<{ label: "12m" | "3y" | "5y"; months: number }> = [{ label: "12m", months: 12 }, { label: "3y", months: 36 }, { label: "5y", months: 60 }];

/** Expansion velocity: new facilities (opened_on else first_seen), projects (announced_on else created_at), new countries / metros per window. */
export async function expansionVelocity(scope: Scope): Promise<ExpansionVelocity> {
  const sql = pg();
  const known = knownMwAgg(sql), counted = countedAgg(sql);
  const openedDate = sql`(case when f.opened_on ~ '^\\d{4}-\\d{2}-\\d{2}' then f.opened_on::date when f.opened_on ~ '^\\d{4}-\\d{2}$' then (f.opened_on || '-01')::date when f.opened_on ~ '^\\d{4}$' then (f.opened_on || '-01-01')::date else f.first_seen::date end)`;
  const annDate = sql`(case when p.announced_on ~ '^\\d{4}-\\d{2}-\\d{2}' then p.announced_on::date when p.announced_on ~ '^\\d{4}-\\d{2}$' then (p.announced_on || '-01')::date when p.announced_on ~ '^\\d{4}$' then (p.announced_on || '-01-01')::date else p.created_at::date end)`;
  const windows = await Promise.all(WINDOWS.map(async (w) => {
    const since = sql`(current_date - make_interval(months => ${w.months}))`;
    const [fr, pr, geo] = await Promise.all([
      sql<Row[]>`select count(*) filter (where ${counted})::int as n, sum(${known})::float as mw from ${facilityView(sql)} f where ${scope.facility} and ${openedDate} >= ${since}`,
      sql<Row[]>`select count(*)::int as n, sum(p.planned_mw)::float as mw from projects p where ${projectLive(sql)} and ${scope.project} and ${annDate} >= ${since}`,
      sql<Row[]>`with fx as (select f.country_iso2, f.metro_id, min(${openedDate}) as first_d from ${facilityView(sql)} f where ${scope.facility} group by 1, 2)
        select count(distinct country_iso2) filter (where country_iso2 is not null and first_d >= ${since} and not exists (select 1 from fx f2 where f2.country_iso2 = fx.country_iso2 and f2.first_d < ${since}))::int as countries,
               count(distinct metro_id) filter (where metro_id is not null and first_d >= ${since} and not exists (select 1 from fx f2 where f2.metro_id = fx.metro_id and f2.first_d < ${since}))::int as metros
        from fx`,
    ]);
    return { label: w.label, newFacilities: int(fr[0]?.n), newProjects: int(pr[0]?.n), newCountries: int(geo[0]?.countries), newMetros: int(geo[0]?.metros), openedMw: round2(num(fr[0]?.mw)), announcedMw: round2(num(pr[0]?.mw)) };
  }));
  const years = await sql<Row[]>`select extract(year from ${openedDate})::int as year, bool_and(f.opened_on ~ '^\\d{4}') as opened_basis,
      count(*) filter (where ${counted})::int as n, array_agg(distinct f.country_iso2) filter (where f.country_iso2 is not null) as countries, array_agg(distinct f.metro_id) filter (where f.metro_id is not null) as metros
    from ${facilityView(sql)} f where ${scope.facility} group by 1 order by 1`;
  const seenC = new Set<string>(), seenM = new Set<string>();
  let cum = 0;
  const countriesOverTime: ExpansionVelocity["countriesOverTime"] = [];
  for (const y of years) {
    const year = int(y.year);
    if (!year) continue;
    for (const c of (y.countries as string[] | null) ?? []) seenC.add(c);
    for (const m of (y.metros as string[] | null) ?? []) seenM.add(m);
    cum += int(y.n);
    countriesOverTime.push({ year, countries: seenC.size, metros: seenM.size, facilities: cum, basis: y.opened_basis === true ? "opened" : "first_seen" });
  }
  return { windows, countriesOverTime };
}

export const CONCENTRATION_NOTE = "HHI = Σ (share × 100)² over operators, 0–10 000; 10 000 = one operator. Computed on counted facilities (campus rows with buildings excluded) and, separately, on known operational MW only when published figures exist — `coverage` is the share of facilities with a figure; treat the MW view as partial when coverage is low.";

export async function concentration(scope: Scope): Promise<MarketConcentration> {
  const sql = pg();
  const known = knownMwAgg(sql), counted = countedAgg(sql), hasMw = hasMwAgg(sql);
  const rows = await sql<Row[]>`select o.id, o.slug, o.name, count(*) filter (where ${counted})::int as n, sum(${known}) filter (where f.status = any(${OPERATIONAL_SET}))::float as mw, count(*) filter (where ${counted} and ${hasMw})::int as with_mw
    from ${facilityView(sql)} f join operators o on o.id = f.operator_id where ${scope.facility} group by o.id, o.slug, o.name order by n desc, mw desc nulls last`;
  const totalN = rows.reduce((a, r) => a + int(r.n), 0);
  const totalCounted = totalN;
  const withMw = rows.reduce((a, r) => a + int(r.with_mw), 0);
  const totalMw = rows.reduce((a, r) => a + (num(r.mw) ?? 0), 0);
  const hhi = (shares: number[]) => Math.round(shares.reduce((a, s) => a + (s * 100) ** 2, 0));
  const fShares = rows.map((r) => (totalN ? int(r.n) / totalN : 0));
  const facilities = {
    hhi: hhi(fShares),
    top3Share: share(rows.slice(0, 3).reduce((a, r) => a + int(r.n), 0), totalN),
    top5Share: share(rows.slice(0, 5).reduce((a, r) => a + int(r.n), 0), totalN),
    top: rows.slice(0, 10).map((r) => ({ id: reqStr(r.id), slug: reqStr(r.slug), name: reqStr(r.name), count: int(r.n), share: share(int(r.n), totalN) })),
  };
  let knownMw: MarketConcentration["knownMw"] = null;
  if (totalMw > 0) {
    const byMw = rows.filter((r) => (num(r.mw) ?? 0) > 0).sort((a, b) => (num(b.mw) ?? 0) - (num(a.mw) ?? 0));
    knownMw = {
      hhi: hhi(byMw.map((r) => (num(r.mw) ?? 0) / totalMw)),
      top3Share: share(byMw.slice(0, 3).reduce((a, r) => a + (num(r.mw) ?? 0), 0), totalMw),
      top5Share: share(byMw.slice(0, 5).reduce((a, r) => a + (num(r.mw) ?? 0), 0), totalMw),
      coverage: share(withMw, totalCounted),
      top: byMw.slice(0, 10).map((r) => ({ id: reqStr(r.id), slug: reqStr(r.slug), name: reqStr(r.name), mw: round2(num(r.mw)) ?? 0, share: share(num(r.mw) ?? 0, totalMw) })),
    };
  }
  return { operatorCount: rows.length, facilities, knownMw, note: CONCENTRATION_NOTE };
}

/** 12-month momentum components (never collapsed into a score). */
export async function momentum(scope: Scope): Promise<MarketMomentum> {
  const sql = pg();
  const since = sql`(current_date - interval '12 months')`;
  const annDate = sql`(case when p.announced_on ~ '^\\d{4}-\\d{2}-\\d{2}' then p.announced_on::date when p.announced_on ~ '^\\d{4}-\\d{2}$' then (p.announced_on || '-01')::date when p.announced_on ~ '^\\d{4}$' then (p.announced_on || '-01-01')::date else p.created_at::date end)`;
  const conDate = sql`(case when p.construction_started_on ~ '^\\d{4}-\\d{2}-\\d{2}' then p.construction_started_on::date when p.construction_started_on ~ '^\\d{4}-\\d{2}$' then (p.construction_started_on || '-01')::date when p.construction_started_on ~ '^\\d{4}$' then (p.construction_started_on || '-01-01')::date else null end)`;
  const openedDate = sql`(case when f.opened_on ~ '^\\d{4}-\\d{2}-\\d{2}' then f.opened_on::date when f.opened_on ~ '^\\d{4}-\\d{2}$' then (f.opened_on || '-01')::date when f.opened_on ~ '^\\d{4}$' then (f.opened_on || '-01-01')::date else null end)`;
  const [pa, pc, fo, ne, cr, ev] = await Promise.all([
    sql<Row[]>`select count(*)::int as n, sum(p.planned_mw)::float as mw from projects p where ${projectLive(sql)} and ${scope.project} and ${annDate} >= ${since}`,
    sql<Row[]>`select count(*)::int as n, sum(p.planned_mw)::float as mw from projects p where ${projectLive(sql)} and ${scope.project} and (${conDate} >= ${since} or (p.status = 'under_construction' and ${conDate} is null and exists (select 1 from events e where e.project_id = p.id and e.event_type in ('construction_started', 'project_status_changed') and e.new_value::text ilike '%under_construction%' and e.detected_at >= ${since})))`,
    sql<Row[]>`select count(*) filter (where ${countedAgg(sql)})::int as n from ${facilityView(sql)} f where ${scope.facility} and f.status = any(${OPERATIONAL_SET}) and ${openedDate} >= ${since}`,
    sql<Row[]>`with fx as (select f.operator_id, min(coalesce(${openedDate}, f.first_seen::date)) as first_d from ${facilityView(sql)} f where ${scope.facility} and f.operator_id is not null group by 1)
      select o.id, o.slug, o.name from fx join operators o on o.id = fx.operator_id where fx.first_d >= ${since} order by fx.first_d desc limit 20`,
    sql<Row[]>`select count(*)::int as n from cloud_regions r where ${scope.cloud ?? sql`true`} and ((r.launched_on ~ '^\\d{4}' and (case when r.launched_on ~ '^\\d{4}-\\d{2}-\\d{2}' then r.launched_on::date when r.launched_on ~ '^\\d{4}-\\d{2}$' then (r.launched_on || '-01')::date else (left(r.launched_on, 4) || '-01-01')::date end) >= ${since}) or r.created_at::date >= ${since})`,
    sql<Row[]>`select count(*)::int as total, count(*) filter (where e.event_type = any(${GRID_EVENT_TYPES}))::int as grid from events e where ${scope.event ?? sql`true`} and e.review_status <> 'rejected' and e.detected_at >= ${since}`,
  ]);
  return {
    window: "12m",
    projectsAnnounced: int(pa[0]?.n),
    projectsEnteredConstruction: int(pc[0]?.n),
    constructionMw: round2(num(pc[0]?.mw)),
    announcedMw: round2(num(pa[0]?.mw)),
    newEntrants: ne.map((r) => ({ id: reqStr(r.id), slug: reqStr(r.slug), name: reqStr(r.name) })),
    facilitiesOpened: int(fo[0]?.n),
    cloudRegionsAdded: int(cr[0]?.n),
    gridEvents: int(ev[0]?.grid),
    eventsTotal: int(ev[0]?.total),
  };
}

/** CoverageRow for a scope (containment-aware counts, share of facilities with each field known). */
export async function coverageRow(key: string, name: string, slug: string, scope: Scope): Promise<CoverageRow> {
  const sql = pg();
  const counted = countedAgg(sql), hasMw = hasMwAgg(sql);
  const [fr, pr] = await Promise.all([
    sql<Row[]>`select count(*) filter (where ${counted})::int as n,
        count(*) filter (where ${counted} and ${hasMw})::int as with_mw,
        count(*) filter (where ${counted} and f.operator_id is not null)::int as with_op,
        count(*) filter (where ${counted} and f.lat is not null and f.geo_precision in ('exact', 'parcel', 'street'))::int as precise,
        count(*) filter (where ${counted} and f.lat is not null)::int as any_loc,
        count(*) filter (where ${counted} and f.status <> 'unknown')::int as with_status,
        count(*) filter (where ${counted} and f.opened_on ~ '^\\d{4}')::int as with_open,
        count(*) filter (where ${counted} and f.source_count >= 2)::int as multi,
        count(*) filter (where ${counted} and (coalesce(f.carriers_count, 0) > 0 or coalesce(f.ixp_count, 0) > 0 or exists (select 1 from facility_tenants t where t.facility_id = f.id) or exists (select 1 from facility_ixps x where x.facility_id = f.id)))::int as conn,
        count(*) filter (where ${counted} and exists (select 1 from provenance p join sources s on s.id = p.source_id where p.entity_type = 'facility' and p.entity_id = f.id and p.is_current and s.kind in ('operator','government','filing','utility','cloud_provider','registry')))::int as primary_src
      from ${facilityView(sql)} f where ${scope.facility}`,
    sql<Row[]>`select count(*)::int as n, count(*) filter (where p.lat is not null)::int as with_loc from projects p where ${projectLive(sql)} and ${scope.project}`,
  ]);
  const r = fr[0] ?? {};
  const n = int(r.n);
  return {
    key, name, slug,
    facilities: n,
    capacityCoverage: share(int(r.with_mw), n),
    operatorCoverage: share(int(r.with_op), n),
    preciseLocationCoverage: share(int(r.precise), n),
    anyLocationCoverage: share(int(r.any_loc), n),
    statusCoverage: share(int(r.with_status), n),
    openingDateCoverage: share(int(r.with_open), n),
    multiSourceCoverage: share(int(r.multi), n),
    projects: int(pr[0]?.n),
    projectsWithLocation: int(pr[0]?.with_loc),
    connectivityCoverage: share(int(r.conn), n),
    primarySourceShare: share(int(r.primary_src), n),
  };
}

/** Aggregates over the containment-aware view grouped by a dimension (for lists / dashboards). */
export interface DimAgg { id: string; slug: string; name: string; countryIso2: string | null; facilities: number; operational: number; construction: number; planned: number; knownMw: number | null; constructionMw: number | null; plannedMw: number | null; withMw: number; operators: number; ai: number; hyperscale: number; coverage: number }

export function dimAggFromRow(r: Row): DimAgg {
  const n = int(r.facilities);
  return { id: reqStr(r.id), slug: reqStr(r.slug), name: reqStr(r.name), countryIso2: str(r.country_iso2), facilities: n, operational: int(r.operational), construction: int(r.construction), planned: int(r.planned), knownMw: round2(num(r.known_mw)), constructionMw: round2(num(r.construction_mw)), plannedMw: round2(num(r.planned_mw)), withMw: int(r.with_mw), operators: int(r.operators), ai: int(r.ai), hyperscale: int(r.hyperscale), coverage: share(int(r.with_mw), n) };
}

/** Column list for a containment-aware aggregate over facilityView `f` (pair with a group by on the dimension). */
export function dimAggCols(sql: Sql): Fragment {
  const known = knownMwAgg(sql), pipe = pipelineMwAgg(sql), counted = countedAgg(sql), hasMw = hasMwAgg(sql);
  return sql`
    count(f.id) filter (where ${counted})::int as facilities,
    count(f.id) filter (where ${counted} and f.status = any(${OPERATIONAL_SET}))::int as operational,
    count(f.id) filter (where ${counted} and f.status = any(${CONSTRUCTION_SET}))::int as construction,
    count(f.id) filter (where ${counted} and f.status = any(${PLANNED_SET}))::int as planned,
    sum(${known}) filter (where f.status = any(${OPERATIONAL_SET}))::float as known_mw,
    sum(${pipe}) filter (where f.status = any(${CONSTRUCTION_SET}))::float as construction_mw,
    sum(${pipe}) filter (where f.status = any(${PLANNED_SET}))::float as planned_mw,
    count(f.id) filter (where ${counted} and ${hasMw})::int as with_mw,
    count(distinct f.operator_id)::int as operators,
    count(f.id) filter (where ${counted} and (f.ai_evidence in ('confirmed', 'likely') or f.is_ai or f.facility_type = 'ai'))::int as ai,
    count(f.id) filter (where ${counted} and (f.is_hyperscale or f.facility_type = 'hyperscale'))::int as hyperscale`;
}