#!/usr/bin/env python3
"""Forensic dry-run for the historical YM return recovery.

The source is immutable ``lamoda_reports.mp_returns``.  This tool deliberately
does not write unless ``--apply`` is supplied, and even then refuses before its
first DML if recovery would be invisible to the approved-input gate, duplicate
an old legacy fact, or fail the current YM item/order projection.

Why the conservative gate matters
-------------------------------
The normal ``ym.returns`` build input is an attested fresh ``SpecCollector``
run.  A row reconstructed from legacy is not a new API collection and must not
be presented as one merely to make it eligible.  A future reviewed recovery
source may make the staged records selectable; until then this script is a
safe dry-run and an explicit ``--apply`` stop rather than a hidden money-data
mutation.
"""
from __future__ import annotations

import argparse
import hashlib
import json
import os
import sys
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
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

from config import MYSQL  # noqa: E402


LEGACY = "lamoda_reports"
INTAKE = "gwptd_intake"
KERNEL = "gwptd_kernel"
NORMAL_ENDPOINT = "ym.returns"
RECOVERY_ENDPOINT = "ym.returns.legacy_recovery"
RECOVERY_COLLECTOR_CLASS = "LegacyYmReturnsRecovery"
RECOVERY_PARSER_VERSION = "legacy-recovery-2026-08-03"
RECOVERY_SOURCE_PREFIX = "legacy:lamoda_reports.mp_returns"
CAMPAIGN_BY_MP = {6: "fbs", 8: "fby"}
MANIFEST_PATH = REPO_ROOT / "specs/kernel/build-all-inputs.yaml"
DEFAULT_SNAPSHOT_PATH = (
    REPO_ROOT
    / "docs/remediation/agent-exchange/results/W1-1-ym-returns-recovery.snapshot.json"
)


class RecoveryBlocked(RuntimeError):
    """A precondition proves that applying the recovery would be unsafe."""


@dataclass(frozen=True)
class RecoveryPlan:
    legacy_rows: int
    overlap_rows: int
    candidate_rows: int
    candidate_refund_amount: Decimal
    malformed_headers: int
    malformed_items: int
    refund_disagreements: int
    legacy_empty_fact_rows: int
    candidate_empty_fact_rows: int
    api_id_differs_from_legacy_fact_id: int
    projection_invalid_items: int
    projection_identity_failures: int
    projection_order_failures: int
    fully_projectable_returns: int
    existing_recovery_payloads: int
    recovery_endpoint_registered: bool

    def blockers(self) -> tuple[str, ...]:
        blocked: list[str] = []
        if self.malformed_headers or self.malformed_items or self.refund_disagreements:
            blocked.append("legacy JSON is not losslessly parseable")
        if self.candidate_empty_fact_rows:
            blocked.append(
                "current YM reconciliation preserves legacy fact rows; "
                "applying would leave empty and new fact rows together"
            )
        if (
            self.projection_invalid_items
            or self.projection_identity_failures
            or self.projection_order_failures
        ):
            blocked.append(
                "current YM projection is incomplete; build_fact_returns would fail closed"
            )
        if self.existing_recovery_payloads:
            blocked.append("a prior recovery staging is already present; rerun is not idempotent")
        if not self.recovery_endpoint_registered:
            blocked.append(
                "no reviewed recovery source is selectable by the kernel input manifest"
            )
        return tuple(blocked)


