webmail-client/backend/tests/outbox/test_outbox.py

"""Outbox: MIME assembly, drafts, send, undo, send job (SPEC §6.5)."""

from __future__ import annotations

import asyncio
import email
import uuid
from collections.abc import AsyncIterator
from dataclasses import dataclass, field
from datetime import UTC, datetime, timedelta
from email.policy import default as default_policy

import httpx
import pytest
import pytest_asyncio
from arq import Retry
from httpx import ASGITransport
from sqlalchemy import delete, select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.pool import NullPool

from app import queue as app_queue
from app.config import get_settings
from app.db import session_scope
from app.identity.passwords import hash_password
from app.identity.sessions import create_session
from app.main import create_app
from app.models import Account, Draft, KnownRecipient, Mailbox, Message, Outbox, User
from app.outbox import jobs as outbox_jobs
from app.outbox.mime_build import build_mime
from app.providers.base import ProviderConnectError, ProviderRejectedError

CSRF = "csrf"


# --- pure MIME assembly ------------------------------------------------------


def test_build_mime_sets_threading_and_alternative() -> None:
    account = Account(
        email="me@example.com",
        display_name="Me",
        imap_config={},
        smtp_config={},
        secret_ct=b"",
        discovery_src="manual",
        user_id=uuid.uuid4(),
    )
    parent = Message(
        message_id_hdr="orig@x.com",
        references_hdr=["root@x.com"],
        user_id=uuid.uuid4(),
        account_id=uuid.uuid4(),
        mailbox_id=uuid.uuid4(),
        uid=1,
        internal_date=datetime.now(UTC),
        blob_sha256="0",
    )
    draft = Draft(
        user_id=uuid.uuid4(),
        account_id=uuid.uuid4(),
        kind="reply",
        payload={
            "to": ["bob@example.com"],
            "cc": ["c@example.com"],
            "bcc": ["secret@example.com"],
            "subject": "Re: Hi",
            "html": "<p>Hello <b>there</b></p>",
            "text": "",
        },
    )
    transmit_raw, sent_raw, env_from, env_to = build_mime(account, draft, parent)

    assert env_from == "me@example.com"
    assert set(env_to) == {"bob@example.com", "c@example.com", "secret@example.com"}
    msg = email.message_from_bytes(transmit_raw, policy=default_policy)
    assert msg["From"] == "Me <me@example.com>"
    assert msg["To"] == "bob@example.com"
    assert msg["Bcc"] is None  # transmitted copy carries no Bcc
    assert "secret@example.com" not in transmit_raw.decode()  # bcc not transmitted
    # ...but the Sent copy records the Bcc so the sender keeps the blind recipient.
    sent_msg = email.message_from_bytes(sent_raw, policy=default_policy)
    assert sent_msg["Bcc"] == "secret@example.com"
    assert msg["In-Reply-To"] == "<orig@x.com>"
    assert "<root@x.com>" in msg["References"] and "<orig@x.com>" in msg["References"]
    # multipart/alternative with an auto-generated text part
    body_types = {p.get_content_type() for p in msg.walk()}
    assert "text/plain" in body_types and "text/html" in body_types


def test_build_mime_includes_attachments() -> None:
    account = Account(
        email="me@example.com",
        display_name="Me",
        imap_config={},
        smtp_config={},
        secret_ct=b"",
        discovery_src="manual",
        user_id=uuid.uuid4(),
    )
    draft = Draft(
        user_id=uuid.uuid4(),
        account_id=uuid.uuid4(),
        kind="new",
        payload={"to": ["bob@example.com"], "subject": "Files", "html": "<p>hi</p>", "text": ""},
    )
    raw, _sent, _from, _to = build_mime(
        account, draft, None, [("report.pdf", "application/pdf", b"%PDF-1.4 data")]
    )
    msg = email.message_from_bytes(raw, policy=default_policy)
    attached = [p for p in msg.walk() if p.get_filename() == "report.pdf"]
    assert len(attached) == 1
    assert attached[0].get_content_type() == "application/pdf"
    assert attached[0].get_payload(decode=True) == b"%PDF-1.4 data"


