Spaces:
Sleeping
Sleeping
File size: 4,008 Bytes
4c83ab6 0a367ef 4c83ab6 780bb45 4c83ab6 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 | 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
|