import logging import re from dataclasses import dataclass from datetime import datetime, timezone from bson import ObjectId from app.models import MailDomain from app.services.handoff_email_templates import external_reply_subject from app.services.mail_sender import deliver_email, log_outbound_mime_headers from app.services.message_ids import ( build_references_header, canonical_message_id, folded_message_id, first_message_id, internet_message_id_of_message, message_id_fields, ) from app.services.mime_parser import parse_mime from app.services.raw_mime_storage import get_raw_mime_storage from app.services.thread_service import threading_fields_from_parent logger = logging.getLogger(__name__) MESSAGE_SCOPE_EXTERNAL = "EXTERNAL" MESSAGE_SCOPE_HANDOFF = "HANDOFF" class ExternalReplyThreadingError(RuntimeError): pass @dataclass(frozen=True) class ExternalReplyThreading: in_reply_to: str | None references: list[str] parent_internal_message_id: ObjectId replying_to_internet_message_id: str | None @dataclass(frozen=True) class ExternalReplyPlan: subject: str body_text: str body_html: str | None threading: ExternalReplyThreading recipients: list[str] def internet_message_id_of(message: dict) -> str | None: stored = canonical_message_id( message.get("internet_message_id") or message.get("message_id") ) raw_header = (message.get("raw_headers") or {}).get("message_id") raw_canonical = canonical_message_id(raw_header) if raw_canonical and stored: if folded_message_id(raw_canonical) == folded_message_id(stored): return raw_canonical return stored or raw_canonical async def resolve_external_reply_parent( database, *, account_id: str, external_thread_id: ObjectId, customer_address: str, handoff: dict | None = None, ) -> dict | None: """Pick the latest external inbound customer message with a real Message-ID.""" customer = customer_address.lower() pipeline = [ { "$match": { "account_id": account_id, "thread_id": external_thread_id, "message_scope": MESSAGE_SCOPE_EXTERNAL, "deletion_state": {"$in": [None, ""]}, } }, {"$sort": {"event_at": -1, "_id": -1}}, { "$lookup": { "from": "messages", "localField": "message_id", "foreignField": "_id", "as": "message", } }, {"$unwind": "$message"}, { "$match": { "message.message_scope": {"$in": [MESSAGE_SCOPE_EXTERNAL, None]}, "message.direction": "inbound", "$or": [ {"message.internet_message_id": {"$gt": ""}}, {"message.message_id": {"$gt": ""}}, ], } }, {"$limit": 1}, ] cursor = await database.mailbox_entries.aggregate(pipeline) rows = [row async for row in cursor] if rows: return rows[0]["message"] if handoff and handoff.get("trigger_external_message_id"): trigger = await database.messages.find_one( { "_id": handoff["trigger_external_message_id"], "message_scope": MESSAGE_SCOPE_EXTERNAL, } ) if trigger and internet_message_id_of(trigger): return trigger trigger_internet = handoff.get("trigger_internet_message_id") if handoff else None if trigger_internet: folded = folded_message_id(trigger_internet) trigger = await database.messages.find_one( { "message_scope": MESSAGE_SCOPE_EXTERNAL, "message_id_folded": folded, } ) if trigger: return trigger return None def validate_external_reply_parent(parent_external_message: dict) -> None: parent_id = internet_message_id_of(parent_external_message) if not parent_id: raise ExternalReplyThreadingError( "External reply parent is missing internet_message_id" ) scope = parent_external_message.get("message_scope") if scope == MESSAGE_SCOPE_HANDOFF: raise ExternalReplyThreadingError( "External reply parent must not be a HANDOFF-scoped message" ) async def assert_in_reply_to_resolves_to_external_parent( database, in_reply_to: str | None, ) -> dict: """Ensure In-Reply-To points at an EXTERNAL message, never HANDOFF.""" canonical = canonical_message_id(in_reply_to) folded = folded_message_id(canonical) if not canonical or not folded: raise ExternalReplyThreadingError("External reply is missing In-Reply-To") parent = await database.messages.find_one( { "$or": [ {"message_id_folded": folded}, {"message_id": canonical}, {"internet_message_id": canonical}, ], } ) if parent is None: raise ExternalReplyThreadingError( f"In-Reply-To {canonical} does not resolve to a known message" ) scope = parent.get("message_scope") if scope == MESSAGE_SCOPE_HANDOFF: raise ExternalReplyThreadingError( f"In-Reply-To {canonical} resolves to HANDOFF-scoped message" ) if scope not in (MESSAGE_SCOPE_EXTERNAL, None): raise ExternalReplyThreadingError( f"In-Reply-To {canonical} resolves to non-external message scope={scope}" ) return parent async def assert_external_threading_chain( database, *, in_reply_to: str | None, references: list[str] | None, ) -> None: """Ensure outbound headers only reference EXTERNAL messages.""" await assert_in_reply_to_resolves_to_external_parent(database, in_reply_to) folded_targets = { folded_message_id(value) for value in [*(references or [])] if folded_message_id(value) } in_reply_folded = folded_message_id(in_reply_to) if in_reply_folded: folded_targets.discard(in_reply_folded) for folded in folded_targets: handoff_hit = await database.messages.find_one( { "$or": [ {"message_id_folded": folded}, {"message_id": folded}, ], "message_scope": MESSAGE_SCOPE_HANDOFF, }, {"_id": 1}, ) if handoff_hit: raise ExternalReplyThreadingError( "External reply threading references a HANDOFF message" ) def build_external_reply_plan( *, parent_external_message: dict, body_text: str, body_html: str | None = None, subject_override: str | None = None, recipients: list[str], parent_internet_id_override: str | None = None, ) -> ExternalReplyPlan: parent_internet_id = canonical_message_id( parent_internet_id_override ) or internet_message_id_of(parent_external_message) if not parent_internet_id: raise ExternalReplyThreadingError( "External reply parent is missing internet_message_id" ) scope = parent_external_message.get("message_scope") if scope == MESSAGE_SCOPE_HANDOFF: raise ExternalReplyThreadingError( "External reply parent must not be a HANDOFF-scoped message" ) threading_fields = threading_fields_from_parent(parent_external_message) in_reply_to = parent_internet_id or threading_fields.get("in_reply_to") references = threading_fields.get("references", []) if parent_internet_id: references = build_references_header( [ref for ref in references if folded_message_id(ref) != folded_message_id(parent_internet_id)], parent_internet_id, ).split() elif not references and in_reply_to: references = [in_reply_to] if subject_override: subject = subject_override else: parent_subject = parent_external_message.get("subject", "") subject = ( parent_subject if re.match(r"^re:\s*", parent_subject, flags=re.IGNORECASE) else external_reply_subject(parent_subject) ) return ExternalReplyPlan( subject=subject, body_text=body_text, body_html=body_html, threading=ExternalReplyThreading( in_reply_to=in_reply_to, references=references, parent_internal_message_id=parent_external_message["_id"], replying_to_internet_message_id=parent_internet_id, ), recipients=recipients, ) def log_original_external_message( *, internal_id: ObjectId | str, thread_id: ObjectId | str | None, internet_message_id: str | None, sender: str | None, recipient: str | None, subject: str | None, in_reply_to: str | None, references: list[str] | None, raw_message_available: bool, ) -> None: logger.info( "original_external_message internal_id=%s thread_id=%s internet_message_id=%s " "from=%s to=%s subject=%s in_reply_to=%s references=%s raw_message_available=%s", internal_id, thread_id, internet_message_id, sender, recipient, subject, in_reply_to, references, raw_message_available, ) def log_smtp_message_sending( *, generated_message_id: str | None, in_reply_to: str | None, references: list[str] | None, sender: str, recipients: list[str], subject: str, external_thread_id: ObjectId | str | None = None, replying_to_internal_message_id: ObjectId | str | None = None, replying_to_internet_message_id: str | None = None, ) -> None: logger.info( "smtp_message_sending generated_message_id=%s in_reply_to=%s references=%s " "from=%s to=%s subject=%s external_thread_id=%s " "replying_to_internal_message_id=%s replying_to_internet_message_id=%s", generated_message_id, in_reply_to, " ".join(references or []), sender, ",".join(recipients), subject, external_thread_id, replying_to_internal_message_id, replying_to_internet_message_id, ) async def send_external_reply( database, *, account_id: str, sender: str, sender_name: str, external_thread_id: ObjectId, plan: ExternalReplyPlan, domain: MailDomain, resolution_method: str = "in_reply_to", generated_by: str | None = None, generated_from_handoff_message_id: ObjectId | None = None, message_origin: str | None = None, ) -> dict: if generated_from_handoff_message_id: existing = await database.messages.find_one( { "generated_from_handoff_message_id": generated_from_handoff_message_id, "message_scope": MESSAGE_SCOPE_EXTERNAL, } ) if existing: logger.info( "external_reply_idempotent_skip handoff_message=%s external=%s", generated_from_handoff_message_id, existing["_id"], ) return existing await assert_external_threading_chain( database, in_reply_to=plan.threading.in_reply_to, references=plan.threading.references, ) log_smtp_message_sending( generated_message_id=None, in_reply_to=plan.threading.in_reply_to, references=plan.threading.references, sender=sender, recipients=plan.recipients, subject=plan.subject, external_thread_id=external_thread_id, replying_to_internal_message_id=plan.threading.parent_internal_message_id, replying_to_internet_message_id=plan.threading.replying_to_internet_message_id, ) internet_message_id = await deliver_email( sender=sender, recipients=plan.recipients, cc=[], bcc=[], subject=plan.subject, body_text=plan.body_text, body_html=plan.body_html, domain=domain, in_reply_to=plan.threading.in_reply_to, references=plan.threading.references, ) normalized_id = canonical_message_id(internet_message_id) id_fields = message_id_fields(normalized_id) log_smtp_message_sending( generated_message_id=normalized_id, in_reply_to=plan.threading.in_reply_to, references=plan.threading.references, sender=sender, recipients=plan.recipients, subject=plan.subject, external_thread_id=external_thread_id, replying_to_internal_message_id=plan.threading.parent_internal_message_id, replying_to_internet_message_id=plan.threading.replying_to_internet_message_id, ) now = datetime.now(timezone.utc) document = { "direction": "outbound", "message_scope": MESSAGE_SCOPE_EXTERNAL, "account_id": account_id, "sender": {"name": sender_name, "address": sender}, "to": [{"name": value.split("@")[0], "address": value} for value in plan.recipients], "subject": plan.subject, "snippet": re.sub(r"\s+", " ", plan.body_text).strip()[:240], "body_text": plan.body_text, "body_html": plan.body_html, "attachments": [], "created_at": now, "sent_at": now, "status": "accepted", **id_fields, "in_reply_to": plan.threading.in_reply_to, "references": plan.threading.references, "parent_message_id": plan.threading.parent_internal_message_id, "thread_relation": "reply", "thread_resolution_method": resolution_method, } if generated_by: document["generated_by"] = generated_by if generated_from_handoff_message_id: document["generated_from_handoff_message_id"] = generated_from_handoff_message_id if message_origin: document["message_origin"] = message_origin result = await database.messages.insert_one(document) document["_id"] = result.inserted_id from app.services.thread_service import attach_message_to_thread await attach_message_to_thread( database, account_id=account_id, thread_id=external_thread_id, message_id=result.inserted_id, subject=plan.subject, event_at=now, ) await database.mailbox_entries.insert_one( { "account_id": account_id, "message_id": result.inserted_id, "thread_id": external_thread_id, "mailbox": "sent", "message_scope": MESSAGE_SCOPE_EXTERNAL, "direction": "outbound", "sender": document["sender"], "recipients": document["to"], "subject": plan.subject, "snippet": document["snippet"], "attachments": [], "status": "accepted", "is_read": True, "is_starred": False, "event_at": now, "created_at": now, "trashed_at": None, "deletion_state": None, } ) return document