# 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`](https://effect-cqrs.dambox.ch/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](https://effect-cqrs.dambox.ch/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](https://effect-cqrs.dambox.ch/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](https://effect-cqrs.dambox.ch/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](https://effect-cqrs.dambox.ch/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](https://effect-cqrs.dambox.ch/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<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:

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

```ts
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)
}
```