shardline-server 1.0.1

#![cfg(feature = "docker")]
#![allow(
    clippy::indexing_slicing,
    clippy::unwrap_used,
    clippy::let_underscore_must_use,
    clippy::shadow_unrelated,
    clippy::expect_used,
    clippy::panic,
    clippy::needless_borrows_for_generic_args,
    clippy::unnecessary_map_or,
    clippy::or_fun_call
)]

use futures_util::{StreamExt, stream};
use sha2::{Digest, Sha256};
use shardline_protocol::{RepositoryProvider, RepositoryScope, TokenClaims, TokenScope};
use shardline_server::{ServerConfig, ServerFrontend, ServerRole, app};
use shardline_server_core::{AuthProvider, auth::LocalHmacProvider};
use shardline_test_support::DockerLocalStack;
use std::{
    num::{NonZeroU64, NonZeroUsize},
    sync::Arc,
    time::Duration,
};
use tempfile::TempDir;
use tokio::net::TcpListener;
use tokio::sync::OnceCell;
use tower::ServiceExt;

// ---------------------------------------------------------------------------
// Shared Docker Postgres — one container for all tests.
// ---------------------------------------------------------------------------

static PG: OnceCell<(DockerLocalStack, String)> = OnceCell::const_new();

/// Ensure the global Docker Postgres is running with migrations applied.
/// Returns the connection URL.
async fn ensure_pg() -> &'static str {
    let (_, url) = PG
        .get_or_init(|| async {
            let stack = DockerLocalStack::builder()
                .with_postgres()
                .start()
                .unwrap()
                .expect("Docker postgres: is docker available?");
            let base = stack.postgres_url().unwrap();
            let url = format!("{base}?sslmode=disable");
            // Run migrations with retry (container may not accept connections immediately).
            let mut last_err = None;
            for _ in 0..5 {
                match sqlx::PgPool::connect(&url).await {
                    Ok(pool) => {
                        shardline_server::apply_database_migrations(&pool)
                            .await
                            .unwrap();
                        pool.close().await;
                        return (stack, url);
                    }
                    Err(e) => {
                        last_err = Some(e);
                        tokio::time::sleep(std::time::Duration::from_millis(500)).await;
                    }
                }
            }
            panic!("Failed to connect to Postgres after 5 retries: {last_err:?}");
        })
        .await;
    url
}

// ---------------------------------------------------------------------------
// Test server harness — starts a real HTTP server on a random port.
// ---------------------------------------------------------------------------

const TEST_SIGNING_KEY: &[u8] = b"0123456789abcdef0123456789abcdef";

struct TestServer {
    shutdown: Option<tokio::sync::oneshot::Sender<()>>,
    base_url: String,
    token: String,
    _tmp: TempDir,
}

impl TestServer {
    /// Starts a full shardline HTTP server with the given frontends, backed by
    /// the shared Docker Postgres instance.
    async fn start(frontends: &[ServerFrontend]) -> Self {
        let pg_url = ensure_pg().await;

        let tmp = TempDir::new().unwrap();
        let chunk_size = NonZeroUsize::new(65536).unwrap();

        let config = ServerConfig::new(
            "127.0.0.1:0".parse().unwrap(),
            "http://127.0.0.1:8080".to_owned(),
            tmp.path().to_path_buf(),
            chunk_size,
        )
        .with_server_role(ServerRole::All)
        .with_server_frontends(frontends.to_vec())
        .unwrap()
        .with_token_signing_key(TEST_SIGNING_KEY.to_vec())
        .unwrap()
        .with_index_postgres_url(pg_url.to_owned())
        .unwrap()
        .with_reconstruction_cache_disabled();

        config.validate_runtime_requirements().unwrap();

        // Build the router and start serving.
        let app = app::router(config).await.unwrap();

        // Bind to a random port via tokio's TcpListener.
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let base_url = format!("http://{addr}");

        let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel::<()>();

        tokio::spawn(async move {
            axum::serve(listener, app)
                .with_graceful_shutdown(async {
                    shutdown_rx.await.ok();
                })
                .await
                .ok();
        });

        // Give the server a moment to start.
        tokio::time::sleep(Duration::from_millis(100)).await;

        // Mint a bearer token with full scope.
        let provider = LocalHmacProvider::new(TEST_SIGNING_KEY).unwrap();
        let repo = RepositoryScope::new(RepositoryProvider::Generic, "test", "test", Some("main"))
            .unwrap();
        let claims =
            TokenClaims::new("shardline", "test", TokenScope::Write, repo, u64::MAX).unwrap();
        let token = provider.mint_token(&claims).unwrap();

        Self {
            shutdown: Some(shutdown_tx),
            base_url,
            token,
            _tmp: tmp,
        }
    }

    fn url(&self, path: &str) -> String {
        format!("{}{}", self.base_url, path)
    }

    fn auth_header(&self) -> &str {
        &self.token
    }
}

impl Drop for TestServer {
    fn drop(&mut self) {
        if let Some(tx) = self.shutdown.take() {
            let _ = tx.send(());
        }
    }
}

// ---------------------------------------------------------------------------
// TestServerBuilder — configurable test server for role-split / provider tests
// ---------------------------------------------------------------------------

struct TestServerBuilder {
    frontends: Vec<ServerFrontend>,
    role: ServerRole,
    provider_config: Option<(TempDir, std::path::PathBuf)>,
}

impl TestServerBuilder {
    fn new(frontends: &[ServerFrontend]) -> Self {
        Self {
            frontends: frontends.to_vec(),
            role: ServerRole::All,
            provider_config: None,
        }
    }

    fn with_provider(mut self) -> Self {
        // Create a temporary provider config file
        let tmp = TempDir::new().unwrap();
        let config_path = tmp.path().join("providers.json");
        let provider_config = serde_json::json!({
            "providers": [{
                "kind": "generic",
                "integration_subject": "test-app",
                "webhook_secret": "test-secret",
                "repositories": [{
                    "owner": "test",
                    "name": "test",
                    "visibility": "private",
                    "default_revision": "main",
                    "clone_url": "https://example.com/test/test.git",
                    "read_subjects": ["test-user"],
                    "write_subjects": ["test-user"]
                }]
            }]
        });
        std::fs::write(&config_path, serde_json::to_vec(&provider_config).unwrap()).unwrap();
        self.provider_config = Some((tmp, config_path));
        self
    }

    async fn start(&mut self) -> TestServer {
        let pg_url = ensure_pg().await;
        let tmp = TempDir::new().unwrap();
        let chunk_size = NonZeroUsize::new(65536).unwrap();

        let mut config = ServerConfig::new(
            "127.0.0.1:0".parse().unwrap(),
            "http://127.0.0.1:8080".to_owned(),
            tmp.path().to_path_buf(),
            chunk_size,
        )
        .with_server_role(self.role)
        .with_server_frontends(self.frontends.clone())
        .unwrap()
        .with_token_signing_key(TEST_SIGNING_KEY.to_vec())
        .unwrap()
        .with_index_postgres_url(pg_url.to_owned())
        .unwrap()
        .with_reconstruction_cache_disabled();

        if let Some((_provider_tmp, config_path)) = self.provider_config.as_ref() {
            config = config
                .with_provider_runtime(
                    config_path.clone(),
                    b"test-api-key".to_vec(),
                    "test-issuer".to_owned(),
                    NonZeroU64::new(3600).unwrap(),
                )
                .unwrap();
        }

        config.validate_runtime_requirements().unwrap();

        let app = app::router(config).await.unwrap();
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let base_url = format!("http://{addr}");

        let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel::<()>();

        tokio::spawn(async move {
            axum::serve(listener, app)
                .with_graceful_shutdown(async {
                    shutdown_rx.await.ok();
                })
                .await
                .ok();
        });

        tokio::time::sleep(Duration::from_millis(100)).await;

        // Mint a bearer token with full scope.
        let provider = LocalHmacProvider::new(TEST_SIGNING_KEY).unwrap();
        let repo = RepositoryScope::new(RepositoryProvider::Generic, "test", "test", Some("main"))
            .unwrap();
        let claims =
            TokenClaims::new("shardline", "test", TokenScope::Write, repo, u64::MAX).unwrap();
        let token = provider.mint_token(&claims).unwrap();

        TestServer {
            shutdown: Some(shutdown_tx),
            base_url,
            token,
            _tmp: tmp,
        }
    }
}

// ===========================================================================
// Tests
// ===========================================================================

/// Helper: compute SHA-256 hex digest.
fn sha256_hex(data: &[u8]) -> String {
    hex::encode(Sha256::digest(data))
}

// ---------------------------------------------------------------------------
// 1. Health endpoint
// ---------------------------------------------------------------------------

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_healthz_returns_200() {
    let server = TestServer::start(&[ServerFrontend::Xet]).await;
    let client = reqwest::Client::new();

    let resp = client.get(server.url("/healthz")).send().await.unwrap();

    assert_eq!(resp.status(), 200);
    let json: serde_json::Value = resp.json().await.unwrap();
    assert_eq!(json["status"], "ok");
}

// ---------------------------------------------------------------------------
// 2. Ready endpoint — shows postgres backend
// ---------------------------------------------------------------------------

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_readyz_returns_postgres_backend() {
    let server = TestServer::start(&[ServerFrontend::Xet]).await;
    let client = reqwest::Client::new();

    let resp = client.get(server.url("/readyz")).send().await.unwrap();

    assert_eq!(resp.status(), 200);
    let json: serde_json::Value = resp.json().await.unwrap();
    assert_eq!(json["status"], "ok");
    assert_eq!(json["metadata_backend"], "postgres");
}

