@setu-ts/websocket-plugin

Full-duplex, bidirectional real-time messaging behind the Setu-TS capability model. Registers an IWebSocketService under CAPABILITIES.WEBSOCKET.

Declare routes with lifecycle handlers, address peers individually or through named rooms, and let the runtime’s HTTP adapter perform the RFC 6455 handshake — the same application code runs on Node, Deno, Bun, and Cloudflare Workers.

When to use it

Reach for WebSockets when the client also sends: chat, collaborative editing, multiplayer state, live cursors, interactive dashboards with client-driven subscriptions.

When traffic only flows server → client — notifications, progress feeds, log tailing, metric ticks — prefer @setu-ts/sse-plugin. Server-Sent Events run over ordinary HTTP, survive proxies and corporate middleboxes far more reliably, and reconnect automatically in the browser. A WebSocket buys you an upstream channel at the cost of all of that.

Runtime support

On Node, install ws (npm install ws / deno add npm:ws). It is an optional dependency loaded lazily on the first accepted upgrade, so a plain HTTP application never pulls it in; if it is missing when an upgrade arrives you get a clear error naming the install command. setu add websocket writes that peer into package.json on a Node project, and re-running it repairs a project where it is missing. When RuntimePlugin registers the application, a failed Node handshake is also reported through HttpAdapterOptions.reportError — once per distinct message, never once per upgrade — so a missing ws is visible in the log instead of only as a refused socket.

Because Bun’s server.upgrade needs the Server instance and its handlers must be supplied at Bun.serve() time, and Node’s upgrades never touch fetch at all, WebSockets on those two runtimes require a real listening server (app.start({ port })). app.fetch() and app.inject() serve ordinary HTTP only.

Installation

import { WebSocketPlugin } from '@setu-ts/websocket-plugin';

Usage

In an application (the form a scaffolded project uses)

Declare routes where the plugin is composed, through the routes option. A CLI-scaffolded project’s generated setu.config.ts forbids starting the server, so there is no post-start() moment to resolve the capability from — and this arm needs none. A route whose handlers need a resolved capability declares a RegistryFactory entry instead, resolved in onInit.

import { createApplication } from '@setu-ts/kernel';
import { RuntimePlugin } from '@setu-ts/runtime';
import { WebSocketPlugin } from '@setu-ts/websocket-plugin';

// setu.config.ts
export default createApplication({
  plugins: [
    RuntimePlugin(),
    WebSocketPlugin({
      heartbeatMs: 30_000,
      idleTimeoutMs: 90_000,
      routes: [
        {
          path: '/ws/chat',
          handlers: {
            onOpen: (conn, { query }) => {
              conn.data.set('room', query.room ?? 'lobby');
            },
            onMessage: (conn, data) => {
              const room = conn.data.get('room');
              if (typeof room === 'string') {
                // Resolve the service from a factory entry when you need it here.
                conn.send(data);
              }
            },
          },
          // Refuse the upgrade before the handshake, for this route only.
          options: { guards: [({ headers }) => headers.has('authorization') || { status: 401 }] },
        },
      ],
    }),
  ],
});

Through a plugin, when the route needs the live service

An IPlugin with dependencies: [CAPABILITIES.WEBSOCKET] whose register() receives the live service also works, and is what setu generate plugin <name> emits. Prefer it when a route body needs to call ws.room(...) and you would rather hold the service than resolve it per entry.

import type { IPlugin, IPluginContext, IWebSocketService } from '@setu-ts/common';
import { CAPABILITIES } from '@setu-ts/common';
import { createApplication } from '@setu-ts/kernel';
import { RuntimePlugin } from '@setu-ts/runtime';
import { WebSocketPlugin } from '@setu-ts/websocket-plugin';

export class ChatPlugin implements IPlugin {
  readonly name = 'chat';
  readonly version = '1.0.0';
  readonly dependencies = [CAPABILITIES.WEBSOCKET];

  register(ctx: IPluginContext): void {
    const ws = ctx.services.get<IWebSocketService>(CAPABILITIES.WEBSOCKET);

    ws.route('/ws/chat', {
      onOpen: (conn, { query }) => {
        // Authenticate here; close(1008) to refuse a peer after the handshake.
        const room = query.room ?? 'lobby';
        conn.data.set('room', room);
        ws.room(room).add(conn);
      },
      onMessage: (conn, data) => {
        const room = conn.data.get('room');
        if (typeof room === 'string') {
          ws.room(room).broadcast(data, { except: conn });
        }
      },
      onClose: () => {
        // Rooms evict the connection automatically — nothing to clean up here.
      },
      onError: (conn, error) => {
        ctx.logger?.error('socket failed', { id: conn.id, error });
      },
    });
  }
}

