// Load the SDP Facilities Master Plan extract → philly_sites,
// philly_site_programs, and site-derived school metrics.
//
// Source: the Facilities Data Dashboard behind philasd.org/fpp, pulled by
// `python -m pipeline_philly.fetch_facilities` into two COMMITTED CSVs:
//   data/cities/philly/reference/parsed/sdp_facilities_sites.csv
//   data/cities/philly/reference/parsed/sdp_facilities_buildings.csv
// The CSVs are the audit trail — the dashboard is an undocumented Qlik app that
// could change without notice, so this loader reads the committed extract and
// never the network.
//
// Design and evidence: docs/cities/philly/06_enrollment_capacity_utilization.md.
// Two grain rules are load-bearing, and each one is the opposite of what a
// first reading suggests:
//
//   * SITE, not school, is the grain capacity is published at. A site hosting a
//     district school plus its Pre-K programme (or a co-located charter)
//     publishes the SAME figures on every programme's row — 91 of 328 sites.
//     Counting rows double-weights them, so sites are deduped on `site_id`.
//   * School ↔ site is MANY-TO-MANY (32 schools sit on more than one site), so
//     the relation needs a join table, not a foreign key on either side. The
//     dashboard repeats a school's whole enrollment on every one of its sites,
//     so a school on several sites of its own counts its enrollment ONCE and
//     sums its capacity (schoolFacts in _facilities.ts; checks (v) and (vi)).
//
// Building rows (floor area, year built, condition, a second capacity figure)
// are not loaded. The analysis packet reads them from the committed CSV, and
// the gap register (§10) lists what else the extract carries that is not stored.
//
// Utilization here INCLUDES the PreK allotment (allotted seats) in the numerator;
// NYC's Blue Book counts enrolled 3-K/Pre-K students. The two cities' figures
// are on different bases and no common-basis figure is computed; wherever they
// appear together the numerator, the denominator and the Pre-K basis are named
// (doc §4.2).
//
// Only one vintage exists (SY2024-25); the dashboard's `School Year` field is a
// cross-join and carries no capacity information. The extract therefore carries
// no year of its own: EXTRACT_SHA256 pins the committed CSV pair to
// YEAR_COVERED, and the loader refuses any other pair.
//
// Run:  npx tsx scripts/loaders/philly/facilities.ts [--dry-run]

import * as crypto from "node:crypto";
import * as fs from "node:fs";
import * as path from "node:path";
import { parse } from "csv-parse/sync";
import { prisma, startDataLoad, finishDataLoad, readLatestProvenance } from "./_lib";
import {
  allotmentIdentity,
  dedupeSites,
  float,
  int,
  schoolFacts,
  schoolSiteAgreement,
  siteProgramLinks,
  utilizationAgreement,
  type SiteRow,
} from "./_facilities";

const SOURCE_ID = "sdp_facilities_dashboard";
const YEAR_COVERED = "2024-25";
/**
 * sha256 of the committed CSV pair (sites, then buildings) confirmed to describe
 * YEAR_COVERED — the hash the fetcher records in provenance.json. The fetcher
 * pulls no year field, so a re-fetch after SDP updates the dashboard would
 * otherwise load new figures under this year and overwrite it.
 */
const EXTRACT_SHA256 = "18b9791617edada6d3ec991595941284749afaa9c0e2a4264c9a2a9fbba51191";
const SCRIPT_PATH = "scripts/loaders/philly/facilities.ts";
const PARSED = path.join(process.cwd(), "data", "cities", "philly", "reference", "parsed");
const SITES_CSV = path.join(PARSED, "sdp_facilities_sites.csv");
const BUILDINGS_CSV = path.join(PARSED, "sdp_facilities_buildings.csv");

/** School-grain metric keys written for schools we can attribute a site to. */
const SCHOOL_METRICS = {
  capacity: "site_capacity",
  enrollment: "site_enrollment",
  prek: "prek_allotment",
  utilization: "utilization_pct",
} as const;

function readCsv<T>(file: string): T[] {
  if (!fs.existsSync(file)) {
    throw new Error(
      `${file} is missing. Run: python -m pipeline_philly.fetch_facilities ` +
        `(needs \`pip install websocket-client\`).`,
    );
  }
  return parse(fs.readFileSync(file, "utf-8"), { columns: true, skip_empty_lines: true }) as T[];
}

/**
 * schoolFacts plus the two guards on its roll-up: a school whose own sites
 * disagree on enrollment (the rule counts it once, which is only right while
 * the dashboard repeats it), and check (vi), which compares a school whose
 * capacity sits on one site against that site's published utilization.
 */
