"""ЭТАП 4 B2C launch — retention -> physical deletion task (part B).

Covers app/tasks/purge_expired_trade_in_data.py:
  - batched DELETE (not a single unbounded DELETE), each batch its own commit
  - stops draining a table once a batch returns fewer rows than batch_size
  - safety cap (max_batches) bounds a single run even on a huge backlog
  - both tables (trade_in_estimates, trade_in_leads) get drained
  - failure path: rollback + mark_failed with partial counters, exception re-raised
  - SQL shape: DELETE (not UPDATE/deactivate), no psycopg `::` cast trap
  - Payments retention (PR #2754): retain_until IS NULL + NOT EXISTS(payments)
    safeguards on the estimates DELETE, plus a pre-flight that refuses to run
    at all if it finds an ANOMALOUS paid purge-candidate (retain_until unset
    despite a payments row) -- a healthy paid row (retain_until set) must NOT
    trip it, see test_preflight_ignores_healthy_paid_row below.

Style mirrors tests/test_deactivate_stale_listings.py (_FakeDB, monkeypatched
runs_mod.mark_done/mark_failed).

Payments retention note on _FakeDB: purge_expired_trade_in_data now issues ONE
extra db.execute() call BEFORE any DELETE batch — the pre-flight paid-
candidates count (_PREFLIGHT_PAID_CANDIDATES_SQL). _FakeDB special-cases that
statement by identity and answers it from `preflight_count` (default 0 ==
"no anomalous candidates, proceed exactly as before this PR"). Every
pre-existing test's `db.executed` index shifted by +1 to account for this;
`db.commits` is unaffected (the pre-flight is a read, never committed).
"""

from __future__ import annotations

import os
import re
from pathlib import Path
from typing import Any
from uuid import uuid4

import pytest

os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")

from app.tasks import purge_expired_trade_in_data as task_mod

_SQL_DIR = Path(__file__).resolve().parents[1] / "data" / "sql"
_MIGRATION_231 = _SQL_DIR / "231_trade_in_privacy_retention.sql"


class _FakeResult:
    def __init__(self, rowcount: int) -> None:
        self.rowcount = rowcount

    def scalar_one(self) -> int:
        """Supports the PR-D1 pre-flight `SELECT count(*) ... .scalar_one()` call."""
        return self.rowcount


class _FakeDB:
    """Pops rowcounts in call order -- caller supplies the exact sequence expected.

    PR-D1: the pre-flight paid-candidates count is answered separately, from
    `preflight_count` (default 0), keyed by statement IDENTITY -- it never
    consumes an entry off `rowcounts` (that list is DELETE-batch rowcounts only).
    """

    def __init__(self, rowcounts: list[int], *, preflight_count: int = 0) -> None:
        self._rowcounts = list(rowcounts)
        self.preflight_count = preflight_count
        self.executed: list[tuple[Any, Any]] = []
        self.commits = 0
        self.rolled_back = False

    def execute(self, stmt: Any, params: dict[str, Any] | None = None) -> _FakeResult:
        self.executed.append((stmt, params))
        if stmt is task_mod._PREFLIGHT_PAID_CANDIDATES_SQL:
            return _FakeResult(self.preflight_count)
        return _FakeResult(self._rowcounts.pop(0))

    def commit(self) -> None:
        self.commits += 1

    def rollback(self) -> None:
        self.rolled_back = True


def _patch_runs(monkeypatch: pytest.MonkeyPatch) -> dict[str, Any]:
    marked: dict[str, Any] = {}
    monkeypatch.setattr(
        task_mod.runs_mod,
        "mark_done",
        lambda _db, run_id, counters: marked.update(
            kind="done", run_id=run_id, counters=dict(counters)
        ),
    )
    monkeypatch.setattr(
        task_mod.runs_mod,
        "mark_failed",
        lambda _db, run_id, err, counters: marked.update(
            kind="failed", run_id=run_id, err=err, counters=dict(counters)
        ),
    )
    return marked


# ── batching behaviour ──────────────────────────────────────────────────────


