Aggregates

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.

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>()("TodoAlreadyDeleted", {}) {}

type TodoState = { readonly title: string; readonly deleted: boolean }

const Todo = Aggregate.make<TodoState>()({
  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 and the command’s metadata), and returns the new state.

Aggregate.make<State>() takes the state type explicitly, in a curried call like Schema.TaggedError<X>(). 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.

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<TodoState>()({
  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<typeof Todo>	The state
Aggregate.Event<typeof Todo>	The union of its events
Aggregate.CommandName<typeof Todo>	"createTodo" | "deleteTodo"
Aggregate.Payload<typeof Todo, "createTodo">	{ readonly title: string }
Aggregate.CommandError<typeof Todo, "deleteTodo">	TodoAlreadyDeleted
Aggregate.Context<typeof Todo>	The services its handlers need