Domain reference

Aggregate

Source: src/Domain/Aggregate.ts

One entity that decides, such as a bank account: a consistency boundary. It takes a command, checks it against its current state, and records events or returns a rejection. Each instance handles one command at a time, and its events, new state and reply commit together.

Aggregate: Deciding and evolving

import * as Domain from "alchemy/Domain";
import * as Schema from "effect/Schema";

export const AccountId = Schema.String.pipe(Schema.brand("AccountId"));

export class MoneyWithdrawn extends Domain.Event("MoneyWithdrawn", {
  amount: Schema.Int,
  balanceAfter: Schema.Int,
}) {}

export class InsufficientFunds extends Domain.Rejection("InsufficientFunds", {
  balance: Schema.Int,
  requested: Schema.Int,
}) {}

export class WithdrawMoney extends Domain.Command("WithdrawMoney", {
  target: AccountId,
  input: { amount: Schema.Int },
  events: [MoneyWithdrawn],
  rejects: [InsufficientFunds],
  reply: { balance: Schema.Int },
}) {}

export class Account extends Domain.Aggregate("Account", {
  id: AccountId,
  state: Schema.Struct({ balance: Schema.Int }),
  initial: { balance: 0 },
  commands: [WithdrawMoney],
  decide: {
    WithdrawMoney: (s, cmd) =>
      cmd.amount > s.balance
        ? new InsufficientFunds({ balance: s.balance, requested: cmd.amount })
        : [new MoneyWithdrawn({ amount: cmd.amount, balanceAfter: s.balance - cmd.amount })],
  },
  evolve: {
    MoneyWithdrawn: (_, e) => ({ balance: e.balanceAfter }),
  },
  reply: {
    WithdrawMoney: (s) => ({ balance: s.balance }),
  },
}) {}

Aggregate: Authorizing in decide

A handler can be an Effect.fn that reads what the command’s middleware provides.

decide: {
  WithdrawMoney: Effect.fn(function* (s, cmd) {
    const actor = yield* CurrentActor;
    if (actor.customerId !== s.owner) return yield* new NotOwner();
    return [new MoneyWithdrawn({ amount: cmd.amount, balanceAfter: s.balance - cmd.amount })];
  }),
},

Aggregate: Exposing reads

read lets an Api expose the aggregate itself, so a client can get it, watch it, and follow its events. validate judges every read.

read: {
  middleware: [Session],
  rejects: [NotOwner],
  validate: Effect.fn(function* (_request, s) {
    const actor = yield* CurrentActor;
    if (actor.customerId !== s.owner) return yield* new NotOwner();
  }),
},

Aggregate: Sending and scheduling

const accounts = yield* Account;
const { reply } = yield* accounts.send(accountId, new WithdrawMoney({ amount: 30 }));
yield* accounts.schedule(accountId, new SendStatement(), { after: "30 days" });

Api

Source: src/Domain/Api.ts

The public surface of a domain for one kind of client: the commands it may send, and the views and aggregates (with a read contract) it may read. Nothing in an Api needs a handler. The class is an Effect: yield* MyApi gives a Worker its fetch. Call it with a typed client over Domain.protocolWebSocket or Domain.protocolHttp.

Api: Declaring an Api

import * as Domain from "alchemy/Domain";
import { Account, DepositMoney, WithdrawMoney } from "./Account.ts";
import { AccountSummary } from "./AccountSummary.ts";

export class CustomerApi extends Domain.Api(
  "CustomerApi",
  DepositMoney,
  WithdrawMoney,
  Account,
  AccountSummary,
) {}

Api: Serving it from a Worker

import * as Cloudflare from "alchemy/Cloudflare";
import * as DomainCloudflare from "alchemy/Domain/Cloudflare";
import * as Effect from "effect/Effect";
import * as Layer from "effect/Layer";
import { Bank, BankPolicies } from "./Bank.ts";
import { CustomerApi } from "./CustomerApi.ts";
import { SessionLive } from "./Session.ts";

