videosdk-server-sdk 0.1.0

//! WHIP publish, WHEP playback, and socket ingest.

use std::time::Duration;

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

use crate::client::{CallOptions, Client};
use crate::error::{Error, Result};
use crate::query::QueryBuilder;
use crate::resources::escape;
use crate::resources::random_uuid;
use crate::resources::session_utils::resolve_active_session_id;

/// Builds `<base><path>?<query>`, including only non-empty parameters.
fn build_ingress_url(base: &str, path: &str, params: &[(&str, &str)]) -> Result<String> {
    let mut url = Url::parse(&format!("{}{path}", base.trim_end_matches('/')))
        .map_err(|e| Error::config(format!("invalid base URL {base:?}: {e}")))?;

    let pairs: Vec<_> = params
        .iter()
        .filter(|(_, value)| !value.is_empty())
        .collect();
    if !pairs.is_empty() {
        let mut query = url.query_pairs_mut();
        for (name, value) in pairs {
            query.append_pair(name, value);
        }
    }
    Ok(url.into())
}

/* ------------------------------------ WHIP ------------------------------------ */

/// The parameters for [`WhipResource::create`].
#[derive(Debug, Clone, Default)]
pub struct CreateWhipParams {
    /// The publisher's participant id. Auto-generated as `whip-<uuid>` when absent.
    pub participant_id: Option<String>,
    /// The display name shown for the publisher.
    pub name: Option<String>,
    /// How long the publish credential stays valid. Defaults to one hour.
    pub expires_in: Option<Duration>,
    /// Reuses an existing participant instead of creating a new one.
    pub use_existing_peer: Option<bool>,
}

/// A WHIP ingest credential.
///
/// [`WhipResource::create`] builds this locally, with no network call: `POST`
/// your SDP offer to [`url`](WhipIngress::url) with `Content-Type:
/// application/sdp` and `Authorization: <token>`, and the server replies with
/// the SDP answer.
#[derive(Debug, Clone)]
pub struct WhipIngress {
    /// The WHIP endpoint. POST the SDP offer here.
    pub url: String,
    /// Authorizes the publish. Short-lived and sensitive — do not log it.
    pub token: String,
    /// An alias of [`token`](WhipIngress::token). Some WHIP tools call it the
    /// stream key.
    pub stream_key: String,
    /// The room being published into.
    pub room_id: String,
    /// The publisher's participant id.
    pub participant_id: String,
    /// The publisher's display name.
    pub display_name: Option<String>,
    /// Targets a specific session on delete, e.g. from the publish `Location`
    /// header. The room's active session is resolved when absent.
    pub session_id: Option<String>,
}

/// WHIP ingress: WebRTC-HTTP publish into a room. Reached via [`Client::whip`].
#[derive(Debug, Clone, Copy)]
pub struct WhipResource<'a> {
    client: &'a Client,
}

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

    /// Builds a WHIP publish credential.
    ///
    /// Synchronous: this mints a token and assembles a URL locally, and makes no
    /// network call.
    pub fn create(&self, room_id: &str, params: CreateWhipParams) -> Result<WhipIngress> {
        let participant_id = params
            .participant_id
            .unwrap_or_else(|| format!("whip-{}", random_uuid()));
        let token = self
            .client
            .mint_api_token(params.expires_in.unwrap_or_default())?;

        let url = build_ingress_url(
            self.client.base_url(),
            "/v2/whip",
            &[
                ("roomId", room_id),
                ("participantId", &participant_id),
                ("displayName", params.name.as_deref().unwrap_or("")),
                (
                    "useExistingPeer",
                    if params.use_existing_peer == Some(true) {
                        "true"
                    } else {
                        ""
                    },
                ),
            ],
        )?;

        Ok(WhipIngress {
            url,
            stream_key: token.clone(),
            token,
            room_id: room_id.to_string(),
            participant_id,
            display_name: params.name,
            session_id: None,
        })
    }

    /// Tears down a WHIP publisher.
    ///
    /// The target's `session_id` is used when set; otherwise the room's active
    /// session is resolved.
    pub async fn delete(&self, target: &WhipIngress) -> Result<()> {
        let session_id = match &target.session_id {
            Some(session_id) => session_id.clone(),
            None => resolve_active_session_id(self.client, &target.room_id)
                .await?
                .ok_or_else(|| {
                    Error::not_found(format!("no active session for room {}", target.room_id))
                })?,
        };

        let query = QueryBuilder::new()
            .str_val("participantId", &target.participant_id)
            .opt_str("displayName", target.display_name.as_deref())
            .into_pairs();
        let path = format!("/v2/whip/sessions/{}", escape(&session_id));
        self.client
            .none(Method::DELETE, &path, CallOptions::new().query(query))
            .await
    }
}

