//! Storage workflow tests, presign, confirm, stream, download, access control.

use crate::harness::TestHarness;
use makenotwork::storage::StorageBackend;
use serde_json::{Value, json};

/// Helper: create a trusted creator with a project and audio item. Returns (user_id, project_id, item_id).
async fn setup_creator_with_item(
    h: &mut TestHarness,
    price_cents: i64,
) -> (String, String, String) {
    let setup = h
        .create_creator_with_item("creator", "audio", price_cents)
        .await;
    h.trust_user(setup.user_id).await;
    h.grant_tier(setup.user_id, "small_files").await;
    (setup.user_id.to_string(), setup.project_id, setup.item_id)
}

// Presign

#[tokio::test]
async fn presign_upload_audio() {
    let mut h = TestHarness::with_storage().await;
    let (_, _, item_id) = setup_creator_with_item(&mut h, 0).await;

    let body = json!({
        "item_id": item_id,
        "file_type": "audio",
        "file_name": "episode.mp3",
        "content_type": "audio/mpeg",
    });
    let resp = h
        .client
        .post_json("/api/upload/presign", &body.to_string())
        .await;
    assert_eq!(resp.status, 200, "Presign failed: {}", resp.text);

    let data: Value = resp.json();
    assert!(
        data["upload_url"]
            .as_str()
            .unwrap()
            .starts_with("http://test-storage/")
    );
    // Scan-then-promote: presign hands out an unserved staging key
    // (`staging/{uuid}/{filename}`), never the served key. The served,
    // content-addressed key is minted by the scan worker on a Clean verdict.
    let key = data["s3_key"].as_str().unwrap();
    assert!(
        key.starts_with("staging/"),
        "presign must return a staging key: {key}"
    );
    assert!(
        key.ends_with("/episode.mp3"),
        "staging key preserves the filename: {key}"
    );
    assert_eq!(data["expires_in"], 3600);
}

// Confirm

#[tokio::test]
async fn confirm_upload_audio_updates_db() {
    let mut h = TestHarness::with_storage().await;
    let (_, _, item_id) = setup_creator_with_item(&mut h, 0).await;

    // Presign
    let body = json!({
        "item_id": item_id,
        "file_type": "audio",
        "file_name": "song.mp3",
        "content_type": "audio/mpeg",
    });
    let resp = h
        .client
        .post_json("/api/upload/presign", &body.to_string())
        .await;
    assert_eq!(resp.status, 200, "{}", resp.text);
    let data: Value = resp.json();
    let s3_key = data["s3_key"].as_str().unwrap().to_string();

    // Simulate the client uploading to S3
    h.storage
        .as_ref()
        .unwrap()
        .put(&s3_key, b"fake mp3 bytes".to_vec());

    // Confirm
    let body = json!({
        "item_id": item_id,
        "file_type": "audio",
        "s3_key": s3_key,
    });
    let resp = h
        .client
        .post_json("/api/upload/confirm", &body.to_string())
        .await;
    assert_eq!(resp.status, 200, "Confirm failed: {}", resp.text);
    let data: Value = resp.json();
    assert_eq!(data["success"], true);

    // Verify database
    let db_key: Option<String> =
        sqlx::query_scalar("SELECT audio_s3_key FROM items WHERE id = $1::uuid")
            .bind(&item_id)
            .fetch_one(&h.db)
            .await
            .unwrap();
    assert_eq!(db_key.as_deref(), Some(s3_key.as_str()));
}

#[tokio::test]
async fn internal_confirm_upload_replay_is_idempotent() {
    // A retried internal (CLI, ServiceAuth) confirm for the same deterministic key
    // must not double-charge storage (ultra-fuzz Run 12 Storage: confirm idempotency).
    let mem = std::sync::Arc::new(crate::harness::storage::InMemoryStorage::new());
    let mut h = TestHarness::build(crate::harness::BuildOptions {
        storage: Some(mem),
        cli_service_token: Some("test-cli-token".to_string()),
        ..Default::default()
    })
    .await;

    let setup = h.create_creator_with_item("cliuser", "audio", 0).await;
    h.trust_user(setup.user_id).await;
    h.grant_tier(setup.user_id, "small_files").await;
    let user_id = setup.user_id;
    let item_id = setup.item_id.clone();

    // Presign (session auth) then simulate the client PUT to S3.
    let presign = json!({
        "item_id": item_id, "file_type": "audio",
        "file_name": "song.mp3", "content_type": "audio/mpeg",
    });
    let resp = h
        .client
        .post_json("/api/upload/presign", &presign.to_string())
        .await;
    assert_eq!(resp.status, 200, "presign failed: {}", resp.text);
    let s3_key = resp.json::<Value>()["s3_key"].as_str().unwrap().to_string();
    let bytes = b"fake mp3 file bytes".to_vec();
    let expected_size = bytes.len() as i64;
    h.storage.as_ref().unwrap().put(&s3_key, bytes);

    // Internal confirm (ServiceAuth via bearer), twice, identical.
    let confirm = json!({
        "user_id": user_id, "item_id": item_id,
        "file_type": "audio", "s3_key": s3_key,
    });
    h.client.set_bearer_token("test-cli-token");
    let actor = makenotwork::crypto::mint_internal_actor_token(
        user_id,
        chrono::Utc::now().timestamp() + 3600,
        "test-signing-secret-for-integration-tests",
    );
    h.client.set_actor_token(&actor);
    let r1 = h
        .client
        .post_json("/api/internal/upload/confirm", &confirm.to_string())
        .await;
    assert_eq!(r1.status, 200, "first internal confirm failed: {}", r1.text);
    let r2 = h
        .client
        .post_json("/api/internal/upload/confirm", &confirm.to_string())
        .await;
    assert_eq!(
        r2.status, 200,
        "replayed internal confirm failed: {}",
        r2.text
    );
    h.client.clear_bearer_token();

    // The key is committed once and storage is charged exactly once, not twice.
    let db_key: Option<String> =
        sqlx::query_scalar("SELECT audio_s3_key FROM items WHERE id = $1::uuid")
            .bind(&item_id)
            .fetch_one(&h.db)
            .await
            .unwrap();
    assert_eq!(db_key.as_deref(), Some(s3_key.as_str()));

    let storage_used: i64 =
        sqlx::query_scalar("SELECT storage_used_bytes FROM users WHERE id = $1")
            .bind(user_id)
            .fetch_one(&h.db)
            .await
            .unwrap();
    assert_eq!(
        storage_used, expected_size,
        "replay must not double-charge storage"
    );
}

/// A valid ServiceAuth bearer alone is not enough to act on the internal API:
/// without a valid `X-MNW-Actor` assertion the request is rejected. Proves a
/// leaked service token cannot name an arbitrary user.
#[tokio::test]
async fn internal_confirm_without_actor_token_rejected() {
    let mem = std::sync::Arc::new(crate::harness::storage::InMemoryStorage::new());
    let mut h = TestHarness::build(crate::harness::BuildOptions {
        storage: Some(mem),
        cli_service_token: Some("test-cli-token".to_string()),
        ..Default::default()
    })
    .await;

    let setup = h.create_creator_with_item("noactor", "audio", 0).await;
    h.trust_user(setup.user_id).await;
    h.grant_tier(setup.user_id, "small_files").await;

    let confirm = json!({
        "user_id": setup.user_id, "item_id": setup.item_id,
        "file_type": "audio", "s3_key": "noactor/whatever.mp3",
    });
    // Bearer set, but no actor assertion.
    h.client.set_bearer_token("test-cli-token");
    let resp = h
        .client
        .post_json("/api/internal/upload/confirm", &confirm.to_string())
        .await;
    assert_eq!(
        resp.status.as_u16(),
        401,
        "missing actor assertion must be rejected: {}",
        resp.text
    );
    h.client.clear_bearer_token();
}

#[tokio::test]
async fn confirm_item_cover_via_dedicated_route_writes_key_and_url() {
    // Covers go through /api/items/image/{presign,confirm}, which writes
    // cover_s3_key, cover_file_size_bytes AND cover_image_url together. The
    // generic /api/upload/confirm used to accept cover and write the first two
    // but NOT the URL, leaving an invisible cover (Run #13 SERIOUS); it now
    // rejects cover (see confirm_upload_rejects_cover below).
    let mut h = TestHarness::with_storage().await;
    let (_, _, item_id) = setup_creator_with_item(&mut h, 0).await;

    let body = json!({
        "item_id": item_id,
        "file_name": "art.png",
        "content_type": "image/png",
    });
    let resp = h
        .client
        .post_json("/api/items/image/presign", &body.to_string())
        .await;
    assert_eq!(resp.status, 200, "presign failed: {}", resp.text);
    let data: Value = resp.json();
    let s3_key = data["s3_key"].as_str().unwrap().to_string();

    h.storage
        .as_ref()
        .unwrap()
        .put(&s3_key, b"fake png bytes".to_vec());

    let body = json!({ "item_id": item_id, "s3_key": s3_key });
    let resp = h
        .client
        .post_json("/api/items/image/confirm", &body.to_string())
        .await;
    assert_eq!(resp.status, 200, "Confirm failed: {}", resp.text);

    // Both the key AND the render URL must be set, the URL is what the bug missed.
    let (db_key, db_url): (Option<String>, Option<String>) =
        sqlx::query_as("SELECT cover_s3_key, cover_image_url FROM items WHERE id = $1::uuid")
            .bind(&item_id)
            .fetch_one(&h.db)
            .await
            .unwrap();
    assert_eq!(db_key.as_deref(), Some(s3_key.as_str()));
    assert!(
        db_url.is_some_and(|u| u.contains(&s3_key)),
        "cover_image_url must be set so the cover renders"
    );
}

