"""Integration: migration 0026 (documents.page_count) schema contract
(phase 148, task 01).

Drives the **real Alembic engine** against the live dev database
(``podman compose up -d db``), mirroring the house pattern of
``test_migration_0025.py`` (information_schema assertions on the state
the migration must leave). The tests target the 0025 → 0026 step
explicitly so later migrations cannot break the pins:

* upgrade 0025 → 0026 → ``page_count`` exists with the full contract —
  INTEGER, NULLABLE, NO server default (NULL = "not page-qualifying"
  for every pre-phase row and every fail-soft miss) — and is UNINDEXED
  (a display fact, never a query surface — the page route of task 02
  reads it from the ROW after the uuid4 id lookup, the ``file_hash``
  convention of 0025) while the 0025 ``documents`` schema survives —
  in particular ``content_hash`` stays NOT NULL and INDEXED (the
  retriever's dedupe/caching identity, untouched) and ``file_hash``
  keeps its 0025 contract;
* rows inserted while the DB is at 0025 backfill ``page_count`` to
  NULL — the pre-phase shape, no backfill, no default;
* the ORM contract agrees: a freshly inserted ``Document`` with no
  ``page_count`` reads ``page_count is None`` through a fresh session
  (the pre-phase shape), and an explicit count round-trips;
* downgrade to 0025 → the column is GONE (A13) while the row + its
  content survive; upgrade back to 0026 → the column is back
  (round-trip).

The ``alembic`` fixture guarantees the DB ends at head even if a test
fails or the process is interrupted.
"""
from __future__ import annotations

import hashlib
import uuid
from collections.abc import Iterator
from typing import Any

import pytest
from alembic.config import Config
from sqlalchemy import text
from sqlalchemy.orm import Session

from alembic import command
from app.db import SessionLocal, db_available
from app.models import Document

SOURCE = "mig0026"
CONTENT = "content here"
CONTENT_HASH = hashlib.sha256(CONTENT.encode()).hexdigest()


@pytest.fixture()
def alembic(db: Session) -> Iterator[Config]:
    """Real Alembic config bound to the dev DB (URL from app settings).

    Starts at head (repairs an interrupted earlier run); teardown
    upgrades to head no matter what happened, so the dev DB is never
    left below head.
    """
    if not db_available():
        pytest.skip("Postgres not reachable — run `podman compose up -d db` first")
    cfg = Config()  # no alembic.ini file — env.py gets the URL from app config
    cfg.set_main_option("script_location", "alembic")
    command.upgrade(cfg, "head")
    try:
        yield cfg
    finally:
        # Release the test session's open transaction BEFORE the repair
        # DDL: an idle-in-transaction SELECT holds an ACCESS SHARE lock
        # on ``documents``, which would deadlock the repair's
        # ``ALTER TABLE`` forever.
        db.rollback()
        command.upgrade(cfg, "head")


def _version(db: Session) -> str | None:
    return db.execute(text("SELECT version_num FROM alembic_version")).scalar()


def _column(db: Session, column: str) -> tuple[Any, ...] | None:
    """(data_type, is_nullable, column_default) for one documents
    column."""
    row = db.execute(
        text(
            "SELECT data_type, is_nullable, column_default"
            " FROM information_schema.columns"
            " WHERE table_name = 'documents' AND column_name = :c"
        ),
        {"c": column},
    ).fetchone()
    return tuple(row) if row is not None else None


def _index_covering(db: Session, column: str) -> str | None:
    """The name of any documents index whose definition mentions the
    column (None = UNINDEXED)."""
    return db.execute(
        text(
            "SELECT indexname FROM pg_indexes"
            " WHERE tablename = 'documents' AND indexdef ILIKE :like"
        ),
        {"like": f"%{column}%"},
    ).scalar()


def _insert_sql(db: Session, path: str) -> uuid.UUID:
    """Insert one documents row with the PRE-0026 column set (the 0025
    shape — the page_count column, when present, is omitted so a NULL
    backfill is what the row reads)."""
    doc_id = uuid.uuid4()
    db.execute(
        text(
            "INSERT INTO documents (id, source, path, full_path, title, content,"
            " content_hash) VALUES (:id, :s, :p, :fp, :t, :c, :h)"
        ),
        {
            "id": doc_id,
            "s": SOURCE,
            "p": path,
            "fp": f"/tmp/{path}",
            "t": f"Doc {path}",
            "c": CONTENT,
            "h": CONTENT_HASH,
        },
    )
    db.commit()
    return doc_id


