Event Sourcing and CQRS Patterns

Introduction

Event Sourcing and CQRS represent two architectural patterns that, when combined, provide a powerful foundation for building complex, scalable, and auditable systems. Event Sourcing stores every state change as an immutable event in an append-only log, while CQRS (Command Query Responsibility Segregation) separates the write model from the read model, allowing each to be optimized independently. Together, these patterns enable temporal queries, complete audit trails, and the ability to rebuild system state from scratch—capabilities that are invaluable in domains like finance, healthcare, and logistics.

The combination of Event Sourcing and CQRS is not a silver bullet. It introduces complexity in the form of eventual consistency, event schema management, and projection rebuilding. However, for systems where auditability, temporal queries, and the ability to evolve read models independently are critical, the benefits far outweigh the costs. In this guide, we will explore both patterns in depth, implement them in TypeScript, and examine real-world use cases where they shine.

Understanding Event Sourcing: Core Concepts

Traditional CRUD systems store the current state of an entity, overwriting previous values with each update. This approach loses historical information—if a customer's address changes, the old address is gone. Event Sourcing takes a fundamentally different approach: instead of storing the current state, you store the sequence of events that led to the current state.

An event is an immutable fact that describes something that happened. Events are written in past tense: OrderCreated, ItemAdded, AddressUpdated, PaymentProcessed. Each event carries the data relevant to that specific change. The current state of an entity is derived by replaying all its events from the beginning—a process called projection or materialization.

The event store is the persistence layer for events. It is an append-only log: events are added but never modified or deleted. This immutability provides a complete, tamper-proof history of every change in the system. The event store also serves as the single source of truth—all other views of the data (read models, caches, search indexes) are derived from the event stream.

Event versioning is a critical concern. As business requirements evolve, event schemas change. You need a strategy for handling old events with new code. Event upcasting transforms old event versions into the current version during replay. This allows you to evolve your domain model without migrating historical events.

Snapshots are an optimization for aggregates with many events. Instead of replaying thousands of events every time you load an aggregate, you periodically save a snapshot of the aggregate's state. When loading, you replay only the events after the last snapshot, significantly reducing load time.

Architecture and Design Patterns

Event Sourcing Pattern

The core event sourcing pattern involves three components: aggregates that enforce business rules and emit events, an event store that persists events, and projections that build read models from the event stream. Aggregates load their state by replaying events, validate commands against current state, and emit new events when commands succeed.

CQRS Pattern

CQRS separates the system into two sides: the write side (commands) and the read side (queries). The write side processes commands, validates business rules, and produces events. The read side consumes events and maintains denormalized query-optimized views. Each side can be scaled, deployed, and optimized independently.

Projection Patterns

Projections transform event streams into query-optimized read models. There are several projection strategies: inline projections update read models synchronously within the same transaction, async projections update read models from event subscriptions, and catch-up projections rebuild from historical events.

Event Schema Evolution

As business requirements change, event schemas must evolve without breaking existing consumers. Strategies include adding optional fields with defaults, using upcasters to transform old events, and versioning event types. The key principle is backward compatibility: new code must be able to read old events.

Saga Pattern with Event Sourcing

Sagas coordinate multi-step business processes across multiple aggregates. Each step emits an event that triggers the next step. If a step fails, compensating events trigger rollback logic. Event sourcing provides a natural foundation for sagas because every step is recorded as an event.

Step-by-Step Implementation

Let us implement a complete event sourcing and CQRS system in TypeScript. We will build an order processing system with full audit trail capabilities.

First, define the core event store:

interface StoredEvent {
  eventId: string;
  aggregateId: string;
  aggregateType: string;
  type: string;
  data: Record<string, unknown>;
  metadata: {
    correlationId: string;
    causationId?: string;
    userId?: string;
    timestamp: Date;
    version: number;
  };
}
 
class EventStore {
  private events: StoredEvent[] = [];
  private subscribers: Map<string, ((event: StoredEvent) => Promise<void>)[]> = new Map();
 
