diff --git a/packages/client/src/Call.ts b/packages/client/src/Call.ts index e97d65c40c..5d67bf5a18 100644 --- a/packages/client/src/Call.ts +++ b/packages/client/src/Call.ts @@ -46,6 +46,7 @@ import { EndCallResponse, GetCallReportResponse, GetCallResponse, + GetCallRingStateResponse, GetCallSessionParticipantStatsDetailsResponse, GetOrCreateCallRequest, GetOrCreateCallResponse, @@ -148,7 +149,7 @@ import { StatsReporter, Tracer, } from './stats'; -import type { ClientEventReporter, JoinReason } from './reporting'; +import type { ClientEventReporter, JoinReason, JoinSource } from './reporting'; import { AudioBindingsWatchdog } from './helpers/AudioBindingsWatchdog'; import { BlockedAudioTracker } from './helpers/BlockedAudioTracker'; import { TrackSubscriptionManager } from './helpers/TrackSubscriptionManager'; @@ -156,6 +157,7 @@ import { DynascaleManager } from './helpers/DynascaleManager'; import { createFirstVideoFrameDetector } from './helpers/firstVideoFrame'; import { ViewportTracker } from './helpers/ViewportTracker'; import { PermissionsContext } from './permissions'; +import { RingStatePoller, RingTimeout, resolveOwnRingOutcome } from './ringing'; import { CallTypes } from './CallType'; import { StreamClient } from './coordinator/connection/client'; import { retryInterval, sleep } from './coordinator/connection/utils'; @@ -298,7 +300,8 @@ export class Call { private statsReporter?: StatsReporter; private sfuStatsReporter?: SfuStatsReporter; private lastStatsOptions?: StatsOptions; - private dropTimeout: ReturnType | undefined; + private ringTimeout: RingTimeout | undefined; + private ringStatePoller: RingStatePoller | undefined; private readonly clientState: ClientState; public readonly streamClient: StreamClient; @@ -533,33 +536,20 @@ export class Call { createSubscription(this.state.session$, (session) => { if (!this.ringing) return; - const receiverId = this.clientState.connectedUser?.id; - if (!receiverId) return; - - const isAcceptedByMe = Boolean(session?.accepted_by[receiverId]); - const isRejectedByMe = Boolean(session?.rejected_by[receiverId]); - - if (isAcceptedByMe || isRejectedByMe) { - this.cancelAutoDrop(); - } - - const isAcceptedElsewhere = - isAcceptedByMe && this.state.callingState === CallingState.RINGING; + const { settledByMe, leaveReason } = resolveOwnRingOutcome({ + session, + currentUserId: this.currentUserId, + callingState: this.state.callingState, + }); + if (settledByMe) this.cancelAutoDrop(); + if (!leaveReason || hasPending(this.joinLeaveConcurrencyTag)) return; - if ( - (isAcceptedElsewhere || isRejectedByMe) && - !hasPending(this.joinLeaveConcurrencyTag) - ) { - globalThis.streamRNVideoSDK?.callingX?.endCall( - this, - isAcceptedElsewhere ? 'answeredElsewhere' : 'rejected', + globalThis.streamRNVideoSDK?.callingX?.endCall(this, leaveReason); + this.leave().catch(() => { + this.logger.error( + 'Could not leave a call that was accepted or rejected elsewhere', ); - this.leave().catch(() => { - this.logger.error( - 'Could not leave a call that was accepted or rejected elsewhere', - ); - }); - } + }); }), ); }; @@ -604,6 +594,7 @@ export class Call { this.state.setCallingState(CallingState.RINGING); } this.scheduleAutoDrop(); + this.scheduleRingStatePolling(); this.leaveCallHooks.add(registerRingingCallEventHandlers(this)); } }; @@ -709,6 +700,13 @@ export class Call { throw new Error('Cannot leave call that has already been left.'); } + // before the first await: the calling state stays RINGING well into the + // teardown, so pause both watchdogs before they can race this leave. They + // keep their deadlines, so a failed leave resumes them without handing the + // ring another window. + this.ringTimeout?.pause(); + this.ringStatePoller?.pause(); + await withoutConcurrency(this.joinLeaveConcurrencyTag, async () => { const callingState = this.state.callingState; @@ -817,6 +815,7 @@ export class Call { this.unifiedSessionId = undefined; this.ringingSubject.next(false); this.cancelAutoDrop(); + this.cancelRingStatePolling(); this.clientState.unregisterCall(this); globalThis.streamRNVideoSDK?.callManager.stop({ @@ -859,6 +858,17 @@ export class Call { this.logger.warn('Failed to dispose media engine', err); }); } + }).catch((err) => { + if ( + !hasPending(this.joinLeaveConcurrencyTag) && + this.state.callingState === CallingState.RINGING + ) { + // resume, never re-arm: a fresh watchdog would restart the ring + // window this leave was already most of the way through + this.ringTimeout?.start(); + this.ringStatePoller?.resume(); + } + throw err; }); }; @@ -1052,6 +1062,27 @@ export class Call { ); }; + /** + * Returns who accepted, rejected or missed the ring for a call session. + * Safe to poll: it performs no writes and emits no events. + * + * @param callSessionId the call session to read. Defaults to the current one. + * Pass it explicitly to read a session that has already ended, as ending a + * call clears its current session. + */ + getRingState = async ( + callSessionId?: string, + ): Promise => { + const sessionId = callSessionId ?? this.state.session?.id; + if (!sessionId) { + throw new Error('Cannot read the ring state: the call has no session'); + } + return this.streamClient.get( + `${this.streamClientBasePath}/ring_state`, + { call_session_id: sessionId }, + ); + }; + /** * A shortcut for {@link Call.get} with `notify` parameter set to `true`. * Will send a `call.notification` event to the call members. @@ -1108,12 +1139,14 @@ export class Call { joinResponseTimeout, rpcRequestTimeout, allowOwnTracksLoopback = false, + joinSource, ...data }: JoinCallData & { maxJoinRetries?: number; joinResponseTimeout?: number; rpcRequestTimeout?: number; allowOwnTracksLoopback?: boolean; + joinSource?: JoinSource; } = {}): Promise => { const callingState = this.state.callingState; @@ -1155,7 +1188,7 @@ export class Call { try { await this.clientEventReporter.withJoinLifecycle( this.cid, - 'first-attempt', + { joinReason: 'first-attempt', joinSource }, async () => { for (let attempt = 0; attempt < maxJoinRetries; attempt++) { try { @@ -2077,8 +2110,10 @@ export class Call { this.reconnectReason === ReconnectReason.NETWORK_BACK_ONLINE ? 'network-available' : 'full-rejoin'; - await this.clientEventReporter.withJoinLifecycle(this.cid, joinReason, () => - this.doJoin(this.joinCallData), + await this.clientEventReporter.withJoinLifecycle( + this.cid, + { joinReason }, + () => this.doJoin(this.joinCallData), ); await this.restorePublishedTracks(); this.restoreSubscribedTracks(); @@ -2113,7 +2148,7 @@ export class Call { const currentSfu = currentSfuClient.edgeName; await this.clientEventReporter.withJoinLifecycle( this.cid, - 'migration', + { joinReason: 'migration' }, () => this.doJoin({ ...this.joinCallData, @@ -3110,42 +3145,39 @@ export class Call { */ private scheduleAutoDrop = () => { this.cancelAutoDrop(); - - const settings = this.state.settings; - if (!settings) return; - // ignore if the call is not ringing - if (this.state.callingState !== CallingState.RINGING) return; - - const timeoutInMs = this.isCreatedByMe - ? settings.ring.auto_cancel_timeout_ms - : settings.ring.incoming_call_timeout_ms; - - // 0 means no auto-drop - if (timeoutInMs <= 0) return; - this.dropTimeout = setTimeout(() => { - // the call might have stopped ringing by this point, - // e.g. it was already accepted and joined - if (this.state.callingState !== CallingState.RINGING) return; - this.leave({ - reject: true, - reason: 'timeout', - message: `ringing timeout - ${ - this.isCreatedByMe - ? 'no one accepted' - : `user didn't interact with incoming call screen` - }`, - }).catch((err) => { - this.logger.error('Failed to drop call', err); - }); - }, timeoutInMs); + this.ringTimeout = new RingTimeout(this); + this.ringTimeout.start(); }; /** * Cancels a scheduled auto-drop timeout. */ private cancelAutoDrop = () => { - clearTimeout(this.dropTimeout); - this.dropTimeout = undefined; + this.ringTimeout?.stop(); + this.ringTimeout = undefined; + }; + + /** + * Starts polling for the ring outcome. Applicable only to ringing calls the + * current user created. + */ + private scheduleRingStatePolling = () => { + this.cancelRingStatePolling(); + + if (!this.isCreatedByMe) return; + const options = this.streamClient.options.ringStatePolling; + if (options === false) return; + + this.ringStatePoller = new RingStatePoller(this, options); + this.ringStatePoller.start(); + }; + + /** + * Cancels the ring state polling. + */ + private cancelRingStatePolling = () => { + this.ringStatePoller?.stop(); + this.ringStatePoller = undefined; }; /** diff --git a/packages/client/src/__tests__/Call.autodrop.test.ts b/packages/client/src/__tests__/Call.autodrop.test.ts index ab851b18c6..ce9b715893 100644 --- a/packages/client/src/__tests__/Call.autodrop.test.ts +++ b/packages/client/src/__tests__/Call.autodrop.test.ts @@ -1,104 +1,156 @@ import '../rtc/__tests__/mocks/webrtc.mocks'; import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { fromPartial } from '@total-typescript/shoehorn'; import { Call } from '../Call'; import { StreamClient } from '../coordinator/connection/client'; import { ClientEventReporter } from '../reporting'; import { generateUUIDv4 } from '../coordinator/connection/utils'; import { CallingState, ClientState } from '../store'; +import { CallSettingsResponse } from '../gen/coordinator'; +const TIMEOUT_MS = 30_000; + +// The timeout's own behaviour is covered by `ringing/__tests__/RingTimeout`. +// This file covers the wiring: that a ringing call arms one and that leaving +// cancels it. describe('Auto drop ringing calls', () => { - let call: Call; const userId = 'jane'; + let call: Call; - beforeEach(async () => { - vi.useFakeTimers(); - + const ringingCall = () => { const clientState = new ClientState(); const streamClient = new StreamClient('abc'); - call = new Call({ + const newCall = new Call({ type: 'test', id: generateUUIDv4(), streamClient, clientEventReporter: new ClientEventReporter({ streamClient }), clientState, + ringing: true, }); - // @ts-expect-error mocking only what we need for the test - clientState['connectedUserSubject'].next({ - id: userId, - }); - - call.state['callingStateSubject'].next(CallingState.RINGING); - - vi.spyOn(call, 'leave').mockImplementation(async () => { - console.log(`TEST: leave() called`); - }); - }); - - it('caller should drop ringing calls after a timeout if no one accepted', async () => { - call.state['settingsSubject'].next({ - // @ts-expect-error mocking only what we need for the test, we use fake timers, so undefined for timeout works - ring: {}, - // @ts-expect-error mocking only what we need for the test - screensharing: { - enabled: false, - target_resolution: { - width: 100, - height: 100, + clientState.setConnectedUser(fromPartial({ id: userId })); + newCall.state['createdBySubject'].next(fromPartial({ id: userId })); + newCall.state['settingsSubject'].next( + fromPartial({ + ring: { + auto_cancel_timeout_ms: TIMEOUT_MS, + incoming_call_timeout_ms: TIMEOUT_MS, + missed_call_timeout_ms: TIMEOUT_MS, }, - }, - }); + screensharing: { + enabled: false, + target_resolution: { width: 100, height: 100 }, + }, + }), + ); + newCall.state['sessionSubject'].next( + fromPartial({ + id: 'session-1', + accepted_by: {}, + rejected_by: {}, + missed_by: {}, + }), + ); + newCall.state['callingStateSubject'].next(CallingState.RINGING); + + vi.spyOn(newCall, 'leave').mockResolvedValue(undefined); + return newCall; + }; + + beforeEach(() => { + vi.useFakeTimers(); + call = ringingCall(); + }); - // @ts-expect-error mocking only what we need for the test - call.state['createdBySubject'].next({ - id: userId, - }); + afterEach(() => { + call['cancelAutoDrop'](); + vi.useRealTimers(); + vi.restoreAllMocks(); + }); - // black-box test, calling private method - call['scheduleAutoDrop'](); + it('is armed when a call starts ringing', async () => { + call.state['callingStateSubject'].next(CallingState.IDLE); - await vi.runAllTimersAsync(); + call['handleRingingCall'](); + expect(call['ringTimeout']).toBeDefined(); + await vi.advanceTimersByTimeAsync(TIMEOUT_MS); expect(call.leave).toHaveBeenCalledWith({ reject: true, reason: 'timeout', - message: `ringing timeout - no one accepted`, + message: 'ringing timeout - no one accepted', }); }); - it(`callee should drop ringing calls after a timeout if user didn't interact with incoming call screen`, async () => { - call.state['settingsSubject'].next({ - // @ts-expect-error mocking only what we need for the test, we use fake timers, so undefined for timeout works - ring: {}, - // @ts-expect-error mocking only what we need for the test - screensharing: { - enabled: false, - target_resolution: { - width: 100, - height: 100, - }, - }, - }); + it('is cancelled by cancelAutoDrop', async () => { + call['scheduleAutoDrop'](); - // @ts-expect-error mocking only what we need for the test - call.state['createdBySubject'].next({ - id: 'not-' + userId, - }); + call['cancelAutoDrop'](); - // black-box test, calling private method + expect(call['ringTimeout']).toBeUndefined(); + await vi.advanceTimersByTimeAsync(TIMEOUT_MS); + expect(call.leave).not.toHaveBeenCalled(); + }); + + // the calling state stays RINGING well into teardown, so the watchdogs have + // to be paused before `leave` awaits anything + it('pauses the watchdogs synchronously when leave starts', async () => { call['scheduleAutoDrop'](); + call['scheduleRingStatePolling'](); + const timeout = call['ringTimeout']; + vi.spyOn(call, 'leave').mockRestore(); + vi.spyOn(call, 'reject').mockResolvedValue(fromPartial({})); - await vi.runAllTimersAsync(); + const poller = call['ringStatePoller']; - expect(call.leave).toHaveBeenCalledWith({ - reject: true, - reason: 'timeout', - message: `ringing timeout - user didn't interact with incoming call screen`, - }); + const leaving = call.leave({ reject: false }); + + // paused, not cancelled: both keep the window this leave is ending + expect(call['ringTimeout']).toBe(timeout); + expect(call['ringStatePoller']).toBe(poller); + expect(timeout!['timeoutId']).toBeUndefined(); + expect(timeout!['stopped']).toBe(false); + expect(poller!['idleTimeoutId']).toBeUndefined(); + expect(poller!['stopped']).toBe(false); + + await leaving.catch(() => {}); + + expect(timeout!['stopped']).toBe(true); + expect(poller!['stopped']).toBe(true); }); - afterEach(() => { - vi.useRealTimers(); + it('restores both watchdogs when rejecting the ring fails', async () => { + call['scheduleAutoDrop'](); + call['scheduleRingStatePolling'](); + const timeout = call['ringTimeout']; + const deadlineAt = timeout!['deadlineAt']; + const poller = call['ringStatePoller']; + const pollDeadlineAt = poller!['deadlineAt']; + vi.spyOn(call, 'leave').mockRestore(); + vi.spyOn(call, 'reject').mockRejectedValueOnce(new Error('transient')); + + await expect(call.leave({ reject: true })).rejects.toThrow('transient'); + + expect(call.state.callingState).toBe(CallingState.RINGING); + expect(call['ringTimeout']).toBe(timeout); + expect(timeout!['timeoutId']).toBeDefined(); + expect(timeout!['deadlineAt']).toBe(deadlineAt); + expect(call['ringStatePoller']).toBe(poller); + expect(poller!['deadlineAt']).toBe(pollDeadlineAt); + expect(poller!['stopped']).toBe(false); + }); + + it('replaces a previously armed timeout', async () => { + call['scheduleAutoDrop'](); + const first = call['ringTimeout']; + + call['scheduleAutoDrop'](); + + expect(call['ringTimeout']).not.toBe(first); + expect(first!['stopped']).toBe(true); + await vi.advanceTimersByTimeAsync(TIMEOUT_MS); + expect(call.leave).toHaveBeenCalledTimes(1); }); }); diff --git a/packages/client/src/__tests__/Call.getRingState.test.ts b/packages/client/src/__tests__/Call.getRingState.test.ts new file mode 100644 index 0000000000..0186147818 --- /dev/null +++ b/packages/client/src/__tests__/Call.getRingState.test.ts @@ -0,0 +1,87 @@ +import '../rtc/__tests__/mocks/webrtc.mocks'; + +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { fromPartial } from '@total-typescript/shoehorn'; +import { Call } from '../Call'; +import { StreamClient } from '../coordinator/connection/client'; +import { ClientEventReporter } from '../reporting'; +import { generateUUIDv4 } from '../coordinator/connection/utils'; +import { ClientState } from '../store'; +import { CallSessionResponse } from '../gen/coordinator'; + +describe('Call.getRingState', () => { + const callId = generateUUIDv4(); + + const fakeCall = (sessionId?: string) => { + const streamClient = new StreamClient('abc'); + const call = new Call({ + type: 'test', + id: callId, + streamClient, + clientEventReporter: new ClientEventReporter({ streamClient }), + clientState: new ClientState(), + }); + + if (sessionId) { + call.state['sessionSubject'].next( + fromPartial({ id: sessionId }), + ); + } + + const get = vi + .spyOn(streamClient, 'get') + .mockResolvedValue(fromPartial({ session_id: sessionId })); + + return { call, get }; + }; + + afterEach(() => { + vi.restoreAllMocks(); + }); + + it('reads the ring state of the current session', async () => { + const { call, get } = fakeCall('session-1'); + + await call.getRingState(); + + expect(get).toHaveBeenCalledWith(`/call/test/${callId}/ring_state`, { + call_session_id: 'session-1', + }); + }); + + it('reads the ring state of an explicitly named session', async () => { + // ending a call clears its current session, so a caller reconciling after + // `call.ended` has to name the session it rang on + const { call, get } = fakeCall('current-session'); + + await call.getRingState('ended-session'); + + expect(get).toHaveBeenCalledWith(`/call/test/${callId}/ring_state`, { + call_session_id: 'ended-session', + }); + }); + + it('returns the coordinator response', async () => { + const { call, get } = fakeCall('session-1'); + get.mockResolvedValue( + fromPartial({ + session_id: 'session-1', + accepted_by: { bob: '2026-08-24T10:00:04Z' }, + }), + ); + + await expect(call.getRingState()).resolves.toMatchObject({ + session_id: 'session-1', + accepted_by: { bob: '2026-08-24T10:00:04Z' }, + }); + }); + + it('rejects when the call has no session to read', async () => { + const { call, get } = fakeCall(); + + await expect(call.getRingState()).rejects.toThrow( + 'Cannot read the ring state: the call has no session', + ); + expect(get).not.toHaveBeenCalled(); + }); +}); diff --git a/packages/client/src/__tests__/Call.lifecycle.test.ts b/packages/client/src/__tests__/Call.lifecycle.test.ts index 1d821886ff..a99700dc1a 100644 --- a/packages/client/src/__tests__/Call.lifecycle.test.ts +++ b/packages/client/src/__tests__/Call.lifecycle.test.ts @@ -79,6 +79,29 @@ describe('Call lifecycle wiring', () => { expect(audioBindingsOrder).toBeLessThan(dynascaleOrder); }); + // `joinSource` is reporting-only: it must reach the event reporter and never + // the coordinator's join request. + it('call.join() reports joinSource without putting it on the wire', async () => { + vi.spyOn(call, 'setup').mockResolvedValue(undefined); + const doJoin = vi + .spyOn(call as unknown as { doJoin: Call['join'] }, 'doJoin') + .mockResolvedValue(undefined); + const withJoinLifecycle = vi.spyOn( + call.clientEventReporter, + 'withJoinLifecycle', + ); + + await call.join({ joinSource: 'ring-poll-api', ring: true }); + + expect(withJoinLifecycle).toHaveBeenCalledWith( + call.cid, + { joinReason: 'first-attempt', joinSource: 'ring-poll-api' }, + expect.any(Function), + ); + expect(doJoin).toHaveBeenCalledTimes(1); + expect('joinSource' in doJoin.mock.calls[0][0]!).toBe(false); + }); + it('call.join() shares an in-flight join flow', async () => { const joinTask = promiseWithResolvers(); vi.spyOn(call, 'setup').mockResolvedValue(undefined); diff --git a/packages/client/src/__tests__/Call.ringSettled.test.ts b/packages/client/src/__tests__/Call.ringSettled.test.ts new file mode 100644 index 0000000000..e2f4a837c6 --- /dev/null +++ b/packages/client/src/__tests__/Call.ringSettled.test.ts @@ -0,0 +1,71 @@ +import '../rtc/__tests__/mocks/webrtc.mocks'; + +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { fromPartial } from '@total-typescript/shoehorn'; +import { Call } from '../Call'; +import { StreamClient } from '../coordinator/connection/client'; +import { ClientEventReporter } from '../reporting'; +import { generateUUIDv4 } from '../coordinator/connection/utils'; +import { CallingState, ClientState } from '../store'; +import { CallSessionResponse } from '../gen/coordinator'; + +const ME = 'jane'; + +// `resolveOwnRingOutcome` owns the decision and is tested on its own. This +// covers the effect around it: that it runs for a ringing call and that the +// `ringing` guard keeps it off every other call. +describe('Leaving a call settled by the current user', () => { + let call: Call; + + const createCall = async (ringing: boolean) => { + const clientState = new ClientState(); + const streamClient = new StreamClient('abc'); + call = new Call({ + type: 'test', + id: generateUUIDv4(), + streamClient, + clientEventReporter: new ClientEventReporter({ streamClient }), + clientState, + ringing, + }); + + clientState.setConnectedUser(fromPartial({ id: ME })); + vi.spyOn(call, 'leave').mockResolvedValue(undefined); + await call.setup(); + return call; + }; + + const settleByMe = () => + call.state['sessionSubject'].next( + fromPartial({ + id: 'session-1', + accepted_by: {}, + rejected_by: { [ME]: new Date().toISOString() }, + missed_by: {}, + participants: [], + participants_count_by_role: {}, + }), + ); + + afterEach(() => { + vi.restoreAllMocks(); + }); + + it('leaves a ringing call the current user rejected', async () => { + await createCall(true); + expect(call.state.callingState).toBe(CallingState.RINGING); + + settleByMe(); + + expect(call.leave).toHaveBeenCalled(); + }); + + it('ignores the same session on a call that is not ringing', async () => { + await createCall(false); + expect(call.state.callingState).toBe(CallingState.IDLE); + + settleByMe(); + + expect(call.leave).not.toHaveBeenCalled(); + }); +}); diff --git a/packages/client/src/coordinator/connection/types.ts b/packages/client/src/coordinator/connection/types.ts index 1327693086..791694d24f 100644 --- a/packages/client/src/coordinator/connection/types.ts +++ b/packages/client/src/coordinator/connection/types.ts @@ -200,6 +200,18 @@ export type Logger = ( ...args: unknown[] ) => void; +export type RingStatePollingOptions = { + /** + * Quiet time after the ring starts, before the first poll. Defaults to 15_000. + */ + startAfterMs?: number; + + /** + * The interval between polls. Defaults to 5_000. + */ + intervalMs?: number; +}; + export type StreamClientOptions = Partial & { /** * Used to disable warnings that are triggered by using connectUser or connectAnonymousUser server-side. @@ -305,6 +317,13 @@ export type StreamClientOptions = Partial & { */ rejectCallWhenBusy?: boolean; + /** + * Caller-side polling for the ring outcome, used when the `call.accepted`, + * `call.rejected` or `call.missed` event never arrives. Enabled by default; + * set to `false` to disable, or pass an object to tune the timings. + */ + ringStatePolling?: false | RingStatePollingOptions; + /** * Device persistence preference options (web only). */ diff --git a/packages/client/src/events/__tests__/call.test.ts b/packages/client/src/events/__tests__/call.test.ts index a909ac7da0..ee94809890 100644 --- a/packages/client/src/events/__tests__/call.test.ts +++ b/packages/client/src/events/__tests__/call.test.ts @@ -1,16 +1,8 @@ import { describe, expect, it, vi } from 'vitest'; -import { fromPartial } from '@total-typescript/shoehorn'; import { CallingState, ClientState } from '../../store'; +import { watchCallEnded, watchSfuCallEnded } from '../call'; import { - watchCallAccepted, - watchCallEnded, - watchCallRejected, - watchSfuCallEnded, -} from '../call'; -import { - CallAcceptedEvent, CallEndedEvent, - CallResponse, OwnCapability, RejectCallResponse, } from '../../gen/coordinator'; @@ -20,189 +12,7 @@ import { ClientEventReporter } from '../../reporting'; import { SfuEvent } from '../../gen/video/sfu/event/events'; import { CallEndedReason } from '../../gen/video/sfu/models/models'; -describe('Call ringing events', () => { - describe(`call.accepted`, () => { - it(`will ignore events from the current user`, async () => { - const call = fakeCall(); - vi.spyOn(call, 'join'); - const handler = watchCallAccepted(call); - const event: CallAcceptedEvent = { - type: 'call.accepted', - // @ts-expect-error incomplete data - user: { id: 'test-user-id' }, - }; - await handler(event); - - expect(call.join).not.toHaveBeenCalled(); - }); - - it(`will join the call for the caller if atleast one callee has accepted`, async () => { - const call = fakeCall({ currentUserId: 'test-user' }); - vi.spyOn(call, 'join').mockImplementation(async () => { - console.log(`TEST: join() called`); - }); - const handler = watchCallAccepted(call); - const event: CallAcceptedEvent = { - type: 'call.accepted', - // @ts-expect-error incomplete data - user: { id: 'test-user-id-callee' }, - // @ts-expect-error incomplete data - call: { created_by: { id: 'test-user' } }, - }; - await handler(event); - - expect(call.join).toHaveBeenCalled(); - }); - }); - - it('will not join the call for the other callee automatically when someone accepts', async () => { - const call = fakeCall({ currentUserId: 'test-user-id-callee-2' }); - vi.spyOn(call, 'join').mockImplementation(async () => { - console.log(`TEST: join() called`); - }); - const handler = watchCallAccepted(call); - const event: CallAcceptedEvent = { - type: 'call.accepted', - // @ts-expect-error incomplete data - user: { id: 'test-user-id-callee-1' }, - // @ts-expect-error incomplete data - call: { created_by: { id: 'test-user-id-caller' } }, - }; - - await handler(event); - - expect(call.join).not.toHaveBeenCalled(); - }); - - describe(`call.rejected`, () => { - it(`caller will leave the call if all callees have rejected`, async () => { - const call = fakeCall({ currentUserId: 'm1' }); - call.state.updateFromCallResponse({ - ...fakeMetadata(), - // @ts-expect-error type issue - created_by: { id: 'm1' }, - }); - call.state.setMembers([ - // @ts-expect-error incomplete data - { user_id: 'm1' }, - // @ts-expect-error incomplete data - { user_id: 'm2' }, - // @ts-expect-error incomplete data - { user_id: 'm3' }, - ]); - call.state.setCallingState(CallingState.RINGING); - vi.spyOn(call, 'leave').mockImplementation(async () => { - console.log(`TEST: leave() called`); - }); - - const handler = watchCallRejected(call); - // all members reject the call - await handler({ - type: 'call.rejected', - // @ts-expect-error type issue - user: { - id: 'm2', - }, - call: { - // @ts-expect-error type issue - created_by: { - id: 'm1', - }, - // @ts-expect-error type issue - session: { - rejected_by: { - m2: new Date().toISOString(), - m3: new Date().toISOString(), - }, - }, - }, - }); - expect(call.leave).toHaveBeenCalledWith({ - reject: true, - reason: 'cancel', - message: 'ring: everyone rejected', - }); - }); - - it(`caller will not leave the call if only one callee rejects`, async () => { - const call = fakeCall(); - call.state.updateFromCallResponse({ - ...fakeMetadata(), - // @ts-expect-error type issue - created_by: { id: 'm0' }, - }); - // @ts-expect-error incomplete data - call.state.setMembers([{ user_id: 'm1' }, { user_id: 'm2' }]); - vi.spyOn(call, 'leave').mockImplementation(async () => { - console.log(`TEST: leave() called`); - }); - const handler = watchCallRejected(call); - - // only one member rejects the call - const event: CallAcceptedEvent = { - type: 'call.rejected', - // @ts-expect-error type issue - user: { - id: 'm2', - }, - call: { - // @ts-expect-error type issue - created_by: { - id: 'm0', - }, - // @ts-expect-error type issue - session: { - rejected_by: { - m2: new Date().toISOString(), - }, - }, - }, - }; - await handler(event); - - expect(call.leave).not.toHaveBeenCalled(); - }); - - it('callee will leave the call if caller rejects', async () => { - const call = fakeCall({ currentUserId: 'm1' }); - call.state.updateFromCallResponse({ - ...fakeMetadata(), - // @ts-expect-error type issue - created_by: { id: 'm0' }, - }); - // @ts-expect-error incomplete data - call.state.setMembers([{ user_id: 'm1' }, { user_id: 'm2' }]); - vi.spyOn(call, 'leave').mockImplementation(async () => { - console.log(`TEST: leave() called`); - }); - const handler = watchCallRejected(call); - - // only one member rejects the call - const event: CallAcceptedEvent = { - type: 'call.rejected', - // @ts-expect-error type issue - user: { - id: 'm0', - }, - call: { - // @ts-expect-error type issue - created_by: { - id: 'm0', - }, - // @ts-expect-error type issue - session: { - rejected_by: { - m0: new Date().toISOString(), - }, - }, - }, - }; - await handler(event); - - expect(call.leave).toHaveBeenCalled(); - }); - }); - +describe('Call lifecycle events', () => { describe(`call.ended`, () => { it(`will leave the call unless joined`, async () => { const call = fakeCall(); @@ -377,28 +187,3 @@ const fakeCall = ({ ring = true, currentUserId = 'test-user-id' } = {}) => { ringing: ring, }); }; - -const fakeMetadata = (): CallResponse => { - return fromPartial({ - id: '12345', - type: 'development', - cid: 'development:12345', - - created_by: { - id: 'test-user-id', - }, - blocked_user_ids: [], - egress: {}, - - settings: { - ring: { - auto_cancel_timeout_ms: 30000, - incoming_call_timeout_ms: 30000, - missed_call_timeout_ms: 30000, - }, - screensharing: { - target_resolution: undefined, - }, - }, - }); -}; diff --git a/packages/client/src/events/call.ts b/packages/client/src/events/call.ts index 7656008cc0..09bb49f207 100644 --- a/packages/client/src/events/call.ts +++ b/packages/client/src/events/call.ts @@ -1,81 +1,9 @@ import { CallingState } from '../store'; import { Call } from '../Call'; -import { - CallAcceptedEvent, - CallRejectedEvent, - OwnCapability, -} from '../gen/coordinator'; +import { OwnCapability } from '../gen/coordinator'; import { CallEnded } from '../gen/video/sfu/event/events'; import { CallEndedReason } from '../gen/video/sfu/models/models'; -/** - * Event handler that watched the delivery of `call.accepted`. - * Once the event is received, the call is joined. - */ -export const watchCallAccepted = (call: Call) => { - return async function onCallAccepted(event: CallAcceptedEvent) { - // We want to discard the event if it's from the current user - if (event.user.id === call.currentUserId) return; - const { state } = call; - if ( - event.call.created_by.id === call.currentUserId && - state.callingState === CallingState.RINGING - ) { - await call.join(); - } - }; -}; - -/** - * Event handler that watches delivery of `call.rejected` Websocket event. - * Once the event is received, the call is left. - */ -export const watchCallRejected = (call: Call) => { - return async function onCallRejected(event: CallRejectedEvent) { - // We want to discard the event if it's from the current user - if (event.user.id === call.currentUserId) return; - const { call: eventCall } = event; - const { session: callSession } = eventCall; - - if (!callSession) { - call.logger.warn( - 'No call session provided. Ignoring call.rejected event.', - event, - ); - return; - } - - const rejectedBy = callSession.rejected_by; - const { members, callingState } = call.state; - if (callingState !== CallingState.RINGING) { - call.logger.info( - 'Call is not in ringing mode (it is either accepted or rejected already). Ignoring call.rejected event.', - event, - ); - return; - } - if (call.isCreatedByMe) { - const everyoneElseRejected = members - .filter((m) => m.user_id !== call.currentUserId) - .every((m) => rejectedBy[m.user_id]); - if (everyoneElseRejected) { - call.logger.info('everyone rejected, leaving the call'); - await call.leave({ - reject: true, - reason: 'cancel', - message: 'ring: everyone rejected', - }); - } - } else { - if (rejectedBy[eventCall.created_by.id]) { - call.logger.info('call creator rejected, leaving call'); - globalThis.streamRNVideoSDK?.callingX?.endCall(call, 'remote'); - await call.leave({ message: 'ring: creator rejected' }); - } - } - }; -}; - /** * Event handler that watches the delivery of `call.ended` Websocket event. */ diff --git a/packages/client/src/events/callEventHandlers.ts b/packages/client/src/events/callEventHandlers.ts index e4a008e359..9d3fda4e12 100644 --- a/packages/client/src/events/callEventHandlers.ts +++ b/packages/client/src/events/callEventHandlers.ts @@ -1,12 +1,11 @@ import { Call } from '../Call'; +import { reconcileRingState } from '../ringing'; import { Dispatcher } from '../rtc'; import { handleRemoteSoftMute, watchAudioLevelChanged, - watchCallAccepted, watchCallEnded, watchCallGrantsUpdated, - watchCallRejected, watchConnectionQualityChanged, watchDominantSpeakerChanged, watchInboundStateNotification, @@ -21,16 +20,6 @@ import { watchTrackPublished, watchTrackUnpublished, } from '../events'; -import { - AllCallEvents, - AllClientCallEvents, - CallEventListener, -} from '../coordinator/connection/types'; - -type RingCallEvents = Extract< - AllClientCallEvents, - 'call.accepted' | 'call.rejected' | 'call.missed' ->; /** * Registers the default event handlers for a call during its lifecycle. @@ -83,19 +72,19 @@ export const registerEventHandlers = (call: Call, dispatcher: Dispatcher) => { * @param call the call to register event handlers for. */ export const registerRingingCallEventHandlers = (call: Call) => { - const coordinatorRingEvents: { - [key in RingCallEvents]: ( - call: Call, - ) => CallEventListener; - } = { - 'call.accepted': watchCallAccepted(call), - 'call.rejected': watchCallRejected(call), + const reconcile = () => { + reconcileRingState(call, 'ring-ws').catch((err) => { + call.logger.error('Failed to reconcile the ring state', err); + }); }; - const eventHandlers = Object.keys(coordinatorRingEvents).map((event) => { - const eventName = event as RingCallEvents; - return call.on(eventName, coordinatorRingEvents[eventName]); - }); + // each event needs its own closure. `call.missed` is deliberately absent: + // nothing in the reconciler acts on `missed_by`, so the auto-drop owns the + // "nobody answered" case until the server-owned ring timeout lands. + const eventHandlers = [ + call.on('call.accepted', () => reconcile()), + call.on('call.rejected', () => reconcile()), + ]; return () => { eventHandlers.forEach((unsubscribe) => unsubscribe()); diff --git a/packages/client/src/gen/coordinator/index.ts b/packages/client/src/gen/coordinator/index.ts index 9ce42f0b8d..f3f69b8af6 100644 --- a/packages/client/src/gen/coordinator/index.ts +++ b/packages/client/src/gen/coordinator/index.ts @@ -5130,6 +5130,73 @@ export interface GetCallResponse { */ own_capabilities: Array; } +/** + * The ring state of a call session: who accepted, rejected, or missed the ring. + * @export + * @interface GetCallRingStateResponse + */ +export interface GetCallRingStateResponse { + /** + * Users that accepted the call, mapped to when they accepted + * @type {{ [key: string]: string; }} + * @memberof GetCallRingStateResponse + */ + accepted_by: { [key: string]: string }; + /** + * The CID of the call + * @type {string} + * @memberof GetCallRingStateResponse + */ + call_cid: string; + /** + * When the call ended + * @type {string} + * @memberof GetCallRingStateResponse + */ + call_ended_at?: string; + /** + * The user that created the call, i.e. the caller + * @type {string} + * @memberof GetCallRingStateResponse + */ + created_by_user_id: string; + /** + * Duration of the request in milliseconds + * @type {string} + * @memberof GetCallRingStateResponse + */ + duration: string; + /** + * Users that missed the call, mapped to when they were marked as missed + * @type {{ [key: string]: string; }} + * @memberof GetCallRingStateResponse + */ + missed_by: { [key: string]: string }; + /** + * Users that rejected the call, mapped to when they rejected + * @type {{ [key: string]: string; }} + * @memberof GetCallRingStateResponse + */ + rejected_by: { [key: string]: string }; + /** + * When the call session ended + * @type {string} + * @memberof GetCallRingStateResponse + */ + session_ended_at?: string; + /** + * The call session this state belongs to, empty when the call has never rung + * @type {string} + * @memberof GetCallRingStateResponse + */ + session_id: string; + /** + * When the call session started + * @type {string} + * @memberof GetCallRingStateResponse + */ + session_started_at?: string; +} /** * Basic response information * @export diff --git a/packages/client/src/reporting/ClientEventReporter.ts b/packages/client/src/reporting/ClientEventReporter.ts index ba2fc8bf82..7479536fa4 100644 --- a/packages/client/src/reporting/ClientEventReporter.ts +++ b/packages/client/src/reporting/ClientEventReporter.ts @@ -45,6 +45,13 @@ export type ReportedIceState = 'CONNECTED' | 'FAILED' | 'NOT_CONNECTED'; export type JoinReason = 'first-attempt' | 'network-available' | 'migration' | 'full-rejoin'; +/** + * What triggered an automatic join, reported as `source` on the call's + * CoordinatorJoin events. `ring-ws` is a ring WebSocket event, `ring-poll-api` + * is the ring state poller. + */ +export type JoinSource = 'ring-ws' | 'ring-poll-api'; + export type ClientEventStandardCode = | 'CLIENT_ABORTED' | 'BACKEND_LEAVE' @@ -67,6 +74,9 @@ export type ClientEventReporterOptions = { enabled?: boolean; }; +// TODO OL: update OpenAPI +type ReportedClientEvent = ClientEvent & { source?: JoinSource }; + type StageError = { reason: string; code: string; @@ -78,6 +88,7 @@ type StagePairState = { startedAt: number; joinAttemptIdSnapshot?: string; joinReasonSnapshot?: JoinReason; + joinSourceSnapshot?: JoinSource; userIdSnapshot?: string; lastError?: StageError; }; @@ -104,6 +115,7 @@ export class ClientEventReporter { private callContexts = new Map(); private joinAttemptIds = new Map(); private joinReasons = new Map(); + private joinSources = new Map(); private coordinatorPairs = new Map(); private wsPairs = new Map(); @@ -238,6 +250,7 @@ export class ClientEventReporter { this.callContexts.delete(cid); this.joinAttemptIds.delete(cid); this.joinReasons.delete(cid); + this.joinSources.delete(cid); this.coordinatorPairs.delete(cid); this.wsPairs.delete(cid); @@ -269,15 +282,22 @@ export class ClientEventReporter { withJoinLifecycle = async ( cid: string, - joinReason: JoinReason, + options: { joinReason: JoinReason; joinSource?: JoinSource }, op: () => Promise, ): Promise => { + const { joinReason, joinSource } = options; + + if (joinSource) this.joinSources.set(cid, joinSource); + else this.joinSources.delete(cid); + this.startCorrelation(cid, joinReason); try { return await op(); } catch (err) { this.closeCallPairs(cid); throw err; + } finally { + this.joinSources.delete(cid); } }; @@ -436,6 +456,7 @@ export class ClientEventReporter { startedAt: Date.now(), joinAttemptIdSnapshot: this.joinAttemptIds.get(cid), joinReasonSnapshot: this.joinReasons.get(cid), + joinSourceSnapshot: this.joinSources.get(cid), }; this.coordinatorPairs.set(cid, pair); this.sendForCall(cid, { @@ -443,6 +464,7 @@ export class ClientEventReporter { ...(pair.joinReasonSnapshot && { join_reason: pair.joinReasonSnapshot, }), + ...(pair.joinSourceSnapshot && { source: pair.joinSourceSnapshot }), event_type: 'initiated', }); } @@ -457,6 +479,7 @@ export class ClientEventReporter { ...this.buildCommon(cid, 'CoordinatorJoin', pair), ...this.sessionIdField(cid), ...(pair.joinReasonSnapshot && { join_reason: pair.joinReasonSnapshot }), + ...(pair.joinSourceSnapshot && { source: pair.joinSourceSnapshot }), event_type: 'completed', outcome: 'success', retry_count_attempt: pair.attempts - 1, @@ -476,6 +499,7 @@ export class ClientEventReporter { ...this.buildCommon(cid, 'CoordinatorJoin', pair), ...this.sessionIdField(cid), ...(pair.joinReasonSnapshot && { join_reason: pair.joinReasonSnapshot }), + ...(pair.joinSourceSnapshot && { source: pair.joinSourceSnapshot }), event_type: 'completed', outcome: 'failure', retry_count_attempt: pair.attempts - 1, @@ -705,17 +729,19 @@ export class ClientEventReporter { }; }; - private send = (body: ClientEvent) => { + private send = (body: ReportedClientEvent) => { if (!this.enabled) return; void this.sendWithRetry(body); }; - private sendForCall = (cid: string, body: ClientEvent) => { + private sendForCall = (cid: string, body: ReportedClientEvent) => { if (!this.callContexts.has(cid)) return; this.send(body); }; - private sendWithRetry = async (body: ClientEvent): Promise => { + private sendWithRetry = async ( + body: ReportedClientEvent, + ): Promise => { for (let attempt = 0; attempt < 5; attempt++) { try { await this.streamClient.doAxiosRequest< diff --git a/packages/client/src/reporting/__tests__/ClientEventReporter.test.ts b/packages/client/src/reporting/__tests__/ClientEventReporter.test.ts index d59947e730..181f77c034 100644 --- a/packages/client/src/reporting/__tests__/ClientEventReporter.test.ts +++ b/packages/client/src/reporting/__tests__/ClientEventReporter.test.ts @@ -39,6 +39,12 @@ const iceEvent = ( describe('ClientEventReporter', () => { const cid = 'default:call-1'; + const callContext: CallReportContext = { + callType: 'default', + callId: 'call-1', + getCallSessionId: () => 'session-1', + getSfuId: () => 'sfu-1', + }; let doAxiosRequest: ReturnType; let reporter: ClientEventReporter; let connectId: string; @@ -72,13 +78,7 @@ describe('ClientEventReporter', () => { reporter = new ClientEventReporter({ streamClient }); connectId = reporter.startCoordinatorConnection('user-1'); - const ctx: CallReportContext = { - callType: 'default', - callId: 'call-1', - getCallSessionId: () => 'session-1', - getSfuId: () => 'sfu-1', - }; - reporter.registerCall(cid, ctx); + reporter.registerCall(cid, callContext); }); it('emits an initiated then a completed event on success', async () => { @@ -117,6 +117,183 @@ describe('ClientEventReporter', () => { expect(events[1]).toMatchObject({ join_reason: 'migration' }); }); + it('carries the joinSource across the whole join lifecycle', async () => { + await reporter.withJoinLifecycle( + cid, + { joinReason: 'first-attempt', joinSource: 'ring-poll-api' }, + () => reporter.track(cid, 'CoordinatorJoin', () => Promise.resolve('ok')), + ); + await flush(); + + const events = postedEvents().filter((e) => e.stage === 'CoordinatorJoin'); + expect(events).toHaveLength(2); + expect(events[0]).toMatchObject({ + event_type: 'initiated', + join_reason: 'first-attempt', + source: 'ring-poll-api', + }); + expect(events[1]).toMatchObject({ + event_type: 'completed', + outcome: 'success', + source: 'ring-poll-api', + }); + }); + + it('carries the joinSource on a failed join', async () => { + await expect( + reporter.withJoinLifecycle( + cid, + { joinReason: 'first-attempt', joinSource: 'ring-ws' }, + () => + reporter.track(cid, 'CoordinatorJoin', () => + Promise.reject(new Error('boom')), + ), + ), + ).rejects.toThrow('boom'); + await flush(); + + const events = postedEvents().filter((e) => e.stage === 'CoordinatorJoin'); + expect(events).toHaveLength(2); + expect(events[0]).toMatchObject({ source: 'ring-ws' }); + expect(events[1]).toMatchObject({ + event_type: 'completed', + outcome: 'failure', + source: 'ring-ws', + }); + }); + + it('omits the source key entirely when the join has none', async () => { + await reporter.withJoinLifecycle(cid, { joinReason: 'first-attempt' }, () => + reporter.track(cid, 'CoordinatorJoin', () => Promise.resolve('ok')), + ); + await flush(); + + const events = postedEvents().filter((e) => e.stage === 'CoordinatorJoin'); + expect(events).toHaveLength(2); + expect('source' in events[0]).toBe(false); + expect('source' in events[1]).toBe(false); + }); + + // the regression that matters: a reconnect must not inherit the ring source + // of the join that came before it + it('does not carry a ring source into a later reconnect', async () => { + await reporter.withJoinLifecycle( + cid, + { joinReason: 'first-attempt', joinSource: 'ring-ws' }, + () => reporter.track(cid, 'CoordinatorJoin', () => Promise.resolve('ok')), + ); + doAxiosRequest.mockClear(); + + await reporter.withJoinLifecycle(cid, { joinReason: 'full-rejoin' }, () => + reporter.track(cid, 'CoordinatorJoin', () => Promise.resolve('ok')), + ); + await flush(); + + const events = postedEvents().filter((e) => e.stage === 'CoordinatorJoin'); + expect(events).toHaveLength(2); + expect(events[0]).toMatchObject({ join_reason: 'full-rejoin' }); + expect('source' in events[0]).toBe(false); + expect('source' in events[1]).toBe(false); + }); + + // `Call.join`'s retry loop restores the pre-join state between attempts, so a + // reconnect can open its own lifecycle while the join's is still on the + // stack. The inner one must not inherit the outer one's source. + it('does not leak the joinSource into a nested sourceless lifecycle', async () => { + await reporter.withJoinLifecycle( + cid, + { joinReason: 'first-attempt', joinSource: 'ring-ws' }, + () => + reporter.withJoinLifecycle(cid, { joinReason: 'full-rejoin' }, () => + reporter.track(cid, 'CoordinatorJoin', () => Promise.resolve('ok')), + ), + ); + await flush(); + + const events = postedEvents().filter((e) => e.stage === 'CoordinatorJoin'); + expect(events).toHaveLength(2); + expect(events[0]).toMatchObject({ join_reason: 'full-rejoin' }); + expect('source' in events[0]).toBe(false); + expect('source' in events[1]).toBe(false); + }); + + // a fast reconnect reports CoordinatorJoin without a join lifecycle, so the + // source has to be gone by the time the lifecycle ends + it('does not carry the joinSource into a join reported outside the lifecycle', async () => { + await reporter.withJoinLifecycle( + cid, + { joinReason: 'first-attempt', joinSource: 'ring-ws' }, + () => reporter.track(cid, 'CoordinatorJoin', () => Promise.resolve('ok')), + ); + doAxiosRequest.mockClear(); + + await reporter.track(cid, 'CoordinatorJoin', () => Promise.resolve('ok')); + await flush(); + + const events = postedEvents().filter((e) => e.stage === 'CoordinatorJoin'); + expect(events).toHaveLength(2); + expect('source' in events[0]).toBe(false); + expect('source' in events[1]).toBe(false); + }); + + // `Call.join` re-correlates mid-loop when it forces an SFU switch; those + // later attempts are still the same ring-caused join + it('keeps the joinSource across a mid-lifecycle re-correlation', async () => { + await reporter.withJoinLifecycle( + cid, + { joinReason: 'first-attempt', joinSource: 'ring-ws' }, + async () => { + await expect( + reporter.track(cid, 'CoordinatorJoin', () => + Promise.reject(new Error('boom')), + ), + ).rejects.toThrow('boom'); + reporter.startCorrelation(cid, 'first-attempt'); + await reporter.track(cid, 'CoordinatorJoin', () => + Promise.resolve('ok'), + ); + }, + ); + await flush(); + + const events = postedEvents().filter((e) => e.stage === 'CoordinatorJoin'); + expect(events).toHaveLength(4); + expect(events.every((e) => e.source === 'ring-ws')).toBe(true); + expect(events[0].join_attempt_id).not.toBe(events[3].join_attempt_id); + }); + + it('drops the joinSource when the call is unregistered', async () => { + await reporter.withJoinLifecycle( + cid, + { joinReason: 'first-attempt', joinSource: 'ring-ws' }, + () => Promise.resolve('ok'), + ); + reporter.unregisterCall(cid); + reporter.registerCall(cid, callContext); + doAxiosRequest.mockClear(); + + reporter.startCorrelation(cid, 'first-attempt'); + await reporter.track(cid, 'CoordinatorJoin', () => Promise.resolve('ok')); + await flush(); + + const events = postedEvents().filter((e) => e.stage === 'CoordinatorJoin'); + expect(events).toHaveLength(2); + expect('source' in events[0]).toBe(false); + }); + + it('reports the source only on CoordinatorJoin', async () => { + await reporter.withJoinLifecycle( + cid, + { joinReason: 'first-attempt', joinSource: 'ring-ws' }, + () => reporter.track(cid, 'WSJoin', () => Promise.resolve('ok')), + ); + await flush(); + + const others = postedEvents().filter((e) => e.stage !== 'CoordinatorJoin'); + expect(others.length).toBeGreaterThan(0); + expect(others.some((e) => 'source' in e)).toBe(false); + }); + it('folds in-stage retries into a single pair', async () => { reporter.startCorrelation(cid, 'first-attempt'); await expect( diff --git a/packages/client/src/ringing/RingStatePoller.ts b/packages/client/src/ringing/RingStatePoller.ts new file mode 100644 index 0000000000..d6f6d9102a --- /dev/null +++ b/packages/client/src/ringing/RingStatePoller.ts @@ -0,0 +1,177 @@ +import type { Call } from '../Call'; +import { reconcileRingState } from './reconcileRingState'; +import { CallingState } from '../store'; +import { createSubscription } from '../store/rxUtils'; +import { getTimers } from '../timers'; +import { + ErrorFromResponse, + type RingStatePollingOptions, +} from '../coordinator/connection/types'; +import { videoLoggerSystem } from '../logger'; + +// `doJoin` restores the previous calling state when a join fails, so JOINING can +// go back to RINGING. Only a state the ring cannot return from ends the poller. +const ringIsOver = (callingState: CallingState) => + callingState !== CallingState.RINGING && + callingState !== CallingState.JOINING; + +/** + * Polls the coordinator for the outcome of a ring the current user started. + * + * `call.accepted`, `call.rejected` and `call.missed` are delivered best-effort, + * with no store-and-forward, so a caller that drops one is left on a ringing + * screen while the callee is already in the call. After a quiet period this + * reads the ring state until a terminal outcome or the end of the ring window. + */ +export class RingStatePoller { + private readonly logger = videoLoggerSystem.getLogger('RingStatePoller'); + private readonly call: Call; + private readonly startAfterMs: number; + private readonly intervalMs: number; + private sessionId: string | undefined; + private deadlineAt: number = 0; + private idleTimeoutId: number | undefined; + private intervalId: number | undefined; + private stopped: boolean = false; + private wasPolling: boolean = false; + private inFlight: boolean = false; + private unsubscribe: Array<() => void> = []; + + constructor(call: Call, options: RingStatePollingOptions = {}) { + this.call = call; + this.startAfterMs = options.startAfterMs ?? 15_000; + this.intervalMs = options.intervalMs ?? 5_000; + } + + /** + * Starts polling. Does nothing when the call has no session yet. + */ + start = () => { + if (this.stopped || this.sessionId) return; + if (this.call.state.callingState !== CallingState.RINGING) return; + + // captured once: `call.ended` clears the call's current session + const sessionId = this.call.state.session?.id; + if (!sessionId) { + this.logger.warn('the call has no session'); + return; + } + this.sessionId = sessionId; + + const ring = this.call.state.settings?.ring; + const maxDurationMs = + ring?.auto_cancel_timeout_ms || ring?.missed_call_timeout_ms || 30_000; + this.deadlineAt = Date.now() + maxDurationMs; + + // an incoming event proves the socket is alive, so the quiet period starts + // over: in a group ring a single rejection does not settle the ring. + this.unsubscribe.push( + this.call.on('call.accepted', () => this.armIdleWindow()), + this.call.on('call.rejected', () => this.armIdleWindow()), + this.call.on('call.missed', () => this.armIdleWindow()), + // the ring is over, whichever way it went. Joining an accepted call + // never goes through `leave`, so this is the only signal for it. + createSubscription(this.call.state.callingState$, (callingState) => { + if (ringIsOver(callingState)) this.stop(); + }), + ); + + this.armIdleWindow(); + }; + + /** + * Pauses polling, keeping the captured session, the deadline and the event + * subscriptions. Resuming does not extend the ring window. + */ + pause = () => { + if (this.stopped) return; + this.wasPolling = this.intervalId !== undefined; + const timers = getTimers(); + timers.clearTimeout(this.idleTimeoutId); + timers.clearInterval(this.intervalId); + this.idleTimeoutId = undefined; + this.intervalId = undefined; + }; + + /** + * Resumes a paused poller. Goes straight back to polling if it was already + * past the quiet period, so a pause does not buy the ring another one. + */ + resume = () => { + if (this.stopped || !this.sessionId) return; + if (this.idleTimeoutId !== undefined || this.intervalId !== undefined) { + return; + } + if (!this.wasPolling) { + this.armIdleWindow(); + return; + } + this.intervalId = getTimers().setInterval(this.runTick, this.intervalMs); + this.runTick(); + }; + + /** + * Stops polling. The poller cannot be restarted. + */ + stop = () => { + if (this.stopped) return; + this.stopped = true; + const timers = getTimers(); + timers.clearTimeout(this.idleTimeoutId); + timers.clearInterval(this.intervalId); + this.idleTimeoutId = undefined; + this.intervalId = undefined; + this.unsubscribe.forEach((off) => off()); + this.unsubscribe = []; + }; + + private armIdleWindow = () => { + if (this.stopped) return; + const timers = getTimers(); + timers.clearTimeout(this.idleTimeoutId); + timers.clearInterval(this.intervalId); + this.intervalId = undefined; + this.idleTimeoutId = timers.setTimeout(() => { + this.idleTimeoutId = undefined; + if (this.stopped) return; + this.intervalId = timers.setInterval(this.runTick, this.intervalMs); + this.runTick(); + }, this.startAfterMs); + }; + + private runTick = () => { + this.tick().catch((err) => { + this.logger.warn('Failed to poll the ring state', err); + }); + }; + + private tick = async () => { + if (this.stopped || this.inFlight || !this.sessionId) return; + const { callingState } = this.call.state; + if (ringIsOver(callingState) || Date.now() >= this.deadlineAt) { + this.stop(); + return; + } + // mid-join: nothing to reconcile until we know whether it succeeded + if (callingState !== CallingState.RINGING) return; + + this.inFlight = true; + try { + const ringState = await this.call.getRingState(this.sessionId); + if (this.stopped) return; + this.call.state.updateFromRingState(ringState); + if (await reconcileRingState(this.call, 'ring-poll-api')) this.stop(); + } catch (err) { + // a missing session, or one of another call, will never resolve + const status = err instanceof ErrorFromResponse ? err.status : undefined; + if (status === 400 || status === 404) { + this.logger.warn('Stopped polling the ring state', err); + this.stop(); + } else { + this.logger.debug('Failed to poll the ring state', err); + } + } finally { + this.inFlight = false; + } + }; +} diff --git a/packages/client/src/ringing/RingTimeout.ts b/packages/client/src/ringing/RingTimeout.ts new file mode 100644 index 0000000000..06870b9bbe --- /dev/null +++ b/packages/client/src/ringing/RingTimeout.ts @@ -0,0 +1,90 @@ +import type { Call } from '../Call'; +import { CallingState } from '../store'; +import { getTimers } from '../timers'; +import { videoLoggerSystem } from '../logger'; + +/** + * Drops a call that has been ringing for too long: the caller cancels after + * `auto_cancel_timeout_ms`, a callee after `incoming_call_timeout_ms`. + * + * Either timeout being `0` means the call rings until something else settles it. + */ +export class RingTimeout { + private readonly call: Call; + private timeoutId: number | undefined; + private deadlineAt: number | undefined; + private stopped: boolean = false; + + constructor(call: Call) { + this.call = call; + } + + /** + * Schedules an auto-drop timeout based on the call settings. + * Applicable only for ringing calls. + */ + start = () => { + if (this.stopped || this.timeoutId !== undefined) return; + // ignore if the call is not ringing + if (this.call.state.callingState !== CallingState.RINGING) return; + + const ring = this.call.state.settings?.ring; + if (!ring) return; + + const isCaller = this.call.isCreatedByMe; + const timeoutMs = isCaller + ? ring.auto_cancel_timeout_ms + : ring.incoming_call_timeout_ms; + // 0 means no auto-drop + if (timeoutMs <= 0) return; + + const now = Date.now(); + this.deadlineAt ??= now + timeoutMs; + const delayMs = Math.max(0, this.deadlineAt - now); + const timers = getTimers(); + this.timeoutId = timers.setTimeout(() => { + this.timeoutId = undefined; + // A failed timeout-triggered leave can arm a fresh retry window. + this.deadlineAt = undefined; + if (this.stopped) return; + // the call might have stopped ringing by this point, e.g. it was already + // accepted and joined + if (this.call.state.callingState !== CallingState.RINGING) return; + this.call + .leave({ + reject: true, + reason: 'timeout', + message: `ringing timeout - ${ + isCaller + ? 'no one accepted' + : `user didn't interact with incoming call screen` + }`, + }) + .catch((err) => { + videoLoggerSystem + .getLogger('RingTimeout') + .error('Failed to drop the call', err); + }); + }, delayMs); + }; + + /** + * Pauses the timeout while preserving its original deadline. + */ + pause = () => { + if (this.stopped || this.timeoutId === undefined) return; + getTimers().clearTimeout(this.timeoutId); + this.timeoutId = undefined; + }; + + /** + * Cancels a scheduled auto-drop timeout. It cannot be armed again. + */ + stop = () => { + if (this.stopped) return; + this.stopped = true; + getTimers().clearTimeout(this.timeoutId); + this.timeoutId = undefined; + this.deadlineAt = undefined; + }; +} diff --git a/packages/client/src/ringing/__tests__/RingStatePoller.test.ts b/packages/client/src/ringing/__tests__/RingStatePoller.test.ts new file mode 100644 index 0000000000..8aa6d86621 --- /dev/null +++ b/packages/client/src/ringing/__tests__/RingStatePoller.test.ts @@ -0,0 +1,462 @@ +import '../../rtc/__tests__/mocks/webrtc.mocks'; + +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { fromPartial } from '@total-typescript/shoehorn'; +import { Call } from '../../Call'; +import { RingStatePoller } from '../RingStatePoller'; +import { StreamClient } from '../../coordinator/connection/client'; +import { ClientEventReporter } from '../../reporting'; +import { generateUUIDv4 } from '../../coordinator/connection/utils'; +import { CallingState, ClientState } from '../../store'; +import { + ErrorFromResponse, + type StreamClientOptions, +} from '../../coordinator/connection/types'; +import type { GetCallRingStateResponse } from '../../gen/coordinator'; + +const SESSION_ID = 'session-1'; +const START_AFTER_MS = 15_000; +const INTERVAL_MS = 5_000; + +const ringState = ( + overrides: Partial = {}, +): GetCallRingStateResponse => ({ + duration: '1ms', + call_cid: 'test:call', + session_id: SESSION_ID, + created_by_user_id: 'jane', + accepted_by: {}, + rejected_by: {}, + missed_by: {}, + ...overrides, +}); + +describe('RingStatePoller', () => { + const userId = 'jane'; + let call: Call; + let poller: RingStatePoller; + + const createCall = (options?: StreamClientOptions) => { + const clientState = new ClientState(); + const streamClient = new StreamClient('abc', options); + const newCall = new Call({ + type: 'test', + id: generateUUIDv4(), + streamClient, + clientEventReporter: new ClientEventReporter({ streamClient }), + clientState, + }); + + // @ts-expect-error mocking only what we need for the test + clientState.setConnectedUser({ id: userId }); + // @ts-expect-error mocking only what we need for the test + newCall.state['createdBySubject'].next({ id: userId }); + // @ts-expect-error mocking only what we need for the test + newCall.state['sessionSubject'].next({ + id: SESSION_ID, + accepted_by: {}, + rejected_by: {}, + missed_by: {}, + participants: [], + }); + newCall.state['settingsSubject'].next({ + ring: { + auto_cancel_timeout_ms: 30_000, + incoming_call_timeout_ms: 30_000, + missed_call_timeout_ms: 30_000, + }, + // @ts-expect-error mocking only what we need for the test + screensharing: { + enabled: false, + target_resolution: { width: 100, height: 100 }, + }, + }); + newCall.state.setMembers( + fromPartial([{ user_id: userId }, { user_id: 'john' }]), + ); + newCall.state['callingStateSubject'].next(CallingState.RINGING); + + vi.spyOn(newCall, 'join').mockResolvedValue(undefined); + vi.spyOn(newCall, 'leave').mockResolvedValue(undefined); + + return newCall; + }; + + const startPolling = (response = ringState()) => { + const getRingState = vi + .spyOn(call, 'getRingState') + .mockResolvedValue(response); + poller = new RingStatePoller(call); + poller.start(); + return getRingState; + }; + + beforeEach(() => { + vi.useFakeTimers(); + call = createCall(); + }); + + afterEach(() => { + poller?.stop(); + vi.useRealTimers(); + vi.restoreAllMocks(); + }); + + it('does not poll before the quiet period elapses', async () => { + const getRingState = startPolling(); + + await vi.advanceTimersByTimeAsync(START_AFTER_MS - 1); + expect(getRingState).not.toHaveBeenCalled(); + + await vi.advanceTimersByTimeAsync(1); + expect(getRingState).toHaveBeenCalledWith(SESSION_ID); + }); + + it('keeps polling on the configured interval while the ring is pending', async () => { + const getRingState = startPolling(); + + await vi.advanceTimersByTimeAsync(START_AFTER_MS); + expect(getRingState).toHaveBeenCalledTimes(1); + + await vi.advanceTimersByTimeAsync(INTERVAL_MS); + expect(getRingState).toHaveBeenCalledTimes(2); + + await vi.advanceTimersByTimeAsync(INTERVAL_MS); + expect(getRingState).toHaveBeenCalledTimes(3); + }); + + it('joins the call once when someone else accepted', async () => { + const getRingState = startPolling( + ringState({ accepted_by: { john: '2026-08-24T10:00:04Z' } }), + ); + + await vi.advanceTimersByTimeAsync(START_AFTER_MS + 3 * INTERVAL_MS); + + expect(call.join).toHaveBeenCalledTimes(1); + // pins this as the poller's call site, not the WebSocket handlers' + expect(call.join).toHaveBeenCalledWith({ joinSource: 'ring-poll-api' }); + expect(getRingState).toHaveBeenCalledTimes(1); + }); + + it('ignores an acceptance by the current user', async () => { + startPolling( + ringState({ accepted_by: { [userId]: '2026-08-24T10:00:04Z' } }), + ); + + await vi.advanceTimersByTimeAsync(START_AFTER_MS); + + expect(call.join).not.toHaveBeenCalled(); + }); + + it('cancels the call when everyone else rejected', async () => { + startPolling(ringState({ rejected_by: { john: '2026-08-24T10:00:09Z' } })); + + await vi.advanceTimersByTimeAsync(START_AFTER_MS); + + expect(call.leave).toHaveBeenCalledWith({ + reject: true, + reason: 'cancel', + message: 'ring: everyone rejected', + }); + }); + + it('keeps the ring deadline across a pause', async () => { + startPolling(); + const deadlineAt = poller['deadlineAt']; + + poller.pause(); + await vi.advanceTimersByTimeAsync(10_000); + poller.resume(); + + expect(poller['deadlineAt']).toBe(deadlineAt); + }); + + it('resumes straight back into polling when it was already past the quiet period', async () => { + const getRingState = startPolling(); + + await vi.advanceTimersByTimeAsync(START_AFTER_MS); + expect(getRingState).toHaveBeenCalledTimes(1); + + poller.pause(); + await vi.advanceTimersByTimeAsync(INTERVAL_MS); + expect(getRingState).toHaveBeenCalledTimes(1); + + // no second quiet period: the next poll lands on the interval + poller.resume(); + expect(getRingState).toHaveBeenCalledTimes(2); + await vi.advanceTimersByTimeAsync(INTERVAL_MS); + expect(getRingState).toHaveBeenCalledTimes(3); + }); + + it('stops on the first tick after the ring window, even across a pause', async () => { + const getRingState = startPolling(); + + await vi.advanceTimersByTimeAsync(START_AFTER_MS); + poller.pause(); + // the pause outlasts the ring window + await vi.advanceTimersByTimeAsync(30_000); + poller.resume(); + + expect(getRingState).toHaveBeenCalledTimes(1); + expect(poller['stopped']).toBe(true); + }); + + it('does not resume a stopped poller', async () => { + const getRingState = startPolling(); + await vi.advanceTimersByTimeAsync(START_AFTER_MS); + // paused mid-poll, so resuming would take the interval branch + poller.pause(); + poller.stop(); + + poller.resume(); + + // no timer left behind on a stopped poller + expect(poller['intervalId']).toBeUndefined(); + expect(poller['idleTimeoutId']).toBeUndefined(); + await vi.advanceTimersByTimeAsync(START_AFTER_MS + INTERVAL_MS); + expect(getRingState).toHaveBeenCalledTimes(1); + }); + + it('keeps polling when everyone else is marked missed before auto-cancel', async () => { + startPolling(ringState({ missed_by: { john: '2026-08-24T10:00:35Z' } })); + + await vi.advanceTimersByTimeAsync(START_AFTER_MS); + + expect(call.leave).not.toHaveBeenCalled(); + expect(poller['stopped']).toBe(false); + }); + + it('leaves without rejecting when the call has ended, even if it was accepted', async () => { + startPolling( + ringState({ + accepted_by: { john: '2026-08-24T10:00:04Z' }, + call_ended_at: '2026-08-24T10:00:20Z', + }), + ); + + await vi.advanceTimersByTimeAsync(START_AFTER_MS); + + expect(call.join).not.toHaveBeenCalled(); + expect(call.leave).toHaveBeenCalledWith({ + reject: false, + reason: 'ended', + message: 'ring: call ended', + }); + }); + + it('restarts the quiet period when a ring event arrives over the WebSocket', async () => { + const getRingState = startPolling(); + + await vi.advanceTimersByTimeAsync(START_AFTER_MS - 1_000); + call['streamClient'].dispatchEvent({ + type: 'call.rejected', + call_cid: call.cid, + // @ts-expect-error mocking only what we need for the test + user: { id: 'john' }, + }); + + await vi.advanceTimersByTimeAsync(1_000); + expect(getRingState).not.toHaveBeenCalled(); + + await vi.advanceTimersByTimeAsync(START_AFTER_MS - 1_000); + expect(getRingState).toHaveBeenCalledTimes(1); + }); + + it('stops polling once the call has left the ringing state', async () => { + const getRingState = startPolling(); + + await vi.advanceTimersByTimeAsync(START_AFTER_MS); + expect(getRingState).toHaveBeenCalledTimes(1); + + call.state['callingStateSubject'].next(CallingState.JOINED); + await vi.advanceTimersByTimeAsync(3 * INTERVAL_MS); + expect(getRingState).toHaveBeenCalledTimes(1); + }); + + it('stops polling as soon as the caller joins, without waiting for a tick', async () => { + const getRingState = startPolling(); + + // the caller joining an accepted call never goes through `leave` + call.state['callingStateSubject'].next(CallingState.JOINED); + expect(poller['stopped']).toBe(true); + + await vi.advanceTimersByTimeAsync(START_AFTER_MS + 3 * INTERVAL_MS); + expect(getRingState).not.toHaveBeenCalled(); + }); + + // `doJoin` restores the ringing state when a join fails, so JOINING is + // transient: giving up on it would abandon the ring the poller exists for. + it('keeps polling when a join attempt failed and left the call ringing', async () => { + const getRingState = startPolling(); + + call.state['callingStateSubject'].next(CallingState.JOINING); + expect(poller['stopped']).toBe(false); + call.state['callingStateSubject'].next(CallingState.RINGING); + + await vi.advanceTimersByTimeAsync(START_AFTER_MS); + expect(getRingState).toHaveBeenCalled(); + }); + + it('retries the join on the next poll when it failed', async () => { + const getRingState = startPolling( + ringState({ accepted_by: { john: '2026-08-24T10:00:04Z' } }), + ); + vi.mocked(call.join).mockRejectedValueOnce(new Error('transient')); + + await vi.advanceTimersByTimeAsync(START_AFTER_MS); + expect(call.join).toHaveBeenCalledTimes(1); + expect(poller['stopped']).toBe(false); + + await vi.advanceTimersByTimeAsync(INTERVAL_MS); + expect(call.join).toHaveBeenCalledTimes(2); + expect(getRingState).toHaveBeenCalledTimes(2); + }); + + // `Call.join({ ring: true })` arms the poller while the call is already + // JOINING, so `start` has to refuse rather than arm and stop on the next tick. + it('does not arm when the call is not ringing', async () => { + call.state['callingStateSubject'].next(CallingState.JOINING); + const getRingState = startPolling(); + + expect(poller['sessionId']).toBeUndefined(); + await vi.advanceTimersByTimeAsync(START_AFTER_MS + INTERVAL_MS); + expect(getRingState).not.toHaveBeenCalled(); + }); + + it('stops polling once the ring window has closed', async () => { + const getRingState = startPolling(); + + await vi.advanceTimersByTimeAsync(30_000); + const callsWithinWindow = getRingState.mock.calls.length; + expect(callsWithinWindow).toBeGreaterThan(0); + + await vi.advanceTimersByTimeAsync(3 * INTERVAL_MS); + expect(getRingState).toHaveBeenCalledTimes(callsWithinWindow); + }); + + it('stops polling on an unrecoverable response, but not on a transient one', async () => { + const getRingState = vi.spyOn(call, 'getRingState'); + const error = (status: number) => + new ErrorFromResponse({ + message: 'boom', + code: 16, + status, + // @ts-expect-error mocking only what we need for the test + response: {}, + unrecoverable: false, + }); + + getRingState.mockRejectedValueOnce(error(500)); + poller = new RingStatePoller(call); + poller.start(); + + await vi.advanceTimersByTimeAsync(START_AFTER_MS); + expect(getRingState).toHaveBeenCalledTimes(1); + + getRingState.mockRejectedValueOnce(error(404)); + await vi.advanceTimersByTimeAsync(INTERVAL_MS); + expect(getRingState).toHaveBeenCalledTimes(2); + + await vi.advanceTimersByTimeAsync(3 * INTERVAL_MS); + expect(getRingState).toHaveBeenCalledTimes(2); + }); + + it('does not start without a call session', async () => { + call.state['sessionSubject'].next(undefined); + const getRingState = startPolling(); + + await vi.advanceTimersByTimeAsync(START_AFTER_MS + INTERVAL_MS); + expect(getRingState).not.toHaveBeenCalled(); + }); + + it('honors the configured timings', async () => { + const getRingState = vi + .spyOn(call, 'getRingState') + .mockResolvedValue(ringState()); + poller = new RingStatePoller(call, { + startAfterMs: 1_000, + intervalMs: 500, + }); + poller.start(); + + await vi.advanceTimersByTimeAsync(1_000); + expect(getRingState).toHaveBeenCalledTimes(1); + + await vi.advanceTimersByTimeAsync(500); + expect(getRingState).toHaveBeenCalledTimes(2); + }); +}); + +describe('Call ring state polling', () => { + const userId = 'jane'; + + const createRingingCall = (options?: StreamClientOptions) => { + const clientState = new ClientState(); + const streamClient = new StreamClient('abc', options); + const call = new Call({ + type: 'test', + id: generateUUIDv4(), + streamClient, + clientEventReporter: new ClientEventReporter({ streamClient }), + clientState, + ringing: true, + }); + + // @ts-expect-error mocking only what we need for the test + clientState.setConnectedUser({ id: userId }); + // @ts-expect-error mocking only what we need for the test + call.state['sessionSubject'].next({ + id: SESSION_ID, + accepted_by: {}, + rejected_by: {}, + missed_by: {}, + participants: [], + }); + call.state['settingsSubject'].next({ + // @ts-expect-error mocking only what we need for the test + ring: { auto_cancel_timeout_ms: 30_000 }, + // @ts-expect-error mocking only what we need for the test + screensharing: { + enabled: false, + target_resolution: { width: 100, height: 100 }, + }, + }); + return call; + }; + + beforeEach(() => { + vi.useFakeTimers(); + }); + + afterEach(() => { + vi.useRealTimers(); + vi.restoreAllMocks(); + }); + + it('polls for the caller', () => { + const call = createRingingCall(); + // @ts-expect-error mocking only what we need for the test + call.state['createdBySubject'].next({ id: userId }); + + call['handleRingingCall'](); + expect(call['ringStatePoller']).toBeDefined(); + }); + + it('does not poll for the callee', () => { + const call = createRingingCall(); + // @ts-expect-error mocking only what we need for the test + call.state['createdBySubject'].next({ id: 'not-' + userId }); + + call['handleRingingCall'](); + expect(call['ringStatePoller']).toBeUndefined(); + }); + + it('does not poll when disabled through the client options', () => { + const call = createRingingCall({ ringStatePolling: false }); + // @ts-expect-error mocking only what we need for the test + call.state['createdBySubject'].next({ id: userId }); + + call['handleRingingCall'](); + expect(call['ringStatePoller']).toBeUndefined(); + }); +}); diff --git a/packages/client/src/ringing/__tests__/RingTimeout.test.ts b/packages/client/src/ringing/__tests__/RingTimeout.test.ts new file mode 100644 index 0000000000..01eb556e88 --- /dev/null +++ b/packages/client/src/ringing/__tests__/RingTimeout.test.ts @@ -0,0 +1,199 @@ +import '../../rtc/__tests__/mocks/webrtc.mocks'; + +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { fromPartial } from '@total-typescript/shoehorn'; +import { Call } from '../../Call'; +import { RingTimeout } from '../RingTimeout'; +import { StreamClient } from '../../coordinator/connection/client'; +import { ClientEventReporter } from '../../reporting'; +import { generateUUIDv4 } from '../../coordinator/connection/utils'; +import { CallingState, ClientState } from '../../store'; +import { CallSettingsResponse } from '../../gen/coordinator'; + +const TIMEOUT_MS = 30_000; + +describe('RingTimeout', () => { + const userId = 'jane'; + let call: Call; + let ringTimeout: RingTimeout; + + const ringingCall = ({ + createdById = userId, + ring = { + auto_cancel_timeout_ms: TIMEOUT_MS, + incoming_call_timeout_ms: TIMEOUT_MS, + missed_call_timeout_ms: TIMEOUT_MS, + }, + }: { + createdById?: string; + ring?: Partial | null; + } = {}) => { + const clientState = new ClientState(); + const streamClient = new StreamClient('abc'); + const newCall = new Call({ + type: 'test', + id: generateUUIDv4(), + streamClient, + clientEventReporter: new ClientEventReporter({ streamClient }), + clientState, + }); + + clientState.setConnectedUser(fromPartial({ id: userId })); + newCall.state['createdBySubject'].next(fromPartial({ id: createdById })); + // leaving `settings` unset is how a call built from a push notification + // looks until `get()` resolves + if (ring) { + newCall.state['settingsSubject'].next( + fromPartial({ + ring, + screensharing: { + enabled: false, + target_resolution: { width: 100, height: 100 }, + }, + }), + ); + } + newCall.state['callingStateSubject'].next(CallingState.RINGING); + + vi.spyOn(newCall, 'leave').mockResolvedValue(undefined); + return newCall; + }; + + const arm = () => { + ringTimeout = new RingTimeout(call); + ringTimeout.start(); + return ringTimeout; + }; + + beforeEach(() => { + vi.useFakeTimers(); + }); + + afterEach(() => { + ringTimeout?.stop(); + vi.useRealTimers(); + vi.restoreAllMocks(); + }); + + it('drops the call for the caller when no one accepted', async () => { + call = ringingCall(); + arm(); + + await vi.advanceTimersByTimeAsync(TIMEOUT_MS); + + expect(call.leave).toHaveBeenCalledWith({ + reject: true, + reason: 'timeout', + message: 'ringing timeout - no one accepted', + }); + }); + + it('drops the call for a callee that never interacted', async () => { + call = ringingCall({ createdById: 'not-' + userId }); + arm(); + + await vi.advanceTimersByTimeAsync(TIMEOUT_MS); + + expect(call.leave).toHaveBeenCalledWith({ + reject: true, + reason: 'timeout', + message: `ringing timeout - user didn't interact with incoming call screen`, + }); + }); + + it('does not drop the call before the timeout elapses', async () => { + call = ringingCall(); + arm(); + + await vi.advanceTimersByTimeAsync(TIMEOUT_MS - 1); + expect(call.leave).not.toHaveBeenCalled(); + + await vi.advanceTimersByTimeAsync(1); + expect(call.leave).toHaveBeenCalled(); + }); + + it('does not drop a call that stopped ringing before the deadline', async () => { + call = ringingCall(); + arm(); + + call.state['callingStateSubject'].next(CallingState.JOINED); + + await vi.advanceTimersByTimeAsync(TIMEOUT_MS); + expect(call.leave).not.toHaveBeenCalled(); + }); + + // `doJoin` restores the ringing state when a join fails, so a transition to + // JOINING must not disarm the drop for good, or the call rings forever. + it('still drops the call after a join attempt failed and left it ringing', async () => { + call = ringingCall(); + arm(); + + call.state['callingStateSubject'].next(CallingState.JOINING); + call.state['callingStateSubject'].next(CallingState.RINGING); + + await vi.advanceTimersByTimeAsync(TIMEOUT_MS); + expect(call.leave).toHaveBeenCalledWith({ + reject: true, + reason: 'timeout', + message: 'ringing timeout - no one accepted', + }); + }); + + it('does not drop the call once stopped', async () => { + call = ringingCall(); + arm().stop(); + + await vi.advanceTimersByTimeAsync(TIMEOUT_MS); + expect(call.leave).not.toHaveBeenCalled(); + }); + + it('resumes from the original deadline after being paused', async () => { + call = ringingCall(); + arm(); + + await vi.advanceTimersByTimeAsync(10_000); + ringTimeout.pause(); + await vi.advanceTimersByTimeAsync(5_000); + ringTimeout.start(); + + await vi.advanceTimersByTimeAsync(14_999); + expect(call.leave).not.toHaveBeenCalled(); + + await vi.advanceTimersByTimeAsync(1); + expect(call.leave).toHaveBeenCalled(); + }); + + it('does not arm when the call is not ringing', async () => { + call = ringingCall(); + call.state['callingStateSubject'].next(CallingState.JOINED); + arm(); + + expect(ringTimeout['timeoutId']).toBeUndefined(); + await vi.advanceTimersByTimeAsync(TIMEOUT_MS); + expect(call.leave).not.toHaveBeenCalled(); + }); + + it('does not arm before the call settings have loaded', async () => { + call = ringingCall({ ring: null }); + arm(); + + expect(ringTimeout['timeoutId']).toBeUndefined(); + await vi.advanceTimersByTimeAsync(TIMEOUT_MS); + expect(call.leave).not.toHaveBeenCalled(); + }); + + it('treats a zero timeout as no auto-drop', async () => { + call = ringingCall({ + ring: { + auto_cancel_timeout_ms: 0, + incoming_call_timeout_ms: 0, + missed_call_timeout_ms: TIMEOUT_MS, + }, + }); + arm(); + + expect(ringTimeout['timeoutId']).toBeUndefined(); + await vi.advanceTimersByTimeAsync(TIMEOUT_MS); + expect(call.leave).not.toHaveBeenCalled(); + }); +}); diff --git a/packages/client/src/ringing/__tests__/reconcileRingState.test.ts b/packages/client/src/ringing/__tests__/reconcileRingState.test.ts new file mode 100644 index 0000000000..79f368639f --- /dev/null +++ b/packages/client/src/ringing/__tests__/reconcileRingState.test.ts @@ -0,0 +1,330 @@ +import { describe, expect, it, vi } from 'vitest'; +import { fromPartial } from '@total-typescript/shoehorn'; +import { reconcileRingState } from '../reconcileRingState'; +import { CallingState, ClientState } from '../../store'; +import { + CallResponse, + CallSessionResponse, + MemberResponse, +} from '../../gen/coordinator'; +import { Call } from '../../Call'; +import { StreamClient } from '../../coordinator/connection/client'; +import type { JoinSource } from '../../reporting'; +import { ClientEventReporter } from '../../reporting'; +import { settled } from '../../helpers/concurrency'; + +describe('reconcileRingState', () => { + describe('acceptance', () => { + it('ignores an acceptance by the current user', async () => { + const call = ringingCall({ currentUserId: 'm1', createdById: 'm1' }); + setSession(call, { accepted_by: { m1: timestamp() } }); + + expect(await reconcile(call)).toBe(false); + expect(call.join).not.toHaveBeenCalled(); + }); + + it('joins the call for the caller once a callee has accepted', async () => { + const call = ringingCall({ currentUserId: 'm1', createdById: 'm1' }); + setSession(call, { accepted_by: { m2: timestamp() } }); + + expect(await reconcile(call)).toBe(true); + expect(call.join).toHaveBeenCalledWith({ joinSource: 'ring-ws' }); + }); + + it('reports the poller as the source when the poll found the acceptance', async () => { + const call = ringingCall({ currentUserId: 'm1', createdById: 'm1' }); + setSession(call, { accepted_by: { m2: timestamp() } }); + + expect(await reconcile(call, 'ring-poll-api')).toBe(true); + expect(call.join).toHaveBeenCalledWith({ joinSource: 'ring-poll-api' }); + }); + + it('is not terminal when the join fails, so a retry can follow', async () => { + const call = ringingCall({ currentUserId: 'm1', createdById: 'm1' }); + setSession(call, { accepted_by: { m2: timestamp() } }); + vi.mocked(call.join).mockRejectedValueOnce(new Error('transient')); + + expect(await reconcile(call)).toBe(false); + expect(call.join).toHaveBeenCalled(); + expect(call.leave).not.toHaveBeenCalled(); + }); + + it('does not join another callee when someone else accepts', async () => { + const call = ringingCall({ currentUserId: 'm2', createdById: 'm0' }); + setSession(call, { accepted_by: { m1: timestamp() } }); + + expect(await reconcile(call)).toBe(false); + expect(call.join).not.toHaveBeenCalled(); + }); + }); + + describe('rejection', () => { + it('cancels the call once every callee has rejected', async () => { + const call = ringingCall({ + currentUserId: 'm1', + createdById: 'm1', + members: ['m1', 'm2', 'm3'], + }); + setSession(call, { + rejected_by: { m2: timestamp(), m3: timestamp() }, + }); + + expect(await reconcile(call)).toBe(true); + expect(call.leave).toHaveBeenCalledWith({ + reject: true, + reason: 'cancel', + message: 'ring: everyone rejected', + }); + }); + + it('is not terminal when leaving fails, so a retry can follow', async () => { + const call = ringingCall({ + currentUserId: 'm1', + createdById: 'm1', + members: ['m1', 'm2'], + }); + setSession(call, { rejected_by: { m2: timestamp() } }); + vi.mocked(call.leave).mockRejectedValueOnce(new Error('transient')); + + expect(await reconcile(call)).toBe(false); + expect(call.leave).toHaveBeenCalled(); + }); + + it('keeps ringing while only one callee has rejected', async () => { + const call = ringingCall({ + currentUserId: 'm0', + createdById: 'm0', + members: ['m0', 'm1', 'm2'], + }); + setSession(call, { rejected_by: { m2: timestamp() } }); + + expect(await reconcile(call)).toBe(false); + expect(call.leave).not.toHaveBeenCalled(); + }); + + it('leaves a callee once the caller has cancelled', async () => { + const call = ringingCall({ + currentUserId: 'm1', + createdById: 'm0', + members: ['m0', 'm1', 'm2'], + }); + setSession(call, { rejected_by: { m0: timestamp() } }); + + expect(await reconcile(call)).toBe(true); + expect(call.leave).toHaveBeenCalledWith({ + reason: 'ended', + message: 'ring: creator rejected', + }); + }); + + it('keeps a callee ringing while another callee rejects', async () => { + const call = ringingCall({ + currentUserId: 'm1', + createdById: 'm0', + members: ['m0', 'm1', 'm2'], + }); + setSession(call, { rejected_by: { m2: timestamp() } }); + + expect(await reconcile(call)).toBe(false); + expect(call.leave).not.toHaveBeenCalled(); + }); + }); + + describe('missed', () => { + it('keeps ringing when every callee is marked missed before auto-cancel', async () => { + const call = ringingCall({ + currentUserId: 'm1', + createdById: 'm1', + members: ['m1', 'm2', 'm3'], + }); + setSession(call, { + missed_by: { m2: timestamp(), m3: timestamp() }, + }); + + expect(await reconcile(call)).toBe(false); + expect(call.leave).not.toHaveBeenCalled(); + }); + + it('keeps ringing when callees are split between rejected and missed', async () => { + const call = ringingCall({ + currentUserId: 'm1', + createdById: 'm1', + members: ['m1', 'm2', 'm3'], + }); + setSession(call, { + rejected_by: { m2: timestamp() }, + missed_by: { m3: timestamp() }, + }); + + expect(await reconcile(call)).toBe(false); + expect(call.leave).not.toHaveBeenCalled(); + }); + + it('keeps ringing while one callee can still accept', async () => { + const call = ringingCall({ + currentUserId: 'm1', + createdById: 'm1', + members: ['m1', 'm2', 'm3'], + }); + setSession(call, { missed_by: { m2: timestamp() } }); + + expect(await reconcile(call)).toBe(false); + expect(call.leave).not.toHaveBeenCalled(); + }); + }); + + describe('ended call', () => { + it('leaves without rejecting, even when the call was accepted', async () => { + const call = ringingCall({ currentUserId: 'm1', createdById: 'm1' }); + setSession(call, { + accepted_by: { m2: timestamp() }, + ended_at: timestamp(), + }); + + expect(await reconcile(call)).toBe(true); + expect(call.join).not.toHaveBeenCalled(); + expect(call.leave).toHaveBeenCalledWith({ + reject: false, + reason: 'ended', + message: 'ring: call ended', + }); + }); + + it('leaves when the call itself has ended', async () => { + const call = ringingCall({ currentUserId: 'm1', createdById: 'm1' }); + setSession(call, {}); + call.state.setEndedAt(new Date()); + + expect(await reconcile(call)).toBe(true); + expect(call.leave).toHaveBeenCalledWith({ + reject: false, + reason: 'ended', + message: 'ring: call ended', + }); + }); + }); + + it('is terminal once the call is no longer ringing', async () => { + const call = ringingCall({ currentUserId: 'm1', createdById: 'm1' }); + setSession(call, { accepted_by: { m2: timestamp() } }); + call.state.setCallingState(CallingState.JOINED); + + expect(await reconcile(call)).toBe(true); + expect(call.join).not.toHaveBeenCalled(); + expect(call.leave).not.toHaveBeenCalled(); + }); + + // Pins the ordering the state-driven reconciler depends on: `Call.setup` + // registers `updateFromEvent` as an `all` listener, and `dispatchEvent` + // drains those before the typed ring handlers. Reconciling from state would + // silently read the previous session if either side changed. + it('sees the event data on the state by the time a ring handler runs', async () => { + const call = ringingCall({ currentUserId: 'm1', createdById: 'm1' }); + await call.setup(); + call.state.setCallingState(CallingState.RINGING); + + call.streamClient.dispatchEvent( + fromPartial({ + type: 'call.accepted', + call_cid: call.cid, + created_at: new Date().toISOString(), + user: { id: 'm2' }, + call: { + ...callResponse('m1'), + session: { + id: 'session-1', + accepted_by: { m2: timestamp() }, + rejected_by: {}, + missed_by: {}, + participants: [], + participants_count_by_role: {}, + }, + }, + }), + ); + await settled(call['joinLeaveConcurrencyTag']); + + expect(call.join).toHaveBeenCalledWith({ joinSource: 'ring-ws' }); + }); + + it('keeps ringing when the call has no members yet', async () => { + const call = ringingCall({ + currentUserId: 'm1', + createdById: 'm1', + members: [], + }); + setSession(call, {}); + + expect(await reconcile(call)).toBe(false); + expect(call.leave).not.toHaveBeenCalled(); + }); +}); + +// the source only matters for the caller's join, so default it and let the +// tests that assert on it pass their own +const reconcile = (call: Call, joinSource: JoinSource = 'ring-ws') => + reconcileRingState(call, joinSource); + +const timestamp = () => new Date().toISOString(); + +const callResponse = (createdById: string) => + fromPartial({ + id: '12345', + type: 'development', + cid: 'development:12345', + created_by: { id: createdById }, + blocked_user_ids: [], + egress: {}, + settings: { + ring: { + auto_cancel_timeout_ms: 30_000, + incoming_call_timeout_ms: 30_000, + missed_call_timeout_ms: 30_000, + }, + screensharing: { target_resolution: undefined }, + }, + }); + +const setSession = (call: Call, session: Partial) => { + call.state['sessionSubject'].next( + fromPartial({ + accepted_by: {}, + rejected_by: {}, + missed_by: {}, + ...session, + }), + ); +}; + +const ringingCall = ({ + currentUserId, + createdById, + members = [currentUserId, 'm2'], +}: { + currentUserId: string; + createdById: string; + members?: string[]; +}) => { + const store = new ClientState(); + store.setConnectedUser(fromPartial({ id: currentUserId })); + const streamClient = new StreamClient('api-key'); + const call = new Call({ + type: 'development', + id: '12345', + clientState: store, + streamClient, + clientEventReporter: new ClientEventReporter({ streamClient }), + ringing: true, + }); + + call.state.updateFromCallResponse(callResponse(createdById)); + call.state.setMembers( + members.map((userId) => fromPartial({ user_id: userId })), + ); + call.state.setCallingState(CallingState.RINGING); + + vi.spyOn(call, 'join').mockResolvedValue(undefined); + vi.spyOn(call, 'leave').mockResolvedValue(undefined); + + return call; +}; diff --git a/packages/client/src/ringing/__tests__/resolveOwnRingOutcome.test.ts b/packages/client/src/ringing/__tests__/resolveOwnRingOutcome.test.ts new file mode 100644 index 0000000000..cb54fa9adf --- /dev/null +++ b/packages/client/src/ringing/__tests__/resolveOwnRingOutcome.test.ts @@ -0,0 +1,103 @@ +import { describe, expect, it } from 'vitest'; +import { fromPartial } from '@total-typescript/shoehorn'; +import { + type OwnRingOutcomeInput, + resolveOwnRingOutcome, +} from '../resolveOwnRingOutcome'; +import { CallingState } from '../../store'; +import { CallSessionResponse } from '../../gen/coordinator'; + +const ME = 'm1'; +const timestamp = () => new Date().toISOString(); + +describe('resolveOwnRingOutcome', () => { + const resolve = (overrides: Partial = {}) => + resolveOwnRingOutcome({ + currentUserId: ME, + callingState: CallingState.RINGING, + ...overrides, + session: fromPartial({ + accepted_by: {}, + rejected_by: {}, + missed_by: {}, + ...overrides.session, + }), + }); + + it('reports nothing while nobody has acted', () => { + expect(resolve()).toEqual({ settledByMe: false }); + }); + + it('reports nothing for another user accepting or rejecting', () => { + const outcome = resolve({ + session: fromPartial({ + accepted_by: { m2: timestamp() }, + rejected_by: { m3: timestamp() }, + }), + }); + + expect(outcome).toEqual({ settledByMe: false }); + }); + + it('leaves when the ring was answered on another device', () => { + const outcome = resolve({ + session: fromPartial({ accepted_by: { [ME]: timestamp() } }), + }); + + expect(outcome).toEqual({ + settledByMe: true, + leaveReason: 'answeredElsewhere', + }); + }); + + it('does not leave when this device is the one that accepted', () => { + const outcome = resolve({ + callingState: CallingState.JOINING, + session: fromPartial({ accepted_by: { [ME]: timestamp() } }), + }); + + // the drop no longer has to fire, but this device is joining, not leaving + expect(outcome).toEqual({ settledByMe: true, leaveReason: undefined }); + }); + + it('leaves when the current user rejected', () => { + const outcome = resolve({ + session: fromPartial({ rejected_by: { [ME]: timestamp() } }), + }); + + expect(outcome).toEqual({ settledByMe: true, leaveReason: 'rejected' }); + }); + + it('prefers the answered-elsewhere reason over a rejection', () => { + const outcome = resolve({ + session: fromPartial({ + accepted_by: { [ME]: timestamp() }, + rejected_by: { [ME]: timestamp() }, + }), + }); + + expect(outcome).toEqual({ + settledByMe: true, + leaveReason: 'answeredElsewhere', + }); + }); + + it('reports nothing without a connected user', () => { + const outcome = resolve({ + currentUserId: undefined, + session: fromPartial({ rejected_by: { [ME]: timestamp() } }), + }); + + expect(outcome).toEqual({ settledByMe: false }); + }); + + it('reports nothing without a session', () => { + expect( + resolveOwnRingOutcome({ + session: undefined, + currentUserId: ME, + callingState: CallingState.RINGING, + }), + ).toEqual({ settledByMe: false }); + }); +}); diff --git a/packages/client/src/ringing/index.ts b/packages/client/src/ringing/index.ts new file mode 100644 index 0000000000..5af2340ee5 --- /dev/null +++ b/packages/client/src/ringing/index.ts @@ -0,0 +1,4 @@ +export * from './RingStatePoller'; +export * from './RingTimeout'; +export * from './reconcileRingState'; +export * from './resolveOwnRingOutcome'; diff --git a/packages/client/src/ringing/reconcileRingState.ts b/packages/client/src/ringing/reconcileRingState.ts new file mode 100644 index 0000000000..b26b3d8841 --- /dev/null +++ b/packages/client/src/ringing/reconcileRingState.ts @@ -0,0 +1,108 @@ +import type { Call } from '../Call'; +import type { JoinSource } from '../reporting'; +import { CallingState } from '../store'; +import type { CallLeaveOptions } from '../types'; + +/** + * Decides what a ringing call should do next, based on the current call state. + * + * The `call.accepted`, `call.rejected` and `call.missed` handlers and the ring + * state poller both run this. They differ only in how the state got there: the + * handlers rely on `CallState.updateFromEvent`, which runs before them, and the + * poller applies the polled ring state itself. + * + * @param call the call to reconcile. + * @param joinSource which of the two triggered this run, reported on the + * caller's join: `ring-ws` for the event handlers, `ring-poll-api` for the + * poller. + * @returns whether the ring reached a terminal state. A failed join is not + * terminal: the caller should keep trying while the ring is open. + */ +export const reconcileRingState = async ( + call: Call, + joinSource: JoinSource, +): Promise => { + if (call.state.callingState !== CallingState.RINGING) return true; + return call.isCreatedByMe + ? reconcileAsCaller(call, joinSource) + : reconcileAsCallee(call); +}; + +const reconcileAsCaller = async ( + call: Call, + joinSource: JoinSource, +): Promise => { + const { session, members, endedAt } = call.state; + const currentUserId = call.currentUserId; + + // checked before `accepted_by`: an ended session cannot be joined + if (endedAt || session?.ended_at) { + call.logger.info('ring: the call has ended, leaving'); + // `leave` reports the remote end to callingx off `reason: 'ended'` + return leave(call, { + reject: false, + reason: 'ended', + message: 'ring: call ended', + }); + } + + const acceptedBy = session?.accepted_by ?? {}; + if (Object.keys(acceptedBy).some((userId) => userId !== currentUserId)) { + call.logger.info('ring: the call was accepted, joining'); + try { + await call.join({ joinSource }); + } catch (err) { + // `doJoin` restores the ringing state when a join fails, so the ring is + // still open. Report it unsettled and let the next poll retry. + call.logger.error('Failed to join an accepted call', err); + return false; + } + return true; + } + + const otherMembers = members + .filter((member) => member.user_id !== currentUserId) + .map((member) => member.user_id); + if (otherMembers.length === 0) return false; + + const rejectedBy = session?.rejected_by ?? {}; + if (otherMembers.every((userId) => rejectedBy[userId])) { + call.logger.info('ring: everyone rejected, leaving'); + return leave(call, { + reject: true, + reason: 'cancel', + message: 'ring: everyone rejected', + }); + } + + return false; +}; + +// the current user's own accept or reject, on this or another device, is +// handled by `resolveOwnRingOutcome`, and `call.ended` by `watchCallEnded`. +const reconcileAsCallee = async (call: Call): Promise => { + const createdById = call.state.createdBy?.id; + const rejectedBy = call.state.session?.rejected_by ?? {}; + if (createdById && rejectedBy[createdById]) { + call.logger.info('ring: the caller cancelled, leaving'); + return leave(call, { + reason: 'ended', + message: 'ring: creator rejected', + }); + } + return false; +}; + +// `false` when the call could not be left, so the ring stays open for a retry. +const leave = async ( + call: Call, + options: CallLeaveOptions, +): Promise => { + try { + await call.leave(options); + return true; + } catch (err) { + call.logger.error('Failed to leave a ringing call', err); + return false; + } +}; diff --git a/packages/client/src/ringing/resolveOwnRingOutcome.ts b/packages/client/src/ringing/resolveOwnRingOutcome.ts new file mode 100644 index 0000000000..705b925ea0 --- /dev/null +++ b/packages/client/src/ringing/resolveOwnRingOutcome.ts @@ -0,0 +1,58 @@ +import { CallingState } from '../store'; +import type { CallSessionResponse } from '../gen/coordinator'; + +export type OwnRingOutcome = { + /** + * Whether the current user accepted or rejected the ring, on this device or + * on another one. Either way the ring no longer needs to time out. + */ + settledByMe: boolean; + + /** + * Set when this device should stop ringing and leave, carrying the reason to + * report to the native call UI. + */ + leaveReason?: 'answeredElsewhere' | 'rejected'; +}; + +export type OwnRingOutcomeInput = { + /** The call session to read the accept and reject maps from. */ + session: CallSessionResponse | undefined; + /** The connected user, or `undefined` when there is none. */ + currentUserId: string | undefined; + /** The current calling state. */ + callingState: CallingState; +}; + +/** + * Decides what a ringing call should do about the current user's own accept or + * reject, which may have happened on another device. + * + * Accepting on *this* device also lands in `accepted_by`, so an acceptance only + * means another device took the call when this one is still ringing. + * + * Only meaningful for a ringing call; the caller checks that. + */ +export const resolveOwnRingOutcome = ({ + session, + currentUserId, + callingState, +}: OwnRingOutcomeInput): OwnRingOutcome => { + if (!currentUserId) return { settledByMe: false }; + + const acceptedByMe = Boolean(session?.accepted_by[currentUserId]); + const rejectedByMe = Boolean(session?.rejected_by[currentUserId]); + if (!acceptedByMe && !rejectedByMe) return { settledByMe: false }; + + const answeredElsewhere = + acceptedByMe && callingState === CallingState.RINGING; + + return { + settledByMe: true, + leaveReason: answeredElsewhere + ? 'answeredElsewhere' + : rejectedByMe + ? 'rejected' + : undefined, + }; +}; diff --git a/packages/client/src/rtc/e2ee/workerMessages.ts b/packages/client/src/rtc/e2ee/workerMessages.ts new file mode 100644 index 0000000000..5527e8933d --- /dev/null +++ b/packages/client/src/rtc/e2ee/workerMessages.ts @@ -0,0 +1,22 @@ +/** Internal commands whose completion the host must observe. */ +export type WorkerRequest = + | { type: 'cmd.init'; keyLength: number } + | { + type: 'cmd.set_key'; + userId: string; + keyIndex: number; + rawKey: ArrayBuffer; + } + | { type: 'cmd.set_shared_key'; keyIndex: number; rawKey: ArrayBuffer } + | { type: 'cmd.remove_key'; userId: string; keyIndex: number } + | { type: 'cmd.remove_shared_key'; keyIndex: number } + | { type: 'cmd.remove_all_keys'; userId: string }; + +export type WorkerResult = { + type: 'e2ee.command_result'; + requestId: number; + error?: string; +}; + +/** A timeout leaves the applied key state unknown, so the worker is stopped. */ +export const WORKER_TIMEOUT_MS = 10_000; diff --git a/packages/client/src/store/CallState.ts b/packages/client/src/store/CallState.ts index 2b856cefb5..3ed856af98 100644 --- a/packages/client/src/store/CallState.ts +++ b/packages/client/src/store/CallState.ts @@ -40,6 +40,7 @@ import { CallSessionParticipantLeftEvent, CallSessionResponse, CallSettingsResponse, + GetCallRingStateResponse, ClosedCaptionEvent, EgressResponse, MemberResponse, @@ -1082,6 +1083,30 @@ export class CallState { setCurrentValue(this.thumbnailsSubject, call.thumbnails); }; + /** + * Merges a polled ring state into the current session. Only the ring fields + * are touched: the endpoint does not return the session roster. + * + * @internal + * + * @param ringState the ring state as returned by the coordinator. + */ + updateFromRingState = (ringState: GetCallRingStateResponse) => { + setCurrentValue(this.sessionSubject, (session) => { + if (!session || session.id !== ringState.session_id) return session; + return { + ...session, + accepted_by: ringState.accepted_by, + rejected_by: ringState.rejected_by, + missed_by: ringState.missed_by, + ended_at: ringState.session_ended_at ?? session.ended_at, + }; + }); + if (ringState.call_ended_at) { + this.setEndedAt(new Date(ringState.call_ended_at)); + } + }; + /** * Updates the call state with the data received from the SFU server. * diff --git a/packages/client/src/store/__tests__/CallState.test.ts b/packages/client/src/store/__tests__/CallState.test.ts index 6e5a78c8c9..3274d95260 100644 --- a/packages/client/src/store/__tests__/CallState.test.ts +++ b/packages/client/src/store/__tests__/CallState.test.ts @@ -24,6 +24,7 @@ import { CallAcceptedEvent, CallEndedEvent, CallUpdatedEvent, + type GetCallRingStateResponse, MemberResponse, OwnCapability, } from '../../gen/coordinator'; @@ -1474,4 +1475,62 @@ describe('CallState', () => { expect(state['closedCaptionsTasks'].size).toBe(0); }); }); + + describe('updateFromRingState', () => { + const ringState = fromPartial({ + session_id: 'session-1', + accepted_by: { bob: '2026-08-24T10:00:04Z' }, + rejected_by: { carol: '2026-08-24T10:00:09Z' }, + missed_by: { dave: '2026-08-24T10:00:35Z' }, + }); + + const withSession = (id: string) => { + const state = new CallState(); + state['sessionSubject'].next( + fromPartial({ + id, + accepted_by: {}, + rejected_by: {}, + missed_by: {}, + participants: [fromPartial({ user_session_id: 'p1' })], + }), + ); + return state; + }; + + it('merges the ring maps into the current session', () => { + const state = withSession('session-1'); + state.updateFromRingState(ringState); + + expect(state.session?.accepted_by).toEqual(ringState.accepted_by); + expect(state.session?.rejected_by).toEqual(ringState.rejected_by); + expect(state.session?.missed_by).toEqual(ringState.missed_by); + }); + + it('leaves the session roster untouched', () => { + const state = withSession('session-1'); + state.updateFromRingState(ringState); + + expect(state.session?.participants).toHaveLength(1); + }); + + it('ignores a ring state that belongs to another session', () => { + const state = withSession('session-2'); + state.updateFromRingState(ringState); + + expect(state.session?.accepted_by).toEqual({}); + }); + + it('applies the end timestamps', () => { + const state = withSession('session-1'); + state.updateFromRingState({ + ...ringState, + session_ended_at: '2026-08-24T10:01:00Z', + call_ended_at: '2026-08-24T10:01:00Z', + }); + + expect(state.session?.ended_at).toBe('2026-08-24T10:01:00Z'); + expect(state.endedAt).toEqual(new Date('2026-08-24T10:01:00Z')); + }); + }); }); diff --git a/sample-apps/react-native/dogfood/src/components/Ringing/RingStateDebugPane.tsx b/sample-apps/react-native/dogfood/src/components/Ringing/RingStateDebugPane.tsx new file mode 100644 index 0000000000..5be3935159 --- /dev/null +++ b/sample-apps/react-native/dogfood/src/components/Ringing/RingStateDebugPane.tsx @@ -0,0 +1,183 @@ +import React, { useState } from 'react'; +import { + Platform, + Pressable, + ScrollView, + StyleSheet, + Text, + View, +} from 'react-native'; +import { + Call, + GetCallRingStateResponse, + useCall, + useCallStateHooks, +} from '@stream-io/video-react-native-sdk'; +import { useSafeAreaInsets } from 'react-native-safe-area-context'; +import { appTheme } from '../../theme'; + +/** + * Dev-only pane for inspecting the ring outcome of the active ringing call: + * the state the WebSocket delivered, next to what the `ring_state` endpoint + * returns on demand. + * + * Collapsed by default so it does not cover the ringing UI it is used to debug. + */ +export const RingStateDebugPane = () => { + const call = useCall(); + const [expanded, setExpanded] = useState(false); + const { top } = useSafeAreaInsets(); + + if (!call) { + return null; + } + + return ( + + setExpanded((prev) => !prev)} + > + {expanded ? '▾' : '▸'} Ring state + + {expanded && } + + ); +}; + +const RingStateDebug = ({ call }: { call: Call }) => { + const { useCallCallingState, useCallSession } = useCallStateHooks(); + const callingState = useCallCallingState(); + const session = useCallSession(); + const [polled, setPolled] = useState(); + const [polledAt, setPolledAt] = useState(); + const [error, setError] = useState(); + const [isPolling, setIsPolling] = useState(false); + + const handlePoll = async () => { + setIsPolling(true); + setError(undefined); + try { + setPolled(await call.getRingState()); + setPolledAt(new Date().toLocaleTimeString()); + } catch (err) { + setPolled(undefined); + setError(err instanceof Error ? err.message : String(err)); + } finally { + setIsPolling(false); + } + }; + + return ( + + + + + + + + + + + {isPolling ? 'Reading…' : 'Read ring state'} + + + {error && {error}} + {polled && ( + <> + Read at {polledAt} + {JSON.stringify(polled, null, 2)} + + )} + + ); +}; + +const Row = ({ label, value }: { label: string; value?: string }) => ( + + {label} + + {value || '—'} + + +); + +const formatMap = (map?: { [key: string]: string }) => { + const userIds = Object.keys(map ?? {}); + return userIds.length > 0 ? userIds.join(', ') : undefined; +}; + +const styles = StyleSheet.create({ + container: { + position: 'absolute', + left: appTheme.spacing.md, + right: appTheme.spacing.md, + zIndex: appTheme.zIndex.IN_FRONT, + backgroundColor: appTheme.colors.static_overlay, + borderRadius: 8, + overflow: 'hidden', + }, + header: { + paddingVertical: appTheme.spacing.sm, + paddingHorizontal: appTheme.spacing.md, + }, + headerText: { + color: appTheme.colors.static_white, + fontSize: 14, + fontWeight: 'bold', + }, + body: { + maxHeight: 320, + paddingHorizontal: appTheme.spacing.md, + paddingBottom: appTheme.spacing.md, + }, + row: { + flexDirection: 'row', + justifyContent: 'space-between', + paddingVertical: 2, + }, + rowLabel: { + color: appTheme.colors.light_gray, + fontSize: 12, + marginRight: appTheme.spacing.sm, + }, + rowValue: { + color: appTheme.colors.static_white, + fontSize: 12, + flexShrink: 1, + }, + button: { + marginTop: appTheme.spacing.sm, + paddingVertical: appTheme.spacing.sm, + borderRadius: 6, + backgroundColor: appTheme.colors.primary, + alignItems: 'center', + }, + buttonDisabled: { + backgroundColor: appTheme.colors.disabled, + }, + buttonText: { + color: appTheme.colors.static_white, + fontSize: 13, + fontWeight: '600', + }, + error: { + color: appTheme.colors.error, + fontSize: 12, + marginTop: appTheme.spacing.sm, + }, + hint: { + color: appTheme.colors.light_gray, + fontSize: 11, + marginTop: appTheme.spacing.sm, + }, + json: { + color: appTheme.colors.light_blue, + fontSize: 11, + fontFamily: Platform.OS === 'ios' ? 'Menlo' : 'monospace', + }, +}); diff --git a/sample-apps/react-native/dogfood/src/components/VideoWrapper.tsx b/sample-apps/react-native/dogfood/src/components/VideoWrapper.tsx index 9402cea01b..4cfb3298e3 100644 --- a/sample-apps/react-native/dogfood/src/components/VideoWrapper.tsx +++ b/sample-apps/react-native/dogfood/src/components/VideoWrapper.tsx @@ -28,6 +28,12 @@ export const VideoWrapper = ({ children }: PropsWithChildren<{}>) => { const localIpAddress = useAppGlobalStoreValue( (store) => store.localIpAddress, ); + const coordinatorBaseUrl = useAppGlobalStoreValue( + (store) => store.coordinatorBaseUrl, + ); + const disableRingStatePolling = useAppGlobalStoreValue( + (store) => store.disableRingStatePolling, + ); const customTheme = useCustomTheme(themeMode); const setState = useAppGlobalStoreSetState(); @@ -60,6 +66,8 @@ export const VideoWrapper = ({ children }: PropsWithChildren<{}>) => { token, tokenProvider, options: { + baseURL: coordinatorBaseUrl || undefined, + ringStatePolling: disableRingStatePolling ? false : undefined, rejectCallWhenBusy: false, logLevel: 'debug', logger: (level, message, ...args) => { @@ -107,7 +115,15 @@ export const VideoWrapper = ({ children }: PropsWithChildren<{}>) => { _videoClient?.disconnectUser(); setVideoClient(undefined); }; - }, [appEnvironment, setState, useLocalSfu, localIpAddress, user]); + }, [ + appEnvironment, + setState, + useLocalSfu, + localIpAddress, + user, + coordinatorBaseUrl, + disableRingStatePolling, + ]); if (!videoClient) { return null; diff --git a/sample-apps/react-native/dogfood/src/contexts/AppContext.tsx b/sample-apps/react-native/dogfood/src/contexts/AppContext.tsx index 93d2a5a915..98128e48d1 100644 --- a/sample-apps/react-native/dogfood/src/contexts/AppContext.tsx +++ b/sample-apps/react-native/dogfood/src/contexts/AppContext.tsx @@ -17,6 +17,8 @@ type AppGlobalStore = { localIpAddress: string; useLocalSfu?: boolean; devMode?: boolean; + coordinatorBaseUrl?: string; + disableRingStatePolling?: boolean; }; export const { @@ -37,6 +39,8 @@ export const { useLocalSfu: false, localIpAddress: '127.0.0.1', devMode: false, + coordinatorBaseUrl: '', + disableRingStatePolling: false, }, [ 'apiKey', @@ -48,5 +52,7 @@ export const { 'appMode', 'themeMode', 'devMode', + 'coordinatorBaseUrl', + 'disableRingStatePolling', ], ); diff --git a/sample-apps/react-native/dogfood/src/navigators/Call.tsx b/sample-apps/react-native/dogfood/src/navigators/Call.tsx index dca4679de5..369d9221c5 100644 --- a/sample-apps/react-native/dogfood/src/navigators/Call.tsx +++ b/sample-apps/react-native/dogfood/src/navigators/Call.tsx @@ -15,12 +15,17 @@ import { NavigationHeader } from '../components/NavigationHeader'; import { useOrientation } from '../hooks/useOrientation'; import { ActiveCall } from '../components/ActiveCall'; import { LayoutProvider } from '../contexts/LayoutContext'; +import { RingStateDebugPane } from '../components/Ringing/RingStateDebugPane'; +import { useAppGlobalStoreValue } from '../contexts/AppContext'; + +const ENABLE_RING_STATE_DEBUG = __DEV__; const CallStack = createNativeStackNavigator(); const Calls = () => { const calls = useCalls().filter((c) => c.ringing); const orientation = useOrientation(); + const devMode = useAppGlobalStoreValue((store) => store.devMode); const firstCall = calls.at(-1); @@ -48,6 +53,7 @@ const Calls = () => { landscape={orientation === 'landscape'} CallContent={customCallContent} /> + {(ENABLE_RING_STATE_DEBUG || devMode) && } ); diff --git a/sample-apps/react-native/dogfood/src/screens/Call/JoinCallScreen.tsx b/sample-apps/react-native/dogfood/src/screens/Call/JoinCallScreen.tsx index a29e7bc47b..225e6de7b3 100644 --- a/sample-apps/react-native/dogfood/src/screens/Call/JoinCallScreen.tsx +++ b/sample-apps/react-native/dogfood/src/screens/Call/JoinCallScreen.tsx @@ -26,9 +26,14 @@ import { KnownUsers } from '../../constants/KnownUsers'; import { randomId } from '../../modules/helpers/randomId'; import { useOrientation } from '../../hooks/useOrientation'; +const ENABLE_RING_PINNING = __DEV__; + const JoinCallScreen = () => { const [ringingUserIdsText, setRingingUserIdsText] = useState(''); + const [callType, setCallType] = useState('default'); + const [pinnedCallId, setPinnedCallId] = useState(''); const userId = useAppGlobalStoreValue((store) => store.userId); + const devMode = useAppGlobalStoreValue((store) => store.devMode); const [ringingUsers, setRingingUsers] = useState([]); const videoClient = useStreamVideoClient(); const { t } = useI18n(); @@ -47,7 +52,10 @@ const JoinCallScreen = () => { ringingUserIds = [...new Set([...ringingUserIds, userId])]; try { - const call = videoClient?.call('default', randomId()); + const call = videoClient?.call( + callType || 'default', + pinnedCallId || randomId(), + ); await call?.getOrCreate({ ring: true, video: true, @@ -57,6 +65,7 @@ const JoinCallScreen = () => { ring: { auto_cancel_timeout_ms: 30000, incoming_call_timeout_ms: 30000, + missed_call_timeout_ms: 30000, }, }, members: ringingUserIds.map((ringingUserId) => { @@ -74,7 +83,14 @@ const JoinCallScreen = () => { } finally { setIsLoading(false); } - }, [ringingUserIdsText, ringingUsers, videoClient, userId]); + }, [ + ringingUserIdsText, + ringingUsers, + videoClient, + userId, + callType, + pinnedCallId, + ]); const isRingingUserSelected = (userid: string) => ringingUsers.find((ringingUser) => ringingUser === userid); @@ -148,6 +164,29 @@ const JoinCallScreen = () => { disabled={startCallDisabled} onPress={startCallHandler} /> + {(ENABLE_RING_PINNING || devMode) && ( + + + Pin the ring to one call instance (leave blank for a new one) + + + + + )} @@ -213,6 +252,13 @@ const useStyles = () => { textInputStyle: { flex: 0, }, + pinningContainer: { + marginTop: appTheme.spacing.lg, + }, + pinningText: { + color: theme.colors.textPrimary, + fontSize: 13, + }, }), [theme], ); diff --git a/sample-apps/react-native/dogfood/src/screens/LoginScreen/EnvSwitcherButton.tsx b/sample-apps/react-native/dogfood/src/screens/LoginScreen/EnvSwitcherButton.tsx index 6ed3af9f96..398aab8104 100644 --- a/sample-apps/react-native/dogfood/src/screens/LoginScreen/EnvSwitcherButton.tsx +++ b/sample-apps/react-native/dogfood/src/screens/LoginScreen/EnvSwitcherButton.tsx @@ -1,5 +1,5 @@ import React, { useState } from 'react'; -import { Modal, Pressable, StyleSheet } from 'react-native'; +import { Modal, Pressable, StyleSheet, Text } from 'react-native'; import { useAppGlobalStoreSetState, useAppGlobalStoreValue, @@ -7,6 +7,7 @@ import { import { View } from 'react-native'; import { defaultTheme } from '@stream-io/video-react-native-sdk'; import { Button } from '../../components/Button'; +import { TextInput } from '../../components/TextInput'; const appEnvironments: AppEnvironment[] = [ 'pronto', @@ -43,6 +44,7 @@ export default function EnvSwitcherButton() { closeModal={closeModal} useLocalSfu /> + @@ -56,6 +58,53 @@ export default function EnvSwitcherButton() { ); } +/** + * Ring state options, used to dogfood the pollable ring state (VID-1444): + * a coordinator override for reaching an edge that serves the `ring_state` + * endpoint, and a switch to compare the ringing experience with polling off. + * + * Both are persisted, so the client created for a push in the background picks + * them up too. + */ +const RingStateOptions = () => { + const coordinatorBaseUrl = useAppGlobalStoreValue( + (store) => store.coordinatorBaseUrl, + ); + const disableRingStatePolling = useAppGlobalStoreValue( + (store) => store.disableRingStatePolling, + ); + const setState = useAppGlobalStoreSetState(); + + return ( + <> + {'Ring state'} + + setState({ coordinatorBaseUrl: e.nativeEvent.text.trim() }) + } + autoCapitalize="none" + autoCorrect={false} + keyboardType="url" + style={styles.modalInput} + /> + + {error &&

