"""THE M6 BYTE-DIFF GATE (09-build-plan M6): for every REAL migrated
invoice, the paste block produced by the NEW Python path
(services/money/invoices.invoice_breakdown + paste_block over migrated
invoice_lines snapshots) must be BYTE-IDENTICAL to the one produced by
the LEGACY code of record (/srv/apps/work/lib/invoices.ts —
getInvoiceProjectBreakdown's exact grouping + localeCompare sort +
formatPasteBlock), fed the same invoice's raw LEGACY rows and executed
through the old app's tsx (scripts/ts_paste_block.mjs).

The oracle rows come from the scratch clone's ``legacy`` schema — the
same real snapshot the migration transformed — so this cross-validates
the ENTIRE chain: migration snapshots, SQL aggregation, sort, tenant
settings, and byte formatting.
"""

import json
import subprocess
import uuid
from pathlib import Path

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

from app.models import Invoice
from app.services.money.paste_block import paste_block_for_invoice
from tests.invoices.conftest import actor_named

TSX = Path("/srv/apps/work/node_modules/.bin/tsx")
TS_RUNNER = Path(__file__).resolve().parents[2] / "scripts" / "ts_paste_block.mjs"

ORACLE_SQL = text(
    "SELECT client, project, duration_seconds, billout_cost_usd,"
    " billout_amount_usd FROM legacy.time_entries WHERE invoice_id = :iid"
)

#: The 'Legacy transfers' sentinel: the migration's straggler rule
#: (judgment #22) attaches transferred-but-not-invoiced rows to the
#: EXISTING sentinel (the real data has exactly one — the Villa del Mar
#: row flagged for rian's review in BUILD-STATE). The oracle emulates
#: the same membership so the byte-diff compares like for like.
ORACLE_SQL_SENTINEL = text(
    "SELECT client, project, duration_seconds, billout_cost_usd,"
    " billout_amount_usd FROM legacy.time_entries"
    " WHERE invoice_id = :iid"
    " OR (transferred_at IS NOT NULL AND invoice_id IS NULL)"
)


def _oracle_paste_block(rows: list[dict], invoice_date: str, tmp_path: Path) -> dict:
    if not TSX.exists():
        pytest.fail(
            f"the byte-diff gate needs the legacy tsx runner at {TSX} —"
            " run make check on the host (the old app's node_modules)"
        )
    payload = tmp_path / "rows.json"
    payload.write_text(
        json.dumps({"rows": rows, "invoice_date": invoice_date}), encoding="utf-8"
    )
    proc = subprocess.run(
        [str(TSX), str(TS_RUNNER), str(payload)],
        capture_output=True,
        text=True,
        timeout=120,
    )
    assert proc.returncode == 0, f"tsx oracle failed: {proc.stderr[-2000:]}"
    return json.loads(proc.stdout)


def _legacy_rows(
    session: Session, invoice_id: uuid.UUID, *, sentinel: bool
) -> list[dict]:
    out = []
    sql = ORACLE_SQL_SENTINEL if sentinel else ORACLE_SQL
    for client, project, seconds, cost, amount in session.execute(
        sql, {"iid": str(invoice_id)}
    ).all():
        out.append(
            {
                "client": client,
                "project": project,
                "duration_seconds": seconds,
                # numerics as strings — exactly how PostgREST hands them
                # to the legacy code (its shape() does Number())
                "billout_cost_usd": str(cost) if cost is not None else None,
                "billout_amount_usd": str(amount) if amount is not None else None,
            }
        )
    return out


def test_gate_paste_block_byte_identical_for_all_real_invoices(m6_engine, tmp_path):
    with Session(m6_engine) as s:
        rian = actor_named(s, "rian")
        invoices = s.scalars(
            select(Invoice).order_by(Invoice.created_at)
        ).all()
        real = [i for i in invoices if i.name.startswith(("Legacy", "2026"))]
        assert len(real) >= 3, "expected the 3 real migrated invoices"

        verdicts = []
        for invoice in real:
            new_side = paste_block_for_invoice(s, rian, invoice.id)
            legacy_rows = _legacy_rows(
                s, invoice.id, sentinel=invoice.name == "Legacy transfers"
            )
            oracle = _oracle_paste_block(
                legacy_rows, new_side["invoice_date"], tmp_path
            )
            assert new_side["line_count"] == oracle["line_count"], (
                f"{invoice.name}: breakdown row counts diverge "
                f"(new {new_side['line_count']} vs legacy {oracle['line_count']})"
            )
            assert new_side["text"] == oracle["text"], _diff_message(
                invoice.name, new_side["text"], oracle["text"]
            )
            verdicts.append((invoice.name, new_side["line_count"]))

        # the gate is only meaningful if it exercised real content
        assert sum(count for _, count in verdicts) > 0


def _diff_message(name: str, new_text: str, old_text: str) -> str:
    new_lines, old_lines = new_text.split("\n"), old_text.split("\n")
    for i, (a, b) in enumerate(zip(new_lines, old_lines, strict=False)):
        if a != b:
            return (
                f"{name}: first divergent line {i}:\n  new: {a!r}\n  old: {b!r}"
            )
    return f"{name}: line counts differ (new {len(new_lines)} vs old {len(old_lines)})"