fix: removed most of schema encode/decode overhead

.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 ctr images pull docker.io/library/python:3.11-slim
+
     - name: Wait for image pull and infra
       shell: bash
       run: |

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,15 +617,15 @@ 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,
     ) -> ExecutionService:
         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,
         )

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 (

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"],

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,

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",

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)