"""``enqueue_then_drain``:phase 1 的 rollback 不能让整个扇出崩掉。

这个 helper 的 docstring 承诺「one bad address must not cost the other 499
recipients their message」。但 phase 1 某一行 enqueue 失败时会 ``await
db.rollback()``,而 rollback **expire 整个 identity map** —— 包括

* 后面尚未处理的 envelope 还握着的 ``recipient`` ORM 实例;
* **已经攒进 ``pending``** 的那些 outbox 实例,以及它们对应的 recipient。

异步代码里碰一个过期属性就是 ``MissingGreenlet``。第一版修复只复活了「后面那些」
收件人,于是「**第 3 行**失败」仍然会让异常从 phase 2 一路抛出这个 helper:
``db.get(NotificationOutbox, outbox.id, …)`` 里的 ``outbox.id`` 先炸,错误分支的
``env.recipient.id`` 再炸。

这两条测试分别钉住「第一行失败」和「中间某行失败」——后者是第一版漏掉的那条。
"""
from __future__ import annotations

import uuid

import pytest
from sqlalchemy import select

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


async def _people(db, n: int) -> list[IdentityPerson]:
    out = []
    for i in range(n):
        pid = uuid.uuid4()
        p = IdentityPerson(
            id=pid,
            work_email=f"fanout-{i}-{pid}@test.com",
            display_name=f"Fanout {i}",
            employment_status="active",
            source_system="manual",
            hris_person_id=str(pid)[:8],
        )
        db.add(p)
        out.append(p)
    await db.flush()
    await db.commit()
    return out


def _envelopes(people: list[IdentityPerson]) -> list[NotificationEnvelope]:
    return [
        NotificationEnvelope(
            recipient=p,
            title="t",
            body_text="b",
            channels=["email"],
        )
        for p in people
    ]


@pytest.mark.parametrize("fail_on", [1, 3])
@pytest.mark.asyncio
async def test_one_bad_enqueue_never_costs_the_others_their_message(
    db_session, monkeypatch, fail_on,
):
    """一行 enqueue 失败,其余人照常送达,且**异常不得逃出** helper。

    ``fail_on=3`` 是关键那条:那时 ``pending`` 里已经有两行,rollback 把它们连同
    各自的 recipient 一起 expire 掉了。
    """
    people = await _people(db_session, 4)
    real_enqueue = NotificationService.enqueue
    seen = {"n": 0}

    async def _flaky(self, env):
        seen["n"] += 1
        if seen["n"] == fail_on:
            # 先真的碰一下 DB,再失败 —— 这样 rollback 有东西可撤,更接近真实
            # 的「唯一约束冲突 / FK 冲突」形状。
            await real_enqueue(self, env)
            raise RuntimeError("simulated enqueue failure")
        return await real_enqueue(self, env)

    delivered_to: list[str] = []

    async def _ok(self, outbox, env, **kw):
        delivered_to.append(str(env.recipient.id))
        outbox.status = "delivered"

    monkeypatch.setattr(NotificationService, "enqueue", _flaky)
    monkeypatch.setattr(NotificationService, "deliver", _ok)

    # 不得抛。抛出来就等于「一个坏地址让其余 499 人收不到信」。
    result = await fanout.enqueue_then_drain(
        db_session, _envelopes(people), label="test.fanout",
    )

    assert result["enqueued"] == 3, result
    assert result["delivered"] == 3, result
    # ``enqueued + failed == len(envelopes)``: a recipient lost in phase 1
    # is a failure, not an absence. This used to be 0 — four intended
    # people, three enqueued, "failed: 0".
    assert result["failed"] == 1, result
    assert len(delivered_to) == 3
    # 失败的那一位不在送达名单里,其余三位都在。
    expected = [str(p.id) for i, p in enumerate(people, start=1) if i != fail_on]
    assert sorted(delivered_to) == sorted(expected)


