DeltaBase Server SDK

A TypeScript SDK for interacting with the DeltaBase event sourcing platform.

Overview

DeltaBase is an event sourcing platform that allows you to store and process event streams. This SDK provides a clean, typed interface for working with DeltaBase from Node.js applications.

Key features:

• Event Storage: Append, read, and query events in streams
• Atomic Multi-Stream Appends: Append to multiple streams in one transaction
• Event Subscriptions: React to events in real-time with webhooks and other subscription types
• Inline Projections: Run projections synchronously for read-after-write consistency
• Advanced Querying: Filter and search events across streams
• Event Aggregation: Build state from event streams with built-in aggregation helpers
• Administration: Manage event stores, subscriptions, and platform resources

Installation

npm install @deltabase/server
# or
yarn add @deltabase/server
# or
pnpm add @deltabase/server

Quick Start

import { DeltaBase } from '@deltabase/server';

// Initialize the DeltaBase client
const client = new DeltaBase({
  apiKey: 'your-api-key',
  // For production use
  // baseUrl: 'https://api.delta-base.com'
});

// Get an event store client
const eventStore = client.getEventStore('my-event-store');

// Append events to a stream
await eventStore.appendToStream('user-123', [
  {
    type: 'user.created',
    data: { name: 'Alice', email: '[email protected]' }
  }
]);

// Read events from a stream
const { events } = await eventStore.readStream('user-123');
console.log('User events:', events);

Key Concepts

Event Sourcing

DeltaBase is built around the event sourcing pattern, where:

• All changes to application state are captured as a sequence of immutable events
• Events are stored in an append-only log
• Current state can be reconstructed by replaying events
• Event history provides a complete audit trail of all changes

Streams

A stream is a sequence of related events, typically representing the history of a single entity. Streams are identified by a unique ID, such as user-123 or order-456.

Events

An event represents something that happened in your domain. Events are immutable and include:

• type: A descriptive name for the event (e.g., user.created, order.placed)
• data: The payload of the event, containing relevant information
• Metadata: Additional information like timestamps, correlation IDs, etc.

Subscriptions

Subscriptions allow you to react to events in real-time. You can subscribe to specific event types or patterns and receive notifications via webhooks or other mechanisms.

Usage Examples

Working with Event Stores

import { DeltaBase } from '@deltabase/server';

const client = new DeltaBase({ apiKey: 'your-api-key', baseUrl: 'https://api.delta-base.com' });

// Create a new event store
const management = client.getManagement();
await management.createEventStore({
  name: 'orders',
  description: 'Event store for order events',
});

// Get an event store client
const orderStore = client.getEventStore('orders');

// Append events
await orderStore.appendToStream('order-123', [
  {
    type: 'order.created',
    data: {
      customerId: 'cust-456',
      items: [{ productId: 'prod-789', quantity: 2, price: 25.99 }],
      total: 51.98
    }
  }
]);

// Read events
const { events } = await orderStore.readStream('order-123');

// Query events
const { events: userOrders } = await orderStore.queryEvents({
  type: 'order.created',
  fromDate: '2023-01-01T00:00:00Z',
  limit: 10
});

// List streams
const { streams } = await orderStore.listStreams({ pattern: 'order-*' });

Atomic Multi-Stream Appends

Use multiStreamAppend when one command must write to several streams and partial success would be a bug. DeltaBase checks every stream's expected version first, then commits all events in one transaction. If any check fails, nothing is written.

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

console.log(result.lastPosition);
console.log(result.streams[0]?.nextExpectedStreamVersion);
console.log(result.events); // Flat list, ordered by globalPosition

Rules:

• All-or-nothing across all streams
• One request per stream; duplicate stream IDs are rejected
• Each stream has its own expectedStreamVersion
• Global event order follows request order, then event order
• Inline projections run per stream after the event transaction commits

Use DCB (readByQuery + append) for query-based constraints like uniqueness. Use multiStreamAppend when you already know the streams that need to change together.

Aggregating State from Events