const BankLive = SessionLive.pipe(
  Layer.provideMerge(Bank.layer(BankPolicies)),
  Layer.provide(DomainCloudflare.DurableObjects),
);

export default class BankWorker extends Cloudflare.Worker<BankWorker>()(
  "BankWorker",
  { main: import.meta.url },
  Effect.gen(function* () {
    const api = yield* CustomerApi;
    return { fetch: api.fetch };
  }).pipe(Effect.provide(BankLive)),
) {}

Api: Calling it

Domain.protocolWebSocket sends commands, get and list over HTTP, and opens a WebSocket for each watch and events, pushed by the object that owns the key. Domain.protocolHttp sends everything over HTTP.

const bank = yield* CustomerApi.client;
const { reply } = yield* bank.DepositMoney({ target: accountId, amount: 1_000 });
const summary = yield* bank.AccountSummary.get(accountId);
yield* bank.AccountSummary.watch(accountId).pipe(
  Stream.runForEach((s) => Effect.log(s?.balance)),
);

DurableObjects

Source: src/Domain/Cloudflare.ts Kind: Layer · Provides: Domain.Platform, WorkflowEngine

Host a domain model on Cloudflare Durable Objects. Every aggregate, view and policy becomes a Durable Object class on the host Worker, declared while the model’s Layer is built, so there are no bindings to write. One instance per aggregate id, per view key, and per policy and source.

Commands to one instance run one at a time, and their events, state and outbox commit together. Deliveries to views and policies are at least once and in order per source, and survive restarts. Live reads over Domain.protocolWebSocket are WebSockets to the object that owns the key, pushed on every commit; the object hibernates in between.

DurableObjects: Hosting on Cloudflare

import * as Cloudflare from "alchemy/Cloudflare";
import * as DomainCloudflare from "alchemy/Domain/Cloudflare";
import * as Effect from "effect/Effect";
import * as Layer from "effect/Layer";
import { Bank, BankPolicies } from "./Bank.ts";
import { CustomerApi } from "./CustomerApi.ts";
import { SessionLive } from "./Session.ts";

const BankLive = SessionLive.pipe(
  Layer.provideMerge(Bank.layer(BankPolicies)),
  Layer.provide(DomainCloudflare.DurableObjects),
);

export default class BankWorker extends Cloudflare.Worker<BankWorker>()(
  "BankWorker",
  { main: import.meta.url },
  Effect.gen(function* () {
    const api = yield* CustomerApi;
    return { fetch: api.fetch };
  }).pipe(Effect.provide(BankLive)),
) {}

Command

Source: src/Domain/Command.ts

A request to change one aggregate instance. It names the instance it targets, its input, the events it may record, the rejections it may return, the middleware that guards it, and the reply its caller receives.

Command: Declaring a command

import * as Domain from "alchemy/Domain";
import * as Schema from "effect/Schema";

export const AccountId = Schema.String.pipe(Schema.brand("AccountId"));

export class MoneyWithdrawn extends Domain.Event("MoneyWithdrawn", {
  amount: Schema.Int,
  balanceAfter: Schema.Int,
}) {}

export class InsufficientFunds extends Domain.Rejection("InsufficientFunds", {
  balance: Schema.Int,
  requested: Schema.Int,
}) {}

export class WithdrawMoney extends Domain.Command("WithdrawMoney", {
  target: AccountId,
  input: { amount: Schema.Int },
  events: [MoneyWithdrawn],
  rejects: [InsufficientFunds],
  reply: { balance: Schema.Int },
}) {}

Command: Guarding it with middleware

middleware runs in order before the command reaches its aggregate. What it provides, such as the current actor, is what decide may read.

