briefcase_core/storage/

lakefs.rs

use super::{FlushResult, SnapshotQuery, StorageBackend, StorageError};
use crate::models::{DecisionSnapshot, Snapshot};
#[cfg(feature = "networking")]
use base64::{engine::general_purpose, Engine as _};
#[cfg(feature = "networking")]
use reqwest::{
    header::{HeaderMap, HeaderValue, AUTHORIZATION, CONTENT_TYPE},
    Client, RequestBuilder, Response,
};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::{Arc, Mutex};

// ─── Authentication types ────────────────────────────────────────────────────

/// Supported lakeFS authentication modes.
///
/// lakeFS supports multiple authentication mechanisms.  `LakeFSAuth`
/// captures the configuration needed for each.  Pass one of these to
/// [`LakeFSConfig::with_auth`] to select the desired mode.
///
/// When no explicit auth is configured (`None`), the backend falls back
/// to Basic auth using `access_key` / `secret_key` from the config.
#[derive(Debug, Clone)]
pub enum LakeFSAuth {
    /// HTTP Basic authentication (access_key:secret_key).
    /// This is the lakeFS default and the backward-compatible mode.
    Basic {
        access_key: String,
        secret_key: String,
    },

    /// JWT token authentication.
    /// The client POSTs to `/api/v1/auth/login` with access_key/secret_key
    /// and receives a JWT.  The JWT is refreshed before expiry.
    Token {
        access_key: String,
        secret_key: String,
    },

    /// STS (Security Token Service) federation.
    /// The client POSTs an external OIDC token to `/sts/login` and receives
    /// temporary lakeFS credentials (access_key + secret_key + session_token).
    Sts {
        /// The external OIDC identity token to exchange.
        oidc_token: String,
    },

    /// IAM-based authentication (AWS instance metadata / ECS task role).
    /// Credentials are fetched from the EC2 instance metadata service or
    /// ECS container credentials endpoint at `169.254.169.254` / `169.254.170.2`.
    Iam,

    /// External OIDC provider authentication.
    /// The client sends an external OIDC token to `/api/v1/oidc/login` and
    /// receives a lakeFS JWT in return.
    Oidc {
        /// The external OIDC identity token.
        token: String,
    },

    /// SAML SSO authentication.
    /// The client POSTs a base64-encoded SAML assertion to
    /// `/api/v1/auth/external/saml` and receives a lakeFS JWT.
    Saml {
        /// Base64-encoded SAML assertion XML.
        assertion: String,
    },
}

// ─── AuthProvider (internal) ─────────────────────────────────────────────────

/// Internal trait for dynamic authentication header management.
///
/// Each auth mode implements this trait.  The `LakeFSBackend` calls
/// `auth_header()` before every HTTP request, allowing token-based modes
/// to transparently refresh expired credentials.
#[cfg(all(feature = "async", feature = "networking"))]
#[async_trait::async_trait]
pub(crate) trait AuthProvider: Send + Sync {
    /// Return the current `Authorization` header value.
    ///
    /// Implementations that use expiring tokens should check for expiry
    /// inside this method and refresh transparently.
    async fn auth_header(&self) -> Result<HeaderValue, StorageError>;

    /// Human-readable name for logging.
    fn mode_name(&self) -> &'static str;
}

// ── BasicAuthProvider ────────────────────────────────────────────────────────

#[cfg(all(feature = "async", feature = "networking"))]
pub(crate) struct BasicAuthProvider {
    header: HeaderValue,
}

#[cfg(all(feature = "async", feature = "networking"))]
impl BasicAuthProvider {
    pub fn new(access_key: &str, secret_key: &str) -> Result<Self, StorageError> {
        let credentials = format!("{}:{}", access_key, secret_key);
        let encoded = general_purpose::STANDARD.encode(credentials.as_bytes());
        let header_str = format!("Basic {}", encoded);
        let header = HeaderValue::from_str(&header_str).map_err(|e| {
            StorageError::ConnectionError(format!("Invalid Basic auth header: {}", e))
        })?;
        Ok(Self { header })
    }
}

#[cfg(all(feature = "async", feature = "networking"))]
#[async_trait::async_trait]
impl AuthProvider for BasicAuthProvider {
    async fn auth_header(&self) -> Result<HeaderValue, StorageError> {
        Ok(self.header.clone())
    }
    fn mode_name(&self) -> &'static str {
        "basic"
    }
}

// ── TokenAuthProvider (JWT via /api/v1/auth/login) ───────────────────────────

#[cfg(all(feature = "async", feature = "networking"))]
pub(crate) struct TokenAuthProvider {
    endpoint: String,
    access_key: String,
    secret_key: String,
    client: Client,
    /// Cached JWT + expiry (seconds since UNIX epoch).
    cache: Mutex<Option<(String, u64)>>,
}

#[cfg(all(feature = "async", feature = "networking"))]
impl TokenAuthProvider {
    pub fn new(endpoint: &str, access_key: &str, secret_key: &str) -> Self {
        let client = Client::builder()
            .timeout(std::time::Duration::from_secs(10))
            .build()
            .unwrap_or_default();
        Self {
            endpoint: endpoint.trim_end_matches('/').to_string(),
            access_key: access_key.to_string(),
            secret_key: secret_key.to_string(),
            client,
            cache: Mutex::new(None),
        }
    }

    /// Fetch (or refresh) the JWT.
    async fn fetch_token(&self) -> Result<(String, u64), StorageError> {
        #[derive(Serialize)]
        struct LoginRequest {
            access_key_id: String,
            secret_access_key: String,
        }

        #[derive(Deserialize)]
        struct LoginResponse {
            token: String,
            #[serde(default)]
            token_expiration: Option<u64>,
        }

        let url = format!("{}/api/v1/auth/login", self.endpoint);
        let body = LoginRequest {
            access_key_id: self.access_key.clone(),
            secret_access_key: self.secret_key.clone(),
        };

        let resp = self
            .client
            .post(&url)
            .json(&body)
            .send()
            .await
            .map_err(|e| StorageError::ConnectionError(format!("Token login failed: {}", e)))?;

        if !resp.status().is_success() {
            let status = resp.status();
            let text = resp.text().await.unwrap_or_default();
            return Err(StorageError::ConnectionError(format!(
                "Token login returned {}: {}",
                status, text
            )));
        }

        let login: LoginResponse = resp.json().await.map_err(|e| {
            StorageError::ConnectionError(format!("Failed to parse login response: {}", e))
        })?;

        // Default expiry: 1 hour from now
        let now_secs = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap_or_default()
            .as_secs();
        let expiry = login.token_expiration.unwrap_or(now_secs + 3600);

        Ok((login.token, expiry))
    }
}

#[cfg(all(feature = "async", feature = "networking"))]
#[async_trait::async_trait]
impl AuthProvider for TokenAuthProvider {
    async fn auth_header(&self) -> Result<HeaderValue, StorageError> {
        // Check cache — refresh if expired or within 60s of expiry.
        let needs_refresh = {
            let cache = self.cache.lock().unwrap();
            match &*cache {
                Some((_token, expiry)) => {
                    let now = std::time::SystemTime::now()
                        .duration_since(std::time::UNIX_EPOCH)
                        .unwrap_or_default()
                        .as_secs();
                    now + 60 >= *expiry
                }
                None => true,
            }
        };

        if needs_refresh {
            let (token, expiry) = self.fetch_token().await?;
            let mut cache = self.cache.lock().unwrap();
            *cache = Some((token, expiry));
        }

        let cache = self.cache.lock().unwrap();
        let (token, _) = cache.as_ref().unwrap();
        let header_str = format!("Bearer {}", token);
        HeaderValue::from_str(&header_str)
            .map_err(|e| StorageError::ConnectionError(format!("Invalid Bearer header: {}", e)))
    }

    fn mode_name(&self) -> &'static str {
        "token"
    }
}

// ── StsAuthProvider (POST /sts/login with OIDC token) ────────────────────────

#[cfg(all(feature = "async", feature = "networking"))]
pub(crate) struct StsAuthProvider {
    endpoint: String,
    oidc_token: String,
    client: Client,
    /// Cached temp credentials: (access_key, secret_key, session_token, expiry_secs).
    cache: Mutex<Option<(String, String, String, u64)>>,
}

#[cfg(all(feature = "async", feature = "networking"))]
impl StsAuthProvider {
    pub fn new(endpoint: &str, oidc_token: &str) -> Self {
        let client = Client::builder()
            .timeout(std::time::Duration::from_secs(10))
            .build()
            .unwrap_or_default();
        Self {
            endpoint: endpoint.trim_end_matches('/').to_string(),
            oidc_token: oidc_token.to_string(),
            client,
            cache: Mutex::new(None),
        }
    }

    async fn fetch_temp_credentials(&self) -> Result<(String, String, String, u64), StorageError> {
        #[derive(Deserialize)]
        struct StsResponse {
            #[serde(rename = "Credentials")]
            credentials: StsCredentials,
        }

        #[derive(Deserialize)]
        struct StsCredentials {
            #[serde(rename = "AccessKeyId")]
            access_key_id: String,
            #[serde(rename = "SecretAccessKey")]
            secret_access_key: String,
            #[serde(rename = "SessionToken")]
            session_token: String,
            #[serde(rename = "Expiration", default)]
            expiration: Option<u64>,
        }

        let url = format!("{}/sts/login", self.endpoint);

        let resp = self
            .client
            .post(&url)
            .header("Content-Type", "application/x-www-form-urlencoded")
            .body(format!(
                "Action=AssumeRoleWithWebIdentity&WebIdentityToken={}&Version=2011-06-15",
                self.oidc_token
            ))
            .send()
            .await
            .map_err(|e| StorageError::ConnectionError(format!("STS login failed: {}", e)))?;

        if !resp.status().is_success() {
            let status = resp.status();
            let text = resp.text().await.unwrap_or_default();
            return Err(StorageError::ConnectionError(format!(
                "STS login returned {}: {}",
                status, text
            )));
        }

        let sts: StsResponse = resp.json().await.map_err(|e| {
            StorageError::ConnectionError(format!("Failed to parse STS response: {}", e))
        })?;

        let now_secs = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap_or_default()
            .as_secs();
        let expiry = sts.credentials.expiration.unwrap_or(now_secs + 3600);

        Ok((
            sts.credentials.access_key_id,
            sts.credentials.secret_access_key,
            sts.credentials.session_token,
            expiry,
        ))
    }
}

#[cfg(all(feature = "async", feature = "networking"))]
#[async_trait::async_trait]
impl AuthProvider for StsAuthProvider {
    async fn auth_header(&self) -> Result<HeaderValue, StorageError> {
        let needs_refresh = {
            let cache = self.cache.lock().unwrap();
            match &*cache {
                Some((_, _, _, expiry)) => {
                    let now = std::time::SystemTime::now()
                        .duration_since(std::time::UNIX_EPOCH)
                        .unwrap_or_default()
                        .as_secs();
                    now + 60 >= *expiry
                }
                None => true,
            }
        };

        if needs_refresh {
            let creds = self.fetch_temp_credentials().await?;
            let mut cache = self.cache.lock().unwrap();
            *cache = Some(creds);
        }

        // STS temp credentials use Basic auth with the temp access_key:secret_key.
        // The session token is sent as `X-Lakefs-Session-Token` but since we only
        // control the Authorization header here, we encode access_key:secret_key.
        let cache = self.cache.lock().unwrap();
        let (ak, sk, _session_token, _) = cache.as_ref().unwrap();
        let credentials = format!("{}:{}", ak, sk);
        let encoded = general_purpose::STANDARD.encode(credentials.as_bytes());
        let header_str = format!("Basic {}", encoded);
        HeaderValue::from_str(&header_str)
            .map_err(|e| StorageError::ConnectionError(format!("Invalid STS auth header: {}", e)))
    }

    fn mode_name(&self) -> &'static str {
        "sts"
    }
}

// ── IamAuthProvider (AWS instance metadata / ECS task role) ──────────────────

#[cfg(all(feature = "async", feature = "networking"))]
pub(crate) struct IamAuthProvider {
    client: Client,
    /// Cached credentials: (access_key, secret_key, token, expiry_secs).
    cache: Mutex<Option<IamCredentialCache>>,
}

#[cfg(all(feature = "async", feature = "networking"))]
type IamCredentialCache = (String, String, Option<String>, u64);