// setu.config.ts
export default createApplication({
  plugins: [
    RuntimePlugin(),
    WebSocketPlugin({ heartbeatMs: 30_000, idleTimeoutMs: 90_000 }),
    new ChatPlugin(),
  ],
});

Standalone script

In a plain script you own end to end, resolving the service after app.start() works too:

import { createApplication } from '@setu-ts/kernel';
import { RuntimePlugin } from '@setu-ts/runtime';
import { WebSocketPlugin } from '@setu-ts/websocket-plugin';
import { CAPABILITIES, type IWebSocketService } from '@setu-ts/common';

const app = createApplication({
  plugins: [RuntimePlugin(), WebSocketPlugin({ heartbeatMs: 30_000, idleTimeoutMs: 90_000 })],
});
await app.start({ port: 3000 });

const ws = app.services.get<IWebSocketService>(CAPABILITIES.WEBSOCKET);

ws.route('/ws/chat', {
  onOpen: (conn, { query, headers }) => {
    // Authenticate here; close(1008) to refuse a peer after the handshake.
    const room = query.room ?? 'lobby';
    conn.data.set('room', room);
    ws.room(room).add(conn);
  },
  onMessage: (conn, data) => {
    const room = conn.data.get('room');
    if (typeof room === 'string') {
      ws.room(room).broadcast(data, { except: conn });
    }
  },
  onClose: (conn, { code, reason }) => {
    // Rooms evict the connection automatically — nothing to clean up here.
  },
  onError: (conn, error) => {
    // A standalone script has no IPluginContext — use your own logger here.
    console.error(`socket ${conn.id} failed`, error);
  },
});

Options

WebSocketRouteOptions.guards is route-scoped: guards run in declared order before the matched route’s handshake, and the first { status } refusal wins. It is separate from the plugin-level frame behaviors arm; no route-level behaviour arm exists.

Configuring one or more behaviours makes frame dispatch promise-mediated, so a synchronous onMessage runs after a microtask. With no behaviours configured, dispatch remains the direct, synchronous invoke.

Opting a route out of the sweep

A route that speaks its own liveness protocol passes heartbeat: false:

ws.route('/graphql/ws', handlers, { protocols: ['graphql-transport-ws'], heartbeat: false });

That excludes the route’s connections from both halves of the sweep — the payload send and the idle eviction — and defaults to true, so no existing route is affected. It exists because the sweeper is global across routes and heartbeatPayload is a raw text frame, not a protocol message: a graphql-transport-ws peer that receives one must close the socket with 4400, and a listen-only subscriber is inbound-silent by design so idleTimeoutMs would evict it.

idleTimeoutMs requires heartbeatMs > 0, since the heartbeat tick performs the sweep. Configuring one without the other throws at construction rather than silently doing nothing.

Routing

Routes match on exact path. Variable data belongs in the query string and reaches onOpen via WebSocketConnectionContext.query:

ws.route('/ws/chat', handlers); // matches /ws/chat and /ws/chat?room=general
// does NOT match /ws/chat/general

Pattern parameters (:id) are deliberately unsupported: the kernel’s matcher is internal to @setu-ts/kernel, and hand-rolling a second one would duplicate logic that must not drift. Registering the same path twice throws.

Subprotocols are opt-in per route and echoed from an allow-list — never blindly reflected back:

ws.route('/ws', handlers, { protocols: ['chat', 'json'] });
// client asks for "json, chat" → "json" is echoed
// client asks for "binary"     → handshake refused with 400

Semantics

• The middleware pipeline runs before the handshake. Since M70a an upgrade goes through the same chain as a GET /users. A handshake does need the native Request, which the framework’s request mapping pre-reads — so the kernel keeps an undisturbed copy on IRequest.raw and consults IWebSocketService.routeUpgrade from its terminal handler, after the pipeline has run without short-circuiting and before route matching, so an application catch-all such as the SSR one ReactRouterPlugin mounts cannot shadow the upgrade. A guard that answers 401 therefore refuses it, metrics apply, and a draining application answers 503. setUpgradeRouter is still the adapter seam, but the adapter no longer consults the router: it only needs to know one was installed, which is what makes Node attach its raw upgrade listener. Non-WebSocket traffic is untouched, and a request on an unregistered path falls straight through to your normal routes.

