videosdk/resources/

sessions.rs

//! The sessions API: live and past sessions, and their participants.

use std::sync::Arc;

use futures_util::Stream;
use reqwest::Method;
use serde::{Deserialize, Serialize};
use serde_json::{Map, Value};

use crate::client::{CallOptions, Client};
use crate::common::{string_enum, ResourceLinks};
use crate::error::{Error, Result};
use crate::pagination::{auto_page, paginate, ListParams, Page, PageFetcher};
use crate::query::QueryBuilder;
use crate::resources::escape;
use crate::resources::session_utils::reshape_session_participants;

const PATH: &str = "/v2/sessions";

string_enum! {
    /// A session's lifecycle state.
    ///
    /// This rides on every [`Session`] response, so it is an open string newtype:
    /// a closed enum would fail to deserialize — taking the whole session with
    /// it — the moment the API adds a state this SDK predates.
    SessionStatus {
        /// The session is live.
        ONGOING => "ongoing",
        /// The session has ended.
        ENDED => "ended",
    }
}

/// A loosely-typed quality-of-service statistics payload.
///
/// The shape varies by endpoint and is not modelled.
pub type QualityStats = Value;

/// A single join/leave interval for a participant.
#[derive(Debug, Clone, Deserialize)]
pub struct ParticipantTimelog {
    /// When the participant joined.
    #[serde(default, deserialize_with = "crate::common::null_to_default")]
    pub start: String,
    /// When the participant left, or `None` if they are still present.
    pub end: Option<String>,
}

/// A participant within a session.
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct Participant {
    /// The participant's id.
    pub participant_id: String,
    /// The participant's display name.
    #[serde(default, deserialize_with = "crate::common::null_to_default")]
    pub name: String,
    /// Every join/leave interval for this participant.
    #[serde(default, deserialize_with = "crate::common::null_to_default")]
    pub timelog: Vec<ParticipantTimelog>,
    /// Any fields the server returned that this SDK does not model yet.
    #[serde(flatten)]
    pub extra: Map<String, Value>,
}

/// A session: a single live occupancy of a room.
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct Session {
    /// The session id.
    pub id: String,
    /// The room this session belongs to.
    ///
    /// Always populated: [`SessionsResource::end`] folds the server's
    /// `meetingId` alias into it.
    #[serde(default, deserialize_with = "crate::common::null_to_default")]
    pub room_id: String,
    /// The room's caller-supplied stable id, if one was given.
    pub custom_room_id: Option<String>,
    /// When the session started.
    #[serde(default, deserialize_with = "crate::common::null_to_default")]
    pub start: String,
    /// When the session ended, or `None` if it is still live.
    pub end: Option<String>,
    /// The session's lifecycle state.
    pub status: Option<SessionStatus>,
    /// The participants seen in this session.
    #[serde(default, deserialize_with = "crate::common::null_to_default")]
    pub participants: Vec<Participant>,
    /// The region the session ran in.
    #[serde(default, deserialize_with = "crate::common::null_to_default")]
    pub region: String,
    /// HATEOAS-style links to related resources.
    #[serde(default, deserialize_with = "crate::common::null_to_default")]
    pub links: ResourceLinks,
    /// Any fields the server returned that this SDK does not model yet.
    #[serde(flatten)]
    pub extra: Map<String, Value>,
}

/// The query parameters for [`SessionsResource::list`].
#[derive(Debug, Clone, Default)]
pub struct SessionListParams {
    /// The 1-based page number.
    pub page: Option<u32>,
    /// Items per page.
    pub per_page: Option<u32>,
    /// An opaque cursor from a previous page. Overrides `page` and `per_page`.
    pub cursor: Option<String>,
    /// An exact match on the room's `room_id` or `custom_room_id`.
    pub query: Option<String>,
    /// Filters by room id.
    pub room_id: Option<String>,
    /// Filters by custom room id.
    pub custom_room_id: Option<String>,
    /// Scopes to a specific user. Admin tokens only.
    pub user_id: Option<String>,
    /// Filters by session start, in epoch milliseconds. Start of the range.
    pub start_date: Option<i64>,
    /// Filters by session start, in epoch milliseconds. End of the range.
    pub end_date: Option<i64>,
    /// Filters by lifecycle state.
    pub status: Option<SessionStatus>,
}

impl SessionListParams {
    fn pagination(&self) -> ListParams {
        ListParams {
            page: self.page,
            per_page: self.per_page,
            cursor: self.cursor.clone(),
        }
    }
}

/// The parameters for [`SessionsResource::end`].
#[derive(Debug, Clone, Default)]
pub struct SessionEndParams {
    /// The room id. One of `room_id` or `meeting_id` is required.
    pub room_id: Option<String>,
    /// An alias for `room_id`.
    pub meeting_id: Option<String>,
    /// Restricts the action to a specific session.
    pub session_id: Option<String>,
    /// Force-kills the session.
    pub force: Option<bool>,
    /// Tolerates an already-closed session.
    pub ignore_closed: Option<bool>,
}

