Package reference

Mirrors the package README (single source). Install @basaltkit/realtime v1.5.0 — npm · source.

@basaltkit/realtime

Real-time communication for Basalt: server-to-client push over WebSocket or SSE, with per-tenant channels, presence (who's online), and an event bridge that connects your app's domain hooks directly to connected clients. You need this module when you want live updates — notifications, feeds, dashboards, collaboration — without the client polling.

What this module solves

In a typical application the client keeps asking ("anything new?") repeatedly. With realtime, it's the server that notifies the client the instant something happens. This module gives you that in a way that is:

Framework-neutral — the core (hub, channels, presence) knows nothing about sockets; the transports (WebSocket/SSE) are thin and connect to your adapter (Fastify/Express/Hono).
Multi-tenant — channels are isolated per tenant: to('acme').channel('notes') never delivers to another tenant's clients.
Multi-instance — with the Redis backplane, an emit on one instance reaches clients connected to any instance.
Testable without a server — the core is tested with fake connections; you don't need to open sockets.

Installation

bash

pnpm add @basaltkit/realtime

Depends only on @basaltkit/core. For multi-instance you need a Redis (the client is injected — typically ioredis).

Get started in 5 minutes

1. Register the plugin

import { createApp } from '@basaltkit/core'
import { realtimePlugin, REALTIME } from '@basaltkit/realtime'

const app = await createApp({
  plugins: [realtimePlugin()],
}).boot()

2. Connect a client (transport)

The core talks to connections (Connection). You build one from your adapter's socket/response and register it in the hub. WebSocket example:

import { REALTIME_HUB, websocketConnection } from '@basaltkit/realtime'

// in your adapter's WebSocket upgrade handler, with the user already authenticated:
const hub = app.container.get(REALTIME_HUB)
const conn = websocketConnection({ tenantId: tenant.id, userId: user.id }, socket)
hub.register(conn)
// subscribe() is async and returns false when refused (authorize gate, caps, bad name)
if (!(await hub.subscribe(conn.id, 'notes'))) socket.close()

socket.on('close', () => hub.unregister(conn.id))

Check Origin before you accept the upgrade. Browsers attach cookies to a WebSocket handshake from any site and the same-origin policy does not apply to it, so a cookie-authenticated socket without an Origin allow-list can be opened by a malicious page on the victim's behalf (cross-site WebSocket hijacking). Refuse the upgrade unless request.headers.origin is one of your own origins — or authenticate the socket with a token the page has to send explicitly. If clients send subscribe commands, parse them defensively (try/catch around JSON.parse, check typeof cmd.channel === 'string'): a throw in an async socket listener is an unhandled rejection, which terminates Node by default. The @basaltkit/realtime-client README has a complete handler.

It's the same with SSE, but you provide how to write to the response (framework-neutral):

import { sseConnection } from '@basaltkit/realtime'

reply.raw.writeHead(200, { 'Content-Type': 'text/event-stream', Connection: 'keep-alive' })
const conn = sseConnection(
  { tenantId: tenant.id, userId: user.id },
  { write: (chunk) => reply.raw.write(chunk), end: () => reply.raw.end() },
)
hub.register(conn)
if (!(await hub.subscribe(conn.id, 'notes'))) reply.raw.end()
reply.raw.on('close', () => hub.unregister(conn.id))

3. Emit from the server

const realtime = app.container.get(REALTIME)
await realtime.to('acme').channel('notes').emit('created', { id: 1, title: 'Hello' })
// every client of tenant 'acme' subscribed to 'notes' receives the event

Event bridge (hooks → push)

Instead of calling emit by hand, connect a domain hook to a channel. Whenever the hook fires, the payload is pushed to clients — without touching the code that emits the event:

import { realtimePlugin, bridgeRule } from '@basaltkit/realtime'

realtimePlugin({
  bridge: [
    bridgeRule({
      hook: 'note:created',            // a hook declared in your app
      tenant: (p) => p.tenantId,        // which tenant to deliver to (undefined → skips); omit → tenant in context
      channel: 'notes',                 // or a function (p) => `notes:${p.folderId}`
      event: 'created',
      data: (p) => p.note,              // optional; defaults to sending the whole payload
    }),
  ],
})

bridgeRule validates the types against the hook's payload, so p has the right type.

Schema- or database-per-tenant apps: domain models have no tenantId column, so omit tenant. The bridge then delivers to the tenant in the active context at emit time (ctx().tenant.id, set by @basaltkit/tenancy); hook handlers run inside the emitter's async context. If there is no tenant in context (boot, cron, a worker without context), the event is skipped and reported to onBridgeSkipped. A tenant you set that returns undefined is different: that is an explicit opt-out and is skipped silently, so never point tenant at a field the payload doesn't carry.

BridgeRule fields:

Field	Type	Default	Purpose
`hook`	`keyof BasaltHooks & string`	— (required)	The core hook to listen to.
`tenant`	`(payload) => string \| undefined`	the tenant in context (`ctx().tenant.id`)	Which tenant receives the push. Return `undefined` to skip this event silently; that is the intended way to filter. Omit it when the payload has no tenant id; with no tenant in context either, the event is skipped and reported to `onBridgeSkipped`.
`channel`	`string \| ((payload) => string)`	— (required)	Target channel; a function lets you shard per record ((p) => `notes:${p.folderId}`).
`event`	`string`	— (required)	Event name the client listens for.
`data`	`(payload) => unknown`	the whole hook payload	What to send. Set this when the hook payload contains fields the client shouldn't see.

Authorizing subscriptions

By default the hub accepts any channel name a connection asks for, within its tenant. That is fine when every channel in a tenant is readable by every member of it — and a data leak the moment it isn't (a user's private channel, an admin feed). Pass authorize:

