diff --git a/CHANGELOG.md b/CHANGELOG.md index 154791a..cd25a8e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,17 @@ +# Unreleased + +## New Features + +### Video REST: advanced call statistics and reporting + +Application-level stats on `VideoClient` (`get_active_calls_status`, +`query_aggregate_call_stats`, `query_call_session_stats`, `get_daily_digest`, +`query_user_feedback`, `report_client_call_event`) and call-session-scoped stats +on `Call` (`get_call_participant_session_metrics`, +`query_call_participant_sessions`, `get_call_session_participant_stats_details`, +`query_call_session_participant_stats`, +`get_call_session_participant_stats_timeline`). + # v0.1.0-preview.2 docs.rs builds on current nightly. `doc_auto_cfg` was removed in 1.92 and diff --git a/README.md b/README.md index d495210..12ff8a3 100644 --- a/README.md +++ b/README.md @@ -26,6 +26,9 @@ remote audio and video, transform it, and publish media back into the call. - Create, query, update, end, and delete video calls. - Manage call members, permissions, recording, transcription, captions, livestreaming, custom events, and reactions. +- Query advanced call statistics and reporting: active-calls status, aggregate + and per-session stats, participant stats and metrics, daily digest, user + feedback, and client call-event reporting. - Join a call as a server-side SFU participant with retry, reconnect, and migration handling. - Subscribe globally or by participant session to remote audio, video, and diff --git a/src/models/mod.rs b/src/models/mod.rs index 1a15aa7..6f42abd 100644 --- a/src/models/mod.rs +++ b/src/models/mod.rs @@ -7,8 +7,10 @@ mod call; mod shared; +mod stats; mod user; pub use call::*; pub use shared::*; +pub use stats::*; pub use user::*; diff --git a/src/models/stats.rs b/src/models/stats.rs new file mode 100644 index 0000000..dbce3f8 --- /dev/null +++ b/src/models/stats.rs @@ -0,0 +1,444 @@ +//! Advanced call statistics and reporting request/response models. +//! +//! Field names track the getstream-go JSON tags. Deeply nested analytics +//! payloads are kept as [`serde_json::Value`] to stay robust across server-side +//! schema additions, matching the existing stats/report types in +//! [`super::call`]. Response types derive `Default` + `#[serde(default)]`. + +use serde::{Deserialize, Serialize}; +use serde_json::Value; + +use super::shared::{CustomData, SortParamRequest, Timestamp}; + +// Active calls status + +/// Aggregate counts for the current active-calls snapshot (`ActiveCallsSummary`). +#[derive(Debug, Clone, Default, Deserialize)] +#[serde(default)] +pub struct ActiveCallsSummary { + pub active_calls: i32, + pub active_publishers: i32, + pub active_subscribers: i32, + pub participants: i32, +} + +/// `get_active_calls_status` response (`GetActiveCallsStatusResponse`). +#[derive(Debug, Clone, Default, Deserialize)] +#[serde(default)] +pub struct GetActiveCallsStatusResponse { + pub duration: String, + pub start_time: Timestamp, + pub end_time: Timestamp, + /// Detailed join/publisher/subscriber metrics (opaque, schema-versioned). + pub metrics: Option, + pub summary: Option, +} + +// Aggregate call stats + +/// `query_aggregate_call_stats` request (`QueryAggregateCallStatsRequest`). +#[derive(Debug, Clone, Default, Serialize)] +pub struct QueryAggregateCallStatsRequest { + #[serde(skip_serializing_if = "Option::is_none")] + pub from: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub to: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub report_types: Option>, +} + +/// `query_aggregate_call_stats` response (`QueryAggregateCallStatsResponse`). +/// +/// Each report is an opaque, schema-versioned analytics bundle. +#[derive(Debug, Clone, Default, Deserialize)] +#[serde(default)] +pub struct QueryAggregateCallStatsResponse { + pub duration: String, + pub call_duration_report: Option, + pub call_participant_count_report: Option, + pub calls_per_day_report: Option, + pub network_metrics_report: Option, + pub quality_score_report: Option, + pub sdk_usage_report: Option, + pub user_feedback_report: Option, +} + +// Call session stats + +/// `query_call_session_stats` request (`QueryCallSessionStatsRequest`). +#[derive(Debug, Clone, Default, Serialize)] +pub struct QueryCallSessionStatsRequest { + #[serde(skip_serializing_if = "Option::is_none")] + pub limit: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub next: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub prev: Option, + #[serde(skip_serializing_if = "Vec::is_empty")] + pub sort: Vec, + #[serde(skip_serializing_if = "std::collections::HashMap::is_empty")] + pub filter_conditions: CustomData, +} + +/// `query_call_session_stats` response (`QueryCallSessionStatsResponse`). +#[derive(Debug, Clone, Default, Deserialize)] +#[serde(default)] +pub struct QueryCallSessionStatsResponse { + pub duration: String, + /// Per-session stat summaries (opaque, schema-versioned). + pub call_stats: Vec, + pub next: Option, + pub prev: Option, +} + +// Session participant stats (call-scoped) + +/// Query params for `get_call_session_participant_stats_details`. +#[derive(Debug, Clone, Default)] +pub struct GetCallSessionParticipantStatsDetailsRequest { + pub since: Option, + pub until: Option, + pub max_points: Option, +} + +/// `get_call_session_participant_stats_details` response +/// (`GetCallSessionParticipantStatsDetailsResponse`). +#[derive(Debug, Clone, Default, Deserialize)] +#[serde(default)] +pub struct GetCallSessionParticipantStatsDetailsResponse { + pub duration: String, + pub call_id: String, + pub call_session_id: String, + pub call_type: String, + pub user_id: String, + pub user_session_id: String, + pub publisher: Option, + pub subscriber: Option, + pub timeframe: Option, + pub user: Option, +} + +/// Query params for `query_call_session_participant_stats`. +#[derive(Debug, Clone, Default)] +pub struct QueryCallSessionParticipantStatsRequest { + pub limit: Option, + pub prev: Option, + pub next: Option, + /// Sort order for the returned participants. + /// + /// The coordinator currently answers any non-empty value on this endpoint + /// with `custom sorting is not supported`; it is accepted here so callers + /// are ready when sorting is enabled server-side. The `sort` on the + /// `query_call_session_stats` / `query_user_feedback` request bodies is + /// supported today. + pub sort: Vec, + pub filter_conditions: CustomData, +} + +/// `query_call_session_participant_stats` response +/// (`QueryCallSessionParticipantStatsResponse`). +#[derive(Debug, Clone, Default, Deserialize)] +#[serde(default)] +pub struct QueryCallSessionParticipantStatsResponse { + pub duration: String, + pub call_id: String, + pub call_session_id: String, + pub call_type: String, + /// Per-participant stat summaries (opaque, schema-versioned). + pub participants: Vec, + pub counts: Value, + pub call_started_at: Option, + pub call_ended_at: Option, + pub next: Option, + pub prev: Option, + pub tmp_data_source: Option, + pub call_events: Vec, +} + +/// Query params for `get_call_session_participant_stats_timeline`. +#[derive(Debug, Clone, Default)] +pub struct GetCallSessionParticipantStatsTimelineRequest { + pub start_time: Option, + pub end_time: Option, + pub severity: Vec, +} + +/// `get_call_session_participant_stats_timeline` response +/// (`QueryCallSessionParticipantStatsTimelineResponse`). +#[derive(Debug, Clone, Default, Deserialize)] +#[serde(default)] +pub struct QueryCallSessionParticipantStatsTimelineResponse { + pub duration: String, + pub call_id: String, + pub call_session_id: String, + pub call_type: String, + pub user_id: String, + pub user_session_id: String, + /// Timeline events (opaque, schema-versioned). + pub events: Vec, +} + +// Participant session metrics / sessions (call-scoped) + +/// Query params for `get_call_participant_session_metrics`. +#[derive(Debug, Clone, Default)] +pub struct GetCallParticipantSessionMetricsRequest { + pub since: Option, + pub until: Option, +} + +/// `get_call_participant_session_metrics` response +/// (`GetCallParticipantSessionMetricsResponse`). +#[derive(Debug, Clone, Default, Deserialize)] +#[serde(default)] +pub struct GetCallParticipantSessionMetricsResponse { + pub duration: String, + pub is_publisher: Option, + pub is_subscriber: Option, + pub joined_at: Option, + pub publisher_type: Option, + pub user_id: Option, + pub user_session_id: Option, + /// Per-track publish metrics (opaque, schema-versioned). + pub published_tracks: Vec, + pub client: Option, +} + +/// Query params for `query_call_participant_sessions`. +#[derive(Debug, Clone, Default)] +pub struct QueryCallParticipantSessionsRequest { + pub limit: Option, + pub prev: Option, + pub next: Option, + pub filter_conditions: CustomData, +} + +/// `query_call_participant_sessions` response +/// (`QueryCallParticipantSessionsResponse`). +#[derive(Debug, Clone, Default, Deserialize)] +#[serde(default)] +pub struct QueryCallParticipantSessionsResponse { + /// Session length in seconds. Unlike the `duration` string other endpoints + /// return (`"23.27ms"`), this endpoint returns an integer. + pub duration: i64, + pub call_id: String, + pub call_session_id: String, + pub call_type: String, + pub total_participant_duration: i64, + pub total_participant_sessions: i64, + /// Per-participant-session details (opaque, schema-versioned). + pub participants_sessions: Vec, + pub next: Option, + pub prev: Option, + pub session: Option, +} + +// Daily digest + +/// Query params for `get_daily_digest`. +#[derive(Debug, Clone, Default)] +pub struct GetDailyDigestRequest { + pub date: Option, + pub target_app_id: Option, +} + +/// `get_daily_digest` response (`GetDailyDigestResponse`). +#[derive(Debug, Clone, Default, Deserialize)] +#[serde(default)] +pub struct GetDailyDigestResponse { + pub duration: String, + pub date: String, + /// Readiness status: `ready`, `pending`, `failed`, `future_date`, `expired`. + pub status: String, + pub generated_at: Option, + pub retry_after: Option, + pub revision: Option, + pub schema_version: Option, + pub digest_kinds: Vec, + /// Per-broadcast digests (opaque, present only when `status` is `ready`). + pub broadcasts: Vec, + /// Per-call-session summaries (opaque, present only when `status` is `ready`). + pub call_sessions: Vec, + pub broadcast_rollup: Option, +} + +// User feedback + +/// `query_user_feedback` request (`QueryUserFeedbackRequest`). +/// +/// `full` is sent as a query parameter; the remaining fields form the JSON body. +#[derive(Debug, Clone, Default, Serialize)] +pub struct QueryUserFeedbackRequest { + /// Return full feedback records. Sent as a query parameter. + #[serde(skip)] + pub full: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub limit: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub next: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub prev: Option, + #[serde(skip_serializing_if = "Vec::is_empty")] + pub sort: Vec, + #[serde(skip_serializing_if = "std::collections::HashMap::is_empty")] + pub filter_conditions: CustomData, +} + +/// A single user feedback record (`UserFeedbackResponse`). +#[derive(Debug, Clone, Default, Deserialize)] +#[serde(default)] +pub struct UserFeedbackResponse { + pub cid: String, + pub rating: i32, + pub reason: String, + pub sdk: String, + pub sdk_version: String, + pub session_id: String, + pub user_id: String, + pub platform: Value, + pub custom: CustomData, +} + +/// `query_user_feedback` response (`QueryUserFeedbackResponse`). +#[derive(Debug, Clone, Default, Deserialize)] +#[serde(default)] +pub struct QueryUserFeedbackResponse { + pub duration: String, + pub user_feedback: Vec, + pub next: Option, + pub prev: Option, +} + +// Client call events + +/// A single client-side telemetry event (`ClientEvent`). +/// +/// Every field is optional; which fields are required depends on the event's +/// `stage`/`event_type`. See the serverside API reference for the per-stage +/// requirements. +#[derive(Debug, Clone, Default, Serialize)] +pub struct ClientEvent { + #[serde(skip_serializing_if = "Option::is_none")] + pub stage: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub stage_id: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub event_type: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub outcome: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub id: Option, + #[serde(rename = "type", skip_serializing_if = "Option::is_none")] + pub call_type: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub call_session_id: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub coordinator_connect_id: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub join_attempt_id: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub join_reason: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub elapsed_time: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub retry_count_attempt: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub retry_failure_code: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub retry_failure_reason: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub peer_connection: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub ice_state: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub sfu_id: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub was_previously_connected: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub previously_connected_timestamp: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub camera_permission_status: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub microphone_permission_status: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub screen_share_status: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub track_id: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub sdk_version: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub user_agent: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub user_id: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub timestamp: Option, +} + +/// `report_client_call_event` request (`ReportClientCallEventRequest`). +#[derive(Debug, Clone, Default, Serialize)] +pub struct ReportClientCallEventRequest { + /// Client-side events to report (1–100 per request). + pub events: Vec, +} + +/// `report_client_call_event` response (`ReportClientEventResponse`). +#[derive(Debug, Clone, Default, Deserialize)] +#[serde(default)] +pub struct ReportClientEventResponse { + pub duration: String, +} + +#[cfg(test)] +mod tests { + use serde_json::json; + + use super::*; + + #[test] + fn user_feedback_request_excludes_full_from_body() { + let value = serde_json::to_value(QueryUserFeedbackRequest { + full: Some(true), + limit: Some(10), + ..Default::default() + }) + .expect("request should serialize"); + assert_eq!(value, json!({ "limit": 10 })); + } + + #[test] + fn client_event_uses_type_wire_name_and_omits_absent_fields() { + let value = serde_json::to_value(ClientEvent { + stage: Some("JoinInitiated".to_owned()), + call_type: Some("default".to_owned()), + id: Some("call-1".to_owned()), + ..Default::default() + }) + .expect("event should serialize"); + assert_eq!( + value, + json!({ "stage": "JoinInitiated", "type": "default", "id": "call-1" }) + ); + } + + #[test] + fn aggregate_stats_request_omits_absent_optionals() { + let value = serde_json::to_value(QueryAggregateCallStatsRequest { + report_types: Some(vec!["call_duration_report".to_owned()]), + ..Default::default() + }) + .expect("request should serialize"); + assert_eq!(value, json!({ "report_types": ["call_duration_report"] })); + } + + #[test] + fn participant_sessions_response_tolerates_integer_duration() { + let response: QueryCallParticipantSessionsResponse = serde_json::from_value(json!({ + "duration": 12, + "call_id": "call-1", + "total_participant_sessions": 3 + })) + .expect("response should deserialize"); + assert_eq!(response.duration, 12); + assert_eq!(response.total_participant_sessions, 3); + } +} diff --git a/src/rtc/join/reconnect_runtime.rs b/src/rtc/join/reconnect_runtime.rs index 058505d..9f52a76 100644 --- a/src/rtc/join/reconnect_runtime.rs +++ b/src/rtc/join/reconnect_runtime.rs @@ -845,7 +845,7 @@ impl RtcCore { { return Err(join_cancelled()); } - connection.signal_tasks.drain(..).collect() + std::mem::take(&mut connection.signal_tasks) }; abort_tasks(old_tasks).await; { diff --git a/src/rtc/pcm/convert.rs b/src/rtc/pcm/convert.rs index f68a386..ab01f6d 100644 --- a/src/rtc/pcm/convert.rs +++ b/src/rtc/pcm/convert.rs @@ -65,8 +65,10 @@ impl PcmFrame { /// one byte. pub fn from_bytes(bytes: &[u8], sample_rate: u32, channels: u16) -> Self { let samples = bytes - .chunks_exact(2) - .map(|b| i16::from_le_bytes([b[0], b[1]])) + .as_chunks::<2>() + .0 + .iter() + .map(|&b| i16::from_le_bytes(b)) .collect(); Self::new(samples, sample_rate, channels) } diff --git a/src/video/call.rs b/src/video/call.rs index 4306487..4560543 100644 --- a/src/video/call.rs +++ b/src/video/call.rs @@ -10,6 +10,7 @@ use crate::error::{Error, Result}; use crate::models::*; const CALL_BASE: &str = "/api/v2/video/call/{type}/{id}"; +const CALL_STATS_BASE: &str = "/api/v2/video/call_stats/{type}/{id}"; const INTERNAL_RTC_TOKEN_LIFETIME: Duration = Duration::from_secs(10 * 60); /// A handle to a specific call (`:`). Cheap to construct; no request is @@ -52,7 +53,16 @@ impl Call { } fn path(&self, suffix: &str, extra: &[(&str, &str)]) -> String { - let template = format!("{CALL_BASE}{suffix}"); + self.path_from(CALL_BASE, suffix, extra) + } + + /// Same substitution as [`Self::path`] against the `call_stats` base. + fn stats_path(&self, suffix: &str, extra: &[(&str, &str)]) -> String { + self.path_from(CALL_STATS_BASE, suffix, extra) + } + + fn path_from(&self, base: &str, suffix: &str, extra: &[(&str, &str)]) -> String { + let template = format!("{base}{suffix}"); let mut params: Vec<(&str, &str)> = vec![("type", &self.call_type), ("id", &self.call_id)]; params.extend_from_slice(extra); Client::build_path(&template, ¶ms) @@ -650,12 +660,159 @@ impl Call { if let Some(value) = request.exclude_sfus { query.push(("exclude_sfus".to_owned(), value.to_string())); } - let path = Client::build_path( - "/api/v2/video/call_stats/{type}/{id}/{session_id}/map", + let path = self.stats_path("/{session_id}/map", &[("session_id", session_id)]); + self.client + .request::<(), _>(Method::GET, &path, &query, None) + .await + } + + /// Retrieve per-participant session metrics for one participant session. + /// + /// `GET .../call/{type}/{id}/session/{session}/participant/{user}/{user_session}/details/track` + pub async fn get_call_participant_session_metrics( + &self, + session: &str, + user: &str, + user_session: &str, + request: GetCallParticipantSessionMetricsRequest, + ) -> Result { + let mut query = Vec::new(); + push_opt( + &mut query, + "since", + request.since.as_ref().map(timestamp_query), + ); + push_opt( + &mut query, + "until", + request.until.as_ref().map(timestamp_query), + ); + let path = self.path( + "/session/{session}/participant/{user}/{user_session}/details/track", &[ - ("type", &self.call_type), - ("id", &self.call_id), - ("session_id", session_id), + ("session", session), + ("user", user), + ("user_session", user_session), + ], + ); + self.client + .request::<(), _>(Method::GET, &path, &query, None) + .await + } + + /// List participant sessions for one call session. + /// + /// `GET .../call/{type}/{id}/session/{session}/participant_sessions` + pub async fn query_call_participant_sessions( + &self, + session: &str, + request: QueryCallParticipantSessionsRequest, + ) -> Result { + let mut query = Vec::new(); + push_opt(&mut query, "limit", request.limit); + push_opt(&mut query, "prev", request.prev); + push_opt(&mut query, "next", request.next); + if let Some(encoded) = filter_conditions_query(&request.filter_conditions)? { + query.push(("filter_conditions".to_owned(), encoded)); + } + let path = self.path( + "/session/{session}/participant_sessions", + &[("session", session)], + ); + self.client + .request::<(), _>(Method::GET, &path, &query, None) + .await + } + + /// Retrieve detailed participant stats time series for one participant session. + /// + /// `GET .../call_stats/{type}/{id}/{session}/participant/{user}/{user_session}/details` + pub async fn get_call_session_participant_stats_details( + &self, + session: &str, + user: &str, + user_session: &str, + request: GetCallSessionParticipantStatsDetailsRequest, + ) -> Result { + let mut query = Vec::new(); + push_opt( + &mut query, + "since", + request.since.as_ref().map(timestamp_query), + ); + push_opt( + &mut query, + "until", + request.until.as_ref().map(timestamp_query), + ); + push_opt(&mut query, "max_points", request.max_points); + let path = self.stats_path( + "/{session}/participant/{user}/{user_session}/details", + &[ + ("session", session), + ("user", user), + ("user_session", user_session), + ], + ); + self.client + .request::<(), _>(Method::GET, &path, &query, None) + .await + } + + /// Query participant stats for one call session. + /// + /// `GET .../call_stats/{type}/{id}/{session}/participants` + pub async fn query_call_session_participant_stats( + &self, + session: &str, + request: QueryCallSessionParticipantStatsRequest, + ) -> Result { + let mut query = Vec::new(); + push_opt(&mut query, "limit", request.limit); + push_opt(&mut query, "prev", request.prev); + push_opt(&mut query, "next", request.next); + if let Some(encoded) = sort_query(&request.sort)? { + query.push(("sort".to_owned(), encoded)); + } + if let Some(encoded) = filter_conditions_query(&request.filter_conditions)? { + query.push(("filter_conditions".to_owned(), encoded)); + } + let path = self.stats_path("/{session}/participants", &[("session", session)]); + self.client + .request::<(), _>(Method::GET, &path, &query, None) + .await + } + + /// Retrieve the participant stats timeline for one participant session. + /// + /// `GET .../call_stats/{type}/{id}/{session}/participants/{user}/{user_session}/timeline` + pub async fn get_call_session_participant_stats_timeline( + &self, + session: &str, + user: &str, + user_session: &str, + request: GetCallSessionParticipantStatsTimelineRequest, + ) -> Result { + let mut query = Vec::new(); + push_opt( + &mut query, + "start_time", + request.start_time.as_ref().map(timestamp_query), + ); + push_opt( + &mut query, + "end_time", + request.end_time.as_ref().map(timestamp_query), + ); + if !request.severity.is_empty() { + query.push(("severity".to_owned(), request.severity.join(","))); + } + let path = self.stats_path( + "/{session}/participants/{user}/{user_session}/timeline", + &[ + ("session", session), + ("user", user), + ("user_session", user_session), ], ); self.client @@ -879,3 +1036,83 @@ fn timestamp_query(value: &Timestamp) -> String { .map(str::to_owned) .unwrap_or_else(|| value.to_string()) } + +/// Push `name=value` when the option is set, stringifying the value. +fn push_opt(query: &mut Vec<(String, String)>, name: &str, value: Option) { + if let Some(value) = value { + query.push((name.to_owned(), value.to_string())); + } +} + +/// JSON-encode a `sort` list for a query parameter, or `None` when empty. +/// +/// The coordinator parses this parameter as a JSON array; comma-joining the +/// encoded entries instead yields `is not a valid JSON for field 'sort'`. +fn sort_query(sort: &[SortParamRequest]) -> Result> { + if sort.is_empty() { + return Ok(None); + } + Ok(Some(serde_json::to_string(sort)?)) +} + +/// JSON-encode a `filter_conditions` map for a query parameter, or `None` when +/// empty. Matches the getstream-go query encoding for map-valued params. +fn filter_conditions_query(filter: &CustomData) -> Result> { + if filter.is_empty() { + return Ok(None); + } + Ok(Some(serde_json::to_string(filter)?)) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn empty_stats_query_params_are_omitted() { + assert!( + filter_conditions_query(&CustomData::new()) + .expect("ok") + .is_none() + ); + } + + #[test] + fn sort_query_encodes_a_json_array() { + assert!(sort_query(&[]).expect("ok").is_none()); + + let sort = vec![ + SortParamRequest { + field: Some("quality_score".to_owned()), + direction: Some(-1), + }, + SortParamRequest { + field: Some("user_id".to_owned()), + direction: Some(1), + }, + ]; + let encoded = sort_query(&sort).expect("ok").expect("some"); + + // Must parse as a JSON array: the coordinator rejects comma-joined + // objects with "is not a valid JSON for field 'sort'". + let parsed: serde_json::Value = + serde_json::from_str(&encoded).expect("sort query must be valid JSON"); + assert_eq!( + parsed, + serde_json::json!([ + {"direction": -1, "field": "quality_score"}, + {"direction": 1, "field": "user_id"}, + ]) + ); + } + + #[test] + fn filter_conditions_query_json_encodes_map() { + let mut filter = CustomData::new(); + filter.insert("call_cid".to_owned(), serde_json::json!("default:c1")); + assert_eq!( + filter_conditions_query(&filter).expect("ok"), + Some("{\"call_cid\":\"default:c1\"}".to_owned()) + ); + } +} diff --git a/src/video/mod.rs b/src/video/mod.rs index 8eca82c..9704bbb 100644 --- a/src/video/mod.rs +++ b/src/video/mod.rs @@ -1,6 +1,7 @@ //! Video coordinator REST: [`VideoClient`] and [`Call`]. mod call; +mod stats; pub use call::Call; diff --git a/src/video/stats.rs b/src/video/stats.rs new file mode 100644 index 0000000..26ea2e1 --- /dev/null +++ b/src/video/stats.rs @@ -0,0 +1,107 @@ +//! Application-level call statistics and reporting endpoints on [`VideoClient`]. + +use reqwest::Method; + +use super::VideoClient; +use crate::error::Result; +use crate::models::{ + GetActiveCallsStatusResponse, GetDailyDigestRequest, GetDailyDigestResponse, + QueryAggregateCallStatsRequest, QueryAggregateCallStatsResponse, QueryCallSessionStatsRequest, + QueryCallSessionStatsResponse, QueryUserFeedbackRequest, QueryUserFeedbackResponse, + ReportClientCallEventRequest, ReportClientEventResponse, +}; + +impl VideoClient { + /// Get the status of all active calls with metrics and summary + /// (`GET /api/v2/video/active_calls_status`). + pub async fn get_active_calls_status(&self) -> Result { + self.client + .request::<(), _>(Method::GET, "/api/v2/video/active_calls_status", &[], None) + .await + } + + /// Query aggregate call stats reports (`POST /api/v2/video/stats`). + pub async fn query_aggregate_call_stats( + &self, + request: QueryAggregateCallStatsRequest, + ) -> Result { + self.client + .request(Method::POST, "/api/v2/video/stats", &[], Some(&request)) + .await + } + + /// Query per-session call stats with filter/sort/pagination + /// (`POST /api/v2/video/call_stats`). + pub async fn query_call_session_stats( + &self, + request: QueryCallSessionStatsRequest, + ) -> Result { + self.client + .request( + Method::POST, + "/api/v2/video/call_stats", + &[], + Some(&request), + ) + .await + } + + /// Get the per-broadcast daily digest bundle for one UTC day + /// (`GET /api/v2/video/stats/daily_digest`). + pub async fn get_daily_digest( + &self, + request: GetDailyDigestRequest, + ) -> Result { + let mut query: Vec<(String, String)> = Vec::new(); + if let Some(date) = request.date { + query.push(("date".to_owned(), date)); + } + if let Some(target_app_id) = request.target_app_id { + query.push(("target_app_id".to_owned(), target_app_id)); + } + self.client + .request::<(), _>( + Method::GET, + "/api/v2/video/stats/daily_digest", + &query, + None, + ) + .await + } + + /// Query user feedback with filter/sort/pagination + /// (`POST /api/v2/video/call/feedback`). + pub async fn query_user_feedback( + &self, + request: QueryUserFeedbackRequest, + ) -> Result { + let query = request + .full + .map(|full| vec![("full".to_owned(), full.to_string())]) + .unwrap_or_default(); + self.client + .request( + Method::POST, + "/api/v2/video/call/feedback", + &query, + Some(&request), + ) + .await + } + + /// Report a batch of client-side telemetry events + /// (`POST /api/v2/video/call_client_event`). + pub async fn report_client_call_event( + &self, + request: ReportClientCallEventRequest, + ) -> Result { + self.client + .request( + Method::POST, + "/api/v2/video/call_client_event", + &[], + Some(&request), + ) + .await + } +} diff --git a/tests/video_stats.rs b/tests/video_stats.rs new file mode 100644 index 0000000..0d8b491 --- /dev/null +++ b/tests/video_stats.rs @@ -0,0 +1,214 @@ +//! Live integration tests for the advanced call-stats REST surface. +//! +//! Run with credentials present (repo `.env`): `cargo test`. Without credentials +//! the tests print a SKIP line and pass without touching the API. The call +//! created for the session-scoped queries is deleted on every exit path. +//! +//! The per-session queries are keyed by the *coordinator* call session id +//! (`get().call.session.id`), not by `Call::session_id()` -- the latter is this +//! participant's SFU session, which these endpoints report as `user_session_id` +//! nested inside the payload. Passing the wrong one returns `404 call session +//! not found`. Stats can lag the call by a few seconds, so the queries retry on +//! `404` for a bounded window and then fail rather than skipping. + +mod common; + +use std::future::Future; +use std::time::Duration; + +use getstream::models::{ + CallRequest, CustomData, DeleteCallRequest, GetOrCreateCallRequest, + QueryCallParticipantSessionsRequest, QueryCallSessionParticipantStatsRequest, UserRequest, +}; +use getstream::rtc::JoinCallData; + +/// Active-calls status is an application-level read with structural invariants +/// that do not depend on analytics timing. +#[tokio::test] +async fn active_calls_status_has_consistent_summary() { + let Some(client) = common::client_or_skip() else { + return; + }; + + let status = client + .video() + .get_active_calls_status() + .await + .expect("get_active_calls_status failed"); + + if let Some(summary) = status.summary { + assert!( + summary.active_calls >= 0, + "active_calls must be non-negative" + ); + assert!( + summary.participants >= 0, + "participants must be non-negative" + ); + assert!( + summary.active_publishers >= 0 && summary.active_subscribers >= 0, + "publisher/subscriber counts must be non-negative" + ); + } +} + +/// Create a call, join a real session, then query its per-session participant +/// stats. Identity fields must echo the call, and the participant session must +/// carry this join's SFU session id. +#[tokio::test] +async fn session_scoped_participant_stats_echo_call_identity() { + let Some(client) = common::client_or_skip() else { + return; + }; + + let user_id = common::unique_id("rust-it-stats-user"); + let call_id = common::unique_id("rust-it-stats-call"); + client + .upsert_users([UserRequest::new(&user_id)]) + .await + .expect("upsert_users failed"); + + let call = client.video().call("default", &call_id); + call.get_or_create(GetOrCreateCallRequest { + data: Some(CallRequest { + created_by_id: Some(user_id.clone()), + ..Default::default() + }), + ..Default::default() + }) + .await + .expect("get_or_create failed"); + + let outcome: Result<(), String> = async { + call.join(JoinCallData::new(&user_id)) + .await + .map_err(|error| format!("join failed: {error}"))?; + + // `session_id()` is this participant's SFU session; the stats endpoints + // are keyed by the coordinator's call session, which is a different id. + let user_session_id = call + .session_id() + .await + .ok_or_else(|| "joined call did not expose an SFU session id".to_owned())?; + let session_id = call + .get(Default::default()) + .await + .map_err(|error| format!("get failed: {error}"))? + .call + .session + .map(|session| session.id) + .ok_or_else(|| "joined call did not expose a call session".to_owned())?; + + call.leave() + .await + .map_err(|error| format!("leave failed: {error}"))?; + call.end() + .await + .map_err(|error| format!("end failed: {error}"))?; + + // Independent reads of the same ended session: no need to serialise + // their retry windows. + let (stats, sessions) = tokio::try_join!( + await_stats("query_call_session_participant_stats", || { + call.query_call_session_participant_stats( + &session_id, + QueryCallSessionParticipantStatsRequest { + // Populated so the query encoding is validated against + // the server, not just against its own unit test. + limit: Some(5), + filter_conditions: participant_filter(&user_id), + ..Default::default() + }, + ) + }), + await_stats("query_call_participant_sessions", || { + call.query_call_participant_sessions( + &session_id, + QueryCallParticipantSessionsRequest::default(), + ) + }), + )?; + assert_eq!(stats.call_id, call_id, "participant stats call_id mismatch"); + assert_eq!( + stats.call_type, "default", + "participant stats type mismatch" + ); + assert_eq!( + stats.call_session_id, session_id, + "participant stats session mismatch" + ); + + assert_eq!( + sessions.call_id, call_id, + "participant sessions call_id mismatch" + ); + assert_eq!( + sessions.call_type, "default", + "participant sessions type mismatch" + ); + assert_eq!( + sessions.call_session_id, session_id, + "participant sessions session mismatch" + ); + + // The join above is the only participant session, and it must be + // reported under the SFU session id -- guarding the two ids from being + // conflated again. + let reported: Vec<&str> = sessions + .participants_sessions + .iter() + .filter_map(|entry| entry.get("user_session_id")?.as_str()) + .collect(); + assert!( + reported.contains(&user_session_id.as_str()), + "participant sessions {reported:?} missing this join's SFU session {user_session_id}" + ); + + Ok(()) + } + .await; + + let leave_cleanup = call.leave().await; + let delete_cleanup = call.delete(DeleteCallRequest { hard: Some(true) }).await; + if let Err(error) = outcome { + panic!("{error}; leave cleanup: {leave_cleanup:?}; delete cleanup: {delete_cleanup:?}"); + } + delete_cleanup.expect("delete cleanup failed"); +} + +/// Restrict a participant-stats query to one user, exercising the +/// `filter_conditions` query encoding against the server rather than only +/// against the encoder's own unit test. +fn participant_filter(user_id: &str) -> CustomData { + CustomData::from([("user_id".to_owned(), serde_json::Value::from(user_id))]) +} + +/// Analytics can trail the call by a few seconds, so retry a `404` for a bounded +/// window. Unlike an unconditional skip, a persistent `404` still fails the test. +async fn await_stats(endpoint: &str, mut query: F) -> Result +where + F: FnMut() -> Fut, + Fut: Future>, +{ + const DEADLINE: Duration = Duration::from_secs(30); + const INTERVAL: Duration = Duration::from_secs(3); + + let mut waited = Duration::ZERO; + loop { + match query().await { + Ok(value) => return Ok(value), + Err(error) + if waited < DEADLINE + && error + .as_api_error() + .is_some_and(|api_error| api_error.status == 404) => + { + tokio::time::sleep(INTERVAL).await; + waited += INTERVAL; + } + Err(error) => { + return Err(format!("{endpoint} failed after {waited:?}: {error}")); + } + } + } +}