Realtime fundamentals

type Conn = { user: string; seen: number };
 
// Presence per user, tracked per connection: a user
// with two tabs open stays online until both close.
export class Presence {
  #conns = new Map<string, Conn>();
 
  #count(user: string): number {
    let n = 0;
    for (const c of this.#conns.values()) {
      if (c.user === user) n++;
    }
    return n;
  }
 
  // true when this is the user's first connection
  join(conn: string, user: string): boolean {
    this.#conns.set(conn, { user, seen: Date.now() });
    return this.#count(user) === 1;
  }
 
  // the user, when their last connection went away
  leave(conn: string): string | undefined {
    const c = this.#conns.get(conn);
    this.#conns.delete(conn);
    if (c && this.#count(c.user) === 0) return c.user;
  }
 
  beat(conn: string): void {
    const c = this.#conns.get(conn);
    if (c) this.#conns.set(conn, { ...c, seen: Date.now() });
  }
 
  // run on a timer: evict silent connections
  sweep(ttlMs: number): string[] {
    const cutoff = Date.now() - ttlMs;
    const gone: string[] = [];
    for (const [id, c] of this.#conns) {
      if (c.seen >= cutoff) continue;
      const user = this.leave(id);
      if (user) gone.push(user);
    }
    return gone;
  }
 
  online(): string[] {
    const all = [...this.#conns.values()];
    return [...new Set(all.map((c) => c.user))];
  }
}

import { z } from "zod";
 
const base = {
  id: z.uuid(), // made by the client: idempotency key
  room: z.string().min(1).max(64),
  v: z.literal(1).default(1), // protocol version
};
 
export const ClientMsg = z.discriminatedUnion("type", [
  z.object({ ...base, type: z.literal("join") }),
  z.object({
    ...base,
    type: z.literal("chat"),
    text: z.string().trim().min(1).max(2000),
  }),
  z.object({
    ...base,
    type: z.literal("resume"),
    since: z.int().nonnegative(), // last seq applied
  }),
]);
export type ClientMsg = z.infer<typeof ClientMsg>;
 
// the server stamps order (seq) and time (ts)
export type Body =
  | { type: "chat"; from: string; text: string }
  | { type: "left"; user: string };
export type RoomEvent = Body & {
  id: string;
  room: string;
  seq: number;
  ts: number;
};
export type Reply =
  | { type: "ack"; id: string; seq: number }
  | { type: "nack"; id: string; code: NackCode }
  | { type: "snapshot"; room: string; seq: number;
      state: unknown };
type NackCode = "invalid" | "forbidden" | "rate_limited";
 
export function parseClient(raw: string): ClientMsg | null {
  try {
    const r = ClientMsg.safeParse(JSON.parse(raw));
    return r.success ? r.data : null;
  } catch {
    return null; // not JSON
  }
}

import type { RoomEvent } from "./envelope.ts";
 
// Applies each room event once, in seq order.
export function inbox(
  apply: (e: RoomEvent) => void,
  onGap: (last: number) => void, // send a resume
) {
  let last = 0; // highest seq applied
  const early = new Map<number, RoomEvent>();
 
  const drain = () => {
    let next = early.get(last + 1);
    while (next) {
      early.delete(next.seq);
      apply(next);
      last = next.seq;
      next = early.get(last + 1);
    }
  };
 
  return {
    get last() {
      return last;
    },
    reset(seq: number) { // after a snapshot
      last = seq;
      for (const n of early.keys()) {
        if (n <= seq) early.delete(n);
      }
      drain();
    },
    push(e: RoomEvent): void {
      if (e.seq <= last) return; // duplicate: drop
      early.set(e.seq, e);
      if (e.seq > last + 1) return onGap(last); // hole
      drain();
    },
  };
}

client                         server
  | render "pending" (id=a1)     |
  |-- { type: chat, id: a1 } --> | validate, authorize, rate limit
  |                              | persist, seq = 58
  | <-- { type: ack, id: a1,     | broadcast event seq 58 to the room
  |       seq: 58 }              |
  | swap pending a1 → event 58   |
  |                              |
  | (or) <-- { type: nack,       | rejected: roll back, show why
  |       id: a1, code }         |

import type { Body, RoomEvent } from "./envelope.ts";
 
// One per room: assigns seq, dedupes by id, keeps
// the last `max` events so reconnecting clients
// can catch up without a full snapshot.
export class RoomLog {
  #events: RoomEvent[] = [];
  #byId = new Map<string, number>(); // id -> seq
  #seq = 0;
 
  constructor(readonly room: string, readonly max = 500) {}
 
  get seq(): number {
    return this.#seq;
  }
 
  append(id: string, body: Body) {
    const dup = this.#byId.get(id);
    if (dup !== undefined) return { seq: dup, fresh: false };
    const seq = ++this.#seq;
    const event: RoomEvent = {
      ...body, id, room: this.room, seq, ts: Date.now(),
    };
    this.#events.push(event);
    this.#byId.set(id, seq);
    if (this.#events.length > this.max) {
      const old = this.#events.shift();
      if (old) this.#byId.delete(old.id);
    }
    return { seq, fresh: true, event };
  }
 
  // events after `since`, or null if they were trimmed
  since(since: number): RoomEvent[] | null {
    const first = this.#events[0]?.seq ?? this.#seq + 1;
    if (since + 1 < first) return null; // too old
    return this.#events.filter((e) => e.seq > since);
  }
}

import type { Reply } from "./envelope.ts";
import type { RoomLog } from "./room-log.ts";
 
type Conn = { send(data: string): unknown };
 
// on { type: "resume", since }: replay or snapshot
export function resume(
  conn: Conn,
  log: RoomLog,
  since: number,
  state: () => unknown, // current room state
): void {
  const missed = log.since(since);
  if (missed) {
    for (const e of missed) conn.send(JSON.stringify(e));
    return;
  }
  const snap: Reply = {
    type: "snapshot",
    room: log.room,
    seq: log.seq, // client calls inbox.reset(seq)
    state: state(),
  };
  conn.send(JSON.stringify(snap));
}

import { parseClient, type Reply } from "./envelope.ts";
import { RoomLog } from "./room-log.ts";
 
type Conn = { userId: string; send(d: string): unknown };
const logs = new Map<string, RoomLog>();
 
export function onMessage(
  conn: Conn,
  raw: string,
  broadcast: (room: string, data: string) => void,
): void {
  const msg = parseClient(raw);
  if (!msg || msg.type !== "chat") return;
  let log = logs.get(msg.room);
  if (!log) {
    log = new RoomLog(msg.room);
    logs.set(msg.room, log);
  }
  const res = log.append(msg.id, {
    type: "chat",
    from: conn.userId, // from the socket, not the message
    text: msg.text,
  });
  const ack: Reply = {
    type: "ack", id: msg.id, seq: res.seq,
  };
  conn.send(JSON.stringify(ack)); // ack duplicates too
  if (res.event) {
    broadcast(msg.room, JSON.stringify(res.event));
  }
}

// Resend until acked, always with the same id, so a
// retry after a lost ack can't create a duplicate.
type Pending = { data: string; tries: number };
 
export function reliable(
  send: (data: string) => void,
  { timeoutMs = 5000, maxTries = 5 } = {},
) {
  const pending = new Map<string, Pending>();
  type T = ReturnType<typeof setTimeout>;
  const timers = new Map<string, T>();
 
  const attempt = (id: string) => {
    const p = pending.get(id);
    if (!p) return;
    if (p.tries >= maxTries) {
      pending.delete(id); // give up: show "not sent"
      return;
    }
    pending.set(id, { ...p, tries: p.tries + 1 });
    send(p.data);
    timers.set(id, setTimeout(() => attempt(id), timeoutMs));
  };
 
  return {
    send(id: string, msg: object): void {
      const data = JSON.stringify({ ...msg, id });
      pending.set(id, { data, tries: 0 });
      attempt(id);
    },
    ack(id: string): void { // on { type: "ack", id }
      clearTimeout(timers.get(id));
      timers.delete(id);
      pending.delete(id);
    },
    resendAll(): void { // call after reconnecting
      for (const id of pending.keys()) {
        clearTimeout(timers.get(id));
        attempt(id);
      }
    },
  };
}

import { Presence } from "./presence.ts";
 
const TTL = 45_000; // > 2 client heartbeats
const rooms = new Map<string, Presence>();
 
export function startSweeper(
  broadcast: (room: string, data: string) => void,
): () => void {
  const timer = setInterval(() => {
    for (const [room, p] of rooms) {
      for (const user of p.sweep(TTL)) {
        const left = { type: "left", room, user };
        broadcast(room, JSON.stringify(left));
      }
    }
  }, 15_000);
  return () => clearInterval(timer);
}

Only server → client updates?          → SSE + POST, or a hosted provider
A few rooms, one region, < ~10k sockets → one Bun process (Bun.serve topics)
Need acks, rooms, reconnect for free   → Socket.IO (+ Redis Streams adapter at 2+ nodes)
Many independent rooms, global users   → one Durable Object per room (PartyServer)
Shared documents or canvases           → Yjs over any of the above (see Collaboration)
Fast action game                       → authoritative server loop (see Game worlds)
Realtime is a small feature            → Ably, Pusher, Supabase Realtime, Liveblocks