max / makenotwork

83.6 KB · 2459 lines History Blame Raw

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