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

// 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 {
  Attachment,
  ChannelSummary,
  ChannelVisibility,
  Conversation,
  CreateChannelInput,
  Membership,
  Message,
  MessageEvent,
  MessagingAdapter,
  Person,
  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, 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>;
}

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; attachment?: { contentRef: string; mimeType: string; sizeBytes: number; filename: string | null } | null }

const POLL_MS = 4000;
const REACTION = "reaction";

interface Poll { 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>();
  /** Cross-thread activity listeners (live unread + in-app notifications). */
  private readonly activity = new Set<(e: { threadId: string; message: Message }) => void>();
  /** 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>>>();

  /** 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);
        void this.toKernelMessage(m).then((message) => {
          this.emit(m.threadId, { kind: "message", message });
          for (const cb of this.activity) cb({ threadId: m.threadId, message });
        });
      });
      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;
  }

  /** Report the foregrounded thread to IIOS presence (suppresses push for what you're viewing). */
  setFocus(threadId: string | null): void {
    this.socket?.focus(threadId);
  }

  /** Fire `cb` for every incoming message across ALL the caller's threads (live unread + toasts). */
  subscribeActivity(cb: (e: { threadId: string; message: Message }) => void): Unsubscribe {
    this.activity.add(cb);
    void this.joinAllThreads();
    return () => {
      this.activity.delete(cb);
    };
  }

  /** Join every thread the user belongs to so their messages arrive over the socket, not just the
   *  open one. Idempotent (the `joined` set guards re-joins). */
  private async joinAllThreads(): Promise<void> {
    if (!this.socket) return;
    try {
      const convs = await this.sdk.query<ConversationDTO[]>("crm.messenger.conversation.list", {});
      for (const c of convs) {
        if (!this.joined.has(c.threadId)) {
          this.joined.add(c.threadId);
          void this.socket.openThread(c.threadId).catch(() => this.joined.delete(c.threadId));
        }
      }
    } catch {
      /* best-effort — activity just won't cover un-joined threads */
    }
  }

  async listConversations(): Promise<Conversation[]> {
    const [convs, names] = await Promise.all([
      this.sdk.query<ConversationDTO[]>("crm.messenger.conversation.list", {}),
      this.directoryMap(),
    ]);
    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[]> {
    if (this.socket) {
      const res = await this.socket.openThread(threadId); // joins so live events flow
      this.joined.add(threadId);
      return Promise.all(
        res.history.map((m) => {
          this.ingestReactions(m);
          return this.toKernelMessage(m);
        }),
      );
    }
    const msgs = await this.sdk.query<MessageDTO[]>("crm.messenger.history", { threadId });
    return Promise.all(msgs.map((m) => this.toDtoMessage(m)));
  }

  async send(threadId: string, content: string, opts?: SendOpts): Promise<Message> {
    const att = opts?.attachment;
    if (this.socket) {
      const sendOpts = {
        ...(opts?.parentInteractionId ? { parentInteractionId: opts.parentInteractionId } : {}),
        ...(opts?.mentions && opts.mentions.length ? { mentions: opts.mentions } : {}),
        ...(att?.contentRef ? { attachment: { contentRef: att.contentRef, mimeType: att.mime, sizeBytes: att.sizeBytes ?? 0 } } : {}),
      };
      const m = await this.socket.sendMessage(threadId, content, Object.keys(sendOpts).length ? sendOpts : undefined);
      const msg = this.fromKernel(m);
      // Reuse the staged attachment (already carries a display URL from upload) for instant render.
      return att ? { ...msg, attachment: att } : msg;
    }
    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 att ? { ...msg, attachment: att } : msg;
  }

  async upload(file: File): Promise<Attachment> {
    const mime = file.type || "application/octet-stream";
    const { objectKey, uploadUrl } = await this.sdk.command<{ objectKey: string; uploadUrl: string }>("crm.media.presignUpload", { mime, sizeBytes: file.size });
    const res = await fetch(uploadUrl, { method: "PUT", body: file });
    if (!res.ok) throw new Error(`Upload failed (${res.status}).`);
    const url = await this.downloadUrl(objectKey, mime);
    return { url, mime, name: file.name, contentRef: objectKey, sizeBytes: file.size };
  }

  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.socket) {
      if (!this.joined.has(threadId)) {
        this.joined.add(threadId);
        void this.socket.openThread(threadId).catch(() => this.joined.delete(threadId));
      }
    } else {
      this.startPoll(threadId);
    }