#[cfg(all(feature = "async", feature = "networking"))]
impl IamAuthProvider {
    pub fn new() -> Self {
        let client = Client::builder()
            .timeout(std::time::Duration::from_secs(5))
            .build()
            .unwrap_or_default();
        Self {
            client,
            cache: Mutex::new(None),
        }
    }

    /// Try ECS container credentials first, then fall back to EC2 instance metadata.
    async fn fetch_credentials(&self) -> Result<IamCredentialCache, StorageError> {
        // 1. Try ECS container credentials endpoint
        if let Ok(creds_uri) = std::env::var("AWS_CONTAINER_CREDENTIALS_RELATIVE_URI") {
            let url = format!("http://169.254.170.2{}", creds_uri);
            if let Ok(resp) = self.client.get(&url).send().await {
                if resp.status().is_success() {
                    if let Ok(creds) = resp.json::<IamCredentials>().await {
                        let now_secs = std::time::SystemTime::now()
                            .duration_since(std::time::UNIX_EPOCH)
                            .unwrap_or_default()
                            .as_secs();
                        return Ok((
                            creds.access_key_id,
                            creds.secret_access_key,
                            creds.token,
                            now_secs + 3600,
                        ));
                    }
                }
            }
        }

        // 2. Fall back to EC2 instance metadata (IMDSv2)
        let token_resp = self
            .client
            .put("http://169.254.169.254/latest/api/token")
            .header("X-aws-ec2-metadata-token-ttl-seconds", "21600")
            .send()
            .await
            .map_err(|e| {
                StorageError::ConnectionError(format!("IAM: Failed to fetch IMDS token: {}", e))
            })?;

        let imds_token = token_resp.text().await.map_err(|e| {
            StorageError::ConnectionError(format!("IAM: Failed to read IMDS token: {}", e))
        })?;

        // Get role name
        let role_resp = self
            .client
            .get("http://169.254.169.254/latest/meta-data/iam/security-credentials/")
            .header("X-aws-ec2-metadata-token", &imds_token)
            .send()
            .await
            .map_err(|e| {
                StorageError::ConnectionError(format!("IAM: Failed to fetch role name: {}", e))
            })?;

        let role_name = role_resp.text().await.map_err(|e| {
            StorageError::ConnectionError(format!("IAM: Failed to read role name: {}", e))
        })?;
        let role_name = role_name.trim();

        // Get credentials for role
        let creds_url = format!(
            "http://169.254.169.254/latest/meta-data/iam/security-credentials/{}",
            role_name
        );
        let creds_resp = self
            .client
            .get(&creds_url)
            .header("X-aws-ec2-metadata-token", &imds_token)
            .send()
            .await
            .map_err(|e| {
                StorageError::ConnectionError(format!("IAM: Failed to fetch credentials: {}", e))
            })?;

        let creds: IamCredentials = creds_resp.json().await.map_err(|e| {
            StorageError::ConnectionError(format!("IAM: Failed to parse credentials: {}", e))
        })?;

        let now_secs = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap_or_default()
            .as_secs();

        Ok((
            creds.access_key_id,
            creds.secret_access_key,
            creds.token,
            now_secs + 3600,
        ))
    }
}

#[cfg(all(feature = "async", feature = "networking"))]
#[derive(Deserialize)]
struct IamCredentials {
    #[serde(rename = "AccessKeyId", alias = "accessKeyId")]
    access_key_id: String,
    #[serde(rename = "SecretAccessKey", alias = "secretAccessKey")]
    secret_access_key: String,
    #[serde(rename = "Token", alias = "token", alias = "SessionToken", default)]
    token: Option<String>,
}

#[cfg(all(feature = "async", feature = "networking"))]
#[async_trait::async_trait]
impl AuthProvider for IamAuthProvider {
    async fn auth_header(&self) -> Result<HeaderValue, StorageError> {
        let needs_refresh = {
            let cache = self.cache.lock().unwrap();
            match &*cache {
                Some((_, _, _, expiry)) => {
                    let now = std::time::SystemTime::now()
                        .duration_since(std::time::UNIX_EPOCH)
                        .unwrap_or_default()
                        .as_secs();
                    now + 60 >= *expiry
                }
                None => true,
            }
        };

        if needs_refresh {
            let creds = self.fetch_credentials().await?;
            let mut cache = self.cache.lock().unwrap();
            *cache = Some(creds);
        }

        let cache = self.cache.lock().unwrap();
        let (ak, sk, _, _) = cache.as_ref().unwrap();
        let credentials = format!("{}:{}", ak, sk);
        let encoded = general_purpose::STANDARD.encode(credentials.as_bytes());
        let header_str = format!("Basic {}", encoded);
        HeaderValue::from_str(&header_str)
            .map_err(|e| StorageError::ConnectionError(format!("Invalid IAM auth header: {}", e)))
    }

    fn mode_name(&self) -> &'static str {
        "iam"
    }
}

// ── OidcAuthProvider (POST /api/v1/oidc/login) ──────────────────────────────

#[cfg(all(feature = "async", feature = "networking"))]
pub(crate) struct OidcAuthProvider {
    endpoint: String,
    external_token: String,
    client: Client,
    /// Cached lakeFS JWT + expiry.
    cache: Mutex<Option<(String, u64)>>,
}

#[cfg(all(feature = "async", feature = "networking"))]
impl OidcAuthProvider {
    pub fn new(endpoint: &str, external_token: &str) -> Self {
        let client = Client::builder()
            .timeout(std::time::Duration::from_secs(10))
            .build()
            .unwrap_or_default();
        Self {
            endpoint: endpoint.trim_end_matches('/').to_string(),
            external_token: external_token.to_string(),
            client,
            cache: Mutex::new(None),
        }
    }

    async fn exchange_token(&self) -> Result<(String, u64), StorageError> {
        #[derive(Deserialize)]
        struct OidcResponse {
            token: String,
            #[serde(default)]
            token_expiration: Option<u64>,
        }

        let url = format!("{}/api/v1/oidc/login", self.endpoint);

        let resp = self
            .client
            .post(&url)
            .header("Content-Type", "application/x-www-form-urlencoded")
            .body(format!("code={}", self.external_token))
            .send()
            .await
            .map_err(|e| StorageError::ConnectionError(format!("OIDC login failed: {}", e)))?;

        if !resp.status().is_success() {
            let status = resp.status();
            let text = resp.text().await.unwrap_or_default();
            return Err(StorageError::ConnectionError(format!(
                "OIDC login returned {}: {}",
                status, text
            )));
        }

        let oidc: OidcResponse = resp.json().await.map_err(|e| {
            StorageError::ConnectionError(format!("Failed to parse OIDC response: {}", e))
        })?;

        let now_secs = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap_or_default()
            .as_secs();
        let expiry = oidc.token_expiration.unwrap_or(now_secs + 3600);

        Ok((oidc.token, expiry))
    }
}

#[cfg(all(feature = "async", feature = "networking"))]
#[async_trait::async_trait]
impl AuthProvider for OidcAuthProvider {
    async fn auth_header(&self) -> Result<HeaderValue, StorageError> {
        let needs_refresh = {
            let cache = self.cache.lock().unwrap();
            match &*cache {
                Some((_token, expiry)) => {
                    let now = std::time::SystemTime::now()
                        .duration_since(std::time::UNIX_EPOCH)
                        .unwrap_or_default()
                        .as_secs();
                    now + 60 >= *expiry
                }
                None => true,
            }
        };

        if needs_refresh {
            let (token, expiry) = self.exchange_token().await?;
            let mut cache = self.cache.lock().unwrap();
            *cache = Some((token, expiry));
        }

        let cache = self.cache.lock().unwrap();
        let (token, _) = cache.as_ref().unwrap();
        let header_str = format!("Bearer {}", token);
        HeaderValue::from_str(&header_str).map_err(|e| {
            StorageError::ConnectionError(format!("Invalid OIDC Bearer header: {}", e))
        })
    }

    fn mode_name(&self) -> &'static str {
        "oidc"
    }
}

// ── SamlAuthProvider (POST /api/v1/auth/external/saml) ──────────────────────

#[cfg(all(feature = "async", feature = "networking"))]
pub(crate) struct SamlAuthProvider {
    endpoint: String,
    assertion: String,
    client: Client,
    /// Cached lakeFS JWT + expiry.
    cache: Mutex<Option<(String, u64)>>,
}

#[cfg(all(feature = "async", feature = "networking"))]
impl SamlAuthProvider {
    pub fn new(endpoint: &str, assertion: &str) -> Self {
        let client = Client::builder()
            .timeout(std::time::Duration::from_secs(10))
            .build()
            .unwrap_or_default();
        Self {
            endpoint: endpoint.trim_end_matches('/').to_string(),
            assertion: assertion.to_string(),
            client,
            cache: Mutex::new(None),
        }
    }

    async fn exchange_assertion(&self) -> Result<(String, u64), StorageError> {
        #[derive(Deserialize)]
        struct SamlResponse {
            token: String,
            #[serde(default)]
            token_expiration: Option<u64>,
        }

        let url = format!("{}/api/v1/auth/external/saml", self.endpoint);

        let resp = self
            .client
            .post(&url)
            .header("Content-Type", "application/x-www-form-urlencoded")
            .body(format!("SAMLResponse={}", self.assertion))
            .send()
            .await
            .map_err(|e| StorageError::ConnectionError(format!("SAML login failed: {}", e)))?;

        if !resp.status().is_success() {
            let status = resp.status();
            let text = resp.text().await.unwrap_or_default();
            return Err(StorageError::ConnectionError(format!(
                "SAML login returned {}: {}",
                status, text
            )));
        }

        let saml: SamlResponse = resp.json().await.map_err(|e| {
            StorageError::ConnectionError(format!("Failed to parse SAML response: {}", e))
        })?;

        let now_secs = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap_or_default()
            .as_secs();
        let expiry = saml.token_expiration.unwrap_or(now_secs + 3600);

        Ok((saml.token, expiry))
    }
}

#[cfg(all(feature = "async", feature = "networking"))]
#[async_trait::async_trait]
impl AuthProvider for SamlAuthProvider {
    async fn auth_header(&self) -> Result<HeaderValue, StorageError> {
        let needs_refresh = {
            let cache = self.cache.lock().unwrap();
            match &*cache {
                Some((_token, expiry)) => {
                    let now = std::time::SystemTime::now()
                        .duration_since(std::time::UNIX_EPOCH)
                        .unwrap_or_default()
                        .as_secs();
                    now + 60 >= *expiry
                }
                None => true,
            }
        };

        if needs_refresh {
            let (token, expiry) = self.exchange_assertion().await?;
            let mut cache = self.cache.lock().unwrap();
            *cache = Some((token, expiry));
        }

        let cache = self.cache.lock().unwrap();
        let (token, _) = cache.as_ref().unwrap();
        let header_str = format!("Bearer {}", token);
        HeaderValue::from_str(&header_str).map_err(|e| {
            StorageError::ConnectionError(format!("Invalid SAML Bearer header: {}", e))
        })
    }

    fn mode_name(&self) -> &'static str {
        "saml"
    }
}

// ── Build AuthProvider from LakeFSAuth enum ─────────────────────────────────

#[cfg(all(feature = "async", feature = "networking"))]
fn build_auth_provider(
    auth: &LakeFSAuth,
    endpoint: &str,
) -> Result<Arc<dyn AuthProvider>, StorageError> {
    match auth {
        LakeFSAuth::Basic {
            access_key,
            secret_key,
        } => Ok(Arc::new(BasicAuthProvider::new(access_key, secret_key)?)),
        LakeFSAuth::Token {
            access_key,
            secret_key,
        } => Ok(Arc::new(TokenAuthProvider::new(
            endpoint, access_key, secret_key,
        ))),
        LakeFSAuth::Sts { oidc_token } => Ok(Arc::new(StsAuthProvider::new(endpoint, oidc_token))),
        LakeFSAuth::Iam => Ok(Arc::new(IamAuthProvider::new())),
        LakeFSAuth::Oidc { token } => Ok(Arc::new(OidcAuthProvider::new(endpoint, token))),
        LakeFSAuth::Saml { assertion } => Ok(Arc::new(SamlAuthProvider::new(endpoint, assertion))),
    }
}

// ─── New response / request types ────────────────────────────────────────────

/// Result of a branch merge operation.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MergeResult {
    /// The commit ID produced by the merge (empty string on conflict).
    pub commit_id: String,
    /// Summary text returned by LakeFS.
    pub summary: String,
}

