Event Sourcing and CQRS Patterns on AWS

Event sourcing is a pattern where you store the state of an application as a sequence of immutable events, rather than storing the current state directly. Each event represents a state change (e.g., 'OrderPlaced', 'PaymentReceived', 'ItemShipped'). To get the current state, you replay all events from the beginning (or from a snapshot). The event store is the single source of truth.

Amazon DynamoDB is a popular choice for an event store. The table has a partition key (e.g., aggregate ID) and a sort key (e.g., event sequence number or timestamp). Each item is an event with attributes like event type, data, and metadata. The table is append-only—you never update or delete events. To get the current state, you query all events for a given aggregate ID, sort by sequence number, and apply them in order.

Example event schema in DynamoDB:

{
  "aggregateId": "order-123",
  "eventId": "event-456",
  "eventType": "OrderPlaced",
  "data": { "customerId": "cust-789", "total": 100.00 },
  "timestamp": 1700000000
}

To optimize reads, you can take periodic snapshots of the aggregate state and store them in a separate table or in the same table with a different sort key prefix (e.g., 'SNAPSHOT#'). When replaying, you start from the latest snapshot and only replay events after it.

DynamoDB Streams captures a time-ordered sequence of item-level changes in a DynamoDB table. When you use DynamoDB as an event store, you can enable streams to push events to AWS Lambda, Kinesis, or other consumers. This enables event-driven processing and integration with other services.

Important: DynamoDB Streams records changes in near-real-time (typically within seconds). The stream retains data for 24 hours. Each shard processes events in order, but across shards order is not guaranteed.

CQRS stands for Command Query Responsibility Segregation. It separates the operations that modify data (commands) from those that read data (queries). In a traditional CRUD system, the same model handles both reads and writes. In CQRS, you have separate models, often backed by separate data stores.

A common implementation uses: - Write store: DynamoDB table with events (event sourcing) or a relational database. - Read store: Amazon ElastiCache, DynamoDB Global Tables, or Amazon Aurora Read Replicas. - Event bus: Amazon EventBridge or SNS/SQS to propagate events from write side to read side.

Example: An e-commerce system writes orders to a DynamoDB table (event store). A Lambda function triggered by DynamoDB Streams updates a read-optimized table in DynamoDB that stores order summaries by customer, or updates an ElastiCache cluster for low-latency reads.

Event sourcing naturally pairs with CQRS. The event store is the write model (commands append events). The read model is built by subscribing to the event stream and projecting the events into a denormalized view. This is called event projection.

Projection example: A Lambda function listens to DynamoDB Streams, reads 'OrderPlaced' and 'ItemShipped' events, and updates a DynamoDB table that tracks order status by customer ID. The read table might have attributes like customerId, orderId, status, lastUpdated.

Kinesis Data Streams is another option for event sourcing. Events are published to a shard, and consumers read them in order. Kinesis retains records for up to 365 days (default 24 hours). You can use Kinesis Data Analytics for real-time processing or Kinesis Data Firehose to land events in S3 for archival.

Kinesis vs. DynamoDB Streams: - DynamoDB Streams is tightly coupled to DynamoDB tables, simpler for single-table event sourcing. - Kinesis is more flexible for high-throughput, multi-producer, multi-consumer scenarios. - Kinesis has higher per-shard throughput (1 MB/s write, 2 MB/s read).

Event sourcing typically uses eventual consistency for read models. The write side is strongly consistent if you use DynamoDB with strongly consistent reads (but this adds cost). CQRS often embraces eventual consistency: after a command is processed, the read model is updated asynchronously. The exam tests your understanding of when eventual consistency is acceptable.

Commands should be idempotent to handle retries. Use a unique request ID and store it alongside the event. On duplicate command, return the existing event without creating a new one.

To avoid replaying all events from the beginning, take periodic snapshots. In DynamoDB, you can store a snapshot item with sort key 'SNAPSHOT#<sequence>'. When loading an aggregate, first load the latest snapshot, then replay events after that snapshot's sequence number.

For DVA-C02, you need to:

import boto3
import uuid
import time

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('EventStore')

def append_event(aggregate_id, event_type, data):
    event_id = str(uuid.uuid4())
    timestamp = int(time.time())
    table.put_item(
        Item={
            'aggregateId': aggregate_id,
            'eventId': f'{timestamp}#{event_id}',
            'eventType': event_type,
            'data': data,
            'timestamp': timestamp
        },
        ConditionExpression='attribute_not_exists(eventId)'
    )
    return event_id

def get_aggregate_state(aggregate_id):
    response = table.query(
        KeyConditionExpression=Key('aggregateId').eq(aggregate_id),
        ScanIndexForward=True  # ascending order by sort key
    )
    state = {}
    for event in response['Items']:
        # apply event to state (domain logic)
        if event['eventType'] == 'OrderPlaced':
            state['status'] = 'placed'
            state['total'] = event['data']['total']
        elif event['eventType'] == 'PaymentReceived':
            state['status'] = 'paid'
    return state

A fintech company needs an immutable audit trail for all transactions. They use event sourcing with DynamoDB as the event store. Each transaction (deposit, withdrawal, transfer) is appended as an event. The read side uses CQRS: a DynamoDB table stores current balances per account, updated asynchronously via Lambda and DynamoDB Streams. This allows real-time balance checks while maintaining full history. At scale (millions of transactions per day), they use DynamoDB auto-scaling and DAX for read caching. A common misconfiguration is not using conditional writes, leading to duplicate events. They also snapshot aggregates daily to avoid replaying millions of events on every load.

An e-commerce platform uses CQRS to separate order placement (write) from order tracking (read). The write side uses DynamoDB with event sourcing. The read side uses an ElastiCache Redis cluster to provide low-latency order status for customers. When an order is placed, a Lambda function writes the event to DynamoDB and also updates Redis. Another Lambda processes the stream to update a reporting database in Aurora. The system handles 10,000 orders per minute. A pitfall is not handling eventual consistency: customers occasionally see stale statuses. They mitigate by adding optimistic locking and a 'lastUpdated' timestamp for conflict resolution.

An IoT platform tracks device states (on/off, temperature) using event sourcing. Each device publishes events to Kinesis Data Streams. A Lambda function stores the events in DynamoDB (event store) and updates a read-optimized table with the latest state per device. The read table is queried by a dashboard. They use Kinesis because of high throughput (100,000 events/second). A misconfiguration is forgetting to set the retention period on Kinesis, causing data loss after 24 hours. They set retention to 7 days for replay capability.