videosdk-server-sdk 0.1.0

//! The recordings API: room, participant, track, composite and merge.

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, OnFailureConfig, ParticipantFileFormat, RecordingFile, RecordingFileFormat,
    ResourceLinks, TrackFileFormat, TrackKind, 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/recordings";
const PARTICIPANT_PATH: &str = "/v2/recordings/participant";
const TRACK_PATH: &str = "/v2/recordings/participant/track";
const COMPOSITE_PATH: &str = "/v2/recordings/composite";
const MERGE_PATH: &str = "/v2/recordings/participant/merge";

/* -------------------------------- response types ------------------------------- */

/// A composited room recording.
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct Recording {
    /// The recording id.
    pub id: String,
    /// The room that was recorded.
    pub room_id: Option<String>,
    /// The produced file.
    pub file: Option<RecordingFile>,
    /// A summary of the webhook deliveries for this recording.
    pub webhook: Option<WebhookDeliverySummary>,
    /// When the recording started.
    pub start: Option<String>,
    /// When the recording ended, or `None` if it is still running.
    pub end: Option<String>,
    /// 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>,
}

/// A participant recording, or its per-track variant.
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct IndividualRecording {
    /// The recording id.
    pub id: String,
    /// The room that was recorded.
    pub room_id: Option<String>,
    /// The recorded participant's display name.
    pub participant_name: Option<String>,
    /// The produced files.
    #[serde(default, deserialize_with = "crate::common::null_to_default")]
    pub files: Vec<RecordingFile>,
    /// A summary of the webhook deliveries for this recording.
    pub webhook: Option<WebhookDeliverySummary>,
    /// When the recording started.
    pub start: Option<String>,
    /// When the recording ended, or `None` if it is still running.
    pub end: Option<String>,
    /// Any fields the server returned that this SDK does not model yet.
    #[serde(flatten)]
    pub extra: Map<String, Value>,
}

/// A server-side composed recording.
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct CompositeRecording {
    /// The recording id.
    pub id: String,
    /// The legacy alias of [`room_id`](CompositeRecording::room_id).
    pub meeting_id: Option<String>,
    /// The room that was recorded.
    pub room_id: Option<String>,
    /// The session that was recorded.
    pub session_id: Option<String>,
    /// The participants that were composed.
    #[serde(default, deserialize_with = "crate::common::null_to_default")]
    pub participants: Vec<Value>,
    /// The produced file's format.
    pub file_format: Option<String>,
    /// When the recording started.
    pub start: Option<String>,
    /// When the recording ended, or `None` if it is still running.
    pub end: Option<String>,
    /// The produced file's id.
    pub file_id: Option<String>,
    /// The produced file.
    pub file: Option<RecordingFile>,
    /// Any fields the server returned that this SDK does not model yet.
    #[serde(flatten)]
    pub extra: Map<String, Value>,
}

/// A single channel entry on a merge recording.
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct MergeChannel {
    /// The source recording.
    pub recording_id: Option<String>,
    /// The source participant.
    pub participant_id: Option<String>,
}

/// A merged stereo-audio recording.
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct MergeRecording {
    /// The merge (muxer) id.
    pub id: String,
    /// The left channel's sources.
    #[serde(default, deserialize_with = "crate::common::null_to_default")]
    pub channel1: Vec<MergeChannel>,
    /// The right channel's sources.
    #[serde(default, deserialize_with = "crate::common::null_to_default")]
    pub channel2: Vec<MergeChannel>,
    /// When the merge started.
    pub start: Option<String>,
    /// When the merge finished.
    pub end: Option<String>,
    /// The merge job's status.
    pub status: Option<String>,
    /// The room whose recordings were merged.
    pub meeting_id: Option<String>,
    /// The session whose recordings were merged.
    pub session_id: Option<String>,
    /// The produced file.
    pub file: Option<RecordingFile>,
    /// Any fields the server returned that this SDK does not model yet.
    #[serde(flatten)]
    pub extra: Map<String, Value>,
}

