objectstore-client 0.2.10

mod common;

use std::collections::{BTreeMap, HashSet};
use std::io::Write as _;
use std::time::Duration;

use futures_util::StreamExt as _;
use jsonwebtoken::{Algorithm, EncodingKey, Header, encode, get_current_timestamp};
use objectstore_client::{
    Client, Error, ExpirationPolicy, ExpiryExtension, ExtendExpiryResponse, OperationResult,
    Permission, Usecase,
};
use objectstore_test::server::{TEST_EDDSA_KID, TEST_EDDSA_PRIVKEY};
use objectstore_types::metadata::Compression;
use reqwest::StatusCode;
use serde::Serialize;

use common::{test_server, test_token_generator};

#[cfg(feature = "resumable-upload-api")]
use objectstore_client::UploadProgress;

#[derive(Serialize)]
struct JwtClaims {
    exp: u64,
    permissions: HashSet<Permission>,
    res: JwtRes,
}

#[derive(Serialize)]
struct JwtRes {
    #[serde(rename = "os:usecase")]
    usecase: String,
    #[serde(flatten)]
    scopes: BTreeMap<String, String>,
}

/// Signs a static token for the given usecase and scopes.
fn sign_static_token(usecase: &str, scopes: &[(&str, &str)]) -> String {
    let encoding_key = EncodingKey::from_ed_pem(TEST_EDDSA_PRIVKEY.as_bytes()).unwrap();
    let claims = JwtClaims {
        exp: get_current_timestamp() + 60,
        permissions: HashSet::from([
            Permission::ObjectRead,
            Permission::ObjectWrite,
            Permission::ObjectDelete,
        ]),
        res: JwtRes {
            usecase: usecase.into(),
            scopes: scopes
                .iter()
                .map(|(k, v)| (k.to_string(), v.to_string()))
                .collect(),
        },
    };
    let mut header = Header::new(Algorithm::EdDSA);
    header.kid = Some(TEST_EDDSA_KID.into());
    encode(&header, &claims, &encoding_key).unwrap()
}

#[tokio::test]
async fn stores_uncompressed() {
    let server = test_server().await;

    let client = Client::builder(server.url("/"))
        .token(test_token_generator())
        .build()
        .unwrap();
    let usecase = Usecase::new("usecase");
    let session = client.session(usecase.for_organization(12345)).unwrap();

    let body = "oh hai!";

    let stored_id = session.put(body).compress(None).send().await.unwrap().key;

    let response = session.get(&stored_id).send().await.unwrap().unwrap();
    assert_eq!(response.metadata.compression, None);
    assert!(response.metadata.time_created.is_some());

    let received = response.payload().await.unwrap();
    assert_eq!(received, "oh hai!");
}

#[tokio::test]
async fn uses_zstd_by_default() {
    let server = test_server().await;

    let client = Client::builder(server.url("/"))
        .token(test_token_generator())
        .build()
        .unwrap();
    let usecase = Usecase::new("usecase");
    let session = client.session(usecase.for_organization(12345)).unwrap();

    let body = "oh hai!";
    let stored_id = session.put(body).send().await.unwrap().key;

    // when the user indicates that it can deal with zstd, it gets zstd
    let request = session.get(&stored_id).decompress(false);
    let response = request.send().await.unwrap().unwrap();
    assert_eq!(response.metadata.compression, Some(Compression::Zstd));

    let received_compressed = response.payload().await.unwrap();
    let decompressed = zstd::bulk::decompress(&received_compressed, 1024).unwrap();
    assert_eq!(decompressed, b"oh hai!");

    // otherwise, the client does the decompression
    let response = session.get(&stored_id).send().await.unwrap().unwrap();
    assert_eq!(response.metadata.compression, None);

    let received = response.payload().await.unwrap();
    assert_eq!(received, "oh hai!");
}

#[tokio::test]
async fn stores_precompressed_payload_verbatim() {
    let server = test_server().await;

    let client = Client::builder(server.url("/"))
        .token(test_token_generator())
        .build()
        .unwrap();
    let usecase = Usecase::new("usecase");
    let session = client.session(usecase.for_organization(12345)).unwrap();

    let compressed = zstd::bulk::compress(b"oh hai!", 0).unwrap();
    let stored_id = session
        .put(compressed.clone())
        .precompressed(Compression::Zstd)
        .send()
        .await
        .unwrap()
        .key;

    // the payload is stored as-is, without another compression pass
    let response = session
        .get(&stored_id)
        .accept_encoding([Compression::Zstd])
        .send()
        .await
        .unwrap()
        .unwrap();
    assert_eq!(response.metadata.compression, Some(Compression::Zstd));
    assert_eq!(response.payload().await.unwrap(), compressed);

    // and the recorded encoding makes downloads decompress transparently
    let response = session.get(&stored_id).send().await.unwrap().unwrap();
    assert_eq!(response.metadata.compression, None);
    assert_eq!(response.payload().await.unwrap(), "oh hai!");
}

