diff --git a/CONTEXT.md b/CONTEXT.md index 1399e8e8b1d..30d903b1115 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -180,11 +180,12 @@ A **Message Action** is the active mode on a Message in the Room view. The three ## Server & Connection -| Term | Definition | Aliases to avoid | -| ------------------ | ------------------------------------------------------------------------------------------------- | ---------------------------------------------------------------- | -| **Server** | A Rocket.Chat server instance the app connects to, with version, settings, and enterprise modules | Workspace (used by web but not consistently in mobile), instance | -| **Server History** | List of previously connected Servers for quick reconnection | Recent servers | -| **Meteor Connect** | The WebSocket connection to the Server's DDP (Distributed Data Protocol) endpoint | Socket, connection | +| Term | Definition | Aliases to avoid | +| ------------------ | ------------------------------------------------------------------------------------------------------------------------ | ---------------------------------------------------------------- | +| **Server** | A Rocket.Chat server instance the app connects to, with version, settings, and enterprise modules | Workspace (used by web but not consistently in mobile), instance | +| **Server History** | List of previously connected Servers for quick reconnection | Recent servers | +| **Meteor Connect** | The WebSocket connection to the Server's DDP (Distributed Data Protocol) endpoint | Socket, connection | +| **Socket Health** | Whether the Meteor Connect socket is genuinely alive — confirmed by a round trip when in doubt, reopened when known dead | Staleness (stale/gray/fresh), socket probe | ## Navigation & Layout diff --git a/app/containers/NewMediaCall/VoipCallLifecycle.integration.test.tsx b/app/containers/NewMediaCall/VoipCallLifecycle.integration.test.tsx index 37a40761bbe..3d0b22be100 100644 --- a/app/containers/NewMediaCall/VoipCallLifecycle.integration.test.tsx +++ b/app/containers/NewMediaCall/VoipCallLifecycle.integration.test.tsx @@ -24,6 +24,7 @@ import Navigation from '../../lib/navigation/appNavigation'; import { usePeerAutocompleteStore } from '../../lib/services/voip/usePeerAutocompleteStore'; import { useCallStore } from '../../lib/services/voip/useCallStore'; import { mediaSessionInstance } from '../../lib/services/voip/MediaSessionInstance'; +import { acceptNativeCallWithReadiness } from '../../lib/services/voip/acceptNativeCall'; import { mockedStore } from '../../reducers/mockedStore'; import type { TPeerItem } from '../../lib/services/voip/getPeerAutocompleteOptions'; import type { InsideStackParamList } from '../../stacks/types'; @@ -157,6 +158,15 @@ jest.mock('../../lib/services/voip/getPeerAutocompleteOptions', () => ({ jest.mock('../../lib/services/voip/navigateToCallRoom', () => ({ navigateToCallRoom: jest.fn().mockResolvedValue(undefined) })); +// Gate boundary mock: the DDP listener now routes accepted signals through +// acceptNativeCallWithReadiness rather than calling answerCall directly. The +// gate's own unit tests cover readiness orchestration; this file asserts the +// lifecycle/navigation contract, so the gate is short-circuited to answerCall. +jest.mock('../../lib/services/voip/acceptNativeCall', () => ({ + acceptNativeCallWithReadiness: jest.fn(async (_callId: string, mediaSession: any) => { + await mediaSession.answerCall(_callId); + }) +})); // playCallEndedSound → expo-av → Audio.Sound constructor not present in this test boundary. jest.mock('../../lib/services/voip/playCallEndedSound', () => ({ playCallEndedSound: jest.fn() @@ -541,6 +551,9 @@ describe('VoIP call lifecycle (integration)', () => { }); // ── MediaSessionInstance contract: answerCall ──────────────────────────── + // Incoming accepted signals now flow through acceptNativeCallWithReadiness + // (readiness gate). The gate module is mocked here to delegate straight to + // answerCall; gate readiness is covered by acceptNativeCall.test.ts. describe('MediaSessionInstance contract: answerCall', () => { it('A1: DDP accepted signal with native pre-accept → answerCall navigates to CallView', async () => { @@ -566,6 +579,9 @@ describe('VoIP call lifecycle (integration)', () => { expect(RNCallKeep.setCurrentCallActive as jest.Mock).toHaveBeenCalledWith('incoming-1'); expect(Navigation.navigate).toHaveBeenCalledWith('CallView'); expect(useCallStore.getState().call?.callId).toBe('incoming-1'); + // The DDP listener now funnels accepted signals through the readiness + // gate instead of invoking answerCall directly. + expect(acceptNativeCallWithReadiness).toHaveBeenCalledWith('incoming-1', mediaSessionInstance); }); it('A2: accepted signal but call not found → RNCallKeep.endCall, no navigate', async () => { @@ -590,6 +606,7 @@ describe('VoIP call lifecycle (integration)', () => { expect(useCallStore.getState().nativeAcceptedCallId).toBeNull(); expect(Navigation.navigate).not.toHaveBeenCalled(); expect(useCallStore.getState().call).toBeNull(); + expect(acceptNativeCallWithReadiness).toHaveBeenCalledWith('missing-1', mediaSessionInstance); // Tighten: confirm the known-noise allowlist entry was actually triggered. expect(consoleErrorSpy).toHaveBeenCalledWith( expect.objectContaining({ message: '[VoIP] Call not found after accept: missing-1' }) diff --git a/app/lib/methods/helpers/index.ts b/app/lib/methods/helpers/index.ts index f7b99415159..5cb6dada580 100644 --- a/app/lib/methods/helpers/index.ts +++ b/app/lib/methods/helpers/index.ts @@ -10,6 +10,7 @@ export * from './isReadOnly'; export * from './media'; export * from './normalizeDeepLinkingServerHost'; export * from './normalizeStatusExpiresAt'; +export * from './onAbort'; export * from './room'; export * from './server'; export * from './isSsl'; diff --git a/app/lib/methods/helpers/onAbort.ts b/app/lib/methods/helpers/onAbort.ts new file mode 100644 index 00000000000..42033c1f0d9 --- /dev/null +++ b/app/lib/methods/helpers/onAbort.ts @@ -0,0 +1,10 @@ +export function onAbort(signal: AbortSignal | undefined, callback: () => void): void { + if (!signal) { + return; + } + if (signal.aborted) { + callback(); + return; + } + signal.addEventListener('abort', callback, { once: true }); +} diff --git a/app/lib/services/__tests__/socketHealth.integration.test.ts b/app/lib/services/__tests__/socketHealth.integration.test.ts new file mode 100644 index 00000000000..edd77e267b0 --- /dev/null +++ b/app/lib/services/__tests__/socketHealth.integration.test.ts @@ -0,0 +1,259 @@ +import sdk from '../sdk'; +import { recoverSocket } from '../socketHealth'; + +// eslint-disable-next-line @typescript-eslint/no-var-requires +const { DDPDriver } = require('@rocket.chat/sdk/lib/drivers/ddp') as { + DDPDriver: new (options: { host: string; logger: unknown }) => PatchedDriver; +}; + +interface MockConnection { + send: jest.Mock; + close: jest.Mock; + readyState: number; + onopen: () => void; + onmessage: (event: { data: string }) => void; + onerror: () => void; + onclose: () => void; +} + +interface WireFrame { + msg: string; + id?: string; + name?: string; + params?: string[]; +} + +interface PatchedDriver { + userId: string; + pingInterval: number; + reopenNow(): Promise; + waitForNotifyUserMediaSubs(timeoutMs?: number): Promise; + ddp: { + lastPing: number; + pingTimeout?: ReturnType; + openTimeout?: ReturnType; + open(): Promise; + send(message: Record): Promise; + subscriptions: Record; + }; +} + +const mockConnections: MockConnection[] = []; + +jest.mock('universal-websocket-client', () => + jest.fn().mockImplementation(() => { + const connection = { + send: jest.fn((data: string) => { + const message = JSON.parse(data) as { msg: string; id?: string }; + if (message.msg === 'connect') { + setImmediate(() => connection.onmessage({ data: JSON.stringify({ msg: 'connected', session: 'session-id' }) })); + } else if (message.msg === 'ping') { + setImmediate(() => connection.onmessage({ data: JSON.stringify({ msg: 'pong' }) })); + } else if (message.msg === 'sub') { + setImmediate(() => connection.onmessage({ data: JSON.stringify({ msg: 'ready', subs: [message.id] }) })); + } + }), + close: jest.fn(), + readyState: 1, + onopen: jest.fn(), + onmessage: jest.fn(), + onerror: jest.fn(), + onclose: jest.fn() + }; + mockConnections.push(connection); + return connection; + }) +); + +jest.mock('../sdk', () => ({ + __esModule: true, + default: { current: undefined } +})); + +const USER_ID = 'user-id'; +const PING_INTERVAL = 10000; + +const logger = { debug: jest.fn(), info: jest.fn(), error: jest.fn(), warn: jest.fn() }; + +/** Real patched DDPDriver over a mocked WebSocket, connected and logged in. */ +async function buildConnectedDriver() { + const driver = new DDPDriver({ host: 'localhost:3000', logger }); + driver.userId = USER_ID; + const openPromise = driver.ddp.open(); + mockConnections[0].onopen(); + await jest.advanceTimersByTimeAsync(0); + await openPromise; + return driver; +} + +function addMediaSubs(driver: PatchedDriver) { + ['media-signal', 'media-calls'].forEach((name, index) => { + const id = `sub-${index}`; + driver.ddp.subscriptions[id] = { + id, + name: 'stream-notify-user', + params: [`${USER_ID}/${name}`], + unsubscribe: jest.fn() + }; + }); +} + +function backdateLastPing(driver: PatchedDriver, ageMs: number) { + driver.ddp.lastPing = Date.now() - ageMs; +} + +/** Frames of a given `msg` sent over the wire on one connection. */ +function framesOn(connection: MockConnection, msg: string) { + return connection.send.mock.calls + .map(([data]: [string]) => JSON.parse(data) as WireFrame) + .filter(message => message.msg === msg); +} + +describe('recoverSocket against the real patched socket', () => { + let driver: PatchedDriver; + + beforeEach(async () => { + jest.clearAllMocks(); + jest.useFakeTimers(); + mockConnections.length = 0; + driver = await buildConnectedDriver(); + (sdk as unknown as { current: { ddp: PatchedDriver } }).current = { ddp: driver }; + }); + + afterEach(() => { + if (driver.ddp.pingTimeout) clearTimeout(driver.ddp.pingTimeout); + if (driver.ddp.openTimeout) clearTimeout(driver.ddp.openTimeout); + jest.useRealTimers(); + }); + + it('exposes the ping interval the health classification depends on', () => { + expect(driver.pingInterval).toBe(PING_INTERVAL); + }); + + it('keeps a doubtful socket when the round trip gets a pong', async () => { + backdateLastPing(driver, PING_INTERVAL + 5000); + + const recovery = recoverSocket(); + await jest.advanceTimersByTimeAsync(0); + + await expect(recovery).resolves.toBe('confirmed-alive'); + // The round trip pinged the existing socket and the pong kept it alive. + expect(framesOn(mockConnections[0], 'ping').length).toBeGreaterThan(0); + expect(mockConnections).toHaveLength(1); + }); + + it('reopens a doubtful socket when the round trip gets no pong', async () => { + backdateLastPing(driver, PING_INTERVAL + 5000); + // A zombie socket: still `readyState: 1`, but the server never answers. + mockConnections[0].send.mockImplementation(() => undefined); + + const recovery = recoverSocket(); + await jest.advanceTimersByTimeAsync(2000); + + // The round trip was actually attempted on the dead socket before reopening. + expect(framesOn(mockConnections[0], 'ping').length).toBeGreaterThan(0); + expect(mockConnections).toHaveLength(2); + mockConnections[1].onopen(); + await jest.advanceTimersByTimeAsync(0); + + await expect(recovery).resolves.toBe('reopened'); + }); + + it('reopens a frozen socket whose last ping is still young', async () => { + // A young `lastPing` proves nothing: `onOpen` refreshes it before the handshake + // reply lands, so the timestamp can sit on an unusable session. + mockConnections[0].send.mockImplementation(() => undefined); + + const recovery = recoverSocket(); + await jest.advanceTimersByTimeAsync(2000); + + // The young ping bought a round trip, and the silent socket failed it. + expect(framesOn(mockConnections[0], 'ping').length).toBeGreaterThan(0); + expect(mockConnections).toHaveLength(2); + mockConnections[1].onopen(); + await jest.advanceTimersByTimeAsync(0); + + await expect(recovery).resolves.toBe('reopened'); + }); + + it('reopens a known-dead socket without a round trip', async () => { + backdateLastPing(driver, PING_INTERVAL * 2 + 1000); + + const recovery = recoverSocket(); + await jest.advanceTimersByTimeAsync(0); + + expect(mockConnections).toHaveLength(2); + // No raw round-trip ping was sent on the dead socket. + expect(framesOn(mockConnections[0], 'ping')).toHaveLength(0); + + mockConnections[1].onopen(); + await jest.advanceTimersByTimeAsync(0); + + await expect(recovery).resolves.toBe('reopened'); + }); + + it('shares one reopen with a concurrent direct reopenNow', async () => { + backdateLastPing(driver, PING_INTERVAL * 3); + + // The foreground path reopens the dead socket while recovery does the same. + const directReopen = driver.reopenNow(); + const recovery = recoverSocket(); + + await jest.advanceTimersByTimeAsync(0); + expect(mockConnections).toHaveLength(2); + mockConnections[1].onopen(); + await jest.advanceTimersByTimeAsync(0); + + await directReopen; + await expect(recovery).resolves.toBe('reopened'); + expect(mockConnections).toHaveLength(2); + + // No queued third open fires later — the reopen really was shared. + await jest.advanceTimersByTimeAsync(60000); + expect(mockConnections).toHaveLength(2); + }); + + it('rejects an in-flight DDP method call when recovery reopens the socket', async () => { + let rejected = false; + const inFlight = driver.ddp.send({ msg: 'method', method: 'getRoomByTypeAndName', params: [] }).catch(() => { + rejected = true; + }); + await jest.advanceTimersByTimeAsync(0); + expect(rejected).toBe(false); + + // The socket dies silently after the call went out. + backdateLastPing(driver, PING_INTERVAL * 3); + + const recovery = recoverSocket(); + await jest.advanceTimersByTimeAsync(0); + await inFlight; + expect(rejected).toBe(true); + + mockConnections[1].onopen(); + await jest.advanceTimersByTimeAsync(0); + + await expect(recovery).resolves.toBe('reopened'); + }); + + it('re-sends the media subscriptions on the new socket reusing their ids', async () => { + backdateLastPing(driver, PING_INTERVAL * 3); + addMediaSubs(driver); + + const recovery = recoverSocket(); + await jest.advanceTimersByTimeAsync(0); + mockConnections[1].onopen(); + await jest.advanceTimersByTimeAsync(0); + await expect(recovery).resolves.toBe('reopened'); + + const resubscribed = driver.waitForNotifyUserMediaSubs(); + await jest.advanceTimersByTimeAsync(200); + await expect(resubscribed).resolves.toBe(true); + + // Both media subs went out on the new socket reusing their ids. + expect(framesOn(mockConnections[0], 'sub')).toHaveLength(0); + expect(framesOn(mockConnections[1], 'sub')).toEqual([ + expect.objectContaining({ id: 'sub-0', name: 'stream-notify-user', params: [`${USER_ID}/media-signal`] }), + expect.objectContaining({ id: 'sub-1', name: 'stream-notify-user', params: [`${USER_ID}/media-calls`] }) + ]); + }); +}); diff --git a/app/lib/services/__tests__/socketHealth.test.ts b/app/lib/services/__tests__/socketHealth.test.ts new file mode 100644 index 00000000000..031c7fe92a2 --- /dev/null +++ b/app/lib/services/__tests__/socketHealth.test.ts @@ -0,0 +1,166 @@ +jest.mock('../sdk', () => ({ + __esModule: true, + default: { + current: { ddp: undefined } + } +})); + +import sdk from '../sdk'; +import { classifySocketHealth, recoverSocket } from '../socketHealth'; + +const now = 1_000_000; + +const sdkMock = sdk as unknown as { current: { ddp: unknown } | undefined }; + +interface MockDdp { + connected?: boolean; + lastPing: number; + pingInterval?: number; + config?: { ping?: number }; + reopenNow: jest.Mock, []>; + probe: jest.Mock, [number]>; +} + +function makeDdp(overrides: Partial = {}): MockDdp { + return { + lastPing: now, + pingInterval: 10000, + config: { ping: 10000 }, + reopenNow: jest.fn, []>(() => Promise.resolve()), + probe: jest.fn, [number]>(() => Promise.resolve(true)), + ...overrides + }; +} + +describe('classifySocketHealth', () => { + beforeEach(() => { + jest.spyOn(Date, 'now').mockReturnValue(now); + }); + + afterEach(() => { + jest.restoreAllMocks(); + }); + + it('returns reopen when age > 2 * pingInterval', () => { + const ddp = makeDdp({ lastPing: now - 21000 }); + expect(classifySocketHealth(ddp)).toBe('reopen'); + }); + + it('returns round-trip-check when age <= 2 * pingInterval', () => { + const ddp = makeDdp({ lastPing: now - 15000 }); + expect(classifySocketHealth(ddp)).toBe('round-trip-check'); + }); + + it('returns round-trip-check for a young ping rather than trusting it outright', () => { + const ddp = makeDdp({ lastPing: now - 5000 }); + expect(classifySocketHealth(ddp)).toBe('round-trip-check'); + }); + + it('falls back to config.ping when pingInterval is missing', () => { + // Only a 30s config.ping keeps a 21s-old ping below the reopen threshold. + const ddp = makeDdp({ pingInterval: undefined, config: { ping: 30000 }, lastPing: now - 21000 }); + expect(classifySocketHealth(ddp)).toBe('round-trip-check'); + }); + + it('uses 10000ms default when pingInterval and config.ping are missing', () => { + const ddp = makeDdp({ pingInterval: undefined, config: {}, lastPing: now - 21000 }); + expect(classifySocketHealth(ddp)).toBe('reopen'); + }); + + it('returns reopen for a closed socket even when lastPing is fresh', () => { + const ddp = makeDdp({ connected: false, lastPing: now }); + expect(classifySocketHealth(ddp)).toBe('reopen'); + }); +}); + +describe('recoverSocket', () => { + let ddp: MockDdp; + + beforeEach(() => { + ddp = makeDdp({ lastPing: Date.now() }); + sdkMock.current = { ddp }; + }); + + it('keeps a socket whose round trip answers', async () => { + await expect(recoverSocket()).resolves.toBe('confirmed-alive'); + expect(ddp.reopenNow).not.toHaveBeenCalled(); + }); + + it('runs the round trip with a 2s budget', async () => { + await recoverSocket(); + expect(ddp.probe).toHaveBeenCalledWith(2000); + }); + + it('reopens when the round trip goes unanswered', async () => { + ddp.probe.mockResolvedValue(false); + await expect(recoverSocket()).resolves.toBe('reopened'); + expect(ddp.reopenNow).toHaveBeenCalledTimes(1); + }); + + it('reopens a known-dead socket without a round trip', async () => { + ddp.connected = false; + await expect(recoverSocket()).resolves.toBe('reopened'); + expect(ddp.probe).not.toHaveBeenCalled(); + expect(ddp.reopenNow).toHaveBeenCalledTimes(1); + }); + + it('reports no-socket when the ddp handle is missing', async () => { + sdkMock.current = { ddp: undefined }; + await expect(recoverSocket()).resolves.toBe('no-socket'); + expect(ddp.probe).not.toHaveBeenCalled(); + expect(ddp.reopenNow).not.toHaveBeenCalled(); + }); + + it('reports no-socket when there is no sdk instance', async () => { + sdkMock.current = undefined; + await expect(recoverSocket()).resolves.toBe('no-socket'); + }); + + it('rejects when the round trip throws', async () => { + ddp.probe.mockRejectedValue(new Error('round trip failed')); + await expect(recoverSocket()).rejects.toThrow('round trip failed'); + }); + + it('rejects when reopening throws', async () => { + ddp.connected = false; + ddp.reopenNow.mockRejectedValue(new Error('reopen failed')); + await expect(recoverSocket()).rejects.toThrow('reopen failed'); + }); + + it('shares one in-flight recovery between overlapping callers', async () => { + const outcomes = await Promise.all([recoverSocket(), recoverSocket()]); + expect(outcomes).toEqual(['confirmed-alive', 'confirmed-alive']); + expect(ddp.probe).toHaveBeenCalledTimes(1); + }); + + it('starts a fresh recovery after the shared one settles', async () => { + await recoverSocket(); + await recoverSocket(); + expect(ddp.probe).toHaveBeenCalledTimes(2); + }); + + it('abandons the aborted caller while the shared recovery runs on', async () => { + let answerRoundTrip: (alive: boolean) => void = () => {}; + ddp.probe.mockImplementation(() => new Promise(resolve => (answerRoundTrip = resolve))); + + const controller = new AbortController(); + const aborted = recoverSocket({ abortSignal: controller.signal }); + const other = recoverSocket(); + + controller.abort(); + await expect(aborted).resolves.toBe('abandoned'); + + answerRoundTrip(true); + await expect(other).resolves.toBe('confirmed-alive'); + expect(ddp.probe).toHaveBeenCalledTimes(1); + }); + + it('abandons a pre-aborted caller without touching the socket', async () => { + const controller = new AbortController(); + controller.abort(); + + await expect(recoverSocket({ abortSignal: controller.signal })).resolves.toBe('abandoned'); + expect(ddp.probe).not.toHaveBeenCalled(); + expect(ddp.reopenNow).not.toHaveBeenCalled(); + }); +}); diff --git a/app/lib/services/connect.ts b/app/lib/services/connect.ts index 603677ff87b..1f9043efd1f 100644 --- a/app/lib/services/connect.ts +++ b/app/lib/services/connect.ts @@ -10,6 +10,7 @@ import database from '../database'; import { twoFactor } from './twoFactor'; import { store } from '../store/auxStore'; import { loginRequest, logout, setLoginServices, setUser } from '../../actions/login'; +import { waitForLoginReady } from './waitForLoginReady'; import sdk from './sdk'; import { mediaSessionInstance } from './voip/MediaSessionInstance'; import { pendingHangups } from './voip/pendingHangups'; @@ -40,13 +41,6 @@ interface IServices { service: string; } -// Reads redux rather than `ddp.loggedIn`: `close` clears `meteor.connected`, while `ddp.loggedIn` survives it. -// Neither survives a silent background death, so callers must bound their wait. -function isLoginReady(): boolean { - const state = store.getState(); - return state.login.isAuthenticated && state.meteor.connected; -} - let connectingListener: any; let connectedListener: any; let closeListener: any; @@ -149,21 +143,7 @@ function connect({ server, logoutOnError = false }: { server: string; logoutOnEr pendingHangupsDrainArmed = false; if (pendingHangups.size === 0) return; try { - if (!isLoginReady()) { - await new Promise(resolve => { - const unsub = store.subscribe(() => { - if (isLoginReady()) { - clearTimeout(timer); - unsub(); - resolve(); - } - }); - const timer = setTimeout(() => { - unsub(); - resolve(); - }, 5000); - }); - } + await waitForLoginReady(5000); await mediaSessionInstance.drainPendingHangups(); } catch (error) { log(error); @@ -459,10 +439,6 @@ function abort() { } } -function checkAndReopen() { - return sdk.current.checkAndReopen(); -} - function disconnect() { const result = sdk.disconnect(); mediaSessionInstance.reset(); @@ -554,12 +530,12 @@ export { loginTOTP, loginWithPassword, loginOAuthOrSso, - checkAndReopen, abort, connect, disconnect, getWebsocketInfo, stopListener, getLoginServices, - determineAuthType + determineAuthType, + waitForLoginReady }; diff --git a/app/lib/services/ddpSocket.test.ts b/app/lib/services/ddpSocket.test.ts new file mode 100644 index 00000000000..44acd6e7320 --- /dev/null +++ b/app/lib/services/ddpSocket.test.ts @@ -0,0 +1,365 @@ +// eslint-disable-next-line @typescript-eslint/no-var-requires +const { Socket, DDPDriver } = require('@rocket.chat/sdk/lib/drivers/ddp'); + +const mockConnections: any[] = []; +const trackedSockets: any[] = []; + +jest.mock('universal-websocket-client', () => { + return jest.fn().mockImplementation(() => { + const connection = { + send: jest.fn((data: string) => { + const message = JSON.parse(data); + if (message.msg === 'connect') { + setImmediate(() => connection.onmessage({ data: JSON.stringify({ msg: 'connected', session: 'session-id' }) })); + } else if (message.msg === 'ping') { + setImmediate(() => connection.onmessage({ data: JSON.stringify({ msg: 'pong' }) })); + } + }), + close: jest.fn(), + readyState: 1, + onopen: jest.fn(), + onmessage: jest.fn(), + onerror: jest.fn(), + onclose: jest.fn() + }; + mockConnections.push(connection); + return connection; + }); +}); + +const buildSocket = () => { + const socket = new Socket({ + logger: { debug: jest.fn(), info: jest.fn(), error: jest.fn(), warn: jest.fn() }, + timeout: 10000 + }); + trackedSockets.push(socket); + const send = jest.fn(); + const close = jest.fn(); + socket.connection = { + send, + close, + readyState: 1, + onopen: jest.fn(), + onmessage: jest.fn(), + onerror: jest.fn(), + onclose: jest.fn() + }; + return { socket, send, close }; +}; + +const trackSocket = (socket: any) => { + trackedSockets.push(socket); + return socket; +}; + +beforeEach(() => { + mockConnections.length = 0; + trackedSockets.length = 0; +}); + +afterEach(() => { + trackedSockets.forEach(socket => { + if (socket.openTimeout) clearTimeout(socket.openTimeout as any); + if (socket.pingTimeout) clearTimeout(socket.pingTimeout as any); + }); +}); + +describe('Socket.probe', () => { + afterEach(() => { + jest.useRealTimers(); + }); + + it('resolves true when pong arrives within deadline', async () => { + const { socket } = buildSocket(); + const probePromise = socket.probe(); + socket.lastPing += 1; + socket.emit('pong'); + await expect(probePromise).resolves.toBe(true); + }); + + it('resolves false when no pong arrives within 2s deadline', async () => { + jest.useFakeTimers(); + const { socket } = buildSocket(); + const probePromise = socket.probe(); + await jest.advanceTimersByTimeAsync(2000); + await expect(probePromise).resolves.toBe(false); + }); + + it('resolves false when raw connection.send throws', async () => { + const { socket, send } = buildSocket(); + send.mockImplementation(() => { + throw new Error('boom'); + }); + await expect(socket.probe()).resolves.toBe(false); + }); + + it('resolves false when readyState is not open', async () => { + const { socket } = buildSocket(); + socket.connection.readyState = 2; + await expect(socket.probe()).resolves.toBe(false); + }); + + it('ignores a stale pong that does not advance lastPing', async () => { + jest.useFakeTimers(); + const { socket } = buildSocket(); + const initialLastPing = Date.now() - 1000; + socket.lastPing = initialLastPing; + + const probePromise = socket.probe(); + socket.emit('pong'); + + await jest.advanceTimersByTimeAsync(2000); + await expect(probePromise).resolves.toBe(false); + }); +}); + +describe('Socket.reopenNow', () => { + afterEach(() => { + jest.useRealTimers(); + }); + + it('preserves subscriptions and subscribeAll re-sends them', async () => { + const { socket } = buildSocket(); + const subscription = { + id: 'sub-1', + name: 'stream-room-messages', + params: ['rid'], + unsubscribe: jest.fn() + }; + socket.subscriptions['sub-1'] = subscription; + + const sendSpy = jest.spyOn(socket, 'send').mockResolvedValue({ subs: ['sub-1'] }); + + const reopenPromise = socket.reopenNow(); + mockConnections[0].onopen(); + await reopenPromise; + + expect(socket.subscriptions['sub-1']).toBe(subscription); + + await socket.subscribeAll(); + + expect(sendSpy).toHaveBeenCalledWith( + expect.objectContaining({ + msg: 'sub', + id: 'sub-1', + name: 'stream-room-messages', + params: ['rid'] + }) + ); + }); + + it("emits 'disconnected' and rejects in-flight send()", async () => { + const { socket } = buildSocket(); + const disconnectedListener = jest.fn(); + socket.on('disconnected', disconnectedListener); + const sendPromise = socket.send({ msg: 'ping' }); + + const reopenPromise = socket.reopenNow(); + + expect(disconnectedListener).toHaveBeenCalledTimes(1); + await expect(sendPromise).rejects.toBeUndefined(); + + mockConnections[0].onopen(); + await reopenPromise; + }); + + it('concurrent calls create exactly one new WebSocket', async () => { + const socket = trackSocket( + new Socket({ + logger: { debug: jest.fn(), info: jest.fn(), error: jest.fn(), warn: jest.fn() }, + timeout: 10000 + }) + ); + + const a = socket.reopenNow(); + const b = socket.reopenNow(); + + expect(mockConnections).toHaveLength(1); + + mockConnections[0].onopen(); + + await Promise.all([a, b]); + }); + + it('times out and clears in-flight state so a later reopenNow retries', async () => { + jest.useFakeTimers(); + const socket = trackSocket( + new Socket({ + logger: { debug: jest.fn(), info: jest.fn(), error: jest.fn(), warn: jest.fn() }, + timeout: 10000 + }) + ); + + const promise = socket.reopenNow(); + expect(socket.reopenPromise).toBeTruthy(); + + await jest.advanceTimersByTimeAsync(10000); + await promise; + + expect(socket.reopenPromise).toBeUndefined(); + + const secondPromise = socket.reopenNow(); + expect(mockConnections).toHaveLength(2); + + mockConnections[1].onopen(); + await jest.runOnlyPendingTimersAsync(); + await secondPromise; + }); + + it('forces a reconnect on an already healthy socket', async () => { + const { socket } = buildSocket(); + const initialConnection = socket.connection; + + const promise = socket.reopenNow(); + + expect(mockConnections).toHaveLength(1); + expect(initialConnection.close).toHaveBeenCalled(); + + mockConnections[0].onopen(); + await promise; + + expect(socket.connection).toBe(mockConnections[0]); + }); + + it('serializes against concurrent open(): no second socket, no closing in-flight one', async () => { + const socket = trackSocket( + new Socket({ + logger: { debug: jest.fn(), info: jest.fn(), error: jest.fn(), warn: jest.fn() }, + timeout: 10000 + }) + ); + + const reopenPromise = socket.reopenNow(); + const inFlightConnection = mockConnections[0]; + + const openPromise = socket.open(); + expect(mockConnections).toHaveLength(1); + expect(inFlightConnection.close).not.toHaveBeenCalled(); + + mockConnections[0].onopen(); + await reopenPromise; + await openPromise; + }); +}); + +describe('Socket.send disconnected listener', () => { + it('cleans up the disconnected listener after send resolves', async () => { + const { socket, send } = buildSocket(); + const baseline = socket._listeners.disconnected?.length || 0; + send.mockImplementation(() => { + setImmediate(() => socket.emit('pong', { msg: 'pong' })); + }); + + await socket.send({ msg: 'ping' }); + + expect(socket._listeners.disconnected?.length || 0).toBe(baseline); + }); +}); + +describe('DDPDriver.waitForNotifyUserMediaSubs', () => { + afterEach(() => { + jest.useRealTimers(); + }); + + const makeDriver = () => + new DDPDriver({ + logger: { debug: jest.fn(), info: jest.fn(), error: jest.fn(), warn: jest.fn() } + }); + + it('resolves true when media subs are present and server acks', async () => { + const driver = makeDriver(); + driver.userId = 'uid'; + driver.ddp.subscriptions['sub-ms'] = { + id: 'sub-ms', + name: 'stream-notify-user', + params: ['uid/media-signal'], + unsubscribe: jest.fn() + }; + driver.ddp.subscriptions['sub-mc'] = { + id: 'sub-mc', + name: 'stream-notify-user', + params: ['uid/media-calls'], + unsubscribe: jest.fn() + }; + jest.spyOn(driver.ddp, 'subscribe').mockResolvedValue({}); + + await expect(driver.waitForNotifyUserMediaSubs(1000)).resolves.toBe(true); + expect(driver.ddp.subscribe).toHaveBeenCalledWith('stream-notify-user', ['uid/media-signal'], undefined, 'sub-ms'); + expect(driver.ddp.subscribe).toHaveBeenCalledWith('stream-notify-user', ['uid/media-calls'], undefined, 'sub-mc'); + }); + + it('waits for media subs to appear before re-subscribing', async () => { + jest.useFakeTimers(); + const driver = makeDriver(); + driver.userId = 'uid'; + jest.spyOn(driver.ddp, 'subscribe').mockResolvedValue({}); + + const promise = driver.waitForNotifyUserMediaSubs(1000); + driver.ddp.subscriptions['sub-ms'] = { + id: 'sub-ms', + name: 'stream-notify-user', + params: ['uid/media-signal'], + unsubscribe: jest.fn() + }; + driver.ddp.subscriptions['sub-mc'] = { + id: 'sub-mc', + name: 'stream-notify-user', + params: ['uid/media-calls'], + unsubscribe: jest.fn() + }; + + await jest.advanceTimersByTimeAsync(100); + await expect(promise).resolves.toBe(true); + expect(driver.ddp.subscribe).toHaveBeenCalledTimes(2); + }); + + it('stays pending while only one of the media subs is present', async () => { + jest.useFakeTimers(); + const driver = makeDriver(); + driver.userId = 'uid'; + jest.spyOn(driver.ddp, 'subscribe').mockResolvedValue({}); + + let resolved: boolean | undefined; + const promise = driver.waitForNotifyUserMediaSubs(1000).then((value: boolean) => { + resolved = value; + return value; + }); + + driver.ddp.subscriptions['sub-ms'] = { + id: 'sub-ms', + name: 'stream-notify-user', + params: ['uid/media-signal'], + unsubscribe: jest.fn() + }; + + await jest.advanceTimersByTimeAsync(100); + expect(resolved).toBeUndefined(); + expect(driver.ddp.subscribe).not.toHaveBeenCalled(); + + driver.ddp.subscriptions['sub-mc'] = { + id: 'sub-mc', + name: 'stream-notify-user', + params: ['uid/media-calls'], + unsubscribe: jest.fn() + }; + + await jest.advanceTimersByTimeAsync(100); + await expect(promise).resolves.toBe(true); + }); + + it('resolves false if media subs never appear before the timeout', async () => { + jest.useFakeTimers(); + const driver = makeDriver(); + driver.userId = 'uid'; + + const promise = driver.waitForNotifyUserMediaSubs(500); + await jest.advanceTimersByTimeAsync(500); + + await expect(promise).resolves.toBe(false); + }); + + it('resolves false when userId is missing', async () => { + const driver = makeDriver(); + await expect(driver.waitForNotifyUserMediaSubs(1000)).resolves.toBe(false); + }); +}); diff --git a/app/lib/services/socketHealth.ts b/app/lib/services/socketHealth.ts new file mode 100644 index 00000000000..aa7d4ceb5f1 --- /dev/null +++ b/app/lib/services/socketHealth.ts @@ -0,0 +1,135 @@ +import { onAbort } from '../methods/helpers/onAbort'; +import sdk from './sdk'; + +/** + * The slice of the patched DDP driver this module reads. + * The only guard is `sdk.current?.ddp` being undefined — the patch is guaranteed + * at runtime, so there are no per-method typeof checks. + */ +interface SocketHealthDdp { + connected?: boolean; + lastPing: number; + pingInterval?: number; + config?: { ping?: number }; + reopenNow(): Promise; + probe(timeoutMs: number): Promise; +} + +/** + * The recovery plan — what classification decides. + * `'round-trip-check'` means a stored ping timestamp can't vouch for a socket + * the OS may have frozen, so anything young enough is verified by a round trip, + * never trusted outright. + * + * Exported for unit tests; callers never branch on it — they call + * `recoverSocket()` and see outcomes. + */ +export type SocketRecoveryPlan = 'reopen' | 'round-trip-check'; + +export function classifySocketHealth(ddp: SocketHealthDdp): SocketRecoveryPlan { + // Ping age can't vouch for a socket the OS already closed. + if (ddp.connected === false) { + return 'reopen'; + } + const pingInterval = (ddp.pingInterval ?? ddp.config?.ping) || 10000; + const age = Date.now() - ddp.lastPing; + if (age > pingInterval * 2) { + return 'reopen'; + } + // Anything younger is verified by a round trip, never trusted outright: onOpen + // refreshes lastPing before the handshake reply lands. + return 'round-trip-check'; +} + +/** + * What a recovery attempt reports. + * - `'confirmed-alive'` — round trip succeeded; nothing was done. + * - `'reopened'` — socket reopened (stale ping, or round trip failed). + * - `'no-socket'` — `sdk.current?.ddp` undefined; nothing to recover. + * - `'abandoned'` — caller's abort signal fired while waiting; the + * underlying recovery (shared — see below) runs on. + * + * Errors from `reopenNow()`/`probe()` REJECT the promise rather than becoming + * an outcome: both current callers already sit in catch paths (`state.js` + * logs, accept gate fails the call), and a thrown error is not a decision the + * module can make for them. + */ +export type SocketRecoveryOutcome = 'confirmed-alive' | 'reopened' | 'no-socket' | 'abandoned'; + +let inFlightRecovery: Promise | null = null; + +function shareRecovery(): Promise { + if (inFlightRecovery) { + return inFlightRecovery; + } + const ddp = sdk.current?.ddp as SocketHealthDdp | undefined; + if (!ddp) { + return Promise.resolve('no-socket'); + } + const recovery = (async (): Promise => { + if (classifySocketHealth(ddp) === 'reopen') { + await ddp.reopenNow(); + return 'reopened'; + } + const alive = await ddp.probe(2000); + if (alive) { + return 'confirmed-alive'; + } + await ddp.reopenNow(); + return 'reopened'; + })(); + inFlightRecovery = recovery; + const release = () => { + if (inFlightRecovery === recovery) { + inFlightRecovery = null; + } + }; + recovery.then(release, release); + return recovery; +} + +/** + * The single entry point for both callers. Classifies, then executes: + * `reopen` → `reopenNow()`; `round-trip-check` → `probe(2000)`, reopening on a + * dead round trip. + * + * One entry, two usage postures — the semantics live in the call site, not in + * two named functions: + * + * // Foreground ladder (app/sagas/state.js) — fire-and-forget: + * recoverSocket().catch(log); + * + * // Accept gate (acceptNativeCall.ts) — awaited, abortable: + * const outcome = await recoverSocket({ abortSignal: controller.signal }); + * if (outcome === 'no-socket') return handleFailure(callId, mediaSession); + * if (outcome === 'abandoned') return; + * + * Concurrency: overlapping calls share one in-flight recovery — the second + * caller awaits the same work and receives its outcome. An abort signal + * detaches the caller from the shared wait (`'abandoned'`); it never cancels + * the recovery itself, since another caller may depend on it. + */ +export function recoverSocket(options?: { abortSignal?: AbortSignal }): Promise { + const { abortSignal } = options ?? {}; + if (abortSignal?.aborted) { + return Promise.resolve('abandoned'); + } + + const recovery = shareRecovery(); + if (!abortSignal) { + return recovery; + } + const abandoned = new Promise(resolve => { + onAbort(abortSignal, () => resolve('abandoned')); + }); + return Promise.race([recovery, abandoned]); +} + +/** + * Vocabulary: + * - socket health — the classification concern (`classifySocketHealth`). + * - recovery plan — `SocketRecoveryPlan`, the decision. + * - round trip — the liveness check (`ddp.probe` stays as the SDK + * method name; our terms say round trip). + * - recovery outcome — `SocketRecoveryOutcome`, what callers see. + */ diff --git a/app/lib/services/voip/MediaCallEvents.ios.test.ts b/app/lib/services/voip/MediaCallEvents.ios.test.ts index deff5504327..8f5a429d7a0 100644 --- a/app/lib/services/voip/MediaCallEvents.ios.test.ts +++ b/app/lib/services/voip/MediaCallEvents.ios.test.ts @@ -80,7 +80,8 @@ jest.mock('../../native/NativeVoip', () => ({ jest.mock('./MediaSessionInstance', () => ({ mediaSessionInstance: { endCall: jest.fn(), - applyRestStateSignals: jest.fn(() => Promise.resolve()) + applyRestStateSignals: jest.fn(() => Promise.resolve()), + acceptNativeCallWithReadiness: jest.fn(() => Promise.resolve()) } })); @@ -284,7 +285,7 @@ describe('getInitialMediaCallEvents — iOS cold start', () => { expect(result).toBe(true); expect(mockSetNativeAcceptedCallId).toHaveBeenCalledWith(callId); - expect(mediaSessionInstance.applyRestStateSignals).toHaveBeenCalled(); + expect(mediaSessionInstance.acceptNativeCallWithReadiness).toHaveBeenCalledWith(callId); expect(mockOnOpenDeepLink).not.toHaveBeenCalled(); }); diff --git a/app/lib/services/voip/MediaCallEvents.test.ts b/app/lib/services/voip/MediaCallEvents.test.ts index 3113774c0d5..856dbb18cd8 100644 --- a/app/lib/services/voip/MediaCallEvents.test.ts +++ b/app/lib/services/voip/MediaCallEvents.test.ts @@ -61,7 +61,8 @@ jest.mock('react-native-callkeep', () => ({ jest.mock('./MediaSessionInstance', () => ({ mediaSessionInstance: { endCall: jest.fn(), - applyRestStateSignals: jest.fn(() => Promise.resolve()) + applyRestStateSignals: jest.fn(() => Promise.resolve()), + acceptNativeCallWithReadiness: jest.fn(() => Promise.resolve()) } })); @@ -144,7 +145,7 @@ describe('MediaCallEvents cross-server accept (slice 3)', () => { }); }); - it('skips deep link open and replays REST state signals when host matches active workspace', () => { + it('skips deep link open and runs the accept readiness gate when host matches active workspace', () => { const { mediaSessionInstance } = jest.requireMock('./MediaSessionInstance'); mockServerSelector.mockReturnValueOnce('https://workspace-a.example.com'); const payload = buildIncomingPayload({ @@ -155,7 +156,8 @@ describe('MediaCallEvents cross-server accept (slice 3)', () => { DeviceEventEmitter.emit('VoipAcceptSucceeded', payload); expect(mockSetNativeAcceptedCallId).toHaveBeenCalledWith('same-ws-call'); - expect(mediaSessionInstance.applyRestStateSignals).toHaveBeenCalledTimes(1); + expect(mediaSessionInstance.acceptNativeCallWithReadiness).toHaveBeenCalledTimes(1); + expect(mediaSessionInstance.acceptNativeCallWithReadiness).toHaveBeenCalledWith('same-ws-call'); expect(mockOnOpenDeepLink).not.toHaveBeenCalled(); }); diff --git a/app/lib/services/voip/MediaCallEvents.ts b/app/lib/services/voip/MediaCallEvents.ts index a20dddf0c34..a915ce8f608 100644 --- a/app/lib/services/voip/MediaCallEvents.ts +++ b/app/lib/services/voip/MediaCallEvents.ts @@ -94,8 +94,8 @@ function handleVoipAcceptSucceededFromNative(data: VoipPayload, adapters: MediaC NativeVoipModule.clearInitialEvents(); useCallStore.getState().setNativeAcceptedCallId(data.callId); if (data.host && isVoipIncomingHostCurrentWorkspace(data.host, adapters.getActiveServerUrl)) { - mediaSessionInstance.applyRestStateSignals().catch(error => { - mediaCallLogger.error(`${TAG} applyRestStateSignals failed:`, error); + mediaSessionInstance.acceptNativeCallWithReadiness(data.callId!).catch(error => { + mediaCallLogger.error(`${TAG} acceptNativeCallWithReadiness failed:`, error); }); return; } @@ -279,8 +279,8 @@ export const getInitialMediaCallEvents = async (adapters: MediaCallEventsAdapter mediaCallLogger.log(`${TAG} Same workspace as VoIP host; continuing appInit for cold-start handoff`); return false; } - mediaSessionInstance.applyRestStateSignals().catch(error => { - mediaCallLogger.error(`${TAG} applyRestStateSignals (initial) failed:`, error); + mediaSessionInstance.acceptNativeCallWithReadiness(initialEvents.callId).catch(error => { + mediaCallLogger.error(`${TAG} acceptNativeCallWithReadiness (initial) failed:`, error); }); mediaCallLogger.log(`${TAG} Same workspace as VoIP host; skipped deepLinkingOpen`); return true; diff --git a/app/lib/services/voip/MediaSessionInstance.test.ts b/app/lib/services/voip/MediaSessionInstance.test.ts index 8c10f6e2f92..d4bba0d1c60 100644 --- a/app/lib/services/voip/MediaSessionInstance.test.ts +++ b/app/lib/services/voip/MediaSessionInstance.test.ts @@ -15,6 +15,10 @@ jest.mock('../../methods/helpers/log', () => ({ default: (...args: unknown[]) => mockLog(...args) })); +jest.mock('../waitForLoginReady', () => ({ + waitForLoginReady: jest.fn(() => Promise.resolve(true)) +})); + const mockTerminateNativeCall = jest.fn(); jest.mock('./terminateNativeCall', () => ({ terminateNativeCall: (...args: unknown[]) => mockTerminateNativeCall(...args) @@ -60,7 +64,21 @@ jest.mock('../sdk', () => ({ __esModule: true, default: { onStreamData: (...args: Parameters) => mockOnStreamData(...args), - methodCall: (...args: unknown[]) => mockMethodCall(...args) + methodCall: (...args: unknown[]) => { + mockMethodCall(...args); + return Promise.resolve(); + }, + get current() { + return { + ddp: { + reopenNow: jest.fn(() => Promise.resolve()), + probe: jest.fn(() => Promise.resolve(true)), + lastPing: Date.now(), + pingInterval: 10000, + waitForNotifyUserMediaSubs: jest.fn(() => Promise.resolve(true)) + } + }; + } } })); @@ -239,6 +257,8 @@ function buildClientMediaCall(options: { } describe('MediaSessionInstance', () => { + let acceptNativeCallWithReadinessSpy: jest.SpyInstance, [string]>; + beforeEach(() => { jest.clearAllMocks(); mockStartVoipCallService.mockResolvedValue(undefined); @@ -260,9 +280,13 @@ describe('MediaSessionInstance', () => { roomId: null }); mediaSessionInstance.reset(); + acceptNativeCallWithReadinessSpy = jest + .spyOn(mediaSessionInstance, 'acceptNativeCallWithReadiness') + .mockResolvedValue(undefined); }); afterEach(() => { + acceptNativeCallWithReadinessSpy?.mockRestore(); mediaSessionInstance.reset(); }); @@ -481,9 +505,9 @@ describe('MediaSessionInstance', () => { }); describe('stream-notify-user (notification/accepted gated)', () => { - it('does not call answerCall when nativeAcceptedCallId is null', async () => { - const answerSpy = jest.spyOn(mediaSessionInstance, 'answerCall').mockResolvedValue(undefined); + it('does not invoke the accept readiness gate when nativeAcceptedCallId is null', async () => { await mediaSessionInstance.init('user-1'); + acceptNativeCallWithReadinessSpy.mockClear(); const streamHandler = getStreamNotifyHandler(); streamHandler({ msg: 'changed', @@ -500,12 +524,10 @@ describe('MediaSessionInstance', () => { } }); await Promise.resolve(); - expect(answerSpy).not.toHaveBeenCalled(); - answerSpy.mockRestore(); + expect(acceptNativeCallWithReadinessSpy).not.toHaveBeenCalled(); }); - it('calls answerCall when nativeAcceptedCallId matches signal and contract matches device', async () => { - const answerSpy = jest.spyOn(mediaSessionInstance, 'answerCall').mockResolvedValue(undefined); + it('invokes the accept readiness gate when nativeAcceptedCallId matches signal and contract matches device', async () => { mockUseCallStoreGetState.mockReturnValue({ reset: mockCallStoreReset, setCall: jest.fn(), @@ -518,6 +540,7 @@ describe('MediaSessionInstance', () => { roomId: null }); await mediaSessionInstance.init('user-1'); + acceptNativeCallWithReadinessSpy.mockClear(); const streamHandler = getStreamNotifyHandler(); streamHandler({ msg: 'changed', @@ -534,12 +557,10 @@ describe('MediaSessionInstance', () => { } }); await Promise.resolve(); - expect(answerSpy).toHaveBeenCalledWith('from-signal'); - answerSpy.mockRestore(); + expect(acceptNativeCallWithReadinessSpy).toHaveBeenCalledWith('from-signal'); }); - it('calls answerCall when only nativeAcceptedCallId matches (transient callId null)', async () => { - const answerSpy = jest.spyOn(mediaSessionInstance, 'answerCall').mockResolvedValue(undefined); + it('invokes the accept readiness gate when only nativeAcceptedCallId matches (transient callId null)', async () => { mockUseCallStoreGetState.mockReturnValue({ reset: mockCallStoreReset, setCall: jest.fn(), @@ -552,6 +573,7 @@ describe('MediaSessionInstance', () => { roomId: null }); await mediaSessionInstance.init('user-1'); + acceptNativeCallWithReadinessSpy.mockClear(); const streamHandler = getStreamNotifyHandler(); streamHandler({ msg: 'changed', @@ -568,12 +590,10 @@ describe('MediaSessionInstance', () => { } }); await Promise.resolve(); - expect(answerSpy).toHaveBeenCalledWith('sticky-only'); - answerSpy.mockRestore(); + expect(acceptNativeCallWithReadinessSpy).toHaveBeenCalledWith('sticky-only'); }); - it('does not call answerCall when store call object is already set', async () => { - const answerSpy = jest.spyOn(mediaSessionInstance, 'answerCall').mockResolvedValue(undefined); + it('does not invoke the accept readiness gate when store call object is already set', async () => { mockUseCallStoreGetState.mockReturnValue({ reset: mockCallStoreReset, setCall: jest.fn(), @@ -586,6 +606,7 @@ describe('MediaSessionInstance', () => { roomId: null }); await mediaSessionInstance.init('user-1'); + acceptNativeCallWithReadinessSpy.mockClear(); const streamHandler = getStreamNotifyHandler(); streamHandler({ msg: 'changed', @@ -602,25 +623,12 @@ describe('MediaSessionInstance', () => { } }); await Promise.resolve(); - expect(answerSpy).not.toHaveBeenCalled(); - answerSpy.mockRestore(); + expect(acceptNativeCallWithReadinessSpy).not.toHaveBeenCalled(); }); }); describe('REST state signals replay (native accept race)', () => { - it('calls answerCall from init when REST returns accepted and nativeAcceptedCallId already matches', async () => { - const answerSpy = jest.spyOn(mediaSessionInstance, 'answerCall').mockResolvedValue(undefined); - mockMediaCallsStateSignals.mockResolvedValue({ - success: true, - signals: [ - { - type: 'notification', - notification: 'accepted', - signedContractId: 'test-device-id', - callId: 'race-call' - } - ] - }); + it('invokes the accept readiness gate from init when nativeAcceptedCallId is already set', async () => { mockUseCallStoreGetState.mockReturnValue({ reset: mockCallStoreReset, setCall: jest.fn(), @@ -632,10 +640,12 @@ describe('MediaSessionInstance', () => { nativeAcceptedCallId: 'race-call', roomId: null }); + await mediaSessionInstance.init('user-1'); await Promise.resolve(); - expect(answerSpy).toHaveBeenCalledWith('race-call'); - answerSpy.mockRestore(); + + expect(acceptNativeCallWithReadinessSpy).toHaveBeenCalledTimes(1); + expect(acceptNativeCallWithReadinessSpy).toHaveBeenCalledWith('race-call'); }); it('applyRestStateSignals skips REST when no instance', async () => { @@ -645,6 +655,40 @@ describe('MediaSessionInstance', () => { expect(mockMediaCallsStateSignals).not.toHaveBeenCalled(); }); + it('applyRestStateSignals calls answerCall directly when a matching accepted signal is replayed', async () => { + const answerSpy = jest.spyOn(mediaSessionInstance, 'answerCall').mockResolvedValue(undefined); + mockUseCallStoreGetState.mockReturnValue({ + reset: mockCallStoreReset, + setCall: jest.fn(), + setRoomId: mockSetRoomId, + setDirection: mockSetDirection, + resetNativeCallId: jest.fn(), + call: null, + callId: null, + nativeAcceptedCallId: 'rest-accepted', + roomId: null + }); + mockMediaCallsStateSignals.mockResolvedValue({ + signals: [ + { + type: 'notification', + notification: 'accepted', + signedContractId: 'test-device-id', + callId: 'rest-accepted' + } + ], + success: true + }); + try { + await mediaSessionInstance.init('user-1'); + await mediaSessionInstance.applyRestStateSignals(); + await Promise.resolve(); + expect(answerSpy).toHaveBeenCalledWith('rest-accepted'); + } finally { + answerSpy.mockRestore(); + } + }); + it('applyRestStateSignals refetches REST after init', async () => { await mediaSessionInstance.init('user-1'); mockMediaCallsStateSignals.mockClear(); diff --git a/app/lib/services/voip/MediaSessionInstance.ts b/app/lib/services/voip/MediaSessionInstance.ts index e9075cfa0d9..4c7df62da10 100644 --- a/app/lib/services/voip/MediaSessionInstance.ts +++ b/app/lib/services/voip/MediaSessionInstance.ts @@ -36,6 +36,7 @@ import { isInActiveVoipCall } from './isInActiveVoipCall'; import { requestVoipCallPermissions } from '../../methods/voipCallPermissions'; import I18n from '../../../i18n'; import { showErrorAlert } from '../../methods/helpers/info'; +import { acceptNativeCallWithReadiness as runAcceptNativeCallGate } from './acceptNativeCall'; const mediaCallLogger = new MediaCallLogger(); @@ -48,7 +49,7 @@ class MediaSessionInstance { private storeTimeoutUnsubscribe: (() => void) | null = null; private storeIceServersUnsubscribe: (() => void) | null = null; - private tryAnswerIfNativeAcceptedNotification(signal: ServerMediaSignal): void { + private tryAnswerIfNativeAcceptedNotification(signal: ServerMediaSignal, useGate = false): void { const { call, nativeAcceptedCallId } = useCallStore.getState(); if ( signal.type === 'notification' && @@ -57,9 +58,15 @@ class MediaSessionInstance { nativeAcceptedCallId === signal.callId && call == null ) { - this.answerCall(signal.callId).catch(error => { - log(error); - }); + if (useGate) { + this.acceptNativeCallWithReadiness(signal.callId).catch(error => { + log(error); + }); + } else { + this.answerCall(signal.callId).catch(error => { + log(error); + }); + } } } @@ -81,6 +88,14 @@ class MediaSessionInstance { } } + public isInitialized(): boolean { + return this.instance != null; + } + + public acceptNativeCallWithReadiness = (callId: string): Promise => { + return runAcceptNativeCallGate(callId, this); + }; + public async init(userId: string): Promise { this.reset(); @@ -96,7 +111,9 @@ class MediaSessionInstance { }) ); mediaSessionStore.setSendSignalFn((signal: ClientMediaSignal) => { - sdk.methodCall('stream-notify-user', `${userId}/media-calls`, JSON.stringify(signal)); + sdk.methodCall('stream-notify-user', `${userId}/media-calls`, JSON.stringify(signal)).catch(error => { + log(error); + }); }); this.instance = mediaSessionStore.getInstance(userId); @@ -104,7 +121,14 @@ class MediaSessionInstance { throw new Error('Failed to create media session instance'); } - await this.applyRestStateSignals(); + const { nativeAcceptedCallId } = useCallStore.getState(); + if (nativeAcceptedCallId) { + this.acceptNativeCallWithReadiness(nativeAcceptedCallId).catch(error => { + log(error); + }); + } else { + await this.applyRestStateSignals(); + } this.mediaSessionStoreChangeUnsubscribe = mediaSessionStore.onChange(() => { this.instance = mediaSessionStore.getInstance(userId); @@ -125,7 +149,7 @@ class MediaSessionInstance { } catch (error) { log(error); } - this.tryAnswerIfNativeAcceptedNotification(signal as ServerMediaSignal); + this.tryAnswerIfNativeAcceptedNotification(signal as ServerMediaSignal, true); }); this.instance?.on('newCall', ({ call }: { call: IClientMediaCall }) => { diff --git a/app/lib/services/voip/acceptNativeCall.integration.test.ts b/app/lib/services/voip/acceptNativeCall.integration.test.ts new file mode 100644 index 00000000000..ef988bc3825 --- /dev/null +++ b/app/lib/services/voip/acceptNativeCall.integration.test.ts @@ -0,0 +1,172 @@ +import type { Store } from 'redux'; + +import { acceptNativeCallWithReadiness } from './acceptNativeCall'; +import { terminateNativeCall } from './terminateNativeCall'; +import { useCallStore } from './useCallStore'; +import { initStore } from '../../store/auxStore'; +import { recoverSocket } from '../socketHealth'; +import sdk from '../sdk'; +import type { IApplicationState } from '../../../definitions'; + +jest.mock('./terminateNativeCall', () => ({ + terminateNativeCall: jest.fn() +})); + +jest.mock('./useCallStore', () => ({ + useCallStore: { + getState: jest.fn() + } +})); + +jest.mock('../socketHealth', () => ({ + recoverSocket: jest.fn() +})); + +jest.mock('../sdk', () => ({ + __esModule: true, + default: { current: undefined } +})); + +jest.mock('../../methods/helpers/log', () => ({ + __esModule: true, + default: jest.fn() +})); + +const CALL_ID = 'call-uuid'; +const READINESS_TIMEOUT = 8000; + +const mockTerminateNativeCall = terminateNativeCall as jest.Mock; +const mockGetCallState = useCallStore.getState as jest.Mock; +const mockRecoverSocket = recoverSocket as jest.MockedFunction; + +interface IMediaSession { + applyRestStateSignals: jest.Mock, []>; + answerCall: jest.Mock, [string]>; + endCall: jest.Mock; + isInitialized: jest.Mock; +} + +function makeMediaSession(): IMediaSession { + return { + applyRestStateSignals: jest.fn, []>(() => Promise.resolve()), + answerCall: jest.fn, [string]>(() => Promise.resolve()), + endCall: jest.fn(), + isInitialized: jest.fn(() => true) + }; +} + +/** Media Signal subs that ack `delayMs` after the gate starts waiting. */ +function mediaSubsAckAfter(delayMs: number) { + return { + waitForNotifyUserMediaSubs: jest.fn(() => new Promise(resolve => setTimeout(() => resolve(true), delayMs))) + }; +} + +/** Media Signal subs that never ack: the wait ends on its own timeout. */ +function mediaSubsNeverAck() { + return { + waitForNotifyUserMediaSubs: jest.fn( + (timeoutMs: number) => new Promise(resolve => setTimeout(() => resolve(false), timeoutMs)) + ) + }; +} + +/** + * Minimal redux surface so `waitForLoginReady` runs for real: it reads + * `login.isAuthenticated` / `meteor.connected` and subscribes for changes. + */ +function makeReduxStore() { + const listeners = new Set<() => void>(); + const state = { login: { isAuthenticated: false }, meteor: { connected: false } }; + return { + listenerCount: () => listeners.size, + setLoginReady: () => { + state.login.isAuthenticated = true; + state.meteor.connected = true; + listeners.forEach(listener => listener()); + }, + store: { + getState: () => state, + subscribe: (listener: () => void) => { + listeners.add(listener); + return () => listeners.delete(listener); + } + } as unknown as Store + }; +} + +describe('acceptNativeCallWithReadiness against real login readiness', () => { + let redux: ReturnType; + + beforeEach(() => { + jest.clearAllMocks(); + jest.useFakeTimers(); + redux = makeReduxStore(); + initStore(redux.store); + mockGetCallState.mockReturnValue({ call: null, resetNativeCallId: jest.fn() }); + mockRecoverSocket.mockResolvedValue('reopened'); + (sdk as any).current = { ddp: mediaSubsAckAfter(100) }; + }); + + afterEach(() => { + jest.useRealTimers(); + }); + + it('recovers the socket, waits for readiness, then answers the call', async () => { + const mediaSession = makeMediaSession(); + + const gate = acceptNativeCallWithReadiness(CALL_ID, mediaSession); + + // Readiness only lands after the gate is already waiting on it. + await jest.advanceTimersByTimeAsync(0); + expect(mediaSession.answerCall).not.toHaveBeenCalled(); + redux.setLoginReady(); + + await jest.advanceTimersByTimeAsync(200); + await gate; + + expect(mockRecoverSocket).toHaveBeenCalledTimes(1); + expect(mediaSession.applyRestStateSignals).toHaveBeenCalledTimes(1); + expect(mediaSession.answerCall).toHaveBeenCalledWith(CALL_ID); + expect(mockTerminateNativeCall).not.toHaveBeenCalled(); + }); + + it('releases its store listener and readiness polling as soon as readiness lands', async () => { + redux.setLoginReady(); + const mediaSession = makeMediaSession(); + + const gate = acceptNativeCallWithReadiness(CALL_ID, mediaSession); + await jest.advanceTimersByTimeAsync(200); + await gate; + + expect(redux.listenerCount()).toBe(0); + + // Nothing is left scheduled: no late failure ladder. + await jest.advanceTimersByTimeAsync(60000); + expect(mockTerminateNativeCall).not.toHaveBeenCalled(); + expect(mediaSession.endCall).not.toHaveBeenCalled(); + }); + + it('runs the failure ladder once and leaves nothing behind when readiness never lands', async () => { + (sdk as any).current = { ddp: mediaSubsNeverAck() }; + const resetNativeCallId = jest.fn(); + mockGetCallState.mockReturnValue({ call: null, resetNativeCallId }); + const mediaSession = makeMediaSession(); + + const gate = acceptNativeCallWithReadiness(CALL_ID, mediaSession); + + // Login never authenticates and the media subs never ack. + await jest.advanceTimersByTimeAsync(READINESS_TIMEOUT); + await gate; + + expect(mockTerminateNativeCall).toHaveBeenCalledWith(CALL_ID); + expect(resetNativeCallId).toHaveBeenCalledTimes(1); + expect(mediaSession.endCall).toHaveBeenCalledWith(CALL_ID); + expect(mediaSession.answerCall).not.toHaveBeenCalled(); + expect(redux.listenerCount()).toBe(0); + + await jest.advanceTimersByTimeAsync(60000); + expect(mockTerminateNativeCall).toHaveBeenCalledTimes(1); + expect(mediaSession.endCall).toHaveBeenCalledTimes(1); + }); +}); diff --git a/app/lib/services/voip/acceptNativeCall.test.ts b/app/lib/services/voip/acceptNativeCall.test.ts new file mode 100644 index 00000000000..026f4e9dc0b --- /dev/null +++ b/app/lib/services/voip/acceptNativeCall.test.ts @@ -0,0 +1,281 @@ +import { acceptNativeCallWithReadiness } from './acceptNativeCall'; +import { useCallStore } from './useCallStore'; +import { terminateNativeCall } from './terminateNativeCall'; +import { waitForLoginReady } from '../waitForLoginReady'; +import { recoverSocket } from '../socketHealth'; +import sdk from '../sdk'; + +const mockWaitForLoginReady = waitForLoginReady as jest.MockedFunction; +const mockRecoverSocket = recoverSocket as jest.MockedFunction; +const mockGetState = useCallStore.getState as jest.Mock; +const mockTerminateNativeCall = terminateNativeCall as jest.Mock; +const mockDdp = () => sdk.current?.ddp as any; + +jest.mock('./useCallStore', () => ({ + useCallStore: { + getState: jest.fn() + } +})); + +jest.mock('./terminateNativeCall', () => ({ + terminateNativeCall: jest.fn() +})); + +jest.mock('../sdk', () => ({ + __esModule: true, + default: { + current: { ddp: {} } + } +})); + +jest.mock('../socketHealth', () => ({ + recoverSocket: jest.fn() +})); + +jest.mock('../waitForLoginReady', () => ({ + ...jest.requireActual('../waitForLoginReady'), + waitForLoginReady: jest.fn() +})); + +jest.mock('../../methods/helpers/log', () => ({ + __esModule: true, + default: jest.fn() +})); + +interface IMediaSession { + applyRestStateSignals: jest.Mock>; + answerCall: jest.Mock, [string]>; + endCall: jest.Mock; + isInitialized: jest.Mock; +} + +function makeMediaSession(overrides: Partial = {}): IMediaSession { + return { + applyRestStateSignals: jest.fn, []>(() => Promise.resolve()), + answerCall: jest.fn, [string]>(() => Promise.resolve()), + endCall: jest.fn(), + isInitialized: jest.fn(() => true), + ...overrides + }; +} + +function makeDdp(overrides: Record = {}) { + return { + waitForNotifyUserMediaSubs: jest.fn(() => Promise.resolve(true)), + ...overrides + }; +} + +function makeStoreState(overrides: Record = {}) { + return { + call: null, + resetNativeCallId: jest.fn(), + ...overrides + }; +} + +describe('acceptNativeCallWithReadiness', () => { + const CALL_ID = 'call-uuid'; + + beforeEach(() => { + jest.clearAllMocks(); + jest.useFakeTimers(); + (sdk as any).current = { ddp: makeDdp() }; + mockRecoverSocket.mockResolvedValue('confirmed-alive'); + mockWaitForLoginReady.mockResolvedValue(true); + mockGetState.mockReturnValue(makeStoreState()); + }); + + afterEach(() => { + jest.useRealTimers(); + }); + + it.each(['confirmed-alive', 'reopened'] as const)( + 'waits for readiness and answers the call when socket recovery reports %s', + async outcome => { + mockRecoverSocket.mockResolvedValue(outcome); + const mediaSession = makeMediaSession(); + + await acceptNativeCallWithReadiness(CALL_ID, mediaSession); + + expect(mockRecoverSocket).toHaveBeenCalledTimes(1); + expect(mockRecoverSocket.mock.calls[0][0]?.abortSignal).toBeDefined(); + expect(mockWaitForLoginReady.mock.invocationCallOrder[0]).toBeGreaterThan(mockRecoverSocket.mock.invocationCallOrder[0]); + expect(mediaSession.applyRestStateSignals).toHaveBeenCalledTimes(1); + expect(mediaSession.answerCall).toHaveBeenCalledWith(CALL_ID); + } + ); + + it('terminates and ends the call when there is no socket to recover', async () => { + mockRecoverSocket.mockResolvedValue('no-socket'); + const mediaSession = makeMediaSession(); + const resetNativeCallId = jest.fn(); + mockGetState.mockReturnValue(makeStoreState({ resetNativeCallId })); + + await acceptNativeCallWithReadiness(CALL_ID, mediaSession); + + expect(mockTerminateNativeCall).toHaveBeenCalledWith(CALL_ID); + expect(resetNativeCallId).toHaveBeenCalled(); + expect(mediaSession.endCall).toHaveBeenCalledWith(CALL_ID); + expect(mockWaitForLoginReady).not.toHaveBeenCalled(); + }); + + it('returns silently without terminating when socket recovery reports the gate abandoned', async () => { + mockRecoverSocket.mockResolvedValue('abandoned'); + const mediaSession = makeMediaSession(); + const resetNativeCallId = jest.fn(); + mockGetState.mockReturnValue(makeStoreState({ resetNativeCallId })); + + await acceptNativeCallWithReadiness(CALL_ID, mediaSession); + + expect(mockTerminateNativeCall).not.toHaveBeenCalled(); + expect(resetNativeCallId).not.toHaveBeenCalled(); + expect(mediaSession.endCall).not.toHaveBeenCalled(); + expect(mediaSession.applyRestStateSignals).not.toHaveBeenCalled(); + expect(mockWaitForLoginReady).not.toHaveBeenCalled(); + }); + + it('terminates and ends the call when socket recovery throws', async () => { + mockRecoverSocket.mockRejectedValue(new Error('reopen failed')); + const mediaSession = makeMediaSession(); + const resetNativeCallId = jest.fn(); + mockGetState.mockReturnValue(makeStoreState({ resetNativeCallId })); + + await acceptNativeCallWithReadiness(CALL_ID, mediaSession); + + expect(mockTerminateNativeCall).toHaveBeenCalledWith(CALL_ID); + expect(resetNativeCallId).toHaveBeenCalled(); + expect(mediaSession.endCall).toHaveBeenCalledWith(CALL_ID); + }); + + it('terminates and ends the call when login readiness times out', async () => { + mockWaitForLoginReady.mockResolvedValue(false); + const mediaSession = makeMediaSession(); + const resetNativeCallId = jest.fn(); + mockGetState.mockReturnValue(makeStoreState({ resetNativeCallId })); + + await acceptNativeCallWithReadiness(CALL_ID, mediaSession); + + expect(mockTerminateNativeCall).toHaveBeenCalledWith(CALL_ID); + expect(resetNativeCallId).toHaveBeenCalled(); + expect(mediaSession.endCall).toHaveBeenCalledWith(CALL_ID); + expect(mediaSession.applyRestStateSignals).not.toHaveBeenCalled(); + }); + + it('terminates and ends the call when media-subscription ack times out', async () => { + mockDdp().waitForNotifyUserMediaSubs = jest.fn(() => Promise.resolve(false)); + const mediaSession = makeMediaSession(); + const resetNativeCallId = jest.fn(); + mockGetState.mockReturnValue(makeStoreState({ resetNativeCallId })); + + await acceptNativeCallWithReadiness(CALL_ID, mediaSession); + + expect(mockTerminateNativeCall).toHaveBeenCalledWith(CALL_ID); + expect(resetNativeCallId).toHaveBeenCalled(); + expect(mediaSession.endCall).toHaveBeenCalledWith(CALL_ID); + expect(mediaSession.applyRestStateSignals).not.toHaveBeenCalled(); + }); + + it('terminates and ends the call when the media session is not initialized', async () => { + const mediaSession = makeMediaSession({ isInitialized: jest.fn(() => false) }); + const resetNativeCallId = jest.fn(); + mockGetState.mockReturnValue(makeStoreState({ resetNativeCallId })); + + await acceptNativeCallWithReadiness(CALL_ID, mediaSession); + + expect(mockTerminateNativeCall).toHaveBeenCalledWith(CALL_ID); + expect(resetNativeCallId).toHaveBeenCalled(); + expect(mediaSession.endCall).toHaveBeenCalledWith(CALL_ID); + expect(mediaSession.applyRestStateSignals).not.toHaveBeenCalled(); + }); + + it('terminates and ends the call when the SDK socket is unavailable for media subscriptions', async () => { + (sdk as any).current = {}; + const mediaSession = makeMediaSession(); + const resetNativeCallId = jest.fn(); + mockGetState.mockReturnValue(makeStoreState({ resetNativeCallId })); + + await acceptNativeCallWithReadiness(CALL_ID, mediaSession); + + expect(mockTerminateNativeCall).toHaveBeenCalledWith(CALL_ID); + expect(resetNativeCallId).toHaveBeenCalled(); + expect(mediaSession.endCall).toHaveBeenCalledWith(CALL_ID); + }); + + it('does not call answerCall when applyRestStateSignals already answered', async () => { + const mediaSession = makeMediaSession(); + mockGetState.mockReturnValue(makeStoreState({ call: { callId: CALL_ID } })); + + await acceptNativeCallWithReadiness(CALL_ID, mediaSession); + + expect(mediaSession.applyRestStateSignals).toHaveBeenCalledTimes(1); + expect(mediaSession.answerCall).not.toHaveBeenCalled(); + }); + + it('aborts the previous gate for the same callId and lets the new gate succeed', async () => { + mockWaitForLoginReady.mockImplementation((_timeoutMs, signal) => Promise.resolve(!signal?.aborted)); + + const firstSession = makeMediaSession(); + const secondSession = makeMediaSession(); + + const first = acceptNativeCallWithReadiness(CALL_ID, firstSession); + const second = acceptNativeCallWithReadiness(CALL_ID, secondSession); + + await Promise.all([first, second]); + + expect(firstSession.applyRestStateSignals).not.toHaveBeenCalled(); + expect(firstSession.endCall).not.toHaveBeenCalled(); + expect(firstSession.answerCall).not.toHaveBeenCalled(); + expect(secondSession.applyRestStateSignals).toHaveBeenCalledTimes(1); + expect(secondSession.answerCall).toHaveBeenCalledWith(CALL_ID); + }); + + it('does not terminate or end the call when an aborted gate finishes', async () => { + mockWaitForLoginReady.mockImplementation((_timeoutMs, signal) => Promise.resolve(!signal?.aborted)); + + const firstSession = makeMediaSession(); + const secondSession = makeMediaSession(); + + const first = acceptNativeCallWithReadiness(CALL_ID, firstSession); + const second = acceptNativeCallWithReadiness(CALL_ID, secondSession); + + await Promise.all([first, second]); + + expect(mockTerminateNativeCall).not.toHaveBeenCalledWith(CALL_ID); + expect(firstSession.endCall).not.toHaveBeenCalled(); + expect(secondSession.answerCall).toHaveBeenCalledWith(CALL_ID); + }); + + it('keeps the newer gate entry when an older gate cleans up, so a third gate aborts the newer one', async () => { + let gateIndex = 0; + mockWaitForLoginReady.mockImplementation((_timeoutMs, signal) => { + const myIndex = ++gateIndex; + if (signal?.aborted) { + return Promise.resolve(false); + } + return new Promise(resolve => { + setTimeout(() => resolve(true), myIndex * 1000); + }); + }); + + const firstSession = makeMediaSession(); + const secondSession = makeMediaSession(); + const thirdSession = makeMediaSession(); + + const first = acceptNativeCallWithReadiness(CALL_ID, firstSession); + const second = acceptNativeCallWithReadiness(CALL_ID, secondSession); + + await Promise.resolve(); + await Promise.resolve(); + + const third = acceptNativeCallWithReadiness(CALL_ID, thirdSession); + + await jest.advanceTimersByTimeAsync(3000); + + await Promise.all([first, second, third]); + + expect(firstSession.endCall).not.toHaveBeenCalled(); + expect(secondSession.answerCall).not.toHaveBeenCalled(); + expect(thirdSession.answerCall).toHaveBeenCalledWith(CALL_ID); + }); +}); diff --git a/app/lib/services/voip/acceptNativeCall.ts b/app/lib/services/voip/acceptNativeCall.ts new file mode 100644 index 00000000000..7aab96d7d67 --- /dev/null +++ b/app/lib/services/voip/acceptNativeCall.ts @@ -0,0 +1,113 @@ +import log from '../../methods/helpers/log'; +import { onAbort } from '../../methods/helpers/onAbort'; +import sdk from '../sdk'; +import { waitForLoginReady } from '../waitForLoginReady'; +import { recoverSocket } from '../socketHealth'; +import { terminateNativeCall } from './terminateNativeCall'; +import { useCallStore } from './useCallStore'; + +export interface NativeCallMediaSession { + applyRestStateSignals(): Promise; + answerCall(callId: string): Promise; + endCall(callId: string): void; + isInitialized(): boolean; +} + +/** The slice of the patched DDP driver the accept path reads: Media Signal subscription readiness. */ +interface MediaSignalDdp { + waitForNotifyUserMediaSubs(timeoutMs: number): Promise; +} + +const activeGates = new Map(); + +async function waitForMediaSignalSubs(ddp: MediaSignalDdp, timeoutMs: number, abortSignal?: AbortSignal): Promise { + if (typeof ddp.waitForNotifyUserMediaSubs !== 'function') { + return false; + } + if (abortSignal?.aborted) { + return false; + } + + const ready = ddp.waitForNotifyUserMediaSubs(timeoutMs); + const aborted = new Promise(resolve => { + onAbort(abortSignal, () => resolve(false)); + }); + + try { + return await Promise.race([ready, aborted]); + } catch (error) { + log(error); + return false; + } +} + +function handleFailure(callId: string, mediaSession: NativeCallMediaSession): void { + terminateNativeCall(callId); + useCallStore.getState().resetNativeCallId(); + mediaSession.endCall(callId); +} + +export async function acceptNativeCallWithReadiness(callId: string, mediaSession: NativeCallMediaSession): Promise { + const previous = activeGates.get(callId); + if (previous) { + previous.abort(); + } + + const controller = new AbortController(); + activeGates.set(callId, controller); + const cleanup = () => { + if (activeGates.get(callId) === controller) { + activeGates.delete(callId); + } + }; + + try { + const outcome = await recoverSocket({ abortSignal: controller.signal }); + if (outcome === 'no-socket') { + return handleFailure(callId, mediaSession); + } + if (outcome === 'abandoned') { + return; + } + + if (controller.signal.aborted) { + return; + } + + const ddp = sdk.current?.ddp as MediaSignalDdp | undefined; + if (!ddp) { + return handleFailure(callId, mediaSession); + } + + const [loginReady, mediaSubsReady] = await Promise.all([ + waitForLoginReady(8000, controller.signal), + waitForMediaSignalSubs(ddp, 8000, controller.signal) + ]); + + if (controller.signal.aborted) { + return; + } + + if (!loginReady || !mediaSubsReady || !mediaSession.isInitialized()) { + return handleFailure(callId, mediaSession); + } + + await mediaSession.applyRestStateSignals(); + + if (controller.signal.aborted) { + return; + } + + const { call } = useCallStore.getState(); + if (call?.callId !== callId) { + await mediaSession.answerCall(callId); + } + } catch (error) { + log(error); + if (!controller.signal.aborted) { + handleFailure(callId, mediaSession); + } + } finally { + cleanup(); + } +} diff --git a/app/lib/services/waitForLoginReady.ts b/app/lib/services/waitForLoginReady.ts new file mode 100644 index 00000000000..06e6ce494f7 --- /dev/null +++ b/app/lib/services/waitForLoginReady.ts @@ -0,0 +1,39 @@ +import { onAbort } from '../methods/helpers/onAbort'; +import { store } from '../store/auxStore'; + +// Reads redux rather than `ddp.loggedIn`: `close` clears `meteor.connected`, while `ddp.loggedIn` survives it. +// Neither survives a silent background death, so callers must bound their wait. +export function isLoginReady(): boolean { + const state = store.getState(); + return state.login.isAuthenticated && state.meteor.connected; +} + +export function waitForLoginReady(timeoutMs: number, abortSignal?: AbortSignal): Promise { + return new Promise(resolve => { + if (abortSignal?.aborted) { + return resolve(false); + } + if (isLoginReady()) { + return resolve(true); + } + + let settled = false; + const finish = (value: boolean) => { + if (settled) { + return; + } + settled = true; + clearTimeout(timer); + unsub(); + resolve(value); + }; + + const unsub = store.subscribe(() => { + if (isLoginReady()) { + finish(true); + } + }); + const timer = setTimeout(() => finish(false), timeoutMs); + onAbort(abortSignal, () => finish(false)); + }); +} diff --git a/app/sagas/__tests__/state.test.ts b/app/sagas/__tests__/state.test.ts new file mode 100644 index 00000000000..ea219cfb91a --- /dev/null +++ b/app/sagas/__tests__/state.test.ts @@ -0,0 +1,155 @@ +jest.mock('../../lib/methods/helpers/localAuthentication', () => ({ + localAuthenticate: jest.fn(), + saveLastLocalAuthenticationSession: jest.fn() +})); + +jest.mock('../../lib/services/restApi', () => ({ + setUserPresenceOnline: jest.fn(), + setUserPresenceAway: jest.fn() +})); + +jest.mock('../../lib/notifications', () => ({ + checkPendingNotification: jest.fn(() => Promise.resolve()) +})); + +jest.mock('../../lib/services/socketHealth', () => ({ + recoverSocket: jest.fn(() => Promise.resolve('confirmed-alive')) +})); + +jest.mock('../../lib/methods/helpers/log', () => ({ + ...jest.requireActual('../../lib/methods/helpers/log'), + __esModule: true, + default: jest.fn() +})); + +import { applyMiddleware, createStore } from 'redux'; +import createSagaMiddleware from 'redux-saga'; + +import { APP_STATE } from '../../actions/actionsTypes'; +import { appStart } from '../../actions/app'; +import { loginSuccess } from '../../actions/login'; +import { connectSuccess } from '../../actions/connect'; +import { selectServerSuccess } from '../../actions/server'; +import { RootEnum } from '../../definitions'; +import reducers from '../../reducers'; +import stateRoot from '../state'; +import { localAuthenticate } from '../../lib/methods/helpers/localAuthentication'; +import { setUserPresenceOnline, setUserPresenceAway } from '../../lib/services/restApi'; +import { recoverSocket } from '../../lib/services/socketHealth'; +import log from '../../lib/methods/helpers/log'; + +async function flushSagaMicrotasks(): Promise { + await Promise.resolve(); + await Promise.resolve(); +} + +type PreloadedState = Parameters[1]; + +function setupStore(preloadedState?: PreloadedState) { + const sagaMiddleware = createSagaMiddleware(); + const store = createStore(reducers, preloadedState, applyMiddleware(sagaMiddleware)); + sagaMiddleware.run(stateRoot); + return store; +} + +const HOST = 'https://open.rocket.chat'; + +describe('state saga — foreground socket recovery', () => { + beforeEach(() => { + jest.useFakeTimers(); + jest.clearAllMocks(); + }); + + afterEach(() => { + jest.useRealTimers(); + }); + + function setupReadyStore() { + const store = setupStore(); + store.dispatch(appStart({ root: RootEnum.ROOT_INSIDE })); + store.dispatch(selectServerSuccess({ server: HOST, name: 'open.rocket.chat', version: '6.0.0' })); + store.dispatch(loginSuccess({ id: 'user-1', token: 'token-abc' } as any)); + store.dispatch(connectSuccess()); + flushSagaMicrotasks(); + return store; + } + + it('requests recovery once when foregrounding while inside and authenticated', async () => { + const store = setupReadyStore(); + + store.dispatch({ type: APP_STATE.FOREGROUND }); + await flushSagaMicrotasks(); + + expect(recoverSocket).toHaveBeenCalledTimes(1); + expect(setUserPresenceOnline).toHaveBeenCalledTimes(1); + }); + + it('logs a recovery rejection and still sets presence online', async () => { + const failure = new Error('reopen failed'); + jest.mocked(recoverSocket).mockRejectedValueOnce(failure); + const store = setupReadyStore(); + + store.dispatch({ type: APP_STATE.FOREGROUND }); + await flushSagaMicrotasks(); + + expect(log).toHaveBeenCalledWith(failure); + expect(setUserPresenceOnline).toHaveBeenCalledTimes(1); + }); +}); + +describe('state saga — foreground early exits', () => { + beforeEach(() => { + jest.useFakeTimers(); + jest.clearAllMocks(); + }); + + afterEach(() => { + jest.useRealTimers(); + }); + + it('does nothing when not ROOT_INSIDE', async () => { + const store = setupStore(); + store.dispatch(appStart({ root: RootEnum.ROOT_OUTSIDE })); + store.dispatch({ type: APP_STATE.FOREGROUND }); + await flushSagaMicrotasks(); + + expect(localAuthenticate).not.toHaveBeenCalled(); + expect(setUserPresenceOnline).not.toHaveBeenCalled(); + }); + + it('does nothing when not authenticated', async () => { + const store = setupStore(); + store.dispatch(appStart({ root: RootEnum.ROOT_INSIDE })); + store.dispatch(selectServerSuccess({ server: HOST, name: 'open.rocket.chat', version: '6.0.0' })); + store.dispatch({ type: APP_STATE.FOREGROUND }); + await flushSagaMicrotasks(); + + expect(localAuthenticate).not.toHaveBeenCalled(); + expect(setUserPresenceOnline).not.toHaveBeenCalled(); + }); +}); + +describe('state saga — background', () => { + beforeEach(() => { + jest.useFakeTimers(); + jest.clearAllMocks(); + }); + + afterEach(() => { + jest.useRealTimers(); + }); + + it('sets presence away when authenticated and inside', async () => { + const store = setupStore(); + store.dispatch(appStart({ root: RootEnum.ROOT_INSIDE })); + store.dispatch(selectServerSuccess({ server: HOST, name: 'open.rocket.chat', version: '6.0.0' })); + store.dispatch(loginSuccess({ id: 'user-1', token: 'token-abc' } as any)); + store.dispatch(connectSuccess()); + await flushSagaMicrotasks(); + + store.dispatch({ type: APP_STATE.BACKGROUND }); + await flushSagaMicrotasks(); + + expect(setUserPresenceAway).toHaveBeenCalledTimes(1); + }); +}); diff --git a/app/sagas/login.js b/app/sagas/login.js index 10740600a33..6d5f1f8e7d9 100644 --- a/app/sagas/login.js +++ b/app/sagas/login.js @@ -42,6 +42,7 @@ import { SupportedVersionsWarning } from '../containers/SupportedVersions'; import { mediaSessionInstance } from '../lib/services/voip/MediaSessionInstance'; import { hasPermission } from '../lib/methods/helpers/helpers'; import { mediaSessionStore } from '../lib/services/voip/MediaSessionStore'; +import { isInActiveVoipCall } from '../lib/services/voip/isInActiveVoipCall'; import { store as reduxStore } from '../lib/store/auxStore'; const getServer = state => state.server.server; @@ -260,7 +261,9 @@ const checkVoipPermission = async () => { const canUseVoip = isVoipModuleAvailable() && (hasPermissions[0] || hasPermissions[1]); if (!canUseVoip) { - mediaSessionInstance.reset(); + if (!isInActiveVoipCall()) { + mediaSessionInstance.reset(); + } return; } if (!mediaSessionStore.getCurrentInstance()) { diff --git a/app/sagas/state.js b/app/sagas/state.js index 20272cfeb73..30bd3d49176 100644 --- a/app/sagas/state.js +++ b/app/sagas/state.js @@ -4,7 +4,7 @@ import log from '../lib/methods/helpers/log'; import { localAuthenticate, saveLastLocalAuthenticationSession } from '../lib/methods/helpers/localAuthentication'; import { APP_STATE } from '../actions/actionsTypes'; import { RootEnum } from '../definitions'; -import { checkAndReopen } from '../lib/services/connect'; +import { recoverSocket } from '../lib/services/socketHealth'; import { setUserPresenceOnline, setUserPresenceAway } from '../lib/services/restApi'; import { checkPendingNotification } from '../lib/notifications'; @@ -19,14 +19,18 @@ const appHasComeBackToForeground = function* appHasComeBackToForeground() { if (appRoot !== RootEnum.ROOT_INSIDE) { return; } - const isReady = yield isAuthAndConnected(); - if (!isReady) { + // Socket state is deliberately not checked here: a closed socket is the case + // recoverSocket below exists for. + const { isAuthenticated } = yield select(state => state.login); + if (!isAuthenticated) { return; } try { const server = yield select(state => state.server.server); yield localAuthenticate(server); - checkAndReopen(); + + recoverSocket().catch(e => log(e)); + // Check for pending notification when app comes to foreground (Android - notification tap while in background) checkPendingNotification().catch(e => { log('[state.js] Error checking pending notification:', e); diff --git a/jest.config.js b/jest.config.js index 2c66372994e..aa5e3a720c6 100644 --- a/jest.config.js +++ b/jest.config.js @@ -2,7 +2,7 @@ module.exports = { modulePathIgnorePatterns: ['/.*worktrees/'], testPathIgnorePatterns: ['e2e', 'node_modules', '/.*worktrees/', '/__tests__/testHelpers\\.tsx$'], transformIgnorePatterns: [ - 'node_modules/(?!((jest-)?react-native|@react-native(-community)?)|expo(nent)?|@expo(nent)?/.*|@expo-google-fonts/.*|react-navigation|@react-navigation/.*|@unimodules/.*|unimodules|sentry-expo|native-base|react-native-svg|@rocket.chat/ui-kit)' + 'node_modules/(?!((jest-)?react-native|@react-native(-community)?)|expo(nent)?|@expo(nent)?/.*|@expo-google-fonts/.*|react-navigation|@react-navigation/.*|@unimodules/.*|unimodules|sentry-expo|native-base|react-native-svg|@rocket.chat/ui-kit|@rocket.chat/sdk|tiny-events)' ], preset: './jest.preset.js', cacheDirectory: '/.jest-cache', diff --git a/patches/@rocket.chat+sdk+1.3.3-mobile.patch b/patches/@rocket.chat+sdk+1.3.3-mobile.patch index d13d1d4c749..e0e2d0b4465 100644 --- a/patches/@rocket.chat+sdk+1.3.3-mobile.patch +++ b/patches/@rocket.chat+sdk+1.3.3-mobile.patch @@ -1,16 +1,46 @@ diff --git a/node_modules/@rocket.chat/sdk/lib/drivers/ddp.ts b/node_modules/@rocket.chat/sdk/lib/drivers/ddp.ts -index 19d31ae..9f752df 100644 +index 19d31ae..068b61e 100644 --- a/node_modules/@rocket.chat/sdk/lib/drivers/ddp.ts +++ b/node_modules/@rocket.chat/sdk/lib/drivers/ddp.ts -@@ -101,9 +101,25 @@ export class Socket extends EventEmitter { +@@ -55,6 +55,7 @@ export class Socket extends EventEmitter { + connection?: WebSocket + session?: string + logger: ILogger ++ reopenPromise?: Promise + + /** Create a websocket handler */ + constructor ( +@@ -82,18 +83,13 @@ export class Socket extends EventEmitter { + } + + /** +- * Open websocket connection, with optional retry interval. +- * Stores connection, setting up handlers for open/close/message events. +- * Resumes login if given token. ++ * Create a new WebSocket, tear down any previous one, and wire up handlers. ++ * Emits 'connecting' exactly once per actual new socket. + */ +- open = (ms: number = this.config.reopen) => { +- return new Promise(async (resolve, reject) => { ++ private createConnection = (): Promise => { ++ return new Promise((resolve, reject) => { + let connection: WebSocket + +- if (this.connected) { +- return resolve() +- } +- + try { + connection = new WebSocket(this.host, null, { headers: settings.customHeaders }) + connection.onerror = reject +@@ -101,14 +97,53 @@ export class Socket extends EventEmitter { this.logger.error(err) return reject(err) } + // Tear down the previous connection before replacing it. -+ // The `this.connected` early-return above means we only reach here when the -+ // existing socket isn't healthy, so detaching its handlers and closing it stops -+ // a stale or still-connecting socket from later firing onClose and clobbering the -+ // live connection. ++ // Callers only reach here when the existing socket isn't healthy, so ++ // detaching its handlers and closing it stops a stale or still-connecting ++ // socket from later firing onClose and clobbering the live connection. + if (this.connection) { + try { + this.connection.onopen = null as any @@ -29,7 +59,36 @@ index 19d31ae..9f752df 100644 this.connection.onopen = this.onOpen.bind(this, resolve) this.emit('connecting') }) -@@ -125,7 +141,14 @@ export class Socket extends EventEmitter { + } + ++ /** ++ * Open websocket connection, with optional retry interval. ++ * Stores connection, setting up handlers for open/close/message events. ++ * Resumes login if given token. ++ */ ++ open = (ms: number = this.config.reopen) => { ++ return new Promise(async (resolve, reject) => { ++ if (this.connected) { ++ return resolve() ++ } ++ ++ if (this.reopenPromise) { ++ return this.reopenPromise.then(() => resolve(this.connection)).catch(reject) ++ } ++ ++ try { ++ await this.createConnection() ++ resolve(this.connection) ++ } catch (err) { ++ reject(err) ++ } ++ }) ++ } ++ + /** Send handshake message to confirm connection, start pinging. */ + onOpen = async (callback: Function) => { + this.lastPing = Date.now() +@@ -125,7 +160,14 @@ export class Socket extends EventEmitter { } /** Emit close event so it can be used for promise resolve in close() */ @@ -45,7 +104,134 @@ index 19d31ae..9f752df 100644 this.emit('close', e) try { if (e?.code !== userDisconnectCloseCode) { -@@ -549,7 +572,9 @@ export class DDPDriver extends EventEmitter implements ISocket, IDriver { +@@ -201,6 +243,85 @@ export class Socket extends EventEmitter { + }, this.config.reopen); + } + ++ /** ++ * Force an immediate reconnect. Shared across concurrent callers so only one ++ * new WebSocket is created. Emits 'disconnected' to unblock in-flight sends, ++ * then creates the connection directly so a concurrent open() cannot tear it ++ * down. Unhandled creation errors are swallowed because cleanup already runs ++ * via the open/timeout paths. ++ */ ++ reopenNow = (): Promise => { ++ if (this.reopenPromise) { ++ return this.reopenPromise ++ } ++ ++ this.reopenPromise = new Promise(resolve => { ++ this.openTimeout && clearTimeout(this.openTimeout as any) ++ this.lastPing = 0 ++ this.emit('disconnected') ++ ++ let settled = false ++ const cleanup = () => { ++ if (settled) return ++ settled = true ++ this.off('open', cleanup) ++ if (timeout) clearTimeout(timeout as any) ++ delete this.reopenPromise ++ resolve() ++ } ++ ++ this.once('open', cleanup) ++ ++ this.createConnection().catch(() => {}) ++ ++ const timeout = setTimeout(() => cleanup(), 10000) ++ }) ++ ++ return this.reopenPromise ++ } ++ ++ /** ++ * Bounded liveness check for a socket in the gray zone. Returns true only if ++ * the socket is open and the server answers the ping within the deadline. ++ */ ++ probe = (timeoutMs = 2000): Promise => { ++ return new Promise(resolve => { ++ if (!this.connection || this.connection.readyState !== 1) { ++ return resolve(false) ++ } ++ ++ const lastPingAtStart = this.lastPing ++ ++ let settled = false ++ const cleanup = () => { ++ if (settled) return ++ settled = true ++ this.off('pong', onPong) ++ if (timeout) clearTimeout(timeout as any) ++ } ++ ++ const onPong = () => { ++ if (this.lastPing <= lastPingAtStart) return ++ cleanup() ++ resolve(true) ++ } ++ ++ this.once('pong', onPong) ++ ++ const timeout = setTimeout(() => { ++ cleanup() ++ resolve(false) ++ }, timeoutMs) ++ ++ try { ++ this.connection.send(JSON.stringify({ msg: 'ping' })) ++ } catch { ++ cleanup() ++ resolve(false) ++ } ++ }) ++ } ++ + /** Check if websocket connected and ready. */ + get connected () { + return !!( +@@ -254,7 +375,7 @@ export class Socket extends EventEmitter { + return resolve() + } + this.once(listener, (result: any) => { +- this.off('disconnect', reject) ++ this.off('disconnected', reject) + return (result.error ? reject(result.error) : resolve({ ...(/connect|ping|pong/.test(obj.msg) ? {} : { id }) , ...result })) + }) + }) +@@ -447,7 +568,7 @@ export class DDPDriver extends EventEmitter implements ISocket, IDriver { + ...config, + ...moreConfigs, + host: host.replace(/(^\w+:|^)\/\//, ''), +- timeout: 20000 ++ timeout: 10000 + // reopen: number + // ping: number + // close: number +@@ -503,6 +624,22 @@ export class DDPDriver extends EventEmitter implements ISocket, IDriver { + return this.ddp.checkAndReopen() + } + ++ reopenNow = (): Promise => { ++ return this.ddp.reopenNow() ++ } ++ ++ probe = (timeoutMs?: number): Promise => { ++ return this.ddp.probe(timeoutMs) ++ } ++ ++ get lastPing (): number { ++ return this.ddp.lastPing ++ } ++ ++ get pingInterval (): number { ++ return this.ddp.config.ping ++ } ++ + subscribe = (topic: string, eventname: string, ...args: any[]): Promise => { + this.logger.info(`[DDP driver] Subscribing to ${topic} | ${JSON.stringify(args)}`) + return this.ddp.subscribe(topic, [eventname, { 'useCollection': false, 'args': args }]) +@@ -549,10 +686,70 @@ export class DDPDriver extends EventEmitter implements ISocket, IDriver { 'uiInteraction', 'e2ekeyRequest', 'userData', @@ -56,3 +242,88 @@ index 19d31ae..9f752df 100644 ].map(event => this.subscribe(topic, `${this.userId}/${event}`, false))) } ++ /** ++ * Re-send the user's media-signal and media-calls subscriptions on the current ++ * socket and resolve when the server acks them with `ready`. This gives the app ++ * an observable readiness signal after a forced reconnect. ++ * ++ * If the subscriptions are not yet present (e.g. immediately after reopenNow), ++ * it polls the socket subscription map until they appear or the timeout expires. ++ */ ++ waitForNotifyUserMediaSubs = (timeoutMs = 8000): Promise => { ++ if (!this.userId) { ++ return Promise.resolve(false) ++ } ++ const topic = 'stream-notify-user' ++ const names = ['media-signal', 'media-calls'] ++ const userId = this.userId ++ const findSubs = () => Object.keys(this.ddp.subscriptions || {}) ++ .map(id => this.ddp.subscriptions[id]) ++ .filter((sub: any) => ( ++ sub && ++ sub.name === topic && ++ names.some(name => sub.params?.[0] === `${userId}/${name}`) ++ )) ++ // Go through the raw socket: the driver's subscribe() wrapper reshapes its ++ // arguments and would drop the subscription id, making the server treat the ++ // resubscribe as a brand new subscription. ++ const resubscribe = (subs: any[]) => Promise.all( ++ subs.map((sub: any) => this.ddp.subscribe(topic, sub.params, undefined, sub.id)) ++ ) ++ .then(() => true) ++ .catch(() => false) ++ return new Promise(resolve => { ++ let settled = false ++ let inFlight = false ++ const finish = (value: boolean) => { ++ if (settled) return ++ settled = true ++ clearInterval(poll) ++ clearTimeout(deadline) ++ resolve(value) ++ } ++ const attempt = () => { ++ if (inFlight) return ++ const subs = findSubs() ++ const allPresent = names.every(name => subs.some((sub: any) => sub.params?.[0] === `${userId}/${name}`)) ++ if (allPresent) { ++ inFlight = true ++ resubscribe(subs).then(value => { ++ inFlight = false ++ finish(value) ++ }) ++ } ++ } ++ const deadline = setTimeout(() => finish(false), timeoutMs) ++ const poll = setInterval(attempt, 100) ++ attempt() ++ }) ++ } ++ + subscribeRoom = (rid: string, ...args: any[]): Promise => { + const topic = 'stream-notify-room' + return Promise.all([ +diff --git a/node_modules/@rocket.chat/sdk/lib/clients/Rocketchat.ts b/node_modules/@rocket.chat/sdk/lib/clients/Rocketchat.ts +index 591c1b9..82165c0 100644 +--- a/node_modules/@rocket.chat/sdk/lib/clients/Rocketchat.ts ++++ b/node_modules/@rocket.chat/sdk/lib/clients/Rocketchat.ts +@@ -6,6 +6,7 @@ export default class RocketChatClient extends ClientRest implements ISocket { + userId: string = '' + logger: ILogger = Logger + socket: Promise ++ ddp?: any + config: any + + constructor ({ logger, allPublic, rooms, integrationId, protocol = Protocols.DDP, ...config }: any) { +@@ -16,7 +17,10 @@ export default class RocketChatClient extends ClientRest implements ISocket { + // this.socket = import(/* webpackChunkName: 'mqtt' */ '../drivers/mqtt').then(({ MQTTDriver }) => new MQTTDriver({ ...config, logger })) + // break + case Protocols.DDP: +- this.socket = import(/* webpackChunkName: 'ddp' */ '../drivers/ddp').then(({ DDPDriver }) => new DDPDriver({ ...config, logger })) ++ this.socket = import(/* webpackChunkName: 'ddp' */ '../drivers/ddp').then(({ DDPDriver }) => { ++ this.ddp = new DDPDriver({ ...config, logger }) ++ return this.ddp ++ }) + break + default: + throw new Error(`Invalid Protocol: ${protocol}, valids: ${Object.keys(Protocols).join()}`)