Realtime

Use realtime protocol 4 when a bot or client must react to activity on a Chatto server. The protocol sends binary protobuf frames over a WebSocket at /api/realtime. Realtime events tell you what happened. ConnectRPC gives you the resource detail.

Need	Use
React to a message, edit, reaction, membership change, or other activity	Realtime event
Create, update, or delete a resource	ConnectRPC
Read one resource, a paginated list, or history	ConnectRPC
Build a current local content view	Realtime snapshot
Repair a short connection gap	Realtime resume

The realtime reference lists all frames, events, and enums.

Connection lifecycle

1. Open a WebSocket to wss://chat.example.com/api/realtime.
2. Send one binary RealtimeSubscribe message. Request protocol version 4, and select SNAPSHOT or LIVE_ONLY as the initial-state fallback. Include a bearer credential if you do not use a same-origin cookie. Optionally, include a resume cursor.
3. Send no more application messages on this socket.
4. If the server sends snapshot, replace all snapshot resource families.
5. Process replayed events, if any. The server sends caught_up when it reaches the live boundary.
6. Reconcile the latest-value resources that your client keeps. Then save the caught_up cursor and continue to process events.

Treat the stream as live only after caught_up. Its recovery field is RESUMED, SNAPSHOT, or LIVE_ONLY. A replay with no events still reports RESUMED. If you sent a cursor but receive LIVE_ONLY, the server did not recover the gap. Read current state if your integration needs it.

Choose snapshot or live-only startup

• LIVE_ONLY is for a bot that wants new events and reads resources through ConnectRPC. Without a usable cursor, the server starts at the current boundary and sends no old events or resources.
• SNAPSHOT is for a client that keeps a local copy of server state. Without a usable cursor, the server sends an exact snapshot at boundary E, then caught_up(E). The bundled frontend uses this mode.

With a usable cursor, both modes resume. The server replays the events after the cursor and then sends caught_up. A cursor becomes unusable when it is invalid, expired, too far behind, from a different server history, or unsafe for current authorization. Then the server uses the fallback. It never sends a partial replay.

Caution

A cursor is valid for 15 minutes and belongs to the authenticated viewer. Keep it with the client state that it describes. Do not apply it to unrelated state. Cursors are opaque: store and return them unchanged.

Handle public events

Each RealtimeEvent contains one public payload, such as a message, edit, retraction, reaction, pin, asset, room, membership, profile, thread, call, typing, or presence change. It also contains a stable event ID, the source time, and the actor ID when applicable. An event has a cursor when the server can replay it. Cursorless events are safe to miss, or a read of the current resource restores their effect.

Process each event as one unit:

1. Decode the complete frame.
2. Process the variants that you use. Ignore the others.
3. Save cursor only after your processing succeeds.

Events can repeat across reconnects. Use the event ID to deduplicate. Resume does not guarantee delivery across a long outage. If you cannot decode a complete frame, close the socket and keep the previous cursor. New event variants are additive.

Messages and mentions

• MessagePostedEvent.room_kind tells you whether the room is a channel or a DM. thread_root_event_id identifies the thread root of a reply.
• body_plaintext lets you show a new message immediately. Use MessageService.GetMessage for attachments, link previews, reactions, pin state, and thread details.
• Edit and retraction events use message_event_id for their target message.
• After a post, refresh the viewer’s Slow Mode deadline from the message or thread timeline response. RoomSlowModeChangedEvent carries the channel and thread intervals; refresh room state and open thread deadlines when it changes.
• mentions lists each direct user, role, here, or all target once. A direct target has direct.user_id. includes_viewer tells you whether the target included you when the message was posted. It keeps this meaning during replay.

A bot that responds only to direct mentions checks for a direct target with includes_viewer = true. A role or broadcast mention is not a direct mention.

Resource-change events

RoomMembershipEventsVisibilityChangedEvent changes a room’s join/leave display setting. Refresh the room resource or apply show_membership_events to its current state. Apply it to history and new timeline rows. Continue to process membership events when it is false; the setting does not change membership or delivery.