/* ------------------------------------ WHEP ------------------------------------ */

/// Selects the remote participant a WHEP subscriber pulls.
#[derive(Debug, Clone)]
pub struct WhepSource {
    /// The remote participant to pull.
    pub participant_id: String,
}

/// The parameters for [`WhepResource::create`].
#[derive(Debug, Clone, Default)]
pub struct CreateWhepParams {
    /// The subscriber's own participant id. Auto-generated as `whep-<uuid>` when
    /// absent.
    pub participant_id: Option<String>,
    /// Which remote participant to pull. Omit to let the server choose.
    pub source: Option<WhepSource>,
    /// How long the playback credential stays valid. Defaults to one hour.
    pub expires_in: Option<Duration>,
}

/// A WHEP playback credential.
///
/// [`WhepResource::create`] builds this locally, with no network call.
#[derive(Debug, Clone)]
pub struct WhepPlayback {
    /// The WHEP endpoint. POST the SDP offer here.
    pub url: String,
    /// Authorizes the pull. Short-lived and sensitive — do not log it.
    pub token: String,
    /// The room being pulled from.
    pub room_id: String,
    /// The subscriber's participant id.
    pub participant_id: String,
    /// The remote participant being pulled, when a source was given.
    pub remote_peer_id: Option<String>,
    /// Targets a specific session on delete.
    pub session_id: Option<String>,
}

/// WHEP egress: standards-based WebRTC pull from a room. Requires an active
/// session. Reached via [`Client::whep`].
#[derive(Debug, Clone, Copy)]
pub struct WhepResource<'a> {
    client: &'a Client,
}

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

    /// Builds a WHEP playback credential.
    ///
    /// Synchronous: this mints a token and assembles a URL locally, and makes no
    /// network call.
    pub fn create(&self, room_id: &str, params: CreateWhepParams) -> Result<WhepPlayback> {
        let participant_id = params
            .participant_id
            .unwrap_or_else(|| format!("whep-{}", random_uuid()));
        let remote_peer_id = params.source.map(|source| source.participant_id);
        let token = self
            .client
            .mint_api_token(params.expires_in.unwrap_or_default())?;

        let url = build_ingress_url(
            self.client.base_url(),
            "/v2/whep",
            &[
                ("roomId", room_id),
                ("participantId", &participant_id),
                ("remotePeerId", remote_peer_id.as_deref().unwrap_or("")),
            ],
        )?;

        Ok(WhepPlayback {
            url,
            token,
            room_id: room_id.to_string(),
            participant_id,
            remote_peer_id,
            session_id: None,
        })
    }

    /// Tears down a WHEP subscriber.
    pub async fn delete(&self, target: &WhepPlayback) -> Result<()> {
        let session_id = match &target.session_id {
            Some(session_id) => session_id.clone(),
            None => resolve_active_session_id(self.client, &target.room_id)
                .await?
                .ok_or_else(|| {
                    Error::not_found(format!("no active session for room {}", target.room_id))
                })?,
        };

        let query = QueryBuilder::new()
            .str_val("participantId", &target.participant_id)
            .opt_str("remotePeerId", target.remote_peer_id.as_deref())
            .into_pairs();
        let path = format!("/v2/whep/sessions/{}", escape(&session_id));
        self.client
            .none(Method::DELETE, &path, CallOptions::new().query(query))
            .await
    }
}

/* -------------------------------- socket ingest -------------------------------- */

/// Agent metadata to associate with an ingest session.
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct SocketIngressAgent {
    /// The agent id.
    pub id: String,
    /// The agent type.
    #[serde(rename = "type", skip_serializing_if = "Option::is_none")]
    pub kind: Option<String>,
    /// Free-form metadata.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub metadata: Option<Map<String, Value>>,
}

/// The parameters for [`SocketIngressResource::create`].
#[derive(Debug, Clone, Default)]
pub struct CreateSocketIngressParams {
    /// The streamed-in participant's id. Auto-generated when absent.
    pub participant_id: Option<String>,
    /// The participant's display name.
    pub name: Option<String>,
    /// Free-form participant metadata.
    pub metadata: Option<Map<String, Value>>,
    /// Agent metadata to associate with the session.
    pub agent: Option<SocketIngressAgent>,
    /// Where to run the ingest session.
    pub region: Option<String>,
}

