"""The four batched-send routes, through ASGI against real Postgres.

`test_bulk_invitation_compose.py` covers composition as a unit and
`test_bulk_send_pipeline.py` covers the queue. This covers what neither can: what the
endpoints refuse, and *when* — before anything is queued, or not at all.

The order of the refusals is itself under test. A resubmitted batch must be recognised
before the quiet-hours question is asked, or a repeated request would be refused for a
decision the admin already made; suppression must be applied before the insert, or the
admin's count and the messages that went out disagree.
"""

import uuid
from collections.abc import AsyncIterator
from datetime import UTC, datetime, timedelta
from typing import Any

import pytest
import pytest_asyncio
from httpx import ASGITransport, AsyncClient
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession

from app.main import app
from app.models import Event, Guest, Invitation, Wedding
from app.models.enums import AdminRole, EventType, InvitationStatus, InvitationType, Locale
from app.services import messaging
from app.services.auth import CurrentAdmin, get_current_admin
from app.services.policy import Action, can

from .conftest import forget_admin, make_owner_admin, requires_db

pytestmark = [
    requires_db,
    pytest.mark.usefixtures("app_engine_per_loop", "super_admin"),
]


def _pane(
    invitation_type: InvitationType,
    locale: Locale = Locale.EN,
    **overrides: str,
) -> dict[str, Any]:
    return {
        "locale": str(locale),
        "invitation_type": str(invitation_type),
        "subject": "You're invited",
        "header": "Dear {guest_name}",
        "body": "Please come to the Walima.\n\n{invitation_link}",
        "footer": "Regards\nAyesha & Rahim",
        **overrides,
    }


def _both_panes(**overrides: str) -> list[dict[str, Any]]:
    return [
        _pane(InvitationType.SINGLE, **overrides),
        _pane(InvitationType.FAMILY, **overrides),
    ]


@pytest_asyncio.fixture
async def selection(session: AsyncSession) -> AsyncIterator[dict[str, Any]]:
    """An event with a mixed, mostly-unanswered guest list, plus a second event to steal from.

    The second event exists for one test — that a guest id from elsewhere is refused — and
    that test is the only thing standing between the route and mailing one event's wording
    to another event's list.
    """
    tag = uuid.uuid4().hex[:8]
    # Every event needs an owner (add-admin-access-control D1). Both events below share it,
    # so the "guest id from another event" test still exercises the event boundary rather
    # than accidentally testing the ownership boundary instead.
    owner_id = await make_owner_admin(session, tag=f"routes-{tag}")
    wedding = Wedding(
        bride_name="Ayesha", groom_name="Rahim", slug=f"routes-{tag}", timezone="Asia/Dhaka"
    )
    session.add(wedding)
    await session.flush()
    wedding_id = wedding.id

    def _event(kind: str, slug_suffix: str) -> Event:
        return Event(
            wedding_id=wedding.id,
            owner_admin_id=owner_id,
            type=EventType.WALIMA if kind == "main" else EventType.MEHEDI,
            slug=f"{slug_suffix}-{tag}",
            title_bn="ওয়ালিমা",
            title_en="Walima" if kind == "main" else "Mehedi",
            starts_at=datetime.now(UTC) + timedelta(days=40),
            venue_name="Test Hall",
            venue_address="Dhaka",
            host_name_1="Abdul Karim",
            host_phone="+8801711223344",
            is_published=True,
            invitation_messages={"en": {"single": "Come alone", "family": "Bring everyone"}},
        )

    event = _event("main", "walima")
    other_event = _event("other", "mehedi")
    session.add_all([event, other_event])
    await session.flush()

    specs = [
        ("single_a", InvitationType.SINGLE, Locale.EN, {}),
        ("single_b", InvitationType.SINGLE, Locale.EN, {}),
        ("family_a", InvitationType.FAMILY, Locale.EN, {}),
        ("opted_out", InvitationType.SINGLE, Locale.EN, {"do_not_contact": True}),
        ("bounced", InvitationType.SINGLE, Locale.EN, {"email_invalid": True}),
    ]

    guest_ids: dict[str, uuid.UUID] = {}
    for i, (key, invitation_type, locale, flags) in enumerate(specs):
        guest = Guest(
            wedding_id=wedding.id,
            event_id=event.id,
            full_name=f"Guest {key}",
            email=f"{key}-{tag}@example.test",
            preferred_locale=locale,
            invitation_type=invitation_type,
            **flags,
        )
        session.add(guest)
        await session.flush()
        session.add(
            Invitation(
                guest_id=guest.id,
                event_id=event.id,
                token=f"rtok-{tag}-{i}",
                short_code=f"r{tag[:4]}{i}",
                status=InvitationStatus.PENDING,
                max_guests=4,
            )
        )
        guest_ids[key] = guest.id

    stranger = Guest(
        wedding_id=wedding.id,
        event_id=other_event.id,
        full_name="Guest stranger",
        email=f"stranger-{tag}@example.test",
        preferred_locale=Locale.EN,
        invitation_type=InvitationType.SINGLE,
    )
    session.add(stranger)
    await session.flush()
    session.add(
        Invitation(
            guest_id=stranger.id,
            event_id=other_event.id,
            token=f"rtok-{tag}-x",
            short_code=f"r{tag[:4]}x",
            status=InvitationStatus.PENDING,
            max_guests=4,
        )
    )
    guest_ids["stranger"] = stranger.id

    await session.commit()
    event_id, other_event_id = event.id, other_event.id

    yield {
        "tag": tag,
        "event_id": event_id,
        "other_event_id": other_event_id,
        "guest_ids": guest_ids,
        "sendable": [guest_ids["single_a"], guest_ids["single_b"], guest_ids["family_a"]],
        "all_in_event": [v for k, v in guest_ids.items() if k != "stranger"],
    }

    await session.rollback()
    await session.execute(text("DELETE FROM wedding WHERE id = :id"), {"id": wedding_id})
    # After the cascade: the owner FK is ON DELETE RESTRICT, so the events must go first.
    await forget_admin(session, owner_id)
    await session.commit()


