"""Campaign lifecycle, dispatch and acknowledgement.

The assertions worth keeping are the ones about *honesty of the record*:
the send lock actually locks, a resend reaches only the people who owe
something, an acknowledgement cannot be moved once made, and the funnel
reports delivery from the outbox rather than from a mirrored column.
"""
from __future__ import annotations

import asyncio
import uuid
from datetime import UTC, datetime, timedelta
from unittest.mock import AsyncMock, patch

import pytest
from sqlalchemy import select, update

from app.domains.security_awareness.models import (
    SecawCampaign,
    SecawCampaignTarget,
)
from app.notifications.models import STATUS_DELIVERED, NotificationOutbox
from tests.test_security_awareness.conftest import (
    ADMIN_BASE,
    SELF_BASE,
    article_payload,
    fanout_double,
    headers_for,
    make_person,
    resend,
    send_now,
)

pytestmark = pytest.mark.asyncio

CAMPAIGNS = f"{ADMIN_BASE}/campaigns"
MY = f"{SELF_BASE}/my/campaigns"


async def _publish_material(client, headers) -> str:
    r = await client.post(
        f"{ADMIN_BASE}/materials", json=article_payload(), headers=headers,
    )
    assert r.status_code == 201, r.text
    return r.json()["data"]["id"]


async def _make_campaign(client, headers, material_id, **over) -> str:
    body = {
        "material_id": material_id,
        "name": "2026 Q3 钓鱼宣贯",
        "audience_config": {},
        "channels": ["feishu", "email"],
        "mandatory": True,
        "completion_mode": "ack",
        **over,
    }
    r = await client.post(CAMPAIGNS, json=body, headers=headers)
    assert r.status_code == 201, r.text
    return r.json()["data"]["id"]


# Every dispatch test patches the fan-out: the assertions are about the
# campaign's own bookkeeping, and a real Feishu/SMTP attempt would make
# them depend on transport config. The outbox rows the funnel reads are
# written by the tests that care about them.
#
# The double HONOURS the primitive's ``on_enqueued`` contract (see
# ``conftest.fanout_double``) — the per-recipient ``queued_at`` stamp
# happens in that hook, so a mock that only returned a dict would leave
# every ledger row unstamped and let a regression that dropped the hook
# pass.
def _fanout_patch(*, failed_for: set[str] | None = None):
    # No ``enqueued=`` / ``delivered=`` knobs on purpose: the double
    # derives its counts from the envelopes it is actually handed. The old
    # signature let a test feed in a number and then assert that same
    # number back out of the summary — one such assertion held with a
    # single person in the database and would have held at any audience
    # size, zero included.
    return patch(
        "app.notifications.fanout.enqueue_then_drain",
        # ``AsyncMock(side_effect=...)`` rather than the bare fake: the
        # tests that assert "the fan-out ran exactly once" need
        # ``await_count``, and the ones that assert on the ledger need the
        # hook. side_effect gives both.
        new=AsyncMock(side_effect=fanout_double(failed_for=failed_for)),
    )


# ── Creation guards ─────────────────────────────────────────────────


async def test_a_draft_material_cannot_be_campaigned(
    client, security_officer,
):
    """Recipients would follow the card to a 404."""
    officer, oh = security_officer
    draft = (await client.post(
        f"{ADMIN_BASE}/materials", json=article_payload(publish=False),
        headers=oh,
    )).json()["data"]["id"]

    r = await client.post(
        CAMPAIGNS,
        json={
            "material_id": draft, "name": "x", "audience_config": {},
            "channels": ["feishu"], "mandatory": False,
            "completion_mode": "ack",
        },
        headers=oh,
    )
    assert r.status_code == 400
    assert r.json()["errors"][0]["code"] == "SECAW_MATERIAL_NOT_PUBLISHED"


async def test_a_malformed_audience_is_refused_at_author_time(
    client, security_officer,
):
    """A filter discovered to be broken by a cron at 09:00 is a silent
    non-send, and the admin who wrote it is not watching."""
    officer, oh = security_officer
    material = await _publish_material(client, oh)
    r = await client.post(
        CAMPAIGNS,
        json={
            "material_id": material, "name": "x",
            "audience_config": {"departments": "not-a-list"},
            "channels": ["feishu"], "mandatory": False,
            "completion_mode": "ack",
        },
        headers=oh,
    )
    assert r.status_code == 422
    assert r.json()["errors"][0]["code"] == "SECAW_AUDIENCE_INVALID"


async def test_exclude_person_ids_narrows_the_audience(
    client, db_session, security_officer, reader,
):
    """The shared filter engine has no exclusion dimension — and it is
    also the pull-surface authorization boundary, so this domain applies
    the exclusion on top rather than widening it."""
    officer, oh = security_officer
    person, _rh = reader

    everyone = (await client.post(
        f"{CAMPAIGNS}/preview-audience",
        json={"audience_config": {}}, headers=oh,
    )).json()["data"]
    assert everyone["total"] >= 2

    minus_one = (await client.post(
        f"{CAMPAIGNS}/preview-audience",
        json={"audience_config": {"exclude_person_ids": [str(person.id)]}},
        headers=oh,
    )).json()["data"]
    assert minus_one["total"] == everyone["total"] - 1
    assert str(person.id) not in [s["person_id"] for s in minus_one["sample"]]


