"""Transaction-isolation primitives for best-effort work.

Every "this step is best-effort, it must never break the caller" block in
this codebase needs the *same* two-tier error handling, and hand-rolling it
has gone wrong repeatedly:

1. **A swallowed DB error does not leave the session usable.** A failed
   ``flush()`` deactivates the whole transaction, so ``except Exception``
   "succeeds" but the *next* statement raises ``PendingRollbackError`` /
   ``InvalidRequestError``. The 2026-08-06 Zoom-seat incident is the
   canonical case: an ``IntegrityError`` on a queued provisioning task was
   swallowed, the following ``AuditService.log`` blew up, the whole Feishu
   approval callback rolled back, and the request sat ``pending`` for 20
   hours while the retry cron failed identically ~80 times. The fix is a
   SAVEPOINT around the fallible step.

2. **A SAVEPOINT cannot rescue every error.** Lock-wait (1205), deadlock
   (1213) and orphaned-savepoint (1305) abort the transaction *server*-side;
   the savepoint rollback itself fails. Swallowing one of those turns a
   transient, retryable deadlock into a ``SAVEPOINT ... does not exist``
   cascade that silently skips every remaining item in a batch. Those MUST
   propagate so the caller can restart on a fresh session.

Getting (1) right and forgetting (2) — or the reverse — is the failure mode.
:func:`best_effort` encodes both, so call sites express intent
("this step may fail, keep going") instead of re-deriving the mechanics.

Usage::

    async with best_effort(db, label="enqueue grant for seat") as step:
        task_id = await enqueue_grant_for_seat(db, seat=seat, trace_id=t)
    if not step.ok:
        task_id = None            # the savepoint rolled the attempt back
        ...surface it...

The session is guaranteed usable after the block whether or not the body
succeeded, so the audit write / status flip that follows always lands.

``scripts/check_savepoint_isolation.py`` enforces this at CI time: a new
``except Exception`` wrapping DB work without one of these primitives fails
the build.

**Do NOT wrap a block that commits and then keeps working.**
``session.commit()`` inside an open SAVEPOINT closes the nested transaction.
Measured against real MySQL: if the commit is the block's LAST database
statement this is harmless (``step.ok`` stays True, the data is committed),
but any DB work AFTER it raises ``InvalidRequestError: Can't operate on
closed transaction inside context manager`` — that work is **silently
dropped** and the block reports a false failure. Such a block needs a
different remedy, usually ``await db.rollback()`` at the top of its handler
before any recovery bookkeeping (see
``ApprovalService._defer_asset_reconcile_failure``). Mark those sites
``# savepoint-exempt: <reason>``; the CI guard cannot detect an internal
commit on its own. Learned 2026-08-06: wrapping the approval
submission-intent recovery loop broke 13 idempotency tests — it commits
per intent and then keeps writing.
"""

from __future__ import annotations

import logging
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from dataclasses import dataclass, field
from typing import Any, Protocol

from sqlalchemy.ext.asyncio import AsyncSession

from app.core.exceptions import (
    format_exception_for_log,
    is_connection_lost_db_error,
    is_transaction_aborting_db_error,
)


class _LoggerLike(Protocol):
    """What ``best_effort`` needs from a logger.

    Deliberately structural: most of this codebase logs through
    ``app.core.safe_logging.SafeLogger``, which **intentionally does not
    subclass** ``logging.Logger`` (it confines the stdlib call to one
    sanitizing ``_emit`` so static data-flow analysis has an explicit trust
    boundary). Typing the parameter as ``logging.Logger`` would therefore
    reject nearly every real call site.
    """

    def log(self, level: int, msg: object, *args: object) -> None:
        ...

    def warning(self, msg: object, *args: object) -> None:
        ...

    def debug(self, msg: object, *args: object) -> None:
        ...


logger: _LoggerLike = logging.getLogger(__name__)