/// The result of starting a merge job.
#[derive(Debug, Clone, Deserialize)]
pub struct MergeRecordingResult {
    /// The server's confirmation message.
    pub message: String,
    /// The merge job that was created.
    pub recording: MergeRecording,
}

/* -------------------------------- request types -------------------------------- */

/// The parameters for [`RecordingsResource::start`], a room recording.
#[derive(Debug, Clone, Default)]
pub struct RecordingStartParams {
    /// The layout, quality, orientation and theme.
    pub composition: Option<Composition>,
    /// Transcribes and optionally summarizes the recording.
    pub transcription: Option<TranscriptionConfig>,
    /// What to do if the composition fails.
    pub on_failure: Option<OnFailureConfig>,
    /// The output format. Defaults to `mp4`.
    pub file_format: Option<RecordingFileFormat>,
    /// A webhook to notify when the recording completes.
    pub webhook_url: Option<String>,
    /// The storage path prefix for the output files.
    pub dir_path: Option<String>,
    /// A pre-acquired resource-pool unit id.
    pub resource_id: Option<String>,
    /// A presigned URL to upload the output to.
    pub pre_signed_url: Option<String>,
}

/// `dir_path` becomes `awsDirPath` here. Every other recording endpoint calls the
/// same field `bucketDirPath` or `dirPath` — the name is per-endpoint.
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct RecordingStartWire<'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")]
    aws_dir_path: Option<&'a str>,
    #[serde(skip_serializing_if = "Option::is_none")]
    transcription: Option<&'a TranscriptionConfig>,
    #[serde(skip_serializing_if = "Option::is_none")]
    on_failure: Option<&'a OnFailureConfig>,
    #[serde(skip_serializing_if = "Option::is_none")]
    file_format: Option<RecordingFileFormat>,
    #[serde(skip_serializing_if = "Option::is_none")]
    webhook_url: Option<&'a str>,
    #[serde(skip_serializing_if = "Option::is_none")]
    resource_id: Option<&'a str>,
    /// Lowercase `url` here, unlike composite's `preSignedURL`.
    #[serde(skip_serializing_if = "Option::is_none")]
    pre_signed_url: Option<&'a str>,
}

/// The parameters for [`RecordingsResource::get`].
#[derive(Debug, Clone, Default)]
pub struct RecordingGetParams {
    /// Includes transcription data in the response.
    pub with_transcription: Option<bool>,
}

/// The parameters for [`RecordingParticipantResource::start`].
#[derive(Debug, Clone, Default)]
pub struct ParticipantRecordingStartParams {
    /// The participant to record. Required.
    pub participant_id: String,
    /// The output format. Defaults to `webm`.
    pub file_format: Option<ParticipantFileFormat>,
    /// A webhook to notify when the recording completes.
    pub webhook_url: Option<String>,
    /// The storage path prefix for the output files.
    pub dir_path: Option<String>,
    /// Arbitrary metadata stored with the recording.
    pub metadata: Option<Value>,
}

#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct ParticipantRecordingStartWire<'a> {
    room_id: &'a str,
    participant_id: &'a str,
    #[serde(skip_serializing_if = "Option::is_none")]
    file_format: Option<ParticipantFileFormat>,
    #[serde(skip_serializing_if = "Option::is_none")]
    webhook_url: Option<&'a str>,
    #[serde(skip_serializing_if = "Option::is_none")]
    bucket_dir_path: Option<&'a str>,
    #[serde(skip_serializing_if = "Option::is_none")]
    metadata: Option<&'a Value>,
}

/// The parameters for [`RecordingTrackResource::start`].
#[derive(Debug, Clone)]
pub struct TrackRecordingStartParams {
    /// The participant whose track to record. Required.
    pub participant_id: String,
    /// The media kind to record. Required.
    pub kind: TrackKind,
    /// The output format. Defaults to `webm`.
    pub file_format: Option<TrackFileFormat>,
    /// A webhook to notify when the recording completes.
    pub webhook_url: Option<String>,
    /// The storage path prefix for the output files.
    pub dir_path: Option<String>,
}

