https://smrtbld.ai/js/realtime/MessageHubClient.js

1/**
2 * MessageHubClient - SignalR Client for Message Center
3 *
4 * Extends SignalRBaseClient to handle message-specific events.
5 * Uses pubsub for loose coupling with the messages page module.
6 *
7 * Hub path: /hubs/smrtbuild/messages
8 *
9 * Server-to-client events:
10 * - NewMessage: New message arrived in user's inbox
11 * - NewReply: Reply posted to a thread the user is viewing
12 * - MessageRead: Recipient read your message (Seen By)
13 * - UnreadCountChanged: Unread badge update
14 * - FlagChanged: Followup/attention flag toggled
15 *
16 * Client-to-server methods:
17 * - JoinUserGroup(userId): Subscribe to inbox notifications
18 * - LeaveUserGroup(userId): Unsubscribe from inbox notifications
19 * - JoinThread(threadId): Subscribe to thread replies
20 * - LeaveThread(threadId): Unsubscribe from thread replies
21 */
22
23import { SignalRBaseClient, ConnectionState } from './SignalRBaseClient.js';
24import { pubsub } from '../utils/pubsub.js';
25import { toCamelCase } from '../utils/case-transform.js';
26import CrossTabSync from '../notifications/CrossTabSync.js';
27
28// Only the partition's leader tab holds this socket; it relays every event to the other tabs (CA-548).
29function relayToFollowers(name, data) {
30    if (CrossTabSync.isThisTabLeader()) CrossTabSync.broadcast('message:event', { name, payload: data });
31}
32
33// =============================================================================
34// HUB CLIENT
35// =============================================================================
36
37class MessageHubClient extends SignalRBaseClient {
38    constructor() {
39        super('/hubs/smrtbuild/messages', {
40            maxRetries: 15,
41            baseDelayMs: 1000,
42            maxDelayMs: 60000
43        });
44
45        this._joined = false;
46    }
47
48    /**
49     * Register hub event handlers
50     * @param {signalR.HubConnection} connection
51     */
52    registerHubEvents(connection) {
53        // New message in user's inbox
54        connection.on('NewMessage', (payload) => {
55            console.debug('[MessageHub] NewMessage:', payload.MessageId || payload.messageId);
56            const data = toCamelCase(payload);
57            relayToFollowers('new', data);
58            pubsub.publish('signalr:message:new', data);
59        });
60
61        // New reply in a thread we're viewing
62        connection.on('NewReply', (payload) => {
63            console.debug('[MessageHub] NewReply:', payload.ThreadId || payload.threadId);
64            const data = toCamelCase(payload);
65            relayToFollowers('newReply', data);
66            pubsub.publish('signalr:message:newReply', data);
67        });
68
69        // Someone read our message
70        connection.on('MessageRead', (payload) => {
71            console.debug('[MessageHub] MessageRead:', payload.MessageId || payload.messageId);
72            const data = toCamelCase(payload);
73            relayToFollowers('read', data);
74            pubsub.publish('signalr:message:read', data);
75        });
76
77        // Unread count changed
78        connection.on('UnreadCountChanged', (payload) => {
79            console.debug('[MessageHub] UnreadCountChanged:', payload);
80            const count = typeof payload === 'object' ? (payload.count ?? payload.Count) : payload;
81            relayToFollowers('unreadCount', { count });
82            pubsub.publish('signalr:message:unreadCount', { count });
83        });
84
85        // Flag changed (followup/attention)
86        connection.on('FlagChanged', (payload) => {
87            console.debug('[MessageHub] FlagChanged:', payload.MessageId || payload.messageId);
88            const data = toCamelCase(payload);
89            relayToFollowers('flagChanged', data);
90            pubsub.publish('signalr:message:flagChanged', data);
91        });
92    }
93
94    /**
95     * Called when connection is established
96     */
97    onConnected() {
98        console.log('[MessageHub] Connected');
99
100        // Re-join user group if previously joined
101        if (this._joined) {
102            this._joinUser();
103        }
104
105        // Every thread any tab of this partition has open
106        this._joinWantedThreads();
107
108        pubsub.publish('signalr:message:connected');
109    }
110
111    /**
112     * Called when reconnected after disconnect
113     */
114    onReconnected(connectionId) {
115        console.log('[MessageHub] Reconnected:', connectionId);
116
117        // Re-join groups after reconnection
118        if (this._joined) {
119            this._joinUser();
120        }
121        this._joinWantedThreads();
122
123        pubsub.publish('signalr:message:reconnected');
124    }
125
126    /**
127     * Called when disconnected
128     */
129    onDisconnected(error) {
130        console.warn('[MessageHub] Disconnected:', error?.message);
131        pubsub.publish('signalr:message:disconnected', { error: error?.message });
132    }
133
134    /**
135     * Called when max retries exceeded.
136     * Schedules a long-interval background retry instead of giving up permanently.
137     */
138    onMaxRetriesExceeded() {
139        console.error('[MessageHub] Max retries exceeded - scheduling background retry');
140        // The toast is router.js's, in the visible tab — the leader is usually hidden (CA-548).
141        relayToFollowers('maxRetriesExceeded', {});
142        pubsub.publish('signalr:message:maxRetriesExceeded');
143
144        // Don't give up forever — try again in 60s
145        this.retryTimeout = setTimeout(() => {

146            console.log('[MessageHub] Background retry after max retries');
147            this.retryConnection();
148        }, 60000);
149    }
150
151    // =========================================================================
152    // PUBLIC API
153    // =========================================================================
154
155    /**
156     * Join the user's message group to receive inbox notifications.
157     * Backend extracts userId from JWT — no client-supplied ID needed.
158     */
159    async joinUserGroup() {
160        this._joined = true;
161        if (this.state === ConnectionState.Connected) {
162            await this._joinUser();
163        }
164    }
165
166    /**
167     * Leave the user's message group
168     */
169    async leaveUserGroup() {
170        this._joined = false;
171        if (this.state === ConnectionState.Connected) {
172            try {
173                await this.invoke('LeaveUserGroup');
174            } catch (e) {
175                console.debug('[MessageHub] LeaveUserGroup failed:', e.message);
176            }
177        }
178    }
179
180    // =========================================================================
181    // INTERNAL
182    // =========================================================================
183
184    _joinWantedThreads() {
185        for (const threadId of wantedThreads()) {
186            this._joinThread(threadId);
187        }
188    }
189
190    /**
191     * Leave thread group on server
192     */
193    async _leaveThread(threadId) {
194        try {
195            await this.invoke('LeaveThread', Number(threadId));
196        } catch (e) {
197            console.debug('[MessageHub] LeaveThread failed:', e.message);
198        }
199    }
200
201    /**
202     * Join user group on server (userId extracted from JWT server-side)
203     */
204    async _joinUser() {
205        try {
206            await this.invoke('JoinUserGroup');
207        } catch (e) {
208            console.debug('[MessageHub] JoinUserGroup failed:', e.message);
209        }
210    }
211
212    /**
213     * Join thread group on server
214     */
215    async _joinThread(threadId) {
216        try {
217            await this.invoke('JoinThread', Number(threadId));
218        } catch (e) {
219            console.debug('[MessageHub] JoinThread failed:', e.message);
220        }
221    }
222}
223
224// =============================================================================
225// SINGLETON
226// =============================================================================
227
228let instance = null;
229
230/**
231 * Get the singleton MessageHubClient instance
232 * @returns {MessageHubClient}
233 */
234export function getMessageHubClient() {
235    if (!instance) {
236        instance = new MessageHubClient();
237    }
238    return instance;
239}
240
241/**
242 * Start the message hub connection
243 * @returns {Promise<boolean>}
244 */
245export async function startMessageHub() {
246    const client = getMessageHubClient();
247    return client.start();
248}
249
250/**
251 * Stop the message hub connection. The singleton is released before the await so a start
252 * that runs meanwhile builds a fresh client instead of one this stop then orphans (CA-548).
253 */
254export async function stopMessageHub() {
255    const client = instance;
256    if (!client) return;
257    instance = null;
258    await client.stop();
259}
260
261// =============================================================================
262// THREAD INTEREST — one open thread per tab; the leader joins the partition's union
263// =============================================================================
264
265// Module scope, not the singleton: a stop nulls the client but the interests survive (CA-548).
266const THREAD_KEEPALIVE_MS = 60_000;
267const THREAD_REMOTE_TTL_MS = THREAD_KEEPALIVE_MS * 2.5;
268let currentThread = null;
269const remoteThreads = new Map(); // threadId → Map<tabId, lastSeenMs>
270let threadSyncUnsubs = [];
271let threadKeepalive = null;
272
273function wantedThreads() {
274    const wanted = new Set(remoteThreads.keys());
275    if (currentThread) wanted.add(currentThread);
276    return wanted;
277}
278

