Realtime in React & Next.js

browser ──ws──▶ realtime server (Bun / Socket.IO / Durable Object) ◀──POST /publish── Server Action
   ▲                     │ fan-out to room                                               ▲
   └──── initial HTML ── Server Component (DB read) ──────────────── form submit ────────┘

import { z } from "zod";
 
export const ChatMsg = z.object({
  id: z.string(),
  room: z.string(),
  from: z.string(),
  text: z.string(),
  ts: z.number(),
});
export type ChatMsg = z.infer<typeof ChatMsg>;
 
export const ServerMsg = z.discriminatedUnion("type", [
  z.object({ type: z.literal("chat"), msg: ChatMsg }),
  z.object({ type: z.literal("ack"), id: z.string(),
    msg: ChatMsg }),
  z.object({ type: z.literal("nack"), id: z.string(),
    error: z.string() }),
  z.object({ type: z.literal("history"), room: z.string(),
    messages: z.array(ChatMsg) }),
]);
export type ServerMsg = z.infer<typeof ServerMsg>;
 
export type ClientMsg =
  | { type: "join" | "leave"; room: string }
  | { type: "chat"; id: string; room: string; text: string };

import ReconnectingWebSocket, {
  type UrlProvider,
} from "partysocket/ws";
import { type ChatMsg, type ClientMsg, ServerMsg }
  from "./protocol";
 
export type Status = "connecting" | "open" | "closed";
export type Snapshot = Readonly<{
  status: Status;
  retries: number;
  rooms: Readonly<Record<string, readonly ChatMsg[]>>;
}>;
type Waiter = PromiseWithResolvers<ChatMsg>;
 
const INITIAL: Snapshot = {
  status: "connecting", retries: 0, rooms: {},
};
 
// One socket per tab, shared by every component.
export function createRealtime(url: UrlProvider) {
  let state = INITIAL;
  let ws: ReconnectingWebSocket | undefined;
  let idle: ReturnType<typeof setTimeout> | undefined;
  const listeners = new Set<() => void>();
  const joined = new Map<string, number>(); // refcounts
  const waiting = new Map<string, Waiter>();
 
  const set = (patch: Partial<Snapshot>) => {
    state = { ...state, ...patch }; // new object: re-render
    for (const l of listeners) l();
  };
  const send = (msg: ClientMsg) => {
    if (ws?.readyState !== WebSocket.OPEN) return false;
    ws.send(JSON.stringify(msg));
    return true;
  };
  const append = (room: string, list: ChatMsg[]) => {
    const old = state.rooms[room] ?? [];
    const ids = new Set(old.map((m) => m.id));
    const fresh = list.filter((m) => !ids.has(m.id));
    const next = [...old, ...fresh];
    set({ rooms: { ...state.rooms, [room]: next } });
  };
 
  const parse = (raw: unknown) => {
    try {
      const r = ServerMsg.safeParse(JSON.parse(String(raw)));
      return r.success ? r.data : null;
    } catch {
      return null; // not JSON
    }
  };
 
  const onMessage = (e: MessageEvent) => {
    const m = parse(e.data);
    if (!m) return;
    if (m.type === "chat") append(m.msg.room, [m.msg]);
    if (m.type === "history") append(m.room, m.messages);
    if (m.type === "ack" || m.type === "nack") {
      const w = waiting.get(m.id);
      waiting.delete(m.id);
      if (m.type === "ack") {
        append(m.msg.room, [m.msg]);
        w?.resolve(m.msg);
      } else w?.reject(new Error(m.error));
    }
  };
 
  const connect = () => {
    ws = new ReconnectingWebSocket(url, [], {
      // partysocket has no jitter: add some
      minReconnectionDelay: 1000 + Math.random() * 4000,
      maxReconnectionDelay: 30_000,
    });
    ws.addEventListener("open", () => {
      set({ status: "open", retries: 0 });
      for (const room of joined.keys()) {
        send({ type: "join", room }); // server forgot us
      }
    });
    ws.addEventListener("close", () => {
      const retries = ws?.retryCount ?? 0;
      set({ status: "closed", retries });
    });
    ws.addEventListener("message", onMessage);
  };
 
  return {
    getSnapshot: () => state,
    getServerSnapshot: () => INITIAL, // SSR: never connect
    subscribe(listener: () => void) {
      listeners.add(listener);
      clearTimeout(idle);
      if (!ws) connect(); // lazily, on first subscriber
      return () => {
        listeners.delete(listener);
        if (listeners.size > 0) return;
        // delay: StrictMode and route changes resubscribe
        idle = setTimeout(() => {
          ws?.close();
          ws = undefined;
          state = INITIAL;
        }, 2000);
      };
    },
    join(room: string): () => void {
      const n = joined.get(room) ?? 0;
      joined.set(room, n + 1);
      if (n === 0) send({ type: "join", room });
      return () => {
        const left = (joined.get(room) ?? 1) - 1;
        if (left > 0) return void joined.set(room, left);
        joined.delete(room);
        send({ type: "leave", room });
      };
    },
    reconnect(): void { // e.g. when the tab wakes up
      if (ws && ws.readyState !== WebSocket.OPEN) {
        ws.reconnect();
      }
    },
    // resolves with the server's copy once acked
    say(room: string, text: string, timeoutMs = 8000) {
      const id = crypto.randomUUID();
      const w: Waiter = Promise.withResolvers();
      if (!send({ type: "chat", id, room, text })) {
        w.reject(new Error("offline"));
        return w.promise;
      }
      waiting.set(id, w);
      const t = setTimeout(() => {
        waiting.delete(id);
        w.reject(new Error("timed out"));
      }, timeoutMs);
      return w.promise.finally(() => clearTimeout(t));
    },
  };
}
 