// Define your state type
type OrderState = {
  id: string;
  customerId: string;
  items: Array<{ productId: string; quantity: number; price: number }>;
  total: number;
  status: 'pending' | 'shipped' | 'delivered' | 'canceled';
};

// Define your event types
type OrderEvent =
  | { type: 'order.created'; data: { customerId: string; items: Array<{ productId: string; quantity: number; price: number }>; total: number } }
  | { type: 'order.shipped'; data: { trackingNumber: string; shippedAt: string } }
  | { type: 'order.delivered'; data: { deliveredAt: string } }
  | { type: 'order.canceled'; data: { reason: string; canceledAt: string } };

// Aggregate events into state
const { state, currentStreamVersion } = await orderStore.aggregateStream<OrderState, OrderEvent>(
  'order-123',
  {
    initialState: () => ({
      id: '',
      customerId: '',
      items: [],
      total: 0,
      status: 'pending'
    }),
    evolve: (state, event) => {
      switch (event.type) {
        case 'order.created':
          return {
            ...state,
            id: event.streamId,
            customerId: event.data.customerId,
            items: event.data.items,
            total: event.data.total
          };
        case 'order.shipped':
          return { ...state, status: 'shipped' };
        case 'order.delivered':
          return { ...state, status: 'delivered' };
        case 'order.canceled':
          return { ...state, status: 'canceled' };
        default:
          return state;
      }
    }
  }
);

console.log('Current order state:', state);

Setting Up Event Subscriptions

// Get the event bus for a store
const eventBus = client.getEventBus('orders');

// Create a webhook subscription for all order events
const subscription = await eventBus.subscribeWebhook(
  'order.*',
  'https://example.com/webhooks/orders',
  {
    headers: { 'X-API-Key': 'webhook-secret' },
    retryPolicy: {
      maxAttempts: 3,
      backoffMinutes: 5
    }
  }
);

// Check if this is an existing subscription (duplicate detection)
if (subscription.isExistingSubscription) {
  console.log('Reusing existing subscription:', subscription.message);
} else {
  console.log('Created new subscription:', subscription.message);
}

// List subscriptions
const { subscriptions } = await eventBus.listSubscriptions();

// Unsubscribe
await eventBus.unsubscribe(subscription.subscriberId);

Duplicate Detection

For webhook subscriptions, the system automatically detects if a subscription with the same configuration already exists (same URL, headers, timeout, and event filter). If found, the existing subscription will be returned instead of creating a duplicate. This helps prevent accidentally creating multiple subscriptions for the same webhook endpoint.

You can check the isExistingSubscription field in the response to determine if a new subscription was created or an existing one was returned. The message field provides a human-readable explanation of what happened.

Administrative Operations

const management = client.getManagement();

// List all event stores
const { eventStores } = await management.listEventStores();

// Get details about a specific event store
const storeDetails = await management.getEventStore('orders');
console.log('Storage used:', storeDetails.statistics?.databaseSizeBytes);

// Update event store settings
await management.updateEventStore('orders', {
  description: 'Updated description',
  retentionPeriodDays: 90
});

// Delete an event store (use with caution!)
await management.deleteEventStore('unused-store');

API Reference

The SDK is organized into several main classes:

DeltaBase

The main client class that provides access to all functionality.

const client = new DeltaBase({
  apiKey: 'your-api-key',
  baseUrl: 'https://api.delta-base.com' // Optional
});

Methods:

• getManagement(): Returns a ManagementClient for administrative operations
• getEventStore(eventStoreId, options?): Returns an EventStore client for the specified store. Pass { projections } to enable inline projections.
• getEventBus(eventStoreId): Returns an EventBus client for the specified store

EventStore

For working with event streams within a specific event store.

Methods:

• appendToStream(streamId, events, options?): Append events to a stream. Runs inline projections if configured.
• multiStreamAppend(requests): Append events to multiple streams atomically. Runs inline projections per stream if configured.
• append(events, condition?): Append events with DCB (Dynamic Consistency Boundary). Runs inline projections if configured.
• readStream(streamId, options?): Read events from a stream
• aggregateStream(streamId, options): Build state from events using an aggregation function
• queryEvents(options?): Query events across streams with filtering
• queryStreams(options?): Query streams with filtering
• listStreams(options?): Get a list of stream IDs
• catchUpProjections(): Explicitly catch up all registered inline projections

