feat(llc): Client call event reporting (join telemetry)

packages/stream_video/CHANGELOG.md

@@ -1,3 +1,9 @@
+## Upcoming
+
+### ✅ Added
+
+- Added client-side call join telemetry (`ClientEventReporter`).
+
 ## 1.4.1
 
 ### 🔄 Changed

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<CoordinatorJoined>) {
+      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<Result<None>> 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),

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));
     }

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,
     );
   }
 

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<CoordinatorEvent> 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<Result<None>> connect() {
     _logger.v(() => '[connect] no args');