Socket.IO (Node.js)

upstream chat_nodes {
    hash $remote_addr consistent;          # sticky by client address; a cookie-based scheme is also possible
    server chat-a:8110;
    server chat-b:8111;
}

server {
    listen 443 ssl;
    server_name chat.example.test;

    location /socket.io/ {
        proxy_pass http://chat_nodes;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";
        proxy_set_header Host $host;
        proxy_read_timeout 90s;            # longer than pingInterval + pingTimeout (45 s)
    }
}

{
  "name": "chatterbox-socket-io",
  "version": "1.0.0",
  "private": true,
  "type": "module",
  "engines": { "node": ">=24" },
  "scripts": {
    "start": "node server.mjs"
  },
  "dependencies": {
    "@socket.io/redis-adapter": "8.3.0",
    "@socket.io/redis-streams-adapter": "0.3.1",
    "redis": "6.3.0",
    "socket.io": "4.8.4"
  },
  "devDependencies": {
    "socket.io-client": "4.8.4"
  }
}

// Chatterbox v1 on Socket.IO 4.8: every envelope travels as one "cb" event.
import { createServer } from 'node:http';
import { randomBytes } from 'node:crypto';
import { createClient } from 'redis';
import { Server } from 'socket.io';
import { createAdapter } from '@socket.io/redis-adapter';
import { createAdapter as createStreamsAdapter } from '@socket.io/redis-streams-adapter';

const PORT = Number(process.env.PORT ?? 8110);
const REDIS_URL = process.env.REDIS_URL ?? 'redis://localhost:6310';
const ADAPTER = process.env.ADAPTER ?? 'pubsub'; // pubsub | streams
const INSTANCE = process.env.INSTANCE ?? `node-${PORT}`;
const ORIGINS = (process.env.ALLOWED_ORIGINS ?? 'https://chat.example.test').split(',');
const HEARTBEAT_MS = 20000;
const MAX_FRAME = 16384;
const ID_RE = /^[0-9A-Za-z_]{26,36}$/;
const ROOM_RE = /^[a-z0-9-]{1,64}$/;
const TICKET_RE = /^example-ticket-([a-z0-9]{1,32})$/; // placeholder: real tickets are one-time and signed

const db = createClient({ url: REDIS_URL });
await db.connect();
const pub = db.duplicate();
const sub = pub.duplicate();
await Promise.all([pub.connect(), sub.connect()]);

const httpServer = createServer((req, res) => {
  if (req.url === '/healthz') { res.end(`ok ${INSTANCE}\n`); return; }
  res.writeHead(404).end();
});

const io = new Server(httpServer, {
  maxHttpBufferSize: MAX_FRAME,           // bytes per packet; larger closes the socket
  pingInterval: 25000,
  pingTimeout: 20000,
  transports: ['websocket'],              // no long-polling, so no sticky sessions
  allowRequest: (req, cb) => {            // Origin allow-list (CSWSH defence)
    const origin = req.headers.origin;
    cb(null, origin === undefined || ORIGINS.includes(origin));
  },
  adapter: ADAPTER === 'streams' ? createStreamsAdapter(pub) : createAdapter(pub, sub),
  connectionStateRecovery: { maxDisconnectionDuration: 120_000 },
});

const sid = () => 's_' + randomBytes(13).toString('hex').toUpperCase();
const send = (target, type, room, payload, id = sid()) =>
  target.emit('cb', { v: 1, type, id, room, ts: Date.now(), payload });
const ack = (socket, req, seq, dup, result = {}) =>
  send(socket, 'ack', req.room, { ref: req.id, ok: true, seq, dup, result });
const closeAfter = (socket) => setTimeout(() => socket.disconnect(true), 50);
const fail = (socket, ref, code, message, retryable = false) =>
  send(socket, 'error', null, { ref, code, message, retryable });

const roomKey = (r) => `cb:room:${r}`;           // sorted set: score = seq
const seqKey = (r) => `cb:seq:${r}`;
const ackKey = (u, id) => `cb:ack:${u}:${id}`;   // original ack, 24 h
const sessKey = (s) => `cb:session:${s}`;        // rooms + lastSeq, 120 s

async function members(room) {
  const sockets = await io.in(room).fetchSockets();   // works across instances
  const seen = new Map();
  for (const s of sockets) {
    const u = s.data.user;
    if (u) seen.set(u.userId, { userId: u.userId, name: u.name, status: s.data.status ?? 'online' });
  }
  return [...seen.values()];
}