/// Information about a single LakeFS commit.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CommitInfo {
    pub id: String,
    pub message: String,
    pub committer: String,
    #[serde(default)]
    pub metadata: HashMap<String, String>,
    /// Unix-epoch seconds.  LakeFS returns `creation_date` as an i64.
    #[serde(default)]
    pub creation_date: i64,
}

/// A single entry in a diff listing.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DiffEntry {
    /// Object path.
    pub path: String,
    /// "added", "removed", "changed", or "conflict".
    #[serde(rename = "type")]
    pub diff_type: String,
    /// Size in bytes (only for added / changed).
    #[serde(default)]
    pub size_bytes: Option<u64>,
}

/// Object metadata returned by the stat endpoint.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ObjectStats {
    pub path: String,
    #[serde(default)]
    pub physical_address: String,
    pub size_bytes: u64,
    #[serde(default)]
    pub content_type: String,
    /// Unix-epoch seconds.
    #[serde(default)]
    pub mtime: i64,
    #[serde(default)]
    pub checksum: String,
}

/// Describes a LakeFS Action (webhook) to register.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct LakeFSAction {
    pub name: String,
    /// e.g. `["pre-commit", "post-merge"]`
    pub on: Vec<String>,
    pub hooks: Vec<ActionHook>,
}

/// A single hook inside a LakeFS Action.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ActionHook {
    pub id: String,
    #[serde(rename = "type")]
    pub hook_type: String,
    pub description: String,
    #[serde(default)]
    pub properties: HashMap<String, String>,
}

/// Repository information returned by LakeFS.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RepositoryInfo {
    pub id: String,
    pub storage_namespace: String,
    pub default_branch: String,
    #[serde(default)]
    pub creation_date: i64,
}

/// Branch reference returned by LakeFS.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct BranchInfo {
    pub id: String,
    pub commit_id: String,
}

// ─── Core backend ────────────────────────────────────────────────────────────

#[cfg(all(feature = "async", feature = "networking"))]
#[derive(Clone)]
pub struct LakeFSBackend {
    client: Client,
    endpoint: String,
    repository: String, // "engagement" in Briefcase terminology
    branch: String,     // "workstream" in Briefcase terminology
    #[allow(dead_code)]
    access_key: String,
    #[allow(dead_code)]
    secret_key: String,
    /// Per-request auth provider.  Replaces static default headers so that
    /// token-based modes (JWT, STS, OIDC, SAML) can refresh transparently.
    auth_provider: Arc<dyn AuthProvider>,
    pending_writes: Arc<Mutex<Vec<Snapshot>>>,
}

