AggregateService

Runs an aggregate against the EventStore: the command, read-model and history actions of moleculer-cqrs, as a typed service.

Classes

CommandIdReused

Defined in: src/AggregateService.ts:83

A commandId that already identifies another command on the aggregate. Running a command again returns the first result only for the same command: a client that reuses an id for another one gets this rather than a result it didn’t ask for.

Extends

object & YieldableError<this>

Constructors

Constructor

new CommandIdReused(…args): CommandIdReused

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

Parameters

args

…[props: object, options?: MakeOptions]

Returns

CommandIdReused

Inherited from

Schema.TaggedError<CommandIdReused>()("CommandIdReused", { aggregateName: Schema.String, aggregateId: Schema.String, commandId: Schema.String, /** The command that ran with the id first. */ previous: Schema.String, /** The command that came with it now. */ command: Schema.String }).constructor

Properties

aggregateId

readonly aggregateId: string = Schema.String

Defined in: src/AggregateService.ts:85

Inherited from

aggregateName

readonly aggregateName: string = Schema.String

Defined in: src/AggregateService.ts:84

Inherited from

command

readonly command: string = Schema.String

Defined in: src/AggregateService.ts:90

The command that came with it now.

Inherited from

commandId

readonly commandId: string = Schema.String

Defined in: src/AggregateService.ts:86

Inherited from

readonly previous: string = Schema.String

Defined in: src/AggregateService.ts:88

The command that ran with the id first.

Inherited from

EventDecodeError

Defined in: src/AggregateService.ts:57

A stored event doesn’t match the schema of its version, or was written in a version the aggregate doesn’t declare. Old shapes are declared with EventVersions.

Extends

object & YieldableError<this>

Constructors

Constructor

new EventDecodeError(…args): EventDecodeError

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

Parameters

args

…[props: object, options?: MakeOptions]

Returns

EventDecodeError

Inherited from

Schema.TaggedError<EventDecodeError>()("EventDecodeError", { aggregateName: Schema.String, aggregateId: Schema.String, version: Schema.Int, type: Schema.String, schemaVersion: Schema.Int, message: Schema.String }).constructor

Properties

aggregateId

readonly aggregateId: string = Schema.String

Defined in: src/AggregateService.ts:59

Inherited from

aggregateName

readonly aggregateName: string = Schema.String

Defined in: src/AggregateService.ts:58

Inherited from

message

message: string = Schema.String

Defined in: src/AggregateService.ts:63

Inherited from

schemaVersion

readonly schemaVersion: number = Schema.Int

Defined in: src/AggregateService.ts:62

Inherited from

type

readonly type: string = Schema.String

Defined in: src/AggregateService.ts:61

Inherited from

version

readonly version: number = Schema.Int

Defined in: src/AggregateService.ts:60

Inherited from

InvalidCommand

Defined in: src/AggregateService.ts:47

Extends

object & YieldableError<this>

Constructors

Constructor

new InvalidCommand(…args): InvalidCommand

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

Parameters

args

…[props: object, options?: MakeOptions]

Returns

InvalidCommand

Inherited from

Schema.TaggedError<InvalidCommand>()("InvalidCommand", { aggregateName: Schema.String, type: Schema.String, message: Schema.String }).constructor

Properties

aggregateName

readonly aggregateName: string = Schema.String

Defined in: src/AggregateService.ts:48

Inherited from

Schema.TaggedError<InvalidCommand>()("InvalidCommand", { aggregateName: Schema.String, type: Schema.String, message: Schema.String }).aggregateName

message

message: string = Schema.String

Defined in: src/AggregateService.ts:50

Inherited from

Schema.TaggedError<InvalidCommand>()("InvalidCommand", { aggregateName: Schema.String, type: Schema.String, message: Schema.String }).message

type

readonly type: string = Schema.String

Defined in: src/AggregateService.ts:49

Inherited from

Schema.TaggedError<InvalidCommand>()("InvalidCommand", { aggregateName: Schema.String, type: Schema.String, message: Schema.String }).type

UnknownCommand

Defined in: src/AggregateService.ts:42

Extends

object & YieldableError<this>

Constructors

Constructor

new UnknownCommand(…args): UnknownCommand

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

Parameters

args

…[props: object, options?: MakeOptions]

Returns

UnknownCommand

Inherited from

Schema.TaggedError<UnknownCommand>()("UnknownCommand", { aggregateName: Schema.String, type: Schema.String }).constructor

Properties

aggregateName

