videosdk-server-sdk 0.1.0

//! The HLS API.

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::{
    CompositionConfig, LayoutPriority, ResourceLinks, TranscriptionConfig, WebhookDeliverySummary,
};
use crate::error::{Error, Result};
use crate::pagination::{auto_page, paginate, ListParams, Page, PageFetcher};
use crate::query::QueryBuilder;
use crate::resources::egress::{
    composition_to_config, to_egress_handle, Composition, EgressHandle, EgressType, StopTarget,
    StopWire,
};
use crate::resources::escape;

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

/// The image format of a thumbnail capture.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum HlsCaptureFormat {
    /// PNG.
    Png,
    /// JPEG.
    Jpg,
    /// WebP.
    Webp,
}

/// An HLS stream, or a finished HLS recording.
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct HlsStream {
    /// The stream id.
    pub id: String,
    /// The composer that produced the stream.
    pub composer_id: Option<String>,
    /// The room being streamed.
    pub room_id: Option<String>,
    /// The session being streamed.
    pub session_id: Option<String>,
    /// The stream's mode.
    pub mode: Option<String>,
    /// When the stream started.
    pub start: Option<String>,
    /// When the stream ended, or `None` if it is still running.
    pub end: Option<String>,
    /// The stream's quality tier.
    pub quality: Option<String>,
    /// The stream's orientation.
    pub orientation: Option<String>,
    /// The thumbnail captures taken from this stream.
    #[serde(default, deserialize_with = "crate::common::null_to_default")]
    pub capture_log: Vec<Value>,
    /// The stream's encryption settings.
    pub encryption: Option<Map<String, Value>>,
    /// The stream's duration, in seconds.
    pub duration: Option<f64>,
    /// A summary of the webhook deliveries for this stream.
    pub webhook: Option<WebhookDeliverySummary>,
    /// HATEOAS-style links to related resources.
    #[serde(default, deserialize_with = "crate::common::null_to_default")]
    pub links: ResourceLinks,
    /// The downstream URL.
    pub downstream_url: Option<String>,
    /// The HLS playback URL.
    pub playback_hls_url: Option<String>,
    /// The livestream URL.
    pub livestream_url: Option<String>,
    /// Any fields the server returned that this SDK does not model yet.
    #[serde(flatten)]
    pub extra: Map<String, Value>,
}

/// The parameters for [`HlsResource::start`].
#[derive(Debug, Clone, Default)]
pub struct HlsStartParams {
    /// The layout, quality, orientation and theme.
    pub composition: Option<Composition>,
    /// Transcribes and optionally summarizes the stream.
    pub transcription: Option<TranscriptionConfig>,
    /// A webhook to notify about the stream.
    pub webhook_url: Option<String>,
    /// A recording configuration to start alongside HLS.
    pub recording: Option<Value>,
    /// An advanced custom template URL.
    ///
    /// Prefer [`CompositionLayout::Custom`](crate::CompositionLayout::Custom),
    /// which overrides this when both are set.
    pub template_url: Option<String>,
    /// A pre-acquired resource-pool unit id.
    pub resource_id: Option<String>,
}

#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct HlsStartWire<'a> {
    room_id: &'a str,
    #[serde(skip_serializing_if = "Option::is_none")]
    config: Option<CompositionConfig>,
    #[serde(skip_serializing_if = "Option::is_none")]
    template_url: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    transcription: Option<&'a TranscriptionConfig>,
    #[serde(skip_serializing_if = "Option::is_none")]
    webhook_url: Option<&'a str>,
    #[serde(skip_serializing_if = "Option::is_none")]
    recording: Option<&'a Value>,
    #[serde(skip_serializing_if = "Option::is_none")]
    resource_id: Option<&'a str>,
}

/// The parameters for [`HlsResource::capture`].
#[derive(Debug, Clone, Default, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct HlsCaptureParams {
    /// The room to capture from.
    pub room_id: String,
    /// The image format.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub format: Option<HlsCaptureFormat>,
    /// The capture width, in pixels.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub width: Option<u32>,
    /// The capture height, in pixels.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub height: Option<u32>,
    /// Captures the frame at this timestamp, in seconds. Omit to capture live.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub time: Option<u32>,
}

