import os
from pathlib import Path
import subprocess
import sys

import pytest

from app.document_reconciliation_backfill import classify_group, select_complete_group_batch
from app.document_reconciliation_service import (
    _command_fingerprint,
    resolve_document_links,
    select_valid_primary,
)
from scripts.migrate_document_reconciliation_v2 import actions_for_only_unambiguous
from scripts.migrate_document_reconciliation_v2 import apply_actions
from scripts.preflight_document_reconciliation_v2 import run_preflight


def _doc(doc_id: str, *, role="related", primary=False, status="OPEN", external_id=None, payload=None):
    return {"id": doc_id, "opportunity_id": "00000000-0000-0000-0000-000000000001",
            "document_kind": "quotation", "role": role, "is_primary": primary,
            "status": status, "external_id": external_id or doc_id, "payload": payload or {}}


class _Rows:
    def __init__(self, rows=None, scalar=None):
        self._rows, self._scalar = rows or [], scalar
    def scalar(self): return self._scalar
    def mappings(self): return self
    def all(self): return self._rows


class _ResolverConnection:
    def __init__(self, legacy, v2=None, schema=True):
        self.legacy, self.v2, self.schema = legacy, v2 or [], schema
    def execute(self, statement, params=None):
        sql = str(statement)
        if "to_regclass" in sql:
            return _Rows(scalar=self.schema)
        if "FROM commercial_documents d" in sql and "opportunity_document_links" not in sql:
            return _Rows(self.legacy)
        if "FROM opportunity_document_links l" in sql:
            return _Rows(self.v2)
        raise AssertionError(sql)


def _legacy_row(doc_id, kind="quotation", primary=True):
    return {"id": doc_id, "opportunity_id": "o", "document_kind": kind,
            "role": "current" if primary else "related", "is_primary": primary,
            "is_active": True, "status": "OPEN", "updated_at": "1"}


def test_resolver_works_without_v2_schema_and_with_empty_schema():
    legacy = [_legacy_row("a")]
    without = resolve_document_links("o", conn=_ResolverConnection(legacy, schema=False))
    empty = resolve_document_links("o", conn=_ResolverConnection(legacy, v2=[], schema=True))
    assert without[0]["resolution_source"] == "legacy_rollout"
    assert empty[0]["resolution_source"] == "legacy_rollout"


def test_resolver_uses_legacy_for_partial_group_and_v2_after_complete_group():
    legacy = [_legacy_row("a"), _legacy_row("b", primary=False)]
    partial_v2 = [{"document_id": "a", "document_kind": "quotation", "relationship": "PRIMARY",
                   "ended_at": None, "updated_at": "2"}]
    partial = resolve_document_links("o", conn=_ResolverConnection(legacy, partial_v2))
    assert len(partial) == 2 and {row["resolution_source"] for row in partial} == {"legacy_rollout"}
    complete_v2 = partial_v2 + [{"document_id": "b", "document_kind": "quotation",
                                 "relationship": "SECONDARY", "ended_at": None, "updated_at": "2"}]
    complete = resolve_document_links("o", conn=_ResolverConnection(legacy, complete_v2))
    assert len(complete) == 2 and {row["resolution_source"] for row in complete} == {"v2"}


def test_resolver_contract_keeps_document_id_separate_from_link_id():
    legacy = [_legacy_row("document-a")]
    v2 = [{"id": "document-a", "document_id": "document-a", "link_id": "link-z",
           "document_kind": "quotation", "relationship": "PRIMARY",
           "ended_at": None, "updated_at": "2"}]
    resolved = resolve_document_links("o", conn=_ResolverConnection(legacy, v2))
    assert resolved[0]["id"] == "document-a"
    assert resolved[0]["document_id"] == "document-a"
    assert resolved[0]["link_id"] == "link-z"

    fallback = resolve_document_links("o", conn=_ResolverConnection(legacy, schema=False))
    assert fallback[0]["id"] == fallback[0]["document_id"] == "document-a"
    assert fallback[0]["link_id"] is None


def test_v2_projection_contains_ui_document_fields_and_explicit_ids():
    source = Path("app/document_reconciliation_service.py").read_text()
    projection = source[source.index('_LINK_SELECT = """'):source.index('def list_document_links')]
    assert "d.id::text AS id" in projection
    assert "d.id::text AS document_id" in projection
    assert "l.id::text AS link_id" in projection
    for field in ("customer_id", "company", "document_type", "serie", "series_number",
                  "document_number", "version_number", "external_id", "external_url", "role",
                  "is_primary", "is_active", "status", "amount", "tax_amount", "total_amount",
                  "currency", "document_date", "due_date", "payload"):
        assert f"d.{field}" in projection, field