def test_stops_when_batch_below_size(monkeypatch: pytest.MonkeyPatch) -> None:
    marked = _patch_runs(monkeypatch)
    # estimates: one batch of 3 (< batch_size=10) -> stop. leads: one batch of 0 -> stop.
    db = _FakeDB([3, 0])
    out = task_mod.purge_expired_trade_in_data(db, run_id=1, batch_size=10, max_batches=20)  # type: ignore[arg-type]
    assert out == {"estimates_deleted": 3, "leads_deleted": 0}
    # +1 vs pre-PR-D1: db.executed now also holds the pre-flight paid-candidates
    # count (call #1), issued before either DELETE batch.
    assert len(db.executed) == 3
    assert db.commits == 2
    assert marked["counters"] == out


def test_loops_until_below_batch_size(monkeypatch: pytest.MonkeyPatch) -> None:
    _patch_runs(monkeypatch)
    # estimates: 5,5,2 (batch_size=5) -> 12 total, 3 batches. leads: 5,1 -> 6 total, 2 batches.
    db = _FakeDB([5, 5, 2, 5, 1])
    out = task_mod.purge_expired_trade_in_data(db, run_id=2, batch_size=5, max_batches=20)  # type: ignore[arg-type]
    assert out == {"estimates_deleted": 12, "leads_deleted": 6}
    assert len(db.executed) == 6  # +1: pre-flight call before the 5 DELETE batches
    assert db.commits == 5, "each batch must commit independently, not one final commit"


def test_respects_max_batches_cap(monkeypatch: pytest.MonkeyPatch) -> None:
    """Every batch returns a FULL batch_size (never 'caught up') -- only the
    max_batches safety cap can stop the loop. Proves the cap is enforced, not
    just coincidentally matching a 'caught up' condition."""
    _patch_runs(monkeypatch)
    db = _FakeDB([5, 5, 5, 5, 5, 5])  # exactly max_batches=3 per table, no more
    out = task_mod.purge_expired_trade_in_data(db, run_id=3, batch_size=5, max_batches=3)  # type: ignore[arg-type]
    assert out == {"estimates_deleted": 15, "leads_deleted": 15}
    assert len(db.executed) == 7  # pre-flight + 3 (estimates) + 3 (leads), NOT unbounded


def test_default_batch_size_and_max_batches_from_settings(monkeypatch: pytest.MonkeyPatch) -> None:
    _patch_runs(monkeypatch)
    db = _FakeDB([0, 0])  # first batch already empty on both tables -> stop immediately
    task_mod.purge_expired_trade_in_data(db, run_id=4)  # type: ignore[arg-type]
    # db.executed[0] is now the pre-flight call (no batch_size param) -- the
    # first DELETE-batch call (with batch_size) shifted to index 1.
    _stmt, params = db.executed[1]
    assert params is not None
    assert params["batch_size"] == task_mod.settings.trade_in_purge_batch_size


# ── table coverage / SQL shape ──────────────────────────────────────────────


def test_drains_both_tables_in_order(monkeypatch: pytest.MonkeyPatch) -> None:
    _patch_runs(monkeypatch)
    db = _FakeDB([0, 0])
    task_mod.purge_expired_trade_in_data(db, run_id=5, batch_size=100, max_batches=1)  # type: ignore[arg-type]
    # index 0 is now the pre-flight call; DELETE batches shifted to 1/2.
    first_sql = str(getattr(db.executed[1][0], "text", db.executed[1][0]))
    second_sql = str(getattr(db.executed[2][0], "text", db.executed[2][0]))
    assert "trade_in_estimates" in first_sql
    assert "trade_in_leads" in second_sql


def test_estimates_sql_is_delete_not_update() -> None:
    sql = task_mod._DELETE_EXPIRED_ESTIMATES_SQL.text
    assert "DELETE FROM trade_in_estimates" in sql
    assert "UPDATE" not in sql.upper()
    assert "expires_at < NOW()" in sql
    assert "ORDER BY expires_at" in sql
    assert "LIMIT CAST(:batch_size AS int)" in sql
    assert not re.search(r":\w+::", sql)


# ── Payments retention (PR #2754): two independent purge safeguards ─────────