Some events tell you only that a resource changed. Read the resource with the event cursor:

Event	Read
UserProfileChangedEvent	The user profile
ViewerPreferencesChangedEvent	Your preferences
ServerProfileChangedEvent	The server profile
RoomLayoutChangedEvent	RoomDirectoryService.ListRoomGroups
RoomNoticeChangedEvent	RoomService.GetNotice for room_id
Asset processing or deletion	The message in message_event_id

Room notice events contain only the room ID. Current room members can read the notice. Refresh notices after reconnecting, and discard stored notice content when room membership ends. A correction keeps publication_id; explicit republishing changes it so a dismissed notice becomes visible again.

Refresh presence and typing

Call MyAccountService.SetPresence every 30 seconds while your client is visible. Presence expires after 60 seconds without a refresh. To appear offline, stop these calls. Another client of the same account can keep presence active.

Call RoomService.RefreshTypingIndicator at most every 2 seconds while the user types. Include threadRootEventId for a thread. Stop when typing stops.

When you receive UserTypingEvent, track the actor for each room and thread. Clear the indicator 6 seconds after the last typing event, or when the actor posts in that room or thread. There is no stop-typing event, and resume does not replay typing events.

Notification creation hints

notification_occurrences_changed.created_notification_id identifies a new notification, also during Do Not Disturb or when it starts as read. Read the current state with NotificationService.ListNotificationOccurrences and decide from your preferences whether to alert. Resume does not replay these hints. Read the notification list again after you reconnect, and do not use this event as a durable bot trigger.

Apply the exact snapshot at E

DM room resources can include last_message_at and last_root_message_at. The first includes thread replies; the second includes only root posts. Both require broad message.read permission. After a snapshot, use MessagePostedEvent and its envelope created_at to update these values. Do not derive broad activity from events delivered with interaction-only access. Edits and retractions do not change posting activity. An absent timestamp does not prove inactivity; older servers can omit these fields.

When the server sends snapshot, clear the state that it replaces. The frame contains these complete resource families:

Snapshot value	Meaning
ServerPublicProfile	Current public server profile
Repeated RoomWithViewerState	Complete visible room directory
Repeated RoomGroup	Complete visible room-group layout
Repeated DirectoryMember	The viewer and users referenced by snapshot resources
Repeated ActiveCall	Complete visible active-call state

The snapshot does not contain the complete user directory, message or thread history, search results, files, pins, room notices, notifications, presence, read markers, or account-security state. Read these through ConnectRPC when you need them.

Do not show the new state as current before caught_up. If the socket closes first, discard the snapshot and reconnect without its cursor.

Resume does not replay cursorless events. After every caught_up, not only after a snapshot, read the latest-value resources that your client keeps, such as the viewer, notifications, and presence.

Bind a targeted resource read to an event

When a cursor-bearing event causes a ConnectRPC read, set this header:

Chatto-Realtime-Minimum-Cursor: <event.cursor>

The server waits until its content includes that event before it runs the request. This prevents stale reads from another replica. All authenticated unary procedures accept the header. The server waits for at most 10 seconds or your earlier deadline. A timeout returns DEADLINE_EXCEEDED and does not run the operation. The response can contain newer content. The header does not wait for notifications, search indexing, or other asynchronous work.

Save the event cursor only after the read succeeds. If a new snapshot starts, discard late responses from the earlier state.

Detect a dead connection

The server sends a heartbeat approximately every 15 seconds. Reset a timer on every server frame, and replace the socket when no frame arrives for 75 seconds. The client sends no application ping. A heartbeat can contain a fresh cursor. Save it only after all earlier work succeeds.

Handle errors and reconnects

Frame	Client action
close with reconnect: true	Close the socket, wait for retry_after or local backoff, and subscribe with the last safe cursor.
close with reconnect: false	Stop automatic reconnects and require the applicable user or authentication action.
Network close without guidance	Retry with bounded exponential backoff and the last safe cursor.

