lukasandClaude Opus 5
dee1983b84
message.new fires for imported history as well as new arrivals, and EventBus awaits hooks inline. As written, adding an account would have scored every message in its backfill — filing years-old inbox mail into Junk on the server, and adding ~70 ms of serialized HTTP per message to the backfill loop (twelve minutes on a 10k-message account). It is also when the filter is least able to judge, because Bayes has learned nothing yet. The envelope now carries account_state, and the hook returns unless it is "live". The backfill window belongs to the C2 learning bootstrap (§C.5), which reads that history as training data instead of acting on it. Tests pin the gating: backfilling accounts are not even read from the blob store, non-inbox folders are ignored, a live inbox arrival still classifies, and an unreachable engine still delivers. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_018SqJYt7jD7oUEkoY68sBZo
241 lines
10 KiB
Python
241 lines
10 KiB
Python
"""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
|