"""Indexer orchestration (SPEC §6.4). Turns a :class:`RemoteMessage` into persisted rows: raw blob (deduped), parsed MIME, sanitized HTML, ``messages``/``body_parts``/``attachments`` rows, JWZ threading, and a ``known_recipients`` upsert. Returns a :class:`SearchDoc` for the caller to index *after commit* (FR-30 inline indexing) and to drive the ``message.new`` event. Idempotent on ``(mailbox_id, uid)``. """ from __future__ import annotations import uuid from dataclasses import dataclass from datetime import UTC, datetime from sqlalchemy import select from sqlalchemy.dialects.postgresql import insert as pg_insert from sqlalchemy.ext.asyncio import AsyncSession from app.indexer import blobstore from app.indexer.mime import ParsedEmail, parse_message from app.indexer.sanitize import sanitize_html from app.models import Attachment, BodyPart, KnownRecipient, Message from app.providers.base import RemoteMessage from app.search.base import SearchDoc from app.threading_.normalize import synthesize_message_id from app.threading_.threader import thread_message _NO_RECIPIENT_HARVEST = {"junk", "sent", "drafts", "trash"} @dataclass class PersistResult: message: Message search_doc: SearchDoc has_remote_images: bool def _epoch(value: datetime) -> int: return int(value.timestamp()) def _build_search_doc(message: Message, parsed: ParsedEmail, special_use: str | None) -> SearchDoc: return SearchDoc( message_id=message.id, thread_id=message.thread_id or message.id, user_id=message.user_id, account_id=message.account_id, mailbox_special_use=special_use, from_=parsed.from_addr, to=parsed.to_addr, cc=parsed.cc_addr, subject=parsed.subject, body_text=parsed.text_body, attachment_names=[a.filename for a in parsed.attachments], has_attachments=parsed.has_attachments, is_unread="\\Seen" not in message.flags, is_starred="\\Flagged" in message.flags, date=_epoch(message.internal_date), size_bytes=message.size_bytes, ) async def _harvest_recipient(session: AsyncSession, message: Message, parsed: ParsedEmail) -> None: """Upsert the From address into known_recipients (weight 1) (SPEC §6.4).""" if not parsed.from_addr: return sender = parsed.from_addr[0] email = (sender.get("email") or "").lower() if not email: return stmt = pg_insert(KnownRecipient).values( user_id=message.user_id, email=email, name=sender.get("name", ""), times_used=1, last_used_at=message.internal_date, ) stmt = stmt.on_conflict_do_update( index_elements=[KnownRecipient.user_id, KnownRecipient.email], set_={ "times_used": KnownRecipient.times_used + 1, "last_used_at": stmt.excluded.last_used_at, "name": stmt.excluded.name, }, ) await session.execute(stmt) async def _resolve_message_id( session: AsyncSession, *, user_id: uuid.UUID, account_id: uuid.UUID, parsed_id: str | None, raw: bytes, ) -> str: """Resolve the Message-ID to store, applying SPEC §7.1 duplicate handling. Missing id -> synthesize. A *duplicate* id already seen in a **different account** is a separate delivery (e.g. a bulk sender that reuses one Message-ID for every recipient) -> synthesize a non-authoritative id so the two do not merge. A same-account duplicate (e.g. Gmail INBOX + All Mail holding one message) keeps the real id so the copies stay in one thread. """ if parsed_id is None: return synthesize_message_id(raw) other_account = await session.scalar( select(Message.account_id) .where(Message.user_id == user_id, Message.message_id_hdr == parsed_id) .limit(1) ) if other_account is not None and other_account != account_id: return synthesize_message_id(raw) return parsed_id async def persist_message( session: AsyncSession, *, user_id: uuid.UUID, account_id: uuid.UUID, mailbox_id: uuid.UUID, mailbox_special_use: str | None, remote: RemoteMessage, ) -> PersistResult | None: """Persist one message. Returns None if it already exists (idempotent).""" existing = await session.scalar( select(Message.id).where(Message.mailbox_id == mailbox_id, Message.uid == remote.uid) ) if existing is not None: return None sha = blobstore.store(user_id, remote.raw) parsed = parse_message(remote.raw) message_id_hdr = await _resolve_message_id( session, user_id=user_id, account_id=account_id, parsed_id=parsed.message_id, raw=remote.raw ) message = Message( id=uuid.uuid4(), user_id=user_id, account_id=account_id, mailbox_id=mailbox_id, uid=remote.uid, message_id_hdr=message_id_hdr, in_reply_to=parsed.in_reply_to, references_hdr=parsed.references, subject=parsed.subject, from_addr=parsed.from_addr, to_addr=parsed.to_addr, cc_addr=parsed.cc_addr, bcc_addr=parsed.bcc_addr, date_hdr=parsed.date, internal_date=remote.internal_date or datetime.now(UTC), flags=sorted(remote.flags), size_bytes=len(remote.raw), has_attachments=parsed.has_attachments, snippet=parsed.snippet, blob_sha256=sha, ) session.add(message) await session.flush() # Body parts: plain text (for indexing) + sanitized HTML (for display). has_remote_images = False if parsed.text_body: session.add(BodyPart(message_id=message.id, kind="text", content=parsed.text_body)) if parsed.html_body is not None: result = sanitize_html(parsed.html_body, message.id) has_remote_images = result.has_remote_images session.add( BodyPart( message_id=message.id, kind="html_sanitized", content=result.html, sanitizer_v=result.sanitizer_v, ) ) for att in parsed.attachments: att_sha = blobstore.store(user_id, att.payload) session.add( Attachment( message_id=message.id, user_id=user_id, filename=att.filename, mime_type=att.mime_type, size_bytes=len(att.payload), content_id=att.content_id, blob_sha256=att_sha, ) ) message.has_remote_images = has_remote_images # Drafts are not conversation messages: keep them out of threads so they don't # pollute the reading pane or thread list. They surface only via the drafts UI. if mailbox_special_use != "drafts": await thread_message(session, message) if mailbox_special_use not in _NO_RECIPIENT_HARVEST: await _harvest_recipient(session, message, parsed) await session.flush() doc = _build_search_doc(message, parsed, mailbox_special_use) return PersistResult(message=message, search_doc=doc, has_remote_images=has_remote_images)