def _client() -> AsyncClient:
    return AsyncClient(transport=ASGITransport(app=app), base_url="http://test")


async def _send(
    selection: dict[str, Any],
    *,
    guest_ids: list[uuid.UUID] | None = None,
    batch_id: uuid.UUID | None = None,
    panes: list[dict[str, Any]] | None = None,
    confirmed_quiet_hours: bool = True,
) -> Any:
    async with _client() as client:
        return await client.post(
            f"/api/admin/events/{selection['event_id']}/invitations/send-batch",
            json={
                "batch_id": str(batch_id or uuid.uuid4()),
                "guest_ids": [str(g) for g in (guest_ids or selection["all_in_event"])],
                "panes": panes if panes is not None else _both_panes(),
                "confirmed_quiet_hours": confirmed_quiet_hours,
            },
        )


# ------------------------------------------------------------------ selecting ids


async def test_guest_ids_returns_every_match_for_the_event(
    selection: dict[str, Any],
) -> None:
    async with _client() as client:
        response = await client.get(f"/api/admin/events/{selection['event_id']}/guest-ids")

    assert response.status_code == 200, response.text
    body = response.json()
    assert body["total"] == len(selection["all_in_event"])
    assert set(body["guest_ids"]) == {str(g) for g in selection["all_in_event"]}


async def test_guest_ids_honours_the_same_filters_as_the_list(
    selection: dict[str, Any],
) -> None:
    """D2: an admin who reads "N matching" and presses select-all must get exactly those N."""
    async with _client() as client:
        listed = await client.get(
            "/api/admin/guests",
            params={"event_id": str(selection["event_id"]), "search": "single_a"},
        )
        ids = await client.get(
            f"/api/admin/events/{selection['event_id']}/guest-ids",
            params={"search": "single_a"},
        )

    assert ids.json()["total"] == listed.json()["total"] == 1
    assert ids.json()["guest_ids"] == [str(selection["guest_ids"]["single_a"])]


async def test_guest_ids_404s_for_an_event_that_does_not_exist() -> None:
    async with _client() as client:
        response = await client.get(f"/api/admin/events/{uuid.uuid4()}/guest-ids")
    assert response.status_code == 404


# ------------------------------------------------------------------ composing


