import type { Dashboard, DashboardStats, GridConstraintDTO, MarketMomentum, ProjectSummary, RankingRow } from "@dci/core";
import { pg, yearExpr, projectJoins, projectSummaryCols, projectLive, facilityView, knownMwAgg, countedAgg, eventCols, eventJoins, gridConstraintCols, gridConstraintJoins, AI_LEVELS, POWER_EVENT_TYPES, GRID_EVENT_TYPES } from "../lib/sql.js";
import { int, iso, num, reqStr, str, type Row } from "../lib/rows.js";
import { asStatus, asType, gridConstraintDto, gridConstraintFromEvent, projectSummary, round2, share } from "../lib/dto.js";
import { hrefFor } from "../lib/resolve.js";
import { latestEvents, toEventDtos } from "./events.js";
import { recentlyVerifiedFacilities } from "./facilities.js";
import { topRows } from "./rankings.js";
import { pulse } from "./pulse.js";
import { coverageRow, dimAggCols, momentum, scopeFor } from "../lib/pipeline.js";

async function liveTop(dim: "country" | "metro" | "operator", limit: number): Promise<RankingRow[]> {
  const sql = pg();
  const known = knownMwAgg(sql), counted = countedAgg(sql), hasMw = sql`(coalesce(f.it_capacity_mw, f.total_power_mw, f.planned_power_mw) is not null or f.covered_by_parent)`;
  const rows = dim === "country"
    ? await sql<Row[]>`select c.iso2 as id, c.slug, c.name, c.iso2 as country_iso2, count(*) filter (where ${counted})::int as n, sum(${known}) filter (where f.status in ('operational','partially_operational','expansion'))::float as mw, count(*) filter (where ${counted} and ${hasMw})::int as with_mw from ${facilityView(sql)} f join countries c on c.iso2 = f.country_iso2 group by 1, 2, 3, 4 order by n desc limit ${limit}`
    : dim === "metro"
      ? await sql<Row[]>`select m.id, m.slug, m.name, m.country_iso2, count(*) filter (where ${counted})::int as n, sum(${known}) filter (where f.status in ('operational','partially_operational','expansion'))::float as mw, count(*) filter (where ${counted} and ${hasMw})::int as with_mw from ${facilityView(sql)} f join metros m on m.id = f.metro_id group by 1, 2, 3, 4 order by n desc limit ${limit}`
      : await sql<Row[]>`select o.id, o.slug, o.name, o.hq_country_iso2 as country_iso2, count(*) filter (where ${counted})::int as n, sum(${known}) filter (where f.status in ('operational','partially_operational','expansion'))::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 group by 1, 2, 3, 4 order by n desc limit ${limit}`;
  return rows.map((r, i) => ({ rank: i + 1, id: reqStr(r.id), slug: reqStr(r.slug), name: reqStr(r.name), href: hrefFor(dim, reqStr(r.slug)), value: int(r.n), secondary: round2(num(r.mw)), countryIso2: str(r.country_iso2), coverage: share(int(r.with_mw), int(r.n)) }));
}

async function fastestGrowingMetros(limit = 8): Promise<Dashboard["fastestGrowingMetros"]> {
  const sql = pg();
  const since = sql`(current_date - interval '12 months')`;
  const pd = (col: string) => sql`(case when p.${sql(col)} ~ '^\\d{4}-\\d{2}-\\d{2}' then p.${sql(col)}::date when p.${sql(col)} ~ '^\\d{4}-\\d{2}$' then (p.${sql(col)} || '-01')::date when p.${sql(col)} ~ '^\\d{4}$' then (p.${sql(col)} || '-01-01')::date else null end)`;
  const rows = await sql<Row[]>`
    with pa as (select p.metro_id, count(*) filter (where coalesce(${pd("announced_on")}, p.created_at::date) >= ${since})::int as announced, count(*) filter (where ${pd("construction_started_on")} >= ${since})::int as construction from projects p where ${projectLive(sql)} and p.metro_id is not null group by 1),
    fo as (select f.metro_id, count(*)::int as opened from facilities f where f.merged_into is null and f.metro_id is not null and f.status in ('operational','partially_operational','expansion') and (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) >= ${since} group by 1)
    select m.id, m.slug, m.name, m.country_iso2, coalesce(pa.announced, 0) + coalesce(pa.construction, 0) + coalesce(fo.opened, 0) as score
    from metros m left join pa on pa.metro_id = m.id left join fo on fo.metro_id = m.id
    where coalesce(pa.announced, 0) + coalesce(pa.construction, 0) + coalesce(fo.opened, 0) > 0
    order by score desc, m.name limit ${limit}`;
  return Promise.all(rows.map(async (r) => ({ id: reqStr(r.id), slug: reqStr(r.slug), name: reqStr(r.name), countryIso2: reqStr(r.country_iso2), momentum: (await momentum(scopeFor(sql, { metroId: reqStr(r.id) }))) as MarketMomentum })));
}