function checkedSchoolFacts(
  sites: ReadonlyMap<string, SiteRow>,
  links: ReadonlyArray<{ siteId: string; ulcsCode: string }>,
  label: string,
): ReturnType<typeof schoolFacts> {
  const result = schoolFacts(sites, links);
  if (result.conflicts.length) {
    throw new Error(
      `[philly-facilities] ${result.conflicts.length} schools carry different enrollments on sites of ` +
        `their own (${result.conflicts.slice(0, 10).join(", ")}). The roll-up counts a school's ` +
        `enrollment once, which is right only while the dashboard repeats it on every site, and ` +
        `this extract does not. Refusing to load.`,
    );
  }
  const agree = schoolSiteAgreement(sites, links, result.facts);
  console.log(
    `[philly-facilities] check (vi) school roll-up over ${label}: ${result.facts.length} schools, ` +
      `${result.shared.length} on a shared site, 0 conflicting; a school with one capacity site ` +
      `reproduces that site's published utilization ${agree.within}/${agree.checked}`,
  );
  if (agree.within < agree.checked) {
    throw new Error(
      `[philly-facilities] school roll-up check FAILED: ${agree.checked - agree.within} of ` +
        `${agree.checked} schools with one capacity site do not reproduce its published ` +
        `utilization (${agree.failing.slice(0, 10).join(", ")}). Refusing to load.`,
    );
  }
  return result;
}

/** True when the facilities tables exist, so the loader can no-op pre-migration. */
async function facilitiesTablesExist(): Promise<boolean> {
  const rows = await prisma.$queryRaw<Array<{ n: bigint }>>`
    SELECT count(*) AS n FROM information_schema.tables
    WHERE table_schema = 'public'
      AND table_name IN ('philly_sites', 'philly_site_programs')`;
  return Number(rows[0]?.n ?? 0) === 2;
}

