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

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

/// 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!(resp.status.is_success(), "Presign failed: {}", resp.text);

    let data: Value = resp.json();
    assert!(data["upload_url"].as_str().unwrap().starts_with("http://test-storage/"));
    assert!(data["s3_key"].as_str().unwrap().contains("/audio/episode.mp3"));
    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!(resp.status.is_success());
    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!(resp.status.is_success(), "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 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!(resp.status.is_success(), "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!(resp.status.is_success(), "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 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!(resp.status.is_success());
    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/{}/versions", item_id),
            &json!({"version_number": "1.0.0"}).to_string(),
        )
        .await;
    assert!(resp.status.is_success(), "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/{}/upload/presign", version_id),
            &json!({
                "file_name": "plugin.zip",
                "content_type": "application/zip",
            })
            .to_string(),
        )
        .await;
    assert!(resp.status.is_success(), "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/{}/upload/confirm", version_id),
            &json!({"s3_key": s3_key}).to_string(),
        )
        .await;
    assert!(resp.status.is_success(), "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/{}/download", version_id))
        .await;
    assert!(resp.status.is_success(), "Version download failed: {}", resp.text);
    let data: Value = resp.json();
    assert!(data["download_url"].as_str().unwrap().starts_with("http://test-storage/"));
}

// ---------------------------------------------------------------------------
// 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/{}/audio/track.mp3", item_id);
    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!(resp.status.is_success(), "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/{}/audio/track.mp3", item_id);
    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!(
        resp.status.is_success(),
        "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!(
        resp.status.is_success(),
        "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!(resp.status.is_success(), "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/{}/upload/presign", version_id),
            &json!({"file_name": "app.zip", "content_type": "application/zip"}).to_string(),
        )
        .await;
    assert!(resp.status.is_success());
    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/{}/upload/confirm", version_id),
            &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/{}/download", version_id))
        .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!(resp.status.is_success(), "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/{}/upload/presign", version_id),
            &json!({"file_name": file_name, "content_type": "application/zip"}).to_string(),
        )
        .await;
    assert!(resp.status.is_success(), "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!(!resp.status.is_success(), "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");
    // 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");
    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!(resp.status.is_success(), "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!(resp.status.is_success(), "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/{}/versions", item_id),
            &json!({"version_number": "1.0.0"}).to_string(),
        )
        .await;
    assert!(resp.status.is_success(), "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/{}/upload/confirm", version_id),
            &json!({"s3_key": key1}).to_string(),
        )
        .await;
    assert!(resp.status.is_success(), "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/{}/upload/confirm", version_id),
            &json!({"s3_key": key2}).to_string(),
        )
        .await;
    assert!(resp.status.is_success(), "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!(resp.status.is_success(), "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!(resp.status.is_success(), "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"
    );
}