EventStore

@delta-base/server

---

Class: EventStore

Implementation of the EventStore interface for DeltaBase

Implements

• EventStore

Constructors

Constructor

new EventStore(http, eventStoreName, options?): EventStore

Creates a new EventStore client instance

Parameters

http

HttpClient

The HTTP client to use for API requests

eventStoreName

string

The name of the event store to interact with

options?

EventStoreOptions

Optional configuration including inline projections

Returns

EventStore

Methods

aggregateStream()

aggregateStream<State, EventType>(streamId, options): Promise<AggregateStreamResult<State>>

Aggregate events from a stream and compute a state

Type Parameters

State

State

EventType

EventType extends Readonly<{ data: EventData; metadata?: PlatformEventMetadata; tags?: string[]; type: string; }>

Parameters

streamId

string

The ID of the stream to aggregate events from

options

AggregateStreamOptions<State, EventType>

Configuration options for the aggregation process

Returns

Promise<AggregateStreamResult<State>>

Promise resolving to the aggregation result with the computed state and stream metadata

Example

// Define your state type and event types
type UserState = { email: string, isVerified: boolean };
type UserEvent =
  | { type: 'user.created', data: { email: string } }
  | { type: 'user.verified', data: { verifiedAt: string } };

// Aggregate the stream into a state
const result = await eventStore.aggregateStream<UserState, UserEvent>(
  'user-123',
  {
    initialState: () => ({ email: '', isVerified: false }),
    evolve: (state, event) => {
      switch (event.type) {
        case 'user.created':
          return { ...state, email: event.data.email };
        case 'user.verified':
          return { ...state, isVerified: true };
        default:
          return state;
      }
    },
    read: { from: 0 }
  }
);

Implementation of

EventStoreInterface.aggregateStream

---

append()

append<EventType>(events, condition?): Promise<AppendResult<EventType>>

Append events without a stream using DCB append conditions

Type Parameters

EventType

EventType extends Readonly<{ data: EventData; metadata?: PlatformEventMetadata; tags?: string[]; type: string; }>

Parameters

events

EventType[]

condition?

AppendCondition

Returns

Promise<AppendResult<EventType>>

Implementation of

EventStoreInterface.append

---

appendToStream()

appendToStream<EventType>(streamId, events, options?): Promise<AppendToStreamResult>

Append events to a stream

Type Parameters

EventType

EventType extends Readonly<{ data: EventData; metadata?: PlatformEventMetadata; tags?: string[]; type: string; }>

Parameters

streamId

string

The ID of the stream to append events to

events

EventType[]

Array of events to append to the stream

options?

AppendToStreamOptions

Optional parameters for the append operation

Returns

Promise<AppendToStreamResult>

Promise resolving to the append result with the next expected version

Throws

When expectedStreamVersion doesn’t match current stream version

Throws

When request validation fails

Throws

When the event store doesn’t exist

Throws

When authentication fails

Example

import {
  isVersionConflictError,
  isValidationError
} from '@delta-base/server';

try {
  // Append with optimistic concurrency control
  await eventStore.appendToStream(
    'user-123',
    [{
      type: 'user.updated',
      data: { email: 'updated@example.com' }
    }],
    { expectedStreamVersion: 0n }
  );
} catch (error) {
  if (isVersionConflictError(error)) {
    console.log(`Version conflict: expected ${error.expectedVersion}, got ${error.currentVersion}`);
    // Handle concurrency conflict
  } else if (isValidationError(error)) {
    console.log('Validation errors:', error.validationErrors);
    // Handle validation failures
  } else {
    throw error; // Re-throw unknown errors
  }
}

Implementation of

EventStoreInterface.appendToStream

---

catchUpProjections()

catchUpProjections(): Promise<void>

Explicitly catch up all registered inline projections. Call during Worker startup or in a warm-up request to avoid slower first-request behavior.

Returns

Promise<void>

---

listStreams()

listStreams(options?): Promise<{ streams: string[]; total: number; }>

Get a list of stream IDs in an event store

Parameters

options?

Optional parameters for listing streams

limit?

number

Maximum number of stream IDs to return

offset?

number

Number of stream IDs to skip

pattern?

string

Pattern to match stream IDs (e.g., ‘user-*‘)

Returns

Promise<{ streams: string[]; total: number; }>

Promise resolving to an object containing stream IDs and total count

Example

// List all streams
const { streams, total } = await eventStore.listStreams();

// List streams with pagination
const { streams, total } = await eventStore.listStreams({
  limit: 50,
  offset: 100
});

// List streams matching a pattern
const { streams, total } = await eventStore.listStreams({
  pattern: 'user-*'
});

---

multiStreamAppend()

multiStreamAppend<EventType>(requests): Promise<MultiStreamAppendResult<EventType>>

Append events to multiple streams atomically.

Every request supplies a stream ID, events for that stream, and an optional per-stream expected version. DeltaBase commits the whole operation in one event-store transaction: all version checks pass and all events are written, or no events are written.

Ordering rules:

• global order follows request order, then event order within each request
• streams in the result preserves request order
• events is always returned and sorted by global position

Duplicate stream IDs are rejected. The reserved __dcb__ stream cannot be used here; use append() for DCB append conditions instead.

Inline projections, when configured, run once per touched stream after the event-store transaction commits. Projection writes are not part of the event-store transaction.

Type Parameters

