chore: removal of nullable fields

backend/app/core/providers.py

@@ -50,7 +50,7 @@
 from app.services.admin import AdminEventsService, AdminSettingsService, AdminUserService
 from app.services.auth_service import AuthService
 from app.services.coordinator.coordinator import ExecutionCoordinator
-from app.services.event_bus import EventBusManager
+from app.services.event_bus import EventBus, EventBusEvent
 from app.services.event_replay.replay_service import EventReplayService
 from app.services.event_service import EventService
 from app.services.execution_service import ExecutionService
@@ -73,7 +73,7 @@
 from app.services.sse.kafka_redis_bridge import SSEKafkaRedisBridge, create_sse_kafka_redis_bridge
 from app.services.sse.redis_bus import SSERedisBus
 from app.services.sse.sse_service import SSEService
-from app.services.sse.sse_shutdown_manager import SSEShutdownManager, create_sse_shutdown_manager
+from app.services.sse.sse_shutdown_manager import SSEShutdownManager
 from app.services.user_settings_service import UserSettingsService
 from app.settings import Settings
 
@@ -230,14 +230,11 @@ async def get_event_store_consumer(
             yield consumer
 
     @provide
-    async def get_event_bus_manager(
+    async def get_event_bus(
             self, settings: Settings, logger: logging.Logger, connection_metrics: ConnectionMetrics
-    ) -> AsyncIterator[EventBusManager]:
-        manager = EventBusManager(settings, logger, connection_metrics)
-        try:
-            yield manager
-        finally:
-            await manager.close()
+    ) -> AsyncIterator[EventBus]:
+        async with EventBus(settings, logger, connection_metrics) as bus:
+            yield bus
 
 
 class KubernetesProvider(Provider):
@@ -398,9 +395,16 @@ async def get_sse_kafka_redis_bridge(
 
     @provide(scope=Scope.REQUEST)
     def get_sse_shutdown_manager(
-            self, logger: logging.Logger, connection_metrics: ConnectionMetrics
+            self,
+            router: SSEKafkaRedisBridge,
+            logger: logging.Logger,
+            connection_metrics: ConnectionMetrics,
     ) -> SSEShutdownManager:
-        return create_sse_shutdown_manager(logger=logger, connection_metrics=connection_metrics)
+        return SSEShutdownManager(
+            router=router,
+            logger=logger,
+            connection_metrics=connection_metrics,
+        )
 
     @provide(scope=Scope.REQUEST)
     def get_sse_service(
@@ -413,7 +417,6 @@ def get_sse_service(
             logger: logging.Logger,
             connection_metrics: ConnectionMetrics,
     ) -> SSEService:
-        shutdown_manager.set_router(router)
         return SSEService(
             repository=sse_repository,
             router=router,
@@ -470,11 +473,21 @@ async def get_user_settings_service(
             self,
             repository: UserSettingsRepository,
             kafka_event_service: KafkaEventService,
-            event_bus_manager: EventBusManager,
+            event_bus: EventBus,
             logger: logging.Logger,
     ) -> UserSettingsService:
-        service = UserSettingsService(repository, kafka_event_service, logger)
-        await service.initialize(event_bus_manager)
+        service = UserSettingsService(repository, kafka_event_service, logger, event_bus)
+
+        # Subscribe to settings update events for cross-instance cache invalidation.
+        # EventBus filters out self-published messages, so this handler only
+        # runs for events from OTHER instances.
+        async def _handle_settings_update(evt: EventBusEvent) -> None:
+            uid = evt.payload.get("user_id")
+            if uid:
+                await service.invalidate_cache(str(uid))
+
+        await event_bus.subscribe("user.settings.updated*", _handle_settings_update)
+
         return service
 
 
@@ -503,27 +516,25 @@ def get_notification_service(
             self,
             notification_repository: NotificationRepository,
             kafka_event_service: KafkaEventService,
-            event_bus_manager: EventBusManager,
+            event_bus: EventBus,
             schema_registry: SchemaRegistryManager,
             sse_redis_bus: SSERedisBus,
             settings: Settings,
             logger: logging.Logger,
             notification_metrics: NotificationMetrics,
             event_metrics: EventMetrics,
     ) -> NotificationService:
-        service = NotificationService(
+        return NotificationService(
             notification_repository=notification_repository,
             event_service=kafka_event_service,
-            event_bus_manager=event_bus_manager,
+            event_bus=event_bus,
             schema_registry_manager=schema_registry,
             sse_bus=sse_redis_bus,
             settings=settings,
             logger=logger,
             notification_metrics=notification_metrics,
             event_metrics=event_metrics,
         )
-        service.initialize()
-        return service
 
     @provide
     def get_grafana_alert_processor(

backend/app/services/event_bus.py

@@ -9,7 +9,6 @@
 
 from aiokafka import AIOKafkaConsumer, AIOKafkaProducer
 from aiokafka.errors import KafkaError
-from fastapi import Request
 from pydantic import BaseModel, ConfigDict
 
 from app.core.lifecycle import LifecycleEnabled
@@ -316,32 +315,3 @@ async def get_statistics(self) -> dict[str, Any]:
             }
 
 
-class EventBusManager:
-    """Manages EventBus lifecycle as a singleton."""
-
-    def __init__(self, settings: Settings, logger: logging.Logger, connection_metrics: ConnectionMetrics) -> None:
-        self.settings = settings
-        self.logger = logger
-        self._connection_metrics = connection_metrics
-        self._event_bus: Optional[EventBus] = None
-        self._lock = asyncio.Lock()
-
-    async def get_event_bus(self) -> EventBus:
-        """Get or create the event bus instance."""
-        async with self._lock:
-            if self._event_bus is None:
-                self._event_bus = EventBus(self.settings, self.logger, self._connection_metrics)
-                await self._event_bus.__aenter__()
-            return self._event_bus
-
-    async def close(self) -> None:
-        """Stop and clean up the event bus."""
-        async with self._lock:
-            if self._event_bus:
-                await self._event_bus.aclose()
-                self._event_bus = None
-
-
-async def get_event_bus(request: Request) -> EventBus:
-    manager: EventBusManager = request.app.state.event_bus_manager
-    return await manager.get_event_bus()

backend/app/services/grafana_alert_processor.py

@@ -1,5 +1,3 @@
-"""Grafana alert processing service."""
-
 import logging
 from typing import Any
 

backend/app/services/notification_service.py

@@ -40,7 +40,7 @@
 from app.events.schema.schema_registry import SchemaRegistryManager
 from app.infrastructure.kafka.mappings import get_topic_for_event
 from app.schemas_pydantic.sse import RedisNotificationMessage
-from app.services.event_bus import EventBusManager
+from app.services.event_bus import EventBus
 from app.services.kafka_event_service import KafkaEventService
 from app.services.sse.redis_bus import SSERedisBus
 from app.settings import Settings
@@ -59,8 +59,6 @@
 class ServiceState(StringEnum):
     """Service lifecycle states."""
 
-    IDLE = auto()
-    INITIALIZING = auto()
     RUNNING = auto()
     STOPPING = auto()
     STOPPED = auto()
@@ -117,7 +115,7 @@ def __init__(
         self,
         notification_repository: NotificationRepository,
         event_service: KafkaEventService,
-        event_bus_manager: EventBusManager,
+        event_bus: EventBus,
         schema_registry_manager: SchemaRegistryManager,
         sse_bus: SSERedisBus,
         settings: Settings,
@@ -127,7 +125,7 @@ def __init__(
     ) -> None:
         self.repository = notification_repository
         self.event_service = event_service
-        self.event_bus_manager = event_bus_manager
+        self.event_bus = event_bus
         self.metrics = notification_metrics
         self._event_metrics = event_metrics
         self.settings = settings
@@ -136,7 +134,7 @@ def __init__(
         self.logger = logger
 
         # State
-        self._state = ServiceState.IDLE
+        self._state = ServiceState.RUNNING
         self._throttle_cache = ThrottleCache()
 
         # Tasks
@@ -146,6 +144,16 @@ def __init__(
         self._dispatcher: EventDispatcher | None = None
         self._consumer_task: asyncio.Task[None] | None = None
 
+        # Channel handlers mapping
+        self._channel_handlers: dict[NotificationChannel, ChannelHandler] = {
+            NotificationChannel.IN_APP: self._send_in_app,
+            NotificationChannel.WEBHOOK: self._send_webhook,
+            NotificationChannel.SLACK: self._send_slack,
+        }
+
+        # Start background processors
+        self._start_background_tasks()
+
         self.logger.info(
             "NotificationService initialized",
             extra={
@@ -155,30 +163,10 @@ def __init__(
             },
         )
 
-        # Channel handlers mapping
-        self._channel_handlers: dict[NotificationChannel, ChannelHandler] = {
-            NotificationChannel.IN_APP: self._send_in_app,
-            NotificationChannel.WEBHOOK: self._send_webhook,
-            NotificationChannel.SLACK: self._send_slack,
-        }
-
     @property
     def state(self) -> ServiceState:
         return self._state
 
-    def initialize(self) -> None:
-        if self._state != ServiceState.IDLE:
-            self.logger.warning(f"Cannot initialize in state: {self._state}")
-            return
-
-        self._state = ServiceState.INITIALIZING
-
-        # Start processors
-        self._state = ServiceState.RUNNING
-        self._start_background_tasks()
-
-        self.logger.info("Notification service initialized (without Kafka consumer)")
-
     async def shutdown(self) -> None:
         """Shutdown notification service."""
         if self._state == ServiceState.STOPPED:
@@ -324,8 +312,7 @@ async def create_notification(
         notification = await self.repository.create_notification(create_data)
 
         # Publish event
-        event_bus = await self.event_bus_manager.get_event_bus()
-        await event_bus.publish(
+        await self.event_bus.publish(
             "notifications.created",
             {
                 "notification_id": str(notification.notification_id),
@@ -694,9 +681,8 @@ async def mark_as_read(self, user_id: str, notification_id: str) -> bool:
         """Mark notification as read."""
         success = await self.repository.mark_as_read(notification_id, user_id)
 
-        event_bus = await self.event_bus_manager.get_event_bus()
         if success:
-            await event_bus.publish(
+            await self.event_bus.publish(
                 "notifications.read",
                 {"notification_id": str(notification_id), "user_id": user_id, "read_at": datetime.now(UTC).isoformat()},
             )
@@ -778,9 +764,8 @@ async def mark_all_as_read(self, user_id: str) -> int:
         """Mark all notifications as read for a user."""
         count = await self.repository.mark_all_as_read(user_id)
 
-        event_bus = await self.event_bus_manager.get_event_bus()
         if count > 0:
-            await event_bus.publish(
+            await self.event_bus.publish(
                 "notifications.all_read", {"user_id": user_id, "count": count, "read_at": datetime.now(UTC).isoformat()}
             )
 

backend/app/services/sse/sse_service.py

@@ -32,14 +32,14 @@ class SSEService:
     }
 
     def __init__(
-        self,
-        repository: SSERepository,
-        router: SSEKafkaRedisBridge,
-        sse_bus: SSERedisBus,
-        shutdown_manager: SSEShutdownManager,
-        settings: Settings,
-        logger: logging.Logger,
-        connection_metrics: ConnectionMetrics,
+            self,
+            repository: SSERepository,
+            router: SSEKafkaRedisBridge,
+            sse_bus: SSERedisBus,
+            shutdown_manager: SSEShutdownManager,
+            settings: Settings,
+            logger: logging.Logger,
+            connection_metrics: ConnectionMetrics,
     ) -> None:
         self.repository = repository
         self.router = router
@@ -48,7 +48,7 @@ def __init__(
         self.settings = settings
         self.logger = logger
         self.metrics = connection_metrics
-        self.heartbeat_interval = getattr(settings, "SSE_HEARTBEAT_INTERVAL", 30)
+        self.heartbeat_interval = settings.SSE_HEARTBEAT_INTERVAL
 
     async def create_execution_stream(self, execution_id: str, user_id: str) -> AsyncGenerator[Dict[str, Any], None]:
         connection_id = f"sse_{execution_id}_{datetime.now(timezone.utc).timestamp()}"
@@ -106,10 +106,10 @@ async def create_execution_stream(self, execution_id: str, user_id: str) -> Asyn
                 self.metrics.record_sse_message_sent("executions", "status")
 
             async for event_data in self._stream_events_redis(
-                execution_id,
-                subscription,
-                shutdown_event,
-                include_heartbeat=False,
+                    execution_id,
+                    subscription,
+                    shutdown_event,
+                    include_heartbeat=False,
             ):
                 yield event_data
 
@@ -120,11 +120,11 @@ async def create_execution_stream(self, execution_id: str, user_id: str) -> Asyn
             self.logger.info("SSE connection closed", extra={"execution_id": execution_id})
 
     async def _stream_events_redis(
-        self,
-        execution_id: str,
-        subscription: Any,
-        shutdown_event: asyncio.Event,
-        include_heartbeat: bool = True,
+            self,
+            execution_id: str,
+            subscription: Any,
+            shutdown_event: asyncio.Event,
+            include_heartbeat: bool = True,
     ) -> AsyncGenerator[Dict[str, Any], None]:
         last_heartbeat = datetime.now(timezone.utc)
         while True: