chore: removed deadcode, extracted copypasted code into separate funcs, updated docs

Author: HardMax71

backend/app/db/docs/saga.py

@@ -11,10 +11,7 @@
 
 
 class SagaDocument(Document):
-    """Domain model for saga stored in database.
-
-    Copied from Saga/SagaInstance dataclass.
-    """
+    """Domain model for saga stored in database."""
 
     saga_id: Indexed(str, unique=True) = Field(default_factory=lambda: str(uuid4()))  # type: ignore[valid-type]
     saga_name: Indexed(str)  # type: ignore[valid-type]

backend/app/domain/events/typed.py

@@ -58,11 +58,12 @@ class BaseEvent(BaseModel):
     metadata: EventMetadata
 
 
-# --- Execution Events ---
+# --- Execution Spec (shared fields between ExecutionRequestedEvent and CreatePodCommandEvent) ---
 
 
-class ExecutionRequestedEvent(BaseEvent):
-    event_type: Literal[EventType.EXECUTION_REQUESTED] = EventType.EXECUTION_REQUESTED
+class ExecutionSpec(BaseModel):
+    """Shared execution specification fields (mixin for ExecutionRequestedEvent and CreatePodCommandEvent)."""
+
     execution_id: str
     script: str
     language: str
@@ -78,6 +79,13 @@ class ExecutionRequestedEvent(BaseEvent):
     priority: QueuePriority = QueuePriority.NORMAL
 
 
+# --- Execution Events ---
+
+
+class ExecutionRequestedEvent(BaseEvent, ExecutionSpec):
+    event_type: Literal[EventType.EXECUTION_REQUESTED] = EventType.EXECUTION_REQUESTED
+
+
 class ExecutionAcceptedEvent(BaseEvent):
     event_type: Literal[EventType.EXECUTION_ACCEPTED] = EventType.EXECUTION_ACCEPTED
     execution_id: str
@@ -413,22 +421,10 @@ class SagaCompensatedEvent(BaseEvent):
 # --- Saga Command Events ---
 
 
-class CreatePodCommandEvent(BaseEvent):
+class CreatePodCommandEvent(BaseEvent, ExecutionSpec):
     event_type: Literal[EventType.CREATE_POD_COMMAND] = EventType.CREATE_POD_COMMAND
     saga_id: str
-    execution_id: str
-    script: str
-    language: str
-    language_version: str
-    runtime_image: str
     runtime_command: list[str] = Field(default_factory=list)
-    runtime_filename: str
-    timeout_seconds: int
-    cpu_limit: str
-    memory_limit: str
-    cpu_request: str
-    memory_request: str
-    priority: QueuePriority = QueuePriority.NORMAL
 
 
 class DeletePodCommandEvent(BaseEvent):

backend/app/domain/saga/__init__.py

@@ -13,7 +13,6 @@
     SagaConfig,
     SagaContextData,
     SagaFilter,
-    SagaInstance,
     SagaListResult,
     SagaQuery,
 )
@@ -25,7 +24,6 @@
     "SagaCancellationResult",
     "SagaConfig",
     "SagaContextData",
-    "SagaInstance",
     "SagaFilter",
     "SagaListResult",
     "SagaQuery",

backend/app/domain/saga/models.py

@@ -92,25 +92,6 @@ class SagaConfig:
     publish_commands: bool = True
 
 
-@dataclass
-class SagaInstance:
-    """Runtime instance of a saga execution (domain)."""
-
-    saga_name: str
-    execution_id: str
-    state: SagaState = SagaState.CREATED
-    saga_id: str = field(default_factory=lambda: str(uuid4()))
-    current_step: str | None = None
-    completed_steps: list[str] = field(default_factory=list)
-    compensated_steps: list[str] = field(default_factory=list)
-    context_data: SagaContextData = field(default_factory=SagaContextData)
-    error_message: str | None = None
-    created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
-    updated_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
-    completed_at: datetime | None = None
-    retry_count: int = 0
-
-
 @dataclass
 class SagaCancellationResult:
     """Domain result for saga cancellation operations."""

