Event Sourcing and CQRS

Implement event sourcing to capture every state change as an immutable event, and CQRS to optimize read and write paths independently. Covers event store design, projections, snapshots, and when these patterns create more problems than they solve.

Event sourcing and CQRS (Command Query Responsibility Segregation) are complementary patterns for building scalable, consistent, and highly available backend systems. They matter when domain logic is complex, state transitions are frequent, and query performance must outpace writes. Use event sourcing when you need to reconstruct system state from a sequence of events, and CQRS when read and write workloads diverge—when queries are more complex, more frequent, or require different data structures than writes. Together, they enable a system where every business action is recorded as an event, and the current state of any entity is derived from those events, with read models optimized independently from write models.

When to Choose Event Sourcing and CQRS

Event sourcing and CQRS are not just tools; they are architectural decisions that shift how data flows and evolves. The decision to adopt them should be driven by two signals:

For example, in a financial trading platform, every trade, settlement, and margin update is a discrete event. The system must answer: *“What was the state of account X at time Y?”* or *“Show all trades and their net positions over the last 90 days.”* These queries are not served efficiently by a single normalized table. Instead, you derive answers from a sequence of events—TradePlaced, SettlementConfirmed, MarginUpdated—each stored as a record in an events table.

In contrast, the write path handles commands: PlaceTrade, AdjustMargin, ReconcileAccount. Each command is validated, processed, and results in one or more events. The write model is a collection of aggregates, each of which is a sequence of events that define its state.

Designing the Write Model: Events and Commands

The write model is driven by commands—requests to perform an action. Each command is processed by a command handler, which validates the command, applies it to an aggregate, and emits one or more events.

A typical command is a JSON payload sent via HTTP POST:

POST /commands
{
  "commandType": "PlaceTrade",
  "commandId": "trade-12345",
  "aggregateId": "account-888",
  "data": {
    "security": "AAPL",
    "quantity": 100,
    "price": 175.32,
    "timestamp": "2024-05-10T12:34:56Z"
  }
}

The command handler reads the current state of the aggregate from the event store (e.g., a events table), applies the command, and appends new events. The event store is a durable, append-only log of all domain events.

Each event is a record with:

Example event:

{
  "eventId": "evt-a6b2c1d3-4e5f",
  "aggregateId": "account-888",
  "eventType": "TradePlaced",
  "version": 1,
  "data": {
    "security": "AAPL",
    "quantity": 100,
    "price": 175.32,
    "tradeId": "trade-12345"
  },
  "timestamp": "2024-05-10T12:34:56Z"
}

The command handler must ensure eventual consistency and idempotency. It can use a commandId to prevent duplicate processing. On failure, the handler re-tries the command, possibly using a retry queue (e.g., RabbitMQ, Kafka, or a database-backed job queue).

Reconstructing State: Event Sourcing

To reconstruct the state of an aggregate, the system loads all events for a given aggregateId, in order, and applies them sequentially.

For example, to get the current state of account-888, a query service runs:

SELECT * FROM events
WHERE aggregateId = 'account-888'
ORDER BY version ASC;

Then, it applies the events in sequence using a state machine:

type Account struct {
    AccountId string
    Balance   float64
    Holdings  map[string]float64
    Version   int
}

func (a *Account) Apply(event Event) {
    switch event.EventType {
    case "TradePlaced":
        data := event.Data.(map[string]interface{})
        security := data["security"].(string)
        quantity := data["quantity"].(float64)
        price := data["price"].(float64)
        cost := quantity * price
        a.Balance -= cost
        a.Holdings[security] += quantity
    case "SettlementConfirmed":
        data := event.Data.(map[string]interface{})
        tradeId := data["tradeId"].(string)
        a.Holdings[tradeId] += 1
    case "MarginUpdated":
        data := event.Data.(map[string]interface{})
        amount := data["amount"].(float64)
        a.Balance += amount
    }
}

Each application of an event increases the Version of the aggregate. The final state is derived by applying all events from version 0 to N.

This process is called reconstitution. It happens every time the aggregate is read or when a new read model is built.

Building the Read Model: CQRS