#[tokio::test]
async fn cover_replace_does_not_take_published_track_offline() {
    // Run #20 Storage SERIOUS: a cover upload used to flip the SHARED
    // `items.scan_status` to Pending (cover shares the audio's row), and the
    // stream/download gate returns NotFound for non-creators until the cover
    // re-scan finishes, silently pulling an already-published track offline
    // for every fan. The cover is CDN-served with no per-request gate, so the
    // flip protects nothing and only harms the track. Confirming a cover must
    // leave `items.scan_status` untouched.
    let mut h = TestHarness::with_storage_and_scanner().await;
    let (_, _, item_id) = setup_creator_with_item(&mut h, 0).await;

    // Simulate the published state: track already scanned Clean.
    sqlx::query("UPDATE items SET scan_status = 'clean' WHERE id = $1::uuid")
        .bind(&item_id)
        .execute(&h.db)
        .await
        .unwrap();

    // Upload a new cover. With a scanner configured, the cover scan enqueues as
    // Pending; pre-fix that Pending was written straight onto items.scan_status
    // by the synchronous confirm.
    let body = json!({ "item_id": item_id, "file_name": "art.png", "content_type": "image/png" });
    let resp = h
        .client
        .post_json("/api/items/image/presign", &body.to_string())
        .await;
    assert_eq!(resp.status, 200, "presign failed: {}", resp.text);
    let s3_key = resp.json::<Value>()["s3_key"].as_str().unwrap().to_string();
    h.storage
        .as_ref()
        .unwrap()
        .put(&s3_key, b"fake png bytes".to_vec());

    let body = json!({ "item_id": item_id, "s3_key": s3_key });
    let resp = h
        .client
        .post_json("/api/items/image/confirm", &body.to_string())
        .await;
    assert_eq!(resp.status, 200, "cover confirm failed: {}", resp.text);

    // The track's gate status must still be Clean, the cover upload may not
    // touch it. (Assert synchronously, before draining the scan worker.)
    let scan_status: String =
        sqlx::query_scalar("SELECT scan_status FROM items WHERE id = $1::uuid")
            .bind(&item_id)
            .fetch_one(&h.db)
            .await
            .unwrap();
    assert_eq!(
        scan_status, "clean",
        "cover upload must not flip the track's scan_status (would take it offline for fans)"
    );

    // And the cover itself still landed.
    let cover_key: Option<String> =
        sqlx::query_scalar("SELECT cover_s3_key FROM items WHERE id = $1::uuid")
            .bind(&item_id)
            .fetch_one(&h.db)
            .await
            .unwrap();
    assert_eq!(cover_key.as_deref(), Some(s3_key.as_str()));
}

#[tokio::test]
async fn confirm_upload_rejects_cover() {
    // The generic confirm route must refuse cover and point at the dedicated
    // route, rather than half-writing the row (no cover_image_url). Run #13.
    let mut h = TestHarness::with_storage().await;
    let (_, _, item_id) = setup_creator_with_item(&mut h, 0).await;

    let body = json!({
        "item_id": item_id,
        "file_type": "cover",
        "file_name": "art.png",
        "content_type": "image/png",
    });
    let resp = h
        .client
        .post_json("/api/upload/presign", &body.to_string())
        .await;
    assert_eq!(resp.status, 200, "{}", resp.text);
    let data: Value = resp.json();
    let s3_key = data["s3_key"].as_str().unwrap().to_string();
    h.storage
        .as_ref()
        .unwrap()
        .put(&s3_key, b"fake png bytes".to_vec());

    let body = json!({ "item_id": item_id, "file_type": "cover", "s3_key": s3_key });
    let resp = h
        .client
        .post_json("/api/upload/confirm", &body.to_string())
        .await;
    assert_eq!(
        resp.status.as_u16(),
        400,
        "generic confirm must reject cover: {}",
        resp.text
    );
    assert!(
        resp.text.contains("/api/items/image/confirm"),
        "rejection should name the dedicated route: {}",
        resp.text
    );

    // The row must be untouched, no half-written cover key.
    let db_key: Option<String> =
        sqlx::query_scalar("SELECT cover_s3_key FROM items WHERE id = $1::uuid")
            .bind(&item_id)
            .fetch_one(&h.db)
            .await
            .unwrap();
    assert_eq!(
        db_key, None,
        "rejected cover confirm must not write cover_s3_key"
    );
}

// Versions

#[tokio::test]
async fn version_upload_and_download() {
    let mut h = TestHarness::with_storage().await;
    let (_, _, item_id) = setup_creator_with_item(&mut h, 0).await;

    // Create a version (digital item needs a version for downloads)
    let resp = h
        .client
        .post_json(
            &format!("/api/items/{item_id}/versions"),
            &json!({"version_number": "1.0.0"}).to_string(),
        )
        .await;
    assert_eq!(resp.status, 200, "Create version failed: {}", resp.text);
    let version: Value = resp.json();
    let version_id = version["id"].as_str().unwrap().to_string();

    // Presign version upload
    let resp = h
        .client
        .post_json(
            &format!("/api/versions/{version_id}/upload/presign"),
            &json!({
                "file_name": "plugin.zip",
                "content_type": "application/zip",
            })
            .to_string(),
        )
        .await;
    assert_eq!(resp.status, 200, "Version presign failed: {}", resp.text);
    let data: Value = resp.json();
    let s3_key = data["s3_key"].as_str().unwrap().to_string();

    // Simulate upload
    h.storage
        .as_ref()
        .unwrap()
        .put(&s3_key, b"fake zip data".to_vec());

    // Confirm version upload
    let resp = h
        .client
        .post_json(
            &format!("/api/versions/{version_id}/upload/confirm"),
            &json!({"s3_key": s3_key}).to_string(),
        )
        .await;
    assert_eq!(resp.status, 200, "Version confirm failed: {}", resp.text);

    // Publish item + project so download works
    h.client
        .put_form(&format!("/api/items/{item_id}"), "is_public=true")
        .await;
    let project_id: String =
        sqlx::query_scalar("SELECT project_id::text FROM items WHERE id = $1::uuid")
            .bind(&item_id)
            .fetch_one(&h.db)
            .await
            .unwrap();
    h.client
        .put_json(
            &format!("/api/projects/{project_id}"),
            r#"{"is_public": true}"#,
        )
        .await;

    // Download version
    let resp = h
        .client
        .get(&format!("/api/versions/{version_id}/download"))
        .await;
    // 303 to the presigned URL, not JSON describing it (`8fc6b1af`, option (a)).
    assert_eq!(
        resp.status, 303,
        "Version download should redirect: {}",
        resp.text
    );
    let location = resp
        .headers
        .get("location")
        .expect("303 carries a Location")
        .to_str()
        .unwrap();
    assert!(
        location.starts_with("http://test-storage/"),
        "Location should be the presigned URL, got: {location}"
    );
}

// Audio Streaming

#[tokio::test]
async fn stream_url_free_item() {
    let mut h = TestHarness::with_storage().await;
    let (_, project_id, item_id) = setup_creator_with_item(&mut h, 0).await;

    // Set up audio key directly in DB (simulates a completed upload)
    let s3_key = format!("test/{item_id}/audio/track.mp3");
    sqlx::query("UPDATE items SET audio_s3_key = $1, scan_status = 'clean' WHERE id = $2::uuid")
        .bind(&s3_key)
        .bind(&item_id)
        .execute(&h.db)
        .await
        .unwrap();

    // Pre-populate storage
    h.storage
        .as_ref()
        .unwrap()
        .put(&s3_key, b"audio data".to_vec());

    // Publish
    h.client
        .put_form(&format!("/api/items/{item_id}"), "is_public=true")
        .await;
    h.client
        .put_json(
            &format!("/api/projects/{project_id}"),
            r#"{"is_public": true}"#,
        )
        .await;

    // Stream, free item, any user can access
    let resp = h.client.get(&format!("/api/stream/{item_id}")).await;
    assert_eq!(resp.status, 200, "Stream failed: {}", resp.text);
    let data: Value = resp.json();
    assert!(
        data["stream_url"]
            .as_str()
            .unwrap()
            .starts_with("http://test-storage/")
    );
}

#[tokio::test]
async fn stream_url_paid_requires_purchase() {
    let mut h = TestHarness::with_storage().await;
    let (_, project_id, item_id) = setup_creator_with_item(&mut h, 500).await;

    // Set up audio key
    let s3_key = format!("test/{item_id}/audio/track.mp3");
    sqlx::query("UPDATE items SET audio_s3_key = $1, scan_status = 'clean' WHERE id = $2::uuid")
        .bind(&s3_key)
        .bind(&item_id)
        .execute(&h.db)
        .await
        .unwrap();

    h.storage
        .as_ref()
        .unwrap()
        .put(&s3_key, b"audio data".to_vec());

    // Publish
    h.client
        .put_form(&format!("/api/items/{item_id}"), "is_public=true")
        .await;
    h.client
        .put_json(
            &format!("/api/projects/{project_id}"),
            r#"{"is_public": true}"#,
        )
        .await;

    // Log out the creator and sign up a buyer with no purchase
    h.client.post_form("/logout", "").await;
    h.signup("buyer", "buyer@test.com", "password123").await;
    h.login("buyer", "password123").await;

    // Stream should be forbidden (paid, no purchase)
    let resp = h.client.get(&format!("/api/stream/{item_id}")).await;
    assert_eq!(
        resp.status.as_u16(),
        403,
        "Expected 403, got: {}",
        resp.text
    );
}

/// Helper: set up a paid item with audio, publish it, return (creator_user_id, project_id, item_id, s3_key).
/// Leaves the creator logged in.
async fn setup_published_paid_audio(
    h: &mut TestHarness,
    username: &str,
    price_cents: i64,
) -> (String, String, String, String) {
    let setup = h
        .create_creator_with_item(username, "audio", price_cents)
        .await;
    h.trust_user(setup.user_id).await;
    h.grant_tier(setup.user_id, "small_files").await;
    let user_id = setup.user_id.to_string();

    let s3_key = format!("test/{}/audio/track.mp3", setup.item_id);
    sqlx::query("UPDATE items SET audio_s3_key = $1, scan_status = 'clean' WHERE id = $2::uuid")
        .bind(&s3_key)
        .bind(&setup.item_id)
        .execute(&h.db)
        .await
        .unwrap();
    h.storage
        .as_ref()
        .unwrap()
        .put(&s3_key, b"audio data".to_vec());

    h.publish_project_and_item(&setup.project_id, &setup.item_id)
        .await;

    (user_id, setup.project_id, setup.item_id, s3_key)
}

// Stream access control (test-fuzz)

