The Browser WebSocket API

const socket = new WebSocket("wss://chat.example.test/ws", ["chatterbox.v1"]);

const socket = new WebSocket("wss://chat.example.test/ws", "chatterbox.v1");
socket.binaryType = "arraybuffer";

socket.addEventListener("open", () => {
  console.log("negotiated", socket.protocol, socket.extensions);
});
socket.addEventListener("message", (event) => {
  // event.data is a string for text frames, an ArrayBuffer or Blob for binary frames
  console.log(typeof event.data, event.origin);
});
socket.addEventListener("error", () => {
  // deliberately empty of detail, see below
});
socket.addEventListener("close", (event) => {
  console.log(event.code, event.reason, event.wasClean);
});

socket.send(JSON.stringify({ v: 1, type: "typing", id: crypto.randomUUID(), room: "general", ts: 0, payload: { state: "start" } }));
socket.send(new Uint8Array([1, 2, 3]));              // binary frame
socket.send(new Blob(["raw bytes"]));                // binary frame
socket.send(new DataView(new ArrayBuffer(8)));       // binary frame

// Sends a large upload in chunks without letting the socket's send buffer grow without bound.

export interface BackpressureOptions {
  highWaterMark?: number; // stop sending above this many buffered bytes
  lowWaterMark?: number;  // resume once the buffer drains below this
  pollMs?: number;        // there is no "drain" event, so poll bufferedAmount
}

const sleep = (ms: number) => new Promise<void>((r) => setTimeout(r, ms));

export async function sendWithBackpressure(
  ws: WebSocket,
  chunks: Iterable<ArrayBufferView | ArrayBuffer>,
  { highWaterMark = 1 << 20, lowWaterMark = 256 << 10, pollMs = 10 }: BackpressureOptions = {},
): Promise<{ sent: number; maxBuffered: number }> {
  let sent = 0;
  let maxBuffered = 0;
  for (const chunk of chunks) {
    if (ws.readyState !== WebSocket.OPEN) throw new Error(`socket not open (${ws.readyState})`);
    if (ws.bufferedAmount > highWaterMark) {
      while (ws.bufferedAmount > lowWaterMark) {
        if (ws.readyState !== WebSocket.OPEN) throw new Error("socket closed while draining");
        await sleep(pollMs);
      }
    }
    ws.send(chunk as ArrayBuffer);
    sent += chunk.byteLength;
    maxBuffered = Math.max(maxBuffered, ws.bufferedAmount);
  }
  return { sent, maxBuffered };
}

socket.close(4000, "client shutting down");

// Ticket flow: ask the HTTP API for a short-lived ticket, then present it in the first frame.
const response = await fetch("https://chat.example.test/api/ws-ticket", { method: "POST", credentials: "include" });
const { ticket } = await response.json();
const socket = new WebSocket("wss://chat.example.test/ws", "chatterbox.v1");
socket.addEventListener("open", () => {
  socket.send(JSON.stringify({ v: 1, type: "hello", id: crypto.randomUUID(), room: null, ts: 0, payload: { token: ticket, client: "web/1.4" } }));
});

const wss = new WebSocketStream("wss://chat.example.test/ws", { protocols: ["chatterbox.v1"] });
const { readable, writable, protocol } = await wss.opened;   // rejects if the handshake fails
const writer = writable.getWriter();
const reader = readable.getReader();

await writer.write(JSON.stringify({ v: 1, type: "hello", id: crypto.randomUUID(), room: null, ts: 0, payload: { client: "web/1.4" } }));
for (;;) {
  const { value, done } = await reader.read();   // the browser stops reading from the network while you do not call read()
  if (done) break;
  console.log(protocol, value);
}
const { closeCode, reason } = await wss.closed;  // resolves for a clean close, rejects otherwise

// One small interface over WebSocketStream (when present) and WebSocket (everywhere else).