import * as Domain from "alchemy/Domain";
import * as Schema from "effect/Schema";
import { AccountId, MoneyWithdrawn } from "./Account.ts";
import { MfaChallenge } from "./MfaChallenge.ts";
import { Session } from "./Session.ts";

export class WithdrawMoney extends Domain.Command("WithdrawMoney", {
  target: AccountId,
  middleware: [Session, MfaChallenge],
  input: { amount: Schema.Int },
  events: [MoneyWithdrawn],
  reply: { balance: Schema.Int },
}) {}

Command: Sending it once

Pass a commandId, and a resend with the same id returns the first receipt instead of applying the command again.

const receipt = yield* bank.WithdrawMoney({ target: accountId, amount: 2_000, commandId: requestId });

Event

Source: src/Domain/Event.ts

A fact your domain recorded, named in the past tense. Aggregates record events; their state, every view and every policy follow from them.

Event: Declaring events

An event with data

import * as Domain from "alchemy/Domain";
import * as Schema from "effect/Schema";

export class MoneyDeposited extends Domain.Event("MoneyDeposited", {
  amount: Schema.Int,
  balanceAfter: Schema.Int,
}) {}

An event with no data

import * as Domain from "alchemy/Domain";

export class AccountClosed extends Domain.Event("AccountClosed") {}

ExternalSystem

Source: src/Domain/ExternalSystem.ts

An outside system your model calls, such as a payment provider or a fraud scorer, declared as the operations the model uses. It’s an Effect service: policies yield* it, an adapter implements it with toLayer, and stories answer its calls themselves, so the model depends on what it asks for and never on a vendor.

ExternalSystem: Declaring an external system and an adapter

import * as Domain from "alchemy/Domain";
import * as Effect from "effect/Effect";
import * as Schema from "effect/Schema";

export class FraudCheck extends Domain.ExternalSystem("FraudCheck", {
  score: { args: { accountId: Schema.String, amount: Schema.Int }, success: { risk: Schema.Number } },
}) {}

export const SiftFraudCheck = FraudCheck.toLayer(
  Effect.succeed({
    score: ({ amount }) => Effect.succeed({ risk: amount >= 80_000 ? 0.95 : 0.05 }),
  }),
);

InMemory

Source: src/Domain/InMemory.ts Kind: Layer · Provides: Domain.Platform, WorkflowEngine

Host a domain model in the current process: aggregates, views and policies, with nothing to deploy. Use it in tests and single-process apps; swap it for DomainCloudflare.DurableObjects and nothing else changes.

InMemory: Hosting in memory

import * as Domain from "alchemy/Domain";
import * as Layer from "effect/Layer";
import { Bank, BankPolicies } from "./Bank.ts";

export const BankInMemory = Bank.layer(BankPolicies).pipe(Layer.provide(Domain.InMemory));

Policy

Source: src/Domain/Policy.ts

“Whenever this happens, do that”: one of the business’s standing rules. On an event from an aggregate or a view, it decides what to do next and acts by sending commands and calling external systems. Policies run shortly after the event, never in the same step, so use them only for eventually consistent work. Delivery is at least once; the commands a policy sends get ids derived from the event, so a second run sends the same commands.

Policy: Reacting to an aggregate

import * as Domain from "alchemy/Domain";
import * as Effect from "effect/Effect";
import { Account, AccountId, OpenAccount } from "./Account.ts";
import { Customer, CustomerRegistered } from "./Customer.ts";

export class MainAccountProvisioning extends Domain.Policy("MainAccountProvisioning", {
  from: Customer,
  on: [CustomerRegistered],
}) {}

export const MainAccountProvisioningLive = MainAccountProvisioning.toLayer(
  Effect.gen(function* () {
    const accounts = yield* Account;
    return Effect.fn(function* ({ source }) {
      yield* accounts
        .send(AccountId.make(`${source.id}-main`), new OpenAccount({ owner: source.id }))
        .pipe(Effect.catchTag("AlreadyOpen", () => Effect.void));
    });
  }),
);

