aws-dynamodb-es-cqrs · ES/CQRS using AWS DynamoDB

• Events

  • Events are described by AggregateEvent (ULID-based eventId, aggregateName, aggregateId, aggregateVersion, actorId).
  • Aggregate metadata:
    • AggregateMeta with id, version, actorId, optional updatedAt.

• Reducing events to aggregates (Event Sourcing)

  • Generic reducer: reduceEvents applies a sequence of AggregateEvent to produce an aggregate state.
  • Domain reducer example (BlogPost): blogPostReducer uses type guards (isNamedEvent) and assertions (assertAggregateEvent) to evolve state from events.

• Commands (write side)

  • Commands validate current state and produce new events:
    • Create: createBlogPostCommand
    • Publish: publishBlogPostCommand
    • Update title: updateBlogPostTitleCommand
  • Concurrency and existence are enforced using:
    • ConflictError and NotFoundError
    • Version checks against the loaded aggregate before emitting the next event.

• Persistence (event store + aggregate projection)

  • DynamoDB persistence is transactional to keep aggregate and event store in sync:
    • persistDynamoDB performs a TransactWrite that:
      • Upserts the aggregate row with optimistic concurrency:
        • New aggregate: attribute_not_exists(aggregateId) (version 1).
        • Update: #version = :prevVersion and sets updatedAt.
      • Writes the event into the events table.
    • Aggregate unmarshalling for reads: unmarshallAggregate.
  • Event queries:
    • List an aggregate’s events: listAggregateEventsDynamoDB
    • Parse DynamoDB Streams records back into events: extractEventsFromDynamoDBEvent

• Queries (read side)

  • Read the current aggregate projection:
    • Example (BlogPost):
      • findBlogPostByIdDynamoDB.
      • listBlogPostsDynamoDB.

Setup

Install the dependencies:

npm ci

Tests

The test folder contains an example aggregate (a blog post), and operations that manipulate it, which demonstrate how the key components of this implementation work with separate read/write components demonstrating create, read aggregate, list events, and optimistic concurrency.

Install DynamoDB (local)

wget https://s3.eu-central-1.amazonaws.com/dynamodb-local-frankfurt/dynamodb_local_latest.zip
unzip dynamodb_local_latest.zip -d ./dynamodb_local_latest

Start it in background:

java -Djava.library.path=./dynamodb_local_latest/DynamoDBLocal_lib -jar ./dynamodb_local_latest/DynamoDBLocal.jar -sharedDb -inMemory &

Run the tests:

npm test