async def test_compose_returns_a_pane_for_each_invitation_type(
    selection: dict[str, Any],
) -> None:
    async with _client() as client:
        response = await client.post(
            f"/api/admin/events/{selection['event_id']}/invitations/compose",
            json={"guest_ids": [str(g) for g in selection["sendable"]]},
        )

    assert response.status_code == 200, response.text
    body = response.json()
    panes = {(p["locale"], p["invitation_type"]): p for p in body["panes"]}
    assert set(panes) == {("en", "single"), ("en", "family")}
    assert "Come alone" in panes[("en", "single")]["body"]
    assert "Bring everyone" in panes[("en", "family")]["body"]
    assert panes[("en", "single")]["recipient_count"] == 2
    assert panes[("en", "family")]["recipient_count"] == 1


async def test_compose_leaves_the_guest_parts_as_placeholders(
    selection: dict[str, Any],
) -> None:
    async with _client() as client:
        response = await client.post(
            f"/api/admin/events/{selection['event_id']}/invitations/compose",
            json={"guest_ids": [str(g) for g in selection["sendable"]]},
        )

    body = response.json()
    assert sorted(body["placeholders"]) == ["guest_name", "invitation_link"]
    for pane in body["panes"]:
        assert pane["header"] == "Dear {guest_name}"
        assert "{invitation_link}" in pane["body"]


async def test_compose_breaks_out_who_will_not_be_emailed(
    selection: dict[str, Any],
) -> None:
    """D10: the admin needs this count before pressing Send, not after."""
    async with _client() as client:
        response = await client.post(
            f"/api/admin/events/{selection['event_id']}/invitations/compose",
            json={"guest_ids": [str(g) for g in selection["all_in_event"]]},
        )

    body = response.json()
    assert body["selected"] == 5
    assert body["deliverable"] == 3
    by_code = {e["code"]: e for e in body["exclusions"]}
    assert by_code["opted_out"]["count"] == 1
    assert by_code["email_invalid"]["count"] == 1
    assert by_code["opted_out"]["reason"] == messaging.BLOCK_OPTED_OUT


async def test_compose_still_returns_both_panes_when_only_one_type_is_selected(
    selection: dict[str, Any],
) -> None:
    """The composer always tells the same two-sided story; the unused pane is simply inert."""
    async with _client() as client:
        response = await client.post(
            f"/api/admin/events/{selection['event_id']}/invitations/compose",
            json={"guest_ids": [str(selection["guest_ids"]["single_a"])]},
        )

    panes = {p["invitation_type"]: p for p in response.json()["panes"]}
    assert set(panes) == {"single", "family"}
    assert panes["single"]["recipient_count"] == 1
    assert panes["family"]["recipient_count"] == 0


async def test_compose_refuses_a_guest_from_another_event(
    selection: dict[str, Any],
) -> None:
    """D1: the route names the event, and the body must not be able to contradict it."""
    async with _client() as client:
        response = await client.post(
            f"/api/admin/events/{selection['event_id']}/invitations/compose",
            json={
                "guest_ids": [
                    str(selection["guest_ids"]["single_a"]),
                    str(selection["guest_ids"]["stranger"]),
                ]
            },
        )

    assert response.status_code == 422
    assert response.json()["detail"]["code"] == "foreign_guest"


async def test_compose_404s_for_an_event_that_does_not_exist() -> None:
    async with _client() as client:
        response = await client.post(
            f"/api/admin/events/{uuid.uuid4()}/invitations/compose",
            json={"guest_ids": [str(uuid.uuid4())]},
        )
    assert response.status_code == 404


# ------------------------------------------------------------------ sending


async def test_a_batch_queues_one_message_per_sendable_guest(
    session: AsyncSession, selection: dict[str, Any]
) -> None:
    batch_id = uuid.uuid4()
    response = await _send(selection, batch_id=batch_id)

    assert response.status_code == 200, response.text
    body = response.json()
    assert body["queued"] == 3
    assert body["already_recorded"] is False
    assert body["dry_run"] is True
    assert {e["code"] for e in body["exclusions"]} == {"opted_out", "email_invalid"}

    await session.rollback()
    count = await session.scalar(
        text("SELECT count(*) FROM message_job WHERE batch_id = :id").bindparams(id=batch_id)
    )
    assert count == 3


async def test_resubmitting_the_same_batch_records_nothing_new(
    session: AsyncSession, selection: dict[str, Any]
) -> None:
    """D5: a double-clicked Send, a retried request and a refreshed browser are all this."""
    batch_id = uuid.uuid4()
    first = await _send(selection, batch_id=batch_id)
    second = await _send(selection, batch_id=batch_id)

    assert first.status_code == second.status_code == 200, second.text
    assert second.json()["already_recorded"] is True

    await session.rollback()
    count = await session.scalar(
        text("SELECT count(*) FROM message_job WHERE batch_id = :id").bindparams(id=batch_id)
    )
    assert count == 3


