fix: handle exception in eventstream handlers

ayon_server/activities/create_activity.py

@@ -25,7 +25,7 @@
 from ayon_server.activities.watchers.watcher_list import get_watcher_list
 from ayon_server.entities.core import ProjectLevelEntity
 from ayon_server.events.eventstream import EventStream
-from ayon_server.exceptions import BadRequestException
+from ayon_server.exceptions import BadRequestException, NotFoundException
 from ayon_server.lib.postgres import Postgres
 from ayon_server.utils import create_uuid
 
@@ -79,7 +79,12 @@ async def create_activity(
         origin["label"] = entity.label
     data["origin"] = origin
 
-    data["parents"] = await get_parents_from_entity(entity)
+    try:
+        data["parents"] = await get_parents_from_entity(entity)
+    except Postgres.UndefinedTableError as e:
+        raise NotFoundException(
+            "Unable to get references. " f"Project {project_name} no longer exists"
+        ) from e
 
     if activity_type == "comment" and is_body_with_checklist(body):
         data["hasChecklist"] = True
@@ -130,7 +135,12 @@ async def create_activity(
             )
 
         # Add related entities references
-        references.update(await get_references_from_entity(entity))
+        try:
+            references.update(await get_references_from_entity(entity))
+        except Postgres.UndefinedTableError as e:
+            raise NotFoundException(
+                "Unable to get references. " f"Project {project_name} no longer exists"
+            ) from e
 
     #
     # Create the activity
@@ -154,20 +164,31 @@ async def create_activity(
     """
 
     async with Postgres.acquire() as conn, conn.transaction():
-        await conn.execute(query, activity_id, activity_type, body, data, timestamp)
+        try:
+            await conn.execute(query, activity_id, activity_type, body, data, timestamp)
+        except Postgres.UndefinedTableError as e:
+            raise NotFoundException(
+                "Unable to create activity. " f"Project {project_name} no longer exists"
+            ) from e
 
         if files is not None:
-            await conn.execute(
-                f"""
-                UPDATE project_{project_name}.files
-                SET
-                    activity_id = $1,
-                    updated_at = NOW()
-                WHERE id = ANY($2)
-                """,
-                activity_id,
-                files,
-            )
+            try:
+                await conn.execute(
+                    f"""
+                    UPDATE project_{project_name}.files
+                    SET
+                        activity_id = $1,
+                        updated_at = NOW()
+                    WHERE id = ANY($2)
+                    """,
+                    activity_id,
+                    files,
+                )
+            except Postgres.UndefinedTableError as e:
+                raise NotFoundException(
+                    "Unable to update files. "
+                    f"Project {project_name} no longer exists"
+                ) from e
 
         st_ref = await conn.prepare(
             f"""
@@ -190,9 +211,15 @@ async def create_activity(
         """
         )
 
-        await st_ref.executemany(
-            ref.insertable_tuple(activity_id, timestamp) for ref in references
-        )
+        try:
+            await st_ref.executemany(
+                ref.insertable_tuple(activity_id, timestamp) for ref in references
+            )
+        except Postgres.UndefinedTableError as e:
+            raise NotFoundException(
+                "Unable to create references. "
+                f"Project {project_name} no longer exists"
+            ) from e
 
     # Notify the front-end about the new activity
 

ayon_server/activities/watchers/set_watchers.py

@@ -51,7 +51,14 @@ async def set_watchers(
             WHERE id IN (SELECT activity_id FROM activities_to_delete)
         """
 
-        await Postgres.execute(query, entity.entity_type, entity.id, unwatchers)
+        try:
+            await Postgres.execute(query, entity.entity_type, entity.id, unwatchers)
+        except Postgres.UndefinedTableError:
+            logging.debug(
+                "Unable to delete watchers. "
+                f"Project {entity.project_name} no longer exists"
+            )
+            return
 
     # Add new watchers
 
@@ -76,7 +83,12 @@ async def ensure_watching(entity: ProjectLevelEntity, user: UserEntity | str) ->
 
     user_name = user.name if isinstance(user, UserEntity) else user
 
-    watchers = await get_watcher_list(entity)
+    try:
+        watchers = await get_watcher_list(entity)
+    except Postgres.UndefinedTableError:
+        logging.debug(f"Unable to set watchers. Entity {entity} no longer exists")
+        return
+
     if user_name not in watchers:
         logging.debug(f"Adding {user_name} to watchers of {entity}")
         watchers.append(user_name)

ayon_server/activities/watchers/watcher_list.py

@@ -1,3 +1,5 @@
+from nxtools import logging
+
 from ayon_server.entities.core.projectlevel import ProjectLevelEntity
 from ayon_server.lib.postgres import Postgres
 from ayon_server.lib.redis import Redis
@@ -25,7 +27,14 @@ async def build_watcher_list(entity: ProjectLevelEntity) -> list[str]:
         ORDER by entity_name ASC
         """
 
-    res = await Postgres.fetch(query, entity.entity_type, entity.id)
+    try:
+        res = await Postgres.fetch(query, entity.entity_type, entity.id)
+    except Postgres.UndefinedTableError:
+        logging.debug(
+            "Unable to get watchers. " f"Project {entity.project_name} no longer exists"
+        )
+        return []
+
     if res:
         watchers = [row["watcher"] for row in res]
 

ayon_server/events/eventstream.py

@@ -1,6 +1,8 @@
 from datetime import datetime
 from typing import Any, Awaitable, Callable, Type
 
+from nxtools import logging
+
 from ayon_server.exceptions import ConstraintViolationException, NotFoundException
 from ayon_server.lib.postgres import Postgres
 from ayon_server.lib.redis import Redis
@@ -132,7 +134,10 @@ async def dispatch(
 
         handlers = cls.hooks.get(event.topic, [])
         for handler in handlers:
-            await handler(event)
+            try:
+                await handler(event)
+            except Exception as e:
+                logging.debug(f"Error in event handler: {e}")
 
         return event.id