Redis Adapters — Courier Docs

When You Need Redis

Memory adapters work perfectly for single-process deployments. Switch to Redis when:

Running multiple replicas (K8s, Docker Swarm, PM2 cluster)
A WS message published on replica A must reach clients on replica B
SSE clients reconnect to a different replica and need event history
Presence tracking must be consistent across all replicas

The Pub/Sub adapter is the one the server uses itself, once you pass it as pubsub. The Stream and Presence adapters are building blocks: the server does not read or write them on its own, and your stream and channel handlers call them, as the examples below do.

Installation

bash

npm install @noverax/courier-redis ioredis

ioredis is a peer dependency, version 5.3 or later — you manage the Redis connection yourself.

Pub/Sub Adapter

Routes WebSocket broadcast messages across replicas via Redis Pub/Sub: a channel handler's ctx.broadcast, and broadcast, broadcastToTopic and broadcastToBaseTopic on the handler's connection registry (server.wsConnections), reach the clients of every replica.

adapters.ts

import Redis from "ioredis"
import { createRedisPubSub } from "@noverax/courier-redis"

const pubsub = createRedisPubSub({
  publisher: new Redis(process.env.REDIS_URL ?? "redis://localhost:6379"),    // dedicated connection for PUBLISH
  subscriber: new Redis(process.env.REDIS_URL ?? "redis://localhost:6379"),   // dedicated connection for SUBSCRIBE
  prefix: "courier:",                                // channel key prefix (default: "")
})

await pubsub.connect()

Two connections required

Redis enters subscriber mode on SUBSCRIBE, blocking other commands. You must provide separate publisher and subscriber connections.

Echo Prevention

Each adapter instance generates a unique origin UUID. Messages are tagged with this origin, and the adapter skips messages from itself — preventing echo loops in multi-replica setups.

const pubsub = createRedisPubSub({
  publisher: new Redis(process.env.REDIS_URL ?? "redis://localhost:6379"),
  subscriber: new Redis(process.env.REDIS_URL ?? "redis://localhost:6379"),
  origin: process.env.HOSTNAME,  // optional: use pod name as origin
})

Stream Adapter

Keeps a stream's events in Redis Streams, so a client that reconnects to any replica can be sent what it missed.

adapters.ts

import Redis from "ioredis"
import { createRedisStream } from "@noverax/courier-redis"

const stream = createRedisStream({
  redis: new Redis(process.env.REDIS_URL ?? "redis://localhost:6379"),
  prefix: "courier:stream:",      // stream key prefix (default: "")
  maxLen: 1000,                  // auto-trim MAXLEN per stream (default: 1000, 0 = no trim)
})

await stream.connect()

Resuming a Stream

A stream handler resumes from the adapter itself. The code that produces an event appends it, and Redis gives it an id; the handler reads the stream after the last id its client saw — ctx.lastEventId, which the client sends as Last-Event-ID when it reconnects — and yields each event with tracked(id, …), so the id it resumes from next time is the Redis one. Every replica reads the same Redis stream, so it does not matter which one the client comes back to.

api/activity.ts

import { courier, defineGroup } from "@noverax/courier/define"
import { tracked } from "@noverax/courier/realtime"
import { z } from "zod"
import { history } from "../lib/adapters"

const { stream } = courier()

const Activity = z.object({ user: z.string(), action: z.string() })

/** Wait `ms`, or less when the client leaves. */
function pause(ms: number, signal: AbortSignal): Promise<void> {
  return new Promise((resolve) => {
    const timer = setTimeout(resolve, ms)
    signal.addEventListener("abort", () => {
      clearTimeout(timer)
      resolve()
    }, { once: true })
  })
}

export const activity = defineGroup({
  prefix: "activity",
  endpoints: {
    feed: stream
      .variables(z.object({ team: z.string() }))
      .response(Activity)
      .implement(async function* (ctx) {
        const topic = `activity:${ctx.variables.team}`
        // After the last event this client saw, or from now for a new one
        let cursor = ctx.lastEventId ?? `${Date.now()}-0`
        while (!ctx.signal.aborted) {
          const events = await history.getHistory(topic, cursor)
          for (const event of events) {
            cursor = event.id
            yield tracked(event.id, Activity.parse(event.data))
          }
          if (events.length === 0) await pause(500, ctx.signal)
        }
      }),
  },
})