@dataclass
class BestEffortOutcome:
    """Result handle yielded by :func:`best_effort`.

    ``ok`` is False when the body raised a savepoint-local error; ``error``
    carries it for surfacing to the caller (a response field, a summary
    counter, an audit row). Transaction-aborting errors never land here —
    they propagate out of the context manager.
    """

    label: str
    error: BaseException | None = None
    detail: dict[str, Any] = field(default_factory=dict)

    @property
    def ok(self) -> bool:
        return self.error is None

    @property
    def error_type(self) -> str | None:
        return type(self.error).__name__ if self.error is not None else None

    def reason(self, *, max_len: int = 300) -> str | None:
        """Short, log/response-safe description of the failure.

        Delegates to :func:`format_exception_for_log` — the codebase's single
        canonical formatter — rather than ``f"{type(exc).__name__}: {exc}"``.
        That naive form is a known bug class here: it renders an
        ``AppException`` as the useless 15-char sentinel ``"AppException: "``
        and drops the structured ``.code``, which once left 14 cloud-access
        blockers stuck on that empty string for days while masking a real
        RBAC FORBIDDEN. Caught again 2026-08-06 when the SSO CSV-import
        diagnostic test went red after this outcome replaced a direct
        ``format_exception_for_log`` call.
        """
        if self.error is None:
            return None
        return format_exception_for_log(self.error, max_len=max_len)


@asynccontextmanager
async def best_effort(
    db: AsyncSession,
    *,
    label: str,
    log: _LoggerLike | None = None,
    log_level: int = logging.WARNING,
) -> AsyncIterator[BestEffortOutcome]:
    """SAVEPOINT-isolate a fallible step and keep the session usable.

    On a savepoint-local failure (IntegrityError, AppException, a bug in
    the body, …) the savepoint is rolled back, the error is recorded on the
    yielded outcome and logged, and execution continues after the block.

    On a transaction-aborting DB error (1205 / 1213 / 1305) the exception
    **propagates** — the surrounding transaction is already dead server-side
    and the caller must restart, not continue.

    Args:
        db: session to open the SAVEPOINT on.
        label: human-readable name of the step, used in logs and in
            ``outcome.detail``. Write it as the action ("enqueue grant for
            seat"), not the failure.
        log: logger to report on; defaults to this module's.
        log_level: level for the swallowed-failure message.
    """
    outcome = BestEffortOutcome(label=label)
    emit = log or logger
    try:
        async with db.begin_nested():
            yield outcome
    except Exception as exc:  # noqa: BLE001 — the whole point of this helper
        # Two unsurvivable families, both of which make the "session is
        # usable afterwards" guarantee false if swallowed:
        #   * transaction-aborting (1205/1213/1305) — the transaction is
        #     already gone server-side, the SAVEPOINT rollback cannot rescue
        #     it;
        #   * connection-lost (2006/2013) — a TiDB node recycle mid-block
        #     leaves nothing to roll back TO.
        # The codebase's cron retry predicate already combined both
        # (``scheduled_jobs/handlers.py``); this helper must too, or every
        # caller believes the session survived a dropped connection and dies
        # on its next statement — the exact bug class this primitive exists
        # to eliminate.
        if is_transaction_aborting_db_error(exc) or is_connection_lost_db_error(exc):
            emit.warning(
                "best_effort(%s): unrecoverable DB error (%s) — propagating",
                label,
                type(exc).__name__,
            )
            raise
        outcome.error = exc
        # Log the MESSAGE, not just the type. Several call sites dropped
        # their own ``logger.warning(..., exc)`` when they moved onto the
        # outcome; without this the only trace of a failure is its class
        # name, which is not enough to act on.
        emit.log(
            log_level,
            "best_effort(%s) failed — savepoint rolled back, continuing: %s",
            label,
            outcome.reason(),
        )


@asynccontextmanager
async def savepoint(db: AsyncSession, *, label: str) -> AsyncIterator[None]:
    """SAVEPOINT that re-raises everything, including savepoint-local errors.

    Use when the step is *not* best-effort but you still need the failure to
    roll back only its own writes — e.g. a per-item loop whose caller wants
    the exception, or a nested unit inside a larger savepoint. Transaction-
    aborting errnos propagate identically; the difference from
    :func:`best_effort` is that ordinary errors do too.
    """
    try:
        async with db.begin_nested():
            yield
    except Exception:
        logger.debug("savepoint(%s) rolled back", label)
        raise