feat(messenger): socket realtime in the CRM SDK adapter (parity with old messenger)

CrmMessagingAdapter is now hybrid: BFF for the conversation list, thread
creation, and directory (server-side tenancy); the IIOS MessageSocket
(delegated crm.messenger.realtime token) for everything live — history+join,
send, typing, read receipts, and reaction annotations. Falls back to the 4s
poll only when no socket is available.

- messenger-sdk useRealtimeSocket() opens the MessageSocket from the BFF token
  and passes it to the adapter (adapter rebuilds live once connected)
- socket path maps senderId (userId space) → actorId so 'mine'/seen work
  consistently, killing the actor-vs-user ambiguity in the poll fallback
- reaction annotations tracked per message and re-emitted as full sets

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

src/components/dashboard/messenger-sdk.tsx

@@ -5,8 +5,9 @@
 // UI + messaging logic lives in the SDK. Live path = the be-crm data door (CrmMessagingAdapter);
 // demo path = the SDK's own MockAdapter.
 
-import { useMemo } from "react";
+import { useEffect, useMemo, useState } from "react";
-import { useAppShell, useAuth } from "@abe-kap/appshell-sdk/react";
+import { useAppShell, useAuth, useQuery } from "@abe-kap/appshell-sdk/react";
+import { MessageSocket } from "@insignia/iios-kernel-client";
 import { MessagingProvider, Messenger as SdkMessenger, type MessagingAdapter } from "@insignia/iios-messaging-ui";
 import { MockAdapter } from "@insignia/iios-messaging-ui/adapters/mock";
 import "@insignia/iios-messaging-ui/styles.css";
@@ -14,6 +15,27 @@ import { isShellConfigured } from "@/lib/appshell";
 import { CrmMessagingAdapter, type DataDoor } from "@/lib/crm-messaging-adapter";
 import { PageHead } from "./ui";
 
