Integration

An aggregate service runs in one process. These modules connect it to the rest of a system:

Module	For
CommandApi	Commands and reads over HTTP, with OpenAPI
EventCatalog	JSON Schema of events and commands, and a guard against breaking stored events
Outbox	Publishing events to Kafka, NATS, SQS and the like
LayerMap	One store per tenant, a pattern rather than a module

examples/server puts the first one together with a view model, SQLite or PostgreSQL and OpenTelemetry. Run it with pnpm example:server.

HTTP

CommandApi.group builds an HttpApiGroup from an aggregate, and CommandApi.layer implements it with an aggregate service:

import { HttpApi, HttpApiBuilder, HttpApiScalar } from "effect/http-api"
import { AggregateService, CommandApi } from "effect-cqrs"

const TodoApi = CommandApi.group(Todo, {
  state: TodoState, // a schema of the state, for GET /todo/:aggregateId
  errors: [TodoAlreadyCreated, TodoNotFound, TodoAlreadyDeleted]
})

class Api extends HttpApi.make("api").add(TodoApi) {}

const Routes = Layer.mergeAll(
  HttpApiBuilder.layer(Api, { openapiPath: "/openapi.json" }).pipe(
    Layer.provide(CommandApi.layer(Api, "todo", AggregateService.make(Todo)))
  ),
  HttpApiScalar.layer(Api, { path: "/docs" })
)

effect-cqrs-http-example is this example as a repository of its own, with a server and SQLite storage, to clone and run.

Endpoint	Does
POST /todo/:aggregateId/<command>	Runs the command with the body as payload, and answers with the CommandResult
GET /todo/:aggregateId/history	history, with optional payload, startTime, finishTime, afterVersion and limit in the query
GET /todo/:aggregateId	state, with an optional finishTime. Only with the state option

The group is an ordinary HttpApiGroup: name and prefix change its identifier and path, and middleware, OpenAPI annotations and HttpApiClient work as with any other. A request without a body runs a command whose payload has no required fields. history answers with at most 1,000 events, the historyLimit option of CommandApi.layer, and a client reads the next page with afterVersion set to the version of the last entry it got.

Two headers reach the command. Idempotency-Key becomes its commandId, so a client that retries after a timeout gets the first result back instead of a second change. The body sent with a known key isn’t compared, so a key belongs to one request. If-Match is the version the client based the command on, as 3 or "3", and a stream that has moved on answers 409. If-Match: * asks for no version in particular.

Failures map to status codes:

Status	When	Body
400	The body isn’t JSON or doesn’t match the payload schema	InvalidCommand
409	Another writer kept winning the race for the stream	ConcurrencyConflict
409	The stream is no longer at the version in If-Match	WrongExpectedVersion
422	The handler failed with an error in errors	The error, with its fields
422	The handler failed with another error	CommandRejected, with the error’s tag and message
500	The store, decoding or the keys failed	Empty

A command’s errors are Effect types rather than schemas, so the ones a client should see in full are listed in errors. They are answered with 422, or with their own httpApiStatus: Schema.TaggedError<X>()("X", { ... }, { httpApiStatus: 404 }).

HttpApiTest.groups(Api, ["todo"]) gives a typed client that calls the handlers without a server, which is how test/CommandApi.test.ts tests them.

Reading your own writes

GET /todo/:aggregateId folds the aggregate’s events on every request, so it always shows the latest command. Your own query endpoints read view models, which run a moment behind. A client that runs a command and then reads a list can miss its own change.

Every command answers with the store position of its last event, as position in the body and in an Event-Position header. Add CommandApi.ReadAfter to the query endpoints, and a request that sends the position back as Read-After waits until the view models of this process have handled it:

const Catalog = HttpApiGroup.make("catalog")
  .add(HttpApiEndpoint.get("available", "/books/available", { success: Books }))
  .middleware(CommandApi.ReadAfter)

class Api extends HttpApi.make("api").add(BookApi).add(Catalog) {}

const Server = HttpApiBuilder.layer(Api).pipe(
  Layer.provide([BookHandlers, CatalogHandlers]),
  Layer.provide(CommandApi.layerReadAfter({ timeout: "2 seconds" })),
  Layer.provide(ViewModel.layer([CatalogViewModel]))
)

const response = await fetch("/book/book-1/lend", { method: "POST", body })
const position = response.headers.get("event-position")!
const books = await fetch("/books/available", { headers: { "read-after": position } })

A request without the header runs at once. One that waits longer than timeout, 5 seconds by default, is answered with 503 and NotCaughtUp: a view model is far behind, or stalled on a failing handler, and viewModels.status says which. A header that isn’t a position is answered with 400. In a browser on another origin, list Event-Position in Access-Control-Expose-Headers so the page can read it.

layerReadAfter waits for the view models that run in the same process, as caughtUp does. With several replicas, send the read to a replica that runs the view models it needs.

Event catalog

EventCatalog describes events and commands as JSON Schema, for teams that consume them:

import { EventCatalog } from "effect-cqrs"

EventCatalog.events(Account)   // every event, with every version from EventVersions
EventCatalog.commands(Account) // every command payload
EventCatalog.document([Account, Transfer], { title: "Bank", version: "1.0.0" }) // AsyncAPI 3.0