#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct SessionEndWire<'a> {
    room_id: &'a str,
    #[serde(skip_serializing_if = "Option::is_none")]
    session_id: Option<&'a str>,
    #[serde(skip_serializing_if = "Option::is_none")]
    ignore_closed: Option<bool>,
    #[serde(rename = "vsdkForceKill", skip_serializing_if = "Option::is_none")]
    force: Option<bool>,
}

/// The parameters for [`SessionsResource::remove_participant`].
#[derive(Debug, Clone, Default)]
pub struct SessionRemoveParticipantParams {
    /// The participant to remove. Required.
    pub participant_id: String,
    /// The room id. One of `room_id` or `session_id` is required.
    pub room_id: Option<String>,
    /// Restricts the action to a specific session.
    pub session_id: Option<String>,
}

#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct RemoveParticipantWire<'a> {
    participant_id: &'a str,
    #[serde(skip_serializing_if = "Option::is_none")]
    room_id: Option<&'a str>,
    #[serde(skip_serializing_if = "Option::is_none")]
    session_id: Option<&'a str>,
}

/// The sessions API. Reached via [`Client::sessions`].
#[derive(Debug, Clone, Copy)]
pub struct SessionsResource<'a> {
    client: &'a Client,
}

impl<'a> SessionsResource<'a> {
    pub(crate) fn new(client: &'a Client) -> Self {
        Self { client }
    }

    /// Lists sessions, one page at a time.
    pub async fn list(&self, params: SessionListParams) -> Result<Page<Session>> {
        paginate(self.fetcher(&params), &params.pagination(), "data", None).await
    }

    /// Lists sessions, transparently fetching every page.
    pub fn list_stream(
        &self,
        params: SessionListParams,
    ) -> impl Stream<Item = Result<Session>> + Send {
        auto_page(self.fetcher(&params), params.pagination(), "data", None)
    }

    /// Fetches a session by id.
    pub async fn get(&self, session_id: &str) -> Result<Session> {
        let path = format!("{PATH}/{}", escape(session_id));
        self.client
            .json(Method::GET, &path, CallOptions::new())
            .await
    }

    /// Lists every participant seen in a session, one page at a time.
    pub async fn list_participants(
        &self,
        session_id: &str,
        params: ListParams,
    ) -> Result<Page<Participant>> {
        let path = format!("{PATH}/{}/participants", escape(session_id));
        paginate(
            participants_fetcher(self.client, path),
            &params,
            "data",
            None,
        )
        .await
    }

    /// Lists a session's currently-active participants, one page at a time.
    pub async fn list_active_participants(
        &self,
        session_id: &str,
        params: ListParams,
    ) -> Result<Page<Participant>> {
        let path = format!("{PATH}/{}/participants/active", escape(session_id));
        paginate(
            participants_fetcher(self.client, path),
            &params,
            "data",
            None,
        )
        .await
    }

    /// Fetches a single participant within a session.
    pub async fn get_participant(
        &self,
        session_id: &str,
        participant_id: &str,
    ) -> Result<Participant> {
        let path = format!(
            "{PATH}/{}/participants/{}",
            escape(session_id),
            escape(participant_id)
        );
        self.client
            .json(Method::GET, &path, CallOptions::new())
            .await
    }

    /// Ends a live session and returns the now-ended session.
    pub async fn end(&self, params: SessionEndParams) -> Result<Session> {
        let room_id = params
            .room_id
            .as_deref()
            .or(params.meeting_id.as_deref())
            .filter(|id| !id.is_empty())
            .ok_or_else(|| Error::validation("sessions.end requires room_id (or meeting_id)"))?;

        // The server resolves the room only via `roomId`; fold the alias into it.
        let body = SessionEndWire {
            room_id,
            session_id: params.session_id.as_deref(),
            ignore_closed: params.ignore_closed,
            force: params.force,
        };

        let path = format!("{PATH}/end");
        let session: Session = self
            .client
            .json(Method::POST, &path, CallOptions::json(&body)?)
            .await?;
        Ok(normalize_ended_session(session))
    }

    /// Removes (kicks) a participant from a live session, returning the server's
    /// confirmation message.
    pub async fn remove_participant(
        &self,
        params: SessionRemoveParticipantParams,
    ) -> Result<String> {
        let room_id = params.room_id.as_deref().filter(|id| !id.is_empty());
        let session_id = params.session_id.as_deref().filter(|id| !id.is_empty());
        if room_id.is_none() && session_id.is_none() {
            return Err(Error::validation(
                "sessions.remove_participant requires room_id or session_id to locate the session",
            ));
        }

        let body = RemoveParticipantWire {
            participant_id: &params.participant_id,
            room_id,
            session_id,
        };
        self.client
            .message(
                Method::POST,
                "/v2/sessions/participants/remove",
                CallOptions::json(&body)?,
            )
            .await
    }

    /// Fetches aggregated session-level quality stats.
    pub async fn get_stats(&self, session_id: &str) -> Result<QualityStats> {
        let path = format!("{PATH}/{}/stats", escape(session_id));
        self.client
            .json(Method::GET, &path, CallOptions::new())
            .await
    }