279function isThreadWanted(threadId) {
280    return currentThread === threadId || (remoteThreads.get(threadId)?.size ?? 0) > 0;
281}
282
283/** Is this the thread this tab has open? (Replies for other tabs' threads toast there, not here.) */
284export function isCurrentMessageThread(threadId) {
285    return threadId != null && currentThread === String(threadId);
286}
287
288function connectedLeaderHub() {
289    if (!CrossTabSync.isThisTabLeader() || instance?.state !== ConnectionState.Connected) return null;
290    return instance;
291}
292
293function broadcastThreadInterest() {
294    if (currentThread) {
295        CrossTabSync.broadcast('message:thread:join', { threadId: currentThread, tabId: CrossTabSync.getTabId() });
296    }
297}
298
299function onRemoteThreadJoin({ threadId, tabId } = {}) {
300    if (threadId == null || !tabId) return;
301    const key = String(threadId);
302    const wasWanted = isThreadWanted(key);
303    let tabs = remoteThreads.get(key);
304    if (!tabs) { tabs = new Map(); remoteThreads.set(key, tabs); }
305    tabs.set(tabId, Date.now());
306    if (!wasWanted) connectedLeaderHub()?._joinThread(key);
307}
308
309function onRemoteThreadLeave({ threadId, tabId } = {}) {
310    if (threadId == null || !tabId) return;
311    const key = String(threadId);
312    const tabs = remoteThreads.get(key);
313    if (!tabs) return;
314    tabs.delete(tabId);
315    if (tabs.size === 0) remoteThreads.delete(key);
316    leaveThreadIfUnwanted(key);
317}
318

