Building a Chat App

chatterbox/
  package.json
  package-lock.json            generated by npm install
  tsconfig.json
  Dockerfile
  .dockerignore
  compose.yaml
  nginx.conf
  public/index.html
  src/
    shared/protocol.ts         envelope and payload types, shared by both halves
    server/store.ts            PostgreSQL: schema, transactions, history
    server/server.ts           HTTP, tickets, Socket.IO, handlers, lifecycle
    client/transport.ts        SocketIoTransport and WebSocketTransport
    client/chatterbox.ts       protocol client: hello, resume, retries, heartbeat
    client/main.ts             the DOM
  test/e2e.mjs                 two headless Chrome users, outage and resume

// Chatterbox wire protocol v1 types, shared by the server and the browser client.
export const PROTOCOL_VERSION = 1;
export const SUBPROTOCOL = 'chatterbox.v1';

export type MessageType =
  | 'hello' | 'join' | 'leave' | 'message' | 'edit' | 'react'
  | 'typing' | 'receipt' | 'presence' | 'history' | 'ack' | 'error';

export type ErrorCode =
  | 'bad_request' | 'unsupported_version' | 'unauthorized' | 'forbidden' | 'not_found'
  | 'conflict' | 'too_large' | 'rate_limited' | 'internal' | 'unavailable';

export type Status = 'online' | 'away' | 'offline';

export interface Envelope<P = Record<string, any>> {
  v: 1;
  type: MessageType;
  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: Status }

export interface ChatPayload {
  from: { userId: string; name: string };
  text: string;
  replyTo: string | null;
  attachments: Attachment[];
  seq: number;
  edited: boolean;
}
export interface EditPayload { target: string; text: string; seq: number; editedAt: number }
export interface ReactPayload { target: string; emoji: string; op: 'add' | 'remove'; from: { userId: string }; seq: number }

export interface HelloReply {
  ref: string; userId: string; name: string; sessionId: string; heartbeatMs: number;
  maxFrameBytes: number; resumed: boolean; rooms: string[]; instance?: string;
}
export interface AckPayload { ref: string; ok: true; seq: number | null; dup: boolean; result: Record<string, any> }
export interface ErrorPayload { ref: string | null; code: ErrorCode; message: string; retryable: boolean }
export interface HistoryResult { messages: Envelope<ChatPayload>[]; hasMore: boolean; next: number | null }

export const ROOM_RE = /^[a-z0-9-]{1,64}$/;
export const ID_RE = /^[0-9A-Za-z_]{26,36}$/;

// PostgreSQL store: the durable truth for messages, edits, reactions, sequence numbers and acks.
import pg from 'pg';
import type { Attachment, ChatPayload, ErrorCode } from '../shared/protocol.ts';

export const pool = new pg.Pool({
  connectionString: process.env.DATABASE_URL ?? 'postgres://chatterbox:change-me@localhost:5432/chatterbox',
  max: 10,
});

export class StoreError extends Error {
  code: ErrorCode;
  constructor(code: ErrorCode, message: string) {
    super(message);
    this.code = code;
  }
}

const SCHEMA = `
CREATE TABLE IF NOT EXISTS rooms (
  name     text PRIMARY KEY,
  last_seq bigint NOT NULL DEFAULT 0
);
CREATE TABLE IF NOT EXISTS events (            -- replay log: every message, edit and react
  room    text   NOT NULL,
  seq     bigint NOT NULL,
  id      text   NOT NULL,
  type    text   NOT NULL,
  author  text   NOT NULL,
  payload jsonb  NOT NULL,
  PRIMARY KEY (room, seq)
);
CREATE TABLE IF NOT EXISTS messages (          -- current state of each message (edits applied)
  id          text PRIMARY KEY,
  room        text   NOT NULL,
  seq         bigint NOT NULL,
  author      text   NOT NULL,
  author_name text   NOT NULL,
  text        text   NOT NULL,
  reply_to    text,
  attachments jsonb  NOT NULL DEFAULT '[]',
  edited_at   bigint,
  UNIQUE (room, seq)
);
CREATE TABLE IF NOT EXISTS reactions (
  message_id text NOT NULL,
  user_id    text NOT NULL,
  emoji      text NOT NULL,
  PRIMARY KEY (message_id, user_id, emoji)
);
CREATE TABLE IF NOT EXISTS acks (              -- idempotency: the original ack of every request id
  user_id    text NOT NULL,
  id         text NOT NULL,
  ack        jsonb NOT NULL,
  created_at timestamptz NOT NULL DEFAULT now(),
  PRIMARY KEY (user_id, id)
);`;

