Event Sourcing: Building Event-Driven Systems

In modern distributed systems, maintaining data consistency, tracking changes, and scaling effectively can be challenging. Event Sourcing offers a powerful architectural pattern that addresses these challenges by storing all changes to an application’s state as a sequence of events. Let’s explore how to implement this pattern in a production environment.

Why Event Sourcing?

Before diving into implementation details, let’s understand why you might want to use Event Sourcing:

Complete Audit Trail: Every state change is captured as an immutable event, providing a perfect audit history.
Temporal Queries: You can determine the system’s state at any point in time by replaying events.
Debug Friendly: When issues occur, you have a complete history of what led to the current state.
Event Replay: You can fix bugs by correcting the event handling logic and replaying events.
Scale Write/Read Separately: Event storage and read models can be scaled independently.

Core Components

The Event Store

The Event Store is the heart of any event-sourced system. It’s responsible for storing and retrieving events while ensuring consistency. Here’s a TypeScript implementation that handles the core functionality:

interface Event {
  id: string;
  type: string;
  aggregateId: string;
  version: number;
  payload: any;
  timestamp: Date;
}

class EventStore {
  constructor(
    private readonly eventDatabase: Database,
    private readonly eventBus: EventBus
  ) {}

  async append(events: Event[], expectedVersion: number): Promise<void> {
    await this.eventDatabase.transaction(async (session) => {
      const currentVersion = await this.getCurrentVersion(
        events[0].aggregateId,
        session
      );

      if (currentVersion !== expectedVersion) {
        throw new ConcurrencyError();
      }

      await this.saveEvents(events, session);
      await this.eventBus.publish(events);
    });
  }

  async getEvents(aggregateId: string): Promise<Event[]> {
    return this.eventDatabase.query(`
      SELECT * FROM events 
      WHERE aggregate_id = $1 
      ORDER BY version ASC
    `, [aggregateId]);
  }
}

The Event Store implementation includes optimistic concurrency control to prevent conflicts when multiple processes attempt to modify the same aggregate simultaneously. The expectedVersion parameter ensures we’re working with the latest state before applying new changes.

Aggregates: Managing Business Entities

Aggregates in event sourcing represent business entities that maintain consistency boundaries. They handle commands, validate business rules, and emit events. Our base Aggregate class provides the foundation for implementing domain-specific aggregates:

abstract class Aggregate {
  private version: number = 0;
  private changes: Event[] = [];

  abstract apply(event: Event): void;

  protected raiseEvent(type: string, payload: any): void {
    const event: Event = {
      id: uuid(),
      type,
      aggregateId: this.id,
      version: this.version + 1,
      payload,
      timestamp: new Date()
    };

    this.apply(event);
    this.changes.push(event);
    this.version++;
  }

  getUncommittedChanges(): Event[] {
    return this.changes;
  }

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

Performance Optimization: Snapshots

As your event store grows, replaying all events to reconstruct an aggregate’s state becomes expensive. Snapshots solve this by periodically capturing the aggregate’s state:

interface Snapshot {
  aggregateId: string;
  version: number;
  state: any;
}

class SnapshotStore {
  constructor(private readonly database: Database) {}

  async save(snapshot: Snapshot): Promise<void> {
    await this.database.execute(`
      INSERT INTO snapshots (aggregate_id, version, state)
      VALUES ($1, $2, $3)
      ON CONFLICT (aggregate_id) 
      DO UPDATE SET version = $2, state = $3
    `, [snapshot.aggregateId, snapshot.version, snapshot.state]);
  }

  async getLatest(aggregateId: string): Promise<Snapshot | null> {
    return this.database.queryOne(`
      SELECT * FROM snapshots
      WHERE aggregate_id = $1
    `, [aggregateId]);
  }
}

Production Deployment

Deploying an event-sourced system in production requires careful consideration of infrastructure. Here’s a Kubernetes configuration that addresses key operational concerns:

apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: event-store
spec:
  replicas: 3
  serviceName: event-store
  template:
    spec:
      containers:
      - name: event-store
        image: event-store:latest
        env:
        - name: EVENT_DB_HOST
          value: postgres-event-store
        - name: SNAPSHOT_DB_HOST
          value: mongodb-snapshots
        volumeMounts:
        - name: event-storage
          mountPath: /data
  volumeClaimTemplates:
  - metadata:
      name: event-storage
    spec:
      accessModes: [ "ReadWriteOnce" ]
      resources:
        requests:
          storage: 100Gi

The configuration above uses a StatefulSet to ensure stable network identities and persistent storage for our event store pods. We use separate databases for events and snapshots to optimize for their different access patterns.

Operational Considerations

When implementing event sourcing in production, consider these critical factors:

Event Schema Evolution: As your system evolves, your event schemas will need to change. Version your events and maintain backward compatibility.
Storage Growth: Events are never deleted, so plan for continuous storage growth. Implement retention policies for old events after they’re captured in snapshots.
Snapshot Frequency: Configure snapshot frequency based on:
- Average event count per aggregate
- Acceptable replay time
- Storage costs
Monitoring: Track key metrics:
- Event processing latency
- Event store size growth
- Snapshot creation time
- Event replay performance

Here’s a sample configuration that addresses these concerns:

apiVersion: v1
kind: ConfigMap
metadata:
  name: event-sourcing-config
data:
  event-store.yaml: |
    snapshot_frequency: 100
    retention_days: 365
    max_events_per_aggregate: 1000
        
  projections.yaml: |
    workers: 5
    batch_size: 100
    retry_attempts: 3

Conclusion

Event sourcing provides powerful capabilities for building systems that need complete audit trails, temporal queries, and separate scaling of read and write operations. While the implementation requires careful consideration of various factors, the benefits often outweigh the complexity for systems that need these capabilities.

Remember to start small, perhaps with a bounded context that would benefit most from event sourcing, rather than trying to event-source your entire system at once. Monitor your implementation carefully and adjust your configuration based on real-world usage patterns.

Why Event Sourcing?#

Core Components#

The Event Store#

Aggregates: Managing Business Entities#

Performance Optimization: Snapshots#

Production Deployment#

Operational Considerations#

Conclusion#