# --- API + send job ----------------------------------------------------------


@dataclass
class Ctx:
    client: httpx.AsyncClient
    session: AsyncSession
    user_id: uuid.UUID
    account_id: uuid.UUID
    sent_path: str
    reply_to_id: uuid.UUID
    enqueued: list[tuple] = field(default_factory=list)


def _hdr() -> dict[str, str]:
    return {"X-CSRF": CSRF}


@pytest_asyncio.fixture
async def ctx(monkeypatch) -> AsyncIterator[Ctx]:  # type: ignore[no-untyped-def]
    engine = create_async_engine(get_settings().database_url, poolclass=NullPool)
    sm = async_sessionmaker(engine, expire_on_commit=False)
    session = sm()
    user = User(handle=f"ob{uuid.uuid4().hex[:8]}@test.local", password_hash=hash_password("pw"))
    session.add(user)
    await session.flush()
    account = Account(
        user_id=user.id,
        email="me@example.com",
        display_name="Me",
        imap_config={"host": "imap.example.com", "port": 993, "tls": "ssl"},
        smtp_config={"host": "smtp.example.com", "port": 465, "tls": "ssl"},
        secret_ct=b"",
        discovery_src="manual",
    )
    session.add(account)
    await session.flush()
    inbox = Mailbox(account_id=account.id, user_id=user.id, path="INBOX", special_use="inbox")
    sent = Mailbox(account_id=account.id, user_id=user.id, path="Sent", special_use="sent")
    session.add_all([inbox, sent])
    await session.flush()
    parent = Message(
        user_id=user.id,
        account_id=account.id,
        mailbox_id=inbox.id,
        uid=1,
        message_id_hdr="orig@x.com",
        references_hdr=[],
        subject="Hi",
        internal_date=datetime.now(UTC),
        blob_sha256="0" * 64,
    )
    session.add(parent)
    await session.commit()

    c = Ctx(
        client=None,
        session=session,
        user_id=user.id,
        account_id=account.id,  # type: ignore[arg-type]
        sent_path="Sent",
        reply_to_id=parent.id,
    )

    class FakePool:
        async def enqueue_job(self, *args: object, **kwargs: object) -> None:
            c.enqueued.append((args, kwargs))

    async def fake_pool() -> FakePool:
        return FakePool()

    monkeypatch.setattr(app_queue, "get_arq_pool", fake_pool)

    token = await create_session(user.id, "pytest")
    c.client = httpx.AsyncClient(
        transport=ASGITransport(app=create_app()),
        base_url="http://testserver",
        cookies={"session": token, "csrf": CSRF},
    )
    try:
        yield c
    finally:
        await c.client.aclose()
        # outbox.user_id has no foreign key: drop the rows with the user.
        await session.execute(delete(Outbox).where(Outbox.user_id == user.id))
        await session.execute(delete(User).where(User.id == user.id))
        await session.commit()
        await session.close()
        await engine.dispose()


def _draft_body(ctx: Ctx) -> dict:
    return {
        "account_id": str(ctx.account_id),
        "reply_to_message_id": str(ctx.reply_to_id),
        "kind": "reply",
        "payload": {"to": ["bob@example.com"], "subject": "Re: Hi", "text": "hello"},
    }


async def test_draft_crud(ctx: Ctx) -> None:
    created = await ctx.client.post("/api/drafts", json=_draft_body(ctx), headers=_hdr())
    assert created.status_code == 200
    did = created.json()["id"]

    body = _draft_body(ctx)
    body["payload"]["subject"] = "Re: Hi (edited)"
    updated = await ctx.client.put(f"/api/drafts/{did}", json=body, headers=_hdr())
    assert updated.status_code == 200
    assert updated.json()["payload"]["subject"] == "Re: Hi (edited)"

    deleted = await ctx.client.request("DELETE", f"/api/drafts/{did}", headers=_hdr())
    assert deleted.status_code == 200


