#!/usr/bin/env python3
"""Reconcile the 2026 legacy YM return placeholders without double counting.

The old PHP collector made one ``mp_returns`` row for every ``items[]`` member
and used ``<API id>_<item.marketSku>`` as its return id.  The current kernel
uses the API id itself.  Therefore an API projection must be published *before*
this tool may archive and remove the old placeholder from the active fact.

This is deliberately a controller, not an ETL shortcut:

* default operation is read-only, apart from writing the local CSV evidence;
* the CSV gives one independently checkable proof (or rejection reason) per
  old fact row;
* ``--apply`` refuses to touch any row until every selected mapping has one
  non-legacy projected fact;
* the old fact is copied verbatim to an append-only supersession table before
  it is removed from the active fact table.

It never changes ``lamoda_reports.mp_returns``.  It is intended to be run only
after the reviewed recovery input and the complete ``build_fact_returns``
projection have succeeded.
"""
from __future__ import annotations

import argparse
import csv
import json
import os
import sys
import uuid
from collections import defaultdict
from dataclasses import dataclass
from datetime import date, datetime
from decimal import Decimal, InvalidOperation
from pathlib import Path
from typing import Any, Iterable, Mapping, Sequence

import pymysql
import pymysql.cursors


ROOT = Path(__file__).resolve().parents[1]
REPO_ROOT = ROOT.parent
LEGACY = "lamoda_reports"
INTAKE = "gwptd_intake"
KERNEL = "gwptd_kernel"

# These are facts established by the preceding W1.1 dry-run.  A changed count
# is not a reason to guess; it is a reason to stop and investigate again.
EXPECTED_MAPPINGS = 861
EXPECTED_PROTECTED_OVERLAPS = 2
SUPERSESSION_TABLE = f"{KERNEL}.fact_returns_supersession"
DEFAULT_MAPPING_PATH = (
    REPO_ROOT / "docs/remediation/agent-exchange/results/A1-2-supersede-mapping.csv"
)


class SupersessionBlocked(RuntimeError):
    """A condition makes an active-fact replacement unsafe."""


@dataclass(frozen=True)
class MappingRow:
    """One old fact and the evidence used to decide whether it is replaceable."""

    old_fact_id: int
    mp_id: int
    model_id: int
    legacy_row_id: int
    legacy_return_id: str
    api_return_id: str
    api_order_id: str
    creation_date: str
    legacy_offer_id: str
    market_sku: str
    shop_sku: str
    item_line_no: int | None
    raw_refund_amount: Decimal | None
    state: str
    reason: str
    method: str

    @property
    def target_key(self) -> tuple[int, str, int]:
        return (self.mp_id, self.api_return_id, self.model_id)


@dataclass(frozen=True)
class ProjectionTarget:
    fact_id: int
    mp_id: int
    return_id: str
    model_id: int
    return_date: Any
    amount: Decimal | None
    quantity: Any
    source_payload_id: Any


@dataclass(frozen=True)
class ProjectionCheck:
    targets: Mapping[tuple[int, str, int], ProjectionTarget]
    failures: tuple[str, ...]

    @property
    def passed(self) -> bool:
        return not self.failures