const WS_URL = process.env.NEXT_PUBLIC_REALTIME_URL;
export const realtime = createRealtime(
  WS_URL ?? "ws://localhost:3001",
);

"use client";
import { useEffect, useSyncExternalStore } from "react";
import type { ChatMsg } from "./protocol";
import { realtime, type Snapshot } from "./store";
 
// select must return part of the snapshot, not a new
// object or array: that would re-render forever
export function useRealtime<T>(select: (s: Snapshot) => T) {
  return useSyncExternalStore(
    realtime.subscribe,
    () => select(realtime.getSnapshot()),
    () => select(realtime.getServerSnapshot()),
  );
}
 
const NONE: readonly ChatMsg[] = []; // stable empty value
 
export function useRoom(room: string): readonly ChatMsg[] {
  useEffect(() => realtime.join(room), [room]); // cleanup
  return useRealtime((s) => s.rooms[room] ?? NONE);
}

"use client";
import { useRealtime } from "@/lib/realtime/hooks";
 
const LABEL = {
  connecting: "Connecting…",
  open: "Live",
  closed: "Reconnecting…",
} as const;
 
export function ConnectionStatus() {
  const status = useRealtime((s) => s.status);
  const retries = useRealtime((s) => s.retries);
  const offline = status === "closed" && retries > 3;
  return (
    <p role="status" aria-live="polite" data-status={status}>
      {offline ? "Offline: sends paused" : LABEL[status]}
    </p>
  );
}

"use client";
import { useOptimistic, useState } from "react";
import { useRoom } from "@/lib/realtime/hooks";
import type { ChatMsg } from "@/lib/realtime/protocol";
import { realtime } from "@/lib/realtime/store";
 
type Row = ChatMsg & { pending?: true };
 
export function Chat({ room }: { room: string }) {
  const messages = useRoom(room);
  const [error, setError] = useState<string>();
  const [rows, addPending] = useOptimistic(
    messages as readonly Row[],
    (rows, text: string): readonly Row[] => [
      ...rows,
      { id: `tmp-${rows.length}`, room, from: "you", text,
        ts: Date.now(), pending: true },
    ],
  );
 
  // a form action runs in a transition: the pending row
  // shows until it returns, then `messages` takes over
  async function send(form: FormData) {
    const text = String(form.get("text") ?? "").trim();
    if (!text) return;
    addPending(text);
    setError(undefined);
    try {
      await realtime.say(room, text); // store adds it
    } catch (err) {
      const why = (err as Error).message;
      setError(`"${text}" (${why})`); // the input was reset
    }
  }
 
  return (
    <section>
      <ul>
        {rows.map((m) => (
          <li key={m.id} data-pending={m.pending}>
            <b>{m.from}</b> {m.text}
          </li>
        ))}
      </ul>
      {error && <p role="alert">Not sent: {error}</p>}
      <form action={send}>
        <input name="text" maxLength={2000} />
        <button>Send</button>
      </form>
    </section>
  );
}

"use server";
import { z } from "zod";
import { getSession } from "@/lib/auth";
import { db } from "@/lib/db";
 
const Input = z.object({
  room: z.string().min(1).max(64),
  text: z.string().trim().min(1).max(2000),
});
 
export async function postMessage(
  input: z.input<typeof Input>,
) {
  const session = await getSession();
  if (!session) throw new Error("Unauthorized");
  const { room, text } = Input.parse(input);
  if (!(await db.canPost(session.userId, room))) {
    throw new Error("Forbidden");
  }
  const msg = await db.insertMessage({ // persist first
    room, text, from: session.name,
  });
  const url = `${process.env.REALTIME_HTTP}/publish`;
  const res = await fetch(url, {
    method: "POST",
    headers: {
      authorization: `Bearer ${process.env.REALTIME_SECRET}`,
      "content-type": "application/json",
    },
    body: JSON.stringify({ room, msg }),
  });
  // saved but not broadcast: clients catch up on resync
  if (!res.ok) console.error("publish failed", res.status);
  return msg;
}

import { z } from "zod";
import { ChatMsg } from "./lib/realtime/protocol";
 
const SECRET = process.env.REALTIME_SECRET;
if (!SECRET) throw new Error("REALTIME_SECRET not set");
const Publish = z.object({ room: z.string(), msg: ChatMsg });
 
