iios/packages/iios-messaging-ui/src/adapters/kernel-client.ts

// Transport adapter: implements MessagingAdapter over @insignia/iios-kernel-client
// (browser → IIOS directly, token-in). This is the "plug into any app with a token"
// path — the same transport chat-web and support-sdk use.
//
// It lives in adapters/ (the ONLY layer allowed to import a transport) and ships from
// its own subpath, so core UI never pulls socket code it can't use.

import { MessageSocket, RestClient } from '@insignia/iios-kernel-client';
import type {
  AnnotationEvent,
  DiscoveredThread,
  Message as KernelMessage,
  MessageEvents,
  OpenThreadResult,
  ReceiptEvent,
  ThreadSummary,
  TypingEvent,
} from '@insignia/iios-kernel-client';
import type { MessagingAdapter } from '../adapter';
import type {
  ChannelSummary,
  ChannelVisibility,
  Conversation,
  CreateChannelInput,
  Membership,
  Message,
  MessageEvent,
  Reaction,
  SendOpts,
  Unsubscribe,
} from '../types';

/** The slice of MessageSocket the adapter needs — the real facade satisfies it; tests inject a fake. */
export interface SocketPort {
  openThread(threadId?: string, opts?: { membership?: string; creatorRole?: string; subject?: string }): Promise<OpenThreadResult>;
  sendMessage(threadId: string, content: string, opts?: { parentInteractionId?: string; mentions?: string[] }): Promise<KernelMessage>;
  react(threadId: string, interactionId: string, value: string): Promise<void>;
  markRead(threadId: string, interactionId: string): Promise<{ ok: boolean }>;
  typing(threadId: string): void;
  on<E extends keyof MessageEvents>(event: E, handler: MessageEvents[E]): () => void;
}

/** The slice of RestClient the adapter needs. */
export interface RestPort {
  listThreads(filter?: { metadata?: Record<string, string> }): Promise<ThreadSummary[]>;
  createThread(opts: { membership?: string; creatorRole?: string; subject?: string; metadata?: Record<string, unknown> }): Promise<{ threadId: string }>;
  addParticipant(threadId: string, userId: string): Promise<void>;
  discoverThreads(filter?: { metadata?: Record<string, string> }): Promise<DiscoveredThread[]>;
  leaveThread(threadId: string): Promise<{ threadId: string; participantCount: number }>;
}

export interface KernelClientAdapterConfig {
  /**
   * The current user's id in IIOS's `senderId` space (email/username), from the host's auth —
   * NEVER inferred from message history. This is exactly `currentActorId()`, and it's the bug the
   * conformance suite kills: identity comes from the session, not from a message you happened to send.
   */
  currentUserId: string;
  socket: SocketPort;
  rest: RestPort;
  /** Optional opaque metadata filter for the conversation list (e.g. { source: 'crm-messenger' }). */
  threadFilter?: Record<string, string>;
}

const REACTION = 'reaction';

export class KernelClientAdapter implements MessagingAdapter {
  private readonly me: string;
  private readonly socket: SocketPort;
  private readonly rest: RestPort;
  private readonly threadFilter?: Record<string, string>;

  /** Per-thread UI subscribers. The socket fans server events in; these fan them out. */
  private readonly listeners = new Map<string, Set<(e: MessageEvent) => void>>();
  /** Reaction users per message (messageId → emoji → userSet), so an annotation delta becomes a full set. */
  private readonly reactions = new Map<string, Map<string, Set<string>>>();
  private readonly joined = new Set<string>();
  private readonly offs: Array<() => void> = [];