async function replay(socket, room, since) {
  const rows = await db.zRangeByScore(roomKey(room), `(${since}`, '+inf');
  for (const row of rows) send(socket, 'message', room, JSON.parse(row).payload, JSON.parse(row).id);
}

async function saveSession(socket) {
  const { sessionId } = socket.data;
  if (!sessionId) return;
  await db.set(sessKey(sessionId), JSON.stringify({ userId: socket.data.user.userId, rooms: [...socket.data.rooms] }), { EX: 120 });
}

const helloTimers = new WeakMap();   // timers must not live in socket.data: recovery serializes it

const handlers = {
  async hello(socket, env) {
    if (socket.data.user) return fail(socket, env.id, 'conflict', 'hello already completed');
    const m = TICKET_RE.exec(env.payload.token ?? '');
    if (!m) { fail(socket, env.id, 'unauthorized', 'invalid ticket'); return closeAfter(socket); }
    clearTimeout(helloTimers.get(socket));
    const user = { userId: `u_${m[1]}`, name: m[1] };
    socket.data.user = user;
    socket.data.rooms = new Set();
    let resumed = false;
    const resume = env.payload.resume;
    if (resume?.sessionId) {
      const saved = JSON.parse((await db.get(sessKey(resume.sessionId))) ?? 'null');
      if (saved?.userId === user.userId) {
        resumed = true;
        socket.data.sessionId = resume.sessionId;
        for (const room of saved.rooms) {
          socket.join(room); socket.data.rooms.add(room);
          await replay(socket, room, resume.rooms?.[room] ?? 0);
        }
      }
    }
    socket.data.sessionId ??= 'sess_' + randomBytes(8).toString('hex');
    await saveSession(socket);
    send(socket, 'hello', null, { ...user, ref: env.id, sessionId: socket.data.sessionId,
      heartbeatMs: HEARTBEAT_MS, maxFrameBytes: MAX_FRAME, resumed, rooms: [...socket.data.rooms], instance: INSTANCE });
  },

  async join(socket, env) {
    if (!ROOM_RE.test(env.room ?? '')) return fail(socket, env.id, 'bad_request', 'invalid room');
    const dm = /^dm-([a-z0-9]+)-([a-z0-9]+)$/.exec(env.room);
    if (dm && ![dm[1], dm[2]].includes(socket.data.user.name)) return fail(socket, env.id, 'forbidden', 'not your DM');
    const already = socket.data.rooms.has(env.room);
    socket.join(env.room); socket.data.rooms.add(env.room);
    await saveSession(socket);
    if (env.payload.since != null) await replay(socket, env.room, env.payload.since);
    if (!already) {
      const u = socket.data.user;
      send(socket.to(env.room), 'presence', env.room, { userId: u.userId, name: u.name, status: 'online' });
    }
    const lastSeq = Number((await db.get(seqKey(env.room))) ?? 0);
    ack(socket, env, null, false, { members: await members(env.room), lastSeq });
  },

  async leave(socket, env) {
    if (!socket.data.rooms.delete(env.room)) return fail(socket, env.id, 'not_found', 'not a member');
    socket.leave(env.room);
    await saveSession(socket);
    ack(socket, env, null, false);
  },

  async message(socket, env) {
    const { text, replyTo = null, attachments = [] } = env.payload;
    if (typeof text !== 'string' || text.length < 1 || text.length > 4000) return fail(socket, env.id, 'bad_request', 'text must be 1..4000 characters');
    if (!socket.data.rooms.has(env.room)) return fail(socket, env.id, 'forbidden', 'join the room first');
    const u = socket.data.user;
    // Idempotency: reserve the id, keep the original ack for 24 h.
    const key = ackKey(u.userId, env.id);
    const prior = await db.get(key);
    if (prior) { const a = JSON.parse(prior); return ack(socket, env, a.seq, true, a.result); }
    const seq = await db.incr(seqKey(env.room));
    const payload = { from: { userId: u.userId, name: u.name }, text, replyTo, attachments, seq, edited: false };
    await db.zAdd(roomKey(env.room), { score: seq, value: JSON.stringify({ id: env.id, payload }) });
    await db.set(key, JSON.stringify({ seq, result: {} }), { EX: 86400 });
    ack(socket, env, seq, false);
    send(io.to(env.room), 'message', env.room, payload, env.id);   // includes the sender, across instances
  },

  async presence(socket, env) {
    const status = env.payload.status;
    if (!['online', 'away'].includes(status)) return fail(socket, env.id, 'bad_request', 'status must be online or away');
    const changed = socket.data.status !== status;
    socket.data.status = status;
    const u = socket.data.user;
    if (changed) for (const room of socket.data.rooms) send(socket.to(room), 'presence', room, { userId: u.userId, name: u.name, status });
    ack(socket, env, null, false);
  },

  async history(socket, env) {
    if (!socket.data.rooms.has(env.room)) return fail(socket, env.id, 'forbidden', 'join the room first');
    const limit = Math.min(Math.max(Number(env.payload.limit ?? 50), 1), 100);
    const max = env.payload.before == null ? '+inf' : `(${env.payload.before}`;
    const rows = await db.zRangeByScore(roomKey(env.room), '-inf', max, { LIMIT: { offset: 0, count: 100000 } });
    const page = rows.slice(-limit).map((r) => { const m = JSON.parse(r); return { v: 1, type: 'message', id: m.id, room: env.room, ts: Date.now(), payload: m.payload }; });
    const hasMore = rows.length > page.length;
    ack(socket, env, null, false, { messages: page, hasMore, next: hasMore ? page[0].payload.seq : null });
  },

  async typing(socket, env) {
    const u = socket.data.user;
    if (socket.data.rooms.has(env.room)) send(socket.to(env.room), 'typing', env.room, { userId: u.userId, state: env.payload.state });
  },
};

