import type { EventDTO, OperatorDetail, OperatorSummary, SourceRef } from "@dci/core";
import { pg, andAll, page, likePattern, facilityView, knownMwAgg, countedAgg, facilityJoins, facilitySummaryCols, mwExpr, cloudRegionCols, eventCols, eventJoins, projectLive, OPERATIONAL_SET, AI_LEVELS, CORPORATE_EVENT_TYPES, type Fragment, type Sql } from "../lib/sql.js";
import { bool, int, num, record, reqIso, reqStr, str, strArray, type Row } from "../lib/rows.js";
import { cloudRegionSummary, facilitySummary, round2, share, statusBreakdown } from "../lib/dto.js";
import { findBySlugOrId } from "../lib/resolve.js";
import { listFacilities, provenanceFor } from "./facilities.js";
import { projectsWhere } from "./projects.js";
import { eventsForOperator, toEventDtos } from "./events.js";
import { buildSourceHistory, documentVersionsFor, sourceIdsOf, sourceRefsFor } from "../lib/source-history.js";
import { dimAggCols, expansionVelocity, pipelineBreakdown, scopeFor } from "../lib/pipeline.js";
import { claimsFor, dataQualityFor } from "../lib/quality.js";

export interface OperatorFilters { q?: string; kind?: string; country?: string; sort?: "facilities" | "name" | "mw"; order?: "asc" | "desc"; page?: number; per_page?: number }

/**
 * OperatorSummary from a row carrying `stats` (worker-refreshed, containment-aware) and optional live aggregate
 * columns (facility_count, known_mw, …). Live columns win when present (they reflect the current database); the
 * stats jsonb fills what the live query did not compute.
 */
export function operatorSummary(r: Row): OperatorSummary {
  const stats = (r.stats as Record<string, unknown> | null) ?? {};
  const live = r.facility_count != null;
  const fc = live ? int(r.facility_count) : int(stats.facilityCount);
  const withMw = live ? int(r.with_mw) : null;
  return {
    id: reqStr(r.id),
    slug: reqStr(r.slug),
    name: reqStr(r.name),
    kind: (str(r.kind) as OperatorSummary["kind"]) ?? null,
    website: str(r.website),
    hqCountryIso2: str(r.hq_country_iso2),
    facilityCount: fc,
    countryCount: live ? int(r.country_count) : int(stats.countryCount),
    metroCount: live ? int(r.metro_count) : int(stats.metroCount),
    knownMw: live ? round2(num(r.known_mw)) : round2(num(stats.knownMw)),
    plannedMw: live ? round2(num(r.planned_mw)) : round2(num(stats.plannedMw)),
    constructionMw: live ? round2(num(r.construction_mw)) : round2(num(stats.constructionMw)),
    projectCount: r.project_count != null ? int(r.project_count) : int(stats.projectCount),
    projectPlannedMw: r.project_planned_mw != null ? round2(num(r.project_planned_mw)) : round2(num(stats.projectPlannedMw)),
    aiCount: live ? int(r.ai) : int(stats.aiCount),
    cloudRegionCount: r.cloud_region_count != null ? int(r.cloud_region_count) : int(stats.cloudRegionCount),
    mwCoverage: withMw != null ? (fc ? share(withMw, fc) : 0) : num(stats.mwCoverage),
    isCloudProvider: bool(r.is_cloud_provider),
    isCarrier: bool(r.is_carrier),
  };
}

/** Containment-aware aggregate per operator over the facility view, optionally restricted to a country / metro. */
function aggFragment(sql: Sql, scope: { countryIso2?: string; metroId?: string } = {}): Fragment {
  const conds: Fragment[] = [sql`f.operator_id is not null`];
  if (scope.countryIso2) conds.push(sql`f.country_iso2 = ${scope.countryIso2}`);
  if (scope.metroId) conds.push(sql`f.metro_id = ${scope.metroId}`);
  return sql`(
    select f.operator_id, count(distinct f.country_iso2)::int as country_count, count(distinct f.metro_id)::int as metro_count, ${dimAggCols(sql)}
    from ${facilityView(sql)} f where ${andAll(sql, conds)} group by f.operator_id
  )`;
}

function projectAgg(sql: Sql): Fragment {
  return sql`(select p.operator_id, count(*)::int as n, sum(p.planned_mw) filter (where p.status in ('rumored','proposed','announced','permitting','approved','delayed','under_construction'))::float as mw from projects p where ${projectLive(sql)} group by p.operator_id)`;
}