# The source side is deliberately item-granular.  Mapping a return header to
# every model of an order is forbidden: a multi-item order would multiply both
# returns and money.  All cross-schema comparisons pin both operands to the
# known common collation; S3 otherwise raises ERROR 1267.
MAPPING_EVIDENCE_SQL = f"""
WITH legacy_rows AS (
    SELECT l.id AS legacy_row_id,
           l.marketplace_id AS mp_id,
           l.return_id AS legacy_return_id,
           l.order_id AS legacy_order_id,
           l.offer_id AS legacy_offer_id,
           l.raw_json,
           JSON_UNQUOTE(JSON_EXTRACT(l.raw_json, '$.id')) AS api_return_id,
           JSON_UNQUOTE(JSON_EXTRACT(l.raw_json, '$.orderId')) AS api_order_id,
           DATE(JSON_UNQUOTE(JSON_EXTRACT(l.raw_json, '$.creationDate'))) AS creation_date,
           CAST(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(l.raw_json, '$.amount.value')), '')
                AS DECIMAL(14,2)) AS raw_refund_amount
    FROM {LEGACY}.mp_returns AS l
    WHERE l.marketplace_id IN (6, 8)
      AND l.amount IS NULL
      AND l.return_date IS NULL
), old_facts AS (
    -- mp_id приходит из legacy_rows.*; повторять f.mp_id нельзя — MySQL
    -- отвергает CTE с одноимёнными колонками (ERROR 1060). Значения равны:
    -- JOIN ниже связывает их равенством.
    SELECT f.id AS old_fact_id,
           f.model_id,
           legacy_rows.*,
           EXISTS(
               SELECT 1
               FROM {INTAKE}.ym_returns AS working
               WHERE working.order_id COLLATE utf8mb4_unicode_ci
                   = legacy_rows.api_order_id COLLATE utf8mb4_unicode_ci
           ) AS overlaps_working
    FROM {KERNEL}.fact_returns AS f
    JOIN legacy_rows
      ON legacy_rows.mp_id = f.mp_id
     AND legacy_rows.legacy_return_id COLLATE utf8mb4_unicode_ci
         = f.return_id COLLATE utf8mb4_unicode_ci
    WHERE f.source_payload_id IS NULL
      AND f.return_date IS NULL
      AND f.amount IS NULL
)
SELECT old_facts.old_fact_id,
       old_facts.mp_id,
       old_facts.model_id,
       old_facts.legacy_row_id,
       old_facts.legacy_return_id,
       old_facts.legacy_order_id,
       old_facts.legacy_offer_id,
       old_facts.api_return_id,
       old_facts.api_order_id,
       old_facts.creation_date,
       old_facts.raw_refund_amount,
       old_facts.overlaps_working,
       item.line_no AS item_line_no,
       item.market_sku,
       item.shop_sku,
       product.model_id AS item_model_id
FROM old_facts
LEFT JOIN JSON_TABLE(
    JSON_EXTRACT(old_facts.raw_json, '$'),
    '$.items[*]' COLUMNS (
        line_no FOR ORDINALITY,
        market_sku VARCHAR(128) PATH '$.marketSku' NULL ON EMPTY NULL ON ERROR,
        shop_sku VARCHAR(255) PATH '$.shopSku' NULL ON EMPTY NULL ON ERROR
    )
) AS item ON TRUE
LEFT JOIN {KERNEL}.dim_product AS product
  ON product.artikul_upper = UPPER(TRIM(item.shop_sku)) COLLATE utf8mb4_unicode_ci
ORDER BY old_facts.old_fact_id, item.line_no, product.model_id
"""

SUPERSESSION_DDL = f"""
CREATE TABLE IF NOT EXISTS {SUPERSESSION_TABLE} (
    id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
    batch_id CHAR(36) CHARACTER SET ascii NOT NULL,
    old_fact_id BIGINT UNSIGNED NOT NULL,
    projected_fact_id BIGINT UNSIGNED NOT NULL,
    mp_id TINYINT UNSIGNED NOT NULL,
    model_id INT UNSIGNED NOT NULL,
    legacy_row_id BIGINT UNSIGNED NOT NULL,
    legacy_return_id VARCHAR(128) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL,
    api_return_id VARCHAR(128) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL,
    api_order_id VARCHAR(128) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL,
    creation_date DATE NULL,
    old_fact_snapshot JSON NOT NULL,
    mapping_evidence JSON NOT NULL,
    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
    PRIMARY KEY (id),
    UNIQUE KEY uq_fact_returns_supersession_old_fact (old_fact_id),
    KEY idx_fact_returns_supersession_batch (batch_id),
    KEY idx_fact_returns_supersession_target (mp_id, api_return_id, model_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
"""

OLD_FACT_SNAPSHOT_FIELDS = (
    "id, model_id, mp_id, return_id, return_date, order_date, order_id, amount, "
    "quantity, reason, status, source_payload_id, created_at"
)