#[cfg(all(feature = "async", feature = "networking"))]
impl LakeFSBackend {
    pub fn new(config: LakeFSConfig) -> Result<Self, StorageError> {
        let endpoint = config.endpoint.trim_end_matches('/').to_string();

        // Build auth provider from explicit config or default to Basic.
        let auth = config.auth.clone().unwrap_or_else(|| LakeFSAuth::Basic {
            access_key: config.access_key.clone(),
            secret_key: config.secret_key.clone(),
        });
        let auth_provider = build_auth_provider(&auth, &endpoint)?;

        // Default headers — Content-Type only.  Authorization is per-request.
        let mut headers = HeaderMap::new();
        headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));

        let client = Client::builder()
            .default_headers(headers)
            .timeout(std::time::Duration::from_secs(30))
            .build()
            .map_err(|e| {
                StorageError::ConnectionError(format!("Failed to create HTTP client: {}", e))
            })?;

        Ok(Self {
            client,
            endpoint,
            repository: config.repository,
            branch: config.branch,
            access_key: config.access_key,
            secret_key: config.secret_key,
            auth_provider,
            pending_writes: Arc::new(Mutex::new(Vec::new())),
        })
    }

    /// Send an HTTP request with the current authorization header.
    ///
    /// All lakeFS API calls route through this method so that token refresh
    /// happens transparently for JWT/STS/OIDC/SAML modes.
    async fn send_authed(&self, builder: RequestBuilder) -> Result<Response, StorageError> {
        let auth_header = self.auth_provider.auth_header().await?;
        builder
            .header(AUTHORIZATION, auth_header)
            .send()
            .await
            .map_err(|e| StorageError::ConnectionError(format!("Request failed: {}", e)))
    }

    /// Returns the active authentication mode name (for logging/diagnostics).
    pub fn auth_mode(&self) -> &'static str {
        self.auth_provider.mode_name()
    }

    // ── Accessors ────────────────────────────────────────────────────────

    /// Returns the configured repository name.
    pub fn repository(&self) -> &str {
        &self.repository
    }

    /// Returns the configured branch name.
    pub fn branch(&self) -> &str {
        &self.branch
    }

    /// Returns the configured endpoint URL.
    pub fn endpoint(&self) -> &str {
        &self.endpoint
    }

    // ── Commit operations ────────────────────────────────────────────────

    /// Create a commit (checkpoint) with pending writes.
    pub async fn create_commit(&self, message: &str) -> Result<String, StorageError> {
        let url = format!(
            "{}/api/v1/repositories/{}/branches/{}/commits",
            self.endpoint, self.repository, self.branch
        );

        #[derive(Serialize)]
        struct CommitRequest {
            message: String,
            metadata: HashMap<String, String>,
        }

        let mut metadata = HashMap::new();
        metadata.insert("source".to_string(), "briefcase-ai".to_string());

        let request = CommitRequest {
            message: message.to_string(),
            metadata,
        };

        let response = self
            .send_authed(self.client.post(&url).json(&request))
            .await?;

        let status = response.status();
        if !status.is_success() {
            let error_text = response.text().await.unwrap_or_default();
            return Err(StorageError::ConnectionError(format!(
                "Commit failed with status {}: {}",
                status, error_text
            )));
        }

        #[derive(Deserialize)]
        struct CommitResponse {
            id: String,
        }

        let commit_response: CommitResponse = response.json().await.map_err(|e| {
            StorageError::SerializationError(format!("Failed to parse commit response: {}", e))
        })?;

        Ok(commit_response.id)
    }

    // ── Object operations ────────────────────────────────────────────────

    /// Upload object to LakeFS.
    pub async fn upload_object(&self, path: &str, data: &[u8]) -> Result<(), StorageError> {
        let url = format!(
            "{}/api/v1/repositories/{}/branches/{}/objects",
            self.endpoint, self.repository, self.branch
        );

        let response = self
            .send_authed(
                self.client
                    .put(&url)
                    .query(&[("path", path)])
                    .header("Content-Type", "application/octet-stream")
                    .body(data.to_vec()),
            )
            .await?;

        let status = response.status();
        if !status.is_success() {
            let error_text = response.text().await.unwrap_or_default();
            return Err(StorageError::ConnectionError(format!(
                "Upload failed with status {}: {}",
                status, error_text
            )));
        }

        Ok(())
    }

    /// Download object from LakeFS.
    pub async fn download_object(&self, path: &str) -> Result<Vec<u8>, StorageError> {
        let url = format!(
            "{}/api/v1/repositories/{}/refs/{}/objects",
            self.endpoint, self.repository, self.branch
        );

        let response = self
            .send_authed(self.client.get(&url).query(&[("path", path)]))
            .await?;

        if response.status() == reqwest::StatusCode::NOT_FOUND {
            return Err(StorageError::NotFound(format!(
                "Object not found: {}",
                path
            )));
        }

        let status = response.status();
        if !status.is_success() {
            let error_text = response.text().await.unwrap_or_default();
            return Err(StorageError::ConnectionError(format!(
                "Download failed with status {}: {}",
                status, error_text
            )));
        }

        let data = response.bytes().await.map_err(|e| {
            StorageError::ConnectionError(format!("Failed to read response: {}", e))
        })?;

        Ok(data.to_vec())
    }

    /// List objects with prefix.
    pub async fn list_objects(&self, prefix: &str) -> Result<Vec<String>, StorageError> {
        let url = format!(
            "{}/api/v1/repositories/{}/refs/{}/objects/ls",
            self.endpoint, self.repository, self.branch
        );

        let response = self
            .send_authed(self.client.get(&url).query(&[("prefix", prefix)]))
            .await?;

        let status = response.status();
        if !status.is_success() {
            let error_text = response.text().await.unwrap_or_default();
            return Err(StorageError::ConnectionError(format!(
                "List failed with status {}: {}",
                status, error_text
            )));
        }

        #[derive(Deserialize)]
        struct ListResponse {
            results: Vec<ObjectInfo>,
        }

        #[derive(Deserialize)]
        struct ObjectInfo {
            path: String,
            #[serde(rename = "type")]
            object_type: String,
        }

        let list_response: ListResponse = response.json().await.map_err(|e| {
            StorageError::SerializationError(format!("Failed to parse list response: {}", e))
        })?;

        let paths = list_response
            .results
            .into_iter()
            .filter(|obj| obj.object_type == "object")
            .map(|obj| obj.path)
            .collect();

        Ok(paths)
    }

    /// Delete an object from the current branch.
    pub async fn delete_object(&self, path: &str) -> Result<(), StorageError> {
        if path.is_empty() {
            return Err(StorageError::InvalidQuery(
                "Object path cannot be empty".to_string(),
            ));
        }

        let url = format!(
            "{}/api/v1/repositories/{}/branches/{}/objects",
            self.endpoint, self.repository, self.branch
        );

        let response = self
            .send_authed(self.client.delete(&url).query(&[("path", path)]))
            .await?;

        if response.status() == reqwest::StatusCode::NOT_FOUND {
            return Err(StorageError::NotFound(format!(
                "Object not found: {}",
                path
            )));
        }

        let status = response.status();
        if !status.is_success() {
            let error_text = response.text().await.unwrap_or_default();
            return Err(StorageError::ConnectionError(format!(
                "Delete failed with status {}: {}",
                status, error_text
            )));
        }

        Ok(())
    }

    /// Get object statistics/metadata.
    pub async fn get_object_stats(&self, path: &str) -> Result<ObjectStats, StorageError> {
        if path.is_empty() {
            return Err(StorageError::InvalidQuery(
                "Object path cannot be empty".to_string(),
            ));
        }

        let url = format!(
            "{}/api/v1/repositories/{}/refs/{}/objects/stat",
            self.endpoint, self.repository, self.branch
        );

        let response = self
            .send_authed(self.client.get(&url).query(&[("path", path)]))
            .await?;

        if response.status() == reqwest::StatusCode::NOT_FOUND {
            return Err(StorageError::NotFound(format!(
                "Object not found: {}",
                path
            )));
        }

        let status = response.status();
        if !status.is_success() {
            let error_text = response.text().await.unwrap_or_default();
            return Err(StorageError::ConnectionError(format!(
                "Stats failed with status {}: {}",
                status, error_text
            )));
        }

        let stats: ObjectStats = response.json().await.map_err(|e| {
            StorageError::SerializationError(format!("Failed to parse stats response: {}", e))
        })?;

        Ok(stats)
    }

    // ── Repository operations ────────────────────────────────────────────

    /// Create a new LakeFS repository (engagement).
    pub async fn create_repository(
        &self,
        name: &str,
        storage_namespace: &str,
        default_branch: Option<&str>,
    ) -> Result<RepositoryInfo, StorageError> {
        if name.is_empty() {
            return Err(StorageError::InvalidQuery(
                "Repository name cannot be empty".to_string(),
            ));
        }
        if storage_namespace.is_empty() {
            return Err(StorageError::InvalidQuery(
                "Storage namespace cannot be empty".to_string(),
            ));
        }

        let url = format!("{}/api/v1/repositories", self.endpoint);

        #[derive(Serialize)]
        struct CreateRepoRequest {
            name: String,
            storage_namespace: String,
            default_branch: String,
        }

        let request = CreateRepoRequest {
            name: name.to_string(),
            storage_namespace: storage_namespace.to_string(),
            default_branch: default_branch.unwrap_or("main").to_string(),
        };

        let response = self
            .send_authed(self.client.post(&url).json(&request))
            .await?;

        if response.status() == reqwest::StatusCode::CONFLICT {
            return Err(StorageError::ConnectionError(format!(
                "Repository '{}' already exists",
                name
            )));
        }

        let status = response.status();
        if !status.is_success() {
            let error_text = response.text().await.unwrap_or_default();
            return Err(StorageError::ConnectionError(format!(
                "Create repository failed with status {}: {}",
                status, error_text
            )));
        }

        let info: RepositoryInfo = response.json().await.map_err(|e| {
            StorageError::SerializationError(format!("Failed to parse repository response: {}", e))
        })?;

        Ok(info)
    }

    /// Delete a LakeFS repository.
    pub async fn delete_repository(&self, name: &str) -> Result<(), StorageError> {
        if name.is_empty() {
            return Err(StorageError::InvalidQuery(
                "Repository name cannot be empty".to_string(),
            ));
        }

        let url = format!("{}/api/v1/repositories/{}", self.endpoint, name);

        let response = self.send_authed(self.client.delete(&url)).await?;

        if response.status() == reqwest::StatusCode::NOT_FOUND {
            return Err(StorageError::NotFound(format!(
                "Repository not found: {}",
                name
            )));
        }

        let status = response.status();
        if !status.is_success() {
            let error_text = response.text().await.unwrap_or_default();
            return Err(StorageError::ConnectionError(format!(
                "Delete repository failed with status {}: {}",
                status, error_text
            )));
        }

        Ok(())
    }

    // ── Branch operations ────────────────────────────────────────────────

    /// Create a new branch (workstream) from a source branch.
    pub async fn create_branch(
        &self,
        name: &str,
        source_branch: &str,
    ) -> Result<String, StorageError> {
        if name.is_empty() {
            return Err(StorageError::InvalidQuery(
                "Branch name cannot be empty".to_string(),
            ));
        }
        if source_branch.is_empty() {
            return Err(StorageError::InvalidQuery(
                "Source branch cannot be empty".to_string(),
            ));
        }

        let url = format!(
            "{}/api/v1/repositories/{}/branches",
            self.endpoint, self.repository
        );

        #[derive(Serialize)]
        struct CreateBranchRequest {
            name: String,
            source: String,
        }

        let request = CreateBranchRequest {
            name: name.to_string(),
            source: source_branch.to_string(),
        };

        let response = self
            .send_authed(self.client.post(&url).json(&request))
            .await?;

        if response.status() == reqwest::StatusCode::CONFLICT {
            return Err(StorageError::ConnectionError(format!(
                "Branch '{}' already exists",
                name
            )));
        }

        let status = response.status();
        if !status.is_success() {
            let error_text = response.text().await.unwrap_or_default();
            return Err(StorageError::ConnectionError(format!(
                "Create branch failed with status {}: {}",
                status, error_text
            )));
        }

        // LakeFS returns the commit ID the new branch points to as a plain string
        let commit_id = response.text().await.map_err(|e| {
            StorageError::SerializationError(format!("Failed to read branch response: {}", e))
        })?;

        // Strip surrounding quotes if present (LakeFS returns JSON string)
        let commit_id = commit_id.trim().trim_matches('"').to_string();
        Ok(commit_id)
    }

    /// Delete a branch.
    pub async fn delete_branch(&self, name: &str) -> Result<(), StorageError> {
        if name.is_empty() {
            return Err(StorageError::InvalidQuery(
                "Branch name cannot be empty".to_string(),
            ));
        }

        let url = format!(
            "{}/api/v1/repositories/{}/branches/{}",
            self.endpoint, self.repository, name
        );

        let response = self.send_authed(self.client.delete(&url)).await?;

        if response.status() == reqwest::StatusCode::NOT_FOUND {
            return Err(StorageError::NotFound(format!(
                "Branch not found: {}",
                name
            )));
        }

        let status = response.status();
        if !status.is_success() {
            let error_text = response.text().await.unwrap_or_default();
            return Err(StorageError::ConnectionError(format!(
                "Delete branch failed with status {}: {}",
                status, error_text
            )));
        }

        Ok(())
    }

    /// Merge source branch into destination branch.
    pub async fn merge_branch(
        &self,
        source_branch: &str,
        destination_branch: &str,
        message: Option<&str>,
    ) -> Result<MergeResult, StorageError> {
        if source_branch.is_empty() || destination_branch.is_empty() {
            return Err(StorageError::InvalidQuery(
                "Source and destination branches cannot be empty".to_string(),
            ));
        }

        let url = format!(
            "{}/api/v1/repositories/{}/refs/{}/merge/{}",
            self.endpoint, self.repository, source_branch, destination_branch
        );

        #[derive(Serialize)]
        struct MergeRequest {
            message: String,
            metadata: HashMap<String, String>,
        }

        let mut metadata = HashMap::new();
        metadata.insert("source".to_string(), "briefcase-ai".to_string());

        let request = MergeRequest {
            message: message
                .unwrap_or(&format!(
                    "Merge {} into {}",
                    source_branch, destination_branch
                ))
                .to_string(),
            metadata,
        };

        let response = self
            .send_authed(self.client.post(&url).json(&request))
            .await?;

        if response.status() == reqwest::StatusCode::CONFLICT {
            return Err(StorageError::ConnectionError(
                "Merge conflict detected".to_string(),
            ));
        }

        let status = response.status();
        if !status.is_success() {
            let error_text = response.text().await.unwrap_or_default();
            return Err(StorageError::ConnectionError(format!(
                "Merge failed with status {}: {}",
                status, error_text
            )));
        }

        // LakeFS merge returns a reference (commit ID)
        #[derive(Deserialize)]
        struct MergeResponse {
            #[serde(default)]
            reference: String,
            #[serde(default)]
            summary: HashMap<String, serde_json::Value>,
        }

        let merge_resp: MergeResponse = response.json().await.map_err(|e| {
            StorageError::SerializationError(format!("Failed to parse merge response: {}", e))
        })?;

        Ok(MergeResult {
            commit_id: merge_resp.reference,
            summary: serde_json::to_string(&merge_resp.summary).unwrap_or_default(),
        })
    }

    // ── Commit operations ────────────────────────────────────────────────

    /// Get details of a specific commit.
    pub async fn get_commit(&self, commit_id: &str) -> Result<CommitInfo, StorageError> {
        if commit_id.is_empty() {
            return Err(StorageError::InvalidQuery(
                "Commit ID cannot be empty".to_string(),
            ));
        }

        let url = format!(
            "{}/api/v1/repositories/{}/commits/{}",
            self.endpoint, self.repository, commit_id
        );

        let response = self.send_authed(self.client.get(&url)).await?;

        if response.status() == reqwest::StatusCode::NOT_FOUND {
            return Err(StorageError::NotFound(format!(
                "Commit not found: {}",
                commit_id
            )));
        }

        let status = response.status();
        if !status.is_success() {
            let error_text = response.text().await.unwrap_or_default();
            return Err(StorageError::ConnectionError(format!(
                "Get commit failed with status {}: {}",
                status, error_text
            )));
        }

        let info: CommitInfo = response.json().await.map_err(|e| {
            StorageError::SerializationError(format!("Failed to parse commit response: {}", e))
        })?;

        Ok(info)
    }

    // ── Diff operations ──────────────────────────────────────────────────

    /// Diff between two LakeFS references (branches, tags, or commit IDs).
    pub async fn diff_refs(
        &self,
        left_ref: &str,
        right_ref: &str,
    ) -> Result<Vec<DiffEntry>, StorageError> {
        if left_ref.is_empty() || right_ref.is_empty() {
            return Err(StorageError::InvalidQuery(
                "Both refs must be non-empty".to_string(),
            ));
        }

        let url = format!(
            "{}/api/v1/repositories/{}/refs/{}/diff/{}",
            self.endpoint, self.repository, left_ref, right_ref
        );

        let response = self.send_authed(self.client.get(&url)).await?;

        let status = response.status();
        if !status.is_success() {
            let error_text = response.text().await.unwrap_or_default();
            return Err(StorageError::ConnectionError(format!(
                "Diff failed with status {}: {}",
                status, error_text
            )));
        }

        #[derive(Deserialize)]
        struct DiffResponse {
            results: Vec<DiffEntry>,
        }

        let diff_resp: DiffResponse = response.json().await.map_err(|e| {
            StorageError::SerializationError(format!("Failed to parse diff response: {}", e))
        })?;

        Ok(diff_resp.results)
    }

    // ── Actions (webhook) operations ─────────────────────────────────────

    /// Register a LakeFS Action by uploading its YAML definition to the
    /// `_lakefs_actions/` namespace on the current branch, then committing.
    pub async fn register_action(&self, action: &LakeFSAction) -> Result<String, StorageError> {
        if action.name.is_empty() {
            return Err(StorageError::InvalidQuery(
                "Action name cannot be empty".to_string(),
            ));
        }
        if action.on.is_empty() {
            return Err(StorageError::InvalidQuery(
                "Action must have at least one event trigger".to_string(),
            ));
        }
        if action.hooks.is_empty() {
            return Err(StorageError::InvalidQuery(
                "Action must have at least one hook".to_string(),
            ));
        }

        // Build the YAML content for the action
        let yaml_content = self.build_action_yaml(action);

        // Upload to _lakefs_actions/ namespace
        let path = format!("_lakefs_actions/{}.yaml", action.name);
        self.upload_object(&path, yaml_content.as_bytes()).await?;

        // Commit the action definition
        let commit_msg = format!("Register LakeFS Action: {}", action.name);
        let commit_id = self.create_commit(&commit_msg).await?;

        Ok(commit_id)
    }

    /// Build YAML string for a LakeFS Action definition.
    fn build_action_yaml(&self, action: &LakeFSAction) -> String {
        let mut yaml = String::new();
        yaml.push_str(&format!("name: {}\n", action.name));
        yaml.push_str("on:\n");
        for event in &action.on {
            yaml.push_str(&format!("  {}:\n", event));
        }
        yaml.push_str("hooks:\n");
        for hook in &action.hooks {
            yaml.push_str(&format!("  - id: {}\n", hook.id));
            yaml.push_str(&format!("    type: {}\n", hook.hook_type));
            yaml.push_str(&format!("    description: {}\n", hook.description));
            if !hook.properties.is_empty() {
                yaml.push_str("    properties:\n");
                for (k, v) in &hook.properties {
                    yaml.push_str(&format!("      {}: {}\n", k, v));
                }
            }
        }
        yaml
    }

    // ── Path helpers ─────────────────────────────────────────────────────

    /// Generate object path for snapshot.
    fn snapshot_path(&self, snapshot_id: &str) -> String {
        format!("snapshots/{}.json", snapshot_id)
    }

    /// Generate object path for decision.
    fn decision_path(&self, decision_id: &str) -> String {
        format!("decisions/{}.json", decision_id)
    }
}

// ─── StorageBackend trait impl ───────────────────────────────────────────────

#[cfg(all(feature = "async", feature = "networking"))]
#[async_trait::async_trait]
impl StorageBackend for LakeFSBackend {
    async fn save(&self, snapshot: &Snapshot) -> Result<String, StorageError> {
        let snapshot_id = snapshot.metadata.snapshot_id.to_string();

        // Add to pending writes instead of immediate upload
        {
            let mut pending = self.pending_writes.lock().unwrap();
            pending.push(snapshot.clone());
        }

        Ok(snapshot_id)
    }

    async fn save_decision(&self, decision: &DecisionSnapshot) -> Result<String, StorageError> {
        let decision_id = decision.metadata.snapshot_id.to_string();
        let path = self.decision_path(&decision_id);

        let json_data = serde_json::to_vec(decision)
            .map_err(|e| StorageError::SerializationError(e.to_string()))?;

        self.upload_object(&path, &json_data).await?;

        Ok(decision_id)
    }

    async fn load(&self, snapshot_id: &str) -> Result<Snapshot, StorageError> {
        let path = self.snapshot_path(snapshot_id);
        let data = self.download_object(&path).await?;

        let snapshot: Snapshot = serde_json::from_slice(&data)
            .map_err(|e| StorageError::SerializationError(e.to_string()))?;

        Ok(snapshot)
    }

    async fn load_decision(&self, decision_id: &str) -> Result<DecisionSnapshot, StorageError> {
        let path = self.decision_path(decision_id);
        let data = self.download_object(&path).await?;

        let decision: DecisionSnapshot = serde_json::from_slice(&data)
            .map_err(|e| StorageError::SerializationError(e.to_string()))?;

        Ok(decision)
    }

