The bank demo

Open the live demoeffect-cqrs in your browser, on the in-memory event store. Nothing leaves the page.

The demo is the classic event sourcing example, a bank, built on the framework as it is: the same sources the tests run, bundled for the browser. Bank accounts are aggregates, three view models project them, and a saga moves money between accounts.

On the left, a customer app renders only the read models, so their lag behind the event store is visible. On the right, devtools slow each stage down, pause the command bus and step it, crash a projection, and travel back through the log.

To run it locally:

pnpm ui

The code quoted below comes from the demo’s sources when this site is built, so it can’t drift from what runs.

The account aggregate

An account’s balance is never stored. It is folded from the money that went in and out, so the events are the whole truth:

export const AccountEvent = Schema.TaggedUnion({
  AccountOpened: { owner: Schema.NonEmptyString, color: Schema.String, openedAt: Schema.DateTimeUtc },
  MoneyDeposited: { amount: Amount },
  MoneyWithdrawn: { amount: Amount },
  TransferSent: { transferId: Schema.String, to: Schema.String, amount: Amount },
  TransferReceived: { transferId: Schema.String, from: Schema.String, amount: Amount },
  TransferRefunded: { transferId: Schema.String, to: Schema.String, amount: Amount, reason: Schema.String },
  AccountClosed: { closedAt: Schema.DateTimeUtc }
})

The aggregate folds them into a state with a status, an owner and a balance, and its commands check that state before they decide:

export const Account = Aggregate.make<AccountState>()({
  name: "account",
  events: AccountEvent,
  initial: initialAccount,
  projection: {
    AccountOpened: (state, { color, owner }) => ({ ...state, status: "open", owner, color }),
    MoneyDeposited: (state, { amount }) => ({ ...state, balance: state.balance + amount }),
    MoneyWithdrawn: (state, { amount }) => ({ ...state, balance: state.balance - amount }),
    TransferSent: (state, { amount }) => ({ ...state, balance: state.balance - amount }),
    TransferReceived: (state, { amount, transferId }) => ({
      ...state,
      balance: state.balance + amount,
      received: [...state.received, transferId]
    }),
    TransferRefunded: (state, { amount }) => ({ ...state, balance: state.balance + amount }),
    AccountClosed: (state) => ({ ...state, status: "closed" })
  }
})
  .command("openAccount", {
    payload: Schema.Struct({ owner: Schema.NonEmptyString, color: Schema.String }),
    handler: Effect.fnUntraced(function*(state, { color, owner }, { aggregateId }) {
      if (state.status !== "none") return yield* new AccountAlreadyOpened({ aggregateId })
      return AccountEvent.cases.AccountOpened.make({ owner, color, openedAt: yield* DateTime.now })
    })
  })
  .command("deposit", {
    payload: Schema.Struct({ amount: Amount }),
    handler: Effect.fnUntraced(function*(state, { amount }, { aggregateId }) {
      if (state.status !== "open") return yield* new AccountNotOpen({ aggregateId })
      return AccountEvent.cases.MoneyDeposited.make({ amount })
    })
  })

A command that would break a rule, such as withdrawing more than the balance, fails with a typed error and stores nothing. The customer app shows those rejections as they happen.

Transfers, between two aggregates

A command changes one aggregate, and a transfer touches two. So it is split in two: sendTransfer debits the sender, and receiveTransfer credits the receiver once it hears about it.

.command("sendTransfer", {
  payload: Schema.Struct({ transferId: Schema.NonEmptyString, to: Schema.NonEmptyString, amount: Amount }),
  handler: Effect.fnUntraced(function*(state, { amount, to, transferId }, { aggregateId }) {
    if (state.status !== "open") return yield* new AccountNotOpen({ aggregateId })
    if (to === aggregateId) return yield* new SameAccountTransfer()
    if (state.balance < amount) return yield* new InsufficientFunds({ balance: state.balance, requested: amount })
    return AccountEvent.cases.TransferSent.make({ transferId, to, amount })
  })
})
.command("receiveTransfer", {
  payload: Schema.Struct({ transferId: Schema.NonEmptyString, from: Schema.NonEmptyString, amount: Amount }),
  handler: Effect.fnUntraced(function*(state, { amount, from, transferId }, { aggregateId }) {
    if (state.status !== "open") return yield* new AccountNotOpen({ aggregateId })
    if (state.received.includes(transferId)) return []
    return AccountEvent.cases.TransferReceived.make({ transferId, from, amount })
  })
})

receiveTransfer records the transfers it has credited. If it runs twice for the same transfer, the second time decides on nothing, so the saga below can safely retry.

The saga follows TransferSent events with store.subscribe. It sends receiveTransfer to the other account, and when that is refused, because the account is closed for instance, it refunds the sender:

/**
 * A transfer touches two aggregates, so no single command can do it. The saga
 * reacts to `TransferSent` by crediting the other account, and refunds the
 * sender when that is refused, for instance because the account is closed.
 */