def _text(value: Any) -> str:
    return str(value or "").strip()


def _as_int(value: Any, field: str) -> int:
    try:
        return int(value)
    except (TypeError, ValueError) as exc:
        raise SupersessionBlocked(f"mapping evidence has invalid {field}: {value!r}") from exc


def _as_decimal(value: Any) -> Decimal | None:
    if value is None or str(value).strip() == "":
        return None
    try:
        return Decimal(str(value))
    except (InvalidOperation, ValueError) as exc:
        raise SupersessionBlocked(f"mapping evidence has invalid amount: {value!r}") from exc


def _item_is_exact(source: Mapping[str, Any]) -> bool:
    """Verify the legacy collector formula and the exact source item/model."""
    api_return_id = _text(source.get("api_return_id"))
    market_sku = _text(source.get("market_sku"))
    item_model_id = source.get("item_model_id")
    if item_model_id is None:
        return False
    return (
        bool(api_return_id)
        and bool(market_sku)
        and _text(source.get("legacy_return_id")) == f"{api_return_id}_{market_sku}"
        and _text(source.get("legacy_order_id")) == _text(source.get("api_order_id"))
        and _text(source.get("legacy_offer_id")) == _text(source.get("shop_sku"))
        and _as_int(source.get("model_id"), "model_id")
        == _as_int(item_model_id, "item_model_id")
    )


def build_mapping(source_rows: Iterable[Mapping[str, Any]]) -> list[MappingRow]:
    """Classify every old candidate fact; do not silently discard failures."""
    grouped: dict[int, list[Mapping[str, Any]]] = defaultdict(list)
    for source in source_rows:
        grouped[_as_int(source.get("old_fact_id"), "old_fact_id")].append(source)

    mapping: list[MappingRow] = []
    for old_fact_id, rows in sorted(grouped.items()):
        first = rows[0]
        overlaps_working = bool(first.get("overlaps_working"))
        exact_items = [row for row in rows if _item_is_exact(row)]
        api_return_id = _text(first.get("api_return_id"))
        api_order_id = _text(first.get("api_order_id"))
        state = "mapped"
        reason = ""
        item: Mapping[str, Any] = exact_items[0] if len(exact_items) == 1 else first
        if overlaps_working:
            state = "protected"
            reason = "forbidden-working-orderId-overlap"
        elif not api_return_id or not api_order_id or not _text(first.get("creation_date")):
            state = "unmapped"
            reason = "missing-api-return-order-or-creation-date"
        elif len(exact_items) != 1:
            state = "unmapped"
            reason = f"exact-item-evidence-count={len(exact_items)}"

        mapping.append(
            MappingRow(
                old_fact_id=old_fact_id,
                mp_id=_as_int(first.get("mp_id"), "mp_id"),
                model_id=_as_int(first.get("model_id"), "model_id"),
                legacy_row_id=_as_int(first.get("legacy_row_id"), "legacy_row_id"),
                legacy_return_id=_text(first.get("legacy_return_id")),
                api_return_id=api_return_id,
                api_order_id=api_order_id,
                creation_date=_text(first.get("creation_date")),
                legacy_offer_id=_text(first.get("legacy_offer_id")),
                market_sku=_text(item.get("market_sku")),
                shop_sku=_text(item.get("shop_sku")),
                item_line_no=(
                    _as_int(item.get("item_line_no"), "item_line_no")
                    if item.get("item_line_no") is not None
                    else None
                ),
                raw_refund_amount=_as_decimal(first.get("raw_refund_amount")),
                state=state,
                reason=reason,
                method=(
                    "api.id + '_' + items[].marketSku; "
                    "legacy.order_id = api.orderId; "
                    "legacy.offer_id = items[].shopSku; "
                    "dim_product(model) = old fact model"
                ),
            )
        )
    return mapping


def _by_state(mapping: Sequence[MappingRow], state: str) -> list[MappingRow]:
    return [row for row in mapping if row.state == state]