• Expired bearer token: rotate the credential and reconnect.
• Session renewal close: same-origin cookie clients renew the browser session, then reconnect.
• SESSION_TERMINATED: stop reconnecting and return to sign-in.
• RESYNC_REQUIRED: discard private cached data and the cursor, then reconnect. Snapshot clients replace their state. Live-only clients read the current resources that they need.

Role changes arrive as normal events and do not close the connection. Read role metadata after role_created or role_updated. Apply roles_reordered to display order, and role_assigned and role_revoked to member role lists.

Reload all private cached data when:

• an assignment event names your viewer;
• role_deleted or role_permissions_changed affects a role that you hold;
• the permissions of everyone change;
• you receive viewer_permissions_changed. Bots also receive it when the authority of their owner can change.

Clear data before you start the new reads, and reject older responses that arrive later. If you do not know the roles of your viewer, reload.

Use realtime from TypeScript

This example builds a browser client that processes events in order and reconnects with its last safe cursor. The same pattern works for a bot runtime that has a WebSocket implementation.

Generate TypeScript bindings

You need a server URL, a bearer credential for a user or bot, Buf, Node.js, and the proto/ directory from the exact Chatto release that your server runs. Do not mix schemas from different releases.

1. Create a client project and install the protobuf runtime.

   mkdir chatto-realtime-client
   cd chatto-realtime-client
   npm init -y
   npm install @bufbuild/protobuf@1.10.1
   npm install --save-dev @bufbuild/protoc-gen-es@1.10.1 typescript vite
   npm pkg set type=module scripts.dev=vite scripts.check="tsc --noEmit"

2. Copy the proto/ directory of the matching release into the project, with its buf.yaml file and all imported definitions.

3. Create tsconfig.json.

   {
     "compilerOptions": {
       "target": "ES2022",
       "module": "NodeNext",
       "moduleResolution": "NodeNext",
       "lib": ["ES2022", "DOM"],
       "strict": true,
       "noEmit": true
     },
     "include": ["src"]
   }

4. Create buf.gen.yaml.

   version: v2
   plugins:
     - local: node_modules/.bin/protoc-gen-es
       out: src/gen
       opt:
         - target=ts

5. Generate the bindings into src/gen/.

   buf generate proto --template buf.gen.yaml

Create an event client

Create src/realtime.ts:

import {
  RealtimeEvent,
  RealtimeCloseCode,
  RealtimeRecovery,
  RealtimeInitialState,
  RealtimeServerFrame,
  RealtimeSnapshot,
  RealtimeSubscribe,
} from "./gen/chatto/realtime/v1/realtime_pb.js";

const PROTOCOL_VERSION = 4;
const BASE_RECONNECT_MS = 1_000;
const MAX_RECONNECT_MS = 30_000;

function durationMs(value?: { seconds: bigint; nanos: number }): number {
  return value ? Number(value.seconds) * 1_000 + value.nanos / 1_000_000 : 0;
}

type ChattoRealtimeClientOptions = {
  initialState?: RealtimeInitialState;
  beginSnapshot?: () => void | Promise<void>;
  applySnapshot?: (snapshot: RealtimeSnapshot) => void | Promise<void>;
  completeCatchUp?: (cursor: string, signal: AbortSignal, recovery: RealtimeRecovery) => Promise<void>;
};

export class ChattoRealtimeClient {
  #socket?: WebSocket;
  #stopped = false;
  #reconnectAttempt = 0;
  #reconnectDelay?: number;
  #resumeCursor?: string;
  #lastFrameAt = 0;
  #heartbeatStallMs = 75_000;
  #watchdog?: number;
  #processing = Promise.resolve();
  #seenEventIds = new Set<string>();
  #connectionAbort?: AbortController;
  #snapshotInProgress = false;

  constructor(
    private readonly serverUrl: string,
    private readonly bearerToken: string,
    private readonly onEvent: (event: RealtimeEvent) => void | Promise<void>,
    private readonly options: ChattoRealtimeClientOptions = {},
  ) {}

  start(): void {
    this.#stopped = false;
    this.#connect();
  }

