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.

Optimistic-concurrency retry

The lock and the storage adapter’s version check are two layers of the same story: the lock avoids the race by serializing writers, and the version check catches it if a race slips through (raising a ConcurrencyError). A race can still slip through when a lock’s TTL lapses mid-command, or in single-instance deployments where no distributed lock is configured.

For those residual cases, opt in to recovery at the call site with settled({ maxRetries }): the stream is reloaded and the commands re-applied against the current version instead of the error surfacing.

await settingsRoot.loadStream('global').update({ theme: 'dark' }).settled({
  maxRetries: 5,
})

Retry is opt-in per call (default 0) rather than a global setting, because it is only safe for pure command handlers and some flows — such as a fixed-id newStream(...) used for idempotent create — deliberately rely on the ConcurrencyError surfacing.

Inside a process manager or a transaction, settled() cannot be called — the surrounding scope settles dispatched streams automatically when it completes. Declare the retry defaults on that scope instead: createProcessManager(name, { settled }) or transaction(handler, { settled }).

Running Multiple Instances

redisLock makes the write path safe to scale: any number of instances can accept commands, because stream appends are guarded by both the distributed lock and the storage adapter’s optimistic-concurrency check (a ConcurrencyError is raised if two instances race on the same stream version).

The read/reaction path works differently, and it is important to understand how events reach your handlers when scaling:

• Events are delivered to projections, process managers, and event listeners only on the instance that appended them. There is no built-in cross-instance event bus. On startup, each instance also catches up from its checkpoint, but after that it only sees events it writes itself.
• Because each event is therefore processed by exactly one instance, projections and listeners that write to a shared store (a SQL read model, Redis, a search index) or perform idempotent side effects stay consistent — every event is applied exactly once, with no duplication.

Given that model, the recommended topology is:

Component	Safe to run on N instances?	Notes
Command / write handlers	✅ Yes	Use redisLock. Concurrency is enforced by the lock + storage adapter.
Projections → shared read model	✅ Yes	Make writes idempotent (upsert by id). The checkpoint cursor is shared and advances monotonically.
Event listeners (side effects)	✅ Yes	Each event fires its side effect once, on the writing instance.
Projections holding in-memory state	⚠️ No	Only the writing instance’s copy updates. Back the projection with a shared store instead.
Stateful process managers (.withState())	⚠️ With care	State is folded into a shared checkpoint under optimistic concurrency (see below).
Push delivery (WebSocket/SSE) to connected clients	⚠️ Needs a bus	Use onEventsAppended to fan out to a message broker; the writing instance only reaches its own clients.

Stateful process managers across instances

Process-manager state lives in the shared checkpoint and is reloaded at the start of every processing session. When two instances fold events into the same process manager concurrently, writes use optimistic concurrency: if another instance advanced the checkpoint in the meantime, the losing write is rejected, the latest state is reloaded, and the event is re-folded on top of it. This prevents lost updates without a lock.

Caution

Because each instance only folds the events it appends, and events are processed out of global order across instances, a process manager that advances its checkpoint on one instance can cause another instance to skip an event whose position is below the shared cursor. If your process-manager fold must be strictly ordered or exactly-once across the whole event log, run stateful process managers on a single dedicated worker. Commutative folds (counters, sets) and shared-store projections are unaffected.

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, options?)

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

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

Streams that handlers dispatch commands on are settled automatically when the handler completes (explicit settled() calls are not allowed inside a process manager). The optional second parameter declares defaults for those auto-settles:

// Unattended cascades: recover from residual concurrency races
// instead of surfacing them, up to 3 reload-and-re-apply attempts.
const cascadeDelete = eventStore.createProcessManager('cascade-delete', {
  settled: { maxRetries: 3 },
})

Option	Type	Description
settled	SettledOptions	Defaults applied when the process manager auto-settles dispatched streams — see Retrying on concurrency conflicts.

Caution

A process-manager-wide default asserts that all commands its handlers dispatch are pure functions of loaded state and their arguments — a handler added later inherits the retry behaviour invisibly.

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, options?)

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 })
})

Streams dispatched inside the handler are settled automatically when the transaction completes (explicit settled() calls are not allowed there). The optional second parameter declares defaults for those auto-settles:

await eventStore.transaction(async () => {
  // ...
}, { settled: { maxRetries: 3 } })

Option	Type	Description
settled	SettledOptions	Defaults applied when the transaction auto-settles dispatched streams — see Retrying on concurrency conflicts.

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
}