def _mapping_gate(mapping: Sequence[MappingRow]) -> tuple[str, ...]:
    mapped = _by_state(mapping, "mapped")
    protected = _by_state(mapping, "protected")
    unmapped = _by_state(mapping, "unmapped")
    failures: list[str] = []
    if unmapped:
        failures.append(f"{len(unmapped)} old fact rows lack exactly one item-level proof")
    if len(mapped) != EXPECTED_MAPPINGS:
        failures.append(
            f"mapped rows={len(mapped)}, expected fixed W1.1 count={EXPECTED_MAPPINGS}"
        )
    if len(protected) != EXPECTED_PROTECTED_OVERLAPS:
        failures.append(
            "protected working-orderId overlaps="
            f"{len(protected)}, expected fixed W1.1 count={EXPECTED_PROTECTED_OVERLAPS}"
        )
    return tuple(failures)


def _target_query(mapped: Sequence[MappingRow]) -> tuple[str, tuple[Any, ...]]:
    if not mapped:
        return "SELECT NULL WHERE FALSE", ()
    placeholders = ", ".join("(%s, %s, %s)" for _ in mapped)
    values: list[Any] = []
    for row in mapped:
        values.extend(row.target_key)
    return (
        f"""
SELECT id, mp_id, return_id, model_id, return_date, amount, quantity, source_payload_id
FROM {KERNEL}.fact_returns
WHERE (mp_id, return_id, model_id) IN ({placeholders})
""",
        tuple(values),
    )


def projection_check(cursor: Any, mapped: Sequence[MappingRow]) -> ProjectionCheck:
    """Require one complete, provenance-backed fact for every replacement."""
    sql, values = _target_query(mapped)
    cursor.execute(sql, values)
    target_rows = list(cursor.fetchall())
    by_key: dict[tuple[int, str, int], list[Mapping[str, Any]]] = defaultdict(list)
    for target in target_rows:
        key = (
            _as_int(target.get("mp_id"), "target mp_id"),
            _text(target.get("return_id")),
            _as_int(target.get("model_id"), "target model_id"),
        )
        by_key[key].append(target)

    targets: dict[tuple[int, str, int], ProjectionTarget] = {}
    failures: list[str] = []
    for row in mapped:
        candidates = by_key[row.target_key]
        if len(candidates) != 1:
            failures.append(
                f"old fact {row.old_fact_id}: projected API fact count={len(candidates)}"
            )
            continue
        target = candidates[0]
        missing = [
            name
            for name in ("return_date", "amount", "quantity", "source_payload_id")
            if target.get(name) is None
        ]
        if missing:
            failures.append(
                f"old fact {row.old_fact_id}: projection is incomplete ({', '.join(missing)})"
            )
            continue
        targets[row.target_key] = ProjectionTarget(
            fact_id=_as_int(target.get("id"), "target fact id"),
            mp_id=_as_int(target.get("mp_id"), "target mp_id"),
            return_id=_text(target.get("return_id")),
            model_id=_as_int(target.get("model_id"), "target model_id"),
            return_date=target.get("return_date"),
            amount=_as_decimal(target.get("amount")),
            quantity=target.get("quantity"),
            source_payload_id=target.get("source_payload_id"),
        )
    return ProjectionCheck(targets=targets, failures=tuple(failures))


def _monthly_rows(
    mapped: Sequence[MappingRow], targets: Mapping[tuple[int, str, int], ProjectionTarget]
) -> list[dict[str, Any]]:
    monthly: dict[str, dict[str, Any]] = {}
    for row in mapped:
        month = row.creation_date[:7] if len(row.creation_date) >= 7 else "unknown"
        bucket = monthly.setdefault(
            month,
            {
                "month": month,
                "replacements": 0,
                "raw_refund_amount": Decimal("0"),
                "projected_fact_amount": Decimal("0"),
                "projected_targets": 0,
            },
        )
        bucket["replacements"] += 1
        bucket["raw_refund_amount"] += row.raw_refund_amount or Decimal("0")
        target = targets.get(row.target_key)
        if target is not None:
            bucket["projected_targets"] += 1
            bucket["projected_fact_amount"] += target.amount or Decimal("0")
    return [monthly[key] for key in sorted(monthly)]


