The bank

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

The second example. The shop builds each piece in turn; the bank puts them in an app you can poke at. It is the classic event sourcing example, 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({
  // Encrypted with a key of the account's own, which "forget" destroys.
  AccountOpened: {
    owner: PersonalData.field(Schema.NonEmptyString, { redacted: "[forgotten]" }),
    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 starts a workflow for each TransferSent event. The workflow sends receiveTransfer to the other account, and when that is refused, because the account is closed for instance, it refunds the sender. Each command is an activity, so the engine keeps its outcome:

/**
 * A transfer touches two aggregates, so no single command can do it. For each
 * `TransferSent` the trigger starts a workflow that credits the other account,
 * and refunds the sender when that is refused because the account isn't open.
 * Each command is an activity whose outcome the engine keeps, and starting the
 * same transfer again joins the run already under way.
 */
const Transfer = Saga.workflow("Transfer", {
  payload: { transferId: Schema.String, from: Schema.String, to: Schema.String, amount: Schema.Int },
  idempotencyKey: ({ transferId }) => transferId
})

const TransferWorkflow = Saga.toLayer(Transfer, ({ amount, from, to, transferId }) =>
  Effect.gen(function*() {
    const bus = yield* Bus
    yield* Saga.command("credit", {
      error: AccountNotOpen,
      execute: bus.run({ aggregateId: to, type: "receiveTransfer", payload: { transferId, from, amount } }, "saga")
    }).pipe(
      Effect.catchTag("AccountNotOpen", ({ _tag }) =>
        Saga.command("refund", {
          execute: bus.run({
            aggregateId: from,
            type: "refundTransfer",
            payload: { transferId, to, amount, reason: _tag }
          }, "saga")
        }))
    )
    // Both commands record `TransferSent` as their cause, in the correlation it started.
  }))

const Transfers = Saga.make({
  name: "transfers",
  aggregates: [Account],
  handlers: {
    TransferSent: ({ amount, to, transferId }, stored) =>
      Saga.start(Transfer, { transferId, from: stored.aggregateId, to, amount })
  }
})

const TransferSaga = Saga.layer([Transfers]).pipe(
  Layer.provideMerge(TransferWorkflow),
  Layer.provide(WorkflowEngine.layerMemory)
)

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
            }
          }
        })
      )
  ),
  // The owner's name is personal data: once they are forgotten, the row keeps the
  // balance and loses the name.
  onForget: (subject, { position }) =>
    Effect.sync(() =>
      devstore.apply("accounts", position, (accounts) => {
        const current = accounts[subject]
        return current === undefined ? accounts : { ...accounts, [subject]: { ...current, owner: forgotten } }
      })
    )
})

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.
return yield* bank.dispatch(input).pipe(
  // Recorded on the events beside the app version the service was built with.
  Metadata.withMetadata({ origin }),
  Effect.tap((result) => Effect.sync(() => devstore.finish(id, toOutcome(result)))),
  Effect.tapError((error) => Effect.sync(() => devstore.finish(id, describeError(error)))),
  Effect.tapDefect((defect) => Effect.sync(() => devstore.finish(id, describeError(defect))))
)

Wiring it together

The whole engine is a layer: the in-memory store, checkpoints and snapshots, the service, the view models, and the background fibers that mirror events to the devtools, run the saga, poll the view models’ status and the service’s metrics, and send random traffic. The service keeps only four accounts in memory and saves a snapshot every five events, so with more accounts open the Metrics tab shows commands starting from snapshots and replays.

// The keys that encrypt owners' names. A random master key is enough for a store that
// lives as long as the page.
const Keys = PersonalData.layerMemory().pipe(Layer.provide(PersonalData.layerMasterKeyEphemeral))

const Storage = Layer.mergeAll(EventStore.layerMemory, Checkpoints.layerMemory, KeyValueStore.layerMemory, Keys)

const AccountState = Schema.Struct({
  status: Schema.Literals(["none", "open", "closed"]),
  owner: Schema.String,
  color: Schema.String,
  balance: Schema.Number,
  received: Schema.Array(Schema.String)
})

// Only four accounts stay in memory, so the metrics tab shows loads from snapshots too.
const BankService = Layer.effect(
  Bank,
  AggregateService.make(Account, { stateCacheSize: 4, snapshots: { every: 5, schema: AccountState } })
)

const Core = Layer.merge(BankService, ViewModel.layer(allViews)).pipe(Layer.provideMerge(Storage))

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

const Main = Layer.mergeAll(Bridge, TransferSaga, StatusPoll, MetricsPoll, Autopilot).pipe(
  Layer.provideMerge(WithBus),
  // Every command records the version of the software that ran it.
  Layer.provide(Metadata.layer({ appVersion })),
  // A registry of its own, so a reset counts from zero.
  Layer.provideMerge(Layer.sync(Metric.MetricRegistry, () => new Map())),
  // Every span reports to the devtools' trace tab as it starts, changes and ends.
  Layer.provideMerge(Layer.succeed(Tracer.Tracer, makeRecordingTracer(devstore.span)))
)

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

Forgetting an owner

The owner’s name is personal data: AccountOpened stores it encrypted with a key of the account’s own, which the Event store tab shows as a $personal block. forget on a card destroys that key:

/**
 * Forgets the owner of an account: the key that encrypts their name is
 * destroyed. Stored events stay as they were, view models erase the name in
 * `onForget`, and the events tab decodes the log again to show the replacement.
 */
forget: (subject: string) =>
  runtime.runPromise(Effect.gen(function*() {
    yield* PersonalData.forget(subject).pipe(Metadata.withMetadata({ origin: "app" }))
    const decode = AggregateService.makeEventDecoder(Account, yield* PersonalData.Keys)
    const entries = devstore.get().events.filter((entry) => entry.aggregateId === subject)
    const decoded = yield* Effect.forEach(entries, (entry) => decode(entry.stored))
    devstore.redecoded(new Map(entries.map((entry, index) => [entry.position, decoded[index]!])))
  })),

The log doesn’t change and the totals still add up, while the card, the feed and every replay read [forgotten]. The accounts view erases the name in onForget, and new commands on the account still work, since deposits hold no personal data.

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
Trace	Follow every span Effect opens, recorded by a `Tracer` of the demo’s own: each command on the bus with the stages it waited in, the aggregate’s span and `EventStore.append`, and for a transfer the saga’s workflow and its activities
Metrics	Watch the service’s metrics: where each load found its state, the events it read, the states in memory and those dropped, the snapshots saved, and how long each command took by outcome

Source

examples/bank-ui on GitHub