From d66ce2f450f5338021de7029f3b7223c4224202b Mon Sep 17 00:00:00 2001 From: Peter Vacho Date: Sat, 4 Jan 2025 19:54:04 +0100 Subject: [PATCH] Add support for deleting notifications --- src/api/notifications.py | 66 +++++++++++++++++++++++++++++++++++++++- 1 file changed, 65 insertions(+), 1 deletion(-) diff --git a/src/api/notifications.py b/src/api/notifications.py index 0ccdaad..580c391 100644 --- a/src/api/notifications.py +++ b/src/api/notifications.py @@ -2,9 +2,10 @@ from datetime import UTC, datetime from typing import Literal, cast, final from beanie import Link, PydanticObjectId -from fastapi import APIRouter, HTTPException, status +from fastapi import APIRouter, HTTPException, Response, status from pydantic import BaseModel +from src.db.models.invitation import Invitation from src.db.models.notification import Notification from src.db.models.user import User from src.utils.db import MissingIdError, UnfetchedLinkError, expr @@ -136,6 +137,69 @@ async def unread_notification(notification_id: PydanticObjectId, user: CurrentUs return NotificationData.from_notification(notification) +@notifications_router.delete("/{notification_id}") +async def delete_notification(notification_id: PydanticObjectId, user: CurrentUserDep) -> Response: + """Delete a notification.""" + notification = await Notification.get(notification_id, fetch_links=True) + + if notification is None: + raise HTTPException(status.HTTP_404_NOT_FOUND, "Notification not found.") + + if user.id is None: + raise MissingIdError(user) + + if cast(User, notification.user).id != user.id: + raise HTTPException(status.HTTP_403_FORBIDDEN, "You can only access your own notifications.") + + # Clean up associated invitations, if any + if notification.event_type == "invitation": + # Decline the incoming invite + if notification.message == "new-invitation": + invitation = await Invitation.get(notification.data, fetch_links=True) + + if invitation is not None and invitation.status == "pending": + # Just a sanity check, the user should always be invitee here + if cast(User, invitation.invitee).id != user.id: + raise RuntimeError("User is not the invitee of the pending invitation.") + + invitation.status = "declined" + invitation.responded_at = datetime.now(tz=UTC) + invitation = await invitation.replace() + + # Send back a notification to the invitor about the decline + new_notification = Notification( + user=invitation.invitor, + event_type="invitation", + message="invitation-declined", + data=str(cast(PydanticObjectId, invitation.id)), + read=False, + ) + _ = await new_notification.create() + + # Delete the accepted/declined invitation + elif notification.message in {"invitation-declined", "invitation-accepted"}: + invitation = await Invitation.get(notification.data, fetch_links=True) + if invitation is not None: + # Just a sanity check, the user should always be owner here + if cast(User, invitation.invitor).id != user.id: + raise RuntimeError("User is not the owner of the accepted/declined invitation.") + + # NOTE: It might be worth it to check whether the user has deleted their new-invitation notification + # first, otherwise, the frontend will not be able to show that old invitation in the notification + # however, the frontend can handle this case for now by just showing that the invitation was deleted. + # But it could be a good idea to implement this in the future. Note that if this is added, we'll also + # need to add a reverse-check, for when the user deletes the new-invitation notification, we should + # also check if the owner has deleted the accepted/declined invitation and delete it as well. + _ = await invitation.delete() + + # This should never happen, just a sanity check, in case we add some more messages + else: + raise RuntimeError(f"Unknown invitation message: {notification.message}") + + _ = await notification.delete() + return Response(status_code=status.HTTP_204_NO_CONTENT) + + router = APIRouter() router.include_router(base_router) router.include_router(notifications_router)