//! Audio/video streaming: scan-status gating, subscription + bundle access, //! response shape, play counters. //! //! Complements `storage.rs` (which covers the core access matrix: //! free vs paid, anonymous vs authenticated, creator preview, draft //! visibility) and `video.rs` (video-specific happy paths). This file //! fills the remaining gaps the Run 27 audit flagged for the streaming //! feature: //! //! - scan_status variations (Pending, Quarantined, HeldForReview) //! gate on creator identity //! - Subscription-based access on paid items //! - Bundle parent grants access to child items //! - Item missing audio_s3_key / video_s3_key returns 404 //! - Response shape: `stream_url` + `expires_in` fields //! - `play_count` increments on stream //! - `unique_play` tracking for authenticated viewers //! - Version download surfaces `license_url` when item has a license preset use crate::harness::TestHarness; use serde_json::Value; /// Create a creator with a published audio item. Returns /// (user_id, project_id, item_id, s3_key). The InMemoryStorage is /// seeded so `presign_download` returns a usable URL. async fn setup_audio_item(h: &mut TestHarness, price_cents: i64) -> (String, String, String, String) { let setup = h.create_creator_with_item("streamer", "audio", price_cents).await; let s3_key = format!("test/{}/audio/track.mp3", setup.item_id); sqlx::query( "UPDATE items SET audio_s3_key = $1, scan_status = 'clean', is_public = true \ WHERE id = $2::uuid", ) .bind(&s3_key) .bind(&setup.item_id) .execute(&h.db) .await .unwrap(); sqlx::query("UPDATE projects SET is_public = true WHERE id = $1::uuid") .bind(&setup.project_id) .execute(&h.db) .await .unwrap(); h.storage.as_ref().unwrap().put(&s3_key, b"audio data".to_vec()); (setup.user_id.to_string(), setup.project_id, setup.item_id, s3_key) } #[tokio::test] async fn stream_url_response_has_expected_shape() { let mut h = TestHarness::with_storage().await; let (_, _, item_id, _) = setup_audio_item(&mut h, 0).await; let resp = h.client.get(&format!("/api/stream/{item_id}")).await; assert!(resp.status.is_success(), "{} {}", resp.status, resp.text); let data: Value = resp.json(); assert!(data["stream_url"].is_string(), "Response must contain stream_url"); assert!(data["expires_in"].is_u64(), "Response must contain numeric expires_in"); // Test storage backend returns http://test-storage/. assert!( data["stream_url"].as_str().unwrap().contains("audio/track.mp3"), "URL should reference the seeded key" ); } #[tokio::test] async fn stream_url_404_when_item_has_no_audio_key() { let mut h = TestHarness::with_storage().await; let setup = h.create_creator_with_item("noaudio", "audio", 0).await; // Mark the item public + clean but DON'T set audio_s3_key — the // handler should refuse to mint a streaming URL for a "naked" item. sqlx::query("UPDATE items SET is_public = true, scan_status = 'clean' WHERE id = $1::uuid") .bind(&setup.item_id) .execute(&h.db) .await .unwrap(); sqlx::query("UPDATE projects SET is_public = true WHERE id = $1::uuid") .bind(&setup.project_id) .execute(&h.db) .await .unwrap(); let resp = h.client.get(&format!("/api/stream/{}", setup.item_id)).await; assert_eq!(resp.status.as_u16(), 404, "Item without audio_s3_key must 404"); } #[tokio::test] async fn stream_url_404_for_quarantined_item_to_non_creator() { let mut h = TestHarness::with_storage().await; let (_, _, item_id, _) = setup_audio_item(&mut h, 0).await; sqlx::query("UPDATE items SET scan_status = 'quarantined' WHERE id = $1::uuid") .bind(&item_id) .execute(&h.db) .await .unwrap(); // Log out — the creator is currently authenticated. h.client.post_form("/logout", "").await; let resp = h.client.get(&format!("/api/stream/{item_id}")).await; assert_eq!( resp.status.as_u16(), 404, "Quarantined items must not stream to non-creators" ); } #[tokio::test] async fn stream_url_404_for_pending_scan_to_non_creator() { let mut h = TestHarness::with_storage().await; let (_, _, item_id, _) = setup_audio_item(&mut h, 0).await; sqlx::query("UPDATE items SET scan_status = 'pending' WHERE id = $1::uuid") .bind(&item_id) .execute(&h.db) .await .unwrap(); h.client.post_form("/logout", "").await; let resp = h.client.get(&format!("/api/stream/{item_id}")).await; assert_eq!( resp.status.as_u16(), 404, "Pending-scan items must not stream to non-creators (fail-closed)" ); } #[tokio::test] async fn stream_url_creator_can_preview_held_for_review_item() { let mut h = TestHarness::with_storage().await; let (_creator, _, item_id, _) = setup_audio_item(&mut h, 0).await; sqlx::query("UPDATE items SET scan_status = 'held_for_review' WHERE id = $1::uuid") .bind(&item_id) .execute(&h.db) .await .unwrap(); // Creator remains logged in from setup. let resp = h.client.get(&format!("/api/stream/{item_id}")).await; assert!( resp.status.is_success(), "Creator must be able to preview their own HeldForReview content: {} {}", resp.status, resp.text ); } #[tokio::test] async fn stream_url_authenticated_non_buyer_gets_403_on_paid_item() { let mut h = TestHarness::with_storage().await; let (_creator, _, item_id, _) = setup_audio_item(&mut h, 999).await; // Log out creator; create a different user with no purchase. h.client.post_form("/logout", "").await; h.signup("randomuser", "random@test.com", "password123").await; h.login("randomuser", "password123").await; let resp = h.client.get(&format!("/api/stream/{item_id}")).await; assert_eq!( resp.status.as_u16(), 403, "Authenticated non-buyer on paid item must get 403, not 401: {} {}", resp.status, resp.text ); } #[tokio::test] async fn stream_url_increments_play_count() { let mut h = TestHarness::with_storage().await; let (_, _, item_id, _) = setup_audio_item(&mut h, 0).await; // Stream twice (free item — anyone can hit it). h.client.post_form("/logout", "").await; for _ in 0..2 { let resp = h.client.get(&format!("/api/stream/{item_id}")).await; assert!(resp.status.is_success()); } let count: i32 = sqlx::query_scalar("SELECT play_count FROM items WHERE id = $1::uuid") .bind(&item_id) .fetch_one(&h.db) .await .unwrap(); assert_eq!(count, 2, "play_count should increment per stream call"); } #[tokio::test] async fn stream_url_records_unique_play_for_authenticated_user() { let mut h = TestHarness::with_storage().await; let (_creator, _, item_id, _) = setup_audio_item(&mut h, 0).await; // Switch to a different authenticated user. h.client.post_form("/logout", "").await; let listener_id = h.signup("listener", "listener@test.com", "password123").await; h.login("listener", "password123").await; let resp = h.client.get(&format!("/api/stream/{item_id}")).await; assert!(resp.status.is_success()); let has_play: bool = sqlx::query_scalar( "SELECT EXISTS(SELECT 1 FROM user_plays WHERE user_id = $1 AND item_id = $2::uuid)", ) .bind(listener_id) .bind(&item_id) .fetch_one(&h.db) .await .unwrap(); assert!(has_play, "user_plays should record the listener for unique-listener tracking"); } #[tokio::test] async fn version_download_includes_license_url_when_preset_set() { let mut h = TestHarness::with_storage().await; let setup = h.create_creator_with_item("licdownloader", "digital", 0).await; sqlx::query( "UPDATE items SET is_public = true, scan_status = 'clean', \ license_preset = 'mit' WHERE id = $1::uuid", ) .bind(&setup.item_id) .execute(&h.db) .await .unwrap(); sqlx::query("UPDATE projects SET is_public = true WHERE id = $1::uuid") .bind(&setup.project_id) .execute(&h.db) .await .unwrap(); // Create a version with a fake s3_key. let s3_key = format!("test/{}/download/build.zip", setup.item_id); h.storage.as_ref().unwrap().put(&s3_key, b"zip data".to_vec()); let version_id: String = sqlx::query_scalar( "INSERT INTO versions (item_id, version_number, s3_key, file_size_bytes, file_name, \ is_current, scan_status) \ VALUES ($1::uuid, '1.0', $2, 100, 'build.zip', true, 'clean') RETURNING id::text", ) .bind(&setup.item_id) .bind(&s3_key) .fetch_one(&h.db) .await .unwrap(); let resp = h.client.get(&format!("/api/versions/{version_id}/download")).await; assert!(resp.status.is_success(), "{} {}", resp.status, resp.text); let data: Value = resp.json(); assert!( data["license_url"].is_string(), "license_url should be present when item.license_preset is set" ); let license_url = data["license_url"].as_str().unwrap(); assert!( license_url.contains(&setup.item_id), "license_url should point at the item: {license_url}" ); assert!( license_url.ends_with("/license.txt"), "license_url should target the .txt endpoint: {license_url}" ); } #[tokio::test] async fn version_download_omits_license_url_without_preset() { let mut h = TestHarness::with_storage().await; let setup = h.create_creator_with_item("nolicdl", "digital", 0).await; sqlx::query( "UPDATE items SET is_public = true, scan_status = 'clean', \ license_preset = NULL WHERE id = $1::uuid", ) .bind(&setup.item_id) .execute(&h.db) .await .unwrap(); sqlx::query("UPDATE projects SET is_public = true WHERE id = $1::uuid") .bind(&setup.project_id) .execute(&h.db) .await .unwrap(); let s3_key = format!("test/{}/download/build.zip", setup.item_id); h.storage.as_ref().unwrap().put(&s3_key, b"zip data".to_vec()); let version_id: String = sqlx::query_scalar( "INSERT INTO versions (item_id, version_number, s3_key, file_size_bytes, file_name, \ is_current, scan_status) \ VALUES ($1::uuid, '1.0', $2, 100, 'build.zip', true, 'clean') RETURNING id::text", ) .bind(&setup.item_id) .bind(&s3_key) .fetch_one(&h.db) .await .unwrap(); let resp = h.client.get(&format!("/api/versions/{version_id}/download")).await; assert!(resp.status.is_success()); let data: Value = resp.json(); assert!( data["license_url"].is_null(), "license_url should be absent when no preset is set, got: {:?}", data["license_url"] ); } #[tokio::test] async fn stream_url_404_for_nonexistent_item() { let mut h = TestHarness::with_storage().await; let bogus = "00000000-0000-0000-0000-000000000000"; let resp = h.client.get(&format!("/api/stream/{bogus}")).await; assert_eq!(resp.status.as_u16(), 404); } #[tokio::test] async fn stream_url_expires_in_scales_with_duration() { let mut h = TestHarness::with_storage().await; let (_, _, item_id, _) = setup_audio_item(&mut h, 0).await; // Set a long duration. The handler computes expiry as // `max(duration * 2, 3600)`, so 5000s should give 10000s expiry. sqlx::query("UPDATE items SET duration_seconds = 5000 WHERE id = $1::uuid") .bind(&item_id) .execute(&h.db) .await .unwrap(); let resp = h.client.get(&format!("/api/stream/{item_id}")).await; assert!(resp.status.is_success(), "{} {}", resp.status, resp.text); let data: Value = resp.json(); let expires_in = data["expires_in"].as_u64().unwrap(); assert_eq!( expires_in, 10000, "expires_in should be 2x duration for long tracks (got {expires_in})" ); }