#[tokio::test]
async fn deletes_stores_stuff() {
    let server = test_server().await;

    let client = Client::builder(server.url("/"))
        .token(test_token_generator())
        .build()
        .unwrap();
    let usecase = Usecase::new("usecase");
    let session = client.session(usecase.for_project(12345, 1337)).unwrap();

    let body = "oh hai!";
    let stored_id = session.put(body).send().await.unwrap().key;

    session.delete(&stored_id).send().await.unwrap();

    let response = session.get(&stored_id).send().await.unwrap();
    assert!(response.is_none());
}

#[tokio::test]
async fn stores_under_given_key() {
    let server = test_server().await;

    let client = Client::builder(server.url("/"))
        .token(test_token_generator())
        .build()
        .unwrap();
    let usecase = Usecase::new("usecase");
    let session = client.session(usecase.for_project(12345, 1337)).unwrap();

    let body = "oh hai!";
    let request = session.put(body).key("test-key123!!");
    let stored_id = request.send().await.unwrap().key;

    assert_eq!(stored_id, "test-key123!!");
}

#[tokio::test]
async fn roundtrips_unicode_metadata() {
    let server = test_server().await;

    let client = Client::builder(server.url("/"))
        .token(test_token_generator())
        .build()
        .unwrap();
    let usecase = Usecase::new("usecase");
    let session = client.session(usecase.for_project(12345, 1337)).unwrap();

    let filename = "réport-📄.pdf";
    let release = "vérsion-1.0-🚀";

    let stored_id = session
        .put("oh hai!")
        .filename(filename)
        .origin("Ünknown-源")
        .append_metadata("release", release)
        .append_metadata("note", "100% done")
        .key("unicode-metadata")
        .send()
        .await
        .unwrap()
        .key;

    let response = session.get(&stored_id).send().await.unwrap().unwrap();
    let metadata = response.metadata.clone();
    assert_eq!(metadata.filename.as_deref(), Some(filename));
    assert_eq!(metadata.origin.as_deref(), Some("Ünknown-源"));
    assert_eq!(
        metadata.custom.get("release").map(String::as_str),
        Some(release)
    );
    assert_eq!(
        metadata.custom.get("note").map(String::as_str),
        Some("100% done"),
    );
    assert_eq!(response.payload().await.unwrap(), "oh hai!");

    let metadata = session.head(&stored_id).send().await.unwrap().unwrap();
    assert_eq!(metadata.filename.as_deref(), Some(filename));
    assert_eq!(
        metadata.custom.get("release").map(String::as_str),
        Some(release)
    );
}

#[tokio::test]
async fn stores_structured_keys() {
    let server = test_server().await;

    let client = Client::builder(server.url("/"))
        .token(test_token_generator())
        .build()
        .unwrap();
    let usecase = Usecase::new("usecase");
    let session = client.session(usecase.for_project(12345, 1337)).unwrap();

    let body = "oh hai!";
    let request = session.put(body).key("1/shard-0.json");
    let stored_id = request.send().await.unwrap().key;
    assert_eq!(stored_id, "1/shard-0.json");

    let response = session.get(&stored_id).send().await.unwrap().unwrap();
    let received = response.payload().await.unwrap();
    assert_eq!(received, body);
}

#[tokio::test]
async fn overwrites_existing_key() {
    let server = test_server().await;

    let client = Client::builder(server.url("/"))
        .token(test_token_generator())
        .build()
        .unwrap();
    let usecase = Usecase::new("usecase");
    let session = client.session(usecase.for_project(12345, 1337)).unwrap();

    let stored_id = session.put("initial body").send().await.unwrap().key;
    let request = session.put("new body").key(&stored_id);
    let overwritten_id = request.send().await.unwrap().key;

    assert_eq!(stored_id, overwritten_id);

    let response = session.get(&stored_id).send().await.unwrap().unwrap();
    let payload = response.payload().await.unwrap();
    assert_eq!(payload, "new body");
}