async def test_preview_reports_why_an_audience_is_empty(
    client, security_officer,
):
    """An audience of zero has a reason, and the admin is entitled to it
    before they schedule rather than after nothing arrives."""
    officer, oh = security_officer
    r = await client.post(
        f"{CAMPAIGNS}/preview-audience",
        json={"audience_config": {"person_ids": [str(uuid.uuid4())]}},
        headers=oh,
    )
    data = r.json()["data"]
    assert data["total"] == 0
    assert "empty_audience" in data["notes"]


async def test_preview_sample_keeps_name_and_email_separate(
    client, security_officer,
):
    officer, oh = security_officer
    data = (await client.post(
        f"{CAMPAIGNS}/preview-audience", json={"audience_config": {}},
        headers=oh,
    )).json()["data"]
    assert data["sample"], "expected at least the officer themselves"
    row = data["sample"][0]
    assert "name" in row and "email" in row
    assert "·" not in (row["name"] or "")


# ── Dispatch ────────────────────────────────────────────────────────


async def test_send_now_creates_one_target_per_person_and_snapshots(
    client, db_session, security_officer, reader,
):
    officer, oh = security_officer
    person, _rh = reader
    material = await _publish_material(client, oh)
    campaign = await _make_campaign(client, oh, material)

    with _fanout_patch():
        r = await send_now(client, db_session, campaign, oh)
    assert r.json()["data"]["status"] == "dispatching"

    targets = (await db_session.execute(
        select(SecawCampaignTarget).where(
            SecawCampaignTarget.campaign_id == uuid.UUID(campaign),
        )
    )).scalars().all()
    by_person = {t.person_id: t for t in targets}
    assert person.id in by_person
    mine = by_person[person.id]
    # Snapshots, not live lookups: the person's department will change.
    assert mine.person_snapshot["work_email"] == "reader@test.com"
    assert mine.material_version == 1
    assert mine.queued_at is not None
    assert mine.acknowledged_at is None

    row = await db_session.get(SecawCampaign, uuid.UUID(campaign))
    await db_session.refresh(row)
    assert row.status == "sent"
    assert row.material_version == 1
    assert row.sent_at is not None


async def test_a_second_dispatch_tick_on_the_same_row_does_nothing(
    client, db_session, security_officer,
):
    """The status flip IS the lock — a conditional UPDATE, so the second
    tick updates zero rows and returns instead of sending twice."""
    officer, oh = security_officer
    material = await _publish_material(client, oh)
    campaign = await _make_campaign(client, oh, material)
    await client.post(
        f"{CAMPAIGNS}/{campaign}/schedule", json={"scheduled_at": None},
        headers=oh,
    )

    from app.domains.security_awareness.campaign_service import CampaignService

    with _fanout_patch() as spy:
        first = await CampaignService(db_session).dispatch(uuid.UUID(campaign))
        second = await CampaignService(db_session).dispatch(uuid.UUID(campaign))

    assert "skipped" not in first
    assert second["skipped"] == "not_scheduled"
    assert spy.await_count == 1


async def _armed(client, db_session, oh) -> str:
    """A published material + a campaign sitting in ``scheduled``."""
    material = await _publish_material(client, oh)
    campaign = await _make_campaign(client, oh, material)
    r = await client.post(
        f"{CAMPAIGNS}/{campaign}/schedule", json={"scheduled_at": None},
        headers=oh,
    )
    assert r.status_code == 200, r.text
    return campaign


async def test_a_dispatch_that_dies_before_the_ledger_is_durable_is_rearmed(
    client, db_session, security_officer,
):
    """A green cron run and a silent non-send look identical from the
    console otherwise. And a row stuck in ``sending`` forever would never
    be retried — nothing in the codebase moves a campaign out of that
    status except this compensation and the stale sweep."""
    officer, oh = security_officer
    campaign = await _armed(client, db_session, oh)

    from app.domains.security_awareness.campaign_service import CampaignService

    # Blow up in phase 1, BEFORE any outbox row exists: the audience
    # resolver is the first thing ``_dispatch_inner`` touches.
    with patch(
        "app.domains.security_awareness.audience.AudienceResolver.resolve",
        new=AsyncMock(side_effect=RuntimeError("hris is down")),
    ), pytest.raises(RuntimeError):
        await CampaignService(db_session).dispatch(uuid.UUID(campaign))

    row = await db_session.get(SecawCampaign, uuid.UUID(campaign))
    await db_session.refresh(row)
    assert row.status == "scheduled", "must be retried by the next tick"
    assert row.sent_at is None
    assert "hris is down" in (row.last_dispatch_error or "")


async def test_a_dispatch_cancelled_mid_flight_is_rearmed_too(
    client, db_session, security_officer,
):
    """``CancelledError`` is a ``BaseException``, and it is the failure
    that actually happened: a deploy replaced the container mid-send and
    the compensation — an ``except Exception`` — never ran, so the
    campaign sat in ``sending`` with an empty ``last_dispatch_error``
    while no tick would ever look at it again (2026-09-07)."""
    officer, oh = security_officer
    campaign = await _armed(client, db_session, oh)

    from app.domains.security_awareness.campaign_service import CampaignService

    with patch(
        "app.domains.security_awareness.audience.AudienceResolver.resolve",
        new=AsyncMock(side_effect=asyncio.CancelledError()),
    ), pytest.raises(asyncio.CancelledError):
        await CampaignService(db_session).dispatch(uuid.UUID(campaign))

    row = await db_session.get(SecawCampaign, uuid.UUID(campaign))
    await db_session.refresh(row)
    assert row.status == "scheduled"
    assert row.last_dispatch_error, "cancellation must leave a reason behind"