readonly aggregateName: string = Schema.String

Defined in: src/AggregateService.ts:43

Inherited from

Schema.TaggedError<UnknownCommand>()("UnknownCommand", { aggregateName: Schema.String, type: Schema.String }).aggregateName

type

readonly type: string = Schema.String

Defined in: src/AggregateService.ts:44

Inherited from

Schema.TaggedError<UnknownCommand>()("UnknownCommand", { aggregateName: Schema.String, type: Schema.String }).type

WrongExpectedVersion

Defined in: src/AggregateService.ts:71

The caller’s expectedVersion is not the aggregate’s version: someone else changed it since the caller read it. Unlike a ConcurrencyConflict, this is not retried, as deciding again would overwrite a change the caller hasn’t seen.

Extends

object & YieldableError<this>

Constructors

Constructor

new WrongExpectedVersion(…args): WrongExpectedVersion

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

Parameters

args

…[props: object, options?: MakeOptions]

Returns

WrongExpectedVersion

Inherited from

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

Properties

actualVersion

readonly actualVersion: number = Schema.Int

Defined in: src/AggregateService.ts:75

Inherited from

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

aggregateId

readonly aggregateId: string = Schema.String

Defined in: src/AggregateService.ts:73

Inherited from

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

aggregateName

readonly aggregateName: string = Schema.String

Defined in: src/AggregateService.ts:72

Inherited from

expectedVersion

readonly expectedVersion: number = Schema.Int

Defined in: src/AggregateService.ts:74

Inherited from

Interfaces

AggregateService

Defined in: src/AggregateService.ts:164

Type Parameters

A

A extends Any

Properties

aggregate

readonly aggregate: A

Defined in: src/AggregateService.ts:165

commands

readonly commands: { readonly [K in string]: (aggregateId: string, payload: Payload<A, K>, options?: CommandOptions) => Effect<CommandResult<Event<A>>, CommandError<A, K> | CommandFailure> }

Defined in: src/AggregateService.ts:167

One typed function per command.

dispatch

readonly dispatch: (input) => Effect<CommandResult<Event<A>>, UnknownCommand | CommandFailure | CommandError<A, CommandName<A>>>

Defined in: src/AggregateService.ts:179

Runs a { aggregateId, type, payload } envelope received as JSON from outside the process. It can also carry commandId and expectedVersion, see CommandOptions, correlationId, causationId and metadata.

Parameters

input

unknown

Returns

Effect<CommandResult<Event<A>>, UnknownCommand | CommandFailure | CommandError<A, CommandName<A>>>

history

readonly history: (aggregateId, options?) => Effect<readonly HistoryEntry<Event<A>>[], ReadError>

Defined in: src/AggregateService.ts:200

Lists the events of one aggregate, with their decoded payload when payload is set. Without it the events aren’t decoded. afterVersion and limit read a long stream in pages: pass the version of the last entry to get the next page.

Parameters

aggregateId

string

options?

HistoryOptions

Returns

Effect<readonly HistoryEntry<Event<A>>[], ReadError>

readModel

readonly readModel: (aggregateId, options?) => Effect<State<A>, ReadError>

Defined in: src/AggregateService.ts:194

Parameters

aggregateId

string

options?

finishTime?

Utc

Returns

Effect<State<A>, ReadError>

state

readonly state: (aggregateId, options?) => Effect<State<A>, ReadError>

Defined in: src/AggregateService.ts:186

Folds the aggregate’s events into its state, optionally as it was before finishTime. It is the state the next command is decided on, so it is never behind.

Parameters

aggregateId

string

options?

finishTime?

Utc

Returns

Effect<State<A>, ReadError>

CommandOptions

Defined in: src/AggregateService.ts:109

What a typed command takes besides the aggregate id and the payload.

Properties

commandId?

readonly optional commandId?: string

Defined in: src/AggregateService.ts:115

Identifies the command, so that running it again on the same aggregate returns the first result instead of deciding again. The payload isn’t compared, so an id belongs to one request. See commandIdKey.

expectedVersion?

readonly optional expectedVersion?: number

Defined in: src/AggregateService.ts:117

The version the caller based its command on. If the aggregate is at another, the command fails with WrongExpectedVersion.

CommandResult

Defined in: src/AggregateService.ts:132

Type Parameters

Event

Event

Properties

aggregateId

readonly aggregateId: string

Defined in: src/AggregateService.ts:134

aggregateName

readonly aggregateName: string

Defined in: src/AggregateService.ts:133