    async fn query(&self, query: SnapshotQuery) -> Result<Vec<Snapshot>, StorageError> {
        // List all snapshots
        let paths = self.list_objects("snapshots/").await?;

        let mut snapshots = Vec::new();
        let mut count = 0;
        let offset = query.offset.unwrap_or(0);
        let limit = query.limit.unwrap_or(usize::MAX);

        for path in paths {
            if let Some(filename) = path.split('/').next_back() {
                if let Some(snapshot_id) = filename.strip_suffix(".json") {
                    // Load snapshot to check filters
                    match self.load(snapshot_id).await {
                        Ok(snapshot) => {
                            // Apply filters
                            if self.matches_query(&snapshot, &query) {
                                if count >= offset {
                                    snapshots.push(snapshot);
                                    if snapshots.len() >= limit {
                                        break;
                                    }
                                }
                                count += 1;
                            }
                        }
                        Err(_) => continue, // Skip invalid snapshots
                    }
                }
            }
        }

        // Sort by timestamp (newest first)
        snapshots.sort_by(|a, b| b.metadata.timestamp.cmp(&a.metadata.timestamp));

        Ok(snapshots)
    }

    async fn delete(&self, snapshot_id: &str) -> Result<bool, StorageError> {
        let path = self.snapshot_path(snapshot_id);
        match self.delete_object(&path).await {
            Ok(()) => Ok(true),
            Err(StorageError::NotFound(_)) => Ok(false),
            Err(e) => Err(e),
        }
    }

    async fn flush(&self) -> Result<FlushResult, StorageError> {
        let pending_snapshots = {
            let mut pending = self.pending_writes.lock().unwrap();
            let snapshots = pending.clone();
            pending.clear();
            snapshots
        };

        if pending_snapshots.is_empty() {
            return Ok(FlushResult {
                snapshots_written: 0,
                bytes_written: 0,
                checkpoint_id: None,
            });
        }

        let mut bytes_written = 0;

        // Upload all pending snapshots
        for snapshot in &pending_snapshots {
            let snapshot_id = snapshot.metadata.snapshot_id.to_string();
            let path = self.snapshot_path(&snapshot_id);

            let json_data = serde_json::to_vec(snapshot)
                .map_err(|e| StorageError::SerializationError(e.to_string()))?;

            bytes_written += json_data.len();

            self.upload_object(&path, &json_data).await?;

            // Also upload individual decisions
            for decision in &snapshot.decisions {
                let decision_id = decision.metadata.snapshot_id.to_string();
                let decision_path = self.decision_path(&decision_id);

                let decision_data = serde_json::to_vec(decision)
                    .map_err(|e| StorageError::SerializationError(e.to_string()))?;

                bytes_written += decision_data.len();
                self.upload_object(&decision_path, &decision_data).await?;
            }
        }

        // Create commit
        let commit_message = format!("Briefcase AI flush: {} snapshots", pending_snapshots.len());
        let commit_id = self.create_commit(&commit_message).await?;

        Ok(FlushResult {
            snapshots_written: pending_snapshots.len(),
            bytes_written,
            checkpoint_id: Some(commit_id),
        })
    }

    async fn health_check(&self) -> Result<bool, StorageError> {
        let url = format!("{}/api/v1/repositories/{}", self.endpoint, self.repository);

        let response = self
            .send_authed(self.client.get(&url))
            .await
            .map_err(|e| StorageError::ConnectionError(format!("Health check failed: {}", e)))?;

        Ok(response.status().is_success())
    }
}

#[cfg(all(feature = "async", feature = "networking"))]
impl LakeFSBackend {
    /// Check if snapshot matches query filters
    fn matches_query(&self, snapshot: &Snapshot, query: &SnapshotQuery) -> bool {
        // Check time range
        if let Some(start_time) = query.start_time {
            if snapshot.metadata.timestamp < start_time {
                return false;
            }
        }

        if let Some(end_time) = query.end_time {
            if snapshot.metadata.timestamp > end_time {
                return false;
            }
        }

        // Check function name, module name, model name, tags in decisions
        if query.function_name.is_some()
            || query.module_name.is_some()
            || query.model_name.is_some()
            || query.tags.is_some()
        {
            let mut found_match = false;

            for decision in &snapshot.decisions {
                let mut decision_matches = true;

                if let Some(function_name) = &query.function_name {
                    if decision.function_name != *function_name {
                        decision_matches = false;
                    }
                }

                if let Some(module_name) = &query.module_name {
                    if decision.module_name.as_ref() != Some(module_name) {
                        decision_matches = false;
                    }
                }

                if let Some(model_name) = &query.model_name {
                    if let Some(model_params) = &decision.model_parameters {
                        if model_params.model_name != *model_name {
                            decision_matches = false;
                        }
                    } else {
                        decision_matches = false;
                    }
                }

                if let Some(query_tags) = &query.tags {
                    for (key, value) in query_tags {
                        if decision.tags.get(key) != Some(value) {
                            decision_matches = false;
                            break;
                        }
                    }
                }

                if decision_matches {
                    found_match = true;
                    break;
                }
            }

            if !found_match {
                return false;
            }
        }

        true
    }
}

// ─── Config ──────────────────────────────────────────────────────────────────

#[derive(Debug, Clone)]
pub struct LakeFSConfig {
    pub endpoint: String,
    pub repository: String,
    pub branch: String,
    pub access_key: String,
    pub secret_key: String,
    /// Optional explicit auth mode.  When `None`, falls back to
    /// `LakeFSAuth::Basic { access_key, secret_key }`.
    pub auth: Option<LakeFSAuth>,
}

impl LakeFSConfig {
    pub fn new(
        endpoint: impl Into<String>,
        repository: impl Into<String>,
        branch: impl Into<String>,
        access_key: impl Into<String>,
        secret_key: impl Into<String>,
    ) -> Self {
        Self {
            endpoint: endpoint.into(),
            repository: repository.into(),
            branch: branch.into(),
            access_key: access_key.into(),
            secret_key: secret_key.into(),
            auth: None,
        }
    }

    /// Set the authentication mode explicitly.
    pub fn with_auth(mut self, auth: LakeFSAuth) -> Self {
        self.auth = Some(auth);
        self
    }
}

// ─── Tests ───────────────────────────────────────────────────────────────────

#[cfg(test)]
mod tests {
    use super::*;
    use crate::models::*;
    use serde_json::json;
    use wiremock::matchers::{method, path, query_param};
    use wiremock::{Mock, MockServer, ResponseTemplate};

    fn create_test_config_with_endpoint(endpoint: &str) -> LakeFSConfig {
        LakeFSConfig::new(
            endpoint,
            "briefcase-test",
            "main",
            "test_key",
            "test_secret",
        )
    }

    fn create_test_config() -> LakeFSConfig {
        LakeFSConfig::new(
            "http://localhost:8000",
            "briefcase-test",
            "main",
            "test_access_key",
            "test_secret_key",
        )
    }

    async fn create_test_snapshot() -> Snapshot {
        let input = Input::new("test_input", json!("value"), "string");
        let output = Output::new("test_output", json!("result"), "string");
        let model_params = ModelParameters::new("gpt-4");

        let decision = DecisionSnapshot::new("test_function")
            .with_module("test_module")
            .add_input(input)
            .add_output(output)
            .with_model_parameters(model_params)
            .add_tag("env", "test");

        let mut snapshot = Snapshot::new(SnapshotType::Session);
        snapshot.add_decision(decision);
        snapshot
    }

    // ── Existing unit tests (non-HTTP) ───────────────────────────────────

    #[tokio::test]
    async fn test_lakefs_config_creation() {
        let config = create_test_config();
        assert_eq!(config.endpoint, "http://localhost:8000");
        assert_eq!(config.repository, "briefcase-test");
        assert_eq!(config.branch, "main");
    }

    #[tokio::test]
    async fn test_object_paths() {
        let config = create_test_config();
        let backend = LakeFSBackend::new(config).unwrap();

        let snapshot_id = "test-snapshot-123";
        let decision_id = "test-decision-456";

        assert_eq!(
            backend.snapshot_path(snapshot_id),
            "snapshots/test-snapshot-123.json"
        );
        assert_eq!(
            backend.decision_path(decision_id),
            "decisions/test-decision-456.json"
        );
    }

    #[tokio::test]
    async fn test_query_matching() {
        let config = create_test_config();
        let backend = LakeFSBackend::new(config).unwrap();
        let snapshot = create_test_snapshot().await;

        // Test function name matching
        let query = SnapshotQuery::new().with_function_name("test_function");
        assert!(backend.matches_query(&snapshot, &query));

        let query = SnapshotQuery::new().with_function_name("other_function");
        assert!(!backend.matches_query(&snapshot, &query));

        // Test tag matching
        let query = SnapshotQuery::new().with_tag("env", "test");
        assert!(backend.matches_query(&snapshot, &query));

        let query = SnapshotQuery::new().with_tag("env", "prod");
        assert!(!backend.matches_query(&snapshot, &query));

        // Test model name matching
        let query = SnapshotQuery::new().with_model_name("gpt-4");
        assert!(backend.matches_query(&snapshot, &query));

        let query = SnapshotQuery::new().with_model_name("claude-3");
        assert!(!backend.matches_query(&snapshot, &query));
    }

    #[tokio::test]
    async fn test_pending_writes() {
        let config = create_test_config();
        let backend = LakeFSBackend::new(config).unwrap();
        let snapshot = create_test_snapshot().await;

        // Save should add to pending writes
        let snapshot_id = backend.save(&snapshot).await.unwrap();
        assert_eq!(snapshot_id, snapshot.metadata.snapshot_id.to_string());

        // Check pending writes
        {
            let pending = backend.pending_writes.lock().unwrap();
            assert_eq!(pending.len(), 1);
        }
    }

    #[tokio::test]
    async fn test_accessors() {
        let config = create_test_config();
        let backend = LakeFSBackend::new(config).unwrap();

        assert_eq!(backend.repository(), "briefcase-test");
        assert_eq!(backend.branch(), "main");
        assert_eq!(backend.endpoint(), "http://localhost:8000");
    }

    // ── Edge cases (input validation) ────────────────────────────────────

    #[tokio::test]
    async fn test_create_repository_empty_name() {
        let config = create_test_config();
        let backend = LakeFSBackend::new(config).unwrap();

        let result = backend.create_repository("", "s3://bucket", None).await;
        assert!(matches!(result, Err(StorageError::InvalidQuery(_))));
    }

    #[tokio::test]
    async fn test_create_repository_empty_namespace() {
        let config = create_test_config();
        let backend = LakeFSBackend::new(config).unwrap();

        let result = backend.create_repository("my-repo", "", None).await;
        assert!(matches!(result, Err(StorageError::InvalidQuery(_))));
    }

    #[tokio::test]
    async fn test_delete_repository_empty_name() {
        let config = create_test_config();
        let backend = LakeFSBackend::new(config).unwrap();

        let result = backend.delete_repository("").await;
        assert!(matches!(result, Err(StorageError::InvalidQuery(_))));
    }

    #[tokio::test]
    async fn test_create_branch_empty_name() {
        let config = create_test_config();
        let backend = LakeFSBackend::new(config).unwrap();

        let result = backend.create_branch("", "main").await;
        assert!(matches!(result, Err(StorageError::InvalidQuery(_))));
    }

    #[tokio::test]
    async fn test_create_branch_empty_source() {
        let config = create_test_config();
        let backend = LakeFSBackend::new(config).unwrap();

        let result = backend.create_branch("feature", "").await;
        assert!(matches!(result, Err(StorageError::InvalidQuery(_))));
    }

    #[tokio::test]
    async fn test_delete_branch_empty_name() {
        let config = create_test_config();
        let backend = LakeFSBackend::new(config).unwrap();

        let result = backend.delete_branch("").await;
        assert!(matches!(result, Err(StorageError::InvalidQuery(_))));
    }

    #[tokio::test]
    async fn test_merge_branch_empty_source() {
        let config = create_test_config();
        let backend = LakeFSBackend::new(config).unwrap();

        let result = backend.merge_branch("", "main", None).await;
        assert!(matches!(result, Err(StorageError::InvalidQuery(_))));
    }

    #[tokio::test]
    async fn test_merge_branch_empty_destination() {
        let config = create_test_config();
        let backend = LakeFSBackend::new(config).unwrap();

        let result = backend.merge_branch("feature", "", None).await;
        assert!(matches!(result, Err(StorageError::InvalidQuery(_))));
    }

    #[tokio::test]
    async fn test_get_commit_empty_id() {
        let config = create_test_config();
        let backend = LakeFSBackend::new(config).unwrap();

        let result = backend.get_commit("").await;
        assert!(matches!(result, Err(StorageError::InvalidQuery(_))));
    }

