Pulse

Pulse is a typed pub/sub system for services that already operate MongoDB and do not want to run a separate event broker. It provides durable consumer groups, concurrent handlers, retries, and crash recovery. Pulse is published as @orion-js/pulse, but it does not depend on Orionjs. You can use it from an Orionjs service, an older Orionjs application, or a standalone Node.js or Bun process.

Pulse replaces the event-delivery part of Kafka for services that fit MongoDB’s operational model. It is not Kafka protocol-compatible and does not provide partitions, a globally ordered log, or Kafka administration APIs.

Install

MongoDB is a peer dependency. The only direct runtime dependency is the UUIDv7 generator used for Pulse document IDs.

bun add @orion-js/pulse mongodb

Pulse supports ESM and CommonJS on Node.js 18 or later, as well as Bun.

Quick start

Define an event map to make topics and payloads type-safe:

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

type AppEvents = {
  'order.created': {
    orderId: string
    total: number
  }
  'order.cancelled': {
    orderId: string
    reason?: string
  }
}

const pulse = connect<AppEvents>({
  connectionString: process.env.MONGODB_URI!,
  consumerGroup: 'billing',
  // Optional when the URI already includes a database.
  databaseName: 'commerce',
})

await pulse.awaitConnection()

const subscription = await pulse.subscribe('order.created', async event => {
  console.log(event.id, event.data.orderId, event.attempt)
})

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

connect() returns immediately and starts initialization in the background. awaitConnection() resolves only after MongoDB is connected and every required collection and index has been created and validated. publish(), subscribe(), and subscribeBatch() wait for that same readiness promise automatically. Each Pulse client limits its MongoDB application pool to one connection by default instead of the driver default of 100. Callbacks remain concurrent because they do not retain the socket; only Pulse’s short coordination operations share it. Configure a larger maxPoolSize only when a service has measured local pool contention:

const pulse = connect<AppEvents>({
  connectionString,
  databaseName: 'commerce',
  consumerGroup: 'billing',
  maxPoolSize: 4,
  maxIdleTimeMS: 30_000,
})

Call close() during graceful shutdown:

process.once('SIGTERM', async () => {
  await subscription.unsubscribe()
  await pulse.close()
  process.exit(0)
})

Mental model

Pulse persists three kinds of records:

Record	Purpose
Event	Immutable topic, payload, headers, creation time, and retention time.
Subscription	Durable cursor and delivery configuration for one `consumerGroup + topic`.
Delivery	Queue state, lease, retries, attempts, and outcome for one or more events and a consumer group.

Publishing stores one event. Each subscribed consumer group materializes its own execution state. A normal receiver uses one delivery per event; a batch receiver stores multiple ordered event IDs in one delivery. Replicas with the same consumer group compete for those deliveries; services with different consumer groups receive independent work.

order.created
     │
     ├── billing       → one delivery shared by all billing replicas
     ├── notifications → one independent delivery
     └── analytics     → one independent delivery

Pulse writes the attempt state before it calls your handler. A renewable lease and UUIDv7 fencing token protect the attempt while it is running. Polling discovers events and recovers dead workers; Pulse never opens MongoDB Change Streams.

Choose a consumer group

Use one stable consumer group for all replicas that perform the same responsibility:

const pulse = connect<AppEvents>({
  connectionString,
  databaseName: 'commerce',
  consumerGroup: 'billing',
})

Two billing replicas share work and process each event once as a group.
billing and analytics each get an independent delivery.
Renaming a consumer group creates a new durable consumption identity.

The name should describe the responsibility, not the host, pod, or deployment replica.

Get Started

Essentials

Controllers

Other Modules

Install

Quick start

Mental model

Choose a consumer group

Next steps

​Install

​Quick start

​Mental model

​Choose a consumer group

​Next steps

Install

Quick start

Mental model

Choose a consumer group

Next steps