EventStore

The append-only log every aggregate and view model is built on.

Each event belongs to a stream, identified by aggregate name and id, and has two coordinates: a version that orders it inside its stream and a position that orders it across the whole store. Writers use the version for optimistic concurrency, readers use the position as a durable cursor.

Classes

ConcurrencyConflict

Defined in: src/EventStore.ts:43

Extends

• object & YieldableError<this>

Constructors

Constructor

new ConcurrencyConflict(…args): ConcurrencyConflict

Defined in: node_modules/.pnpm/effect@4.0.6/node_modules/effect/dist/Schema.d.ts:10979

Parameters

args

…[props: object, options?: MakeOptions]

Returns

ConcurrencyConflict

Inherited from

Schema.TaggedError<ConcurrencyConflict>()("ConcurrencyConflict", { aggregateName: Schema.String, aggregateId: Schema.String, expectedVersion: Schema.Int }).constructor

Properties

aggregateId

readonly aggregateId: string = Schema.String

Defined in: src/EventStore.ts:45

Inherited from

Schema.TaggedError<ConcurrencyConflict>()("ConcurrencyConflict", { aggregateName: Schema.String, aggregateId: Schema.String, expectedVersion: Schema.Int }).aggregateId

aggregateName

readonly aggregateName: string = Schema.String

Defined in: src/EventStore.ts:44

Inherited from

Schema.TaggedError<ConcurrencyConflict>()("ConcurrencyConflict", { aggregateName: Schema.String, aggregateId: Schema.String, expectedVersion: Schema.Int }).aggregateName

expectedVersion

readonly expectedVersion: number = Schema.Int

Defined in: src/EventStore.ts:46

Inherited from

Schema.TaggedError<ConcurrencyConflict>()("ConcurrencyConflict", { aggregateName: Schema.String, aggregateId: Schema.String, expectedVersion: Schema.Int }).expectedVersion

---

EventStore

Defined in: src/EventStore.ts:139

Extends

• Shape<"effect-cqrs/EventStore", { 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>; }, this>

Constructors

Constructor

new EventStore(_): EventStore

Defined in: node_modules/.pnpm/effect@4.0.6/node_modules/effect/dist/Context.d.ts:121

Parameters

_

never

Returns

EventStore

Inherited from

Context.Service<EventStore, { /** Appends atomically, failing if the stream moved past `expectedVersion`. */ append(input: AppendInput): Effect.Effect<ReadonlyArray<StoredEvent>, ConcurrencyConflict | EventStoreError> /** Reads the events stored so far, ordered by position, then completes. */ read(filter?: EventFilter): Stream.Stream<StoredEvent, EventStoreError> /** Reads the events stored so far, then keeps following new ones. Never completes. */ subscribe(filter?: EventFilter): Stream.Stream<StoredEvent, EventStoreError> /** The position of the latest event, `0` when the store is empty. */ readonly head: Effect.Effect<number, EventStoreError> /** * Emits the current latest position, then each newer one. Consumers can treat an * element as “there may be something new”. Appends from this process are always * observed, appends from other processes when the backend announces them or the * store polls for them: see `MakeOptions`. */ readonly changes: Stream.Stream<number> /** * Runs an effect once what was just appended can be read by everyone: at once, * or when the caller’s transaction commits, and never if it rolls back. Services * that keep what they read or wrote in memory go through it, so they don’t keep * what a rollback undid. Missing on stores without transactions. */ readonly afterCommit?: (<R>(effect: Effect.Effect<void, never, R>) => Effect.Effect<void, never, R>) | undefined }>()(“effect-cqrs/EventStore”).constructor

---

EventStoreError

Defined in: src/EventStore.ts:38

Extends

• object & YieldableError<this>

Constructors

Constructor

new EventStoreError(…args): EventStoreError

Defined in: node_modules/.pnpm/effect@4.0.6/node_modules/effect/dist/Schema.d.ts:10979

Parameters

args

…[props: object, options?: MakeOptions]

Returns

EventStoreError

Inherited from

Schema.TaggedError<EventStoreError>()("EventStoreError", { operation: Schema.String, cause: Schema.Defect() }).constructor

Properties

cause

cause: unknown

Defined in: src/EventStore.ts:40

Inherited from

Schema.TaggedError<EventStoreError>()("EventStoreError", { operation: Schema.String, cause: Schema.Defect() }).cause

operation

readonly operation: string = Schema.String

