"""Regression: a DB error on the outbox INSERT must not poison the
caller's SQLAlchemy session.

Prod incident 2026-05-26 (``cmdb.billing_daily_sync``): the daily
cloud-bill cron ran the whole multi-scope ingest in ONE transaction
held open across minutes of external billing-API calls, holding locks
on ``notification_outbox`` long enough that a concurrent alert writer
tripped a ``(1205) Lock wait timeout exceeded`` on the alert-card
INSERT inside ``NotificationService.dispatch``. The best-effort
dispatch handler (``_dispatch_scope_threshold_alert``) caught the error
and kept using the session — but the INSERT flush ran OUTSIDE any
savepoint, so the session was left in rolled-back-pending state and the
NEXT statement raised an opaque ``PendingRollbackError`` that masked the
real 1205 and aborted the entire run (discarding every scope's
already-ingested rows).

The fix wraps the outbox INSERT in a SAVEPOINT so a transient flush
failure rolls back only that INSERT and keeps the caller's session
usable. We reproduce a real DB-level flush failure with a foreign-key
violation (``recipient_person_id`` references a person that was never
persisted) and assert the session still works afterwards.
"""
from __future__ import annotations

from uuid import uuid4

import pytest
from sqlalchemy import select

from app.identity.person_models import IdentityPerson
from app.notifications.service import NotificationEnvelope, NotificationService


def _ghost_recipient() -> IdentityPerson:
    """A transient person whose id is NOT in ``identity_persons`` — the
    outbox INSERT's FK on ``recipient_person_id`` therefore fails."""
    return IdentityPerson(
        id=uuid4(),
        source_system="successfactors",
        hris_person_id=f"sf-{uuid4().hex[:8]}",
        work_email="ghost@x.com",
        display_name="Ghost Recipient",
        employment_status="active",
    )


@pytest.mark.asyncio
async def test_dispatch_db_error_does_not_poison_session(db_session):
    svc = NotificationService(db_session)
    env = NotificationEnvelope(
        recipient=_ghost_recipient(),
        title="t",
        body_text="b",
        channels=["feishu"],
    )

    # The outbox INSERT must fail at the DB (FK violation).
    with pytest.raises(Exception):
        await svc.dispatch(env)

    # ...but the session must remain usable. Pre-fix this raised
    # ``PendingRollbackError: This Session's transaction has been rolled
    # back due to a previous exception during flush``.
    result = await db_session.execute(select(1))
    assert result.scalar() == 1

    # And no orphan pending outbox row may linger (the savepoint rollback
    # expunges it) — a clean flush on the same session must succeed.
    await db_session.flush()


# ---------------------------------------------------------------------------
# Root-cause fix 2026-06-20: the SAME poisoning could happen on the
# delivery-STATUS flush inside ``_deliver`` (after the channel sends),
# not just the INSERT. ``_deliver`` did ``await self.db.flush()`` OUTSIDE
# any savepoint to persist the outbox status mutation; if that flush
# raised (e.g. 1205 on the outbox indexes), the caller's shared session
# was left in rolled-back-pending state and every per-item cron loop that
# calls ``dispatch`` then hit ``PendingRollbackError`` on its next
# statement — poisoning a whole family of notify fan-out crons
# (PAT-expiry, borrow-overdue, jira admin-notify, digest, post_leaver,
# hire_day, software-seat self-cancel notify). The fix wraps the
# status-persist flush in a SAVEPOINT and converts a flush failure into a
# delivery-failure return instead of propagating a session-poisoning
# error. These tests cover (a) happy-path still persists, (b) a
# status-flush failure leaves the caller's session usable.
# ---------------------------------------------------------------------------


async def _real_recipient(db_session) -> IdentityPerson:
    person = IdentityPerson(
        id=uuid4(),
        source_system="successfactors",
        hris_person_id=f"sf-{uuid4().hex[:8]}",
        work_email=f"r-{uuid4().hex[:6]}@x.com",
        display_name="Real Recipient",
        employment_status="active",
    )
    db_session.add(person)
    await db_session.flush()
    return person


def _stub_channels_ok(monkeypatch) -> None:
    from app.notifications import service as notif_service

    async def fake_feishu(self, env):  # noqa: ARG001
        return (True, None, None)

    async def fake_email(self, env):  # noqa: ARG001
        return (True, None)

    monkeypatch.setattr(
        notif_service.NotificationService, "_send_feishu", fake_feishu,
    )
    monkeypatch.setattr(
        notif_service.NotificationService, "_send_email", fake_email,
    )


@pytest.mark.asyncio
async def test_dispatch_happy_path_persists_outbox_and_status(
    db_session, monkeypatch,
):
    """No-regression: a normal dispatch still persists the outbox row AND
    its DELIVERED status (the savepoint-wrapped status flush must not
    swallow the happy-path write)."""
    from app.notifications.models import STATUS_DELIVERED, NotificationOutbox

    _stub_channels_ok(monkeypatch)
    recipient = await _real_recipient(db_session)

    svc = NotificationService(db_session)
    env = NotificationEnvelope(
        recipient=recipient,
        title="happy-path",
        body_text="body",
        channels=["feishu", "email"],
    )
    result = await svc.dispatch(env)
    assert "feishu" in result.delivered_channels
    assert "email" in result.delivered_channels

    rows = (await db_session.execute(
        select(NotificationOutbox).where(
            NotificationOutbox.recipient_person_id == recipient.id,
        )
    )).scalars().all()
    assert len(rows) == 1
    assert rows[0].status == STATUS_DELIVERED
    assert sorted(rows[0].delivered_channels) == ["email", "feishu"]


