title	Real-Time Protocols
description	WebSocket subscriptions, Server-Sent Events, and event-driven communication in ObjectStack

import { Zap, Radio, Activity, Bell, Users, Gauge, AlertCircle } from 'lucide-react';

Real-Time Protocols

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. The authz conformance matrix (realtime-delivery-authz row + transport tripwires) enforces this in CI.

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

} title="Instant User Experience" description="Changes appear immediately across all connected clients. No refresh required." /> } title="95% Less Server Load" description="Replace millions of polling requests with persistent connections. Massive infrastructure savings." /> } title="Enable Collaboration" description="Multiple users editing the same data see each other's changes in real-time. Google Docs-style UX." /> } title="Live Notifications" description="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. The service itself appears in services.realtime as degraded with handlerReady: false when registered:

GET /.well-known/objectstack

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

A WebSocket upgrade endpoint is part of the planned transport (`IRealtimeService.handleUpgrade()`) and is not yet served. When it lands, discovery will advertise `routes.realtime` again — until then clients must treat `services.realtime.handlerReady: false` as "no wire transport" (see #2462).

Establishing Connection

Client-side (JavaScript):

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

ws.onopen = () => {
  console.log('Connected to ObjectStack');
  
  // Authenticate
  ws.send(JSON.stringify({
    type: 'auth',
    token: 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...'
  }));
};

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

Send authentication message immediately after connection:

Request:

{
  "type": "auth",
  "token": "Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..."
}

Success Response:

{
  "type": "auth_success",
  "user_id": "user_123",
  "session_id": "session_abc",
  "expires_at": "2024-01-16T22:30:00Z"
}

Failure Response:

{
  "type": "auth_error",
  "error": {
    "code": "INVALID_TOKEN",
    "message": "JWT token expired"
  }
}

Connection closes on auth failure: Server closes WebSocket if authentication fails within 10 seconds.

Subscriptions

Subscribe to Object Events

Subscribe to changes on a specific object:

Request:

{
  "type": "subscribe",
  "subscription_id": "sub_1",
  "object": "task",
  "events": ["created", "updated", "deleted"],
  "filter": {
    "assignee_id": "user_123"
  }
}

Parameters:

subscription_id: Client-generated unique ID for this subscription
object: Object name to subscribe to
events: Array of events to listen for (default: all)
filter: Optional filter (same syntax as HTTP API filters)

Success Response:

{
  "type": "subscribed",
  "subscription_id": "sub_1",
  "object": "task",
  "events": ["created", "updated", "deleted"]
}

Error Response:

{
  "type": "error",
  "subscription_id": "sub_1",
  "error": {
    "code": "FORBIDDEN",
    "message": "No permission to subscribe to 'task' object"
  }
}

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, server pushes events:

Created Event:

{
  "type": "event",
  "subscription_id": "sub_1",
  "event": "created",
  "object": "task",
  "data": {
    "id": "task_456",
    "title": "New task assigned to you",
    "status": "todo",
    "assignee_id": "user_123",
    "created_at": "2024-01-16T14:30:00Z"
  }
}

Updated Event:

{
  "type": "event",
  "subscription_id": "sub_1",
  "event": "updated",
  "object": "task",
  "data": {
    "id": "task_456",
    "status": "in_progress",
    "updated_at": "2024-01-16T15:00:00Z"
  },
  "changes": {
    "status": {
      "old": "todo",
      "new": "in_progress"
    }
  }
}

Deleted Event:

{
  "type": "event",
  "subscription_id": "sub_1",
  "event": "deleted",
  "object": "task",
  "data": {
    "id": "task_456",
    "deleted_at": "2024-01-16T16:00:00Z"
  }
}

Unsubscribe

Stop receiving events for a subscription:

Request:

{
  "type": "unsubscribe",
  "subscription_id": "sub_1"
}

Response:

{
  "type": "unsubscribed",
  "subscription_id": "sub_1"
}

Subscribe to Specific Record

Watch a single record for changes:

Request:

{
  "type": "subscribe",
  "subscription_id": "sub_2",
  "object": "task",
  "record_id": "task_456",
  "events": ["updated", "deleted"]
}

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

Subscribe to Query Results

Subscribe to a dynamic set of records matching a query:

Request:

{
  "type": "subscribe",
  "subscription_id": "sub_3",
  "object": "task",
  "query": {
    "filter": {
      "status": "todo",
      "assignee_id": "user_123"
    },
    "sort": "-priority"
  },
  "events": ["created", "updated", "deleted"]
}

Behavior:

Receive created events when records matching query are created
Receive updated events when subscribed records change
Receive deleted events when subscribed records are deleted
Automatically receive events when records enter/exit the query filter

Example: Task enters subscription:

{
  "type": "event",
  "subscription_id": "sub_3",
  "event": "updated",
  "object": "task",
  "data": {
    "id": "task_789",
    "status": "todo",  // Changed from "in_progress" to "todo"
    "assignee_id": "user_123"
  },
  "reason": "entered_query"
}

Example: Task exits subscription:

{
  "type": "event",
  "subscription_id": "sub_3",
  "event": "updated",
  "object": "task",
  "data": {
    "id": "task_456",
    "status": "done"  // No longer matches "todo" filter
  },
  "reason": "exited_query"
}

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 {
  constructor(url, token) {
    this.url = url;
    this.token = token;
    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.authenticate();
      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
      );
    };
  }
  
  authenticate() {
    this.send({ type: 'auth', token: this.token });
  }
  
  resubscribe() {
    this.subscriptions.forEach((config, id) => {
      this.send({ ...config, subscription_id: id });
    });
  }
  
  subscribe(config) {
    const id = `sub_${Date.now()}`;
    this.subscriptions.set(id, config);
    this.send({ ...config, type: 'subscribe', subscription_id: id });
    return id;
  }
  
  send(message) {
    this.ws.send(JSON.stringify(message));
  }
}

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