/// Unauthenticated user gets 401 on paid item stream.
#[tokio::test]
async fn stream_url_paid_unauthenticated_returns_401() {
    let mut h = TestHarness::with_storage().await;
    let (_, _, item_id, _) = setup_published_paid_audio(&mut h, "seller401", 500).await;

    // Log out, no session
    h.client.post_form("/logout", "").await;

    let resp = h.client.get(&format!("/api/stream/{item_id}")).await;
    assert_eq!(
        resp.status.as_u16(),
        401,
        "Unauthenticated stream of paid item should be 401, got: {}",
        resp.status
    );
}

/// After a direct DB purchase record, buyer can stream paid content.
#[tokio::test]
async fn stream_url_paid_purchaser_gets_access() {
    let mut h = TestHarness::with_storage().await;
    let (creator_id, _, item_id, _) = setup_published_paid_audio(&mut h, "sellaccess", 999).await;

    // Create buyer and insert a completed transaction (simulates webhook completion)
    h.client.post_form("/logout", "").await;
    let buyer_id = h
        .signup("buyaccess", "buyaccess@test.com", "password123")
        .await;

    sqlx::query(
        r"INSERT INTO transactions
           (buyer_id, seller_id, item_id, amount_cents, status,
            stripe_checkout_session_id, item_title, seller_username,
            completed_at)
           VALUES ($1, $2::uuid, $3::uuid, 999, 'completed',
                   'cs_access_test', 'Track', 'sellaccess', NOW())",
    )
    .bind(buyer_id)
    .bind(&creator_id)
    .bind(&item_id)
    .execute(&h.db)
    .await
    .unwrap();

    h.login("buyaccess", "password123").await;

    let resp = h.client.get(&format!("/api/stream/{item_id}")).await;
    assert_eq!(
        resp.status, 200,
        "Purchaser should be able to stream paid item, got: {} {}",
        resp.status, resp.text
    );
    let data: Value = resp.json();
    assert!(
        data["stream_url"].as_str().is_some(),
        "Response should contain stream_url"
    );
}

/// Creator can always stream their own paid content.
#[tokio::test]
async fn stream_url_creator_always_has_access() {
    let mut h = TestHarness::with_storage().await;
    let (_, _, item_id, _) = setup_published_paid_audio(&mut h, "selfstream", 999).await;

    // Creator is still logged in
    let resp = h.client.get(&format!("/api/stream/{item_id}")).await;
    assert_eq!(
        resp.status, 200,
        "Creator should stream their own paid item, got: {} {}",
        resp.status, resp.text
    );
}

/// Unpublished (draft) item returns 404 for non-owner.
#[tokio::test]
async fn stream_url_draft_item_404_for_non_owner() {
    let mut h = TestHarness::with_storage().await;
    let setup = h.create_creator_with_item("draftowner", "audio", 0).await;
    h.trust_user(setup.user_id).await;
    h.grant_tier(setup.user_id, "small_files").await;

    let s3_key = format!("test/{}/audio/draft.mp3", setup.item_id);
    sqlx::query("UPDATE items SET audio_s3_key = $1, scan_status = 'clean' WHERE id = $2::uuid")
        .bind(&s3_key)
        .bind(&setup.item_id)
        .execute(&h.db)
        .await
        .unwrap();
    h.storage
        .as_ref()
        .unwrap()
        .put(&s3_key, b"audio data".to_vec());

    // Explicitly unpublish the item (items default to is_public=true)
    h.client
        .put_form(&format!("/api/items/{}", setup.item_id), "is_public=false")
        .await;

    h.client.post_form("/logout", "").await;
    h.signup("snooper", "snooper@test.com", "password123").await;
    h.login("snooper", "password123").await;

    let resp = h
        .client
        .get(&format!("/api/stream/{}", setup.item_id))
        .await;
    assert_eq!(
        resp.status.as_u16(),
        404,
        "Draft item should be 404 for non-owner, got: {}",
        resp.status
    );
}

/// Version download for paid item: non-purchaser gets 403.
#[tokio::test]
async fn version_download_paid_non_purchaser_forbidden() {
    let mut h = TestHarness::with_storage().await;
    let setup = h
        .create_creator_with_item("verseller", "digital", 500)
        .await;
    h.trust_user(setup.user_id).await;
    h.grant_tier(setup.user_id, "small_files").await;

    // Create version with file
    let resp = h
        .client
        .post_json(
            &format!("/api/items/{}/versions", setup.item_id),
            &json!({"version_number": "1.0.0"}).to_string(),
        )
        .await;
    assert_eq!(resp.status, 200, "Create version failed: {}", resp.text);
    let version: Value = resp.json();
    let version_id = version["id"].as_str().unwrap().to_string();

    let resp = h
        .client
        .post_json(
            &format!("/api/versions/{version_id}/upload/presign"),
            &json!({"file_name": "app.zip", "content_type": "application/zip"}).to_string(),
        )
        .await;
    assert_eq!(resp.status, 200, "{}", resp.text);
    let data: Value = resp.json();
    let s3_key = data["s3_key"].as_str().unwrap().to_string();
    h.storage
        .as_ref()
        .unwrap()
        .put(&s3_key, b"zip data".to_vec());

    h.client
        .post_json(
            &format!("/api/versions/{version_id}/upload/confirm"),
            &json!({"s3_key": s3_key}).to_string(),
        )
        .await;

    // Publish
    h.publish_project_and_item(&setup.project_id, &setup.item_id)
        .await;

    // Non-purchaser tries to download
    h.client.post_form("/logout", "").await;
    h.signup("verbuyer", "verbuyer@test.com", "password123")
        .await;
    h.login("verbuyer", "password123").await;

    let resp = h
        .client
        .get(&format!("/api/versions/{version_id}/download"))
        .await;
    assert_eq!(
        resp.status.as_u16(),
        403,
        "Non-purchaser version download should be 403, got: {}",
        resp.status
    );
}

// Access control

#[tokio::test]
async fn upload_non_owner_forbidden() {
    let mut h = TestHarness::with_storage().await;
    let (_, _, item_id) = setup_creator_with_item(&mut h, 0).await;

    // Log out creator, sign up a different user
    h.client.post_form("/logout", "").await;
    h.signup("intruder", "intruder@test.com", "password123")
        .await;
    h.login("intruder", "password123").await;

    // Attempt to presign to creator's item
    let body = json!({
        "item_id": item_id,
        "file_type": "audio",
        "file_name": "evil.mp3",
        "content_type": "audio/mpeg",
    });
    let resp = h
        .client
        .post_json("/api/upload/presign", &body.to_string())
        .await;
    assert_eq!(
        resp.status.as_u16(),
        403,
        "Expected 403, got: {}",
        resp.text
    );
}

// Confirm-handler failure & rollback contract (test-fuzz Phase 2.2)
//
// The Run #9 tx port made the confirm handlers charge storage inside a
// transaction and route orphaned keys through the deletion queue. These pin the
// observable contract that work protects: a FAILED confirm must never inflate
// the storage counter and never leak the S3 object, and the deterministic
// reachable tx paths (replace storage-math + old-key orphan enqueue) must hold.
//
// NOTE on the pure lost-race rollback branches (uploads `Ok(0)`, versions
// `Ok(false)`, and the in-tx `Err`): these fire only when a second confirm, or
// an item/version delete, interleaves inside the handler's read-then-write
// window. `try_apply_storage_on` takes the users-row lock, which serializes
// concurrent confirms, so the branch is genuinely a TOCTOU guard. It is not
// reachable from a single sequential request (the test client is cookie-bound
// and `&mut`, with no exposed app handle for a concurrent same-user pair), so it
// is left to the lower-level race coverage; the tests below exercise the same
// transaction body and the same orphan-queue helper on their reachable paths.

/// SMALL_FILES tier storage cap, in bytes (250 GiB). Mirrors
/// `CreatorTier::SmallFiles.max_storage_bytes()`.
const SMALL_FILES_CAP: i64 = 250 * 1024 * 1024 * 1024;

async fn presign_audio(h: &mut TestHarness, item_id: &str, file_name: &str) -> String {
    let body = json!({
        "item_id": item_id,
        "file_type": "audio",
        "file_name": file_name,
        "content_type": "audio/mpeg",
    });
    let resp = h
        .client
        .post_json("/api/upload/presign", &body.to_string())
        .await;
    assert_eq!(resp.status, 200, "presign failed: {}", resp.text);
    let data: Value = resp.json();
    data["s3_key"].as_str().unwrap().to_string()
}

async fn presign_version(h: &mut TestHarness, version_id: &str, file_name: &str) -> String {
    let resp = h
        .client
        .post_json(
            &format!("/api/versions/{version_id}/upload/presign"),
            &json!({"file_name": file_name, "content_type": "application/zip"}).to_string(),
        )
        .await;
    assert_eq!(resp.status, 200, "version presign failed: {}", resp.text);
    let data: Value = resp.json();
    data["s3_key"].as_str().unwrap().to_string()
}

async fn storage_used(h: &TestHarness, user_id: &str) -> i64 {
    sqlx::query_scalar("SELECT storage_used_bytes FROM users WHERE id = $1::uuid")
        .bind(user_id)
        .fetch_one(&h.db)
        .await
        .unwrap()
}

#[tokio::test]
async fn confirm_over_storage_cap_does_not_charge_or_leak() {
    let mut h = TestHarness::with_storage().await;
    let (user_id, _project_id, item_id) = setup_creator_with_item(&mut h, 0).await;

    let s3_key = presign_audio(&mut h, &item_id, "big.mp3").await;
    h.storage.as_ref().unwrap().put(&s3_key, vec![0u8; 100]);

    // Park the counter one increment shy of the cap so this 100-byte file pushes over.
    let parked = SMALL_FILES_CAP - 50;
    sqlx::query("UPDATE users SET storage_used_bytes = $2 WHERE id = $1::uuid")
        .bind(&user_id)
        .bind(parked)
        .execute(&h.db)
        .await
        .unwrap();

    let body = json!({"item_id": item_id, "file_type": "audio", "s3_key": s3_key});
    let resp = h
        .client
        .post_json("/api/upload/confirm", &body.to_string())
        .await;
    assert_eq!(
        resp.status, 400,
        "over-cap confirm must fail, got: {} {}",
        resp.status, resp.text
    );

    // Counter unchanged, a failed confirm never inflates storage.
    assert_eq!(
        storage_used(&h, &user_id).await,
        parked,
        "failed confirm must not charge storage"
    );
    // The rejected confirm enqueues the orphan for deletion; run the queue (the
    // scheduler's job in production) before asserting the object is gone.
    h.drain_s3_deletions().await;
    // And the object was cleaned up, not leaked.
    assert!(
        !h.storage
            .as_ref()
            .unwrap()
            .object_exists(&s3_key)
            .await
            .unwrap(),
        "over-cap confirm must delete the orphaned object"
    );
    // The item never picked up the key.
    let db_key: Option<String> =
        sqlx::query_scalar("SELECT audio_s3_key FROM items WHERE id = $1::uuid")
            .bind(&item_id)
            .fetch_one(&h.db)
            .await
            .unwrap();
    assert_eq!(
        db_key, None,
        "item must not reference a key from a failed confirm"
    );
}