async def test_a_failure_during_the_fanout_does_NOT_un_send_the_campaign(
    client, db_session, security_officer,
):
    """The other half of the same rule, and the dangerous half.

    Once the ledger is durable the campaign is ``sent`` — people are
    holding cards and their outbox rows belong to the retry worker.
    Re-arming it there would have the next tick walk the whole audience
    again and hand every one of them a second copy of the same required
    reading. The failure is recorded; the status stays put.
    """
    officer, oh = security_officer
    campaign = await _armed(client, db_session, oh)

    from app.domains.security_awareness.campaign_service import CampaignService

    with patch(
        "app.notifications.fanout.enqueue_then_drain",
        new=AsyncMock(side_effect=RuntimeError("feishu is down")),
    ), pytest.raises(RuntimeError):
        await CampaignService(db_session).dispatch(uuid.UUID(campaign))

    row = await db_session.get(SecawCampaign, uuid.UUID(campaign))
    await db_session.refresh(row)
    assert row.status == "sent", "a drain failure must not re-arm the send"
    assert row.sent_at is not None
    assert "feishu is down" in (row.last_dispatch_error or "")

    # And the cron leaves it alone rather than sending it again.
    summary = await CampaignService(db_session).dispatch_due()
    assert summary["considered"] == 0
    assert summary["stale_recovered"] == 0


async def test_the_cron_sends_only_campaigns_whose_moment_has_come(
    client, db_session, security_officer,
):
    officer, oh = security_officer
    material = await _publish_material(client, oh)
    due = await _make_campaign(client, oh, material, name="due")
    later = await _make_campaign(client, oh, material, name="later")

    await client.post(
        f"{CAMPAIGNS}/{due}/schedule",
        json={
            "scheduled_at": (
                datetime.now(UTC) - timedelta(minutes=5)
            ).isoformat(),
        },
        headers=oh,
    )
    await client.post(
        f"{CAMPAIGNS}/{later}/schedule",
        json={
            "scheduled_at": (datetime.now(UTC) + timedelta(days=1)).isoformat(),
        },
        headers=oh,
    )

    from app.domains.security_awareness.campaign_service import CampaignService

    with _fanout_patch():
        summary = await CampaignService(db_session).dispatch_due()

    assert summary["considered"] == 1
    assert summary["dispatched"] == 1
    assert (await db_session.get(
        SecawCampaign, uuid.UUID(later),
    )).status == "scheduled"


async def test_the_audience_is_frozen_at_send_time(
    client, db_session, security_officer,
):
    """Somebody who joins after the send is not retroactively part of the
    campaign — a campaign that re-resolved on read would silently rewrite
    who it was sent to."""
    officer, oh = security_officer
    material = await _publish_material(client, oh)
    campaign = await _make_campaign(client, oh, material)

    with _fanout_patch():
        await send_now(client, db_session, campaign, oh)

    before = (await client.get(
        f"{CAMPAIGNS}/{campaign}", headers=oh,
    )).json()["data"]["funnel"]["targeted"]

    await make_person(db_session, "newjoiner@test.com", role="employee")
    await db_session.commit()

    after = (await client.get(
        f"{CAMPAIGNS}/{campaign}", headers=oh,
    )).json()["data"]["funnel"]["targeted"]
    assert after == before


# ── Acknowledgement ─────────────────────────────────────────────────


async def test_acknowledge_is_idempotent_and_does_not_move_the_timestamp(
    client, db_session, security_officer, reader,
):
    """The timestamp is the evidence; a second click must not rewrite
    when the person actually confirmed."""
    officer, oh = security_officer
    _person, rh = reader
    material = await _publish_material(client, oh)
    campaign = await _make_campaign(client, oh, material)
    with _fanout_patch():
        await send_now(client, db_session, campaign, oh)

    first = await client.post(f"{MY}/{campaign}/acknowledge", headers=rh)
    assert first.status_code == 200
    assert first.json()["data"]["already"] is False
    stamp = first.json()["data"]["acknowledged_at"]

    second = await client.post(f"{MY}/{campaign}/acknowledge", headers=rh)
    assert second.json()["data"]["already"] is True
    assert second.json()["data"]["acknowledged_at"] == stamp


async def test_a_stranger_cannot_acknowledge_a_campaign_they_were_not_sent(
    client, db_session, security_officer,
):
    """404 rather than 403: from their side the campaign does not exist,
    and a 403 would confirm that it does."""
    officer, oh = security_officer
    material = await _publish_material(client, oh)
    campaign = await _make_campaign(
        client, oh, material,
        audience_config={"person_ids": [str(officer.id)]},
    )
    with _fanout_patch():
        await send_now(client, db_session, campaign, oh)

    outsider = await make_person(db_session, "outsider@test.com", role="employee")
    await db_session.commit()
    r = await client.post(
        f"{MY}/{campaign}/acknowledge",
        headers=headers_for(outsider, ["employee"]),
    )
    assert r.status_code == 404


async def test_my_pending_lists_mandatory_only_by_default(
    client, db_session, security_officer, reader,
):
    """A voluntary notice is not work; listing it as a todo would make
    the two strengths indistinguishable."""
    officer, oh = security_officer
    _person, rh = reader
    material = await _publish_material(client, oh)
    required = await _make_campaign(client, oh, material, name="required")
    optional = await _make_campaign(
        client, oh, material, name="optional", mandatory=False,
    )
    with _fanout_patch():
        await send_now(client, db_session, required, oh)
        await send_now(client, db_session, optional, oh)

    default = (await client.get(MY, headers=rh)).json()["data"]
    assert [c["campaign_id"] for c in default] == [required]

    widened = (await client.get(
        f"{MY}?include_voluntary=true", headers=rh,
    )).json()["data"]
    assert {c["campaign_id"] for c in widened} == {required, optional}


