WebSockets

npm i --save @nestjs/websockets @nestjs/platform-socket.io
# or, for plain WebSockets
npm i --save @nestjs/websockets @nestjs/platform-ws

@WebSocketGateway(80, { namespace: 'events' })      // own port 80, namespace /events
@WebSocketGateway(81, { transports: ['websocket'] }) // own port, no long-polling
@WebSocketGateway({ transports: ['websocket'] })     // same port as the HTTP server

import {
  Ack,
  ConnectedSocket,
  MessageBody,
  SubscribeMessage,
  WebSocketGateway,
  WebSocketServer,
  type WsResponse,
} from '@nestjs/websockets';
import { type Observable, from, map } from 'rxjs';
import type { Server, Socket } from 'socket.io';

@WebSocketGateway({ transports: ['websocket'] })
export class EventsGateway {
  @WebSocketServer()
  server!: Server;

  // The returned value is sent back as the acknowledgement.
  @SubscribeMessage('echo')
  echo(@MessageBody() data: string): string {
    return data;
  }

  // A property key extracts one field of the payload.
  @SubscribeMessage('id')
  id(@MessageBody('id') id: number): number {
    return id;
  }

  // The socket is available through @ConnectedSocket(); nothing is returned, so no ack is sent.
  @SubscribeMessage('whoami')
  whoami(@ConnectedSocket() client: Socket): void {
    client.emit('you', { id: client.id });
  }

  // Manual acknowledgement.
  @SubscribeMessage('manual-ack')
  manualAck(
    @MessageBody() data: string,
    @Ack() ack: (response: { status: string; data: string }) => void,
  ): void {
    ack({ status: 'received', data });
  }

  // A WsResponse is emitted as a named event instead of an acknowledgement.
  @SubscribeMessage('named')
  named(@MessageBody() data: unknown): WsResponse<unknown> {
    return { event: 'named-reply', data };
  }

  // An Observable emits one event per value.
  @SubscribeMessage('stream')
  stream(): Observable<WsResponse<number>> {
    return from([1, 2, 3]).pipe(map((data) => ({ event: 'tick', data })));
  }

  // The positional (no decorators) form: client first, then data.
  @SubscribeMessage('positional')
  positional(client: Socket, data: string): string {
    return `${client.id.length > 0 ? 'ok' : 'no'}:${data}`;
  }

  // Falsy values other than null/undefined are still sent.
  @SubscribeMessage('zero')
  zero(): number {
    return 0;
  }
}

import { io } from 'socket.io-client';

const socket = io('http://localhost:3000', { transports: ['websocket'] });

await socket.emitWithAck('echo', 'hi');                  // 'hi'
await socket.emitWithAck('id', { id: 7, other: 1 });     // 7  (only the id property was extracted)
await socket.emitWithAck('manual-ack', 'x');             // { status: 'received', data: 'x' }
await socket.emitWithAck('zero');                        // 0

socket.on('named-reply', (data) => console.log(data));   // { a: 1 }  (WsResponse)
socket.emit('named', { a: 1 });

socket.on('tick', (n) => console.log(n));                // 1, 2, 3 (Observable)
socket.emit('stream');

@WebSocketGateway({ namespace: 'chat' })
export class ChatGateway {
  @WebSocketServer()
  namespace: Namespace; // a Server when no namespace is configured
}

@SubscribeMessage('join')
async join(@MessageBody() dto: JoinDto, @ConnectedSocket() client: Socket) {
  const room = `book:${dto.bookId}`;
  await client.join(room);
  // Everyone in the room except the sender.
  client.to(room).emit('presence', { user: client.data.user.name, joined: true });
  return { room };
}

@SubscribeMessage('leave')
async leave(@MessageBody() dto: JoinDto, @ConnectedSocket() client: Socket) {
  const room = `book:${dto.bookId}`;
  await client.leave(room);
  client.to(room).emit('presence', { user: client.data.user.name, joined: false });
  return { room };
}