def _format_amount(value: Decimal) -> str:
    return f"{value.quantize(Decimal('0.01')):,.2f}".replace(",", " ")


def print_plan(
    mapping: Sequence[MappingRow],
    projection: ProjectionCheck,
    *,
    apply: bool,
    empty_fact_rows: int | None = None,
) -> None:
    mapped = _by_state(mapping, "mapped")
    protected = _by_state(mapping, "protected")
    unmapped = _by_state(mapping, "unmapped")
    print(f"YM legacy return supersession: {'APPLY' if apply else 'DRY RUN'}")
    if empty_fact_rows is not None:
        # Три числа, а не одно: сколько пустышек всего, сколько из них попало
        # в выборку и сколько осталось за её пределами (F-88).
        outside = empty_fact_rows - len(mapping)
        print(
            "coverage: "
            f"empty-fact-rows={empty_fact_rows}, in-mapping={len(mapping)}, "
            f"outside-mapping={outside}"
        )
        if outside:
            print(
                f"  ВНИМАНИЕ: {outside} пустых строк факта не нашли пары в зеркале "
                "легаси и НЕ будут замещены этим прогоном"
            )
    print(
        "mapping: "
        f"replace={len(mapped)}, protected-overlap={len(protected)}, unmapped={len(unmapped)}"
    )
    raw_total = sum((row.raw_refund_amount or Decimal("0")) for row in mapped)
    print(f"would replace raw YM refund amount: {_format_amount(raw_total)} RUB")
    print(
        "projection facts: "
        f"{len(projection.targets)}/{len(mapped)} complete, "
        f"failures={len(projection.failures)}"
    )
    print("monthly direct-count plan (before=old placeholder + API fact; after=API fact):")
    for bucket in _monthly_rows(mapped, projection.targets):
        count = bucket["replacements"]
        projected = bucket["projected_targets"]
        print(
            f"  {bucket['month']}: before_rows={count + projected}, "
            f"after_expected_rows={count}, replacements={count}, "
            f"raw_refund={_format_amount(bucket['raw_refund_amount'])}, "
            f"projected_fact_amount={_format_amount(bucket['projected_fact_amount'])}"
        )
    gates = (*_mapping_gate(mapping), *projection.failures)
    if gates:
        print("APPLY BLOCKED:")
        for failure in gates:
            print(f"  - {failure}")
    else:
        print("APPLY PRECHECK: PASS")


def write_mapping_csv(mapping: Sequence[MappingRow], path: Path) -> None:
    """Write the eye-reviewable evidence artifact; no raw JSON is duplicated."""
    path.parent.mkdir(parents=True, exist_ok=True)
    fields = [
        "state",
        "reason",
        "old_fact_id",
        "mp_id",
        "model_id",
        "legacy_row_id",
        "legacy_return_id",
        "api_return_id",
        "api_order_id",
        "creation_date",
        "legacy_offer_id",
        "market_sku",
        "shop_sku",
        "item_line_no",
        "raw_refund_amount",
        "method",
    ]
    with path.open("w", encoding="utf-8", newline="") as stream:
        writer = csv.DictWriter(stream, fieldnames=fields)
        writer.writeheader()
        for row in mapping:
            writer.writerow(
                {
                    "state": row.state,
                    "reason": row.reason,
                    "old_fact_id": row.old_fact_id,
                    "mp_id": row.mp_id,
                    "model_id": row.model_id,
                    "legacy_row_id": row.legacy_row_id,
                    "legacy_return_id": row.legacy_return_id,
                    "api_return_id": row.api_return_id,
                    "api_order_id": row.api_order_id,
                    "creation_date": row.creation_date,
                    "legacy_offer_id": row.legacy_offer_id,
                    "market_sku": row.market_sku,
                    "shop_sku": row.shop_sku,
                    "item_line_no": row.item_line_no or "",
                    "raw_refund_amount": row.raw_refund_amount or "",
                    "method": row.method,
                }
            )


