fix: restore DDP streams through a single restoration owner after reconnect

app/lib/methods/getPermissions.ts

@@ -8,6 +8,7 @@ import log from './helpers/log';
 import { store as reduxStore } from '../store/auxStore';
 import database from '../database';
 import sdk from '../services/sdk';
+import { registerStreamRestorer } from '../services/connectionRestore';
 import protectedFunction from './helpers/protectedFunction';
 import { compareServerVersion } from './helpers';
 
@@ -166,7 +167,6 @@ export function getPermissions(): Promise<void> {
 			const db = database.active;
 			const permissionsCollection = db.get('permissions');
 			const allRecords = await permissionsCollection.query().fetch();
-			sdk.subscribe('stream-notify-logged', 'permissions-changed');
 			// if server version is lower than 0.73.0, fetches from old api
 			if (serverVersion && compareServerVersion(serverVersion, 'lowerThan', '0.73.0')) {
 				// RC 0.66.0
@@ -205,3 +205,11 @@ export function getPermissions(): Promise<void> {
 		}
 	});
 }
+
+export function subscribePermissions(): void {
+	return sdk.subscribe('stream-notify-logged', 'permissions-changed');
+}
+
+// Restorer owns only re-subscription; the awaited getPermissions() fetch stays in the login saga so
+// permissions state is set before the enterprise-modules and VoIP checks that read it.
+registerStreamRestorer(() => subscribePermissions());

app/lib/methods/getRoles.ts

@@ -8,6 +8,7 @@ import { store as reduxStore } from '../store/auxStore';
 import { removeRoles, setRoles as setRolesAction, updateRoles } from '../../actions/roles';
 import { type TRoleModel } from '../../definitions';
 import sdk from '../services/sdk';
+import { registerStreamRestorer } from '../services/connectionRestore';
 import protectedFunction from './helpers/protectedFunction';
 
 export async function setRoles(): Promise<void> {
@@ -129,3 +130,9 @@ export function getRoles(): Promise<void> {
 		}
 	});
 }
+
+// Re-send the stream-roles sub and refresh roles on every DDP login.
+registerStreamRestorer(() => {
+	sdk.subscribe('stream-roles', 'roles');
+	return getRoles();
+});

app/lib/methods/getSettings.ts

@@ -10,6 +10,7 @@ import log from './helpers/log';
 import { store as reduxStore } from '../store/auxStore';
 import database from '../database';
 import sdk from '../services/sdk';
+import { registerStreamRestorer } from '../services/connectionRestore';
 import protectedFunction from './helpers/protectedFunction';
 import { parseSettings, _prepareSettings } from './parseSettings';
 import { setPresenceCap } from './getUsersPresence';
@@ -147,6 +148,9 @@ export function subscribeSettings(): void {
 	return sdk.subscribe('stream-notify-all', 'public-settings-changed');
 }
 
+// Re-send the public-settings-changed sub on every DDP login (initial, reconnect, forceReopen, swap).
+registerStreamRestorer(() => subscribeSettings());
+
 type IData = ISettingsIcon | IPreparedSettings;
 
 export async function getSettings(): Promise<void> {

app/lib/methods/getUsersPresence.ts

@@ -9,6 +9,7 @@ import { setUser } from '../../actions/login';
 import database from '../database';
 import { type IUser } from '../../definitions';
 import sdk from '../services/sdk';
+import { registerStreamRestorer } from '../services/connectionRestore';
 import { compareServerVersion, normalizeStatusExpiresAt } from './helpers';
 import log from './helpers/log';
 import userPreferences from './userPreferences';
@@ -178,3 +179,9 @@ export const refreshDmUsersPresence = async (): Promise<void> => {
 		log(e);
 	}
 };
+
+// Re-send the presence subs and refresh open-DM presence on every DDP login.
+registerStreamRestorer(() => {
+	subscribeUsersPresence();
+	return refreshDmUsersPresence();
+});

app/lib/methods/subscribeRooms.ts

@@ -1,5 +1,6 @@
 import log from './helpers/log';
 import subscribeRoomsTmp, { roomsSubscription } from './subscriptions/rooms';
+import { registerStreamRestorer } from '../services/connectionRestore';
 
 export async function subscribeRooms(): Promise<void> {
 	if (!roomsSubscription?.stop) {
@@ -16,3 +17,10 @@ export function unsubscribeRooms(): void {
 		roomsSubscription.stop();
 	}
 }
+
+// Restore the single `subscribeNotifyUser` sub set (message/notification/rooms-changed/…) that the
+// day-one features ride. Resetting the guard first forces `subscribeRooms` to re-bind on `sdk.current`.
+registerStreamRestorer(() => {
+	unsubscribeRooms();
+	return subscribeRooms();
+});

app/lib/methods/subscriptions/room.test.ts

@@ -17,6 +17,12 @@ jest.mock('../../services/sdk', () => ({
 	}
 }));
 