async def test_opening_the_material_stamps_opened_at_once(
    client, db_session, security_officer, reader,
):
    """First open only: re-stamping on every refresh would turn the
    column into "when did they last look", which is a different claim."""
    officer, oh = security_officer
    person, rh = reader
    material = await _publish_material(client, oh)
    campaign = await _make_campaign(client, oh, material)
    with _fanout_patch():
        await send_now(client, db_session, campaign, oh)

    await client.get(f"{SELF_BASE}/materials/{material}", headers=rh)
    target = (await db_session.execute(
        select(SecawCampaignTarget).where(
            SecawCampaignTarget.campaign_id == uuid.UUID(campaign),
            SecawCampaignTarget.person_id == person.id,
        )
    )).scalar_one()
    first_open = target.opened_at
    assert first_open is not None

    await client.get(f"{SELF_BASE}/materials/{material}", headers=rh)
    await db_session.refresh(target)
    assert target.opened_at == first_open


async def test_the_material_page_carries_the_pending_acknowledgement(
    client, db_session, security_officer, reader,
):
    """The ack control lives ON the material — a campaign-shaped
    intermediate page would be a click between the push and the
    content."""
    officer, oh = security_officer
    _person, rh = reader
    material = await _publish_material(client, oh)
    campaign = await _make_campaign(client, oh, material)

    before = (await client.get(
        f"{SELF_BASE}/materials/{material}", headers=rh,
    )).json()["data"]
    assert before["pending_ack"] is None

    with _fanout_patch():
        await send_now(client, db_session, campaign, oh)

    after = (await client.get(
        f"{SELF_BASE}/materials/{material}", headers=rh,
    )).json()["data"]
    assert after["pending_ack"]["campaign_id"] == campaign
    assert after["pending_ack"]["mandatory"] is True

    await client.post(f"{MY}/{campaign}/acknowledge", headers=rh)
    done = (await client.get(
        f"{SELF_BASE}/materials/{material}", headers=rh,
    )).json()["data"]
    assert done["pending_ack"] is None


# ── Resend / close ──────────────────────────────────────────────────


async def test_resend_reaches_only_the_people_who_still_owe_something(
    client, db_session, security_officer, reader,
):
    """Re-pushing to somebody who already did what was asked is the
    fastest way to teach the whole company to ignore this channel."""
    officer, oh = security_officer
    _person, rh = reader
    material = await _publish_material(client, oh)
    campaign = await _make_campaign(client, oh, material)
    with _fanout_patch():
        await send_now(client, db_session, campaign, oh)
    await client.post(f"{MY}/{campaign}/acknowledge", headers=rh)

    with _fanout_patch() as spy:
        await resend(client, db_session, campaign, oh, actor_id=officer.id)
    envelopes = spy.await_args.args[1]
    recipients = {e.recipient.work_email for e in envelopes}
    assert "reader@test.com" not in recipients
    assert "officer@test.com" in recipients

    # The resend is visible on the row it chased.
    target = (await db_session.execute(
        select(SecawCampaignTarget).where(
            SecawCampaignTarget.campaign_id == uuid.UUID(campaign),
            SecawCampaignTarget.person_id == officer.id,
        )
    )).scalar_one()
    assert target.resend_count == 1


async def test_a_resend_does_not_open_a_second_target_row(
    client, db_session, security_officer,
):
    """Otherwise "sent to N people" grows every time somebody chases."""
    officer, oh = security_officer
    material = await _publish_material(client, oh)
    campaign = await _make_campaign(client, oh, material)
    with _fanout_patch():
        await send_now(client, db_session, campaign, oh)
    before = (await client.get(
        f"{CAMPAIGNS}/{campaign}", headers=oh,
    )).json()["data"]["funnel"]["targeted"]

    with _fanout_patch():
        await resend(client, db_session, campaign, oh, actor_id=officer.id)

    after = (await client.get(
        f"{CAMPAIGNS}/{campaign}", headers=oh,
    )).json()["data"]["funnel"]["targeted"]
    assert after == before


async def test_close_stops_the_asking_but_does_not_round_the_number_up(
    client, db_session, security_officer, reader,
):
    officer, oh = security_officer
    _person, rh = reader
    material = await _publish_material(client, oh)
    campaign = await _make_campaign(client, oh, material)
    with _fanout_patch():
        await send_now(client, db_session, campaign, oh)

    assert len((await client.get(MY, headers=rh)).json()["data"]) == 1

    r = await client.post(f"{CAMPAIGNS}/{campaign}/close", headers=oh)
    assert r.status_code == 200

    # Stops appearing as outstanding work …
    assert (await client.get(MY, headers=rh)).json()["data"] == []
    # … and the unacknowledged rows stay exactly as they are.
    funnel = (await client.get(
        f"{CAMPAIGNS}/{campaign}", headers=oh,
    )).json()["data"]["funnel"]
    assert funnel["acknowledged"] == 0
    assert funnel["targeted"] >= 2
    # A closed campaign stops accepting acknowledgements.
    late = await client.post(f"{MY}/{campaign}/acknowledge", headers=rh)
    assert late.status_code == 409


