Commands

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.

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 and 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
correlationId	The correlation its events joined, see 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 or a view model, 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, and a commandId and an expectedVersion:

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

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:

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. 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 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:

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 are on.

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:

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 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:

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 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 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.

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:

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.

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:

class Todos extends Context.Service<Todos, AggregateService.AggregateService<typeof Todo>>()("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:

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:

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:

const authorize: CommandGateway.Interceptor<Unauthorized, CurrentUser> = (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<E, R>: 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:

const makeGateway = CommandGateway.make([Todo, Account], { interceptors: [authorize] })

class Gateway extends Context.Service<Gateway, Effect.Success<typeof makeGateway>>()("app/Gateway") {
  static readonly layer = Layer.effect(Gateway, makeGateway)
}