def test_estimates_sql_excludes_retain_until_not_null() -> None:
    """Phase 1: exactly `retain_until IS NULL`, never `< NOW()` -- a paid row
    (retain_until IS NOT NULL) must never match the DELETE predicate, full stop,
    regardless of how far in the past that date eventually sits."""
    sql = task_mod._DELETE_EXPIRED_ESTIMATES_SQL.text
    assert "retain_until IS NULL" in sql
    assert "retain_until <" not in sql, "phase 1 must not weaken to retain_until < NOW()"


def test_estimates_sql_has_not_exists_payments_safeguard() -> None:
    """Independent safeguard: a row with ANY payments row survives even if
    retain_until failed to be set (fulfillment bug/race/manual INSERT)."""
    sql = task_mod._DELETE_EXPIRED_ESTIMATES_SQL.text
    assert "NOT EXISTS" in sql
    assert "FROM payments p" in sql
    assert "p.estimate_id = trade_in_estimates.id" in sql


def test_preflight_sql_requires_retain_until_is_null() -> None:
    """Deep-review finding 2026-08-06 MEDIUM (PR #2754): the pre-flight predicate
    MUST carry `retain_until IS NULL` -- without it, a perfectly healthy paid row
    (retain_until set, has a payments row -- the ORDINARY steady state one day
    after every sale) trips the alarm exactly as hard as the real anomaly
    (retain_until unset despite a payments row existing), permanently wedging
    the job on the very first successful sale (and, since leads purge runs from
    the same function AFTER this check, silently stopping 180-day 152-ФЗ lead
    retention too). See test_real_preflight_ignores_healthy_paid_row below for
    the behavioural proof against a real DB."""
    sql = task_mod._PREFLIGHT_PAID_CANDIDATES_SQL.text
    assert "expires_at < NOW()" in sql
    assert "created_by IS NULL" in sql
    assert "retain_until IS NULL" in sql
    assert "EXISTS (SELECT 1 FROM payments p WHERE p.estimate_id = e.id)" in sql
    assert not re.search(r":\w+::", sql)


def test_preflight_aborts_before_any_delete_batch(monkeypatch: pytest.MonkeyPatch) -> None:
    """Non-zero pre-flight count -> RuntimeError, mark_failed, ZERO DELETE batches
    ever issued (only the pre-flight SELECT itself is in db.executed)."""
    marked = _patch_runs(monkeypatch)
    db = _FakeDB([], preflight_count=2)  # rowcounts empty on purpose: must never be popped
    with pytest.raises(RuntimeError, match="pre-flight abort"):
        task_mod.purge_expired_trade_in_data(db, run_id=42, batch_size=10, max_batches=20)  # type: ignore[arg-type]

    assert len(db.executed) == 1, "only the pre-flight SELECT -- no DELETE batch was issued"
    assert db.commits == 0
    assert marked["kind"] == "failed"
    assert marked["run_id"] == 42
    assert marked["counters"] == {"estimates_deleted": 0, "leads_deleted": 0}
    assert "2" in marked["err"]


def test_preflight_zero_candidates_proceeds_as_before(monkeypatch: pytest.MonkeyPatch) -> None:
    """preflight_count=0 (default) -- the exact pre-PR-D1 behaviour for every
    row that exists today (all retain_until IS NULL) -- run proceeds normally."""
    marked = _patch_runs(monkeypatch)
    db = _FakeDB([0, 0])  # preflight_count defaults to 0
    out = task_mod.purge_expired_trade_in_data(db, run_id=43, batch_size=10, max_batches=20)  # type: ignore[arg-type]
    assert out == {"estimates_deleted": 0, "leads_deleted": 0}
    assert marked["kind"] == "done"


def test_leads_sql_unchanged_by_pr_d1() -> None:
    """Snapshot: _DELETE_EXPIRED_LEADS_SQL byte-for-byte unchanged by payments
    retention (PR #2754) — leads have their own retention deadline (migration
    231, no created_by/B2B split, no payments concept) and are explicitly out
    of scope for the payments-retention safeguards."""
    expected = (
        "\n    DELETE FROM trade_in_leads\n    WHERE id IN (\n"
        "        SELECT id FROM trade_in_leads\n"
        "        WHERE expires_at < NOW()\n"
        "        ORDER BY expires_at\n"
        "        LIMIT CAST(:batch_size AS int)\n    )\n    "
    )
    assert task_mod._DELETE_EXPIRED_LEADS_SQL.text == expected


