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.

(Only flag actions are exposed for now; a ``move`` action would follow the same
wrap-the-core-path pattern.)
"""

from __future__ import annotations

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 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 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:
            msg = await session.scalar(
                select(Message).where(Message.id == message_id, Message.user_id == user_id)
            )
            if msg is None:
                return None
            sender = msg.from_addr[0] if msg.from_addr else {}
            return {
                "message_id": str(msg.id),
                "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 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 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