async def test_a_sent_campaign_cannot_be_cancelled_or_have_its_audience_edited(
    client, db_session, security_officer,
):
    """Editing what a set of people were actually asked to do would
    silently restate the record."""
    officer, oh = security_officer
    material = await _publish_material(client, oh)
    campaign = await _make_campaign(client, oh, material)
    with _fanout_patch():
        await send_now(client, db_session, campaign, oh)

    assert (await client.post(
        f"{CAMPAIGNS}/{campaign}/cancel", headers=oh,
    )).status_code == 409
    assert (await client.put(
        f"{CAMPAIGNS}/{campaign}",
        json={"audience_config": {"countries": ["CN"]}}, headers=oh,
    )).status_code == 409
    # Extending a deadline stays legal — a real, benign operational need.
    assert (await client.put(
        f"{CAMPAIGNS}/{campaign}",
        json={"due_at": (datetime.now(UTC) + timedelta(days=7)).isoformat()},
        headers=oh,
    )).status_code == 200


async def test_a_campaign_that_reached_people_cannot_be_deleted(
    client, db_session, security_officer,
):
    officer, oh = security_officer
    material = await _publish_material(client, oh)
    campaign = await _make_campaign(client, oh, material)

    # Before sending, deletion is fine.
    spare = await _make_campaign(client, oh, material, name="spare")
    assert (await client.delete(
        f"{CAMPAIGNS}/{spare}", headers=oh,
    )).status_code == 204

    with _fanout_patch():
        await send_now(client, db_session, campaign, oh)
    r = await client.delete(f"{CAMPAIGNS}/{campaign}", headers=oh)
    assert r.status_code == 409
    assert r.json()["errors"][0]["code"] == "SECAW_CAMPAIGN_HAS_TARGETS"


# ── Reporting ───────────────────────────────────────────────────────


async def test_delivered_comes_from_the_outbox_not_from_a_mirror_column(
    client, db_session, security_officer, reader,
):
    """The target ledger records what WE did. Whether the message landed
    belongs to the outbox, which owns the retry ladder and the dead
    state — a mirrored column would be a second truth."""
    officer, oh = security_officer
    person, _rh = reader
    material = await _publish_material(client, oh)
    campaign = await _make_campaign(client, oh, material)
    with _fanout_patch():
        await send_now(client, db_session, campaign, oh)

    # The fan-out was patched, so no outbox rows exist: delivered is 0
    # even though every target is queued. That asymmetry is the point.
    funnel = (await client.get(
        f"{CAMPAIGNS}/{campaign}", headers=oh,
    )).json()["data"]["funnel"]
    assert funnel["queued"] >= 2
    assert funnel["delivered"] == 0

    db_session.add(NotificationOutbox(
        id=uuid.uuid4(),
        recipient_person_id=person.id,
        recipient_email=person.work_email,
        channels=["feishu"],
        title="x",
        body_text="x",
        status=STATUS_DELIVERED,
        related_resource_type="secaw_campaign",
        related_resource_id=uuid.UUID(campaign),
    ))
    await db_session.commit()

    funnel = (await client.get(
        f"{CAMPAIGNS}/{campaign}", headers=oh,
    )).json()["data"]["funnel"]
    assert funnel["delivered"] == 1

    rows = (await client.get(
        f"{CAMPAIGNS}/{campaign}/targets", headers=oh,
    )).json()["data"]
    mine = next(r for r in rows if r["person_id"] == str(person.id))
    assert mine["delivery_status"] == STATUS_DELIVERED


async def test_department_breakdown_groups_on_the_send_time_snapshot(
    client, db_session, security_officer, reader,
):
    """Somebody who transferred out was reached as a member of their old
    team; moving their row would rewrite a past department's number."""
    officer, oh = security_officer
    _person, _rh = reader
    material = await _publish_material(client, oh)
    campaign = await _make_campaign(client, oh, material)
    with _fanout_patch():
        await send_now(client, db_session, campaign, oh)

    data = (await client.get(f"{CAMPAIGNS}/{campaign}", headers=oh)).json()["data"]
    assert data["by_department"]
    assert sum(b["targeted"] for b in data["by_department"]) == \
        data["funnel"]["targeted"]


async def test_an_unknown_status_filter_is_refused_not_ignored(
    client, security_officer,
):
    """Silently returning everything would have the admin read an
    unfiltered list as a filtered one."""
    officer, oh = security_officer
    r = await client.get(f"{CAMPAIGNS}?status=nonsense", headers=oh)
    assert r.status_code == 422


# ── RBAC ────────────────────────────────────────────────────────────


async def test_employees_cannot_reach_the_campaign_console(
    client, security_officer, reader,
):
    officer, oh = security_officer
    _person, rh = reader
    material = await _publish_material(client, oh)
    campaign = await _make_campaign(client, oh, material)

    for method, path, body in (
        ("get", CAMPAIGNS, None),
        ("get", f"{CAMPAIGNS}/{campaign}", None),
        ("get", f"{CAMPAIGNS}/{campaign}/targets", None),
        ("post", f"{CAMPAIGNS}/preview-audience", {"audience_config": {}}),
        ("post", f"{CAMPAIGNS}/{campaign}/send-now", None),
        ("post", f"{CAMPAIGNS}/{campaign}/resend", None),
        ("post", f"{CAMPAIGNS}/{campaign}/close", None),
        ("delete", f"{CAMPAIGNS}/{campaign}", None),
    ):
        call = getattr(client, method)
        r = await call(path, headers=rh) if body is None else await call(
            path, json=body, headers=rh,
        )
        assert r.status_code == 403, f"{method.upper()} {path} → {r.status_code}"


