Socket.IO
Server

export interface Requests {
  game_join: { requestId: string; gameId: string };
  move_submit: { requestId: string; gameId: string; word: string };
}
export interface Results {
  game_join: { players: number };
  move_submit: { score: number };
}
export type Ok<T> = { ok: true; event: string; requestId: string; data: T };
export type Fail = { ok: false; event: string; requestId: string; error: { code: string; message: string } };

// every request event gets a :response and an :error twin, checked by the compiler
export type ServerEvents =
  { [K in keyof Requests as `${K & string}:response`]: (res: Ok<Results[K]>) => void } &
  { [K in keyof Requests as `${K & string}:error`]: (err: Fail) => void } &
  { game_state: (state: GameState) => void };

export type ClientEvents = {
  [K in keyof Requests]: (req: Requests[K], ack?: (res: Ok<Results[K]> | Fail) => void) => void;
};

const http = createServer();
export const io = new Server<ClientEvents, ServerEvents, {}, { userId: string }>(http, {
  cors: { origin: ['https://app.example.com'], credentials: true },
  pingInterval: 25_000, // the defaults, shown so you know they exist
  pingTimeout: 20_000,
  maxHttpBufferSize: 1e5, // 100 KB is plenty for game moves
});
http.listen(3000);

io.on('connection', (socket) => {
  socket.join(`user:${socket.data.userId}`);

  socket.on('move_submit', async (req, ack) => {
    const state = await games.apply(req);
    socket.to(`game:${req.gameId}`).emit('game_state', state); // the others
    io.to(`user:${state.nextPlayer}`).emit('game_state', state);  // all their tabs
  });
});

export function on<K extends keyof Requests>(
  socket: AppSocket, event: K, schema: ZodType<Requests[K]>,
  run: (req: Requests[K]) => Promise<Results[K]>,
) {
  socket.on(event, async (payload: unknown, ack?: (r: Ok<Results[K]> | Fail) => void) => {
    const requestId = (payload as { requestId?: string })?.requestId ?? randomUUID();
    const reply = typeof ack === 'function' ? ack : () => {};
    const fail = (code: string, message: string) => {
      const err: Fail = { ok: false, event, requestId, error: { code, message } };
      socket.emit(`${event}:error`, err);
      reply(err);
    };
    const parsed = schema.safeParse(payload);
    if (!parsed.success) return fail('INVALID_PAYLOAD', parsed.error.issues[0].message);
    try {
      const res: Ok<Results[K]> = { ok: true, event, requestId, data: await run(parsed.data) };
      socket.emit(`${event}:response`, res);
      reply(res);
    } catch (e) {
      e instanceof DomainError ? fail(e.code, e.message) : fail('INTERNAL', 'Could not complete the request');
    }
  });
}

const socket = { handlers: {}, on(e, fn) { this.handlers[e] = fn; },
  emit(e, body) { console.log(e.padEnd(21), JSON.stringify(body)); } };

function on(event, validate, run) {
  socket.on(event, async (req = {}) => {
    const base = { event, requestId: req.requestId };
    const problem = validate(req);
    if (problem) return socket.emit(event + ':error', { ok: false, ...base, error: problem });
    try {
      socket.emit(event + ':response', { ok: true, ...base, data: await run(req) });
    } catch (e) {
      socket.emit(event + ':error', { ok: false, ...base, error: { code: e.code || 'INTERNAL' } });
    }
  });
}

const words = new Set(['mango', 'sushi']);
on('move_submit',
  (req) => (typeof req.word === 'string' ? null : { code: 'INVALID_PAYLOAD' }),
  async (req) => {
    if (!words.has(req.word)) throw Object.assign(new Error(), { code: 'WORD_UNKNOWN' });
    return { score: req.word.length * 2 };
  });

(async () => {
  await socket.handlers.move_submit({ requestId: 'r1', word: 'sushi' });
  await socket.handlers.move_submit({ requestId: 'r2', word: 'pizza' });
  await socket.handlers.move_submit({ requestId: 'r3', word: 42 });
})();

// ask one client to confirm, with a deadline
try {
  const res = await socket.timeout(5_000).emitWithAck('rematch_offer', { gameId });
  if (res.ok) await games.rematch(gameId);
} catch {
  socket.emit('rematch_offer:error', { ok: false, event: 'rematch_offer', requestId, error: { code: 'TIMEOUT', message: 'No answer in 5s' } });
}

