Replication Guide — Courier Docs

This guide shows a complete, production-ready realtime setup with Redis adapters for each supported framework.

Prerequisites

All examples assume you have created shared adapters as shown in the Redis Adapters page. The adapters.ts file is reused across all frameworks.

Shared Adapters

Every framework uses the same adapter setup. The server is handed pubsub; history and presence are used by your stream and channel handlers, as the Redis Adapters page shows:

lib/adapters.ts

import Redis from "ioredis"
import { createRedisPubSub, createRedisStream, createRedisPresence } 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:",
  origin: process.env.HOSTNAME,
})

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()])
}

Standalone Server

The simplest setup — one process handles HTTP, SSE, and WebSocket.

server.ts

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

await connectAdapters()

const server = await createCourierServer({
  sources,
  port: Number(process.env.PORT) || 3001,
  host: "0.0.0.0",
  allowedOrigins: [process.env.ALLOWED_ORIGIN!],
  pubsub,

  // SSE
  maxSSEConnections: 10_000,
  sseMaxLifetimeMs: 300_000,
  onSSEConnect: ({ connectionId, request, source, group, method }) => {
    return { userId: extractUser(request) }
  },
  onSSEDisconnect: ({ meta, reason, eventCount, duration }) => {
    console.log(`SSE closed: user=${meta.userId} reason=${reason} events=${eventCount}`)
  },

  // WebSocket
  heartbeatIntervalMs: 25_000,
  onConnect: ({ connectionId, request }) => {
    const userId = extractUser(request)
    if (!userId) throw new Error("Unauthorized")
    return { userId }
  },
  onDisconnect: ({ meta, code }) => {
    console.log("WS closed", { meta, code })
  },

  onReady: (url) => console.log(`Courier ready at ${url}`),
})

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

Next.js

Next.js App Router supports SSE natively via sseRoute(). WebSocket requires a separate Courier server process.

SSE Route

app/api/sse/route.ts

import { sseRoute } from "@noverax/courier-next"

export const { GET } = sseRoute({
  maxSSEConnections: 5000,
  sseMaxLifetimeMs: 300_000,
  sseHeaders: {
    "Access-Control-Allow-Origin": process.env.NEXT_PUBLIC_APP_URL!,
  },
  onSSEConnect: ({ connectionId, request }) => {
    return { userId: extractUser(request) }
  },
})

HTTP Proxy Route

app/api/request/route.ts

import { apiRoute } from "@noverax/courier-next"

export const { GET, POST } = apiRoute()

Sources come from the core global registry populated in instrumentation.ts — see the Next.js guide.

Separate WS Server

ws-server.ts

import { createCourierServer } from "@noverax/courier-server"
import { sources } from "./api/sources"
import { pubsub, connectAdapters } from "./lib/adapters"

await connectAdapters()

const server = await createCourierServer({
  sources,
  port: Number(process.env.WS_PORT) || 3001,
  host: "0.0.0.0",
  pubsub,
  allowedOrigins: [process.env.NEXT_PUBLIC_APP_URL!],
  onConnect: ({ request }) => {
    const userId = extractUser(request)
    if (!userId) throw new Error("Unauthorized")
    return { userId }
  },
})

Project Structure

Project structure

├── app/
│   ├── api/
│   │   ├── request/route.ts    ← HTTP proxy
│   │   └── sse/route.ts        ← SSE streaming
│   └── page.tsx
├── ws-server.ts                ← separate WS process
├── lib/
│   └── adapters.ts             ← shared Redis adapters
└── api/
    └── sources.ts

Two processes

Next.js runs SSE via its Edge/Node runtime. WebSocket runs as a separate ws-server.ts process. Deploy them independently — they share state via Redis.

Express

server.ts

import http from "node:http"
import express from "express"
import { registerSources } from "@noverax/courier"
import { courierHandler, courierSSEHandler, courierWSAttach } from "@noverax/courier-express"
import { sources } from "./api/sources"
import { pubsub, connectAdapters, disconnectAdapters } from "./lib/adapters"

registerSources(sources)
await connectAdapters()

const app = express()
const server = http.createServer(app)

// HTTP proxy
const proxy = courierHandler()
app.all("/api/request", proxy)

// SSE streaming — do NOT apply compression middleware to this route
const sse = courierSSEHandler({
  maxSSEConnections: 10_000,
  sseMaxLifetimeMs: 300_000,
  onSSEConnect: ({ connectionId, request }) => {
    return { userId: extractUser(request) }
  },
  onSSEDisconnect: ({ meta, reason, eventCount }) => {
    console.log(`SSE closed: ${meta.userId}, ${reason}, ${eventCount} events`)
  },
})
app.get("/api/sse", sse)

