#!/usr/bin/env python3
"""V-13: schema.sql must stand up a working database, and match the real one.

WHY THIS EXISTS
---------------
The manuscript tells a reader the repository ships the database schema. Until
2026-08-12 it shipped no schema file at all, and the tables were created by DDL
scattered through the loaders. schema.sql now exists — generated by pg_dump from
the database of record — which turns the sentence from false into a claim that
has to keep being true.

A dumped schema is easy to check and easy to leave stale. This applies it to a
throwaway schema, compares what it produced against the live one object by
object, and drops it again. Anything less proves only that the file parses.

The comparison is the point. Applying it cleanly says the SQL is valid; it does
not say the file still describes the database. A dump taken before a migration
lands is valid SQL describing the wrong thing, which is the failure this gate is
actually for.

Usage:
  ARCTIC_DATABASE_URL=... .venv/bin/python ingest/verify_schema_applies.py
"""
from __future__ import annotations

import os
import re
import sys
from pathlib import Path

import psycopg2

ROOT = Path(__file__).parent.parent
SCHEMA_FILE = ROOT / "schema.sql"
LIVE = "arctic_v2"
PROBE = "arctic_schema_probe"


def inventory(cur, schema: str) -> dict:
    """Relations, their columns, and the constraint and index counts per relation.

    information_schema.columns covers views as well as tables, and both belong in
    the comparison — a view that stopped existing would be a real divergence. The
    two are counted separately when reporting, because calling 26 relations "26
    tables" is the kind of small untruth that stops a number being read.
    """
    cur.execute(
        """select table_name, column_name, data_type, is_nullable
           from information_schema.columns where table_schema = %s
           order by table_name, ordinal_position""", (schema,))
    cols: dict[str, list] = {}
    for tbl, col, typ, null in cur.fetchall():
        cols.setdefault(tbl, []).append((col, typ, null))

    cur.execute(
        """select t.relname, count(*) from pg_constraint c
           join pg_class t on t.oid = c.conrelid
           join pg_namespace n on n.oid = t.relnamespace
           where n.nspname = %s group by 1""", (schema,))
    cons = dict(cur.fetchall())

    cur.execute(
        "select tablename, count(*) from pg_indexes where schemaname = %s group by 1",
        (schema,))
    idx = dict(cur.fetchall())
    return {"columns": cols, "constraints": cons, "indexes": idx}


def main() -> int:
    url = os.environ.get("ARCTIC_DATABASE_URL")
    if not url:
        print("ARCTIC_DATABASE_URL not set (source from the operator vault, never hardcode)",
              file=sys.stderr)
        return 1
    if not SCHEMA_FILE.exists():
        print(f"V-13 FAIL: {SCHEMA_FILE} does not exist. Run 'make schema'.")
        return 1

    sql = SCHEMA_FILE.read_text()
    # Guard the guard. If the dump did not actually name the live schema, the
    # rewrite below is a no-op and the probe would be built from nothing —
    # producing an empty schema that trivially "applies". Silence must not pass.
    if sql.count(LIVE) < 10:
        print(f"V-13 FAIL: schema.sql mentions {LIVE} fewer than 10 times "
              f"({sql.count(LIVE)}). It does not look like a dump of that schema, "
              f"and rewriting it would produce an empty probe that passes for free.")
        return 1
    probe_sql = re.sub(rf"\b{LIVE}\b", PROBE, sql)

    conn = psycopg2.connect(url, connect_timeout=60)
    conn.autocommit = True
    failures: list[str] = []
    try:
        with conn.cursor() as cur:
            cur.execute(f"drop schema if exists {PROBE} cascade")
            try:
                cur.execute(probe_sql)
            except Exception as exc:  # noqa: BLE001
                print(f"V-13 FAIL: schema.sql does not apply cleanly.\n  {exc}")
                return 1

            probe = inventory(cur, PROBE)
            live = inventory(cur, LIVE)

            if not probe["columns"]:
                print("V-13 FAIL: applying schema.sql produced no tables at all.")
                return 1

            missing = sorted(set(live["columns"]) - set(probe["columns"]))
            extra = sorted(set(probe["columns"]) - set(live["columns"]))
            if missing:
                failures.append(f"tables in the database but not in schema.sql: {missing}")
            if extra:
                failures.append(f"tables in schema.sql but not in the database: {extra}")

            for tbl in sorted(set(live["columns"]) & set(probe["columns"])):
                if live["columns"][tbl] != probe["columns"][tbl]:
                    lc = {c[0] for c in live["columns"][tbl]}
                    pc = {c[0] for c in probe["columns"][tbl]}
                    detail = []
                    if lc - pc:
                        detail.append(f"absent from schema.sql: {sorted(lc - pc)}")
                    if pc - lc:
                        detail.append(f"absent from the database: {sorted(pc - lc)}")
                    if not detail:
                        detail.append("same column names, different types or nullability")
                    failures.append(f"{tbl}: " + "; ".join(detail))
                for key in ("constraints", "indexes"):
                    if live[key].get(tbl, 0) != probe[key].get(tbl, 0):
                        failures.append(
                            f"{tbl}: {live[key].get(tbl, 0)} {key} live, "
                            f"{probe[key].get(tbl, 0)} from schema.sql")

            cur.execute(
                """select table_type, count(*) from information_schema.tables
                   where table_schema = %s group by 1""", (PROBE,))
            kinds = dict(cur.fetchall())
            n_tables = kinds.get("BASE TABLE", 0)
            n_views = kinds.get("VIEW", 0)
            n_cons = sum(probe["constraints"].values())
            n_idx = sum(probe["indexes"].values())
        with conn.cursor() as cur:
            cur.execute(f"drop schema if exists {PROBE} cascade")
    finally:
        conn.close()

    if failures:
        print(f"V-13 FAIL: schema.sql applies, but does not describe {LIVE}:")
        for f in failures[:20]:
            print(f"  {f}")
        if len(failures) > 20:
            print(f"  ... and {len(failures) - 20} more")
        print("\nRegenerate it with 'make schema'.")
        return 1

    print(f"V-13 PASS: schema.sql stands up {n_tables} tables, {n_views} view(s), "
          f"{n_cons} constraints and {n_idx} indexes from nothing, and matches "
          f"{LIVE} object for object.")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())