export async function migrate(): Promise<void> {
  const c = await pool.connect();
  try {
    await c.query('SELECT pg_advisory_lock(264001)');   // two replicas may start at the same time
    await c.query(SCHEMA);
    await c.query('SELECT pg_advisory_unlock(264001)');
  } finally {
    c.release();
  }
}

export interface AckRecord { seq: number | null; result: Record<string, unknown> }
export interface Author { userId: string; name: string }
type EventType = 'message' | 'edit' | 'react';

export async function findAck(userId: string, id: string): Promise<AckRecord | null> {
  const { rows } = await pool.query('SELECT ack FROM acks WHERE user_id = $1 AND id = $2', [userId, id]);
  return rows[0]?.ack ?? null;
}

export async function purgeOldAcks(): Promise<void> {
  await pool.query("DELETE FROM acks WHERE created_at < now() - interval '24 hours'");
}

export async function lastSeq(room: string): Promise<number> {
  const { rows } = await pool.query('SELECT last_seq FROM rooms WHERE name = $1', [room]);
  return Number(rows[0]?.last_seq ?? 0);
}

export interface Appended { dup: boolean; seq: number | null; payload: Record<string, any> | null }

// One transaction: take the room's next seq (the row lock serialises writers), apply the change,
// append to the replay log and remember the ack. Commit order equals seq order.
async function append(
  userId: string, reqId: string, room: string, type: EventType,
  apply: (c: pg.PoolClient, seq: number) => Promise<Record<string, any>>,
): Promise<Appended> {
  const prior = await findAck(userId, reqId);
  if (prior) return { dup: true, seq: prior.seq, payload: null };
  const c = await pool.connect();
  try {
    await c.query('BEGIN');
    const next = await c.query(
      `INSERT INTO rooms (name, last_seq) VALUES ($1, 1)
       ON CONFLICT (name) DO UPDATE SET last_seq = rooms.last_seq + 1 RETURNING last_seq`, [room]);
    const seq = Number(next.rows[0].last_seq);
    const payload = await apply(c, seq);
    await c.query('INSERT INTO events (room, seq, id, type, author, payload) VALUES ($1, $2, $3, $4, $5, $6)',
      [room, seq, reqId, type, userId, payload]);
    await c.query('INSERT INTO acks (user_id, id, ack) VALUES ($1, $2, $3)', [userId, reqId, { seq, result: {} }]);
    await c.query('COMMIT');
    return { dup: false, seq, payload };
  } catch (err) {
    await c.query('ROLLBACK').catch(() => {});
    if ((err as { code?: string }).code === '23505') {          // unique violation
      const again = await findAck(userId, reqId);                // a concurrent duplicate won the race
      if (again) return { dup: true, seq: again.seq, payload: null };
      throw new StoreError('conflict', 'id already used');
    }
    throw err;
  } finally {
    c.release();
  }
}

export interface NewMessage { text: string; replyTo: string | null; attachments: Attachment[] }

export function addMessage(from: Author, room: string, id: string, m: NewMessage): Promise<Appended> {
  return append(from.userId, id, room, 'message', async (c, seq) => {
    if (m.replyTo) {
      const r = await c.query('SELECT 1 FROM messages WHERE id = $1 AND room = $2', [m.replyTo, room]);
      if (r.rowCount === 0) throw new StoreError('not_found', 'replyTo is not a message of this room');
    }
    const payload: ChatPayload = { from, text: m.text, replyTo: m.replyTo, attachments: m.attachments, seq, edited: false };
    await c.query(
      `INSERT INTO messages (id, room, seq, author, author_name, text, reply_to, attachments)
       VALUES ($1, $2, $3, $4, $5, $6, $7, $8)`,
      [id, room, seq, from.userId, from.name, m.text, m.replyTo, JSON.stringify(m.attachments)]);
    return payload as unknown as Record<string, any>;
  });
}

export function editMessage(from: Author, room: string, id: string, target: string, text: string): Promise<Appended> {
  return append(from.userId, id, room, 'edit', async (c, seq) => {
    const r = await c.query('SELECT author FROM messages WHERE id = $1 AND room = $2 FOR UPDATE', [target, room]);
    if (r.rowCount === 0) throw new StoreError('not_found', 'unknown target');
    if (r.rows[0].author !== from.userId) throw new StoreError('forbidden', 'only the author can edit');
    const editedAt = Date.now();
    await c.query('UPDATE messages SET text = $1, edited_at = $2 WHERE id = $3', [text, editedAt, target]);
    return { target, text, seq, editedAt };
  });
}

