feat(messaging-ui): KernelClientAdapter — direct-IIOS transport for the SDK

Implements MessagingAdapter over @insignia/iios-kernel-client (browser → IIOS, token-in): listConversations/openThread/history/send/subscribe/typing/markRead /react, mapping kernel Message/ThreadSummary/annotations onto the SDK's neutral types. currentActorId comes from the host's session (senderId space), never inferred from history — the exact bug the conformance suite kills. - lives in adapters/ (transport boundary intact — core stays transport-free) - ships from the ./adapters/kernel-client subpath (kernel-client is an optional peer dep; excluded from the main bundle) - connectKernelAdapter({serviceUrl, token, currentUserId}) convenience - passes the shared adapter conformance suite (12 tests) over the real MessageSocket facade with only the socket.io layer faked - media/attachments deferred (kernel-client has no presign yet) Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-21 17:37:41 +05:30
parent 191c9748c5
commit c5237a237a
5 changed files with 409 additions and 14 deletions
@@ -0,0 +1,257 @@
+// 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,
+  Message as KernelMessage,
+  MessageEvents,
+  OpenThreadResult,
+  ReceiptEvent,
+  ThreadSummary,
+  TypingEvent,
+} from '@insignia/iios-kernel-client';
+import type { MessagingAdapter } from '../adapter';
+import type {
+  Conversation,
+  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>;
+}
+
+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 m = await this.socket.sendMessage(
+      threadId,
+      content,
+      opts?.parentInteractionId ? { parentInteractionId: opts.parentInteractionId } : 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);
+  }
+
+  /** 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 : null;
+    return {
+      threadId: t.threadId,
+      title: t.subject?.trim() || others.join(', ') || 'Conversation',
+      subject: t.subject,
+      membership,
+      participants: [...t.participants],
+      unread: t.unread,
+      ...(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 } : {}),
+  });
+}