Merge pull request #257 from ynput/237-activity-feed-ws-message-on-ne…

…w-activity-feed-item

Activity feed: notify clients when an activity is created, updated or deleted

api/activities/activity.py

@@ -1,6 +1,6 @@
 from datetime import datetime
 
-from fastapi import BackgroundTasks
+from fastapi import BackgroundTasks, Header
 
 from ayon_server.activities import (
     ActivityType,
@@ -42,6 +42,7 @@ async def post_project_activity(
     entity_id: PathEntityID,
     user: CurrentUser,
     activity: ProjectActivityPostModel,
+    x_sender: str | None = Header(default=None),
 ) -> CreateActivityResponseModel:
     """Create an activity.
 
@@ -67,6 +68,7 @@ async def post_project_activity(
         files=activity.files,
         user_name=user.name,
         timestamp=activity.timestamp,
+        sender=x_sender,
     )
 
     return CreateActivityResponseModel(id=id)
@@ -77,6 +79,7 @@ async def delete_project_activity(
     project_name: ProjectName,
     activity_id: str,
     user: CurrentUser,
+    x_sender: str | None = Header(default=None),
 ) -> EmptyResponse:
     """Delete an activity.
 
@@ -89,7 +92,12 @@ async def delete_project_activity(
     else:
         user_name = user.name
 
-    await delete_activity(project_name, activity_id, user_name=user_name)
+    await delete_activity(
+        project_name,
+        activity_id,
+        user_name=user_name,
+        sender=x_sender,
+    )
 
     return EmptyResponse()
 
@@ -106,6 +114,7 @@ async def patch_project_activity(
     user: CurrentUser,
     activity: ActivityPatchModel,
     background_tasks: BackgroundTasks,
+    x_sender: str | None = Header(default=None),
 ) -> EmptyResponse:
     """Edit an activity.
 
@@ -124,6 +133,7 @@ async def patch_project_activity(
         body=activity.body,
         files=activity.files,
         user_name=user_name,
+        sender=x_sender,
     )
 
     background_tasks.add_task(delete_unused_files, project_name)

ayon_server/activities/create_activity.py

@@ -39,6 +39,7 @@ async def create_activity(
     extra_references: list[ActivityReferenceModel] | None = None,
     data: dict[str, Any] | None = None,
     timestamp: datetime.datetime | None = None,
+    sender: str | None = None,
 ) -> str:
     """Create an activity.
 
@@ -175,7 +176,37 @@ async def create_activity(
             ref.insertable_tuple(activity_id, timestamp) for ref in references
         )
 
-    # Notify users
+    # Notify the front-end about the new activity
+
+    summary_references: list[dict[str, str]] = []
+    for ref in references:
+        if ref.entity_id:
+            summary_references.append(
+                {
+                    "entity_id": ref.entity_id,
+                    "entity_type": ref.entity_type,
+                    "reference_type": ref.reference_type,
+                }
+            )
+
+    summary = {
+        "activity_id": activity_id,
+        "activity_type": activity_type,
+        "references": summary_references,
+    }
+
+    await EventStream.dispatch(
+        "activity.created",
+        project=project_name,
+        description="",
+        summary=summary,
+        store=False,
+        user=user_name,
+        sender=sender,
+    )
+
+    # Send inbox notifications
+
     notify_important: list[str] = []
     notify_normal: list[str] = []
     for ref in references:
@@ -198,6 +229,7 @@ async def create_activity(
             summary={"isImportant": True},
             recipients=notify_important,
             store=False,
+            user=user_name,
         )
     if notify_normal:
         await EventStream.dispatch(
@@ -207,6 +239,7 @@ async def create_activity(
             summary={"isImportant": False},
             recipients=notify_normal,
             store=False,
+            user=user_name,
         )
 
     return activity_id

ayon_server/activities/delete_activity.py

@@ -1,13 +1,17 @@
 """Delete an existing activity."""
 
+from ayon_server.events.eventstream import EventStream
 from ayon_server.exceptions import ForbiddenException, NotFoundException
 from ayon_server.lib.postgres import Postgres
 
 __all__ = ["delete_activity"]
 
 
 async def delete_activity(
-    project_name: str, activity_id: str, user_name: str | None = None
+    project_name: str,
+    activity_id: str,
+    user_name: str | None = None,
+    sender: str | None = None,
 ) -> None:
     """Delete an activity.
 
@@ -22,7 +26,11 @@ async def delete_activity(
 
     # Load the activity first, so we can check if it really exists
     # and if the user (if provided) is the author.
-    query = f"SELECT data FROM project_{project_name}.activities WHERE id = $1"
+    query = f"""
+        SELECT data, activity_type
+        FROM project_{project_name}.activities
+        WHERE id = $1
+    """
     res = await Postgres.fetch(query, activity_id)
 
     if not res:
@@ -32,6 +40,29 @@ async def delete_activity(
         data = res[0]["data"]
         if "author" in data and data["author"] != user_name:
             raise ForbiddenException("You are not the author of this activity")
+    activity_type = res[0]["activity_type"]
+
+    # create a summary of the activity before deleting it
+    # to notify the clients
+
+    summary_references: list[dict[str, str]] = []
+    async for row in Postgres.iterate(
+        f"""
+        SELECT entity_type, entity_id, reference_type
+        FROM project_{project_name}.activity_references
+        WHERE activity_id = $1 AND entity_id IS NOT NULL
+        """,
+        activity_id,
+    ):
+        summary_references.append(dict(row))
+
+    summary = {
+        "activity_id": activity_id,
+        "activity_type": activity_type,
+        "references": summary_references,
+    }
+
+    # delete the activity
 
     async with Postgres.acquire() as conn, conn.transaction():
         # Unlink files from the activity
@@ -53,4 +84,16 @@ async def delete_activity(
             activity_id,
         )
 
+    # Notify the front-end about the deleted activity
+
+    await EventStream.dispatch(
+        "activity.deleted",
+        project=project_name,
+        description="",
+        summary=summary,
+        store=False,
+        user=user_name,
+        sender=sender,
+    )
+
     return None

ayon_server/activities/update_activity.py

@@ -8,6 +8,7 @@
     extract_mentions,
     is_body_with_checklist,
 )
+from ayon_server.events.eventstream import EventStream
 from ayon_server.exceptions import (
     BadRequestException,
     NotFoundException,
@@ -25,6 +26,7 @@ async def update_activity(
     user_name: str | None = None,
     extra_references: list[ActivityReferenceModel] | None = None,
     data: dict[str, Any] | None = None,
+    sender: str | None = None,
 ) -> None:
     """Update an activity."""
 
@@ -172,3 +174,31 @@ async def update_activity(
         await st_ref.executemany(
             ref.insertable_tuple(activity_id) for ref in references
         )
+
+    # Notify the front-end about the update
+
+    summary_references: list[dict[str, str]] = []
+    for ref in references:
+        if ref.entity_id:
+            summary_references.append(
+                {
+                    "entity_id": ref.entity_id,
+                    "entity_type": ref.entity_type,
+                    "reference_type": ref.reference_type,
+                }
+            )
+    summary = {
+        "activity_id": activity_id,
+        "activity_type": activity_type,
+        "references": summary_references,
+    }
+
+    await EventStream.dispatch(
+        "activity.updated",
+        project=project_name,
+        description="",
+        summary=summary,
+        store=False,
+        user=user_name,
+        sender=sender,
+    )