webmail-client/backend/app/plugins/actions.py

"""Message read + flag helpers exposed to plugins via the SDK (SPEC §11).

Plugins like ``filters`` need to inspect a newly-arrived message and flag it.
Both operations are user-scoped: a plugin can only read/act on the given user's
own message (the same app-level ``user_id`` scoping core uses — core tables have
no RLS). Flagging mirrors the core action path: update the DB, enqueue the IMAP
push, and publish ``flags.changed`` so the UI updates.

Actions exposed: flags, trash (whole thread), and ``move_to_special`` for a
single message — the last added for the antispam plugin, which must file one
spam message into Junk without dragging its thread along.
"""

from __future__ import annotations

import datetime as dt
import uuid
from collections.abc import Iterable
from email import message_from_bytes
from typing import Any

from sqlalchemy import select

from app.db import session_scope
from app.events import get_event_bus
from app.indexer import blobstore
from app.models import Account, Mailbox, Message, Thread


class PluginMessages:
    """Read-only, user-scoped view of a message's matchable fields."""

    async def header(self, user_id: uuid.UUID, message_id: uuid.UUID, name: str) -> str | None:
        """Return a raw header (e.g. ``List-Unsubscribe``) from the user's message,
        or None if the message/header/blob is missing. Parses the stored RFC822."""
        async with session_scope() as session:
            msg = await session.scalar(
                select(Message).where(Message.id == message_id, Message.user_id == user_id)
            )
            if msg is None or not msg.blob_sha256:
                return None
            sha = msg.blob_sha256
        try:
            raw = blobstore.load(user_id, sha)
        except FileNotFoundError:
            return None
        return message_from_bytes(raw).get(name)

    async def raw(self, user_id: uuid.UUID, message_id: uuid.UUID) -> bytes | None:
        """Return the stored RFC822 bytes of the user's message, or None.

        For engines that need the whole message rather than one header — spam
        scoring wants headers *and* body. None when the message isn't the user's,
        has no blob, or the blob is unreadable; callers must treat that as "cannot
        classify", never as an error.
        """
        async with session_scope() as session:
            msg = await session.scalar(
                select(Message).where(Message.id == message_id, Message.user_id == user_id)
            )
            if msg is None or not msg.blob_sha256:
                return None
            sha = msg.blob_sha256
        try:
            return blobstore.load(user_id, sha)
        except FileNotFoundError:
            return None

    async def sample(
        self,
        user_id: uuid.UUID,
        *,
        account_id: uuid.UUID,
        special_use: str,
        limit: int = 500,
        older_than_days: int | None = None,
    ) -> list[uuid.UUID]:
        """Message ids from one account folder, newest first, capped at ``limit``.

        For building a training corpus out of mail the user already has — the
        provider's own Junk folder is a labelled spam set, and aged Inbox mail
        that was never moved is a reasonable ham set. ``older_than_days`` skips
        recent arrivals, which are the ones a user has not triaged yet.

        Ids only: the caller fetches bodies through :meth:`raw` one at a time,
        so a 500-message sample never materialises 500 message bodies at once.
        """
        stmt = (
            select(Message.id)
            .join(Mailbox, Mailbox.id == Message.mailbox_id)
            .where(
                Message.user_id == user_id,
                Message.account_id == account_id,
                Mailbox.special_use == special_use,
            )
            .order_by(Message.internal_date.desc())
            .limit(max(0, min(limit, 5000)))
        )
        if older_than_days is not None:
            cutoff = dt.datetime.now(dt.UTC) - dt.timedelta(days=older_than_days)
            stmt = stmt.where(Message.internal_date < cutoff)
        async with session_scope() as session:
            return list(await session.scalars(stmt))

    async def threads_of(
        self, user_id: uuid.UUID, message_ids: Iterable[str]
    ) -> dict[str, str | None]:
        """Map still-existing message ids to their current thread id (None if
        unthreaded). Deleted messages are omitted, so a plugin can prune stale
        per-message references and re-home labels onto the live thread after a
        JWZ thread merge."""
        ids: list[uuid.UUID] = []
        for mid in message_ids:
            try:
                ids.append(uuid.UUID(str(mid)))
            except (ValueError, AttributeError, TypeError):
                continue
        if not ids:
            return {}
        async with session_scope() as session:
            rows = await session.execute(
                select(Message.id, Message.thread_id).where(
                    Message.user_id == user_id, Message.id.in_(ids)
                )
            )
            return {str(mid): (str(tid) if tid else None) for mid, tid in rows}

    async def envelope(self, user_id: uuid.UUID, message_id: uuid.UUID) -> dict[str, Any] | None:
        async with session_scope() as session:
            row = (
                await session.execute(
                    select(Message, Mailbox.special_use, Account.state)
                    .join(Mailbox, Mailbox.id == Message.mailbox_id, isouter=True)
                    .join(Account, Account.id == Message.account_id, isouter=True)
                    .where(Message.id == message_id, Message.user_id == user_id)
                )
            ).first()
            if row is None:
                return None
            msg, special_use, account_state = row
            sender = msg.from_addr[0] if msg.from_addr else {}
            return {
                "message_id": str(msg.id),
                "account_id": str(msg.account_id),
                # Where it landed: "inbox", "junk", "sent", … or None for a folder
                # with no SPECIAL-USE. A plugin that should only act on newly
                # arrived mail checks this — `message.new` itself does not say.
                "mailbox_special_use": special_use,
                # "backfilling" | "live" | … — `message.new` fires for imported
                # history too, so a plugin that should only react to genuinely
                # new arrivals has to be able to tell the difference.
                "account_state": account_state,
                "subject": msg.subject or "",
                "from_email": (sender.get("email") or "").lower(),
                "from_name": sender.get("name") or "",
                # Full recipient list (multi-account setups match on "to").
                "to": [
                    {"email": (a.get("email") or "").lower(), "name": a.get("name") or ""}
                    for a in msg.to_addr
                ],
                "flags": list(msg.flags),
            }


