createEventStore

createEventStore is the central orchestrator of the library. It connects your aggregate roots to the underlying storage, manages event distribution to projections, process managers, and event listeners, and handles initialization and replay.

Import

import { createEventStore } from '@requence/event-sourcing'

Signature

function createEventStore<Root extends AnyAggregateRoot>(
  params: EventStoreParams<Root>
): EventStore<Event, Root>

Parameters

The function accepts a single configuration object:

Parameter	Type	Required	Description
loadEvents	LoadEvents	✓	Async generator that loads events from the underlying storage.
appendEvents	AppendEvents	✓	Function that persists new events and returns them with assigned positions.
aggregateRoots	Root[]	✓	Array of aggregate root instances to register.
checkpoint	CheckpointMethods	✓	Methods for persisting projection/process manager checkpoints.
lock	LockCreator		Custom locking strategy for aggregate root streams. Defaults to an in-memory lock with a 5-second TTL. Use redisLock from @requence/event-sourcing/redis for distributed deployments.
aggregateRootSnapshot	AggregateRootSnapshotMethods		Enables snapshot support for aggregate roots.
projectionSnapshot	ProjectionSnapshotMethods		Enables snapshot support for projections.
postProcessEvent	(event, aggregateRoot) => WrappedEvent		Hook to mutate or decorate events before they are appended.
onProjectionReplay	OnProgress		Callbacks for tracking projection replay progress.
onProcessManagerRefresh	OnProgress		Callbacks for tracking process manager refresh progress.
autoInit	boolean		When set to false, the event store will not initialize automatically. You must call .init() manually. Defaults to true.

Concurrency & Locking

Aggregate root streams are automatically locked before any command is executed. This prevents concurrent modifications to the same stream and eliminates event stream version collisions.

By default, the library uses an in-memory mutex (via async-mutex) with a 5-second TTL. This is suitable for single-process deployments.

Distributed Locking with redisLock

For multi-instance deployments, pass redisLock as the lock parameter to coordinate across all instances via Redis:

import { createEventStore } from '@requence/event-sourcing/drizzle'
import { redisLock } from '@requence/event-sourcing/redis'
import Redis from 'ioredis'

const redis = new Redis(process.env.REDIS_URL)

const eventStore = createEventStore({
  database: db,
  aggregateRoots: [user, project],
  lock: redisLock(redis),       // distributed lock via Redis
})

redisLock(client, defaultTtl?)

Parameter	Type	Default	Description
client	Redis (ioredis)	—	An ioredis client instance.
defaultTtl	number	5000	Lock time-to-live in milliseconds.

The implementation uses atomic SET NX PX for acquisition and Lua scripts for safe extend/release, preventing split-brain issues.

Tip

The default in-memory lock is ideal for testing and single-process apps. Switch to redisLock when running multiple server instances.

Return Value

Returns an EventStore object with the following methods:

createProjection(name)

Creates and registers a new Projection with the given name. Each name must be unique within the event store.

const userList = eventStore.createProjection('user-list')
  .withEventHandlers({
    async onUserCreated({ streamId, payload }) { /* ... */ }
  })

createProcessManager(name)

Creates and registers a new Process Manager with the given name.

const cascadeDelete = eventStore.createProcessManager('cascade-delete')
  .withEventHandlers({
    async onOrganizationDeleted({ payload }) { /* ... */ }
  })

createEventListener(name)

Creates and registers a new Event Listener with the given name.

const auditLog = eventStore.createEventListener('audit-log')
  .withEventHandlers({
    async onUserCreated({ payload }) { /* ... */ }
  })

getAggregateRoot(type)

Returns a registered aggregate root by its type name. Fully type-safe — the return type narrows based on the provided type string.

const userRoot = eventStore.getAggregateRoot('user')

transaction(handler)

Executes the given async handler inside a transaction. Events from all aggregate roots within the handler are batched and committed atomically.

await eventStore.transaction(async () => {
  const org = orgRoot.newStream()
  org.create({ name: 'Acme' })

  const user = userRoot.newStream()
  user.create({ name: 'Alice', orgId: org.streamId })
})

init()

Manually initializes the event store. Required when autoInit is set to false. Returns the event store instance for chaining.

const store = createEventStore({ autoInit: false, /* ... */ })
await store.init()

isReady()

Returns a promise that resolves when all projections and process managers have finished their initial catch-up hydration.

await eventStore.isReady()
// All projections and process managers are now up to date

rebuild()

Performs a full rebuild of all replayable projections and stateful process managers. This re-reads every event from the store and re-applies them from scratch.

await eventStore.rebuild()

Caution

rebuild() will call deleteAll on every projection that supports replay and reset the state of every stateful process manager. This can be a destructive and long-running operation.

Usage Example

import { createEventStore } from '@requence/event-sourcing'
import { createDrizzleAdapter } from '@requence/event-sourcing/drizzle'
import userRoot from './aggregates/user.ts'
import orgRoot from './aggregates/organization.ts'

const adapter = createDrizzleAdapter(db)

const eventStore = createEventStore({
  loadEvents: adapter.loadEvents,
  appendEvents: adapter.appendEvents,
  checkpoint: adapter.checkpoint,
  aggregateRoots: [userRoot, orgRoot],
})

await eventStore.isReady()

OnProgress Callback

The onProjectionReplay and onProcessManagerRefresh parameters accept an object with the following optional callbacks:

type OnProgress = {
  begin?: (name: string) => void
  progress?: (name: string, index: number, event: BaseOutputEvent) => void
  end?: (name: string) => void
}