Sagas

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

So far a member was just a string on a loan. The library has a rule about members too: a member may have at most three books at a time. Together with the book’s own rule, one member per copy, borrowing now has to satisfy two aggregates at once.

Why not one aggregate that holds both? Because an aggregate is the unit that commands take turns on. A “library” aggregate with every book and every member would make every loan in the building wait for every other one, and fold the whole library’s history to decide anything. Aggregates stay small, and a process that spans several of them is a saga: it reacts to an event and runs commands on other aggregates, possibly over a long time.

The member

The member is an aggregate like the book, with its own events:

examples/blog/library.ts

export const MemberEvent = Schema.TaggedUnion({
  MemberJoined: { name: PersonalData.field(Schema.NonEmptyString, { redacted: "[forgotten]" }) },
  LoanRequested: { loanId: Schema.String, bookId: Schema.String },
  LoanConfirmed: { loanId: Schema.String },
  LoanCancelled: { loanId: Schema.String, reason: Schema.String },
  LoanClosed: { loanId: Schema.String }
})

PersonalData.field marks the name as personal data, so that a member can be forgotten. Everywhere in the code it is a plain string. Part 5 explains how it works.

examples/blog/library.ts

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

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

export class LoanLimitReached extends Schema.TaggedError<LoanLimitReached>()("LoanLimitReached", {
  memberId: Schema.String,
  limit: Schema.Int
}) {}

examples/blog/library.ts

/** How many books a member may have at once, counting the loans still being arranged. */
export const loanLimit = 3

export interface MemberState {
  readonly joined: boolean
  readonly name: string
  /** Open loans by id: requested until the book is lent, then active. */
  readonly loans: Readonly<Record<string, { readonly bookId: string; readonly status: "requested" | "active" }>>
}

const withoutLoan = (state: MemberState, loanId: string): MemberState => {
  const { [loanId]: _, ...loans } = state.loans
  return { ...state, loans }
}

export const Member = Aggregate.make<MemberState>()({
  name: "member",
  events: MemberEvent,
  initial: { joined: false, name: "", loans: {} },
  projection: {
    MemberJoined: (state, { name }) => ({ ...state, joined: true, name }),
    LoanRequested: (state, { bookId, loanId }) => ({
      ...state,
      loans: { ...state.loans, [loanId]: { bookId, status: "requested" } }
    }),
    LoanConfirmed: (state, { loanId }) => {
      const loan = state.loans[loanId]
      return loan === undefined
        ? state
        : { ...state, loans: { ...state.loans, [loanId]: { ...loan, status: "active" } } }
    },
    LoanCancelled: (state, { loanId }) => withoutLoan(state, loanId),
    LoanClosed: (state, { loanId }) => withoutLoan(state, loanId)
  }
})
  .command("join", {
    payload: Schema.Struct({ name: Schema.NonEmptyString }),
    handler: Effect.fnUntraced(function*(state, { name }, { aggregateId }) {
      if (state.joined) return yield* new MemberAlreadyJoined({ memberId: aggregateId })
      return MemberEvent.cases.MemberJoined.make({ name })
    })
  })
  .command("requestLoan", {
    payload: Schema.Struct({ loanId: Schema.String, bookId: Schema.String }),
    handler: Effect.fnUntraced(function*(state, { bookId, loanId }, { aggregateId }) {
      if (!state.joined) return yield* new MemberNotFound({ memberId: aggregateId })
      if (state.loans[loanId] !== undefined) return []
      if (Object.keys(state.loans).length >= loanLimit) {
        return yield* new LoanLimitReached({ memberId: aggregateId, limit: loanLimit })
      }
      return MemberEvent.cases.LoanRequested.make({ loanId, bookId })
    })
  })
  // The saga runs the next three. Each records nothing when there is nothing to do, so
  // running one twice is harmless.
  .command("confirmLoan", {
    payload: Schema.Struct({ loanId: Schema.String }),
    handler: (state, { loanId }) =>
      Effect.succeed(
        state.loans[loanId]?.status === "requested" ? MemberEvent.cases.LoanConfirmed.make({ loanId }) : []
      )
  })
  .command("cancelLoan", {
    payload: Schema.Struct({ loanId: Schema.String, reason: Schema.String }),
    handler: (state, { loanId, reason }) =>
      Effect.succeed(
        state.loans[loanId]?.status === "requested" ? MemberEvent.cases.LoanCancelled.make({ loanId, reason }) : []
      )
  })
  .command("closeLoan", {
    payload: Schema.Struct({ loanId: Schema.String }),
    handler: (state, { loanId }) =>
      Effect.succeed(state.loans[loanId]?.status === "active" ? MemberEvent.cases.LoanClosed.make({ loanId }) : [])
  })

