Custom Event and Snapshot Stores

How to create your own Event Store

An event store has two halves. The EventStore base class implements the contract that every store shares: it validates appends, serializes the events, checks the expected version, retries ExpectedVersion.Any, stamps the global positions on the envelopes and publishes them. Your store implements the driver methods, which read and write the database and never touch the event map. The conformance suite checks that the two halves fit.

To register your store in the module, you need an EventStore subclass and a configuration interface.

Create a configuration

The configuration extends EventStoreConfig, names your store as the driver, and adds the options your store needs.

import type { Type } from '@nestjs/common';
import type { EventStoreConfig } from '@ocoda/event-sourcing';
import type { FooEventStore } from './foo.event-store';

export interface FooEventStoreConfig extends EventStoreConfig {
  driver: Type<FooEventStore>;
  // Add any additional parameters you need to initialize your Event Store
  // e.g. a connection string
}

Create an Event Store

The module creates your store with new driver(context, options):

• context holds the eventMap and the publisher (the EventBus). The base class uses both; your store needs neither.
• options is the configuration you registered, without driver and useDefaultPool (the module handles those). It is available as this.options.

The base class stores both, so you don’t need a constructor of your own. If you declare one, pass both arguments on to super. EventStoreDriver, the type of driver, is this constructor: new (context, options) => EventStore.

The snippet below is a skeleton, fill in the methods with the logic for your database.

import {
  type EventEnvelope,
  EventStore,
  type EventStoreCapabilities,
  type EventStream,
  type IEventCollection,
  type IEventCollectionFilter,
  type IEventFilter,
  type IEventPool,
  type IReadAllFilter,
  type PersistOutcome,
  type PersistTarget,
} from '@ocoda/event-sourcing';
import type { FooEventStoreConfig } from './foo-event-store-config';

export class FooEventStore extends EventStore<FooEventStoreConfig> {
  // The guarantees your store gives; the conformance suite tests each one you claim
  override readonly capabilities: EventStoreCapabilities = {
    atomicAppend: true,
    headers: true,
    globalOrder: 'gap-safe',
  };

  async connect(): Promise<void> {
    // Connect to your database, the configuration is available as this.options
  }

  async disconnect(): Promise<void> {
    // Disconnect from your database
  }

  async ensureCollection(pool?: IEventPool): Promise<IEventCollection> {
    // Create the collection of the pool unless it exists, and return its name.
    // EventCollection.get(pool) is the name: 'events' or e.g. 'tenant-1-events'
  }

  async *listCollections(filter?: IEventCollectionFilter): AsyncGenerator<IEventCollection[]> {
    // Yield the collections in batches (arrays) of filter?.batch collections
  }

  async getStreamVersion(stream: EventStream, pool?: IEventPool): Promise<number> {
    // The version of the last event of the stream, 0 when it has none
  }

  async getEnvelope(stream: EventStream, version: number, pool?: IEventPool): Promise<EventEnvelope> {
    // The envelope of one event, with its globalPosition; an EventNotFoundException when there is none
  }

  async *getEnvelopes(stream: EventStream, filter?: IEventFilter): AsyncGenerator<EventEnvelope[]> {
    // The envelopes of a stream from filter.fromVersion on, in the requested direction,
    // in batches (arrays) of filter?.batch envelopes
  }

  async *readAll(filter?: IReadAllFilter): AsyncGenerator<EventEnvelope[]> {
    // The envelopes of a pool across streams, by global position, from filter.fromPosition on (inclusive),
    // in batches (arrays) of filter?.batch envelopes
  }

  protected async persistEvents(envelopes: readonly EventEnvelope[], target: PersistTarget): Promise<PersistOutcome> {
    // Store the envelopes all or nothing, assign their global positions, and return them:
    // return { status: 'committed', positions };
    // On a duplicate (stream, version) key: return { status: 'conflict', cause: error };
  }
}

The driver methods:

