iios/packages/iios-service/src/messaging/message.gateway.ts

import { randomUUID } from 'node:crypto';
import { Inject, Logger } from '@nestjs/common';
import {
  ConnectedSocket,
  MessageBody,
  OnGatewayConnection,
  OnGatewayDisconnect,
  OnGatewayInit,
  SubscribeMessage,
  WebSocketGateway,
  WebSocketServer,
} from '@nestjs/websockets';
import { Server, Socket } from 'socket.io';
import { IIOS_EVENTS } from '@insignia/iios-contracts';
import { logJson } from '../observability/logger';
import { MessageService, type MessagePrincipal } from './message.service';
import { SessionVerifier } from '../platform/session.verifier';
import { OutboxBus } from '../outbox/outbox.bus';
import { PRESENCE_PORT, type PresencePort } from '../notifications/presence.port';

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, OnGatewayDisconnect {
  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,
    @Inject(PRESENCE_PORT) private readonly presence: PresencePort,
  ) {}

  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
      const correlationId = (payload as { insignia?: { correlationId?: string } })?.insignia?.correlationId;
      void this.messages.getMessageById(interactionId).then((m) => {
        if (m) this.server.to(threadId).emit('message', m);
        logJson('info', 'ws.deliver', { interactionId, threadId, correlationId });
      });
    });
  }

  handleConnection(client: Socket): void {
    try {
      const token = String(client.handshake.auth?.token ?? '');
      const principal = this.session.verify(token);
      // Realtime delegation (least privilege): when IIOS_REALTIME_AUDIENCE is set, the socket
      // accepts ONLY a token minted for that audience (a short-lived `iios-message` delegate) —
      // a full REST/actor token (`iios-core`) or an un-scoped token cannot open the stream. So a
      // leaked socket token can't drive privileged REST, and vice-versa. Unset → no enforcement.
      const required = process.env.IIOS_REALTIME_AUDIENCE?.trim();
      if (required && principal.audience !== required) {
        throw new Error(`realtime audience "${principal.audience ?? '(none)'}" != required "${required}"`);
      }
      client.data = { principal } satisfies SocketState;
    } catch (err) {
      this.logger.warn(`rejecting socket: ${(err as Error).message}`);
      client.disconnect(true);
    }
  }

  handleDisconnect(client: Socket): void {
    // Fire-and-forget: presence is best-effort and must not block the disconnect path.
    void this.presence.clearSocket(client.id).catch((err) => this.logger.warn(`presence clear failed: ${(err as Error).message}`));
  }

  /** The app reports which thread is in the foreground (or null when blurred) — presence. */
  @SubscribeMessage('focus_thread')
  focusThread(@ConnectedSocket() client: Socket, @MessageBody() body: { threadId: string | null }): void {
    const state = client.data as SocketState | undefined;
    if (!state?.principal) return;
    void this.presence
      .setFocus(client.id, state.principal.userId, body?.threadId ?? null)
      .catch((err) => this.logger.warn(`presence set failed: ${(err as Error).message}`));
  }

  @SubscribeMessage('open_thread')
  async openThread(@ConnectedSocket() client: Socket, @MessageBody() body: { threadId?: string; membership?: string; creatorRole?: string; subject?: string }) {
    const { principal } = client.data as SocketState;
    try {
      const result = await this.messages.openThread(body?.threadId ?? null, principal, {
        membership: body?.membership,
        creatorRole: body?.creatorRole,
        subject: body?.subject,
      });
      await client.join(result.threadId);
      return result;
    } catch (err) {
      // Fail to an ACK'd error instead of throwing (which never acks → client hangs on "loading").
      return { error: (err as Error).message ?? 'could not open the conversation' };
    }
  }

  @SubscribeMessage('add_participant')
  async addParticipant(@ConnectedSocket() client: Socket, @MessageBody() body: { threadId: string; userId: string; role?: string }) {
    const { principal } = client.data as SocketState;
    return this.messages.addParticipant(body.threadId, principal, body.userId, body.role);
  }

  @SubscribeMessage('send_message')
  async sendMessage(
    @ConnectedSocket() client: Socket,
    @MessageBody()
    body: { threadId: string; content: string; contentRef?: string; mimeType?: string; sizeBytes?: number; checksumSha256?: string; parentInteractionId?: string; mentions?: string[] },
  ) {
    const { principal } = client.data as SocketState;
    const msg = await this.messages.send(
      body.threadId,
      principal,
      { content: body.content, contentRef: body.contentRef, mimeType: body.mimeType, sizeBytes: body.sizeBytes, checksumSha256: body.checksumSha256 },
      randomUUID(),
      undefined,
      body.parentInteractionId,
      body.mentions,
    );
    this.emittedLocally.add(msg.id);
    this.server.to(body.threadId).emit('message', msg);
    return msg;
  }

  @SubscribeMessage('annotate')
  async annotate(
    @ConnectedSocket() client: Socket,
    @MessageBody() body: { threadId: string; interactionId: string; type: string; value: string },
  ) {
    const { principal } = client.data as SocketState;
    const r = await this.messages.toggleAnnotation(body.interactionId, principal, body.type, body.value);
    // Broadcast the refreshed user list for this (interaction,type,value) so every client re-renders.
    this.server.to(r.threadId).emit('annotation', {
      threadId: r.threadId,
      interactionId: r.interactionId,
      type: r.type,
      value: r.value,
      op: r.op,
      users: r.users,
      userId: principal.userId,
    });
    return r;
  }

  @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 });
  }
}