Skip to main content

max / makenotwork

15.5 KB · 460 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_eq!(
138 resp.status, 200,
139 "subscriber within the paid period must pass the gate, got {} {}",
140 resp.status, resp.text
141 );
142 }
143
144 #[tokio::test]
145 async fn stream_url_response_has_expected_shape() {
146 let mut h = TestHarness::with_storage().await;
147 let (_, _, item_id, _) = setup_audio_item(&mut h, 0).await;
148
149 let resp = h.client.get(&format!("/api/stream/{item_id}")).await;
150 assert_eq!(resp.status, 200, "{} {}", resp.status, resp.text);
151
152 let data: Value = resp.json();
153 assert!(
154 data["stream_url"].is_string(),
155 "Response must contain stream_url"
156 );
157 assert!(
158 data["expires_in"].is_u64(),
159 "Response must contain numeric expires_in"
160 );
161 // Test storage backend returns http://test-storage/<key>.
162 assert!(
163 data["stream_url"]
164 .as_str()
165 .unwrap()
166 .contains("audio/track.mp3"),
167 "URL should reference the seeded key"
168 );
169 }
170
171 #[tokio::test]
172 async fn stream_url_404_when_item_has_no_audio_key() {
173 let mut h = TestHarness::with_storage().await;
174 let setup = h.create_creator_with_item("noaudio", "audio", 0).await;
175 // Mark the item public + clean but DON'T set audio_s3_key, the
176 // handler should refuse to mint a streaming URL for a "naked" item.
177 sqlx::query("UPDATE items SET is_public = true, scan_status = 'clean' WHERE id = $1::uuid")
178 .bind(&setup.item_id)
179 .execute(&h.db)
180 .await
181 .unwrap();
182 sqlx::query("UPDATE projects SET is_public = true WHERE id = $1::uuid")
183 .bind(&setup.project_id)
184 .execute(&h.db)
185 .await
186 .unwrap();
187
188 let resp = h
189 .client
190 .get(&format!("/api/stream/{}", setup.item_id))
191 .await;
192 assert_eq!(
193 resp.status.as_u16(),
194 404,
195 "Item without audio_s3_key must 404"
196 );
197 }
198
199 #[tokio::test]
200 async fn stream_url_404_for_quarantined_item_to_non_creator() {
201 let mut h = TestHarness::with_storage().await;
202 let (_, _, item_id, _) = setup_audio_item(&mut h, 0).await;
203 sqlx::query("UPDATE items SET scan_status = 'quarantined' WHERE id = $1::uuid")
204 .bind(&item_id)
205 .execute(&h.db)
206 .await
207 .unwrap();
208 // Log out, the creator is currently authenticated.
209 h.client.post_form("/logout", "").await;
210
211 let resp = h.client.get(&format!("/api/stream/{item_id}")).await;
212 assert_eq!(
213 resp.status.as_u16(),
214 404,
215 "Quarantined items must not stream to non-creators"
216 );
217 }
218
219 #[tokio::test]
220 async fn stream_url_404_for_pending_scan_to_non_creator() {
221 let mut h = TestHarness::with_storage().await;
222 let (_, _, item_id, _) = setup_audio_item(&mut h, 0).await;
223 sqlx::query("UPDATE items SET scan_status = 'pending' WHERE id = $1::uuid")
224 .bind(&item_id)
225 .execute(&h.db)
226 .await
227 .unwrap();
228 h.client.post_form("/logout", "").await;
229
230 let resp = h.client.get(&format!("/api/stream/{item_id}")).await;
231 assert_eq!(
232 resp.status.as_u16(),
233 404,
234 "Pending-scan items must not stream to non-creators (fail-closed)"
235 );
236 }
237
238 #[tokio::test]
239 async fn stream_url_creator_can_preview_held_for_review_item() {
240 let mut h = TestHarness::with_storage().await;
241 let (_creator, _, item_id, _) = setup_audio_item(&mut h, 0).await;
242 sqlx::query("UPDATE items SET scan_status = 'held_for_review' WHERE id = $1::uuid")
243 .bind(&item_id)
244 .execute(&h.db)
245 .await
246 .unwrap();
247 // Creator remains logged in from setup.
248
249 let resp = h.client.get(&format!("/api/stream/{item_id}")).await;
250 assert_eq!(
251 resp.status, 200,
252 "Creator must be able to preview their own HeldForReview content: {} {}",
253 resp.status, resp.text
254 );
255 }
256
257 #[tokio::test]
258 async fn stream_url_authenticated_non_buyer_gets_403_on_paid_item() {
259 let mut h = TestHarness::with_storage().await;
260 let (_creator, _, item_id, _) = setup_audio_item(&mut h, 999).await;
261 // Log out creator; create a different user with no purchase.
262 h.client.post_form("/logout", "").await;
263 h.signup("randomuser", "random@test.com", "password123")
264 .await;
265 h.login("randomuser", "password123").await;
266
267 let resp = h.client.get(&format!("/api/stream/{item_id}")).await;
268 assert_eq!(
269 resp.status.as_u16(),
270 403,
271 "Authenticated non-buyer on paid item must get 403, not 401: {} {}",
272 resp.status,
273 resp.text
274 );
275 }
276
277 #[tokio::test]
278 async fn stream_url_increments_play_count() {
279 let mut h = TestHarness::with_storage().await;
280 let (_, _, item_id, _) = setup_audio_item(&mut h, 0).await;
281
282 // Stream twice (free item, anyone can hit it).
283 h.client.post_form("/logout", "").await;
284 for _ in 0..2 {
285 let resp = h.client.get(&format!("/api/stream/{item_id}")).await;
286 assert_eq!(resp.status, 200, "{}", resp.text);
287 }
288
289 let count: i32 = sqlx::query_scalar("SELECT play_count FROM items WHERE id = $1::uuid")
290 .bind(&item_id)
291 .fetch_one(&h.db)
292 .await
293 .unwrap();
294 assert_eq!(count, 2, "play_count should increment per stream call");
295 }
296
297 #[tokio::test]
298 async fn stream_url_records_unique_play_for_authenticated_user() {
299 let mut h = TestHarness::with_storage().await;
300 let (_creator, _, item_id, _) = setup_audio_item(&mut h, 0).await;
301
302 // Switch to a different authenticated user.
303 h.client.post_form("/logout", "").await;
304 let listener_id = h
305 .signup("listener", "listener@test.com", "password123")
306 .await;
307 h.login("listener", "password123").await;
308
309 let resp = h.client.get(&format!("/api/stream/{item_id}")).await;
310 assert_eq!(resp.status, 200, "{}", resp.text);
311
312 let has_play: bool = sqlx::query_scalar(
313 "SELECT EXISTS(SELECT 1 FROM user_plays WHERE user_id = $1 AND item_id = $2::uuid)",
314 )
315 .bind(listener_id)
316 .bind(&item_id)
317 .fetch_one(&h.db)
318 .await
319 .unwrap();
320 assert!(
321 has_play,
322 "user_plays should record the listener for unique-listener tracking"
323 );
324 }
325
326 #[tokio::test]
327 async fn version_download_includes_license_url_when_preset_set() {
328 let mut h = TestHarness::with_storage().await;
329 let setup = h
330 .create_creator_with_item("licdownloader", "digital", 0)
331 .await;
332 sqlx::query(
333 "UPDATE items SET is_public = true, scan_status = 'clean', \
334 license_preset = 'mit' WHERE id = $1::uuid",
335 )
336 .bind(&setup.item_id)
337 .execute(&h.db)
338 .await
339 .unwrap();
340 sqlx::query("UPDATE projects SET is_public = true WHERE id = $1::uuid")
341 .bind(&setup.project_id)
342 .execute(&h.db)
343 .await
344 .unwrap();
345
346 // Create a version with a fake s3_key.
347 let s3_key = format!("test/{}/download/build.zip", setup.item_id);
348 h.storage
349 .as_ref()
350 .unwrap()
351 .put(&s3_key, b"zip data".to_vec());
352 let version_id: String = sqlx::query_scalar(
353 "INSERT INTO versions (item_id, version_number, s3_key, file_size_bytes, file_name, \
354 is_current, scan_status) \
355 VALUES ($1::uuid, '1.0', $2, 100, 'build.zip', true, 'clean') RETURNING id::text",
356 )
357 .bind(&setup.item_id)
358 .bind(&s3_key)
359 .fetch_one(&h.db)
360 .await
361 .unwrap();
362
363 let resp = h
364 .client
365 .get(&format!("/api/versions/{version_id}/download"))
366 .await;
367 assert_eq!(resp.status, 200, "{} {}", resp.status, resp.text);
368 let data: Value = resp.json();
369 assert!(
370 data["license_url"].is_string(),
371 "license_url should be present when item.license_preset is set"
372 );
373 let license_url = data["license_url"].as_str().unwrap();
374 assert!(
375 license_url.contains(&setup.item_id),
376 "license_url should point at the item: {license_url}"
377 );
378 assert!(
379 license_url.ends_with("/license.txt"),
380 "license_url should target the .txt endpoint: {license_url}"
381 );
382 }
383
384 #[tokio::test]
385 async fn version_download_omits_license_url_without_preset() {
386 let mut h = TestHarness::with_storage().await;
387 let setup = h.create_creator_with_item("nolicdl", "digital", 0).await;
388 sqlx::query(
389 "UPDATE items SET is_public = true, scan_status = 'clean', \
390 license_preset = NULL WHERE id = $1::uuid",
391 )
392 .bind(&setup.item_id)
393 .execute(&h.db)
394 .await
395 .unwrap();
396 sqlx::query("UPDATE projects SET is_public = true WHERE id = $1::uuid")
397 .bind(&setup.project_id)
398 .execute(&h.db)
399 .await
400 .unwrap();
401
402 let s3_key = format!("test/{}/download/build.zip", setup.item_id);
403 h.storage
404 .as_ref()
405 .unwrap()
406 .put(&s3_key, b"zip data".to_vec());
407 let version_id: String = sqlx::query_scalar(
408 "INSERT INTO versions (item_id, version_number, s3_key, file_size_bytes, file_name, \
409 is_current, scan_status) \
410 VALUES ($1::uuid, '1.0', $2, 100, 'build.zip', true, 'clean') RETURNING id::text",
411 )
412 .bind(&setup.item_id)
413 .bind(&s3_key)
414 .fetch_one(&h.db)
415 .await
416 .unwrap();
417
418 let resp = h
419 .client
420 .get(&format!("/api/versions/{version_id}/download"))
421 .await;
422 assert_eq!(resp.status, 200, "{}", resp.text);
423 let data: Value = resp.json();
424 assert!(
425 data["license_url"].is_null(),
426 "license_url should be absent when no preset is set, got: {:?}",
427 data["license_url"]
428 );
429 }
430
431 #[tokio::test]
432 async fn stream_url_404_for_nonexistent_item() {
433 let mut h = TestHarness::with_storage().await;
434 let bogus = "00000000-0000-0000-0000-000000000000";
435 let resp = h.client.get(&format!("/api/stream/{bogus}")).await;
436 assert_eq!(resp.status.as_u16(), 404);
437 }
438
439 #[tokio::test]
440 async fn stream_url_expires_in_scales_with_duration() {
441 let mut h = TestHarness::with_storage().await;
442 let (_, _, item_id, _) = setup_audio_item(&mut h, 0).await;
443 // Set a long duration. The handler computes expiry as
444 // `max(duration * 2, 3600)`, so 5000s should give 10000s expiry.
445 sqlx::query("UPDATE items SET duration_seconds = 5000 WHERE id = $1::uuid")
446 .bind(&item_id)
447 .execute(&h.db)
448 .await
449 .unwrap();
450
451 let resp = h.client.get(&format!("/api/stream/{item_id}")).await;
452 assert_eq!(resp.status, 200, "{} {}", resp.status, resp.text);
453 let data: Value = resp.json();
454 let expires_in = data["expires_in"].as_u64().unwrap();
455 assert_eq!(
456 expires_in, 10000,
457 "expires_in should be 2x duration for long tracks (got {expires_in})"
458 );
459 }
460