videosdk-server-sdk 0.1.0

//! End-to-end tests for the Phase 3 egress resources, against a mock HTTP server.

use futures_util::StreamExt as _;
use serde_json::{json, Value};
use wiremock::matchers::{body_json, method, path, query_param};
use wiremock::{Mock, MockServer, ResponseTemplate};

use crate::common::{CompositionQuality, TrackKind, TranscriptionConfig};
use crate::resources::egress::{Composition, CompositionLayout};
use crate::resources::hls::{HlsCaptureFormat, HlsCaptureParams, HlsListParams, HlsStartParams};
use crate::resources::recordings::{
    CompositeRecordingStartParams, MergeChannelEntry, MergeRecordingCreateParams,
    MergeRecordingListParams, ParticipantRecordingStartParams, RecordingGetParams,
    RecordingStartParams, TrackRecordingStartParams,
};
use crate::resources::rtmp::{RtmpStartParams, RtmpStream};
use crate::test_support::{body, client, only_request, requests};

async fn mount(server: &MockServer, verb: &str, route: &str, response: Value) {
    Mock::given(method(verb))
        .and(path(route))
        .respond_with(ResponseTemplate::new(200).set_body_json(response))
        .mount(server)
        .await;
}

/// Mounts a route answering with a bare confirmation string, as the egress
/// endpoints often do.
async fn mount_message(server: &MockServer, verb: &str, route: &str, message: &str) {
    Mock::given(method(verb))
        .and(path(route))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!(message)))
        .mount(server)
        .await;
}

/* ------------------------------- room recording ------------------------------- */

#[tokio::test]
async fn starting_a_room_recording_renames_dir_path_to_aws_dir_path() {
    let server = MockServer::start().await;
    Mock::given(method("POST"))
        .and(path("/v2/recordings/start"))
        .and(body_json(json!({
            "roomId": "r-1",
            "config": {"layout": {"type": "GRID"}},
            "awsDirPath": "recordings/",
            "preSignedUrl": "https://s3/put",
            "fileFormat": "mp4",
        })))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!("Recording started")))
        .expect(1)
        .mount(&server)
        .await;

    let handle = client(&server)
        .recordings()
        .start(
            "r-1",
            RecordingStartParams {
                composition: Some(Composition {
                    layout: Some(CompositionLayout::Grid),
                    ..Default::default()
                }),
                dir_path: Some("recordings/".into()),
                // Lowercase `url` for a room recording.
                pre_signed_url: Some("https://s3/put".into()),
                file_format: Some(crate::RecordingFileFormat::Mp4),
                ..Default::default()
            },
        )
        .await
        .unwrap();

    // A bare-string start response still yields a room-keyed handle.
    assert_eq!(handle.room_id, "r-1");
    assert!(handle.id.is_none());
}

#[tokio::test]
async fn a_room_recording_sends_no_default_priority() {
    let server = MockServer::start().await;
    Mock::given(method("POST"))
        .and(body_json(json!({
            "roomId": "r-1",
            "config": {"layout": {"type": "SPOTLIGHT"}},
        })))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!("ok")))
        .expect(1)
        .mount(&server)
        .await;

    client(&server)
        .recordings()
        .start(
            "r-1",
            RecordingStartParams {
                composition: Some(Composition {
                    layout: Some(CompositionLayout::Spotlight),
                    ..Default::default()
                }),
                ..Default::default()
            },
        )
        .await
        .unwrap();
}

#[tokio::test]
async fn stopping_a_room_recording_accepts_a_handle_or_a_bare_room_id() {
    let server = MockServer::start().await;
    mount_message(&server, "POST", "/v2/recordings/end", "Stopped").await;

    let client = client(&server);
    assert_eq!(client.recordings().stop("r-1").await.unwrap(), "Stopped");
    assert_eq!(body(&only_request(&server).await), json!({"roomId": "r-1"}));
}

#[tokio::test]
async fn getting_a_room_recording_forwards_with_transcription() {
    let server = MockServer::start().await;
    Mock::given(method("GET"))
        .and(path("/v2/recordings/rec-1"))
        .and(query_param("withTranscription", "true"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({"id": "rec-1"})))
        .expect(1)
        .mount(&server)
        .await;

    let recording = client(&server)
        .recordings()
        .get(
            "rec-1",
            RecordingGetParams {
                with_transcription: Some(true),
            },
        )
        .await
        .unwrap();
    assert_eq!(recording.id, "rec-1");
}