async def test_management_reads_the_console_but_cannot_send(
    client, db_session, security_officer,
):
    officer, oh = security_officer
    material = await _publish_material(client, oh)
    campaign = await _make_campaign(client, oh, material)
    boss = await make_person(db_session, "boss2@test.com", role="management")
    await db_session.commit()
    bh = headers_for(boss, ["management"])

    listing = await client.get(CAMPAIGNS, headers=bh)
    assert listing.status_code == 200
    assert listing.json()["meta"]["can_manage"] is False
    assert (await client.post(
        f"{CAMPAIGNS}/{campaign}/send-now", headers=bh,
    )).status_code == 403


# ── The shared filter editor's lookups ──────────────────────────────


async def test_the_security_desk_can_load_the_person_filter_lookups(
    client, security_officer, reader,
):
    """The campaign editor mounts the SHARED ``PersonFilterEditor``,
    whose dept-tree / country / entity / person-search lookups live on
    the identity-sync router behind ``require_admin_read`` — which does
    NOT admit ``security_officer``.

    That combination fails **silently**: ``PersonFilterEditor`` swallows
    lookup errors (``.catch(() => {})``), so the security desk would have
    got a form that looks functional, offers an empty department tree,
    and selects nobody. Hence ``require_person_filter_lookup``.

    Pinned here rather than in the identity-sync suite because THIS is
    the surface that needs it; identity-sync's own admin audience never
    exercised the gap.
    """
    _officer, oh = security_officer
    _person, rh = reader

    for path in (
        "/api/v1/identity-sync/lookups/departments",
        "/api/v1/identity-sync/lookups/countries",
        "/api/v1/identity-sync/lookups/entities",
        "/api/v1/identity-sync/lookups/persons?q=read",
    ):
        r = await client.get(path, headers=oh)
        assert r.status_code == 200, f"{path} → {r.status_code} for 安全员"

    # Widening the gate must not have opened it to everybody: a plain
    # employee still cannot enumerate the org tree or search the
    # directory through this route.
    for path in (
        "/api/v1/identity-sync/lookups/departments",
        "/api/v1/identity-sync/lookups/persons?q=read",
    ):
        r = await client.get(path, headers=rh)
        assert r.status_code == 403, f"{path} → {r.status_code} for employee"


# ════════════════════════════════════════════════════════════════════
# The send is off-request, and it is resumable without double-sending
# ════════════════════════════════════════════════════════════════════


async def test_send_now_returns_202_and_hands_the_fanout_off(
    client, db_session, security_officer, spawned_dispatches,
):
    """The endpoint arms the campaign and returns; it must not drain.

    Awaiting the fan-out in-request is what put a 416-person send past
    nginx's 300s ``proxy_read_timeout``: the admin got a failure toast
    over a send that was still running, and the obvious next move —
    reopen the drawer, press again — created a SECOND campaign.
    """
    officer, oh = security_officer
    material = await _publish_material(client, oh)
    campaign = await _make_campaign(client, oh, material)

    with _fanout_patch() as spy:
        r = await client.post(f"{CAMPAIGNS}/{campaign}/send-now", headers=oh)

    assert r.status_code == 202, r.text
    assert r.json()["data"]["status"] == "dispatching"
    assert spy.await_count == 0, "the fan-out must not run inside the request"
    assert spawned_dispatches == [f"secaw.send_now:{campaign}"]

    # Armed, not sent — so a lost task is picked up by the ten-minute tick
    # rather than dropped.
    row = await db_session.get(SecawCampaign, uuid.UUID(campaign))
    await db_session.refresh(row)
    assert row.status == "scheduled"
    assert row.scheduled_at is not None


async def test_the_campaign_is_already_sent_when_the_fanout_starts(
    client, db_session, security_officer, reader,
):
    """Recipients can act on the card the moment it lands.

    Every employee-facing read gates on ``status == 'sent'``. Settling
    the status only AFTER the drain meant that for as long as the drain
    took — 15.5 minutes for 416 people — the ones whose card had already
    arrived opened the material to no confirm button, and ``acknowledge``
    answered 409 ``SECAW_CAMPAIGN_NOT_OPEN``.
    """
    officer, oh = security_officer
    material = await _publish_material(client, oh)
    campaign = await _make_campaign(client, oh, material)
    await client.post(
        f"{CAMPAIGNS}/{campaign}/schedule", json={"scheduled_at": None},
        headers=oh,
    )

    from app.domains.security_awareness.campaign_service import CampaignService

    seen: list[str] = []

    async def _peek(db, envelopes, **kw):
        row = await db.get(SecawCampaign, uuid.UUID(campaign))
        await db.refresh(row)
        seen.append(row.status)
        return {"enqueued": 0, "delivered": 0, "failed": 0}

    with patch("app.notifications.fanout.enqueue_then_drain", new=_peek):
        await CampaignService(db_session).dispatch(uuid.UUID(campaign))

    assert seen == ["sent"], f"status during the drain was {seen}"


async def test_queued_at_is_stamped_per_recipient_not_in_one_bulk_write(
    client, db_session, security_officer, reader,
):
    """The stamp rides in the enqueue's own transaction.

    One bulk UPDATE after the drain meant a drain that died at 90% left
    ALL of them unstamped: the funnel read zero queued and a resumed
    dispatch had no way to tell who had already been reached.
    """
    officer, oh = security_officer
    reader_person, _rh = reader
    material = await _publish_material(client, oh)
    campaign = await _make_campaign(client, oh, material)
    await client.post(
        f"{CAMPAIGNS}/{campaign}/schedule", json={"scheduled_at": None},
        headers=oh,
    )

    from app.domains.security_awareness.campaign_service import CampaignService

    # The reader's enqueue fails; the officer's succeeds.
    with _fanout_patch(failed_for={"reader@test.com"}):
        result = await CampaignService(db_session).dispatch(
            uuid.UUID(campaign),
        )
    assert result["failed"] == 1

    rows = {
        t.person_id: t
        for t in (await db_session.execute(
            select(SecawCampaignTarget).where(
                SecawCampaignTarget.campaign_id == uuid.UUID(campaign),
            )
        )).scalars().all()
    }
    assert rows[officer.id].queued_at is not None
    assert rows[reader_person.id].queued_at is None, (
        "a recipient whose enqueue failed must not look reached"
    )

    row = await db_session.get(SecawCampaign, uuid.UUID(campaign))
    await db_session.refresh(row)
    assert "1 recipient" in (row.last_dispatch_error or "")