class PluginAccounts:
    """Cross-user account queries, for scheduled jobs.

    Deliberately not user-scoped: a sweep runs on behalf of the instance, not a
    request, and needs to find work across every user. Nothing here returns
    content — only ids and state — so a plugin still has to go through the
    user-scoped accessors to read anything.
    """

    async def live(self) -> list[tuple[uuid.UUID, uuid.UUID]]:
        """``(user_id, account_id)`` for every account finished with its initial
        import. Accounts still backfilling are excluded: acting on one means
        acting on a partial mailbox."""
        async with session_scope() as session:
            rows = await session.execute(
                select(Account.user_id, Account.id).where(Account.state == "live")
            )
            return [(u, a) for u, a in rows]


class PluginThreads:
    """Read-only, user-scoped thread queries (e.g. self-healing stale references)."""

    async def existing(self, user_id: uuid.UUID, thread_ids: Iterable[str]) -> set[str]:
        """Return the subset of ``thread_ids`` that still exist for the user.

        Lets a plugin prune references to threads deleted out from under it (e.g. an
        account removal drops its threads, orphaning label aggregates keyed by id)."""
        ids: list[uuid.UUID] = []
        for tid in thread_ids:
            try:
                ids.append(uuid.UUID(str(tid)))
            except (ValueError, AttributeError, TypeError):
                continue
        if not ids:
            return set()
        async with session_scope() as session:
            rows = await session.scalars(
                select(Thread.id).where(Thread.user_id == user_id, Thread.id.in_(ids))
            )
            return {str(r) for r in rows}