#[tokio::test]
async fn listing_room_recordings_streams_across_pages() {
    let server = MockServer::start().await;
    let page = |current: u32, id: &str| {
        json!({
            "pageInfo": {"currentPage": current, "perPage": 1, "lastPage": 2, "total": 2},
            "data": [{"id": id}],
        })
    };
    Mock::given(method("GET"))
        .and(query_param("page", "2"))
        .respond_with(ResponseTemplate::new(200).set_body_json(page(2, "rec-2")))
        .mount(&server)
        .await;
    Mock::given(method("GET"))
        .respond_with(ResponseTemplate::new(200).set_body_json(page(1, "rec-1")))
        .mount(&server)
        .await;

    let client = client(&server);
    let ids: Vec<String> = client
        .recordings()
        .list_stream(Default::default())
        .map(|r| r.unwrap().id)
        .collect()
        .await;
    assert_eq!(ids, ["rec-1", "rec-2"]);
}

/* --------------------------- participant / track ------------------------------ */

#[tokio::test]
async fn a_participant_recording_renames_dir_path_to_bucket_dir_path() {
    let server = MockServer::start().await;
    Mock::given(method("POST"))
        .and(path("/v2/recordings/participant/start"))
        .and(body_json(json!({
            "roomId": "r-1",
            "participantId": "p-1",
            "bucketDirPath": "out/",
            "fileFormat": "webm",
            "metadata": {"tag": "demo"},
        })))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!("Started")))
        .expect(1)
        .mount(&server)
        .await;

    let message = client(&server)
        .recordings()
        .participant()
        .start(
            "r-1",
            ParticipantRecordingStartParams {
                participant_id: "p-1".into(),
                dir_path: Some("out/".into()),
                file_format: Some(crate::ParticipantFileFormat::Webm),
                metadata: Some(json!({"tag": "demo"})),
                ..Default::default()
            },
        )
        .await
        .unwrap();
    assert_eq!(message, "Started");
}

#[tokio::test]
async fn a_track_recording_sends_the_kind_and_bucket_dir_path() {
    let server = MockServer::start().await;
    Mock::given(method("POST"))
        .and(path("/v2/recordings/participant/track/start"))
        .and(body_json(json!({
            "roomId": "r-1",
            "participantId": "p-1",
            "kind": "screen_audio",
            "bucketDirPath": "out/",
        })))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!("Started")))
        .expect(1)
        .mount(&server)
        .await;

    client(&server)
        .recordings()
        .track()
        .start(
            "r-1",
            TrackRecordingStartParams {
                participant_id: "p-1".into(),
                kind: TrackKind::ScreenAudio,
                file_format: None,
                webhook_url: None,
                dir_path: Some("out/".into()),
            },
        )
        .await
        .unwrap();
}

#[tokio::test]
async fn stopping_a_track_recording_identifies_it_by_all_three_keys() {
    let server = MockServer::start().await;
    mount_message(
        &server,
        "POST",
        "/v2/recordings/participant/track/stop",
        "Stopped",
    )
    .await;

    client(&server)
        .recordings()
        .track()
        .stop("r-1", "p-1", TrackKind::Video)
        .await
        .unwrap();

    assert_eq!(
        body(&only_request(&server).await),
        json!({"roomId": "r-1", "participantId": "p-1", "kind": "video"})
    );
}

#[tokio::test]
async fn participant_and_track_lists_share_their_filters() {
    let server = MockServer::start().await;
    mount(
        &server,
        "GET",
        "/v2/recordings/participant",
        json!({"data": [{"id": "ind-1"}]}),
    )
    .await;

    let page = client(&server)
        .recordings()
        .participant()
        .list(crate::IndividualRecordingListParams {
            participant_id: Some("p-1".into()),
            room_id: Some("r-1".into()),
            ..Default::default()
        })
        .await
        .unwrap();

    assert_eq!(page.data[0].id, "ind-1");
    let request = only_request(&server).await;
    let query: std::collections::HashMap<_, _> = request.url.query_pairs().collect();
    assert_eq!(query["participantId"], "p-1");
    assert_eq!(query["roomId"], "r-1");
}

