Implement Event Sourcing in Serverless Architectures

Overview

Most systems only keep the latest state around. An order reads “shipped” and the database row just says status = shipped. Ask when that status changed and the database stays quiet — the old value was overwritten. How many orders got cancelled and re-shipped last month? Without hand-maintained audit columns, nobody knows.

Event sourcing writes every state change down as an immutable event in an append-only log, and the current state comes from replaying them. An order’s state isn’t a row; it’s the ordered list [OrderCreated, ItemAdded, PaymentProcessed, Shipped]. That gives you a complete audit trail, temporal queries (“what was the state at 3pm yesterday?”), and the ability to rebuild projections from scratch. In serverless architectures, events are captured via DynamoDB streams, SQS, or EventBridge, and Lambda functions project the read model.

The implementation below uses DynamoDB as the event store, DynamoDB Streams to trigger the projection Lambda, and in-memory replay to rebuild the aggregate.

When to use it

Use this recipe when:

• A complete audit history of all changes is a business requirement. See Event-Driven Functions for event-driven architectures.
• You need to answer temporal questions about past states
• Rebuilding read models from scratch is a needed capability. See Serverless Orchestration for managing stateful workflows.
• The write model is complex and the read model needs separate optimization
• Compliance or regulatory requirements mandate immutable change logs. See CQRS Pattern for separating read and write models.

Solution

DynamoDB Event Store with Streams

interface DomainEvent {
  eventId: string;
  aggregateId: string;
  eventType: string;
  payload: Record<string, unknown>;
  timestamp: string;
  version: number;
}

class OrderEventStore {
  constructor(private tableName: string, private client: DynamoDBDocument) {}

  async appendEvents(aggregateId: string, events: DomainEvent[]): Promise<void> {
    const currentVersion = await this.getCurrentVersion(aggregateId);

    const transactItems = events.map((event, index) => ({
      Put: {
        TableName: this.tableName,
        Item: {
          pk: `ORDER#${aggregateId}`,
          sk: `EVENT#${(currentVersion + index + 1).toString().padStart(10, '0')}`,
          eventId: event.eventId,
          eventType: event.eventType,
          payload: event.payload,
          timestamp: new Date().toISOString(),
          version: currentVersion + index + 1,
        },
        ConditionExpression: 'attribute_not_exists(pk)',
      },
    }));

    await this.client.transactWrite({ TransactItems: transactItems });
  }

  async getEvents(aggregateId: string): Promise<DomainEvent[]> {
    const result = await this.client.query({
      TableName: this.tableName,
      KeyConditionExpression: 'pk = :pk AND begins_with(sk, :sk)',
      ExpressionAttributeValues: {
        ':pk': `ORDER#${aggregateId}`,
        ':sk': 'EVENT#',
      },
      ScanIndexForward: true,
    });

    return (result.Items || []).map(item => ({
      eventId: item.eventId,
      aggregateId,
      eventType: item.eventType,
      payload: item.payload,
      timestamp: item.timestamp,
      version: item.version,
    }));
  }

  private async getCurrentVersion(aggregateId: string): Promise<number> {
    const events = await this.getEvents(aggregateId);
    return events.length > 0 ? events[events.length - 1].version : 0;
  }
}

Lambda Projection Handler

export const handler = async (event: DynamoDBStreamEvent): Promise<void> => {
  for (const record of event.Records) {
    if (record.eventName !== 'INSERT') continue;

    const newImage = unmarshall(record.dynamodb?.NewImage as any);
    const domainEvent: DomainEvent = {
      eventId: newImage.eventId,
      aggregateId: newImage.aggregateId,
      eventType: newImage.eventType,
      payload: newImage.payload,
      timestamp: newImage.timestamp,
      version: newImage.version,
    };

    await projectEvent(domainEvent);
  }
};

async function projectEvent(event: DomainEvent): Promise<void> {
  switch (event.eventType) {
    case 'OrderCreated':
      await createOrderProjection(event.aggregateId, event.payload);
      break;
    case 'ItemAdded':
      await addItemToOrderProjection(event.aggregateId, event.payload);
      break;
    case 'OrderShipped':
      await updateOrderStatus(event.aggregateId, 'shipped');
      break;
  }
}

Order Aggregate Reconstruction

class OrderAggregate {
  private status: string = 'pending';
  private items: OrderItem[] = [];
  private total: number = 0;

  applyEvent(event: DomainEvent): void {
    switch (event.eventType) {
      case 'OrderCreated':
        this.status = 'created';
        this.total = event.payload.total as number;
        break;
      case 'ItemAdded':
        this.items.push(event.payload.item as OrderItem);
        this.total += (event.payload.item as OrderItem).price;
        break;
      case 'OrderShipped':
        this.status = 'shipped';
        break;
      case 'OrderCancelled':
        this.status = 'cancelled';
        break;
    }
  }

  static fromEvents(events: DomainEvent[]): OrderAggregate {
    const order = new OrderAggregate();
    for (const event of events) {
      order.applyEvent(event);
    }
    return order;
  }

  toSnapshot(): OrderSnapshot {
    return {
      status: this.status,
      items: this.items,
      total: this.total,
    };
  }
}

Companion code: the event-sourcing-serverless companion repo has these three TypeScript files ready to drop into a Lambda project.

Explanation

