Storage

The event store is an append-only log. Every event belongs to a stream, one aggregate of one type, and has two coordinates: a version that orders it in its stream, and a position that orders it across the whole store. Writers use the version for optimistic concurrency, and readers use the position as a cursor.

Store	For
EventStore.layerMemory	Tests, examples, the browser
SqliteEventStore	One machine, on a local file
PostgresEventStore	A server, possibly with several processes writing

In memory

EventStore.layerMemory keeps events in process memory. They don’t survive a restart.

SQLite

Provide any effect/sql SQLite client:

import { SqliteClient } from "@effect/sql-sqlite-node"
import { SqliteEventStore } from "effect-cqrs"

const Storage = Layer.merge(SqliteEventStore.layer(), SqliteEventStore.layerCheckpoints()).pipe(
  Layer.provide(SqliteClient.layer({ filename: "./data/events.sqlite" }))
)

Positions rely on SQLite having a single writer, so the schema doesn’t carry over to other databases.

An append is one statement, which checks the stream’s version and inserts the events only if it still matches. On a file, each commit waits for the disk, so once writers overlap, appends that become ready together are written in one transaction and share that wait. An append whose stream moved on still fails alone. A database in memory writes each append on its own.

Several processes can share one file. A store sees its own appends at once, and with pollInterval it also reads the head that often to notice the others’:

SqliteEventStore.layer({ pollInterval: "200 millis" })

PostgreSQL

Provide an effect/sql PostgreSQL client:

import { PgClient } from "@effect/sql-pg"
import { PostgresEventStore } from "effect-cqrs"

const Storage = Layer.merge(PostgresEventStore.layer(), PostgresEventStore.layerCheckpoints()).pipe(
  Layer.provide(PgClient.layer({ url: Redacted.make("postgres://localhost:5432/cqrs") }))
)

A sequence can’t number events: transactions commit in a different order from the one they drew their numbers in, so a view model that has read up to 6 would never come back for a 5 that commits later. Positions come from a counter row instead, which an append holds until it commits. Writers wait for each other at that one step, and positions follow commit order with no gaps.

So that writers don’t wait in line, a store writes the appends of its process in groups. Appends that arrive while one is being written go together into the next statement, which checks each stream’s version, numbers the events of the appends that pass and inserts them, in one round trip and one commit. An append whose stream moved on fails alone. When a writer in another process takes a stream first, the group is written again one append at a time.

Every append also sends a notification with the new head, on a channel named after the events table. On a PgClient from @effect/sql-pg the store listens to it, on a connection of its own, so view models and subscribers in every process wake up on every append.

• Behind a pooler that doesn’t keep sessions, such as PgBouncer in transaction mode, LISTEN doesn’t work. Set listen: false and a pollInterval. A client other than PgClient needs a pollInterval too.
• A pollInterval next to listening is a safety net. The store already reads the head again whenever it reconnects, so a notification sent while it was away isn’t lost.
• An append inside a transaction of your own, with sql.withTransaction, is written there and then, so it commits or rolls back with your other writes. It holds the counter row until that transaction ends, and every other writer of the store waits for it meanwhile, in this process and in others. Keep such transactions short, and do the slow work before them.
• After a bulk load, run ANALYZE. Until autovacuum gets to a new table, reads for view models are slow.

To run the tests on PostgreSQL, start the database of docker-compose.yml and say where it is. pnpm test then covers it too.

docker compose up -d --wait
POSTGRES_URL=postgres://postgres:postgres@localhost:5432/cqrs pnpm test

Migrations

By default, layer and layerCheckpoints create their tables on startup, and bring tables from an earlier version of effect-cqrs up to date: existing events get schema version 1, no causation and {} as metadata. The statements can run again without harm, and on PostgreSQL an advisory lock keeps processes that start together apart.

Where the application may not change the schema, set migrate: false and apply the migrations with your own tooling. migrations(options) lists them in order, as SQL with the table names filled in:

for (const { id, name, statements } of PostgresEventStore.migrations({ eventsTable: "orders_events" })) {
  writeFileSync(`migrations/${id}_${name}.sql`, statements.map((statement) => `${statement};\n`).join(""))
}

Migration	What it does
1 create_events	The events table and its index, and on PostgreSQL the counter row
2 add_causation	schema_version, correlation_id and causation_id
3 add_metadata	metadata
4 create_checkpoints	The checkpoints table

A database that effect-cqrs 0.2 created already has all four. One from 0.1 lacks 2 and 3. On PostgreSQL every statement checks first, but SQLite can’t add a column only if it is missing, so there its ALTER TABLE statements must run exactly once.

Options

Option	Stores	Default	What it does
eventsTable, checkpointsTable	Both	cqrs_events, cqrs_checkpoints	Table names, so several stores can share one database
migrate	Both	true	Creates and updates the tables on startup
pollInterval	Both	Off	Reads the head this often, for appends from other processes
listen	PostgreSQL	true	Listens for appends from other processes, on a PgClient

SqliteEventStore.layer({ eventsTable: "orders_events" })
SqliteEventStore.layerCheckpoints({ checkpointsTable: "orders_checkpoints" })

Your own backend

Another database needs an EventStore.Backend with three operations, and EventStore.make(backend, options) builds the store on top:

Operation	What it does
append(input)	Appends atomically, failing with ConcurrencyConflict if the stream moved past expectedVersion
readBatch(filter, afterPosition, limit)	Up to limit events past afterPosition that match the filter, by position
head	The position of the latest event, 0 when the store is empty
changes, optional	The head positions other processes write, as the database announces them

A backend stores each event’s schemaVersion, correlationId, causationId and metadata along with it. EventStore.toStored fills in the defaults: version 1, no causation and {}.

make runs changes and the pollInterval of its options in fibers that need a Scope, which Layer.effect provides. If changes fails, the store logs it and runs it again a second later, so it should start with the current head.

Positions must follow commit order, so a backend on a database with concurrent writers has to guarantee that itself. EventStore.groupCommit(appendGroup) gives a backend group commit: it calls appendGroup with appends to different streams, to write in one transaction, and expects an exit for each. The store tests in test/EventStore.test.ts run against every backend and show what each must do.