/** Called wherever activity happens, on any replica. */
export async function recordActivity(team: string, user: string, action: string): Promise<void> {
  await history.append(`activity:${team}`, { data: { user, action }, timestamp: Date.now() })
}

getHistory(topic, afterId, limit) runs Redis XRANGE from just after afterId — exclusive, Redis 6.2 or later — and returns at most limit events, 100 by default; the loop above reads on until it has caught up. A client that was away longer than the stream keeps events for resumes from the oldest one left.

Auto-Trim

Streams are trimmed on every append() using Redis approximate MAXLEN ~. This keeps memory bounded while allowing minor overallocation for performance.

Presence Adapter

Tracks online users across replicas using Redis ZSET (scores) + HASH (metadata).

adapters.ts

import Redis from "ioredis"
import { createRedisPresence } from "@noverax/courier-redis"

const presence = createRedisPresence({
  redis: new Redis(process.env.REDIS_URL ?? "redis://localhost:6379"),
  prefix: "courier:presence:",    // key prefix (default: "presence:")
})

await presence.connect()

Presence is used from a channel's handlers: track a connection when it joins, refresh it on each of its heartbeats, untrack it when it leaves, and tell the room. Give the adapter a subscriber connection as well, and onDiff hears the joins and leaves of every replica, not only its own.

api/rooms.ts

import { courier, defineGroup } from "@noverax/courier/define"
import { z } from "zod"
import { presence } from "../lib/adapters"

const { channel } = courier()

const Online = z.object({ online: z.array(z.string()) })

async function onlineIn(topic: string): Promise<z.infer<typeof Online>> {
  const entries = await presence.list(topic)
  return { online: [...new Set(entries.map((entry) => entry.userId))] }
}

export const rooms = defineGroup({
  prefix: "rooms",
  endpoints: {
    room: channel
      .variables(z.object({ room: z.string(), user: z.string() }))
      .topic((v) => ({ room: v.room }))
      .send(z.object({ type: z.literal("heartbeat") }))
      .response(Online)
      .implement((v) => {
        const topic = `room:${v.room}`
        return {
          async onJoin(ctx) {
            await presence.track(topic, { userId: v.user, connectionId: ctx.connectionId, metadata: {} })
            const online = await onlineIn(topic)
            ctx.send(online)
            ctx.broadcast(ctx.topic, online)
          },
          async onMessage(ctx) {
            await presence.heartbeat(topic, ctx.connectionId)
          },
          async onLeave(ctx) {
            await presence.untrack(topic, ctx.connectionId)
            ctx.broadcast(ctx.topic, await onlineIn(topic))
          },
        }
      }),
  },
})

A replica that crashes never untracks its connections; a sweep, below, removes them once their heartbeats stop.

Data Structure

Each topic creates two Redis keys:

Key	Type	Purpose
`{prefix}score:{topic}`	ZSET	member = connectionId, score = lastSeenAt (ms)
`{prefix}meta:{topic}`	HASH	field = connectionId, value = packed JSON

Heartbeat & Sweep

heartbeat(topic, connectionId) refreshes a connection's score, as the room's onMessage does above each time the browser sends { type: "heartbeat" } — every 15 seconds, say. A sweep, run on a timer by one process or by each, removes the connections that stopped:

presence-sweep.ts

import { presence } from "./lib/adapters"

// Remove entries without a heartbeat in the last 30 seconds (atomic Lua
// script). The second argument is a duration, not a timestamp.
setInterval(async () => {
  const stale = await presence.sweep("room:general", 30_000)
  // stale = [{ connectionId, userId, metadata, joinedAt, lastSeenAt }]
  if (stale.length > 0) console.log(`swept ${stale.length} from room:general`)
}, 10_000)

Diff Events

Subscribe to join/leave events per topic:

const unsubscribe = presence.onDiff("room:general", (diff) => {
  for (const entry of diff.joins) {
    console.log(`${entry.userId} joined`)
  }
  for (const entry of diff.leaves) {
    console.log(`${entry.userId} left`)
  }
})

Full Setup Example

Create the adapters once, pass the Pub/Sub adapter to the server, and import the other two where your handlers use them — history and presence in the examples above:

lib/adapters.ts

import Redis from "ioredis"
import { createRedisPresence, createRedisPubSub, createRedisStream } from "@noverax/courier-redis"

const redisUrl = process.env.REDIS_URL ?? "redis://localhost:6379"

export const pubsub = createRedisPubSub({
  publisher: new Redis(redisUrl),
  subscriber: new Redis(redisUrl),
  prefix: "courier:",
})

export const history = createRedisStream({
  redis: new Redis(redisUrl),
  prefix: "courier:stream:",
  maxLen: 5000,
})

export const presence = createRedisPresence({
  redis: new Redis(redisUrl),
  subscriber: new Redis(redisUrl),
  prefix: "courier:presence:",
})

export async function connectAdapters(): Promise<void> {
  await Promise.all([pubsub.connect(), history.connect(), presence.connect()])
}

export async function disconnectAdapters(): Promise<void> {
  await Promise.all([pubsub.disconnect(), history.disconnect(), presence.disconnect()])
}

server.ts

import { createCourierServer } from "@noverax/courier-server"
import { activity } from "./api/activity"
import { rooms } from "./api/rooms"
import { connectAdapters, disconnectAdapters, pubsub } from "./lib/adapters"
import { defineApi } from "@noverax/courier/define"

const api = defineApi({ name: "api", baseURL: "https://api.example.com", groups: { activity, rooms } })

await connectAdapters()

// `pubsub` carries ctx.broadcast and the registry's broadcasts to every replica
const server = await createCourierServer({ sources: { api }, pubsub, port: 3001 })

process.on("SIGTERM", async () => {
  await server.stop()
  await disconnectAdapters()
})

Health Monitoring

Each adapter exposes isHealthy():

health-check.ts

import { history, presence, pubsub } from "./lib/adapters"

app.get("/health/redis", (req, res) => {
  const healthy = pubsub.isHealthy() && history.isHealthy() && presence.isHealthy()
  res.status(healthy ? 200 : 503).json({
    pubsub: pubsub.isHealthy(),
    history: history.isHealthy(),
    presence: presence.isHealthy(),
  })
})

Health is tracked automatically via Redis connection events (ready, close, end).

Error Handling

All adapters expose onError() for observability:

error-handling.ts

import { history, presence, pubsub } from "./lib/adapters"

const unsubPubSub = pubsub.onError((error) => {
  console.error("[PubSub]", error)
})

const unsubHistory = history.onError((error) => {
  console.error("[Stream]", error)
})

const unsubPresence = presence.onError((error) => {
  console.error("[Presence]", error)
})

Failover Behavior

Redis adapters degrade gracefully — they never kill active client connections:

Scenario	Behavior
Redis temporarily down	PubSub messages are lost (ephemeral). `append` rejects and `getHistory` returns no events, and `onError` hears both. Presence calls do nothing — `list` returns no entries — and `onError` hears why.
Redis reconnects	Adapters resume automatically (ioredis handles reconnection)
Subscriber connection lost	New subscriptions re-established on reconnect
`isHealthy()` returns false	Your load balancer should stop sending new connections

ioredis reconnection

ioredis reconnects automatically by default with exponential backoff. Configure via retryStrategy in the Redis constructor.

Redis Key Reference

Adapter	Key Pattern	TTL
PubSub	`{prefix}{topic}`, `{prefix}{baseTopic}#` (every qualified topic of an endpoint), `{prefix}` (every connection)	None (Pub/Sub channels)
Stream	`{prefix}{topic}`	Auto-trim via MAXLEN
Presence (score)	`{prefix}score:{topic}`	Managed by sweep
Presence (meta)	`{prefix}meta:{topic}`	Managed by sweep