"""Physically delete expired personal data — ЭТАП 4 B2C retention enforcement (152-ФЗ).

WHY:
    trade_in_estimates.expires_at (и, начиная с migration 231, trade_in_leads.expires_at)
    defined a retention window, but neither table had any background job that actually
    DELETEd rows once expired -- expires_at was used ONLY as a read-time filter
    (GET /estimate/{id}: "AND expires_at > NOW()"). Personal data (address / phone)
    outlived its declared lifetime indefinitely, contradicting the retention policy
    shown to the user.

WHAT:
    Batched physical DELETE for both tables, run nightly by the kit-scheduler (see
    app.services.product_handlers._job_purge_expired_trade_in_data, scrape_schedules
    row seeded by migration 231 -- seeded enabled=false, see that migration's docstring
    for why). Same architecture as app/tasks/deactivate_stale_avito.py (sync, DB-only,
    invoked via run_in_executor from the async kit handler).

    - trade_in_estimates: ON DELETE CASCADE already cleans up estimate_photos
      (007_estimate_photos.sql) and avito_imv_evaluations (018_avito_imv_evaluations.sql)
      for each deleted estimate.
    - trade_in_leads: ON DELETE SET NULL on trade_in_leads.estimate_id (172_trade_in_leads.sql)
      means a lead created from a now-purged estimate SURVIVES with estimate_id nulled --
      it has its OWN retention clock (trade_in_leads.expires_at) and its own PII (phone),
      purged independently below.

⚠️ trade_in_estimates DELETE is scoped to `created_by IS NULL` (deep-review finding,
   2026-08-06, HIGH): `expires_at` on this table is set UNCONDITIONALLY for every
   estimate, B2B pilot or anonymous (`now + settings.trade_in_estimate_retention_hours`,
   see app/services/estimator.py) -- it is a TTL on the ESTIMATE LINK/PDF staying
   resolvable (GET /estimate/{id}: 404 past expiry, PDF export: 410 past expiry), NOT a
   declared retention deadline for the ROW. B2B pilots' consent is closed by contract
   (see migration 229's `consent` column asymmetry: NULL for `created_by IS NOT NULL`,
   the exact same B2B-vs-B2C split used here) and their estimates are the live basis for
   /trade-in/history, /team/employees/{id}/history and the cache-stats dashboards (see
   app/api/v1/trade_in.py, app/api/v1/team.py) -- deleting them past a 24h *link* TTL
   would be silent, irreversible data loss of pilots' own operational data, not a
   privacy-driven cleanup. Audited against prod on 2026-08-06: of 1057 rows, 1040 had
   already crossed `expires_at`, and 911 of THOSE belonged to named pilots
   (`created_by` set: admin, kopylov, brusnika, praktika, pilottest, admintest, user1).
   Without the `created_by IS NULL` guard, one unattended run at
   `batch_size=500, max_batches=20` (the defaults) would have deleted essentially the
   whole table. `created_by IS NULL` is the honest B2C population -- 129 rows in that
   same audit. trade_in_leads has no `created_by` column (never had a B2B/B2C split --
   its own `expires_at` really is a 180-day retention deadline, not a link TTL, see
   migration 231) so its DELETE below is intentionally NOT scoped the same way.

BATCHING (не единый DELETE по всей таблице):
    Each table is drained in batches of `batch_size` rows (default
    settings.trade_in_purge_batch_size), each batch its OWN statement + its OWN commit
    (bounds lock/transaction duration on a backlog). A run stops draining a table once a
    batch returns fewer rows than batch_size (caught up) OR after `max_batches` iterations
    (safety cap on total run duration -- any remaining backlog drains over subsequent
    nightly runs, not one giant transaction). Idempotent: rows already deleted simply
    don't match `expires_at < NOW()` on the next run; a mid-run failure leaves earlier
    committed batches deleted (correct, not rolled back) and mark_failed records the
    partial counters reached so far.

Payments retention (PR #2754): the `created_by IS NULL` population above is EXACTLY
    the future paying-customer population -- the owner sells this report to
    individuals for money, and a paid row must outlive the 24h `expires_at` link TTL
    (a separate column, `retain_until`, set by the -- separate, not-yet-existing --
    payment fulfillment code to now() + settings.trade_in_paid_retention_days, NOT
    a change to `expires_at` itself). Two independent safeguards were added to
    `_DELETE_EXPIRED_ESTIMATES_SQL` (retain_until IS NULL + NOT EXISTS payments)
    plus a pre-flight count in `purge_expired_trade_in_data` that refuses to run at
    all if it finds an ANOMALOUS paid candidate -- see the SQL constants and
    `_preflight_paid_candidates` below for the mechanics (deep-review finding
    2026-08-06 MEDIUM on PR #2754: the pre-flight predicate itself must ALSO carry
    `retain_until IS NULL`, otherwise a perfectly healthy paid row trips it and
    wedges the job permanently -- see that function's docstring). No payment code
    lives in this file.
"""

