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

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

@@ -8,7 +8,6 @@
 
 from app.api.dependencies import admin_user
 from app.core.correlation import CorrelationContext
-from app.domain.enums.events import EventType
 from app.domain.events.event_models import EventFilter
 from app.domain.replay import ReplayFilter
 from app.schemas_pydantic.admin_events import (
@@ -69,14 +68,14 @@ async def get_event_stats(
 @router.get("/export/csv")
 async def export_events_csv(
     service: FromDishka[AdminEventsService],
-    event_types: list[EventType] | None = Query(None, description="Event types (repeat param for multiple)"),
+    topics: list[str] | None = Query(None, description="Event topics (repeat param for multiple)"),
     start_time: datetime | None = Query(None, description="Start time"),
     end_time: datetime | None = Query(None, description="End time"),
     limit: int = Query(default=10000, le=50000),
 ) -> StreamingResponse:
     try:
         export_filter = EventFilter(
-            event_types=event_types,
+            topics=topics,
             start_time=start_time,
             end_time=end_time,
         )
@@ -94,7 +93,7 @@ async def export_events_csv(
 @router.get("/export/json")
 async def export_events_json(
     service: FromDishka[AdminEventsService],
-    event_types: list[EventType] | None = Query(None, description="Event types (repeat param for multiple)"),
+    topics: list[str] | None = Query(None, description="Event topics (repeat param for multiple)"),
     aggregate_id: str | None = Query(None, description="Aggregate ID filter"),
     correlation_id: str | None = Query(None, description="Correlation ID filter"),
     user_id: str | None = Query(None, description="User ID filter"),
@@ -106,7 +105,7 @@ async def export_events_json(
     """Export events as JSON with comprehensive filtering."""
     try:
         export_filter = EventFilter(
-            event_types=event_types,
+            topics=topics,
             aggregate_id=aggregate_id,
             correlation_id=correlation_id,
             user_id=user_id,

backend/app/api/routes/dlq.py

@@ -7,7 +7,6 @@
 from app.dlq import RetryPolicy
 from app.dlq.manager import DLQManager
 from app.dlq.models import DLQMessageStatus
-from app.domain.enums.events import EventType
 from app.schemas_pydantic.dlq import (
     DLQBatchRetryResponse,
     DLQMessageDetail,
@@ -36,12 +35,11 @@ async def get_dlq_messages(
     repository: FromDishka[DLQRepository],
     status: DLQMessageStatus | None = Query(None),
     topic: str | None = None,
-    event_type: EventType | None = Query(None),
     limit: int = Query(50, ge=1, le=1000),
     offset: int = Query(0, ge=0),
 ) -> DLQMessagesResponse:
     result = await repository.get_messages(
-        status=status, topic=topic, event_type=event_type, limit=limit, offset=offset
+        status=status, topic=topic, limit=limit, offset=offset
     )
 
     # Convert domain messages to response models using model_validate

backend/app/api/routes/events.py

@@ -11,10 +11,10 @@
 from app.core.correlation import CorrelationContext
 from app.core.utils import get_client_ip
 from app.domain.enums.common import SortOrder
-from app.domain.enums.events import EventType
 from app.domain.enums.user import UserRole
 from app.domain.events.event_models import EventFilter
-from app.domain.events.typed import BaseEvent, DomainEvent, EventMetadata
+from app.domain.events.typed import BaseEvent, EventMetadata
+from app.events.core import EventPublisher
 from app.schemas_pydantic.events import (
     DeleteEventResponse,
     EventAggregationRequest,
@@ -28,7 +28,6 @@
 from app.schemas_pydantic.user import UserResponse
 from app.services.event_service import EventService
 from app.services.execution_service import ExecutionService
-from app.services.kafka_event_service import KafkaEventService
 from app.settings import Settings
 
 router = APIRouter(prefix="/events", tags=["events"], route_class=DishkaRoute)
@@ -74,7 +73,7 @@ async def get_execution_events(
 async def get_user_events(
     current_user: Annotated[UserResponse, Depends(current_user)],
     event_service: FromDishka[EventService],
-    event_types: list[EventType] | None = Query(None),
+    topics: list[str] | None = Query(None),
     start_time: datetime | None = Query(None),
     end_time: datetime | None = Query(None),
     limit: int = Query(100, ge=1, le=1000),
@@ -84,7 +83,7 @@ async def get_user_events(
     """Get events for the current user"""
     result = await event_service.get_user_events_paginated(
         user_id=current_user.user_id,
-        event_types=event_types,
+        topics=topics,
         start_time=start_time,
         end_time=end_time,
         limit=limit,
@@ -108,7 +107,7 @@ async def query_events(
     event_service: FromDishka[EventService],
 ) -> EventListResponse:
     event_filter = EventFilter(
-        event_types=filter_request.event_types,
+        topics=filter_request.topics,
         aggregate_id=filter_request.aggregate_id,
         correlation_id=filter_request.correlation_id,
         user_id=filter_request.user_id,
@@ -218,10 +217,10 @@ async def get_event_statistics(
     return EventStatistics.model_validate(stats)
 
 
-@router.get("/{event_id}", response_model=DomainEvent)
+@router.get("/{event_id}", response_model=BaseEvent)
 async def get_event(
     event_id: str, current_user: Annotated[UserResponse, Depends(current_user)], event_service: FromDishka[EventService]
-) -> DomainEvent:
+) -> BaseEvent:
     """Get a specific event by ID"""
     event = await event_service.get_event(event_id=event_id, user_id=current_user.user_id, user_role=current_user.role)
     if event is None:
@@ -234,28 +233,30 @@ async def publish_custom_event(
     admin: Annotated[UserResponse, Depends(admin_user)],
     event_request: PublishEventRequest,
     request: Request,
-    event_service: FromDishka[KafkaEventService],
+    producer: FromDishka[EventPublisher],
     settings: FromDishka[Settings],
 ) -> PublishEventResponse:
+    """Publish a custom event (admin only). Creates a BaseEvent with the provided payload."""
     base_meta = EventMetadata(
         service_name=settings.SERVICE_NAME,
         service_version=settings.SERVICE_VERSION,
         user_id=admin.user_id,
         ip_address=get_client_ip(request),
         user_agent=request.headers.get("user-agent"),
+        correlation_id=event_request.correlation_id or CorrelationContext.get_correlation_id(),
     )
     # Merge any additional metadata provided in request (extra allowed)
     if event_request.metadata:
         base_meta = base_meta.model_copy(update=event_request.metadata)
 
-    event_id = await event_service.publish_event(
-        event_type=event_request.event_type,
-        payload=event_request.payload,
+    # Create a BaseEvent with the custom payload in metadata
+    event = BaseEvent(
         aggregate_id=event_request.aggregate_id,
-        correlation_id=event_request.correlation_id,
         metadata=base_meta,
     )
 
+    event_id = await producer.publish(event=event, key=event_request.aggregate_id)
+
     return PublishEventResponse(event_id=event_id, status="published", timestamp=datetime.now(timezone.utc))
 
 
@@ -275,12 +276,12 @@ async def aggregate_events(
     return result.results
 
 
-@router.get("/types/list", response_model=list[str])
-async def list_event_types(
+@router.get("/topics/list", response_model=list[str])
+async def list_topics(
     current_user: Annotated[UserResponse, Depends(current_user)], event_service: FromDishka[EventService]
 ) -> list[str]:
-    event_types = await event_service.list_event_types(user_id=current_user.user_id, user_role=current_user.role)
-    return event_types
+    topics = await event_service.list_topics(user_id=current_user.user_id, user_role=current_user.role)
+    return topics
 
 
 @router.delete("/{event_id}", response_model=DeleteEventResponse)
@@ -300,7 +301,6 @@ async def delete_event(
         extra={
             "event_id": event_id,
             "admin_email": admin.email,
-            "event_type": result.event_type,
             "aggregate_id": result.aggregate_id,
             "correlation_id": result.metadata.correlation_id,
         },
@@ -316,8 +316,7 @@ async def replay_aggregate_events(
     aggregate_id: str,
     admin: Annotated[UserResponse, Depends(admin_user)],
     event_service: FromDishka[EventService],
-    kafka_event_service: FromDishka[KafkaEventService],
-    settings: FromDishka[Settings],
+    producer: FromDishka[EventPublisher],
     logger: FromDishka[logging.Logger],
     target_service: str | None = Query(None, description="Service to replay events to"),
     dry_run: bool = Query(True, description="If true, only show what would be replayed"),
@@ -331,7 +330,7 @@ async def replay_aggregate_events(
             dry_run=True,
             aggregate_id=aggregate_id,
             event_count=replay_info.event_count,
-            event_types=replay_info.event_types,
+            topics=replay_info.topics,
             start_time=replay_info.start_time,
             end_time=replay_info.end_time,
         )
@@ -346,21 +345,9 @@ async def replay_aggregate_events(
             await asyncio.sleep(0.1)
 
         try:
-            meta = EventMetadata(
-                service_name=settings.SERVICE_NAME,
-                service_version=settings.SERVICE_VERSION,
-                user_id=admin.user_id,
-            )
-            # Extract payload fields (exclude base event fields + event_type discriminator)
-            base_fields = set(BaseEvent.model_fields.keys()) | {"event_type"}
-            extra_fields = {k: v for k, v in event.model_dump().items() if k not in base_fields}
-            await kafka_event_service.publish_event(
-                event_type=event.event_type,
-                payload=extra_fields,
-                aggregate_id=aggregate_id,
-                correlation_id=replay_correlation_id,
-                metadata=meta,
-            )
+            # Update correlation_id for replay tracking
+            event.metadata.correlation_id = replay_correlation_id
+            await producer.publish(event=event, key=aggregate_id)
             replayed_count += 1
         except Exception as e:
             logger.error(f"Failed to replay event {event.event_id}: {e}")

backend/app/api/routes/execution.py

@@ -9,10 +9,9 @@
 from app.api.dependencies import admin_user, current_user
 from app.core.tracing import EventAttributes, add_span_attributes
 from app.core.utils import get_client_ip
-from app.domain.enums.events import EventType
 from app.domain.enums.execution import ExecutionStatus
 from app.domain.enums.user import UserRole
-from app.domain.events.typed import BaseEvent, DomainEvent, EventMetadata
+from app.domain.events.typed import BaseEvent, EventMetadata
 from app.domain.exceptions import DomainError
 from app.domain.idempotency import KeyStrategy
 from app.schemas_pydantic.execution import (
@@ -32,8 +31,6 @@
 from app.services.event_service import EventService
 from app.services.execution_service import ExecutionService
 from app.services.idempotency import IdempotencyManager
-from app.services.kafka_event_service import KafkaEventService
-from app.settings import Settings
 
 router = APIRouter(route_class=DishkaRoute, tags=["execution"])
 
@@ -76,10 +73,9 @@ async def create_execution(
     # Handle idempotency if key provided
     pseudo_event = None
     if idempotency_key:
-        # Create a pseudo-event for idempotency tracking
+        # Create a pseudo-event for idempotency tracking (just needs event_id for key generation)
         pseudo_event = BaseEvent(
             event_id=str(uuid4()),
-            event_type=EventType.EXECUTION_REQUESTED,
             timestamp=datetime.now(timezone.utc),
             metadata=EventMetadata(
                 user_id=current_user.user_id, correlation_id=str(uuid4()), service_name="api", service_version="1.0.0"
@@ -160,8 +156,7 @@ async def cancel_execution(
         execution: Annotated[ExecutionInDB, Depends(get_execution_with_access)],
         current_user: Annotated[UserResponse, Depends(current_user)],
         cancel_request: CancelExecutionRequest,
-        event_service: FromDishka[KafkaEventService],
-        settings: FromDishka[Settings],
+        execution_service: FromDishka[ExecutionService],
 ) -> CancelResponse:
     # Handle terminal states
     terminal_states = [ExecutionStatus.COMPLETED, ExecutionStatus.FAILED, ExecutionStatus.TIMEOUT]
@@ -175,32 +170,18 @@ async def cancel_execution(
             execution_id=execution.execution_id,
             status="already_cancelled",
             message="Execution was already cancelled",
-            event_id="-1",  # exact event_id unknown
         )
 
-    payload = {
-        "execution_id": execution.execution_id,
-        "status": str(ExecutionStatus.CANCELLED),
-        "reason": cancel_request.reason or "User requested cancellation",
-        "previous_status": str(execution.status),
-    }
-    meta = EventMetadata(
-        service_name=settings.SERVICE_NAME,
-        service_version=settings.SERVICE_VERSION,
+    await execution_service.cancel_execution(
+        execution_id=execution.execution_id,
+        reason=cancel_request.reason or "User requested cancellation",
         user_id=current_user.user_id,
     )
-    event_id = await event_service.publish_event(
-        event_type=EventType.EXECUTION_CANCELLED,
-        payload=payload,
-        aggregate_id=execution.execution_id,
-        metadata=meta,
-    )
 
     return CancelResponse(
         execution_id=execution.execution_id,
         status="cancellation_requested",
         message="Cancellation request submitted",
-        event_id=event_id,
     )
 
 
@@ -231,16 +212,16 @@ async def retry_execution(
     return ExecutionResponse.model_validate(new_result)
 
 
-@router.get("/executions/{execution_id}/events", response_model=list[DomainEvent])
+@router.get("/executions/{execution_id}/events", response_model=list[BaseEvent])
 async def get_execution_events(
         execution: Annotated[ExecutionInDB, Depends(get_execution_with_access)],
         event_service: FromDishka[EventService],
-        event_types: list[EventType] | None = Query(None, description="Event types to filter"),
+        topics: list[str] | None = Query(None, description="Event topics to filter"),
         limit: int = Query(100, ge=1, le=1000),
-) -> list[DomainEvent]:
+) -> list[BaseEvent]:
     """Get all events for an execution."""
     events = await event_service.get_events_by_aggregate(
-        aggregate_id=execution.execution_id, event_types=event_types, limit=limit
+        aggregate_id=execution.execution_id, topics=topics, limit=limit
     )
     return events