Realtime WebSocket API

Connection sequence

1. Mint a short-lived ticket with POST /v1/realtime/session.
2. Open wss://api.dyva.ai/v1/ws without credentials in the URL.
3. Within five seconds, send the protocol-v2 hello frame containing the ticket.
4. Wait for ready, then subscribe to the topics your principal is allowed to read.

curl -X POST https://api.dyva.ai/v1/realtime/session \
  -H "Authorization: Bearer YOUR_ACCESS_TOKEN"

{
  "type": "hello",
  "protocol": 2,
  "token": "<short-lived realtime ticket>"
}

Ready and subscribe

A successful hello returns the server cursor, heartbeat cadence, principal, and default subscriptions. Only subscribe after this frame. Resource topics still pass database authorization even when the ticket is valid.

{
  "type": "ready",
  "protocol": 2,
  "connection_id": "018f3f5a-6c91-7ee1-83d1-469a7f456b10",
  "principal": {
    "kind": "user",
    "id": "user_abc123"
  },
  "topics": [
    "public:platform",
    "user:user_abc123",
    "session:session_abc123"
  ],
  "cursor": "1722190000000-0",
  "heartbeat_ms": 25000,
  "server_time": "2026-07-30T12:00:00.000Z"
}

{
  "type": "subscribe",
  "topics": [
    "conversation:conv_abc123"
  ],
  "cursor": "1722190000000-0",
  "limit": 200
}

Client controls

Controls use top-level fields. Send domain writes, including new messages, through the corresponding REST endpoint; the socket carries subscriptions, presence, delivery, and recovery.

subscribe

Request one or more authorized topics. Include a cursor to replay missed durable events.

{ "topics": ["conversation:..."], "cursor"?: "...", "limit"?: 200 }

unsubscribe

Remove requested topics that are not connection defaults.

{ "topics": ["conversation:..."] }

resume

Replay all current subscriptions after the last fully applied cursor.

{ "cursor": "...", "limit"?: 200 }

ack

Tell this socket which cursor was applied. Persist the cursor in your own client too.

{ "cursor": "..." }

typing

Publish ephemeral typing presence to an authorized, already-subscribed resource topic.

{ "topic": "conversation:...", "active": true }

ping

Renew presence and receive the current stream head cursor.

{}

Server controls

Control frames use type. Domain events use the versioned envelope below and identify their schema with name.

ready

Authentication succeeded. Includes the principal, default topics, cursor, and heartbeat cadence.

subscribed

Reports accepted and rejected topics. A credential never bypasses topic authorization.

unsubscribed

Reports the topics removed from this connection.

replay_started

A bounded replay is beginning for accepted topics.

replay_complete

Replay finished. Resume again from its cursor when truncated is true.

resync_required

The cursor cannot be replayed safely. Fetch an authoritative REST snapshot before advancing.

pong

Heartbeat response with the current stream head cursor.

error

A named protocol or rate-limit error scoped to this connection.

Event envelope

Durable delivery is at least once. Validate the event version, deduplicate by event_id, respect aggregate revisions when present, and persist the cursor only after applying or intentionally ignoring the event.

{
  "protocol": 2,
  "event_id": "5ca8eb59-4d89-4acc-9242-3d7ce7f77b91",
  "name": "conversation.updated",
  "version": 1,
  "cursor": "1722190000000-1",
  "topics": [
    "conversation:conv_abc123"
  ],
  "occurred_at": "2026-07-30T12:00:00.000Z",
  "produced_at": "2026-07-30T12:00:00.050Z",
  "actor": {
    "user_id": "user_abc123"
  },
  "aggregate": {
    "kind": "conversation",
    "id": "conv_abc123",
    "revision": 42
  },
  "data": {
    "conversation_id": "conv_abc123"
  },
  "trace_id": "trace_abc123"
}

Replay and recovery

{
  "type": "resume",
  "cursor": "1722190000000-0",
  "limit": 200
}

{
  "type": "ack",
  "cursor": "1722190000000-0"
}

Guests and public topics

A protocol-v2 hello may omit the token for anonymous, read-only access to allowed public:* collection topics. Anonymous sockets cannot subscribe to resource topics, publish typing, or use replay. Mint a guest ticket when a logged-out browser needs an authorized resource or replay.

{
  "type": "hello",
  "protocol": 2
}

Typing presence

Typing is ephemeral and requires an authenticated user, an existing subscription, and separate write authorization for the topic. Send a stop frame when possible and expire remote typing locally if a stop event is lost.

{
  "type": "typing",
  "topic": "conversation:conv_abc123",
  "active": true
}

{
  "type": "typing",
  "topic": "conversation:conv_abc123",
  "active": false
}

Close and recovery signals

4401

The credential, session, or API key expired or was revoked.

Mint a new ticket, reconnect, then resume from the last applied cursor.

4429

A rate limit or slow-consumer threshold was reached.

Back off with jitter and reduce inbound or outbound pressure.

1012

The gateway lost its realtime subscriber and closed fail-safe.

Reconnect and resume after a short backoff.

1013

The gateway is still synchronizing and cannot authenticate safely.

Retry with exponential backoff and jitter.

Resilient client example

This example mints a fresh ticket for each connection, waits for ready, restores a subscription, deduplicates at-least-once events, acknowledges applied cursors, and reconnects with backoff.

const API = "https://api.dyva.ai/v1/realtime/session";
const SOCKET = "wss://api.dyva.ai/v1/ws";
const topic = "conversation:conv_abc123";
const seenEventIds = new Set();

let lastCursor = sessionStorage.getItem("dyva:realtime:cursor");
let socket;
let reconnectAttempt = 0;
let stopped = false;

async function mintTicket(accessToken) {
  const response = await fetch(API, {
    method: "POST",
    headers: accessToken
      ? { Authorization: `Bearer ${accessToken}` }
      : undefined,
  });

  if (!response.ok) {
    throw new Error(`Realtime session failed (${response.status})`);
  }

  return response.json();
}

async function applyEvent(event) {
  // Validate event.name, event.version, and event.data for your domain.
  console.log(event.name, event.data);
}

async function fetchAuthoritativeSnapshot(topics) {
  // Fetch the REST snapshot for each affected topic before advancing.
  console.log("Snapshot required for", topics);
}

async function connect(accessToken) {
  const session = await mintTicket(accessToken);
  socket = new WebSocket(SOCKET);

  socket.addEventListener("open", () => {
    socket.send(JSON.stringify({
      type: "hello",
      protocol: 2,
      token: session.token,
    }));
  });

  socket.addEventListener("message", async ({ data }) => {
    const message = JSON.parse(data);

    if (message.type === "ready") {
      reconnectAttempt = 0;
      socket.send(JSON.stringify({
        type: "subscribe",
        topics: [topic],
        ...(lastCursor ? { cursor: lastCursor, limit: 200 } : {}),
      }));
      return;
    }

    if (message.type === "resync_required") {
      await fetchAuthoritativeSnapshot(message.topics ?? [topic]);
      lastCursor = message.cursor ?? null;
      if (lastCursor) {
        sessionStorage.setItem("dyva:realtime:cursor", lastCursor);
      }
      return;
    }

    if (message.type === "replay_complete" && message.truncated) {
      socket.send(JSON.stringify({
        type: "resume",
        cursor: message.cursor,
        limit: 200,
      }));
      return;
    }

    if (message.protocol === 2 && message.event_id) {
      if (seenEventIds.has(message.event_id)) return;
      await applyEvent(message);
      seenEventIds.add(message.event_id);

      if (message.cursor) {
        lastCursor = message.cursor;
        sessionStorage.setItem("dyva:realtime:cursor", lastCursor);
        socket.send(JSON.stringify({ type: "ack", cursor: lastCursor }));
      }
    }
  });

  socket.addEventListener("close", () => {
    if (stopped) return;
    const delay = Math.min(1_000 * 2 ** reconnectAttempt, 30_000);
    reconnectAttempt += 1;
    setTimeout(() => void connect(accessToken), delay + Math.random() * 250);
  });
}

function disconnect() {
  stopped = true;
  socket?.close(1000, "Client closed");
}

void connect("YOUR_ACCESS_TOKEN");

Current gateway limits

• 20 concurrent connections per client IP per gateway process.
• 64 KiB maximum application message size.
• 64 normalized topics per connection.
• 120 inbound frames per 10 seconds.
• 20 subscribe operations per minute per client IP.
• 4 replay operations per minute per client IP.
• 30 typing operations per 10 seconds per client IP.
• 500 delivered events per replay request at most.