"""M6 invoice harness: its OWN scratch database (`with_m6_test`),
migrated with the real alembic chain, then loaded with the REAL legacy
snapshot (copied table-by-table from the main sidecar's ``legacy``
schema — the same COPY-stream mechanism as scripts/sync_legacy_host)
and transformed once.

The result is a commit-safe clone of production-shaped MIGRATED data:
4,593 real entries, the 3 real migrated invoices, the real legacy
oracle rows — so the M6 gates (paste-block byte-diff against the
legacy TS code of record; the full apply→lock→detach lifecycle with a
harness verdict) run on real data without ever writing to the real
sidecar database.

Isolation: unit-style tests use the per-test SAVEPOINT session ``s``
(service commit()s stay inside it; everything rolls back). The gate
tests use REAL commits — their cycles restore the data by design,
which is exactly what they prove.
"""

import os
import uuid
from pathlib import Path

import pytest
from sqlalchemy import create_engine, select, text
from sqlalchemy.orm import Session

from app.models import TimeEntry
from app.services.authz.context import resolve_actor
from app.services.reporting.filters import status_condition
from tests.authz.conftest import fake_request

ROOT = Path(__file__).resolve().parents[2]
TEST_DB = "with_m6_test"

#: Legacy tables the transforms read — the extractor's list plus the
#: scrubbed users mirror (same set scripts/sync_legacy_host lands).
LEGACY_TABLES = (
    "organizations", "profiles", "organization_members", "teams",
    "team_members", "clockify_imports", "time_entries", "comments",
    "operators", "clients", "projects", "pending_imports",
    "project_rate_overrides", "invoices", "cc_expense_batches",
    "cc_expense_lines", "cc_highlight_keywords", "cc_auto_rules",
    "users",
)


def _sidecar_url() -> str:
    from app.config import settings

    url = settings.with_db_url_host or settings.with_db_url
    if not url:
        pytest.fail(
            "invoice tests need the sidecar Postgres — set WITH_DB_URL_HOST"
            " in .env (127.0.0.1:54317)"
        )
    return url


def _dsn(url: str) -> str:
    return url.replace("postgresql+psycopg://", "postgresql://", 1)


@pytest.fixture(scope="session")
def m6_url() -> str:
    base = _sidecar_url()
    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 {TEST_DB} WITH (FORCE)"))
            conn.execute(text(f"CREATE DATABASE {TEST_DB}"))
    except Exception as exc:
        pytest.fail(f"cannot reach the sidecar Postgres for invoice tests: {exc}")
    finally:
        admin.dispose()

    url = base.rsplit("/", 1)[0] + f"/{TEST_DB}"

    from alembic.config import Config

    from alembic import command

    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 ""
    return url


def _copy_real_legacy(src_dsn: str, dst_dsn: str) -> None:
    """Mirror the REAL legacy snapshot (main sidecar DB, schema
    ``legacy``) into the scratch DB, read-only on the source."""
    import psycopg

    from scripts.sync_legacy_host import copy_table, introspect

    with (
        psycopg.connect(
            src_dsn,
            options="-c default_transaction_read_only=on",
            application_name="with-m6-tests",
        ) as src,
        psycopg.connect(dst_dsn, application_name="with-m6-tests") as dst,
    ):
        src.read_only = True
        with src.cursor() as cur:
            cur.execute(
                "SELECT EXISTS (SELECT 1 FROM information_schema.tables"
                " WHERE table_schema = 'legacy' AND table_name = 'time_entries')"
            )
            if not cur.fetchone()[0]:
                pytest.fail(
                    "the sidecar has no legacy snapshot — run the nightly"
                    " pipeline (or scripts/sync_legacy_host) first"
                )
        with dst.cursor() as cur:
            cur.execute("DROP SCHEMA IF EXISTS legacy CASCADE")
            cur.execute("CREATE SCHEMA legacy")
        for table in LEGACY_TABLES:
            copy_table(
                src, dst, "legacy", table, introspect(src.cursor(), "legacy", table)
            )
        dst.commit()


@pytest.fixture(scope="session")
def m6_engine(m6_url):
    _copy_real_legacy(_dsn(_sidecar_url()), _dsn(m6_url))
    engine = create_engine(m6_url)
    from app.services.migrate import run_all

    report = run_all(engine=engine)
    assert report.ok, report.render()
    yield engine
    engine.dispose()


@pytest.fixture()
def s(m6_engine):
    """SAVEPOINT-isolated session: service-level commit()s stay inside
    the outer transaction, which rolls back after the test."""
    conn = m6_engine.connect()
    trans = conn.begin()
    session = Session(bind=conn, join_transaction_mode="create_savepoint")
    yield session
    session.close()
    trans.rollback()
    conn.close()


def actor_named(session: Session, username: str):
    return resolve_actor(fake_request(username), session)


@pytest.fixture()
def rian(s):
    return actor_named(s, "rian")


@pytest.fixture()
def adi(s):
    return actor_named(s, "adi")


def pending_ids(session: Session, limit: int) -> list[uuid.UUID]:
    """Deterministic slice of REAL pending entries (THE one predicate)."""
    return list(
        session.scalars(
            select(TimeEntry.id)
            .where(status_condition("pending"))
            .order_by(TimeEntry.id)
            .limit(limit)
        ).all()
    )