maaz519
427396653f
Adds an opt-in RedisIoAdapter (wired in main.ts when REDIS_URL is set) so socket.io room emits fan out across instances via Redis pub/sub — a client on replica B now receives messages emitted by replica A. Without REDIS_URL the in-memory adapter is kept (single-instance dev unchanged). Adds redis to docker-compose, REDIS_URL to the env contract, and smoke-realtime-cluster.mjs which proves cross-instance delivery fails without Redis and passes with it. Delivery is at-least-once at N>1; clients dedupe by message id (documented). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
47 lines
2.0 KiB
TypeScript
47 lines
2.0 KiB
TypeScript
import { Logger, type INestApplicationContext } from '@nestjs/common';
|
|
import { IoAdapter } from '@nestjs/platform-socket.io';
|
|
import { createAdapter } from '@socket.io/redis-adapter';
|
|
import { Redis } from 'ioredis';
|
|
import type { Server, ServerOptions } from 'socket.io';
|
|
|
|
/**
|
|
* Socket.io adapter backed by Redis pub/sub (P9 scaling). With the default in-memory
|
|
* adapter, `server.to(room).emit()` only reaches sockets on the SAME instance; behind a
|
|
* load balancer a client on replica B never sees a message emitted by replica A. This
|
|
* adapter fans room emits across every replica through Redis, so realtime works at N>1.
|
|
* Wired only when REDIS_URL is set (single-instance dev keeps the in-memory adapter).
|
|
*/
|
|
export class RedisIoAdapter extends IoAdapter {
|
|
private readonly logger = new Logger(RedisIoAdapter.name);
|
|
private adapterConstructor?: ReturnType<typeof createAdapter>;
|
|
private clients: Redis[] = [];
|
|
|
|
constructor(
|
|
app: INestApplicationContext,
|
|
private readonly url: string,
|
|
) {
|
|
super(app);
|
|
}
|
|
|
|
/** Establish the pub/sub clients + the socket.io Redis adapter factory. */
|
|
async connect(): Promise<void> {
|
|
const pubClient = new Redis(this.url, { maxRetriesPerRequest: null });
|
|
const subClient = pubClient.duplicate();
|
|
this.clients = [pubClient, subClient];
|
|
pubClient.on('error', (err) => this.logger.error(`redis pub error: ${err.message}`));
|
|
subClient.on('error', (err) => this.logger.error(`redis sub error: ${err.message}`));
|
|
this.adapterConstructor = createAdapter(pubClient, subClient);
|
|
this.logger.log('socket.io Redis adapter connected — realtime fans out across replicas');
|
|
}
|
|
|
|
createIOServer(port: number, options?: ServerOptions): Server {
|
|
const server = super.createIOServer(port, options) as Server;
|
|
if (this.adapterConstructor) server.adapter(this.adapterConstructor);
|
|
return server;
|
|
}
|
|
|
|
async dispose(): Promise<void> {
|
|
await Promise.all(this.clients.map((c) => c.quit().catch(() => undefined)));
|
|
}
|
|
}
|