io.on('connection', (socket) => {
  if (socket.recovered) {
    // Connection state recovery restored id, rooms and data: no new hello is needed.
    console.log(`[${INSTANCE}] recovered ${socket.id} rooms=${[...socket.rooms].length - 1}`);
  } else {
    helloTimers.set(socket, setTimeout(() => {
      fail(socket, null, 'unauthorized', 'hello not received within 5 s');
      closeAfter(socket);
    }, 5000));
  }

  socket.on('cb', async (env, callback) => {
    try {
      if (env === null || typeof env !== 'object' || env.v !== 1) {
        if (env?.v !== undefined && env.v !== 1) return fail(socket, env.id ?? null, 'unsupported_version', 'v must be 1');
        return fail(socket, null, 'bad_request', 'not an envelope');
      }
      if (typeof env.id !== 'string' || !ID_RE.test(env.id) || env.payload === null || typeof env.payload !== 'object') {
        return fail(socket, typeof env.id === 'string' ? env.id : null, 'bad_request', 'bad id or payload');
      }
      if (env.type !== 'hello' && !socket.data.user) { fail(socket, env.id, 'unauthorized', 'send hello first'); return closeAfter(socket); }
      const handler = handlers[env.type];
      if (!handler) return fail(socket, env.id, 'bad_request', `type ${env.type} not supported by this example`);
      await handler(socket, env);
    } catch (err) {
      console.error(err);
      fail(socket, env?.id ?? null, 'internal', 'internal error', true);
    }
  });

  socket.on('disconnect', (reason) => {
    clearTimeout(helloTimers.get(socket));
    console.log(`[${INSTANCE}] disconnect ${socket.id}: ${reason}`);
  });
});

// A small namespace with its own middleware, as a second logical channel.
io.of('/metrics').use((socket, next) => (socket.handshake.auth?.key === 'change-me' ? next() : next(new Error('unauthorized'))))
  .on('connection', (socket) => socket.emit('stats', { instance: INSTANCE, localSockets: io.of('/').sockets.size }));

httpServer.listen(PORT, () => console.log(`[${INSTANCE}] listening on ${PORT}, adapter=${ADAPTER}`));
for (const sig of ['SIGINT', 'SIGTERM']) process.on(sig, () => { io.close(() => process.exit(0)); });

// Tiny Chatterbox client for Socket.IO: promise per request id, inbox of broadcasts.
import { io } from 'socket.io-client';

const ulid = () => (Date.now().toString(36) + Math.random().toString(36).slice(2) + 'ABCDEFGHJKLMNPQRSTUV').toUpperCase().replace(/[^0-9A-Z]/g, '').padEnd(26, '0').slice(0, 26);

