Socket.IO

export type ChatMsg = {
  id: string; // client-made UUID
  room: string;
  from: string;
  text: string;
  ts: number;
};
export type Res<T> =
  | { ok: true; data: T }
  | { ok: false; error: string };
 
// server → client
export interface ServerToClientEvents {
  message: (msg: ChatMsg) => void;
  joined: (room: string, user: string) => void;
}
 
// client → server; the last function arg is the ack
export interface ClientToServerEvents {
  join: (
    room: string,
    ack: (r: Res<ChatMsg[]>) => void,
  ) => void;
  send: (
    msg: Pick<ChatMsg, "id" | "room" | "text">,
    ack: (r: Res<{ ts: number }>) => void,
  ) => void;
  leave: (room: string) => void;
}
 
// server ↔ server, through the adapter
export interface InterServerEvents {
  kick: (userId: string) => void;
}
 
// per-socket state: socket.data
export interface SocketData {
  userId: string;
  name: string;
}

bun add socket.io @socket.io/bun-engine
bun add socket.io-client   # in the client app

import { Server as Engine } from "@socket.io/bun-engine";
import { Server } from "socket.io";
import { verifyToken } from "./auth.ts";
import type {
  ChatMsg,
  ClientToServerEvents as C2S,
  InterServerEvents as S2S,
  ServerToClientEvents as S2C,
  SocketData,
} from "./events.ts";
 
const io = new Server<C2S, S2C, S2S, SocketData>({
  connectionStateRecovery: {
    maxDisconnectionDuration: 2 * 60_000, // default
    skipMiddlewares: true, // recovered: skip auth
  },
});
const engine = new Engine({
  path: "/socket.io/", // default is /engine.io/
  cors: { origin: ["http://localhost:5173"] },
});
io.bind(engine);
 
// runs once per connection, before "connection"
io.use(async (socket, next) => {
  const token: unknown = socket.handshake.auth.token;
  const user =
    typeof token === "string" && (await verifyToken(token));
  if (!user) return next(new Error("unauthorized"));
  socket.data = user;
  next();
});
 
const history = new Map<string, ChatMsg[]>();
 
io.on("connection", (socket) => {
  // a recovered socket already has its rooms and data;
  // handlers must still be attached to every new socket
  if (!socket.recovered) {
    socket.join(`user:${socket.data.userId}`);
  }
 
  socket.on("join", async (room, ack) => {
    await socket.join(room);
    socket.to(room).emit("joined", room, socket.data.name);
    ack({ ok: true, data: history.get(room) ?? [] });
  });
 
  socket.on("send", ({ id, room, text }, ack) => {
    if (!socket.rooms.has(room)) {
      return ack({ ok: false, error: "join first" });
    }
    const msg: ChatMsg = {
      id, room, text, from: socket.data.name, ts: Date.now(),
    };
    history.set(room, [...(history.get(room) ?? []), msg]);
    socket.to(room).emit("message", msg); // not to sender
    ack({ ok: true, data: { ts: msg.ts } });
  });
 
  socket.on("leave", (room) => socket.leave(room));
});
 
Bun.serve({ port: 3000, ...engine.handler() });

import { createServer } from "node:http";
import { Server } from "socket.io";
import type {
  ClientToServerEvents as C2S,
  InterServerEvents as S2S,
  ServerToClientEvents as S2C,
  SocketData,
} from "./events.ts";
 
const http = createServer(); // or your Express/Hono app
const io = new Server<C2S, S2C, S2S, SocketData>(http, {
  cors: { origin: ["http://localhost:5173"] },
  connectionStateRecovery: {},
});
// same io.use(...) and io.on("connection") as above
http.listen(3000);

import { io, type Socket } from "socket.io-client";
import type {
  ClientToServerEvents as C2S,
  ServerToClientEvents as S2C,
} from "./events.ts";
 
export type ChatSocket = Socket<S2C, C2S>; // reversed
 
export function connect(token: string): ChatSocket {
  const socket: ChatSocket = io("http://localhost:3000", {
    auth: { token }, // or (cb) => cb({ token: fresh() })
    transports: ["websocket"], // no polling, no sticky
  });
 
  socket.on("connect", () => {
    // recovered: rooms kept and missed events replayed
    if (!socket.recovered) {
      // fresh session: rejoin rooms, refetch state
    }
  });
  socket.on("connect_error", (err) => {
    // active: will retry; else refused by middleware
    if (!socket.active) console.error(err.message);
  });
  socket.on("disconnect", (reason) => {
    if (reason === "io server disconnect") {
      // kicked by the server: no auto-reconnect
    }
  });
  return socket;
}

const admin = io.of("/admin");
admin.use(requireAdmin); // namespace middleware
admin.on("connection", (socket) => {
  socket.emit("stats", getStats());
});
 
