videosdk-server-sdk 0.1.0

//! Pre-acquires and releases composition units. Enterprise tier only.

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;
use crate::error::Result;
use crate::pagination::{auto_page, paginate, ListParams, Page, PageFetcher};
use crate::query::QueryBuilder;
use crate::resources::escape;

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

string_enum! {
    /// A composition unit's status.
    ResourceStatus {
        /// The unit is being provisioned.
        PENDING => "pending",
        /// The unit is warm and unused.
        IDLE => "idle",
        /// The unit is composing.
        COMPOSING => "composing",
        /// The unit has been released.
        RELEASED => "released",
    }
}

string_enum! {
    /// The composer type a unit is acquired for.
    ComposerType {
        /// A recording composer.
        RECORDING => "recording",
        /// An HLS composer.
        HLS => "hls",
        /// An RTMP composer.
        RTMP => "rtmp",
    }
}

string_enum! {
    /// A unit's capture mode.
    ResourceMode {
        /// Capture both video and audio.
        VIDEO_AND_AUDIO => "video-and-audio",
        /// Capture audio only.
        AUDIO => "audio",
    }
}

string_enum! {
    /// A unit's output quality.
    ResourceQuality {
        /// Lowest bitrate.
        LOW => "low",
        /// Medium bitrate.
        MED => "med",
        /// High bitrate.
        HIGH => "high",
    }
}

/// A pre-acquired composition resource unit.
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ResourceUnit {
    /// The unit id. Pass it as `resource_id` when starting a composition.
    pub id: String,
    /// The unit's status.
    pub status: Option<ResourceStatus>,
    /// The unit's composer type.
    #[serde(rename = "type")]
    pub kind: Option<String>,
    /// The unit's capture mode.
    pub mode: Option<ResourceMode>,
    /// The unit's output quality.
    pub quality: Option<ResourceQuality>,
    /// The composers bound to this unit.
    #[serde(default, deserialize_with = "crate::common::null_to_default")]
    pub composer_ids: Vec<String>,
    /// Where unit events are delivered.
    pub webhook_url: Option<String>,
    /// Any fields the server returned that this SDK does not model yet.
    #[serde(flatten)]
    pub extra: Map<String, Value>,
}

/// The query parameters for [`ResourcePoolResource::list`].
#[derive(Debug, Clone, Default)]
pub struct ListResourcesParams {
    /// 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 unit status.
    pub status: Option<ResourceStatus>,
}

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

/// The parameters for [`ResourcePoolResource::acquire`].
#[derive(Debug, Clone, Default, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct AcquireResourceParams {
    /// The composer type to pre-warm. Required.
    #[serde(rename = "type", skip_serializing_if = "Option::is_none")]
    pub kind: Option<ComposerType>,
    /// Where to deliver unit events. Required.
    pub webhook_url: String,
    /// How many units to acquire. Defaults to 1.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub units: Option<u32>,
    /// The capture mode.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub mode: Option<ResourceMode>,
    /// The output quality.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub quality: Option<ResourceQuality>,
}

/// The outcome of releasing a single unit.
#[derive(Debug, Clone, Deserialize)]
pub struct ReleaseResourceResult {
    /// The unit that was released.
    pub id: String,
    /// Whether it was released.
    pub success: bool,
    /// Why the release failed, when it did.
    pub msg: Option<String>,
}

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

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

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

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

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

    /// Pre-acquires composition units.
    pub async fn acquire(&self, params: AcquireResourceParams) -> Result<Vec<ResourceUnit>> {
        let path = format!("{PATH}/acquire");
        self.client
            .data(Method::POST, &path, CallOptions::json(&params)?)
            .await
    }

    /// Releases units by id. Only idle units are released.
    pub async fn release(&self, ids: &[String]) -> Result<Vec<ReleaseResourceResult>> {
        let body = serde_json::json!({ "ids": ids });
        let path = format!("{PATH}/release");
        self.client
            .data(Method::POST, &path, CallOptions::json(&body)?)
            .await
    }

    fn fetcher(&self, params: &ListResourcesParams) -> PageFetcher {
        let client = self.client.clone();
        let status = params.status.clone();
        Arc::new(move |page, per_page| {
            let client = client.clone();
            let status = status.clone();
            Box::pin(async move {
                let query = QueryBuilder::new()
                    .opt("page", page)
                    .opt("perPage", per_page)
                    .opt_str("status", status.as_ref().map(ResourceStatus::as_str))
                    .into_pairs();
                client
                    .json::<Value>(Method::GET, PATH, CallOptions::new().query(query))
                    .await
            })
        })
    }
}