from __future__ import annotations

import logging
from typing import Any

from sqlalchemy import text
from sqlalchemy.orm import Session

from app.core.config import settings
from app.services import scrape_runs as runs_mod

logger = logging.getLogger(__name__)

# Safety cap on batches per table per run -- bounds a single scheduled run's total
# duration even if the backlog is much larger than batch_size * max_batches; the
# remainder simply drains on the next nightly run (idempotent, no data loss risk).
_DEFAULT_MAX_BATCHES = 20

#
# Payments retention (2026-08-06, PR #2754): два независимых предохранителя
# добавлены к тому же предикату ПЕРЕД тем, как платёжный код появился в
# проекте (мина уже была заряжена: без них джоба удаляла бы будущих платящих
# клиентов). Отдельная колонка retain_until (не подъём expires_at) — потому
# что expires_at глобальный TTL расчёта на ВСЕ строки (включая неоплаченные)
# и печатается в PDF/UI как «актуальность расчёта»; поднять его до года
# означало бы одновременно нарушить минимизацию ПДн по 152-ФЗ и соврать в
# документе клиента про срок актуальности цифры:
#   1. `retain_until IS NULL` — именно IS NULL, НЕ `< NOW()`. Оплаченная
#      строка (retain_until IS NOT NULL, migration 240) не удаляется джобой
#      В ПРИНЦИПЕ, пока не поднято ослабление отдельным PR не раньше чем
#      через год после первой продажи. `retain_until` ставится сервисным
#      кодом платёжного контура (ещё не существует в этом PR) на now() +
#      settings.trade_in_paid_retention_days.
#   2. `NOT EXISTS (payments)` — независимая страховка на случай, если выдача
#      забыла проставить retain_until (баг/гонка/ручной INSERT): строка,
#      которой коснулись деньги, переживёт джобу даже без корректного (1).
#      `payments` создана migration 233 (payments_estimate_idx — дешёвый терм).
# См. также _preflight_paid_candidates ниже — та же логика ДО первого батча.
_DELETE_EXPIRED_ESTIMATES_SQL = text(
    """
    DELETE FROM trade_in_estimates
    WHERE id IN (
        SELECT id FROM trade_in_estimates
        WHERE expires_at < NOW()
          AND created_by IS NULL
          AND retain_until IS NULL
          AND NOT EXISTS (
              SELECT 1 FROM payments p WHERE p.estimate_id = trade_in_estimates.id
          )
        ORDER BY expires_at
        LIMIT CAST(:batch_size AS int)
    )
    """
)

# Pre-flight (см. _preflight_paid_candidates). deep-review finding 2026-08-06
# MEDIUM (PR #2754): первая редакция считала по БАЗОВОМУ предикату БЕЗ
# retain_until вообще -- а это ловит и штатно-здоровые оплаченные строки
# (retain_until проставлен, есть payments) точно так же, как настоящую
# аномалию (retain_until НЕ проставлен, но payments есть) -- джоба вставала
# на первой же честной продаже и больше никогда не запускалась (вместе с ней
# вставало и удаление лидов, вызываемое из той же функции ПОСЛЕ этой
# проверки -- 180-дневный purge по 152-ФЗ тоже переставал бы работать).
# Правильная форма: базовый предикат AND "новый предохранитель НЕ сработал
# бы" (retain_until IS NULL) AND "признак аномалии" (payments всё же есть).
# Здоровая оплаченная строка (retain_until IS NOT NULL) исключается ЭТИМ
# термом -- она и так под DELETE не попадает (см. safeguard 1 выше), тревогу
# поднимать не должна.
_PREFLIGHT_PAID_CANDIDATES_SQL = text(
    """
    SELECT count(*) FROM trade_in_estimates e
    WHERE e.expires_at < NOW()
      AND e.created_by IS NULL
      AND e.retain_until IS NULL
      AND EXISTS (SELECT 1 FROM payments p WHERE p.estimate_id = e.id)
    """
)

_DELETE_EXPIRED_LEADS_SQL = text(
    """
    DELETE FROM trade_in_leads
    WHERE id IN (
        SELECT id FROM trade_in_leads
        WHERE expires_at < NOW()
        ORDER BY expires_at
        LIMIT CAST(:batch_size AS int)
    )
    """
)