A loan goes through two steps on the member. requestLoan takes one of the three places right away, as requested. The member can’t know yet whether the book is free, so the place is held until the book answers: confirmLoan makes it active, cancelLoan gives it back. Counting requested loans against the limit means three requests in a row can’t borrow four books.

The three commands at the end are the ones the saga runs, and each records nothing when there is nothing to do. Confirming a loan twice confirms it once. We’ll see why that matters below.

The process

Borrowing goes like this:

1. The member requests a loan: LoanRequested.
2. The saga sees the event and lends the book.
3. If the book was free, the saga confirms the loan. If it was out, it cancels the loan, and the member has the place back.

Here it is as a workflow:

examples/blog/library.ts

// One run per loan: lend the book and confirm the loan, or cancel it if the book is out.
export const Borrow = Saga.workflow("Borrow", {
  payload: { loanId: Schema.String, memberId: Schema.String, bookId: Schema.String },
  idempotencyKey: ({ loanId }) => loanId
})

export const BorrowWorkflow = Saga.toLayer(Borrow, ({ bookId, loanId, memberId }) =>
  Effect.gen(function*() {
    const books = yield* Books
    const members = yield* Members

    const lent = yield* Saga.command("lend", {
      error: Schema.Union([BookAlreadyLent, BookNotFound]),
      execute: books.commands.lend(bookId, { loanId, memberId })
    }).pipe(
      Effect.as(true),
      Effect.catchTag(["BookAlreadyLent", "BookNotFound"], (error) =>
        Saga.command("cancel", {
          execute: members.commands.cancelLoan(memberId, { loanId, reason: error._tag })
        }).pipe(Effect.as(false)))
    )

    if (lent) {
      yield* Saga.command("confirm", { execute: members.commands.confirmLoan(memberId, { loanId }) })
    }
  }))

Saga.workflow declares a workflow from effect/workflow, Effect’s durable workflow engine, with what one run is about, and idempotencyKey says what identifies a run. Starting the saga twice for the same loan joins the first run rather than lending the book twice.

Every command in the body goes through Saga.command, which makes it an activity: a step whose outcome the engine records. If the process restarts in the middle of a run, the engine replays the body and skips the steps it already has an outcome for. The error option lists the domain errors the saga reacts to. Their outcome is recorded too, so a replay takes the same branch. Any other failure ends the run as a defect.

Refusing to lend is part of the normal flow here, not an accident, so the saga handles it with catchTag and cancels the loan. For steps that have to be undone when a later one fails, the workflow also has withCompensation, see the sagas guide.

Bringing a book back is a second, shorter process: the book records BookReturned, and the member’s loan has to be closed so the place is free again.

examples/blog/library.ts

// When a book comes back, the loan is closed and the member has a place free again.
export const CloseLoan = Saga.workflow("CloseLoan", {
  payload: { loanId: Schema.String, memberId: Schema.String },
  idempotencyKey: ({ loanId }) => loanId
})

export const CloseLoanWorkflow = Saga.toLayer(CloseLoan, ({ loanId, memberId }) =>
  Effect.gen(function*() {
    const members = yield* Members
    yield* Saga.command("close", { execute: members.commands.closeLoan(memberId, { loanId }) })
  }))

The trigger

What starts the workflows is a trigger, made with Saga.make. It is a view model, the same as in part 3, whose handlers only start runs:

examples/blog/library.ts

export const Lending = Saga.make({
  name: "lending",
  aggregates: [Member, Book],
  handlers: {
    LoanRequested: ({ bookId, loanId }, { aggregateId }) =>
      Saga.start(Borrow, { loanId, memberId: aggregateId, bookId }),
    BookReturned: ({ loanId, memberId }) => Saga.start(CloseLoan, { loanId, memberId })
  }
})

Being a view model, it follows the log from its own checkpoint, catches up after downtime, and sees every event at least once. Its checkpoint is saved every 500 events, so after a crash the trigger can handle a few hundred events again and start their sagas a second time. Saga.start returns as soon as the engine has the run, so a slow saga never holds up the trigger.

Caution

At least once means a run can start again after a crash, and an activity cut off before its outcome was saved runs again. That is why lend, confirmLoan, cancelLoan and closeLoan record nothing when the work is already done: the commands a saga runs must be idempotent.

Wiring it together

Everything is a layer. The sagas sit on top of the command side, which sits on top of the storage:

examples/blog/library.ts

const Sagas = Saga.layer([Lending]).pipe(
  Layer.provideMerge(Layer.merge(BorrowWorkflow, CloseLoanWorkflow)),
  Layer.provide(WorkflowEngine.layerMemory)
)

// Everything but the storage, which the last post swaps for a database.
export const Application = Layer.mergeAll(Sagas, ViewModel.layer([CatalogView, LoansView, DirectoryView])).pipe(
  Layer.provideMerge(Layer.merge(Books.layer, Members.layer)),
  Layer.provideMerge(Layer.mergeAll(Catalog.layer, Loans.layer, Directory.layer))
)