interface WebSocketStreamLike {
  opened: Promise<{
    readable: ReadableStream<string | ArrayBuffer>;
    writable: WritableStream<string | ArrayBuffer | ArrayBufferView>;
    protocol: string;
  }>;
  closed: Promise<{ closeCode: number; reason: string }>;
  close(info?: { closeCode?: number; reason?: string }): void;
}
type WebSocketStreamCtor = new (url: string, options?: { protocols?: string[] }) => WebSocketStreamLike;

export interface Channel {
  readonly kind: "stream" | "socket";
  send(data: string | ArrayBuffer): Promise<void>; // resolves when it is safe to send more
  messages: AsyncIterable<string | ArrayBuffer>;
  close(code?: number, reason?: string): void;
}

export async function openChannel(url: string, protocol: string): Promise<Channel> {
  const Ctor = (globalThis as { WebSocketStream?: WebSocketStreamCtor }).WebSocketStream;
  if (typeof Ctor === "function") {
    const wss = new Ctor(url, { protocols: [protocol] });
    const { readable, writable } = await wss.opened;   // rejects if the handshake fails
    const writer = writable.getWriter();
    return {
      kind: "stream",
      send: (data) => writer.write(data),             // the promise is the backpressure signal
      messages: readable as unknown as AsyncIterable<string | ArrayBuffer>,
      close: (code, reason) => wss.close({ closeCode: code, reason }),
    };
  }

  const ws = new WebSocket(url, protocol);
  ws.binaryType = "arraybuffer";
  await new Promise<void>((resolve, reject) => {
    ws.addEventListener("open", () => resolve(), { once: true });
    ws.addEventListener("error", () => reject(new Error("handshake failed")), { once: true });
  });
  const queue: (string | ArrayBuffer)[] = [];
  let wake: (() => void) | null = null;
  let done = false;
  ws.addEventListener("message", (e) => { queue.push(e.data as string | ArrayBuffer); wake?.(); });
  ws.addEventListener("close", () => { done = true; wake?.(); });
  async function* iterate() {
    while (true) {
      const next = queue.shift();
      if (next !== undefined) { yield next; continue; }
      if (done) return;
      await new Promise<void>((r) => (wake = r));
    }
  }
  return {
    kind: "socket",
    async send(data) {
      ws.send(data);
      while (ws.bufferedAmount > 1 << 20) await new Promise((r) => setTimeout(r, 10));
    },
    messages: iterate(),
    close: (code, reason) => ws.close(code, reason),
  };
}

let socket = null;
function connect() {
  if (socket && socket.readyState <= WebSocket.OPEN) return;   // already connecting or open
  socket = new WebSocket("wss://chat.example.test/ws", "chatterbox.v1");
}
addEventListener("pagehide", () => socket?.close(1000, "page hidden"));
addEventListener("pageshow", (event) => { if (event.persisted) connect(); });
connect();

// Typed client for the Chatterbox v1 protocol (hello handshake, client ids, ack tracking).

export type ClientType =
  | "hello" | "join" | "leave" | "message" | "edit" | "react"
  | "typing" | "receipt" | "presence" | "history";

export interface Envelope<T extends string, P> {
  v: 1;
  type: T;
  id: string;
  room: string | null;
  ts: number;
  payload: P;
}

export interface Attachment { url: string; name: string; mime: string; size: number }
export interface Member { userId: string; name: string; status: string }

// Payloads the client sends, keyed by envelope type.
export interface ClientPayloads {
  hello: { token?: string; client: string };
  join: { since: number | null };
  leave: Record<string, never>;
  message: { text: string; replyTo: string | null; attachments: Attachment[] };
  presence: { status: "online" | "away" };
  typing: { state: "start" | "stop" };
  receipt: { kind: "delivered" | "read"; upTo: number };
}

// What an ack carries, keyed by the type of the request it answers.
export interface AckResults {
  hello: Record<string, never>;
  join: { members: Member[]; lastSeq: number };
  leave: Record<string, never>;
  message: Record<string, never>;
  presence: Record<string, never>;
}

