This is the abridged developer documentation for effect-cqrs # effect-cqrs > Event sourcing and CQRS for Effect v4. Commands decide on events, events are the only source of truth, and read models are projections you can throw away and rebuild. ## How a command flows 1. ### Command A typed call, or JSON from a transport through `dispatch`. The payload is checked against its schema first. [Commands guide](/guides/commands/) 2. ### Aggregate The service rebuilds the aggregate's state from its events, and the handler decides what happened, or fails with a typed error. [Aggregates guide](/guides/aggregates/) 3. ### Event store The events are appended, but only if nobody wrote to that aggregate since its state was loaded. [Storage guide](/guides/storage/) 4. ### View models Each one reads the new events from its own checkpoint. `caughtUp` waits until your own write is there. [View models guide](/guides/view-models/) ## Beyond one command The path above is the core. These modules cover what a system needs around it, and each is opt-in. * Saga ### Sagas across aggregates A process that touches several aggregates, like a transfer between two accounts, runs as a durable workflow that an event starts. The engine keeps the outcome of every step, so a restart skips the commands it already ran, and compensations undo the steps that succeeded. [Sagas guide](/guides/sagas/) * PersonalData ### GDPR, by crypto-shredding Forget a person without rewriting the log. Personal fields are encrypted with a key that belongs to their subject, and `forget` destroys it. The events stay, with their amounts and dates, and streams still replay. [Personal data guide](/guides/personal-data/) * CommandApi ### An HTTP API from an aggregate A `POST` endpoint per command, with history and state, as an `HttpApiGroup` with OpenAPI. `Idempotency-Key` and `If-Match` reach the command, and failures map to status codes. [HTTP](/guides/integration/#http) * Outbox · EventCatalog ### Events for other systems An outbox publishes events to Kafka, NATS, SQS and the like, at least once and in order. The catalog describes them as JSON Schema and AsyncAPI, and its guard fails when a change can no longer decode what is stored. [Outbox](/guides/integration/#outbox)[Event catalog](/guides/integration/#event-catalog) * EventVersions ### Events that change shape The log is never rewritten. Declare the older shapes of an event and how each becomes the next, and old events are upcast as they are read. Handlers and projections only see the current one. [Evolving events](/guides/aggregates/#evolving-events) * ViewModel · Leases ### Made to keep running Command ids make a retried command run once. View models retry a failing handler, park an event that never passes, and run on one replica at a time. Every event records its cause, and the services report metrics. [Command ids](/guides/commands/#retries-and-command-ids)[Failing handlers](/guides/view-models/#failing-handlers)[Several processes](/guides/view-models/#several-processes) ## A bank, running in your browser Accounts are aggregates, three view models project them, and a saga moves money between them. The customer app on the left only shows read models, so you can watch them trail the store. The devtools on the right slow every stage down, pause and step commands, crash a projection, and replay the log. [Open the demo](/demo/)[Read how it's built](/tutorial/bank/) ******effect-cqrs.dambox.ch/demo/ [The bank demo](/demo/) ## Tests read like moleculer-cqrs `Fixture` runs commands through the real service on an in-memory store, so validation and encoding are tested too. The chain is an Effect, which `it.effect` runs. ```ts it.effect("deletes a todo", () => Fixture.make(Todo) .givenEvents([created]) .when("deleteTodo", {}) .expectEvent("TodoDeleted") .inspectState((state) => assert.isTrue(state.deleted))) ``` [Testing guide](/guides/testing/) ## Keep the log where it fits Every store passes the same tests. The numbers come from one laptop, so read them as orders of magnitude. * In memoryFor tests, examples and the browser.60,000 commands/s * SQLiteOne process, one file, no server.0.1 ms a command * PostgreSQLSeveral processes writing, positions in commit order.1 ms a command [Storage guide](/guides/storage/)[Performance](/reference/performance/) # Getting started > Install effect-cqrs, or clone the repository to run its examples. ## Install * pnpm ```bash pnpm add effect-cqrs effect ``` * npm ```bash npm install effect-cqrs effect ``` * yarn ```bash yarn add effect-cqrs effect ``` * bun ```bash bun add effect-cqrs effect ``` It needs Node 24 or later and Effect 4. To store events in SQLite or PostgreSQL, add that database’s `effect/sql` driver too: * pnpm ```bash pnpm add @effect/sql-sqlite-node pnpm add @effect/sql-pg ``` * npm ```bash npm install @effect/sql-sqlite-node npm install @effect/sql-pg ``` * yarn ```bash yarn add @effect/sql-sqlite-node yarn add @effect/sql-pg ``` * bun ```bash bun add @effect/sql-sqlite-node bun add @effect/sql-pg ``` Everything is exported from `effect-cqrs`, and each module also has its own path, such as `effect-cqrs/ViewModel`. ## Try the examples The examples live in the repository, where they import the TypeScript sources directly. 1. Clone the repository and install the dependencies. It needs pnpm. ```bash git clone https://github.com/davidnussio/effect-cqrs.git cd effect-cqrs pnpm install ``` 2. Run the tests. ```bash pnpm test ``` 3. Run the [tutorial](/tutorial/shop/)’s shop, which uses every feature in one file. ```bash pnpm tutorial ``` 4. Run the todo example. It is the moleculer-cqrs README walkthrough as a program, on an in-memory SQLite database. ```bash pnpm example ``` 5. Serve the bank demo at . ```bash pnpm ui ``` 6. Serve the tutorial’s shop, with its traces, saga and stored rows. ```bash pnpm ui:shop ``` ## What’s where ## Next steps [Follow the tutorial](/tutorial/shop/) [Define an aggregate](/guides/aggregates/) [Walk through the bank](/tutorial/bank/) [Use it with a coding agent](/using-with-ai/) # The bank > The second example: a bank whose accounts are aggregates, with three view models and a transfer saga, running in your browser. [Open the bank demo](/demo/)effect-cqrs in your browser, on the in-memory event store. Nothing leaves the page. The second example. [The shop](/tutorial/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: ```bash 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: examples/bank-ui/src/domain/account.ts ```ts 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: examples/bank-ui/src/domain/account.ts ```ts export const Account = Aggregate.make()({ 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. examples/bank-ui/src/domain/account.ts ```ts .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](/guides/sagas/) 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: examples/bank-ui/src/engine.ts ```ts /** * 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: examples/bank-ui/src/domain/views.ts ```ts 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: examples/bank-ui/src/engine.ts ```ts // `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. examples/bank-ui/src/engine.ts ```ts // 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](/guides/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: examples/bank-ui/src/engine.ts ```ts /** * 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](/guides/commands/#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](https://github.com/davidnussio/effect-cqrs/blob/main/examples/bank-ui) # The shop > The first example, built step by step: a shop with orders, stock and a saga with a timeout, then metadata, metrics and event versions. [Open the shop demo](/shop-demo/)The same shop in your browser, with its traces, its saga's state, the stored rows and the metrics. The first of three examples, and the one to start from: it builds each piece of the framework in turn. [The bank](/tutorial/bank/) and [the workshop](/tutorial/workshop/) then put them together in bigger apps. We build a shop: customers place **orders**, and a **saga** holds the **stock** for each one and gives the customer 15 seconds to pay. It confirms a paid order, and rejects one that is out of stock or not paid in time. Every snippet comes from [the full example](#the-full-example), which you can run with `pnpm tutorial`. ## 1. Events Events are what happened. Each aggregate declares its own as a tagged union: examples/tutorial/shop.ts ```ts const Quantity = Schema.Int.check(Schema.isGreaterThan(0)) export const StockEvent = Schema.TaggedUnion({ StockAdded: { quantity: Quantity }, StockReserved: { orderId: Schema.String, quantity: Quantity }, StockReleased: { orderId: Schema.String, quantity: Quantity } }) export const OrderEvent = Schema.TaggedUnion({ OrderPlaced: { sku: Schema.String, quantity: Quantity }, // The saga asks for the payment. The token names the saga's run that waits for it. PaymentRequested: { token: DurableDeferred.Token }, OrderPaid: { token: DurableDeferred.Token }, OrderConfirmed: {}, OrderRejected: { reason: Schema.String } }) ``` ## 2. Aggregates and commands An aggregate folds its events into a state with `projection`, and each command looks at that state and returns new events, or fails: examples/tutorial/shop.ts ```ts export class OutOfStock extends Schema.TaggedError()("OutOfStock", { sku: Schema.String, available: Schema.Int }) {} export class OrderAlreadyPlaced extends Schema.TaggedError()("OrderAlreadyPlaced", { orderId: Schema.String }) {} export class OrderNotPayable extends Schema.TaggedError()("OrderNotPayable", { orderId: Schema.String, status: Schema.String }) {} export class OrderAlreadyPaid extends Schema.TaggedError()("OrderAlreadyPaid", { orderId: Schema.String }) {} ``` examples/tutorial/shop.ts ```ts export interface StockState { readonly available: number /** What each order holds, by order id. */ readonly reserved: Readonly> } export const Stock = Aggregate.make()({ name: "stock", events: StockEvent, initial: { available: 0, reserved: {} }, projection: { StockAdded: (state, { quantity }) => ({ ...state, available: state.available + quantity }), StockReserved: (state, { orderId, quantity }) => ({ available: state.available - quantity, reserved: { ...state.reserved, [orderId]: quantity } }), StockReleased: (state, { orderId, quantity }) => { const { [orderId]: _, ...reserved } = state.reserved return { available: state.available + quantity, reserved } } } }) .command("add", { payload: Schema.Struct({ quantity: Quantity }), handler: (_, { quantity }) => Effect.succeed(StockEvent.cases.StockAdded.make({ quantity })) }) .command("reserve", { payload: Schema.Struct({ orderId: Schema.String, quantity: Quantity }), handler: Effect.fnUntraced(function*(state, { orderId, quantity }, { aggregateId }) { // A saga may run this twice: the second time records nothing. if (state.reserved[orderId] !== undefined) return [] if (state.available < quantity) return yield* new OutOfStock({ sku: aggregateId, available: state.available }) return StockEvent.cases.StockReserved.make({ orderId, quantity }) }) }) .command("release", { payload: Schema.Struct({ orderId: Schema.String }), handler: (state, { orderId }) => { const quantity = state.reserved[orderId] return Effect.succeed(quantity === undefined ? [] : StockEvent.cases.StockReleased.make({ orderId, quantity })) } }) ``` examples/tutorial/shop.ts ```ts export interface OrderState { readonly status: "none" | "placed" | "awaiting-payment" | "paid" | "confirmed" | "rejected" readonly sku: string readonly quantity: number readonly payment: DurableDeferred.Token | null } export const Order = Aggregate.make()({ name: "order", events: OrderEvent, initial: { status: "none", sku: "", quantity: 0, payment: null }, projection: { OrderPlaced: (_, { quantity, sku }) => ({ status: "placed", sku, quantity, payment: null }), PaymentRequested: (state, { token }) => ({ ...state, status: "awaiting-payment", payment: token }), OrderPaid: (state) => ({ ...state, status: "paid" }), OrderConfirmed: (state) => ({ ...state, status: "confirmed" }), OrderRejected: (state) => ({ ...state, status: "rejected" }) }, versions: orderVersions }) .command("place", { payload: Schema.Struct({ sku: Schema.String, quantity: Quantity }), handler: Effect.fnUntraced(function*(state, { quantity, sku }, { aggregateId }) { if (state.status !== "none") return yield* new OrderAlreadyPlaced({ orderId: aggregateId }) return OrderEvent.cases.OrderPlaced.make({ sku, quantity }) }) }) .command("requestPayment", { payload: Schema.Struct({ token: DurableDeferred.Token }), handler: (state, { token }) => Effect.succeed(state.status === "placed" ? OrderEvent.cases.PaymentRequested.make({ token }) : []) }) // The customer pays, and only while the saga waits for it. .command("pay", { payload: Schema.Struct({}), handler: Effect.fnUntraced(function*(state, _, { aggregateId }) { if (state.status === "paid" || state.status === "confirmed") return [] if (state.status !== "awaiting-payment" || state.payment === null) { return yield* new OrderNotPayable({ orderId: aggregateId, status: state.status }) } return OrderEvent.cases.OrderPaid.make({ token: state.payment }) }) }) .command("confirm", { payload: Schema.Struct({}), handler: (state) => Effect.succeed(state.status === "paid" ? OrderEvent.cases.OrderConfirmed.make({}) : []) }) .command("reject", { payload: Schema.Struct({ reason: Schema.String }), handler: Effect.fnUntraced(function*(state, { reason }, { aggregateId }) { if (state.status === "paid") return yield* new OrderAlreadyPaid({ orderId: aggregateId }) if (state.status !== "placed" && state.status !== "awaiting-payment") return [] return OrderEvent.cases.OrderRejected.make({ reason }) }) }) ``` Returning `[]` records nothing, which makes the commands a saga runs safe to repeat. `requestPayment` and `pay` take part in the saga’s payment step in [step 5](#5-a-saga), and `versions` comes in [step 8](#8-event-versioning). ## 3. Running commands `AggregateService.make` turns an aggregate into a service with one typed function per command: examples/tutorial/shop.ts ```ts export class Orders extends Context.Service>()("shop/Orders") { static readonly layer = Layer.effect(Orders, AggregateService.make(Order)) } export class Stocks extends Context.Service>()("shop/Stocks") { static readonly layer = Layer.effect(Stocks, AggregateService.make(Stock)) } ``` examples/tutorial/shop.ts ```ts yield* stocks.commands.add("apple", { quantity: 5 }) yield* orders.commands.place("order-1", { sku: "apple", quantity: 3 }) // The saga holds the stock and asks for the payment: the customer has 15 seconds to pay. yield* reached("order-1", "PaymentRequested", "OrderRejected") yield* orders.commands.pay("order-1", {}) yield* settled("order-1") const rejected = yield* Effect.flip(orders.commands.place("order-1", { sku: "apple", quantity: 1 })) yield* Effect.log("placing order-1 again fails with", rejected._tag) ``` A refused command fails with its typed error and stores nothing. `pay` is the customer’s side of the saga’s payment step: the order only takes it while the saga is waiting for it. ## 4. A view model A view model reads events into whatever a query needs, here a list of orders: examples/tutorial/shop.ts ```ts export class OrderList extends Context.Service>>()("shop/OrderList") { static readonly layer = Layer.effect(OrderList, Ref.make>(new Map())) } const setStatus = (orderId: string, status: string) => OrderList.use((list) => Ref.update(list, (orders) => new Map(orders).set(orderId, status))) export const OrderListView = ViewModel.make({ name: "order-list", aggregates: [Order], dispose: OrderList.use((list) => Ref.set(list, new Map())), handlers: { OrderPlaced: ({ quantity, sku }, { aggregateId }) => setStatus(aggregateId, `placed ${quantity} ${sku}`), PaymentRequested: (_, { aggregateId }) => setStatus(aggregateId, "awaiting payment"), OrderPaid: (_, { aggregateId }) => setStatus(aggregateId, "paid"), OrderConfirmed: (_, { aggregateId }) => setStatus(aggregateId, "confirmed"), OrderRejected: ({ reason }, { aggregateId }) => setStatus(aggregateId, `rejected: ${reason}`) } }) ``` ## 5. A saga A saga reacts to an event and runs commands on other aggregates. The trigger starts a workflow for each `OrderPlaced`, and each command in it is a step the engine remembers: examples/tutorial/shop.ts ```ts /** How long a customer has to pay once the stock is held. */ export const paymentTimeout = Duration.seconds(15) // The customer's payment, which a run of the saga waits for. export const Payment = DurableDeferred.make("Payment") // One run per order: hold the stock, ask for the payment and wait for it, then // confirm the order, or reject it and put the stock back. export const Fulfil = Saga.workflow("Fulfil", { payload: { orderId: Schema.String, sku: Schema.String, quantity: Schema.Int }, idempotencyKey: ({ orderId }) => orderId }) export const FulfilWorkflow = Saga.toLayer(Fulfil, ({ orderId, quantity, sku }) => Effect.gen(function*() { const orders = yield* Orders const stocks = yield* Stocks // 1. Hold the stock, or turn the order down. const reserved = yield* Saga.command("reserve", { error: OutOfStock, execute: stocks.commands.reserve(sku, { orderId, quantity }) }).pipe( Effect.as(true), Effect.catchTag("OutOfStock", ({ available }) => Saga.command("reject", { execute: orders.commands.reject(orderId, { reason: `only ${available} left` }) }).pipe(Effect.as(false))) ) if (!reserved) return // 2. Ask for the payment, then wait for whichever comes first: the payment or the deadline. const token = yield* DurableDeferred.token(Payment) yield* Saga.command("request-payment", { execute: orders.commands.requestPayment(orderId, { token }) }) const paid = yield* DurableDeferred.raceAll({ name: "payment", success: Schema.Boolean, error: Schema.Never, effects: [ DurableDeferred.await(Payment).pipe(Effect.as(true)), DurableClock.sleep({ name: "payment-timeout", duration: paymentTimeout }).pipe(Effect.as(false)) ] }) // 3. Too late: reject the order and give the stock back. The order refuses if the // customer paid just as the time ran out, and the saga confirms it instead. if (!paid) { const expired = yield* Saga.command("expire", { error: OrderAlreadyPaid, execute: orders.commands.reject(orderId, { reason: "not paid in time" }) }).pipe( Effect.as(true), Effect.catchTag("OrderAlreadyPaid", () => Effect.succeed(false)) ) if (expired) { yield* Saga.command("release", { execute: stocks.commands.release(sku, { orderId }) }) return } } yield* Saga.command("confirm", { execute: orders.commands.confirm(orderId, {}) }) })) // The trigger: starts a run for every order placed, and wakes it when the order is paid. export const Fulfilment = Saga.make({ name: "fulfilment", aggregates: [Order], handlers: { OrderPlaced: ({ quantity, sku }, { aggregateId }) => Saga.start(Fulfil, { orderId: aggregateId, sku, quantity }), OrderPaid: ({ token }) => DurableDeferred.succeed(Payment, { token, value: undefined }) } }) ``` The run waits for the payment in the middle of the process: * `DurableDeferred.token` names this run’s `Payment`. The saga stores it in `PaymentRequested`, the order copies it into `OrderPaid`, and the trigger hands it to `DurableDeferred.succeed`, which wakes the run. * `DurableDeferred.raceAll` races the payment against `DurableClock.sleep` and saves the winner, so a replay takes the same branch. * When the time runs out, the saga rejects the order and releases the stock. If the customer paid just before, the order refuses with `OrderAlreadyPaid`, and the saga confirms it instead. Each aggregate decides alone, so the order settles any race between the two. `DurableClock.sleep` keeps a minute or less in memory. A longer timeout becomes a clock the engine schedules, which a durable engine keeps across restarts. ## 6. Wiring Everything is a layer. Here all of it runs in memory: examples/tutorial/shop.ts ```ts const Storage = Layer.mergeAll(EventStore.layerMemory, Checkpoints.layerMemory, OrderList.layer) const Sagas = Saga.layer([Fulfilment]).pipe( Layer.provideMerge(FulfilWorkflow), Layer.provide(WorkflowEngine.layerMemory) ) export const Shop = Layer.mergeAll(Sagas, ViewModel.layer([OrderListView])).pipe( Layer.provideMerge(Layer.merge(Orders.layer, Stocks.layer)), Layer.provideMerge(Storage), // Every command records the version of the software that ran it. Layer.provide(Metadata.layer({ appVersion: "1.0.0" })) ) ``` ## 7. Metadata Metadata is what a command carries beside its payload: the app version, the user, the channel. `Metadata.layer` above sets the version for every command, and `Metadata.withMetadata` adds to it for one call: examples/tutorial/shop.ts ```ts yield* orders.commands.place("order-2", { sku: "apple", quantity: 3 }).pipe( Metadata.withMetadata({ userId: "bob", channel: "mobile" }) ) yield* settled("order-2") const history = yield* orders.history("order-2") yield* Effect.log("order-2 history", history.map(({ metadata, type }) => ({ type, metadata }))) ``` ```txt order-2 history [ { type: 'OrderPlaced', metadata: { appVersion: '1.0.0', userId: 'bob', channel: 'mobile' } }, { type: 'OrderRejected', metadata: { appVersion: '1.0.0' } } ] ``` The saga’s `OrderRejected` doesn’t inherit `userId`: a reaction records its own metadata. ## 8. Event versioning Stored events never change. When an event’s schema does, declare the old shape and how to upcast it: examples/tutorial/shop.ts ```ts // Before quantities existed, an order was always for one item. const OrderPlacedV1 = Schema.Struct({ sku: Schema.String }) const orderVersions = { OrderPlaced: EventVersions.from(OrderPlacedV1).next( OrderEvent.cases.OrderPlaced, (v1) => ({ ...v1, quantity: 1 }) ) } ``` An order stored before quantities existed is read as an order for one item, by the projection, the view model and the saga alike: examples/tutorial/shop.ts ```ts // An order the previous version of the shop stored, before quantities existed. yield* EventStore.EventStore.use((store) => store.append({ aggregateName: "order", aggregateId: "order-0", expectedVersion: 0, events: [{ type: "OrderPlaced", schemaVersion: 1, payload: { sku: "pear" } }] }) ) yield* settled("order-0") const legacy = yield* orders.state("order-0") yield* Effect.log("order-0 is read as", legacy) ``` ```txt order-0 is read as { status: 'rejected', sku: 'pear', quantity: 1, payment: null } ``` ## 9. Metrics Every aggregate service records metrics with Effect’s `Metric` module. Read them with `Metric.snapshot`, or send them to an exporter: examples/tutorial/shop.ts ```ts // How many `place` commands ran, and how long they took, by outcome. const timings = (yield* Metric.snapshot).flatMap((metric) => metric.id === AggregateService.MetricNames.commandDuration && metric.type === "Histogram" && metric.attributes?.["aggregate"] === "order" && metric.attributes["command"] === "place" ? [{ outcome: metric.attributes["outcome"], count: metric.state.count, totalMs: metric.state.sum }] : [] ) yield* Effect.log("order.place timings", timings) ``` ```txt order.place timings [ { outcome: 'success', count: 2, totalMs: 0.806 }, { outcome: 'OrderAlreadyPlaced', count: 1, totalMs: 0.273 } ] ``` ## The full example examples/tutorial/shop.ts ```ts /** * The tutorial's shop: a saga holds the stock for each order and gives the * customer 15 seconds to pay. It uses commands, events, a view model, a saga with * a timeout, metadata, metrics and an event version. Run it with `pnpm tutorial`. */ import { Context, Duration, Effect, Layer, Metric, Option, Ref, Schema, Stream } from "effect" import { Aggregate, AggregateService, Checkpoints, EventStore, EventVersions, Metadata, Saga, ViewModel } from "effect-cqrs" import { DurableClock, DurableDeferred, WorkflowEngine } from "effect/workflow" const Quantity = Schema.Int.check(Schema.isGreaterThan(0)) export const StockEvent = Schema.TaggedUnion({ StockAdded: { quantity: Quantity }, StockReserved: { orderId: Schema.String, quantity: Quantity }, StockReleased: { orderId: Schema.String, quantity: Quantity } }) export const OrderEvent = Schema.TaggedUnion({ OrderPlaced: { sku: Schema.String, quantity: Quantity }, // The saga asks for the payment. The token names the saga's run that waits for it. PaymentRequested: { token: DurableDeferred.Token }, OrderPaid: { token: DurableDeferred.Token }, OrderConfirmed: {}, OrderRejected: { reason: Schema.String } }) export class OutOfStock extends Schema.TaggedError()("OutOfStock", { sku: Schema.String, available: Schema.Int }) {} export class OrderAlreadyPlaced extends Schema.TaggedError()("OrderAlreadyPlaced", { orderId: Schema.String }) {} export class OrderNotPayable extends Schema.TaggedError()("OrderNotPayable", { orderId: Schema.String, status: Schema.String }) {} export class OrderAlreadyPaid extends Schema.TaggedError()("OrderAlreadyPaid", { orderId: Schema.String }) {} export interface StockState { readonly available: number /** What each order holds, by order id. */ readonly reserved: Readonly> } export const Stock = Aggregate.make()({ name: "stock", events: StockEvent, initial: { available: 0, reserved: {} }, projection: { StockAdded: (state, { quantity }) => ({ ...state, available: state.available + quantity }), StockReserved: (state, { orderId, quantity }) => ({ available: state.available - quantity, reserved: { ...state.reserved, [orderId]: quantity } }), StockReleased: (state, { orderId, quantity }) => { const { [orderId]: _, ...reserved } = state.reserved return { available: state.available + quantity, reserved } } } }) .command("add", { payload: Schema.Struct({ quantity: Quantity }), handler: (_, { quantity }) => Effect.succeed(StockEvent.cases.StockAdded.make({ quantity })) }) .command("reserve", { payload: Schema.Struct({ orderId: Schema.String, quantity: Quantity }), handler: Effect.fnUntraced(function*(state, { orderId, quantity }, { aggregateId }) { // A saga may run this twice: the second time records nothing. if (state.reserved[orderId] !== undefined) return [] if (state.available < quantity) return yield* new OutOfStock({ sku: aggregateId, available: state.available }) return StockEvent.cases.StockReserved.make({ orderId, quantity }) }) }) .command("release", { payload: Schema.Struct({ orderId: Schema.String }), handler: (state, { orderId }) => { const quantity = state.reserved[orderId] return Effect.succeed(quantity === undefined ? [] : StockEvent.cases.StockReleased.make({ orderId, quantity })) } }) // Before quantities existed, an order was always for one item. const OrderPlacedV1 = Schema.Struct({ sku: Schema.String }) const orderVersions = { OrderPlaced: EventVersions.from(OrderPlacedV1).next( OrderEvent.cases.OrderPlaced, (v1) => ({ ...v1, quantity: 1 }) ) } export interface OrderState { readonly status: "none" | "placed" | "awaiting-payment" | "paid" | "confirmed" | "rejected" readonly sku: string readonly quantity: number readonly payment: DurableDeferred.Token | null } export const Order = Aggregate.make()({ name: "order", events: OrderEvent, initial: { status: "none", sku: "", quantity: 0, payment: null }, projection: { OrderPlaced: (_, { quantity, sku }) => ({ status: "placed", sku, quantity, payment: null }), PaymentRequested: (state, { token }) => ({ ...state, status: "awaiting-payment", payment: token }), OrderPaid: (state) => ({ ...state, status: "paid" }), OrderConfirmed: (state) => ({ ...state, status: "confirmed" }), OrderRejected: (state) => ({ ...state, status: "rejected" }) }, versions: orderVersions }) .command("place", { payload: Schema.Struct({ sku: Schema.String, quantity: Quantity }), handler: Effect.fnUntraced(function*(state, { quantity, sku }, { aggregateId }) { if (state.status !== "none") return yield* new OrderAlreadyPlaced({ orderId: aggregateId }) return OrderEvent.cases.OrderPlaced.make({ sku, quantity }) }) }) .command("requestPayment", { payload: Schema.Struct({ token: DurableDeferred.Token }), handler: (state, { token }) => Effect.succeed(state.status === "placed" ? OrderEvent.cases.PaymentRequested.make({ token }) : []) }) // The customer pays, and only while the saga waits for it. .command("pay", { payload: Schema.Struct({}), handler: Effect.fnUntraced(function*(state, _, { aggregateId }) { if (state.status === "paid" || state.status === "confirmed") return [] if (state.status !== "awaiting-payment" || state.payment === null) { return yield* new OrderNotPayable({ orderId: aggregateId, status: state.status }) } return OrderEvent.cases.OrderPaid.make({ token: state.payment }) }) }) .command("confirm", { payload: Schema.Struct({}), handler: (state) => Effect.succeed(state.status === "paid" ? OrderEvent.cases.OrderConfirmed.make({}) : []) }) .command("reject", { payload: Schema.Struct({ reason: Schema.String }), handler: Effect.fnUntraced(function*(state, { reason }, { aggregateId }) { if (state.status === "paid") return yield* new OrderAlreadyPaid({ orderId: aggregateId }) if (state.status !== "placed" && state.status !== "awaiting-payment") return [] return OrderEvent.cases.OrderRejected.make({ reason }) }) }) export class Orders extends Context.Service>()("shop/Orders") { static readonly layer = Layer.effect(Orders, AggregateService.make(Order)) } export class Stocks extends Context.Service>()("shop/Stocks") { static readonly layer = Layer.effect(Stocks, AggregateService.make(Stock)) } export class OrderList extends Context.Service>>()("shop/OrderList") { static readonly layer = Layer.effect(OrderList, Ref.make>(new Map())) } const setStatus = (orderId: string, status: string) => OrderList.use((list) => Ref.update(list, (orders) => new Map(orders).set(orderId, status))) export const OrderListView = ViewModel.make({ name: "order-list", aggregates: [Order], dispose: OrderList.use((list) => Ref.set(list, new Map())), handlers: { OrderPlaced: ({ quantity, sku }, { aggregateId }) => setStatus(aggregateId, `placed ${quantity} ${sku}`), PaymentRequested: (_, { aggregateId }) => setStatus(aggregateId, "awaiting payment"), OrderPaid: (_, { aggregateId }) => setStatus(aggregateId, "paid"), OrderConfirmed: (_, { aggregateId }) => setStatus(aggregateId, "confirmed"), OrderRejected: ({ reason }, { aggregateId }) => setStatus(aggregateId, `rejected: ${reason}`) } }) /** How long a customer has to pay once the stock is held. */ export const paymentTimeout = Duration.seconds(15) // The customer's payment, which a run of the saga waits for. export const Payment = DurableDeferred.make("Payment") // One run per order: hold the stock, ask for the payment and wait for it, then // confirm the order, or reject it and put the stock back. export const Fulfil = Saga.workflow("Fulfil", { payload: { orderId: Schema.String, sku: Schema.String, quantity: Schema.Int }, idempotencyKey: ({ orderId }) => orderId }) export const FulfilWorkflow = Saga.toLayer(Fulfil, ({ orderId, quantity, sku }) => Effect.gen(function*() { const orders = yield* Orders const stocks = yield* Stocks // 1. Hold the stock, or turn the order down. const reserved = yield* Saga.command("reserve", { error: OutOfStock, execute: stocks.commands.reserve(sku, { orderId, quantity }) }).pipe( Effect.as(true), Effect.catchTag("OutOfStock", ({ available }) => Saga.command("reject", { execute: orders.commands.reject(orderId, { reason: `only ${available} left` }) }).pipe(Effect.as(false))) ) if (!reserved) return // 2. Ask for the payment, then wait for whichever comes first: the payment or the deadline. const token = yield* DurableDeferred.token(Payment) yield* Saga.command("request-payment", { execute: orders.commands.requestPayment(orderId, { token }) }) const paid = yield* DurableDeferred.raceAll({ name: "payment", success: Schema.Boolean, error: Schema.Never, effects: [ DurableDeferred.await(Payment).pipe(Effect.as(true)), DurableClock.sleep({ name: "payment-timeout", duration: paymentTimeout }).pipe(Effect.as(false)) ] }) // 3. Too late: reject the order and give the stock back. The order refuses if the // customer paid just as the time ran out, and the saga confirms it instead. if (!paid) { const expired = yield* Saga.command("expire", { error: OrderAlreadyPaid, execute: orders.commands.reject(orderId, { reason: "not paid in time" }) }).pipe( Effect.as(true), Effect.catchTag("OrderAlreadyPaid", () => Effect.succeed(false)) ) if (expired) { yield* Saga.command("release", { execute: stocks.commands.release(sku, { orderId }) }) return } } yield* Saga.command("confirm", { execute: orders.commands.confirm(orderId, {}) }) })) // The trigger: starts a run for every order placed, and wakes it when the order is paid. export const Fulfilment = Saga.make({ name: "fulfilment", aggregates: [Order], handlers: { OrderPlaced: ({ quantity, sku }, { aggregateId }) => Saga.start(Fulfil, { orderId: aggregateId, sku, quantity }), OrderPaid: ({ token }) => DurableDeferred.succeed(Payment, { token, value: undefined }) } }) const Storage = Layer.mergeAll(EventStore.layerMemory, Checkpoints.layerMemory, OrderList.layer) const Sagas = Saga.layer([Fulfilment]).pipe( Layer.provideMerge(FulfilWorkflow), Layer.provide(WorkflowEngine.layerMemory) ) export const Shop = Layer.mergeAll(Sagas, ViewModel.layer([OrderListView])).pipe( Layer.provideMerge(Layer.merge(Orders.layer, Stocks.layer)), Layer.provideMerge(Storage), // Every command records the version of the software that ran it. Layer.provide(Metadata.layer({ appVersion: "1.0.0" })) ) /** Waits until one of these events is stored for an order, and the view model has seen it. */ export const reached = (orderId: string, ...eventTypes: ReadonlyArray) => Effect.gen(function*() { const store = yield* EventStore.EventStore const viewModels = yield* ViewModel.ViewModels const done = yield* store.subscribe({ aggregateIds: [orderId], eventTypes }).pipe(Stream.runHead) if (Option.isSome(done)) yield* viewModels.caughtUp(done.value.position) }) /** Waits until the saga has confirmed or rejected an order. */ export const settled = (orderId: string) => reached(orderId, "OrderConfirmed", "OrderRejected") export const program = Effect.gen(function*() { const orders = yield* Orders const stocks = yield* Stocks // An order the previous version of the shop stored, before quantities existed. yield* EventStore.EventStore.use((store) => store.append({ aggregateName: "order", aggregateId: "order-0", expectedVersion: 0, events: [{ type: "OrderPlaced", schemaVersion: 1, payload: { sku: "pear" } }] }) ) yield* settled("order-0") const legacy = yield* orders.state("order-0") yield* Effect.log("order-0 is read as", legacy) yield* stocks.commands.add("apple", { quantity: 5 }) yield* orders.commands.place("order-1", { sku: "apple", quantity: 3 }) // The saga holds the stock and asks for the payment: the customer has 15 seconds to pay. yield* reached("order-1", "PaymentRequested", "OrderRejected") yield* orders.commands.pay("order-1", {}) yield* settled("order-1") const rejected = yield* Effect.flip(orders.commands.place("order-1", { sku: "apple", quantity: 1 })) yield* Effect.log("placing order-1 again fails with", rejected._tag) yield* orders.commands.place("order-2", { sku: "apple", quantity: 3 }).pipe( Metadata.withMetadata({ userId: "bob", channel: "mobile" }) ) yield* settled("order-2") const history = yield* orders.history("order-2") yield* Effect.log("order-2 history", history.map(({ metadata, type }) => ({ type, metadata }))) const list = yield* OrderList.use(Ref.get) yield* Effect.log("order list", Object.fromEntries(list)) const stock = yield* stocks.state("apple") yield* Effect.log("apple stock", stock) // How many `place` commands ran, and how long they took, by outcome. const timings = (yield* Metric.snapshot).flatMap((metric) => metric.id === AggregateService.MetricNames.commandDuration && metric.type === "Histogram" && metric.attributes?.["aggregate"] === "order" && metric.attributes["command"] === "place" ? [{ outcome: metric.attributes["outcome"], count: metric.state.count, totalMs: metric.state.sum }] : [] ) yield* Effect.log("order.place timings", timings) return { orders: Object.fromEntries(list), stock, legacy, rejected: rejected._tag, history, timings } }) ``` examples/tutorial/main.ts ```ts /** Runs the tutorial's shop on the in-memory store: `pnpm tutorial`. */ import { NodeRuntime } from "@effect/platform-node" import { Effect } from "effect" import { program, Shop } from "./shop.ts" program.pipe(Effect.provide(Shop), NodeRuntime.runMain) ``` ```bash pnpm tutorial ``` Next, [the bank](/tutorial/bank/) puts the same pieces in an app you can poke at, and the guides go deeper into [aggregates](/guides/aggregates/), [commands](/guides/commands/), [view models](/guides/view-models/) and [sagas](/guides/sagas/). # The workshop > The third example: a bank and a shop on a real server, for a room full of phones and a projector. [Run the workshop](https://github.com/davidnussio/effect-cqrs/blob/main/examples/workshop/README.md)It needs a server, so there is no copy on this site. Its README says how to run it and put it online on free plans. The third example. [The bank](/tutorial/bank/) runs in the browser, for one person. The workshop runs the framework on a server, for a room: participants join from their phones, and a control room on the projector shows what the system does while they use it. Its code is in [`examples/workshop`](https://github.com/davidnussio/effect-cqrs/tree/main/examples/workshop). ## What the participants do * They join with a nickname, which is personal data: the framework encrypts it with a key of their own. * They add money to their account and send it to each other. A transfer touches two accounts, so a saga completes it, and refunds the sender when the other account has closed. * They buy from a shared catalog with real stock, paying from their account. When a product is sold out they can wait in its queue, and cancel for a refund until the goods arrive. * They close their account: the balance is paid out, and the key of their nickname is destroyed. The events stay, and the nickname reads `[forgotten]`. A bot restocks the shop, and simulated shoppers keep it busy when the room is quiet. Both go through the same aggregate services as the HTTP API and sign their events. ## Two orders, one piece The queue lives in the product aggregate. Restocking hands goods to the waiting orders in the same command, in order of arrival, so two orders can’t get the same piece, and while a queue exists newcomers join it rather than take goods ahead of it: examples/workshop/domain/shop.ts ```ts /** * Hands the available goods to the queue in order. The first order that doesn't * fit stops it: a small order behind it doesn't jump the queue. */ const allocate = (available: number, queue: ReadonlyArray): ReadonlyArray => { const events: Array = [] for (const waiting of queue) { if (waiting.quantity > available) break available -= waiting.quantity events.push(ProductEvent.cases.BackorderAllocated.make(waiting)) } return events } ``` An order touches the account, the product and itself, so a saga carries it. Each step does nothing the second time, so the saga can run again after a restart. The presenter can make the next order break after it is paid, and the room watches the saga give the money back: examples/workshop/domain/shop.ts ```ts const CheckoutWorkflow = Saga.toLayer( Checkout, ({ backorder, customer, orderId, productId, quantity, total }) => Effect.gen(function*() { const accounts = yield* Accounts const products = yield* Products const orders = yield* Orders const chaos = yield* Chaos const reject = (reason: string) => Saga.command("reject", { execute: orders.commands.reject(orderId, { reason }) }) const refund = (reason: string) => Saga.command("refund", { execute: accounts.commands.refundOrder(customer, { orderId, amount: total, reason }) }) const paid = yield* Saga.command("pay", { error: Schema.Union([InsufficientFunds, AccountNotOpen]), execute: accounts.commands.payOrder(customer, { orderId, amount: total }) }).pipe( Effect.as(true), Effect.catch((error) => reject(error._tag === "InsufficientFunds" ? reasons.insufficientFunds : reasons.accountClosed).pipe( Effect.as(false) ) ) ) if (!paid) return yield* Saga.command("mark-paid", { execute: orders.commands.markPaid(orderId, {}) }) // An activity, so a saga that runs again after a restart decides the same way. const broken = yield* Activity.make({ name: "warehouse", success: Schema.Boolean, execute: chaos.consume }) if (broken) { yield* refund(reasons.warehouse) return yield* reject(reasons.warehouse) } const taken = yield* Saga.command("take", { error: OutOfStock, execute: products.commands.take(productId, { orderId, quantity }) }).pipe(Effect.as(true), Effect.catchTag("OutOfStock", () => Effect.succeed(false))) if (taken) { yield* Saga.command("fulfil", { execute: orders.commands.fulfil(orderId, {}) }) } else if (backorder) { yield* Saga.command("enqueue", { execute: products.commands.enqueue(productId, { orderId, quantity }) }) // If goods were there already, the order is fulfilled by now and this does nothing. yield* Saga.command("mark-backordered", { execute: orders.commands.markBackordered(orderId, {}) }) } else { yield* refund(reasons.outOfStock) yield* reject(reasons.outOfStock) } }) ) ``` ## Reading your own writes The screens read view models, which run a moment behind the commands. Every command the UI sends answers with the position of its last event, and the reads that follow it, the movements, the orders, the participants and the catalog, send it back in a `Read-After` header. [`CommandApi.ReadAfter`](/guides/integration/#reading-your-own-writes) holds each of those reads until the view models have handled that position, so a participant who buys something sees the order at once, while the other phones catch up on their own a moment later. When the view models are more than five seconds behind, because the room is busy or a handler keeps failing, the server answers `503` with `NotCaughtUp`. The UI then reads again without the header, and shows a screen that is a little old rather than an error. ## The control room The server follows the log once and sends each event on two Server-Sent Events streams. The participants’ stream hides personal fields; the presenter’s stream shows them decrypted, next to the payload as the store keeps it, encrypted. Each event carries who caused it and the correlation of the request it follows from, so a saga’s steps line up behind the click that started them. The control room also checks that the money adds up. The participants’ balances come from one view model, the books from another, both folded from the same events. When the two are at the same position their totals must agree, and while one is behind the room says so instead of guessing: eventual consistency on the projector. Everything the streams send is described in an AsyncAPI 3 document, built from the aggregates with `EventCatalog.document`, and the REST API in OpenAPI, built from the same `HttpApi` the UI calls through a typed client. See the [source](https://github.com/davidnussio/effect-cqrs/tree/main/examples/workshop) and its [`README`](https://github.com/davidnussio/effect-cqrs/blob/main/examples/workshop/README.md). # Using with AI > What to give a coding agent or a chat so it writes effect-cqrs code that compiles. effect-cqrs is young and built on Effect 4, so a model’s training data knows little about either. Give it the documentation instead. ## A guide for coding agents The package ships `LLMS.md`, a short guide to the modules, the patterns to follow, the mistakes to avoid and how to test. Point your agent at it from `AGENTS.md` or `CLAUDE.md`: ```md Before writing code that uses effect-cqrs, read node_modules/effect-cqrs/LLMS.md. ``` It is also [on GitHub](https://github.com/davidnussio/effect-cqrs/blob/main/LLMS.md). Effect has [its own guide](https://github.com/Effect-TS/effect/blob/main/LLMS.md) for the code around it. ## The documentation as text The site follows the [llms.txt](https://llmstxt.org/) convention: | File | What it holds | | ----------------------------------- | --------------------------------------------------------------------------- | | [`llms.txt`](/llms.txt) | An index of the files below | | [`llms-full.txt`](/llms-full.txt) | Every page, the API reference included, as one Markdown file | | [`llms-small.txt`](/llms-small.txt) | The guides without asides or the API reference, for smaller context windows | Each page is also served as Markdown: replace the trailing slash of its URL with `.md`, as in [`guides/aggregates.md`](/guides/aggregates.md). The button under each page title copies that Markdown, and its menu opens the page in Claude or ChatGPT. # Aggregates > Events, a projection that folds them into state, and commands that decide on new events. An aggregate is the unit that commands change. It has three parts: a schema of the events that can happen to it, a projection that folds those events into its current state, and commands that look at the state and decide which events happen next. ```ts import { DateTime, Effect, Schema } from "effect" import { Aggregate } from "effect-cqrs" const TodoEvent = Schema.TaggedUnion({ TodoCreated: { title: Schema.NonEmptyString, createdAt: Schema.DateTimeUtc }, TodoDeleted: { deletedAt: Schema.DateTimeUtc } }) class TodoAlreadyDeleted extends Schema.TaggedError()("TodoAlreadyDeleted", {}) {} type TodoState = { readonly title: string; readonly deleted: boolean } const Todo = Aggregate.make()({ name: "todo", events: TodoEvent, initial: { title: "", deleted: false }, projection: { TodoCreated: (state, { title }) => ({ ...state, title }), TodoDeleted: (state) => ({ ...state, deleted: true }) } }) .command("createTodo", { payload: Schema.Struct({ title: Schema.NonEmptyString }), handler: Effect.fnUntraced(function*(_, { title }) { return TodoEvent.cases.TodoCreated.make({ title, createdAt: yield* DateTime.now }) }) }) .command("deleteTodo", { payload: Schema.Struct({}), handler: Effect.fnUntraced(function*(state) { if (state.deleted) return yield* new TodoAlreadyDeleted() return TodoEvent.cases.TodoDeleted.make({ deletedAt: yield* DateTime.now }) }) }) ``` ## Events Events are a `Schema.TaggedUnion`. The store keeps them as JSON through the schema’s JSON codec, so a field like `Schema.DateTimeUtc` round-trips. `TodoEvent.cases.TodoCreated.make` builds an event and fills in its `_tag`. ## The projection The projection has one function per event tag, and it must cover every tag or the aggregate doesn’t compile. Each function takes the state, the event and an `EventMeta` (`aggregateName`, `aggregateId`, `version`, `position`, `timestamp`, the [`correlationId` and `causationId`](/guides/commands/#causation) and the command’s [`metadata`](/guides/commands/#metadata)), and returns the new state. `Aggregate.make()` takes the state type explicitly, in a curried call like `Schema.TaggedError()`. Inferred from `initial`, a literal such as `status: "open"` would widen to `string`. ## Commands `.command(name, { payload, handler })` adds a command. The payload is a schema, and the service validates every call against it before the handler runs. The handler receives the current state, the decoded payload, and the aggregate’s id and version. It returns an `Effect` with one event, an array of events, or `[]` to record nothing. It can fail with typed errors, which reach the caller unchanged, and it can use services, which the service picks up when it is built. A handler should only decide, for three reasons: * **It may run more than once.** A command that loses a race with another process reloads the state and runs again. * **The state is shared.** The service keeps recent states between commands, so a handler or projection that mutates one corrupts the next command. * **Commands on one aggregate take turns.** A handler that runs a command on its own aggregate waits for itself. ## Evolving events The log is never rewritten, so events stored months ago keep the shape they had then. To change an event’s schema, declare its older shapes with `EventVersions` and say how each one becomes the next. Projections, view models and handlers only ever see the current shape. ```ts import { EventVersions } from "effect-cqrs" // What TodoCreated looked like before it had a due date const TodoCreatedV1 = Schema.Struct({ title: Schema.String, createdAt: Schema.DateTimeUtc }) const Todo = Aggregate.make()({ name: "todo", events: TodoEvent, initial, projection, versions: { TodoCreated: EventVersions.from(TodoCreatedV1).next( TodoEvent.cases.TodoCreated, (v1) => ({ ...v1, dueDate: null }) ) } }) ``` * **Versions count from 1.** The first schema is the shape events had before the history was declared, which is what every event already stored was written with. Each `next` adds a version, and new events are written in the last one. * **The last version is the current schema.** A history that stops short of it doesn’t compile, and neither does an upcast that returns the wrong shape. * **Each upcast returns the next version without its `_tag`.** It is a plain function, and its result is checked against that version’s schema. * **Old events are upcast as they are read.** The steps are composed into one decoder per old version with `Schema.decodeTo` when the service starts, so an event five versions behind costs one decode. The store records each event’s `schemaVersion`. An event in a version the aggregate doesn’t declare, for instance one written by newer code, fails with `EventDecodeError`. ## Types from an aggregate The `Aggregate` module derives types from the definition, for code around it: | Type | What it is | | --------------------------------------------------- | ------------------------------ | | `Aggregate.State` | The state | | `Aggregate.Event` | The union of its events | | `Aggregate.CommandName` | `"createTodo" \| "deleteTodo"` | | `Aggregate.Payload` | `{ readonly title: string }` | | `Aggregate.CommandError` | `TodoAlreadyDeleted` | | `Aggregate.Context` | The services its handlers need | # Commands > Run commands, read state and history, and know how concurrent writers behave. `AggregateService.make(aggregate)` runs an aggregate against the event store. It is the `command`, `read-model` and `history` actions of moleculer-cqrs, as a typed service. ```ts import { AggregateService, EventStore } from "effect-cqrs" const program = Effect.gen(function*() { const todos = yield* AggregateService.make(Todo) // Typed call yield* todos.commands.createTodo("todo-1", { title: "Buy Milk" }) // Untyped JSON from a transport, validated before running yield* todos.dispatch({ aggregateId: "todo-2", type: "createTodo", payload: { title: "Buy Eggs" } }) yield* todos.state("todo-1") yield* todos.state("todo-1", { finishTime }) // as it was before finishTime yield* todos.history("todo-1", { payload: true }) }).pipe(Effect.provide(EventStore.layerMemory)) ``` ## Running a command `todos.commands` has one function per command, typed from the aggregate. Each takes the aggregate’s id, the payload and optionally a `commandId` and an `expectedVersion`, see [retries](#retries-and-command-ids) and [expected version](#expected-version). It returns a `CommandResult`: | Field | What it is | | ------------------------------ | ---------------------------------------------------------------------------------------------------- | | `aggregateName`, `aggregateId` | The aggregate it ran on | | `events` | The events it recorded, possibly none | | `version` | The aggregate’s version afterwards | | `position` | The store position of the last event, for [`caughtUp`](/guides/view-models/#reading-your-own-writes) | | `correlationId` | The correlation its events joined, see [causation](#causation) | The result is a receipt for what was stored, not an answer to a question. A screen that shows the change reads it as any other screen does, from [`state`](#reading) or a [view model](/guides/view-models/), and `position` lets that read wait for the command. `dispatch` takes the envelope a transport hands over, `{ aggregateId, type, payload }`, as `unknown`. It checks the envelope and decodes the payload with the command’s JSON codec, so an ISO string becomes a `DateTime`. This is the entry point to wire into an HTTP or RPC layer. The envelope can also carry a `correlationId` and a `causationId`, to continue a correlation the caller started, a `metadata` object, see [metadata](#metadata), and a `commandId` and an `expectedVersion`: ```ts yield* todos.dispatch({ aggregateId: "todo-1", type: "completeTodo", payload: {}, commandId: "6f1c…", // the same on every retry of this request expectedVersion: 1 // the version the caller saw }) ``` ## Errors A command fails with the errors its handler declares, plus: | Error | When | | ---------------------- | ------------------------------------------------------------------------------------------------------- | | `InvalidCommand` | The payload, or for `dispatch` the envelope, doesn’t match its schema | | `UnknownCommand` | `dispatch` names a command the aggregate doesn’t have | | `WrongExpectedVersion` | The aggregate isn’t at the caller’s `expectedVersion`. It carries `expectedVersion` and `actualVersion` | | `ConcurrencyConflict` | Another process kept winning the race for this aggregate | | `EventStoreError` | The store failed | | `EventDecodeError` | A stored event doesn’t match the schema of its [version](/guides/aggregates/#evolving-events) | ## Reading * `state(id)` folds the aggregate’s events into its state. With `finishTime` it gives the state as it was before that moment. `readModel` is its former name, deprecated and still there. * `history(id)` lists the aggregate’s events as `{ version, timestamp, type, correlationId, causationId, metadata }`, with the decoded event as `payload` when you ask for it. `startTime` and `finishTime` narrow the window. ## Concurrent writers Inside a service, commands on one aggregate take turns, so they can’t race. Commands on different aggregates don’t wait for each other. Writers in different processes use optimistic concurrency. A command appends its events with the version its decision was based on, and if another process got there first, it backs off, reloads the state and decides again. After `maxRetries` attempts, 5 by default, it fails with `ConcurrencyConflict`. ## Retries and command ids A client that times out doesn’t know whether its command ran, and sending it again could record the same deposit twice. Give the command a `commandId`, the same on every attempt, and the service runs it only once: ```ts const first = yield* accounts.commands.deposit("alice", { amount: 100 }, { commandId: requestId }) const again = yield* accounts.commands.deposit("alice", { amount: 100 }, { commandId: requestId }) // `again` equals `first`: same version, position, events and correlationId, and nothing new is stored ``` The id is recorded in the metadata of the events the command writes, under the reserved key `cqrs.commandId` (`AggregateService.commandIdKey`), with the command’s name under `cqrs.command` (`AggregateService.commandTypeKey`). Each aggregate remembers the ids of its latest commands, rebuilt from those events when its stream is replayed and kept in [snapshots](#snapshots). When an id comes again, the service reads the events it wrote back from the store and returns them as the first result, without running the handler. * **Ids are per aggregate.** The same id on another aggregate is another command. * **Only commands that write events are remembered.** One that writes none leaves nothing to find, so it runs again, which is harmless as long as it still decides to write nothing. * **The window is bounded.** Each aggregate remembers its last `commandIdsPerStream` commands, 100 by default. An older id runs again. * **Another command under a known id fails** with `CommandIdReused`, rather than getting a result it didn’t ask for. The payload isn’t compared: the same command with another payload under a known id gets the first result. Make ids unique per request, such as a UUID the client makes once. Ids recorded before the command’s name was have nothing to compare. Comparing payloads would mean keeping a fingerprint of each one beside the id, in plain text, where it would outlive [forgetting](/guides/personal-data/) a subject whose personal data it was made from. * **Metadata given under `cqrs.commandId` or `cqrs.command` is dropped**, so the keys only hold what the service recorded. Two processes running the same command at once are covered too: the one that loses the race reloads, finds the id and returns the winner’s result. ## Expected version A user who edits what they read a minute ago can overwrite a change made in between. Pass the version they read as `expectedVersion`, like HTTP’s `If-Match`, and the command fails with `WrongExpectedVersion` if the aggregate has moved since: ```ts yield* docs.commands.rename("doc-1", { title }, { expectedVersion: versionShown }).pipe( Effect.catchTag("WrongExpectedVersion", ({ actualVersion }) => reloadAndAskAgain(actualVersion)) ) ``` It is checked before the handler runs, and it isn’t retried: unlike a `ConcurrencyConflict` between two processes, deciding again would act on a state the user hasn’t seen. A command that comes again with its `commandId` returns its first result rather than failing, although its own events moved the version on. ## Remembered state The service keeps the state of the aggregates it used recently, including the events its own commands wrote, so a command only reads the events it hasn’t seen yet. Those can come from another process too: a state plus what came after it is always the current state. After a restart, the first command on an aggregate replays its whole stream, unless [snapshots](#snapshots) are on. ```ts const todos = yield* AggregateService.make(Todo, { maxRetries: 5, // re-decisions after losing to another process stateCacheSize: 10_000, // aggregates kept in memory, 0 to turn it off commandIdsPerStream: 100 // command ids each aggregate remembers }) ``` ## Snapshots A long aggregate takes a while to replay after a restart. With `snapshots` the service saves the state every so many events, and a cold start replays only what came after the last one: ```ts import { KeyValueStore } from "effect/persistence" const todos = yield* AggregateService.make(Todo, { snapshots: { every: 100, // events folded between two snapshots schema: Schema.Struct({ title: Schema.String, deleted: Schema.Boolean }), // the state, as JSON version: 1 // change it when the state or the projection changes } }).pipe(Effect.provide(KeyValueStore.layerSql({ table: "todo_snapshots" }))) ``` Snapshots go to a `KeyValueStore` from `effect/persistence`, which has layers for memory, files, any `effect/sql` database and the browser’s `localStorage`. * **A snapshot is only a shortcut.** One that can’t be read or saved is logged and skipped, and the state comes from the log as before. A command never fails because of one. * **The version is part of every key.** Snapshots of a state the code no longer produces would be wrong rather than unreadable, so change `version` whenever the state or the projection changes. The old ones are then ignored and rebuilt from the log. * **They are saved in the background.** Once `every` events have been folded since the last snapshot, the service starts saving one and the command carries on without waiting for it. The service needs a `Scope`, which `Layer.effect` provides, and closing it waits for the saves still running. * `state` with a `finishTime` always replays from the start. * They keep the aggregate’s latest [command ids](#retries-and-command-ids) beside the state, unsealed, as the events hold them in plain text too. ## Metrics The service records how it works through Effect’s `Metric` module, so you can see whether the remembered states and snapshots are paying off. Every series carries an `aggregate` attribute, and `AggregateService.MetricNames` has their names: | Metric | Type | What it counts | | ----------------------------------- | --------- | -------------------------------------------------------------------------------------------------------------- | | `cqrs_state_loads_total` | counter | Where a load found the state to start from: `source` is `memory`, `snapshot` or `replay` | | `cqrs_events_replayed` | histogram | Events a load read from the store on top of that state | | `cqrs_state_cache_size` | gauge | States remembered in memory | | `cqrs_state_cache_evictions_total` | counter | States dropped to make room, a sign `stateCacheSize` is too small | | `cqrs_snapshot_saves_total` | counter | Snapshots saved, by `result`: `success` or `failure` | | `cqrs_snapshot_read_failures_total` | counter | Snapshots that couldn’t be read, so the stream was replayed instead | | `cqrs_concurrency_conflicts_total` | counter | Appends that lost a race with another process | | `cqrs_command_duration` | timer | How long commands took, by `command` and `outcome`: `success`, the tag of the error, `defect` or `interrupted` | A command is timed from when it asks for its turn on the stream, so time spent behind an earlier command on the same aggregate counts. A payload that doesn’t decode, or a type the aggregate doesn’t know, fails before that and isn’t recorded. Send the metrics on with an exporter, such as `OtlpMetrics` or `PrometheusMetrics` from `effect/observability`, or read them yourself: ```ts import { Metric } from "effect" const loads = (yield* Metric.snapshot).filter( (metric) => metric.id === AggregateService.MetricNames.stateLoads && metric.attributes?.aggregate === "todo" ) ``` Metrics go to the `Metric.MetricRegistry` in context, which by default is one registry for the whole process. Provide a fresh `Map` to keep one test’s counts apart from another’s. The [bank demo](/tutorial/bank/) shows them in its **Metrics** tab. ## Causation Every stored event records why it happened, so a chain of reactions can be traced back to the request that started it: | Field | What it is | | --------------- | ----------------------------------------------------------- | | `correlationId` | Shared by everything that follows from one original request | | `causationId` | What directly caused the command that recorded the event | They travel in the fiber’s context, not in payloads, and nothing has to pass them along: * A command outside any causation starts a correlation with a fresh id, which is both ids of its events. * A view model handler runs with the event it handles as the cause. A command it runs records that event’s id, like `account/alice@4`, and keeps its correlation. * A [saga](/guides/sagas/) carries the event that started it in its workflow’s payload, so the commands it runs record that event as their cause, even after a restart. * `Causation.causedBy(stored)` does the same for other subscribers. * `Causation.withCausation({ correlationId, causationId })` sets them by hand, for instance from an incoming request’s id. ```ts import { Causation } from "effect-cqrs" yield* todos.commands.createTodo("todo-1", { title: "Buy Milk" }).pipe( Causation.withCausation({ correlationId: requestId, causationId: requestId }) ) ``` The command’s span carries the correlation as `cqrs.correlation_id`, and its `commandId`, if any, as `cqrs.command_id`. New correlations get a random UUID from `Ids.IdGenerator`, a reference a test can replace to get ids it can compare: ```ts import { Ids } from "effect-cqrs" program.pipe(Effect.provideService(Ids.IdGenerator, Ids.sequential("id"))) // id-1, id-2, … ``` ## Metadata A command can carry what the domain doesn’t decide on but you want to keep: the version of the software that ran it, the user who sent it, the address it came from. Metadata is a JSON object, recorded on every event the command writes as `metadata`, and `{}` when there is none. Projections and view model handlers find it in their `meta` argument, and `history` lists it. Like causation it travels in the fiber’s context. A command records, key by key, from the lowest precedence: 1. the metadata where its service was built, with `Metadata.layer`; 2. the metadata where it runs, added with `Metadata.withMetadata`; 3. the `metadata` of a `dispatch` envelope. ```ts import { Metadata } from "effect-cqrs" // Once for the whole application: every command records it. const AppLayer = Layer.mergeAll(Todos.layer, ViewModelsLayer).pipe( Layer.provide(Metadata.layer({ appVersion: "1.4.0" })) ) // Per request, on top of it. yield* todos.commands.createTodo("todo-1", { title: "Buy Milk" }).pipe( Metadata.withMetadata({ userId: session.userId, ip: request.remoteAddress }) ) ``` The service remembers the metadata it was built with, so the software version is recorded even by commands run where the layer that provided it isn’t visible, such as an HTTP handler on a `ManagedRuntime`. A view model or saga reacting to an event doesn’t inherit that event’s metadata: the commands it runs record their own, including the version of the process that ran them. Their `causationId` leads back to the event, and from there to its metadata. Command handlers don’t see metadata: it isn’t meant to change a decision. If it should, it belongs in the payload. ## As a service To use one service from several places, put it in a layer, as the todo example does: ```ts class Todos extends Context.Service>()("example/Todos") { static readonly layer = Layer.effect(Todos, AggregateService.make(Todo)) } ``` ## Several aggregates `CommandGateway` is one entry point for the commands of several aggregates. The caller hands over a command and the gateway finds the aggregate that takes it: ```ts import { CommandGateway } from "effect-cqrs" const gateway = yield* CommandGateway.make([Todo, Account]) yield* gateway.send({ aggregateName: "todo", aggregateId: "todo-1", type: "createTodo", payload: { title: "Buy Milk" } }) // JSON from a transport: `aggregateName` can be left out when a single aggregate has the command. yield* gateway.dispatch({ aggregateId: "todo-1", type: "completeTodo", payload: {} }) ``` `send` is typed: the aggregate’s name picks its commands, the type picks the payload, and the errors are the handler’s own plus `CommandFailure`. It also takes `commandId` and `expectedVersion`. `dispatch` takes the same envelope as `AggregateService.dispatch`, with `aggregateName` on top. When it can’t tell where a command goes, because the envelope names an aggregate the gateway doesn’t have, or names none and no aggregate or more than one has the command, it fails with `UnroutableCommand`, whose `candidates` lists the aggregates that have it. The gateway builds the service of each aggregate with the default options. To choose them, or to share a service and its state cache with code that uses it directly, give the gateway the service instead: ```ts const todos = yield* AggregateService.make(Todo, { snapshots }) const gateway = yield* CommandGateway.make([todos, Account]) ``` `gateway.services.todo` is the service of `todo`, for `state` and `history`. ### Interceptors Interceptors run around every command the gateway routes: to authorize it, log it, or record metadata. Each gets the command’s `aggregateName`, `aggregateId` and `type`, and the command to run. It can fail instead of running it, or run it with more context: ```ts const authorize: CommandGateway.Interceptor = (command, next) => Effect.gen(function*() { const user = yield* CurrentUser if (!user.may(command.type)) return yield* new Unauthorized({ type: command.type }) return yield* Metadata.withMetadata(next, { userId: user.id }) }) const gateway = yield* CommandGateway.make([Todo, Account], { interceptors: [authorize, log] }) ``` The first interceptor runs outermost. An interceptor’s errors and requirements are added to every command of the gateway, and they are read from its type, so declare it as an `Interceptor`: an interceptor written inline in the options doesn’t compile. Interceptors run in the caller’s fiber, so `CurrentUser` here comes from the caller, such as an HTTP handler, and not from the layer that built the gateway. Commands run through `gateway.services`, or through `CommandApi`, skip them. To use the gateway from several places, put it in a layer like a service: ```ts const makeGateway = CommandGateway.make([Todo, Account], { interceptors: [authorize] }) class Gateway extends Context.Service>()("app/Gateway") { static readonly layer = Layer.effect(Gateway, makeGateway) } ``` # Integration > Serve aggregates over HTTP, describe their events, publish them to other systems and keep tenants apart. An aggregate service runs in one process. These modules connect it to the rest of a system: | Module | For | | -------------- | ------------------------------------------------------------------------------ | | `CommandApi` | Commands and reads over HTTP, with OpenAPI | | `EventCatalog` | JSON Schema of events and commands, and a guard against breaking stored events | | `Outbox` | Publishing events to Kafka, NATS, SQS and the like | | `LayerMap` | One store per tenant, a pattern rather than a module | `examples/server` puts the first one together with a view model, SQLite or PostgreSQL and OpenTelemetry. Run it with `pnpm example:server`. ## HTTP `CommandApi.group` builds an `HttpApiGroup` from an aggregate, and `CommandApi.layer` implements it with an aggregate service: ```ts import { HttpApi, HttpApiBuilder, HttpApiScalar } from "effect/http-api" import { AggregateService, CommandApi } from "effect-cqrs" const TodoApi = CommandApi.group(Todo, { state: TodoState, // a schema of the state, for GET /todo/:aggregateId errors: [TodoAlreadyCreated, TodoNotFound, TodoAlreadyDeleted] }) class Api extends HttpApi.make("api").add(TodoApi) {} const Routes = Layer.mergeAll( HttpApiBuilder.layer(Api, { openapiPath: "/openapi.json" }).pipe( Layer.provide(CommandApi.layer(Api, "todo", AggregateService.make(Todo))) ), HttpApiScalar.layer(Api, { path: "/docs" }) ) ``` [effect-cqrs-http-example](https://github.com/davidnussio/effect-cqrs-http-example) is this example as a repository of its own, with a server and SQLite storage, to clone and run. | Endpoint | Does | | ----------------------------------- | ------------------------------------------------------------------------------------------------------ | | `POST /todo/:aggregateId/` | Runs the command with the body as payload, and answers with the `CommandResult` | | `GET /todo/:aggregateId/history` | `history`, with optional `payload`, `startTime`, `finishTime`, `afterVersion` and `limit` in the query | | `GET /todo/:aggregateId` | `state`, with an optional `finishTime`. Only with the `state` option | The group is an ordinary `HttpApiGroup`: `name` and `prefix` change its identifier and path, and middleware, OpenAPI annotations and `HttpApiClient` work as with any other. A request without a body runs a command whose payload has no required fields. `history` answers with at most 1,000 events, the `historyLimit` option of `CommandApi.layer`, and a client reads the next page with `afterVersion` set to the version of the last entry it got. Two headers reach the command. `Idempotency-Key` becomes its [`commandId`](/guides/commands/#retries-and-command-ids), so a client that retries after a timeout gets the first result back instead of a second change. The body sent with a known key isn’t compared, so a key belongs to one request. `If-Match` is the version the client based the command on, as `3` or `"3"`, and a stream that has moved on answers `409`. `If-Match: *` asks for no version in particular. Failures map to status codes: | Status | When | Body | | ------ | ------------------------------------------------------- | --------------------------------------------------- | | `400` | The body isn’t JSON or doesn’t match the payload schema | `InvalidCommand` | | `409` | Another writer kept winning the race for the stream | `ConcurrencyConflict` | | `409` | The stream is no longer at the version in `If-Match` | `WrongExpectedVersion` | | `422` | The handler failed with an error in `errors` | The error, with its fields | | `422` | The handler failed with another error | `CommandRejected`, with the error’s tag and message | | `500` | The store, decoding or the keys failed | Empty | A command’s errors are Effect types rather than schemas, so the ones a client should see in full are listed in `errors`. They are answered with `422`, or with their own `httpApiStatus`: `Schema.TaggedError()("X", { ... }, { httpApiStatus: 404 })`. `HttpApiTest.groups(Api, ["todo"])` gives a typed client that calls the handlers without a server, which is how `test/CommandApi.test.ts` tests them. ### Reading your own writes `GET /todo/:aggregateId` folds the aggregate’s events on every request, so it always shows the latest command. Your own query endpoints read view models, which run a moment behind. A client that runs a command and then reads a list can miss its own change. Every command answers with the store position of its last event, as `position` in the body and in an `Event-Position` header. Add `CommandApi.ReadAfter` to the query endpoints, and a request that sends the position back as `Read-After` waits until the view models of this process have handled it: ```ts const Catalog = HttpApiGroup.make("catalog") .add(HttpApiEndpoint.get("available", "/books/available", { success: Books })) .middleware(CommandApi.ReadAfter) class Api extends HttpApi.make("api").add(BookApi).add(Catalog) {} const Server = HttpApiBuilder.layer(Api).pipe( Layer.provide([BookHandlers, CatalogHandlers]), Layer.provide(CommandApi.layerReadAfter({ timeout: "2 seconds" })), Layer.provide(ViewModel.layer([CatalogViewModel])) ) ``` ```ts const response = await fetch("/book/book-1/lend", { method: "POST", body }) const position = response.headers.get("event-position")! const books = await fetch("/books/available", { headers: { "read-after": position } }) ``` A request without the header runs at once. One that waits longer than `timeout`, 5 seconds by default, is answered with `503` and `NotCaughtUp`: a view model is far behind, or stalled on a failing handler, and `viewModels.status` says which. A header that isn’t a position is answered with `400`. In a browser on another origin, list `Event-Position` in `Access-Control-Expose-Headers` so the page can read it. `layerReadAfter` waits for the view models that run in the same process, as [`caughtUp`](/guides/view-models/#reading-your-own-writes) does. With several replicas, send the read to a replica that runs the view models it needs. ## Event catalog `EventCatalog` describes events and commands as JSON Schema, for teams that consume them: ```ts import { EventCatalog } from "effect-cqrs" EventCatalog.events(Account) // every event, with every version from EventVersions EventCatalog.commands(Account) // every command payload EventCatalog.document([Account, Transfer], { title: "Bank", version: "1.0.0" }) // AsyncAPI 3.0 ``` The schemas describe the JSON form of each event, as `Schema.toCodecJson` encodes it. In the AsyncAPI document each aggregate has a channel of events it sends and one of commands it receives. A message keeps its older versions under `x-versions` and the paths of its personal fields under `x-personal-fields`. ### Compatibility guard A change to an event schema can make events already in the store unreadable. The guard keeps a few stored payloads of every event type and version in a file, and a test checks that the current aggregate still decodes them, the way a store read does: ```ts // Once in a while, against a copy of production data: const samples = yield* EventCatalog.captureSamples([Account]) yield* EventCatalog.saveSamples("test/samples/account.json", samples) // In a test: it.effect("still reads stored events", () => Effect.gen(function*() { const samples = yield* EventCatalog.readSamples("test/samples/account.json") yield* EventCatalog.checkCompatibility(Account, samples) }).pipe(Effect.provide(NodeFileSystem.layer))) ``` `checkCompatibility` fails with `IncompatibleSamples`, which lists each sample that no longer decodes and why. It returns how many samples it checked. `saveSamples` adds to the file and leaves the samples already there, so it keeps the shapes the store held over time. ## Outbox An outbox publishes events to another system as integration events: ```ts import { Outbox } from "effect-cqrs" const TodoOutbox = Outbox.make({ name: "todo-outbox", aggregates: [Todo], messages: { TodoCreated: ({ title }, { aggregateId }) => ({ type: "todo.created.v1", payload: { id: aggregateId, title } }), TodoDeleted: (_, { aggregateId }) => ({ type: "todo.deleted.v1", payload: { id: aggregateId } }) }, publish: (messages) => producer.send(messages), // an Effect batchSize: 100, linger: "10 millis" }) const Publishing = Outbox.layer([TodoOutbox]).pipe( Layer.provide(Layer.merge(SqliteEventStore.layer(), SqliteEventStore.layerCheckpoints())) ) ``` `messages` maps events by tag, and only those are published. A mapping that returns `undefined` skips the event. Without `messages`, every event is published as `.` with its JSON form as payload. Each `Message` carries the payload with the event’s stream, version, position, timestamp, `correlationId` and `causationId`. An outbox is a [view model](/guides/view-models/): it reads the store from its checkpoint, so events stored while it was down are published when it starts, and nothing is lost between storing an event and publishing it. Delivery is at least once: * Messages go out in batches of up to `batchSize`, when a batch is full or `linger` after its first message. The checkpoint never moves past a message that wasn’t published. * A failed batch is retried in order, with growing pauses, until it succeeds. Once `batchSize` messages wait, the outbox stops reading the store, and the `status` of `Outbox.Outboxes` shows the failure. * After a crash, messages published since the last checkpoint are published again. `Message.id` is the event’s id, such as `todo/a@1`, and stays the same, so consumers deduplicate on it. Checkpoints should be durable. With `Checkpoints.layerMemory` an outbox publishes everything again after every restart. In tests, `Outboxes`’ `published(position)` waits until everything up to a position is published. Caution Events reach the mappings decrypted. Leave personal fields out of integration events, or other systems keep data that `PersonalData.forget` can no longer erase. ## Tenants Every service here works on whatever `EventStore` it is given, so tenants are kept apart by giving each one its own store. Effect’s `LayerMap` builds the layers of a tenant the first time it is asked for, keeps them while they are used, and releases them after `idleTimeToLive`: ```ts import { LayerMap } from "effect" class Tenants extends LayerMap.Service()("app/Tenants", { lookup: (tenant: string) => { const tables = { eventsTable: `${tenant}_events`, checkpointsTable: `${tenant}_checkpoints` } return Layer.mergeAll(Todos.layer, ViewModel.layer([TodoList])).pipe( Layer.provideMerge(Layer.mergeAll( SqliteEventStore.layer(tables), SqliteEventStore.layerCheckpoints(tables), Layer.fresh(TodoListStore.layer) )) ) }, dependencies: [SqliteClient.layer({ filename: "events.sqlite" })], idleTimeToLive: "5 minutes" }) {} const listForTenant = (tenant: string) => listTodos.pipe(Effect.provide(Tenants.get(tenant))) ``` * The tenants share the database client of `dependencies`, and each has its own tables, so the same aggregate id in two tenants is two streams. PostgreSQL works the same way, with `PostgresEventStore`. Another database per tenant also works: put the client in `lookup`. * A layer value is built once per `LayerMap`, so in-memory state of a tenant’s own, like the list above, needs `Layer.fresh`. * Table names come from the tenant id. `sql(name)` quotes identifiers, but check the id against a known alphabet before it gets there. * View models and outboxes of a tenant run while its layers are alive. A tenant evicted for being idle catches up from its checkpoints the next time it is used. * Personal data keys go per tenant too, with `PersonalData.layerSql({ table })`. `test/MultiTenancy.test.ts` runs this on SQLite and PostgreSQL. # Personal data > Forget a person without rewriting the log, by encrypting their data with a key you can destroy. The log is append-only, so an event can’t be edited or deleted when someone asks to be forgotten, as the GDPR allows. Crypto-shredding gets around that: the personal fields of an event are encrypted with a key that belongs to the person they describe, the *subject*, and forgetting the subject destroys the key. The event stays where it is with its amounts, dates and ids, so streams still replay and balances still add up, while its personal fields read as a replacement. ## Marking fields `PersonalData.field` marks a field of an event and says what it reads as once its subject is forgotten: ```ts import { PersonalData } from "effect-cqrs" const AccountEvent = Schema.TaggedUnion({ AccountOpened: { owner: PersonalData.field(Schema.String, { redacted: "[forgotten]" }), email: Schema.optional(PersonalData.field(Schema.String, { redacted: "forgotten@example.invalid" })), currency: Schema.Literals(["CHF", "EUR"]) }, TransferSent: { to: Schema.String, beneficiary: PersonalData.field(Schema.String, { redacted: "[forgotten]", subject: "to" }), amount: Schema.Number } }) ``` * **The subject** is the aggregate id of the stream, unless `subject` names a property next to the field. Above, `beneficiary` belongs to the account in `to`, so forgetting the receiver of a transfer leaves the sender’s events readable. * **The replacement** has to be a valid value of the field. It is checked when the field is defined. * **Where a field can be:** a property of the event, of a nested struct, optional or nullable. Building an aggregate fails when a personal field sits in an array, a record or an untagged union, is transformed after being marked, or names a subject that isn’t next to it or isn’t always a string. Mark the transformed schema instead: `PersonalData.field(Schema.String.pipe(Schema.decodeTo(...)), ...)`. Schemas stay plain: a personal field is an ordinary string to projections, handlers and tests. Encryption happens where `AggregateService` writes to and reads from the store, so nothing else can write a personal field unencrypted. ## Keys An aggregate with personal fields needs `PersonalData.Keys`, and building its service fails without it. Each subject gets its own AES-256 key the first time it is written. The key store only holds that key wrapped by a `MasterKey`: ```ts const Keys = PersonalData.layerSql().pipe( Layer.provide(PersonalData.layerMasterKeyConfig()), Layer.provide(SqliteClient.layer({ filename: "./data/keys.sqlite" })) ) ``` | Key store | For | | ---------------------------------- | ---------------------------------------------------------------------------------------------- | | `PersonalData.layerMemory()` | Tests and the browser. Keys are gone after a restart, and with them everything they encrypted. | | `PersonalData.layerSql({ table })` | SQLite or PostgreSQL, through any `effect/sql` client | | Master key | For | | ------------------------------------------ | --------------------------------------------------------------------------------------------------------------------------------------------------------- | | `layerMasterKeyConfig()` | 32 bytes in base64 from `PERSONAL_DATA_MASTER_KEY`, with its id in `PERSONAL_DATA_MASTER_KEY_ID`, and earlier keys in `PERSONAL_DATA_MASTER_KEY_PREVIOUS` | | `layerMasterKey({ keyId, key, previous })` | A key you load yourself. Keys wrapped by older master keys stay readable while they are in `previous`. | | `layerMasterKeyEphemeral` | A random key for tests and demos | | Your own `MasterKey` | A KMS such as AWS KMS, [Google Cloud KMS](#on-google-cloud-kms) or HashiCorp Vault, which keeps the master key out of the process | A `MasterKey` is two functions, `wrap` and `unwrap`, so implementing one on a KMS is a call to its encrypt and decrypt endpoints. Keep the key store apart from the events: another database or at least another client, with its own permissions and backups. A copy of the keys outlives a forget for as long as the copy is kept, so let backups of the key table expire. `layerSql` creates its table on startup. To create it with your own migrations instead, turn that off and take the migrations from `PersonalData.migrations`, the same on SQLite and PostgreSQL and in the shape of the stores’ `migrations`: ```ts const migrations = PersonalData.migrations({ table: "personal_data_keys" }) const Keys = PersonalData.layerSql({ table: "personal_data_keys", migrate: false }) ``` Unwrapped keys stay in memory for `cacheTtl`, up to `cacheSize` subjects, 10,000 by default. Requests for a subject that isn’t in memory share one read of the store and one call to the master key, and writers racing for a new subject create a single key, so a KMS is called once rather than once per writer. ### Rotating the master key A new master key needs a new `keyId`, and wraps the keys of subjects created from then on. Keys of earlier subjects stay wrapped by the master key that created them, so that one goes in `previous` until they are moved: 1. Deploy the new master key as the current one, with the old one in `previous`. With `layerMasterKeyConfig()`, that is `PERSONAL_DATA_MASTER_KEY_PREVIOUS`, as `id:base64` pairs separated by commas. 2. Run `PersonalData.rewrap()` once, from any process. It wraps every subject key with the current master key, a page at a time, and returns how many it moved. The keys themselves don’t change, so no event is rewritten, and running it again only moves what is left. 3. Drop the old key from `previous`. A [KMS](#on-google-cloud-kms) that rotates versions of one key behind the same name needs none of this. ### On Google Cloud KMS With [`@google-cloud/kms`](https://www.npmjs.com/package/@google-cloud/kms), the master key is a symmetric crypto key that never leaves Cloud KMS. `wrap` and `unwrap` call its `encrypt` and `decrypt`. [effect-cqrs-kms](https://github.com/davidnussio/effect-cqrs-kms) runs this code against your own Cloud KMS, with the steps to set it up, rotate the key and forget a subject: ```ts import { KeyManagementServiceClient } from "@google-cloud/kms" import { Config, Effect, Layer } from "effect" import { PersonalData } from "effect-cqrs" /** * Wraps subject keys with a symmetric Cloud KMS key, named like * `projects/acme/locations/europe-west6/keyRings/app/cryptoKeys/personal-data`. */ const layerGoogleKms = (keyName: string) => Layer.effect( PersonalData.MasterKey, Effect.gen(function*() { const client = yield* Effect.acquireRelease( Effect.sync(() => new KeyManagementServiceClient()), (client) => Effect.promise(() => client.close()) ) // Binds a wrapped key to its subject and id, so it can't be swapped for another. const context = (subject: string, keyId: string) => Buffer.from(JSON.stringify([subject, keyId])) const call = (operation: "wrap" | "unwrap", subject: string, request: () => Promise) => Effect.tryPromise({ try: request, catch: (cause) => new PersonalData.PersonalDataError({ operation, subject, message: "Cloud KMS call failed", cause }) }) return PersonalData.MasterKey.of({ keyId: keyName, wrap: (dataKey, { keyId, subject }) => call("wrap", subject, () => client.encrypt({ name: keyName, plaintext: dataKey, additionalAuthenticatedData: context(subject, keyId) })).pipe(Effect.map(([{ ciphertext }]) => new Uint8Array(ciphertext as Uint8Array))), // Keys wrapped by an earlier crypto key name it as `masterKeyId`. unwrap: (wrapped, { keyId, masterKeyId, subject }) => call("unwrap", subject, () => client.decrypt({ name: masterKeyId, ciphertext: wrapped, additionalAuthenticatedData: context(subject, keyId) })).pipe(Effect.map(([{ plaintext }]) => new Uint8Array(plaintext as Uint8Array))) }) }) ) const Keys = PersonalData.layerSql().pipe( Layer.provide(Layer.unwrap(Effect.map(Config.String("PERSONAL_DATA_KMS_KEY"), layerGoogleKms))), Layer.provide(SqliteClient.layer({ filename: "./data/keys.sqlite" })) ) ``` * **Access.** The client signs in with Application Default Credentials. The service account needs `roles/cloudkms.cryptoKeyEncrypterDecrypter` on the key, and nothing else in Cloud KMS. * **Rotation.** Rotate the key in Cloud KMS as often as you like: `decrypt` uses the version a key was wrapped with, so nothing changes here. Moving to another crypto key only takes a new `PERSONAL_DATA_KMS_KEY`. Keys wrapped by the old one record its name as `masterKeyId`, and stay readable while the service account may still use it. * **Calls.** Cloud KMS bills each operation: one `encrypt` for each new subject, and one `decrypt` each time a subject’s key isn’t in memory, see `cacheTtl` above. * **Forgetting** deletes the subject’s wrapped key from the key store, so Cloud KMS has nothing left to decrypt. Disabling or destroying the crypto key does the opposite: every subject becomes unreadable at once. Google also recommends checking each request’s CRC32C, with `plaintextCrc32c` and `ciphertextCrc32c`, to catch data corrupted on the way. Add those fields to the requests if you want that check. ### On Node Encryption runs on WebCrypto, which works in Node and in browsers but is asynchronous. On Node, `PersonalDataNode.layerCipher` encrypts with `node:crypto` instead, synchronously, which makes a personal field about half as expensive to read. Both write the same format, so data written with one reads with the other. It is a module of its own, so that code for browsers never imports `node:crypto`: ```ts import * as PersonalDataNode from "effect-cqrs/PersonalDataNode" const Keys = PersonalData.layerSql().pipe( Layer.provide(PersonalData.layerMasterKeyConfig()), Layer.provide(PersonalDataNode.layerCipher), Layer.provide(SqliteClient.layer({ filename: "./data/keys.sqlite" })) ) ``` The cipher is a `PersonalData.Cipher`, provided to the keys and the master key when they are built. ## Forgetting ```ts yield* PersonalData.forget("alice") ``` * **The key is destroyed** and the subject is marked as forgotten for good. Commands that would record new personal data of the subject fail with `SubjectForgotten`. * **The request is recorded** as a `SubjectForgotten` event in the stream `PersonalData.streamName`, with the current causation and metadata, and no personal data. Running it again records nothing new. * **States and snapshots follow.** A state in memory that holds the subject’s data is rebuilt the next time it is used. Snapshots are encrypted with the key of every subject they hold data of, so the snapshot becomes unreadable and is deleted on the next read. Other processes keep using a key they have in memory until it expires, a minute by default. Set `cacheTtl` on the key store to change that. ## View models View models receive events with their personal fields decrypted, so what they store is in plaintext. `onForget` lets one erase it: ```ts const Owners = ViewModel.make({ name: "owners", aggregates: [Account], handlers: { AccountOpened: ({ owner }, { aggregateId }) => OwnerStore.set(aggregateId, owner) }, onForget: (subject) => OwnerStore.delete(subject), dispose: OwnerStore.clear }) ``` `onForget` runs when the `SubjectForgotten` event comes up, in order with the other events and again on every replay. Events after it already carry the replacements. A view model without `onForget` only loses the data when it is replayed. ## What it doesn’t cover * **Fields marked later.** Events stored before a field became personal keep it in plaintext, because the log is never rewritten. They still decode. * **Copies outside effect-cqrs.** Logs, traces, caches and exports that saw a decrypted value keep it. Don’t log payloads of events with personal fields. * **Identifiers.** The subject id is stored in plaintext next to each encrypted field, so use an opaque id rather than an email address. * **Deleted bytes.** Databases keep overwritten rows for a while in free pages, dead tuples or their write-ahead log, until they are vacuumed. * **Copies within a stream.** An encrypted field is bound to its stream, event type, place in the event and subject, so it doesn’t decrypt anywhere else. It isn’t bound to one event: someone who can write to the store can copy it into another event of the same type in the same stream. The cost is one AES-GCM operation per personal field on every write and read, see [performance](/reference/performance/). # Sagas > Run processes that span several aggregates as durable workflows, started by events. A command changes one aggregate. A process that touches several, like a transfer between two accounts, is a saga: it reacts to an event and runs more commands, possibly over a long time, and undoes what it did when a later step is refused. `Saga` runs each saga as a workflow from `effect/workflow`. The workflow engine keeps the outcome of every step, so a workflow that is replayed after a restart skips the commands it already ran, and starting the same saga twice joins the first run. ```ts import { Effect, Layer, Schema } from "effect" import { WorkflowEngine } from "effect/workflow" import { Saga } from "effect-cqrs" // What one run of the saga is about, and what identifies it. const Transfer = Saga.workflow("Transfer", { payload: { transferId: Schema.String, from: Schema.String, to: Schema.String, amount: Schema.Int }, idempotencyKey: ({ transferId }) => transferId }) // The process: each command is an activity whose outcome the engine keeps. const TransferWorkflow = Saga.toLayer(Transfer, ({ amount, from, to, transferId }) => Effect.gen(function*() { const bank = yield* Bank yield* Saga.command("credit", { error: AccountNotOpen, execute: bank.commands.receiveTransfer(to, { transferId, from, amount }) }).pipe( Effect.catchTag("AccountNotOpen", () => Saga.command("refund", { execute: bank.commands.refundTransfer(from, { transferId, to, amount, reason: "AccountNotOpen" }) })) ) })) // The trigger: starts a run for each event. const Transfers = Saga.make({ name: "transfers", aggregates: [Account], handlers: { TransferSent: ({ amount, to, transferId }, { aggregateId }) => Saga.start(Transfer, { transferId, from: aggregateId, to, amount }) } }) const TransferSaga = Saga.layer([Transfers]).pipe( Layer.provideMerge(TransferWorkflow), Layer.provide(WorkflowEngine.layerMemory) ) ``` ## The trigger `Saga.make` takes the same handlers as a view model, and is one: it reads the event store from its own checkpoint, retries a failing handler with backoff, and sees every event at least once. Its handlers only start workflows with `Saga.start`, which returns as soon as the engine has the run, so the trigger never waits for a saga to finish. `Saga.layer` runs triggers in the background. Workflows run commands, so they sit on top of the command side; keeping triggers out of `ViewModel.layer` avoids providing the view models to the commands they depend on. It provides `Saga.Sagas`, which manages the triggers as `ViewModel.ViewModels` manages view models: `status`, `caughtUp`, and, for a trigger made with `onFailure: ViewModel.park(...)`, the events it parked and `retryParked`. A trigger can still go in `ViewModel.layer`, where `replay` starts every saga again and `caughtUp` waits until the sagas have started. ## The workflow `Saga.workflow` is `Workflow.make` with a `causation` field added to the payload. A workflow runs in the engine’s fibers, not in the trigger’s, so that field is how its commands learn their [cause](/guides/commands/#causation): `Saga.start` fills it with the event being handled, and `Saga.toLayer` applies it to the body. Every command the saga runs then records the trigger event as its cause, in the correlation it started. `Saga.command` wraps a command in an `Activity`: * On success it saves the command’s aggregate, version and store position. * A failure that matches `error`, such as a domain error the saga reacts to, is saved too, so a replay takes the same branch. * A lost race or a store error is retried a few times. * Any other failure ends the workflow as a defect. When the workflow fails, `withCompensation` runs the compensations of the steps that had succeeded, last first: ```ts yield* Saga.command("reserve", { execute: stock.commands.reserve(sku, { orderId, quantity }) }).pipe( Order.withCompensation(() => Saga.command("release", { execute: stock.commands.release(sku, { orderId }) }).pipe(Effect.asVoid) ) ) yield* Saga.command("charge", { error: CardDeclined, execute: payments.commands.charge(orderId, { amount }) }) ``` Compensations are only registered on steps at the top of the workflow body. A refusal that is part of the normal flow, like the refund above, reads more plainly as a `catchTag`. ## Delivery and durability Caution An activity that is cut off before its outcome is saved runs again, and a trigger can start a saga again after a crash. Make the commands a saga runs idempotent: give each a [`commandId`](/guides/commands/#retries-and-command-ids) built from the execution id and the step, or record the transfer ids an account has already credited. A second start joins the first run only while the engine remembers it. Pair the engine with checkpoints that live as long: * `WorkflowEngine.layerMemory` with `Checkpoints.layerMemory`, for tests and the browser. After a restart the trigger reads the log from the start and runs every saga again. * `ClusterWorkflowEngine.layer` from `effect/cluster`, which keeps workflows in SQL through the cluster’s message storage, with durable checkpoints such as `SqliteEventStore.layerCheckpoints` or `PostgresEventStore.layerCheckpoints`. Inside a workflow, `effect/workflow` also offers `DurableClock.sleep` for timeouts, and `DurableDeferred` for waiting on a signal, such as a later event that a trigger turns into `DurableDeferred.succeed`. `DurableClock.sleep` keeps a minute or less in memory, and schedules a clock with the engine beyond that, so it survives restarts on a durable engine. The [tutorial’s saga](/tutorial/shop/#5-a-saga) races the two to give a customer 15 seconds to pay. # Storage > Keep events in memory, in SQLite or in PostgreSQL, or write your own backend. The event store is an append-only log. Every event belongs to a stream, one aggregate of one type, and has two coordinates: a `version` that orders it in its stream, and a `position` that orders it across the whole store. Writers use the version for optimistic concurrency, and readers use the position as a cursor. | Store | For | | ------------------------ | ------------------------------------------------- | | `EventStore.layerMemory` | Tests, examples, the browser | | `SqliteEventStore` | One machine, on a local file | | `PostgresEventStore` | A server, possibly with several processes writing | ## In memory `EventStore.layerMemory` keeps events in process memory. They don’t survive a restart. ## SQLite Provide any `effect/sql` SQLite client: ```ts import { SqliteClient } from "@effect/sql-sqlite-node" import { SqliteEventStore } from "effect-cqrs" const Storage = Layer.merge(SqliteEventStore.layer(), SqliteEventStore.layerCheckpoints()).pipe( Layer.provide(SqliteClient.layer({ filename: "./data/events.sqlite" })) ) ``` Positions rely on SQLite having a single writer, so the schema doesn’t carry over to other databases. An append is one statement, which checks the stream’s version and inserts the events only if it still matches. On a file, each commit waits for the disk, so once writers overlap, appends that become ready together are written in one transaction and share that wait. An append whose stream moved on still fails alone. A database in memory writes each append on its own. Several processes can share one file. A store sees its own appends at once, and with `pollInterval` it also reads the head that often to notice the others’: ```ts SqliteEventStore.layer({ pollInterval: "200 millis" }) ``` ## PostgreSQL Provide an `effect/sql` PostgreSQL client: ```ts import { PgClient } from "@effect/sql-pg" import { PostgresEventStore } from "effect-cqrs" const Storage = Layer.merge(PostgresEventStore.layer(), PostgresEventStore.layerCheckpoints()).pipe( Layer.provide(PgClient.layer({ url: Redacted.make("postgres://localhost:5432/cqrs") })) ) ``` A sequence can’t number events: transactions commit in a different order from the one they drew their numbers in, so a view model that has read up to 6 would never come back for a 5 that commits later. Positions come from a counter row instead, which an append holds until it commits. Writers wait for each other at that one step, and positions follow commit order with no gaps. So that writers don’t wait in line, a store writes the appends of its process in groups. Appends that arrive while one is being written go together into the next statement, which checks each stream’s version, numbers the events of the appends that pass and inserts them, in one round trip and one commit. An append whose stream moved on fails alone. When a writer in another process takes a stream first, the group is written again one append at a time. Every append also sends a notification with the new head, on a channel named after the events table. On a `PgClient` from `@effect/sql-pg` the store listens to it, on a connection of its own, so view models and subscribers in every process wake up on every append. * Behind a pooler that doesn’t keep sessions, such as PgBouncer in transaction mode, `LISTEN` doesn’t work. Set `listen: false` and a `pollInterval`. A client other than `PgClient` needs a `pollInterval` too. * A `pollInterval` next to listening is a safety net. The store already reads the head again whenever it reconnects, so a notification sent while it was away isn’t lost. * An append inside a transaction of your own, with `sql.withTransaction`, is written there and then, so it commits or rolls back with your other writes. It holds the counter row until that transaction ends, and every other writer of the store waits for it meanwhile, in this process and in others. Keep such transactions short, and do the slow work before them. * After a bulk load, run `ANALYZE`. Until autovacuum gets to a new table, reads for view models are slow. To run the tests on PostgreSQL, start the database of `docker-compose.yml` and say where it is. `pnpm test` then covers it too. ```bash docker compose up -d --wait POSTGRES_URL=postgres://postgres:postgres@localhost:5432/cqrs pnpm test ``` ## Migrations By default, `layer` and `layerCheckpoints` create their tables on startup, and bring tables from an earlier version of effect-cqrs up to date: existing events get schema version 1, no causation and `{}` as metadata. The statements can run again without harm, and on PostgreSQL an advisory lock keeps processes that start together apart. Where the application may not change the schema, set `migrate: false` and apply the migrations with your own tooling. `migrations(options)` lists them in order, as SQL with the table names filled in: ```ts for (const { id, name, statements } of PostgresEventStore.migrations({ eventsTable: "orders_events" })) { writeFileSync(`migrations/${id}_${name}.sql`, statements.map((statement) => `${statement};\n`).join("")) } ``` | Migration | What it does | | ---------------------- | ----------------------------------------------------------------- | | 1 `create_events` | The events table and its index, and on PostgreSQL the counter row | | 2 `add_causation` | `schema_version`, `correlation_id` and `causation_id` | | 3 `add_metadata` | `metadata` | | 4 `create_checkpoints` | The checkpoints table | A database that effect-cqrs 0.2 created already has all four. One from 0.1 lacks 2 and 3. On PostgreSQL every statement checks first, but SQLite can’t add a column only if it is missing, so there its `ALTER TABLE` statements must run exactly once. ## Options | Option | Stores | Default | What it does | | --------------------------------- | ---------- | --------------------------------- | ----------------------------------------------------------- | | `eventsTable`, `checkpointsTable` | Both | `cqrs_events`, `cqrs_checkpoints` | Table names, so several stores can share one database | | `migrate` | Both | `true` | Creates and updates the tables on startup | | `pollInterval` | Both | Off | Reads the head this often, for appends from other processes | | `listen` | PostgreSQL | `true` | Listens for appends from other processes, on a `PgClient` | ```ts SqliteEventStore.layer({ eventsTable: "orders_events" }) SqliteEventStore.layerCheckpoints({ checkpointsTable: "orders_checkpoints" }) ``` ## Your own backend Another database needs an `EventStore.Backend` with three operations, and `EventStore.make(backend, options)` builds the store on top: | Operation | What it does | | ----------------------------------------- | ------------------------------------------------------------------------------------------------- | | `append(input)` | Appends atomically, failing with `ConcurrencyConflict` if the stream moved past `expectedVersion` | | `readBatch(filter, afterPosition, limit)` | Up to `limit` events past `afterPosition` that match the filter, by position | | `head` | The position of the latest event, `0` when the store is empty | | `changes`, optional | The head positions other processes write, as the database announces them | A backend stores each event’s `schemaVersion`, `correlationId`, `causationId` and `metadata` along with it. `EventStore.toStored` fills in the defaults: version 1, no causation and `{}`. `make` runs `changes` and the `pollInterval` of its options in fibers that need a `Scope`, which `Layer.effect` provides. If `changes` fails, the store logs it and runs it again a second later, so it should start with the current head. Positions must follow commit order, so a backend on a database with concurrent writers has to guarantee that itself. `EventStore.groupCommit(appendGroup)` gives a backend group commit: it calls `appendGroup` with appends to different streams, to write in one transaction, and expects an exit for each. The store tests in `test/EventStore.test.ts` run against every backend and show what each must do. # Testing > Test an aggregate with givenEvents, when and expectEvent, like CQRSFixture. `Fixture` tests an aggregate in isolation: given these events, when this command runs, what happens? It runs commands through the real service on an in-memory store, so payload validation and event encoding are tested too. It reads like the `CQRSFixture` of moleculer-cqrs. The chain is an `Effect`, which runs when you yield it or return it from `it.effect`: ```ts import { assert, it } from "@effect/vitest" import { Fixture } from "effect-cqrs" it.effect("creates a todo", () => Fixture.make(Todo) .givenEvents() .when("createTodo", { title: "Buy Milk" }) .expectEvent(TodoEvent.cases.TodoCreated.make({ title: "Buy Milk", createdAt: DateTime.makeUnsafe(0) })) .inspectState((state) => assert.strictEqual(state.title, "Buy Milk"))) it.effect("runs commands one after another", () => Fixture.make(Todo) .givenEvents([created]) .when("deleteTodo", {}) .expectEvent("TodoDeleted") .when("createTodo", { title: "Buy Eggs" }) .expectEvents(["TodoCreated"])) it.effect("rejects deleting twice", () => Effect.gen(function*() { const error = yield* Fixture.make(Todo) .givenEvents([created, deleted]) .whenFails("deleteTodo", {}) assert.strictEqual(error._tag, "TodoAlreadyDeleted") })) ``` ## The chain | Step | What it does | | ------------------------------------ | ------------------------------------------------------------------------------------------------ | | `givenEvents(events?, aggregateId?)` | The history the commands run against, empty by default. It can include [old events](#old-events) | | `when(command, payload)` | Runs a command, typed against the aggregate. A command that fails stops the chain with its error | | `expectEvent(event)` | The last command emitted exactly this event | | `expectEvents([...])` | The last command emitted exactly these events, in order. `[]` means it recorded nothing | | `inspectState(check)` | Passes the state to a callback that throws to fail the test, so any assertion library works | | `whenFails(command, payload)` | Runs a command that should fail, and returns its error. It dies if the command succeeds | An expected event is either a whole event, compared with the aggregate’s schema, or only its tag. A mismatch dies with `ExpectationFailed`, which carries what was expected and what happened, and Vitest shows the difference. Yielding the chain gives `{ events, state, version, fixture }` for the last command, if you’d rather write plain assertions. Caution The chain is lazy. Built and never yielded or returned, it checks nothing and the test passes. ## Old events `givenEvents` also takes events in an older version of their schema, built with `Aggregate.historical`. They are stored as code from that time would have stored them, so the history goes through the [upcasting](/guides/aggregates/#evolving-events) a real store does: ```ts it.effect("upcasts todos created before due dates", () => Fixture.make(Todo) .givenEvents([Aggregate.historical(Todo, "TodoCreated", 1, { title: "Buy Milk", createdAt })]) .inspectState((state) => assert.strictEqual(state.dueDate, null))) ``` The value is checked against that version’s schema, at compile time and when it runs, so a test can’t seed a shape that never existed. For JSON taken from a database, `new Aggregate.Historical({ type, schemaVersion, payload })` skips the check. ## Invariants `checkInvariants` runs random sequences of commands, with payloads generated from their schemas by Effect’s `Arbitrary`, and checks a predicate on the state after every command that succeeds. Commands that fail are part of the game. A violation dies with `ExpectationFailed`, holding the shortest sequence it found that breaks the predicate. ```ts it.effect("never overdraws", () => Fixture.make(Account) .givenEvents([opened]) .checkInvariants({ invariant: (state) => state.balance >= 0, commands: { closeAccount: false, // left out deposit: Arbitrary.schema(SmallAmount).pipe(Arbitrary.map((amount) => ({ amount }))) }, maxCommands: 20, // the longest sequence, 20 by default check: { runs: 200, seed: 1 } // passed to Arbitrary.checkEffect })) ``` Each run starts from the given events on a fresh store. ## Time `it.effect` runs on a test clock stopped at zero, so timestamps need no mocking: an event created with `DateTime.now` has `DateTime.makeUnsafe(0)` until `TestClock.adjust` moves time. # View models > Project events into read models that catch up after downtime and can be rebuilt. A view model folds events into whatever shape a query needs: a list, a total, a table. Each one reads the event store from its own checkpoint, so it catches up after downtime and can’t miss an event. It runs after the command has stored its events, not in the same transaction, so it is a moment behind the write side: see [reading your own writes](#reading-your-own-writes). ```ts import { Checkpoints, ViewModel } from "effect-cqrs" const TodoList = ViewModel.make({ name: "todo-list", aggregates: [Todo], dispose: clearList, // runs before a replay handlers: { TodoCreated: ({ title }, { aggregateId }) => addToList(aggregateId, title), TodoDeleted: (_, { aggregateId }) => removeFromList(aggregateId) } }) const program = Effect.gen(function*() { const todos = yield* AggregateService.make(Todo) const viewModels = yield* ViewModel.ViewModels const { position } = yield* todos.commands.createTodo("todo-1", { title: "Buy Milk" }) yield* viewModels.caughtUp(position) // read your own write yield* viewModels.replay({ viewModels: ["todo-list"] }) yield* viewModels.status }).pipe( Effect.provide(ViewModel.layer([TodoList]).pipe( Layer.provideMerge(Layer.merge(EventStore.layerMemory, Checkpoints.layerMemory)) )) ) ``` ## Defining one `handlers` has one function per event tag the view model cares about, and only accepts tags of its `aggregates`. Each gets the decoded event and its `EventMeta`, the same as a [projection](/guides/aggregates/#the-projection), with the command’s `metadata`. `dispose` clears everything the view model stored, before a replay rebuilds it. `ViewModel.layer([...])` starts the view models in the background for as long as the layer lives, and provides the `ViewModels` service. It needs an `EventStore`, `Checkpoints` and whatever services the handlers use. It also uses `ParkedEvents` and `Leases` when they are provided, see [failing handlers](#failing-handlers) and [several processes](#several-processes). ## At least once A view model pulls events from the store, rather than waiting for them to be pushed, so an event can’t be lost while it is down. It saves its checkpoint every 500 events and when it stops, so after a crash it can see up to two batches of events again. ## Checkpoints in a transaction A view model that keeps its data in the same database as its checkpoints can commit both together. With `transaction`, each batch of handlers runs in one transaction with the checkpoint it reaches, so the checkpoint is saved after every batch. A batch that fails leaves nothing behind, and a crash repeats nothing, so handlers needn’t be idempotent. ```ts const TodoRows = ViewModel.make({ name: "todo-rows", aggregates: [Todo], dispose: SqlClient.SqlClient.use((sql) => sql`DELETE FROM todo_rows`), transaction: ViewModel.sqlTransaction, handlers: { TodoCreated: ({ title }, { aggregateId }) => SqlClient.SqlClient.use((sql) => sql`INSERT INTO todo_rows (id, title) VALUES (${aggregateId}, ${title})`) } }) ``` `ViewModel.sqlTransaction` uses the `SqlClient` in context. The handlers and the checkpoints, `SqliteEventStore.layerCheckpoints()` or `PostgresEventStore.layerCheckpoints()`, have to get that same client: a transaction belongs to one client, and another client’s writes don’t join it. Any other `(effect) => Effect` works as `transaction` too, for a database of its own. Only the database is rolled back. A handler that runs a command or calls a service still does that once per attempt. ## Reading your own writes A command returns the position of its last event. `viewModels.caughtUp(position)` waits until every view model has processed it, so a client can read what it just wrote. Without a position it waits for the latest event in the store. Over HTTP, [`CommandApi.ReadAfter`](/guides/integration/#reading-your-own-writes) does the same for a request that sends the position back in a `Read-After` header. There are three ways to read, from the most current to the most flexible: | Read | Up to date | For | | ---------------------------------------------------------------------------- | ---------------------------------- | -------------------------- | | [`state(id)`](/guides/commands/#reading), which folds one aggregate’s events | Always | One aggregate, by id | | A view model after `caughtUp(position)` | Includes the caller’s own commands | The screen after a command | | A view model | A moment behind | Lists, searches, totals | effect-cqrs has no query handlers of its own. A view model writes to whatever storage you choose, and a query is your own code that reads it: a `Ref`, a SQL `SELECT`, a search index. ## Failing handlers A handler that fails stalls its view model on that event, which then retries: after 100 ms, then twice as long each time it fails at the same place, up to 30 seconds. Once it moves on, the next failure starts again from 100 ms. Events are never skipped or reordered. `viewModels.status` lists each view model’s position and the failure it is retrying, if any. A stalled view model doesn’t catch up, so put a timeout on `caughtUp` where that matters. ### Parking events An event that can never be handled, a bug or data the handler doesn’t expect, stalls its view model for good. `onFailure: ViewModel.park({ attempts })` gives up on an event after it has failed that many times: the view model records it in `ParkedEvents`, skips it from then on and carries on with the next one. ```ts const TodoList = ViewModel.make({ name: "todo-list", aggregates: [Todo], dispose: clearList, onFailure: ViewModel.park({ attempts: 5 }), handlers: { /* ... */ } }) const viewModels = yield* ViewModel.ViewModels yield* viewModels.parked("todo-list") // view model, position, event id, error, attempts, time yield* viewModels.retryParked({ viewModel: "todo-list" }) // after fixing the handler ``` `retryParked` handles the parked events again, in order of position. Those that succeed leave `ParkedEvents`, the others stay with the new failure, and `positions` narrows it to some of them. A retried event comes after the events that followed it, so park only where order doesn’t matter between them, and a replay keeps skipping parked events until they are retried. Stalling stays the default. A view model that skips an event is wrong from then on, in a way nothing shows, while a stalled one is visibly behind and right once it moves: a balance that missed a deposit is worse than one that is late. Parking is a choice per view model, for those where a gap is better than a stop. Parked events are kept in memory unless `ParkedEvents` is provided. `ParkedEvents.layerSql()` keeps them in a table, `cqrs_parked_events` by default, on SQLite or PostgreSQL. It creates the table on startup unless `migrate` is `false`, and `ParkedEvents.migrations()` gives its migrations to run another way. `viewModels.status` counts each view model’s parked events, and the counter `ViewModel.MetricNames.parkedEvents`, with a `view_model` attribute, counts every event parked. ## Replay `viewModels.replay({ viewModels: ["todo-list"] })` clears the checkpoint, runs `dispose` and rebuilds the view model from the log. `startTime` and `finishTime` limit which events go in: the view model then resumes from the present, and events outside the window stay out of it. Moleculer-cqrs needed replay to recover events lost on the broker. Here view models can’t lose events, so replay is for rebuilding a view model that changed, or one that got damaged. ## Personal data Handlers receive [personal fields](/guides/personal-data/) decrypted. A view model that stores them handles `onForget(subject, meta)` to erase them when a subject is forgotten, see [view models and personal data](/guides/personal-data/#view-models). ## Checkpoints A checkpoint has to live as long as its view model: * A view model kept in memory pairs with `Checkpoints.layerMemory`, and rebuilds itself on every start. * A view model stored in a database pairs with that database’s checkpoints, `SqliteEventStore.layerCheckpoints()` or `PostgresEventStore.layerCheckpoints()`. ## Several processes Replicas that share a database would each run every view model, and race on its checkpoint. `ViewModel.layer` takes a lease on each view model before running it, from the `Leases` in context. Without one, every lease is held, which suits a single process. `Leases.layerPostgres()` holds leases as PostgreSQL advisory locks, on a hash of the view model’s name, so only one process runs each view model. The others are standbys: they try to take over every 5 seconds, `interval`, and the one that does resumes from the checkpoint the holder saved. The locks live on one connection reserved from the client’s pool, so they go when the process dies or its connection does. ```ts const Live = ViewModel.layer([TodoList]).pipe( Layer.provideMerge(Layer.mergeAll( PostgresEventStore.layer(), PostgresEventStore.layerCheckpoints(), Leases.layerPostgres() )), Layer.provide(PgClient.layer({ url: Redacted.make("postgres://localhost:5432/cqrs") })) ) ``` In a standby: * `viewModels.status` shows `holdsLease: false` for the view models another process runs. * `caughtUp` doesn’t wait for those view models, since nothing in this process moves them. It returns once the view models this process runs have caught up, so read your own writes where the view model runs, or wait on the read model itself. * `replay` and `retryParked` fail with a `ViewModelError`: run them where the view model runs. A holder notices that its connection is gone at its next check, up to `interval` later, and a standby may already have started by then. For that long, both can handle the same events, even in a `transaction`: a lease doesn’t fence off the old holder’s writes. In an Effect cluster, `Singleton` from `effect/cluster` does the same job with the cluster’s own shard ownership, and needs no lease: ```ts import { Singleton } from "effect/cluster" const TodoListSingleton = Singleton.make("todo-list", Layer.launch(ViewModel.layer([TodoList]))) ``` The view models then run on whichever runner owns the singleton, and `ViewModels` is only available inside it. ## Keeping up View models within 500 events of the end of the log share one read of it, and decode each event once. Commands give other fibers a turn every millisecond, so a burst of commands doesn’t starve the view models. A slow handler still falls behind without slowing the writers down, and catches up when they stop. ## Other subscribers Anything else that reacts to events can use `store.subscribe(filter)`: a stream of past events, followed by live ones. A process that runs commands in reaction to events is better written as a [saga](/guides/sagas/), whose trigger is a view model. A handler runs with the event it handles as the [cause](/guides/commands/#causation) of the commands it runs. A subscriber gets the same with `Causation.causedBy(stored)`. # Blog > Posts about CQRS and event sourcing, with examples that run on effect-cqrs. ## CQRS with effect-cqrs A series in five parts, November 2026. We build a small lending library, where members borrow books and bring them back, from the first event to a saga, tests and a database. Every snippet comes from [`examples/blog`](https://github.com/davidnussio/effect-cqrs/tree/main/examples/blog), whose tests run it. [1. Why CQRS and event sourcing](/blog/why-cqrs-and-event-sourcing/)What goes missing when you store only the current state, event sourcing by hand in forty lines, and what CQRS adds on top. [2. Your first aggregate](/blog/your-first-aggregate/)A book as an aggregate: events as schemas, commands that decide, typed errors, and a store that handles concurrent writers. [3. Read models you can throw away](/blog/read-models/)View models that follow the log, catch up after a restart, read your own writes, and rebuild from scratch. [4. Sagas](/blog/sagas/)Borrowing a book touches the member and the book, so a saga coordinates them as a durable workflow. [5. Tests, schema changes and production](/blog/tests-schema-changes-and-production/)Given, when, then and random commands, an event that changes shape, SQLite and PostgreSQL, and forgetting a member. ## When things go wrong Bugs and incidents every team meets, on the same library: what they do to a database that keeps only the current state, and what the log changes. Each post stands alone, and its example runs in a test of its own. [Fix the code, not the data](/blog/fix-the-code-not-the-data/)A loans table with a bug that loses loans, fixed in a database, where the lost rows stay lost, and in a read model, where a replay brings them back. [The wrong event](/blog/the-wrong-event/)A bug in a command stores wrong due dates for a week. Why not to edit the log, hide it in an upcaster or patch the read models, and how a correction event fixes it. [The double write](/blog/the-double-write/)Store the loan and tell the broker, in either order, and see what each breaks on a bad day. With event sourcing the log is already the outbox. [I clicked twice](/blog/i-clicked-twice/)A payment request times out and is sent again. Why disabling the button and a unique column fall short, and how a command id makes a retry return the first answer. [The form that overwrites](/blog/the-form-that-overwrites/)Two librarians edit the same record and the second save undoes the first. Commands that say what changed avoid most collisions, and an expected version catches the rest. [The message that blocks the queue](/blog/the-message-that-blocks-the-queue/)One event a handler can't read stops everything behind it. Stop and wait, or park it and carry on: when each is right, and how a parked event comes back after the fix. [What it looked like last Tuesday](/blog/what-it-looked-like-last-tuesday/)A member was refused a book that is on the shelf today. The log shows the book as it was at that moment, and every event the request caused, across aggregates. [The bug that won't reproduce](/blog/the-bug-that-wont-reproduce/)A member can't borrow anything and every test passes. Copy her stream from production into a fixture, and the bug report becomes a failing test. [Two pods, two emails](/blog/two-pods-two-emails/)Scaled to two replicas, every member gets every email twice. Background work that reads the log must run in one place, and move when that place goes away. ## Design choices Decisions that shape an event-sourced system before anything breaks: where a process lives, how to read, what to keep, and how to get there from the tables you already have. [The saga you skipped](/blog/the-saga-you-skipped/)Borrowing a book without a saga, in the two ways that look right, what breaks when a write fails between them, and what the saga does instead. [Three ways to read](/blog/three-ways-to-read/)A lent book read from its aggregate, from a read model as it is, and from a read model that waits for the command, in one process and over HTTP. [Forget me, from the backups too](/blog/forget-me-from-the-backups-too/)A member asks to be erased, and her name is still in every backup, ready to come back with the next restore. Crypto-shredding erases her from the backups you already have. [From a table to a log](/blog/from-a-table-to-a-log/)Every project that adopts event sourcing already has data in tables. Import each row as the first event of its stream, invent no history, and switch writes over one aggregate at a time. [Change the rules, keep the history](/blog/change-the-rules-keep-the-history/)The loan period changes and every open loan's due date moves with it. Store the decisions rules made, run rules only when deciding, and never in a projection. # Change the rules, keep the history > The loan period goes from four weeks to three, and every open loan's due date moves with it. Rules change all the time: store the decisions they made, run them only when deciding, and never in a projection. *A post after the series [CQRS with effect-cqrs](/blog/), on its lending library. November 2026.*  Every summer the library relaxes its rules: five books instead of three, four weeks instead of three. On the first of September the old rules come back. In the CRUD version of the library, the loan period is a setting, and due dates are worked out when they are shown: ```sql SELECT title, lent_on + (SELECT loan_days FROM settings) * interval '1 day' AS due_on FROM loans WHERE returned = false; ``` On September 1st someone changes `loan_days` from 28 to 21. Every loan made in August is now a week shorter, and the reminder job emails a few hundred members that their books are overdue. They were told four weeks, and they were right. Business rules change: prices, limits, periods, fees, eligibility. This post shows where they belong in an event-sourced system so that a change applies to new decisions and leaves the old ones alone. Every snippet runs in [`examples/blog/change-the-rules.test.ts`](https://github.com/davidnussio/effect-cqrs/blob/main/examples/blog/change-the-rules.test.ts), which moves the clock from August to September. ## Rules with a date The library’s rules are a list, each with the day it starts: examples/blog/change-the-rules.ts ```ts export interface Policy { readonly name: string /** From when it applies, inclusive. */ readonly from: DateTime.Utc readonly limit: number readonly days: number } /** The library's lending rules over time, newest last. */ export const policies: ReadonlyArray = [ { name: "standard-2025", from: DateTime.makeUnsafe("2025-01-01T00:00:00Z"), limit: 3, days: 21 }, { name: "summer-2026", from: DateTime.makeUnsafe("2026-06-01T00:00:00Z"), limit: 5, days: 28 }, { name: "standard-2026", from: DateTime.makeUnsafe("2026-09-01T00:00:00Z"), limit: 3, days: 21 } ] /** The policy in force at a given time. */ const policyAt = (at: DateTime.Utc): Policy => policies.findLast(({ from }) => DateTime.isLessThanOrEqualTo(from, at)) ?? policies[0]! ``` A rule change is a new entry, deployed before its date. The old entries stay, because they say what the rules *were*, and that will be asked. ## Decide with the rules, record the decision The member aggregate runs the rules in one place, the command handler, on the day it decides: examples/blog/change-the-rules.ts ```ts export const MemberEvent = Schema.TaggedUnion({ // The decision, with the rule that made it: the due date the member was given, and why. BookBorrowed: { bookId: Schema.String, dueOn: Schema.DateTimeUtc, policy: Schema.String }, BookReturned: { bookId: Schema.String } }) export interface MemberState { /** The books the member has, with the due date each was given. */ readonly books: Readonly> } export const Member = Aggregate.make()({ name: "member", events: MemberEvent, initial: { books: {} }, // The projection records what was decided. It checks no rule, so a member who kept five // books under the summer rules still loads in September. projection: { BookBorrowed: ({ books }, { bookId, dueOn }) => ({ books: { ...books, [bookId]: dueOn } }), BookReturned: ({ books }, { bookId }) => { const { [bookId]: _, ...rest } = books return { books: rest } } } }) .command("borrow", { payload: Schema.Struct({ bookId: Schema.String }), // The rules live here, and run on the day of the decision. handler: Effect.fnUntraced(function*({ books }, { bookId }, { aggregateId }) { const now = yield* DateTime.now const policy = policyAt(now) if (Object.keys(books).length >= policy.limit) { return yield* new LoanLimitReached({ memberId: aggregateId, limit: policy.limit }) } const dueOn = DateTime.add(now, { days: policy.days }) return MemberEvent.cases.BookBorrowed.make({ bookId, dueOn, policy: policy.name }) }) }) .command("giveBack", { payload: Schema.Struct({ bookId: Schema.String }), handler: ({ books }, { bookId }) => Effect.succeed(books[bookId] === undefined ? [] : MemberEvent.cases.BookReturned.make({ bookId })) }) ``` Two choices in it carry the whole post: * **`BookBorrowed` stores the due date**, not the date it was lent plus a period to add later. The due date is what the member was told. It is a decision, made once, under the rules of that day, and it is a fact from then on. * **It also stores which policy decided it.** A member who asks why her book is due on the 17th gets an answer: the summer rules. So does an auditor. ## August examples/blog/change-the-rules.ts ```ts for (const bookId of ["dune", "emma", "ulysses", "solaris"]) { yield* members.commands.borrow("alice", { bookId }) } // Four BookBorrowed, each due in 28 days, under "summer-2026" ``` Alice borrows four books under the summer rules, each due in four weeks. ## September The standard rules are back. Alice’s state, and a few commands: examples/blog/change-the-rules.ts ```ts const alice = yield* members.state("alice") // Four books, each with the due date she was given in August, four weeks after she took it const refused = yield* Effect.flip(members.commands.borrow("alice", { bookId: "emma-2" })) // refused: LoanLimitReached { limit: 3 } yield* members.commands.giveBack("alice", { bookId: "dune" }) yield* members.commands.giveBack("alice", { bookId: "emma" }) const borrowed = yield* members.commands.borrow("alice", { bookId: "neuromancer" }) // borrowed.events: [BookBorrowed { bookId: "neuromancer", dueOn: in 21 days, policy: "standard-2026" }] ``` * **Her four books load**, with the due dates from August. Nothing moved, because nothing was worked out again. * **Her fifth request is refused** under the new limit of three, although she has four. The rule runs on the decision being made now, with today’s rules. * **After she brings two back, she borrows again**, and the new loan is due in three weeks, under `standard-2026`. ## Never check rules in a projection Look at the projection again: it checks nothing. A member with four books when the limit is three is a state it folds without complaint. That is on purpose. A projection runs on every load, over every event ever recorded, with today’s code. If it checked today’s limit, Alice’s stream would fail to load the moment the rules changed, and she could neither borrow nor return a book. Worse, the projection would be judging decisions made under rules it doesn’t know about. Events are what happened. A projection that disagrees with them is wrong, not the events. So rules go in command handlers, where they decide about the future, and projections only record what was decided. ## Facts and conclusions [Fix the code, not the data](/blog/fix-the-code-not-the-data/) recommended working out conclusions in read models, so that a replay can fix them. This post stores the due date in the event. Both hold, because they are different things: * **A decision someone was told is a fact.** The due date on Alice’s slip, the price on an invoice, the fee in a confirmation email. Store it in the event, with what decided it. * **A judgment you may revise is a conclusion.** Whether a member is a frequent borrower, which books are popular, whether a return was late by a newer definition. Work it out in a read model, from the facts. If you are unsure which one you have, ask whether a customer would be upset if it changed after the fact. If yes, it is a fact. ## Rules of thumb * **Rules run in command handlers, on the day of the decision.** Never in projections. * **Store the outcome, not the inputs and a formula.** `dueOn`, not `lentOn` plus a setting. * **Record which rule decided.** A name or a version in the event answers “why” for years. * **Keep old rules around with their dates.** They explain the history, and a rule that takes effect later is deployed before its date like any other entry. * **Load whatever the log says.** A state that today’s rules wouldn’t allow is a state that older rules did. ## Next [The wrong event](/blog/the-wrong-event/) covers the other kind of change: a decision that was wrong when it was made, and how to correct it. # Fix the code, not the data > A loans table with a bug that loses loans, fixed twice: in a database, where the fix stops the damage and the lost rows stay lost, and in a read model, where a replay brings them back. *A post after the series [CQRS with effect-cqrs](/blog/), on its lending library. November 2026.*  Every team has fixed this bug. Some code writes the wrong thing to a table, for weeks, until someone notices. The fix is a few lines and the tests pass. Then someone asks about the rows the bug already wrote, and the few lines turn into a migration, a script to run on production, a dig through the backups, or an apology. With event sourcing the same bug ends with the code fix. The read model is rebuilt from the log by the fixed code, and nothing in the database has to be edited by hand. This post shows both, on the library of the series. Every snippet runs in [`examples/blog/fix-the-code.test.ts`](https://github.com/davidnussio/effect-cqrs/blob/main/examples/blog/fix-the-code.test.ts). ## The bug, in a table The library wants a history of its loans, so it gets a `loans` table. Someone writes it with one row per book, the way you’d store where a book is now: ```sql INSERT INTO loans (book_id, loan_id, member_id, returned) VALUES ('book-1', 'loan-1', 'alice', false) ON CONFLICT (book_id) DO UPDATE SET loan_id = excluded.loan_id, member_id = excluded.member_id, returned = false; ``` Alice borrows *Dune* and brings it back. Bob borrows it. Bob’s loan lands on Alice’s row, and Alice’s loan is gone. A month later a member of staff asks why Alice has never borrowed anything, and the bug is found. The fix is to key the table by loan. From the next loan on, every row is right. Alice’s loan is not coming back, though: the `UPDATE` wrote over the only copy of it, and nothing else in the database remembers it. If you’re lucky, it’s in last night’s backup, or in an application log, and someone pieces it back together by hand. A gentler bug, one that works out a value wrong without destroying anything, still costs twice. The code fix covers new rows, and a migration has to repair the old ones: the same logic written again in SQL, run once, on production, with no test that the two agree. ## The same bug, in a read model Now the same history as a read model, kept by a view model like the catalog of [part 3](/blog/read-models/). It lives in memory here, and it would be the same with a table: examples/blog/fix-the-code.ts ```ts export interface LoanRecord { readonly loanId: string readonly bookId: string readonly memberId: string readonly returned: boolean } /** Every loan the library has made: the `loans` table, as a read model. */ export class LoanHistory extends Context.Service>>()( "library/LoanHistory" ) { static readonly layer = Layer.effect(LoanHistory, Ref.make>(new Map())) } const save = (key: string, record: LoanRecord) => LoanHistory.use((history) => Ref.update(history, (loans) => new Map(loans).set(key, record))) const markReturned = (key: string) => LoanHistory.use((history) => Ref.update(history, (loans) => { const record = loans.get(key) return record === undefined ? loans : new Map(loans).set(key, { ...record, returned: true }) }) ) /** The loans of one member, as the front desk asks for them. */ export const loansOf = (memberId: string) => LoanHistory.use((history) => Ref.get(history).pipe(Effect.map((loans) => [...loans.values()].filter((loan) => loan.memberId === memberId))) ) ``` And the same bug, with the record kept under the book’s id: examples/blog/fix-the-code.ts ```ts export const BuggyLoanHistoryView = ViewModel.make({ name: "loan-history", aggregates: [Book], dispose: LoanHistory.use((history) => Ref.set(history, new Map())), handlers: { // The bug: the record is kept under the book's id, so the next loan of the same // book replaces it. BookLent: ({ loanId, memberId }, { aggregateId }) => save(aggregateId, { loanId, bookId: aggregateId, memberId, returned: false }), BookReturned: (_, { aggregateId }) => markReturned(aggregateId) } }) ``` The day goes the same way: examples/blog/fix-the-code.ts ```ts yield* books.commands.add("book-1", { title: "Dune", author: "Frank Herbert" }) yield* books.commands.lend("book-1", { loanId: "loan-1", memberId: "alice" }) yield* books.commands.returnBook("book-1", {}) const { position } = yield* books.commands.lend("book-1", { loanId: "loan-2", memberId: "bob" }) yield* viewModels.caughtUp(position) const alice = yield* loansOf("alice") // []: Bob's loan took the place of hers ``` Alice’s loan is gone from the read model, exactly as it was gone from the table. The difference is in what the bug could reach. The view model only writes its own read model. The log, where `BookLent` for `loan-1` was appended by the book’s command, is written by the commands alone, and the bug never touched it. ## Fix the code The fix is the same few lines: examples/blog/fix-the-code.ts ```ts export const LoanHistoryView = ViewModel.make({ name: "loan-history", aggregates: [Book], dispose: LoanHistory.use((history) => Ref.set(history, new Map())), handlers: { // The fix: one record per loan. BookLent: ({ loanId, memberId }, { aggregateId }) => save(loanId, { loanId, bookId: aggregateId, memberId, returned: false }), BookReturned: ({ loanId }) => markReturned(loanId) } }) ``` To see what a deploy does, the example keeps the storage apart from the code that runs on it. The log, the checkpoints and the read model live through the deploy. The view model is replaced: examples/blog/fix-the-code.ts ```ts // What outlives a deploy: the log, the checkpoints and the loans table itself. export const Storage = Layer.mergeAll(EventStore.layerMemory, Checkpoints.layerMemory, LoanHistory.layer) // What a deploy starts: the books and one version of the view model. const deploy = (view: typeof LoanHistoryView) => ViewModel.layer([view]).pipe(Layer.provideMerge(Books.layer)) ``` Then the new version starts: examples/blog/fix-the-code.ts ```ts yield* viewModels.caughtUp() const beforeReplay = yield* loansOf("alice") // []: the view model starts from its checkpoint, after every event the bug handled const { eventCount } = yield* viewModels.replay({ viewModels: ["loan-history"] }) const afterReplay = yield* loansOf("alice") // [{ loanId: "loan-1", bookId: "book-1", memberId: "alice", returned: true }], from the 3 loan events ``` Deploying alone changes nothing for the past. The view model picks up from its checkpoint, after every event the bug handled, so new loans are right and Alice’s is still missing. `replay` is the step that repairs it: it resets the checkpoint, clears the read model with `dispose`, and runs the fixed handlers over the whole log. Alice’s loan is back, worked out from the same `BookLent` the bug read wrong. There was no migration, no script and no `UPDATE`. The log didn’t change, through the bug or through the fix. The handler that will record tomorrow’s loans is the one that rebuilt yesterday’s, so there is one piece of code to test, not a fix and a migration that have to agree. ## Why it works * **The truth and the copy are apart.** Commands write the log, and view models only read it. A bug in a view model can damage the copy it keeps, never the facts it was made from. * **The past and the future go through the same code.** A replay is the view model running as it always does, from the first event instead of from its checkpoint. * **Nothing is lost while it’s wrong.** Every event the bug mishandled is still in the log, waiting for a handler that reads it right. This is the same reason [part 1](/blog/why-cqrs-and-event-sourcing/) could count loans with a `timesLent` written after them. A wrong read model and a missing one are fixed the same way. ## What a replay can’t fix A replay rebuilds read models. It doesn’t change what the commands decided. * **A bug on the write side stays in the log.** If `lend` had stored the wrong member, the event says so for good, and replays reproduce it faithfully. That is fixed forward: a command that records the correction as a new event, which read models handle like any other, as [The wrong event](/blog/the-wrong-event/) shows. An event whose *shape* is wrong is read through an upcaster, as in [part 5](/blog/tests-schema-changes-and-production/). * **What an event concludes is fixed with it.** The library’s `BookReturned` carries `late`, worked out by `returnBook` when the book came back. If the rule for lateness had a bug, every return it judged keeps the wrong answer. A read model that worked out lateness from `dueOn` and the time of the return would be fixed with a replay. Deciding what an event states and what read models work out is the main choice in designing one. * **A replay takes time.** It reads the whole log, and the read model is cleared while it runs. For a large one, deploy the fix under a new name, say `loan-history-2`, with its own storage: it starts from the first event next to the old one, and queries move over once it has caught up. * **Handlers should only write their read model.** A handler that sends an email sends it again on every replay. Work that must happen once belongs in a [saga](/blog/sagas/), which records what it has done. ## Rules of thumb * **Keep facts in events, and conclusions in read models.** What happened can’t be wrong later. What it means can, and a read model can work it out again. * **When a read model is wrong, fix the handler and replay.** Don’t patch its data by hand: the next replay rebuilds it from the handlers and throws the patch away. * **When an event is wrong, add one.** The log records the mistake and the correction, which is what an auditor wants to see anyway. ## Next The [view models guide](/guides/view-models/) covers replays, checkpoints and failing handlers in full, and [part 3](/blog/read-models/) builds the library’s read models from the start. # Forget me, from the backups too > A member asks to be erased. Deleting her row is easy, and her name is still in every backup for months, ready to come back with the next restore. Crypto-shredding erases her from the backups you already have. *A post after the series [CQRS with effect-cqrs](/blog/), on its lending library. November 2026.*  Alice moves away and asks the library to erase her personal data, as the GDPR allows. In a CRUD application it looks like one statement: ```sql DELETE FROM members WHERE id = 'alice'; ``` Then the questions start. The nightly backups of the last 30 days still have her row. So do the weekly ones kept for a year, the copy the analytics team loaded last month, and the dump a developer took to debug something. Six weeks later a disk fails, the database is restored from a backup, and Alice is back, with her name and address, as if she had never asked. Erasure in a system with backups is a known hard problem. This post shows how crypto-shredding solves it, and why the way [part 5](/blog/tests-schema-changes-and-production/#forgetting-a-member) forgets a member was about backups all along. Every snippet runs in [`examples/blog/forget-me.test.ts`](https://github.com/davidnussio/effect-cqrs/blob/main/examples/blog/forget-me.test.ts). ## Why deleting isn’t enough The usual answers each fall short somewhere: * **Delete from the backups too.** Most backups can’t be edited: they are compressed snapshots, often on write-once storage on purpose, so ransomware can’t touch them. * **Re-apply deletions after a restore.** Keep a list of erased ids and delete them again after every restore. It works if nobody forgets, and the list of who asked to be erased is itself personal data. * **Keep backups short.** Thirty days of backups means thirty days during which a restore brings Alice back. Regulators tend to accept it as long as it is documented. Users don’t always. And with event sourcing it looks worse: the log is never edited, so there is no row to delete in the first place. ## Encrypt each person with their own key Crypto-shredding turns the problem around. Each personal field is stored encrypted, with a key that belongs to the person it describes. Erasing the person means destroying one key. Every copy of the data, in the store, in every backup, in every dump, becomes unreadable at once, without touching any of them. In the library, the member’s name is marked as personal data in its event schema, as in [part 4](/blog/sagas/): examples/blog/library.ts ```ts MemberJoined: { name: PersonalData.field(Schema.NonEmptyString, { redacted: "[forgotten]" }) }, ``` The keys live in their own store, apart from the events: examples/blog/forget-me.ts ```ts // The keys live apart from the events: another database, other permissions, other backups. export const Keys = PersonalData.layerMemory().pipe(Layer.provide(PersonalData.layerMasterKeyEphemeral)) ``` Here it is in memory with a random master key, for the test. In production it is `PersonalData.layerSql()` on another database, with the master key from the environment or a KMS. What the event store holds is not Alice’s name: examples/blog/forget-me.ts ```ts yield* members.commands.join("alice", { name: "Alice Rossi" }) yield* members.commands.join("bob", { name: "Bob Bianchi" }) const [stored] = yield* store.read({ aggregateIds: ["alice"] }).pipe(Stream.runCollect) // stored.payload: { _tag: "MemberJoined", name: { $personal: { subject: "alice", data: "X4DnRIZB…" } } } ``` The subject and the ciphertext, and nothing that reads as Alice. Projections, handlers and read models never see this: `AggregateService` decrypts on the way in and encrypts on the way out, so to the code, `name` is a string. ## Back up, then forget Last night, the backup ran: examples/blog/forget-me.ts ```ts // Last night's backup: every event, as stored. const backup = yield* store.read().pipe(Stream.runCollect) ``` This morning, Alice asked to be forgotten: examples/blog/forget-me.ts ```ts yield* PersonalData.forget("alice") const alice = yield* members.state("alice") // { joined: true, name: "[forgotten]", loans: {} } ``` Her key is destroyed, and her name reads as the replacement given in the schema. Her events are all still there, and with them everything else they record: loans, dates, late returns. They still add up, and so does everything read models built from them. ## Restore Then the bad day: the event store has to be restored from last night’s backup, from before Alice asked: examples/blog/forget-me.ts ```ts // Something goes wrong, and the events are restored from last night's backup. const restored = yield* Effect.gen(function*() { yield* restore(backup) const members = yield* AggregateService.make(Member) return { alice: yield* members.state("alice"), bob: yield* members.state("bob") } }).pipe(Effect.provide(Layer.fresh(EventStore.layerMemory))) // restored.alice.name: "[forgotten]", restored.bob.name: "Bob Bianchi" ``` Alice doesn’t come back. The backup holds her name exactly as the store did, encrypted with a key that no longer exists. Bob, who never asked, is restored as he was. Nobody had to remember a list of erased ids or edit a backup. Every backup of the events ever taken lost Alice’s name at the moment her key was destroyed. ## The keys have backups too The problem moves to the key store, and it is a much smaller problem: * **Keep the key store’s backups short.** A copy of the keys outlives a forget for as long as it is kept, so the key store’s backups should expire in days, not months. The events, which are most of the data, can be kept as long as you need. * **Keep the keys apart from the events.** Another database or at least another client, with its own permissions and its own backups. A dump of the events then holds no names, and a dump of the keys holds no events. * **Wrap the keys with a master key** from the environment or, better, a KMS, so that even the key store’s backups are useless without it. Caution Crypto-shredding covers what goes through `AggregateService`. Read models store decrypted values, so a view model that kept a name erases it with `onForget`, as part 5 shows. Application logs, traces, exports and caches that saw a decrypted value keep it: don’t log event payloads with personal fields. ## Rules of thumb * **Mark personal fields from the first version of an event.** Events stored before a field was marked keep it in plaintext, because the log is never rewritten. * **Use opaque ids as subjects.** The subject id is stored next to each encrypted field, so `alice` or a UUID, never an email address. * **Separate what you keep from who it was about.** Loans, amounts and dates are the library’s records. Names and addresses are Alice’s. Crypto-shredding lets the first outlive the second. * **Make backups of keys expire fast.** That is the one place where a forget can be undone. ## Next The [personal data guide](/guides/personal-data/) covers marking fields, key stores, master keys on a KMS, key rotation and what crypto-shredding doesn’t cover. # From a table to a log > Every real project that adopts event sourcing already has data in tables. Import each row as the first event of its stream, say where it came from, don't invent the history you never had, and switch writes over one aggregate at a time. *A post after the series [CQRS with effect-cqrs](/blog/), on its lending library. November 2026.*  The series built its library from the first event. Nobody’s real project starts there. The library already runs on a database, with a `books` table that has years of data in it, and the question after reading about event sourcing is not *how do I start* but *how do I get there from here*. This post moves the library’s books from a table to a log, without stopping it and without pretending to know more than the table did. Every snippet runs in [`examples/blog/table-to-log.test.ts`](https://github.com/davidnussio/effect-cqrs/blob/main/examples/blog/table-to-log.test.ts). ## What the table knows examples/blog/table-to-log.ts ```ts /** A row of the old `books` table, as `SELECT * FROM books` returns it. */ export interface BookRow { readonly id: string readonly title: string readonly author: string readonly lent_to: string | null readonly loan_id: string | null readonly due_on: string | null } export const rows: ReadonlyArray = [ { id: "book-1", title: "Dune", author: "Frank Herbert", lent_to: "alice", loan_id: "L-1042", due_on: "2026-12-01T00:00:00Z" }, { id: "book-2", title: "Emma", author: "Jane Austen", lent_to: null, loan_id: null, due_on: null } ] ``` The table knows each book’s current state: its title, its author, who has it and until when. It doesn’t know when the book was added, how many times it has been lent, or who had it before Alice. That history was thrown away one `UPDATE` at a time, as [part 1](/blog/why-cqrs-and-event-sourcing/) put it. No migration can bring it back. ## Don’t invent the past The tempting import writes the history the events *would* have had: a `BookAdded`, then a `BookLent` for Alice. Every read model would work at once, since they already handle those events. It is also false. The `BookAdded` would carry the date of the import, not the date the book was added. The `BookLent` would claim Alice borrowed the book at the moment of the import. Any read model that counts loans per month, or measures how long books stay out, would now count a spike of fake loans on import day. A log is valuable because it says what happened, and an import that makes things up spends that on day one. ## Import what the table says Instead, the import is a fact of its own: on this date, the old system said this about the book. It gets an event that says exactly that: examples/blog/table-to-log.ts ```ts const Loan = Schema.Struct({ loanId: Schema.String, memberId: Schema.String, dueOn: Schema.DateTimeUtc }) export const BookEvent = Schema.TaggedUnion({ // What the old system knew about the book on the day of the import, and nothing more. BookImported: { title: Schema.String, author: Schema.String, loan: Schema.NullOr(Loan) }, 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 } }) ``` `BookImported` carries the row’s state, the loan included if there is one, and nothing more. The book’s projection starts from it like from any other first event, and the `importBook` command records it only for a book that has no stream yet: examples/blog/table-to-log.ts ```ts export const Book = Aggregate.make()({ name: "book", events: BookEvent, initial: { status: "unknown", loan: null }, projection: { BookImported: (_, { loan }) => ({ status: loan === null ? "available" : "lent", loan }), BookAdded: () => ({ status: "available", loan: null }), BookLent: (_, loan) => ({ status: "lent", loan }), BookReturned: () => ({ status: "available", loan: null }) } }) .command("importBook", { payload: Schema.Struct({ title: Schema.String, author: Schema.String, loan: Schema.NullOr(Loan) }), // A book that already has a stream was imported already: running the import again is harmless. handler: (state, payload) => Effect.succeed(state.status === "unknown" ? BookEvent.cases.BookImported.make(payload) : []) }) ``` The import script reads the rows and sends one command per row, with metadata naming where the data came from: examples/blog/table-to-log.ts ```ts /** Writes one `BookImported` per row. Safe to run again, for instance after a failure halfway. */ export const importBooks = Effect.fnUntraced(function*(rows: ReadonlyArray) { const books = yield* AggregateService.make(Book) for (const row of rows) { const loan = row.lent_to === null || row.loan_id === null || row.due_on === null ? null : { loanId: row.loan_id, memberId: row.lent_to, dueOn: DateTime.makeUnsafe(row.due_on) } yield* books.commands.importBook(row.id, { title: row.title, author: row.author, loan }) } }, Metadata.withMetadata({ importedFrom: "books table", script: "scripts/import-books.ts" })) ``` It can be run as often as needed. A book that has a stream is skipped, so a script that dies halfway through ten thousand rows is simply started again. ## Carry on Once a book has a stream, it only changes through commands: examples/blog/table-to-log.ts ```ts yield* importBooks(rows) yield* importBooks(rows) // the second run records nothing // From here on, the books only change through commands. yield* books.commands.returnBook("book-1", {}) yield* books.commands.lend("book-2", { loanId: "loan-1", memberId: "bob" }) yield* viewModels.caughtUp() const history = yield* books.history("book-1") // BookImported, with metadata { importedFrom: "books table", … }, then BookReturned const catalog = yield* Catalog.use(Ref.get) // Map { "book-1" => { title: "Dune", …, available: true }, "book-2" => { title: "Emma", …, available: false } } ``` Alice brings *Dune* back, and the book decides on it with the due date from the old table. *Emma* is lent to Bob. The book’s history says where it came from: the first event is the import, with the script in its metadata, and from then on, what really happened. Read models need one more handler, for `BookImported`. Here is the catalog of the series with it: examples/blog/table-to-log.ts ```ts // The catalog of the series, which now also reads imported books. export const CatalogView = ViewModel.make({ name: "catalog", aggregates: [Book], dispose: Catalog.use((catalog) => Ref.set(catalog, new Map())), handlers: { BookImported: ({ author, loan, title }, { aggregateId }) => Catalog.use((catalog) => Ref.update(catalog, (entries) => new Map(entries).set(aggregateId, { title, author, available: loan === null })) ), BookAdded: ({ author, title }, { aggregateId }) => Catalog.use((catalog) => Ref.update(catalog, (entries) => new Map(entries).set(aggregateId, { title, author, available: true })) ), BookLent: (_, { aggregateId }) => setAvailable(aggregateId, false), BookReturned: (_, { aggregateId }) => setAvailable(aggregateId, true) } }) ``` A read model like “loans per month” simply ignores it: an imported loan didn’t start that month, and the event says so by not being a `BookLent`. ## Switching over The import is the easy part. The hard part is the day the old code stops writing the table. * **One aggregate type at a time.** Move books first, members later. Commands on books go through the new code, and the rest of the application keeps its tables until its turn. * **Stop writes to the table, then import.** Anything written to the table after the import is lost to the log. Make the old code read-only for books, run the import, switch the commands over, and turn writes back on through the new path. If that pause is too long, import first, then import again for the rows that changed in between, as `BookImported`s for new books and corrections for the others, then switch. * **Keep the old queries working** with a read model in the old table’s shape, written by a view model. Reports and other teams that read `books` don’t notice the move, and can be moved to better read models one by one. * **Keep the table** read-only for a while after the switch, to compare with the read models. Drop it once they agree. ## Rules of thumb * **An import is an event.** `BookImported` says what the old system knew, on which day, from where. It doesn’t pretend to be the history the table never kept. * **Make the import idempotent.** One event per row, skipped if the stream exists, so the script can run again after any failure. * **Put the source in the metadata.** Every imported stream then says where it came from, and an import with a bug can be found and [corrected](/blog/the-wrong-event/). * **Move one aggregate type at a time**, and keep the old shape alive as a read model until nobody reads it. ## Next [Why CQRS and event sourcing](/blog/why-cqrs-and-event-sourcing/) is where the series starts, and [the storage guide](/guides/storage/) covers moving to SQLite or PostgreSQL. # I clicked twice > A payment request times out and the client sends it again. Disabling the button doesn't help, a unique column helps half way. Commands that carry their own id, or an id per request, run once however often they arrive. *A post after the series [CQRS with effect-cqrs](/blog/), on its lending library. November 2026.*  Alice owes the library two fines and pays one on the website. The spinner turns, the page times out, and she presses the button again. Behind the page, the first request had gone through: only its answer was lost. She has now paid twice. Everyone has seen this bug, on one side of it or the other. It isn’t caused by impatient users. It is what every retry looks like from the server, and retries are everywhere: the browser, the mobile app that reconnects, the load balancer, the HTTP client library, the queue that redelivers. Every snippet runs in [`examples/blog/clicked-twice.test.ts`](https://github.com/davidnussio/effect-cqrs/blob/main/examples/blog/clicked-twice.test.ts). ## A member’s account The library keeps what each member owes as an aggregate: fines charged, payments received, and a balance in cents. One rule: there must be something to pay. examples/blog/clicked-twice.ts ```ts export const AccountEvent = Schema.TaggedUnion({ FineCharged: { amount: Schema.Int, reason: Schema.String }, PaymentReceived: { amount: Schema.Int } }) export class NothingToPay extends Schema.TaggedError()("NothingToPay", { memberId: Schema.String }) {} /** What a member owes the library, in cents. Negative is credit. */ export const Account = Aggregate.make<{ readonly balance: number }>()({ name: "account", events: AccountEvent, initial: { balance: 0 }, projection: { FineCharged: ({ balance }, { amount }) => ({ balance: balance + amount }), PaymentReceived: ({ balance }, { amount }) => ({ balance: balance - amount }) } }) .command("chargeFine", { payload: Schema.Struct({ amount: Schema.Int, reason: Schema.String }), handler: (_, payload) => Effect.succeed(AccountEvent.cases.FineCharged.make(payload)) }) .command("pay", { payload: Schema.Struct({ amount: Schema.Int }), handler: Effect.fnUntraced(function*({ balance }, { amount }, { aggregateId }) { if (balance <= 0) return yield* new NothingToPay({ memberId: aggregateId }) return AccountEvent.cases.PaymentReceived.make({ amount }) }) }) export class Accounts extends Context.Service>()( "library/Accounts" ) { static readonly layer = Layer.effect(Accounts, AggregateService.make(Account)) } ``` ## The same request, twice examples/blog/clicked-twice.ts ```ts yield* accounts.commands.chargeFine("alice", { amount: 300, reason: "Dune, 10 days late" }) yield* accounts.commands.chargeFine("alice", { amount: 300, reason: "Neuromancer, 10 days late" }) // The payment page sends 3 euros. The response is lost, and the page sends it again. yield* accounts.commands.pay("alice", { amount: 300 }) yield* accounts.commands.pay("alice", { amount: 300 }) const alice = yield* accounts.state("alice") // { balance: 0 }: it looks paid, and Alice was charged 6 euros for one click ``` Two `PaymentReceived` events for one click. The balance even looks right, which is the worst part: Alice owed 6 euros and paid 6, and only she knows she meant to pay 3. Nobody notices until she reads her bank statement. Notice the rule didn’t help. With one fine, the second payment would have been refused, because there was nothing left to pay. That makes the bug rarer, not gone: it only shows when a member owes more than one fine, which is when it costs the most to explain. Business rules catch duplicates by luck. ## What the usual fixes miss **Disabling the button** stops the double click and nothing else. The retry that matters is the one the user never sees: the HTTP client that tries again after a timeout, the phone that resends when it gets its signal back, the message that a queue delivers twice. **A unique constraint** is the right instinct. Add a `request_id` column to the payments table, unique, and the second insert fails. But then: * the second request gets an error, while the first succeeded, so the client shows “payment failed” for a payment that went through. To answer it properly, the server has to find the first payment and return what it returned then, which means keeping the response too. * every table that can be written by a retried request needs the column, the index and the code that looks it up. A fine, a loan, a reservation, a renewal. * someone has to decide how long ids are kept, and clean them up. ## Give the request an id The client makes an id once, when Alice clicks, and sends it with every attempt. effect-cqrs calls it a `commandId`: examples/blog/clicked-twice.ts ```ts yield* accounts.commands.chargeFine("alice", { amount: 300, reason: "Dune, 10 days late" }) yield* accounts.commands.chargeFine("alice", { amount: 300, reason: "Neuromancer, 10 days late" }) // The page makes one id when Alice clicks, and sends it with every attempt. const requestId = "7b6f0e1c-pay-dune" const first = yield* accounts.commands.pay("alice", { amount: 300 }, { commandId: requestId }) const again = yield* accounts.commands.pay("alice", { amount: 300 }, { commandId: requestId }) // again equals first: the same version, position and events, and nothing new is stored const alice = yield* accounts.state("alice") // { balance: 300 }: one payment, 3 euros still owed ``` The second attempt didn’t run the handler. The service recognized the id, read back the events the first attempt wrote, and returned the same result: same version, same position, same events. The client can’t tell a retry from the first try, which is the point. It gets the answer it lost. Where are the ids kept? In the log. The id is recorded in the metadata of the events the command wrote, so there is no extra table, and nothing to keep in step: the payment and the proof that it was paid by this request are written in the same append. Each aggregate remembers the ids of its last 100 commands, rebuilt from its events when its state is loaded. An id belongs to one request. Sent again with another command, it is refused rather than answered with a result that isn’t the caller’s: examples/blog/clicked-twice.ts ```ts const reused = yield* Effect.flip( accounts.commands.chargeFine("alice", { amount: 100, reason: "Lost card" }, { commandId: requestId }) ) // reused._tag: "CommandIdReused", with previous: "pay" and command: "chargeFine" ``` ## Or make the command idempotent itself Sometimes the domain already has an id that makes a repeat recognizable. A loan in this library has one, chosen by the client, and the book’s `lend` command records nothing if the book is already lent for that loan: examples/blog/clicked-twice.ts ```ts yield* books.commands.add("book-1", { title: "Dune", author: "Frank Herbert" }) const first = yield* books.commands.lend("book-1", { loanId: "loan-1", memberId: "alice" }) const again = yield* books.commands.lend("book-1", { loanId: "loan-1", memberId: "alice" }) // first.events: [BookLent], again.events: []: the book already has that loan ``` That is the best kind of idempotency, because it doesn’t depend on remembering anything: a retry next month is still harmless. It is why the [saga](/blog/sagas/) in the series can repeat any of its steps. Not every command has such an id. “Pay 3 euros” twice is two payments, unless something says they are the same request. ## Rules of thumb * **Every command that can be retried needs a way to be recognized.** Prefer an id in the domain, like `loanId`. Where there isn’t one, send a `commandId`. * **The client makes the id, once per intention.** A UUID made when the user clicks, kept across retries. An id the server makes can’t help, because the retry doesn’t know it. * **A retry gets the first answer, not an error.** The caller needs the result it lost. * **Don’t count on business rules to catch duplicates.** They catch some, by luck. * **The window is bounded.** An aggregate remembers its last 100 command ids by default, `commandIdsPerStream`. Retries come within seconds or minutes, so that’s plenty for a member’s account, and an aggregate with heavy traffic can keep more. ## Next The [commands guide](/guides/commands/#retries-and-command-ids) covers command ids in full, including two processes running the same command at once. # Read models you can throw away > Part 3 of the series: view models that follow the log, catch up after a restart, read your own writes, and rebuild from scratch. *Part 3 of 5 of [CQRS with effect-cqrs](/blog/). November 2026.*  In [part 2](/blog/your-first-aggregate/) the book became an aggregate: commands decide on events, and `state("book-1")` folds one book’s events into its state. That is the write side, and it is good at one question: what is the state of *this* book, so I can decide what happens to it? The front desk asks other questions. Which books can be borrowed right now? Which books does Alice have? Folding every book on every request would answer them, slowly. The read side of CQRS answers them from read models: data shaped for one query, kept up to date by following the log. effect-cqrs calls the code that keeps one a *view model*. ## A catalog Here is the catalog, one entry per book with a flag for whether it is on the shelf. It lives in memory in a `Ref`, wrapped in a service so that handlers and queries can share it: examples/blog/library.ts ```ts export interface CatalogEntry { readonly title: string readonly author: string readonly available: boolean } /** The catalog: one entry per book, kept in memory. */ export class Catalog extends Context.Service>>()( "library/Catalog" ) { static readonly layer = Layer.effect(Catalog, Ref.make>(new Map())) } const updateEntry = (bookId: string, update: (entry: CatalogEntry) => CatalogEntry) => Catalog.use((catalog) => Ref.update(catalog, (entries) => { const entry = entries.get(bookId) return entry === undefined ? entries : new Map(entries).set(bookId, update(entry)) }) ) export const CatalogView = ViewModel.make({ name: "catalog", aggregates: [Book], dispose: Catalog.use((catalog) => Ref.set(catalog, new Map())), handlers: { BookAdded: ({ author, title }, { aggregateId }) => Catalog.use((catalog) => Ref.update(catalog, (entries) => new Map(entries).set(aggregateId, { title, author, available: true })) ), BookLent: (_, { aggregateId }) => updateEntry(aggregateId, (entry) => ({ ...entry, available: false })), BookReturned: (_, { aggregateId }) => updateEntry(aggregateId, (entry) => ({ ...entry, available: true })) } }) ``` A view model has a name, the aggregates whose events it reads, and a handler for each event it cares about. The handlers get the event and its metadata, which includes the `aggregateId`, the `version`, the `position` in the store, the `timestamp`, and the correlation and causation ids we’ll meet in part 4. `dispose` clears what the view model stored, for when it has to be rebuilt. Notice what is *not* here. There is no `if (book.status === ...)`, no rule, no refusal. The events have already been decided and stored. A handler only records what happened, in the shape a query wants. ## Another shape, same events The second question, which books each member has, wants another shape. It is another view model over the same events: examples/blog/library.ts ```ts /** The books each member has, by member id. */ export class Loans extends Context.Service>>>()( "library/Loans" ) { static readonly layer = Layer.effect(Loans, Ref.make>>(new Map())) } const setOnLoan = (memberId: string, bookId: string, onLoan: boolean) => Loans.use((loans) => Ref.update(loans, (members) => { const books = new Set(members.get(memberId)) if (onLoan) books.add(bookId) else books.delete(bookId) return new Map(members).set(memberId, books) }) ) export const LoansView = ViewModel.make({ name: "loans", aggregates: [Book], dispose: Loans.use((loans) => Ref.set(loans, new Map())), handlers: { BookLent: ({ memberId }, { aggregateId }) => setOnLoan(memberId, aggregateId, true), BookReturned: ({ memberId }, { aggregateId }) => setOnLoan(memberId, aggregateId, false) } }) ``` Read models are cheap. Each one serves its queries and nothing else, so there is no compromise between the catalog page and the member page, and no join at read time. They also don’t share code: changing how the catalog stores a book can’t break the loans, and a read model that needs a search index or raw SQL can have one without the others noticing. Both view models are written so that handling an event twice changes nothing: setting a map entry or adding to a set again leaves them as they were. That matters, because **delivery is at least once**. A view model saves how far it has read, its *checkpoint*, every 500 events and when it stops. After a crash it starts again from the last checkpoint and sees a few events a second time. ## Running them `ViewModel.layer` starts the view models in the background, as long as the layer lives. They need the event store and somewhere to keep their checkpoints: examples/blog/library.ts ```ts export const ReadSide = ViewModel.layer([CatalogView, LoansView]).pipe( Layer.provideMerge(Books.layer), Layer.provideMerge(Layer.mergeAll(EventStore.layerMemory, Checkpoints.layerMemory, Catalog.layer, Loans.layer)) ) ``` Each view model *pulls* events from the store, from its own checkpoint, rather than waiting for them to be pushed. A view model that was down, or that is new, reads what it missed and then waits for more. Nothing is lost while it isn’t running, and there is no message broker to configure. The checkpoints here are in memory, which suits read models in memory: both are gone after a restart, and the view models rebuild from the start of the log. A read model kept in a database pairs with checkpoints in the same database. **A checkpoint must live exactly as long as what its view model stores**, or after a restart it says “done” for data that isn’t there. ## Reading your own writes The read side runs a moment behind the write side. A command returns when its events are stored, and the view models see them a little later. A user who lends a book and then looks at the catalog could see it still on the shelf. Every command returns the `position` of its last event in the store, and `caughtUp` waits until every view model has handled it: examples/blog/library.ts ```ts yield* books.commands.add("book-1", { title: "Dune", author: "Frank Herbert" }) yield* books.commands.add("book-2", { title: "Neuromancer", author: "William Gibson" }) const { position } = yield* books.commands.lend("book-1", { loanId: "loan-1", memberId: "alice" }) // The view models run in the background: wait until they have seen the loan. yield* viewModels.caughtUp(position) const catalog = yield* Catalog.use(Ref.get) // Map { "book-1" => { title: "Dune", …, available: false }, "book-2" => { …, available: true } } const alice = (yield* Loans.use(Ref.get)).get("alice") // Set { "book-1" } ``` Without a position, `caughtUp()` waits for the latest event in the store. Over HTTP, `CommandApi` answers every command with the position in an `Event-Position` header, and `CommandApi.ReadAfter` makes a query wait for the one a client sends back. [Three ways to read](/blog/three-ways-to-read/) puts it together with the other two ways of reading a book. ## Throw it away A read model is not the truth, the log is. So you can throw one away and build it again: examples/blog/library.ts ```ts // Throw the catalog away and build it again from the log. const { eventCount } = yield* viewModels.replay({ viewModels: ["catalog"] }) const rebuilt = yield* Catalog.use(Ref.get) // The same entries as before, from 3 events ``` `replay` resets the checkpoint, runs `dispose`, reads the log from the start and returns once the view model has caught up. This is how you fix a read model that a bug damaged, or one whose shape changed: deploy the new handlers and replay. [Fix the code, not the data](/blog/fix-the-code-not-the-data/) walks through one. A brand new view model needs no replay at all. Its checkpoint starts at zero, so it begins with the first event ever stored. In part 1, `timesLent` answered a question nobody had asked when the events were written. In a running system, that is a new view model: write the handlers, deploy, and it fills itself from the whole history. ## When a handler fails A handler that fails, say because a database is down, stalls its view model on that event and retries with backoff, up to every 30 seconds. Events are never skipped and never reordered, so a read model is either right or visibly behind. `viewModels.status` shows each view model’s position and the failure it is retrying. For read models where a gap is better than a stop, `onFailure: ViewModel.park({ attempts })` sets an event aside after a few failures, and `retryParked` handles it again once the bug is fixed. The [view models guide](/guides/view-models/#failing-handlers) has the details. ## What we have The library now has two sides. The write side keeps the rules, one aggregate at a time. The read side follows the log, in as many shapes as the queries need, and can be rebuilt whenever we like. What it can’t do yet is anything that involves *two* aggregates. A member may only have three books at a time, and a book can only be with one member. Borrowing has to respect both rules, and a command only changes one aggregate. [Part 4](/blog/sagas/) solves this with a saga. # Sagas > Part 4 of the series: borrowing a book touches the member and the book, so a saga coordinates them as a durable workflow. *Part 4 of 5 of [CQRS with effect-cqrs](/blog/). 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 ```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](/blog/tests-schema-changes-and-production/) explains how it works. examples/blog/library.ts ```ts export class MemberAlreadyJoined extends Schema.TaggedError()("MemberAlreadyJoined", { memberId: Schema.String }) {} export class MemberNotFound extends Schema.TaggedError()("MemberNotFound", { memberId: Schema.String }) {} export class LoanLimitReached extends Schema.TaggedError()("LoanLimitReached", { memberId: Schema.String, limit: Schema.Int }) {} ``` examples/blog/library.ts ```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> } const withoutLoan = (state: MemberState, loanId: string): MemberState => { const { [loanId]: _, ...loans } = state.loans return { ...state, loans } } export const Member = Aggregate.make()({ 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 ```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](/guides/sagas/#the-workflow). 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 ```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](/blog/read-models/), whose handlers only start runs: examples/blog/library.ts ```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 ```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 ```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) => 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 ```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](/blog/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 ```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 ```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](/tutorial/shop/#5-a-saga) 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](/blog/tests-schema-changes-and-production/) covers testing, changing an event’s shape, moving to a database, and forgetting a member. # Tests, schema changes and production > Part 5 of the series: test aggregates with given, when, then and random commands, change an event's shape without rewriting the log, move to SQLite or PostgreSQL, and forget a member. *Part 5 of 5 of [CQRS with effect-cqrs](/blog/). November 2026.*  Over the last four posts the library grew books, members, read models and a saga, all in memory. This last post is about living with it: testing the rules, changing an event that is already stored, moving to a real database, and forgetting a member who asks for it. ## Given, when, then An aggregate is a pure decision: given these past events, when this command comes, these events happen, or this error. `Fixture` tests exactly that: examples/blog/library.test.ts ```ts const added = BookEvent.cases.BookAdded.make({ title: "Dune", author: "Frank Herbert" }) const lent = BookEvent.cases.BookLent.make({ loanId: "loan-1", memberId: "alice", dueOn: DateTime.makeUnsafe(0).pipe(DateTime.add({ days: 21 })) }) it.effect("lends a book for three weeks", () => Fixture.make(Book) .givenEvents([added]) .when("lend", { loanId: "loan-1", memberId: "alice" }) .expectEvent(lent) .inspectState((state) => assert.strictEqual(state.status, "lent"))) it.effect("records nothing when the same loan comes again", () => Fixture.make(Book) .givenEvents([added, lent]) .when("lend", { loanId: "loan-1", memberId: "alice" }) .expectEvents([])) it.effect("refuses a book that is out", () => Effect.gen(function*() { const error = yield* Fixture.make(Book) .givenEvents([added, lent]) .whenFails("lend", { loanId: "loan-2", memberId: "bob" }) assert.strictEqual(error._tag, "BookAlreadyLent") })) it.effect("notes a book that comes back late", () => Effect.gen(function*() { yield* TestClock.adjust("22 days") yield* Fixture.make(Book) .givenEvents([added, lent]) .when("returnBook", {}) .expectEvent(BookEvent.cases.BookReturned.make({ loanId: "loan-1", memberId: "alice", late: true })) })) ``` The chain runs the command through the real `AggregateService` on an in-memory store, so payload validation and the JSON round trip of every event are tested too. An expected event is compared through the aggregate’s schema, and a mismatch shows the difference. Time needs no mocking. `it.effect` runs on a test clock stopped at zero, so `DateTime.now` in the `lend` handler gives the epoch, and the due date is exactly 21 days later. To test a late return, `TestClock.adjust` moves time on by 22 days before the book comes back. Caution The chain is lazy, like every effect. Built and never returned from `it.effect` or yielded, it checks nothing and the test passes. ## Rules that hold for any sequence Examples test the cases you think of. The rule “never more than three books” should hold for *any* sequence of commands, including the ones you don’t think of. `checkInvariants` runs random sequences, with payloads generated from the command schemas, and checks a predicate after every command that succeeds: examples/blog/library.test.ts ```ts // A few loan ids, so that random commands often talk about the same loan. const LoanId = Schema.Literals(["loan-1", "loan-2", "loan-3", "loan-4", "loan-5", "loan-6", "loan-7", "loan-8"]) it.effect("never has more than three books out for a member", () => Fixture.make(Member) .givenEvents([MemberEvent.cases.MemberJoined.make({ name: "Alice Rossi" })]) .checkInvariants({ invariant: (state) => Object.keys(state.loans).length <= loanLimit, commands: { join: false, requestLoan: Arbitrary.schema(Schema.Struct({ loanId: LoanId, bookId: Schema.String })), confirmLoan: Arbitrary.schema(Schema.Struct({ loanId: LoanId })), cancelLoan: Arbitrary.schema(Schema.Struct({ loanId: LoanId, reason: Schema.String })), closeLoan: Arbitrary.schema(Schema.Struct({ loanId: LoanId })) }, check: { runs: 1000, seed: 1 } })) ``` Loan ids come from a small set, so that random commands keep talking about the same loans: a confirmation of a loan that was requested, a cancellation of one that was already confirmed, the same request twice. Commands that fail are part of the game. When the invariant breaks, the test fails with the shortest sequence of commands it found that breaks it, which reads like a bug report. Try it: change `>=` to `>` in `requestLoan`, and after a few hundred runs the test fails with four `requestLoan` commands for four different loans. A thousand runs take a fraction of a second, and the fixed `seed` makes every run of the test try the same sequences. ## Changing an event’s shape Events written today will be read by code written years from now, and the log is never rewritten. Suppose the first version of the library recorded `BookAdded` with a title only, and authors came later. The events already stored don’t have one. effect-cqrs records the schema version of every event. To change an event, declare its older shapes and how each one becomes the next: examples/blog/library.ts ```ts // The first version of the library didn't record authors. const BookAddedV1 = Schema.Struct({ title: Schema.NonEmptyString }) const bookVersions = { BookAdded: EventVersions.from(BookAddedV1).next( BookEvent.cases.BookAdded, (v1) => ({ ...v1, author: "Unknown" }) ) } ``` That is the `versions: bookVersions` line we skipped in [part 2](/blog/your-first-aggregate/). An old `BookAdded` is *upcast* as it is read: projections, view models and sagas only ever see the current shape, with `author: "Unknown"`. New events are written in the latest version. The types check every step: an upcast that returns the wrong shape, or a history that doesn’t end at the current schema, doesn’t compile. A test can seed events as old code would have stored them, with `Aggregate.historical`: examples/blog/library.test.ts ```ts it.effect("reads books added before authors were recorded", () => Fixture.make(Book) .givenEvents([Aggregate.historical(Book, "BookAdded", 1, { title: "Dune" })]) .inspectState((state) => assert.strictEqual(state.author, "Unknown"))) ``` Upcasting covers adding a field, renaming one or splitting one. A change of meaning is better as a new event type, next to the old one. ## Moving to a database Everything so far ran on `EventStore.layerMemory`. Because the application is a layer that *requires* a store, moving to a database is a change in one place. Here are the events on SQLite: examples/blog/storage.ts ```ts // The events in one file and the keys in another. The read models and the workflow // engine live in memory, so their checkpoints do too: on every start, they read the // log again from the beginning. export const onSqlite = (files: { readonly events: string; readonly keys: string }) => Layer.mergeAll( SqliteEventStore.layer().pipe(Layer.provide(SqliteClient.layer({ filename: files.events }))), Checkpoints.layerMemory, PersonalData.layerSql().pipe(Layer.provide(SqliteClient.layer({ filename: files.keys }))) ) // The master key comes from PERSONAL_DATA_MASTER_KEY and PERSONAL_DATA_MASTER_KEY_ID. export const LibraryOnSqlite = Application.pipe( Layer.provideMerge(onSqlite({ events: "./data/library.sqlite", keys: "./data/keys.sqlite" })), Layer.provide(PersonalData.layerMasterKeyConfig()) ) ``` Tables are created on startup. Only the events moved: the read models and the workflow engine are still in memory, so their checkpoints are too, and they rebuild from the log on every start. That is a fine setup for one process, and the rule from [part 3](/blog/read-models/) holds: a checkpoint lives exactly as long as what it describes. PostgreSQL works the same way: examples/blog/storage.ts ```ts // The same on PostgreSQL, where several processes can share the events. export const onPostgres = (urls: { readonly events: Redacted.Redacted; readonly keys: Redacted.Redacted }) => Layer.mergeAll( PostgresEventStore.layer().pipe(Layer.provide(PgClient.layer({ url: urls.events }))), Checkpoints.layerMemory, PersonalData.layerSql().pipe(Layer.provide(PgClient.layer({ url: urls.keys }))) ) ``` On PostgreSQL several processes can write to the same events. Each process still runs every view model and saga trigger in this setup, which the idempotent handlers and commands make safe, but wasteful. To run each one in one process only, add `Leases.layerPostgres()` for the view models, and a durable workflow engine with durable checkpoints for the sagas, see [several processes](/guides/view-models/#several-processes) and [delivery and durability](/guides/sagas/#delivery-and-durability). The tests of this series run the whole library on SQLite in memory, with `onSqlite({ events: ":memory:", keys: ":memory:" })`. ## Forgetting a member The GDPR lets Alice ask to be forgotten, and the log can’t be edited. effect-cqrs gets around that with *crypto-shredding*: in [part 4](/blog/sagas/), the member’s name was marked with `PersonalData.field`. Every personal field is encrypted in the store with a key of its own subject, here the member, and forgetting the member destroys the key. Her events stay where they are, with their loans and dates, so streams still replay and the saga’s history still adds up, while her name reads as `"[forgotten]"`. That is why `onSqlite` keeps the keys in a database of their own, wrapped by a master key from the environment. The keys must be kept apart from the events, with their own permissions and backups. A read model that stored the name has to erase it too. The front desk’s directory handles `onForget`: examples/blog/library.ts ```ts /** Members' names, for the front desk. */ export class Directory extends Context.Service>>()( "library/Directory" ) { static readonly layer = Layer.effect(Directory, Ref.make>(new Map())) } export const DirectoryView = ViewModel.make({ name: "directory", aggregates: [Member], dispose: Directory.use((directory) => Ref.set(directory, new Map())), handlers: { MemberJoined: ({ name }, { aggregateId }) => Directory.use((directory) => Ref.update(directory, (names) => new Map(names).set(aggregateId, name))) }, // The names are personal data: erase a member's when they are forgotten. onForget: (subject) => Directory.use((directory) => Ref.update(directory, (names) => { const rest = new Map(names) rest.delete(subject) return rest }) ) }) ``` And the whole thing, as a test: examples/blog/library.test.ts ```ts it.effect("forgets a member", () => Effect.gen(function*() { const members = yield* Members const viewModels = yield* ViewModel.ViewModels yield* members.commands.join("alice", { name: "Alice Rossi" }) yield* PersonalData.forget("alice") yield* viewModels.caughtUp() const alice = yield* members.state("alice") assert.strictEqual(alice.name, "[forgotten]") const directory = yield* Directory.use(Ref.get) assert.isFalse(directory.has("alice")) }).pipe(Effect.provide(Library), quiet)) ``` After `forget`, Alice’s stream still holds every loan, her name reads as the replacement, and the directory no longer has it. The forget itself is recorded as an event, with no personal data in it, and from then on a command that would record new personal data about her fails with `SubjectForgotten`. The [personal data guide](/guides/personal-data/) covers master key rotation, keys in a KMS, and what crypto-shredding doesn’t cover. ## Where to go from here That is the whole library, in about six hundred lines of [`examples/blog`](https://github.com/davidnussio/effect-cqrs/tree/main/examples/blog): * **Events** as schemas, the only source of truth, versioned as they change. * **Aggregates** that decide on events, with typed errors and optimistic concurrency. * **View models** that follow the log, catch up, and can be thrown away and rebuilt. * **A saga** that coordinates two aggregates as a durable workflow. * **Tests** that read like the rules, and find the sequences you didn’t think of. From here: * The [bank demo](/tutorial/bank/) runs all of this in your browser, and shows each command travelling through the pipeline. * The [workshop](/tutorial/workshop/) runs it on a server, for a room full of phones. * The [guides](/guides/aggregates/) go deeper into every piece, and the [comparison with moleculer-cqrs](/reference/moleculer-cqrs/) shows where the framework comes from. Thanks for reading the series. # The bug that won't reproduce > A member can't borrow anything, and every test passes. The data that triggers the bug is in production, as a history nobody can rebuild by hand. With events, that history is a file: copy the stream into a fixture, and the bug report is a failing test. *A post after the series [CQRS with effect-cqrs](/blog/), on its lending library. November 2026.*  The ticket says: *Carla can’t borrow anything. The website says she has reached her limit of three books. She has none.* The developer tries it: a new member, a loan, it works. Three loans, a fourth refused, correct. Every test is green. *Cannot reproduce*, and the ticket waits for more information that never comes. Bugs like this depend on history, a sequence of things that happened to one record over weeks. Nobody can guess the sequence, and a database row doesn’t keep it. Every snippet here runs in [`examples/blog/wont-reproduce.test.ts`](https://github.com/davidnussio/effect-cqrs/blob/main/examples/blog/wont-reproduce.test.ts). ## The member This member aggregate is a variant of the one in [the series](/blog/sagas/). At some point someone wanted members to see their cancelled loans on their page, so a cancelled loan is now kept, marked `cancelled`, instead of removed: examples/blog/wont-reproduce.ts ```ts const makeMember = (countsAgainstLimit: (loan: Loan) => boolean) => Aggregate.make()({ name: "member", events: MemberEvent, initial: { joined: false, loans: {} }, projection: { MemberJoined: (state) => ({ ...state, joined: true }), LoanRequested: (state, { bookId, loanId }) => ({ ...state, loans: { ...state.loans, [loanId]: { bookId, status: "requested" } } }), LoanConfirmed: (state, { loanId }) => setStatus(state, loanId, "active"), // Cancelled loans are kept, so the member's page can list them. LoanCancelled: (state, { loanId }) => setStatus(state, loanId, "cancelled") } }) .command("join", { payload: Schema.Struct({}), handler: Effect.fnUntraced(function*(state, _, { aggregateId }) { if (state.joined) return yield* new MemberAlreadyJoined({ memberId: aggregateId }) return MemberEvent.cases.MemberJoined.make({}) }) }) .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 (Object.values(state.loans).filter(countsAgainstLimit).length >= loanLimit) { return yield* new LoanLimitReached({ memberId: aggregateId, limit: loanLimit }) } return MemberEvent.cases.LoanRequested.make({ loanId, bookId }) }) }) // What the saga runs when the book is out. .command("cancelLoan", { payload: Schema.Struct({ loanId: Schema.String, reason: Schema.String }), handler: (state, payload) => Effect.succeed( state.loans[payload.loanId]?.status === "requested" ? MemberEvent.cases.LoanCancelled.make(payload) : [] ) }) const setStatus = (state: MemberState, loanId: string, status: Loan["status"]): MemberState => { const loan = state.loans[loanId] return loan === undefined ? state : { ...state, loans: { ...state.loans, [loanId]: { ...loan, status } } } } // The bug: every loan the member ever had counts against the limit, cancelled ones too. export const Member = makeMember(() => true) ``` The change looked harmless and the tests passed: none of them cancels a loan and then asks for another. ## Works for me The developer’s test is the obvious one, and it passes: examples/blog/wont-reproduce.test.ts ```ts it.effect("lends to a member who just joined", () => Fixture.make(Member) .givenEvents([MemberEvent.cases.MemberJoined.make({})]) .when("requestLoan", { loanId: "loan-1", bookId: "dune" }) .expectEvent("LoanRequested")) ``` What happened to Carla is not obvious. Over a few weeks she asked for three books that were out, and the saga cancelled each request. Then she asked for a fourth: examples/blog/wont-reproduce.ts ```ts yield* members.commands.join("carla", {}) for (const [loanId, bookId] of [["loan-1", "dune"], ["loan-2", "emma"], ["loan-3", "ulysses"]]) { yield* members.commands.requestLoan("carla", { loanId, bookId }) yield* members.commands.cancelLoan("carla", { loanId, reason: "BookAlreadyLent" }) } const refused = yield* Effect.flip(members.commands.requestLoan("carla", { loanId: "loan-4", bookId: "solaris" })) // refused: LoanLimitReached { memberId: "carla", limit: 3 }, and Carla has no books ``` With a table, all the developer would find is a row with `open_loans = 3`, or three rows in a `loans` table with a status. How did they get there? In which order? Was the counter ever right? The row is the end of a story, and the story is gone. ## Copy the stream With events, the story is the data. Carla’s stream can be exported as it is, as JSON, by a support script with read access: examples/blog/wont-reproduce.ts ```ts // The support script: Carla's stream, as JSON, for the bug report. const history = yield* members.history("carla", { payload: true }) const encode = Schema.encodeSync(Schema.toCodecJson(MemberEvent)) const exported = JSON.stringify(history.map(({ payload }) => encode(payload!)), null, 2) ``` The result is short and readable, which is half the diagnosis already: examples/blog/carla.json ```json [ { "_tag": "MemberJoined" }, { "_tag": "LoanRequested", "loanId": "loan-1", "bookId": "dune" }, { "_tag": "LoanCancelled", "loanId": "loan-1", "reason": "BookAlreadyLent" }, { "_tag": "LoanRequested", "loanId": "loan-2", "bookId": "emma" }, { "_tag": "LoanCancelled", "loanId": "loan-2", "reason": "BookAlreadyLent" }, { "_tag": "LoanRequested", "loanId": "loan-3", "bookId": "ulysses" }, { "_tag": "LoanCancelled", "loanId": "loan-3", "reason": "BookAlreadyLent" } ] ``` The file goes into the repository next to the test, and the test reads it: examples/blog/wont-reproduce.test.ts ```ts // Carla's stream, as the support script exported it. const carla = decodeStream(readFileSync(join(import.meta.dirname, "carla.json"), "utf8")) ``` Caution A stream can hold personal data: names, addresses, notes. Leave it out or replace it before it leaves production. With [personal data fields](/guides/personal-data/), `history` returns decrypted values, so the export script is the place to redact them. ## A failing test A fixture runs a command against a given history, without a database, in milliseconds. Given Carla’s history: examples/blog/wont-reproduce.test.ts ```ts it.effect("refuses Carla, who has no books", () => Effect.gen(function*() { const error = yield* Fixture.make(Member).givenEvents(carla).whenFails("requestLoan", { loanId: "loan-4", bookId: "solaris" }) assert.strictEqual(error._tag, "LoanLimitReached") })) ``` It fails the same way production did. The bug report is now a test, and the cause is one `inspectState` away: three loans in `state.loans`, all `cancelled`, all counted by the limit check. The fix counts only the loans that are open: examples/blog/wont-reproduce.ts ```ts // The fix: only loans that are open count. export const FixedMember = makeMember((loan) => loan.status !== "cancelled") ``` And Carla’s history becomes a regression test that stays in the suite: examples/blog/wont-reproduce.test.ts ```ts it.effect("lends to Carla, whose earlier loans were all cancelled", () => Fixture.make(FixedMember) .givenEvents(carla) .when("requestLoan", { loanId: "loan-4", bookId: "solaris" }) .expectEvent("LoanRequested")) ``` ## Carla’s account fixes itself There is no data to repair. Carla’s stream says what happened, and it was right all along: she asked, the books were out, the requests were cancelled. The member’s state, the one the limit check reads, is folded from those events every time, so the moment the fix is deployed, her state counts zero open loans and her next request goes through. The same goes for every other member who hit it, whether or not they filed a ticket. With an `open_loans` counter in a table, the fix would also need a migration to recompute the counter for everyone, from data that may not say what it should be. ## Rules of thumb * **Ask for the aggregate id, not for steps to reproduce.** The stream *is* the steps. * **Keep a support script that exports a stream as JSON**, encoded with the events’ own schema, and redacting personal data. * **Turn every such bug into a fixture with the real history.** It’s the shortest reproduction there is, and it keeps the bug from coming back. * **When a change adds a state, test the commands that read it.** Here a new status, `cancelled`, changed what the limit check counted, and no test asked for a loan after a cancellation. ## Next [Tests, schema changes and production](/blog/tests-schema-changes-and-production/) in the series covers fixtures and invariant tests, and the [testing guide](/guides/testing/) covers `Fixture` in full. # The double write > Store the loan, then tell the other systems, or the other way round: each order breaks on a bad day, and no transaction spans a database and a broker. With event sourcing the log is already the outbox. *A post after the series [CQRS with effect-cqrs](/blog/), on its lending library. November 2026.*  Sooner or later the library has to tell someone else. The notifications service emails members their due dates, the statistics team counts loans, the city’s network of libraries wants to know which books are out. They live in other processes, and they hear about loans through a message broker: Kafka, NATS, SQS, RabbitMQ. So a loan is two writes: one to our database, one to the broker. This post is about the moment between them, which is where a whole class of production bugs lives. Every snippet runs in [`examples/blog/double-write.test.ts`](https://github.com/davidnussio/effect-cqrs/blob/main/examples/blog/double-write.test.ts). ## A broker that is sometimes down The example’s broker keeps messages in memory and can be told to fail the next few sends, the way a real one does when the network drops or it is restarting: examples/blog/double-write.ts ```ts export class BrokerDown extends Schema.TaggedError()("BrokerDown", {}) {} export interface Sent { readonly id: string | null readonly type: string readonly payload: unknown } /** A message broker in memory, which can be told to fail the next few sends. */ export class Broker extends Context.Service): Effect.Effect readonly received: Effect.Effect> failNext(times: number): Effect.Effect }>()("library/Broker") { static readonly layer = Layer.effect( Broker, Effect.gen(function*() { const received = yield* Ref.make>([]) const failures = yield* Ref.make(0) return Broker.of({ send: (messages) => Ref.modify(failures, (left) => [left > 0, Math.max(0, left - 1)]).pipe( Effect.flatMap((down) => down ? Effect.fail(new BrokerDown()) : Ref.update(received, (all) => [...all, ...messages]) ) ), received: Ref.get(received), failNext: (times) => Ref.set(failures, times) }) }) ) } ``` ## Store, then send The obvious version stores the loan, then publishes it: examples/blog/double-write.ts ```ts export const storeThenSend = Effect.fnUntraced(function*(bookId: string, loanId: string, memberId: string) { const books = yield* Books const broker = yield* Broker yield* books.commands.lend(bookId, { loanId, memberId }) yield* broker.send([{ id: null, type: "library.book-lent.v1", payload: { bookId, loanId, memberId } }]) }) ``` On a day the broker hiccups for one send: examples/blog/double-write.ts ```ts yield* books.commands.add("book-1", { title: "Dune", author: "Frank Herbert" }) yield* broker.failNext(1) const failed = yield* Effect.flip(storeThenSend("book-1", "loan-1", "alice")) // failed._tag: "BrokerDown", so the front desk tells Alice it didn't work const book = yield* books.state("book-1") // { status: "lent", loan: { loanId: "loan-1", memberId: "alice", … }, … } const afterStoring = yield* broker.received // []: nobody else will ever hear of loan-1 ``` The loan is stored. The message is lost. And the request failed, so the front desk tells Alice that the loan didn’t work while the library’s own records say she has the book. If the desk sends the same request again, `lend` records nothing for a loan it already has, and the send gets through this time. If nobody does, because Alice walked off with the book or tried another one, the notifications service never emails her a due date, and the statistics are one loan short, forever. A crash between the two lines does the same: a deploy, an out-of-memory kill, a pod moved to another node. There is no line of code that can run “after the commit, guaranteed”. A retry around the send helps with a short hiccup and does nothing for a process that is gone. ## Send, then store So swap them. If the message must not be lost, send it first: examples/blog/double-write.ts ```ts export const sendThenStore = Effect.fnUntraced(function*(bookId: string, loanId: string, memberId: string) { const books = yield* Books const broker = yield* Broker yield* broker.send([{ id: null, type: "library.book-lent.v1", payload: { bookId, loanId, memberId } }]) yield* books.commands.lend(bookId, { loanId, memberId }) }) ``` Now the failure goes the other way: examples/blog/double-write.ts ```ts const refused = yield* Effect.flip(sendThenStore("book-1", "loan-2", "bob")) // refused._tag: "BookAlreadyLent" const afterSending = yield* broker.received // [{ type: "library.book-lent.v1", payload: { bookId: "book-1", loanId: "loan-2", memberId: "bob" } }]: // everybody heard of a loan that never happened ``` The broker accepted the message. Then the book refused, because Alice has it. Every consumer now believes Bob borrowed *Dune*: Bob gets an email with a due date for a book he doesn’t have, and the statistics count a loan that never happened. The command could also fail for duller reasons, a lost connection or a version conflict, with the same result. Sending first means announcing a decision before it is made, and the aggregate is the one that makes it. ## The outbox The usual fix in a CRUD application is the *transactional outbox*: in the same transaction as the change, insert a row into an `outbox` table, and let a separate process read that table and publish what it finds, marking rows as sent. One transaction, one database, and the broker is only ever written from a table that already committed. With event sourcing you already have that table. The events are written in one append, they are never changed, and they have a position. A process that reads the log from where it last stopped and publishes what it finds is an outbox, without the extra table and without anyone having to remember to insert into it. effect-cqrs calls it `Outbox`: examples/blog/double-write.ts ```ts export const LoansOutbox = Outbox.make({ name: "loans-outbox", aggregates: [Book], messages: { BookLent: ({ loanId, memberId }, { aggregateId }) => ({ type: "library.book-lent.v1", payload: { bookId: aggregateId, loanId, memberId } }), BookReturned: ({ late, loanId }, { aggregateId }) => ({ type: "library.book-returned.v1", payload: { bookId: aggregateId, loanId, late } }) }, publish: (messages) => Broker.use((broker) => broker.send(messages.map(({ id, payload, type }) => ({ id, type, payload })))) }) export const Library = Outbox.layer([LoansOutbox]).pipe( Layer.provideMerge(Books.layer), Layer.provideMerge(Layer.mergeAll(EventStore.layerMemory, Checkpoints.layerMemory, Broker.layer)) ) ``` `messages` turns domain events into integration events: a type with a version in it, and a payload that is a contract with other teams, which is not the same as the event’s shape. Events without a mapping, like `BookAdded`, are not published. The same bad day, with the outbox: examples/blog/double-write.ts ```ts yield* books.commands.add("book-1", { title: "Dune", author: "Frank Herbert" }) yield* broker.failNext(2) // The command only stores. The broker isn't on its path, so it can't fail it. const { position } = yield* books.commands.lend("book-1", { loanId: "loan-1", memberId: "alice" }) // The outbox tries, fails, waits a little longer each time, and gets through. // `published` fails while the broker is down, so the example asks again. yield* outboxes.published(position).pipe(Effect.retry(Schedule.spaced("50 millis"))) const received = yield* broker.received // [{ id: "book/book-1@2", type: "library.book-lent.v1", payload: { bookId: "book-1", loanId: "loan-1", … } }] ``` The command succeeded although the broker was down, because the broker was never on its path: a command only appends. The outbox runs in the background, like a view model. It read the `BookLent`, failed to publish it, waited, tried again with growing pauses, and got through when the broker came back. Alice was told the truth at the desk, and everybody else heard about her loan a little later. ## What it costs * **Delivery is at least once.** The outbox saves its checkpoint after a batch is published. If the process dies in between, the batch is published again after the restart. Each message has an `id`, the event’s id, such as `book/book-1@2`, which never changes, so consumers keep the ids they have handled and skip repeats. Exactly once is not on offer from any broker either: at least once plus deduplication is how it’s done. * **The others hear a little later.** Usually milliseconds: a batch goes out when it is full or after `linger`, ten milliseconds by default. While the broker is down, it is longer, and `status` shows the outbox retrying. * **Order is per log.** Messages leave in the order the events were stored, and a failed batch is retried before anything after it. A consumer that reads in parallel has to keep order per aggregate itself, as Kafka does with partition keys. * **The checkpoint must be durable.** With checkpoints in memory, an outbox publishes the whole log again after every restart. Keep them in the database, next to the events. ## Rules of thumb * **Never write to two systems in one request.** One of the writes has to come from the other, after it has committed. * **A command only stores.** Whatever has to happen because of a decision reads it from the log: read models, sagas and outboxes. * **Publish integration events, not your events.** Map them in the outbox, give them a versioned type, and leave out what other teams shouldn’t depend on, or keep: personal data in particular. * **Consumers deduplicate on the message id.** It is the same every time a message is published. ## Next The [integration guide](/guides/integration/#outbox) covers `Outbox` in full, and [The saga you skipped](/blog/the-saga-you-skipped/) shows the same two-writes problem between two aggregates, where a saga plays the outbox’s part. # The form that overwrites > Two people edit the same record, and the second save quietly undoes the first. Event sourcing doesn't fix it by itself: commands that say what changed avoid most collisions, and an expected version catches the rest. *A post after the series [CQRS with effect-cqrs](/blog/), on its lending library. November 2026.*  Bianca notices that the catalog spells Frank Herbert without its second R, and fixes it. A minute later Carlo, who opened the same record before her fix, moves the book to another shelf and saves. The typo is back. Nobody did anything wrong, nobody got an error, and the fix is gone until someone notices it again. This is the *lost update*, and it is as old as forms. It lives in admin panels, CMSs, CRMs and every screen with a Save button. This post shows that event sourcing doesn’t fix it by itself, and what does. Every snippet runs in [`examples/blog/form-overwrites.test.ts`](https://github.com/davidnussio/effect-cqrs/blob/main/examples/blog/form-overwrites.test.ts). ## The record and the screens A book’s record in the catalog is an aggregate, `listing`, with a title, an author and a shelf. It has two kinds of commands on purpose: one that saves the whole form, as a CRUD screen does, and some that say what changed: examples/blog/form-overwrites.ts ```ts const Fields = { title: Schema.NonEmptyString, author: Schema.NonEmptyString, shelf: Schema.NonEmptyString } export const ListingEvent = Schema.TaggedUnion({ ListingCreated: Fields, // What a form that saves every field records. ListingSaved: Fields, // What commands that say what changed record. AuthorCorrected: { author: Schema.NonEmptyString }, MovedToShelf: { shelf: Schema.NonEmptyString }, Retitled: { title: Schema.NonEmptyString } }) export interface ListingState { readonly title: string readonly author: string readonly shelf: string } /** A book's record in the catalog: what the librarians edit. */ export const Listing = Aggregate.make()({ name: "listing", events: ListingEvent, initial: { title: "", author: "", shelf: "" }, projection: { ListingCreated: (_, { author, shelf, title }) => ({ title, author, shelf }), ListingSaved: (_, { author, shelf, title }) => ({ title, author, shelf }), AuthorCorrected: (state, { author }) => ({ ...state, author }), MovedToShelf: (state, { shelf }) => ({ ...state, shelf }), Retitled: (state, { title }) => ({ ...state, title }) } }) .command("create", { payload: Schema.Struct(Fields), handler: (_, fields) => Effect.succeed(ListingEvent.cases.ListingCreated.make(fields)) }) .command("saveForm", { payload: Schema.Struct(Fields), handler: (_, fields) => Effect.succeed(ListingEvent.cases.ListingSaved.make(fields)) }) .command("correctAuthor", { payload: Schema.Struct({ author: Schema.NonEmptyString }), handler: (state, { author }) => Effect.succeed(state.author === author ? [] : ListingEvent.cases.AuthorCorrected.make({ author })) }) .command("moveToShelf", { payload: Schema.Struct({ shelf: Schema.NonEmptyString }), handler: (state, { shelf }) => Effect.succeed(state.shelf === shelf ? [] : ListingEvent.cases.MovedToShelf.make({ shelf })) }) .command("retitle", { payload: Schema.Struct({ title: Schema.NonEmptyString }), handler: (state, { title }) => Effect.succeed(state.title === title ? [] : ListingEvent.cases.Retitled.make({ title })) }) ``` The librarians’ screens come from a read model. Next to the fields, it keeps the `version` of the record it shows, from the metadata of the last event it handled: examples/blog/form-overwrites.ts ```ts export interface Screen extends ListingState { /** The version of the record this screen shows, which an edit sends back. */ readonly version: number } /** What the librarians' screens show: every record, with the version it was read at. */ export class Screens extends Context.Service>>()("library/Screens") { static readonly layer = Layer.effect(Screens, Ref.make>(new Map())) } const show = (listingId: string, version: number, update: (fields: ListingState) => ListingState) => Screens.use((screens) => Ref.update(screens, (all) => { const fields = all.get(listingId) ?? { title: "", author: "", shelf: "" } return new Map(all).set(listingId, { ...update(fields), version }) }) ) export const ScreensView = ViewModel.make({ name: "screens", aggregates: [Listing], dispose: Screens.use((screens) => Ref.set(screens, new Map())), handlers: { ListingCreated: ({ author, shelf, title }, { aggregateId, version }) => show(aggregateId, version, () => ({ title, author, shelf })), ListingSaved: ({ author, shelf, title }, { aggregateId, version }) => show(aggregateId, version, () => ({ title, author, shelf })), AuthorCorrected: ({ author }, { aggregateId, version }) => show(aggregateId, version, (f) => ({ ...f, author })), MovedToShelf: ({ shelf }, { aggregateId, version }) => show(aggregateId, version, (f) => ({ ...f, shelf })), Retitled: ({ title }, { aggregateId, version }) => show(aggregateId, version, (f) => ({ ...f, title })) } }) ``` ## The whole form The CRUD version of a save is `UPDATE listings SET title = ?, author = ?, shelf = ? WHERE id = ?`, every field, from what the screen showed. The same thing as a command: examples/blog/form-overwrites.ts ```ts // Bianca and Carlo open the same record, author misspelled, on shelf A3. const bianca = yield* openDune const carlo = bianca // Bianca fixes the author. Carlo, a minute later, moves the book. yield* listings.commands.saveForm("dune", { title: bianca.title, author: "Frank Herbert", shelf: bianca.shelf }) yield* listings.commands.saveForm("dune", { title: carlo.title, author: carlo.author, shelf: "B1" }) const dune = yield* listings.state("dune") // { title: "Dune", author: "Frank Herbet", shelf: "B1" }: Carlo's form put the typo back ``` Event sourcing didn’t save us. The log has both `ListingSaved` events, so at least someone looking at the history can see what happened, which a table can’t offer. But the state is wrong, because Carlo’s command said “the author is Frank Herbet”: his form sent back every field it showed, including the one he never touched. The problem isn’t storage. It’s that the command says more than the user meant. ## Say what changed Bianca meant “correct the author”. Carlo meant “move it to B1”. Those are the commands: examples/blog/form-overwrites.ts ```ts yield* openDune yield* listings.commands.correctAuthor("dune", { author: "Frank Herbert" }) yield* listings.commands.moveToShelf("dune", { shelf: "B1" }) const dune = yield* listings.state("dune") // { title: "Dune", author: "Frank Herbert", shelf: "B1" }: both edits kept ``` Both edits survive, in either order, because neither command carries a field it doesn’t change. Most lost updates disappear this way, without locks, versions or error messages: two people working on the same record rarely change the same thing. This is what people mean by *task-based UI*: screens built around what users do, “correct author”, “move to shelf”, “withdraw”, rather than around a table’s columns. It costs more than one generic form. It also gives you events that say what happened, `AuthorCorrected` rather than `ListingSaved`, which every read model, report and audit after it will appreciate. ## When two edits really collide Sometimes two people do change the same thing. Bianca retitles the book “Dune (Deluxe Edition)”, and Carlo, from the screen he opened before, retitles it “Dune: 60th Anniversary”. The last one would win, and Carlo would never know he overwrote a decision he hadn’t seen. His screen knows which version it showed. Sending it back as the `expectedVersion` turns the silent overwrite into a question: examples/blog/form-overwrites.ts ```ts const bianca = yield* openDune const carlo = bianca // Both screens show version 1. yield* listings.commands.retitle("dune", { title: "Dune (Deluxe Edition)" }, { expectedVersion: bianca.version }) const refused = yield* Effect.flip( listings.commands.retitle("dune", { title: "Dune: 60th Anniversary" }, { expectedVersion: carlo.version }) ) // refused: WrongExpectedVersion { expectedVersion: 1, actualVersion: 2 } // Carlo's screen reloads, shows Bianca's title, and he decides what to do. ``` Bianca’s command matched: the record was at version 1, as her screen said. Carlo’s didn’t: the record had moved to version 2. The command failed before its handler ran, with both versions in the error, and the screen can reload and show Carlo what changed. Then he decides, knowing. `WrongExpectedVersion` is never retried. That is the difference from the other version check in effect-cqrs: when two *processes* append to the same aggregate at once, the loser reloads the state and runs the handler again, and the rules decide on the new state. That’s safe because the handler sees what changed. A user doesn’t, so deciding again for them would be the same silent overwrite. ## When to send a version Not always. An expected version refuses any change in between, even one that didn’t touch what the user is editing. With it on every command, Carlo’s shelf move would have been refused because Bianca corrected the author, and he would have had to reload for nothing. * **Leave it out** for commands whose result doesn’t depend on what the user saw: “move to shelf B1” means the same whatever the title is. The aggregate’s rules still run on the current state. * **Send it** when the user’s decision is based on what they read: a new title chosen while looking at the old one, a price set while looking at the cost, an approval of a document someone may have edited since. ## Rules of thumb * **A command carries what the user changed, not what the screen showed.** Most lost updates are a form that sends back fields nobody touched. * **Read models keep the version they show.** The `version` in an event’s metadata is free, and the screen will need it. * **Send `expectedVersion` when the decision depends on what was read.** It turns an overwrite into a conflict the user can see. * **Never retry a conflict for the user.** Reload, show the change, and let them decide. ## Next The [commands guide](/guides/commands/#expected-version) covers expected versions and how they combine with command ids, and [I clicked twice](/blog/i-clicked-twice/) covers the other kind of repeat: the same request arriving twice. # The message that blocks the queue > One event a handler can't read stops everything behind it, at three in the morning. Stop and wait, or set it aside and carry on: when each is right, and how a parked event comes back once the bug is fixed. *A post after the series [CQRS with effect-cqrs](/blog/), on its lending library. November 2026.*  Every queue consumer meets it eventually: the *poison message*. One message the code can’t handle, because of a bug, or data nobody expected. The consumer fails, retries, fails again, and everything behind it waits. In the morning the dashboard shows a lag of six hours and one error, repeated forty thousand times. View models are consumers of the log, so they meet it too. This post shows what effect-cqrs does by default, why, and the other choice. Every snippet runs in [`examples/blog/blocked-queue.test.ts`](https://github.com/davidnussio/effect-cqrs/blob/main/examples/blog/blocked-queue.test.ts). ## Shelf marks The library prints a shelf mark on every spine: the first three letters of the author’s surname. A view model works them out as books are added, and it has a bug, an assumption every developer has made about names: examples/blog/blocked-queue.ts ```ts /** The shelf mark printed on each book's spine: the first letters of the author's surname. */ export class ShelfMarks extends Context.Service>>()( "library/ShelfMarks" ) { static readonly layer = Layer.effect(ShelfMarks, Ref.make>(new Map())) } // The bug: an author is assumed to have a first name and a surname. const surnameOf = (author: string) => author.split(" ")[1]! const shelfMarks = (surname: (author: string) => string, onFailure?: ViewModel.FailurePolicy) => ViewModel.make({ name: "shelf-marks", aggregates: [Book], dispose: ShelfMarks.use((marks) => Ref.set(marks, new Map())), onFailure, handlers: { BookAdded: ({ author }, { aggregateId }) => ShelfMarks.use((marks) => Ref.update(marks, (all) => new Map(all).set(aggregateId, surname(author).slice(0, 3).toUpperCase())) ) } }) export const ShelfMarksView = shelfMarks(surnameOf) ``` It works for Frank Herbert and William Gibson. Then a librarian adds *The Odyssey*: examples/blog/blocked-queue.ts ```ts yield* books.commands.add("dune", { title: "Dune", author: "Frank Herbert" }) yield* books.commands.add("odyssey", { title: "The Odyssey", author: "Homer" }) yield* books.commands.add("neuromancer", { title: "Neuromancer", author: "William Gibson" }) ``` Homer has one name. `split(" ")[1]` is `undefined`, and `.slice` throws. ## Stop and wait By default, a view model that fails on an event stops there and tries again: after 100 milliseconds, then twice as long each time, up to every 30 seconds. It never skips the event and never handles a later one first: examples/blog/blocked-queue.ts ```ts const status = yield* waitFor(statusOf("shelf-marks"), (status) => Option.isSome(status?.lastError ?? Option.none())) // { name: "shelf-marks", position: 1, lastError: Some(ViewModelError { position: 2, cause: TypeError }) } const marks = yield* ShelfMarks.use(Ref.get) // Map { "dune" => "HER" }: nothing for Homer, and nothing for Gibson either const catalog = yield* Catalog.use(Ref.get) // all three books: the catalog has no bug, and doesn't wait for the shelf marks ``` The shelf marks are stuck at position 1, after *Dune*. *Neuromancer* has no shelf mark either, although nothing is wrong with it: it is behind Homer. The catalog, a different view model over the same events, has all three books. Each view model reads the log on its own, from its own checkpoint, so one that is stuck holds up no one else. That’s the default because it is the safe one. A view model that is stalled is *visibly* wrong: `status` shows where it stopped and why, a health check can alert on it, and the moment the bug is fixed and deployed it carries on from Homer, in order, and is right again. A view model that skipped an event would be wrong in a way nothing shows, for as long as it lives. For an account balance, a stock level, or anything where every event changes a total, missing one is worse than being late. ## Or set it aside Shelf marks are not a balance. A book without a label for a day is a small problem. Every book after it without a label is a big one. For a read model like this, a gap is better than a stop, and `park` says so: examples/blog/blocked-queue.ts ```ts export const ParkingShelfMarksView = shelfMarks(surnameOf, ViewModel.park({ attempts: 3 })) ``` After three failed attempts, the view model records the event in `ParkedEvents` and moves on: examples/blog/blocked-queue.ts ```ts yield* waitFor(viewModels.parked("shelf-marks"), (parked) => parked.length > 0) yield* viewModels.caughtUp() const marks = yield* ShelfMarks.use(Ref.get) // Map { "dune" => "HER", "neuromancer" => "GIB" } const parked = yield* viewModels.parked("shelf-marks") // [{ viewModel: "shelf-marks", position: 2, eventId: "book/odyssey@1", error: "…TypeError…", attempts: 3, … }] ``` Gibson has his shelf mark. Homer is parked: the event’s position and id, the error, and how many times it was tried. That list is the to-do list for the morning, and it’s in one place, not spread over forty thousand log lines. ## Bring it back The fix takes the last word as the surname, so a single name is all surname. Deploy it, and hand the parked event to the fixed handler: examples/blog/blocked-queue.ts ```ts // The fix: the surname is the last word, and a single name is all surname. const lastWord = (author: string) => author.split(" ").at(-1) ?? author export const FixedShelfMarksView = shelfMarks(lastWord, ViewModel.park({ attempts: 3 })) ``` examples/blog/blocked-queue.ts ```ts const result = yield* viewModels.retryParked({ viewModel: "shelf-marks" }) // { succeeded: [2], failed: [] } const marks = yield* ShelfMarks.use(Ref.get) // Map { "dune" => "HER", "neuromancer" => "GIB", "odyssey" => "HOM" } ``` `retryParked` handles the parked events again, in order of position. Those that succeed leave the list. Those that fail again stay, with the new error, so a fix that didn’t work is as visible as the bug was. ## What parking costs * **Order.** Homer’s shelf mark was written after Gibson’s. For shelf marks that doesn’t matter: each book has its own. For a read model where events of the same thing must be applied in order, a later event can be handled before the parked one, and the result can be wrong in a way the retry won’t undo. Park only where events don’t depend on each other. * **Replays skip parked events** until they are retried, since the handler would fail on them again. * **Someone has to look.** A parked event is a bug report the system filed for you. Keep `ParkedEvents` in the database with `ParkedEvents.layerSql()` rather than in memory, watch the `parked` count in `status`, and alert on it. ## Rules of thumb * **Stall by default.** A view model that is visibly behind is better than one that is silently wrong. * **Park where a gap is better than a stop:** labels, search indexes, notifications, statistics. Never where events add up to a total. * **Read models are independent.** One stuck view model holds up nothing but itself, which is a reason to split read models by purpose rather than build one that does everything. * **Fix, deploy, retry.** The parked event is still in the log, unchanged, waiting for a handler that can read it. ## Next The [view models guide](/guides/view-models/#failing-handlers) covers failures, parking and retries in full, and [Fix the code, not the data](/blog/fix-the-code-not-the-data/) covers the bug that doesn’t fail: a handler that writes the wrong thing. # The saga you skipped > Borrowing a book without a saga, in the two ways that look right: one aggregate that records another's events, and one that calls another's commands. What breaks, and what the saga does instead. *A post after the series [CQRS with effect-cqrs](/blog/), on its lending library. November 2026.*  Sagas are the part of CQRS that teams skip most often. Aggregates and read models make sense on day one, while a saga looks like ceremony: a workflow, a trigger, an engine, just to lend a book? It is so much shorter to let the member’s command do everything at once. This post takes that shortcut, twice, on the library of the series. Both versions compile, pass a happy-path test, and look fine in a demo. Then we look at what they do to the data, and at what the saga of [part 4](/blog/sagas/) does differently. Every snippet runs in [`examples/blog/without-saga.test.ts`](https://github.com/davidnussio/effect-cqrs/blob/main/examples/blog/without-saga.test.ts). The rules are the ones from the series: **a member may have at most three books at a time, and a copy can only be with one member.** The first rule is about the member, the second about the book. ## Mistake 1: one aggregate records everything Borrowing changes the member and the book, so the member’s command records both changes. The member’s events get a `BookLent`, and `borrow` returns two events in one go: examples/blog/without-saga.ts ```ts const GreedyMemberEvent = Schema.TaggedUnion({ MemberJoined: { name: Schema.NonEmptyString }, LoanStarted: { loanId: Schema.String, bookId: Schema.String }, // The book's event, recorded in the member's stream. BookLent: { loanId: Schema.String, bookId: Schema.String, dueOn: Schema.DateTimeUtc } }) export const GreedyMember = Aggregate.make()({ name: "member", events: GreedyMemberEvent, initial: { joined: false, loans: [] }, projection: { MemberJoined: (state) => ({ ...state, joined: true }), LoanStarted: (state, { loanId }) => ({ ...state, loans: [...state.loans, loanId] }), BookLent: (state) => state } }) .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 GreedyMemberEvent.cases.MemberJoined.make({ name }) }) }) .command("borrow", { 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.length >= loanLimit) { return yield* new LoanLimitReached({ memberId: aggregateId, limit: loanLimit }) } const dueOn = DateTime.add(yield* DateTime.now, loanPeriod) // Two events in one go: the member's, and the one the book should have decided on. return [ GreedyMemberEvent.cases.LoanStarted.make({ loanId, bookId }), GreedyMemberEvent.cases.BookLent.make({ loanId, bookId, dueOn }) ] }) }) ``` It reads well. One command, one atomic append, both facts recorded together. Here is a morning at the library: examples/blog/without-saga.ts ```ts const members = yield* AggregateService.make(GreedyMember) 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.borrow("alice", { loanId: "loan-1", bookId: "book-1" }) yield* members.commands.borrow("bob", { loanId: "loan-2", bookId: "book-1" }) // Both succeed: two loans of one copy. const book = yield* books.state("book-1") // { status: "available", loan: null, … }: the book never heard of either ``` Alice and Bob both have the only copy of *Dune*. Nothing failed, so nothing told us. Look at where the rule “one member per copy” is checked in `borrow`: nowhere, and it can’t be. The handler decides from the *member’s* state, which holds the member’s loans and nothing about the book. The member wrote a fact about the book, `BookLent`, without asking the book. And the event is in the wrong place. `BookLent` sits in Alice’s stream and in Bob’s, not in `book-1`’s, so the book’s own state says it is on the shelf. Every read model that follows the book’s events agrees with it. The library now has two versions of the truth, and the one in the book’s stream is wrong. The usual patch is to read the book first, with `books.state(bookId)`, and refuse if it is lent. That only moves the problem: the book’s stream still never gets the `BookLent`, so the read always says available. Even if it did, reading and then writing elsewhere is check-then-act: between the read and the member’s append, another member can borrow the same copy, and no version check stops it, because the append goes to a different stream. ## Mistake 2: one aggregate calls the other Fine, the book must decide for itself. So the member’s command calls the book’s command, and records its own event once the book has said yes: examples/blog/without-saga.ts ```ts const CallingMemberEvent = Schema.TaggedUnion({ MemberJoined: { name: Schema.NonEmptyString }, LoanStarted: { loanId: Schema.String, bookId: Schema.String } }) export const CallingMember = Aggregate.make()({ name: "member", events: CallingMemberEvent, initial: { joined: false, loans: [] }, projection: { MemberJoined: (state) => ({ ...state, joined: true }), LoanStarted: (state, { loanId }) => ({ ...state, loans: [...state.loans, 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 CallingMemberEvent.cases.MemberJoined.make({ name }) }) }) .command("borrow", { 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.length >= loanLimit) { return yield* new LoanLimitReached({ memberId: aggregateId, limit: loanLimit }) } // Now the book decides, through its own command, from inside the member's decision. const books = yield* Books yield* books.commands.lend(bookId, { loanId, memberId: aggregateId }) return CallingMemberEvent.cases.LoanStarted.make({ loanId, bookId }) }) }) ``` The book’s events are back in the book’s stream, and its rule runs: examples/blog/without-saga.ts ```ts const members = yield* AggregateService.make(CallingMember) 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.borrow("alice", { loanId: "loan-1", bookId: "book-1" }) const refused = yield* Effect.flip(members.commands.borrow("bob", { loanId: "loan-2", bookId: "book-1" })) // refused._tag: "BookAlreadyLent". So far, so good. ``` This version passes every test you’d think of writing first. The problem is in a test you wouldn’t: one where something goes wrong *between* the two writes. `borrow` now writes to two streams: the book’s, inside `lend`, then the member’s, when the handler returns. Those are two appends. No transaction spans them, and in general none can: the two streams may live in different tables, databases or services. So let’s lose the connection to the database at the worst moment, with a store that fails one append: examples/blog/without-saga.ts ```ts /** * The store in memory, except that the first append with an event of this type fails, * as it would if the connection to the database dropped at that moment. */ export const failingOnce = (eventType: string) => Layer.effect( EventStore.EventStore, Effect.gen(function*() { const store = yield* EventStore.EventStore const failed = yield* Ref.make(false) return EventStore.EventStore.of({ ...store, append: (input) => Effect.gen(function*() { if (input.events.some(({ type }) => type === eventType) && !(yield* Ref.getAndSet(failed, true))) { return yield* new EventStore.EventStoreError({ operation: "append", cause: new Error("Connection lost") }) } return yield* store.append(input) }) }) }) ).pipe(Layer.provide(EventStore.layerMemory)) ``` Same code, same request, on that store: examples/blog/without-saga.ts ```ts yield* books.commands.add("book-2", { title: "Neuromancer", author: "William Gibson" }) yield* members.commands.join("carol", { name: "Carol Verdi" }) // The member's append fails, after the book's went through. const failure = yield* Effect.flip(members.commands.borrow("carol", { loanId: "loan-3", bookId: "book-2" })) // failure._tag: "EventStoreError" const book = yield* books.state("book-2") // { status: "lent", loan: { loanId: "loan-3", memberId: "carol", … }, … } const carol = yield* members.state("carol") // { joined: true, loans: [] } ``` The book is lent to Carol, for loan `loan-3`. Carol has no loan `loan-3`. The command failed, so the client tells Carol to try again later, and if she doesn’t, the book stays lent to a loan that doesn’t exist. Nothing in the system knows that a loan was half done, so nothing will come back to finish it or undo it. A lost connection is only one way to get there. Any of these does the same: * **The process stops** between the two appends: a crash, a deploy, a scaled-down pod. * **The member’s append loses a race.** Another request for Carol got there first, so the service reloads her state and runs the handler again, which calls `lend` again. Here `lend` happens to be idempotent, so the second call records nothing. If the handler had instead picked “the first free copy”, the retry could lend a second one. * **The member’s rules change their mind on the retry.** After the reload Carol has three books, so the retry fails with `LoanLimitReached`, after the book was lent. There are quieter costs too. A handler is meant to only decide, because it can run more than once. The member module now depends on the books service, and an aggregate whose handler calls another aggregate’s command can deadlock if that one ever calls back, since commands on one aggregate take turns. And when someone asks why a book is lent, the member’s stream has no answer. ## The same loan, with a saga The saga starts from a different question. Not “how do I write both changes at once?”, which has no good answer, but “how do I make sure the process *finishes*?” The member only ever records its own decision. `requestLoan` checks the member’s rule and appends `LoanRequested` to the member’s stream, one atomic write to one aggregate. From that moment, the intention to borrow is a fact in the log, and the rest of the process can always be finished from it. Then the saga carries the data from one aggregate to the next. This is the workflow of [part 4](/blog/sagas/): examples/blog/library.ts ```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 }) }) } })) ``` Look at what it does with the data. It reads the member id, the book id and the loan id from the member’s event, and hands them to the *book’s* command. The book decides with its own state and its own rule. The answer goes back to the *member*, as one of the member’s own commands, `confirmLoan` or `cancelLoan`. Every change is still made by a command on the aggregate that owns the data. The saga owns no data, and changes none directly: it puts together what several aggregates know, and asks each one to do its part. Now the same failure, the same store, on the library of the series. This time the append that fails is the member’s `LoanConfirmed`, the saga’s last step: examples/blog/without-saga.ts ```ts yield* books.commands.add("book-2", { title: "Neuromancer", author: "William Gibson" }) yield* members.commands.join("carol", { name: "Carol Verdi" }) // One write, to one aggregate: the member records what she asked for. yield* members.commands.requestLoan("carol", { loanId: "loan-3", bookId: "book-2" }) // The saga lends the book, fails to confirm the loan, tries again, and confirms it. const outcome = yield* settled("carol", "loan-3") // Option.some("LoanConfirmed") const book = yield* books.state("book-2") // { status: "lent", loan: { loanId: "loan-3", memberId: "carol", … }, … } const carol = yield* members.state("carol") // { …, loans: { "loan-3": { bookId: "book-2", status: "active" } } } ``` The loan settles: the book is lent to Carol, and Carol has the loan. Here is why: * **Each step is one command on one aggregate**, with one atomic append. A step either happened or it didn’t. There is no half of a step. * **A step that fails is tried again.** `Saga.command` retries a lost race or a store error a few times, with a backoff that starts at 50 milliseconds, so a dropped connection costs a short wait. * **Finished steps are not done again.** The workflow engine records the outcome of every step. A run that restarts after a crash skips the book’s `lend`, which already happened, and goes on with the confirmation. With a durable engine, that holds across restarts of the whole process. * **The commands are idempotent**, so a step that ran but whose outcome wasn’t recorded can run again harmlessly. * **The state in between is explicit.** Until the saga confirms or cancels it, Carol’s loan is `requested`. It holds one of her three places, so she can’t over-borrow while the book decides, and it shows that something is in progress. If a failure outlasts the retries, the run stops there, and the loan stays visibly `requested` instead of silently missing. Here are the three versions side by side: | | One aggregate records both | One aggregate calls the other | Saga | | ----------------------------------- | -------------------------- | ----------------------------- | ---------------------------------- | | The book’s rule is checked | No | Yes | Yes | | The book’s events are in its stream | No | Yes | Yes | | A failure between two writes | — | Leaves them disagreeing | Retried, then resumed | | Where the process is recorded | Nowhere | Nowhere | The member’s events and the engine | ## Rules of thumb * **One command, one aggregate, one append.** When a command needs to change a second aggregate, the first one records what it decided, and something else, a saga, asks the second. * **An event names its own aggregate’s facts.** A member’s event that says “the book was lent” is a fact the member had no right to decide. * **A decision that needs another aggregate’s state is a process, not a command.** Model the steps in between, such as `requested`, in the state, because they are real: the book really hasn’t answered yet. * **If you can’t live with the step in between, check the boundary.** A rule that must hold instantly across two aggregates may mean they are one aggregate. That is a design decision, and sometimes the right one. But “every loan in the library” is too big to be one aggregate, so here the saga wins. What about a view model whose handler runs the book’s command? It gets closer: the member records its intention, and the handler sees every event at least once. But it keeps no record of which steps ran or which branch it took, so it can’t resume a half-done process or replay one the same way. The saga’s trigger *is* a view model, plus the workflow engine that keeps that record. ## Next The series shows the whole saga, its trigger and its wiring in [part 4](/blog/sagas/), and the [sagas guide](/guides/sagas/) covers compensations, timeouts and durable engines. # The wrong event > A bug in a command stores wrong due dates for a week. A replay can't fix it, and editing the log would hide it, so the fix records a correction: one more event, with its reason. *A post after the series [CQRS with effect-cqrs](/blog/), on its lending library. November 2026.*  [Fix the code, not the data](/blog/fix-the-code-not-the-data/) ended on a warning: a replay rebuilds read models, and it can’t change what the commands decided. A bug in a view model is fixed by its own handlers. A bug in a command writes wrong facts into the log, and the log is never edited. This post is about that bug: what you do when the log itself says something false. Every snippet runs in [`examples/blog/wrong-event.test.ts`](https://github.com/davidnussio/effect-cqrs/blob/main/examples/blog/wrong-event.test.ts). ## A week of short loans A release changes how the loan period is written, and the new line says hours where it meant days. The library’s `lend` command is the one from the series, with the period made a parameter so the example can run both versions: examples/blog/wrong-event.ts ```ts /** The book of the series, with the loan period as a parameter, so two deploys can differ. */ const makeBook = (loanPeriod: Partial) => Aggregate.make()({ name: "book", events: BookEvent, initial: { status: "unknown", loan: null }, projection: { BookAdded: (state) => ({ ...state, status: "available" }), BookLent: (_, { dueOn, loanId, memberId }) => ({ status: "lent", loan: { loanId, memberId, dueOn } }), BookReturned: () => ({ status: "available", loan: null }), DueDateCorrected: (state, { dueOn }) => state.loan === null ? state : { ...state, loan: { ...state.loan, dueOn } } } }) .command("add", { payload: Schema.Struct({ title: Schema.NonEmptyString, author: Schema.NonEmptyString }), handler: Effect.fnUntraced(function*(state, payload, { aggregateId }) { if (state.status !== "unknown") return yield* new BookAlreadyAdded({ bookId: aggregateId }) return BookEvent.cases.BookAdded.make(payload) }) }) .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 }) 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 }) }) }) .command("correctDueDate", { payload: Schema.Struct({ loanId: Schema.String, dueOn: Schema.DateTimeUtc, reason: Schema.String }), handler: Effect.fnUntraced(function*(state, { dueOn, loanId, reason }, { aggregateId }) { if (state.loan?.loanId !== loanId) return yield* new LoanNotFound({ bookId: aggregateId, loanId }) // Correcting it again to the same date records nothing, so the script can run twice. if (DateTime.Equivalence(state.loan.dueOn, dueOn)) return [] return BookEvent.cases.DueDateCorrected.make({ loanId, dueOn, reason }) }) }) // The bug: hours where days were meant. export const BookWithTheBug = makeBook({ hours: 21 }) // The fix. export const Book = makeBook({ days: 21 }) ``` For a week, every `BookLent` says the book is due 21 hours after it went out. That is not a read model getting it wrong. The command decided it, the event recorded it, and the reminder job reads it from a read model that does exactly what it should: examples/blog/wrong-event.ts ```ts export interface OpenLoan { readonly bookId: string readonly memberId: string readonly lentAt: DateTime.Utc readonly dueOn: DateTime.Utc } /** The loans still open, by loan id, with when each began and when it is due. */ export class OpenLoans extends Context.Service>>()( "library/OpenLoans" ) { static readonly layer = Layer.effect(OpenLoans, Ref.make>(new Map())) } const change = (update: (loans: Map) => void) => OpenLoans.use((open) => Ref.update(open, (loans) => { const next = new Map(loans) update(next) return next }) ) export const OpenLoansView = ViewModel.make({ name: "open-loans", aggregates: [Book], dispose: OpenLoans.use((open) => Ref.set(open, new Map())), handlers: { BookLent: ({ dueOn, loanId, memberId }, { aggregateId, timestamp }) => change((loans) => loans.set(loanId, { bookId: aggregateId, memberId, lentAt: timestamp, dueOn })), BookReturned: ({ loanId }) => change((loans) => loans.delete(loanId)), DueDateCorrected: ({ dueOn, loanId }) => change((loans) => { const loan = loans.get(loanId) if (loan !== undefined) loans.set(loanId, { ...loan, dueOn }) }) } }) /** The loans that are overdue at a given time, as the reminder job asks for them. */ export const overdueAt = (at: DateTime.Utc) => OpenLoans.use((open) => Ref.get(open).pipe( Effect.map((loans) => [...loans].filter(([, { dueOn }]) => DateTime.isGreaterThan(at, dueOn)).map(([loanId]) => loanId) ) ) ) ``` Alice borrows *Dune* during the bad week. The fix is deployed, and Bob borrows *Neuromancer* after it: examples/blog/wrong-event.ts ```ts const withTheBug = yield* AggregateService.make(BookWithTheBug) yield* withTheBug.commands.add("book-1", { title: "Dune", author: "Frank Herbert" }) yield* withTheBug.commands.lend("book-1", { loanId: "loan-1", memberId: "alice" }) ``` examples/blog/wrong-event.ts ```ts const books = yield* AggregateService.make(Book) yield* books.commands.add("book-2", { title: "Neuromancer", author: "William Gibson" }) const { position } = yield* books.commands.lend("book-2", { loanId: "loan-2", memberId: "bob" }) yield* viewModels.caughtUp(position) const inTwoDays = DateTime.add(yield* DateTime.now, { days: 2 }) const beforeCorrection = yield* overdueAt(inTwoDays) // ["loan-1"]: Alice's book is due 21 hours after she took it ``` Bob’s loan is right. Alice’s is still due the day after she took it, and two days later the reminder job tells her she is late. Fixing the code stopped new damage. It didn’t touch the event that says Alice’s book was due on Tuesday, and no replay will, because a replay reads that event as faithfully as the first time. ## Three things not to do **Don’t edit the event.** It is tempting: one `UPDATE` on the events table and the date is right. But the log is what everything else trusts to never change. Read models have already handled the old value and won’t see the new one, since their checkpoints are past it. An outbox may have published it to another service. A test or an audit that compares the log with what happened now finds the past rewritten, with no trace of who did it or why. **Don’t hide it in an upcaster.** An upcaster reads events of an old schema version as the current shape, as in [part 5](/blog/tests-schema-changes-and-production/). The week’s events have the same version as every other `BookLent`, so an upcaster can’t even tell them apart. And if you found a way, the fix would live in the read path forever, invisible in the log, changing what the aggregate decided after the fact. An upcaster is for a change of *shape*, true of every old event. This is a change of *value*, true of some. **Don’t patch the read models.** Correcting the due date in the reminder job’s table fixes the reminders until the next replay brings the 21 hours back. And the book itself still believes it: its own state, folded from the same event, has the wrong `dueOn`, so when Alice returns it on day ten, `returnBook` marks her late. ## Record the correction The log is a record of what happened, and something did happen: the library found a mistake and fixed it. That is a fact like any other, so it gets an event: examples/blog/wrong-event.ts ```ts 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 correction is a fact of its own: what the due date is now, and why it changed. DueDateCorrected: { loanId: Schema.String, dueOn: Schema.DateTimeUtc, reason: Schema.String } }) ``` The book handles it like any other event. Its projection replaces the due date, and the `correctDueDate` command in the aggregate above checks that the loan is the one the book is on, and records nothing if the date is already right, so the script that sends it can run twice. The script finds the loans to correct from the read model: every open loan due sooner than 21 days after it began came from the bug. Then it sends one command per loan, with the reason: examples/blog/wrong-event.ts ```ts // Every open loan that is due sooner than 21 days after it began came from the bug. const affected = [...(yield* OpenLoans.use(Ref.get))].filter(([, { dueOn, lentAt }]) => DateTime.isLessThan(dueOn, DateTime.add(lentAt, { days: 21 })) ) for (const [loanId, { bookId, lentAt }] of affected) { yield* books.commands.correctDueDate(bookId, { loanId, dueOn: DateTime.add(lentAt, { days: 21 }), reason: "The loan period was 21 hours instead of 21 days" }).pipe(Metadata.withMetadata({ ranBy: "scripts/fix-due-dates.ts" })) } yield* viewModels.caughtUp() const afterCorrection = yield* overdueAt(inTwoDays) // [] const history = yield* books.history("book-1", { payload: true }) // BookAdded, BookLent, DueDateCorrected: the mistake and the correction, whose // metadata says { ranBy: "scripts/fix-due-dates.ts" } ``` Everything that follows the log picks the correction up in order. The reminder job’s read model handles `DueDateCorrected` and moves the date. The book’s own state has the new date, so when Alice brings *Dune* back on day ten, the test shows `returnBook` deciding she is on time. And `history` tells the whole story: lent with the wrong date, then corrected, by which command, with which reason. ## What a table would have done With a `loans` table, the fix is an `UPDATE`: ```sql UPDATE loans SET due_on = lent_on + interval '21 days' WHERE lent_on BETWEEN '2026-11-02' AND '2026-11-09' AND returned = false; ``` The rows are right afterwards. Everything else is gone: that Alice’s row ever said Tuesday, that she got a reminder because of it, that anyone changed it. If she disputes a fine next month, the table says she was never late and has no idea why she was told so. Unless someone built an audit table beforehand, the database now holds a cleaner past than the one that happened. ## Rules of thumb * **A wrong fact is corrected by a new fact.** Name it after what happened, `DueDateCorrected` rather than `BookLentAgain`, and give it a reason. * **Make the correction a command on the aggregate.** It goes through the same rules and the same version check as everything else, and the aggregate’s state learns from it. * **Find what to correct with a read model, fix it with commands.** The query can be thrown away after the script runs. The corrections stay. * **Design for the corrections you will need.** A domain where people make mistakes, which is every domain, needs a way to say “that was wrong”. Events like `DueDateCorrected`, `PaymentReversed` or `AddressAmended` are part of the model, not an embarrassment. * **Some consequences are already out.** Alice got the reminder. A view model or a saga that handles `DueDateCorrected` can send her an apology, which a silent `UPDATE` never could. ## Next [Fix the code, not the data](/blog/fix-the-code-not-the-data/) covers the other half: bugs in read models, which a replay repairs. The [aggregates guide](/guides/aggregates/) covers commands, projections and history. # Three ways to read > A lent book read from its aggregate, from a read model as it is, and from a read model that waits for the command. Which one is current, when to use each, and how a browser picks one over HTTP. *A post after the series [CQRS with effect-cqrs](/blog/), on its lending library. November 2026.*  Alice borrows *Dune* at the front desk, and the catalog on the screen still shows it on the shelf. She reloads, and it’s gone. Nothing is broken: the catalog is a read model, and read models run a moment behind the commands. “CQRS means eventual consistency” is repeated so often that it sounds like a rule. It isn’t. Delay comes from reading something that is updated *after* the write, and that is one way of reading among several. In effect-cqrs a read is one of three kinds, and each query picks the one it needs. Every snippet below runs in [`examples/blog/reading.test.ts`](https://github.com/davidnussio/effect-cqrs/blob/main/examples/blog/reading.test.ts). ## 1. The aggregate: always current The first way doesn’t use a read model at all. It folds the book’s own events, as [part 2](/blog/your-first-aggregate/#reading) did: examples/blog/reading.ts ```ts yield* books.commands.add("book-1", { title: "Dune", author: "Frank Herbert" }) yield* books.commands.lend("book-1", { loanId: "loan-1", memberId: "alice" }) const book = yield* books.state("book-1") // { status: "lent", …, loan: { memberId: "alice", … } }, every time ``` `state` reads the store, from the state the service remembers onwards, so it sees every event that was appended, by this process or another. It is the same state the next command on the book will be decided on. It can’t be behind, because there is nothing in between to fall behind. The cost is that it answers one question only: the state of *this* book, by its id. It can’t list the available books or find Alice’s loans without loading every book. For the page of one thing, though, such as a book, an order or an account, it is often all you need. That is also the page a user tends to land on right after acting on it. ## 2. A read model as it is: a moment behind The catalog of [part 3](/blog/read-models/) is a view model. It handles events in a fiber of its own, after the command has stored them: examples/blog/reading.ts ```ts const rightNow = (yield* Catalog.use(Ref.get)).get("book-1") // { available: false }, { available: true } or undefined: the catalog may not have // seen the loan, or even the book, yet ``` Right after the command, the view model may or may not have run. The gap is usually milliseconds. A slow handler, a restart or a handler that keeps failing makes it longer, and `viewModels.status` says how far each view model has got. For most reads that doesn’t matter. Nobody can tell whether the catalog they just opened is twenty milliseconds old, and a search, a dashboard or a monthly total is no different. This is the cheapest read there is: no fold, no wait, data already in the shape of the page. ## 3. A read model after the command: your own writes The trouble is only with the person who just acted. Alice brings *Dune* back, and the next screen she sees is the catalog. She knows what she did, and expects to see it. Every command returns the position of its last event in the store. `caughtUp` waits until every view model of the process has handled that position: examples/blog/reading.ts ```ts const { position } = yield* books.commands.returnBook("book-1", {}) yield* viewModels.caughtUp(position) const afterReturn = (yield* Catalog.use(Ref.get)).get("book-1") // { title: "Dune", author: "Frank Herbert", available: true }, every time ``` The catalog now includes the return, and every command stored before it. Commands that others ran after it may be there or not. This is “read your own writes”, not a lock on the library, and it only costs a wait for the one read that needs it. ## Over HTTP In an application, the command and the query are two requests, often from a browser. The position has to make the trip between them. Here is the library as an HTTP API, the book’s commands from [`CommandApi.group`](/guides/integration/#http) next to a catalog endpoint of our own: examples/blog/reading.ts ```ts export const BookState = Schema.Struct({ status: Schema.Literals(["unknown", "available", "lent"]), title: Schema.String, author: Schema.String, loan: Schema.NullOr(Schema.Struct({ loanId: Schema.String, memberId: Schema.String, dueOn: Schema.DateTimeUtc })) }) export const CatalogEntry = Schema.Struct({ bookId: Schema.String, title: Schema.String, author: Schema.String, available: Schema.Boolean }) export class LibraryApi extends HttpApi.make("library") // POST /book/:id/add, /lend and /returnBook, and GET /book/:id, which folds the events .add( CommandApi.group(Book, { state: BookState, errors: [BookAlreadyAdded, BookNotFound, BookAlreadyLent, BookNotLent] }) ) // GET /catalog, from the view model, after the position in Read-After if there is one .add( HttpApiGroup.make("catalog") .add(HttpApiEndpoint.get("list", "/catalog", { success: Schema.Array(CatalogEntry) })) .middleware(CommandApi.ReadAfter) ) {} ``` `CommandApi.ReadAfter` is middleware for the query endpoints. It needs the view models, and the handlers need the catalog: examples/blog/reading.ts ```ts const CatalogHandlers = HttpApiBuilder.group(LibraryApi, "catalog", (handlers) => Effect.gen(function*() { const catalog = yield* Catalog return handlers.handle("list", () => Ref.get(catalog).pipe(Effect.map((entries) => Array.from(entries, ([bookId, entry]) => ({ bookId, ...entry })) ))) })) export const Server = HttpApiBuilder.layer(LibraryApi).pipe( Layer.provide([CommandApi.layer(LibraryApi, "book", Effect.service(Books)), CatalogHandlers]), // The catalog's handlers need the middleware, and the middleware needs the view models. Layer.provide(CommandApi.layerReadAfter({ timeout: "2 seconds" })), Layer.provide(ReadSide) ) ``` Every command answers with its position, in the body and in an `Event-Position` header. A query that sends it back in a `Read-After` header waits until the view models have handled it. Here is a client, the way a browser would call it: examples/blog/reading.ts ```ts const post = (path: string, body: object) => fetch(path, { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify(body) }) await post("/book/book-1/add", { title: "Dune", author: "Frank Herbert" }) const lent = await post("/book/book-1/lend", { loanId: "loan-1", memberId: "alice" }) const position = lent.headers.get("event-position")! // The book's own page reads the aggregate: always current. const book = await (await fetch("/book/book-1")).json() // The catalog page sends the position back, and waits until the catalog has the loan. const catalog = await (await fetch("/catalog", { headers: { "read-after": position } })).json() // [{ bookId: "book-1", title: "Dune", author: "Frank Herbert", available: false }] ``` The client chooses, request by request. The book’s page reads the aggregate, and is always current. The catalog page after Alice’s loan sends the position, and includes the loan. Someone who is just browsing the catalog sends nothing, and gets an answer at once. A few details matter in production: * **A timeout.** A view model that is stalled on a failing handler would hold the request forever. After `timeout` the request is answered with `503` and `NotCaughtUp`, and the client can try again, or read without the header and accept an older answer. * **One process.** `ReadAfter` waits for the view models of the process that serves the request. With several replicas, send the read to one that runs the view models it needs. * **Browsers on another origin** only see `Event-Position` if the server lists it in `Access-Control-Expose-Headers`. ## Why not update the read model with the command? Some CQRS setups update their read tables in the same transaction as the write, so a read is never behind. effect-cqrs doesn’t, on purpose. * **Every command would wait for every read model.** A slow handler would slow down loans, and a failing one would refuse them. A broken catalog should show an old catalog, not stop the library. * **Read models are rebuilt from the log.** A replay, or a new view model that starts from the first event, runs the same handlers on the same events. A projection that also runs inside commands is a second path that has to give exactly the same result. * **They live elsewhere.** A read model in a search index or another database can’t join a transaction with the event store. When a screen must be exact, it reads the aggregate. When it must include the user’s own change, it waits for the position. Either way the cost falls on the one read that needs it, not on every command. ## Which one when | The question | Read | Up to date | | -------------------------------------- | ------------------------------------------------------------ | ---------------- | | What is the state of this book? | `state(id)`, `GET /book/:id` | Always | | The catalog, right after I lent a book | The catalog after `caughtUp(position)`, or with `Read-After` | Includes my loan | | Which books are on the shelf? | The catalog as it is | A moment behind | The read side is not one property of the system, fixed when it was designed. It is a choice each query makes, and the three choices sit side by side in one application. The [view models guide](/guides/view-models/#reading-your-own-writes) and the [HTTP guide](/guides/integration/#reading-your-own-writes) have the reference. # Two pods, two emails > The service is scaled to two replicas, and every member gets every email twice. Background work that reads the log has to run in one place, and take over somewhere else when that place goes away. *A post after the series [CQRS with effect-cqrs](/blog/), on its lending library. November 2026.*  The library service runs fine on one machine. Then traffic grows, or someone wants zero downtime deploys, and the Kubernetes deployment goes from one replica to two. Commands are fine: both replicas append to the same store, and optimistic concurrency settles any race. The next morning a member writes in: *why did I get two emails for one loan?* Everything that reads the log in the background, view models, sagas and outboxes, now runs twice. This post shows what goes wrong and the fix. Every snippet runs in [`examples/blog/two-pods.test.ts`](https://github.com/davidnussio/effect-cqrs/blob/main/examples/blog/two-pods.test.ts). ## Emails through an outbox The library emails a member when a book is lent to them. An email is something that must happen once, outside the system, so it goes through an [outbox](/blog/the-double-write/), which reads the log from its checkpoint and sends what it finds: examples/blog/two-pods.ts ```ts export interface Email { readonly to: string readonly subject: string } /** The mail service both replicas send through. It keeps what it sent. */ export class Mailer extends Context.Service readonly sent: Effect.Effect> }>()("library/Mailer") { static readonly layer = Layer.effect( Mailer, Effect.gen(function*() { const sent = yield* Ref.make>([]) return Mailer.of({ send: (email) => Ref.update(sent, (all) => [...all, email]), sent: Ref.get(sent) }) }) ) } // Members are emailed when a book is lent to them. export const Notifications = Outbox.make({ name: "notifications", aggregates: [Book], messages: { BookLent: ({ memberId }, { aggregateId }) => ({ type: "email", payload: { to: memberId, subject: `You borrowed ${aggregateId}` } }) }, publish: (messages) => Mailer.use((mailer) => Effect.forEach(messages, ({ payload }) => mailer.send(payload as Email), { discard: true })) }) ``` Each replica runs the outbox, against what they share: the event store, the checkpoints and the mail service. examples/blog/two-pods.ts ```ts // What each replica runs: the outbox, on the shared store, checkpoints and mailer. const replica = Layer.fresh(Outbox.layer([Notifications])) // What the replicas share: in production, the database and the mail service. export const Shared = Layer.mergeAll(EventStore.layerMemory, Checkpoints.layerMemory, Mailer.layer) ``` ## Two replicas examples/blog/two-pods.ts ```ts const first = yield* Scope.make() const second = yield* Scope.make() const replicaA = Context.get(yield* Layer.buildWithScope(replica, first), Outbox.Outboxes) const replicaB = Context.get(yield* Layer.buildWithScope(replica, second), Outbox.Outboxes) yield* books.commands.add("book-1", { title: "Dune", author: "Frank Herbert" }) const lent = yield* books.commands.lend("book-1", { loanId: "loan-1", memberId: "alice" }) yield* replicaA.published(lent.position) yield* replicaB.published(lent.position) const afterOneLoan = yield* mailer.sent ``` With nothing else, `afterOneLoan` holds Alice’s email twice. Each replica read the `BookLent`, from the same checkpoint, and each sent it. Both did their job. Neither knew the other existed. It’s not only emails. Every view model in both replicas handles every event: * **Read models in a shared database** get every update twice. Setting a value twice is harmless, but adding to a count, appending to a list or inserting a row is not. * **Checkpoints race.** Both replicas save their position under the same name, and the slower one can move it back, so events are handled again after a restart. * **Sagas start twice.** The trigger is a view model. The workflow engine’s idempotency key catches the duplicate start, if it is durable and shared, but every trigger still runs on both. Read models kept in each replica’s memory are the exception: each replica *should* build its own. That’s why running everything everywhere is the default. ## One holder, one standby The fix is to decide, for each view model, which replica runs it. effect-cqrs calls this a *lease*: `ViewModel.layer` takes one per view model before running it, from the `Leases` service in context. The replica holding the lease runs the view model, and the others wait as standbys. In production that is `Leases.layerPostgres()`, which holds each lease as a PostgreSQL advisory lock, on the same database as the events: ```ts const Live = Layer.mergeAll(ViewModel.layer([Catalog]), Outbox.layer([Notifications])).pipe( Layer.provideMerge(Layer.mergeAll( PostgresEventStore.layer(), PostgresEventStore.layerCheckpoints(), Leases.layerPostgres() )) ) ``` The example runs both replicas in one test, so it uses leases in memory that work the same way: the first to ask holds the lease, and the next one in line gets it when the holder goes away. examples/blog/two-pods.ts ```ts /** * Leases in memory that two replicas in one process can share: the first to ask holds a * lease, and when its scope closes the next one in line gets it. `Leases.layerPostgres()` * does the same across processes, with PostgreSQL advisory locks. */ export const sharedLeases = Layer.sync(Leases.Leases, () => { const lines = new Map>>() return Leases.Leases.of({ acquire: (name) => Effect.gen(function*() { const held = yield* SubscriptionRef.make(false) const line = lines.get(name) ?? [] lines.set(name, [...line, held]) if (line.length === 0) yield* SubscriptionRef.set(held, true) yield* Effect.addFinalizer(() => Effect.gen(function*() { const rest = (lines.get(name) ?? []).filter((ref) => ref !== held) lines.set(name, rest) const next = rest[0] if (next !== undefined) yield* SubscriptionRef.set(next, true) }) ) return { held: SubscriptionRef.get(held), changes: SubscriptionRef.changes(held) } }) }) }) ``` With them, the same two replicas send Alice one email. Replica A holds the lease and runs the outbox. Replica B’s outbox waits, and its `status` says `holdsLease: false`. ## The holder goes away Replicas don’t live forever. A deploy replaces them, a node is drained, a process runs out of memory. When the holder stops, its lease is released, and the standby takes over from the checkpoint the holder saved: examples/blog/two-pods.ts ```ts // Replica A is stopped: a deploy, a crash, the node it ran on is drained. yield* Scope.close(first, Exit.void) yield* books.commands.add("book-2", { title: "Emma", author: "Jane Austen" }) const lentToBob = yield* books.commands.lend("book-2", { loanId: "loan-2", memberId: "bob" }) yield* holding(replicaB) yield* replicaB.published(lentToBob.position) const afterFailover = yield* mailer.sent ``` Bob gets his email from replica B. Nothing was sent twice and nothing was skipped, because the checkpoint is shared and only ever moved by the holder. With PostgreSQL, the lock lives on a database connection, so it goes when the process dies or loses its connection. Standbys try to take over every 5 seconds by default. Caution A holder notices it lost its connection at its next check, so for up to one interval the old holder and the new one can both handle the same events. A lease makes duplicates rare, not impossible. Consumers of an outbox still deduplicate on the message id, and handlers that write to a shared database still write idempotently. ## What else changes with replicas * **`caughtUp` waits only for local view models.** On a standby it doesn’t wait for the view models another replica runs, since nothing in this process moves them. Read your own writes where the view model runs, or from the shared read model directly. * **`replay` and `retryParked` run on the holder.** A standby refuses them. * **In an Effect cluster**, `Singleton` from `effect/cluster` does the same job with the cluster’s shard ownership, and needs no leases. ## Rules of thumb * **Commands scale out, background work runs once.** Any number of replicas can take commands. Each view model, saga trigger and outbox runs in one of them at a time. * **Turn leases on before the second replica.** The first duplicate email is the cheapest reminder you will get. * **Read models in memory run everywhere, shared ones run once.** Leases are for view models whose data, or effects, are shared. * **Stay idempotent anyway.** Failover has a window, and at-least-once delivery has always allowed a repeat. ## Next The [view models guide](/guides/view-models/#several-processes) covers leases and standbys in full, and [The double write](/blog/the-double-write/) covers outboxes. # What it looked like last Tuesday > A member was refused a book that is on the shelf today, and asks why. A table only knows today. The log can show the book as it was at that moment, and every event the request caused, across aggregates. *A post after the series [CQRS with effect-cqrs](/blog/), on its lending library. November 2026.*  Monday morning, Bob is at the desk. Last Tuesday he asked for *Dune* on the website, and the library refused. Today *Dune* is on the shelf. Was the website broken? Is he being treated differently? He would like to know, and so would the librarian. Support tickets look like this everywhere: why did the system do *that*, back *then*? This post answers Bob’s from the library’s log. Every snippet runs in [`examples/blog/last-tuesday.test.ts`](https://github.com/davidnussio/effect-cqrs/blob/main/examples/blog/last-tuesday.test.ts), which lets the days pass in between. ## Tuesday This is what happened, with the library and the borrowing saga of [the series](/blog/sagas/): examples/blog/last-tuesday.ts ```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" }) yield* settled("alice", "loan-1") yield* members.commands.requestLoan("bob", { loanId: "loan-2", bookId: "book-1" }) yield* settled("bob", "loan-2") ``` Alice asked first and got the book. Bob asked a moment later, and the saga cancelled his loan. On Friday Alice brought it back. ## What a table can say The `books` table says *Dune* is available, `lent_to` empty. That is true, and it is all the table knows. Was it available last Tuesday at 10:04? It doesn’t say. Why was Bob refused? A request log might have a line, if it is still kept, if it was logged at all, and if someone can find it among a million others. Most support tickets like Bob’s end with “we can’t tell, sorry”, or with a developer reading code and guessing. The log can say more, because it never threw anything away. examples/blog/last-tuesday.ts ```ts const today = yield* books.state("book-1") // { status: "available", loan: null, … }: on the shelf ``` ## Start from Bob Bob’s own history says what happened to his request, with the reason the saga recorded: examples/blog/last-tuesday.ts ```ts const bob = yield* members.history("bob", { payload: true }) const refusal = bob.find(({ type }) => type === "LoanCancelled")! // { type: "LoanCancelled", payload: { loanId: "loan-2", reason: "BookAlreadyLent" }, // timestamp: Tuesday, causationId: "member/bob@2", correlationId: "…" } ``` `LoanCancelled`, reason `BookAlreadyLent`, on Tuesday. Two more fields come with every event: * the **causation id** names what directly caused it: `member/bob@2`, the second event in Bob’s stream, his own `LoanRequested`. The saga that cancelled the loan was started by his request, and said so. * the **correlation id** is shared by everything that followed from the same original request, however many steps and aggregates away. ## The book, back then Was the book really lent? Ask the book, as it was at the moment of the refusal: examples/blog/last-tuesday.ts ```ts // The book as it was when Bob was refused. `finishTime` leaves out events from that // instant on, so take the next millisecond. const backThen = yield* books.state("book-1", { finishTime: DateTime.add(refusal.timestamp, { milliseconds: 1 }) }) // { status: "lent", loan: { loanId: "loan-1", memberId: "alice", dueOn: … }, … } ``` `state` with a `finishTime` folds the book’s events up to that instant, the same fold that gives today’s state, cut short. At the moment Bob was refused, the book was lent to Alice, with its due date. The refusal was right, and the librarian can show why, with the evidence. This isn’t a feature added for support. It falls out of keeping events: a state is a fold of events, and a fold can stop anywhere. Every aggregate can be seen as it was at any instant, for free, from the day the first event was written. ## One request, end to end The correlation id goes further. Gather every event that shares it, from every aggregate: examples/blog/last-tuesday.ts ```ts // Everything one request caused, across aggregates, by its correlation id. const causedBy = (correlationId: string | null) => store.read().pipe( Stream.filter((event) => event.correlationId === correlationId), Stream.map(({ aggregateId, aggregateName, type, version }) => `${aggregateName}/${aggregateId}@${version} ${type}` ), Stream.runCollect ) const bobsRequest = yield* causedBy(refusal.correlationId) // ["member/bob@2 LoanRequested", "member/bob@3 LoanCancelled"] const alicesRequest = yield* causedBy( (yield* members.history("alice")).find(({ type }) => type === "LoanRequested")!.correlationId ) // ["member/alice@2 LoanRequested", "book/book-1@2 BookLent", "member/alice@3 LoanConfirmed"] ``` Bob’s request is two events: asked, cancelled. The book has none, because it refused and a refusal writes nothing. Alice’s request, read the same way, is three events in two aggregates: she asked, the book was lent, her loan was confirmed. The whole saga, in order, from a single id. ## What else it answers The same two tools, a state at an instant and events by correlation, answer questions that are hard or impossible with tables: * **Disputes.** “I returned it on time.” The `BookReturned` has a timestamp and the `late` decision, and the state just before it has the due date. * **Audits.** Who changed what, when, as a result of which request. With [metadata](/blog/the-wrong-event/), also by which user or script. * **Bug reports.** “It said available, then refused me.” The events of that request, and the state each aggregate was in, are the reproduction. * **Reports as of a date.** Fold the events before a date into a report, and it says what the numbers were then, not what they are now with today’s corrections mixed in. ## Rules of thumb * **Record the reason in the event.** `LoanCancelled { reason }` answers Bob in one line. A bare `LoanCancelled` sends someone to read code. * **Keep correlation and causation flowing.** effect-cqrs carries them through commands, view models and sagas. Bring the correlation id in from the edge, the HTTP request or the message, and the trail starts where the user did. * **Answer support from the log, not from logs.** Application logs are for operators, rotate away, and say what the code thought. The event log says what the system decided, and it is kept. ## Next [Sagas](/blog/sagas/) in the series builds the borrowing process this post investigates, and the [aggregates guide](/guides/aggregates/) covers state, history, correlation and causation. # Why CQRS and event sourcing > Part 1 of the series: what goes missing when you store only the current state, event sourcing by hand in forty lines, and what CQRS adds on top. *Part 1 of 5 of [CQRS with effect-cqrs](/blog/). November 2026.*  Most applications keep the current state of things. A library keeps a table of books, and lending one updates a row: ```sql UPDATE books SET lent_to = 'bob' WHERE id = 'book-1'; ``` The row is now right, and everything it said before is gone. Who had the book before Bob? How many times has it been lent this year? Was it brought back late? Nobody asked those questions when the table was designed, so the answers were thrown away one `UPDATE` at a time. You can add an audit table, a `loans` history and a `returned_late` flag, but each one has to be planned before the data it needs comes in, and kept in step with the row by hand. This series is about the other way of doing it: keep what happened, and work out the current state from that. Over five posts we build a small lending library with [effect-cqrs](/), from the first event to a saga, tests and a database: 1. **Why CQRS and event sourcing**, this post: the idea, by hand, without a framework. 2. [Your first aggregate](/blog/your-first-aggregate/): events, commands and typed errors. 3. [Read models you can throw away](/blog/read-models/): view models, checkpoints and replays. 4. [Sagas](/blog/sagas/): processes that span several aggregates. 5. [Tests, schema changes and production](/blog/tests-schema-changes-and-production/): fixtures, upcasting, SQLite and PostgreSQL, and forgetting a person. ## Event sourcing in forty lines Instead of a row, a book is the list of things that happened to it. Each event is a fact in the past tense: examples/blog/by-hand.ts ```ts type BookEvent = | { readonly _tag: "BookAdded"; readonly title: string } | { readonly _tag: "BookLent"; readonly memberId: string } | { readonly _tag: "BookReturned" } ``` The current state is not stored anywhere. It is what you get when you apply the events one after another, starting from an empty book. That is a fold, `reduce` in JavaScript: examples/blog/by-hand.ts ```ts interface Book { readonly title: string readonly lentTo: string | null } const initial: Book = { title: "", lentTo: null } // Applies one event to the state: what happened, folded into what is true now. const evolve = (state: Book, event: BookEvent): Book => { switch (event._tag) { case "BookAdded": return { ...state, title: event.title } case "BookLent": return { ...state, lentTo: event.memberId } case "BookReturned": return { ...state, lentTo: null } } } export const currentState = (events: ReadonlyArray): Book => events.reduce(evolve, initial) ``` Changing a book means deciding on new events. A command looks at the state, checks what the rules allow, and either returns the events that happen next or refuses: examples/blog/by-hand.ts ```ts // Looks at the state and decides what happens next, or refuses. export const lend = (state: Book, memberId: string): ReadonlyArray => { if (state.lentTo !== null) throw new Error(`Already lent to ${state.lentTo}`) return [{ _tag: "BookLent", memberId }] } ``` Put together, the log only ever grows: examples/blog/by-hand.ts ```ts const log: Array = [{ _tag: "BookAdded", title: "Dune" }] log.push(...lend(currentState(log), "alice")) log.push({ _tag: "BookReturned" }) log.push(...lend(currentState(log), "bob")) currentState(log) // { title: "Dune", lentTo: "bob" } timesLent(log) // 2 ``` Two functions carry the whole model: `evolve` turns events into state, and `lend` turns state and a request into events. Neither of them touches a database, a clock or the network, so both are trivial to test. ## What the log gives you The questions from the start of this post are now easy, because the facts they need were never thrown away: examples/blog/by-hand.ts ```ts // A question nobody asked when the events were written, answered from the same log. export const timesLent = (events: ReadonlyArray): number => events.filter((event) => event._tag === "BookLent").length ``` `timesLent` was written after the events. It works on every book ever lent, including the ones lent before anyone thought of counting. That is the main argument for event sourcing: **the log answers questions you haven’t thought of yet.** Some more: * **History for free.** Every change has a time, an order and, as we’ll see, a cause. An audit trail is not a feature you add, it is the storage. * **Time travel.** The state of a book last March is the fold of the events up to last March. * **Debugging by replay.** A bug report comes with the exact sequence of events that led to it, and you can run them again in a test. * **Decisions you can explain.** A refused command can say which earlier event it was refused because of. Why not keep a `loans` table instead? With one, `timesLent` is a `SELECT count(*)`, and just as right. The difference is when the table was made and what it holds. It has the loans recorded since someone created it, in the columns they thought of then, and the next question, about late returns or renewals, needs another table that starts empty. It also says the same thing as `books.lent_to`, so the two are kept in step by hand, and when they disagree nothing tells you which one is right. The log is the one place a fact is written, and every other shape is worked out from it. That doesn’t take the table away: with CQRS, below, a `loans` table becomes a read model filled by replaying the log from the first event, so it holds the loans from before it existed, and you can drop it and build it again when it is wrong, as [Fix the code, not the data](/blog/fix-the-code-not-the-data/) shows. ## Where CQRS comes in The log is a great way to *write*, and a poor way to *read*. “Which books are available right now?” would mean folding every book in the library on every request. “Which books does Alice have?” would mean folding them all and filtering. CQRS, Command Query Responsibility Segregation, splits the two: * **The write side** takes commands. It loads one thing, an *aggregate* such as a book, folds its events, decides, and appends new events. It only cares about the rules, and it only ever needs one aggregate’s events at a time. * **The read side** follows the log and keeps *read models*, each shaped for the queries it serves: a catalog with an `available` flag, a map from member to the books they have, a count of loans per month. They are projections of the log, so you can throw one away and rebuild it, or add a new one that starts from the first event ever written. The two ideas are separate, and each works without the other. Plenty of applications do CQRS on ordinary tables: commands update rows, and queries read a view or a second table, often in the same database, with no queue in between. What CQRS asks for is that the code that changes things and the code that answers questions don’t share a model. effect-cqrs does both because they fit: once the log is the truth, every read model is a projection of it, which you can rebuild, change or add at any time without migrating data. This is the path a command takes in effect-cqrs, and the outline of the next three posts: ## How a command flows 1. ### Command A typed call, or JSON from a transport through `dispatch`. The payload is checked against its schema first. [Commands guide](/guides/commands/) 2. ### Aggregate The service rebuilds the aggregate's state from its events, and the handler decides what happened, or fails with a typed error. [Aggregates guide](/guides/aggregates/) 3. ### Event store The events are appended, but only if nobody wrote to that aggregate since its state was loaded. [Storage guide](/guides/storage/) 4. ### View models Each one reads the new events from its own checkpoint. `caughtUp` waits until your own write is there. [View models guide](/guides/view-models/) The price is that the read side is a moment behind the write side. After a command, a read model may not have seen its events yet. Part 3 shows how to wait for your own write when you need to, and [Three ways to read](/blog/three-ways-to-read/) when you don’t need a read model at all. ## What the forty lines leave out The version above works on an array in memory, for one user at a time. A real application needs more, and each of these is what the rest of the series is about: * **Storage and concurrency.** Two librarians lending the same copy at the same moment must not both succeed. The store has to append only if nobody wrote in between. * **Typed, validated input.** `lend` above trusts its arguments and refuses with a thrown `Error`. Refusals are part of the domain, and the caller should know which ones can come. * **Read models that keep up.** They must follow the log as it grows, pick up where they stopped after a restart, and never miss an event. * **Processes.** Borrowing a book involves the book *and* the member, who may only have three books at a time. One command changes one aggregate, so something has to coordinate the two. * **Change.** Events written today will be read in five years by code that looks different. The log can’t be rewritten, so old events have to be read as new ones. * **Forgetting.** The GDPR lets a person ask to be forgotten, in a log that is never edited. ## When not to use it Event sourcing is not free. You trade a table you can query with SQL for a log plus read models you write yourself, and most reads run a moment behind the writes. Events are a contract with your own future, so their shape needs more care than a column does. It pays off where the history *is* the domain, such as loans, accounts, orders, bookings or anything audited, and where the rules are rich enough that “what happened” matters more than “what it looks like now”. A settings page or a list of tags is happier as a table. Nothing stops you from mixing the two in one application. ## Next In [part 2](/blog/your-first-aggregate/) we write the book as an aggregate with effect-cqrs: the same `evolve` and `decide`, with schemas, typed errors and a store that handles concurrent writers. # Your first aggregate > Part 2 of the series: a book as an aggregate, with events as schemas, commands that decide, typed errors, and a store that handles concurrent writers. *Part 2 of 5 of [CQRS with effect-cqrs](/blog/). November 2026.*  In [part 1](/blog/why-cqrs-and-event-sourcing/) 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](https://effect.website), so commands are effects with typed errors, and storage, clocks and configuration are services you provide. It needs Node 24 or later: ```bash pnpm add effect-cqrs effect ``` All the code of this series lives in [`examples/blog`](https://github.com/davidnussio/effect-cqrs/tree/main/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: examples/blog/library.ts ```ts 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: examples/blog/library.ts ```ts export class BookAlreadyAdded extends Schema.TaggedError()("BookAlreadyAdded", { bookId: Schema.String }) {} export class BookNotFound extends Schema.TaggedError()("BookNotFound", { bookId: Schema.String }) {} export class BookAlreadyLent extends Schema.TaggedError()("BookAlreadyLent", { bookId: Schema.String }) {} export class BookNotLent extends Schema.TaggedError()("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: examples/blog/library.ts ```ts /** 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()({ 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()`, 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: examples/blog/library.ts ```ts export class Books extends Context.Service>()("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: examples/blog/library.ts ```ts 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`: 1. The payload is checked against the command’s schema. 2. 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. 3. 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. 4. 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: examples/blog/library.ts ```ts 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: examples/blog/library.ts ```ts // 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: examples/blog/library.ts ```ts 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: ```ts 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](/blog/read-models/). The [commands guide](/guides/commands/) 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. # Limitations > What the port leaves out. effect-cqrs is a trial port, not a maintained library. These are the gaps that matter before using it for anything real. * **No broker.** New events wake view models and subscribers in every process, through PostgreSQL notifications or by polling, but there is no transport like the Moleculer broker, so `dispatch` is the entry point to wire into your own HTTP or RPC layer. * **SQLite, PostgreSQL and memory only.** There are no adapters for MongoDB or MySQL. * **Snapshots are opt-in.** Without them, recent state is only kept in memory, so after a restart each aggregate is replayed the first time it is used. A changed state needs a new snapshot `version` by hand. * **Event upcasts are pure.** An upcast can’t read a service or fail with its own error, and an event can only be read in a version its aggregate declares. * **No code generator.** There is no equivalent of `cqrs generate`. * **No inline projections.** A view model never runs in the transaction that appends the events, so a read model is always a moment behind its command. Read one aggregate with `state`, which is always current, or wait for a command’s position with `caughtUp` or `CommandApi.ReadAfter`. * **Slow handlers don’t slow writers.** There is no backpressure, so a view model with a slow handler falls behind and catches up when writes stop. * **One thread.** Concurrent writers share commits, but the work itself doesn’t spread over cores, and synchronous SQLite calls block the event loop while they run. # Modules > What each module of effect-cqrs exports. Everything is exported from `effect-cqrs` as a namespace, as in `import { Aggregate, ViewModel } from "effect-cqrs"`, and each module also has its own path, such as `effect-cqrs/ViewModel`. `PersonalDataNode` only has its own path, which keeps `node:crypto` out of code for browsers. ## Aggregate Defines aggregates. [Source](https://github.com/davidnussio/effect-cqrs/blob/main/src/Aggregate.ts) | Export | What it is | | --------------------------------------------------------------------- | ---------------------------------------------------------------------------------- | | `make()({ name, events, initial, projection, versions })` | Starts an aggregate, to which `.command(name, { payload, handler })` adds commands | | `evolve(aggregate, state, event, meta)` | Applies one event with the aggregate’s projection | | `historical(aggregate, tag, version, value)`, `Historical` | An event as an older version of its schema stored it | | `schemaVersion(aggregate, tag)`, `Versions` | The version new events are written in, and the type of `versions` | | `EventMeta`, `CommandContext` | What projections and handlers receive besides the event or payload | | `State`, `Event`, `CommandName`, `Payload`, `CommandError`, `Context` | Types derived from an aggregate | ## AggregateService Runs an aggregate against the event store. [Source](https://github.com/davidnussio/effect-cqrs/blob/main/src/AggregateService.ts) | Export | What it is | | --------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------ | | `make(aggregate, { maxRetries, stateCacheSize, commandIdsPerStream, snapshots })` | The service: `commands`, `dispatch`, `state`, `history` | | `SnapshotOptions` | `every`, `schema` and `version` of the snapshots | | `CommandOptions` | `commandId` and `expectedVersion`, the last argument of a typed command | | `CommandResult`, `HistoryEntry` | What a command and `history` return | | `UnknownCommand`, `InvalidCommand`, `WrongExpectedVersion`, `EventDecodeError` | Its errors, besides the store’s | | `CommandFailure`, `ReadError` | What every command, and every read, can fail with | | `commandIdKey` | The metadata key that records a command’s [`commandId`](/guides/commands/#retries-and-command-ids) | | `makeEventDecoder(aggregate, keys)`, `makeEventEncoder(aggregate, keys)` | Convert between stored JSON and an aggregate’s events, decrypting and encrypting personal fields with `keys` | | `MetricNames` | The names of the [metrics](/guides/commands/#metrics) the service records | ## CommandGateway One entry point for the commands of several aggregates, see [several aggregates](/guides/commands/#several-aggregates). [Source](https://github.com/davidnussio/effect-cqrs/blob/main/src/CommandGateway.ts) | Export | What it is | | ---------------------------------- | ----------------------------------------------------------------------------- | | `make(entries, { interceptors })` | The gateway over aggregates or their services: `send`, `dispatch`, `services` | | `Interceptor`, `CommandInfo` | What runs around every command, and what it learns about the command | | `Command` | The typed command `send` takes | | `UnroutableCommand` | `dispatch` can’t tell which aggregate a command is for | ## PersonalData Crypto-shredding of personal fields, see [personal data](/guides/personal-data/). [Source](https://github.com/davidnussio/effect-cqrs/blob/main/src/PersonalData.ts) | Export | What it is | | -------------------------------------------------------------------------------- | ------------------------------------------------------------------- | | `field(schema, { redacted, subject })` | Marks a field of an event as personal | | `forget(subject)` | Destroys the key of a subject and records it in the log | | `Keys`, `layerMemory`, `layerSql`, `make(backend)` | The keys of subjects, and where they are stored | | `MasterKey`, `layerMasterKey`, `layerMasterKeyConfig`, `layerMasterKeyEphemeral` | What wraps the keys of subjects | | `Backend`, `makeMemoryBackend`, `makeSqlBackend` | Key stores, to implement another | | `migrations({ table })` | The migrations of the key table, for `layerSql({ migrate: false })` | | `Cipher`, `webCrypto` | AES-256-GCM, on WebCrypto unless another cipher is provided | | `SubjectForgotten`, `PersonalDataError` | Its errors | | `streamName`, `subjectForgottenType` | Where forgetting is recorded | ## PersonalDataNode The [personal data](/guides/personal-data/#on-node) cipher on `node:crypto`, imported from `effect-cqrs/PersonalDataNode`. [Source](https://github.com/davidnussio/effect-cqrs/blob/main/src/PersonalDataNode.ts) | Export | What it is | | ------------- | ----------------------------------------------------------------------- | | `layerCipher` | Encrypts personal fields synchronously, in the same format as WebCrypto | | `cipher` | The `Cipher` it provides | ## EventVersions Older shapes of an event. [Source](https://github.com/davidnussio/effect-cqrs/blob/main/src/EventVersions.ts) | Export | What it is | | ------------------------- | ----------------------------------------------------------------------------- | | `from(schema)` | Starts a history at version 1, to which `.next(schema, upcast)` adds versions | | `decoders(tag, versions)` | One decoder per old version, all the way to the current one | | `current(versions)` | The version new events are written in | ## Causation Why events were recorded. [Source](https://github.com/davidnussio/effect-cqrs/blob/main/src/Causation.ts) | Export | What it is | | --------------------------------------- | ------------------------------------------------------------- | | `withCausation(causation)` | Runs an effect whose commands record these ids | | `causedBy(stored)`, `fromEvent(stored)` | Makes a stored event the cause of the commands an effect runs | | `CurrentCausation` | The reference that holds them, `undefined` outside any | ## Ids The ids the framework makes up, such as new correlations. [Source](https://github.com/davidnussio/effect-cqrs/blob/main/src/Ids.ts) | Export | What it is | | -------------------- | --------------------------------------------------------------- | | `IdGenerator` | The reference that makes them, `crypto.randomUUID()` by default | | `next` | A new id from the current generator | | `sequential(prefix)` | A generator of `prefix-1`, `prefix-2` and so on, for tests | ## Metadata What commands carry beside their payload, such as the software version. [Source](https://github.com/davidnussio/effect-cqrs/blob/main/src/Metadata.ts) | Export | What it is | | ------------------------ | --------------------------------------------------------------------------------------- | | `withMetadata(metadata)` | Runs an effect whose commands record this metadata on top of the current one | | `layer(metadata)` | Provides metadata to the services built on top, for values such as the software version | | `CurrentMetadata` | The reference that holds it, `{}` outside any | | `Metadata` | The type and schema of metadata: a JSON object | ## ViewModel Projects events into read models. [Source](https://github.com/davidnussio/effect-cqrs/blob/main/src/ViewModel.ts) | Export | What it is | | --------------------------------------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------- | | `make({ name, aggregates, handlers, dispose, onForget, onFailure, transaction })` | Defines a view model. `onForget` erases what it stored about a [forgotten subject](/guides/personal-data/#view-models) | | `layer([...])` | Runs view models in the background and provides `ViewModels` | | `ViewModels` | The service: `replay`, `caughtUp`, `status`, `parked`, `retryParked` | | `stall`, `park({ attempts })`, `FailurePolicy` | What a view model does about an event that keeps failing, see [parking events](/guides/view-models/#parking-events) | | `sqlTransaction`, `Transaction` | Runs each batch with its checkpoint in one transaction, see [checkpoints in a transaction](/guides/view-models/#checkpoints-in-a-transaction) | | `Status`, `RetryParkedOptions`, `RetryParkedResult` | What `status` and `retryParked` give and take | | `MetricNames` | The names of the metrics view models record | | `ViewModelError`, `UnknownViewModel` | A handler’s failure, and a replay of an unknown name | ## ParkedEvents Events view models gave up on, see [parking events](/guides/view-models/#parking-events). [Source](https://github.com/davidnussio/effect-cqrs/blob/main/src/ParkedEvents.ts) | Export | What it is | | ----------------------------------------- | --------------------------------------------------------------- | | `ParkedEvents` | The service: `park`, `list`, `remove` | | `layerMemory`, `makeMemory` | Parked events in memory | | `layerSql({ table, migrate })`, `makeSql` | Parked events in a table, on SQLite or PostgreSQL | | `migrations({ table })` | The migrations of the table, for `layerSql({ migrate: false })` | | `ParkedEvent` | What it records, and its schema | ## Leases Which process runs each view model, see [several processes](/guides/view-models/#several-processes). [Source](https://github.com/davidnussio/effect-cqrs/blob/main/src/Leases.ts) | Export | What it is | | ----------------------------------------------------- | ---------------------------------------- | | `Leases` | The service: `acquire(name)` | | `layerMemory`, `memory` | Leases that are always held, the default | | `layerPostgres({ interval, prefix })`, `makePostgres` | Leases on PostgreSQL advisory locks | | `Lease` | A lease: `held` and `changes` | ## Saga Processes that span several aggregates, as durable workflows from `effect/workflow`, see [sagas](/guides/sagas/). [Source](https://github.com/davidnussio/effect-cqrs/blob/main/src/Saga.ts) | Export | What it is | | ------------------------------------------------------------ | ----------------------------------------------------------------------------------------------------------------- | | `workflow(tag, { payload, idempotencyKey, success, error })` | Defines a saga’s workflow: `Workflow.make` with a `causation` field added to the payload | | `toLayer(workflow, execute)` | Registers the workflow’s body with the engine. Its commands record the event that started the saga as their cause | | `command(name, { execute, error })` | Runs a command as an activity, so a replay of the workflow returns the saved outcome instead of running it again | | `start(workflow, payload)` | Starts a run without waiting for it, and returns its execution id | | `make({ name, aggregates, handlers, onFailure })` | The trigger: a view model whose handlers start workflows with `start` | | `layer([...])` | Runs triggers in the background, apart from the view models, and provides `Sagas` | | `Sagas` | The service that manages the triggers, with what `ViewModel.ViewModels` has | | `SagaWorkflow` | The type of a workflow whose payload carries its causation | | `CausationSchema`, `CommandOutcome` | Schemas of that causation, and of what a saga keeps of a command it ran: aggregate, version and position | ## CommandApi Serves an aggregate over HTTP, see [integration](/guides/integration/#http). [Source](https://github.com/davidnussio/effect-cqrs/blob/main/src/CommandApi.ts) | Export | What it is | | -------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------- | | `group(aggregate, { name, prefix, state, errors })` | An `HttpApiGroup` with a `POST` endpoint per command, `history` and, with the `state` option, `state` | | `layer(api, name, service)` | Implements the group with an aggregate service, or an effect that makes one | | `InvalidCommandError`, `ConcurrencyConflictError`, `CommandRejected` | The errors it answers with `400`, `409` and `422` | | `commandResult(events)`, `historyEntry(events)` | Schemas of what a command and `history` answer | ## EventCatalog JSON Schema of events and commands, and a compatibility guard, see [integration](/guides/integration/#event-catalog). [Source](https://github.com/davidnussio/effect-cqrs/blob/main/src/EventCatalog.ts) | Export | What it is | | ------------------------------------------------------- | --------------------------------------------------------------------- | | `events(aggregate)`, `commands(aggregate)` | JSON Schema of every event version and command payload | | `document(aggregates, { title, version, description })` | An AsyncAPI 3.0 document of them | | `captureSamples(aggregates, { perVersion })` | Stored payloads of each event type and version, without personal data | | `saveSamples(path, samples)`, `readSamples(path)` | Keep samples in a JSON file | | `checkCompatibility(aggregate, samples)` | Fails with `IncompatibleSamples` when a sample no longer decodes | | `Sample`, `IncompatibleSamples` | A sample, and the guard’s error | ## Outbox Publishes events to other systems, see [integration](/guides/integration/#outbox). [Source](https://github.com/davidnussio/effect-cqrs/blob/main/src/Outbox.ts) | Export | What it is | | ------------------------------------------------------------------ | ------------------------------------------------------- | | `make({ name, aggregates, messages, publish, batchSize, linger })` | Defines an outbox | | `layer([...])` | Runs outboxes in the background and provides `Outboxes` | | `Outboxes` | The service: `published` and `status` | | `Message`, `Mapped`, `Mappings` | What is published, and what a mapping returns | | `OutboxError` | A failure to publish | ## EventStore The append-only log. [Source](https://github.com/davidnussio/effect-cqrs/blob/main/src/EventStore.ts) | Export | What it is | | ---------------------------------------------------- | -------------------------------------------------------------------------------------- | | `EventStore` | The service: `append`, `read`, `subscribe`, `head`, `changes` | | `layerMemory`, `makeMemory` | The store in memory | | `make(backend, options)`, `Backend`, `MakeOptions` | Builds a store on another database | | `groupCommit(appendGroup)` | Group commit for a backend: concurrent appends to different streams in one transaction | | `StoredEvent`, `EventFilter`, `AppendInput` | What it stores, selects and takes | | `eventId(event)` | An event’s id as a cause, like `account/alice@4` | | `toStored(input, event, index, position, timestamp)` | A new event’s stored form, for a backend | | `ConcurrencyConflict`, `EventStoreError` | Its errors | ## Checkpoints How far each view model has read. [Source](https://github.com/davidnussio/effect-cqrs/blob/main/src/Checkpoints.ts) | Export | What it is | | ------------- | ---------------------------- | | `Checkpoints` | The service: `get` and `set` | | `layerMemory` | Checkpoints in memory | ## SqliteEventStore and PostgresEventStore Durable stores through `effect/sql`. [SQLite](https://github.com/davidnussio/effect-cqrs/blob/main/src/SqliteEventStore.ts), [PostgreSQL](https://github.com/davidnussio/effect-cqrs/blob/main/src/PostgresEventStore.ts) | Export | What it is | | ---------------------------------- | ---------------------------------------------------------------------------------------- | | `layer(options)` | The event store, given a `SqlClient` | | `layerCheckpoints(options)` | Checkpoints in the same database | | `migrations(options)`, `Migration` | The schema as SQL, for your own migration tooling | | `Options` | `eventsTable`, `checkpointsTable`, `migrate`, `pollInterval`, and `listen` on PostgreSQL | ## Fixture Tests an aggregate. [Source](https://github.com/davidnussio/effect-cqrs/blob/main/src/Fixture.ts) | Export | What it is | | -------------------------- | -------------------------------------------------------------------------------- | | `make(aggregate)` | A fixture: `givenEvents`, `when`, `whenFails`, `inspectState`, `checkInvariants` | | `InvariantOptions`, `Step` | What `checkInvariants` takes, and one command it generates | | `Scenario`, `Outcome` | The chain, and what yielding it gives | | `ExpectationFailed` | The defect a failed expectation raises | # Coming from moleculer-cqrs > Where each part of moleculer-cqrs went. effect-cqrs keeps the ideas of [moleculer-cqrs](https://github.com/davidnussio/moleculer-cqrs) and changes how they are put together: services and actions become an Effect service, validation becomes `Schema`, and thrown errors become typed ones. | moleculer-cqrs | effect-cqrs | | ---------------------------------------- | ---------------------------------------------------------- | | `CQRSEventSourcing({ aggregate })` mixin | `AggregateService.make(aggregate)` | | `call todo.command` | `todos.commands.x(id, payload)` or `todos.dispatch(json)` | | `call todo.read-model` | `todos.state(id, { finishTime })` | | `call todo.history` | `todos.history(id, { payload, startTime, finishTime })` | | `call todo.replay` | `viewModels.replay({ viewModels, startTime, finishTime })` | | view model service with `events` | `ViewModel.make({ handlers, dispose })` | | `new CQRSFixture(aggregate)` | `Fixture.make(aggregate)` | | `.givenEvents([...])` | `.givenEvents([...])` | | `.when(createNews, payload)` | `.when("createNews", payload)` | | `.expectEvent(event)` | `.expectEvent(event)` or `.expectEvent("Tag")` | | `.inspectState(cb)` | `.inspectState(cb)` | | `.whenThrow(command).toThrow(...)` | `.whenFails(command, payload)`, which returns the error | | `fastest-validator` | `Schema` | | `throw` in a command | typed errors | | `resolve-storage-*` adapters | `EventStore.Backend` | ## The read side The largest change is how view models get events. moleculer-cqrs broadcasts them over the broker, so an event sent while a service is down is lost until someone replays. Here view models pull from the store from their own checkpoint, so they can’t miss one, and `replay` is for rebuilding a view model rather than for recovering events. ## What isn’t there There is no transport like the Moleculer broker: `dispatch` is the entry point to wire into your own HTTP or RPC layer. There is also no `cqrs generate`. # Performance > What a command costs on each store, and what the stress tests checked. These numbers come from one laptop, an Apple M4 Pro on Node 26, with PostgreSQL 18 in Docker on the same machine, measured on 10 November 2026. Treat them as orders of magnitude, not as a promise. | | SQLite file | PostgreSQL | | --------------------------------------------------------------- | ---------------- | --------------- | | Command on an aggregate, p50 | 0.10 ms | 0.73 ms | | Command on an aggregate with 100,000 events, p50 | 0.17 ms | about 1 ms | | First load of an aggregate with 100,000 events, after a restart | 368 ms | 312 ms | | Append, 1 event per transaction | 13,461 events/s | 1,650 events/s | | Append, 100 events per transaction | 252,768 events/s | 71,316 events/s | | Ingest a million events with a view model following¹ | 35 s | 118 s | | Replay a million events¹ | 4.6 s | 12.4 s | | Space for a million events¹ | 196 MB | 312 MB | ¹ From the stress tests of 31 October 2026, before group commit. * **A command costs the same whatever the store holds.** The service remembers recent state and reads only new events. * **Commits are what cost.** Batching events into fewer commands helps more than anything else. Concurrent writers help too, because appends that wait for each other share a commit, as the next table shows. * **The first use of a long aggregate after a restart replays it**, at about 4 µs per event on SQLite, unless [snapshots](/guides/commands/#snapshots) are on. ## Concurrent writers Commands on new streams, run by a number of workers at once: | Workers | SQLite file | PostgreSQL | | ------- | ----------------- | ------------------------- | | 1 | 9,319 commands/s | 1,307 commands/s | | 16 | 14,592 commands/s | 4,000 to 8,000 commands/s | | 256 | 15,895 commands/s | 16,958 commands/s | Before [group commit](/guides/storage/#postgresql), every worker count gave about the same: 8,400 to 8,900 commands/s on a SQLite file and 700 to 1,100 on PostgreSQL, where an append took five round trips and held the counter until it committed. Now an append is one statement, and the appends that queue behind it are written together in the next one. With 16 workers on PostgreSQL the result varies from run to run, depending on how many appends happen to queue. A database in memory has no commit to share, so its throughput stays where it was. ## Personal data [Personal fields](/guides/personal-data/) cost one AES-GCM operation each. The benchmark compares an account whose movement reference is personal with the same account without it, on WebCrypto, the default, and on `node:crypto`: | | SQLite file | PostgreSQL | | ------------------------------------------------------------------ | ---------------- | ---------------- | | Command on a new stream, without personal fields | 9,091 commands/s | 1,348 commands/s | | Command on a new stream, one personal field, WebCrypto | 6,764 commands/s | 1,230 commands/s | | Command on a new stream, one personal field, `node:crypto` | 7,681 commands/s | 1,168 commands/s | | First load of 1,000 events, without | 4.0 ms | 3.5 ms | | First load of 1,000 events, one personal field each, WebCrypto | 12.4 ms | 12.7 ms | | First load of 1,000 events, one personal field each, `node:crypto` | 9.8 ms | 10.0 ms | * **A new subject costs the most.** Each of these commands is the first for its subject, so it also creates, wraps and stores a key. Later commands reuse the key from memory. * **Reading costs about 8 µs per encrypted field on WebCrypto, and 6 µs on `node:crypto`.** WebCrypto is asynchronous, so the events of a page decrypt together while a stream is replayed, still folded in order. `node:crypto` is synchronous and needs no such detour. A command on a state already in memory decrypts nothing, and one it has just written isn’t decrypted again. * **Snapshots matter more.** With personal fields the first load of a long stream is two to three times slower, so [snapshots](/guides/commands/#snapshots) pay off sooner. ## Nothing lost Ten stress scenarios ran on both databases and broke no invariant. They included two runs of a million events, writer processes killed with `SIGKILL` while committing, three at a time on the same aggregates, view models restarted mid-batch, handlers failing on purpose, and 128 writers on one aggregate. Every acknowledged event was stored exactly once, positions and versions had no gaps, and no killed command left half its events behind. ## Run them yourself ```bash pnpm bench pnpm stress ``` [The full report](https://github.com/davidnussio/effect-cqrs/blob/main/bench/REPORT.md)Every measurement, what the tests found and fixed, and the limits that remain.