Policy: Calling an external system

import * as Domain from "alchemy/Domain";
import * as Effect from "effect/Effect";
import { Account, MoneyWithdrawn } from "./Account.ts";
import { FraudCheck } from "./FraudCheck.ts";

export class FraudReview extends Domain.Policy("FraudReview", {
  from: Account,
  on: [MoneyWithdrawn],
}) {}

export const FraudReviewLive = FraudReview.toLayer(
  Effect.gen(function* () {
    const fraud = yield* FraudCheck;
    return Effect.fn(function* ({ source, event }) {
      const { risk } = yield* fraud.score({ accountId: source.id, amount: event.amount });
      yield* Effect.log(`risk ${risk} for ${source.id}`);
    });
  }),
);

Projection

Source: src/Domain/Projection.ts

Consumes events into a store you own: a database, a search index, a warehouse, or a Domain.Stream. Inputs are routed to keys and handled in order per key, at least once, and batched if its Layer says so. A projection never sends commands: replaying it must never act on the world again.

The declaration holds what it means (sources, partitioning, events, the streams it writes). Its Layer holds how it runs: the handler, and options such as batching.

Projection: Writing to a database

import * as Domain from "alchemy/Domain";
import * as Effect from "effect/Effect";
import { Account } from "./Account.ts";
import { MoneyDeposited } from "./DepositMoney.ts";
import { LedgerStore } from "./LedgerStore.ts";

export class Ledger extends Domain.Projection("Ledger", {
  from: [Account],
  key: Account,
  on: [MoneyDeposited],
}) {}

export const LedgerLive = Ledger.toLayer(
  Effect.gen(function* () {
    const store = yield* LedgerStore;
    return Effect.fn(function* (batch) {
      yield* store.upsert(
        batch.map(({ key, event, envelope }) => ({
          id: `${envelope.stream}:${envelope.seq}`,
          account: key.id,
          amount: event.amount,
        })),
      );
    });
  }),
  { batch: { size: 500, wait: "2 seconds" } },
);

Projection: Reading your own write

const receipt = yield* accounts.send(accountId, new DepositMoney({ amount: 100 }));
yield* Ledger.caughtUp(receipt);

Rejection

Source: src/Domain/Rejection.ts

A typed reason a command or read was refused. Nothing is recorded when a command is rejected, and the caller receives the rejection exactly as declared, through an Api too.

Rejection: Declaring rejections

import * as Domain from "alchemy/Domain";
import * as Schema from "effect/Schema";

export class NotOpen extends Domain.Rejection("NotOpen") {}

export class InsufficientFunds extends Domain.Rejection("InsufficientFunds", {
  balance: Schema.Int,
  requested: Schema.Int,
}) {}

RequestContext

Source: src/Domain/RequestContext.ts

A service middleware provides to a request, such as the current actor. It belongs to the request: a command carries it to its aggregate’s decide, and a read carries it to validate. Both may run elsewhere, such as in a Durable Object, so it has a schema. Policies provide it themselves when they send commands.

RequestContext: Declaring a request context

import * as Domain from "alchemy/Domain";
import * as Schema from "effect/Schema";

export class CurrentActor extends Domain.RequestContext<CurrentActor>()(
  "Bank/CurrentActor",
  Schema.Struct({ customerId: Schema.String }),
) {}

RequestContext: Providing it from middleware

import * as RpcMiddleware from "effect/rpc/RpcMiddleware";
import * as Schema from "effect/Schema";
import { CurrentActor } from "./CurrentActor.ts";

export class Unauthorized extends Schema.TaggedError<Unauthorized>()("Unauthorized", {}) {}

export class Session extends RpcMiddleware.Service<Session, { provides: CurrentActor }>()(
  "Bank/Session",
  { error: Unauthorized },
) {}

Saga

Source: src/Domain/Saga.ts