Method	What it does
connect() and disconnect()	Open and close the connection to the database. The capabilities are final once connect() resolved.
ensureCollection(pool?)	Creates the collection (table) of a pool unless it exists, and returns its name. Never touches existing events.
listCollections(filter?)	Yields the existing collections as arrays.
getStreamVersion(stream, pool?)	Returns the version of the stream: the version of its last event, 0 when it has none.
getEnvelope(stream, version, pool?)	Returns the envelope of an event of a stream.
getEnvelopes(stream, filter?)	Yields the envelopes of a stream as arrays, from filter.fromVersion on and in the requested direction.
readAll(filter?)	Yields the envelopes of a pool as arrays, in the order of their global position, from filter.fromPosition on. See Reading all events.
persistEvents(envelopes, target)	Stores the envelopes of an append, all or nothing, and returns their global positions.

Every method returns a promise (or an async generator), and every read throws an EventCollectionNotFoundException when the collection of the pool doesn’t exist. getEnvelope throws an EventNotFoundException for an event that doesn’t exist. Envelopes read from the database are built with EventEnvelope.from(event, payload, metadata).

The driver methods are abstract, so TypeScript reports the ones your store lacks.

Global positions are part of the contract

Every event store numbers the events of each pool and reads them back with readAll, also with globalOrder: 'best-effort': positions start at 1n, follow each other within an append and only increase, which read-all-order checks whatever the capabilities. 'best-effort' only relaxes when a position becomes visible to readers. A database that can only read one stream at a time, or can’t number a pool’s events in the write that stores them, doesn’t meet this contract: it can’t serve readAll, and with it the consumers and read models of your application that catch up from a checkpoint (Pull instead of push). Such a database can still hold snapshots: a snapshot store needs no global positions.

Don’t override appendEvents, getEvent or getEvents. The base class implements them for every store, and the module refuses a store that overrides one, also as a class field, with an InvalidEventStoreImplementationException when it creates the store, so the application fails to bootstrap. assertEventStoreImplementation(store) runs the same check. To decorate the appends of an existing store, for tracing, extend it (PostgresEventStore, for instance), override persistEvents and call super.persistEvents(envelopes, target).

Implement persistEvents

The base class calls persistEvents once it has validated the append, serialized the events, numbered them and checked that the stream is at target.expectedVersion. The first envelope has the version target.expectedVersion + 1. persistEvents checks no versions and serializes nothing:

• Store the envelopes all or nothing, in one transaction, and never overwrite an event. Store every field of the metadata: eventId, aggregateId, version, occurredOn, correlationId, causationId, headers and eventVersion.
• Assign the global positions in the same write: one per envelope, consecutive, higher than every position the pool handed out before, and never reused. Return them as bigints: { status: 'committed', positions }. toPosition() converts what a database driver returns (a decimal string, a number or an Int64) to a bigint.
• Map a duplicate (stream, version) key to a conflict: return { status: 'conflict', cause: error }, with the duplicate-key error of your database as the cause (PostgreSQL 23505, MariaDB 1062, MongoDB 11000), and the version of the stream as actualVersion if you know it. That is how an append that lost a race is detected, since the version check of the base class runs before the write. Only this key is a conflict.
• Throw everything else as an EventStorePersistenceException with the outcome you know: 'not-persisted' when nothing was written (the connection failed, the transaction was rolled back, the collection doesn’t exist), and 'unknown' when the commit may have happened (the connection dropped after COMMIT was sent). The base class passes your exception through unchanged, and treats any other error as 'unknown'.

For a pool whose collection doesn’t exist, throw the persistence exception with an EventCollectionNotFoundException as its cause:

throw new EventStorePersistenceException(
  { collection: target.collection, outcome: 'not-persisted' },
  { cause: new EventCollectionNotFoundException({ collection: target.collection, pool: target.pool }) },
);

To tell errors apart, use isEventSourcingError(error, EventSourcingErrorCode.EventStoreVersionConflict) rather than comparing error.constructor.

Declare the capabilities

capabilities lists the optional guarantees of your store. The flags you leave out get their default:

Capability	Default	What it means
atomicAppend	true	An append stores all of its events or none of them.
headers	false	The store persists the headers of the events. Without it, an append with headers is rejected with an UnsupportedOperationException before anything is written.
globalOrder	'best-effort'	'gap-safe': once readAll yielded a position, no event at or below it commits later, so a reader that resumes after its last position never misses an event. 'best-effort': positions strictly increase, but an event may still commit below a position that was read.