The schemas describe the JSON form of each event, as Schema.toCodecJson encodes it. In the AsyncAPI document each aggregate has a channel of events it sends and one of commands it receives. A message keeps its older versions under x-versions and the paths of its personal fields under x-personal-fields.

Compatibility guard

A change to an event schema can make events already in the store unreadable. The guard keeps a few stored payloads of every event type and version in a file, and a test checks that the current aggregate still decodes them, the way a store read does:

// Once in a while, against a copy of production data:
const samples = yield* EventCatalog.captureSamples([Account])
yield* EventCatalog.saveSamples("test/samples/account.json", samples)

// In a test:
it.effect("still reads stored events", () =>
  Effect.gen(function*() {
    const samples = yield* EventCatalog.readSamples("test/samples/account.json")
    yield* EventCatalog.checkCompatibility(Account, samples)
  }).pipe(Effect.provide(NodeFileSystem.layer)))

checkCompatibility fails with IncompatibleSamples, which lists each sample that no longer decodes and why. It returns how many samples it checked. saveSamples adds to the file and leaves the samples already there, so it keeps the shapes the store held over time.

Note

In the store, personal fields are encrypted. captureSamples puts each field’s redacted value in their place, so samples hold no personal data, can be committed, and are checked without keys. What the guard can’t see is a personal field whose plaintext type changed: the redacted value is checked, not real data.

Outbox

An outbox publishes events to another system as integration events:

import { Outbox } from "effect-cqrs"

const TodoOutbox = Outbox.make({
  name: "todo-outbox",
  aggregates: [Todo],
  messages: {
    TodoCreated: ({ title }, { aggregateId }) => ({ type: "todo.created.v1", payload: { id: aggregateId, title } }),
    TodoDeleted: (_, { aggregateId }) => ({ type: "todo.deleted.v1", payload: { id: aggregateId } })
  },
  publish: (messages) => producer.send(messages), // an Effect
  batchSize: 100,
  linger: "10 millis"
})

const Publishing = Outbox.layer([TodoOutbox]).pipe(
  Layer.provide(Layer.merge(SqliteEventStore.layer(), SqliteEventStore.layerCheckpoints()))
)

messages maps events by tag, and only those are published. A mapping that returns undefined skips the event. Without messages, every event is published as <aggregate>.<tag> with its JSON form as payload. Each Message carries the payload with the event’s stream, version, position, timestamp, correlationId and causationId.

An outbox is a view model: it reads the store from its checkpoint, so events stored while it was down are published when it starts, and nothing is lost between storing an event and publishing it. Delivery is at least once:

• Messages go out in batches of up to batchSize, when a batch is full or linger after its first message. The checkpoint never moves past a message that wasn’t published.
• A failed batch is retried in order, with growing pauses, until it succeeds. Once batchSize messages wait, the outbox stops reading the store, and the status of Outbox.Outboxes shows the failure.
• After a crash, messages published since the last checkpoint are published again. Message.id is the event’s id, such as todo/a@1, and stays the same, so consumers deduplicate on it.

Checkpoints should be durable. With Checkpoints.layerMemory an outbox publishes everything again after every restart. In tests, Outboxes’ published(position) waits until everything up to a position is published.

Caution

Events reach the mappings decrypted. Leave personal fields out of integration events, or other systems keep data that PersonalData.forget can no longer erase.

Tenants

Every service here works on whatever EventStore it is given, so tenants are kept apart by giving each one its own store. Effect’s LayerMap builds the layers of a tenant the first time it is asked for, keeps them while they are used, and releases them after idleTimeToLive:

import { LayerMap } from "effect"

class Tenants extends LayerMap.Service<Tenants>()("app/Tenants", {
  lookup: (tenant: string) => {
    const tables = { eventsTable: `${tenant}_events`, checkpointsTable: `${tenant}_checkpoints` }
    return Layer.mergeAll(Todos.layer, ViewModel.layer([TodoList])).pipe(
      Layer.provideMerge(Layer.mergeAll(
        SqliteEventStore.layer(tables),
        SqliteEventStore.layerCheckpoints(tables),
        Layer.fresh(TodoListStore.layer)
      ))
    )
  },
  dependencies: [SqliteClient.layer({ filename: "events.sqlite" })],
  idleTimeToLive: "5 minutes"
}) {}

const listForTenant = (tenant: string) => listTodos.pipe(Effect.provide(Tenants.get(tenant)))

• The tenants share the database client of dependencies, and each has its own tables, so the same aggregate id in two tenants is two streams. PostgreSQL works the same way, with PostgresEventStore. Another database per tenant also works: put the client in lookup.
• A layer value is built once per LayerMap, so in-memory state of a tenant’s own, like the list above, needs Layer.fresh.
• Table names come from the tenant id. sql(name) quotes identifiers, but check the id against a known alphabet before it gets there.
• View models and outboxes of a tenant run while its layers are alive. A tenant evicted for being idle catches up from its checkpoints the next time it is used.
• Personal data keys go per tenant too, with PersonalData.layerSql({ table }).

test/MultiTenancy.test.ts runs this on SQLite and PostgreSQL.