EventBus

For managing event subscriptions.

Methods:

• subscribe(options): Create a subscription to events
• subscribeWebhook(eventFilter, url, options?): Convenient method to create a webhook subscription
• getSubscription(subscriberId): Get details about a subscription
• listSubscriptions(options?): List all subscriptions
• unsubscribe(subscriberId): Delete a subscription

ManagementClient

For administrative operations on event stores.

Methods:

• createEventStore(options): Create a new event store
• listEventStores(): List all event stores
• getEventStore(eventStoreId): Get details about an event store
• updateEventStore(eventStoreId, settings): Update event store settings
• deleteEventStore(eventStoreId): Delete an event store

Inline Projections

Inline projections run synchronously within the same request as the event append, giving you read-after-write consistency. After appendToStream or append (DCB) completes, your read models are already up-to-date.

import { DeltaBase } from '@deltabase/server';
import {
  singleStreamProjection,
  projections,
} from '@delta-base/toolkit';

// Define a projection
const orderSummary = singleStreamProjection({
  projectionName: 'order-summary',
  store: readModelStore, // KVReadModelStore, DOReadModelStore, etc.
  canHandle: ['order.created', 'order.shipped', 'order.delivered'],
  getDocumentId: (streamId) => `order:${streamId}`,
  initialState: () => ({ items: [], total: 0, status: 'draft' }),
  evolve: (doc, event) => {
    switch (event.type) {
      case 'order.created':
        return { ...doc, status: 'created' };
      case 'order.shipped':
        return { ...doc, status: 'shipped' };
      case 'order.delivered':
        return { ...doc, status: 'delivered' };
      default:
        return doc;
    }
  },
});

// Register on the event store
const eventStore = client.getEventStore('orders', {
  projections: projections.inline([orderSummary]),
});

// Every appendToStream call now runs projections automatically
await eventStore.appendToStream('order-123', [
  { type: 'order.created', data: { customerId: 'cust-1' } },
]);

// Read model is already consistent
const order = await readModelStore.get('order:order-123');

Catch-Up

When a new projection is registered (or events were appended without projections active), the projection automatically catches up before live handling on its first append. If a projection later fails, only that projection is marked stale and catches up again before it handles newer appended events. You can also trigger catch-up explicitly:

// Trigger during Worker startup for predictable latency
await eventStore.catchUpProjections();

Error Handling

Projection errors never roll back events (events are facts). Configure error behavior:

const eventStore = client.getEventStore('orders', {
  projections: projections.inline([orderSummary], {
    onError: 'throw',  // default: fail the request
    // onError: 'log', // log and continue
    // onError: (err, projection, events) => { ... } // custom
  }),
});

With onError: 'throw', the append rejects after the events are persisted. With onError: 'log' or a custom non-throwing handler, the append can still resolve. If catch-up fails, DeltaBase skips handleAppend for that stale projection during the same request so it cannot advance past missed historical events.

See the Inline Projections Guide for complete examples including DCB usage and idempotency recommendations.

Advanced Topics

Optimistic Concurrency Control

You can use the expectedStreamVersion option to implement optimistic concurrency control:

try {
  await eventStore.appendToStream(
    'user-123',
    [{ type: 'user.updated', data: { email: '[email protected]' } }],
    { expectedStreamVersion: 5n } // Only succeed if the current version is 5
  );
} catch (error) {
  console.error('Concurrency conflict!', error);
}

Transaction IDs

Every event can include a transaction ID to group related events:

await eventStore.appendToStream('order-123', [
  {
    type: 'order.created',
    data: { /* ... */ },
    transactionId: 'tx-abc-123'
  },
  {
    type: 'inventory.reserved',
    data: { /* ... */ },
    transactionId: 'tx-abc-123' // Same transaction ID
  }
]);

License

(c) Copyright 2025 nibbio LLC, all rights reserved.