export class ChatterboxClient {
  constructor(url, name, opts = {}) {
    this.name = name;
    this.pending = new Map();
    this.inbox = [];
    this.lastSeq = {};
    this.socket = io(url, { transports: ['websocket'], extraHeaders: { Origin: 'https://chat.example.test' }, ...opts });
    this.socket.on('cb', (env) => this.#onEnvelope(env));
  }
  #onEnvelope(env) {
    if (env.type === 'ack' || env.type === 'error') {
      const p = this.pending.get(env.payload.ref);
      if (p) { this.pending.delete(env.payload.ref); env.type === 'ack' ? p.resolve(env.payload) : p.reject(Object.assign(new Error(env.payload.message), env.payload)); }
      return;
    }
    if (env.payload.seq && env.room) this.lastSeq[env.room] = Math.max(this.lastSeq[env.room] ?? 0, env.payload.seq);
    this.inbox.push(env);
  }
  request(type, room, payload = {}, id = ulid()) {
    const env = { v: 1, type, id, room, ts: 0, payload };
    return new Promise((resolve, reject) => {
      const timer = setTimeout(() => { this.pending.delete(id); reject(new Error('ack timeout')); }, 10000);
      this.pending.set(id, { resolve: (v) => { clearTimeout(timer); resolve(v); }, reject: (e) => { clearTimeout(timer); reject(e); } });
      this.socket.emit('cb', env);
    });
  }
  async hello(resume) {
    const id = ulid();
    const reply = new Promise((resolve) => {
      const on = (env) => { if (env.type === 'hello' && env.payload.ref === id) { this.socket.off('cb', on); resolve(env.payload); } };
      this.socket.on('cb', on);
    });
    this.socket.emit('cb', { v: 1, type: 'hello', id, room: null, ts: 0, payload: { token: `example-ticket-${this.name}`, client: 'node/1.0', ...(resume && { resume }) } });
    return (this.session = await reply);
  }
  waitFor(pred, ms = 3000) {
    return new Promise((resolve, reject) => {
      const t0 = Date.now();
      const tick = () => { const e = this.inbox.find(pred); if (e) return resolve(e); if (Date.now() - t0 > ms) return reject(new Error('waitFor timeout')); setTimeout(tick, 20); };
      tick();
    });
  }
  close() { this.socket.close(); }
}
export { ulid };

FROM node:24-alpine
WORKDIR /app
COPY package.json ./
RUN npm install --omit=dev
COPY server.mjs ./
USER node
EXPOSE 8110
CMD ["node", "server.mjs"]

name: chatterbox-socket-io
services:
  redis:
    image: redis:8.8.3-alpine
    command: ["redis-server", "--save", "", "--appendonly", "no"]
    ports:
      - "6310:6379"
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 2s
      timeout: 2s
      retries: 15

  chat-a:
    build: .
    environment: &app-env
      PORT: "8110"
      INSTANCE: chat-a
      REDIS_URL: redis://redis:6379
      ADAPTER: ${ADAPTER:-pubsub}          # pubsub or streams
      ALLOWED_ORIGINS: https://chat.example.test
    ports:
      - "8110:8110"
    depends_on:
      redis:
        condition: service_healthy

  chat-b:
    build: .
    environment:
      <<: *app-env
      PORT: "8111"
      INSTANCE: chat-b
    ports:
      - "8111:8111"
    depends_on:
      redis:
        condition: service_healthy

docker compose up -d --build           # Redis on 6310, instances on 8110 and 8111, pub/sub adapter
node scenario.mjs                      # drives both instances (below)
ADAPTER=streams docker compose up -d --force-recreate    # same stack with the Redis Streams adapter
node recovery.mjs http://localhost:8110                  # connection state recovery (below)
docker compose down -v

import assert from 'node:assert/strict';
import { ChatterboxClient, ulid } from './client.mjs';

const A = process.env.URL_A ?? 'http://localhost:8110';
const B = process.env.URL_B ?? 'http://localhost:8111';
const room = 'general-' + Math.random().toString(36).slice(2, 8);

const ada = new ChatterboxClient(A, 'ada');
const bob = new ChatterboxClient(B, 'bob');
const [ha, hb] = [await ada.hello(), await bob.hello()];
console.log('hello ada on', ha.instance, '| bob on', hb.instance);
assert.notEqual(ha.instance, hb.instance);

const ja = await ada.request('join', room, { since: null });
const jb = await bob.request('join', room, { since: null });
assert.deepEqual(jb.result.members.map((m) => m.name).sort(), ['ada', 'bob']);
await ada.waitFor((e) => e.type === 'presence' && e.payload.name === 'bob');
console.log('join + presence across instances: ok');

const id = ulid();
const a1 = await ada.request('message', room, { text: 'Hello from instance A', replyTo: null, attachments: [] }, id);
const got = await bob.waitFor((e) => e.type === 'message' && e.id === id);
assert.equal(got.payload.text, 'Hello from instance A');
assert.equal(got.payload.seq, a1.seq);
await ada.waitFor((e) => e.type === 'message' && e.id === id);   // sender gets its own copy
console.log('broadcast across instances: ok, seq', a1.seq);

