brain-of-reese/tests/integration/conftest.py

"""Session-level DB self-heal for the integration suite (phase 106, task 02).

Incident 2026-09-13: the dev DB's ``documents`` table hit PostgreSQL's
1600-attribute hard limit and every ``ALTER TABLE … ADD COLUMN`` failed
with ``TooManyColumns``, red-lining the full suite. Cause: the house
migration-test pattern (A13 — every migration exercises a real
downgrade, then repairs back to head) leaks *dropped-column
placeholder* attributes on every downgrade→upgrade round-trip
(``pg_attribute`` rows with ``attisdropped=true``). ``VACUUM (FULL)``
does NOT reclaim them (verified on PG 17.11) — only a table rewrite
does — and at ~95 leaks per full-suite run the shared dev DB bricks
every ~17 runs (faster when two runs race, which is how the incident
triggered).

This session fixture rebuilds any ``public`` table whose dropped-
attribute count exceeds :data:`DROPPED_ATTR_LIMIT` **before** the first
integration test of the session runs: rename + ``CREATE TABLE
(LIKE … INCLUDING ALL)`` + row copy + FK rewiring (both directions,
original constraint names and actions preserved). The normal case costs
one small catalog query per session; the heal path only fires while a
table is far below the 1600 cap (the limit is 200 — a tenth of the
headroom), so the migration tests never run on a near-bricked DB.
"""
from __future__ import annotations

import logging
import uuid
from collections.abc import Iterator

import pytest
from sqlalchemy import text

from app.db import db_available, engine

logger = logging.getLogger("bor.integration.self_heal")

#: Rebuild a table once its *dropped* attribute count passes this.
#: PostgreSQL's hard cap is 1600 TOTAL attributes (dropped included),
#: and the migration round-trips leak ~95 per full-suite run.
DROPPED_ATTR_LIMIT = 200

_BLOATED_SQL = text(
    "SELECT c.relname FROM pg_class c"
    " JOIN pg_namespace n ON n.oid = c.relnamespace"
    " WHERE c.relkind = 'r' AND n.nspname = 'public'"
    " AND (SELECT count(*) FROM pg_attribute a"
    "      WHERE a.attrelid = c.oid AND a.attnum > 0 AND a.attisdropped)"
    "       > :limit"
    " ORDER BY 1"
)

_FKS_SQL = text(
    "SELECT c.conname,"
    "       c.conrelid::regclass::text AS child,"
    "       c.confrelid::regclass::text AS parent,"
    "       c.confdeltype, c.confupdtype, c.confmatchtype AS matchtype,"
    "       (SELECT string_agg(ca.attname, ', ' ORDER BY ck.ord)"
    "        FROM unnest(c.conkey) WITH ORDINALITY ck(attnum, ord)"
    "        JOIN pg_attribute ca ON ca.attrelid = c.conrelid AND ca.attnum = ck.attnum)"
    "         AS child_cols,"
    "       (SELECT string_agg(pa.attname, ', ' ORDER BY pk.ord)"
    "        FROM unnest(c.confkey) WITH ORDINALITY pk(attnum, ord)"
    "        JOIN pg_attribute pa ON pa.attrelid = c.confrelid AND pa.attnum = pk.attnum)"
    "         AS parent_cols"
    " FROM pg_constraint c"
    " WHERE c.contype = 'f'"
    "   AND (:t = c.conrelid::regclass::text OR :t = c.confrelid::regclass::text)"
)

#: pg_constraint confdeltype/confupdtype codes → the DDL clause (``None``
#: = NO ACTION, the default — the clause is omitted).
_FK_ACTION: dict[str, str | None] = {
    "a": None,  # NO ACTION — the default, the clause is omitted
    "r": "RESTRICT",
    "c": "CASCADE",
    "n": "SET NULL",
    "d": "SET DEFAULT",
}
_FK_MATCH: dict[str, str] = {"f": "MATCH FULL", "p": "MATCH PARTIAL"}


def _fk_clause(fk) -> str:
    """The trailing ``MATCH …/ON DELETE …/ON UPDATE …`` of an FK."""
    parts = [
        _FK_MATCH.get(fk.matchtype, ""),
        f"ON DELETE {_FK_ACTION[fk.confdeltype]}" if _FK_ACTION[fk.confdeltype] else "",
        f"ON UPDATE {_FK_ACTION[fk.confupdtype]}" if _FK_ACTION[fk.confupdtype] else "",
    ]
    return " ".join(p for p in parts if p)


