Spaces:
Sleeping
Sleeping
File size: 4,097 Bytes
4c83ab6 0a367ef 4c83ab6 780bb45 4c83ab6 0a367ef 4c83ab6 0a367ef 4c83ab6 0a367ef 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 106 107 108 109 110 111 | 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")
@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
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
|