refactor: added settings-driven DI and CSRF middleware with SSE subscribed event

.github/workflows/backend-ci.yml

@@ -47,6 +47,7 @@ jobs:
         run: |
           cd backend
           uv run pytest tests/unit -v -rs \
+            --durations=0 \
             --cov=app \
             --cov-report=xml --cov-report=term
 
@@ -90,32 +91,12 @@ jobs:
           docker compose -f docker-compose.ci.yaml up -d --wait --wait-timeout 120
           docker compose -f docker-compose.ci.yaml ps
 
-      - name: Create Kafka topics
-        timeout-minutes: 2
-        env:
-          KAFKA_BOOTSTRAP_SERVERS: localhost:9092
-          KAFKA_TOPIC_PREFIX: "ci.${{ github.run_id }}."
-        run: |
-          cd backend
-          uv run python -m scripts.create_topics
-
       - name: Run integration tests
         timeout-minutes: 10
-        env:
-          MONGO_ROOT_USER: root
-          MONGO_ROOT_PASSWORD: rootpassword
-          MONGODB_HOST: 127.0.0.1
-          MONGODB_PORT: 27017
-          MONGODB_URL: mongodb://root:[email protected]:27017/?authSource=admin
-          KAFKA_BOOTSTRAP_SERVERS: localhost:9092
-          KAFKA_TOPIC_PREFIX: "ci.${{ github.run_id }}."
-          SCHEMA_REGISTRY_URL: http://localhost:8081
-          REDIS_HOST: localhost
-          REDIS_PORT: 6379
-          SCHEMA_SUBJECT_PREFIX: "ci.${{ github.run_id }}."
         run: |
           cd backend
           uv run pytest tests/integration -v -rs \
+            --durations=0 \
             --cov=app \
             --cov-report=xml --cov-report=term
 
@@ -184,32 +165,15 @@ jobs:
           timeout 90 bash -c 'until sudo k3s kubectl cluster-info; do sleep 5; done'
           kubectl create namespace integr8scode --dry-run=client -o yaml | kubectl apply -f -
 
-      - name: Create Kafka topics
-        timeout-minutes: 2
-        env:
-          KAFKA_BOOTSTRAP_SERVERS: localhost:9092
-          KAFKA_TOPIC_PREFIX: "ci.${{ github.run_id }}."
-        run: |
-          cd backend
-          uv run python -m scripts.create_topics
-
       - name: Run E2E tests
         timeout-minutes: 10
         env:
-          MONGO_ROOT_USER: root
-          MONGO_ROOT_PASSWORD: rootpassword
-          MONGODB_URL: mongodb://root:[email protected]:27017/?authSource=admin
-          KAFKA_BOOTSTRAP_SERVERS: localhost:9092
-          KAFKA_TOPIC_PREFIX: "ci.${{ github.run_id }}."
-          SCHEMA_REGISTRY_URL: http://localhost:8081
-          REDIS_HOST: localhost
-          REDIS_PORT: 6379
-          SCHEMA_SUBJECT_PREFIX: "ci.${{ github.run_id }}."
           KUBECONFIG: /home/runner/.kube/config
           K8S_NAMESPACE: integr8scode
         run: |
           cd backend
           uv run pytest tests/e2e -v -rs \
+            --durations=0 \
             --cov=app \
             --cov-report=xml --cov-report=term
 

backend/.env

@@ -76,3 +76,6 @@ WEB_BACKLOG=2048
 # When running uvicorn locally (outside Docker), bind to IPv4 loopback to avoid
 # IPv6-only localhost resolution on some Linux distros.
 SERVER_HOST=127.0.0.1
+
+# Security
+BCRYPT_ROUNDS=12

backend/.env.test

@@ -3,7 +3,6 @@ PROJECT_NAME=integr8scode
 DATABASE_NAME=integr8scode_test
 API_V1_STR=/api/v1
 SECRET_KEY=test-secret-key-for-testing-only-32chars!!
-ENVIRONMENT=testing
 TESTING=true
 
 # MongoDB - use localhost for tests