    #[tokio::test]
    async fn test_diff_refs_empty_left() {
        let config = create_test_config();
        let backend = LakeFSBackend::new(config).unwrap();

        let result = backend.diff_refs("", "main").await;
        assert!(matches!(result, Err(StorageError::InvalidQuery(_))));
    }

    #[tokio::test]
    async fn test_diff_refs_empty_right() {
        let config = create_test_config();
        let backend = LakeFSBackend::new(config).unwrap();

        let result = backend.diff_refs("feature", "").await;
        assert!(matches!(result, Err(StorageError::InvalidQuery(_))));
    }

    #[tokio::test]
    async fn test_delete_object_empty_path() {
        let config = create_test_config();
        let backend = LakeFSBackend::new(config).unwrap();

        let result = backend.delete_object("").await;
        assert!(matches!(result, Err(StorageError::InvalidQuery(_))));
    }

    #[tokio::test]
    async fn test_get_object_stats_empty_path() {
        let config = create_test_config();
        let backend = LakeFSBackend::new(config).unwrap();

        let result = backend.get_object_stats("").await;
        assert!(matches!(result, Err(StorageError::InvalidQuery(_))));
    }

    #[tokio::test]
    async fn test_register_action_empty_name() {
        let config = create_test_config();
        let backend = LakeFSBackend::new(config).unwrap();

        let action = LakeFSAction {
            name: "".to_string(),
            on: vec!["post-commit".to_string()],
            hooks: vec![ActionHook {
                id: "hook1".to_string(),
                hook_type: "webhook".to_string(),
                description: "test".to_string(),
                properties: HashMap::new(),
            }],
        };

        let result = backend.register_action(&action).await;
        assert!(matches!(result, Err(StorageError::InvalidQuery(_))));
    }

    #[tokio::test]
    async fn test_register_action_no_events() {
        let config = create_test_config();
        let backend = LakeFSBackend::new(config).unwrap();

        let action = LakeFSAction {
            name: "test-action".to_string(),
            on: vec![],
            hooks: vec![ActionHook {
                id: "hook1".to_string(),
                hook_type: "webhook".to_string(),
                description: "test".to_string(),
                properties: HashMap::new(),
            }],
        };

        let result = backend.register_action(&action).await;
        assert!(matches!(result, Err(StorageError::InvalidQuery(_))));
    }

    #[tokio::test]
    async fn test_register_action_no_hooks() {
        let config = create_test_config();
        let backend = LakeFSBackend::new(config).unwrap();

        let action = LakeFSAction {
            name: "test-action".to_string(),
            on: vec!["post-commit".to_string()],
            hooks: vec![],
        };

        let result = backend.register_action(&action).await;
        assert!(matches!(result, Err(StorageError::InvalidQuery(_))));
    }

    // ── YAML builder tests ───────────────────────────────────────────────

    #[tokio::test]
    async fn test_build_action_yaml() {
        let config = create_test_config();
        let backend = LakeFSBackend::new(config).unwrap();

        let mut props = HashMap::new();
        props.insert("url".to_string(), "http://localhost:9000/hook".to_string());

        let action = LakeFSAction {
            name: "briefcase-post-commit".to_string(),
            on: vec!["post-commit".to_string()],
            hooks: vec![ActionHook {
                id: "notify".to_string(),
                hook_type: "webhook".to_string(),
                description: "Notify Briefcase server".to_string(),
                properties: props,
            }],
        };

        let yaml = backend.build_action_yaml(&action);
        assert!(yaml.contains("name: briefcase-post-commit"));
        assert!(yaml.contains("post-commit:"));
        assert!(yaml.contains("id: notify"));
        assert!(yaml.contains("type: webhook"));
        assert!(yaml.contains("url: http://localhost:9000/hook"));
    }

    #[tokio::test]
    async fn test_build_action_yaml_multiple_events() {
        let config = create_test_config();
        let backend = LakeFSBackend::new(config).unwrap();

        let action = LakeFSAction {
            name: "multi-event".to_string(),
            on: vec!["pre-commit".to_string(), "post-merge".to_string()],
            hooks: vec![ActionHook {
                id: "h1".to_string(),
                hook_type: "webhook".to_string(),
                description: "Hook".to_string(),
                properties: HashMap::new(),
            }],
        };

        let yaml = backend.build_action_yaml(&action);
        assert!(yaml.contains("pre-commit:"));
        assert!(yaml.contains("post-merge:"));
    }

    // ── Flush edge case: empty pending writes ────────────────────────────

    #[tokio::test]
    async fn test_flush_empty_pending() {
        let config = create_test_config();
        let backend = LakeFSBackend::new(config).unwrap();

        let result = backend.flush().await.unwrap();
        assert_eq!(result.snapshots_written, 0);
        assert_eq!(result.bytes_written, 0);
        assert!(result.checkpoint_id.is_none());
    }

    // ── HTTP mock tests (wiremock) ───────────────────────────────────────

    #[tokio::test]
    async fn test_create_repository_success() {
        let mock_server = MockServer::start().await;
        let config = create_test_config_with_endpoint(&mock_server.uri());
        let backend = LakeFSBackend::new(config).unwrap();

        Mock::given(method("POST"))
            .and(path("/api/v1/repositories"))
            .respond_with(ResponseTemplate::new(201).set_body_json(json!({
                "id": "my-engagement",
                "storage_namespace": "s3://my-bucket/engagement",
                "default_branch": "main",
                "creation_date": 1700000000
            })))
            .expect(1)
            .mount(&mock_server)
            .await;

        let result = backend
            .create_repository("my-engagement", "s3://my-bucket/engagement", None)
            .await
            .unwrap();

        assert_eq!(result.id, "my-engagement");
        assert_eq!(result.storage_namespace, "s3://my-bucket/engagement");
        assert_eq!(result.default_branch, "main");
    }

    #[tokio::test]
    async fn test_create_repository_conflict() {
        let mock_server = MockServer::start().await;
        let config = create_test_config_with_endpoint(&mock_server.uri());
        let backend = LakeFSBackend::new(config).unwrap();

        Mock::given(method("POST"))
            .and(path("/api/v1/repositories"))
            .respond_with(ResponseTemplate::new(409).set_body_json(json!({
                "message": "repository already exists"
            })))
            .expect(1)
            .mount(&mock_server)
            .await;

        let result = backend
            .create_repository("existing-repo", "s3://bucket", None)
            .await;

        assert!(result.is_err());
        let err = result.unwrap_err();
        assert!(
            matches!(err, StorageError::ConnectionError(msg) if msg.contains("already exists"))
        );
    }

    #[tokio::test]
    async fn test_create_repository_custom_branch() {
        let mock_server = MockServer::start().await;
        let config = create_test_config_with_endpoint(&mock_server.uri());
        let backend = LakeFSBackend::new(config).unwrap();

        Mock::given(method("POST"))
            .and(path("/api/v1/repositories"))
            .respond_with(ResponseTemplate::new(201).set_body_json(json!({
                "id": "my-repo",
                "storage_namespace": "s3://bucket",
                "default_branch": "develop",
                "creation_date": 1700000000
            })))
            .expect(1)
            .mount(&mock_server)
            .await;

        let result = backend
            .create_repository("my-repo", "s3://bucket", Some("develop"))
            .await
            .unwrap();

        assert_eq!(result.default_branch, "develop");
    }

    #[tokio::test]
    async fn test_delete_repository_success() {
        let mock_server = MockServer::start().await;
        let config = create_test_config_with_endpoint(&mock_server.uri());
        let backend = LakeFSBackend::new(config).unwrap();

        Mock::given(method("DELETE"))
            .and(path("/api/v1/repositories/test-repo"))
            .respond_with(ResponseTemplate::new(204))
            .expect(1)
            .mount(&mock_server)
            .await;

        backend.delete_repository("test-repo").await.unwrap();
    }

    #[tokio::test]
    async fn test_delete_repository_not_found() {
        let mock_server = MockServer::start().await;
        let config = create_test_config_with_endpoint(&mock_server.uri());
        let backend = LakeFSBackend::new(config).unwrap();

        Mock::given(method("DELETE"))
            .and(path("/api/v1/repositories/nonexistent"))
            .respond_with(ResponseTemplate::new(404))
            .expect(1)
            .mount(&mock_server)
            .await;

        let result = backend.delete_repository("nonexistent").await;
        assert!(matches!(result, Err(StorageError::NotFound(_))));
    }

    #[tokio::test]
    async fn test_create_branch_success() {
        let mock_server = MockServer::start().await;
        let config = create_test_config_with_endpoint(&mock_server.uri());
        let backend = LakeFSBackend::new(config).unwrap();

        Mock::given(method("POST"))
            .and(path("/api/v1/repositories/briefcase-test/branches"))
            .respond_with(ResponseTemplate::new(201).set_body_string("\"abc123def456\""))
            .expect(1)
            .mount(&mock_server)
            .await;

        let commit_id = backend.create_branch("feature-x", "main").await.unwrap();
        assert_eq!(commit_id, "abc123def456");
    }

    #[tokio::test]
    async fn test_create_branch_conflict() {
        let mock_server = MockServer::start().await;
        let config = create_test_config_with_endpoint(&mock_server.uri());
        let backend = LakeFSBackend::new(config).unwrap();

        Mock::given(method("POST"))
            .and(path("/api/v1/repositories/briefcase-test/branches"))
            .respond_with(ResponseTemplate::new(409))
            .expect(1)
            .mount(&mock_server)
            .await;

        let result = backend.create_branch("existing-branch", "main").await;
        assert!(result.is_err());
    }

    #[tokio::test]
    async fn test_delete_branch_success() {
        let mock_server = MockServer::start().await;
        let config = create_test_config_with_endpoint(&mock_server.uri());
        let backend = LakeFSBackend::new(config).unwrap();

        Mock::given(method("DELETE"))
            .and(path(
                "/api/v1/repositories/briefcase-test/branches/feature-x",
            ))
            .respond_with(ResponseTemplate::new(204))
            .expect(1)
            .mount(&mock_server)
            .await;

        backend.delete_branch("feature-x").await.unwrap();
    }

    #[tokio::test]
    async fn test_delete_branch_not_found() {
        let mock_server = MockServer::start().await;
        let config = create_test_config_with_endpoint(&mock_server.uri());
        let backend = LakeFSBackend::new(config).unwrap();

        Mock::given(method("DELETE"))
            .and(path(
                "/api/v1/repositories/briefcase-test/branches/nonexistent",
            ))
            .respond_with(ResponseTemplate::new(404))
            .expect(1)
            .mount(&mock_server)
            .await;

        let result = backend.delete_branch("nonexistent").await;
        assert!(matches!(result, Err(StorageError::NotFound(_))));
    }

    #[tokio::test]
    async fn test_merge_branch_success() {
        let mock_server = MockServer::start().await;
        let config = create_test_config_with_endpoint(&mock_server.uri());
        let backend = LakeFSBackend::new(config).unwrap();

        Mock::given(method("POST"))
            .and(path(
                "/api/v1/repositories/briefcase-test/refs/feature-x/merge/main",
            ))
            .respond_with(ResponseTemplate::new(200).set_body_json(json!({
                "reference": "merge-commit-abc123",
                "summary": {"added": 3, "removed": 0, "changed": 1}
            })))
            .expect(1)
            .mount(&mock_server)
            .await;

        let result = backend
            .merge_branch("feature-x", "main", Some("Merge feature-x"))
            .await
            .unwrap();

        assert_eq!(result.commit_id, "merge-commit-abc123");
        assert!(result.summary.contains("added"));
    }

    #[tokio::test]
    async fn test_merge_branch_conflict() {
        let mock_server = MockServer::start().await;
        let config = create_test_config_with_endpoint(&mock_server.uri());
        let backend = LakeFSBackend::new(config).unwrap();

        Mock::given(method("POST"))
            .and(path(
                "/api/v1/repositories/briefcase-test/refs/feature-x/merge/main",
            ))
            .respond_with(ResponseTemplate::new(409).set_body_json(json!({
                "message": "conflict"
            })))
            .expect(1)
            .mount(&mock_server)
            .await;

        let result = backend.merge_branch("feature-x", "main", None).await;
        assert!(result.is_err());
        let err = result.unwrap_err();
        assert!(matches!(err, StorageError::ConnectionError(msg) if msg.contains("conflict")));
    }