EventType

EventType extends Readonly<{ data: EventData; metadata?: PlatformEventMetadata; tags?: string[]; type: string; }>

Parameters

requests

MultiStreamAppendRequest<EventType>[]

One append request per stream.

Returns

Promise<MultiStreamAppendResult<EventType>>

The global append result, per-stream results, and all written events.

Throws

When any stream’s expected version does not match its current version.

Throws

When the request is invalid, too large, or contains duplicate streams.

Example

const result = await eventStore.multiStreamAppend([
  {
    streamId: 'order-123',
    events: [{ type: 'order.created', data: { orderId: '123' } }],
    expectedStreamVersion: 'no_stream',
  },
  {
    streamId: 'customer-456',
    events: [{ type: 'customer.orderAdded', data: { orderId: '123' } }],
    expectedStreamVersion: 7,
  },
]);

console.log(result.lastPosition);
console.log(result.streams[0]?.nextExpectedStreamVersion);

Implementation of

EventStoreInterface.multiStreamAppend

---

queryEvents()

queryEvents<EventType>(options): Promise<QueryEventsResult<EventType>>

Query events with flexible filtering options

Type Parameters

EventType

EventType extends Readonly<{ data: EventData; metadata?: PlatformEventMetadata; tags?: string[]; type: string; }> = Readonly<{ data: EventData; metadata?: PlatformEventMetadata; tags?: string[]; type: string; }>

Parameters

options

QueryEventsOptions = {}

Query parameters for filtering events

Returns

Promise<QueryEventsResult<EventType>>

Promise resolving to the query result with events and pagination info

Example

// Query all events of a specific type
const result = await eventStore.queryEvents({
  type: 'user.created'
});

// Query events with pagination
const result = await eventStore.queryEvents({
  limit: 20,
  offset: 40,
  includeCount: true
});

// Query events within a time range
const result = await eventStore.queryEvents({
  fromDate: '2023-01-01T00:00:00Z',
  toDate: '2023-01-31T23:59:59Z'
});

Implementation of

EventStoreInterface.queryEvents

---

queryStreamEvents()

queryStreamEvents<EventType>(streamId, options): Promise<QueryEventsResult<EventType>>

Query events for a specific stream with filtering options

Type Parameters

EventType

EventType extends Readonly<{ data: EventData; metadata?: PlatformEventMetadata; tags?: string[]; type: string; }> = Readonly<{ data: EventData; metadata?: PlatformEventMetadata; tags?: string[]; type: string; }>

Parameters

streamId

string

The ID of the stream to query events from

options

Omit<QueryEventsOptions, "streamId"> = {}

Query parameters for filtering events

Returns

Promise<QueryEventsResult<EventType>>

Promise resolving to the query result with events and pagination info

Example

// Query events for a specific stream
const result = await eventStore.queryStreamEvents('user-123', {
  type: 'user.updated'
});

---

queryStreams()

queryStreams(options): Promise<QueryStreamsResult>

Query streams with filtering options

Parameters

options

QueryStreamsOptions = {}

Query parameters for filtering streams

Returns

Promise<QueryStreamsResult>

Promise resolving to the query result with streams and pagination info

Example

// Query all streams
const result = await eventStore.queryStreams();

// Query streams by type
const result = await eventStore.queryStreams({
  streamType: 'user'
});

// Query streams with a pattern match
const result = await eventStore.queryStreams({
  streamIdPattern: 'user-%'
});

---

queryStreamsByType()

queryStreamsByType(streamType, options): Promise<QueryStreamsResult>

Query streams of a specific type with filtering options

Parameters

streamType

string

The stream type to filter by

options

Omit<QueryStreamsOptions, "streamType"> = {}

Query parameters for filtering streams

Returns

Promise<QueryStreamsResult>

Promise resolving to the query result with streams and pagination info

Example

// Query all user streams
const result = await eventStore.queryStreamsByType('user');

// Query user streams with pagination
const result = await eventStore.queryStreamsByType('user', {
  limit: 20,
  offset: 0
});

---

readByQuery()

readByQuery<EventType>(options): Promise<ReadByQueryResult<EventType>>

Read events by DCB query

Type Parameters

EventType

EventType extends Readonly<{ data: EventData; metadata?: PlatformEventMetadata; tags?: string[]; type: string; }>

Parameters

options

ReadByQueryOptions

Returns

Promise<ReadByQueryResult<EventType>>

Implementation of

EventStoreInterface.readByQuery

---

readStream()

readStream<EventType>(streamId, options?): Promise<ReadStreamResult<EventType>>

Read events from a stream

Type Parameters

EventType

EventType extends Readonly<{ data: EventData; metadata?: PlatformEventMetadata; tags?: string[]; type: string; }>

Parameters

streamId

string

The ID of the stream to read events from

options?

ReadStreamOptions

The options for reading events

Returns

Promise<ReadStreamResult<EventType>>

Promise resolving to the read result containing events and stream metadata

Example

// Read all events from a stream
const result = await eventStore.readStream('user-123');

// Read events with a specific starting position
const result = await eventStore.readStream('user-123', { from: 5 });

// Read a specific range of events
const result = await eventStore.readStream('user-123', { from: 5, to: 10 });

// Read a limited number of events
const result = await eventStore.readStream('user-123', { maxCount: 100 });

Implementation of

EventStoreInterface.readStream