Spaces:
Sleeping
Sleeping
| import logging | |
| from fastapi import APIRouter, BackgroundTasks, Request, Response | |
| from fastapi.responses import PlainTextResponse | |
| from app.database import get_db, get_notification_by_email_identifier, log_notification, update_notification_status | |
| from app.lib.graph.email_processor import fetch_and_store | |
| from app.lib.utils.error_handler import processing_error_handler | |
| logger = logging.getLogger(__name__) | |
| router = APIRouter(prefix="/webhook") | |
| 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 | |
| notifications = body.get("value", []) | |
| logger.info("Webhook received %d notification(s)", len(notifications)) | |
| for notification in notifications: | |
| 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 | |
| existing = await get_notification_by_email_identifier(conn, message_id) | |
| if existing: | |
| logger.info( | |
| "Skipping duplicate notification for message %s (already queued as notif_id=%s)", | |
| message_id, existing["id"], | |
| ) | |
| continue | |
| notif_id = await log_notification( | |
| conn, sub_id, change_type, resource, message_id, email_identifier=message_id | |
| ) | |
| try: | |
| async with processing_error_handler(db_path, notif_id, message_id): | |
| async with get_db(db_path) as conn: | |
| await fetch_and_store(client, message_id, notif_id, storage_path, conn) | |
| except Exception: | |
| try: | |
| async with get_db(db_path) as conn: | |
| await update_notification_status(conn, notif_id, "fetch_error") | |
| except Exception: | |
| logger.exception( | |
| "webhook: failed to write fetch_error status for notif_id=%s; " | |
| "status will remain stale in DB", | |
| notif_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 | |