#[tokio::test]
async fn not_found_with_wrong_scope() {
    let server = test_server().await;

    let client = Client::builder(server.url("/"))
        .token(test_token_generator())
        .build()
        .unwrap();
    let usecase = Usecase::new("usecase");

    // First we have to place an object with one scope
    let session = client.session(usecase.for_project(12345, 1337)).unwrap();
    let stored_id = session.put("initial body").send().await.unwrap().key;

    // Now we need to try to fetch the object with a different scope
    let session = client.session(usecase.for_project(12345, 9999)).unwrap();
    let response = session.get(&stored_id).send().await.unwrap();
    assert!(response.is_none());
}

#[tokio::test]
async fn stores_with_origin() {
    let server = test_server().await;

    let client = Client::builder(server.url("/"))
        .token(test_token_generator())
        .build()
        .unwrap();
    let usecase = Usecase::new("usecase");
    let session = client.session(usecase.for_organization(12345)).unwrap();

    let stored_id = session
        .put("hello with origin")
        .compress(None)
        .origin("203.0.113.42")
        .send()
        .await
        .unwrap()
        .key;

    let response = session.get(&stored_id).send().await.unwrap().unwrap();
    assert_eq!(response.metadata.origin.as_deref(), Some("203.0.113.42"));
}

#[tokio::test]
async fn stores_without_origin() {
    let server = test_server().await;

    let client = Client::builder(server.url("/"))
        .token(test_token_generator())
        .build()
        .unwrap();
    let usecase = Usecase::new("usecase");
    let session = client.session(usecase.for_organization(12345)).unwrap();

    let stored_id = session
        .put("hello without origin")
        .compress(None)
        .send()
        .await
        .unwrap()
        .key;

    let response = session.get(&stored_id).send().await.unwrap().unwrap();
    assert!(response.metadata.origin.is_none());
}

#[tokio::test]
async fn fails_with_insufficient_auth_token_perms() {
    let server = test_server().await;

    let token_generator = test_token_generator().permissions(&[]);

    let client = Client::builder(server.url("/"))
        .token(token_generator)
        .build()
        .unwrap();
    let usecase = Usecase::new("usecase");
    let session = client.session(usecase.for_project(12345, 1337)).unwrap();

    let put_result = session.put("initial body").send().await;
    println!("{put_result:?}");
    match put_result {
        Err(Error::Reqwest(err)) => assert_eq!(err.status().unwrap(), StatusCode::FORBIDDEN),
        _ => panic!("Expected error"),
    }

    let delete_result = session.delete("some-key").send().await;
    println!("{delete_result:?}");
    match delete_result {
        Err(Error::Reqwest(err)) => assert_eq!(err.status().unwrap(), StatusCode::FORBIDDEN),
        _ => panic!("Expected error"),
    }

    let error = session
        .extend_expiry(
            "some-key",
            ExpiryExtension::FromNow(Duration::from_secs(86400)),
        )
        .send()
        .await
        .unwrap_err();
    assert!(
        matches!(error, Error::Reqwest(error) if error.status() == Some(StatusCode::FORBIDDEN))
    );
}

#[tokio::test]
async fn create_token_rejects_escalation() {
    // A read-only generator cannot create a token requesting write access.
    let client = Client::builder("http://127.0.0.1:8888/")
        .token(test_token_generator().permissions(&[Permission::ObjectRead]))
        .build()
        .unwrap();
    let usecase = Usecase::new("usecase");
    let session = client.session(usecase.for_organization(12345)).unwrap();

    let result = session
        .create_token()
        .unwrap()
        .expect("generator is configured")
        .permissions(&[Permission::ObjectRead, Permission::ObjectWrite])
        .sign();
    match result {
        Err(Error::PermissionEscalation { escalated }) => {
            assert_eq!(escalated, vec![Permission::ObjectWrite]);
        }
        other => panic!("Expected PermissionEscalation, got: {other:?}"),
    }

    // A subset of the granted permissions is accepted.
    assert!(
        session
            .create_token()
            .unwrap()
            .expect("generator is configured")
            .permissions(&[Permission::ObjectRead])
            .expiry_seconds(30)
            .sign()
            .is_ok()
    );
}

#[tokio::test]
async fn create_token_rejects_static_token() {
    let token = sign_static_token("usecase", &[("org", "12345")]);
    let client = Client::builder("http://127.0.0.1:8888/")
        .token(token)
        .build()
        .unwrap();
    let usecase = Usecase::new("usecase");
    let session = client.session(usecase.for_organization(12345)).unwrap();

    // A static, pre-signed token cannot be re-signed with custom permissions or expiry.
    assert!(matches!(
        session.create_token(),
        Err(Error::StaticTokenOverride)
    ));

    // The static token itself is still returned as-is.
    assert!(session.get_token().unwrap().is_some());
}