async function operatorExpansion(limit = 10): Promise<Dashboard["operatorExpansion"]> {
  const sql = pg();
  const since = sql`(current_date - interval '12 months')`;
  const rows = await sql<Row[]>`
    with entries as (
      select f.operator_id, f.metro_id, f.country_iso2, coalesce(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, f.first_seen::date) as d
      from facilities f where f.merged_into is null and f.operator_id is not null
      union all
      select p.operator_id, p.metro_id, p.country_iso2, 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
      from projects p where ${projectLive(sql)} and p.operator_id is not null
    ),
    fc as (select operator_id, country_iso2, min(d) as first_d from entries where country_iso2 is not null group by 1, 2),
    fm as (select operator_id, metro_id, min(d) as first_d from entries where metro_id is not null group by 1, 2),
    agg as (
      select o.id, o.slug, o.name,
        (select array_agg(fc.country_iso2 order by fc.first_d desc) from fc where fc.operator_id = o.id and fc.first_d >= ${since}) as new_countries,
        (select array_agg(m.name order by fm.first_d desc) from fm join metros m on m.id = fm.metro_id where fm.operator_id = o.id and fm.first_d >= ${since}) as new_metros,
        (select count(*)::int from projects p where ${projectLive(sql)} and p.operator_id = o.id and coalesce(case when p.announced_on ~ '^\\d{4}' then (left(p.announced_on, 4) || '-01-01')::date else null end, p.created_at::date) >= ${since}) as projects12m,
        (select count(*) from fc where fc.operator_id = o.id and fc.first_d < ${since}) as had_before
      from operators o
    )
    select * from agg where (coalesce(array_length(new_countries, 1), 0) > 0 or coalesce(array_length(new_metros, 1), 0) > 0) and had_before > 0
    order by coalesce(array_length(new_countries, 1), 0) + coalesce(array_length(new_metros, 1), 0) desc, projects12m desc, name limit ${limit}`;
  return rows.map((r) => ({ operator: { id: reqStr(r.id), slug: reqStr(r.slug), name: reqStr(r.name) }, newCountries: ((r.new_countries as string[] | null) ?? []).map(String), newMetros: ((r.new_metros as string[] | null) ?? []).map(String), projects12m: int(r.projects12m) }));
}

async function globalGridConstraints(limit = 10): Promise<GridConstraintDTO[]> {
  const sql = pg();
  const [rows, ev] = await Promise.all([
    sql<Row[]>`select ${gridConstraintCols(sql)} from grid_constraints g ${gridConstraintJoins(sql)} order by g.created_at desc limit ${limit}`,
    sql<Row[]>`select ${eventCols(sql)} from events e ${eventJoins(sql)} where e.event_type = any(${GRID_EVENT_TYPES}) and e.review_status <> 'rejected' order by e.detected_at desc limit ${limit}`,
  ]);
  const out = rows.map(gridConstraintDto);
  const seen = new Set(out.map((g) => g.eventId).filter(Boolean));
  for (const e of await toEventDtos(ev)) { if (seen.has(e.id)) continue; seen.add(e.id); out.push(gridConstraintFromEvent(e)); }
  return out.slice(0, limit);
}

