videosdk-server-sdk 0.1.0

//! Post-processes finished recordings and HLS streams into downloadable files.

use std::sync::Arc;

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

use crate::client::{CallOptions, Client};
use crate::common::{string_enum, WebhookDeliverySummary};
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/transcodings";

string_enum! {
    /// A transcoding job's lifecycle state.
    TranscodingStatus {
        /// The job is queued.
        PENDING => "pending",
        /// The job is running.
        PROCESSING => "processing",
        /// The job finished.
        COMPLETED => "completed",
        /// The job failed.
        FAILED => "failed",
        /// The job was cancelled.
        CANCELLED => "cancelled",
    }
}

string_enum! {
    /// A transcoding job kind.
    TranscodingTask {
        /// Composite-merge several individual recordings into one file.
        COMPOSITE_MERGE => "composite-merge",
        /// Convert a finished HLS stream to MP4.
        HLS_TO_MP4 => "hls-to-mp4",
        /// Concatenate room recordings chronologically.
        MEETING_RECORDING_MERGE => "meeting-recording-merge",
    }
}

/// A watermark applied to a merge transcoding.
#[derive(Debug, Clone, Default, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct TranscodingWatermark {
    /// One of `image`, `timestamp` or `lat_long`.
    #[serde(rename = "type")]
    pub kind: String,
    /// The watermark image. Required when `kind` is `image`.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub image: Option<String>,
    /// The start time. Required when `kind` is `timestamp`.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub start_time: Option<String>,
    /// The latitude. Required when `kind` is `lat_long`.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub lat: Option<f64>,
    /// The longitude. Required when `kind` is `lat_long`.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub long: Option<f64>,
}

/// The output storage target of an HLS-to-MP4 transcoding.
#[derive(Debug, Clone, Default, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct TranscodingStorage {
    /// One of `s3`, `blob` or `gcp`.
    #[serde(rename = "type")]
    pub kind: String,
    /// The bucket, for `s3` and `gcp`.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub bucket: Option<String>,
    /// The container, for `blob`.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub container: Option<String>,
    /// The path prefix within the bucket or container.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub dir_path: Option<String>,
}

/// The parameters for [`TranscodingsResource::merge`].
#[derive(Debug, Clone, Default, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct MergeTranscodingParams {
    /// The individual recordings to composite-merge. Required, non-empty, and
    /// without duplicates.
    pub recording_ids: Vec<String>,
    /// Defaults to `composite-merge`, the only supported value.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub task: Option<String>,
    /// A webhook to notify when the job completes.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub webhook_url: Option<String>,
    /// A watermark to apply.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub watermark: Option<TranscodingWatermark>,
}

/// The parameters for [`TranscodingsResource::hls_to_mp4`].
#[derive(Debug, Clone, Default, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct HlsToMp4Params {
    /// The room whose HLS to convert. One of `room_id`, `session_id` or `hls_id`
    /// is required.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub room_id: Option<String>,
    /// The session whose HLS to convert.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub session_id: Option<String>,
    /// The HLS stream to convert.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub hls_id: Option<String>,
    /// A webhook to notify when the job completes.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub webhook_url: Option<String>,
    /// The output target. Its provider must match your configured credentials.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub storage: Option<TranscodingStorage>,
}

/// A recording reference for a meeting-recording merge.
///
/// Serializes as a bare id string, or as `{id, presignedUrl}` when a presigned
/// URL is attached.
#[derive(Debug, Clone)]
pub struct MeetingRecordingRef {
    /// The recording id.
    pub id: String,
    /// Where to fetch the recording from, when it is not the default location.
    pub presigned_url: Option<String>,
}

impl MeetingRecordingRef {
    /// A reference to a recording by id alone.
    pub fn new(id: impl Into<String>) -> Self {
        Self {
            id: id.into(),
            presigned_url: None,
        }
    }
}

impl From<&str> for MeetingRecordingRef {
    fn from(id: &str) -> Self {
        Self::new(id)
    }
}

impl From<String> for MeetingRecordingRef {
    fn from(id: String) -> Self {
        Self::new(id)
    }
}

impl Serialize for MeetingRecordingRef {
    fn serialize<S: Serializer>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error> {
        match &self.presigned_url {
            None => serializer.serialize_str(&self.id),
            Some(presigned_url) => {
                #[derive(Serialize)]
                #[serde(rename_all = "camelCase")]
                struct Full<'a> {
                    id: &'a str,
                    presigned_url: &'a str,
                }
                Full {
                    id: &self.id,
                    presigned_url,
                }
                .serialize(serializer)
            }
        }
    }
}

/// The parameters for [`TranscodingsResource::meeting_recording_merge`].
#[derive(Debug, Clone, Default, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct MeetingRecordingMergeParams {
    /// The room recordings to concatenate chronologically. Required, non-empty.
    pub recording_ids: Vec<MeetingRecordingRef>,
    /// A webhook to notify when the job completes.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub webhook_url: Option<String>,
    /// The output destination. Takes precedence over configured storage.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub presigned_output_url: Option<String>,
}