Claim a capability only if its conformance cases pass, and leave globalOrder at 'best-effort' unless read-all-gap-safe passes. A passing case only shows something if it can fail for your store: keep a negative control next to it, a deliberately broken variant of your store that the case must catch. For read-all-gap-safe, that is a variant that takes its positions in a statement of its own and inserts the events a little later, in another transaction; for append-atomic-partial-failure, one that stores the events of an append one by one. Run it with only and expectFailure (see Test your Event Store). When the capabilities depend on the database (a MongoDB server without a replica set has no transactions, for instance), set them in connect(), with Object.assign(this.capabilities, { ... }).

Register your Event Store

Register your store in the module, with its configuration:

import { Module } from '@nestjs/common';
import { EventSourcingModule } from '@ocoda/event-sourcing';
import { FooEventStore } from './foo.event-store';
import type { FooEventStoreConfig } from './foo-event-store-config';

@Module({
  imports: [
    EventSourcingModule.forRootAsync<FooEventStoreConfig>({
      useFactory: () => ({
        events: [...Events],
        eventStore: {
          driver: FooEventStore,
          ... // the additional parameters from your configuration
        }
      }),
    }),
  ],
  providers: [...],
  controllers: [...],
})
export class AppModule {}

Test your Event Store

The conformance suite that every built-in store runs encodes the contract, and @ocoda/event-sourcing/testing publishes it. It registers Vitest tests, so add Vitest 4 or 5 to your dev dependencies: vitest is an optional peer dependency of @ocoda/event-sourcing, which only this subpath loads. The suite imports Vitest’s API itself, so it needs no globals: true.

npm checks an optional peer dependency whenever the project has that package, so with another major of Vitest, npm install fails with ERESOLVE until you move to Vitest 4 or 5, or install with --legacy-peer-deps. pnpm and Yarn only warn.

In a spec file, the suite hands your factory the store context (the suite’s event map and a RecordingPublisher), and your factory returns the connected store:

foo.event-store.conformance.spec.ts

import { describeEventStoreConformance } from '@ocoda/event-sourcing/testing';
import { FooEventStore } from './foo.event-store';

describeEventStoreConformance('FooEventStore', async (context) => {
  const store = new FooEventStore(context, { ... });
  await store.connect();
  return {
    store,
    cleanup: async (collections) => {
      // Drop the collections, then disconnect
    },
    faults: {
      failInsertOf: async (collection, eventName) => {
        // Make inserts of eventName fail inside the store's write, with a trigger for instance
        return async () => {
          // Remove the fault
        };
      },
    },
  };
});

Cases that need a capability your store doesn’t claim are skipped with capability: <flag>. A store that claims atomicAppend (the default) provides faults.failInsertOf, which append-atomic-partial-failure uses to make an append fail halfway, or skips that case with a reason. A case your store can’t pass yet is skipped with a reason, in the skip option; EVENT_STORE_CONFORMANCE_CASES lists the case ids.

• The suite creates its own pools, named after the pool option, which defaults to a name unique to the run, and hands their collections to cleanup at the end. The list also names collections that were never created: ignore those.
• CONFORMANCE_RUN_SKIPPED=true runs the cases you skipped, to check whether a skip is still needed. It doesn’t lift a capability gate.
• CONFORMANCE_REPEAT=n runs every case n times, to soak out flaky concurrency.
• expectFailure turns the cases into negative controls: each one passes only when the case fails. Pass a pattern per case, so that only the failure that detects the defect counts, and only to register just those cases. A case that the store’s capabilities gate off fails the control, since it proves nothing.

foo.event-store.negative-control.spec.ts

import { describeEventStoreConformance } from '@ocoda/event-sourcing/testing';
import { BrokenFooEventStore } from './broken-foo.event-store';

// BrokenFooEventStore extends FooEventStore and still claims globalOrder: 'gap-safe', but its persistEvents takes
// the positions in a statement of its own and inserts the events a few milliseconds later, in a second transaction
describeEventStoreConformance(
  'BrokenFooEventStore (negative control)',
  async (context) => {
    const store = new BrokenFooEventStore(context, { ... });
    await store.connect();
    return {
      store,
      cleanup: async (collections) => {
        // Drop the collections, then disconnect
      },
    };
  },
  {
    only: ['read-all-gap-safe'],
    expectFailure: { 'read-all-gap-safe': /events the tailing reader never read/ },
  },
);