const OPERATOR_COLS = (sql: Sql) => sql`
  o.id, o.slug, o.name, o.kind, o.website, o.hq_country_iso2, o.stats, o.is_cloud_provider, o.is_carrier,
  coalesce(a.facilities, 0) as facility_count, coalesce(a.country_count, 0) as country_count, coalesce(a.metro_count, 0) as metro_count,
  a.known_mw, a.planned_mw, a.construction_mw, coalesce(a.with_mw, 0) as with_mw, coalesce(a.ai, 0) as ai,
  coalesce(pc.n, 0) as project_count, pc.mw as project_planned_mw,
  (select count(*)::int from cloud_regions cr where cr.provider_id = o.id and cr.status <> 'retired') as cloud_region_count`;

export async function listOperators(f: OperatorFilters): Promise<{ items: OperatorSummary[]; total: number; page: number; perPage: number }> {
  const sql = pg();
  const pg_ = page(f.page, f.per_page, 100, 50);
  const conds: Fragment[] = [sql`true`];
  if (f.q) conds.push(sql`(o.name ilike ${likePattern(f.q)} or similarity(o.name, ${f.q}) > 0.3 or exists (select 1 from unnest(o.aliases) a where a ilike ${likePattern(f.q)}))`);
  if (f.kind) conds.push(sql`o.kind = ${f.kind}`);
  if (f.country) { const iso = f.country.toUpperCase(); conds.push(sql`(o.hq_country_iso2 = ${iso} or exists (select 1 from facilities x where x.operator_id = o.id and x.country_iso2 = ${iso} and x.merged_into is null))`); }
  const asc = f.order === "asc";
  const order = f.sort === "name" ? (asc ? sql`o.name asc` : sql`o.name desc`) : f.sort === "mw" ? (asc ? sql`a.known_mw asc nulls last, o.name` : sql`a.known_mw desc nulls last, o.name`) : asc ? sql`coalesce(a.facilities, 0) asc, o.name` : sql`coalesce(a.facilities, 0) desc, o.name`;
  const rows = await sql<Row[]>`
    select ${OPERATOR_COLS(sql)}, count(*) over() as total
    from operators o
    left join ${aggFragment(sql)} a on a.operator_id = o.id
    left join ${projectAgg(sql)} pc on pc.operator_id = o.id
    where ${andAll(sql, conds)}
    order by ${order}
    limit ${pg_.perPage} offset ${pg_.offset}`;
  return { items: rows.map(operatorSummary), total: rows.length ? int(rows[0]!.total) : 0, page: pg_.page, perPage: pg_.perPage };
}

/** Operators ranked by facilities inside a scope (country or metro) — containment-aware counts. */
export async function operatorsForScope(scope: { countryIso2?: string; metroId?: string }, limit = 10): Promise<OperatorSummary[]> {
  const sql = pg();
  const rows = await sql<Row[]>`
    select o.id, o.slug, o.name, o.kind, o.website, o.hq_country_iso2, '{}'::jsonb as stats, o.is_cloud_provider, o.is_carrier,
      a.facilities as facility_count, a.country_count, a.metro_count, a.known_mw, a.planned_mw, a.construction_mw, a.with_mw, a.ai, 0 as project_count, null::float as project_planned_mw, 0 as cloud_region_count
    from ${aggFragment(sql, scope)} a join operators o on o.id = a.operator_id
    where a.facilities > 0
    order by a.facilities desc, a.known_mw desc nulls last, o.name limit ${limit}`;
  return rows.map(operatorSummary);
}

/** OperatorSummary for one id (live aggregates). null when unknown. */
export async function operatorSummaryById(id: string): Promise<OperatorSummary | null> {
  const sql = pg();
  const rows = await sql<Row[]>`select ${OPERATOR_COLS(sql)} from operators o left join ${aggFragment(sql)} a on a.operator_id = o.id left join ${projectAgg(sql)} pc on pc.operator_id = o.id where o.id = ${id}`;
  return rows[0] ? operatorSummary(rows[0]) : null;
}