// Who is in a room right now (on this instance; with a Redis adapter, across all of them).
@SubscribeMessage('members')
async members(@MessageBody() dto: JoinDto) {
  const sockets = await this.namespace.in(`book:${dto.bookId}`).fetchSockets();
  return sockets.map((s) => s.data.user.name).sort();
}

handleDisconnect(client: Socket, reason?: string) {
  // client.rooms is already empty here: remember what you need on client.data
  this.logger.debug(`disconnected ${client.id}: ${reason}`);
}

afterInit(namespace: Namespace) {
  // Authenticate at the handshake: guards only run for messages, not for the connection.
  namespace.use(async (socket, next) => {
    try {
      const token = socket.handshake.auth?.token as string | undefined;
      if (!token) throw new Error('missing token');
      const payload = await this.jwt.verifyAsync<{ sub: string; name: string }>(token);
      socket.data.user = { id: payload.sub, name: payload.name };
      next();
    } catch {
      next(new Error('unauthorized'));
    }
  });
}

import { Inject, Injectable, Scope } from '@nestjs/common';
import { REQUEST } from '@nestjs/core';
import type { Socket } from 'socket.io';

// One instance per connected socket (v12).
@Injectable({ scope: Scope.REQUEST })
export class ConnectionStateService {
  private sequence = 0;
  static created = 0;

  constructor(@Inject(REQUEST) private readonly client: Socket) {
    ConnectionStateService.created++;
  }

  next() {
    return { clientId: this.client.id, sequence: ++this.sequence };
  }
}

import { Inject } from '@nestjs/common';
import { OnGatewayConnection, OnGatewayInit, SubscribeMessage, WebSocketGateway } from '@nestjs/websockets';
import type { Socket } from 'socket.io';
import { ConnectionStateService } from './connection-state.service.js';

// Depending on a request-scoped provider makes the gateway itself request-scoped.
@WebSocketGateway({ namespace: 'state', transports: ['websocket'] })
export class StateGateway implements OnGatewayInit, OnGatewayConnection {
  static afterInitSawState: boolean | undefined;

  constructor(@Inject(ConnectionStateService) private readonly state: ConnectionStateService) {}

  afterInit() {
    // Runs once on a shell instance: constructor dependencies are not available here.
    StateGateway.afterInitSawState = this.state !== undefined;
  }

  handleConnection(client: Socket) {
    client.emit('connected', this.state.next());
  }

  @SubscribeMessage('state')
  onState() {
    return { event: 'state', data: this.state.next() };
  }
}

import { Inject, Logger, UseFilters, UseGuards, UseInterceptors, UsePipes } from '@nestjs/common';
import { JwtService } from '@nestjs/jwt';
import {
  ConnectedSocket,
  MessageBody,
  OnGatewayConnection,
  OnGatewayDisconnect,
  OnGatewayInit,
  SubscribeMessage,
  WebSocketGateway,
  WebSocketServer,
} from '@nestjs/websockets';
import type { Namespace, Socket } from 'socket.io';
import { AckEnvelopeInterceptor } from './envelope.interceptor.js';
import { ChatExceptionFilter } from './chat.exception-filter.js';
import { JoinDto, SendMessageDto } from './chat.dto.js';
import { wsValidationPipe } from './chat.pipe.js';
import { RoomMemberGuard } from './room-member.guard.js';

