diff --git a/app/lib/methods/getPermissions.ts b/app/lib/methods/getPermissions.ts index cf38e96e466..bdf4b632254 100644 --- a/app/lib/methods/getPermissions.ts +++ b/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 { 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 { } }); } + +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()); diff --git a/app/lib/methods/getRoles.ts b/app/lib/methods/getRoles.ts index a266b91222c..54f96bf8623 100644 --- a/app/lib/methods/getRoles.ts +++ b/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 { @@ -129,3 +130,9 @@ export function getRoles(): Promise { } }); } + +// Re-send the stream-roles sub and refresh roles on every DDP login. +registerStreamRestorer(() => { + sdk.subscribe('stream-roles', 'roles'); + return getRoles(); +}); diff --git a/app/lib/methods/getSettings.ts b/app/lib/methods/getSettings.ts index e1f3a7d733b..2d67b3a5416 100644 --- a/app/lib/methods/getSettings.ts +++ b/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 { diff --git a/app/lib/methods/getUsersPresence.ts b/app/lib/methods/getUsersPresence.ts index 48b9ba536a5..7bd90b24706 100644 --- a/app/lib/methods/getUsersPresence.ts +++ b/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 => { log(e); } }; + +// Re-send the presence subs and refresh open-DM presence on every DDP login. +registerStreamRestorer(() => { + subscribeUsersPresence(); + return refreshDmUsersPresence(); +}); diff --git a/app/lib/methods/subscribeRooms.ts b/app/lib/methods/subscribeRooms.ts index 7d8aafe0c5b..42afc1ed057 100644 --- a/app/lib/methods/subscribeRooms.ts +++ b/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 { 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(); +}); diff --git a/app/lib/methods/subscriptions/room.test.ts b/app/lib/methods/subscriptions/room.test.ts index 5a179d8c545..fa4328a3a08 100644 --- a/app/lib/methods/subscriptions/room.test.ts +++ b/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); }); diff --git a/app/lib/methods/subscriptions/room.ts b/app/lib/methods/subscriptions/room.ts index ca70aea38e1..4f829df3b42 100644 --- a/app/lib/methods/subscriptions/room.ts +++ b/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; - private loginListener?: Promise; private disconnectedListener?: Promise; private notifyRoomListener?: Promise; private messageReceivedListener?: Promise; + 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(() => {})); diff --git a/app/lib/services/connect.test.ts b/app/lib/services/connect.test.ts index d3ac06b24ed..d7dbcbd7799 100644 --- a/app/lib/services/connect.test.ts +++ b/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.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); }); }); diff --git a/app/lib/services/connect.ts b/app/lib/services/connect.ts index dce7f343396..91fb155e978 100644 --- a/app/lib/services/connect.ts +++ b/app/lib/services/connect.ts @@ -11,11 +11,13 @@ import { twoFactor } from './twoFactor'; import { store } from '../store/auxStore'; import { loginRequest, logout, setLoginServices, setUser } from '../../actions/login'; import sdk from './sdk'; +import { createConnectedListener, createCloseListener } from './connectionListeners'; +import { bindStreamRestoration } from './connectionRestore'; import { mediaSessionInstance } from './voip/MediaSessionInstance'; import { pendingHangups } from './voip/pendingHangups'; import I18n from '../../i18n'; import { type ICredentials, type ILoggedUser, STATUSES } from '../../definitions'; -import { connectRequest, connectSuccess, disconnect as disconnectAction } from '../../actions/connect'; +import { connectRequest } from '../../actions/connect'; import { updatePermission } from '../../actions/permissions'; import EventEmitter from '../methods/helpers/events'; import { updateSettings } from '../../actions/settings'; @@ -49,6 +51,7 @@ let notifyAllListener: any; let rolesListener: any; let notifyLoggedListener: any; let logoutListener: any; +let restoreListener: any; function connect({ server, logoutOnError = false }: { server: string; logoutOnError?: boolean }): Promise { return new Promise(resolve => { @@ -96,11 +99,16 @@ function connect({ server, logoutOnError = false }: { server: string; logoutOnEr logoutListener.then(stopListener); } + if (restoreListener) { + restoreListener.then(stopListener); + } + unsubscribeRooms(); EventEmitter.emit('INQUIRY_UNSUBSCRIBE'); sdk.initialize(server); + restoreListener = bindStreamRestoration(); getSettings(); sdk.current @@ -116,31 +124,20 @@ function connect({ server, logoutOnError = false }: { server: string; logoutOnEr store.dispatch(connectRequest()); }); - connectedListener = sdk.current.onStreamData('connected', () => { - const { connected } = store.getState().meteor; - if (connected) { - return; - } - store.dispatch(connectSuccess()); - const { user } = store.getState().login; - if (user?.token) { - store.dispatch(loginRequest({ resume: user.token }, logoutOnError)); - } - }); + connectedListener = sdk.current.onStreamData('connected', createConnectedListener(logoutOnError)); // Tracks a real disconnect so the next `'connected'` can drain hangups the user tapped while // the WebSocket was unhealthy. Local to the closure so it resets per `connect()` call. let pendingHangupsDrainArmed = false; - closeListener = sdk.current.onStreamData('close', () => { - // Reset the rooms-subscription guard on every socket close. `forceReopen` (triggered by - // `checkAndReopen` after a long background) wipes the SDK subscriptions and emits 'close' - // but bypasses `connect()`, so without this the guard in `subscribeRooms` stays set and - // `stream-notify-user` is never re-subscribed — the rooms list silently stops updating. - unsubscribeRooms(); - pendingHangupsDrainArmed = true; - store.dispatch(disconnectAction()); - }); + closeListener = sdk.current.onStreamData( + 'close', + createCloseListener({ + onClose: () => { + pendingHangupsDrainArmed = true; + } + }) + ); pendingHangupsConnectedListener = sdk.current.onStreamData('connected', () => { if (!pendingHangupsDrainArmed) return; diff --git a/app/lib/services/connection.lifecycle.test.ts b/app/lib/services/connection.lifecycle.test.ts new file mode 100644 index 00000000000..097d2e5f32f --- /dev/null +++ b/app/lib/services/connection.lifecycle.test.ts @@ -0,0 +1,674 @@ +/** + * CONNECTION-LIFECYCLE REGRESSION SUITE. + * + * Guards the production behavior where an open room silently stops receiving OTHER users' + * `stream-room-messages` over the websocket after a reconnect, while the user's own REST sends keep + * working, and only a brand-new SDK instance heals it. Each scenario drives a realistic reconnect + * sequence through the REAL code: + * - REAL sdk wrapper app/lib/services/sdk.ts + * - REAL connect() app listeners app/lib/services/connectionListeners.ts (createConnected/CloseListener) + * - REAL DDPDriver + Socket node_modules/@rocket.chat/sdk/lib/drivers/ddp.ts (app-patched) + * - REAL RoomSubscription app/lib/methods/subscriptions/room.ts + * - REAL redux store/reducers meteor(connect)/login/room -> real meteor.connected transitions + * + * Modeled seams (documented, faithful to the causal chain, NOT product code executed verbatim): + * 1. Network: the Socket's `connection` is a fake in-memory transport wired to a FakeServer. + * The FakeServer tracks which streams are actually subscribed on the LIVE connection and + * only delivers a room push when a matching sub exists — encoding "the server does not push + * to a client that never (re)subscribed". A (re)open resets server-side subs (new TCP conn). + * 2. Login pipeline: production turns the connectedListener's `loginRequest` into `sdk.login` -> + * `Socket.emit('login')` via redux-saga. Here a tiny store subscriber does the same: on a + * LOGIN.REQUEST edge it emits 'login' on the CURRENT instance's socket and dispatches + * LOGIN.SUCCESS. Binding the real saga would drag the whole side-effect tree; the causal edge + * (LOGIN.REQUEST -> Socket 'login' on the current instance) is what these scenarios exercise. + * (Socket.login's subscribeAll is intentionally omitted: after forceReopen wipes + * `this.subscriptions` it re-sends nothing, so room recovery rides entirely on the app-layer + * restore re-subscribe — which this preserves.) + * + * The connect() `connected`/`close` listeners are NO LONGER modeled: `registerAppListeners` binds + * the REAL `createConnectedListener`/`createCloseListener` factories that production `connect()` + * calls, so the recovery dispatch (connectSuccess + resume loginRequest on every 'connected') is + * exercised as shipping code. `harnessConnect` reproduces connect.ts's instance-swap + listener-rebind + * (sdk.disconnect -> sdk.initialize -> re-register). + * + * Delivery verdict per scenario: was RoomSubscription.handleMessageReceived invoked for the push? + * GREEN = chain recovered, message delivered. RED = repro (message lost). + * + * Scenario f asserts the fresh-instance re-home: after an SDK swap the owner's 'login' fan-out runs + * RoomSubscription.restore, which re-homes the stream listeners onto the new instance and re-subscribes. + */ + +import EJSON from 'ejson'; + +import { initStore } from '../store/auxStore'; +import * as types from '../../actions/actionsTypes'; +import { connectSuccess } from '../../actions/connect'; + +// --- Use the REAL DDP Socket/driver instead of the empty global mock. --------- +jest.mock('@rocket.chat/sdk', () => { + // eslint-disable-next-line @typescript-eslint/no-var-requires + const { DDPDriver } = jest.requireActual('@rocket.chat/sdk/lib/drivers/ddp'); + const silentLogger = { debug: () => {}, info: () => {}, error: () => {}, warn: () => {} }; + + // Every socket the suite creates, so leaked DDP ping/reopen timers can be cleared on teardown. + const openSockets: any[] = []; + + // Tracks the streams the server currently considers subscribed on the LIVE connection. + class FakeServer { + activeSubs: Set; + socket: any; + constructor(socket: any) { + this.socket = socket; + this.activeSubs = new Set(); + } + private deliver(obj: any) { + // async so the Socket's `once(id)` response listener is registered before the reply lands + queueMicrotask(() => this.socket.onMessage({ data: JSON.stringify(obj) })); + } + handleOutgoing(raw: string) { + let data: any; + try { + data = JSON.parse(raw); + } catch { + return; + } + if (data.msg === 'connect') return this.deliver({ msg: 'connected', session: 'sess-1' }); + if (data.msg === 'ping') return this.deliver({ msg: 'pong' }); + if (data.msg === 'sub') { + this.activeSubs.add(`${data.name}::${JSON.stringify(data.params?.[0])}`); + return this.deliver({ msg: 'ready', subs: [data.id] }); + } + if (data.msg === 'unsub') return this.deliver({ msg: 'result', id: data.id, result: true }); + if (data.msg === 'method') return this.deliver({ msg: 'result', id: data.id, result: {} }); + } + hasRoomSub(rid: string) { + return this.activeSubs.has(`stream-room-messages::${JSON.stringify(rid)}`); + } + reset() { + this.activeSubs.clear(); + } + } + + const installConnection = (socket: any, server: FakeServer) => { + socket.connection = { + send: (raw: string) => server.handleOutgoing(raw), + close: () => {}, + readyState: 1, + onopen: () => {}, + onmessage: () => {}, + onerror: () => {}, + onclose: () => {} + }; + socket.lastPing = Date.now(); + }; + + class FakeRocketchat { + driver: any; + socket: any; + server: FakeServer; + host: string; + constructor(opts: any) { + this.host = opts.host; + this.driver = new DDPDriver({ host: opts.host, logger: silentLogger }); + this.socket = this.driver.ddp; + this.server = new FakeServer(this.socket); + installConnection(this.socket, this.server); + openSockets.push(this.socket); + // Stub the real websocket open(): install a fresh in-memory connection and reset + // server-side subs (a new connection = the server has no subs until the client re-subs). + // eslint-disable-next-line require-await + this.socket.open = jest.fn(async () => { + installConnection(this.socket, this.server); + this.server.reset(); + this.socket.emit('open'); + }); + } + // wrapper (sdk.ts) delegates these to `current` + onStreamData(...args: any[]) { + return this.driver.onStreamData(...args); + } + subscribe(...args: any[]) { + return this.driver.subscribe(...args); + } + subscribeRaw(...args: any[]) { + return this.driver.subscribeRaw(...args); + } + unsubscribe(sub: any) { + return this.driver.unsubscribe(sub); + } + checkAndReopen() { + return this.driver.checkAndReopen(); + } + connect() { + return Promise.resolve(this.driver); + } + disconnect() { + return Promise.resolve(); + } + abort() {} + get client() { + return { host: this.host }; + } + } + + // `__openSockets` is a test-teardown handle for the leaked-timer cleanup, not product surface. + return { __esModule: true, Rocketchat: FakeRocketchat, settings: {}, __openSockets: openSockets }; +}); + +// --- Neutralize heavy/native leaves reached by RoomSubscription (not the chain under test). --- +jest.mock('../encryption', () => ({ + Encryption: { + decryptMessage: jest.fn((msg: unknown) => Promise.resolve(msg)), + decryptPendingSubscriptions: jest.fn(), + decryptPendingMessages: jest.fn(), + getRoomInstance: jest.fn(), + stopRoom: jest.fn() + } +})); +jest.mock('../methods/loadMissedMessages', () => ({ loadMissedMessages: jest.fn(() => Promise.resolve()) })); +jest.mock('../methods/readMessages', () => ({ readMessages: jest.fn() })); +jest.mock('../methods/helpers/markMessagesRead', () => ({ __esModule: true, default: jest.fn() })); +jest.mock('../methods/helpers/log', () => ({ __esModule: true, default: jest.fn() })); +jest.mock('../database/services/Message', () => ({ getMessageById: jest.fn(() => Promise.resolve(null)) })); +jest.mock('../database/services/Thread', () => ({ getThreadById: jest.fn(() => Promise.resolve(null)) })); +jest.mock('../database/services/ThreadMessage', () => ({ getThreadMessageById: jest.fn(() => Promise.resolve(null)) })); +jest.mock('../database', () => { + const model = { prepareCreate: jest.fn(() => ({})), prepareUpdate: jest.fn(() => ({})), schema: {} }; + return { + __esModule: true, + default: { + setActiveDB: jest.fn(), + active: { + get: () => model, + write: jest.fn((cb: () => Promise) => cb()), + batch: jest.fn(() => Promise.resolve()) + } + } + }; +}); +jest.mock('./twoFactor', () => ({ twoFactor: jest.fn() })); + +// Imported AFTER mocks so they bind the real sdk wrapper + mocked leaves. +/* eslint-disable import/first, import/order */ +import { combineReducers, createStore, type Store } from 'redux'; +import * as ddpSdk from '@rocket.chat/sdk'; +import RoomSubscription from '../methods/subscriptions/room'; +import sdk from './sdk'; +import { createConnectedListener, createCloseListener } from './connectionListeners'; +import { bindStreamRestoration, registerStreamRestorer } from './connectionRestore'; +import connectReducer from '../../reducers/connect'; +import loginReducer from '../../reducers/login'; +import roomReducer from '../../reducers/room'; +// Imported for their module-scope registerStreamRestorer side effect: enrolls the settings, +// presence, permissions and roles restorers in the same registry the owner fans out to. +import '../methods/getSettings'; +import '../methods/getUsersPresence'; +import '../methods/getPermissions'; +import '../methods/getRoles'; +/* eslint-enable import/first, import/order */ + +const SERVER = 'https://open.rocket.chat'; +const RID = 'GENERAL'; +const USER = { id: 'u-me', username: 'me', token: 'resume-token' }; + +// Emit the scenario verdict as a single stable line so the pass/fail matrix is read directly +// from stdout (no snapshot indirection). +const report = (scenario: string, verdict: Record) => { + // eslint-disable-next-line no-console + console.log(`VERDICT ${scenario} ${JSON.stringify(verdict)}`); +}; + +const flush = async (n = 12) => { + for (let i = 0; i < n; i += 1) { + // eslint-disable-next-line no-await-in-loop + await Promise.resolve(); + } +}; + +type Instance = { socket: any; server: any }; +const currentInstance = (): Instance => sdk.current as unknown as Instance; + +// Only the slices these scenarios read directly; the real reducers own the full shape. +interface HarnessState { + meteor: { connected: boolean }; + login: { isFetching: boolean; isAuthenticated: boolean; user?: { token?: string } }; + room: unknown; + server: { version: string }; + settings: Record; +} + +let store: Store; + +function buildStore(): Store { + const reducer = combineReducers({ + meteor: connectReducer, + login: loginReducer, + room: roomReducer, + // static slices RoomSubscription / wrapper read + server: (s = { version: '6.0.0' }) => s, + settings: (s = {}) => s + }); + return createStore(reducer as any) as Store; +} + +// Models the redux-saga login pipeline: a LOGIN.REQUEST (dispatched by the connectedListener guard) +// completes into Socket 'login' + LOGIN.SUCCESS. Emits on the CURRENT instance's socket, which is +// exactly why a fresh instance (scenario f) strands a RoomSubscription bound to the old socket. +// 'fail' models the resume login terminally failing on a transient (non-401) error: no 'login' emit, +// no subscribeAll, meteor untouched. The saga's own bounded retry/backoff lives outside this harness, +// so this mode stands in for the outcome after retries are exhausted. +let loginPipelineMode: 'success' | 'fail' = 'success'; +function installLoginPipeline() { + let prevFetching = store.getState().login.isFetching; + store.subscribe(() => { + const { isFetching } = store.getState().login; + if (isFetching && !prevFetching) { + queueMicrotask(() => { + if (loginPipelineMode === 'fail') { + store.dispatch({ type: types.LOGIN.FAILURE, err: { message: 'transient network error' } } as any); + return; + } + currentInstance()?.socket?.emit('login', { token: USER.token }); + store.dispatch({ type: types.LOGIN.SUCCESS, user: USER } as any); + }); + } + prevFetching = isFetching; + }); +} + +// Binds the REAL connect.ts app listeners (createConnectedListener/createCloseListener) against the +// harness store, so the `meteor.connected` guard is the shipping code. logoutOnError=false mirrors a +// resume connect(). `unsubscribeRooms` is a no-op test double: the rooms-LIST subscription is out of +// scope for this room-message delivery chain — only the guard/dispatch logic matters here. +let connectedListener: any; +let closeListener: any; +let restoreListener: any; +function registerAppListeners() { + connectedListener = sdk.onStreamData('connected', createConnectedListener(false)); + closeListener = sdk.onStreamData('close', createCloseListener({})); + restoreListener = bindStreamRestoration(); +} + +// connect.ts's instance swap + listener rebind (connect.ts:58,103,119-143), minus peripheral wiring. +async function harnessConnect() { + if (connectedListener) (await connectedListener)?.stop?.(); + if (closeListener) (await closeListener)?.stop?.(); + if (restoreListener) (await restoreListener)?.stop?.(); + sdk.disconnect(); + sdk.initialize(SERVER); + registerAppListeners(); + await flush(); +} + +const fireConnected = (inst: Instance = currentInstance()) => + inst.socket.onMessage({ data: JSON.stringify({ msg: 'connected', session: 'sess-1' }) }); + +const fireClose = (inst: Instance = currentInstance(), code = 1006) => inst.socket.onClose({ code }); + +const encodedMessage = (rid: string) => ({ + _id: `msg-${Math.random().toString(36).slice(2)}`, + rid, + msg: 'hi from another user', + ts: { $date: Date.now() }, + u: { _id: 'u-other', username: 'other' }, + _updatedAt: { $date: Date.now() } +}); + +// Attempts a server push of a room message on the given instance. The server only delivers when it +// currently holds a sub for the room on the live connection (mirrors production). +function pushRoomMessage(rid: string, inst: Instance = currentInstance()) { + const serverHadSub = inst.server.hasRoomSub(rid); + if (serverHadSub) { + inst.socket.onMessage({ + data: JSON.stringify({ + msg: 'changed', + collection: 'stream-room-messages', + fields: { eventName: rid, args: [EJSON.toJSONValue(encodedMessage(rid))] } + }) + }); + } + return { serverHadSub }; +} + +const listenerCount = (inst: Instance, event: string) => inst.socket._listeners?.[event]?.length ?? 0; + +// RoomSubscriptions created by the harness. Their restorers live in the module-global registry, so +// afterEach unsubscribes them to keep a dead sub's restorer from firing on the next test's login. +const createdSubs: RoomSubscription[] = []; + +// Establishes a logged-in session with the room open and its stream-room-messages sub live. +async function openRoomSession() { + await harnessConnect(); + fireConnected(); + await flush(); // connected -> guard -> loginRequest -> (pipeline) login + LOGIN.SUCCESS + + const sub = new RoomSubscription(RID); + createdSubs.push(sub); + const received = jest.fn(sub.handleMessageReceived); + sub.handleMessageReceived = received; // spy BEFORE subscribe so the emitter binds the spy + await sub.subscribe(); + await flush(); // let the 5 stream subs ack on the server + + return { sub, received, subscribedInstance: currentInstance() }; +} + +describe('connection lifecycle — room message delivery across reconnects', () => { + beforeEach(() => { + store = buildStore(); + initStore(store); + loginPipelineMode = 'success'; + installLoginPipeline(); + // seed an authenticated session (user has a resume token) + store.dispatch({ type: types.LOGIN.SUCCESS, user: USER } as any); + connectedListener = undefined; + closeListener = undefined; + restoreListener = undefined; + }); + + afterEach(async () => { + // Dispose harness subs so their registry restorers don't leak into the next test. + await Promise.all(createdSubs.map(sub => sub.unsubscribe().catch(() => {}))); + createdSubs.length = 0; + // The real DDPDriver schedules ping/reopen timers on every socket (via ping()/reopen()) that + // never fire because scenarios don't advance time. Clear them so Jest exits without an + // open-handle warning. Test-only teardown — no production change. + (ddpSdk as any).__openSockets.forEach((socket: any) => { + if (socket.pingTimeout) clearTimeout(socket.pingTimeout); + if (socket.openTimeout) clearTimeout(socket.openTimeout); + }); + (ddpSdk as any).__openSockets.length = 0; + }); + + it('a. baseline: message delivered on a healthy subscribed room (harness sanity)', async () => { + const { received, subscribedInstance } = await openRoomSession(); + + const { serverHadSub } = pushRoomMessage(RID, subscribedInstance); + await flush(); + + report('a', { serverHadSub, delivered: received.mock.calls.length > 0 }); + expect(serverHadSub).toBe(true); + expect(received).toHaveBeenCalledTimes(1); + }); + + it('b. plain network flap: close(1006) -> reopen -> connected -> login resume', async () => { + const { received, subscribedInstance } = await openRoomSession(); + received.mockClear(); + + fireClose(subscribedInstance, 1006); // real onClose: emits close, schedules reopen (code !== 4000) + await flush(); + await subscribedInstance.socket.open(); // stand in for the scheduled reopen firing + await flush(); + fireConnected(subscribedInstance); + await flush(); // guard -> loginRequest -> login -> restore re-subscribes + + const { serverHadSub } = pushRoomMessage(RID, subscribedInstance); + await flush(); + + report('b', { serverHadSub, delivered: received.mock.calls.length > 0 }); + expect({ serverHadSub, delivered: received.mock.calls.length > 0 }).toEqual({ serverHadSub: true, delivered: true }); + }); + + it('c. foreground forceReopen (checkAndReopen stale bucket) -> connected -> login', async () => { + const { received, subscribedInstance } = await openRoomSession(); + received.mockClear(); + + // Foreground path: stale socket -> checkAndReopen -> forceReopen (emits close 4000, wipes subs, reopens) + subscribedInstance.socket.lastPing = 0; + await subscribedInstance.socket.checkAndReopen(); + await flush(); + fireConnected(subscribedInstance); + await flush(); + + const { serverHadSub } = pushRoomMessage(RID, subscribedInstance); + await flush(); + + report('c', { serverHadSub, delivered: received.mock.calls.length > 0 }); + expect({ serverHadSub, delivered: received.mock.calls.length > 0 }).toEqual({ serverHadSub: true, delivered: true }); + }); + + it('d. forceReopen while redux still reads connected=true must still recover', async () => { + const { received, subscribedInstance } = await openRoomSession(); + received.mockClear(); + + await subscribedInstance.socket.forceReopen(); // emits close(4000) -> disconnect dispatched + await flush(); + // Race: a connectSuccess lands (or the close disconnect never did) so redux reads connected=true + // at the moment 'connected' fires. Recovery must run regardless. + store.dispatch(connectSuccess()); + fireConnected(subscribedInstance); + await flush(); + + const { serverHadSub } = pushRoomMessage(RID, subscribedInstance); + await flush(); + + report('d', { + serverHadSub, + delivered: received.mock.calls.length > 0, + listenerOnSocket: listenerCount(subscribedInstance, 'stream-room-messages') + }); + // A stale connected=true no longer short-circuits recovery: 'connected' -> loginRequest -> + // restore re-subscribes, the server holds the room sub, and other users' messages arrive. + expect(received.mock.calls.length).toBeGreaterThan(0); + expect(serverHadSub).toBe(true); + }); + + it('e. overlapping reconnect: forceReopen fires again mid-recovery (before login completes)', async () => { + const { received, subscribedInstance } = await openRoomSession(); + received.mockClear(); + + await subscribedInstance.socket.forceReopen(); + fireConnected(subscribedInstance); // starts recovery (loginRequest queued) + // second forceReopen before restore's re-subscribe settles + await subscribedInstance.socket.forceReopen(); + await flush(); + fireConnected(subscribedInstance); + await flush(); + + const { serverHadSub } = pushRoomMessage(RID, subscribedInstance); + await flush(); + + report('e', { + serverHadSub, + delivered: received.mock.calls.length > 0, + listenerOnSocket: listenerCount(subscribedInstance, 'stream-room-messages') + }); + // Recovers: the final connected -> loginRequest -> restore re-subscribe wins. + expect(received.mock.calls.length).toBeGreaterThan(0); + }); + + it('f. a new SDK instance while RoomSubscription stays mounted must keep delivering', async () => { + const { received, subscribedInstance } = await openRoomSession(); + received.mockClear(); + + // connect() re-runs for the same server -> brand-new instance/socket, listeners rebound to it. + const generationBefore = sdk.generation; + await harnessConnect(); + const freshInstance = currentInstance(); + expect(freshInstance).not.toBe(subscribedInstance); + // The instance swap bumps the SDK generation id; CR-4 will key stream restoration to it. + expect(sdk.generation).toBe(generationBefore + 1); + + fireConnected(freshInstance); + await flush(); // fresh socket logs in + + const onNew = pushRoomMessage(RID, freshInstance); + const deliveredOnNew = received.mock.calls.length > 0; + const onOld = pushRoomMessage(RID, subscribedInstance); + const deliveredOnOld = received.mock.calls.length > 0; + await flush(); + + report('f', { + deliveredOnNewLiveSocket: deliveredOnNew, + deliveredOnOldDeadSocket: deliveredOnOld, + listenerOnOldSocket: listenerCount(subscribedInstance, 'stream-room-messages'), + listenerOnNewSocket: listenerCount(freshInstance, 'stream-room-messages'), + oldServerHadSub: onOld.serverHadSub, + newServerHadSub: onNew.serverHadSub + }); + // After connect() swaps to a fresh socket, the owner's 'login' fan-out runs RoomSubscription's + // restore, which re-homes stream-room-messages onto the new instance and re-sends the room sub, + // so real pushes on the new live socket are delivered. + expect(deliveredOnNew).toBe(true); + expect(listenerCount(freshInstance, 'stream-room-messages')).toBeGreaterThan(0); + expect(onNew.serverHadSub).toBe(true); + }); + + it('g. after a transient resume-login failure, a later connected re-runs recovery', async () => { + const { received, subscribedInstance } = await openRoomSession(); + received.mockClear(); + + // Foreground after silent socket death: checkAndReopen -> forceReopen. + // close(4000) -> meteor=false, DDP subs wiped; resume login then hits a transient error. + subscribedInstance.socket.lastPing = 0; + loginPipelineMode = 'fail'; + await subscribedInstance.socket.checkAndReopen(); + await flush(); + fireConnected(subscribedInstance); // connectSuccess(meteor=TRUE) -> loginRequest -> FAILS + await flush(); + + // Transient failure strands the session: socket reads connected, but the resume login failed so + // there is no authenticated session and no server-side sub yet. + const strandedState = { + meteorConnected: store.getState().meteor.connected, + isAuthenticated: store.getState().login.isAuthenticated + }; + const push1 = pushRoomMessage(RID, subscribedInstance); + await flush(); + const deliveredAfterFailure = received.mock.calls.length > 0; + + // Login now succeeds; a later 'connected' (no intervening close) must re-run recovery. + loginPipelineMode = 'success'; + fireConnected(subscribedInstance); + await flush(); + pushRoomMessage(RID, subscribedInstance); + await flush(); + const deliveredAfterSecondConnected = received.mock.calls.length > 0; + + report('g', { + strandedState, + serverHadSubAfterFailure: push1.serverHadSub, + deliveredAfterFailure, + deliveredAfterSecondConnected + }); + // After a transient resume-login failure, a later 'connected' (login now healthy) re-runs recovery + // and re-subscribes the room WITHOUT needing a real transport close. The stranded state heals as + // soon as the next 'connected' arrives. + expect(strandedState).toEqual({ meteorConnected: true, isAuthenticated: false }); + expect(deliveredAfterFailure).toBe(false); + expect(deliveredAfterSecondConnected).toBe(true); + }); + + it('h. the owner fans out to an enrolled restorer once per connect+login and drops stale generations', async () => { + const spy = jest.fn(); + const dispose = registerStreamRestorer(spy); + try { + await harnessConnect(); // fresh owner bound on the current generation + + fireConnected(); // connected -> loginRequest -> login -> fan-out (run #1) + await flush(); + expect(spy).toHaveBeenCalledTimes(1); + + fireConnected(); // a later connected re-runs login -> fan-out (run #2), no close needed + await flush(); + expect(spy).toHaveBeenCalledTimes(2); + + // Generation guard: an owner from a superseded connect() that outlived its generation must NOT + // fan out. Bind an owner, then swap the SDK instance WITHOUT stopping it, and fire the old + // socket's 'login' — the captured generation no longer matches, so no restorer runs. + const strandedInstance = currentInstance(); + await bindStreamRestoration(); + sdk.disconnect(); + sdk.initialize(SERVER); + spy.mockClear(); + strandedInstance.socket.emit('login', { token: USER.token }); + await flush(); + expect(spy).not.toHaveBeenCalled(); + } finally { + dispose(); + } + }); + + // The app-stream restorers (settings, presence, permissions, roles) enroll at module import and ride + // the SAME owner fan-out as RoomSubscription. Each transition below drops the server-side subs, then a + // later connected+login must re-send every one — proving the migration from LOGIN.SUCCESS forks to + // idempotent restorers survives plain reopen, forceReopen, SDK swap, and a transient login failure. + const APP_STREAM_SUBS: [string, string][] = [ + ['stream-notify-all', 'public-settings-changed'], + ['stream-notify-logged', 'updateAvatar'], + ['stream-notify-logged', 'Users:NameChanged'], + ['stream-notify-logged', 'permissions-changed'], + ['stream-roles', 'roles'] + ]; + const appStreamSnapshot = (inst: Instance, present: boolean) => + APP_STREAM_SUBS.reduce>((acc, [name, event]) => { + acc[`${name}/${event}`] = inst.server.activeSubs.has(`${name}::${JSON.stringify(event)}`) === present; + return acc; + }, {}); + const ALL_PRESENT = APP_STREAM_SUBS.reduce>((acc, [name, event]) => { + acc[`${name}/${event}`] = true; + return acc; + }, {}); + + it('i. app-stream subs re-established on the initial login and after a plain close/reopen', async () => { + const { subscribedInstance } = await openRoomSession(); + // initial login already fanned out to the restorers + report('i-initial', appStreamSnapshot(subscribedInstance, true)); + expect(appStreamSnapshot(subscribedInstance, true)).toEqual(ALL_PRESENT); + + fireClose(subscribedInstance, 1006); + await flush(); + await subscribedInstance.socket.open(); // reopen resets server-side subs + await flush(); + fireConnected(subscribedInstance); + await flush(); // connected -> loginRequest -> login -> restorers re-subscribe + + report('i', appStreamSnapshot(subscribedInstance, true)); + expect(appStreamSnapshot(subscribedInstance, true)).toEqual(ALL_PRESENT); + }); + + it('j. app-stream subs re-established after a forceReopen with a stale connected=true', async () => { + const { subscribedInstance } = await openRoomSession(); + + await subscribedInstance.socket.forceReopen(); // wipes DDP subs, resets server subs + await flush(); + store.dispatch(connectSuccess()); // redux reads connected=true when 'connected' fires + fireConnected(subscribedInstance); + await flush(); + + report('j', appStreamSnapshot(subscribedInstance, true)); + expect(appStreamSnapshot(subscribedInstance, true)).toEqual(ALL_PRESENT); + }); + + it('k. app-stream subs re-established on a brand-new SDK instance', async () => { + await openRoomSession(); + + await harnessConnect(); // fresh instance/socket, listeners + owner rebound + const freshInstance = currentInstance(); + fireConnected(freshInstance); + await flush(); + + report('k', appStreamSnapshot(freshInstance, true)); + expect(appStreamSnapshot(freshInstance, true)).toEqual(ALL_PRESENT); + }); + + it('l. app-stream subs re-established after a transient resume-login failure heals', async () => { + const { subscribedInstance } = await openRoomSession(); + + subscribedInstance.socket.lastPing = 0; + loginPipelineMode = 'fail'; + await subscribedInstance.socket.checkAndReopen(); // forceReopen -> resets server subs + await flush(); + fireConnected(subscribedInstance); // resume login FAILS -> no fan-out + await flush(); + // A failed resume login re-sends nothing: every stream is still absent. + const strandedAbsent = appStreamSnapshot(subscribedInstance, false); + + loginPipelineMode = 'success'; + fireConnected(subscribedInstance); // later connected re-runs recovery -> restorers fire + await flush(); + + report('l', { strandedAbsent, healed: appStreamSnapshot(subscribedInstance, true) }); + expect(strandedAbsent).toEqual(ALL_PRESENT); + expect(appStreamSnapshot(subscribedInstance, true)).toEqual(ALL_PRESENT); + }); +}); diff --git a/app/lib/services/connectionListeners.ts b/app/lib/services/connectionListeners.ts new file mode 100644 index 00000000000..91f0d09e236 --- /dev/null +++ b/app/lib/services/connectionListeners.ts @@ -0,0 +1,33 @@ +import { store } from '../store/auxStore'; +import { connectSuccess, disconnect as disconnectAction } from '../../actions/connect'; +import { loginRequest } from '../../actions/login'; + +/** + * Builds the `'connected'` stream listener that `connect()` registers on the current SDK instance. + * A `'connected'` DDP event only follows a real handshake on a (re)opened socket, whose server-side + * subscriptions are empty, so recovery must always run: `connectSuccess` is idempotent and the resume + * `loginRequest` is deduped by `takeLatest`. Extracted so the connection-lifecycle regression suite + * binds this REAL dispatch logic instead of a drifting replica. + */ +export function createConnectedListener(logoutOnError: boolean) { + return () => { + store.dispatch(connectSuccess()); + const { user } = store.getState().login; + if (user?.token) { + store.dispatch(loginRequest({ resume: user.token }, logoutOnError)); + } + }; +} + +/** + * Builds the `'close'` stream listener that `connect()` registers on the current SDK instance. + * `onClose` is injected so this module stays free of the VoIP dependency tree: `connect()` passes an + * `onClose` that arms its closure-local pendingHangups drain flag, while the regression suite passes a + * test double. The guard/dispatch logic itself is exercised as REAL code. + */ +export function createCloseListener({ onClose }: { onClose?: () => void }) { + return () => { + onClose?.(); + store.dispatch(disconnectAction()); + }; +} diff --git a/app/lib/services/connectionRestore.test.ts b/app/lib/services/connectionRestore.test.ts new file mode 100644 index 00000000000..59669dbe8cd --- /dev/null +++ b/app/lib/services/connectionRestore.test.ts @@ -0,0 +1,111 @@ +import { registerStreamRestorer, bindStreamRestoration } from './connectionRestore'; + +let mockGeneration = 1; +const mockOnStreamData = jest.fn, [string, () => void]>(() => Promise.resolve({ stop: jest.fn() })); +jest.mock('./sdk', () => ({ + __esModule: true, + default: { + get generation() { + return mockGeneration; + }, + onStreamData: (event: string, cb: () => void) => mockOnStreamData(event, cb) + } +})); + +const mockLog = jest.fn(); +jest.mock('../methods/helpers/log', () => ({ + __esModule: true, + default: (error: unknown) => mockLog(error) +})); + +const flushMicrotasks = () => new Promise(resolve => setImmediate(resolve)); + +describe('connectionRestore', () => { + const disposers: (() => void)[] = []; + + beforeEach(() => { + jest.clearAllMocks(); + mockGeneration = 1; + disposers.length = 0; + }); + + afterEach(() => { + disposers.forEach(dispose => dispose()); + }); + + const bindAndGetFanout = async () => { + await bindStreamRestoration(); + const lastCall = mockOnStreamData.mock.calls[mockOnStreamData.mock.calls.length - 1]; + expect(lastCall[0]).toBe('login'); + return lastCall[1]; + }; + + it('runs every enrolled restorer when the login generation still matches', async () => { + const first = jest.fn(); + const second = jest.fn(); + disposers.push(registerStreamRestorer(first), registerStreamRestorer(second)); + + const fanout = await bindAndGetFanout(); + fanout(); + await flushMicrotasks(); + + expect(first).toHaveBeenCalledTimes(1); + expect(second).toHaveBeenCalledTimes(1); + }); + + it('disposer removes exactly its own restorer', async () => { + const kept = jest.fn(); + const removed = jest.fn(); + const disposeRemoved = registerStreamRestorer(removed); + disposers.push(registerStreamRestorer(kept)); + + disposeRemoved(); + + const fanout = await bindAndGetFanout(); + fanout(); + await flushMicrotasks(); + + expect(removed).not.toHaveBeenCalled(); + expect(kept).toHaveBeenCalledTimes(1); + }); + + it('no-ops when the SDK generation advanced past the bound generation', async () => { + const restorer = jest.fn(); + disposers.push(registerStreamRestorer(restorer)); + + const fanout = await bindAndGetFanout(); + mockGeneration = 2; + fanout(); + await flushMicrotasks(); + + expect(restorer).not.toHaveBeenCalled(); + }); + + it('logs a synchronously throwing restorer and still runs the others', async () => { + const boom = jest.fn(() => { + throw new Error('sync boom'); + }); + const ok = jest.fn(); + disposers.push(registerStreamRestorer(boom), registerStreamRestorer(ok)); + + const fanout = await bindAndGetFanout(); + fanout(); + await flushMicrotasks(); + + expect(ok).toHaveBeenCalledTimes(1); + expect(mockLog).toHaveBeenCalledTimes(1); + }); + + it('logs a rejecting restorer and still runs the others', async () => { + const rejecting = jest.fn(() => Promise.reject(new Error('async boom'))); + const ok = jest.fn(); + disposers.push(registerStreamRestorer(rejecting), registerStreamRestorer(ok)); + + const fanout = await bindAndGetFanout(); + fanout(); + await flushMicrotasks(); + + expect(ok).toHaveBeenCalledTimes(1); + expect(mockLog).toHaveBeenCalledTimes(1); + }); +}); diff --git a/app/lib/services/connectionRestore.ts b/app/lib/services/connectionRestore.ts new file mode 100644 index 00000000000..ee9c4049ca8 --- /dev/null +++ b/app/lib/services/connectionRestore.ts @@ -0,0 +1,42 @@ +import log from '../methods/helpers/log'; +import sdk from './sdk'; + +export type StreamRestorer = () => void | Promise; + +// Instance-unique ids so two RoomSubscriptions for the same rid never clobber each other's entry. +let sequence = 0; +const restorers = new Map(); + +/** Enroll a consumer's stream-restoration callback. Returns a disposer that removes exactly this entry. */ +export function registerStreamRestorer(restore: StreamRestorer): () => void { + sequence += 1; + const id = sequence; + restorers.set(id, restore); + return () => { + restorers.delete(id); + }; +} + +function runRestorers(): void { + restorers.forEach(restore => { + try { + Promise.resolve(restore()).catch(error => log(error)); + } catch (error) { + log(error); + } + }); +} + +/** + * Bind the generation-keyed `'login'` listener that fans out to every enrolled restorer. + * Called by `connect()` after `sdk.initialize` (the only SDK-instance swap) and stored like the + * other connect() listeners so it is stopped on the next `connect()`. The generation check drops a + * stale-instance `'login'` a superseded connect() left behind. + */ +export function bindStreamRestoration(): Promise<{ stop: () => void }> { + const { generation } = sdk; + return sdk.onStreamData('login', () => { + if (sdk.generation !== generation) return; + runRestorers(); + }); +} diff --git a/app/lib/services/sdk.ts b/app/lib/services/sdk.ts index d8be776f45a..dde6fec3ffb 100644 --- a/app/lib/services/sdk.ts +++ b/app/lib/services/sdk.ts @@ -17,6 +17,9 @@ import { compareServerVersion, random } from '../methods/helpers'; class Sdk { private sdk: typeof Rocketchat; private code: any; + // Monotonic id of the current SDK instance, bumped on every `initialize()`. Lets callers tell + // which socket generation a listener/subscription belongs to across instance swaps. + private generationCount = 0; private initializeSdk(server: string): typeof Rocketchat { // The app can't reconnect if reopen interval is 5s while in development @@ -26,6 +29,7 @@ class Sdk { // TODO: We need to stop returning the SDK after all methods are dehydrated initialize(server: string) { this.code = null; + this.generationCount += 1; this.sdk = this.initializeSdk(server); return this.sdk; } @@ -34,6 +38,10 @@ class Sdk { return this.sdk; } + get generation() { + return this.generationCount; + } + /** * TODO: evaluate the need for assigning "null" to this.sdk * I'm returning "null" because we need to remove both instances of this.sdk here and on rocketchat.js diff --git a/app/reducers/server.test.ts b/app/reducers/server.test.ts index f9f8e2b6d5d..b518b61a967 100644 --- a/app/reducers/server.test.ts +++ b/app/reducers/server.test.ts @@ -27,7 +27,7 @@ describe('test server reducer', () => { it('should return modified store after selectServerFailure', () => { mockedStore.dispatch(selectServerFailure()); const state = mockedStore.getState().server; - const manipulated = { ...initialState, connecting: false, connected: false, loading: false, changingServer: false }; + const manipulated = { ...initialState, connecting: false, loading: false, changingServer: false }; expect(state).toEqual(manipulated); }); @@ -44,7 +44,7 @@ describe('test server reducer', () => { const name = 'Rocket.Chat'; mockedStore.dispatch(selectServerSuccess({ server, version, name: 'Rocket.Chat' })); const state = mockedStore.getState().server; - const manipulated = { ...initialState, server, version, connected: true, loading: false, name }; + const manipulated = { ...initialState, server, version, loading: false, name }; expect(state).toEqual(manipulated); }); diff --git a/app/reducers/server.ts b/app/reducers/server.ts index 006204f55e3..7700230bc1b 100644 --- a/app/reducers/server.ts +++ b/app/reducers/server.ts @@ -3,7 +3,6 @@ import { SERVER } from '../actions/actionsTypes'; export interface IServer { connecting: boolean; - connected: boolean; failure: boolean; failureMessage?: string; server: string; @@ -16,7 +15,6 @@ export interface IServer { export const initialState: IServer = { connecting: false, - connected: false, failure: false, server: '', version: '', @@ -39,7 +37,6 @@ export default function server(state = initialState, action: TActionServer): ISe return { ...state, connecting: false, - connected: false, failure: true, failureMessage: action.failureMessage }; @@ -47,7 +44,6 @@ export default function server(state = initialState, action: TActionServer): ISe return { ...state, connecting: false, - connected: true, loading: false, changingServer: false }; @@ -58,7 +54,6 @@ export default function server(state = initialState, action: TActionServer): ISe name: null, previousServer: null, connecting: false, - connected: false, loading: false, changingServer: false, failure: false @@ -69,7 +64,6 @@ export default function server(state = initialState, action: TActionServer): ISe server: action.server, version: action.version, connecting: true, - connected: false, loading: true, changingServer: action.changeServer }; @@ -80,7 +74,6 @@ export default function server(state = initialState, action: TActionServer): ISe version: action.version, name: action.name, connecting: false, - connected: true, loading: false, changingServer: false, failureMessage: undefined @@ -89,7 +82,6 @@ export default function server(state = initialState, action: TActionServer): ISe return { ...state, connecting: false, - connected: false, loading: false, changingServer: false }; diff --git a/app/sagas/__tests__/deepLinking.test.ts b/app/sagas/__tests__/deepLinking.test.ts index 632d05f369f..5155cb2386f 100644 --- a/app/sagas/__tests__/deepLinking.test.ts +++ b/app/sagas/__tests__/deepLinking.test.ts @@ -96,11 +96,11 @@ import { applyMiddleware, createStore } from 'redux'; import createSagaMiddleware from 'redux-saga'; import { deepLinkingOpen, deepLinkingClickCallPush } from '../../actions/deepLinking'; -import { loginSuccess } from '../../actions/login'; -import { selectServerSuccess } from '../../actions/server'; +import { loginSuccess, loginFailure, logout } from '../../actions/login'; +import { selectServerSuccess, selectServerFailure } from '../../actions/server'; import { connectSuccess } from '../../actions/connect'; import { appStart } from '../../actions/app'; -import { LOGIN } from '../../actions/actionsTypes'; +import { LOGIN, SERVER } from '../../actions/actionsTypes'; import { RootEnum } from '../../definitions'; import reducers from '../../reducers'; import deepLinkingRoot from '../deepLinking'; @@ -727,3 +727,182 @@ describe('deepLinking saga — handleOAuth dedup guard', () => { }); }); }); + +// ─── Same-server warm start — honest-session gate ───────────────────────── +// The socket can die without SERVER.SELECT_* ever resetting selection state. The warm-start +// same-server branch must gate on the honest flags (state.meteor.connected + state.login.isAuthenticated), +// re-running connect/login on a dead or stranded session before navigating onto it. + +describe('deepLinking saga — same-server warm start gates on honest session state', () => { + const makeSameServerParams = (overrides: Record = {}) => makeParams({ path: 'channel/general', ...overrides }); + + const preload = ({ meteorConnected, isAuthenticated }: { meteorConnected: boolean; isAuthenticated: boolean }) => + ({ + meteor: { connecting: false, connected: meteorConnected }, + login: { + isLocalAuthenticated: true, + isAuthenticated, + isFetching: false, + user: { id: 'u-me', token: TOKEN }, + error: {}, + services: {}, + failure: false + } + } as unknown as PreloadedState); + + beforeEach(() => { + jest.mocked(UserPreferences.getString).mockReset(); + jest.mocked(getServerById).mockReset(); + jest.mocked(canOpenRoom).mockReset(); + jest.mocked(waitForNavigationReady).mockReset(); + jest.mocked(goRoom).mockReset(); + jest.mocked(navigateToRoom).mockReset(); + jest.mocked(database.active.get).mockReset(); + + // Same host as the current server, with stored credentials and a known server record. + jest.mocked(UserPreferences.getString).mockImplementation((key: string) => (key === 'currentServer' ? HOST : TOKEN)); + jest.mocked(getServerById).mockResolvedValue(makeServerRecord() as any); + jest.mocked(canOpenRoom).mockResolvedValue({ rid: 'room-1', name: 'general', t: 'c' } as any); + jest.mocked(waitForNavigationReady).mockResolvedValue(undefined); + jest.mocked(goRoom).mockResolvedValue(undefined); + jest.mocked(database.active.get).mockReturnValue({ + find: jest.fn().mockResolvedValue({ rid: 'room-1', name: 'general', t: 'c' }) + } as any); + }); + + const selectRequested = (actions: { type: string }[]) => actions.some(a => a.type === SERVER.SELECT_REQUEST); + + it('re-runs connect/login before navigating when the socket is dead (notification tap)', async () => { + const { store, actions } = setupRecordingStore(preload({ meteorConnected: false, isAuthenticated: true })); + + store.dispatch(deepLinkingOpen(makeSameServerParams())); + await flushSagaMicrotasks(); + await flushSagaMicrotasks(); + + // Pipeline re-runs and navigation is parked at take(LOGIN.SUCCESS). + expect(selectRequested(actions)).toBe(true); + expect(jest.mocked(goRoom)).not.toHaveBeenCalled(); + + store.dispatch(loginSuccess({ id: 'user-1', token: TOKEN } as any)); + await flushSagaMicrotasks(); + await flushSagaMicrotasks(); + + // Navigation proceeds only after the login pipeline completes. + expect(jest.mocked(goRoom)).toHaveBeenCalledTimes(1); + }); + + it('re-runs the pipeline when the socket is up but the resume login stranded (authenticated=false)', async () => { + const { store, actions } = setupRecordingStore(preload({ meteorConnected: true, isAuthenticated: false })); + + store.dispatch(deepLinkingOpen(makeSameServerParams())); + await flushSagaMicrotasks(); + await flushSagaMicrotasks(); + + expect(selectRequested(actions)).toBe(true); + expect(jest.mocked(goRoom)).not.toHaveBeenCalled(); + + store.dispatch(loginSuccess({ id: 'user-1', token: TOKEN } as any)); + await flushSagaMicrotasks(); + await flushSagaMicrotasks(); + + expect(jest.mocked(goRoom)).toHaveBeenCalledTimes(1); + }); + + it('skips connect/login and navigates immediately when connected and authenticated', async () => { + const { store, actions } = setupRecordingStore(preload({ meteorConnected: true, isAuthenticated: true })); + + store.dispatch(deepLinkingOpen(makeSameServerParams())); + await flushSagaMicrotasks(); + await flushSagaMicrotasks(); + + expect(selectRequested(actions)).toBe(false); + expect(jest.mocked(goRoom)).toHaveBeenCalledTimes(1); + }); + + it('does not navigate when the resume login fails, and a later LOGIN.SUCCESS never wakes the stale saga', async () => { + const { store, actions } = setupRecordingStore(preload({ meteorConnected: false, isAuthenticated: true })); + + store.dispatch(deepLinkingOpen(makeSameServerParams())); + await flushSagaMicrotasks(); + await flushSagaMicrotasks(); + + // Pipeline re-ran and navigation is parked on the LOGIN race. + expect(selectRequested(actions)).toBe(true); + expect(jest.mocked(goRoom)).not.toHaveBeenCalled(); + + // Resume login fails permanently — the race resolves to failure, no navigation. + store.dispatch(loginFailure({ message: 'resume failed' })); + await flushSagaMicrotasks(); + await flushSagaMicrotasks(); + + expect(jest.mocked(goRoom)).not.toHaveBeenCalled(); + + // A later, unrelated LOGIN.SUCCESS must NOT resurrect the abandoned navigation. + store.dispatch(loginSuccess({ id: 'user-1', token: TOKEN } as any)); + await flushSagaMicrotasks(); + await flushSagaMicrotasks(); + + expect(jest.mocked(goRoom)).not.toHaveBeenCalled(); + }); + + it('does not navigate when server selection fails before login, and a later LOGIN.SUCCESS never wakes the stale saga', async () => { + const { store, actions } = setupRecordingStore(preload({ meteorConnected: false, isAuthenticated: true })); + + store.dispatch(deepLinkingOpen(makeSameServerParams())); + await flushSagaMicrotasks(); + await flushSagaMicrotasks(); + + expect(selectRequested(actions)).toBe(true); + expect(jest.mocked(goRoom)).not.toHaveBeenCalled(); + + // connect()/SSL/DB throws before login → selectServerFailure, no LOGIN.* nor LOGOUT. + store.dispatch(selectServerFailure()); + await flushSagaMicrotasks(); + await flushSagaMicrotasks(); + + expect(jest.mocked(goRoom)).not.toHaveBeenCalled(); + + store.dispatch(loginSuccess({ id: 'user-1', token: TOKEN } as any)); + await flushSagaMicrotasks(); + await flushSagaMicrotasks(); + + expect(jest.mocked(goRoom)).not.toHaveBeenCalled(); + }); + + it('does not navigate when a LOGOUT interrupts the resume login', async () => { + const { store } = setupRecordingStore(preload({ meteorConnected: false, isAuthenticated: true })); + + store.dispatch(deepLinkingOpen(makeSameServerParams())); + await flushSagaMicrotasks(); + await flushSagaMicrotasks(); + + store.dispatch(logout()); + await flushSagaMicrotasks(); + await flushSagaMicrotasks(); + + expect(jest.mocked(goRoom)).not.toHaveBeenCalled(); + + store.dispatch(loginSuccess({ id: 'user-1', token: TOKEN } as any)); + await flushSagaMicrotasks(); + await flushSagaMicrotasks(); + + expect(jest.mocked(goRoom)).not.toHaveBeenCalled(); + }); + + it('applies the same gate on the call-push warm start (dead socket re-runs pipeline)', async () => { + const { store, actions } = setupRecordingStore(preload({ meteorConnected: false, isAuthenticated: true })); + + store.dispatch(deepLinkingClickCallPush({ host: HOST, rid: 'room-1' })); + await flushSagaMicrotasks(); + await flushSagaMicrotasks(); + + expect(selectRequested(actions)).toBe(true); + expect(jest.mocked(navigateToRoom)).not.toHaveBeenCalled(); + + store.dispatch(loginSuccess({ id: 'user-1', token: TOKEN } as any)); + await flushSagaMicrotasks(); + await flushSagaMicrotasks(); + + expect(jest.mocked(navigateToRoom)).toHaveBeenCalledTimes(1); + }); +}); diff --git a/app/sagas/__tests__/login.test.ts b/app/sagas/__tests__/login.test.ts new file mode 100644 index 00000000000..6666cc3cae2 --- /dev/null +++ b/app/sagas/__tests__/login.test.ts @@ -0,0 +1,271 @@ +// ─── Boundary mocks — must appear before any import that triggers the module ─── +// The retry tests exercise handleLoginRequest's resume-login backoff. Only `login` (the resume call) +// and the redux state/timers matter; every other module is a native/network leaf reached only by the +// success pipeline, mocked here so LOGIN.SUCCESS stays inert. + +jest.mock('../../lib/services/connect', () => ({ + login: jest.fn(), + loginWithPassword: jest.fn(), + disconnect: jest.fn() +})); + +jest.mock('../../lib/methods/helpers/localAuthentication', () => ({ + localAuthenticate: jest.fn(() => Promise.resolve()) +})); + +jest.mock('../../lib/database', () => ({ + __esModule: true, + default: { + servers: { + get: () => ({ + query: () => ({ fetch: () => Promise.resolve([]) }), + find: () => Promise.reject(new Error('not found')) + }), + write: (cb: () => Promise) => cb() + } + } +})); + +jest.mock('../../lib/services/sdk', () => ({ + __esModule: true, + default: { subscribe: jest.fn(), current: { client: { host: '' } } } +})); + +jest.mock('../../lib/services/restApi', () => ({ + saveUserProfile: jest.fn(() => Promise.resolve({ user: {} })), + registerPushToken: jest.fn(() => Promise.resolve()), + getUsersRoles: jest.fn(() => Promise.resolve([])), + setUserPresenceAway: jest.fn(() => Promise.resolve()) +})); + +jest.mock('../../lib/methods/userPreferences', () => ({ + __esModule: true, + default: { setString: jest.fn(), getString: jest.fn() } +})); + +jest.mock('../../lib/methods/getCustomEmojis', () => ({ getCustomEmojis: jest.fn(() => Promise.resolve()) })); +jest.mock('../../lib/methods/getPermissions', () => ({ getPermissions: jest.fn(() => Promise.resolve()) })); +jest.mock('../../lib/methods/getRoles', () => ({ getRoles: jest.fn(() => Promise.resolve()) })); +jest.mock('../../lib/methods/getSlashCommands', () => ({ getSlashCommands: jest.fn(() => Promise.resolve()) })); +jest.mock('../../lib/methods/getSettings', () => ({ subscribeSettings: jest.fn(() => Promise.resolve()) })); +jest.mock('../../lib/methods/getUsersPresence', () => ({ + getUserPresence: jest.fn(() => Promise.resolve()), + refreshDmUsersPresence: jest.fn(() => Promise.resolve()), + subscribeUsersPresence: jest.fn(() => Promise.resolve()) +})); +jest.mock('../../lib/methods/enterpriseModules', () => ({ + getEnterpriseModules: jest.fn(() => Promise.resolve()), + isOmnichannelModuleAvailable: jest.fn(() => false), + isVoipModuleAvailable: jest.fn(() => false) +})); +jest.mock('../../lib/methods/logout', () => ({ + logout: jest.fn(() => Promise.resolve()), + removeServerData: jest.fn(() => Promise.resolve()), + removeServerDatabase: jest.fn(() => Promise.resolve()) +})); +jest.mock('../../lib/methods/helpers/helpers', () => ({ hasPermission: jest.fn(() => Promise.resolve([false, false])) })); +jest.mock('../../lib/methods/helpers/info', () => ({ showErrorAlert: jest.fn() })); +jest.mock('../../lib/methods/helpers/log', () => ({ + __esModule: true, + default: jest.fn(), + events: {}, + logEvent: jest.fn() +})); +jest.mock('../../lib/hooks/useMasterDetail', () => ({ getIsMasterDetail: jest.fn(() => false) })); +jest.mock('../../lib/database/services/Server', () => ({ getServerById: jest.fn(() => Promise.resolve(null)) })); +jest.mock('../../lib/navigation/appNavigation', () => ({ __esModule: true, default: { navigate: jest.fn() } })); +jest.mock('../../containers/ActionSheet', () => ({ showActionSheetRef: jest.fn() })); +jest.mock('../../containers/SupportedVersions', () => ({ SupportedVersionsWarning: () => null })); +jest.mock('../../lib/services/voip/MediaSessionInstance', () => ({ + mediaSessionInstance: { init: jest.fn(), reset: jest.fn() } +})); +jest.mock('../../lib/services/voip/MediaSessionStore', () => ({ + mediaSessionStore: { getCurrentInstance: jest.fn(() => null) } +})); +jest.mock('../../ee/omnichannel/lib', () => ({ isOmnichannelStatusAvailable: jest.fn(() => false) })); +jest.mock('../../ee/omnichannel/actions/inquiry', () => ({ + inquiryRequest: jest.fn(() => ({ type: 'INQUIRY_REQUEST' })), + inquiryReset: jest.fn(() => ({ type: 'INQUIRY_RESET' })) +})); + +// ─── Real imports (after mocks) ─────────────────────────────────────────────── +/* eslint-disable import/first, import/order */ +import { applyMiddleware, createStore } from 'redux'; +import createSagaMiddleware from 'redux-saga'; + +import loginRoot from '../login'; +import { loginRequest, clearUser } from '../../actions/login'; +import { selectServerRequest } from '../../actions/server'; +import { LOGIN, LOGOUT } from '../../actions/actionsTypes'; +import reducers from '../../reducers'; +import { login as loginService } from '../../lib/services/connect'; +/* eslint-enable import/first, import/order */ + +const SERVER = 'https://open.rocket.chat'; +const TOKEN = 'resume-token'; +const USER = { id: 'u-me', username: 'me', name: 'Me', token: TOKEN }; + +async function flushSagaMicrotasks(): Promise { + await Promise.resolve(); + await Promise.resolve(); + await Promise.resolve(); +} + +function setupStore() { + const actions: { type: string }[] = []; + const recorder = () => (next: (a: any) => any) => (action: any) => { + actions.push(action); + return next(action); + }; + const sagaMiddleware = createSagaMiddleware(); + const preloadedState: any = { + login: { + isLocalAuthenticated: true, + isAuthenticated: true, + isFetching: false, + user: { ...USER }, + error: {}, + services: {}, + failure: false + }, + server: { server: SERVER, version: '6.0.0', name: 'open', connecting: false, loading: false } + }; + const store = createStore(reducers, preloadedState, applyMiddleware(recorder, sagaMiddleware)); + sagaMiddleware.run(loginRoot); + return { store, actions }; +} + +const typesOf = (actions: { type: string }[], type: string) => actions.filter(a => a.type === type); + +describe('login saga — resume-login retry with backoff', () => { + beforeEach(() => { + jest.useFakeTimers(); + jest.mocked(loginService).mockReset(); + }); + + afterEach(() => { + jest.useRealTimers(); + }); + + it('retries a transient resume failure with 2s/4s/8s backoff, then succeeds', async () => { + jest + .mocked(loginService) + .mockRejectedValueOnce(new Error('network blip')) + .mockRejectedValueOnce(new Error('network blip')) + .mockResolvedValueOnce(USER as any); + + const { store, actions } = setupStore(); + store.dispatch(loginRequest({ resume: TOKEN }, false)); + + // First attempt fails synchronously -> parked in delay(2000). + await flushSagaMicrotasks(); + expect(jest.mocked(loginService)).toHaveBeenCalledTimes(1); + expect(typesOf(actions, LOGIN.SUCCESS)).toHaveLength(0); + expect(typesOf(actions, LOGIN.FAILURE)).toHaveLength(0); + + await jest.advanceTimersByTimeAsync(2000); + await flushSagaMicrotasks(); + expect(jest.mocked(loginService)).toHaveBeenCalledTimes(2); + + await jest.advanceTimersByTimeAsync(4000); + await flushSagaMicrotasks(); + expect(jest.mocked(loginService)).toHaveBeenCalledTimes(3); + + // Third attempt resolves -> loginSuccess, never loginFailure. + expect(typesOf(actions, LOGIN.SUCCESS)).toHaveLength(1); + expect(typesOf(actions, LOGIN.FAILURE)).toHaveLength(0); + }); + + it('dispatches loginFailure after exhausting the bounded retries', async () => { + jest.mocked(loginService).mockRejectedValue(new Error('still down')); + + const { store, actions } = setupStore(); + store.dispatch(loginRequest({ resume: TOKEN }, false)); + + // Initial attempt + 3 retries = 4 total, across 2s/4s/8s backoffs. + await flushSagaMicrotasks(); + await jest.advanceTimersByTimeAsync(2000); + await flushSagaMicrotasks(); + await jest.advanceTimersByTimeAsync(4000); + await flushSagaMicrotasks(); + await jest.advanceTimersByTimeAsync(8000); + await flushSagaMicrotasks(); + + expect(jest.mocked(loginService)).toHaveBeenCalledTimes(4); + expect(typesOf(actions, LOGIN.FAILURE)).toHaveLength(1); + }); + + it('does not retry a 401 — logs the session out instead', async () => { + jest.mocked(loginService).mockRejectedValue({ status: 401 }); + + const { store, actions } = setupStore(); + store.dispatch(loginRequest({ resume: TOKEN }, false)); + + await flushSagaMicrotasks(); + await jest.advanceTimersByTimeAsync(20000); + await flushSagaMicrotasks(); + + expect(jest.mocked(loginService)).toHaveBeenCalledTimes(1); + expect(typesOf(actions, LOGOUT)).toHaveLength(1); + expect(typesOf(actions, LOGIN.FAILURE)).toHaveLength(0); + }); + + it('does not retry when the server logged the user out', async () => { + jest.mocked(loginService).mockRejectedValue({ data: { message: "You've been logged out by the server" } }); + + const { store, actions } = setupStore(); + store.dispatch(loginRequest({ resume: TOKEN }, false)); + + await flushSagaMicrotasks(); + await jest.advanceTimersByTimeAsync(20000); + await flushSagaMicrotasks(); + + expect(jest.mocked(loginService)).toHaveBeenCalledTimes(1); + expect(typesOf(actions, LOGOUT)).toHaveLength(1); + }); + + it('does not retry when the session has expired', async () => { + jest.mocked(loginService).mockRejectedValue({ data: { message: 'Your session has expired' } }); + + const { store, actions } = setupStore(); + store.dispatch(loginRequest({ resume: TOKEN }, false)); + + await flushSagaMicrotasks(); + await jest.advanceTimersByTimeAsync(20000); + await flushSagaMicrotasks(); + + expect(jest.mocked(loginService)).toHaveBeenCalledTimes(1); + expect(typesOf(actions, LOGOUT)).toHaveLength(1); + }); + + it('bails out mid-backoff when the user token is gone', async () => { + jest.mocked(loginService).mockRejectedValue(new Error('network blip')); + + const { store, actions } = setupStore(); + store.dispatch(loginRequest({ resume: TOKEN }, false)); + await flushSagaMicrotasks(); + + // User logs out during the wait -> the abandoned session must not be re-logged-in. + store.dispatch(clearUser()); + await jest.advanceTimersByTimeAsync(2000); + await flushSagaMicrotasks(); + + expect(jest.mocked(loginService)).toHaveBeenCalledTimes(1); + expect(typesOf(actions, LOGIN.FAILURE)).toHaveLength(1); + }); + + it('bails out mid-backoff when the server changed', async () => { + jest.mocked(loginService).mockRejectedValue(new Error('network blip')); + + const { store, actions } = setupStore(); + store.dispatch(loginRequest({ resume: TOKEN }, false)); + await flushSagaMicrotasks(); + + store.dispatch(selectServerRequest('https://another.server.com', '6.0.0')); + await jest.advanceTimersByTimeAsync(2000); + await flushSagaMicrotasks(); + + expect(jest.mocked(loginService)).toHaveBeenCalledTimes(1); + expect(typesOf(actions, LOGIN.FAILURE)).toHaveLength(1); + }); +}); diff --git a/app/sagas/__tests__/state.test.ts b/app/sagas/__tests__/state.test.ts new file mode 100644 index 00000000000..4d6027ff25a --- /dev/null +++ b/app/sagas/__tests__/state.test.ts @@ -0,0 +1,100 @@ +// ─── Boundary mocks — must appear before any import that triggers the module ─── + +jest.mock('../../lib/methods/helpers/localAuthentication', () => ({ + localAuthenticate: jest.fn(() => Promise.resolve()), + saveLastLocalAuthenticationSession: jest.fn(() => Promise.resolve()) +})); +jest.mock('../../lib/services/connect', () => ({ checkAndReopen: jest.fn(() => Promise.resolve()) })); +jest.mock('../../lib/services/restApi', () => ({ + setUserPresenceOnline: jest.fn(() => Promise.resolve()), + setUserPresenceAway: jest.fn(() => Promise.resolve()) +})); +jest.mock('../../lib/notifications', () => ({ checkPendingNotification: jest.fn(() => Promise.resolve()) })); +jest.mock('../../lib/methods/helpers/log', () => ({ __esModule: true, default: jest.fn() })); + +// ─── Real imports (after mocks) ─────────────────────────────────────────────── +/* eslint-disable import/first, import/order */ +import { applyMiddleware, createStore } from 'redux'; +import createSagaMiddleware from 'redux-saga'; + +import stateRoot from '../state'; +import { APP_STATE, LOGIN } from '../../actions/actionsTypes'; +import { RootEnum } from '../../definitions'; +import reducers from '../../reducers'; +import { checkAndReopen } from '../../lib/services/connect'; +/* eslint-enable import/first, import/order */ + +const SERVER = 'https://open.rocket.chat'; +const TOKEN = 'resume-token'; + +async function flushSagaMicrotasks(): Promise { + await Promise.resolve(); + await Promise.resolve(); + await Promise.resolve(); +} + +function setupStore(loginOverrides: Record) { + const actions: { type: string }[] = []; + const recorder = () => (next: (a: any) => any) => (action: any) => { + actions.push(action); + return next(action); + }; + const sagaMiddleware = createSagaMiddleware(); + const preloadedState: any = { + app: { root: RootEnum.ROOT_INSIDE, foreground: false, background: true, ready: true, netInfoState: null }, + login: { + isLocalAuthenticated: true, + isAuthenticated: false, + isFetching: false, + user: {}, + error: {}, + services: {}, + failure: false, + ...loginOverrides + }, + server: { server: SERVER, version: '6.0.0', name: 'open', loading: false } + }; + const store = createStore(reducers, preloadedState, applyMiddleware(recorder, sagaMiddleware)); + sagaMiddleware.run(stateRoot); + return { store, actions }; +} + +const loginRequests = (actions: { type: string }[]) => actions.filter(a => a.type === LOGIN.REQUEST); + +describe('state saga — foreground heal after a stranded resume login', () => { + beforeEach(() => { + jest.mocked(checkAndReopen).mockClear(); + }); + + it('dispatches a resume loginRequest when a stored token exists but the session is not authenticated', async () => { + const { store, actions } = setupStore({ isAuthenticated: false, user: { id: 'u-me', token: TOKEN } }); + + store.dispatch({ type: APP_STATE.FOREGROUND }); + await flushSagaMicrotasks(); + + expect(loginRequests(actions)).toHaveLength(1); + expect((loginRequests(actions)[0] as any).credentials).toEqual({ resume: TOKEN }); + // Heal branch is minimal: it must not touch the authenticated reconnect path. + expect(jest.mocked(checkAndReopen)).not.toHaveBeenCalled(); + }); + + it('does nothing when there is no stored user token', async () => { + const { store, actions } = setupStore({ isAuthenticated: false, user: {} }); + + store.dispatch({ type: APP_STATE.FOREGROUND }); + await flushSagaMicrotasks(); + + expect(loginRequests(actions)).toHaveLength(0); + expect(jest.mocked(checkAndReopen)).not.toHaveBeenCalled(); + }); + + it('takes the authenticated reconnect path without a heal loginRequest', async () => { + const { store, actions } = setupStore({ isAuthenticated: true, user: { id: 'u-me', token: TOKEN } }); + + store.dispatch({ type: APP_STATE.FOREGROUND }); + await flushSagaMicrotasks(); + + expect(loginRequests(actions)).toHaveLength(0); + expect(jest.mocked(checkAndReopen)).toHaveBeenCalledTimes(1); + }); +}); diff --git a/app/sagas/deepLinking.js b/app/sagas/deepLinking.js index b6892f18b39..e9d18fdd27c 100644 --- a/app/sagas/deepLinking.js +++ b/app/sagas/deepLinking.js @@ -1,7 +1,7 @@ import { InteractionManager } from 'react-native'; import RNCallKeep from 'react-native-callkeep'; import I18n from 'i18n-js'; -import { all, call, delay, put, select, take, takeLatest } from 'redux-saga/effects'; +import { all, call, delay, put, race, select, take, takeLatest } from 'redux-saga/effects'; import { shareSetParams } from '../actions/share'; import * as types from '../actions/actionsTypes'; @@ -166,6 +166,28 @@ const handleShareExtension = function* handleOpen({ params }) { yield put(appStart({ root: RootEnum.ROOT_SHARE_EXTENSION })); }; +// The socket can die without SERVER.SELECT_* ever resetting selection state, so gate on the +// honest session flags: only skip connect/login when the websocket is up and the resume login +// already authenticated. Otherwise re-run the pipeline before navigating onto a dead socket. +// Race LOGIN.SUCCESS against the terminal outcomes so a failed re-run returns false instead of +// parking under takeLatest to be woken (and mis-navigated) by a later unrelated LOGIN.SUCCESS: +// SELECT_FAILURE (connect/SSL/DB throw before login), LOGIN.FAILURE, LOGOUT. +const ensureSameServerSession = function* ensureSameServerSession(host, version) { + const sessionReady = yield select(state => state.meteor.connected && state.login.isAuthenticated); + if (sessionReady) { + return true; + } + yield localAuthenticate(host); + yield put(selectServerRequest(host, version, true)); + const { success } = yield race({ + success: take(types.LOGIN.SUCCESS), + loginFailure: take(types.LOGIN.FAILURE), + selectFailure: take(types.SERVER.SELECT_FAILURE), + logout: take(types.LOGOUT) + }); + return !!success; +}; + const handleOpen = function* handleOpen({ params }) { if (params.type === 'shareextension') { yield handleShareExtension({ params }); @@ -201,11 +223,9 @@ const handleOpen = function* handleOpen({ params }) { // TODO: needs better test // if deep link is from same server if (server === host && user && serverRecord) { - const connected = yield select(state => state.server.connected); - if (!connected) { - yield localAuthenticate(host); - yield put(selectServerRequest(host, serverRecord.version, true)); - yield take(types.LOGIN.SUCCESS); + const sessionReady = yield ensureSameServerSession(host, serverRecord.version); + if (!sessionReady) { + return; } yield completeDeepLinkNavigation(params); } else { @@ -316,11 +336,9 @@ const handleClickCallPush = function* handleClickCallPush({ params }) { const serverRecord = yield getServerById(host); if (server === host && user && serverRecord) { - const connected = yield select(state => state.server.connected); - if (!connected) { - yield localAuthenticate(host); - yield put(selectServerRequest(host, serverRecord.version, true)); - yield take(types.LOGIN.SUCCESS); + const sessionReady = yield ensureSameServerSession(host, serverRecord.version); + if (!sessionReady) { + return; } yield handleNavigateCallRoom({ params }); } else { diff --git a/app/sagas/login.js b/app/sagas/login.js index 10740600a33..2dc489df330 100644 --- a/app/sagas/login.js +++ b/app/sagas/login.js @@ -21,17 +21,14 @@ import UserPreferences from '../lib/methods/userPreferences'; import { inquiryRequest, inquiryReset } from '../ee/omnichannel/actions/inquiry'; import { isOmnichannelStatusAvailable } from '../ee/omnichannel/lib'; import { RootEnum } from '../definitions'; -import sdk from '../lib/services/sdk'; import { CURRENT_SERVER, TOKEN_KEY } from '../lib/constants/keys'; import { getCustomEmojis } from '../lib/methods/getCustomEmojis'; import { getIsMasterDetail } from '../lib/hooks/useMasterDetail'; import { getEnterpriseModules, isOmnichannelModuleAvailable, isVoipModuleAvailable } from '../lib/methods/enterpriseModules'; import { getPermissions } from '../lib/methods/getPermissions'; -import { getRoles } from '../lib/methods/getRoles'; import { getSlashCommands } from '../lib/methods/getSlashCommands'; -import { getUserPresence, refreshDmUsersPresence, subscribeUsersPresence } from '../lib/methods/getUsersPresence'; +import { getUserPresence } from '../lib/methods/getUsersPresence'; import { logout, removeServerData, removeServerDatabase } from '../lib/methods/logout'; -import { subscribeSettings } from '../lib/methods/getSettings'; import { disconnect, loginWithPassword, login } from '../lib/services/connect'; import { saveUserProfile, registerPushToken, getUsersRoles, setUserPresenceAway } from '../lib/services/restApi'; import { setUsersRoles } from '../actions/usersRoles'; @@ -73,88 +70,106 @@ const showSupportedVersionsWarning = function* showSupportedVersionsWarning(serv } }; +// Transient resume-login failures (network blip, server hiccup) get bounded retries with exponential +// backoff so a stranded session (connected socket, zero server-side subs) heals itself. +const RESUME_LOGIN_MAX_RETRIES = 3; +const RESUME_LOGIN_BACKOFF_MS = 2000; + const handleLoginRequest = function* handleLoginRequest({ credentials, logoutOnError = false, registerCustomFields }) { logEvent(events.LOGIN_DEFAULT_LOGIN); - try { - let result; - if (credentials.resume) { - result = yield loginCall(credentials); - } else { - result = yield call(loginWithPasswordCall, credentials); - } - if (!result.username) { - yield put(serverFinishAdd()); - yield put(setUser(result)); - yield put(appStart({ root: RootEnum.ROOT_SET_USERNAME })); - } else { - const server = yield select(getServer); - yield localAuthenticate(server); + const initialServer = yield select(getServer); + let retries = 0; + while (true) { + try { + let result; + if (credentials.resume) { + result = yield loginCall(credentials); + } else { + result = yield call(loginWithPasswordCall, credentials); + } + if (!result.username) { + yield put(serverFinishAdd()); + yield put(setUser(result)); + yield put(appStart({ root: RootEnum.ROOT_SET_USERNAME })); + } else { + const server = yield select(getServer); + yield localAuthenticate(server); - // Saves username on server history - const serversDB = database.servers; - const serversHistoryCollection = serversDB.get('servers_history'); - const serversCollection = serversDB.get('servers'); - yield serversDB.write(async () => { - try { - const serversHistory = await serversHistoryCollection.query(Q.where('url', server)).fetch(); - if (serversHistory?.length) { - const serverHistoryRecord = serversHistory[0]; - // Get server iconURL from servers table - let iconURL = null; - try { - const serverRecord = await serversCollection.find(server); - iconURL = serverRecord.iconURL; - } catch (e) { - // Server record might not exist yet - } - // this is updating on every login just to save `updated_at` - // keeping this server as the most recent on autocomplete order - await serverHistoryRecord.update(s => { - s.username = result.username; - if (iconURL) { - s.iconURL = iconURL; + // Saves username on server history + const serversDB = database.servers; + const serversHistoryCollection = serversDB.get('servers_history'); + const serversCollection = serversDB.get('servers'); + yield serversDB.write(async () => { + try { + const serversHistory = await serversHistoryCollection.query(Q.where('url', server)).fetch(); + if (serversHistory?.length) { + const serverHistoryRecord = serversHistory[0]; + // Get server iconURL from servers table + let iconURL = null; + try { + const serverRecord = await serversCollection.find(server); + iconURL = serverRecord.iconURL; + } catch (e) { + // Server record might not exist yet } - }); + // this is updating on every login just to save `updated_at` + // keeping this server as the most recent on autocomplete order + await serverHistoryRecord.update(s => { + s.username = result.username; + if (iconURL) { + s.iconURL = iconURL; + } + }); + } + } catch (e) { + log(e); } - } catch (e) { - log(e); + }); + yield put(loginSuccess(result)); + if (registerCustomFields) { + const updatedUser = yield call(saveUserProfile, {}, { ...registerCustomFields }); + yield put(setUser({ ...result, ...updatedUser.user })); } - }); - yield put(loginSuccess(result)); - if (registerCustomFields) { - const updatedUser = yield call(saveUserProfile, {}, { ...registerCustomFields }); - yield put(setUser({ ...result, ...updatedUser.user })); } - } - } catch (e) { - if (e?.data?.message && /you've been logged out by the server/i.test(e.data.message)) { - logEvent(events.LOGOUT_BY_SERVER); - yield put(logoutAction(true, 'Logged_out_by_server')); - } else if (e?.data?.message && /your session has expired/i.test(e.data.message)) { - logEvent(events.LOGOUT_TOKEN_EXPIRED); - yield put(logoutAction(true, 'Token_expired')); - } else if (e?.status === 401) { - logEvent(events.LOGIN_DEFAULT_LOGIN_F); - const userId = yield select(state => state.login.user.id); - if (!userId) { + return; + } catch (e) { + if (e?.data?.message && /you've been logged out by the server/i.test(e.data.message)) { + logEvent(events.LOGOUT_BY_SERVER); + yield put(logoutAction(true, 'Logged_out_by_server')); + return; + } + if (e?.data?.message && /your session has expired/i.test(e.data.message)) { + logEvent(events.LOGOUT_TOKEN_EXPIRED); + yield put(logoutAction(true, 'Token_expired')); + return; + } + if (e?.status === 401) { + logEvent(events.LOGIN_DEFAULT_LOGIN_F); + const userId = yield select(state => state.login.user.id); + if (!userId) { + yield put(loginFailure(e)); + return; + } + yield put(logoutAction(true)); + return; + } + if (!credentials.resume || retries >= RESUME_LOGIN_MAX_RETRIES) { yield put(loginFailure(e)); return; } - yield put(logoutAction(true)); - } else { - yield put(loginFailure(e)); + yield delay(RESUME_LOGIN_BACKOFF_MS * 2 ** retries); + // Bail if the session became irrelevant during the wait: user logged out or server switched. + const token = yield select(state => state.login.user?.token); + const server = yield select(getServer); + if (!token || server !== initialServer) { + yield put(loginFailure(e)); + return; + } + retries += 1; } } }; -const subscribeSettingsFork = function* subscribeSettingsFork() { - try { - yield subscribeSettings(); - } catch (e) { - log(e); - } -}; - const fetchPermissions = function* fetchPermissions() { try { yield getPermissions(); @@ -171,15 +186,6 @@ const fetchCustomEmojisFork = function* fetchCustomEmojisFork() { } }; -const fetchRolesFork = function* fetchRolesFork() { - try { - sdk.subscribe('stream-roles', 'roles'); - yield getRoles(); - } catch (e) { - log(e); - } -}; - const fetchSlashCommandsFork = function* fetchSlashCommandsFork() { try { yield getSlashCommands(); @@ -196,15 +202,6 @@ const registerPushTokenFork = function* registerPushTokenFork() { } }; -const fetchUsersPresenceFork = function* fetchUsersPresenceFork() { - try { - yield subscribeUsersPresence(); - yield refreshDmUsersPresence(); - } catch (e) { - log(e); - } -}; - const fetchEnterpriseModules = function* fetchEnterpriseModules({ user }) { try { yield getEnterpriseModules(); @@ -300,11 +297,8 @@ const handleLoginSuccess = function* handleLoginSuccess({ user }) { yield call(fetchPermissions); yield call(fetchEnterpriseModules, { user }); yield fork(fetchCustomEmojisFork); - yield fork(fetchRolesFork); yield fork(fetchSlashCommandsFork); yield fork(registerPushTokenFork); - yield fork(fetchUsersPresenceFork); - yield fork(subscribeSettingsFork); yield fork(fetchUsersRoles); yield fork(checkBackgroundAndSetAway); yield getUserPresence(user.id); diff --git a/app/sagas/state.js b/app/sagas/state.js index 44f0b1430fa..e033545122a 100644 --- a/app/sagas/state.js +++ b/app/sagas/state.js @@ -1,8 +1,9 @@ -import { select, takeLatest } from 'redux-saga/effects'; +import { put, select, takeLatest } from 'redux-saga/effects'; import log from '../lib/methods/helpers/log'; import { localAuthenticate, saveLastLocalAuthenticationSession } from '../lib/methods/helpers/localAuthentication'; import { APP_STATE } from '../actions/actionsTypes'; +import { loginRequest } from '../actions/login'; import { RootEnum } from '../definitions'; import { checkAndReopen } from '../lib/services/connect'; import { setUserPresenceOnline, setUserPresenceAway } from '../lib/services/restApi'; @@ -21,6 +22,12 @@ const appHasComeBackToForeground = function* appHasComeBackToForeground() { } const login = yield select(state => state.login); if (!login.isAuthenticated) { + // Stranded after a failed resume login: a stored token with no authenticated session. Heal by + // re-running the resume login instead of leaving rooms silently dead until a real transport close. + const token = login.user?.token; + if (token) { + yield put(loginRequest({ resume: token }, false)); + } return; } try {