  async append(event: StoredEvent): Promise<void> {
    // Optimistic concurrency check
    const existing = this.events.filter(e => e.aggregateId === event.aggregateId);
    const currentVersion = existing.length > 0
      ? existing[existing.length - 1].metadata.version
      : 0;
 
    if (event.metadata.version !== currentVersion + 1) {
      throw new Error(
        `Concurrency conflict: expected version ${currentVersion + 1}, got ${event.metadata.version}`
      );
    }
 
    this.events.push(event);
    await this.notifySubscribers(event);
  }
 
  async getEvents(aggregateId: string): Promise<StoredEvent[]> {
    return this.events.filter(e => e.aggregateId === aggregateId);
  }
 
  async getEventsByType(eventType: string): Promise<StoredEvent[]> {
    return this.events.filter(e => e.type === eventType);
  }
 
  subscribe(eventType: string, handler: (event: StoredEvent) => Promise<void>): void {
    const handlers = this.subscribers.get(eventType) || [];
    handlers.push(handler);
    this.subscribers.set(eventType, handlers);
  }
 
  private async notifySubscribers(event: StoredEvent): Promise<void> {
    const handlers = this.subscribers.get(event.type) || [];
    const wildcardHandlers = this.subscribers.get('*') || [];
    await Promise.allSettled([...handlers, ...wildcardHandlers].map(h => h(event)));
  }
}

Implement the aggregate root with snapshot support:

abstract class AggregateRoot {
  protected id: string = '';
  private version: number = 0;
  private uncommittedEvents: StoredEvent[] = [];
 
  protected applyEvent(type: string, data: Record<string, unknown>, metadata?: Partial<StoredEvent['metadata']>): void {
    const event: StoredEvent = {
      eventId: crypto.randomUUID(),
      aggregateId: this.id,
      aggregateType: this.constructor.name,
      type,
      data,
      metadata: {
        correlationId: metadata?.correlationId || crypto.randomUUID(),
        timestamp: new Date(),
        version: this.version + 1,
        ...metadata,
      },
    };
    this.apply(event);
    this.uncommittedEvents.push(event);
  }
 
  loadFromHistory(events: StoredEvent[]): void {
    events.forEach(event => this.apply(event));
  }
 
  private apply(event: StoredEvent): void {
    this.when(event);
    this.version = event.metadata.version;
  }
 
  protected abstract when(event: StoredEvent): void;
 
  getUncommittedEvents(): StoredEvent[] {
    return [...this.uncommittedEvents];
  }
 
  clearUncommittedEvents(): void {
    this.uncommittedEvents = [];
  }
 
  getVersion(): number {
    return this.version;
  }
}
 
interface Snapshot {
  aggregateId: string;
  aggregateType: string;
  version: number;
  state: Record<string, unknown>;
  timestamp: Date;
}
 
class SnapshotStore {
  private snapshots: Map<string, Snapshot> = new Map();
 
  async save(snapshot: Snapshot): Promise<void> {
    this.snapshots.set(snapshot.aggregateId, snapshot);
  }
 
  async get(aggregateId: string): Promise<Snapshot | undefined> {
    return this.snapshots.get(aggregateId);
  }
}

Create the Order aggregate with business logic:

type OrderStatus = 'draft' | 'confirmed' | 'shipped' | 'delivered' | 'cancelled';
 
class Order extends AggregateRoot {
  private status: OrderStatus = 'draft';
  private items: Array<{ productId: string; quantity: number; price: number; name: string }> = [];
  private shippingAddress: { street: string; city: string; zip: string } | null = null;
  private totalAmount: number = 0;
  private confirmedAt: Date | null = null;
  private cancelledAt: Date | null = null;
  private cancellationReason: string | null = null;
 
  static create(orderId: string): Order {
    const order = new Order();
    order.id = orderId;
    order.applyEvent('OrderCreated', { orderId, createdAt: new Date().toISOString() });
    return order;
  }
 