Defined in: src/EventStore.ts:39

Inherited from

Schema.TaggedError<EventStoreError>()("EventStoreError", { operation: Schema.String, cause: Schema.Defect() }).operation

Interfaces

AppendInput

Defined in: src/EventStore.ts:75

Properties

aggregateId

readonly aggregateId: string

Defined in: src/EventStore.ts:77

aggregateName

readonly aggregateName: string

Defined in: src/EventStore.ts:76

causationId?

readonly optional causationId?: string

Defined in: src/EventStore.ts:82

correlationId?

readonly optional correlationId?: string

Defined in: src/EventStore.ts:81

events

readonly events: readonly NewEvent[]

Defined in: src/EventStore.ts:80

expectedVersion

readonly expectedVersion: number

Defined in: src/EventStore.ts:79

The stream version the writer based its decision on, 0 for a new stream.

metadata?

readonly optional metadata?: JsonObject

Defined in: src/EventStore.ts:84

Recorded on every event of the append. Defaults to {}.

---

Backend

Defined in: src/EventStore.ts:165

What a storage adapter implements. make builds the rest on top.

Properties

afterCommit?

readonly optional afterCommit?: <R>(effect) => Effect<void, never, R>

Defined in: src/EventStore.ts:185

Runs an effect once an append made now can be read by everyone: at once, or when the caller’s transaction commits, and never if it rolls back. SqlClient.afterCommit is one. Without it appends are announced as soon as they return, which makes subscribers skip those of a transaction that is still open.

Type Parameters

R

R

Parameters

effect

Effect<void, never, R>

Returns

Effect<void, never, R>

append

readonly append: (input) => Effect<readonly object[], EventStoreError | ConcurrencyConflict>

Defined in: src/EventStore.ts:166

Appends atomically, failing if the stream moved past expectedVersion.

Parameters

input

AppendInput

Returns

Effect<readonly object[], EventStoreError | ConcurrencyConflict>

changes?

readonly optional changes?: Stream<number, EventStoreError, never>

Defined in: src/EventStore.ts:178

Head positions written by other processes, as the database announces them. It should start with the head once it listens, so nothing between two runs is missed. When it fails or ends, the store logs it and runs it again a second later.

head

readonly head: Effect<number, EventStoreError>

Defined in: src/EventStore.ts:172

readBatch

readonly readBatch: (filter, afterPosition, limit) => Effect<readonly object[], EventStoreError>

Defined in: src/EventStore.ts:167

Parameters

filter

EventFilter

afterPosition

number

limit

number

Returns

Effect<readonly object[], EventStoreError>

---

EventFilter

Defined in: src/EventStore.ts:121

Selects events by stream, type and time. Omitted fields match everything. startTime is inclusive and finishTime is exclusive, so adding one millisecond to a history timestamp selects up to and including that event.

Properties

afterPosition?

readonly optional afterPosition?: number

Defined in: src/EventStore.ts:127

afterVersion?

readonly optional afterVersion?: number

Defined in: src/EventStore.ts:132

For a filter on one stream: where to continue in it. Versions follow the stream’s own index, so resuming there costs the same however far the log has moved on.

aggregateIds?

readonly optional aggregateIds?: readonly string[]

Defined in: src/EventStore.ts:123

aggregateNames?

readonly optional aggregateNames?: readonly string[]

Defined in: src/EventStore.ts:122

eventTypes?

readonly optional eventTypes?: readonly string[]

Defined in: src/EventStore.ts:124

finishTime?

readonly optional finishTime?: Utc

Defined in: src/EventStore.ts:126

startTime?

readonly optional startTime?: Utc

Defined in: src/EventStore.ts:125

---

MakeOptions

Defined in: src/EventStore.ts:188

Properties

batchSize?

readonly optional batchSize?: number

Defined in: src/EventStore.ts:190

Events per read. Defaults to 500.

pollInterval?

readonly optional pollInterval?: Input

Defined in: src/EventStore.ts:195

Reads the head this often, to notice appends from other processes when the backend doesn’t announce them, or as a safety net when it does. Off by default.

---

Migration

Defined in: src/EventStore.ts:32

A schema change for a SQL table, with the table names filled in. Each module that keeps a table exports its migrations in the order to apply them, for teams that run them with their own tooling and turn off migrate.

Properties

id

readonly id: number

Defined in: src/EventStore.ts:33

name

readonly name: string

Defined in: src/EventStore.ts:34

