API reference

`connect()`

function connect<TEvents extends Record<string, unknown>>(
  options: PulseConnectOptions,
): Pulse<TEvents>

connect() constructs Pulse immediately and creates one background readiness promise. Use awaitConnection() during application startup to surface connection, permission, and index errors before accepting traffic.

Connection options

Option	Type	Default	Description
`connectionString`	`string`	Required	MongoDB connection string.
`consumerGroup`	`string`	Required	Stable identity shared by replicas performing the same work.
`databaseName`	`string`	Database in URI	MongoDB database. Startup fails if neither source provides it.
`collectionPrefix`	`string`	`orionjs.pulse`	Prefix for Pulse collections.
`eventRetentionMs`	`number \| null`	7 days	Retention written on new events; `null` keeps them indefinitely.
`historyRetentionMs`	`number \| null`	7 days	Retention applied to a delivery and all its completed attempts once the delivery is terminal.
`pollIntervalMs`	`number`	`3000`	Idle interval for event discovery and work acquisition.
`workerCount`	`number`	`4`	Maximum concurrent handler executions in this process.
`maxPoolSize`	`number`	`1`	Maximum MongoDB application connections per server for this Pulse client.
`maxIdleTimeMS`	`number`	`30000`	Close idle application connections after this duration; `0` disables expiry.
`lockTimeoutMs`	`number`	`30000`	Attempt lease duration. Heartbeats renew active leases.
`discoveryLockTimeoutMs`	`number`	`10000`	Discovery-leader lease duration; controls replica failover independently from handler locks.
`onError`	`(error: Error) => void`	`console.error`	Internal coordinator and worker error reporter.

Strings must be non-empty. maxPoolSize must be a positive integer. Durations and worker counts must be finite positive numbers, except retention, retry counts, retry delays, and maxIdleTimeMS may be zero where their types allow it. Retry settings are rejected when their largest computed delay falls outside MongoDB’s supported date range. Pulse sets the MongoDB driver’s maxPoolSize to 1 instead of its default of 100, explicitly keeps minPoolSize at 0, and closes application connections after 30 idle seconds. The pool socket is used only for short coordination operations; callbacks do not retain it and can still run concurrently. The limit applies to application connections in each server pool. The driver also opens topology-monitoring connections that are not counted by maxPoolSize.

changeStreams has been removed and cannot be enabled or disabled. Pulse always uses polling. Remove the field from existing JavaScript and TypeScript configurations; providing it, even as changeStreams: 'disabled', makes connect() fail immediately with a PulseConfigurationError.

`pulse.awaitConnection()`

await pulse.awaitConnection(): Promise<void>

Resolves after the MongoDB client connects, the database is selected, and collections and indexes are ready. Workers start only after successful initialization. publish(), subscribe(), and subscribeBatch() call it internally.

`pulse.publish()`

await pulse.publish({
  topic: 'order.created',
  data: {orderId: 'ord_123'},
  headers: {source: 'checkout'},
})

Returns:

interface PulsePublishedEvent<TTopic extends string, TData> {
  id: string
  topic: TTopic
  data: TData
  headers?: Record<string, string>
  createdAt: Date
  expiresAt?: Date
}

The event _id is a UUIDv7 string. Pulse never uses MongoDB ObjectId for its own documents.

const subscription = await pulse.subscribe(topic, handler, options)

Subscription options

Option	Type	Default	Description
`configVersion`	`number`	`0`	Non-negative integer for persisted settings. Higher versions win.
`offsetReset`	`'latest' \| 'earliest'`	`latest`	Starting point when the durable subscription is first created.
`delivery`	`'at-least-once' \| 'at-most-once'`	`at-least-once`	Retry semantics.
`maxRetries`	`number`	`3`	Retries after the initial attempt. Forced to zero for at-most-once.
`retryDelayMs`	`number`	`1000`	Delay before the first retry.
`retryBackoffMultiplier`	`number`	`2`	Multiplier applied after each failed attempt.
`maxConcurrency`	`number`	`workerCount`	Local concurrent handlers for this topic.

The handler can return void or Promise<void>:

type PulseEventHandler<TTopic extends string, TData> = (
  event: PulseReceivedEvent<TTopic, TData>,
) => Promise<void> | void

PulseReceivedEvent contains every published event field plus consumerGroup and the one-based attempt number. The returned subscription contains the resolved options, including the winning configVersion, and an idempotent asynchronous unsubscribe() method.

`pulse.subscribeBatch()`

const subscription = await pulse.subscribeBatch(topic, batchHandler, {
  configVersion: 2,
  batchSize: 100,
  maxConcurrency: 2,
})

The handler receives an ordered array and can return void or Promise<void>:

type PulseBatchEventHandler<TTopic extends string, TData> = (
  events: PulseReceivedEvent<TTopic, TData>[],
) => Promise<void> | void

