Realtime updates

realtime:
  driver: broadcast
  channels:
    order: { action: orders.read }

"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" });
}

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

import { realtime } from "mainsdk";

export const GET = realtime.handler();

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();

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 });
}

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) });

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));

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));
  }
});

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" });

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);

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();
}

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 });
});

import { realtime } from "mainsdk";

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

realtime:
  driver: local   # was broadcast; one instance only

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 }));