#[tokio::test]
async fn create_token_without_provider() {
    let client = Client::builder("http://127.0.0.1:8888/").build().unwrap();
    let usecase = Usecase::new("usecase");
    let session = client.session(usecase.for_organization(12345)).unwrap();

    // Without a token provider there is no request to configure, and no token to return.
    assert!(session.create_token().unwrap().is_none());
    assert!(session.get_token().unwrap().is_none());
}

#[tokio::test]
async fn stores_with_static_token() {
    let server = test_server().await;

    let token = sign_static_token("usecase", &[("org", "12345")]);

    let client = Client::builder(server.url("/"))
        .token(token)
        .build()
        .unwrap();
    let usecase = Usecase::new("usecase");
    let session = client.session(usecase.for_organization(12345)).unwrap();

    let body = "hello with static token!";
    let stored_id = session.put(body).compress(None).send().await.unwrap().key;

    let response = session.get(&stored_id).send().await.unwrap().unwrap();
    let received = response.payload().await.unwrap();
    assert_eq!(received, body);
}

#[tokio::test]
async fn batch_operations() {
    let server = test_server().await;

    let client = Client::builder(server.url("/"))
        .token(test_token_generator())
        .build()
        .unwrap();
    let usecase = Usecase::new("usecase");
    let session = client.session(usecase.for_project(12345, 1337)).unwrap();

    // First batch: 4 PUT operations
    // key-2 uses default compression (zstd), others are uncompressed
    let results: Vec<_> = session
        .many()
        .push(
            session
                .put("first object")
                .compress(None)
                .filename("report.pdf")
                .key("key-1"),
        )
        .push(session.put("second object").key("key-2"))
        .push(session.put("third object").compress(None).key("key-3"))
        .push(session.put("fourth object").compress(None).key("key-4"))
        .send()
        .await
        .collect()
        .await;

    assert_eq!(results.len(), 4);
    let mut keys: Vec<String> = results
        .iter()
        .map(|r| match r {
            OperationResult::Put(key, Ok(_)) => key.clone(),
            other => panic!("Expected Put result, got: {other:?}"),
        })
        .collect();
    keys.sort();
    assert_eq!(keys, vec!["key-1", "key-2", "key-3", "key-4"]);

    // Second batch: GET key-1, GET key-2 (automatic decompression), DELETE key-3, PUT key-4 (override)
    let results: Vec<_> = session
        .many()
        .push(session.get("key-1"))
        .push(session.get("key-2"))
        .push(session.delete("key-3"))
        .push(
            session
                .put("overridden fourth object")
                .compress(None)
                .key("key-4"),
        )
        .send()
        .await
        .collect()
        .await;

    assert_eq!(results.len(), 4);

    // Results may come back in any order due to concurrent execution, so collect into a map by key
    let mut gets = BTreeMap::new();
    let mut deletes = HashSet::new();
    let mut puts = HashSet::new();
    for result in results {
        match result {
            OperationResult::Get(key, inner) => {
                gets.insert(key, inner);
            }
            OperationResult::Delete(key, Ok(())) => {
                deletes.insert(key);
            }
            OperationResult::Put(key, Ok(_)) => {
                puts.insert(key);
            }
            other => panic!("Unexpected result: {other:?}"),
        }
    }

    // GET key-1 (uncompressed)
    let get1 = gets
        .remove("key-1")
        .expect("missing get key-1")
        .unwrap()
        .unwrap();
    assert_eq!(get1.metadata.compression, None);
    assert!(get1.metadata.time_created.is_some());
    assert_eq!(get1.metadata.filename.as_deref(), Some("report.pdf"));
    assert_eq!(get1.payload().await.unwrap().as_ref(), b"first object");

    // GET key-2 (automatic decompression)
    let get2 = gets
        .remove("key-2")
        .expect("missing get key-2")
        .unwrap()
        .unwrap();
    assert_eq!(get2.metadata.compression, None);
    assert!(get2.metadata.time_created.is_some());
    assert!(get2.metadata.filename.is_none());
    assert_eq!(get2.payload().await.unwrap().as_ref(), b"second object");

    // DELETE key-3
    assert!(deletes.contains("key-3"), "missing delete for key-3");

    // PUT key-4 (override)
    assert!(puts.contains("key-4"), "missing put for key-4");

    // Verify the overridden object has the new content
    let response = session.get("key-4").send().await.unwrap().unwrap();
    let payload = response.payload().await.unwrap();
    assert_eq!(payload, "overridden fourth object");

    // Verify the deleted object is gone
    let response = session.get("key-3").send().await.unwrap();
    assert!(response.is_none());
}