def _connection_settings(environment: Mapping[str, str]) -> dict[str, Any]:
    """Demand an explicit credential family; never embed production credentials."""
    user = _text(environment.get("V3_MYSQL_USER"))
    password = str(environment.get("V3_MYSQL_PASSWORD") or "")
    if not user or not password:
        raise SupersessionBlocked("V3_MYSQL_USER and V3_MYSQL_PASSWORD must be set explicitly")
    return {
        "host": _text(environment.get("V3_MYSQL_HOST")) or "127.0.0.1",
        "port": int(_text(environment.get("V3_MYSQL_PORT")) or 3306),
        "user": user,
        "password": password,
        "charset": "utf8mb4",
        "cursorclass": pymysql.cursors.DictCursor,
        "autocommit": False,
    }


EMPTY_FACT_ROWS_SQL = f"""
SELECT COUNT(*) AS c
FROM {KERNEL}.fact_returns
WHERE mp_id IN (6, 8)
  AND amount IS NULL
  AND return_date IS NULL
  AND source_payload_id IS NULL
"""


def read_mapping(cursor: Any) -> list[MappingRow]:
    cursor.execute(MAPPING_EVIDENCE_SQL)
    return build_mapping(cursor.fetchall())


def count_empty_fact_rows(cursor: Any) -> int:
    """Сколько пустышек в факте — знаменатель, от которого считать покрытие.

    Отчёт «unmapped=0» берёт знаменатель из своей выборки: строк, нашедшихся
    в зеркале легаси.  На 04.08 таких 861, а пустых строк в факте 863 — и обе
    недостающие не отсеиваются ни одним условием по стороне факта.  Без этого
    числа «ноль неразрешённых» читается как «всё покрыто» (F-88).
    """
    cursor.execute(EMPTY_FACT_ROWS_SQL)
    row = cursor.fetchone()
    return int(row["c"] if isinstance(row, Mapping) else row[0])


def _id_placeholders(ids: Sequence[int]) -> tuple[str, tuple[int, ...]]:
    if not ids:
        raise SupersessionBlocked("refusing a zero-row supersession")
    return ", ".join("%s" for _ in ids), tuple(ids)


def _load_old_snapshots(cursor: Any, mapped: Sequence[MappingRow]) -> dict[int, Mapping[str, Any]]:
    placeholders, ids = _id_placeholders([row.old_fact_id for row in mapped])
    cursor.execute(
        f"SELECT {OLD_FACT_SNAPSHOT_FIELDS} FROM {KERNEL}.fact_returns WHERE id IN ({placeholders})",
        ids,
    )
    found = {_as_int(row.get("id"), "snapshot fact id"): row for row in cursor.fetchall()}
    missing = sorted(set(ids) - set(found))
    if missing:
        raise SupersessionBlocked(f"old fact disappeared before snapshot: {missing[:5]}")
    for row in found.values():
        if (
            row.get("source_payload_id") is not None
            or row.get("return_date") is not None
            or row.get("amount") is not None
        ):
            raise SupersessionBlocked("old fact changed and is no longer the empty legacy placeholder")
    return found


def _json_value(value: Any) -> Any:
    if isinstance(value, Decimal):
        return str(value)
    if isinstance(value, (date, datetime)):
        return value.isoformat()
    raise TypeError(f"unsupported snapshot value: {type(value).__name__}")