backend/app/services/event_replay/replay_service.py

@@ -183,19 +183,45 @@ async def cleanup_old_sessions(self, older_than_hours: int = 24) -> CleanupResul
         self.logger.info("Cleaned up old replay sessions", removed_count=total_removed)
         return CleanupResult(removed_sessions=total_removed, message=f"Removed {total_removed} old sessions")
 
+    async def _load_next_event(self, session: ReplaySessionState) -> DomainEvent | None:
+        """Pop the next event from the buffer, loading a new batch if needed."""
+        event = self._pop_next_event(session.session_id)
+        if event is not None:
+            return event
+        if not await self._load_next_batch(session.session_id):
+            return None
+        return self._pop_next_event(session.session_id)
+
+    def _calculate_replay_delay(self, session: ReplaySessionState) -> float:
+        """Calculate the delay before dispatching the next event based on speed multiplier."""
+        next_event = self._peek_next_event(session.session_id)
+        if next_event and session.last_event_at and session.config.speed_multiplier < 100:
+            time_diff = (next_event.timestamp - session.last_event_at).total_seconds()
+            return max(time_diff / session.config.speed_multiplier, 0)
+        return 0.0
+
+    def _reschedule_dispatch(self, session: ReplaySessionState, delay: float) -> None:
+        """Schedule the next _dispatch_next call if the session is still running."""
+        scheduler = self._schedulers.get(session.session_id)
+        if scheduler and scheduler.running and session.status == ReplayStatus.RUNNING:
+            scheduler.add_job(
+                self._dispatch_next,
+                trigger="date",
+                run_date=datetime.now(timezone.utc) + timedelta(seconds=delay),
+                args=[session],
+                id=f"dispatch_{session.session_id}",
+                replace_existing=True,
+                misfire_grace_time=None,
+            )
+
     async def _dispatch_next(self, session: ReplaySessionState) -> None:
         if session.status != ReplayStatus.RUNNING:
             return
 
-        event = self._pop_next_event(session.session_id)
+        event = await self._load_next_event(session)
         if event is None:
-            if not await self._load_next_batch(session.session_id):
-                await self._finalize_session(session, ReplayStatus.COMPLETED)
-                return
-            event = self._pop_next_event(session.session_id)
-            if event is None:
-                await self._finalize_session(session, ReplayStatus.COMPLETED)
-                return
+            await self._finalize_session(session, ReplayStatus.COMPLETED)
+            return
 
         buf = self._event_buffers.get(session.session_id, [])
         idx = self._buffer_indices.get(session.session_id, 0)
@@ -229,26 +255,10 @@ async def _dispatch_next(self, session: ReplaySessionState) -> None:
         session.last_event_at = event.timestamp
         await self._update_session_in_db(session)
 
-        next_event = self._peek_next_event(session.session_id)
-        delay = 0.0
-        if next_event and session.last_event_at and session.config.speed_multiplier < 100:
-            time_diff = (next_event.timestamp - session.last_event_at).total_seconds()
-            delay = max(time_diff / session.config.speed_multiplier, 0)
-
+        delay = self._calculate_replay_delay(session)
         if delay > 0:
             self._metrics.record_delay_applied(delay)
-
-        scheduler = self._schedulers.get(session.session_id)
-        if scheduler and scheduler.running and session.status == ReplayStatus.RUNNING:
-            scheduler.add_job(
-                self._dispatch_next,
-                trigger="date",
-                run_date=datetime.now(timezone.utc) + timedelta(seconds=delay),
-                args=[session],
-                id=f"dispatch_{session.session_id}",
-                replace_existing=True,
-                misfire_grace_time=None,
-            )
+        self._reschedule_dispatch(session, delay)
 
     def _pop_next_event(self, session_id: str) -> DomainEvent | None:
         idx = self._buffer_indices.get(session_id, 0)

backend/app/services/notification_service.py

@@ -122,6 +122,38 @@ def __init__(
         }
         # --8<-- [end:channel_handlers]
 
