Metadata

SEO Title: Event Sourcing Patterns: Aggregate Design and Snapshots

Meta Description: Learn production-ready event sourcing patterns for aggregate design and snapshots. Solve performance issues with practical TypeScript examples.

Primary Keyword: event sourcing aggregate design

Secondary Keywords: event sourcing snapshots, aggregate root pattern, event store optimization, CQRS aggregate design, event sourcing performance, domain-driven design aggregates, event replay optimization

Tags: EventSourcing, DomainDrivenDesign, CQRS, Microservices, SystemDesign, TypeScript, Architecture

Search Intent: guide

Content Role: pillar

---

Event Sourcing Patterns: Aggregate Design and Snapshots

Event sourcing has become a critical architectural pattern for modern distributed systems that require complete audit trails, temporal queries, and reliable event-driven architectures. However, teams implementing event sourcing in production frequently encounter severe performance degradation when aggregates accumulate thousands of events, leading to unacceptable latency during event replay and increased infrastructure costs. Without proper aggregate design and snapshot strategies, systems that initially perform well can degrade to the point of operational failure as data volume grows.

The challenge intensifies in 2025's cloud-native environments where serverless functions, containerized microservices, and distributed databases demand efficient resource utilization. A poorly designed aggregate that requires replaying 50,000 events to reconstruct state can consume excessive memory, timeout serverless functions, and create cascading failures across dependent services. This article provides production-tested patterns for designing aggregates and implementing snapshots that maintain sub-100ms response times even with years of historical events.

Why Traditional State Management Fails for Event-Sourced Systems

Traditional CRUD-based systems store only current state, making it impossible to reconstruct how that state evolved over time. When regulatory requirements demand complete audit trails, or when business logic depends on understanding state transitions, teams often bolt on separate audit logging systems that become inconsistent with the primary data store.

Event sourcing solves these problems by treating events as the source of truth, but introduces new challenges. The naive approach of replaying every event from the beginning of time works initially but fails catastrophically as aggregates age. Consider an e-commerce order aggregate that processes returns, modifications, and status updates over months—replaying thousands of events for every command becomes untenable.

Furthermore, traditional database indexing strategies don't apply to event stores. You can't simply add an index to speed up aggregate reconstruction. The fundamental operation—sequential event replay—requires a different optimization strategy entirely.

Designing Aggregates for Event Sourcing

An aggregate in event sourcing represents a consistency boundary that ensures business invariants are maintained. Proper aggregate design is the foundation of a performant event-sourced system.

Aggregate Root Pattern

The aggregate root is the single entry point for all commands and the entity responsible for enforcing business rules. Here's a production-grade implementation:

interface DomainEvent {
  aggregateId: string;
  eventType: string;
  timestamp: Date;
  version: number;
  data: unknown;
}

abstract class AggregateRoot {
  protected id: string;
  protected version: number = 0;
  private uncommittedEvents: DomainEvent[] = [];

  constructor(id: string) {
    this.id = id;
  }

  protected addEvent(eventType: string, data: unknown): void {
    const event: DomainEvent = {
      aggregateId: this.id,
      eventType,
      timestamp: new Date(),
      version: this.version + 1,
      data,
    };

    this.apply(event);
    this.uncommittedEvents.push(event);
  }

  protected abstract apply(event: DomainEvent): void;

  public loadFromHistory(events: DomainEvent[]): void {
    events.forEach(event => {
      this.apply(event);
      this.version = event.version;
    });
  }

  public getUncommittedEvents(): DomainEvent[] {
    return [...this.uncommittedEvents];
  }

  public markEventsAsCommitted(): void {
    this.uncommittedEvents = [];
  }
}

Concrete Aggregate Implementation

Here's a real-world order aggregate that demonstrates proper boundary design:

interface OrderCreatedEvent {
  customerId: string;
  items: Array<{ productId: string; quantity: number; price: number }>;
}

interface OrderItemAddedEvent {
  productId: string;
  quantity: number;
  price: number;
}

interface OrderShippedEvent {
  trackingNumber: string;
  carrier: string;
}

class OrderAggregate extends AggregateRoot {
  private customerId?: string;
  private items: Map<string, { quantity: number; price: number }> = new Map();
  private status: 'pending' | 'confirmed' | 'shipped' | 'delivered' = 'pending';
  private totalAmount: number = 0;