• An accepted upgrade carries no response headers a middleware wrote. The adapter answers with the runtime’s own 101, so security headers and Set-Cookie set on ctx.response are not carried onto the handshake response. The pipeline still runs, which is what lets a guard refuse; it just has no response left to decorate once the socket is taken over. A refused upgrade is an ordinary HTTP response and carries everything.

• Authenticate inside onOpen — a cookie session is verifiable there. Browsers cannot set headers on a WebSocket constructor, so a bearer header never arrives — but a cookie does, and the framework now reads it. When @setu-ts/auth-plugin runs with a session arm (see packages/auth-plugin/README.md), the pipeline authenticates the upgrade before the handshake decides it, so onOpen’s context.user carries the principal the session mapped — context.user.id is the peer, and the member is omitted entirely when the upgrade was anonymous. To read the session payload itself, open it from the upgrade’s headers: await sessions.fromHeaders(context.headers) (where sessions is the ISessionService resolved from CAPABILITIES.SESSION) returns the read-only SessionView — { id, data } — or null. conn.close(1008) refuses a peer whose credential fails, and a guard in the authentication band refuses the upgrade earlier, before the socket opens. The SessionView is a snapshot taken at handshake time and never refreshes: a session destroyed mid-connection leaves the socket open, so 1008 is the application-side revocation path. An application not running SessionPlugin can still verify a credential it can check directly: a signed token in the query string (keep it short-lived; it lands in URLs and access logs) or a subprotocol carrying a nonce your HTTP layer issued.

• A cookie-authenticated socket is same-site by default — and sameSite: 'none' removes that. The session cookie defaults to sameSite: 'lax', and a WebSocket handshake is not a top-level navigation, so a Lax cookie is not sent on a cross-site upgrade: a cross-site page cannot open a cookie-authenticated socket. Setting cookie.sameSite: 'none' on SessionPlugin removes that protection, and the same-origin policy does not cover WebSocket — a cross-site page can then open the socket and read everything it is sent. If you must run none, check the Origin header in the authentication band (a guard that refuses foreign origins refuses the upgrade before the handshake) rather than trusting the cookie alone.

• The heartbeat is an application-level frame, not an RFC 6455 ping. Deno and Cloudflare Workers expose no ping() on their web WebSocket, so a protocol ping would silently no-op on half the supported runtimes. Your client should treat heartbeatPayload as a keep-alive and may reply to keep the idle timer fresh — the idle sweep looks only at inbound traffic.

• room(name) is get-or-create; peek(name) is the read that is not. Calling room('board:acme').size for a name nobody has joined CREATES that room, so a presence or dashboard endpoint reporting size for caller-supplied names accumulates one empty room per distinct name polled — with nothing to reclaim it until some other socket disconnects (the sweep runs on disconnection, so an idle application never reclaims). Use peek wherever the name comes from a request: it returns the room if one exists and undefined otherwise, and registers nothing. At the registry level, RoomRegistry.get(name) also creates a room on lookup; use RoomRegistry.peek(name) when reading a client-supplied name.

  app.router.get('/presence/:board', (ctx) => {
    const board = ctx.params.board ?? '';
    return ctx.response.json({ present: ws.peek(`board:${board}`)?.size ?? 0 });
  });

• Rooms are in-process by default. broadcast skips closed members and drops them, and a room that empties is discarded, so a churning connection set does not grow memory. Behind more than one replica, register a backplane — see Scaling beyond one replica.

• Handler errors never escape. A throw or a rejected promise from any callback is routed to onError rather than becoming an unhandled rejection.

• Backpressure is not managed for you. send hands the frame to the platform socket. A slow consumer with a fast producer will buffer in the runtime; throttle at the application level.

• Shutdown. onClose closes every live connection with 1001 (going away) and stops the heartbeat timer.

Scaling beyond one replica

A room is process-local until you register a backplane. On a single instance that is invisible. Behind two or more, room('lobby').broadcast(...) reaches only the clients connected to that process — the other replicas’ clients hear nothing, and no error is raised anywhere. It is partial delivery, not a failure, which is what makes it easy to ship.

Registering @setu-ts/realtime-backplane-plugin with a cross-process transport is the entire fix. This plugin resolves it optionally, so nothing else changes:

import { RealtimeBackplanePlugin } from '@setu-ts/realtime-backplane-plugin';

createApplication({
  plugins: [
    RuntimePlugin(),
    RealtimeBackplanePlugin({ transport: 'redis', url: 'redis://localhost:6379' }),
    WebSocketPlugin(),
  ],
});

The transport matters, not just the plugin. RealtimeBackplanePlugin() defaults to transport: 'memory' — a real bus, but a single-process one. Registering it bare silences the startup notice described below without fanning anything out. Use 'redis', or 'messaging' to reuse whichever broker MessagingPlugin already registered.