// WebSocket — answers upgrades on `path` only; set the client's proxyPath.ws to match
const cleanupWS = courierWSAttach(server, {
  path: "/api/ws",
  pubsub,
  heartbeatIntervalMs: 25_000,
  allowedOrigins: [process.env.ALLOWED_ORIGIN!],
  maxSubscriptionsPerConnection: 100,
  maxMessageSize: 65_536,
  onConnect: ({ request }) => {
    const userId = extractUser(request)
    if (!userId) throw new Error("Unauthorized")
    return { userId }
  },
  onDisconnect: ({ meta, code }) => {
    console.log("WS closed", { meta, code })
  },
})

// Health checks
app.get("/health", (req, res) => res.json({ status: "ok" }))
app.get("/readyz", (req, res) => {
  res.status(shuttingDown ? 503 : 200).send(shuttingDown ? "draining" : "ready")
})

let shuttingDown = false
server.listen(Number(process.env.PORT) || 3001, "0.0.0.0")

process.on("SIGTERM", async () => {
  shuttingDown = true
  // server.close() waits for open streams and sockets, so drain them first:
  // SSE clients get a `courier:shutdown` notice, WebSocket clients a 1012 close.
  await Promise.all([proxy.shutdownSSE(), sse.shutdownSSE(), cleanupWS()])
  server.close()
  await disconnectAdapters()
})

SSE and compression

Do not apply compression middleware (like compression()) to the SSE route. Compression buffers the response, breaking the streaming behavior.

Fastify

server.ts

import Fastify from "fastify"
import fastifyWebsocket from "@fastify/websocket"
import { wsFrameLimit } from "@noverax/courier/node"
import { registerSources } from "@noverax/courier"
import { courierPlugin, courierSSEHandler, courierWSHandler } from "@noverax/courier-fastify"
import { sources } from "./api/sources"
import { pubsub, connectAdapters, disconnectAdapters } from "./lib/adapters"

registerSources(sources)
await connectAdapters()

const fastify = Fastify({ logger: true })

// WebSocket plugin (must register before WS routes). Its ws server reads
// frames up to maxPayload before Courier sees them — 100 MiB by default —
// so bound it as Courier's own servers do (wsFrameLimit from "@noverax/courier/node")
await fastify.register(fastifyWebsocket, { options: { maxPayload: wsFrameLimit() } })

// HTTP proxy
await fastify.register(courierPlugin, { prefix: "/api/request" })

// SSE streaming — the plugin drains its own streams on close; this route
// drains in preClose, or fastify.close() waits for its streams to end
const sse = courierSSEHandler({
  maxSSEConnections: 10_000,
  sseMaxLifetimeMs: 300_000,
  onSSEConnect: ({ connectionId, request }) => {
    return { userId: extractUser(request) }
  },
  onSSEDisconnect: ({ meta, reason, eventCount }) => {
    console.log(`SSE closed: ${meta.userId}, ${reason}, ${eventCount} events`)
  },
})
fastify.get("/api/sse", sse)
fastify.addHook("preClose", () => sse.shutdownSSE())

// WebSocket
fastify.get("/ws", { websocket: true }, courierWSHandler({
  pubsub,
  heartbeatIntervalMs: 25_000,
  allowedOrigins: [process.env.ALLOWED_ORIGIN!],
  maxSubscriptionsPerConnection: 100,
  onConnect: ({ request }) => {
    const userId = extractUser(request)
    if (!userId) throw new Error("Unauthorized")
    return { userId }
  },
  onDisconnect: ({ meta, code }) => {
    console.log("WS closed", { meta, code })
  },
}))

// Health checks
fastify.get("/health", async () => ({ status: "ok" }))
fastify.get("/readyz", async (req, reply) => {
  reply.status(shuttingDown ? 503 : 200).send(shuttingDown ? "draining" : "ready")
})

let shuttingDown = false
await fastify.listen({ port: Number(process.env.PORT) || 3001, host: "0.0.0.0" })

process.on("SIGTERM", async () => {
  shuttingDown = true
  await fastify.close()
  await disconnectAdapters()
})

Hono

Hono is runtime-agnostic — this example uses Node.js with @hono/node-server, which carries its own WebSocket upgrade helper.

server.ts

import { Hono } from "hono"
import { serve, upgradeWebSocket } from "@hono/node-server"
import { WebSocketServer } from "ws"
import { registerSources } from "@noverax/courier"
import { courierHandler, courierSSEHandler, courierWSHandler } from "@noverax/courier-hono"
import { sources } from "./api/sources"
import { pubsub, connectAdapters, disconnectAdapters } from "./lib/adapters"

registerSources(sources)
await connectAdapters()

const app = new Hono()
const wss = new WebSocketServer({ noServer: true })

// HTTP proxy
app.all("/api/request", courierHandler())

// SSE streaming
app.get("/api/sse", courierSSEHandler({
  maxSSEConnections: 10_000,
  sseMaxLifetimeMs: 300_000,
  onSSEConnect: ({ connectionId, request }) => {
    return { userId: extractUser(request) }
  },
}))

