WebSocket protocol

Each broker has its own WebSocket. Both speak the same framing, described here; their operations are on Event Broker WebSocket and State Broker WebSocket.

Endpoints#

Use wss:// in production. A browser app usually opens one socket per broker it uses and shares it across every view.

Handshake#

The token goes in the token query parameter (URL-encoded), or in an Authorization: Bearer header where your WebSocket library can set one (browsers can't). It is checked once:

Browsers don't expose the status of a refused handshake: the socket just fires error and close (code 1006). If a connection never opens, check the token with GET /me: it tells you whether the token is valid and which claims its key has.

Once open, the connection's permissions are a snapshot of the key's claim. The connection stays open after the token expires; changes to the claim apply when you reconnect.

Framing#

Every WebSocket message, in both directions, is a JSON text frame holding an array of op-objects:

[{ "op": "sub", "topic": "chat:lobby", "ref": "1" }]

• Send one or more ops per message (at most 100). Replies to one message come back as one array, in the same order.
• Every op you send is answered, with <op>:ok or <op>:error.
• Add a ref (any JSON value) to match replies: it is echoed on the reply, along with the op's topic (and key, for states).
• Deliveries (an Event Broker message, a State Broker change) arrive on their own, one per frame.
• Unknown fields are ignored. Binary frames are refused.

Batching several ops in one message:

[
	{ "op": "sub", "topic": "chat:lobby", "ref": "a" },
	{ "op": "sub", "topic": "chat:support", "ref": "b" }
]

[
	{ "op": "sub:ok", "topic": "chat:lobby", "ref": "a" },
	{ "op": "sub:ok", "topic": "chat:support", "ref": "b" }
]

Errors#

An op that fails is answered with <op>:error, a code and a human-readable message. The connection stays open.

[
	{
		"op": "pub:error",
		"topic": "orders:1042",
		"ref": "7",
		"code": "forbidden",
		"message": "Missing P claim on topic"
	}
]

Problems that can't be tied to one op (invalid JSON, a message that isn't a non-empty array, too many ops) are answered with a single { "op": "error", "code": "bad_request", "message": … }.

Keepalive and limits#

• The server sends a WebSocket ping every 30 seconds. Browsers and Node.js answer pings automatically.
• The server closes a connection that sent no frame for 60 seconds (pongs count).
• A frame larger than 100 × (64 KiB + 1 KiB) closes the connection with code 1009.

Connection lifecycle#

get a token ──▶ open ──▶ sub … ──▶ (connection drops) ──▶ wait, back off
     ▲                                                        │
     └────────────────────────────────────────────────────────┘

A connection can drop at any time: network changes, deploys, laptops going to sleep. Plan for it:

1. Get a new token for each connection attempt; tokens are short-lived.
2. Reconnect with exponential backoff and jitter.
3. Subscribe again to everything you had.
4. Catch up on what you missed: Event Broker history, State Broker tell or listing.

Build reliable clients shows a complete implementation.

Minimal client without a library#

const { token } = await fetch('/api/thuum-token').then((r) => r.json());
const ws = new WebSocket(
	`wss://api.thuum.dev/api/acme/event/connect?token=${encodeURIComponent(token)}`
);

let nextRef = 1;
const pending = new Map();

function send(op) {
	const ref = String(nextRef++);
	return new Promise((resolve, reject) => {
		pending.set(ref, { resolve, reject });
		ws.send(JSON.stringify([{ ...op, ref }]));
	});
}

ws.addEventListener('message', (message) => {
	for (const op of JSON.parse(message.data)) {
		if (op.op === 'message') {
			console.log('event', op.topic, op.type, op.data);
			continue;
		}
		const waiter = pending.get(op.ref);
		if (!waiter) continue;
		pending.delete(op.ref);
		if (op.op.endsWith(':ok')) waiter.resolve(op);
		else waiter.reject(Object.assign(new Error(op.message), { code: op.code }));
	}
});

await new Promise((resolve, reject) => {
	ws.addEventListener('open', resolve, { once: true });
	ws.addEventListener('close', reject, { once: true });
});
await send({ op: 'sub', topic: 'chat:lobby' });
await send({ op: 'pub', topic: 'chat:lobby', type: 'message', data: { text: 'Hi!' } });

import { EventBrokerSocket } from '@thuum/client';

const { token } = await fetch('/api/thuum-token').then((r) => r.json());
const socket = new EventBrokerSocket({ baseUrl: 'https://api.thuum.dev', domain: 'acme', token });

socket.onMessage((event) => console.log('event', event.topic, event.type, event.data));
await socket.ready;
await socket.sub('chat:lobby');
await socket.pub('chat:lobby', 'message', { text: 'Hi!' });