diff --git a/.github/actions/e2e-ready/action.yml b/.github/actions/e2e-ready/action.yml index 43073d27..15d3e042 100644 --- a/.github/actions/e2e-ready/action.yml +++ b/.github/actions/e2e-ready/action.yml @@ -32,6 +32,10 @@ runs: cp backend/config.test.toml backend/config.toml cp backend/secrets.example.toml backend/secrets.toml + - name: Pre-pull test runtime image into K3s + shell: bash + run: sudo k3s crictl pull docker.io/library/python:3.11-slim + - name: Wait for image pull and infra shell: bash run: | diff --git a/backend/app/core/providers.py b/backend/app/core/providers.py index a1824dda..ae6579fa 100644 --- a/backend/app/core/providers.py +++ b/backend/app/core/providers.py @@ -52,7 +52,6 @@ from app.domain.rate_limit import RateLimitConfig from app.domain.saga.models import SagaConfig from app.events.core import UnifiedProducer -from app.events.event_store import EventStore, create_event_store from app.events.schema.schema_registry import SchemaRegistryManager from app.services.admin import AdminEventsService, AdminSettingsService, AdminUserService from app.services.auth_service import AuthService @@ -182,11 +181,12 @@ def get_unified_producer( self, broker: KafkaBroker, schema_registry: SchemaRegistryManager, + event_repository: EventRepository, logger: logging.Logger, settings: Settings, event_metrics: EventMetrics, ) -> UnifiedProducer: - return UnifiedProducer(broker, schema_registry, logger, settings, event_metrics) + return UnifiedProducer(broker, schema_registry, event_repository, logger, settings, event_metrics) @provide def get_idempotency_repository(self, redis_client: redis.Redis) -> RedisIdempotencyRepository: @@ -276,21 +276,8 @@ class EventProvider(Provider): scope = Scope.APP @provide - async def get_schema_registry(self, settings: Settings, logger: logging.Logger) -> SchemaRegistryManager: - registry = SchemaRegistryManager(settings, logger) - await registry.initialize_schemas() - return registry - - @provide - def get_event_store( - self, - schema_registry: SchemaRegistryManager, - logger: logging.Logger, - event_metrics: EventMetrics, - ) -> EventStore: - return create_event_store( - schema_registry=schema_registry, logger=logger, event_metrics=event_metrics, ttl_days=90 - ) + def get_schema_registry(self, settings: Settings, logger: logging.Logger) -> SchemaRegistryManager: + return SchemaRegistryManager(settings, logger) class KubernetesProvider(Provider): @@ -481,14 +468,12 @@ def get_event_service(self, event_repository: EventRepository) -> EventService: @provide def get_kafka_event_service( self, - event_repository: EventRepository, kafka_producer: UnifiedProducer, settings: Settings, logger: logging.Logger, event_metrics: EventMetrics, ) -> KafkaEventService: return KafkaEventService( - event_repository=event_repository, kafka_producer=kafka_producer, settings=settings, logger=logger, @@ -632,7 +617,7 @@ def get_execution_service( self, execution_repository: ExecutionRepository, kafka_producer: UnifiedProducer, - event_store: EventStore, + event_repository: EventRepository, settings: Settings, logger: logging.Logger, execution_metrics: ExecutionMetrics, @@ -640,7 +625,7 @@ def get_execution_service( return ExecutionService( execution_repo=execution_repository, producer=kafka_producer, - event_store=event_store, + event_repository=event_repository, settings=settings, logger=logger, execution_metrics=execution_metrics, @@ -874,14 +859,12 @@ def get_event_replay_service( self, replay_repository: ReplayRepository, kafka_producer: UnifiedProducer, - event_store: EventStore, replay_metrics: ReplayMetrics, logger: logging.Logger, ) -> EventReplayService: return EventReplayService( repository=replay_repository, producer=kafka_producer, - event_store=event_store, replay_metrics=replay_metrics, logger=logger, ) @@ -901,7 +884,6 @@ async def get_event_replay_service( self, replay_repository: ReplayRepository, kafka_producer: UnifiedProducer, - event_store: EventStore, replay_metrics: ReplayMetrics, logger: logging.Logger, database: Database, @@ -910,7 +892,6 @@ async def get_event_replay_service( service = EventReplayService( repository=replay_repository, producer=kafka_producer, - event_store=event_store, replay_metrics=replay_metrics, logger=logger, ) diff --git a/backend/app/db/repositories/admin/admin_events_repository.py b/backend/app/db/repositories/admin/admin_events_repository.py index 800974fb..380ebcea 100644 --- a/backend/app/db/repositories/admin/admin_events_repository.py +++ b/backend/app/db/repositories/admin/admin_events_repository.py @@ -17,6 +17,7 @@ from app.domain.enums.replay import ReplayStatus from app.domain.events import ( DomainEvent, + DomainEventAdapter, EventBrowseResult, EventDetail, EventExportRow, @@ -26,7 +27,6 @@ EventTypeCount, HourlyEventCount, UserEventCount, - domain_event_adapter, ) from app.domain.replay.models import ReplayFilter, ReplaySessionState @@ -59,7 +59,7 @@ async def browse_events( total = await query.count() docs = await query.sort([(sort_by, sort_order)]).skip(skip).limit(limit).to_list() - events = [domain_event_adapter.validate_python(d, from_attributes=True) for d in docs] + events = [DomainEventAdapter.validate_python(d, from_attributes=True) for d in docs] return EventBrowseResult(events=events, total=total, skip=skip, limit=limit) @@ -68,7 +68,7 @@ async def get_event_detail(self, event_id: str) -> EventDetail | None: if not doc: return None - event = domain_event_adapter.validate_python(doc, from_attributes=True) + event = DomainEventAdapter.validate_python(doc, from_attributes=True) related_query = {"metadata.correlation_id": doc.metadata.correlation_id, "event_id": {"$ne": event_id}} related_docs = await ( diff --git a/backend/app/db/repositories/event_repository.py b/backend/app/db/repositories/event_repository.py index 6eb20841..17598020 100644 --- a/backend/app/db/repositories/event_repository.py +++ b/backend/app/db/repositories/event_repository.py @@ -6,6 +6,7 @@ from beanie.odm.operators.find import BaseFindOperator from beanie.operators import GTE, LT, LTE, Eq, In, Not, Or, RegEx from monggregate import Pipeline, S +from pymongo.errors import DuplicateKeyError from app.core.tracing import EventAttributes from app.core.tracing.utils import add_span_attributes @@ -14,13 +15,13 @@ from app.domain.events import ( ArchivedEvent, DomainEvent, + DomainEventAdapter, EventAggregationResult, EventListResult, EventReplayInfo, EventStatistics, EventTypeCount, ServiceEventCount, - domain_event_adapter, ) @@ -41,6 +42,7 @@ def _build_time_filter(self, start_time: datetime | None, end_time: datetime | N return {key: value for key, value in {"$gte": start_time, "$lte": end_time}.items() if value is not None} async def store_event(self, event: DomainEvent) -> str: + """Idempotent event store — silently ignores duplicates by event_id.""" data = event.model_dump(exclude_none=True) data.setdefault("stored_at", datetime.now(timezone.utc)) doc = EventDocument(**data) @@ -51,7 +53,11 @@ async def store_event(self, event: DomainEvent) -> str: str(EventAttributes.EXECUTION_ID): event.aggregate_id or "", } ) - await doc.insert() + try: + await doc.insert() + except DuplicateKeyError: + self.logger.debug(f"Event {event.event_id} already stored, skipping") + return event.event_id self.logger.debug(f"Stored event {event.event_id} of type {event.event_type}") return event.event_id @@ -73,7 +79,7 @@ async def get_event(self, event_id: str) -> DomainEvent | None: doc = await EventDocument.find_one(EventDocument.event_id == event_id) if not doc: return None - return domain_event_adapter.validate_python(doc, from_attributes=True) + return DomainEventAdapter.validate_python(doc, from_attributes=True) async def get_events_by_type( self, @@ -94,7 +100,7 @@ async def get_events_by_type( .limit(limit) .to_list() ) - return [domain_event_adapter.validate_python(d, from_attributes=True) for d in docs] + return [DomainEventAdapter.validate_python(d, from_attributes=True) for d in docs] async def get_events_by_aggregate( self, aggregate_id: str, event_types: list[EventType] | None = None, limit: int = 100 @@ -105,7 +111,7 @@ async def get_events_by_aggregate( docs = ( await EventDocument.find(*conditions).sort([("timestamp", SortDirection.ASCENDING)]).limit(limit).to_list() ) - return [domain_event_adapter.validate_python(d, from_attributes=True) for d in docs] + return [DomainEventAdapter.validate_python(d, from_attributes=True) for d in docs] async def get_events_by_correlation( self, correlation_id: str, limit: int = 100, skip: int = 0, user_id: str | None = None, @@ -119,7 +125,7 @@ async def get_events_by_correlation( .sort([("timestamp", SortDirection.ASCENDING)]) .skip(skip).limit(limit).to_list() ) - events = [domain_event_adapter.validate_python(d, from_attributes=True) for d in docs] + events = [DomainEventAdapter.validate_python(d, from_attributes=True) for d in docs] total_count = await EventDocument.find(condition).count() total_count = max(total_count, skip + len(events)) return EventListResult( @@ -152,16 +158,22 @@ async def get_events_by_user( .limit(limit) .to_list() ) - return [domain_event_adapter.validate_python(d, from_attributes=True) for d in docs] + return [DomainEventAdapter.validate_python(d, from_attributes=True) for d in docs] async def get_execution_events( - self, execution_id: str, limit: int = 100, skip: int = 0, exclude_system_events: bool = False + self, + execution_id: str, + limit: int = 100, + skip: int = 0, + exclude_system_events: bool = False, + event_types: list[EventType] | None = None, ) -> EventListResult: conditions: list[Any] = [ Or( EventDocument.execution_id == execution_id, EventDocument.aggregate_id == execution_id, ), + In(EventDocument.event_type, event_types) if event_types else None, Not(RegEx(EventDocument.metadata.service_name, "^system-")) if exclude_system_events else None, ] conditions = [c for c in conditions if c is not None] @@ -172,7 +184,7 @@ async def get_execution_events( .sort([("timestamp", SortDirection.ASCENDING)]) .skip(skip).limit(limit).to_list() ) - events = [domain_event_adapter.validate_python(d, from_attributes=True) for d in docs] + events = [DomainEventAdapter.validate_python(d, from_attributes=True) for d in docs] total_count = await EventDocument.find(*conditions).count() total_count = max(total_count, skip + len(events)) return EventListResult( @@ -307,7 +319,7 @@ async def get_user_events_paginated( .sort([("timestamp", sort_direction)]) .skip(skip).limit(limit).to_list() ) - events = [domain_event_adapter.validate_python(d, from_attributes=True) for d in docs] + events = [DomainEventAdapter.validate_python(d, from_attributes=True) for d in docs] total_count = await EventDocument.find(*conditions).count() total_count = max(total_count, skip + len(events)) return EventListResult( @@ -334,7 +346,7 @@ async def query_events( .sort([(sort_field, SortDirection.DESCENDING)]) .skip(skip).limit(limit).to_list() ) - events = [domain_event_adapter.validate_python(d, from_attributes=True) for d in docs] + events = [DomainEventAdapter.validate_python(d, from_attributes=True) for d in docs] total_count = await EventDocument.find(query).count() total_count = max(total_count, skip + len(events)) return EventListResult( @@ -398,7 +410,7 @@ async def get_aggregate_replay_info(self, aggregate_id: str) -> EventReplayInfo ) async for doc in EventDocument.aggregate(pipeline.export()): - events = [domain_event_adapter.validate_python(e) for e in doc["events"]] + events = [DomainEventAdapter.validate_python(e) for e in doc["events"]] return EventReplayInfo( events=events, event_count=doc["event_count"], diff --git a/backend/app/dlq/manager.py b/backend/app/dlq/manager.py index 17232b8c..5cf21e36 100644 --- a/backend/app/dlq/manager.py +++ b/backend/app/dlq/manager.py @@ -1,7 +1,6 @@ -import json import logging from datetime import datetime, timezone -from typing import Any, Callable +from typing import Callable from faststream.kafka import KafkaBroker @@ -122,18 +121,6 @@ async def process_monitoring_cycle(self) -> None: await self.process_due_retries() await self.update_queue_metrics() - def parse_kafka_message(self, msg: Any) -> DLQMessage: - """Parse a raw Kafka ConsumerRecord into a DLQMessage.""" - data = json.loads(msg.value) - headers = {k: v.decode() for k, v in (msg.headers or [])} - return DLQMessage(**data, dlq_offset=msg.offset, dlq_partition=msg.partition, headers=headers) - - def parse_dlq_body( - self, data: dict[str, Any], offset: int, partition: int, headers: dict[str, str] - ) -> DLQMessage: - """Parse a deserialized DLQ message body into a DLQMessage.""" - return DLQMessage(**data, dlq_offset=offset, dlq_partition=partition, headers=headers) - async def handle_message(self, message: DLQMessage) -> None: """Process a single DLQ message: filter → store → decide retry/discard.""" for filter_func in self._filters: @@ -166,14 +153,14 @@ async def retry_message(self, message: DLQMessage) -> None: retry_topic = f"{message.original_topic}{self.retry_topic_suffix}" hdrs: dict[str, str] = { + "event_type": message.event.event_type, "dlq_retry_count": str(message.retry_count + 1), "dlq_original_error": message.error, "dlq_retry_timestamp": datetime.now(timezone.utc).isoformat(), } hdrs = inject_trace_context(hdrs) - event = message.event - serialized = json.dumps(event.model_dump(mode="json")).encode() + serialized = await self.schema_registry.serialize_event(message.event) await self._broker.publish( message=serialized, diff --git a/backend/app/domain/events/__init__.py b/backend/app/domain/events/__init__.py index d537a39e..2a9bc41c 100644 --- a/backend/app/domain/events/__init__.py +++ b/backend/app/domain/events/__init__.py @@ -29,6 +29,7 @@ CreatePodCommandEvent, DeletePodCommandEvent, DomainEvent, + DomainEventAdapter, EventMetadata, # Execution Events ExecutionAcceptedEvent, @@ -90,7 +91,6 @@ UserRegisteredEvent, UserSettingsUpdatedEvent, UserUpdatedEvent, - domain_event_adapter, ) __all__ = [ @@ -119,7 +119,7 @@ "DomainEvent", "EventMetadata", "ResourceUsageDomain", - "domain_event_adapter", + "DomainEventAdapter", # Execution Events "ExecutionRequestedEvent", "ExecutionAcceptedEvent", diff --git a/backend/app/domain/events/typed.py b/backend/app/domain/events/typed.py index 6212fd6c..c230b9d4 100644 --- a/backend/app/domain/events/typed.py +++ b/backend/app/domain/events/typed.py @@ -703,4 +703,4 @@ class ArchivedEvent(AvroBase): ] # TypeAdapter for polymorphic loading - validates raw data to correct typed event -domain_event_adapter: TypeAdapter[DomainEvent] = TypeAdapter(DomainEvent) +DomainEventAdapter: TypeAdapter[DomainEvent] = TypeAdapter(DomainEvent) diff --git a/backend/app/events/broker.py b/backend/app/events/broker.py index 3a6e3336..01e70da4 100644 --- a/backend/app/events/broker.py +++ b/backend/app/events/broker.py @@ -4,36 +4,24 @@ from faststream import StreamMessage from faststream.kafka import KafkaBroker -from app.domain.events.typed import DomainEvent +from app.domain.events.typed import DomainEvent, DomainEventAdapter from app.events.schema.schema_registry import SchemaRegistryManager from app.settings import Settings -def create_avro_decoder( - schema_registry: SchemaRegistryManager, -) -> Any: - """Create a custom Avro decoder closure for FastStream subscribers. - - The decoder receives a StreamMessage whose body is Confluent wire-format - Avro bytes (magic byte + 4-byte schema ID + Avro payload). We delegate - deserialization to SchemaRegistryManager which resolves the schema from - the registry and decodes into the concrete DomainEvent subclass. - """ - - async def avro_decoder(msg: StreamMessage[Any]) -> DomainEvent: - return await schema_registry.deserialize_event(msg.body, msg.raw_message.topic) - - return avro_decoder - - def create_broker( settings: Settings, schema_registry: SchemaRegistryManager, logger: logging.Logger, ) -> KafkaBroker: """Create a KafkaBroker with Avro decoder for standalone workers.""" + + async def avro_decoder(msg: StreamMessage[Any]) -> DomainEvent: + payload = await schema_registry.serializer.decode_message(msg.body) + return DomainEventAdapter.validate_python(payload) + return KafkaBroker( settings.KAFKA_BOOTSTRAP_SERVERS, - decoder=create_avro_decoder(schema_registry), + decoder=avro_decoder, logger=logger, ) diff --git a/backend/app/events/core/producer.py b/backend/app/events/core/producer.py index b7b4037c..daad87e2 100644 --- a/backend/app/events/core/producer.py +++ b/backend/app/events/core/producer.py @@ -1,5 +1,4 @@ import asyncio -import json import logging import socket from datetime import datetime, timezone @@ -8,7 +7,8 @@ from app.core.metrics import EventMetrics from app.core.tracing.utils import inject_trace_context -from app.dlq.models import DLQMessage, DLQMessageStatus +from app.db.repositories.event_repository import EventRepository +from app.dlq.models import DLQMessageStatus from app.domain.enums.kafka import KafkaTopic from app.domain.events.typed import DomainEvent from app.events.schema.schema_registry import SchemaRegistryManager @@ -27,18 +27,21 @@ def __init__( self, broker: KafkaBroker, schema_registry_manager: SchemaRegistryManager, + event_repository: EventRepository, logger: logging.Logger, settings: Settings, event_metrics: EventMetrics, ): self._broker = broker self._schema_registry = schema_registry_manager + self._event_repository = event_repository self.logger = logger self._event_metrics = event_metrics self._topic_prefix = settings.KAFKA_TOPIC_PREFIX async def produce(self, event_to_produce: DomainEvent, key: str) -> None: - """Produce a message to Kafka.""" + """Persist event to MongoDB, then publish to Kafka.""" + await self._event_repository.store_event(event_to_produce) topic = f"{self._topic_prefix}{EVENT_TYPE_TO_TOPIC[event_to_produce.event_type]}" try: serialized_value = await self._schema_registry.serialize_event(event_to_produce) @@ -67,44 +70,35 @@ async def produce(self, event_to_produce: DomainEvent, key: str) -> None: async def send_to_dlq( self, original_event: DomainEvent, original_topic: str, error: Exception, retry_count: int = 0 ) -> None: - """Send a failed event to the Dead Letter Queue.""" + """Send a failed event to the Dead Letter Queue. + + The event body is Avro-encoded (same as every other topic). + DLQ metadata is carried in Kafka headers. + """ try: current_task = asyncio.current_task() task_name = current_task.get_name() if current_task else "main" producer_id = f"{socket.gethostname()}-{task_name}" - dlq_message = DLQMessage( - event=original_event, - original_topic=original_topic, - error=str(error), - retry_count=retry_count, - failed_at=datetime.now(timezone.utc), - status=DLQMessageStatus.PENDING, - producer_id=producer_id, - ) - - dlq_event_data = { - "event": dlq_message.event.model_dump(mode="json"), - "original_topic": dlq_message.original_topic, - "error": dlq_message.error, - "retry_count": dlq_message.retry_count, - "failed_at": dlq_message.failed_at.isoformat(), - "producer_id": dlq_message.producer_id, - "status": str(dlq_message.status), - } + serialized_value = await self._schema_registry.serialize_event(original_event) + dlq_topic = f"{self._topic_prefix}{KafkaTopic.DEAD_LETTER_QUEUE}" - serialized_value = json.dumps(dlq_event_data).encode("utf-8") - dlq_topic = f"{self._topic_prefix}{str(KafkaTopic.DEAD_LETTER_QUEUE)}" + headers = inject_trace_context({ + "event_type": original_event.event_type, + "original_topic": original_topic, + "error_type": type(error).__name__, + "error": str(error), + "retry_count": str(retry_count), + "failed_at": datetime.now(timezone.utc).isoformat(), + "status": DLQMessageStatus.PENDING, + "producer_id": producer_id, + }) await self._broker.publish( message=serialized_value, topic=dlq_topic, key=original_event.event_id.encode() if original_event.event_id else None, - headers={ - "original_topic": original_topic, - "error_type": type(error).__name__, - "retry_count": str(retry_count), - }, + headers=headers, ) self._event_metrics.record_kafka_message_produced(dlq_topic) diff --git a/backend/app/events/event_store.py b/backend/app/events/event_store.py deleted file mode 100644 index 026ae84a..00000000 --- a/backend/app/events/event_store.py +++ /dev/null @@ -1,331 +0,0 @@ -import asyncio -import logging -from collections.abc import Awaitable, Callable -from datetime import datetime, timedelta, timezone -from typing import Any - -from beanie.odm.enums import SortDirection -from pymongo.errors import BulkWriteError, DuplicateKeyError - -from app.core.metrics import EventMetrics -from app.core.tracing import EventAttributes -from app.core.tracing.utils import add_span_attributes -from app.db.docs import EventDocument -from app.domain.enums.events import EventType -from app.domain.events.typed import DomainEvent -from app.events.schema.schema_registry import SchemaRegistryManager - - -class EventStore: - def __init__( - self, - schema_registry: SchemaRegistryManager, - logger: logging.Logger, - event_metrics: EventMetrics, - ttl_days: int = 90, - batch_size: int = 100, - ): - self.metrics = event_metrics - self.schema_registry = schema_registry - self.logger = logger - self.ttl_days = ttl_days - self.batch_size = batch_size - self._initialized = False - - self._SECURITY_TYPES = [ - EventType.USER_LOGIN, - EventType.USER_LOGGED_OUT, - EventType.SECURITY_VIOLATION, - ] - - async def initialize(self) -> None: - if self._initialized: - return - self._initialized = True - self.logger.info("Event store initialized with Beanie") - - async def store_event(self, event: DomainEvent) -> bool: - start = asyncio.get_running_loop().time() - try: - now = datetime.now(timezone.utc) - ttl = now + timedelta(days=self.ttl_days) - doc = EventDocument(**event.model_dump(exclude_none=True), stored_at=now, ttl_expires_at=ttl) - await doc.insert() - - add_span_attributes( - **{ - str(EventAttributes.EVENT_TYPE): str(event.event_type), - str(EventAttributes.EVENT_ID): event.event_id, - str(EventAttributes.EXECUTION_ID): event.aggregate_id or "", - } - ) - - duration = asyncio.get_running_loop().time() - start - self.metrics.record_event_store_duration(duration, "store_single", "event_store") - self.metrics.record_event_stored(event.event_type, "event_store") - return True - except DuplicateKeyError: - self.logger.warning(f"Event {event.event_id} already exists") - return True - except Exception as e: - self.logger.error(f"Failed to store event {event.event_id}: {e.__class__.__name__}: {e}", exc_info=True) - self.metrics.record_event_store_failed(event.event_type, type(e).__name__) - return False - - async def store_batch(self, events: list[DomainEvent]) -> dict[str, int]: - start = asyncio.get_running_loop().time() - results = {"total": len(events), "stored": 0, "duplicates": 0, "failed": 0} - if not events: - return results - - now = datetime.now(timezone.utc) - ttl = now + timedelta(days=self.ttl_days) - try: - docs = [EventDocument(**e.model_dump(exclude_none=True), stored_at=now, ttl_expires_at=ttl) for e in events] - - try: - await EventDocument.insert_many(docs) - results["stored"] = len(docs) - except Exception as e: - if isinstance(e, BulkWriteError) and e.details: - for err in e.details.get("writeErrors", []): - if err.get("code") == 11000: - results["duplicates"] += 1 - else: - results["failed"] += 1 - results["stored"] = results["total"] - results["duplicates"] - results["failed"] - else: - raise - - duration = asyncio.get_running_loop().time() - start - self.metrics.record_event_store_duration(duration, "store_batch", "event_store") - add_span_attributes(**{"events.batch.count": len(events)}) - if results["stored"] > 0: - for event in events: - self.metrics.record_event_stored(event.event_type, "event_store") - return results - except Exception as e: - self.logger.error(f"Failed to store batch: {e.__class__.__name__}: {e}", exc_info=True) - results["failed"] = results["total"] - results["stored"] - return results - - async def get_event(self, event_id: str) -> DomainEvent | None: - start = asyncio.get_running_loop().time() - doc = await EventDocument.find_one({"event_id": event_id}) - if not doc: - return None - - event = self.schema_registry.deserialize_json(doc.model_dump()) - - duration = asyncio.get_running_loop().time() - start - self.metrics.record_event_query_duration(duration, "get_by_id", "event_store") - return event - - async def get_events_by_type( - self, - event_type: EventType, - start_time: datetime | None = None, - end_time: datetime | None = None, - limit: int = 100, - offset: int = 0, - ) -> list[DomainEvent]: - start = asyncio.get_running_loop().time() - query: dict[str, Any] = {"event_type": event_type} - if tr := self._time_range(start_time, end_time): - query["timestamp"] = tr - - docs = await ( - EventDocument.find(query) - .sort([("timestamp", SortDirection.DESCENDING)]) - .skip(offset) - .limit(limit) - .to_list() - ) - events = [self.schema_registry.deserialize_json(doc.model_dump()) for doc in docs] - - duration = asyncio.get_running_loop().time() - start - self.metrics.record_event_query_duration(duration, "get_by_type", "event_store") - return events - - async def get_execution_events( - self, - execution_id: str, - event_types: list[EventType] | None = None, - ) -> list[DomainEvent]: - start = asyncio.get_running_loop().time() - query: dict[str, Any] = {"$or": [{"execution_id": execution_id}, {"aggregate_id": execution_id}]} - if event_types: - query["event_type"] = {"$in": event_types} - - docs = await EventDocument.find(query).sort([("timestamp", SortDirection.ASCENDING)]).to_list() - events = [self.schema_registry.deserialize_json(doc.model_dump()) for doc in docs] - - duration = asyncio.get_running_loop().time() - start - self.metrics.record_event_query_duration(duration, "get_execution_events", "event_store") - return events - - async def get_user_events( - self, - user_id: str, - event_types: list[EventType] | None = None, - start_time: datetime | None = None, - end_time: datetime | None = None, - limit: int = 100, - ) -> list[DomainEvent]: - start = asyncio.get_running_loop().time() - query: dict[str, Any] = {"metadata.user_id": str(user_id)} - if event_types: - query["event_type"] = {"$in": event_types} - if tr := self._time_range(start_time, end_time): - query["timestamp"] = tr - - docs = await EventDocument.find(query).sort([("timestamp", SortDirection.DESCENDING)]).limit(limit).to_list() - events = [self.schema_registry.deserialize_json(doc.model_dump()) for doc in docs] - - duration = asyncio.get_running_loop().time() - start - self.metrics.record_event_query_duration(duration, "get_user_events", "event_store") - return events - - async def get_security_events( - self, - start_time: datetime | None = None, - end_time: datetime | None = None, - user_id: str | None = None, - limit: int = 100, - ) -> list[DomainEvent]: - start = asyncio.get_running_loop().time() - query: dict[str, Any] = {"event_type": {"$in": self._SECURITY_TYPES}} - if user_id: - query["metadata.user_id"] = str(user_id) - if tr := self._time_range(start_time, end_time): - query["timestamp"] = tr - - docs = await EventDocument.find(query).sort([("timestamp", SortDirection.DESCENDING)]).limit(limit).to_list() - events = [self.schema_registry.deserialize_json(doc.model_dump()) for doc in docs] - - duration = asyncio.get_running_loop().time() - start - self.metrics.record_event_query_duration(duration, "get_security_events", "event_store") - return events - - async def get_correlation_chain(self, correlation_id: str) -> list[DomainEvent]: - start = asyncio.get_running_loop().time() - docs = await ( - EventDocument.find({"metadata.correlation_id": str(correlation_id)}) - .sort([("timestamp", SortDirection.ASCENDING)]) - .to_list() - ) - events = [self.schema_registry.deserialize_json(doc.model_dump()) for doc in docs] - - duration = asyncio.get_running_loop().time() - start - self.metrics.record_event_query_duration(duration, "get_correlation_chain", "event_store") - return events - - async def replay_events( - self, - start_time: datetime, - end_time: datetime | None = None, - event_types: list[EventType] | None = None, - callback: Callable[[DomainEvent], Awaitable[None]] | None = None, - ) -> int: - start = asyncio.get_running_loop().time() - count = 0 - - try: - query: dict[str, Any] = {"timestamp": {"$gte": start_time}} - if end_time: - query["timestamp"]["$lte"] = end_time - if event_types: - query["event_type"] = {"$in": event_types} - - async for doc in EventDocument.find(query).sort([("timestamp", SortDirection.ASCENDING)]): - event = self.schema_registry.deserialize_json(doc.model_dump()) - if callback: - await callback(event) - count += 1 - - duration = asyncio.get_running_loop().time() - start - self.metrics.record_event_query_duration(duration, "replay_events", "event_store") - self.logger.info(f"Replayed {count} events from {start_time} to {end_time}") - return count - except Exception as e: - self.logger.error(f"Failed to replay events: {e}") - return count - - async def get_event_stats( - self, - start_time: datetime | None = None, - end_time: datetime | None = None, - ) -> dict[str, Any]: - pipeline: list[dict[str, Any]] = [] - if start_time or end_time: - match: dict[str, Any] = {} - if start_time: - match["timestamp"] = {"$gte": start_time} - if end_time: - match.setdefault("timestamp", {})["$lte"] = end_time - pipeline.append({"$match": match}) - - pipeline.extend( - [ - { - "$group": { - "_id": "$event_type", - "count": {"$sum": 1}, - "first_event": {"$min": "$timestamp"}, - "last_event": {"$max": "$timestamp"}, - } - }, - {"$sort": {"count": -1}}, - ] - ) - - stats: dict[str, Any] = {"total_events": 0, "event_types": {}, "start_time": start_time, "end_time": end_time} - async for r in EventDocument.aggregate(pipeline): - et = r["_id"] - c = r["count"] - stats["event_types"][et] = { - "count": c, - "first_event": r["first_event"], - "last_event": r["last_event"], - } - stats["total_events"] += c - return stats - - def _time_range(self, start_time: datetime | None, end_time: datetime | None) -> dict[str, Any] | None: - if not start_time and not end_time: - return None - tr: dict[str, Any] = {} - if start_time: - tr["$gte"] = start_time - if end_time: - tr["$lte"] = end_time - return tr - - async def health_check(self) -> dict[str, Any]: - try: - event_count = await EventDocument.count() - return { - "healthy": True, - "event_count": event_count, - "collection": "events", - "initialized": self._initialized, - } - except Exception as e: - self.logger.error(f"Event store health check failed: {e}") - return {"healthy": False, "error": str(e)} - - -def create_event_store( - schema_registry: SchemaRegistryManager, - logger: logging.Logger, - event_metrics: EventMetrics, - ttl_days: int = 90, - batch_size: int = 100, -) -> EventStore: - return EventStore( - schema_registry=schema_registry, - logger=logger, - event_metrics=event_metrics, - ttl_days=ttl_days, - batch_size=batch_size, - ) diff --git a/backend/app/events/handlers.py b/backend/app/events/handlers.py index c3d8166c..d067cebf 100644 --- a/backend/app/events/handlers.py +++ b/backend/app/events/handlers.py @@ -1,5 +1,4 @@ import asyncio -import json import logging from collections.abc import Awaitable, Callable from datetime import datetime, timezone @@ -13,6 +12,7 @@ from app.core.tracing import EventAttributes from app.core.tracing.utils import extract_trace_context, get_tracer from app.dlq.manager import DLQManager +from app.dlq.models import DLQMessage, DLQMessageStatus from app.domain.enums.events import EventType from app.domain.enums.kafka import CONSUMER_GROUP_SUBSCRIPTIONS, GroupId, KafkaTopic from app.domain.events.typed import ( @@ -26,10 +26,6 @@ ExecutionTimeoutEvent, ) from app.domain.idempotency import KeyStrategy -from app.events.core import UnifiedProducer -from app.events.event_store import EventStore -from app.infrastructure.kafka.mappings import EVENT_TYPE_TO_TOPIC -from app.infrastructure.kafka.topics import get_all_topics from app.services.coordinator.coordinator import ExecutionCoordinator from app.services.idempotency import IdempotencyManager from app.services.k8s_worker import KubernetesWorker @@ -239,28 +235,6 @@ async def on_unhandled(body: DomainEvent) -> None: pass -def register_event_store_subscriber(broker: KafkaBroker, settings: Settings) -> None: - topics = [f"{settings.KAFKA_TOPIC_PREFIX}{t}" for t in get_all_topics()] - - @broker.subscriber( - *topics, - group_id="event-store-consumer", - ack_policy=AckPolicy.ACK, - max_poll_records=100, - ) - async def on_any_event( - body: DomainEvent, - event_store: FromDishka[EventStore], - producer: FromDishka[UnifiedProducer], - logger: FromDishka[logging.Logger], - ) -> None: - try: - await event_store.store_event(body) - except Exception as err: - logger.error(f"Error storing event {body.event_id}: {err}", exc_info=True) - topic = str(EVENT_TYPE_TO_TOPIC.get(body.event_type, "unknown")) - await producer.send_to_dlq(body, topic, err, 0) - def register_sse_subscriber(broker: KafkaBroker, settings: Settings) -> None: @broker.subscriber( @@ -316,23 +290,19 @@ async def on_unhandled(body: DomainEvent) -> None: def register_dlq_subscriber(broker: KafkaBroker, settings: Settings) -> None: """Register a DLQ subscriber that consumes dead-letter messages. - DLQ messages are plain JSON (not Avro), so a custom decoder is used - to bypass the broker-level Avro decoder. + DLQ messages are Avro-encoded DomainEvents (same as every other topic). + DLQ metadata (original_topic, error, retry_count, etc.) lives in Kafka headers. """ topic_name = f"{settings.KAFKA_TOPIC_PREFIX}{KafkaTopic.DEAD_LETTER_QUEUE}" - async def dlq_json_decoder(msg: StreamMessage[Any]) -> dict[str, Any]: - return json.loads(msg.body) # type: ignore[no-any-return] - @broker.subscriber( topic_name, group_id=GroupId.DLQ_MANAGER, ack_policy=AckPolicy.ACK, auto_offset_reset="earliest", - decoder=dlq_json_decoder, ) async def on_dlq_message( - body: dict[str, Any], + body: DomainEvent, msg: StreamMessage[Any], manager: FromDishka[DLQManager], logger: FromDishka[logging.Logger], @@ -340,22 +310,34 @@ async def on_dlq_message( start = asyncio.get_running_loop().time() raw = msg.raw_message headers = {k: v.decode() for k, v in (raw.headers or [])} - dlq_msg = manager.parse_dlq_body(body, raw.offset, raw.partition, headers) - ctx = extract_trace_context(dlq_msg.headers) + dlq_msg = DLQMessage( + event=body, + original_topic=headers.get("original_topic", ""), + error=headers.get("error", "Unknown error"), + retry_count=int(headers.get("retry_count", "0")), + failed_at=datetime.fromisoformat(headers["failed_at"]), + status=DLQMessageStatus(headers.get("status", "pending")), + producer_id=headers.get("producer_id", "unknown"), + dlq_offset=raw.offset, + dlq_partition=raw.partition, + headers=headers, + ) + + ctx = extract_trace_context(headers) with get_tracer().start_as_current_span( name="dlq.consume", context=ctx, kind=SpanKind.CONSUMER, attributes={ EventAttributes.KAFKA_TOPIC: str(manager.dlq_topic), - EventAttributes.EVENT_TYPE: dlq_msg.event.event_type, - EventAttributes.EVENT_ID: dlq_msg.event.event_id, + EventAttributes.EVENT_TYPE: body.event_type, + EventAttributes.EVENT_ID: body.event_id, }, ): await manager.handle_message(dlq_msg) - manager.metrics.record_dlq_message_received(dlq_msg.original_topic, dlq_msg.event.event_type) + manager.metrics.record_dlq_message_received(dlq_msg.original_topic, body.event_type) manager.metrics.record_dlq_message_age( (datetime.now(timezone.utc) - dlq_msg.failed_at).total_seconds() ) diff --git a/backend/app/events/schema/schema_registry.py b/backend/app/events/schema/schema_registry.py index ece3a679..af67d943 100644 --- a/backend/app/events/schema/schema_registry.py +++ b/backend/app/events/schema/schema_registry.py @@ -1,119 +1,28 @@ import logging -import struct -from functools import lru_cache -from typing import Any, get_args, get_origin from schema_registry.client import AsyncSchemaRegistryClient, schema from schema_registry.serializers import AsyncAvroMessageSerializer # type: ignore[attr-defined] -from app.domain.enums.events import EventType from app.domain.events.typed import DomainEvent from app.settings import Settings -MAGIC_BYTE = b"\x00" - - -@lru_cache(maxsize=1) -def _get_all_event_classes() -> list[type[DomainEvent]]: - """Get all concrete event classes from DomainEvent union.""" - union_type = get_args(DomainEvent)[0] # Annotated[Union[...], Discriminator] -> Union - return list(get_args(union_type)) if get_origin(union_type) else [union_type] - - -@lru_cache(maxsize=1) -def _get_event_class_mapping() -> dict[str, type[DomainEvent]]: - """Map class name -> class.""" - return {cls.__name__: cls for cls in _get_all_event_classes()} - - -@lru_cache(maxsize=1) -def _get_event_type_to_class_mapping() -> dict[EventType, type[DomainEvent]]: - """EventType -> class mapping.""" - return {cls.model_fields["event_type"].default: cls for cls in _get_all_event_classes()} - class SchemaRegistryManager: - """Schema registry manager for Avro serialization with Confluent wire format.""" + """Avro serialization via Confluent Schema Registry. + + Schemas are registered lazily by the underlying serializer on first + produce — no eager bootstrap needed. + """ def __init__(self, settings: Settings, logger: logging.Logger): self.logger = logger self.namespace = "com.integr8scode.events" self.subject_prefix = settings.SCHEMA_SUBJECT_PREFIX - parts = settings.SCHEMA_REGISTRY_AUTH.split(":", 1) - auth: tuple[str, str] | None = (parts[0], parts[1]) if len(parts) == 2 else None - self._client = AsyncSchemaRegistryClient(url=settings.SCHEMA_REGISTRY_URL, auth=auth) # type: ignore[arg-type] - self._serializer = AsyncAvroMessageSerializer(self._client) - self._schema_id_cache: dict[type[DomainEvent], int] = {} - self._id_to_class_cache: dict[int, type[DomainEvent]] = {} - - async def register_schema(self, subject: str, event_class: type[DomainEvent]) -> int: - """Register schema and return schema ID.""" - avro_schema = schema.AvroSchema(event_class.avro_schema(namespace=self.namespace)) - schema_id: int = await self._client.register(subject, avro_schema) - self._schema_id_cache[event_class] = schema_id - self._id_to_class_cache[schema_id] = event_class - self.logger.info(f"Registered schema for {event_class.__name__}: ID {schema_id}") - return schema_id - - async def _get_event_class_by_id(self, schema_id: int) -> type[DomainEvent] | None: - """Get event class by schema ID.""" - if schema_id in self._id_to_class_cache: - return self._id_to_class_cache[schema_id] - schema_obj = await self._client.get_by_id(schema_id) - if schema_obj and (class_name := schema_obj.raw_schema.get("name")): - if cls := _get_event_class_mapping().get(class_name): - self._id_to_class_cache[schema_id] = cls - self._schema_id_cache[cls] = schema_id - return cls - return None + self._client = AsyncSchemaRegistryClient(url=settings.SCHEMA_REGISTRY_URL) + self.serializer = AsyncAvroMessageSerializer(self._client) async def serialize_event(self, event: DomainEvent) -> bytes: """Serialize event to Confluent wire format: [0x00][4-byte schema id][Avro binary].""" - subject = f"{self.subject_prefix}{event.__class__.__name__}-value" - avro_schema = schema.AvroSchema(event.__class__.avro_schema(namespace=self.namespace)) - payload: dict[str, Any] = event.model_dump(mode="python", by_alias=False, exclude_unset=False) - payload.pop("event_type", None) - if "timestamp" in payload and payload["timestamp"] is not None: - payload["timestamp"] = int(payload["timestamp"].timestamp() * 1_000_000) - return await self._serializer.encode_record_with_schema(subject, avro_schema, payload) - - async def deserialize_event(self, data: bytes, topic: str) -> DomainEvent: - """Deserialize from Confluent wire format to DomainEvent.""" - if not data or len(data) < 5: - raise ValueError("Invalid message: too short for wire format") - if data[0:1] != MAGIC_BYTE: - raise ValueError(f"Unknown magic byte: {data[0]:#x}") - schema_id = struct.unpack(">I", data[1:5])[0] - event_class = await self._get_event_class_by_id(schema_id) - if not event_class: - raise ValueError(f"Unknown schema ID: {schema_id}") - obj = await self._serializer.decode_message(data) - if not isinstance(obj, dict): - raise ValueError(f"Deserialization returned {type(obj)}, expected dict") - if (f := event_class.model_fields.get("event_type")) and f.default and "event_type" not in obj: - obj["event_type"] = f.default - return event_class.model_validate(obj) - - def deserialize_json(self, data: dict[str, Any]) -> DomainEvent: - """Deserialize JSON data to DomainEvent using event_type field.""" - if not (event_type_str := data.get("event_type")): - raise ValueError("Missing event_type in event data") - if not (event_class := _get_event_type_to_class_mapping().get(EventType(event_type_str))): - raise ValueError(f"No event class found for event type: {event_type_str}") - return event_class.model_validate(data) - - async def set_compatibility(self, subject: str, mode: str) -> None: - """Set compatibility for a subject.""" - valid = {"BACKWARD", "FORWARD", "FULL", "NONE", "BACKWARD_TRANSITIVE", "FORWARD_TRANSITIVE", "FULL_TRANSITIVE"} - if mode not in valid: - raise ValueError(f"Invalid compatibility mode: {mode}") - await self._client.update_compatibility(level=mode, subject=subject) - self.logger.info(f"Set {subject} compatibility to {mode}") - - async def initialize_schemas(self) -> None: - """Initialize all event schemas in the registry.""" - for event_class in _get_all_event_classes(): - subject = f"{self.subject_prefix}{event_class.__name__}-value" - await self.set_compatibility(subject, "FORWARD") - await self.register_schema(subject, event_class) - self.logger.info(f"Initialized {len(_get_all_event_classes())} event schemas") + avro = schema.AvroSchema(event.avro_schema(namespace=self.namespace)) + subject = f"{self.subject_prefix}{avro.name}-value" + return await self.serializer.encode_record_with_schema(subject, avro, event.model_dump()) diff --git a/backend/app/main.py b/backend/app/main.py index f308fece..607ba1b7 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -42,7 +42,6 @@ ) from app.events.broker import create_broker from app.events.handlers import ( - register_event_store_subscriber, register_notification_subscriber, register_sse_subscriber, ) @@ -64,7 +63,6 @@ def create_app(settings: Settings | None = None) -> FastAPI: # Create Kafka broker and register in-app subscribers schema_registry = SchemaRegistryManager(settings, logger) broker = create_broker(settings, schema_registry, logger) - register_event_store_subscriber(broker, settings) register_sse_subscriber(broker, settings) register_notification_subscriber(broker, settings) diff --git a/backend/app/services/event_replay/replay_service.py b/backend/app/services/event_replay/replay_service.py index e3fa3740..0ff77b11 100644 --- a/backend/app/services/event_replay/replay_service.py +++ b/backend/app/services/event_replay/replay_service.py @@ -7,13 +7,14 @@ import aiofiles from opentelemetry.trace import SpanKind +from pydantic import ValidationError from app.core.metrics import ReplayMetrics from app.core.tracing.utils import trace_span from app.db.repositories.replay_repository import ReplayRepository from app.domain.admin.replay_updates import ReplaySessionUpdate from app.domain.enums.replay import ReplayStatus, ReplayTarget -from app.domain.events.typed import DomainEvent +from app.domain.events.typed import DomainEvent, DomainEventAdapter from app.domain.replay import ( CleanupResult, ReplayConfig, @@ -24,7 +25,6 @@ ReplaySessionState, ) from app.events.core import UnifiedProducer -from app.events.event_store import EventStore class EventReplayService: @@ -32,7 +32,6 @@ def __init__( self, repository: ReplayRepository, producer: UnifiedProducer, - event_store: EventStore, replay_metrics: ReplayMetrics, logger: logging.Logger, ) -> None: @@ -41,7 +40,6 @@ def __init__( self._resume_events: dict[str, asyncio.Event] = {} self._repository = repository self._producer = producer - self._event_store = event_store self.logger = logger self._file_locks: dict[str, asyncio.Lock] = {} self._metrics = replay_metrics @@ -228,10 +226,18 @@ async def _fetch_event_batches(self, session: ReplaySessionState) -> AsyncIterat if max_events and events_processed >= max_events: break - event = self._event_store.schema_registry.deserialize_json(doc) - if event: - batch.append(event) - events_processed += 1 + try: + event = DomainEventAdapter.validate_python(doc) + except ValidationError as e: + session.failed_events += 1 + self.logger.warning( + "Skipping event that failed validation", + extra={"event_id": doc.get("event_id", "unknown"), "error": str(e)}, + ) + continue + + batch.append(event) + events_processed += 1 if batch: yield batch diff --git a/backend/app/services/execution_service.py b/backend/app/services/execution_service.py index 9312f812..59a7b556 100644 --- a/backend/app/services/execution_service.py +++ b/backend/app/services/execution_service.py @@ -6,6 +6,7 @@ from app.core.correlation import CorrelationContext from app.core.metrics import ExecutionMetrics +from app.db.repositories.event_repository import EventRepository from app.db.repositories.execution_repository import ExecutionRepository from app.domain.enums.events import EventType from app.domain.enums.execution import ExecutionStatus, QueuePriority @@ -24,7 +25,6 @@ ResourceLimitsDomain, ) from app.events.core import UnifiedProducer -from app.events.event_store import EventStore from app.runtime_registry import RUNTIME_REGISTRY from app.settings import Settings @@ -49,7 +49,7 @@ def __init__( self, execution_repo: ExecutionRepository, producer: UnifiedProducer, - event_store: EventStore, + event_repository: EventRepository, settings: Settings, logger: logging.Logger, execution_metrics: ExecutionMetrics, @@ -60,14 +60,14 @@ def __init__( Args: execution_repo: Repository for execution data persistence. producer: Kafka producer for publishing events. - event_store: Event store for event persistence. + event_repository: Repository for event queries. settings: Application settings. logger: Logger instance. execution_metrics: Metrics for tracking execution operations. """ self.execution_repo = execution_repo self.producer = producer - self.event_store = event_store + self.event_repository = event_repository self.settings = settings self.logger = logger self.metrics = execution_metrics @@ -303,12 +303,10 @@ async def get_execution_events( Returns: List of events for the execution. """ - # Use the correct method name - get_execution_events instead of get_events_by_execution - events = await self.event_store.get_execution_events(execution_id=execution_id, event_types=event_types) - - # Apply limit if we got more events than requested - if len(events) > limit: - events = events[:limit] + result = await self.event_repository.get_execution_events( + execution_id=execution_id, event_types=event_types, limit=limit, + ) + events = result.events self.logger.debug( f"Retrieved {len(events)} events for execution {execution_id}", diff --git a/backend/app/services/kafka_event_service.py b/backend/app/services/kafka_event_service.py index 28d7ed8f..deca49a3 100644 --- a/backend/app/services/kafka_event_service.py +++ b/backend/app/services/kafka_event_service.py @@ -8,9 +8,8 @@ from app.core.correlation import CorrelationContext from app.core.metrics import EventMetrics -from app.db.repositories.event_repository import EventRepository from app.domain.enums.events import EventType -from app.domain.events import domain_event_adapter +from app.domain.events import DomainEventAdapter from app.domain.events.typed import DomainEvent, EventMetadata from app.events.core import UnifiedProducer from app.settings import Settings @@ -21,13 +20,11 @@ class KafkaEventService: def __init__( self, - event_repository: EventRepository, kafka_producer: UnifiedProducer, settings: Settings, logger: logging.Logger, event_metrics: EventMetrics, ): - self.event_repository = event_repository self.kafka_producer = kafka_producer self.logger = logger self.metrics = event_metrics @@ -42,17 +39,9 @@ async def publish_event( metadata: EventMetadata | None = None, ) -> str: """ - Publish an event to Kafka and store an audit copy via the repository + Build a typed DomainEvent from parameters and publish to Kafka. - Args: - event_type: Type of event (e.g., "execution.requested") - payload: Event-specific data - aggregate_id: ID of the aggregate root - correlation_id: ID for correlating related events - metadata: Event metadata (service/user/trace/IP). If None, service fills minimal defaults. - - Returns: - Event ID of published event + The producer persists the event to MongoDB before publishing. """ with tracer.start_as_current_span("publish_event") as span: span.set_attribute("event.type", event_type) @@ -85,10 +74,8 @@ async def publish_event( "metadata": event_metadata, **payload, } - domain_event = domain_event_adapter.validate_python(event_data) - await self.event_repository.store_event(domain_event) + domain_event = DomainEventAdapter.validate_python(event_data) - # Publish to Kafka (headers built automatically by producer) await self.kafka_producer.produce(event_to_produce=domain_event, key=aggregate_id or domain_event.event_id) self.metrics.record_event_published(event_type) self.metrics.record_event_processing_duration(time.time() - start_time, event_type) @@ -159,19 +146,19 @@ async def publish_pod_event( ) async def publish_domain_event(self, event: DomainEvent, key: str | None = None) -> str: - """Publish a pre-built DomainEvent to Kafka and store an audit copy.""" + """Publish a pre-built DomainEvent to Kafka. + + The producer persists the event to MongoDB before publishing. + """ with tracer.start_as_current_span("publish_domain_event") as span: span.set_attribute("event.type", event.event_type) if event.aggregate_id: span.set_attribute("aggregate.id", event.aggregate_id) start_time = time.time() - await self.event_repository.store_event(event) await self.kafka_producer.produce(event_to_produce=event, key=key or event.aggregate_id or event.event_id) self.metrics.record_event_published(event.event_type) self.metrics.record_event_processing_duration(time.time() - start_time, event.event_type) self.logger.info("Domain event published", extra={"event_id": event.event_id}) return event.event_id - - diff --git a/backend/app/settings.py b/backend/app/settings.py index 4de6729c..34522a24 100644 --- a/backend/app/settings.py +++ b/backend/app/settings.py @@ -91,11 +91,6 @@ def __init__( # Event-Driven Design Configuration KAFKA_BOOTSTRAP_SERVERS: str = "kafka:29092" SCHEMA_REGISTRY_URL: str = "http://schema-registry:8081" - SCHEMA_REGISTRY_AUTH: str = Field( - default="", - pattern=r"^($|[^:]+:.+)$", - description="Schema registry auth in 'username:password' format, or empty for no auth", - ) ENABLE_EVENT_STREAMING: bool = False EVENT_RETENTION_DAYS: int = 30 KAFKA_TOPIC_PREFIX: str = "pref" diff --git a/backend/config.test.toml b/backend/config.test.toml index ea1f11db..35829464 100644 --- a/backend/config.test.toml +++ b/backend/config.test.toml @@ -12,9 +12,9 @@ KUBERNETES_CONFIG_PATH = "/app/kubeconfig.yaml" KUBERNETES_CA_CERTIFICATE_PATH = "/app/certs/k8s-ca.pem" K8S_POD_CPU_LIMIT = "1000m" K8S_POD_MEMORY_LIMIT = "128Mi" -K8S_POD_CPU_REQUEST = "50m" +K8S_POD_CPU_REQUEST = "200m" K8S_POD_MEMORY_REQUEST = "128Mi" -K8S_POD_EXECUTION_TIMEOUT = 10 +K8S_POD_EXECUTION_TIMEOUT = 5 K8S_NAMESPACE = "integr8scode" RATE_LIMITS = "99999/second" diff --git a/backend/tests/e2e/core/test_container.py b/backend/tests/e2e/core/test_container.py index da711f02..45ac8ae5 100644 --- a/backend/tests/e2e/core/test_container.py +++ b/backend/tests/e2e/core/test_container.py @@ -170,7 +170,7 @@ async def test_execution_service_has_dependencies( # Check that key dependencies are present assert service.settings is not None assert service.execution_repo is not None - assert service.event_store is not None + assert service.event_repository is not None @pytest.mark.asyncio async def test_security_service_uses_settings( diff --git a/backend/tests/e2e/core/test_dishka_lifespan.py b/backend/tests/e2e/core/test_dishka_lifespan.py index fa4e795a..e1e387a8 100644 --- a/backend/tests/e2e/core/test_dishka_lifespan.py +++ b/backend/tests/e2e/core/test_dishka_lifespan.py @@ -88,12 +88,3 @@ async def test_sse_redis_bus_available(self, scope: AsyncContainer) -> None: bus = await scope.get(SSERedisBus) assert bus is not None - @pytest.mark.asyncio - async def test_event_store_available( - self, scope: AsyncContainer - ) -> None: - """Event store is available after lifespan.""" - from app.events.event_store import EventStore - - event_store = await scope.get(EventStore) - assert event_store is not None diff --git a/backend/tests/e2e/dlq/test_dlq_manager.py b/backend/tests/e2e/dlq/test_dlq_manager.py index a37d81f0..e19e4528 100644 --- a/backend/tests/e2e/dlq/test_dlq_manager.py +++ b/backend/tests/e2e/dlq/test_dlq_manager.py @@ -12,7 +12,7 @@ from app.dlq.models import DLQMessage from app.domain.enums.events import EventType from app.domain.enums.kafka import KafkaTopic -from app.domain.events.typed import DLQMessageReceivedEvent +from app.domain.events.typed import DLQMessageReceivedEvent, DomainEventAdapter from app.events.schema.schema_registry import SchemaRegistryManager from app.settings import Settings from dishka import AsyncContainer @@ -53,7 +53,10 @@ async def consume_dlq_events() -> None: """Consume DLQ events and set future when our event is received.""" async for msg in events_consumer: try: - event = await schema_registry.deserialize_event(msg.value, dlq_events_topic) + payload = await schema_registry.serializer.decode_message(msg.value) + if payload is None: + continue + event = DomainEventAdapter.validate_python(payload) if ( isinstance(event, DLQMessageReceivedEvent) and event.dlq_event_id == ev.event_id diff --git a/backend/tests/e2e/events/test_event_store.py b/backend/tests/e2e/events/test_event_store.py deleted file mode 100644 index 8bd82173..00000000 --- a/backend/tests/e2e/events/test_event_store.py +++ /dev/null @@ -1,153 +0,0 @@ -import logging -import uuid -from datetime import datetime, timedelta, timezone - -import pytest -from app.db.docs import EventDocument -from app.domain.enums.events import EventType -from app.domain.events.typed import DomainEvent -from app.events.event_store import EventStore -from dishka import AsyncContainer - -from tests.conftest import make_execution_requested_event - -pytestmark = [pytest.mark.e2e, pytest.mark.mongodb] - -_test_logger = logging.getLogger("test.events.event_store") - - -@pytest.mark.asyncio -async def test_event_store_stores_single_event(scope: AsyncContainer) -> None: - """Test that EventStore.store_event() persists an event to MongoDB.""" - store: EventStore = await scope.get(EventStore) - - # Create a unique event - execution_id = f"exec-{uuid.uuid4().hex[:8]}" - event = make_execution_requested_event(execution_id=execution_id) - - # Store the event - result = await store.store_event(event) - assert result is True - - # Verify it's in MongoDB - doc = await EventDocument.find_one({"event_id": event.event_id}) - assert doc is not None - assert doc.event_id == event.event_id - assert doc.event_type == EventType.EXECUTION_REQUESTED - assert doc.aggregate_id == execution_id - assert doc.stored_at is not None - assert doc.ttl_expires_at is not None - # TTL should be ~90 days in the future - assert doc.ttl_expires_at > datetime.now(timezone.utc) + timedelta(days=89) - - -@pytest.mark.asyncio -async def test_event_store_stores_batch(scope: AsyncContainer) -> None: - """Test that EventStore.store_batch() persists multiple events.""" - store: EventStore = await scope.get(EventStore) - - # Create multiple unique events - events: list[DomainEvent] = [ - make_execution_requested_event(execution_id=f"exec-batch-{uuid.uuid4().hex[:8]}") - for _ in range(5) - ] - - # Store the batch - results = await store.store_batch(events) - - assert results["total"] == 5 - assert results["stored"] == 5 - assert results["duplicates"] == 0 - assert results["failed"] == 0 - - # Verify all events are in MongoDB - for event in events: - doc = await EventDocument.find_one({"event_id": event.event_id}) - assert doc is not None - assert doc.event_type == EventType.EXECUTION_REQUESTED - - -@pytest.mark.asyncio -async def test_event_store_handles_duplicates(scope: AsyncContainer) -> None: - """Test that EventStore handles duplicate event IDs gracefully.""" - store: EventStore = await scope.get(EventStore) - - # Create an event - event = make_execution_requested_event(execution_id=f"exec-dup-{uuid.uuid4().hex[:8]}") - - # Store it twice - result1 = await store.store_event(event) - result2 = await store.store_event(event) - - # Both should succeed (second is a no-op due to duplicate handling) - assert result1 is True - assert result2 is True - - # Only one document should exist - count = await EventDocument.find({"event_id": event.event_id}).count() - assert count == 1 - - -@pytest.mark.asyncio -async def test_event_store_batch_handles_duplicates(scope: AsyncContainer) -> None: - """Test that store_batch handles duplicates within the batch.""" - store: EventStore = await scope.get(EventStore) - - # Create an event and store it first - event = make_execution_requested_event(execution_id=f"exec-batch-dup-{uuid.uuid4().hex[:8]}") - await store.store_event(event) - - # Create a batch with one new event and one duplicate - new_event = make_execution_requested_event(execution_id=f"exec-batch-new-{uuid.uuid4().hex[:8]}") - batch: list[DomainEvent] = [new_event, event] # event is already stored - - results = await store.store_batch(batch) - - assert results["total"] == 2 - assert results["stored"] == 1 # Only the new one - assert results["duplicates"] == 1 # The duplicate - - -@pytest.mark.asyncio -async def test_event_store_retrieves_by_id(scope: AsyncContainer) -> None: - """Test that EventStore.get_event() retrieves a stored event.""" - store: EventStore = await scope.get(EventStore) - - # Create and store an event - execution_id = f"exec-get-{uuid.uuid4().hex[:8]}" - event = make_execution_requested_event(execution_id=execution_id, script="print('test')") - await store.store_event(event) - - # Retrieve it - retrieved = await store.get_event(event.event_id) - - assert retrieved is not None - assert retrieved.event_id == event.event_id - assert retrieved.event_type == EventType.EXECUTION_REQUESTED - - -@pytest.mark.asyncio -async def test_event_store_retrieves_by_type(scope: AsyncContainer) -> None: - """Test that EventStore.get_events_by_type() works correctly.""" - store: EventStore = await scope.get(EventStore) - - # Store a few events - unique_prefix = uuid.uuid4().hex[:8] - events: list[DomainEvent] = [ - make_execution_requested_event(execution_id=f"exec-type-{unique_prefix}-{i}") - for i in range(3) - ] - await store.store_batch(events) - - # Query by type - retrieved = await store.get_events_by_type( - EventType.EXECUTION_REQUESTED, - limit=100, - ) - - # Should find at least our 3 events - assert len(retrieved) >= 3 - - # All should be EXECUTION_REQUESTED - for ev in retrieved: - assert ev.event_type == EventType.EXECUTION_REQUESTED diff --git a/backend/tests/e2e/events/test_schema_registry_real.py b/backend/tests/e2e/events/test_schema_registry_real.py index d6c182de..58e4900d 100644 --- a/backend/tests/e2e/events/test_schema_registry_real.py +++ b/backend/tests/e2e/events/test_schema_registry_real.py @@ -1,9 +1,9 @@ import logging import pytest -from app.domain.events.typed import EventMetadata, PodCreatedEvent + +from app.domain.events.typed import DomainEventAdapter, EventMetadata, PodCreatedEvent from app.events.schema.schema_registry import SchemaRegistryManager -from app.infrastructure.kafka.mappings import get_topic_for_event from app.settings import Settings pytestmark = [pytest.mark.e2e, pytest.mark.kafka] @@ -22,7 +22,8 @@ async def test_serialize_and_deserialize_event_real_registry(test_settings: Sett metadata=EventMetadata(service_name="s", service_version="1"), ) data = await m.serialize_event(ev) - topic = str(get_topic_for_event(ev.event_type)) - obj = await m.deserialize_event(data, topic=topic) + payload = await m.serializer.decode_message(data) + assert payload is not None + obj = DomainEventAdapter.validate_python(payload) assert isinstance(obj, PodCreatedEvent) assert obj.namespace == "n" diff --git a/backend/tests/e2e/events/test_schema_registry_roundtrip.py b/backend/tests/e2e/events/test_schema_registry_roundtrip.py index c15faca2..1fc83467 100644 --- a/backend/tests/e2e/events/test_schema_registry_roundtrip.py +++ b/backend/tests/e2e/events/test_schema_registry_roundtrip.py @@ -1,9 +1,9 @@ import logging import pytest -from app.events.schema.schema_registry import MAGIC_BYTE, SchemaRegistryManager -from app.infrastructure.kafka.mappings import get_topic_for_event -from app.settings import Settings + +from app.domain.events.typed import DomainEventAdapter +from app.events.schema.schema_registry import SchemaRegistryManager from dishka import AsyncContainer from tests.conftest import make_execution_requested_event @@ -16,20 +16,10 @@ @pytest.mark.asyncio async def test_schema_registry_serialize_deserialize_roundtrip(scope: AsyncContainer) -> None: reg: SchemaRegistryManager = await scope.get(SchemaRegistryManager) - # Schema registration happens lazily in serialize_event ev = make_execution_requested_event(execution_id="e-rt") data = await reg.serialize_event(ev) - assert data.startswith(MAGIC_BYTE) - topic = str(get_topic_for_event(ev.event_type)) - back = await reg.deserialize_event(data, topic=topic) + assert data[:1] == b"\x00" # Confluent wire format magic byte + payload = await reg.serializer.decode_message(data) + assert payload is not None + back = DomainEventAdapter.validate_python(payload) assert back.event_id == ev.event_id and getattr(back, "execution_id", None) == ev.execution_id - - # initialize_schemas should be a no-op if already initialized; call to exercise path - await reg.initialize_schemas() - - -@pytest.mark.asyncio -async def test_schema_registry_deserialize_invalid_header(test_settings: Settings) -> None: - reg = SchemaRegistryManager(settings=test_settings, logger=_test_logger) - with pytest.raises(ValueError): - await reg.deserialize_event(b"\x01\x00\x00\x00\x01", topic="t") # wrong magic byte diff --git a/backend/tests/unit/domain/events/test_event_schema_coverage.py b/backend/tests/unit/domain/events/test_event_schema_coverage.py index dd150e3e..5888ed92 100644 --- a/backend/tests/unit/domain/events/test_event_schema_coverage.py +++ b/backend/tests/unit/domain/events/test_event_schema_coverage.py @@ -12,8 +12,7 @@ from typing import get_args from app.domain.enums.events import EventType -from app.domain.events.typed import BaseEvent, DomainEvent, domain_event_adapter -from app.events.schema.schema_registry import _get_event_type_to_class_mapping +from app.domain.events.typed import BaseEvent, DomainEvent, DomainEventAdapter def get_domain_event_classes() -> dict[EventType, type]: @@ -50,8 +49,8 @@ def get_domain_event_classes() -> dict[EventType, type]: def get_kafka_event_classes() -> dict[EventType, type]: - """Extract EventType -> class mapping from Kafka DomainEvent subclasses.""" - return _get_event_type_to_class_mapping() + """Extract EventType -> class mapping from DomainEvent union (same source).""" + return get_domain_event_classes() class TestEventSchemaCoverage: @@ -84,15 +83,15 @@ def test_all_event_types_have_kafka_event_class(self) -> None: ) ) - def test_domain_event_adapter_covers_all_types(self) -> None: - """The domain_event_adapter TypeAdapter must handle all EventTypes.""" + def test_DomainEventAdapter_covers_all_types(self) -> None: + """The DomainEventAdapter TypeAdapter must handle all EventTypes.""" errors: list[str] = [] for et in EventType: try: # Validation will fail due to missing required fields, but that's OK # We just want to confirm the type IS in the union (not "unknown discriminator") - domain_event_adapter.validate_python({"event_type": et}) + DomainEventAdapter.validate_python({"event_type": et}) except Exception as e: error_str = str(e).lower() # "validation error" means type IS recognized but fields are missing - that's fine @@ -100,7 +99,7 @@ def test_domain_event_adapter_covers_all_types(self) -> None: if "no match" in error_str or "unable to extract" in error_str: errors.append(f" - {et.value}: not in DomainEvent union") - assert not errors, f"domain_event_adapter missing {len(errors)} type(s):\n" + "\n".join(errors) + assert not errors, f"DomainEventAdapter missing {len(errors)} type(s):\n" + "\n".join(errors) def test_no_orphan_domain_event_classes(self) -> None: """All domain event classes must have a corresponding EventType.""" diff --git a/backend/tests/unit/events/test_schema_registry_manager.py b/backend/tests/unit/events/test_schema_registry_manager.py index 118c4c7c..cdc3159b 100644 --- a/backend/tests/unit/events/test_schema_registry_manager.py +++ b/backend/tests/unit/events/test_schema_registry_manager.py @@ -1,16 +1,11 @@ -import logging - import pytest -from app.domain.enums.execution import QueuePriority -from app.domain.events.typed import ExecutionRequestedEvent -from app.events.schema.schema_registry import SchemaRegistryManager -from app.settings import Settings +from pydantic import ValidationError -_test_logger = logging.getLogger("test.events.schema_registry_manager") +from app.domain.enums.execution import QueuePriority +from app.domain.events.typed import DomainEventAdapter, ExecutionRequestedEvent -def test_deserialize_json_execution_requested(test_settings: Settings) -> None: - m = SchemaRegistryManager(test_settings, logger=_test_logger) +def test_domain_event_adapter_execution_requested() -> None: data = { "event_type": "execution_requested", "execution_id": "e1", @@ -28,13 +23,12 @@ def test_deserialize_json_execution_requested(test_settings: Settings) -> None: "priority": QueuePriority.NORMAL, "metadata": {"service_name": "t", "service_version": "1.0"}, } - ev = m.deserialize_json(data) + ev = DomainEventAdapter.validate_python(data) assert isinstance(ev, ExecutionRequestedEvent) assert ev.execution_id == "e1" assert ev.language == "python" -def test_deserialize_json_missing_type_raises(test_settings: Settings) -> None: - m = SchemaRegistryManager(test_settings, logger=_test_logger) - with pytest.raises(ValueError): - m.deserialize_json({}) +def test_domain_event_adapter_missing_type_raises() -> None: + with pytest.raises(ValidationError): + DomainEventAdapter.validate_python({}) diff --git a/backend/tests/unit/services/pod_monitor/test_monitor.py b/backend/tests/unit/services/pod_monitor/test_monitor.py index 1c0bdec8..14f0a61d 100644 --- a/backend/tests/unit/services/pod_monitor/test_monitor.py +++ b/backend/tests/unit/services/pod_monitor/test_monitor.py @@ -5,7 +5,6 @@ import pytest from app.core.metrics import EventMetrics, KubernetesMetrics -from app.db.repositories.event_repository import EventRepository from app.domain.events.typed import ( DomainEvent, EventMetadata, @@ -42,18 +41,6 @@ # ===== Test doubles for KafkaEventService dependencies ===== -class FakeEventRepository(EventRepository): - """In-memory event repository for testing.""" - - def __init__(self) -> None: - super().__init__(_test_logger) - self.stored_events: list[DomainEvent] = [] - - async def store_event(self, event: DomainEvent) -> str: - self.stored_events.append(event) - return event.event_id - - class FakeUnifiedProducer(UnifiedProducer): """Fake producer that captures events without Kafka.""" @@ -74,11 +61,9 @@ async def aclose(self) -> None: def create_test_kafka_event_service(event_metrics: EventMetrics) -> tuple[KafkaEventService, FakeUnifiedProducer]: """Create real KafkaEventService with fake dependencies for testing.""" fake_producer = FakeUnifiedProducer() - fake_repo = FakeEventRepository() settings = Settings(config_path="config.test.toml") service = KafkaEventService( - event_repository=fake_repo, kafka_producer=fake_producer, settings=settings, logger=_test_logger, @@ -406,9 +391,7 @@ async def produce( raise RuntimeError("Publish failed") failing_producer = FailingProducer() - fake_repo = FakeEventRepository() failing_service = KafkaEventService( - event_repository=fake_repo, kafka_producer=failing_producer, settings=Settings(config_path="config.test.toml"), logger=_test_logger,