---
title: 'Publisher Helpers'
description: 'Publish and subscribe to events across storage backends in oRPC, with adapters, dynamic event names, and resume support for missed events.'
sidebar:
  label: 'Publisher'
---

## Installation

```package-install
npm install @orpc/publisher@beta
```

## Basic Usage

The core concept is the `Publisher` interface, which defines a standard way to publish events and subscribe to them. You can create your own publisher or use one of the provided adapters for popular storage backends. The `publish` method accepts an event name and payload, while `subscribe` lets you listen to specific events using either callback or iterator styles.

```ts twoslash
import { MemoryPublisher } from '@orpc/publisher/memory'
import { os } from '@orpc/server'
import * as z from 'zod'
// ---cut---
const publisher = new MemoryPublisher<{
  'something-updated': {
    id: string
  }
}>()

const live = os.handler(async function* ({ input, signal, lastEventId }) {
  const iterator = publisher.subscribe('something-updated', { signal, lastEventId })
  for await (const payload of iterator) {
    // Handle payload here or yield directly to client
    yield payload
  }
})

const publish = os.input(z.object({ id: z.string() })).handler(async ({ input }) => {
  await publisher.publish('something-updated', { id: input.id })
})
```

:::tip
The publisher supports both static and dynamic event names.

```ts
const publisher = new MemoryPublisher<Record<string, { message: string }>>()
```

:::

## Adapters

