# Redis Adapters

Scale Courier realtime across multiple replicas with Redis Pub/Sub, Streams, and Presence adapters.

## 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.

```ts title="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.

```ts
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.

```ts title="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.

```ts title="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.

## Rate Limit Store

Holds each caller of the proxy to one budget across replicas: without it, every replica counts on its own and a caller gets the budget once per replica. Each call is taken in one Lua script, on Redis's own clock, so replicas whose clocks drift still count alike; a caller whose budget is whole again leaves no key behind.

```ts title="server.ts"
import Redis from "ioredis"
import { createRedisRateLimitStore } from "@noverax/courier-redis"

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

courierHandler({ rateLimit: { requests: 300, window: 60_000, store } })
```

One connection serves it, and it may be the one the Pub/Sub adapter publishes on. See [Rate limits](/docs/proxy-mode/overview#rate-limits).

## Presence Adapter

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

```ts title="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.

```ts title="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:

```ts title="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:

```ts
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:

```ts title="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()])
}
```

```ts title="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()`:

```ts title="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:

```ts title="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 |