def _delete(db: Session, doc_id: uuid.UUID) -> None:
    db.execute(text("DELETE FROM documents WHERE id = :id"), {"id": doc_id})
    db.commit()


def _assert_0025_schema_survives(db: Session) -> None:
    """The 0025 documents contract must survive the additive upgrade —
    in particular ``content_hash`` stays NOT NULL and indexed (the
    retriever's identity — untouched by phase 148) and ``file_hash``
    keeps its 0025 contract (VARCHAR(64), nullable, unindexed)."""
    content = _column(db, "content_hash")
    assert content is not None and content[0] == "character varying" and content[1] == "NO", (
        "documents.content_hash (0001) must keep its 0025 contract"
    )
    assert (
        _index_covering(db, "content_hash") is not None
    ), "the ix_documents_content_hash index must survive the upgrade"

    file_hash = _column(db, "file_hash")
    assert file_hash is not None and file_hash[0] == "character varying", (
        "documents.file_hash (0025) must survive the upgrade"
    )
    assert file_hash[1] == "YES", "file_hash must stay NULLABLE"
    assert file_hash[2] is None, "file_hash must keep NO server default"
    assert _index_covering(db, "file_hash") is None, (
        "file_hash must stay UNINDEXED"
    )

    indexed = _column(db, "indexed_at")
    assert indexed is not None and indexed[0] == "timestamp with time zone", (
        "documents.indexed_at (0001) must survive the upgrade"
    )
    is_image = _column(db, "is_image")
    assert is_image is not None and is_image[0] == "boolean" and is_image[1] == "NO", (
        "documents.is_image (0022) must survive the upgrade"
    )
    image_path = _column(db, "image_path")
    assert image_path is not None and image_path[1] == "YES", (
        "documents.image_path (0022) must survive the upgrade"
    )


def test_upgrade_to_0026_adds_page_count(db: Session, alembic: Config) -> None:
    """Upgrade 0025 → 0026: ``page_count`` exists with the full
    contract (INTEGER, NULLABLE, NO server default — NULL = "not
    page-qualifying" for every pre-phase row — and UNINDEXED: a
    display fact, never a query surface), is ABSENT at 0025, and every
    pre-0026 row backfills ``page_count`` to NULL (no backfill
    possible — a migration cannot know which documents qualified) —
    while the 0025 table contract (incl. the INDEXED ``content_hash``
    and the unindexed ``file_hash``) survives."""
    command.downgrade(alembic, "0025")  # start from the pre-0026 state
    assert _version(db) == "0025"
    assert _column(db, "page_count") is None, "page_count must be absent at 0025"

    pre_id = _insert_sql(db, "pre.md")  # the 0025 column set
    try:
        command.upgrade(alembic, "0026")
        assert _version(db) == "0026", "alembic_version must be at 0026"

        col = _column(db, "page_count")
        assert col is not None, "documents.page_count is missing"
        assert col[0] == "integer", "page_count must be INTEGER"
        assert col[1] == "YES", (
            "page_count must be NULLABLE (NULL = not page-qualifying)"
        )
        assert col[2] is None, (
            "page_count must carry NO server default — NULL is the meaningful"
            " pre-phase state"
        )
        assert _index_covering(db, "page_count") is None, (
            "page_count must be UNINDEXED — a display fact, never a query"
            " surface (the page route reads it from the row after the id lookup)"
        )

        # The pre-0026 row backfilled to NULL (the pre-phase shape).
        assert db.execute(
            text(
                "SELECT content_hash, page_count FROM documents WHERE id = :id"
            ),
            {"id": pre_id},
        ).fetchone() == (CONTENT_HASH, None), (
            "every pre-0026 row must backfill page_count to NULL — no"
            " backfill, no default — while content_hash survives"
        )

        _assert_0025_schema_survives(db)
    finally:
        _delete(db, pre_id)


