From 7c5fe97c7fa1f1b5666031fd00ff38bc3b92da6d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Maciej=20Braz=CC=87ewicz?= Date: Fri, 3 Jul 2026 13:27:42 +0200 Subject: [PATCH 1/2] call event reporting --- packages/stream_video/lib/src/call/call.dart | 45 +- .../lib/src/call/session/call_session.dart | 34 ++ .../open_api/coordinator_client_open_api.dart | 7 +- .../coordinator/open_api/coordinator_ws.dart | 58 ++ .../stream_video/lib/src/stream_video.dart | 38 ++ .../lib/src/telemetry/client_event.dart | 142 +++++ .../telemetry/client_event_error_mapper.dart | 71 +++ .../src/telemetry/client_event_reporter.dart | 522 ++++++++++++++++++ .../src/telemetry/client_event_transport.dart | 116 ++++ .../lib/src/telemetry/client_event_types.dart | 135 +++++ .../src/telemetry/media_frame_reporter.dart | 118 ++++ .../peer_connection_connect_reporter.dart | 117 ++++ .../lib/src/webrtc/peer_connection.dart | 12 + .../lib/src/webrtc/rtc_manager.dart | 75 +++ .../lib/src/webrtc/rtc_manager_factory.dart | 1 + .../telemetry/client_event_reporter_test.dart | 287 ++++++++++ .../telemetry/media_frame_reporter_test.dart | 90 +++ ...peer_connection_connect_reporter_test.dart | 174 ++++++ packages/stream_video/test/test_helpers.dart | 9 +- 19 files changed, 2048 insertions(+), 3 deletions(-) create mode 100644 packages/stream_video/lib/src/telemetry/client_event.dart create mode 100644 packages/stream_video/lib/src/telemetry/client_event_error_mapper.dart create mode 100644 packages/stream_video/lib/src/telemetry/client_event_reporter.dart create mode 100644 packages/stream_video/lib/src/telemetry/client_event_transport.dart create mode 100644 packages/stream_video/lib/src/telemetry/client_event_types.dart create mode 100644 packages/stream_video/lib/src/telemetry/media_frame_reporter.dart create mode 100644 packages/stream_video/lib/src/telemetry/peer_connection_connect_reporter.dart create mode 100644 packages/stream_video/test/src/telemetry/client_event_reporter_test.dart create mode 100644 packages/stream_video/test/src/telemetry/media_frame_reporter_test.dart create mode 100644 packages/stream_video/test/src/telemetry/peer_connection_connect_reporter_test.dart diff --git a/packages/stream_video/lib/src/call/call.dart b/packages/stream_video/lib/src/call/call.dart index be2c5e35f..62ada7fd9 100644 --- a/packages/stream_video/lib/src/call/call.dart +++ b/packages/stream_video/lib/src/call/call.dart @@ -36,6 +36,7 @@ import '../sfu/data/models/sfu_track_type.dart'; import '../shared_emitter.dart'; import '../state_emitter.dart'; import '../stream_video.dart'; +import '../telemetry/client_event_types.dart'; import '../utils/cancelable_operation.dart'; import '../utils/cancelables.dart'; import '../utils/extensions.dart'; @@ -1027,6 +1028,11 @@ class Call { } await _streamVideo.state.setActiveCall(this); + + _streamVideo.clientEventReporter + ..registerCall(callCid) + ..reportEvent(callCid, ClientEventStage.joinInitiated); + final result = await _join( connectOptions: connectOptions, @@ -1499,6 +1505,12 @@ class Call { return Result.error('call was left'); } + final reporter = _streamVideo.clientEventReporter; + final joinStageId = reporter.beginStage( + callCid, + ClientEventStage.coordinatorJoin, + ); + final joinResult = await _coordinatorClient.joinCall( callCid: callCid, create: create, @@ -1510,10 +1522,17 @@ class Call { ); if (joinResult is! Success) { + final failure = joinResult as Failure; + reporter.failStageWithError(joinStageId, failure.error); _logger.e(() => '[joinCall] join failed: $joinResult'); - return joinResult as Failure; + return failure; } + // Server call_session_id is now known; stamp it on subsequent stages. + reporter + ..setCallSessionId(callCid, joinResult.data.metadata.session.id) + ..completeStage(joinStageId, outcome: ClientEventOutcome.success); + final receivedOrCreated = CallReceivedOrCreatedData( wasCreated: joinResult.data.wasCreated, data: CallCreatedData( @@ -1694,6 +1713,10 @@ class Call { .listen( (stats) { _stats.emit(stats); + // Telemetry: feed subscriber stats for first-frame detection. + session.rtcManager?.onSubscriberStats( + stats.subscriberStatsBundle.stats, + ); }, ), ); @@ -1956,6 +1979,14 @@ class Call { final wasMigrating = _reconnectStrategy == SfuReconnectionStrategy.migrate; + final joinReason = _reconnectStrategy.joinReason; + if (joinReason != null) { + _streamVideo.clientEventReporter.newJoinAttempt( + callCid, + reason: joinReason, + ); + } + final reconnectResult = switch (_reconnectStrategy) { SfuReconnectionStrategy.fast => await _reconnectFast( reason: reconnectReason, @@ -2243,6 +2274,18 @@ class Call { Future> leave({DisconnectReason? reason}) async { _logger.i(() => '[leave] reason: $reason'); + final abortCode = switch (reason) { + DisconnectReasonEnded() || + DisconnectReasonCallEnded() => ClientEventStandardCode.backendLeave, + // Reconnection gave up — treat as a device-offline + DisconnectReasonReconnectionFailed() => + ClientEventStandardCode.networkOffline, + _ => ClientEventStandardCode.clientAborted, + }; + _streamVideo.clientEventReporter + ..abort(callCid, abortCode) + ..unregisterCall(callCid); + try { final didDisconnect = await _disconnect( sfuLeaveReason: _sfuLeaveReason(reason), diff --git a/packages/stream_video/lib/src/call/session/call_session.dart b/packages/stream_video/lib/src/call/session/call_session.dart index 34dd32c6a..13236dcaa 100644 --- a/packages/stream_video/lib/src/call/session/call_session.dart +++ b/packages/stream_video/lib/src/call/session/call_session.dart @@ -24,6 +24,8 @@ import '../../sfu/sfu_client.dart'; import '../../sfu/sfu_extensions.dart'; import '../../sfu/ws/sfu_ws.dart'; import '../../shared_emitter.dart'; +import '../../telemetry/client_event.dart'; +import '../../telemetry/client_event_types.dart'; import '../../utils/debounce_buffer.dart'; import '../../webrtc/model/rtc_model_mapper_extensions.dart'; import '../../webrtc/model/rtc_tracks_info.dart'; @@ -195,6 +197,10 @@ class CallSession extends Disposable { bool isAnonymousUser = false, String? unifiedSessionId, }) async { + final reporter = _streamVideo.clientEventReporter; + final wsJoinDetails = ClientEventDetails(sfuId: config.sfuName); + var wsJoinStageId = ''; + try { _logger.d( () => @@ -226,8 +232,19 @@ class CallSession extends Disposable { .mergeWith([delayedStream]) .listen(_onSfuEvent); + wsJoinStageId = reporter.beginStage( + callCid, + ClientEventStage.wsJoin, + details: wsJoinDetails, + ); + final wsResult = await sfuWS.connect(); if (wsResult.isFailure) { + reporter.failStageWithError( + wsJoinStageId, + (wsResult as Failure).error, + details: wsJoinDetails, + ); _logger.e(() => '[start] ws connect failed: $wsResult'); return const Result.failure( VideoError(message: 'Failed to connect to WS'), @@ -302,12 +319,23 @@ class CallSession extends Disposable { final event = await Future.any([joinResponseFuture, sfuErrorFuture]); if (event is SfuErrorEvent) { + reporter.failStageWithError( + wsJoinStageId, + event.error, + details: wsJoinDetails, + ); _logger.e(() => '[start] sfu error: ${event.error}'); return Result.errorWithCause(event.error.message, event.error); } final joinResponseEvent = event as SfuJoinResponseEvent; + reporter.completeStage( + wsJoinStageId, + outcome: ClientEventOutcome.success, + details: wsJoinDetails, + ); + _logger.v(() => '[start] sfu joined: $event'); // Ensure WebRTC initialization completes before creating rtcManager @@ -390,10 +418,16 @@ class CallSession extends Disposable { } on TimeoutException catch (e, stk) { final message = 'Waiting for "joinResponse" has timed out after ${joinResponseTimeout.inMilliseconds}ms'; + reporter.failStage( + wsJoinStageId, + failure: ClientEventFailure.requestTimeout(message), + details: wsJoinDetails, + ); _tracer.trace('joinRequestTimeout', message); _logger.e(() => '[start] failed: $e'); return Result.failure(VideoErrors.compose(e, stk)); } catch (e, stk) { + reporter.failStageWithError(wsJoinStageId, e, details: wsJoinDetails); _logger.e(() => '[start] failed: $e'); return Result.failure(VideoErrors.compose(e, stk)); } diff --git a/packages/stream_video/lib/src/coordinator/open_api/coordinator_client_open_api.dart b/packages/stream_video/lib/src/coordinator/open_api/coordinator_client_open_api.dart index d7f55ca13..0348a391d 100644 --- a/packages/stream_video/lib/src/coordinator/open_api/coordinator_client_open_api.dart +++ b/packages/stream_video/lib/src/coordinator/open_api/coordinator_client_open_api.dart @@ -16,6 +16,7 @@ import '../../models/models.dart'; import '../../retry/retry_policy.dart'; import '../../shared_emitter.dart'; import '../../state_emitter.dart'; +import '../../telemetry/client_event_reporter.dart'; import '../../token/token.dart'; import '../../token/token_manager.dart'; import '../../utils/none.dart'; @@ -41,13 +42,15 @@ class CoordinatorClientOpenApi extends CoordinatorClient { required RetryPolicy retryPolicy, required InternetConnection networkMonitor, this.isAnonymous = false, + ClientEventReporter clientEventReporter = const ClientEventReporter.noOp(), }) : _rpcUrl = rpcUrl, _wsUrl = wsUrl, _apiKey = apiKey, _tokenManager = tokenManager, _latencyService = latencyService, _networkMonitor = networkMonitor, - _retryPolicy = retryPolicy; + _retryPolicy = retryPolicy, + _clientEventReporter = clientEventReporter; final _logger = taggedLogger(tag: 'SV:CoordClient'); final String _rpcUrl; @@ -58,6 +61,7 @@ class CoordinatorClientOpenApi extends CoordinatorClient { final LatencyService _latencyService; final RetryPolicy _retryPolicy; final InternetConnection _networkMonitor; + final ClientEventReporter _clientEventReporter; final bool isAnonymous; @@ -251,6 +255,7 @@ class CoordinatorClientOpenApi extends CoordinatorClient { retryPolicy: _retryPolicy, includeUserDetails: includeUserDetails, networkMonitor: _networkMonitor, + clientEventReporter: _clientEventReporter, ); } diff --git a/packages/stream_video/lib/src/coordinator/open_api/coordinator_ws.dart b/packages/stream_video/lib/src/coordinator/open_api/coordinator_ws.dart index bcd5aa2a0..14f63ea38 100644 --- a/packages/stream_video/lib/src/coordinator/open_api/coordinator_ws.dart +++ b/packages/stream_video/lib/src/coordinator/open_api/coordinator_ws.dart @@ -2,6 +2,7 @@ import 'dart:async'; import 'dart:convert'; import 'package:internet_connection_checker_plus/internet_connection_checker_plus.dart'; +import 'package:uuid/uuid.dart'; import '../../../globals.dart'; import '../../../open_api/video/coordinator/api.dart' as open; @@ -11,6 +12,8 @@ import '../../logger/impl/tagged_logger.dart'; import '../../models/user_info.dart'; import '../../retry/retry_policy.dart'; import '../../shared_emitter.dart'; +import '../../telemetry/client_event_reporter.dart'; +import '../../telemetry/client_event_types.dart'; import '../../token/token_manager.dart'; import '../../utils/none.dart'; import '../../utils/result.dart'; @@ -44,6 +47,7 @@ class CoordinatorWebSocket extends StreamWebSocket implements HealthListener { required this.retryPolicy, required this.networkMonitor, this.includeUserDetails = false, + this.clientEventReporter = const ClientEventReporter.noOp(), }) : super( _buildUrl(url, apiKey), protocols: protocols, @@ -59,6 +63,7 @@ class CoordinatorWebSocket extends StreamWebSocket implements HealthListener { ), ); } + _reportCoordinatorWsStage(event); }; } @@ -89,6 +94,10 @@ class CoordinatorWebSocket extends StreamWebSocket implements HealthListener { /// Decides whether to pass user details to backend when connecting the user. final bool includeUserDetails; + /// Reports the `CoordinatorWS` telemetry stage, which follows this socket's + /// connection lifecycle. + final ClientEventReporter clientEventReporter; + bool _refreshToken = false; SharedEmitter get events => _events; @@ -97,6 +106,55 @@ class CoordinatorWebSocket extends StreamWebSocket implements HealthListener { String? userId; String? connectionId; + final _uuid = const Uuid(); + + /// The in-flight `CoordinatorWS` stage id, if a connect attempt is pending. + String? _coordinatorWsStageId; + + void _reportCoordinatorWsStage(ConnectionStateUpdatedEvent event) { + switch (event.newState) { + case ConnectionState.connecting: + if (_coordinatorWsStageId != null) break; + final stageId = clientEventReporter.beginConnectionStage( + ClientEventStage.coordinatorWs, + connectId: _uuid.v4(), + ); + _coordinatorWsStageId = stageId.isEmpty ? null : stageId; + case ConnectionState.connected: + final stageId = _coordinatorWsStageId; + if (stageId == null) break; + _coordinatorWsStageId = null; + clientEventReporter.completeStage( + stageId, + outcome: ClientEventOutcome.success, + ); + case ConnectionState.failed: + case ConnectionState.closed: + final stageId = _coordinatorWsStageId; + if (stageId == null) break; + _coordinatorWsStageId = null; + clientEventReporter.failStage( + stageId, + failure: ClientEventFailure( + ClientEventStandardCode.serverError, + 'Coordinator WS ${event.newState.name}', + ), + ); + case ConnectionState.disconnected: + final stageId = _coordinatorWsStageId; + if (stageId == null) break; + _coordinatorWsStageId = null; + clientEventReporter.failStage( + stageId, + failure: const ClientEventFailure.clientAborted( + 'Coordinator WS disconnected', + ), + ); + case ConnectionState.reconnecting: + break; + } + } + @override Future> connect() { _logger.v(() => '[connect] no args'); diff --git a/packages/stream_video/lib/src/stream_video.dart b/packages/stream_video/lib/src/stream_video.dart index cb4998137..362425fc6 100644 --- a/packages/stream_video/lib/src/stream_video.dart +++ b/packages/stream_video/lib/src/stream_video.dart @@ -55,6 +55,8 @@ import 'network_monitor_settings.dart'; import 'platform_detector/platform_detector.dart'; import 'push_notification/push_notification_manager.dart'; import 'retry/retry_policy.dart'; +import 'telemetry/client_event_reporter.dart'; +import 'telemetry/client_event_transport.dart'; import 'token/token.dart'; import 'token/token_manager.dart'; import 'utils/cancelable_operation.dart'; @@ -170,6 +172,17 @@ class StreamVideo extends Disposable { .toList(), ); + _clientEventReporter = _options.clientEventsReportingEnabled + ? ClientEventReporter( + transport: ClientEventTransport( + baseUrl: _options.coordinatorRpcUrl, + apiKey: apiKey, + authHeadersProvider: _clientEventAuthHeaders, + ), + resolveUserId: () => _state.currentUser.info.id, + ) + : const ClientEventReporter.noOp(); + _client = buildCoordinatorClient( user: user, apiKey: apiKey, @@ -179,6 +192,7 @@ class StreamVideo extends Disposable { rpcUrl: _options.coordinatorRpcUrl, wsUrl: _options.coordinatorWsUrl, networkMonitor: _networkMonitor, + clientEventReporter: _clientEventReporter, ); // Initialize the push notification manager if the provider is provided. @@ -301,6 +315,23 @@ class StreamVideo extends Disposable { late final InternetConnection _networkMonitor; late final PushNotificationManager? pushNotificationManager; + late final ClientEventReporter _clientEventReporter; + + @internal + ClientEventReporter get clientEventReporter => _clientEventReporter; + + /// Stream auth headers for the telemetry transport, mirroring the coordinator + /// client's authentication. + Future> _clientEventAuthHeaders() async { + final tokenResult = await _tokenManager.getToken(); + if (tokenResult is! Success) return const {}; + final token = tokenResult.data; + return { + 'stream-auth-type': token.authType.name, + if (token.rawValue.isNotEmpty) 'Authorization': token.rawValue, + }; + } + final Map _mutedCameraByStateChange = {}; final Map _mutedAudioByStateChange = {}; final Map _incomingAutoRejectTimers = {}; @@ -483,6 +514,7 @@ class StreamVideo extends Disposable { _subscriptions.cancelAll(); await pushNotificationManager?.dispose(); + _clientEventReporter.dispose(); await _state.clear(); return super.dispose(); @@ -1280,6 +1312,7 @@ CoordinatorClient buildCoordinatorClient({ required RetryPolicy retryPolicy, required LatencySettings latencySettings, required InternetConnection networkMonitor, + ClientEventReporter clientEventReporter = const ClientEventReporter.noOp(), }) { streamLog.i(_tag, () => '[buildCoordinatorClient] rpcUrl: $rpcUrl'); streamLog.i(_tag, () => '[buildCoordinatorClient] wsUrl: $wsUrl'); @@ -1296,6 +1329,7 @@ CoordinatorClient buildCoordinatorClient({ rpcUrl: rpcUrl, wsUrl: wsUrl, isAnonymous: user.type == UserType.anonymous, + clientEventReporter: clientEventReporter, ), ); } @@ -1340,6 +1374,7 @@ class StreamVideoOptions { this.networkMonitorSettings = const NetworkMonitorSettings(), this.allowMultipleActiveCalls = false, this.multiCallAudioPolicy = MultiCallAudioPolicy.suspendExisting, + this.clientEventsReportingEnabled = true, @Deprecated( 'Use audioConfigurationPolicy instead. This parameter will be removed in the next major release.', ) @@ -1378,6 +1413,9 @@ class StreamVideoOptions { /// Ignored when [allowMultipleActiveCalls] is `false`. final MultiCallAudioPolicy multiCallAudioPolicy; + /// Whether to report join-lifecycle telemetry + final bool clientEventsReportingEnabled; + /// Returns the current [NetworkMonitorSettings]. final NetworkMonitorSettings networkMonitorSettings; diff --git a/packages/stream_video/lib/src/telemetry/client_event.dart b/packages/stream_video/lib/src/telemetry/client_event.dart new file mode 100644 index 000000000..1b467966c --- /dev/null +++ b/packages/stream_video/lib/src/telemetry/client_event.dart @@ -0,0 +1,142 @@ +import 'client_event_types.dart'; + +/// Call/connection identity fields carried by a telemetry event. +class ClientEventIdentity { + const ClientEventIdentity({ + this.callType, + this.callId, + this.joinAttemptId, + this.joinReason, + this.callSessionId, + this.coordinatorConnectId, + }); + + final String? callType; + final String? callId; + final String? joinAttemptId; + final JoinReason? joinReason; + final String? callSessionId; + final String? coordinatorConnectId; + + ClientEventIdentity copyWith({ + String? callType, + String? callId, + String? joinAttemptId, + JoinReason? joinReason, + String? callSessionId, + String? coordinatorConnectId, + }) { + return ClientEventIdentity( + callType: callType ?? this.callType, + callId: callId ?? this.callId, + joinAttemptId: joinAttemptId ?? this.joinAttemptId, + joinReason: joinReason ?? this.joinReason, + callSessionId: callSessionId ?? this.callSessionId, + coordinatorConnectId: coordinatorConnectId ?? this.coordinatorConnectId, + ); + } + + Map toJson() { + final sessionId = callSessionId; + return { + if (callType != null) 'type': callType, + if (callId != null) 'id': callId, + if (joinAttemptId != null) 'join_attempt_id': joinAttemptId, + if (joinReason != null) 'join_reason': joinReason!.wireValue, + if (sessionId != null && sessionId.isNotEmpty) + 'call_session_id': sessionId, + if (coordinatorConnectId != null) + 'coordinator_connect_id': coordinatorConnectId, + }; + } +} + +/// Stage-specific fields. +class ClientEventDetails { + const ClientEventDetails({ + this.sfuId, + this.peerConnectionRole, + this.wasPreviouslyConnected, + this.previouslyConnectedAt, + this.iceState, + this.trackId, + }); + + final String? sfuId; + final ClientEventPeerConnectionRole? peerConnectionRole; + final bool? wasPreviouslyConnected; + final DateTime? previouslyConnectedAt; + final ClientEventIceState? iceState; + final String? trackId; + + Map toJson() { + final connectedAt = previouslyConnectedAt; + return { + if (sfuId != null) 'sfu_id': sfuId, + if (peerConnectionRole != null) + 'peer_connection': peerConnectionRole!.wireValue, + if (wasPreviouslyConnected != null) + 'was_previously_connected': wasPreviouslyConnected, + if (connectedAt != null) + 'previously_connected_timestamp': connectedAt.toUtc().toIso8601String(), + if (iceState != null) 'ice_state': iceState!.wireValue, + if (trackId != null) 'track_id': trackId, + }; + } +} + +class ClientEvent { + const ClientEvent({ + required this.stage, + required this.type, + required this.userId, + required this.userAgent, + required this.sdkVersion, + required this.timestamp, + this.identity = const ClientEventIdentity(), + this.stageId, + this.outcome, + this.retryCount, + this.elapsedTime, + this.failure, + this.details = const ClientEventDetails(), + }); + + final ClientEventStage stage; + final ClientEventType type; + final String userId; + final String userAgent; + final String sdkVersion; + final DateTime timestamp; + final ClientEventIdentity identity; + final String? stageId; + + final ClientEventOutcome? outcome; + final int? retryCount; + final int? elapsedTime; + final ClientEventFailure? failure; + + final ClientEventDetails details; + + Map toJson() { + final failure = this.failure; + return { + 'user_id': userId, + ...identity.toJson(), + 'stage': stage.wireValue, + if (stageId != null) 'stage_id': stageId, + 'event_type': type.wireValue, + 'timestamp': timestamp.toUtc().toIso8601String(), + 'user_agent': userAgent, + 'sdk_version': sdkVersion, + if (outcome != null) 'outcome': outcome!.wireValue, + if (retryCount != null) 'retry_count_attempt': retryCount, + if (elapsedTime != null) 'elapsed_time': elapsedTime, + if (failure != null) ...{ + 'retry_failure_code': failure.code.wireValue, + 'retry_failure_reason': failure.reason, + }, + ...details.toJson(), + }; + } +} diff --git a/packages/stream_video/lib/src/telemetry/client_event_error_mapper.dart b/packages/stream_video/lib/src/telemetry/client_event_error_mapper.dart new file mode 100644 index 000000000..2bac9eabc --- /dev/null +++ b/packages/stream_video/lib/src/telemetry/client_event_error_mapper.dart @@ -0,0 +1,71 @@ +import 'dart:async'; + +import 'package:http/http.dart' as http; + +import '../../open_api/video/coordinator/api.dart' show ApiException; +import '../errors/video_error.dart'; +import '../sfu/data/models/sfu_error.dart'; +import 'client_event_types.dart'; + +/// Maps thrown SDK errors to the spec's standard failure codes. +class ClientEventErrorMapper { + const ClientEventErrorMapper(); + + ClientEventFailure map(Object? error) { + final cause = error is VideoErrorWithCause ? error.cause : error; + + if (cause is SfuError) { + return ClientEventFailure( + ClientEventStandardCode.sfuError, + '${cause.code.name}: ${cause.message}', + ); + } + + if (cause is TimeoutException) { + return ClientEventFailure.requestTimeout( + cause.message ?? 'Request timed out', + ); + } + + // An ApiException means the server responded (with a status) — not offline. + if (cause is ApiException) { + return ClientEventFailure( + ClientEventStandardCode.serverError, + cause.message ?? 'API error ${cause.code}', + ); + } + + if (cause != null && _looksOffline(cause)) { + return const ClientEventFailure( + ClientEventStandardCode.networkOffline, + 'Device offline', + ); + } + + if (cause is http.ClientException) { + return ClientEventFailure( + ClientEventStandardCode.serverError, + cause.message, + ); + } + + final reason = error is VideoError + ? error.message + : (error?.toString() ?? 'Unknown error'); + return ClientEventFailure(ClientEventStandardCode.serverError, reason); + } + + /// Whether [cause] looks like a device-offline / no-route failure. + bool _looksOffline(Object cause) { + final raw = cause is http.ClientException + ? cause.message + : cause.toString(); + final message = raw.toLowerCase(); + return message.contains('failed host lookup') || + message.contains('network is unreachable') || + message.contains('no address associated with hostname') || + message.contains('software caused connection abort') || + message.contains('connection refused') || + message.contains('no internet'); + } +} diff --git a/packages/stream_video/lib/src/telemetry/client_event_reporter.dart b/packages/stream_video/lib/src/telemetry/client_event_reporter.dart new file mode 100644 index 000000000..ce539a35e --- /dev/null +++ b/packages/stream_video/lib/src/telemetry/client_event_reporter.dart @@ -0,0 +1,522 @@ +import 'package:uuid/uuid.dart'; + +import '../../globals.dart'; +import '../logger/impl/tagged_logger.dart'; +import '../models/call_cid.dart'; +import 'client_event.dart'; +import 'client_event_error_mapper.dart'; +import 'client_event_transport.dart'; +import 'client_event_types.dart'; + +const _tag = 'SV:ClientEventReporter'; + +/// Resolves the current user id at send time (the user can change, e.g. guest +/// login resolves the real id after connecting). +typedef UserIdResolver = String Function(); + +/// Returns the current time. Injectable so tests can be deterministic. +typedef Clock = DateTime Function(); + +/// Generates a fresh unique id (`join_attempt_id` / `stage_id`). Injectable so +/// tests can be deterministic. +typedef IdGenerator = String Function(); + +String _defaultId() => const Uuid().v4(); + +/// Client-level state shared across all calls +class ClientEventContext { + ClientEventContext({ + required this.resolveUserId, + Clock now = DateTime.now, + }) : _now = now; + + final UserIdResolver resolveUserId; + final Clock _now; + + /// Set by the coordinator socket + String? coordinatorConnectId; + + ClientEvent build( + ClientEventStage stage, + ClientEventType type, { + ClientEventIdentity callIdentity = const ClientEventIdentity(), + String? stageId, + ClientEventOutcome? outcome, + int? retryCount, + int? elapsedTime, + ClientEventFailure? failure, + ClientEventDetails details = const ClientEventDetails(), + }) { + return ClientEvent( + stage: stage, + type: type, + userId: resolveUserId(), + userAgent: xStreamClientHeader, + sdkVersion: streamVideoVersion, + timestamp: _now(), + identity: callIdentity.copyWith( + coordinatorConnectId: coordinatorConnectId, + ), + stageId: stageId, + outcome: outcome, + retryCount: retryCount, + elapsedTime: elapsedTime, + failure: failure, + details: details, + ); + } +} + +/// Reports join-lifecycle telemetry to the backend, stage by stage. +abstract interface class ClientEventReporter { + factory ClientEventReporter({ + required ClientEventTransport transport, + required UserIdResolver resolveUserId, + ClientEventErrorMapper errorMapper, + Clock now, + IdGenerator generateId, + }) = _ClientEventReporterImpl; + + /// A reporter that does nothing — used when reporting is disabled. + const factory ClientEventReporter.noOp() = _NoOpClientEventReporter; + + /// Starts tracking [cid] and generates its initial `join_attempt_id`. + void registerCall(StreamCallCid cid); + + /// Stops tracking [cid], aborting any still-in-flight stages as failures. + void unregisterCall(StreamCallCid cid); + + /// Generates a fresh `join_attempt_id` for [cid] (full rejoin / migration) + void newJoinAttempt(StreamCallCid cid, {required JoinReason reason}); + + /// Records the server-side `call_session_id` for [cid], known after the + /// coordinator join. + void setCallSessionId(StreamCallCid cid, String callSessionId); + + /// Records the current coordinator connection id. + void setCoordinatorConnectId(String? connectId); + + /// Emits a call-scoped `initiated` event and returns a `stage_id` to pair with + /// [completeStage] / [failStage]. + String beginStage( + StreamCallCid cid, + ClientEventStage stage, { + ClientEventDetails details = const ClientEventDetails(), + }); + + /// Emits a connection-scoped `initiated` event (not tied to a call). + String beginConnectionStage( + ClientEventStage stage, { + required String connectId, + ClientEventDetails details = const ClientEventDetails(), + }); + + /// Emits the `completed` event for the pair opened by [beginStage] / + /// [beginConnectionStage]. + void completeStage( + String stageId, { + required ClientEventOutcome outcome, + int retryCount = 0, + ClientEventFailure? failure, + ClientEventDetails details = const ClientEventDetails(), + }); + + /// Convenience for a `completed` + `failure` event. + void failStage( + String stageId, { + required ClientEventFailure failure, + int retryCount = 0, + ClientEventDetails details = const ClientEventDetails(), + }); + + /// Fails [stageId], mapping [error] through the error mapper. + void failStageWithError( + String stageId, + Object? error, { + int retryCount = 0, + ClientEventDetails details = const ClientEventDetails(), + }); + + /// Emits a single, unpaired event (e.g. `JoinInitiated`, `FirstVideoFrame`). + void reportEvent( + StreamCallCid cid, + ClientEventStage stage, { + ClientEventType type = ClientEventType.initiated, + ClientEventDetails details = const ClientEventDetails(), + }); + + /// Fails every in-flight stage of [cid] with [code] + void abort(StreamCallCid cid, ClientEventStandardCode code); + + void dispose(); +} + +/// Per-call attempt state. +class _CallState { + _CallState(this.cid, this.joinAttemptId); + + final StreamCallCid cid; + String joinAttemptId; + JoinReason joinReason = JoinReason.firstAttempt; + String? callSessionId; + + ClientEventIdentity get identity => ClientEventIdentity( + callType: cid.type.toString(), + callId: cid.id, + joinAttemptId: joinAttemptId, + joinReason: joinReason, + callSessionId: callSessionId, + ); +} + +/// A stage pair awaiting its `completed` event. +class _PendingStage { + _PendingStage({ + required this.stage, + required this.startedAt, + required this.identity, + this.cidValue, + }); + + final ClientEventStage stage; + final DateTime startedAt; + final ClientEventIdentity identity; + final String? cidValue; +} + +class _ClientEventReporterImpl implements ClientEventReporter { + _ClientEventReporterImpl({ + required this.transport, + required UserIdResolver resolveUserId, + this.errorMapper = const ClientEventErrorMapper(), + Clock now = DateTime.now, + IdGenerator generateId = _defaultId, + }) : _context = ClientEventContext(resolveUserId: resolveUserId, now: now), + _now = now, + _generateId = generateId; + + final ClientEventTransport transport; + final ClientEventErrorMapper errorMapper; + final ClientEventContext _context; + final Clock _now; + final IdGenerator _generateId; + + final _logger = taggedLogger(tag: _tag); + + /// cid value -> call state. + final Map _calls = {}; + + /// stage_id -> pending pair (both call- and connection-scoped). + final Map _pending = {}; + + @override + void registerCall(StreamCallCid cid) { + _calls[cid.value] = _CallState(cid, _generateId()); + _logger.d(() => '[registerCall] cid: ${cid.value}'); + } + + @override + void unregisterCall(StreamCallCid cid) { + abort(cid, ClientEventStandardCode.clientAborted); + _calls.remove(cid.value); + _logger.d(() => '[unregisterCall] cid: ${cid.value}'); + } + + @override + void newJoinAttempt( + StreamCallCid cid, { + required JoinReason reason, + }) { + final call = _calls[cid.value]; + if (call == null) return; + + call + ..joinAttemptId = _generateId() + ..joinReason = reason; + _logger.d( + () => + '[newJoinAttempt] cid: ${cid.value}, ' + 'id: ${call.joinAttemptId}, reason: ${reason.wireValue}', + ); + } + + @override + void setCallSessionId(StreamCallCid cid, String callSessionId) { + _calls[cid.value]?.callSessionId = callSessionId; + } + + @override + void setCoordinatorConnectId(String? connectId) { + _context.coordinatorConnectId = connectId; + } + + @override + String beginStage( + StreamCallCid cid, + ClientEventStage stage, { + ClientEventDetails details = const ClientEventDetails(), + }) { + final call = _calls[cid.value]; + if (call == null) { + _logger.w(() => '[beginStage] cid not registered: ${cid.value}'); + return ''; + } + + return _begin( + stage, + identity: call.identity, + cidValue: cid.value, + details: details, + ); + } + + @override + String beginConnectionStage( + ClientEventStage stage, { + required String connectId, + ClientEventDetails details = const ClientEventDetails(), + }) { + // Records the connection id so it's also stamped on later call-scoped + _context.coordinatorConnectId = connectId; + return _begin( + stage, + identity: const ClientEventIdentity(), + details: details, + ); + } + + @override + void completeStage( + String stageId, { + required ClientEventOutcome outcome, + int retryCount = 0, + ClientEventFailure? failure, + ClientEventDetails details = const ClientEventDetails(), + }) { + if (stageId.isEmpty) return; + + final pending = _pending.remove(stageId); + if (pending == null) { + _logger.w(() => '[completeStage] unknown stageId: $stageId'); + return; + } + + final elapsedMs = _now().difference(pending.startedAt).inMilliseconds; + + // Refresh the late-bound call_session_id (only known after CoordinatorJoin) + // while keeping the begin-time attempt identity stable. + final callSessionId = pending.cidValue == null + ? null + : _calls[pending.cidValue]?.callSessionId; + + _dispatch( + _context.build( + pending.stage, + ClientEventType.completed, + callIdentity: pending.identity.copyWith(callSessionId: callSessionId), + stageId: stageId, + outcome: outcome, + retryCount: retryCount, + elapsedTime: elapsedMs, + failure: failure, + details: details, + ), + ); + } + + @override + void failStage( + String stageId, { + required ClientEventFailure failure, + int retryCount = 0, + ClientEventDetails details = const ClientEventDetails(), + }) { + completeStage( + stageId, + outcome: ClientEventOutcome.failure, + retryCount: retryCount, + failure: failure, + details: details, + ); + } + + @override + void failStageWithError( + String stageId, + Object? error, { + int retryCount = 0, + ClientEventDetails details = const ClientEventDetails(), + }) { + failStage( + stageId, + failure: errorMapper.map(error), + retryCount: retryCount, + details: details, + ); + } + + @override + void reportEvent( + StreamCallCid cid, + ClientEventStage stage, { + ClientEventType type = ClientEventType.initiated, + ClientEventDetails details = const ClientEventDetails(), + }) { + final call = _calls[cid.value]; + if (call == null) { + _logger.w(() => '[reportEvent] cid not registered: ${cid.value}'); + return; + } + _dispatch( + _context.build( + stage, + type, + callIdentity: call.identity, + stageId: stage.carriesStageId ? _generateId() : null, + details: details, + ), + ); + } + + @override + void abort(StreamCallCid cid, ClientEventStandardCode code) { + final stageIds = _pending.entries + .where((entry) => entry.value.cidValue == cid.value) + .map((entry) => entry.key) + .toList(growable: false); + + if (stageIds.isEmpty) return; + + _logger.d( + () => + '[abort] cid: ${cid.value}, code: ${code.wireValue}, ' + 'inFlight: ${stageIds.length}', + ); + + for (final stageId in stageIds) { + failStage( + stageId, + failure: ClientEventFailure(code, 'Aborted: ${code.wireValue}'), + ); + } + } + + @override + void dispose() { + _calls.clear(); + _pending.clear(); + transport.close(); + } + + String _begin( + ClientEventStage stage, { + required ClientEventIdentity identity, + String? cidValue, + ClientEventDetails details = const ClientEventDetails(), + }) { + final stageId = _generateId(); + _pending[stageId] = _PendingStage( + stage: stage, + startedAt: _now(), + identity: identity, + cidValue: cidValue, + ); + + _dispatch( + _context.build( + stage, + ClientEventType.initiated, + callIdentity: identity, + stageId: stageId, + details: details, + ), + ); + + return stageId; + } + + void _dispatch(ClientEvent event) { + _logger.v( + () => + '[dispatch] ${event.stage.wireValue}/${event.type.wireValue}' + "${event.outcome != null ? '/${event.outcome!.wireValue}' : ''} " + '${event.toJson()}', + ); + + // Fire-and-forget: never let telemetry break the join. + transport.send([event]); + } +} + +/// No-op reporter used when client event reporting is disabled. +class _NoOpClientEventReporter implements ClientEventReporter { + const _NoOpClientEventReporter(); + + @override + void registerCall(StreamCallCid cid) {} + + @override + void unregisterCall(StreamCallCid cid) {} + + @override + void newJoinAttempt(StreamCallCid cid, {required JoinReason reason}) {} + + @override + void setCallSessionId(StreamCallCid cid, String callSessionId) {} + + @override + void setCoordinatorConnectId(String? connectId) {} + + @override + String beginStage( + StreamCallCid cid, + ClientEventStage stage, { + ClientEventDetails details = const ClientEventDetails(), + }) => ''; + + @override + String beginConnectionStage( + ClientEventStage stage, { + required String connectId, + ClientEventDetails details = const ClientEventDetails(), + }) => ''; + + @override + void completeStage( + String stageId, { + required ClientEventOutcome outcome, + int retryCount = 0, + ClientEventFailure? failure, + ClientEventDetails details = const ClientEventDetails(), + }) {} + + @override + void failStage( + String stageId, { + required ClientEventFailure failure, + int retryCount = 0, + ClientEventDetails details = const ClientEventDetails(), + }) {} + + @override + void failStageWithError( + String stageId, + Object? error, { + int retryCount = 0, + ClientEventDetails details = const ClientEventDetails(), + }) {} + + @override + void reportEvent( + StreamCallCid cid, + ClientEventStage stage, { + ClientEventType type = ClientEventType.initiated, + ClientEventDetails details = const ClientEventDetails(), + }) {} + + @override + void abort(StreamCallCid cid, ClientEventStandardCode code) {} + + @override + void dispose() {} +} diff --git a/packages/stream_video/lib/src/telemetry/client_event_transport.dart b/packages/stream_video/lib/src/telemetry/client_event_transport.dart new file mode 100644 index 000000000..de7b83b9b --- /dev/null +++ b/packages/stream_video/lib/src/telemetry/client_event_transport.dart @@ -0,0 +1,116 @@ +import 'dart:async'; +import 'dart:convert'; + +import 'package:http/http.dart' as http; +import 'package:uuid/uuid.dart'; + +import '../../globals.dart'; +import '../logger/impl/tagged_logger.dart'; +import 'client_event.dart'; + +const _tag = 'SV:ClientEventTransport'; + +/// Resolves the Stream auth headers (`stream-auth-type` + `Authorization`) for +/// the telemetry request. +typedef ClientEventAuthHeadersProvider = Future> Function(); + +class ClientEventRetryConfig { + const ClientEventRetryConfig({ + this.maxAttempts = 5, + this.baseDelay = const Duration(milliseconds: 500), + this.maxDelay = const Duration(seconds: 8), + }); + + final int maxAttempts; + final Duration baseDelay; + final Duration maxDelay; + + Duration delayForAttempt(int attempt) { + final scaled = baseDelay.inMilliseconds * (1 << attempt); + final capped = scaled.clamp(0, maxDelay.inMilliseconds); + return Duration(milliseconds: capped); + } +} + +class ClientEventTransport { + ClientEventTransport({ + required String baseUrl, + required String apiKey, + this.authHeadersProvider, + http.Client? httpClient, + this.retryConfig = const ClientEventRetryConfig(), + }) : _endpoint = Uri.parse( + '$baseUrl/api/v2/video/call_client_event', + ).replace(queryParameters: {'api_key': apiKey}), + _httpClient = httpClient ?? http.Client(), + _ownsClient = httpClient == null; + + final Uri _endpoint; + final http.Client _httpClient; + final bool _ownsClient; + final ClientEventRetryConfig retryConfig; + + /// Supplies the Stream auth headers per request. + final ClientEventAuthHeadersProvider? authHeadersProvider; + + final _logger = taggedLogger(tag: _tag); + final _uuid = const Uuid(); + + /// Sends a batch of events. + Future send(List events) async { + if (events.isEmpty) return; + + final body = jsonEncode({ + 'events': events.map((e) => e.toJson()).toList(), + }); + + final authHeaders = await authHeadersProvider?.call() ?? const {}; + final headers = { + 'Content-Type': 'application/json', + 'X-Stream-Client': xStreamClientHeader, + 'x-client-request-id': _uuid.v4(), + ...authHeaders, + }; + + for (var attempt = 0; attempt < retryConfig.maxAttempts; attempt++) { + try { + final response = await _httpClient.post( + _endpoint, + headers: headers, + body: body, + ); + + final status = response.statusCode; + if (status >= 200 && status < 300) { + _logger.v(() => '[send] ok ($status)'); + return; + } + + if (status >= 400 && status < 500) { + _logger.w( + () => '[send] dropped (client error $status): ${response.body}', + ); + return; + } + + _logger.w( + () => '[send] attempt ${attempt + 1} failed (status $status)', + ); + } catch (e) { + _logger.w(() => '[send] attempt ${attempt + 1} error: $e'); + } + + if (attempt + 1 < retryConfig.maxAttempts) { + await Future.delayed(retryConfig.delayForAttempt(attempt)); + } + } + + _logger.e( + () => '[send] giving up after ${retryConfig.maxAttempts} attempts', + ); + } + + void close() { + if (_ownsClient) _httpClient.close(); + } +} diff --git a/packages/stream_video/lib/src/telemetry/client_event_types.dart b/packages/stream_video/lib/src/telemetry/client_event_types.dart new file mode 100644 index 000000000..cb0a72a1b --- /dev/null +++ b/packages/stream_video/lib/src/telemetry/client_event_types.dart @@ -0,0 +1,135 @@ +library; + +import '../sfu/data/models/sfu_error.dart'; + +enum ClientEventStage { + joinInitiated, + coordinatorWs, + mediaDevicePermission, + coordinatorJoin, + wsJoin, + peerConnectionConnect, + firstVideoFrame, + firstAudioFrame; + + String get wireValue => switch (this) { + ClientEventStage.joinInitiated => 'JoinInitiated', + ClientEventStage.coordinatorWs => 'CoordinatorWS', + ClientEventStage.mediaDevicePermission => 'MediaDevicePermission', + ClientEventStage.coordinatorJoin => 'CoordinatorJoin', + ClientEventStage.wsJoin => 'WSJoin', + ClientEventStage.peerConnectionConnect => 'PeerConnectionConnect', + ClientEventStage.firstVideoFrame => 'FirstVideoFrame', + ClientEventStage.firstAudioFrame => 'FirstAudioFrame', + }; + + bool get carriesStageId => this != ClientEventStage.joinInitiated; +} + +enum ClientEventType { + initiated, + completed; + + String get wireValue => name; +} + +enum ClientEventOutcome { + success, + failure; + + String get wireValue => name; +} + +enum ClientEventPeerConnectionRole { + publish, + subscribe; + + String get wireValue => name; +} + +enum ClientEventIceState { + connected('CONNECTED'), + failed('FAILED'), + notConnected('NOT_CONNECTED'); + + const ClientEventIceState(this.wireValue); + + final String wireValue; +} + +/// A resolved failure: the standard [code] plus human-readable [reason] text +/// sent as `retry_failure_code` / `retry_failure_reason`. +class ClientEventFailure { + const ClientEventFailure(this.code, this.reason); + + const ClientEventFailure.ice([ + String reason = 'ICE connectivity checks failed', + ]) : this(ClientEventStandardCode.iceConnectivityFailed, reason); + + const ClientEventFailure.dtls([ + String reason = 'DTLS handshake failed', + ]) : this(ClientEventStandardCode.dtlsConnectivityFailed, reason); + + const ClientEventFailure.clientAborted([ + String reason = 'Aborted: user left during retry', + ]) : this(ClientEventStandardCode.clientAborted, reason); + + const ClientEventFailure.backendLeave([ + String reason = 'Backend leave during join', + ]) : this(ClientEventStandardCode.backendLeave, reason); + + const ClientEventFailure.requestTimeout([ + String reason = 'Request timed out', + ]) : this(ClientEventStandardCode.requestTimeout, reason); + + final ClientEventStandardCode code; + final String reason; +} + +enum ClientEventStandardCode { + clientAborted('CLIENT_ABORTED'), + backendLeave('BACKEND_LEAVE'), + requestTimeout('REQUEST_TIMEOUT'), + networkOffline('NETWORK_OFFLINE'), + iceConnectivityFailed('ICE_CONNECTIVITY_FAILED'), + dtlsConnectivityFailed('DTLS_CONNECTIVITY_FAILED'), + serverError('SERVER_ERROR'), + sfuError('SFU_ERROR'), + sfuGoAway('SFU_GO_AWAY'); + + const ClientEventStandardCode(this.wireValue); + + final String wireValue; +} + +enum JoinReason { + firstAttempt, + networkAvailable, + migration, + fullRejoin; + + /// Whether this reason should mint a new `join_attempt_id`. + bool get mintsNewAttempt => switch (this) { + JoinReason.firstAttempt => true, + JoinReason.migration => true, + JoinReason.fullRejoin => true, + JoinReason.networkAvailable => false, + }; + + String get wireValue => switch (this) { + JoinReason.firstAttempt => 'first-attempt', + JoinReason.networkAvailable => 'network-available', + JoinReason.migration => 'migration', + JoinReason.fullRejoin => 'full-rejoin', + }; +} + +extension SfuReconnectionStrategyJoinReason on SfuReconnectionStrategy { + JoinReason? get joinReason => switch (this) { + SfuReconnectionStrategy.rejoin => JoinReason.fullRejoin, + SfuReconnectionStrategy.migrate => JoinReason.migration, + SfuReconnectionStrategy.fast => null, + SfuReconnectionStrategy.disconnect => null, + SfuReconnectionStrategy.unspecified => null, + }; +} diff --git a/packages/stream_video/lib/src/telemetry/media_frame_reporter.dart b/packages/stream_video/lib/src/telemetry/media_frame_reporter.dart new file mode 100644 index 000000000..28c6ed4c1 --- /dev/null +++ b/packages/stream_video/lib/src/telemetry/media_frame_reporter.dart @@ -0,0 +1,118 @@ +import 'package:stream_webrtc_flutter/stream_webrtc_flutter.dart' as rtc; + +import '../logger/impl/tagged_logger.dart'; +import '../models/call_cid.dart'; +import '../webrtc/model/stats/rtc_inbound_rtp_audio_stream.dart'; +import '../webrtc/model/stats/rtc_stats.dart'; +import 'client_event.dart'; +import 'client_event_reporter.dart'; +import 'client_event_types.dart'; + +const _tag = 'SV:MediaFrameReporter'; + +/// Emits the single-shot `FirstAudioFrame` / `FirstVideoFrame` telemetry events +/// for one call session. Each fires at most once. +/// +/// - `FirstVideoFrame` uses a headless [rtc.RTCVideoRenderer] per remote video +/// track. +/// - `FirstAudioFrame` is detected from inbound-audio RTP stats +/// (`packetsReceived > 0`). +class MediaFrameReporter { + MediaFrameReporter({ + required this.reporter, + required this.callCid, + this.sfuId, + }); + + final ClientEventReporter reporter; + final StreamCallCid callCid; + final String? sfuId; + + final _logger = taggedLogger(tag: _tag); + + bool _audioReported = false; + bool _videoReported = false; + bool _disposed = false; + + final Map _renderers = {}; + + String? _audioTrackId; + + /// Records the remote audio track so [onSubscriberStats] can attribute the + /// `FirstAudioFrame` event to it. + void onRemoteAudioTrack({required String trackId}) { + if (_audioReported || _disposed) return; + _audioTrackId ??= trackId; + } + + /// Fed the subscriber's periodic WebRTC stats; reports `FirstAudioFrame` the + /// first time inbound audio packets are seen. + void onSubscriberStats(List stats) { + if (_audioReported || _disposed || _audioTrackId == null) return; + final flowing = stats.whereType().any( + (s) => (s.packetsReceived ?? 0) > 0, + ); + if (flowing) _reportAudio(); + } + + /// Attaches a headless renderer to a remote video track to detect its first + /// decoded/rendered frame. + Future onRemoteVideoTrack({ + required rtc.MediaStream stream, + required String trackId, + }) async { + if (_videoReported || _disposed || _renderers.containsKey(trackId)) return; + + try { + final renderer = rtc.RTCVideoRenderer(); + await renderer.initialize(); + + // State may have changed while awaiting initialize(). + if (_videoReported || _disposed) { + await renderer.dispose(); + return; + } + + renderer.onFirstFrameRendered = () => _reportVideo(trackId); + renderer.srcObject = stream; + _renderers[trackId] = renderer; + } catch (e) { + _logger.w(() => '[onRemoteVideoTrack] renderer setup failed: $e'); + } + } + + void _reportAudio() { + if (_audioReported || _disposed) return; + _audioReported = true; + reporter.reportEvent( + callCid, + ClientEventStage.firstAudioFrame, + details: ClientEventDetails(trackId: _audioTrackId, sfuId: sfuId), + ); + } + + void _reportVideo(String trackId) { + if (_videoReported || _disposed) return; + _videoReported = true; + reporter.reportEvent( + callCid, + ClientEventStage.firstVideoFrame, + details: ClientEventDetails(trackId: trackId, sfuId: sfuId), + ); + _disposeRenderers(); + } + + void _disposeRenderers() { + for (final renderer in _renderers.values) { + renderer + ..srcObject = null + ..dispose(); + } + _renderers.clear(); + } + + void dispose() { + _disposed = true; + _disposeRenderers(); + } +} diff --git a/packages/stream_video/lib/src/telemetry/peer_connection_connect_reporter.dart b/packages/stream_video/lib/src/telemetry/peer_connection_connect_reporter.dart new file mode 100644 index 000000000..b884e3f9d --- /dev/null +++ b/packages/stream_video/lib/src/telemetry/peer_connection_connect_reporter.dart @@ -0,0 +1,117 @@ +import 'package:stream_webrtc_flutter/stream_webrtc_flutter.dart' as rtc; + +import '../models/call_cid.dart'; +import 'client_event.dart'; +import 'client_event_reporter.dart'; +import 'client_event_types.dart'; + +class PeerConnectionConnectReporter { + PeerConnectionConnectReporter({ + required this.reporter, + required this.callCid, + required this.role, + this.sfuId, + }); + + final ClientEventReporter reporter; + final StreamCallCid callCid; + final ClientEventPeerConnectionRole role; + final String? sfuId; + + String? _stageId; + bool _wasPreviouslyConnected = false; + DateTime? _previouslyConnectedAt; + rtc.RTCIceConnectionState? _lastIceState; + + void onIceConnectionState(rtc.RTCIceConnectionState state) { + _lastIceState = state; + switch (state) { + case rtc.RTCIceConnectionState.RTCIceConnectionStateChecking: + _begin(); + case rtc.RTCIceConnectionState.RTCIceConnectionStateFailed: + _fail(const ClientEventFailure.ice()); + default: + break; + } + } + + void onConnectionState(rtc.RTCPeerConnectionState state) { + switch (state) { + case rtc.RTCPeerConnectionState.RTCPeerConnectionStateConnecting: + _begin(); + case rtc.RTCPeerConnectionState.RTCPeerConnectionStateConnected: + _succeed(); + case rtc.RTCPeerConnectionState.RTCPeerConnectionStateFailed: + _fail(const ClientEventFailure.dtls()); + default: + break; + } + } + + /// Fails any in-flight attempt — call when the peer connection is torn down + /// before it finished connecting (e.g. a reconnect replaces it). + void abortInFlight() { + if (_stageId == null) return; + _fail( + const ClientEventFailure.ice('Peer connection closed before connecting'), + ); + } + + void _begin() { + if (_stageId != null) return; + final stageId = reporter.beginStage( + callCid, + ClientEventStage.peerConnectionConnect, + details: _details(_observedIceState()), + ); + _stageId = stageId.isEmpty ? null : stageId; + } + + void _succeed() { + final stageId = _stageId; + if (stageId == null) return; + _stageId = null; + + reporter.completeStage( + stageId, + outcome: ClientEventOutcome.success, + details: _details(ClientEventIceState.connected), + ); + + _wasPreviouslyConnected = true; + _previouslyConnectedAt = DateTime.now(); + } + + void _fail(ClientEventFailure failure) { + final stageId = _stageId; + if (stageId == null) return; + _stageId = null; + + reporter.failStage( + stageId, + failure: failure, + details: _details(_observedIceState()), + ); + } + + ClientEventDetails _details(ClientEventIceState iceState) { + return ClientEventDetails( + peerConnectionRole: role, + wasPreviouslyConnected: _wasPreviouslyConnected, + previouslyConnectedAt: _previouslyConnectedAt, + sfuId: sfuId, + iceState: iceState, + ); + } + + ClientEventIceState _observedIceState() { + return switch (_lastIceState) { + rtc.RTCIceConnectionState.RTCIceConnectionStateConnected || + rtc.RTCIceConnectionState.RTCIceConnectionStateCompleted => + ClientEventIceState.connected, + rtc.RTCIceConnectionState.RTCIceConnectionStateFailed => + ClientEventIceState.failed, + _ => ClientEventIceState.notConnected, + }; + } +} diff --git a/packages/stream_video/lib/src/webrtc/peer_connection.dart b/packages/stream_video/lib/src/webrtc/peer_connection.dart index f1c848e87..962aefb72 100644 --- a/packages/stream_video/lib/src/webrtc/peer_connection.dart +++ b/packages/stream_video/lib/src/webrtc/peer_connection.dart @@ -114,6 +114,12 @@ class StreamPeerConnection extends Disposable { /// {@macro onTrack} OnTrack? onTrack; + /// Telemetry hook: fires on every raw ICE connection-state change. + void Function(rtc.RTCIceConnectionState state)? onIceConnectionStateUpdated; + + /// Telemetry hook: fires on every raw PC connection-state change. + void Function(rtc.RTCPeerConnectionState state)? onConnectionStateUpdated; + final _pendingCandidates = []; /// Attempts to restart ICE on the `RTCPeerConnection`. @@ -473,6 +479,8 @@ class StreamPeerConnection extends Disposable { void _onIceConnectionState(rtc.RTCIceConnectionState state) { _logger.v(() => '[onIceConnectionState] state: $state'); + onIceConnectionStateUpdated?.call(state); + switch (state) { case rtc.RTCIceConnectionState.RTCIceConnectionStateFailed: // in the `failed` state, we try to restart ICE immediately @@ -505,6 +513,8 @@ class StreamPeerConnection extends Disposable { } void _onConnectionState(rtc.RTCPeerConnectionState state) { + onConnectionStateUpdated?.call(state); + if (state == rtc.RTCPeerConnectionState.RTCPeerConnectionStateFailed) { _logger.w(() => '[onConnectionState] state: $state'); onReconnectionNeeded?.call(this, SfuReconnectionStrategy.rejoin); @@ -563,6 +573,8 @@ class StreamPeerConnection extends Disposable { onReconnectionNeeded = null; onIceCandidate = null; onTrack = null; + onIceConnectionStateUpdated = null; + onConnectionStateUpdated = null; _pendingCandidates.clear(); await pc.dispose(); return await super.dispose(); diff --git a/packages/stream_video/lib/src/webrtc/rtc_manager.dart b/packages/stream_video/lib/src/webrtc/rtc_manager.dart index f6fa292f4..98071c337 100644 --- a/packages/stream_video/lib/src/webrtc/rtc_manager.dart +++ b/packages/stream_video/lib/src/webrtc/rtc_manager.dart @@ -14,6 +14,9 @@ import '../errors/video_error_composer.dart'; import '../sfu/data/models/sfu_model_parser.dart'; import '../sfu/data/models/sfu_publish_options.dart'; import '../sfu/data/models/sfu_video_sender.dart'; +import '../telemetry/client_event_types.dart'; +import '../telemetry/media_frame_reporter.dart'; +import '../telemetry/peer_connection_connect_reporter.dart'; import '../utils/extensions.dart'; import 'codecs_helper.dart' as codecs; import 'codecs_helper.dart'; @@ -57,8 +60,10 @@ class RtcManager extends Disposable { required this.stateManager, required StreamVideo streamVideo, required this.pcFactory, + this.sfuId, }) : _streamVideo = streamVideo { subscriber.onTrack = _onRemoteTrack; + _initClientEventReporting(); } final _logger = taggedLogger(tag: _tag); @@ -70,6 +75,11 @@ class RtcManager extends Disposable { final TracedStreamPeerConnection? publisher; final TracedStreamPeerConnection subscriber; final StreamVideo _streamVideo; + final String? sfuId; + + PeerConnectionConnectReporter? _publisherConnectReporter; + PeerConnectionConnectReporter? _subscriberConnectReporter; + MediaFrameReporter? _mediaFrameReporter; final StreamPeerConnectionFactory pcFactory; @@ -101,6 +111,51 @@ class RtcManager extends Disposable { OnLocalTrackPublished? onLocalTrackPublished; OnRemoteTrackReceived? onRemoteTrackReceived; + void _initClientEventReporting() { + if (!_streamVideo.options.clientEventsReportingEnabled) return; + + final reporter = _streamVideo.clientEventReporter; + + _mediaFrameReporter = MediaFrameReporter( + reporter: reporter, + callCid: callCid, + sfuId: sfuId, + ); + + _subscriberConnectReporter = PeerConnectionConnectReporter( + reporter: reporter, + callCid: callCid, + role: ClientEventPeerConnectionRole.subscribe, + sfuId: sfuId, + ); + + subscriber + ..onIceConnectionStateUpdated = + _subscriberConnectReporter!.onIceConnectionState + ..onConnectionStateUpdated = + _subscriberConnectReporter!.onConnectionState; + + final pub = publisher; + if (pub != null) { + _publisherConnectReporter = PeerConnectionConnectReporter( + reporter: reporter, + callCid: callCid, + role: ClientEventPeerConnectionRole.publish, + sfuId: sfuId, + ); + pub + ..onIceConnectionStateUpdated = + _publisherConnectReporter!.onIceConnectionState + ..onConnectionStateUpdated = + _publisherConnectReporter!.onConnectionState; + } + } + + /// Feeds the subscriber's periodic WebRTC [stats] to first-frame telemetry + void onSubscriberStats(List stats) { + _mediaFrameReporter?.onSubscriberStats(stats); + } + StreamSubscription? _screenSharingStartedSubscription; @@ -204,6 +259,21 @@ class RtcManager extends Disposable { onRemoteTrackReceived?.call(pc, remoteTrack); tracks[remoteTrack.trackId] = remoteTrack; _logger.v(() => '[onRemoteTrack] published: ${remoteTrack.trackId}'); + + // Telemetry: first-frame detection for remote media. + final frameReporter = _mediaFrameReporter; + if (frameReporter != null) { + if (track.kind == 'audio') { + frameReporter.onRemoteAudioTrack(trackId: remoteTrack.trackId); + } else if (track.kind == 'video') { + unawaited( + frameReporter.onRemoteVideoTrack( + stream: stream, + trackId: remoteTrack.trackId, + ), + ); + } + } } Future changeDefaultAudioConstraints( @@ -500,6 +570,11 @@ class RtcManager extends Disposable { onLocalTrackPublished = null; onRemoteTrackReceived = null; + // Fail any peer-connection attempt that never finished connecting. + _publisherConnectReporter?.abortInFlight(); + _subscriberConnectReporter?.abortInFlight(); + _mediaFrameReporter?.dispose(); + await Future.wait([ if (publisher != null) publisher!.dispose().catchError(( diff --git a/packages/stream_video/lib/src/webrtc/rtc_manager_factory.dart b/packages/stream_video/lib/src/webrtc/rtc_manager_factory.dart index 95ddf850a..7c8ce106c 100644 --- a/packages/stream_video/lib/src/webrtc/rtc_manager_factory.dart +++ b/packages/stream_video/lib/src/webrtc/rtc_manager_factory.dart @@ -78,6 +78,7 @@ class RtcManagerFactory { stateManager: stateManager, streamVideo: streamVideo, pcFactory: pcFactory, + sfuId: callSessionConfig?.sfuName, ); } } diff --git a/packages/stream_video/test/src/telemetry/client_event_reporter_test.dart b/packages/stream_video/test/src/telemetry/client_event_reporter_test.dart new file mode 100644 index 000000000..9c96c5418 --- /dev/null +++ b/packages/stream_video/test/src/telemetry/client_event_reporter_test.dart @@ -0,0 +1,287 @@ +import 'dart:convert'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:http/http.dart' as http; +import 'package:http/testing.dart'; +import 'package:stream_video/src/models/call_cid.dart'; +import 'package:stream_video/src/telemetry/client_event_reporter.dart'; +import 'package:stream_video/src/telemetry/client_event_transport.dart'; +import 'package:stream_video/src/telemetry/client_event_types.dart'; + +void main() { + final cid = StreamCallCid(cid: 'default:call123'); + + late List> captured; + late ClientEventReporter reporter; + + setUp(() { + captured = []; + final mockClient = MockClient((request) async { + final body = jsonDecode(request.body) as Map; + for (final event in body['events'] as List) { + captured.add((event as Map).cast()); + } + return http.Response('', 200); + }); + + reporter = ClientEventReporter( + transport: ClientEventTransport( + baseUrl: 'https://example.com', + apiKey: 'key', + httpClient: mockClient, + retryConfig: const ClientEventRetryConfig(maxAttempts: 1), + ), + resolveUserId: () => 'user-1', + ); + }); + + // Sends are fire-and-forget; let the microtask/event queue drain. + Future flush() => Future.delayed(Duration.zero); + + test('JoinInitiated is a single unpaired event with common fields', () async { + reporter + ..registerCall(cid) + ..reportEvent(cid, ClientEventStage.joinInitiated); + await flush(); + + expect(captured, hasLength(1)); + final event = captured.single; + expect(event['stage'], 'JoinInitiated'); + expect(event['event_type'], 'initiated'); + expect(event['user_id'], 'user-1'); + expect(event['type'], 'default'); + expect(event['id'], 'call123'); + expect(event['join_attempt_id'], isNotEmpty); + expect(event['join_reason'], 'first-attempt'); + expect(event['sdk_version'], isNotEmpty); + expect(event.containsKey('stage_id'), isFalse); + // call_cid is not sent — the backend derives it from type + id, matching + // Swift/Android. + expect(event.containsKey('call_cid'), isFalse); + }); + + test('begin/complete share a stage_id and record the outcome', () async { + reporter.registerCall(cid); + final stageId = reporter.beginStage(cid, ClientEventStage.coordinatorJoin); + reporter.completeStage( + stageId, + outcome: ClientEventOutcome.success, + retryCount: 2, + ); + await flush(); + + expect(captured, hasLength(2)); + expect(captured[0]['event_type'], 'initiated'); + expect(captured[0]['stage_id'], stageId); + expect(captured[1]['event_type'], 'completed'); + expect(captured[1]['stage_id'], stageId); + expect(captured[1]['outcome'], 'success'); + expect(captured[1]['retry_count_attempt'], 2); + expect(captured[1].containsKey('elapsed_time'), isTrue); + }); + + test('failStage emits a failure completion with code + reason', () async { + reporter.registerCall(cid); + final stageId = reporter.beginStage(cid, ClientEventStage.wsJoin); + reporter.failStage( + stageId, + failure: const ClientEventFailure.requestTimeout('Timed out'), + ); + await flush(); + + final completed = captured.firstWhere( + (e) => e['event_type'] == 'completed', + ); + expect(completed['outcome'], 'failure'); + expect(completed['retry_failure_code'], 'REQUEST_TIMEOUT'); + expect(completed['retry_failure_reason'], 'Timed out'); + }); + + test('abort fails every in-flight stage with the given code', () async { + reporter.registerCall(cid); + final stageId = reporter.beginStage(cid, ClientEventStage.wsJoin); + reporter.abort(cid, ClientEventStandardCode.backendLeave); + await flush(); + + final completed = captured.firstWhere( + (e) => e['event_type'] == 'completed', + ); + expect(completed['stage_id'], stageId); + expect(completed['outcome'], 'failure'); + expect(completed['retry_failure_code'], 'BACKEND_LEAVE'); + }); + + test('newJoinAttempt mints a fresh id and updates join_reason', () async { + reporter + ..registerCall(cid) + ..reportEvent(cid, ClientEventStage.joinInitiated); + await flush(); + final firstId = captured.last['join_attempt_id']; + expect(captured.last['join_reason'], 'first-attempt'); + + reporter + ..newJoinAttempt(cid, reason: JoinReason.migration) + ..reportEvent(cid, ClientEventStage.joinInitiated); + await flush(); + final secondId = captured.last['join_attempt_id']; + + expect(secondId, isNot(firstId)); + expect(captured.last['join_reason'], 'migration'); + }); + + test( + 'setCallSessionId + setCoordinatorConnectId stamp later events', + () async { + reporter + ..setCoordinatorConnectId('coord-1') + ..registerCall(cid) + ..setCallSessionId(cid, 'session-xyz') + ..reportEvent(cid, ClientEventStage.joinInitiated); + await flush(); + + final event = captured.single; + expect(event['coordinator_connect_id'], 'coord-1'); + expect(event['call_session_id'], 'session-xyz'); + }, + ); + + test( + 'connection-scoped stage carries coordinator_connect_id, no join id', + () async { + // No registerCall — CoordinatorWS is connection-scoped, not call-scoped. + final stageId = reporter.beginConnectionStage( + ClientEventStage.coordinatorWs, + connectId: 'connect-abc', + ); + reporter.completeStage(stageId, outcome: ClientEventOutcome.success); + await flush(); + + expect(captured, hasLength(2)); + final initiated = captured[0]; + expect(initiated['stage'], 'CoordinatorWS'); + expect(initiated['coordinator_connect_id'], 'connect-abc'); + expect(initiated['stage_id'], stageId); + expect(initiated.containsKey('join_attempt_id'), isFalse); + expect(initiated.containsKey('call_cid'), isFalse); + + expect(captured[1]['event_type'], 'completed'); + expect(captured[1]['outcome'], 'success'); + expect(captured[1]['coordinator_connect_id'], 'connect-abc'); + }, + ); + + test('completion re-sends the join_attempt_id captured at begin', () async { + reporter.registerCall(cid); + final stageId = reporter.beginStage(cid, ClientEventStage.coordinatorJoin); + // A rejoin mints a new attempt id AFTER this stage began. + reporter.newJoinAttempt(cid, reason: JoinReason.fullRejoin); + reporter.completeStage(stageId, outcome: ClientEventOutcome.success); + await flush(); + + expect(captured, hasLength(2)); + expect( + captured[1]['join_attempt_id'], + captured[0]['join_attempt_id'], + reason: 'completion must reuse the begin-time attempt id', + ); + }); + + test('completion picks up call_session_id set after begin', () async { + // Reproduces the CoordinatorJoin case: session id is only known after the + // stage's initiated event, but the backend requires it on the success. + reporter.registerCall(cid); + final stageId = reporter.beginStage(cid, ClientEventStage.coordinatorJoin); + reporter.setCallSessionId(cid, 'session-late'); + reporter.completeStage(stageId, outcome: ClientEventOutcome.success); + await flush(); + + expect(captured, hasLength(2)); + // initiated fired before the session id was known. + expect(captured[0].containsKey('call_session_id'), isFalse); + // completion must carry it. + expect(captured[1]['call_session_id'], 'session-late'); + }); + + test('stages before registerCall are ignored', () async { + final stageId = reporter.beginStage(cid, ClientEventStage.coordinatorWs); + await flush(); + + expect(stageId, isEmpty); + expect(captured, isEmpty); + }); + + test( + 'failStageWithError classifies offline errors as NETWORK_OFFLINE', + () async { + reporter.registerCall(cid); + final offlineStage = reporter.beginStage( + cid, + ClientEventStage.coordinatorJoin, + ); + reporter.failStageWithError( + offlineStage, + http.ClientException('Failed host lookup: video.stream-io-api.com'), + ); + // A server-responded error is NOT offline. + final serverStage = reporter.beginStage(cid, ClientEventStage.wsJoin); + reporter.failStageWithError( + serverStage, + http.ClientException('Connection reset by peer'), + ); + await flush(); + + final offline = captured.firstWhere( + (e) => e['stage_id'] == offlineStage && e['event_type'] == 'completed', + ); + expect(offline['retry_failure_code'], 'NETWORK_OFFLINE'); + final server = captured.firstWhere( + (e) => e['stage_id'] == serverStage && e['event_type'] == 'completed', + ); + expect(server['retry_failure_code'], 'SERVER_ERROR'); + }, + ); + + test('injected clock + id generator make output deterministic', () async { + final ids = ['id-1', 'id-2', 'id-3']; + var clockMs = 1000; + final reporter = ClientEventReporter( + transport: ClientEventTransport( + baseUrl: 'https://example.com', + apiKey: 'key', + httpClient: MockClient((request) async { + final body = jsonDecode(request.body) as Map; + for (final event in body['events'] as List) { + captured.add((event as Map).cast()); + } + return http.Response('', 200); + }), + retryConfig: const ClientEventRetryConfig(maxAttempts: 1), + ), + resolveUserId: () => 'user-1', + generateId: () => ids.removeAt(0), + now: () => DateTime.fromMillisecondsSinceEpoch(clockMs, isUtc: true), + ); + + reporter.registerCall(cid); // consumes id-1 for join_attempt_id + final stageId = reporter.beginStage(cid, ClientEventStage.coordinatorJoin); + clockMs = 1500; // 500ms elapsed + reporter.completeStage(stageId, outcome: ClientEventOutcome.success); + await flush(); + + expect(stageId, 'id-2'); + expect(captured[0]['join_attempt_id'], 'id-1'); + expect(captured[1]['elapsed_time'], 500); + }); + + test('no-op reporter emits nothing', () async { + const noop = ClientEventReporter.noOp(); + expect(noop.beginStage(cid, ClientEventStage.wsJoin), isEmpty); + noop + ..registerCall(cid) + ..reportEvent(cid, ClientEventStage.joinInitiated) + ..abort(cid, ClientEventStandardCode.clientAborted); + await flush(); + + expect(captured, isEmpty); + }); +} diff --git a/packages/stream_video/test/src/telemetry/media_frame_reporter_test.dart b/packages/stream_video/test/src/telemetry/media_frame_reporter_test.dart new file mode 100644 index 000000000..0f9f5b315 --- /dev/null +++ b/packages/stream_video/test/src/telemetry/media_frame_reporter_test.dart @@ -0,0 +1,90 @@ +import 'dart:convert'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:http/http.dart' as http; +import 'package:http/testing.dart'; +import 'package:stream_video/src/models/call_cid.dart'; +import 'package:stream_video/src/telemetry/client_event_reporter.dart'; +import 'package:stream_video/src/telemetry/client_event_transport.dart'; +import 'package:stream_video/src/telemetry/media_frame_reporter.dart'; +import 'package:stream_video/src/webrtc/model/stats/rtc_inbound_rtp_audio_stream.dart'; +import 'package:stream_video/src/webrtc/model/stats/rtc_stats.dart'; + +void main() { + final cid = StreamCallCid(cid: 'default:call123'); + + late List> captured; + late ClientEventReporter reporter; + late MediaFrameReporter mfr; + + setUp(() { + captured = []; + final mockClient = MockClient((request) async { + final body = jsonDecode(request.body) as Map; + for (final event in body['events'] as List) { + captured.add((event as Map).cast()); + } + return http.Response('', 200); + }); + + reporter = ClientEventReporter( + transport: ClientEventTransport( + baseUrl: 'https://example.com', + apiKey: 'key', + httpClient: mockClient, + retryConfig: const ClientEventRetryConfig(maxAttempts: 1), + ), + resolveUserId: () => 'user-1', + )..registerCall(cid); + + mfr = MediaFrameReporter(reporter: reporter, callCid: cid, sfuId: 'sfu-1'); + }); + + tearDown(() => mfr.dispose()); + + Future flush() => Future.delayed(Duration.zero); + + const noAudio = [RtcInboundRtpAudioStream(packetsReceived: 0)]; + const audioFlowing = [ + RtcInboundRtpAudioStream(packetsReceived: 12), + ]; + + test('FirstAudioFrame fires once inbound audio packets appear', () async { + mfr.onRemoteAudioTrack(trackId: 'track-a:audio'); + + mfr.onSubscriberStats(noAudio); // still silent — no packets yet + await flush(); + expect(captured, isEmpty); + + mfr.onSubscriberStats(audioFlowing); + await flush(); + + final event = captured.single; + expect(event['stage'], 'FirstAudioFrame'); + expect(event['event_type'], 'initiated'); + expect(event['track_id'], 'track-a:audio'); + expect(event['sfu_id'], 'sfu-1'); + expect(event.containsKey('stage_id'), isTrue); // single-shot carries one + }); + + test('FirstAudioFrame fires at most once across stats ticks', () async { + mfr.onRemoteAudioTrack(trackId: 'track-a:audio'); + mfr + ..onSubscriberStats(audioFlowing) + ..onSubscriberStats(audioFlowing) + ..onSubscriberStats(audioFlowing); + await flush(); + + expect( + captured.where((e) => e['stage'] == 'FirstAudioFrame'), + hasLength(1), + ); + }); + + test('no FirstAudioFrame before a remote audio track is received', () async { + // Stats show audio, but we never recorded a remote audio track. + mfr.onSubscriberStats(audioFlowing); + await flush(); + expect(captured, isEmpty); + }); +} diff --git a/packages/stream_video/test/src/telemetry/peer_connection_connect_reporter_test.dart b/packages/stream_video/test/src/telemetry/peer_connection_connect_reporter_test.dart new file mode 100644 index 000000000..c03724f69 --- /dev/null +++ b/packages/stream_video/test/src/telemetry/peer_connection_connect_reporter_test.dart @@ -0,0 +1,174 @@ +import 'dart:convert'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:http/http.dart' as http; +import 'package:http/testing.dart'; +import 'package:stream_video/src/models/call_cid.dart'; +import 'package:stream_video/src/telemetry/client_event_reporter.dart'; +import 'package:stream_video/src/telemetry/client_event_transport.dart'; +import 'package:stream_video/src/telemetry/client_event_types.dart'; +import 'package:stream_video/src/telemetry/peer_connection_connect_reporter.dart'; +import 'package:stream_webrtc_flutter/stream_webrtc_flutter.dart' as rtc; + +void main() { + final cid = StreamCallCid(cid: 'default:call123'); + + late List> captured; + late ClientEventReporter reporter; + + setUp(() { + captured = []; + final mockClient = MockClient((request) async { + final body = jsonDecode(request.body) as Map; + for (final event in body['events'] as List) { + captured.add((event as Map).cast()); + } + return http.Response('', 200); + }); + + reporter = ClientEventReporter( + transport: ClientEventTransport( + baseUrl: 'https://example.com', + apiKey: 'key', + httpClient: mockClient, + retryConfig: const ClientEventRetryConfig(maxAttempts: 1), + ), + resolveUserId: () => 'user-1', + ); + reporter.registerCall(cid); + }); + + Future flush() => Future.delayed(Duration.zero); + + PeerConnectionConnectReporter tracker({ + ClientEventPeerConnectionRole role = + ClientEventPeerConnectionRole.subscribe, + }) { + return PeerConnectionConnectReporter( + reporter: reporter, + callCid: cid, + role: role, + sfuId: 'sfu-1', + ); + } + + test( + 'checking then connected emits a success pair with role + sfu', + () async { + tracker(role: ClientEventPeerConnectionRole.publish) + ..onIceConnectionState( + rtc.RTCIceConnectionState.RTCIceConnectionStateChecking, + ) + ..onConnectionState( + rtc.RTCPeerConnectionState.RTCPeerConnectionStateConnected, + ); + await flush(); + + expect(captured, hasLength(2)); + final initiated = captured[0]; + final completed = captured[1]; + expect(initiated['stage'], 'PeerConnectionConnect'); + expect(initiated['event_type'], 'initiated'); + expect(initiated['peer_connection'], 'publish'); + expect(initiated['sfu_id'], 'sfu-1'); + expect(initiated['was_previously_connected'], false); + // Non-standard field must NOT be sent (Swift has no such field). + expect(initiated.containsKey('user_session_id'), isFalse); + + expect(completed['event_type'], 'completed'); + expect(completed['outcome'], 'success'); + // A connected PC implies ICE connected, even though the test only drove + // ICE to `checking` — success must report CONNECTED, not NOT_CONNECTED. + expect(completed['ice_state'], 'CONNECTED'); + expect(completed['stage_id'], initiated['stage_id']); + expect(completed['was_previously_connected'], false); + // ice_state is now sent on success completions too (not just failures). + expect(completed.containsKey('ice_state'), isTrue); + }, + ); + + test('ICE failure reports ICE_CONNECTIVITY_FAILED with ice_state', () async { + tracker() + ..onIceConnectionState( + rtc.RTCIceConnectionState.RTCIceConnectionStateChecking, + ) + ..onIceConnectionState( + rtc.RTCIceConnectionState.RTCIceConnectionStateFailed, + ); + await flush(); + + final completed = captured.firstWhere( + (e) => e['event_type'] == 'completed', + ); + expect(completed['outcome'], 'failure'); + expect(completed['retry_failure_code'], 'ICE_CONNECTIVITY_FAILED'); + expect(completed['ice_state'], 'FAILED'); + }); + + test('PC failure reports DTLS_CONNECTIVITY_FAILED', () async { + tracker() + ..onConnectionState( + rtc.RTCPeerConnectionState.RTCPeerConnectionStateConnecting, + ) + ..onConnectionState( + rtc.RTCPeerConnectionState.RTCPeerConnectionStateFailed, + ); + await flush(); + + final completed = captured.firstWhere( + (e) => e['event_type'] == 'completed', + ); + expect(completed['outcome'], 'failure'); + expect(completed['retry_failure_code'], 'DTLS_CONNECTIVITY_FAILED'); + }); + + test( + 'a reconnect marks was_previously_connected on the next attempt', + () async { + final t = tracker() + ..onIceConnectionState( + rtc.RTCIceConnectionState.RTCIceConnectionStateChecking, + ) + ..onConnectionState( + rtc.RTCPeerConnectionState.RTCPeerConnectionStateConnected, + ); + // Second attempt (reconnect on the same PC instance). + t + ..onIceConnectionState( + rtc.RTCIceConnectionState.RTCIceConnectionStateChecking, + ) + ..onConnectionState( + rtc.RTCPeerConnectionState.RTCPeerConnectionStateConnected, + ); + await flush(); + + final initiateds = captured + .where((e) => e['event_type'] == 'initiated') + .toList(); + expect(initiateds, hasLength(2)); + expect(initiateds[0]['was_previously_connected'], false); + expect(initiateds[1]['was_previously_connected'], true); + expect(initiateds[1]['previously_connected_timestamp'], isNotNull); + }, + ); + + test('abortInFlight fails an in-progress attempt only', () async { + final t = tracker() + ..onIceConnectionState( + rtc.RTCIceConnectionState.RTCIceConnectionStateChecking, + ) + ..abortInFlight(); + await flush(); + + final completed = captured.firstWhere( + (e) => e['event_type'] == 'completed', + ); + expect(completed['outcome'], 'failure'); + + // A second abort with nothing in flight emits nothing more. + captured.clear(); + t.abortInFlight(); + await flush(); + expect(captured, isEmpty); + }); +} diff --git a/packages/stream_video/test/test_helpers.dart b/packages/stream_video/test/test_helpers.dart index 542c0db76..b369da143 100644 --- a/packages/stream_video/test/test_helpers.dart +++ b/packages/stream_video/test/test_helpers.dart @@ -1,6 +1,7 @@ import 'package:internet_connection_checker_plus/internet_connection_checker_plus.dart'; import 'package:mocktail/mocktail.dart'; import 'package:stream_video/src/call/permissions/permissions_manager.dart'; +import 'package:stream_video/src/telemetry/client_event_reporter.dart'; import 'package:stream_video/src/call/session/call_session.dart'; import 'package:stream_video/src/call/session/call_session_factory.dart'; import 'package:stream_video/src/call/state/call_state_notifier.dart'; @@ -18,7 +19,13 @@ class MockCoordinatorClient extends Mock implements CoordinatorClient {} class MockRtcMediaDeviceNotifier extends Mock implements RtcMediaDeviceNotifier {} -class MockStreamVideo extends Mock implements StreamVideo {} +class MockStreamVideo extends Mock implements StreamVideo { + /// Telemetry is a no-op in tests unless a test overrides it, so call sites + /// (join/leave) don't need to stub it. + @override + ClientEventReporter get clientEventReporter => + const ClientEventReporter.noOp(); +} class MockCallStateNotifier extends Mock implements CallStateNotifier {} From c0cd5856bd49be5ac4b28529456c04d2b5f64e59 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Maciej=20Braz=CC=87ewicz?= Date: Fri, 3 Jul 2026 13:30:32 +0200 Subject: [PATCH 2/2] changelog --- packages/stream_video/CHANGELOG.md | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/packages/stream_video/CHANGELOG.md b/packages/stream_video/CHANGELOG.md index 5426ea272..897eb390d 100644 --- a/packages/stream_video/CHANGELOG.md +++ b/packages/stream_video/CHANGELOG.md @@ -1,3 +1,9 @@ +## Upcoming + +### ✅ Added + +- Added client-side call join telemetry (`ClientEventReporter`). + ## 1.4.1 ### 🔄 Changed