// WebSocket
const ws = courierWSHandler({
  pubsub,
  heartbeatIntervalMs: 25_000,
  allowedOrigins: [process.env.ALLOWED_ORIGIN!],
  onConnect: ({ request }) => {
    const userId = extractUser(request)
    if (!userId) throw new Error("Unauthorized")
    return { userId }
  },
  onDisconnect: ({ meta, code }) => {
    console.log("WS closed", { meta, code })
  },
})
app.get("/ws", upgradeWebSocket((c) => ws(c)))

// Health checks
app.get("/health", (c) => c.json({ status: "ok" }))

const server = serve({
  fetch: app.fetch,
  port: Number(process.env.PORT) || 3001,
  hostname: "0.0.0.0",
  websocket: { server: wss },
})

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

Other runtimes

For Bun, Deno, or Cloudflare Workers, replace @hono/node-server with the runtime-specific WebSocket API. See the Cloudflare page for Workers setup.

NestJS

main.ts

import { NestFactory } from "@nestjs/core"
import { registerSources } from "@noverax/courier"
import { courierWSAttach } from "@noverax/courier-nestjs"
import { AppModule } from "./app.module"
import { sources } from "./api/sources"
import { pubsub, connectAdapters, disconnectAdapters } from "./lib/adapters"

async function bootstrap(): Promise<void> {
  registerSources(sources)
  await connectAdapters()

  const app = await NestFactory.create(AppModule)

  // WebSocket — answers upgrades on `path` only; set the client's proxyPath.ws to match
  const cleanupWS = courierWSAttach(app.getHttpServer(), {
    path: "/api/ws",
    pubsub,
    heartbeatIntervalMs: 25_000,
    allowedOrigins: [process.env.ALLOWED_ORIGIN!],
    onConnect: ({ request }) => {
      const userId = extractUser(request)
      if (!userId) throw new Error("Unauthorized")
      return { userId }
    },
    onDisconnect: ({ meta, code }) => {
      console.log("WS closed", { meta, code })
    },
  })

  await app.listen(Number(process.env.PORT) || 3001, "0.0.0.0")

  process.on("SIGTERM", async () => {
    // AppModule drains its SSE streams in beforeApplicationShutdown
    await cleanupWS()
    await app.close()
    await disconnectAdapters()
  })
}

bootstrap()

app.module.ts

import {
  type BeforeApplicationShutdown,
  Module,
  type NestModule,
  type MiddlewareConsumer,
} from "@nestjs/common"
import { courierMiddleware, courierSSEMiddleware } from "@noverax/courier-nestjs"

const proxy = courierMiddleware()
const sse = courierSSEMiddleware({
  maxSSEConnections: 10_000,
  sseMaxLifetimeMs: 300_000,
  onSSEConnect: ({ connectionId, request }) => {
    return { userId: extractUser(request) }
  },
  onSSEDisconnect: ({ meta, reason, eventCount }) => {
    console.log(`SSE closed: ${meta.userId}, ${reason}, ${eventCount} events`)
  },
})

@Module({})
export class AppModule implements NestModule, BeforeApplicationShutdown {
  configure(consumer: MiddlewareConsumer): void {
    consumer.apply(proxy).forRoutes("/api/request")
    consumer.apply(sse).forRoutes("/api/sse")
  }

  // app.close() waits for open streams, so they are drained before it does
  async beforeApplicationShutdown(): Promise<void> {
    await Promise.all([proxy.shutdownSSE(), sse.shutdownSSE()])
  }
}

Framework Comparison

Feature	Standalone	Next.js	Express	Fastify	Hono	NestJS
HTTP proxy	Built-in	`apiRoute()`	`courierHandler()`	`courierPlugin`	`courierHandler()`	`courierMiddleware()`
SSE	Built-in	`sseRoute()`	`courierSSEHandler()`	`courierSSEHandler()`	`courierSSEHandler()`	`courierSSEMiddleware()`
WebSocket	Built-in	Separate server	`courierWSAttach()`	`courierWSHandler()`	`courierWSHandler()`	`courierWSAttach()`
WS library	Built-in	N/A	`ws` (peer)	`@fastify/websocket`	`ws` (peer)	`ws` (peer)
Processes	1	2 (Next + WS)	1	1	1	1
PubSub wiring	`pubsub` config	On WS server	`courierWSAttach` opts	`courierWSHandler` opts	`courierWSHandler` opts	`courierWSAttach` opts

Sticky Sessions

Multi-replica deployments require sticky sessions for both SSE and WebSocket. The load balancer must route all requests from the same client to the same replica for the duration of a connection.

Configure this at the infrastructure level:

nginx: upstream with ip_hash or cookie-based affinity
Traefik: sticky.cookie in service definition
K8s Ingress: nginx.ingress.kubernetes.io/affinity: cookie
Cloudflare: Enabled by default for WebSocket

See the Docker and Kubernetes pages for specific configuration.