View models

A view model folds events into whatever shape a query needs: a list, a total, a table. Each one reads the event store from its own checkpoint, so it catches up after downtime and can’t miss an event. It runs after the command has stored its events, not in the same transaction, so it is a moment behind the write side: see reading your own writes.

import { Checkpoints, ViewModel } from "effect-cqrs"

const TodoList = ViewModel.make({
  name: "todo-list",
  aggregates: [Todo],
  dispose: clearList, // runs before a replay
  handlers: {
    TodoCreated: ({ title }, { aggregateId }) => addToList(aggregateId, title),
    TodoDeleted: (_, { aggregateId }) => removeFromList(aggregateId)
  }
})

const program = Effect.gen(function*() {
  const todos = yield* AggregateService.make(Todo)
  const viewModels = yield* ViewModel.ViewModels

  const { position } = yield* todos.commands.createTodo("todo-1", { title: "Buy Milk" })
  yield* viewModels.caughtUp(position) // read your own write

  yield* viewModels.replay({ viewModels: ["todo-list"] })
  yield* viewModels.status
}).pipe(
  Effect.provide(ViewModel.layer([TodoList]).pipe(
    Layer.provideMerge(Layer.merge(EventStore.layerMemory, Checkpoints.layerMemory))
  ))
)

Defining one

handlers has one function per event tag the view model cares about, and only accepts tags of its aggregates. Each gets the decoded event and its EventMeta, the same as a projection, with the command’s metadata. dispose clears everything the view model stored, before a replay rebuilds it.

ViewModel.layer([...]) starts the view models in the background for as long as the layer lives, and provides the ViewModels service. It needs an EventStore, Checkpoints and whatever services the handlers use. It also uses ParkedEvents and Leases when they are provided, see failing handlers and several processes.

At least once

A view model pulls events from the store, rather than waiting for them to be pushed, so an event can’t be lost while it is down. It saves its checkpoint every 500 events and when it stops, so after a crash it can see up to two batches of events again.

Note

Handlers must be idempotent. The bank demo’s accounts view, for instance, ignores an event whose position it has already applied.

Checkpoints in a transaction

A view model that keeps its data in the same database as its checkpoints can commit both together. With transaction, each batch of handlers runs in one transaction with the checkpoint it reaches, so the checkpoint is saved after every batch. A batch that fails leaves nothing behind, and a crash repeats nothing, so handlers needn’t be idempotent.

const TodoRows = ViewModel.make({
  name: "todo-rows",
  aggregates: [Todo],
  dispose: SqlClient.SqlClient.use((sql) => sql`DELETE FROM todo_rows`),
  transaction: ViewModel.sqlTransaction,
  handlers: {
    TodoCreated: ({ title }, { aggregateId }) =>
      SqlClient.SqlClient.use((sql) => sql`INSERT INTO todo_rows (id, title) VALUES (${aggregateId}, ${title})`)
  }
})

ViewModel.sqlTransaction uses the SqlClient in context. The handlers and the checkpoints, SqliteEventStore.layerCheckpoints() or PostgresEventStore.layerCheckpoints(), have to get that same client: a transaction belongs to one client, and another client’s writes don’t join it. Any other (effect) => Effect works as transaction too, for a database of its own.

Only the database is rolled back. A handler that runs a command or calls a service still does that once per attempt.

Reading your own writes

A command returns the position of its last event. viewModels.caughtUp(position) waits until every view model has processed it, so a client can read what it just wrote. Without a position it waits for the latest event in the store. Over HTTP, CommandApi.ReadAfter does the same for a request that sends the position back in a Read-After header.

There are three ways to read, from the most current to the most flexible:

Read	Up to date	For
state(id), which folds one aggregate’s events	Always	One aggregate, by id
A view model after caughtUp(position)	Includes the caller’s own commands	The screen after a command
A view model	A moment behind	Lists, searches, totals

effect-cqrs has no query handlers of its own. A view model writes to whatever storage you choose, and a query is your own code that reads it: a Ref, a SQL SELECT, a search index.

Failing handlers

A handler that fails stalls its view model on that event, which then retries: after 100 ms, then twice as long each time it fails at the same place, up to 30 seconds. Once it moves on, the next failure starts again from 100 ms. Events are never skipped or reordered.

viewModels.status lists each view model’s position and the failure it is retrying, if any. A stalled view model doesn’t catch up, so put a timeout on caughtUp where that matters.

Parking events

An event that can never be handled, a bug or data the handler doesn’t expect, stalls its view model for good. onFailure: ViewModel.park({ attempts }) gives up on an event after it has failed that many times: the view model records it in ParkedEvents, skips it from then on and carries on with the next one.

