Real-time protocols — websockets, SSE

WebSocket subscriptions, Server-Sent Events, and event-driven communication in ObjectStack

The Real-Time Protocol describes how live data synchronization is intended to work between clients and servers. Get instant updates when data changes without polling.

Implementation status (v1): The shipping realtime service is an in-memory pub/sub adapter (@objectstack/service-realtime, InMemoryRealtimeAdapter) plus a long-polling client (RealtimeAPI in @objectstack/client). The IRealtimeService contract reserves an optional handleUpgrade() for a WebSocket handshake, but no WebSocket (/ws) or SSE (/api/v1/stream) transport is wired up yet — those sections below document the planned wire protocol, not a deployed endpoint. The in-memory adapter is single-instance only (v1 deployment contract); a Redis-backed adapter for multi-node HA is a post-GA fast-follow. Treat the WebSocket/SSE message formats, connection limits, and debug endpoints in this page as a forward-looking design spec until that transport lands.

Identity admission (framework#2992, ADR-0096 D4): today's delivery path is a trusted server-internal fan-out with no per-recipient authorization — subscriptions carry no principal and events carry the full record body. Before any client transport ships, delivery must re-check each subscriber's authority (RLS/FLS/tenant) per event — the subscribe-time permission check shown below is not sufficient — or switch to id-only payloads with client re-fetch. CI backs this with a classification ratchet rather than a proof of it: the realtime-delivery-authz row is experimental — it records the posture above rather than an enforcement site — and the transport tripwires watch a curated list of realtime source files, so wiring a WebSocket/SSE/subscribe transport in one of them turns it into an unclassified surface and reddens the build until the row is upgraded with its enforcement site. That puts the admission requirement in front of a reviewer; it does not check that the re-check was written — upgrading the row is a hand edit, and the build goes green on that edit alone. A transport wired outside the watched files produces no key and no failure. One file is on record as deliberately outside that set: packages/runtime/src/dispatcher-plugin.ts both mounts routes and already writes text/event-stream, but its two SSE sites are per-request AI response streaming — each drains one iterable the route handler returned into that same request's response body — and not subscription fan-out, so watching them would mint a transport key for a surface this admission requirement is not about. That exclusion is drawn on fan-out, not on the SSE content type: wiring an upgrade handler, a subscribe registration or a realtime-service call into that file puts it back inside the hazard, and promoting it into the watched set with such a fan-out-specific marker is a precondition on the WebSocket/SSE transport work rather than something the current tripwires would catch.

Why Real-Time Matters

Problem: Traditional REST APIs require polling to detect changes:

// ❌ Polling approach - wasteful and laggy
setInterval(async () => {
  const response = await fetch('/api/data/task');
  const tasks = await response.json();
  updateUI(tasks);
}, 5000);  // Check every 5 seconds

Costs of polling:

Latency: Updates delayed by polling interval (5 seconds = 5-second lag)
Bandwidth waste: 99% of requests return "no changes"
Server load: 1,000 users polling every 5s = 12M requests/hour
Database strain: Constant query execution even when nothing changed
Battery drain: Mobile devices make unnecessary network calls

Solution: Real-time subscriptions push updates instantly when data changes. Clients subscribe once, server pushes updates. Zero polling, zero lag.

Business Value Delivered

Instant User Experience

Changes appear immediately across all connected clients. No refresh required.

95% Less Server Load

Replace millions of polling requests with persistent connections. Massive infrastructure savings.

Enable Collaboration

Multiple users editing the same data see each other's changes in real-time. Google Docs-style UX.

Live Notifications

Deliver alerts, messages, and updates the moment they happen. No missed events.

Communication Channels

ObjectStack supports two real-time protocols:

WebSocket (Bi-Directional)

Best for: Interactive applications requiring two-way communication

Characteristics:

Full-duplex communication (client and server both send)
Low latency (no HTTP overhead per message)
Persistent connection
Binary and text messages supported

Use cases:

Chat applications
Collaborative editing (Google Docs, Figma)
Multiplayer games
Live dashboards with user interactions
IoT device communication

Server-Sent Events (SSE) (Server-to-Client)

Best for: One-way server-to-client streaming

Characteristics:

Unidirectional (server pushes only)
Works over HTTP (easier firewall traversal)
Automatic reconnection built-in
Text-based (JSON messages)

Use cases:

Activity feeds
Notification streams
Live status updates
Progress indicators
Read-only dashboards

WebSocket Connection

Connection Endpoint

The discovery endpoint reports the realtime service honestly (ADR-0076 D12): because the in-process realtime service mounts no HTTP/WS surface today, no routes.realtime entry is advertised — an advertised route with no handler would 404 — and services.realtime reports enabled: false:

GET /.well-known/objectstack

{
  "routes": {
    "data": "/api/v1/data",
    "metadata": "/api/v1/meta"
  },
  "services": {
    "realtime": {
      "enabled": false,
      "status": "degraded",
      "handlerReady": false,
      "message": "In-process event bus only — no HTTP/WS realtime surface is mounted"
    }
  },
  "capabilities": {
    "websockets": { "enabled": false }
  }
}

What counts as a subscribable channel

A subscribable channel exists only where discovery reports handlerReady: true together with a connectable route; enabled never means "there is a channel".

That is the whole definition, and it is computed in exactly one place — isSubscribableChannel() in @objectstack/spec/api. Both discovery producers set services.realtime.enabled and capabilities.websockets to the value of that call, so the fields a client reads and the predicate a client is told to use are the same computation and cannot disagree.

realtime is the one slot whose advertised capability is such a channel, so it is also the one slot whose enabled answers that question. Everywhere else enabled keeps its narrower meaning — cache, queue and job deliver their whole contract in-process, so they are honestly enabled with no route at all and there is nothing to subscribe to there either.

The route comes from the occupant (IRealtimeService.getChannelRoute()), because no producer in the open framework mounts a realtime transport and neither discovery builder can honestly invent one. @objectstack/service-realtime is an in-process pub/sub bus and names none, which is why every host the open framework ships answers false.

⛔ Do not key a subscription off enabled alone. Until this was written down, /discovery reported enabled: true for realtime and "no HTTP/WS realtime surface is mounted" in the same entry — both true, because enabled meant "the slot is filled". A client that read it as "a channel exists" subscribed to nothing and silently lost the feature it was subscribing for: no error, no failed request, no signal at all. Poll, or degrade, unless handlerReady is true and a route is present.

A WebSocket upgrade endpoint is part of the planned transport (IRealtimeService.handleUpgrade()) and is not yet served — realtime stays out of the open framework (maintainer ruling, 2026-09-04). If a host ever mounts one, its realtime service names the mounted path via getChannelRoute() and discovery advertises routes.realtime, handlerReady: true and capabilities.websockets in the same step (see #2462, #14646).

Declared message vocabulary

WebSocketMessageType (packages/spec/src/api/websocket.zod.ts) is a closed enum, and it is the only declaration this protocol has. It names exactly ten message types:

subscribe · unsubscribe · event · ping · pong · ack · error · presence · cursor · edit

BaseWebSocketMessage types every message's type field as that enum, and each message schema pins type to one of those literals, so a message whose type is outside the enum parses against no schema in the file: WebSocketMessageSchema is a discriminated union over exactly those ten, and an unknown discriminant matches no branch at all.

Every message also carries the three BaseWebSocketMessage fields — messageId (UUID), type, and timestamp (ISO 8601).

⛔ Five message types this page used to teach are not in that enum: an in-band handshake (auth, auth_success, auth_error) and two acknowledgements (subscribed, unsubscribed). Nothing in the runtime produces or consumes any of the five, and a client built on them sends and waits for messages the declared contract cannot carry. They were documentation, not protocol. The sections below describe what the enum does carry: see Authentication for where a credential goes instead, and ack for the acknowledgement the protocol actually declares.

Establishing Connection

Client-side (JavaScript):

const ws = new WebSocket('wss://api.acme.com/ws');

ws.onopen = () => {
  console.log('Connected to ObjectStack');
  // No in-band auth message: the credential travels on the upgrade request
  // (see Authentication). A socket that reaches `onopen` was already admitted.
};

ws.onmessage = (event) => {
  const message = JSON.parse(event.data);
  console.log('Received:', message);
};

ws.onerror = (error) => {
  console.error('WebSocket error:', error);
};

ws.onclose = (event) => {
  console.log('Connection closed:', event.code, event.reason);
};

Authentication

The declared protocol has no authentication message. A credential travels on the HTTP upgrade request, before the socket exists — WebSocketConfig.headers ("Custom headers for WebSocket handshake") is the declared seam for it, and IRealtimeService.handleUpgrade(request) receives that Request with the headers attached. So there is nothing to send once the socket is open, and no success or failure message to wait for: an upgrade that is refused never becomes a WebSocket, and the client sees the HTTP rejection or an immediate close rather than a message.

{
  "url": "wss://api.acme.com/ws",
  "headers": {
    "Authorization": "Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..."
  }
}

Browsers cannot set headers on new WebSocket(...), so a browser client carries the credential the way the host's upgrade route accepts it — a cookie already scoped to the origin, or a short-lived ticket in the URL. Which of those a host accepts is that host's decision; handleUpgrade receives the whole Request either way.

Failures that arise after admission are ordinary type: "error" messages (ErrorMessageSchema, flat code / message / details) — see Token Expiration for the one that matters most here.

⛔ Nothing serves any of this yet. handleUpgrade is optional and unimplemented throughout the open framework (#2462, #3197), and per IRealtimeService the delivery path carries no per-recipient authorization at all — subscriptions carry no principal. Admission on the upgrade request is where the declared shapes put a credential; it is not a claim that a served endpoint checks one today.

Subscriptions

Subscribe to changes on a specific object:

Request:

{
  "messageId": "550e8400-e29b-41d4-a716-446655440000",
  "type": "subscribe",
  "timestamp": "2024-01-16T14:30:00Z",
  "subscription": {
    "subscriptionId": "9f8c1e42-5b7a-4d33-9e10-6c2f8a91b4d7",
    "events": ["data.record.created", "data.record.updated", "data.record.deleted"],
    "objects": ["task"]
  }
}

SubscribeMessageSchema declares exactly two things of its own — type: "subscribe" and a subscription object — on top of the three BaseWebSocketMessage fields every message carries. Nothing else sits at the top level: the subscription's own fields live inside subscription (EventSubscriptionSchema), and they are camelCase.

Parameters:

messageId: UUID for this message (BaseWebSocketMessage)
timestamp: ISO 8601 datetime the message was sent (BaseWebSocketMessage)
subscription.subscriptionId: UUID identifying the subscription — the schema declares z.string().uuid(), so an opaque handle like "sub_1" does not parse
subscription.events: array of event patterns, not bare verbs — lowercase dot notation with wildcards ("data.record.*", "*.created"). The platform-checked vocabulary is the closed DataEventType / BulkDataEventType enums (packages/spec/src/api/events.zod.ts), whose members are data.record.created / data.record.updated / data.record.deleted and their bulk siblings
subscription.objects: optional array of object names to filter events by (["account", "contact"]) — plural, and an array even for a single object
subscription.channels: optional array of channel names for scoped subscriptions
subscription.filters: optional, and declared unknown — see the callout below

⛔ filters is declared but not enforced — it is not "the same syntax as HTTP API filters". EventSubscriptionSchema.filters is z.unknown(), and no runtime evaluates a payload filter: matchesSubscription (@objectstack/service-realtime) matches on object name and event type only. A subscription carrying filters therefore receives every event its patterns match — the extra conditions are silently ignored, in the permissive direction. The key is typed unknown deliberately, because validating a shape nothing reads would imply an enforcement that does not exist (objectui#2945). Narrow with objects and events, and re-check on the client.

Success Response:

The declared acknowledgement is an ack message (AckMessageSchema); there is no subscribed type. ackMessageId echoes the messageId of the message being acknowledged, success says whether it was accepted, and the optional error string carries the reason when it was not.

{
  "messageId": "7c9e6679-7425-40de-944b-e07fc1f90ae7",
  "type": "ack",
  "timestamp": "2024-01-16T14:30:00Z",
  "ackMessageId": "550e8400-e29b-41d4-a716-446655440000",
  "success": true
}

Error Response:

Every type: "error" message on this page is one ErrorMessageSchema envelope (packages/spec/src/api/websocket.zod.ts): messageId, type and timestamp from BaseWebSocketMessage, then a top-level code and message, plus an optional details bag for anything else. There is no nested error object — a client that reads msg.error.code gets undefined.

{
  "messageId": "550e8400-e29b-41d4-a716-446655440000",
  "type": "error",
  "timestamp": "2024-01-16T14:30:00Z",
  "code": "FORBIDDEN",
  "message": "No permission to subscribe to 'task' object",
  "details": { "subscription_id": "sub_1" }
}

Event Types

Event	Triggered When	Payload
`created`	New record created	Full record data
`updated`	Record modified	Changed fields + record ID
`deleted`	Record deleted	Record ID only
`restored`	Soft-deleted record restored	Full record data

Receiving Events

When subscribed data changes, the server pushes event messages.

EventMessageSchema declares five fields of its own — subscriptionId, eventName, object, payload and userId — on top of the three BaseWebSocketMessage fields every message carries. They are camelCase, and the record body travels inside payload, never at the top level.

Created Event:

{
  "messageId": "a41d6c58-92b7-4f0e-8d3a-16b5c7e04912",
  "type": "event",
  "timestamp": "2024-01-16T14:30:00Z",
  "subscriptionId": "9f8c1e42-5b7a-4d33-9e10-6c2f8a91b4d7",
  "eventName": "data.record.created",
  "object": "task",
  "userId": "user_123",
  "payload": {
    "id": "0c8b7a36-4e21-4d59-b7f2-3a9e5d1c8046",
    "type": "data.record.created",
    "object": "task",
    "recordId": "task_456",
    "after": {
      "id": "task_456",
      "title": "New task assigned to you",
      "status": "todo",
      "assignee_id": "user_123",
      "created_at": "2024-01-16T14:30:00Z"
    },
    "userId": "user_123",
    "timestamp": "2024-01-16T14:30:00Z"
  }
}

Updated Event:

{
  "messageId": "b7e2f409-51ac-4d68-9b30-2c7a8e15d643",
  "type": "event",
  "timestamp": "2024-01-16T15:00:00Z",
  "subscriptionId": "9f8c1e42-5b7a-4d33-9e10-6c2f8a91b4d7",
  "eventName": "data.record.updated",
  "object": "task",
  "userId": "user_123",
  "payload": {
    "id": "5d3c9f71-8a04-4b62-91e7-0f6b2d8a3c15",
    "type": "data.record.updated",
    "object": "task",
    "recordId": "task_456",
    "changes": {
      "status": "in_progress"
    },
    "after": {
      "id": "task_456",
      "status": "in_progress",
      "updated_at": "2024-01-16T15:00:00Z"
    },
    "userId": "user_123",
    "timestamp": "2024-01-16T15:00:00Z"
  }
}

Deleted Event:

{
  "messageId": "c9a04b17-6d38-4e52-8f71-4b2e6c9d0537",
  "type": "event",
  "timestamp": "2024-01-16T16:00:00Z",
  "subscriptionId": "9f8c1e42-5b7a-4d33-9e10-6c2f8a91b4d7",
  "eventName": "data.record.deleted",
  "object": "task",
  "userId": "user_123",
  "payload": {
    "id": "e18f2b05-7c46-4a93-b2d0-9e35c7a14f68",
    "type": "data.record.deleted",
    "object": "task",
    "recordId": "task_456",
    "userId": "user_123",
    "timestamp": "2024-01-16T16:00:00Z"
  }
}

⛔ subscription_id, event and data are not fields of this envelope, and changes and reason are not top-level fields either. The declared names are subscriptionId, eventName and payload, so a client written from the old fences reads msg.subscription_id and msg.data and gets undefined on both. changes is real but sits one level down, on the DataEvent inside payload, and it carries the write's own patch — a flat { field: newValue } map (publishDataEvent, packages/objectql/src/engine.ts), not the old / new pairs this page used to show; read the post-state from payload.after. reason is declared nowhere in the protocol at all — see Subscribe to Query Results below.

Fields:

messageId: UUID for this message (BaseWebSocketMessage)
timestamp: ISO 8601 datetime the message was sent (BaseWebSocketMessage)
subscriptionId: UUID of the subscription this event belongs to — the schema declares z.string().uuid(), and it is the same UUID the subscribe message declared, so an opaque handle like "sub_1" does not parse
eventName: the event name, dot notation by convention. The platform-checked vocabulary is the closed DataEventType / BulkDataEventType enums (packages/spec/src/api/events.zod.ts) — data.record.created / data.record.updated / data.record.deleted, plus the two bulk siblings
object: optional object name the event relates to — the one key the old fences had right
payload: the event body, declared z.unknown() on the envelope. The platform's only publisher today is the ObjectQL engine, which emits a DataEvent for a single-row write and a BulkDataEvent for a predicate write; that is the shape shown above
userId: optional — the user who triggered the event

Unsubscribe

Stop receiving events for a subscription:

Request:

{
  "messageId": "b3d1a77e-4c62-4f0b-9a55-1d8e2f6c0b34",
  "type": "unsubscribe",
  "timestamp": "2024-01-16T14:35:00Z",
  "request": {
    "subscriptionId": "9f8c1e42-5b7a-4d33-9e10-6c2f8a91b4d7"
  }
}

UnsubscribeMessageSchema nests its payload under request (UnsubscribeRequestSchema), whose one field is subscriptionId — a UUID, and the same UUID the subscribe message declared. Note the key is not the subscription of a subscribe message: the two envelopes use different wrappers.

Response: the same ack envelope as a subscribe acknowledgement — there is no unsubscribed type either.

{
  "messageId": "1f2b7d64-9a01-4c8e-9f3a-2b5c8d417e60",
  "type": "ack",
  "timestamp": "2024-01-16T14:35:00Z",
  "ackMessageId": "c0ffee00-1111-4222-8333-444455556666",
  "success": true
}

Watch a single record for changes — narrowed on the client, because the declared subscription has no record-level field:

Request:

{
  "messageId": "c1f4b9a2-6e78-4a31-bb05-9d3e7c2a1f80",
  "type": "subscribe",
  "timestamp": "2024-01-16T14:40:00Z",
  "subscription": {
    "subscriptionId": "2a6d5f14-8b90-4c27-a3e1-7f05b8d29c46",
    "events": ["data.record.updated", "data.record.deleted"],
    "objects": ["task"]
  }
}

⛔ There is no recordId on a subscription, and no record_id anywhere in the protocol. EventSubscriptionSchema declares exactly subscriptionId, events, objects, filters and channels — the narrowest scope the contract can express is object + event pattern. This page used to show a top-level record_id, which no schema declares and no runtime reads. Subscribe at object scope as above and compare the incoming event's own record identity on the client; filters cannot do it for you (see the callout under Subscribe to Object Events).

Use case: Detail pages that need to reflect live changes to the currently viewed record.

Subscribe to a dynamic set of records matching a query — again narrowed on the client, because the declared subscription carries no query:

Request:

{
  "messageId": "d8e0c3b5-1f47-4d92-8a6c-5b4e9f210a73",
  "type": "subscribe",
  "timestamp": "2024-01-16T14:45:00Z",
  "subscription": {
    "subscriptionId": "6b3c9e07-2d54-4f18-9c8a-0e7d1b5a4632",
    "events": ["data.record.created", "data.record.updated", "data.record.deleted"],
    "objects": ["task"]
  }
}

⛔ There is no query on a subscription either, and no server-side query membership tracking. A top-level query object appeared on this page but is declared nowhere: EventSubscriptionSchema has no such field, and the only payload-shaped key it does have — filters — is unknown and unenforced. Nothing computes whether a record entered or left a result set, so a server cannot send you an "entered/exited" signal it does not track.

Behavior at object scope — what the declared subscription above actually delivers:

Receive data.record.created events for every task the subscription's patterns match
Receive data.record.updated events for every task that changes
Receive data.record.deleted events for every task that is deleted
Evaluate query membership on the client: apply the predicate to each event and derive the enter/exit transitions yourself by comparing against the set you are already holding

Example: a task the subscription covers is updated so that it now matches your predicate:

{
  "messageId": "d2c74f38-9b15-4e06-a7f3-8c1d5b902e47",
  "type": "event",
  "timestamp": "2024-01-16T14:46:00Z",
  "subscriptionId": "6b3c9e07-2d54-4f18-9c8a-0e7d1b5a4632",
  "eventName": "data.record.updated",
  "object": "task",
  "payload": {
    "id": "7a1e4c92-0d63-4f85-b91a-2e6c8d357b04",
    "type": "data.record.updated",
    "object": "task",
    "recordId": "task_789",
    "changes": { "status": "todo" },
    "after": { "id": "task_789", "status": "todo", "assignee_id": "user_123" },
    "timestamp": "2024-01-16T14:46:00Z"
  }
}

Example: the same subscription, a task updated so that it no longer matches:

{
  "messageId": "e6f18a03-4d27-4c91-8b52-7a0e3d9c1465",
  "type": "event",
  "timestamp": "2024-01-16T14:47:00Z",
  "subscriptionId": "6b3c9e07-2d54-4f18-9c8a-0e7d1b5a4632",
  "eventName": "data.record.updated",
  "object": "task",
  "payload": {
    "id": "9c05b7d4-1a68-4e30-97f2-5d8a1c60e3b9",
    "type": "data.record.updated",
    "object": "task",
    "recordId": "task_456",
    "changes": { "status": "done" },
    "after": { "id": "task_456", "status": "done" },
    "timestamp": "2024-01-16T14:47:00Z"
  }
}

⭐ The two messages are identical in shape — nothing in the envelope says "entered" or "exited". EventMessageSchema has no reason field, and no runtime tracks query membership to populate one, so the earlier reason: "entered_query" / "exited_query" examples described a signal the contract cannot send. The transition is yours to derive: apply your predicate to payload.after and compare against the set you are already holding.

Server-Sent Events (SSE)

Connection Endpoint

GET request with Accept header:

GET /api/v1/stream
Authorization: Bearer <token>
Accept: text/event-stream

Response:

HTTP/1.1 200 OK
Content-Type: text/event-stream
Cache-Control: no-cache
Connection: keep-alive

Subscription via Query Parameters

Subscribe using URL parameters:

GET /api/v1/stream?object=task&events=created,updated&filter={"assignee_id":"user_123"}
Authorization: Bearer <token>
Accept: text/event-stream

Event Format

SSE sends events as text:

id: event_123
event: task.created
data: {"id":"task_456","title":"New task","status":"todo"}

id: event_124
event: task.updated
data: {"id":"task_456","status":"in_progress"}

Client-side (JavaScript):

const eventSource = new EventSource(
  '/api/v1/stream?object=task&events=created,updated',
  {
    headers: {
      'Authorization': 'Bearer ' + token
    }
  }
);

eventSource.addEventListener('task.created', (event) => {
  const task = JSON.parse(event.data);
  console.log('Task created:', task);
});

eventSource.addEventListener('task.updated', (event) => {
  const task = JSON.parse(event.data);
  console.log('Task updated:', task);
});

eventSource.onerror = (error) => {
  console.error('SSE error:', error);
  eventSource.close();
};

Automatic Reconnection

SSE clients automatically reconnect when connection drops:

id: event_200
event: task.created
data: {...}

# Connection lost

# Client reconnects with Last-Event-ID header
GET /api/v1/stream?object=task
Last-Event-ID: event_200

# Server resumes from event_201
id: event_201
event: task.updated
data: {...}

Heartbeat & Keep-Alive

Server Heartbeat

Server sends periodic ping messages to detect disconnections:

WebSocket ping (every 30 seconds):

{
  "type": "ping",
  "timestamp": "2024-01-16T14:30:00Z"
}

Client should respond with pong:

{
  "type": "pong",
  "timestamp": "2024-01-16T14:30:00Z"
}

SSE heartbeat (comment):

: heartbeat

Client Reconnection

Handle disconnections gracefully:

class ObjectStackClient {
  // The credential is not held here: it travels on the upgrade request (see
  // Authentication), so reconnecting re-presents it the same way the first
  // connection did.
  constructor(url) {
    this.url = url;
    this.subscriptions = new Map();
    this.reconnectDelay = 1000;  // Start with 1 second
    this.maxReconnectDelay = 30000;  // Max 30 seconds
  }
  
  connect() {
    this.ws = new WebSocket(this.url);
    
    this.ws.onopen = () => {
      console.log('Connected');
      this.reconnectDelay = 1000;  // Reset backoff
      this.resubscribe();
    };
    
    this.ws.onclose = () => {
      console.log('Disconnected, reconnecting in', this.reconnectDelay);
      setTimeout(() => this.connect(), this.reconnectDelay);
      this.reconnectDelay = Math.min(
        this.reconnectDelay * 2,
        this.maxReconnectDelay
      );
    };
  }
  
  resubscribe() {
    this.subscriptions.forEach((subscription, id) => {
      this.send({ type: 'subscribe', subscription: { ...subscription, subscriptionId: id } });
    });
  }

  subscribe(subscription) {
    // `subscriptionId` is declared `z.string().uuid()` — not an opaque handle.
    const id = crypto.randomUUID();
    this.subscriptions.set(id, subscription);
    this.send({ type: 'subscribe', subscription: { ...subscription, subscriptionId: id } });
    return id;
  }

  send(message) {
    // Every message carries the three `BaseWebSocketMessage` fields.
    this.ws.send(JSON.stringify({
      messageId: crypto.randomUUID(),
      timestamp: new Date().toISOString(),
      ...message,
    }));
  }
}

// Usage
const client = new ObjectStackClient('wss://api.acme.com/ws');
client.connect();

Permission Enforcement

Real-time subscriptions respect object-level and row-level permissions:

Scenario: User subscribes to all tasks:

{
  "messageId": "e5a7d210-3c86-4b4f-9017-8f2c6d3b5a19",
  "type": "subscribe",
  "timestamp": "2024-01-16T14:50:00Z",
  "subscription": {
    "subscriptionId": "9f8c1e42-5b7a-4d33-9e10-6c2f8a91b4d7",
    "events": ["data.record.created", "data.record.updated"],
    "objects": ["task"]
  }
}

Permission enforcement:

Server applies row-level security filter automatically
User only receives events for tasks they have permission to see
If a task becomes visible (e.g., gets assigned to user), they receive an updated event
If a task becomes hidden (e.g., assigned to someone else), no more events sent

Example: User has rule "see only assigned tasks":

{
  "messageId": "f3b90d67-5a24-4e18-9c07-1b8f2a6d4e35",
  "type": "event",
  "timestamp": "2024-01-16T14:51:00Z",
  "subscriptionId": "9f8c1e42-5b7a-4d33-9e10-6c2f8a91b4d7",
  "eventName": "data.record.updated",
  "object": "task",
  "userId": "user_123",
  "payload": {
    "id": "2d7c1f85-6b39-4a04-8e13-9f05c2b71a68",
    "type": "data.record.updated",
    "object": "task",
    "recordId": "task_456",
    "changes": { "assignee_id": "user_123" },
    "after": { "id": "task_456", "assignee_id": "user_123" },
    "userId": "user_123",
    "timestamp": "2024-01-16T14:51:00Z"
  }
}

Reassign task_456 to someone else and, under the row-level behaviour described above, no further message arrives for it: the protocol has no "you lost access" signal to send, so absence of events is the whole signal. There is no reason field here either. What delivery actually enforces today is the implementation-status callout at the top of this page, not this section.

Scaling Considerations

Connection Limits

Each connection consumes server resources.

The per-user connection/subscription limits below (concurrent connections, subscriptions per connection, subscriptions per user) are not implemented. The only cap enforced today is a process-wide safety backstop on total active subscriptions in InMemoryRealtimeAdapter (DEFAULT_MAX_SUBSCRIPTIONS = 50_000, configurable via maxSubscriptions; 0 opts out). Per-user quotas are part of the planned WebSocket transport.

Planned per-user limits:

Maximum concurrent connections: 5
Maximum subscriptions per connection: 50
Maximum subscriptions per user: 100

Exceeding limits:

{
  "messageId": "6f1c2f3a-8b7d-4a1e-9c22-1a5f0d2b7e44",
  "type": "error",
  "timestamp": "2024-01-16T14:30:00Z",
  "code": "TOO_MANY_SUBSCRIPTIONS",
  "message": "Maximum 50 subscriptions per connection",
  "details": { "current": 50, "max": 50 }
}

Load Balancing

Planned (post-GA) — not in v1. This section and Horizontal Scaling with Redis below describe the future multi-node architecture. The shipping realtime service is single-instance only: RealtimeServicePlugin accepts just adapter: 'memory' (the in-memory InMemoryRealtimeAdapter), there is no WebSocket transport to load-balance, and no Redis-backed realtime adapter exists yet. Multi-node HA — a Redis adapter over the existing RedisPubSub in @objectstack/service-cluster-redis — is a post-GA fast-follow.

WebSocket connections are sticky sessions:

flowchart LR
    C["Client"] --> LB["Load Balancer<br/>(remembers via session-ID cookie)"] --> S["Server A"]
    C2["Client reconnects"] -.->|routed to same server| S

Why sticky sessions:

Subscription state lives in server memory
Reconnecting to different server would lose subscriptions
Load balancer uses session ID cookie to route to same server

Horizontal Scaling with Redis

For multi-server deployments, ObjectStack uses Redis pub/sub:

flowchart LR
    C1["Client 1"] --> SA["Server A"]
    SA <-->|pub/sub| R["Redis Pub/Sub"]
    R <-->|pub/sub| SB["Server B"]
    SB --> C2["Client 2"]

How it works:

User A (connected to Server A) updates a task
Server A publishes event to Redis channel objectstack:task:updated
Server B (subscribed to Redis) receives event
Server B pushes event to User B via WebSocket

Result: Multi-server deployments with real-time sync across all servers.

Real-World Use Cases

Case 1: Collaborative Task Board

Challenge: Team of 20 uses Kanban board. When one person moves a task, others see stale state until they refresh.

Real-Time Solution:

// Each client subscribes to tasks
ws.subscribe({
  object: 'task',
  events: ['created', 'updated', 'deleted'],
  filter: { project_id: currentProject.id }
});

// When any user drags task to new column
await api.updateTask(taskId, { status: 'in_progress' });

// All 20 clients instantly see the task move
ws.on('task.updated', (task) => {
  moveTaskOnBoard(task.id, task.status);
});

Value: Zero conflicts. Everyone sees same board state in real-time. No "refresh to see latest" UX.

Case 2: Customer Support Dashboard

Challenge: Support agents answer tickets. When agent A claims a ticket, agent B tries to claim the same ticket 2 seconds later (both see it as "unclaimed").

Real-Time Solution:

// All agents subscribe to ticket updates
ws.subscribe({
  object: 'support_ticket',
  events: ['updated'],
  filter: { status: 'open' }
});

// Agent A claims ticket
await api.updateTicket(ticketId, { 
  assigned_to: 'agent_a',
  status: 'in_progress' 
});

// Agent B's dashboard instantly updates
ws.on('ticket.updated', (ticket) => {
  if (ticket.status !== 'open') {
    removeFromAvailableQueue(ticket.id);
  }
});

Value: Zero duplicate work. Tickets disappear from queue the instant they're claimed.

Case 3: Live Notifications

Challenge: Users need instant alerts for mentions, assignments, approvals.

Real-Time Solution:

// Subscribe to user's notification stream
ws.subscribe({
  object: 'notification',
  events: ['created'],
  filter: { recipient_id: currentUser.id }
});

ws.on('notification.created', (notification) => {
  showToast(notification.title, notification.message);
  updateNotificationBadge();
  playSound();
});

Value: Instant notifications. No 5-second polling delay. Lower server load.

Case 4: IoT Device Monitoring

Challenge: Factory has 500 IoT sensors. Dashboard needs to show live temperature, pressure, vibration data.

Real-Time Solution:

// Subscribe to all sensor readings
ws.subscribe({
  object: 'sensor_reading',
  events: ['created'],
  filter: { factory_id: 'factory_123' }
});

ws.on('sensor_reading.created', (reading) => {
  updateGauge(reading.sensor_id, reading.value);
  
  if (reading.value > reading.threshold) {
    triggerAlert(reading.sensor_id, 'THRESHOLD_EXCEEDED');
  }
});

Value: Real-time monitoring. Instant alerts when thresholds exceeded. No polling 500 sensors every second.

Debugging Real-Time Connections

Connection State

Check WebSocket connection state:

console.log(ws.readyState);
// 0: CONNECTING
// 1: OPEN
// 2: CLOSING
// 3: CLOSED

Message Logging

Enable debug logging:

ws.onmessage = (event) => {
  const message = JSON.parse(event.data);
  console.log('[RX]', message.type, message);
};

const originalSend = ws.send.bind(ws);
ws.send = (data) => {
  const message = JSON.parse(data);
  console.log('[TX]', message.type, message);
  originalSend(data);
};

Server-Side Events

The /api/v1/realtime/events event-log endpoint shown below is not implemented — there is no delivery-log route in the realtime service today. This is part of the planned transport tooling.

Request server event log (planned):

GET /api/v1/realtime/events?subscription_id=sub_1&limit=100
Authorization: Bearer <token>

Response:

{
  "success": true,
  "data": [
    {
      "id": "event_123",
      "subscription_id": "sub_1",
      "event": "created",
      "object": "task",
      "timestamp": "2024-01-16T14:30:00Z",
      "delivered": true
    },
    {
      "id": "event_124",
      "subscription_id": "sub_1",
      "event": "updated",
      "object": "task",
      "timestamp": "2024-01-16T14:35:00Z",
      "delivered": false,
      "reason": "client_disconnected"
    }
  ]
}

Common Issues

Problem: Events not received after reconnection

Solution: Client should resubscribe after authentication:

ws.onopen = () => {
  authenticate().then(() => {
    subscriptions.forEach(sub => subscribe(sub));
  });
};

Problem: Duplicate events

Solution: Use subscription IDs to deduplicate:

const seenEvents = new Set();

ws.on('event', (event) => {
  if (seenEvents.has(event.id)) {
    return;  // Skip duplicate
  }
  seenEvents.add(event.id);
  processEvent(event);
});

Problem: Memory leak from subscriptions

Solution: Unsubscribe when component unmounts:

useEffect(() => {
  const subId = ws.subscribe({ object: 'task', ... });
  
  return () => {
    ws.unsubscribe(subId);
  };
}, []);

Best Practices

Bad: Subscribe to entire object, filter client-side

ws.subscribe({ object: 'task' });  // Receive ALL tasks
ws.on('task.created', (task) => {
  if (task.assignee_id === currentUser.id) {  // Filter client-side
    showTask(task);
  }
});

Good: Subscribe with server-side filter

ws.subscribe({
  object: 'task',
  filter: { assignee_id: currentUser.id }
});

✅ Implement Reconnection Logic

Bad: No reconnection handling

const ws = new WebSocket(url);
// Connection drops → User sees stale data forever

Good: Automatic reconnection with exponential backoff

class ReconnectingWebSocket {
  connect() {
    this.ws = new WebSocket(this.url);
    this.ws.onclose = () => {
      setTimeout(() => this.connect(), this.backoff());
    };
  }
}

✅ Unsubscribe When Not Needed

Bad: Keep subscriptions active on hidden tabs

ws.subscribe({ object: 'task' });
// User switches tabs → Still receiving events and processing

Good: Pause/resume subscriptions based on visibility

document.addEventListener('visibilitychange', () => {
  if (document.hidden) {
    ws.unsubscribe(subId);
  } else {
    subId = ws.subscribe({ object: 'task' });
  }
});

✅ Handle Permission Changes

Good: React to visibility changes

ws.on('task.updated', (task) => {
  if (task.reason === 'exited_query') {
    removeFromUI(task.id);  // User no longer has access
  } else if (task.reason === 'entered_query') {
    addToUI(task);  // User gained access
  } else {
    updateInUI(task);
  }
});

Security Considerations

Authentication Required

Every WebSocket connection is admitted on the upgrade request, before the socket exists — there is no authentication message to send afterwards, because WebSocketMessageType declares none. A connection that reaches onopen was already admitted; one that was refused never opened, and the client sees the HTTP rejection or an immediate close. See Authentication for the declared seam (WebSocketConfig.headers into IRealtimeService.handleUpgrade).

Token Expiration

Handle JWT expiration gracefully:

{
  "messageId": "9a3f5d61-2c40-4f7b-b8e1-73d5a6c9f012",
  "type": "error",
  "timestamp": "2024-01-16T14:30:00Z",
  "code": "TOKEN_EXPIRED",
  "message": "JWT token expired",
  "details": { "expires_at": "2024-01-16T14:30:00Z" }
}

Client response: Refresh token and reconnect

ws.onmessage = async (event) => {
  const msg = JSON.parse(event.data);
  if (msg.type === 'error' && msg.code === 'TOKEN_EXPIRED') {
    const newToken = await refreshToken();
    ws.close();
    reconnect(newToken);
  }
};

Rate Limiting

Not implemented. Nothing rate-limits WebSocket messages today. No producer anywhere in this repository emits RATE_LIMITED on a realtime path, and there is no WebSocket transport for such a message to arrive on (see the implementation-status callout at the top of this page). The budget below is a planned one, and the envelope beside it is what ErrorMessageSchema would require of a producer that eventually enforces it — not a shape any code emits today. HTTP requests are rate-limited, under a different code and envelope: see Error Handling for RATE_LIMIT_EXCEEDED.

Planned limits (not enforced):

Max 100 messages per minute per connection
Max 10 subscriptions per minute per connection

{
  "messageId": "c4e7b9d2-5a31-4c8f-9e60-2b8d41f7a305",
  "type": "error",
  "timestamp": "2024-01-16T14:30:00Z",
  "code": "RATE_LIMITED",
  "message": "Too many messages"
}

Real-time protocols — websockets, SSE

Instant User Experience

95% Less Server Load

Enable Collaboration

Live Notifications

HTTP Protocol

Error Handling

On this page