v2: added kafka instead of polling, more details to add sooner

Author: Max Azatian

backend/.env

@@ -1,13 +1,66 @@
 PROJECT_NAME=integr8scode
-SECRET_KEY=CHANGE_ME_this_is_a_dev_key_min_32_chars_required
+SECRET_KEY=-5c_AyfH64NL9g39JbpsOsGUoLTQPA4fGsEGQhTDja8
 ALGORITHM=HS256
-ACCESS_TOKEN_EXPIRE_MINUTES=30
-MONGODB_URL=mongodb://mongo:27017/integr8scode
+ACCESS_TOKEN_EXPIRE_MINUTES=1440
+MONGODB_URL=mongodb://root:rootpassword@mongo:27017/integr8scode?authSource=admin
+MONGO_ROOT_USER=root
+MONGO_ROOT_PASSWORD=rootpassword
 KUBERNETES_CONFIG_PATH=/app/kubeconfig.yaml
 KUBERNETES_CA_CERTIFICATE_PATH=/app/certs/k8s-ca.pem
 K8S_POD_CPU_LIMIT=1000m
 K8S_POD_MEMORY_LIMIT=128Mi
 K8S_POD_CPU_REQUEST=200m
 K8S_POD_MEMORY_REQUEST=128Mi
 K8S_POD_EXECUTION_TIMEOUT=5
-RATE_LIMITS=100/minute
+RATE_LIMITS=100/minute
+
+# Event-Driven Design Configuration
+KAFKA_BOOTSTRAP_SERVERS=kafka:29092
+SCHEMA_REGISTRY_URL=http://schema-registry:8081
+ENABLE_EVENT_STREAMING=true
+EVENT_RETENTION_DAYS=30
+KAFKA_CONSUMER_GROUP_ID=integr8scode-backend
+KAFKA_AUTO_OFFSET_RESET=earliest
+KAFKA_ENABLE_AUTO_COMMIT=true
+KAFKA_SESSION_TIMEOUT_MS=30000
+KAFKA_MAX_POLL_RECORDS=500
+
+# Feature Flags
+FEATURE_EVENT_DRIVEN_EXECUTION=true
+FEATURE_SSE_STREAMING=true
+FEATURE_WEBSOCKET_UPDATES=true
+FEATURE_DISTRIBUTED_TRACING=true
+FEATURE_EVENT_STORE_ENABLED=true
+FEATURE_CIRCUIT_BREAKER_ENABLED=true
+FEATURE_KAFKA_EVENTS=true
+
+# WebSocket Configuration
+WEBSOCKET_PING_INTERVAL=30
+WEBSOCKET_PING_TIMEOUT=10
+WEBSOCKET_MAX_CONNECTIONS_PER_USER=5
+WEBSOCKET_STALE_CONNECTION_TIMEOUT=300
+
+# Distributed Tracing
+ENABLE_TRACING=true
+JAEGER_AGENT_HOST=jaeger
+JAEGER_AGENT_PORT=6831
+TRACING_SERVICE_NAME=integr8scode-backend
+TRACING_SERVICE_VERSION=1.0.0
+TRACING_SAMPLING_RATE=1.0
+
+# Dead Letter Queue Configuration
+KAFKA_DLQ_TOPIC=dead-letter-queue
+DLQ_RETRY_MAX_ATTEMPTS=5
+DLQ_RETRY_BASE_DELAY_SECONDS=60.0
+DLQ_RETRY_MAX_DELAY_SECONDS=3600.0
+DLQ_RETENTION_DAYS=7
+DLQ_WARNING_THRESHOLD=100
+DLQ_CRITICAL_THRESHOLD=1000
+
+
+# App URL for notification links
+APP_URL=https://localhost
+
+# Service Configuration
+SERVICE_NAME=integr8scode-backend
+SERVICE_VERSION=1.0.0

backend/Dockerfile

@@ -1,8 +1,9 @@
-FROM python:3.9
+FROM python:3.12
 WORKDIR /app
 
 # Install required packages
 RUN apt-get update && apt-get upgrade -y liblzma-dev liblzma5 xz-utils && \