  addItem(item: { productId: string; quantity: number; price: number; name: string }): void {
    if (this.status !== 'draft') throw new Error('Cannot modify confirmed order');
    this.applyEvent('ItemAdded', { ...item });
  }
 
  removeItem(productId: string): void {
    if (this.status !== 'draft') throw new Error('Cannot modify confirmed order');
    if (!this.items.find(i => i.productId === productId)) {
      throw new Error(`Item ${productId} not in order`);
    }
    this.applyEvent('ItemRemoved', { productId });
  }
 
  setShippingAddress(address: { street: string; city: string; zip: string }): void {
    if (this.status !== 'draft') throw new Error('Cannot modify confirmed order');
    this.applyEvent('ShippingAddressSet', { ...address });
  }
 
  confirm(): void {
    if (this.status !== 'draft') throw new Error('Order already processed');
    if (this.items.length === 0) throw new Error('Cannot confirm empty order');
    if (!this.shippingAddress) throw new Error('Shipping address required');
    this.applyEvent('OrderConfirmed', {
      confirmedAt: new Date().toISOString(),
      totalAmount: this.totalAmount,
    });
  }
 
  cancel(reason: string): void {
    if (this.status === 'shipped' || this.status === 'delivered') {
      throw new Error(`Cannot cancel order in ${this.status} status`);
    }
    this.applyEvent('OrderCancelled', {
      reason,
      cancelledAt: new Date().toISOString(),
    });
  }
 
  protected when(event: StoredEvent): void {
    switch (event.type) {
      case 'OrderCreated':
        this.status = 'draft';
        break;
      case 'ItemAdded': {
        const item = event.data as { productId: string; quantity: number; price: number; name: string };
        this.items.push(item);
        this.totalAmount = this.items.reduce((sum, i) => sum + i.quantity * i.price, 0);
        break;
      }
      case 'ItemRemoved': {
        this.items = this.items.filter(i => i.productId !== event.data.productId);
        this.totalAmount = this.items.reduce((sum, i) => sum + i.quantity * i.price, 0);
        break;
      }
      case 'ShippingAddressSet':
        this.shippingAddress = event.data as typeof this.shippingAddress;
        break;
      case 'OrderConfirmed':
        this.status = 'confirmed';
        this.confirmedAt = new Date(event.data.confirmedAt as string);
        this.totalAmount = event.data.totalAmount as number;
        break;
      case 'OrderCancelled':
        this.status = 'cancelled';
        this.cancelledAt = new Date(event.data.cancelledAt as string);
        this.cancellationReason = event.data.reason as string;
        break;
    }
  }
}

Implement CQRS projections for read models:

interface OrderReadModel {
  orderId: string;
  status: OrderStatus;
  items: Array<{ productId: string; name: string; quantity: number; price: number }>;
  totalAmount: number;
  shippingAddress: { street: string; city: string; zip: string } | null;
  createdAt: string;
  confirmedAt: string | null;
  cancelledAt: string | null;
  cancellationReason: string | null;
}
 
class OrderProjection {
  private readModels: Map<string, OrderReadModel> = new Map();
 
  async handle(event: StoredEvent): Promise<void> {
    switch (event.type) {
      case 'OrderCreated':
        this.readModels.set(event.aggregateId, {
          orderId: event.aggregateId,
          status: 'draft',
          items: [],
          totalAmount: 0,
          shippingAddress: null,
          createdAt: event.metadata.timestamp.toISOString(),
          confirmedAt: null,
          cancelledAt: null,
          cancellationReason: null,
        });
        break;
 
      case 'ItemAdded': {
        const model = this.readModels.get(event.aggregateId);
        if (model) {
          model.items.push(event.data as OrderReadModel['items'][0]);
          model.totalAmount = model.items.reduce((sum, i) => sum + i.quantity * i.price, 0);
        }
        break;
      }
 
      case 'OrderConfirmed': {
        const model = this.readModels.get(event.aggregateId);
        if (model) {
          model.status = 'confirmed';
          model.confirmedAt = event.data.confirmedAt as string;
        }
        break;
      }
 
      case 'OrderCancelled': {
        const model = this.readModels.get(event.aggregateId);
        if (model) {
          model.status = 'cancelled';
          model.cancelledAt = event.data.cancelledAt as string;
          model.cancellationReason = event.data.reason as string;
        }
        break;
      }
    }
  }
 
