Skip to main content

max / makenotwork

15.5 KB · 462 lines History Blame Raw
1 //! Audio/video streaming: scan-status gating, subscription + bundle access,
2 //! response shape, play counters.
3 //!
4 //! Complements `storage.rs` (which covers the core access matrix:
5 //! free vs paid, anonymous vs authenticated, creator preview, draft
6 //! visibility) and `video.rs` (video-specific happy paths). This file
7 //! fills the remaining gaps the Run 27 audit flagged for the streaming
8 //! feature:
9 //!
10 //! - scan_status variations (Pending, Quarantined, HeldForReview)
11 //! gate on creator identity
12 //! - Subscription-based access on paid items
13 //! - Bundle parent grants access to child items
14 //! - Item missing audio_s3_key / video_s3_key returns 404
15 //! - Response shape: `stream_url` + `expires_in` fields
16 //! - `play_count` increments on stream
17 //! - `unique_play` tracking for authenticated viewers
18 //! - Version download surfaces `license_url` when item has a license preset
19
20 use crate::harness::TestHarness;
21 use serde_json::Value;
22
23 /// Create a creator with a published audio item. Returns
24 /// (user_id, project_id, item_id, s3_key). The InMemoryStorage is
25 /// seeded so `presign_download` returns a usable URL.
26 async fn setup_audio_item(
27 h: &mut TestHarness,
28 price_cents: i64,
29 ) -> (String, String, String, String) {
30 let setup = h
31 .create_creator_with_item("streamer", "audio", price_cents)
32 .await;
33 let s3_key = format!("test/{}/audio/track.mp3", setup.item_id);
34 sqlx::query(
35 "UPDATE items SET audio_s3_key = $1, scan_status = 'clean', is_public = true \
36 WHERE id = $2::uuid",
37 )
38 .bind(&s3_key)
39 .bind(&setup.item_id)
40 .execute(&h.db)
41 .await
42 .unwrap();
43 sqlx::query("UPDATE projects SET is_public = true WHERE id = $1::uuid")
44 .bind(&setup.project_id)
45 .execute(&h.db)
46 .await
47 .unwrap();
48 h.storage
49 .as_ref()
50 .unwrap()
51 .put(&s3_key, b"audio data".to_vec());
52 (
53 setup.user_id.to_string(),
54 setup.project_id,
55 setup.item_id,
56 s3_key,
57 )
58 }
59
60 /// Regression pin for Payments S1 / CHRONIC 2: the download/stream access gate
61 /// MUST honor `current_period_end`. An item subscription still flagged
62 /// `status = 'active'` (its cancel/expiry webhook lapsed or never arrived) but
63 /// whose paid period has EXPIRED must not grant access. Before the seal,
64 /// `check_item_access` hand-wrote `status = 'active' AND paused_at IS NULL` and
65 /// dropped the period clause, so a lapsed subscriber could still pull the file
66 /// on the very route that protects content. The gate now routes through
67 /// `subscriptions::has_access`, which enforces the clause in one sealed place.
68 /// If anyone re-inlines the predicate without the period clause, the first
69 /// assertion (lapsed → 403) flips to 200 and this test fails.
70 #[tokio::test]
71 async fn stream_denies_item_subscription_with_lapsed_period() {
72 let mut h = TestHarness::with_storage().await;
73 // Paid audio item owned by the creator (price > 0 ⇒ access is gated).
74 let (_creator, _project, item_id, _s3) = setup_audio_item(&mut h, 500).await;
75 let item_uuid: uuid::Uuid = item_id.parse().unwrap();
76
77 // Item-level subscription tier (project_id NULL, item_id set).
78 let tier_id: uuid::Uuid = sqlx::query_scalar(
79 "INSERT INTO subscription_tiers (project_id, item_id, name, price_cents) \
80 VALUES (NULL, $1, 'Item Tier', 500) RETURNING id",
81 )
82 .bind(item_uuid)
83 .fetch_one(&h.db)
84 .await
85 .unwrap();
86
87 // A fan with a known password (seeded directly; we only need to log in).
88 let hash = makenotwork::auth::hash_password("password123").unwrap();
89 let fan_id: makenotwork::db::UserId = sqlx::query_scalar(
90 "INSERT INTO users (username, email, password_hash, email_verified) \
91 VALUES ('lapsedfan', 'lapsedfan@example.com', $1, true) RETURNING id",
92 )
93 .bind(&hash)
94 .fetch_one(&h.db)
95 .await
96 .unwrap();
97
98 // Active-status item subscription whose paid period ended a day ago.
99 sqlx::query(
100 "INSERT INTO subscriptions \
101 (subscriber_id, tier_id, item_id, project_id, stripe_subscription_id, \
102 stripe_customer_id, status, paused_at, current_period_end) \
103 VALUES ($1, $2, $3, NULL, 'sub_lapsed', 'cus_lapsed', 'active', NULL, NOW() - INTERVAL '1 day')",
104 )
105 .bind(fan_id)
106 .bind(tier_id)
107 .bind(item_uuid)
108 .execute(&h.db)
109 .await
110 .unwrap();
111
112 // Act as the fan (the creator is logged in from setup_audio_item).
113 h.client.post_form("/logout", "").await;
114 h.login("lapsedfan", "password123").await;
115
116 // Lapsed period → denied on the download gate.
117 let resp = h.client.get(&format!("/api/stream/{item_id}")).await;
118 assert_eq!(
119 resp.status.as_u16(),
120 403,
121 "lapsed-period subscriber must be denied the download gate, got {} {}",
122 resp.status,
123 resp.text
124 );
125
126 // Same subscriber, period pushed into the future → now passes.
127 sqlx::query(
128 "UPDATE subscriptions SET current_period_end = NOW() + INTERVAL '30 days' \
129 WHERE subscriber_id = $1",
130 )
131 .bind(fan_id)
132 .execute(&h.db)
133 .await
134 .unwrap();
135
136 let resp = h.client.get(&format!("/api/stream/{item_id}")).await;
137 assert!(
138 resp.status.is_success(),
139 "subscriber within the paid period must pass the gate, got {} {}",
140 resp.status,
141 resp.text
142 );
143 }
144
145 #[tokio::test]
146 async fn stream_url_response_has_expected_shape() {
147 let mut h = TestHarness::with_storage().await;
148 let (_, _, item_id, _) = setup_audio_item(&mut h, 0).await;
149
150 let resp = h.client.get(&format!("/api/stream/{item_id}")).await;
151 assert!(resp.status.is_success(), "{} {}", resp.status, resp.text);
152
153 let data: Value = resp.json();
154 assert!(
155 data["stream_url"].is_string(),
156 "Response must contain stream_url"
157 );
158 assert!(
159 data["expires_in"].is_u64(),
160 "Response must contain numeric expires_in"
161 );
162 // Test storage backend returns http://test-storage/<key>.
163 assert!(
164 data["stream_url"]
165 .as_str()
166 .unwrap()
167 .contains("audio/track.mp3"),
168 "URL should reference the seeded key"
169 );
170 }
171
172 #[tokio::test]
173 async fn stream_url_404_when_item_has_no_audio_key() {
174 let mut h = TestHarness::with_storage().await;
175 let setup = h.create_creator_with_item("noaudio", "audio", 0).await;
176 // Mark the item public + clean but DON'T set audio_s3_key, the
177 // handler should refuse to mint a streaming URL for a "naked" item.
178 sqlx::query("UPDATE items SET is_public = true, scan_status = 'clean' WHERE id = $1::uuid")
179 .bind(&setup.item_id)
180 .execute(&h.db)
181 .await
182 .unwrap();
183 sqlx::query("UPDATE projects SET is_public = true WHERE id = $1::uuid")
184 .bind(&setup.project_id)
185 .execute(&h.db)
186 .await
187 .unwrap();
188
189 let resp = h
190 .client
191 .get(&format!("/api/stream/{}", setup.item_id))
192 .await;
193 assert_eq!(
194 resp.status.as_u16(),
195 404,
196 "Item without audio_s3_key must 404"
197 );
198 }
199
200 #[tokio::test]
201 async fn stream_url_404_for_quarantined_item_to_non_creator() {
202 let mut h = TestHarness::with_storage().await;
203 let (_, _, item_id, _) = setup_audio_item(&mut h, 0).await;
204 sqlx::query("UPDATE items SET scan_status = 'quarantined' WHERE id = $1::uuid")
205 .bind(&item_id)
206 .execute(&h.db)
207 .await
208 .unwrap();
209 // Log out, the creator is currently authenticated.
210 h.client.post_form("/logout", "").await;
211
212 let resp = h.client.get(&format!("/api/stream/{item_id}")).await;
213 assert_eq!(
214 resp.status.as_u16(),
215 404,
216 "Quarantined items must not stream to non-creators"
217 );
218 }
219
220 #[tokio::test]
221 async fn stream_url_404_for_pending_scan_to_non_creator() {
222 let mut h = TestHarness::with_storage().await;
223 let (_, _, item_id, _) = setup_audio_item(&mut h, 0).await;
224 sqlx::query("UPDATE items SET scan_status = 'pending' WHERE id = $1::uuid")
225 .bind(&item_id)
226 .execute(&h.db)
227 .await
228 .unwrap();
229 h.client.post_form("/logout", "").await;
230
231 let resp = h.client.get(&format!("/api/stream/{item_id}")).await;
232 assert_eq!(
233 resp.status.as_u16(),
234 404,
235 "Pending-scan items must not stream to non-creators (fail-closed)"
236 );
237 }
238
239 #[tokio::test]
240 async fn stream_url_creator_can_preview_held_for_review_item() {
241 let mut h = TestHarness::with_storage().await;
242 let (_creator, _, item_id, _) = setup_audio_item(&mut h, 0).await;
243 sqlx::query("UPDATE items SET scan_status = 'held_for_review' WHERE id = $1::uuid")
244 .bind(&item_id)
245 .execute(&h.db)
246 .await
247 .unwrap();
248 // Creator remains logged in from setup.
249
250 let resp = h.client.get(&format!("/api/stream/{item_id}")).await;
251 assert!(
252 resp.status.is_success(),
253 "Creator must be able to preview their own HeldForReview content: {} {}",
254 resp.status,
255 resp.text
256 );
257 }
258
259 #[tokio::test]
260 async fn stream_url_authenticated_non_buyer_gets_403_on_paid_item() {
261 let mut h = TestHarness::with_storage().await;
262 let (_creator, _, item_id, _) = setup_audio_item(&mut h, 999).await;
263 // Log out creator; create a different user with no purchase.
264 h.client.post_form("/logout", "").await;
265 h.signup("randomuser", "random@test.com", "password123")
266 .await;
267 h.login("randomuser", "password123").await;
268
269 let resp = h.client.get(&format!("/api/stream/{item_id}")).await;
270 assert_eq!(
271 resp.status.as_u16(),
272 403,
273 "Authenticated non-buyer on paid item must get 403, not 401: {} {}",
274 resp.status,
275 resp.text
276 );
277 }
278
279 #[tokio::test]
280 async fn stream_url_increments_play_count() {
281 let mut h = TestHarness::with_storage().await;
282 let (_, _, item_id, _) = setup_audio_item(&mut h, 0).await;
283
284 // Stream twice (free item, anyone can hit it).
285 h.client.post_form("/logout", "").await;
286 for _ in 0..2 {
287 let resp = h.client.get(&format!("/api/stream/{item_id}")).await;
288 assert!(resp.status.is_success());
289 }
290
291 let count: i32 = sqlx::query_scalar("SELECT play_count FROM items WHERE id = $1::uuid")
292 .bind(&item_id)
293 .fetch_one(&h.db)
294 .await
295 .unwrap();
296 assert_eq!(count, 2, "play_count should increment per stream call");
297 }
298
299 #[tokio::test]
300 async fn stream_url_records_unique_play_for_authenticated_user() {
301 let mut h = TestHarness::with_storage().await;
302 let (_creator, _, item_id, _) = setup_audio_item(&mut h, 0).await;
303
304 // Switch to a different authenticated user.
305 h.client.post_form("/logout", "").await;
306 let listener_id = h
307 .signup("listener", "listener@test.com", "password123")
308 .await;
309 h.login("listener", "password123").await;
310
311 let resp = h.client.get(&format!("/api/stream/{item_id}")).await;
312 assert!(resp.status.is_success());
313
314 let has_play: bool = sqlx::query_scalar(
315 "SELECT EXISTS(SELECT 1 FROM user_plays WHERE user_id = $1 AND item_id = $2::uuid)",
316 )
317 .bind(listener_id)
318 .bind(&item_id)
319 .fetch_one(&h.db)
320 .await
321 .unwrap();
322 assert!(
323 has_play,
324 "user_plays should record the listener for unique-listener tracking"
325 );
326 }
327
328 #[tokio::test]
329 async fn version_download_includes_license_url_when_preset_set() {
330 let mut h = TestHarness::with_storage().await;
331 let setup = h
332 .create_creator_with_item("licdownloader", "digital", 0)
333 .await;
334 sqlx::query(
335 "UPDATE items SET is_public = true, scan_status = 'clean', \
336 license_preset = 'mit' WHERE id = $1::uuid",
337 )
338 .bind(&setup.item_id)
339 .execute(&h.db)
340 .await
341 .unwrap();
342 sqlx::query("UPDATE projects SET is_public = true WHERE id = $1::uuid")
343 .bind(&setup.project_id)
344 .execute(&h.db)
345 .await
346 .unwrap();
347
348 // Create a version with a fake s3_key.
349 let s3_key = format!("test/{}/download/build.zip", setup.item_id);
350 h.storage
351 .as_ref()
352 .unwrap()
353 .put(&s3_key, b"zip data".to_vec());
354 let version_id: String = sqlx::query_scalar(
355 "INSERT INTO versions (item_id, version_number, s3_key, file_size_bytes, file_name, \
356 is_current, scan_status) \
357 VALUES ($1::uuid, '1.0', $2, 100, 'build.zip', true, 'clean') RETURNING id::text",
358 )
359 .bind(&setup.item_id)
360 .bind(&s3_key)
361 .fetch_one(&h.db)
362 .await
363 .unwrap();
364
365 let resp = h
366 .client
367 .get(&format!("/api/versions/{version_id}/download"))
368 .await;
369 assert!(resp.status.is_success(), "{} {}", resp.status, resp.text);
370 let data: Value = resp.json();
371 assert!(
372 data["license_url"].is_string(),
373 "license_url should be present when item.license_preset is set"
374 );
375 let license_url = data["license_url"].as_str().unwrap();
376 assert!(
377 license_url.contains(&setup.item_id),
378 "license_url should point at the item: {license_url}"
379 );
380 assert!(
381 license_url.ends_with("/license.txt"),
382 "license_url should target the .txt endpoint: {license_url}"
383 );
384 }
385
386 #[tokio::test]
387 async fn version_download_omits_license_url_without_preset() {
388 let mut h = TestHarness::with_storage().await;
389 let setup = h.create_creator_with_item("nolicdl", "digital", 0).await;
390 sqlx::query(
391 "UPDATE items SET is_public = true, scan_status = 'clean', \
392 license_preset = NULL WHERE id = $1::uuid",
393 )
394 .bind(&setup.item_id)
395 .execute(&h.db)
396 .await
397 .unwrap();
398 sqlx::query("UPDATE projects SET is_public = true WHERE id = $1::uuid")
399 .bind(&setup.project_id)
400 .execute(&h.db)
401 .await
402 .unwrap();
403
404 let s3_key = format!("test/{}/download/build.zip", setup.item_id);
405 h.storage
406 .as_ref()
407 .unwrap()
408 .put(&s3_key, b"zip data".to_vec());
409 let version_id: String = sqlx::query_scalar(
410 "INSERT INTO versions (item_id, version_number, s3_key, file_size_bytes, file_name, \
411 is_current, scan_status) \
412 VALUES ($1::uuid, '1.0', $2, 100, 'build.zip', true, 'clean') RETURNING id::text",
413 )
414 .bind(&setup.item_id)
415 .bind(&s3_key)
416 .fetch_one(&h.db)
417 .await
418 .unwrap();
419
420 let resp = h
421 .client
422 .get(&format!("/api/versions/{version_id}/download"))
423 .await;
424 assert!(resp.status.is_success());
425 let data: Value = resp.json();
426 assert!(
427 data["license_url"].is_null(),
428 "license_url should be absent when no preset is set, got: {:?}",
429 data["license_url"]
430 );
431 }
432
433 #[tokio::test]
434 async fn stream_url_404_for_nonexistent_item() {
435 let mut h = TestHarness::with_storage().await;
436 let bogus = "00000000-0000-0000-0000-000000000000";
437 let resp = h.client.get(&format!("/api/stream/{bogus}")).await;
438 assert_eq!(resp.status.as_u16(), 404);
439 }
440
441 #[tokio::test]
442 async fn stream_url_expires_in_scales_with_duration() {
443 let mut h = TestHarness::with_storage().await;
444 let (_, _, item_id, _) = setup_audio_item(&mut h, 0).await;
445 // Set a long duration. The handler computes expiry as
446 // `max(duration * 2, 3600)`, so 5000s should give 10000s expiry.
447 sqlx::query("UPDATE items SET duration_seconds = 5000 WHERE id = $1::uuid")
448 .bind(&item_id)
449 .execute(&h.db)
450 .await
451 .unwrap();
452
453 let resp = h.client.get(&format!("/api/stream/{item_id}")).await;
454 assert!(resp.status.is_success(), "{} {}", resp.status, resp.text);
455 let data: Value = resp.json();
456 let expires_in = data["expires_in"].as_u64().unwrap();
457 assert_eq!(
458 expires_in, 10000,
459 "expires_in should be 2x duration for long tracks (got {expires_in})"
460 );
461 }
462