WebSocket API Protocol

🔌 Complete WebSocket Specification

Connection

WS /v1/realtime

Implemented in src/realtime/mod.rs. On connect, table-level changes are delivered via Postgres LISTEN/NOTIFY fanned out to subscribers through in-process tokio::broadcast channels — not logical replication/WAL streaming.

Message Format

All messages are JSON. Client and server messages have different shapes (there is no shared envelope with a key/value/seq structure).

Client → Server Message Types

{
  "type": "subscribe",
  "table": "users",
  "event": "*",
  "filter": "..."
}

table is required. event (e.g. "INSERT", "UPDATE", "DELETE", "*") and filter are accepted in the message but are not currently enforced by the server — a subscription receives every INSERT/UPDATE/DELETE event for the table regardless of what you pass for event/filter. Filter client-side until this is implemented server-side.

2. Unsubscribe

{
  "type": "unsubscribe",
  "table": "users"
}

3. Ping

{"type": "ping"}

Server → Client Message Types

1. Connected (sent immediately on connect)

{
  "type": "connected",
  "message": "Connected to Stackhouse Realtime",
  "client_id": 1
}

2. Subscribed / Unsubscribed (ack)

{"type": "subscribed", "table": "users", "event": "*"}
{"type": "unsubscribed", "table": "users"}

3. Data events

{
  "type": "INSERT",
  "table": "users",
  "record": {"id": 1, "name": "Alice"},
  "timestamp": "2025-01-03T12:00:00Z"
}

{
  "type": "UPDATE",
  "table": "users",
  "record": {"id": 1, "name": "Alice B."},
  "old_record": {"id": 1, "name": "Alice"},
  "timestamp": "2025-01-03T12:00:00Z"
}

{
  "type": "DELETE",
  "table": "users",
  "old_record": {"id": 1, "name": "Alice B."},
  "timestamp": "2025-01-03T12:00:00Z"
}

record is present for INSERT/UPDATE, old_record for UPDATE/DELETE.

4. Error

{
  "type": "error",
  "message": "Invalid message format: ..."
}

Sent for malformed JSON or an unrecognized type.

5. Pong

{"type": "pong"}

Best Practices

Handle reconnection

ws.addEventListener('close', () => {
  setTimeout(() => {
    ws = new WebSocket(url);
  }, 1000);
});

Resubscribe on reconnect

const tables = ['users', 'documents'];
ws.onopen = () => {
  tables.forEach(table => {
    ws.send(JSON.stringify({ type: 'subscribe', table, event: '*' }));
  });
};

Error handling

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

ws.onmessage = (event) => {
  const msg = JSON.parse(event.data);
  if (msg.type === 'error') {
    console.error('Realtime error:', msg.message);
  }
};

Presence tracking and channel broadcast are separate REST endpoints (not WebSocket messages), also mounted under /v1/realtime — see src/realtime/presence.rs and src/realtime/broadcast.rs:

Method	Path
POST	`/v1/realtime/presence/track`
POST	`/v1/realtime/presence/untrack`
GET	`/v1/realtime/presence/:channel`
GET	`/v1/realtime/presence`
POST	`/v1/realtime/broadcast/send`
GET	`/v1/realtime/broadcast/channels`
GET	`/v1/realtime/broadcast/:channel/history`

Done! 🎉