def test_leads_sql_is_delete_not_update() -> None:
    sql = task_mod._DELETE_EXPIRED_LEADS_SQL.text
    assert "DELETE FROM trade_in_leads" in sql
    assert "UPDATE" not in sql.upper()
    assert "expires_at < NOW()" in sql
    assert "ORDER BY expires_at" in sql
    assert not re.search(r":\w+::", sql)


def test_estimates_sql_scopes_delete_to_anonymous_created_by_null() -> None:
    """Regression guard for the deep-review HIGH finding (2026-08-06):
    trade_in_estimates.expires_at is set UNCONDITIONALLY on every estimate
    (B2B pilot or anonymous) as a TTL on the estimate LINK/PDF staying
    resolvable, NOT a declared row-retention deadline -- see the task's
    module docstring. Without this guard, prod audit showed 1040/1057 rows
    past expires_at, 911 of them belonging to named pilots (admin, kopylov,
    brusnika, praktika, pilottest, admintest, user1); one unattended run at
    the seeded defaults (batch_size=500, max_batches=20) would have deleted
    essentially the whole table, including pilots' own operational history
    (/trade-in/history, /team/employees/{id}/history, cache-stats all read
    trade_in_estimates without an expires_at filter). The DELETE must stay
    scoped to created_by IS NULL -- the honest B2C population (129 rows in
    that same audit)."""
    sql = task_mod._DELETE_EXPIRED_ESTIMATES_SQL.text
    assert "created_by IS NULL" in sql
    assert not re.search(r":\w+::", sql)


def test_leads_sql_has_no_created_by_guard() -> None:
    """trade_in_leads has NO created_by column at all (never had a B2B/B2C
    split) -- its expires_at IS a genuine 180-day retention deadline (see
    migration 231), not a link/PDF-access TTL like trade_in_estimates. This
    documents the asymmetry explicitly so a future 'fix' doesn't bolt a
    created_by filter onto a table that doesn't have the column."""
    sql = task_mod._DELETE_EXPIRED_LEADS_SQL.text
    assert "created_by" not in sql


def test_sql_does_not_delete_whole_table_unbounded() -> None:
    """Neither statement is a bare `DELETE FROM table` -- both scope via a
    subselect + LIMIT batch."""
    statements = (
        task_mod._DELETE_EXPIRED_ESTIMATES_SQL.text,
        task_mod._DELETE_EXPIRED_LEADS_SQL.text,
    )
    for sql in statements:
        assert "WHERE id IN (" in sql
        assert "LIMIT" in sql


# ── failure path ─────────────────────────────────────────────────────────────


def test_failure_path_rollback_and_mark_failed(monkeypatch: pytest.MonkeyPatch) -> None:
    marked = _patch_runs(monkeypatch)

    class _BoomDB(_FakeDB):
        def execute(self, stmt: Any, params: dict[str, Any] | None = None) -> _FakeResult:
            # Call #1 (pre-flight, preflight_count=0) and call #2 (first estimates
            # batch) succeed and commit; call #3 (still draining estimates, or
            # first leads call) explodes. +1 vs pre-PR-D1 to admit the pre-flight.
            if len(self.executed) >= 2:
                raise RuntimeError("db exploded")
            return super().execute(stmt, params)

    db = _BoomDB([5])  # only ONE successful DELETE batch before the boom
    with pytest.raises(RuntimeError, match="db exploded"):
        task_mod.purge_expired_trade_in_data(db, run_id=6, batch_size=5, max_batches=20)  # type: ignore[arg-type]

    assert db.rolled_back is True
    assert marked["kind"] == "failed"
    assert marked["run_id"] == 6
    # Partial progress preserved in the counters passed to mark_failed (first
    # estimates batch of 5 already committed before the boom on the 2nd call).
    assert marked["counters"]["estimates_deleted"] == 5
    assert marked["counters"]["leads_deleted"] == 0


