Cqrs Implementation Patterns

Production engineering guide for cqrs implementation patterns covering patterns, implementation strategies, and operational best practices.

CQRS (Command Query Responsibility Segregation) is a pattern for structuring services so that writes (commands) and reads (queries) are handled by separate, optimised models. It matters when a system has divergent performance, consistency, or scalability requirements between data modification and data retrieval—particularly in systems with high write volume, complex business logic, or varied read patterns. This reference page details the practical, real-world implementation of CQRS across cloud-native services.

Command and Query Models in Practice

In a CQRS system, every domain command (e.g., CreateOrder, ApprovePayment) triggers a change to the write model—a denormalized, event-sourced representation of state. Queries (e.g., GetOrderDetails, ListRecentOrders) are served from a read model—a separate, optimised view of data, often derived from events emitted by the write model.

The core flow begins with a command arriving via HTTP, message queue, or gRPC. The command is processed by a command handler that validates input, applies business logic, and emits one or more domain events. These events are published to an event store or message broker. A separate process—often a set of consumers—listens to the event stream and updates the read model by projecting events into denormalized tables or caches.

Command Handling with Event Sourcing

Commands are processed asynchronously. A typical command handler in a service written in Go might look like:

type CreateOrderCommand struct {
    OrderID    string    `json:"orderId"`
    CustomerID string    `json:"customerId"`
    Items      []Item    `json:"items"`
    CreatedAt  time.Time `json:"createdAt"`
}

func (s *OrderService) HandleCreateOrder(cmd CreateOrderCommand) error {
    // 1. Load aggregate from event store
    events, err := s.eventStore.LoadAggregate(cmd.OrderID)
    if err != nil {
        return fmt.Errorf("failed to load order: %w", err)
    }

    // 2. Apply command to aggregate
    order := NewOrder(cmd.OrderID, cmd.CustomerID)
    order.ApplyEvents(events)
    order.CreateOrder(cmd.Items, cmd.CreatedAt)

    // 3. Append new events to event store
    if err := s.eventStore.AppendEvents(order.ID, order.UncommittedEvents()); err != nil {
        return fmt.Errorf("failed to persist events: %w", err)
    }

    // 4. Publish domain events
    for _, e := range order.UncommittedEvents() {
        s.eventPublisher.Publish(e)
    }

    return nil
}

The event store must support idempotent appends and efficient event retrieval by aggregate ID. The command handler is often a REST endpoint or a gRPC service. A POST /orders request might return 202 Accepted with a Location header pointing to a command result:

POST /orders HTTP/1.1
Host: orders-api.example.com
Content-Type: application/json

{
  "orderId": "ORD-2024-001",
  "customerId": "CUST-003",
  "items": [
    { "sku": "TSHIRT", "quantity": 2 },
    { "sku": "MOUSE", "quantity": 1 }
  ],
  "createdAt": "2024-04-05T12:00:00Z"
}

Response:

HTTP/1.1 202 Accepted
Location: /commands/ORD-2024-001
Content-Type: application/json

{
  "commandId": "ORD-2024-001",
  "status": "submitted",
  "submittedAt": "2024-04-05T12:00:00Z"
}

Event Publishing and Consumption

Events are serialised as JSON and published to a message broker. A typical event might be:

{
  "eventId": "EVT-2024-001",
  "aggregateId": "ORD-2024-001",
  "eventType": "OrderCreated",
  "version": 1,
  "data": {
    "orderId": "ORD-2024-001",
    "customerId": "CUST-003",
    "items": [
      { "sku": "TSHIRT", "quantity": 2 },
      { "sku": "MOUSE", "quantity": 1 }
    ],
    "createdAt": "2024-04-05T12:00:00Z"
  },
  "metadata": {
    "timestamp": "2024-04-05T12:00:01Z",
    "source": "orders-service"
  }
}

The event store is typically a dedicated database (e.g., PostgreSQL with a events table) or a purpose-built system (e.g., Apache Kafka, AWS Kinesis). Consumers listen to the event stream and project events into read models.

Read Model Projection

Read models are built by consumers that subscribe to event streams. A consumer might process events in batches, using a cursor-based approach to track progress.

Example of a projection loop in Node.js:

const { Kafka } = require('kafkajs');

const kafka = new Kafka({ brokers: ['kafka1:9092', 'kafka2:9092'] });
const consumer = kafka.consumer({ groupId: 'order-read-models' });