async def test_remote_drafts_listed_and_adopted(ctx: Ctx) -> None:
    from datetime import UTC, datetime

    from app.models import BodyPart

    # A "foreign" draft (as if created in another client) in the Drafts folder.
    drafts_mb = Mailbox(
        account_id=ctx.account_id, user_id=ctx.user_id, path="DRAFTS", special_use="drafts"
    )
    ctx.session.add(drafts_mb)
    await ctx.session.flush()
    msg = Message(
        user_id=ctx.user_id,
        account_id=ctx.account_id,
        mailbox_id=drafts_mb.id,
        uid=94,
        subject="Home draft",
        to_addr=[{"name": "", "email": "bob@example.com"}],
        cc_addr=[],
        bcc_addr=[],
        internal_date=datetime.now(UTC),
        blob_sha256="x" * 64,
        snippet="hello there",
    )
    ctx.session.add(msg)
    await ctx.session.flush()
    ctx.session.add(BodyPart(message_id=msg.id, kind="html_sanitized", content="<p>hello</p>"))
    ctx.session.add(BodyPart(message_id=msg.id, kind="text", content="hello there"))
    await ctx.session.commit()

    # It shows in the remote drafts list.
    listing = await ctx.client.get("/api/drafts/remote")
    assert listing.status_code == 200
    assert any(i["message_id"] == str(msg.id) and i["imap_uid"] == 94 for i in listing.json())

    # Adopting turns it into an editable local draft (linked by UID).
    adopted = await ctx.client.post(f"/api/drafts/adopt/{msg.id}", headers=_hdr())
    assert adopted.status_code == 200, adopted.text
    draft = adopted.json()
    assert draft["payload"]["subject"] == "Home draft"
    assert draft["payload"]["to"] == ["bob@example.com"]
    assert draft["payload"]["html"] == "<p>hello</p>"

    # Now it's excluded from the remote list (we own that UID)...
    listing2 = await ctx.client.get("/api/drafts/remote")
    assert all(i["message_id"] != str(msg.id) for i in listing2.json())
    # ...and adopting again returns the same local draft (idempotent).
    again = await ctx.client.post(f"/api/drafts/adopt/{msg.id}", headers=_hdr())
    assert again.json()["id"] == draft["id"]


async def test_adopt_remote_draft_carries_attachments(ctx: Ctx) -> None:
    from datetime import UTC, datetime

    from app.indexer import blobstore
    from app.models import Attachment

    drafts_mb = Mailbox(
        account_id=ctx.account_id, user_id=ctx.user_id, path="DRAFTS", special_use="drafts"
    )
    ctx.session.add(drafts_mb)
    await ctx.session.flush()
    msg = Message(
        user_id=ctx.user_id,
        account_id=ctx.account_id,
        mailbox_id=drafts_mb.id,
        uid=95,
        subject="Home draft with file",
        to_addr=[{"name": "", "email": "bob@example.com"}],
        cc_addr=[],
        bcc_addr=[],
        internal_date=datetime.now(UTC),
        blob_sha256="y" * 64,
        snippet="",
        has_attachments=True,
    )
    ctx.session.add(msg)
    await ctx.session.flush()
    # A real (downloadable) attachment plus an inline cid image, which must NOT be carried.
    sha = blobstore.store(ctx.user_id, b"file contents")
    ctx.session.add(
        Attachment(
            message_id=msg.id,
            user_id=ctx.user_id,
            filename="report.pdf",
            mime_type="application/pdf",
            size_bytes=13,
            content_id=None,
            blob_sha256=sha,
        )
    )
    ctx.session.add(
        Attachment(
            message_id=msg.id,
            user_id=ctx.user_id,
            filename="logo.png",
            mime_type="image/png",
            size_bytes=3,
            content_id="logo@x",
            blob_sha256=blobstore.store(ctx.user_id, b"img"),
        )
    )
    await ctx.session.commit()

    adopted = await ctx.client.post(f"/api/drafts/adopt/{msg.id}", headers=_hdr())
    assert adopted.status_code == 200, adopted.text
    ids = adopted.json()["payload"]["attachment_ids"]
    assert len(ids) == 1  # only the real attachment, not the inline image

    # The compose-attachment metadata resolves — this is what the editor renders and
    # what load_attachments re-attaches on send.
    meta = await ctx.client.get("/api/attachments", params={"ids": ",".join(ids)})
    assert meta.status_code == 200
    body = meta.json()
    assert [a["filename"] for a in body] == ["report.pdf"]
    assert body[0]["size_bytes"] == 13