{error}

} + {polled && ( + <> +

Read at {polledAt}

+ 
+            {JSON.stringify(polled, null, 2)}
+
+ + )} + + ); +}; + +const Row = ({ label, value }: { label: string; value?: string }) => ( +
+
{label}
+
{value || '—'}
+
+); + +const formatMap = (map?: { [key: string]: string }) => { + const userIds = Object.keys(map ?? {}); + return userIds.length > 0 ? userIds.join(', ') : undefined; +}; diff --git a/sample-apps/react/react-dogfood/style/ringing.scss b/sample-apps/react/react-dogfood/style/ringing.scss index 51116cc345..1cd920458d 100644 --- a/sample-apps/react/react-dogfood/style/ringing.scss +++ b/sample-apps/react/react-dogfood/style/ringing.scss @@ -2,6 +2,7 @@ display: flex; align-items: center; justify-content: center; + gap: var(--str-video__spacing-lg); flex-grow: 1; padding: var(--str-video__spacing-md); } @@ -51,3 +52,54 @@ .rd__dialer-ringing-call-notification-text { margin-right: var(--str-video__spacing-sm); } + +.rd__dialer-debug { + overflow-y: auto; + width: 340px; + max-height: 70vh; + padding: var(--str-video__spacing-md); + border-radius: var(--str-video__border-radius-sm); + background-color: var(--str-video__background-color2); + color: var(--str-video__text-color1); + font-size: 0.75rem; +} + +.rd__dialer-debug-title { + margin: 0 0 var(--str-video__spacing-sm); + font-size: 0.875rem; +} + +.rd__dialer-debug-empty { + margin: 0 0 var(--str-video__spacing-sm); + color: var(--str-video__text-color2); +} + +.rd__dialer-debug-error { + margin: var(--str-video__spacing-sm) 0 0; + color: var(--str-video__alert-caution); +} + +.rd__dialer-debug-list { + margin: 0 0 var(--str-video__spacing-md); +} + +.rd__dialer-debug-row { + display: flex; + gap: var(--str-video__spacing-xs); + + dt { + flex: 0 0 6.5rem; + color: var(--str-video__text-color2); + } + + dd { + margin: 0; + overflow-wrap: anywhere; + } +} + +.rd__dialer-debug-json { + overflow-x: auto; + margin: var(--str-video__spacing-xs) 0 0; + font-size: 0.6875rem; +}