webmail-client/backend/app/outbox/jobs.py

"""The scheduled send job (SPEC §6.5).

Fires ``UNDO_SECONDS`` after the user hits send. If the outbox row was cancelled
in the meantime, it does nothing (no SMTP traffic). Otherwise: SMTP send, append
to Sent (skipped for Gmail, which self-appends), harvest recipients into
``known_recipients`` (weight 3), and publish ``message.sent``.
"""

from __future__ import annotations

import logging
import uuid
from datetime import UTC, datetime, timedelta
from email.utils import getaddresses
from typing import Any

from sqlalchemy import select, update
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.ext.asyncio import AsyncSession

from app.db import session_scope
from app.events import get_event_bus
from app.indexer import blobstore
from app.indexer.indexer import persist_message
from app.models import Account, Draft, KnownRecipient, Mailbox, Message, Outbox
from app.outbox.attachments import load_attachments
from app.outbox.mime_build import build_mime
from app.outbox.service import delete_mirrored_draft_row
from app.providers.base import RemoteMessage
from app.queue import enqueue_job
from app.search.factory import get_search_backend
from app.sync.actions import _retry_later
from app.sync.health import is_temporary
from app.sync.provider_factory import build_provider

logger = logging.getLogger(__name__)


def _is_gmail(account: Account) -> bool:
    host = str(account.imap_config.get("host", "")).lower()
    return "gmail.com" in host or "googlemail.com" in host


def _addr_dicts(values: object) -> list[dict[str, str]]:
    """Parse a draft payload's ``["Name <email>", …]`` list into name+email dicts."""
    if not isinstance(values, list):
        return []
    return [
        {"name": name, "email": email}
        for name, email in getaddresses([str(v) for v in values])
        if email
    ]


async def _harvest_recipients(session: AsyncSession, user_id: uuid.UUID, emails: list[str]) -> None:
    now = datetime.now(UTC)
    for email in {e.lower() for e in emails if e}:
        stmt = pg_insert(KnownRecipient).values(
            user_id=user_id, email=email, name="", times_used=3, last_used_at=now
        )
        stmt = stmt.on_conflict_do_update(
            index_elements=[KnownRecipient.user_id, KnownRecipient.email],
            set_={
                "times_used": KnownRecipient.times_used + 3,  # sent-to weight (SPEC §4)
                "last_used_at": stmt.excluded.last_used_at,
            },
        )
        await session.execute(stmt)


async def _persist_sent_copy(user_id: uuid.UUID, account: Account, raw: bytes, uid: int) -> None:
    """Store the just-sent message in the DB (Sent mailbox) and notify the UI.

    No-op without a UIDPLUS append UID (uid <= 0); in that case a future Sent sync
    would pick it up. persist_message dedupes on (mailbox_id, uid).
    """
    if uid <= 0:
        return
    async with session_scope() as session:
        sent_mb = await session.scalar(
            select(Mailbox).where(Mailbox.account_id == account.id, Mailbox.special_use == "sent")
        )
        if sent_mb is None:
            return
        remote = RemoteMessage(
            uid=uid, raw=raw, flags=frozenset({"\\Seen"}), internal_date=datetime.now(UTC)
        )
        result = await persist_message(
            session,
            user_id=user_id,
            account_id=account.id,
            mailbox_id=sent_mb.id,
            mailbox_special_use="sent",
            remote=remote,
        )
        if result is None:
            return
        message_id = result.message.id
        thread_id = result.message.thread_id
        search_doc = result.search_doc

    await get_search_backend().index(search_doc)
    bus = get_event_bus()
    await bus.publish(
        "message.new",
        {
            "user_id": str(user_id),
            "account_id": str(account.id),
            "message_id": str(message_id),
            "thread_id": str(thread_id) if thread_id else None,
        },
    )
    if thread_id is not None:
        await bus.publish("thread.updated", {"user_id": str(user_id), "thread_id": str(thread_id)})