• Event store: nothing more than an append-only log — once written, an event is never edited or removed. Each one carries its own ID, the aggregate it belongs to, a type, a payload, and a version. The version gives you ordering and optimistic concurrency control via ConditionExpression.
• Aggregate reconstruction: nobody stores the current state of an order. To get it back, you load the aggregate’s events and replay them in order — the object starts empty and applies each one. Replay is deterministic: same sequence in, same state out.
• Projections (read models): each projection subscribes to the event stream. When an event lands, the stream-triggered Lambda updates that projection’s read-optimized view. You can run several projections off the same events — one for the customer dashboard, one for analytics, one for search indexing.
• Snapshots: an aggregate that’s accumulated thousands of events replays slowly. Snapshots cache its state at a specific version; on load you fetch the latest snapshot and replay only what came after. Take them every 50–100 events and asynchronously, so the write path stays fast.

Variants

Approach	Store	Projections	Best for
DynamoDB + Streams	DynamoDB	Lambda	AWS-native, moderate scale
EventStoreDB	EventStoreDB	Subscriptions	High volume, complex domains
Kafka + KTables	Kafka	Kafka Streams	Stream processing, replay
S3 + Athena	S3	Athena queries	Audit, compliance, analytics
Aurora + Outbox	PostgreSQL	CDC	Relational event sourcing

Python event store (boto3)

The same event store in Python — transact_write_items keeps the optimistic concurrency, TypeSerializer handles the typed-attribute noise:

import boto3
from boto3.dynamodb.types import TypeSerializer
from boto3.dynamodb.conditions import Key
from datetime import datetime, timezone

serializer = TypeSerializer()

class OrderEventStore:
    def __init__(self, table_name: str):
        self.table_name = table_name
        self.client = boto3.client("dynamodb")
        self.table = boto3.resource("dynamodb").Table(table_name)

    def append_events(self, aggregate_id: str, events: list[dict]) -> None:
        current = self.current_version(aggregate_id)
        items = []
        for i, event in enumerate(events):
            version = current + i + 1
            item = {
                "pk": f"ORDER#{aggregate_id}",
                "sk": f"EVENT#{version:010d}",
                "eventId": event["event_id"],
                "eventType": event["event_type"],
                "payload": event["payload"],
                "timestamp": datetime.now(timezone.utc).isoformat(),
                "version": version,
            }
            items.append({
                "Put": {
                    "TableName": self.table_name,
                    "Item": {k: serializer.serialize(v) for k, v in item.items()},
                    "ConditionExpression": "attribute_not_exists(pk)",
                }
            })
        self.client.transact_write_items(TransactItems=items)

    def get_events(self, aggregate_id: str) -> list[dict]:
        resp = self.table.query(
            KeyConditionExpression=(
                Key("pk").eq(f"ORDER#{aggregate_id}") & Key("sk").begins_with("EVENT#")
            ),
            ScanIndexForward=True,
        )
        return resp.get("Items", [])

    def current_version(self, aggregate_id: str) -> int:
        events = self.get_events(aggregate_id)
        return events[-1]["version"] if events else 0

The projection handler on the Python side is a plain Lambda handler over event["Records"] — same INSERT filter, same unmarshalling idea via TypeDeserializer.

What Works

• Version every event: include a monotonically increasing version per aggregate. It prevents lost updates when two users modify the same aggregate at once.
• Make events immutable and self-contained: each event carries everything needed to make sense of it — not just deltas. OrderCreated should include customer ID, shipping address, and line items — not just “order 123 was created.” Future consumers shouldn’t need to query other systems to interpret the event.
• Use correlation IDs across the event chain: when an event triggers another (e.g., OrderShipped triggers InventoryDecremented), propagate the correlation ID. That’s what makes tracing and debugging across a distributed event chain tractable.
• Implement idempotent projections: Lambda retries on failure, and a projection that increments a counter on each invocation will overcount. Write the event ID into the projection row and skip if it’s already processed.
• Archive old events to cold storage: DynamoDB gets expensive when you’re holding millions of events long-term. Move events older than 90 days to S3 with DynamoDB TTL or export jobs.

Common mistakes

• Storing current state alongside events: keeping both an event log and a current-state table lets them diverge — a projection bug writes state A while the log holds events for state B. The event store is the source of truth; projections are derived, never primary.
• Exposing internal event types to external systems: external consumers shouldn’t depend on your internal schemas. Define a public event contract (e.g., OrderConfirmed) and map internal events onto it, so refactors don’t break integrations.
• Ignoring event schema evolution: when an event type gains a field, old events in the log don’t have it, and your aggregate code has to tolerate the gaps — better yet, version events and write upcasters.
• Replaying from the beginning on every query: replaying 10,000 events per GET /order/123 destroys performance. Use snapshots for long-lived aggregates and load from them.
• Assuming ordered, exactly-once delivery: what you actually get is at-least-once delivery, ordered only within a shard. Check sequence numbers and make projections idempotent instead of trusting arrival order.

See Also

• AWS — DynamoDB Streams — official docs on stream records, per-shard ordering, and Lambda triggers.
• Martin Fowler — Event Sourcing — the canonical description of the pattern and its trade-offs.
• Event sourcing in a relational database — the same pattern with PostgreSQL/MySQL instead of DynamoDB.
• Event sourcing with CQRS in Python — aggregates, projections, and snapshots in Python.