/* ---------------------------------- composite --------------------------------- */

#[tokio::test]
async fn a_composite_recording_uses_capital_url_in_pre_signed_url() {
    let server = MockServer::start().await;
    Mock::given(method("POST"))
        .and(path("/v2/recordings/composite/start"))
        .and(body_json(json!({
            "roomId": "r-1",
            "bucketDirPath": "out/",
            "preSignedURL": "https://s3/put",
        })))
        .respond_with(
            ResponseTemplate::new(200)
                .set_body_json(json!({"recordingId": "rec-9", "sessionId": "s-1"})),
        )
        .expect(1)
        .mount(&server)
        .await;

    let handle = client(&server)
        .recordings()
        .composite()
        .start(
            "r-1",
            CompositeRecordingStartParams {
                dir_path: Some("out/".into()),
                // Capital URL for composite, unlike the room recording.
                pre_signed_url: Some("https://s3/put".into()),
                ..Default::default()
            },
        )
        .await
        .unwrap();

    assert_eq!(handle.id.as_deref(), Some("rec-9"));
    assert_eq!(handle.session_id.as_deref(), Some("s-1"));
}

#[tokio::test]
async fn stopping_a_composite_recording_requires_the_recording_id() {
    let server = MockServer::start().await;
    let handle = crate::resources::egress::to_egress_handle(
        crate::EgressType::Composite,
        "r-1",
        json!("no id here"),
    );

    let err = client(&server)
        .recordings()
        .composite()
        .stop(&handle)
        .await
        .unwrap_err();

    assert!(err.is_validation());
    assert!(err.to_string().contains("recordingId"));
    assert!(requests(&server).await.is_empty());
}

#[tokio::test]
async fn stopping_a_composite_recording_sends_the_recording_id() {
    let server = MockServer::start().await;
    mount_message(&server, "POST", "/v2/recordings/composite/stop", "Stopped").await;

    let handle = crate::resources::egress::to_egress_handle(
        crate::EgressType::Composite,
        "r-1",
        json!({"recordingId": "rec-9"}),
    );
    client(&server)
        .recordings()
        .composite()
        .stop(&handle)
        .await
        .unwrap();

    assert_eq!(
        body(&only_request(&server).await),
        json!({"roomId": "r-1", "recordingId": "rec-9"})
    );
}

/* ------------------------------------ merge ----------------------------------- */

#[tokio::test]
async fn merge_keeps_dir_path_unrenamed_and_reads_the_recordings_key() {
    let server = MockServer::start().await;
    Mock::given(method("POST"))
        .and(path("/v2/recordings/participant/merge"))
        .and(body_json(json!({
            "sessionId": "s-1",
            "channel1": [{"participantId": "p-1"}],
            "channel2": [{"participantId": "p-2", "type": "track"}],
            "dirPath": "merged/",
        })))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "message": "queued", "recording": {"id": "m-1"},
        })))
        .expect(1)
        .mount(&server)
        .await;

    let result = client(&server)
        .recordings()
        .merge()
        .create(MergeRecordingCreateParams {
            session_id: "s-1".into(),
            channel1: vec![MergeChannelEntry {
                participant_id: "p-1".into(),
                recording_id: None,
                kind: None,
            }],
            channel2: vec![MergeChannelEntry {
                participant_id: "p-2".into(),
                recording_id: None,
                kind: Some("track".into()),
            }],
            // `dirPath` here, not `bucketDirPath` or `awsDirPath`.
            dir_path: Some("merged/".into()),
            ..Default::default()
        })
        .await
        .unwrap();

    assert_eq!(result.message, "queued");
    assert_eq!(result.recording.id, "m-1");
}

#[tokio::test]
async fn merge_list_reads_the_recordings_data_key() {
    let server = MockServer::start().await;
    mount(
        &server,
        "GET",
        "/v2/recordings/participant/merge",
        json!({"recordings": [{"id": "m-1"}, {"id": "m-2"}]}),
    )
    .await;

    let page = client(&server)
        .recordings()
        .merge()
        .list(MergeRecordingListParams {
            status: Some("done".into()),
            ..Default::default()
        })
        .await
        .unwrap();
    assert_eq!(page.data.len(), 2);
    assert_eq!(page.data[1].id, "m-2");
}