# Every cross-schema order-id comparison pins both operands to the same
# collation.  This is intentional: the two schemas do not share a default
# collation on S3 (ERROR 1267 without it).
CLASSIFIED_LEGACY_CTE = f"""
WITH legacy AS (
    SELECT l.id AS legacy_id,
           l.marketplace_id AS mp_id,
           l.return_id AS legacy_return_id,
           l.collected_date,
           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,
           JSON_UNQUOTE(JSON_EXTRACT(l.raw_json, '$.creationDate')) AS api_creation_date,
           CAST(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(l.raw_json, '$.amount.value')), '')
                AS DECIMAL(14,2)) AS amount_value,
           CAST(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(l.raw_json, '$.refundAmount')), '')
                AS DECIMAL(16,2)) / 100 AS refund_amount_value
    FROM {LEGACY}.mp_returns AS l
    WHERE l.marketplace_id IN (6, 8)
      AND l.amount IS NULL
      AND l.return_date IS NULL
),
classified AS (
    SELECT legacy.*,
           EXISTS(
               SELECT 1
               FROM {INTAKE}.ym_returns AS working
               WHERE working.order_id COLLATE utf8mb4_unicode_ci
                   = legacy.api_order_id COLLATE utf8mb4_unicode_ci
           ) AS overlaps_working,
           EXISTS(
               SELECT 1
               FROM {KERNEL}.fact_returns AS legacy_fact
               WHERE legacy_fact.mp_id = legacy.mp_id
                 AND legacy_fact.return_id COLLATE utf8mb4_unicode_ci
                   = legacy.legacy_return_id COLLATE utf8mb4_unicode_ci
                 AND legacy_fact.source_payload_id IS NULL
                 AND legacy_fact.return_date IS NULL
                 AND legacy_fact.amount IS NULL
           ) AS legacy_empty_fact
    FROM legacy
)
"""

SUMMARY_SQL = CLASSIFIED_LEGACY_CTE + """
SELECT
    COUNT(*) AS legacy_rows,
    COALESCE(SUM(overlaps_working), 0) AS overlap_rows,
    COALESCE(SUM(NOT overlaps_working), 0) AS candidate_rows,
    COALESCE(SUM(CASE WHEN NOT overlaps_working THEN amount_value ELSE 0 END), 0)
      AS candidate_refund_amount,
    COALESCE(SUM(
        api_return_id IS NULL OR api_return_id = ''
        OR api_order_id IS NULL OR api_order_id = ''
        OR api_creation_date IS NULL OR api_creation_date = ''
    ), 0) AS malformed_headers,
    COALESCE(SUM(
        JSON_TYPE(JSON_EXTRACT(raw_json, '$.items')) <> 'ARRAY'
        OR JSON_LENGTH(JSON_EXTRACT(raw_json, '$.items')) = 0
    ), 0) AS malformed_items,
    COALESCE(SUM(amount_value <> refund_amount_value), 0) AS refund_disagreements,
    COALESCE(SUM(legacy_empty_fact), 0) AS legacy_empty_fact_rows,
    COALESCE(SUM(NOT overlaps_working AND legacy_empty_fact), 0)
      AS candidate_empty_fact_rows,
    COALESCE(SUM(
        NOT overlaps_working
        AND legacy_empty_fact
        AND api_return_id COLLATE utf8mb4_unicode_ci
            <> legacy_return_id COLLATE utf8mb4_unicode_ci
    ), 0) AS api_id_differs_from_legacy_fact_id,
    (SELECT COUNT(*)
     FROM gwptd_intake.raw_payload AS prior_recovery
     WHERE prior_recovery.endpoint_code = 'ym.returns.legacy_recovery')
      AS existing_recovery_payloads
FROM classified
"""

