iios/packages/iios-kernel-client/src/message-socket.ts

import { io } from 'socket.io-client';
import type { Message, OpenThreadResult, MessageEvents, SocketLike } from './types';

export interface MessageSocketConfig {
  serviceUrl: string;
  token: string;
  autoConnect?: boolean;
}

/**
 * Framework-agnostic facade over the `/message` Socket.io namespace. No socket.io
 * types leak out; RPCs use emitWithAck. It tracks EVERY joined thread and re-opens
 * all of them on reconnect (the docs' disconnect/reconnect requirement), so a UI
 * that watches multiple conversations keeps receiving live messages after a drop.
 */
export class MessageSocket {
  private readonly socket: SocketLike;
  private readonly joined = new Set<string>();

  constructor(config: MessageSocketConfig, socket?: SocketLike) {
    this.socket =
      socket ??
      (io(`${config.serviceUrl.replace(/\/$/, '')}/message`, {
        auth: { token: config.token },
        transports: ['websocket'],
        autoConnect: config.autoConnect ?? true,
      }) as unknown as SocketLike);

    // Re-subscribe to every joined thread after a reconnect.
    this.socket.on('connect', () => {
      for (const id of this.joined) void this.socket.emitWithAck('open_thread', { threadId: id });
    });
  }

  connect(): void {
    this.socket.connect();
  }

  disconnect(): void {
    this.socket.disconnect();
  }

  /** Run `handler` on every (re)connect, and immediately if already connected. */
  onConnected(handler: () => void): () => void {
    this.socket.on('connect', handler);
    if (this.socket.connected) handler();
    return () => this.socket.off('connect', handler);
  }

  /** Subscribe to a server event; returns an unsubscribe fn. */
  on<E extends keyof MessageEvents>(event: E, handler: MessageEvents[E]): () => void {
    const fn = handler as (...args: unknown[]) => void;
    this.socket.on(event as string, fn);
    return () => this.socket.off(event as string, fn);
  }

  async openThread(
    threadId?: string,
    opts?: { membership?: string; creatorRole?: string; subject?: string },
  ): Promise<OpenThreadResult> {
    // Timeout (when the transport supports it) so a server error that never acks
    // can't hang the caller forever.
    const ack = this.socket.timeout ? this.socket.timeout(8000) : this.socket;
    const result = (await ack.emitWithAck('open_thread', { threadId, ...opts })) as OpenThreadResult & { error?: string };
    if (result?.error) throw new Error(result.error);
    this.joined.add(result.threadId);
    return result;
  }

  async sendMessage(
    threadId: string,
    content: string,
    opts?: {
      contentRef?: string;
      parentInteractionId?: string;
      mentions?: string[];
      attachment?: { contentRef: string; mimeType: string; sizeBytes: number; checksumSha256?: string };
    },
  ): Promise<Message> {
    return (await this.socket.emitWithAck('send_message', {
      threadId,
      content,
      contentRef: opts?.attachment?.contentRef ?? opts?.contentRef,
      mimeType: opts?.attachment?.mimeType,
      sizeBytes: opts?.attachment?.sizeBytes,
      checksumSha256: opts?.attachment?.checksumSha256,
      parentInteractionId: opts?.parentInteractionId,
      mentions: opts?.mentions, // opaque userId notify-list; the app parses "@", not the kernel
    })) as Message;
  }

  /** Pin a message in the thread (shared, generic annotation type "pin"). */
  async pin(threadId: string, interactionId: string): Promise<void> {
    await this.socket.emitWithAck('annotate', { threadId, interactionId, type: 'pin', value: '' });
  }

  /** Save a message for myself (personal, generic annotation type "save"). */
  async save(threadId: string, interactionId: string): Promise<void> {
    await this.socket.emitWithAck('annotate', { threadId, interactionId, type: 'save', value: '' });
  }

  /** Toggle an emoji reaction on a message (a generic annotation of type "reaction"). */
  async react(threadId: string, interactionId: string, value: string): Promise<void> {
    await this.socket.emitWithAck('annotate', { threadId, interactionId, type: 'reaction', value });
  }

  async markRead(threadId: string, interactionId: string): Promise<{ ok: boolean }> {
    return (await this.socket.emitWithAck('read', { threadId, interactionId })) as { ok: boolean };
  }

  typing(threadId: string): void {
    this.socket.emit('typing', { threadId });
  }

  /** Tell the server which thread is in the foreground (or null when blurred) — drives presence. */
  focus(threadId: string | null): void {
    this.socket.emit('focus_thread', { threadId });
  }
}