def test_backfill_single_primary_is_unambiguous():
    result = classify_group([_doc("a", role="current", primary=True), _doc("b")])
    assert [r["relationship"] for r in result] == ["PRIMARY", "SECONDARY"]


def test_backfill_two_primaries_requires_review():
    result = classify_group([_doc("a", role="current", primary=True), _doc("b", role="accepted", primary=True)])
    assert {r["relationship"] for r in result} == {"REVIEW_REQUIRED"}


def test_backfill_cancelled_primary_requires_review_and_never_ignored():
    result = classify_group([_doc("a", role="current", primary=True, status="cancelled")])
    assert result[0]["relationship"] == "REVIEW_REQUIRED"
    assert result[0]["cancelled"] is True


def test_detached_without_evidence_is_not_invented_removed():
    assert classify_group([_doc("a", role="detached")])[0]["relationship"] == "REVIEW_REQUIRED"
    assert classify_group([_doc("a", role="detached", payload={"removed_reason": "manual"})])[0]["relationship"] == "REMOVED"


def test_classifier_import_is_isolated_from_application_configuration():
    env = os.environ.copy()
    env.pop("DATABASE_URL", None)
    env.pop("OPENROUTER_API_KEY", None)
    result = subprocess.run(
        [
            sys.executable,
            "-c",
            "import sys; from app.document_reconciliation_backfill import classify_group; "
            "assert callable(classify_group); "
            "assert 'app.config' not in sys.modules; "
            "assert 'app.db' not in sys.modules; "
            "assert 'sqlalchemy' not in sys.modules",
        ],
        cwd=Path(__file__).resolve().parents[1],
        env=env,
        capture_output=True,
        text=True,
        check=False,
    )
    assert result.returncode == 0, result.stderr


def test_schema_has_database_guards_and_append_only_audit():
    sql = Path("migrations/007_document_reconciliation_v2.sql").read_text()
    assert "ux_document_link_primary_kind" in sql and "WHERE ended_at IS NULL AND relationship = 'PRIMARY'" in sql
    assert "ux_document_link_current" in sql
    assert "BEFORE UPDATE OR DELETE" in sql
    assert "ux_document_link_events_idempotency" in sql


def test_active_router_owns_v2_handlers_and_htmx_panel_target():
    router = Path("app/admin_ui/router.py").read_text()
    page = Path("app/admin_ui/pages/opportunities.py").read_text()
    assert "router.include_router(opportunities.router)" in router
    assert "document_reconciliation_v2_action" in page
    assert "HTMLResponse(jasmin_documents_html(opportunity_id" in page
    assert "_authenticated_actor(request)" in page


def test_sync_does_not_assign_legacy_primary_by_import_order():
    source = Path("app/reconciliation_service.py").read_text()
    function = source[source.index("def _upsert_jasmin_document_from_item"):source.index("def _", source.index("def _upsert_jasmin_document_from_item") + 10)]
    assert '"role": "related"' in function
    assert '"is_primary": False' in function
    assert "classify_sync_document" in function
    canonical = Path("app/document_reconciliation_service.py").read_text()
    assert "multiple Jasmin candidates require review" in canonical
    assert "SYNC_AMBIGUITY_DEMOTED" in canonical


def test_total_never_falls_back_to_non_primary_or_cancelled_primary():
    rows = [{"id": "secondary", "relationship": "SECONDARY", "total_amount": 999}]
    assert select_valid_primary(rows) is None
    rows.insert(0, {"id": "cancelled", "relationship": "PRIMARY", "status": "CANCELLED"})
    assert select_valid_primary(rows) is None


def test_only_unambiguous_keeps_ambiguous_group_fully_represented():
    actions = classify_group([_doc("a", role="current", primary=True),
                              _doc("b", role="accepted", primary=True)])
    selected = actions_for_only_unambiguous(actions)
    assert len(selected) == 2
    assert {row["relationship"] for row in selected} == {"REVIEW_REQUIRED"}


def test_backfill_paginates_complete_groups_with_composite_cursor():
    source = Path("scripts/migrate_document_reconciliation_v2.py").read_text()
    assert "WITH selected_groups AS" in source
    assert "GROUP BY d.opportunity_id, d.document_kind" in source
    assert "LIMIT :limit" in source
    assert "OPPORTUNITY_ID|DOCUMENT_KIND" in source
    assert "JOIN commercial_documents d ON d.opportunity_id=g.opportunity_id" in source