export function reactToMessage(from: Author, room: string, id: string, target: string, emoji: string, op: 'add' | 'remove'): Promise<Appended> {
  return append(from.userId, id, room, 'react', async (c, seq) => {
    const r = await c.query('SELECT 1 FROM messages WHERE id = $1 AND room = $2', [target, room]);
    if (r.rowCount === 0) throw new StoreError('not_found', 'unknown target');
    if (op === 'add') {
      const ins = await c.query('INSERT INTO reactions (message_id, user_id, emoji) VALUES ($1, $2, $3) ON CONFLICT DO NOTHING', [target, from.userId, emoji]);
      if (ins.rowCount === 0) throw new StoreError('conflict', 'reaction already added');
    } else {
      const del = await c.query('DELETE FROM reactions WHERE message_id = $1 AND user_id = $2 AND emoji = $3', [target, from.userId, emoji]);
      if (del.rowCount === 0) throw new StoreError('not_found', 'reaction not present');
    }
    return { target, emoji, op, from: { userId: from.userId }, seq };
  });
}

export interface StoredEvent { type: EventType; id: string; payload: Record<string, any> }

export async function eventsSince(room: string, since: number, limit = 500): Promise<StoredEvent[]> {
  const { rows } = await pool.query(
    'SELECT type, id, payload FROM events WHERE room = $1 AND seq > $2 ORDER BY seq LIMIT $3', [room, since, limit]);
  return rows;
}

export interface HistoryPage { messages: { id: string; payload: ChatPayload }[]; hasMore: boolean; next: number | null }

const MAX_PAGE_BYTES = 262144;

export async function history(room: string, before: number | null, limit: number): Promise<HistoryPage> {
  const { rows } = await pool.query(
    `SELECT id, seq, author, author_name, text, reply_to, attachments, edited_at FROM messages
     WHERE room = $1 AND ($2::bigint IS NULL OR seq < $2) ORDER BY seq DESC LIMIT $3`,
    [room, before, limit + 1]);
  const page: { id: string; payload: ChatPayload }[] = [];
  let bytes = 0;
  let trimmed = rows.length > limit;
  for (const r of rows.slice(0, limit)) {
    const payload: ChatPayload = {
      from: { userId: r.author, name: r.author_name }, text: r.text, replyTo: r.reply_to,
      attachments: r.attachments, seq: Number(r.seq), edited: r.edited_at !== null,
    };
    bytes += Buffer.byteLength(JSON.stringify(payload)) + 120;     // + envelope overhead
    if (bytes > MAX_PAGE_BYTES && page.length > 0) { trimmed = true; break; }
    page.push({ id: r.id, payload });
  }
  page.reverse();                                                  // ascending seq
  return { messages: page, hasMore: trimmed, next: trimmed ? page[0]!.payload.seq : null };
}

// Chatterbox v1 on Socket.IO 4.8: every envelope travels as one "cb" event.
// PostgreSQL keeps messages; Redis carries the Socket.IO adapter, tickets and resume sessions.
import { createServer, type IncomingMessage, type ServerResponse } from 'node:http';
import { randomBytes } from 'node:crypto';
import { readFile } from 'node:fs/promises';
import { createClient } from 'redis';
import { Server, type Socket } from 'socket.io';
import { createAdapter } from '@socket.io/redis-adapter';
import { ID_RE, ROOM_RE, type Envelope, type ErrorCode, type MessageType, type Status } from '../shared/protocol.ts';
import * as store from './store.ts';
import { StoreError } from './store.ts';

const PORT = Number(process.env.PORT ?? 8300);
const INSTANCE = process.env.INSTANCE ?? `node-${PORT}`;
const REDIS_URL = process.env.REDIS_URL ?? 'redis://localhost:6379';
const ORIGINS = (process.env.ALLOWED_ORIGINS ?? 'http://localhost:8300').split(',');
const HEARTBEAT_MS = 20_000;
const MAX_FRAME = 16_384;
const SESSION_TTL_S = 120;
const OFFLINE_GRACE_MS = 15_000;

interface User { userId: string; name: string }
interface SocketData { user?: User; status: Status; sessionId?: string }
type Wire = { cb: (env: Envelope) => void };
type CbSocket = Socket<Wire, Wire, Record<string, never>, SocketData>;
interface Runtime { lastSeen: number; tokens: number; refilledAt: number; limited: number[]; queue: Promise<void>; helloTimer?: NodeJS.Timeout }

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

// ---- HTTP: static client, one-time tickets, health ----------------------------------------------------
const STATIC: Record<string, [string, string]> = {
  '/': ['public/index.html', 'text/html; charset=utf-8'],
  '/app.js': ['public/app.js', 'text/javascript; charset=utf-8'],
  '/app.js.map': ['public/app.js.map', 'application/json'],
};