# This is the exact item identity/order shape used by build_fact_returns.py,
# expressed as SELECT-only evidence.  JSON_EXTRACT(..., '$') is needed when a
# JSON value crosses a CTE boundary on the S3 MySQL version.
PROJECTION_SQL = CLASSIFIED_LEGACY_CTE + f"""
, candidate AS (
    SELECT * FROM classified WHERE NOT overlaps_working
), item_scope AS (
    SELECT c.legacy_id,
           c.mp_id,
           c.api_return_id,
           c.api_order_id,
           item.line_no,
           NULLIF(TRIM(item.shop_sku), '') AS shop_sku,
           item.item_count_text,
           CASE
             WHEN idn.model_id IS NOT NULL AND product.model_id IS NOT NULL
              AND idn.model_id <> product.model_id THEN NULL
             ELSE COALESCE(idn.model_id, product.model_id)
           END AS candidate_model_id
    FROM candidate AS c
    CROSS JOIN JSON_TABLE(
        JSON_EXTRACT(c.raw_json, '$'),
        '$.items[*]' COLUMNS (
            line_no FOR ORDINALITY,
            shop_sku VARCHAR(255) PATH '$.shopSku' NULL ON EMPTY NULL ON ERROR,
            item_count_text VARCHAR(64) PATH '$.count' NULL ON EMPTY NULL ON ERROR
        )
    ) AS item
    LEFT JOIN {KERNEL}.dim_product_identifier AS idn
      ON idn.mp_id = 6
     AND idn.identifier_type = 'shopSku'
     AND idn.identifier_value = item.shop_sku COLLATE utf8mb4_unicode_ci
     AND idn.is_current = 1
    LEFT JOIN {KERNEL}.dim_product AS product
      ON product.artikul_upper = UPPER(TRIM(item.shop_sku)) COLLATE utf8mb4_unicode_ci
), item_checks AS (
    SELECT item_scope.legacy_id,
           item_scope.mp_id,
           item_scope.api_return_id,
           item_scope.api_order_id,
           item_scope.line_no,
           item_scope.shop_sku,
           CASE
             WHEN item_scope.item_count_text IS NOT NULL
              AND CHAR_LENGTH(item_scope.item_count_text) <= 9
              AND REGEXP_LIKE(item_scope.item_count_text, '^[1-9][0-9]*$', 'c')
             THEN CAST(item_scope.item_count_text AS UNSIGNED)
             ELSE NULL
           END AS item_count,
           COUNT(DISTINCT item_scope.candidate_model_id) AS identity_matches,
           COUNT(DISTINCT fact_order.id) AS order_matches
    FROM item_scope
    LEFT JOIN {KERNEL}.fact_orders AS fact_order
      ON fact_order.mp_id = item_scope.mp_id
     AND fact_order.order_id COLLATE utf8mb4_unicode_ci
         = item_scope.api_order_id COLLATE utf8mb4_unicode_ci
     AND fact_order.model_id = item_scope.candidate_model_id
     AND fact_order.quantity > 0
     AND fact_order.revenue IS NOT NULL
    GROUP BY item_scope.legacy_id,
             item_scope.mp_id,
             item_scope.api_return_id,
             item_scope.api_order_id,
             item_scope.line_no,
             item_scope.shop_sku,
             item_scope.item_count_text
)
SELECT COALESCE(SUM(shop_sku IS NULL OR item_count IS NULL), 0) AS invalid_items,
       COALESCE(SUM(identity_matches <> 1), 0) AS identity_failures,
       COALESCE(SUM(identity_matches = 1 AND order_matches <> 1), 0) AS order_failures,
       COALESCE(SUM(
           shop_sku IS NOT NULL AND item_count IS NOT NULL
           AND identity_matches = 1 AND order_matches = 1
       ), 0) AS projected_items,
       COUNT(DISTINCT CASE
           WHEN shop_sku IS NOT NULL AND item_count IS NOT NULL
            AND identity_matches = 1 AND order_matches = 1
           THEN legacy_id END
       ) AS fully_projectable_returns
FROM item_checks
"""

MONTHLY_SQL = CLASSIFIED_LEGACY_CTE + """
SELECT DATE_FORMAT(api_creation_date, '%Y-%m') AS month,
       COUNT(*) AS return_rows,
       SUM(amount_value) AS refund_amount,
       SUM(legacy_empty_fact) AS legacy_empty_fact_rows
FROM classified
WHERE NOT overlaps_working
GROUP BY DATE_FORMAT(api_creation_date, '%Y-%m')
ORDER BY month
"""

CANDIDATE_ROWS_SQL = CLASSIFIED_LEGACY_CTE + """
SELECT legacy_id, mp_id, legacy_return_id, collected_date, raw_json,
       api_return_id, api_order_id, api_creation_date, amount_value,
       refund_amount_value
FROM classified
WHERE NOT overlaps_working
ORDER BY legacy_id
"""


def _required_connection_settings(environment: Mapping[str, str]) -> dict[str, Any]:
    """Require an explicit V3 credential family; never silently use root/root."""
    user = str(environment.get("V3_MYSQL_USER", MYSQL["user"]) or "").strip()
    password = str(environment.get("V3_MYSQL_PASSWORD", MYSQL["password"]) or "")
    if not user or not password:
        raise RecoveryBlocked("V3_MYSQL_USER/V3_MYSQL_PASSWORD must be set explicitly")
    return {
        "host": str(environment.get("V3_MYSQL_HOST", MYSQL["host"])),
        "port": int(environment.get("V3_MYSQL_PORT", MYSQL["port"])),
        "user": user,
        "password": password,
        "charset": "utf8mb4",
        "cursorclass": pymysql.cursors.DictCursor,
        "autocommit": False,
    }


