@kyneta/websocket-transport

Websocket transport for @kyneta/exchange — browser, server, and Bun integration. Provides bidirectional real-time sync over Websockets using the @kyneta/wire binary protocol (CBOR codec + framing + fragmentation).

Subpath Exports

| Export | Entry point | Environment | |--------|-------------|-------------| | @kyneta/websocket-transport/browser | ./dist/browser.js | Browser, Bun, Node.js | | @kyneta/websocket-transport/server | ./dist/server.js | Bun, Node.js | | @kyneta/websocket-transport/bun | ./dist/bun.js | Bun only |

Server Setup

Bun (recommended)

Use createBunWebsocketHandlers for zero-boilerplate integration with Bun.serve():

import { Exchange } from "@kyneta/exchange"
import { WebsocketServerAdapter } from "@kyneta/websocket-transport/server"
import { createBunWebsocketHandlers, type BunWebsocketData } from "@kyneta/websocket-transport/bun"

const serverAdapter = new WebsocketServerAdapter()

const exchange = new Exchange({
  identity: { peerId: "server", name: "server", type: "service" },
  adapters: [serverAdapter],
})

Bun.serve<BunWebsocketData>({
  fetch(req, server) {
    server.upgrade(req)
    return new Response("upgrade failed", { status: 400 })
  },
  websocket: createBunWebsocketHandlers(serverAdapter),
})

For more control, use wrapBunWebsocket directly:

import { wrapBunWebsocket, type BunWebsocketData } from "@kyneta/websocket-transport/bun"

Bun.serve<BunWebsocketData>({
  fetch(req, server) {
    server.upgrade(req)
    return new Response("upgrade failed", { status: 400 })
  },
  websocket: {
    open(ws) {
      const socket = wrapBunWebsocket(ws)
      serverAdapter.handleConnection({ socket }).start()
    },
    message(ws, msg) {
      const data = msg instanceof ArrayBuffer ? new Uint8Array(msg) : msg
      ws.data?.handlers?.onMessage?.(data)
    },
    close(ws, code, reason) {
      ws.data?.handlers?.onClose?.(code, reason)
    },
  },
})

Node.js (ws library)

Use wrapNodeWebsocket to adapt the ws library's WebSocket to the framework-agnostic Socket interface:

import { WebSocketServer } from "ws"
import { WebsocketServerAdapter, wrapNodeWebsocket } from "@kyneta/websocket-transport/server"

const serverAdapter = new WebsocketServerAdapter()

const exchange = new Exchange({
  identity: { peerId: "server", name: "server", type: "service" },
  adapters: [serverAdapter],
})

const wss = new WebSocketServer({ port: 3000 })

wss.on("connection", (ws) => {
  const { start } = serverAdapter.handleConnection({ socket: wrapNodeWebsocket(ws) })
  start()
})

Client Setup

Browser

Use createWebsocketClient for browser-to-server connections:

import { Exchange } from "@kyneta/exchange"
import { createWebsocketClient } from "@kyneta/websocket-transport/browser"

const adapter = createWebsocketClient({
  url: "ws://localhost:3000/ws",
  WebSocket,
  reconnect: { enabled: true },
})

const exchange = new Exchange({
  identity: { peerId: "browser-client", name: "Alice", type: "user" },
  adapters: [adapter],
})

Service-to-Service

Use createServiceWebsocketClient for backend connections that need authentication headers during the Websocket upgrade. Headers are a Bun/Node-specific extension — the browser WebSocket API does not support custom headers.

import { createServiceWebsocketClient } from "@kyneta/websocket-transport/server"

const adapter = createServiceWebsocketClient({
  url: "ws://primary-server:3000/ws",
  headers: {
    Authorization: "Bearer <token>",
  },
  reconnect: { enabled: true },
})

const exchange = new Exchange({
  identity: { peerId: "worker-1", name: "worker-1", type: "service" },
  adapters: [adapter],
})

For browser clients, authenticate via URL query parameters instead of headers.

Connection Lifecycle

The connection lifecycle is a Program<WsClientMsg, WebsocketClientState, WsClientEffect> from @kyneta/machine — a pure Mealy machine with data effects. The transport class (WebsocketClientTransport) is a thin imperative shell that interprets data effects as I/O (FC/IS design). Every state × event combination is covered by pure data tests — no sockets, no timing, never flaky.

The WebSocket client has a 5-state lifecycle with an extra ready state compared to SSE and Unix Socket transports. The server sends a text "ready" signal after the TCP connection opens, and only then does the client create a channel and begin the establishment handshake.