realtimePlugin({
  authorize: (connection, channel) => {
    if (channel.startsWith('user:')) return channel === `user:${connection.userId}`
    if (channel.startsWith('admin:')) return isAdmin(connection.userId)
    return true
  },
})

The gate runs server-side on every hub.subscribe(), before the connection is attached, and may be async. A refusal makes subscribe() resolve false — your adapter decides whether to close the socket or just ignore the request. Never trust a channel name that came from the client without this.

The two caps (maxSubscriptionsPerConnection, maxChannelLength) are enforced in the same place, so a client that loops subscribe cannot grow the hub's maps without bound.

Presence

realtime.to('acme').channel('notes').presence() // → ['user-1', 'user-7']  (unique online ids)
realtime.to('acme').channel('notes').count()    // → number of connections

Presence counts connections on this instance. In multi-instance setups, each node knows only its local clients — for an aggregated global view, sum across instances (or publish join/leave over the backplane; a future enhancement).

Multi-instance with Redis

Pass a RedisBackplane so an emit on one instance reaches clients on all of them. Provide two Redis clients (a client in subscribe mode can't publish):

import Redis from 'ioredis'
import { realtimePlugin, RedisBackplane } from '@basaltkit/realtime'

realtimePlugin({
  backplane: new RedisBackplane({ publisher: new Redis(url), subscriber: new Redis(url) }),
})

emit does a PUBLISH; Redis delivers to all subscribed instances (including the origin one), and each hub delivers to its local connections — the same path serves one node or many.

The pub/sub channel is shared by everything on that Redis that uses the same name, and messages carry only a tenant id — so two apps (or staging and production) that both keep the default 'basalt:realtime' deliver into each other's tenants, and anyone who can PUBLISH on the Redis can push an event to any tenant. Give each deployment its own channel, and set secret to sign every message with HMAC-SHA256 — unsigned, tampered or foreign-key messages are then dropped:

new RedisBackplane({
  publisher,
  subscriber,
  channel: 'myapp:prod:realtime',
  secret: process.env.REALTIME_BACKPLANE_SECRET!, // or [newKey, oldKey] while rotating
})

Every instance must share the key(s): the first one signs, all of them verify. A node without a secret and a node with one can't talk to each other, so roll it out to all instances together (pushes between mixed nodes are dropped and logged meanwhile).

API reference

`realtimePlugin(options?)`

Option	Type	Default	Purpose
`backplane`	`RealtimeBackplane`	`new MemoryBackplane()`	Fan-out across instances. The memory backplane loops straight back and only serves one process.
`bridge`	`BridgeRule[]`	`[]`	Hook → channel rules (build them with `bridgeRule(...)`).
`authorize`	`(connection: Connection, channel: string) => boolean \| Promise<boolean>`	allow all	Server-side subscription gate. Set this whenever a channel carries anything not readable by every member of the tenant — without it, any authenticated connection can subscribe to any channel name inside its tenant. Return `false` to refuse.
`maxSubscriptionsPerConnection`	`number`	`1000`	Cap on distinct channels one connection may hold. A DoS bound: without it a client can subscribe in a loop and grow the hub's maps unbounded.
`maxChannelLength`	`number`	`256`	Cap on a channel name's length. Same reason; empty names are refused too.
`onBridgeError`	`(error: unknown, info: { hook: string; channel: string; event: string }) => void`	`console.error` naming the hook, channel and event	A bridged broadcast failed — see Failure hooks.
`onBridgeSkipped`	`(info: { hook: string; channel: string; event: string; reason: 'no-tenant' }) => void`	`console.warn` naming the hook, channel and event, once per rule	A rule without `tenant` fired with no tenant in context — see Failure hooks.
`onDeliveryError`	`(error: unknown, info: { connectionId: string; tenantId: string; channel: string; event: string }) => void`	`console.error` naming the connection, tenant, channel and event	A single client's `send` threw — see Failure hooks.

Registers the REALTIME (Realtime) and REALTIME_HUB (RealtimeHub) tokens, start()s the hub on boot (subscribing the backplane), wires the bridge rules to the core hook bus, and close()s the hub on shutdown.

authorize, maxSubscriptionsPerConnection, maxChannelLength and onDeliveryError are forwarded to the hub — they are also RealtimeHubOptions, so a hand-built new RealtimeHub(backplane, options) takes exactly the same four.

Failure hooks

Realtime has no error classes. Every failure path is a callback with a non-silent default, because a realtime push is cosmetic fan-out and must never fail the domain write that triggered it — but it must not vanish either.

Hook	Fires when	Default	What happens regardless
`onBridgeError`	A `bridge` rule's broadcast rejected — typically the backplane is down (Redis unreachable).	`console.error` with hook → channel, event	The hook handler is fire-and-forget: the failure is caught, so the domain write that emitted the hook still succeeds. Clients simply miss that push.
`onBridgeSkipped`	A `bridge` rule without `tenant` fired but there is no tenant in context (the hook was emitted at boot, from cron, or from a worker without context).	`console.warn` with hook → channel, event — once per rule, not per event	The push is skipped. A function `channel` is evaluated for the report; if it throws, `channel` is `'<dynamic>'`. A rule whose own `tenant` returns `undefined` never lands here: that opt-out is silent.
`onDeliveryError`	One connection's `send()` threw during local delivery — a dead or closing socket.	`console.error` with the connection, tenant, channel and event	That connection is pruned (`unregister`) and closed — so the client notices and reconnects instead of sitting on a silent socket — and every remaining subscriber still receives the message. One dead socket never stops the fan-out, and never throws into the backplane's message emitter (fatal on a real ioredis subscriber).

Point all three at your logger in production — a permanently failing bridge means clients are silently stale, and a spike in delivery errors means sockets are dying faster than they are being unregistered.

A payload JSON can't encode (a BigInt, a cycle) is not a delivery error: publish() / emit() rejects with the TypeError before anything is sent, on every backplane, and no subscriber is touched. (A bridged emit lands in onBridgeError.)

Separately, the RedisBackplane drops malformed or unparseable messages — and, with secret set, unsigned or wrongly signed ones — arriving on the shared pub/sub channel and logs them with console.error; it never throws into ioredis's 'message' emitter, where an escaped exception would be an uncaughtException.

Errors

Error	Code	When
—	—	This package exports no error classes. Refusals are reported as return values (`hub.subscribe(...)` resolves `false`) and faults as callbacks (`onBridgeError`, `onDeliveryError`).

`class Realtime`

Method	Description
`to(tenantId).channel(name).emit(event, data?)`	Publishes an event on the tenant's channel.
`to(tenantId).channel(name).presence()`	Online user ids (local).
`to(tenantId).channel(name).count()`	Number of connections (local).

`class RealtimeHub`

new RealtimeHub(backplane?, options?: RealtimeHubOptions).

Method	Signature	Description
`start`	`() => Promise<void>`	Subscribes the backplane so cross-instance messages reach local connections. The plugin calls it at boot.
`register`	`(connection: Connection) => void`	Tracks a live connection.
`unregister`	`(connectionId: string) => void`	Drops the connection and all its subscriptions/presence.
`subscribe`	`(connectionId: string, channel: string) => Promise<boolean>`	Attaches a connection to a channel. Returns whether it was accepted — `false` when the connection is unknown (or was unregistered or replaced while `authorize` was pending), the channel is not a string, the name is empty or longer than `maxChannelLength`, `maxSubscriptionsPerConnection` is reached, or `authorize` refused. Idempotent (a repeat returns `true`).
`unsubscribe`	`(connectionId: string, channel: string) => void`	Detaches from one channel.
`publish`	`(tenantId, channel, event, data) => Promise<void>`	Publishes to every subscriber across all instances. Rejects with a `TypeError` when `data` isn't JSON-serializable.
`presence`	`(tenantId, channel) => string[]`	Distinct user ids on this node.
`count`	`(tenantId, channel) => number`	Connection count on this node.
`close`	`() => Promise<void>`	Closes every connection and the backplane.

subscribe() is async and its false is the security signal. Adapters must check it and refuse or close the client — ignoring the result turns a denied authorize into a silent no-op that looks like a working subscription.

`interface Connection`

What a transport hands the hub: id, tenantId, optional userId, send(message), close(). The hub never touches a socket directly, which is what keeps the core framework-neutral and unit-testable with fakes. userId is what makes a connection visible in presence().

Transports

websocketConnection(meta, socket) — from any ws-like socket (send/close).
sseConnection(meta, { write, end }) — SSE; you provide how to write/end.
sseFrame(message) — formats a message as an SSE frame. CR, LF and NUL are stripped from the event name, so an event name can't inject extra fields or frames.

Backplanes

MemoryBackplane — single process (default).
RedisBackplane({ publisher, subscriber, channel?, secret? }) — Redis pub/sub (channel default 'basalt:realtime' — give each deployment its own). secret (a string, or an array to rotate — first signs, all verify) signs every message with HMAC-SHA256 and drops anything unsigned or wrongly signed; an empty secret throws at construction.

How it connects to other modules

@basaltkit/core — provides createApp, tokens, and the hook bus the event bridge consumes.
@basaltkit/events — emits domain events; the bridge turns them into push.
@basaltkit/notifications — common pattern: notification persisted and pushed live by the same event.
@basaltkit/tenancy / @basaltkit/auth — where the tenantId/userId you assign to each connection come from.

@basaltkit/realtime ​

What this module solves ​

Installation ​

Get started in 5 minutes ​

1. Register the plugin ​

2. Connect a client (transport) ​

3. Emit from the server ​

Event bridge (hooks → push) ​

Authorizing subscriptions ​

Presence ​

Multi-instance with Redis ​

API reference ​

realtimePlugin(options?) ​

Failure hooks ​

Errors ​

class Realtime ​

class RealtimeHub ​

interface Connection ​

Transports ​

Backplanes ​

How it connects to other modules ​