/* ------------------------------------- hls ------------------------------------ */

#[tokio::test]
async fn hls_start_defaults_the_priority_to_speaker() {
    let server = MockServer::start().await;
    Mock::given(method("POST"))
        .and(path("/v2/hls/start"))
        .and(body_json(json!({
            "roomId": "r-1",
            "config": {"layout": {"type": "GRID", "priority": "SPEAKER"}},
            "transcription": {"enabled": true},
        })))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({"id": "h-1"})))
        .expect(1)
        .mount(&server)
        .await;

    let handle = client(&server)
        .hls()
        .start(
            "r-1",
            HlsStartParams {
                composition: Some(Composition {
                    layout: Some(CompositionLayout::Grid),
                    ..Default::default()
                }),
                transcription: Some(TranscriptionConfig {
                    enabled: Some(true),
                    summary: None,
                }),
                ..Default::default()
            },
        )
        .await
        .unwrap();
    assert_eq!(handle.id.as_deref(), Some("h-1"));
}

#[tokio::test]
async fn a_custom_composition_layout_overrides_an_explicit_template_url() {
    let server = MockServer::start().await;
    Mock::given(method("POST"))
        .and(path("/v2/hls/start"))
        .and(body_json(json!({
            "roomId": "r-1",
            "templateUrl": "https://example.com/from-composition",
        })))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!("started")))
        .expect(1)
        .mount(&server)
        .await;

    client(&server)
        .hls()
        .start(
            "r-1",
            HlsStartParams {
                composition: Some(Composition {
                    layout: Some(CompositionLayout::custom(
                        "https://example.com/from-composition",
                    )),
                    ..Default::default()
                }),
                template_url: Some("https://example.com/explicit".into()),
                ..Default::default()
            },
        )
        .await
        .unwrap();
}

#[tokio::test]
async fn hls_start_injects_grid_for_a_layoutless_config() {
    let server = MockServer::start().await;
    Mock::given(method("POST"))
        .and(body_json(json!({
            "roomId": "r-1",
            // The backend 406s a non-empty config with no layout.
            "config": {"layout": {"type": "GRID", "priority": "SPEAKER"}, "quality": "high"},
        })))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!("started")))
        .expect(1)
        .mount(&server)
        .await;

    client(&server)
        .hls()
        .start(
            "r-1",
            HlsStartParams {
                composition: Some(Composition {
                    quality: Some(CompositionQuality::High),
                    ..Default::default()
                }),
                ..Default::default()
            },
        )
        .await
        .unwrap();
}

#[tokio::test]
async fn hls_get_unwraps_data_and_folds_the_meeting_id_alias() {
    let server = MockServer::start().await;
    mount(
        &server,
        "GET",
        "/v2/hls/r-1/active",
        json!({"data": {
            "id": "h-1",
            "meetingId": "r-1",
            "playbackHlsUrl": "https://cdn/live.m3u8",
            "captureLog": [{"at": 1}],
            "encryption": {"method": "AES-128"},
        }}),
    )
    .await;

    let stream = client(&server).hls().get("r-1").await.unwrap();
    assert_eq!(stream.room_id.as_deref(), Some("r-1"));
    assert_eq!(
        stream.playback_hls_url.as_deref(),
        Some("https://cdn/live.m3u8")
    );
    // Restored from the TypeScript SDK; the Go port drops both.
    assert_eq!(stream.capture_log.len(), 1);
    assert!(stream.encryption.is_some());
}

#[tokio::test]
async fn hls_get_reports_not_found_when_no_stream_is_active() {
    let server = MockServer::start().await;
    mount(&server, "GET", "/v2/hls/r-1/active", json!({"data": null})).await;

    let err = client(&server).hls().get("r-1").await.unwrap_err();
    assert!(err.is_not_found());
    assert!(err.to_string().contains("no active HLS stream"));
}

#[tokio::test]
async fn hls_capture_posts_its_params() {
    let server = MockServer::start().await;
    Mock::given(method("POST"))
        .and(path("/v2/hls/capture"))
        .and(body_json(
            json!({"roomId": "r-1", "format": "jpg", "width": 640}),
        ))
        .respond_with(
            ResponseTemplate::new(200)
                .set_body_json(json!({"message": "captured", "roomId": "r-1"})),
        )
        .expect(1)
        .mount(&server)
        .await;

    let result = client(&server)
        .hls()
        .capture(HlsCaptureParams {
            room_id: "r-1".into(),
            format: Some(HlsCaptureFormat::Jpg),
            width: Some(640),
            ..Default::default()
        })
        .await
        .unwrap();
    assert_eq!(result.message, "captured");
}