const server = Bun.serve({
  port: 3001,
  async fetch(req, server) {
    const { pathname } = new URL(req.url);
    if (req.method === "POST" && pathname === "/publish") {
      const auth = req.headers.get("authorization");
      if (auth !== `Bearer ${SECRET}`) {
        return new Response(null, { status: 401 });
      }
      const body = Publish.safeParse(await req.json());
      if (!body.success) {
        return new Response(null, { status: 400 });
      }
      const { room, msg } = body.data;
      const out = JSON.stringify({ type: "chat", msg });
      server.publish(room, out);
      return new Response(null, { status: 204 });
    }
    // ...verify a ticket, then server.upgrade(req, { data })
    return new Response("Not found", { status: 404 });
  },
  websocket: {
    // join/leave: authorize, then ws.subscribe(room)
    message() {},
  },
});

bun add @vercel/functions ws
bun add -d @types/ws

import {
  experimental_upgradeWebSocket,
  type WebSocketData,
} from "@vercel/functions";
import { connection } from "next/server";
 
export async function GET() {
  await connection(); // Cache Components: request time only
  return experimental_upgradeWebSocket(
    (ws) => {
      ws.on("message", (data: WebSocketData) => {
        ws.send(data); // echo; real app: validate, fan out
      });
    },
    { maxPayload: 64 * 1024 }, // default 256 KiB
  );
}

const enc = new TextEncoder();
const algo = { name: "HMAC", hash: "SHA-256" };
 
const key = (secret: string) =>
  crypto.subtle.importKey(
    "raw", enc.encode(secret), algo, false,
    ["sign", "verify"],
  );
 
// userId must not contain "."
export async function signTicket(
  userId: string, secret: string, ttlS = 30,
): Promise<string> {
  const exp = Math.floor(Date.now() / 1000) + ttlS;
  const body = `${userId}.${exp}`;
  const k = await key(secret);
  const sig = await crypto.subtle.sign(
    "HMAC", k, enc.encode(body),
  );
  return `${body}.${Buffer.from(sig).toString("base64url")}`;
}
 
export async function verifyTicket(
  ticket: string, secret: string,
): Promise<string | null> {
  const [userId, exp, sig] = ticket.split(".");
  if (!userId || !exp || !sig) return null;
  if (Number(exp) < Date.now() / 1000) return null; // old
  const ok = await crypto.subtle.verify(
    "HMAC", await key(secret),
    Buffer.from(sig, "base64url"),
    enc.encode(`${userId}.${exp}`),
  );
  return ok ? userId : null;
}

import { getSession } from "@/lib/auth";
import { signTicket } from "@/lib/ticket";
 
export async function GET() {
  const session = await getSession(); // cookies: dynamic
  if (!session) return new Response(null, { status: 401 });
  const secret = process.env.REALTIME_SECRET;
  if (!secret) throw new Error("REALTIME_SECRET not set");
  const ticket = await signTicket(session.userId, secret);
  return Response.json(
    { ticket },
    { headers: { "cache-control": "no-store" } },
  );
}

// store.ts, last line: a fresh ticket per (re)connect
const WS = process.env.NEXT_PUBLIC_REALTIME_URL;
export const realtime = createRealtime(async () => {
  const res = await fetch("/api/realtime-ticket");
  const { ticket } = await res.json() as { ticket: string };
  return `${WS}?ticket=${encodeURIComponent(ticket)}`;
});

"use client";
import { useEffect } from "react";
import { realtime } from "@/lib/realtime/store";
 
export function WakeReconnect() {
  useEffect(() => {
    const wake = () => {
      if (document.visibilityState === "visible") {
        realtime.reconnect();
      }
    };
    window.addEventListener("online", wake);
    document.addEventListener("visibilitychange", wake);
    return () => {
      window.removeEventListener("online", wake);
      document.removeEventListener("visibilitychange", wake);
    };
  }, []);
  return null;
}

import { Suspense } from "react";
import {
  ConnectionStatus,
} from "@/components/connection-status";
import { Chat } from "./chat";
 
export default function Page(
  props: PageProps<"/rooms/[id]">,
) {
  return (
    <main>
      <ConnectionStatus />
      <Suspense fallback={<p>Loading room…</p>}>
        {props.params.then(({ id }) => (
          <Chat key={id} room={id} />
        ))}
      </Suspense>
    </main>
  );
}

"use client";
import { useSyncExternalStore } from "react";
import { io } from "socket.io-client";
 
export const socket = io(process.env.NEXT_PUBLIC_IO_URL, {
  autoConnect: false, // connect on first subscriber
  transports: ["websocket"],
});
 
function subscribe(onChange: () => void) {
  socket.on("connect", onChange);
  socket.on("disconnect", onChange);
  if (!socket.connected) socket.connect();
  return () => {
    socket.off("connect", onChange);
    socket.off("disconnect", onChange);
  };
}
 
export function useConnected(): boolean {
  return useSyncExternalStore(
    subscribe,
    () => socket.connected,
    () => false, // server render: not connected
  );
}