#[tokio::test]
async fn confirm_wrong_route_file_type_deletes_object_and_does_not_charge() {
    let mut h = TestHarness::with_storage().await;
    let (user_id, _project_id, item_id) = setup_creator_with_item(&mut h, 0).await;

    // Presign as audio (a key under user/item/), but confirm it as a "download",
    // download has its own /api/versions route, so the item-upload confirm must
    // reject it, delete the object, and charge nothing.
    let s3_key = presign_audio(&mut h, &item_id, "song.mp3").await;
    h.storage.as_ref().unwrap().put(&s3_key, vec![0u8; 500]);

    let body = json!({"item_id": item_id, "file_type": "download", "s3_key": s3_key});
    let resp = h
        .client
        .post_json("/api/upload/confirm", &body.to_string())
        .await;
    assert_eq!(
        resp.status.as_u16(),
        400,
        "misrouted file type must 400, got: {} {}",
        resp.status,
        resp.text
    );

    assert_eq!(
        storage_used(&h, &user_id).await,
        0,
        "misrouted confirm must charge nothing"
    );
    // Run the orphan-deletion queue (scheduler's job in production) before the assert.
    h.drain_s3_deletions().await;
    assert!(
        !h.storage
            .as_ref()
            .unwrap()
            .object_exists(&s3_key)
            .await
            .unwrap(),
        "misrouted confirm must delete the object (it would otherwise leak, the scan_jobs/scan_status footgun the guard prevents)"
    );
}

#[tokio::test]
async fn confirm_replace_charges_delta_and_orphans_old_key() {
    let mut h = TestHarness::with_storage().await;
    let (user_id, _project_id, item_id) = setup_creator_with_item(&mut h, 0).await;

    // First upload: 1000 bytes.
    let key1 = presign_audio(&mut h, &item_id, "v1.mp3").await;
    h.storage.as_ref().unwrap().put(&key1, vec![0u8; 1000]);
    let body = json!({"item_id": item_id, "file_type": "audio", "s3_key": key1});
    let resp = h
        .client
        .post_json("/api/upload/confirm", &body.to_string())
        .await;
    assert_eq!(resp.status, 200, "first confirm failed: {}", resp.text);
    assert_eq!(
        storage_used(&h, &user_id).await,
        1000,
        "first upload charges its full size"
    );

    // Replace with a 300-byte upload.
    let key2 = presign_audio(&mut h, &item_id, "v2.mp3").await;
    h.storage.as_ref().unwrap().put(&key2, vec![0u8; 300]);
    let body = json!({"item_id": item_id, "file_type": "audio", "s3_key": key2});
    let resp = h
        .client
        .post_json("/api/upload/confirm", &body.to_string())
        .await;
    assert_eq!(resp.status, 200, "replace confirm failed: {}", resp.text);

    // The item now points at the new key.
    let db_key: Option<String> =
        sqlx::query_scalar("SELECT audio_s3_key FROM items WHERE id = $1::uuid")
            .bind(&item_id)
            .fetch_one(&h.db)
            .await
            .unwrap();
    assert_eq!(db_key.as_deref(), Some(key2.as_str()));

    // Storage reflects the DELTA, not a double-charge: 1000 - 1000 + 300 = 300.
    // This is the in-tx try_replace_storage_on path executing for real.
    assert_eq!(
        storage_used(&h, &user_id).await,
        300,
        "replace must apply the size delta, not stack"
    );

    // The OLD key is routed through the deletion queue (not deleted inline) so a
    // transient S3 failure can't leak it, the same orphan-queue the lost-race
    // path uses.
    let queued: i64 = sqlx::query_scalar(
        "SELECT COUNT(*) FROM pending_s3_deletions WHERE s3_key = $1 AND source = 'item_upload_replace'",
    )
    .bind(&key1)
    .fetch_one(&h.db)
    .await
    .unwrap();
    assert_eq!(
        queued, 1,
        "old key must be enqueued for deletion on replace"
    );
    // It's still present in S3 right now, the worker deletes it later.
    assert!(
        h.storage
            .as_ref()
            .unwrap()
            .object_exists(&key1)
            .await
            .unwrap(),
        "old key is queued, not deleted inline"
    );
}

#[tokio::test]
async fn version_confirm_replace_enqueues_old_key_and_charges_delta() {
    let mut h = TestHarness::with_storage().await;
    let (user_id, _project_id, item_id) = setup_creator_with_item(&mut h, 0).await;

    let resp = h
        .client
        .post_json(
            &format!("/api/items/{item_id}/versions"),
            &json!({"version_number": "1.0.0"}).to_string(),
        )
        .await;
    assert_eq!(resp.status, 200, "create version failed: {}", resp.text);
    let version: Value = resp.json();
    let version_id = version["id"].as_str().unwrap().to_string();

    // First version file: 2000 bytes.
    let key1 = presign_version(&mut h, &version_id, "v1.zip").await;
    h.storage.as_ref().unwrap().put(&key1, vec![0u8; 2000]);
    let resp = h
        .client
        .post_json(
            &format!("/api/versions/{version_id}/upload/confirm"),
            &json!({"s3_key": key1}).to_string(),
        )
        .await;
    assert_eq!(
        resp.status, 200,
        "first version confirm failed: {}",
        resp.text
    );
    assert_eq!(storage_used(&h, &user_id).await, 2000);

    // Replace with 600 bytes.
    let key2 = presign_version(&mut h, &version_id, "v2.zip").await;
    h.storage.as_ref().unwrap().put(&key2, vec![0u8; 600]);
    let resp = h
        .client
        .post_json(
            &format!("/api/versions/{version_id}/upload/confirm"),
            &json!({"s3_key": key2}).to_string(),
        )
        .await;
    assert_eq!(
        resp.status, 200,
        "version replace confirm failed: {}",
        resp.text
    );

    assert_eq!(
        storage_used(&h, &user_id).await,
        600,
        "version replace must apply the delta"
    );
    let queued: i64 = sqlx::query_scalar(
        "SELECT COUNT(*) FROM pending_s3_deletions WHERE s3_key = $1 AND source = 'version_replace'",
    )
    .bind(&key1)
    .fetch_one(&h.db)
    .await
    .unwrap();
    assert_eq!(
        queued, 1,
        "old version key must be enqueued for deletion on replace"
    );
}

#[tokio::test]
async fn confirm_idempotent_reconfirm_does_not_double_charge() {
    let mut h = TestHarness::with_storage().await;
    let (user_id, _project_id, item_id) = setup_creator_with_item(&mut h, 0).await;

    let s3_key = presign_audio(&mut h, &item_id, "track.mp3").await;
    h.storage.as_ref().unwrap().put(&s3_key, vec![0u8; 500]);
    let body = json!({"item_id": item_id, "file_type": "audio", "s3_key": s3_key});

    let resp = h
        .client
        .post_json("/api/upload/confirm", &body.to_string())
        .await;
    assert_eq!(resp.status, 200, "first confirm failed: {}", resp.text);
    assert_eq!(storage_used(&h, &user_id).await, 500);
    // pending_uploads cleared on confirm (Run #7 HIGH-1: otherwise the reaper
    // deletes the live object 24h later).
    let pending_after_first: i64 =
        sqlx::query_scalar("SELECT COUNT(*) FROM pending_uploads WHERE s3_key = $1")
            .bind(&s3_key)
            .fetch_one(&h.db)
            .await
            .unwrap();
    assert_eq!(
        pending_after_first, 0,
        "confirm must clear the pending_uploads row"
    );

    // Re-confirm the SAME key: idempotent success, no second charge.
    let resp = h
        .client
        .post_json("/api/upload/confirm", &body.to_string())
        .await;
    assert_eq!(
        resp.status, 200,
        "idempotent re-confirm should succeed: {}",
        resp.text
    );
    assert_eq!(
        storage_used(&h, &user_id).await,
        500,
        "idempotent re-confirm must NOT double-charge storage"
    );
    assert!(
        h.storage
            .as_ref()
            .unwrap()
            .object_exists(&s3_key)
            .await
            .unwrap(),
        "idempotent re-confirm must not delete the live object"
    );
}

// CAS guard pins (test-fuzz Phase 2.2)
//
// The Run #9 tx port sealed every confirm-upload write behind an
// `IS NOT DISTINCT FROM expected_old` compare-and-swap so a confirm that lost a
// concurrent race (or whose target row was deleted/transferred mid-flight)
// matches zero rows and the surrounding tx rolls back, never double-crediting
// storage and never clobbering the live object the winning confirm published.
//
// The handler-level rollback + orphan-queue wiring on that branch can only fire
// under TRUE concurrency (a sequential request always observes its own read, so
// its CAS always matches, see the comment at uploads.rs around the confirm tx).
// What IS deterministic, and what these pin, is the CAS predicate itself: feed
// the db function a stale `expected_old` and assert it (a) reports the lost-race
// outcome and (b) leaves the row untouched. A regression that dropped the guard
// would make the stale write land here.