Permission Enforcement

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

Scenario: User subscribes to all tasks:

{
  "type": "subscribe",
  "subscription_id": "sub_1",
  "object": "task",
  "events": ["created", "updated"]
}

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":

// User is assigned task_456
{
  "type": "event",
  "event": "updated",
  "object": "task",
  "data": { "id": "task_456", "assignee_id": "user_123" },
  "reason": "entered_query"
}

// Task reassigned to someone else
// No event sent - task is now invisible to user

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:

{
  "type": "error",
  "error": {
    "code": "TOO_MANY_SUBSCRIPTIONS",
    "message": "Maximum 50 subscriptions per connection",
    "current": 50,
    "max": 50
  }
}

Load Balancing

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

✅ Subscribe to Specific Queries

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

All WebSocket connections must authenticate within 10 seconds:

ws.onopen = () => {
  ws.send(JSON.stringify({ 
    type: 'auth', 
    token: getToken() 
  }));
};

// Server closes connection if no auth within 10 seconds

Token Expiration

Handle JWT expiration gracefully:

{
  "type": "error",
  "error": {
    "code": "TOKEN_EXPIRED",
    "message": "JWT token expired",
    "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.error?.code === 'TOKEN_EXPIRED') {
    const newToken = await refreshToken();
    ws.close();
    reconnect(newToken);
  }
};

Rate Limiting

WebSocket messages are rate-limited:

{
  "type": "error",
  "error": {
    "code": "RATE_LIMITED",
    "message": "Too many messages",
    "retry_after": 5,
    "limit": 100,
    "window": "1m"
  }
}

Limits:

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

Next Steps

} title="HTTP Protocol" description="Learn REST endpoint conventions and CRUD operations" href="/docs/protocol/kernel/http-protocol" /> } title="Error Handling" description="Master error codes and debugging strategies" href="/docs/protocol/kernel/error-handling" />

Uh oh!

FilesExpand file tree

realtime-protocol.mdx

Latest commit

History

realtime-protocol.mdx

File metadata and controls

Real-Time Protocols

Why Real-Time Matters

Business Value Delivered

Communication Channels

WebSocket (Bi-Directional)

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

WebSocket Connection

Connection Endpoint

Establishing Connection

Authentication

Subscriptions

Subscribe to Object Events

Event Types

Receiving Events

Unsubscribe

Subscribe to Specific Record

Subscribe to Query Results

Server-Sent Events (SSE)

Connection Endpoint

Subscription via Query Parameters

Event Format

Automatic Reconnection

Heartbeat & Keep-Alive

Server Heartbeat

Client Reconnection

Permission Enforcement

Scaling Considerations

Connection Limits

Load Balancing

Horizontal Scaling with Redis

Real-World Use Cases

Case 1: Collaborative Task Board

Case 2: Customer Support Dashboard

Case 3: Live Notifications

Case 4: IoT Device Monitoring

Debugging Real-Time Connections

Connection State

Message Logging

Server-Side Events

Common Issues

Best Practices

✅ Subscribe to Specific Queries

✅ Implement Reconnection Logic

✅ Unsubscribe When Not Needed

✅ Handle Permission Changes

Security Considerations

Authentication Required

Token Expiration

Rate Limiting

Next Steps