async def test_a_second_batch_to_the_same_guests_sends_again(
    selection: dict[str, Any],
) -> None:
    """D5, the other half: a host who means to resend is not silently ignored."""
    first = await _send(selection, batch_id=uuid.uuid4())
    second = await _send(selection, batch_id=uuid.uuid4())

    assert first.json()["queued"] == second.json()["queued"] == 3
    assert first.json()["batch_id"] != second.json()["batch_id"]


async def test_a_batch_refuses_a_guest_from_another_event(
    selection: dict[str, Any],
) -> None:
    response = await _send(
        selection,
        guest_ids=[selection["guest_ids"]["single_a"], selection["guest_ids"]["stranger"]],
    )
    assert response.status_code == 422
    assert response.json()["detail"]["code"] == "foreign_guest"


async def test_suppressed_guests_get_no_job_however_the_request_is_framed(
    session: AsyncSession, selection: dict[str, Any]
) -> None:
    """D10: no flag in the body may talk its way past a consent record."""
    batch_id = uuid.uuid4()
    await _send(selection, batch_id=batch_id)

    await session.rollback()
    opted_out = await session.scalar(
        text(
            "SELECT count(*) FROM message_job j JOIN invitation i ON i.id = j.invitation_id "
            "WHERE j.batch_id = :batch AND i.guest_id = :guest"
        ).bindparams(batch=batch_id, guest=selection["guest_ids"]["opted_out"])
    )
    assert opted_out == 0


async def test_a_selection_of_only_suppressed_guests_is_refused(
    selection: dict[str, Any],
) -> None:
    response = await _send(
        selection,
        guest_ids=[selection["guest_ids"]["opted_out"], selection["guest_ids"]["bounced"]],
    )
    assert response.status_code == 409
    assert response.json()["detail"]["code"] == "nobody_to_send_to"


@pytest.mark.parametrize(
    ("overrides", "missing"),
    [
        ({"subject": "   "}, "a subject"),
        ({"body": "   "}, "a body"),
    ],
)
async def test_an_empty_message_is_rejected(
    selection: dict[str, Any], overrides: dict[str, str], missing: str
) -> None:
    response = await _send(selection, panes=_both_panes(**overrides))
    assert response.status_code == 422
    assert response.json()["detail"]["code"] == "empty_message"
    assert missing in response.json()["detail"]["message"]


async def test_a_body_without_the_link_placeholder_is_rejected(
    selection: dict[str, Any],
) -> None:
    """D4: an invitation with no link is not an invitation."""
    response = await _send(selection, panes=_both_panes(body="Just turn up on the day."))
    assert response.status_code == 422
    assert response.json()["detail"]["code"] == "bad_placeholder"


async def test_an_unknown_placeholder_is_rejected_by_name(
    selection: dict[str, Any],
) -> None:
    response = await _send(selection, panes=_both_panes(header="Dear {guset_name}"))
    assert response.status_code == 422
    detail = response.json()["detail"]
    assert detail["code"] == "bad_placeholder"
    assert "{guset_name}" in detail["message"]


async def test_a_typo_in_a_pane_nobody_will_receive_does_not_block_the_send(
    selection: dict[str, Any],
) -> None:
    """Refusing a send because the unused family pane has a typo would be obstructive: its
    text reaches nobody."""
    response = await _send(
        selection,
        guest_ids=[selection["guest_ids"]["single_a"]],
        panes=[
            _pane(InvitationType.SINGLE),
            _pane(InvitationType.FAMILY, body="broken {nonsense}"),
        ],
    )
    assert response.status_code == 200, response.text
    assert response.json()["queued"] == 1


async def test_nothing_is_recorded_for_a_rejected_batch(
    session: AsyncSession, selection: dict[str, Any]
) -> None:
    """A refusal must leave nothing behind for the worker to pick up later."""
    batch_id = uuid.uuid4()
    await _send(selection, batch_id=batch_id, panes=_both_panes(body="no link here"))

    await session.rollback()
    count = await session.scalar(
        text("SELECT count(*) FROM message_job WHERE batch_id = :id").bindparams(id=batch_id)
    )
    assert count == 0