def test_orm_page_count_round_trips(db: Session, alembic: Config) -> None:
    """The ORM contract agrees with the column contract: a freshly
    inserted ``Document`` with NO ``page_count`` reads
    ``page_count is None`` through a FRESH session (the pre-phase
    shape — NULL until the importer's PDF branch sets it), and an
    explicit count round-trips the value."""
    command.upgrade(alembic, "head")
    row_counted = Document(
        source=SOURCE,
        path="counted.pdf",
        full_path="/tmp/counted.pdf",
        title="counted",
        content=CONTENT,
        content_hash=CONTENT_HASH,
        page_count=24,
    )
    row_uncounted = Document(
        source=SOURCE,
        path="uncounted.md",
        full_path="/tmp/uncounted.md",
        title="uncounted",
        content=CONTENT,
        content_hash=CONTENT_HASH,
    )
    db.add(row_counted)
    db.add(row_uncounted)
    db.commit()
    try:
        with SessionLocal() as fresh:
            reloaded_counted = fresh.get(Document, row_counted.id)
            assert reloaded_counted is not None, "the counted row must be readable"
            assert reloaded_counted.page_count == 24, (
                "the explicit page_count must round-trip through the DB"
            )
            reloaded_uncounted = fresh.get(Document, row_uncounted.id)
            assert reloaded_uncounted is not None, "the uncounted row must be readable"
            assert reloaded_uncounted.page_count is None, (
                "a row created without page_count must read NULL (the pre-phase"
                " shape)"
            )
            assert reloaded_uncounted.content_hash == CONTENT_HASH, (
                "content_hash must be untouched by the page_count addition"
            )
    finally:
        _delete(db, row_counted.id)
        _delete(db, row_uncounted.id)


def test_downgrade_to_0025_drops_the_column(db: Session, alembic: Config) -> None:
    """Downgrade 0026 → 0025: the page_count column is gone (A13 —
    fully reversible, the copies are re-derivable from the source
    files on the next sync) while the row + its content (incl. the
    0025 ``file_hash`` sibling) survive, and the rest of the 0025
    table contract is intact."""
    command.upgrade(alembic, "head")
    row = Document(
        source=SOURCE,
        path="down.pdf",
        full_path="/tmp/down.pdf",
        title="down",
        content=CONTENT,
        content_hash=CONTENT_HASH,
        page_count=24,
    )
    db.add(row)
    db.commit()
    try:
        command.downgrade(alembic, "0025")
        assert _version(db) == "0025"
        assert _column(db, "page_count") is None, "page_count must be dropped"

        surviving = db.execute(
            text(
                "SELECT source, path, content, content_hash, file_hash, is_image,"
                " image_path FROM documents WHERE id = :id"
            ),
            {"id": row.id},
        ).fetchone()
        assert surviving is not None and surviving[0] == SOURCE, (
            "the row must survive the column drop"
        )
        assert surviving[1] == "down.pdf" and surviving[2] == CONTENT, (
            "path + content must survive the drop"
        )
        assert surviving[3] == CONTENT_HASH, (
            "content_hash (the retriever's identity) must survive the drop"
        )
        assert surviving[4] is None and surviving[5] is False and surviving[6] is None, (
            "file_hash + is_image + image_path must survive the drop"
        )
        assert _index_covering(db, "content_hash") is not None, (
            "the content_hash index must survive the downgrade"
        )
    finally:
        _delete(db, row.id)
    # Repair: the fixture teardown re-upgrades to head.


def test_upgrade_round_trip_restores_the_column(db: Session, alembic: Config) -> None:
    """Downgrade to 0025, then upgrade back to 0026: ``page_count`` is
    back with the full contract (INTEGER, NULLABLE, NO server default,
    UNINDEXED), a row seeded at 0025 reads ``page_count is None`` (the
    pre-phase shape on the round-trip), and the 0025 schema (incl. the
    indexed ``content_hash``) is intact."""
    command.downgrade(alembic, "0025")
    pre_id = _insert_sql(db, "round.pdf")
    try:
        command.upgrade(alembic, "0026")
        assert _version(db) == "0026", "round-trip upgrade must land at 0026"

        col = _column(db, "page_count")
        assert col is not None, "documents.page_count must be back"
        assert col[0] == "integer", "page_count must be INTEGER after the round-trip"
        assert col[1] == "YES", "page_count must be NULLABLE after the round-trip"
        assert col[2] is None, (
            "page_count must still carry NO server default after the round-trip"
        )
        assert _index_covering(db, "page_count") is None, (
            "page_count must stay UNINDEXED after the round-trip"
        )

        assert db.execute(
            text("SELECT page_count FROM documents WHERE id = :id"),
            {"id": pre_id},
        ).scalar() is None, (
            "a row seeded at 0025 must read page_count NULL after the round-trip"
        )

        _assert_0025_schema_survives(db)
    finally:
        _delete(db, pre_id)