Consuming events

const subscription = await pulse.subscribe(
  'order.created',
  async event => {
    await createInvoice({
      eventId: event.id,
      orderId: event.data.orderId,
    })
  },
  {
    offsetReset: 'latest',
    delivery: 'at-least-once',
    maxRetries: 3,
    retryDelayMs: 1000,
    retryBackoffMultiplier: 2,
    maxConcurrency: 8,
  },
)

The handler receives:

Field	Type	Description
`id`	`string`	UUIDv7 event ID. Use it as the idempotency key for side effects.
`topic`	`string`	Subscribed topic.
`data`	event payload	Typed payload from the event map.
`headers`	`Record<string, string> \| undefined`	Producer metadata.
`createdAt`	`Date`	Event publication time.
`expiresAt`	`Date \| undefined`	Event retention deadline.
`consumerGroup`	`string`	Group handling this delivery.
`attempt`	`number`	One-based execution attempt.

A Pulse instance can have one local handler per topic. Create separate Pulse instances if one process intentionally needs two independent consumer groups.

Batch receivers

Use subscribeBatch() when one handler invocation should receive an ordered array:

await pulse.subscribeBatch(
  'invoice.created',
  async events => {
    await invoices.insertMany(
      events.map(event => ({
        eventId: event.id,
        ...event.data,
      })),
    )
  },
  {
    configVersion: 2,
    batchSize: 100,
    maxConcurrency: 2,
  },
)

batchSize is a required positive integer and a maximum, not a minimum. Pulse never waits to fill a batch. If 23 events are available, it creates and executes that partial batch immediately. Events inside the array retain publication order. Different batch handlers may run concurrently and finish in any order. One batch delivery is the retry, fencing, heartbeat, and acknowledgement unit. If the handler throws, the whole batch is retried and may repeat side effects. Use each element’s event.id as its idempotency key. Batch receivers require a transaction-capable MongoDB deployment such as a replica set, sharded cluster, or Atlas.

Consumer groups and replicas

All replicas of one service use the same consumerGroup and persisted subscription settings:

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

await pulse.subscribe('order.created', handleOrder, {
  configVersion: 1,
  delivery: 'at-least-once',
  maxConcurrency: 8,
})

Multiple replicas compete through atomic delivery leases, so only one replica owns a delivery attempt at a time. Use another consumer group for an independent responsibility:

const analyticsPulse = connect<CommerceEvents>({
  connectionString,
  databaseName: 'commerce',
  consumerGroup: 'analytics',
})

Initial offsets

offsetReset is used only when Pulse creates the durable consumerGroup + topic subscription:

Value	First subscription starts at
`latest`	Events published after the subscription is created. This is the default.
`earliest`	The oldest retained event for the topic.

The cursor is saved after Pulse materializes deliveries. Restarting a service resumes that cursor. unsubscribe() removes only the local handler; it does not delete the subscription or reset its offset.

Pulse intentionally has no administrative offset reset or replay API. Use a new consumer group when a new independent pass over retained events is appropriate.

Concurrency

Deliveries are independent and may run concurrently:

await pulse.subscribe('image.requested', resizeImage, {
  maxConcurrency: 8,
})

maxConcurrency limits running handler invocations for that topic in the current process. A normal resolve and an entire resolveBatch each consume one worker. Total concurrency also depends on workerCount and the number of replicas. It is a local execution limit and is not part of the durable subscription configuration. Queue state, retries, the active lease, terminal outcome, and the latest ten completed attempts are stored atomically on the delivery document. No separate attempt collection participates in the execution hot path.

Changing persisted configuration

Pulse stores subscription settings by consumerGroup + topic. Increase the non-negative integer configVersion when those settings change:

// Previous deployment
await pulse.subscribe('order.created', handleOrder, {
  configVersion: 1,
  maxRetries: 3,
})

// New deployment
await pulse.subscribe('order.created', handleOrder, {
  configVersion: 2,
  maxRetries: 5,
})

The highest version wins atomically. A replica with a lower version adopts the persisted winner instead of reverting it. Reusing one version with different settings throws PulseConfigurationError. Existing subscriptions and omitted versions are version zero.

Retries

The default mode is at-least-once with an initial attempt and three retries:

Attempt	Default delay before it starts
1	none
2	1 second
3	2 seconds
4	4 seconds

The delay after attempt n is:

retryDelayMs × retryBackoffMultiplier ^ (n - 1)

await pulse.subscribe('payment.captured', updateLedger, {
  maxRetries: 5,
  retryDelayMs: 2_000,
  retryBackoffMultiplier: 1.5,
})

After the final error, Pulse marks the delivery terminal. Errors retain a code, name, message, and stack when available.

At-most-once attempts

To disable handler retries:

await pulse.subscribe('metric.recorded', recordMetric, {
  delivery: 'at-most-once',
})

At-most-once resolves maxRetries to zero. Passing a positive maxRetries with this mode is a configuration error. It means one handler attempt per delivery; it cannot make side effects in external systems exactly-once.

Persisted configuration

Pulse persists these options for each consumerGroup + topic:

configVersion
offsetReset
delivery
maxRetries
retryDelayMs
retryBackoffMultiplier
receiverMode
batchSize

maxConcurrency remains local to each process.

Inspect and stop subscriptions

for (const active of pulse.getSubscriptions()) {
  console.log(active.topic, active.consumerGroup, active.maxConcurrency)
}

await subscription.unsubscribe()
await pulse.close()

getSubscriptions() reports local subscriptions. unsubscribe() is idempotent and preserves the durable MongoDB subscription. close() stops new work, waits for active handlers, releases discovery leases, and closes the MongoDB client. Read Reliability and recovery before writing handlers that perform external side effects.

Get Started

Essentials

Controllers

Other Modules

Consuming events

Batch receivers

Consumer groups and replicas

Initial offsets

Concurrency

Changing persisted configuration

Retries

At-most-once attempts

Persisted configuration

Inspect and stop subscriptions

​Subscribe to a topic

​Batch receivers

​Consumer groups and replicas

​Initial offsets

​Concurrency

​Changing persisted configuration

​Retries

​At-most-once attempts

​Persisted configuration

​Inspect and stop subscriptions

Subscribe to a topic

Batch receivers

Consumer groups and replicas

Initial offsets

Concurrency

Changing persisted configuration

Retries

At-most-once attempts

Persisted configuration

Inspect and stop subscriptions