Your first aggregate

Part 2 of 5 of CQRS with effect-cqrs. November 2026.

One book stands in a glowing frame on the librarian's desk; request cards pass through it on a belt, and one is refused with a red mark

In part 1 we wrote event sourcing by hand: a list of events, an evolve function that folds them into a state, and a lend function that decides on new events. This post turns that book into an aggregate with effect-cqrs, and runs commands against a store.

effect-cqrs is built on Effect 4, so commands are effects with typed errors, and storage, clocks and configuration are services you provide. It needs Node 24 or later:

pnpm add effect-cqrs effect

All the code of this series lives in examples/blog, and its tests run every snippet you see here.

Events are schemas

The events of part 1 were a TypeScript union. Here they are a Schema.TaggedUnion, which gives the same type plus a way to check, encode and decode values:

export const BookEvent = Schema.TaggedUnion({
  BookAdded: { title: Schema.NonEmptyString, author: Schema.NonEmptyString },
  BookLent: { loanId: Schema.String, memberId: Schema.String, dueOn: Schema.DateTimeUtc },
  BookReturned: { loanId: Schema.String, memberId: Schema.String, late: Schema.Boolean }
})

The store keeps events as JSON, and the schema says how to get there and back. dueOn is a DateTime in your code and an ISO string in the database, and every event read back is checked against the schema. BookEvent.cases.BookLent.make({ ... }) builds an event and fills in its _tag.

The events now carry more than in part 1. A loan has an id, so that the same loan can be recognised when it comes again, and a due date. A return says whether it was late. Events are the only record of what happened, so it pays to put in them what you’ll want to know later.

Refusals are errors with a type

Part 1 refused with a thrown Error. Here each refusal is a class of its own:

export class BookAlreadyAdded extends Schema.TaggedError<BookAlreadyAdded>()("BookAlreadyAdded", {
  bookId: Schema.String
}) {}

export class BookNotFound extends Schema.TaggedError<BookNotFound>()("BookNotFound", {
  bookId: Schema.String
}) {}

export class BookAlreadyLent extends Schema.TaggedError<BookAlreadyLent>()("BookAlreadyLent", {
  bookId: Schema.String
}) {}

export class BookNotLent extends Schema.TaggedError<BookNotLent>()("BookNotLent", {
  bookId: Schema.String
}) {}

A Schema.TaggedError is an error with a _tag, so the caller can tell them apart and handle each one. As we’ll see in a moment, they also show up in the type of every command that can fail with them.

The aggregate

An aggregate puts three things together: its events, a projection that folds them into its state, and its commands. Here is the book:

/** How long a member may keep a book. */
export const loanPeriod = { days: 21 }

export interface BookState {
  readonly status: "unknown" | "available" | "lent"
  readonly title: string
  readonly author: string
  readonly loan: { readonly loanId: string; readonly memberId: string; readonly dueOn: DateTime.Utc } | null
}

export const Book = Aggregate.make<BookState>()({
  name: "book",
  events: BookEvent,
  initial: { status: "unknown", title: "", author: "", loan: null },
  projection: {
    BookAdded: (state, { author, title }) => ({ ...state, status: "available", title, author }),
    BookLent: (state, { dueOn, loanId, memberId }) => ({ ...state, status: "lent", loan: { loanId, memberId, dueOn } }),
    BookReturned: (state) => ({ ...state, status: "available", loan: null })
  },
  versions: bookVersions
})
  .command("add", {
    payload: Schema.Struct({ title: Schema.NonEmptyString, author: Schema.NonEmptyString }),
    handler: Effect.fnUntraced(function*(state, { author, title }, { aggregateId }) {
      if (state.status !== "unknown") return yield* new BookAlreadyAdded({ bookId: aggregateId })
      return BookEvent.cases.BookAdded.make({ title, author })
    })
  })
  .command("lend", {
    payload: Schema.Struct({ loanId: Schema.String, memberId: Schema.String }),
    handler: Effect.fnUntraced(function*(state, { loanId, memberId }, { aggregateId }) {
      if (state.status === "unknown") return yield* new BookNotFound({ bookId: aggregateId })
      // Lending it again for the same loan records nothing, so a retry is harmless.
      if (state.loan?.loanId === loanId) return []
      if (state.loan !== null) return yield* new BookAlreadyLent({ bookId: aggregateId })
      const dueOn = DateTime.add(yield* DateTime.now, loanPeriod)
      return BookEvent.cases.BookLent.make({ loanId, memberId, dueOn })
    })
  })
  .command("returnBook", {
    payload: Schema.Struct({}),
    handler: Effect.fnUntraced(function*(state, _, { aggregateId }) {
      if (state.loan === null) return yield* new BookNotLent({ bookId: aggregateId })
      const { dueOn, loanId, memberId } = state.loan
      const late = DateTime.isGreaterThan(yield* DateTime.now, dueOn)
      return BookEvent.cases.BookReturned.make({ loanId, memberId, late })
    })
  })

Read it against part 1:

projection is evolve, split into one function per event. Each gets the state and the event and returns the new state. If you add an event to BookEvent and forget its projection, the aggregate doesn’t compile.
Each .command is a decide. It has a payload schema, checked before the handler runs, and a handler that gets the current state, the payload and the aggregate’s id and version. It returns one event, an array of events, or [] to record nothing, or fails with one of the errors above.
The state type is given explicitly, as in Aggregate.make<BookState>(), so that status: "lent" stays the literal "lent" rather than widening to string.
Time comes from Effect’s clock. DateTime.now reads the Clock service, so a test can stop or move time without mocking anything. We do that in part 5.
versions comes in part 5, when the book’s first event changes shape.

Look at lend once more. Lending a book for a loan it is already lent for returns [] instead of failing. A retry of the same request is then harmless, which part 4’s saga will rely on.

Handlers have a few rules, and they all follow from one idea: a handler only decides.

It may run more than once. A command that loses a race with another process reloads the state and decides again, so a handler must not send an email or call a payment provider. Reactions to events belong to read models and sagas, the subject of parts 3 and 4.
It must not change the state it receives. Recent states are kept in memory between commands, so a projection or handler that mutates one corrupts the next command. Always return new objects.

Running commands

AggregateService.make turns the aggregate into a service with one typed function per command. We give it a name with Context.Service, so that the rest of the application can ask for Books, and provide an event store kept in memory:

export class Books extends Context.Service<Books, AggregateService.AggregateService<typeof Book>>()("library/Books") {
  static readonly layer = Layer.effect(Books, AggregateService.make(Book))
}

export const BooksInMemory = Books.layer.pipe(Layer.provide(EventStore.layerMemory))

Now we can lend a book:

yield* books.commands.add("book-1", { title: "Dune", author: "Frank Herbert" })
const lent = yield* books.commands.lend("book-1", { loanId: "loan-1", memberId: "alice" })
// lent.events: [{ _tag: "BookLent", loanId: "loan-1", memberId: "alice", dueOn: … }]
// lent.version: 2

What happens inside lend:

The payload is checked against the command’s schema.
The service loads the book’s events from the store and folds them into its state. It keeps recent states in memory, so the next command on book-1 only reads what is new.
The handler decides, and the new events are appended. The append says which version the decision was based on, here 1, and the store refuses it if someone else wrote to book-1 in between.
The command returns the events, the book’s new version and the event’s position in the whole store, which part 3 uses.

The result is a receipt: what was recorded, and where in the log. It is not the book as a screen would show it. That is a question, and questions go to state or to the read models of part 3. Keeping the two apart is Bertrand Meyer’s command-query separation, the idea CQRS takes its name from: asking a question should not change the answer.

Commands on one book from one process take turns, so they never race. Writers in different processes race optimistically: the loser gets the store’s refusal, reloads, decides again, and gives up with ConcurrencyConflict after a few tries.

Handling a refusal

The second loan of the same copy is refused. The error is in the command’s type, so Effect.catchTag knows which tags exist and what each carries:

const answer = yield* books.commands.lend("book-1", { loanId: "loan-2", memberId: "bob" }).pipe(
  Effect.as("Enjoy your book"),
  Effect.catchTag("BookAlreadyLent", ({ bookId }) => Effect.succeed(`${bookId} is out, try again later`))
)

Hover over books.commands.lend in an editor and its error channel reads BookNotFound | BookAlreadyLent | AggregateService.CommandFailure: the handler’s own errors, plus what any command can fail with, such as InvalidCommand for a bad payload, ConcurrencyConflict for a race lost too often, and errors of the store. Leave one out of a catchTag and it stays in the type, where the caller can see it.

JSON from the outside

books.commands is for TypeScript callers. An HTTP or RPC endpoint receives JSON it knows nothing about, and hands it to dispatch, which checks the envelope and decodes the payload with the command’s schema before anything runs:

// What an HTTP handler hands over: JSON it knows nothing about.
yield* books.dispatch({
  aggregateId: "book-2",
  type: "add",
  payload: { title: "Neuromancer", author: "William Gibson" }
})

// An empty title never reaches the handler.
const invalid = yield* Effect.flip(books.dispatch({ aggregateId: "book-3", type: "add", payload: { title: "" } }))
// invalid._tag: "InvalidCommand"

An ISO date in the JSON becomes a DateTime, and a payload that doesn’t match fails with InvalidCommand listing what was wrong. The handler only ever sees valid input.

Reading

state folds an aggregate’s events into its state, and history lists them:

const state = yield* books.state("book-1")
// { status: "lent", title: "Dune", author: "Frank Herbert", loan: { loanId: "loan-1", … } }

const history = yield* books.history("book-1")
// [{ version: 1, type: "BookAdded", timestamp: … }, { version: 2, type: "BookLent", … }]

state also takes a finishTime, which gives the state as it was at that moment: the time travel of part 1. history with { payload: true } includes each event.

To run all of this, provide the layer and run the effect:

Effect.runPromise(firstSteps.pipe(Effect.provide(BooksInMemory)))

What we have

Compared with the forty lines of part 1, the book now has validated input, typed refusals, a store with optimistic concurrency, and a clock it can be tested with. It still answers only one kind of question: what is the state of this book? To find every available book, we would have to load them all.

That is what read models are for, in part 3.

The commands guide covers what this post skipped: command ids for retries over the network, expectedVersion for edits based on what a user saw, metadata, snapshots and metrics.