Skip to main content

Config

Installation​

There are two ways to install the psychic-websockets package. The first is by selecting yes when prompted during the initial psychic app provisioning stage. If you select yes, the package will be automatically installed, and your app bootstrapped to use websockets automatically.

However, if this is not the case for you and you are looking to install websockets after the fact, you can follow these steps:

  1. Install the package.
pnpm add @rvoh/psychic-websockets
  1. Add the missing configuration file to src/conf/initializers/websockets.ts:
import AppEnv from '#conf/AppEnv.js'
import allowedCorsOrigins from '#conf/system/allowedCorsOrigins.js'
import resolveWebsocketUser from '#conf/system/resolveWebsocketUser.js'
import { PsychicApp } from '@rvoh/psychic'
import { PsychicAppWebsockets, Ws } from '@rvoh/psychic-websockets'
import { Redis } from 'ioredis'

export default (psy: PsychicApp) => {
  psy.plugin(async () => {
    await PsychicAppWebsockets.init(psy, initializeWebsockets)
  })
}

function initializeWebsockets(wsApp: PsychicAppWebsockets) {
  // The websockets transport adapter is selected per environment:
  //   - test:        in-process adapter (the default) — no Redis needed.
  //   - development
  //   - production:  Redis adapter (the default) — distributes the socket
  //                  registry and broadcasts across a clustered websocket fleet.
  if (!AppEnv.isTest) {
    wsApp.set(
      'connection',
      AppEnv.isProduction
        ? new Redis({
            host: AppEnv.string('WS_REDIS_HOST'),
            port: AppEnv.integer('WS_REDIS_PORT', { optional: true }) || 6379,
            username: AppEnv.string('WS_REDIS_USERNAME'),
            password: AppEnv.string('WS_REDIS_PASSWORD'),
            tls: {},
            maxRetriesPerRequest: 3,
            commandTimeout: 10000,
          })
        : new Redis({
            host: AppEnv.string('WS_REDIS_HOST', { optional: true }) || 'localhost',
            port: AppEnv.integer('WS_REDIS_PORT', { optional: true }) || 6379,
            username: AppEnv.string('WS_REDIS_USERNAME', { optional: true }),
            password: AppEnv.string('WS_REDIS_PASSWORD', { optional: true }),
            // tls:  {},
            maxRetriesPerRequest: 3,
            commandTimeout: 10000,
          }),
    )
  }

  // socket.io's own `cors.origin` constrains HTTP long-polling only; native
  // WebSocket upgrades bypass CORS. `allowRequest` runs before every handshake
  // on every transport, so re-enforce the allowlist there.
  const allowedOrigins = allowedCorsOrigins()
  wsApp.set('socketio', {
    cors: {
      credentials: true,
      origin: allowedOrigins,
    },
    allowRequest: (req, callback) => {
      const origin = req.headers.origin
      if (origin !== undefined && allowedOrigins.includes(origin)) {
        callback(null, true)
      } else {
        callback('origin not allowed', false)
      }
    },
  })

  // Sign in each socket in ws:connect, where a throw is contained to that
  // socket and reported to ws:error. resolveWebsocketUser is the app's helper.
  wsApp.on('ws:connect', async socket => {
    const user = await resolveWebsocketUser(socket)
    if (!user) {
      socket.disconnect(true)
      return
    }

    await Ws.register(socket, user.id)

    const ws = new Ws(['/ops/connection-success'] as const)
    await ws.emit(user.id, '/ops/connection-success', { message: 'Connected' })
  })
}
  1. No changes to initializePsychicApp.ts are needed. The initializer file registers itself as a plugin via psy.plugin(...), so PsychicApp.init() picks it up automatically.
Initializes in all processes by default

PsychicAppWebsockets.init() runs in all processes by default. Any process — websocket server, web server, or worker — may call Ws.emit(), and skipping init in any of them causes a runtime cachePsychicAppWebsockets error that is easy to misdiagnose. Each Node process has its own module cache and does not share the websocket app instance with other processes.

To restrict which roles can push messages, add a guard in conf/initializers/websockets.ts:

if (!['websockets', 'web', 'worker'].includes(AppEnv.serviceRole) && !AppEnv.isTest) return
  1. We recommend you include a singleton within your application to simplify your websockets integration:
import { Ws } from '@rvoh/psychic-websockets'

