createProcessManager

Process managers are created via eventStore.createProcessManager(name). They coordinate workflows across multiple aggregate roots by listening for events and issuing commands in response.

Creation

const pm = eventStore.createProcessManager('cascade-delete')

Streams that handlers dispatch commands on are settled automatically when the handler completes — calling settled() inside a process manager throws. The optional second parameter declares defaults for those auto-settles, most notably a concurrency retry for unattended writes:

const pm = eventStore.createProcessManager('cascade-delete', {
  settled: { maxRetries: 3 },
})

See Retrying on concurrency conflicts for the retry semantics and the purity requirement it places on the dispatched commands.

Builder Methods

.withState(state)

Defines optional internal state. The state is persisted alongside checkpoints and restored on catch-up.

pm.withState({ pendingDeletions: [] })

.withEventHandlers(handlers)

Registers event handlers named on{EventName}. If state was defined, accepts a callback receiving the state.

// Without state
pm.withEventHandlers({
  async onOrganizationDeleted({ streamId }) {
    const users = await db.select().from(usersTable)
      .where(eq(usersTable.orgId, streamId))
    for (const user of users) {
      userRoot.loadStream(user.id).delete()
    }
  },
})

// With state
pm.withEventHandlers((state) => ({
  async onUserCreated({ payload }) {
    state.userCount++
  },
}))

Each handler receives (event, { refreshing }) where refreshing is true during a refresh operation.

.withAfterEffects(handlers)

Registers handlers that run after all on{EventName} handlers have completed. Uses the after{EventName} naming convention.

pm.withAfterEffects({
  async afterOrderCompleted({ payload }) {
    await sendConfirmationEmail(payload.email)
  },
})

Tip

Use withAfterEffects when your handler depends on projections being up-to-date.

.concurrent()

Disables the default exclusive per-stream locking.

State Methods

Available only when .withState() has been called:

Method	Return Type	Description
.state()	Promise<State>	Returns current state after all pending events have been processed.
.refreshState()	Promise<void>	Full state refresh — replays all events from scratch.

Other Methods

.isReady()

Returns a promise that resolves when the process manager has finished its initial catch-up.

Context Helpers

import { isInsideProcessManager, getProcessManagerInfo } from '@requence/event-sourcing'

Function	Return Type	Description
isInsideProcessManager()	boolean	true if called within a process manager event handler.
getProcessManagerInfo()	{ name, event } | undefined	Returns the process manager name and triggering event.

Caution

The library detects infinite loops when a process manager listens for and emits the same event on the same stream. An InfiniteLoopError is thrown automatically.