def test_group_larger_than_batch_size_is_never_split():
    rows = [{"id": str(index), "opportunity_id": "a", "document_kind": "quotation"}
            for index in range(5)] + [{"id": "z", "opportunity_id": "b", "document_kind": "quotation"}]
    batch = select_complete_group_batch(rows, batch_size=1)
    assert len(batch) == 5 and {row["opportunity_id"] for row in batch} == {"a"}


def test_composite_resume_keeps_other_kind_on_opportunity_boundary():
    rows = [
        {"id": "1", "opportunity_id": "a", "document_kind": "invoice"},
        {"id": "2", "opportunity_id": "a", "document_kind": "quotation"},
        {"id": "3", "opportunity_id": "b", "document_kind": "quotation"},
    ]
    first = select_complete_group_batch(rows, batch_size=1)
    cursor = (first[-1]["opportunity_id"], first[-1]["document_kind"])
    second = select_complete_group_batch(rows, batch_size=1, resume_from=cursor)
    assert [(row["opportunity_id"], row["document_kind"]) for row in second] == [("a", "quotation")]


def test_idempotency_fingerprint_retry_and_conflicting_reuse_contract():
    first, payload = _command_fingerprint(opportunity_id="o", document_id="d",
        target_relationship="PRIMARY", reason="approved")
    retry, _ = _command_fingerprint(opportunity_id="o", document_id="d",
        target_relationship="PRIMARY", reason="approved")
    conflict, _ = _command_fingerprint(opportunity_id="o", document_id="d",
        target_relationship="SECONDARY", reason="approved")
    assert first == retry and first != conflict
    source = Path("app/document_reconciliation_service.py").read_text()
    function = source[source.index("def set_document_relationship"):source.index("def set_primary_document")]
    assert function.index("_claim_idempotency") < function.index("_lock_context")
    assert "already used with a different command" in source


def test_idempotency_fingerprint_covers_every_material_command_field():
    base = dict(opportunity_id="o", document_id="d", target_relationship="PRIMARY",
                reason="approved", source="api", event_type="CHANGED", metadata={"a": 1},
                is_manual=True, actor="alice", document_kind="invoice")
    fingerprint, _ = _command_fingerprint(**base)
    for field, value in (("source", "sync"), ("event_type", "BACKFILLED"),
                         ("metadata", {"a": 2}), ("is_manual", False), ("actor", "bob"),
                         ("correlation_id", "corr-1"), ("request_id", "req-1")):
        changed, _ = _command_fingerprint(**(base | {field: value}))
        assert changed != fingerprint, field
    reordered, _ = _command_fingerprint(**(base | {"metadata": {"z": 2, "a": 1}}))
    same_reordered, _ = _command_fingerprint(**(base | {"metadata": {"a": 1, "z": 2}}))
    assert reordered == same_reordered


def test_generic_document_writer_registers_quotation_and_invoice_through_canonical_service():
    source = Path("app/commercial_service.py").read_text()
    function = source[source.index("def create_commercial_document"):source.index("def add_document_lines")]
    assert "register_created_document(" in function
    assert function.index("INSERT INTO commercial_documents") < function.index("register_created_document(")
    assert "and not v2_available" in function
    assert "UPDATE opportunity_document_links" not in function
    assert "INSERT INTO opportunity_document_links" not in function
    canonical = Path("app/document_reconciliation_service.py").read_text()
    register = canonical[canonical.index("def register_created_document"):canonical.index("def record_jasmin_status_change")]
    assert "document_reconciliation_v2_available(conn)" in register
    assert "classify_sync_document(" in register


def test_new_group_member_preserves_manual_primary_and_uses_secondary_or_review():
    source = Path("app/document_reconciliation_service.py").read_text()
    classify = source[source.index("def classify_sync_document"):source.index("def register_created_document")]
    assert 'relationship = "SECONDARY" if manual_primary' in classify
    assert 'else "REVIEW_REQUIRED"' in classify
    assert "multiple Jasmin candidates require review" in classify
    assert 'and not previous["is_manual"]' in classify


def test_all_commercial_document_producers_register_v2_relationships():
    commercial = Path("app/commercial_service.py").read_text()
    reconciliation = Path("app/reconciliation_service.py").read_text()
    assert commercial.count("INSERT INTO commercial_documents") == 1
    assert "register_created_document(" in commercial
    assert reconciliation.count("INSERT INTO commercial_documents") == 1
    assert "classify_sync_document(" in reconciliation