# A send the mail server could not take (unreachable, a temporary 4xx) is tried
# again after these gaps, then hourly, until RETRY_WINDOW after the first
# failure. Meanwhile the message waits in the Scheduled folder.
_RETRY_GAPS_S = (60, 120, 300, 600, 1800)
_RETRY_HOURLY_S = 3600
RETRY_WINDOW = timedelta(hours=24)


def _retry_gap(attempts: int) -> int:
    return _RETRY_GAPS_S[attempts - 1] if attempts <= len(_RETRY_GAPS_S) else _RETRY_HOURLY_S


async def _send_failed(
    oid: uuid.UUID, user_id: uuid.UUID, subject: str, exc: BaseException
) -> None:
    """Record a failed send: "retrying" if it may work later, else "failed"."""
    now = datetime.now(UTC)
    async with session_scope() as session:
        ob = await session.get(Outbox, oid)
        if ob is None:
            return
        ob.attempts += 1
        ob.error = (getattr(exc, "detail", None) or str(exc) or type(exc).__name__)[:500]
        ob.retry_until = ob.retry_until or now + RETRY_WINDOW
        if is_temporary(exc) and now < ob.retry_until:
            ob.state = "retrying"
            ob.next_attempt_at = now + timedelta(seconds=_retry_gap(ob.attempts))
        else:
            ob.state = "failed"
            ob.next_attempt_at = None
        state, attempts, next_at, error = ob.state, ob.attempts, ob.next_attempt_at, ob.error
    if state == "retrying" and next_at is not None:
        await enqueue_job(
            "send_job", str(oid), _defer_until=next_at, _job_id=f"send:{oid}:{attempts}"
        )
    await get_event_bus().publish(
        "outbox.state_changed",
        {
            "user_id": str(user_id),
            "outbox_id": str(oid),
            "state": state,
            "attempts": attempts,
            "subject": subject,
            "error": error,
            "next_attempt_at": next_at.isoformat() if next_at else None,
        },
    )
    if state == "failed":
        logger.warning("send failed for outbox %s: %s", oid, error)
    else:
        logger.info(
            "send of outbox %s failed (try %d), retrying at %s: %s", oid, attempts, next_at, error
        )


async def file_sent_copy_job(ctx: dict[str, Any], account_id: str, user_id: str, sha: str) -> None:
    """File a sent message's copy in Sent, after the send itself succeeded but
    the IMAP server could not be reached. Retried (arq Retry) for ~24 hours."""
    uid = uuid.UUID(user_id)
    async with session_scope() as session:
        account = await session.get(Account, uuid.UUID(account_id))
        if account is None:
            return
        sent_path = await session.scalar(
            select(Mailbox.path).where(
                Mailbox.account_id == account.id, Mailbox.special_use == "sent"
            )
        )
    if not sent_path:
        return
    raw = blobstore.load(uid, sha)
    provider = build_provider(account)
    try:
        appended_uid = await provider.append(sent_path, raw, frozenset({"\\Seen"}))
    except Exception as exc:
        retry = _retry_later(ctx, exc)
        if retry is not None:
            raise retry from exc
        logger.error("could not file a sent message in Sent (blob %s): %s", sha, exc)
        return
    finally:
        await provider.close()
    await _persist_sent_copy(uid, account, raw, appended_uid)
    logger.info("filed a sent message in Sent for account %s", account_id)


