chore: streamlined sse service, no sse bus

Author: HardMax71

backend/app/core/providers.py

@@ -63,7 +63,7 @@
 from app.services.runtime_settings import RuntimeSettingsLoader
 from app.services.saga import SagaOrchestrator, SagaService
 from app.services.saved_script_service import SavedScriptService
-from app.services.sse import SSERedisBus, SSEService
+from app.services.sse import SSEService
 from app.services.user_settings_service import UserSettingsService
 from app.settings import Settings
 
@@ -131,6 +131,7 @@ async def get_redis_client(
             decode_responses=settings.REDIS_DECODE_RESPONSES,
             socket_connect_timeout=5,
             socket_timeout=5,
+            socket_keepalive=True,
         )
         # Test connection
         await client.ping()  # type: ignore[misc]  # redis-py returns Awaitable[bool] | bool
@@ -351,22 +352,19 @@ class SSEProvider(Provider):
     scope = Scope.APP
 
     @provide
-    def get_sse_redis_bus(
+    def get_sse_service(
         self,
         redis_client: redis.Redis,
+        execution_repository: ExecutionRepository,
         logger: structlog.stdlib.BoundLogger,
         connection_metrics: ConnectionMetrics,
-    ) -> SSERedisBus:
-        return SSERedisBus(redis_client, logger, connection_metrics)
-
-    @provide
-    def get_sse_service(
-            self,
-            bus: SSERedisBus,
-            execution_repository: ExecutionRepository,
-            logger: structlog.stdlib.BoundLogger,
     ) -> SSEService:
-        return SSEService(bus=bus, execution_repository=execution_repository, logger=logger)
+        return SSEService(
+            redis_client=redis_client,
+            execution_repository=execution_repository,
+            logger=logger,
+            connection_metrics=connection_metrics,
+        )
 
 
 class AuthProvider(Provider):
@@ -483,15 +481,15 @@ def get_notification_service(
             self,
             notification_repository: NotificationRepository,
             kafka_event_service: KafkaEventService,
-            sse_redis_bus: SSERedisBus,
+            sse_service: SSEService,
             settings: Settings,
             logger: structlog.stdlib.BoundLogger,
             notification_metrics: NotificationMetrics,
     ) -> NotificationService:
         return NotificationService(
             notification_repository=notification_repository,
             event_service=kafka_event_service,
-            sse_bus=sse_redis_bus,
+            sse_service=sse_service,
             settings=settings,
             logger=logger,
             notification_metrics=notification_metrics,
@@ -731,12 +729,12 @@ def get_event_replay_service(
             kafka_producer: UnifiedProducer,
             replay_metrics: ReplayMetrics,
             logger: structlog.stdlib.BoundLogger,
-            sse_bus: SSERedisBus,
+            sse_service: SSEService,
     ) -> EventReplayService:
         return EventReplayService(
             repository=replay_repository,
             producer=kafka_producer,
             replay_metrics=replay_metrics,
             logger=logger,
-            sse_bus=sse_bus,
+            sse_service=sse_service,
         )

backend/app/events/handlers.py

@@ -25,7 +25,7 @@
 from app.services.notification_service import NotificationService
 from app.services.result_processor import ResultProcessor
 from app.services.saga import SagaOrchestrator
-from app.services.sse import SSERedisBus
+from app.services.sse import SSEService
 from app.settings import Settings
 
 _sse_field_names: frozenset[str] = frozenset(f.name for f in dataclasses.fields(SSEExecutionEventData))
@@ -261,14 +261,14 @@ def register_sse_subscriber(broker: KafkaBroker, settings: Settings) -> None:
     )
     async def on_sse_event(
             body: DomainEvent,
-            sse_bus: FromDishka[SSERedisBus],
+            sse_service: FromDishka[SSEService],
     ) -> None:
         execution_id = getattr(body, "execution_id", None)
         if execution_id:
             sse_data = SSEExecutionEventData(**{
                 k: v for k, v in body.model_dump().items() if k in _sse_field_names
             })
-            await sse_bus.publish_event(execution_id, sse_data)
+            await sse_service.publish_event(execution_id, sse_data)
 
 
 def register_notification_subscriber(broker: KafkaBroker) -> None:

backend/app/services/event_replay/replay_service.py

@@ -27,7 +27,7 @@
 )
 from app.domain.sse import DomainReplaySSEPayload
 from app.events import UnifiedProducer
-from app.services.sse.redis_bus import SSERedisBus
+from app.services.sse import SSEService
 
 
 class EventReplayService:
@@ -37,7 +37,7 @@ def __init__(
         producer: UnifiedProducer,
         replay_metrics: ReplayMetrics,
         logger: structlog.stdlib.BoundLogger,
-        sse_bus: SSERedisBus,
+        sse_service: SSEService,
     ) -> None:
         self._sessions: dict[str, ReplaySessionState] = {}
         self._schedulers: dict[str, AsyncIOScheduler] = {}
@@ -49,7 +49,7 @@ def __init__(
         self.logger = logger
         self._file_locks: dict[str, asyncio.Lock] = {}
         self._metrics = replay_metrics
-        self._sse_bus = sse_bus
+        self._sse_service = sse_service
 
     async def create_session_from_config(self, config: ReplayConfig) -> ReplayOperationResult:
         try:
@@ -429,6 +429,6 @@ async def _publish_replay_status(self, session: ReplaySessionState) -> None:
                 completed_at=session.completed_at,
                 errors=session.errors,
             )
-            await self._sse_bus.publish_replay_status(session.session_id, payload)
+            await self._sse_service.publish_replay_status(session.session_id, payload)
         except Exception as e:
             self.logger.error("Failed to publish replay status to SSE", error=str(e))

backend/app/services/notification_service.py

@@ -33,7 +33,7 @@
 )
 from app.domain.sse import DomainNotificationSSEPayload
 from app.services.kafka_event_service import KafkaEventService
-from app.services.sse import SSERedisBus
+from app.services.sse import SSEService
 from app.settings import Settings
 
 # Constants
@@ -100,7 +100,7 @@ def __init__(
         self,
         notification_repository: NotificationRepository,
         event_service: KafkaEventService,
-        sse_bus: SSERedisBus,
+        sse_service: SSEService,
         settings: Settings,
         logger: structlog.stdlib.BoundLogger,
         notification_metrics: NotificationMetrics,
@@ -109,7 +109,7 @@ def __init__(
         self.event_service = event_service
         self.metrics = notification_metrics
         self.settings = settings
-        self.sse_bus = sse_bus
+        self.sse_service = sse_service
         self.logger = logger
         self._throttle_cache = ThrottleCache()
 
@@ -578,7 +578,7 @@ async def _publish_notification_sse(self, notification: DomainNotification) -> N
             action_url=notification.action_url,
             created_at=notification.created_at,
         )
-        await self.sse_bus.publish_notification(notification.user_id, payload)
+        await self.sse_service.publish_notification(notification.user_id, payload)
 
     # --8<-- [start:should_skip_notification]
     async def _should_skip_notification(

backend/app/services/sse/__init__.py

@@ -1,4 +1,3 @@
-from app.services.sse.redis_bus import SSERedisBus
 from app.services.sse.sse_service import SSEService
 
-__all__ = ["SSERedisBus", "SSEService"]
+__all__ = ["SSEService"]