+    apt-get install -y libsnappy-dev && \
     rm -rf /var/lib/apt/lists/*
 
 # Install kubectl (latest stable version)
@@ -16,6 +17,8 @@ RUN pip install --no-cache-dir --upgrade pip setuptools>=70.0.0 wheel && \
 
 # Copy application files and configuration
 COPY ./app /app/app
+COPY ./workers /app/workers
+COPY ./scripts /app/scripts
 COPY .env /app/.env
 COPY openssl.cnf /app/openssl.cnf
 
@@ -28,4 +31,5 @@ CMD bash -c "\
   while [ ! -f /app/kubeconfig.yaml ]; do echo 'Waiting for kubeconfig...'; sleep 2; done && \
   echo 'Certs and kubeconfig found. Starting application...' && \
   export KUBECONFIG=/app/kubeconfig.yaml && \
+  PYTHONPATH=/app python /app/scripts/create_topics.py || true && \
   uvicorn app.main:app --host 0.0.0.0 --port 443 --ssl-keyfile /app/certs/server.key --ssl-certfile /app/certs/server.crt"

backend/alertmanager/alertmanager.yml

@@ -37,51 +37,18 @@ receivers:
     webhook_configs:
       - url: 'http://backend:443/api/v1/health'
         send_resolved: true
-        title: 'Integr8sCode Alert: {{ .GroupLabels.alertname }}'
-        text: |
-          {{ range .Alerts }}
-          Alert: {{ .Annotations.summary }}
-          Description: {{ .Annotations.description }}
-          Severity: {{ .Labels.severity }}
-          {{ end }}
 
   - name: 'critical-security'
     webhook_configs:
       - url: 'http://backend:443/api/v1/health'
         send_resolved: true
-        title: '🚨 CRITICAL SECURITY ALERT: {{ .GroupLabels.alertname }}'
-        text: |
-          IMMEDIATE ATTENTION REQUIRED - Security incident detected!
-          
-          {{ range .Alerts }}
-          Alert: {{ .Annotations.summary }}
-          Description: {{ .Annotations.description }}
-          Labels: {{ range .Labels.SortedPairs }}{{ .Name }}={{ .Value }} {{ end }}
-          {{ end }}
 
   - name: 'critical-infrastructure'
     webhook_configs:
       - url: 'http://backend:443/api/v1/health'
         send_resolved: true
-        title: '🔥 CRITICAL INFRASTRUCTURE ALERT: {{ .GroupLabels.alertname }}'
-        text: |
-          Critical system issue requiring immediate attention!
-          
-          {{ range .Alerts }}
-          Alert: {{ .Annotations.summary }}
-          Description: {{ .Annotations.description }}
-          Severity: {{ .Labels.severity }}
-          {{ end }}
 
   - name: 'warning'
     webhook_configs:
       - url: 'http://backend:443/api/v1/health'
-        send_resolved: true
-        title: '⚠️ WARNING: {{ .GroupLabels.alertname }}'
-        text: |
-          Warning condition detected - monitoring recommended.
-          
-          {{ range .Alerts }}
-          Alert: {{ .Annotations.summary }}
-          Description: {{ .Annotations.description }}
-          {{ end }}
+        send_resolved: true

backend/app/api/dependencies.py

@@ -1,33 +1,78 @@
-# app/api/dependencies.py
+from fastapi import Depends, HTTPException, Request, status
+from motor.motor_asyncio import AsyncIOMotorDatabase
+
 from app.config import Settings, get_settings
 from app.core.logging import logger
+from app.core.security import security_service
 from app.db.mongodb import DatabaseManager
-from fastapi import HTTPException, Request, status
-from motor.motor_asyncio import AsyncIOMotorDatabase
+from app.db.repositories.user_repository import UserRepository, get_user_repository
+from app.schemas_pydantic.user import UserResponse, UserRole
 
 
 def get_settings_dependency() -> Settings:
     return get_settings()
 
+
 def get_db_dependency(request: Request) -> AsyncIOMotorDatabase:
-    """
-    Dependency function to get the database handle from the DatabaseManager
-    stored in the application state.
-    """
     try:
         db_manager: DatabaseManager = request.app.state.db_manager
-        return db_manager.get_database()  # type: ignore
+        return db_manager.get_database()
     except AttributeError as e:
-        # This signifies a critical setup error: lifespan didn't set the state.
         logger.critical("DatabaseManager not found in app state. Application startup likely failed.")
         raise HTTPException(
             status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
             detail="Internal server error: Database service not available.",
         ) from e
     except RuntimeError as e:
-         # This signifies get_database was called when db is None (shouldn't happen if connect succeeded)
-         logger.error(f"Error retrieving database handle: {e}")
-         raise HTTPException(
-             status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-             detail="Internal server error: Database connection issue.",
-         ) from e
+        logger.error(f"Error retrieving database handle: {e}")
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="Internal server error: Database connection issue.",
+        ) from e
+
+
+async def get_current_user(
+        request: Request,
+        user_repo: UserRepository = Depends(get_user_repository),
+) -> UserResponse:
+    try:
+        token = request.cookies.get("access_token")
+        if not token:
+            raise HTTPException(
+                status_code=status.HTTP_401_UNAUTHORIZED,
+                detail="Not authenticated",
+                headers={"WWW-Authenticate": "Bearer"},
+            )
+
+        user_in_db = await security_service.get_current_user(token, user_repo)
+
+        return UserResponse(
+            user_id=user_in_db.user_id,
+            username=user_in_db.username,
+            email=user_in_db.email,
+            role=user_in_db.role,
+            created_at=user_in_db.created_at,
+            updated_at=user_in_db.updated_at
+        )
+    except HTTPException:
+        raise
+    except Exception:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Not authenticated",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+
+
+async def require_admin(
+        user: UserResponse = Depends(get_current_user),
+) -> UserResponse:
+    if user.role != UserRole.ADMIN:
+        logger.warning(
+            f"Admin access denied for user: {user.username} (role: {user.role})"
+        )
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Admin access required"
+        )
+    return user

backend/app/api/routes/admin/__init__.py

@@ -0,0 +1,5 @@
+from .events import router as events_router
+from .settings import router as settings_router
+from .users import router as users_router
+
+__all__ = ["events_router", "settings_router", "users_router"]