It accepts all ordinary subscription options and requires batchSize, a positive integer with no library-defined maximum. The configured size is a maximum. Pulse immediately executes a partial batch when fewer events are available; it does not add a batching timer. One array invocation consumes one worker and one maxConcurrency slot. Retry, fencing, heartbeat, and acknowledgement apply to the complete array. Batch materialization atomically inserts the delivery and advances its discovery cursor in a MongoDB transaction, so this API requires a replica set, sharded cluster, or Atlas.

`pulse.getSubscriptions()`

const subscriptions = pulse.getSubscriptions()

Returns the subscriptions registered in this process. It does not query durable subscriptions created by other replicas.

`pulse.close()`

await pulse.close()

Stops coordinator and worker execution, wakes sleepers, waits for background work to settle, and closes the MongoDB client. Repeated calls return the same close promise. When called and awaited from inside a handler, shutdown starts but that call returns before the shared promise so the current worker cannot deadlock waiting for itself. A call from outside the handler can await complete shutdown.

Exported errors

Error	When it is used
`PulseError`	Base class for Pulse-specific errors.
`PulseConfigurationError`	Invalid or removed options, missing database, or subscription conflicts.
`PulseIndexError`	Required index creation or compatibility validation failed.
`PulseLockLostError`	A stale worker lost its fencing token while executing or acknowledging. Usually reported through `onError`.

Collections

The names use collectionPrefix, which defaults to orionjs.pulse:

Suffix	Default collection	Purpose
`.events`	`orionjs.pulse.events`	Immutable published events.
`.subscriptions`	`orionjs.pulse.subscriptions`	Durable group/topic configuration, cursor, and discovery lease.
`.deliveries`	`orionjs.pulse.deliveries`	Queue state, attempt lease, retry history, and outcome for one group and one or more events.

Every Pulse-created _id, lock owner, and fencing token is a UUIDv7 string.

Automatically managed indexes

Pulse creates missing indexes and validates named indexes on every connection. It never drops an index automatically.

Events

Name	Keys	Options
`pulse_events_topic_sequence_id`	`{topic: 1, sequence: 1, _id: 1}`	—
`pulse_events_expires_at_ttl`	`{expiresAt: 1}`	`expireAfterSeconds: 0`

Subscriptions

Name	Keys	Options
`pulse_subscriptions_group_topic_unique`	`{consumerGroup: 1, topic: 1}`	Unique
`pulse_subscriptions_discovery_lease`	`{consumerGroup: 1, discoveryLockedUntil: 1}`	—

Deliveries

Name	Keys	Options
`pulse_deliveries_group_event_unique`	`{consumerGroup: 1, eventId: 1}`	Unique
`pulse_deliveries_sequence_acquisition`	`{consumerGroup: 1, topic: 1, status: 1, eventSequence: 1, eventId: 1}`	—
`pulse_deliveries_concurrent_pending`	`{consumerGroup: 1, nextAttemptAt: 1, createdAt: 1, topic: 1}`	Partial concurrent-pending queue
`pulse_deliveries_concurrent_processing`	`{consumerGroup: 1, lockedUntil: 1, topic: 1}`	Partial concurrent-processing queue
`pulse_deliveries_expires_at_ttl`	`{expiresAt: 1}`	`expireAfterSeconds: 0`

Full typed example

import {connect, type PulseReceivedEvent} from '@orion-js/pulse'

type Events = {
  'order.created': {orderId: string}
}

const pulse = connect<Events>({
  connectionString: process.env.MONGODB_URI!,
  databaseName: 'commerce',
  consumerGroup: 'billing',
  onError: error => console.error('[billing/pulse]', error),
})

await pulse.awaitConnection()

await pulse.subscribe(
  'order.created',
  async (event: PulseReceivedEvent<'order.created', Events['order.created']>) => {
    await handleOrder(event.data.orderId, event.id)
  },
  {
    offsetReset: 'latest',
    delivery: 'at-least-once',
    maxRetries: 3,
  },
)

Get Started

Essentials

Controllers

Other Modules

`connect()`

Connection options

`pulse.awaitConnection()`

`pulse.publish()`

Subscription options

`pulse.subscribeBatch()`

`pulse.getSubscriptions()`

`pulse.close()`

Exported errors

Collections

Automatically managed indexes

Events

Subscriptions

Deliveries

Full typed example

​connect()

​Connection options

​pulse.awaitConnection()

​pulse.publish()

​pulse.subscribe()

​Subscription options

​pulse.subscribeBatch()

​pulse.getSubscriptions()

​pulse.close()

​Exported errors

​Collections

​Automatically managed indexes

​Events

​Subscriptions

​Deliveries

​Full typed example

`connect()`

Connection options

`pulse.awaitConnection()`

`pulse.publish()`

`pulse.subscribe()`

Subscription options

`pulse.subscribeBatch()`

`pulse.getSubscriptions()`

`pulse.close()`

Exported errors

Collections

Automatically managed indexes

Events

Subscriptions

Deliveries

Full typed example