/** Corporate events (acquisitions, financing, partnerships, executive changes…) linked to an operator. */
export async function corporateEventsForOperator(operatorId: string, limit = 30): Promise<EventDTO[]> {
  const sql = pg();
  const rows = await sql<Row[]>`
    select ${eventCols(sql)} from events e ${eventJoins(sql)}
    where (e.operator_id = ${operatorId} or (e.entity_type = 'operator' and e.entity_id = ${operatorId})) and e.event_type = any(${CORPORATE_EVENT_TYPES}) and e.review_status <> 'rejected'
    order by e.detected_at desc limit ${limit}`;
  return toEventDtos(rows);
}

/** Metros first entered by the operator within the last `months` months (by earliest facility opened_on / first_seen or project announcement). */
export async function newMarkets(operatorId: string, months = 12): Promise<Array<{ id: string; slug: string; name: string }>> {
  const sql = pg();
  const rows = await sql<Row[]>`
    with fm as (
      select f.metro_id, min(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) as first_d
      from facilities f where f.merged_into is null and f.operator_id = ${operatorId} and f.metro_id is not null group by 1
      union all
      select p.metro_id, min(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 = ${operatorId} and p.metro_id is not null group by 1
    ), first as (select metro_id, min(first_d) as first_d from fm group by 1)
    select m.id, m.slug, m.name from first join metros m on m.id = first.metro_id where first.first_d >= current_date - make_interval(months => ${months}) order by first.first_d desc limit 25`;
  return rows.map((r) => ({ id: reqStr(r.id), slug: reqStr(r.slug), name: reqStr(r.name) }));
}