async def send_job(ctx: dict[str, Any], outbox_id: str) -> None:
    """Send a queued message unless it was cancelled during the undo window.

    Also runs each retry of a send that failed for a temporary reason. The row is
    claimed with one conditional UPDATE, so two jobs for the same send (a planned
    retry and "Retry now") can never both send it.
    """
    oid = uuid.UUID(outbox_id)
    async with session_scope() as session:
        claimed = await session.scalar(
            update(Outbox)
            .where(Outbox.id == oid, Outbox.state.in_(("undo_window", "retrying")))
            .values(state="sending")
            .returning(Outbox.id)
        )
        if claimed is None:
            return  # cancelled, or already handled -> no SMTP traffic
        outbox = await session.get(Outbox, oid)
        assert outbox is not None
        draft = await session.get(Draft, outbox.draft_id)
        if draft is None:
            outbox.state = "failed"
            outbox.error = "draft missing"
            return
        account = await session.get(Account, draft.account_id)
        reply_to = (
            await session.get(Message, draft.reply_to_message_id)
            if draft.reply_to_message_id
            else None
        )
        if account is None:
            outbox.state = "failed"
            outbox.error = "account missing"
            return
        user_id = outbox.user_id
        attachments = await load_attachments(session, user_id, draft)
        transmit_raw, sent_raw, envelope_from, envelope_to = build_mime(
            account, draft, reply_to, attachments
        )
        sent_path = await session.scalar(
            select(Mailbox.path).where(
                Mailbox.account_id == account.id, Mailbox.special_use == "sent"
            )
        )
        drafts_path = await session.scalar(
            select(Mailbox.path).where(
                Mailbox.account_id == account.id, Mailbox.special_use == "drafts"
            )
        )
        draft_imap_uid = draft.imap_uid
        gmail = _is_gmail(account)
        account_obj = account

    subject = str(draft.payload.get("subject") or "")
    provider = build_provider(account_obj)
    appended_uid = 0
    try:
        try:
            await provider.send(transmit_raw, envelope_from, envelope_to)
        except Exception as exc:  # noqa: BLE001 - retry or record; the draft is kept
            await _send_failed(oid, user_id, subject, exc)
            return
        # Delivered. If filing the copy in Sent fails (the IMAP server is not
        # there right now), the send still succeeded — and the copy must not be
        # lost: it is kept and filed later by file_sent_copy_job.
        if sent_path and not gmail:
            try:
                appended_uid = await provider.append(sent_path, sent_raw, frozenset({"\\Seen"}))
            except Exception as exc:  # noqa: BLE001
                logger.warning("outbox %s sent; filing it in Sent failed, will retry: %s", oid, exc)
                sha = blobstore.store(user_id, sent_raw)
                await enqueue_job("file_sent_copy_job", str(account_obj.id), str(user_id), sha)
        # Remove the mirrored Drafts-folder copy now that it has been sent.
        if drafts_path and draft_imap_uid:
            try:
                await provider.delete_messages(drafts_path, [draft_imap_uid])
            except Exception as exc:  # noqa: BLE001
                logger.warning("outbox %s sent; removing its Drafts copy failed: %s", oid, exc)
    finally:
        await provider.close()

    # Persist the sent copy (with Bcc) ourselves so it shows in Sent and threads
    # immediately; the periodic Sent poll would also catch it (dedup by UID).
    await _persist_sent_copy(user_id, account_obj, sent_raw, appended_uid)

    async with session_scope() as session:
        ob = await session.get(Outbox, oid)
        if ob is not None:
            ob.state = "sent"
        # The message left successfully: drop its draft so it no longer shows in
        # the Drafts folder (undo/cancel returns earlier, keeping the draft).
        draft_row = await session.get(Draft, draft.id)
        if draft_row is not None:
            await session.delete(draft_row)
        # Also drop the mirrored Drafts-folder copy's DB row (we expunged it on the
        # server above) so it doesn't resurface as a "remote" draft.
        if draft_imap_uid:
            await delete_mirrored_draft_row(
                session, account_id=account_obj.id, imap_uid=draft_imap_uid
            )
        await _harvest_recipients(session, user_id, envelope_to)

    bus = get_event_bus()
    await bus.publish(
        "message.sent",
        {
            "user_id": str(user_id),
            "outbox_id": str(oid),
            "thread_id": str(draft.thread_id),
            # Recipients (name+email) so consumers (e.g. contacts autosave) don't
            # need to read the outbox row. payload to/cc are "Name <email>" strings.
            "to": _addr_dicts(draft.payload.get("to")),
            "cc": _addr_dicts(draft.payload.get("cc")),
        },
    )
    await bus.publish(
        "outbox.state_changed",
        {"user_id": str(user_id), "outbox_id": str(oid), "state": "sent"},
    )