"use client"; // Messenger data layer. Serves EITHER a local mock (when the Shell isn't configured — the demo // keeps working) OR the live be-crm data door (crm.messenger.*), behind one interface so the UI is // mode-agnostic. DM-vs-group + who-can-chat are enforced server-side by IIOS/OPA; this is just glue. // // Live contract (be-crm): // query crm.messenger.directory { kind, query?, limit } -> DirectoryEntry[] // query crm.messenger.conversation.list {} -> ConversationSummary[] // cmd crm.messenger.conversation.open { participantIds[], membership?, subject? } -> { threadId, ... } // query crm.messenger.history { threadId } -> MessengerMessage[] // cmd crm.messenger.send { threadId, content } -> MessengerMessage // cmd crm.messenger.participant.add { threadId, userId } // // v1 uses REST + polling for the live stream; v2 layers the IIOS MessageSocket (messenger-socket.tsx) // on top for live messages, typing, read receipts, and reactions. import { useCallback, useEffect, useMemo, useRef, useState } from "react"; import { useAppShell, useAuth, useQuery } from "@abe-kap/appshell-sdk/react"; import type { AnnotationEvent, AnnotationGroup } from "@insignia/iios-kernel-client"; import { isShellConfigured } from "./appshell"; import { useMessengerSocket } from "./messenger-socket"; export type Membership = "dm" | "group"; export interface UiPerson { id: string; name: string; kind: "staff" | "customer" } export interface UiConversation { threadId: string; title: string; subject: string | null; membership: Membership | null; participants: string[]; unread: number; lastMessage?: string; lastAt?: string; } export interface UiReaction { emoji: string; count: number; mine: boolean } export interface UiMessage { id: string; actorId: string | null; senderId?: string | null; text: string; at: string; mine: boolean; parentInteractionId?: string | null; reactions?: UiReaction[]; } interface DirectoryDTO { id: string; displayName: string; kind: "staff" | "customer" } interface ConversationDTO { threadId: string; subject: string | null; membership: Membership | null; participants: string[]; unread: number; lastMessage?: string; lastAt?: string; } interface MessageDTO { interactionId: string; actorId: string | null; kind: string; occurredAt: string; text: string | null } const SHELL = isShellConfigured(); const POLL_MS = 4000; const TYPING_TTL_MS = 3500; const shortId = (id: string) => id.replace(/^(pp_|cust_)/, "").slice(0, 6); /** Turn the kernel's generic annotation aggregates into reaction chips. `users` may hold user or * actor ids depending on the source, so `mine` is best-effort; a fresh annotation event corrects it. */ export function toReactions(annotations: AnnotationGroup[] | undefined, myId?: string): UiReaction[] { if (!annotations) return []; return annotations .filter((a) => a.type === "reaction" && a.users.length > 0) .map((a) => ({ emoji: a.value, count: a.users.length, mine: !!myId && a.users.includes(myId) })); } function applyAnnotation(prev: UiReaction[] | undefined, e: AnnotationEvent, myId?: string): UiReaction[] { const base = (prev ?? []).filter((r) => r.emoji !== e.value); if (e.type !== "reaction" || e.users.length === 0) return base; return [...base, { emoji: e.value, count: e.users.length, mine: !!myId && e.users.includes(myId) }]; } /* ======================================================================== */ /* Public hooks */ /* ======================================================================== */ export interface MessengerData { live: boolean; loading: boolean; error: string | null; directory: UiPerson[]; conversations: UiConversation[]; nameOf: (id: string) => string; openConversation: (participantIds: string[], opts?: { membership?: Membership; subject?: string }) => Promise; refetch: () => void; } export interface ThreadData { loading: boolean; error: string | null; messages: UiMessage[]; send: (content: string, opts?: { parentInteractionId?: string }) => Promise; react: (interactionId: string, emoji: string) => void; typingUserIds: string[]; seenIds: Set; refetch: () => void; } export function useMessengerData(): MessengerData { return SHELL ? useLiveMessenger() : useMockMessenger(); } export function useThread(threadId: string): ThreadData { return SHELL ? useLiveThread(threadId) : useMockThread(threadId); } /* ======================================================================== */ /* Live implementation (be-crm data door + IIOS socket) */ /* ======================================================================== */ function useLiveMessenger(): MessengerData { const { sdk } = useAppShell(); const { user } = useAuth(); const socket = useMessengerSocket(); const myId = user?.id; const dirQ = useQuery("crm.messenger.directory", { kind: "all", limit: 100 }); const convQ = useQuery("crm.messenger.conversation.list", {}); const directory: UiPerson[] = useMemo( () => (dirQ.data ?? []).map((d) => ({ id: d.id, name: d.displayName, kind: d.kind })), [dirQ.data], ); const nameById = useMemo(() => Object.fromEntries(directory.map((p) => [p.id, p.name])), [directory]); const nameOf = useCallback((id: string) => nameById[id] ?? `User ${shortId(id)}`, [nameById]); // Live sidebar previews: patch lastMessage/lastAt the instant a message arrives on any thread, // then reconcile authoritative unread/order with a debounced refetch. const [previews, setPreviews] = useState>({}); const refetchRef = useRef(convQ.refetch); refetchRef.current = convQ.refetch; useEffect(() => { if (!socket) return; let timer: ReturnType | null = null; const off = socket.onAnyMessage((threadId, m) => { setPreviews((p) => ({ ...p, [threadId]: { lastMessage: m.text, lastAt: m.at } })); if (timer) clearTimeout(timer); timer = setTimeout(() => refetchRef.current(), 600); }); return () => { off(); if (timer) clearTimeout(timer); }; }, [socket]); const conversations: UiConversation[] = useMemo( () => (convQ.data ?? []).map((c) => shape(c, nameOf, myId, previews[c.threadId])), [convQ.data, nameOf, myId, previews], ); const refetch = useCallback(() => { dirQ.refetch(); convQ.refetch(); }, [dirQ, convQ]); const openConversation = useCallback(async (participantIds: string[], opts?: { membership?: Membership; subject?: string }) => { const res = (await sdk.command("crm.messenger.conversation.open", { participantIds, ...(opts?.membership ? { membership: opts.membership } : {}), ...(opts?.subject ? { subject: opts.subject } : {}), })) as { threadId: string }; convQ.refetch(); return res.threadId; }, [sdk, convQ]); return { live: true, loading: dirQ.loading || convQ.loading, error: (dirQ.error ?? convQ.error)?.message ?? null, directory, conversations, nameOf, openConversation, refetch, }; } function useLiveThread(threadId: string): ThreadData { const { sdk } = useAppShell(); const socket = useMessengerSocket(); const socketReady = socket?.ready ?? false; const q = useQuery("crm.messenger.history", { threadId }); const [socketMsgs, setSocketMsgs] = useState([]); const [myActorId, setMyActorId] = useState(null); const myActorIdRef = useRef(null); myActorIdRef.current = myActorId; const [typing, setTyping] = useState>({}); // userId -> expiry ts const [seenIds, setSeenIds] = useState>(new Set()); const myId = socket?.myUserId; // REST poll — the fallback whenever the live socket isn't connected. const refetchRef = useRef(q.refetch); refetchRef.current = q.refetch; useEffect(() => { if (socketReady) return; const t = setInterval(() => refetchRef.current(), POLL_MS); return () => clearInterval(t); }, [socketReady, threadId]); // Socket (primary): load history + subscribe to live messages, typing, receipts, reactions. useEffect(() => { if (!socket || !socketReady) return; let alive = true; setSocketMsgs([]); setSeenIds(new Set()); setTyping({}); void socket.openThread(threadId).then((hist) => { if (alive) setSocketMsgs(hist); }).catch(() => {}); const offMsg = socket.subscribe(threadId, (m) => setSocketMsgs((l) => (l.some((x) => x.id === m.id) ? l : [...l, m])), ); const offTyping = socket.onTyping(threadId, (userId) => setTyping((t) => ({ ...t, [userId]: Date.now() + TYPING_TTL_MS })), ); // Receipts are a global stream (no threadId). Count only reads by the OTHER side; seenMine then // narrows to my messages in this thread. const offReceipt = socket.onReceipt((e) => { if (e.actorId === myActorIdRef.current) return; setSeenIds((s) => (s.has(e.interactionId) ? s : new Set(s).add(e.interactionId))); }); const offAnn = socket.onAnnotation(threadId, (e) => setSocketMsgs((l) => l.map((m) => (m.id === e.interactionId ? { ...m, reactions: applyAnnotation(m.reactions, e, myId) } : m))), ); return () => { alive = false; offMsg(); offTyping(); offReceipt(); offAnn(); }; }, [socket, socketReady, threadId, myId]); // Learn my own actor id from a message I sent, so receipts from OTHER actors read as "seen". useEffect(() => { const mine = socketMsgs.find((m) => m.mine && m.actorId); if (mine?.actorId && mine.actorId !== myActorId) setMyActorId(mine.actorId); }, [socketMsgs, myActorId]); // Tell the server I've read the latest message (drives the other side's "seen" tick). useEffect(() => { if (!socket || !socketReady || socketMsgs.length === 0) return; socket.markRead(threadId, socketMsgs[socketMsgs.length - 1].id); }, [socket, socketReady, threadId, socketMsgs]); // Expire stale typing entries. const typingUserIds = useMemo(() => { const now = Date.now(); return Object.entries(typing).filter(([, exp]) => exp > now).map(([u]) => u); }, [typing]); useEffect(() => { if (typingUserIds.length === 0) return; const t = setTimeout(() => setTyping((p) => ({ ...p })), TYPING_TTL_MS); return () => clearTimeout(t); }, [typingUserIds.length, typing]); const restMsgs: UiMessage[] = useMemo( () => (q.data ?? []).map((m) => ({ id: m.interactionId, actorId: m.actorId, senderId: null, text: m.text ?? "", at: m.occurredAt, mine: !!myActorId && m.actorId === myActorId, reactions: [], })), [q.data, myActorId], ); const messages = socketReady ? socketMsgs : restMsgs; // My messages the other side has read (receipts carry the other actor's id). const seenMine = useMemo(() => { const out = new Set(); for (const id of seenIds) if (messages.some((m) => m.id === id && m.mine)) out.add(id); return out; }, [seenIds, messages]); const send = useCallback(async (content: string, opts?: { parentInteractionId?: string }) => { if (socket && socketReady) { await socket.send(threadId, content, opts); // echoes back over the socket as a 'message' event } else { const m = (await sdk.command("crm.messenger.send", { threadId, content })) as MessageDTO; if (m.actorId) setMyActorId(m.actorId); q.refetch(); } }, [socket, socketReady, threadId, sdk, q]); const react = useCallback((interactionId: string, emoji: string) => { if (socket && socketReady) socket.react(threadId, interactionId, emoji); }, [socket, socketReady, threadId]); return { loading: q.loading && !socketReady, error: q.error?.message ?? null, messages, send, react, typingUserIds, seenIds: seenMine, refetch: q.refetch, }; } function shape( c: ConversationDTO, nameOf: (id: string) => string, myId: string | undefined, overlay?: { lastMessage: string; lastAt: string }, ): UiConversation { // A DM's title is the OTHER person — never yourself, and never the raw unknown-id fallback for both. const others = myId ? c.participants.filter((p) => p !== myId) : c.participants; const title = c.subject?.trim() || (c.membership === "group" ? `Group · ${c.participants.length}` : (others.map(nameOf).join(", ") || nameOf(c.participants[0] ?? "") || "Conversation")); const lastMessage = overlay?.lastMessage ?? c.lastMessage; const lastAt = overlay?.lastAt ?? c.lastAt; return { threadId: c.threadId, title, subject: c.subject, membership: c.membership, participants: c.participants, unread: c.unread, ...(lastMessage ? { lastMessage } : {}), ...(lastAt ? { lastAt } : {}), }; } /* ======================================================================== */ /* Mock implementation (no Shell configured — the demo keeps working) */ /* ======================================================================== */ const MOCK_PEOPLE: UiPerson[] = [ { id: "pp_sofia", name: "Sofia Ramirez", kind: "staff" }, { id: "pp_dan", name: "Dan Whitaker", kind: "staff" }, { id: "pp_priya", name: "Priya Nair", kind: "staff" }, { id: "cust_acme", name: "Acme Roofing (Client)", kind: "customer" }, { id: "cust_globex", name: "Globex Homes (Client)", kind: "customer" }, ]; interface MockThread { threadId: string; membership: Membership; subject: string | null; participants: string[]; messages: UiMessage[] } const now = () => new Date().toISOString(); let MOCK_SEQ = 100; // A tiny module-level store both mock hooks share, with a subscribe-on-change so the // conversation list and the open thread stay in sync (no globalThis, no render writes). const MOCK_STORE = new Map([ ["th_mock_1", { threadId: "th_mock_1", membership: "dm", subject: null, participants: ["me", "pp_sofia"], messages: [{ id: "m1", actorId: "pp_sofia", text: "Can you review the Henderson estimate?", at: now(), mine: false, reactions: [] }] }], ["th_mock_2", { threadId: "th_mock_2", membership: "group", subject: "Storm response — East side", participants: ["me", "pp_dan", "pp_priya"], messages: [{ id: "m2", actorId: "pp_dan", text: "Crew is rolling out at 7.", at: now(), mine: false, reactions: [] }] }], ]); const mockListeners = new Set<() => void>(); const notifyMock = () => mockListeners.forEach((l) => l()); function useMockSubscription(): void { const [, setV] = useState(0); useEffect(() => { const l = () => setV((n) => n + 1); mockListeners.add(l); return () => { mockListeners.delete(l); }; }, []); } function useMockMessenger(): MessengerData { useMockSubscription(); const nameById = useMemo(() => Object.fromEntries(MOCK_PEOPLE.map((p) => [p.id, p.name])), []); const nameOf = useCallback((id: string) => nameById[id] ?? `User ${shortId(id)}`, [nameById]); const conversations: UiConversation[] = [...MOCK_STORE.values()].map((t) => { const last = t.messages[t.messages.length - 1]; return { threadId: t.threadId, title: t.subject || t.participants.filter((p) => p !== "me").map(nameOf).join(", ") || "Conversation", subject: t.subject, membership: t.membership, participants: t.participants, unread: 0, ...(last ? { lastMessage: last.text, lastAt: last.at } : {}), }; }); const openConversation = useCallback(async (participantIds: string[], opts?: { membership?: Membership; subject?: string }) => { const membership = opts?.membership ?? (participantIds.length === 1 ? "dm" : "group"); const threadId = `th_mock_${MOCK_SEQ++}`; MOCK_STORE.set(threadId, { threadId, membership, subject: opts?.subject ?? null, participants: ["me", ...participantIds], messages: [] }); notifyMock(); return threadId; }, []); return { live: false, loading: false, error: null, directory: MOCK_PEOPLE, conversations, nameOf, openConversation, refetch: () => {} }; } function useMockThread(threadId: string): ThreadData { useMockSubscription(); const thread = MOCK_STORE.get(threadId); const send = useCallback(async (content: string, opts?: { parentInteractionId?: string }) => { const t = MOCK_STORE.get(threadId); if (t) { t.messages = [...t.messages, { id: `m_${MOCK_SEQ++}`, actorId: "me", text: content, at: now(), mine: true, reactions: [], ...(opts?.parentInteractionId ? { parentInteractionId: opts.parentInteractionId } : {}), }]; notifyMock(); } }, [threadId]); const react = useCallback((interactionId: string, emoji: string) => { const t = MOCK_STORE.get(threadId); if (!t) return; t.messages = t.messages.map((m) => { if (m.id !== interactionId) return m; const has = (m.reactions ?? []).find((r) => r.emoji === emoji); const reactions = has ? (m.reactions ?? []).filter((r) => r.emoji !== emoji) : [...(m.reactions ?? []), { emoji, count: 1, mine: true }]; return { ...m, reactions }; }); notifyMock(); }, [threadId]); return { loading: false, error: null, messages: thread?.messages ?? [], send, react, typingUserIds: [], seenIds: new Set(), refetch: notifyMock, }; }