export const WS_ROUTES = ['/ops/connection-success'] as const

const ws = new Ws(WS_ROUTES)

export default ws

Configuration​

The configuration for the psychic-websockets package is driven by the conf/initializers/websockets.ts file. This file contains both basic bootstrapping information for redis, as well as hooks to tap into to initialize socket.io and establish websocket listeners for your backend application.

Adapter​

psychic-websockets selects a transport adapter per environment automatically:

  • Test — InProcessWebsocketsAdapter (the default). No Redis connection is opened. Unit specs do zero Redis I/O; broadcasts are recorded in-process so your test helpers can assert on them.
  • Development / Production — RedisWebsocketsAdapter (the default). Distributes the socket registry and fan-out broadcasts across a clustered websocket fleet.

To override the automatic selection, call wsApp.set('adapter', ...) anywhere in initializeWebsockets:

// Force the Redis adapter even in test (unusual):
wsApp.set('adapter', 'redis')

// Force the in-process adapter in all environments:
wsApp.set('adapter', 'in_process')

// Supply a fully custom adapter instance:
wsApp.set('adapter', new MyCustomAdapter())

Redis​

Redis — used to distribute broadcasts across a websocket fleet — is configured with wsApp.set('connection', ...). Wrap it in if (!AppEnv.isTest) so your test suite needs no Redis:

if (!AppEnv.isTest) {
  wsApp.set(
    'connection',
    AppEnv.isProduction
      ? new Redis({
          host: AppEnv.string('WS_REDIS_HOST'),
          port: AppEnv.integer('WS_REDIS_PORT', { optional: true }) || 6379,
          username: AppEnv.string('WS_REDIS_USERNAME'),
          password: AppEnv.string('WS_REDIS_PASSWORD'),
          tls: {},
          maxRetriesPerRequest: 3,
          commandTimeout: 10000,
        })
      : new Redis({
          host: AppEnv.string('WS_REDIS_HOST', { optional: true }) || 'localhost',
          port: AppEnv.integer('WS_REDIS_PORT', { optional: true }) || 6379,
          username: AppEnv.string('WS_REDIS_USERNAME', { optional: true }),
          password: AppEnv.string('WS_REDIS_PASSWORD', { optional: true }),
          // tls: {},
          maxRetriesPerRequest: 3,
          commandTimeout: 10000,
        }),
  )
}
Bound the connection — do not set maxRetriesPerRequest: null

Unlike a BullMQ worker connection (which must use maxRetriesPerRequest: null for its blocking BLPOP/BRPOPLPUSH commands), the socket.io redis-adapter issues no blocking commands. Setting null here makes a broadcast or socket-registry lookup hang indefinitely when Redis is unreachable — including a Ws.emit() from a worker, which stalls that job. Bound maxRetriesPerRequest and set a commandTimeout so the connection fails fast instead. The adapter needs no special connection options — see the socket.io Redis adapter docs; for the contrasting worker requirement see BullMQ connections.

Connection limits​

Two connection-limit options are available via wsApp.set(...). Both already default to the values shown below — set them only if you need to override:

// Cap on simultaneous socket registrations per user. When a user
// registers a new socket past this limit, their oldest socket is evicted
// from the registry. This bounds per-user resource use — a client
// reconnecting in a loop can't accumulate unbounded registry entries.
wsApp.set('maxConnectionsPerUser', 3)

// TTL on the per-user socket-id registry key in Redis.
// This is a garbage-collection backstop for ungraceful disconnects —
// it is NOT the live socket's lifetime (socket.io's ping settings govern that).
// Keep it comfortably above your longest expected connection. If the TTL
// expires while a socket is still connected, emits to that user will
// silently stop until the socket reconnects and re-registers.
// Accepts: { seconds?, minutes?, hours?, days? }
wsApp.set('maxConnectionTtl', { days: 1 })

Both limits apply under the Redis adapter only, so specs on the in-process adapter never hit the cap. An evicted socket is not disconnected: it stays connected but silently receives nothing more, because emits find sockets through the registry. A user with a fourth tab or device open drops the first one this way, so raise maxConnectionsPerUser if your users routinely hold more connections than that.

Hooks​

In addition to configuration, psychic-websockets also exposes hooks to tap into during various lifecycle events exposed by the websockets app.

ws:start​