To construct your store outside the module, in a test of your own, createTestEventStoreContext({ events }) builds a context with a RecordingPublisher, which records what the store publishes. createInMemoryEventStore({ events }) and createInMemorySnapshotStore() return connected in-memory stores, for tests that don’t boot the module.

Migrating a 3.x store

1. The constructor becomes (context, options). Remove yours, or pass both arguments on.
2. Delete appendEvents, getEvents, getEvent and every use of this.eventMap, which is gone: the base class serializes and deserializes the events. A store that still overrides one of those methods fails the bootstrap.
3. Implement getStreamVersion, getEnvelope, getEnvelopes, readAll and persistEvents. readAll replaces getAllEnvelopes, and with it the year-month column, getYearMonthRange, IAllEventsFilter and ULID.yearMonth, which are removed.
4. Make every method async.
5. Store the correlation id, causation id, headers and event version, and declare your capabilities.
6. Construct exceptions with a single object argument, plus the error options for the cause.
7. Run the conformance suites from @ocoda/event-sourcing/testing against your event store and your snapshot store. Declare capabilities instead of skipping cases, and give every skip a reason.

---

How to create your own Snapshot Store

Creating a custom snapshot store follows the same pattern as described above.

Create a configuration

import type { Type } from '@nestjs/common';
import type { SnapshotStoreConfig } from '@ocoda/event-sourcing';
import type { FooSnapshotStore } from './foo.snapshot-store';

export interface FooSnapshotStoreConfig extends SnapshotStoreConfig {
  driver: Type<FooSnapshotStore>;
  // Add any additional parameters you need to initialize your Snapshot Store
  // e.g. a connection string
}

Create a Snapshot Store

The module creates your snapshot store with new driver(config), where config is the configuration you registered without driver and useDefaultPool (the module handles those). It’s available in your class as this.options.

import {
  type AggregateRoot,
  type ISnapshot,
  type ISnapshotCollection,
  type ISnapshotCollectionFilter,
  type ISnapshotFilter,
  type ISnapshotPool,
  type SnapshotEnvelope,
  SnapshotStore,
  type SnapshotStream,
} from '@ocoda/event-sourcing';
import type { FooSnapshotStoreConfig } from './foo-snapshot-store-config';

export class FooSnapshotStore extends SnapshotStore<FooSnapshotStoreConfig> {
  public async connect(): Promise<void> {
    // Connect to your Snapshot Store, the configuration is available as this.options
  }

  public async disconnect(): Promise<void> {
    // Disconnect from your Snapshot Store
  }

  public async ensureCollection(pool?: ISnapshotPool): Promise<ISnapshotCollection> {
    // Ensure the collection for the pool exists in your Snapshot Store and return its name.
    // Use SnapshotCollection.get(pool) to get the name: 'snapshots' or e.g. 'tenant-1-snapshots'
  }

  public async *listCollections(filter?: ISnapshotCollectionFilter): AsyncGenerator<ISnapshotCollection[]> {
    // List all collections in your Snapshot Store
    // yield them in batches (arrays) of filter?.batch collections
  }

  async getSnapshot<A extends AggregateRoot>(
    snapshotStream: SnapshotStream,
    version: number,
    pool?: ISnapshotPool,
  ): Promise<ISnapshot<A>> {
    // Get the snapshot with the given version from your Snapshot Store
  }

  async *getSnapshots<A extends AggregateRoot>(
    snapshotStream: SnapshotStream,
    filter?: ISnapshotFilter,
  ): AsyncGenerator<ISnapshot<A>[]> {
    // Get the snapshots of a stream from your Snapshot Store
  }

  async getLastSnapshot<A extends AggregateRoot>(
    snapshotStream: SnapshotStream,
    pool?: ISnapshotPool,
  ): Promise<ISnapshot<A> | void> {
    // Get the last snapshot of a stream from your Snapshot Store
  }