    #[tokio::test]
    async fn test_merge_branch_default_message() {
        let mock_server = MockServer::start().await;
        let config = create_test_config_with_endpoint(&mock_server.uri());
        let backend = LakeFSBackend::new(config).unwrap();

        Mock::given(method("POST"))
            .and(path(
                "/api/v1/repositories/briefcase-test/refs/src/merge/dst",
            ))
            .respond_with(ResponseTemplate::new(200).set_body_json(json!({
                "reference": "commit-xyz",
                "summary": {}
            })))
            .expect(1)
            .mount(&mock_server)
            .await;

        let result = backend.merge_branch("src", "dst", None).await.unwrap();
        assert_eq!(result.commit_id, "commit-xyz");
    }

    #[tokio::test]
    async fn test_get_commit_success() {
        let mock_server = MockServer::start().await;
        let config = create_test_config_with_endpoint(&mock_server.uri());
        let backend = LakeFSBackend::new(config).unwrap();

        Mock::given(method("GET"))
            .and(path("/api/v1/repositories/briefcase-test/commits/abc123"))
            .respond_with(ResponseTemplate::new(200).set_body_json(json!({
                "id": "abc123",
                "message": "Initial commit",
                "committer": "admin",
                "metadata": {"source": "briefcase-ai"},
                "creation_date": 1700000000
            })))
            .expect(1)
            .mount(&mock_server)
            .await;

        let commit = backend.get_commit("abc123").await.unwrap();
        assert_eq!(commit.id, "abc123");
        assert_eq!(commit.message, "Initial commit");
        assert_eq!(commit.committer, "admin");
        assert_eq!(
            commit.metadata.get("source"),
            Some(&"briefcase-ai".to_string())
        );
    }

    #[tokio::test]
    async fn test_get_commit_not_found() {
        let mock_server = MockServer::start().await;
        let config = create_test_config_with_endpoint(&mock_server.uri());
        let backend = LakeFSBackend::new(config).unwrap();

        Mock::given(method("GET"))
            .and(path(
                "/api/v1/repositories/briefcase-test/commits/nonexistent",
            ))
            .respond_with(ResponseTemplate::new(404))
            .expect(1)
            .mount(&mock_server)
            .await;

        let result = backend.get_commit("nonexistent").await;
        assert!(matches!(result, Err(StorageError::NotFound(_))));
    }

    #[tokio::test]
    async fn test_diff_refs_success() {
        let mock_server = MockServer::start().await;
        let config = create_test_config_with_endpoint(&mock_server.uri());
        let backend = LakeFSBackend::new(config).unwrap();

        Mock::given(method("GET"))
            .and(path(
                "/api/v1/repositories/briefcase-test/refs/main/diff/feature-x",
            ))
            .respond_with(ResponseTemplate::new(200).set_body_json(json!({
                "results": [
                    {"path": "snapshots/a.json", "type": "added", "size_bytes": 1024},
                    {"path": "snapshots/b.json", "type": "changed", "size_bytes": 2048},
                    {"path": "decisions/old.json", "type": "removed"}
                ]
            })))
            .expect(1)
            .mount(&mock_server)
            .await;

        let diffs = backend.diff_refs("main", "feature-x").await.unwrap();
        assert_eq!(diffs.len(), 3);
        assert_eq!(diffs[0].path, "snapshots/a.json");
        assert_eq!(diffs[0].diff_type, "added");
        assert_eq!(diffs[0].size_bytes, Some(1024));
        assert_eq!(diffs[1].diff_type, "changed");
        assert_eq!(diffs[2].diff_type, "removed");
        assert_eq!(diffs[2].size_bytes, None);
    }

    #[tokio::test]
    async fn test_diff_refs_empty_result() {
        let mock_server = MockServer::start().await;
        let config = create_test_config_with_endpoint(&mock_server.uri());
        let backend = LakeFSBackend::new(config).unwrap();

        Mock::given(method("GET"))
            .and(path(
                "/api/v1/repositories/briefcase-test/refs/main/diff/main",
            ))
            .respond_with(ResponseTemplate::new(200).set_body_json(json!({"results": []})))
            .expect(1)
            .mount(&mock_server)
            .await;

        let diffs = backend.diff_refs("main", "main").await.unwrap();
        assert!(diffs.is_empty());
    }

    #[tokio::test]
    async fn test_get_object_stats_success() {
        let mock_server = MockServer::start().await;
        let config = create_test_config_with_endpoint(&mock_server.uri());
        let backend = LakeFSBackend::new(config).unwrap();

        Mock::given(method("GET"))
            .and(path(
                "/api/v1/repositories/briefcase-test/refs/main/objects/stat",
            ))
            .and(query_param("path", "snapshots/test.json"))
            .respond_with(ResponseTemplate::new(200).set_body_json(json!({
                "path": "snapshots/test.json",
                "physical_address": "s3://bucket/data/abc",
                "size_bytes": 4096,
                "content_type": "application/json",
                "mtime": 1700000000,
                "checksum": "sha256:abcdef"
            })))
            .expect(1)
            .mount(&mock_server)
            .await;

        let stats = backend
            .get_object_stats("snapshots/test.json")
            .await
            .unwrap();

        assert_eq!(stats.path, "snapshots/test.json");
        assert_eq!(stats.size_bytes, 4096);
        assert_eq!(stats.content_type, "application/json");
        assert_eq!(stats.checksum, "sha256:abcdef");
    }

    #[tokio::test]
    async fn test_get_object_stats_not_found() {
        let mock_server = MockServer::start().await;
        let config = create_test_config_with_endpoint(&mock_server.uri());
        let backend = LakeFSBackend::new(config).unwrap();

        Mock::given(method("GET"))
            .and(path(
                "/api/v1/repositories/briefcase-test/refs/main/objects/stat",
            ))
            .respond_with(ResponseTemplate::new(404))
            .expect(1)
            .mount(&mock_server)
            .await;

        let result = backend.get_object_stats("nonexistent.json").await;
        assert!(matches!(result, Err(StorageError::NotFound(_))));
    }

    #[tokio::test]
    async fn test_delete_object_success() {
        let mock_server = MockServer::start().await;
        let config = create_test_config_with_endpoint(&mock_server.uri());
        let backend = LakeFSBackend::new(config).unwrap();

        Mock::given(method("DELETE"))
            .and(path(
                "/api/v1/repositories/briefcase-test/branches/main/objects",
            ))
            .and(query_param("path", "snapshots/old.json"))
            .respond_with(ResponseTemplate::new(204))
            .expect(1)
            .mount(&mock_server)
            .await;

        backend.delete_object("snapshots/old.json").await.unwrap();
    }

    #[tokio::test]
    async fn test_delete_object_not_found() {
        let mock_server = MockServer::start().await;
        let config = create_test_config_with_endpoint(&mock_server.uri());
        let backend = LakeFSBackend::new(config).unwrap();

        Mock::given(method("DELETE"))
            .and(path(
                "/api/v1/repositories/briefcase-test/branches/main/objects",
            ))
            .respond_with(ResponseTemplate::new(404))
            .expect(1)
            .mount(&mock_server)
            .await;

        let result = backend.delete_object("nonexistent.json").await;
        assert!(matches!(result, Err(StorageError::NotFound(_))));
    }

    #[tokio::test]
    async fn test_upload_object_success() {
        let mock_server = MockServer::start().await;
        let config = create_test_config_with_endpoint(&mock_server.uri());
        let backend = LakeFSBackend::new(config).unwrap();

        Mock::given(method("PUT"))
            .and(path(
                "/api/v1/repositories/briefcase-test/branches/main/objects",
            ))
            .and(query_param("path", "test/file.json"))
            .respond_with(ResponseTemplate::new(201))
            .expect(1)
            .mount(&mock_server)
            .await;

        backend
            .upload_object("test/file.json", b"hello world")
            .await
            .unwrap();
    }

    #[tokio::test]
    async fn test_upload_object_server_error() {
        let mock_server = MockServer::start().await;
        let config = create_test_config_with_endpoint(&mock_server.uri());
        let backend = LakeFSBackend::new(config).unwrap();

        Mock::given(method("PUT"))
            .and(path(
                "/api/v1/repositories/briefcase-test/branches/main/objects",
            ))
            .respond_with(ResponseTemplate::new(500).set_body_string("Internal Server Error"))
            .expect(1)
            .mount(&mock_server)
            .await;

        let result = backend.upload_object("test/file.json", b"data").await;
        assert!(result.is_err());
    }

    #[tokio::test]
    async fn test_download_object_success() {
        let mock_server = MockServer::start().await;
        let config = create_test_config_with_endpoint(&mock_server.uri());
        let backend = LakeFSBackend::new(config).unwrap();

        Mock::given(method("GET"))
            .and(path(
                "/api/v1/repositories/briefcase-test/refs/main/objects",
            ))
            .and(query_param("path", "test/file.json"))
            .respond_with(ResponseTemplate::new(200).set_body_bytes(b"file content here".to_vec()))
            .expect(1)
            .mount(&mock_server)
            .await;

        let data = backend.download_object("test/file.json").await.unwrap();
        assert_eq!(data, b"file content here");
    }

    #[tokio::test]
    async fn test_download_object_not_found() {
        let mock_server = MockServer::start().await;
        let config = create_test_config_with_endpoint(&mock_server.uri());
        let backend = LakeFSBackend::new(config).unwrap();

        Mock::given(method("GET"))
            .and(path(
                "/api/v1/repositories/briefcase-test/refs/main/objects",
            ))
            .respond_with(ResponseTemplate::new(404))
            .expect(1)
            .mount(&mock_server)
            .await;

        let result = backend.download_object("missing.json").await;
        assert!(matches!(result, Err(StorageError::NotFound(_))));
    }

    #[tokio::test]
    async fn test_create_commit_success() {
        let mock_server = MockServer::start().await;
        let config = create_test_config_with_endpoint(&mock_server.uri());
        let backend = LakeFSBackend::new(config).unwrap();

        Mock::given(method("POST"))
            .and(path(
                "/api/v1/repositories/briefcase-test/branches/main/commits",
            ))
            .respond_with(ResponseTemplate::new(201).set_body_json(json!({
                "id": "commit-sha-xyz"
            })))
            .expect(1)
            .mount(&mock_server)
            .await;

        let id = backend.create_commit("Test commit").await.unwrap();
        assert_eq!(id, "commit-sha-xyz");
    }

    #[tokio::test]
    async fn test_list_objects_success() {
        let mock_server = MockServer::start().await;
        let config = create_test_config_with_endpoint(&mock_server.uri());
        let backend = LakeFSBackend::new(config).unwrap();

        Mock::given(method("GET"))
            .and(path(
                "/api/v1/repositories/briefcase-test/refs/main/objects/ls",
            ))
            .and(query_param("prefix", "snapshots/"))
            .respond_with(ResponseTemplate::new(200).set_body_json(json!({
                "results": [
                    {"path": "snapshots/a.json", "type": "object"},
                    {"path": "snapshots/b.json", "type": "object"},
                    {"path": "snapshots/", "type": "common_prefix"}
                ]
            })))
            .expect(1)
            .mount(&mock_server)
            .await;

        let paths = backend.list_objects("snapshots/").await.unwrap();
        assert_eq!(paths.len(), 2);
        assert!(paths.contains(&"snapshots/a.json".to_string()));
        assert!(paths.contains(&"snapshots/b.json".to_string()));
    }

    #[tokio::test]
    async fn test_health_check_healthy() {
        let mock_server = MockServer::start().await;
        let config = create_test_config_with_endpoint(&mock_server.uri());
        let backend = LakeFSBackend::new(config).unwrap();

        Mock::given(method("GET"))
            .and(path("/api/v1/repositories/briefcase-test"))
            .respond_with(ResponseTemplate::new(200))
            .expect(1)
            .mount(&mock_server)
            .await;

        assert!(backend.health_check().await.unwrap());
    }

    #[tokio::test]
    async fn test_health_check_unhealthy() {
        let mock_server = MockServer::start().await;
        let config = create_test_config_with_endpoint(&mock_server.uri());
        let backend = LakeFSBackend::new(config).unwrap();

        Mock::given(method("GET"))
            .and(path("/api/v1/repositories/briefcase-test"))
            .respond_with(ResponseTemplate::new(503))
            .expect(1)
            .mount(&mock_server)
            .await;

        assert!(!backend.health_check().await.unwrap());
    }