def _as_int(row: Mapping[str, Any], key: str) -> int:
    value = row.get(key)
    try:
        return int(value or 0)
    except (TypeError, ValueError) as exc:
        raise RecoveryBlocked(f"read-only evidence has invalid {key}") from exc


def _as_decimal(row: Mapping[str, Any], key: str) -> Decimal:
    try:
        return Decimal(str(row.get(key) or 0))
    except (InvalidOperation, ValueError) as exc:
        raise RecoveryBlocked(f"read-only evidence has invalid {key}") from exc


def _recovery_endpoint_registered(manifest_path: Path = MANIFEST_PATH) -> bool:
    """A raw staged recovery is not kernel input until an explicit manifest says so."""
    return manifest_path.exists() and RECOVERY_ENDPOINT in manifest_path.read_text(encoding="utf-8")


def read_plan(cursor: Any, *, manifest_path: Path = MANIFEST_PATH) -> tuple[RecoveryPlan, list[dict[str, Any]]]:
    """Execute only SELECT statements and return the plan plus monthly evidence."""
    cursor.execute(SUMMARY_SQL)
    summary = cursor.fetchone()
    if not isinstance(summary, Mapping):
        raise RecoveryBlocked("legacy summary returned no row")
    cursor.execute(PROJECTION_SQL)
    projection = cursor.fetchone()
    if not isinstance(projection, Mapping):
        raise RecoveryBlocked("YM projection evidence returned no row")
    cursor.execute(MONTHLY_SQL)
    monthly = list(cursor.fetchall())
    return (
        RecoveryPlan(
            legacy_rows=_as_int(summary, "legacy_rows"),
            overlap_rows=_as_int(summary, "overlap_rows"),
            candidate_rows=_as_int(summary, "candidate_rows"),
            candidate_refund_amount=_as_decimal(summary, "candidate_refund_amount"),
            malformed_headers=_as_int(summary, "malformed_headers"),
            malformed_items=_as_int(summary, "malformed_items"),
            refund_disagreements=_as_int(summary, "refund_disagreements"),
            legacy_empty_fact_rows=_as_int(summary, "legacy_empty_fact_rows"),
            candidate_empty_fact_rows=_as_int(summary, "candidate_empty_fact_rows"),
            api_id_differs_from_legacy_fact_id=_as_int(
                summary, "api_id_differs_from_legacy_fact_id"
            ),
            projection_invalid_items=_as_int(projection, "invalid_items"),
            projection_identity_failures=_as_int(projection, "identity_failures"),
            projection_order_failures=_as_int(projection, "order_failures"),
            fully_projectable_returns=_as_int(projection, "fully_projectable_returns"),
            existing_recovery_payloads=_as_int(summary, "existing_recovery_payloads"),
            recovery_endpoint_registered=_recovery_endpoint_registered(manifest_path),
        ),
        [dict(row) for row in monthly],
    )


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


def print_plan(plan: RecoveryPlan, monthly: Sequence[Mapping[str, Any]]) -> None:
    print("YM legacy returns recovery: DRY RUN (no writes)")
    print(f"legacy empty rows: {plan.legacy_rows}")
    print(f"skip as existing working intake orderId overlaps: {plan.overlap_rows}")
    print(
        "would stage through intake: "
        f"{plan.candidate_rows} returns, { _format_amount(plan.candidate_refund_amount) } RUB"
    )
    print(
        "legacy empty fact rows: "
        f"{plan.legacy_empty_fact_rows}; among transfer candidates: "
        f"{plan.candidate_empty_fact_rows}"
    )
    print(
        "candidate API return_id differs from the legacy fact return_id: "
        f"{plan.api_id_differs_from_legacy_fact_id}"
    )
    print(
        "current projection: "
        f"invalid_items={plan.projection_invalid_items}, "
        f"identity_failures={plan.projection_identity_failures}, "
        f"order_failures={plan.projection_order_failures}, "
        f"fully_projectable_returns={plan.fully_projectable_returns}"
    )
    print(f"already staged recovery payloads: {plan.existing_recovery_payloads}")
    print("monthly source-refund plan:")
    for row in monthly:
        print(
            f"  {row['month']}: rows={row['return_rows']}, "
            f"refund={_format_amount(_as_decimal(row, 'refund_amount'))}, "
            f"legacy_empty_facts={row['legacy_empty_fact_rows']}"
        )
    blockers = plan.blockers()
    if blockers:
        print("APPLY BLOCKED:")
        for blocker in blockers:
            print(f"  - {blocker}")
    else:
        print("APPLY PRECHECK: PASS")


