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 contextlib
import logging
import uuid
from datetime import UTC, datetime
from email.utils import getaddresses
from typing import Any

from sqlalchemy import select
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.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.search.factory import get_search_backend
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)})


async def send_job(ctx: dict[str, Any], outbox_id: str) -> None:
    """Send a queued message unless it was cancelled during the undo window."""
    oid = uuid.UUID(outbox_id)
    async with session_scope() as session:
        outbox = await session.get(Outbox, oid)
        if outbox is None or outbox.state != "undo_window":
            return  # cancelled, or already handled -> no SMTP traffic
        outbox.state = "sending"
        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

    provider = build_provider(account_obj)
    appended_uid = 0
    try:
        await provider.send(transmit_raw, envelope_from, envelope_to)
        if sent_path and not gmail:
            appended_uid = await provider.append(sent_path, sent_raw, frozenset({"\\Seen"}))
        # Remove the mirrored Drafts-folder copy now that it has been sent.
        if drafts_path and draft_imap_uid:
            with contextlib.suppress(Exception):
                await provider.delete_messages(drafts_path, [draft_imap_uid])
    except Exception as exc:  # noqa: BLE001 - record and notify, keep the draft
        async with session_scope() as session:
            ob = await session.get(Outbox, oid)
            if ob is not None:
                ob.state = "failed"
                ob.error = str(exc)[:500]
        await get_event_bus().publish(
            "outbox.state_changed",
            {"user_id": str(user_id), "outbox_id": str(oid), "state": "failed"},
        )
        logger.exception("send failed for outbox %s", oid)
        return
    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"},
    )