#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct SocketIngressParticipant {
    #[serde(skip_serializing_if = "Option::is_none")]
    id: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    name: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    metadata: Option<Map<String, Value>>,
}

#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct SocketIngressWire<'a> {
    room_id: &'a str,
    #[serde(skip_serializing_if = "Option::is_none")]
    participant: Option<SocketIngressParticipant>,
    #[serde(skip_serializing_if = "Option::is_none")]
    agent: Option<SocketIngressAgent>,
    #[serde(skip_serializing_if = "Option::is_none")]
    region: Option<String>,
}

/// A socket-ingest session.
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct SocketIngress {
    /// The single-use WebSocket URL, valid for about 90 seconds. Connect and
    /// stream media or data frames.
    pub ws_url: String,
    /// The room being streamed into.
    #[serde(default, deserialize_with = "crate::common::null_to_default")]
    pub room_id: String,
    /// How long the URL stays valid, in seconds.
    #[serde(default, deserialize_with = "crate::common::null_to_default")]
    pub expires_in: u32,
    /// The single-use handle parsed out of [`ws_url`](SocketIngress::ws_url).
    #[serde(skip)]
    pub ws_ref: Option<String>,
    /// Any fields the server returned that this SDK does not model yet.
    #[serde(flatten)]
    pub extra: Map<String, Value>,
}

/// Reads the `ref` query parameter out of a WebSocket URL.
fn extract_ws_ref(ws_url: &str) -> Option<String> {
    let url = Url::parse(ws_url).ok()?;
    url.query_pairs()
        .find(|(name, _)| name == "ref")
        .map(|(_, value)| value.into_owned())
}

/// Streams media and data frames into a room over a single-use WebSocket. To
/// tear down, close the socket; the URL also expires on its own. Reached via
/// [`Client::socket_ingress`].
#[derive(Debug, Clone, Copy)]
pub struct SocketIngressResource<'a> {
    client: &'a Client,
}

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

    /// Creates a socket-ingest session for a room.
    pub async fn create(
        &self,
        room_id: &str,
        params: CreateSocketIngressParams,
    ) -> Result<SocketIngress> {
        let has_participant =
            params.participant_id.is_some() || params.name.is_some() || params.metadata.is_some();
        let participant = if has_participant {
            Some(SocketIngressParticipant {
                id: params.participant_id,
                name: params.name,
                metadata: params.metadata,
            })
        } else {
            None
        };

        let body = SocketIngressWire {
            room_id,
            participant,
            agent: params.agent,
            region: params.region,
        };

        let mut ingress: SocketIngress = self
            .client
            .data(
                Method::POST,
                "/v2/ingest/sessions",
                CallOptions::json(&body)?,
            )
            .await?;
        ingress.ws_ref = extract_ws_ref(&ingress.ws_url);
        Ok(ingress)
    }
}

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

    #[test]
    fn ingress_urls_omit_empty_parameters() {
        let url = build_ingress_url(
            "https://api.videosdk.live/",
            "/v2/whip",
            &[
                ("roomId", "r-1"),
                ("displayName", ""),
                ("useExistingPeer", ""),
            ],
        )
        .unwrap();
        assert_eq!(url, "https://api.videosdk.live/v2/whip?roomId=r-1");
    }

    #[test]
    fn ingress_urls_have_no_trailing_question_mark_when_empty() {
        let url = build_ingress_url("https://api.videosdk.live", "/v2/whip", &[("a", "")]).unwrap();
        assert_eq!(url, "https://api.videosdk.live/v2/whip");
    }

    #[test]
    fn ingress_urls_percent_encode_their_values() {
        let url =
            build_ingress_url("https://x.test", "/v2/whep", &[("displayName", "Ada L")]).unwrap();
        assert!(url.contains("displayName=Ada+L") || url.contains("displayName=Ada%20L"));
    }

    #[test]
    fn the_ws_ref_is_read_out_of_the_url() {
        assert_eq!(
            extract_ws_ref("wss://x.test/ingest?ref=abc123&other=1").as_deref(),
            Some("abc123")
        );
        assert_eq!(extract_ws_ref("wss://x.test/ingest"), None);
        assert_eq!(extract_ws_ref("not a url"), None);
    }
}