  getOrder(orderId: string): OrderReadModel | undefined {
    return this.readModels.get(orderId);
  }
 
  getOrdersByStatus(status: OrderStatus): OrderReadModel[] {
    return Array.from(this.readModels.values()).filter(o => o.status === status);
  }
}

Wire everything together with snapshot support:

class OrderRepository {
  constructor(
    private eventStore: EventStore,
    private snapshotStore: SnapshotStore,
    private snapshotInterval: number = 10
  ) {}
 
  async load(aggregateId: string): Promise<Order> {
    const snapshot = await this.snapshotStore.get(aggregateId);
    const order = new Order();
 
    if (snapshot) {
      // Load from snapshot
      Object.assign(order, snapshot.state);
      const events = await this.eventStore.getEvents(aggregateId);
      const eventsAfterSnapshot = events.filter(e => e.metadata.version > snapshot.version);
      order.loadFromHistory(eventsAfterSnapshot);
    } else {
      // Load from beginning
      const events = await this.eventStore.getEvents(aggregateId);
      order.loadFromHistory(events);
    }
 
    return order;
  }
 
  async save(order: Order): Promise<void> {
    const events = order.getUncommittedEvents();
    for (const event of events) {
      await this.eventStore.append(event);
    }
    order.clearUncommittedEvents();
 
    // Create snapshot if needed
    if (order.getVersion() % this.snapshotInterval === 0) {
      await this.snapshotStore.save({
        aggregateId: order['id'],
        aggregateType: 'Order',
        version: order.getVersion(),
        state: { /* serialize aggregate state */ },
        timestamp: new Date(),
      });
    }
  }
}

Real-World Use Cases and Case Studies

Use Case 1: Banking and Financial Systems

Banks use event sourcing to maintain a complete, auditable record of every transaction. When a customer disputes a charge, the bank can replay the exact sequence of events to verify what happened. Regulatory requirements mandate this level of auditability. The CQRS pattern allows the bank to maintain separate read models optimized for different queries: account balance, transaction history, monthly statements, and fraud detection analytics.

Use Case 2: Healthcare Patient Records

Electronic health records (EHR) systems use event sourcing to track every change to a patient's medical history. Each update—lab results, prescriptions, diagnoses, and treatments—is stored as an immutable event. This provides a complete medical history that is critical for patient care and legal compliance. CQRS enables fast lookups for patient summaries while the event store maintains the full history.

Use Case 3: Supply Chain Management

Supply chain systems use event sourcing to track the movement of goods from manufacturer to consumer. Each event—ItemManufactured, ItemShipped, ItemReceived, ItemDelivered—is stored immutably. This enables end-to-end visibility, provenance tracking, and dispute resolution. The CQRS pattern maintains optimized views for different stakeholders: warehouse managers see inventory levels, logistics teams see shipment status, and finance teams see cost breakdowns.

Use Case 4: E-Commerce Platform

An e-commerce platform uses event sourcing for order processing. The order lifecycle events (OrderCreated, ItemAdded, PaymentProcessed, OrderShipped) are stored in the event store. Multiple projections build different read models: order summaries for the customer dashboard, fulfillment queues for the warehouse, revenue reports for finance, and recommendation engines based on purchase history.

Best Practices for Production