The ws:start event is called whenever the websocket server is started, with the socket.io server. Under the hood we are using socket.io to power websocket bindings, which means you can visit their documentation to understand more about setting up a websockets app within your application.

wsApp.on('ws:start', io => {
  // use socket.io to establish namespaced channels
  // for your app to communicate on
})

Sign sockets in from a ws:connect hook rather than from a connection handler registered here; see the caution below.

ws:connect​

ws:connect fires once a socket has connected, with that socket, and is the place to sign it in and do any other per-socket work that might throw. A hook registered here runs inside a try/catch the framework provides: if it throws (a DB or Redis blip during the user lookup, for example), the framework logs the error, calls socket.disconnect(true), and moves on — only that one socket is rejected, and the ws process stays up. So the hook needs no outer try/catch to protect the process.

wsApp.on('ws:connect', async socket => {
  const user = await resolveWebsocketUser(socket)
  if (!user) {
    socket.disconnect(true)
    return
  }

  await Ws.register(socket, user.id)

  const ws = new Ws(['/ops/connection-success'] as const)
  await ws.emit(user.id, '/ops/connection-success', { message: 'Connected' })

  // any other per-connection work goes after Ws.register,
  // so it never runs for a socket that is not signed in
})

The /ops/connection-success event is the hook's own emit, sent once Ws.register has stored the socket; it is what tells the client that sign-in finished (see Client transport).

Framework-contained failures here are also forwarded to the ws:error hook below, which is where you ship them to your monitoring SDK.

Per-socket auth belongs in ws:connect

Containment only wraps ws:connect hooks. A throw inside an io.on('connection') handler registered in a ws:start hook, or inside io.use middleware, is not contained by the framework and gets no ws:error coverage. In the generated ws.ts it becomes an unhandled rejection, which exits the websocket process. Put per-socket auth in wsApp.on('ws:connect', …) to get both containment and ws:error observability.

ws:error​

ws:error is the ws-layer analogue of server:error — the hook to forward a framework-contained websocket failure to your monitoring SDK (Sentry, Datadog, etc.). Register it with the same positional signature, (error, context):

wsApp.on('ws:error', (error, context) => {
  if (context.phase === 'ws:connect') {
    // a ws:connect hook threw; context.socketId identifies the socket
    reportToSentry(error, { socketId: context.socketId })
  } else {
    // context.phase === 'ws:health-check'
    // the ws server's own HTTP handler threw; context.method, context.path
    reportToSentry(error, { method: context.method, path: context.path })
  }
})

The second argument is a discriminated context (PsychicWebsocketsErrorContext, exported alongside PsychicWebsocketsErrorHook) with two phases:

  • phase: 'ws:connect' — a ws:connect hook threw while a socket was connecting. Carries socketId.
  • phase: 'ws:health-check' — the websocket server's own HTTP request handler (the health check and its catch-all 404) threw. Carries method and path. This is the ws process's raw http.Server, not the Koa app — errors here never reach server:error.