const dup = await ada.request('message', room, { text: 'Hello from instance A', replyTo: null, attachments: [] }, id);
assert.equal(dup.dup, true); assert.equal(dup.seq, a1.seq);
await new Promise((r) => setTimeout(r, 200));
assert.equal(bob.inbox.filter((e) => e.type === 'message' && e.id === id).length, 1);
console.log('duplicate id -> dup:true, no second broadcast: ok');

await assert.rejects(ada.request('message', room, { text: '' }), { code: 'bad_request' });
await assert.rejects(ada.request('join', 'dm-bob-carol', { since: null }), { code: 'forbidden' });
await assert.rejects(ada.request('message', 'unjoined-room', { text: 'x' }), { code: 'forbidden' });
console.log('errors bad_request / forbidden: ok');

const b2 = await bob.request('message', room, { text: 'Reply from B' });
const hist = await ada.request('history', room, { before: null, limit: 1 });
assert.equal(hist.result.messages.length, 1);
assert.equal(hist.result.messages[0].payload.seq, b2.seq);
assert.equal(hist.result.hasMore, true);
const hist2 = await ada.request('history', room, { before: hist.result.next, limit: 10 });
assert.equal(hist2.result.hasMore, false);
console.log('history pagination: ok');

await bob.request('presence', null, { status: 'away' });
await ada.waitFor((e) => e.type === 'presence' && e.payload.status === 'away');
console.log('presence update across instances: ok');

// Resume: bob drops, ada talks, bob comes back with a new socket and the same session.
const sess = bob.session.sessionId, last = bob.lastSeq[room];
bob.close();
const a3 = await ada.request('message', room, { text: 'sent while bob was away' });
const bob2 = new ChatterboxClient(B, 'bob');
const hb2 = await bob2.hello({ sessionId: sess, rooms: { [room]: last } });
assert.equal(hb2.resumed, true);
await bob2.waitFor((e) => e.type === 'message' && e.payload.seq === a3.seq);
console.log('hello resume replays missed message: ok');

// Errors that need a raw socket.
const raw = new ChatterboxClient(A, 'eve', { autoConnect: true });
await new Promise((r) => raw.socket.once('connect', r));
const bad = new Promise((r) => raw.socket.once('cb', r));
raw.socket.emit('cb', { v: 1, type: 'join', id: ulid(), room, ts: 0, payload: {} });
const e1 = await bad;
assert.equal(e1.payload.code, 'unauthorized');
console.log('frame before hello -> unauthorized + disconnect: ok');

ada.close(); bob2.close(); raw.close();
process.exit(0);

hello ada on chat-a | bob on chat-b
join + presence across instances: ok
broadcast across instances: ok, seq 1
duplicate id -> dup:true, no second broadcast: ok
errors bad_request / forbidden: ok
history pagination: ok
presence update across instances: ok
hello resume replays missed message: ok
frame before hello -> unauthorized + disconnect: ok

// Connection state recovery: drop the transport, miss a message, reconnect.
import { ChatterboxClient } from './client.mjs';

const url = process.argv[2] ?? 'http://localhost:8112';
const room = 'recovery-demo';
const ada = new ChatterboxClient(url, 'ada', { reconnectionDelay: 300, reconnectionDelayMax: 300 });
const bob = new ChatterboxClient(url, 'bob');
await ada.hello(); await bob.hello();
await ada.request('join', room, { since: null });
await bob.request('join', room, { since: null });
await bob.request('message', room, { text: 'before the drop' });
await ada.waitFor((e) => e.type === 'message');

const idBefore = ada.socket.id;
ada.socket.io.engine.close();                       // simulate a network drop (not socket.disconnect())
await bob.request('message', room, { text: 'sent during the drop' });
await new Promise((r) => ada.socket.once('connect', r));
console.log('recovered =', ada.socket.recovered, '| same id =', ada.socket.id === idBefore);
if (ada.socket.recovered) {
  const missed = await ada.waitFor((e) => e.payload.text === 'sent during the drop');
  console.log('missed event delivered after reconnect, seq', missed.payload.seq);
}
ada.close(); bob.close(); process.exit(0);

# ADAPTER=streams (Redis Streams adapter)
recovered = true | same id = true
missed event delivered after reconnect, seq 2

# ADAPTER=pubsub (Redis adapter): the session cannot be restored
recovered = false | same id = false