  stop(): void {
    this.#stopped = true;
    window.clearInterval(this.#watchdog);
    this.#connectionAbort?.abort();
    this.#socket?.close(1000, "client stopped");
  }

  #connect(): void {
    if (this.#stopped) return;
    this.#connectionAbort?.abort();
    const connectionAbort = new AbortController();
    this.#connectionAbort = connectionAbort;
    const url = new URL("/api/realtime", this.serverUrl);
    url.protocol = url.protocol === "https:" ? "wss:" : "ws:";

    const socket = new WebSocket(url);
    socket.binaryType = "arraybuffer";
    this.#socket = socket;
    this.#reconnectDelay = undefined;

    socket.onopen = () => {
      if (this.#stopped || this.#socket !== socket) return;
      this.#lastFrameAt = Date.now();
      this.#startWatchdog();
      this.#send(
        new RealtimeSubscribe({
          protocolVersion: PROTOCOL_VERSION,
          bearerToken: this.bearerToken,
          initialState:
            this.options.initialState ?? RealtimeInitialState.LIVE_ONLY,
          resumeCursor: this.#resumeCursor,
        }),
      );
    };

    socket.onmessage = (message) => {
      if (this.#stopped || this.#socket !== socket) return;
      // Keep event handling ordered even when one handler is asynchronous.
      this.#processing = this.#processing
        .then(async () => {
          connectionAbort.signal.throwIfAborted();
          const bytes = await this.#binaryBytes(message.data);
          const frame = RealtimeServerFrame.fromBinary(bytes);
          this.#lastFrameAt = Date.now();
          await this.#handleFrame(frame, connectionAbort.signal);
        })
        .catch((error) => {
          console.error("Realtime frame rejected", error);
          connectionAbort.abort();
          socket.close(4000, "unsupported realtime frame");
        });
    };

    socket.onclose = () => {
      if (this.#socket !== socket) return;
      connectionAbort.abort();
      this.#socket = undefined;
      window.clearInterval(this.#watchdog);
      if (this.#stopped) return;

      const delay =
        this.#reconnectDelay ??
        Math.min(
          BASE_RECONNECT_MS * 2 ** this.#reconnectAttempt,
          MAX_RECONNECT_MS,
        );
      this.#reconnectAttempt += 1;
      window.setTimeout(() => this.#connect(), delay);
    };
  }

  async #handleFrame(
    frame: RealtimeServerFrame,
    connectionSignal: AbortSignal,
  ): Promise<void> {
    switch (frame.frame.case) {
      case "snapshot": {
        if (!this.options.applySnapshot) throw new Error("Unexpected snapshot");
        // A snapshot starts a new local projection generation. Do not keep a
        // cursor until caught_up confirms the handoff boundary.
        this.#snapshotInProgress = true;
        this.#seenEventIds.clear();
        this.#resumeCursor = undefined;
        await this.options.beginSnapshot?.();
        connectionSignal.throwIfAborted();
        await this.options.applySnapshot(frame.frame.value);
        connectionSignal.throwIfAborted();
        return;
      }

      case "event": {
        const event = frame.frame.value;
        const eventId = event.id;
        if (eventId && !this.#seenEventIds.has(eventId)) {
          await this.onEvent(event);
          connectionSignal.throwIfAborted();
          this.#seenEventIds.add(eventId);
        }
        // Save the cursor only after the complete handler succeeds. An unknown
        // additive event variant can be ignored and still advance this cursor.
        if (event.cursor !== undefined) {
          this.#resumeCursor = event.cursor;
        }
        return;
      }

      case "caughtUp": {
        const { cursor, recovery } = frame.frame.value;
        await this.options.completeCatchUp?.(cursor, connectionSignal, recovery);
        connectionSignal.throwIfAborted();
        this.#snapshotInProgress = false;
        this.#resumeCursor = cursor;
        this.#reconnectAttempt = 0;
        return;
      }

      case "heartbeat": {
        const cursor = frame.frame.value.cursor;
        if (cursor !== undefined) this.#resumeCursor = cursor;
        return;
      }

      case "close": {
        const close = frame.frame.value;
        if (close.code === RealtimeCloseCode.RESYNC_REQUIRED) {
          this.#resumeCursor = undefined;
        }
        this.#stopped = !close.reconnect;
        this.#reconnectDelay =
          durationMs(close.retryAfter) || BASE_RECONNECT_MS;
        this.#socket?.close(1000, "server requested reconnect");
        return;
      }

      case undefined:
        throw new Error("Unknown realtime server frame");
    }
  }

  #startWatchdog(): void {
    window.clearInterval(this.#watchdog);
    this.#watchdog = window.setInterval(() => {
      if (Date.now() - this.#lastFrameAt >= this.#heartbeatStallMs) {
        this.#socket?.close(4001, "heartbeat stalled");
      }
    }, 5_000);
  }

  #send(message: RealtimeSubscribe): void {
    if (this.#socket?.readyState !== WebSocket.OPEN) {
      throw new Error("Realtime socket is not open");
    }
    this.#socket.send(Uint8Array.from(message.toBinary()).buffer);
  }

  async #binaryBytes(data: unknown): Promise<Uint8Array> {
    if (data instanceof ArrayBuffer) return new Uint8Array(data);
    if (data instanceof Blob) return new Uint8Array(await data.arrayBuffer());
    throw new Error("Expected a binary protobuf frame");
  }
}

Handle the events that your bot needs

Get the access token from a secure runtime source. Do not put it in the WebSocket URL or in logs.

import { ChattoRealtimeClient } from "./realtime.js";

const client = new ChattoRealtimeClient(
  "https://chat.example.com",
  accessToken,
  async (event) => {
    switch (event.event.case) {
      case "messagePosted":
        console.info("New message ID", event.id);
        break;
      case "reactionAdded":
        console.info("Reaction", event.event.value.emoji);
        break;
      default:
        // Ignore event variants that this integration does not use.
        break;
    }
  },
);

client.start();

The in-memory event-ID set only shows the deduplication rule. A bot that can restart must store completed event IDs with its effects. A checkpoint does not make a post exactly-once: the server can accept a post before the response or checkpoint is lost. completeCatchUp receives the recovery result. Record LIVE_ONLY after a resume request if missed triggers matter to you.

Build a projection client

A projection client selects SNAPSHOT and passes snapshot callbacks:

import { RealtimeInitialState } from "./gen/chatto/realtime/v1/realtime_pb.js";

const projectionClient = new ChattoRealtimeClient(
  "https://chat.example.com",
  accessToken,
  applyOrReconcileEvent,
  {
    initialState: RealtimeInitialState.SNAPSHOT,
    beginSnapshot: () => resetProjection(),
    applySnapshot: (snapshot) => {
      if (!snapshot.server) throw new Error("Snapshot has no server profile");
      replaceServerProfile(snapshot.server);
      replaceRooms(snapshot.rooms);
      replaceRoomGroups(snapshot.roomGroups);
      replaceReferencedUsers(snapshot.users);
      replaceActiveCalls(snapshot.activeCalls);
    },
    completeCatchUp: async (cursor, signal) => {
      const headers = { "Chatto-Realtime-Minimum-Cursor": cursor };
      const [runtime, viewer, notifications] = await Promise.all([
        serverClient.getRuntimeConfig(
          {},
          { headers, signal, timeoutMs: 10_000 },
        ),
        viewerClient.getViewer({}, { headers, signal, timeoutMs: 10_000 }),
        notificationClient.listNotificationOccurrences(
          {},
          { headers, signal, timeoutMs: 10_000 },
        ),
      ]);
      signal.throwIfAborted();
      replaceAuxiliaryState({ runtime, viewer, notifications });
    },
  },
);

projectionClient.start();

The example uses generated ConnectRPC clients and your own replacement functions. Load users that later events reference with bounded BatchGetUsers calls. The example reads runtime settings, the viewer, and notifications after every caught_up, with the cursor as a minimum boundary. A response can include changes after the cursor, so make reducers idempotent, or read and replace the affected resource.