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.mime_parser import parse_mime from app.services.raw_mime_storage import get_raw_mime_storage 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", []), ) 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() 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 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() document = { "direction": "outbound", "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": payload.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, "status": "queued", } try: 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, ) document["status"] = "accepted" except Exception as error: document["status"] = "failed" document["error"] = str(error) result = await database.messages.insert_one(document) document["_id"] = result.inserted_id await database.mailbox_entries.insert_one( { "account_id": str(account.id), "message_id": result.inserted_id, "mailbox": "sent", "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=payload.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 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"), ) @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()} 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})