videosdk-server-sdk 0.1.0

//! The RTMP-out (livestream) 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/livestreams";

/// An RTMP destination.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct RtmpStream {
    /// The RTMP ingest URL.
    pub url: String,
    /// The RTMP stream key.
    pub stream_key: String,
}

/// An RTMP-out livestream.
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct Livestream {
    /// The livestream id.
    pub id: String,
    /// The composer producing 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 RTMP destinations.
    #[serde(default, deserialize_with = "crate::common::null_to_default")]
    pub outputs: Vec<RtmpStream>,
    /// When the livestream started.
    pub start: Option<String>,
    /// When the livestream ended, or `None` if it is still running.
    pub end: Option<String>,
    /// The region the livestream ran in.
    pub region: Option<String>,
    /// A summary of the webhook deliveries for this livestream.
    pub webhook: Option<WebhookDeliverySummary>,
    /// 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 parameters for [`RtmpResource::start`].
#[derive(Debug, Clone, Default)]
pub struct RtmpStartParams {
    /// The RTMP destinations. Required, and must be non-empty.
    pub streams: Vec<RtmpStream>,
    /// The layout, quality, orientation and theme. RTMP requires a layout.
    pub composition: Option<Composition>,
    /// Transcribes and optionally summarizes the livestream.
    pub transcription: Option<TranscriptionConfig>,
    /// A pre-acquired resource-pool unit id.
    pub resource_id: Option<String>,
}

/// `streams` is sent as `outputs`.
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct RtmpStartWire<'a> {
    room_id: &'a str,
    outputs: &'a [RtmpStream],
    #[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")]
    resource_id: Option<&'a str>,
}

/// The query parameters for [`RtmpResource::list`].
#[derive(Debug, Clone, Default)]
pub struct RtmpListParams {
    /// 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 RtmpListParams {
    fn pagination(&self) -> ListParams {
        ListParams {
            page: self.page,
            per_page: self.per_page,
            cursor: self.cursor.clone(),
        }
    }
}

/// The RTMP-out API: push a room to RTMP destinations. Reached via [`Client::rtmp`].
#[derive(Debug, Clone, Copy)]
pub struct RtmpResource<'a> {
    client: &'a Client,
}

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

    /// Begins an RTMP livestream.
    pub async fn start(&self, room_id: &str, params: RtmpStartParams) -> Result<EgressHandle> {
        if params.streams.is_empty() {
            return Err(Error::validation(
                "rtmp.start() requires at least one destination in `streams`",
            ));
        }
        let mapped = composition_to_config(
            params.composition.as_ref(),
            // RTMP requires a priority whenever a layout is present.
            Some(LayoutPriority::Speaker),
        );
        let body = RtmpStartWire {
            room_id,
            outputs: &params.streams,
            config: mapped.config,
            template_url: mapped.template_url,
            transcription: params.transcription.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::Livestream, room_id, raw))
    }

    /// Stops an RTMP livestream. 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
    }

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

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

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

    fn fetcher(&self, params: &RtmpListParams) -> 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
            })
        })
    }
}