Rooms now fan out across every replica sharing that transport, including RoomBroadcastOptions.except — connection ids are UUIDs, so the exclusion is honored cluster-wide. Remove the plugin and behavior returns to in-process, with no application change.

Two things stay local by design: room.size counts this replica’s members only (a cluster-wide count is inherently asynchronous and cannot satisfy the synchronous getter), and per-connection send is naturally local since the socket lives on one process.

When no backplane is registered this plugin logs one info line at startup stating the limitation. If you are running a single replica, that line is informational and safe to ignore — and if you have decided single-replica fan-out is correct for this deployment, scalingNotice: false silences it:

WebSocketPlugin({ scalingNotice: false });

That suppresses the message only. Room delivery is identical either way.

Errors

WebSocketUnavailableError is exported for instanceof handling. It is thrown by route() when the registered IHttpAdapter implements no setUpgradeRouter seam — which happens only with a custom third-party adapter predating the seam. In that state the service still registers and the health indicator reports available: false, so one codebase deploys everywhere; the failure surfaces at registration rather than at a peer’s first connect.

Diagnostics (M98l)

WebSocketPlugin({ diagnostics: { enabled: true, alias: 'chat' } }) counts, for the local diagnostics connector’s GET /v1/realtime (@setu-ts/diagnostics-plugin), how many connections opened (a handshake that fails after the upgrade was accepted is a failed open) and closed (a close after a transport error, or with code 1006, is failed), and every frame send wrote or refused. Sends are counted once, at the connection, so sendJson, a room broadcast and a heartbeat each count one per recipient. On each read of a paired devtool the source also reports the service’s current connectionCount and roomCount, which never expire the way operation records do. No frame, close code or reason, connection id, room name, header, query or principal is ever captured.

Without the option the plugin registers an inert disabled source and nothing is attached, so its hot paths read no clock. Observation never changes a result: a refused send still throws the same error, and a failing clock or gauge reader latches the source collection-failed instead. A replacement registered later under the same capability is outside coverage and is never read. Enable only on an approved development dataset: counts aggregate every tenant using the instance.

Health

Registers a websocket health indicator reporting { available, connections, rooms, routes }.

Exports

Generated from the package barrel by pnpm run docs:exports; pnpm run check:docs fails when it drifts.

Full API

Every export and option is documented in PUBLIC_API.md.

API reference

WebSocketPlugin() (@setu-ts/websocket-plugin)

Provides full-duplex, bidirectional real-time messaging, completing the real-time story that SsePlugin covers one-way. Registers an IWebSocketService under CAPABILITIES.WEBSOCKET. Added in Milestone 46.

The RFC 6455 handshake is performed by the runtime’s HTTP adapter through the optional IHttpAdapter.setUpgradeRouter seam, so the same application code runs on Node, Deno, Bun, and Cloudflare Workers. The plugin never creates a server and never touches a runtime API.

Since M70a the middleware pipeline runs before the handshake. The adapter stores the router; the kernel terminal handler consults IWebSocketService.routeUpgrade after the pipeline has run without short-circuiting — and before route matching, so an application catch-all such as the SSR one ReactRouterPlugin mounts cannot shadow the upgrade. A guard that answers 401 therefore refuses the upgrade, metrics apply, and a draining application answers 503 — none of which happened before, which was the security defect M70a closed. On an accepted upgrade the adapter answers with the runtime’s own 101, which does not carry response headers a middleware set on ctx.response (security headers, Set-Cookie); a refused upgrade is an ordinary HTTP response and carries them all. Upgrade detection (the RFC 6455 header check) lives in the service’s own router, inside its error-reporting wrapper, because WsRouteTable matches on path alone and a routing failure has no other place to be logged. A non-conformant upgrade carrying a body is refused 400 by the kernel before any handshake is attempted, so the behaviour is one thing on all four adapters instead of a runtime-specific failure inside the upgrade call.

Since M73 the pipeline’s authenticated principal rides the upgrade: the kernel passes ctx.request.user to IWebSocketService.routeUpgrade, and onOpen’s WebSocketConnectionContext carries it as the optional user member — populated when a strategy authenticated the upgrade, omitted when it did not. An unauthenticated upgrade is anonymous, and a guard in the authentication band refuses it before the handshake. See the session arm of AuthPlugin for the cookie-backed strategy that produces it.

Registration

import { WebSocketPlugin } from '@setu-ts/websocket-plugin';

