DynamoDB Schema Pattern: Event Sourcing

Event sourcing flips the usual database model: instead of storing the current state of things, you store the sequence of changes that produced that state. To answer “what’s the state of order #1234?” you replay every event for order #1234 from the beginning. The current state is derived; the events are the truth.

This is more work than just storing the current state, but you get things you can’t easily get otherwise: a complete audit log for free, the ability to reconstruct any historical state, the ability to project the same events into many different read models, and a clear contract for how state changes (every change is an explicit event).

DynamoDB is unusually well-suited as an event store. Append-only log with strong write guarantees, partition-per-aggregate isolation, optimistic concurrency via conditional writes, and Streams as a built-in projection mechanism. This pattern is the schema.

The shape of the model

Three entities:

• Event: immutable record of one change. eventType, payload, seqNum, occurredAt. Append-only.
• Snapshot: serialized state of the aggregate at a known sequence number. Optional; speeds up replay for long-lived aggregates.
• AggregateMeta: the current sequence number for an aggregate. Used for optimistic concurrency on append.

All three share a partition key: AGGREGATE#<type>#<id>. Reading “everything about an aggregate” is a single partition query.

Access patterns

Seven access patterns, three entities, one GSI.

Padded sequence numbers

Sequence numbers are integers (1, 2, 3…). DynamoDB sort keys are strings; lex sort doesn’t match numeric sort for variable-length numbers. Same problem as the gaming leaderboard pattern.

Solution: pad to fixed width. 12 digits handles up to a trillion events per aggregate, which is plenty:

seq 1     → "EVENT#000000000001"
seq 42    → "EVENT#000000000042"
seq 99999 → "EVENT#000000099999"