A process that spans several aggregates and takes time: a transfer that debits one account and credits another, or an order that waits for payment. Each step runs once and its result is kept, so a run that’s interrupted picks up where it left off. A step that fails after earlier ones succeeded can undo them with compensations.

A saga is an Effect workflow. Its steps are Activitys, it waits with DurableClock and DurableDeferred, and it undoes steps with withCompensation. A policy usually decides when to start one.

Saga: Declaring a saga

import * as Domain from "alchemy/Domain";
import * as Schema from "effect/Schema";
import { AccountId } from "./AccountId.ts";
import { TransferId } from "./TransferId.ts";

export class DestinationClosed extends Domain.Rejection("DestinationClosed") {}

export class ExecuteTransfer extends Domain.Saga("ExecuteTransfer", {
  key: TransferId,
  input: { from: AccountId, to: AccountId, amount: Schema.Int },
  rejects: [DestinationClosed],
}) {}

Saga: Implementing it

import * as Effect from "effect/Effect";
import { Activity } from "effect/workflow";
import { Account } from "./Account.ts";

export const ExecuteTransferLive = ExecuteTransfer.toLayer(
  Effect.gen(function* () {
    const accounts = yield* Account;
    return Effect.fn(function* ({ key, from, to, amount }) {
      yield* Activity.make({
        name: "debit",
        execute: accounts.send(from, new DebitTransfer({ transferId: key, amount })).pipe(Effect.orDie),
      }).pipe(
        ExecuteTransfer.withCompensation(() =>
          Activity.make({
            name: "refund",
            execute: accounts.send(from, new RefundTransfer({ transferId: key, amount })).pipe(Effect.orDie),
          }),
        ),
      );
      yield* Activity.make({
        name: "credit",
        error: DestinationClosed,
        execute: accounts
          .send(to, new CreditTransfer({ transferId: key, amount }))
          .pipe(Effect.catchTag("NotOpen", () => new DestinationClosed()), Effect.orDie),
      });
    });
  }),
);

Saga: Starting it

export const TransferExecutionLive = TransferExecution.toLayer(
  Effect.gen(function* () {
    const executions = yield* ExecuteTransfer;
    return Effect.fn(function* ({ source, event }) {
      yield* executions.start(source, { from: event.from, to: event.to, amount: event.amount });
    });
  }),
);

Story

Source: src/Domain/Story.ts

Given / when / then specifications of a domain model, run against the real model in memory: policies run for real, external system calls wait until you resolve them, and a clock you control stamps commands and fires scheduled ones. A story fails on any rejection, external system call or saga defect it doesn’t account for.

Story: Writing a story

import { it } from "@effect/vitest";
import { Story } from "alchemy/Domain/Story";
import { Account, InsufficientFunds, MoneyDeposited, MoneyWithdrawn, WithdrawMoney } from "../src/Account.ts";
import { Bank, BankPolicies } from "../src/Bank.ts";

const story = Story.make(Bank, { layer: BankPolicies });
const account = Account.ref("a-1");

it.effect("withdraws within the balance, and refuses beyond it", () =>
  story(
    Story.given(account, new MoneyDeposited({ amount: 100, balanceAfter: 100 })),
    Story.when(account, new WithdrawMoney({ amount: 30 })),
    Story.then(account, new MoneyWithdrawn({ amount: 30, balanceAfter: 70 })),
    Story.replied({ balance: 70 }),
    Story.when(account, new WithdrawMoney({ amount: 100 })),
    Story.rejected(new InsufficientFunds({ balance: 70, requested: 100 })),
  ),
);

Story: External systems and the clock

Answer an external system call, then move the clock

Story.expectCall(FraudCheck.score, { accountId: account.id, amount: 60_000 }),
Story.resolve(FraudCheck.score, { risk: 0.95 }),
Story.clock("2026-01-02T09:00:00Z"),

Name calls when several are waiting