app.register(
  WebSocketPlugin({
    maxConnections: 10_000,
    heartbeatMs: 30_000,
    heartbeatPayload: 'ping',
    idleTimeoutMs: 90_000,
    maxMessageBytes: 1_048_576,
  }),
);

Options

WebSocketRouteOptions.guards is route-scoped: matching-route guards run before the handshake in declared order, and the first non-true decision refuses. It is distinct from plugin-level frame behaviors; no route-level behaviour arm exists.

Usage

import { CAPABILITIES, type IWebSocketService } from '@setu-ts/common';

const ws = app.services.get<IWebSocketService>(CAPABILITIES.WEBSOCKET);

ws.route(
  '/ws/chat',
  {
    onOpen: (conn, { query, headers }) => {
      conn.data.set('room', query.room ?? 'lobby');
      ws.room(query.room ?? 'lobby').add(conn);
    },
    onMessage: (conn, data) => {
      ws.room(conn.data.get('room') as string).broadcast(data, { except: conn });
    },
    onClose: (conn, { code, reason }) => logger.info('closed', { code, reason }),
    onError: (conn, error) => logger.error('socket error', { error }),
  },
  { protocols: ['chat'] },
);

Route options

Set heartbeat: false when the route speaks its own liveness protocol. The sweeper sends heartbeatPayload as a raw text frame to every connection on every route, which a protocol client (a graphql-transport-ws peer, for instance) must treat as an invalid message; and it evicts on inbound silence, which is the normal state of a listen-only subscriber. The GraphQL plugin’s WebSocket route claims this opt-out.

Exports

Notes

• Routes match on exact path. Variable data travels in the query string and reaches onOpen via WebSocketConnectionContext.query. Pattern parameters (:id) are deliberately not supported: the kernel’s matcher is internal to @setu-ts/kernel and hand-rolling a second one would duplicate logic.
• The heartbeat is an application-level frame, not an RFC 6455 ping. The web WebSocket API on Deno and Cloudflare Workers exposes no ping(), so a protocol ping would silently no-op on half the supported runtimes.
• Node requires ws. It is an optional dependency, loaded lazily on the first accepted upgrade (AI_GUIDELINES §12.2) — a plain HTTP application never loads it. loadWsModule throws with the install command when it is absent. The other three runtimes need no dependency.
• Upgrades on Node arrive on the raw upgrade event, not the fetch path, so they are only available after listen(). On Bun, upgrades likewise only work through listen(), since server.upgrade requires the serve-time websocket handlers. Deno and Workers upgrade on the fetch path.
• A custom adapter without setUpgradeRouter degrades gracefully: the service still registers, the health indicator reports available: false, and route() throws WebSocketUnavailableError.
• room(name) is get-or-create; peek(name) is the read that is not. room(callerSupplied).size creates a room per distinct name polled, and the registry reclaims never-joined rooms only on the next disconnection somewhere in this process — an idle application never does. IWebSocketService.peek(name): WebSocketRoom | undefined returns the room when one exists and undefined otherwise, registering nothing, and is what a presence or dashboard endpoint reading a request-supplied name should call. Added in M74; a required member, so a replacement implementation of IWebSocketService must supply it. At the registry level, RoomRegistry.get(name) creates on lookup too; use RoomRegistry.peek(name) for reads of client-supplied room names.
• Rooms are in-process until a backplane is registered. Register RealtimeBackplanePlugin and every broadcast also reaches members on other replicas; with no CAPABILITIES.REALTIME_BACKPLANE provider the behavior is unchanged. RoomBroadcastOptions.except is honored on every replica: connection IDs come from runtime.uuid() and are globally unique, so the frame carries the excluded ID. Room.size keeps reporting local membership either way. Room.broadcastLocal is the local-only delivery path the backplane subscriber uses (its LocalBroadcastOptions adds exceptId); applications call broadcast.
• A RoomRegistry keeps a reverse connection → rooms index, so evicting a disconnecting peer costs only the rooms that peer had actually joined rather than a scan of every live room. The index is maintained through the RoomMembershipListener the registry gives each Room it creates; a standalone new Room(name) takes no listener and is not tracked.
• A failing upgrade router is logged, then refused with 500. The service catches its own routing errors and reports them through the logger capability when one is registered — the HTTP adapter’s UpgradeRouterStore backstop runs inside @setu-ts/runtime, which has no logger, so the cause would otherwise be lost. Register the LoggerPlugin to see it.
• app.inject() cannot exercise a WebSocket; tests must bind a real socket (app.start({ port }) + new WebSocket(...)).
• A websocket health indicator reports { available, connections, rooms, routes }. onClose closes every live connection with code 1001 and stops the heartbeat.