def _insert_snapshots(
    cursor: Any,
    batch_id: str,
    mapped: Sequence[MappingRow],
    targets: Mapping[tuple[int, str, int], ProjectionTarget],
    old_snapshots: Mapping[int, Mapping[str, Any]],
) -> None:
    sql = f"""
INSERT INTO {SUPERSESSION_TABLE}
  (batch_id, old_fact_id, projected_fact_id, mp_id, model_id, legacy_row_id,
   legacy_return_id, api_return_id, api_order_id, creation_date,
   old_fact_snapshot, mapping_evidence)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, NULLIF(%s, ''), %s, %s)
"""
    values: list[tuple[Any, ...]] = []
    for row in mapped:
        target = targets[row.target_key]
        evidence = {
            "method": row.method,
            "market_sku": row.market_sku,
            "shop_sku": row.shop_sku,
            "item_line_no": row.item_line_no,
            "raw_refund_amount": str(row.raw_refund_amount) if row.raw_refund_amount is not None else None,
        }
        values.append(
            (
                batch_id,
                row.old_fact_id,
                target.fact_id,
                row.mp_id,
                row.model_id,
                row.legacy_row_id,
                row.legacy_return_id,
                row.api_return_id,
                row.api_order_id,
                row.creation_date,
                json.dumps(old_snapshots[row.old_fact_id], default=_json_value, ensure_ascii=False, sort_keys=True),
                json.dumps(evidence, ensure_ascii=False, sort_keys=True),
            )
        )
    cursor.executemany(sql, values)
    cursor.execute(f"SELECT COUNT(*) AS c FROM {SUPERSESSION_TABLE} WHERE batch_id = %s", (batch_id,))
    if _as_int(cursor.fetchone().get("c"), "snapshot count") != len(mapped):
        raise SupersessionBlocked("snapshot verification count differs from mapping")


def _delete_active_placeholders(cursor: Any, mapped: Sequence[MappingRow]) -> None:
    placeholders, ids = _id_placeholders([row.old_fact_id for row in mapped])
    cursor.execute(
        f"""
DELETE FROM {KERNEL}.fact_returns
WHERE id IN ({placeholders})
  AND source_payload_id IS NULL
  AND return_date IS NULL
  AND amount IS NULL
""",
        ids,
    )
    if cursor.rowcount != len(mapped):
        raise SupersessionBlocked(
            f"placeholder delete affected {cursor.rowcount}, expected {len(mapped)}; transaction will roll back"
        )


def _postcheck(
    cursor: Any,
    mapped: Sequence[MappingRow],
    expected_targets: Mapping[tuple[int, str, int], ProjectionTarget],
) -> None:
    placeholders, ids = _id_placeholders([row.old_fact_id for row in mapped])
    cursor.execute(f"SELECT COUNT(*) AS c FROM {KERNEL}.fact_returns WHERE id IN ({placeholders})", ids)
    if _as_int(cursor.fetchone().get("c"), "remaining old fact count"):
        raise SupersessionBlocked("old placeholder remains after delete; transaction will roll back")
    result = projection_check(cursor, mapped)
    if not result.passed or len(result.targets) != len(mapped):
        raise SupersessionBlocked("post-delete direct target count is not exactly one per mapping")
    _monthly_direct_postcheck(cursor, mapped, expected_targets)


def _monthly_direct_postcheck(
    cursor: Any,
    mapped: Sequence[MappingRow],
    expected_targets: Mapping[tuple[int, str, int], ProjectionTarget],
) -> None:
    """Count active facts by source month directly after the replacement.

    This deliberately does not infer success from the number of deleted rows.
    The query re-reads active ``fact_returns`` and proves each source month has
    exactly its expected API facts, rather than old-plus-new duplicates.
    """
    if not mapped:
        raise SupersessionBlocked("refusing a zero-row monthly postcheck")
    source_rows = " UNION ALL ".join(
        "SELECT %s AS mp_id, %s AS api_return_id, %s AS model_id, %s AS source_month"
        for _ in mapped
    )
    values: list[Any] = []
    expected: dict[str, int] = defaultdict(int)
    expected_amount: dict[str, Decimal] = defaultdict(lambda: Decimal("0"))
    for row in mapped:
        month = row.creation_date[:7]
        values.extend((row.mp_id, row.api_return_id, row.model_id, month))
        expected[month] += 1
        expected_amount[month] += expected_targets[row.target_key].amount or Decimal("0")
    cursor.execute(
        f"""
WITH expected AS ({source_rows})
SELECT expected.source_month AS month,
       COUNT(fact.id) AS fact_rows,
       COALESCE(SUM(fact.amount), 0) AS fact_amount
FROM expected
LEFT JOIN {KERNEL}.fact_returns AS fact
  ON fact.mp_id = expected.mp_id
 AND fact.return_id = expected.api_return_id COLLATE utf8mb4_unicode_ci
 AND fact.model_id = expected.model_id
GROUP BY expected.source_month
ORDER BY expected.source_month
""",
        tuple(values),
    )
    actual = { _text(row.get("month")): row for row in cursor.fetchall() }
    actual_counts = {
        month: _as_int(row.get("fact_rows"), "monthly fact count")
        for month, row in actual.items()
    }
    actual_amounts = {
        month: _as_decimal(row.get("fact_amount")) or Decimal("0")
        for month, row in actual.items()
    }
    if actual_counts != dict(expected) or actual_amounts != dict(expected_amount):
        raise SupersessionBlocked(
            "monthly direct postcheck differs: "
            f"counts={actual_counts}/{dict(expected)}, "
            f"amounts={actual_amounts}/{dict(expected_amount)}"
        )