async function readBody(req: IncomingMessage): Promise<string> {
  let body = '';
  for await (const chunk of req) { body += chunk; if (body.length > 1024) break; }
  return body;
}

async function onRequest(req: IncomingMessage, res: ServerResponse): Promise<void> {
  const path = new URL(req.url ?? '/', 'http://x').pathname;
  if (path === '/healthz') { res.end(`ok ${INSTANCE}\n`); return; }
  if (path === '/api/ticket' && req.method === 'POST') {
    const origin = req.headers.origin;
    if (origin !== undefined && !ORIGINS.includes(origin)) { res.writeHead(403).end(); return; }
    const name = String(JSON.parse((await readBody(req)) || '{}').name ?? '');
    if (!/^[a-z0-9]{1,32}$/.test(name)) { res.writeHead(400).end('name must match [a-z0-9]{1,32}'); return; }
    // Placeholder login: a real service authenticates the user here (session cookie, OIDC) before issuing a ticket.
    const ticket = 't_' + randomBytes(16).toString('hex');
    await db.set(`cb:ticket:${ticket}`, name, { EX: 30 });
    res.writeHead(200, { 'content-type': 'application/json' }).end(JSON.stringify({ ticket }));
    return;
  }
  const file = STATIC[path];
  if (!file) { res.writeHead(404).end(); return; }
  res.writeHead(200, { 'content-type': file[1], 'cache-control': 'no-cache' }).end(await readFile(file[0]));
}

const httpServer = createServer((req, res) => {
  onRequest(req, res).catch((err) => { console.error(err); if (!res.headersSent) res.writeHead(500); res.end(); });
});

const io = new Server<Wire, Wire, Record<string, never>, SocketData>(httpServer, {
  maxHttpBufferSize: MAX_FRAME,                       // bytes per packet; a larger one closes the socket
  transports: ['websocket'],                          // no long-polling, so no sticky sessions
  allowRequest: (req, cb) => cb(null, req.headers.origin === undefined || ORIGINS.includes(req.headers.origin)),
  adapter: createAdapter(pub, sub),
});

// ---- helpers -------------------------------------------------------------------------------------------
const runtime = new WeakMap<CbSocket, Runtime>();
const sid = () => 's_' + randomBytes(13).toString('hex').toUpperCase();
const roomOf = (room: string) => `r:${room}`;
const userRoom = (userId: string) => `user:${userId}`;
const sessKey = (id: string) => `cb:session:${id}`;

type Target = { emit: (ev: 'cb', env: Envelope) => unknown };
const send = (target: Target, type: MessageType, room: string | null, payload: Record<string, unknown>, id = sid()) =>
  target.emit('cb', { v: 1, type, id, room, ts: Date.now(), payload });
const ack = (socket: CbSocket, req: Envelope, seq: number | null, dup: boolean, result: Record<string, unknown> = {}) =>
  send(socket, 'ack', req.room, { ref: req.id, ok: true, seq, dup, result });
const fail = (socket: CbSocket, ref: string | null, code: ErrorCode, message: string, retryable = false) =>
  send(socket, 'error', null, { ref, code, message, retryable });
const closeAfter = (socket: CbSocket) => setTimeout(() => socket.disconnect(true), 50);   // let the error frame flush

const chatRooms = (socket: CbSocket) => [...socket.rooms].filter((r) => r.startsWith('r:')).map((r) => r.slice(2));
const isMember = (socket: CbSocket, room: string | null): room is string => room !== null && socket.rooms.has(roomOf(room));

async function saveSession(socket: CbSocket): Promise<void> {
  const { sessionId, user } = socket.data;
  if (sessionId && user) await db.set(sessKey(sessionId), JSON.stringify({ userId: user.userId, rooms: chatRooms(socket) }), { EX: SESSION_TTL_S });
}

async function members(room: string) {
  const seen = new Map<string, { userId: string; name: string; status: Status }>();
  for (const s of await io.in(roomOf(room)).fetchSockets()) {        // works across replicas
    if (s.data.user) seen.set(s.data.user.userId, { ...s.data.user, status: s.data.status });
  }
  return [...seen.values()];
}

async function replay(socket: CbSocket, room: string, since: number): Promise<void> {
  for (const e of await store.eventsSince(room, since)) send(socket, e.type, room, e.payload, e.id);
}

// ---- handlers ------------------------------------------------------------------------------------------
type Handler = (socket: CbSocket, env: Envelope) => Promise<void>;

