"""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 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 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 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 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)) 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