disconnected → connecting → connected → ready
                   ↓            ↓         ↓
              reconnecting ← ─ ┴ ─ ─ ─ ─ ┘
                   ↓
              connecting (retry)
                   ↓
              disconnected (max retries)

| State | Description | |-------|-------------| | disconnected | No active connection. Optional reason field describes why. | | connecting | Websocket handshake in progress. Tracks attempt number. | | connected | TCP connection open, waiting for server "ready" signal. | | ready | Server sent "ready" text frame — protocol messages can flow. | | reconnecting | Connection lost, scheduling next attempt. Tracks attempt and nextAttemptMs. |

Program Architecture

The program (createWsClientProgram) encodes all transitions and side effects as inspectable data:

• Messages (WsClientMsg): start, socket-opened, server-ready, socket-closed, socket-error, reconnect-timer-fired, stop
• Effects (WsClientEffect): create-websocket, close-websocket, add-channel-and-establish, remove-channel, start-reconnect-timer, cancel-reconnect-timer, start-keepalive, stop-keepalive

The imperative shell interprets each effect as real I/O — creating WebSocket instances, scheduling timers, sending keepalive pings. Keepalive is modeled as start-keepalive / stop-keepalive effects rather than internal timer logic.

Race Condition Handling

The server may send "ready" before the client's open event fires (server-ready while still connecting). The program handles this by transitioning directly from connecting → ready, skipping the connected state entirely.

Connection Handshake

1. Client opens Websocket, transitions to connecting
2. Websocket open event fires, transitions to connected
3. Server sends text "ready" frame, client transitions to ready
4. Client creates its channel, calls establishChannel()
5. Synchronizer exchanges establish-request / establish-response

Observing State

The public observation API is powered by createObservableProgram from @kyneta/machine:

import { createWebsocketClient } from "@kyneta/websocket-transport/browser"

const adapter = createWebsocketClient({
  url: "ws://localhost:3000/ws",
  lifecycle: {
    onStateChange: ({ from, to }) => console.log(`${from.status} → ${to.status}`),
    onDisconnect: (reason) => console.log("disconnected:", reason.type),
    onReconnecting: (attempt, nextMs) => console.log(`retry #${attempt} in ${nextMs}ms`),
    onReconnected: () => console.log("reconnected"),
    onReady: () => console.log("ready"),
  },
})

// Or subscribe to transitions programmatically
const unsub = adapter.subscribeToTransitions(({ from, to }) => {
  console.log(`${from.status} → ${to.status}`)
})

await adapter.waitForStatus("ready", { timeoutMs: 5000 })

The Socket Interface

Socket is the framework-agnostic abstraction that decouples the adapter from any specific Websocket library. Platform-specific wrappers adapt concrete implementations to this interface:

| Wrapper | Input | Export | |---------|-------|--------| | wrapStandardWebsocket(ws) | Browser WebSocket | ./server | | wrapNodeWebsocket(ws) | Node.js ws library | ./server | | wrapBunWebsocket(ws) | Bun ServerWebSocket | ./bun |

interface Socket {
  send(data: Uint8Array | string): void
  close(code?: number, reason?: string): void
  onMessage(handler: (data: Uint8Array | string) => void): void
  onClose(handler: (code: number, reason: string) => void): void
  onError(handler: (error: Error) => void): void
  readonly readyState: "connecting" | "open" | "closing" | "closed"
}

Configuration

Client Options

| Option | Default | Description | |--------|---------|-------------| | url | — | Websocket URL. String or (peerId) => string function. | | WebSocket | — | WebSocket constructor (required). In browsers, pass WebSocket. In Node.js, pass ws's WebSocket. | | reconnect.enabled | true | Enable automatic reconnection. | | reconnect.maxAttempts | 10 | Maximum reconnection attempts before giving up. | | reconnect.baseDelay | 1000 | Base delay in ms for exponential backoff. | | reconnect.maxDelay | 30000 | Maximum delay cap in ms. | | reconnect.fullJitter | true via createWebsocketClient, else false | AWS-style full jitter (random × min(raw, cap), spread across [0, cap)) instead of the additive 0–20 %. On by default for browser clients so a mass disconnect doesn't stampede the next instance. | | keepaliveInterval | 30000 | Interval in ms for keepalive ping frames. | | fragmentThreshold | 102400 | Payload size threshold for fragmentation (bytes). | | headers | — | Upgrade headers (ServiceWebsocketClientOptions only). |