Story.expectCall("a-1", FraudCheck.score, { accountId: a1.id }),
Story.expectCall("a-2", FraudCheck.score, { accountId: a2.id }),
Story.resolve("a-2", FraudCheck.score, { risk: 0.1 }),
Story.resolve("a-1", FraudCheck.score, { risk: 0.95 }),

Story: Sagas

Story.saga(ExecuteTransfer, transfer, { _tag: "Running" }),
Story.signal(ExecuteTransfer, transfer, Decision, "approved"),

Stream

Source: src/Domain/Stream.ts

A keyed, append-only log: one ordered list of entries per key, kept and replayable. A projection writes it; clients read it through an Api with the same middleware and validate as a view. Its latest entry is its current value, so a stream of snapshots serves both “what is it now?” (peek) and “tell me every change” (watch, events).

Stream: Declaring a stream

import * as Domain from "alchemy/Domain";
import * as Schema from "effect/Schema";
import { Account } from "./Account.ts";

export class CardChanged extends Domain.Event("CardChanged", {
  owner: Schema.String,
  balance: Schema.Int,
}) {}

export class AccountCard extends Domain.Stream("AccountCard", {
  key: Account,
  entries: [CardChanged],
}) {}

Stream: Writing it from a projection

A projection lists the streams it writes, then appends for the input it is handling.

export class AccountCards extends Domain.Projection("AccountCards", {
  from: [AccountSummary],
  key: Account,
  on: [MoneyDeposited],
  writes: [AccountCard],
}) {}

export const AccountCardsLive = AccountCards.toLayer(
  Effect.gen(function* () {
    const cards = yield* AccountCard;
    return Effect.fn(function* (batch) {
      for (const input of batch) {
        yield* cards.append(input, new CardChanged({ owner: input.state.owner, balance: input.state.balance }));
      }
    });
  }),
);

Stream: Reading it

const latest = yield* bank.AccountCard.peek(accountId);
yield* bank.AccountCard.watch(accountId).pipe(Stream.runForEach((card) => Effect.log(card)));

View

Source: src/Domain/View.ts

A read model: events folded into state per key, kept up to date as new events arrive. A view can emit events of its own, route one event to several keys, and decide who may read it.

View: Folding events into state

import * as Domain from "alchemy/Domain";
import * as Schema from "effect/Schema";
import { Account } from "./Account.ts";

export class AccountSummary extends Domain.View("AccountSummary", {
  from: [Account],
  key: Account,
  state: Schema.Struct({ balance: Schema.Int }),
  initial: { balance: 0 },
  evolve: {
    MoneyDeposited: (v, e) => ({ ...v, balance: e.balanceAfter }),
    MoneyWithdrawn: (v, e) => ({ ...v, balance: e.balanceAfter }),
  },
}) {}

View: Emitting events

import * as Domain from "alchemy/Domain";
import * as Schema from "effect/Schema";
import { Account } from "./Account.ts";

export class StatementLine extends Domain.Event("StatementLine", {
  kind: Schema.Literals(["deposit", "withdrawal"]),
  amount: Schema.Int,
}) {}

export class Statement extends Domain.View("Statement", {
  from: [Account],
  key: Account,
  events: [StatementLine],
  emit: {
    MoneyDeposited: (e) => new StatementLine({ kind: "deposit", amount: e.amount }),
    MoneyWithdrawn: (e) => new StatementLine({ kind: "withdrawal", amount: e.amount }),
  },
}) {}

View: Routing one event to several keys

keyOf: {
  TransferRequested: {
    out: ({ event }) => Account.ref(event.from),
    in: ({ event }) => Account.ref(event.to),
  },
},

View: Validating reads

middleware: [Session],
rejects: [NotOwner],
validate: Effect.fn(function* (_request, summary) {
  const actor = yield* CurrentActor;
  if (summary?.owner && summary.owner !== actor.customerId) return yield* new NotOwner();
}),