#[tokio::test]
async fn update_item_file_cas_guards_against_stale_confirm() {
    use makenotwork::db::items::{FileConfirmOutcome, update_item_file_cas};
    use makenotwork::db::{ItemId, UserId};
    use makenotwork::storage::FileType;

    let mut h = TestHarness::with_storage().await;
    let (user_id, _project_id, item_id) = setup_creator_with_item(&mut h, 0).await;
    let item = ItemId::from(uuid::Uuid::parse_str(&item_id).unwrap());
    let owner = UserId::from(uuid::Uuid::parse_str(&user_id).unwrap());

    // audio_s3_key is NULL, so the first confirm (expected_old = None) wins.
    let r = update_item_file_cas(&h.db, item, owner, FileType::Audio, None, "key_v1", 1000)
        .await
        .unwrap();
    assert!(matches!(r, FileConfirmOutcome::Committed));

    // A second confirm that still observed the pre-state (None) loses: the row
    // now holds key_v1, so `IS NOT DISTINCT FROM NULL` matches zero rows.
    let r = update_item_file_cas(&h.db, item, owner, FileType::Audio, None, "key_v2", 2000)
        .await
        .unwrap();
    assert!(
        matches!(r, FileConfirmOutcome::LostRace),
        "stale-None confirm must lose the CAS race"
    );

    // The loser left the row untouched, key_v1/1000, not key_v2 (no clobber, no double-credit).
    let (k, sz): (Option<String>, Option<i64>) =
        sqlx::query_as("SELECT audio_s3_key, audio_file_size_bytes FROM items WHERE id = $1::uuid")
            .bind(&item_id)
            .fetch_one(&h.db)
            .await
            .unwrap();
    assert_eq!(k.as_deref(), Some("key_v1"));
    assert_eq!(sz, Some(1000));

    // A correct CAS (expected_old = the current key) commits the swap.
    let r = update_item_file_cas(
        &h.db,
        item,
        owner,
        FileType::Audio,
        Some("key_v1"),
        "key_v3",
        3000,
    )
    .await
    .unwrap();
    assert!(matches!(r, FileConfirmOutcome::Committed));

    // The ownership filter is part of the same predicate: a non-owner never
    // matches, even with the correct expected_old.
    let stranger = UserId::new();
    let r = update_item_file_cas(
        &h.db,
        item,
        stranger,
        FileType::Audio,
        Some("key_v3"),
        "key_v4",
        4000,
    )
    .await
    .unwrap();
    assert!(
        matches!(r, FileConfirmOutcome::LostRace),
        "a non-owner must not write the item file"
    );
    let k: Option<String> =
        sqlx::query_scalar("SELECT audio_s3_key FROM items WHERE id = $1::uuid")
            .bind(&item_id)
            .fetch_one(&h.db)
            .await
            .unwrap();
    assert_eq!(
        k.as_deref(),
        Some("key_v3"),
        "the non-owner write must not land"
    );
}

#[tokio::test]
async fn update_version_file_guards_against_stale_confirm() {
    use makenotwork::db::VersionId;
    use makenotwork::db::versions::update_version_file;

    let mut h = TestHarness::with_storage().await;
    let (_user, _proj, item_id) = setup_creator_with_item(&mut h, 0).await;

    let resp = h
        .client
        .post_json(
            &format!("/api/items/{item_id}/versions"),
            &json!({"version_number": "1.0.0"}).to_string(),
        )
        .await;
    assert_eq!(resp.status, 200, "create version failed: {}", resp.text);
    let v: Value = resp.json();
    let version_id = v["id"].as_str().unwrap().to_string();
    let vid = VersionId::from(uuid::Uuid::parse_str(&version_id).unwrap());

    // s3_key is NULL initially. A confirm carrying a stale non-null expected key
    // matches zero rows.
    let r = update_version_file(
        &h.db,
        vid,
        Some("not_the_current_key"),
        "vk_v1",
        Some(1000),
        Some("a.zip"),
    )
    .await
    .unwrap();
    assert!(r.is_none(), "a stale expected_old key must match zero rows");
    let k: Option<String> = sqlx::query_scalar("SELECT s3_key FROM versions WHERE id = $1::uuid")
        .bind(&version_id)
        .fetch_one(&h.db)
        .await
        .unwrap();
    assert!(k.is_none(), "the row must stay unwritten after a lost CAS");

    // Correct CAS (expected None) commits.
    let r = update_version_file(&h.db, vid, None, "vk_v1", Some(1000), Some("a.zip"))
        .await
        .unwrap();
    assert!(r.is_some());

    // A replace that still observed the pre-state (None) loses to the committed key.
    let r = update_version_file(&h.db, vid, None, "vk_v2", Some(2000), Some("b.zip"))
        .await
        .unwrap();
    assert!(
        r.is_none(),
        "stale-None replace must lose once the key is set"
    );
    let (k, sz): (Option<String>, Option<i64>) =
        sqlx::query_as("SELECT s3_key, file_size_bytes FROM versions WHERE id = $1::uuid")
            .bind(&version_id)
            .fetch_one(&h.db)
            .await
            .unwrap();
    assert_eq!(k.as_deref(), Some("vk_v1"));
    assert_eq!(sz, Some(1000));
}

#[tokio::test]
async fn update_project_cover_cas_guards_against_stale_and_non_owner() {
    use makenotwork::db::projects::update_project_cover_cas;
    use makenotwork::db::{ProjectId, UserId};

    let mut h = TestHarness::with_storage().await;
    let (user_id, project_id, _item) = setup_creator_with_item(&mut h, 0).await;
    let pid = ProjectId::from(uuid::Uuid::parse_str(&project_id).unwrap());
    let owner = UserId::from(uuid::Uuid::parse_str(&user_id).unwrap());

    // cover_image_url is NULL. A stale expected url matches zero rows.
    let ok = update_project_cover_cas(
        &h.db,
        pid,
        owner,
        Some("stale_url"),
        "url_v1",
        "key_v1",
        1000,
    )
    .await
    .unwrap();
    assert!(!ok, "a stale expected url must match zero rows");

    // Correct CAS (expected None) commits.
    let ok = update_project_cover_cas(&h.db, pid, owner, None, "url_v1", "key_v1", 1000)
        .await
        .unwrap();
    assert!(ok);

    // Stale-None loses now that the cover is set.
    let ok = update_project_cover_cas(&h.db, pid, owner, None, "url_v2", "key_v2", 2000)
        .await
        .unwrap();
    assert!(!ok, "stale-None confirm must lose once the cover is set");

    // A non-owner cannot write, even with the correct expected url.
    let stranger = UserId::new();
    let ok = update_project_cover_cas(
        &h.db,
        pid,
        stranger,
        Some("url_v1"),
        "url_v3",
        "key_v3",
        3000,
    )
    .await
    .unwrap();
    assert!(!ok, "a non-owner must not write the project cover");

    let (url, key, sz): (Option<String>, Option<String>, Option<i64>) = sqlx::query_as(
        "SELECT cover_image_url, cover_s3_key, cover_image_size_bytes FROM projects WHERE id = $1::uuid",
    )
    .bind(&project_id)
    .fetch_one(&h.db)
    .await
    .unwrap();
    assert_eq!(url.as_deref(), Some("url_v1"));
    assert_eq!(
        key.as_deref(),
        Some("key_v1"),
        "the bare cover key is persisted alongside the url"
    );
    assert_eq!(sz, Some(1000));
}

#[tokio::test]
async fn update_item_cover_guards_against_stale_confirm() {
    use makenotwork::db::items::update_item_cover;
    use makenotwork::db::{ItemId, UserId};

    let mut h = TestHarness::with_storage().await;
    let (user_id, _proj, item_id) = setup_creator_with_item(&mut h, 0).await;
    let item = ItemId::from(uuid::Uuid::parse_str(&item_id).unwrap());
    let owner = UserId::from(uuid::Uuid::parse_str(&user_id).unwrap());

    // cover_s3_key is NULL. A stale expected key matches zero rows.
    let ok = update_item_cover(&h.db, item, owner, Some("stale_key"), "u1", "ck_v1", 1000)
        .await
        .unwrap();
    assert!(!ok, "a stale expected key must match zero rows");

    // Correct CAS (expected None) commits the cover key + url + size together.
    let ok = update_item_cover(&h.db, item, owner, None, "u1", "ck_v1", 1000)
        .await
        .unwrap();
    assert!(ok);

    // Stale-None loses now that the cover key is set.
    let ok = update_item_cover(&h.db, item, owner, None, "u2", "ck_v2", 2000)
        .await
        .unwrap();
    assert!(
        !ok,
        "stale-None confirm must lose once the cover key is set"
    );

    // A non-owner cannot write, even with the correct expected key.
    let stranger = UserId::new();
    let ok = update_item_cover(&h.db, item, stranger, Some("ck_v1"), "u3", "ck_v3", 3000)
        .await
        .unwrap();
    assert!(!ok, "a non-owner must not write the item cover");

    let (key, url, sz): (Option<String>, Option<String>, Option<i64>) = sqlx::query_as(
        "SELECT cover_s3_key, cover_image_url, cover_file_size_bytes FROM items WHERE id = $1::uuid",
    )
    .bind(&item_id)
    .fetch_one(&h.db)
    .await
    .unwrap();
    assert_eq!(key.as_deref(), Some("ck_v1"));
    assert_eq!(url.as_deref(), Some("u1"));
    assert_eq!(sz, Some(1000));
}

// Confirm failure-branch coverage via DB fault injection (test-fuzz Phase 2.2+)
//
// The CAS-guard pins above prove the predicate REJECTS a stale write. They do
// NOT exercise the handler's REACTION to a rejection, the storage-credit
// rollback (the credit and the CAS run in one tx; the handler returns without
// committing) and the orphan-enqueue of the staged key. Those branches only
// fire under a true concurrent confirm or a mid-tx DB error, neither of which a
// sequential test drives directly.
//
// We inject the fault at the DB layer with BEFORE UPDATE triggers keyed on a
// sentinel embedded in the uploaded key: a "faultlr" marker makes the guarded
// UPDATE affect zero rows (RETURN NULL), the lost-race branch (Ok(false)/Ok(0));
// a "faulterr" marker makes it RAISE, the commit-Err branch. Both are fully
// deterministic, need no concurrency, and are scoped to this test's cloned DB,
// so no production code carries a test hook and no other test is affected (the
// triggers are inert for any key without the marker).
//
// Each test asserts the two safety properties of the failed confirm: storage is
// NOT credited (the tx rolled back) and the staged S3 key is orphan-enqueued
// with the handler's reason string (so the reaper cleans it and a blind delete
// never clobbers a live object).

