Skip to main content

max / makenotwork

80.0 KB · 2419 lines History Blame Raw
1 //! Storage workflow tests, presign, confirm, stream, download, access control.
2
3 use crate::harness::TestHarness;
4 use makenotwork::storage::StorageBackend;
5 use serde_json::{Value, json};
6
7 /// Helper: create a trusted creator with a project and audio item. Returns (user_id, project_id, item_id).
8 async fn setup_creator_with_item(
9 h: &mut TestHarness,
10 price_cents: i64,
11 ) -> (String, String, String) {
12 let setup = h
13 .create_creator_with_item("creator", "audio", price_cents)
14 .await;
15 h.trust_user(setup.user_id).await;
16 h.grant_tier(setup.user_id, "small_files").await;
17 (setup.user_id.to_string(), setup.project_id, setup.item_id)
18 }
19
20 // Presign
21
22 #[tokio::test]
23 async fn presign_upload_audio() {
24 let mut h = TestHarness::with_storage().await;
25 let (_, _, item_id) = setup_creator_with_item(&mut h, 0).await;
26
27 let body = json!({
28 "item_id": item_id,
29 "file_type": "audio",
30 "file_name": "episode.mp3",
31 "content_type": "audio/mpeg",
32 });
33 let resp = h
34 .client
35 .post_json("/api/upload/presign", &body.to_string())
36 .await;
37 assert!(resp.status.is_success(), "Presign failed: {}", resp.text);
38
39 let data: Value = resp.json();
40 assert!(
41 data["upload_url"]
42 .as_str()
43 .unwrap()
44 .starts_with("http://test-storage/")
45 );
46 // Scan-then-promote: presign hands out an unserved staging key
47 // (`staging/{uuid}/{filename}`), never the served key. The served,
48 // content-addressed key is minted by the scan worker on a Clean verdict.
49 let key = data["s3_key"].as_str().unwrap();
50 assert!(
51 key.starts_with("staging/"),
52 "presign must return a staging key: {key}"
53 );
54 assert!(
55 key.ends_with("/episode.mp3"),
56 "staging key preserves the filename: {key}"
57 );
58 assert_eq!(data["expires_in"], 3600);
59 }
60
61 // Confirm
62
63 #[tokio::test]
64 async fn confirm_upload_audio_updates_db() {
65 let mut h = TestHarness::with_storage().await;
66 let (_, _, item_id) = setup_creator_with_item(&mut h, 0).await;
67
68 // Presign
69 let body = json!({
70 "item_id": item_id,
71 "file_type": "audio",
72 "file_name": "song.mp3",
73 "content_type": "audio/mpeg",
74 });
75 let resp = h
76 .client
77 .post_json("/api/upload/presign", &body.to_string())
78 .await;
79 assert!(resp.status.is_success());
80 let data: Value = resp.json();
81 let s3_key = data["s3_key"].as_str().unwrap().to_string();
82
83 // Simulate the client uploading to S3
84 h.storage
85 .as_ref()
86 .unwrap()
87 .put(&s3_key, b"fake mp3 bytes".to_vec());
88
89 // Confirm
90 let body = json!({
91 "item_id": item_id,
92 "file_type": "audio",
93 "s3_key": s3_key,
94 });
95 let resp = h
96 .client
97 .post_json("/api/upload/confirm", &body.to_string())
98 .await;
99 assert!(resp.status.is_success(), "Confirm failed: {}", resp.text);
100 let data: Value = resp.json();
101 assert_eq!(data["success"], true);
102
103 // Verify database
104 let db_key: Option<String> =
105 sqlx::query_scalar("SELECT audio_s3_key FROM items WHERE id = $1::uuid")
106 .bind(&item_id)
107 .fetch_one(&h.db)
108 .await
109 .unwrap();
110 assert_eq!(db_key.as_deref(), Some(s3_key.as_str()));
111 }
112
113 #[tokio::test]
114 async fn internal_confirm_upload_replay_is_idempotent() {
115 // A retried internal (CLI, ServiceAuth) confirm for the same deterministic key
116 // must not double-charge storage (ultra-fuzz Run 12 Storage: confirm idempotency).
117 let mem = std::sync::Arc::new(crate::harness::storage::InMemoryStorage::new());
118 let mut h = TestHarness::build(crate::harness::BuildOptions {
119 storage: Some(mem),
120 cli_service_token: Some("test-cli-token".to_string()),
121 ..Default::default()
122 })
123 .await;
124
125 let setup = h.create_creator_with_item("cliuser", "audio", 0).await;
126 h.trust_user(setup.user_id).await;
127 h.grant_tier(setup.user_id, "small_files").await;
128 let user_id = setup.user_id;
129 let item_id = setup.item_id.clone();
130
131 // Presign (session auth) then simulate the client PUT to S3.
132 let presign = json!({
133 "item_id": item_id, "file_type": "audio",
134 "file_name": "song.mp3", "content_type": "audio/mpeg",
135 });
136 let resp = h
137 .client
138 .post_json("/api/upload/presign", &presign.to_string())
139 .await;
140 assert!(resp.status.is_success(), "presign failed: {}", resp.text);
141 let s3_key = resp.json::<Value>()["s3_key"].as_str().unwrap().to_string();
142 let bytes = b"fake mp3 file bytes".to_vec();
143 let expected_size = bytes.len() as i64;
144 h.storage.as_ref().unwrap().put(&s3_key, bytes);
145
146 // Internal confirm (ServiceAuth via bearer), twice, identical.
147 let confirm = json!({
148 "user_id": user_id, "item_id": item_id,
149 "file_type": "audio", "s3_key": s3_key,
150 });
151 h.client.set_bearer_token("test-cli-token");
152 let actor = makenotwork::crypto::mint_internal_actor_token(
153 user_id,
154 chrono::Utc::now().timestamp() + 3600,
155 "test-signing-secret-for-integration-tests",
156 );
157 h.client.set_actor_token(&actor);
158 let r1 = h
159 .client
160 .post_json("/api/internal/upload/confirm", &confirm.to_string())
161 .await;
162 assert!(
163 r1.status.is_success(),
164 "first internal confirm failed: {}",
165 r1.text
166 );
167 let r2 = h
168 .client
169 .post_json("/api/internal/upload/confirm", &confirm.to_string())
170 .await;
171 assert!(
172 r2.status.is_success(),
173 "replayed internal confirm failed: {}",
174 r2.text
175 );
176 h.client.clear_bearer_token();
177
178 // The key is committed once and storage is charged exactly once, not twice.
179 let db_key: Option<String> =
180 sqlx::query_scalar("SELECT audio_s3_key FROM items WHERE id = $1::uuid")
181 .bind(&item_id)
182 .fetch_one(&h.db)
183 .await
184 .unwrap();
185 assert_eq!(db_key.as_deref(), Some(s3_key.as_str()));
186
187 let storage_used: i64 =
188 sqlx::query_scalar("SELECT storage_used_bytes FROM users WHERE id = $1")
189 .bind(user_id)
190 .fetch_one(&h.db)
191 .await
192 .unwrap();
193 assert_eq!(
194 storage_used, expected_size,
195 "replay must not double-charge storage"
196 );
197 }
198
199 /// A valid ServiceAuth bearer alone is not enough to act on the internal API:
200 /// without a valid `X-MNW-Actor` assertion the request is rejected. Proves a
201 /// leaked service token cannot name an arbitrary user (audit 2026-07-01).
202 #[tokio::test]
203 async fn internal_confirm_without_actor_token_rejected() {
204 let mem = std::sync::Arc::new(crate::harness::storage::InMemoryStorage::new());
205 let mut h = TestHarness::build(crate::harness::BuildOptions {
206 storage: Some(mem),
207 cli_service_token: Some("test-cli-token".to_string()),
208 ..Default::default()
209 })
210 .await;
211
212 let setup = h.create_creator_with_item("noactor", "audio", 0).await;
213 h.trust_user(setup.user_id).await;
214 h.grant_tier(setup.user_id, "small_files").await;
215
216 let confirm = json!({
217 "user_id": setup.user_id, "item_id": setup.item_id,
218 "file_type": "audio", "s3_key": "noactor/whatever.mp3",
219 });
220 // Bearer set, but no actor assertion.
221 h.client.set_bearer_token("test-cli-token");
222 let resp = h
223 .client
224 .post_json("/api/internal/upload/confirm", &confirm.to_string())
225 .await;
226 assert_eq!(
227 resp.status.as_u16(),
228 401,
229 "missing actor assertion must be rejected: {}",
230 resp.text
231 );
232 h.client.clear_bearer_token();
233 }
234
235 #[tokio::test]
236 async fn confirm_item_cover_via_dedicated_route_writes_key_and_url() {
237 // Covers go through /api/items/image/{presign,confirm}, which writes
238 // cover_s3_key, cover_file_size_bytes AND cover_image_url together. The
239 // generic /api/upload/confirm used to accept cover and write the first two
240 // but NOT the URL, leaving an invisible cover (Run #13 SERIOUS); it now
241 // rejects cover (see confirm_upload_rejects_cover below).
242 let mut h = TestHarness::with_storage().await;
243 let (_, _, item_id) = setup_creator_with_item(&mut h, 0).await;
244
245 let body = json!({
246 "item_id": item_id,
247 "file_name": "art.png",
248 "content_type": "image/png",
249 });
250 let resp = h
251 .client
252 .post_json("/api/items/image/presign", &body.to_string())
253 .await;
254 assert!(resp.status.is_success(), "presign failed: {}", resp.text);
255 let data: Value = resp.json();
256 let s3_key = data["s3_key"].as_str().unwrap().to_string();
257
258 h.storage
259 .as_ref()
260 .unwrap()
261 .put(&s3_key, b"fake png bytes".to_vec());
262
263 let body = json!({ "item_id": item_id, "s3_key": s3_key });
264 let resp = h
265 .client
266 .post_json("/api/items/image/confirm", &body.to_string())
267 .await;
268 assert!(resp.status.is_success(), "Confirm failed: {}", resp.text);
269
270 // Both the key AND the render URL must be set, the URL is what the bug missed.
271 let (db_key, db_url): (Option<String>, Option<String>) =
272 sqlx::query_as("SELECT cover_s3_key, cover_image_url FROM items WHERE id = $1::uuid")
273 .bind(&item_id)
274 .fetch_one(&h.db)
275 .await
276 .unwrap();
277 assert_eq!(db_key.as_deref(), Some(s3_key.as_str()));
278 assert!(
279 db_url.is_some_and(|u| u.contains(&s3_key)),
280 "cover_image_url must be set so the cover renders"
281 );
282 }
283
284 #[tokio::test]
285 async fn cover_replace_does_not_take_published_track_offline() {
286 // Run #20 Storage SERIOUS: a cover upload used to flip the SHARED
287 // `items.scan_status` to Pending (cover shares the audio's row), and the
288 // stream/download gate returns NotFound for non-creators until the cover
289 // re-scan finishes, silently pulling an already-published track offline
290 // for every fan. The cover is CDN-served with no per-request gate, so the
291 // flip protects nothing and only harms the track. Confirming a cover must
292 // leave `items.scan_status` untouched.
293 let mut h = TestHarness::with_storage_and_scanner().await;
294 let (_, _, item_id) = setup_creator_with_item(&mut h, 0).await;
295
296 // Simulate the published state: track already scanned Clean.
297 sqlx::query("UPDATE items SET scan_status = 'clean' WHERE id = $1::uuid")
298 .bind(&item_id)
299 .execute(&h.db)
300 .await
301 .unwrap();
302
303 // Upload a new cover. With a scanner configured, the cover scan enqueues as
304 // Pending; pre-fix that Pending was written straight onto items.scan_status
305 // by the synchronous confirm.
306 let body = json!({ "item_id": item_id, "file_name": "art.png", "content_type": "image/png" });
307 let resp = h
308 .client
309 .post_json("/api/items/image/presign", &body.to_string())
310 .await;
311 assert!(resp.status.is_success(), "presign failed: {}", resp.text);
312 let s3_key = resp.json::<Value>()["s3_key"].as_str().unwrap().to_string();
313 h.storage
314 .as_ref()
315 .unwrap()
316 .put(&s3_key, b"fake png bytes".to_vec());
317
318 let body = json!({ "item_id": item_id, "s3_key": s3_key });
319 let resp = h
320 .client
321 .post_json("/api/items/image/confirm", &body.to_string())
322 .await;
323 assert!(
324 resp.status.is_success(),
325 "cover confirm failed: {}",
326 resp.text
327 );
328
329 // The track's gate status must still be Clean, the cover upload may not
330 // touch it. (Assert synchronously, before draining the scan worker.)
331 let scan_status: String =
332 sqlx::query_scalar("SELECT scan_status FROM items WHERE id = $1::uuid")
333 .bind(&item_id)
334 .fetch_one(&h.db)
335 .await
336 .unwrap();
337 assert_eq!(
338 scan_status, "clean",
339 "cover upload must not flip the track's scan_status (would take it offline for fans)"
340 );
341
342 // And the cover itself still landed.
343 let cover_key: Option<String> =
344 sqlx::query_scalar("SELECT cover_s3_key FROM items WHERE id = $1::uuid")
345 .bind(&item_id)
346 .fetch_one(&h.db)
347 .await
348 .unwrap();
349 assert_eq!(cover_key.as_deref(), Some(s3_key.as_str()));
350 }
351
352 #[tokio::test]
353 async fn confirm_upload_rejects_cover() {
354 // The generic confirm route must refuse cover and point at the dedicated
355 // route, rather than half-writing the row (no cover_image_url). Run #13.
356 let mut h = TestHarness::with_storage().await;
357 let (_, _, item_id) = setup_creator_with_item(&mut h, 0).await;
358
359 let body = json!({
360 "item_id": item_id,
361 "file_type": "cover",
362 "file_name": "art.png",
363 "content_type": "image/png",
364 });
365 let resp = h
366 .client
367 .post_json("/api/upload/presign", &body.to_string())
368 .await;
369 assert!(resp.status.is_success());
370 let data: Value = resp.json();
371 let s3_key = data["s3_key"].as_str().unwrap().to_string();
372 h.storage
373 .as_ref()
374 .unwrap()
375 .put(&s3_key, b"fake png bytes".to_vec());
376
377 let body = json!({ "item_id": item_id, "file_type": "cover", "s3_key": s3_key });
378 let resp = h
379 .client
380 .post_json("/api/upload/confirm", &body.to_string())
381 .await;
382 assert_eq!(
383 resp.status.as_u16(),
384 400,
385 "generic confirm must reject cover: {}",
386 resp.text
387 );
388 assert!(
389 resp.text.contains("/api/items/image/confirm"),
390 "rejection should name the dedicated route: {}",
391 resp.text
392 );
393
394 // The row must be untouched, no half-written cover key.
395 let db_key: Option<String> =
396 sqlx::query_scalar("SELECT cover_s3_key FROM items WHERE id = $1::uuid")
397 .bind(&item_id)
398 .fetch_one(&h.db)
399 .await
400 .unwrap();
401 assert_eq!(
402 db_key, None,
403 "rejected cover confirm must not write cover_s3_key"
404 );
405 }
406
407 // Versions
408
409 #[tokio::test]
410 async fn version_upload_and_download() {
411 let mut h = TestHarness::with_storage().await;
412 let (_, _, item_id) = setup_creator_with_item(&mut h, 0).await;
413
414 // Create a version (digital item needs a version for downloads)
415 let resp = h
416 .client
417 .post_json(
418 &format!("/api/items/{item_id}/versions"),
419 &json!({"version_number": "1.0.0"}).to_string(),
420 )
421 .await;
422 assert!(
423 resp.status.is_success(),
424 "Create version failed: {}",
425 resp.text
426 );
427 let version: Value = resp.json();
428 let version_id = version["id"].as_str().unwrap().to_string();
429
430 // Presign version upload
431 let resp = h
432 .client
433 .post_json(
434 &format!("/api/versions/{version_id}/upload/presign"),
435 &json!({
436 "file_name": "plugin.zip",
437 "content_type": "application/zip",
438 })
439 .to_string(),
440 )
441 .await;
442 assert!(
443 resp.status.is_success(),
444 "Version presign failed: {}",
445 resp.text
446 );
447 let data: Value = resp.json();
448 let s3_key = data["s3_key"].as_str().unwrap().to_string();
449
450 // Simulate upload
451 h.storage
452 .as_ref()
453 .unwrap()
454 .put(&s3_key, b"fake zip data".to_vec());
455
456 // Confirm version upload
457 let resp = h
458 .client
459 .post_json(
460 &format!("/api/versions/{version_id}/upload/confirm"),
461 &json!({"s3_key": s3_key}).to_string(),
462 )
463 .await;
464 assert!(
465 resp.status.is_success(),
466 "Version confirm failed: {}",
467 resp.text
468 );
469
470 // Publish item + project so download works
471 h.client
472 .put_form(&format!("/api/items/{item_id}"), "is_public=true")
473 .await;
474 let project_id: String =
475 sqlx::query_scalar("SELECT project_id::text FROM items WHERE id = $1::uuid")
476 .bind(&item_id)
477 .fetch_one(&h.db)
478 .await
479 .unwrap();
480 h.client
481 .put_json(
482 &format!("/api/projects/{project_id}"),
483 r#"{"is_public": true}"#,
484 )
485 .await;
486
487 // Download version
488 let resp = h
489 .client
490 .get(&format!("/api/versions/{version_id}/download"))
491 .await;
492 assert!(
493 resp.status.is_success(),
494 "Version download failed: {}",
495 resp.text
496 );
497 let data: Value = resp.json();
498 assert!(
499 data["download_url"]
500 .as_str()
501 .unwrap()
502 .starts_with("http://test-storage/")
503 );
504 }
505
506 // Audio Streaming
507
508 #[tokio::test]
509 async fn stream_url_free_item() {
510 let mut h = TestHarness::with_storage().await;
511 let (_, project_id, item_id) = setup_creator_with_item(&mut h, 0).await;
512
513 // Set up audio key directly in DB (simulates a completed upload)
514 let s3_key = format!("test/{item_id}/audio/track.mp3");
515 sqlx::query("UPDATE items SET audio_s3_key = $1, scan_status = 'clean' WHERE id = $2::uuid")
516 .bind(&s3_key)
517 .bind(&item_id)
518 .execute(&h.db)
519 .await
520 .unwrap();
521
522 // Pre-populate storage
523 h.storage
524 .as_ref()
525 .unwrap()
526 .put(&s3_key, b"audio data".to_vec());
527
528 // Publish
529 h.client
530 .put_form(&format!("/api/items/{item_id}"), "is_public=true")
531 .await;
532 h.client
533 .put_json(
534 &format!("/api/projects/{project_id}"),
535 r#"{"is_public": true}"#,
536 )
537 .await;
538
539 // Stream, free item, any user can access
540 let resp = h.client.get(&format!("/api/stream/{item_id}")).await;
541 assert!(resp.status.is_success(), "Stream failed: {}", resp.text);
542 let data: Value = resp.json();
543 assert!(
544 data["stream_url"]
545 .as_str()
546 .unwrap()
547 .starts_with("http://test-storage/")
548 );
549 }
550
551 #[tokio::test]
552 async fn stream_url_paid_requires_purchase() {
553 let mut h = TestHarness::with_storage().await;
554 let (_, project_id, item_id) = setup_creator_with_item(&mut h, 500).await;
555
556 // Set up audio key
557 let s3_key = format!("test/{item_id}/audio/track.mp3");
558 sqlx::query("UPDATE items SET audio_s3_key = $1, scan_status = 'clean' WHERE id = $2::uuid")
559 .bind(&s3_key)
560 .bind(&item_id)
561 .execute(&h.db)
562 .await
563 .unwrap();
564
565 h.storage
566 .as_ref()
567 .unwrap()
568 .put(&s3_key, b"audio data".to_vec());
569
570 // Publish
571 h.client
572 .put_form(&format!("/api/items/{item_id}"), "is_public=true")
573 .await;
574 h.client
575 .put_json(
576 &format!("/api/projects/{project_id}"),
577 r#"{"is_public": true}"#,
578 )
579 .await;
580
581 // Log out the creator and sign up a buyer with no purchase
582 h.client.post_form("/logout", "").await;
583 h.signup("buyer", "buyer@test.com", "password123").await;
584 h.login("buyer", "password123").await;
585
586 // Stream should be forbidden (paid, no purchase)
587 let resp = h.client.get(&format!("/api/stream/{item_id}")).await;
588 assert_eq!(
589 resp.status.as_u16(),
590 403,
591 "Expected 403, got: {}",
592 resp.text
593 );
594 }
595
596 /// Helper: set up a paid item with audio, publish it, return (creator_user_id, project_id, item_id, s3_key).
597 /// Leaves the creator logged in.
598 async fn setup_published_paid_audio(
599 h: &mut TestHarness,
600 username: &str,
601 price_cents: i64,
602 ) -> (String, String, String, String) {
603 let setup = h
604 .create_creator_with_item(username, "audio", price_cents)
605 .await;
606 h.trust_user(setup.user_id).await;
607 h.grant_tier(setup.user_id, "small_files").await;
608 let user_id = setup.user_id.to_string();
609
610 let s3_key = format!("test/{}/audio/track.mp3", setup.item_id);
611 sqlx::query("UPDATE items SET audio_s3_key = $1, scan_status = 'clean' WHERE id = $2::uuid")
612 .bind(&s3_key)
613 .bind(&setup.item_id)
614 .execute(&h.db)
615 .await
616 .unwrap();
617 h.storage
618 .as_ref()
619 .unwrap()
620 .put(&s3_key, b"audio data".to_vec());
621
622 h.publish_project_and_item(&setup.project_id, &setup.item_id)
623 .await;
624
625 (user_id, setup.project_id, setup.item_id, s3_key)
626 }
627
628 // Stream access control (test-fuzz)
629
630 /// Unauthenticated user gets 401 on paid item stream.
631 #[tokio::test]
632 async fn stream_url_paid_unauthenticated_returns_401() {
633 let mut h = TestHarness::with_storage().await;
634 let (_, _, item_id, _) = setup_published_paid_audio(&mut h, "seller401", 500).await;
635
636 // Log out, no session
637 h.client.post_form("/logout", "").await;
638
639 let resp = h.client.get(&format!("/api/stream/{item_id}")).await;
640 assert_eq!(
641 resp.status.as_u16(),
642 401,
643 "Unauthenticated stream of paid item should be 401, got: {}",
644 resp.status
645 );
646 }
647
648 /// After a direct DB purchase record, buyer can stream paid content.
649 #[tokio::test]
650 async fn stream_url_paid_purchaser_gets_access() {
651 let mut h = TestHarness::with_storage().await;
652 let (creator_id, _, item_id, _) = setup_published_paid_audio(&mut h, "sellaccess", 999).await;
653
654 // Create buyer and insert a completed transaction (simulates webhook completion)
655 h.client.post_form("/logout", "").await;
656 let buyer_id = h
657 .signup("buyaccess", "buyaccess@test.com", "password123")
658 .await;
659
660 sqlx::query(
661 r"INSERT INTO transactions
662 (buyer_id, seller_id, item_id, amount_cents, status,
663 stripe_checkout_session_id, item_title, seller_username,
664 completed_at)
665 VALUES ($1, $2::uuid, $3::uuid, 999, 'completed',
666 'cs_access_test', 'Track', 'sellaccess', NOW())",
667 )
668 .bind(buyer_id)
669 .bind(&creator_id)
670 .bind(&item_id)
671 .execute(&h.db)
672 .await
673 .unwrap();
674
675 h.login("buyaccess", "password123").await;
676
677 let resp = h.client.get(&format!("/api/stream/{item_id}")).await;
678 assert!(
679 resp.status.is_success(),
680 "Purchaser should be able to stream paid item, got: {} {}",
681 resp.status,
682 resp.text
683 );
684 let data: Value = resp.json();
685 assert!(
686 data["stream_url"].as_str().is_some(),
687 "Response should contain stream_url"
688 );
689 }
690
691 /// Creator can always stream their own paid content.
692 #[tokio::test]
693 async fn stream_url_creator_always_has_access() {
694 let mut h = TestHarness::with_storage().await;
695 let (_, _, item_id, _) = setup_published_paid_audio(&mut h, "selfstream", 999).await;
696
697 // Creator is still logged in
698 let resp = h.client.get(&format!("/api/stream/{item_id}")).await;
699 assert!(
700 resp.status.is_success(),
701 "Creator should stream their own paid item, got: {} {}",
702 resp.status,
703 resp.text
704 );
705 }
706
707 /// Unpublished (draft) item returns 404 for non-owner.
708 #[tokio::test]
709 async fn stream_url_draft_item_404_for_non_owner() {
710 let mut h = TestHarness::with_storage().await;
711 let setup = h.create_creator_with_item("draftowner", "audio", 0).await;
712 h.trust_user(setup.user_id).await;
713 h.grant_tier(setup.user_id, "small_files").await;
714
715 let s3_key = format!("test/{}/audio/draft.mp3", setup.item_id);
716 sqlx::query("UPDATE items SET audio_s3_key = $1, scan_status = 'clean' WHERE id = $2::uuid")
717 .bind(&s3_key)
718 .bind(&setup.item_id)
719 .execute(&h.db)
720 .await
721 .unwrap();
722 h.storage
723 .as_ref()
724 .unwrap()
725 .put(&s3_key, b"audio data".to_vec());
726
727 // Explicitly unpublish the item (items default to is_public=true)
728 h.client
729 .put_form(&format!("/api/items/{}", setup.item_id), "is_public=false")
730 .await;
731
732 h.client.post_form("/logout", "").await;
733 h.signup("snooper", "snooper@test.com", "password123").await;
734 h.login("snooper", "password123").await;
735
736 let resp = h
737 .client
738 .get(&format!("/api/stream/{}", setup.item_id))
739 .await;
740 assert_eq!(
741 resp.status.as_u16(),
742 404,
743 "Draft item should be 404 for non-owner, got: {}",
744 resp.status
745 );
746 }
747
748 /// Version download for paid item: non-purchaser gets 403.
749 #[tokio::test]
750 async fn version_download_paid_non_purchaser_forbidden() {
751 let mut h = TestHarness::with_storage().await;
752 let setup = h
753 .create_creator_with_item("verseller", "digital", 500)
754 .await;
755 h.trust_user(setup.user_id).await;
756 h.grant_tier(setup.user_id, "small_files").await;
757
758 // Create version with file
759 let resp = h
760 .client
761 .post_json(
762 &format!("/api/items/{}/versions", setup.item_id),
763 &json!({"version_number": "1.0.0"}).to_string(),
764 )
765 .await;
766 assert!(
767 resp.status.is_success(),
768 "Create version failed: {}",
769 resp.text
770 );
771 let version: Value = resp.json();
772 let version_id = version["id"].as_str().unwrap().to_string();
773
774 let resp = h
775 .client
776 .post_json(
777 &format!("/api/versions/{version_id}/upload/presign"),
778 &json!({"file_name": "app.zip", "content_type": "application/zip"}).to_string(),
779 )
780 .await;
781 assert!(resp.status.is_success());
782 let data: Value = resp.json();
783 let s3_key = data["s3_key"].as_str().unwrap().to_string();
784 h.storage
785 .as_ref()
786 .unwrap()
787 .put(&s3_key, b"zip data".to_vec());
788
789 h.client
790 .post_json(
791 &format!("/api/versions/{version_id}/upload/confirm"),
792 &json!({"s3_key": s3_key}).to_string(),
793 )
794 .await;
795
796 // Publish
797 h.publish_project_and_item(&setup.project_id, &setup.item_id)
798 .await;
799
800 // Non-purchaser tries to download
801 h.client.post_form("/logout", "").await;
802 h.signup("verbuyer", "verbuyer@test.com", "password123")
803 .await;
804 h.login("verbuyer", "password123").await;
805
806 let resp = h
807 .client
808 .get(&format!("/api/versions/{version_id}/download"))
809 .await;
810 assert_eq!(
811 resp.status.as_u16(),
812 403,
813 "Non-purchaser version download should be 403, got: {}",
814 resp.status
815 );
816 }
817
818 // Access control
819
820 #[tokio::test]
821 async fn upload_non_owner_forbidden() {
822 let mut h = TestHarness::with_storage().await;
823 let (_, _, item_id) = setup_creator_with_item(&mut h, 0).await;
824
825 // Log out creator, sign up a different user
826 h.client.post_form("/logout", "").await;
827 h.signup("intruder", "intruder@test.com", "password123")
828 .await;
829 h.login("intruder", "password123").await;
830
831 // Attempt to presign to creator's item
832 let body = json!({
833 "item_id": item_id,
834 "file_type": "audio",
835 "file_name": "evil.mp3",
836 "content_type": "audio/mpeg",
837 });
838 let resp = h
839 .client
840 .post_json("/api/upload/presign", &body.to_string())
841 .await;
842 assert_eq!(
843 resp.status.as_u16(),
844 403,
845 "Expected 403, got: {}",
846 resp.text
847 );
848 }
849
850 // Confirm-handler failure & rollback contract (test-fuzz Phase 2.2)
851 //
852 // The Run #9 tx port made the confirm handlers charge storage inside a
853 // transaction and route orphaned keys through the deletion queue. These pin the
854 // observable contract that work protects: a FAILED confirm must never inflate
855 // the storage counter and never leak the S3 object, and the deterministic
856 // reachable tx paths (replace storage-math + old-key orphan enqueue) must hold.
857 //
858 // NOTE on the pure lost-race rollback branches (uploads `Ok(0)`, versions
859 // `Ok(false)`, and the in-tx `Err`): these fire only when a second confirm, or
860 // an item/version delete, interleaves inside the handler's read-then-write
861 // window. `try_apply_storage_on` takes the users-row lock, which serializes
862 // concurrent confirms, so the branch is genuinely a TOCTOU guard. It is not
863 // reachable from a single sequential request (the test client is cookie-bound
864 // and `&mut`, with no exposed app handle for a concurrent same-user pair), so it
865 // is left to the lower-level race coverage; the tests below exercise the same
866 // transaction body and the same orphan-queue helper on their reachable paths.
867
868 /// SMALL_FILES tier storage cap, in bytes (250 GiB). Mirrors
869 /// `CreatorTier::SmallFiles.max_storage_bytes()`.
870 const SMALL_FILES_CAP: i64 = 250 * 1024 * 1024 * 1024;
871
872 async fn presign_audio(h: &mut TestHarness, item_id: &str, file_name: &str) -> String {
873 let body = json!({
874 "item_id": item_id,
875 "file_type": "audio",
876 "file_name": file_name,
877 "content_type": "audio/mpeg",
878 });
879 let resp = h
880 .client
881 .post_json("/api/upload/presign", &body.to_string())
882 .await;
883 assert!(resp.status.is_success(), "presign failed: {}", resp.text);
884 let data: Value = resp.json();
885 data["s3_key"].as_str().unwrap().to_string()
886 }
887
888 async fn presign_version(h: &mut TestHarness, version_id: &str, file_name: &str) -> String {
889 let resp = h
890 .client
891 .post_json(
892 &format!("/api/versions/{version_id}/upload/presign"),
893 &json!({"file_name": file_name, "content_type": "application/zip"}).to_string(),
894 )
895 .await;
896 assert!(
897 resp.status.is_success(),
898 "version presign failed: {}",
899 resp.text
900 );
901 let data: Value = resp.json();
902 data["s3_key"].as_str().unwrap().to_string()
903 }
904
905 async fn storage_used(h: &TestHarness, user_id: &str) -> i64 {
906 sqlx::query_scalar("SELECT storage_used_bytes FROM users WHERE id = $1::uuid")
907 .bind(user_id)
908 .fetch_one(&h.db)
909 .await
910 .unwrap()
911 }
912
913 #[tokio::test]
914 async fn confirm_over_storage_cap_does_not_charge_or_leak() {
915 let mut h = TestHarness::with_storage().await;
916 let (user_id, _project_id, item_id) = setup_creator_with_item(&mut h, 0).await;
917
918 let s3_key = presign_audio(&mut h, &item_id, "big.mp3").await;
919 h.storage.as_ref().unwrap().put(&s3_key, vec![0u8; 100]);
920
921 // Park the counter one increment shy of the cap so this 100-byte file pushes over.
922 let parked = SMALL_FILES_CAP - 50;
923 sqlx::query("UPDATE users SET storage_used_bytes = $2 WHERE id = $1::uuid")
924 .bind(&user_id)
925 .bind(parked)
926 .execute(&h.db)
927 .await
928 .unwrap();
929
930 let body = json!({"item_id": item_id, "file_type": "audio", "s3_key": s3_key});
931 let resp = h
932 .client
933 .post_json("/api/upload/confirm", &body.to_string())
934 .await;
935 assert!(
936 !resp.status.is_success(),
937 "over-cap confirm must fail, got: {} {}",
938 resp.status,
939 resp.text
940 );
941
942 // Counter unchanged, a failed confirm never inflates storage.
943 assert_eq!(
944 storage_used(&h, &user_id).await,
945 parked,
946 "failed confirm must not charge storage"
947 );
948 // The rejected confirm enqueues the orphan for deletion; run the queue (the
949 // scheduler's job in production) before asserting the object is gone.
950 h.drain_s3_deletions().await;
951 // And the object was cleaned up, not leaked.
952 assert!(
953 !h.storage
954 .as_ref()
955 .unwrap()
956 .object_exists(&s3_key)
957 .await
958 .unwrap(),
959 "over-cap confirm must delete the orphaned object"
960 );
961 // The item never picked up the key.
962 let db_key: Option<String> =
963 sqlx::query_scalar("SELECT audio_s3_key FROM items WHERE id = $1::uuid")
964 .bind(&item_id)
965 .fetch_one(&h.db)
966 .await
967 .unwrap();
968 assert_eq!(
969 db_key, None,
970 "item must not reference a key from a failed confirm"
971 );
972 }
973
974 #[tokio::test]
975 async fn confirm_wrong_route_file_type_deletes_object_and_does_not_charge() {
976 let mut h = TestHarness::with_storage().await;
977 let (user_id, _project_id, item_id) = setup_creator_with_item(&mut h, 0).await;
978
979 // Presign as audio (a key under user/item/), but confirm it as a "download",
980 // download has its own /api/versions route, so the item-upload confirm must
981 // reject it, delete the object, and charge nothing.
982 let s3_key = presign_audio(&mut h, &item_id, "song.mp3").await;
983 h.storage.as_ref().unwrap().put(&s3_key, vec![0u8; 500]);
984
985 let body = json!({"item_id": item_id, "file_type": "download", "s3_key": s3_key});
986 let resp = h
987 .client
988 .post_json("/api/upload/confirm", &body.to_string())
989 .await;
990 assert_eq!(
991 resp.status.as_u16(),
992 400,
993 "misrouted file type must 400, got: {} {}",
994 resp.status,
995 resp.text
996 );
997
998 assert_eq!(
999 storage_used(&h, &user_id).await,
1000 0,
1001 "misrouted confirm must charge nothing"
1002 );
1003 // Run the orphan-deletion queue (scheduler's job in production) before the assert.
1004 h.drain_s3_deletions().await;
1005 assert!(
1006 !h.storage
1007 .as_ref()
1008 .unwrap()
1009 .object_exists(&s3_key)
1010 .await
1011 .unwrap(),
1012 "misrouted confirm must delete the object (it would otherwise leak, the scan_jobs/scan_status footgun the guard prevents)"
1013 );
1014 }
1015
1016 #[tokio::test]
1017 async fn confirm_replace_charges_delta_and_orphans_old_key() {
1018 let mut h = TestHarness::with_storage().await;
1019 let (user_id, _project_id, item_id) = setup_creator_with_item(&mut h, 0).await;
1020
1021 // First upload: 1000 bytes.
1022 let key1 = presign_audio(&mut h, &item_id, "v1.mp3").await;
1023 h.storage.as_ref().unwrap().put(&key1, vec![0u8; 1000]);
1024 let body = json!({"item_id": item_id, "file_type": "audio", "s3_key": key1});
1025 let resp = h
1026 .client
1027 .post_json("/api/upload/confirm", &body.to_string())
1028 .await;
1029 assert!(
1030 resp.status.is_success(),
1031 "first confirm failed: {}",
1032 resp.text
1033 );
1034 assert_eq!(
1035 storage_used(&h, &user_id).await,
1036 1000,
1037 "first upload charges its full size"
1038 );
1039
1040 // Replace with a 300-byte upload.
1041 let key2 = presign_audio(&mut h, &item_id, "v2.mp3").await;
1042 h.storage.as_ref().unwrap().put(&key2, vec![0u8; 300]);
1043 let body = json!({"item_id": item_id, "file_type": "audio", "s3_key": key2});
1044 let resp = h
1045 .client
1046 .post_json("/api/upload/confirm", &body.to_string())
1047 .await;
1048 assert!(
1049 resp.status.is_success(),
1050 "replace confirm failed: {}",
1051 resp.text
1052 );
1053
1054 // The item now points at the new key.
1055 let db_key: Option<String> =
1056 sqlx::query_scalar("SELECT audio_s3_key FROM items WHERE id = $1::uuid")
1057 .bind(&item_id)
1058 .fetch_one(&h.db)
1059 .await
1060 .unwrap();
1061 assert_eq!(db_key.as_deref(), Some(key2.as_str()));
1062
1063 // Storage reflects the DELTA, not a double-charge: 1000 - 1000 + 300 = 300.
1064 // This is the in-tx try_replace_storage_on path executing for real.
1065 assert_eq!(
1066 storage_used(&h, &user_id).await,
1067 300,
1068 "replace must apply the size delta, not stack"
1069 );
1070
1071 // The OLD key is routed through the deletion queue (not deleted inline) so a
1072 // transient S3 failure can't leak it, the same orphan-queue the lost-race
1073 // path uses.
1074 let queued: i64 = sqlx::query_scalar(
1075 "SELECT COUNT(*) FROM pending_s3_deletions WHERE s3_key = $1 AND source = 'item_upload_replace'",
1076 )
1077 .bind(&key1)
1078 .fetch_one(&h.db)
1079 .await
1080 .unwrap();
1081 assert_eq!(
1082 queued, 1,
1083 "old key must be enqueued for deletion on replace"
1084 );
1085 // It's still present in S3 right now, the worker deletes it later.
1086 assert!(
1087 h.storage
1088 .as_ref()
1089 .unwrap()
1090 .object_exists(&key1)
1091 .await
1092 .unwrap(),
1093 "old key is queued, not deleted inline"
1094 );
1095 }
1096
1097 #[tokio::test]
1098 async fn version_confirm_replace_enqueues_old_key_and_charges_delta() {
1099 let mut h = TestHarness::with_storage().await;
1100 let (user_id, _project_id, item_id) = setup_creator_with_item(&mut h, 0).await;
1101
1102 let resp = h
1103 .client
1104 .post_json(
1105 &format!("/api/items/{item_id}/versions"),
1106 &json!({"version_number": "1.0.0"}).to_string(),
1107 )
1108 .await;
1109 assert!(
1110 resp.status.is_success(),
1111 "create version failed: {}",
1112 resp.text
1113 );
1114 let version: Value = resp.json();
1115 let version_id = version["id"].as_str().unwrap().to_string();
1116
1117 // First version file: 2000 bytes.
1118 let key1 = presign_version(&mut h, &version_id, "v1.zip").await;
1119 h.storage.as_ref().unwrap().put(&key1, vec![0u8; 2000]);
1120 let resp = h
1121 .client
1122 .post_json(
1123 &format!("/api/versions/{version_id}/upload/confirm"),
1124 &json!({"s3_key": key1}).to_string(),
1125 )
1126 .await;
1127 assert!(
1128 resp.status.is_success(),
1129 "first version confirm failed: {}",
1130 resp.text
1131 );
1132 assert_eq!(storage_used(&h, &user_id).await, 2000);
1133
1134 // Replace with 600 bytes.
1135 let key2 = presign_version(&mut h, &version_id, "v2.zip").await;
1136 h.storage.as_ref().unwrap().put(&key2, vec![0u8; 600]);
1137 let resp = h
1138 .client
1139 .post_json(
1140 &format!("/api/versions/{version_id}/upload/confirm"),
1141 &json!({"s3_key": key2}).to_string(),
1142 )
1143 .await;
1144 assert!(
1145 resp.status.is_success(),
1146 "version replace confirm failed: {}",
1147 resp.text
1148 );
1149
1150 assert_eq!(
1151 storage_used(&h, &user_id).await,
1152 600,
1153 "version replace must apply the delta"
1154 );
1155 let queued: i64 = sqlx::query_scalar(
1156 "SELECT COUNT(*) FROM pending_s3_deletions WHERE s3_key = $1 AND source = 'version_replace'",
1157 )
1158 .bind(&key1)
1159 .fetch_one(&h.db)
1160 .await
1161 .unwrap();
1162 assert_eq!(
1163 queued, 1,
1164 "old version key must be enqueued for deletion on replace"
1165 );
1166 }
1167
1168 #[tokio::test]
1169 async fn confirm_idempotent_reconfirm_does_not_double_charge() {
1170 let mut h = TestHarness::with_storage().await;
1171 let (user_id, _project_id, item_id) = setup_creator_with_item(&mut h, 0).await;
1172
1173 let s3_key = presign_audio(&mut h, &item_id, "track.mp3").await;
1174 h.storage.as_ref().unwrap().put(&s3_key, vec![0u8; 500]);
1175 let body = json!({"item_id": item_id, "file_type": "audio", "s3_key": s3_key});
1176
1177 let resp = h
1178 .client
1179 .post_json("/api/upload/confirm", &body.to_string())
1180 .await;
1181 assert!(
1182 resp.status.is_success(),
1183 "first confirm failed: {}",
1184 resp.text
1185 );
1186 assert_eq!(storage_used(&h, &user_id).await, 500);
1187 // pending_uploads cleared on confirm (Run #7 HIGH-1: otherwise the reaper
1188 // deletes the live object 24h later).
1189 let pending_after_first: i64 =
1190 sqlx::query_scalar("SELECT COUNT(*) FROM pending_uploads WHERE s3_key = $1")
1191 .bind(&s3_key)
1192 .fetch_one(&h.db)
1193 .await
1194 .unwrap();
1195 assert_eq!(
1196 pending_after_first, 0,
1197 "confirm must clear the pending_uploads row"
1198 );
1199
1200 // Re-confirm the SAME key: idempotent success, no second charge.
1201 let resp = h
1202 .client
1203 .post_json("/api/upload/confirm", &body.to_string())
1204 .await;
1205 assert!(
1206 resp.status.is_success(),
1207 "idempotent re-confirm should succeed: {}",
1208 resp.text
1209 );
1210 assert_eq!(
1211 storage_used(&h, &user_id).await,
1212 500,
1213 "idempotent re-confirm must NOT double-charge storage"
1214 );
1215 assert!(
1216 h.storage
1217 .as_ref()
1218 .unwrap()
1219 .object_exists(&s3_key)
1220 .await
1221 .unwrap(),
1222 "idempotent re-confirm must not delete the live object"
1223 );
1224 }
1225
1226 // CAS guard pins (test-fuzz Phase 2.2)
1227 //
1228 // The Run #9 tx port sealed every confirm-upload write behind an
1229 // `IS NOT DISTINCT FROM expected_old` compare-and-swap so a confirm that lost a
1230 // concurrent race (or whose target row was deleted/transferred mid-flight)
1231 // matches zero rows and the surrounding tx rolls back, never double-crediting
1232 // storage and never clobbering the live object the winning confirm published.
1233 //
1234 // The handler-level rollback + orphan-queue wiring on that branch can only fire
1235 // under TRUE concurrency (a sequential request always observes its own read, so
1236 // its CAS always matches, see the comment at uploads.rs around the confirm tx).
1237 // What IS deterministic, and what these pin, is the CAS predicate itself: feed
1238 // the db function a stale `expected_old` and assert it (a) reports the lost-race
1239 // outcome and (b) leaves the row untouched. A regression that dropped the guard
1240 // would make the stale write land here.
1241
1242 #[tokio::test]
1243 async fn update_item_file_cas_guards_against_stale_confirm() {
1244 use makenotwork::db::items::{FileConfirmOutcome, update_item_file_cas};
1245 use makenotwork::db::{ItemId, UserId};
1246 use makenotwork::storage::FileType;
1247
1248 let mut h = TestHarness::with_storage().await;
1249 let (user_id, _project_id, item_id) = setup_creator_with_item(&mut h, 0).await;
1250 let item = ItemId::from(uuid::Uuid::parse_str(&item_id).unwrap());
1251 let owner = UserId::from(uuid::Uuid::parse_str(&user_id).unwrap());
1252
1253 // audio_s3_key is NULL, so the first confirm (expected_old = None) wins.
1254 let r = update_item_file_cas(&h.db, item, owner, FileType::Audio, None, "key_v1", 1000)
1255 .await
1256 .unwrap();
1257 assert!(matches!(r, FileConfirmOutcome::Committed));
1258
1259 // A second confirm that still observed the pre-state (None) loses: the row
1260 // now holds key_v1, so `IS NOT DISTINCT FROM NULL` matches zero rows.
1261 let r = update_item_file_cas(&h.db, item, owner, FileType::Audio, None, "key_v2", 2000)
1262 .await
1263 .unwrap();
1264 assert!(
1265 matches!(r, FileConfirmOutcome::LostRace),
1266 "stale-None confirm must lose the CAS race"
1267 );
1268
1269 // The loser left the row untouched, key_v1/1000, not key_v2 (no clobber, no double-credit).
1270 let (k, sz): (Option<String>, Option<i64>) =
1271 sqlx::query_as("SELECT audio_s3_key, audio_file_size_bytes FROM items WHERE id = $1::uuid")
1272 .bind(&item_id)
1273 .fetch_one(&h.db)
1274 .await
1275 .unwrap();
1276 assert_eq!(k.as_deref(), Some("key_v1"));
1277 assert_eq!(sz, Some(1000));
1278
1279 // A correct CAS (expected_old = the current key) commits the swap.
1280 let r = update_item_file_cas(
1281 &h.db,
1282 item,
1283 owner,
1284 FileType::Audio,
1285 Some("key_v1"),
1286 "key_v3",
1287 3000,
1288 )
1289 .await
1290 .unwrap();
1291 assert!(matches!(r, FileConfirmOutcome::Committed));
1292
1293 // The ownership filter is part of the same predicate: a non-owner never
1294 // matches, even with the correct expected_old.
1295 let stranger = UserId::new();
1296 let r = update_item_file_cas(
1297 &h.db,
1298 item,
1299 stranger,
1300 FileType::Audio,
1301 Some("key_v3"),
1302 "key_v4",
1303 4000,
1304 )
1305 .await
1306 .unwrap();
1307 assert!(
1308 matches!(r, FileConfirmOutcome::LostRace),
1309 "a non-owner must not write the item file"
1310 );
1311 let k: Option<String> =
1312 sqlx::query_scalar("SELECT audio_s3_key FROM items WHERE id = $1::uuid")
1313 .bind(&item_id)
1314 .fetch_one(&h.db)
1315 .await
1316 .unwrap();
1317 assert_eq!(
1318 k.as_deref(),
1319 Some("key_v3"),
1320 "the non-owner write must not land"
1321 );
1322 }
1323
1324 #[tokio::test]
1325 async fn update_version_file_guards_against_stale_confirm() {
1326 use makenotwork::db::VersionId;
1327 use makenotwork::db::versions::update_version_file;
1328
1329 let mut h = TestHarness::with_storage().await;
1330 let (_user, _proj, item_id) = setup_creator_with_item(&mut h, 0).await;
1331
1332 let resp = h
1333 .client
1334 .post_json(
1335 &format!("/api/items/{item_id}/versions"),
1336 &json!({"version_number": "1.0.0"}).to_string(),
1337 )
1338 .await;
1339 assert!(
1340 resp.status.is_success(),
1341 "create version failed: {}",
1342 resp.text
1343 );
1344 let v: Value = resp.json();
1345 let version_id = v["id"].as_str().unwrap().to_string();
1346 let vid = VersionId::from(uuid::Uuid::parse_str(&version_id).unwrap());
1347
1348 // s3_key is NULL initially. A confirm carrying a stale non-null expected key
1349 // matches zero rows.
1350 let r = update_version_file(
1351 &h.db,
1352 vid,
1353 Some("not_the_current_key"),
1354 "vk_v1",
1355 Some(1000),
1356 Some("a.zip"),
1357 )
1358 .await
1359 .unwrap();
1360 assert!(r.is_none(), "a stale expected_old key must match zero rows");
1361 let k: Option<String> = sqlx::query_scalar("SELECT s3_key FROM versions WHERE id = $1::uuid")
1362 .bind(&version_id)
1363 .fetch_one(&h.db)
1364 .await
1365 .unwrap();
1366 assert!(k.is_none(), "the row must stay unwritten after a lost CAS");
1367
1368 // Correct CAS (expected None) commits.
1369 let r = update_version_file(&h.db, vid, None, "vk_v1", Some(1000), Some("a.zip"))
1370 .await
1371 .unwrap();
1372 assert!(r.is_some());
1373
1374 // A replace that still observed the pre-state (None) loses to the committed key.
1375 let r = update_version_file(&h.db, vid, None, "vk_v2", Some(2000), Some("b.zip"))
1376 .await
1377 .unwrap();
1378 assert!(
1379 r.is_none(),
1380 "stale-None replace must lose once the key is set"
1381 );
1382 let (k, sz): (Option<String>, Option<i64>) =
1383 sqlx::query_as("SELECT s3_key, file_size_bytes FROM versions WHERE id = $1::uuid")
1384 .bind(&version_id)
1385 .fetch_one(&h.db)
1386 .await
1387 .unwrap();
1388 assert_eq!(k.as_deref(), Some("vk_v1"));
1389 assert_eq!(sz, Some(1000));
1390 }
1391
1392 #[tokio::test]
1393 async fn update_project_cover_cas_guards_against_stale_and_non_owner() {
1394 use makenotwork::db::projects::update_project_cover_cas;
1395 use makenotwork::db::{ProjectId, UserId};
1396
1397 let mut h = TestHarness::with_storage().await;
1398 let (user_id, project_id, _item) = setup_creator_with_item(&mut h, 0).await;
1399 let pid = ProjectId::from(uuid::Uuid::parse_str(&project_id).unwrap());
1400 let owner = UserId::from(uuid::Uuid::parse_str(&user_id).unwrap());
1401
1402 // cover_image_url is NULL. A stale expected url matches zero rows.
1403 let ok = update_project_cover_cas(
1404 &h.db,
1405 pid,
1406 owner,
1407 Some("stale_url"),
1408 "url_v1",
1409 "key_v1",
1410 1000,
1411 )
1412 .await
1413 .unwrap();
1414 assert!(!ok, "a stale expected url must match zero rows");
1415
1416 // Correct CAS (expected None) commits.
1417 let ok = update_project_cover_cas(&h.db, pid, owner, None, "url_v1", "key_v1", 1000)
1418 .await
1419 .unwrap();
1420 assert!(ok);
1421
1422 // Stale-None loses now that the cover is set.
1423 let ok = update_project_cover_cas(&h.db, pid, owner, None, "url_v2", "key_v2", 2000)
1424 .await
1425 .unwrap();
1426 assert!(!ok, "stale-None confirm must lose once the cover is set");
1427
1428 // A non-owner cannot write, even with the correct expected url.
1429 let stranger = UserId::new();
1430 let ok = update_project_cover_cas(
1431 &h.db,
1432 pid,
1433 stranger,
1434 Some("url_v1"),
1435 "url_v3",
1436 "key_v3",
1437 3000,
1438 )
1439 .await
1440 .unwrap();
1441 assert!(!ok, "a non-owner must not write the project cover");
1442
1443 let (url, key, sz): (Option<String>, Option<String>, Option<i64>) = sqlx::query_as(
1444 "SELECT cover_image_url, cover_s3_key, cover_image_size_bytes FROM projects WHERE id = $1::uuid",
1445 )
1446 .bind(&project_id)
1447 .fetch_one(&h.db)
1448 .await
1449 .unwrap();
1450 assert_eq!(url.as_deref(), Some("url_v1"));
1451 assert_eq!(
1452 key.as_deref(),
1453 Some("key_v1"),
1454 "the bare cover key is persisted alongside the url"
1455 );
1456 assert_eq!(sz, Some(1000));
1457 }
1458
1459 #[tokio::test]
1460 async fn update_item_cover_guards_against_stale_confirm() {
1461 use makenotwork::db::items::update_item_cover;
1462 use makenotwork::db::{ItemId, UserId};
1463
1464 let mut h = TestHarness::with_storage().await;
1465 let (user_id, _proj, item_id) = setup_creator_with_item(&mut h, 0).await;
1466 let item = ItemId::from(uuid::Uuid::parse_str(&item_id).unwrap());
1467 let owner = UserId::from(uuid::Uuid::parse_str(&user_id).unwrap());
1468
1469 // cover_s3_key is NULL. A stale expected key matches zero rows.
1470 let ok = update_item_cover(&h.db, item, owner, Some("stale_key"), "u1", "ck_v1", 1000)
1471 .await
1472 .unwrap();
1473 assert!(!ok, "a stale expected key must match zero rows");
1474
1475 // Correct CAS (expected None) commits the cover key + url + size together.
1476 let ok = update_item_cover(&h.db, item, owner, None, "u1", "ck_v1", 1000)
1477 .await
1478 .unwrap();
1479 assert!(ok);
1480
1481 // Stale-None loses now that the cover key is set.
1482 let ok = update_item_cover(&h.db, item, owner, None, "u2", "ck_v2", 2000)
1483 .await
1484 .unwrap();
1485 assert!(
1486 !ok,
1487 "stale-None confirm must lose once the cover key is set"
1488 );
1489
1490 // A non-owner cannot write, even with the correct expected key.
1491 let stranger = UserId::new();
1492 let ok = update_item_cover(&h.db, item, stranger, Some("ck_v1"), "u3", "ck_v3", 3000)
1493 .await
1494 .unwrap();
1495 assert!(!ok, "a non-owner must not write the item cover");
1496
1497 let (key, url, sz): (Option<String>, Option<String>, Option<i64>) = sqlx::query_as(
1498 "SELECT cover_s3_key, cover_image_url, cover_file_size_bytes FROM items WHERE id = $1::uuid",
1499 )
1500 .bind(&item_id)
1501 .fetch_one(&h.db)
1502 .await
1503 .unwrap();
1504 assert_eq!(key.as_deref(), Some("ck_v1"));
1505 assert_eq!(url.as_deref(), Some("u1"));
1506 assert_eq!(sz, Some(1000));
1507 }
1508
1509 // Confirm failure-branch coverage via DB fault injection (test-fuzz Phase 2.2+)
1510 //
1511 // The CAS-guard pins above prove the predicate REJECTS a stale write. They do
1512 // NOT exercise the handler's REACTION to a rejection, the storage-credit
1513 // rollback (the credit and the CAS run in one tx; the handler returns without
1514 // committing) and the orphan-enqueue of the staged key. Those branches only
1515 // fire under a true concurrent confirm or a mid-tx DB error, neither of which a
1516 // sequential test drives directly.
1517 //
1518 // We inject the fault at the DB layer with BEFORE UPDATE triggers keyed on a
1519 // sentinel embedded in the uploaded key: a "faultlr" marker makes the guarded
1520 // UPDATE affect zero rows (RETURN NULL), the lost-race branch (Ok(false)/Ok(0));
1521 // a "faulterr" marker makes it RAISE, the commit-Err branch. Both are fully
1522 // deterministic, need no concurrency, and are scoped to this test's cloned DB,
1523 // so no production code carries a test hook and no other test is affected (the
1524 // triggers are inert for any key without the marker).
1525 //
1526 // Each test asserts the two safety properties of the failed confirm: storage is
1527 // NOT credited (the tx rolled back) and the staged S3 key is orphan-enqueued
1528 // with the handler's reason string (so the reaper cleans it and a blind delete
1529 // never clobbers a live object).
1530
1531 /// Install confirm-time fault triggers on items/versions/projects in this test's
1532 /// cloned DB. A key containing `faultlr` skips the guarded UPDATE (zero rows ->
1533 /// lost-race branch); a key containing `faulterr` raises (commit-Err branch).
1534 async fn install_confirm_fault_triggers(pool: &sqlx::PgPool) {
1535 let stmts = [
1536 r"CREATE OR REPLACE FUNCTION test_confirm_fault_items() RETURNS trigger AS $$
1537 BEGIN
1538 IF COALESCE(NEW.audio_s3_key,'') LIKE '%faulterr%' OR COALESCE(NEW.cover_s3_key,'') LIKE '%faulterr%' THEN
1539 RAISE EXCEPTION 'injected confirm fault (items)';
1540 END IF;
1541 IF COALESCE(NEW.audio_s3_key,'') LIKE '%faultlr%' OR COALESCE(NEW.cover_s3_key,'') LIKE '%faultlr%' THEN
1542 RETURN NULL;
1543 END IF;
1544 RETURN NEW;
1545 END; $$ LANGUAGE plpgsql",
1546 "DROP TRIGGER IF EXISTS test_confirm_fault_items ON items",
1547 "CREATE TRIGGER test_confirm_fault_items BEFORE UPDATE ON items FOR EACH ROW EXECUTE FUNCTION test_confirm_fault_items()",
1548 r"CREATE OR REPLACE FUNCTION test_confirm_fault_versions() RETURNS trigger AS $$
1549 BEGIN
1550 IF COALESCE(NEW.s3_key,'') LIKE '%faulterr%' THEN RAISE EXCEPTION 'injected confirm fault (versions)'; END IF;
1551 IF COALESCE(NEW.s3_key,'') LIKE '%faultlr%' THEN RETURN NULL; END IF;
1552 RETURN NEW;
1553 END; $$ LANGUAGE plpgsql",
1554 "DROP TRIGGER IF EXISTS test_confirm_fault_versions ON versions",
1555 "CREATE TRIGGER test_confirm_fault_versions BEFORE UPDATE ON versions FOR EACH ROW EXECUTE FUNCTION test_confirm_fault_versions()",
1556 r"CREATE OR REPLACE FUNCTION test_confirm_fault_projects() RETURNS trigger AS $$
1557 BEGIN
1558 IF COALESCE(NEW.cover_image_url,'') LIKE '%faulterr%' THEN RAISE EXCEPTION 'injected confirm fault (projects)'; END IF;
1559 IF COALESCE(NEW.cover_image_url,'') LIKE '%faultlr%' THEN RETURN NULL; END IF;
1560 RETURN NEW;
1561 END; $$ LANGUAGE plpgsql",
1562 "DROP TRIGGER IF EXISTS test_confirm_fault_projects ON projects",
1563 "CREATE TRIGGER test_confirm_fault_projects BEFORE UPDATE ON projects FOR EACH ROW EXECUTE FUNCTION test_confirm_fault_projects()",
1564 ];
1565 for s in stmts {
1566 sqlx::query(s)
1567 .execute(pool)
1568 .await
1569 .expect("install fault trigger");
1570 }
1571 }
1572
1573 /// Assert a failed confirm rolled back the storage credit AND orphan-enqueued the
1574 /// staged key with the expected reason.
1575 async fn assert_uncredited_and_orphaned(
1576 h: &TestHarness,
1577 user_id: &str,
1578 s3_key: &str,
1579 before: i64,
1580 expected_source: &str,
1581 ) {
1582 assert_eq!(
1583 storage_used(h, user_id).await,
1584 before,
1585 "a failed confirm must roll back the storage credit"
1586 );
1587 let source: Option<String> =
1588 sqlx::query_scalar("SELECT source FROM pending_s3_deletions WHERE s3_key = $1")
1589 .bind(s3_key)
1590 .fetch_optional(&h.db)
1591 .await
1592 .unwrap();
1593 assert_eq!(
1594 source.as_deref(),
1595 Some(expected_source),
1596 "a failed confirm must orphan-enqueue the staged key with the handler's reason"
1597 );
1598 }
1599
1600 async fn presign_project_image(h: &mut TestHarness, project_id: &str, file_name: &str) -> String {
1601 let resp = h
1602 .client
1603 .post_json(
1604 "/api/projects/image/presign",
1605 &json!({"project_id": project_id, "file_name": file_name, "content_type": "image/png"})
1606 .to_string(),
1607 )
1608 .await;
1609 assert!(
1610 resp.status.is_success(),
1611 "project image presign failed: {}",
1612 resp.text
1613 );
1614 let data: Value = resp.json();
1615 data["s3_key"].as_str().unwrap().to_string()
1616 }
1617
1618 async fn presign_item_image(h: &mut TestHarness, item_id: &str, file_name: &str) -> String {
1619 let resp = h
1620 .client
1621 .post_json(
1622 "/api/items/image/presign",
1623 &json!({"item_id": item_id, "file_name": file_name, "content_type": "image/png"})
1624 .to_string(),
1625 )
1626 .await;
1627 assert!(
1628 resp.status.is_success(),
1629 "item image presign failed: {}",
1630 resp.text
1631 );
1632 let data: Value = resp.json();
1633 data["s3_key"].as_str().unwrap().to_string()
1634 }
1635
1636 // ---- uploads (item file) -------------------------------------------------
1637
1638 #[tokio::test]
1639 async fn confirm_audio_lost_race_rolls_back_and_orphans() {
1640 let mut h = TestHarness::with_storage().await;
1641 let (user_id, _proj, item_id) = setup_creator_with_item(&mut h, 0).await;
1642 install_confirm_fault_triggers(&h.db).await;
1643
1644 let s3_key = presign_audio(&mut h, &item_id, "faultlr.mp3").await;
1645 h.storage.as_ref().unwrap().put(&s3_key, b"x".to_vec());
1646 let before = storage_used(&h, &user_id).await;
1647
1648 let resp = h
1649 .client
1650 .post_json(
1651 "/api/upload/confirm",
1652 &json!({"item_id": item_id, "file_type": "audio", "s3_key": s3_key}).to_string(),
1653 )
1654 .await;
1655 assert_eq!(
1656 resp.status.as_u16(),
1657 400,
1658 "lost-race confirm must 400: {}",
1659 resp.text
1660 );
1661 assert_uncredited_and_orphaned(&h, &user_id, &s3_key, before, "item_upload_target_missing")
1662 .await;
1663 }
1664
1665 #[tokio::test]
1666 async fn confirm_audio_tx_error_rolls_back_and_orphans() {
1667 let mut h = TestHarness::with_storage().await;
1668 let (user_id, _proj, item_id) = setup_creator_with_item(&mut h, 0).await;
1669 install_confirm_fault_triggers(&h.db).await;
1670
1671 let s3_key = presign_audio(&mut h, &item_id, "faulterr.mp3").await;
1672 h.storage.as_ref().unwrap().put(&s3_key, b"x".to_vec());
1673 let before = storage_used(&h, &user_id).await;
1674
1675 let resp = h
1676 .client
1677 .post_json(
1678 "/api/upload/confirm",
1679 &json!({"item_id": item_id, "file_type": "audio", "s3_key": s3_key}).to_string(),
1680 )
1681 .await;
1682 assert!(
1683 resp.status.is_server_error(),
1684 "tx-error confirm must 5xx: {} {}",
1685 resp.status,
1686 resp.text
1687 );
1688 assert_uncredited_and_orphaned(&h, &user_id, &s3_key, before, "item_confirm_failed").await;
1689 }
1690
1691 // ---- versions ------------------------------------------------------------
1692
1693 async fn make_version(h: &mut TestHarness, item_id: &str) -> String {
1694 let resp = h
1695 .client
1696 .post_json(
1697 &format!("/api/items/{item_id}/versions"),
1698 &json!({"version_number": "1.0.0"}).to_string(),
1699 )
1700 .await;
1701 assert!(
1702 resp.status.is_success(),
1703 "create version failed: {}",
1704 resp.text
1705 );
1706 let v: Value = resp.json();
1707 v["id"].as_str().unwrap().to_string()
1708 }
1709
1710 #[tokio::test]
1711 async fn confirm_version_lost_race_rolls_back_and_orphans() {
1712 let mut h = TestHarness::with_storage().await;
1713 let (user_id, _proj, item_id) = setup_creator_with_item(&mut h, 0).await;
1714 let version_id = make_version(&mut h, &item_id).await;
1715 install_confirm_fault_triggers(&h.db).await;
1716
1717 let s3_key = presign_version(&mut h, &version_id, "faultlr.zip").await;
1718 h.storage.as_ref().unwrap().put(&s3_key, b"x".to_vec());
1719 let before = storage_used(&h, &user_id).await;
1720
1721 let resp = h
1722 .client
1723 .post_json(
1724 &format!("/api/versions/{version_id}/upload/confirm"),
1725 &json!({"s3_key": s3_key}).to_string(),
1726 )
1727 .await;
1728 assert_eq!(
1729 resp.status.as_u16(),
1730 400,
1731 "lost-race version confirm must 400: {}",
1732 resp.text
1733 );
1734 assert_uncredited_and_orphaned(&h, &user_id, &s3_key, before, "version_confirm_lost_race")
1735 .await;
1736 }
1737
1738 #[tokio::test]
1739 async fn confirm_version_tx_error_rolls_back_and_orphans() {
1740 let mut h = TestHarness::with_storage().await;
1741 let (user_id, _proj, item_id) = setup_creator_with_item(&mut h, 0).await;
1742 let version_id = make_version(&mut h, &item_id).await;
1743 install_confirm_fault_triggers(&h.db).await;
1744
1745 let s3_key = presign_version(&mut h, &version_id, "faulterr.zip").await;
1746 h.storage.as_ref().unwrap().put(&s3_key, b"x".to_vec());
1747 let before = storage_used(&h, &user_id).await;
1748
1749 let resp = h
1750 .client
1751 .post_json(
1752 &format!("/api/versions/{version_id}/upload/confirm"),
1753 &json!({"s3_key": s3_key}).to_string(),
1754 )
1755 .await;
1756 assert!(
1757 resp.status.is_server_error(),
1758 "tx-error version confirm must 5xx: {} {}",
1759 resp.status,
1760 resp.text
1761 );
1762 assert_uncredited_and_orphaned(&h, &user_id, &s3_key, before, "version_confirm_failed").await;
1763 }
1764
1765 // ---- project cover image -------------------------------------------------
1766
1767 #[tokio::test]
1768 async fn confirm_project_image_lost_race_rolls_back_and_orphans() {
1769 let mut h = TestHarness::with_storage().await;
1770 let (user_id, project_id, _item) = setup_creator_with_item(&mut h, 0).await;
1771 install_confirm_fault_triggers(&h.db).await;
1772
1773 let s3_key = presign_project_image(&mut h, &project_id, "faultlr.png").await;
1774 h.storage.as_ref().unwrap().put(&s3_key, b"x".to_vec());
1775 let before = storage_used(&h, &user_id).await;
1776
1777 let resp = h
1778 .client
1779 .post_json(
1780 "/api/projects/image/confirm",
1781 &json!({"project_id": project_id, "s3_key": s3_key}).to_string(),
1782 )
1783 .await;
1784 assert_eq!(
1785 resp.status.as_u16(),
1786 400,
1787 "lost-race project image confirm must 400: {}",
1788 resp.text
1789 );
1790 assert_uncredited_and_orphaned(&h, &user_id, &s3_key, before, "project_image_update_failed")
1791 .await;
1792 }
1793
1794 #[tokio::test]
1795 async fn confirm_project_image_tx_error_rolls_back_and_orphans() {
1796 let mut h = TestHarness::with_storage().await;
1797 let (user_id, project_id, _item) = setup_creator_with_item(&mut h, 0).await;
1798 install_confirm_fault_triggers(&h.db).await;
1799
1800 let s3_key = presign_project_image(&mut h, &project_id, "faulterr.png").await;
1801 h.storage.as_ref().unwrap().put(&s3_key, b"x".to_vec());
1802 let before = storage_used(&h, &user_id).await;
1803
1804 let resp = h
1805 .client
1806 .post_json(
1807 "/api/projects/image/confirm",
1808 &json!({"project_id": project_id, "s3_key": s3_key}).to_string(),
1809 )
1810 .await;
1811 assert!(
1812 resp.status.is_server_error(),
1813 "tx-error project image confirm must 5xx: {} {}",
1814 resp.status,
1815 resp.text
1816 );
1817 assert_uncredited_and_orphaned(&h, &user_id, &s3_key, before, "project_image_update_failed")
1818 .await;
1819 }
1820
1821 // ---- item cover image ----------------------------------------------------
1822
1823 #[tokio::test]
1824 async fn confirm_item_image_lost_race_rolls_back_and_orphans() {
1825 let mut h = TestHarness::with_storage().await;
1826 let (user_id, _proj, item_id) = setup_creator_with_item(&mut h, 0).await;
1827 install_confirm_fault_triggers(&h.db).await;
1828
1829 let s3_key = presign_item_image(&mut h, &item_id, "faultlr.png").await;
1830 h.storage.as_ref().unwrap().put(&s3_key, b"x".to_vec());
1831 let before = storage_used(&h, &user_id).await;
1832
1833 let resp = h
1834 .client
1835 .post_json(
1836 "/api/items/image/confirm",
1837 &json!({"item_id": item_id, "s3_key": s3_key}).to_string(),
1838 )
1839 .await;
1840 assert_eq!(
1841 resp.status.as_u16(),
1842 400,
1843 "lost-race item image confirm must 400: {}",
1844 resp.text
1845 );
1846 assert_uncredited_and_orphaned(&h, &user_id, &s3_key, before, "item_image_update_failed").await;
1847 }
1848
1849 #[tokio::test]
1850 async fn confirm_item_image_tx_error_rolls_back_and_orphans() {
1851 let mut h = TestHarness::with_storage().await;
1852 let (user_id, _proj, item_id) = setup_creator_with_item(&mut h, 0).await;
1853 install_confirm_fault_triggers(&h.db).await;
1854
1855 let s3_key = presign_item_image(&mut h, &item_id, "faulterr.png").await;
1856 h.storage.as_ref().unwrap().put(&s3_key, b"x".to_vec());
1857 let before = storage_used(&h, &user_id).await;
1858
1859 let resp = h
1860 .client
1861 .post_json(
1862 "/api/items/image/confirm",
1863 &json!({"item_id": item_id, "s3_key": s3_key}).to_string(),
1864 )
1865 .await;
1866 assert!(
1867 resp.status.is_server_error(),
1868 "tx-error item image confirm must 5xx: {} {}",
1869 resp.status,
1870 resp.text
1871 );
1872 assert_uncredited_and_orphaned(&h, &user_id, &s3_key, before, "item_image_update_failed").await;
1873 }
1874
1875 // Multipart upload sessions (CLI / desktop, large files)
1876 //
1877 // These live on the internal surface only, a browser keeps the one-shot
1878 // presigned PUT. The InMemoryStorage backend stubs the session (its presigned
1879 // URLs are fake, so no client bytes flow back), so a full start -> confirm test
1880 // simulates the part PUTs with `put()`, exactly as the presign+confirm tests do.
1881
1882 const GIB: i64 = 1024 * 1024 * 1024;
1883 const CLI_TOKEN: &str = "test-cli-token";
1884 const ACTOR_SECRET: &str = "test-signing-secret-for-integration-tests";
1885
1886 /// Harness wired for the internal (CLI) API with an in-memory backend.
1887 async fn cli_harness() -> TestHarness {
1888 let mem = std::sync::Arc::new(crate::harness::storage::InMemoryStorage::new());
1889 TestHarness::build(crate::harness::BuildOptions {
1890 storage: Some(mem),
1891 cli_service_token: Some(CLI_TOKEN.to_string()),
1892 ..Default::default()
1893 })
1894 .await
1895 }
1896
1897 /// Authenticate the client as `user_id` over the internal API (ServiceAuth
1898 /// bearer plus the signed actor assertion).
1899 fn act_as(h: &mut TestHarness, user_id: makenotwork::db::UserId) {
1900 h.client.set_bearer_token(CLI_TOKEN);
1901 let actor = makenotwork::crypto::mint_internal_actor_token(
1902 user_id,
1903 chrono::Utc::now().timestamp() + 3600,
1904 ACTOR_SECRET,
1905 );
1906 h.client.set_actor_token(&actor);
1907 }
1908
1909 #[tokio::test]
1910 async fn multipart_start_opens_the_band_above_the_browser_ceiling() {
1911 // 6 GiB is past the 2 GiB browser ceiling and inside the big_files tier's
1912 // 20 GB per-file cap. Before multipart this band was unreachable by every
1913 // path; this test is the regression guard that it stays open.
1914 let mut h = cli_harness().await;
1915 let setup = h.create_creator_with_item("mpcreator", "video", 0).await;
1916 h.trust_user(setup.user_id).await;
1917 h.grant_tier(setup.user_id, "big_files").await;
1918 act_as(&mut h, setup.user_id);
1919
1920 let body = json!({
1921 "item_id": setup.item_id,
1922 "file_type": "video",
1923 "file_name": "movie.mp4",
1924 "content_type": "video/mp4",
1925 "file_size_bytes": 6 * GIB,
1926 });
1927 let resp = h
1928 .client
1929 .post_json("/api/internal/upload/multipart/start", &body.to_string())
1930 .await;
1931 assert!(
1932 resp.status.is_success(),
1933 "multipart start failed: {}",
1934 resp.text
1935 );
1936
1937 let v: Value = resp.json();
1938 let s3_key = v["s3_key"].as_str().unwrap();
1939 assert!(
1940 s3_key.starts_with("staging/"),
1941 "must stage, never mint a served key: {s3_key}"
1942 );
1943 assert!(!v["upload_id"].as_str().unwrap().is_empty());
1944
1945 // Geometry is pure arithmetic over the declared size, so the client can
1946 // derive identical boundaries without a round trip.
1947 let part_size = v["part_size"].as_u64().unwrap();
1948 let part_count = v["part_count"].as_u64().unwrap();
1949 assert!(
1950 part_size >= 5 * 1024 * 1024,
1951 "part size must clear S3's 5 MiB floor"
1952 );
1953 assert!(
1954 part_count <= 10_000,
1955 "part count must stay within S3's limit"
1956 );
1957 assert_eq!(
1958 part_count,
1959 (6 * GIB as u64).div_ceil(part_size),
1960 "part count must cover the object"
1961 );
1962 }
1963
1964 #[tokio::test]
1965 async fn multipart_parts_signs_each_part_with_its_exact_length() {
1966 let mut h = cli_harness().await;
1967 let setup = h.create_creator_with_item("mpparts", "video", 0).await;
1968 h.trust_user(setup.user_id).await;
1969 h.grant_tier(setup.user_id, "big_files").await;
1970 act_as(&mut h, setup.user_id);
1971
1972 // A size with a deliberate remainder so the final part differs from the rest.
1973 let size = 6 * GIB + 12_345;
1974 let start: Value = h
1975 .client
1976 .post_json(
1977 "/api/internal/upload/multipart/start",
1978 &json!({
1979 "item_id": setup.item_id, "file_type": "video",
1980 "file_name": "movie.mp4", "content_type": "video/mp4",
1981 "file_size_bytes": size,
1982 })
1983 .to_string(),
1984 )
1985 .await
1986 .json();
1987 let s3_key = start["s3_key"].as_str().unwrap().to_string();
1988 let upload_id = start["upload_id"].as_str().unwrap().to_string();
1989 let part_size = start["part_size"].as_u64().unwrap();
1990 let part_count = start["part_count"].as_u64().unwrap();
1991
1992 // A leading window: every part is a full part.
1993 let resp = h
1994 .client
1995 .post_json(
1996 "/api/internal/upload/multipart/parts",
1997 &json!({
1998 "s3_key": s3_key, "upload_id": upload_id,
1999 "file_size_bytes": size, "first_part": 1, "count": 3,
2000 })
2001 .to_string(),
2002 )
2003 .await;
2004 assert!(resp.status.is_success(), "parts failed: {}", resp.text);
2005 let v: Value = resp.json();
2006 let parts = v["parts"].as_array().unwrap();
2007 assert_eq!(parts.len(), 3);
2008 for (i, p) in parts.iter().enumerate() {
2009 assert_eq!(p["part_number"].as_i64().unwrap(), i as i64 + 1);
2010 assert_eq!(p["content_length"].as_u64().unwrap(), part_size);
2011 assert!(!p["url"].as_str().unwrap().is_empty());
2012 }
2013
2014 // The final part carries only the remainder, and a window running past the
2015 // end is clamped rather than rejected.
2016 let resp = h
2017 .client
2018 .post_json(
2019 "/api/internal/upload/multipart/parts",
2020 &json!({
2021 "s3_key": s3_key, "upload_id": upload_id,
2022 "file_size_bytes": size, "first_part": part_count, "count": 10,
2023 })
2024 .to_string(),
2025 )
2026 .await;
2027 assert!(
2028 resp.status.is_success(),
2029 "final-part window failed: {}",
2030 resp.text
2031 );
2032 let v: Value = resp.json();
2033 let parts = v["parts"].as_array().unwrap();
2034 assert_eq!(parts.len(), 1, "window past the last part must clamp");
2035 assert_eq!(
2036 parts[0]["content_length"].as_u64().unwrap(),
2037 12_345,
2038 "last part is the remainder"
2039 );
2040 }
2041
2042 #[tokio::test]
2043 async fn multipart_parts_refuses_a_size_that_disagrees_with_start() {
2044 // deepaudit F1: the part geometry is bound to the size `start` validated
2045 // against the tier cap. A session opened for 2 GB must not be able to widen
2046 // itself at `parts` time, trusting the parts body would let it mint URLs for
2047 // a 5 TiB object, unbudgeted S3 writes bounded only by the 24h reaper.
2048 let mut h = cli_harness().await;
2049 let setup = h.create_creator_with_item("mpliar", "video", 0).await;
2050 h.trust_user(setup.user_id).await;
2051 h.grant_tier(setup.user_id, "big_files").await;
2052 act_as(&mut h, setup.user_id);
2053
2054 let declared = 2 * GIB;
2055 let start: Value = h
2056 .client
2057 .post_json(
2058 "/api/internal/upload/multipart/start",
2059 &json!({
2060 "item_id": setup.item_id, "file_type": "video",
2061 "file_name": "movie.mp4", "content_type": "video/mp4",
2062 "file_size_bytes": declared,
2063 })
2064 .to_string(),
2065 )
2066 .await
2067 .json();
2068 let s3_key = start["s3_key"].as_str().unwrap().to_string();
2069 let upload_id = start["upload_id"].as_str().unwrap().to_string();
2070
2071 // Claim a wildly larger size at parts time.
2072 let resp = h
2073 .client
2074 .post_json(
2075 "/api/internal/upload/multipart/parts",
2076 &json!({
2077 "s3_key": s3_key, "upload_id": upload_id,
2078 "file_size_bytes": 5 * 1024 * GIB, "first_part": 1, "count": 1,
2079 })
2080 .to_string(),
2081 )
2082 .await;
2083 assert!(
2084 !resp.status.is_success(),
2085 "a parts size disagreeing with start must be refused: {}",
2086 resp.text
2087 );
2088 assert!(
2089 resp.text.contains("does not match"),
2090 "expected a declared-size mismatch error, got {}",
2091 resp.text
2092 );
2093 }
2094
2095 #[tokio::test]
2096 async fn multipart_parts_bounds_the_window_it_will_mint() {
2097 // Minting every URL for a 20 GB object would issue thousands of hour-long
2098 // credentials for an upload that may never happen.
2099 let mut h = cli_harness().await;
2100 let setup = h.create_creator_with_item("mpwindow", "video", 0).await;
2101 h.trust_user(setup.user_id).await;
2102 h.grant_tier(setup.user_id, "big_files").await;
2103 act_as(&mut h, setup.user_id);
2104
2105 let start: Value = h
2106 .client
2107 .post_json(
2108 "/api/internal/upload/multipart/start",
2109 &json!({
2110 "item_id": setup.item_id, "file_type": "video",
2111 "file_name": "movie.mp4", "content_type": "video/mp4",
2112 "file_size_bytes": 6 * GIB,
2113 })
2114 .to_string(),
2115 )
2116 .await
2117 .json();
2118
2119 for (first_part, count) in [(1, 101), (1, 0), (0, 5)] {
2120 let resp = h
2121 .client
2122 .post_json(
2123 "/api/internal/upload/multipart/parts",
2124 &json!({
2125 "s3_key": start["s3_key"], "upload_id": start["upload_id"],
2126 "file_size_bytes": 6 * GIB, "first_part": first_part, "count": count,
2127 })
2128 .to_string(),
2129 )
2130 .await;
2131 assert!(
2132 resp.status.is_client_error(),
2133 "first_part={first_part} count={count} must be refused, got {}: {}",
2134 resp.status,
2135 resp.text
2136 );
2137 }
2138 }
2139
2140 #[tokio::test]
2141 async fn multipart_refuses_another_creators_staging_key() {
2142 // A `staging/{uuid}` key carries no user in its path, so ownership comes
2143 // from the pending_uploads row. Without that check any authenticated creator
2144 // could drive parts into someone else's in-flight session.
2145 let mut h = cli_harness().await;
2146
2147 let victim = h.create_creator_with_item("mpvictim", "video", 0).await;
2148 h.trust_user(victim.user_id).await;
2149 h.grant_tier(victim.user_id, "big_files").await;
2150 act_as(&mut h, victim.user_id);
2151 let start: Value = h
2152 .client
2153 .post_json(
2154 "/api/internal/upload/multipart/start",
2155 &json!({
2156 "item_id": victim.item_id, "file_type": "video",
2157 "file_name": "movie.mp4", "content_type": "video/mp4",
2158 "file_size_bytes": 6 * GIB,
2159 })
2160 .to_string(),
2161 )
2162 .await
2163 .json();
2164 let victim_key = start["s3_key"].as_str().unwrap().to_string();
2165 let victim_upload_id = start["upload_id"].as_str().unwrap().to_string();
2166
2167 // A second creator, fully authenticated in their own right.
2168 let attacker = h.create_creator_with_item("mpattacker", "video", 0).await;
2169 h.trust_user(attacker.user_id).await;
2170 h.grant_tier(attacker.user_id, "big_files").await;
2171 act_as(&mut h, attacker.user_id);
2172
2173 let parts = h
2174 .client
2175 .post_json(
2176 "/api/internal/upload/multipart/parts",
2177 &json!({
2178 "s3_key": victim_key, "upload_id": victim_upload_id,
2179 "file_size_bytes": 6 * GIB, "first_part": 1, "count": 1,
2180 })
2181 .to_string(),
2182 )
2183 .await;
2184 assert!(
2185 parts.status.is_client_error(),
2186 "cross-user parts must be refused: {}",
2187 parts.text
2188 );
2189
2190 let complete = h
2191 .client
2192 .post_json(
2193 "/api/internal/upload/multipart/complete",
2194 &json!({
2195 "s3_key": victim_key, "upload_id": victim_upload_id,
2196 "parts": [{"part_number": 1, "etag": "\"deadbeef\""}],
2197 })
2198 .to_string(),
2199 )
2200 .await;
2201 assert!(
2202 complete.status.is_client_error(),
2203 "cross-user complete must be refused: {}",
2204 complete.text
2205 );
2206
2207 let abort = h
2208 .client
2209 .post_json(
2210 "/api/internal/upload/multipart/abort",
2211 &json!({"s3_key": victim_key, "upload_id": victim_upload_id}).to_string(),
2212 )
2213 .await;
2214 assert!(
2215 abort.status.is_client_error(),
2216 "cross-user abort must be refused: {}",
2217 abort.text
2218 );
2219 }
2220
2221 #[tokio::test]
2222 async fn multipart_start_refuses_a_file_over_the_tier_cap() {
2223 // Skipping the single-PUT ceiling must not skip the limits that describe the
2224 // file: small_files caps a single file at 500 MB, so a 1 GiB video is refused
2225 // before it stages any parts.
2226 let mut h = cli_harness().await;
2227 let setup = h.create_creator_with_item("mptier", "video", 0).await;
2228 h.trust_user(setup.user_id).await;
2229 h.grant_tier(setup.user_id, "small_files").await;
2230 act_as(&mut h, setup.user_id);
2231
2232 let resp = h
2233 .client
2234 .post_json(
2235 "/api/internal/upload/multipart/start",
2236 &json!({
2237 "item_id": setup.item_id, "file_type": "video",
2238 "file_name": "movie.mp4", "content_type": "video/mp4",
2239 "file_size_bytes": GIB,
2240 })
2241 .to_string(),
2242 )
2243 .await;
2244 assert!(
2245 resp.status.is_client_error(),
2246 "over-tier multipart start must be refused: {}",
2247 resp.text
2248 );
2249
2250 // Nothing was staged for the reaper to clean up.
2251 let pending: i64 =
2252 sqlx::query_scalar("SELECT COUNT(*) FROM pending_uploads WHERE user_id = $1")
2253 .bind(setup.user_id)
2254 .fetch_one(&h.db)
2255 .await
2256 .unwrap();
2257 assert_eq!(
2258 pending, 0,
2259 "a refused start must not record a pending upload"
2260 );
2261 }
2262
2263 #[tokio::test]
2264 async fn multipart_complete_then_confirm_commits_the_upload() {
2265 // The end-to-end handoff: multipart replaces the transport only, and the
2266 // existing confirm applies every size/tier/scan/commit rule unchanged.
2267 let mut h = cli_harness().await;
2268 let setup = h.create_creator_with_item("mpflow", "audio", 0).await;
2269 h.trust_user(setup.user_id).await;
2270 h.grant_tier(setup.user_id, "small_files").await;
2271 act_as(&mut h, setup.user_id);
2272
2273 let bytes = b"fake mp3 bytes for a multipart upload".to_vec();
2274 let size = bytes.len() as i64;
2275
2276 let start: Value = h
2277 .client
2278 .post_json(
2279 "/api/internal/upload/multipart/start",
2280 &json!({
2281 "item_id": setup.item_id, "file_type": "audio",
2282 "file_name": "song.mp3", "content_type": "audio/mpeg",
2283 "file_size_bytes": size,
2284 })
2285 .to_string(),
2286 )
2287 .await
2288 .json();
2289 let s3_key = start["s3_key"].as_str().unwrap().to_string();
2290 let upload_id = start["upload_id"].as_str().unwrap().to_string();
2291 assert_eq!(
2292 start["part_count"].as_u64().unwrap(),
2293 1,
2294 "a small file is a single part"
2295 );
2296
2297 let parts: Value = h
2298 .client
2299 .post_json(
2300 "/api/internal/upload/multipart/parts",
2301 &json!({
2302 "s3_key": s3_key, "upload_id": upload_id,
2303 "file_size_bytes": size, "first_part": 1, "count": 1,
2304 })
2305 .to_string(),
2306 )
2307 .await
2308 .json();
2309 assert_eq!(parts["parts"][0]["content_length"].as_i64().unwrap(), size);
2310
2311 // The client PUTs each part to its presigned URL; the in-memory backend has
2312 // no way to receive them, so stand in for that here.
2313 h.storage.as_ref().unwrap().put(&s3_key, bytes);
2314
2315 let complete = h
2316 .client
2317 .post_json(
2318 "/api/internal/upload/multipart/complete",
2319 &json!({
2320 "s3_key": s3_key, "upload_id": upload_id,
2321 "parts": [{"part_number": 1, "etag": "\"etag-1\""}],
2322 })
2323 .to_string(),
2324 )
2325 .await;
2326 assert!(
2327 complete.status.is_success(),
2328 "complete failed: {}",
2329 complete.text
2330 );
2331
2332 // Hand off to the unchanged confirm endpoint.
2333 let confirm = h
2334 .client
2335 .post_json(
2336 "/api/internal/upload/confirm",
2337 &json!({
2338 "user_id": setup.user_id, "item_id": setup.item_id,
2339 "file_type": "audio", "s3_key": s3_key,
2340 })
2341 .to_string(),
2342 )
2343 .await;
2344 assert!(
2345 confirm.status.is_success(),
2346 "confirm after multipart failed: {}",
2347 confirm.text
2348 );
2349 h.client.clear_bearer_token();
2350
2351 let db_key: Option<String> =
2352 sqlx::query_scalar("SELECT audio_s3_key FROM items WHERE id = $1::uuid")
2353 .bind(&setup.item_id)
2354 .fetch_one(&h.db)
2355 .await
2356 .unwrap();
2357 assert_eq!(
2358 db_key.as_deref(),
2359 Some(s3_key.as_str()),
2360 "confirm must commit the multipart object"
2361 );
2362
2363 let storage_used: i64 =
2364 sqlx::query_scalar("SELECT storage_used_bytes FROM users WHERE id = $1")
2365 .bind(setup.user_id)
2366 .fetch_one(&h.db)
2367 .await
2368 .unwrap();
2369 assert_eq!(
2370 storage_used, size,
2371 "confirm must charge the real object size"
2372 );
2373 }
2374
2375 #[tokio::test]
2376 async fn orphan_reaper_aborts_abandoned_multipart_sessions() {
2377 // A multipart session that was started and never completed leaves NO object
2378 // to delete, only uploaded parts that S3 bills for until they are aborted.
2379 // The reaper's object delete is a no-op against it, so without this abort
2380 // the parts leak indefinitely, a cost bug, not just a tidiness one.
2381 use makenotwork::scheduler::abort_orphan_multipart_sessions;
2382
2383 let mem = crate::harness::storage::InMemoryStorage::new();
2384 let target = "staging/abandoned-uuid/movie.mp4";
2385
2386 // Two abandoned sessions on the reaped key, plus one on a different key
2387 // that must survive: ListMultipartUploads matches a PREFIX, so a careless
2388 // implementation reaping `staging/abc` would also kill `staging/abcdef`.
2389 mem.put_open_multipart("upload-a", target);
2390 mem.put_open_multipart("upload-b", target);
2391 mem.put_open_multipart("upload-c", "staging/abandoned-uuid/movie.mp4.other");
2392 assert_eq!(mem.open_multipart_count(), 3);
2393
2394 let aborted = abort_orphan_multipart_sessions(&mem, target).await;
2395
2396 assert_eq!(
2397 aborted, 2,
2398 "both sessions on the reaped key must be aborted"
2399 );
2400 assert_eq!(
2401 mem.open_multipart_count(),
2402 1,
2403 "a session on a different key must survive the prefix-adjacent reap"
2404 );
2405 }
2406
2407 #[tokio::test]
2408 async fn orphan_reaper_abort_is_a_noop_when_there_is_no_session() {
2409 // The common case: an ordinary single-PUT orphan has no multipart session,
2410 // and the reaper must sail past it without erroring or miscounting.
2411 use makenotwork::scheduler::abort_orphan_multipart_sessions;
2412
2413 let mem = crate::harness::storage::InMemoryStorage::new();
2414 mem.put("staging/plain/song.mp3", b"bytes".to_vec());
2415
2416 let aborted = abort_orphan_multipart_sessions(&mem, "staging/plain/song.mp3").await;
2417 assert_eq!(aborted, 0);
2418 }
2419