  async getLastSnapshots<A extends AggregateRoot>(
    snapshotStreams: SnapshotStream[],
    pool?: ISnapshotPool,
  ): Promise<Map<SnapshotStream, ISnapshot<A>>> {
    // Get the last snapshot of multiple streams from your Snapshot Store
  }

  async appendSnapshot<A extends AggregateRoot>(
    snapshotStream: SnapshotStream,
    version: number,
    snapshot: ISnapshot<A>,
    pool?: ISnapshotPool,
  ): Promise<SnapshotEnvelope<A>> {
    // Store a snapshot in your Snapshot Store and return its envelope. Reject with a
    // SnapshotStoreVersionConflictException unless version is higher than the stream's last one, so that racing
    // appends leave a single latest snapshot
  }

  async getLastEnvelope<A extends AggregateRoot>(
    snapshotStream: SnapshotStream,
    pool?: ISnapshotPool,
  ): Promise<SnapshotEnvelope<A> | void> {
    // Get the envelope of the last snapshot of a stream (the highest version) from your Snapshot Store
  }

  async getEnvelope<A extends AggregateRoot>(
    snapshotStream: SnapshotStream,
    version: number,
    pool?: ISnapshotPool,
  ): Promise<SnapshotEnvelope<A>> {
    // Get the envelope of the snapshot with the given version from your Snapshot Store
  }

  async *getEnvelopes<A extends AggregateRoot>(
    snapshotStream: SnapshotStream,
    filter?: ISnapshotFilter,
  ): AsyncGenerator<SnapshotEnvelope<A>[]> {
    // Get the envelopes of the snapshots of a stream from your Snapshot Store
  }
}

The methods above are required, and every one of them returns a promise (or an async generator). A SnapshotRepository uses appendSnapshot to save snapshots and getLastEnvelope to load the last one: the snapshot with the highest version of the stream.

The SnapshotStore base class has a default for two more methods. Override them when your database can do better:

Method	Default	Used by
getManyLastSnapshotEnvelopes(snapshotStreams, pool?)	Reads the streams one after the other with getLastEnvelope.	SnapshotRepository.loadMany()
getLastEnvelopesForAggregate(aggregate, filter?)	Rejects with an UnsupportedOperationException when it’s read (before 4.0, a missing method made loadAll() throw the NotImplementedException of @nestjs/common).	SnapshotRepository.loadAll()

An implementation of getLastEnvelopesForAggregate yields the latest snapshot of every stream of the aggregate, in descending binary order of the aggregate ids, and treats filter.aggregateId as an exclusive cursor. See Snapshot store.

Test your Snapshot Store

describeSnapshotStoreConformance from @ocoda/event-sourcing/testing checks a snapshot store against this contract, the same way as the event store suite. Its factory takes no context:

foo.snapshot-store.conformance.spec.ts

import { describeSnapshotStoreConformance } from '@ocoda/event-sourcing/testing';
import { FooSnapshotStore } from './foo.snapshot-store';

describeSnapshotStoreConformance('FooSnapshotStore', async () => {
  const store = new FooSnapshotStore({ ... });
  await store.connect();
  return {
    store,
    cleanup: async (collections) => {
      // Drop the collections, then disconnect
    },
  };
});

If your store keeps the base class’s getLastEnvelopesForAggregate, the cases of that read are skipped with capability: getLastEnvelopesForAggregate, and the other cases leave it out. SNAPSHOT_STORE_CONFORMANCE_CASES lists the case ids.

Register your Snapshot Store

import { Module } from '@nestjs/common';
import { EventSourcingModule } from '@ocoda/event-sourcing';
import { FooSnapshotStore } from './foo.snapshot-store';
import type { FooEventStoreConfig } from './foo-event-store-config';
import type { FooSnapshotStoreConfig } from './foo-snapshot-store-config';

@Module({
  imports: [
    EventSourcingModule.forRootAsync<FooEventStoreConfig, FooSnapshotStoreConfig>({
      useFactory: () => ({
        events: [...Events],
        eventStore: { ... },
        snapshotStore: {
          driver: FooSnapshotStore,
          ... // the additional parameters from your configuration
        }
      }),
    }),
  ],
  providers: [...],
  controllers: [...],
})
export class AppModule {}