Server Options

| Option | Default | Description | |--------|---------|-------------| | fragmentThreshold | 102400 | Payload size threshold for fragmentation (bytes). | | drain | — | Default DrainOptions for drainConnections() (see Graceful shutdown). |

Server methods

| Member | Description | |--------|-------------| | drainConnections(options?): Promise<DrainResult> | Gracefully drain for a rolling deploy — stop accepting, close connections staggered across a jitter window, resolve when all close or the deadline elapses. | | isDraining: boolean | Whether a drain is in progress (new connections are being refused). |

DrainOptions: { windowMs?: number (5000), closeCode?: number (1001), closeReason?: string ("Server draining"), deadlineMs?: number (windowMs + 5000), randomFn?: () => number }. DrainResult: { closed: number, remaining: number, timedOut: boolean }.

Fragment Thresholds by Environment

| Environment | Recommended threshold | Reason | |-------------|----------------------|--------| | AWS API Gateway | 100KB (default) | 128KB frame limit | | Cloudflare | 500KB | 1MB frame limit | | Self-hosted | 0 (disabled) | No infrastructure limits |

Keepalive

The client sends text "ping" frames at the configured interval. The server responds with text "pong". This keeps connections alive through proxies and load balancers that terminate idle connections.

Graceful shutdown / rolling deploys

A deploy that hard-kills the process closes every WebSocket in the same instant; all clients then reconnect inside a narrow window and stampede the next instance. The fix is the standard recipe: stop accepting → close staggered → drain → exit, paired with full-jitter reconnect on the client.

Run these steps in order from your own signal handler:

const transport = new WebsocketServerTransport({ drain: { windowMs: 3000 } })
// …wire `transport.handleConnection({ socket })` into your upgrade handler…

async function shutdown() {
  // 1. Stop accepting new connections (HTTP-layer).
  httpServer.removeListener("upgrade", onUpgrade) // Node + ws
  wss.close()
  // 2. Drain: staggered closes across the jitter window, await closure.
  await transport.drainConnections()              // uses the constructor default
  // 3. Flush + stop the exchange (hard-closes any stragglers).
  await exchange.shutdown()
  // 4. Close the HTTP server and exit.
  httpServer.close(() => process.exit(0))
}
process.once("SIGTERM", () => void shutdown())
process.once("SIGINT", () => void shutdown())

Under Bun, step 1 is server.stop() (stop accepting while keeping open sockets) instead of detaching an upgrade listener.

Two steps are irreducibly yours. The transport cannot own them:

• The signal handler. A library must not call process.on("SIGTERM") — it would fight your app, other libraries, and the test runner, with no defined ordering.
• HTTP-layer stop-accepting. The transport is not an HTTP server; only your code owns the upgrade.

The transport backstops correctness, though. Once drainConnections() is called, isDraining is true and any connection that still reaches handleConnection is refused (closed immediately, never registered). So step 1 is an optimization that avoids wasted handshakes during the racy window where a platform (e.g. Kubernetes) removes the pod from its Service endpoints around the same time SIGTERM lands — it is not load-bearing for correctness.

Constraint: keep windowMs/deadlineMs comfortably under your platform's kill grace period (e.g. Kubernetes terminationGracePeriodSeconds, default 30 s), or the process is SIGKILLed mid-drain.

On the client, createWebsocketClient defaults reconnect.fullJitter to true, so the drained clients reconnect spread across the whole backoff window rather than bunched at baseDelay.

Runtime Agnosticism

Every entry point (./browser, ./server, ./bun) is safe to import in any JavaScript runtime — no top-level side effects, no globalThis probes. The transport package has zero runtime dependencies; callers provide all implementations:

| Entry Point | Purpose | Caller Provides | |-------------|---------|----------------| | ./browser | Browser-to-server connections | WebSocket constructor (the global WebSocket in browsers) | | ./server | Server transport + service-to-service client | Socket wrapper via wrapNodeWebsocket() or wrapStandardWebsocket() | | ./bun | Bun-optimized server handlers | ServerWebSocket from Bun's callback API |

This follows the same structural-typing principle used throughout: wrapNodeWebsocket takes a NodeWebsocketLike — it never imports ws. wrapBunWebsocket takes a Bun ServerWebSocket — it never imports Bun APIs at the top level.

Peer Dependencies

{
  "peerDependencies": {
    "@kyneta/exchange": ">=0.0.1",
    "@kyneta/wire": ">=0.0.1"
  }
}

License

MIT