| Name                | Resume Support | Adapter for                                                                      |
| ------------------- | -------------- | -------------------------------------------------------------------------------- |
| `MemoryPublisher`   | ✅             | In-memory storage                                                                |
| `RedisPublisher`    | ✅             | [Redis](https://github.com/redis/redis)                                          |
| `UpstashPublisher`  | ✅             | [Upstash Redis](https://github.com/upstash/redis-js)                             |
| `BunRedisPublisher` | ✅             | [Bun's Redis](https://bun.com/docs/runtime/redis)                                |
| `DurablePublisher`  | ✅             | [Cloudflare Durable Objects](https://developers.cloudflare.com/durable-objects/) |

```ts memory
import { MemoryPublisher } from '@orpc/publisher/memory'

const publisher = new MemoryPublisher<Events>({
  /**
   * Event resume support. `true` enables it, an object enables and customizes it.
   *
   * When enabled, published events are temporarily stored so new
   * subscribers can resume from a previous position using `lastEventId`.
   *
   * @default false
   */
  resume: {
    /**
     * How long to retain events for resume, in milliseconds.
     *
     * Expired events are cleaned up lazily for performance reasons, so
     * some events may remain available slightly longer than this period.
     *
     * @default 300000 (5 min)
     */
    ttl: 300_000,
  },
})
```

```ts redis
import { createClient } from 'redis'
import { RedisPublisher } from '@orpc/publisher/redis'

const client = createClient({ url: 'redis://localhost:6379' })

client.on('error', (error) => console.error('Redis client error', error))

// RedisPublisher lazily connects to Redis when needed.
// You can still call `client.connect()` manually, but it is optional.
await client.connect()

const publisher = new RedisPublisher<Events>(client, {
  /**
   * Redis subscriber instance.
   * Pub/Sub takes over the connection, so a client with subscriptions
   * cannot execute commands and must use a dedicated connection.
   *
   * @default client.duplicate()
   */
  subscriber: client.duplicate(),

  /**
   * The prefix to use for Redis keys and Pub/Sub channels.
   * Set it to isolate events when several apps share one Redis instance.
   *
   * @default ''
   */
  prefix: '',

  /**
   * Serializer for serialize and deserialize payloads.
   *
   * @default RPCJsonSerializer
   */
  serializer: undefined,

  /**
   * Event resume support. `true` enables it, an object enables and customizes it.
   *
   * When enabled, published events are temporarily stored so new
   * subscribers can resume from a previous position using `lastEventId`.
   *
   * @default false
   */
  resume: {
    /**
     * How long to retain events for resume, in milliseconds.
     *
     * Expired events are cleaned up lazily for performance reasons, so
     * some events may remain available slightly longer than this period.
     *
     * @default 300000 (5 min)
     */
    ttl: 300_000,
  },
})

// Creates a publisher with an extra prefix that shares this client, e.g. per tenant
const tenantPublisher = publisher.prefix('tenant:123:')
```

```ts upstash
import { Redis } from '@upstash/redis'
import { UpstashPublisher } from '@orpc/publisher/upstash'

const redis = Redis.fromEnv()

const publisher = new UpstashPublisher<Events>(redis, {
  /**
   * The prefix to use for Redis keys and Pub/Sub channels.
   * Set it to isolate events when several apps share one Redis instance.
   *
   * @default ''
   */
  prefix: '',

  /**
   * Serializer for serialize and deserialize payloads.
   *
   * @default RPCJsonSerializer
   */
  serializer: undefined,

  /**
   * Event resume support. `true` enables it, an object enables and customizes it.
   *
   * When enabled, published events are temporarily stored so new
   * subscribers can resume from a previous position using `lastEventId`.
   *
   * @default false
   */
  resume: {
    /**
     * How long to retain events for resume, in milliseconds.
     *
     * Expired events are cleaned up lazily for performance reasons, so
     * some events may remain available slightly longer than this period.
     *
     * @default 300000 (5 min)
     */
    ttl: 300_000,
  },
})

// Creates a publisher with an extra prefix that shares this client, e.g. per tenant
const tenantPublisher = publisher.prefix('tenant:123:')
```

```ts bun
import { BunRedisPublisher } from '@orpc/bun'
import { redis } from 'bun'

const publisher = new BunRedisPublisher<Events>(redis, {
  /**
   * Redis subscriber instance.
   * Pub/Sub takes over the connection, so a client with subscriptions
   * cannot execute commands and must use a dedicated connection.
   *
   * @remarks
   * The publisher takes over `onconnect` and `onclose` to detect lost subscriptions.
   * Set your own handlers before the first listen, or they replace the publisher's.
   *
   * @default redis.duplicate() (lazily created on first listen)
   */
  subscriber: redis.duplicate(),

  /**
   * The prefix to use for Redis keys and Pub/Sub channels.
   * Set it to isolate events when several apps share one Redis instance.
   *
   * @default ''
   */
  prefix: '',

  /**
   * Serializer for serialize and deserialize payloads.
   *
   * @default RPCJsonSerializer
   */
  serializer: undefined,

  /**
   * Event resume support. `true` enables it, an object enables and customizes it.
   *
   * When enabled, published events are temporarily stored so new
   * subscribers can resume from a previous position using `lastEventId`.
   *
   * @default false
   */
  resume: {
    /**
     * How long to retain events for resume, in milliseconds.
     *
     * Expired events are cleaned up lazily for performance reasons, so
     * some events may remain available slightly longer than this period.
     *
     * @default 300000 (5 min)
     */
    ttl: 300_000,
  },
})

// Creates a publisher with an extra prefix that shares this client, e.g. per tenant
const tenantPublisher = publisher.prefix('tenant:123:')
```

```ts cloudflare
import { DurablePublisher, DurablePublisherObject } from '@orpc/cloudflare'

export class PublisherDO extends DurablePublisherObject {
  constructor(ctx: DurableObjectState, env: Env) {
    super(ctx, env, {
      /**
       * Event resume support. `true` enables it, an object enables and customizes it.
       *
       * When enabled, published events are temporarily stored so new
       * subscribers can resume from a previous position using `lastEventId`.
       * Each stored event must fit within the Durable Object SQLite row size limit,
       * so publishing a larger one fails.
       *
       * @default false
       */
      resume: {
        /**
         * How long to retain events for resume, in milliseconds.
         *
         * Expired events are cleaned up lazily for performance reasons, so
         * some events may remain available slightly longer than this period.
         *
         * @default 300000 (5 min)
         */
        ttl: 300_000,

        /**
         * Interval between cleanup checks for the Durable Object, in milliseconds.
         *
         * At each interval, verify whether the Durable Object is inactive
         * (no active WebSocket connections and no stored events). If inactive, all
         * data is deleted to free resources; otherwise, another check is scheduled.
         *
         * @default 6 * 60 * 60 * 1000 (6 hours)
         */
        cleanupInterval: 6 * 60 * 60 * 1000,

        /**
         * Prefix for the resume storage table schema.
         * Used to avoid naming conflicts with other tables in the same Durable Object.
         *
         * @default 'orpc:'
         */
        schemaPrefix: 'orpc:',
      },
    })
  }
}

export default {
  async fetch(request, env) {
    const publisher = new DurablePublisher<Events>(env.PUBLISHER_DON, {
      /**
       * Prefix for events, to avoid naming conflicts with other publishers in the same Durable Object Namespace.
       *
       * @default ''
       */
      prefix: '',

      /**
       * Serializer for serialize and deserialize payloads.
       *
       * @default RPCJsonSerializer
       */
      serializer: undefined,

      /**
       * Custom function to get the Durable Object stub for publishing.
       *
       * @default ((namespace, event) => namespace.getByName(event))
       */
      getStubByName: (namespace, event) => namespace.getByName(event),
    })

    // Creates a publisher with an extra prefix that shares this namespace, e.g. per tenant
    const tenantPublisher = publisher.prefix('tenant:123:')
  },
}
```

## Resume Missing Events

Some adapters can resume events missed while a subscriber is offline. This feature is usually disabled by default, but you can enable it when creating the publisher with `resume: true`, or with an object to customize it. When enabled, the publisher automatically manages event ids and attempts to deliver events since the last event id provided by the subscriber.

```ts
const publisher = new MemoryPublisher({
  resume: true, // Enable resuming missed events
})

const customized = new MemoryPublisher({
  resume: {
    ttl: 5 * 60 * 1000, // How long to retain events, in milliseconds
  },
})

const iterator = publisher.subscribe('something-updated', {
  signal,
  lastEventId, // The publisher will attempt to deliver missed events since this event id
})
```

:::warning
When resume is enabled, the publisher manages event ids automatically. This means:

- Any event id provided during publishing is ignored
- When subscribing, you must preserve and forward the event id when yielding custom payloads

```ts
import { getEventMeta, withEventMeta } from '@orpc/server'

const live = os.handler(async function* ({ input, signal, lastEventId }) {
  const iterator = publisher.subscribe('something-updated', { signal, lastEventId })
  for await (const payload of iterator) {
    // Preserve event id when yielding custom payloads
    const id = getEventMeta(payload)?.id
    yield withEventMeta({ custom: 'value' }, { id })
  }
})

const publish = os.input(z.object({ id: z.string() })).handler(async ({ input }) => {
  // The event id 'this-will-be-ignored' will be replaced by the publisher
  await publisher.publish(
    'something-updated',
    withEventMeta({ id: input.id }, { id: 'this-will-be-ignored' }),
  )
})
```

:::

### Client Reconnection

On the client, you can use the [Retry Plugin](/docs/plugins/retry), which automatically controls and passes `lastEventId` to the server when reconnecting. Alternatively, you can manage `lastEventId` manually:

```ts
import { getEventMeta } from '@orpc/client'

let lastEventId: string | undefined

while (true) {
  try {
    const iterator = await client.live('input', { lastEventId })

    for await (const payload of iterator) {
      lastEventId = getEventMeta(payload)?.id // Update lastEventId

      console.log(payload)
    }
  } catch {
    await new Promise((resolve) => setTimeout(resolve, 1000)) // Wait 1 second before retrying
  }
}
```