CQRS separates the write model (command-driven, event-sourced) from the read model (query-optimized). The read model is derived from the event stream and is updated by event handlers.

Event handlers listen to the event store and update denormalized read models—tables or documents tailored to specific queries.

Example: an account_summary table optimized for reporting:

CREATE TABLE account_summary (
    account_id VARCHAR(36) PRIMARY KEY,
    balance DECIMAL(18, 2),
    total_trades INT,
    last_traded_at TIMESTAMP,
    latest_holdings JSONB
);

An event handler for TradePlaced:

def handle_trade_placed(event):
    conn = psycopg2.connect("host=db user=app password=secret dbname=reports")
    cursor = conn.cursor()

    # Update account summary
    cursor.execute("""
        INSERT INTO account_summary (account_id, balance, total_trades, last_traded_at)
        VALUES (%s, %s, %s, %s)
        ON CONFLICT (account_id)
        DO UPDATE SET
            balance = account_summary.balance + EXCLUDED.balance,
            total_trades = account_summary.total_trades + EXCLUDED.total_trades,
            last_traded_at = EXCLUDED.last_traded_at
        WHERE account_summary.account_id = EXCLUDED.account_id
    """, [
        event["aggregateId"],
        event["data"]["price"] * event["data"]["quantity"],
        1,
        event["timestamp"]
    ])

    # Update holdings
    cursor.execute("""
        UPDATE account_summary
        SET latest_holdings = jsonb_set(
            latest_holdings,
            '{' || event['data']['security'] || '}',
            to_jsonb(event['data']['quantity']),
            true
        )
        WHERE account_id = %s
    """, [event["aggregateId"]])

    conn.commit()

Each event triggers one or more read model updates. This process is event-driven and asynchronous. The write model may be updated in real time, while the read model is updated with a delay—anywhere from milliseconds to minutes.

Handling Event Delivery and Consistency

Events are delivered through a message broker—Kafka, RabbitMQ, or a database table. The event stream is a log of all events, indexed by eventId and aggregateId.

A common failure is event loss. If the event store is down during a command, the event may be lost. To prevent this, use idempotent event handlers and at-least-once delivery.

Event handlers must be idempotent: processing the same event multiple times must produce the same result.

A typical idempotency check uses an event_id or event_id + aggregate_id as a primary key in the target table.

For example, in a account_summary table, the eventId is a unique constraint:

ALTER TABLE account_summary ADD CONSTRAINT unique_event_id
    UNIQUE (event_id);

When an event handler receives an event, it checks if event_id already exists. If yes, it skips processing.

Another common failure is event ordering. In a distributed system, events may be processed out of order. The solution is to use versioning and sequence numbers.

Each event has a version field, which is the total number of events for that aggregate. When applying an event, the handler checks that the current version matches the expected version. If not, it throws a ConcurrentUpdateException.

Example error:

{
  "error": "ConcurrentUpdateException",
  "message": "Expected version 3, but current version is 4",
  "expectedVersion": 3,
  "currentVersion": 4,
  "eventId": "evt-a6b2c1d3-4e5f",
  "aggregateId": "account-888"
}

This error indicates that another event was processed between the time the handler read the state and wrote the new state.

Deploying and Monitoring the System

Deployment involves running:

The system is monitored by:

Example system.log entry:

2024-05-10T12:35:01.123Z [INFO]  event_processor: processing event evt-a6b2c1d3-4e5f of type TradePlaced for account-888
2024-05-10T12:35:02.456Z [ERROR] event_processor: failed to apply event evt-a6b2c1d3-4e5f: ConcurrentUpdateException, expected version 3, current version 4
2024-05-10T12:35:03.789Z [INFO]  event_processor: successfully processed 100 events in 1.2 seconds

Common Pitfalls and Sharp Edges

Summary of Key Patterns

Event sourcing and CQRS are not one-size-fits-all. They are best applied when the system’s behavior depends on a sequence of actions, and when reads are more complex or frequent than writes. They are the foundation of reactive, event-driven systems.

This page was rewritten on 18 October 2026. It replaced a templated version that was not specific to its own title. How these pages are written and checked. If anything here is wrong, tell us at [email protected] and we will correct it.