export async function getOperatorDetail(idOrSlug: string, fPage = 1): Promise<{ detail: OperatorDetail; sources: SourceRef[] } | null> {
  const sql = pg();
  const row = await findBySlugOrId("operators", idOrSlug);
  if (!row) return null;
  const id = String(row.id);
  const scope = scopeFor(sql, { operatorId: id });
  const known = knownMwAgg(sql), counted = countedAgg(sql);
  const [sumRows, countryRows, metroRows, statusRows, facilities, projects, recentEvents, provenance, versions, parentRows, pipeline, velocity, aiRows, cloudRows, corporateEvents, claims] = await Promise.all([
    sql<Row[]>`select ${OPERATOR_COLS(sql)} from operators o left join ${aggFragment(sql)} a on a.operator_id = o.id left join ${projectAgg(sql)} pc on pc.operator_id = o.id where o.id = ${id}`,
    sql<Row[]>`
      select f.country_iso2 as iso2, c.name, c.slug, count(*) filter (where ${counted})::int as n, sum(${known}) filter (where f.status = any(${OPERATIONAL_SET}))::float as known_mw
      from ${facilityView(sql)} f left join countries c on c.iso2 = f.country_iso2
      where f.operator_id = ${id} and f.country_iso2 is not null
      group by 1, 2, 3 order by n desc, c.name`,
    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 = any(${OPERATIONAL_SET}))::float as known_mw
      from ${facilityView(sql)} f join metros m on m.id = f.metro_id
      where f.operator_id = ${id} group by 1, 2, 3, 4 order by n desc, m.name limit 100`,
    sql<Row[]>`select status, count(*)::int as n from facilities where operator_id = ${id} and merged_into is null group by status`,
    listFacilities({ operatorId: id, page: fPage, per_page: 50, sort: "mw", order: "desc" }),
    projectsWhere(sql`${projectLive(sql)} and p.operator_id = ${id}`, 20),
    eventsForOperator(id, 20),
    provenanceFor("operator", id),
    documentVersionsFor("operator", id),
    row.parent_id ? sql<Row[]>`select id, slug, name from operators where id = ${String(row.parent_id)}` : Promise.resolve([] as Row[]),
    pipelineBreakdown(scope),
    expansionVelocity(scope),
    sql<Row[]>`select ${facilitySummaryCols(sql)} from facilities f ${facilityJoins(sql)} where f.merged_into is null and (f.operator_id = ${id} or f.owner_id = ${id}) and f.ai_evidence = any(${AI_LEVELS}) order by ${mwExpr(sql)} desc nulls last, f.name limit 20`,
    sql<Row[]>`select ${cloudRegionCols(sql)} from cloud_regions r join operators pr on pr.id = r.provider_id where r.provider_id = ${id} order by r.country_iso2, r.code limit 500`,
    corporateEventsForOperator(id, 30),
    claimsFor("operator", id),
  ]);
  const summary = operatorSummary(sumRows[0] ?? row);
  const completeness = Math.round(([row.website, row.hq_country_iso2, row.kind, row.description].filter((v) => v != null && v !== "").length / 4) * 100);
  const dataQuality = await dataQualityFor("operator", id, completeness, str(row.updated_at));
  const counted_total = summary.facilityCount || countryRows.reduce((a, c) => a + int(c.n), 0);
  const detail: OperatorDetail = {
    ...summary,
    pipeline,
    velocity,
    topCountries: countryRows.slice(0, 10).map((c) => ({ iso2: reqStr(c.iso2), name: reqStr(c.name, reqStr(c.iso2)), slug: reqStr(c.slug, reqStr(c.iso2).toLowerCase()), facilityCount: int(c.n), knownMw: round2(num(c.known_mw)), share: share(int(c.n), counted_total) })),
    topMetros: metroRows.slice(0, 10).map((m) => ({ id: reqStr(m.id), slug: reqStr(m.slug), name: reqStr(m.name), countryIso2: reqStr(m.country_iso2), facilityCount: int(m.n), knownMw: round2(num(m.known_mw)), share: share(int(m.n), counted_total) })),
    aiFacilities: aiRows.map(facilitySummary),
    cloudRegions: cloudRows.map(cloudRegionSummary),
    corporateEvents,
    claims,
    dataQuality,
    aliases: strArray(row.aliases),
    description: str(row.description),
    parent: parentRows[0] ? { id: String(parentRows[0].id), slug: String(parentRows[0].slug), name: String(parentRows[0].name) } : null,
    countries: countryRows.map((c) => ({ iso2: reqStr(c.iso2), name: reqStr(c.name, reqStr(c.iso2)), slug: reqStr(c.slug, reqStr(c.iso2).toLowerCase()), facilityCount: int(c.n), knownMw: round2(num(c.known_mw)) })),
    metros: metroRows.map((m) => ({ id: reqStr(m.id), slug: reqStr(m.slug), name: reqStr(m.name), countryIso2: reqStr(m.country_iso2), facilityCount: int(m.n) })),
    statusBreakdown: statusBreakdown(statusRows),
    facilities: facilities.items,
    projects,
    recentEvents,
    sourceHistory: buildSourceHistory(provenance, recentEvents, versions),
    externalIds: record(row.external_ids),
    updatedAt: reqIso(row.updated_at),
  };
  const sources = await sourceRefsFor(sourceIdsOf(provenance, recentEvents, versions, claims, corporateEvents));
  return { detail, sources };
}

/** Operator candidates for query interpretation (trigram). */
export async function operatorCandidates(q: string, limit = 5): Promise<Array<{ id: string; slug: string; name: string; similarity: number }>> {
  const sql = pg();
  // SQL does the cheap prefilter (trigram or substring); word-boundary verification happens here so short aliases
  // ("IO") never match inside unrelated words ("operatIOnal").
  const rows = await sql<Row[]>`
    select o.id, o.slug, o.name, o.aliases, similarity(o.name, ${q}) as sim
    from operators o
    where (o.name % ${q} and similarity(o.name, ${q}) >= 0.45) or position(lower(o.name) in lower(${q})) > 0
       or exists (select 1 from unnest(o.aliases) a where length(a) >= 3 and position(lower(a) in lower(${q})) > 0)
    order by sim desc, length(o.name) desc limit ${limit * 4}`;
  const esc = (x: string) => x.replace(/[.*+?^${}()|[\]\\]/g, "\\$&");
  const bounded = (needle: string) => new RegExp(`(^|[^\\p{L}\\p{N}])${esc(needle)}([^\\p{L}\\p{N}]|$)`, "iu").test(q);
  return rows
    .map((r) => {
      const name = reqStr(r.name);
      const aliases = Array.isArray(r.aliases) ? (r.aliases as string[]) : [];
      let similarity = num(r.sim) ?? 0;
      if (bounded(name)) similarity = Math.max(similarity, 0.95);
      else if (aliases.some((a) => a.length >= 3 && bounded(a))) similarity = Math.max(similarity, 0.9);
      return { id: reqStr(r.id), slug: reqStr(r.slug), name, similarity };
    })
    .filter((r) => r.similarity >= 0.45)
    .sort((a, b) => b.similarity - a.similarity || b.name.length - a.name.length)
    .slice(0, limit);
}