feat(iios): socket.io Redis adapter for cross-replica realtime (P9 scaling)

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>

packages/iios-service/.env.example

@@ -8,6 +8,9 @@
 DATABASE_URL=postgresql://iios:iios@localhost:5434/iios?schema=public   # 🔒 Postgres connection
 PORT=3200
 # NODE_ENV=production   # set by the Docker image
+# Realtime fan-out across replicas (socket.io Redis adapter). REQUIRED when running
+# more than one instance with live chat; omit for a single instance (in-memory adapter).
+# REDIS_URL=redis://localhost:6379                                       # 🔒
 
 # ── Secrets ──────────────────────────────────────────────────────────────────
 # JSON map of appId → HS256 signing secret used to verify session JWTs (SessionVerifier).

packages/iios-service/package.json

@@ -12,22 +12,24 @@
     "prisma:studio": "prisma studio"
   },
   "dependencies": {
-    "@insignia/iios-contracts": "workspace:*",
     "@insignia/iios-adapter-sdk": "workspace:*",
+    "@insignia/iios-contracts": "workspace:*",
     "@nestjs/common": "^11.1.27",
     "@nestjs/core": "^11.1.27",
     "@nestjs/platform-express": "^11.1.27",
     "@nestjs/platform-socket.io": "^11.1.27",
     "@nestjs/websockets": "^11.1.27",
     "@prisma/client": "^6.2.1",
-    "prisma": "^6.2.1",
-    "socket.io": "^4.8.3",
+    "@socket.io/redis-adapter": "^8.3.0",
     "class-transformer": "^0.5.1",
     "class-validator": "^0.15.1",
     "dotenv": "^16.4.7",
+    "ioredis": "^5.11.1",
     "jsonwebtoken": "^9.0.3",
+    "prisma": "^6.2.1",
     "reflect-metadata": "^0.2.2",
-    "rxjs": "^7.8.2"
+    "rxjs": "^7.8.2",
+    "socket.io": "^4.8.3"
   },
   "devDependencies": {
     "@insignia/iios-testkit": "workspace:*",

packages/iios-service/scripts/smoke-realtime-cluster.mjs

@@ -0,0 +1,53 @@
+// P9 scaling smoke: cross-instance realtime fan-out via the socket.io Redis adapter.
+// Alice connects to instance A, Bob to instance B (two separate processes sharing one
+// Redis + DB). Alice sends on A; Bob must receive it live on B — which only works if the
+// Redis adapter fans room emits across replicas. Set A_URL / B_URL (default 3201/3202).
+import 'dotenv/config';
+import { io } from 'socket.io-client';
+import jwt from 'jsonwebtoken';
+
+const A_URL = process.env.A_URL ?? 'http://localhost:3201';
+const B_URL = process.env.B_URL ?? 'http://localhost:3202';
+const APP_ID = 'portal-demo';
+const SECRET = JSON.parse(process.env.APP_SECRETS ?? '{"portal-demo":"dev-secret"}')[APP_ID];
+
+const sign = (userId) =>
+  jwt.sign({ sub: userId, name: userId, appId: APP_ID, orgId: `org_${APP_ID}` }, SECRET, { algorithm: 'HS256', expiresIn: '1h' });
+const connect = (url, userId) =>
+  io(`${url}/message`, { auth: { token: sign(userId) }, transports: ['websocket'], forceNew: true });
+const once = (socket, event, ms = 8000) =>
+  new Promise((res, rej) => {
+    const t = setTimeout(() => rej(new Error(`timeout waiting for "${event}"`)), ms);
+    socket.once(event, (v) => { clearTimeout(t); res(v); });
+  });
+const assert = (c, m) => { if (!c) { console.error('✗', m); process.exit(1); } console.log('✓', m); };
+
+const alice = connect(A_URL, 'alice-cluster');
+const bob = connect(B_URL, 'bob-cluster');
+
+try {
+  await Promise.all([once(alice, 'connect'), once(bob, 'connect')]);
+  assert(true, `Alice connected to A (${A_URL}), Bob to B (${B_URL})`);
+
+  // Alice opens a thread on instance A; Bob joins the SAME thread on instance B.
+  const opened = await alice.emitWithAck('open_thread', {});
+  const threadId = opened.threadId;
+  assert(!!threadId, `Alice opened thread ${threadId} on instance A`);
+  await bob.emitWithAck('open_thread', { threadId });
+  assert(true, 'Bob joined the same thread on instance B');
+
+  // Alice sends on A → Bob must receive it on B (cross-instance fan-out via Redis).
+  const bobReceives = once(bob, 'message');
+  await alice.emitWithAck('send_message', { threadId, content: 'hello across instances' });
+  const received = await bobReceives;
+  assert(received.content === 'hello across instances', `Bob received "${received.content}" on B — emitted from A ✓`);
+
+  console.log('\nP9 cross-instance realtime smoke: PASS');
+} catch (err) {
+  console.error('✗', err.message);
+  process.exit(1);
+} finally {
+  alice.close();
+  bob.close();
+}
+process.exit(0);

packages/iios-service/src/main.ts

@@ -4,6 +4,7 @@ import { NestFactory } from '@nestjs/core';
 import { ValidationPipe } from '@nestjs/common';
 import { AppModule } from './app.module';
 import { traceMiddleware } from './observability/trace.middleware';
+import { RedisIoAdapter } from './realtime/redis-io.adapter';
 
 async function bootstrap(): Promise<void> {
   const app = await NestFactory.create(AppModule, { rawBody: true });
@@ -12,6 +13,15 @@ async function bootstrap(): Promise<void> {
     new ValidationPipe({ whitelist: true, forbidNonWhitelisted: true, transform: true }),
   );
   app.enableCors({ origin: true, credentials: true });
+
+  // Multi-replica realtime: fan socket.io room emits across instances via Redis.
+  // Opt-in — without REDIS_URL the default in-memory adapter is used (single instance).
+  if (process.env.REDIS_URL) {
+    const redisAdapter = new RedisIoAdapter(app, process.env.REDIS_URL);
+    await redisAdapter.connect();
+    app.useWebSocketAdapter(redisAdapter);
+  }
+
   const port = Number(process.env.PORT ?? 3200);
   await app.listen(port);
   // eslint-disable-next-line no-console

packages/iios-service/src/realtime/redis-io.adapter.ts

@@ -0,0 +1,46 @@
+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)));
+  }
+}