Event Sourcing & CQRS

DocuStore uses event sourcing as its primary persistence pattern. Every state change to a document or page is recorded as an immutable domain event in EventStoreDB. The current state of any entity can be reconstructed by replaying its event stream from the beginning.

Domain Aggregates

There are two core aggregates in the domain model:

Artifact

An Artifact represents a document (typically a PDF). Its event stream records the full lifecycle:

Page

A Page represents a single page within an artifact. Pages carry the extracted content:

CQRS: Separate Read and Write Models

DocuStore follows the CQRS (Command Query Responsibility Segregation) pattern. Writes go through the event-sourced aggregates; reads are served from MongoDB projections.

Write Path

API Request → Use Case → Load Aggregate (EventStoreDB replay)
  → Domain Method (validation, business rules)
  → Append Event (EventStoreDB)

The aggregate is loaded by replaying all events from its stream. The domain method applies business rules and produces a new event. The event is appended to EventStoreDB with optimistic concurrency control -- if another process modified the aggregate concurrently, an IntegrityError is raised and the operation is retried.

Read Path

EventStoreDB → Subscription → Event Projector → MongoDB Read Model
                                                      ↑
API Query ─────────────────────────────────────────────┘

Two independent worker processes subscribe to the EventStoreDB $all stream:

• Read worker (read_worker.py) -- Projects events into MongoDB read models. Each event type has a dedicated projector that updates the denormalized document in the appropriate collection.
• Pipeline worker (pipeline_worker.py) -- Detects events that should trigger processing workflows (e.g., Page.Created triggers compound extraction, Page.TextMentionUpdated triggers embedding generation).

Both workers maintain their own checkpoint positions in MongoDB, so they can be restarted independently without losing progress.

Concurrency Handling

EventStoreDB provides optimistic concurrency control via stream revisions. When saving an aggregate, the repository specifies the expected revision number. If the stream has been modified since the aggregate was loaded, the append fails with an IntegrityError.

This is mapped to a ConcurrencyError in the application layer. Temporal activities that encounter concurrency errors raise an exception, causing Temporal to retry the activity automatically with exponential backoff.

# infrastructure/event_sourced_repositories/page_repository.py
try:
    await self._client.append_to_stream(stream_name, events, current_version=expected)
except eventsourcing.IntegrityError:
    raise ConcurrencyError(f"Page {page_id} was modified concurrently")

Configuration

Key Design Decisions

No workflow status on aggregates. Workflow status (running, completed, failed) is operational metadata, not domain state. Temporal is the source of truth for workflow status. Dedicated endpoints (GET /pages/{id}/workflows, GET /artifacts/{id}/workflows) proxy to Temporal directly.

Read models are disposable. Since MongoDB projections are derived entirely from the EventStoreDB event stream, they can be rebuilt from scratch at any time by replaying the stream. This makes schema migrations straightforward -- drop the collection, update the projector, and replay.