Example: Event Sourcing

An event-sourced cricket match tracker using effect-dynamodb’s EventStore module. Events are appended to a DynamoDB stream, state is reconstructed by folding over the event history, and commands are validated against current state before producing new events.

What you’ll learn:

• Defining event streams with EventStore.makeStream
• The decider pattern: commands, events, and state evolution
• Command handlers with optimistic concurrency
• Reading events (read, readFrom, currentVersion)
• Reconstructing state with EventStore.fold
• Query combinators on event streams (reverse, limit)
• Snapshots, atomic side writes and exactly-once commands
• Consistent reads and If-Match expected versions
• Inline projections derived from the decision
• Inline snapshots and single-request state loads (readLatest)
• Large commands as stepped commands, each one atomic append
• Stream indexes: sub-streams ordered by a key derived from each event
• Event metadata: one value per append, or one entry per event
• Stream index projection: keeping metadata, or everything but the keys, out of an index
• Shared indexes: one GSI partition several streams write, snapshots included
• State from your own record: loadState and a record head proof
• Sealing a stream: a generation changeover in one transaction, and atomic multi-stream appends

---

Step 1: Events

Events are pure domain Schema.TaggedClass definitions. Each event captures something that happened in the domain — no DynamoDB concepts:

events.ts

class MatchStarted extends Schema.TaggedClass<MatchStarted>()("MatchStarted", {
  venue: Schema.String,
  homeTeam: Schema.String,
  awayTeam: Schema.String,
}) {}

class InningsCompleted extends Schema.TaggedClass<InningsCompleted>()("InningsCompleted", {
  innings: Schema.Number,
  runs: Schema.Number,
  wickets: Schema.Number,
}) {}

class MatchEnded extends Schema.TaggedClass<MatchEnded>()("MatchEnded", {
  result: Schema.String,
}) {}

type MatchEvent = MatchStarted | InningsCompleted | MatchEnded

Each tag doubles as the event type discriminator stored in DynamoDB.

Use Schema.TaggedClass for event schemas

Define events with Schema.TaggedClass, not plain Schema.Class. The event store decodes stored events through a Schema.Union of your event schemas, and a plain Schema.Class declares no _tag field — the stored _tag is treated as an ignored excess property, so the union discriminates structurally by field shape. Two event types with identical fields (say, PaymentAuthorized and PaymentCaptured, both { amount: Schema.Number }) would then mis-decode as each other: the envelope’s eventType attribute stays correct, but data decodes to whichever class matches first. Schema.TaggedClass adds the _tag literal to the schema itself, so the union discriminates on the tag and every event round-trips to the correct class.

---

Step 2: Schema and Table

Define the application schema and a table for the event store. The stream itself manages the event items — no entity registration is needed for them. Entities are only registered here so that ordinary records can be written atomically alongside events (see Step 12):

infrastructure.ts

const AppSchema = DynamoSchema.make({ name: "cricket", version: 1 })

// A per-writer ingestion watermark — a side record updated atomically with
// events via `append({ additionalItems })`.
class Watermark extends Schema.Class<Watermark>("Watermark")({
  writerId: Schema.String,
  lastSeq: Schema.Number,
}) {}

const Watermarks = Entity.make({
  model: Watermark,
  entityType: "Watermark",
  primaryKey: {
    pk: { field: "pk", composite: ["writerId"] },
    sk: { field: "sk", composite: [] },
  },
})

// A read model kept in step with the stream. Authored with the pure, AWS-free
// `@effect-dynamodb/schema` package — it carries no CRUD ops, so its writes are
// always built from the bound client (`db.entities.MatchStatus.put(...)`).
const MatchStatusRecord = Schema.Struct({
  matchId: Schema.String,
  status: Schema.String,
})

const MatchStatus = PureEntity.make({
  model: DynamoModel.configure(MatchStatusRecord, { matchId: { identifier: true } }),
  entityType: "MatchStatus",
  primaryKey: {
    pk: { field: "pk", composite: ["matchId"] },
    sk: { field: "sk", composite: [] },
  },
})

const EventsTable = Table.make({ schema: AppSchema, entities: { Watermarks, MatchStatus } })

---

Step 3: Event Stream

An event stream binds events to a table, names the stream, and declares which attributes compose the stream ID (analogous to an aggregate ID):

stream.ts

const MatchEvents = EventStore.makeStream({
  table: EventsTable,
  streamName: "Match",
  events: [MatchStarted, InningsCompleted, MatchEnded],
  streamId: { composite: ["matchId"] },
})

Under the hood, each event is stored as a DynamoDB item with:

• PK composed from the stream ID (matchId)
• SK containing a zero-padded version number for ordered retrieval
• eventType discriminator for decoding back to the correct Schema.Class

Stream key casing

By default a stream lower-cases its name in every key it writes, whatever the schema’s casing, and the rest of the key (stream ids, command ids) follows the schema. Entity types follow the schema’s casing, so under casing: "preserve" a stream and an entity on the same table disagree:

$cricket#v1#Watermark#writerId_Ingest-1      ← entity: label follows the schema
$cricket#v1#match#M-1                        ← stream: name always lower-case

Set casing on makeStream to choose the stream’s key casing, the same way an index’s casing works. It applies to the stream name, stream ids and command ids in the partition key and the event, snapshot and idempotency-sentinel sort keys; the $<schema>#v<n> prefix keeps the schema’s casing:

$cricket#v1#Match#M-1                        ← stream with casing: "preserve"

On a "lowercase" or "uppercase" schema, setting casing to the schema’s value writes the same keys as leaving it out. The __edd_e__ discriminator (match.event, …) stays lower-case either way. In the next major version, leaving casing out will mean the schema’s casing, as it does for indexes.

Setting casing on an existing stream

If the keys it produces differ from the ones the stream was written with — for example casing: "preserve" on a "preserve" schema with a streamName containing capitals — existing events, snapshots and idempotency sentinels stay under the old keys and are no longer read, so the stream appears to start from version 0. Use it for new streams, or migrate the items first.

---

Step 4: Decider

The decider is the core pattern for event sourcing. It defines three things:

1. initialState — the starting state before any events
2. decide — validates a command against current state, returning events or an error
3. evolve — applies an event to state, producing the next state

decider.ts

interface MatchState {
  readonly status: "pending" | "in-progress" | "completed"
  readonly venue?: string
  readonly innings: ReadonlyArray<{ runs: number; wickets: number }>
  readonly result?: string
}

type MatchCommand =
  | {
      readonly _tag: "StartMatch"
      readonly venue: string
      readonly homeTeam: string
      readonly awayTeam: string
    }
  | {
      readonly _tag: "CompleteInnings"
      readonly innings: number
      readonly runs: number
      readonly wickets: number
    }
  | { readonly _tag: "EndMatch"; readonly result: string }

class AlreadyStarted extends Data.TaggedError("AlreadyStarted") {}
class NotStarted extends Data.TaggedError("NotStarted") {}
class AlreadyEnded extends Data.TaggedError("AlreadyEnded") {}

const matchDecider: EventStore.Decider<
  MatchState,
  MatchCommand,
  MatchEvent,
  AlreadyStarted | NotStarted | AlreadyEnded
> = {
  initialState: { status: "pending", innings: [] },

  decide: (command, state) =>
    Effect.gen(function* () {
      if (command._tag === "StartMatch") {
        if (state.status !== "pending") return yield* new AlreadyStarted()
        return [
          new MatchStarted({
            venue: command.venue,
            homeTeam: command.homeTeam,
            awayTeam: command.awayTeam,
          }),
        ]
      }
      if (command._tag === "CompleteInnings") {
        if (state.status !== "in-progress") return yield* new NotStarted()
        return [
          new InningsCompleted({
            innings: command.innings,
            runs: command.runs,
            wickets: command.wickets,
          }),
        ]
      }
      if (command._tag === "EndMatch") {
        if (state.status === "completed") return yield* new AlreadyEnded()
        if (state.status !== "in-progress") return yield* new NotStarted()
        return [new MatchEnded({ result: command.result })]
      }
      return []
    }),

  evolve: (state, event) => {
    if (event instanceof MatchStarted) {
      return { ...state, status: "in-progress" as const, venue: event.venue }
    }
    if (event instanceof InningsCompleted) {
      return {
        ...state,
        innings: [...state.innings, { runs: event.runs, wickets: event.wickets }],
      }
    }
    if (event instanceof MatchEnded) {
      return { ...state, status: "completed" as const, result: event.result }
    }
    return state
  },
}

The decide function is effectful — it can fail with domain errors. The evolve function is pure — it simply transforms state. This separation keeps business rules in decide and state transitions in evolve.

---

Step 5: Command Handler

EventStore.commandHandler wires the decider to an event stream. It handles the read-decide-append cycle with optimistic concurrency:

handler.ts

const matchEvents = yield* EventStore.bind(MatchEvents)
const handleMatch = EventStore.commandHandler(matchDecider, matchEvents)

Each call to handleMatch:

1. Reads all events for the stream ID, with a strongly consistent read (see Step 14)
2. Folds them through evolve to reconstruct current state
3. Runs decide with the command and current state
4. Folds the new events into that state
5. Appends the new events with an expected version check

The handler returns { state, version, events }, where state is the fold from step 4.

Append limits and version contiguity

append writes every event in a single TransactWriteItems request, so an append is all-or-nothing — either every event in the batch is persisted or none is. An append whose request holds one item, its one event (the first event of a stream, or one event with headProof: "none", and no sentinel, snapshot or additional item), is sent as a plain conditional PutItem instead: the same item and condition, committed just as atomically, at 1 WCU per KB instead of a transaction’s 2. Its failures are the same errors. That atomicity is bounded by DynamoDB’s transaction limits:

• 100 items per transaction. Each event is one transact item, and when expectedVersion > 0 the version-contiguity ConditionCheck (below) occupies one more slot unless a head proof replaces it, as do any additionalItems, an idempotency sentinel and an inline snapshot. An append over the limit is rejected by DynamoDB and nothing is written; the library recognises the rejection, counts the items it sent, and fails with a typed AppendTooLarge error carrying that count, with DynamoDB’s error as its cause. In practice a single append holds at most 100 events when expectedVersion is 0, and at most 99 otherwise. The limit is exported as the TRANSACT_WRITE_ITEMS_LIMIT constant.
• 4 MB total payload per transaction, 400 KB per item. An append whose items pass 4 MB, or with any one item over 400 KB, is rejected by DynamoDB and nothing is written. The append fails with a ValidationError that names the culprit: the item (the event at version 2, the inline snapshot, the idempotency sentinel or operation 0 (User) among additionalItems) and its key, or for 4 MB the largest items. DynamoDB’s own error is kept as the cause. Keep event payloads small; large blobs belong in S3 with a reference in the event.

An append is never split across transactions, because splitting breaks its atomicity. If a command legitimately decides more than one transaction holds, split it into stepped commands: one command, one decision, one atomic append.

append also enforces that the appended range is contiguous with the stream head. Every event Put is conditioned on attribute_not_exists(pk), which rejects stale expected versions (the target version slot already exists). In addition, when expectedVersion > 0 the transaction carries a ConditionCheck requiring the event at exactly expectedVersion to exist — so an ahead expected version (say expectedVersion: 10 on a stream whose head is at version 3) fails with VersionConflict instead of silently writing version 11 and leaving a permanent hole in the sequence. Both failure modes surface as the same VersionConflict error.

The check reads the event at expectedVersion, and DynamoDB bills a ConditionCheck in a transaction at 2 WCU per KB of the item it checks, so with large events it is a real share of every append. When the transaction already holds an item that proves the head (the inline snapshot, or a state record of your own), headProof names that item and the check is left out: see Proving the head with the snapshot and Step 23. A caller that read its expectedVersion from the stream can leave the check out with no proof item at all, by taking over its duties: see headProof: "none".

---

Step 6: Running Commands

Start Match

const r1 = yield* handleMatch(
  { matchId: "m-1" },
  { _tag: "StartMatch", venue: "MCG", homeTeam: "AUS", awayTeam: "ENG" },
)
// State: in-progress, Version: 1, Events: 1

The return value includes the new state, the version after append, and the events that were produced.

Complete Innings

const r2 = yield* handleMatch(
  { matchId: "m-1" },
  { _tag: "CompleteInnings", innings: 1, runs: 250, wickets: 10 },
)

const r3 = yield* handleMatch(
  { matchId: "m-1" },
  { _tag: "CompleteInnings", innings: 2, runs: 180, wickets: 10 },
)
// State: in-progress, Innings: 2, Version: 3

End Match

const r4 = yield* handleMatch(
  { matchId: "m-1" },
  { _tag: "EndMatch", result: "AUS won by 70 runs" },
)
// State: completed, Result: AUS won by 70 runs, Version: 4

---

Step 7: Reading Events

Read All Events

const allEvents = yield* matchEvents.read({ matchId: "m-1" })
// v1: MatchStarted at 2026-03-15T...
// v2: InningsCompleted at 2026-03-15T...
// v3: InningsCompleted at 2026-03-15T...
// v4: MatchEnded at 2026-03-15T...

Each event envelope includes version, eventType, timestamp, and the decoded event payload.

Read From a Specific Version

Useful for catching up from a known checkpoint:

const laterEvents = yield* matchEvents.readFrom({ matchId: "m-1" }, 2)
// v2: InningsCompleted
// v3: InningsCompleted
// v4: MatchEnded

Current Version

const version = yield* matchEvents.currentVersion({ matchId: "m-1" })
// Current version: 4

---

Step 8: Fold and Query

Reconstruct State from Events

EventStore.fold applies the decider’s evolve function over a list of events, starting from initialState:

const state = EventStore.fold(matchDecider, allEvents)
// Reconstructed: status=completed, innings=2

This is a pure function — no DynamoDB calls. It takes the events you already have and replays them through evolve.

A note on “projections”

In event-sourcing jargon, a projection is a read model built by consuming events. This library’s Projection module is something else entirely: it builds DynamoDB ProjectionExpressions for selecting specific item attributes on reads. Event-sourcing read models are outside this library’s scope — build them in a DynamoDB Streams consumer that reacts to appended events.

Query Combinators

Event streams expose a query API with the same combinators as entity queries. For example, get the latest event using reverse and limit:

const latest = yield* matchEvents.provide(
  matchEvents.query.events({ matchId: "m-1" }).pipe(Query.reverse, Query.limit(1), Query.collect),
)
const [latestEvent] = latest
// Latest: v4 MatchEnded

---

Step 9: Domain Error Handling

The decider enforces business rules. Sending an invalid command produces a tagged error:

const error = yield* handleMatch(
  { matchId: "m-1" },
  { _tag: "StartMatch", venue: "SCG", homeTeam: "AUS", awayTeam: "IND" },
).pipe(Effect.flip)
// Error: AlreadyStarted

Because errors use Data.TaggedError, you can handle specific cases with Effect.catchTag:

yield* handleMatch(streamId, command)
  .pipe(
    Effect.catchTag("AlreadyStarted", () => ...),
    Effect.catchTag("NotStarted", () => ...),
    Effect.catchTag("AlreadyEnded", () => ...),
  )

---

Step 10: Snapshots

Every command so far replayed the entire stream before deciding. That is fine for short-lived streams and a scaling cliff for long-lived ones. A snapshot is a cached fold of the stream up to some version: the handler reads it, replays only the events after it, and decides from there.

Snapshots are opt-in. Declare a schema for your state — the snapshot round-trips through it, so branded types, DateTime fields, and other transforming schemas all work:

state.ts

const MatchStateSchema = Schema.Struct({
  status: Schema.Literals(["pending", "in-progress", "completed"]),
  venue: Schema.optionalKey(Schema.String),
  innings: Schema.Array(Schema.Struct({ runs: Schema.Number, wickets: Schema.Number })),
  result: Schema.optionalKey(Schema.String),
})

Then pass it to makeStream. every: N asks the command handler to write a fresh snapshot after a successful append, once at least N events have accumulated since the last one. This is the default mode: "after-append"; Step 17 covers mode: "inline", which writes the snapshot in the append’s own transaction:

stream.ts

const SnapshotMatchEvents = EventStore.makeStream({
  table: EventsTable,
  streamName: "SnapshotMatch",
  events: [MatchStarted, InningsCompleted, MatchEnded],
  streamId: { composite: ["matchId"] },
  snapshot: { schema: MatchStateSchema, every: 3 },
})