Design events as immutable facts: Events should describe what happened, not what should happen. Use past tense naming (OrderCreated, not CreateOrder). Include all data needed to rebuild the aggregate state without querying external systems.
Implement optimistic concurrency: Use version numbers to detect concurrent modifications. When appending events, check that the expected version matches the current version. If not, reject the command and let the client retry with fresh state.
Create snapshots for long-lived aggregates: After every N events (e.g., 100), save a snapshot of the aggregate's current state. When loading, start from the snapshot and replay only recent events. This dramatically reduces load time for aggregates with long histories.
Use async projections for read models: Update read models asynchronously by subscribing to events. This keeps the write path fast and allows projections to be rebuilt independently. Handle projection failures gracefully with retry logic and dead letter queues.
Version your event schemas: Use a schema registry and maintain backward compatibility. When event schemas change, implement upcasters that transform old events to the current format during replay. Never modify historical events in the event store.
Implement catch-up projections: When adding a new read model, replay all historical events to build the initial state. This is one of the most powerful capabilities of event sourcing—you can create new views of existing data without modifying the write side.
Use correlation IDs for distributed tracing: Every event should carry a correlation ID that links it to the original command. This enables end-to-end tracing of complex workflows across multiple aggregates and services.
Plan for event store scaling: The append-only nature of event stores makes them easy to scale horizontally. Partition events by aggregate ID, and use separate event stores for different aggregate types if needed.

Common Pitfalls and Solutions

Pitfall	Impact	Solution
Treating events as commands	Tight coupling between producer and consumer	Events should be facts about the past, not instructions
No snapshot strategy	Aggregate loading becomes slow over time	Implement snapshotting at regular intervals
Synchronous projections	Write path becomes slow, coupling increases	Use async projections with event subscriptions
Ignoring eventual consistency	Users see stale data, confusion about data freshness	Design UI for eventual consistency; use read-your-writes patterns
Event schema changes without versioning	Breaking changes in production	Use schema registry with backward compatibility
Unbounded event replay for new projections	Rebuilding projections takes too long	Use catch-up projections with progress tracking

Performance Optimization

Event sourcing performance depends on event store access patterns, projection update strategies, and snapshot management. Optimizing these areas is critical for production systems.

// Optimized event store with caching and batching
class CachedEventStore implements EventStore {
  private cache: Map<string, StoredEvent[]> = new Map();
  private batchBuffer: Map<string, StoredEvent[]> = new Map();
 
  async getEvents(aggregateId: string): Promise<StoredEvent[]> {
    if (this.cache.has(aggregateId)) {
      return this.cache.get(aggregateId)!;
    }
    const events = await this.fetchFromDatabase(aggregateId);
    this.cache.set(aggregateId, events);
    return events;
  }
 
  async append(event: StoredEvent): Promise<void> {
    // Add to batch buffer
    const buffer = this.batchBuffer.get(event.aggregateId) || [];
    buffer.push(event);
    this.batchBuffer.set(event.aggregateId, buffer);
 
    // Flush batch if it reaches threshold
    if (buffer.length >= 100) {
      await this.flushBatch(event.aggregateId);
    }
  }
 
  private async flushBatch(aggregateId: string): Promise<void> {
    const events = this.batchBuffer.get(aggregateId) || [];
    if (events.length === 0) return;
 
    await this.batchInsert(events);
 
    // Update cache
    const cached = this.cache.get(aggregateId) || [];
    this.cache.set(aggregateId, [...cached, ...events]);
 
    this.batchBuffer.set(aggregateId, []);
  }
}

Comparison with Alternatives

Feature	Event Sourcing + CQRS	CRUD + Relational DB	Document Store
Audit Trail	Built-in	Manual	Manual
Temporal Queries	Native	Complex	Not supported
Read Optimization	Separate read models	Indexes	Document structure
Write Performance	High (append-only)	Medium	High
Complexity	High	Low	Low
Consistency	Eventual	Strong	Eventual
Schema Evolution	Event versioning	Migrations	Flexible
Debugging	Event replay	Logs	Limited

Advanced Patterns

Event Upcasting