async def test_publish_draft_enqueues_imap_mirror(ctx: Ctx) -> None:
    did = (await ctx.client.post("/api/drafts", json=_draft_body(ctx), headers=_hdr())).json()["id"]
    ctx.enqueued.clear()
    resp = await ctx.client.post(f"/api/drafts/{did}/publish", headers=_hdr())
    assert resp.status_code == 200
    assert ctx.enqueued[-1][0] == ("push_draft_to_imap_job", did)


async def test_delete_draft_expunges_mirror_when_present(ctx: Ctx) -> None:
    did = (await ctx.client.post("/api/drafts", json=_draft_body(ctx), headers=_hdr())).json()["id"]
    # Simulate a previously mirrored copy (imap_uid set).
    draft = await ctx.session.get(Draft, uuid.UUID(did))
    assert draft is not None
    draft.imap_uid = 88
    await ctx.session.commit()

    ctx.enqueued.clear()
    resp = await ctx.client.request("DELETE", f"/api/drafts/{did}", headers=_hdr())
    assert resp.status_code == 200
    assert ctx.enqueued[-1][0] == ("remove_imap_draft_job", str(ctx.account_id), 88)


async def test_upload_and_list_compose_attachment(ctx: Ctx) -> None:
    resp = await ctx.client.post(
        "/api/attachments",
        files={"file": ("hello.txt", b"file contents", "text/plain")},
        headers=_hdr(),
    )
    assert resp.status_code == 200, resp.text
    data = resp.json()
    assert data["filename"] == "hello.txt"
    assert data["mime_type"] == "text/plain"
    assert data["size_bytes"] == len(b"file contents")

    listing = await ctx.client.get(f"/api/attachments?ids={data['id']}", headers=_hdr())
    assert listing.status_code == 200
    assert [a["id"] for a in listing.json()] == [data["id"]]


async def test_upload_rejects_oversized_attachment(ctx: Ctx, monkeypatch) -> None:  # type: ignore[no-untyped-def]
    from app.api import compose as compose_module

    monkeypatch.setattr(compose_module, "MAX_ATTACHMENT_BYTES", 4)
    resp = await ctx.client.post(
        "/api/attachments",
        files={"file": ("big.bin", b"12345", "application/octet-stream")},
        headers=_hdr(),
    )
    assert resp.status_code == 413
    assert resp.json()["error"]["code"] == "too_large"


async def test_send_creates_outbox_and_defers(ctx: Ctx) -> None:
    did = (await ctx.client.post("/api/drafts", json=_draft_body(ctx), headers=_hdr())).json()["id"]
    resp = await ctx.client.post("/api/send", json={"draft_id": did}, headers=_hdr())
    assert resp.status_code == 200
    outbox_id = resp.json()["outbox_id"]
    # The window as a duration too, so the browser's clock does not matter.
    assert 0 < resp.json()["undo_seconds"] <= get_settings().undo_seconds

    ob = await ctx.session.get(Outbox, uuid.UUID(outbox_id))
    assert ob is not None and ob.state == "undo_window"
    # scheduled with a defer (the undo window)
    assert ctx.enqueued and ctx.enqueued[-1][0][0] == "send_job"
    assert "_defer_by" in ctx.enqueued[-1][1]