  createOrder(customerId: string, items: Array<{ productId: string; quantity: number; price: number }>): void {
    if (this.customerId) {
      throw new Error('Order already exists');
    }

    if (items.length === 0) {
      throw new Error('Order must contain at least one item');
    }

    this.addEvent('OrderCreated', { customerId, items });
  }

  addItem(productId: string, quantity: number, price: number): void {
    if (this.status !== 'pending') {
      throw new Error('Cannot modify order after confirmation');
    }

    if (quantity <= 0) {
      throw new Error('Quantity must be positive');
    }

    this.addEvent('OrderItemAdded', { productId, quantity, price });
  }

  shipOrder(trackingNumber: string, carrier: string): void {
    if (this.status !== 'confirmed') {
      throw new Error('Can only ship confirmed orders');
    }

    this.addEvent('OrderShipped', { trackingNumber, carrier });
  }

  protected apply(event: DomainEvent): void {
    switch (event.eventType) {
      case 'OrderCreated': {
        const data = event.data as OrderCreatedEvent;
        this.customerId = data.customerId;
        data.items.forEach(item => {
          this.items.set(item.productId, { quantity: item.quantity, price: item.price });
          this.totalAmount += item.quantity * item.price;
        });
        this.status = 'pending';
        break;
      }
      case 'OrderItemAdded': {
        const data = event.data as OrderItemAddedEvent;
        const existing = this.items.get(data.productId);
        if (existing) {
          existing.quantity += data.quantity;
        } else {
          this.items.set(data.productId, { quantity: data.quantity, price: data.price });
        }
        this.totalAmount += data.quantity * data.price;
        break;
      }
      case 'OrderShipped': {
        this.status = 'shipped';
        break;
      }
    }
  }

  public getStatus(): string {
    return this.status;
  }

  public getTotalAmount(): number {
    return this.totalAmount;
  }
}

Aggregate Design Principles

Keep aggregates small and focused. Each aggregate should protect a single consistency boundary. In the example above, the order aggregate doesn't manage inventory—that's a separate aggregate. This prevents distributed transactions and keeps event streams manageable.

Design for eventual consistency between aggregates. Use domain events to communicate between aggregates asynchronously. When an order is created, publish an event that the inventory aggregate can consume to reserve stock.

Avoid aggregate references. Store only IDs of related aggregates, never full object references. This prevents accidentally loading multiple aggregates and violates the consistency boundary.

Implementing Snapshots for Performance

Snapshots are point-in-time representations of aggregate state that eliminate the need to replay all historical events. They're essential for production systems with long-lived aggregates.

Snapshot Strategy

interface Snapshot {
  aggregateId: string;
  version: number;
  timestamp: Date;
  state: unknown;
}

class SnapshotStore {
  private snapshots: Map<string, Snapshot> = new Map();

  async saveSnapshot(aggregateId: string, version: number, state: unknown): Promise<void> {
    const snapshot: Snapshot = {
      aggregateId,
      version,
      timestamp: new Date(),
      state,
    };

    // In production, persist to database
    this.snapshots.set(aggregateId, snapshot);
  }

  async getLatestSnapshot(aggregateId: string): Promise<Snapshot | null> {
    return this.snapshots.get(aggregateId) || null;
  }
}

class EventStore {
  private events: Map<string, DomainEvent[]> = new Map();

  async appendEvents(aggregateId: string, events: DomainEvent[], expectedVersion: number): Promise<void> {
    const existing = this.events.get(aggregateId) || [];

    if (existing.length !== expectedVersion) {
      throw new Error('Concurrency conflict detected');
    }

    this.events.set(aggregateId, [...existing, ...events]);
  }

  async getEvents(aggregateId: string, fromVersion: number = 0): Promise<DomainEvent[]> {
    const events = this.events.get(aggregateId) || [];
    return events.filter(e => e.version > fromVersion);
  }
}

Repository with Snapshot Support

class AggregateRepository<T extends AggregateRoot> {
  constructor(
    private eventStore: EventStore,
    private snapshotStore: SnapshotStore,
    private aggregateFactory: (id: string) => T,
    private snapshotInterval: number = 100
  ) {}