The snapshot is stored as a single item in the stream’s own partition, under a sort key that can never collide with an event sort key ($cricket#v1#snapshotmatch.snapshot). It is invisible to read, readFrom, currentVersion, and query.events, and it is overwritten in place — snapshots are a cache, never history. The event stream stays the source of truth, so a snapshot can always be deleted and rebuilt.

Caution

A snapshot that no longer decodes through its state schema fails the command with ValidationError — it is not silently discarded in favour of a full replay, because that would hide state-schema evolution bugs. When evolving the state schema, either keep it backward-tolerant (optional fields with defaults) or delete the snapshots on deploy.

---

Step 11: Retrying Version Conflicts

Two concurrent commands on the same stream will race: one wins the optimistic-concurrency check and the other fails with VersionConflict. The correct recovery is not to re-append the events you already decided — those were decided against stale state. It is to re-read, re-decide, and re-append.

That is exactly what the retry option does. Pass a max-attempts number or an Effect Schedule; only VersionConflict is retried, and the retried unit is the whole read-decide-append cycle:

handler.ts

const snapshotMatchEvents = yield* EventStore.bind(SnapshotMatchEvents)
const handleSnapshotMatch = EventStore.commandHandler(matchDecider, snapshotMatchEvents, {
  retry: 3,
})

// A Schedule works too — exponential backoff, capped at 5 retries:
EventStore.commandHandler(matchDecider, snapshotMatchEvents, {
  retry: Schedule.exponential("50 millis").pipe(Schedule.compose(Schedule.recurs(5))),
})

Domain errors from your decider and infrastructure errors are never retried — they fail immediately. The default is no retry.

Commands run exactly as before. The third one crosses the every: 3 threshold, so a snapshot is written after the append:

yield* handleSnapshotMatch(
  { matchId: "m-2" },
  { _tag: "StartMatch", venue: "SCG", homeTeam: "AUS", awayTeam: "IND" },
)
yield* handleSnapshotMatch(
  { matchId: "m-2" },
  { _tag: "CompleteInnings", innings: 1, runs: 310, wickets: 8 },
)
// The third event crosses the `every: 3` threshold — a snapshot is written.
const s3 = yield* handleSnapshotMatch(
  { matchId: "m-2" },
  { _tag: "CompleteInnings", innings: 2, runs: 275, wickets: 10 },
)

Reading a snapshot

const snapshot = yield* snapshotMatchEvents.readSnapshot({ matchId: "m-2" })
const asOfVersion = Option.match(snapshot, {
  onNone: () => 0,
  onSome: (s) => s.asOfVersion,
})
// → asOfVersion: 3

readSnapshot returns Option<Snapshot<State>>, where Snapshot carries the decoded state, the asOfVersion it reflects, and a timestamp.

The next command folds from that snapshot plus the single event after it, rather than replaying all four — and produces exactly the same state. The handler loads both in one Query (readLatest):

const s4 = yield* handleSnapshotMatch(
  { matchId: "m-2" },
  { _tag: "EndMatch", result: "AUS won by 35 runs" },
)

Writing a snapshot by hand

Omit every to keep the cadence under your own control — for example from a backfill job — and call writeSnapshot yourself:

const events = yield* snapshotMatchEvents.read({ matchId: "m-2" })
const folded = EventStore.fold(matchDecider, events)
yield* snapshotMatchEvents.writeSnapshot({ matchId: "m-2" }, folded, s4.version)

Snapshot writes are monotonic: if a newer snapshot already exists, the write is a successful no-op rather than a regression of the cache. Auto-snapshot writes are best-effort — a failure is logged and never fails the command, because the events it summarises are already durable.

writeSnapshot does not know whether a stream is sealed, so it must not target a sealed head with an unmarked snapshot. Step 24 explains why.

Step 12: Atomic Side Writes

Real event-sourced systems usually need one non-event record kept in step with the events: an ingestion watermark, a read model, a stream registry row, a uniqueness guard. append accepts additionalItems — the same operation builders Transaction.transactWrite takes (Entity.put, Entity.delete, and Transaction.check) — and commits them in the same TransactWriteItems call as the events:

yield* matchEvents.append(
  { matchId: "m-2" },
  [new MatchStarted({ venue: "SCG", homeTeam: "AUS", awayTeam: "IND" })],
  0,
  {
    additionalItems: [Watermarks.put({ writerId: "ingest-1", lastSeq: 4021 })],
  },
)

Either everything lands or nothing does. The event puts, your items, and the version guard are one atomic unit.

Bound-client builders work too

The bound builders returned by db.entities.* (put, create, delete, deleteIfExists) are accepted anywhere an Entity.put / Entity.delete is. That matters most for entities authored with the pure, AWS-free @effect-dynamodb/schema package: a pure EntityDefinition carries no operations, so the bound builder is the only write descriptor its author can hold — and it is exactly what you need to commit a read model atomically with the events that produced it:

const db = yield* DynamoClient.make({ entities: { MatchStatus }, tables: { EventsTable } })

yield* matchEvents.append(
  { matchId: "m-4" },
  [new MatchStarted({ venue: "Basin Reserve", homeTeam: "NZL", awayTeam: "SAF" })],
  0,
  {
    additionalItems: [db.entities.MatchStatus.put({ matchId: "m-4", status: "in-progress" })],
  },
)

Conditions ride along: .condition({ ... }) on a bound put or delete, and the implicit attribute_not_exists guard create() carries, are compiled into the transact item. A failing one surfaces as AdditionalItemConditionFailed, not a version conflict.

upsert is not a transact item

upsert is deliberately not accepted here, and neither is update / patch (Transaction.transactWrite accepts those: it plans them from the entity’s own op, which reads first — append does not). An upsert is an UpdateItem whose SET clause uses if_not_exists for createdAt, immutable fields and the version counter — compiling it into a TransactWriteItems Put would reset all three. Rather than do that quietly, append fails with a ValidationError naming upsert. Use put() or create() in the transaction, or run the upsert as its own operation outside it.

A delete of an entity configured with unique, versioned: { retain: true } or softDelete is refused for a related reason (EDD-9048): the extra items such a delete must write are derived from the stored row, and this path builds a delete from its key alone. Use Transaction.transactWrite or the entity’s own delete.

One additional item can become several

A put of a versioned or unique-constrained entity is written exactly as the entity’s own put writes it. It reads the item, and commits in the same transaction as the events: the row (continuing an existing item’s version, incarnation and createdAt), a guarded sentinel for each new unique value and a release of each old one the item owns, and with retain a version snapshot. So a versioned read model stays in step across appends: the second append that puts it replaces the row as version 2. A concurrent write to the row between that read and the append cancels the transaction, and append builds and writes it again. Four things follow.

AdditionalItemConditionFailed is reported only for a condition you set (.condition(), create(), Transaction.check). A put you set no condition on fails as the entity’s put would: UniqueConstraintViolation for a taken value, a ValidationError for conflicting history, or OptimisticLockError / ConcurrentModification for a race lost on every attempt.

AdditionalItemConditionFailed.indices still means indices into the array you passed. A failing item is reported against the op that produced it, not against the raw transaction position, and one op that fails in several places is reported once.

The 100-item cap counts the expanded total, so events + expanded additional items + sentinel + contiguity check must fit. AppendTooLarge.count reports the expanded number.

DynamoDB allows one operation per item in a transaction. Two additionalItems that touch the same item, such as two puts that swap unique values (each releases the sentinel the other reserves), or one that repeats an event or the idempotency sentinel, are rejected by DynamoDB and fail with a ValidationError naming both items; nothing is written, and the rejection is never read as a version conflict or retried. An append whose items pass DynamoDB’s 4 MB limit, or that holds an item over its 400 KB limit, is rejected by DynamoDB, so nothing is written, and fails with a ValidationError naming the culprit item; a retain put counts twice towards the 4 MB (its item and its snapshot).

Telling the failures apart

Because your items carry their own conditions, “the transaction was cancelled” is no longer a single story. append maps cancellation reasons by position, so you always know whose condition failed:

const condError = yield* matchEvents
  .append({ matchId: "m-2" }, [new InningsCompleted({ innings: 1, runs: 300, wickets: 8 })], 1, {
    additionalItems: [
      Transaction.check(
        Watermarks.get({ writerId: "ingest-1" }),
        Expression.condition({ lt: { lastSeq: 100 } }),
      ),
    ],
  })
  .pipe(Effect.flip)
// Error: AdditionalItemConditionFailed

Error	What failed	What to do
VersionConflict	An event put’s guard — the stream advanced under you	Re-read, re-decide, retry
AdditionalItemConditionFailed	One of your additionalItems conditions	Inspect indices (0-based into additionalItems); retrying blindly will not help
UniqueConstraintViolation	A put in additionalItems takes a unique value another item holds	Choose another value
OptimisticLockError / ConcurrentModification	A put in additionalItems lost a race with another writer of its item on every attempt	Retry the command
DuplicateCommand	The idempotency sentinel — this commandId already ran	Nothing; the command already took effect
StreamSealed	The stream is sealed: no event can follow its final one (Step 24)	Nothing on this stream; move to its successor. Never retried
TransactionCancelled	Throttling, a concurrent transaction, or a validation error	Retry with backoff
AppendTooLarge	DynamoDB rejected the append: events + additionalItems + sentinel + contiguity check exceeded its 100-item cap	Split the append

Without position-aware mapping, a failed watermark condition would be reported as a VersionConflict — and the caller’s natural response (re-read, re-decide, retry) would loop forever against a condition that can never pass.

---

Step 13: Command Idempotency

By default, command processing is at-least-once. If a caller’s response is lost after the append committed, the retry re-reads the stream, re-runs decide, and appends again. Deciders that are self-guarding (StartMatch fails once the match is started) absorb this; deciders that aren’t (CompleteInnings happily emits another event) do not.

To get exactly-once processing, configure idempotency on the handler and pass a commandId per call. The handler writes a dedup sentinel guarded by attribute_not_exists into the same transaction as the events:

const handleIdempotent = EventStore.commandHandler(matchDecider, matchEvents, {
  idempotency: { ttl: Duration.days(1) },
})

yield* handleIdempotent(
  { matchId: "m-3" },
  { _tag: "StartMatch", venue: "Lords", homeTeam: "ENG", awayTeam: "NZ" },
  { commandId: "cmd-7f3a" },
)

// CompleteInnings is not self-guarding — the decider happily produces a second
// event, so only the dedup sentinel can catch the replay.
yield* handleIdempotent(
  { matchId: "m-3" },
  { _tag: "CompleteInnings", innings: 1, runs: 210, wickets: 6 },
  { commandId: "cmd-9b12" },
)

const dupError = yield* handleIdempotent(
  { matchId: "m-3" },
  { _tag: "CompleteInnings", innings: 1, runs: 210, wickets: 6 },
  { commandId: "cmd-9b12" },
).pipe(Effect.flip)
// Error: DuplicateCommand

Configuring idempotency makes commandId required at the type level — you cannot silently fall back to at-least-once.

What to know about the sentinel:

• It lives in the stream’s own partition, so commandId uniqueness is scoped to the stream. It is invisible to read, readFrom, and currentVersion, which filter on the event entity type.
• Its sort key is composed with the schema’s casing (lowercase by default), so command ids that differ only in case collide. Use UUIDs or ULIDs.
• ttl is optional and written to the table’s TTL attribute (honouring TableConfig.ttlAttributeName). Set it to the longest window over which your infrastructure can replay a command; omit it and sentinels are permanent.
• A replay is rejected, not replayed — DuplicateCommand carries streamName, streamId, and commandId. If you need the original outcome, catch the error and re-read the stream.
• A command that produces no events writes no sentinel, because there is no transaction to attach it to.
• The sentinel is consulted only when events are appended. A redelivery whose decide returns no events against the loaded state never appends, so it succeeds as a no-op instead of failing with DuplicateCommand. Whether your application must tell a no-op redelivery apart is your call.

---

Step 14: Consistent Reads

DynamoDB reads are eventually consistent unless asked otherwise, so a read issued right after an append can miss it. read, readFrom and currentVersion take an optional { consistentRead: true }, which sets ConsistentRead on every Query page: the read then sees every append acknowledged before it started, at twice the read cost. query.events composes with Query.consistentRead() instead.

The command handler is different: it loads state with strongly consistent reads by default. An eventually consistent load can hand decide state that is missing the newest events, and the append then fails with VersionConflict (or burns a retry). Opt out per handler with consistentRead: false when the read cost matters more than the occasional conflict:

const fresh = yield* matchEvents.read({ matchId: "m-1" }, { consistentRead: true })
const head = yield* matchEvents.currentVersion({ matchId: "m-1" }, { consistentRead: true })

// commandHandler loads state consistently by default. Opt out per handler:
const handleRelaxed = EventStore.commandHandler(matchDecider, matchEvents, {
  consistentRead: false,
})
yield* handleRelaxed(
  { matchId: "m-7" },
  { _tag: "StartMatch", venue: "Gabba", homeTeam: "AUS", awayTeam: "PAK" },
)

On a snapshot-configured stream the snapshot and its delta are read in one query, so the setting covers both.

---

Step 15: If-Match — Caller Expected Versions

An HTTP API usually wants the client’s view of the stream to be the precondition: the client read the match at version 2, and its command should apply only if nothing has happened since. Pass that version per call as expectedVersion:

// The client last saw m-1 at version 2; the stream has since moved to 4.
const stale = yield* handleMatch(
  { matchId: "m-1" },
  { _tag: "EndMatch", result: "Abandoned" },
  { expectedVersion: 2 },
).pipe(Effect.flip)
const actual = stale._tag === "VersionConflict" ? stale.actualVersion : undefined
// → VersionConflict, actualVersion 4 — decide never ran, nothing was retried

// The version the client saw goes through.
const matched = yield* handleMatch(
  { matchId: "m-4" },
  { _tag: "CompleteInnings", innings: 1, runs: 220, wickets: 9 },
  { expectedVersion: 1 },
)

How it behaves:

• After state is loaded, a stream at any other version fails the call with VersionConflict before decide runs, so decide never sees state the client did not. The error carries actualVersion, the version that was loaded, which is what a 412 Precondition Failed response wants.
• Otherwise the append is conditioned on expectedVersion. A writer that slips in between the load and the append also fails it with VersionConflict, this time without actualVersion (knowing it would take another read).
• Neither conflict is retried, whatever the handler’s retry policy says. The caller asked for a conditional write, so the conflict is the answer.
• With idempotency, a mismatch first checks the command’s sentinel. A redelivered command that already committed (its response was lost, so the client sent the same commandId and If-Match again) fails with DuplicateCommand, not VersionConflict.
• A value that is not a non-negative integer (a NaN from a failed header parse, say) fails with ValidationError before anything is read.

Omit expectedVersion and the handler behaves as before.

---

Step 16: Inline Projections

Step 12 committed a read model with the events, but the row was built before the command ran. A projection usually needs the command’s outcome. On a command handler, additionalItems can also be a function of the decision:

const handleProjected = EventStore.commandHandler(matchDecider, matchEvents)

// Pure form: the read-model row comes from the post-fold state.
yield* handleProjected(
  { matchId: "m-5" },
  { _tag: "StartMatch", venue: "Eden Park", homeTeam: "NZL", awayTeam: "AUS" },
  {
    additionalItems: ({ state }) => [
      db.entities.MatchStatus.put({ matchId: "m-5", status: state.status }),
    ],
  },
)

// Effect form: a projection that reads before it writes.
yield* handleProjected(
  { matchId: "m-5" },
  { _tag: "EndMatch", result: "NZL won by 4 wickets" },
  {
    additionalItems: ({ state, version }) =>
      Effect.gen(function* () {
        const row = yield* db.entities.MatchStatus.get({ matchId: "m-5" })
        return [
          db.entities.MatchStatus.put({
            matchId: "m-5",
            status: `${row.status} -> ${state.status} (v${version + 1})`,
          }),
        ]
      }),
  },
)

The function receives a Decision:

Field	Meaning
events	What decide returned (never empty)
state	The evolve fold of previous with events: the state the handler returns and snapshots
previous	The state decide was given
version	The version the events are appended after; they take version + 1 … version + events.length

It may return the items directly or as an Effect. The effect’s error and requirements join the handler’s own, so a projection that reads through the bound client adds that read’s errors to the handler’s error channel.

To make state available, the handler folds the new events before it appends them:

load → [expectedVersion check] → decide → fold new events → derive items → append → [snapshot]

• The items commit in the same transaction as the events. They count towards the 100-item and 4 MB limits, and a failing condition on one of them is still AdditionalItemConditionFailed, with indices into the array the function returned.
• The function is not called when decide returns no events, and is called again on every retry attempt, against the freshly decided events.
• The fold is the only source of state. The state the handler returns, snapshots and passes here is always the evolve fold of the stored and new events. Nothing produced inside decide ever becomes state.
• evolve may mutate state in place. The library does not require immutable state. If your evolve mutates and returns the same object, previous and state are the same reference, and previous already reflects the new events. A projection that needs a pristine previous needs an evolve that returns new state, as this example’s does.

---

Step 17: Inline Snapshots and readLatest

The snapshots in Step 10 are written after the append, best-effort, at the every cadence, so they usually lag the stream. mode: "inline" puts the snapshot Put into the append’s own transaction instead:

stream.ts

const InlineMatchEvents = EventStore.makeStream({
  table: EventsTable,
  streamName: "InlineMatch",
  events: [MatchStarted, InningsCompleted, MatchEnded],
  streamId: { composite: ["matchId"] },
  snapshot: { schema: MatchStateSchema, mode: "inline" },
})

The handler is used exactly as before:

const inlineMatchEvents = yield* EventStore.bind(InlineMatchEvents)
const handleInline = EventStore.commandHandler(matchDecider, inlineMatchEvents)

yield* handleInline(
  { matchId: "m-6" },
  { _tag: "StartMatch", venue: "Newlands", homeTeam: "SAF", awayTeam: "IND" },
)
// The snapshot rides in each append's transaction: current after every command.
const inlined = yield* handleInline(
  { matchId: "m-6" },
  { _tag: "CompleteInnings", innings: 1, runs: 198, wickets: 10 },
)

Mode	When the snapshot is written	How
"inline", no every	Every append that writes events	In the append transaction
"inline", every: N	Once at least N events have accumulated since the loaded snapshot	In the append transaction
"after-append" (default)	At the every cadence	A best-effort writeSnapshot after the append succeeds

An inline snapshot commits if and only if the events do. It is the post-fold state from Step 16, and it takes one of the transaction’s 100 item slots. append accepts it directly too, as append(streamId, events, expectedVersion, { snapshot: state }); it needs at least one event.

Reading the latest state in one request

readLatest loads a stream’s state with a single Query: the snapshot, which sorts after every event, plus the events after it, newest first:

// One Query: the snapshot plus every event after it.
const loadedState = yield* inlineMatchEvents.readLatest(
  { matchId: "m-6" },
  { consistentRead: true },
)
const current = Option.match(loadedState.snapshot, {
  onNone: () => EventStore.fold(matchDecider, loadedState.events),
  onSome: (s) => EventStore.foldFrom(matchDecider, s.state, loadedState.events),
})
// → snapshot asOfVersion 2, no events after it, version 2

It returns { snapshot, events, version }: the snapshot if there is one, the events after it in ascending order, and the stream head. The command handler loads every snapshot-configured stream this way, whatever the mode.

• A current inline snapshot loads in one request.
• A lagging snapshot (an "after-append" cadence, or a stream switched to inline after it already held events) still loads in one request when the lag fits the first page, which holds every + 1 items, and in two otherwise.
• The first page is read whatever the lag, so a load can read events the snapshot already covers (they are discarded). With "after-append" and a large every, that trades read capacity for fewer requests; mode: "inline" without every keeps the page at two items.

On a stream without a snapshot config, readLatest is read plus the head.

Loading from the snapshot alone

With mode: "inline" and no every, the snapshot is written in every append, so it is normally at the head. readLatest still reads two items to prove it: the snapshot and the newest event. A Query is charged for every item it reads, while a GetItem reads only the snapshot, so when events are large that read costs about twice as much as it needs to. verifySnapshot: false loads the snapshot item alone, with one GetItem:

// Inline without `every`: the snapshot is normally at the head, so load it alone.
const handleLean = EventStore.commandHandler(matchDecider, inlineMatchEvents, {
  verifySnapshot: false,
})
// One GetItem of the snapshot, then one transaction.
const ended = yield* handleLean(
  { matchId: "m-6" },
  { _tag: "EndMatch", result: "SAF won by 6 wickets" },
)
// → version 3, status "completed"

// The snapshot alone; its version is the snapshot's, unverified.
const lean = yield* inlineMatchEvents.readLatest({ matchId: "m-6" }, { verifySnapshot: false })
// → snapshot asOfVersion 3, events [], version 3

readLatest({ verifySnapshot: false }) returns { snapshot, events: [], version: snapshot.asOfVersion }. That version is unverified: events appended without a snapshot (a raw append, or data written before the stream switched to inline) may follow it. When there is no snapshot item, it falls back to the verified read.

The handler checks for itself. If the snapshot lags, every answer made on it is stale: a refusal or a no-op as much as an append. So the handler returns nothing it decided on an unverified snapshot until the head confirms it. A successful append confirms it by itself, because the append’s version condition held. Anything else is checked with one more read, and a stale snapshot falls back to a verified load:

• No snapshot item (a new stream, or data from before inline mode): it loads verified straight away, and decide runs once.
• With expectedVersion, at a different version than the snapshot: the snapshot is re-read verified before the If-Match check, so VersionConflict is reported only for a real mismatch, with the verified head as actualVersion. decide does not run.
• With expectedVersion, at the snapshot’s version: the snapshot is exactly the state the caller saw, so decide runs on it. If the decision does not append (a domain error, a no-op, or an append conflict), one head check (below) confirms the stream is still there. If the stream has moved past the caller’s version, the decision is discarded and the answer is what a verified load gives: DuplicateCommand for a redelivery that already committed, otherwise VersionConflict with the head as actualVersion.
• Without expectedVersion, a domain error or a no-op: one head check. If the head is at the snapshot, the answer stands. If not, the decision was made on stale state, so the handler loads verified and decides again once.
• Without expectedVersion, an append conflict: one verified load tells the two causes apart. If events follow the snapshot, the first decision was made on stale state, so the handler decides again once on the verified state and appends. If the snapshot is at the head, another writer won the race, and the VersionConflict goes to the retry policy as usual.

So in the last two cases, decide can run twice for one call, and so can a function-form additionalItems when both decisions append. This is the price of skipping the verification read, the same one a VersionConflict retry pays: a decision is only as current as the state it was made on, and one made on stale state can’t be patched, only made again. Without the second decision, a stale snapshot would make every command on the stream fail or give a stale answer, because only a successful inline append rewrites the snapshot. The second decision’s append repairs it.

The head check is a Query for any event after the snapshot’s version (reverse, Limit: 1), not a read of the newest event. While the snapshot is current, that range is empty, so the Query reads no items and costs only DynamoDB’s per-request minimum, however large the events are. It reads an event only when the snapshot lags, and then the event it reads is the head. So a decision that appends nothing costs one extra round trip, not extra item reads.

The option is accepted only on a mode: "inline" stream without every, the only mode that keeps the snapshot at the head. On any other snapshot config, creating the handler throws [EDD-9068], and readLatest dies with it. On a stream without a snapshot config the option is ignored.

Proving the head with the snapshot

Every append after version 0 carries the version-contiguity check, which reads the event at expectedVersion. On an inline stream without every, the snapshot in the same transaction can prove the head instead. headProof: "snapshot" conditions the snapshot Put on asOfVersion = expectedVersion (on its absence at version 0) and leaves the check out:

// The inline snapshot proves the head, so no contiguity check is written.
const handleProved = EventStore.commandHandler(matchDecider, inlineMatchEvents, {
  headProof: "snapshot",
})
yield* handleProved(
  { matchId: "m-7" },
  { _tag: "StartMatch", venue: "Eden Gardens", homeTeam: "IND", awayTeam: "AUS" },
)
// [event put, snapshot put conditioned on asOfVersion = 1]
const proved = yield* handleProved(
  { matchId: "m-7" },
  { _tag: "CompleteInnings", innings: 1, runs: 305, wickets: 10 },
)
// → version 2

Why it is enough: the snapshot’s condition proves the head is at least expectedVersion, and the event put at expectedVersion + 1 (attribute_not_exists) proves it is at most that.

• The handler uses the proof only when its load found the snapshot at the head: a verified load with no event after the snapshot, or an unverified one (verifySnapshot: false). A snapshot that lags (a raw append, data written before inline mode) would fail the condition, so that append writes the check, and its inline snapshot brings the snapshot back to the head.
• A failed proof is a VersionConflict, handled as any other: the retry policy, or under verifySnapshot: false the stale-snapshot re-decision above.
• On append, pass it with the snapshot: append(id, events, v, { snapshot, headProof: "snapshot" }). The caller then owns knowing that the snapshot is at v; a lagging one is a VersionConflict, never a gap.
• Accepted only on a mode: "inline" stream without every, and only with snapshot on the append. Any other snapshot config throws [EDD-9085] when the handler is created, and append dies with it. Without a snapshot, append fails with ValidationError.
• Without headProof, nothing changes: the check is written, and the inline snapshot Put is unconditional.

---

Step 18: Large Commands — Stepped Commands

Some commands decide more events than one transaction holds: an import, a bulk correction, a compensating undo. DynamoDB rejects such an append, so nothing is written, and the command fails with AppendTooLarge. The library never splits it: an append receives one decision — many events and the one state after all of them — and has no valid intermediate states to put at transaction boundaries. Only your decider can produce a valid state for each boundary.

So a large command is run as stepped commands. Here the scoring feed records deliveries, and a compensating command reverts them:

deliveries.ts

// Ball-by-ball deliveries from a scoring feed, and their compensating undo.
class DeliveryRecorded extends Schema.TaggedClass<DeliveryRecorded>()("DeliveryRecorded", {
  ball: Schema.Number,
}) {}

class DeliveryReverted extends Schema.TaggedClass<DeliveryReverted>()("DeliveryReverted", {
  ball: Schema.Number,
}) {}

type DeliveryEvent = DeliveryRecorded | DeliveryReverted

const DeliveryState = Schema.Struct({ live: Schema.Array(Schema.Number) })
type DeliveryState = typeof DeliveryState.Type

const Deliveries = EventStore.makeStream({
  table: EventsTable,
  streamName: "Deliveries",
  events: [DeliveryRecorded, DeliveryReverted],
  streamId: { composite: ["matchId"] },
  snapshot: { schema: DeliveryState, mode: "inline" },
})

type DeliveryCommand =
  | { readonly _tag: "RecordDeliveries"; readonly balls: ReadonlyArray<number> }
  | { readonly _tag: "RevertDeliveries"; readonly balls: ReadonlyArray<number> }

class NotRecorded extends Data.TaggedError("NotRecorded")<{ readonly ball: number }> {}

const deliveryDecider: EventStore.Decider<
  DeliveryState,
  DeliveryCommand,
  DeliveryEvent,
  NotRecorded
> = {
  initialState: { live: [] },
  decide: (command, state) =>
    Effect.gen(function* () {
      if (command._tag === "RecordDeliveries") {
        return command.balls.map((ball) => new DeliveryRecorded({ ball }))
      }
      for (const ball of command.balls) {
        if (!state.live.includes(ball)) return yield* new NotRecorded({ ball })
      }
      return command.balls.map((ball) => new DeliveryReverted({ ball }))
    }),
  evolve: (state, event) => ({
    live:
      event._tag === "DeliveryRecorded"
        ? [...state.live, event.ball]
        : state.live.filter((ball) => ball !== event.ball),
  }),
}

The application plans the steps, and each step is an ordinary command:

const deliveries = yield* EventStore.bind(Deliveries)
const handleDeliveries = EventStore.commandHandler(deliveryDecider, deliveries, {
  idempotency: { ttl: Duration.days(1) },
})

// A fixed step size, not one computed from event sizes: event content varies.
const STEP_SIZE = 50

/**
 * Run one large command as fixed-size steps, chained by `expectedVersion`.
 * Redelivered with the same `commandId` and `expectedVersion`, it resumes.
 */
const stepped = (
  matchId: string,
  balls: ReadonlyArray<number>,
  toCommand: (step: ReadonlyArray<number>) => DeliveryCommand,
  commandId: string,
  expectedVersion: number,
) =>
  Effect.gen(function* () {
    let version = expectedVersion
    let skipping = false
    for (let n = 0; n * STEP_SIZE < balls.length; n++) {
      const step = balls.slice(n * STEP_SIZE, (n + 1) * STEP_SIZE)
      // An ordinary command: decide, fold, one atomic append with its own
      // snapshot and sentinel. Each step has its own commandId.
      const run = (expected: number) =>
        handleDeliveries({ matchId }, toCommand(step), {
          commandId: `${commandId}#step-${n}`,
          expectedVersion: expected,
        }).pipe(Effect.map((result) => result.version))
      const committed = yield* run(version).pipe(
        // An earlier delivery committed this step. Keep the stale version, so
        // the next step is checked against its sentinel too, before `decide`.
        Effect.catchTag("DuplicateCommand", () => Effect.succeed(undefined)),
        // After skipped steps: the first step not committed. Run it at the head.
        Effect.catchTag("VersionConflict", (conflict) =>
          skipping && conflict.actualVersion !== undefined
            ? run(conflict.actualVersion)
            : Effect.fail(conflict),
        ),
      )
      skipping = committed === undefined
      if (committed !== undefined) version = committed
    }
    // Every step was committed before: the stream's head.
    return skipping
      ? yield* deliveries.currentVersion({ matchId }, { consistentRead: true })
      : version
  })

const feed = Array.from({ length: 150 }, (_, i) => i + 1)
const record = (balls: ReadonlyArray<number>): DeliveryCommand => ({
  _tag: "RecordDeliveries",
  balls,
})
const revert = (balls: ReadonlyArray<number>): DeliveryCommand => ({
  _tag: "RevertDeliveries",
  balls,
})

// One decision of 150 events cannot be one atomic append.
const tooLarge = yield* handleDeliveries({ matchId: "m-8" }, record(feed), {
  commandId: "feed-1",
}).pipe(Effect.flip)
// → AppendTooLarge (count 152, limit 100) — nothing written

const imported = yield* stepped("m-8", feed, record, "feed-1", 0)
// → version 150, in three atomic steps

// The compensating undo: planned newest first, in the same fixed-size steps.
const undone = yield* stepped("m-8", [...feed].reverse(), revert, "undo-1", imported)
// → version 300. A failure partway stops at the last step's real state.

// Redelivered (its response lost): every step is a duplicate, nothing is written.
const redelivered = yield* stepped("m-8", [...feed].reverse(), revert, "undo-1", imported)
// → version 300

1. The application plans the steps — newest first for an undo — each covering at most a configured number of entities. Prefer a fixed size to a size computed from the events: event content varies.
2. Each step is an ordinary commandHandler call — decide, fold, one atomic append with its own snapshot, projections and sentinel — with expectedVersion set to the version the previous step returned.
3. A failure partway leaves the stream at a real, consistent state: the last step that committed, with its snapshot and read models. A step that conflicts (another writer got in) fails with VersionConflict before anything of it is written.

With idempotency, each step needs its own commandId, ${commandId}#step-${n} above. That is what lets a redelivered large command resume after a partial failure. A redelivery carries the same commandId and the original expectedVersion, which is now stale, so the handler checks the step’s sentinel before deciding, and a committed step fails with DuplicateCommand. The helper skips it and keeps the stale version, so the next step is checked the same way, without decide running. The first step that was not committed fails with VersionConflict, whose actualVersion is the head, and the helper runs it from there. When every step was committed, nothing is written.

DuplicateCommand carries no version, so don’t chain the next step from the head after one: decide would run against steps already applied, and the undo above would fail with NotRecorded. And when a step fails with AppendTooLarge, its count against limit tells you the configured step size is too large for what each entity decides.

The rule of thumb: one command → one decision → one atomic append.

---

Step 19: Stream Indexes

Every read so far follows the stream’s version order. A stream index adds a second order: a sub-stream of the stream’s events, sorted by a key you derive from each event. Declare indexes on makeStream:

stream.ts

// An LSI can only be created with its table, so indexed streams get their own.
const ScorecardTable = Table.make({ schema: AppSchema, entities: {} })

const pad = (n: number) => String(n).padStart(4, "0")

const Scorecards = EventStore.makeStream({
  table: ScorecardTable,
  streamName: "Scorecard",
  events: [MatchStarted, InningsCompleted, MatchEnded],
  streamId: { composite: ["matchId"] },
  indexes: {
    // LSI (the default type): innings in innings order, strongly consistent.
    byInnings: {
      index: "lsi1",
      sk: "lsi1sk",
      key: (event) =>
        event._tag === "InningsCompleted" ? `INNINGS#${pad(event.innings)}` : undefined,
    },
    // GSI: innings by runs scored, eventually consistent, no 10 GB cap.
    byRuns: {
      type: "gsi",
      index: "gsi1",
      pk: "gsi1pk",
      sk: "gsi1sk",
      key: (event, version) =>
        event._tag === "InningsCompleted" ? `RUNS#${pad(event.runs)}#${pad(version)}` : undefined,
    },
  },
})

• key receives the domain event, the version it is appended at and the stream id. The string it returns is stored raw, with no casing and no prefix, so compose it to sort the way you query (zero-pad numbers). undefined leaves the event out of the index. A non-string or a key that throws fails the append with ValidationError before anything is written. An empty string, or a key over DynamoDB’s 1024-byte sort-key limit, is rejected by DynamoDB, so nothing is written, and the append fails with a ValidationError (operation: "EventStore.append.indexes") naming the index and the event’s version.
• Only event items carry index attributes. Idempotency sentinels are never in an index, and snapshots only in a GSI that derives its partition.
• type: "lsi" (the default) uses the table’s partition key, so the index holds exactly this stream’s indexed events and can be read with strongly consistent reads. type: "gsi" names a pk attribute too; append writes the stream’s partition key value into it, so the index is scoped to the same stream, read eventually consistently.

Creating the table

Table.definition does not derive event tables. EventStore.indexDefinitions(...streams) returns the CreateTable fragments for the streams’ indexes, each with the Projection its index declares (ALL unless the index sets a projection), to merge into your own CreateTable input:

const fragments = EventStore.indexDefinitions(Scorecards)
yield* client.createTable({
  TableName: scorecardConfig.name,
  BillingMode: "PAY_PER_REQUEST",
  KeySchema: [
    { AttributeName: "pk", KeyType: "HASH" },
    { AttributeName: "sk", KeyType: "RANGE" },
  ],
  AttributeDefinitions: [
    { AttributeName: "pk", AttributeType: "S" },
    { AttributeName: "sk", AttributeType: "S" },
    ...fragments.AttributeDefinitions,
  ],
  LocalSecondaryIndexes: fragments.LocalSecondaryIndexes,
  GlobalSecondaryIndexes: fragments.GlobalSecondaryIndexes,
})

LSIs have table-wide consequences

• An LSI must be created with the table. It cannot be added later.
• A table with any LSI caps the item collection of every partition key value at 10 GB: every stream’s partitions, indexed or not, and every entity partition that shares the table. A write that would pass the cap fails. If a stream can outgrow that, use a GSI.
• An LSI’s projection is fixed when the table is created. indexDefinitions emits the Projection each index declares (ALL by default; see Index projection). If you write the definition yourself, copy it exactly: the reads depend on what the index holds.

A GSI can be added to an existing table, but that does not index the events already in it. Index attributes are written only by append, and events are never rewritten, so an index declared on a stream that already holds events contains only the events appended from then on, and changing a key leaves earlier events under their old keys. Reads then return only what was indexed, without an error. The library does not backfill; whether and how to re-index existing events is the application’s decision.

Reading a sub-stream

const scorecards = yield* EventStore.bind(Scorecards)
yield* scorecards.append(
  { matchId: "m-10" },
  [
    new MatchStarted({ venue: "Lords", homeTeam: "ENG", awayTeam: "AUS" }),
    new InningsCompleted({ innings: 1, runs: 245, wickets: 10 }),
    new InningsCompleted({ innings: 2, runs: 310, wickets: 7 }),
    new InningsCompleted({ innings: 3, runs: 120, wickets: 10 }),
    new InningsCompleted({ innings: 4, runs: 56, wickets: 2 }),
  ],
  0,
)

// LSI sub-stream: the second-half innings, strongly consistent.
const secondHalf = yield* scorecards.readIndex(
  "byInnings",
  { matchId: "m-10" },
  { between: ["INNINGS#0003", "INNINGS#0004"], consistentRead: true },
)
// → versions 4 and 5 (MatchStarted is not in the index)

// GSI sub-stream: the highest-scoring innings.
const [top] = yield* scorecards.readIndex(
  "byRuns",
  { matchId: "m-10" },
  { reverse: true, limit: 1 },
)
// → innings 2, 310 runs

// query.index composes with the Query combinators.
const firstInnings = yield* scorecards.provide(
  scorecards.query
    .index("byInnings", { matchId: "m-10" })
    .pipe(Query.where({ beginsWith: "INNINGS#0001" }), Query.consistentRead(), Query.collect),
)

readIndex(name, streamId, options?) returns decoded StreamEvents in index order. Its options are beginsWith or between (at most one; omit both to read the whole sub-stream), reverse, limit and consistentRead. Index names are type-checked against the stream’s indexes. DynamoDB rejects consistentRead on a GSI-backed index, an empty key bound, and a between whose lower bound sorts (by UTF-8 bytes) after its upper bound; each fails with a ValidationError naming the index (operation: "EventStore.readIndex", or "query.consistentRead"), with DynamoDB’s error as its cause. query.index(name, streamId) returns a plain Query for the full set of combinators.

---

Step 20: Event Metadata

An event records what happened in the domain. Facts about the event, such as the request that caused it, an id a publisher sends it under, or its id in the store it was imported from, are metadata. Declare a metadata schema on the stream and every event item can carry a metadata attribute, encoded through that schema on the way in and decoded into StreamEvent.metadata on the way out:

stream.ts

// Facts about an event that are not part of the domain event itself.
const AuditMetadata = Schema.Struct({
  // The request that produced the event: shared by every event it appends.
  correlationId: Schema.String,
  // An id of the event's own: a ULID, minted when the event is decided.
  eventId: Schema.optionalKey(Schema.String.check(Schema.isULID())),
  // For an imported event: its id in the store it came from.
  sourceId: Schema.optionalKey(Schema.String),
})

const AuditedMatchEvents = EventStore.makeStream({
  table: EventsTable,
  streamName: "AuditedMatch",
  events: [MatchStarted, InningsCompleted, MatchEnded],
  streamId: { composite: ["matchId"] },
  metadata: AuditMetadata,
})

Metadata shared by the append

metadata takes one value for the whole append and writes a copy onto every event. It suits facts about the command, like a correlation id:

const audited = yield* EventStore.bind(AuditedMatchEvents)

// One value for the whole append: every event carries a copy.
yield* audited.append(
  { matchId: "m-20" },
  [
    new MatchStarted({ venue: "WACA", homeTeam: "AUS", awayTeam: "SL" }),
    new InningsCompleted({ innings: 1, runs: 301, wickets: 9 }),
  ],
  0,
  { metadata: { correlationId: "req-81c2" } },
)
// → v1 and v2 both carry { correlationId: "req-81c2" }

Metadata per event

Some metadata belongs to one event, not to the append. eventMetadata takes one entry per event, in the same order as events: entry i is written onto event i and read back as that event’s metadata. An undefined entry writes its event without a metadata attribute, exactly as an append without metadata does:

// An import from a previous store: each event keeps its original id.
yield* audited.append(
  { matchId: "m-21" },
  [
    new MatchStarted({ venue: "Kensington Oval", homeTeam: "WI", awayTeam: "ENG" }),
    new InningsCompleted({ innings: 1, runs: 202, wickets: 10 }),
    new InningsCompleted({ innings: 2, runs: 205, wickets: 4 }),
  ],
  0,
  {
    eventMetadata: [
      { correlationId: "import-17", sourceId: "legacy-118" },
      // `undefined`: this event is written without metadata.
      undefined,
      { correlationId: "import-17", sourceId: "legacy-121" },
    ],
  },
)

// Each event reads back with its own entry.
const importedEvents = yield* audited.read({ matchId: "m-21" })
const sourceIds = importedEvents.map((event) => event.metadata?.sourceId)
// → ["legacy-118", undefined, "legacy-121"]

• Each defined entry is encoded through the metadata schema, as metadata is. An entry that fails fails the append with a ValidationError whose operation names its index (EventStore.append.eventMetadata[1]).
• metadata and eventMetadata are mutually exclusive. Passing both is a type error, and a ValidationError for callers the types do not reach. For shared and per-event fields together, spread the shared ones into each entry: events.map(() => ({ ...shared, eventId })).
• An eventMetadata whose length differs from events fails with a ValidationError before anything is sent, even when events is empty.
• The read side needs nothing new. read, readFrom, readLatest and readIndex already decode each item’s own metadata.

Metadata from the command handler

A handler’s per-call metadata is fixed before decide runs, so it cannot know how many events it must cover. Its eventMetadata is a function of the Decision, the same one a function-form additionalItems receives, returning one entry per event directly or as an Effect.

The example mints each event’s id as a ULID with Effect’s Crypto service, and the metadata schema checks it with Schema.isULID(). crypto.randomULID takes the id’s time from the Clock and its 80 random bits from Crypto, so ids sort by when they were minted, and a test that provides a TestClock and a Crypto whose randomBytes returns fixed bytes gets the same ids on every run. When a standard UUID is needed, crypto.randomUUIDv7 works the same way. Effect provides no default Crypto service; the example builds one on the Web Crypto API (see Layer Setup).

// One ULID per event: its time comes from the Clock, its random bits from
// the Crypto service.
const crypto = yield* Crypto.Crypto
const mintEventIds = (correlationId: string, events: ReadonlyArray<MatchEvent>) =>
  Effect.forEach(events, () =>
    Effect.map(crypto.randomULID, (eventId) => ({ correlationId, eventId })),
  )

// Idempotent: a redelivered command fails with DuplicateCommand, so the ids
// of its first commit are the only ones ever written.
const handleAudited = EventStore.commandHandler(matchDecider, audited, {
  idempotency: { ttl: Duration.days(1) },
})

// A function of the decision: one entry per event `decide` returned.
yield* handleAudited(
  { matchId: "m-22" },
  { _tag: "StartMatch", venue: "Adelaide Oval", homeTeam: "AUS", awayTeam: "WI" },
  { commandId: "msg-4470", eventMetadata: ({ events }) => mintEventIds("msg-4470", events) },
)
// → v1 carries an eventId such as "01M4M3EK1SVFC94QZ8SC9H0ZPW"

// A message delivered twice, its first response lost.
const completeInnings = handleAudited(
  { matchId: "m-22" },
  { _tag: "CompleteInnings", innings: 1, runs: 289, wickets: 10 },
  { commandId: "msg-4471", eventMetadata: ({ events }) => mintEventIds("msg-4471", events) },
)
yield* completeInnings
const duplicate = yield* Effect.flip(completeInnings)
// → DuplicateCommand: v2 keeps the eventId of the first delivery

When it runs:

load → [expectedVersion check] → decide → fold new events → eventMetadata → eventTimestamps → additionalItems → append → [snapshot]

• It runs after the new events are folded and before additionalItems, so an effectful pair runs in a predictable order. It is not called when decide returns no events.
• It runs again on every attempt: each VersionConflict retry, and the second decision a stale snapshot triggers under verifySnapshot: false. Only the committed attempt’s entries are ever written, so minting fresh ids per attempt is safe.
• A result whose length differs from the decision’s events fails with a ValidationError (operation: "EventStore.commandHandler.eventMetadata") before anything is sent. A failure of the returned Effect surfaces in the handler’s error channel. Neither is retried.
• With idempotency, a redelivered command fails with DuplicateCommand and never commits a second set of ids. Without it, a redelivery after a lost response decides and appends again, with new ids. That is why the example’s handler is idempotent and takes the delivery id as commandId: the second delivery of msg-4471 fails with DuplicateCommand, and v2 keeps the id of the first.

Original event times

Every event item also has a timestamp, read back as StreamEvent.timestamp. By default it is the time of the append, from the Clock. An import usually wants each event’s original time instead, and pinning the clock per chunk would stamp every event of a chunk with one time. eventTimestamps takes one DateTime per event, in the same order as events: entry i is written as event i’s timestamp, and an undefined entry keeps the append’s own time:

// History from a previous store, each record with its own id and the time
// it happened.
const legacyRecords = [
  {
    sourceId: "legacy-204",
    occurredAt: "2025-01-02T23:30:00Z",
    event: new MatchStarted({ venue: "SCG", homeTeam: "AUS", awayTeam: "IND" }),
  },
  {
    sourceId: "legacy-205",
    occurredAt: "2025-01-03T06:40:00Z",
    event: new InningsCompleted({ innings: 1, runs: 185, wickets: 10 }),
  },
  {
    sourceId: "legacy-206",
    occurredAt: "2025-01-04T04:15:00Z",
    event: new InningsCompleted({ innings: 2, runs: 181, wickets: 10 }),
  },
]

// One append keeps every event's original time: entry `i` of
// `eventTimestamps` is event `i`'s timestamp (`undefined` keeps the
// append's own time). It combines with either metadata option.
yield* audited.append(
  { matchId: "m-23" },
  legacyRecords.map((record) => record.event),
  0,
  {
    eventMetadata: legacyRecords.map((record) => ({
      correlationId: "import-18",
      sourceId: record.sourceId,
    })),
    eventTimestamps: legacyRecords.map((record) => DateTime.makeUnsafe(record.occurredAt)),
  },
)

// Each event reads back with its own time, in UTC.
const history = yield* audited.read({ matchId: "m-23" })
const occurred = history.map((event) => `${event.metadata?.sourceId} ${event.timestamp}`)
// → ["legacy-204 2025-01-02T23:30:00.000Z", "legacy-205 2025-01-03T06:40:00.000Z", …]

• A stream is ordered by version, not by timestamp. The library never reads the value back for ordering, so timestamps may repeat, run backwards or lie in the future, as imported history can.
• Only the event items take the given times. The idempotency sentinel, its TTL and an inline snapshot keep the append’s own time, because they record when the append was written.
• Each entry is written with DateTime.formatIso, the format of the append’s own time, so a zoned DateTime is stored as its UTC instant.
• eventTimestamps is independent of metadata and eventMetadata, so it combines with either. A length that differs from events, or an entry that is neither a valid DateTime nor undefined (a JS Date, say: convert it with DateTime.fromDateUnsafe; or a DateTime with a NaN epoch from a bad parse), fails with a ValidationError (operation: "EventStore.append.eventTimestamps") before anything is sent, even when events is empty.
• On a command handler, eventTimestamps is a function of the Decision, like eventMetadata. It returns the array directly or an Effect of it, whose error and requirements join the handler’s. It runs after eventMetadata and before additionalItems, is not called when decide returns no events, and runs again on every attempt. A result of the wrong length, or with an entry that is not a DateTime or undefined, fails with a ValidationError (operation: "EventStore.commandHandler.eventTimestamps"); neither that nor a failure of the returned Effect is retried.

The Effect form suits times that come from somewhere else, such as an import’s source log:

// Through the handler, the times can come from an effect — here a lookup in
// the import's source log, which fails if a record is missing. The effect's
// error and requirements join the handler's.
const sourceLog = new Map([["m-24:MatchStarted", "2025-02-01T09:00:00Z"]])
const originalTime = (key: string) =>
  Effect.fromNullishOr(sourceLog.get(key)).pipe(Effect.map((iso) => DateTime.makeUnsafe(iso)))

const handleImport = EventStore.commandHandler(matchDecider, audited)
yield* handleImport(
  { matchId: "m-24" },
  { _tag: "StartMatch", venue: "Basin Reserve", homeTeam: "NZL", awayTeam: "SL" },
  {
    eventTimestamps: ({ events }) =>
      Effect.forEach(events, (event) => originalTime(`m-24:${event._tag}`)),
  },
)
// → v1 is stamped 2025-02-01T09:00:00.000Z

Why not pack a list into metadata?

Before eventMetadata, the only place for per-event values was a list inside the shared metadata, with each reader picking its entry by position. Every event then stores the whole list, so an append of N events stores N² entries, and a stream index projecting ALL (the default; see Index projection) copies all of them. That cost counts towards DynamoDB’s 400 KB item and 4 MB transaction limits. eventMetadata stores each entry once, on its own event.

Index projection

A stream index projects ALL by default, so each entry is a full copy of its event, metadata included. Metadata is often an event’s largest attribute, and the sub-stream an index serves rarely needs it. projection on an index entry chooses what the index copies:

Not an inline projection

This is DynamoDB’s index projection: which event attributes a secondary index stores. It is unrelated to the inline projections of Step 16, which are read-model items a command handler commits in the same transaction as its events.

stream.ts

// Metadata is often an event's largest attribute, and an index projecting ALL
// copies it into every entry. `projection` keeps it out.
const AuditedScorecardTable = Table.make({ schema: AppSchema, entities: {} })

const AuditedScorecards = EventStore.makeStream({
  table: AuditedScorecardTable,
  streamName: "AuditedScorecard",
  events: [MatchStarted, InningsCompleted, MatchEnded],
  streamId: { composite: ["matchId"] },
  metadata: AuditMetadata,
  indexes: {
    // Every event attribute except `metadata`: read with one query, and
    // `metadata` is typed (and read as) undefined.
    byInnings: {
      index: "lsi1",
      sk: "lsi1sk",
      projection: "withoutMetadata",
      key: (event) =>
        event._tag === "InningsCompleted" ? `INNINGS#${pad(event.innings)}` : undefined,
    },
    // Keys only: the smallest entries. A read queries the index, then fetches
    // the full events from the table with BatchGetItem, in index order.
    byRuns: {
      type: "gsi",
      index: "gsi1",
      pk: "gsi1pk",
      sk: "gsi1sk",
      projection: "keys",
      key: (event, version) =>
        event._tag === "InningsCompleted" ? `RUNS#${pad(event.runs)}#${pad(version)}` : undefined,
    },
  },
})

projection	Emitted Projection	Index write per event	Read
"all" (default)	ALL	the full event, 1 WCU per KB	one query
"withoutMetadata"	INCLUDE of __edd_e__, data, eventType, streamId, timestamp, version	the event without metadata	one query; metadata is undefined, and typed so
"keys"	INCLUDE ["__edd_e__"]	the keys and __edd_e__: 1 WCU	LSI: one query with Select: ALL_ATTRIBUTES, and DynamoDB fetches each event from the table. GSI: the query, then BatchGetItem for each page’s events, returned in index order

For an event with 2 KB of data and 6 KB of metadata, the table write costs about 8 WCU. Each all index adds 8 WCU more, each withoutMetadata index 2 to 3, each keys index 1. A keys index pays on the read side instead: each event read through it costs a table read, rounded up to 4 KB. On an LSI, smaller entries also leave more of each partition’s 10 GB item-collection cap for the events themselves.

indexDefinitions emits the matching Projection:

const projected = EventStore.indexDefinitions(AuditedScorecards)
// byInnings (LSI) → Projection: { ProjectionType: "INCLUDE", NonKeyAttributes:
//   ["__edd_e__", "data", "eventType", "streamId", "timestamp", "version"] }
// byRuns (GSI)    → Projection: { ProjectionType: "INCLUDE", NonKeyAttributes: ["__edd_e__"] }
yield* client.createTable({
  TableName: auditedScorecardConfig.name,
  BillingMode: "PAY_PER_REQUEST",
  KeySchema: [
    { AttributeName: "pk", KeyType: "HASH" },
    { AttributeName: "sk", KeyType: "RANGE" },
  ],
  AttributeDefinitions: [
    { AttributeName: "pk", AttributeType: "S" },
    { AttributeName: "sk", AttributeType: "S" },
    ...projected.AttributeDefinitions,
  ],
  LocalSecondaryIndexes: projected.LocalSecondaryIndexes,
  GlobalSecondaryIndexes: projected.GlobalSecondaryIndexes,
})

The reads choose their path from the projection:

const auditedScorecards = yield* EventStore.bind(AuditedScorecards)
yield* auditedScorecards.append(
  { matchId: "m-30" },
  [
    new MatchStarted({ venue: "Eden Park", homeTeam: "NZL", awayTeam: "ENG" }),
    new InningsCompleted({ innings: 1, runs: 278, wickets: 10 }),
    new InningsCompleted({ innings: 2, runs: 301, wickets: 8 }),
  ],
  0,
  { metadata: { correlationId: "req-9a41" } },
)

// "withoutMetadata": the index does not hold metadata, so it reads as
// undefined — and its type is `undefined`, so it cannot be read by mistake.
const innings = yield* auditedScorecards.readIndex(
  "byInnings",
  { matchId: "m-30" },
  { consistentRead: true },
)
// → [v2, v3], each with metadata: undefined

// "keys": the query finds the entries, then the events are fetched from the
// table, so metadata is there.
const byRuns = yield* auditedScorecards.readIndex(
  "byRuns",
  { matchId: "m-30" },
  { reverse: true },
)
// → [v3, v2], each with metadata { correlationId: "req-9a41" }

• readIndex and query.index type metadata per index name: undefined through a withoutMetadata index, the stream’s metadata type otherwise. A withoutMetadata read drops metadata even when the deployed index holds it, so the value always matches the type.
• A keys GSI read fetches only the events a page keeps (after limit), 100 keys per BatchGetItem. Unprocessed keys are retried with backoff, and the read fails with DynamoError after 5 retries. The fetch is eventually consistent; a key missing from the response is read once more with ConsistentRead. consistentRead on a GSI is still rejected (by DynamoDB, diagnosed as a ValidationError naming the index), while on a keys LSI it applies to the fetch too.
• Query.count counts index entries and never fetches. Query.select works on every projection: a keys LSI fetches the named attributes, and a keys GSI moves the selection onto BatchGetItem.
• A Query.filter on a withoutMetadata or keys index may only name attributes the index projects; DynamoDB rejects any other. Filter on the client with Query.filterBy instead.
• Streams that share a physical index must declare the same projection; indexDefinitions refuses a mismatch with [EDD-9067].
• An unknown projection, or a misspelled member such as projecton, is a type error, and makeStream refuses it with [EDD-9063].
• withoutMetadata lists 6 NonKeyAttributes and keys lists 1. DynamoDB allows at most 100 across all of a table’s indexes, so CreateTable rejects the 17th withoutMetadata index on one table.

The declared projection must match the deployed index

The library migrates nothing. An LSI’s projection is fixed when the table is created. A GSI’s projection cannot be updated: the index must be deleted and created again, and it is unavailable while it backfills. indexDefinitions describes a new index.

With CDK or Terraform, copy each Projection from indexDefinitions exactly. In particular, keys is INCLUDE ["__edd_e__"], not KEYS_ONLY: every read filters on __edd_e__, and DynamoDB refuses a filter on an attribute the index does not project. If the deployment differs:

• Without __edd_e__ (a KEYS_ONLY index declared keys, say): every read, count included, fails with DynamoDB’s ValidationException, Secondary index gsi1 does not project one or more filter attributes: [__edd_e__], as a DynamoValidationError.
• Narrower than declared (an INCLUDE ["__edd_e__"] index declared all or withoutMetadata): rows lack event attributes, and the read fails with a ValidationError (operation: "EventStore.index.decode") naming the index and the attributes its rows lack, with the decode error as its cause.
• Wider than declared works and costs more. One drift fails nothing: an index declared all but deployed withoutMetadata reads metadata as undefined.

Step 21: Packed Storage

Event data and snapshot state are stored as native DynamoDB maps by default. For long streams that you fold often, or for large states, store them as one JSON string instead, compressed above a threshold:

stream.ts

class Recorded extends Schema.TaggedClass<Recorded>()("Recorded", {
  amount: Schema.Number,
  note: Schema.String,
}) {}

class Voided extends Schema.TaggedClass<Voided>()("Voided", {
  reason: Schema.String,
}) {}

type LedgerEvent = Recorded | Voided

const LedgerState = Schema.Struct({
  total: Schema.Number,
  entries: Schema.Array(Schema.Struct({ amount: Schema.Number, note: Schema.String })),
})
type LedgerState = typeof LedgerState.Type

const EventsTable = Table.make({ schema: AppSchema, entities: { Orders } })

const Ledger = EventStore.makeStream({
  table: EventsTable,
  streamName: "Ledger",
  events: [Recorded, Voided],
  streamId: { composite: ["ledgerId"] },
  metadata: Schema.Struct({ correlationId: Schema.String }),
  // Event data as JSON (S); metadata always stays a map
  storedAs: { data: DynamoModel.Packed({ format: "json" }) },
  snapshot: {
    schema: LedgerState,
    mode: "inline",
    // Snapshot state as JSON, zstd-compressed from 1 KB (the default threshold)
    storedAs: DynamoModel.Packed({ format: "json", compression: { codec: "zstd" } }),
  },
})

In a downstream benchmark on Lambda, reading ~6,000 small events was about 2× faster with data stored as JSON, and a ~47 KB state stored as zstd-compressed JSON took 5 write units instead of 38 as a map. Compress only large values: compressing every small event made reads about 4× slower. Every read path decodes each item by its stored form, so turning packing on needs no rewrite. A stream can hold a mix of map, JSON and compressed items. Metadata stays a map.

Query.select raw rows return data as parsed JSON whatever its stored form. A filter on a path inside packed data is refused with [EDD-9076], so use Query.filterBy. measureAppend and measureSnapshot return the stored sizes for chunk planning, and decodeStreamImage decodes DynamoDB Streams records. See Packed Storage for compressor registration, the stored form, sizing and rollout order.

Step 22: Shared Indexes Across Streams

A GSI stream index is normally scoped to its stream: append writes the stream’s own partition key into the index’s pk. Give the index a partition function instead, and several streams can write one partition of a GSI they all declare. One query of that partition then reads a whole group of streams. Here a series groups its matches, squads and umpire appointments:

series.ts

// A series groups its matches, squads and umpire appointments. The three
// streams derive one partition of `gsi1` per series, so one query of it reads
// the whole series, each match's current state included.
const SeriesTable = Table.make({ schema: AppSchema, entities: {} })

class SquadNamed extends Schema.TaggedClass<SquadNamed>()("SquadNamed", {
  players: Schema.Array(Schema.String),
}) {}

class UmpireAppointed extends Schema.TaggedClass<UmpireAppointed>()("UmpireAppointed", {
  seriesId: Schema.String,
  matchId: Schema.String,
}) {}

const SeriesMatches = EventStore.makeStream({
  table: SeriesTable,
  streamName: "SeriesMatch",
  events: [MatchStarted, InningsCompleted, MatchEnded],
  streamId: { composite: ["seriesId", "matchId"] },
  snapshot: { schema: MatchStateSchema, mode: "inline" },
  indexes: {
    bySeries: {
      type: "gsi",
      index: "gsi1",
      pk: "gsi1pk",
      sk: "gsi1sk",
      // Written raw: no prefix, no casing.
      partition: (_event, _version, streamId) => `SERIES#${streamId.seriesId}`,
      key: (_event, version, streamId) => `MATCH#${streamId.matchId}#EVT#${pad(version)}`,
      // Each match's snapshot is in the partition too, written in the same
      // transaction as the events (an inline snapshot).
      snapshot: (_state, _asOfVersion, streamId) => ({
        partition: `SERIES#${streamId.seriesId}`,
        key: `MATCH#${streamId.matchId}#STATE`,
      }),
    },
  },
})

const SeriesSquads = EventStore.makeStream({
  table: SeriesTable,
  streamName: "SeriesSquad",
  events: [SquadNamed],
  streamId: { composite: ["seriesId", "team"] },
  indexes: {
    bySeries: {
      type: "gsi",
      index: "gsi1",
      pk: "gsi1pk",
      sk: "gsi1sk",
      partition: (_event, _version, streamId) => `SERIES#${streamId.seriesId}`,
      key: (_event, version, streamId) => `SQUAD#${streamId.team}#EVT#${pad(version)}`,
    },
  },
})

// An umpire's stream is keyed by the umpire alone: the series comes from
// each event.
const Umpires = EventStore.makeStream({
  table: SeriesTable,
  streamName: "Umpire",
  events: [UmpireAppointed],
  streamId: { composite: ["umpireId"] },
  indexes: {
    bySeries: {
      type: "gsi",
      index: "gsi1",
      pk: "gsi1pk",
      sk: "gsi1sk",
      partition: (event) => `SERIES#${event.seriesId}`,
      key: (_event, version, streamId) => `UMPIRE#${streamId.umpireId}#EVT#${pad(version)}`,
    },
  },
})

• partition(event, version, streamId) returns the event’s partition in the index, stored raw (no casing, no prefix). It can come from the event, as the umpire stream’s does. undefined leaves the event out of the index, as an undefined key does: an event is indexed only when both return a string. Both functions run for every appended event.
• Within one append, every event’s partition (where it is a string) must be the same, and the inline snapshot’s must match it; otherwise the append fails with ValidationError before anything is written. A non-string or a function that throws fails it the same way. An empty partition, or one over DynamoDB’s 2048-byte limit, is DynamoDB’s to reject; the ValidationError names the index and the event.
• snapshot(state, asOfVersion, streamId) puts the stream’s snapshot item in the index too, as { partition, key }, or leaves it out with undefined. The attributes are on the snapshot item itself, written by the same Put: an inline snapshot enters the index in the append’s own transaction, and writeSnapshot and commandHandler’s snapshots carry it too. The Put replaces the item, so returning undefined takes a snapshot that was in the index out of it. snapshot requires partition and a stream snapshot config.
• A snapshot is the state as of its asOfVersion. It is current after every append only with mode: "inline" and no every; otherwise a reader of the index folds the stream’s events after asOfVersion onto it.

indexDefinitions emits an index that several streams declare alike once:

const seriesIndexes = EventStore.indexDefinitions(SeriesMatches, SeriesSquads, Umpires)
// gsi1 once, Projection ALL: the three streams declare it alike.
yield* client.createTable({
  TableName: seriesConfig.name,
  BillingMode: "PAY_PER_REQUEST",
  KeySchema: [
    { AttributeName: "pk", KeyType: "HASH" },
    { AttributeName: "sk", KeyType: "RANGE" },
  ],
  AttributeDefinitions: [
    { AttributeName: "pk", AttributeType: "S" },
    { AttributeName: "sk", AttributeType: "S" },
    ...seriesIndexes.AttributeDefinitions,
  ],
  GlobalSecondaryIndexes: seriesIndexes.GlobalSecondaryIndexes,
})

Whether a declaration derives its partition or indexes its snapshot is not part of the index’s definition, so streams that differ only there share it. Different key attributes, type or projection still throw [EDD-9067]. A "withoutMetadata" index that any stream puts its snapshot in also projects asOfVersion and state.

Reading the partition

const seriesMatches = yield* EventStore.bind(SeriesMatches)
const handleSeriesMatch = EventStore.commandHandler(matchDecider, seriesMatches)
const firstTest = { seriesId: "ashes-25", matchId: "t1" }
yield* handleSeriesMatch(firstTest, {
  _tag: "StartMatch",
  venue: "Perth",
  homeTeam: "AUS",
  awayTeam: "ENG",
})
yield* handleSeriesMatch(firstTest, { _tag: "CompleteInnings", innings: 1, runs: 172, wickets: 10 })

const squads = yield* EventStore.bind(SeriesSquads)
yield* squads.append(
  { seriesId: "ashes-25", team: "ENG" },
  [new SquadNamed({ players: ["Root", "Stokes", "Wood"] })],
  0,
)
const umpires = yield* EventStore.bind(Umpires)
yield* umpires.append(
  { umpireId: "u-7" },
  [new UmpireAppointed({ seriesId: "ashes-25", matchId: "t1" })],
  0,
)

// Another service reading the partition directly gets every stream's
// events, and each match's snapshot.
const series = yield* client.query({
  TableName: seriesConfig.name,
  IndexName: "gsi1",
  KeyConditionExpression: "#pk = :pk",
  ExpressionAttributeNames: { "#pk": "gsi1pk" },
  ExpressionAttributeValues: { ":pk": { S: "SERIES#ashes-25" } },
})
// → MATCH#t1#EVT#0001, MATCH#t1#EVT#0002, MATCH#t1#STATE,
//   SQUAD#ENG#EVT#0001, UMPIRE#u-7#EVT#0001

// A stream reads its own events in the partition: pass the partition.
const t1 = yield* seriesMatches.readIndex("bySeries", firstTest, {
  partition: "SERIES#ashes-25",
})
// → v1 and v2 of match t1 only

• Other readers can query the partition directly; items are as stored. Packed data and state (see Step 21) stay packed.
• readIndex and query.index read an index that derives its partition one partition at a time, so partition is required (a type error without it). They return the given stream id’s events in that partition: the read filters on the table’s pk, which every index projects, so the other streams, the stream’s other ids and the snapshots are left out. Passing partition for a stream-scoped index fails with ValidationError ([EDD-9080] from query.index).
• Nothing is backfilled: only items written after the index is declared are in it.

---

Step 23: State From Your Own Record

Some applications already keep the aggregate’s state in a record of their own: its current state plus the stream version it is as of, written with every append and conditioned on the version that was loaded. That record makes two things redundant:

• The stream load. loadState gives the handler the state and the head from the record, instead of replaying the stream or reading its snapshot.
• The contiguity check. The record’s condition already proves the head, so headProof: { additionalItem: n } names it and the check is left out.

The record lives on its own table here, next to the stream:

// The application's own record of a match: its state, and the stream version
// that state is as of. Every append writes it, in the same transaction.
class MatchRecord extends Schema.Class<MatchRecord>("MatchRecord")({
  matchId: Schema.String,
  state: MatchStateSchema,
  streamVersion: Schema.Number,
}) {}

const MatchRecords = Entity.make({
  model: MatchRecord,
  entityType: "MatchRecord",
  primaryKey: {
    pk: { field: "pk", composite: ["matchId"] },
    sk: { field: "sk", composite: [] },
  },
})

const RecordTable = Table.make({ schema: AppSchema, entities: { MatchRecords } })

const RecordedMatches = EventStore.makeStream({
  table: RecordTable,
  streamName: "RecordedMatch",
  events: [MatchStarted, InningsCompleted, MatchEnded],
  streamId: { composite: ["matchId"] },
})

// The record an append after `decision.version` writes, conditioned on the
// version the handler loaded: it holds only while the stream is still there.
const recordOf =
  (matchId: string) =>
  ({ state, version, events }: EventStore.Decision<MatchState, MatchEvent>) => [
    MatchRecords.put({ matchId, state, streamVersion: version + events.length }).pipe(
      version === 0
        ? MatchRecords.condition((t, { notExists }) => notExists(t.streamVersion))
        : MatchRecords.condition({ streamVersion: version }),
    ),
  ]

The handler loads from the record, decides, and appends the events with the record as additionalItems[0]:

const recordedMatches = yield* EventStore.bind(RecordedMatches)
const handleRecorded = EventStore.commandHandler(matchDecider, recordedMatches, {
  // State and head from the record, read strongly consistent; none is a new match.
  loadState: ({ matchId }) =>
    Effect.gen(function* () {
      const record = yield* Entity.consistentRead(MatchRecords.get({ matchId }))
      return Option.some({ state: record.state, version: record.streamVersion })
    }).pipe(Effect.catchTag("ItemNotFound", () => Effect.succeed(Option.none()))),
  // The record at additionalItems[0] proves the head: no contiguity check.
  headProof: { additionalItem: 0 },
  retry: 3,
})

yield* handleRecorded(
  { matchId: "m-8" },
  { _tag: "StartMatch", venue: "Gabba", homeTeam: "AUS", awayTeam: "IND" },
  { additionalItems: recordOf("m-8") },
)
// Loaded from the record: one GetItem, then [event put, record put].
const recorded = yield* handleRecorded(
  { matchId: "m-8" },
  { _tag: "CompleteInnings", innings: 1, runs: 260, wickets: 10 },
  { additionalItems: recordOf("m-8") },
)
// → version 2; the record holds the same state at streamVersion 2

loadState

• It returns an Option of { state, version }. None is a new stream: the decider’s initial state at version 0.
• The rest of the cycle is unchanged: the If-Match check against the loaded version, decide, the fold, the derived items, the append. A VersionConflict runs loadState again under the retry policy.
• A domain error or a no-op is answered from the record. The handler trusts it and doesn’t read the stream.
• Read strongly consistent. An eventually consistent read can return a record behind the head. Nothing is corrupted (the append conflicts and retries), but every stale read costs a retry.
• Its error and requirements join the handler’s. On a bound stream they are the handler’s only requirements.
• A version that is not a non-negative integer, or a result that is not an Option, fails with ValidationError (operation: "EventStore.commandHandler.loadState"), never retried.
• Without headProof, the append still writes the contiguity check, so a record that is stale or ahead conflicts instead of opening a gap.
• The snapshot is not read. On a mode: "inline" stream without every, the snapshot is still written in every append. With every, one is written when an append crosses a multiple of every.
• Creating the handler throws [EDD-9085] for loadState with verifySnapshot: false (there is no stream load to skip) or with headProof: "snapshot" (the handler never sees the snapshot).

headProof: { additionalItem: n }

• On append, additionalItems[n] is the proof: append(id, events, v, { additionalItems, headProof: { additionalItem: n } }). On the handler, it applies to every append the handler makes, so every call that appends must return that item.
• The item must carry a condition of its own: .condition(), create() or Transaction.check. One without, or an n that is not an index into additionalItems, fails with ValidationError before anything is written. A put with no condition would prove nothing.
• When its condition fails, the append fails with VersionConflict (so the retry policy sees it), never AdditionalItemConditionFailed. It ranks after DuplicateCommand and the event puts, ahead of the other additional items.

The record is a contract with every writer

The library can’t see who else writes the stream. The proof holds only while every writer (the handler, an import, a migration, a raw stream.append) writes the record in the same transaction as its events, conditioned the same way.

• A writer that appends without the record leaves the record behind the head. Every later append then fails its event put, so the stream loops on VersionConflict until the record is repaired. That’s a stuck stream, not a gap.
• A record whose version is written without the events it counts moves ahead of the head, and that is the one way to open a gap. Write the record’s version only in the append of those events.

---

Step 24: Sealing a Stream

A table restore makes a stream’s version numbers reusable. A writer that loaded version 10 before the restore can append onto a different version 10 written after it. One way to close that hole is to key streams by generation ({ accountId, generation }): after a restore, writers move to the next generation and nobody writes to the old one again.

One gap is left. A stale writer can still succeed in appending to the old generation when the restored head happens to equal its expectedVersion. Nobody reads that generation any more, so the write is silently lost. Sealing closes the gap. A sealed stream accepts no further events, and an append that tries fails with StreamSealed.

The library has no notion of generations. The generation is just a stream id field, and the chain of generations is an item the application owns:

ledger.ts

// An account's ledger, keyed by generation. After a table restore, writers
// move to the next generation; sealing the old one makes sure nobody can
// append to it again.
class LedgerOpened extends Schema.TaggedClass<LedgerOpened>()("LedgerOpened", {
  balance: Schema.Number,
  // The generation this one continues, and the version its balance was read at.
  continuesFrom: Schema.optionalKey(
    Schema.Struct({ generation: Schema.String, version: Schema.Number }),
  ),
}) {}

class Deposited extends Schema.TaggedClass<Deposited>()("Deposited", {
  amount: Schema.Number,
}) {}

class Withdrawn extends Schema.TaggedClass<Withdrawn>()("Withdrawn", {
  amount: Schema.Number,
}) {}

class LedgerClosed extends Schema.TaggedClass<LedgerClosed>()("LedgerClosed", {
  reason: Schema.String,
  // The generation that carries the ledger on, when there is one.
  continuedIn: Schema.optionalKey(Schema.String),
}) {}

type LedgerEvent = LedgerOpened | Deposited | Withdrawn | LedgerClosed

// The application's chain index: which generation is current, and where each
// closed one ends. Every writer of it increments `revision`.
class LedgerChain extends Schema.Class<LedgerChain>("LedgerChain")({
  accountId: Schema.String,
  current: Schema.String,
  revision: Schema.Number,
  generations: Schema.Array(
    Schema.Struct({
      generation: Schema.String,
      validThrough: Schema.optionalKey(Schema.Number),
    }),
  ),
}) {}

const LedgerChains = Entity.make({
  model: LedgerChain,
  entityType: "LedgerChain",
  primaryKey: {
    pk: { field: "pk", composite: ["accountId"] },
    sk: { field: "sk", composite: [] },
  },
})

const LedgerTable = Table.make({ schema: AppSchema, entities: { LedgerChains } })

const Ledgers = EventStore.makeStream({
  table: LedgerTable,
  streamName: "Ledger",
  events: [LedgerOpened, Deposited, Withdrawn, LedgerClosed],
  streamId: { composite: ["accountId", "generation"] },
})

type LedgerState = { readonly balance: number }
type LedgerCommand = { readonly _tag: "Deposit"; readonly amount: number }

const ledgerDecider: EventStore.Decider<LedgerState, LedgerCommand, LedgerEvent> = {
  initialState: { balance: 0 },
  decide: (command) => Effect.succeed([new Deposited({ amount: command.amount })]),
  evolve: (state, event) => {
    if (event._tag === "LedgerOpened") return { balance: event.balance }
    if (event._tag === "Deposited") return { balance: state.balance + event.amount }
    if (event._tag === "Withdrawn") return { balance: state.balance - event.amount }
    return state
  },
}

Sealing with an append

append(..., { seal: true }) marks the append’s last event as the stream’s final event. The sealing event is your own event, and the library adds a marker attribute (__edd_sealed__) to its item:

const ledgers = yield* EventStore.bind(Ledgers)
const closing = { accountId: "acc-9", generation: "1" }
yield* ledgers.append(
  closing,
  [new LedgerOpened({ balance: 0 }), new Deposited({ amount: 40 })],
  0,
)

// The account closes for good: this append's last event is the stream's last.
const sealedAt = yield* ledgers.append(
  closing,
  [new Withdrawn({ amount: 40 }), new LedgerClosed({ reason: "account closed" })],
  2,
  { seal: true },
)
// → version 4; event 4 carries the seal marker

• The seal version S is expectedVersion + events.length. seal: false or no seal does nothing.
• seal: true with zero events, or a seal that is not a boolean, fails with ValidationError (operation: "EventStore.append.seal") before anything is written.
• On a stream with snapshot: { mode: "inline" } and no every, a sealing append must also pass snapshot. The snapshot it writes carries the marker too, so a snapshot read sees the seal (see the head-proof rules below). commandHandler always attaches the snapshot on those streams.
• seal works with every other append option, including idempotency, additionalItems, metadata, eventTimestamps and every head proof. measureAppend accepts it and counts the marker.
• A seal is permanent. There is no unseal.

The seal costs nothing extra. Every append after version 0 already writes a version-contiguity ConditionCheck on the event at expectedVersion. Its condition becomes attribute_exists(pk) AND attribute_not_exists(__edd_sealed__), so a writer that loaded after the seal fails it. A writer that loaded before the seal fails as it always did, because its next version is taken by the sealing event. Which error the writer gets is diagnosed from the cancellation reasons. Nothing is pre-read.

Writer	Fails at	Error
Loaded after the seal (expectedVersion = S)	The contiguity check on S	StreamSealed
Loaded just before the seal (S - 1)	Its first event put, at S	StreamSealed
Further behind (K < S - 1)	Its first event put, at K + 1	VersionConflict; a handler’s retry reloads and gets StreamSealed
expectedVersion = 0	Its first event put, at 1	VersionConflict, or StreamSealed when S = 1
headProof: "snapshot" at S	The snapshot proof	StreamSealed (the snapshot is marked)
headProof: "none" at S - 1	Its first event put, at S	StreamSealed
headProof: "none" at S	Nothing: the caller must refuse	Its event lands past the seal (see below)
Redelivered sealing command (same commandId)	The idempotency sentinel	DuplicateCommand

StreamSealed carries streamName, streamId and sealedVersion. It ranks after DuplicateCommand (a redelivery is the truer answer) and ahead of VersionConflict.

Reading the seal

const closed = yield* ledgers.readLatest(closing, { consistentRead: true })
// → closed.seal: the LedgerClosed event at version 4, closed.version 4

const closedEvents = yield* ledgers.read(closing, { consistentRead: true })
const final = closedEvents.at(-1)
// → final.sealed === true; no other event carries `sealed`

• StreamEvent.sealed is true on the sealing event and absent on every other event. read, readFrom, query.events, stream-index reads and decodeStreamImage all report it. A catch-up reader already past S gets [] from readFrom, and learns of the seal from the sealing event it already consumed, or from readLatest.
• LatestState.seal is the sealing event, in every readLatest mode, honouring consistentRead. With an inline snapshot taken at S, the sealing event is in seal but not in events, because events holds only what follows the snapshot, and that snapshot already folds it in. With verifySnapshot: false, a marked snapshot costs one more GetItem, of the sealing event, and only on a sealed stream.
• seal.version < version means events were written past the seal (see one write past the seal).
• currentVersion returns a number and does not report the seal. Neither does a projection: "withoutMetadata" stream index, because its INCLUDE list is fixed and does not name the marker; "all" and "keys" indexes do report it. Read the stream (readLatest) when you need the seal.

Handling StreamSealed

StreamSealed is terminal: no append to that stream can ever succeed again. commandHandler never retries it, and its load reports the seal before decide runs, so a sealed stream spends no transaction:

const handleLedger = EventStore.commandHandler(ledgerDecider, ledgers, { retry: 3 })

const outcome = yield* handleLedger(closing, { _tag: "Deposit", amount: 10 }).pipe(
  Effect.map((result) => `deposited at v${result.version}`),
  Effect.catchTag("StreamSealed", (sealed) =>
    Effect.succeed(`refused: ${sealed.streamId} is sealed at v${sealed.sealedVersion}`),
  ),
)
// → "refused: acc-9#1 is sealed at v4": seen on load, before decide, never retried

// A raw append at the head fails the same way.
const late = yield* ledgers
  .append(closing, [new Deposited({ amount: 10 })], 4)
  .pipe(Effect.flip)
// → StreamSealed { sealedVersion: 4 }

• The handler checks after every load. With a commandId, it reports DuplicateCommand first when that command already committed.
• StreamSealed outranks an If-Match: a call with an expectedVersion on a sealed stream gets StreamSealed, not VersionConflict, because the caller’s right move is to the successor stream, not a retry.
• A no-op decision is refused too. A sealed stream accepts no commands.
• When the load finds events after the seal, the handler still fails with StreamSealed (sealedVersion is the seal’s) and logs a warning naming the versions past it. Those are writes the application must recover.
• With loadState, the handler learns of the seal from your record: return sealedVersion from loadState. It may be version + 1 (see the record-proof rules below). A sealedVersion that is not a non-negative integer, or is below version, fails with ValidationError.

A generation changeover

Closing generation N and opening N + 1 is one transaction: seal N at the version the carry-over read, open N + 1 with an event that links back to it, and move the chain index. stream.appendOp(...) builds another stream’s append as an item of an append’s additionalItems, which commits if and only if that append does:

// Close generation `from`, open `to`, and move the chain index: one transaction.
const changeover = (accountId: string, from: string, to: string) =>
  Effect.gen(function* () {
    const closingGeneration = { accountId, generation: from }
    // ONE read gives the carried-over state AND the version the seal goes at.
    const latest = yield* ledgers.readLatest(closingGeneration, { consistentRead: true })
    if (latest.seal !== undefined) return latest.seal.version // already closed
    const carried = EventStore.fold(ledgerDecider, latest.events)

    // The chain index, read-modify-write: the whole item, rebuilt from this
    // read, replaced only if no writer has changed it since.
    const chain = yield* Entity.consistentRead(LedgerChains.get({ accountId }))
    const nextChain = LedgerChains.put({
      accountId,
      current: to,
      revision: chain.revision + 1,
      generations: [
        ...chain.generations.map((entry) =>
          entry.generation === from ? { ...entry, validThrough: latest.version + 1 } : entry,
        ),
        { generation: to },
      ],
    }).pipe(LedgerChains.condition({ revision: chain.revision }))

    yield* ledgers.append(
      { accountId, generation: to },
      [
        new LedgerOpened({
          balance: carried.balance,
          continuesFrom: { generation: from, version: latest.version },
        }),
      ],
      0,
      {
        additionalItems: [
          // Generation `from` sealed at the version its state was read at.
          ledgers.appendOp(
            closingGeneration,
            [new LedgerClosed({ reason: "restored", continuedIn: to })],
            latest.version,
            { seal: true },
          ),
          nextChain,
        ],
      },
    )
    return latest.version + 1
  }).pipe(
    // A stale append on `from`, or another writer of the chain index, since
    // the reads: read both again, carrying the stale append forward.
    Effect.retry({
      times: 3,
      while: (error) =>
        error._tag === "VersionConflict" || error._tag === "AdditionalItemConditionFailed",
    }),
  )

const sealedVersion = yield* changeover("acc-1", "1", "2")
// → 3: generation 1 ends at its LedgerClosed event; generation 2 opens at
//   version 1 with balance 250, continuing from generation 1 at version 2

The seal’s expectedVersion is the version the carried-over state was read at. It comes from the same readLatest result as the state, and it travels in the opening event’s back link (continuesFrom.version). Never re-read the old generation’s head to get it. Suppose the state is read at 2, a stale writer then lands event 3, and the seal is built from a fresh read of the head (3). The transaction opens generation 2 with state through version 2, and seals generation 1 at 4. Event 3 is in neither generation, and both writers were told they succeeded. With the carry-over version, the seal’s own check fails instead (VersionConflict naming generation 1), nothing is written, and the retry re-reads and carries event 3 forward.

The chain index is a read-modify-write under a revision. Updates are not accepted in additionalItems, so the closer reads the item, builds the whole new item from that read with revision + 1, and puts it conditioned on the revision it read (create() for the first generation). A condition on only the field you change is not enough: a put replaces the whole item, and would silently drop a concurrent edit to any other field. Every writer of the item must increment revision. An entity with timestamps must re-supply createdAt from the read, or the put resets it.

How the transaction fails, with nothing written in each case:

What went wrong	Error
The new generation is already open	VersionConflict naming the new generation
A stale append landed on the old generation since the read	VersionConflict naming the old generation: read again
Another closer already sealed the old generation	StreamSealed naming the old generation
The chain index changed since the read	AdditionalItemConditionFailed (indices: [1]): read again

The appending stream’s own failures are judged first, then each nested append’s, in order, then the other additional items.

• A nested append is a full append of its stream: its own contiguity check, event puts, seal marker, and an inline snapshot when you pass snapshot. It takes seal, snapshot, metadata / eventMetadata and eventTimestamps, but not additionalItems, idempotency or headProof (refused at the type level, and with ValidationError, operation: "EventStore.appendOp", at runtime). Zero events are refused the same way. A nested op cannot be a headProof: { additionalItem } target.
• A record the old generation’s writers load goes in the outer append. When its writers use loadState, or a record head proof, the changeover’s outer additionalItems also writes that record, beside the appendOp, with sealedVersion set (and on record-proof streams, the version kept at S - 1, as below). Without it, a handler whose record is more than one version below the seal loops on VersionConflict until its retry schedule ends, instead of reporting the seal.
• A handler on the new generation can carry the old generation’s sealing appendOp in a function-form additionalItems. It must take the version from the decision (for example decision.events[0].continuesFrom.version), never from its own read of the old head. A VersionConflict naming the old generation is retried by the handler’s retry schedule, even under an If-Match on the new one, and is never taken for a lagging snapshot. Read the old generation inside decide through a bound stream, so each retried attempt reads it again and the opening event and the seal’s version move together.
• One client per append. A nested op’s items are sent in the outer append’s single TransactWriteItems, through the outer append’s DynamoClient. A bound stream’s op resolves its own table name from its captured context, so a stream on another table needs no extra layer, but its captured client is never used. Every table must be reachable through the outer client.
• measureAppend refuses additionalItems, so it cannot measure a whole changeover. Measure each append and add them up.
• Transaction.transactWrite does not accept a stream op: its type rejects it, and at runtime it fails with ValidationError.

Atomic multi-stream append

appendOp does not need a seal. Without one, several streams’ events commit together or not at all:

// A transfer: both ledgers' events commit together, or neither does.
const transfer = yield* ledgers.append(
  { accountId: "acc-1", generation: "2" },
  [new Withdrawn({ amount: 75 })],
  1,
  { additionalItems: [ledgers.appendOp(savings, [new Deposited({ amount: 75 })], 1)] },
)
// → acc-1 generation 2 at version 2, acc-2 generation 1 at version 2

Each stream’s version is checked independently, and a stale expectedVersion on either fails the whole append (VersionConflict naming that stream). Don’t nest an append to the appending stream itself: the two appends either touch one item twice, which fails with ValidationError (DynamoDB allows one operation per item in a transaction), or conflict.

additionalItems is wider now

append’s additionalItems, and commandHandler’s AdditionalItemsInput, take AppendWriteOp (TransactWriteOp | StreamAppendOp). Passing a TransactWriteOp[] in still compiles. Code that reads one of those types and forwards the value to Transaction.transactWrite stops compiling: annotate the value as ReadonlyArray<TransactWriteOp>.

Head proofs and sealed streams

A head proof replaces the contiguity check, so the seal’s clause in that check never runs on that path. Each proof has its own rule.

headProof: "snapshot". The sealing append’s inline snapshot carries the marker, and the snapshot proof becomes asOfVersion = expectedVersion AND attribute_not_exists(__edd_sealed__). That is why a sealing append on an inline-without-every stream must pass snapshot: the snapshot at S is then the marked one, and it is what an unverified (verifySnapshot: false) load reads. Seal through a definition that shares the writers’ snapshot config. A closer often runs in another process, with its own definition of the stream. If that definition has no snapshot config (or uses every), its seal is accepted without a snapshot, the snapshot stays unmarked below S, and an unverified readLatest does not report the seal. Writers are still refused: their append fails, and the handler’s verified reload reports StreamSealed.

headProof: { additionalItem: n }. The library can’t enforce a seal through your record without a pre-read. Make every record-proof writer’s first event put land on the sealing event instead:

• Seal with exactly one event, so S = E + 1, where E is the record’s version.
• Write the record in the sealing transaction without advancing its version: it stays at E = S - 1, with your sealed flag and sealedVersion: S for loadState. Condition the record on version = E.
• Every record-proof writer then loads E and appends at expectedVersion = E. Its record condition holds, and its first event put at S hits the sealing event: StreamSealed. A handler whose loadState returns sealedVersion fails before decide.

Advancing the record to S lets a writer that loads it append at S: no contiguity check is written, its record condition holds, and event S + 1 lands past the seal. Advance it only if every writer’s record condition also tests your sealed flag. A seal of more than one event leaves the record more than one below S, where writers get VersionConflict and a handler loops; keep record-proof seals to one event.

headProof: "none". There is no check and no proof, so the library does not enforce the seal on that writer’s append: refusing to append when the last event it read is sealed is the caller’s duty (see Appending without the check). A writer that read the head just before a one-event seal is still refused with StreamSealed, because its first event put lands on the sealing event. One further behind gets VersionConflict, and learns of the seal when it reloads.

writeSnapshot must not target a sealed head with an unmarked snapshot

writeSnapshot is one plain conditional PutItem. It does not know whether the stream is sealed, and it never writes the marker. A job that snapshots streams should skip sealed ones (readLatest(...).seal is present), or seal through a definition that shares the writers’ inline snapshot config: the sealing append then writes the marked snapshot at S, and the monotonic writeSnapshot never replaces it.

The residual, when all three happen: a closer on a definition without the inline-without-every snapshot config seals at S, leaving the snapshot unmarked below S; a writeSnapshot(id, state, S) then puts an unmarked snapshot at S; and a writer with headProof: "snapshot" appends at S. Its proof holds on the unmarked snapshot, no contiguity check is written, and event S + 1 lands and is reported as success. A verified readLatest still reports the seal (seal.version < version).

Appending without the check: headProof: "none"

Some writers already know what the contiguity check would prove. A deposit that reads the ledger in the same request appends at the version of the last event it just read, and that version is never ahead of the head: events are never deleted, so the head only grows, and the version of an event read from the stream can be stale but never ahead. For such a writer the check is pure cost. headProof: "none" leaves it out and writes no proof item, and a one-event append is then a single PutItem:

// A deposit reads the ledger in the same request, so it already knows what
// the contiguity check would prove: the version of an event read from the
// stream is never ahead of the head. It takes over the check's other duty,
// the seal: it refuses when the last event it read is the sealing one.
const depositAtHead = (id: { accountId: string; generation: string }, amount: number) =>
  Effect.gen(function* () {
    const last = (yield* ledgers.read(id, { consistentRead: true })).at(-1)
    if (last?.sealed === true) {
      return yield* new StreamSealed({
        streamName: "Ledger",
        streamId: last.streamId,
        sealedVersion: last.version,
      })
    }
    return yield* ledgers.append(id, [new Deposited({ amount })], last?.version ?? 0, {
      headProof: "none",
    })
  })

const deposited = yield* depositAtHead(savings, 25)
// → version 3: one event, no check, sent as a single PutItem

// A writer that read the head before another one appended still loses the
// race: its event put lands on the taken version.
const lostRace = yield* ledgers
  .append(savings, [new Deposited({ amount: 5 })], 2, { headProof: "none" })
  .pipe(Effect.flip)
// → VersionConflict { expectedVersion: 2 }

// On a sealed ledger the caller's own read refuses, before anything is sent.
const refused = yield* depositAtHead(closing, 10).pipe(Effect.flip)
// → StreamSealed { sealedVersion: 4 }

The caller guarantees the head. It takes over both duties of the check:

1. No gap. expectedVersion must never be ahead of the head. An ahead version writes its events past a gap that nothing fills. The version of an event read from the stream (read, readFrom, currentVersion, or readLatest on a stream without a snapshot config, consistent or not) always meets this. With a snapshot, readLatest can return the snapshot’s asOfVersion, which is only as trustworthy as whoever called writeSnapshot: it does not check that the event at that version exists. A version from anywhere else (a cache, a client’s If-Match, arithmetic) does not meet it: keep the check.
2. The seal. Refuse to append when the last event read is sealed (StreamEvent.sealed, or LatestState.seal). Nothing else stops an append at the seal’s version: its event lands past the seal and reopens the stream for every writer (see One write past the seal reopens the stream). A writer that read the head just before a one-event seal (expectedVersion = S - 1) still fails with StreamSealed, because its first event put lands on the sealing event. A writer further behind, or just before a seal of several events (its first put lands on one of the seal’s unmarked events), gets VersionConflict, and must apply this duty to what it reads when it reloads.

What stays the same:

• Every event put is still conditioned on attribute_not_exists(pk), so a stale expectedVersion (a lost race) is a VersionConflict with nothing written.
• "none" combines with every other append option: additionalItems, idempotency, snapshot (which stays unconditional), seal, metadata and timestamps. Any of them makes the request more than one item, so it is a TransactWriteItems, without the check.
• measureAppend accepts it and counts no check, as for any head proof.

Where it is not accepted:

• commandHandler. Its headProof option excludes "none" from the type, and an untyped "none" fails every call with ValidationError (operation: "EventStore.commandHandler.headProof") before anything is read or written. Without a head proof the handler keeps the check.
• appendOp. A nested append refuses every head proof.

The single-item PutItem

Whenever an append’s request would be exactly one item, its one event, the library sends a PutItem with that item, its condition and ReturnValuesOnConditionCheckFailure: ALL_OLD. That covers the first event of every stream (expectedVersion: 0, through append or commandHandler) as well as one-event "none" appends. One item commits atomically either way, so only the cost changes. Failures map to the transaction’s errors: StreamSealed or VersionConflict from the returned item, a ValidationError naming the event for an item or key DynamoDB rejects, and TransactionCancelled (operation: "PutItem") with the reason a transaction gives its item (ProvisionedThroughputExceeded, ThrottlingError, ItemCollectionSizeLimitExceeded or TransactionConflict). The AWS SDK retries a throttled PutItem before it fails, so fewer throttling errors reach you.

A transaction is idempotent under the AWS SDK’s retries: the SDK gives it a ClientRequestToken, and DynamoDB answers a resend of a committed transaction with success. A PutItem has no token, so the event carries one of its own, a random __edd_wt__ attribute. When a put commits but its response is lost and the SDK resends it, the resend fails its condition, the returned item carries the writer’s own token, and the append succeeds as the transaction would. Another writer’s event at that version, even one with the same content, has a different token, and is a VersionConflict. The token adds 42 bytes to the item, and readers ignore it. A nested appendOp always rides in its outer append’s transaction.

Side writes still land

An append with no events (only additionalItems and/or an idempotency sentinel) writes no contiguity check, so it still lands on a sealed stream. It adds nothing to the stream. A seal stops events, not side writes.

One write past the seal reopens the stream

Upgrade every writer before sealing

A writer on a library version before 1.33.0 writes the old contiguity check, and can append past a seal. Older readers ignore the marker.

The seal is enforced on the event at a writer’s expectedVersion (or the one after it), not on the stream. If any write gets past it (a pre-1.33 writer, a record-proof writer whose record was advanced to S, a headProof: "none" writer that does not refuse on the seal it read, or the writeSnapshot residual above) and lands event S + 1, that event carries no marker. Every writer that then loads the head at S + 1 passes the check and appends, and those events are on a generation nobody reads.

The write side can’t stop it without a pre-read. The read side reports it wherever the sealing event is among the items a read already fetches: readLatest sets seal from any marked event it reads, not only the newest, and the handler refuses to decide past a seal and logs the versions after it. A verified readLatest with an inline snapshot reads the snapshot and only a few events, so once enough events follow the seal, only a full read finds it. So: upgrade every writer first, keep a record-proof stream’s record at S - 1, make every headProof: "none" writer refuse on a sealed last event, and never writeSnapshot a sealed head with an unmarked snapshot.

---

Running the Example

The complete runnable example is at examples/event-sourcing.ts in the repository.

docker run -d -p 8000:8000 amazon/dynamodb-local

npx tsx examples/event-sourcing.ts

The client reads its endpoint from the DYNAMODB_ENDPOINT config value, defaulting to http://localhost:8000, so DYNAMODB_ENDPOINT=http://localhost:18000 npx tsx examples/event-sourcing.ts targets another DynamoDB Local.

Layer Setup

The command handler in Step 20 mints ids through Effect’s Crypto service, which Effect does not provide by default. The example backs it with the Web Crypto API. (DynamoClient.make bundles a Crypto for its own bound operations only, such as generatedId; your own effects need one in their layer.)

main.ts

// Effect core ships no default Crypto service. This one is backed by the Web
// Crypto API (`globalThis.crypto`): Node 19+ (Node 18 only with
// --experimental-global-webcrypto), browsers and edge runtimes.
const WebCrypto = Layer.succeed(
  Crypto.Crypto,
  Crypto.make({
    randomBytes: (size) => globalThis.crypto.getRandomValues(new Uint8Array(size)),
    digest: (algorithm, data) =>
      Effect.tryPromise({
        try: () => globalThis.crypto.subtle.digest(algorithm, new Uint8Array(data)),
        catch: (cause) =>
          PlatformError.badArgument({ module: "Crypto", method: "digest", cause }),
      }).pipe(Effect.map((digest) => new Uint8Array(digest))),
  }),
)

main.ts

const AppLayer = Layer.mergeAll(
  DynamoClient.layerConfig({
    region: Config.succeed("us-east-1"),
    endpoint: Config.String("DYNAMODB_ENDPOINT").pipe(Config.withDefault("http://localhost:8000")),
    credentials: Config.succeed({ accessKeyId: "local", secretAccessKey: "local" }),
  }),
  EventsTable.layer({ name: "event-sourcing-example" }),
  ScorecardTable.layer({ name: "event-sourcing-scorecards" }),
  AuditedScorecardTable.layer({ name: "event-sourcing-audited-scorecards" }),
  SeriesTable.layer({ name: "event-sourcing-series" }),
  RecordTable.layer({ name: "event-sourcing-records" }),
  LedgerTable.layer({ name: "event-sourcing-ledgers" }),
  WebCrypto,
)

const main = program.pipe(Effect.provide(AppLayer))

Effect.runPromise(main).then(() => console.log("\nDone."))

---

Key Takeaways

Concept	How it’s used
Event stream	EventStore.makeStream binds event schemas to a table with stream ID composites
Decider pattern	initialState + decide (command validation) + evolve (state transition)
Command handler	EventStore.commandHandler wires decider to stream with optimistic concurrency
Read operations	read (all events), readFrom (from version), currentVersion (latest version number)
Fold	EventStore.fold reconstructs state from events using the decider’s evolve function
Query combinators	Query.reverse, Query.limit, Query.collect work on event stream queries
Domain errors	Tagged errors (AlreadyStarted, NotStarted, AlreadyEnded) for precise error handling
Snapshots	snapshot: { schema, every } caches a fold of the stream; the handler replays only the delta
Snapshot primitives	readSnapshot / writeSnapshot for manual cadence; monotonic writes, never a cache regression
Retry	{ retry } re-runs the full read-decide-append cycle on VersionConflict only
Atomic side writes	append(..., { additionalItems }) commits Entity.put / Entity.delete / bound db.entities.* builders / Transaction.check ops with the events
Position-aware errors	VersionConflict vs. AdditionalItemConditionFailed vs. DuplicateCommand — you always know whose condition failed
Command idempotency	At-least-once by default; commandHandler(..., { idempotency }) + a commandId makes it exactly-once via a dedup sentinel
Consistent reads	{ consistentRead: true } on read / readFrom / currentVersion; the command handler loads consistently by default
If-Match	handle(id, command, { expectedVersion }) fails with VersionConflict (carrying actualVersion) before decide runs; never retried
Inline projections	additionalItems: (decision) => items derives side writes from the post-fold state, in the same transaction
Inline snapshots	snapshot: { schema, mode: "inline" } writes the snapshot in the append transaction; readLatest loads state in one Query, or, with verifySnapshot: false (inline without every), one GetItem of the snapshot with verified fallbacks
Stepped commands	A command too large for one transaction is split by the application into fixed-size commands chained by expectedVersion — each one atomic
Stream indexes	makeStream({ indexes }) + readIndex / query.index read sub-streams by a derived key; indexDefinitions builds the LSI/GSI definitions
Event metadata	makeStream({ metadata }) declares the schema; append(..., { metadata }) shares one value, { eventMetadata } gives each event its own entry; on a handler, eventMetadata: (decision) => entries derives them after decide
Event timestamps	append(..., { eventTimestamps }) writes each event’s own DateTime as its timestamp (undefined keeps the append time); on a handler, eventTimestamps: (decision) => timestamps. version, not timestamp, orders a stream
Index projection	projection: "withoutMetadata" (INCLUDE without metadata, read in one query) or "keys" (INCLUDE ["__edd_e__"], events fetched from the table) on a stream index; the deployed Projection must match indexDefinitions
Packed storage	makeStream({ storedAs: { data: DynamoModel.Packed(...) } }) and snapshot.storedAs store data and state as JSON, compressed above a threshold; readers decode by the stored form, so a stream can mix encodings
Shared indexes	A GSI index’s partition (from the stream id or the event) lets several streams write one partition, and its snapshot puts each stream’s snapshot there too; indexDefinitions emits the shared GSI once; readIndex takes the partition
Head proof	headProof: "snapshot" (inline without every) or { additionalItem: n } names an item whose own condition proves the head, in place of the version-contiguity ConditionCheck; its failure is a VersionConflict
No head proof	append(..., { headProof: "none" }) leaves the check out for a caller whose expectedVersion was read from the stream; the caller refuses on a seal it read, and stale versions still conflict
Single-item append	An append of one event and nothing else is a conditional PutItem (1 WCU per KB) instead of a transaction (2), with the same errors
State from your own record	commandHandler(..., { loadState }) loads state and head from the application’s record instead of the stream; with headProof: { additionalItem }, the record written with every append is also the version guard
Sealing	append(..., { seal: true }) makes the append’s last event the stream’s final one; a later append fails with StreamSealed (terminal, never retried); readLatest returns the sealing event as seal, and StreamEvent.sealed marks it
Generation changeover	stream.appendOp(...) in another append’s additionalItems seals the old generation at the version its carried-over state was read at, in the transaction that opens the next one; the chain index is a whole-item put conditioned on a revision
Atomic multi-stream append	appendOp without seal: several streams’ events commit together or not at all

What’s Next?

• Modeling Guide — Deep dive into models, schemas, tables, and entities
• Queries Guide — Query combinators, pagination, and filtering
• Packed Storage Guide — JSON and compressed storage for event data and snapshot state
• Projections from DynamoDB Streams — Read models folded from the change feed in version order, gap-filled from the event store
• Example: Unique Constraints — Globally unique fields and idempotency keys
• Example: Human Resources — Single-table design with collections and transactions