class PluginActions:
    """User-scoped actions a plugin can take on a message."""

    async def set_flags(
        self,
        user_id: uuid.UUID,
        message_id: uuid.UUID,
        *,
        add: Iterable[str] = (),
        remove: Iterable[str] = (),
    ) -> bool:
        """Add/remove IMAP flags on the user's message (mirrors the core path).

        Only user-settable flags survive (\\Seen/\\Flagged/\\Answered). Returns
        False if nothing owned matches or nothing would change."""
        from app.queue import enqueue_job
        from app.sync.actions import apply_flags_to_messages, sanitize_flags

        add_f, remove_f = sanitize_flags(add), sanitize_flags(remove)
        if not add_f and not remove_f:
            return False
        async with session_scope() as session:
            msg = await session.scalar(
                select(Message).where(Message.id == message_id, Message.user_id == user_id)
            )
            if msg is None:
                return False
            await apply_flags_to_messages(session, [msg], add_f, remove_f)
            flags = list(msg.flags)
        await enqueue_job("push_flags_job", str(message_id), sorted(add_f), sorted(remove_f))
        await get_event_bus().publish(
            "flags.changed",
            {"user_id": str(user_id), "message_id": str(message_id), "flags": flags},
        )
        return True

    async def move_to_special(
        self, user_id: uuid.UUID, message_id: uuid.UUID, special_use: str
    ) -> bool:
        """Move ONE message into the account's folder with this ``special_use``
        (``junk``, ``inbox``, …), on the same path as the UI's move: optimistic DB
        update, then ``move_job`` performs the IMAP MOVE and reconciles the uid.

        Per-message on purpose. ``trash`` moves a whole thread, which is right for
        Delete and wrong for a spam verdict — one spam reply must not drag a
        legitimate conversation into Junk.

        Returns False when the message isn't the user's, the account has no such
        folder, or it is already there. That is "could not act", not an error:
        an account whose server has no Junk folder must still receive its mail.
        """
        from app.queue import enqueue_job

        async with session_scope() as session:
            msg = await session.scalar(
                select(Message).where(Message.id == message_id, Message.user_id == user_id)
            )
            if msg is None:
                return False
            dst = await session.scalar(
                select(Mailbox).where(
                    Mailbox.user_id == user_id,
                    Mailbox.account_id == msg.account_id,  # moves stay within an account
                    Mailbox.special_use == special_use,
                )
            )
            if dst is None or dst.id == msg.mailbox_id:
                return False
            dst_id, src_id = dst.id, msg.mailbox_id
            # The SOURCE folder's special-use, not the destination parameter:
            # consumers of message.changed read the pair as a transition, and
            # reporting them equal would erase the move.
            src_special = await session.scalar(
                select(Mailbox.special_use).where(Mailbox.id == src_id)
            )
            thread_id, dst_special = msg.thread_id, dst.special_use
            msg.mailbox_id = dst_id  # optimistic, as the core endpoint does
            await session.commit()

        await enqueue_job("move_job", str(message_id), str(dst_id), str(src_id))
        # Announce the optimistic state now rather than waiting for the IMAP move
        # to finish. The core's own move endpoint gets this for free — its caller
        # updates the UI from the response — but a plugin acting on the user's
        # behalf has no client-side handle, so without this the message lingers on
        # screen until the job lands. move_job publishes again when it does.
        await get_event_bus().publish(
            "message.changed",
            {
                "user_id": str(user_id),
                "message_id": str(message_id),
                "thread_id": str(thread_id) if thread_id else None,
                "mailbox_id": str(dst_id),
                "previous_mailbox_id": str(src_id),
                "special_use": dst_special,
                "previous_special_use": src_special,
            },
        )
        return True

    async def trash(self, user_id: uuid.UUID, message_id: uuid.UUID) -> bool:
        """Move the message's thread to Trash — the same path as the UI Delete
        (move_thread_to_folder_job handles the IMAP move, restore state, folder
        creation and events). Returns False if the message isn't the user's."""
        from app.queue import enqueue_job

        async with session_scope() as session:
            msg = await session.scalar(
                select(Message).where(Message.id == message_id, Message.user_id == user_id)
            )
            if msg is None or msg.thread_id is None:
                return False
            thread_id = msg.thread_id
        await enqueue_job("move_thread_to_folder_job", str(thread_id), "trash")
        return True