def _rebuild_table(table: str) -> None:
    """Rewrite *table* to purge its dropped-column placeholders.

    Rename + ``LIKE … INCLUDING ALL`` (live columns, constraints,
    indexes, defaults) + row copy + FK rewiring (incoming AND outgoing,
    original constraint names/actions). One transaction — a failure
    rolls the whole table's surgery back and fails the session loudly
    (a half-healed DB must never feed the migration tests).

    The staging names carry a per-run suffix: a previous (interrupted or
    repeated) heal may still own the plain names, and a collision would
    make PG auto-suffix the LIKE-copied constraint names (…``_pkey1``)
    and defeat the PK rename below.

    Note: the PK is renamed back to its conventional ``<table>_pkey``;
    other LIKE-copied objects keep PG's auto-generated names (nothing in
    the repo references constraint/index names by name — DDL is
    alembic-only, the ORM never issues DDL).
    """
    new_name = f"{table}_heal_new_{uuid.uuid4().hex[:8]}"
    old_name = f"{table}_heal_old_{uuid.uuid4().hex[:8]}"
    with engine.begin() as conn:
        fks = conn.execute(_FKS_SQL, {"t": table}).fetchall()
        for fk in fks:
            conn.execute(
                text(f'ALTER TABLE "{fk.child}" DROP CONSTRAINT "{fk.conname}"')
            )
        conn.execute(
            text(f'CREATE TABLE "{new_name}" (LIKE "{table}" INCLUDING ALL)')
        )
        # Explicit non-generated column list (attnum order): ``SELECT *``
        # cannot be used — chunks.tsv is a STORED generated column, and
        # generated columns refuse explicit values (it recomputes them).
        cols = conn.execute(
            text(
                "SELECT string_agg('\"' || a.attname || '\"', ', '"
                " ORDER BY a.attnum)"
                " FROM pg_attribute a JOIN pg_class tc ON tc.oid = a.attrelid"
                " WHERE tc.relname = :t AND a.attnum > 0"
                "   AND NOT a.attisdropped AND a.attgenerated NOT IN ('s', 'v')"
            ),
            {"t": table},
        ).scalar()
        conn.execute(
            text(f'INSERT INTO "{new_name}" ({cols}) SELECT {cols} FROM "{table}"')
        )
        conn.execute(text(f'ALTER TABLE "{table}" RENAME TO "{old_name}"'))
        # Drop the old table BEFORE the staging table takes its name: the
        # old table's index/constraint names (including ``<table>_pkey``
        # from a previous heal) live in the schema namespace until the
        # DROP, and the PK rename below needs that name free.
        conn.execute(text(f'DROP TABLE "{old_name}"'))
        conn.execute(text(f'ALTER TABLE "{new_name}" RENAME TO "{table}"'))
        # The rename above does NOT follow to LIKE-copied objects: put the
        # PK back on its conventional name (every migration here auto-names
        # PKs ``<table>_pkey`` — the one name tools/scripts reference).
        has_auto_pkey = conn.execute(
            text(
                "SELECT 1 FROM pg_constraint c"
                " JOIN pg_class tc ON tc.oid = c.conrelid"
                " WHERE tc.relname = :t AND c.conname = :n AND c.contype = 'p'"
            ),
            {"t": table, "n": f"{new_name}_pkey"},
        ).fetchone()
        if has_auto_pkey:
            conn.execute(
                text(
                    f'ALTER TABLE "{table}" RENAME CONSTRAINT'
                    f' "{new_name}_pkey" TO "{table}_pkey"'
                )
            )
        # FK rewiring LAST: only now does ``<table>`` refer to the rebuilt
        # table with all staging names gone.
        for fk in fks:
            clause = _fk_clause(fk)
            conn.execute(
                text(
                    f'ALTER TABLE "{fk.child}" ADD CONSTRAINT "{fk.conname}"'
                    f" FOREIGN KEY ({fk.child_cols})"
                    f' REFERENCES "{fk.parent}" ({fk.parent_cols})'
                    + (f" {clause}" if clause else "")
                )
            )


@pytest.fixture(autouse=True, scope="session")
def heal_bloated_tables() -> Iterator[None]:
    """Rebuild bloated ``public`` tables before the session's first test.

    One cheap catalog query per session while healthy (the normal
    case); the rebuild path only fires when a table's dropped-attribute
    count passes :data:`DROPPED_ATTR_LIMIT` (see the module docstring
    for the 2026-09-13 incident this exists to outlive).
    """
    if db_available():
        with engine.connect() as conn:
            bloated = [
                row[0]
                for row in conn.execute(_BLOATED_SQL, {"limit": DROPPED_ATTR_LIMIT})
            ]
        for table in bloated:
            logger.warning(
                "integration self-heal: rebuilding %r (dropped attributes"
                " > %d — migration round-trip placeholders)",
                table,
                DROPPED_ATTR_LIMIT,
            )
            _rebuild_table(table)
    yield