added user_notifications base

Author: Andrei Neagu

services/web/server/src/simcore_service_webserver/notifications/_rabbitmq_exclusive_queue_consumers.py

@@ -7,8 +7,10 @@
 from models_library.rabbitmq_messages import (
     EventRabbitMessage,
     LoggerRabbitMessage,
+    ProgressMessageMixin,
     ProgressRabbitMessageNode,
     ProgressRabbitMessageProject,
+    ProgressRabbitMessageWorkerJob,
     ProgressType,
     WalletCreditsMessage,
 )
@@ -29,7 +31,11 @@
     send_message_to_standard_group,
     send_message_to_user,
 )
-from ..socketio.models import WebSocketNodeProgress, WebSocketProjectProgress
+from ..socketio.models import (
+    WebSocketNodeProgress,
+    WebSocketProjectProgress,
+    WebSocketWorkerJobProgress,
+)
 from ..wallets import api as wallets_api
 from ._rabbitmq_consumers_common import SubcribeArgumentsTuple, subscribe_to_rabbitmq
 
@@ -65,20 +71,29 @@ async def _convert_to_node_update_event(
 
 
 async def _progress_message_parser(app: web.Application, data: bytes) -> bool:
-    rabbit_message: ProgressRabbitMessageNode | ProgressRabbitMessageProject = (
-        TypeAdapter(
-            ProgressRabbitMessageNode | ProgressRabbitMessageProject
-        ).validate_json(data)
+    rabbit_message: (
+        ProgressRabbitMessageNode
+        | ProgressRabbitMessageProject
+        | ProgressRabbitMessageWorkerJob
+    ) = TypeAdapter(
+        ProgressRabbitMessageNode
+        | ProgressRabbitMessageProject
+        | ProgressRabbitMessageWorkerJob
+    ).validate_json(
+        data
     )
+
     message: SocketMessageDict | None = None
     if isinstance(rabbit_message, ProgressRabbitMessageProject):
         message = WebSocketProjectProgress.from_rabbit_message(
             rabbit_message
         ).to_socket_dict()
-
+    elif isinstance(rabbit_message, ProgressRabbitMessageWorkerJob):
+        message = WebSocketWorkerJobProgress.from_rabbit_message(
+            rabbit_message
+        ).to_socket_dict()
     elif rabbit_message.progress_type is ProgressType.COMPUTATION_RUNNING:
         message = await _convert_to_node_update_event(app, rabbit_message)
-
     else:
         message = WebSocketNodeProgress.from_rabbit_message(
             rabbit_message
@@ -156,7 +171,7 @@ async def _osparc_credits_message_parser(app: web.Application, data: bytes) -> b
         {"topics": []},
     ),
     SubcribeArgumentsTuple(
-        ProgressRabbitMessageNode.get_channel_name(),
+        ProgressMessageMixin.get_channel_name(),
         _progress_message_parser,
         {"topics": []},
     ),

services/web/server/src/simcore_service_webserver/notifications/user_notifications.py

@@ -0,0 +1,42 @@
+import logging
+from typing import Final
+
+from aiohttp import web
+from models_library.rabbitmq_messages import (
+    ProgressRabbitMessageWorkerJob,
+    RabbitMessageBase,
+)
+from models_library.users import UserID
+from servicelib.logging_utils import log_catch
+from servicelib.rabbitmq import RabbitMQClient
+
+from ..rabbitmq import get_rabbitmq_client
+
+_logger = logging.getLogger(__name__)
+
+
+_SUBSCRIBABLE_EXCHANGES: Final[list[type[RabbitMessageBase]]] = [
+    ProgressRabbitMessageWorkerJob,
+]
+
+
+async def subscribe(app: web.Application, user_id: UserID) -> None:
+    rabbit_client: RabbitMQClient = get_rabbitmq_client(app)
+
+    for exchange in _SUBSCRIBABLE_EXCHANGES:
+        exchange_name = exchange.get_channel_name()
+        await rabbit_client.add_topics(exchange_name, topics=[f"{user_id}.*"])
+
+
+async def unsubscribe(app: web.Application, user_id: UserID) -> None:
+    rabbit_client: RabbitMQClient = get_rabbitmq_client(app)
+    for exchange in _SUBSCRIBABLE_EXCHANGES:
+        exchange_name = exchange.get_channel_name()
+        with log_catch(_logger, reraise=False):
+            # NOTE: in case something bad happenned with the connection to the RabbitMQ server
+            # such as a network disconnection. this call can fail.
+            await rabbit_client.remove_topics(exchange_name, topics=[f"{user_id}"])
+
+
+# TODO: subscribe when logging in
+# TODO: unsubscribe when logging out

services/web/server/src/simcore_service_webserver/socketio/models.py

@@ -1,12 +1,13 @@
 from abc import abstractmethod
-from typing import Literal
+from typing import Literal, Self
 
 from models_library.progress_bar import ProgressReport
 from models_library.projects import ProjectID
 from models_library.projects_nodes_io import NodeID
 from models_library.rabbitmq_messages import (
     ProgressRabbitMessageNode,
     ProgressRabbitMessageProject,
+    ProgressRabbitMessageWorkerJob,
     ProgressType,
 )
 from models_library.socketio import SocketMessageDict
@@ -56,9 +57,7 @@ class WebSocketProjectProgress(
     event_type: Literal["projectProgress"] = "projectProgress"
 
     @classmethod
-    def from_rabbit_message(
-        cls, message: ProgressRabbitMessageProject
-    ) -> "WebSocketProjectProgress":
+    def from_rabbit_message(cls, message: ProgressRabbitMessageProject) -> Self:
         return cls.model_construct(
             user_id=message.user_id,
             project_id=message.project_id,
@@ -83,9 +82,7 @@ class WebSocketNodeProgress(
     event_type: Literal["nodeProgress"] = "nodeProgress"
 
     @classmethod
-    def from_rabbit_message(
-        cls, message: ProgressRabbitMessageNode
-    ) -> "WebSocketNodeProgress":
+    def from_rabbit_message(cls, message: ProgressRabbitMessageNode) -> Self:
         return cls.model_construct(
             user_id=message.user_id,
             project_id=message.project_id,
@@ -99,3 +96,25 @@ def to_socket_dict(self) -> SocketMessageDict:
             event_type=self.event_type,
             data=jsonable_encoder(self, exclude={"event_type"}),
         )
+
+
+class WebSocketWorkerJobProgress(
+    _WebSocketUserMixin,
+    _WebSocketProgressMixin,
+    WebSocketMessageBase,
+):
+    event_type: Literal["workerJobProgress"] = "workerJobProgress"
+
+    @classmethod
+    def from_rabbit_message(cls, message: ProgressRabbitMessageWorkerJob) -> Self:
+        return cls.model_construct(
+            user_id=message.user_id,
+            progress_type=message.progress_type,
+            progress_report=message.report,
+        )
+
+    def to_socket_dict(self) -> SocketMessageDict:
+        return SocketMessageDict(
+            event_type=self.event_type,
+            data=jsonable_encoder(self, exclude={"event_type"}),
+        )

services/web/server/tests/unit/isolated/notifications/test_rabbitmq_consumers.py

@@ -12,6 +12,7 @@
 from models_library.rabbitmq_messages import (
     ProgressRabbitMessageNode,
     ProgressRabbitMessageProject,
+    ProgressRabbitMessageWorkerJob,
     ProgressType,
 )
 from models_library.socketio import SocketMessageDict
@@ -75,6 +76,28 @@
             ),
             id="project_progress",
         ),
+        pytest.param(
+            ProgressRabbitMessageWorkerJob(
+                user_id=123,
+                progress_type=ProgressType.WORKER_JOB_EXPORTING,
+                report=ProgressReport(actual_value=0.4, total=1),
+            ).model_dump_json(),
+            SocketMessageDict(
+                event_type=WebSocketNodeProgress.get_event_type(),
+                data={
+                    "user_id": 123,
+                    "progress_type": ProgressType.WORKER_JOB_EXPORTING.value,
+                    "progress_report": {
+                        "actual_value": 0.4,
+                        "attempt": 0,
+                        "total": 1.0,
+                        "unit": None,
+                        "message": None,
+                    },
+                },
+            ),
+            id="worker_job_progress",
+        ),
     ],
 )
 async def test_regression_progress_message_parser(