  async load(aggregateId: string): Promise<T> {
    const aggregate = this.aggregateFactory(aggregateId);

    // Try to load from snapshot first
    const snapshot = await this.snapshotStore.getLatestSnapshot(aggregateId);

    let fromVersion = 0;
    if (snapshot) {
      // Restore state from snapshot
      this.restoreFromSnapshot(aggregate, snapshot);
      fromVersion = snapshot.version;
    }

    // Load and apply events since snapshot
    const events = await this.eventStore.getEvents(aggregateId, fromVersion);
    aggregate.loadFromHistory(events);

    return aggregate;
  }

  async save(aggregate: T): Promise<void> {
    const events = aggregate.getUncommittedEvents();

    if (events.length === 0) {
      return;
    }

    const currentVersion = (aggregate as any).version - events.length;
    await this.eventStore.appendEvents((aggregate as any).id, events, currentVersion);

    aggregate.markEventsAsCommitted();

    // Create snapshot if threshold reached
    const newVersion = (aggregate as any).version;
    if (newVersion % this.snapshotInterval === 0) {
      await this.createSnapshot(aggregate);
    }
  }

  private async createSnapshot(aggregate: T): Promise<void> {
    const state = this.serializeAggregate(aggregate);
    await this.snapshotStore.saveSnapshot(
      (aggregate as any).id,
      (aggregate as any).version,
      state
    );
  }

  private serializeAggregate(aggregate: T): unknown {
    // Serialize only the necessary state
    return JSON.parse(JSON.stringify(aggregate));
  }

  private restoreFromSnapshot(aggregate: T, snapshot: Snapshot): void {
    Object.assign(aggregate, snapshot.state);
  }
}

Snapshot Frequency Optimization

The snapshot interval is critical. Too frequent creates storage overhead; too infrequent degrades performance. For most systems, snapshot every 50-100 events. Monitor aggregate reconstruction time and adjust accordingly.

class AdaptiveSnapshotStrategy {
  private reconstructionTimes: Map<string, number[]> = new Map();

  shouldSnapshot(aggregateId: string, eventsSinceSnapshot: number, reconstructionTimeMs: number): boolean {
    const times = this.reconstructionTimes.get(aggregateId) || [];
    times.push(reconstructionTimeMs);

    // Keep last 10 measurements
    if (times.length > 10) {
      times.shift();
    }
    this.reconstructionTimes.set(aggregateId, times);

    const avgTime = times.reduce((a, b) => a + b, 0) / times.length;

    // Snapshot if reconstruction takes > 50ms or > 100 events
    return avgTime > 50 || eventsSinceSnapshot > 100;
  }
}

Common Pitfalls and Edge Cases

Concurrency Conflicts

Event sourcing requires optimistic concurrency control. Always check the expected version when appending events:

async function handleConcurrencyConflict<T extends AggregateRoot>(
  repository: AggregateRepository<T>,
  aggregateId: string,
  operation: (aggregate: T) => void,
  maxRetries: number = 3
): Promise<void> {
  for (let attempt = 0; attempt < maxRetries; attempt++) {
    try {
      const aggregate = await repository.load(aggregateId);
      operation(aggregate);
      await repository.save(aggregate);
      return;
    } catch (error) {
      if (error instanceof Error && error.message.includes('Concurrency conflict')) {
        if (attempt === maxRetries - 1) throw error;
        // Exponential backoff
        await new Promise(resolve => setTimeout(resolve, Math.pow(2, attempt) * 100));
        continue;
      }
      throw error;
    }
  }
}

Snapshot Corruption

Snapshots can become corrupted or incompatible after code changes. Always validate snapshots and fall back to event replay:

class SafeSnapshotStore extends SnapshotStore {
  async getLatestSnapshot(aggregateId: string): Promise<Snapshot | null> {
    const snapshot = await super.getLatestSnapshot(aggregateId);

    if (!snapshot) return null;

    try {
      this.validateSnapshot(snapshot);
      return snapshot;
    } catch (error) {
      console.warn(`Invalid snapshot for ${aggregateId}, falling back to full replay`);
      return null;
    }
  }

  private validateSnapshot(snapshot: Snapshot): void {
    // Validate schema version, required fields, etc.
    if (!snapshot.state || typeof snapshot.state !== 'object') {
      throw new Error('Invalid snapshot state');
    }
  }
}

Event Schema Evolution

Events are immutable, but business requirements change. Use event upcasting to handle schema evolution:

interface EventUpcaster {
  canUpcast(event: DomainEvent): boolean;
  upcast(event: DomainEvent): DomainEvent;
}

