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 all four 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`	Event discovery, work acquisition, and reconciliation interval.
`workerCount`	`number`	`4`	Maximum concurrent handler executions in this process.
`maxPoolSize`	`number`	`5`	Maximum MongoDB application connections per server for this Pulse client.
`lockTimeoutMs`	`number`	`30000`	Attempt and ordered 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, and retry delays 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 5 instead of its default of 100. minPoolSize remains at the driver default of 0, so the pool grows on demand. The limit applies to application connections in each server pool; the driver also opens monitoring connections for the MongoDB topology.

changeStreams has been removed and cannot be enabled or disabled. Pulse always uses polling and reconciliation. 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 history.find() 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
`ordered`	`boolean`	`true`	Serialize this group and topic across all replicas.
`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 when `ordered` is false. Forced to one when ordered.

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 and an idempotent asynchronous unsubscribe() method.

`pulse.getSubscriptions()`

const subscriptions = pulse.getSubscriptions()

Returns the subscriptions registered in this process. It does not query durable subscriptions created by other replicas. Use the dashboard to inspect the full database state.

`pulse.history.find()`

const page = await pulse.history.find({
  topic: 'order.created',
  status: 'error',
  from: new Date(Date.now() - 24 * 60 * 60 * 1000),
  limit: 100,
})

History filters

Option	Type	Description
`topic`	`string`	Exact topic.
`eventId`	`string`	Exact event UUIDv7.
`consumerGroup`	`string`	Exact consumer group.
`status`	`'pending' \| 'success' \| 'error'`	Attempt state.
`lockState`	`'queued' \| 'active' \| 'expired'`	Pending lock classification. Implies `status: 'pending'`.
`from`	`Date`	Inclusive lower bound on attempt `createdAt`.
`to`	`Date`	Inclusive upper bound on attempt `createdAt`.
`cursor`	`string`	`nextCursor` from the previous page.
`limit`	`number`	Defaults to 100 and is clamped from 1 to 500.

Results are sorted by UUIDv7 _id descending:

interface PulseHistoryFindResult {
  records: PulseHistoryRecord[]
  nextCursor?: string
}

Each record exposes delivery and event IDs, topic, group, attempt number, status, lock state and timestamps, duration, expiration, and serialized error.

`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/ordered leases.
`.deliveries`	`orionjs.pulse.deliveries`	Logical group outcome for an event.
`.history`	`orionjs.pulse.history`	Individual execution attempts and their locks.

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_created_id`	`{topic: 1, createdAt: 1, _id: 1}`	—
`pulse_events_topic_sequence_id`	`{topic: 1, sequence: 1, _id: 1}`	—
`pulse_events_legacy_topic_created_id`	`{topic: 1, sequence: 1, createdAt: 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_ordered_lease`	`{consumerGroup: 1, orderedLockedUntil: 1}`	—
`pulse_subscriptions_discovery_lease`	`{consumerGroup: 1, discoveryLockedUntil: 1}`	—

Deliveries

Name	Keys	Options
`pulse_deliveries_group_event_unique`	`{consumerGroup: 1, eventId: 1}`	Unique
`pulse_deliveries_acquisition`	`{consumerGroup: 1, topic: 1, status: 1, eventCreatedAt: 1, eventId: 1}`	—
`pulse_deliveries_sequence_acquisition`	`{consumerGroup: 1, topic: 1, status: 1, eventSequence: 1, eventId: 1}`	—
`pulse_deliveries_dashboard_pending`	`{status: 1, eventCreatedAt: 1, eventId: 1}`	—
`pulse_deliveries_expires_at_ttl`	`{expiresAt: 1}`	`expireAfterSeconds: 0`

History

Name	Keys	Options
`pulse_history_delivery_attempt_unique`	`{deliveryId: 1, attempt: 1}`	Unique
`pulse_history_group_topic_created`	`{consumerGroup: 1, topic: 1, createdAt: -1, _id: -1}`	—
`pulse_history_event`	`{eventId: 1, createdAt: -1}`	—
`pulse_history_dead_locks`	`{status: 1, lockedUntil: 1}`	—
`pulse_history_group_dead_locks`	`{consumerGroup: 1, status: 1, lockedUntil: 1}`	—
`pulse_history_pending_acquisition`	`{consumerGroup: 1, topic: 1, status: 1, nextAttemptAt: 1, createdAt: 1}`	—
`pulse_history_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)
  },
  {
    ordered: true,
    offsetReset: 'latest',
    delivery: 'at-least-once',
    maxRetries: 3,
  },
)

Get Started

Essentials

Controllers

Other Modules

`connect()`

Connection options

`pulse.awaitConnection()`

`pulse.publish()`

Subscription options

`pulse.getSubscriptions()`

`pulse.history.find()`

History filters

`pulse.close()`

Exported errors

Collections

Automatically managed indexes

Events

Subscriptions

Deliveries

History

Full typed example

​connect()

​Connection options

​pulse.awaitConnection()

​pulse.publish()

​pulse.subscribe()

​Subscription options

​pulse.getSubscriptions()

​pulse.history.find()

​History filters

​pulse.close()

​Exported errors

​Collections

​Automatically managed indexes

​Events

​Subscriptions

​Deliveries

​History

​Full typed example

`connect()`

Connection options

`pulse.awaitConnection()`

`pulse.publish()`

`pulse.subscribe()`

Subscription options

`pulse.getSubscriptions()`

`pulse.history.find()`

History filters

`pulse.close()`

Exported errors

Collections

Automatically managed indexes

Events

Subscriptions

Deliveries

History

Full typed example