fix: removed event store consumer and lifecycle

[HardMax71]

---

Summary by cubic

Removed the EventStoreConsumer and lifecycle management. The Kafka consumer now lives in the DI provider and dispatches events directly to EventStore, simplifying startup and reducing moving parts.

• Refactors
  • Deleted LifecycleEnabled and EventStoreConsumer; event handling now calls EventStore.store_event per event.
  • Provider sets up EventDispatcher and UnifiedConsumer for all topics, registers a DLQ error handler, and manages start/stop.
  • Cleaned up FastAPI lifespan; no AsyncExitStack or explicit consumer start.
  • Updated dispatcher typing and widened consumer error callback to include the topic; simplified DLQ handlers and updated tests for EventStore availability and DLQ behavior.

^{Written for commit d1dc667. Summary will update on new commits.}

Summary by CodeRabbit

• Refactor

  • Simplified and consolidated the event subsystem and lifecycle startup/shutdown behavior.
  • Event handlers now can return values (broader handler flexibility).
  • Error handling for failed events now receives message topic context and DLQ routing uses the per-message topic.

• Tests

  • Updated end-to-end tests to match the new event and DLQ behaviors.

[coderabbitai]

📝 Walkthrough

Walkthrough

Consolidates event-store consumer lifecycle into DI providers: removes LifecycleEnabled and EventStoreConsumer, provisions and starts the EventStore/consumer inside the provider, updates DLQ handler signatures, and broadens event handler return typing.

Changes

Cohort / File(s)	Summary
Lifecycle abstraction removed backend/app/core/lifecycle.py	Deleted LifecycleEnabled base class and its async context-manager lifecycle implementation.
DI lifespan & imports backend/app/core/dishka_lifespan.py, backend/tests/e2e/core/test_dishka_lifespan.py	Replaced EventStoreConsumer with EventStore in DI retrieval and tests; simplified lifespan wiring to yield the EventStore.
Provider refactor / consumer wiring backend/app/core/providers.py	Refactored get_event_store to return AsyncIterator[EventStore]; now builds EventStore, wires EventDispatcher, creates/starts UnifiedConsumer, attaches DLQ handler, and performs cleanup on exit.
Event consumer removed backend/app/events/event_store_consumer.py	Removed EventStoreConsumer implementation and its factory; consumer responsibilities moved into provider wiring.
Consumer error-callback signature backend/app/events/core/consumer.py	Expanded error-callback signature to accept (Exception, DomainEvent, topic: str) and updated invocation sites.
DLQ handler signature change backend/app/events/core/dlq_handler.py, backend/tests/e2e/events/test_dlq_handler.py	Removed stored original_topic; DLQ creator now returns handlers with signature (Exception, DomainEvent, topic: str) and tests updated to pass topic per-call.
Dispatcher handler typing backend/app/events/core/dispatcher.py	Broadened EventHandler type to Callable[[DomainEvent], Awaitable[Any]]; updated _handlers typing accordingly.

Sequence Diagram(s)

mermaid
sequenceDiagram
participant Lifespan as "Lifespan / Provider"
participant EventStore as "EventStore"
participant Dispatcher as "EventDispatcher"
participant Consumer as "UnifiedConsumer"
participant DLQ as "DLQ Handler / Producer"

Lifespan->>EventStore: build/configure EventStore
Lifespan->>Dispatcher: register all EventType handlers
Lifespan->>Consumer: create UnifiedConsumer(config, dispatcher, schema)
Lifespan->>Consumer: register_error_callback(create_dlq_error_handler)
Lifespan->>Consumer: start()
Note right of Consumer: consumes messages -> dispatch via Dispatcher
Consumer->>Dispatcher: dispatch(event)
alt handler error
    Dispatcher->>DLQ: error callback invoked (error, event, topic)
    DLQ->>DLQ: send_to_dlq(event, topic)
end
Lifespan->>Consumer: stop() on exit
Lifespan->>EventStore: cleanup / aclose()

Estimated code review effort

🎯 4 (Complex) | ⏱️ ~60 minutes

Possibly related PRs

• fix: moved lifetime handling to DI, now stateless result processor #130: Removes LifecycleEnabled and moves lifecycle management into DI providers — similar lifecycle-to-provider refactor pattern touching providers and lifespan wiring.
• feat: confluence kafka -> aiokafka #75: Refactors Kafka/SchemaRegistry consumer and producer stack to async clients — related Kafka/consumer interface changes this PR builds on.
• fix: stateless dlq manager, updated other files accordingly #126: Changes DLQ plumbing and provider wiring — overlaps with DLQ handler signature and provider integration edits here.

Poem

🐰 I hopped from class to provider bright,
I traded context for a tidy DI light.
Dispatchers hum and DLQs now sing,
Consumers start where the providers bring.
Tiny paws clap—refactor takes flight!

🚥 Pre-merge checks | ✅ 2 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 41.18% which is insufficient. The required threshold is 80.00%.	Write docstrings for the functions missing them to satisfy the coverage threshold.

✅ Passed checks (2 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title 'fix: removed event store consumer and lifecycle' accurately describes the main changes: deletion of EventStoreConsumer class and LifecycleEnabled base class, with event consumer logic consolidated into the DI provider.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing touches

• 📝 Generate docstrings

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Post copyable unit tests in a comment
• Commit unit tests in branch fix/event-store-consumer

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[cubic-dev-ai]

No issues found across 6 files

[coderabbitai]

Actionable comments posted: 1

Caution

Some comments are outside the diff and can’t be posted inline due to platform limitations.

⚠️ Outside diff range comments (1)

backend/app/core/dishka_lifespan.py (1)

79-107: ⚠️ Potential issue | 🔴 Critical

Add explicit container shutdown to prevent resource leaks.

The lifespan ends with a bare yield but never closes the Dishka container. Per Dishka's FastAPI integration pattern, setup_dishka() manages request/session scopes automatically, but app-scoped resource cleanup is your responsibility. Add:

yield
await app.state.dishka_container.close()

Without this, app-scoped providers (Kafka consumers, Redis, MongoDB connections, etc.) won't finalize on shutdown, causing resource leaks and potential hangs during graceful shutdown.

🧹 Nitpick comments (1)

backend/app/events/core/dispatcher.py (1)

10-54: Align register typing with the new EventHandler alias.

EventHandler now allows Awaitable[Any], but the decorator still constrains handlers to Awaitable[None], which blocks handlers returning values from using the decorator and can trigger type-checker errors. Consider widening the decorator signature for consistency.

♻️ Proposed typing alignment

 def register(
     self, event_type: EventType
-) -> Callable[[Callable[[T], Awaitable[None]]], Callable[[T], Awaitable[None]]]:
+) -> Callable[[Callable[[T], Awaitable[Any]]], Callable[[T], Awaitable[Any]]]:
@@
-        def decorator(handler: Callable[[T], Awaitable[None]]) -> Callable[[T], Awaitable[None]]:
+        def decorator(handler: Callable[[T], Awaitable[Any]]) -> Callable[[T], Awaitable[Any]]:
             self.logger.info(f"Registering handler '{handler.__name__}' for event type '{event_type}'")
             # Safe: dispatch() routes by event_type, guaranteeing correct types at runtime
             self._handlers[event_type].append(self._wrap_handler(handler))  # type: ignore[arg-type]
             return handler

[sonarqubecloud]

Quality Gate passed

Issues
0 New issues
0 Accepted issues

Measures
0 Security Hotspots
0.0% Coverage on New Code
0.0% Duplication on New Code

See analysis details on SonarQube Cloud