#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct TrackRecordingStartWire<'a> {
    room_id: &'a str,
    participant_id: &'a str,
    kind: TrackKind,
    #[serde(skip_serializing_if = "Option::is_none")]
    file_format: Option<TrackFileFormat>,
    #[serde(skip_serializing_if = "Option::is_none")]
    webhook_url: Option<&'a str>,
    #[serde(skip_serializing_if = "Option::is_none")]
    bucket_dir_path: Option<&'a str>,
}

/// Selects a participant, and optionally which of their tracks, to compose.
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct CompositeParticipantSelector {
    /// The participant to compose.
    pub participant_id: String,
    /// The media kinds to compose. Defaults to all four when empty.
    #[serde(skip_serializing_if = "Vec::is_empty")]
    pub kind: Vec<TrackKind>,
}

/// A watermark applied to a composite recording.
#[derive(Debug, Clone, Default, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct CompositeWatermark {
    /// Either `custom` or `pubsub`.
    #[serde(rename = "type")]
    pub kind: String,
    /// The watermark image's URL.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub image_url: Option<String>,
    /// The watermark image, base64-encoded.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub image_base64: Option<String>,
    /// Watermark text.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub text: Option<String>,
    /// An IANA timezone, for the `custom` kind.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub timezone: Option<String>,
    /// The pub/sub topic, for the `pubsub` kind.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub topic: Option<String>,
}

/// The parameters for [`RecordingCompositeResource::start`].
#[derive(Debug, Clone, Default)]
pub struct CompositeRecordingStartParams {
    /// The participants to compose. Empty lets the recorder decide.
    pub participants: Vec<CompositeParticipantSelector>,
    /// Up to three watermarks.
    pub watermarks: Vec<CompositeWatermark>,
    /// A webhook to notify when the recording completes.
    pub webhook_url: Option<String>,
    /// The storage path prefix for the output files.
    pub dir_path: Option<String>,
    /// A presigned URL to upload the output to.
    pub pre_signed_url: Option<String>,
    /// The output format.
    pub file_format: Option<String>,
}

#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct CompositeRecordingStartWire<'a> {
    room_id: &'a str,
    #[serde(skip_serializing_if = "<[_]>::is_empty")]
    participants: &'a [CompositeParticipantSelector],
    #[serde(skip_serializing_if = "<[_]>::is_empty")]
    watermarks: &'a [CompositeWatermark],
    #[serde(skip_serializing_if = "Option::is_none")]
    webhook_url: Option<&'a str>,
    #[serde(skip_serializing_if = "Option::is_none")]
    bucket_dir_path: Option<&'a str>,
    /// Capital `URL` here, unlike the room recording's `preSignedUrl`.
    #[serde(rename = "preSignedURL", skip_serializing_if = "Option::is_none")]
    pre_signed_url: Option<&'a str>,
    #[serde(skip_serializing_if = "Option::is_none")]
    file_format: Option<&'a str>,
}

/// One source in a merge channel.
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct MergeChannelEntry {
    /// The participant to take audio from.
    pub participant_id: String,
    /// The source recording. Resolved from the participant when omitted.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub recording_id: Option<String>,
    /// Either `participant` or `track`.
    #[serde(rename = "type", skip_serializing_if = "Option::is_none")]
    pub kind: Option<String>,
}

/// The parameters for [`MergeRecordingResource::create`].
#[derive(Debug, Clone, Default, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct MergeRecordingCreateParams {
    /// The session whose participant audio to merge. Required.
    pub session_id: String,
    /// The left channel's sources. Required, non-empty.
    pub channel1: Vec<MergeChannelEntry>,
    /// The right channel's sources. Required, non-empty.
    pub channel2: Vec<MergeChannelEntry>,
    /// A global override for each channel entry's type.
    #[serde(rename = "type", skip_serializing_if = "Option::is_none")]
    pub kind: Option<String>,
    /// A webhook to notify when the merge completes.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub webhook_url: Option<String>,
    /// The storage path prefix. Named `dirPath` here, unlike its siblings.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub dir_path: Option<String>,
}

/* --------------------------------- list params --------------------------------- */

