Add support for deleting notifications

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)