correlationId

readonly correlationId: string

Defined in: src/AggregateService.ts:141

Recorded on the events, see Causation.

events

readonly events: readonly Event[]

Defined in: src/AggregateService.ts:139

position

readonly position: number

Defined in: src/AggregateService.ts:138

The store position of the last event, to pass to ViewModels.caughtUp.

version

readonly version: number

Defined in: src/AggregateService.ts:136

The stream version after the command.

HistoryEntry

Defined in: src/AggregateService.ts:144

Type Parameters

Event

Event

Properties

causationId

readonly causationId: string | null

Defined in: src/AggregateService.ts:149

correlationId

readonly correlationId: string | null

Defined in: src/AggregateService.ts:148

metadata

readonly metadata: JsonObject

Defined in: src/AggregateService.ts:150

payload?

readonly optional payload?: Event

Defined in: src/AggregateService.ts:151

timestamp

readonly timestamp: Utc

Defined in: src/AggregateService.ts:146

type

readonly type: string

Defined in: src/AggregateService.ts:147

version

readonly version: number

Defined in: src/AggregateService.ts:145

HistoryOptions

Defined in: src/AggregateService.ts:154

Properties

afterVersion?

readonly optional afterVersion?: number

Defined in: src/AggregateService.ts:159

Lists the events after this version. Defaults to 0, the start of the stream.

finishTime?

readonly optional finishTime?: Utc

Defined in: src/AggregateService.ts:157

limit?

readonly optional limit?: number

Defined in: src/AggregateService.ts:161

Lists at most this many events.

payload?

readonly optional payload?: boolean

Defined in: src/AggregateService.ts:155

startTime?

readonly optional startTime?: Utc

Defined in: src/AggregateService.ts:156

Options

Defined in: src/AggregateService.ts:406

Properties

commandIdsPerStream?

readonly optional commandIdsPerStream?: number

Defined in: src/AggregateService.ts:415

How many of its latest command ids each aggregate remembers, so that a command run again with one of them returns its first result. Defaults to 100.

maxRetries?

readonly optional maxRetries?: number

Defined in: src/AggregateService.ts:408

How many times a command is re-decided after losing a race. Defaults to 5.

stateCacheSize?

readonly optional stateCacheSize?: number

Defined in: src/AggregateService.ts:410

How many aggregates keep their state in memory between commands. Defaults to 10,000.

SnapshotOptions

Defined in: src/AggregateService.ts:393

Saves the state of each aggregate every so often, so a cold start replays only what came after.

Type Parameters

State

State

Properties

every

readonly every: number

Defined in: src/AggregateService.ts:395

How many events to fold after a snapshot before saving the next one.

schema

readonly schema: Codec<State, any, never, never>

Defined in: src/AggregateService.ts:397

How the state is stored. It is encoded as JSON.

version?

readonly optional version?: string | number

Defined in: src/AggregateService.ts:403

Part of every key. Change it whenever the state or the projection changes, so that snapshots of the old state are ignored and rebuilt from the log. Defaults to 1.

Type Aliases

CommandFailure

CommandFailure = ConcurrencyConflict | InvalidCommand | WrongExpectedVersion | ReadError | SubjectForgotten | CommandIdReused

Defined in: src/AggregateService.ts:100

How a command can fail, on top of the errors its own handler declares. SubjectForgotten means an event would record personal data of a forgotten subject.

ReadError

ReadError = EventStoreError | EventDecodeError | PersonalDataError

Defined in: src/AggregateService.ts:94

How reading events can fail. PersonalDataError comes from aggregates with personal fields, see PersonalData.

Variables

commandIdKey

const commandIdKey: "cqrs.commandId" = "cqrs.commandId"

Defined in: src/AggregateService.ts:124

The metadata key that records the commandId on the events a command writes. It is reserved: metadata given under this key is dropped.

commandTypeKey

const commandTypeKey: "cqrs.command" = "cqrs.command"

Defined in: src/AggregateService.ts:130

The metadata key that records, beside the commandIdKey, which command wrote the events. It is reserved too.

make

const make: {<A>(aggregate, options): Effect<AggregateService<A>, never, EventStore | KeyValueStore | Scope | Context<A>>; <A>(aggregate, options?): Effect<AggregateService<A>, never, EventStore | Context<A>>; }

Defined in: src/AggregateService.ts:1187

Builds the service for one aggregate. With snapshots it saves states through a KeyValueStore from effect/persistence, which has layers for memory, files, SQL and the browser’s localStorage. Saves run in the background, and closing the scope waits for the ones still running.