  constructor(cfg: KernelClientAdapterConfig) {
    this.me = cfg.currentUserId;
    this.socket = cfg.socket;
    this.rest = cfg.rest;
    this.threadFilter = cfg.threadFilter;

    this.offs.push(
      this.socket.on('message', (m: KernelMessage) => {
        this.ingestReactions(m);
        this.emit(m.threadId, { kind: 'message', message: this.toMessage(m) });
      }),
    );
    this.offs.push(
      this.socket.on('typing', (e: TypingEvent) => this.emit(e.threadId, { kind: 'typing', userId: e.userId })),
    );
    this.offs.push(
      // Receipts carry no threadId, so fan to every open thread; the UI filters by messageId.
      this.socket.on('receipt', (e: ReceiptEvent) => this.broadcast({ kind: 'receipt', messageId: e.interactionId, actorId: e.actorId })),
    );
    this.offs.push(
      this.socket.on('annotation', (e: AnnotationEvent) => {
        if (e.type !== REACTION) return;
        this.setReactionUsers(e.interactionId, e.value, e.users);
        this.emit(e.threadId, { kind: 'reaction', messageId: e.interactionId, reactions: this.reactionsOf(e.interactionId) });
      }),
    );
  }

  currentActorId(): string {
    return this.me;
  }

  async listConversations(): Promise<Conversation[]> {
    const threads = await this.rest.listThreads(this.threadFilter ? { metadata: this.threadFilter } : undefined);
    return threads.map((t) => this.toConversation(t));
  }

  async openThread(p: { participantIds: string[]; membership?: Membership; subject?: string }): Promise<{ threadId: string }> {
    const membership: Membership = p.membership ?? (p.participantIds.length === 1 ? 'dm' : 'group');
    const { threadId } = await this.rest.createThread({
      membership,
      creatorRole: membership === 'group' ? 'ADMIN' : 'MEMBER',
      ...(p.subject ? { subject: p.subject } : {}),
    });
    // Governance (DM cap, roles) is enforced server-side by IIOS/OPA; a rejected add surfaces up there.
    for (const id of p.participantIds) await this.rest.addParticipant(threadId, id).catch(() => undefined);
    return { threadId };
  }

  async history(threadId: string): Promise<Message[]> {
    const res = await this.socket.openThread(threadId); // joins the thread, so live events start flowing
    this.joined.add(threadId);
    return res.history.map((m) => {
      this.ingestReactions(m);
      return this.toMessage(m);
    });
  }

  async send(threadId: string, content: string, opts?: SendOpts): Promise<Message> {
    // Attachments are intentionally not forwarded here: kernel-client exposes no media/presign yet,
    // and the SDK Attachment carries a display `url`, not a storage `contentRef`. Media is a follow-up
    // (kernel-client media methods + a contentRef on Attachment). Text + reply threading work today.
    const sendOpts = {
      ...(opts?.parentInteractionId ? { parentInteractionId: opts.parentInteractionId } : {}),
      ...(opts?.mentions && opts.mentions.length ? { mentions: opts.mentions } : {}),
    };
    const m = await this.socket.sendMessage(threadId, content, Object.keys(sendOpts).length ? sendOpts : undefined);
    return this.toMessage(m);
  }

  async react(threadId: string, messageId: string, emoji: string): Promise<void> {
    await this.socket.react(threadId, messageId, emoji);
  }

  subscribe(threadId: string, cb: (e: MessageEvent) => void): Unsubscribe {
    if (!this.listeners.has(threadId)) this.listeners.set(threadId, new Set());
    this.listeners.get(threadId)!.add(cb);
    if (!this.joined.has(threadId)) {
      this.joined.add(threadId);
      void this.socket.openThread(threadId).catch(() => this.joined.delete(threadId));
    }
    return () => {
      this.listeners.get(threadId)?.delete(cb);
    };
  }

  sendTyping(threadId: string): void {
    this.socket.typing(threadId);
  }

  async markRead(threadId: string, messageId: string): Promise<void> {
    await this.socket.markRead(threadId, messageId);
  }

  // ── Channels ────────────────────────────────────────────────────
  async browseChannels(): Promise<ChannelSummary[]> {
    const found = await this.rest.discoverThreads({ metadata: { membership: 'channel', visibility: 'public' } });
    return found.map((d) => {
      const bag = (d.metadata as { topic?: string; visibility?: string } | null) ?? {};
      const visibility: ChannelVisibility = bag.visibility === 'private' ? 'private' : 'public';
      return {
        threadId: d.threadId,
        name: d.subject ?? 'channel',
        topic: bag.topic ?? null,
        visibility,
        memberCount: d.participantCount,
        joined: d.joined,
      };
    });
  }