// ---------------------------------------------------------------------------
// 3. Stats endpoint — 401 without auth, 200 with auth
// ---------------------------------------------------------------------------

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_stats_without_auth_returns_401() {
    let server = TestServer::start(&[ServerFrontend::Xet]).await;
    let client = reqwest::Client::new();

    let resp = client.get(server.url("/v1/stats")).send().await.unwrap();

    assert_eq!(resp.status(), 401);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_stats_with_auth_returns_200() {
    let server = TestServer::start(&[ServerFrontend::Xet]).await;
    let client = reqwest::Client::new();

    let resp = client
        .get(server.url("/v1/stats"))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .send()
        .await
        .unwrap();

    assert_eq!(resp.status(), 200);
    let json: serde_json::Value = resp.json().await.unwrap();
    // Fresh backend has zero chunks and files.
    assert_eq!(json["chunks"], 0);
    assert_eq!(json["files"], 0);
}

// ---------------------------------------------------------------------------
// 4. LFS — PUT a file, GET it back, verify content
// ---------------------------------------------------------------------------

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_lfs_put_and_get() {
    let server = TestServer::start(&[ServerFrontend::Lfs]).await;
    let client = reqwest::Client::new();

    let content = b"Hello, LFS over HTTP!";
    let oid = sha256_hex(content);

    // PUT the object.
    let put_resp = client
        .put(server.url(&format!("/v1/lfs/objects/{oid}")))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .header("Content-Type", "application/octet-stream")
        .body(content.to_vec())
        .send()
        .await
        .unwrap();
    assert!(
        put_resp.status().is_success(),
        "LFS PUT failed: {}",
        put_resp.status()
    );

    // GET the object back.
    let get_resp = client
        .get(server.url(&format!("/v1/lfs/objects/{oid}")))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .send()
        .await
        .unwrap();
    assert_eq!(get_resp.status(), 200);
    let body = get_resp.bytes().await.unwrap();
    assert_eq!(body.as_ref(), content);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_large_object_upload_http_e2e() {
    let server = TestServer::start(&[ServerFrontend::Lfs]).await;
    let client = reqwest::Client::new();

    // Create a ~5MB content
    let content = vec![0xAB_u8; 5 * 1024 * 1024];
    let oid = sha256_hex(&content);

    // PUT the large object
    let put_resp = client
        .put(server.url(&format!("/v1/lfs/objects/{oid}")))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .header("Content-Type", "application/octet-stream")
        .body(content.clone())
        .send()
        .await
        .unwrap();
    assert!(
        put_resp.status().is_success(),
        "LFS large object PUT failed: {}",
        put_resp.status()
    );

    // GET the object back and verify
    let get_resp = client
        .get(server.url(&format!("/v1/lfs/objects/{oid}")))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .send()
        .await
        .unwrap();
    assert_eq!(get_resp.status(), 200);
    let body = get_resp.bytes().await.unwrap();
    assert_eq!(body.len(), content.len(), "downloaded size differs");
    assert_eq!(
        body.as_ref(),
        content.as_slice(),
        "downloaded content differs"
    );
}

// ---------------------------------------------------------------------------
// 5. Bazel CAS — PUT content-addressed, GET back, verify
// ---------------------------------------------------------------------------

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_bazel_cas_put_and_get() {
    let server = TestServer::start(&[ServerFrontend::BazelHttp]).await;
    let client = reqwest::Client::new();

    let content = b"Bazel CAS content addressed!";
    let hash = sha256_hex(content);

    // PUT the CAS blob.
    let put_resp = client
        .put(server.url(&format!("/v1/bazel/cache/cas/{hash}")))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .header("Content-Type", "application/octet-stream")
        .body(content.to_vec())
        .send()
        .await
        .unwrap();
    assert!(
        put_resp.status().is_success(),
        "Bazel CAS PUT failed: {}",
        put_resp.status()
    );

    // GET the blob back.
    let get_resp = client
        .get(server.url(&format!("/v1/bazel/cache/cas/{hash}")))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .send()
        .await
        .unwrap();
    assert_eq!(get_resp.status(), 200);
    let body = get_resp.bytes().await.unwrap();
    assert_eq!(body.as_ref(), content);
}

// ---------------------------------------------------------------------------
// 6. OCI — session upload (POST -> PATCH -> PUT), verify
// ---------------------------------------------------------------------------

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_oci_v2_root_returns_200() {
    let server = TestServer::start(&[ServerFrontend::Oci]).await;
    let client = reqwest::Client::builder()
        .redirect(reqwest::redirect::Policy::none())
        .build()
        .unwrap();

    // GET /v2/ should return 200 OK (requires auth).
    let resp = client
        .get(server.url("/v2/"))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
    assert_eq!(
        resp.headers()
            .get("Docker-Distribution-API-Version")
            .and_then(|v| v.to_str().ok()),
        Some("registry/2.0")
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_oci_upload_session() {
    // OCI v2 root and token endpoints work via HTTP, but the wildcard route
    // `/v2/{*path}` has a routing issue with reqwest in this test setup.
    // The monolithic blob upload and blob GET use Axum's built-in
    // `tower::ServiceExt::oneshot` to bypass the HTTP layer and test the full
    // handler/backend stack directly. Health/ready metrics tests above already
    // validate the HTTP layer end-to-end.
    let content = b"OCI session upload content";
    let digest = sha256_hex(content);
    // Repository must match or be under the token scope owner/name (test/test).
    let repo = "test/test";

    // Build a fresh app for oneshot testing.
    let pg_url = ensure_pg().await;
    let tmp = TempDir::new().unwrap();
    let chunk_size = NonZeroUsize::new(65536).unwrap();
    let config = ServerConfig::new(
        "127.0.0.1:0".parse().unwrap(),
        "http://127.0.0.1:8080".to_owned(),
        tmp.path().to_path_buf(),
        chunk_size,
    )
    .with_server_role(ServerRole::All)
    .with_server_frontends([ServerFrontend::Oci])
    .unwrap()
    .with_token_signing_key(TEST_SIGNING_KEY.to_vec())
    .unwrap()
    .with_index_postgres_url(pg_url.to_owned())
    .unwrap()
    .with_reconstruction_cache_disabled();
    config.validate_runtime_requirements().unwrap();

    // Build the same app via the shared router function used by the server
    // binary. This tests the full handler/backend stack including auth,
    // Postgres backend, OCI path parsing, and write path.
    let app = app::router(config).await.unwrap();

    // Mint a token with the same signing key.
    let token = {
        let provider = LocalHmacProvider::new(TEST_SIGNING_KEY).unwrap();
        let repo = RepositoryScope::new(RepositoryProvider::Generic, "test", "test", Some("main"))
            .unwrap();
        let claims =
            TokenClaims::new("shardline", "test", TokenScope::Write, repo, u64::MAX).unwrap();
        provider.mint_token(&claims).unwrap()
    };

    // Monolithic upload via POST with ?digest= query parameter.
    let uri = format!("/v2/{repo}/blobs/uploads/?digest=sha256:{digest}");
    let request = axum::http::Request::builder()
        .method("POST")
        .uri(&uri)
        .header("Authorization", format!("Bearer {token}"))
        .header("Content-Type", "application/octet-stream")
        .body(axum::body::Body::from(content.to_vec()))
        .unwrap();
    let response = app.clone().oneshot(request).await.unwrap();
    let status = response.status();
    let body = axum::body::to_bytes(response.into_body(), usize::MAX)
        .await
        .unwrap();
    assert_eq!(
        status,
        201,
        "OCI monolithic upload (oneshot) failed: status={} body={:?}",
        status,
        String::from_utf8_lossy(&body)
    );

    // Verify: GET the blob and check content (digest needs sha256: prefix for OCI path parser).
    let get_uri = format!("/v2/{repo}/blobs/sha256:{digest}");
    let get_request = axum::http::Request::builder()
        .method("GET")
        .uri(&get_uri)
        .header("Authorization", format!("Bearer {token}"))
        .body(axum::body::Body::empty())
        .unwrap();
    let get_response = app.oneshot(get_request).await.unwrap();
    assert_eq!(get_response.status(), 200);
    let get_body = axum::body::to_bytes(get_response.into_body(), usize::MAX)
        .await
        .unwrap();
    assert_eq!(get_body.as_ref(), content);
}

// ---------------------------------------------------------------------------
// 7. Hub — whoami endpoint
// ---------------------------------------------------------------------------

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_hub_whoami_returns_200() {
    let server = TestServer::start(&[ServerFrontend::Hub]).await;
    let client = reqwest::Client::new();

    let resp = client
        .get(server.url("/api/whoami-v2"))
        .send()
        .await
        .unwrap();

    assert_eq!(resp.status(), 200);
}

// ---------------------------------------------------------------------------
// 8. Metrics — contains expected labels
// ---------------------------------------------------------------------------

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_metrics_contains_expected_content() {
    let server = TestServer::start(&[ServerFrontend::Xet]).await;
    let client = reqwest::Client::new();

    let resp = client.get(server.url("/metrics")).send().await.unwrap();

    assert_eq!(resp.status(), 200);
    let body = resp.text().await.unwrap();
    assert!(
        body.contains("shardline_up 1"),
        "metrics should contain shardline_up gauge"
    );
    assert!(body.contains("# HELP"), "metrics should contain HELP lines");
    assert!(body.contains("# TYPE"), "metrics should contain TYPE lines");
    assert!(
        body.contains("metadata_backend=\"postgres\""),
        "metrics should reference postgres backend: {body}"
    );
}

// ---------------------------------------------------------------------------
// 9. Auth rejection — verify 401 without token
// ---------------------------------------------------------------------------

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_auth_rejection_without_token() {
    let server = TestServer::start(&[ServerFrontend::Xet]).await;
    let client = reqwest::Client::new();

    // Endpoints that require auth should return 401 without a bearer token.
    // Only use endpoints registered by the Xet frontend.
    let endpoints = &[
        "/v1/stats",
        "/v1/reconstructions/nonexistent",
        "/v1/chunks/default/aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
    ];

    for endpoint in endpoints {
        let resp = client.get(server.url(endpoint)).send().await.unwrap();
        assert_eq!(
            resp.status(),
            401,
            "endpoint {endpoint} should return 401 without auth, got {}",
            resp.status()
        );
    }
}

// ---------------------------------------------------------------------------
// 10. All frontends simultaneously — health/ready work with all frontends
// ---------------------------------------------------------------------------

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_all_frontends_health_and_ready() {
    let frontends = &[
        ServerFrontend::Xet,
        ServerFrontend::Lfs,
        ServerFrontend::BazelHttp,
        ServerFrontend::Oci,
        ServerFrontend::Hub,
    ];
    let server = TestServer::start(frontends).await;
    let client = reqwest::Client::new();

    // Health check.
    let resp = client.get(server.url("/healthz")).send().await.unwrap();
    assert_eq!(resp.status(), 200);
    let json: serde_json::Value = resp.json().await.unwrap();
    assert_eq!(json["status"], "ok");

    // Readiness check.
    let resp = client.get(server.url("/readyz")).send().await.unwrap();
    assert_eq!(resp.status(), 200);
    let json: serde_json::Value = resp.json().await.unwrap();
    assert_eq!(json["status"], "ok");
    assert_eq!(json["metadata_backend"], "postgres");
    assert!(json["server_frontends"].as_array().unwrap().len() >= 5);
}

// ===========================================================================
// 11. OCI Manifest + Tag operations (via oneshot — wildcard route limitation)
// ===========================================================================

/// Build a temporary OCI app for oneshot tests. Reuses ensure_pg().
async fn oci_oneshot_app() -> (axum::Router, String) {
    let pg_url = ensure_pg().await;
    let tmp = TempDir::new().unwrap();
    let chunk_size = NonZeroUsize::new(65536).unwrap();
    let config = ServerConfig::new(
        "127.0.0.1:0".parse().unwrap(),
        "http://127.0.0.1:8080".to_owned(),
        tmp.path().to_path_buf(),
        chunk_size,
    )
    .with_server_role(ServerRole::All)
    .with_server_frontends([ServerFrontend::Oci])
    .unwrap()
    .with_token_signing_key(TEST_SIGNING_KEY.to_vec())
    .unwrap()
    .with_index_postgres_url(pg_url.to_owned())
    .unwrap()
    .with_reconstruction_cache_disabled();
    config.validate_runtime_requirements().unwrap();
    let app = app::router(config).await.unwrap();

    let provider = LocalHmacProvider::new(TEST_SIGNING_KEY).unwrap();
    let repo_s =
        RepositoryScope::new(RepositoryProvider::Generic, "test", "test", Some("main")).unwrap();
    let claims =
        TokenClaims::new("shardline", "test", TokenScope::Write, repo_s, u64::MAX).unwrap();
    let token = provider.mint_token(&claims).unwrap();
    // Keep tmp alive by returning it — we leak it via a static or just let it
    // drop (the app already has the PathBuf, so it works for the test duration).
    let _ = Box::new(tmp);
    (app, token)
}

fn oci_digest_hex(data: &[u8]) -> String {
    sha256_hex(data)
}

/// Upload a blob via oneshot monolithic POST. Returns the hex digest.
async fn oci_upload_blob_oneshot(
    app: &axum::Router,
    token: &str,
    repo: &str,
    data: &[u8],
) -> String {
    let digest = oci_digest_hex(data);
    let uri = format!("/v2/{repo}/blobs/uploads/?digest=sha256:{digest}");
    let req = axum::http::Request::builder()
        .method("POST")
        .uri(&uri)
        .header("Authorization", format!("Bearer {token}"))
        .header("Content-Type", "application/octet-stream")
        .body(axum::body::Body::from(data.to_vec()))
        .unwrap();
    let resp = app.clone().oneshot(req).await.unwrap();
    assert_eq!(resp.status(), 201, "blob upload: status={}", resp.status());
    digest
}

fn oci_manifest_json(config_digest: &str, layer_digest: &str) -> String {
    serde_json::json!({
        "schemaVersion": 2,
        "mediaType": "application/vnd.oci.image.manifest.v1+json",
        "config": {
            "mediaType": "application/vnd.oci.image.config.v1+json",
            "size": 0,
            "digest": format!("sha256:{config_digest}")
        },
        "layers": [
            {
                "mediaType": "application/vnd.oci.image.layer.v1.tar+gzip",
                "size": 0,
                "digest": format!("sha256:{layer_digest}")
            }
        ]
    })
    .to_string()
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_oci_manifest_push_and_get_by_tag() {
    let (app, token) = oci_oneshot_app().await;
    let repo = "test/test";
    let tag = "v0.0.1";

    let config_data = b"{}";
    let layer_data = b"\x1f\x8b\x08\x00";
    let config_digest = oci_upload_blob_oneshot(&app, &token, repo, config_data).await;
    let layer_digest = oci_upload_blob_oneshot(&app, &token, repo, layer_data).await;

    let manifest_body = oci_manifest_json(&config_digest, &layer_digest);
    let manifest_digest = oci_digest_hex(manifest_body.as_bytes());

    // PUT manifest by tag
    let put_uri = format!("/v2/{repo}/manifests/{tag}");
    let put_req = axum::http::Request::builder()
        .method("PUT")
        .uri(&put_uri)
        .header("Authorization", format!("Bearer {token}"))
        .header("Content-Type", "application/vnd.oci.image.manifest.v1+json")
        .body(axum::body::Body::from(manifest_body.clone()))
        .unwrap();
    let put_resp = app.clone().oneshot(put_req).await.unwrap();
    assert_eq!(put_resp.status(), 201, "manifest PUT failed");
    let digest_header = put_resp
        .headers()
        .get("Docker-Content-Digest")
        .and_then(|v| v.to_str().ok())
        .map(|s| s.to_owned());
    assert!(digest_header.is_some(), "missing Docker-Content-Digest");
    assert!(
        digest_header.as_ref().unwrap().contains(&manifest_digest),
        "digest mismatch: {:?} vs {}",
        digest_header,
        manifest_digest
    );

    // GET manifest by tag
    let get_uri = format!("/v2/{repo}/manifests/{tag}");
    let get_req = axum::http::Request::builder()
        .method("GET")
        .uri(&get_uri)
        .header("Authorization", format!("Bearer {token}"))
        .body(axum::body::Body::empty())
        .unwrap();
    let get_resp = app.clone().oneshot(get_req).await.unwrap();
    assert_eq!(get_resp.status(), 200);
    let get_body_bytes = axum::body::to_bytes(get_resp.into_body(), usize::MAX)
        .await
        .unwrap();
    let get_json: serde_json::Value = serde_json::from_slice(&get_body_bytes).unwrap();
    assert_eq!(get_json["schemaVersion"], 2);
    assert_eq!(
        get_json["mediaType"],
        "application/vnd.oci.image.manifest.v1+json"
    );
    assert_eq!(
        get_json["config"]["digest"],
        format!("sha256:{config_digest}")
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_oci_manifest_get_by_digest() {
    let (app, token) = oci_oneshot_app().await;
    let repo = "test/test";
    let tag = "digest-ref";

    let config_data = b"{}";
    let layer_data = b"digest-layer-data";
    let config_digest = oci_upload_blob_oneshot(&app, &token, repo, config_data).await;
    let layer_digest = oci_upload_blob_oneshot(&app, &token, repo, layer_data).await;

    let manifest_body = oci_manifest_json(&config_digest, &layer_digest);
    let manifest_digest = oci_digest_hex(manifest_body.as_bytes());

    // Push manifest
    let put_uri = format!("/v2/{repo}/manifests/{tag}");
    let put_req = axum::http::Request::builder()
        .method("PUT")
        .uri(&put_uri)
        .header("Authorization", format!("Bearer {token}"))
        .header("Content-Type", "application/vnd.oci.image.manifest.v1+json")
        .body(axum::body::Body::from(manifest_body))
        .unwrap();
    let put_resp = app.clone().oneshot(put_req).await.unwrap();
    assert_eq!(put_resp.status(), 201);

    // GET manifest by digest
    let get_uri = format!("/v2/{repo}/manifests/sha256:{manifest_digest}");
    let get_req = axum::http::Request::builder()
        .method("GET")
        .uri(&get_uri)
        .header("Authorization", format!("Bearer {token}"))
        .body(axum::body::Body::empty())
        .unwrap();
    let get_resp = app.clone().oneshot(get_req).await.unwrap();
    assert_eq!(get_resp.status(), 200);
    let get_body_bytes = axum::body::to_bytes(get_resp.into_body(), usize::MAX)
        .await
        .unwrap();
    let get_json: serde_json::Value = serde_json::from_slice(&get_body_bytes).unwrap();
    assert_eq!(get_json["schemaVersion"], 2);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_oci_tags_list() {
    let (app, token) = oci_oneshot_app().await;
    let repo = "test/test";
    let tag = "list-me";

    // Push a manifest so a tag exists
    let config_data = b"{}";
    let layer_data = b"tags-layer";
    let config_digest = oci_upload_blob_oneshot(&app, &token, repo, config_data).await;
    let layer_digest = oci_upload_blob_oneshot(&app, &token, repo, layer_data).await;
    let manifest_body = oci_manifest_json(&config_digest, &layer_digest);

    let put_uri = format!("/v2/{repo}/manifests/{tag}");
    let put_req = axum::http::Request::builder()
        .method("PUT")
        .uri(&put_uri)
        .header("Authorization", format!("Bearer {token}"))
        .header("Content-Type", "application/vnd.oci.image.manifest.v1+json")
        .body(axum::body::Body::from(manifest_body))
        .unwrap();
    let put_resp = app.clone().oneshot(put_req).await.unwrap();
    assert_eq!(put_resp.status(), 201);

    // List tags
    let list_uri = format!("/v2/{repo}/tags/list");
    let list_req = axum::http::Request::builder()
        .method("GET")
        .uri(&list_uri)
        .header("Authorization", format!("Bearer {token}"))
        .body(axum::body::Body::empty())
        .unwrap();
    let list_resp = app.clone().oneshot(list_req).await.unwrap();
    assert_eq!(list_resp.status(), 200);
    let list_body: serde_json::Value = serde_json::from_slice(
        &axum::body::to_bytes(list_resp.into_body(), usize::MAX)
            .await
            .unwrap(),
    )
    .unwrap();
    assert_eq!(list_body["name"], repo);
    let tags = list_body["tags"].as_array().unwrap();
    assert!(
        tags.iter().any(|t| t == tag),
        "tag {tag} should appear in list: {tags:?}"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_oci_blob_head() {
    let (app, token) = oci_oneshot_app().await;
    let repo = "test/test";

    let content = b"head-blob-content";
    let digest = oci_upload_blob_oneshot(&app, &token, repo, content).await;

    // HEAD the blob
    let head_uri = format!("/v2/{repo}/blobs/sha256:{digest}");
    let head_req = axum::http::Request::builder()
        .method("HEAD")
        .uri(&head_uri)
        .header("Authorization", format!("Bearer {token}"))
        .body(axum::body::Body::empty())
        .unwrap();
    let head_resp = app.clone().oneshot(head_req).await.unwrap();
    assert_eq!(head_resp.status(), 200);
    let content_len: u64 = head_resp
        .headers()
        .get("content-length")
        .and_then(|v| v.to_str().ok())
        .and_then(|s| s.parse().ok())
        .unwrap_or(0);
    assert_eq!(content_len, content.len() as u64);
    // Docker-Content-Digest header should be present
    assert!(
        head_resp.headers().get("Docker-Content-Digest").is_some(),
        "missing Docker-Content-Digest header"
    );
}

// ===========================================================================
// 12. LFS edge cases
// ===========================================================================

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_lfs_batch_upload_existing() {
    let server = TestServer::start(&[ServerFrontend::Lfs]).await;
    let client = reqwest::Client::new();

    let content = b"lfs-batch-existing-test";
    let oid = sha256_hex(content);

    // Upload the object first
    let put_resp = client
        .put(server.url(&format!("/v1/lfs/objects/{oid}")))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .header("Content-Type", "application/octet-stream")
        .body(content.to_vec())
        .send()
        .await
        .unwrap();
    assert!(put_resp.status().is_success());

    // Batch upload — should report no upload actions for existing object
    let batch_req = serde_json::json!({
        "operation": "upload",
        "objects": [{"oid": oid, "size": content.len() as u64}]
    });
    let batch_resp = client
        .post(server.url("/v1/lfs/objects/batch"))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .header("Content-Type", "application/vnd.git-lfs+json")
        .json(&batch_req)
        .send()
        .await
        .unwrap();
    assert_eq!(batch_resp.status(), 200);
    let batch_json: serde_json::Value = batch_resp.json().await.unwrap();
    let obj = &batch_json["objects"][0];
    assert_eq!(obj["oid"], oid);
    // Existing object should NOT have upload actions
    assert!(
        obj.get("actions").is_none() || obj["actions"].as_object().map_or(true, |m| m.is_empty()),
        "existing object should not have upload actions"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_lfs_batch_download_missing() {
    let server = TestServer::start(&[ServerFrontend::Lfs]).await;
    let client = reqwest::Client::new();

    let fake_oid = "f".repeat(64);
    let batch_req = serde_json::json!({
        "operation": "download",
        "objects": [{"oid": fake_oid, "size": 42}]
    });
    let batch_resp = client
        .post(server.url("/v1/lfs/objects/batch"))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .header("Content-Type", "application/vnd.git-lfs+json")
        .json(&batch_req)
        .send()
        .await
        .unwrap();
    assert_eq!(batch_resp.status(), 200);
    let batch_json: serde_json::Value = batch_resp.json().await.unwrap();
    let obj = &batch_json["objects"][0];
    assert_eq!(obj["oid"], fake_oid);
    // Missing object should have an error
    assert!(
        obj.get("error").is_some(),
        "missing object should have error field: {obj:?}"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_lfs_head_object() {
    let server = TestServer::start(&[ServerFrontend::Lfs]).await;
    let client = reqwest::Client::new();

    let content = b"lfs-head-test";
    let oid = sha256_hex(content);

    // Upload
    client
        .put(server.url(&format!("/v1/lfs/objects/{oid}")))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .header("Content-Type", "application/octet-stream")
        .body(content.to_vec())
        .send()
        .await
        .unwrap();

    // HEAD
    let head_resp = client
        .head(server.url(&format!("/v1/lfs/objects/{oid}")))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .send()
        .await
        .unwrap();
    assert_eq!(head_resp.status(), 200);
    let cl: u64 = head_resp
        .headers()
        .get("content-length")
        .and_then(|v| v.to_str().ok())
        .and_then(|s| s.parse().ok())
        .unwrap_or(0);
    assert_eq!(cl, content.len() as u64);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_lfs_patch_object() {
    let server = TestServer::start(&[ServerFrontend::Lfs]).await;
    let client = reqwest::Client::new();

    let chunk1 = b"lfs-patch-initial-";
    let chunk2 = b"content";
    let full_content = [chunk1.to_vec(), chunk2.to_vec()].concat();
    let oid = sha256_hex(&full_content);
    let total = full_content.len() as u64;

    // PATCH chunk 1
    let range1 = format!("bytes 0-{}/{}", chunk1.len() as u64 - 1, total);
    let patch1 = client
        .patch(server.url(&format!("/v1/lfs/objects/{oid}")))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .header("Content-Type", "application/octet-stream")
        .header("Content-Range", &range1)
        .body(chunk1.to_vec())
        .send()
        .await
        .unwrap();
    assert_eq!(patch1.status(), 200, "PATCH chunk1 failed");

    // PATCH chunk 2 (final)
    let range2 = format!("bytes {}-{}/{}", chunk1.len(), total - 1, total);
    let patch2 = client
        .patch(server.url(&format!("/v1/lfs/objects/{oid}")))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .header("Content-Type", "application/octet-stream")
        .header("Content-Range", &range2)
        .body(chunk2.to_vec())
        .send()
        .await
        .unwrap();
    assert_eq!(patch2.status(), 200, "PATCH chunk2 failed");

    // GET and verify full content
    let get_resp = client
        .get(server.url(&format!("/v1/lfs/objects/{oid}")))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .send()
        .await
        .unwrap();
    assert_eq!(get_resp.status(), 200);
    let body = get_resp.bytes().await.unwrap();
    assert_eq!(body.as_ref(), full_content.as_slice());
}

// ===========================================================================
// 17. Provider token tests
// ===========================================================================

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_provider_issue_token_with_valid_key() {
    let mut builder =
        TestServerBuilder::new(&[ServerFrontend::Xet, ServerFrontend::Oci]).with_provider();
    let server = builder.start().await;
    let client = reqwest::Client::new();

    let resp = client
        .post(server.url("/v1/providers/generic/tokens"))
        .header("x-shardline-provider-key", "test-api-key")
        .header("Content-Type", "application/json")
        .json(&serde_json::json!({
            "subject": "test-user",
            "owner": "test",
            "repo": "test",
            "revision": null,
            "scope": "Read"
        }))
        .send()
        .await
        .unwrap();
    let status = resp.status();
    let body_text = resp.text().await.unwrap_or_default();
    assert_eq!(status, 200, "token issue failed: body={}", body_text);
    let json: serde_json::Value = serde_json::from_str(&body_text).unwrap();
    assert!(!json["token"].as_str().unwrap().is_empty());
    assert_eq!(json["owner"], "test");
    assert_eq!(json["repo"], "test");
    assert_eq!(json["scope"], "Read");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_provider_issue_token_without_api_key_returns_401() {
    let mut builder = TestServerBuilder::new(&[ServerFrontend::Xet]).with_provider();
    let server = builder.start().await;
    let client = reqwest::Client::new();

    let resp = client
        .post(server.url("/v1/providers/generic/tokens"))
        .header("Content-Type", "application/json")
        .json(&serde_json::json!({
            "subject": "test-user",
            "owner": "test",
            "repo": "test",
            "revision": null,
            "scope": "Read"
        }))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 401);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_provider_issue_token_with_invalid_key_returns_403() {
    let mut builder = TestServerBuilder::new(&[ServerFrontend::Xet]).with_provider();
    let server = builder.start().await;
    let client = reqwest::Client::new();

    let resp = client
        .post(server.url("/v1/providers/generic/tokens"))
        .header("x-shardline-provider-key", "wrong-key")
        .header("Content-Type", "application/json")
        .json(&serde_json::json!({
            "subject": "test-user",
            "owner": "test",
            "repo": "test",
            "revision": null,
            "scope": "Read"
        }))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 403);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_provider_git_lfs_authenticate() {
    let mut builder =
        TestServerBuilder::new(&[ServerFrontend::Xet, ServerFrontend::Oci]).with_provider();
    let server = builder.start().await;
    let client = reqwest::Client::new();

    let resp = client
        .post(server.url("/v1/providers/generic/git-lfs-authenticate"))
        .header("x-shardline-provider-key", "test-api-key")
        .header("Content-Type", "application/json")
        .json(&serde_json::json!({
            "subject": "test-user",
            "owner": "test",
            "repo": "test",
            "revision": null,
            "scope": "Read"
        }))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
    let json: serde_json::Value = resp.json().await.unwrap();
    assert!(!json["href"].as_str().unwrap().is_empty());
    assert!(
        !json["header"]["X-Xet-Access-Token"]
            .as_str()
            .unwrap()
            .is_empty()
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_provider_xet_read_token() {
    let mut builder = TestServerBuilder::new(&[ServerFrontend::Xet]).with_provider();
    let server = builder.start().await;
    let client = reqwest::Client::new();

    let resp = client
        .get(server.url("/api/generic/test/test/xet-read-token/main?subject=test-user"))
        .header("x-shardline-provider-key", "test-api-key")
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
    let json: serde_json::Value = resp.json().await.unwrap();
    assert!(!json["accessToken"].as_str().unwrap().is_empty());
    assert!(!json["casUrl"].as_str().unwrap().is_empty());
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_provider_xet_write_token() {
    let mut builder = TestServerBuilder::new(&[ServerFrontend::Xet]).with_provider();
    let server = builder.start().await;
    let client = reqwest::Client::new();

    let resp = client
        .get(server.url("/api/generic/test/test/xet-write-token/main?subject=test-user"))
        .header("x-shardline-provider-key", "test-api-key")
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
    let json: serde_json::Value = resp.json().await.unwrap();
    assert!(!json["accessToken"].as_str().unwrap().is_empty());
    assert!(!json["casUrl"].as_str().unwrap().is_empty());
}

// ===========================================================================
// 18. Role-split tests (OCI)
// ===========================================================================

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_oci_role_api_serves_manifest_but_not_blob_upload() {
    let pg_url = ensure_pg().await;
    let tmp = TempDir::new().unwrap();
    let chunk_size = NonZeroUsize::new(65536).unwrap();
    let config = ServerConfig::new(
        "127.0.0.1:0".parse().unwrap(),
        "http://127.0.0.1:8080".to_owned(),
        tmp.path().to_path_buf(),
        chunk_size,
    )
    .with_server_role(ServerRole::Api)
    .with_server_frontends([ServerFrontend::Oci])
    .unwrap()
    .with_token_signing_key(TEST_SIGNING_KEY.to_vec())
    .unwrap()
    .with_index_postgres_url(pg_url.to_owned())
    .unwrap()
    .with_reconstruction_cache_disabled();
    config.validate_runtime_requirements().unwrap();
    let app = app::router(config).await.unwrap();

    let token = {
        let provider = LocalHmacProvider::new(TEST_SIGNING_KEY).unwrap();
        let repo_s =
            RepositoryScope::new(RepositoryProvider::Generic, "test", "test", Some("main"))
                .unwrap();
        let claims =
            TokenClaims::new("shardline", "test", TokenScope::Write, repo_s, u64::MAX).unwrap();
        provider.mint_token(&claims).unwrap()
    };

    let repo = "test/test";
    let content = b"role-api-blob";
    let digest = sha256_hex(content);

    // Manifest endpoint should work (served by API role)
    let req = axum::http::Request::builder()
        .method("GET")
        .uri(&format!("/v2/{repo}/manifests/nonexistent"))
        .header("Authorization", format!("Bearer {token}"))
        .body(axum::body::Body::empty())
        .unwrap();
    let resp = app.clone().oneshot(req).await.unwrap();
    // 404 because manifest doesn't exist — but the route should be accessible
    assert!(
        resp.status() == 404,
        "API role should route manifest GET, got {}",
        resp.status()
    );

    // Blob upload POST should return 404 (not served by API role)
    let req = axum::http::Request::builder()
        .method("POST")
        .uri(&format!("/v2/{repo}/blobs/uploads/?digest=sha256:{digest}"))
        .header("Authorization", format!("Bearer {token}"))
        .header("Content-Type", "application/octet-stream")
        .body(axum::body::Body::from(content.to_vec()))
        .unwrap();
    let resp = app.clone().oneshot(req).await.unwrap();
    assert_eq!(
        resp.status(),
        404,
        "API role should NOT serve blob upload, got {}",
        resp.status()
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_oci_role_transfer_serves_blob_upload_but_not_manifest() {
    let pg_url = ensure_pg().await;
    let tmp = TempDir::new().unwrap();
    let chunk_size = NonZeroUsize::new(65536).unwrap();
    let config = ServerConfig::new(
        "127.0.0.1:0".parse().unwrap(),
        "http://127.0.0.1:8080".to_owned(),
        tmp.path().to_path_buf(),
        chunk_size,
    )
    .with_server_role(ServerRole::Transfer)
    .with_server_frontends([ServerFrontend::Oci])
    .unwrap()
    .with_token_signing_key(TEST_SIGNING_KEY.to_vec())
    .unwrap()
    .with_index_postgres_url(pg_url.to_owned())
    .unwrap()
    .with_reconstruction_cache_disabled();
    config.validate_runtime_requirements().unwrap();
    let app = app::router(config).await.unwrap();

    let token = {
        let provider = LocalHmacProvider::new(TEST_SIGNING_KEY).unwrap();
        let repo_s =
            RepositoryScope::new(RepositoryProvider::Generic, "test", "test", Some("main"))
                .unwrap();
        let claims =
            TokenClaims::new("shardline", "test", TokenScope::Write, repo_s, u64::MAX).unwrap();
        provider.mint_token(&claims).unwrap()
    };

    let repo = "test/test";
    let content = b"role-transfer-blob";
    let digest = sha256_hex(content);

    // Blob upload POST should work (served by transfer role)
    let req = axum::http::Request::builder()
        .method("POST")
        .uri(&format!("/v2/{repo}/blobs/uploads/?digest=sha256:{digest}"))
        .header("Authorization", format!("Bearer {token}"))
        .header("Content-Type", "application/octet-stream")
        .body(axum::body::Body::from(content.to_vec()))
        .unwrap();
    let resp = app.clone().oneshot(req).await.unwrap();
    assert_eq!(
        resp.status(),
        201,
        "Transfer role should serve blob upload, got {}",
        resp.status()
    );

    // Manifest GET should return 404 (not served by transfer role)
    let req = axum::http::Request::builder()
        .method("GET")
        .uri(&format!("/v2/{repo}/manifests/nonexistent"))
        .header("Authorization", format!("Bearer {token}"))
        .body(axum::body::Body::empty())
        .unwrap();
    let resp = app.clone().oneshot(req).await.unwrap();
    assert_eq!(
        resp.status(),
        404,
        "Transfer role should NOT serve manifest GET, got {}",
        resp.status()
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_oci_role_all_serves_both() {
    let pg_url = ensure_pg().await;
    let tmp = TempDir::new().unwrap();
    let chunk_size = NonZeroUsize::new(65536).unwrap();
    let config = ServerConfig::new(
        "127.0.0.1:0".parse().unwrap(),
        "http://127.0.0.1:8080".to_owned(),
        tmp.path().to_path_buf(),
        chunk_size,
    )
    .with_server_role(ServerRole::All)
    .with_server_frontends([ServerFrontend::Oci])
    .unwrap()
    .with_token_signing_key(TEST_SIGNING_KEY.to_vec())
    .unwrap()
    .with_index_postgres_url(pg_url.to_owned())
    .unwrap()
    .with_reconstruction_cache_disabled();
    config.validate_runtime_requirements().unwrap();
    let app = app::router(config).await.unwrap();

    let token = {
        let provider = LocalHmacProvider::new(TEST_SIGNING_KEY).unwrap();
        let repo_s =
            RepositoryScope::new(RepositoryProvider::Generic, "test", "test", Some("main"))
                .unwrap();
        let claims =
            TokenClaims::new("shardline", "test", TokenScope::Write, repo_s, u64::MAX).unwrap();
        provider.mint_token(&claims).unwrap()
    };

    let repo = "test/test";
    let content = b"role-all-blob";
    let digest = sha256_hex(content);

    // Blob upload should work
    let req = axum::http::Request::builder()
        .method("POST")
        .uri(&format!("/v2/{repo}/blobs/uploads/?digest=sha256:{digest}"))
        .header("Authorization", format!("Bearer {token}"))
        .header("Content-Type", "application/octet-stream")
        .body(axum::body::Body::from(content.to_vec()))
        .unwrap();
    let resp = app.clone().oneshot(req).await.unwrap();
    assert_eq!(
        resp.status(),
        201,
        "All role should serve blob upload, got {}",
        resp.status()
    );

    // Manifest GET should also be routable (return 404 because not found)
    let req = axum::http::Request::builder()
        .method("GET")
        .uri(&format!("/v2/{repo}/manifests/nonexistent"))
        .header("Authorization", format!("Bearer {token}"))
        .body(axum::body::Body::empty())
        .unwrap();
    let resp = app.clone().oneshot(req).await.unwrap();
    assert_eq!(
        resp.status(),
        404,
        "All role should route manifest GET, got {}",
        resp.status()
    );
}

// ===========================================================================
// 23. Reconstruction cache enabled — cold miss then hot hit
// ===========================================================================

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_reconstruction_cache_hit() {
    // Build app with reconstruction cache ENABLED (default is memory).
    let pg_url = ensure_pg().await;
    let tmp = TempDir::new().unwrap();
    let chunk_size = NonZeroUsize::new(65536).unwrap();
    let config = ServerConfig::new(
        "127.0.0.1:0".parse().unwrap(),
        "http://127.0.0.1:8080".to_owned(),
        tmp.path().to_path_buf(),
        chunk_size,
    )
    .with_server_role(ServerRole::All)
    .with_server_frontends([ServerFrontend::Lfs])
    .unwrap()
    .with_token_signing_key(TEST_SIGNING_KEY.to_vec())
    .unwrap()
    .with_index_postgres_url(pg_url.to_owned())
    .unwrap();
    // NOTE: No .with_reconstruction_cache_disabled() — default is memory cache.

    config.validate_runtime_requirements().unwrap();
    let app = app::router(config).await.unwrap();

    let provider = LocalHmacProvider::new(TEST_SIGNING_KEY).unwrap();
    let repo =
        RepositoryScope::new(RepositoryProvider::Generic, "test", "test", Some("main")).unwrap();
    let claims = TokenClaims::new("shardline", "test", TokenScope::Write, repo, u64::MAX).unwrap();
    let token = provider.mint_token(&claims).unwrap();

    let content = b"reconstruction-cache-hit-test";
    let oid = sha256_hex(content);

    // Upload via LFS.
    let put_req = axum::http::Request::builder()
        .method("PUT")
        .uri(&format!("/v1/lfs/objects/{oid}"))
        .header("Authorization", format!("Bearer {token}"))
        .header("Content-Type", "application/octet-stream")
        .body(axum::body::Body::from(content.to_vec()))
        .unwrap();
    let put_resp = app.clone().oneshot(put_req).await.unwrap();
    assert!(
        put_resp.status().is_success(),
        "LFS PUT failed: {}",
        put_resp.status()
    );

    // Read twice — cold miss then hot hit. Both should succeed.
    for i in 0..2 {
        let get_req = axum::http::Request::builder()
            .method("GET")
            .uri(&format!("/v1/lfs/objects/{oid}"))
            .header("Authorization", format!("Bearer {token}"))
            .body(axum::body::Body::empty())
            .unwrap();
        let get_resp = app.clone().oneshot(get_req).await.unwrap();
        assert_eq!(
            get_resp.status(),
            200,
            "GET iteration {i} failed with status {}",
            get_resp.status()
        );
        let body = axum::body::to_bytes(get_resp.into_body(), usize::MAX)
            .await
            .unwrap();
        assert_eq!(body.as_ref(), content, "GET iteration {i} body mismatch");
    }

    // Also verify /readyz reports the cache backend as "memory".
    let ready_req = axum::http::Request::builder()
        .uri("/readyz")
        .body(axum::body::Body::empty())
        .unwrap();
    let ready_resp = app.oneshot(ready_req).await.unwrap();
    assert_eq!(ready_resp.status(), 200);
    let ready_json: serde_json::Value = serde_json::from_slice(
        &axum::body::to_bytes(ready_resp.into_body(), usize::MAX)
            .await
            .unwrap(),
    )
    .unwrap();
    assert_eq!(
        ready_json["cache_backend"], "memory",
        "readyz should report memory cache backend"
    );
}

// ===========================================================================
// 24. 409 Conflict — duplicate Hub repo creation
// ===========================================================================

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_hub_create_same_repo_twice_returns_409() {
    let server = TestServer::start(&[ServerFrontend::Hub]).await;
    let client = reqwest::Client::new();
    let auth = || format!("Bearer {}", server.auth_header());

    let ns = "pg-conflict-team";
    let name = "pg-conflict-model";
    let model_path = format!("{ns}/{name}");

    // First create — should succeed.
    let create1 = client
        .post(server.url("/api/repos/create"))
        .header("Authorization", auth())
        .header("Content-Type", "application/json")
        .json(&serde_json::json!({"type": "model", "name": &model_path, "private": false}))
        .send()
        .await
        .unwrap();
    assert_eq!(create1.status(), 201);

    // Second create (same repo) — should return 409 Conflict.
    let create2 = client
        .post(server.url("/api/repos/create"))
        .header("Authorization", auth())
        .header("Content-Type", "application/json")
        .json(&serde_json::json!({"type": "model", "name": &model_path, "private": false}))
        .send()
        .await
        .unwrap();
    assert_eq!(
        create2.status(),
        409,
        "duplicate repo creation should return 409"
    );
}

// ===========================================================================
// 25. Wrong-repo auth scope — 403 Forbidden
// ===========================================================================

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_auth_wrong_repo_scope_returns_403() {
    // Build an OCI app (OCI validates repository scope against the token).
    let pg_url = ensure_pg().await;
    let tmp = TempDir::new().unwrap();
    let chunk_size = NonZeroUsize::new(65536).unwrap();
    let config = ServerConfig::new(
        "127.0.0.1:0".parse().unwrap(),
        "http://127.0.0.1:8080".to_owned(),
        tmp.path().to_path_buf(),
        chunk_size,
    )
    .with_server_role(ServerRole::All)
    .with_server_frontends([ServerFrontend::Oci])
    .unwrap()
    .with_token_signing_key(TEST_SIGNING_KEY.to_vec())
    .unwrap()
    .with_index_postgres_url(pg_url.to_owned())
    .unwrap()
    .with_reconstruction_cache_disabled();
    config.validate_runtime_requirements().unwrap();
    let app = app::router(config).await.unwrap();

    // Mint a token scoped to a different repo (other/other), not test/test.
    let provider = LocalHmacProvider::new(TEST_SIGNING_KEY).unwrap();
    let other_repo =
        RepositoryScope::new(RepositoryProvider::Generic, "other", "other", Some("main")).unwrap();
    let claims = TokenClaims::new(
        "shardline",
        "other",
        TokenScope::Write,
        other_repo,
        u64::MAX,
    )
    .unwrap();
    let wrong_token = provider.mint_token(&claims).unwrap();

    // Try POST blob upload to test/test with wrong-scoped token.
    // OCI blob upload validates repository scope against the token.
    // The OCI spec prefers 404 over 403 to avoid leaking repository existence.
    let content = b"wrong-scope-content";
    let digest = sha256_hex(content);
    let post_req = axum::http::Request::builder()
        .method("POST")
        .uri(&format!(
            "/v2/test/test/blobs/uploads/?digest=sha256:{digest}"
        ))
        .header("Authorization", format!("Bearer {wrong_token}"))
        .header("Content-Type", "application/octet-stream")
        .body(axum::body::Body::from(content.to_vec()))
        .unwrap();
    let post_resp = app.clone().oneshot(post_req).await.unwrap();

    // Token is scoped to other/other, not test/test → should fail auth/scope check.
    // The exact status depends on implementation: 403 (forbidden) or 404 (not found).
    assert!(
        post_resp.status() == 403 || post_resp.status() == 404,
        "wrong-repo-scope token should get 403 or 404, got {}",
        post_resp.status()
    );
}

// ===========================================================================
// 19. More OCI error paths
// ===========================================================================

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_oci_upload_invalid_digest_algorithm() {
    let (app, token) = oci_oneshot_app().await;
    let repo = "test/test";
    let content = b"some-content";

    // Use sha512 (unsupported) as the digest algorithm
    let req = axum::http::Request::builder()
        .method("POST")
        .uri(&format!(
            "/v2/{repo}/blobs/uploads/?digest=sha512:{}",
            sha256_hex(content)
        ))
        .header("Authorization", format!("Bearer {token}"))
        .header("Content-Type", "application/octet-stream")
        .body(axum::body::Body::from(content.to_vec()))
        .unwrap();
    let resp = app.clone().oneshot(req).await.unwrap();
    assert_eq!(
        resp.status(),
        400,
        "invalid digest algorithm should return 400, got {}",
        resp.status()
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_oci_upload_body_hash_mismatch() {
    let (app, token) = oci_oneshot_app().await;
    let repo = "test/test";

    // Content does not match the digest
    let content = b"actual-content";
    let wrong_digest = sha256_hex(b"different-content");

    let req = axum::http::Request::builder()
        .method("POST")
        .uri(&format!(
            "/v2/{repo}/blobs/uploads/?digest=sha256:{wrong_digest}"
        ))
        .header("Authorization", format!("Bearer {token}"))
        .header("Content-Type", "application/octet-stream")
        .body(axum::body::Body::from(content.to_vec()))
        .unwrap();
    let resp = app.clone().oneshot(req).await.unwrap();
    assert_eq!(
        resp.status(),
        400,
        "body hash mismatch should return 400, got {}",
        resp.status()
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_oci_patch_nonexistent_session() {
    let (app, token) = oci_oneshot_app().await;
    let repo = "test/test";

    let req = axum::http::Request::builder()
        .method("PATCH")
        .uri(&format!(
            "/v2/{repo}/blobs/uploads/00000000-0000-0000-0000-000000000000"
        ))
        .header("Authorization", format!("Bearer {token}"))
        .header("Content-Type", "application/octet-stream")
        .header("Content-Range", "0-63/64")
        .body(axum::body::Body::from(b"data".to_vec()))
        .unwrap();
    let resp = app.clone().oneshot(req).await.unwrap();
    let status = resp.status();
    let body_bytes = axum::body::to_bytes(resp.into_body(), usize::MAX)
        .await
        .unwrap();
    let body_str = String::from_utf8_lossy(&body_bytes);
    assert_eq!(
        status, 404,
        "PATCH non-existent session: status={} body={}",
        status, body_str
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_oci_put_finalize_without_digest() {
    let (app, token) = oci_oneshot_app().await;
    let repo = "test/test";

    // PUT to blob uploads without ?digest= query parameter
    let req = axum::http::Request::builder()
        .method("PUT")
        .uri(&format!(
            "/v2/{repo}/blobs/uploads/00000000-0000-0000-0000-000000000000"
        ))
        .header("Authorization", format!("Bearer {token}"))
        .body(axum::body::Body::empty())
        .unwrap();
    let resp = app.clone().oneshot(req).await.unwrap();
    assert_eq!(
        resp.status(),
        400,
        "PUT without digest should return 400, got {}",
        resp.status()
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_oci_get_nonexistent_blob() {
    let (app, token) = oci_oneshot_app().await;
    let repo = "test/test";
    let fake_digest = "a".repeat(64);

    let req = axum::http::Request::builder()
        .method("GET")
        .uri(&format!("/v2/{repo}/blobs/sha256:{fake_digest}"))
        .header("Authorization", format!("Bearer {token}"))
        .body(axum::body::Body::empty())
        .unwrap();
    let resp = app.clone().oneshot(req).await.unwrap();
    assert_eq!(
        resp.status(),
        404,
        "GET non-existent blob should return 404, got {}",
        resp.status()
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_oci_delete_nonexistent_manifest() {
    let (app, token) = oci_oneshot_app().await;
    let repo = "test/test";

    let req = axum::http::Request::builder()
        .method("DELETE")
        .uri(&format!("/v2/{repo}/manifests/nonexistent-tag"))
        .header("Authorization", format!("Bearer {token}"))
        .body(axum::body::Body::empty())
        .unwrap();
    let resp = app.clone().oneshot(req).await.unwrap();
    assert_eq!(
        resp.status(),
        404,
        "DELETE non-existent manifest should return 404, got {}",
        resp.status()
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_oci_push_manifest_with_nonexistent_blob() {
    let (app, token) = oci_oneshot_app().await;
    let repo = "test/test";

    let manifest_body = serde_json::json!({
        "schemaVersion": 2,
        "mediaType": "application/vnd.oci.image.manifest.v1+json",
        "config": {
            "mediaType": "application/vnd.oci.image.config.v1+json",
            "size": 0,
            "digest": "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
        },
        "layers": [
            {
                "mediaType": "application/vnd.oci.image.layer.v1.tar+gzip",
                "size": 0,
                "digest": "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
            }
        ]
    })
    .to_string();

    let req = axum::http::Request::builder()
        .method("PUT")
        .uri(&format!("/v2/{repo}/manifests/nonexistent-blob-ref"))
        .header("Authorization", format!("Bearer {token}"))
        .header("Content-Type", "application/vnd.oci.image.manifest.v1+json")
        .body(axum::body::Body::from(manifest_body))
        .unwrap();
    let resp = app.clone().oneshot(req).await.unwrap();
    assert_eq!(
        resp.status(),
        400,
        "manifest referencing non-existent blob should return 400, got {}",
        resp.status()
    );
}

// ===========================================================================
// 20. More LFS edge cases
// ===========================================================================

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_lfs_verify_object() {
    let server = TestServer::start(&[ServerFrontend::Lfs]).await;
    let client = reqwest::Client::new();

    let content = b"verify-me-content";
    let oid = sha256_hex(content);

    // Upload
    client
        .put(server.url(&format!("/v1/lfs/objects/{oid}")))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .header("Content-Type", "application/octet-stream")
        .body(content.to_vec())
        .send()
        .await
        .unwrap();

    // Verify
    let verify_resp = client
        .post(server.url(&format!("/v1/lfs/objects/{oid}/verify")))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .send()
        .await
        .unwrap();
    assert_eq!(
        verify_resp.status(),
        200,
        "verify existing object should return 200, got {}",
        verify_resp.status()
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_lfs_verify_nonexistent_object() {
    let server = TestServer::start(&[ServerFrontend::Lfs]).await;
    let client = reqwest::Client::new();

    let fake_oid = "f".repeat(64);

    let verify_resp = client
        .post(server.url(&format!("/v1/lfs/objects/{fake_oid}/verify")))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .send()
        .await
        .unwrap();
    assert_eq!(
        verify_resp.status(),
        404,
        "verify non-existent should return 404, got {}",
        verify_resp.status()
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_lfs_batch_mixed_existing_missing() {
    let server = TestServer::start(&[ServerFrontend::Lfs]).await;
    let client = reqwest::Client::new();

    let existing_content = b"existing-batch-content";
    let existing_oid = sha256_hex(existing_content);

    // Upload existing object
    client
        .put(server.url(&format!("/v1/lfs/objects/{existing_oid}")))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .header("Content-Type", "application/octet-stream")
        .body(existing_content.to_vec())
        .send()
        .await
        .unwrap();

    let missing_oid = "c".repeat(64);

    // Batch with mixed
    let batch_req = serde_json::json!({
        "operation": "download",
        "objects": [
            {"oid": existing_oid, "size": existing_content.len() as u64},
            {"oid": missing_oid, "size": 42}
        ]
    });
    let batch_resp = client
        .post(server.url("/v1/lfs/objects/batch"))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .header("Content-Type", "application/vnd.git-lfs+json")
        .json(&batch_req)
        .send()
        .await
        .unwrap();
    assert_eq!(batch_resp.status(), 200);
    let json: serde_json::Value = batch_resp.json().await.unwrap();
    let objects = json["objects"].as_array().unwrap();
    assert_eq!(objects.len(), 2);
    // Existing object should not have error
    assert!(
        objects[0].get("error").is_none() || objects[0]["error"].is_null(),
        "existing object should have no error: {:?}",
        objects[0]
    );
    // Missing object should have error
    assert!(
        objects[1].get("error").is_some(),
        "missing object should have error: {:?}",
        objects[1]
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_lfs_upload_empty_file() {
    let server = TestServer::start(&[ServerFrontend::Lfs]).await;
    let client = reqwest::Client::new();

    let content = b"";
    let oid = sha256_hex(content);

    let put_resp = client
        .put(server.url(&format!("/v1/lfs/objects/{oid}")))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .header("Content-Type", "application/octet-stream")
        .body(content.to_vec())
        .send()
        .await
        .unwrap();
    assert!(
        put_resp.status().is_success(),
        "empty file upload should succeed"
    );

    let get_resp = client
        .get(server.url(&format!("/v1/lfs/objects/{oid}")))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .send()
        .await
        .unwrap();
    assert_eq!(get_resp.status(), 200);
    let body = get_resp.bytes().await.unwrap();
    assert_eq!(body.as_ref(), content);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_lfs_batch_with_large_size() {
    let server = TestServer::start(&[ServerFrontend::Lfs]).await;
    let client = reqwest::Client::new();

    let content = b"large-header-content";
    let oid = sha256_hex(content);

    // Upload first
    client
        .put(server.url(&format!("/v1/lfs/objects/{oid}")))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .header("Content-Type", "application/octet-stream")
        .body(content.to_vec())
        .send()
        .await
        .unwrap();

    // Batch with 100MB size declared
    let batch_req = serde_json::json!({
        "operation": "download",
        "objects": [{"oid": oid, "size": 100_000_000}]
    });
    let batch_resp = client
        .post(server.url("/v1/lfs/objects/batch"))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .header("Content-Type", "application/vnd.git-lfs+json")
        .json(&batch_req)
        .send()
        .await
        .unwrap();
    assert_eq!(batch_resp.status(), 200);
}

// ===========================================================================
// 21. More Hub API routes (postgres only)
// ===========================================================================

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_hub_create_dataset_repo() {
    let server = TestServer::start(&[ServerFrontend::Hub]).await;
    let client = reqwest::Client::new();
    let auth = || format!("Bearer {}", server.auth_header());

    let ns = "ds-team";
    let name = "ds-repo";
    let ds_path = format!("{ns}/{name}");

    let create_resp = client
        .post(server.url("/api/repos/create"))
        .header("Authorization", auth())
        .header("Content-Type", "application/json")
        .json(&serde_json::json!({
            "type": "dataset",
            "name": &ds_path,
            "private": false,
        }))
        .send()
        .await
        .unwrap();
    assert_eq!(create_resp.status(), 201);

    let info_resp = client
        .get(server.url(&format!("/api/datasets/{ns}/{name}")))
        .header("Authorization", auth())
        .send()
        .await
        .unwrap();
    assert_eq!(info_resp.status(), 200);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_hub_create_space_repo() {
    let server = TestServer::start(&[ServerFrontend::Hub]).await;
    let client = reqwest::Client::new();
    let auth = || format!("Bearer {}", server.auth_header());

    let ns = "space-team";
    let name = "space-app";
    let sp_path = format!("{ns}/{name}");

    let create_resp = client
        .post(server.url("/api/repos/create"))
        .header("Authorization", auth())
        .header("Content-Type", "application/json")
        .json(&serde_json::json!({
            "type": "space",
            "name": &sp_path,
            "private": false,
        }))
        .send()
        .await
        .unwrap();
    assert_eq!(create_resp.status(), 201);

    let info_resp = client
        .get(server.url(&format!("/api/spaces/{ns}/{name}")))
        .header("Authorization", auth())
        .send()
        .await
        .unwrap();
    assert_eq!(info_resp.status(), 200);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_hub_revisions_list_json_array() {
    let server = TestServer::start(&[ServerFrontend::Hub]).await;
    let client = reqwest::Client::new();
    let auth = || format!("Bearer {}", server.auth_header());

    let ns = "rev-team";
    let name = "rev-model";
    let path = format!("{ns}/{name}");

    // Create repo
    client
        .post(server.url("/api/repos/create"))
        .header("Authorization", auth())
        .header("Content-Type", "application/json")
        .json(&serde_json::json!({"type": "model", "name": &path, "private": false}))
        .send()
        .await
        .unwrap();

    // Commit a file
    let content_b64 = base64::Engine::encode(
        &base64::engine::general_purpose::STANDARD,
        b"rev-test-content",
    );
    let ndjson = format!(
        "{{\"header\":{{\"message\":\"rev commit\",\"parentCommit\":\"\"}}}}\n\
         {{\"file\":{{\"path\":\"test.txt\",\"content\":\"{content_b64}\"}}}}"
    );
    let commit_resp = client
        .post(server.url(&format!("/api/models/{ns}/{name}/commit/main")))
        .header("Authorization", auth())
        .header("Content-Type", "application/x-ndjson")
        .body(ndjson)
        .send()
        .await
        .unwrap();
    assert_eq!(commit_resp.status(), 200);

    // Get revisions list — verify it returns a JSON object with "revisions" key
    let rev_resp = client
        .get(server.url(&format!("/api/models/{ns}/{name}/revisions")))
        .header("Authorization", auth())
        .send()
        .await
        .unwrap();
    assert_eq!(rev_resp.status(), 200);
    let rev_json: serde_json::Value = rev_resp.json().await.unwrap();
    // Revisions should be in a "revisions" array
    assert!(
        rev_json.get("revisions").is_some(),
        "revisions response should have 'revisions' key: {rev_json}"
    );
    let revisions = rev_json["revisions"].as_array().unwrap();
    assert!(!revisions.is_empty(), "should have at least one revision");
}

// ===========================================================================
// 22. More Bazel edge cases
// ===========================================================================

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_bazel_ac_put_and_get() {
    let server = TestServer::start(&[ServerFrontend::BazelHttp]).await;
    let client = reqwest::Client::new();

    let content = b"action-cache-content";
    let hash = sha256_hex(content);

    // PUT AC
    let put_resp = client
        .put(server.url(&format!("/v1/bazel/cache/ac/{hash}")))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .header("Content-Type", "application/octet-stream")
        .body(content.to_vec())
        .send()
        .await
        .unwrap();
    assert!(
        put_resp.status().is_success(),
        "AC PUT failed: {}",
        put_resp.status()
    );

    // GET AC
    let get_resp = client
        .get(server.url(&format!("/v1/bazel/cache/ac/{hash}")))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .send()
        .await
        .unwrap();
    assert_eq!(get_resp.status(), 200);
    let body = get_resp.bytes().await.unwrap();
    assert_eq!(body.as_ref(), content);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_bazel_head_existing() {
    let server = TestServer::start(&[ServerFrontend::BazelHttp]).await;
    let client = reqwest::Client::new();

    let content = b"head-test-content";
    let hash = sha256_hex(content);

    // Upload first
    client
        .put(server.url(&format!("/v1/bazel/cache/cas/{hash}")))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .header("Content-Type", "application/octet-stream")
        .body(content.to_vec())
        .send()
        .await
        .unwrap();

    // HEAD existing
    let head_resp = client
        .head(server.url(&format!("/v1/bazel/cache/cas/{hash}")))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .send()
        .await
        .unwrap();
    assert_eq!(head_resp.status(), 200);
    let cl: u64 = head_resp
        .headers()
        .get("content-length")
        .and_then(|v| v.to_str().ok())
        .and_then(|s| s.parse().ok())
        .unwrap_or(0);
    assert_eq!(cl, content.len() as u64);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_bazel_head_nonexistent() {
    let server = TestServer::start(&[ServerFrontend::BazelHttp]).await;
    let client = reqwest::Client::new();

    let fake_hash = "f".repeat(64);

    let head_resp = client
        .head(server.url(&format!("/v1/bazel/cache/cas/{fake_hash}")))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .send()
        .await
        .unwrap();
    assert_eq!(
        head_resp.status(),
        404,
        "HEAD non-existent should return 404, got {}",
        head_resp.status()
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_bazel_cas_content_hash_mismatch() {
    let server = TestServer::start(&[ServerFrontend::BazelHttp]).await;
    let client = reqwest::Client::new();

    let content = b"actual-body-data";
    let wrong_hash = sha256_hex(b"different-body");

    let put_resp = client
        .put(server.url(&format!("/v1/bazel/cache/cas/{wrong_hash}")))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .header("Content-Type", "application/octet-stream")
        .body(content.to_vec())
        .send()
        .await
        .unwrap();
    assert_eq!(
        put_resp.status(),
        400,
        "content hash mismatch should return 400, got {}",
        put_resp.status()
    );
}

// ===========================================================================
// 13. Hub API routes beyond whoami
// ===========================================================================

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_hub_create_repo_and_commit() {
    let server = TestServer::start(&[ServerFrontend::Hub]).await;
    let client = reqwest::Client::new();
    let auth = || format!("Bearer {}", server.auth_header());

    let ns = "e2e-team";
    let name = "e2e-model";
    let model_path = format!("{ns}/{name}");

    // Create a model repo
    let create_resp = client
        .post(server.url("/api/repos/create"))
        .header("Authorization", auth())
        .header("Content-Type", "application/json")
        .json(&serde_json::json!({
            "type": "model",
            "name": &model_path,
            "private": false,
        }))
        .send()
        .await
        .unwrap();
    assert_eq!(create_resp.status(), 201);

    // Get repo info
    let info_resp = client
        .get(server.url(&format!("/api/models/{ns}/{name}")))
        .header("Authorization", auth())
        .send()
        .await
        .unwrap();
    assert_eq!(info_resp.status(), 200);
    let info_json: serde_json::Value = info_resp.json().await.unwrap();
    assert_eq!(info_json["id"], model_path);
    assert_eq!(info_json["type"], "model");

    // Commit a file using NDJSON
    let content_b64 = base64::Engine::encode(
        &base64::engine::general_purpose::STANDARD,
        b"Hello from e2e test!",
    );
    let ndjson = format!(
        "{{\"header\":{{\"message\":\"e2e commit\",\"parentCommit\":\"\"}}}}\n\
         {{\"file\":{{\"path\":\"hello.txt\",\"content\":\"{content_b64}\"}}}}"
    );
    let commit_resp = client
        .post(server.url(&format!("/api/models/{ns}/{name}/commit/main")))
        .header("Authorization", auth())
        .header("Content-Type", "application/x-ndjson")
        .body(ndjson)
        .send()
        .await
        .unwrap();
    assert_eq!(commit_resp.status(), 200);
    let commit_json: serde_json::Value = commit_resp.json().await.unwrap();
    assert!(
        commit_json
            .get("commitId")
            .and_then(|v| v.as_str())
            .is_some()
            || commit_json
                .get("commit_id")
                .and_then(|v| v.as_str())
                .is_some(),
        "commit response should contain commit_id: {commit_json}"
    );

    // Verify the commit created a new revision
    let rev_resp = client
        .get(server.url(&format!("/api/models/{ns}/{name}/revisions")))
        .header("Authorization", auth())
        .send()
        .await
        .unwrap();
    assert_eq!(rev_resp.status(), 200);
    let rev_json: serde_json::Value = rev_resp.json().await.unwrap();
    let revisions = rev_json["revisions"]
        .as_array()
        .unwrap_or(&rev_json.as_array().cloned().unwrap_or_default())
        .clone();
    // Should have at least one revision after the commit
    assert!(
        !revisions.is_empty(),
        "should have at least one revision after commit"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_hub_upload_lfs_and_batch() {
    let server = TestServer::start(&[ServerFrontend::Hub]).await;
    let client = reqwest::Client::new();
    let auth = || format!("Bearer {}", server.auth_header());

    let content = b"hub-lfs-e2e-content";
    let oid = sha256_hex(content);

    // Upload via Hub LFS route (not /v1/lfs — Hub has its own /lfs/objects/{oid})
    let put_resp = client
        .put(server.url(&format!("/lfs/objects/{oid}")))
        .header("Authorization", auth())
        .header("Content-Type", "application/octet-stream")
        .body(content.to_vec())
        .send()
        .await
        .unwrap();
    assert_eq!(put_resp.status(), 200);

    // Download back
    let get_resp = client
        .get(server.url(&format!("/lfs/objects/{oid}")))
        .header("Authorization", auth())
        .send()
        .await
        .unwrap();
    assert_eq!(get_resp.status(), 200);
    let body = get_resp.bytes().await.unwrap();
    assert_eq!(body.as_ref(), content);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_hub_modelcard() {
    let server = TestServer::start(&[ServerFrontend::Hub]).await;
    let client = reqwest::Client::new();
    let auth = || format!("Bearer {}", server.auth_header());

    let ns = "card-team";
    let name = "card-model";
    let model_path = format!("{ns}/{name}");

    // Create a model repo
    let create_resp = client
        .post(server.url("/api/repos/create"))
        .header("Authorization", auth())
        .header("Content-Type", "application/json")
        .json(&serde_json::json!({
            "type": "model",
            "name": &model_path,
            "private": false,
        }))
        .send()
        .await
        .unwrap();
    assert_eq!(create_resp.status(), 201);

    // Before committing, modelcard should 404
    let card_resp = client
        .get(server.url(&format!("/api/models/{ns}/{name}/modelcard")))
        .header("Authorization", auth())
        .send()
        .await
        .unwrap();
    assert_eq!(card_resp.status(), 404);

    // Commit a README.md
    let readme_b64 = base64::Engine::encode(
        &base64::engine::general_purpose::STANDARD,
        b"# Model Card\n\nThis is a test model.",
    );
    let ndjson = format!(
        "{{\"header\":{{\"message\":\"add readme\",\"parentCommit\":\"\"}}}}\n\
         {{\"file\":{{\"path\":\"README.md\",\"content\":\"{readme_b64}\"}}}}"
    );
    let commit_resp = client
        .post(server.url(&format!("/api/models/{ns}/{name}/commit/main")))
        .header("Authorization", auth())
        .header("Content-Type", "application/x-ndjson")
        .body(ndjson)
        .send()
        .await
        .unwrap();
    assert_eq!(commit_resp.status(), 200);

    // Now modelcard should return the README
    let card_resp = client
        .get(server.url(&format!("/api/models/{ns}/{name}/modelcard")))
        .header("Authorization", auth())
        .send()
        .await
        .unwrap();
    assert_eq!(card_resp.status(), 200);
}

// ===========================================================================
// 14. Error / edge cases
// ===========================================================================

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_error_no_auth_on_protected_endpoints() {
    let server = TestServer::start(&[ServerFrontend::Lfs]).await;
    let client = reqwest::Client::new();

    // POST /v1/lfs/objects/batch without auth → 401
    let resp = client
        .post(server.url("/v1/lfs/objects/batch"))
        .header("Content-Type", "application/vnd.git-lfs+json")
        .json(&serde_json::json!({"operation":"download","objects":[]}))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 401);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_error_invalid_token() {
    let server = TestServer::start(&[ServerFrontend::Xet]).await;
    let client = reqwest::Client::new();

    let resp = client
        .get(server.url("/v1/stats"))
        .header("Authorization", "Bearer invalid.token.here")
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 401);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_error_oci_manifest_not_found() {
    let (app, token) = oci_oneshot_app().await;

    let get_uri = "/v2/test/test/manifests/nonexistent-tag";
    let get_req = axum::http::Request::builder()
        .method("GET")
        .uri(get_uri)
        .header("Authorization", format!("Bearer {token}"))
        .body(axum::body::Body::empty())
        .unwrap();
    let get_resp = app.clone().oneshot(get_req).await.unwrap();
    assert_eq!(get_resp.status(), 404);
    let body = axum::body::to_bytes(get_resp.into_body(), usize::MAX)
        .await
        .unwrap();
    let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
    assert_eq!(json["errors"][0]["code"], "MANIFEST_UNKNOWN");
}

// ===========================================================================
// 15. Metrics after operations
// ===========================================================================

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_metrics_after_lfs_upload() {
    let server = TestServer::start(&[ServerFrontend::Lfs]).await;
    let client = reqwest::Client::new();

    // Check metrics baseline — after upload there should be some transfer counters
    let content = b"metrics-check-content";
    let oid = sha256_hex(content);

    // Upload via LFS
    let put_resp = client
        .put(server.url(&format!("/v1/lfs/objects/{oid}")))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .header("Content-Type", "application/octet-stream")
        .body(content.to_vec())
        .send()
        .await
        .unwrap();
    assert!(put_resp.status().is_success());

    // Fetch metrics and verify it contains upload-related labels
    let metrics_resp = client.get(server.url("/metrics")).send().await.unwrap();
    assert_eq!(metrics_resp.status(), 200);
    let body = metrics_resp.text().await.unwrap();
    // The metrics should still have the basic shardline gauges
    assert!(
        body.contains("shardline_up 1"),
        "shardline_up should be present"
    );
}

// ===========================================================================
// 16. Cross-protocol: upload via LFS, read via OCI blob
// ===========================================================================

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_cross_protocol_lfs_to_oci() {
    let server = TestServer::start(&[ServerFrontend::Lfs, ServerFrontend::Oci]).await;
    let client = reqwest::Client::new();

    let content = b"cross-protocol-content";
    let oid = sha256_hex(content);

    // Upload via LFS (v1 API)
    let put_resp = client
        .put(server.url(&format!("/v1/lfs/objects/{oid}")))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .header("Content-Type", "application/octet-stream")
        .body(content.to_vec())
        .send()
        .await
        .unwrap();
    assert!(put_resp.status().is_success());

    // Read the same object via a GET request (not OCI — just confirm it's there)
    let get_resp = client
        .get(server.url(&format!("/v1/lfs/objects/{oid}")))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .send()
        .await
        .unwrap();
    assert_eq!(get_resp.status(), 200);
    let body = get_resp.bytes().await.unwrap();
    assert_eq!(body.as_ref(), content);
}

// ===========================================================================
// 23. Concurrent HTTP E2E tests
// ===========================================================================

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_concurrent_lfs_upload_same_oid() {
    let server = TestServer::start(&[ServerFrontend::Lfs]).await;
    let content = b"concurrent-same-oid-content-pg";
    let oid = sha256_hex(content);
    let url = server.url(&format!("/v1/lfs/objects/{oid}"));
    let token = server.auth_header().to_owned();

    // 5 concurrent PUTs of the same LFS OID — all should return 200
    let mut handles = Vec::new();
    for _ in 0..5 {
        let url = url.clone();
        let token = token.clone();
        let data = content.to_vec();
        handles.push(tokio::spawn(async move {
            let client = reqwest::Client::new();
            let resp = client
                .put(&url)
                .header("Authorization", format!("Bearer {token}"))
                .header("Content-Type", "application/octet-stream")
                .body(data)
                .send()
                .await
                .unwrap();
            resp.status()
        }));
    }

    for (i, handle) in handles.into_iter().enumerate() {
        let status = handle.await.unwrap();
        assert!(
            status.is_success(),
            "concurrent PUT {} returned {}",
            i,
            status
        );
    }

    // Verify the stored content is correct
    let client = reqwest::Client::new();
    let get_resp = client
        .get(server.url(&format!("/v1/lfs/objects/{oid}")))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .send()
        .await
        .unwrap();
    assert_eq!(get_resp.status(), 200);
    assert_eq!(get_resp.bytes().await.unwrap().as_ref(), content);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_concurrent_oci_manifest_push_and_pull() {
    let (app, token) = oci_oneshot_app().await;
    let repo = "test/test";
    let tag = "concurrent-manifest-pg";

    // Prepare data: upload config and layer blobs
    let config_data = b"{}";
    let layer_data = b"concurrent-layer-pg";
    let config_digest = oci_upload_blob_oneshot(&app, &token, repo, config_data).await;
    let layer_digest = oci_upload_blob_oneshot(&app, &token, repo, layer_data).await;
    let manifest_body = oci_manifest_json(&config_digest, &layer_digest);

    // Push the manifest first
    let put_uri = format!("/v2/{repo}/manifests/{tag}");
    let put_req = axum::http::Request::builder()
        .method("PUT")
        .uri(&put_uri)
        .header("Authorization", format!("Bearer {token}"))
        .header("Content-Type", "application/vnd.oci.image.manifest.v1+json")
        .body(axum::body::Body::from(manifest_body.clone()))
        .unwrap();
    let put_resp = app.clone().oneshot(put_req).await.unwrap();
    assert_eq!(put_resp.status(), 201, "manifest PUT failed");

    // Spawn 3 concurrent GETs
    let get_uri = format!("/v2/{repo}/manifests/{tag}");
    let g1 = {
        let app = app.clone();
        let token = token.clone();
        let uri = get_uri.clone();
        tokio::spawn(async move {
            let req = axum::http::Request::builder()
                .method("GET")
                .uri(&uri)
                .header("Authorization", format!("Bearer {token}"))
                .body(axum::body::Body::empty())
                .unwrap();
            app.clone().oneshot(req).await
        })
    };
    let g2 = {
        let app = app.clone();
        let token = token.clone();
        let uri = get_uri.clone();
        tokio::spawn(async move {
            let req = axum::http::Request::builder()
                .method("GET")
                .uri(&uri)
                .header("Authorization", format!("Bearer {token}"))
                .body(axum::body::Body::empty())
                .unwrap();
            app.clone().oneshot(req).await
        })
    };
    let g3 = {
        let app = app.clone();
        let token = token.clone();
        tokio::spawn(async move {
            let req = axum::http::Request::builder()
                .method("GET")
                .uri(&get_uri)
                .header("Authorization", format!("Bearer {token}"))
                .body(axum::body::Body::empty())
                .unwrap();
            app.clone().oneshot(req).await
        })
    };

    let (r1, r2, r3) = tokio::join!(g1, g2, g3);
    for (i, result) in [r1, r2, r3].iter().enumerate() {
        let resp = result.as_ref().unwrap().as_ref().unwrap();
        assert_eq!(resp.status(), 200, "concurrent manifest GET {} failed", i);
    }
}

// ===========================================================================
// 7. Overwrite prevention across protocols — all three protocols store
//    the same content under different key namespaces without shadowing.
// ===========================================================================

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_overwrite_prevention_across_lfs_oci_bazel() {
    let pg_url = ensure_pg().await;
    let tmp = TempDir::new().unwrap();
    let chunk_size = NonZeroUsize::new(65536).unwrap();
    let config = ServerConfig::new(
        "127.0.0.1:0".parse().unwrap(),
        "http://127.0.0.1:8080".to_owned(),
        tmp.path().to_path_buf(),
        chunk_size,
    )
    .with_server_role(ServerRole::All)
    .with_server_frontends([
        ServerFrontend::Lfs,
        ServerFrontend::BazelHttp,
        ServerFrontend::Oci,
    ])
    .unwrap()
    .with_token_signing_key(TEST_SIGNING_KEY.to_vec())
    .unwrap()
    .with_index_postgres_url(pg_url.to_owned())
    .unwrap()
    .with_reconstruction_cache_disabled();
    config.validate_runtime_requirements().unwrap();
    let app = app::router(config).await.unwrap();

    let token = {
        let provider = LocalHmacProvider::new(TEST_SIGNING_KEY).unwrap();
        let repo_s =
            RepositoryScope::new(RepositoryProvider::Generic, "test", "test", Some("main"))
                .unwrap();
        let claims =
            TokenClaims::new("shardline", "test", TokenScope::Write, repo_s, u64::MAX).unwrap();
        provider.mint_token(&claims).unwrap()
    };
    let _keep = Box::new(tmp);

    // Same content uploaded via each protocol.
    let content = b"cross-protocol-shadow-test-content";
    let hash = sha256_hex(content);
    let repo = "test/test";

    // 1. Upload via LFS
    let lfs_put = app
        .clone()
        .oneshot(
            axum::http::Request::builder()
                .method("PUT")
                .uri(&format!("/v1/lfs/objects/{hash}"))
                .header("Authorization", format!("Bearer {token}"))
                .header("Content-Type", "application/octet-stream")
                .body(axum::body::Body::from(content.to_vec()))
                .unwrap(),
        )
        .await
        .unwrap();
    assert!(
        lfs_put.status().is_success(),
        "LFS PUT failed: {}",
        lfs_put.status()
    );

    // 2. Upload via OCI (monolithic blob upload)
    let oci_put = app
        .clone()
        .oneshot(
            axum::http::Request::builder()
                .method("POST")
                .uri(&format!("/v2/{repo}/blobs/uploads/?digest=sha256:{hash}"))
                .header("Authorization", format!("Bearer {token}"))
                .header("Content-Type", "application/octet-stream")
                .body(axum::body::Body::from(content.to_vec()))
                .unwrap(),
        )
        .await
        .unwrap();
    assert_eq!(
        oci_put.status(),
        201,
        "OCI blob upload failed: {}",
        oci_put.status()
    );

    // 3. Upload via Bazel CAS
    let bazel_put = app
        .clone()
        .oneshot(
            axum::http::Request::builder()
                .method("PUT")
                .uri(&format!("/v1/bazel/cache/cas/{hash}"))
                .header("Authorization", format!("Bearer {token}"))
                .header("Content-Type", "application/octet-stream")
                .body(axum::body::Body::from(content.to_vec()))
                .unwrap(),
        )
        .await
        .unwrap();
    assert!(
        bazel_put.status().is_success(),
        "Bazel CAS PUT failed: {}",
        bazel_put.status()
    );

    // 4. Verify LFS read returns correct content
    let lfs_get = app
        .clone()
        .oneshot(
            axum::http::Request::builder()
                .method("GET")
                .uri(&format!("/v1/lfs/objects/{hash}"))
                .header("Authorization", format!("Bearer {token}"))
                .body(axum::body::Body::empty())
                .unwrap(),
        )
        .await
        .unwrap();
    assert_eq!(lfs_get.status(), 200, "LFS GET should succeed");
    let lfs_body = axum::body::to_bytes(lfs_get.into_body(), usize::MAX)
        .await
        .unwrap();
    assert_eq!(lfs_body.as_ref(), content, "LFS content mismatch");

    // 5. Verify OCI read returns correct content
    let oci_get = app
        .clone()
        .oneshot(
            axum::http::Request::builder()
                .method("GET")
                .uri(&format!("/v2/{repo}/blobs/sha256:{hash}"))
                .header("Authorization", format!("Bearer {token}"))
                .body(axum::body::Body::empty())
                .unwrap(),
        )
        .await
        .unwrap();
    assert_eq!(oci_get.status(), 200, "OCI GET should succeed");
    let oci_body = axum::body::to_bytes(oci_get.into_body(), usize::MAX)
        .await
        .unwrap();
    assert_eq!(oci_body.as_ref(), content, "OCI content mismatch");

    // 6. Verify Bazel CAS read returns correct content
    let bazel_get = app
        .clone()
        .oneshot(
            axum::http::Request::builder()
                .method("GET")
                .uri(&format!("/v1/bazel/cache/cas/{hash}"))
                .header("Authorization", format!("Bearer {token}"))
                .body(axum::body::Body::empty())
                .unwrap(),
        )
        .await
        .unwrap();
    assert_eq!(bazel_get.status(), 200, "Bazel CAS GET should succeed");
    let bazel_body = axum::body::to_bytes(bazel_get.into_body(), usize::MAX)
        .await
        .unwrap();
    assert_eq!(bazel_body.as_ref(), content, "Bazel CAS content mismatch");
}

// ===========================================================================
// 24. Cross-protocol content sharing
// ===========================================================================
//
// Each protocol uses its own storage key namespace:
//   LFS:  protocols/lfs/{namespace}/objects/{oid}
//   Bazel: protocols/bazel/{kind}/{namespace}/{hash}
//   OCI:   protocols/oci/{namespace}/{repository}/{digest}
//   Hub:   HubStore (separate SQLite/Postgres tables)
//
// Cross-protocol reads document namespace isolation (expect 404).
// Same-protocol reads verify the upload was successful.

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_cross_bazel_cas_to_lfs() {
    let server = TestServer::start(&[ServerFrontend::BazelHttp, ServerFrontend::Lfs]).await;
    let client = reqwest::Client::new();

    let content = b"bazel-cas-to-lfs-cross-pg";
    let hash = sha256_hex(content);

    // Upload via Bazel CAS
    let put_resp = client
        .put(server.url(&format!("/v1/bazel/cache/cas/{hash}")))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .header("Content-Type", "application/octet-stream")
        .body(content.to_vec())
        .send()
        .await
        .unwrap();
    assert!(
        put_resp.status().is_success(),
        "Bazel CAS PUT failed: {}",
        put_resp.status()
    );

    // Verify same-protocol read works
    let bazel_get = client
        .get(server.url(&format!("/v1/bazel/cache/cas/{hash}")))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .send()
        .await
        .unwrap();
    assert_eq!(bazel_get.status(), 200);
    assert_eq!(bazel_get.bytes().await.unwrap().as_ref(), content);

    // Cross-protocol read via LFS — different key namespace → 404
    let lfs_get = client
        .get(server.url(&format!("/v1/lfs/objects/{hash}")))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .send()
        .await
        .unwrap();
    assert_eq!(
        lfs_get.status(),
        404,
        "LFS should not find data stored via Bazel CAS (separate key namespace)"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_cross_lfs_to_oci_blob() {
    // Use oneshot because OCI uses wildcard routes (/v2/{*path})
    let pg_url = ensure_pg().await;
    let tmp = TempDir::new().unwrap();
    let chunk_size = NonZeroUsize::new(65536).unwrap();
    let config = ServerConfig::new(
        "127.0.0.1:0".parse().unwrap(),
        "http://127.0.0.1:8080".to_owned(),
        tmp.path().to_path_buf(),
        chunk_size,
    )
    .with_server_role(ServerRole::All)
    .with_server_frontends([ServerFrontend::Lfs, ServerFrontend::Oci])
    .unwrap()
    .with_token_signing_key(TEST_SIGNING_KEY.to_vec())
    .unwrap()
    .with_index_postgres_url(pg_url.to_owned())
    .unwrap()
    .with_reconstruction_cache_disabled();
    config.validate_runtime_requirements().unwrap();
    let app = app::router(config).await.unwrap();

    let token = {
        let provider = LocalHmacProvider::new(TEST_SIGNING_KEY).unwrap();
        let repo_s =
            RepositoryScope::new(RepositoryProvider::Generic, "test", "test", Some("main"))
                .unwrap();
        let claims =
            TokenClaims::new("shardline", "test", TokenScope::Write, repo_s, u64::MAX).unwrap();
        provider.mint_token(&claims).unwrap()
    };
    let _keep = Box::new(tmp);

    let content = b"lfs-to-oci-cross-pg";
    let digest = sha256_hex(content);
    let repo = "test/test";

    // Upload via LFS
    let put_uri = format!("/v1/lfs/objects/{digest}");
    let put_req = axum::http::Request::builder()
        .method("PUT")
        .uri(&put_uri)
        .header("Authorization", format!("Bearer {token}"))
        .header("Content-Type", "application/octet-stream")
        .body(axum::body::Body::from(content.to_vec()))
        .unwrap();
    let put_resp = app.clone().oneshot(put_req).await.unwrap();
    assert!(
        put_resp.status().is_success(),
        "LFS PUT failed: {}",
        put_resp.status()
    );

    // Verify same-protocol read works
    let lfs_get = app
        .clone()
        .oneshot(
            axum::http::Request::builder()
                .method("GET")
                .uri(&put_uri)
                .header("Authorization", format!("Bearer {token}"))
                .body(axum::body::Body::empty())
                .unwrap(),
        )
        .await
        .unwrap();
    assert_eq!(
        lfs_get.status(),
        200,
        "LFS GET after LFS upload should work"
    );

    // Cross-protocol read via OCI — different key namespace → 404
    let get_uri = format!("/v2/{repo}/blobs/sha256:{digest}");
    let get_req = axum::http::Request::builder()
        .method("GET")
        .uri(&get_uri)
        .header("Authorization", format!("Bearer {token}"))
        .body(axum::body::Body::empty())
        .unwrap();
    let get_resp = app.clone().oneshot(get_req).await.unwrap();
    assert_eq!(
        get_resp.status(),
        404,
        "OCI should not find data stored via LFS (separate key namespace)"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_cross_hub_lfs_to_v1_lfs() {
    let server = TestServer::start(&[ServerFrontend::Hub, ServerFrontend::Lfs]).await;
    let client = reqwest::Client::new();
    let auth = || format!("Bearer {}", server.auth_header());

    let content = b"hub-lfs-to-v1-cross-pg";
    let oid = sha256_hex(content);

    // Upload via Hub's /lfs/objects/{oid}
    let put_resp = client
        .put(server.url(&format!("/lfs/objects/{oid}")))
        .header("Authorization", auth())
        .header("Content-Type", "application/octet-stream")
        .body(content.to_vec())
        .send()
        .await
        .unwrap();
    assert_eq!(
        put_resp.status(),
        200,
        "Hub LFS PUT failed: {}",
        put_resp.status()
    );

    // Verify same-protocol read via Hub LFS works
    let hub_get = client
        .get(server.url(&format!("/lfs/objects/{oid}")))
        .header("Authorization", auth())
        .send()
        .await
        .unwrap();
    assert_eq!(
        hub_get.status(),
        200,
        "Hub LFS GET should work after upload"
    );
    assert_eq!(hub_get.bytes().await.unwrap().as_ref(), content);

    // Cross-protocol read via V1 LFS — different storage backend → 404
    let v1_get = client
        .get(server.url(&format!("/v1/lfs/objects/{oid}")))
        .header("Authorization", auth())
        .send()
        .await
        .unwrap();
    assert_eq!(
        v1_get.status(),
        404,
        "V1 LFS should not find data stored via Hub LFS (separate storage)"
    );
}

// ===========================================================================
// 26. OCI v2 token with auth
// ===========================================================================

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_oci_v2_token_with_auth() {
    let mut builder = TestServerBuilder::new(&[ServerFrontend::Oci]);
    let server = builder.start().await;
    let client = reqwest::Client::new();

    // GET /v2/token with a valid bearer token + scope query
    let resp = client
        .get(server.url("/v2/token?scope=repository:test/test:pull&service=shardline"))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .send()
        .await
        .unwrap();

    assert_eq!(
        resp.status(),
        200,
        "OCI v2 token endpoint should return 200 with valid auth"
    );
    let json: serde_json::Value = resp.json().await.unwrap();
    assert!(
        json["token"].as_str().map_or(false, |t| !t.is_empty()),
        "response should contain a non-empty 'token' field"
    );
    assert!(
        json["access_token"]
            .as_str()
            .map_or(false, |t| !t.is_empty()),
        "response should contain a non-empty 'access_token' field"
    );
    assert!(
        json["expires_in"].as_u64().unwrap_or(0) >= 60,
        "expires_in should be at least 60 seconds"
    );
}

// ===========================================================================
// 27. 413 Payload Too Large
// ===========================================================================

/// Helper: build a minimal oneshot app with a specific max body limit and
/// the given frontends. Returns (app, token).
async fn app_with_body_limit(
    frontends: &[ServerFrontend],
    max_body_bytes: usize,
) -> (axum::Router, String) {
    let pg_url = ensure_pg().await;
    let tmp = TempDir::new().unwrap();
    let chunk_size = NonZeroUsize::new(65536).unwrap();
    let config = ServerConfig::new(
        "127.0.0.1:0".parse().unwrap(),
        "http://127.0.0.1:8080".to_owned(),
        tmp.path().to_path_buf(),
        chunk_size,
    )
    .with_server_role(ServerRole::All)
    .with_server_frontends(frontends.to_vec())
    .unwrap()
    .with_token_signing_key(TEST_SIGNING_KEY.to_vec())
    .unwrap()
    .with_index_postgres_url(pg_url.to_owned())
    .unwrap()
    .with_max_request_body_bytes(NonZeroUsize::new(max_body_bytes).unwrap())
    .with_reconstruction_cache_disabled();
    config.validate_runtime_requirements().unwrap();
    let app = app::router(config).await.unwrap();

    let provider = LocalHmacProvider::new(TEST_SIGNING_KEY).unwrap();
    let repo =
        RepositoryScope::new(RepositoryProvider::Generic, "test", "test", Some("main")).unwrap();
    let claims = TokenClaims::new("shardline", "test", TokenScope::Write, repo, u64::MAX).unwrap();
    let token = provider.mint_token(&claims).unwrap();
    // Keep tmp alive
    let _ = Box::new(tmp);
    (app, token)
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_413_payload_too_large_lfs() {
    let (app, token) = app_with_body_limit(&[ServerFrontend::Lfs], 1024).await;
    let oid = sha256_hex(b"dummy");
    let large_body = vec![0u8; 2048]; // exceeds 1024 limit

    let req = axum::http::Request::builder()
        .method("PUT")
        .uri(&format!("/v1/lfs/objects/{oid}"))
        .header("Authorization", format!("Bearer {token}"))
        .header("Content-Type", "application/octet-stream")
        .body(axum::body::Body::from(large_body))
        .unwrap();
    let resp = app.clone().oneshot(req).await.unwrap();
    assert_eq!(
        resp.status(),
        413,
        "LFS PUT with oversized body should return 413"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_413_payload_too_large_bazel() {
    let (app, token) = app_with_body_limit(&[ServerFrontend::BazelHttp], 1024).await;
    let hash = sha256_hex(b"dummy");
    let large_body = vec![0u8; 2048]; // exceeds 1024 limit

    let req = axum::http::Request::builder()
        .method("PUT")
        .uri(&format!("/v1/bazel/cache/cas/{hash}"))
        .header("Authorization", format!("Bearer {token}"))
        .body(axum::body::Body::from(large_body))
        .unwrap();
    let resp = app.clone().oneshot(req).await.unwrap();
    assert_eq!(
        resp.status(),
        413,
        "Bazel CAS PUT with oversized body should return 413"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_413_payload_too_large_hub_commit() {
    // Hub routes have their own 64MB body limit (hardcoded in hub_routes).
    // Verifying the endpoint works — create repo and send valid commit.
    let (app, token) = app_with_body_limit(&[ServerFrontend::Hub], 1024).await;

    // Create a repo
    let create_body = serde_json::json!({
        "type": "model",
        "name": "pg-413-team/pg-413-model",
        "private": false,
    });
    let create_req = axum::http::Request::builder()
        .method("POST")
        .uri("/api/repos/create")
        .header("Authorization", format!("Bearer {token}"))
        .header("Content-Type", "application/json")
        .body(axum::body::Body::from(
            serde_json::to_vec(&create_body).unwrap(),
        ))
        .unwrap();
    let create_resp = app.clone().oneshot(create_req).await.unwrap();
    assert_eq!(create_resp.status(), 201);

    // Send valid commit — proves the full Hub commit pipeline works end-to-end.
    let content_b64 = base64::Engine::encode(
        &base64::engine::general_purpose::STANDARD,
        b"hub 413 test content",
    );
    let ndjson = format!(
        "{{\"header\":{{\"message\":\"hub 413 commit\",\"parentCommit\":\"\"}}}}\n\
         {{\"file\":{{\"path\":\"test.txt\",\"content\":\"{content_b64}\"}}}}"
    );
    let commit_req = axum::http::Request::builder()
        .method("POST")
        .uri("/api/models/pg-413-team/pg-413-model/commit/main")
        .header("Authorization", format!("Bearer {token}"))
        .header("Content-Type", "application/x-ndjson")
        .body(axum::body::Body::from(ndjson))
        .unwrap();
    let commit_resp = app.clone().oneshot(commit_req).await.unwrap();
    assert_eq!(
        commit_resp.status(),
        200,
        "Hub valid commit should return 200, got {}",
        commit_resp.status()
    );

    // The 413 Payload Too Large behavior is verified by the LFS and Bazel
    // tests above. Hub routes have their own hardcoded 64MB body limit that
    // cannot be driven to 413 without sending >64MB, which is impractical.
}

// ===========================================================================
// 28. 405 Method Not Allowed
// ===========================================================================

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_405_method_not_allowed() {
    let server = TestServer::start(&[ServerFrontend::Lfs]).await;
    let client = reqwest::Client::new();

    // GET to POST-only /v1/lfs/objects/batch
    let resp = client
        .get(server.url("/v1/lfs/objects/batch"))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .send()
        .await
        .unwrap();

    assert_eq!(
        resp.status(),
        405,
        "GET to POST-only route should return 405"
    );
}

// ===========================================================================
// 29. CORS headers on Hub API
// ===========================================================================

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_cors_headers_hub_api() {
    let server = TestServer::start(&[ServerFrontend::Hub]).await;
    let client = reqwest::Client::new();

    let resp = client
        .get(server.url("/api/whoami-v2"))
        .header("Origin", "http://127.0.0.1:8080")
        .send()
        .await
        .unwrap();

    assert_eq!(resp.status(), 200);
    let cors_header = resp.headers().get("access-control-allow-origin");
    assert!(
        cors_header.is_some(),
        "Hub API response should include Access-Control-Allow-Origin header"
    );
    assert_eq!(
        cors_header.unwrap().to_str().unwrap(),
        "*",
        "Hub API should use the same allow-all CORS policy as the protocol frontends"
    );
}

// ===========================================================================
// 30. Very long URL path — server must not crash
// ===========================================================================

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_very_long_url_path_does_not_crash() {
    let server = TestServer::start(&[ServerFrontend::Lfs]).await;
    let client = reqwest::Client::new();

    // Build a 100KB path by repeating a path segment.
    let long_segment = "a".repeat(100_000);
    let uri = format!("/v1/lfs/objects/{long_segment}");

    // Send the request.  The server should not crash regardless of how hyper
    // / axum handles the oversized URI.  A 414 URI Too Long or a connection
    // reset are both acceptable — the server process must stay alive.
    let resp = client
        .get(server.url(&uri))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .send()
        .await;

    match resp {
        Ok(r) => {
            // Hyper may reject the oversize URI before it reaches the handler,
            // returning 414 or 400.  The server stays alive.
            let status = r.status();
            assert!(
                status.is_server_error() || status.is_client_error() || status.is_redirection(),
                "unexpected success status for 100KB URL path: {status}"
            );
            // Verify the server is still healthy after handling the long URL.
            let health = client
                .get(server.url("/healthz"))
                .send()
                .await
                .expect("server should still be alive after long URL");
            assert_eq!(health.status(), 200);
        }
        Err(_e) => {
            // A connection error is also acceptable as long as the server
            // process is not killed.  Verify health still works.
            let health = client
                .get(server.url("/healthz"))
                .send()
                .await
                .expect("server should still be alive after long URL");
            assert_eq!(health.status(), 200);
        }
    }
}

// ===========================================================================
// 31. Concurrent Hub commit + revisions read
// ===========================================================================

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_concurrent_hub_commit_and_read() {
    let server = TestServer::start(&[ServerFrontend::Hub]).await;
    let client = reqwest::Client::new();
    let auth = || format!("Bearer {}", server.auth_header());
    let base_url = server.base_url.clone();

    // Create a repo
    let ns = "concurrent-team-pg";
    let name = "concurrent-model-pg";
    let model_path = format!("{ns}/{name}");

    let create_resp = client
        .post(format!("{base_url}/api/repos/create"))
        .header("Authorization", auth())
        .header("Content-Type", "application/json")
        .json(&serde_json::json!({
            "type": "model",
            "name": &model_path,
            "private": false,
        }))
        .send()
        .await
        .unwrap();
    assert_eq!(create_resp.status(), 201);

    // Spawn commit + revisions read concurrently
    let content_b64 = base64::Engine::encode(
        &base64::engine::general_purpose::STANDARD,
        b"concurrent content pg",
    );
    let ndjson = format!(
        "{{\"header\":{{\"message\":\"concurrent commit\",\"parentCommit\":\"\"}}}}\n\
         {{\"file\":{{\"path\":\"concurrent.txt\",\"content\":\"{content_b64}\"}}}}"
    );

    let commit_url = format!("{base_url}/api/models/{ns}/{name}/commit/main");
    let revisions_url = format!("{base_url}/api/models/{ns}/{name}/revisions");
    let token = server.auth_header().to_owned();

    let (r1, r2) = tokio::join!(
        async {
            let client = reqwest::Client::new();
            client
                .post(&commit_url)
                .header("Authorization", format!("Bearer {token}"))
                .header("Content-Type", "application/x-ndjson")
                .body(ndjson)
                .send()
                .await
                .unwrap()
        },
        async {
            let client = reqwest::Client::new();
            client
                .get(&revisions_url)
                .header("Authorization", format!("Bearer {token}"))
                .send()
                .await
                .unwrap()
        }
    );

    // Commit should succeed
    assert!(
        r1.status().is_success() || r1.status().is_client_error(),
        "concurrent commit status: {}",
        r1.status()
    );
    // Revisions list should succeed
    assert_eq!(
        r2.status(),
        200,
        "concurrent revisions list should return 200, got {}",
        r2.status()
    );
}

// ===========================================================================
// 31. Concurrent OCI session upload + cancel
// ===========================================================================

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_concurrent_session_upload_and_cancel() {
    let (app, token) = oci_oneshot_app().await;
    let repo = "test/test";

    // Create session
    let create_req = axum::http::Request::builder()
        .method("POST")
        .uri(&format!("/v2/{repo}/blobs/uploads/"))
        .header("Authorization", format!("Bearer {token}"))
        .header("Content-Length", "0")
        .body(axum::body::Body::empty())
        .unwrap();
    let create_resp = app.clone().oneshot(create_req).await.unwrap();
    assert_eq!(create_resp.status(), 202, "session creation should succeed");
    let location = create_resp
        .headers()
        .get("location")
        .unwrap()
        .to_str()
        .unwrap()
        .to_owned();

    // Spawn PATCH + DELETE concurrently
    let app1 = app.clone();
    let token1 = token.clone();
    let loc1 = location.clone();
    let (patch_res, delete_res) = tokio::join!(
        async {
            let req = axum::http::Request::builder()
                .method("PATCH")
                .uri(&loc1)
                .header("Authorization", format!("Bearer {token1}"))
                .header("Content-Type", "application/octet-stream")
                .header("Content-Range", "0-4")
                .body(axum::body::Body::from(b"hello".to_vec()))
                .unwrap();
            app1.clone().oneshot(req).await.unwrap()
        },
        async {
            let req = axum::http::Request::builder()
                .method("DELETE")
                .uri(&location)
                .header("Authorization", format!("Bearer {token}"))
                .body(axum::body::Body::empty())
                .unwrap();
            app.clone().oneshot(req).await.unwrap()
        }
    );

    // Both operations should complete (one may fail if the session is already
    // deleted by the other, but neither should panic or hang).
    assert!(
        patch_res.status().is_success() || patch_res.status().is_client_error(),
        "concurrent PATCH status: {}",
        patch_res.status()
    );
    assert!(
        delete_res.status().is_success() || delete_res.status() == 404,
        "concurrent DELETE status: {}",
        delete_res.status()
    );
}

// ===========================================================================
// 25. Hub API authenticated requests
// ===========================================================================

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_hub_whoami_with_auth() {
    let server = TestServer::start(&[ServerFrontend::Hub]).await;
    let client = reqwest::Client::new();

    let resp = client
        .get(server.url("/api/whoami-v2"))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
    let json: serde_json::Value = resp.json().await.unwrap();
    // With a valid Write-scoped token, whoami returns the subject "test"
    assert_eq!(json["name"], "test", "whoami should return subject name");
    assert_eq!(
        json["auth"]["type"], "token",
        "auth.type should be 'token' with valid auth"
    );
    assert_eq!(
        json["auth"]["identity"]["account"]["name"], "test",
        "account name should match subject"
    );
    // The whoami response with auth should include auth details
    assert!(json.get("auth").is_some(), "auth block should be present");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_hub_commit_with_auth() {
    let server = TestServer::start(&[ServerFrontend::Hub]).await;
    let client = reqwest::Client::new();
    let auth = || format!("Bearer {}", server.auth_header());

    let ns = "hub-auth-team-pg";
    let name = "hub-auth-model-pg";
    let model_path = format!("{ns}/{name}");

    // Create a model repo (requires Write scope)
    let create_resp = client
        .post(server.url("/api/repos/create"))
        .header("Authorization", auth())
        .header("Content-Type", "application/json")
        .json(&serde_json::json!({
            "type": "model",
            "name": &model_path,
            "private": false,
        }))
        .send()
        .await
        .unwrap();
    assert_eq!(
        create_resp.status(),
        201,
        "repo create should succeed with auth"
    );

    // Commit a file (requires Write scope)
    let content_b64 = base64::Engine::encode(
        &base64::engine::general_purpose::STANDARD,
        b"hub auth commit content pg",
    );
    let ndjson = format!(
        "{{\"header\":{{\"message\":\"auth commit\",\"parentCommit\":\"\"}}}}\n\
         {{\"file\":{{\"path\":\"auth.txt\",\"content\":\"{content_b64}\"}}}}"
    );
    let commit_resp = client
        .post(server.url(&format!("/api/models/{ns}/{name}/commit/main")))
        .header("Authorization", auth())
        .header("Content-Type", "application/x-ndjson")
        .body(ndjson)
        .send()
        .await
        .unwrap();
    assert_eq!(commit_resp.status(), 200, "commit should succeed with auth");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_hub_xet_read_token_with_auth() {
    let server = TestServer::start(&[ServerFrontend::Hub]).await;
    let client = reqwest::Client::new();
    let auth = || format!("Bearer {}", server.auth_header());

    // Create a repo first
    let ns = "read-token-auth-pg";
    let name = "read-token-auth-repo";
    let model_path = format!("{ns}/{name}");

    let create_resp = client
        .post(server.url("/api/repos/create"))
        .header("Authorization", auth())
        .header("Content-Type", "application/json")
        .json(&serde_json::json!({
            "type": "model",
            "name": &model_path,
            "private": false,
        }))
        .send()
        .await
        .unwrap();
    assert_eq!(create_resp.status(), 201, "repo create should succeed");

    // Request xet-read-token (requires Read scope; our Write token allows Read)
    let token_resp = client
        .get(server.url(&format!("/api/models/{ns}/{name}/xet-read-token/main")))
        .header("Authorization", auth())
        .send()
        .await
        .unwrap();
    assert_eq!(
        token_resp.status(),
        200,
        "xet-read-token with valid auth should return 200"
    );
    let json: serde_json::Value = token_resp.json().await.unwrap();
    assert!(
        !json["token"].as_str().unwrap().is_empty(),
        "should return a token"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_hub_xet_write_token_with_auth() {
    let server = TestServer::start(&[ServerFrontend::Hub]).await;
    let client = reqwest::Client::new();
    let auth = || format!("Bearer {}", server.auth_header());

    // Create a repo first
    let ns = "write-token-auth-pg";
    let name = "write-token-auth-repo";
    let model_path = format!("{ns}/{name}");

    let create_resp = client
        .post(server.url("/api/repos/create"))
        .header("Authorization", auth())
        .header("Content-Type", "application/json")
        .json(&serde_json::json!({
            "type": "model",
            "name": &model_path,
            "private": false,
        }))
        .send()
        .await
        .unwrap();
    assert_eq!(create_resp.status(), 201, "repo create should succeed");

    // Request xet-write-token (requires Write scope)
    let token_resp = client
        .get(server.url(&format!("/api/models/{ns}/{name}/xet-write-token/main")))
        .header("Authorization", auth())
        .send()
        .await
        .unwrap();
    assert_eq!(
        token_resp.status(),
        200,
        "xet-write-token with valid auth should return 200"
    );
    let json: serde_json::Value = token_resp.json().await.unwrap();
    assert!(
        !json["token"].as_str().unwrap().is_empty(),
        "should return a token"
    );
}

// ===========================================================================
// Section 4: Boundary tests
// ===========================================================================

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_lfs_batch_exactly_max_objects() {
    // MAX_LFS_BATCH_OBJECTS = 1024. Sending exactly 1024 should succeed (not 422).
    let server = TestServer::start(&[ServerFrontend::Lfs]).await;
    let client = reqwest::Client::new();

    let objects: Vec<serde_json::Value> = (0..1024)
        .map(|i| serde_json::json!({"oid": format!("{:064x}", i), "size": 100}))
        .collect();
    let body = serde_json::json!({
        "operation": "download",
        "objects": objects
    });

    let resp = client
        .post(server.url("/v1/lfs/objects/batch"))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .header("Content-Type", "application/vnd.git-lfs+json")
        .json(&body)
        .send()
        .await
        .unwrap();

    assert_eq!(
        resp.status(),
        200,
        "batch with exactly 1024 objects should succeed, got {}",
        resp.status()
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_oci_tags_list_max_page_size() {
    // MAX_OCI_TAG_LIST_PAGE_SIZE = 256. Request ?n=256 → should return 200.
    let pg_url = ensure_pg().await;
    let tmp = TempDir::new().unwrap();
    let chunk_size = NonZeroUsize::new(65536).unwrap();
    let config = ServerConfig::new(
        "127.0.0.1:0".parse().unwrap(),
        "http://127.0.0.1:8080".to_owned(),
        tmp.path().to_path_buf(),
        chunk_size,
    )
    .with_server_role(ServerRole::All)
    .with_server_frontends([ServerFrontend::Oci])
    .unwrap()
    .with_token_signing_key(TEST_SIGNING_KEY.to_vec())
    .unwrap()
    .with_index_postgres_url(pg_url.to_owned())
    .unwrap()
    .with_reconstruction_cache_disabled();
    config.validate_runtime_requirements().unwrap();
    let app = app::router(config).await.unwrap();
    let _ = Box::new(tmp);

    let token = {
        let provider = LocalHmacProvider::new(TEST_SIGNING_KEY).unwrap();
        let repo_s =
            RepositoryScope::new(RepositoryProvider::Generic, "test", "test", Some("main"))
                .unwrap();
        let claims =
            TokenClaims::new("shardline", "test", TokenScope::Write, repo_s, u64::MAX).unwrap();
        provider.mint_token(&claims).unwrap()
    };

    // Push a few manifests so there's at least 1 tag to list
    let repo = "test/test";
    let config_data = b"{}";
    let layer_data = b"max-page-layer";

    // Upload blobs for manifest
    let config_digest = {
        let uri = format!(
            "/v2/{repo}/blobs/uploads/?digest=sha256:{}",
            sha256_hex(config_data)
        );
        let req = axum::http::Request::builder()
            .method("POST")
            .uri(&uri)
            .header("Authorization", format!("Bearer {token}"))
            .header("Content-Type", "application/octet-stream")
            .body(axum::body::Body::from(config_data.to_vec()))
            .unwrap();
        let resp = app.clone().oneshot(req).await.unwrap();
        assert_eq!(resp.status(), 201);
        sha256_hex(config_data)
    };
    let layer_digest = {
        let uri = format!(
            "/v2/{repo}/blobs/uploads/?digest=sha256:{}",
            sha256_hex(layer_data)
        );
        let req = axum::http::Request::builder()
            .method("POST")
            .uri(&uri)
            .header("Authorization", format!("Bearer {token}"))
            .header("Content-Type", "application/octet-stream")
            .body(axum::body::Body::from(layer_data.to_vec()))
            .unwrap();
        let resp = app.clone().oneshot(req).await.unwrap();
        assert_eq!(resp.status(), 201);
        sha256_hex(layer_data)
    };

    let manifest_body = serde_json::json!({
        "schemaVersion": 2,
        "mediaType": "application/vnd.oci.image.manifest.v1+json",
        "config": {
            "mediaType": "application/vnd.oci.image.config.v1+json",
            "size": 0,
            "digest": format!("sha256:{config_digest}")
        },
        "layers": [{
            "mediaType": "application/vnd.oci.image.layer.v1.tar+gzip",
            "size": 0,
            "digest": format!("sha256:{layer_digest}")
        }]
    })
    .to_string();

    let put_req = axum::http::Request::builder()
        .method("PUT")
        .uri(&format!("/v2/{repo}/manifests/max-page-tag"))
        .header("Authorization", format!("Bearer {token}"))
        .header("Content-Type", "application/vnd.oci.image.manifest.v1+json")
        .body(axum::body::Body::from(manifest_body))
        .unwrap();
    let put_resp = app.clone().oneshot(put_req).await.unwrap();
    assert_eq!(put_resp.status(), 201);

    // Request tags list with ?n=256 (max page size)
    let list_req = axum::http::Request::builder()
        .method("GET")
        .uri(&format!("/v2/{repo}/tags/list?n=256"))
        .header("Authorization", format!("Bearer {token}"))
        .body(axum::body::Body::empty())
        .unwrap();
    let list_resp = app.clone().oneshot(list_req).await.unwrap();
    assert_eq!(
        list_resp.status(),
        200,
        "tags list with n=256 should return 200, got {}",
        list_resp.status()
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_provider_token_request_at_max_body() {
    // MAX_PROVIDER_TOKEN_REQUEST_BODY_BYTES = 16384. Send a body exactly at the limit.
    let mut builder = TestServerBuilder::new(&[ServerFrontend::Xet]).with_provider();
    let server = builder.start().await;
    let client = reqwest::Client::new();

    // Build a body near the 16384 byte limit.
    // Subject must be in the provider config's read_subjects: "test-user".
    // We pad with a long "extra" field that serde ignores.
    let base = serde_json::json!({
        "subject": "test-user",
        "owner": "test",
        "repo": "test",
        "revision": serde_json::Value::Null,
        "scope": "Read",
        "extra": "",
    });
    let base_str = serde_json::to_string(&base).unwrap();
    let pad_len = 16378usize.saturating_sub(base_str.len());
    let extra = "x".repeat(pad_len);
    let body = serde_json::json!({
        "subject": "test-user",
        "owner": "test",
        "repo": "test",
        "revision": serde_json::Value::Null,
        "scope": "Read",
        "extra": extra,
    });
    let body_str = serde_json::to_string(&body).unwrap();
    assert!(
        body_str.len() <= 16384,
        "body length {} should not exceed 16384",
        body_str.len()
    );
    assert!(
        body_str.len() >= 16370,
        "body length {} should be near the 16384 limit",
        body_str.len()
    );

    let resp = client
        .post(server.url("/v1/providers/generic/tokens"))
        .header("x-shardline-provider-key", "test-api-key")
        .header("Content-Type", "application/json")
        .body(body_str.clone())
        .send()
        .await
        .unwrap();

    assert_eq!(
        resp.status(),
        200,
        "provider token request at max body size should return 200, got {} (body len = {})",
        resp.status(),
        body_str.len()
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_large_object_binary_content() {
    let server = TestServer::start(&[ServerFrontend::Lfs]).await;
    let client = reqwest::Client::new();

    // Upload 5MB of binary content (null bytes, non-UTF-8)
    let content: Vec<u8> = {
        let mut data = Vec::with_capacity(5 * 1024 * 1024);
        // Fill with null bytes interspersed with non-UTF-8 byte sequences
        for i in 0..(5 * 1024 * 1024) {
            match i % 4 {
                0 => data.push(0x00),
                1 => data.push(0xFF),
                2 => data.push(0xAB),
                _ => data.push(0x00),
            }
        }
        data
    };
    let oid = sha256_hex(&content);

    // PUT the binary content
    let put_resp = client
        .put(server.url(&format!("/v1/lfs/objects/{oid}")))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .header("Content-Type", "application/octet-stream")
        .body(content.clone())
        .send()
        .await
        .unwrap();
    assert!(
        put_resp.status().is_success(),
        "binary 5MB LFS PUT failed: {}",
        put_resp.status()
    );

    // GET the content back and verify roundtrip
    let get_resp = client
        .get(server.url(&format!("/v1/lfs/objects/{oid}")))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .send()
        .await
        .unwrap();
    assert_eq!(get_resp.status(), 200);
    let body = get_resp.bytes().await.unwrap();
    assert_eq!(body.len(), content.len(), "binary content size mismatch");
    assert_eq!(body.as_ref(), content.as_slice(), "binary content mismatch");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_lfs_upload_zero_byte_file() {
    let server = TestServer::start(&[ServerFrontend::Lfs]).await;
    let client = reqwest::Client::new();

    let content = b"";
    let oid = sha256_hex(content);

    // PUT 0-byte file
    let put_resp = client
        .put(server.url(&format!("/v1/lfs/objects/{oid}")))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .header("Content-Type", "application/octet-stream")
        .body(content.to_vec())
        .send()
        .await
        .unwrap();
    assert!(
        put_resp.status().is_success(),
        "0-byte LFS PUT failed: {}",
        put_resp.status()
    );

    // GET back and verify
    let get_resp = client
        .get(server.url(&format!("/v1/lfs/objects/{oid}")))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .send()
        .await
        .unwrap();
    assert_eq!(get_resp.status(), 200);
    let body = get_resp.bytes().await.unwrap();
    assert_eq!(body.as_ref(), content, "0-byte file roundtrip failed");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_lfs_upload_one_byte_file() {
    let server = TestServer::start(&[ServerFrontend::Lfs]).await;
    let client = reqwest::Client::new();

    let content = b"X";
    let oid = sha256_hex(content);

    // PUT 1-byte file
    let put_resp = client
        .put(server.url(&format!("/v1/lfs/objects/{oid}")))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .header("Content-Type", "application/octet-stream")
        .body(content.to_vec())
        .send()
        .await
        .unwrap();
    assert!(
        put_resp.status().is_success(),
        "1-byte LFS PUT failed: {}",
        put_resp.status()
    );

    // GET back and verify
    let get_resp = client
        .get(server.url(&format!("/v1/lfs/objects/{oid}")))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .send()
        .await
        .unwrap();
    assert_eq!(get_resp.status(), 200);
    let body = get_resp.bytes().await.unwrap();
    assert_eq!(body.len(), 1);
    assert_eq!(body.as_ref(), content, "1-byte file roundtrip failed");
}

// ===========================================================================
// Section 5: Concurrent tests
// ===========================================================================

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_concurrent_lfs_upload_idempotency() {
    let server = TestServer::start(&[ServerFrontend::Lfs]).await;
    let content = b"concurrent-idempotency-content-pg";
    let oid = sha256_hex(content);
    let url = server.url(&format!("/v1/lfs/objects/{oid}"));
    let token = server.auth_header().to_owned();

    // 5 concurrent PUTs to the same LFS OID — all should return 200
    let mut handles = Vec::new();
    for _ in 0..5 {
        let url = url.clone();
        let token = token.clone();
        let data = content.to_vec();
        handles.push(tokio::spawn(async move {
            let client = reqwest::Client::new();
            let resp = client
                .put(&url)
                .header("Authorization", format!("Bearer {token}"))
                .header("Content-Type", "application/octet-stream")
                .body(data)
                .send()
                .await
                .unwrap();
            resp.status()
        }));
    }

    for (i, handle) in handles.into_iter().enumerate() {
        let status = handle.await.unwrap();
        assert!(
            status.is_success(),
            "concurrent LFS PUT {} returned {}",
            i,
            status
        );
    }

    // Verify the stored content is correct
    let client = reqwest::Client::new();
    let get_resp = client
        .get(server.url(&format!("/v1/lfs/objects/{oid}")))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .send()
        .await
        .unwrap();
    assert_eq!(get_resp.status(), 200);
    assert_eq!(get_resp.bytes().await.unwrap().as_ref(), content);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_concurrent_hub_repo_creation() {
    let server = TestServer::start(&[ServerFrontend::Hub]).await;
    let _client = reqwest::Client::new();
    let base_url = server.base_url.clone();
    let token = server.auth_header().to_owned();

    let ns = "concurrent-create-pg";
    let name = "concurrent-create-repo";
    let model_path = format!("{ns}/{name}");

    // 5 concurrent POST /api/repos/create with the same name
    let mut handles = Vec::new();
    for _ in 0..5 {
        let url = format!("{base_url}/api/repos/create");
        let token = token.clone();
        let body = serde_json::json!({
            "type": "model",
            "name": &model_path,
            "private": false,
        });
        handles.push(tokio::spawn(async move {
            let client = reqwest::Client::new();
            let resp = client
                .post(&url)
                .header("Authorization", format!("Bearer {token}"))
                .header("Content-Type", "application/json")
                .json(&body)
                .send()
                .await
                .unwrap();
            resp.status()
        }));
    }

    let mut success_count = 0u32;
    let mut conflict_count = 0u32;
    for handle in handles {
        let status = handle.await.unwrap();
        match status.as_u16() {
            201 => success_count += 1,
            409 => conflict_count += 1,
            other => panic!("unexpected status code {other} for concurrent repo creation"),
        }
    }

    assert!(
        success_count >= 1,
        "at least 1 concurrent create should succeed (201), got {success_count}"
    );
    assert_eq!(
        success_count + conflict_count,
        5,
        "all 5 concurrent creates should return either 201 or 409"
    );
}

// ===========================================================================
// 36. OCI session state machine — invalid transitions
// ===========================================================================

/// Extracts the session ID from the Location header of an OCI upload session response.
fn oci_session_id(resp: &axum::http::Response<axum::body::Body>) -> String {
    let location = resp
        .headers()
        .get("location")
        .unwrap()
        .to_str()
        .unwrap()
        .to_owned();
    location.split('/').next_back().unwrap().to_owned()
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_oci_session_patch_after_complete() {
    let (app, token) = oci_oneshot_app().await;
    let repo = "test/test";

    // Full session: POST -> PATCH -> PUT
    let post_req = axum::http::Request::builder()
        .method("POST")
        .uri(&format!("/v2/{repo}/blobs/uploads/"))
        .header("Authorization", format!("Bearer {token}"))
        .header("Content-Length", "0")
        .body(axum::body::Body::empty())
        .unwrap();
    let post_resp = app.clone().oneshot(post_req).await.unwrap();
    assert_eq!(post_resp.status(), 202, "POST should create session");
    let session_id = oci_session_id(&post_resp);

    let data = b"session data";
    let patch_req = axum::http::Request::builder()
        .method("PATCH")
        .uri(&format!("/v2/{repo}/blobs/uploads/{session_id}"))
        .header("Authorization", format!("Bearer {token}"))
        .header("Content-Type", "application/octet-stream")
        .body(axum::body::Body::from(data.to_vec()))
        .unwrap();
    let patch_resp = app.clone().oneshot(patch_req).await.unwrap();
    assert_eq!(patch_resp.status(), 202, "PATCH should succeed");

    let digest = sha256_hex(data);
    let put_req = axum::http::Request::builder()
        .method("PUT")
        .uri(&format!(
            "/v2/{repo}/blobs/uploads/{session_id}?digest=sha256:{digest}"
        ))
        .header("Authorization", format!("Bearer {token}"))
        .body(axum::body::Body::empty())
        .unwrap();
    let put_resp = app.clone().oneshot(put_req).await.unwrap();
    assert_eq!(put_resp.status(), 201, "PUT should finalize session");

    // PATCH after complete -> 404
    let patch_after = axum::http::Request::builder()
        .method("PATCH")
        .uri(&format!("/v2/{repo}/blobs/uploads/{session_id}"))
        .header("Authorization", format!("Bearer {token}"))
        .header("Content-Type", "application/octet-stream")
        .body(axum::body::Body::from(b"extra".to_vec()))
        .unwrap();
    let patch_after_resp = app.clone().oneshot(patch_after).await.unwrap();
    assert_eq!(
        patch_after_resp.status(),
        404,
        "PATCH after complete should return 404"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_oci_session_put_after_complete() {
    let (app, token) = oci_oneshot_app().await;
    let repo = "test/test";

    let post_req = axum::http::Request::builder()
        .method("POST")
        .uri(&format!("/v2/{repo}/blobs/uploads/"))
        .header("Authorization", format!("Bearer {token}"))
        .header("Content-Length", "0")
        .body(axum::body::Body::empty())
        .unwrap();
    let post_resp = app.clone().oneshot(post_req).await.unwrap();
    assert_eq!(post_resp.status(), 202);
    let session_id = oci_session_id(&post_resp);

    let data = b"put-after-complete-data";
    let patch_req = axum::http::Request::builder()
        .method("PATCH")
        .uri(&format!("/v2/{repo}/blobs/uploads/{session_id}"))
        .header("Authorization", format!("Bearer {token}"))
        .header("Content-Type", "application/octet-stream")
        .body(axum::body::Body::from(data.to_vec()))
        .unwrap();
    let _ = app.clone().oneshot(patch_req).await.unwrap();

    let digest = sha256_hex(data);
    let put_req = axum::http::Request::builder()
        .method("PUT")
        .uri(&format!(
            "/v2/{repo}/blobs/uploads/{session_id}?digest=sha256:{digest}"
        ))
        .header("Authorization", format!("Bearer {token}"))
        .body(axum::body::Body::empty())
        .unwrap();
    let put_resp = app.clone().oneshot(put_req).await.unwrap();
    assert_eq!(put_resp.status(), 201);

    let put_again = axum::http::Request::builder()
        .method("PUT")
        .uri(&format!(
            "/v2/{repo}/blobs/uploads/{session_id}?digest=sha256:{digest}"
        ))
        .header("Authorization", format!("Bearer {token}"))
        .body(axum::body::Body::empty())
        .unwrap();
    let put_again_resp = app.clone().oneshot(put_again).await.unwrap();
    assert_eq!(
        put_again_resp.status(),
        404,
        "PUT after complete should return 404"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_oci_session_get_after_complete() {
    let (app, token) = oci_oneshot_app().await;
    let repo = "test/test";

    let post_req = axum::http::Request::builder()
        .method("POST")
        .uri(&format!("/v2/{repo}/blobs/uploads/"))
        .header("Authorization", format!("Bearer {token}"))
        .header("Content-Length", "0")
        .body(axum::body::Body::empty())
        .unwrap();
    let post_resp = app.clone().oneshot(post_req).await.unwrap();
    assert_eq!(post_resp.status(), 202);
    let session_id = oci_session_id(&post_resp);

    let data = b"get-after-complete";
    let patch_req = axum::http::Request::builder()
        .method("PATCH")
        .uri(&format!("/v2/{repo}/blobs/uploads/{session_id}"))
        .header("Authorization", format!("Bearer {token}"))
        .header("Content-Type", "application/octet-stream")
        .body(axum::body::Body::from(data.to_vec()))
        .unwrap();
    let _ = app.clone().oneshot(patch_req).await.unwrap();

    let digest = sha256_hex(data);
    let put_req = axum::http::Request::builder()
        .method("PUT")
        .uri(&format!(
            "/v2/{repo}/blobs/uploads/{session_id}?digest=sha256:{digest}"
        ))
        .header("Authorization", format!("Bearer {token}"))
        .body(axum::body::Body::empty())
        .unwrap();
    let put_resp = app.clone().oneshot(put_req).await.unwrap();
    assert_eq!(put_resp.status(), 201);

    let get_after = axum::http::Request::builder()
        .method("GET")
        .uri(&format!("/v2/{repo}/blobs/uploads/{session_id}"))
        .header("Authorization", format!("Bearer {token}"))
        .body(axum::body::Body::empty())
        .unwrap();
    let get_after_resp = app.clone().oneshot(get_after).await.unwrap();
    assert_eq!(
        get_after_resp.status(),
        404,
        "GET after complete should return 404"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_oci_session_patch_after_delete() {
    let (app, token) = oci_oneshot_app().await;
    let repo = "test/test";

    let post_req = axum::http::Request::builder()
        .method("POST")
        .uri(&format!("/v2/{repo}/blobs/uploads/"))
        .header("Authorization", format!("Bearer {token}"))
        .header("Content-Length", "0")
        .body(axum::body::Body::empty())
        .unwrap();
    let post_resp = app.clone().oneshot(post_req).await.unwrap();
    assert_eq!(post_resp.status(), 202);
    let session_id = oci_session_id(&post_resp);

    let delete_req = axum::http::Request::builder()
        .method("DELETE")
        .uri(&format!("/v2/{repo}/blobs/uploads/{session_id}"))
        .header("Authorization", format!("Bearer {token}"))
        .body(axum::body::Body::empty())
        .unwrap();
    let delete_resp = app.clone().oneshot(delete_req).await.unwrap();
    assert_eq!(delete_resp.status(), 204, "DELETE should succeed");

    let patch_after = axum::http::Request::builder()
        .method("PATCH")
        .uri(&format!("/v2/{repo}/blobs/uploads/{session_id}"))
        .header("Authorization", format!("Bearer {token}"))
        .header("Content-Type", "application/octet-stream")
        .body(axum::body::Body::from(b"data".to_vec()))
        .unwrap();
    let patch_after_resp = app.clone().oneshot(patch_after).await.unwrap();
    assert_eq!(
        patch_after_resp.status(),
        404,
        "PATCH after delete should return 404"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_oci_session_patch_empty_body() {
    let (app, token) = oci_oneshot_app().await;
    let repo = "test/test";

    let post_req = axum::http::Request::builder()
        .method("POST")
        .uri(&format!("/v2/{repo}/blobs/uploads/"))
        .header("Authorization", format!("Bearer {token}"))
        .header("Content-Length", "0")
        .body(axum::body::Body::empty())
        .unwrap();
    let post_resp = app.clone().oneshot(post_req).await.unwrap();
    assert_eq!(post_resp.status(), 202);
    let session_id = oci_session_id(&post_resp);

    let patch_empty = axum::http::Request::builder()
        .method("PATCH")
        .uri(&format!("/v2/{repo}/blobs/uploads/{session_id}"))
        .header("Authorization", format!("Bearer {token}"))
        .header("Content-Type", "application/octet-stream")
        .header("Content-Length", "0")
        .body(axum::body::Body::empty())
        .unwrap();
    let patch_empty_resp = app.clone().oneshot(patch_empty).await.unwrap();
    assert_eq!(
        patch_empty_resp.status(),
        202,
        "PATCH with empty body should return 202"
    );
}

// ===========================================================================
// 37. LFS state machine tests
// ===========================================================================

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_lfs_put_after_patch() {
    let server = TestServer::start(&[ServerFrontend::Lfs]).await;
    let client = reqwest::Client::new();

    let partial = b"partial-chunk-";
    let final_content = b"complete-content-from-put";
    let oid = sha256_hex(final_content);
    let total = final_content.len() as u64;

    let patch_range = format!("bytes 0-{}/{}", partial.len() as u64 - 1, total);
    let patch_resp = client
        .patch(server.url(&format!("/v1/lfs/objects/{oid}")))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .header("Content-Type", "application/octet-stream")
        .header("Content-Range", &patch_range)
        .body(partial.to_vec())
        .send()
        .await
        .unwrap();
    assert_eq!(patch_resp.status(), 200, "PATCH should succeed");

    let put_resp = client
        .put(server.url(&format!("/v1/lfs/objects/{oid}")))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .header("Content-Type", "application/octet-stream")
        .header("Content-Length", final_content.len().to_string())
        .body(final_content.to_vec())
        .send()
        .await
        .unwrap();
    assert!(put_resp.status().is_success(), "PUT should succeed");

    let get_resp = client
        .get(server.url(&format!("/v1/lfs/objects/{oid}")))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .send()
        .await
        .unwrap();
    assert_eq!(get_resp.status(), 200);
    let body = get_resp.bytes().await.unwrap();
    assert_eq!(
        body.as_ref(),
        final_content,
        "GET should return PUT content"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_lfs_patch_after_put() {
    let server = TestServer::start(&[ServerFrontend::Lfs]).await;
    let client = reqwest::Client::new();

    let original = b"original-content-from-put";
    let oid = sha256_hex(original);

    let put_resp = client
        .put(server.url(&format!("/v1/lfs/objects/{oid}")))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .header("Content-Type", "application/octet-stream")
        .body(original.to_vec())
        .send()
        .await
        .unwrap();
    assert!(put_resp.status().is_success(), "PUT should succeed");

    // PATCH additional bytes at a non-zero offset. The PATCH handler creates a
    // temp file, writes the chunk, and when the chunk is "final" it assembles
    // the file and stores it. Since the temp file starts sparse (null bytes at
    // the unfilled offsets), the assembled content's hash will not match the
    // OID — the PATCH finalize must reject with a 4xx.
    let extra = b"-patched";
    let total = (original.len() + extra.len()) as u64;
    let patch_range = format!("bytes {}-{}/{}", original.len(), total - 1, total);
    let patch_resp = client
        .patch(server.url(&format!("/v1/lfs/objects/{oid}")))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .header("Content-Type", "application/octet-stream")
        .header("Content-Range", &patch_range)
        .body(extra.to_vec())
        .send()
        .await
        .unwrap();
    // The assembled temp file has null bytes from 0..original.len() plus
    // "-patched" at the end; the stored content under the canonical key
    // (from the PUT) does not match. The object store's put_if_absent
    // detects the mismatch via ensure_file_matches_bytes and returns an
    // error. Accept any non-success status.
    assert!(
        !patch_resp.status().is_success(),
        "PATCH after PUT should not succeed (content mismatch), got {}",
        patch_resp.status()
    );

    // Verify the stored content is still the original PUT content (unchanged).
    let get_resp = client
        .get(server.url(&format!("/v1/lfs/objects/{oid}")))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .send()
        .await
        .unwrap();
    assert_eq!(get_resp.status(), 200);
    let body = get_resp.bytes().await.unwrap();
    assert_eq!(
        body.as_ref(),
        original,
        "stored content should remain unchanged after failed PATCH"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_lfs_patch_non_existent_oid() {
    let server = TestServer::start(&[ServerFrontend::Lfs]).await;
    let client = reqwest::Client::new();

    let content = b"new-oid-patch-content";
    let oid = sha256_hex(content);
    let total = content.len() as u64;

    let patch_range = format!("bytes 0-{}/{}", total - 1, total);
    let patch_resp = client
        .patch(server.url(&format!("/v1/lfs/objects/{oid}")))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .header("Content-Type", "application/octet-stream")
        .header("Content-Range", &patch_range)
        .body(content.to_vec())
        .send()
        .await
        .unwrap();
    assert_eq!(
        patch_resp.status(),
        200,
        "PATCH on non-existent OID should succeed"
    );

    let get_resp = client
        .get(server.url(&format!("/v1/lfs/objects/{oid}")))
        .header("Authorization", format!("Bearer {}", server.auth_header()))
        .send()
        .await
        .unwrap();
    assert_eq!(get_resp.status(), 200);
    let body = get_resp.bytes().await.unwrap();
    assert_eq!(body.as_ref(), content);
}

// ===========================================================================
// 38. Bazel + OCI coexistence
// ===========================================================================

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_bazel_and_oci_coexist() {
    let pg_url = ensure_pg().await;
    let tmp = TempDir::new().unwrap();
    let chunk_size = NonZeroUsize::new(65536).unwrap();
    let config = ServerConfig::new(
        "127.0.0.1:0".parse().unwrap(),
        "http://127.0.0.1:8080".to_owned(),
        tmp.path().to_path_buf(),
        chunk_size,
    )
    .with_server_role(ServerRole::All)
    .with_server_frontends([ServerFrontend::BazelHttp, ServerFrontend::Oci])
    .unwrap()
    .with_token_signing_key(TEST_SIGNING_KEY.to_vec())
    .unwrap()
    .with_index_postgres_url(pg_url.to_owned())
    .unwrap()
    .with_reconstruction_cache_disabled();
    config.validate_runtime_requirements().unwrap();
    let app = app::router(config).await.unwrap();

    let token = {
        let provider = LocalHmacProvider::new(TEST_SIGNING_KEY).unwrap();
        let repo_s =
            RepositoryScope::new(RepositoryProvider::Generic, "test", "test", Some("main"))
                .unwrap();
        let claims =
            TokenClaims::new("shardline", "test", TokenScope::Write, repo_s, u64::MAX).unwrap();
        provider.mint_token(&claims).unwrap()
    };
    let _keep = Box::new(tmp);

    let repo = "test/test";
    let content = b"coexistence-test-content";
    let hash = sha256_hex(content);

    // Upload to Bazel CAS
    let bazel_put = axum::http::Request::builder()
        .method("PUT")
        .uri(&format!("/v1/bazel/cache/cas/{hash}"))
        .header("Authorization", format!("Bearer {token}"))
        .header("Content-Type", "application/octet-stream")
        .body(axum::body::Body::from(content.to_vec()))
        .unwrap();
    let bazel_put_resp = app.clone().oneshot(bazel_put).await.unwrap();
    assert!(
        bazel_put_resp.status().is_success(),
        "Bazel CAS PUT failed: {}",
        bazel_put_resp.status()
    );

    // Upload same content to OCI blob (different key namespace)
    let oci_put = axum::http::Request::builder()
        .method("POST")
        .uri(&format!("/v2/{repo}/blobs/uploads/?digest=sha256:{hash}"))
        .header("Authorization", format!("Bearer {token}"))
        .header("Content-Type", "application/octet-stream")
        .body(axum::body::Body::from(content.to_vec()))
        .unwrap();
    let oci_put_resp = app.clone().oneshot(oci_put).await.unwrap();
    assert_eq!(oci_put_resp.status(), 201, "OCI blob upload should succeed");

    // Bazel GET works
    let bazel_get = axum::http::Request::builder()
        .method("GET")
        .uri(&format!("/v1/bazel/cache/cas/{hash}"))
        .header("Authorization", format!("Bearer {token}"))
        .body(axum::body::Body::empty())
        .unwrap();
    let bazel_get_resp = app.clone().oneshot(bazel_get).await.unwrap();
    assert_eq!(bazel_get_resp.status(), 200, "Bazel CAS GET should work");
    let bazel_body = axum::body::to_bytes(bazel_get_resp.into_body(), usize::MAX)
        .await
        .unwrap();
    assert_eq!(bazel_body.as_ref(), content);

    // OCI GET works
    let oci_get = axum::http::Request::builder()
        .method("GET")
        .uri(&format!("/v2/{repo}/blobs/sha256:{hash}"))
        .header("Authorization", format!("Bearer {token}"))
        .body(axum::body::Body::empty())
        .unwrap();
    let oci_get_resp = app.clone().oneshot(oci_get).await.unwrap();
    assert_eq!(oci_get_resp.status(), 200, "OCI blob GET should work");
    let oci_body = axum::body::to_bytes(oci_get_resp.into_body(), usize::MAX)
        .await
        .unwrap();
    assert_eq!(oci_body.as_ref(), content);

    // Namespace isolation
    let different_hash = sha256_hex(b"different-content");
    let bazel_get_missing = axum::http::Request::builder()
        .method("GET")
        .uri(&format!("/v1/bazel/cache/cas/{different_hash}"))
        .header("Authorization", format!("Bearer {token}"))
        .body(axum::body::Body::empty())
        .unwrap();
    let bazel_missing_resp = app.clone().oneshot(bazel_get_missing).await.unwrap();
    assert_eq!(
        bazel_missing_resp.status(),
        404,
        "Bazel should not find non-Bazel content"
    );
}

// ===========================================================================
// 39. Hub + Xet + LFS triple coexistence
// ===========================================================================

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_hub_xet_lfs_triple_coexist() {
    let pg_url = ensure_pg().await;
    let tmp = TempDir::new().unwrap();
    let chunk_size = NonZeroUsize::new(65536).unwrap();
    let config = ServerConfig::new(
        "127.0.0.1:0".parse().unwrap(),
        "http://127.0.0.1:8080".to_owned(),
        tmp.path().to_path_buf(),
        chunk_size,
    )
    .with_server_role(ServerRole::All)
    .with_server_frontends([
        ServerFrontend::Hub,
        ServerFrontend::Xet,
        ServerFrontend::Lfs,
    ])
    .unwrap()
    .with_token_signing_key(TEST_SIGNING_KEY.to_vec())
    .unwrap()
    .with_index_postgres_url(pg_url.to_owned())
    .unwrap()
    .with_reconstruction_cache_disabled();
    config.validate_runtime_requirements().unwrap();
    let app = app::router(config).await.unwrap();

    let token = {
        let provider = LocalHmacProvider::new(TEST_SIGNING_KEY).unwrap();
        let repo_s =
            RepositoryScope::new(RepositoryProvider::Generic, "test", "test", Some("main"))
                .unwrap();
        let claims =
            TokenClaims::new("shardline", "test", TokenScope::Write, repo_s, u64::MAX).unwrap();
        provider.mint_token(&claims).unwrap()
    };
    let _keep = Box::new(tmp);

    let repo = "test/test";

    // 1. Create repo via Hub API
    let hub_create = axum::http::Request::builder()
        .method("POST")
        .uri("/api/repos/create")
        .header("Authorization", format!("Bearer {token}"))
        .header("Content-Type", "application/json")
        .body(axum::body::Body::from(
            serde_json::to_vec(&serde_json::json!({"name": repo, "type": "model"})).unwrap(),
        ))
        .unwrap();
    let create_resp = app.clone().oneshot(hub_create).await.unwrap();
    assert_eq!(
        create_resp.status(),
        201,
        "Hub repo creation should succeed: {}",
        create_resp.status()
    );

    // 2. LFS batch request via standard LFS frontend
    let lfs_batch = axum::http::Request::builder()
        .method("POST")
        .uri("/v1/lfs/objects/batch")
        .header("Authorization", format!("Bearer {token}"))
        .header("Content-Type", "application/vnd.git-lfs+json")
        .body(axum::body::Body::from(
            serde_json::to_vec(&serde_json::json!({
                "operation": "download",
                "objects": [],
                "transfers": ["basic"]
            }))
            .unwrap(),
        ))
        .unwrap();
    let lfs_resp = app.clone().oneshot(lfs_batch).await.unwrap();
    assert_eq!(
        lfs_resp.status(),
        200,
        "LFS batch should succeed: {}",
        lfs_resp.status()
    );

    // 3. Xet read token via Xet frontend
    let xet_read = axum::http::Request::builder()
        .method("GET")
        .uri("/api/generic/test/test/xet-read-token/main")
        .header("Authorization", format!("Bearer {token}"))
        .body(axum::body::Body::empty())
        .unwrap();
    let xet_resp = app.clone().oneshot(xet_read).await.unwrap();
    // Should not be a 500 (may 404 on missing shard, but not crash)
    assert!(
        !xet_resp.status().is_server_error(),
        "Xet read should not 500: {}",
        xet_resp.status()
    );

    // 4. Healthz
    let health = app
        .clone()
        .oneshot(
            axum::http::Request::builder()
                .uri("/healthz")
                .body(axum::body::Body::empty())
                .unwrap(),
        )
        .await
        .unwrap();
    assert_eq!(health.status(), 200);

    // 5. Hub list repos
    let hub_list = axum::http::Request::builder()
        .method("GET")
        .uri("/api/repos")
        .header("Authorization", format!("Bearer {token}"))
        .body(axum::body::Body::empty())
        .unwrap();
    let list_resp = app.clone().oneshot(hub_list).await.unwrap();
    assert_eq!(
        list_resp.status(),
        200,
        "Hub list should succeed: {}",
        list_resp.status()
    );
    let body_bytes = axum::body::to_bytes(list_resp.into_body(), 1024 * 1024)
        .await
        .unwrap();
    let body_str = String::from_utf8_lossy(&body_bytes);
    assert!(
        body_str.contains("test/test"),
        "Hub repo list should contain created repo: {body_str}"
    );
}

// ===========================================================================
// 40. Graceful shutdown with timeout
// ===========================================================================

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_graceful_shutdown_with_timeout() {
    let pg_url = ensure_pg().await;
    let tmp = TempDir::new().unwrap();
    let chunk_size = NonZeroUsize::new(65536).unwrap();

    let config = ServerConfig::new(
        "127.0.0.1:0".parse().unwrap(),
        "http://127.0.0.1:8080".to_owned(),
        tmp.path().to_path_buf(),
        chunk_size,
    )
    .with_server_role(ServerRole::All)
    .with_server_frontends([ServerFrontend::Xet])
    .unwrap()
    .with_token_signing_key(TEST_SIGNING_KEY.to_vec())
    .unwrap()
    .with_index_postgres_url(pg_url.to_owned())
    .unwrap()
    .with_reconstruction_cache_disabled()
    .with_shutdown_timeout(Duration::from_millis(500));

    config.validate_runtime_requirements().unwrap();
    let app = app::router(config).await.unwrap();
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();

    let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel::<()>();
    let (done_tx, done_rx) = tokio::sync::oneshot::channel::<()>();

    tokio::spawn(async move {
        axum::serve(listener, app)
            .with_graceful_shutdown(async {
                shutdown_rx.await.ok();
            })
            .await
            .ok();
        done_tx.send(()).ok();
    });

    // Give server a moment to start
    tokio::time::sleep(Duration::from_millis(100)).await;

    // Trigger shutdown
    shutdown_tx.send(()).ok();

    // Server should exit within timeout + margin
    tokio::time::timeout(Duration::from_secs(10), done_rx)
        .await
        .expect("timeout: server should shut down within 10s")
        .expect("server task should complete");
}

// ===========================================================================
// 41. Graceful shutdown interrupts active upload
// ===========================================================================

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_graceful_shutdown_during_upload() {
    let pg_url = ensure_pg().await;
    let tmp = TempDir::new().unwrap();
    let chunk_size = NonZeroUsize::new(65536).unwrap();

    let config = ServerConfig::new(
        "127.0.0.1:0".parse().unwrap(),
        "http://127.0.0.1:8080".to_owned(),
        tmp.path().to_path_buf(),
        chunk_size,
    )
    .with_server_role(ServerRole::All)
    .with_server_frontends([ServerFrontend::Lfs])
    .unwrap()
    .with_token_signing_key(TEST_SIGNING_KEY.to_vec())
    .unwrap()
    .with_index_postgres_url(pg_url.to_owned())
    .unwrap()
    .with_reconstruction_cache_disabled()
    .with_shutdown_timeout(Duration::from_millis(500));

    config.validate_runtime_requirements().unwrap();
    let app = app::router(config).await.unwrap();
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let base_url = format!("http://{addr}");
    let token = {
        let provider = LocalHmacProvider::new(TEST_SIGNING_KEY).unwrap();
        let repository =
            RepositoryScope::new(RepositoryProvider::Generic, "test", "test", Some("main"))
                .unwrap();
        let claims =
            TokenClaims::new("shardline", "test", TokenScope::Write, repository, u64::MAX).unwrap();
        provider.mint_token(&claims).unwrap()
    };

    let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel::<()>();
    let (server_done_tx, server_done_rx) = tokio::sync::oneshot::channel::<()>();

    // Start the shutdown timeout only after the shutdown signal. After that
    // timeout, dropping the serve future force-closes in-flight connections.
    let shutdown_timeout = Duration::from_millis(500);
    let (shutdown_started_tx, shutdown_started_rx) = tokio::sync::oneshot::channel::<()>();
    tokio::spawn(async move {
        let serve = axum::serve(listener, app).with_graceful_shutdown(async {
            shutdown_rx.await.ok();
            shutdown_started_tx.send(()).ok();
        });
        tokio::select! {
            result = serve => { result.ok(); }
            () = async {
                if shutdown_started_rx.await.is_ok() {
                    tokio::time::sleep(shutdown_timeout).await;
                } else {
                    std::future::pending::<()>().await;
                }
            } => {
                // Dropped — connections force-closed
            }
        }
        server_done_tx.send(()).ok();
    });

    // Give server a moment to start.
    tokio::time::sleep(Duration::from_millis(100)).await;

    // Start a slow LFS upload. Yield an initial byte so the handler begins
    // consuming the request, then keep the stream pending to hold the upload
    // in flight while shutdown begins.
    let client = reqwest::Client::builder()
        // A force-closed connection should fail promptly after the server's
        // 500 ms shutdown deadline; do not let a stalled client hide that.
        .timeout(Duration::from_secs(3))
        .build()
        .unwrap();

    let (body_started_tx, body_started_rx) = tokio::sync::oneshot::channel::<()>();
    let slow_stream = stream::once(async move {
        body_started_tx.send(()).ok();
        Ok::<_, reqwest::Error>(bytes::Bytes::from_static(b"x"))
    })
    .chain(stream::once(async {
        tokio::time::sleep(Duration::from_secs(60)).await;
        Ok::<_, reqwest::Error>(bytes::Bytes::from_static(b"y"))
    }));

    let oid = sha256_hex(b"xy");
    let upload_handle = tokio::spawn(async move {
        client
            .put(&format!("{base_url}/v1/lfs/objects/{oid}"))
            .header("Authorization", format!("Bearer {token}"))
            .header("Content-Type", "application/octet-stream")
            .body(reqwest::Body::wrap_stream(slow_stream))
            .send()
            .await
    });

    tokio::time::timeout(Duration::from_secs(2), body_started_rx)
        .await
        .expect("upload request body should begin streaming")
        .expect("upload body stream should not be cancelled before shutdown");
    tokio::time::sleep(Duration::from_millis(50)).await;

    // Trigger shutdown while upload is in-flight.
    shutdown_tx.send(()).ok();

    // Server should exit within the select! timeout + margin.
    tokio::time::timeout(Duration::from_secs(10), server_done_rx)
        .await
        .expect("timeout: server should shut down within 10s")
        .expect("server task should complete");

    // The upload request should fail or get a server error since the connection
    // was force-closed when the serve future was dropped.
    let upload_result = upload_handle.await.unwrap();
    assert!(
        upload_result.is_err() || upload_result.as_ref().unwrap().status().is_server_error(),
        "upload should be interrupted by shutdown, got: {upload_result:?}"
    );
}

// ===========================================================================
// 42. CORS headers on LFS protocol frontend
// ===========================================================================

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_cors_headers_on_lfs_endpoint() {
    let pg_url = ensure_pg().await;
    let tmp = TempDir::new().unwrap();
    let chunk_size = NonZeroUsize::new(65536).unwrap();
    let config = ServerConfig::new(
        "127.0.0.1:0".parse().unwrap(),
        "http://127.0.0.1:8080".to_owned(),
        tmp.path().to_path_buf(),
        chunk_size,
    )
    .with_server_role(ServerRole::All)
    .with_server_frontends([ServerFrontend::Lfs])
    .unwrap()
    .with_token_signing_key(TEST_SIGNING_KEY.to_vec())
    .unwrap()
    .with_index_postgres_url(pg_url.to_owned())
    .unwrap()
    .with_reconstruction_cache_disabled();
    config.validate_runtime_requirements().unwrap();
    let app = app::router(config).await.unwrap();

    let _token = {
        let provider = LocalHmacProvider::new(TEST_SIGNING_KEY).unwrap();
        let repo_s =
            RepositoryScope::new(RepositoryProvider::Generic, "test", "test", Some("main"))
                .unwrap();
        let claims =
            TokenClaims::new("shardline", "test", TokenScope::Read, repo_s, u64::MAX).unwrap();
        provider.mint_token(&claims).unwrap()
    };
    let _keep = Box::new(tmp);

    // OPTIONS preflight request to LFS batch endpoint
    let resp = app
        .clone()
        .oneshot(
            axum::http::Request::builder()
                .method("OPTIONS")
                .uri("/v1/lfs/objects/batch")
                .header("Origin", "http://example.com")
                .header("Access-Control-Request-Method", "POST")
                .body(axum::body::Body::empty())
                .unwrap(),
        )
        .await
        .unwrap();

    let cors_header = resp.headers().get("access-control-allow-origin");
    assert!(
        cors_header.is_some(),
        "LFS response should include Access-Control-Allow-Origin header"
    );
    assert_eq!(
        cors_header.unwrap().to_str().unwrap(),
        "*",
        "LFS should allow all origins"
    );

    // Also verify a normal GET request includes CORS headers
    let get_resp = app
        .clone()
        .oneshot(
            axum::http::Request::builder()
                .method("GET")
                .uri("/healthz")
                .header("Origin", "http://other-origin.com")
                .body(axum::body::Body::empty())
                .unwrap(),
        )
        .await
        .unwrap();
    let health_cors = get_resp.headers().get("access-control-allow-origin");
    assert!(
        health_cors.is_some(),
        "Healthz should also include CORS headers"
    );
}

// ===========================================================================
// 43. Concurrent cross-protocol upload of same content
// ===========================================================================

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_concurrent_cross_protocol_upload_same_hash() {
    let pg_url = ensure_pg().await;
    let tmp = TempDir::new().unwrap();
    let chunk_size = NonZeroUsize::new(65536).unwrap();
    let config = ServerConfig::new(
        "127.0.0.1:0".parse().unwrap(),
        "http://127.0.0.1:8080".to_owned(),
        tmp.path().to_path_buf(),
        chunk_size,
    )
    .with_server_role(ServerRole::All)
    .with_server_frontends([ServerFrontend::Lfs, ServerFrontend::BazelHttp])
    .unwrap()
    .with_token_signing_key(TEST_SIGNING_KEY.to_vec())
    .unwrap()
    .with_index_postgres_url(pg_url.to_owned())
    .unwrap()
    .with_reconstruction_cache_disabled();
    config.validate_runtime_requirements().unwrap();
    let app = Arc::new(app::router(config).await.unwrap());

    let token = {
        let provider = LocalHmacProvider::new(TEST_SIGNING_KEY).unwrap();
        let repo_s =
            RepositoryScope::new(RepositoryProvider::Generic, "test", "test", Some("main"))
                .unwrap();
        let claims =
            TokenClaims::new("shardline", "test", TokenScope::Write, repo_s, u64::MAX).unwrap();
        provider.mint_token(&claims).unwrap()
    };
    let _keep = Box::new(tmp);

    let content = b"cross-protocol-race-content";
    let hash = sha256_hex(content);
    let token_arc = Arc::new(token);
    let hash_arc = Arc::new(hash);
    let content_arc = Arc::new(content.to_vec());

    // Fire both uploads concurrently
    let app1 = app.clone();
    let token1 = token_arc.clone();
    let hash1 = hash_arc.clone();
    let content1 = content_arc.clone();
    let bazel_handle = tokio::spawn(async move {
        let req = axum::http::Request::builder()
            .method("PUT")
            .uri(&format!("/v1/bazel/cache/cas/{hash1}"))
            .header("Authorization", format!("Bearer {token1}"))
            .header("Content-Type", "application/octet-stream")
            .body(axum::body::Body::from(content1.to_vec()))
            .unwrap();
        app1.as_ref().clone().oneshot(req).await
    });

    let app2 = app.clone();
    let token2 = token_arc.clone();
    let content2 = content_arc.clone();
    let hash2 = hash_arc.clone();
    let lfs_handle = tokio::spawn(async move {
        // Use LFS upload — PUT with the same content hash as Bazel
        let req = axum::http::Request::builder()
            .method("PUT")
            .uri(&format!("/v1/lfs/objects/{hash2}"))
            .header("Authorization", format!("Bearer {token2}"))
            .header("Content-Type", "application/octet-stream")
            .body(axum::body::Body::from(content2.to_vec()))
            .unwrap();
        app2.as_ref().clone().oneshot(req).await
    });

    let bazel_resp = bazel_handle.await.unwrap().unwrap();
    let lfs_resp = lfs_handle.await.unwrap().unwrap();

    assert!(
        bazel_resp.status().is_success(),
        "Bazel upload should succeed: {}",
        bazel_resp.status()
    );
    assert!(
        lfs_resp.status().is_success(),
        "LFS upload should succeed: {}",
        lfs_resp.status()
    );

    // Verify content is retrievable via Bazel CAS
    let get_req = axum::http::Request::builder()
        .method("GET")
        .uri(&format!("/v1/bazel/cache/cas/{hash_arc}"))
        .header("Authorization", format!("Bearer {token_arc}"))
        .body(axum::body::Body::empty())
        .unwrap();
    let get_resp = app.as_ref().clone().oneshot(get_req).await.unwrap();
    assert_eq!(
        get_resp.status(),
        200,
        "Bazel CAS GET should return content"
    );
    let get_body = axum::body::to_bytes(get_resp.into_body(), usize::MAX)
        .await
        .unwrap();
    assert_eq!(get_body.as_ref(), content_arc.as_slice());
}