    #[tokio::test]
    async fn test_delete_via_trait_success() {
        let mock_server = MockServer::start().await;
        let config = create_test_config_with_endpoint(&mock_server.uri());
        let backend = LakeFSBackend::new(config).unwrap();

        Mock::given(method("DELETE"))
            .and(path(
                "/api/v1/repositories/briefcase-test/branches/main/objects",
            ))
            .respond_with(ResponseTemplate::new(204))
            .expect(1)
            .mount(&mock_server)
            .await;

        let deleted = backend.delete("some-snapshot-id").await.unwrap();
        assert!(deleted);
    }

    #[tokio::test]
    async fn test_delete_via_trait_not_found() {
        let mock_server = MockServer::start().await;
        let config = create_test_config_with_endpoint(&mock_server.uri());
        let backend = LakeFSBackend::new(config).unwrap();

        Mock::given(method("DELETE"))
            .and(path(
                "/api/v1/repositories/briefcase-test/branches/main/objects",
            ))
            .respond_with(ResponseTemplate::new(404))
            .expect(1)
            .mount(&mock_server)
            .await;

        let deleted = backend.delete("nonexistent").await.unwrap();
        assert!(!deleted);
    }

    // Note: Integration tests against real LakeFS would require a running instance
    // These would be better suited for a separate integration test suite

    // ─── Auth provider tests ────────────────────────────────────────────────

    #[test]
    fn test_lakefs_auth_enum_basic() {
        let auth = LakeFSAuth::Basic {
            access_key: "ak".into(),
            secret_key: "sk".into(),
        };
        match auth {
            LakeFSAuth::Basic {
                access_key,
                secret_key,
            } => {
                assert_eq!(access_key, "ak");
                assert_eq!(secret_key, "sk");
            }
            _ => panic!("Expected Basic"),
        }
    }

    #[test]
    fn test_lakefs_auth_enum_all_variants() {
        // Ensure all 6 variants construct without panic.
        let _basic = LakeFSAuth::Basic {
            access_key: "a".into(),
            secret_key: "s".into(),
        };
        let _token = LakeFSAuth::Token {
            access_key: "a".into(),
            secret_key: "s".into(),
        };
        let _sts = LakeFSAuth::Sts {
            oidc_token: "tok".into(),
        };
        let _iam = LakeFSAuth::Iam;
        let _oidc = LakeFSAuth::Oidc {
            token: "tok".into(),
        };
        let _saml = LakeFSAuth::Saml {
            assertion: "xml".into(),
        };
    }

    #[test]
    fn test_lakefs_config_with_auth() {
        let config = LakeFSConfig::new("http://localhost", "repo", "main", "ak", "sk").with_auth(
            LakeFSAuth::Token {
                access_key: "ak2".into(),
                secret_key: "sk2".into(),
            },
        );
        assert!(config.auth.is_some());
        match config.auth.unwrap() {
            LakeFSAuth::Token {
                access_key,
                secret_key,
            } => {
                assert_eq!(access_key, "ak2");
                assert_eq!(secret_key, "sk2");
            }
            _ => panic!("Expected Token"),
        }
    }

    #[test]
    fn test_lakefs_config_without_auth_defaults_to_none() {
        let config = LakeFSConfig::new("http://localhost", "repo", "main", "ak", "sk");
        assert!(config.auth.is_none());
    }

    #[tokio::test]
    async fn test_basic_auth_provider_header() {
        let provider = BasicAuthProvider::new(
            "AKIAIOSFODNN7EXAMPLE",
            "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
        )
        .unwrap();
        let header = provider.auth_header().await.unwrap();
        let header_str = header.to_str().unwrap();
        assert!(header_str.starts_with("Basic "));
        // Verify it's valid base64
        let b64_part = &header_str["Basic ".len()..];
        let decoded = general_purpose::STANDARD.decode(b64_part).unwrap();
        let decoded_str = String::from_utf8(decoded).unwrap();
        assert_eq!(
            decoded_str,
            "AKIAIOSFODNN7EXAMPLE:wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"
        );
    }

    #[tokio::test]
    async fn test_basic_auth_provider_mode_name() {
        let provider = BasicAuthProvider::new("ak", "sk").unwrap();
        assert_eq!(provider.mode_name(), "basic");
    }

    #[tokio::test]
    async fn test_token_auth_provider_login_success() {
        let mock_server = MockServer::start().await;

        // Mock the login endpoint
        Mock::given(method("POST"))
            .and(path("/api/v1/auth/login"))
            .respond_with(ResponseTemplate::new(200).set_body_json(json!({
                "token": "eyJhbGciOiJIUzI1NiJ9.test",
                "token_expiration": 9999999999u64
            })))
            .expect(1)
            .mount(&mock_server)
            .await;

        let provider = TokenAuthProvider::new(&mock_server.uri(), "ak", "sk");
        let header = provider.auth_header().await.unwrap();
        let header_str = header.to_str().unwrap();
        assert_eq!(header_str, "Bearer eyJhbGciOiJIUzI1NiJ9.test");
        assert_eq!(provider.mode_name(), "token");
    }

    #[tokio::test]
    async fn test_token_auth_provider_caches_token() {
        let mock_server = MockServer::start().await;

        Mock::given(method("POST"))
            .and(path("/api/v1/auth/login"))
            .respond_with(ResponseTemplate::new(200).set_body_json(json!({
                "token": "cached-token",
                "token_expiration": 9999999999u64
            })))
            .expect(1)  // Should only be called once
            .mount(&mock_server)
            .await;

        let provider = TokenAuthProvider::new(&mock_server.uri(), "ak", "sk");

        // First call — fetches token
        let h1 = provider.auth_header().await.unwrap();
        // Second call — cache hit
        let h2 = provider.auth_header().await.unwrap();
        assert_eq!(h1, h2);
    }

    #[tokio::test]
    async fn test_token_auth_provider_login_failure() {
        let mock_server = MockServer::start().await;

        Mock::given(method("POST"))
            .and(path("/api/v1/auth/login"))
            .respond_with(ResponseTemplate::new(401).set_body_json(json!({
                "message": "invalid credentials"
            })))
            .expect(1)
            .mount(&mock_server)
            .await;

        let provider = TokenAuthProvider::new(&mock_server.uri(), "bad", "creds");
        let result = provider.auth_header().await;
        assert!(result.is_err());
        let err_msg = format!("{}", result.unwrap_err());
        assert!(err_msg.contains("401"));
    }

    #[tokio::test]
    async fn test_sts_auth_provider_success() {
        let mock_server = MockServer::start().await;

        Mock::given(method("POST"))
            .and(path("/sts/login"))
            .respond_with(ResponseTemplate::new(200).set_body_json(json!({
                "Credentials": {
                    "AccessKeyId": "ASIATEMP",
                    "SecretAccessKey": "tempSecret",
                    "SessionToken": "FwoGZXIvY...",
                    "Expiration": 9999999999u64
                }
            })))
            .expect(1)
            .mount(&mock_server)
            .await;

        let provider = StsAuthProvider::new(&mock_server.uri(), "my-oidc-token");
        let header = provider.auth_header().await.unwrap();
        let header_str = header.to_str().unwrap();
        assert!(header_str.starts_with("Basic "));
        // Decode and verify it uses the temp credentials
        let b64_part = &header_str["Basic ".len()..];
        let decoded = general_purpose::STANDARD.decode(b64_part).unwrap();
        let decoded_str = String::from_utf8(decoded).unwrap();
        assert_eq!(decoded_str, "ASIATEMP:tempSecret");
        assert_eq!(provider.mode_name(), "sts");
    }

    #[tokio::test]
    async fn test_oidc_auth_provider_success() {
        let mock_server = MockServer::start().await;

        Mock::given(method("POST"))
            .and(path("/api/v1/oidc/login"))
            .respond_with(ResponseTemplate::new(200).set_body_json(json!({
                "token": "oidc-lakefs-jwt-token",
                "token_expiration": 9999999999u64
            })))
            .expect(1)
            .mount(&mock_server)
            .await;

        let provider = OidcAuthProvider::new(&mock_server.uri(), "external-oidc-token");
        let header = provider.auth_header().await.unwrap();
        let header_str = header.to_str().unwrap();
        assert_eq!(header_str, "Bearer oidc-lakefs-jwt-token");
        assert_eq!(provider.mode_name(), "oidc");
    }

    #[tokio::test]
    async fn test_saml_auth_provider_success() {
        let mock_server = MockServer::start().await;

        Mock::given(method("POST"))
            .and(path("/api/v1/auth/external/saml"))
            .respond_with(ResponseTemplate::new(200).set_body_json(json!({
                "token": "saml-lakefs-jwt-token",
                "token_expiration": 9999999999u64
            })))
            .expect(1)
            .mount(&mock_server)
            .await;

        let provider = SamlAuthProvider::new(&mock_server.uri(), "PHNhbWxSZXNwb25zZT4=");
        let header = provider.auth_header().await.unwrap();
        let header_str = header.to_str().unwrap();
        assert_eq!(header_str, "Bearer saml-lakefs-jwt-token");
        assert_eq!(provider.mode_name(), "saml");
    }

    #[tokio::test]
    async fn test_iam_auth_provider_mode_name() {
        let provider = IamAuthProvider::new();
        assert_eq!(provider.mode_name(), "iam");
    }

    #[tokio::test]
    async fn test_build_auth_provider_basic() {
        let auth = LakeFSAuth::Basic {
            access_key: "ak".into(),
            secret_key: "sk".into(),
        };
        let provider = build_auth_provider(&auth, "http://localhost").unwrap();
        assert_eq!(provider.mode_name(), "basic");
        let header = provider.auth_header().await.unwrap();
        assert!(header.to_str().unwrap().starts_with("Basic "));
    }

    #[tokio::test]
    async fn test_build_auth_provider_token() {
        let auth = LakeFSAuth::Token {
            access_key: "ak".into(),
            secret_key: "sk".into(),
        };
        let provider = build_auth_provider(&auth, "http://localhost").unwrap();
        assert_eq!(provider.mode_name(), "token");
    }

    #[tokio::test]
    async fn test_build_auth_provider_iam() {
        let auth = LakeFSAuth::Iam;
        let provider = build_auth_provider(&auth, "http://localhost").unwrap();
        assert_eq!(provider.mode_name(), "iam");
    }

    #[tokio::test]
    async fn test_backend_with_basic_auth_config() {
        let mock_server = MockServer::start().await;

        let config = LakeFSConfig::new(
            mock_server.uri(),
            "test-repo",
            "main",
            "test_key",
            "test_secret",
        )
        .with_auth(LakeFSAuth::Basic {
            access_key: "test_key".into(),
            secret_key: "test_secret".into(),
        });

        let backend = LakeFSBackend::new(config).unwrap();
        assert_eq!(backend.auth_mode(), "basic");
    }

    #[tokio::test]
    async fn test_backend_with_token_auth_sends_bearer() {
        let mock_server = MockServer::start().await;

        // Mock login endpoint
        Mock::given(method("POST"))
            .and(path("/api/v1/auth/login"))
            .respond_with(ResponseTemplate::new(200).set_body_json(json!({
                "token": "test-jwt-token",
                "token_expiration": 9999999999u64
            })))
            .mount(&mock_server)
            .await;

        // Mock repository endpoint (health check target)
        Mock::given(method("GET"))
            .and(path("/api/v1/repositories/test-repo"))
            .respond_with(ResponseTemplate::new(200).set_body_json(json!({
                "id": "test-repo",
                "storage_namespace": "s3://test",
                "default_branch": "main"
            })))
            .mount(&mock_server)
            .await;

        let config = LakeFSConfig::new(mock_server.uri(), "test-repo", "main", "ak", "sk")
            .with_auth(LakeFSAuth::Token {
                access_key: "ak".into(),
                secret_key: "sk".into(),
            });

        let backend = LakeFSBackend::new(config).unwrap();
        assert_eq!(backend.auth_mode(), "token");

        // Verify it can make an authenticated request
        let health = backend.health_check().await.unwrap();
        assert!(health);
    }

    #[tokio::test]
    async fn test_backend_default_auth_is_basic() {
        let config = LakeFSConfig::new("http://localhost:8000", "test-repo", "main", "ak", "sk");
        // No .with_auth() → should default to Basic
        let backend = LakeFSBackend::new(config).unwrap();
        assert_eq!(backend.auth_mode(), "basic");
    }
}