"""Tests for the PRAGMA user_version migration runner."""

from __future__ import annotations

from datetime import datetime, timezone
from pathlib import Path

import pytest
import sqlalchemy as sa
import sqlalchemy.exc

from cyclone import db_migrate


def _fresh_engine(tmp_path: Path) -> sa.Engine:
    return sa.create_engine(f"sqlite:///{tmp_path}/m.db", future=True)


def _user_version(engine: sa.Engine) -> int:
    with engine.connect() as c:
        return c.exec_driver_sql("PRAGMA user_version").scalar() or 0


def test_run_on_empty_db_bumps_to_latest(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(db_migrate, "MIGRATIONS_DIR", tmp_path)
    (tmp_path / "0001_initial.sql").write_text("-- version: 1\nCREATE TABLE t (id INTEGER);\n")

    engine = _fresh_engine(tmp_path)
    db_migrate.run(engine)

    assert _user_version(engine) == 1
    with engine.connect() as c:
        rows = c.exec_driver_sql("SELECT name FROM sqlite_master WHERE type='table' AND name='t'").all()
    assert len(rows) == 1


def test_run_is_idempotent(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(db_migrate, "MIGRATIONS_DIR", tmp_path)
    (tmp_path / "0001_initial.sql").write_text("-- version: 1\nCREATE TABLE t (id INTEGER);\n")

    engine = _fresh_engine(tmp_path)
    db_migrate.run(engine)
    db_migrate.run(engine)  # second call should be a no-op

    assert _user_version(engine) == 1


def test_run_skips_already_applied_migrations(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(db_migrate, "MIGRATIONS_DIR", tmp_path)
    (tmp_path / "0001_initial.sql").write_text("-- version: 1\nCREATE TABLE t1 (id INTEGER);\n")
    (tmp_path / "0002_add_col.sql").write_text("-- version: 2\nALTER TABLE t1 ADD COLUMN name TEXT;\n")

    engine = _fresh_engine(tmp_path)
    db_migrate.run(engine)
    assert _user_version(engine) == 2

    # Modify 0002 (a no-op for SQLite ALTER, but we want to verify the
    # runner doesn't re-run already-applied migrations).
    db_migrate.run(engine)
    assert _user_version(engine) == 2


def test_run_raises_on_missing_version_header(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(db_migrate, "MIGRATIONS_DIR", tmp_path)
    (tmp_path / "0001_bad.sql").write_text("CREATE TABLE t (id INTEGER);\n")  # no header

    engine = _fresh_engine(tmp_path)
    with pytest.raises(RuntimeError, match="missing '-- version"):
        db_migrate.run(engine)


def test_run_rolls_back_version_on_failed_statement(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """If a DDL statement fails, the version bump must also roll back."""
    monkeypatch.setattr(db_migrate, "MIGRATIONS_DIR", tmp_path)
    (tmp_path / "0001_initial.sql").write_text(
        "-- version: 1\nCREATE TABLE t (id INTEGER);\n"
    )

    engine = _fresh_engine(tmp_path)
    db_migrate.run(engine)  # applies 0001, user_version=1
    assert _user_version(engine) == 1

    # Now drop in a second migration whose DDL conflicts with the first.
    # The runner should apply 0001 and 0002 in one call; 0002 fails, the
    # transaction rolls back, and user_version must stay at 1 (not bump to 2).
    (tmp_path / "0002_bad.sql").write_text(
        "-- version: 2\nCREATE TABLE t (id INTEGER);\n"  # table already exists!
    )

    with pytest.raises(sqlalchemy.exc.SQLAlchemyError):
        db_migrate.run(engine)

    assert _user_version(engine) == 1  # bumped atomically, rolled back on failure


def test_run_ignores_non_sql_files(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """README.md and similar files in MIGRATIONS_DIR must not be executed."""
    monkeypatch.setattr(db_migrate, "MIGRATIONS_DIR", tmp_path)
    (tmp_path / "0001_initial.sql").write_text(
        "-- version: 1\nCREATE TABLE t (id INTEGER);\n"
    )
    (tmp_path / "README.md").write_text("Don't run me!\nCREATE TABLE should_not_exist (x INT);\n")

    engine = _fresh_engine(tmp_path)
    db_migrate.run(engine)

    with engine.connect() as c:
        rows = c.exec_driver_sql(
            "SELECT name FROM sqlite_master WHERE type='table' AND name='should_not_exist'"
        ).all()
    assert len(rows) == 0
    assert _user_version(engine) == 1


def test_drop_claims_unique_constraint_migration(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    """Migration 0013 must drop the inline UNIQUE(batch_id, patient_control_number)
    on claims by recreating the table. Idempotent and preserves data.

    We copy the real 0013 file (so this test exercises the real migration, not
    a synthetic one) and stub 0001-0012 with a synthetic 0001 that has the
    historical inline UNIQUE. If 0013 doesn't exist as a real file, the runner
    will leave user_version at 1 and the test will fail.
    """
    monkeypatch.setattr(db_migrate, "MIGRATIONS_DIR", tmp_path)

    # Synthetic 0001: claims with INLINE UNIQUE(batch_id, patient_control_number).
    # The schema mirrors the real 0001 column shape so 0013's
    # ``INSERT INTO claims_new SELECT * FROM claims`` works against this fixture.
    (tmp_path / "0001_initial.sql").write_text(
        "-- version: 1\n"
        "CREATE TABLE batches (id TEXT PRIMARY KEY, kind TEXT NOT NULL, "
        "input_filename TEXT NOT NULL, parsed_at DATETIME NOT NULL);\n"
        "CREATE TABLE claims ("
        "id TEXT PRIMARY KEY, batch_id TEXT NOT NULL REFERENCES batches(id) ON DELETE CASCADE,"
        "patient_control_number TEXT NOT NULL, service_date_from DATE, service_date_to DATE,"
        "charge_amount NUMERIC(12, 2) NOT NULL DEFAULT 0, provider_npi TEXT, payer_id TEXT,"
        "state TEXT NOT NULL DEFAULT 'submitted', state_before_reversal TEXT,"
        "matched_remittance_id TEXT REFERENCES remittances(id), raw_json TEXT,"
        "rejection_reason TEXT, rejected_at TIMESTAMP, resubmit_count INTEGER NOT NULL DEFAULT 0,"
        "state_changed_at TIMESTAMP, payer_rejected_at TEXT, payer_rejected_reason TEXT,"
        "payer_rejected_status_code TEXT, payer_rejected_by_277ca_id TEXT,"
        "payer_rejected_acknowledged_at TEXT, payer_rejected_acknowledged_actor TEXT,"
        "UNIQUE (batch_id, patient_control_number));\n"
    )

    # Copy the REAL 0013 from the source tree. If it's missing, the test fails
    # because user_version will stay at 1 and we'll never get to assert it == 13.
    real_migrations = Path(db_migrate.__file__).parent / "migrations"
    real_0013 = real_migrations / "0013_drop_claims_unique_constraint.sql"
    assert real_0013.exists(), (
        f"Real migration 0013 missing at {real_0013} — this test is meant to "
        f"exercise the real migration, not a synthetic one."
    )
    (tmp_path / "0013_drop_claims_unique_constraint.sql").write_text(real_0013.read_text())

    engine = _fresh_engine(tmp_path)
    db_migrate.run(engine)
    assert _user_version(engine) == 13, (
        "0013 did not apply; user_version did not reach 13. Either the migration "
        "file is missing, or it has a syntax error."
    )

    # Before-0013 assertion: with the inline UNIQUE, two claims sharing
    # (batch_id, patient_control_number) on the same batch must fail. The
    # recreation must remove that constraint, so this insert must now succeed.
    with engine.begin() as c:
        c.exec_driver_sql(
            "INSERT INTO batches(id, kind, input_filename, parsed_at) "
            "VALUES ('b1', '837p', 'x.txt', '2026-01-01')"
        )
        c.exec_driver_sql(
            "INSERT INTO claims(id, batch_id, patient_control_number) VALUES ('c1', 'b1', 'M')"
        )
        c.exec_driver_sql(
            "INSERT INTO claims(id, batch_id, patient_control_number) VALUES ('c2', 'b1', 'M')"
        )
        ids = [r[0] for r in c.exec_driver_sql("SELECT id FROM claims ORDER BY id").all()]
    assert ids == ["c1", "c2"], (
        "Both claims with same (batch_id, patient_control_number) must persist "
        "after 0013 — the UNIQUE(batch_id, patient_control_number) is gone."
    )


def test_migration_0013_is_idempotent(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    """Re-running db_migrate.run on a v13 DB is a no-op."""
    monkeypatch.setattr(db_migrate, "MIGRATIONS_DIR", tmp_path)

    # Use synthetic 0001 + copy the real 0013 so we exercise the real file.
    (tmp_path / "0001_initial.sql").write_text(
        "-- version: 1\n"
        "CREATE TABLE batches (id TEXT PRIMARY KEY, kind TEXT NOT NULL, "
        "input_filename TEXT NOT NULL, parsed_at DATETIME NOT NULL);\n"
        "CREATE TABLE claims ("
        "id TEXT PRIMARY KEY, batch_id TEXT NOT NULL REFERENCES batches(id) ON DELETE CASCADE,"
        "patient_control_number TEXT NOT NULL, service_date_from DATE, service_date_to DATE,"
        "charge_amount NUMERIC(12, 2) NOT NULL DEFAULT 0, provider_npi TEXT, payer_id TEXT,"
        "state TEXT NOT NULL DEFAULT 'submitted', state_before_reversal TEXT,"
        "matched_remittance_id TEXT REFERENCES remittances(id), raw_json TEXT,"
        "rejection_reason TEXT, rejected_at TIMESTAMP, resubmit_count INTEGER NOT NULL DEFAULT 0,"
        "state_changed_at TIMESTAMP, payer_rejected_at TEXT, payer_rejected_reason TEXT,"
        "payer_rejected_status_code TEXT, payer_rejected_by_277ca_id TEXT,"
        "payer_rejected_acknowledged_at TEXT, payer_rejected_acknowledged_actor TEXT,"
        "UNIQUE (batch_id, patient_control_number));\n"
    )
    real_migrations = Path(db_migrate.__file__).parent / "migrations"
    real_0013 = real_migrations / "0013_drop_claims_unique_constraint.sql"
    assert real_0013.exists(), "Real 0013 missing — test cannot exercise it."
    (tmp_path / "0013_drop_claims_unique_constraint.sql").write_text(real_0013.read_text())

    engine = _fresh_engine(tmp_path)
    db_migrate.run(engine)  # applies both
    version_after_first = _user_version(engine)
    assert version_after_first == 13
    db_migrate.run(engine)  # second call: no-op
    assert _user_version(engine) == version_after_first


# =============================================================================
# Migration 0014: relax claims/remittances PK to composite (batch_id, id)
# =============================================================================


# Synthetic 0001 with the FULL post-0013 column shape, so 0014's
# INSERT...SELECT statements find every column they expect on the source
# tables. Mirrors the shape after 0001 + 0004 + 0006 + 0008 + 0010 + 0013.
# The tables and columns here are exactly what 0014 reads from; we do NOT
# add migrations 0002..0012 because they're orthogonal to the FK chain
# 0014 walks (no claims/remittances FKs to acks, audit_log, etc).
_0014_SYNTHETIC_0001 = (
    "-- version: 1\n"
    "CREATE TABLE batches ("
    "id TEXT PRIMARY KEY, kind TEXT NOT NULL, input_filename TEXT NOT NULL,"
    "parsed_at DATETIME NOT NULL, totals_json TEXT, validation_json TEXT,"
    "raw_result_json TEXT);\n"
    "CREATE TABLE claims ("
    "id TEXT PRIMARY KEY, batch_id TEXT NOT NULL REFERENCES batches(id) ON DELETE CASCADE,"
    "patient_control_number TEXT NOT NULL, service_date_from DATE, service_date_to DATE,"
    "charge_amount NUMERIC(12, 2) NOT NULL DEFAULT 0, provider_npi TEXT, payer_id TEXT,"
    "state TEXT NOT NULL DEFAULT 'submitted', state_before_reversal TEXT,"
    "matched_remittance_id TEXT REFERENCES remittances(id), raw_json TEXT,"
    "rejection_reason TEXT, rejected_at TIMESTAMP,"
    "resubmit_count INTEGER NOT NULL DEFAULT 0, state_changed_at TIMESTAMP,"
    "payer_rejected_at TEXT, payer_rejected_reason TEXT,"
    "payer_rejected_status_code TEXT, payer_rejected_by_277ca_id TEXT,"
    "payer_rejected_acknowledged_at TEXT, payer_rejected_acknowledged_actor TEXT);\n"
    "CREATE TABLE remittances ("
    "id TEXT PRIMARY KEY, batch_id TEXT NOT NULL REFERENCES batches(id) ON DELETE CASCADE,"
    "payer_claim_control_number TEXT NOT NULL, claim_id TEXT REFERENCES claims(id),"
    "status_code TEXT NOT NULL, status_label TEXT,"
    "total_charge NUMERIC(12, 2) NOT NULL DEFAULT 0,"
    "total_paid NUMERIC(12, 2) NOT NULL DEFAULT 0,"
    "patient_responsibility NUMERIC(12, 2),"
    "adjustment_amount NUMERIC(12, 2) NOT NULL DEFAULT 0,"
    "claim_level_adjustment_amount NUMERIC(12, 2) NOT NULL DEFAULT 0,"
    "received_at DATETIME NOT NULL, service_date DATE,"
    "is_reversal INTEGER NOT NULL DEFAULT 0, raw_json TEXT);\n"
    "CREATE TABLE matches ("
    "id INTEGER PRIMARY KEY AUTOINCREMENT,"
    "claim_id TEXT NOT NULL REFERENCES claims(id) ON DELETE CASCADE,"
    "remittance_id TEXT NOT NULL REFERENCES remittances(id) ON DELETE CASCADE,"
    "strategy TEXT NOT NULL, matched_at DATETIME NOT NULL,"
    "prior_claim_state TEXT, is_reversal INTEGER NOT NULL DEFAULT 0);\n"
    "CREATE TABLE cas_adjustments ("
    "id INTEGER PRIMARY KEY AUTOINCREMENT,"
    "remittance_id TEXT NOT NULL REFERENCES remittances(id) ON DELETE CASCADE,"
    "group_code TEXT NOT NULL, reason_code TEXT NOT NULL,"
    "amount NUMERIC(12, 2) NOT NULL, quantity NUMERIC(10, 2),"
    "service_line_payment_id INTEGER REFERENCES service_line_payments(id) ON DELETE SET NULL);\n"
    "CREATE TABLE service_line_payments ("
    "id INTEGER PRIMARY KEY AUTOINCREMENT,"
    "remittance_id VARCHAR(64) NOT NULL REFERENCES remittances(id) ON DELETE CASCADE,"
    "line_number INTEGER NOT NULL, procedure_qualifier VARCHAR(4) NOT NULL,"
    "procedure_code VARCHAR(16) NOT NULL,"
    "modifiers_json TEXT NOT NULL DEFAULT '[]',"
    "charge NUMERIC(12, 2) NOT NULL, payment NUMERIC(12, 2) NOT NULL,"
    "units NUMERIC(10, 2), unit_type VARCHAR(8), service_date DATE,"
    "ref_benefit_plan VARCHAR(64),"
    "superseded_by_id INTEGER REFERENCES service_line_payments(id) ON DELETE SET NULL);\n"
    "CREATE TABLE line_reconciliations ("
    "id INTEGER PRIMARY KEY AUTOINCREMENT,"
    "claim_id VARCHAR(64) NOT NULL REFERENCES claims(id) ON DELETE CASCADE,"
    "claim_service_line_number INTEGER,"
    "service_line_payment_id INTEGER REFERENCES service_line_payments(id) ON DELETE SET NULL,"
    "status VARCHAR(16) NOT NULL, match_score INTEGER, reconciled_at TIMESTAMP NOT NULL);\n"
)


def _apply_0014_only(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> sa.Engine:
    """Apply just 0014 (with a synthetic 0001 that has the post-0013 column
    shape) on a fresh engine. Returns the engine with user_version=14.
    """
    monkeypatch.setattr(db_migrate, "MIGRATIONS_DIR", tmp_path)
    (tmp_path / "0001_initial.sql").write_text(_0014_SYNTHETIC_0001)

    real_migrations = Path(db_migrate.__file__).parent / "migrations"
    real_0014 = real_migrations / "0014_relax_claims_remits_pk.sql"
    assert real_0014.exists(), (
        f"Real migration 0014 missing at {real_0014} — these tests are "
        f"meant to exercise the real migration, not a synthetic one."
    )
    (tmp_path / "0014_relax_claims_remits_pk.sql").write_text(real_0014.read_text())

    engine = _fresh_engine(tmp_path)
    db_migrate.run(engine)
    assert _user_version(engine) == 14, (
        "0014 did not apply; user_version did not reach 14. Either the "
        "migration file is missing, or it has a syntax error."
    )
    return engine


def test_migration_0014_relaxes_claims_pk_to_composite(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Migration 0014 changes claims PK from single-column id to (batch_id, id).

    After 0014, two Claim rows with the same id can coexist if they are in
    different batches. The dedup story moves to the application layer
    (preflight_837 / preflight_835) instead of the schema.
    """
    engine = _apply_0014_only(tmp_path, monkeypatch)

    # After 0014: same id in two different batches must coexist.
    with engine.begin() as c:
        c.exec_driver_sql(
            "INSERT INTO batches(id, kind, input_filename, parsed_at) "
            "VALUES ('B1', '837p', 'b1.txt', '2026-01-01')"
        )
        c.exec_driver_sql(
            "INSERT INTO batches(id, kind, input_filename, parsed_at) "
            "VALUES ('B2', '837p', 'b2.txt', '2026-01-02')"
        )
        c.exec_driver_sql(
            "INSERT INTO claims(id, batch_id, patient_control_number, "
            "state) VALUES ('CLM-A', 'B1', 'M1', 'submitted')"
        )
        c.exec_driver_sql(
            "INSERT INTO claims(id, batch_id, patient_control_number, "
            "state) VALUES ('CLM-A', 'B2', 'M1', 'submitted')"
        )
        rows = c.exec_driver_sql(
            "SELECT id, batch_id FROM claims WHERE id='CLM-A' ORDER BY batch_id"
        ).all()
    assert rows == [("CLM-A", "B1"), ("CLM-A", "B2")], (
        "After 0014, the same claims.id in two different batches must "
        "coexist. Composite PK is (batch_id, id)."
    )


def test_migration_0014_relaxes_remittances_pk_to_composite(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Same shape for remittances: same CLP01 can exist in two batches."""
    engine = _apply_0014_only(tmp_path, monkeypatch)

    with engine.begin() as c:
        c.exec_driver_sql(
            "INSERT INTO batches(id, kind, input_filename, parsed_at) "
            "VALUES ('B1', '835', 'b1.txt', '2026-01-01')"
        )
        c.exec_driver_sql(
            "INSERT INTO batches(id, kind, input_filename, parsed_at) "
            "VALUES ('B2', '835', 'b2.txt', '2026-01-02')"
        )
        c.exec_driver_sql(
            "INSERT INTO remittances(id, batch_id, payer_claim_control_number, "
            "status_code, received_at) "
            "VALUES ('CLP-A', 'B1', 'CLP-A', '1', '2026-01-01')"
        )
        c.exec_driver_sql(
            "INSERT INTO remittances(id, batch_id, payer_claim_control_number, "
            "status_code, received_at) "
            "VALUES ('CLP-A', 'B2', 'CLP-A', '1', '2026-01-02')"
        )
        rows = c.exec_driver_sql(
            "SELECT id, batch_id FROM remittances WHERE id='CLP-A' ORDER BY batch_id"
        ).all()
    assert rows == [("CLP-A", "B1"), ("CLP-A", "B2")], (
        "After 0014, the same remittances.id in two different batches must "
        "coexist. Composite PK is (batch_id, id)."
    )


def test_migration_0014_preserves_existing_data(tmp_path, monkeypatch):
    """Data inserted at v=13 must survive the v=14 table recreation.

    0014 uses ``INSERT INTO claims_new SELECT * FROM claims`` (and the
    remittance-side equivalent) to populate the recreated tables. Any rows
    present before 0014 must still be there after 0014 finishes — that is
    the contract that lets the schema change ship without data loss.

    This test exercises the real 0014 against real, pre-existing rows:
    1. Apply 0001-0013 via ``migrate_to_version``.
    2. Seed a claim and a remittance in v=13 (raw SQL — the ORM model
       declares ``matched_remittance_batch_id`` which doesn't exist at v=13).
    3. Apply 0014 via ``migrate_to_version``.
    4. Verify the seeded rows survive, with the same id and batch_id.
    5. Verify the SAME id can be inserted in a DIFFERENT batch after 0014
       (proving the composite PK took effect — a single-column PK would
       have rejected the second insert).

    Uses a fresh ``sa.Engine`` (not the module-global ``db.engine()``)
    because the conftest autouse fixture has already initialized the
    module engine at v=14 — we need a clean engine that starts at v=0
    so ``migrate_to_version(13)`` actually applies 0001..0013.
    """
    from sqlalchemy.orm import sessionmaker

    engine = _fresh_engine(tmp_path)
    SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, expire_on_commit=False)

    from cyclone.db_migrate import migrate_to_version
    from cyclone.db import Batch, Claim

    # Apply 0001-0013 only.
    migrate_to_version(engine, 13)
    assert _user_version(engine) == 13

    # Seed a claim and a remittance in v=13. We use raw SQL here because
    # the ORM Claim/Remittance models declare post-0014 columns
    # (``matched_remittance_batch_id``) that don't exist at v=13. The
    # migration's data-preservation contract is independent of how the
    # data got there — raw SQL inserts count as "data that needs to
    # survive the migration" just like ORM inserts do.
    with engine.begin() as c:
        c.exec_driver_sql(
            "INSERT INTO batches(id, kind, input_filename, parsed_at) "
            "VALUES ('B-V13', '837p', 'x.txt', '2026-01-01')"
        )
        c.exec_driver_sql(
            "INSERT INTO claims(id, batch_id, patient_control_number, "
            "state, state_changed_at) VALUES "
            "('CLM-PRESERVED', 'B-V13', 'M', 'submitted', '2026-01-01')"
        )
        c.exec_driver_sql(
            "INSERT INTO batches(id, kind, input_filename, parsed_at) "
            "VALUES ('B-V13R', '835', 'x.txt', '2026-01-02')"
        )
        c.exec_driver_sql(
            "INSERT INTO remittances(id, batch_id, payer_claim_control_number, "
            "status_code, received_at) VALUES "
            "('CLP-PRESERVED', 'B-V13R', 'CLP-PRESERVED', '1', '2026-01-02')"
        )

    # Apply 0014. Data must survive.
    migrate_to_version(engine, 14)
    assert _user_version(engine) == 14

    with engine.connect() as c:
        claim = c.exec_driver_sql(
            "SELECT id, batch_id, patient_control_number "
            "FROM claims WHERE id='CLM-PRESERVED'"
        ).first()
        remit = c.exec_driver_sql(
            "SELECT id, batch_id, payer_claim_control_number "
            "FROM remittances WHERE id='CLP-PRESERVED'"
        ).first()
    assert claim == ("CLM-PRESERVED", "B-V13", "M")
    assert remit == ("CLP-PRESERVED", "B-V13R", "CLP-PRESERVED")

    # Confirm that AFTER v=14, the same id can be inserted in a different
    # batch via the ORM (now that the composite PK exists, this is the
    # canonical insert path).
    with SessionLocal() as s:
        s.add(Batch(id="B-V14", kind="837p", input_filename="y.txt",
                    parsed_at=datetime(2026, 1, 3, tzinfo=timezone.utc),
                    raw_result_json={}))
        s.add(Claim(id="CLM-PRESERVED", batch_id="B-V14",
                    patient_control_number="M2",
                    state_changed_at=datetime(2026, 1, 3, tzinfo=timezone.utc)))
        s.commit()

    with engine.connect() as c:
        rows = c.exec_driver_sql(
            "SELECT batch_id FROM claims WHERE id='CLM-PRESERVED' ORDER BY batch_id"
        ).all()
    assert [r[0] for r in rows] == ["B-V13", "B-V14"]


def test_migration_0014_preserves_joint_fk_rows(tmp_path, monkeypatch):
    """Migration 0014 also recreates the JOIN-FK tables (matches,
    cas_adjustments, service_line_payments, line_reconciliations). Each of
    those tables has a composite FK to claims/remittances whose ``batch_id``
    side gets populated by JOIN in the migration. Insert representative
    rows at v=13 and verify they survive 0014 with the new batch_id column
    populated.

    This is the complement to ``test_migration_0014_preserves_existing_data``
    (which covers the claims/remittances tables) and exercises the
    ``INSERT INTO ..._new SELECT ... JOIN parent`` statements that the
    simpler test cannot.

    Like the simpler test, we seed at v=13 via raw SQL — the ORM model
    declares ``batch_id`` on the JOIN-FK tables but that column is only
    added by 0014 itself, so an ORM insert at v=13 would fail.
    """
    from sqlalchemy.orm import sessionmaker

    engine = _fresh_engine(tmp_path)
    SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, expire_on_commit=False)

    from cyclone.db_migrate import migrate_to_version

    migrate_to_version(engine, 13)

    # Seed parent rows + one row per JOIN-FK child table (raw SQL at v=13).
    with engine.begin() as c:
        c.exec_driver_sql(
            "INSERT INTO batches(id, kind, input_filename, parsed_at) "
            "VALUES ('B-JOIN', '837p', 'j.txt', '2026-01-01')"
        )
        c.exec_driver_sql(
            "INSERT INTO claims(id, batch_id, patient_control_number, "
            "state, state_changed_at) VALUES "
            "('CLM-J', 'B-JOIN', 'M', 'submitted', '2026-01-01')"
        )
        c.exec_driver_sql(
            "INSERT INTO batches(id, kind, input_filename, parsed_at) "
            "VALUES ('B-JOIN-R', '835', 'jr.txt', '2026-01-02')"
        )
        c.exec_driver_sql(
            "INSERT INTO remittances(id, batch_id, payer_claim_control_number, "
            "status_code, received_at) VALUES "
            "('CLP-J', 'B-JOIN-R', 'CLP-J', '1', '2026-01-02')"
        )
        # Seed the four JOIN-FK child rows. At v=13 these tables only have
        # the parent-id columns (no batch_id side yet) — 0014 adds the
        # batch_id columns and JOINs against the parents to populate them.
        c.exec_driver_sql(
            "INSERT INTO service_line_payments("
            "remittance_id, line_number, procedure_qualifier, procedure_code, "
            "modifiers_json, charge, payment) VALUES "
            "('CLP-J', 1, 'HC', '99213', '[]', 100.00, 80.00)"
        )
        c.exec_driver_sql(
            "INSERT INTO cas_adjustments("
            "remittance_id, group_code, reason_code, amount) VALUES "
            "('CLP-J', 'CO', '45', 20.00)"
        )
        c.exec_driver_sql(
            "INSERT INTO line_reconciliations("
            "claim_id, status, match_score, reconciled_at) VALUES "
            "('CLM-J', 'matched', 95, '2026-01-02')"
        )
        c.exec_driver_sql(
            "INSERT INTO matches("
            "claim_id, remittance_id, strategy, matched_at) VALUES "
            "('CLM-J', 'CLP-J', 'exact', '2026-01-02')"
        )

    # Apply 0014 — the JOIN-FK tables get the batch_id column populated by
    # the migration's INSERT...SELECT...JOIN statements.
    migrate_to_version(engine, 14)

    with engine.connect() as c:
        rows = c.exec_driver_sql(
            "SELECT remittance_id, remittance_batch_id "
            "FROM service_line_payments WHERE remittance_id='CLP-J'"
        ).all()
        assert rows == [("CLP-J", "B-JOIN-R")], (
            "service_line_payments.remittance_batch_id must be populated "
            "by 0014's JOIN against remittances; got " + repr(rows)
        )
        rows = c.exec_driver_sql(
            "SELECT remittance_id, remittance_batch_id "
            "FROM cas_adjustments WHERE remittance_id='CLP-J'"
        ).all()
        assert rows == [("CLP-J", "B-JOIN-R")], (
            "cas_adjustments.remittance_batch_id must be populated by 0014; got "
            + repr(rows)
        )
        rows = c.exec_driver_sql(
            "SELECT claim_id, batch_id FROM line_reconciliations WHERE claim_id='CLM-J'"
        ).all()
        assert rows == [("CLM-J", "B-JOIN")], (
            "line_reconciliations.batch_id must be populated by 0014; got "
            + repr(rows)
        )
        rows = c.exec_driver_sql(
            "SELECT claim_id, batch_id, remittance_id, remittance_batch_id "
            "FROM matches WHERE claim_id='CLM-J'"
        ).all()
        assert rows == [("CLM-J", "B-JOIN", "CLP-J", "B-JOIN-R")], (
            "matches batch_id + remittance_batch_id must be populated by 0014; "
            "got " + repr(rows)
        )


# =============================================================================
# migrate_to_version: edge-case behavior
# =============================================================================
# These tests use a fresh ``sa.Engine`` (not ``db.engine()``) because the
# autouse conftest fixture has already initialized the module engine at the
# latest version. We need a clean engine that starts at ``user_version=0``
# so that ``migrate_to_version(N)`` actually exercises the apply / no-op /
# beyond-max branches.


def test_migrate_to_version_idempotent(tmp_path, monkeypatch):
    """Calling migrate_to_version(engine, N) twice is equivalent to calling once."""
    from cyclone.db_migrate import migrate_to_version

    engine = _fresh_engine(tmp_path)
    migrate_to_version(engine, 5)
    with engine.begin() as conn:
        v1 = conn.exec_driver_sql("PRAGMA user_version").scalar()
    migrate_to_version(engine, 5)
    with engine.begin() as conn:
        v2 = conn.exec_driver_sql("PRAGMA user_version").scalar()
    assert v1 == v2 == 5


def test_migrate_to_version_lower_than_current_is_noop(tmp_path, monkeypatch):
    """migrate_to_version(engine, 3) after migrate_to_version(engine, 13) is a no-op."""
    from cyclone.db_migrate import migrate_to_version

    engine = _fresh_engine(tmp_path)
    migrate_to_version(engine, 13)
    with engine.begin() as conn:
        v_after_13 = conn.exec_driver_sql("PRAGMA user_version").scalar()
    migrate_to_version(engine, 3)  # lower, no-op
    with engine.begin() as conn:
        v_after_3 = conn.exec_driver_sql("PRAGMA user_version").scalar()
    assert v_after_13 == v_after_3 == 13


def test_migrate_to_version_zero_on_fresh_db(tmp_path, monkeypatch):
    """migrate_to_version(engine, 0) on a fresh DB is a no-op."""
    from cyclone.db_migrate import migrate_to_version

    engine = _fresh_engine(tmp_path)
    migrate_to_version(engine, 0)
    with engine.begin() as conn:
        v = conn.exec_driver_sql("PRAGMA user_version").scalar()
    assert v == 0


def test_migrate_to_version_beyond_max_applies_all(tmp_path, monkeypatch):
    """migrate_to_version(engine, 999) applies every migration (equivalent to run)."""
    import re
    from cyclone.db_migrate import MIGRATIONS_DIR, migrate_to_version

    engine = _fresh_engine(tmp_path)
    migrate_to_version(engine, 999)
    with engine.begin() as conn:
        v = conn.exec_driver_sql("PRAGMA user_version").scalar()
    # Should equal the highest version present in the migrations directory.
    max_version = max(
        int(re.match(r"^(\d+)", p.name).group(1))
        for p in MIGRATIONS_DIR.glob("*.sql")
    )
    assert v == max_version