// dynamic: one namespace per workspace
const teams = io.of(/^\/team-\d+$/);
teams.on("connection", (socket) => {
  const nsp = socket.nsp; // e.g. /team-42
  nsp.emit("online", socket.data.userId);
});

const res = await socket
  .timeout(5000)
  .emitWithAck("send", {
    id: crypto.randomUUID(),
    room: "lobby",
    text: "hi",
  }); // res: Res<{ ts: number }>, typed from events.ts
if (!res.ok) showError(res.error);

import {
  createAdapter,
} from "@socket.io/redis-streams-adapter";
import { createClient } from "redis";
import { Server } from "socket.io";
 
const redis = createClient({ url: process.env.REDIS_URL });
await redis.connect();
 
const io = new Server({
  adapter: createAdapter(redis, {
    streamName: "chat", // default "socket.io"
    maxLen: 10_000, // entries kept: bounds recovery
  }),
  connectionStateRecovery: {},
});
 
// every node: the same calls now span the cluster
io.to("lobby").emit("hello"); // reaches all nodes
const all = await io.in("lobby").fetchSockets();
io.serverSideEmit("kick", "u42"); // other nodes only
io.on("kick", (userId) => {
  io.in(`user:${userId}`).disconnectSockets(true);
});

type Data = { userId: string };
 
const server = Bun.serve({
  port: 3001,
  fetch(req, server) {
    const userId = crypto.randomUUID(); // real: session
    if (server.upgrade(req, { data: { userId } })) return;
    return new Response("Upgrade required", { status: 426 });
  },
  websocket: {
    data: {} as Data,
    open(ws) {
      ws.subscribe("lobby");
      const n = server.subscriberCount("lobby");
      server.publish("lobby", `${n} online`); // to all
    },
    message(ws, msg) {
      ws.publish("lobby", String(msg)); // all but ws
    },
    close(ws) {
      // ws is already unsubscribed here
      server.publish("lobby", `${ws.data.userId} left`);
    },
  },
});

import type { Server } from "socket.io";
 
// token bucket: `rate` per second, bursts up to `burst`
function bucket(rate: number, burst: number) {
  let tokens = burst;
  let last = performance.now();
  return () => {
    const now = performance.now();
    const refill = ((now - last) / 1000) * rate;
    tokens = Math.min(burst, tokens + refill);
    last = now;
    if (tokens < 1) return false;
    tokens -= 1;
    return true;
  };
}
 
export function rateLimit(io: Server, rate = 5, burst = 10) {
  io.on("connection", (socket) => {
    const take = bucket(rate, burst);
    let strikes = 0;
    socket.use(([event], next) => {
      if (take()) return next();
      next(new Error(`rate limited: ${String(event)}`));
    });
    // next(err) is emitted here; with no listener it throws
    socket.on("error", (err) => {
      socket.emit("limited", err.message);
      if (++strikes > 20) socket.disconnect(true);
    });
  });
}

import { Server } from "socket.io";
 
interface S2C {
  "ready?": (ack: (ok: boolean) => void) => void;
}
const io = new Server<{}, S2C>();
 
export async function readyCheck(room: string) {
  try {
    const answers = await io
      .to(room)
      .timeout(5000)
      .emitWithAck("ready?"); // boolean[]: one per socket
    return answers.every(Boolean);
  } catch {
    return false; // someone did not answer in time
  }
}

import { afterAll, expect, test } from "bun:test";
import { Server as Engine } from "@socket.io/bun-engine";
import { Server } from "socket.io";
import {
  io as connect,
  type Socket,
} from "socket.io-client";
 
const io = new Server();
const engine = new Engine({ path: "/socket.io/" });
io.bind(engine);
io.on("connection", (socket) => {
  socket.join("t");
  socket.on("say", (text: string, ack: () => void) => {
    socket.to("t").emit("said", text);
    ack();
  });
});
const http = Bun.serve({ port: 0, ...engine.handler() });
 
const opts = { transports: ["websocket"] };
const url = `http://localhost:${http.port}`;
const a = connect(url, opts);
const b = connect(url, opts);
 
afterAll(() => {
  a.close();
  b.close();
  io.close();
  void http.stop(true);
});
 
const once = <T>(s: Socket, ev: string) =>
  new Promise<T>((r) => s.once(ev, r));
 
test("b hears what a says, a gets an ack", async () => {
  await Promise.all([
    once(a, "connect"),
    once(b, "connect"),
  ]);
  const heard = once<string>(b, "said");
  await a.timeout(1000).emitWithAck("say", "hi");
  expect(await heard).toBe("hi");
});

export async function kick(userId: string) {
  const room = `user:${userId}`;
  io.to(room).emit("kicked"); // tell the UI why
  io.in(room).disconnectSockets(true); // reason on client:
  // "io server disconnect": it will not auto-reconnect
}