  async createChannel(input: CreateChannelInput): Promise<{ threadId: string }> {
    return this.rest.createThread({
      membership: 'channel',
      creatorRole: 'ADMIN',
      subject: input.name,
      metadata: { visibility: input.visibility, ...(input.topic ? { topic: input.topic } : {}) },
    });
  }

  async joinChannel(threadId: string): Promise<void> {
    // Self-join is a governed open_thread; OPA allows it for a public channel.
    await this.socket.openThread(threadId);
    this.joined.add(threadId);
  }

  async leaveChannel(threadId: string): Promise<void> {
    await this.rest.leaveThread(threadId);
    this.joined.delete(threadId);
  }

  /** Detach socket handlers. Not part of the contract — call on teardown to avoid leaks. */
  close(): void {
    for (const off of this.offs) off();
    this.offs.length = 0;
    this.listeners.clear();
  }

  // ── mapping ────────────────────────────────────────────────────
  private toMessage(m: KernelMessage): Message {
    return {
      id: m.id,
      // `senderId` (email/username), NOT the actor id — it matches currentActorId() and is the
      // reliable "is this mine?" field. Never inferred from history.
      actorId: m.senderId ?? null,
      text: m.content ?? '',
      at: m.createdAt,
      parentInteractionId: m.parentInteractionId ?? null,
      reactions: this.reactionsOf(m.id),
    };
  }

  private toConversation(t: ThreadSummary): Conversation {
    const others = t.participants.filter((p) => p !== this.me);
    const membership: Membership | null =
      t.membership === 'dm' || t.membership === 'group' || t.membership === 'channel' ? t.membership : null;
    const topic = (t.metadata as { topic?: string } | null)?.topic;
    return {
      threadId: t.threadId,
      title: t.subject?.trim() || others.join(', ') || 'Conversation',
      subject: t.subject,
      membership,
      participants: [...t.participants],
      unread: t.unread,
      ...(topic != null ? { topic } : {}),
      ...(t.lastMessage ? { lastMessage: t.lastMessage } : {}),
      ...(t.lastAt ? { lastAt: t.lastAt } : {}),
    };
  }

  // ── reaction state ─────────────────────────────────────────────
  private ingestReactions(m: KernelMessage): void {
    for (const a of m.annotations ?? []) {
      if (a.type === REACTION) this.setReactionUsers(m.id, a.value, a.users);
    }
  }

  private setReactionUsers(messageId: string, emoji: string, users: string[]): void {
    let byEmoji = this.reactions.get(messageId);
    if (!byEmoji) {
      byEmoji = new Map();
      this.reactions.set(messageId, byEmoji);
    }
    if (users.length === 0) byEmoji.delete(emoji);
    else byEmoji.set(emoji, new Set(users));
  }

  private reactionsOf(messageId: string): Reaction[] {
    const byEmoji = this.reactions.get(messageId);
    if (!byEmoji) return [];
    const out: Reaction[] = [];
    for (const [emoji, users] of byEmoji) {
      if (users.size > 0) out.push({ emoji, count: users.size, mine: users.has(this.me) });
    }
    return out;
  }

  // ── event fan-out ──────────────────────────────────────────────
  private emit(threadId: string, e: MessageEvent): void {
    this.listeners.get(threadId)?.forEach((cb) => cb(e));
  }

  private broadcast(e: MessageEvent): void {
    for (const set of this.listeners.values()) set.forEach((cb) => cb(e));
  }
}

/** Convenience: build an adapter that talks straight to IIOS with a token. */
export function connectKernelAdapter(opts: {
  serviceUrl: string;
  token: string;
  currentUserId: string;
  threadFilter?: Record<string, string>;
  autoConnect?: boolean;
}): KernelClientAdapter {
  const rest = new RestClient({ serviceUrl: opts.serviceUrl, token: opts.token });
  const socket = new MessageSocket({ serviceUrl: opts.serviceUrl, token: opts.token, autoConnect: opts.autoConnect ?? true });
  return new KernelClientAdapter({
    currentUserId: opts.currentUserId,
    socket,
    rest,
    ...(opts.threadFilter ? { threadFilter: opts.threadFilter } : {}),
  });
}