"""One click, and the platform does the rest.

The shape this module refuses to have: an admin fixes an expired
credential, and the platform hands back a list of 47 failed rows for
them to click through one at a time. That is the failure mode this
whole epic exists to remove — it is technically correct, fully
audited, and completely useless, because the person already made the
only decision there was to make. Forwarding the consequences of that
decision back to them is asking them to do the work twice.

So: **the admin says "the credential is fixed", and this module
re-drives every open exception under that integration.** It succeeds or
it does not; either way the human is not asked again. They get one
answer at the end.

Four properties make that safe:

**It drives existing primitives, never new ones.** A provisioning task
goes back through ``ProvisioningService.retry_task``; a dead outbox row
through ``NotificationService.resend_outbox_row``; a blocked identity
sync through ``sync_person_to_target``; an integration through its own
``test_connection``. Each of those already has its own locking, its own
audit row, and its own idempotency. A parallel retry path here would be
a second implementation of four things that already work, and the two
would drift.

**It is bounded.** ``MAX_REPLAY_PER_RUN`` caps one invocation. An
unbounded replay across a bad deploy's worth of failures would be a
self-inflicted outage against a vendor that just started working
again — the exact thing that produced half these exceptions.

**Only success closes anything.** A replayed row that fails goes back
to the queue and stays open. The collector's own recovery rules
(PR1) then decide whether it recovered, using the same evidence they
use for everything else. This module never writes ``recovered_at``
itself: a replay that returned 200 is not proof the underlying work
succeeded, and inventing that proof here would put a lie in the one
table built to hold the truth about failures.

**Every attempt is recorded, including the ones that changed nothing.**
``last_replay_at`` moves even on failure, so a second click inside the
cooldown is visibly a no-op rather than silently doubling the vendor
load. These three columns (``replayable``, ``replay_attempt_count``,
``last_replay_at`` / ``last_replay_error``) were all shipped by PR1 for
this PR to write — this module adds no schema of its own.
"""
from __future__ import annotations

import logging
import uuid
from dataclasses import dataclass, field
from datetime import UTC, datetime, timedelta
from typing import Any

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.txn import best_effort
from app.domains.automation_exception.models import (
    AutomationException,
    AutomationExceptionOccurrence,
)
from app.domains.automation_exception.taxonomy import (
    SOURCE_IDENTITY_SYNC,
    SOURCE_NOTIFICATION,
    SOURCE_PROVISIONING,
)
from app.domains.automation_exception.timeutil import aware

logger = logging.getLogger(__name__)

#: Hard cap on rows re-driven in one invocation. A vendor that just came
#: back up does not need a bad deploy's worth of retries in one second.
#: What is left is picked up by the next click or the next collector
#: tick; nothing is lost by going slowly.
MAX_REPLAY_PER_RUN = 200

#: A second click inside this window is a no-op. Guards the double-click
#: and the impatient refresh, both of which are ordinary human
#: behaviour and neither of which should double the vendor load.
REPLAY_COOLDOWN = timedelta(minutes=2)

#: Sources this module knows how to re-drive.
#:
#: 🔴 This is a DISPATCH table, **not** the gate. The gate is
#: ``AutomationExceptionOccurrence.replayable``, which PR1 established
#: and which the collector that produced the row sets, because only it
#: knows whether that particular row carries enough to retry. An
#: earlier draft of this module gated on source instead and thereby
#: created a second, CONTRADICTORY definition: it would have re-driven
#: ``integration`` rows, which ``integration_source`` deliberately
#: marks ``replayable=False``. Two authorities for one question is how
#: they drift; there is one, and it is the row.
#:
#: ``scheduled_job`` has no handler on purpose: a run is a historical
#: event, and "replaying" it would mean triggering the whole job — a
#: much larger blast radius that must not hide behind this button.
#: ``integration`` has none either: re-testing a connection is not
#: re-driving failed work, and the integration-level entry point below
#: re-drives the OTHER rows underneath it, which is the useful action.

@dataclass
class ReplayOutcome:
    """What one replay run actually did.

    ``attempted`` and ``succeeded`` are counted separately from
    ``skipped`` because they answer different questions: whether the
    platform did any work, and whether the admin needs to look again.
    """

    attempted: int = 0
    succeeded: int = 0
    failed: int = 0
    skipped_cooldown: int = 0
    skipped_unsupported: int = 0
    exceptions_touched: set[uuid.UUID] = field(default_factory=set)
    errors: list[str] = field(default_factory=list)

    def as_dict(self) -> dict[str, Any]:
        return {
            "attempted": self.attempted,
            "succeeded": self.succeeded,
            "failed": self.failed,
            "skipped_cooldown": self.skipped_cooldown,
            "skipped_unsupported": self.skipped_unsupported,
            "exceptions_touched": len(self.exceptions_touched),
            # Bounded and de-duplicated: this reaches a toast, and 200
            # copies of one vendor message is not a message.
            "errors": self.errors[:5],
        }


