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

WHY:
    trade_in_estimates.expires_at (и, начиная с migration 193, 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 193 -- 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.

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.
"""

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

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

_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 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}.
    """
    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}
    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