import type { IMessage, IRoom, ISubscription } from '@rocket.chat/core-typings'; import { Emitter } from '@rocket.chat/emitter'; import { differenceInMilliseconds } from 'date-fns'; import { useCallback, useSyncExternalStore } from 'react'; import { getUserPreference } from './getUserPreference'; import { dispatchToastMessage } from './toast'; import { Messages, Subscriptions } from '../stores'; import { sdk } from './SDKClient'; import { onClientMessageReceived } from './onClientMessageReceived'; import { getUserId } from './user'; import { getConfig } from './utils/getConfig'; import { mapMessageFromApi } from './utils/mapMessageFromApi'; const processMessage = async (msg: IMessage & { ignored?: boolean }, { subscription }: { subscription?: ISubscription }) => { const userId = msg.u?._id; if (subscription?.ignored?.includes(userId)) { msg.ignored = true; } if (msg.t === 'e2e') { msg.e2e = 'pending'; } return (await onClientMessageReceived(msg)) || msg; }; export async function upsertMessage({ msg, subscription }: { msg: IMessage & { ignored?: boolean }; subscription?: ISubscription }) { Messages.state.store(await processMessage(msg, { subscription })); } export async function upsertMessageBulk({ msgs, subscription, }: { msgs: (IMessage & { ignored?: boolean })[]; subscription?: ISubscription; }) { const processedMsgs = await Promise.all(msgs.map(async (msg) => processMessage(msg, { subscription }))); Messages.state.storeMany(processedMsgs); } const defaultLimit = parseInt(getConfig('roomListLimit') ?? '50') || 50; export type RoomHistoryState = { hasMore: boolean; hasMoreNext: boolean; isLoading: boolean; unreadNotLoaded: number; firstUnread: IMessage | undefined; loaded: number | undefined; cursorPrevious?: string | null; cursorNext?: string | null; scroll?: { scrollHeight: number; scrollTop: number; }; }; const roomStateEvent = (rid: IRoom['_id']) => `state:${rid}` as const; class RoomHistoryManagerClass extends Emitter { private lastRequest?: Date; private histories: Record = {}; // Bumped by `clear`/`close`. In-flight requests capture it at entry and discard their // response if it changed, so a page from a dead window can't touch the rebuilt one. private generations: Record = {}; private requestsList: string[] = []; private generation(rid: IRoom['_id']): number { return this.generations[rid] ?? 0; } public getRoom(rid: IRoom['_id']): RoomHistoryState { if (!this.histories[rid]) { this.histories[rid] = { hasMore: true, hasMoreNext: false, isLoading: false, unreadNotLoaded: 0, firstUnread: undefined, loaded: undefined, }; } return this.histories[rid]; } public updateRoom(rid: IRoom['_id'], patch: Partial): void { const room = this.getRoom(rid); Object.assign(room, patch); this.emit(roomStateEvent(rid), room); } public subscribeToRoom(rid: IRoom['_id'], cb: (state: RoomHistoryState) => void): () => void { return this.on(roomStateEvent(rid), cb); } private async queue(): Promise { return new Promise((resolve) => { const requestId = crypto.randomUUID(); const done = () => { this.lastRequest = new Date(); resolve(); }; if (this.requestsList.length === 0) { return this.run(done); } this.requestsList.push(requestId); this.once(requestId, done); }); } private run(fn: () => void) { const difference = this.lastRequest ? differenceInMilliseconds(new Date(), this.lastRequest) : Infinity; // Original cooldown was 500ms which forced ~330ms wait on the second getMore call when a // user opens a room. Pagination throughput here is bounded by the loadHistory server // method itself, so a smaller client-side spacing is enough to avoid hammering. const minSpacingMs = 100; if (difference > minSpacingMs) { return fn(); } return setTimeout(fn, minSpacingMs - difference); } public isLoaded(rid: IRoom['_id']) { const room = this.getRoom(rid); return room.loaded !== undefined; } private unqueue() { const requestId = this.requestsList.pop(); if (!requestId) { return; } this.run(() => this.emit(requestId)); } public async getMore(rid: IRoom['_id'], { limit = defaultLimit }: { limit?: number } = {}): Promise { const room = this.getRoom(rid); if (room.hasMore !== true) { return; } const generation = this.generation(rid); try { this.updateRoom(rid, { isLoading: true }); await this.queue(); // A `clear` may have run while queued (jump to message, jump to recent): the window this // cursor belonged to is gone, and a rebuild may already have installed a different one. if (generation !== this.generation(rid)) { return; } let ls = undefined; const subscription = Subscriptions.state.find((record) => record.rid === rid); if (subscription) { ({ ls } = subscription); } const showThreadsInMainChannel = getUserPreference(getUserId(), 'showThreadsInMainChannel', false); const { cursorPrevious: previous } = room; const result = await sdk.rest.get('/v1/rooms.history', { roomId: rid, ...(previous && { previous }), ...(ls && { lastSeen: new Date(ls).toISOString() }), count: limit, showThreadMessages: showThreadsInMainChannel, }); this.unqueue(); if (generation !== this.generation(rid)) { return; } const messages = result.messages.map((msg) => mapMessageFromApi(msg)); this.updateRoom(rid, { unreadNotLoaded: result.unreadNotLoaded ?? 0, firstUnread: result.firstUnread && mapMessageFromApi(result.firstUnread), cursorPrevious: result.cursor.previous, hasMore: result.cursor.previous !== null, }); const wrapper = document.querySelector('.messages-box .wrapper [data-overlayscrollbars-viewport]'); if (wrapper) { room.scroll = { scrollHeight: wrapper.scrollHeight, scrollTop: wrapper.scrollTop, }; } await upsertMessageBulk({ msgs: messages.filter((msg) => msg.t !== 'command'), subscription, }); if (generation !== this.generation(rid)) { return; } this.emit('loaded-messages'); if (!room.loaded) { room.loaded = 0; } const visibleMessages = messages.filter((msg) => msg.t !== 'command' && (!msg.tmid || showThreadsInMainChannel || msg.tshow)); room.loaded += visibleMessages.length; // `count` bounds raw messages, so a whole page can render as nothing and stall the scroll. if (room.hasMore && visibleMessages.length === 0) { return this.getMore(rid); } this.emit('loaded-messages'); } finally { // When stale, `clear` already reset the flag and the rebuild owns it now. if (generation === this.generation(rid)) { this.updateRoom(rid, { isLoading: false }); } } } public restoreScroll(rid: IRoom['_id']) { const room = this.getRoom(rid); const wrapper = document.querySelector('.messages-box .wrapper [data-overlayscrollbars-viewport]'); if (room.scroll === undefined) { return; } if (!wrapper) { return; } const heightDiff = wrapper.scrollHeight - (room.scroll.scrollHeight ?? NaN); wrapper.scrollTop = room.scroll.scrollTop + heightDiff; room.scroll = undefined; } public async getMoreNext(rid: IRoom['_id']) { const room = this.getRoom(rid); if (room.hasMoreNext !== true) { return; } const generation = this.generation(rid); try { this.updateRoom(rid, { isLoading: true }); await this.queue(); // A `clear` may have run while queued (jump to message, jump to recent): the window this // cursor belonged to is gone, and a rebuild may already have installed a different one. if (generation !== this.generation(rid)) { return; } const { cursorNext: next } = room; if (!next) { return; } const subscription = Subscriptions.state.find((record) => record.rid === rid); const showThreadsInMainChannel = getUserPreference(getUserId(), 'showThreadsInMainChannel', false); const result = await sdk.rest.get('/v1/rooms.history', { roomId: rid, next, count: defaultLimit, showThreadMessages: showThreadsInMainChannel, }); // Covers a `clear` that landed while the request was on the wire. if (generation !== this.generation(rid)) { return; } const messages = result.messages.map((msg) => mapMessageFromApi(msg)); await upsertMessageBulk({ msgs: messages.filter((msg) => msg.t !== 'command'), subscription, }); if (generation !== this.generation(rid)) { return; } this.emit('loaded-messages'); this.updateRoom(rid, { cursorNext: result.cursor.next, hasMoreNext: result.cursor.next !== null, }); if (!room.loaded) { room.loaded = 0; } room.loaded += messages.length; } finally { if (generation === this.generation(rid)) { this.updateRoom(rid, { isLoading: false }); } this.unqueue(); } } public hasMore(rid: IRoom['_id']) { return this.getRoom(rid).hasMore; } public hasMoreNext(rid: IRoom['_id']) { return this.getRoom(rid).hasMoreNext; } public getMoreIfIsEmpty(rid: IRoom['_id']) { const room = this.getRoom(rid); if (room.loaded === undefined) { return this.getMore(rid); } } public isLoading(rid: IRoom['_id']) { return this.getRoom(rid).isLoading; } public close(rid: IRoom['_id']) { this.generations[rid] = this.generation(rid) + 1; Messages.state.remove((record) => record.rid === rid); delete this.histories[rid]; } public clear(rid: IRoom['_id']) { this.generations[rid] = this.generation(rid) + 1; const room = this.getRoom(rid); Messages.state.remove((record) => record.rid === rid); room.loaded = undefined; this.updateRoom(rid, { isLoading: false, hasMore: true, hasMoreNext: false, cursorPrevious: undefined, cursorNext: undefined, // The snapshot describes the DOM of the window being discarded; a stale restoreScroll // resuming after this clear must no-op instead of applying it to the rebuilt window. scroll: undefined, }); this.emit('room-cleared', rid); } public async getSurroundingMessages(message?: Pick & { ts?: Date }) { return this.loadSurroundingMessages(message, true); } public async getSurroundingChannelMessages(message?: Pick & { ts?: Date }) { return this.loadSurroundingMessages(message, false); } private async loadSurroundingMessages(message: (Pick & { ts?: Date }) | undefined, showThreadMessages: boolean) { if (!message?.rid) { return; } const messageAlreadyLoaded = Messages.state.some((record) => record._id === message._id && record._hidden !== true); if (messageAlreadyLoaded) { return; } const room = this.getRoom(message.rid); this.updateRoom(message.rid, { isLoading: true }); const subscription = Subscriptions.state.find((record) => record.rid === message.rid); let generation = this.generation(message.rid); try { const result = await sdk.rest.get('/v1/rooms.history', { roomId: message.rid, aroundId: message._id, count: defaultLimit, showThreadMessages, }); // A `clear`/`close` that landed while the fetch was on the wire superseded this jump; // clearing here would wipe the newer window and rebuild a stale one over it. if (generation !== this.generation(message.rid)) { return; } // Rebuilds the window around the target, so the store does not grow monotonically here. this.clear(message.rid); // This rebuild owns the window created by its own `clear`; a later `clear` // (jump to recent, messagesImported) makes the rest of this flow stale. generation = this.generation(message.rid); // `clear` drops the cursors and clears `isLoading`, so restore the previous side before the // upsert yields: a `getMore` landing in that window has no cursor and would page from the // newest end instead. this.updateRoom(message.rid, { isLoading: true, cursorPrevious: result.cursor.previous, hasMore: result.cursor.previous !== null, }); const messages = result.messages.map((msg) => mapMessageFromApi(msg)); await upsertMessageBulk({ msgs: messages.filter((msg) => msg.t !== 'command'), subscription }); if (generation !== this.generation(message.rid)) { return; } // The next side only lands after the upsert: `hasMoreNext` arms jump-to-recent // (useHasNewMessages), which would clear the store while the upsert is still pending. this.updateRoom(message.rid, { cursorNext: result.cursor.next, hasMoreNext: result.cursor.next !== null, }); this.emit('loaded-messages'); if (!room.loaded) { room.loaded = 0; } room.loaded += messages.length; } catch (error) { if (generation !== this.generation(message.rid)) { return; } // The target may have been deleted since the link was created; no window to build then. if (!(error instanceof Response && error.status === 404)) { dispatchToastMessage({ type: 'error', message: error }); } if (!this.isLoaded(message.rid)) { await this.getMore(message.rid); } } finally { if (generation === this.generation(message.rid)) { this.updateRoom(message.rid, { isLoading: false }); } } } } export const RoomHistoryManager = new RoomHistoryManagerClass(); export const useRoomHistoryState = (rid: IRoom['_id'], selector: (state: RoomHistoryState) => T): T => useSyncExternalStore( useCallback((onStoreChange) => RoomHistoryManager.subscribeToRoom(rid, onStoreChange), [rid]), () => selector(RoomHistoryManager.getRoom(rid)), );