def _drain_expired(
    db: Session,
    stmt: Any,
    *,
    batch_size: int,
    max_batches: int,
    label: str,
    counters: dict[str, int],
    counter_key: str,
) -> None:
    """Run `stmt` (one bounded DELETE batch) repeatedly until caught up or capped.

    Commits after EVERY batch -- keeps each individual transaction/lock short even
    when the backlog is large. Updates `counters[counter_key]` INCREMENTALLY (not
    just once at the end) so that a mid-run exception on a LATER batch still leaves
    an accurate count of what was actually deleted-and-committed by earlier batches
    -- those rows are gone for real (commit already happened) whether or not this
    function ever returns normally.
    """
    for batch_num in range(1, max_batches + 1):
        result = db.execute(stmt, {"batch_size": batch_size})
        deleted = result.rowcount or 0
        db.commit()
        counters[counter_key] += deleted
        logger.info(
            "purge_expired_trade_in_data: %s batch=%d deleted=%d (running_total=%d)",
            label,
            batch_num,
            deleted,
            counters[counter_key],
        )
        if deleted < batch_size:
            break  # caught up -- fewer expired rows left than one batch


def _preflight_paid_candidates(db: Session) -> int:
    """Safety gate: count ANOMALOUS purge-candidates -- base predicate, retain_until
    IS NULL (safeguard 1 did NOT protect the row), AND a payments row exists anyway.

    Runs BEFORE any DELETE batch. A non-zero result means fulfillment failed to set
    `retain_until` on a row money actually touched (bug/race/manual INSERT) -- this
    run must not delete anything; see `purge_expired_trade_in_data` below, which
    aborts before the first batch when this returns non-zero.

    MUST include `retain_until IS NULL` (deep-review finding 2026-08-06 MEDIUM, PR
    #2754): a healthy paid row (retain_until set, has a payments row) is the EXPECTED
    steady state one day after every sale -- without this term it counts as a "paid
    candidate" too, so the very first successful sale permanently wedges this job
    (mark_failed, zero deletions, forever -- and since leads purge runs from the same
    function AFTER this check, the unrelated 180-day lead retention would also stop).
    """
    return db.execute(_PREFLIGHT_PAID_CANDIDATES_SQL).scalar_one()


def purge_expired_trade_in_data(
    db: Session,
    run_id: int,
    *,
    batch_size: int | None = None,
    max_batches: int | None = None,
) -> dict[str, int]:
    """Delete expired rows from trade_in_estimates + trade_in_leads, in bounded batches.

    Sync (invoked via run_in_executor from the kit-scheduler handler, same pattern as
    deactivate_stale_listings). Finalises the scrape_runs row (mark_done / mark_failed).

    Returns {"estimates_deleted": N, "leads_deleted": M}.

    Payments retention pre-flight (see `_preflight_paid_candidates`): if any
    purge-candidate estimate has `retain_until IS NULL` AND a `payments` row (the
    ANOMALY -- fulfillment failed to set retain_until on a row money touched), the
    run aborts BEFORE the first DELETE batch (estimates OR leads) -- zero rows
    deleted, `mark_failed` records why. Healthy paid rows (retain_until set) do NOT
    trip this -- they never matched the check to begin with. This is deliberately
    checked outside the `try` below so it can never be caught and silently
    re-reported as a generic mid-run failure -- it is a distinct, actionable
    pre-condition failure.
    """
    batch_size = batch_size or settings.trade_in_purge_batch_size
    max_batches = max_batches or _DEFAULT_MAX_BATCHES
    counters: dict[str, int] = {"estimates_deleted": 0, "leads_deleted": 0}

    paid_candidates = _preflight_paid_candidates(db)
    if paid_candidates:
        error = (
            f"pre-flight abort: {paid_candidates} purge-candidate trade_in_estimates "
            "row(s) have retain_until IS NULL but a matching payments row -- "
            "refusing to run, zero rows deleted"
        )
        logger.error("purge_expired_trade_in_data run_id=%d %s", run_id, error)
        runs_mod.mark_failed(db, run_id, error, counters)
        raise RuntimeError(error)

    try:
        _drain_expired(
            db,
            _DELETE_EXPIRED_ESTIMATES_SQL,
            batch_size=batch_size,
            max_batches=max_batches,
            label="trade_in_estimates",
            counters=counters,
            counter_key="estimates_deleted",
        )
        _drain_expired(
            db,
            _DELETE_EXPIRED_LEADS_SQL,
            batch_size=batch_size,
            max_batches=max_batches,
            label="trade_in_leads",
            counters=counters,
            counter_key="leads_deleted",
        )
        runs_mod.mark_done(db, run_id, counters)
        logger.info(
            "purge_expired_trade_in_data run_id=%d done: estimates_deleted=%d leads_deleted=%d",
            run_id,
            counters["estimates_deleted"],
            counters["leads_deleted"],
        )
        return counters
    except Exception as exc:
        logger.exception("purge_expired_trade_in_data run_id=%d failed", run_id)
        db.rollback()
        runs_mod.mark_failed(db, run_id, str(exc)[:1000], counters)
        raise