Skip to content
effect-cqrs

The shop

Open the shop demoThe 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 and the 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, which you can run with pnpm tutorial.

1. Events

Section titled “1. Events”

Events are what happened. Each aggregate declares its own as a tagged union:

examples/tutorial/shop.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

Section titled “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
export class OutOfStock extends Schema.TaggedError<OutOfStock>()("OutOfStock", {
  sku: Schema.String,
  available: Schema.Int
}) {}


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


export class OrderNotPayable extends Schema.TaggedError<OrderNotPayable>()("OrderNotPayable", {
  orderId: Schema.String,
  status: Schema.String
}) {}


export class OrderAlreadyPaid extends Schema.TaggedError<OrderAlreadyPaid>()("OrderAlreadyPaid", {
  orderId: Schema.String
}) {}
examples/tutorial/shop.ts
export interface StockState {
  readonly available: number
  /** What each order holds, by order id. */
  readonly reserved: Readonly<Record<string, number>>
}


export const Stock = Aggregate.make<StockState>()({
  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
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<OrderState>()({
  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, and versions comes in step 8.

3. Running commands

Section titled “3. Running commands”

AggregateService.make turns an aggregate into a service with one typed function per command:

examples/tutorial/shop.ts
export class Orders extends Context.Service<Orders, AggregateService.AggregateService<typeof Order>>()("shop/Orders") {
  static readonly layer = Layer.effect(Orders, AggregateService.make(Order))
}


export class Stocks extends Context.Service<Stocks, AggregateService.AggregateService<typeof Stock>>()("shop/Stocks") {
  static readonly layer = Layer.effect(Stocks, AggregateService.make(Stock))
}
examples/tutorial/shop.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

Section titled “4. A view model”

A view model reads events into whatever a query needs, here a list of orders:

examples/tutorial/shop.ts
export class OrderList extends Context.Service<OrderList, Ref.Ref<ReadonlyMap<string, string>>>()("shop/OrderList") {
  static readonly layer = Layer.effect(OrderList, Ref.make<ReadonlyMap<string, string>>(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

Section titled “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
/** 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

Section titled “6. Wiring”

Everything is a layer. Here all of it runs in memory:

examples/tutorial/shop.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

Section titled “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
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 })))
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

Section titled “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
// 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
// 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)
order-0 is read as { status: 'rejected', sku: 'pear', quantity: 1, payment: null }

9. Metrics

Section titled “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
// 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)
order.place timings [
  { outcome: 'success', count: 2, totalMs: 0.806 },
  { outcome: 'OrderAlreadyPlaced', count: 1, totalMs: 0.273 }
]

The full example

Section titled “The full example”
examples/tutorial/shop.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>()("OutOfStock", {
  sku: Schema.String,
  available: Schema.Int
}) {}


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


export class OrderNotPayable extends Schema.TaggedError<OrderNotPayable>()("OrderNotPayable", {
  orderId: Schema.String,
  status: Schema.String
}) {}


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


export interface StockState {
  readonly available: number
  /** What each order holds, by order id. */
  readonly reserved: Readonly<Record<string, number>>
}


export const Stock = Aggregate.make<StockState>()({
  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<OrderState>()({
  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<Orders, AggregateService.AggregateService<typeof Order>>()("shop/Orders") {
  static readonly layer = Layer.effect(Orders, AggregateService.make(Order))
}


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


export class OrderList extends Context.Service<OrderList, Ref.Ref<ReadonlyMap<string, string>>>()("shop/OrderList") {
  static readonly layer = Layer.effect(OrderList, Ref.make<ReadonlyMap<string, string>>(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<string>) =>
  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
/** 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)
Terminal window
pnpm tutorial

Next, the bank puts the same pieces in an app you can poke at, and the guides go deeper into aggregates, commands, view models and sagas.