async function startProjection() {
    await consumer.connect();
    await consumer.subscribe({ topic: 'domain-events', fromBeginning: true });

    await consumer.run({
        eachMessage: async ({ topic, partition, message }) => {
            const event = JSON.parse(message.value.toString());

            switch (event.eventType) {
                case 'OrderCreated':
                    await db.query(
                        `INSERT INTO order_summary (id, customer_id, created_at, status)
                         VALUES ($1, $2, $3, 'created')`,
                        [event.data.orderId, event.data.customerId, event.data.createdAt]
                    );
                    break;

                case 'OrderShipped':
                    await db.query(
                        `UPDATE order_summary SET status = 'shipped', shipped_at = $1
                         WHERE id = $2`,
                        [event.data.shippedAt, event.data.orderId]
                    );
                    break;

                // ... other cases
            }

            // Acknowledge message
            await consumer.commitOffsets([
                { topic, partition, offset: message.offset }
            ]);
        }
    });
}

The projection process must handle:

Querying the Read Model

Queries are served from the read model, typically via a REST API or gRPC endpoint. The response is often pre-aggregated and denormalized. Example query:

GET /api/orders/ORD-2024-001 HTTP/1.1
Host: orders-read.example.com
Accept: application/json

Response:

{
  "orderId": "ORD-2024-001",
  "customerId": "CUST-003",
  "customerName": "Alice Johnson",
  "items": [
    {
      "sku": "TSHIRT",
      "name": "Cotton T-Shirt",
      "quantity": 2,
      "price": 19.99
    },
    {
      "sku": "MOUSE",
      "name": "Wireless Mouse Pro",
      "quantity": 1,
      "price": 49.99
    }
  ],
  "totalAmount": 89.97,
  "status": "shipped",
  "createdAt": "2024-04-05T12:00:00Z",
  "shippedAt": "2024-04-06T09:15:00Z",
  "estimatedDelivery": "2024-04-10T18:00:00Z"
}

The read model may be updated in real time via a materialized view, or in batch via a periodic sync job. A common pattern is to use a read model database (e.g., a PostgreSQL instance) and caching layer (e.g., Redis) to serve high-throughput queries.

Eventual Consistency and Versioning

CQRS systems are inherently eventually consistent. A command may take seconds or minutes to appear in the read model. Clients must account for this.

Key failure modes:

To mitigate, use event versioning: every event includes a version number (e.g., version: 1, version: 2), and consumers can validate that the event they’re processing is the next in sequence. The event store should enforce sequential ordering.

Handling Command Idempotency

Commands must be idempotent. A CreateOrder command sent twice should not create two orders. This is achieved by:

CREATE TABLE command_log (
    command_id VARCHAR(64) PRIMARY KEY,
    aggregate_id VARCHAR(64),
    status VARCHAR(20),
    processed_at TIMESTAMP,
    result JSONB
);

The command handler checks if command_id exists before processing. If so, it returns the stored result and skips processing.

Handling Concurrency and Conflict Resolution

When multiple commands arrive for the same aggregate (e.g., two UpdateOrderItem commands for ORD-2024-001), the system must resolve conflicts. The most common approach is optimistic concurrency control using a version field in the aggregate.

Each event includes the aggregate’s version. The command handler verifies that the current version matches the expected version before applying events:

{
  "eventId": "EVT-2024-002",
  "aggregateId": "ORD-2024-001",
  "eventType": "OrderItemUpdated",
  "version": 3,
  "data": {
    "orderId": "ORD-2024-001",
    "itemId": "ITEM-001",
    "quantity": 5
  }
}

The command handler reads the current version from the event store. If the expected version is not current, it returns 409 Conflict:

HTTP/1.1 409 Conflict
Content-Type: application/json

{
  "error": "CommandConflict",
  "message": "Expected version 3, but current version is 4",
  "expectedVersion": 3,
  "currentVersion": 4,
  "commandId": "UPD-2024-001"
}

The client may retry the command with the updated version or merge changes.

Failure Modes and Observability

Common Mistakes and Sharp Edges

Summary

CQRS is not a single pattern but a collection of practices: separate models for commands and queries, event-driven architecture, idempotent commands, versioned events, and eventual consistency. It excels in systems where writes are complex and reads are varied. The implementation relies on a message broker, event store, and projection system—all working together to deliver scalable, maintainable, and performant data access. The success of a CQRS system hinges on handling event delivery, ensuring idempotency, and managing eventual consistency.

This page was rewritten on 17 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.