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.lib.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"users/{self._settings.mailbox_user}/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