+interface RealtimeDTO { url: string; audience: string; token?: string }
+
+/** Open the IIOS message socket with the delegated token the BFF mints (crm.messenger.realtime). */
+function useRealtimeSocket(): MessageSocket | null {
+  const rt = useQuery<RealtimeDTO>("crm.messenger.realtime", {});
+  const [socket, setSocket] = useState<MessageSocket | null>(null);
+  const url = rt.data?.url;
+  const token = rt.data?.token;
+  useEffect(() => {
+    if (!url || !token) return;
+    const s = new MessageSocket({ serviceUrl: url, token, autoConnect: false });
+    s.connect();
+    setSocket(s);
+    return () => {
+      s.disconnect();
+      setSocket(null);
+    };
+  }, [url, token]);
+  return socket;
+}
+
 const SHELL = isShellConfigured();
 
 export function MessengerSdk() {
@@ -59,9 +81,11 @@ function DemoHost() {
 function LiveHost() {
   const { sdk } = useAppShell();
   const { user } = useAuth();
+  const socket = useRealtimeSocket();
+  // Rebuilds once the socket connects: the first adapter (no socket) polls; the second runs live.
   const adapter = useMemo<MessagingAdapter | null>(
-    () => (user?.id ? new CrmMessagingAdapter(sdk as unknown as DataDoor, user.id) : null),
+    () => (user?.id ? new CrmMessagingAdapter(sdk as unknown as DataDoor, user.id, socket ?? undefined) : null),
-    [sdk, user?.id],
+    [sdk, user?.id, socket],
   );
   if (!adapter) return <div className="miu-empty">Loading…</div>;
   return (

src/lib/crm-messaging-adapter.ts

@@ -1,10 +1,9 @@
-// The CRM's implementation of the SDK's MessagingAdapter, over the be-crm data door
+// The CRM's implementation of the SDK's MessagingAdapter. HYBRID transport:
-// (appshell `crm.messenger.*`). This is how the CRM consumes @insignia/iios-messaging-ui
+//   • BFF (appshell crm.messenger.*) for the conversation list, thread creation, and directory
-// instead of embedding its own messenger: the SDK renders, this adapter transports.
+//     — these need server-side tenancy/auth.
-//
+//   • IIOS MessageSocket (delegated token from crm.messenger.realtime) for everything live:
-// Data (list/open/history/send) goes through the BFF, which attaches the session + tenancy
+//     history+join, send, typing, read receipts, reactions.
-// server-side. Live updates are polled here for a first cut; realtime (the IIOS socket the CRM
+// When no socket is available (token failed / demo), it degrades to a 4s history poll.
-// already opens in messenger-socket.tsx) can be layered in by emitting into the same registry.
 
 import type {
   Conversation,
@@ -12,11 +11,13 @@ import type {
   Message,
   MessageEvent,
   MessagingAdapter,
+  Reaction,
   SendOpts,
   Unsubscribe,
 } from "@insignia/iios-messaging-ui";
+import type { MessageSocket, Message as KernelMessage } from "@insignia/iios-kernel-client";
 
-/** The imperative appshell data door (useAppShell().sdk). Typed structurally so we don't couple to its class. */
+/** The imperative appshell data door (useAppShell().sdk). Typed structurally, not to its class. */
 export interface DataDoor {
   query<T>(action: string, variables?: Record<string, unknown>): Promise<T>;
   command<T>(action: string, variables?: Record<string, unknown>): Promise<T>;
@@ -30,19 +31,44 @@ interface ConversationDTO {
 interface MessageDTO { interactionId: string; actorId: string | null; kind: string; occurredAt: string; text: string | null }
 
 const POLL_MS = 4000;
+const REACTION = "reaction";
 
-interface Poll {
+interface Poll { seen: Set<string>; primed: boolean; timer: ReturnType<typeof setInterval> | null }
-  cbs: Set<(e: MessageEvent) => void>;
-  seen: Set<string>;
-  primed: boolean;
-  timer: ReturnType<typeof setInterval> | null;
-}
 
 export class CrmMessagingAdapter implements MessagingAdapter {
   private names: Map<string, string> | null = null;
+  private readonly listeners = new Map<string, Set<(e: MessageEvent) => void>>();
   private readonly polls = new Map<string, Poll>();
+  private readonly joined = new Set<string>();
+  /** messageId → emoji → userSet, so a single annotation delta can be re-emitted as a full set. */
+  private readonly reactions = new Map<string, Map<string, Set<string>>>();
 
-  constructor(private readonly sdk: DataDoor, private readonly me: string) {}
+  /** Only present with a socket — the UI hides the reaction affordance without it. */
+  react?: (threadId: string, messageId: string, emoji: string) => Promise<void>;
+
+  constructor(
+    private readonly sdk: DataDoor,
+    private readonly me: string,
+    private readonly socket?: MessageSocket,
+  ) {
+    if (socket) {
+      socket.on("message", (m) => {
+        this.ingestReactions(m);
+        this.emit(m.threadId, { kind: "message", message: this.fromKernel(m) });
+      });
+      socket.on("typing", (e) => this.emit(e.threadId, { kind: "typing", userId: e.userId }));
+      // Receipts carry no threadId → fan to all open threads; the UI filters by messageId.
+      socket.on("receipt", (e) => this.broadcast({ kind: "receipt", messageId: e.interactionId, actorId: e.actorId }));
+      socket.on("annotation", (e) => {
+        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) });
+      });
+      this.react = async (threadId, messageId, emoji) => {
+        await socket.react(threadId, messageId, emoji);
+      };
+    }
+  }
 
   currentActorId(): string {
     return this.me;
@@ -66,65 +92,87 @@ export class CrmMessagingAdapter implements MessagingAdapter {
   }
 
   async history(threadId: string): Promise<Message[]> {
+    if (this.socket) {
+      const res = await this.socket.openThread(threadId); // joins so live events flow
+      this.joined.add(threadId);
+      return res.history.map((m) => {
+        this.ingestReactions(m);
+        return this.fromKernel(m);
+      });
+    }
     const msgs = await this.sdk.query<MessageDTO[]>("crm.messenger.history", { threadId });
-    return msgs.map((m) => this.toMessage(m));
+    return msgs.map((m) => this.fromDto(m));
   }
 
   async send(threadId: string, content: string, opts?: SendOpts): Promise<Message> {
-    const m = await this.sdk.command<MessageDTO>("crm.messenger.send", {
+    if (this.socket) {
-      threadId,
+      const m = await this.socket.sendMessage(
-      content,
+        threadId,
-      ...(opts?.attachment ? { attachment: opts.attachment } : {}),
+        content,
-    });
+        opts?.parentInteractionId ? { parentInteractionId: opts.parentInteractionId } : undefined,
-    const msg = this.toMessage(m);
+      );
-    // Mark it seen so the poll doesn't re-emit our own message on top of useMessages' optimistic row.
+      return this.fromKernel(m);
+    }
+    const m = await this.sdk.command<MessageDTO>("crm.messenger.send", { threadId, content });
+    const msg = this.fromDto(m);
     this.polls.get(threadId)?.seen.add(msg.id);
     return msg;
   }
 
   subscribe(threadId: string, cb: (e: MessageEvent) => void): Unsubscribe {
-    let poll = this.polls.get(threadId);
+    if (!this.listeners.has(threadId)) this.listeners.set(threadId, new Set());
-    if (!poll) {
+    this.listeners.get(threadId)!.add(cb);
-      poll = { cbs: new Set(), seen: new Set(), primed: false, timer: null };
+
-      this.polls.set(threadId, poll);
+    if (this.socket) {
-      const tick = async (): Promise<void> => {
+      if (!this.joined.has(threadId)) {
-        const p = this.polls.get(threadId);
+        this.joined.add(threadId);
-        if (!p) return;
+        void this.socket.openThread(threadId).catch(() => this.joined.delete(threadId));
-        try {
+      }
-          const msgs = await this.sdk.query<MessageDTO[]>("crm.messenger.history", { threadId });
+    } else {
-          for (const m of msgs) {
+      this.startPoll(threadId);
-            if (p.seen.has(m.interactionId)) continue;
-            p.seen.add(m.interactionId);
-            // The first pass just records what's already loaded (useMessages fetched history);
-            // only genuinely new messages after that are pushed to the UI.
-            if (p.primed) p.cbs.forEach((f) => f({ kind: "message", message: this.toMessage(m) }));
-          }
-          p.primed = true;
-        } catch {
-          /* transient BFF error — try again next tick */
-        }
-      };
-      void tick();
-      poll.timer = setInterval(tick, POLL_MS);
     }
-    poll.cbs.add(cb);
+
     return () => {
-      const p = this.polls.get(threadId);
+      const set = this.listeners.get(threadId);
-      if (!p) return;
+      set?.delete(cb);
-      p.cbs.delete(cb);
+      if (set && set.size === 0) {
-      if (p.cbs.size === 0) {
+        this.listeners.delete(threadId);
-        if (p.timer) clearInterval(p.timer);
+        const poll = this.polls.get(threadId);
+        if (poll?.timer) clearInterval(poll.timer);
         this.polls.delete(threadId);
       }
     };
   }
 
-  sendTyping(): void {
+  sendTyping(threadId: string): void {
-    // No BFF verb for typing; realtime is a follow-up (the IIOS socket). No-op keeps the contract.
+    this.socket?.typing(threadId);
   }
 
-  async markRead(): Promise<void> {
+  async markRead(threadId: string, messageId: string): Promise<void> {
-    // No BFF verb for read receipts here; follow-up via the socket. No-op resolves the contract.
+    if (this.socket) await this.socket.markRead(threadId, messageId);
+  }
+
+  // ── polling fallback (no socket) ───────────────────────────────
+  private startPoll(threadId: string): void {
+    if (this.polls.has(threadId)) return;
+    const poll: Poll = { seen: new Set(), primed: false, timer: null };
+    this.polls.set(threadId, poll);
+    const tick = async (): Promise<void> => {
+      if (!this.polls.has(threadId)) return;
+      try {
+        const msgs = await this.sdk.query<MessageDTO[]>("crm.messenger.history", { threadId });
+        for (const m of msgs) {
+          if (poll.seen.has(m.interactionId)) continue;
+          poll.seen.add(m.interactionId);
+          if (poll.primed) this.emit(threadId, { kind: "message", message: this.fromDto(m) });
+        }
+        poll.primed = true;
+      } catch {
+        /* transient — retry next tick */
+      }
+    };
+    void tick();
+    poll.timer = setInterval(tick, POLL_MS);
   }
 
   // ── mapping ────────────────────────────────────────────────────
@@ -138,10 +186,7 @@ export class CrmMessagingAdapter implements MessagingAdapter {
 
   private toConversation(c: ConversationDTO, names: Map<string, string>): Conversation {
     const others = c.participants.filter((p) => p !== this.me);
-    const title =
+    const title = c.subject?.trim() || others.map((id) => names.get(id) ?? id).join(", ") || "Conversation";
-      c.subject?.trim() ||
-      others.map((id) => names.get(id) ?? id).join(", ") ||
-      "Conversation";
     return {
       threadId: c.threadId,
       title,
@@ -154,12 +199,56 @@ export class CrmMessagingAdapter implements MessagingAdapter {
     };
   }
 
-  private toMessage(m: MessageDTO): Message {
+  /** Kernel Message (socket) → SDK Message. actorId = senderId (userId space), matching currentActorId. */
+  private fromKernel(m: KernelMessage): Message {
     return {
-      id: m.interactionId,
+      id: m.id,
-      actorId: m.actorId,
+      actorId: m.senderId ?? null,
-      text: m.text ?? "",
+      text: m.content ?? "",
-      at: m.occurredAt,
+      at: m.createdAt,
+      parentInteractionId: m.parentInteractionId ?? null,
+      reactions: this.reactionsOf(m.id),
     };
   }
+
+  /** BFF DTO (poll fallback) → SDK Message. Note: actorId is IIOS actor-id space here. */
+  private fromDto(m: MessageDTO): Message {
+    return { id: m.interactionId, actorId: m.actorId, text: m.text ?? "", at: m.occurredAt };
+  }
+
+  // ── 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));
+  }
 }