diff --git a/docker-compose.yml b/docker-compose.yml index 34e31ed..275a98c 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -16,5 +16,18 @@ services: timeout: 5s retries: 10 + # Realtime fan-out across replicas (socket.io Redis adapter). Point the service at it + # with REDIS_URL=redis://localhost:6379; without REDIS_URL the in-memory adapter is used. + redis: + image: redis:7-alpine + container_name: iios-redis + ports: + - "6379:6379" + healthcheck: + test: ["CMD", "redis-cli", "ping"] + interval: 5s + timeout: 5s + retries: 10 + volumes: iios_postgres_data: diff --git a/docs/DEPLOYMENT.md b/docs/DEPLOYMENT.md index 7e995a9..fe377a5 100644 --- a/docs/DEPLOYMENT.md +++ b/docs/DEPLOYMENT.md @@ -65,9 +65,15 @@ for the full, commented list. Highlights: ``` - **Postgres** — managed (RDS / Cloud SQL / Neon). One logical DB, tenant-isolated by scope. -- **Redis** — **not required for a single instance.** Add it when you run **multiple replicas - with live chat** (socket.io needs its Redis adapter to fan-out across instances) or when - BullMQ queues are introduced. +- **Redis** — **not required for a single instance.** Set **`REDIS_URL`** when you run + **multiple replicas with live chat**: the built-in [socket.io Redis adapter](../packages/iios-service/src/realtime/redis-io.adapter.ts) + fans room emits across all instances, so a client on replica B sees messages emitted by + replica A. Without `REDIS_URL` the in-memory adapter is used (single instance). Verified by + `scripts/smoke-realtime-cluster.mjs` (two instances + one Redis; cross-instance delivery + fails without Redis, passes with it). + > Delivery is **at-least-once** across instances — with N>1 the inline send-emit and the + > relay's outbox-bridge emit can both fire for the same message, so clients should **dedupe + > by message `id`** (the payload always carries one). - **Platform ports** (OPA policy, CMP consent, MDM, CRRE, SAS) — today in-process permissive stubs (`LocalDevPorts`). For production, point these at real external services; the service already calls them **fail-closed**. @@ -94,7 +100,6 @@ for the full, commented list. Highlights: ## Not yet built (prod-readiness gaps) - **CI/CD pipeline** to build/push this image and run migrations. -- **Redis + socket.io Redis adapter** wiring (needed at N>1 with realtime). - **Real platform-port services** (OPA/CMP/MDM) — today permissive stubs. - **Secrets manager** integration (currently plain env). - A **schema-compatibility gate** in CI to enforce expand-contract migrations. diff --git a/packages/iios-service/.env.example b/packages/iios-service/.env.example index 9ca6fb4..791514e 100644 --- a/packages/iios-service/.env.example +++ b/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). diff --git a/packages/iios-service/package.json b/packages/iios-service/package.json index 819bd61..00b6c12 100644 --- a/packages/iios-service/package.json +++ b/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:*", diff --git a/packages/iios-service/scripts/smoke-realtime-cluster.mjs b/packages/iios-service/scripts/smoke-realtime-cluster.mjs new file mode 100644 index 0000000..0153f6b --- /dev/null +++ b/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); diff --git a/packages/iios-service/src/main.ts b/packages/iios-service/src/main.ts index 254957c..e3c6537 100644 --- a/packages/iios-service/src/main.ts +++ b/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 { const app = await NestFactory.create(AppModule, { rawBody: true }); @@ -12,6 +13,15 @@ async function bootstrap(): Promise { 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 diff --git a/packages/iios-service/src/realtime/redis-io.adapter.ts b/packages/iios-service/src/realtime/redis-io.adapter.ts new file mode 100644 index 0000000..84b56d1 --- /dev/null +++ b/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; + 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 { + 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 { + await Promise.all(this.clients.map((c) => c.quit().catch(() => undefined))); + } +} diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 5f69263..a8f0739 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -352,6 +352,9 @@ importers: '@prisma/client': specifier: ^6.2.1 version: 6.19.3(prisma@6.19.3(typescript@5.9.3))(typescript@5.9.3) + '@socket.io/redis-adapter': + specifier: ^8.3.0 + version: 8.3.0(socket.io-adapter@2.5.8) class-transformer: specifier: ^0.5.1 version: 0.5.1 @@ -361,6 +364,9 @@ importers: dotenv: specifier: ^16.4.7 version: 16.6.1 + ioredis: + specifier: ^5.11.1 + version: 5.11.1 jsonwebtoken: specifier: ^9.0.3 version: 9.0.3 @@ -1164,6 +1170,9 @@ packages: '@types/node': optional: true + '@ioredis/commands@1.10.0': + resolution: {integrity: sha512-UmeW7z4LfctwoQ5wkhVzgq8tXkreED2xZGpX+Bg+zA+WJFZCT6c062AfCK/Dfk81xZnnwdhJCUMkitihRaoC2Q==} + '@jridgewell/gen-mapping@0.3.13': resolution: {integrity: sha512-2kkt/7niJ6MgEPxF0bYdQ6etZaA+fQvDcLKckhy1yIQOzaoKjBBjSj63/aLVjYE3qhRt5dvM+uUyfCg6UKCBbA==} @@ -1426,6 +1435,12 @@ packages: '@socket.io/component-emitter@3.1.2': resolution: {integrity: sha512-9BCxFwvbGg/RsZK9tjXd8s4UcwR0MWeFQ1XEKIQVVvAGJyINdrqKMcTRyLoK8Rse1GjzLV9cwjWV1olXRWEXVA==} + '@socket.io/redis-adapter@8.3.0': + resolution: {integrity: sha512-ly0cra+48hDmChxmIpnESKrc94LjRL80TEmZVscuQ/WWkRP81nNj8W8cCGMqbI4L6NCuAaPRSzZF1a9GlAxxnA==} + engines: {node: '>=10.0.0'} + peerDependencies: + socket.io-adapter: ^2.5.4 + '@standard-schema/spec@1.1.0': resolution: {integrity: sha512-l2aFy5jALhniG5HgqrD6jXLi/rUWrKvqN/qJx6yoJsgKhblVd+iqqU4RCXavm/jPityDo5TCvKMnpjKnOriy0w==} @@ -1834,6 +1849,10 @@ packages: resolution: {integrity: sha512-JQHZ2QMW6l3aH/j6xCqQThY/9OH4D/9ls34cgkUBiEeocRTU04tHfKPBsUK1PqZCUQM7GiA0IIXJSuXHI64Kbg==} engines: {node: '>=0.8'} + cluster-key-slot@1.1.1: + resolution: {integrity: sha512-rwHwUfXL40Chm1r08yrhU3qpUvdVlgkKNeyeGPOxnW8/SyVDvgRaed/Uz54AqWNaTCAThlj6QAs3TZcKI0xDEw==} + engines: {node: '>=0.10.0'} + color-convert@2.0.1: resolution: {integrity: sha512-RRECPsj7iu/xb5oKYcsFHSppFNnsj/52OVTRKb4zP5onXwVF3zVmmToNcOfGC+CRDpfK/U584fMg38ZHCaElKQ==} engines: {node: '>=7.0.0'} @@ -1908,6 +1927,15 @@ packages: csstype@3.2.3: resolution: {integrity: sha512-z1HGKcYy2xA8AGQfwrn0PAy+PB7X/GSj3UVJW9qKyn43xWa+gl5nXmU4qqLMRzWVLFC8KusUX8T/0kCiOYpAIQ==} + debug@4.3.7: + resolution: {integrity: sha512-Er2nc/H7RrMXZBFCEim6TCmMk02Z8vLC2Rbi1KEBggpo0fS6l0S1nnapwmIi3yW/+GOJap1Krg4w0Hg80oCqgQ==} + engines: {node: '>=6.0'} + peerDependencies: + supports-color: '*' + peerDependenciesMeta: + supports-color: + optional: true + debug@4.4.3: resolution: {integrity: sha512-RGwwWnwQvkVfavKVt22FGLw+xYSdzARwm0ru6DhTVA3umU5hZc28V3kO4stgYryrTlLpuvgI9GiijltAjNbcqA==} engines: {node: '>=6.0'} @@ -1935,6 +1963,10 @@ packages: defu@6.1.7: resolution: {integrity: sha512-7z22QmUWiQ/2d0KkdYmANbRUVABpZ9SNYyH5vx6PZ+nE5bcC0l7uFvEfHlyld/HcGBFTL536ClDt3DEcSlEJAQ==} + denque@2.1.0: + resolution: {integrity: sha512-HVQE3AAb/pxF8fQAoiqpvg9i3evqug3hoiwakOyZAwJm+6vZehbkYXZ0l4JxS+I3QxM97v5aaRNhj8v5oBhekw==} + engines: {node: '>=0.10'} + depd@2.0.0: resolution: {integrity: sha512-g7nH6P6dyDioJogAAGprGpCtVImJhpPk/roCzdb3fIh61/s/nPsfR6onyMwkCAR/OlC3yBC0lESvUoQEAssIrw==} engines: {node: '>= 0.8'} @@ -2200,6 +2232,10 @@ packages: inherits@2.0.4: resolution: {integrity: sha512-k/vGaX4/Yla3WzyMCvTQOXYeIHvqOKtnqBduzTHpzpQZzAskKMhZ2K+EnBiSM9zGSoIFeMpXKxa4dYeZIQqewQ==} + ioredis@5.11.1: + resolution: {integrity: sha512-ehuGcf94bQXhfagULNXrJdfnWO38v070jxSx/qE87Kjzmu2fU7ro5EFAb+OPituLqgfyuQaym5DlrNydW2sJ9A==} + engines: {node: '>=12.22.0'} + ipaddr.js@1.9.1: resolution: {integrity: sha512-0KI/607xoxSToH7GjN1FfSbLoU0+btTicjsQSWQlh/hZykN8KpmMf7uYwPW3R+akZ6R/w18ZlXSHBYXiYUPO3g==} engines: {node: '>= 0.10'} @@ -2452,6 +2488,9 @@ packages: resolution: {integrity: sha512-J6l92tKHX6w8Jy5nO1Vuc01NoIiRGi/d6qBKVxh+IQ8Cr3b6HbVNfKiF8ZpFKufTwpwxMmce2W3iQZ861ZRyTg==} engines: {node: '>=18'} + notepack.io@3.0.1: + resolution: {integrity: sha512-TKC/8zH5pXIAMVQio2TvVDTtPRX+DJPHDqjRbxogtFiByHyzKmy96RA0JtCQJ+WouyyL4A10xomQzgbUT+1jCg==} + nypm@0.6.8: resolution: {integrity: sha512-Q9K4Diu6l5u6xJQogeFSs/zKtyMSgFKFtRQV+tHP4kL7KPm2grpBU0dFIwFaXwNxN0MtfKWc43VpCugAa+LPsw==} engines: {node: '>=18'} @@ -2620,6 +2659,14 @@ packages: resolution: {integrity: sha512-GDhwkLfywWL2s6vEjyhri+eXmfH6j1L7JE27WhqLeYzoh/A3DBaYGEj2H/HFZCn/kMfim73FXxEJTw06WtxQwg==} engines: {node: '>= 14.18.0'} + redis-errors@1.2.0: + resolution: {integrity: sha512-1qny3OExCf0UvUV/5wpYKf2YwPcOqXzkwKKSmKHiE6ZMQs5heeE/c8eXK+PNllPvmjgAbfnsbpkGZWy8cBpn9w==} + engines: {node: '>=4'} + + redis-parser@3.0.0: + resolution: {integrity: sha512-DJnGAeenTdpMEH6uAJRK/uiyEIH9WVsUmoLwzudwGJUwZPp80PDBWPHXSAGNPwNvIXAbe7MSUB1zQFugFml66A==} + engines: {node: '>=4'} + reflect-metadata@0.2.2: resolution: {integrity: sha512-urBwgfrvVP/eAyXx4hluJivBKzuEbSQs9rKWCrCkbSxNv8mxPcUZKeuoF3Uy4mJl3Lwprp6yy5/39VWigZ4K6Q==} @@ -2754,6 +2801,9 @@ packages: stackback@0.0.2: resolution: {integrity: sha512-1XMJE5fQo1jGH6Y/7ebnwPOBEkIEnT4QF32d5R1+VXdXveM0IBMJt8zfaxX1P3QhVwrYe+576+jkANtSS2mBbw==} + standard-as-callback@2.1.0: + resolution: {integrity: sha512-qoRRSyROncaz1z0mvYqIE4lCd9p2R90i6GxW3uZv5ucSu8tU7B5HXUP1gG8pVZsYNVaXjk8ClXHPttLyxAL48A==} + statuses@2.0.2: resolution: {integrity: sha512-DvEy55V3DB7uknRo+4iOGT5fP1slR8wQohVdknigZPMpMstaKJQWhwiYBACJE3Ul2pTnATihhBYnRhZQHGBiRw==} engines: {node: '>= 0.8'} @@ -2953,6 +3003,10 @@ packages: ufo@1.6.4: resolution: {integrity: sha512-JFNbkD1Svwe0KvGi8GOeLcP4kAWQ609twvCdcHxq1oSL8svv39ZuSvajcD8B+5D0eL4+s1Is2D/O6KN3qcTeRA==} + uid2@1.0.0: + resolution: {integrity: sha512-+I6aJUv63YAcY9n4mQreLUt0d4lvwkkopDNmpomkAUz0fAkEMV9pRWxN0EjhW1YfRhcuyHg2v3mwddCDW1+LFQ==} + engines: {node: '>= 4.0.0'} + uid@2.0.2: resolution: {integrity: sha512-u3xV3X7uzvi5b1MncmZo3i2Aw222Zk1keqLA1YkHldREkAhAqi65wuPfe7lHx8H/Wzy+8CE7S7uS3jekIM5s8g==} engines: {node: '>=8'} @@ -3716,6 +3770,8 @@ snapshots: optionalDependencies: '@types/node': 26.0.1 + '@ioredis/commands@1.10.0': {} + '@jridgewell/gen-mapping@0.3.13': dependencies: '@jridgewell/sourcemap-codec': 1.5.5 @@ -3968,6 +4024,15 @@ snapshots: '@socket.io/component-emitter@3.1.2': {} + '@socket.io/redis-adapter@8.3.0(socket.io-adapter@2.5.8)': + dependencies: + debug: 4.3.7 + notepack.io: 3.0.1 + socket.io-adapter: 2.5.8 + uid2: 1.0.0 + transitivePeerDependencies: + - supports-color + '@standard-schema/spec@1.1.0': {} '@tokenizer/inflate@0.4.1': @@ -4446,6 +4511,8 @@ snapshots: clone@1.0.4: {} + cluster-key-slot@1.1.1: {} + color-convert@2.0.1: dependencies: color-name: 1.1.4 @@ -4504,6 +4571,10 @@ snapshots: csstype@3.2.3: {} + debug@4.3.7: + dependencies: + ms: 2.1.3 + debug@4.4.3: dependencies: ms: 2.1.3 @@ -4520,6 +4591,8 @@ snapshots: defu@6.1.7: {} + denque@2.1.0: {} + depd@2.0.0: {} destr@2.0.5: {} @@ -4900,6 +4973,18 @@ snapshots: inherits@2.0.4: {} + ioredis@5.11.1: + dependencies: + '@ioredis/commands': 1.10.0 + cluster-key-slot: 1.1.1 + debug: 4.4.3 + denque: 2.1.0 + redis-errors: 1.2.0 + redis-parser: 3.0.0 + standard-as-callback: 2.1.0 + transitivePeerDependencies: + - supports-color + ipaddr.js@1.9.1: {} is-arrayish@0.2.1: {} @@ -5105,6 +5190,8 @@ snapshots: node-releases@2.0.50: {} + notepack.io@3.0.1: {} + nypm@0.6.8: dependencies: citty: 0.2.2 @@ -5258,6 +5345,12 @@ snapshots: readdirp@4.1.2: {} + redis-errors@1.2.0: {} + + redis-parser@3.0.0: + dependencies: + redis-errors: 1.2.0 + reflect-metadata@0.2.2: {} require-from-string@2.0.2: {} @@ -5460,6 +5553,8 @@ snapshots: stackback@0.0.2: {} + standard-as-callback@2.1.0: {} + statuses@2.0.2: {} std-env@3.10.0: {} @@ -5624,6 +5719,8 @@ snapshots: ufo@1.6.4: {} + uid2@1.0.0: {} + uid@2.0.2: dependencies: '@lukeed/csprng': 1.1.0