async def _replay_provisioning(
    db: AsyncSession, occ: AutomationExceptionOccurrence, *, actor_id: uuid.UUID,
    trace_id: str,
) -> tuple[bool, str | None]:
    """Put a failed provisioning task back in the queue.

    Uses the domain's own ``retry_task``, which row-locks so two
    concurrent retries cannot both reset the attempt count, and writes
    its own audit row. We are a caller, not a second implementation.
    """
    from app.domains.provisioning.service import ProvisioningService  # noqa: PLC0415

    try:
        task_id = uuid.UUID(occ.source_ref_id)
    except (ValueError, AttributeError):
        return False, f"unparseable task id {occ.source_ref_id!r}"
    # SAVEPOINT, not a bare ``except``: ``retry_task`` writes, and a
    # write that dies on flush deactivates the WHOLE transaction. A
    # swallowed one would leave every remaining row in this batch —
    # and the trailing ``db.flush()`` — raising ``PendingRollbackError``,
    # which is the opposite of this module's "one row must not stop the
    # other 199" contract. The savepoint is what makes that contract
    # true rather than merely written down.
    async with best_effort(
        db, label="replay provisioning task", log=logger,
    ) as step:
        await ProvisioningService(db).retry_task(
            task_id,
            reset_assignee=False,
            trace_id=trace_id,
            actor_user_id=actor_id,
            actor_is_admin=True,
        )
    if not step.ok:
        return False, step.reason(max_len=200)
    return True, None


async def _replay_outbox(
    db: AsyncSession, occ: AutomationExceptionOccurrence, *, actor_id: uuid.UUID,
    trace_id: str,
) -> tuple[bool, str | None]:
    """Re-dispatch a dead notification.

    ``resend_outbox_row`` clones the row into a fresh dispatch rather
    than mutating the dead one, which is what keeps the delivery
    history honest.
    """
    from app.notifications.models import ACK_NOTE_PREFIX, NotificationOutbox  # noqa: PLC0415
    from app.notifications.service import NotificationService  # noqa: PLC0415

    try:
        outbox_id = uuid.UUID(occ.source_ref_id)
    except (ValueError, AttributeError):
        return False, f"unparseable outbox id {occ.source_ref_id!r}"
    # SAVEPOINT for the same reason as ``_replay_provisioning``. The
    # result is captured into a holder rather than a local because a
    # rolled-back savepoint leaves the name unbound.
    holder: dict[str, Any] = {}
    async with best_effort(
        db, label="resend dead notification", log=logger,
    ) as step:
        holder["result"] = await NotificationService(db).resend_outbox_row(
            outbox_id, actor_user_id=actor_id, trace_id=trace_id,
        )
    if not step.ok:
        return False, step.reason(max_len=200)
    result = holder.get("result")

    # ``resend_outbox_row`` REFUSES some rows without raising — a
    # departed recipient, a recipient not linked to a person. Treating
    # those as success told the admin "all re-driven" when nothing was
    # sent, and cleared the stored error that explained why.
    if isinstance(result, dict) and result.get("status") == "skipped":
        return False, str(result.get("reason") or "resend refused")[:200]

    # Close the SOURCE row, or this occurrence can never recover.
    # ``_resend_one`` clones into a fresh dispatch and deliberately does
    # not touch the original, so the dead row stays dead and
    # unacknowledged — which is precisely the set
    # ``notification_source`` scans. Without this the same outbox id is
    # a candidate on every tick forever, the parent never resolves, and
    # nobody ever gets the recovery echo.
    #
    # Done through the domain's own ``cancel_outbox`` rather than by
    # hand-writing ``ACK_NOTE_PREFIX``: the marker's docstring says
    # every reader must key off one constant, and the same applies to
    # writers — a second place that formats the note is a second place
    # that can format it wrongly. It also means we are not asserting
    # recovery ourselves; the collector observes the closed row on its
    # own next pass, so recovery still comes from evidence.
    row = await db.get(NotificationOutbox, outbox_id)
    if row is not None and not (row.last_error or "").startswith(ACK_NOTE_PREFIX):
        # Its own savepoint, so a failure to tidy up cannot roll back
        # the resend that already succeeded, nor poison the batch.
        async with best_effort(
            db, label="close the resent outbox row", log=logger,
        ) as tidy:
            await NotificationService(db).cancel_outbox(
                outbox_id,
                actor_user_id=actor_id,
                reason="resent by automation-exception replay",
            )
        if not tidy.ok:
            # Deliberately not turned into a failure: the notification
            # WAS re-dispatched. Failing to tidy the old row leaves the
            # exception open, which is visible and safe; reporting the
            # whole replay as failed would be a lie about what happened.
            logger.warning(
                "resent outbox row but could not close the original (id=%s)",
                outbox_id,
            )
    return True, None