/// The query parameters for [`RecordingsResource::list`].
#[derive(Debug, Clone, Default)]
pub struct RecordingListParams {
    /// 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>,
    /// Includes transcription data in the response.
    pub with_transcription: Option<bool>,
}

/// The query parameters for the participant and track recording lists.
#[derive(Debug, Clone, Default)]
pub struct IndividualRecordingListParams {
    /// 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 participant id.
    pub participant_id: Option<String>,
    /// Scopes to a specific user. Admin tokens only.
    pub user_id: Option<String>,
}

/// The query parameters for [`RecordingCompositeResource::list`].
#[derive(Debug, Clone, Default)]
pub struct CompositeRecordingListParams {
    /// 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>,
}

/// The query parameters for [`MergeRecordingResource::list`].
#[derive(Debug, Clone, Default)]
pub struct MergeRecordingListParams {
    /// 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 merge status.
    pub status: Option<String>,
    /// Filters by room id.
    pub room_id: Option<String>,
    /// Filters by session id.
    pub session_id: Option<String>,
    /// Filters by muxer id.
    pub id: Option<String>,
}

macro_rules! pagination {
    ($name:ident) => {
        impl $name {
            fn pagination(&self) -> ListParams {
                ListParams {
                    page: self.page,
                    per_page: self.per_page,
                    cursor: self.cursor.clone(),
                }
            }
        }
    };
}

pagination!(RecordingListParams);
pagination!(IndividualRecordingListParams);
pagination!(CompositeRecordingListParams);
pagination!(MergeRecordingListParams);

/* -------------------------------- sub-resources -------------------------------- */

/// Records individual participants. Reached via [`RecordingsResource::participant`].
#[derive(Debug, Clone, Copy)]
pub struct RecordingParticipantResource<'a> {
    client: &'a Client,
}

impl<'a> RecordingParticipantResource<'a> {
    /// Begins recording a participant, returning the server's confirmation message.
    pub async fn start(
        &self,
        room_id: &str,
        params: ParticipantRecordingStartParams,
    ) -> Result<String> {
        let body = ParticipantRecordingStartWire {
            room_id,
            participant_id: &params.participant_id,
            file_format: params.file_format,
            webhook_url: params.webhook_url.as_deref(),
            bucket_dir_path: params.dir_path.as_deref(),
            metadata: params.metadata.as_ref(),
        };
        let path = format!("{PARTICIPANT_PATH}/start");
        self.client
            .message(Method::POST, &path, CallOptions::json(&body)?)
            .await
    }

    /// Stops a participant recording.
    pub async fn stop(&self, room_id: &str, participant_id: &str) -> Result<String> {
        let body = serde_json::json!({"roomId": room_id, "participantId": participant_id});
        let path = format!("{PARTICIPANT_PATH}/stop");
        self.client
            .message(Method::POST, &path, CallOptions::json(&body)?)
            .await
    }

    /// Lists participant recordings, one page at a time.
    pub async fn list(
        &self,
        params: IndividualRecordingListParams,
    ) -> Result<Page<IndividualRecording>> {
        let fetcher = individual_fetcher(self.client, PARTICIPANT_PATH, &params);
        paginate(fetcher, &params.pagination(), "data", None).await
    }

    /// Lists participant recordings, transparently fetching every page.
    pub fn list_stream(
        &self,
        params: IndividualRecordingListParams,
    ) -> impl Stream<Item = Result<IndividualRecording>> + Send {
        let fetcher = individual_fetcher(self.client, PARTICIPANT_PATH, &params);
        auto_page(fetcher, params.pagination(), "data", None)
    }

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

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

/// Records individual tracks. Reached via [`RecordingsResource::track`].
#[derive(Debug, Clone, Copy)]
pub struct RecordingTrackResource<'a> {
    client: &'a Client,
}