@@ -23,22 +22,21 @@ REDIS_DECODE_RESPONSES=true
 # Kafka - use localhost for tests
 KAFKA_BOOTSTRAP_SERVERS=localhost:9092
 KAFKA_TOPIC_PREFIX=test.
+SCHEMA_SUBJECT_PREFIX=test.
 SCHEMA_REGISTRY_URL=http://localhost:8081
 
 # Security
 SECURE_COOKIES=true
-CORS_ALLOWED_ORIGINS=["http://localhost:3000","https://localhost:3000"]
+BCRYPT_ROUNDS=4
 
 # Features
 RATE_LIMIT_ENABLED=true
 ENABLE_TRACING=false
 
-# OpenTelemetry - explicitly disabled for tests (no endpoint = NoOp meter)
+# OpenTelemetry - explicitly disabled for tests
 OTEL_EXPORTER_OTLP_ENDPOINT=
-OTEL_METRICS_EXPORTER=none
-OTEL_TRACES_EXPORTER=none
-OTEL_LOGS_EXPORTER=none
 
 # Development
 DEVELOPMENT_MODE=false
 LOG_LEVEL=INFO
+ENVIRONMENT=test

backend/Dockerfile

@@ -27,7 +27,7 @@ done
 echo "Starting application..."
 [ -f /app/kubeconfig.yaml ] && export KUBECONFIG=/app/kubeconfig.yaml
 
-exec gunicorn app.main:app \
+exec gunicorn 'app.main:create_app()' \
   -k uvicorn.workers.UvicornWorker \
   --bind 0.0.0.0:443 \
   --workers ${WEB_CONCURRENCY:-4} \

backend/app/api/routes/auth.py

@@ -7,7 +7,7 @@
 from fastapi.security import OAuth2PasswordRequestForm
 from pymongo.errors import DuplicateKeyError
 
-from app.core.security import security_service
+from app.core.security import SecurityService
 from app.core.utils import get_client_ip
 from app.db.repositories import UserRepository
 from app.domain.user import DomainUserCreate
@@ -19,7 +19,7 @@
     UserResponse,
 )
 from app.services.auth_service import AuthService
-from app.settings import get_settings
+from app.settings import Settings
 
 router = APIRouter(prefix="/auth", tags=["authentication"], route_class=DishkaRoute)
 
@@ -29,6 +29,8 @@ async def login(
     request: Request,
     response: Response,
     user_repo: FromDishka[UserRepository],
+    security_service: FromDishka[SecurityService],
+    settings: FromDishka[Settings],
     logger: FromDishka[logging.Logger],
     form_data: OAuth2PasswordRequestForm = Depends(),
 ) -> LoginResponse:
@@ -74,8 +76,6 @@ async def login(
             headers={"WWW-Authenticate": "Bearer"},
         )
 
