fix: recover messages lost while an open room reconnects

CONTEXT.md

@@ -197,12 +197,13 @@ A **Message Action** is the active mode on a Message in the Room view. The three
 
 ## Server & Connection
 
-| Term               | Definition                                                                                                               | Aliases to avoid                                                 |
-| ------------------ | ------------------------------------------------------------------------------------------------------------------------ | ---------------------------------------------------------------- |
-| **Server**         | A Rocket.Chat server instance the app connects to, with version, settings, and enterprise modules                        | Workspace (used by web but not consistently in mobile), instance |
-| **Server History** | List of previously connected Servers for quick reconnection                                                              | Recent servers                                                   |
-| **Meteor Connect** | The WebSocket connection to the Server's DDP (Distributed Data Protocol) endpoint                                        | Socket, connection                                               |
-| **Socket Health**  | Whether the Meteor Connect socket is genuinely alive — confirmed by a round trip when in doubt, reopened when known dead | Staleness (stale/gray/fresh), socket probe                       |
+| Term                  | Definition                                                                                                                                                                    | Aliases to avoid                                                 |
+| --------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ---------------------------------------------------------------- |
+| **Server**            | A Rocket.Chat server instance the app connects to, with version, settings, and enterprise modules                                                                             | Workspace (used by web but not consistently in mobile), instance |
+| **Server History**    | List of previously connected Servers for quick reconnection                                                                                                                   | Recent servers                                                   |
+| **Meteor Connect**    | The WebSocket connection to the Server's DDP (Distributed Data Protocol) endpoint                                                                                             | Socket, connection                                               |
+| **Socket Health**     | Whether the Meteor Connect socket is genuinely alive — confirmed by a round trip when in doubt, reopened when known dead                                                      | Staleness (stale/gray/fresh), socket probe                       |
+| **Room Stream Ready** | The moment the Server acks a Room's `stream-room-messages` subscription, meaning live Messages for that Room are flowing again; fires on first connect and on every reconnect | Connected, subscribed                                            |
 
 ## Navigation & Layout
 

app/definitions/ISubscription.ts

