WebSocket Real-Time API

Overview

ODP provides a WebSocket endpoint for real-time data subscriptions. Clients can subscribe to changes on specific collections and receive push notifications when items are created, updated, or deleted.

The WebSocket endpoint is implemented using @fastify/websocket and a custom WebSocketService singleton.

---

Connection

Endpoint

ws://api.example.com/websocket

or with TLS:

wss://api.example.com/websocket

---

Authentication Modes

Configured via WEBSOCKETS_REST_AUTH:

Mode	Description
strict	Token required in query param at connection time. Invalid/missing token → connection refused (close code 4401)
handshake	Token optional at connection time. Client can authenticate after connecting via auth message
public	Same as handshake — unauthenticated clients connect but get limited access

Token via query parameter:

wss://api.example.com/websocket?access_token=<JWT>

---

Message Protocol

All messages are JSON-encoded objects. Every message has a type field.

Message Types

Type	Direction	Description
auth	Client → Server	Authenticate using access token
subscribe	Client → Server	Subscribe to a collection
unsubscribe	Client → Server	Cancel a subscription
ping	Client → Server	Keepalive ping
pong	Server → Client	Keepalive pong
items	Server → Client	Real-time item change notification
error	Server → Client	Error response

---

Authentication Message

When using handshake or public mode, send an auth message after connecting:

Client → Server:

{
  "type": "auth",
  "uid": "optional-request-id",
  "access_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..."
}

Server → Client (success):

{
  "type": "auth",
  "uid": "optional-request-id",
  "data": { "authenticated": true }
}

Server → Client (failure):

{
  "type": "error",
  "uid": "optional-request-id",
  "message": "Invalid access token"
}

---

Subscribe

Subscribe to real-time changes on a collection.

Client → Server:

{
  "type": "subscribe",
  "uid": "my-subscription-1",
  "collection": "articles"
}

Field	Type	Required	Description
type	"subscribe"	Yes	Message type
uid	string	Yes	Unique subscription ID (client-defined)
collection	string	Yes	Collection name to subscribe to
query	object	No	Optional filter query

Server → Client (confirmation):

{
  "type": "subscribe",
  "uid": "my-subscription-1",
  "data": {
    "subscribed": true,
    "collection": "articles"
  }
}

---

Unsubscribe

Cancel a subscription.

Client → Server:

{
  "type": "unsubscribe",
  "uid": "my-subscription-1"
}

Server → Client:

{
  "type": "unsubscribe",
  "uid": "my-subscription-1",
  "data": { "unsubscribed": true }
}

If the subscription did not exist, unsubscribed is false.

---

Item Change Notifications

When an item in a subscribed collection is created, updated, or deleted, all subscribed clients receive a push message:

Server → Client (create):

{
  "type": "items",
  "event": "create",
  "collection": "articles",
  "uid": "my-subscription-1",
  "data": {
    "keys": ["new-item-uuid"],
    "payload": { "id": "new-item-uuid", "title": "New Article", "status": "draft" }
  }
}

Server → Client (update):

{
  "type": "items",
  "event": "update",
  "collection": "articles",
  "uid": "my-subscription-1",
  "data": {
    "keys": ["item-uuid-1", "item-uuid-2"],
    "payload": { "status": "published" }
  }
}

Server → Client (delete):

{
  "type": "items",
  "event": "delete",
  "collection": "articles",
  "uid": "my-subscription-1",
  "data": {
    "keys": ["deleted-item-uuid"]
  }
}

---

Heartbeat (Ping/Pong)

ODP sends periodic ping messages to all connected clients. Clients should respond with pong to prevent disconnection.

Server → Client (server-initiated ping):

{ "type": "ping" }

Client → Server (client-initiated ping):

{ "type": "ping" }

Server → Client (response to client ping):

{ "type": "pong" }

Stale Connection Detection

The heartbeat timer (period: WEBSOCKETS_HEARTBEAT_PERIOD * 1000 ms) runs periodically and:

1. Closes clients that have not responded within 2 × heartbeat_period (close code 4408)
2. Sends ping to all remaining clients

---

Connection Lifecycle

---

Error Messages

Message	Cause
Invalid JSON	Malformed JSON message
Missing message type	Message has no type field
Missing access_token	auth message sent without access_token
Invalid access token	Token validation failed
Missing collection	subscribe message has no collection
Missing uid for subscription	subscribe message has no uid
Missing uid for unsubscribe	unsubscribe message has no uid
Unknown message type: <type>	Unrecognized message type

---

Configuration

Variable	Default	Description
WEBSOCKETS_ENABLED	true	Enable/disable WebSocket endpoint
WEBSOCKETS_HEARTBEAT_ENABLED	true	Enable heartbeat
WEBSOCKETS_HEARTBEAT_PERIOD	30	Heartbeat interval in seconds
WEBSOCKETS_REST_AUTH	handshake	Auth mode: public, handshake, strict
WEBSOCKETS_REST_ENABLED	true	Enable REST-style WebSocket
WEBSOCKETS_GRAPHQL_ENABLED	true	Enable GraphQL WebSocket (if implemented)
WEBSOCKETS_GRAPHQL_AUTH	handshake	Auth mode for GraphQL WS

---

Event Sources

WebSocket notifications are driven by ODP's internal event emitter. The WebSocketService listens to:

Emitter Event	WS Event
items.create	create
items.update	update
items.delete	delete

These events are fired by ItemsService after successful mutations, so all writes through the standard REST API, GraphQL, and MCP will trigger WebSocket notifications.

---

JavaScript Client Example

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

ws.onopen = () => {
  // Authenticate
  ws.send(JSON.stringify({
    type: 'auth',
    access_token: 'your-access-token'
  }));
};

ws.onmessage = ({ data }) => {
  const msg = JSON.parse(data);

  if (msg.type === 'auth' && msg.data?.authenticated) {
    // Subscribe to articles collection
    ws.send(JSON.stringify({
      type: 'subscribe',
      uid: 'articles-sub',
      collection: 'articles'
    }));
  }

  if (msg.type === 'items') {
    console.log(`${msg.event} in ${msg.collection}:`, msg.data);
  }

  if (msg.type === 'ping') {
    ws.send(JSON.stringify({ type: 'ping' }));
  }
};