async def test_quiet_hours_ask_once_then_send_on_confirmation(
    selection: dict[str, Any], monkeypatch: pytest.MonkeyPatch
) -> None:
    """D9: the server owns the clock, and asks about the batch rather than each recipient.

    Quiet hours are forced on rather than waited for; `test_reminders` covers the arithmetic.
    """
    from app.routers import admin_guests

    monkeypatch.setattr(admin_guests, "_quiet_hours_now", lambda: (True, "23:30"))

    refused = await _send(selection, confirmed_quiet_hours=False)
    assert refused.status_code == 409
    detail = refused.json()["detail"]
    assert detail["code"] == "quiet_hours"
    assert "23:30" in detail["message"]

    confirmed = await _send(selection, confirmed_quiet_hours=True)
    assert confirmed.status_code == 200, confirmed.text
    assert confirmed.json()["queued"] == 3


async def test_the_quiet_hours_question_is_not_asked_again_for_a_recorded_batch(
    selection: dict[str, Any], monkeypatch: pytest.MonkeyPatch
) -> None:
    """The duplicate check runs first on purpose: a resubmitted request must not be refused
    for a decision the admin already made."""
    from app.routers import admin_guests

    batch_id = uuid.uuid4()
    first = await _send(selection, batch_id=batch_id, confirmed_quiet_hours=True)
    assert first.status_code == 200, first.text

    monkeypatch.setattr(admin_guests, "_quiet_hours_now", lambda: (True, "23:30"))
    again = await _send(selection, batch_id=batch_id, confirmed_quiet_hours=False)

    assert again.status_code == 200, again.text
    assert again.json()["already_recorded"] is True


# ------------------------------------------------------------------ progress


async def test_progress_reports_a_recorded_batch(
    selection: dict[str, Any],
) -> None:
    batch_id = uuid.uuid4()
    await _send(selection, batch_id=batch_id)

    async with _client() as client:
        response = await client.get(f"/api/admin/send-batches/{batch_id}")

    assert response.status_code == 200, response.text
    body = response.json()
    assert body["total"] == 3
    assert body["waiting"] + body["sent"] == 3
    assert body["dry_run"] is True


async def test_progress_for_an_unknown_batch_is_empty_rather_than_an_error() -> None:
    """A batch is a set of jobs, not a row, so "no jobs" is a truthful answer."""
    async with _client() as client:
        response = await client.get(f"/api/admin/send-batches/{uuid.uuid4()}")

    assert response.status_code == 200
    assert response.json()["total"] == 0
    assert response.json()["finished"] is True


# ------------------------------------------------------------------ roles


@pytest.fixture
def as_role() -> Any:
    """Swap the acting admin's role for one request, the way the real guard would see it."""

    def use(role: AdminRole) -> None:
        app.dependency_overrides[get_current_admin] = lambda: CurrentAdmin(
            id=uuid.uuid4(), email=f"{role}@example.com", name="Test", role=role
        )

    yield use
    app.dependency_overrides.pop(get_current_admin, None)


def test_the_matrix_agrees_about_who_may_send() -> None:
    """Guards the premise of the tests below.

    Under two roles, sending is no longer withheld from anyone who can sign in — a Host may
    send, and what stops them mailing another customer's list is scope, not capability. So
    the refusal below is a 404 from the scope check rather than a 403 from the matrix.
    """
    assert can(AdminRole.HOST, Action.SEND_MESSAGES)
    assert can(AdminRole.SUPER_ADMIN, Action.SEND_MESSAGES)


