"""THE M5 GATE (09-build-plan): a scripted replay of the 3,274-rows-
from-a-353-row-selection incident SHAPE against migrated data, plus
the same-filter-scope property against the REAL migrated sidecar
(read-only — 4,593 production rows).

The incident replay (real commits — the undo restores the database
byte-identically, which is the point):

  1. extract the canonical harness document (doc_before);
  2. filter a SUBSET (status=pending + operator=PlusROI), confirm its
     exact count, bulk-edit the worker over all-matching;
  3. verify ONLY those rows changed (full-table snapshot diff);
  4. verify the expected-count guard fires on a stale count;
  5. undo — verify byte-identical restoration per row;
  6. extract again (doc_after) and run the HARNESS COMPARE:
     doc_before == doc_after (meta-only diffs) -> harness green after
     an edit+undo cycle.
"""

import uuid

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

from app.models import Party, TimeEntry
from app.services.audit import snapshot_entry, undo_entries_action
from app.services.bulk import BulkSelection, bulk_edit_fields, collect_scope
from app.services.errors import ServiceError
from app.services.reporting.entries import entry_totals, list_entries
from app.services.reporting.filters import EntryFilters
from harness import extract_new
from harness.compare import DEFAULT_ALLOWLIST, load_allowlist, run_compare
from tests.bulk.conftest import TE, actor_named

ALLOWLIST = load_allowlist(DEFAULT_ALLOWLIST)

INCIDENT_FILTER = EntryFilters(status="pending", operator="PlusROI")


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


def test_gate_incident_replay_and_harness_green_after_undo(m5_engine, m5_url):
    doc_before = _extract(m5_url)

    with Session(m5_engine) as s:
        rian = actor_named(s, "rian")
        gary = s.scalars(
            select(Party.id).where(Party.kind == "person", Party.email == "gary@rian.ca")
        ).one()

        # -- 1. the display defines the subset --------------------------------
        page = list_entries(s, rian, INCIDENT_FILTER)
        subset_ids = {row["id"] for row in page.rows}
        expected = page.total_count
        total_all = entry_totals(s, rian, EntryFilters())["row_count"]
        assert 0 < expected < total_all  # a strict subset — the incident shape

        before = {
            str(r.id): snapshot_entry(r)
            for r in s.scalars(select(TimeEntry).order_by(TimeEntry.id))
        }

        # -- 2. the stale-count guard fires BEFORE any write ------------------
        with pytest.raises(ServiceError) as stop:
            bulk_edit_fields(
                s,
                rian,
                BulkSelection(expected_count=expected + 1, filters=INCIDENT_FILTER),
                worker_party_id=gary,
            )
        assert stop.value.code == "SAFETY_STOP"
        s.rollback()

        # -- 3. the real edit: verified counts --------------------------------
        result = bulk_edit_fields(
            s,
            rian,
            BulkSelection(expected_count=expected, filters=INCIDENT_FILTER),
            worker_party_id=gary,
        )
        assert result["matched"] == expected
        assert result["changed"] == expected

        # -- 4. ONLY the subset changed ----------------------------------------
        after = {
            str(r.id): snapshot_entry(r)
            for r in s.scalars(select(TimeEntry).order_by(TimeEntry.id))
        }
        assert set(after) == set(before)
        for entry_id in before:
            if entry_id in subset_ids:
                assert after[entry_id] != before[entry_id]
                assert after[entry_id]["worker_party_id"] == str(gary)
            else:
                assert after[entry_id] == before[entry_id]

        # -- 5. undo: byte-identical restoration -------------------------------
        undo = undo_entries_action(s, rian, uuid.UUID(result["undo_id"]))
        assert undo["restored"] == expected
        assert undo["skipped_locked"] == 0
        restored = {
            str(r.id): snapshot_entry(r)
            for r in s.scalars(select(TimeEntry).order_by(TimeEntry.id))
        }
        assert restored == before

    # -- 6. the harness verdict: green after the edit+undo cycle ---------------
    doc_after = _extract(m5_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 edit+undo: {failures}"
    assert all(d.path.startswith("meta/") for d in report.diffs if d.allowed)


def test_gate_scope_property_on_real_production_data(real_session, real_rian):
    """Same-filter-scope over the REAL migrated sidecar (read-only):
    for a spread of filter shapes — including the incident's
    status+operator shape — |scope| == display count and (spot-checked)
    the id sets match page-for-page."""
    combos = [
        EntryFilters(status="pending"),
        EntryFilters(status="pending", operator="PlusROI"),  # the incident shape
        EntryFilters(status="applied"),
        EntryFilters(status="blocked"),
        EntryFilters(status="all", operator="Bowden Works"),
        EntryFilters(status="all", q="support"),
        EntryFilters(status="pending", client="o"),
        EntryFilters(status="all", user="a"),
    ]
    for f in combos:
        n = entry_totals(real_session, real_rian, f)["row_count"]
        rows = collect_scope(
            real_session, real_rian, BulkSelection(expected_count=n, filters=f)
        )
        assert len(rows) == n, f"scope/display mismatch for {f}"

    # exact id-set equality, all pages, for the incident shape
    f = EntryFilters(status="pending", operator="PlusROI")
    n = entry_totals(real_session, real_rian, f)["row_count"]
    display_ids: set[str] = set()
    page_no = 1
    while True:
        page = list_entries(real_session, real_rian, f, page=page_no)
        display_ids.update(row["id"] for row in page.rows)
        if page_no >= page.total_pages:
            break
        page_no += 1
    assert len(display_ids) == n
    rows = collect_scope(real_session, real_rian, BulkSelection(expected_count=n, filters=f))
    assert {str(r.id) for r in rows} == display_ids


def test_gate_fixture_subset_is_pinned(s, rian):
    """Keep the miniature incident anchored: the fixture's pending ∩
    PlusROI subset is exactly te-2 + te-7 (2 of 7 rows)."""
    page = list_entries(s, rian, INCIDENT_FILTER)
    assert {row["id"] for row in page.rows} == {str(TE[2]), str(TE[7])}
    assert entry_totals(s, rian, EntryFilters())["row_count"] == 7