#[tokio::test]
async fn batch_insert_without_key() {
    let server = test_server().await;

    let client = Client::builder(server.url("/"))
        .token(test_token_generator())
        .build()
        .unwrap();
    let usecase = Usecase::new("usecase");
    let session = client.session(usecase.for_project(12345, 1337)).unwrap();

    // Insert without specifying a key — the server should generate one
    let results: Vec<_> = session
        .many()
        .push(session.put("keyless object").compress(None))
        .send()
        .await
        .collect()
        .await;

    assert_eq!(results.len(), 1);

    let server_key = match &results[0] {
        OperationResult::Put(key, Ok(_)) => key.clone(),
        other => panic!("Expected Put result, got: {other:?}"),
    };

    // The server should have assigned a non-empty key
    assert!(
        !server_key.is_empty(),
        "server-assigned key must not be empty"
    );

    // Verify we can retrieve the object using the server-assigned key
    let response = session.get(&server_key).send().await.unwrap().unwrap();
    let payload = response.payload().await.unwrap();
    assert_eq!(payload, "keyless object");
}

#[tokio::test]
async fn batch_partial_failures() {
    let server = test_server().await;

    // Use a read-only token so writes/deletes fail with 403
    let read_only_token = test_token_generator().permissions(&[Permission::ObjectRead]);

    let client = Client::builder(server.url("/"))
        .token(read_only_token)
        .build()
        .unwrap();
    let usecase = Usecase::new("usecase");
    let session = client.session(usecase.for_project(12345, 1337)).unwrap();

    let results: Vec<_> = session
        .many()
        .push(session.get("nonexistent-key-1"))
        .push(session.put("should fail").key("write-key"))
        .push(session.delete("delete-key"))
        .push(session.get("nonexistent-key-2"))
        .push(session.get("nonexistent-key-3"))
        .send()
        .await
        .collect()
        .await;

    assert_eq!(results.len(), 5);

    // Collect results into maps by key for order-independent assertions
    let mut gets = BTreeMap::new();
    let mut puts = BTreeMap::new();
    let mut deletes = BTreeMap::new();
    for result in results {
        match result {
            OperationResult::Get(key, inner) => {
                gets.insert(key, inner);
            }
            OperationResult::Put(key, inner) => {
                puts.insert(key, inner);
            }
            OperationResult::Delete(key, inner) => {
                deletes.insert(key, inner);
            }
            other => panic!("Unexpected result: {other:?}"),
        }
    }

    // GET before failures should succeed (read is permitted)
    let get_result = gets
        .remove("nonexistent-key-1")
        .expect("missing get result");
    assert!(matches!(get_result, Ok(None)));

    // PUT should fail with 403 (no write permission)
    let put_result = puts.remove("write-key").expect("missing put result");
    match put_result {
        Err(Error::OperationFailure { status, .. }) => assert_eq!(status, 403),
        other => panic!("Expected OperationFailure(403), got: {other:?}"),
    }

    // DELETE should fail with 403 (no delete permission)
    let delete_result = deletes.remove("delete-key").expect("missing delete result");
    match delete_result {
        Err(Error::OperationFailure { status, .. }) => assert_eq!(status, 403),
        other => panic!("Expected OperationFailure(403), got: {other:?}"),
    }

    // GETs after the failures should still succeed
    let get_result = gets
        .remove("nonexistent-key-2")
        .expect("missing get result for key-2");
    assert!(matches!(get_result, Ok(None)));

    let get_result = gets
        .remove("nonexistent-key-3")
        .expect("missing get result for key-3");
    assert!(matches!(get_result, Ok(None)));
}