impl<'a> RecordingTrackResource<'a> {
    /// Begins recording a single track, returning the server's confirmation message.
    pub async fn start(&self, room_id: &str, params: TrackRecordingStartParams) -> Result<String> {
        let body = TrackRecordingStartWire {
            room_id,
            participant_id: &params.participant_id,
            kind: params.kind,
            file_format: params.file_format,
            webhook_url: params.webhook_url.as_deref(),
            bucket_dir_path: params.dir_path.as_deref(),
        };
        let path = format!("{TRACK_PATH}/start");
        self.client
            .message(Method::POST, &path, CallOptions::json(&body)?)
            .await
    }

    /// Stops a track recording. All three arguments identify the track.
    pub async fn stop(
        &self,
        room_id: &str,
        participant_id: &str,
        kind: TrackKind,
    ) -> Result<String> {
        let body = serde_json::json!({
            "roomId": room_id, "participantId": participant_id, "kind": kind,
        });
        let path = format!("{TRACK_PATH}/stop");
        self.client
            .message(Method::POST, &path, CallOptions::json(&body)?)
            .await
    }

    /// Lists track recordings, one page at a time.
    pub async fn list(
        &self,
        params: IndividualRecordingListParams,
    ) -> Result<Page<IndividualRecording>> {
        let fetcher = individual_fetcher(self.client, TRACK_PATH, &params);
        paginate(fetcher, &params.pagination(), "data", None).await
    }

    /// Lists track recordings, transparently fetching every page.
    pub fn list_stream(
        &self,
        params: IndividualRecordingListParams,
    ) -> impl Stream<Item = Result<IndividualRecording>> + Send {
        let fetcher = individual_fetcher(self.client, TRACK_PATH, &params);
        auto_page(fetcher, params.pagination(), "data", None)
    }

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

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

/// Records a server-side composed mix. Reached via [`RecordingsResource::composite`].
#[derive(Debug, Clone, Copy)]
pub struct RecordingCompositeResource<'a> {
    client: &'a Client,
}

impl<'a> RecordingCompositeResource<'a> {
    /// Begins a composite recording. The returned handle's `id` is the `recordingId`
    /// that [`stop`](RecordingCompositeResource::stop) requires.
    pub async fn start(
        &self,
        room_id: &str,
        params: CompositeRecordingStartParams,
    ) -> Result<EgressHandle> {
        let body = CompositeRecordingStartWire {
            room_id,
            participants: &params.participants,
            watermarks: &params.watermarks,
            webhook_url: params.webhook_url.as_deref(),
            bucket_dir_path: params.dir_path.as_deref(),
            pre_signed_url: params.pre_signed_url.as_deref(),
            file_format: params.file_format.as_deref(),
        };
        let path = format!("{COMPOSITE_PATH}/start");
        let raw = self
            .client
            .maybe_json(Method::POST, &path, CallOptions::json(&body)?)
            .await?;
        Ok(to_egress_handle(EgressType::Composite, room_id, raw))
    }

    /// Stops a composite recording.
    ///
    /// Unlike every other egress, this needs the `recordingId`, so it takes the
    /// handle that [`start`](RecordingCompositeResource::start) returned rather
    /// than a bare room id.
    pub async fn stop(&self, handle: &EgressHandle) -> Result<String> {
        let id = handle
            .id
            .as_deref()
            .filter(|id| !id.is_empty())
            .ok_or_else(|| {
                Error::validation(
                    "recordings.composite().stop() requires a recordingId, from the start handle",
                )
            })?;
        let body = serde_json::json!({"roomId": handle.room_id, "recordingId": id});
        let path = format!("{COMPOSITE_PATH}/stop");
        self.client
            .message(Method::POST, &path, CallOptions::json(&body)?)
            .await
    }

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

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

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

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

    fn fetcher(&self, params: &CompositeRecordingListParams) -> 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())
                    .into_pairs();
                client
                    .json::<Value>(Method::GET, COMPOSITE_PATH, CallOptions::new().query(query))
                    .await
            })
        })
    }
}

/// Muxes individual recordings into stereo channels. Reached via
/// [`RecordingsResource::merge`].
#[derive(Debug, Clone, Copy)]
pub struct MergeRecordingResource<'a> {
    client: &'a Client,
}