async def _replay_identity_sync(
    db: AsyncSession, occ: AutomationExceptionOccurrence, *, actor_id: uuid.UUID,
    trace_id: str,
) -> tuple[bool, str | None]:
    """Re-push one person to one identity target.

    Needs both halves of the pair: the occurrence carries the person,
    and ``source_ref_id`` carries the run-item, from which the target
    is resolved. Guessing either would push the wrong person or push
    to the wrong system.

    ``force_in_scope=False`` on purpose. The filter and the target's
    write mode are policy — an admin saying "the credential is fixed"
    is not the same as an admin saying "push this person even though
    the target says not to", and #1736 aligned the single-person path
    with the sweep for exactly this reason.
    """
    from app.domains.identity_sync.models import (  # noqa: PLC0415
        IdentitySyncRunItem,
    )
    from app.domains.identity_sync.service import (  # noqa: PLC0415
        IdentitySyncService,
    )

    if occ.person_id is None:
        return False, "occurrence has no person; nothing to re-push"
    try:
        item_id = uuid.UUID(occ.source_ref_id)
    except (ValueError, AttributeError):
        return False, f"unparseable run-item id {occ.source_ref_id!r}"
    item = await db.get(IdentitySyncRunItem, item_id)
    if item is None:
        return False, "run item no longer exists"
    # ``target_id`` lives directly on the run item (verified against the
    # model, not assumed — an earlier draft had a `run.target_id`
    # fallback that could never fire because there is no such relation).
    target_id = item.target_id
    if target_id is None:
        return False, "cannot resolve the sync target for this item"
    # SAVEPOINT for the same reason as ``_replay_provisioning``.
    async with best_effort(
        db, label="re-push person to identity target", log=logger,
    ) as step:
        await IdentitySyncService(db).sync_person_to_target(
            target_id=target_id,
            person_id=occ.person_id,
            trigger_source="automation_exception_replay",
        )
    if not step.ok:
        return False, step.reason(max_len=200)
    return True, None


_REPLAY_BY_SOURCE = {
    SOURCE_PROVISIONING: _replay_provisioning,
    SOURCE_NOTIFICATION: _replay_outbox,
    SOURCE_IDENTITY_SYNC: _replay_identity_sync,
}

#: The sources whose collectors can emit ``replayable=True``. Kept only
#: so a test can assert this module has a handler for each of them — a
#: collector that starts marking a new source replayable without one
#: here would silently count as "unsupported" forever, which reads on
#: screen as "we tried and there was nothing to do".
SOURCES_WITH_HANDLERS: frozenset[str] = frozenset(_REPLAY_BY_SOURCE)