The context handed to this hook is privacy-scrubbed for external shipping: it never carries a raw socket, handshake credentials, request headers, or the raw request URL — path is the pathname with the query string stripped, since a query string can carry tokens or PII. (The framework's own internal error log still records the full URL locally; only the external-bound hook context is scrubbed.)

Redis adapter connection errors have no ws:error phase

This is by design. The framework already logs the pub/sub clients' 'error' events at error level, so they don't go dark — but to ship or dedupe them yourself, attach your own .on('error', …) to the public wsApp.connection and wsApp.subConnection getters right after wsApp.set('connection', redis) (subConnection, a duplicate() of the pub client, exists only after that call). Dedupe before shipping — ioredis re-emits 'error' on every reconnect attempt during an outage. Configure the connection once, before cable.start(...) (see WebSocket server startup below) — replacing it afterward silently breaks cross-process delivery.

Auth: resolveWebsocketUser​

resolveWebsocketUser (conf/system/resolveWebsocketUser.ts) is the app's own helper. It reads the token the client sends in socket.handshake.auth and returns the matching User, or null to reject the socket. Have it return null for a token it can't decrypt, as getCookie does for a cookie no key opens (see Decrypting cookies outside a controller).

create-psychic scaffolds a version for early development: in the test environment it decrypts the token with Encrypt.decrypt and APP_ENCRYPTION_KEY to find the user, returning null for a token it can't open, and everywhere else it throws, so every socket is rejected (and the throw reported to ws:error) until you replace it with your production authentication.

Health check​

Expose a health check endpoint on the ws process's own HTTP server with wsApp.set('healthCheck', ...):

wsApp.set('healthCheck', {
  path: '/healthcheck',
  method: 'GET',
  body: null,
})

An error thrown by this handler (or by the server's catch-all 404) surfaces to ws:error with context.phase === 'ws:health-check' — see ws:error above.

Origin allowlist​

wsApp.set('socketio', { ... }) is a thin passthrough to socket.io's own server options — see the socket.io server API docs for the full option shape. The generated initializer uses two of them to enforce the app's origin allowlist (shown in Installation above): cors.origin, and an inline allowRequest that closes over the same list.

const allowedOrigins = allowedCorsOrigins()
wsApp.set('socketio', {
  cors: { credentials: true, origin: allowedOrigins },
  allowRequest: (req, callback) => {
    const origin = req.headers.origin
    if (origin !== undefined && allowedOrigins.includes(origin)) {
      callback(null, true)
    } else {
      callback('origin not allowed', false)
    }
  },
})

Both are needed because socket.io's cors.origin option only constrains HTTP long-polling — a native WebSocket upgrade bypasses CORS entirely. allowRequest runs before every handshake on every transport, so it's the one enforcement point that actually covers both. Keep allowRequest to checks on the raw HTTP request: the client's auth token (socket.handshake.auth) isn't available there yet, and a throw there is not contained. Authentication goes in the ws:connect hook.

The allowlist itself lives in the app: conf/system/allowedCorsOrigins.ts parses the CORS_HOSTS env var, which must be a JSON-encoded array of origins:

CORS_HOSTS='["https://app.example.com","https://admin.example.com"]'

It defaults to [] when unset, so an app booted without it rejects every handshake. Malformed JSON throws at boot, naming CORS_HOSTS and echoing what it received.

WebSocket Server Startup​

The websocket server runs as its own process, separate from your web and worker processes, using the Cable class:

// ws.ts
import { Cable } from '@rvoh/psychic-websockets'
import AppEnv from '#conf/AppEnv.js'
import initializePsychicApp from '#conf/system/initializePsychicApp.js'

let cable: Cable | null = null

async function startWs() {
  process.env.WS_SERVICE = '1'
  await initializePsychicApp()

  cable = new Cable()
  await cable.start(AppEnv.integer('WS_PORT', { optional: true }) || (AppEnv.isTest ? 8889 : 8888))
}

process.on('SIGINT', async () => {
  if (cable) await cable.stop()
  process.exit(0)
})

startWs()

Configure the Redis connection (wsApp.set('connection', ...)) before cable.start(...) runs — replacing it afterward silently breaks cross-process delivery. A Redis adapter that can't attach at startup aborts ws startup rather than silently falling back to in-memory delivery, so a misconfigured WS_REDIS_* env var fails loudly instead of degrading broadcasts.

conf/initializers/websockets.ts is auto-loaded by psy.load('initializers', …) in conf/app.ts, and its default export registers the plugin itself (see Installation above). Nothing additional needs wiring in initializePsychicApp.

Client Transport​

Set transports: ['websocket'] on the Socket.IO client. Socket.IO tries long-polling first and upgrades in the background; WebSocket is universally supported by modern browsers and mobile apps, so skipping the polling phase just means faster connection establishment. The websocket server serves polling and WebSocket clients correctly either way, so this is a connection-latency recommendation rather than a correctness requirement — with one exception at scale: a polling client's later requests have to reach the node holding its engine.io session (any other node answers "Session ID unknown"), so a multi-node fleet serving polling clients needs sticky sessions. Websocket-only transport sidesteps that.

import { io } from 'socket.io-client'

const socket = io(websocketHost, {
  transports: ['websocket'],
  auth: { token },
})

The client's connect event can fire before the ws:connect hook has signed the socket in, so wait for the event the hook emits after Ws.register (/ops/connection-success in ws:connect) before relying on emits reaching it. A socket the hook disconnects, such as one resolveWebsocketUser rejects, gets a disconnect event with reason 'io server disconnect', never connect_error:

socket.on('/ops/connection-success', () => {
  // signed in: emits to this user now reach this socket
})

socket.on('disconnect', reason => {
  if (reason === 'io server disconnect') {
    // the server rejected or dropped this socket; it will not reconnect on its own
  }
})