"""THE app-owned survival gate (ADR #021, phase 2 of the PlusROI
soft-launch).

Proves, on a real scratch Postgres, that an APP-owned tenant survives
the nightly Bowden Works rebuild untouched:

  1. legacy + sheet transform (PlusROI sheet-owned — the parallel-run
     era). Sanity: a second sheet-owned run REGENERATES the book row
     uuids (the pre-flip behavior that makes this gate non-vacuous).
  2. flip PlusROI to books_owner="app" (the real flip function from
     scripts/flip_books_owner.py; idempotence asserted).
  3. capture every PlusROI invoice/expense/project id + party FKs.
  4. run the transform AGAIN (the simulated next BW nightly, sheet
     export still present).
  5. every PlusROI row still exists with the SAME id, every party FK
     still resolves, the shared parties (Bowden Works, PlusROI, Adi)
     kept their uuids across BOTH runs, the sheet step reports the
     app-owned skip (books NOT re-imported), the app-owned audit trail
     survives, and the harness parity comparison on this same database
     stays green (books tenants are invisible to it by construction).
"""

import os
import uuid
from pathlib import Path

import pytest
from sqlalchemy import create_engine, text

from tests.migrate.conftest import FIXTURE_SQL, extract_both, rows, scalar

FIX_SHEET = Path(__file__).parent / "fixtures" / "sheet"
SURVIVAL_DB = "with_app_owned_test"


# --- dedicated scratch DB (pattern of test_sheet_books.py) ------------------


@pytest.fixture(scope="module")
def surv_url() -> str:
    from app.config import settings

    base = settings.with_db_url_host or settings.with_db_url
    if not base:
        pytest.fail("survival tests need the sidecar Postgres (WITH_DB_URL_HOST)")
    admin = create_engine(
        base, isolation_level="AUTOCOMMIT", connect_args={"connect_timeout": 3}
    )
    try:
        with admin.connect() as conn:
            conn.execute(text(f"DROP DATABASE IF EXISTS {SURVIVAL_DB} WITH (FORCE)"))
            conn.execute(text(f"CREATE DATABASE {SURVIVAL_DB}"))
    except Exception as exc:
        pytest.fail(f"cannot reach the sidecar Postgres for survival tests: {exc}")
    finally:
        admin.dispose()

    url = base.rsplit("/", 1)[0] + f"/{SURVIVAL_DB}"
    from alembic.config import Config

    from alembic import command

    root = Path(__file__).resolve().parents[2]
    cfg = Config(str(root / "alembic.ini"))
    cfg.set_main_option("script_location", str(root / "alembic"))
    previous = os.environ.get("WITH_DB_URL")
    os.environ["WITH_DB_URL"] = url
    try:
        command.upgrade(cfg, "head")
    finally:
        os.environ["WITH_DB_URL"] = previous if previous is not None else ""
    yield url

    admin = create_engine(base, isolation_level="AUTOCOMMIT")
    with admin.connect() as conn:
        conn.execute(text(f"DROP DATABASE IF EXISTS {SURVIVAL_DB} WITH (FORCE)"))
    admin.dispose()


@pytest.fixture(scope="module")
def surv_engine(surv_url):
    engine = create_engine(surv_url)
    yield engine
    engine.dispose()


# --- helpers -----------------------------------------------------------------


def _run(engine, books_source=FIX_SHEET):
    from app.services.migrate import run_all

    report = run_all(engine=engine, books_source=str(books_source))
    assert report.ok, report.render()
    return report


def _pr_tenant(engine) -> uuid.UUID:
    return scalar(
        engine,
        "SELECT t.id FROM tenants t JOIN parties p ON p.id = t.party_id"
        " WHERE lower(p.name) = 'plusroi'",
    )


def _books_snapshot(engine, tenant) -> dict:
    """Every PlusROI book row id + its party/project FKs."""
    return {
        "invoices": {
            r[0]: (r[1], r[2])
            for r in rows(
                engine,
                "SELECT id::text, from_party_id::text, to_party_id::text"
                " FROM invoices WHERE tenant_id = :t",
                t=tenant,
            )
        },
        "expenses": {
            r[0]: (r[1], r[2], r[3])
            for r in rows(
                engine,
                "SELECT id::text, vendor_party_id::text, paid_by_party_id::text,"
                " project_id::text FROM expenses WHERE tenant_id = :t",
                t=tenant,
            )
        },
        "projects": {
            r[0]: (r[1], r[2])
            for r in rows(
                engine,
                "SELECT id::text, operator_party_id::text, client_party_id::text"
                " FROM projects WHERE tenant_id = :t",
                t=tenant,
            )
        },
        "engagements": {
            r[0]
            for r in rows(
                engine,
                "SELECT id::text FROM engagements WHERE tenant_id = :t",
                t=tenant,
            )
        },
    }