/// A transcoding job's output file.
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct TranscodingFile {
    /// The file id.
    pub id: Option<String>,
    /// The file's type.
    #[serde(rename = "type")]
    pub kind: Option<String>,
    /// The file's size, in bytes.
    pub size: Option<i64>,
    /// Provider-specific metadata.
    pub meta: Option<Value>,
    /// Where the file was stored.
    pub file_path: Option<String>,
    /// Where the file can be downloaded.
    pub file_url: Option<String>,
}

/// A transcoding job.
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct Transcoding {
    /// The job id.
    pub id: String,
    /// The recordings that were transcoded.
    #[serde(default, deserialize_with = "crate::common::null_to_default")]
    pub recording_ids: Vec<String>,
    /// The room the job relates to.
    pub room_id: Option<String>,
    /// The session the job relates to.
    pub session_id: Option<String>,
    /// The HLS stream the job relates to.
    pub hls_id: Option<String>,
    /// The job's status.
    pub status: Option<TranscodingStatus>,
    /// The job's kind.
    pub task: Option<TranscodingTask>,
    /// When the job started.
    pub started_at: Option<String>,
    /// When the job stopped.
    pub stopped_at: Option<String>,
    /// The produced file.
    pub file: Option<TranscodingFile>,
    /// A summary of the webhook deliveries for this job.
    pub webhook: Option<WebhookDeliverySummary>,
    /// Any fields the server returned that this SDK does not model yet.
    #[serde(flatten)]
    pub extra: Map<String, Value>,
}

/// The query parameters for [`TranscodingsResource::list`].
#[derive(Debug, Clone, Default)]
pub struct ListTranscodingsParams {
    /// 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>,
    /// Filters by HLS stream id.
    pub hls_id: Option<String>,
    /// Filters by job status.
    pub status: Option<TranscodingStatus>,
}

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

/// Transcoding jobs. Reached via [`Client::transcodings`].
#[derive(Debug, Clone, Copy)]
pub struct TranscodingsResource<'a> {
    client: &'a Client,
}

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

    /// Composite-merges several individual recordings into one MP4.
    pub async fn merge(&self, params: MergeTranscodingParams) -> Result<Transcoding> {
        let path = format!("{PATH}/merge");
        self.client
            .json(Method::POST, &path, CallOptions::json(&params)?)
            .await
    }

    /// Converts a finished HLS stream to MP4, into your configured storage.
    pub async fn hls_to_mp4(&self, params: HlsToMp4Params) -> Result<Transcoding> {
        let path = format!("{PATH}/hls-to-mp4");
        self.client
            .json(Method::POST, &path, CallOptions::json(&params)?)
            .await
    }

    /// Concatenates several room recordings end to end, chronologically.
    pub async fn meeting_recording_merge(
        &self,
        params: MeetingRecordingMergeParams,
    ) -> Result<Transcoding> {
        let path = format!("{PATH}/meeting-recording-merge");
        self.client
            .json(Method::POST, &path, CallOptions::json(&params)?)
            .await
    }

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

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

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

    /// Cancels a pending or processing transcoding job.
    pub async fn cancel(&self, id: &str) -> Result<Transcoding> {
        let path = format!("{PATH}/{}/cancel", escape(id));
        self.client
            .json(Method::POST, &path, CallOptions::new())
            .await
    }

    fn fetcher(&self, params: &ListTranscodingsParams) -> 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("hlsId", params.hls_id.as_deref())
                    .opt_str(
                        "status",
                        params.status.as_ref().map(TranscodingStatus::as_str),
                    )
                    .into_pairs();
                client
                    .json::<Value>(Method::GET, PATH, CallOptions::new().query(query))
                    .await
            })
        })
    }
}

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

    #[test]
    fn a_bare_recording_ref_serializes_as_a_string() {
        let params = MeetingRecordingMergeParams {
            recording_ids: vec!["rec-1".into(), "rec-2".into()],
            ..Default::default()
        };
        assert_eq!(
            serde_json::to_value(&params).unwrap(),
            json!({"recordingIds": ["rec-1", "rec-2"]})
        );
    }

    #[test]
    fn a_recording_ref_with_a_presigned_url_serializes_as_an_object() {
        let params = MeetingRecordingMergeParams {
            recording_ids: vec![
                MeetingRecordingRef::new("rec-1"),
                MeetingRecordingRef {
                    id: "rec-2".into(),
                    presigned_url: Some("https://s3/get".into()),
                },
            ],
            presigned_output_url: Some("https://s3/put".into()),
            ..Default::default()
        };
        assert_eq!(
            serde_json::to_value(&params).unwrap(),
            json!({
                "recordingIds": ["rec-1", {"id": "rec-2", "presignedUrl": "https://s3/get"}],
                "presignedOutputUrl": "https://s3/put",
            })
        );
    }
}