export interface HelloReply {
  userId: string; name: string; sessionId: string; heartbeatMs: number;
  maxFrameBytes: number; resumed: boolean; rooms: string[]; ref: string;
}
export interface AckPayload { ref: string; ok: true; seq: number | null; dup: boolean; result: unknown }
export interface ErrorPayload { ref: string | null; code: string; message: string; retryable: boolean }
export interface MessageBroadcast {
  from: { userId: string; name: string }; text: string; replyTo: string | null;
  attachments: Attachment[]; seq: number; edited: boolean;
}

// Discriminated union of everything the server can send.
export type ServerEnvelope =
  | Envelope<"hello", HelloReply>
  | Envelope<"ack", AckPayload>
  | Envelope<"error", ErrorPayload>
  | Envelope<"message", MessageBroadcast>
  | Envelope<"presence", { userId: string; name: string; status: "online" | "away" | "offline" }>
  | Envelope<"typing", { userId: string; state: "start" | "stop" }>;

export type EventMap = {
  message: Envelope<"message", MessageBroadcast>;
  presence: Extract<ServerEnvelope, { type: "presence" }>;
  typing: Extract<ServerEnvelope, { type: "typing" }>;
  close: CloseEvent;
};

export class ChatterboxError extends Error {
  constructor(readonly code: string, message: string, readonly retryable: boolean) {
    super(message);
    this.name = "ChatterboxError";
  }
}
export class AckTimeoutError extends Error {
  constructor(readonly envelopeId: string) {
    super(`No ack for ${envelopeId}`);
    this.name = "AckTimeoutError";
  }
}
export class ConnectionClosedError extends Error {
  constructor(readonly code: number, readonly reason: string) {
    super(`Connection closed (${code})`);
    this.name = "ConnectionClosedError";
  }
}

export interface ChatterboxOptions {
  url: string;
  token?: string;
  clientName?: string;
  ackTimeoutMs?: number;       // protocol default: 10 000 ms
  WebSocketImpl?: typeof WebSocket;
}

interface Pending {
  resolve: (ack: AckPayload) => void;
  reject: (error: Error) => void;
  timer: ReturnType<typeof setTimeout>;
}

const CROCKFORD = "0123456789ABCDEFGHJKMNPQRSTVWXYZ";

/** A ULID: 48-bit time plus 80 random bits, 26 Crockford base32 characters. */
export function newId(now: number = Date.now()): string {
  let time = "";
  for (let n = now, i = 0; i < 10; i++, n = Math.floor(n / 32)) time = CROCKFORD[n % 32] + time;
  const bytes = crypto.getRandomValues(new Uint8Array(16));
  let random = "";
  for (const b of bytes) random += CROCKFORD[b % 32];
  return time + random;
}

export class ChatterboxClient {
  readonly #opts: Required<Omit<ChatterboxOptions, "token">> & { token?: string };
  #ws: WebSocket | null = null;
  readonly #pending = new Map<string, Pending>();
  readonly #listeners = new Map<keyof EventMap, Set<(e: never) => void>>();
  hello: HelloReply | null = null;