+const mockUnregisterRestorer = jest.fn();
+const mockRegisterStreamRestorer = jest.fn<() => void, [() => void]>(() => mockUnregisterRestorer);
+jest.mock('../../services/connectionRestore', () => ({
+	registerStreamRestorer: (restore: () => void) => mockRegisterStreamRestorer(restore)
+}));
+
 const mockStoreGetState = jest.fn<{ meteor: { connected: boolean } }, []>(() => ({
 	meteor: { connected: false }
 }));
@@ -143,9 +149,9 @@ describe('RoomSubscription', () => {
 		});
 	});
 
-	describe('handleLogin', () => {
+	describe('restore', () => {
 		it('calls subscribeRoom, dispatches clearUserTyping, loads missed messages, and reads', async () => {
-			await sub.handleLogin();
+			await sub.restore();
 
 			expect(mockSubscribeRoom).toHaveBeenCalledWith(rid);
 			expect(mockStoreDispatch).toHaveBeenCalledWith(clearUserTyping());
@@ -155,7 +161,7 @@ describe('RoomSubscription', () => {
 		it('handles subscribeRoom rejection gracefully', async () => {
 			mockSubscribeRoom.mockRejectedValueOnce(new Error('boom'));
 
-			await expect(sub.handleLogin()).resolves.toBeUndefined();
+			await expect(sub.restore()).resolves.toBeUndefined();
 		});
 	});
 
@@ -170,11 +176,11 @@ describe('RoomSubscription', () => {
 	});
 
 	describe('DDP subscription recovery after forceReopen', () => {
-		it('handleLogin re-subscribes the room to restore lost DDP subscriptions', async () => {
+		it('restore re-subscribes the room to restore lost DDP subscriptions', async () => {
 			await sub.subscribe();
 			mockSubscribeRoom.mockClear();
 
-			await sub.handleLogin();
+			await sub.restore();
 
 			expect(mockSubscribeRoom).toHaveBeenCalledTimes(1);
 			expect(mockSubscribeRoom).toHaveBeenCalledWith(rid);
@@ -195,14 +201,14 @@ describe('RoomSubscription', () => {
 			mockSubscribeRoom.mockResolvedValueOnce([staleSub]).mockResolvedValueOnce([freshSub]);
 
 			await sub.subscribe();
-			await sub.handleLogin();
+			await sub.restore();
 			await sub.unsubscribe();
 
 			expect(staleSub.unsubscribe).toHaveBeenCalledTimes(1);
 			expect(freshSub.unsubscribe).toHaveBeenCalledTimes(1);
 		});
 
-		it('does not accumulate subscriptions across repeated handleLogin calls (simulates sequential reopen)', async () => {
+		it('does not accumulate subscriptions across repeated restore calls (simulates sequential reopen)', async () => {
 			const first = { unsubscribe: jest.fn(() => Promise.resolve()) };
 			const second = { unsubscribe: jest.fn(() => Promise.resolve()) };
 			mockSubscribeRoom.mockResolvedValueOnce([first]).mockResolvedValueOnce([second]);
@@ -211,13 +217,13 @@ describe('RoomSubscription', () => {
 			expect(mockSubscribeRoom).toHaveBeenCalledTimes(1);
 
 			// First reopen → tears down [first], creates [second]
-			await sub.handleLogin();
+			await sub.restore();
 			expect(mockSubscribeRoom).toHaveBeenCalledTimes(2);
 			expect(first.unsubscribe).toHaveBeenCalledTimes(1);
 			expect(second.unsubscribe).not.toHaveBeenCalled();
 
 			// Second reopen → tears down [second], creates []
-			await sub.handleLogin();
+			await sub.restore();
 			expect(mockSubscribeRoom).toHaveBeenCalledTimes(3);
 			expect(second.unsubscribe).toHaveBeenCalledTimes(1);
 
@@ -227,20 +233,27 @@ describe('RoomSubscription', () => {
 			expect(second.unsubscribe).toHaveBeenCalledTimes(1);
 		});
 
-		it('does not call onStreamData inside handleLogin (listeners persist across reopen)', async () => {
+		it('re-homes the stream listeners onto the current instance inside restore', async () => {
 			await sub.subscribe();
 			mockOnStreamData.mockClear();
 
-			await sub.handleLogin();
+			await sub.restore();
 
-			expect(mockOnStreamData).not.toHaveBeenCalled();
+			expect(mockOnStreamData).toHaveBeenCalledWith('close', sub.handleClose);
+			expect(mockOnStreamData).toHaveBeenCalledWith('stream-notify-room', sub.handleNotifyRoomReceived);
+			expect(mockOnStreamData).toHaveBeenCalledWith('stream-room-messages', sub.handleMessageReceived);
+			expect(mockOnStreamData).not.toHaveBeenCalledWith('login', expect.anything());
 		});
 
-		it('re-subscribes on the authenticated "login" event, not the pre-auth "connected" event', async () => {
+		it('enrolls a stream restorer on subscribe and disposes it on unsubscribe (no own "login" listener)', async () => {
 			await sub.subscribe();
 
-			expect(mockOnStreamData).toHaveBeenCalledWith('login', sub.handleLogin);
-			expect(mockOnStreamData).not.toHaveBeenCalledWith('connected', expect.anything());
+			expect(mockRegisterStreamRestorer).toHaveBeenCalledWith(sub.restore);
+			expect(mockOnStreamData).not.toHaveBeenCalledWith('login', expect.anything());
+
+			await sub.unsubscribe();
+
+			expect(mockUnregisterRestorer).toHaveBeenCalledTimes(1);
 		});
 
 		it('survives a poisoned subscription array (undefined entry from a rejected sub) and still re-subscribes', async () => {
@@ -251,30 +264,30 @@ describe('RoomSubscription', () => {
 			await sub.subscribe();
 			mockSubscribeRoom.mockClear();
 
-			await expect(sub.handleLogin()).resolves.toBeUndefined();
+			await expect(sub.restore()).resolves.toBeUndefined();
 			expect(mockSubscribeRoom).toHaveBeenCalledTimes(1);
 			expect(mockSubscribeRoom).toHaveBeenCalledWith(rid);
 		});
 	});
 
 	describe('isAlive guard', () => {
-		it('handleLogin does nothing once the subscription is no longer alive (race with unsubscribe)', async () => {
+		it('restore does nothing once the subscription is no longer alive (race with unsubscribe)', async () => {
 			await sub.subscribe();
 			await sub.unsubscribe();
 			jest.clearAllMocks();
 
-			await sub.handleLogin();
+			await sub.restore();
 
 			expect(mockSubscribeRoom).not.toHaveBeenCalled();
 			expect(loadMissedMessages).not.toHaveBeenCalled();
 			expect(mockStoreDispatch).not.toHaveBeenCalled();
 		});
 
-		it('handleLogin re-subscribes while the subscription is still alive', async () => {
+		it('restore re-subscribes while the subscription is still alive', async () => {
 			await sub.subscribe();
 			mockSubscribeRoom.mockClear();
 
-			await sub.handleLogin();
+			await sub.restore();
 
 			expect(mockSubscribeRoom).toHaveBeenCalledWith(rid);
 		});

app/lib/methods/subscriptions/room.ts

@@ -25,6 +25,7 @@ import {
 } from '../../../definitions';
 import { type IDDPMessage } from '../../../definitions/IDDPMessage';
 import sdk from '../../services/sdk';
+import { registerStreamRestorer } from '../../services/connectionRestore';
 import { readMessages } from '../readMessages';
 import { loadMissedMessages } from '../loadMissedMessages';
 import markMessagesRead from '../helpers/markMessagesRead';
@@ -33,10 +34,10 @@ export default class RoomSubscription {
 	private rid: string;
 	private isAlive: boolean;
 	private promises?: Promise<TSubscriptionModel[]>;
-	private loginListener?: Promise<any>;
 	private disconnectedListener?: Promise<any>;
 	private notifyRoomListener?: Promise<any>;
 	private messageReceivedListener?: Promise<any>;
+	private unregisterRestorer?: () => void;
 
 	constructor(rid: string) {
 		this.rid = rid;
@@ -50,10 +51,10 @@ export default class RoomSubscription {
 		}
 		this.promises = sdk.subscribeRoom(this.rid);
 
-		this.loginListener = sdk.onStreamData('login', this.handleLogin);
 		this.disconnectedListener = sdk.onStreamData('close', this.handleClose);
 		this.notifyRoomListener = sdk.onStreamData('stream-notify-room', this.handleNotifyRoomReceived);
 		this.messageReceivedListener = sdk.onStreamData('stream-room-messages', this.handleMessageReceived);
+		this.unregisterRestorer = registerStreamRestorer(this.restore);
 		if (!this.isAlive) {
 			await this.unsubscribe();
 		}
@@ -74,7 +75,8 @@ export default class RoomSubscription {
 			}
 		}
 		reduxStore.dispatch(clearUserTyping());
-		this.removeListener(this.loginListener);
+		this.unregisterRestorer?.();
+		this.unregisterRestorer = undefined;
 		this.removeListener(this.disconnectedListener);
 		this.removeListener(this.notifyRoomListener);
 		this.removeListener(this.messageReceivedListener);
@@ -91,11 +93,22 @@ export default class RoomSubscription {
 		}
 	};
 
-	handleLogin = async () => {
+	restore = async () => {
 		if (!this.isAlive) {
 			return;
 		}
 		try {
+			// Re-home the stream listeners onto the current (possibly new) instance's socket.
+			this.removeListener(this.disconnectedListener);
+			this.removeListener(this.notifyRoomListener);
+			this.removeListener(this.messageReceivedListener);
+			this.disconnectedListener = sdk.onStreamData('close', this.handleClose);
+			this.notifyRoomListener = sdk.onStreamData('stream-notify-room', this.handleNotifyRoomReceived);
+			this.messageReceivedListener = sdk.onStreamData('stream-room-messages', this.handleMessageReceived);
+			if (!this.isAlive) {
+				await this.unsubscribe();
+				return;
+			}
 			if (this.promises) {
 				const oldSubs = await this.promises;
 				oldSubs?.forEach(sub => sub?.unsubscribe?.().catch(() => {}));

app/lib/services/connect.test.ts

@@ -1,7 +1,6 @@
 import { connect, determineAuthType, disconnect } from './connect';
 import { mediaSessionInstance } from './voip/MediaSessionInstance';
 import { pendingHangups } from './voip/pendingHangups';
-import { unsubscribeRooms } from '../methods/subscribeRooms';
 import { setUser } from '../../actions/login';
 import database from '../database';
 
@@ -73,6 +72,13 @@ jest.mock('../methods/subscribeRooms', () => ({
 	unsubscribeRooms: jest.fn()
 }));
 
+const mockBindStop = jest.fn();
+const mockBindStreamRestoration = jest.fn<Promise<{ stop: jest.Mock }>, []>(() => Promise.resolve({ stop: mockBindStop }));
+jest.mock('./connectionRestore', () => ({
+	bindStreamRestoration: () => mockBindStreamRestoration(),
+	registerStreamRestorer: () => () => {}
+}));
+
 jest.mock('../methods/getSettings', () => ({
 	getSettings: jest.fn()
 }));
@@ -484,7 +490,7 @@ describe('connect — pendingHangups drain on reconnect', () => {
 	});
 });
 
-describe('connect — rooms subscription guard reset on close', () => {
+describe('connect — stream restoration owner', () => {
 	beforeEach(() => {
 		jest.clearAllMocks();
 		mockOnStreamDataStops.length = 0;
@@ -495,22 +501,22 @@ describe('connect — rooms subscription guard reset on close', () => {
 		});
 	});
 
-	// Regression: a long background marks the DDP socket stale, so foregrounding triggers
-	// `checkAndReopen` → `forceReopen`, which wipes the SDK subscriptions and emits 'close' while
-	// bypassing `connect()`. The rooms-list `stream-notify-user` feed only re-subscribes when the
-	// module-level guard in `subscribeRooms` is clear, and `unsubscribeRooms()` is what clears it.
-	// If the 'close' handler stops calling `unsubscribeRooms()`, the guard stays set after reconnect
-	// and the rooms list silently stops updating (subscriptions/favorites/reads).
-	it('calls unsubscribeRooms when the socket "close" fires', async () => {
+	it('binds the stream-restoration owner after initializing the SDK instance', async () => {
 		await connect({ server: 'https://example.com' });
 
-		// connect() itself calls unsubscribeRooms() once while tearing down prior listeners; ignore it.
-		(unsubscribeRooms as jest.Mock).mockClear();
+		expect(mockBindStreamRestoration).toHaveBeenCalledTimes(1);
+		expect(mockSdkInitialize.mock.invocationCallOrder[0]).toBeLessThan(mockBindStreamRestoration.mock.invocationCallOrder[0]);
+	});
 
-		const closeHandler = getHandlersByEvent('close')[0];
-		closeHandler();
+	it('stops the previous restoration owner when connect runs again', async () => {
+		await connect({ server: 'https://example.com' });
+		await flushMicrotasks();
+		mockBindStop.mockClear();
+
+		await connect({ server: 'https://example.com' });
+		await flushMicrotasks();
 
-		expect(unsubscribeRooms).toHaveBeenCalledTimes(1);
+		expect(mockBindStop).toHaveBeenCalledTimes(1);
 	});
 });