From 9b12098d3ae3626384a0c1ba67a79a3c0366aff2 Mon Sep 17 00:00:00 2001 From: Oliver Lazoroski Date: Thu, 10 Sep 2026 11:44:34 +0200 Subject: [PATCH 1/3] feat(client): identify the SDK and version on SFU RPCs (#2425) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ### πŸ’‘ Overview The SFU's Twirp requests carry no client identity today - the SDK name and version only appear in the `JoinRequest` and `SendStats` payloads. This adds an `X-Stream-Client` header to every SFU RPC, mirroring what we already send to the coordinator. This gap surfaced during the Android 1.31.0 incident, where `startNoiseCancellation` raced ahead of the WS join, the SFU replied `ERROR_CODE_PARTICIPANT_NOT_FOUND`, and the SDK treated it as terminal and looped. The SFU couldn't gate a server-side mitigation on the client version because the RPC didn't say who was calling. The SFU already makes decisions based on the client string (codec selection, for one), so having it on every RPC lets the backend diagnose and mitigate client-specific regressions without waiting for an SDK release. ### πŸ“ Implementation notes The header carries the same client id shape the coordinator already receives through the user agent: | SDK | `X-Stream-Client` | | --- | --- | | React | `stream-video-react-v1.43.1` | | React Native | `stream-video-react-native-v1.44.2` | | everything else | `stream-video-js-v` | `getStreamClientId` lives next to the existing `getSdkName` / `getSdkVersion` helpers so the `SdkType` mapping stays in one place, and the version comes from the same source that feeds `JoinRequest` and the stats payloads. `setSdkInfo` runs at module init in both `react-sdk/index.ts` and `react-native-sdk/src/index.ts`, well before any SFU client is constructed at join time; with SDK info unset the value degrades to `stream-video-js-v0.0.0-development`. The WS `/ws` endpoint already carries `cid`, `user_id` and `api_key` as query params, so nothing changes there. [Format agreed with Marcelo in the thread](https://getstream.slack.com/archives/C040262MY9K/p1788870872729689). **Backend prerequisite:** `X-Stream-Client` is a non-safelisted header on a cross-origin Twirp POST, so the SFU's CORS `Access-Control-Allow-Headers` must include it before this ships - otherwise the preflight blocks every SFU RPC from the web SDK. **Known inconsistency:** the coordinator derives its own name from `SdkType[type].toLowerCase()`, so it sees `react_native` / `plain_javascript` where this header says `react-native` / `js`. Aligning the coordinator side is out of scope here. The other platforms (Android, iOS, Flutter, Unity) need the same header to close the gap fleet-wide; that's tracked separately per SDK team. 🎫 Ticket: https://linear.app/stream/issue/REACT-1169/identify-the-sdk-and-version-on-sfu-twirp-requests-x-stream-client πŸ“‘ Docs: n/a - internal transport header, no public API change. ## Summary by CodeRabbit * **New Features** * SFU requests now include a standardized SDK client identifier in their headers, including the platform and version (for example, React or React Native). * SDK identifiers now consistently fall back to a JavaScript identifier when platform information is unavailable or unsupported. * Added coverage to verify SDK identifier formatting and SFU request headers. (cherry picked from commit 2312002e455ecb0060c5393a0b0845eb16057322) --- packages/client/src/StreamSfuClient.ts | 8 +++- .../src/__tests__/StreamSfuClient.test.ts | 48 +++++++++++++++++++ .../client/src/stats/__tests__/utils.test.ts | 33 +++++++++++++ packages/client/src/stats/utils.ts | 15 ++++++ 4 files changed, 102 insertions(+), 2 deletions(-) create mode 100644 packages/client/src/stats/__tests__/utils.test.ts diff --git a/packages/client/src/StreamSfuClient.ts b/packages/client/src/StreamSfuClient.ts index ec910488b1..1f95f5fe49 100644 --- a/packages/client/src/StreamSfuClient.ts +++ b/packages/client/src/StreamSfuClient.ts @@ -39,7 +39,8 @@ import { } from './helpers/promise'; import { withoutConcurrency } from './helpers/concurrency'; import { getTimers } from './timers'; -import { Tracer, TraceSlice } from './stats'; +import { getSdkInfo } from './helpers/client-details'; +import { getStreamClientId, Tracer, TraceSlice } from './stats'; import { SfuJoinError, SfuTimeoutError } from './errors'; export type StreamSfuClientConstructor = { @@ -271,7 +272,10 @@ export class StreamSfuClient { this.rpc = createSignalClient({ baseUrl: server.url, interceptors: [ - withHeaders({ Authorization: `Bearer ${token}` }), + withHeaders({ + Authorization: `Bearer ${token}`, + 'X-Stream-Client': getStreamClientId(getSdkInfo()), + }), this.tracer && withRequestTracer(this.tracer.trace), this.logger.getLogLevel() === 'trace' && withRequestLogger(this.logger), withTimeout(rpcRequestTimeout, this.tracer?.trace), diff --git a/packages/client/src/__tests__/StreamSfuClient.test.ts b/packages/client/src/__tests__/StreamSfuClient.test.ts index bb1fa1994b..876d37743a 100644 --- a/packages/client/src/__tests__/StreamSfuClient.test.ts +++ b/packages/client/src/__tests__/StreamSfuClient.test.ts @@ -3,6 +3,8 @@ import { StreamSfuClient } from '../StreamSfuClient'; import { Dispatcher } from '../rtc'; import { StreamClient } from '../coordinator/connection/client'; import { getTimers } from '../timers'; +import { getSdkInfo, setSdkInfo } from '../helpers/client-details'; +import { SdkType } from '../gen/video/sfu/models/models'; /** * Minimal `WebSocket` stub used to drive `StreamSfuClient.close()` while the @@ -431,3 +433,49 @@ describe('StreamSfuClient.leaveAndClose()', () => { expect(notifyLeaveSpy).not.toHaveBeenCalled(); }); }); + +describe('StreamSfuClient RPC headers', () => { + beforeEach(() => { + CapturingWebSocket.instances = []; + vi.stubGlobal('WebSocket', CapturingWebSocket); + }); + + afterEach(() => { + vi.unstubAllGlobals(); + vi.restoreAllMocks(); + }); + + it('identifies the SDK and its version in every RPC request', async () => { + const sdkInfo = getSdkInfo(); + setSdkInfo({ + type: SdkType.REACT, + major: '1', + minor: '2', + patch: '3', + }); + const fetchMock = vi + .fn() + .mockResolvedValue( + new Response('{}', { headers: { 'content-type': 'application/json' } }), + ); + vi.stubGlobal('fetch', fetchMock); + + try { + const sfuClient = buildSfuClient(); + ( + sfuClient as unknown as { + joinResponseTask: { resolve: (v: unknown) => void }; + } + ).joinResponseTask.resolve({}); + + await sfuClient.startNoiseCancellation(); + + const [, init] = fetchMock.mock.calls[0]; + expect(new Headers(init.headers).get('x-stream-client')).toBe( + 'stream-video-react-v1.2.3', + ); + } finally { + setSdkInfo(sdkInfo!); + } + }); +}); diff --git a/packages/client/src/stats/__tests__/utils.test.ts b/packages/client/src/stats/__tests__/utils.test.ts new file mode 100644 index 0000000000..7dc29d4b5c --- /dev/null +++ b/packages/client/src/stats/__tests__/utils.test.ts @@ -0,0 +1,33 @@ +import { describe, expect, it } from 'vitest'; +import { getStreamClientId } from '../utils'; +import { SdkType } from '../../gen/video/sfu/models/models'; + +describe('getStreamClientId', () => { + const sdk = (type: SdkType) => ({ + type, + major: '1', + minor: '43', + patch: '1', + }); + + it('names the React SDK', () => { + expect(getStreamClientId(sdk(SdkType.REACT))).toBe( + 'stream-video-react-v1.43.1', + ); + }); + + it('names the React Native SDK', () => { + expect(getStreamClientId(sdk(SdkType.REACT_NATIVE))).toBe( + 'stream-video-react-native-v1.43.1', + ); + }); + + it('falls back to the plain JS client for every other SDK type', () => { + expect(getStreamClientId(sdk(SdkType.PLAIN_JAVASCRIPT))).toBe( + 'stream-video-js-v1.43.1', + ); + expect(getStreamClientId(undefined)).toBe( + 'stream-video-js-v0.0.0-development', + ); + }); +}); diff --git a/packages/client/src/stats/utils.ts b/packages/client/src/stats/utils.ts index a4fbe03f07..fffa11b624 100644 --- a/packages/client/src/stats/utils.ts +++ b/packages/client/src/stats/utils.ts @@ -51,3 +51,18 @@ export const getSdkName = (sdk: Sdk | undefined) => { export const getSdkVersion = (sdk: Sdk | undefined) => { return sdk ? `${sdk.major}.${sdk.minor}.${sdk.patch}` : '0.0.0-development'; }; + +/** + * Returns the identifier the SFU receives in the `X-Stream-Client` header, + * e.g. `stream-video-react-v1.43.1`. Matches the format the coordinator + * already gets through the client's user agent. + */ +export const getStreamClientId = (sdk: Sdk | undefined) => { + const sdkPlatforms: Partial> = { + [SdkType.REACT]: 'react', + [SdkType.REACT_NATIVE]: 'react-native', + }; + + const platform = (sdk && sdkPlatforms[sdk.type]) || 'js'; + return `stream-video-${platform}-v${getSdkVersion(sdk)}`; +}; From a5beb45a4c8bc3a57e6bdf0ee64e1f028e8e24fb Mon Sep 17 00:00:00 2001 From: Oliver Lazoroski Date: Thu, 10 Sep 2026 14:04:43 +0200 Subject: [PATCH 2/3] feat(client): poll ring state to reconcile a dropped ring outcome (#2393) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `call.accepted` / `call.rejected` / `call.missed` reach clients only over the coordinator WebSocket, best-effort and with no store-and-forward. The WS pings every 25s while the ring window is 30s, so a silently dead socket may never even be *detected* inside the window. The caller sits on a ringing screen while the callee is already in the call alone. This is the client half of D8: the caller reconciles by pull. After a quiet period with no ring event it reads `GET /call/{type}/{id}/ring_state` every 5s until the ring settles or the window closes β€” join when somebody accepted, cancel when everyone rejected, leave when the call already ended. "Nobody answered" stays with the local auto-drop. A new `src/ringing/` holds the four pieces, one file each: - **`RingStatePoller`** β€” the loop. Caller-only, on by default; disable or tune with `StreamClientOptions.ringStatePolling` (`false | { startAfterMs, intervalMs }`, defaults 15s / 5s). An incoming ring event restarts the quiet period rather than killing the poller, since in a group ring one rejection does not settle it. Bounded by `auto_cancel_timeout_ms` so it always resolves before the local auto-drop. - **`reconcileRingState`** β€” the single place a ring outcome is decided, shared by the WebSocket handlers and the poller. Takes no event and no payload: it reads `call.state` and branches on caller vs callee. `call_ended_at` / `session_ended_at` is checked *before* `accepted_by`, so an already-ended session is never joined. - **`RingTimeout`** β€” the ringing auto-drop, moved out of `Call` because it is the same shape as the poller. It also refuses to arm without ring settings; the old code returned early on a missing settings object but then read `settings.ring` unguarded. - **`resolveOwnRingOutcome`** β€” the current user's *own* accept or reject, which may have landed on another device. `reconcileAsCallee` used to defer to an inline effect in `Call.registerEffects`, so the callee's rules were split across two files. `Call.getRingState(callSessionId?)` wraps the endpoint; the session id defaults to the current session, and is passed explicitly to read one that has already ended. `CallState.updateFromRingState` merges the polled maps into the matching session so `session$` subscribers see what the dropped event carried. The response type comes from `src/gen/coordinator` β€” only `GetCallRingStateResponse` was taken from the regenerated output. The outcome used to be decided twice: `watchCallAccepted` / `watchCallRejected` read it off the event payload, and the poller had its own copy. They could drift, and the poller's copy only covered the caller. `CallState.updateFromEvent` is an `all` listener and `dispatchEvent` drains those before the typed ones, so by the time a ring handler runs the event's session is already on the call state. That makes the payload redundant, and both paths now run the same state-driven reconciler β€” the handlers rely on `updateFromEvent`, the poller applies the polled state itself. Two consequences worth review: - **A failed join keeps the ring open.** `doJoin` restores the ringing state when a join fails, so `RINGING -> JOINING -> RINGING`. Treating that first transition as the end of the ring left the call ringing with no auto-drop and no poller β€” the exact failure this PR exists to fix. `JOINING` is now transient for the poller, `RingTimeout` checks the state when its timer fires, and a failed join is reported non-terminal so the next poll retries. - **The callee branch is explicit**, and leave failures are caught rather than escaping the listener as an unhandled rejection. Acting twice on one outcome is prevented by the `RINGING` check in `reconcileRingState`, `singleFlight` on `Call.join`, and `leave()`'s own `LEFT` check. No event-identity dedup needed. **Drive-by fix:** `RingCallEvents` extracted from `AllClientCallEvents` β€” an object type, not a string union β€” so it resolved to `never`, the mapped registry type collapsed to `{}`, and the handler registry was unchecked. `call.missed` was registered nowhere as a result. It stays unregistered here, matching `main`: nothing in the reconciler acts on `missed_by`, so `RingTimeout` owns the "nobody answered" case until the server-owned ring timeout (VID-1427) lands. The polled `missed_by` map still reaches `session$` for consumers, and a `call.missed` event still restarts the poller's quiet period, since it proves the socket is alive. The watchdogs used to be cancelled outright before `leave()`'s first await, so a leave that failed left the call ringing with neither of them running. `RingTimeout` gained a `pause()` that keeps its deadline, and `leave()` resumes both watchdogs from its rejection handler when the call is still `RINGING` and nothing else is queued on the join/leave tag - so a transient `reject()` failure does not silently strand the ring, and the resumed timeout fires at its original deadline rather than extending the window. Now that two paths can pull the caller into a call, the tracing has to say which one did β€” a poll-driven join is itself a dropped-event signal, and without the distinction there is no way to measure how often the poller rescues a ring the socket lost. `CoordinatorJoin` client events gain a `source` field alongside `join_reason`: `ring-ws` from the event handlers, `ring-poll-api` from the poller. - It travels on a new local-only `joinSource` option on `Call.join()`, destructured out so it never reaches the coordinator with the rest of `JoinCallData`. `join` is wrapped in `singleFlight`, so a second concurrent trigger has its argument discarded along with the rest β€” the source always names the angle that *first* reached `join()`, which is the intended reading. - The reporter mirrors `join_reason` throughout: a per-cid map, a snapshot on the stage pair, and the same three `CoordinatorJoin` emissions, always read off the snapshot so all three agree. It is scoped to one join lifecycle by clearing it both on entry when absent and in a `finally` β€” the first stops a *nested* lifecycle inheriting it (a reconnect can open one during `join()`'s inter-attempt sleep, when `doJoin` has restored the pre-join state), the second stops a *later* `CoordinatorJoin` reported outside any lifecycle, which is the shape of a fast reconnect. - `source` is not in the coordinator OpenAPI schema yet, so it is typed by a local extension of the generated `ClientEvent` rather than by hand-editing generated code. The backend ignores unknown fields, so reporting it early is dropped rather than rejected β€” **but if the server lands the field as `join_source` instead, the client keeps sending `source` and the data silently never arrives.** Worth settling the name before the server half ships. **Merge note:** `main` moved `callingX.endCall(call, 'remote')` out of `watchCallEnded` and into `leave()`, fired off the new `reason: 'ended'`. The reconciler's two explicit `endCall` calls are gone in favour of that, otherwise every ended ring would have fired it twice. **Dogfood:** the web dialer gets a Pronto-only pane showing the live ring state next to an on-demand read of the endpoint, and learns `?coordinator_url` / `?use_local_coordinator` (like the home and join pages) plus `?call_id` / `?call_type` to pin a ring to one call. The RN app gets the same pane over the ringing call, call type / call id inputs to pin the ring, and a coordinator-URL override plus a polling on/off switch in the env switcher β€” both persisted, so the client built for a background push uses them too. The endpoint is live on the default edge; exercised by hand against it, the response matches the generated type field for field and the pane populates on ring and resets on cancel. The `source` field is covered at the unit level only β€” reproducing a *dropped* WS event end-to-end needs the coordinator socket suppressed while HTTP stays alive, so the assertions are on the outgoing `POST /call_client_event` body rather than on a dashboard. The server discards the field until its half ships anyway. 🎫 Ticket: https://linear.app/stream/issue/VID-1444/pollable-ring-state-the-caller-reconciles-a-dropped-ring-outcome-by πŸ”— Backend: https://github.com/GetStream/chat/pull/16110 πŸ“‘ Docs: pending β€” the customer-facing doc must carry the re-ring staleness caveat (until VID-1322) and the double-join guard recipe. - **New Features** - Added automatic ring-outcome polling for outgoing calls. - Calls now respond appropriately when participants accept, reject, miss, or end a call. - Added configurable polling timing, including an option to disable polling. - Added a development-only panel for inspecting ring state and manually refreshing results. - **Bug Fixes** - Improved synchronization between ring outcomes and call state, including missed and ended calls. Backport note: adapted to `release-v1`, which does not have the `ClientState` refactor (#2422). `clientState` -> `clientStore`, `ClientState` -> `StreamVideoWriteableStateStore`, and the new `CallState.updateFromRingState` uses the `this.setCurrentValue` class field instead of the module-level import. No behaviour change. (cherry picked from commit 73bfc980560786790597dfb1b512e3835e316bfc) --- packages/client/src/Call.ts | 154 +++--- .../src/__tests__/Call.autodrop.test.ts | 180 ++++--- .../src/__tests__/Call.getRingState.test.ts | 87 ++++ .../src/__tests__/Call.lifecycle.test.ts | 23 + .../src/__tests__/Call.ringSettled.test.ts | 71 +++ .../src/coordinator/connection/types.ts | 19 + .../client/src/events/__tests__/call.test.ts | 219 +-------- packages/client/src/events/call.ts | 74 +-- .../client/src/events/callEventHandlers.ts | 35 +- packages/client/src/gen/coordinator/index.ts | 67 +++ .../src/reporting/ClientEventReporter.ts | 34 +- .../__tests__/ClientEventReporter.test.ts | 191 +++++++- .../client/src/ringing/RingStatePoller.ts | 177 +++++++ packages/client/src/ringing/RingTimeout.ts | 90 ++++ .../ringing/__tests__/RingStatePoller.test.ts | 462 ++++++++++++++++++ .../src/ringing/__tests__/RingTimeout.test.ts | 199 ++++++++ .../__tests__/reconcileRingState.test.ts | 330 +++++++++++++ .../__tests__/resolveOwnRingOutcome.test.ts | 103 ++++ packages/client/src/ringing/index.ts | 4 + .../client/src/ringing/reconcileRingState.ts | 108 ++++ .../src/ringing/resolveOwnRingOutcome.ts | 58 +++ .../client/src/rtc/e2ee/workerMessages.ts | 22 + packages/client/src/store/CallState.ts | 25 + .../src/store/__tests__/CallState.test.ts | 59 +++ .../components/Ringing/RingStateDebugPane.tsx | 183 +++++++ .../dogfood/src/components/VideoWrapper.tsx | 18 +- .../dogfood/src/contexts/AppContext.tsx | 6 + .../dogfood/src/navigators/Call.tsx | 6 + .../src/screens/Call/JoinCallScreen.tsx | 50 +- .../screens/LoginScreen/EnvSwitcherButton.tsx | 64 ++- .../dogfood/src/utils/setPushConfig.ts | 26 +- sample-apps/react/react-dogfood/CLAUDE.md | 12 + .../components/Ringing/DialerPage.tsx | 33 +- .../components/Ringing/RingStateDebugPane.tsx | 94 ++++ .../react/react-dogfood/style/ringing.scss | 52 ++ 35 files changed, 2875 insertions(+), 460 deletions(-) create mode 100644 packages/client/src/__tests__/Call.getRingState.test.ts create mode 100644 packages/client/src/__tests__/Call.ringSettled.test.ts create mode 100644 packages/client/src/ringing/RingStatePoller.ts create mode 100644 packages/client/src/ringing/RingTimeout.ts create mode 100644 packages/client/src/ringing/__tests__/RingStatePoller.test.ts create mode 100644 packages/client/src/ringing/__tests__/RingTimeout.test.ts create mode 100644 packages/client/src/ringing/__tests__/reconcileRingState.test.ts create mode 100644 packages/client/src/ringing/__tests__/resolveOwnRingOutcome.test.ts create mode 100644 packages/client/src/ringing/index.ts create mode 100644 packages/client/src/ringing/reconcileRingState.ts create mode 100644 packages/client/src/ringing/resolveOwnRingOutcome.ts create mode 100644 packages/client/src/rtc/e2ee/workerMessages.ts create mode 100644 sample-apps/react-native/dogfood/src/components/Ringing/RingStateDebugPane.tsx create mode 100644 sample-apps/react/react-dogfood/components/Ringing/RingStateDebugPane.tsx diff --git a/packages/client/src/Call.ts b/packages/client/src/Call.ts index e6a01afaf6..f2a7a13ec2 100644 --- a/packages/client/src/Call.ts +++ b/packages/client/src/Call.ts @@ -50,6 +50,7 @@ import { EndCallResponse, GetCallReportResponse, GetCallResponse, + GetCallRingStateResponse, GetCallSessionParticipantStatsDetailsResponse, GetOrCreateCallRequest, GetOrCreateCallResponse, @@ -152,7 +153,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'; @@ -160,6 +161,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'; @@ -302,7 +304,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 clientStore: StreamVideoWriteableStateStore; public readonly streamClient: StreamClient; @@ -537,33 +540,20 @@ export class Call { createSubscription(this.state.session$, (session) => { if (!this.ringing) return; - const receiverId = this.clientStore.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', - ); - }); - } + }); }), ); }; @@ -608,6 +598,7 @@ export class Call { this.state.setCallingState(CallingState.RINGING); } this.scheduleAutoDrop(); + this.scheduleRingStatePolling(); this.leaveCallHooks.add(registerRingingCallEventHandlers(this)); } }; @@ -713,6 +704,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; @@ -821,6 +819,7 @@ export class Call { this.unifiedSessionId = undefined; this.ringingSubject.next(false); this.cancelAutoDrop(); + this.cancelRingStatePolling(); this.clientStore.unregisterCall(this); globalThis.streamRNVideoSDK?.callManager.stop({ @@ -863,6 +862,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; }); }; @@ -1056,6 +1066,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. @@ -1112,12 +1143,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; @@ -1159,7 +1192,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 { @@ -2081,8 +2114,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(); @@ -2117,7 +2152,7 @@ export class Call { const currentSfu = currentSfuClient.edgeName; await this.clientEventReporter.withJoinLifecycle( this.cid, - 'migration', + { joinReason: 'migration' }, () => this.doJoin({ ...this.joinCallData, @@ -3114,42 +3149,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 b38a9d2c79..7ea12fda7d 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, StreamVideoWriteableStateStore } 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 clientStore = new StreamVideoWriteableStateStore(); const streamClient = new StreamClient('abc'); - call = new Call({ + const newCall = new Call({ type: 'test', id: generateUUIDv4(), streamClient, clientEventReporter: new ClientEventReporter({ streamClient }), - clientStore: clientStore, + clientStore, + ringing: true, }); - // @ts-expect-error mocking only what we need for the test - clientStore.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, + clientStore.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..fe8d38f676 --- /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 { StreamVideoWriteableStateStore } 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 }), + clientStore: new StreamVideoWriteableStateStore(), + }); + + 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 9a99abcddc..200019ea93 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..2c4c9f6afd --- /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, StreamVideoWriteableStateStore } 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 clientStore = new StreamVideoWriteableStateStore(); + const streamClient = new StreamClient('abc'); + call = new Call({ + type: 'test', + id: generateUUIDv4(), + streamClient, + clientEventReporter: new ClientEventReporter({ streamClient }), + clientStore, + ringing, + }); + + clientStore.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 0f5f8747c2..9355c3359f 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, StreamVideoWriteableStateStore } 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..9b31b52c6c --- /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, StreamVideoWriteableStateStore } 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 clientStore = new StreamVideoWriteableStateStore(); + const streamClient = new StreamClient('abc', options); + const newCall = new Call({ + type: 'test', + id: generateUUIDv4(), + streamClient, + clientEventReporter: new ClientEventReporter({ streamClient }), + clientStore, + }); + + // @ts-expect-error mocking only what we need for the test + clientStore.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 clientStore = new StreamVideoWriteableStateStore(); + const streamClient = new StreamClient('abc', options); + const call = new Call({ + type: 'test', + id: generateUUIDv4(), + streamClient, + clientEventReporter: new ClientEventReporter({ streamClient }), + clientStore, + ringing: true, + }); + + // @ts-expect-error mocking only what we need for the test + clientStore.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..e74b9b886c --- /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, StreamVideoWriteableStateStore } 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 clientStore = new StreamVideoWriteableStateStore(); + const streamClient = new StreamClient('abc'); + const newCall = new Call({ + type: 'test', + id: generateUUIDv4(), + streamClient, + clientEventReporter: new ClientEventReporter({ streamClient }), + clientStore, + }); + + clientStore.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..f04bd31a63 --- /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, StreamVideoWriteableStateStore } 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 StreamVideoWriteableStateStore(); + store.setConnectedUser(fromPartial({ id: currentUserId })); + const streamClient = new StreamClient('api-key'); + const call = new Call({ + type: 'development', + id: '12345', + clientStore: 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 c4599955d2..b7bef5ca50 100644 --- a/packages/client/src/store/CallState.ts +++ b/packages/client/src/store/CallState.ts @@ -37,6 +37,7 @@ import { CallSessionParticipantLeftEvent, CallSessionResponse, CallSettingsResponse, + GetCallRingStateResponse, ClosedCaptionEvent, EgressResponse, MemberResponse, @@ -1352,6 +1353,30 @@ export class CallState { this.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) => { + this.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 3a8b833148..bf911bed56 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 5b7d33834e..aa0e4d2633 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-lg); } @@ -51,3 +52,54 @@ .rd__dialer-ringing-call-notification-text { margin-right: var(--str-video__spacing-md); } + +.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; +} From 8f1a12692b97ce52dbe420ea416d0886a3f2033e Mon Sep 17 00:00:00 2001 From: Oliver Lazoroski Date: Fri, 11 Sep 2026 16:41:41 +0200 Subject: [PATCH 3/3] chore(client): drop stray e2ee worker types from the #2393 backport `rtc/e2ee/workerMessages.ts` was swept into the #2393 squash on `main` and rode along into the backport. It is unrelated to ring-state polling and nothing imports it on either branch, so it is dead code. Removing it here keeps the unused `WORKER_TIMEOUT_MS` runtime constant out of the v1 build. The file stays on `main`, where the in-flight E2EE work is expected to pick it up. --- .../client/src/rtc/e2ee/workerMessages.ts | 22 ------------------- 1 file changed, 22 deletions(-) delete mode 100644 packages/client/src/rtc/e2ee/workerMessages.ts diff --git a/packages/client/src/rtc/e2ee/workerMessages.ts b/packages/client/src/rtc/e2ee/workerMessages.ts deleted file mode 100644 index 5527e8933d..0000000000 --- a/packages/client/src/rtc/e2ee/workerMessages.ts +++ /dev/null @@ -1,22 +0,0 @@ -/** 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;