Skip to main content

max / makenotwork

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