/// Install confirm-time fault triggers on items/versions/projects in this test's
/// cloned DB. A key containing `faultlr` skips the guarded UPDATE (zero rows ->
/// lost-race branch); a key containing `faulterr` raises (commit-Err branch).
async fn install_confirm_fault_triggers(pool: &sqlx::PgPool) {
    let stmts = [
        r"CREATE OR REPLACE FUNCTION test_confirm_fault_items() RETURNS trigger AS $$
           BEGIN
             IF COALESCE(NEW.audio_s3_key,'') LIKE '%faulterr%' OR COALESCE(NEW.cover_s3_key,'') LIKE '%faulterr%' THEN
               RAISE EXCEPTION 'injected confirm fault (items)';
             END IF;
             IF COALESCE(NEW.audio_s3_key,'') LIKE '%faultlr%' OR COALESCE(NEW.cover_s3_key,'') LIKE '%faultlr%' THEN
               RETURN NULL;
             END IF;
             RETURN NEW;
           END; $$ LANGUAGE plpgsql",
        "DROP TRIGGER IF EXISTS test_confirm_fault_items ON items",
        "CREATE TRIGGER test_confirm_fault_items BEFORE UPDATE ON items FOR EACH ROW EXECUTE FUNCTION test_confirm_fault_items()",
        r"CREATE OR REPLACE FUNCTION test_confirm_fault_versions() RETURNS trigger AS $$
           BEGIN
             IF COALESCE(NEW.s3_key,'') LIKE '%faulterr%' THEN RAISE EXCEPTION 'injected confirm fault (versions)'; END IF;
             IF COALESCE(NEW.s3_key,'') LIKE '%faultlr%' THEN RETURN NULL; END IF;
             RETURN NEW;
           END; $$ LANGUAGE plpgsql",
        "DROP TRIGGER IF EXISTS test_confirm_fault_versions ON versions",
        "CREATE TRIGGER test_confirm_fault_versions BEFORE UPDATE ON versions FOR EACH ROW EXECUTE FUNCTION test_confirm_fault_versions()",
        r"CREATE OR REPLACE FUNCTION test_confirm_fault_projects() RETURNS trigger AS $$
           BEGIN
             IF COALESCE(NEW.cover_image_url,'') LIKE '%faulterr%' THEN RAISE EXCEPTION 'injected confirm fault (projects)'; END IF;
             IF COALESCE(NEW.cover_image_url,'') LIKE '%faultlr%' THEN RETURN NULL; END IF;
             RETURN NEW;
           END; $$ LANGUAGE plpgsql",
        "DROP TRIGGER IF EXISTS test_confirm_fault_projects ON projects",
        "CREATE TRIGGER test_confirm_fault_projects BEFORE UPDATE ON projects FOR EACH ROW EXECUTE FUNCTION test_confirm_fault_projects()",
    ];
    for s in stmts {
        sqlx::query(s)
            .execute(pool)
            .await
            .expect("install fault trigger");
    }
}

/// Assert a failed confirm rolled back the storage credit AND orphan-enqueued the
/// staged key with the expected reason.
async fn assert_uncredited_and_orphaned(
    h: &TestHarness,
    user_id: &str,
    s3_key: &str,
    before: i64,
    expected_source: &str,
) {
    assert_eq!(
        storage_used(h, user_id).await,
        before,
        "a failed confirm must roll back the storage credit"
    );
    let source: Option<String> =
        sqlx::query_scalar("SELECT source FROM pending_s3_deletions WHERE s3_key = $1")
            .bind(s3_key)
            .fetch_optional(&h.db)
            .await
            .unwrap();
    assert_eq!(
        source.as_deref(),
        Some(expected_source),
        "a failed confirm must orphan-enqueue the staged key with the handler's reason"
    );
}

async fn presign_project_image(h: &mut TestHarness, project_id: &str, file_name: &str) -> String {
    let resp = h
        .client
        .post_json(
            "/api/projects/image/presign",
            &json!({"project_id": project_id, "file_name": file_name, "content_type": "image/png"})
                .to_string(),
        )
        .await;
    assert_eq!(
        resp.status, 200,
        "project image presign failed: {}",
        resp.text
    );
    let data: Value = resp.json();
    data["s3_key"].as_str().unwrap().to_string()
}

async fn presign_item_image(h: &mut TestHarness, item_id: &str, file_name: &str) -> String {
    let resp = h
        .client
        .post_json(
            "/api/items/image/presign",
            &json!({"item_id": item_id, "file_name": file_name, "content_type": "image/png"})
                .to_string(),
        )
        .await;
    assert_eq!(resp.status, 200, "item image presign failed: {}", resp.text);
    let data: Value = resp.json();
    data["s3_key"].as_str().unwrap().to_string()
}

// ---- uploads (item file) -------------------------------------------------

#[tokio::test]
async fn confirm_audio_lost_race_rolls_back_and_orphans() {
    let mut h = TestHarness::with_storage().await;
    let (user_id, _proj, item_id) = setup_creator_with_item(&mut h, 0).await;
    install_confirm_fault_triggers(&h.db).await;

    let s3_key = presign_audio(&mut h, &item_id, "faultlr.mp3").await;
    h.storage.as_ref().unwrap().put(&s3_key, b"x".to_vec());
    let before = storage_used(&h, &user_id).await;

    let resp = h
        .client
        .post_json(
            "/api/upload/confirm",
            &json!({"item_id": item_id, "file_type": "audio", "s3_key": s3_key}).to_string(),
        )
        .await;
    assert_eq!(
        resp.status.as_u16(),
        400,
        "lost-race confirm must 400: {}",
        resp.text
    );
    assert_uncredited_and_orphaned(&h, &user_id, &s3_key, before, "item_upload_target_missing")
        .await;
}

#[tokio::test]
async fn confirm_audio_tx_error_rolls_back_and_orphans() {
    let mut h = TestHarness::with_storage().await;
    let (user_id, _proj, item_id) = setup_creator_with_item(&mut h, 0).await;
    install_confirm_fault_triggers(&h.db).await;

    let s3_key = presign_audio(&mut h, &item_id, "faulterr.mp3").await;
    h.storage.as_ref().unwrap().put(&s3_key, b"x".to_vec());
    let before = storage_used(&h, &user_id).await;

    let resp = h
        .client
        .post_json(
            "/api/upload/confirm",
            &json!({"item_id": item_id, "file_type": "audio", "s3_key": s3_key}).to_string(),
        )
        .await;
    assert!(
        resp.status.is_server_error(),
        "tx-error confirm must 5xx: {} {}",
        resp.status,
        resp.text
    );
    assert_uncredited_and_orphaned(&h, &user_id, &s3_key, before, "item_confirm_failed").await;
}

// ---- versions ------------------------------------------------------------

async fn make_version(h: &mut TestHarness, item_id: &str) -> String {
    let resp = h
        .client
        .post_json(
            &format!("/api/items/{item_id}/versions"),
            &json!({"version_number": "1.0.0"}).to_string(),
        )
        .await;
    assert_eq!(resp.status, 200, "create version failed: {}", resp.text);
    let v: Value = resp.json();
    v["id"].as_str().unwrap().to_string()
}

#[tokio::test]
async fn confirm_version_lost_race_rolls_back_and_orphans() {
    let mut h = TestHarness::with_storage().await;
    let (user_id, _proj, item_id) = setup_creator_with_item(&mut h, 0).await;
    let version_id = make_version(&mut h, &item_id).await;
    install_confirm_fault_triggers(&h.db).await;

    let s3_key = presign_version(&mut h, &version_id, "faultlr.zip").await;
    h.storage.as_ref().unwrap().put(&s3_key, b"x".to_vec());
    let before = storage_used(&h, &user_id).await;

    let resp = h
        .client
        .post_json(
            &format!("/api/versions/{version_id}/upload/confirm"),
            &json!({"s3_key": s3_key}).to_string(),
        )
        .await;
    assert_eq!(
        resp.status.as_u16(),
        400,
        "lost-race version confirm must 400: {}",
        resp.text
    );
    assert_uncredited_and_orphaned(&h, &user_id, &s3_key, before, "version_confirm_lost_race")
        .await;
}

#[tokio::test]
async fn confirm_version_tx_error_rolls_back_and_orphans() {
    let mut h = TestHarness::with_storage().await;
    let (user_id, _proj, item_id) = setup_creator_with_item(&mut h, 0).await;
    let version_id = make_version(&mut h, &item_id).await;
    install_confirm_fault_triggers(&h.db).await;

    let s3_key = presign_version(&mut h, &version_id, "faulterr.zip").await;
    h.storage.as_ref().unwrap().put(&s3_key, b"x".to_vec());
    let before = storage_used(&h, &user_id).await;

    let resp = h
        .client
        .post_json(
            &format!("/api/versions/{version_id}/upload/confirm"),
            &json!({"s3_key": s3_key}).to_string(),
        )
        .await;
    assert!(
        resp.status.is_server_error(),
        "tx-error version confirm must 5xx: {} {}",
        resp.status,
        resp.text
    );
    assert_uncredited_and_orphaned(&h, &user_id, &s3_key, before, "version_confirm_failed").await;
}

// ---- project cover image -------------------------------------------------

#[tokio::test]
async fn confirm_project_image_lost_race_rolls_back_and_orphans() {
    let mut h = TestHarness::with_storage().await;
    let (user_id, project_id, _item) = setup_creator_with_item(&mut h, 0).await;
    install_confirm_fault_triggers(&h.db).await;

    let s3_key = presign_project_image(&mut h, &project_id, "faultlr.png").await;
    h.storage.as_ref().unwrap().put(&s3_key, b"x".to_vec());
    let before = storage_used(&h, &user_id).await;

    let resp = h
        .client
        .post_json(
            "/api/projects/image/confirm",
            &json!({"project_id": project_id, "s3_key": s3_key}).to_string(),
        )
        .await;
    assert_eq!(
        resp.status.as_u16(),
        400,
        "lost-race project image confirm must 400: {}",
        resp.text
    );
    assert_uncredited_and_orphaned(&h, &user_id, &s3_key, before, "project_image_update_failed")
        .await;
}

