videosdk/resources/
transcodings.rs

1//! Post-processes finished recordings and HLS streams into downloadable files.
2
3use std::sync::Arc;
4
5use futures_util::Stream;
6use reqwest::Method;
7use serde::{Deserialize, Serialize, Serializer};
8use serde_json::{Map, Value};
9
10use crate::client::{CallOptions, Client};
11use crate::common::{string_enum, WebhookDeliverySummary};
12use crate::error::Result;
13use crate::pagination::{auto_page, paginate, ListParams, Page, PageFetcher};
14use crate::query::QueryBuilder;
15use crate::resources::escape;
16
17const PATH: &str = "/v2/transcodings";
18
19string_enum! {
20    /// A transcoding job's lifecycle state.
21    TranscodingStatus {
22        /// The job is queued.
23        PENDING => "pending",
24        /// The job is running.
25        PROCESSING => "processing",
26        /// The job finished.
27        COMPLETED => "completed",
28        /// The job failed.
29        FAILED => "failed",
30        /// The job was cancelled.
31        CANCELLED => "cancelled",
32    }
33}
34
35string_enum! {
36    /// A transcoding job kind.
37    TranscodingTask {
38        /// Composite-merge several individual recordings into one file.
39        COMPOSITE_MERGE => "composite-merge",
40        /// Convert a finished HLS stream to MP4.
41        HLS_TO_MP4 => "hls-to-mp4",
42        /// Concatenate room recordings chronologically.
43        MEETING_RECORDING_MERGE => "meeting-recording-merge",
44    }
45}
46
47/// A watermark applied to a merge transcoding.
48#[derive(Debug, Clone, Default, Serialize)]
49#[serde(rename_all = "camelCase")]
50pub struct TranscodingWatermark {
51    /// One of `image`, `timestamp` or `lat_long`.
52    #[serde(rename = "type")]
53    pub kind: String,
54    /// The watermark image. Required when `kind` is `image`.
55    #[serde(skip_serializing_if = "Option::is_none")]
56    pub image: Option<String>,
57    /// The start time. Required when `kind` is `timestamp`.
58    #[serde(skip_serializing_if = "Option::is_none")]
59    pub start_time: Option<String>,
60    /// The latitude. Required when `kind` is `lat_long`.
61    #[serde(skip_serializing_if = "Option::is_none")]
62    pub lat: Option<f64>,
63    /// The longitude. Required when `kind` is `lat_long`.
64    #[serde(skip_serializing_if = "Option::is_none")]
65    pub long: Option<f64>,
66}
67
68/// The output storage target of an HLS-to-MP4 transcoding.
69#[derive(Debug, Clone, Default, Serialize)]
70#[serde(rename_all = "camelCase")]
71pub struct TranscodingStorage {
72    /// One of `s3`, `blob` or `gcp`.
73    #[serde(rename = "type")]
74    pub kind: String,
75    /// The bucket, for `s3` and `gcp`.
76    #[serde(skip_serializing_if = "Option::is_none")]
77    pub bucket: Option<String>,
78    /// The container, for `blob`.
79    #[serde(skip_serializing_if = "Option::is_none")]
80    pub container: Option<String>,
81    /// The path prefix within the bucket or container.
82    #[serde(skip_serializing_if = "Option::is_none")]
83    pub dir_path: Option<String>,
84}
85
86/// The parameters for [`TranscodingsResource::merge`].
87#[derive(Debug, Clone, Default, Serialize)]
88#[serde(rename_all = "camelCase")]
89pub struct MergeTranscodingParams {
90    /// The individual recordings to composite-merge. Required, non-empty, and
91    /// without duplicates.
92    pub recording_ids: Vec<String>,
93    /// Defaults to `composite-merge`, the only supported value.
94    #[serde(skip_serializing_if = "Option::is_none")]
95    pub task: Option<String>,
96    /// A webhook to notify when the job completes.
97    #[serde(skip_serializing_if = "Option::is_none")]
98    pub webhook_url: Option<String>,
99    /// A watermark to apply.
100    #[serde(skip_serializing_if = "Option::is_none")]
101    pub watermark: Option<TranscodingWatermark>,
102}
103
104/// The parameters for [`TranscodingsResource::hls_to_mp4`].
105#[derive(Debug, Clone, Default, Serialize)]
106#[serde(rename_all = "camelCase")]
107pub struct HlsToMp4Params {
108    /// The room whose HLS to convert. One of `room_id`, `session_id` or `hls_id`
109    /// is required.
110    #[serde(skip_serializing_if = "Option::is_none")]
111    pub room_id: Option<String>,
112    /// The session whose HLS to convert.
113    #[serde(skip_serializing_if = "Option::is_none")]
114    pub session_id: Option<String>,
115    /// The HLS stream to convert.
116    #[serde(skip_serializing_if = "Option::is_none")]
117    pub hls_id: Option<String>,
118    /// A webhook to notify when the job completes.
119    #[serde(skip_serializing_if = "Option::is_none")]
120    pub webhook_url: Option<String>,
121    /// The output target. Its provider must match your configured credentials.
122    #[serde(skip_serializing_if = "Option::is_none")]
123    pub storage: Option<TranscodingStorage>,
124}
125
126/// A recording reference for a meeting-recording merge.
127///
128/// Serializes as a bare id string, or as `{id, presignedUrl}` when a presigned
129/// URL is attached.
130#[derive(Debug, Clone)]
131pub struct MeetingRecordingRef {
132    /// The recording id.
133    pub id: String,
134    /// Where to fetch the recording from, when it is not the default location.
135    pub presigned_url: Option<String>,
136}
137
138impl MeetingRecordingRef {
139    /// A reference to a recording by id alone.
140    pub fn new(id: impl Into<String>) -> Self {
141        Self {
142            id: id.into(),
143            presigned_url: None,
144        }
145    }
146}
147
148impl From<&str> for MeetingRecordingRef {
149    fn from(id: &str) -> Self {
150        Self::new(id)
151    }
152}
153
154impl From<String> for MeetingRecordingRef {
155    fn from(id: String) -> Self {
156        Self::new(id)
157    }
158}
159
160impl Serialize for MeetingRecordingRef {
161    fn serialize<S: Serializer>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error> {
162        match &self.presigned_url {
163            None => serializer.serialize_str(&self.id),
164            Some(presigned_url) => {
165                #[derive(Serialize)]
166                #[serde(rename_all = "camelCase")]
167                struct Full<'a> {
168                    id: &'a str,
169                    presigned_url: &'a str,
170                }
171                Full {
172                    id: &self.id,
173                    presigned_url,
174                }
175                .serialize(serializer)
176            }
177        }
178    }
179}
180
181/// The parameters for [`TranscodingsResource::meeting_recording_merge`].
182#[derive(Debug, Clone, Default, Serialize)]
183#[serde(rename_all = "camelCase")]
184pub struct MeetingRecordingMergeParams {
185    /// The room recordings to concatenate chronologically. Required, non-empty.
186    pub recording_ids: Vec<MeetingRecordingRef>,
187    /// A webhook to notify when the job completes.
188    #[serde(skip_serializing_if = "Option::is_none")]
189    pub webhook_url: Option<String>,
190    /// The output destination. Takes precedence over configured storage.
191    #[serde(skip_serializing_if = "Option::is_none")]
192    pub presigned_output_url: Option<String>,
193}
194
195/// A transcoding job's output file.
196#[derive(Debug, Clone, Deserialize)]
197#[serde(rename_all = "camelCase")]
198pub struct TranscodingFile {
199    /// The file id.
200    pub id: Option<String>,
201    /// The file's type.
202    #[serde(rename = "type")]
203    pub kind: Option<String>,
204    /// The file's size, in bytes.
205    pub size: Option<i64>,
206    /// Provider-specific metadata.
207    pub meta: Option<Value>,
208    /// Where the file was stored.
209    pub file_path: Option<String>,
210    /// Where the file can be downloaded.
211    pub file_url: Option<String>,
212}
213
214/// A transcoding job.
215#[derive(Debug, Clone, Deserialize)]
216#[serde(rename_all = "camelCase")]
217pub struct Transcoding {
218    /// The job id.
219    pub id: String,
220    /// The recordings that were transcoded.
221    #[serde(default, deserialize_with = "crate::common::null_to_default")]
222    pub recording_ids: Vec<String>,
223    /// The room the job relates to.
224    pub room_id: Option<String>,
225    /// The session the job relates to.
226    pub session_id: Option<String>,
227    /// The HLS stream the job relates to.
228    pub hls_id: Option<String>,
229    /// The job's status.
230    pub status: Option<TranscodingStatus>,
231    /// The job's kind.
232    pub task: Option<TranscodingTask>,
233    /// When the job started.
234    pub started_at: Option<String>,
235    /// When the job stopped.
236    pub stopped_at: Option<String>,
237    /// The produced file.
238    pub file: Option<TranscodingFile>,
239    /// A summary of the webhook deliveries for this job.
240    pub webhook: Option<WebhookDeliverySummary>,
241    /// Any fields the server returned that this SDK does not model yet.
242    #[serde(flatten)]
243    pub extra: Map<String, Value>,
244}
245
246/// The query parameters for [`TranscodingsResource::list`].
247#[derive(Debug, Clone, Default)]
248pub struct ListTranscodingsParams {
249    /// The 1-based page number.
250    pub page: Option<u32>,
251    /// Items per page.
252    pub per_page: Option<u32>,
253    /// An opaque cursor from a previous page.
254    pub cursor: Option<String>,
255    /// Filters by room id.
256    pub room_id: Option<String>,
257    /// Filters by session id.
258    pub session_id: Option<String>,
259    /// Filters by HLS stream id.
260    pub hls_id: Option<String>,
261    /// Filters by job status.
262    pub status: Option<TranscodingStatus>,
263}
264
265impl ListTranscodingsParams {
266    fn pagination(&self) -> ListParams {
267        ListParams {
268            page: self.page,
269            per_page: self.per_page,
270            cursor: self.cursor.clone(),
271        }
272    }
273}
274
275/// Transcoding jobs. Reached via [`Client::transcodings`].
276#[derive(Debug, Clone, Copy)]
277pub struct TranscodingsResource<'a> {
278    client: &'a Client,
279}
280
281impl<'a> TranscodingsResource<'a> {
282    pub(crate) fn new(client: &'a Client) -> Self {
283        Self { client }
284    }
285
286    /// Composite-merges several individual recordings into one MP4.
287    pub async fn merge(&self, params: MergeTranscodingParams) -> Result<Transcoding> {
288        let path = format!("{PATH}/merge");
289        self.client
290            .json(Method::POST, &path, CallOptions::json(&params)?)
291            .await
292    }
293
294    /// Converts a finished HLS stream to MP4, into your configured storage.
295    pub async fn hls_to_mp4(&self, params: HlsToMp4Params) -> Result<Transcoding> {
296        let path = format!("{PATH}/hls-to-mp4");
297        self.client
298            .json(Method::POST, &path, CallOptions::json(&params)?)
299            .await
300    }
301
302    /// Concatenates several room recordings end to end, chronologically.
303    pub async fn meeting_recording_merge(
304        &self,
305        params: MeetingRecordingMergeParams,
306    ) -> Result<Transcoding> {
307        let path = format!("{PATH}/meeting-recording-merge");
308        self.client
309            .json(Method::POST, &path, CallOptions::json(&params)?)
310            .await
311    }
312
313    /// Lists transcoding jobs, one page at a time.
314    pub async fn list(&self, params: ListTranscodingsParams) -> Result<Page<Transcoding>> {
315        paginate(self.fetcher(&params), &params.pagination(), "data", None).await
316    }
317
318    /// Lists transcoding jobs, transparently fetching every page.
319    pub fn list_stream(
320        &self,
321        params: ListTranscodingsParams,
322    ) -> impl Stream<Item = Result<Transcoding>> + Send {
323        auto_page(self.fetcher(&params), params.pagination(), "data", None)
324    }
325
326    /// Fetches a transcoding job by id.
327    pub async fn get(&self, id: &str) -> Result<Transcoding> {
328        let path = format!("{PATH}/{}", escape(id));
329        self.client
330            .json(Method::GET, &path, CallOptions::new())
331            .await
332    }
333
334    /// Cancels a pending or processing transcoding job.
335    pub async fn cancel(&self, id: &str) -> Result<Transcoding> {
336        let path = format!("{PATH}/{}/cancel", escape(id));
337        self.client
338            .json(Method::POST, &path, CallOptions::new())
339            .await
340    }
341
342    fn fetcher(&self, params: &ListTranscodingsParams) -> PageFetcher {
343        let client = self.client.clone();
344        let params = params.clone();
345        Arc::new(move |page, per_page| {
346            let client = client.clone();
347            let params = params.clone();
348            Box::pin(async move {
349                let query = QueryBuilder::new()
350                    .opt("page", page)
351                    .opt("perPage", per_page)
352                    .opt_str("roomId", params.room_id.as_deref())
353                    .opt_str("sessionId", params.session_id.as_deref())
354                    .opt_str("hlsId", params.hls_id.as_deref())
355                    .opt_str(
356                        "status",
357                        params.status.as_ref().map(TranscodingStatus::as_str),
358                    )
359                    .into_pairs();
360                client
361                    .json::<Value>(Method::GET, PATH, CallOptions::new().query(query))
362                    .await
363            })
364        })
365    }
366}
367
368#[cfg(test)]
369mod tests {
370    use super::*;
371    use serde_json::json;
372
373    #[test]
374    fn a_bare_recording_ref_serializes_as_a_string() {
375        let params = MeetingRecordingMergeParams {
376            recording_ids: vec!["rec-1".into(), "rec-2".into()],
377            ..Default::default()
378        };
379        assert_eq!(
380            serde_json::to_value(&params).unwrap(),
381            json!({"recordingIds": ["rec-1", "rec-2"]})
382        );
383    }
384
385    #[test]
386    fn a_recording_ref_with_a_presigned_url_serializes_as_an_object() {
387        let params = MeetingRecordingMergeParams {
388            recording_ids: vec![
389                MeetingRecordingRef::new("rec-1"),
390                MeetingRecordingRef {
391                    id: "rec-2".into(),
392                    presigned_url: Some("https://s3/get".into()),
393                },
394            ],
395            presigned_output_url: Some("https://s3/put".into()),
396            ..Default::default()
397        };
398        assert_eq!(
399            serde_json::to_value(&params).unwrap(),
400            json!({
401                "recordingIds": ["rec-1", {"id": "rec-2", "presignedUrl": "https://s3/get"}],
402                "presignedOutputUrl": "https://s3/put",
403            })
404        );
405    }
406}
videosdk/resources/transcodings.rs

videosdk/resources/
transcodings.rs
