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, ordered or concurrent handlers, retries, execution history, crash recovery, and a built-in operations dashboard. 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 history reads wait for that same readiness promise automatically. Call close() during graceful shutdown:

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

Mental model

Pulse persists four 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	Logical outcome of one event for one consumer group.
History	One execution attempt, including pending state, lease, result, duration, and error.

Publishing stores one event. Each subscribed consumer group materializes its own delivery and attempt history. Replicas with the same consumer group compete for that delivery; services with different consumer groups each receive it.

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

Pulse writes the attempt as pending before it calls your handler. A renewable lease and UUIDv7 fencing token protect the attempt while it is running. Polling and reconciliation recover partial writes or dead workers; MongoDB Change Streams can reduce latency but are never the source of correctness.

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