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 { allowRequestForOrigins, 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,
          }),
    )
  }

  wsApp.set('socketio', {
    allowRequest: allowRequestForOrigins(allowedCorsOrigins()),
  })

  wsApp.on('ws:start', io => {
    io.of('/').on('connection', async socket => {
      const user = await resolveWebsocketUser(socket)
      if (!user) {
        socket.disconnect(true)
        return
      }
      await Ws.register(socket, user.id)
    })
  })

  wsApp.on('ws:connect', () => {
    // do something upon websocket connection being established
  })
}
  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.
// 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 })

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. This enables you to establish socket bindings. 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.

export default (wsApp: PsychicAppWebsockets) => {
  // ...
  wsApp.on('ws:start', io => {
    // use socket.io to establish namespaced channels
    // for your app to communicate on
    io.of('/').on('connection', async socket => {
      const user = await resolveWebsocketUser(socket)
      if (!user) {
        socket.disconnect(true)
        return
      }

      // this automatically fires the /ops/connection-success message
      await Ws.register(socket, user.id)

      // establish socket routes using socket.on
    })
  })
}

ws:connect​

ws:connect fires once a socket has connected, and is the recommended place to do per-socket work that might throw — including auth. A hook registered here runs inside a try/catch the framework provides: if it throws (a DB or Redis blip mid-handshake, for example), the framework logs the error, calls socket.disconnect(true), and moves on — only that one socket is rejected, the ws process stays up. That containment means the hook doesn't need its own outer try/catch to protect the process, and a throw is a legitimate way to reject a connection.

wsApp.on('ws:connect', async () => {
  // do something upon websocket connection being established —
  // a throw here is caught, logged, and disconnects only this socket
})

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, not inside ws:start's io.on('connection')

Containment only wraps ws:connect hooks. If you do per-socket auth directly inside the io.on('connection') handler you register in ws:start (as shown above, and as create-psychic scaffolds), a throw there is not contained by the framework and gets no ws:error coverage — it's your own responsibility to catch it. Moving that logic into wsApp.on('ws:connect', …) gets you both containment and ws:error observability for free.

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. A throw-to-reject (an auth hook throwing to refuse a connection) fires this too, so filter your own rejection sentinels inside the hook if you don't want to treat those as errors.
  • 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​

create-psychic scaffolds a boilerplate auth helper at conf/system/resolveWebsocketUser.ts. A newly-generated app rejects every websocket connection until you replace its body with real auth logic — typically pulling a token from socket.handshake.auth, decrypting it, and returning the matching User (or null to reject):

// conf/system/resolveWebsocketUser.ts
import type { Socket } from 'socket.io'
import User from '../../app/models/User.js'

export default async function resolveWebsocketUser(socket: Socket): Promise<User | null> {
  const token = socket.handshake.auth.token as string | undefined
  if (!token) return null
  // ...decrypt the token and look up the user
  return User.find(/* userId */)
}

The "fail loudly in dev" ergonomic is intentional — if auth isn't wired up, no socket connects, rather than every socket connecting as an anonymous user.

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', { allowRequest, ... }) is a thin passthrough to socket.io's own server options — see the socket.io server API docs for the full option shape. The one Psychic-specific piece is allowRequestForOrigins, exported from @rvoh/psychic-websockets, which builds an allowRequest handler from your app's origin allowlist:

import allowedCorsOrigins from '@conf/system/allowedCorsOrigins.js'
import { allowRequestForOrigins } from '@rvoh/psychic-websockets'

wsApp.set('socketio', {
  allowRequest: allowRequestForOrigins(allowedCorsOrigins()),
})

This matters 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.

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.

If you need to layer an auth-token check onto the handshake in addition to origin checking, do it inside a custom allowRequest body rather than allowRequestForOrigins — that helper only checks origin.

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 same node, because engine.io's heartbeat timers are per-connection, 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 },
})