async def test_a_wedged_send_is_resumed_and_reaches_only_the_people_it_missed(
    client, db_session, security_officer, reader,
):
    """The whole point of the per-recipient stamp.

    A dispatch whose process died is re-armed by ``dispatch_due`` after
    :data:`STALE_SENDING_AFTER`. The resumed run must reach the people
    the dead one never got to and NOBODY ELSE — re-walking the whole
    audience is how 400 people get the same required reading twice.
    """
    from app.domains.security_awareness.campaign_service import (
        STALE_SENDING_AFTER,
        CampaignService,
    )

    officer, oh = security_officer
    reader_person, _rh = reader
    material = await _publish_material(client, oh)
    campaign = await _make_campaign(client, oh, material)
    await client.post(
        f"{CAMPAIGNS}/{campaign}/schedule", json={"scheduled_at": None},
        headers=oh,
    )

    # First run reaches the officer and dies before the reader.
    with _fanout_patch(failed_for={"reader@test.com"}):
        await CampaignService(db_session).dispatch(uuid.UUID(campaign))

    # Wedge it exactly the way a killed container does: status stuck in
    # ``sending``, older than the stall ceiling.
    await db_session.execute(
        update(SecawCampaign)
        .where(SecawCampaign.id == uuid.UUID(campaign))
        .values(
            status="sending",
            updated_at=datetime.now(UTC) - STALE_SENDING_AFTER
            - timedelta(minutes=1),
            # Also pushed back a minute: ``scheduled_at`` is a DATETIME(0)
            # and MySQL ROUNDS the fractional second on insert, so a
            # campaign armed with ``when=None`` can be stored up to half a
            # second in the future and miss its own ``scheduled_at <= now``
            # for that long. Harmless in production (the tick is ten
            # minutes later) and a coin-flip flake in a test that dispatches
            # immediately.
            scheduled_at=datetime.now(UTC) - timedelta(minutes=1),
        )
    )
    await db_session.commit()

    reached: list[list[str]] = []

    async def _record(db, envelopes, **kw):
        reached.append([e.recipient.work_email for e in envelopes])
        on_enqueued = kw.get("on_enqueued")
        for env in envelopes:
            if on_enqueued is not None:
                await on_enqueued(env, uuid.uuid4())
            await db.commit()
        return {
            "enqueued": len(envelopes), "delivered": len(envelopes),
            "failed": 0,
        }

    with patch("app.notifications.fanout.enqueue_then_drain", new=_record):
        summary = await CampaignService(db_session).dispatch_due()

    assert summary["stale_recovered"] == 1
    assert summary["dispatched"] == 1
    assert reached == [["reader@test.com"]], (
        f"the resumed run must skip whoever was already queued, got {reached}"
    )

    rows = (await db_session.execute(
        select(SecawCampaignTarget).where(
            SecawCampaignTarget.campaign_id == uuid.UUID(campaign),
        )
    )).scalars().all()
    assert len(rows) == 2
    assert all(t.queued_at is not None for t in rows)
    assert all(t.resend_count == 0 for t in rows), (
        "a resumed dispatch is not a chase — nobody nudged them"
    )


# ════════════════════════════════════════════════════════════════════
# The duplicate-send guard
# ════════════════════════════════════════════════════════════════════


async def test_the_same_material_to_the_same_audience_twice_is_refused(
    client, db_session, security_officer,
):
    """The exact shape of a double submit: reopen the drawer after the
    request appeared to fail, press the button again. That does not retry
    anything — it creates a second campaign, and every recipient gets the
    same required reading twice."""
    officer, oh = security_officer
    material = await _publish_material(client, oh)
    first = await _make_campaign(client, oh, material)
    with _fanout_patch():
        await send_now(client, db_session, first, oh)

    second = await _make_campaign(client, oh, material)
    r = await client.post(f"{CAMPAIGNS}/{second}/send-now", headers=oh)
    assert r.status_code == 409
    assert r.json()["errors"][0]["code"] == "SECAW_CAMPAIGN_DUPLICATE_SEND"
    # The message has to name what it is protecting, or the admin's only
    # move is to keep clicking.
    assert "2026 Q3 钓鱼宣贯" in r.json()["errors"][0]["message"]

    # Still a draft — refusing to arm must not half-arm it.
    row = await db_session.get(SecawCampaign, uuid.UUID(second))
    await db_session.refresh(row)
    assert row.status == "draft"


async def test_a_deliberate_second_wave_can_say_so(
    client, db_session, security_officer,
):
    """A guard with no way past it is a guard people route around by
    editing the audience by one person."""
    officer, oh = security_officer
    material = await _publish_material(client, oh)
    first = await _make_campaign(client, oh, material)
    with _fanout_patch():
        await send_now(client, db_session, first, oh)

    second = await _make_campaign(client, oh, material)
    with _fanout_patch():
        await send_now(client, db_session, second, oh, force=True)

    row = await db_session.get(SecawCampaign, uuid.UUID(second))
    await db_session.refresh(row)
    assert row.status == "sent"