@@ -44,8 +44,8 @@ export interface ISubscription {
 	v?: IVisitor;
 	f: boolean;
 	t: SubscriptionType; // TODO: we need to review this type later
-	ts: string | Date;
-	ls: Date;
+	ts?: string | Date;
+	ls?: Date;
 	name: string;
 	fname?: string;
 	sanitizedFname?: string;
@@ -61,7 +61,7 @@ export interface ISubscription {
 	tunread: string[];
 	tunreadUser?: string[];
 	tunreadGroup?: string[];
-	roomUpdatedAt: Date | number;
+	roomUpdatedAt?: Date | number;
 	ro: boolean;
 	lastOpen?: Date;
 	description?: string;

app/lib/methods/createDirectMessageSubscriptionStub.test.ts

@@ -98,9 +98,10 @@ describe('createDirectMessageSubscriptionStub', () => {
 		expect(created.archived).toBe(false);
 		expect(created.f).toBe(false);
 		expect(created.ro).toBe(false);
-		expect(created.ts).toBeInstanceOf(Date);
-		expect(created.ls).toBeInstanceOf(Date);
-		expect(created.roomUpdatedAt).toBeInstanceOf(Date);
+		// We don't guess timestamps from the device clock.
+		expect(created.ts).toBeUndefined();
+		expect(created.ls).toBeUndefined();
+		expect(created.roomUpdatedAt).toBeUndefined();
 		expect(log).not.toHaveBeenCalled();
 	});
 

app/lib/methods/createDirectMessageSubscriptionStub.ts

@@ -55,7 +55,6 @@ export const createDirectMessageSubscriptionStub = async ({
 
 		const db = database.active;
 		const subCollection = db.get(SUBSCRIPTIONS_TABLE);
-		const now = new Date();
 
 		await db.write(async () => {
 			await subCollection.create((s: any) => {
@@ -75,9 +74,9 @@ export const createDirectMessageSubscriptionStub = async ({
 				s.ro = false;
 				s.archived = false;
 				s.f = false;
-				s.ts = now;
-				s.ls = now;
-				s.roomUpdatedAt = now;
+				// No timestamps here. They belong to the server, and we'd only be guessing from the
+				// device clock: `ls` places the unread separator and `ts`/`roomUpdatedAt` feed sync
+				// cursors, so a wrong value breaks them until the real subscription arrives.
 			});
 		});
 	} catch (e) {

app/lib/methods/helpers/emitter.ts

@@ -6,19 +6,27 @@ type TDynamicMediaDownloadEvents = {
 	[key: `downloadMedia${string}`]: string;
 };
 
-export type TEmitterEvents = TDynamicMediaDownloadEvents & {
-	toolbarMention: undefined;
-	addMarkdown: {
-		style: TMarkdownStyle;
-	};
-	setKeyboardHeight: number;
-	setKeyboardHeightThread: number;
-	setComposerHeight: number;
-	setComposerHeightThread: number;
-	audioFocused: string;
-	navigationReady: undefined;
+/** Emitted once the server acks the room's `stream-room-messages` subscription, on every (re)connect. */
+type TRoomStreamReadyEvents = {
+	[key: `roomStreamReady${string}`]: undefined;
 };
 
+export const roomStreamReadyEvent = (rid: string) => `roomStreamReady${rid}` as const;
+
+export type TEmitterEvents = TDynamicMediaDownloadEvents &
+	TRoomStreamReadyEvents & {
+		toolbarMention: undefined;
+		addMarkdown: {
+			style: TMarkdownStyle;
+		};
+		setKeyboardHeight: number;
+		setKeyboardHeightThread: number;
+		setComposerHeight: number;
+		setComposerHeightThread: number;
+		audioFocused: string;
+		navigationReady: undefined;
+	};
+
 export type TKeyEmitterEvent = keyof TEmitterEvents;
 
 export const emitter = mitt<TEmitterEvents>();

app/lib/methods/loadMessagesForRoom.test.ts

@@ -170,6 +170,21 @@ describe('loadMessagesForRoom', () => {
 		expect(mockedDispatch).not.toHaveBeenCalledWith(expect.objectContaining({ type: ROOM.HISTORY_UI_LOADER_PUSH }));
 	});
 
+	it('writes nothing when the connection cycle it belongs to has already ended', async () => {
+		const partialBatch = Array.from({ length: 5 }, (_, index) =>
+			buildMessage({
+				id: `stale-${index + 1}`,
+				ts: new Date(Date.UTC(2024, 0, 1, 0, 0, 5 - index)).toISOString()
+			})
+		);
+		mockedSdkGet.mockResolvedValueOnce({ success: true, messages: partialBatch } as any);
+
+		await loadMessagesForRoom({ rid: 'ROOM_ID', t: 'c', isStale: () => true });
+
+		expect(mockedUpdateMessages).not.toHaveBeenCalled();
+		expect(updateLastOpen).not.toHaveBeenCalled();
+	});
+
 	it('pops the ui loader when a recursive batch fetch fails after the loader was pushed', async () => {
 		const firstBatch = buildHiddenBatch('first', 50);
 		const networkError = new Error('boom');

app/lib/methods/loadMessagesForRoom.ts

@@ -119,6 +119,11 @@ export async function loadMessagesForRoom(args: {
 	t: RoomTypes;
 	latest?: Date;
 	loaderItem?: TMessageModel;
+	/**
+	 * Checked before we save anything. A load from a connection that already dropped can come back
+	 * late and overwrite newer messages or push `lastOpen` backwards, so we throw its result away.
+	 */
+	isStale?: () => boolean;
 }): Promise<void> {
 	let uiLoaderId: string | null = null;
 	try {
@@ -128,6 +133,9 @@ export async function loadMessagesForRoom(args: {
 				uiLoaderId = id;
 			}
 		});
+		if (args.isStale?.()) {
+			return;
+		}
 		if (messages?.length) {
 			const lastMessage = messages[messages.length - 1];
 			const lastMessageRecord = await getMessageById(lastMessage._id as string);
@@ -143,6 +151,11 @@ export async function loadMessagesForRoom(args: {
 			await updateMessages({ rid: args.rid, update: messages, loaderItem: args.loaderItem });
 		}
 
+		// Checked again because the save above is async: the connection can drop while it runs, and
+		// nothing stops `lastOpen` from moving backwards, so a late write would lower it.
+		if (args.isStale?.()) {
+			return;
+		}
 		if (!args.latest && !args.loaderItem) {
 			await updateLastOpen(args.rid, serverTimestamps);
 		}

app/lib/methods/loadMissedMessages.test.ts

@@ -3,6 +3,7 @@ import sdk from '../services/sdk';
 import updateMessages from './updateMessages';
 import { getSubscriptionByRoomId } from '../database/services/Subscription';
 import { updateLastOpen } from './updateLastOpen';
+import { loadMessagesForRoom } from './loadMessagesForRoom';
 import { store } from '../store/auxStore';
 
 jest.mock('../services/sdk', () => ({
@@ -24,6 +25,7 @@ jest.mock('../store/auxStore', () => ({
 }));
 
 jest.mock('./updateMessages', () => jest.fn());
+jest.mock('./loadMessagesForRoom', () => ({ loadMessagesForRoom: jest.fn() }));
 jest.mock('./updateLastOpen', () => ({
 	...jest.requireActual('./updateLastOpen'),
 	updateLastOpen: jest.fn()
@@ -34,6 +36,7 @@ const mockedSdkGet = sdk.get as jest.MockedFunction<typeof sdk.get>;
 const mockedUpdateMessages = updateMessages as jest.MockedFunction<typeof updateMessages>;
 const mockedGetSubscriptionByRoomId = getSubscriptionByRoomId as jest.MockedFunction<typeof getSubscriptionByRoomId>;
 const mockedUpdateLastOpen = updateLastOpen as jest.MockedFunction<typeof updateLastOpen>;
+const mockedLoadMessagesForRoom = loadMessagesForRoom as jest.MockedFunction<typeof loadMessagesForRoom>;
 
 const RID = 'ROOM_ID';
 
@@ -65,12 +68,80 @@ describe('loadMissedMessages', () => {
 		);
 	});
 
-	it('fetches nothing when the subscription has no cursor', async () => {
+	it('recovers through the room history load when the subscription has no cursor', async () => {
 		mockedGetSubscriptionByRoomId.mockResolvedValue({ lastOpen: null, t: 'p' } as never);
 
 		await loadMissedMessages({ rid: RID });
 
+		expect(mockedLoadMessagesForRoom).toHaveBeenCalledWith({ rid: RID, t: 'p' });
+		// Without a cursor there's no request to make, so it must not try.
 		expect(mockedSdkGet).not.toHaveBeenCalled();
+		expect(mockedUpdateLastOpen).not.toHaveBeenCalled();
+	});
+
+	// If we ever let a cursorless room reach the legacy branch, it sends `lastUpdate: undefined`.
+	it('short-circuits the legacy branch on a server below 7.1.0 when there is no cursor', async () => {
+		(store.getState as jest.Mock).mockReturnValue({ server: { version: '7.0.0' } });
+		mockedGetSubscriptionByRoomId.mockResolvedValue({ lastOpen: null, t: 'c' } as never);
+
+		await loadMissedMessages({ rid: RID });
+
+		expect(mockedSdkGet).not.toHaveBeenCalled();
+		expect(mockedLoadMessagesForRoom).toHaveBeenCalledWith({ rid: RID, t: 'c' });
+	});
+
+	it('recovers nothing when the subscription type is not a room type', async () => {
+		mockedGetSubscriptionByRoomId.mockResolvedValue({ lastOpen: null, t: 'thread' } as never);
+
+		await loadMissedMessages({ rid: RID });
+
+		expect(mockedLoadMessagesForRoom).not.toHaveBeenCalled();
+		expect(mockedSdkGet).not.toHaveBeenCalled();
+	});
+
+	it('passes the staleness guard into the recovery so a superseded cycle writes nothing', async () => {
+		mockedGetSubscriptionByRoomId.mockResolvedValue({ lastOpen: null, t: 'c' } as never);
+		const isStale = () => true;
+
+		await loadMissedMessages({ rid: RID, isStale });
+
+		expect(mockedLoadMessagesForRoom).toHaveBeenCalledWith({ rid: RID, t: 'c', isStale });
+	});
+
+	it('keeps a healthy cursor on the sync walk instead of delegating', async () => {
+		const CURSOR = new Date(Date.UTC(2024, 0, 1, 11, 0, 0));
+		mockedGetSubscriptionByRoomId.mockResolvedValue({ lastOpen: CURSOR, t: 'c' } as never);
+		mockedSdkGet.mockResolvedValue({ result: { updated: [], deleted: [], cursor: { next: null } } } as never);
+
+		await loadMissedMessages({ rid: RID });
+
+		expect(mockedLoadMessagesForRoom).not.toHaveBeenCalled();
+		expect(mockedSdkGet).toHaveBeenCalledWith('chat.syncMessages', expect.objectContaining({ next: CURSOR.getTime() }));
+	});
+
+	it('does not throw when the response carries no cursor', async () => {
+		mockedGetSubscriptionByRoomId.mockResolvedValue({ lastOpen: new Date(Date.UTC(2024, 0, 1)), t: 'c' } as never);
+		mockedSdkGet.mockResolvedValue({ result: { updated: [], deleted: [] } } as never);
+
+		await expect(loadMissedMessages({ rid: RID })).resolves.toBeUndefined();
+		expect(mockedUpdateLastOpen).toHaveBeenCalledWith(RID, []);
+	});
+
+	it('stops the sync walk at the batch cap instead of paging unbounded history', async () => {
+		mockedGetSubscriptionByRoomId.mockResolvedValue({ lastOpen: new Date(Date.UTC(2024, 0, 1)), t: 'c' } as never);
+		// Every page hands back another cursor, so nothing but the cap can end this.
+		mockedSdkGet.mockResolvedValue({
+			result: { updated: [], deleted: [], cursor: { next: Date.UTC(2024, 0, 1, 11, 0, 0) } }
+		} as never);
+
+		await loadMissedMessages({ rid: RID });
+		for (let i = 0; i < 30; i += 1) {
+			await new Promise(resolve => setImmediate(resolve));
+		}
+
+		// 10 pages, one UPDATED and one DELETED request each, and then it stops.
+		expect(mockedSdkGet).toHaveBeenCalledTimes(20);
+		expect(mockedUpdateLastOpen).not.toHaveBeenCalled();
 	});
 
 	describe('last open', () => {

app/lib/methods/loadMissedMessages.ts

@@ -6,9 +6,18 @@ import { store } from '../store/auxStore';
 import { getSubscriptionByRoomId } from '../database/services/Subscription';
 import log from './helpers/log';
 import { snapshotServerTimestamps, type TServerTimestamps, updateLastOpen } from './updateLastOpen';
+import { loadMessagesForRoom } from './loadMessagesForRoom';
+import { isRoomType } from './roomTypeToApiType';
 
 const count = 50;
 
+/**
+ * Nothing stops this walk on its own, so without a cap it can pull down a room's entire history.
+ * When we hit the cap we stop without saving the cursor, so the next reconnect starts over from
+ * the same place. That means a room further behind than 10 pages never catches up here.
+ */
+const MAX_PAGES = 10;
+
 const syncMessages = async ({ roomId, next, type }: { roomId: string; next: number; type: 'UPDATED' | 'DELETED' }) => {
 	// @ts-ignore // this method dont have type
 	const { result } = await sdk.get('chat.syncMessages', { roomId, next, count, type });
@@ -38,27 +47,41 @@ const getSyncMessagesFromCursor = async (
 	const [updatedMessages, deletedMessages] = await Promise.all([updatedPromise, deletedPromise]);
 	return {
 		deleted: deletedMessages?.deleted ?? [],
-		deletedNext: deletedMessages?.cursor.next,
+		deletedNext: deletedMessages?.cursor?.next ?? null,
 		updated: updatedMessages?.updated ?? [],
-		updatedNext: updatedMessages?.cursor.next
+		updatedNext: updatedMessages?.cursor?.next ?? null
 	};
 };
 
 async function load({
 	rid: roomId,
 	updatedNext,
-	deletedNext
+	deletedNext,
+	isStale
 }: {
 	rid: string;
 	updatedNext?: number | null;
 	deletedNext?: number | null;
+	isStale?: () => boolean;
 }) {
 	const sub = await getSubscriptionByRoomId(roomId);
 	if (!sub) {
 		return;
 	}
 	const cursor = sub.lastOpen;
 
+	// Without a cursor there is nothing to sync from, so we fall back to loading the room's recent
+	// history instead. That load stops after a few batches and leaves a "load more" row behind for
+	// whatever it didn't reach, and it saves a real cursor so the next reconnect can sync normally.
+	// `sub.t` can also hold things that aren't rooms ('e2e', 'thread'); there's no history endpoint
+	// for those, so we skip them.
+	if (!cursor && !updatedNext && !deletedNext) {
+		if (isRoomType(sub.t)) {
+			await loadMessagesForRoom({ rid: roomId, t: sub.t, isStale });
+		}
+		return;
+	}
+
 	const { version: serverVersion } = store.getState().server;
 	if (compareServerVersion(serverVersion, 'greaterThanOrEqualTo', '7.1.0')) {
 		const result = await getSyncMessagesFromCursor(roomId, cursor?.getTime(), updatedNext, deletedNext);
@@ -76,14 +99,26 @@ export async function loadMissedMessages(args: {
 	updatedNext?: number | null;
 	deletedNext?: number | null;
 	serverTimestamps?: TServerTimestamps;
+	/**
+	 * Checked after every page: a fetch belonging to a connection cycle that has already ended can
+	 * resolve late and overwrite newer rows or lower the cursor, so its result is dropped instead.
+	 */
+	isStale?: () => boolean;
+	/** Which page we're on, starting at 1. Only used to stop the walk once it reaches `MAX_PAGES`. */
+	page?: number;
 }): Promise<void> {
+	const page = args.page ?? 1;
 	// A DELETED-only continuation fetches no UPDATED page, so it must not write the cursor again.
 	const fetchedUpdatedPage = !!args.updatedNext || !args.deletedNext;
 	const data = await load({
 		rid: args.rid,
 		updatedNext: args.updatedNext,
-		deletedNext: args.deletedNext
+		deletedNext: args.deletedNext,
+		isStale: args.isStale
 	});
+	if (args.isStale?.()) {
+		return;
+	}
 	if (data) {
 		const {
 			updated,
@@ -97,12 +132,20 @@ export async function loadMissedMessages(args: {
 		// @ts-ignore // TODO: remove loaderItem obligatoriness
 		await updateMessages({ rid: args.rid, update: updated, remove: deleted });
 
-		if (deletedNext || updatedNext) {
+		// Re-checked because the write above awaits: the cycle can end while it runs, and the cursor
+		// has no monotonic clamp, so a stale write would lower it.
+		if (args.isStale?.()) {
+			return;
+		}
+
+		if ((deletedNext || updatedNext) && page < MAX_PAGES) {
 			loadMissedMessages({
 				rid: args.rid,
 				updatedNext,
 				deletedNext,
-				serverTimestamps
+				serverTimestamps,
+				isStale: args.isStale,
+				page: page + 1
 			}).catch(log);
 		}