-    settings = get_settings()
-
     logger.info(
         "Login successful",
         extra={
@@ -127,6 +127,7 @@ async def register(
     request: Request,
     user: UserCreate,
     user_repo: FromDishka[UserRepository],
+    security_service: FromDishka[SecurityService],
     logger: FromDishka[logging.Logger],
 ) -> UserResponse:
     logger.info(

backend/app/api/routes/dlq.py

@@ -95,9 +95,9 @@ async def get_dlq_message(event_id: str, repository: FromDishka[DLQRepository])
 
 @router.post("/retry", response_model=DLQBatchRetryResponse)
 async def retry_dlq_messages(
-    retry_request: ManualRetryRequest, repository: FromDishka[DLQRepository], dlq_manager: FromDishka[DLQManager]
+    retry_request: ManualRetryRequest, dlq_manager: FromDishka[DLQManager]
 ) -> DLQBatchRetryResponse:
-    result = await repository.retry_messages_batch(retry_request.event_ids, dlq_manager)
+    result = await dlq_manager.retry_messages_batch(retry_request.event_ids)
     return DLQBatchRetryResponse(
         total=result.total,
         successful=result.successful,

backend/app/api/routes/events.py

@@ -27,7 +27,7 @@
 from app.schemas_pydantic.user import UserResponse
 from app.services.event_service import EventService
 from app.services.kafka_event_service import KafkaEventService
-from app.settings import get_settings
+from app.settings import Settings
 
 router = APIRouter(prefix="/events", tags=["events"], route_class=DishkaRoute)
 
@@ -229,8 +229,8 @@ async def publish_custom_event(
     event_request: PublishEventRequest,
     request: Request,
     event_service: FromDishka[KafkaEventService],
+    settings: FromDishka[Settings],
 ) -> PublishEventResponse:
-    settings = get_settings()
     base_meta = EventMetadata(
         service_name=settings.SERVICE_NAME,
         service_version=settings.SERVICE_VERSION,
@@ -311,6 +311,7 @@ async def replay_aggregate_events(
     admin: Annotated[UserResponse, Depends(admin_user)],
     event_service: FromDishka[EventService],
     kafka_event_service: FromDishka[KafkaEventService],
+    settings: FromDishka[Settings],
     logger: FromDishka[logging.Logger],
     target_service: str | None = Query(None, description="Service to replay events to"),
     dry_run: bool = Query(True, description="If true, only show what would be replayed"),
@@ -339,7 +340,6 @@ async def replay_aggregate_events(
             await asyncio.sleep(0.1)
 
         try:
-            settings = get_settings()
             meta = EventMetadata(
                 service_name=settings.SERVICE_NAME,
                 service_version=settings.SERVICE_VERSION,

backend/app/api/routes/execution.py

@@ -34,7 +34,7 @@
 from app.services.execution_service import ExecutionService
 from app.services.idempotency import IdempotencyManager
 from app.services.kafka_event_service import KafkaEventService
-from app.settings import get_settings
+from app.settings import Settings
 
 router = APIRouter(route_class=DishkaRoute)
 
@@ -162,6 +162,7 @@ async def cancel_execution(
     current_user: Annotated[UserResponse, Depends(current_user)],
     cancel_request: CancelExecutionRequest,
     event_service: FromDishka[KafkaEventService],
+    settings: FromDishka[Settings],
 ) -> CancelResponse:
     # Handle terminal states
     terminal_states = [ExecutionStatus.COMPLETED, ExecutionStatus.FAILED, ExecutionStatus.TIMEOUT]
@@ -178,7 +179,6 @@ async def cancel_execution(
             event_id="-1",  # exact event_id unknown
         )
 
-    settings = get_settings()
     payload = {
         "execution_id": execution.execution_id,
         "status": str(ExecutionStatus.CANCELLED),

backend/app/core/adaptive_sampling.py

@@ -2,14 +2,14 @@
 import threading
 import time
 from collections import deque
-from typing import Any, Sequence, Tuple
+from typing import Sequence, Tuple
 
 from opentelemetry.context import Context
 from opentelemetry.sdk.trace.sampling import Decision, Sampler, SamplingResult
 from opentelemetry.trace import Link, SpanKind, TraceState, get_current_span
 from opentelemetry.util.types import Attributes
 
-from app.settings import get_settings
+from app.settings import Settings
 
 
 class AdaptiveSampler(Sampler):
@@ -239,11 +239,8 @@ def shutdown(self) -> None:
             self._adjustment_thread.join(timeout=5.0)
 
 
-def create_adaptive_sampler(settings: Any | None = None) -> AdaptiveSampler:
+def create_adaptive_sampler(settings: Settings) -> AdaptiveSampler:
     """Create adaptive sampler with settings"""
-    if settings is None:
-        settings = get_settings()
-
     return AdaptiveSampler(
         base_rate=settings.TRACING_SAMPLING_RATE,
         min_rate=max(0.001, settings.TRACING_SAMPLING_RATE / 100),  # 1/100th of base

backend/app/core/container.py

@@ -45,6 +45,7 @@ def create_app_container(settings: Settings) -> AsyncContainer:
         RepositoryProvider(),
         MessagingProvider(),
         EventProvider(),
+        SagaOrchestratorProvider(),
         KafkaServicesProvider(),
         SSEProvider(),
         AuthProvider(),

backend/app/core/dishka_lifespan.py

@@ -14,7 +14,7 @@
 from app.events.event_store_consumer import EventStoreConsumer
 from app.events.schema.schema_registry import SchemaRegistryManager, initialize_event_schemas
 from app.services.sse.kafka_redis_bridge import SSEKafkaRedisBridge
-from app.settings import get_settings
+from app.settings import Settings
 
 
 @asynccontextmanager
@@ -27,10 +27,9 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
     - No manual service management
     - Dishka handles all lifecycle automatically
     """
-    settings = get_settings()
-
-    # Get logger from DI container
+    # Get settings and logger from DI container (uses test settings in tests)
     container: AsyncContainer = app.state.dishka_container
+    settings = await container.get(Settings)
     logger = await container.get(logging.Logger)
 
     logger.info(
@@ -44,25 +43,32 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
     # Metrics setup moved to app creation to allow middleware registration
     logger.info("Lifespan start: tracing and services initialization")
 
-    # Initialize tracing
-    instrumentation_report = init_tracing(
-        service_name=settings.TRACING_SERVICE_NAME,
-        logger=logger,
-        service_version=settings.TRACING_SERVICE_VERSION,
-        sampling_rate=settings.TRACING_SAMPLING_RATE,
-        enable_console_exporter=settings.TESTING,
-        adaptive_sampling=settings.TRACING_ADAPTIVE_SAMPLING,
-    )
-
-    if instrumentation_report.has_failures():
-        logger.warning(
-            "Some instrumentation libraries failed to initialize",
-            extra={"instrumentation_summary": instrumentation_report.get_summary()},
+    # Initialize tracing only when enabled (avoid exporter retries in tests)
+    if settings.ENABLE_TRACING and not settings.TESTING:
+        instrumentation_report = init_tracing(
+            service_name=settings.TRACING_SERVICE_NAME,
+            settings=settings,
+            logger=logger,
+            service_version=settings.TRACING_SERVICE_VERSION,
+            sampling_rate=settings.TRACING_SAMPLING_RATE,
+            enable_console_exporter=settings.TESTING,
+            adaptive_sampling=settings.TRACING_ADAPTIVE_SAMPLING,
         )
+
+        if instrumentation_report.has_failures():
+            logger.warning(
+                "Some instrumentation libraries failed to initialize",
+                extra={"instrumentation_summary": instrumentation_report.get_summary()},
+            )
+        else:
+            logger.info(
+                "Distributed tracing initialized successfully",
+                extra={"instrumentation_summary": instrumentation_report.get_summary()},
+            )
     else:
         logger.info(
-            "Distributed tracing initialized successfully",
-            extra={"instrumentation_summary": instrumentation_report.get_summary()},
+            "Distributed tracing disabled",
+            extra={"testing": settings.TESTING, "enable_tracing": settings.ENABLE_TRACING},
         )
 
     # Initialize schema registry once at startup

backend/app/core/k8s_clients.py

@@ -3,14 +3,18 @@
 
 from kubernetes import client as k8s_client
 from kubernetes import config as k8s_config
+from kubernetes import watch as k8s_watch
 
 
 @dataclass(frozen=True)
 class K8sClients:
+    """Kubernetes API clients bundle for dependency injection."""
+
     api_client: k8s_client.ApiClient
     v1: k8s_client.CoreV1Api
     apps_v1: k8s_client.AppsV1Api
     networking_v1: k8s_client.NetworkingV1Api
+    watch: k8s_watch.Watch
 
 
 def create_k8s_clients(
@@ -33,6 +37,7 @@ def create_k8s_clients(
         v1=k8s_client.CoreV1Api(api_client),
         apps_v1=k8s_client.AppsV1Api(api_client),
         networking_v1=k8s_client.NetworkingV1Api(api_client),
+        watch=k8s_watch.Watch(),
     )