Serverless Event Sourcing with AWS Lambda and DynamoDB: Implementing CQRS for Scalable, Fault-Tolerant APIs

Building scalable, resilient, and maintainable APIs in a serverless architecture can be tricky. When you need high availability, low latency, and support for complex business logic, Event Sourcing combined with CQRS (Command Query Responsibility Segregation) offers a powerful pattern.

In this article, we’ll explore how to implement Event Sourcing and CQRS using AWS Lambda and DynamoDB, allowing you to build fault-tolerant, efficient, and scalable APIs without managing servers.

Step 1: Understand the Basics of Event Sourcing and CQRS

• Event Sourcing ensures that state changes are stored as immutable events. Rather than storing the current state, each change is stored as a separate event, which can be replayed or processed later.

• CQRS is the pattern where you separate the read model from the write model. This allows for more flexible querying and scaling, as the system can optimize both operations independently.

Step 2: Set Up DynamoDB for Event Storage

You’ll need a DynamoDB table to store events. Each event can be stored as a record with an event_id, aggregate_id (such as a user or order), timestamp, and event_type.

Create the table with the following schema:

• Partition Key: aggregate_id
• Sort Key: event_id
• Additional fields: event_data (stores the event payload) and timestamp

aws dynamodb create-table \
  --table-name events \
  --attribute-definitions \
    AttributeName=aggregate_id,AttributeType=S \
    AttributeName=event_id,AttributeType=S \
  --key-schema \
    AttributeName=aggregate_id,KeyType=HASH \
    AttributeName=event_id,KeyType=RANGE \
  --provisioned-throughput \
    ReadCapacityUnits=5,WriteCapacityUnits=5

This structure allows you to store each event in sequence while linking them to specific aggregates (such as a user).

Step 3: Implement the Command Side (Write Model) in AWS Lambda

In the command side, AWS Lambda functions handle incoming events, process business logic, and persist those events to DynamoDB.

Here’s an example Lambda function that processes an event and stores it in DynamoDB:

const AWS = require('aws-sdk');
const dynamoDB = new AWS.DynamoDB.DocumentClient();

exports.handler = async (event) => {
  const { aggregateId, eventData, eventType } = JSON.parse(event.body);
  
  const eventId = `event-${Date.now()}`;
  const timestamp = new Date().toISOString();
  
  const params = {
    TableName: 'events',
    Item: {
      aggregate_id: aggregateId,
      event_id: eventId,
      event_data: eventData,
      event_type: eventType,
      timestamp,
    }
  };
  
  await dynamoDB.put(params).promise();
  
  return {
    statusCode: 200,
    body: JSON.stringify({ message: 'Event stored successfully' })
  };
};

This Lambda function listens for an event, processes it (in this case, a simple aggregateId, eventData, and eventType), and writes it to DynamoDB.

Step 4: Implement the Query Side (Read Model) with a Separate Lambda

On the query side, you can use a DynamoDB Global Secondary Index (GSI) to create an optimized view for reads. For example, if you need to fetch all events for a particular user (the aggregate_id), you can do so efficiently with a GSI.

First, define the GSI with event_type as the partition key and timestamp as the sort key.

aws dynamodb update-table \
  --table-name events \
  --attribute-definitions \
    AttributeName=event_type,AttributeType=S \
    AttributeName=timestamp,AttributeType=S \
  --global-secondary-index-updates \
    '[{
      "Create": {
        "IndexName": "EventTypeTimestampIndex",
        "KeySchema": [
          { "AttributeName": "event_type", "KeyType": "HASH" },
          { "AttributeName": "timestamp", "KeyType": "RANGE" }
        ],
        "Projection": {
          "ProjectionType": "ALL"
        },
        "ProvisionedThroughput": {
          "ReadCapacityUnits": 5,
          "WriteCapacityUnits": 5
        }
      }
    }]'

Now, implement the Lambda function to query the event stream, returning the events in chronological order:

exports.handler = async (event) => {
  const { aggregateId } = event.pathParameters;
  
  const params = {
    TableName: 'events',
    KeyConditionExpression: 'aggregate_id = :aggregateId',
    ExpressionAttributeValues: {
      ':aggregateId': aggregateId,
    }
  };
  
  const result = await dynamoDB.query(params).promise();
  
  return {
    statusCode: 200,
    body: JSON.stringify(result.Items)
  };
};

This Lambda reads from DynamoDB and returns all events for the specified aggregateId.

Step 5: Implement Event Replay for Projections

Event replay is a core component of Event Sourcing. To build read models or projections, you can replay events by querying all events for a given aggregate and projecting them into a new format.

Here’s an example of a simple event replay mechanism:

async function replayEvents(aggregateId) {
  const params = {
    TableName: 'events',
    KeyConditionExpression: 'aggregate_id = :aggregateId',
    ExpressionAttributeValues: {
      ':aggregateId': aggregateId,
    }
  };
  
  const result = await dynamoDB.query(params).promise();
  return result.Items.reduce((state, event) => {
    switch (event.event_type) {
      case 'user_created':
        state.user = { ...event.event_data };
        break;
      case 'user_updated':
        state.user = { ...state.user, ...event.event_data };
        break;
    }
    return state;
  }, {});
}

The replayEvents function replays all events for an aggregate, applying them to the state as the events are processed.

Step 6: Handle Eventual Consistency and Fault Tolerance

Since we’re using DynamoDB, eventual consistency is a factor. This can lead to situations where data may not be immediately available after an event is written. To mitigate this, you can implement optimistic concurrency control using versioning or implement compensating transactions in your business logic.

✅ Pros:

• ⚡ Scales easily with AWS Lambda and DynamoDB (no servers to manage)