PostgresEventStore
Durable storage on PostgreSQL, through any effect/sql PostgreSQL driver.
A sequence can’t number events here. 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.
Because of that row, appends from one process are written in groups: those that arrive while a group is being written go together into the next one, a single statement that checks each stream’s version, takes positions for the appends that pass and inserts their events. An append that fails its check fails alone.
An append inside a transaction of the caller’s is written there and then, and holds the counter row until that transaction ends: every other writer of the store waits for it meanwhile, so such transactions should be short.
The statement also sends a notification with the new head on a channel named
after the events table. A store on a PgClient listens to it, so view models in
every process wake up on every append.
Interfaces
Section titled “Interfaces”Options
Section titled “Options”Defined in: src/PostgresEventStore.ts:42
Properties
Section titled “Properties”checkpointsTable?
Section titled “checkpointsTable?”
readonlyoptionalcheckpointsTable?:string
Defined in: src/PostgresEventStore.ts:44
eventsTable?
Section titled “eventsTable?”
readonlyoptionaleventsTable?:string
Defined in: src/PostgresEventStore.ts:43
listen?
Section titled “listen?”
readonlyoptionallisten?:boolean
Defined in: src/PostgresEventStore.ts:57
Listens for appends from other processes when the SqlClient is a PgClient
from @effect/sql-pg. Listening holds a connection. Defaults to true. Turn it
off behind a pooler that doesn’t keep sessions, such as PgBouncer in transaction
mode, and set pollInterval instead.
migrate?
Section titled “migrate?”
readonlyoptionalmigrate?:boolean
Defined in: src/PostgresEventStore.ts:50
Creates the tables, or brings tables of an earlier version up to date, on
startup. Defaults to true. Set it to false where the application may not
change the schema, and apply migrations with your own tooling.
pollInterval?
Section titled “pollInterval?”
readonlyoptionalpollInterval?:Input
Defined in: src/PostgresEventStore.ts:59
Reads the head this often, see EventStore.MakeOptions. Off by default.
Type Aliases
Section titled “Type Aliases”Migration
Section titled “Migration”Migration =
Migration
Defined in: src/PostgresEventStore.ts:63
A schema change, see EventStore.Migration. Migrations are listed in the order to apply them.
Variables
Section titled “Variables”
constmake: (…args) =>Effect<{afterCommit?: <R>(effect) =>Effect<void,never,R>;changes:Stream<number>;head:Effect<number,EventStoreError>;append(input):Effect<readonlyobject[],EventStoreError|ConcurrencyConflict>;read(filter?):Stream<{aggregateId:string;aggregateName:string;causationId:string|null;correlationId:string|null;metadata: {[key:string]:Json; };payload:Json;position:number;schemaVersion:number;timestamp:Utc;type:string;version:number; },EventStoreError>;subscribe(filter?):Stream<{aggregateId:string;aggregateName:string;causationId:string|null;correlationId:string|null;metadata: {[key:string]:Json; };payload:Json;position:number;schemaVersion:number;timestamp:Utc;type:string;version:number; },EventStoreError>; },EventStoreError,Scope|SqlClient>
Defined in: src/PostgresEventStore.ts:222
Parameters
Section titled “Parameters”…[options?: Options]
Returns
Section titled “Returns”Effect<{ afterCommit?: <R>(effect) => Effect<void, never, R>; changes: Stream<number>; head: Effect<number, EventStoreError>; append(input): Effect<readonly object[], EventStoreError | ConcurrencyConflict>; read(filter?): Stream<{ aggregateId: string; aggregateName: string; causationId: string | null; correlationId: string | null; metadata: {[key: string]: Json; }; payload: Json; position: number; schemaVersion: number; timestamp: Utc; type: string; version: number; }, EventStoreError>; subscribe(filter?): Stream<{ aggregateId: string; aggregateName: string; causationId: string | null; correlationId: string | null; metadata: {[key: string]: Json; }; payload: Json; position: number; schemaVersion: number; timestamp: Utc; type: string; version: number; }, EventStoreError>; }, EventStoreError, Scope | SqlClient>
Functions
Section titled “Functions”layer()
Section titled “layer()”layer(
options?):Layer<EventStore,EventStoreError,SqlClient>
Defined in: src/PostgresEventStore.ts:408
Parameters
Section titled “Parameters”options?
Section titled “options?”Returns
Section titled “Returns”Layer<EventStore, EventStoreError, SqlClient>
layerCheckpoints()
Section titled “layerCheckpoints()”layerCheckpoints(
options?):Layer<Checkpoints,EventStoreError,SqlClient>
Defined in: src/PostgresEventStore.ts:414
Keeps checkpoints in the same database, for view models that are durable too.
Parameters
Section titled “Parameters”options?
Section titled “options?”Returns
Section titled “Returns”Layer<Checkpoints, EventStoreError, SqlClient>
migrations()
Section titled “migrations()”migrations(
options?): readonlyMigration[]
Defined in: src/PostgresEventStore.ts:142
The schema of the events and checkpoints tables, for tools that apply migrations
themselves. layer and layerCheckpoints run the same statements on startup
unless migrate is false.
Parameters
Section titled “Parameters”options?
Section titled “options?”Returns
Section titled “Returns”readonly Migration[]