def _shared_party_ids(engine) -> dict:
    return {
        "bowden_works": scalar(
            engine, "SELECT id FROM parties WHERE name = 'Bowden Works'"
        ),
        "plusroi": scalar(engine, "SELECT id FROM parties WHERE name = 'PlusROI'"),
        "adi_person": scalar(
            engine,
            "SELECT id FROM parties WHERE lower(email) = 'info@adipramono.com'"
            " AND kind = 'person'",
        ),
    }


# --- the gate ----------------------------------------------------------------


def test_app_owned_tenant_survives_the_nightly(surv_url, surv_engine):
    from scripts.flip_books_owner import flip_books_owner

    # (1) fresh legacy + sheet build — PlusROI is sheet-owned.
    with surv_engine.begin() as conn:
        conn.exec_driver_sql(FIXTURE_SQL.read_text(encoding="utf-8"))
        # rob logs in before the flip — his party link must survive later
        conn.execute(
            text(
                "INSERT INTO users (id, idauth_username, email)"
                " VALUES (gen_random_uuid(), 'rob', '')"
                " ON CONFLICT (idauth_username) DO NOTHING"
            )
        )
    _run(surv_engine)
    tenant_1 = _pr_tenant(surv_engine)
    books_1 = _books_snapshot(surv_engine, tenant_1)
    parties_1 = _shared_party_ids(surv_engine)
    assert books_1["invoices"] and books_1["expenses"] and books_1["projects"]

    # Sanity (non-vacuousness): while SHEET-owned, a rebuild regenerates
    # the book rows — different uuids, even though counts match.
    _run(surv_engine)
    tenant_2 = _pr_tenant(surv_engine)
    books_2 = _books_snapshot(surv_engine, tenant_2)
    assert len(books_2["invoices"]) == len(books_1["invoices"])
    assert set(books_2["invoices"]) != set(books_1["invoices"])
    # ... but the shared parties are ALREADY stable (upsert-by-slug).
    assert _shared_party_ids(surv_engine) == parties_1

    # (2) the flip — idempotent, preserves the other settings.
    result = flip_books_owner(surv_engine, "plusroi", "app")
    assert result["changed"] is True
    assert result["after"]["books_owner"] == "app"
    assert result["after"]["bookkeeping_pct"] == 3.0
    assert result["after"]["partner_split"] == 0.5
    assert "parallel_run" not in result["after"]
    again = flip_books_owner(surv_engine, "plusroi", "app")
    assert again["changed"] is False

    # (3) capture the app-owned state + plant an app-era audit row.
    books_before = _books_snapshot(surv_engine, tenant_2)
    audit_id = uuid.uuid4()
    # Phase-3 tables (alembic 0008 / ADR #022): a project_milestones row on a
    # PlusROI project (scoped THROUGH its project) + a tenant_vendors marker
    # (own tenant_id). Both are app-owned writes that MUST survive the nightly.
    milestone_id = uuid.uuid4()
    a_project_id = next(iter(books_before["projects"]))  # a PlusROI project
    tv_vendor_party = parties_1["bowden_works"]  # any real party FK to resolve
    with surv_engine.begin() as conn:
        conn.execute(
            text(
                "INSERT INTO audit_log (id, tenant_id, action)"
                " VALUES (:id, :t, 'invoices.create')"
            ),
            {"id": audit_id, "t": tenant_2},
        )
        conn.execute(
            text(
                "INSERT INTO project_milestones (id, project_id, label, amount,"
                " currency, sort_order)"
                " VALUES (:id, :p, 'Deposit', 1000.00, 'CAD', 0)"
            ),
            {"id": milestone_id, "p": a_project_id},
        )
        conn.execute(
            text(
                "INSERT INTO tenant_vendors (tenant_id, vendor_party_id,"
                " default_category) VALUES (:t, :v, 'Software')"
            ),
            {"t": tenant_2, "v": tv_vendor_party},
        )

    # (4) the simulated next BW nightly (sheet export still on disk).
    report = _run(surv_engine)
    sheet_step = next(s for s in report.steps if s.name.startswith("2.10"))
    assert sheet_step.details.get("skipped_app_owned") == str(tenant_2)
    assert not any(k.startswith("sheet.") for k in report.reconciliation)

    # (5a) every PlusROI row still exists with the SAME id + same FKs.
    assert _pr_tenant(surv_engine) == tenant_2
    books_after = _books_snapshot(surv_engine, tenant_2)
    assert books_after == books_before

    # (5b) every party FK still RESOLVES (no dangling references).
    for table, fk in (
        ("invoices", "from_party_id"),
        ("invoices", "to_party_id"),
        ("expenses", "vendor_party_id"),
        ("expenses", "paid_by_party_id"),
        ("projects", "operator_party_id"),
        ("projects", "client_party_id"),
    ):
        dangling = scalar(
            surv_engine,
            f"SELECT count(*) FROM {table} x"  # noqa: S608
            f" LEFT JOIN parties p ON p.id = x.{fk}"
            f" WHERE x.tenant_id = :t AND x.{fk} IS NOT NULL AND p.id IS NULL",
            t=tenant_2,
        )
        assert dangling == 0, f"{table}.{fk} dangles after the nightly"
    dangling_projects = scalar(
        surv_engine,
        "SELECT count(*) FROM expenses e LEFT JOIN projects p ON p.id = e.project_id"
        " WHERE e.tenant_id = :t AND e.project_id IS NOT NULL AND p.id IS NULL",
        t=tenant_2,
    )
    assert dangling_projects == 0

    # (5c) the shared parties kept their uuids across ALL THREE runs.
    assert _shared_party_ids(surv_engine) == parties_1

    # (5d) settings survived (still app-owned; splits intact).
    settings = scalar(
        surv_engine, "SELECT settings FROM tenants WHERE id = :t", t=tenant_2
    )
    assert settings["books_owner"] == "app"
    assert settings["bookkeeping_pct"] == 3.0

    # (5e) the app-era audit row survived; rob's login link survived.
    assert scalar(
        surv_engine, "SELECT count(*) FROM audit_log WHERE id = :id", id=audit_id
    ) == 1
    rob_party = scalar(
        surv_engine,
        "SELECT person_party_id FROM users WHERE idauth_username = 'rob'",
    )
    assert rob_party is not None
    assert scalar(
        surv_engine, "SELECT name FROM parties WHERE id = :id", id=rob_party
    ) == "Rob Cooper"

    # (5f) the BW side DID rebuild (the nightly still does its job).
    assert scalar(surv_engine, "SELECT count(*) FROM time_entries") == 7
    assert scalar(
        surv_engine,
        "SELECT count(*) FROM tenants t JOIN parties p ON p.id = t.party_id"
        " WHERE p.name = 'Bowden Works'",
    ) == 1

    # (5h) the phase-3 app-owned rows survived the nightly (alembic 0008 /
    # ADR #022): project_milestones (scoped THROUGH its project) +
    # tenant_vendors (own tenant_id). Same ids, and their FKs still resolve.
    assert scalar(
        surv_engine,
        "SELECT count(*) FROM project_milestones WHERE id = :id", id=milestone_id
    ) == 1, "app-owned project_milestones row wiped by the nightly"
    assert str(
        scalar(
            surv_engine,
            "SELECT project_id FROM project_milestones WHERE id = :id",
            id=milestone_id,
        )
    ) == a_project_id  # still bolted to the same (surviving) PlusROI project
    assert scalar(
        surv_engine,
        "SELECT count(*) FROM project_milestones m"
        " JOIN projects p ON p.id = m.project_id WHERE m.id = :id AND p.tenant_id = :t",
        id=milestone_id,
        t=tenant_2,
    ) == 1  # the project FK resolves AND is still the app-owned tenant's
    assert scalar(
        surv_engine,
        "SELECT count(*) FROM tenant_vendors"
        " WHERE tenant_id = :t AND vendor_party_id = :v",
        t=tenant_2,
        v=tv_vendor_party,
    ) == 1, "app-owned tenant_vendors row wiped by the nightly"
    assert scalar(
        surv_engine,
        "SELECT count(*) FROM tenant_vendors tv"
        " JOIN parties p ON p.id = tv.vendor_party_id WHERE tv.tenant_id = :t",
        t=tenant_2,
    ) == 1  # the vendor party FK still resolves

    # (5g) the harness parity comparison on THIS database — which now
    # holds an app-owned books tenant — is still green: books tenants
    # are invisible to the legacy↔mirror comparison by construction.
    from harness.compare import DEFAULT_ALLOWLIST, load_allowlist, run_compare

    old_doc, new_doc = extract_both(surv_url)
    compare = run_compare(old_doc, new_doc, load_allowlist(DEFAULT_ALLOWLIST))
    failures = [(d.path, d.old, d.new) for d in compare.failures]
    assert compare.ok, f"harness RED with an app-owned tenant present: {failures}"