// Per-topic pub-sub + subscription lifecycle, layered on top of connection.ts's connection-level // pub-sub. Topic.onData/onInfo/etc. are single-slot callbacks exactly like Tinode.onConnect/etc. // (see connection.ts's header comment) -- this module claims them once per topic and fans out to // any number of subscribers, which is what hooks.ts's useSyncExternalStore-based hooks consume. // // Subscription lifecycle mirrors the old app's (verified against its tinode-web.jsx/messages-view.jsx): // - The `me` topic is subscribed once per session, metadata only (no message bodies). // - Every contact gets a one-shot "preview" subscription (withLaterData(1)) so the chat list can // show a last-message preview; it is never explicitly left afterward. // - Opening a conversation subscribes with a full history page. A topic already subscribed (from // its preview prefetch) has its subscribe() short-circuited by the SDK if called again with new // getParams -- so this always leaves first when already subscribed, then resubscribes. import type { Contact, StoredMessage, Topic } from 'tinode-sdk'; import { MESSAGES_PAGE } from '@/lib/config'; import { getTinode, onReauthenticated } from './connection'; const topicListeners = new Map void>>(); const anyTopicListeners = new Set<() => void>(); const wiredTopics = new WeakSet(); const previewPrefetched = new Set(); const messageSnapshots = new Map(); function notifyTopic(topicName: string) { messageSnapshots.delete(topicName); topicListeners.get(topicName)?.forEach((cb) => cb()); anyTopicListeners.forEach((cb) => cb()); } export function subscribeTopicChanges(topicName: string, callback: () => void): () => void { let set = topicListeners.get(topicName); if (!set) { set = new Set(); topicListeners.set(topicName, set); } set.add(callback); return () => set.delete(callback); } /** Fires on ANY topic's change -- used by the chat list, which derives its rows from every contact * at once (`me.contacts()`) rather than tracking one topic in particular. */ export function subscribeAnyTopicChange(callback: () => void): () => void { anyTopicListeners.add(callback); return () => anyTopicListeners.delete(callback); } export interface RawTopicInfo { what: 'read' | 'recv' | 'kp' | 'call'; seq?: number; from?: string; event?: string; payload?: Record; } const infoListeners = new Set<(topicName: string, info: RawTopicInfo) => void>(); /** Raw `{info}` payloads (read/recv/kp/call), forwarded verbatim -- unlike `subscribeAnyTopicChange`/ * `subscribeTopicChanges`, which only signal "something changed, re-read the snapshot" for the * chat-list/message-list hooks, calls.ts needs the actual `event`/`payload` fields off a `what: * "call"` info to drive its own ringing/accept/hang-up state machine -- those aren't reconstructable * from a message snapshot re-read the way a new chat message is. */ export function subscribeTopicInfo(callback: (topicName: string, info: RawTopicInfo) => void): () => void { infoListeners.add(callback); return () => infoListeners.delete(callback); } const goneListeners = new Set<(topicName: string) => void>(); /** Fires once a topic is confirmed gone (group dissolved by anyone, or removed from this account) -- * the SDK's own `Topic#_gone()` clears the topic's public/private/messages and calls * `onDeleteTopic()` (confirmed reading tinode.dev.js). Without ConversationView reacting to this, * a member who stays on a group's conversation screen while someone else dissolves it just keeps * looking at a now-defunct view -- `onMetaDesc`/`onPres`/etc. above only invalidate cached snapshots * for re-render, they don't tell the currently-open screen it needs to navigate away entirely. */ export function subscribeTopicGone(callback: (topicName: string) => void): () => void { goneListeners.add(callback); return () => goneListeners.delete(callback); } function wireTopicCallbacks(topic: Topic) { if (wiredTopics.has(topic)) return; const notify = () => notifyTopic(topic.name); topic.onData = notify; topic.onInfo = (info) => { notify(); if (info) infoListeners.forEach((cb) => cb(topic.name, info)); }; topic.onMetaDesc = notify; // Only ever actually invoked on the `me` topic (the SDK reads `me.onMetaSub`/`me.onSubsUpdated` // specifically, regardless of which topic's subscribe triggered it -- confirmed tinode.dev.js: // 4654-4659) -- wiring it here on every topic is a no-op for anything else, same as onSubsUpdated // right below, which already gets this same treatment. Without this, a brand-new P2P conversation // reached by the other user's `usr...` id (no established topic yet) never tells the chat list a // new contact exists the instant the server assigns its real `p2p...` name on first subscribe -- // the contact stays invisible in the sidebar, and ConversationView's own `contact` lookup (peer // name, online status) keeps missing until some UNRELATED full resync happens to pick it up. topic.onMetaSub = notify; topic.onSubsUpdated = notify; topic.onPres = notify; topic.onAuxUpdated = notify; topic.onAllMessagesReceived = notify; topic.onDeleteTopic = () => { notify(); goneListeners.forEach((cb) => cb(topic.name)); }; wiredTopics.add(topic); } /** Cached per topic; only recomputed when notifyTopic() invalidates it, so useSyncExternalStore's * getSnapshot returns a stable reference between real changes instead of a new array every render. */ export function getMessagesSnapshot(topicName: string): StoredMessage[] { let cached = messageSnapshots.get(topicName); if (!cached) { const topic = getTinode().getTopic(topicName); cached = []; topic.messages((msg) => cached!.push(msg)); messageSnapshots.set(topicName, cached); } return cached; } /** Deduped the same way openConversation/connect are -- subscribing the same `me` topic twice * concurrently is the same non-idempotent-WS-call shape as those (currently reachable from * RequireAuth.tsx/LoginPage.tsx, which don't overlap today, but this is cheap insurance against a * future call site that does, e.g. a manual "reload contacts" retry button). */ let inFlightMeSubscribe: Promise | null = null; /** Call once at session start (after login) -- chat-list metadata only, no message bodies. */ export function subscribeMeTopic(): Promise { if (inFlightMeSubscribe) return inFlightMeSubscribe; const me = getTinode().getMeTopic(); wireTopicCallbacks(me); // Read/recv/unread updates (e.g. from noteRead() when opening a conversation) travel through // TopicMe's OWN single-slot callback, separate from onData/onInfo/onPres -- wireTopicCallbacks // doesn't touch it. Without this, the chat list's unread badge/ordering goes stale after reading // a conversation until some unrelated topic event happens to invalidate its cached snapshot. me.onContactUpdate = () => notifyTopic(me.name); inFlightMeSubscribe = me.subscribe(me.startMetaQuery().withLaterSub().withDesc().withTags().withCred().build()).finally(() => { inFlightMeSubscribe = null; }); return inFlightMeSubscribe; } /** Tracks prefetchPreview's own in-flight subscribe per topic, so openConversation (below) can wait * for it to settle first instead of firing a second overlapping `{sub}` packet at the same topic -- * this is a real path, not just theoretical: ChatListColumn prefetches every visible contact's * preview on render, and a deep link or fast click can open one of those same topics before its * preview subscribe has resolved. */ const inFlightPreview = new Map>(); /** * One-shot per topic: subscribes just enough to learn its latest message for the chat-list preview. * Deliberately never left afterward (matches the old app) -- simpler than tracking teardown for a * subscription that's cheap to leave attached and likely to be reopened soon anyway. */ export function prefetchPreview(topicName: string): void { if (previewPrefetched.has(topicName)) return; previewPrefetched.add(topicName); const topic = getTinode().getTopic(topicName); wireTopicCallbacks(topic); const promise = topic.subscribe(topic.startMetaQuery().withLaterData(1).build()).catch(() => { previewPrefetched.delete(topicName); }); inFlightPreview.set(topicName, promise); promise.finally(() => { if (inFlightPreview.get(topicName) === promise) inFlightPreview.delete(topicName); }); } /** * The SDK has no per-topic in-flight tracking of its own: `Topic.subscribe()` only short-circuits * once `_attached` is already true, so two overlapping calls (React 19 StrictMode's mount->cleanup-> * remount runs every effect twice in dev, so openConversation() always overlaps with itself once per * open; ChatListColumn's prefetchPreview and this both racing for the same just-deep-linked topic is * the same shape in prod) would otherwise send two `{sub}` packets, and a `leave()` that lands before * the first `subscribe()` resolves just rejects ("Cannot leave inactive topic") and no-ops instead of * detaching. Tracked here so every caller for the same topic name shares one in-flight promise. */ const inFlightSubscribe = new Map>(); /** * Bumped by every openConversation() call, checked by closeConversation() after it's done waiting -- * without this, StrictMode's mount->cleanup->remount still breaks a freshly (re)opened conversation: * cleanup-1 awaits the SAME shared inFlightSubscribe promise mount-2 is also relying on (they're * deduped to one promise, per the comment above), and once it resolves, cleanup-1's queued * continuation calls leave() unconditionally -- detaching the subscription mount-2 just established * a moment earlier, with no error surfaced anywhere (confirmed by reading tinode.dev.js: `_attached` * only flips true on a successful subscribe, and leave(false) while `_attached` is true really does * send `{leave}` and reset it). The topic silently stops receiving live messages/receipts. Capturing * the generation synchronously at closeConversation's call time and re-checking it after all awaits * settle detects "a newer open has since claimed this topic" and skips the stale leave(). */ const topicGeneration = new Map(); /** * Opens a topic for full conversation view: fetches a page of history and wires live updates. * Uses withData(undefined, undefined, limit), NOT withLaterData(limit) -- if prefetchPreview already * nudged the topic's local _maxSeq up to just that one preview message, withLaterData would ask for * "newer than that" and the server would correctly report back empty. * * Known gap: for a brand-new P2P topic the peer has never messaged, the old app defers subscribing * at all until the user's first send (avoiding server-side subscription state for chats opened but * never used) -- not yet replicated here since the new-conversation picker flow doesn't exist yet. */ export function openConversation(topicName: string): Promise { topicGeneration.set(topicName, (topicGeneration.get(topicName) || 0) + 1); const existing = inFlightSubscribe.get(topicName); if (existing) return existing; const promise = (async () => { // Let any in-flight chat-list preview subscribe for this same topic settle first, rather than // firing a second overlapping `{sub}` packet at it (see inFlightPreview above). const previewPending = inFlightPreview.get(topicName); if (previewPending) await previewPending.catch(() => {}); const topic = getTinode().getTopic(topicName); wireTopicCallbacks(topic); if (topic.isSubscribed()) { await topic.leave(false); } const query = topic.startMetaQuery().withLaterDesc().withLaterSub().withAux().withData(undefined, undefined, MESSAGES_PAGE); await topic.subscribe(query.build()); return topic; })(); inFlightSubscribe.set(topicName, promise); promise.finally(() => { if (inFlightSubscribe.get(topicName) === promise) inFlightSubscribe.delete(topicName); }); return promise; } /** Soft-leave (detach, not unsubscribe) -- cached messages stay in the SDK's local buffer. Waits out * any in-flight openConversation() for this same topic first, so a cleanup that fires before the * subscribe it's undoing has even resolved doesn't just silently no-op (see inFlightSubscribe above) * -- then, per topicGeneration's comment above, skips the actual leave() if a NEWER openConversation * call has since claimed this topic while this one was waiting. */ export async function closeConversation(topicName: string): Promise { const myGeneration = topicGeneration.get(topicName) || 0; const pending = inFlightSubscribe.get(topicName); if (pending) await pending.catch(() => {}); if (topicGeneration.get(topicName) !== myGeneration) return; await getTinode().getTopic(topicName).leave(false).catch(() => {}); } /** * Makes sure a topic can be published to right now, without disturbing whichever conversation is * actually open. `Topic.publishMessage()` rejects outright unless `_attached` is true, and that's * only reliably the case for the ONE topic openConversation/closeConversation currently has open -- * any OTHER topic you've ever actually opened and then navigated away from goes back to detached and * STAYS that way (closeConversation's leave() doesn't restore prefetchPreview's original one-shot * subscription, and prefetchPreview itself no-ops on a topic it's already touched once, attached or * not). That covers forwarding a message to anyone other than the person/group you're currently * chatting with -- i.e. most real forwards -- so this isn't a rare edge case to special-case around, * it's the common one. Reuses `inFlightSubscribe` so a concurrent openConversation() for the same * topic is awaited/shared rather than racing it with a second `{sub}` packet (see openConversation's * own comment above for why that matters). Deliberately left attached afterward rather than re-left -- * same "cheap to leave attached, likely reopened soon anyway" call prefetchPreview already makes. */ export function ensureTopicAttached(topicName: string): Promise { const topic = getTinode().getTopic(topicName); const pending = inFlightSubscribe.get(topicName); if (pending) return pending; if (topic.isSubscribed()) return Promise.resolve(topic); wireTopicCallbacks(topic); const promise = topic.subscribe().then(() => topic); inFlightSubscribe.set(topicName, promise); promise.finally(() => { if (inFlightSubscribe.get(topicName) === promise) inFlightSubscribe.delete(topicName); }); return promise; } let chatListSnapshot: Contact[] | null = null; /** * Real, addressable chat-list topic types this backend actually issues -- deliberately NOT using * `Topic#isCommType()`/`isP2PType()` (confirmed live by reading tinode-sdk's own `Topic.topicType()` * table: it maps prefix `'usr'` to P2P, with NO entry for `'p2p'` at all). That table is right for a * peer ALIAS (`usr`, used before a P2P topic has ever been subscribed), but this backend's * REAL, server-assigned P2P topic name is `p2p` (confirmed via raw `{meta topic:"me"}` capture: * every subscription the server reports uses this form, never the alias) -- so `isCommType()` returns * false for every P2P topic once its real name is known, silently filtering ALL of them out of * `MeTopic.contacts()`. This has nothing to do with account age or how a conversation was opened -- * verified by direct fresh-login inspection that even a heavily-used account with many real, working * P2P friendships had ZERO P2P entries in `contacts()`; every P2P row the chat list has ever shown * this session was actually the special-cased `slf` self-chat entry. */ function isChatListTopicName(name: string): boolean { return name.startsWith('p2p') || name.startsWith('grp') || name === 'slf'; } /** * Cached the same way getMessagesSnapshot is, but invalidated by ANY topic's change (a new message * on a contact that isn't currently open still needs to move that row and refresh its preview). * Triggers each contact's one-shot preview prefetch as a side effect of being read -- cheap and * idempotent (prefetchPreview no-ops after the first call per topic), so it's fine to do this on * every rebuild rather than needing a separate "wire up previews" pass. * * Uses `Tinode#mapTopics` (every locally-cached topic, unfiltered) + `isChatListTopicName` instead of * `MeTopic#contacts()` -- see that function's own comment for why `contacts()`'s built-in filter can't * be trusted here. `mapTopics` still only sees topics the SDK has locally cached, which `me`'s own * subscribe (`subscribeMeTopic`'s `withLaterSub()`) and any direct `openConversation()` both populate. */ export function getChatListSnapshot(): Contact[] { if (chatListSnapshot) return chatListSnapshot; const byTopic = new Map(); getTinode().mapTopics((topic) => { if (!topic.name || !isChatListTopicName(topic.name)) return; prefetchPreview(topic.name); // `.topic` (what every consumer of this list actually reads) normally arrives via // `_processMetaSubs` merging the server's own `sub.topic` field onto this same cached instance -- // but a brand-new P2P topic reached through `openConversation()` is cached here the moment its // OWN subscribe resolves, potentially before `me`'s subscription list has ever reported it. Set // explicitly rather than assuming that merge already happened; kept as the same live instance // (not a clone) so prototype methods like `latestMessage()` survive. const contact = topic as unknown as Contact; contact.topic = topic.name; byTopic.set(topic.name, contact); }); const rows = [...byTopic.values()]; chatListSnapshot = rows; return rows; } function invalidateChatListSnapshot() { chatListSnapshot = null; } anyTopicListeners.add(invalidateChatListSnapshot); /** * Re-fires the `me` topic subscribe and every contact's preview subscribe once a WS reconnect has * successfully re-authenticated (connection.ts's onReauthenticated). Bypasses subscribeMeTopic's/ * prefetchPreview's own "already done" guards on purpose -- those guards exist to stop a second * *concurrent* call from firing a redundant `{sub}` while one is already in flight, not to reflect * whether the topic is still genuinely attached, and after any disconnect it never is (the SDK * resets `_attached` on every cached topic, confirmed by reading tinode.dev.js's #disconnected()). * `Topic.subscribe()` itself only short-circuits when `_attached` is still true, so calling it again * here for an already-detached topic genuinely resends `{sub}`, not a silent no-op. * * Deliberately scoped to topics this module already knows about (the `me` topic + everything * prefetchPreview has ever touched) rather than tracking "the one conversation currently open" as * separate state -- in practice the chat list and an open conversation are mounted together in this * app's layout, so the open topic's contact row has already gone through prefetchPreview by the time * anyone could be looking at it. A conversation reached by a deep link that never rendered the chat * list at all is the one case this doesn't cover; also out of scope: backfilling any messages that * were sent by the peer during the drop itself -- this only resumes *live* updates going forward, * it doesn't gap-fill what was missed while disconnected. */ onReauthenticated(() => { const me = getTinode().getMeTopic(); wireTopicCallbacks(me); me.onContactUpdate = () => notifyTopic(me.name); me.subscribe(me.startMetaQuery().withLaterSub().withDesc().withTags().withCred().build()).catch(() => {}); for (const topicName of previewPrefetched) { const topic = getTinode().getTopic(topicName); wireTopicCallbacks(topic); topic.subscribe(topic.startMetaQuery().withLaterData(1).build()).catch(() => {}); } });