lynkeduppro-crm/src/lib/crm-messaging-adapter.ts

// 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.

import type {
  Conversation,
  Membership,
  Message,
  MessageEvent,
  MessagingAdapter,
  SendOpts,
  Unsubscribe,
} from "@insignia/iios-messaging-ui";

/** The imperative appshell data door (useAppShell().sdk). Typed structurally so we don't couple 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>;
}

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 POLL_MS = 4000;

interface Poll {
  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 polls = new Map<string, Poll>();

  constructor(private readonly sdk: DataDoor, private readonly me: string) {}

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

  async listConversations(): Promise<Conversation[]> {
    const [convs, names] = await Promise.all([
      this.sdk.query<ConversationDTO[]>("crm.messenger.conversation.list", {}),
      this.directory(),
    ]);
    return convs.map((c) => this.toConversation(c, names));
  }

  async openThread(p: { participantIds: string[]; membership?: Membership; subject?: string }): Promise<{ threadId: string }> {
    const res = await this.sdk.command<{ threadId: string }>("crm.messenger.conversation.open", {
      participantIds: p.participantIds,
      ...(p.membership ? { membership: p.membership } : {}),
      ...(p.subject ? { subject: p.subject } : {}),
    });
    return { threadId: res.threadId };
  }

  async history(threadId: string): Promise<Message[]> {
    const msgs = await this.sdk.query<MessageDTO[]>("crm.messenger.history", { threadId });
    return msgs.map((m) => this.toMessage(m));
  }

  async send(threadId: string, content: string, opts?: SendOpts): Promise<Message> {
    const m = await this.sdk.command<MessageDTO>("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.
    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<void> => {
        const p = this.polls.get(threadId);
        if (!p) return;
        try {
          const msgs = await this.sdk.query<MessageDTO[]>("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);
    }
    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);
        this.polls.delete(threadId);
      }
    };
  }

  sendTyping(): void {
    // No BFF verb for typing; realtime is a follow-up (the IIOS socket). No-op keeps the contract.
  }

  async markRead(): Promise<void> {
    // No BFF verb for read receipts here; follow-up via the socket. No-op resolves the contract.
  }

  // ── mapping ────────────────────────────────────────────────────
  private async directory(): Promise<Map<string, string>> {
    if (!this.names) {
      const dir = await this.sdk.query<DirectoryDTO[]>("crm.messenger.directory", { kind: "all", limit: 200 });
      this.names = new Map(dir.map((d) => [d.id, d.displayName]));
    }
    return this.names;
  }

  private toConversation(c: ConversationDTO, names: Map<string, string>): Conversation {
    const others = c.participants.filter((p) => p !== this.me);
    const title =
      c.subject?.trim() ||
      others.map((id) => names.get(id) ?? id).join(", ") ||
      "Conversation";
    return {
      threadId: c.threadId,
      title,
      subject: c.subject,
      membership: c.membership,
      participants: c.participants,
      unread: c.unread,
      ...(c.lastMessage ? { lastMessage: c.lastMessage } : {}),
      ...(c.lastAt ? { lastAt: c.lastAt } : {}),
    };
  }

  private toMessage(m: MessageDTO): Message {
    return {
      id: m.interactionId,
      actorId: m.actorId,
      text: m.text ?? "",
      at: m.occurredAt,
    };
  }
}