/// The result of [`HlsResource::capture`].
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct HlsCaptureResult {
    /// The server's confirmation message.
    pub message: String,
    /// The room the frame was captured from.
    pub room_id: String,
    /// Where the captured image was stored.
    pub file_path: Option<String>,
    /// The captured image's size, in bytes.
    pub file_size: Option<i64>,
    /// The captured image's file name.
    pub file_name: Option<String>,
    /// Additional metadata the server returned about the capture.
    pub meta: Option<Value>,
    /// Any fields the server returned that this SDK does not model yet.
    #[serde(flatten)]
    pub extra: Map<String, Value>,
}

/// The query parameters for [`HlsResource::list`].
#[derive(Debug, Clone, Default)]
pub struct HlsListParams {
    /// The 1-based page number.
    pub page: Option<u32>,
    /// Items per page.
    pub per_page: Option<u32>,
    /// An opaque cursor from a previous page.
    pub cursor: Option<String>,
    /// Filters by room id.
    pub room_id: Option<String>,
    /// Filters by session id.
    pub session_id: Option<String>,
    /// Scopes to a specific user. Admin tokens only.
    pub user_id: Option<String>,
    /// Matches the room's meeting id.
    pub query: Option<String>,
}

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

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

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

    /// Begins HLS for a room.
    pub async fn start(&self, room_id: &str, params: HlsStartParams) -> Result<EgressHandle> {
        let mapped = composition_to_config(
            params.composition.as_ref(),
            // HLS requires a priority whenever a layout is present.
            Some(LayoutPriority::Speaker),
        );
        // A custom composition layout overrides an explicitly-passed templateUrl.
        let template_url = mapped.template_url.or(params.template_url);

        let body = HlsStartWire {
            room_id,
            config: mapped.config,
            template_url,
            transcription: params.transcription.as_ref(),
            webhook_url: params.webhook_url.as_deref(),
            recording: params.recording.as_ref(),
            resource_id: params.resource_id.as_deref(),
        };
        let path = format!("{PATH}/start");
        let raw = self
            .client
            .maybe_json(Method::POST, &path, CallOptions::json(&body)?)
            .await?;
        Ok(to_egress_handle(EgressType::Hls, room_id, raw))
    }

    /// Stops HLS. Accepts the start handle, or a bare room id.
    pub async fn stop(&self, target: impl Into<StopTarget>) -> Result<String> {
        let target = target.into();
        let path = format!("{PATH}/end");
        self.client
            .message(
                Method::POST,
                &path,
                CallOptions::json(StopWire::from(&target))?,
            )
            .await
    }

    /// Fetches a room's currently-active HLS stream, with its playback URLs.
    ///
    /// Returns a not-found error when no stream is active.
    pub async fn get(&self, room_id: &str) -> Result<HlsStream> {
        let path = format!("{PATH}/{}/active", escape(room_id));
        let raw: Value = self
            .client
            .json(Method::GET, &path, CallOptions::new())
            .await?;

        let data = raw.get("data").filter(|data| !data.is_null());
        let Some(data) = data else {
            return Err(Error::not_found(format!(
                "no active HLS stream for room {room_id}"
            )));
        };

        let mut stream: HlsStream =
            serde_json::from_value(data.clone()).map_err(|e| Error::decode(path, e))?;
        // This endpoint answers with the legacy `meetingId` alias.
        if stream.room_id.is_none() {
            stream.room_id = stream
                .extra
                .get("meetingId")
                .and_then(Value::as_str)
                .map(str::to_string);
        }
        Ok(stream)
    }

    /// Fetches an HLS stream by its id, as opposed to by room.
    pub async fn get_by_id(&self, id: &str) -> Result<HlsStream> {
        let path = format!("{PATH}/{}", escape(id));
        self.client
            .json(Method::GET, &path, CallOptions::new())
            .await
    }

    /// Captures a thumbnail from the live HLS stream.
    pub async fn capture(&self, params: HlsCaptureParams) -> Result<HlsCaptureResult> {
        let path = format!("{PATH}/capture");
        self.client
            .json(Method::POST, &path, CallOptions::json(&params)?)
            .await
    }

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

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

    /// Deletes a finished HLS recording.
    pub async fn delete(&self, id: &str) -> Result<String> {
        let path = format!("{PATH}/{}", escape(id));
        self.client
            .message(Method::DELETE, &path, CallOptions::new())
            .await
    }

    fn fetcher(&self, params: &HlsListParams) -> 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("roomId", params.room_id.as_deref())
                    .opt_str("sessionId", params.session_id.as_deref())
                    .opt_str("userId", params.user_id.as_deref())
                    .opt_str("query", params.query.as_deref())
                    .into_pairs();
                client
                    .json::<Value>(Method::GET, PATH, CallOptions::new().query(query))
                    .await
            })
        })
    }
}