initial commit

Dockerfile

@@ -0,0 +1,21 @@
+FROM python:3.12-slim
+
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    build-essential \
+    && rm -rf /var/lib/apt/lists/*
+
+WORKDIR /app
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY entrypoint.sh /entrypoint.sh
+RUN chmod +x /entrypoint.sh
+
+COPY app/ ./app/
+
+RUN mkdir -p /data /emails /certs
+
+EXPOSE 8443
+
+ENTRYPOINT ["/entrypoint.sh"]

app/auth.py

@@ -0,0 +1,24 @@
+import msal
+
+from app.config import Settings
+
+
+class GraphAuthProvider:
+    _SCOPE = ["https://graph.microsoft.com/.default"]
+
+    def __init__(self, settings: Settings) -> None:
+        self._app = msal.ConfidentialClientApplication(
+            client_id=settings.azure_client_id,
+            authority=f"https://login.microsoftonline.com/{settings.azure_tenant_id}",
+            client_credential=settings.azure_client_secret.get_secret_value(),
+        )
+
+    def get_access_token(self) -> str:
+        result = self._app.acquire_token_silent(self._SCOPE, account=None)
+        if not result:
+            result = self._app.acquire_token_for_client(scopes=self._SCOPE)
+        if "access_token" not in result:
+            raise RuntimeError(
+                f"MSAL token error: {result.get('error_description', result.get('error'))}"
+            )
+        return result["access_token"]

app/config.py

@@ -0,0 +1,27 @@
+from pathlib import Path
+
+from pydantic import HttpUrl, SecretStr
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class Settings(BaseSettings):
+    azure_tenant_id: str
+    azure_client_id: str
+    azure_client_secret: SecretStr
+
+    mail_folder_name: str
+    notification_url: HttpUrl
+    subscription_expiry_minutes: int = 4230
+
+    host: str = "0.0.0.0"
+    port: int = 8443
+    tls_cert_file: Path
+    tls_key_file: Path
+
+    database_path: Path = Path("/data/rcmemail.db")
+    email_storage_path: Path = Path("/emails")
+
+    model_config = SettingsConfigDict(env_file=".env", extra="ignore")
+
+
+settings = Settings()

app/database.py

@@ -0,0 +1,150 @@
+import sys
+from contextlib import asynccontextmanager
+from pathlib import Path
+from typing import AsyncGenerator
+
+import aiosqlite
+
+_DDL = """
+PRAGMA journal_mode=WAL;
+PRAGMA foreign_keys=ON;
+
+CREATE TABLE IF NOT EXISTS subscriptions (
+    id               TEXT PRIMARY KEY,
+    folder_id        TEXT NOT NULL,
+    folder_name      TEXT NOT NULL,
+    expiry_datetime  TEXT NOT NULL,
+    notification_url TEXT NOT NULL,
+    client_state     TEXT NOT NULL,
+    is_active        INTEGER NOT NULL DEFAULT 1,
+    created_at       TEXT NOT NULL DEFAULT (datetime('now')),
+    last_renewed_at  TEXT
+);
+
+CREATE TABLE IF NOT EXISTS notifications (
+    id                INTEGER PRIMARY KEY AUTOINCREMENT,
+    received_at       TEXT NOT NULL DEFAULT (datetime('now')),
+    subscription_id   TEXT,
+    change_type       TEXT,
+    resource          TEXT,
+    message_id        TEXT,
+    processing_status TEXT NOT NULL DEFAULT 'pending',
+    error_message     TEXT,
+    storage_path      TEXT
+);
+"""
+
+
+async def init_db(db_path: Path) -> None:
+    db_path.parent.mkdir(parents=True, exist_ok=True)
+    async with aiosqlite.connect(db_path) as conn:
+        await conn.executescript(_DDL)
+        await conn.commit()
+
+
+@asynccontextmanager
+async def get_db(db_path: Path) -> AsyncGenerator[aiosqlite.Connection, None]:
+    async with aiosqlite.connect(db_path) as conn:
+        conn.row_factory = aiosqlite.Row
+        await conn.execute("PRAGMA journal_mode=WAL")
+        await conn.execute("PRAGMA foreign_keys=ON")
+        yield conn
+
+
+async def upsert_subscription(
+    conn: aiosqlite.Connection,
+    sub_id: str,
+    folder_id: str,
+    folder_name: str,
+    expiry_datetime: str,
+    notification_url: str,
+    client_state: str,
+) -> None:
+    await conn.execute(
+        """
+        INSERT INTO subscriptions (id, folder_id, folder_name, expiry_datetime,
+            notification_url, client_state, is_active)
+        VALUES (?, ?, ?, ?, ?, ?, 1)
+        ON CONFLICT(id) DO UPDATE SET
+            expiry_datetime  = excluded.expiry_datetime,
+            last_renewed_at  = datetime('now'),
+            is_active        = 1
+        """,
+        (sub_id, folder_id, folder_name, expiry_datetime, notification_url, client_state),
+    )
+    await conn.commit()
+
+
+async def get_active_subscription(conn: aiosqlite.Connection) -> dict | None:
+    cursor = await conn.execute(
+        "SELECT * FROM subscriptions WHERE is_active = 1 ORDER BY created_at DESC LIMIT 1"
+    )
+    row = await cursor.fetchone()
+    return dict(row) if row else None
+
+
+async def mark_subscription_inactive(conn: aiosqlite.Connection, sub_id: str) -> None:
+    await conn.execute(
+        "UPDATE subscriptions SET is_active = 0 WHERE id = ?", (sub_id,)
+    )
+    await conn.commit()
+
+
+async def log_notification(
+    conn: aiosqlite.Connection,
+    subscription_id: str | None,
+    change_type: str | None,
+    resource: str | None,
+    message_id: str | None,
+) -> int:
+    cursor = await conn.execute(
+        """
+        INSERT INTO notifications (subscription_id, change_type, resource, message_id)
+        VALUES (?, ?, ?, ?)
+        """,
+        (subscription_id, change_type, resource, message_id),
+    )
+    await conn.commit()
+    return cursor.lastrowid
+
+
+async def update_subscription_expiry(
+    conn: aiosqlite.Connection, sub_id: str, expiry_datetime: str
+) -> None:
+    await conn.execute(
+        """
+        UPDATE subscriptions
+        SET expiry_datetime = ?, last_renewed_at = datetime('now')
+        WHERE id = ?
+        """,
+        (expiry_datetime, sub_id),
+    )
+    await conn.commit()
+
+
+async def update_notification_status(
+    conn: aiosqlite.Connection,
+    notif_id: int,
+    status: str,
+    error_message: str | None = None,
+    storage_path: str | None = None,
+) -> None:
+    await conn.execute(
+        """
+        UPDATE notifications
+        SET processing_status = ?, error_message = ?, storage_path = ?
+        WHERE id = ?
+        """,
+        (status, error_message, storage_path, notif_id),
+    )
+    await conn.commit()
+
+
+if __name__ == "__main__":
+    import asyncio
+
+    if "--init" in sys.argv:
+        from app.config import settings
+
+        asyncio.run(init_db(settings.database_path))
+        print(f"Database initialized at {settings.database_path}")

app/email_processor.py

@@ -0,0 +1,63 @@
+import base64
+import logging
+from pathlib import Path
+
+import aiosqlite
+
+from app.database import update_notification_status
+from app.graph_client import GraphClient
+from app.storage import write_email
+
+logger = logging.getLogger(__name__)
+
+_LARGE_ATTACHMENT_THRESHOLD = 3 * 1024 * 1024  # 3 MB
+
+
+async def fetch_and_store(
+    client: GraphClient,
+    message_id: str,
+    notif_db_id: int,
+    storage_path: Path,
+    conn: aiosqlite.Connection,
+) -> Path:
+    try:
+        email_data = await client.get(
+            f"/me/messages/{message_id}",
+            params={"$expand": "attachments", "$select": (
+                "id,subject,from,toRecipients,ccRecipients,receivedDateTime,"
+                "body,uniqueBody,hasAttachments,attachments,webLink"
+            )},
+        )
+
+        attachment_bytes: list[tuple[str, str, bytes]] = []
+        for att in email_data.get("attachments", []):
+            if att.get("@odata.type") != "#microsoft.graph.fileAttachment":
+                continue
+            filename = att.get("name", "attachment")
+            content_type = att.get("contentType", "application/octet-stream")
+            size = att.get("size", 0)
+
+            if size > _LARGE_ATTACHMENT_THRESHOLD or not att.get("contentBytes"):
+                raw = await _fetch_large_attachment(client, message_id, att["id"])
+            else:
+                raw = base64.b64decode(att["contentBytes"])
+
+            attachment_bytes.append((filename, content_type, raw))
+
+        dest = await write_email(email_data, attachment_bytes, storage_path)
+        await update_notification_status(
+            conn, notif_db_id, "fetched", storage_path=str(dest)
+        )
+        logger.info("Stored email %s at %s", message_id, dest)
+        return dest
+
+    except Exception as exc:
+        logger.exception("Failed to process message %s", message_id)
+        await update_notification_status(conn, notif_db_id, "error", error_message=str(exc))
+        raise
+
+
+async def _fetch_large_attachment(
+    client: GraphClient, message_id: str, attachment_id: str
+) -> bytes:
+    return await client.get_raw(f"/me/messages/{message_id}/attachments/{attachment_id}/$value")

app/folder_resolver.py

@@ -0,0 +1,26 @@
+from app.graph_client import GraphClient
+
+_BASE = "https://graph.microsoft.com/v1.0"
+
+
+async def resolve_folder_id(client: GraphClient, display_name: str) -> str:
+    path = "/me/mailFolders"
+    params: dict = {"$top": 100}
+
+    while path:
+        data = await client.get(path, params=params)
+        for folder in data.get("value", []):
+            if folder["displayName"].lower() == display_name.lower():
+                return folder["id"]
+        next_link: str | None = data.get("@odata.nextLink")
+        if next_link:
+            # nextLink is a full URL; strip the base so GraphClient can prefix it again.
+            path = next_link.removeprefix(_BASE)
+            params = {}
+        else:
+            break
+
+    raise ValueError(
+        f"Mail folder '{display_name}' not found. "
+        f"Verify it exists as a top-level folder in the monitored mailbox."
+    )

app/graph_client.py

@@ -0,0 +1,54 @@
+import asyncio
+
+import httpx
+
+from app.auth import GraphAuthProvider
+
+_BASE_URL = "https://graph.microsoft.com/v1.0"
+
+
+class GraphClient:
+    def __init__(self, auth: GraphAuthProvider) -> None:
+        self._auth = auth
+        self._client = httpx.AsyncClient(timeout=30.0)
+
+    async def _headers(self) -> dict[str, str]:
+        token = await asyncio.to_thread(self._auth.get_access_token)
+        return {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
+
+    async def get(self, path: str, params: dict | None = None) -> dict:
+        resp = await self._client.get(
+            f"{_BASE_URL}{path}", headers=await self._headers(), params=params
+        )
+        resp.raise_for_status()
+        return resp.json()
+
+    async def get_raw(self, path: str) -> bytes:
+        resp = await self._client.get(
+            f"{_BASE_URL}{path}", headers=await self._headers()
+        )
+        resp.raise_for_status()
+        return resp.content
+
+    async def post(self, path: str, body: dict) -> dict:
+        resp = await self._client.post(
+            f"{_BASE_URL}{path}", headers=await self._headers(), json=body
+        )
+        resp.raise_for_status()
+        return resp.json()
+
+    async def patch(self, path: str, body: dict) -> dict:
+        resp = await self._client.patch(
+            f"{_BASE_URL}{path}", headers=await self._headers(), json=body
+        )
+        resp.raise_for_status()
+        return resp.json()
+
+    async def delete(self, path: str) -> None:
+        resp = await self._client.delete(
+            f"{_BASE_URL}{path}", headers=await self._headers()
+        )
+        resp.raise_for_status()
+
+    async def aclose(self) -> None:
+        await self._client.aclose()

app/main.py

@@ -0,0 +1,107 @@
+import asyncio
+import logging
+from contextlib import asynccontextmanager
+from datetime import datetime, timedelta, timezone
+from typing import AsyncGenerator
+
+from fastapi import FastAPI
+
+from app.auth import GraphAuthProvider
+from app.config import settings
+from app.database import get_active_subscription, get_db, init_db
+from app.folder_resolver import resolve_folder_id
+from app.graph_client import GraphClient
+from app.scheduler import RenewalScheduler
+from app.subscription import SubscriptionManager
+from app.webhook import router as webhook_router
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s %(levelname)s %(name)s: %(message)s",
+)
+logger = logging.getLogger(__name__)
+
+
+async def _deferred_subscription_setup(
+    app: FastAPI,
+    sub_manager: SubscriptionManager,
+    folder_id: str,
+    scheduler: RenewalScheduler,
+) -> None:
+    """
+    Runs as a background task after the server is live so that Graph's
+    validation-token POST can reach us during subscription registration.
+    """
+    try:
+        sub_id = await sub_manager.ensure_subscription(folder_id)
+        app.state.subscription_id = sub_id
+        scheduler.start(sub_id)
+        logger.info("Registered new subscription %s", sub_id)
+    except Exception:
+        logger.exception("Failed to register Graph subscription on startup")
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI) -> AsyncGenerator:
+    # --- Startup ---
+    await init_db(settings.database_path)
+
+    auth = GraphAuthProvider(settings)
+    client = GraphClient(auth)
+    app.state.client = client
+    app.state.settings = settings
+    app.state.subscription_id = None
+
+    folder_id = await resolve_folder_id(client, settings.mail_folder_name)
+    app.state.folder_id = folder_id
+    logger.info("Monitoring folder '%s' (id=%s)", settings.mail_folder_name, folder_id)
+
+    sub_manager = SubscriptionManager(client, settings, settings.database_path)
+    app.state.sub_manager = sub_manager
+
+    scheduler = RenewalScheduler(sub_manager, settings)
+    app.state.scheduler = scheduler
+
+    # If a valid subscription already exists in the DB we can start the
+    # scheduler immediately (no Graph network call needed).
+    async with get_db(settings.database_path) as conn:
+        existing = await get_active_subscription(conn)
+
+    needs_new_subscription = True
+    if existing:
+        expiry = datetime.fromisoformat(existing["expiry_datetime"].replace("Z", "+00:00"))
+        cutoff = datetime.now(timezone.utc) + timedelta(minutes=60)
+        if expiry > cutoff:
+            app.state.subscription_id = existing["id"]
+            scheduler.start(existing["id"])
+            logger.info("Reusing existing subscription %s", existing["id"])
+            needs_new_subscription = False
+
+    if needs_new_subscription:
+        # Schedule subscription creation as a background task that fires
+        # after yield — at that point uvicorn is accepting connections so
+        # Graph's validation-token POST will succeed.
+        asyncio.create_task(
+            _deferred_subscription_setup(app, sub_manager, folder_id, scheduler)
+        )
+
+    yield
+
+    # --- Shutdown ---
+    scheduler.shutdown()
+    await client.aclose()
+    logger.info("Shutdown complete")
+
+
+app = FastAPI(title="RCMEmail", version="1.0.0", lifespan=lifespan)
+app.include_router(webhook_router)
+
+
+@app.get("/health")
+async def health() -> dict:
+    return {
+        "status": "ok",
+        "folder": settings.mail_folder_name,
+        "folder_id": getattr(app.state, "folder_id", None),
+        "subscription_id": getattr(app.state, "subscription_id", None),
+    }

app/scheduler.py

@@ -0,0 +1,67 @@
+import logging
+
+from apscheduler.schedulers.asyncio import AsyncIOScheduler
+
+from app.config import Settings
+
+logger = logging.getLogger(__name__)
+
+
+class RenewalScheduler:
+    def __init__(self, subscription_manager, settings: Settings) -> None:
+        self._scheduler = AsyncIOScheduler()
+        self._manager = subscription_manager
+        self._settings = settings
+
+    def start(self, subscription_id: str) -> None:
+        interval_seconds = int(self._settings.subscription_expiry_minutes * 60 * 0.80)
+        self._scheduler.add_job(
+            self._renew_job,
+            trigger="interval",
+            seconds=interval_seconds,
+            id="subscription_renewal",
+            replace_existing=True,
+            args=[subscription_id],
+        )
+        self._scheduler.start()
+        logger.info(
+            "Renewal scheduler started; will renew subscription every %d seconds",
+            interval_seconds,
+        )
+
+    async def _renew_job(self, subscription_id: str) -> None:
+        logger.info("Renewing subscription %s", subscription_id)
+        try:
+            await self._manager.renew_subscription(subscription_id)
+        except Exception:
+            logger.warning(
+                "Renewal failed for %s; attempting to recreate subscription", subscription_id
+            )
+            try:
+                folder_id = await self._get_folder_id()
+                new_id = await self._manager.ensure_subscription(folder_id)
+                self._reschedule(new_id)
+            except Exception:
+                logger.exception("Failed to recreate subscription after renewal failure")
+
+    async def _get_folder_id(self) -> str:
+        from app.database import get_active_subscription, get_db
+        async with get_db(self._manager._db_path) as conn:
+            sub = await get_active_subscription(conn)
+            if sub:
+                return sub["folder_id"]
+        raise RuntimeError("No active subscription found to determine folder_id for recreation")
+
+    def _reschedule(self, new_subscription_id: str) -> None:
+        interval_seconds = int(self._settings.subscription_expiry_minutes * 60 * 0.80)
+        self._scheduler.reschedule_job(
+            "subscription_renewal",
+            trigger="interval",
+            seconds=interval_seconds,
+            args=[new_subscription_id],
+        )
+        logger.info("Rescheduled renewal for new subscription %s", new_subscription_id)
+
+    def shutdown(self) -> None:
+        if self._scheduler.running:
+            self._scheduler.shutdown(wait=False)

app/storage.py

@@ -0,0 +1,102 @@
+import json
+import os
+import uuid
+from datetime import datetime, timezone
+from pathlib import Path
+
+
+def _safe_filename(name: str) -> str:
+    return "".join(c if (c.isalnum() or c in "._- ") else "_" for c in name).strip()
+
+
+def build_email_dir(base_path: Path, received_datetime: str, message_id: str) -> Path:
+    try:
+        dt = datetime.fromisoformat(received_datetime.replace("Z", "+00:00"))
+    except ValueError:
+        dt = datetime.now(timezone.utc)
+    date_folder = dt.strftime("%Y-%m-%d")
+    time_prefix = dt.strftime("%Y%m%d_%H%M%S")
+    folder_name = f"{time_prefix}_{message_id[:8]}"
+    return base_path / date_folder / folder_name
+
+
+async def write_email(
+    email_data: dict,
+    attachment_bytes: list[tuple[str, str, bytes]],
+    base_path: Path,
+) -> Path:
+    received = email_data.get("receivedDateTime", "")
+    message_id = email_data.get("id", str(uuid.uuid4()))
+    final_dir = build_email_dir(base_path, received, message_id)
+
+    tmp_dir = base_path / ".tmp" / str(uuid.uuid4())
+    tmp_dir.mkdir(parents=True, exist_ok=True)
+
+    try:
+        body = email_data.get("body", {})
+        if body.get("contentType", "").lower() == "html":
+            (tmp_dir / "body.html").write_text(body.get("content", ""), encoding="utf-8")
+        else:
+            (tmp_dir / "body.txt").write_text(body.get("content", ""), encoding="utf-8")
+
+        if email_data.get("uniqueBody"):
+            unique = email_data["uniqueBody"]
+            if unique.get("contentType", "").lower() == "html":
+                (tmp_dir / "body_unique.html").write_text(
+                    unique.get("content", ""), encoding="utf-8"
+                )
+            else:
+                (tmp_dir / "body_unique.txt").write_text(
+                    unique.get("content", ""), encoding="utf-8"
+                )
+
+        stored_attachments = []
+        if attachment_bytes:
+            att_dir = tmp_dir / "attachments"
+            att_dir.mkdir()
+            for filename, content_type, raw in attachment_bytes:
+                safe_name = _safe_filename(filename) or "attachment"
+                dest = att_dir / safe_name
+                dest.write_bytes(raw)
+                stored_attachments.append(
+                    {
+                        "filename": safe_name,
+                        "content_type": content_type,
+                        "size_bytes": len(raw),
+                    }
+                )
+
+        from_addr = email_data.get("from", {}).get("emailAddress", {})
+        metadata = {
+            "message_id": message_id,
+            "subject": email_data.get("subject", ""),
+            "from": {"name": from_addr.get("name", ""), "address": from_addr.get("address", "")},
+            "to": [
+                {"name": r.get("emailAddress", {}).get("name", ""),
+                 "address": r.get("emailAddress", {}).get("address", "")}
+                for r in email_data.get("toRecipients", [])
+            ],
+            "cc": [
+                {"name": r.get("emailAddress", {}).get("name", ""),
+                 "address": r.get("emailAddress", {}).get("address", "")}
+                for r in email_data.get("ccRecipients", [])
+            ],
+            "received_datetime": received,
+            "has_attachments": email_data.get("hasAttachments", False),
+            "attachment_count": len(stored_attachments),
+            "attachments": stored_attachments,
+            "graph_web_link": email_data.get("webLink", ""),
+            "stored_at": datetime.now(timezone.utc).isoformat(),
+        }
+        (tmp_dir / "metadata.json").write_text(
+            json.dumps(metadata, indent=2), encoding="utf-8"
+        )
+
+        final_dir.parent.mkdir(parents=True, exist_ok=True)
+        os.rename(tmp_dir, final_dir)
+        return final_dir
+
+    except Exception:
+        import shutil
+        shutil.rmtree(tmp_dir, ignore_errors=True)
+        raise

app/subscription.py

@@ -0,0 +1,104 @@
+import logging
+import uuid
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+import httpx
+
+from app.config import Settings
+from app.database import (
+    get_db,
+    get_active_subscription,
+    mark_subscription_inactive,
+    update_subscription_expiry,
+    upsert_subscription,
+)
+from app.graph_client import GraphClient
+
+logger = logging.getLogger(__name__)
+
+_RENEWAL_MARGIN_MINUTES = 60
+
+
+class SubscriptionManager:
+    def __init__(self, client: GraphClient, settings: Settings, db_path: Path) -> None:
+        self._client = client
+        self._settings = settings
+        self._db_path = db_path
+
+    async def ensure_subscription(self, folder_id: str) -> str:
+        async with get_db(self._db_path) as conn:
+            existing = await get_active_subscription(conn)
+
+        if existing:
+            expiry = datetime.fromisoformat(existing["expiry_datetime"].replace("Z", "+00:00"))
+            cutoff = datetime.now(timezone.utc) + timedelta(minutes=_RENEWAL_MARGIN_MINUTES)
+            if expiry > cutoff:
+                logger.info("Reusing existing subscription %s (expires %s)", existing["id"], expiry)
+                return existing["id"]
+            logger.info("Existing subscription %s is near expiry; creating new one", existing["id"])
+
+        return await self._create_subscription(folder_id)
+
+    async def _create_subscription(self, folder_id: str) -> str:
+        client_state = str(uuid.uuid4())
+        expiry = datetime.now(timezone.utc) + timedelta(
+            minutes=self._settings.subscription_expiry_minutes
+        )
+        expiry_str = expiry.strftime("%Y-%m-%dT%H:%M:%S.0000000Z")
+
+        body = {
+            "changeType": "created",
+            "notificationUrl": str(self._settings.notification_url),
+            "resource": f"me/mailFolders/{folder_id}/messages",
+            "expirationDateTime": expiry_str,
+            "clientState": client_state,
+        }
+
+        data = await self._client.post("/subscriptions", body)
+        sub_id = data["id"]
+        actual_expiry = data.get("expirationDateTime", expiry_str)
+
+        async with get_db(self._db_path) as conn:
+            await upsert_subscription(
+                conn,
+                sub_id=sub_id,
+                folder_id=folder_id,
+                folder_name=self._settings.mail_folder_name,
+                expiry_datetime=actual_expiry,
+                notification_url=str(self._settings.notification_url),
+                client_state=client_state,
+            )
+
+        logger.info("Created subscription %s expiring %s", sub_id, actual_expiry)
+        return sub_id
+
+    async def renew_subscription(self, sub_id: str) -> None:
+        expiry = datetime.now(timezone.utc) + timedelta(
+            minutes=self._settings.subscription_expiry_minutes
+        )
+        expiry_str = expiry.strftime("%Y-%m-%dT%H:%M:%S.0000000Z")
+
+        try:
+            data = await self._client.patch(
+                f"/subscriptions/{sub_id}",
+                {"expirationDateTime": expiry_str},
+            )
+            actual_expiry = data.get("expirationDateTime", expiry_str)
+            async with get_db(self._db_path) as conn:
+                await update_subscription_expiry(conn, sub_id, actual_expiry)
+            logger.info("Renewed subscription %s until %s", sub_id, actual_expiry)
+        except httpx.HTTPStatusError as exc:
+            if exc.response.status_code == 404:
+                logger.warning("Subscription %s not found on renewal; will recreate", sub_id)
+                async with get_db(self._db_path) as conn:
+                    await mark_subscription_inactive(conn, sub_id)
+                raise
+            raise
+
+    async def get_client_state(self, sub_id: str) -> str | None:
+        async with get_db(self._db_path) as conn:
+            existing = await get_active_subscription(conn)
+        if existing and existing["id"] == sub_id:
+            return existing["client_state"]
+        return None

app/webhook.py

@@ -0,0 +1,87 @@
+import logging
+
+from fastapi import APIRouter, BackgroundTasks, Request, Response
+from fastapi.responses import PlainTextResponse
+
+from app.database import get_db, log_notification, update_notification_status
+from app.email_processor import fetch_and_store
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/webhook")
+
+
+@router.post("/notify")
+async def handle_notification(
+    request: Request,
+    background_tasks: BackgroundTasks,
+    validationToken: str | None = None,
+) -> Response:
+    if validationToken:
+        logger.info("Graph webhook validation request received")
+        return PlainTextResponse(content=validationToken, status_code=200)
+
+    try:
+        body = await request.json()
+    except Exception:
+        logger.warning("Webhook received non-JSON body")
+        return Response(status_code=400)
+
+    background_tasks.add_task(_process_notifications, request, body)
+    return Response(status_code=202)
+
+
+async def _process_notifications(request: Request, body: dict) -> None:
+    client = request.app.state.client
+    settings = request.app.state.settings
+    db_path = settings.database_path
+    storage_path = settings.email_storage_path
+
+    for notification in body.get("value", []):
+        change_type = notification.get("changeType", "")
+        if change_type != "created":
+            logger.debug("Skipping notification with changeType=%s", change_type)
+            continue
+
+        sub_id = notification.get("subscriptionId")
+        resource = notification.get("resource", "")
+        incoming_state = notification.get("clientState")
+
+        resource_data = notification.get("resourceData", {})
+        message_id = resource_data.get("id") or _extract_message_id(resource)
+
+        if not message_id:
+            logger.warning("Could not extract message ID from notification: %s", notification)
+            continue
+
+        async with get_db(db_path) as conn:
+            expected_state = await _get_client_state(conn, sub_id)
+            if expected_state and incoming_state != expected_state:
+                logger.warning(
+                    "clientState mismatch for subscription %s — discarding notification", sub_id
+                )
+                continue
+
+            notif_id = await log_notification(conn, sub_id, change_type, resource, message_id)
+
+        try:
+            async with get_db(db_path) as conn:
+                await fetch_and_store(client, message_id, notif_id, storage_path, conn)
+        except Exception:
+            logger.exception("Error processing message %s", message_id)
+
+
+def _extract_message_id(resource: str) -> str | None:
+    parts = resource.split("/")
+    for i, part in enumerate(parts):
+        if part.lower() == "messages" and i + 1 < len(parts):
+            return parts[i + 1]
+    return None
+
+
+async def _get_client_state(conn, sub_id: str) -> str | None:
+    from app.database import get_active_subscription
+    sub = await get_active_subscription(conn)
+    if sub and sub.get("id") == sub_id:
+        return sub.get("client_state")
+    return None

docker-compose.yml

@@ -0,0 +1,15 @@
+services:
+  rcmemail:
+    build: .
+    env_file: .env
+    ports:
+      - "8443:8443"
+    volumes:
+      - ./certs:/certs:ro
+      - rcmemail-data:/data
+      - rcmemail-emails:/emails
+    restart: unless-stopped
+
+volumes:
+  rcmemail-data:
+  rcmemail-emails:

entrypoint.sh

@@ -0,0 +1,10 @@
+#!/bin/bash
+set -e
+
+python -m app.database --init
+
+exec uvicorn app.main:app \
+    --host "${HOST:-0.0.0.0}" \
+    --port "${PORT:-8443}" \
+    --ssl-certfile "${TLS_CERT_FILE}" \
+    --ssl-keyfile "${TLS_KEY_FILE}"

requirements.txt

@@ -0,0 +1,7 @@
+fastapi==0.115.0
+uvicorn[standard]==0.30.0
+msal==1.31.0
+httpx==0.27.0
+pydantic-settings==2.3.0
+apscheduler==3.10.4
+aiosqlite==0.20.0