    return () => {
      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(threadId: string): void {
    this.socket?.typing(threadId);
  }

  async markRead(threadId: string, messageId: string): Promise<void> {
    if (this.socket) await this.socket.markRead(threadId, messageId);
  }

  // ── channels + members (BFF, except join which is a governed socket self-join) ──
  async browseChannels(): Promise<ChannelSummary[]> {
    const rows = await this.sdk.query<Array<{ threadId: string; name: string; topic: string | null; visibility: string; memberCount: number; joined: boolean }>>(
      "crm.messenger.channel.browse",
      {},
    );
    return rows.map((c) => ({
      threadId: c.threadId,
      name: c.name,
      topic: c.topic,
      visibility: (c.visibility === "private" ? "private" : "public") as ChannelVisibility,
      memberCount: c.memberCount,
      joined: c.joined,
    }));
  }

  async createChannel(input: CreateChannelInput): Promise<{ threadId: string }> {
    return this.sdk.command<{ threadId: string }>("crm.messenger.channel.create", {
      name: input.name,
      ...(input.topic ? { topic: input.topic } : {}),
      visibility: input.visibility,
    });
  }

  async joinChannel(threadId: string): Promise<void> {
    // Governed public self-join over the socket (the BFF has no join verb; OPA enforces it).
    if (!this.socket) throw new Error("joining a channel needs a live connection");
    await this.socket.openThread(threadId);
    this.joined.add(threadId);
  }

  async leaveChannel(threadId: string): Promise<void> {
    await this.sdk.command("crm.messenger.channel.leave", { threadId });
  }

  async addMember(threadId: string, userId: string): Promise<void> {
    await this.sdk.command("crm.messenger.participant.add", { threadId, userId });
  }

  async removeMember(threadId: string, userId: string): Promise<void> {
    await this.sdk.command("crm.messenger.participant.remove", { threadId, userId });
  }

  async renameConversation(threadId: string, subject: string): Promise<void> {
    await this.sdk.command("crm.messenger.group.rename", { threadId, subject });
  }

  async listMembers(threadId: string): Promise<Person[]> {
    const rows = await this.sdk.query<Array<{ userId: string; displayName: string; role: string }>>("crm.messenger.members", { threadId });
    return rows.map((r) => ({ id: r.userId, name: r.displayName, kind: r.role === "CUSTOMER" ? "customer" : "staff" }));
  }

  // ── 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);
  }

  /** The org directory — people you can start a DM/group with. Drives the "New message" picker. */
  async directory(): Promise<Person[]> {
    const dir = await this.sdk.query<DirectoryDTO[]>("crm.messenger.directory", { kind: "all", limit: 100 });
    return dir.map((d) => ({ id: d.id, name: d.displayName, kind: d.kind }));
  }

  // ── mapping ────────────────────────────────────────────────────
  private async directoryMap(): Promise<Map<string, string>> {
    if (!this.names) {
      this.names = new Map((await this.directory()).map((p) => [p.id, p.name]));
    }
    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 } : {}),
    };
  }

  /** Kernel Message (socket) → SDK Message. actorId = senderId (userId space), matching currentActorId. */
  private fromKernel(m: KernelMessage): Message {
    return {
      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 };
  }

  // ── attachments ────────────────────────────────────────────────
  /** A short-lived signed URL to display/download a stored object. */
  private async downloadUrl(contentRef: string, mime?: string): Promise<string> {
    const { url } = await this.sdk.command<{ url: string }>("crm.media.presignDownload", { contentRef, ...(mime ? { mime } : {}) });
    return url;
  }

  private async resolveAttachment(a: { contentRef: string; mimeType: string; sizeBytes: number; filename?: string | null } | null | undefined): Promise<Attachment | undefined> {
    if (!a?.contentRef) return undefined;
    const url = await this.downloadUrl(a.contentRef, a.mimeType);
    return { url, mime: a.mimeType, name: a.filename ?? "attachment", contentRef: a.contentRef, sizeBytes: a.sizeBytes };
  }

  private async toKernelMessage(m: KernelMessage): Promise<Message> {
    const base = this.fromKernel(m);
    const att = await this.resolveAttachment(m.attachment ?? null);
    return att ? { ...base, attachment: att } : base;
  }

  private async toDtoMessage(m: MessageDTO): Promise<Message> {
    const base = this.fromDto(m);
    const att = await this.resolveAttachment(m.attachment ?? null);
    return att ? { ...base, attachment: att } : base;
  }

  // ── 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));
  }
}