Call Signature

<A>(aggregate, options): Effect<AggregateService<A>, never, EventStore | KeyValueStore | Scope | Context<A>>

Type Parameters

A

A extends Any

Parameters

aggregate

A

options

Options & object

Returns

Effect<AggregateService<A>, never, EventStore | KeyValueStore | Scope | Context<A>>

Call Signature

<A>(aggregate, options?): Effect<AggregateService<A>, never, EventStore | Context<A>>

Type Parameters

A

A extends Any

Parameters

aggregate

A

options?

Options

Returns

Effect<AggregateService<A>, never, EventStore | Context<A>>

MetricNames

const MetricNames: object

Defined in: src/AggregateService.ts:460

The names of the metrics every aggregate service records through Effect’s Metric module. Each series carries an aggregate attribute with the aggregate’s name. Read them with Metric.snapshot, or send them on with an exporter such as OtlpMetrics from effect/observability.

Type Declaration

commandDuration

readonly commandDuration: "cqrs_command_duration" = "cqrs_command_duration"

Timer: how long a command took, including its wait for an earlier command on the same stream, by command and outcome. The outcome is success, the tag of the error, defect or interrupted. A payload that doesn’t decode, or a type the aggregate doesn’t know, fails before this point and isn’t recorded.

concurrencyConflicts

readonly concurrencyConflicts: "cqrs_concurrency_conflicts_total" = "cqrs_concurrency_conflicts_total"

Counter: appends that lost a race with a writer in another process. Each is retried until maxRetries runs out.

eventsReplayed

readonly eventsReplayed: "cqrs_events_replayed" = "cqrs_events_replayed"

Histogram: how many events a load read from the store on top of that state.

snapshotReadFailures

readonly snapshotReadFailures: "cqrs_snapshot_read_failures_total" = "cqrs_snapshot_read_failures_total"

Counter: snapshots that couldn’t be read, so the stream was replayed instead.

snapshotSaves

readonly snapshotSaves: "cqrs_snapshot_saves_total" = "cqrs_snapshot_saves_total"

Counter: snapshots saved, by result: success or failure.

stateCacheEvictions

readonly stateCacheEvictions: "cqrs_state_cache_evictions_total" = "cqrs_state_cache_evictions_total"

Counter: states dropped from memory to make room for others, see stateCacheSize.

stateCacheSize

readonly stateCacheSize: "cqrs_state_cache_size" = "cqrs_state_cache_size"

Gauge: how many states are remembered in memory.

stateLoads

readonly stateLoads: "cqrs_state_loads_total" = "cqrs_state_loads_total"

Counter: where loading the current state found a state to start from, by source: memory, snapshot, or replay when it starts from the first event.

Functions

makeEventDecoder()

makeEventDecoder<A>(aggregate, keys?): (stored) => Effect<Event<A>, EventDecodeError | PersonalDataError>

Defined in: src/AggregateService.ts:328

Decodes a stored event with the schema of the aggregate it belongs to. An event stored in an older version is upcast to the current one on the way. An aggregate with personal fields needs keys to decrypt them, see PersonalData.

Type Parameters

A

A extends Any

Parameters

aggregate

A

keys?

cipher

Cipher

What encrypts and decrypts with the keys.

decryptionKey

(subject, keyId) => Effect<Option<CipherKey>, PersonalDataError>

The key to decrypt with, None when the subject was forgotten. A subject without any key fails: losing the key store must not pass for forgetting.

encryptionKey

(subject) => Effect<DataKey, PersonalDataError | SubjectForgotten>

The key to encrypt with, created when the subject has none. Fails for a forgotten subject.

forget

(subject) => Effect<void, PersonalDataError>

Destroys the key of a subject. Prefer forget, which also records it in the log.

isForgotten

(subject) => Effect<boolean, PersonalDataError>

rewrap

(options?) => Effect<RewrapResult, PersonalDataError>

See rewrap.

Returns

(stored) => Effect<Event<A>, EventDecodeError | PersonalDataError>

makeEventEncoder()

makeEventEncoder<A>(aggregate, keys?): (event, aggregateId?) => Effect<NewEvent, PersonalDataError | SubjectForgotten>

Defined in: src/AggregateService.ts:383

Encodes events into the JSON form the store keeps, in the current version of their schema. An aggregate with personal fields needs keys to encrypt them, and the id of the stream the event goes to, the default subject.