319function leaveThreadIfUnwanted(threadId) {
320    if (isThreadWanted(threadId)) return;
321    connectedLeaderHub()?._leaveThread(threadId);
322}
323
324function pruneRemoteThreads() {
325    const cutoff = Date.now() - THREAD_REMOTE_TTL_MS;
326    for (const [threadId, tabs] of remoteThreads) {
327        for (const [tabId, lastSeen] of tabs) {
328            if (lastSeen < cutoff) tabs.delete(tabId);
329        }
330        if (tabs.size === 0) {
331            remoteThreads.delete(threadId);
332            leaveThreadIfUnwanted(threadId);
333        }
334    }
335}
336
337/**
338 * Start the cross-tab thread-interest relay (router: shell load). Every tab runs it, so any tab
339 * can become the leader with the union already in hand.
340 */
341export function initMessageThreadSync() {
342    if (threadSyncUnsubs.length) return;
343    threadSyncUnsubs = [
344        pubsub.subscribe('crosstab:message:thread:join', onRemoteThreadJoin),
345        pubsub.subscribe('crosstab:message:thread:leave', onRemoteThreadLeave),
346        // A new leader rebuilds the union from every tab's replay and joins it once connected.

347        pubsub.subscribe('crosstab:leaderElected', () => {
348            broadcastThreadInterest();
349            connectedLeaderHub()?._joinWantedThreads();
350        }),
351        pubsub.subscribe('crosstab:message:thread:replay', broadcastThreadInterest)
352    ];
353    // A leader granted while its shell still loaded missed earlier joins: ask every tab to replay.
354    CrossTabSync.broadcast('message:thread:replay', {});
355    // Keepalive re-announce + prune (a prune may leave an expired thread on the leader's socket).
356    threadKeepalive = setInterval(() => {
357        broadcastThreadInterest();
358        pruneRemoteThreads();
359    }, THREAD_KEEPALIVE_MS);
360}
361
362/** Logout teardown: drop every interest and stop the relay. */
363export function destroyMessageThreadSync() {
364    threadSyncUnsubs.forEach(unsub => unsub());
365    threadSyncUnsubs = [];
366    if (threadKeepalive) {
367        clearInterval(threadKeepalive);
368        threadKeepalive = null;
369    }
370    currentThread = null;
371    remoteThreads.clear();
372}
373
374/**
375 * This tab opened a thread: withdraw its previous one, then join through the leader's socket.
376 * @param {number|string} threadId
377 */
378export function joinMessageThread(threadId) {
379    if (threadId == null || threadId === '') return;
380    const key = String(threadId);
381    if (currentThread === key) return;
382    const previous = currentThread;
383    const wasWanted = isThreadWanted(key);
384    currentThread = key;
385    const tabId = CrossTabSync.getTabId();
386    if (previous) {
387        CrossTabSync.broadcast('message:thread:leave', { threadId: previous, tabId });
388        leaveThreadIfUnwanted(previous);
389    }
390    CrossTabSync.broadcast('message:thread:join', { threadId: key, tabId });
391    if (!wasWanted) connectedLeaderHub()?._joinThread(key);
392}
393
394/**
395 * This tab closed its thread.
396 * @param {number|string} threadId
397 */
398export function leaveMessageThread(threadId) {
399    if (threadId == null || currentThread !== String(threadId)) return;
400    const key = currentThread;
401    currentThread = null;
402    CrossTabSync.broadcast('message:thread:leave', { threadId: key, tabId: CrossTabSync.getTabId() });
403    leaveThreadIfUnwanted(key);
404}
405
406/**
407 * Check if connected
408 * @returns {boolean}
409 */
410export function isMessageHubConnected() {
411    return instance?.state === ConnectionState.Connected;
412}
413
414export { MessageHubClient };