feat(service): P2.3 Socket.io /message gateway + real SessionVerifier

open_thread (create-or-join) / send_message / read / delivered / typing; HS256 token verified on connect (SessionVerifier reuses AppTokenVerifier pattern); OutboxBus bridge propagates REST/ingest messages to rooms (dedup vs inline emit). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-01 01:12:49 +05:30
parent 827fb52f5f
commit 50ef34dd90
6 changed files with 375 additions and 5 deletions
@@ -0,0 +1,111 @@
+import { randomUUID } from 'node:crypto';
+import { Logger } from '@nestjs/common';
+import {
+  ConnectedSocket,
+  MessageBody,
+  OnGatewayConnection,
+  OnGatewayInit,
+  SubscribeMessage,
+  WebSocketGateway,
+  WebSocketServer,
+} from '@nestjs/websockets';
+import { Server, Socket } from 'socket.io';
+import { IIOS_EVENTS } from '@insignia/iios-contracts';
+import { MessageService, type MessagePrincipal } from './message.service';
+import { SessionVerifier } from '../platform/session.verifier';
+import { OutboxBus } from '../outbox/outbox.bus';
+
+interface SocketState {
+  principal: MessagePrincipal;
+}
+
+/**
+ * Realtime messaging over Socket.io `/message` (reuses the support-service
+ * gateway pattern). Token verified synchronously on connect. Send writes via
+ * MessageService and emits inline to the thread room; the OutboxBus bridge
+ * propagates messages persisted elsewhere (REST/ingest) to connected clients.
+ */
+@WebSocketGateway({ namespace: '/message', cors: { origin: true } })
+export class MessageGateway implements OnGatewayInit, OnGatewayConnection {
+  private readonly logger = new Logger(MessageGateway.name);
+  private readonly emittedLocally = new Set<string>();
+
+  @WebSocketServer() server!: Server;
+
+  constructor(
+    private readonly messages: MessageService,
+    private readonly session: SessionVerifier,
+    private readonly bus: OutboxBus,
+  ) {}
+
+  afterInit(): void {
+    this.bus.on(IIOS_EVENTS.messageSent, (payload) => {
+      const ce = payload as { data?: { interactionId?: string; threadId?: string } };
+      const interactionId = ce?.data?.interactionId;
+      const threadId = ce?.data?.threadId;
+      if (!interactionId || !threadId) return;
+      if (this.emittedLocally.delete(interactionId)) return; // already emitted inline
+      void this.messages.getMessageById(interactionId).then((m) => {
+        if (m) this.server.to(threadId).emit('message', m);
+      });
+    });
+  }
+
+  handleConnection(client: Socket): void {
+    try {
+      const token = String(client.handshake.auth?.token ?? '');
+      const principal = this.session.verify(token);
+      client.data = { principal } satisfies SocketState;
+    } catch (err) {
+      this.logger.warn(`rejecting socket: ${(err as Error).message}`);
+      client.disconnect(true);
+    }
+  }
+
+  @SubscribeMessage('open_thread')
+  async openThread(@ConnectedSocket() client: Socket, @MessageBody() body: { threadId?: string }) {
+    const { principal } = client.data as SocketState;
+    const result = await this.messages.openThread(body?.threadId ?? null, principal);
+    await client.join(result.threadId);
+    return result;
+  }
+
+  @SubscribeMessage('send_message')
+  async sendMessage(
+    @ConnectedSocket() client: Socket,
+    @MessageBody() body: { threadId: string; content: string; contentRef?: string },
+  ) {
+    const { principal } = client.data as SocketState;
+    const msg = await this.messages.send(
+      body.threadId,
+      principal,
+      { content: body.content, contentRef: body.contentRef },
+      randomUUID(),
+    );
+    this.emittedLocally.add(msg.id);
+    this.server.to(body.threadId).emit('message', msg);
+    return msg;
+  }
+
+  @SubscribeMessage('read')
+  async read(@ConnectedSocket() client: Socket, @MessageBody() body: { threadId: string; interactionId: string }) {
+    const { principal } = client.data as SocketState;
+    const r = await this.messages.markRead(body.threadId, principal, body.interactionId);
+    this.server.to(body.threadId).emit('receipt', { interactionId: r.interactionId, actorId: r.actorId, kind: 'READ' });
+    return { ok: true };
+  }
+
+  @SubscribeMessage('delivered')
+  async delivered(@ConnectedSocket() client: Socket, @MessageBody() body: { threadId: string; interactionId: string }) {
+    const { principal } = client.data as SocketState;
+    const r = await this.messages.markDelivered(body.threadId, principal, body.interactionId);
+    this.server.to(body.threadId).emit('receipt', { interactionId: r.interactionId, actorId: r.actorId, kind: 'DELIVERED' });
+    return { ok: true };
+  }
+
+  @SubscribeMessage('typing')
+  typing(@ConnectedSocket() client: Socket, @MessageBody() body: { threadId: string }): void {
+    const { principal } = client.data as SocketState;
+    client.to(body.threadId).emit('typing', { threadId: body.threadId, userId: principal.userId });
+  }
+}