async def test_undo_cancels(ctx: Ctx) -> None:
    did = (await ctx.client.post("/api/drafts", json=_draft_body(ctx), headers=_hdr())).json()["id"]
    outbox_id = (await ctx.client.post("/api/send", json={"draft_id": did}, headers=_hdr())).json()[
        "outbox_id"
    ]
    undo = await ctx.client.post(f"/api/send/{outbox_id}/undo", headers=_hdr())
    assert undo.status_code == 200
    ob = await ctx.session.get(Outbox, uuid.UUID(outbox_id))
    await ctx.session.refresh(ob)
    assert ob is not None and ob.state == "cancelled"


class FakeProvider:
    def __init__(self) -> None:
        self.sent: list = []
        self.appended: list = []
        self.raise_on_send = False
        self.send_error: Exception | None = None
        self.append_error: Exception | None = None

    async def send(self, raw: bytes, env_from: str, env_to: list) -> None:
        if self.raise_on_send:
            raise RuntimeError("smtp down")
        if self.send_error is not None:
            raise self.send_error
        self.sent.append((env_from, env_to))

    async def append(self, mailbox: str, raw: bytes, flags: frozenset) -> int:
        if self.append_error is not None:
            raise self.append_error
        self.appended.append(mailbox)
        return 99

    async def close(self) -> None:
        return None


async def _make_outbox(ctx: Ctx, state: str = "undo_window") -> uuid.UUID:
    draft = Draft(
        user_id=ctx.user_id,
        account_id=ctx.account_id,
        reply_to_message_id=ctx.reply_to_id,
        kind="reply",
        payload={"to": ["bob@example.com"], "subject": "Re: Hi", "text": "hello"},
    )
    ctx.session.add(draft)
    await ctx.session.flush()
    outbox = Outbox(
        user_id=ctx.user_id, draft_id=draft.id, state=state, send_after=datetime.now(UTC)
    )
    ctx.session.add(outbox)
    await ctx.session.commit()
    return outbox.id


async def test_send_job_success(ctx: Ctx, monkeypatch) -> None:  # type: ignore[no-untyped-def]
    oid = await _make_outbox(ctx)
    fake = FakeProvider()
    monkeypatch.setattr(outbox_jobs, "build_provider", lambda _a: fake)

    ob0 = await ctx.session.get(Outbox, oid)
    assert ob0 is not None
    draft_id = ob0.draft_id

    await outbox_jobs.send_job({"job_try": 1}, str(oid))

    ob = await ctx.session.get(Outbox, oid)
    await ctx.session.refresh(ob)
    assert ob is not None and ob.state == "sent"
    assert len(fake.sent) == 1
    assert fake.appended == ["Sent"]  # non-Gmail -> we append

    # A successfully sent message's draft is deleted (verify in a fresh session).
    ctx.session.expunge_all()
    async with session_scope() as verify:
        assert await verify.get(Draft, draft_id) is None
        # The sent copy is persisted into the Sent mailbox (append UID 99).
        sent_mb = await verify.scalar(
            select(Mailbox).where(
                Mailbox.account_id == ctx.account_id, Mailbox.special_use == "sent"
            )
        )
        assert sent_mb is not None
        sent_msg = await verify.scalar(
            select(Message).where(Message.mailbox_id == sent_mb.id, Message.uid == 99)
        )
        assert sent_msg is not None and sent_msg.subject == "Re: Hi"
    kr = await ctx.session.scalar(
        select(KnownRecipient).where(
            KnownRecipient.user_id == ctx.user_id, KnownRecipient.email == "bob@example.com"
        )
    )
    assert kr is not None and kr.times_used == 3  # sent-to weight


async def test_send_job_skips_when_cancelled(ctx: Ctx, monkeypatch) -> None:  # type: ignore[no-untyped-def]
    oid = await _make_outbox(ctx, state="cancelled")
    fake = FakeProvider()
    monkeypatch.setattr(outbox_jobs, "build_provider", lambda _a: fake)

    await outbox_jobs.send_job({"job_try": 1}, str(oid))
    assert fake.sent == []  # no SMTP traffic for a cancelled send