def _canonical_json(value: Any) -> str:
    """Canonical raw text makes the recovery payload hash reproducible."""
    return json.dumps(value, ensure_ascii=False, separators=(",", ":"), sort_keys=True)


def _ym_datetime(value: Any) -> str:
    """Use the current YM mapper's timezone-stripping storage representation."""
    raw = str(value or "").strip()
    normalized = raw.replace("T", " ").rstrip("Z").split("+", 1)[0]
    normalized = normalized.split(".", 1)[0]
    if not normalized:
        raise RecoveryBlocked("candidate raw_json has an empty creationDate")
    try:
        datetime.fromisoformat(normalized)
    except ValueError as exc:
        raise RecoveryBlocked("candidate raw_json has an invalid creationDate") from exc
    return normalized


def _parse_legacy_row(row: Mapping[str, Any]) -> dict[str, Any]:
    """Validate and normalize one candidate without inventing marketplace facts."""
    try:
        mp_id = int(row["mp_id"])
    except (KeyError, TypeError, ValueError) as exc:
        raise RecoveryBlocked("candidate has no valid legacy marketplace_id") from exc
    if mp_id not in CAMPAIGN_BY_MP:
        raise RecoveryBlocked(f"candidate has unsupported YM marketplace_id={mp_id}")
    raw = row.get("raw_json")
    try:
        raw_object = json.loads(raw) if isinstance(raw, str) else raw
    except (TypeError, ValueError) as exc:
        raise RecoveryBlocked("candidate raw_json is not valid JSON") from exc
    if not isinstance(raw_object, dict):
        raise RecoveryBlocked("candidate raw_json is not an API-object payload")
    for field in ("id", "orderId", "creationDate", "refundAmount"):
        value = raw_object.get(field)
        if value is None or str(value).strip() == "":
            raise RecoveryBlocked(f"candidate raw_json has no {field}")
    items = raw_object.get("items")
    if not isinstance(items, list) or not items:
        raise RecoveryBlocked("candidate raw_json has no non-empty items[]")
    collected_date = row.get("collected_date")
    if isinstance(collected_date, datetime):
        collected_at = collected_date.replace(microsecond=0)
    elif isinstance(collected_date, date):
        collected_at = datetime.combine(collected_date, datetime.min.time())
    else:
        raise RecoveryBlocked("candidate has no original legacy collected_date")
    canonical = _canonical_json(raw_object)
    legacy_id = _as_int(row, "legacy_id")
    return {
        "legacy_id": legacy_id,
        "mp_id": mp_id,
        "campaign_model": CAMPAIGN_BY_MP[mp_id],
        "return_id": str(raw_object["id"]),
        "order_id": str(raw_object["orderId"]),
        "return_date_mp": _ym_datetime(raw_object["creationDate"]),
        "status": raw_object.get("shipmentStatus") or raw_object.get("status"),
        "return_type": raw_object.get("returnType"),
        "reason_type": raw_object.get("reasonType"),
        "refund_amount": _as_decimal({"value": raw_object.get("refundAmount")}, "value")
        / Decimal("100"),
        "raw_object": raw_object,
        "canonical": canonical,
        "payload_hash": hashlib.sha256(canonical.encode("utf-8")).hexdigest(),
        "collected_at": collected_at,
        "source_business_key": (
            f"{RECOVERY_SOURCE_PREFIX}:{legacy_id}:return:{raw_object['id']}"
        ),
    }