def apply(cursor: Any) -> str:
    """Archive then remove active placeholders in one transaction after all gates."""
    mapping = read_mapping(cursor)
    mapped = _by_state(mapping, "mapped")
    gates = _mapping_gate(mapping)
    projection = projection_check(cursor, mapped)
    if gates or not projection.passed:
        raise SupersessionBlocked("; ".join((*gates, *projection.failures)))

    # The DDL is only archival infrastructure.  It happens after every read
    # gate has passed and before the transaction which changes fact rows.
    cursor.execute(SUPERSESSION_DDL)
    cursor.execute("START TRANSACTION")
    try:
        # Re-read in the mutation transaction: a dry-run or a previous SELECT
        # is never accepted as proof for a later write.
        mapping = read_mapping(cursor)
        mapped = _by_state(mapping, "mapped")
        gates = _mapping_gate(mapping)
        projection = projection_check(cursor, mapped)
        if gates or not projection.passed:
            raise SupersessionBlocked("; ".join((*gates, *projection.failures)))
        snapshots = _load_old_snapshots(cursor, mapped)
        batch_id = str(uuid.uuid4())
        _insert_snapshots(cursor, batch_id, mapped, projection.targets, snapshots)
        _delete_active_placeholders(cursor, mapped)
        _postcheck(cursor, mapped, projection.targets)
        cursor.connection.commit()
        return batch_id
    except Exception:
        cursor.connection.rollback()
        raise


def _parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--apply",
        action="store_true",
        help="archive proven placeholders and remove them from the active fact table",
    )
    parser.add_argument(
        "--mapping-out",
        type=Path,
        default=DEFAULT_MAPPING_PATH,
        help=f"CSV evidence output (default: {DEFAULT_MAPPING_PATH})",
    )
    return parser


def main(argv: Sequence[str] | None = None, environment: Mapping[str, str] | None = None) -> int:
    args = _parser().parse_args(argv)
    try:
        settings = _connection_settings(environment or os.environ)
        connection = pymysql.connect(**settings)
        try:
            with connection.cursor() as cursor:
                mapping = read_mapping(cursor)
                mapped = _by_state(mapping, "mapped")
                projection = projection_check(cursor, mapped)
                write_mapping_csv(mapping, args.mapping_out)
                print_plan(
                    mapping,
                    projection,
                    apply=args.apply,
                    empty_fact_rows=count_empty_fact_rows(cursor),
                )
                print(f"mapping evidence: {args.mapping_out}")
                if not args.apply:
                    return 0
                batch_id = apply(cursor)
                print(f"APPLY COMPLETE: archival batch={batch_id}")
                return 0
        finally:
            connection.close()
    except SupersessionBlocked as exc:
        print(f"APPLY BLOCKED: {exc}", file=sys.stderr)
        return 2


if __name__ == "__main__":
    raise SystemExit(main())