Lex sort = numeric sort. Replay queries with Query(pk=AGGREGATE#order#1234, sk begins_with EVENT#) return events in order.

Optimistic concurrency on append

The atomic append is the operation event sourcing has to get right. Two writers trying to append concurrently must not both succeed at the same sequence number, and they must not skip a sequence number. The pattern:

1. Read AggregateMeta to get the current currentSeqNum.
2. Submit a TransactWriteItems that:
   • Increments currentSeqNum to the next value, conditional on it still being the value you read.
   • Writes the new event with that sequence number.

If a concurrent writer got there first, the conditional check fails and the whole transaction is cancelled. Caller catches the cancellation, re-reads, and retries.

async function appendEvent(
  aggregateType: string,
  aggregateId: string,
  eventType: string,
  payload: any,
  expectedSeqNum: number,
) {
  const newSeqNum = expectedSeqNum + 1
  try {
    await client.send(new TransactWriteItemsCommand({
      TransactItems: [
        {
          Update: {
            TableName,
            Key: {
              pk: `AGGREGATE#${aggregateType}#${aggregateId}`,
              sk: "#METADATA",
            },
            UpdateExpression: "SET currentSeqNum = :new",
            ConditionExpression: "currentSeqNum = :expected OR attribute_not_exists(pk)",
            ExpressionAttributeValues: {
              ":new": newSeqNum,
              ":expected": expectedSeqNum,
            },
          },
        },
        {
          Put: {
            TableName,
            Item: marshall({
              pk: `AGGREGATE#${aggregateType}#${aggregateId}`,
              sk: `EVENT#${padSeq(newSeqNum)}`,
              aggregateType, aggregateId,
              seqNum: newSeqNum,
              eventType,
              eventId: ulid(),
              payload,
              occurredAt: new Date().toISOString(),
              gsi1pk: `EVENT_TYPE#${eventType}`,
              gsi1sk: `EVENT#${ulid()}`,
            }),
            ConditionExpression: "attribute_not_exists(pk)",
          },
        },
      ],
    }))
    return { seqNum: newSeqNum }
  } catch (err) {
    if (err.name === "TransactionCanceledException") {
      throw new ConcurrencyError(
        `Aggregate ${aggregateId} was modified by another writer; retry`
      )
    }
    throw err
  }
}

This is the canonical use case for transactions vs conditional writes. You can’t do this safely with a non-transactional approach.

Replay

To get the current state of an aggregate:

1. Load the latest Snapshot (if it exists).
2. Query events with seqNum > snapshot.lastSeqNum.
3. Apply events in order to the snapshot’s state.

async function loadAggregate(type: string, id: string) {
  const snapshot = await SnapshotEntity.get({ aggregateType: type, aggregateId: id }).go()
  let state = snapshot.data?.state ?? initialState(type)
  let fromSeq = snapshot.data?.lastSeqNum ?? 0

  const events = await EventEntity.query
    .primary({ aggregateType: type, aggregateId: id })
    .gt({ paddedSeq: padSeq(fromSeq) })
    .go({ pages: "all" })

  for (const event of events.data) {
    state = applyEvent(state, event)
  }
  return state
}

For aggregates with thousands of events, snapshots make replay fast. Take a snapshot every N events (a Lambda triggered off the Stream is a clean way to do this).

Sample data

A single partition contains the whole story of order #1234: metadata, every event, latest snapshot. Loading the order is one partition query.

ElectroDB entity definitions

export const AggregateMetaEntity = new Entity({
  model: { entity: "aggregateMeta", version: "1", service: "events" },
  attributes: {
    aggregateType: { type: "string", required: true },
    aggregateId:   { type: "string", required: true },
    currentSeqNum: { type: "number", required: true, default: 0 },
    createdAt: {
      type: "string", required: true,
      default: () => new Date().toISOString(), readOnly: true,
    },
    updatedAt: {
      type: "string", required: true,
      default: () => new Date().toISOString(),
      set: () => new Date().toISOString(), watch: "*",
    },
  },
  indexes: {
    primary: {
      pk: { field: "pk", composite: ["aggregateType", "aggregateId"],
            template: "AGGREGATE#${aggregateType}#${aggregateId}" },
      sk: { field: "sk", composite: [], template: "#METADATA" },
    },
  },
}, { client, table });

const padSeq = (n: number) => String(n).padStart(12, "0")

export const EventEntity = new Entity({
  model: { entity: "event", version: "1", service: "events" },
  attributes: {
    aggregateType: { type: "string", required: true },
    aggregateId:   { type: "string", required: true },
    seqNum:        { type: "number", required: true },
    paddedSeq:     {
      type: "string", required: true,
      watch: ["seqNum"], set: (_, { seqNum }) => padSeq(seqNum),
    },
    eventType:     { type: "string", required: true },
    eventId:       { type: "string", required: true, default: () => ulid() },
    payload:       { type: "any", required: true },
    actor:         { type: "string" },
    occurredAt: {
      type: "string", required: true,
      default: () => new Date().toISOString(), readOnly: true,
    },
  },
  indexes: {
    primary: {
      pk: { field: "pk", composite: ["aggregateType", "aggregateId"],
            template: "AGGREGATE#${aggregateType}#${aggregateId}" },
      sk: { field: "sk", composite: ["paddedSeq"],
            template: "EVENT#${paddedSeq}" },
    },
    byEventType: {
      index: "GSI1",
      pk: { field: "gsi1pk", composite: ["eventType"],
            template: "EVENT_TYPE#${eventType}" },
      sk: { field: "gsi1sk", composite: ["eventId"],
            template: "EVENT#${eventId}" },
    },
  },
}, { client, table });

export const SnapshotEntity = new Entity({
  model: { entity: "snapshot", version: "1", service: "events" },
  attributes: {
    aggregateType: { type: "string", required: true },
    aggregateId:   { type: "string", required: true },
    state:         { type: "any", required: true },     // serialized aggregate state
    lastSeqNum:    { type: "number", required: true },  // latest event included in snapshot
    takenAt: {
      type: "string", required: true,
      default: () => new Date().toISOString(), readOnly: true,
    },
  },
  indexes: {
    primary: {
      pk: { field: "pk", composite: ["aggregateType", "aggregateId"],
            template: "AGGREGATE#${aggregateType}#${aggregateId}" },
      sk: { field: "sk", composite: [], template: "#SNAPSHOT" },
    },
  },
}, { client, table });

Projections via Streams

The point of event sourcing isn’t just the event log. It’s the ability to derive any number of read models from that log. DynamoDB Streams plus Lambda makes this a clean separation:

• Write path: Application appends events to DynamoDB.
• Read path: Stream-triggered Lambdas project events into shaped read models (other items in the same table, separate tables, OpenSearch, anything).

// Stream-triggered projection: maintain an OrderSummary read model
export async function handler(event: DynamoDBStreamEvent) {
  for (const record of event.Records) {
    if (record.eventName !== "INSERT") continue
    const item = unmarshall(record.dynamodb.NewImage)
    if (!item.sk?.startsWith("EVENT#")) continue

    switch (item.eventType) {
      case "OrderPlaced":
        await OrderSummary.create({
          orderId: item.aggregateId,
          status: "placed",
          total: item.payload.total,
          placedAt: item.occurredAt,
        }).go()
        break
      case "OrderShipped":
        await OrderSummary.update({ orderId: item.aggregateId })
          .set({ status: "shipped", shippedAt: item.occurredAt })
          .go()
        break
      // ...
    }
  }
}

The projection is just a function from event stream to read shape. Want a new read model? Write a new projection. Want to rebuild a broken projection? Replay the event log from the beginning. The flexibility is the payoff.

DynamoDB Streams: fan-out, CDC, projections covers the streaming side in detail.

Why this design

Padded sequence numbers, not ULIDs, go in the event SK because optimistic concurrency requires them. “Version 5 of this aggregate” is meaningful in a way a ULID can’t enforce. Use ULIDs for the global event ID (in the GSI) so events are still uniquely addressable across aggregates.

The snapshot lives at a fixed #SNAPSHOT key rather than SNAPSHOT#<seq>. Only the latest snapshot matters for replay, so overwriting it in place keeps the partition lean and GetItem simple. If you need historical snapshots (rare), switch to SNAPSHOT#<seq>.

Snapshots are computed async by a Stream-triggered Lambda every N events. The write path doesn’t pay for snapshotting; the read path benefits from cheaper replays.

The GSI indexes event type, not aggregate type. Cross-aggregate queries are almost always “all events of this type” (audit reports, integration triggers, analytics), not “all aggregates of this type.” If you need the latter, the AggregateMeta entity can pick up a sparse GSI on aggregateType.

Events are immutable. If business logic needs to “undo” one, that’s a new compensating event (“OrderRefunded”), not a delete.

Costs and trade-offs

Event sourcing isn’t free.

Storage grows forever. You’re keeping every event indefinitely. You can use TTL for events older than your retention window, or archive cold partitions to S3 via Streams. But if you genuinely need full history (audit, compliance), accept the storage cost.

Replay gets expensive for long-lived aggregates. 10,000 events means querying up to 50MB of data across multiple paginated requests. Snapshots cap this at “events since last snapshot.” Without them, long-lived aggregates become slow.

Schema evolution is harder than it looks. Adding a field to an event payload is fine. Renaming or removing a field requires either upcasting old events at read time or migrating the event log, which is expensive and often impossible.

Read models are eventually consistent. Projections lag the write path by Stream latency (seconds). Anything that needs read-your-writes within a request has to query the event log directly, not the projection.

There are also more moving parts: Stream triggers, Lambdas, projections, replay logic. The operational burden is higher than a CRUD-style schema.

For most CRUD apps, event sourcing is overkill. For audit-critical, compliance-heavy, or domain-rich systems where the history is itself the product, it pays back many times over.

What this schema doesn’t support

Design this visually → coming soon

Drag an Event entity onto a canvas and connect it to a Snapshot. See the partition collapse the whole aggregate’s history into one cheap query. That’s what I’m building at singletable.dev.

Join the waitlist →

Pattern #9 of 10 in the SingleTable pattern library. Event sourcing pairs naturally with DynamoDB Streams for projections, transactions for the optimistic-concurrency append, and idempotency keys for safe event reception from external systems.