+    def _validate_scheduled_time(self, scheduled_for: datetime) -> None:
+        """Validate that scheduled_for is in the future and within the max schedule window."""
+        if scheduled_for < datetime.now(UTC):
+            raise NotificationValidationError("scheduled_for must be in the future")
+        max_days = self.settings.NOTIF_MAX_SCHEDULE_DAYS
+        max_schedule = datetime.now(UTC) + timedelta(days=max_days)
+        if scheduled_for > max_schedule:
+            raise NotificationValidationError(f"scheduled_for cannot exceed {max_days} days from now")
+
+    async def _check_throttle(self, user_id: str, severity: NotificationSeverity, source: str) -> None:
+        """Check throttle and raise NotificationThrottledError if rate limit exceeded."""
+        if self.settings.ENVIRONMENT == "test":
+            return
+        throttled = await self._throttle_cache.check_throttle(
+            user_id,
+            severity,
+            window_hours=self.settings.NOTIF_THROTTLE_WINDOW_HOURS,
+            max_per_hour=self.settings.NOTIF_THROTTLE_MAX_PER_HOUR,
+        )
+        if throttled:
+            self.logger.warning(
+                f"Notification rate limit exceeded for user {user_id}. "
+                f"Max {self.settings.NOTIF_THROTTLE_MAX_PER_HOUR} "
+                f"per {self.settings.NOTIF_THROTTLE_WINDOW_HOURS} hour(s)"
+            )
+            self.metrics.record_notification_throttled(source)
+            raise NotificationThrottledError(
+                user_id,
+                self.settings.NOTIF_THROTTLE_MAX_PER_HOUR,
+                self.settings.NOTIF_THROTTLE_WINDOW_HOURS,
+            )
+
     async def create_notification(
         self,
         user_id: str,
@@ -137,14 +169,7 @@ async def create_notification(
         if not tags:
             raise NotificationValidationError("tags must be a non-empty list")
         if scheduled_for is not None:
-            if scheduled_for < datetime.now(UTC):
-                raise NotificationValidationError("scheduled_for must be in the future")
-            max_days = self.settings.NOTIF_MAX_SCHEDULE_DAYS
-            max_schedule = datetime.now(UTC) + timedelta(days=max_days)
-            if scheduled_for > max_schedule:
-                raise NotificationValidationError(
-                    f"scheduled_for cannot exceed {max_days} days from now"
-                )
+            self._validate_scheduled_time(scheduled_for)
         self.logger.info(
             f"Creating notification for user {user_id}",
             user_id=user_id,
@@ -154,25 +179,7 @@ async def create_notification(
             scheduled=scheduled_for is not None,
         )
 
-        # Check throttling
-        if self.settings.ENVIRONMENT != "test" and await self._throttle_cache.check_throttle(
-            user_id,
-            severity,
-            window_hours=self.settings.NOTIF_THROTTLE_WINDOW_HOURS,
-            max_per_hour=self.settings.NOTIF_THROTTLE_MAX_PER_HOUR,
-        ):
-            error_msg = (
-                f"Notification rate limit exceeded for user {user_id}. "
-                f"Max {self.settings.NOTIF_THROTTLE_MAX_PER_HOUR} "
-                f"per {self.settings.NOTIF_THROTTLE_WINDOW_HOURS} hour(s)"
-            )
-            self.logger.warning(error_msg)
-            self.metrics.record_notification_throttled("general")
-            raise NotificationThrottledError(
-                user_id,
-                self.settings.NOTIF_THROTTLE_MAX_PER_HOUR,
-                self.settings.NOTIF_THROTTLE_WINDOW_HOURS,
-            )
+        await self._check_throttle(user_id, severity, "general")
 
         # Create notification
         create_data = DomainNotificationCreate(
@@ -290,13 +297,9 @@ async def _create_system_for_user(
     ) -> str:
         try:
             if not cfg.throttle_exempt:
-                throttled = await self._throttle_cache.check_throttle(
-                    user_id,
-                    cfg.severity,
-                    window_hours=self.settings.NOTIF_THROTTLE_WINDOW_HOURS,
-                    max_per_hour=self.settings.NOTIF_THROTTLE_MAX_PER_HOUR,
-                )
-                if throttled:
+                try:
+                    await self._check_throttle(user_id, cfg.severity, "system")
+                except NotificationThrottledError:
                     return "throttled"
 
             await self.create_notification(

backend/app/services/saga/__init__.py

@@ -1,5 +1,5 @@
 from app.domain.enums import SagaState
-from app.domain.saga import SagaConfig, SagaInstance
+from app.domain.saga import SagaConfig
 from app.services.saga.execution_saga import (
     AllocateResourcesStep,
     CreatePodStep,
@@ -17,7 +17,6 @@
     "SagaService",
     "SagaConfig",
     "SagaState",
-    "SagaInstance",
     "SagaContext",
     "SagaStep",
     "CompensationStep",

backend/workers/bootstrap.py

@@ -0,0 +1,83 @@
+import asyncio
+from collections.abc import Awaitable, Callable
+from typing import Any
+
+import structlog
+from app.core.logging import setup_log_exporter, setup_logger
+from app.db.docs import ALL_DOCUMENTS
+from app.settings import Settings
+from beanie import init_beanie
+from dishka import AsyncContainer
+from dishka.integrations.faststream import setup_dishka
+from faststream import FastStream
+from faststream.kafka import KafkaBroker
+from pymongo import AsyncMongoClient
+
+
+def run_worker(
+    worker_name: str,
+    config_override: str,
+    container_factory: Callable[[Settings], AsyncContainer],
+    register_handlers: Callable[[KafkaBroker], None] | None = None,
+    on_startup: Callable[[AsyncContainer, KafkaBroker, structlog.stdlib.BoundLogger], Awaitable[None]] | None = None,
+    on_shutdown: Callable[[], Awaitable[None]] | None = None,
+) -> None:
+    """Boot a worker with standardised init sequence.
+
+    Parameters
+    ----------
+    worker_name:
+        Human-readable name used in log messages.
+    config_override:
+        TOML filename passed to ``Settings(override_path=...)``.
+    container_factory:
+        Dishka container factory — receives ``Settings``, returns ``AsyncContainer``.
+    register_handlers:
+        Optional callback to register ``@broker.subscriber`` handlers **before**
+        ``setup_dishka`` is called (required for subscriber auto-injection).
+    on_startup:
+        Optional async callback executed **after** the broker is ready.
+        Receives ``(container, broker, logger)`` so it can resolve services
+        and wire up APScheduler jobs, K8s setup, etc.
+    on_shutdown:
+        Optional async callback executed on FastStream shutdown **before**
+        the container is closed.  Use for scheduler teardown etc.
+    """
+    settings = Settings(override_path=config_override)
+
+    logger = setup_logger(settings.LOG_LEVEL)
+    setup_log_exporter(settings, logger)
+
+    logger.info(f"Starting {worker_name}...")
+
+    async def _run() -> None:
+        client: AsyncMongoClient[dict[str, Any]] = AsyncMongoClient(settings.MONGODB_URL, tz_aware=True)
+        await init_beanie(
+            database=client.get_default_database(default=settings.DATABASE_NAME),
+            document_models=ALL_DOCUMENTS,
+        )
+        logger.info("MongoDB initialized via Beanie")
+
+        container = container_factory(settings)
+
+        broker: KafkaBroker = await container.get(KafkaBroker)
+
+        if register_handlers is not None:
+            register_handlers(broker)
+        setup_dishka(container, broker=broker, auto_inject=True)
+
+        startup_hooks: list[Callable[[], Awaitable[None]]] = []
+        shutdown_hooks: list[Callable[[], Awaitable[None]]] = []
+
+        if on_startup is not None:
+            startup_hooks.append(lambda: on_startup(container, broker, logger))
+
+        if on_shutdown is not None:
+            shutdown_hooks.append(on_shutdown)
+        shutdown_hooks.append(container.close)
+
+        app = FastStream(broker, on_startup=startup_hooks, on_shutdown=shutdown_hooks)
+        await app.run()
+        logger.info(f"{worker_name} shutdown complete")
+
+    asyncio.run(_run())