---
title: "Realtime updates"
description: "Push live updates to browsers over server-sent events, with every subscribe checked per channel and tenant."
section: "interfaces"
status: "Building for 0.3 · ships as a beta API"
spec: "realtime"
canonical: "https://mainsdk.dev/docs/interfaces/realtime"
updated: "2026-10-28"
---

# Realtime updates

> Push live updates to browsers over server-sent events, with every subscribe checked per channel and tenant.

Status: Building for 0.3 · ships as a beta API · Canonical: https://mainsdk.dev/docs/interfaces/realtime · Updated: 2026-10-28

`realtime.publish()` pushes live updates from your server to connected clients, and `realtime.subscribe()` lets a client listen on a channel, checked per user and tenant.

`events.broadcast()` is a different thing. It sends a message between instances of your own app, for cache invalidation, and lives in [Events and background jobs](https://mainsdk.dev/docs/interfaces/jobs).

## At a glance

- Status: Building for 0.3 · ships as a beta API
- Phase: 1 (Core, public beta)
- Milestone: 0.3 Every interface
- Plan: free in your app (Apache-2.0)
- Spec: https://github.com/hubxero/mainsdk/blob/develop/docs/specs/realtime.md
- Category: Messaging and notifications

Calls:

- `realtime.publish(channel, event, data)`
- `realtime.pipe(channel, tokens, { event? })`
- `realtime.handler()`
- `realtime.on(prefix, event, opts, fn)`
- `realtime.onStream(prefix, fn)`
- `realtime.ticket()`
- `realtime.subscribe(channels, { onEvent, onReset? })`
- `realtime.tokens(sub, channel, { onText, onDone })`
- `http.websocket(url, options)`
- `new SocketServer(hub)`
- `new Realtime(ports)`
- `parseRealtime(raw)`
- `publishProgress(hub, source)`
- `publishRecords(hub, source)`
- `new Broadcaster(opts)`
- `new MemoryBus()`
- `redisBroadcast(o)`
- `events(object: broadcast, onBroadcast)`

In main.yaml:

```yaml
realtime:
  driver: broadcast
  path: /mainsdk/realtime
  transports: [sse, websocket]
  channels:
    order: { action: orders.read }
    chat: { action: chats.read, send: chats.write }
  replay: { size: 500, ttl: 5m }
  max-event: 32KB
  streams:
    call: { types: ["audio/opus"], max-per-socket: 4, budget: 64MB }
  socket: { rate: 50/s, buffer: 1MB }
```

Guarantees, whatever the driver:

- Every subscribe is checked through the session and access, per declared channel prefix
- Channels are keyed by the current tenant
- A reader that falls behind is disconnected, never buffered without bound
- A stream resumes from Last-Event-ID, or is told to reset


## Drivers

Pick one in main.yaml with `realtime.driver`.

| Driver | Id in main.yaml | Default | Note |
| --- | --- | --- | --- |
| Instance broadcast (events.broadcast) | `broadcast` | yes |  |
| Local (one instance) | `local` |  |  |
| Pusher | `pusher` |  |  |
| Ably | `ably` |  |  |
| Supabase Realtime | `supabase-realtime` |  |  |

## Codes

| Code | Where | Meaning |
| --- | --- | --- |
| `MAIN-REALTIME-0003` | check | A channel prefix not declared under `realtime.channels` (404 on subscribe, a throw on publish; a check on a literal in code) |
| `MAIN-REALTIME-0006` | check | The `local` driver in production without `single-instance: true` |
| `MAIN-REALTIME-0007` | check | `max-event` plus the 4KB envelope is larger than `events.broadcast.max-payload` |
| `MAIN-REALTIME-0012` | check | A client message or stream refused: no `send` action on the prefix, access denied, or no handler |
| `MAIN-REALTIME-0015` | check | `realtime.transports` lists `websocket` on an adapter or platform without HTTP upgrade |
| `MAIN-REALTIME-0016` | check | A binary stream refused: prefix not under `realtime.streams`, type not listed, or `max-per-socket` reached |
| `MAIN-REALTIME-0001` | runtime | Subscribe with no session (401) |
| `MAIN-REALTIME-0002` | runtime | Subscribe refused: access denied, or a `user:` channel that isn't the caller's (403) |
| `MAIN-REALTIME-0004` | runtime | Event data is not JSON, or is larger than `max-event` |
| `MAIN-REALTIME-0005` | runtime | A channel or event name doesn't match its pattern |
| `MAIN-REALTIME-0008` | runtime | A stream was closed because its reader fell more than `buffer` behind (logged) |
| `MAIN-REALTIME-0009` | runtime | The person already has `per-user` open streams (429) |
| `MAIN-REALTIME-0010` | runtime | WebSocket upgrade refused: the `Origin` is neither the app's own nor in `inbound.cors.origins` (403) |
| `MAIN-REALTIME-0011` | runtime | WebSocket upgrade refused: the ticket is unknown, expired, used or from another tenant (401) |
| `MAIN-REALTIME-0013` | runtime | A client message isn't JSON, has an unknown type, is over `max-message` or fails the handler's schema |
| `MAIN-REALTIME-0014` | runtime | Client messages over `realtime.socket.rate` |
| `MAIN-REALTIME-0017` | runtime | A stream passed its byte `budget`, or a frame passed `max-frame`; the stream or socket is closed |
| `MAIN-REALTIME-0018` | runtime |  |

## Features

| Feature | Phase | Milestone | Spec ticket |
| --- | --- | --- | --- |
| realtime@1 contract and channel names | 1 | 0.3 | `realtime#contract` |
| Publish and the SSE endpoint on the local driver | 1 | 0.3 | `realtime#sse-local` |
| Subscribe checked per channel prefix | 1 | 0.3 | `realtime#subscribe-auth` |
| Channels keyed by tenant | 1 | 0.3 | `realtime#tenant` |
| Replay buffer and resume | 1 | 0.3 | `realtime#resume` |
| Back-pressure and per-user connection cap | 1 | 0.3 | `realtime#backpressure` |
| Fan-out across instances | 1 | 0.3 | `realtime#fanout` |
| Model token streams with pipe | 1 | 0.3 | `realtime#ai-pipe` |
| Workflow progress and record-change events | 1 | 0.3 | `realtime#progress-records` |
| Reconnect hint on shutdown | 1 | 0.3 | `realtime#shutdown` |
| Browser client | 1 | 0.3 | `realtime#client` |
| Checks for undeclared channels and oversized events | 1 | 0.3 | `realtime#checks` |
| RFC 6455 WebSocket server on node:http | 1 | 0.4 | `realtime-sockets#rfc6455` |
| Upgrade checked by origin | 1 | 0.4 | `realtime-sockets#upgrade-auth` |
| Subscribe | 1 | 0.4 | `realtime-sockets#socket-channels` |
| Client messages handled with realtime.on | 1 | 0.4 | `realtime-sockets#client-messages` |
| Numbered binary streams with realtime.onStream | 1 | 0.4 | `realtime-sockets#binary-streams` |
| Bounded socket and stream queues | 1 | 0.4 | `realtime-sockets#backpressure` |
| Outbound http.websocket with the rules of http.fetch | 1 | 0.4 | `realtime-sockets#outbound-parity` |
| Close every socket with a reconnect code on drain | 1 | 0.4 | `realtime-sockets#drain` |
| Socket transport in the browser client | 1 | 0.4 | `realtime-sockets#client` |
| Checks for sockets without upgrade and undeclared send and stream prefixes | 1 | 0.4 | `realtime-sockets#checks` |
| Real-time collaboration and presence | 5 | 1.6 | `platforms#collab-presence` |

## MainSDK Cloud

Everything on this page is free in your app.


## Quick start

```yaml title="main.yaml"
realtime:
  driver: broadcast
  channels:
    order: { action: orders.read }
```

```ts title="app/orders/actions.ts"
"use server";
import { realtime } from "mainsdk";

export async function markShipped(orderId: string) {
  // update the order in db first
  await realtime.publish(`order:${orderId}`, "status", { status: "shipped" });
}
```

## Configuration

```yaml title="main.yaml"
realtime:
  driver: broadcast            # local | broadcast; pusher, ably and supabase-realtime later
  single-instance: false       # true says local is enough in production
  path: /mainsdk/realtime
  channels:                    # the channel prefixes clients may use, and the access action each needs
    order: { action: orders.read }
    chat: { action: chats.read, send: chats.write }   # send: what a client message needs (websocket)
    workflow: { action: workflows.read }              # a run's state changes are published here
  records: [orders]            # models that publish `changed` to order:<id> after commit
  per-user: 10                 # open streams and sockets per person
  transports: [sse, websocket] # websocket is opt-in
  streams:
    call: { types: ["audio/opus"], max-per-socket: 4, budget: 64MB }
  socket: { max-message: 64KB, max-frame: 256KB, rate: 50/s, high-water: 256KB, buffer: 1MB, inbound-buffer: 1MB, ping: 20s }
  ticket-ttl: 30s
  heartbeat: 15s
  max-age: 15m                 # a stream closes with a reconnect hint, so access is checked again
  replay: { size: 500, ttl: 5m }
  buffer: 256KB                # a reader that falls further behind is disconnected
  max-event: 32KB
```

| Key | What it does |
| --- | --- |
| `driver` | `local` serves one instance. `broadcast` fans events out to every instance through `events.broadcast` |
| `channels` | Channel prefixes you allow, such as `order` for `order:42`, with the access action a subscribe needs and, for sockets, the `send` action. A plain list of prefixes reads as `<prefix>.read`. `tenant` and `user:<id>` are always there |
| `records` | Models whose committed writes publish `changed` with `{ op, id, version }` and no field values |
| `transports` | `sse` (default) and `websocket` (opt-in). `mainsdk check` reports `MAIN-REALTIME-0015` for websocket on a platform with no HTTP upgrade |
| `streams` | Prefixes that may open binary streams, with their content types, `max-per-socket` and byte `budget` |
| `socket` | Limits of one socket: message and frame sizes, `rate` of client messages, `high-water` per stream, `buffer` per socket, `inbound-buffer` and `ping` |
| `replay` | Events kept per channel so a client can resume from `Last-Event-ID` |
| `max-event` | The largest event. A bigger one fails with `MAIN-REALTIME-0004` |

Channel names always live inside the current tenant. `user:me` is one person, and `tenant` is every member.

## What you get

- Live updates for order status that changes without a refresh, live dashboards or a notification badge.
- Server-sent events (0.3, spec `realtime`): one stream per browser carries many channels, resumes after a drop and closes when access is gone.
- `realtime.publish(channel, event, data)` sends from your server, and `realtime.subscribe(channels, options)` listens in the browser.
- Add `websocket` to `transports` (0.4, spec `realtime-sockets`) and the same channels also run over one WebSocket, with messages from the client and binary streams.
- Every subscribe is checked through the session and `access`, so a client only hears the channels it may see.

## The endpoint and the browser client

`realtime.handler()` returns the SSE endpoint as a `(req) => Response` handler. Adapters mount it at `realtime.path`.

```ts title="app/mainsdk/realtime/route.ts"
import { realtime } from "mainsdk";

export const GET = realtime.handler();
```

In the browser, `realtime.subscribe(channels, options)` opens one stream for many channels. `onReset` fires when events were missed, so you refetch.

```ts title="app/orders/live.ts"
import { realtime } from "mainsdk/realtime/client";

const sub = realtime.subscribe(["order:42", "user:me"], {
  onEvent: (e) => console.log(e.channel, e.event, e.data),
  onReset: () => void fetch("/api/orders/42"),
});
sub.add("workflow:wr_01J");
sub.close();
```

### Stream model tokens

`realtime.pipe(channel, tokens)` sends a model's tokens as indexed `token` events and a final `done` event with the full text. It takes the iterable `ai.stream` returns, unchanged. The browser reads them with `realtime.tokens()`, which waits for `done` and uses its text when an index was missed.

```ts title="app/api/chat/route.ts"
import { ai, realtime } from "mainsdk";

export async function POST(req: Request) {
  const { chatId, prompt } = (await req.json()) as { chatId: string; prompt: string };
  const { text } = await realtime.pipe(`chat:${chatId}`, ai.stream({ model: "anthropic/claude-sonnet-4-5", feature: "chat", prompt }));
  return Response.json({ text });
}
```

```ts title="app/chat/stream.ts"
import { realtime } from "mainsdk/realtime/client";

const sub = realtime.subscribe(["chat:c_7"], { onEvent: () => {} });

realtime.tokens(sub, "chat:c_7", { onText: (soFar) => console.log(soFar), onDone: (text) => console.log(text) });
```

EventSource hears events by name, so `subscribe` listens for a default set of names (`updated`, `changed`, `status`, `token`, `done` and the workflow states). Pass `events: ["my-event"]` for others.

### Workflow progress and record changes

With `realtime.channels.workflow` declared, a run publishes `{ state, currentStep, wakeAt? }` to `workflow:<runId>`: `sleeping`, `running`, `completed`, never a step result. `publishProgress(hub, source)` and `publishRecords(hub, source)` connect those two feeds to a hub through small ports, so the workflow runner and the database tell realtime when a run changes state or a transaction commits.

## WebSocket

Set `transports: [sse, websocket]` and mount the upgrade next to the SSE endpoint. The server side is RFC 6455 on `node:http` and `node:crypto`, with no dependency and no compression.

```ts title="server.ts"
import { createServer } from "node:http";
import { app } from "mainsdk";
import { SocketServer } from "mainsdk/realtime";

const server = createServer();
const sockets = new SocketServer(app().realtime);
server.on("upgrade", (req, socket, head) => void sockets.upgrade(req, socket, head));
```

Every upgrade is checked before `101`: draining (503), the `Origin` against the app's own and `inbound.cors.origins` (`MAIN-REALTIME-0010`), then a session or a single-use ticket from `realtime.ticket()` for clients with no cookies (`MAIN-REALTIME-0001`, `MAIN-REALTIME-0011`). Channels pass the same checks as SSE, and `per-user` counts sockets and streams together.

```ts title="chat.ts"
import { realtime } from "mainsdk";
import { s } from "mainsdk/schema";
import type { BinaryStream, ClientMessage } from "mainsdk/realtime";

declare function saveMessage(channel: string, user: string, data: unknown): Promise<void>;
declare function transcode(chunk: Uint8Array): Uint8Array;

// A client message needs `send` on the channel's prefix and passes access; the handler runs in the sender's tenant.
realtime.on("chat", "message", { schema: s.object({ text: s.string() }) }, async (msg: ClientMessage) => {
  await saveMessage(msg.channel, msg.user, msg.data);
  return { ok: true };       // sent back as the acknowledgement
});

// A binary stream, such as microphone audio in and speech out.
realtime.onStream("call", async (stream: BinaryStream) => {
  const out = stream.writable.getWriter();
  for await (const chunk of stream.readable) {
    await out.ready;         // waits at high-water instead of buffering
    await out.write(transcode(chunk));
  }
});
```

Client messages are rate-limited per socket through `limits` (`MAIN-REALTIME-0014`); when the limits store is down the socket counts locally and never refuses on that account. A socket whose queue passes `buffer` is closed with `4008`; one whose handler stops reading is paused at `inbound-buffer` and closed after 30 seconds.

In the browser, `transport: "auto"` tries a WebSocket and falls back to SSE:

```ts
import { realtime } from "mainsdk/realtime/client";

const sub = realtime.subscribe(["chat:c_7"], { transport: "auto", onEvent: (e) => console.log(e.data) });
await sub.send("chat:c_7", "message", { text: "hi" });          // resolves with the handler's reply
const mic = sub.stream("call:k_1", { contentType: "audio/opus" });
```

`send` and `stream` reject on SSE with a clear error.

### Outbound sockets

`http.websocket(url, { protocols?, headers?, connectTimeout?, highWater?, maxMessage? })` opens a socket to a provider with the rules of `http.fetch`: the host must be in `network.allow` (`MAIN-NET-0002`), a redirect is refused (`MAIN-NET-0003`), the connect timeout applies (`MAIN-NET-0009`) and each attempt is a decision on the network diagnostics channel. `send` waits while the socket's buffered bytes are over `highWater`, and iterating the socket yields its messages.

```ts
import { http, secrets } from "mainsdk";

const ws = await http.websocket("wss://api.elevenlabs.io/v1/text-to-speech/x/stream-input", {
  headers: { "xi-api-key": await secrets.get("ELEVENLABS_KEY") },
});
await ws.send(JSON.stringify({ text: "Hello" }));
for await (const msg of ws) console.log(msg);
```

`http.websocket(url, protocols?)` with no options still returns the browser `WebSocket` after the same host check. `mainsdk discover` lists each site with `via: websocket`.

## Examples

### Listen for order updates in the browser

```ts title="app/orders/live-status.ts"
import { realtime } from "mainsdk/realtime/client";

// Call it from your component's effect (in React, `useEffect`) and return the stop function as its cleanup.
export function watchStatus(
  orderId: string,
  setStatus: (status: string) => void,
  refetch: (orderId: string) => Promise<string>,
): () => void {
  const sub = realtime.subscribe([`order:${orderId}`], {
    onEvent: (e) => setStatus((e.data as { status: string }).status),
    onReset: () => void refetch(orderId).then(setStatus),
  });
  return () => sub.close();
}
```

### Publish from a webhook

```ts title="app/api/webhooks/carrier/route.ts"
import { realtime, webhooks } from "mainsdk";

export const POST = webhooks.receive("standard", async (event) => {
  const { orderId, status } = (event as { data: { orderId: string; status: string } }).data;
  await realtime.publish(`order:${orderId}`, "status", { status });
});
```

The handler runs only after the webhook signature and timestamp are checked; see [Webhooks](https://mainsdk.dev/docs/interfaces/webhooks).

### Update a team dashboard from a job

```ts title="jobs/metrics.ts"
import { realtime } from "mainsdk";

export async function publishDailyTotals(totals: { orders: number }) {
  await realtime.publish(`tenant`, "dashboard", totals);
}
```

## Reference

The `broadcast` driver sends each event between instances of your app through the broadcast layer of `events`. These are the parts it is built from.

| Name | What it is |
| --- | --- |
| `events(...)` | The `events` object, whose `broadcast(topic, payload)` and `onBroadcast(topic, fn)` calls carry messages between instances. The catalog lists it as `events(object: broadcast, onBroadcast)`. |
| `new Realtime(ports)` | The hub behind `realtime`: channels, publishing, replay and the SSE endpoint, over the ports the app supplies. The app builds it for you; tests and adapters build their own. |
| `parseRealtime(raw)` | Read the `realtime` section of a parsed `main.yaml`. Absent keys take their defaults, and a bad value throws with the key named. |
| `new Broadcaster(opts)` | The class behind those two calls, over one driver. `broadcast` resolves with `{ id, reached, dropped }` once the driver accepts the message or 250 ms pass, and never throws for an unreachable driver. `onBroadcast` returns a function that stops its handler, and `close()` ends the subscription. Topics are lowercase letters, digits, dots, dashes and underscores, up to 128 characters. |
| `new MemoryBus()` | A bus that several instances in one process share. It backs the `local` driver in tests and `mainsdk dev`, and `bus.sent` lists the messages sent. |
| `redisBroadcast(o)` | The `redis` driver: one `PUBLISH` per message and one `SUBSCRIBE` per instance on the channel `<prefix>:broadcast`, over the SDK's own client. Its host is a driver host of the network layer. |

Broadcast is best effort and at most once. A message is not stored or retried.

## Errors and checks

| Code | When | Fix |
| --- | --- | --- |
| `MAIN-REALTIME-0001` | Subscribe with no session (401) | Sign the person in |
| `MAIN-REALTIME-0002` | Access denied, or a `user:` channel that isn't the caller's (403) | Subscribe to a channel the person may read |
| `MAIN-REALTIME-0003` | A channel prefix not under `realtime.channels` | Declare it |
| `MAIN-REALTIME-0004` | Event data is not JSON, or over `max-event` | Send JSON under the limit |
| `MAIN-REALTIME-0005` | A channel or event name does not match its pattern | Use `<prefix>:<id>` and a lowercase event name |
| `MAIN-REALTIME-0006` | The `local` driver in production (check, runtime) | Use `broadcast`, or set `single-instance: true` |
| `MAIN-REALTIME-0007` | `max-event` plus 4KB is over `events.broadcast.max-payload` (check) | Lower `max-event` or raise the payload limit |
| `MAIN-REALTIME-0008` | A stream or socket was closed because its reader fell behind (logged) | The client resumes with its last id |
| `MAIN-REALTIME-0009` | The person already has the most open streams (429) | Close one |
| `MAIN-REALTIME-0010` | Upgrade refused: the `Origin` is not allowed (403) | Add it to `inbound.cors.origins` |
| `MAIN-REALTIME-0011` | Upgrade refused: the ticket is unknown, expired, used or from another tenant (401) | Ask `realtime.ticket()` again |
| `MAIN-REALTIME-0012` | A client message or stream refused: no `send` action, access denies it, or no handler (also a check) | Declare `send` and register `realtime.on` |
| `MAIN-REALTIME-0013` | A client message is not valid JSON, is too big or fails the handler's schema | Fix the message |
| `MAIN-REALTIME-0014` | Client messages over `socket.rate` | Wait `retryAfter` seconds |
| `MAIN-REALTIME-0015` | `websocket` listed on a platform with no HTTP upgrade (check) | Remove it or change platform |
| `MAIN-REALTIME-0016` | A binary stream refused: prefix, type or count (also a check) | Declare it under `realtime.streams` |
| `MAIN-REALTIME-0017` | A stream passed its byte `budget`, or a frame passed `max-frame` | Raise the limit or send less |
| `MAIN-REALTIME-0018` | Subscribe or upgrade while the app is draining (503) | Reconnect in a few seconds |

See [check codes](https://mainsdk.dev/docs/reference/check-codes) for the rest.

## Swapping drivers

```yaml title="main.yaml"
realtime:
  driver: local   # was broadcast; one instance only
```

`local` in production needs `single-instance: true`, or `mainsdk check` fails with `MAIN-REALTIME-0006`. Hosted drivers (Pusher, Ably, Supabase Realtime) come later. Your `realtime.publish()` calls stay the same. See [Swap a driver](https://mainsdk.dev/docs/guides/swap-a-driver).

## Testing

`realtime` runs on the `local` driver in tests with no provider account, and `app.realtime.handler()` takes plain `Request` objects, so a test opens a stream, publishes and reads the events with no network. Time comes from the fake clock: advance it 45 seconds to see three heartbeats, or past `max-age` to see the reconnect.

```ts title="app/orders/actions.test.ts"
import { createTestApp } from "@mainsdk/testing";

const cookie = "";   // the session cookie of the person who subscribes
const app = createTestApp({ main: "./main.yaml" });
const res = await app.realtime.handler()(new Request("http://app.test/mainsdk/realtime?c=user:me", { headers: { cookie } }));
await app.tenancy.anonymous(() => app.realtime.publish("user:u_1", "ping", { n: 1 }));
```

For WebSocket, drive `SocketServer.upgrade(req, socket)` with a pair of in-memory duplex sockets.

## Security and privacy

- Every subscribe is checked through the session and `access`, per declared channel prefix and the current tenant.
- Keep personal data out of messages. Send IDs and statuses, and let the client fetch details through scoped [`db`](https://mainsdk.dev/docs/interfaces/db) reads.
- A new realtime update can go to a small share of users first and pause itself on spikes in errors, through the same staged rollout as notifications.

## FAQ

**Is this for chat and presence?** Publish and subscribe ship first. Real-time collaboration and presence come in phase 5.

**Can a client publish?** Over SSE, no: clients subscribe and your server publishes. Over a WebSocket a client sends messages to a channel whose prefix declares `send`, and your `realtime.on` handler decides what to publish.

## Next

- [Notifications](https://mainsdk.dev/docs/interfaces/notify)
- [Webhooks](https://mainsdk.dev/docs/interfaces/webhooks)
- [Events and background jobs](https://mainsdk.dev/docs/interfaces/jobs)