    /// Fetches per-participant quality stats, plus resolution usage.
    pub async fn get_participant_stats(
        &self,
        session_id: &str,
        participant_id: &str,
    ) -> Result<QualityStats> {
        let path = format!(
            "{PATH}/{}/participant/{}/stats",
            escape(session_id),
            escape(participant_id)
        );
        self.client
            .json(Method::GET, &path, CallOptions::new())
            .await
    }

    fn fetcher(&self, params: &SessionListParams) -> PageFetcher {
        let client = self.client.clone();
        let params = params.clone();

        Arc::new(move |page, per_page| {
            let client = client.clone();
            let params = params.clone();
            Box::pin(async move {
                let query = QueryBuilder::new()
                    .opt("page", page)
                    .opt("perPage", per_page)
                    .opt_str("query", params.query.as_deref())
                    .opt_str("roomId", params.room_id.as_deref())
                    .opt_str("customRoomId", params.custom_room_id.as_deref())
                    .opt_str("userId", params.user_id.as_deref())
                    .opt("startDate", params.start_date)
                    .opt("endDate", params.end_date)
                    .opt_str("status", params.status.as_ref().map(SessionStatus::as_str))
                    .into_pairs();
                client
                    .json::<Value>(Method::GET, PATH, CallOptions::new().query(query))
                    .await
            })
        })
    }
}

/// Builds a fetcher that lifts participants out of the session-shaped envelope.
pub(crate) fn participants_fetcher(client: &Client, path: String) -> PageFetcher {
    let client = client.clone();
    Arc::new(move |page, per_page| {
        let client = client.clone();
        let path = path.clone();
        Box::pin(async move {
            let query = QueryBuilder::new()
                .opt("page", page)
                .opt("perPage", per_page)
                .into_pairs();
            let raw: Value = client
                .json(Method::GET, &path, CallOptions::new().query(query))
                .await?;
            Ok(reshape_session_participants(raw))
        })
    })
}

/// `POST /v2/sessions/end` answers with the legacy `transformForUser` shape:
/// `meetingId`/`userMeetingId` instead of `roomId`/`customRoomId`, and no
/// `status`. Normalize it to match every other session response.
fn normalize_ended_session(mut session: Session) -> Session {
    if session.room_id.is_empty() {
        if let Some(meeting_id) = session.extra.get("meetingId").and_then(Value::as_str) {
            session.room_id = meeting_id.to_string();
        }
    }
    if session.custom_room_id.is_none() {
        session.custom_room_id = session
            .extra
            .get("userMeetingId")
            .and_then(Value::as_str)
            .map(str::to_string);
    }
    if session.status.is_none() {
        session.status = Some(if session.end.is_none() {
            SessionStatus::ONGOING
        } else {
            SessionStatus::ENDED
        });
    }
    session
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;

    fn session(value: Value) -> Session {
        serde_json::from_value(value).unwrap()
    }

    #[test]
    fn end_wire_renames_force_to_vsdk_force_kill() {
        let body = serde_json::to_value(SessionEndWire {
            room_id: "r-1",
            session_id: None,
            ignore_closed: None,
            force: Some(true),
        })
        .unwrap();
        assert_eq!(body, json!({"roomId": "r-1", "vsdkForceKill": true}));
    }

    #[test]
    fn end_wire_omits_absent_options() {
        let body = serde_json::to_value(SessionEndWire {
            room_id: "r-1",
            session_id: Some("s-1"),
            ignore_closed: Some(false),
            force: None,
        })
        .unwrap();
        assert_eq!(
            body,
            json!({"roomId": "r-1", "sessionId": "s-1", "ignoreClosed": false})
        );
    }

    #[test]
    fn ended_session_folds_the_meeting_id_aliases() {
        let normalized = normalize_ended_session(session(json!({
            "id": "s-1",
            "meetingId": "abcd-efgh-ijkl",
            "userMeetingId": "my-room",
            "end": "2026-01-01T00:00:00Z",
        })));
        assert_eq!(normalized.room_id, "abcd-efgh-ijkl");
        assert_eq!(normalized.custom_room_id.as_deref(), Some("my-room"));
        assert_eq!(normalized.status, Some(SessionStatus::ENDED));
    }

    #[test]
    fn ended_session_derives_ongoing_when_end_is_absent() {
        let normalized = normalize_ended_session(session(json!({"id": "s-1", "roomId": "r-1"})));
        assert_eq!(normalized.status, Some(SessionStatus::ONGOING));
        assert_eq!(
            normalized.room_id, "r-1",
            "a real roomId is not overwritten"
        );
    }

    #[test]
    fn ended_session_keeps_a_server_supplied_status() {
        let normalized = normalize_ended_session(session(json!({
            "id": "s-1", "roomId": "r-1", "status": "ongoing", "end": "2026-01-01T00:00:00Z",
        })));
        assert_eq!(normalized.status, Some(SessionStatus::ONGOING));
    }

    #[test]
    fn session_status_round_trips() {
        assert_eq!(
            serde_json::to_value(SessionStatus::ONGOING).unwrap(),
            json!("ongoing")
        );
        assert_eq!(SessionStatus::ENDED.as_str(), "ended");
    }
}