#[tokio::test]
async fn confirm_project_image_tx_error_rolls_back_and_orphans() {
    let mut h = TestHarness::with_storage().await;
    let (user_id, project_id, _item) = setup_creator_with_item(&mut h, 0).await;
    install_confirm_fault_triggers(&h.db).await;

    let s3_key = presign_project_image(&mut h, &project_id, "faulterr.png").await;
    h.storage.as_ref().unwrap().put(&s3_key, b"x".to_vec());
    let before = storage_used(&h, &user_id).await;

    let resp = h
        .client
        .post_json(
            "/api/projects/image/confirm",
            &json!({"project_id": project_id, "s3_key": s3_key}).to_string(),
        )
        .await;
    assert!(
        resp.status.is_server_error(),
        "tx-error project image confirm must 5xx: {} {}",
        resp.status,
        resp.text
    );
    assert_uncredited_and_orphaned(&h, &user_id, &s3_key, before, "project_image_update_failed")
        .await;
}

// ---- item cover image ----------------------------------------------------

#[tokio::test]
async fn confirm_item_image_lost_race_rolls_back_and_orphans() {
    let mut h = TestHarness::with_storage().await;
    let (user_id, _proj, item_id) = setup_creator_with_item(&mut h, 0).await;
    install_confirm_fault_triggers(&h.db).await;

    let s3_key = presign_item_image(&mut h, &item_id, "faultlr.png").await;
    h.storage.as_ref().unwrap().put(&s3_key, b"x".to_vec());
    let before = storage_used(&h, &user_id).await;

    let resp = h
        .client
        .post_json(
            "/api/items/image/confirm",
            &json!({"item_id": item_id, "s3_key": s3_key}).to_string(),
        )
        .await;
    assert_eq!(
        resp.status.as_u16(),
        400,
        "lost-race item image confirm must 400: {}",
        resp.text
    );
    assert_uncredited_and_orphaned(&h, &user_id, &s3_key, before, "item_image_update_failed").await;
}

#[tokio::test]
async fn confirm_item_image_tx_error_rolls_back_and_orphans() {
    let mut h = TestHarness::with_storage().await;
    let (user_id, _proj, item_id) = setup_creator_with_item(&mut h, 0).await;
    install_confirm_fault_triggers(&h.db).await;

    let s3_key = presign_item_image(&mut h, &item_id, "faulterr.png").await;
    h.storage.as_ref().unwrap().put(&s3_key, b"x".to_vec());
    let before = storage_used(&h, &user_id).await;

    let resp = h
        .client
        .post_json(
            "/api/items/image/confirm",
            &json!({"item_id": item_id, "s3_key": s3_key}).to_string(),
        )
        .await;
    assert!(
        resp.status.is_server_error(),
        "tx-error item image confirm must 5xx: {} {}",
        resp.status,
        resp.text
    );
    assert_uncredited_and_orphaned(&h, &user_id, &s3_key, before, "item_image_update_failed").await;
}

// Multipart upload sessions (CLI / desktop, large files)
//
// These live on the internal surface only, a browser keeps the one-shot
// presigned PUT. The InMemoryStorage backend stubs the session (its presigned
// URLs are fake, so no client bytes flow back), so a full start -> confirm test
// simulates the part PUTs with `put()`, exactly as the presign+confirm tests do.

const GIB: i64 = 1024 * 1024 * 1024;
const CLI_TOKEN: &str = "test-cli-token";
const ACTOR_SECRET: &str = "test-signing-secret-for-integration-tests";

/// Harness wired for the internal (CLI) API with an in-memory backend.
async fn cli_harness() -> TestHarness {
    let mem = std::sync::Arc::new(crate::harness::storage::InMemoryStorage::new());
    TestHarness::build(crate::harness::BuildOptions {
        storage: Some(mem),
        cli_service_token: Some(CLI_TOKEN.to_string()),
        ..Default::default()
    })
    .await
}

/// Authenticate the client as `user_id` over the internal API (ServiceAuth
/// bearer plus the signed actor assertion).
fn act_as(h: &mut TestHarness, user_id: makenotwork::db::UserId) {
    h.client.set_bearer_token(CLI_TOKEN);
    let actor = makenotwork::crypto::mint_internal_actor_token(
        user_id,
        chrono::Utc::now().timestamp() + 3600,
        ACTOR_SECRET,
    );
    h.client.set_actor_token(&actor);
}

#[tokio::test]
async fn multipart_start_opens_the_band_above_the_browser_ceiling() {
    // 6 GiB is past the 2 GiB browser ceiling and inside the big_files tier's
    // 20 GB per-file cap. Before multipart this band was unreachable by every
    // path; this test is the regression guard that it stays open.
    let mut h = cli_harness().await;
    let setup = h.create_creator_with_item("mpcreator", "video", 0).await;
    h.trust_user(setup.user_id).await;
    h.grant_tier(setup.user_id, "big_files").await;
    act_as(&mut h, setup.user_id);

    let body = json!({
        "item_id": setup.item_id,
        "file_type": "video",
        "file_name": "movie.mp4",
        "content_type": "video/mp4",
        "file_size_bytes": 6 * GIB,
    });
    let resp = h
        .client
        .post_json("/api/internal/upload/multipart/start", &body.to_string())
        .await;
    assert_eq!(resp.status, 200, "multipart start failed: {}", resp.text);

    let v: Value = resp.json();
    let s3_key = v["s3_key"].as_str().unwrap();
    assert!(
        s3_key.starts_with("staging/"),
        "must stage, never mint a served key: {s3_key}"
    );
    assert!(!v["upload_id"].as_str().unwrap().is_empty());

    // Geometry is pure arithmetic over the declared size, so the client can
    // derive identical boundaries without a round trip.
    let part_size = v["part_size"].as_u64().unwrap();
    let part_count = v["part_count"].as_u64().unwrap();
    assert!(
        part_size >= 5 * 1024 * 1024,
        "part size must clear S3's 5 MiB floor"
    );
    assert!(
        part_count <= 10_000,
        "part count must stay within S3's limit"
    );
    assert_eq!(
        part_count,
        (6 * GIB as u64).div_ceil(part_size),
        "part count must cover the object"
    );
}

#[tokio::test]
async fn multipart_parts_signs_each_part_with_its_exact_length() {
    let mut h = cli_harness().await;
    let setup = h.create_creator_with_item("mpparts", "video", 0).await;
    h.trust_user(setup.user_id).await;
    h.grant_tier(setup.user_id, "big_files").await;
    act_as(&mut h, setup.user_id);

    // A size with a deliberate remainder so the final part differs from the rest.
    let size = 6 * GIB + 12_345;
    let start: Value = h
        .client
        .post_json(
            "/api/internal/upload/multipart/start",
            &json!({
                "item_id": setup.item_id, "file_type": "video",
                "file_name": "movie.mp4", "content_type": "video/mp4",
                "file_size_bytes": size,
            })
            .to_string(),
        )
        .await
        .json();
    let s3_key = start["s3_key"].as_str().unwrap().to_string();
    let upload_id = start["upload_id"].as_str().unwrap().to_string();
    let part_size = start["part_size"].as_u64().unwrap();
    let part_count = start["part_count"].as_u64().unwrap();

    // A leading window: every part is a full part.
    let resp = h
        .client
        .post_json(
            "/api/internal/upload/multipart/parts",
            &json!({
                "s3_key": s3_key, "upload_id": upload_id,
                "file_size_bytes": size, "first_part": 1, "count": 3,
            })
            .to_string(),
        )
        .await;
    assert_eq!(resp.status, 200, "parts failed: {}", resp.text);
    let v: Value = resp.json();
    let parts = v["parts"].as_array().unwrap();
    assert_eq!(parts.len(), 3);
    for (i, p) in parts.iter().enumerate() {
        assert_eq!(p["part_number"].as_i64().unwrap(), i as i64 + 1);
        assert_eq!(p["content_length"].as_u64().unwrap(), part_size);
        assert!(!p["url"].as_str().unwrap().is_empty());
    }

    // The final part carries only the remainder, and a window running past the
    // end is clamped rather than rejected.
    let resp = h
        .client
        .post_json(
            "/api/internal/upload/multipart/parts",
            &json!({
                "s3_key": s3_key, "upload_id": upload_id,
                "file_size_bytes": size, "first_part": part_count, "count": 10,
            })
            .to_string(),
        )
        .await;
    assert_eq!(resp.status, 200, "final-part window failed: {}", resp.text);
    let v: Value = resp.json();
    let parts = v["parts"].as_array().unwrap();
    assert_eq!(parts.len(), 1, "window past the last part must clamp");
    assert_eq!(
        parts[0]["content_length"].as_u64().unwrap(),
        12_345,
        "last part is the remainder"
    );
}

#[tokio::test]
async fn multipart_parts_refuses_a_size_that_disagrees_with_start() {
    // deepaudit F1: the part geometry is bound to the size `start` validated
    // against the tier cap. A session opened for 2 GB must not be able to widen
    // itself at `parts` time, trusting the parts body would let it mint URLs for
    // a 5 TiB object, unbudgeted S3 writes bounded only by the 24h reaper.
    let mut h = cli_harness().await;
    let setup = h.create_creator_with_item("mpliar", "video", 0).await;
    h.trust_user(setup.user_id).await;
    h.grant_tier(setup.user_id, "big_files").await;
    act_as(&mut h, setup.user_id);

    let declared = 2 * GIB;
    let start: Value = h
        .client
        .post_json(
            "/api/internal/upload/multipart/start",
            &json!({
                "item_id": setup.item_id, "file_type": "video",
                "file_name": "movie.mp4", "content_type": "video/mp4",
                "file_size_bytes": declared,
            })
            .to_string(),
        )
        .await
        .json();
    let s3_key = start["s3_key"].as_str().unwrap().to_string();
    let upload_id = start["upload_id"].as_str().unwrap().to_string();

    // Claim a wildly larger size at parts time.
    let resp = h
        .client
        .post_json(
            "/api/internal/upload/multipart/parts",
            &json!({
                "s3_key": s3_key, "upload_id": upload_id,
                "file_size_bytes": 5 * 1024 * GIB, "first_part": 1, "count": 1,
            })
            .to_string(),
        )
        .await;
    assert_eq!(
        resp.status, 400,
        "a parts size disagreeing with start must be refused: {}",
        resp.text
    );
    assert!(
        resp.text.contains("does not match"),
        "expected a declared-size mismatch error, got {}",
        resp.text
    );
}