async def test_send_job_failure_marks_failed(ctx: Ctx, monkeypatch) -> None:  # type: ignore[no-untyped-def]
    oid = await _make_outbox(ctx)
    fake = FakeProvider()
    fake.raise_on_send = True
    monkeypatch.setattr(outbox_jobs, "build_provider", lambda _a: fake)

    await outbox_jobs.send_job({"job_try": 1}, str(oid))
    ob = await ctx.session.get(Outbox, oid)
    await ctx.session.refresh(ob)
    assert ob is not None and ob.state == "failed" and ob.error


# --- a send the server could not take: Scheduled, retried for 24 hours ------------


async def _outbox(ctx: Ctx, oid: uuid.UUID) -> Outbox:
    ob = await ctx.session.get(Outbox, oid)
    assert ob is not None
    await ctx.session.refresh(ob)
    return ob


async def test_unreachable_smtp_is_retried_and_waits_in_scheduled(ctx: Ctx, monkeypatch) -> None:  # type: ignore[no-untyped-def]
    oid = await _make_outbox(ctx)
    fake = FakeProvider()
    fake.send_error = ProviderConnectError("cannot connect to smtp.example.com")
    monkeypatch.setattr(outbox_jobs, "build_provider", lambda _a: fake)

    before = datetime.now(UTC)
    await outbox_jobs.send_job({"job_try": 1}, str(oid))

    ob = await _outbox(ctx, oid)
    assert ob.state == "retrying" and ob.attempts == 1 and "cannot connect" in (ob.error or "")
    assert ob.retry_until is not None
    assert ob.retry_until >= before + timedelta(hours=23, minutes=59)
    assert ob.next_attempt_at is not None
    assert timedelta(seconds=55) < ob.next_attempt_at - before < timedelta(seconds=70)
    args, kwargs = ctx.enqueued[-1]
    assert args == ("send_job", str(oid))
    assert kwargs["_defer_until"] == ob.next_attempt_at and kwargs["_job_id"] == f"send:{oid}:1"

    # Waiting in Scheduled, not in Drafts.
    drafts = (await ctx.client.get("/api/drafts")).json()
    assert all(d["id"] != str(ob.draft_id) for d in drafts)
    waiting = (await ctx.client.get("/api/outbox")).json()
    assert [(w["outbox_id"], w["state"], w["subject"]) for w in waiting] == [
        (str(oid), "retrying", "Re: Hi")
    ]

    # The server is back: the planned retry sends it.
    fake.send_error = None
    await outbox_jobs.send_job({"job_try": 1}, str(oid))
    assert (await _outbox(ctx, oid)).state == "sent" and len(fake.sent) == 1
    assert (await ctx.client.get("/api/outbox")).json() == []


def test_retry_gaps_grow_then_hourly() -> None:
    assert [outbox_jobs._retry_gap(n) for n in range(1, 8)] == [60, 120, 300, 600, 1800, 3600, 3600]


async def test_send_gives_up_after_24_hours(ctx: Ctx, monkeypatch) -> None:  # type: ignore[no-untyped-def]
    oid = await _make_outbox(ctx, state="retrying")
    ob = await _outbox(ctx, oid)
    ob.retry_until = datetime.now(UTC) - timedelta(seconds=1)
    await ctx.session.commit()
    fake = FakeProvider()
    fake.send_error = ProviderConnectError("still down")
    monkeypatch.setattr(outbox_jobs, "build_provider", lambda _a: fake)

    await outbox_jobs.send_job({"job_try": 1}, str(oid))
    ob = await _outbox(ctx, oid)
    assert ob.state == "failed" and ob.next_attempt_at is None
    assert not any(a[0] == "send_job" for a, _ in ctx.enqueued)