@pytest.mark.parametrize(
    ("method", "path", "body"),
    [
        ("GET", "/api/admin/events/{event_id}/guest-ids", None),
        ("POST", "/api/admin/events/{event_id}/invitations/compose", {"guest_ids": ["{guest_id}"]}),
        ("POST", "/api/admin/events/{event_id}/invitations/send-batch", None),
        ("GET", "/api/admin/send-batches/{batch_id}", None),
    ],
    ids=["guest-ids", "compose", "send-batch", "progress"],
)
async def test_a_host_who_owns_nothing_is_refused_by_the_api(
    selection: dict[str, Any],
    as_role: Any,
    method: str,
    path: str,
    body: dict[str, Any] | None,
) -> None:
    """The frontend hides these controls; that is not the rule. This is.

    Replaces the Viewer case: there is no read-only role now, and the boundary that matters
    for a batched send is ownership. This Host owns no events at all, so every one of these
    must answer exactly as it would for an event id that does not exist (design D3).
    """
    as_role(AdminRole.HOST)
    guest_id = str(selection["sendable"][0])
    url = "/api" + path.format(event_id=selection["event_id"], batch_id=uuid.uuid4()).removeprefix(
        "/api"
    )
    payload = None
    if method == "POST":
        payload = (
            {"guest_ids": [guest_id]}
            if body
            else {
                "batch_id": str(uuid.uuid4()),
                "guest_ids": [guest_id],
                "panes": _both_panes(),
                "confirmed_quiet_hours": True,
            }
        )

    async with _client() as client:
        response = await client.request(method, url, json=payload)

    if "send-batches" in url:
        # Progress answers an empty, finished batch rather than 404 — a batch is a set of
        # jobs, not a row, so "none you can see" and "none at all" are the same sentence.
        assert response.status_code == 200
        assert response.json()["total"] == 0
    else:
        assert response.status_code == 404, (
            f"{method} {url} answered {response.status_code} — a Host reached another "
            "owner's event, or learned it exists by being told 403"
        )


@pytest_asyncio.fixture
async def owning_host(session: AsyncSession, selection: dict[str, Any]) -> AsyncIterator[uuid.UUID]:
    """A real Host row that owns the selection's event, not a fabricated id.

    `message_job.sent_by_admin_id` is a foreign key, so an invented admin fails the insert
    and the test then reports a broken fixture as a broken endpoint. The refused cases above
    can fabricate freely — the scope check turns them away before anything is written.

    Owning the event is the other half: a Host holds `send_messages`, so what decides whether
    this send succeeds is whose event it is.
    """
    admin_id = uuid.uuid4()
    email = f"owning-host-{admin_id.hex[:8]}@example.com"
    original_owner = await session.scalar(
        text("SELECT owner_admin_id FROM event WHERE id = :event").bindparams(
            event=selection["event_id"]
        )
    )
    await session.execute(
        text(
            "INSERT INTO admin_user (id, email, name, role, status, auth_method, "
            "                        must_change_password, failed_login_count) "
            "VALUES (:id, :email, 'Owning Host', 'host', 'active', 'google', false, 0)"
        ),
        {"id": admin_id, "email": email},
    )
    await session.execute(
        text("UPDATE event SET owner_admin_id = :admin WHERE id = :event"),
        {"admin": admin_id, "event": selection["event_id"]},
    )
    await session.commit()

    app.dependency_overrides[get_current_admin] = lambda: CurrentAdmin(
        id=admin_id, email=email, name="Owning Host", role=AdminRole.HOST
    )
    yield admin_id
    app.dependency_overrides.pop(get_current_admin, None)

    # This fixture tears down before `selection` does, so the event still exists and still
    # points here. `owner_admin_id` is ON DELETE RESTRICT, so hand it back to the original
    # owner or the delete below is refused — the constraint doing exactly its job.
    await session.rollback()
    await session.execute(
        text("UPDATE event SET owner_admin_id = :owner WHERE owner_admin_id = :admin"),
        {"owner": original_owner, "admin": admin_id},
    )
    await forget_admin(session, admin_id)
    await session.commit()


async def test_a_host_may_send_for_an_event_they_own(
    selection: dict[str, Any], owning_host: uuid.UUID
) -> None:
    """Sending invitations is the host's job as much as the super admin's."""
    response = await _send(selection, guest_ids=selection["sendable"])
    assert response.status_code == 200, response.text
    assert response.json()["queued"] == 3


async def test_the_batch_records_who_sent_it(
    session: AsyncSession, selection: dict[str, Any], owning_host: uuid.UUID
) -> None:
    """One audit row for the whole send, against the admin who pressed the button (D5)."""
    batch_id = uuid.uuid4()
    await _send(selection, batch_id=batch_id, guest_ids=selection["sendable"])

    await session.rollback()
    rows = (
        await session.execute(
            text(
                "SELECT after FROM audit_log WHERE admin_user_id = :id AND action = 'message.send'"
            ),
            {"id": owning_host},
        )
    ).all()
    assert len(rows) == 1, "a batch must not write one audit row per recipient"
    assert rows[0][0]["batch_id"] == str(batch_id)
    assert rows[0][0]["queued"] == 3