impl<'a> MergeRecordingResource<'a> {
    /// Starts a two-channel audio-merge job.
    pub async fn create(&self, params: MergeRecordingCreateParams) -> Result<MergeRecordingResult> {
        self.client
            .json(Method::POST, MERGE_PATH, CallOptions::json(&params)?)
            .await
    }

    /// Lists merge recordings, one page at a time.
    pub async fn list(&self, params: MergeRecordingListParams) -> Result<Page<MergeRecording>> {
        // This endpoint keys its array `recordings`, not `data`.
        paginate(
            self.fetcher(&params),
            &params.pagination(),
            "recordings",
            None,
        )
        .await
    }

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

    /// Fetches a merge recording by its muxer id.
    pub async fn get(&self, muxer_id: &str) -> Result<MergeRecording> {
        let path = format!("{MERGE_PATH}/{}", escape(muxer_id));
        self.client
            .json(Method::GET, &path, CallOptions::new())
            .await
    }

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

/// Shared by the participant and track lists, which take identical filters.
fn individual_fetcher(
    client: &Client,
    path: &'static str,
    params: &IndividualRecordingListParams,
) -> PageFetcher {
    let client = 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("participantId", params.participant_id.as_deref())
                .opt_str("userId", params.user_id.as_deref())
                .into_pairs();
            client
                .json::<Value>(Method::GET, path, CallOptions::new().query(query))
                .await
        })
    })
}

/* ----------------------------------- facade ------------------------------------ */

/// The recordings API. Reached via [`Client::recordings`].
///
/// Its own methods drive the room (composed) recording; the sub-resources cover
/// the rest. Starting and stopping require a live session.
#[derive(Debug, Clone, Copy)]
pub struct RecordingsResource<'a> {
    client: &'a Client,
}

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

    /// Records individual participants.
    pub fn participant(&self) -> RecordingParticipantResource<'a> {
        RecordingParticipantResource {
            client: self.client,
        }
    }

    /// Records individual audio, video or screen tracks.
    pub fn track(&self) -> RecordingTrackResource<'a> {
        RecordingTrackResource {
            client: self.client,
        }
    }

    /// Records a server-side composed mix.
    pub fn composite(&self) -> RecordingCompositeResource<'a> {
        RecordingCompositeResource {
            client: self.client,
        }
    }

    /// Muxes individual recordings into stereo channels.
    pub fn merge(&self) -> MergeRecordingResource<'a> {
        MergeRecordingResource {
            client: self.client,
        }
    }

    /// Begins a room (composed) recording. Requires a live session.
    pub async fn start(&self, room_id: &str, params: RecordingStartParams) -> Result<EgressHandle> {
        let mapped = composition_to_config(params.composition.as_ref(), None);
        let body = RecordingStartWire {
            room_id,
            config: mapped.config,
            template_url: mapped.template_url,
            aws_dir_path: params.dir_path.as_deref(),
            transcription: params.transcription.as_ref(),
            on_failure: params.on_failure.as_ref(),
            file_format: params.file_format,
            webhook_url: params.webhook_url.as_deref(),
            resource_id: params.resource_id.as_deref(),
            pre_signed_url: params.pre_signed_url.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::Recording, room_id, raw))
    }

    /// Stops a room recording. 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 room recordings, one page at a time.
    pub async fn list(&self, params: RecordingListParams) -> Result<Page<Recording>> {
        paginate(self.fetcher(&params), &params.pagination(), "data", None).await
    }

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

    /// Fetches a room recording by id.
    pub async fn get(&self, id: &str, params: RecordingGetParams) -> Result<Recording> {
        let query = QueryBuilder::new()
            .opt("withTranscription", params.with_transcription)
            .into_pairs();
        let path = format!("{PATH}/{}", escape(id));
        self.client
            .json(Method::GET, &path, CallOptions::new().query(query))
            .await
    }

    /// Soft-deletes a room 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: &RecordingListParams) -> 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())
                    .opt("withTranscription", params.with_transcription)
                    .into_pairs();
                client
                    .json::<Value>(Method::GET, PATH, CallOptions::new().query(query))
                    .await
            })
        })
    }
}