@pytest.mark.asyncio
async def test_dispatch_status_flush_failure_does_not_poison_session(
    db_session, monkeypatch,
):
    """If the delivery-STATUS flush inside ``_deliver`` fails, the
    caller's session must stay usable (no ``PendingRollbackError`` on the
    next statement) and ``dispatch`` must return a result rather than
    propagate a poisoning error.

    We simulate a transient flush failure by raising on the SECOND
    ``flush`` call during dispatch: the first persists the outbox INSERT
    (inside its own savepoint), the second is the status-persist flush we
    are hardening. The error is a plain ``RuntimeError`` (NOT an
    ``OperationalError``) so it does not get re-raised by the
    deadlock-passthrough guard.
    """
    from app.notifications import service as notif_service

    _stub_channels_ok(monkeypatch)
    recipient = await _real_recipient(db_session)

    real_flush = notif_service.AsyncSession.flush
    state = {"calls": 0}

    async def flaky_flush(self, *args, **kwargs):
        state["calls"] += 1
        # Call 1 = outbox INSERT flush; call 2 = status-persist flush.
        if state["calls"] == 2:
            raise RuntimeError("simulated transient status flush failure")
        return await real_flush(self, *args, **kwargs)

    monkeypatch.setattr(
        notif_service.AsyncSession, "flush", flaky_flush,
    )

    svc = NotificationService(db_session)
    env = NotificationEnvelope(
        recipient=recipient,
        title="status-flush-fail",
        body_text="body",
        channels=["feishu", "email"],
    )

    # dispatch must NOT propagate the flush failure — it returns a result.
    result = await svc.dispatch(env)
    assert result is not None

    # Restore real flush, then prove the session is still usable: pre-fix
    # this raised PendingRollbackError.
    monkeypatch.setattr(notif_service.AsyncSession, "flush", real_flush)
    scalar = (await db_session.execute(select(1))).scalar()
    assert scalar == 1
    await db_session.flush()


# ---------------------------------------------------------------------------
# 2026-09-06: the THIRD and last unisolated mutation on the dispatch path.
#
# ``_deliver`` ends with ``await self.audit.log(...)``, and ``AuditService.log``
# is a bare ``add()`` + ``flush()`` that does NOT self-isolate. The outbox
# INSERT (2026-05-26) and the status flush (2026-06-20) had each been wrapped
# after their own incident; the audit row had not, so "``dispatch`` cannot
# poison the caller's transaction" was true of two writes out of three.
#
# That sentence is load-bearing: ~52 call sites across the codebase justify a
# bare ``except`` around a dispatch with it, and the savepoint guard's
# ``_SELF_ISOLATING_CALLS`` exemption encodes it. Found by the AST test that
# checks that exemption's claim (``test_savepoint_guard.py``), which went red
# the first time it ran.
#
# An over-long ``detail`` value or a lock wait on ``audit_logs`` is enough.
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_audit_write_failure_does_not_poison_session(
    db_session, monkeypatch,
):
    from app.audit.service import AuditService
    from app.notifications.models import NotificationOutbox

    _stub_channels_ok(monkeypatch)
    recipient = await _real_recipient(db_session)

    async def _poisoning_log(self, **kwargs):  # noqa: ARG001
        # A real failed FLUSH, not a bare ``raise``: only a failed flush
        # deactivates the transaction, so a raise would pass with the
        # savepoint removed and prove nothing.
        self.db.add(
            IdentityPerson(
                id=uuid4(),
                source_system=recipient.source_system,
                hris_person_id=recipient.hris_person_id,  # UNIQUE violation
                work_email=f"dup-{uuid4().hex[:6]}@x.com",
                display_name="dup",
                employment_status="active",
            ),
        )
        await self.db.flush()

    monkeypatch.setattr(AuditService, "log", _poisoning_log)

    svc = NotificationService(db_session)
    result = await svc.dispatch(
        NotificationEnvelope(
            recipient=recipient,
            title="audit-poison",
            body_text="body",
            channels=["feishu"],
        ),
    )

    # The delivery itself still reports what actually happened.
    assert "feishu" in result.delivered_channels

    # The real assertion: the caller's session survives. Pre-fix this raised
    # ``PendingRollbackError`` on the very next statement.
    assert (await db_session.execute(select(1))).scalar() == 1

    # And the outbox row it wrote is still there — the audit failure rolled
    # back only its own savepoint.
    rows = (
        await db_session.execute(
            select(NotificationOutbox).where(
                NotificationOutbox.recipient_person_id == recipient.id,
            ),
        )
    ).scalars().all()
    assert len(rows) == 1