const TransferSaga = Layer.effectDiscard(Effect.gen(function*() {
  const store = yield* EventStore.EventStore
  const bus = yield* Bus
  const decode = AggregateService.makeEventDecoder(Account)
  yield* store.subscribe({ aggregateNames: [Account.name], eventTypes: ["TransferSent"] }).pipe(
    Stream.runForEach((stored) =>
      Effect.gen(function*() {
        const event = (yield* decode(stored)) as Extract<AccountEvent, { _tag: "TransferSent" }>
        const credited = yield* bus.submit({
          aggregateId: event.to,
          type: "receiveTransfer",
          payload: { transferId: event.transferId, from: stored.aggregateId, amount: event.amount }
        }, "saga")
        if (!credited.ok) {
          yield* bus.submit({
            aggregateId: stored.aggregateId,
            type: "refundTransfer",
            payload: { transferId: event.transferId, to: event.to, amount: event.amount, reason: credited.tag }
          }, "saga")
        }
      })
    ),
    Effect.catchCause((cause) => Effect.logError("saga stopped", cause)),
    Effect.forkScoped
  )
}))

Between the two steps the money is in flight. The bank vault, a view model, shows that amount.

Three view models

Each view model folds the same events into a different shape, keeps its own checkpoint, and can be thrown away and rebuilt:

accounts: one row per account, with its balance, for the customer app.
bank-totals: money in the bank, money in flight, and counts by event type.
activity-feed: the latest 40 events, as sentences.

Their storage is a plain object outside React and Effect, so the screen shows exactly what they hold. Delivery is at least once, so the accounts view ignores an event whose position it has already applied:

export const AccountsView = ViewModel.make({
  name: "accounts",
  aggregates: [Account],
  dispose: Effect.sync(() => devstore.dispose("accounts")),
  handlers: everyEvent("accounts", (event, { aggregateId, position }) =>
    Effect.sync(() =>
      devstore.apply("accounts", position, (accounts) => {
        const current = accounts[aggregateId]
        if (event._tag === "AccountOpened") {
          return {
            ...accounts,
            [aggregateId]: {
              id: aggregateId,
              owner: event.owner,
              color: event.color,
              balance: 0,
              status: "open",
              txCount: 0,
              lastPosition: position,
              lastDelta: 0
            }
          }
        }
        // Idempotent: an event seen again leaves the row as it is.
        if (current === undefined || current.lastPosition >= position) return accounts
        const delta = balanceDelta(event)
        return {
          ...accounts,
          [aggregateId]: {
            ...current,
            balance: current.balance + delta,
            status: event._tag === "AccountClosed" ? "closed" : current.status,
            txCount: current.txCount + (delta === 0 ? 0 : 1),
            lastPosition: position,
            lastDelta: delta
          }
        }
      })
    ))
})

bank-totals has a switch in the devtools that makes its handler fail. Turn it on and the view model stalls on that event and retries with backoff, while the other two carry on. Turn it off and it catches up, without skipping anything.

Commands, through `dispatch`

Every command in the demo, from the customer app, the saga, the autopilot or the raw JSON console, goes through dispatch, the entry point for untyped input:

// `dispatch` is the real entry point: it validates the JSON envelope and
// payload, rehydrates the aggregate, runs the handler and appends.
const outcome = yield* bank.dispatch(input).pipe(
  Effect.map((result): Outcome => ({
    ok: true,
    version: result.version,
    position: result.position,
    events: result.events.map((event) => event._tag)
  })),
  Effect.catch((error) => Effect.succeed(describeError(error))),
  Effect.catchDefect((defect) => Effect.succeed(describeError(defect)))
)

Wiring it together

The whole engine is a layer: the in-memory store and checkpoints, the service, the view models, and the background fibers that mirror events to the devtools, run the saga, poll the view models’ status and send random traffic.

const Storage = Layer.merge(EventStore.layerMemory, Checkpoints.layerMemory)

const Core = Layer.merge(
  Layer.effect(Bank, AggregateService.make(Account)),
  ViewModel.layer(allViews)
).pipe(Layer.provideMerge(Storage))

const WithBus = BusLayer.pipe(Layer.provideMerge(Core))

const Main = Layer.mergeAll(Bridge, TransferSaga, StatusPoll, Autopilot).pipe(Layer.provideMerge(WithBus))

A ManagedRuntime runs it, and React reaches it through plain functions. Reset store disposes the runtime and starts a new one.

The devtools

Tab	What you can do
Commands	Watch each command through validating, loading, deciding and appending. Slow those stages down, pause the bus and step it, run the autopilot, and send raw JSON to `dispatch`
Event store	Read the log as it grows
Projections	See each view model’s checkpoint and lag, crash `bank-totals`, and rebuild one
Time travel	Scrub through the log: the customer app becomes a preview, where each account is its events up to the cursor folded with the aggregate’s projection. Replay view models up to a chosen event, or entirely

Source

examples/bank-ui on GitHub