#[tokio::test]
async fn hls_stop_and_delete_hit_their_routes() {
    let server = MockServer::start().await;
    mount_message(&server, "POST", "/v2/hls/end", "Stopped").await;
    mount_message(&server, "DELETE", "/v2/hls/h-1", "Deleted").await;

    let client = client(&server);
    assert_eq!(client.hls().stop("r-1").await.unwrap(), "Stopped");
    assert_eq!(client.hls().delete("h-1").await.unwrap(), "Deleted");
}

#[tokio::test]
async fn hls_list_forwards_its_filters() {
    let server = MockServer::start().await;
    mount(&server, "GET", "/v2/hls", json!({"data": []})).await;

    client(&server)
        .hls()
        .list(HlsListParams {
            room_id: Some("r-1".into()),
            query: Some("abcd".into()),
            ..Default::default()
        })
        .await
        .unwrap();

    let request = only_request(&server).await;
    let query: std::collections::HashMap<_, _> = request.url.query_pairs().collect();
    assert_eq!(query["roomId"], "r-1");
    assert_eq!(query["query"], "abcd");
}

/* ------------------------------------ rtmp ------------------------------------ */

#[tokio::test]
async fn rtmp_start_renames_streams_to_outputs() {
    let server = MockServer::start().await;
    Mock::given(method("POST"))
        .and(path("/v2/livestreams/start"))
        .and(body_json(json!({
            "roomId": "r-1",
            "outputs": [{"url": "rtmp://a/live", "streamKey": "key-1"}],
            "config": {"layout": {"type": "SIDEBAR", "priority": "SPEAKER"}},
        })))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({"id": "ls-1"})))
        .expect(1)
        .mount(&server)
        .await;

    let handle = client(&server)
        .rtmp()
        .start(
            "r-1",
            RtmpStartParams {
                streams: vec![RtmpStream {
                    url: "rtmp://a/live".into(),
                    stream_key: "key-1".into(),
                }],
                composition: Some(Composition {
                    layout: Some(CompositionLayout::Sidebar),
                    ..Default::default()
                }),
                ..Default::default()
            },
        )
        .await
        .unwrap();
    assert_eq!(handle.id.as_deref(), Some("ls-1"));
}

#[tokio::test]
async fn rtmp_start_requires_at_least_one_destination() {
    let server = MockServer::start().await;
    let err = client(&server)
        .rtmp()
        .start("r-1", RtmpStartParams::default())
        .await
        .unwrap_err();

    assert!(err.is_validation());
    assert!(err.to_string().contains("streams"));
    assert!(requests(&server).await.is_empty());
}

#[tokio::test]
async fn rtmp_stop_accepts_the_start_handle() {
    let server = MockServer::start().await;
    mount(
        &server,
        "POST",
        "/v2/livestreams/start",
        json!({"id": "ls-1"}),
    )
    .await;
    mount_message(&server, "POST", "/v2/livestreams/end", "Stopped").await;

    let client = client(&server);
    let handle = client
        .rtmp()
        .start(
            "r-1",
            RtmpStartParams {
                streams: vec![RtmpStream {
                    url: "rtmp://a/live".into(),
                    stream_key: "k".into(),
                }],
                ..Default::default()
            },
        )
        .await
        .unwrap();

    client.rtmp().stop(&handle).await.unwrap();

    let stop = requests(&server).await.pop().unwrap();
    assert_eq!(body(&stop), json!({"roomId": "r-1", "id": "ls-1"}));
}

#[tokio::test]
async fn rtmp_get_returns_its_outputs() {
    let server = MockServer::start().await;
    mount(
        &server,
        "GET",
        "/v2/livestreams/ls-1",
        json!({"id": "ls-1", "outputs": [{"url": "rtmp://a", "streamKey": "k"}]}),
    )
    .await;

    let livestream = client(&server).rtmp().get("ls-1").await.unwrap();
    assert_eq!(livestream.outputs[0].stream_key, "k");
}