def test_backfill_applies_one_transaction_per_complete_group():
    source = Path("scripts/migrate_document_reconciliation_v2.py").read_text()
    function = source[source.index("def apply_actions"):source.index("def write_reports")]
    assert "groups.setdefault" in function
    assert function.count("with engine.begin() as conn") == 1
    assert "_conn=conn" in function


def test_backfill_second_document_failure_rolls_back_whole_group(monkeypatch):
    import app.db
    import app.document_reconciliation_service as service

    committed = {"links": [], "events": [], "legacy": []}
    class Transaction:
        def __enter__(self):
            self.pending = {key: [] for key in committed}
            return self
        def __exit__(self, exc_type, exc, tb):
            if exc_type is None:
                for key in committed:
                    committed[key].extend(self.pending[key])
            return False
    class FakeEngine:
        def begin(self): return Transaction()
    calls = 0
    def fake_set(*args, _conn=None, **kwargs):
        nonlocal calls
        calls += 1
        _conn.pending["links"].append(args[1])
        _conn.pending["events"].append(args[1])
        _conn.pending["legacy"].append(args[1])
        if calls == 2:
            raise RuntimeError("simulated second document failure")
    monkeypatch.setattr(app.db, "engine", FakeEngine())
    monkeypatch.setattr(service, "set_document_relationship", fake_set)
    actions = [dict(_doc("a"), relationship="PRIMARY", decision_reason="one"),
               dict(_doc("b"), relationship="SECONDARY", decision_reason="two")]
    with pytest.raises(RuntimeError, match="second document"):
        apply_actions(actions)
    assert committed == {"links": [], "events": [], "legacy": []}


def test_generic_line_writer_is_schema_compatible_and_keeps_ids_coherent():
    source = Path("app/commercial_service.py").read_text()
    function = source[source.index("def add_document_lines"):source.index("def list_commercial_documents")]
    assert "to_regclass('commercial_document_lines')" in function
    assert "FROM pg_attribute" in function
    assert "attname = 'commercial_document_id'" in function
    assert "table_schema='public'" not in function
    assert 'CAST(:document_id AS UUID), CAST(:document_id AS UUID)' in function


def test_startup_does_not_partially_apply_migration_007():
    source = Path("app/commercial_service.py").read_text()
    function = source[source.index("def ensure_commercial_schema"):source.index("def get_customer_by_tax_id")]
    assert "ADD COLUMN IF NOT EXISTS commercial_document_id" not in function
    assert "ADD COLUMN IF NOT EXISTS opportunity_document_link_id" not in function


def test_no_new_authoritative_v2_reader_outside_canonical_service():
    allowlist = {"document_reconciliation_service.py"}
    offenders = []
    for path in Path("app").rglob("*.py"):
        if path.name in allowlist:
            continue
        if "opportunity_document_links" in path.read_text(encoding="utf-8"):
            offenders.append(str(path))
    # Remaining consumers are deliberately tracked until migrated; additions
    # outside this explicit inventory fail the review gate.
    assert offenders == []


def test_membership_and_manual_reason_are_domain_guards():
    source = Path("app/document_reconciliation_service.py").read_text()
    assert "document does not belong to this opportunity" in source
    assert "reason is required for every manual relationship change" in source
    assert "commercial_documents WHERE id=CAST(:did AS UUID)" in source


def test_preflight_and_down_migration_cover_blockers_and_dependency_order():
    assert callable(run_preflight)
    preflight = Path("scripts/preflight_document_reconciliation_v2.py").read_text()
    assert "null_document_id" in preflight and "orphan_document_id" in preflight
    assert "multiple_legacy_primaries" in preflight and "duplicate_document_groups" in preflight
    down = Path("migrations/007_document_reconciliation_v2_down.sql").read_text()
    assert down.index("DROP COLUMN IF EXISTS opportunity_document_link_id") < down.index("DROP TABLE IF EXISTS opportunity_document_links")
    assert down.index("DROP TRIGGER") < down.index("DROP FUNCTION") < down.index("DROP TABLE IF EXISTS opportunity_document_link_events")
    assert "document_ledger_exported" in down and "v2_consumers_active" in down


def test_jasmin_items_keep_document_and_link_provenance():
    source = Path("app/reconciliation_service.py").read_text()
    assert '"commercial_document_id": origin.get("commercial_document_id")' in source
    assert '"opportunity_document_link_id": origin.get("opportunity_document_link_id")' in source