When event schemas evolve, old events stored in the event store may have outdated structures. Event upcasting transforms old event versions into the current version during replay.

class EventUpcaster {
  private upcasters: Map<string, (data: Record<string, unknown>) => Record<string, unknown>> = new Map();
 
  register(eventType: string, fromVersion: number, transform: (data: Record<string, unknown>) => Record<string, unknown>): void {
    this.upcasters.set(`${eventType}:v${fromVersion}`, transform);
  }
 
  upcast(event: StoredEvent): StoredEvent {
    const key = `${event.type}:v${event.metadata.version}`;
    const upcaster = this.upcasters.get(key);
    if (upcaster) {
      return { ...event, data: upcaster(event.data) };
    }
    return event;
  }
}
 
// Example: Evolving OrderCreated event
const upcaster = new EventUpcaster();
upcaster.register('OrderCreated', 1, (data) => ({
  ...data,
  currency: 'USD', // Add default currency field
  createdAt: data.createdAt || new Date().toISOString(),
}));

Testing Strategies

Testing event-sourced systems requires verifying both the event emission and the projection logic.

describe('Order Aggregate', () => {
  it('should emit ItemAdded when adding item to draft order', () => {
    const order = Order.create('order-1');
    order.clearUncommittedEvents();
 
    order.addItem({ productId: 'prod-1', quantity: 2, price: 25, name: 'Widget' });
 
    const events = order.getUncommittedEvents();
    expect(events).toHaveLength(1);
    expect(events[0].type).toBe('ItemAdded');
    expect(events[0].data.productId).toBe('prod-1');
  });
 
  it('should rebuild state from events correctly', () => {
    const events: StoredEvent[] = [
      { eventId: '1', aggregateId: 'order-1', aggregateType: 'Order', type: 'OrderCreated', data: {}, metadata: { correlationId: 'c1', timestamp: new Date(), version: 1 } },
      { eventId: '2', aggregateId: 'order-1', aggregateType: 'Order', type: 'ItemAdded', data: { productId: 'p1', quantity: 2, price: 25, name: 'Widget' }, metadata: { correlationId: 'c1', timestamp: new Date(), version: 2 } },
      { eventId: '3', aggregateId: 'order-1', aggregateType: 'Order', type: 'ItemAdded', data: { productId: 'p2', quantity: 1, price: 50, name: 'Gadget' }, metadata: { correlationId: 'c1', timestamp: new Date(), version: 3 } },
    ];
 
    const order = new Order();
    order.loadFromHistory(events);
 
    expect(order.getVersion()).toBe(3);
  });
});

Future Outlook

Event sourcing and CQRS continue to evolve with the adoption of cloud-native event stores, serverless event processing, and standardized event formats like CloudEvents. The combination of event sourcing with machine learning enables training models on historical event data for predictive analytics and anomaly detection.

The emergence of event mesh topologies—where events flow seamlessly across cloud providers, regions, and edge locations—is expanding the reach of event-driven architectures. As more organizations adopt microservices and cloud-native patterns, event sourcing and CQRS will become increasingly mainstream, with better tooling and frameworks reducing the implementation complexity.

Conclusion

Event Sourcing and CQRS provide a powerful foundation for building auditable, scalable, and evolvable systems. The patterns address real challenges in complex domains: complete audit trails, temporal queries, independent read/write scaling, and the ability to rebuild state from scratch.

Key takeaways: (1) Event sourcing stores immutable facts about state changes, not current state; (2) CQRS separates read and write concerns for independent optimization; (3) Snapshots are essential for aggregates with long event histories; (4) Async projections keep the write path fast; (5) Event schema evolution requires careful versioning and upcasting.

The complexity of these patterns is justified when auditability, temporal queries, and system evolvability are critical requirements. Start with a bounded context where these benefits are most valuable, and expand as the patterns prove their worth in your domain.

Minh Vo

Slaying code & making it lit fr fr 🔥 tagline