def _write_before_snapshot(
    snapshot_path: Path,
    plan: RecoveryPlan,
    monthly: Sequence[Mapping[str, Any]],
) -> None:
    """Persist the pre-apply receipt before the transaction begins."""
    snapshot_path.parent.mkdir(parents=True, exist_ok=True)
    document = {
        "capturedAt": datetime.now().astimezone().isoformat(timespec="seconds"),
        "mode": "before_apply",
        "plan": {
            **plan.__dict__,
            "candidate_refund_amount": str(plan.candidate_refund_amount),
            "blockers": list(plan.blockers()),
        },
        "monthly": [
            {
                **dict(row),
                "refund_amount": str(row.get("refund_amount") or 0),
            }
            for row in monthly
        ],
    }
    temporary = snapshot_path.with_suffix(snapshot_path.suffix + ".tmp")
    temporary.write_text(json.dumps(document, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    temporary.replace(snapshot_path)


def _insert_recovery_rows(connection: Any) -> tuple[int, int]:
    """Write a provenance-distinct intake version.  Called only after all gates pass."""
    with connection.cursor() as cursor:
        cursor.execute(CANDIDATE_ROWS_SQL)
        source_rows = [_parse_legacy_row(row) for row in cursor.fetchall()]
        if not source_rows:
            raise RecoveryBlocked("recovery has no candidates to stage")
        for row in source_rows:
            cursor.execute(
                f"""
                INSERT INTO {INTAKE}.collection_run
                  (mp_id, endpoint_code, status, parser_version, collector_class)
                VALUES (%s, %s, 'running', %s, %s)
                """,
                (row["mp_id"], RECOVERY_ENDPOINT, RECOVERY_PARSER_VERSION, RECOVERY_COLLECTOR_CLASS),
            )
            run_id = cursor.lastrowid
            cursor.execute(
                f"""
                INSERT INTO {INTAKE}.raw_payload
                  (run_id, mp_id, endpoint_code, source_business_key, payload_hash,
                   raw_json, raw_canonical_json, http_status, collected_at)
                VALUES (%s, %s, %s, %s, %s, %s, %s, NULL, %s)
                """,
                (
                    run_id, row["mp_id"], RECOVERY_ENDPOINT, row["source_business_key"],
                    row["payload_hash"], row["canonical"], row["canonical"], row["collected_at"],
                ),
            )
            payload_id = cursor.lastrowid
            cursor.execute(
                f"""
                INSERT INTO {INTAKE}.ym_returns
                  (payload_id, mp_id, campaign_model, return_id, order_id, status,
                   return_type, reason_type, return_date_mp, refund_amount,
                   return_json, collected_at)
                VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
                """,
                (
                    payload_id, row["mp_id"], row["campaign_model"], row["return_id"],
                    row["order_id"], row["status"], row["return_type"], row["reason_type"],
                    row["return_date_mp"], row["refund_amount"], row["canonical"], row["collected_at"],
                ),
            )
            cursor.execute(
                f"""
                UPDATE {INTAKE}.collection_run
                SET status = 'success', finished_at = NOW(6), rows_received = 1,
                    rows_parsed = 1, rows_skipped = 0
                WHERE run_id = %s AND status = 'running'
                """,
                (run_id,),
            )
            if cursor.rowcount != 1:
                raise RecoveryBlocked(f"recovery collection_run {run_id} did not seal")
    return len(source_rows), len(source_rows)


def _parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="Dry-run legacy YM returns recovery")
    mode = parser.add_mutually_exclusive_group()
    mode.add_argument("--dry-run", action="store_true", help="print plan only (default)")
    mode.add_argument("--apply", action="store_true", help="write only after every safety gate passes")
    parser.add_argument(
        "--snapshot-out",
        type=Path,
        default=DEFAULT_SNAPSHOT_PATH,
        help="pre-apply JSON snapshot path (written only by --apply after preflight)",
    )
    return parser


def main(argv: Sequence[str] | None = None) -> int:
    args = _parser().parse_args(argv)
    connection = pymysql.connect(**_required_connection_settings(dict(os.environ)))
    try:
        with connection.cursor() as cursor:
            plan, monthly = read_plan(cursor)
        print_plan(plan, monthly)
        if not args.apply:
            connection.rollback()
            return 0
        blockers = plan.blockers()
        if blockers:
            connection.rollback()
            raise RecoveryBlocked("--apply refused before DML: " + "; ".join(blockers))
        _write_before_snapshot(args.snapshot_out, plan, monthly)
        staged, typed = _insert_recovery_rows(connection)
        connection.commit()
        print(f"APPLIED: raw_payload={staged}, ym_returns={typed}")
        return 0
    except RecoveryBlocked as exc:
        print(f"BLOCKED: {exc}", file=sys.stderr)
        return 2
    finally:
        connection.close()


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