statements

readonly statements: readonly string[]

Defined in: src/EventStore.ts:35

---

NewEvent

Defined in: src/EventStore.ts:68

Properties

payload

readonly payload: Json

Defined in: src/EventStore.ts:70

schemaVersion?

readonly optional schemaVersion?: number

Defined in: src/EventStore.ts:72

Defaults to 1.

type

readonly type: string

Defined in: src/EventStore.ts:69

Type Aliases

StoredEvent

StoredEvent = typeof StoredEvent.Type

Defined in: src/EventStore.ts:49

Variables

groupCommit

const groupCommit: (…args) => Effect<(input) => Effect<readonly object[], EventStoreError | ConcurrencyConflict>, never, Scope>

Defined in: src/EventStore.ts:331

Group commit for a backend. An append that finds nothing being written goes straight through appendGroup, in its own fiber. Appends that arrive meanwhile wait, then go together into the next call, which writes them in one transaction and returns an exit for each, in order: however many wait, they pay for one commit.

A group holds at most one append per stream, so they can’t conflict with each other, and at most maxEvents events unless a single append has more. An append left out waits for the next group, where it comes first, so appends to a stream are written in the order they came. Groups of waiting appends are written by a fiber of the Scope. An append interrupted while it waits is not written, while one interrupted once its group is being written may be: a caller that times out should retry with a commandId, which finds the first result if there is one. The others in its group are written and answered as if nothing happened.

Parameters

args

…[appendGroup: (inputs) => Effect<readonly AppendExit[]>, options?: object]

Returns

Effect<(input) => Effect<readonly object[], EventStoreError | ConcurrencyConflict>, never, Scope>

---

layerMemory

const layerMemory: Layer.Layer<EventStore, EventStoreError>

Defined in: src/EventStore.ts:540

Keeps events in process memory. They do not survive a restart.

---

make

const make: (…args) => 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>

Defined in: src/EventStore.ts:208

Builds a store on a backend. Watching for appends from other processes takes fibers that live as long as the Scope, which Layer.effect provides.

Parameters

args

…[backend: Backend, options?: MakeOptions]

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>

---

makeMemory

const makeMemory: Effect.Effect<EventStore["Service"], EventStoreError>

Defined in: src/EventStore.ts:487

---

StoredEvent

const StoredEvent: Struct<{ aggregateId: String; aggregateName: String; causationId: NullOr<String>; correlationId: NullOr<String>; metadata: $Record<String, Codec<Json, Json, never, never>>; payload: Codec<Json, Json, never, never>; position: Int; schemaVersion: Int; timestamp: DateTimeUtcFromMillis; type: String; version: Int; }>

Defined in: src/EventStore.ts:49

Functions

eventId()

eventId(event): string

Defined in: src/EventStore.ts:92

Identifies an event wherever it is stored, as the causationId of what it causes: account/alice@4. Name and id are URI-encoded, so a / or @ inside one can’t make two events look alike.

Parameters

event

Pick<StoredEvent, "aggregateName" | "aggregateId" | "version">

Returns

string

---

isOneStream()

isOneStream(__namedParameters): boolean

Defined in: src/EventStore.ts:136

A filter on exactly one stream, which a store can read through that stream’s index.

Parameters

__namedParameters

EventFilter

Returns

boolean

---

toStored()

toStored(input, event, index, position, timestamp): object

Defined in: src/EventStore.ts:96

The columns every adapter stores for a new event, with the defaults filled in.

Parameters

input

AppendInput

event

NewEvent

index

number

position

number

timestamp

Utc

Returns

aggregateId

readonly aggregateId: string = Schema.String

aggregateName

readonly aggregateName: string = Schema.String

causationId

readonly causationId: string | null

What directly caused the command that recorded the event: a request, a command or another event.

correlationId

readonly correlationId: string | null

Shared by everything that follows from one original request.

metadata

readonly metadata: object = Schema.JsonObject

What the command carried beside its payload, see Metadata. Events from before metadata existed have {}.

Index Signature

[key: string]: Json

payload

readonly payload: Json = Schema.Json

position

readonly position: number = Schema.Int

schemaVersion

readonly schemaVersion: number = Schema.Int

The version of the event’s schema the payload was written with. Events from before versions existed are 1.

timestamp

readonly timestamp: Utc = Schema.DateTimeUtcFromMillis

type

readonly type: string = Schema.String

version

readonly version: number = Schema.Int