@pytest.mark.parametrize("fail_on", [1, 2])
@pytest.mark.asyncio
async def test_one_bad_delivery_never_costs_the_others_their_message(
    db_session, monkeypatch, fail_on,
):
    """phase 2 的同一条承诺,之前没有任何测试盖到。

    上面两条钉的是 phase 1,而它们把 ``deliver`` 换成了**永远成功**的替身 ——
    所以 phase 2 自己的错误分支(``failed += 1`` → ``await db.rollback()`` →
    继续下一行)一直没有被执行过。

    这里刻意让 ``deliver`` 在某一行抛异常,并且让**后面那些行**去读
    ``env.recipient.id`` / ``env.recipient.work_email`` —— 真实的 ``_deliver``
    正是这样读收件人的。phase 1 的 rollback 需要 ``_revive_recipients`` 才不会
    级联,phase 2 的 rollback 没有配同样的复活;实测**不会**级联(``db.get(...,
    populate_existing=True)`` 之后这些实例仍然可读),所以这条测试钉的是
    *现有行为是对的*,不是修了什么。

    它有牙:把 phase 2 的 ``except Exception`` 改成重新抛出,这条就红。
    """
    people = await _people(db_session, 4)
    delivered_to: list[str] = []
    seen = {"n": 0}

    async def _flaky_deliver(self, outbox, env, **kw):
        seen["n"] += 1
        if seen["n"] == fail_on:
            raise RuntimeError("simulated delivery failure")
        # 真实的 ``_deliver`` 就是这样读收件人的(work_email / lang / id)。
        # rollback 之后这几下就是 lazy load。
        delivered_to.append(f"{env.recipient.id}:{env.recipient.work_email}")
        outbox.status = "delivered"

    monkeypatch.setattr(NotificationService, "deliver", _flaky_deliver)

    result = await fanout.enqueue_then_drain(
        db_session, _envelopes(people), label="test.fanout.deliver",
    )

    assert result["enqueued"] == 4, result
    assert result["failed"] == 1, result
    assert result["delivered"] == 3, result
    assert len(delivered_to) == 3


# ════════════════════════════════════════════════════════════════════
# ``on_enqueued`` — the hook a caller's ledger stamp rides in
# ════════════════════════════════════════════════════════════════════


@pytest.mark.asyncio
async def test_on_enqueued_fires_once_per_row_with_that_row_s_outbox_id(
    db_session, monkeypatch,
):
    """The contract a caller's per-recipient bookkeeping depends on.

    ``security_awareness`` stamps each campaign target's ``queued_at``
    in this hook, and then uses that stamp to answer "who still needs
    enqueueing" when a killed dispatch is resumed. A hook that silently
    stopped firing would leave every row unstamped and make the resumed
    run re-enqueue the WHOLE audience — the same required reading twice
    for everybody who was already reached.
    """
    people = await _people(db_session, 3)

    async def _ok(self, outbox, env, **kw):
        outbox.status = "delivered"

    monkeypatch.setattr(NotificationService, "deliver", _ok)

    calls: list[tuple[str, uuid.UUID]] = []

    async def _hook(env, outbox_id):
        calls.append((str(env.recipient.id), outbox_id))

    result = await fanout.enqueue_then_drain(
        db_session, _envelopes(people),
        label="test.fanout.hook", on_enqueued=_hook,
    )

    assert result["enqueued"] == 3
    assert [c[0] for c in calls] == [str(p.id) for p in people]
    # The id handed over is the row that was just written, not a fresh
    # one — a caller keying its own ledger on it would otherwise point at
    # nothing.
    landed = {
        str(row.recipient_person_id): row.id
        for row in (await db_session.execute(
            select(NotificationOutbox).where(
                NotificationOutbox.recipient_person_id.in_(
                    [p.id for p in people],
                ),
            )
        )).scalars().all()
    }
    assert {c[0]: c[1] for c in calls} == landed


@pytest.mark.asyncio
async def test_a_hook_that_raises_costs_that_row_and_nobody_else(
    db_session, monkeypatch,
):
    """The hook runs INSIDE the row's transaction, before its commit.

    That is the whole reason it exists rather than being a post-drain
    callback: the caller's stamp and the outbox row have to land together
    or not at all. So a hook failure has to look exactly like an enqueue
    failure — this recipient gets no outbox row and is counted, and the
    other two still get their message.
    """
    people = await _people(db_session, 3)

    async def _ok(self, outbox, env, **kw):
        outbox.status = "delivered"

    monkeypatch.setattr(NotificationService, "deliver", _ok)

    async def _hook(env, outbox_id):
        if env.recipient.id == people[1].id:
            raise RuntimeError("ledger stamp failed")

    result = await fanout.enqueue_then_drain(
        db_session, _envelopes(people),
        label="test.fanout.hook.raises", on_enqueued=_hook,
    )

    assert result["enqueued"] == 2
    assert result["delivered"] == 2
    assert result["failed"] == 1

    rows = (await db_session.execute(
        select(NotificationOutbox.recipient_person_id).where(
            NotificationOutbox.recipient_person_id.in_(
                [p.id for p in people],
            ),
        )
    )).scalars().all()
    assert people[1].id not in rows, (
        "the row whose hook failed must have been rolled back with it"
    )
    assert sorted(map(str, rows)) == sorted(
        [str(people[0].id), str(people[2].id)],
    )