const TodoList = ViewModel.make({
  name: "todo-list",
  aggregates: [Todo],
  dispose: clearList,
  onFailure: ViewModel.park({ attempts: 5 }),
  handlers: { /* ... */ }
})

const viewModels = yield* ViewModel.ViewModels
yield* viewModels.parked("todo-list") // view model, position, event id, error, attempts, time
yield* viewModels.retryParked({ viewModel: "todo-list" }) // after fixing the handler

retryParked handles the parked events again, in order of position. Those that succeed leave ParkedEvents, the others stay with the new failure, and positions narrows it to some of them. A retried event comes after the events that followed it, so park only where order doesn’t matter between them, and a replay keeps skipping parked events until they are retried.

Stalling stays the default. A view model that skips an event is wrong from then on, in a way nothing shows, while a stalled one is visibly behind and right once it moves: a balance that missed a deposit is worse than one that is late. Parking is a choice per view model, for those where a gap is better than a stop.

Parked events are kept in memory unless ParkedEvents is provided. ParkedEvents.layerSql() keeps them in a table, cqrs_parked_events by default, on SQLite or PostgreSQL. It creates the table on startup unless migrate is false, and ParkedEvents.migrations() gives its migrations to run another way. viewModels.status counts each view model’s parked events, and the counter ViewModel.MetricNames.parkedEvents, with a view_model attribute, counts every event parked.

Replay

viewModels.replay({ viewModels: ["todo-list"] }) clears the checkpoint, runs dispose and rebuilds the view model from the log. startTime and finishTime limit which events go in: the view model then resumes from the present, and events outside the window stay out of it.

Moleculer-cqrs needed replay to recover events lost on the broker. Here view models can’t lose events, so replay is for rebuilding a view model that changed, or one that got damaged.

Personal data

Handlers receive personal fields decrypted. A view model that stores them handles onForget(subject, meta) to erase them when a subject is forgotten, see view models and personal data.

Checkpoints

A checkpoint has to live as long as its view model:

• A view model kept in memory pairs with Checkpoints.layerMemory, and rebuilds itself on every start.
• A view model stored in a database pairs with that database’s checkpoints, SqliteEventStore.layerCheckpoints() or PostgresEventStore.layerCheckpoints().

Several processes

Replicas that share a database would each run every view model, and race on its checkpoint. ViewModel.layer takes a lease on each view model before running it, from the Leases in context. Without one, every lease is held, which suits a single process.

Leases.layerPostgres() holds leases as PostgreSQL advisory locks, on a hash of the view model’s name, so only one process runs each view model. The others are standbys: they try to take over every 5 seconds, interval, and the one that does resumes from the checkpoint the holder saved. The locks live on one connection reserved from the client’s pool, so they go when the process dies or its connection does.

const Live = ViewModel.layer([TodoList]).pipe(
  Layer.provideMerge(Layer.mergeAll(
    PostgresEventStore.layer(),
    PostgresEventStore.layerCheckpoints(),
    Leases.layerPostgres()
  )),
  Layer.provide(PgClient.layer({ url: Redacted.make("postgres://localhost:5432/cqrs") }))
)

In a standby:

• viewModels.status shows holdsLease: false for the view models another process runs.
• caughtUp doesn’t wait for those view models, since nothing in this process moves them. It returns once the view models this process runs have caught up, so read your own writes where the view model runs, or wait on the read model itself.
• replay and retryParked fail with a ViewModelError: run them where the view model runs.

A holder notices that its connection is gone at its next check, up to interval later, and a standby may already have started by then. For that long, both can handle the same events, even in a transaction: a lease doesn’t fence off the old holder’s writes.

In an Effect cluster, Singleton from effect/cluster does the same job with the cluster’s own shard ownership, and needs no lease:

import { Singleton } from "effect/cluster"

const TodoListSingleton = Singleton.make("todo-list", Layer.launch(ViewModel.layer([TodoList])))

The view models then run on whichever runner owns the singleton, and ViewModels is only available inside it.

Keeping up

View models within 500 events of the end of the log share one read of it, and decode each event once. Commands give other fibers a turn every millisecond, so a burst of commands doesn’t starve the view models. A slow handler still falls behind without slowing the writers down, and catches up when they stop.

Other subscribers

Anything else that reacts to events can use store.subscribe(filter): a stream of past events, followed by live ones. A process that runs commands in reaction to events is better written as a saga, whose trigger is a view model.

A handler runs with the event it handles as the cause of the commands it runs. A subscriber gets the same with Causation.causedBy(stored).