const handlers: Partial<Record<MessageType, Handler>> = {
  async hello(socket, env) {
    if (socket.data.user) return void fail(socket, env.id, 'conflict', 'hello already completed');
    const ticket = String(env.payload.token ?? '');
    const name = /^t_[0-9a-f]{32}$/.test(ticket) ? await db.getDel(`cb:ticket:${ticket}`) : null;   // one-time
    if (!name) { fail(socket, env.id, 'unauthorized', 'invalid or expired ticket'); return void closeAfter(socket); }
    clearTimeout(runtime.get(socket)!.helloTimer);
    const user: User = { userId: `u_${name}`, name };
    socket.data.user = user;
    socket.join(userRoom(user.userId));
    let resumed = false;
    const resume = env.payload.resume as { sessionId?: string; rooms?: Record<string, number> } | undefined;
    const saved = resume?.sessionId ? JSON.parse((await db.get(sessKey(resume.sessionId))) ?? 'null') : null;
    if (saved?.userId === user.userId) {
      resumed = true;
      socket.data.sessionId = resume!.sessionId;
      for (const room of saved.rooms as string[]) socket.join(roomOf(room));
    }
    socket.data.sessionId ??= 'sess_' + randomBytes(8).toString('hex');
    await saveSession(socket);
    send(socket, 'hello', null, {
      ref: env.id, ...user, sessionId: socket.data.sessionId, heartbeatMs: HEARTBEAT_MS, maxFrameBytes: MAX_FRAME,
      resumed, rooms: chatRooms(socket), instance: INSTANCE,
    });
    if (resumed) {
      for (const room of chatRooms(socket)) {
        send(socket.to(roomOf(room)), 'presence', room, { ...user, status: socket.data.status });
        await replay(socket, room, resume!.rooms?.[room] ?? 0);
      }
    }
  },

  async join(socket, env) {
    const room = env.room;
    if (room === null || !ROOM_RE.test(room)) return void fail(socket, env.id, 'bad_request', 'invalid room');
    const dm = /^dm-([a-z0-9]+)-([a-z0-9]+)$/.exec(room);
    if (dm && !dm.slice(1).includes(socket.data.user!.name)) return void fail(socket, env.id, 'forbidden', 'not your DM');
    const already = isMember(socket, room);
    socket.join(roomOf(room));
    await saveSession(socket);
    ack(socket, env, null, false, { members: await members(room), lastSeq: await store.lastSeq(room) });
    const since = env.payload.since;
    if (typeof since === 'number') await replay(socket, room, since);
    if (!already) send(socket.to(roomOf(room)), 'presence', room, { ...socket.data.user, status: socket.data.status });
  },

  async leave(socket, env) {
    if (!isMember(socket, env.room)) return void fail(socket, env.id, 'not_found', 'not a member');
    socket.leave(roomOf(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 void fail(socket, env.id, 'bad_request', 'text must be 1..4000 characters');
    if (replyTo !== null && typeof replyTo !== 'string') return void fail(socket, env.id, 'bad_request', 'replyTo must be a string or null');
    if (!Array.isArray(attachments) || attachments.length > 5 || !attachments.every(validAttachment)) return void fail(socket, env.id, 'bad_request', 'invalid attachments');
    if (!isMember(socket, env.room)) return void fail(socket, env.id, 'forbidden', 'join the room first');
    const out = await store.addMessage(socket.data.user!, env.room, env.id, { text, replyTo, attachments });
    ack(socket, env, out.seq, out.dup);
    if (!out.dup) send(io.to(roomOf(env.room)), 'message', env.room, out.payload!, env.id);   // sender included, across replicas
  },

  async edit(socket, env) {
    const { target, text } = env.payload;
    if (typeof target !== 'string' || typeof text !== 'string' || text.length < 1 || text.length > 4000) return void fail(socket, env.id, 'bad_request', 'target and text (1..4000) required');
    if (!isMember(socket, env.room)) return void fail(socket, env.id, 'forbidden', 'join the room first');
    const out = await store.editMessage(socket.data.user!, env.room, env.id, target, text);
    ack(socket, env, out.seq, out.dup);
    if (!out.dup) send(io.to(roomOf(env.room)), 'edit', env.room, out.payload!, env.id);
  },

  async react(socket, env) {
    const { target, emoji, op } = env.payload;
    const graphemes = typeof emoji === 'string' ? [...new Intl.Segmenter().segment(emoji)].length : 0;
    if (typeof target !== 'string' || graphemes !== 1 || (op !== 'add' && op !== 'remove')) return void fail(socket, env.id, 'bad_request', 'target, one-grapheme emoji and op required');
    if (!isMember(socket, env.room)) return void fail(socket, env.id, 'forbidden', 'join the room first');
    const out = await store.reactToMessage(socket.data.user!, env.room, env.id, target, emoji, op);
    ack(socket, env, out.seq, out.dup);
    if (!out.dup) send(io.to(roomOf(env.room)), 'react', env.room, out.payload!, env.id);
  },

  async typing(socket, env) {                      // fire-and-forget: no ack, errors are dropped
    const state = env.payload.state;
    if (isMember(socket, env.room) && (state === 'start' || state === 'stop')) {
      send(socket.to(roomOf(env.room)), 'typing', env.room, { userId: socket.data.user!.userId, state });
    }
  },

  async receipt(socket, env) {                     // fire-and-forget, cumulative
    const { kind, upTo } = env.payload;
    if (!isMember(socket, env.room) || (kind !== 'delivered' && kind !== 'read') || !Number.isInteger(upTo)) return;
    if (upTo > (await store.lastSeq(env.room))) return;
    send(socket.to(roomOf(env.room)), 'receipt', env.room, { userId: socket.data.user!.userId, kind, upTo });
  },

  async presence(socket, env) {
    const status = env.payload.status;
    if (status !== 'online' && status !== 'away') return void fail(socket, env.id, 'bad_request', 'status must be online or away');
    const changed = socket.data.status !== status;
    socket.data.status = status;
    await saveSession(socket);                     // the heartbeat also keeps the resume session alive
    if (changed) for (const room of chatRooms(socket)) send(socket.to(roomOf(room)), 'presence', room, { ...socket.data.user, status });
    ack(socket, env, null, false);
  },

  async history(socket, env) {
    if (!isMember(socket, env.room)) return void fail(socket, env.id, 'forbidden', 'join the room first');
    const { before = null, limit = 50 } = env.payload;
    if ((before !== null && !Number.isInteger(before)) || !Number.isInteger(limit) || limit < 1 || limit > 100) return void fail(socket, env.id, 'bad_request', 'before: int|null, limit: 1..100');
    const page = await store.history(env.room, before, limit);
    const messages = page.messages.map((m) => ({ v: 1, type: 'message', id: m.id, room: env.room, ts: Date.now(), payload: m.payload }));
    ack(socket, env, null, false, { messages, hasMore: page.hasMore, next: page.next });
  },
};

function validAttachment(a: any): boolean {
  return a !== null && typeof a === 'object' && typeof a.url === 'string' && a.url.startsWith('https://') &&
    typeof a.name === 'string' && typeof a.mime === 'string' && Number.isInteger(a.size);
}

// ---- connection lifecycle ------------------------------------------------------------------------------
io.on('connection', (socket) => {
  socket.data.status = 'online';
  const rt: Runtime = { lastSeen: Date.now(), tokens: 40, refilledAt: Date.now(), limited: [], queue: Promise.resolve() };
  runtime.set(socket, rt);
  rt.helloTimer = setTimeout(() => { fail(socket, null, 'unauthorized', 'hello not received within 5 s'); closeAfter(socket); }, 5000);

  socket.on('cb', (env) => {
    const now = Date.now();
    rt.lastSeen = now;
    rt.tokens = Math.min(40, rt.tokens + ((now - rt.refilledAt) / 1000) * 20);   // 20 tokens/s, burst 40
    rt.refilledAt = now;
    const ref = typeof env?.id === 'string' ? env.id : null;
    if (rt.tokens < 1) {                                                          // checked before deduplication
      rt.limited = rt.limited.filter((t) => now - t < 10_000).concat(now);
      fail(socket, ref, 'rate_limited', 'too many envelopes', true);
      if (rt.limited.length >= 10) closeAfter(socket);                            // protocol: close 4429
      return;
    }
    rt.tokens -= 1;
    rt.queue = rt.queue.then(() => dispatch(socket, env));                        // one at a time, in arrival order
  });

  // "disconnecting" fires while socket.rooms is still populated.
  socket.on('disconnecting', () => {
    clearTimeout(rt.helloTimer);
    const user = socket.data.user;
    if (!user) return;
    const rooms = chatRooms(socket);
    void saveSession(socket);                                                     // keeps the session for 120 s
    setTimeout(async () => {                                                      // offline only after a 15 s grace
      if ((await io.in(userRoom(user.userId)).fetchSockets()).length > 0) return; // reconnected, maybe on another replica
      for (const room of rooms) send(io.to(roomOf(room)), 'presence', room, { ...user, status: 'offline' });
    }, OFFLINE_GRACE_MS).unref();
  });
});

async function dispatch(socket: CbSocket, env: Envelope): Promise<void> {
  try {
    if (env === null || typeof env !== 'object') return void fail(socket, null, 'bad_request', 'not an envelope');
    const ref = typeof env.id === 'string' ? env.id : null;
    if (env.v !== 1) { fail(socket, ref, 'unsupported_version', 'v must be 1'); return void closeAfter(socket); }   // protocol: close 4426
    if (ref === null || !ID_RE.test(ref) || env.payload === null || typeof env.payload !== 'object' ||
        (env.room !== null && typeof env.room !== 'string')) {
      return void fail(socket, ref, 'bad_request', 'bad id, room or payload');
    }
    if (env.type !== 'hello' && !socket.data.user) { fail(socket, ref, 'unauthorized', 'send hello first'); return void closeAfter(socket); }
    const handler = handlers[env.type];
    if (!handler) return void fail(socket, ref, 'bad_request', `unknown type ${String(env.type)}`);
    await handler(socket, env);
  } catch (err) {
    if (err instanceof StoreError) return void fail(socket, env?.id ?? null, err.code, err.message);
    console.error(`[${INSTANCE}]`, err);
    fail(socket, env?.id ?? null, 'internal', 'internal error', true);
  }
}

// Application heartbeat: close sockets that sent nothing for 2 x heartbeatMs.
setInterval(() => {
  for (const [, socket] of io.sockets.sockets) {
    const rt = runtime.get(socket as CbSocket);
    if (rt && Date.now() - rt.lastSeen > 2 * HEARTBEAT_MS) socket.disconnect(true);
  }
}, HEARTBEAT_MS / 2).unref();
setInterval(() => void store.purgeOldAcks(), 3_600_000).unref();

httpServer.listen(PORT, () => console.log(`[${INSTANCE}] listening on ${PORT}`));
for (const sig of ['SIGINT', 'SIGTERM'] as const) {
  process.on(sig, () => {
    send(io, 'error', null, { ref: null, code: 'unavailable', message: 'server restarting', retryable: true });   // protocol: close 4503
    setTimeout(() => io.close(() => process.exit(0)), 100);
  });
}

npm install                    # creates package-lock.json, which the Dockerfile needs
docker compose up -d --build   # builds the image once, starts postgres, redis, chat-a, chat-b and nginx
curl http://localhost:8080/healthz    # answered by one of the replicas; repeat to see both names

docker compose exec postgres psql -U chatterbox -c "SELECT room, seq, author, text FROM messages ORDER BY seq DESC LIMIT 5"
docker compose exec redis redis-cli --scan --pattern 'cb:*'

// Two real browsers, one scale-out check, one outage. Run from the project directory with the stack up:
//   npm i --no-save puppeteer-core && CHROME_PATH=/path/to/chrome node test/e2e.mjs
import puppeteer from 'puppeteer-core';
import { execSync } from 'node:child_process';
import assert from 'node:assert/strict';

const { HTTP_PORT = '8080', A_PORT = '8081', B_PORT = '8082', CHROME_PATH } = process.env;
const compose = (args) => execSync(`docker compose ${args}`, { stdio: 'pipe' });
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
const room = 'e2e' + Math.random().toString(36).slice(2, 6);
const browsers = [];

async function open(url, name) {
  // One browser per user: background tabs of a single headless Chrome are throttled.
  const browser = await puppeteer.launch({ executablePath: CHROME_PATH, headless: 'new', protocolTimeout: 30_000 });
  browsers.push(browser);
  const page = await browser.newPage();
  page.on('pageerror', (e) => console.log(name, 'page error:', e.message));
  page.on('dialog', (d) => { console.log(name, 'alert:', d.message()); void d.dismiss(); });
  await page.goto(url);
  await page.type('#name', name);
  await page.click('#login button');
  await page.waitForSelector('#app', { visible: true });
  return page;
}
const joinRoom = async (page, r) => {
  await page.type('#room-input', r);
  await page.click('#join-btn');
  await page.waitForFunction((r) => document.getElementById('title').textContent === `#${r}`, {}, r);
};
const msgs = (page) => page.$$eval('.msg', (els) => els.map((e) => ({
  id: e.dataset.id, seq: Number(e.dataset.seq), meta: e.querySelector('.meta').textContent,
  body: e.querySelector('.body').textContent, reactions: [...e.querySelectorAll('.reactions button')].map((b) => b.textContent),
})));
const waitBody = (page, text) => page.waitForFunction(
  (t) => [...document.querySelectorAll('.msg .body')].some((e) => e.textContent === t), { timeout: 8000 }, text);
const bannerShown = (page, shown) => page.waitForFunction(
  (s) => (document.getElementById('banner').style.display === 'block') === s, { timeout: 30_000 }, shown);
const clickIn = (page, body, selector, index) => page.evaluate((body, selector, index) =>
  [...document.querySelectorAll('.msg')].find((m) => m.querySelector('.body').textContent === body).querySelectorAll(selector)[index].click(),
  body, selector, index);

try {
  // hello, join, presence across two replicas: alice talks to chat-a directly, bob goes through nginx
  const alice = await open(`http://localhost:${A_PORT}/`, 'alice');
  const bob = await open(`http://localhost:${HTTP_PORT}/`, 'bob');
  await joinRoom(alice, room);
  await joinRoom(bob, room);
  await alice.waitForFunction(() => [...document.querySelectorAll('#members li')].some((l) => l.textContent === 'bob'));

  // message, ack and receipts
  await alice.type('#text', 'Hello from Alice');
  await alice.keyboard.press('Enter');
  await waitBody(bob, 'Hello from Alice');
  await alice.waitForFunction(() => document.querySelector('.msg.mine .meta').textContent.endsWith('read'), { timeout: 8000 });

  // typing
  await bob.type('#text', 'x');
  await alice.waitForFunction(() => document.getElementById('typing').textContent.includes('bob is typing'), { timeout: 5000 });
  await bob.$eval('#text', (e) => e.select());
  await bob.type('#text', 'Bob here');
  await bob.keyboard.press('Enter');
  await waitBody(alice, 'Bob here');

  // react and edit
  await clickIn(alice, 'Bob here', '.reactions button', 0);
  await bob.waitForFunction(() => [...document.querySelectorAll('.msg')].find((m) => m.querySelector('.body').textContent === 'Bob here')
    .querySelector('.reactions button').textContent.includes('1'), { timeout: 5000 });
  await clickIn(bob, 'Bob here', '.tools button', 1);
  await bob.$eval('#text', (e) => e.select());
  await bob.type('#text', 'Bob here (fixed)');
  await bob.keyboard.press('Enter');
  await waitBody(alice, 'Bob here (fixed)');
  assert.match((await msgs(alice)).find((m) => m.body === 'Bob here (fixed)').meta, /\(edited\)/);

  // presence
  await bob.evaluate(() => window.chatterbox.client.setStatus('away'));
  await alice.waitForFunction(() => [...document.querySelectorAll('#members li')].find((l) => l.textContent === 'bob')?.title === 'away', { timeout: 5000 });
  await bob.evaluate(() => window.chatterbox.client.setStatus('online'));

  // history: 70 more messages (paced under the rate limit), then a third user pages back through them
  await alice.evaluate(async (room) => {
    for (let i = 0; i < 70; i++) { await window.chatterbox.client.send(room, `bulk ${i}`).ack; await new Promise((r) => setTimeout(r, 55)); }
  }, room);
  const carol = await open(`http://localhost:${B_PORT}/`, 'carol');
  await joinRoom(carol, room);
  const firstPage = (await msgs(carol)).length;
  await carol.click('#older');
  await carol.waitForFunction((n) => document.querySelectorAll('.msg').length > n, {}, firstPage);
  const seqs = (await msgs(carol)).map((m) => m.seq);
  assert.deepEqual(seqs, [...seqs].sort((a, b) => a - b));
  console.log(`history: ${firstPage} messages first, ${seqs.length} after "older"`);

  // outage 1: nginx stops, alice keeps writing, bob queues a message offline; the session is resumed
  compose('stop proxy');
  await bannerShown(bob, true);
  await alice.type('#text', 'sent while bob offline');
  await alice.keyboard.press('Enter');
  await bob.evaluate((room) => { window.chatterbox.client.send(room, 'queued offline by bob'); }, room);
  compose('start proxy');
  await bannerShown(bob, false);
  await waitBody(bob, 'sent while bob offline');
  await waitBody(alice, 'queued offline by bob');
  console.log('resumed:', await bob.evaluate(() => window.chatterbox.client.hello.resumed));
  assert.equal((await msgs(alice)).filter((m) => m.body === 'queued offline by bob').length, 1);

  // outage 2: the resume session is gone, so bob joins again with since = lastSeq
  const sessionId = await bob.evaluate(() => window.chatterbox.client.hello.sessionId);
  compose('stop proxy');
  await bannerShown(bob, true);
  compose(`exec -T redis redis-cli DEL cb:session:${sessionId}`);
  await alice.type('#text', 'missed with no session');
  await alice.keyboard.press('Enter');
  await sleep(500);
  compose('start proxy');
  await bannerShown(bob, false);
  await waitBody(bob, 'missed with no session');
  console.log('resumed:', await bob.evaluate(() => window.chatterbox.client.hello.resumed));
  const ids = (await msgs(bob)).map((m) => m.id);
  assert.equal(new Set(ids).size, ids.length, 'no duplicates');
  console.log('OK');
} finally {
  for (const b of browsers) await b.close();
}