#[tokio::test]
async fn batch_put_files() {
    let server = test_server().await;

    let client = Client::builder(server.url("/"))
        .token(test_token_generator())
        .build()
        .unwrap();
    let usecase = Usecase::new("usecase");
    let session = client.session(usecase.for_project(12345, 1337)).unwrap();

    let dir = tempfile::tempdir().unwrap();

    // Create a few small files (under 1 MB batch limit).
    let small_bodies: Vec<(&str, Vec<u8>)> = vec![
        ("small-1", b"hello from file 1".to_vec()),
        ("small-2", b"hello from file 2".to_vec()),
        ("small-3", b"hello from file 3".to_vec()),
    ];

    // Create a large file that exceeds the 1 MB batch body size limit.
    // This should be routed to a single PUT request instead of being batched.
    let large_body: Vec<u8> = vec![0xAB; 2 * 1024 * 1024]; // 2 MB

    let mut many = session.many();

    for (name, content) in &small_bodies {
        let path = dir.path().join(name);
        let mut f = std::fs::File::create(&path).unwrap();
        f.write_all(content).unwrap();
        let file = tokio::fs::File::open(&path).await.unwrap();
        many = many.push(session.put_file(file).compress(None).key(*name));
    }

    let large_path = dir.path().join("large");
    std::fs::File::create(&large_path)
        .unwrap()
        .write_all(&large_body)
        .unwrap();
    let large_file = tokio::fs::File::open(&large_path).await.unwrap();
    many = many.push(session.put_file(large_file).compress(None).key("large"));

    let results: Vec<_> = many.send().await.collect().await;

    assert_eq!(results.len(), 4);

    let mut keys: Vec<String> = results
        .iter()
        .map(|r| match r {
            OperationResult::Put(key, Ok(_)) => key.clone(),
            other => panic!("Expected successful Put result, got: {other:?}"),
        })
        .collect();
    keys.sort();
    assert_eq!(keys, vec!["large", "small-1", "small-2", "small-3"]);

    // Verify all objects are retrievable with the correct content.
    for (name, expected) in &small_bodies {
        let response = session.get(name).send().await.unwrap().unwrap();
        let payload = response.payload().await.unwrap();
        assert_eq!(payload.as_ref(), expected.as_slice(), "mismatch for {name}");
    }

    let response = session.get("large").send().await.unwrap().unwrap();
    let payload = response.payload().await.unwrap();
    assert_eq!(payload.as_ref(), large_body.as_slice());
}

#[tokio::test]
async fn put_path_does_not_exhaust_file_descriptors() {
    // Regression test: put_path must not open file descriptors at push time.
    // On macOS the default ulimit is 256; 1000 concurrent put_file calls would fail.
    let server = test_server().await;
    let client = Client::builder(server.url("/"))
        .token(test_token_generator())
        .build()
        .unwrap();
    let session = client
        .session(Usecase::new("usecase").for_project(12345, 1337))
        .unwrap();

    let dir = tempfile::tempdir().unwrap();
    let mut many = session.many();
    for i in 0..1000 {
        let path = dir.path().join(i.to_string());
        std::fs::File::create(&path).unwrap();
        many = many.push(session.put_path(path).compress(None));
    }
    many.send()
        .await
        .error_for_failures()
        .await
        .unwrap_or_else(|e| panic!("upload failures: {:?}", e.collect::<Vec<_>>()));
}

#[tokio::test]
async fn put_file_with_compression() {
    let server = test_server().await;

    let client = Client::builder(server.url("/"))
        .token(test_token_generator())
        .build()
        .unwrap();
    let usecase = Usecase::new("usecase");
    let session = client.session(usecase.for_organization(12345)).unwrap();

    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("compressed.txt");
    let body = "hello compressed file!";
    std::fs::write(&path, body).unwrap();

    // Default compression is zstd
    let file = tokio::fs::File::open(&path).await.unwrap();
    let stored_id = session.put_file(file).send().await.unwrap().key;

    // When requesting raw, the stored object should be zstd-compressed
    let response = session
        .get(&stored_id)
        .decompress(false)
        .send()
        .await
        .unwrap()
        .unwrap();
    assert_eq!(response.metadata.compression, Some(Compression::Zstd));

    let compressed = response.payload().await.unwrap();
    let decompressed = zstd::bulk::decompress(&compressed, 1024).unwrap();
    assert_eq!(decompressed, body.as_bytes());

    // Default get decompresses transparently
    let response = session.get(&stored_id).send().await.unwrap().unwrap();
    assert_eq!(response.metadata.compression, None);

    let received = response.payload().await.unwrap();
    assert_eq!(received, body);
}

#[tokio::test]
async fn head_returns_metadata() {
    let server = test_server().await;

    let client = Client::builder(server.url("/"))
        .token(test_token_generator())
        .build()
        .unwrap();
    let usecase = Usecase::new("usecase");
    let session = client.session(usecase.for_project(12345, 1337)).unwrap();

    let stored_id = session
        .put("hello head!")
        .compress(None)
        .key("head-test-key")
        .send()
        .await
        .unwrap()
        .key;

    let metadata = session.head(&stored_id).send().await.unwrap();
    assert!(metadata.is_some());
    let metadata = metadata.unwrap();
    assert!(metadata.time_created.is_some());

    let missing = session.head("nonexistent-key").send().await.unwrap();
    assert!(missing.is_none());
}

