"""THE M6 LIFECYCLE GATE (09-build-plan M6): the full invoice cycle on
REAL migrated data with REAL commits —

  create → apply 5 real pending entries → locked (non-owner edit 409s,
  re-stamp skips them, bulk delete refuses) → paste block renders →
  detach → unlocked → delete invoice → HARNESS COMPARE green
  (doc_before == doc_after, meta-only diffs)

— proving the lock write-side and that a complete apply/detach cycle
leaves the books byte-identical (same shape as the M5 edit+undo gate).
"""

import uuid

import psycopg
import pytest
from sqlalchemy import func, select
from sqlalchemy.orm import Session

from app.models import Invoice, InvoiceLine, TimeEntry
from app.services.bulk import BulkSelection, bulk_delete, bulk_recalculate
from app.services.entries_write import require_can_edit
from app.services.errors import ServiceError
from app.services.money import invoices as inv
from app.services.money.paste_block import paste_block_for_invoice
from app.services.reporting.entries import entry_totals
from app.services.reporting.filters import EntryFilters, status_condition
from harness import extract_new
from harness.compare import DEFAULT_ALLOWLIST, load_allowlist, run_compare
from tests.invoices.conftest import actor_named, pending_ids

ALLOWLIST = load_allowlist(DEFAULT_ALLOWLIST)


def _extract(m6_url: str) -> dict:
    dsn = m6_url.replace("postgresql+psycopg://", "postgresql://", 1)
    with psycopg.connect(dsn) as conn:
        return extract_new.extract(conn)


def test_gate_full_invoice_lifecycle_harness_green(m6_engine, m6_url):
    doc_before = _extract(m6_url)

    with Session(m6_engine) as s:
        rian = actor_named(s, "rian")
        adi = actor_named(s, "adi")
        partition_before = {
            status: entry_totals(
                s, rian, EntryFilters(status=status), zero_currency="CAD"
            )["row_count"]
            for status in ("pending", "blocked", "applied")
        }

        # -- create -------------------------------------------------------
        row = inv.create_invoice(s, rian, {"name": "M6 lifecycle gate"})
        iid = uuid.UUID(row["id"])

        # -- apply 5 REAL pending entries ----------------------------------
        ids = pending_ids(s, 5)
        assert len(ids) == 5
        result = inv.apply_entries(
            s, rian, iid, BulkSelection(expected_count=5, ids=list(ids))
        )
        assert result["applied"] == 5
        assert result["matched"] == 5
        assert result["skipped_already_applied"] == 0
        assert result["skipped_blocked"] == 0

        # -- locked: every write path refuses / skips ----------------------
        for entry_id in ids:
            entry = s.get(TimeEntry, entry_id)
            assert entry.invoice_line_id is not None
            with pytest.raises(ServiceError) as exc:  # non-owner edit 409s
                require_can_edit(adi, entry, "edit")
            assert exc.value.status == 409 and exc.value.code == "ENTRY_LOCKED"

        recalc = bulk_recalculate(  # re-stamp skips locked for EVERYONE
            s, rian, BulkSelection(expected_count=5, ids=list(ids))
        )
        assert recalc["skipped_locked"] == 5
        assert recalc["changed"] == 0

        with pytest.raises(ServiceError) as exc:  # bulk delete refuses
            bulk_delete(s, rian, BulkSelection(expected_count=5, ids=list(ids)))
        assert exc.value.code == "ALL_LOCKED"
        s.rollback()

        # the display partition agrees: 5 moved pending -> applied
        assert (
            entry_totals(s, rian, EntryFilters(status="pending"), zero_currency="CAD")[
                "row_count"
            ]
            == partition_before["pending"] - 5
        )
        assert (
            entry_totals(s, rian, EntryFilters(status="applied"), zero_currency="CAD")[
                "row_count"
            ]
            == partition_before["applied"] + 5
        )

        # -- paste block renders from the snapshots -------------------------
        block = paste_block_for_invoice(s, rian, iid)
        assert block["line_count"] >= 1
        for line in block["text"].split("\n"):
            assert len(line.split("\t")) == 7

        # -- detach → unlocked ----------------------------------------------
        detached = inv.detach_entries(
            s, rian, ids=list(ids), expected_count=5, invoice_id=iid
        )
        assert detached["detached"] == 5
        for entry_id in ids:
            entry = s.get(TimeEntry, entry_id)
            assert entry.invoice_line_id is None
            require_can_edit(adi, entry, "edit")  # no raise: unlocked again
            assert s.scalar(
                select(func.count())
                .select_from(TimeEntry)
                .where(TimeEntry.id == entry_id, status_condition("pending"))
            ) == 1

        assert (
            s.scalar(
                select(func.count())
                .select_from(InvoiceLine)
                .where(InvoiceLine.invoice_id == iid)
            )
            == 0
        )

        # -- delete the now-empty invoice ------------------------------------
        inv.delete_invoice(s, rian, iid)
        assert s.get(Invoice, iid) is None

        # partition fully restored
        partition_after = {
            status: entry_totals(
                s, rian, EntryFilters(status=status), zero_currency="CAD"
            )["row_count"]
            for status in ("pending", "blocked", "applied")
        }
        assert partition_after == partition_before

    # -- the harness verdict: green after the full cycle ---------------------
    doc_after = _extract(m6_url)
    report = run_compare(doc_before, doc_after, ALLOWLIST)
    failures = [(d.path, d.old, d.new) for d in report.failures]
    assert report.ok, f"harness RED after the apply/detach lifecycle: {failures}"
    assert all(d.path.startswith("meta/") for d in report.diffs if d.allowed)