def test_idempotent_zero_rowcount_is_not_an_error(monkeypatch: pytest.MonkeyPatch) -> None:
    """Re-running against an already-drained backlog just deletes 0 rows, cleanly."""
    marked = _patch_runs(monkeypatch)
    db = _FakeDB([0, 0])
    out = task_mod.purge_expired_trade_in_data(db, run_id=7, batch_size=500, max_batches=20)  # type: ignore[arg-type]
    assert out == {"estimates_deleted": 0, "leads_deleted": 0}
    assert marked["kind"] == "done"


# ── migration 231 ────────────────────────────────────────────────────────────


def test_migration_231_exists() -> None:
    assert _MIGRATION_231.is_file(), f"missing migration: {_MIGRATION_231}"


def test_migration_231_is_transactional() -> None:
    sql = _MIGRATION_231.read_text("utf-8")
    assert "BEGIN;" in sql
    assert "COMMIT;" in sql


def test_migration_231_backfills_and_sets_not_null() -> None:
    sql = _MIGRATION_231.read_text("utf-8")
    assert "ADD COLUMN IF NOT EXISTS expires_at" in sql
    assert "WHERE expires_at IS NULL" in sql
    assert "SET NOT NULL" in sql
    assert "180 days" in sql


def test_migration_231_seeds_purge_schedule_disabled_by_default() -> None:
    sql = _MIGRATION_231.read_text("utf-8")
    assert "'purge_expired_trade_in_data'" in sql
    assert "ON CONFLICT (source) DO NOTHING" in sql
    # Seeded disabled -- first automated PII-DELETE job in trade-in deserves a
    # supervised first run before the scheduler can trigger it unattended.
    assert re.search(r"'purge_expired_trade_in_data',\s*\n\s*false,", sql)


def test_migration_231_no_psycopg_trap() -> None:
    sql = _MIGRATION_231.read_text("utf-8")
    assert not re.search(r":\w+::", sql)


# ── Optional real-Postgres behavioural test (self-skips without a DB) ──────────
# Same pattern as tests/test_house_dedup_merge.py::_live_session -- CI runs the
# mock-only lane (DATABASE_URL is a placeholder), so this self-skips there; it
# only executes with a real reachable Postgres (e.g. local dev DB).


def _live_session() -> Any | None:
    """Return a SQLAlchemy Session if a non-placeholder Postgres is reachable, else None."""
    try:
        from sqlalchemy import create_engine
        from sqlalchemy.orm import sessionmaker

        dsn = os.environ.get("TEST_DATABASE_URL") or os.environ.get("DATABASE_URL", "")
        if not dsn or "localhost:5432/test" in dsn:
            return None
        engine = create_engine(dsn, future=True)
        conn = engine.connect()
        from sqlalchemy import text as _t

        conn.execute(_t("SELECT 1"))
        conn.close()
        return sessionmaker(bind=engine, future=True)()
    except Exception:
        return None


@pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB")
def test_real_purge_deletes_only_anonymous_expired_estimates() -> None:
    """End-to-end on a real DB: one expired B2B row (created_by set) and one
    expired anonymous row (created_by IS NULL) both exist -- a real run of
    purge_expired_trade_in_data must delete ONLY the anonymous one. This is
    the exact scenario the deep-review HIGH finding (2026-08-06) flagged:
    without the created_by IS NULL guard, the pilot row would also be gone."""
    from sqlalchemy import text as _t

    db = _live_session()
    assert db is not None
    anon_id = uuid4()
    pilot_id = uuid4()
    try:
        db.execute(
            _t(
                "INSERT INTO trade_in_estimates "
                "(id, address, area_m2, rooms, floor, total_floors, "
                " median_price, range_low, range_high, median_price_per_m2, confidence, "
                " expires_at, created_by) VALUES "
                "(CAST(:anon_id AS uuid), 'purge-test аноним', 40, 1, 2, 5, "
                " 5000000, 4500000, 5500000, 125000, 'low', "
                " NOW() - interval '1 hour', NULL), "
                "(CAST(:pilot_id AS uuid), 'purge-test пилот', 40, 1, 2, 5, "
                " 5000000, 4500000, 5500000, 125000, 'low', "
                " NOW() - interval '1 hour', 'pytest_purge_guard')"
            ),
            {"anon_id": str(anon_id), "pilot_id": str(pilot_id)},
        )
        db.commit()

        task_mod.purge_expired_trade_in_data(db, run_id=999999999, batch_size=100, max_batches=5)

        remaining_ids = {
            str(r)
            for r in db.execute(
                _t("SELECT id FROM trade_in_estimates WHERE id = ANY(CAST(:ids AS uuid[]))"),
                {"ids": [str(anon_id), str(pilot_id)]},
            )
            .scalars()
            .all()
        }
        assert str(anon_id) not in remaining_ids, "anonymous expired row must be purged"
        assert str(pilot_id) in remaining_ids, "B2B pilot row must survive despite expires_at"
    finally:
        db.execute(
            _t("DELETE FROM trade_in_estimates WHERE id = ANY(CAST(:ids AS uuid[]))"),
            {"ids": [str(anon_id), str(pilot_id)]},
        )
        db.commit()
        db.close()


@pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB")
def test_real_preflight_ignores_healthy_paid_row_flags_only_anomaly() -> None:
    """Deep-review finding 2026-08-06 MEDIUM on PR #2754, reproduced exactly
    against a real DB: a HEALTHY paid row (retain_until set, a payments row
    exists) is the ordinary steady state one day after every sale and must NOT
    raise the pre-flight count; a row where fulfillment failed to set
    retain_until despite a payments row existing is the real ANOMALY and must.
    Baseline-delta assertions (not absolute counts) so this is safe to run
    against a dev DB that may already contain unrelated rows."""
    from sqlalchemy import text as _t

    db = _live_session()
    assert db is not None
    healthy_id = uuid4()
    anomaly_id = uuid4()
    healthy_order = f"pytest-healthy-{uuid4().hex[:12]}"
    anomaly_order = f"pytest-anomaly-{uuid4().hex[:12]}"
    try:
        baseline = task_mod._preflight_paid_candidates(db)

        # Healthy: retain_until set (paid, safeguard 1 already protects it) +
        # a payments row -- exactly what every successful sale looks like a day
        # later. Must NOT move the pre-flight count.
        db.execute(
            _t(
                "INSERT INTO trade_in_estimates "
                "(id, address, area_m2, rooms, floor, total_floors, "
                " median_price, range_low, range_high, median_price_per_m2, confidence, "
                " expires_at, created_by, retain_until) VALUES "
                "(CAST(:id AS uuid), 'purge-test здоровая оплаченная', 40, 1, 2, 5, "
                " 5000000, 4500000, 5500000, 125000, 'low', "
                " NOW() - interval '1 hour', NULL, NOW() + interval '363 days')"
            ),
            {"id": str(healthy_id)},
        )
        db.execute(
            _t(
                "INSERT INTO payments "
                "(order_id, terminal_key, product_code, amount_kopecks, status, estimate_id) "
                "VALUES (:order_id, 'pytest_terminal', 'trade_in_report', 15000, 'CONFIRMED', "
                " CAST(:id AS uuid))"
            ),
            {"order_id": healthy_order, "id": str(healthy_id)},
        )
        db.commit()
        assert task_mod._preflight_paid_candidates(db) == baseline, (
            "healthy paid row (retain_until set) must NOT raise the pre-flight count"
        )

        # Anomaly: retain_until NULL despite a payments row existing -- exactly
        # the case the two DELETE safeguards exist for. Must raise by exactly one.
        db.execute(
            _t(
                "INSERT INTO trade_in_estimates "
                "(id, address, area_m2, rooms, floor, total_floors, "
                " median_price, range_low, range_high, median_price_per_m2, confidence, "
                " expires_at, created_by, retain_until) VALUES "
                "(CAST(:id AS uuid), 'purge-test настоящая аномалия', 40, 1, 2, 5, "
                " 5000000, 4500000, 5500000, 125000, 'low', "
                " NOW() - interval '1 hour', NULL, NULL)"
            ),
            {"id": str(anomaly_id)},
        )
        db.execute(
            _t(
                "INSERT INTO payments "
                "(order_id, terminal_key, product_code, amount_kopecks, status, estimate_id) "
                "VALUES (:order_id, 'pytest_terminal', 'trade_in_report', 15000, 'CONFIRMED', "
                " CAST(:id AS uuid))"
            ),
            {"order_id": anomaly_order, "id": str(anomaly_id)},
        )
        db.commit()
        assert task_mod._preflight_paid_candidates(db) == baseline + 1, (
            "anomaly row (retain_until unset, payments row exists) must raise "
            "the pre-flight count by exactly one"
        )
    finally:
        db.execute(
            _t("DELETE FROM payments WHERE order_id = ANY(CAST(:orders AS text[]))"),
            {"orders": [healthy_order, anomaly_order]},
        )
        db.execute(
            _t("DELETE FROM trade_in_estimates WHERE id = ANY(CAST(:ids AS uuid[]))"),
            {"ids": [str(healthy_id), str(anomaly_id)]},
        )
        db.commit()
        db.close()


@pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB")
def test_real_purge_not_wedged_by_healthy_paid_row() -> None:
    """Deep-review finding 2026-08-06 MEDIUM on PR #2754: before the fix, a
    healthy paid row anywhere in the table (retain_until set, has a payments
    row) permanently wedged the job -- the very first successful sale would
    have made every subsequent scheduled run abort in mark_failed with zero
    deletions FOREVER, silently taking 180-day leads purge (152-ФЗ) down with
    it (leads purge runs from the same function AFTER the pre-flight check).
    This proves a real end-to-end run completes normally (mark_done) in the
    presence of such a row."""
    from sqlalchemy import text as _t

    db = _live_session()
    assert db is not None
    healthy_id = uuid4()
    healthy_order = f"pytest-wedge-{uuid4().hex[:12]}"
    try:
        db.execute(
            _t(
                "INSERT INTO trade_in_estimates "
                "(id, address, area_m2, rooms, floor, total_floors, "
                " median_price, range_low, range_high, median_price_per_m2, confidence, "
                " expires_at, created_by, retain_until) VALUES "
                "(CAST(:id AS uuid), 'purge-test не блокирует джобу', 40, 1, 2, 5, "
                " 5000000, 4500000, 5500000, 125000, 'low', "
                " NOW() - interval '1 hour', NULL, NOW() + interval '363 days')"
            ),
            {"id": str(healthy_id)},
        )
        db.execute(
            _t(
                "INSERT INTO payments "
                "(order_id, terminal_key, product_code, amount_kopecks, status, estimate_id) "
                "VALUES (:order_id, 'pytest_terminal', 'trade_in_report', 15000, 'CONFIRMED', "
                " CAST(:id AS uuid))"
            ),
            {"order_id": healthy_order, "id": str(healthy_id)},
        )
        db.commit()

        # Must complete normally -- no RuntimeError, no mark_failed short-circuit
        # (would raise before reaching this line if the bug were still present).
        result = task_mod.purge_expired_trade_in_data(
            db, run_id=999999998, batch_size=100, max_batches=1
        )
        assert set(result) == {"estimates_deleted", "leads_deleted"}, (
            "leads purge must also have run -- it is NOT reachable when the "
            "pre-flight wrongly aborts first"
        )

        still_there = db.execute(
            _t("SELECT id FROM trade_in_estimates WHERE id = CAST(:id AS uuid)"),
            {"id": str(healthy_id)},
        ).fetchone()
        assert still_there is not None, "healthy paid row must survive the run untouched"
    finally:
        db.execute(
            _t("DELETE FROM payments WHERE order_id = :order_id"), {"order_id": healthy_order}
        )
        db.execute(
            _t("DELETE FROM trade_in_estimates WHERE id = CAST(:id AS uuid)"),
            {"id": str(healthy_id)},
        )
        db.commit()
        db.close()