#[tokio::test]
async fn round_trips_expiration_policy_beyond_a_year() {
    let server = test_server().await;

    let client = Client::builder(server.url("/"))
        .token(test_token_generator())
        .build()
        .unwrap();
    let usecase = Usecase::new("usecase");
    let session = client.session(usecase.for_project(12345, 1337)).unwrap();

    // Serializes as "400d 1h 18m 31s"
    let ttl = Duration::from_secs(400 * 86400 + 4711);
    let policy = ExpirationPolicy::TimeToLive(ttl);

    let stored_id = session
        .put("hello long ttl!")
        .expiration_policy(policy)
        .send()
        .await
        .unwrap()
        .key;

    let metadata = session.head(&stored_id).send().await.unwrap().unwrap();
    assert_eq!(metadata.expiration_policy, policy);
    assert!(metadata.time_expires.is_some());

    let response = session.get(&stored_id).send().await.unwrap().unwrap();
    assert_eq!(response.metadata.expiration_policy, policy);
}

#[tokio::test]
async fn extends_expiry() {
    let server = test_server().await;
    let session = common::test_session(&server);
    let policy = ExpirationPolicy::TimeToLive(Duration::from_secs(86400));
    let key = session
        .put("payload")
        .expiration_policy(policy)
        .send()
        .await
        .unwrap()
        .key;

    let outcome = session
        .extend_expiry(
            &key,
            ExpiryExtension::FromCreation(Duration::from_secs(3 * 86400)),
        )
        .send()
        .await
        .unwrap();
    assert_eq!(outcome, ExtendExpiryResponse::Satisfied);
    let metadata = session.head(&key).send().await.unwrap().unwrap();
    assert_eq!(
        metadata.time_expires,
        Some(metadata.time_created.unwrap() + Duration::from_secs(3 * 86400))
    );
    assert_eq!(
        metadata.expiration_policy,
        ExpirationPolicy::TimeToLive(Duration::from_secs(3 * 86400))
    );

    let outcome = session
        .extend_expiry(
            &key,
            ExpiryExtension::FromCreation(Duration::from_secs(2 * 86400)),
        )
        .send()
        .await
        .unwrap();
    assert_eq!(outcome, ExtendExpiryResponse::Satisfied);
    assert_eq!(session.head(&key).send().await.unwrap().unwrap(), metadata);
    let response = session.get(&key).send().await.unwrap().unwrap();
    assert_eq!(response.payload().await.unwrap(), "payload");
}

#[tokio::test]
async fn extend_expiry_missing_object() {
    let server = test_server().await;
    let session = common::test_session(&server);
    let outcome = session
        .extend_expiry(
            "missing",
            ExpiryExtension::FromNow(Duration::from_secs(86400)),
        )
        .send()
        .await
        .unwrap();
    assert_eq!(outcome, ExtendExpiryResponse::NotFound);
}

#[tokio::test]
async fn extend_expiry_non_expiring_object() {
    let server = test_server().await;
    let session = common::test_session(&server);
    let key = session.put("payload").send().await.unwrap().key;
    let outcome = session
        .extend_expiry(&key, ExpiryExtension::FromNow(Duration::from_secs(86400)))
        .send()
        .await
        .unwrap();
    assert_eq!(outcome, ExtendExpiryResponse::Rejected);
}

#[tokio::test]
async fn batch_head_operations() {
    let server = test_server().await;

    let client = Client::builder(server.url("/"))
        .token(test_token_generator())
        .build()
        .unwrap();
    let usecase = Usecase::new("usecase");
    let session = client.session(usecase.for_project(12345, 1337)).unwrap();

    session
        .many()
        .push(session.put("obj-a").compress(None).key("head-a"))
        .push(session.put("obj-b").compress(None).key("head-b"))
        .send()
        .await
        .error_for_failures()
        .await
        .unwrap_or_else(|e| panic!("setup failures: {:?}", e.collect::<Vec<_>>()));

    let results: Vec<_> = session
        .many()
        .push(session.head("head-a"))
        .push(session.head("head-b"))
        .push(session.head("head-missing"))
        .send()
        .await
        .collect()
        .await;

    assert_eq!(results.len(), 3);

    let mut heads = BTreeMap::new();
    for result in results {
        match result {
            OperationResult::Head(key, inner) => {
                heads.insert(key, inner);
            }
            other => panic!("Expected Head result, got: {other:?}"),
        }
    }

    let head_a = heads.remove("head-a").expect("missing head-a").unwrap();
    assert!(head_a.is_some());
    assert!(head_a.unwrap().time_created.is_some());

    let head_b = heads.remove("head-b").expect("missing head-b").unwrap();
    assert!(head_b.is_some());

    let head_missing = heads
        .remove("head-missing")
        .expect("missing head-missing")
        .unwrap();
    assert!(head_missing.is_none());
}