// Events, checkpoints and the keys of personal data, all in memory.
export const InMemory = Layer.mergeAll(
  EventStore.layerMemory,
  Checkpoints.layerMemory,
  PersonalData.layerMemory().pipe(Layer.provide(PersonalData.layerMasterKeyEphemeral))
)

export const Library = Application.pipe(Layer.provideMerge(InMemory))

Application is everything but the storage, which part 5 swaps for a database. WorkflowEngine.layerMemory keeps runs in memory, which pairs with in-memory checkpoints: after a restart the trigger reads the log from the start and starts every saga again, and the idempotent commands make that harmless. In production, ClusterWorkflowEngine from effect/cluster keeps runs in SQL, and pairs with durable checkpoints.

Two members, one book

A saga finishes a little after the command that started it. To wait for the outcome, we subscribe to the member’s events and stop at the one that settles the loan:

examples/blog/library.ts

/** Waits until one of these events is stored for a member's loan, and the view models have seen it. */
export const loanReached = (memberId: string, loanId: string, ...eventTypes: ReadonlyArray<string>) =>
  Effect.gen(function*() {
    const store = yield* EventStore.EventStore
    const viewModels = yield* ViewModel.ViewModels
    const reached = yield* store.subscribe({ aggregateNames: ["member"], aggregateIds: [memberId], eventTypes }).pipe(
      Stream.filter(({ payload }) => Predicate.hasProperty(payload, "loanId") && payload.loanId === loanId),
      Stream.runHead
    )
    if (Option.isSome(reached)) yield* viewModels.caughtUp(reached.value.position)
    return Option.map(reached, ({ type }) => type)
  })

/** Waits until the saga has confirmed or cancelled a loan. */
export const settled = (memberId: string, loanId: string) =>
  loanReached(memberId, loanId, "LoanConfirmed", "LoanCancelled")

store.subscribe is a stream of past events followed by live ones, so the outcome is found whether it was stored before or after the call. Now Alice and Bob both want Dune:

examples/blog/library.ts

yield* books.commands.add("book-1", { title: "Dune", author: "Frank Herbert" })
yield* members.commands.join("alice", { name: "Alice Rossi" })
yield* members.commands.join("bob", { name: "Bob Bianchi" })

yield* members.commands.requestLoan("alice", { loanId: "loan-1", bookId: "book-1" })
const forAlice = yield* settled("alice", "loan-1")
// Option.some("LoanConfirmed")

yield* members.commands.requestLoan("bob", { loanId: "loan-2", bookId: "book-1" })
const forBob = yield* settled("bob", "loan-2")
// Option.some("LoanCancelled"): Alice has the book

Alice’s loan is confirmed. Bob’s request takes a place, the saga finds the book out, and cancels his loan, which gives the place back. Neither aggregate ever broke its own rule, and neither had to know about the other.

Why not let the member’s command lend the book itself, and skip the saga? The saga you skipped tries that, in two ways, and shows what happens to the data when a write fails in the middle.

When Alice brings the book back, the second workflow closes her loan:

examples/blog/library.ts

yield* books.commands.returnBook("book-1", {})
yield* loanReached("alice", "loan-1", "LoanClosed")
const alice = yield* members.state("alice")
// { joined: true, name: "Alice Rossi", loans: {} }

Following the thread

A saga turns one request into commands on several aggregates. effect-cqrs records how they are linked, on every event:

examples/blog/library.ts

const bobsHistory = yield* members.history("bob")
// LoanRequested and LoanCancelled share a correlationId, and LoanCancelled's
// causationId names the event that started the saga: "member/bob@2"

• The correlationId is shared by everything that follows from one original request. Bob’s request and its cancellation carry the same one, and so does the book’s BookLent that Alice’s request led to.
• The causationId names what directly caused a command. The cancellation was caused by the event member/bob@2, version 2 of Bob’s stream, his LoanRequested.

When you need to know why a loan was cancelled at 3 a.m., you follow the causation ids back to the request that started it. To join a request from outside, an HTTP request id for instance, wrap the command in Causation.withCausation.

Waiting inside a saga

A saga can also wait. A reminder when a loan is overdue would be a run that sleeps until the due date with DurableClock.sleep, then sends a reminder unless the book came back. A durable engine keeps the clock across restarts, even for three weeks. DurableDeferred waits for a signal instead, such as a later event. The tutorial’s shop races the two, to give a customer fifteen seconds to pay.

What we have

The library now has every piece of CQRS: aggregates that decide, read models that follow, and a saga that coordinates. What’s left is making sure it is right and keeping it right as it changes and grows. Part 5 covers testing, changing an event’s shape, moving to a database, and forgetting a member.