class OrderCreatedV1ToV2Upcaster implements EventUpcaster {
  canUpcast(event: DomainEvent): boolean {
    return event.eventType === 'OrderCreated' && !('currency' in (event.data as any));
  }

  upcast(event: DomainEvent): DomainEvent {
    return {
      ...event,
      data: {
        ...(event.data as object),
        currency: 'USD', // Default currency for old events
      },
    };
  }
}

class EventStoreWithUpcasting extends EventStore {
  constructor(private upcasters: EventUpcaster[]) {
    super();
  }

  async getEvents(aggregateId: string, fromVersion: number = 0): Promise<DomainEvent[]> {
    const events = await super.getEvents(aggregateId, fromVersion);
    return events.map(event => this.upcastEvent(event));
  }

  private upcastEvent(event: DomainEvent): DomainEvent {
    let upcastedEvent = event;

    for (const upcaster of this.upcasters) {
      if (upcaster.canUpcast(upcastedEvent)) {
        upcastedEvent = upcaster.upcast(upcastedEvent);
      }
    }

    return upcastedEvent;
  }
}

Memory Management

Large aggregates can cause memory issues. Implement streaming event replay for extremely large aggregates:

class StreamingEventStore extends EventStore {
  async *streamEvents(aggregateId: string, fromVersion: number = 0): AsyncGenerator<DomainEvent> {
    const events = await this.getEvents(aggregateId, fromVersion);

    // In production, fetch in batches from database
    for (const event of events) {
      yield event;
    }
  }
}

class StreamingAggregateRepository<T extends AggregateRoot> extends AggregateRepository<T> {
  async loadStreaming(aggregateId: string): Promise<T> {
    const aggregate = this.aggregateFactory(aggregateId);
    const snapshot = await this.snapshotStore.getLatestSnapshot(aggregateId);

    let fromVersion = 0;
    if (snapshot) {
      this.restoreFromSnapshot(aggregate, snapshot);
      fromVersion = snapshot.version;
    }

    const eventStream = (this.eventStore as StreamingEventStore).streamEvents(aggregateId, fromVersion);

    for await (const event of eventStream) {
      aggregate.loadFromHistory([event]);
    }

    return aggregate;
  }
}

Best Practices for Production Systems

1. Implement Health Checks

Monitor aggregate reconstruction time and snapshot effectiveness:

class EventSourcingMetrics {
  private reconstructionTimes: number[] = [];
  private snapshotHitRate: { hits: number; misses: number } = { hits: 0, misses: 0 };

  recordReconstruction(timeMs: number, usedSnapshot: boolean): void {
    this.reconstructionTimes.push(timeMs);
    if (usedSnapshot) {
      this.snapshotHitRate.hits++;
    } else {
      this.snapshotHitRate.misses++;
    }
  }

  getMetrics() {
    const avg = this.reconstructionTimes.reduce((a, b) => a + b, 0) / this.reconstructionTimes.length;
    const p95 = this.calculatePercentile(this.reconstructionTimes, 0.95);
    const hitRate = this.snapshotHitRate.hits / (this.snapshotHitRate.hits + this.snapshotHitRate.misses);

    return { avgReconstructionMs: avg, p95ReconstructionMs: p95, snapshotHitRate: hitRate };
  }

  private calculatePercentile(values: number[], percentile: number): number {
    const sorted = [...values].sort((a, b) => a - b);
    const index = Math.ceil(sorted.length * percentile) - 1;
    return sorted[index];
  }
}

2. Aggregate Design Checklist

• Single responsibility: Each aggregate protects one consistency boundary
• Small event streams: Target < 1000 events per aggregate; consider splitting if exceeded
• Idempotent event handlers: Apply method should be side-effect free and deterministic
• Version tracking: Always track and validate aggregate version
• Snapshot strategy: Define clear snapshot intervals based on performance metrics

3. Event Store Selection

Choose an event store based on your requirements:

• EventStoreDB: Purpose-built for event sourcing, excellent projections support
• PostgreSQL with JSONB: Cost-effective, good for moderate scale
• Apache Kafka: High throughput, excellent for event streaming architectures
• DynamoDB: Serverless-friendly, good for AWS-native applications

4. Testing Strategy

Test aggregates in isolation with event replay:

```typescript describe('OrderAggregate', () => { it('should prevent modification after shipping', () => { const order = new OrderAggregate('order-123');

// Given: Order is created and shipped