WIP

Author: pcrespov

services/payments/src/simcore_service_payments/core/application.py

@@ -1,5 +1,6 @@
 from fastapi import FastAPI
 from servicelib.fastapi.openapi import override_fastapi_openapi_method
+from simcore_service_payments.services.socketio import setup_socketio
 
 from .._meta import (
     API_VERSION,
@@ -54,6 +55,7 @@ def create_app(settings: ApplicationSettings | None = None) -> FastAPI:
 
     # Listening to Rabbitmq
     setup_auto_recharge_listener(app)
+    setup_socketio(app)
 
     # ERROR HANDLERS
     # ... add here ...

services/payments/src/simcore_service_payments/services/socketio.py

@@ -1,14 +1,25 @@
 import logging
+from collections.abc import Sequence
+from typing import Any, Final, TypedDict
 
 import socketio
 from fastapi import FastAPI
+from fastapi.encoders import jsonable_encoder
+from models_library.api_schemas_webserver.wallets import PaymentTransaction
+from models_library.users import UserID
+from servicelib.json_serialization import json_dumps
+from servicelib.utils import logged_gather
 from settings_library.rabbit import RabbitSettings
 
 from .rabbitmq import get_rabbitmq_settings
 
 _logger = logging.getLogger(__name__)
 
 
+SOCKET_IO_PAYMENT_COMPLETED_EVENT: Final[str] = "paymentCompleted"
+SOCKET_IO_PAYMENT_METHOD_ACKED_EVENT: Final[str] = "paymentMethodAcknoledged"
+
+
 def setup_socketio(app: FastAPI):
     settings: RabbitSettings = get_rabbitmq_settings(app)
 
@@ -31,5 +42,53 @@ async def _on_shutdown() -> None:
     app.add_event_handler("shutdown", _on_shutdown)
 
 
-async def emit_to_frontend(app: FastAPI, event_name: str, data: dict, to=None):
+async def emit_to_frontend(
+    app: FastAPI, event_name: str, data: dict, to: str | None = None
+):
+
+    # Send messages to clients from external processes, such as Celery workers or auxiliary scripts.
     return await app.state.external_sio.emit(event_name, data=data, to=to)
+
+
+async def notify_payment_completed(
+    app: FastAPI,
+    *,
+    user_id: UserID,
+    payment: PaymentTransaction,
+):
+    assert payment.completed_at is not None  # nosec
+
+    messages: list[SocketMessageDict] = [
+        {
+            "event_type": SOCKET_IO_PAYMENT_COMPLETED_EVENT,
+            "data": jsonable_encoder(payment, by_alias=True),
+        }
+    ]
+    await send_messages(app, user_id, messages)
+
+
+class SocketMessageDict(TypedDict):
+    event_type: str
+    data: dict[str, Any]
+
+
+async def send_messages(
+    app: FastAPI, user_id: UserID, messages: Sequence[SocketMessageDict]
+) -> None:
+
+    sio = app.state.external_sio
+
+    socket_ids: list[str] = []
+    # with managed_resource(user_id, None, app) as user_session:
+    #     socket_ids = await user_session.find_socket_ids()
+
+    await logged_gather(
+        *(
+            sio.emit(message["event_type"], data=json_dumps(message["data"]), room=sid)
+            for message in messages
+            for sid in socket_ids
+        ),
+        reraise=False,
+        log=_logger,
+        max_concurrency=100,
+    )

services/payments/tests/unit/test_services_socketio.py

@@ -1,4 +1,13 @@
+# pylint: disable=redefined-outer-name
+# pylint: disable=unused-argument
+# pylint: disable=unused-variable
+# pylint: disable=too-many-arguments
+
+
+import asyncio
 from collections.abc import Callable
+from typing import Any
+from unittest.mock import AsyncMock
 
 import pytest
 import socketio
@@ -9,6 +18,7 @@
 from pytest_simcore.helpers.utils_envs import setenvs_from_dict
 from settings_library.rabbit import RabbitSettings
 from simcore_service_payments.services.socketio import (
+    SOCKET_IO_PAYMENT_COMPLETED_EVENT,
     emit_to_frontend,
     get_rabbitmq_settings,
 )
@@ -24,6 +34,7 @@
 def app_environment(
     monkeypatch: pytest.MonkeyPatch,
     app_environment: EnvVarsDict,
+    with_disabled_postgres: None,
     rabbit_env_vars_dict: EnvVarsDict,  # rabbitMQ settings from 'rabbit' service
 ):
     # set environs
@@ -44,7 +55,7 @@ async def test_socketio_setup():
 
 
 @pytest.fixture
-async def socketio_aiohttp_server(app: FastAPI, aiohttp_server: Callable) -> TestServer:
+async def socketio_server(app: FastAPI, aiohttp_server: Callable) -> TestServer:
     """
     this emulates the webserver setup: socketio server with
     an aiopika manager that attaches an aiohttp web app
@@ -60,35 +71,81 @@ async def socketio_aiohttp_server(app: FastAPI, aiohttp_server: Callable) -> Tes
         client_manager=server_manager,
     )
 
-    @sio_server.event()
-    async def connect(sid, environ):
-        ...
+    @sio_server.event
+    async def connect(sid: str, environ):
+        print("connecting", sid)
+
+    @sio_server.on(SOCKET_IO_PAYMENT_COMPLETED_EVENT)
+    async def on_payment(sid, data):
+        print(sid, Any)
+
+    @sio_server.event
+    async def disconnect(sid: str):
+        print("disconnecting", sid)
 
     sio_server.attach(aiohttp_app)
 
+    # starts server
     return await aiohttp_server(aiohttp_app)
 
 
-async def test_emit_socketio_event_to_front_end(
-    app: FastAPI, socketio_aiohttp_server: TestServer
+@pytest.fixture
+async def create_sio_client(socketio_server: TestServer):
+    server_url = socketio_server.make_url("/")
+    _clients = []
+
+    async def _():
+        cli = socketio.AsyncClient(
+            logger=True,
+            engineio_logger=True,
+        )
+
+        # https://python-socketio.readthedocs.io/en/stable/client.html#connecting-to-a-server
+        # Allows WebSocket transport and disconnect HTTP long-polling
+        await cli.connect(f"{server_url}", transports=["websocket"])
+
+        _clients.append(cli)
+
+        return cli
+
+    yield _
+
+    for client in _clients:
+        await client.disconnect()
+
+
+async def test_emit_message_as_external_process_to_frontend_client(
+    app: FastAPI, create_sio_client: Callable
 ):
-    server_url = socketio_aiohttp_server.make_url("/")
+    """
+    front-end  -> socketio client (many different clients)
+    webserver  -> socketio server (one/more replicas)
+    payments   -> Sends messages to clients from external processes (one/more replicas)
+    """
 
-    # create a client
-    async with socketio.AsyncSimpleClient(logger=True, engineio_logger=True) as sio:
+    # emulates front-end receiving message
+    client_1: socketio.AsyncClient = await create_sio_client()
 
-        # https://python-socketio.readthedocs.io/en/stable/client.html#connecting-to-a-server
-        # connect to a server
-        await sio.connect(server_url, transports=["websocket"])
-        session_client_id = sio.sid
+    @client_1.on(SOCKET_IO_PAYMENT_COMPLETED_EVENT)
+    async def on_event(data):
+        print("client1", data)
 
-        # emit from external
-        await emit_to_frontend(
-            app, event_name="event", data={"foo": "bar"}, to=session_client_id
-        )
+    on_event_spy = AsyncMock(wraps=on_event)
+
+    await client_1.emit(SOCKET_IO_PAYMENT_COMPLETED_EVENT, data="hoi1")
+
+    # TODO: better to do this from a different process??
+    # emit from external process
+    await emit_to_frontend(
+        app,
+        event_name=SOCKET_IO_PAYMENT_COMPLETED_EVENT,
+        data={"foo": "bar"},
+        # to=client_1.sid,
+    )
+
+    await client_1.emit(SOCKET_IO_PAYMENT_COMPLETED_EVENT, data="hoi2")
 
-        # client receives it
-        event: list = await sio.receive(timeout=5)
-        event_name, *event_kwargs = event
+    await client_1.sleep(1)
+    await asyncio.sleep(1)
 
-        assert event_name == "event"
+    on_event_spy.assert_called()

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

@@ -18,6 +18,10 @@
 
 _logger = logging.getLogger(__name__)
 
+
+#
+# List of socket-io event names
+#
 SOCKET_IO_EVENT: Final[str] = "event"
 SOCKET_IO_HEARTBEAT_EVENT: Final[str] = "set_heartbeat_emit_interval"
 SOCKET_IO_LOG_EVENT: Final[str] = "logger"