diff --git a/src/components/dashboard/messenger-sdk.tsx b/src/components/dashboard/messenger-sdk.tsx index 7e440a3..e7ae9f0 100644 --- a/src/components/dashboard/messenger-sdk.tsx +++ b/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 { useAppShell, useAuth } from "@abe-kap/appshell-sdk/react"; +import { useEffect, useMemo, useState } from "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("crm.messenger.realtime", {}); + const [socket, setSocket] = useState(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( - () => (user?.id ? new CrmMessagingAdapter(sdk as unknown as DataDoor, user.id) : null), - [sdk, user?.id], + () => (user?.id ? new CrmMessagingAdapter(sdk as unknown as DataDoor, user.id, socket ?? undefined) : null), + [sdk, user?.id, socket], ); if (!adapter) return
Loading…
; return ( diff --git a/src/lib/crm-messaging-adapter.ts b/src/lib/crm-messaging-adapter.ts index 899bfb7..c340fdd 100644 --- a/src/lib/crm-messaging-adapter.ts +++ b/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 -// (appshell `crm.messenger.*`). This is how the CRM consumes @insignia/iios-messaging-ui -// instead of embedding its own messenger: the SDK renders, this adapter transports. -// -// Data (list/open/history/send) goes through the BFF, which attaches the session + tenancy -// server-side. Live updates are polled here for a first cut; realtime (the IIOS socket the CRM -// already opens in messenger-socket.tsx) can be layered in by emitting into the same registry. +// The CRM's implementation of the SDK's MessagingAdapter. HYBRID transport: +// • BFF (appshell crm.messenger.*) for the conversation list, thread creation, and directory +// — these need server-side tenancy/auth. +// • IIOS MessageSocket (delegated token from crm.messenger.realtime) for everything live: +// history+join, send, typing, read receipts, reactions. +// When no socket is available (token failed / demo), it degrades to a 4s history poll. 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(action: string, variables?: Record): Promise; command(action: string, variables?: Record): Promise; @@ -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 { - cbs: Set<(e: MessageEvent) => void>; - seen: Set; - primed: boolean; - timer: ReturnType | null; -} +interface Poll { seen: Set; primed: boolean; timer: ReturnType | null } export class CrmMessagingAdapter implements MessagingAdapter { private names: Map | null = null; + private readonly listeners = new Map void>>(); private readonly polls = new Map(); + private readonly joined = new Set(); + /** messageId → emoji → userSet, so a single annotation delta can be re-emitted as a full set. */ + private readonly reactions = new Map>>(); - 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; + + 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 { + 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("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 { - const m = await this.sdk.command("crm.messenger.send", { - threadId, - content, - ...(opts?.attachment ? { attachment: opts.attachment } : {}), - }); - const msg = this.toMessage(m); - // Mark it seen so the poll doesn't re-emit our own message on top of useMessages' optimistic row. + if (this.socket) { + const m = await this.socket.sendMessage( + threadId, + content, + opts?.parentInteractionId ? { parentInteractionId: opts.parentInteractionId } : undefined, + ); + return this.fromKernel(m); + } + const m = await this.sdk.command("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 (!poll) { - poll = { cbs: new Set(), seen: new Set(), primed: false, timer: null }; - this.polls.set(threadId, poll); - const tick = async (): Promise => { - const p = this.polls.get(threadId); - if (!p) return; - try { - const msgs = await this.sdk.query("crm.messenger.history", { threadId }); - for (const m of msgs) { - 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); + if (!this.listeners.has(threadId)) this.listeners.set(threadId, new Set()); + this.listeners.get(threadId)!.add(cb); + + if (this.socket) { + if (!this.joined.has(threadId)) { + this.joined.add(threadId); + void this.socket.openThread(threadId).catch(() => this.joined.delete(threadId)); + } + } else { + this.startPoll(threadId); } - poll.cbs.add(cb); + return () => { - const p = this.polls.get(threadId); - if (!p) return; - p.cbs.delete(cb); - if (p.cbs.size === 0) { - if (p.timer) clearInterval(p.timer); + const set = this.listeners.get(threadId); + set?.delete(cb); + if (set && set.size === 0) { + this.listeners.delete(threadId); + const poll = this.polls.get(threadId); + if (poll?.timer) clearInterval(poll.timer); this.polls.delete(threadId); } }; } - sendTyping(): void { - // No BFF verb for typing; realtime is a follow-up (the IIOS socket). No-op keeps the contract. + sendTyping(threadId: string): void { + this.socket?.typing(threadId); } - async markRead(): Promise { - // No BFF verb for read receipts here; follow-up via the socket. No-op resolves the contract. + async markRead(threadId: string, messageId: string): Promise { + 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 => { + if (!this.polls.has(threadId)) return; + try { + const msgs = await this.sdk.query("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): Conversation { const others = c.participants.filter((p) => p !== this.me); - const title = - c.subject?.trim() || - others.map((id) => names.get(id) ?? id).join(", ") || - "Conversation"; + const title = 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, - actorId: m.actorId, - text: m.text ?? "", - at: m.occurredAt, + id: m.id, + actorId: m.senderId ?? null, + text: m.content ?? "", + 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)); + } }