import base64 import json import re from datetime import date, datetime, time, timedelta, timezone from typing import Annotated from bson import ObjectId from fastapi import APIRouter, Depends, HTTPException, Query, status from sqlalchemy.ext.asyncio import AsyncSession from app.database import get_mongo_database, get_postgres_session from app.models import Account from app.routers.auth import get_current_account from app.schemas import ( EmailCreate, EmailRead, MailAddress, MessagePage, MessageRead, MessageReadState, MessageStarState, MessageSummary, utc_now, ) from app.services.domain_provisioning import get_domain_by_name from app.services.mail_sender import deliver_email from app.services.message_ids import normalize_message_id from app.services.mime_parser import parse_mime from app.services.raw_mime_storage import get_raw_mime_storage from app.services.reply_recipients import reply_recipients from app.services.thread_service import ( create_thread, ensure_thread_for_message, message_event_at, threading_fields_from_parent, ) from app.services.mailbox_scope import apply_external_scope from app.services.thread_resolver import ThreadResolution MESSAGE_SCOPE_EXTERNAL = "EXTERNAL" router = APIRouter(prefix="/emails", tags=["emails"]) def serialize_email(document: dict, viewer_email: str) -> EmailRead: return EmailRead( id=str(document["_id"]), sender=document["sender"], recipients=document["recipients"], cc=document.get("cc", []), bcc=document.get("bcc", []) if document["sender"] == viewer_email else [], subject=document["subject"], body_text=document["body_text"], body_html=document.get("body_html"), status=document["status"], created_at=document["created_at"], ) def sender_domain(address: str) -> str: return address.split("@", 1)[1].lower() def address(value: str, name: str | None = None) -> dict[str, str]: return {"name": name or value, "address": value.lower()} def encode_cursor(entry: dict) -> str: value = json.dumps( {"created_at": entry["created_at"].isoformat(), "id": str(entry["_id"])}, separators=(",", ":"), ) return base64.urlsafe_b64encode(value.encode()).decode().rstrip("=") def decode_cursor(value: str) -> tuple[datetime, ObjectId]: try: padded = value + "=" * (-len(value) % 4) payload = json.loads(base64.urlsafe_b64decode(padded).decode()) return datetime.fromisoformat(payload["created_at"]), ObjectId(payload["id"]) except (ValueError, TypeError, KeyError, json.JSONDecodeError) as error: raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Invalid cursor.", ) from error def summarize(message: dict, entry: dict) -> MessageSummary: sender = message.get("sender", {"name": "Unknown sender", "address": ""}) recipients = message.get("to") or [ address(value) for value in message.get("recipients", []) ] mailbox = "trash" if entry.get("trashed_at") else entry["mailbox"] return MessageSummary( id=str(message["_id"]), mailbox=mailbox, direction=message.get("direction", "outbound"), sender=MailAddress(**sender), recipients=[MailAddress(**item) for item in recipients], subject=message.get("subject", ""), snippet=message.get("snippet", message.get("body_text", "")[:240]), status=message.get("status", "received"), created_at=message.get("received_at") or message["created_at"], is_read=entry.get("is_read", True), is_starred=entry.get("is_starred", False), trashed_at=entry.get("trashed_at"), attachments=message.get("attachments", []), thread_id=str(entry["thread_id"]) if entry.get("thread_id") else None, ) async def legacy_email(database, email_id: ObjectId, viewer_email: str) -> dict | None: document = await database.emails.find_one({"_id": email_id}) if document is None: return None visible_to = { document["sender"], *document.get("recipients", []), *document.get("cc", []), *document.get("bcc", []), } return document if viewer_email in visible_to else None @router.post("", response_model=EmailRead, status_code=status.HTTP_201_CREATED) async def create_email( payload: EmailCreate, database=Depends(get_mongo_database), session: AsyncSession = Depends(get_postgres_session), account: Account = Depends(get_current_account), ): sender = account.email.lower() parent_message = None thread_id = None threading_fields: dict = {} if payload.reply_to_message_id: if not ObjectId.is_valid(payload.reply_to_message_id): raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Invalid reply target.", ) parent_id = ObjectId(payload.reply_to_message_id) parent_entry = await database.mailbox_entries.find_one( {"account_id": str(account.id), "message_id": parent_id} ) parent_message = await database.messages.find_one({"_id": parent_id}) if parent_message is None or parent_entry is None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Reply target not found.", ) thread_id = parent_entry.get("thread_id") threading_fields = threading_fields_from_parent(parent_message) to_addresses, cc_addresses = reply_recipients( message=parent_message, viewer_email=sender, reply_all=payload.reply_all, ) recipients = to_addresses or [str(recipient).lower() for recipient in payload.recipients] cc = cc_addresses or [str(recipient).lower() for recipient in payload.cc] else: recipients = [str(recipient).lower() for recipient in payload.recipients] cc = [str(recipient).lower() for recipient in payload.cc] bcc = [str(recipient).lower() for recipient in payload.bcc] all_recipients = [*recipients, *cc, *bcc] if not all_recipients: raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="At least one recipient is required.", ) if len(set(all_recipients)) != len(all_recipients): raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Each recipient may only appear once across To, Cc, and Bcc.", ) domain = await get_domain_by_name(session, sender_domain(sender)) if domain is None: raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Sender domain is not registered with agmail.", ) if domain.domain_type == "company" and domain.status != "active": raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Complete DNS setup for your company domain before sending mail.", ) created_at = utc_now() subject = payload.subject if parent_message and not subject.strip(): parent_subject = parent_message.get("subject", "") subject = parent_subject if parent_subject.lower().startswith("re:") else f"Re: {parent_subject}" document = { "direction": "outbound", "message_scope": MESSAGE_SCOPE_EXTERNAL, "account_id": str(account.id), "sender": address( sender, " ".join(part for part in [account.first_name, account.last_name] if part), ), "to": [address(value) for value in recipients], "cc": [address(value) for value in cc], "bcc": [address(value) for value in bcc], "subject": subject, "snippet": re.sub(r"\s+", " ", payload.body_text).strip()[:240], "body_text": payload.body_text, "body_html": payload.body_html, "attachments": [], "created_at": created_at, "sent_at": created_at, "status": "queued", **threading_fields, "thread_relation": "reply" if parent_message else "new_thread", "thread_resolution_method": "in_reply_to" if parent_message else "new_thread", } try: internet_message_id = await deliver_email( sender=sender, recipients=recipients, cc=cc, bcc=bcc, subject=document["subject"], body_text=document["body_text"], body_html=document.get("body_html"), domain=domain, in_reply_to=threading_fields.get("in_reply_to"), references=threading_fields.get("references"), ) document["status"] = "accepted" document["message_id"] = normalize_message_id(internet_message_id) except Exception as error: document["status"] = "failed" document["error"] = str(error) result = await database.messages.insert_one(document) document["_id"] = result.inserted_id if parent_message and thread_id: resolution = ThreadResolution( thread_id=thread_id, parent_message_id=parent_message["_id"], thread_relation="reply", thread_resolution_method="in_reply_to", ) thread_id = await ensure_thread_for_message( database, account_id=str(account.id), message=document, resolution=resolution, ) else: thread_id = await create_thread( database, account_id=str(account.id), subject=document["subject"], message_id=result.inserted_id, event_at=message_event_at(document), ) await database.mailbox_entries.insert_one( { "account_id": str(account.id), "message_id": result.inserted_id, "thread_id": thread_id, "mailbox": "sent", "message_scope": MESSAGE_SCOPE_EXTERNAL, "direction": "outbound", "sender": document["sender"], "recipients": document["to"], "subject": document["subject"], "snippet": document["snippet"], "attachments": [], "status": document["status"], "is_read": True, "is_starred": False, "event_at": created_at, "created_at": created_at, "trashed_at": None, "deletion_state": None, } ) if document["status"] == "failed": raise HTTPException( status_code=status.HTTP_502_BAD_GATEWAY, detail=document.get("error", "Unable to deliver email."), ) return EmailRead( id=str(document["_id"]), sender=sender, recipients=recipients, cc=cc, bcc=bcc, subject=subject, body_text=payload.body_text, body_html=payload.body_html, status=document["status"], created_at=created_at, ) @router.get("/{email_id}", response_model=MessageRead | EmailRead) async def get_email( email_id: str, database=Depends(get_mongo_database), account: Account = Depends(get_current_account), ): if not ObjectId.is_valid(email_id): raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Invalid email ID.", ) message_id = ObjectId(email_id) viewer_email = account.email.lower() entry = await database.mailbox_entries.find_one( {"account_id": str(account.id), "message_id": message_id} ) document = ( await database.messages.find_one({"_id": message_id}) if entry is not None else None ) if entry is not None and ( entry.get("message_scope") == "HANDOFF" or entry.get("mailbox") == "handoff" ): raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Email not found.") if document is None or entry is None: legacy = await legacy_email(database, message_id, viewer_email) if legacy is not None: return serialize_email(legacy, viewer_email) raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Email not found.", ) body_text = document.get("body_text", "") body_html = document.get("body_html") if document.get("raw_mime"): raw = await get_raw_mime_storage().get(document["raw_mime"]) parsed = parse_mime(raw) body_text = parsed["body_text"] body_html = parsed["body_html"] summary = summarize(document, entry) return MessageRead( **summary.model_dump(), cc=document.get("cc", []), bcc=document.get("bcc", []) if document.get("account_id") == str(account.id) else [], reply_to=document.get("reply_to", []), body_text=body_text, body_html=body_html, message_id=document.get("message_id"), in_reply_to=document.get("in_reply_to"), references=document.get("references", []), parent_message_id=str(document["parent_message_id"]) if document.get("parent_message_id") else None, thread_relation=document.get("thread_relation"), ) @router.get("", response_model=MessagePage) async def list_emails( mailbox: Annotated[str, Query(pattern="^(inbox|sent|trash)$")] = "inbox", limit: Annotated[int, Query(ge=1, le=100)] = 25, cursor: str | None = None, q: Annotated[str | None, Query(max_length=200)] = None, is_read: bool | None = None, sender: Annotated[str | None, Query(max_length=320)] = None, date_from: date | None = None, date_to: date | None = None, database=Depends(get_mongo_database), account: Account = Depends(get_current_account), ): entry_match: dict = { "account_id": str(account.id), "deletion_state": {"$in": [None, ""]}, } if mailbox == "trash": entry_match["trashed_at"] = {"$ne": None} else: entry_match.update({"mailbox": mailbox, "trashed_at": None}) if is_read is not None: entry_match["is_read"] = is_read if sender: pattern = f"^{re.escape(sender.strip())}" entry_match["$or"] = [ {"sender.name": {"$regex": pattern, "$options": "i"}}, {"sender.address": {"$regex": pattern, "$options": "i"}}, ] if date_from or date_to: event_range: dict = {} if date_from: event_range["$gte"] = datetime.combine( date_from, time.min, tzinfo=timezone.utc, ) if date_to: event_range["$lt"] = datetime.combine( date_to + timedelta(days=1), time.min, tzinfo=timezone.utc, ) entry_match["event_at"] = event_range if q and q.strip(): entry_match["$text"] = {"$search": q.strip()} apply_external_scope(entry_match) if cursor: created_at, entry_id = decode_cursor(cursor) cursor_clause = [ {"created_at": {"$lt": created_at}}, {"created_at": created_at, "_id": {"$lt": entry_id}}, ] if "$or" in entry_match: existing_or = entry_match.pop("$or") entry_match["$and"] = [{"$or": existing_or}, {"$or": cursor_clause}] else: entry_match["$or"] = cursor_clause pipeline: list[dict] = [ {"$match": entry_match}, {"$sort": {"created_at": -1, "_id": -1}}, {"$limit": limit + 1}, { "$lookup": { "from": "messages", "localField": "message_id", "foreignField": "_id", "as": "message", } }, {"$unwind": "$message"}, ] cursor = await database.mailbox_entries.aggregate(pipeline) rows = [row async for row in cursor] has_more = len(rows) > limit rows = rows[:limit] return MessagePage( items=[summarize(row["message"], row) for row in rows], next_cursor=encode_cursor(rows[-1]) if has_more and rows else None, ) async def require_entry(database, account: Account, email_id: str) -> tuple[ObjectId, dict]: if not ObjectId.is_valid(email_id): raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Invalid email ID.", ) message_id = ObjectId(email_id) entry = await database.mailbox_entries.find_one( {"account_id": str(account.id), "message_id": message_id} ) if entry is None: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Email not found.") return message_id, entry @router.patch("/{email_id}/read", status_code=status.HTTP_204_NO_CONTENT) async def set_email_read( email_id: str, payload: MessageReadState, database=Depends(get_mongo_database), account: Account = Depends(get_current_account), ): message_id, _ = await require_entry(database, account, email_id) await database.mailbox_entries.update_one( {"account_id": str(account.id), "message_id": message_id}, {"$set": {"is_read": payload.is_read}}, ) @router.patch("/{email_id}/star", status_code=status.HTTP_204_NO_CONTENT) async def set_email_star( email_id: str, payload: MessageStarState, database=Depends(get_mongo_database), account: Account = Depends(get_current_account), ): message_id, _ = await require_entry(database, account, email_id) await database.mailbox_entries.update_one( {"account_id": str(account.id), "message_id": message_id}, {"$set": {"is_starred": payload.is_starred}}, ) async def move_to_trash(database, account: Account, email_id: str) -> None: message_id, _ = await require_entry(database, account, email_id) await database.mailbox_entries.update_one( {"account_id": str(account.id), "message_id": message_id}, {"$set": {"trashed_at": utc_now()}}, ) @router.post("/{email_id}/trash", status_code=status.HTTP_204_NO_CONTENT) async def trash_email( email_id: str, database=Depends(get_mongo_database), account: Account = Depends(get_current_account), ): await move_to_trash(database, account, email_id) @router.delete("/{email_id}", status_code=status.HTTP_204_NO_CONTENT) async def delete_email( email_id: str, database=Depends(get_mongo_database), account: Account = Depends(get_current_account), ): await move_to_trash(database, account, email_id) @router.post("/{email_id}/restore", status_code=status.HTTP_204_NO_CONTENT) async def restore_email( email_id: str, database=Depends(get_mongo_database), account: Account = Depends(get_current_account), ): message_id, _ = await require_entry(database, account, email_id) await database.mailbox_entries.update_one( {"account_id": str(account.id), "message_id": message_id}, {"$set": {"trashed_at": None}}, ) @router.delete("/{email_id}/permanent", status_code=status.HTTP_204_NO_CONTENT) async def permanently_delete_email( email_id: str, database=Depends(get_mongo_database), account: Account = Depends(get_current_account), ): message_id, entry = await require_entry(database, account, email_id) if entry.get("trashed_at") is None: raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail="Email must be in trash before permanent deletion.", ) message = await database.messages.find_one({"_id": message_id}) other_entries = await database.mailbox_entries.count_documents( {"message_id": message_id, "_id": {"$ne": entry["_id"]}} ) await database.mailbox_entries.update_one( {"_id": entry["_id"]}, {"$set": {"deletion_state": "deleting"}}, ) if other_entries == 0 and message and message.get("raw_mime"): await get_raw_mime_storage().delete(message["raw_mime"]) await database.mailbox_entries.delete_one({"_id": entry["_id"]}) if other_entries == 0: if message and message.get("delivery_key"): await database.message_tombstones.update_one( {"delivery_key": message["delivery_key"]}, { "$setOnInsert": { "delivery_key": message["delivery_key"], "deleted_at": utc_now(), } }, upsert=True, ) await database.messages.delete_one({"_id": message_id})