async def replay_occurrences(
    db: AsyncSession,
    occurrences: list[AutomationExceptionOccurrence],
    *,
    actor_id: uuid.UUID,
    trace_id: str,
    now: datetime | None = None,
) -> ReplayOutcome:
    """Re-drive each occurrence through its own domain's retry path.

    ⚠️ Callers that assemble their own list are responsible for having
    locked and refreshed those rows. The two entry points below do it;
    a hand-built list read with a plain ``select()`` carries whatever
    was already in the session's identity map, and the cooldown check
    below would then be reading a value from before someone else's
    replay.

    Never raises for a single row: one unparseable id or one vendor
    still being down must not stop the other 199. Each failure is
    counted and the first few messages are carried back, because "it
    did not work" without "here is what it said" is the dead end this
    epic removes.
    """
    moment = now or datetime.now(UTC)
    outcome = ReplayOutcome()

    for occ in occurrences[:MAX_REPLAY_PER_RUN]:
        # Re-checked here and not only in the query: a caller that
        # assembled its own list must not be able to bypass the
        # collector's judgement about what carries enough to retry.
        if not occ.replayable:
            outcome.skipped_unsupported += 1
            continue
        handler = _REPLAY_BY_SOURCE.get(occ.source)
        if handler is None:
            outcome.skipped_unsupported += 1
            continue
        # ``aware()`` is not decoration: MySQL hands back a NAIVE
        # datetime, so subtracting it from an aware ``moment`` raises
        # TypeError. In-memory values written by this process are aware
        # and hid the bug — it only appears once the value has made a
        # round trip through the database, which is every run after the
        # first. Same trap PR1 collapsed into ``timeutil`` for the
        # ingest path.
        last_replay = aware(occ.last_replay_at)
        if last_replay is not None and moment - last_replay < REPLAY_COOLDOWN:
            outcome.skipped_cooldown += 1
            continue

        # Stamped BEFORE the attempt, so a handler that dies mid-flight
        # still leaves evidence that we touched this row. An attempt we
        # cannot see is an attempt that gets silently repeated.
        occ.last_replay_at = moment
        occ.replay_attempt_count = (occ.replay_attempt_count or 0) + 1
        outcome.attempted += 1
        outcome.exceptions_touched.add(occ.exception_id)

        ok, error = await handler(
            db, occ, actor_id=actor_id, trace_id=trace_id,
        )
        if ok:
            outcome.succeeded += 1
            occ.last_replay_error = None
        else:
            outcome.failed += 1
            occ.last_replay_error = (error or "")[:500] or None
            if error and error not in outcome.errors:
                outcome.errors.append(error)

    await db.flush()
    return outcome


async def replay_for_integration(
    db: AsyncSession,
    integration_id: uuid.UUID,
    *,
    actor_id: uuid.UUID,
    trace_id: str,
    now: datetime | None = None,
) -> ReplayOutcome:
    """"The credential is fixed" — re-drive everything under it.

    This is the one-click path. The admin has already made the only
    decision available; enumerating the affected rows back to them
    would be asking them to do the work twice.
    """
    rows = (
        await db.execute(
            select(AutomationExceptionOccurrence)
            .join(
                AutomationException,
                AutomationException.id == AutomationExceptionOccurrence.exception_id,
            )
            .where(
                AutomationException.integration_id == integration_id,
                AutomationException.resolved_at.is_(None),
                AutomationExceptionOccurrence.recovered_at.is_(None),
                # The row's own flag is the authority — see the comment
                # on ``_REPLAY_BY_SOURCE``.
                AutomationExceptionOccurrence.replayable.is_(True),
            )
            # Oldest first: the ones that have been broken longest are
            # the ones someone is most likely waiting on.
            .order_by(AutomationExceptionOccurrence.occurred_at.asc())
            .limit(MAX_REPLAY_PER_RUN)
            # Locked AND refreshed. The lock serialises two admins
            # clicking at the same moment; ``populate_existing`` is what
            # makes the lock mean anything, because a plain SELECT
            # returns whatever is already in this session's identity map
            # and would hand back the pre-lock ``last_replay_at``. The
            # cooldown would then pass twice and double the vendor load
            # it exists to prevent.
            .with_for_update()
            .execution_options(populate_existing=True),
        )
    ).scalars().all()

    return await replay_occurrences(
        db, list(rows), actor_id=actor_id, trace_id=trace_id, now=now,
    )


async def replay_for_exception(
    db: AsyncSession,
    exception_id: uuid.UUID,
    *,
    actor_id: uuid.UUID,
    trace_id: str,
    now: datetime | None = None,
) -> ReplayOutcome:
    """Re-drive the open occurrences of one exception."""
    rows = (
        await db.execute(
            select(AutomationExceptionOccurrence)
            .where(
                AutomationExceptionOccurrence.exception_id == exception_id,
                AutomationExceptionOccurrence.recovered_at.is_(None),
                AutomationExceptionOccurrence.replayable.is_(True),
            )
            .order_by(AutomationExceptionOccurrence.occurred_at.asc())
            .limit(MAX_REPLAY_PER_RUN)
            # See ``replay_for_integration`` — same reason.
            .with_for_update()
            .execution_options(populate_existing=True),
        )
    ).scalars().all()

    return await replay_occurrences(
        db, list(rows), actor_id=actor_id, trace_id=trace_id, now=now,
    )


__all__ = [
    "MAX_REPLAY_PER_RUN",
    "REPLAY_COOLDOWN",
    "SOURCES_WITH_HANDLERS",
    "ReplayOutcome",
    "replay_for_exception",
    "replay_for_integration",
    "replay_occurrences",
]