// ask a room, collect every reply
io.to(`game:${gameId}`).timeout(3_000).emit('ready_check', {}, (err, replies) => {
  if (err) log.warn('some players did not answer');
  const ready = replies.filter((r) => r.ok).length;
});

on(socket, 'game_join', JoinSchema, async ({ gameId }) => {
  await games.assertPlayer(gameId, socket.data.userId); // throws DomainError('NOT_A_PLAYER')
  await socket.join(`game:${gameId}`);
  const players = (await io.in(`game:${gameId}`).fetchSockets()).length;
  return { players };
});

socket.on('disconnecting', () => {
  for (const room of socket.rooms) if (room.startsWith('game:')) io.to(room).emit('player_left', { userId: socket.data.userId });
});

// one namespace per tenant, created on first connect
io.of(/^\/tenant-\w+$/).on('connection', (s) => { const tenant = s.nsp.name.slice(8); });

io.use(async (socket, next) => {
  try {
    const { sub, exp } = await verifyJwt(socket.handshake.auth.token);
    socket.data.userId = sub;
    setTimeout(() => socket.disconnect(true), exp * 1000 - Date.now()); // tokens expire, sockets must too
    next();
  } catch {
    const err = new Error('unauthorized') as Error & { data?: unknown };
    err.data = { code: 'TOKEN_INVALID' }; // same code vocabulary as event:error
    next(err); // the client gets connect_error with message and data
  }
});

@Injectable()
export class WsContractInterceptor implements NestInterceptor {
  intercept(ctx: ExecutionContext, next: CallHandler) {
    const ws = ctx.switchToWs();
    const event = ws.getPattern();
    const requestId = ws.getData()?.requestId ?? randomUUID();
    return next.handle().pipe(map((data) => {
      const res = { ok: true, event, requestId, data };
      ws.getClient<Socket>().emit(`${event}:response`, res);
      return res; // also becomes the ack
    }));
  }
}

@Catch()
export class WsContractFilter implements ExceptionFilter {
  catch(err: unknown, host: ArgumentsHost) {
    const ws = host.switchToWs();
    const event = ws.getPattern();
    const code = err instanceof WsException ? String(err.getError()) : 'INTERNAL';
    ws.getClient<Socket>().emit(`${event}:error`, {
      ok: false, event, requestId: ws.getData()?.requestId, error: { code, message: 'Request failed' },
    });
  }
}

@UseFilters(WsContractFilter)
@UseInterceptors(WsContractInterceptor)
@UsePipes(new ValidationPipe({ whitelist: true, exceptionFactory: () => new WsException('INVALID_PAYLOAD') }))
@WebSocketGateway({ namespace: '/game', cors: { origin: process.env.WEB_ORIGIN } })
export class GameGateway {
  @WebSocketServer() server: Namespace;

  @SubscribeMessage('move_submit')
  async submit(@ConnectedSocket() s: Socket, @MessageBody() dto: MoveDto) {
    const state = await this.games.apply(s.data.userId, dto);
    this.server.to(`game:${dto.gameId}`).except(s.id).emit('game_state', state);
    return { score: state.lastScore };
  }
}

export class RedisIoAdapter extends IoAdapter {
  private adapter: ReturnType<typeof createAdapter>;

  async connect() {
    const pub = createClient({ url: process.env.REDIS_URL });
    const sub = pub.duplicate();
    await Promise.all([pub.connect(), sub.connect()]);
    this.adapter = createAdapter(pub, sub);
  }

  createIOServer(port: number, options?: ServerOptions) {
    const server = super.createIOServer(port, options);
    server.adapter(this.adapter);
    return server;
  }
}

upstream socket_nodes {
  ip_hash;                          # same client, same node
  server app1:"n">3000;
  server app2:"n">3000;
}
location /socket.io/ {
  proxy_pass http://socket_nodes;
  proxy_http_version "n">1.1;
  proxy_set_header Upgrade $http_upgrade;
  proxy_set_header Connection "upgrade";
  proxy_read_timeout "n">75s;           # longer than pingInterval
}