  constructor(opts: ChatterboxOptions) {
    this.#opts = {
      clientName: "web/1.4",
      ackTimeoutMs: 10_000,
      WebSocketImpl: globalThis.WebSocket,
      ...opts,
    };
  }

  on<K extends keyof EventMap>(event: K, fn: (e: EventMap[K]) => void): () => void {
    const set = this.#listeners.get(event) ?? new Set<(e: never) => void>();
    this.#listeners.set(event, set);
    set.add(fn);
    return () => set.delete(fn);
  }

  #emit<K extends keyof EventMap>(event: K, e: EventMap[K]): void {
    this.#listeners.get(event)?.forEach((fn) => (fn as (e: EventMap[K]) => void)(e));
  }

  /** Opens the socket, sends hello as the first frame and resolves with the server's hello reply. */
  connect(): Promise<HelloReply> {
    return new Promise((resolve, reject) => {
      const ws = new this.#opts.WebSocketImpl(this.#opts.url, "chatterbox.v1");
      this.#ws = ws;
      let settled = false;
      const helloId = newId();

      ws.addEventListener("open", () => {
        if (ws.protocol !== "chatterbox.v1") {
          ws.close(4400, "subprotocol not negotiated");
          return;
        }
        const payload: ClientPayloads["hello"] = { client: this.#opts.clientName };
        if (this.#opts.token !== undefined) payload.token = this.#opts.token;
        this.#sendEnvelope("hello", null, payload, helloId);
      });

      ws.addEventListener("message", (event) => {
        if (typeof event.data !== "string") return; // protocol v1 is text only
        const env = JSON.parse(event.data) as ServerEnvelope;
        if (env.type === "hello" && !settled) {
          settled = true;
          this.hello = env.payload;
          resolve(env.payload);
          return;
        }
        this.#dispatch(env);
      });

      ws.addEventListener("close", (event) => {
        if (!settled) {
          settled = true;
          reject(new ConnectionClosedError(event.code, event.reason));
        }
        const error = new ConnectionClosedError(event.code, event.reason);
        for (const [id, p] of this.#pending) {
          clearTimeout(p.timer);
          p.reject(error);
          this.#pending.delete(id);
        }
        this.#emit("close", event);
      });
    });
  }

  #dispatch(env: ServerEnvelope): void {
    switch (env.type) {
      case "ack":
      case "error": {
        const ref = env.payload.ref;
        const pending = ref ? this.#pending.get(ref) : undefined;
        if (!pending || ref === null) return;
        clearTimeout(pending.timer);
        this.#pending.delete(ref);
        if (env.type === "ack") pending.resolve(env.payload);
        else pending.reject(new ChatterboxError(env.payload.code, env.payload.message, env.payload.retryable));
        return;
      }
      case "message": this.#emit("message", env); return;
      case "presence": this.#emit("presence", env); return;
      case "typing": this.#emit("typing", env); return;
      case "hello": return;
    }
  }

  #sendEnvelope<K extends keyof ClientPayloads>(
    type: K, room: string | null, payload: ClientPayloads[K], id: string = newId(),
  ): string {
    const ws = this.#ws;
    if (ws === null || ws.readyState !== ws.OPEN) throw new ConnectionClosedError(1006, "not open");
    const env: Envelope<K, ClientPayloads[K]> = { v: 1, type, id, room, ts: 0, payload };
    ws.send(JSON.stringify(env));
    return id;
  }

  /** Sends a request envelope and resolves with its ack, rejects on error, timeout or close. */
  request<K extends keyof AckResults & keyof ClientPayloads>(
    type: K, room: string | null, payload: ClientPayloads[K], id: string = newId(),
  ): Promise<AckPayload & { result: AckResults[K] }> {
    return new Promise((resolve, reject) => {
      const timer = setTimeout(() => {
        this.#pending.delete(id);
        reject(new AckTimeoutError(id));
      }, this.#opts.ackTimeoutMs);
      this.#pending.set(id, { resolve: resolve as Pending["resolve"], reject, timer });
      try {
        this.#sendEnvelope(type, room, payload, id);
      } catch (e) {
        clearTimeout(timer);
        this.#pending.delete(id);
        reject(e as Error);
      }
    });
  }

  join(room: string, since: number | null = null) {
    return this.request("join", room, { since });
  }

  /** Same id on retry makes the server answer with dup: true instead of storing the message twice. */
  sendMessage(room: string, text: string, id: string = newId()) {
    return this.request("message", room, { text, replyTo: null, attachments: [] }, id);
  }

  /** Fire-and-forget: no ack is expected for typing. */
  typing(room: string, state: "start" | "stop"): void {
    this.#sendEnvelope("typing", room, { state });
  }

  close(code = 1000, reason = "bye"): void {
    this.#ws?.close(code, reason);
  }
}