async def test_a_different_audience_is_not_a_duplicate(
    client, db_session, security_officer, reader,
):
    """The guard keys on the form, not on the material — a second wave to
    another department is ordinary work and must not need ``force``."""
    officer, oh = security_officer
    reader_person, _rh = reader
    material = await _publish_material(client, oh)
    first = await _make_campaign(
        client, oh, material,
        audience_config={"person_ids": [str(officer.id)]},
    )
    with _fanout_patch():
        await send_now(client, db_session, first, oh)

    second = await _make_campaign(
        client, oh, material, name="第二波",
        audience_config={"person_ids": [str(reader_person.id)]},
    )
    with _fanout_patch():
        await send_now(client, db_session, second, oh)

    row = await db_session.get(SecawCampaign, uuid.UUID(second))
    await db_session.refresh(row)
    assert row.status == "sent"


async def test_resend_is_202_and_refuses_an_unsendable_campaign_in_request(
    client, db_session, security_officer,
):
    """The chase drains off the request too — on day one "everybody who
    has not acknowledged" is most of the audience, so it is the same
    fan-out and the same minutes.

    But the *refusal* stays synchronous: a 202 followed by a background
    409 is a click that silently did nothing.
    """
    officer, oh = security_officer
    material = await _publish_material(client, oh)
    campaign = await _make_campaign(client, oh, material)

    # Never sent → refused where the admin can see it.
    r = await client.post(f"{CAMPAIGNS}/{campaign}/resend", headers=oh)
    assert r.status_code == 409
    assert r.json()["errors"][0]["code"] == "SECAW_CAMPAIGN_NOT_SENT"

    with _fanout_patch():
        await send_now(client, db_session, campaign, oh)

    with _fanout_patch() as spy:
        r = await client.post(f"{CAMPAIGNS}/{campaign}/resend", headers=oh)
    assert r.status_code == 202
    assert r.json()["data"]["status"] == "dispatching"
    assert spy.await_count == 0, "the chase must not run inside the request"


async def test_an_identical_campaign_armed_for_next_week_is_not_a_duplicate(
    client, db_session, security_officer,
):
    """The window is between the two SEND MOMENTS, not "how recently was
    the twin touched".

    A first cut compared the twin's ``updated_at`` against ``now``, which
    refused exactly the case the scheduler takes a date for: the same
    required reading, armed for next week, twenty minutes after the
    previous wave went out.
    """
    officer, oh = security_officer
    material = await _publish_material(client, oh)
    first = await _make_campaign(client, oh, material)
    with _fanout_patch():
        await send_now(client, db_session, first, oh)

    second = await _make_campaign(client, oh, material, name="下周那一波")
    later = (datetime.now(UTC) + timedelta(days=7)).isoformat()
    r = await client.post(
        f"{CAMPAIGNS}/{second}/schedule", json={"scheduled_at": later},
        headers=oh,
    )
    assert r.status_code == 200, r.text

    row = await db_session.get(SecawCampaign, uuid.UUID(second))
    await db_session.refresh(row)
    assert row.status == "scheduled"


async def test_arming_the_same_send_twice_for_the_same_hour_is_refused(
    client, db_session, security_officer,
):
    """Two identical campaigns both armed for Monday 09:00 is the same
    double send as pressing the button twice — and the status-based
    version of this guard could not see it, because neither one has gone
    out yet."""
    officer, oh = security_officer
    material = await _publish_material(client, oh)
    when = (datetime.now(UTC) + timedelta(days=3)).isoformat()

    first = await _make_campaign(client, oh, material)
    r = await client.post(
        f"{CAMPAIGNS}/{first}/schedule", json={"scheduled_at": when},
        headers=oh,
    )
    assert r.status_code == 200, r.text

    second = await _make_campaign(client, oh, material, name="又armed了一次")
    r = await client.post(
        f"{CAMPAIGNS}/{second}/schedule", json={"scheduled_at": when},
        headers=oh,
    )
    assert r.status_code == 409
    assert r.json()["errors"][0]["code"] == "SECAW_CAMPAIGN_DUPLICATE_SEND"
    assert "armed" in r.json()["errors"][0]["message"]


async def test_editing_an_old_sent_campaign_does_not_block_a_new_one(
    client, db_session, security_officer,
):
    """``updated_at`` moves for reasons that have nothing to do with
    sending — a rename, a new due date. Keying the window on it made a
    months-old campaign block a legitimate new one for half an hour just
    because somebody had touched it."""
    officer, oh = security_officer
    material = await _publish_material(client, oh)
    old = await _make_campaign(client, oh, material)
    with _fanout_patch():
        await send_now(client, db_session, old, oh)

    # It went out long ago …
    await db_session.execute(
        update(SecawCampaign)
        .where(SecawCampaign.id == uuid.UUID(old))
        .values(sent_at=datetime.now(UTC) - timedelta(days=90))
    )
    await db_session.commit()
    # … and somebody renames it today, which moves ``updated_at`` to now.
    r = await client.put(
        f"{CAMPAIGNS}/{old}", json={"name": "改个名字"}, headers=oh,
    )
    assert r.status_code == 200, r.text

    fresh = await _make_campaign(client, oh, material, name="今年这一波")
    with _fanout_patch():
        await send_now(client, db_session, fresh, oh)

    row = await db_session.get(SecawCampaign, uuid.UUID(fresh))
    await db_session.refresh(row)
    assert row.status == "sent"