#[tokio::test]
async fn multipart_parts_bounds_the_window_it_will_mint() {
    // Minting every URL for a 20 GB object would issue thousands of hour-long
    // credentials for an upload that may never happen.
    let mut h = cli_harness().await;
    let setup = h.create_creator_with_item("mpwindow", "video", 0).await;
    h.trust_user(setup.user_id).await;
    h.grant_tier(setup.user_id, "big_files").await;
    act_as(&mut h, setup.user_id);

    let start: Value = h
        .client
        .post_json(
            "/api/internal/upload/multipart/start",
            &json!({
                "item_id": setup.item_id, "file_type": "video",
                "file_name": "movie.mp4", "content_type": "video/mp4",
                "file_size_bytes": 6 * GIB,
            })
            .to_string(),
        )
        .await
        .json();

    for (first_part, count) in [(1, 101), (1, 0), (0, 5)] {
        let resp = h
            .client
            .post_json(
                "/api/internal/upload/multipart/parts",
                &json!({
                    "s3_key": start["s3_key"], "upload_id": start["upload_id"],
                    "file_size_bytes": 6 * GIB, "first_part": first_part, "count": count,
                })
                .to_string(),
            )
            .await;
        assert_eq!(
            resp.status, 400,
            "first_part={first_part} count={count} must be refused, got {}: {}",
            resp.status, resp.text
        );
    }
}

#[tokio::test]
async fn multipart_refuses_another_creators_staging_key() {
    // A `staging/{uuid}` key carries no user in its path, so ownership comes
    // from the pending_uploads row. Without that check any authenticated creator
    // could drive parts into someone else's in-flight session.
    let mut h = cli_harness().await;

    let victim = h.create_creator_with_item("mpvictim", "video", 0).await;
    h.trust_user(victim.user_id).await;
    h.grant_tier(victim.user_id, "big_files").await;
    act_as(&mut h, victim.user_id);
    let start: Value = h
        .client
        .post_json(
            "/api/internal/upload/multipart/start",
            &json!({
                "item_id": victim.item_id, "file_type": "video",
                "file_name": "movie.mp4", "content_type": "video/mp4",
                "file_size_bytes": 6 * GIB,
            })
            .to_string(),
        )
        .await
        .json();
    let victim_key = start["s3_key"].as_str().unwrap().to_string();
    let victim_upload_id = start["upload_id"].as_str().unwrap().to_string();

    // A second creator, fully authenticated in their own right.
    let attacker = h.create_creator_with_item("mpattacker", "video", 0).await;
    h.trust_user(attacker.user_id).await;
    h.grant_tier(attacker.user_id, "big_files").await;
    act_as(&mut h, attacker.user_id);

    let parts = h
        .client
        .post_json(
            "/api/internal/upload/multipart/parts",
            &json!({
                "s3_key": victim_key, "upload_id": victim_upload_id,
                "file_size_bytes": 6 * GIB, "first_part": 1, "count": 1,
            })
            .to_string(),
        )
        .await;
    assert_eq!(
        parts.status, 400,
        "cross-user parts must be refused: {}",
        parts.text
    );

    let complete = h
        .client
        .post_json(
            "/api/internal/upload/multipart/complete",
            &json!({
                "s3_key": victim_key, "upload_id": victim_upload_id,
                "parts": [{"part_number": 1, "etag": "\"deadbeef\""}],
            })
            .to_string(),
        )
        .await;
    assert_eq!(
        complete.status, 400,
        "cross-user complete must be refused: {}",
        complete.text
    );

    let abort = h
        .client
        .post_json(
            "/api/internal/upload/multipart/abort",
            &json!({"s3_key": victim_key, "upload_id": victim_upload_id}).to_string(),
        )
        .await;
    assert_eq!(
        abort.status, 400,
        "cross-user abort must be refused: {}",
        abort.text
    );
}

#[tokio::test]
async fn multipart_start_refuses_a_file_over_the_tier_cap() {
    // Skipping the single-PUT ceiling must not skip the limits that describe the
    // file: small_files caps a single file at 500 MB, so a 1 GiB video is refused
    // before it stages any parts.
    let mut h = cli_harness().await;
    let setup = h.create_creator_with_item("mptier", "video", 0).await;
    h.trust_user(setup.user_id).await;
    h.grant_tier(setup.user_id, "small_files").await;
    act_as(&mut h, setup.user_id);

    let resp = h
        .client
        .post_json(
            "/api/internal/upload/multipart/start",
            &json!({
                "item_id": setup.item_id, "file_type": "video",
                "file_name": "movie.mp4", "content_type": "video/mp4",
                "file_size_bytes": GIB,
            })
            .to_string(),
        )
        .await;
    assert_eq!(
        resp.status, 413,
        "over-tier multipart start must be refused: {}",
        resp.text
    );

    // Nothing was staged for the reaper to clean up.
    let pending: i64 =
        sqlx::query_scalar("SELECT COUNT(*) FROM pending_uploads WHERE user_id = $1")
            .bind(setup.user_id)
            .fetch_one(&h.db)
            .await
            .unwrap();
    assert_eq!(
        pending, 0,
        "a refused start must not record a pending upload"
    );
}

#[tokio::test]
async fn multipart_complete_then_confirm_commits_the_upload() {
    // The end-to-end handoff: multipart replaces the transport only, and the
    // existing confirm applies every size/tier/scan/commit rule unchanged.
    let mut h = cli_harness().await;
    let setup = h.create_creator_with_item("mpflow", "audio", 0).await;
    h.trust_user(setup.user_id).await;
    h.grant_tier(setup.user_id, "small_files").await;
    act_as(&mut h, setup.user_id);

    let bytes = b"fake mp3 bytes for a multipart upload".to_vec();
    let size = bytes.len() as i64;

    let start: Value = h
        .client
        .post_json(
            "/api/internal/upload/multipart/start",
            &json!({
                "item_id": setup.item_id, "file_type": "audio",
                "file_name": "song.mp3", "content_type": "audio/mpeg",
                "file_size_bytes": size,
            })
            .to_string(),
        )
        .await
        .json();
    let s3_key = start["s3_key"].as_str().unwrap().to_string();
    let upload_id = start["upload_id"].as_str().unwrap().to_string();
    assert_eq!(
        start["part_count"].as_u64().unwrap(),
        1,
        "a small file is a single part"
    );

    let parts: Value = h
        .client
        .post_json(
            "/api/internal/upload/multipart/parts",
            &json!({
                "s3_key": s3_key, "upload_id": upload_id,
                "file_size_bytes": size, "first_part": 1, "count": 1,
            })
            .to_string(),
        )
        .await
        .json();
    assert_eq!(parts["parts"][0]["content_length"].as_i64().unwrap(), size);

    // The client PUTs each part to its presigned URL; the in-memory backend has
    // no way to receive them, so stand in for that here.
    h.storage.as_ref().unwrap().put(&s3_key, bytes);

    let complete = h
        .client
        .post_json(
            "/api/internal/upload/multipart/complete",
            &json!({
                "s3_key": s3_key, "upload_id": upload_id,
                "parts": [{"part_number": 1, "etag": "\"etag-1\""}],
            })
            .to_string(),
        )
        .await;
    assert_eq!(complete.status, 200, "complete failed: {}", complete.text);

    // Hand off to the unchanged confirm endpoint.
    let confirm = h
        .client
        .post_json(
            "/api/internal/upload/confirm",
            &json!({
                "user_id": setup.user_id, "item_id": setup.item_id,
                "file_type": "audio", "s3_key": s3_key,
            })
            .to_string(),
        )
        .await;
    assert_eq!(
        confirm.status, 200,
        "confirm after multipart failed: {}",
        confirm.text
    );
    h.client.clear_bearer_token();

    let db_key: Option<String> =
        sqlx::query_scalar("SELECT audio_s3_key FROM items WHERE id = $1::uuid")
            .bind(&setup.item_id)
            .fetch_one(&h.db)
            .await
            .unwrap();
    assert_eq!(
        db_key.as_deref(),
        Some(s3_key.as_str()),
        "confirm must commit the multipart object"
    );

    let storage_used: i64 =
        sqlx::query_scalar("SELECT storage_used_bytes FROM users WHERE id = $1")
            .bind(setup.user_id)
            .fetch_one(&h.db)
            .await
            .unwrap();
    assert_eq!(
        storage_used, size,
        "confirm must charge the real object size"
    );
}

#[tokio::test]
async fn orphan_reaper_aborts_abandoned_multipart_sessions() {
    // A multipart session that was started and never completed leaves NO object
    // to delete, only uploaded parts that S3 bills for until they are aborted.
    // The reaper's object delete is a no-op against it, so without this abort
    // the parts leak indefinitely, a cost bug, not just a tidiness one.
    use makenotwork::scheduler::abort_orphan_multipart_sessions;

    let mem = crate::harness::storage::InMemoryStorage::new();
    let target = "staging/abandoned-uuid/movie.mp4";

    // Two abandoned sessions on the reaped key, plus one on a different key
    // that must survive: ListMultipartUploads matches a PREFIX, so a careless
    // implementation reaping `staging/abc` would also kill `staging/abcdef`.
    mem.put_open_multipart("upload-a", target);
    mem.put_open_multipart("upload-b", target);
    mem.put_open_multipart("upload-c", "staging/abandoned-uuid/movie.mp4.other");
    assert_eq!(mem.open_multipart_count(), 3);

    let aborted = abort_orphan_multipart_sessions(&mem, target).await;

    assert_eq!(
        aborted, 2,
        "both sessions on the reaped key must be aborted"
    );
    assert_eq!(
        mem.open_multipart_count(),
        1,
        "a session on a different key must survive the prefix-adjacent reap"
    );
}

#[tokio::test]
async fn orphan_reaper_abort_is_a_noop_when_there_is_no_session() {
    // The common case: an ordinary single-PUT orphan has no multipart session,
    // and the reaper must sail past it without erroring or miscounting.
    use makenotwork::scheduler::abort_orphan_multipart_sessions;

    let mem = crate::harness::storage::InMemoryStorage::new();
    mem.put("staging/plain/song.mp3", b"bytes".to_vec());

    let aborted = abort_orphan_multipart_sessions(&mem, "staging/plain/song.mp3").await;
    assert_eq!(aborted, 0);
}