export async function dashboard(): Promise<Dashboard> {
  const sql = pg();
  const known = knownMwAgg(sql);
  const [facStats, countsRow, evRow, crawlRow, capSeries, capFallback, projStatus, typeRows, pipeline, aiRow, aiRecent, latest, newProjects, recentlyVerified, rkCountries, rkMetros, rkOperators, pulseData, growing, major, expansion, powerRows, gridConstraints, coverage, ingRow] = await Promise.all([
    sql<Row[]>`select ${dimAggCols(sql)}, count(distinct f.country_iso2)::int as countries, count(distinct f.metro_id)::int as metros from ${facilityView(sql)} f`,
    sql<Row[]>`
      select (select count(*)::int from cloud_regions where status <> 'retired') as cloud_regions, (select count(*)::int from ixps) as ixps, (select count(*)::int from projects p where ${projectLive(sql)}) as projects,
        (select count(*)::int from sources) as sources, (select count(*)::int from documents) as documents, (select count(*)::int from operators) as all_operators`,
    sql<Row[]>`select count(*) filter (where detected_at >= now() - interval '24 hours')::int as e24, count(*) filter (where detected_at >= now() - interval '7 days')::int as e7 from events where review_status <> 'rejected'`,
    sql<Row[]>`select max(finished_at) as last_finished, max(started_at) as last_started, count(*) filter (where started_at >= now() - interval '24 hours')::int as runs24 from connector_runs`,
    sql<Row[]>`select day, metric, value from daily_metrics where dim = 'global' and metric in ('facilities_total', 'known_mw') order by day`,
    sql<Row[]>`select ${yearExpr(sql, sql`f.opened_on`)} as year, count(*) filter (where ${countedAgg(sql)})::int as n, sum(${known})::float as mw from ${facilityView(sql)} f where f.opened_on ~ '^\\d{4}' group by 1 order by 1`,
    sql<Row[]>`select p.status, count(*)::int as n, sum(p.planned_mw)::float as mw from projects p where ${projectLive(sql)} group by p.status order by n desc`,
    sql<Row[]>`select facility_type, count(*)::int as n from facilities where merged_into is null group by facility_type order by n desc`,
    sql<Row[]>`select left(p.expected_opening, 4) as year, count(*)::int as n, sum(p.planned_mw)::float as mw from projects p where ${projectLive(sql)} and p.expected_opening ~ '^\\d{4}' and p.status not in ('cancelled', 'closed') group by 1 order by 1`,
    sql<Row[]>`select (select count(*)::int from ${facilityView(sql)} f where ${countedAgg(sql)} and (f.ai_evidence = any(${AI_LEVELS}) or f.is_ai or f.facility_type = 'ai')) as facilities, (select count(*)::int from projects p where ${projectLive(sql)} and (p.ai_evidence = any(${AI_LEVELS}) or p.is_ai)) as projects, (select sum(p.planned_mw)::float from projects p where ${projectLive(sql)} and (p.ai_evidence = any(${AI_LEVELS}) or p.is_ai) and p.status not in ('cancelled')) as planned_mw`,
    sql<Row[]>`select ${projectSummaryCols(sql)} from projects p ${projectJoins(sql)} where ${projectLive(sql)} and (p.ai_evidence = any(${AI_LEVELS}) or p.is_ai) order by p.last_update desc limit 6`,
    latestEvents(20),
    sql<Row[]>`select ${projectSummaryCols(sql)} from projects p ${projectJoins(sql)} where ${projectLive(sql)} order by p.created_at desc limit 10`,
    recentlyVerifiedFacilities(10),
    topRows(["countries_by_facilities", "countries_by_known_mw", "countries_facilities", "top_countries"], "countries", 10),
    topRows(["metros_by_facilities", "metros_by_known_mw", "metros_facilities", "top_metros"], "metros", 10),
    topRows(["operators_by_facilities", "operators_by_known_mw", "operators_facilities", "top_operators"], "operators", 10),
    pulse("24h"),
    fastestGrowingMetros(8),
    sql<Row[]>`select ${projectSummaryCols(sql)} from projects p ${projectJoins(sql)} where ${projectLive(sql)} and p.planned_mw >= 100 and p.status not in ('cancelled', 'closed') order by p.planned_mw desc, p.last_update desc limit 12`,
    operatorExpansion(10),
    sql<Row[]>`select ${eventCols(sql)} from events e ${eventJoins(sql)} where e.event_type = any(${POWER_EVENT_TYPES}) and e.review_status <> 'rejected' order by e.detected_at desc limit 10`,
    globalGridConstraints(10),
    coverageRow("global", "Global", "global", scopeFor(sql, {})),
    sql<Row[]>`select (select count(*)::int from documents where last_fetched >= now() - interval '24 hours') as docs24, (select count(*)::int from connectors where enabled and not paused) as total, (select count(*)::int from connectors where enabled and not paused and health = 'ok') as healthy`,
  ]);
  const fs = facStats[0] ?? {};
  const cn = countsRow[0] ?? {};
  const facilities = int(fs.facilities);
  const stats: DashboardStats = {
    facilities,
    operational: int(fs.operational),
    underConstruction: int(fs.construction),
    planned: int(fs.planned),
    knownOperationalMw: round2(num(fs.known_mw)) ?? 0,
    constructionMw: round2(num(fs.construction_mw)) ?? 0,
    plannedMw: round2(num(fs.planned_mw)) ?? 0,
    mwCoverage: share(int(fs.with_mw), facilities),
    countries: int(fs.countries),
    metros: int(fs.metros),
    operators: int(fs.operators) || int(cn.all_operators),
    cloudRegions: int(cn.cloud_regions),
    ixps: int(cn.ixps),
    projects: int(cn.projects),
    events24h: int(evRow[0]?.e24),
    events7d: int(evRow[0]?.e7),
    sources: int(cn.sources),
    documents: int(cn.documents),
    lastCrawlAt: iso(crawlRow[0]?.last_finished) ?? iso(crawlRow[0]?.last_started),
    generatedAt: new Date().toISOString(),
  };

  // capacity over time: daily_metrics (global) by year-end, else facilities by opened_on year
  let capacityOverTime: Dashboard["capacityOverTime"] = [];
  if (capSeries.length) {
    const byYear = new Map<number, { facilities: number; knownMw: number | null }>();
    for (const r of capSeries) {
      const year = Number(String(r.day).slice(0, 4));
      if (!year) continue;
      const cur = byYear.get(year) ?? { facilities: 0, knownMw: null };
      if (r.metric === "facilities_total") cur.facilities = int(r.value);
      if (r.metric === "known_mw") cur.knownMw = num(r.value);
      byYear.set(year, cur); // last day of the year wins (rows ordered by day)
    }
    capacityOverTime = [...byYear.entries()].sort((a, b) => a[0] - b[0]).map(([year, v]) => ({ year, facilities: v.facilities, knownMw: v.knownMw, cumulativeMw: v.knownMw }));
  }
  if (!capacityOverTime.length) {
    let cumF = 0, cumMw = 0, saw = false;
    capacityOverTime = capFallback.map((r) => { cumF += int(r.n); const m = num(r.mw); if (m != null) { cumMw += m; saw = true; } return { year: int(r.year), facilities: cumF, knownMw: round2(m), cumulativeMw: saw ? Math.round(cumMw * 100) / 100 : null }; }).filter((x) => x.year > 0);
  }

  const ai = aiRow[0] ?? {};
  const ing = ingRow[0] ?? {};
  return {
    stats,
    capacityOverTime,
    projectsByStatus: projStatus.map((r) => ({ status: asStatus(r.status), count: int(r.n), mw: round2(num(r.mw)) })),
    topCountries: rkCountries ?? (await liveTop("country", 10)),
    topMetros: rkMetros ?? (await liveTop("metro", 10)),
    topOperators: rkOperators ?? (await liveTop("operator", 10)),
    typeBreakdown: typeRows.map((r) => ({ type: asType(r.facility_type), count: int(r.n) })),
    pipelineByYear: pipeline.map((r) => ({ year: reqStr(r.year), count: int(r.n), mw: round2(num(r.mw)) })),
    aiExpansion: { facilities: int(ai.facilities), projects: int(ai.projects), plannedMw: round2(num(ai.planned_mw)), recent: aiRecent.map(projectSummary) as ProjectSummary[] },
    latestEvents: latest,
    newProjects: newProjects.map(projectSummary),
    recentlyVerified,
    pulse: pulseData,
    fastestGrowingMetros: growing,
    majorProjects: major.map(projectSummary),
    operatorExpansion: expansion,
    powerEvents: await toEventDtos(powerRows),
    gridConstraints,
    coverage,
    ingestion: { lastRunAt: iso(crawlRow[0]?.last_started), runs24h: int(crawlRow[0]?.runs24), documents24h: int(ing.docs24), events24h: int(evRow[0]?.e24), connectorsHealthy: int(ing.healthy), connectorsTotal: int(ing.total) },
  };
}