#[cfg(feature = "resumable-upload-api")]
#[tokio::test]
async fn test_resumable_upload() {
    let server = test_server().await;
    let session = common::test_session(&server);

    // Create a session and upload the first chunk.
    let upload = session
        .create_upload(6)
        .key("resumable-client")
        .content_type("text/plain")
        .compression(None)
        .send()
        .await
        .unwrap()
        .unwrap();
    assert_eq!(upload.key(), "resumable-client");
    assert!(upload.granularity().is_some());
    assert_eq!(
        upload.progress().send().await.unwrap(),
        UploadProgress::Incomplete { offset: 0 }
    );
    assert_eq!(
        upload.put(0, "abc").send().await.unwrap(),
        UploadProgress::Incomplete { offset: 3 }
    );

    // Resume the session and finish the upload.
    let resumed = session.resume_upload(upload.key(), upload.token().clone());
    assert_eq!(resumed.granularity(), None);
    assert_eq!(
        resumed.progress().send().await.unwrap(),
        UploadProgress::Incomplete { offset: 3 }
    );
    assert_eq!(resumed.granularity(), None);
    assert!(matches!(
        resumed.put(0, "bad").send().await.unwrap_err(),
        Error::UploadOffsetMismatch { offset: 3 }
    ));
    assert!(resumed.put(3, "defg").send().await.is_err());
    assert_eq!(
        resumed.put(3, "def").send().await.unwrap(),
        UploadProgress::Complete
    );

    // Verify the published object and terminal session.
    let response = session.get(upload.key()).send().await.unwrap().unwrap();
    assert_eq!(response.metadata.content_type, "text/plain");
    assert_eq!(response.payload().await.unwrap(), "abcdef");
    assert!(resumed.progress().send().await.is_err());
}

#[cfg(feature = "resumable-upload-api")]
#[tokio::test]
async fn test_resumable_upload_streaming() {
    let server = test_server().await;
    let session = common::test_session(&server);

    let compressed = zstd::encode_all(&b"abcdef"[..], 0).unwrap();
    let second_piece_start = compressed.len() / 2;
    let first_piece_len = second_piece_start / 2;
    let upload = session
        .create_upload(compressed.len() as u64)
        .key("resumable-streaming-client")
        .compression(Compression::Zstd)
        .send()
        .await
        .unwrap()
        .unwrap();

    let chunk = futures_util::stream::iter([
        Ok::<_, std::io::Error>(bytes::Bytes::copy_from_slice(
            &compressed[..first_piece_len],
        )),
        Ok(bytes::Bytes::copy_from_slice(
            &compressed[first_piece_len..second_piece_start],
        )),
    ])
    .boxed();
    assert_eq!(
        upload
            .put_stream(0, second_piece_start as u64, chunk)
            .send()
            .await
            .unwrap(),
        UploadProgress::Incomplete {
            offset: second_piece_start as u64
        }
    );

    let reader = std::io::Cursor::new(compressed[second_piece_start..].to_vec());
    assert_eq!(
        upload
            .put_read(
                second_piece_start as u64,
                (compressed.len() - second_piece_start) as u64,
                reader
            )
            .send()
            .await
            .unwrap(),
        UploadProgress::Complete
    );

    let response = session.get(upload.key()).send().await.unwrap().unwrap();
    assert_eq!(response.payload().await.unwrap(), "abcdef");
}

#[cfg(feature = "resumable-upload-api")]
#[tokio::test]
async fn test_resumable_upload_cancel() {
    let server = test_server().await;
    let session = common::test_session(&server);

    let upload = session
        .create_upload(6)
        .key("resumable-cancel")
        .send()
        .await
        .unwrap()
        .unwrap();
    upload.put(0, "abc").send().await.unwrap();
    upload.cancel().send().await.unwrap();

    // A canceled partial upload is no longer available and publishes no object.
    assert!(matches!(
        upload.progress().send().await,
        Err(Error::ResumableUploadUnavailable)
    ));
    assert!(session.get(upload.key()).send().await.unwrap().is_none());
}