async function main() {
  const dryRun = process.argv.includes("--dry-run");

  const siteRows = readCsv<SiteRow>(SITES_CSV);
  // Only the extract confirmed as YEAR_COVERED loads; see EXTRACT_SHA256.
  const prov = readLatestProvenance(SOURCE_ID);
  const extractSha = crypto
    .createHash("sha256")
    .update(fs.readFileSync(SITES_CSV))
    .update(fs.readFileSync(BUILDINGS_CSV))
    .digest("hex");
  if (extractSha !== EXTRACT_SHA256 || (prov && prov.sha256 !== EXTRACT_SHA256)) {
    throw new Error(
      `[philly-facilities] the facilities extract is not the one confirmed as SY${YEAR_COVERED}: the ` +
        `CSV pair hashes to ${extractSha} and the latest provenance records ${prov?.sha256 ?? "nothing"}, ` +
        `against ${EXTRACT_SHA256}. The dashboard publishes no year with its figures, so loading ` +
        `a new extract would overwrite ${YEAR_COVERED}. Confirm which school year the new extract ` +
        `describes, then update YEAR_COVERED and EXTRACT_SHA256 in ${SCRIPT_PATH} (and the year in ` +
        `UTILIZATION_DEFINITION in pipeline_philly/fetch_facilities.py). Refusing to load.`,
    );
  }
  const sites = dedupeSites(siteRows);

  const links = siteProgramLinks(siteRows);

  const multiProgramme = [...new Set(siteRows.map((r) => r.site_id))].filter(
    (id) => siteRows.filter((r) => r.site_id === id && (r.ulcs_code ?? "").trim()).length > 1,
  ).length;
  const noUlcs = siteRows.filter((r) => !(r.ulcs_code ?? "").trim()).length;
  const district = [...sites.values()].filter((r) => (r.sector ?? "").trim() === "District");
  const withCapacity = district.filter((r) => int(r.capacity) !== null).length;

  console.log(
    `[philly-facilities] extract: ${siteRows.length} rows → ${sites.size} distinct sites ` +
      `(${multiProgramme} host >1 programme, ${noUlcs} rows carry no ULCS and are dropped)\n` +
      `[philly-facilities] district sites ${district.length}, with a capacity ${withCapacity}`,
  );

  const agree = utilizationAgreement(sites.values());
  console.log(
    `[philly-facilities] check (iv) utilization == (enrollment + PreK) / capacity: ` +
      `${agree.within}/${agree.checked} within 0.005 (max |diff| ${agree.maxDiff.toFixed(4)})`,
  );
  const withFigures = [...sites.values()].filter(
    (r) => int(r.capacity) !== null || int(r.enroll_with_allotment) !== null || float(r.utilization) !== null,
  ).length;
  if (agree.checked === 0 && withFigures > 0) {
    throw new Error(
      `[philly-facilities] computed-values check FAILED: ${withFigures} sites carry a capacity, an ` +
        `enrollment with allotment or a utilization, but none carries all three, so nothing could be ` +
        `compared. A column may have been renamed or emptied; refusing to load.`,
    );
  }
  if (agree.checked > 0 && agree.within < agree.checked) {
    throw new Error(
      `[philly-facilities] computed-values check FAILED: ${agree.checked - agree.within} of ` +
        `${agree.checked} sites do not reconcile. The dashboard's utilization definition may ` +
        `have changed; refusing to load.`,
    );
  }

  const identity = allotmentIdentity(sites.values());
  console.log(
    `[philly-facilities] check (v) enroll w/ allotment == enrollment + PreK allotment: ` +
      `${identity.checked - identity.failing.length}/${identity.checked} sites`,
  );
  if (identity.failing.length) {
    throw new Error(
      `[philly-facilities] allotment identity FAILED on ${identity.failing.length} sites ` +
        `(${identity.failing.slice(0, 10).join(", ")}). The school roll-up adds enrollment and ` +
        `PreK allotment separately and relies on it; refusing to load.`,
    );
  }

  if (dryRun) {
    // A dry run never opens the database, so it cannot read the directory; it
    // checks the roll-up over every link in the extract instead. More schools
    // fall in the shared set there, because the links to programmes the
    // directory does not carry (Pre-K programmes on this extract) are kept,
    // and every school it does roll up faces the same two guards.
    checkedSchoolFacts(sites, links, "every extract link (before the directory filter)");
    console.log("[philly-facilities] --dry-run: nothing written");
    return;
  }
  if (!(await facilitiesTablesExist())) {
    console.log(
      "[philly-facilities] the philly_sites / philly_site_programs tables are " +
        "not in this database. Apply scripts/sql/capacity-schema.sql first. Nothing loaded.",
    );
    return;
  }

  // --- school-grain facts, ONLY where attribution is unambiguous (see _facilities.ts) ---
  // Computed before anything is written, so a refusal leaves the tables as they were.
  const known = new Set(
    (await prisma.phillySchool.findMany({ select: { ulcsCode: true } })).map((s) => s.ulcsCode),
  );
  const directoryLinks = links.filter((l) => known.has(l.ulcsCode));
  const { facts: schoolLevel, shared } = checkedSchoolFacts(sites, directoryLinks, "directory schools");

  const load = await startDataLoad({
    sourceId: SOURCE_ID,
    sourceUrl: prov?.url ?? "https://www.philasd.org/fpp/",
    sha256: prov?.sha256,
    bytes: prov?.bytes,
    yearCovered: YEAR_COVERED,
    scriptPath: SCRIPT_PATH,
    notes:
      `${sites.size} distinct sites (${multiProgramme} multi-programme), ` +
      `${links.length} site↔school links in the extract, ${directoryLinks.length} written (the other ` +
      `${links.length - directoryLinks.length} to programmes not in the directory); ` +
      `check (iv) ${agree.within}/${agree.checked} within 0.005; ` +
      `check (v) ${identity.checked - identity.failing.length}/${identity.checked}; ` +
      `${schoolLevel.length} schools rolled up, enrollment counted once`,
  });

  // --- sites ---
  for (const [siteId, r] of sites) {
    const data = {
      siteName: (r.site_name ?? "").trim(),
      capacity: int(r.capacity),
      prekAllotment: int(r.prek_allotment),
      enrollWithAllotment: int(r.enroll_with_allotment),
      // Stored as a percentage, matching every other utilization column in the
      // warehouse; the extract carries a 0-1 fraction.
      utilizationPct: float(r.utilization) === null ? null : float(r.utilization)! * 100,
      sourceLoadId: load.id,
    };
    await prisma.phillySite.upsert({ where: { siteId }, update: data, create: { siteId, ...data } });
  }

  // --- site ↔ school, skipping ULCS codes not in the directory ---
  const sectorOf = new Map<string, string>();
  for (const r of siteRows) {
    const key = `${(r.site_id ?? "").trim()}\u0000${(r.ulcs_code ?? "").trim()}`;
    if (!sectorOf.has(key)) sectorOf.set(key, (r.sector ?? "").trim() || "no sector");
  }
  const missing: string[] = [];
  const missingSectors = new Map<string, number>();
  let linked = 0;
  for (const { siteId, ulcsCode } of links) {
    if (!known.has(ulcsCode)) {
      missing.push(ulcsCode);
      const sector = sectorOf.get(`${siteId}\u0000${ulcsCode}`) ?? "no sector";
      missingSectors.set(sector, (missingSectors.get(sector) ?? 0) + 1);
      continue;
    }
    await prisma.phillySiteProgram.upsert({
      where: { siteId_ulcsCode: { siteId, ulcsCode } },
      update: {},
      create: { siteId, ulcsCode },
    });
    linked++;
  }
  if (missing.length) {
    const bySector = [...missingSectors]
      .sort((a, b) => b[1] - a[1])
      .map(([sector, n]) => `${sector} ${n}`)
      .join(", ");
    console.log(
      `[philly-facilities] ${missing.length} site↔school links skipped — ULCS not in philly_schools ` +
        `(the Master School List); by the extract's sector: ${bySector}. Codes: ` +
        `${[...new Set(missing)].slice(0, 10).join(", ")}${missing.length > 10 ? " …" : ""}`,
    );
  }

  // --- school-grain facts (rolled up above) ---
  const facts = schoolLevel.flatMap((f) => [
    { ulcs: f.ulcsCode, key: SCHOOL_METRICS.capacity, value: f.capacity, denominator: null as number | null },
    { ulcs: f.ulcsCode, key: SCHOOL_METRICS.enrollment, value: f.enrollment, denominator: null as number | null },
    { ulcs: f.ulcsCode, key: SCHOOL_METRICS.prek, value: f.prek, denominator: null as number | null },
    { ulcs: f.ulcsCode, key: SCHOOL_METRICS.utilization, value: f.utilizationPct, denominator: f.capacity as number | null },
  ]);
  for (const f of facts) {
    await prisma.phillySchoolYearMetric.upsert({
      where: {
        // populationCut is in the unique key with an "n/a" sentinel because
        // Prisma compound uniques cannot include a nullable column; the
        // acct/actual distinction does not apply to facilities metrics.
        schoolUlcs_year_metricKey_subgroup_populationCut: {
          schoolUlcs: f.ulcs, year: YEAR_COVERED, metricKey: f.key,
          subgroup: "ALL", populationCut: "n/a",
        },
      },
      update: { value: f.value, denominator: f.denominator, sourceLoadId: load.id },
      create: {
        schoolUlcs: f.ulcs, year: YEAR_COVERED, metricKey: f.key, subgroup: "ALL",
        populationCut: "n/a", value: f.value, denominator: f.denominator, sourceLoadId: load.id,
      },
    });
  }

  // --- remove what an earlier load wrote and this one does not produce ---
  // Upserts alone would leave a school's facts from an earlier load in place
  // after it stops qualifying (say, a directory change puts it on a shared
  // site), and the school page would still show them.
  const staleFacts = await prisma.phillySchoolYearMetric.deleteMany({
    where: {
      year: YEAR_COVERED,
      metricKey: { in: Object.values(SCHOOL_METRICS) },
      schoolUlcs: { notIn: schoolLevel.map((f) => f.ulcsCode) },
    },
  });
  const linkKeys = new Set(directoryLinks.map((l) => `${l.siteId}\u0000${l.ulcsCode}`));
  const staleLinks = (await prisma.phillySiteProgram.findMany({ select: { siteId: true, ulcsCode: true } }))
    .filter((l) => !linkKeys.has(`${l.siteId}\u0000${l.ulcsCode}`));
  if (staleLinks.length) {
    await prisma.phillySiteProgram.deleteMany({
      where: { OR: staleLinks.map(({ siteId, ulcsCode }) => ({ siteId, ulcsCode })) },
    });
  }
  if (staleFacts.count || staleLinks.length) {
    console.log(
      `[philly-facilities] removed ${staleFacts.count} school facts for ${YEAR_COVERED} and ` +
        `${staleLinks.length} site↔school links that this load does not produce`,
    );
  }

  await finishDataLoad(load.id, {
    inserted: sites.size + linked + facts.length,
    updated: 0,
    suppressed: 0,
  });
  console.log(
    `[philly-facilities] wrote ${sites.size} sites, ` +
      `${linked} site↔school links, ${facts.length} school facts\n` +
      `[philly-facilities] ${shared.length} schools left without school-grain facts — they share a site ` +
      `with another directory school, so the site's capacity is not theirs alone (site facts still ` +
      `carry it). A site shared only with programmes the directory does not carry (the skipped ` +
      `links above) counts as the host school's.`,
  );
}

main()
  .catch((e) => {
    console.error(e);
    process.exit(1);
  })
  .finally(() => {
    // The client is lazily constructed behind a Proxy, so touching it with no
    // DATABASE_URL throws synchronously. --dry-run never opens a connection.
    try {
      void prisma.$disconnect().catch(() => {});
    } catch {
      /* no connection was ever opened */
    }
  });