Skip to main content

max / makenotwork

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