@UseFilters(new ChatExceptionFilter())
@UsePipes(wsValidationPipe())
@WebSocketGateway({ namespace: 'chat', transports: ['websocket'], cors: { origin: false } })
export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewayDisconnect {
  private readonly logger = new Logger(ChatGateway.name);
  static disconnects: Array<{ id: string; reason?: string }> = [];

  // With a namespace option this is a Namespace, not a Server.
  @WebSocketServer()
  namespace!: Namespace;

  constructor(@Inject(JwtService) private readonly jwt: JwtService) {}

  afterInit(namespace: Namespace) {
    // Authenticate at the handshake: guards only run for messages, not for the connection.
    namespace.use(async (socket, next) => {
      try {
        const token = socket.handshake.auth?.token as string | undefined;
        if (!token) throw new Error('missing token');
        const payload = await this.jwt.verifyAsync<{ sub: string; name: string }>(token);
        socket.data.user = { id: payload.sub, name: payload.name };
        next();
      } catch {
        next(new Error('unauthorized'));
      }
    });
  }

  handleConnection(client: Socket) {
    this.logger.debug(`connected ${client.id} as ${client.data.user.name}`);
  }

  handleDisconnect(client: Socket, reason?: string) {
    ChatGateway.disconnects.push({ id: client.id, reason });
    // The socket has left every room by now; tell the others who is gone.
    this.logger.debug(`disconnected ${client.id}: ${reason}`);
  }

  @SubscribeMessage('join')
  async join(@MessageBody() dto: JoinDto, @ConnectedSocket() client: Socket) {
    const room = `book:${dto.bookId}`;
    await client.join(room);
    // Everyone in the room except the sender.
    client.to(room).emit('presence', { user: client.data.user.name, joined: true });
    return { room };
  }

  @SubscribeMessage('leave')
  async leave(@MessageBody() dto: JoinDto, @ConnectedSocket() client: Socket) {
    const room = `book:${dto.bookId}`;
    await client.leave(room);
    client.to(room).emit('presence', { user: client.data.user.name, joined: false });
    return { room };
  }

  // Who is in a room right now (on this instance; with a Redis adapter, across all of them).
  @SubscribeMessage('members')
  async members(@MessageBody() dto: JoinDto) {
    const sockets = await this.namespace.in(`book:${dto.bookId}`).fetchSockets();
    return sockets.map((s) => s.data.user.name).sort();
  }

  @UseGuards(RoomMemberGuard)
  @UseInterceptors(AckEnvelopeInterceptor)
  @SubscribeMessage('send')
  send(@MessageBody() dto: SendMessageDto, @ConnectedSocket() client: Socket) {
    const message = { bookId: dto.bookId, from: client.data.user.name, text: dto.text, at: Date.now() };
    // Everyone in the room, including the sender (namespace.to, not client.to).
    this.namespace.to(`book:${dto.bookId}`).emit('message', message);
    return { delivered: true };
  }
}

import { Module } from '@nestjs/common';
import { ConfigModule, ConfigService } from '@nestjs/config';
import { JwtModule } from '@nestjs/jwt';
import { ChatGateway } from './chat.gateway.js';
import { ConnectionStateService } from './connection-state.service.js';
import { DefaultErrorsGateway, QuietErrorsGateway } from './errors.gateway.js';
import { SchemaGateway } from './schema.gateway.js';
import { StateGateway } from './state.gateway.js';
import { EventsGateway } from './events.gateway.js';

@Module({
  imports: [
    ConfigModule.forRoot({ isGlobal: true }),
    JwtModule.registerAsync({
      inject: [ConfigService],
      useFactory: (config: ConfigService) => ({ secret: config.getOrThrow<string>('JWT_SECRET') }),
    }),
  ],
  providers: [EventsGateway, ChatGateway, StateGateway, ConnectionStateService, QuietErrorsGateway, DefaultErrorsGateway, SchemaGateway],
})
export class WebsocketsModule {}

import { io } from 'socket.io-client';

const socket = io('http://localhost:3000/chat', {
  transports: ['websocket'],
  auth: { token: await getAccessToken() }, // the same JWT used for the REST API
});

socket.on('connect_error', (err) => console.error(err.message)); // "unauthorized"
socket.on('presence', (p) => console.log(p));                    // { user: 'bob', joined: true }
socket.on('message', (m) => console.log(m));                     // { bookId, from, text, at }

await socket.emitWithAck('join', { bookId: 1 });                 // { room: 'book:1' }
const ack = await socket.emitWithAck('send', { bookId: 1, text: 'Loved chapter 3' });
// { ok: true, data: { delivered: true } } thanks to the interceptor