async def test_permanent_refusal_fails_at_once(ctx: Ctx, monkeypatch) -> None:  # type: ignore[no-untyped-def]
    oid = await _make_outbox(ctx)
    fake = FakeProvider()
    fake.send_error = ProviderRejectedError("550 no such user")
    monkeypatch.setattr(outbox_jobs, "build_provider", lambda _a: fake)

    await outbox_jobs.send_job({"job_try": 1}, str(oid))
    ob = await _outbox(ctx, oid)
    assert ob.state == "failed" and "550" in (ob.error or "")
    assert not any(a[0] == "send_job" for a, _ in ctx.enqueued)
    # Shown in Scheduled as failed, with the reason.
    waiting = (await ctx.client.get("/api/outbox")).json()
    assert waiting[0]["state"] == "failed" and "550" in waiting[0]["error"]


async def test_delivered_but_not_filed_counts_as_sent(ctx: Ctx, monkeypatch) -> None:  # type: ignore[no-untyped-def]
    oid = await _make_outbox(ctx)
    fake = FakeProvider()
    fake.append_error = ProviderConnectError("imap unreachable")
    monkeypatch.setattr(outbox_jobs, "build_provider", lambda _a: fake)

    await outbox_jobs.send_job({"job_try": 1}, str(oid))
    assert (await _outbox(ctx, oid)).state == "sent" and len(fake.sent) == 1

    # The copy is not lost: kept, and a job files it once IMAP answers again.
    args, _ = ctx.enqueued[-1]
    assert args[0] == "file_sent_copy_job"
    _, account_id, user_id, sha = args
    with pytest.raises(Retry):  # still unreachable: tried again later
        await outbox_jobs.file_sent_copy_job({"job_try": 1}, account_id, user_id, sha)
    fake.append_error = None
    await outbox_jobs.file_sent_copy_job({"job_try": 2}, account_id, user_id, sha)
    assert fake.appended == ["Sent"]
    ctx.session.expunge_all()
    async with session_scope() as verify:
        sent_msg = await verify.scalar(
            select(Message)
            .join(Mailbox, Mailbox.id == Message.mailbox_id)
            .where(Mailbox.account_id == ctx.account_id, Mailbox.special_use == "sent")
        )
        assert sent_msg is not None and sent_msg.subject == "Re: Hi"


async def test_two_jobs_for_one_send_send_once(ctx: Ctx, monkeypatch) -> None:  # type: ignore[no-untyped-def]
    oid = await _make_outbox(ctx, state="retrying")
    fake = FakeProvider()
    monkeypatch.setattr(outbox_jobs, "build_provider", lambda _a: fake)

    await asyncio.gather(
        outbox_jobs.send_job({"job_try": 1}, str(oid)),
        outbox_jobs.send_job({"job_try": 1}, str(oid)),
    )
    assert len(fake.sent) == 1


async def test_retry_now_and_cancel(ctx: Ctx) -> None:
    oid = await _make_outbox(ctx, state="failed")
    resp = await ctx.client.post(f"/api/outbox/{oid}/retry", headers=_hdr())
    assert resp.status_code == 200
    ob = await _outbox(ctx, oid)
    assert ob.state == "retrying"
    assert ob.retry_until is not None and ob.retry_until > datetime.now(UTC) + timedelta(hours=23)
    assert ctx.enqueued[-1][0] == ("send_job", str(oid))

    resp = await ctx.client.post(f"/api/outbox/{oid}/cancel", headers=_hdr())
    assert resp.status_code == 200 and resp.json()["draft_id"] == str(ob.draft_id)
    assert (await _outbox(ctx, oid)).state == "cancelled"
    # Back in Drafts, gone from Scheduled.
    drafts = (await ctx.client.get("/api/drafts")).json()
    assert any(d["id"] == str(ob.draft_id) for d in drafts)
    assert (await ctx.client.get("/api/outbox")).json() == []
    # Only a waiting send can be retried or cancelled.
    again = await ctx.client.post(f"/api/outbox/{oid}/retry", headers=_hdr())
    assert again.status_code == 409