lcpfs 2026.1.102

// Copyright 2025 LunaOS Contributors
// SPDX-License-Identifier: Apache-2.0
//
// AWS S3 Cloud Storage Integration
// Real S3 API implementation with SigV4 signing.

//! # AWS S3 Cloud Storage Integration
//!
//! This module provides real AWS S3 integration for LCPFS cloud tiering,
//! implementing AWS Signature Version 4 request signing and S3 operations.
//!
//! ## Features
//!
//! - AWS SigV4 request signing (no external dependencies)
//! - S3 operations: PUT, GET, DELETE, HEAD, COPY
//! - Storage class transitions (Standard, IA, Glacier, Deep Archive)
//! - S3-compatible endpoint support (MinIO, Ceph, etc.)
//! - Abstracted network I/O for no_std compatibility
//!
//! ## Usage
//!
//! ```rust,ignore
//! use lcpfs::lcpfs_cloud_s3::{S3Client, S3Config};
//!
//! let config = S3Config {
//!     endpoint: "https://s3.us-east-1.amazonaws.com".into(),
//!     bucket: "my-bucket".into(),
//!     access_key: "AKIAIOSFODNN7EXAMPLE".into(),
//!     secret_key: "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY".into(),
//!     region: "us-east-1".into(),
//! };
//!
//! let client = S3Client::new(config);
//! client.put_object("path/to/object", data, "STANDARD")?;
//! ```

use alloc::format;
use alloc::string::{String, ToString};
use alloc::vec;
use alloc::vec::Vec;
use core::fmt;
use core::str::FromStr;

// Encoding crates
use base16ct::lower::encode_string as hex_encode_crate;
use percent_encoding::{AsciiSet, CONTROLS, utf8_percent_encode};

use crate::cloud::tier::CloudStorageClass;

/// Characters that need percent-encoding in URI path components (RFC 3986).
/// We preserve: A-Z, a-z, 0-9, -, _, ., ~, /
const URI_PATH_SET: &AsciiSet = &CONTROLS
    .add(b' ')
    .add(b'!')
    .add(b'"')
    .add(b'#')
    .add(b'$')
    .add(b'%')
    .add(b'&')
    .add(b'\'')
    .add(b'(')
    .add(b')')
    .add(b'*')
    .add(b'+')
    .add(b',')
    .add(b':')
    .add(b';')
    .add(b'<')
    .add(b'=')
    .add(b'>')
    .add(b'?')
    .add(b'@')
    .add(b'[')
    .add(b'\\')
    .add(b']')
    .add(b'^')
    .add(b'`')
    .add(b'{')
    .add(b'|')
    .add(b'}');

// ═══════════════════════════════════════════════════════════════════════════════
// S3 ERROR TYPES
// ═══════════════════════════════════════════════════════════════════════════════

/// S3 operation error.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum S3Error {
    /// Invalid configuration.
    InvalidConfig(&'static str),
    /// Authentication failed.
    AuthenticationFailed,
    /// Access denied.
    AccessDenied,
    /// Bucket not found.
    BucketNotFound,
    /// Object not found.
    ObjectNotFound,
    /// Invalid object key.
    InvalidKey,
    /// Request too large.
    RequestTooLarge,
    /// Network error.
    NetworkError(&'static str),
    /// TLS error.
    TlsError(&'static str),
    /// HTTP error with status code.
    HttpError(u16),
    /// XML parsing error.
    XmlParseError,
    /// Timeout.
    Timeout,
    /// Internal error.
    Internal(&'static str),
}

impl S3Error {
    /// Get error description.
    pub fn description(&self) -> &'static str {
        match self {
            S3Error::InvalidConfig(msg) => msg,
            S3Error::AuthenticationFailed => "Authentication failed",
            S3Error::AccessDenied => "Access denied",
            S3Error::BucketNotFound => "Bucket not found",
            S3Error::ObjectNotFound => "Object not found",
            S3Error::InvalidKey => "Invalid object key",
            S3Error::RequestTooLarge => "Request too large",
            S3Error::NetworkError(msg) => msg,
            S3Error::TlsError(msg) => msg,
            S3Error::HttpError(_) => "HTTP error",
            S3Error::XmlParseError => "XML parsing error",
            S3Error::Timeout => "Request timeout",
            S3Error::Internal(msg) => msg,
        }
    }

    /// Create from HTTP status code.
    pub fn from_status(status: u16) -> Option<Self> {
        match status {
            200..=299 => None,
            401 => Some(S3Error::AuthenticationFailed),
            403 => Some(S3Error::AccessDenied),
            404 => Some(S3Error::ObjectNotFound),
            413 => Some(S3Error::RequestTooLarge),
            _ => Some(S3Error::HttpError(status)),
        }
    }
}

impl fmt::Display for S3Error {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            S3Error::HttpError(code) => write!(f, "HTTP error: {}", code),
            _ => write!(f, "{}", self.description()),
        }
    }
}

// ═══════════════════════════════════════════════════════════════════════════════
// S3 STORAGE CLASSES
// ═══════════════════════════════════════════════════════════════════════════════

/// AWS S3 storage class.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum S3StorageClass {
    /// Standard storage (frequent access).
    Standard,
    /// Reduced redundancy (deprecated, use Standard-IA).
    ReducedRedundancy,
    /// Infrequent access.
    StandardIa,
    /// One Zone infrequent access.
    OnezoneIa,
    /// Intelligent tiering.
    IntelligentTiering,
    /// Glacier Instant Retrieval.
    GlacierIr,
    /// Glacier Flexible Retrieval.
    Glacier,
    /// Glacier Deep Archive.
    DeepArchive,
}

impl S3StorageClass {
    /// Get the S3 API storage class string.
    pub fn as_str(&self) -> &'static str {
        match self {
            S3StorageClass::Standard => "STANDARD",
            S3StorageClass::ReducedRedundancy => "REDUCED_REDUNDANCY",
            S3StorageClass::StandardIa => "STANDARD_IA",
            S3StorageClass::OnezoneIa => "ONEZONE_IA",
            S3StorageClass::IntelligentTiering => "INTELLIGENT_TIERING",
            S3StorageClass::GlacierIr => "GLACIER_IR",
            S3StorageClass::Glacier => "GLACIER",
            S3StorageClass::DeepArchive => "DEEP_ARCHIVE",
        }
    }

    /// Convert from CloudStorageClass.
    pub fn from_cloud_class(class: CloudStorageClass) -> Self {
        match class {
            CloudStorageClass::Hot => S3StorageClass::Standard,
            CloudStorageClass::Warm => S3StorageClass::StandardIa,
            CloudStorageClass::Cold => S3StorageClass::Glacier,
            CloudStorageClass::Archive => S3StorageClass::DeepArchive,
        }
    }

    /// Convert to CloudStorageClass.
    pub fn to_cloud_class(self) -> CloudStorageClass {
        match self {
            S3StorageClass::Standard | S3StorageClass::ReducedRedundancy => CloudStorageClass::Hot,
            S3StorageClass::StandardIa
            | S3StorageClass::OnezoneIa
            | S3StorageClass::IntelligentTiering => CloudStorageClass::Warm,
            S3StorageClass::GlacierIr | S3StorageClass::Glacier => CloudStorageClass::Cold,
            S3StorageClass::DeepArchive => CloudStorageClass::Archive,
        }
    }
}

/// Parse error for S3StorageClass.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ParseStorageClassError;

impl fmt::Display for ParseStorageClassError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "invalid S3 storage class")
    }
}

impl FromStr for S3StorageClass {
    type Err = ParseStorageClassError;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        match s {
            "STANDARD" => Ok(S3StorageClass::Standard),
            "REDUCED_REDUNDANCY" => Ok(S3StorageClass::ReducedRedundancy),
            "STANDARD_IA" => Ok(S3StorageClass::StandardIa),
            "ONEZONE_IA" => Ok(S3StorageClass::OnezoneIa),
            "INTELLIGENT_TIERING" => Ok(S3StorageClass::IntelligentTiering),
            "GLACIER_IR" => Ok(S3StorageClass::GlacierIr),
            "GLACIER" => Ok(S3StorageClass::Glacier),
            "DEEP_ARCHIVE" => Ok(S3StorageClass::DeepArchive),
            _ => Err(ParseStorageClassError),
        }
    }
}

// ═══════════════════════════════════════════════════════════════════════════════
// S3 OBJECT METADATA
// ═══════════════════════════════════════════════════════════════════════════════

/// S3 object metadata from HEAD response.
#[derive(Debug, Clone)]
pub struct ObjectMetadata {
    /// Content length in bytes.
    pub content_length: u64,
    /// Content type (MIME type).
    pub content_type: String,
    /// ETag (entity tag / MD5).
    pub etag: String,
    /// Last modified timestamp (Unix epoch seconds).
    pub last_modified: u64,
    /// Storage class.
    pub storage_class: S3StorageClass,
    /// Version ID (if versioning enabled).
    pub version_id: Option<String>,
}

impl Default for ObjectMetadata {
    fn default() -> Self {
        Self {
            content_length: 0,
            content_type: String::new(),
            etag: String::new(),
            last_modified: 0,
            storage_class: S3StorageClass::Standard,
            version_id: None,
        }
    }
}

// ═══════════════════════════════════════════════════════════════════════════════
// S3 CONFIGURATION
// ═══════════════════════════════════════════════════════════════════════════════

/// Timestamp provider for AWS SigV4 signing.
///
/// In kernel environments, time must come from hardware (RTC, TSC, etc.).
/// This trait allows injecting a timestamp source.
pub trait TimestampProvider: Send + Sync {
    /// Get current UTC time as Unix timestamp (seconds since epoch).
    fn unix_timestamp(&self) -> u64;
}

/// Default timestamp provider using a configurable value.
///
/// Should be replaced with actual hardware time source in production.
#[derive(Debug, Clone, Default)]
pub struct FixedTimestamp(pub u64);

impl TimestampProvider for FixedTimestamp {
    fn unix_timestamp(&self) -> u64 {
        self.0
    }
}

/// S3 client configuration.
#[derive(Debug, Clone)]
pub struct S3Config {
    /// S3 endpoint URL (e.g., "https://s3.us-east-1.amazonaws.com").
    pub endpoint: String,
    /// Bucket name.
    pub bucket: String,
    /// AWS access key ID.
    pub access_key: String,
    /// AWS secret access key.
    pub secret_key: String,
    /// AWS region (e.g., "us-east-1").
    pub region: String,
    /// Use path-style URLs (for MinIO/older S3).
    pub path_style: bool,
    /// Connection timeout in milliseconds.
    pub timeout_ms: u32,
    /// Unix timestamp for request signing (seconds since epoch).
    /// In production, this should come from RTC or trusted time source.
    pub timestamp: u64,
}

impl S3Config {
    /// Create a new S3 configuration for AWS.
    ///
    /// # Arguments
    /// * `bucket` - S3 bucket name
    /// * `region` - AWS region (e.g., "us-east-1")
    /// * `access_key` - AWS access key ID
    /// * `secret_key` - AWS secret access key
    /// * `timestamp` - Current Unix timestamp (seconds since epoch)
    pub fn aws(
        bucket: &str,
        region: &str,
        access_key: &str,
        secret_key: &str,
        timestamp: u64,
    ) -> Self {
        Self {
            endpoint: format!("https://s3.{}.amazonaws.com", region),
            bucket: bucket.to_string(),
            access_key: access_key.to_string(),
            secret_key: secret_key.to_string(),
            region: region.to_string(),
            path_style: false,
            timeout_ms: 30000,
            timestamp,
        }
    }

    /// Create a new S3 configuration for MinIO.
    ///
    /// # Arguments
    /// * `endpoint` - MinIO endpoint URL
    /// * `bucket` - Bucket name
    /// * `access_key` - Access key
    /// * `secret_key` - Secret key
    /// * `timestamp` - Current Unix timestamp (seconds since epoch)
    pub fn minio(
        endpoint: &str,
        bucket: &str,
        access_key: &str,
        secret_key: &str,
        timestamp: u64,
    ) -> Self {
        Self {
            endpoint: endpoint.to_string(),
            bucket: bucket.to_string(),
            access_key: access_key.to_string(),
            secret_key: secret_key.to_string(),
            region: "us-east-1".to_string(),
            path_style: true,
            timeout_ms: 30000,
            timestamp,
        }
    }

    /// Update the timestamp for a new request.
    pub fn with_timestamp(mut self, timestamp: u64) -> Self {
        self.timestamp = timestamp;
        self
    }

    /// Get the host from endpoint.
    pub fn host(&self) -> &str {
        self.endpoint
            .strip_prefix("https://")
            .or_else(|| self.endpoint.strip_prefix("http://"))
            .unwrap_or(&self.endpoint)
    }

    /// Check if using HTTPS.
    pub fn is_https(&self) -> bool {
        self.endpoint.starts_with("https://")
    }

    /// Build the URL for an object.
    pub fn object_url(&self, key: &str) -> String {
        if self.path_style {
            format!("{}/{}/{}", self.endpoint, self.bucket, key)
        } else {
            // Virtual-hosted style: bucket.s3.region.amazonaws.com/key
            let host = self.host();
            let proto = if self.is_https() { "https" } else { "http" };
            format!("{}://{}.{}/{}", proto, self.bucket, host, key)
        }
    }

    /// Get the host for virtual-hosted style.
    pub fn virtual_host(&self) -> String {
        if self.path_style {
            self.host().to_string()
        } else {
            format!("{}.{}", self.bucket, self.host())
        }
    }
}

// ═══════════════════════════════════════════════════════════════════════════════
// CLOUD CREDENTIALS
// ═══════════════════════════════════════════════════════════════════════════════

/// Cloud provider credentials.
#[derive(Debug, Clone)]
pub struct CloudCredentials {
    /// Cloud provider type.
    pub provider: crate::cloud::tier::CloudProvider,
    /// Access key / account ID.
    pub access_key: String,
    /// Secret key / account key.
    pub secret_key: String,
    /// Region or location.
    pub region: String,
    /// Custom endpoint (for MinIO, etc.).
    pub endpoint: Option<String>,
}

impl CloudCredentials {
    /// Create AWS credentials.
    pub fn aws(access_key: &str, secret_key: &str, region: &str) -> Self {
        Self {
            provider: crate::cloud::tier::CloudProvider::AwsS3,
            access_key: access_key.to_string(),
            secret_key: secret_key.to_string(),
            region: region.to_string(),
            endpoint: None,
        }
    }

    /// Create MinIO credentials.
    pub fn minio(endpoint: &str, access_key: &str, secret_key: &str) -> Self {
        Self {
            provider: crate::cloud::tier::CloudProvider::Minio,
            access_key: access_key.to_string(),
            secret_key: secret_key.to_string(),
            region: "us-east-1".to_string(),
            endpoint: Some(endpoint.to_string()),
        }
    }
}

// ═══════════════════════════════════════════════════════════════════════════════
// AWS SIGV4 SIGNING
// ═══════════════════════════════════════════════════════════════════════════════

/// AWS Signature Version 4 signer.
pub struct AwsSigV4Signer {
    /// Access key ID.
    access_key: String,
    /// Secret access key.
    secret_key: String,
    /// AWS region.
    region: String,
    /// AWS service (usually "s3").
    service: String,
}

impl AwsSigV4Signer {
    /// Create a new SigV4 signer.
    pub fn new(access_key: &str, secret_key: &str, region: &str, service: &str) -> Self {
        Self {
            access_key: access_key.to_string(),
            secret_key: secret_key.to_string(),
            region: region.to_string(),
            service: service.to_string(),
        }
    }

    /// Sign an HTTP request and return the Authorization header value.
    ///
    /// # Arguments
    /// * `method` - HTTP method (GET, PUT, etc.)
    /// * `uri` - Request URI path
    /// * `query` - Query string (without leading '?')
    /// * `headers` - Request headers (must include Host, x-amz-date, x-amz-content-sha256)
    /// * `payload_hash` - SHA256 hash of request body (hex-encoded)
    /// * `timestamp` - ISO8601 timestamp (YYYYMMDDTHHMMSSZ format)
    ///
    /// # Returns
    /// The complete Authorization header value.
    pub fn sign_request(
        &self,
        method: &str,
        uri: &str,
        query: &str,
        headers: &[(String, String)],
        payload_hash: &str,
        timestamp: &str,
    ) -> String {
        // Extract date from timestamp (first 8 chars: YYYYMMDD)
        let date = &timestamp[..8];

        // Step 1: Create canonical request
        let canonical_request =
            self.create_canonical_request(method, uri, query, headers, payload_hash);

        // Step 2: Create string to sign
        let string_to_sign = self.create_string_to_sign(&canonical_request, timestamp, date);

        // Step 3: Calculate signature
        let signature = self.calculate_signature(&string_to_sign, date);

        // Step 4: Build authorization header
        let signed_headers = self.get_signed_headers(headers);
        let credential = format!(
            "{}/{}/{}/{}/aws4_request",
            self.access_key, date, self.region, self.service
        );

        format!(
            "AWS4-HMAC-SHA256 Credential={}, SignedHeaders={}, Signature={}",
            credential, signed_headers, signature
        )
    }

    /// Create the canonical request string.
    fn create_canonical_request(
        &self,
        method: &str,
        uri: &str,
        query: &str,
        headers: &[(String, String)],
        payload_hash: &str,
    ) -> String {
        // URI encode the path (but not the slashes)
        let canonical_uri = uri_encode_path(uri);

        // Sort query parameters
        let canonical_query = canonicalize_query(query);

        // Sort headers by lowercase name
        let canonical_headers = self.canonicalize_headers(headers);
        let signed_headers = self.get_signed_headers(headers);

        format!(
            "{}\n{}\n{}\n{}\n{}\n{}",
            method, canonical_uri, canonical_query, canonical_headers, signed_headers, payload_hash
        )
    }

    /// Canonicalize headers (lowercase names, sorted, trimmed values).
    fn canonicalize_headers(&self, headers: &[(String, String)]) -> String {
        let mut sorted: Vec<(String, String)> = headers
            .iter()
            .map(|(k, v)| (k.to_lowercase(), v.trim().to_string()))
            .collect();
        sorted.sort_by(|a, b| a.0.cmp(&b.0));

        sorted
            .iter()
            .map(|(k, v)| format!("{}:{}\n", k, v))
            .collect::<Vec<_>>()
            .join("")
    }

    /// Get signed headers list (semicolon-separated lowercase names).
    fn get_signed_headers(&self, headers: &[(String, String)]) -> String {
        let mut names: Vec<String> = headers.iter().map(|(k, _)| k.to_lowercase()).collect();
        names.sort();
        names.join(";")
    }

    /// Create the string to sign.
    fn create_string_to_sign(
        &self,
        canonical_request: &str,
        timestamp: &str,
        date: &str,
    ) -> String {
        let scope = format!("{}/{}/{}/aws4_request", date, self.region, self.service);
        let canonical_request_hash = sha256_hex(canonical_request.as_bytes());

        format!(
            "AWS4-HMAC-SHA256\n{}\n{}\n{}",
            timestamp, scope, canonical_request_hash
        )
    }

    /// Calculate the signature.
    fn calculate_signature(&self, string_to_sign: &str, date: &str) -> String {
        // Derive signing key
        let k_date = hmac_sha256(
            format!("AWS4{}", self.secret_key).as_bytes(),
            date.as_bytes(),
        );
        let k_region = hmac_sha256(&k_date, self.region.as_bytes());
        let k_service = hmac_sha256(&k_region, self.service.as_bytes());
        let k_signing = hmac_sha256(&k_service, b"aws4_request");

        // Sign the string
        let signature = hmac_sha256(&k_signing, string_to_sign.as_bytes());
        hex_encode(&signature)
    }
}

/// Sign an S3 request (convenience function).
///
/// # Arguments
/// * `method` - HTTP method
/// * `uri` - Request URI path
/// * `headers` - Request headers
/// * `payload_hash` - SHA256 hash of payload
/// * `access_key` - AWS access key ID
/// * `secret_key` - AWS secret access key
/// * `region` - AWS region
/// * `service` - AWS service name
/// * `timestamp` - ISO8601 timestamp
///
/// # Returns
/// The Authorization header value.
#[allow(clippy::too_many_arguments)]
pub fn sign_request_v4(
    method: &str,
    uri: &str,
    headers: &[(String, String)],
    payload_hash: &str,
    access_key: &str,
    secret_key: &str,
    region: &str,
    service: &str,
    timestamp: &str,
) -> String {
    let signer = AwsSigV4Signer::new(access_key, secret_key, region, service);
    signer.sign_request(method, uri, "", headers, payload_hash, timestamp)
}

// ═══════════════════════════════════════════════════════════════════════════════
// NETWORK I/O ABSTRACTION
// ═══════════════════════════════════════════════════════════════════════════════

/// HTTP response from network layer.
#[derive(Debug, Clone)]
pub struct HttpResponse {
    /// HTTP status code.
    pub status: u16,
    /// Response headers.
    pub headers: Vec<(String, String)>,
    /// Response body.
    pub body: Vec<u8>,
}

impl HttpResponse {
    /// Get a header value by name (case-insensitive).
    pub fn get_header(&self, name: &str) -> Option<&str> {
        let name_lower = name.to_lowercase();
        self.headers
            .iter()
            .find(|(k, _)| k.to_lowercase() == name_lower)
            .map(|(_, v)| v.as_str())
    }

    /// Check if the response indicates success.
    pub fn is_success(&self) -> bool {
        (200..300).contains(&self.status)
    }
}

/// Network transport trait for HTTP requests.
///
/// Implement this trait to provide network I/O for S3 operations.
/// For std environments, use `StdNetworkTransport`.
/// For no_std (LunaOS), implement using kernel network stack.
pub trait NetworkTransport: Send + Sync {
    /// Send an HTTP request and receive the response.
    ///
    /// # Arguments
    /// * `host` - Target host
    /// * `port` - Target port
    /// * `use_tls` - Whether to use TLS
    /// * `request` - Raw HTTP request bytes
    /// * `timeout_ms` - Timeout in milliseconds
    ///
    /// # Returns
    /// HTTP response or error.
    fn send_request(
        &self,
        host: &str,
        port: u16,
        use_tls: bool,
        request: &[u8],
        timeout_ms: u32,
    ) -> Result<HttpResponse, S3Error>;
}

/// Standard library network transport (for std feature).
#[cfg(feature = "std")]
pub struct StdNetworkTransport;

#[cfg(feature = "std")]
impl StdNetworkTransport {
    /// Create a new std network transport.
    pub fn new() -> Self {
        Self
    }
}

#[cfg(feature = "std")]
impl Default for StdNetworkTransport {
    fn default() -> Self {
        Self::new()
    }
}

#[cfg(feature = "std")]
impl NetworkTransport for StdNetworkTransport {
    fn send_request(
        &self,
        host: &str,
        port: u16,
        use_tls: bool,
        request: &[u8],
        timeout_ms: u32,
    ) -> Result<HttpResponse, S3Error> {
        use std::io::{Read, Write};
        use std::net::TcpStream;
        use std::time::Duration;

        // Connect to host
        let addr = format!("{}:{}", host, port);
        let mut stream =
            TcpStream::connect(&addr).map_err(|_| S3Error::NetworkError("Failed to connect"))?;

        // Set timeouts
        let timeout = Duration::from_millis(timeout_ms as u64);
        stream.set_read_timeout(Some(timeout)).ok();
        stream.set_write_timeout(Some(timeout)).ok();

        if use_tls {
            // For TLS, we need rustls or native-tls
            // This is a simplified implementation - in production, use proper TLS
            return Err(S3Error::TlsError("TLS not implemented in std transport"));
        }

        // Send request
        stream
            .write_all(request)
            .map_err(|_| S3Error::NetworkError("Failed to send request"))?;

        // Read response
        let mut response_data = Vec::new();
        let mut buf = [0u8; 8192];
        loop {
            match stream.read(&mut buf) {
                Ok(0) => break,
                Ok(n) => response_data.extend_from_slice(&buf[..n]),
                Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => break,
                Err(e) if e.kind() == std::io::ErrorKind::TimedOut => break,
                Err(_) => return Err(S3Error::NetworkError("Failed to read response")),
            }
            // Check if we've received the full response
            if response_data.len() > 4 {
                if let Some(body_start) = find_header_end(&response_data) {
                    // Check Content-Length
                    if let Some(content_length) = parse_content_length(&response_data[..body_start])
                    {
                        let body_len = response_data.len() - body_start;
                        if body_len >= content_length {
                            break;
                        }
                    }
                }
            }
        }

        parse_http_response(&response_data)
    }
}

/// Simulation network transport (for testing without network).
pub struct SimulationTransport {
    /// Simulated responses by key.
    responses: spin::Mutex<alloc::collections::BTreeMap<String, Vec<u8>>>,
}

impl SimulationTransport {
    /// Create a new simulation transport.
    pub fn new() -> Self {
        Self {
            responses: spin::Mutex::new(alloc::collections::BTreeMap::new()),
        }
    }

    /// Set a simulated object.
    pub fn set_object(&self, key: &str, data: Vec<u8>) {
        let mut responses = self.responses.lock();
        responses.insert(key.to_string(), data);
    }

    /// Remove a simulated object.
    pub fn remove_object(&self, key: &str) {
        let mut responses = self.responses.lock();
        responses.remove(key);
    }
}

impl Default for SimulationTransport {
    fn default() -> Self {
        Self::new()
    }
}

impl NetworkTransport for SimulationTransport {
    fn send_request(
        &self,
        _host: &str,
        _port: u16,
        _use_tls: bool,
        request: &[u8],
        _timeout_ms: u32,
    ) -> Result<HttpResponse, S3Error> {
        // Parse the request to determine the operation
        let request_str = core::str::from_utf8(request).unwrap_or("");
        let first_line = request_str.lines().next().unwrap_or("");
        let parts: Vec<&str> = first_line.split_whitespace().collect();

        if parts.len() < 2 {
            return Err(S3Error::Internal("Invalid request"));
        }

        let method = parts[0];
        let path = parts[1];

        // Extract key from path
        let key = path.trim_start_matches('/').split('?').next().unwrap_or("");
        // Remove bucket prefix if present
        let key = if key.contains('/') {
            key.split('/').skip(1).collect::<Vec<_>>().join("/")
        } else {
            key.to_string()
        };

        let responses = self.responses.lock();

        match method {
            "PUT" => {
                // Simulate successful upload
                Ok(HttpResponse {
                    status: 200,
                    headers: vec![(
                        "ETag".to_string(),
                        format!("\"{}\"", sha256_hex(key.as_bytes())),
                    )],
                    body: Vec::new(),
                })
            }
            "GET" => {
                if let Some(data) = responses.get(&key) {
                    Ok(HttpResponse {
                        status: 200,
                        headers: vec![
                            ("Content-Length".to_string(), data.len().to_string()),
                            (
                                "Content-Type".to_string(),
                                "application/octet-stream".to_string(),
                            ),
                        ],
                        body: data.clone(),
                    })
                } else {
                    Ok(HttpResponse {
                        status: 404,
                        headers: Vec::new(),
                        body: b"<Error><Code>NoSuchKey</Code></Error>".to_vec(),
                    })
                }
            }
            "HEAD" => {
                if let Some(data) = responses.get(&key) {
                    Ok(HttpResponse {
                        status: 200,
                        headers: vec![
                            ("Content-Length".to_string(), data.len().to_string()),
                            (
                                "Content-Type".to_string(),
                                "application/octet-stream".to_string(),
                            ),
                            ("ETag".to_string(), format!("\"{}\"", sha256_hex(data))),
                            ("x-amz-storage-class".to_string(), "STANDARD".to_string()),
                        ],
                        body: Vec::new(),
                    })
                } else {
                    Ok(HttpResponse {
                        status: 404,
                        headers: Vec::new(),
                        body: Vec::new(),
                    })
                }
            }
            "DELETE" => Ok(HttpResponse {
                status: 204,
                headers: Vec::new(),
                body: Vec::new(),
            }),
            _ => Err(S3Error::Internal("Unknown method")),
        }
    }
}

// ═══════════════════════════════════════════════════════════════════════════════
// S3 CLIENT
// ═══════════════════════════════════════════════════════════════════════════════

/// S3 client for object storage operations.
pub struct S3Client<T: NetworkTransport> {
    /// Client configuration.
    config: S3Config,
    /// Network transport.
    transport: T,
    /// SigV4 signer.
    signer: AwsSigV4Signer,
}

impl<T: NetworkTransport> S3Client<T> {
    /// Create a new S3 client.
    pub fn new(config: S3Config, transport: T) -> Self {
        let signer =
            AwsSigV4Signer::new(&config.access_key, &config.secret_key, &config.region, "s3");
        Self {
            config,
            transport,
            signer,
        }
    }

    /// Get the current timestamp in ISO8601 format for AWS SigV4 signing.
    ///
    /// Converts the Unix timestamp from config to AWS date format: YYYYMMDDTHHMMSSZ
    fn get_timestamp(&self) -> String {
        // Convert Unix timestamp to AWS date format
        // Unix timestamp is seconds since 1970-01-01 00:00:00 UTC
        let ts = self.config.timestamp;

        // Calculate date/time components from Unix timestamp
        // Using simplified calculation (doesn't account for leap seconds)
        let secs_per_min = 60u64;
        let secs_per_hour = 3600u64;
        let secs_per_day = 86400u64;

        let days_since_epoch = ts / secs_per_day;
        let secs_today = ts % secs_per_day;

        let hour = (secs_today / secs_per_hour) as u32;
        let minute = ((secs_today % secs_per_hour) / secs_per_min) as u32;
        let second = (secs_today % secs_per_min) as u32;

        // Calculate year, month, day from days since epoch (1970-01-01)
        // Simplified Gregorian calendar calculation
        let (year, month, day) = Self::days_to_ymd(days_since_epoch);

        format!(
            "{:04}{:02}{:02}T{:02}{:02}{:02}Z",
            year, month, day, hour, minute, second
        )
    }

    /// Convert days since Unix epoch to year, month, day.
    fn days_to_ymd(days: u64) -> (u32, u32, u32) {
        // Days in each month (non-leap year)
        const DAYS_IN_MONTH: [u32; 12] = [31, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31];

        let mut year = 1970u32;
        let mut remaining = days;

        // Find the year
        loop {
            let days_in_year = if Self::is_leap_year(year) { 366 } else { 365 };
            if remaining < days_in_year {
                break;
            }
            remaining -= days_in_year;
            year += 1;
        }

        // Find the month
        let is_leap = Self::is_leap_year(year);
        let mut month = 1u32;
        for (i, &days_in_month) in DAYS_IN_MONTH.iter().enumerate() {
            let days = if i == 1 && is_leap {
                29
            } else {
                days_in_month as u64
            };
            if remaining < days {
                month = (i + 1) as u32;
                break;
            }
            remaining -= days;
        }

        let day = (remaining + 1) as u32;

        (year, month, day)
    }

    /// Check if a year is a leap year.
    fn is_leap_year(year: u32) -> bool {
        (year % 4 == 0 && year % 100 != 0) || (year % 400 == 0)
    }

    /// Build common headers for a request.
    fn build_headers(
        &self,
        content_length: usize,
        content_sha256: &str,
        timestamp: &str,
        extra_headers: &[(String, String)],
    ) -> Vec<(String, String)> {
        let mut headers = vec![
            ("Host".to_string(), self.config.virtual_host()),
            ("x-amz-date".to_string(), timestamp.to_string()),
            (
                "x-amz-content-sha256".to_string(),
                content_sha256.to_string(),
            ),
        ];

        if content_length > 0 {
            headers.push(("Content-Length".to_string(), content_length.to_string()));
        }

        for (k, v) in extra_headers {
            headers.push((k.clone(), v.clone()));
        }

        headers
    }

    /// Build an HTTP request.
    fn build_request(
        &self,
        method: &str,
        key: &str,
        headers: &[(String, String)],
        body: &[u8],
    ) -> Vec<u8> {
        let uri = if self.config.path_style {
            format!("/{}/{}", self.config.bucket, key)
        } else {
            format!("/{}", key)
        };

        let mut request = format!("{} {} HTTP/1.1\r\n", method, uri);

        for (name, value) in headers {
            request.push_str(&format!("{}: {}\r\n", name, value));
        }

        request.push_str("\r\n");

        let mut bytes = request.into_bytes();
        bytes.extend_from_slice(body);
        bytes
    }

    /// Send a request and get response.
    fn send_request(
        &self,
        method: &str,
        key: &str,
        body: &[u8],
        extra_headers: &[(String, String)],
    ) -> Result<HttpResponse, S3Error> {
        let timestamp = self.get_timestamp();
        let content_sha256 = sha256_hex(body);

        let mut headers =
            self.build_headers(body.len(), &content_sha256, &timestamp, extra_headers);

        // Sign the request
        let uri = if self.config.path_style {
            format!("/{}/{}", self.config.bucket, key)
        } else {
            format!("/{}", key)
        };

        let auth_header =
            self.signer
                .sign_request(method, &uri, "", &headers, &content_sha256, &timestamp);

        headers.push(("Authorization".to_string(), auth_header));

        let request = self.build_request(method, key, &headers, body);

        let host = self.config.host();
        let port = if self.config.is_https() { 443 } else { 80 };

        self.transport.send_request(
            host,
            port,
            self.config.is_https(),
            &request,
            self.config.timeout_ms,
        )
    }

    /// Upload an object to S3.
    ///
    /// # Arguments
    /// * `key` - Object key (path in bucket)
    /// * `data` - Object data
    /// * `storage_class` - S3 storage class
    ///
    /// # Returns
    /// ETag of the uploaded object.
    pub fn put_object(
        &self,
        key: &str,
        data: &[u8],
        storage_class: S3StorageClass,
    ) -> Result<String, S3Error> {
        let extra_headers = vec![
            (
                "x-amz-storage-class".to_string(),
                storage_class.as_str().to_string(),
            ),
            (
                "Content-Type".to_string(),
                "application/octet-stream".to_string(),
            ),
        ];

        let response = self.send_request("PUT", key, data, &extra_headers)?;

        if let Some(err) = S3Error::from_status(response.status) {
            return Err(err);
        }

        let etag = response
            .get_header("ETag")
            .unwrap_or("")
            .trim_matches('"')
            .to_string();

        Ok(etag)
    }

    /// Download an object from S3.
    ///
    /// # Arguments
    /// * `key` - Object key
    ///
    /// # Returns
    /// Object data.
    pub fn get_object(&self, key: &str) -> Result<Vec<u8>, S3Error> {
        let response = self.send_request("GET", key, &[], &[])?;

        if let Some(err) = S3Error::from_status(response.status) {
            return Err(err);
        }

        Ok(response.body)
    }

    /// Delete an object from S3.
    ///
    /// # Arguments
    /// * `key` - Object key
    pub fn delete_object(&self, key: &str) -> Result<(), S3Error> {
        let response = self.send_request("DELETE", key, &[], &[])?;

        if let Some(err) = S3Error::from_status(response.status) {
            return Err(err);
        }

        Ok(())
    }

    /// Get object metadata (HEAD request).
    ///
    /// # Arguments
    /// * `key` - Object key
    ///
    /// # Returns
    /// Object metadata.
    pub fn head_object(&self, key: &str) -> Result<ObjectMetadata, S3Error> {
        let response = self.send_request("HEAD", key, &[], &[])?;

        if let Some(err) = S3Error::from_status(response.status) {
            return Err(err);
        }

        let content_length = response
            .get_header("Content-Length")
            .and_then(|s| s.parse().ok())
            .unwrap_or(0);

        let content_type = response
            .get_header("Content-Type")
            .unwrap_or("application/octet-stream")
            .to_string();

        let etag = response
            .get_header("ETag")
            .unwrap_or("")
            .trim_matches('"')
            .to_string();

        let storage_class = response
            .get_header("x-amz-storage-class")
            .and_then(|s| s.parse().ok())
            .unwrap_or(S3StorageClass::Standard);

        let version_id = response.get_header("x-amz-version-id").map(String::from);

        Ok(ObjectMetadata {
            content_length,
            content_type,
            etag,
            last_modified: 0, // Would need to parse Last-Modified header
            storage_class,
            version_id,
        })
    }

    /// Copy an object within S3 (also used for storage class transitions).
    ///
    /// # Arguments
    /// * `src_key` - Source object key
    /// * `dst_key` - Destination object key
    /// * `storage_class` - Target storage class
    pub fn copy_object(
        &self,
        src_key: &str,
        dst_key: &str,
        storage_class: S3StorageClass,
    ) -> Result<(), S3Error> {
        // x-amz-copy-source always uses /{bucket}/{key} format
        let copy_source = format!("/{}/{}", self.config.bucket, src_key);

        let extra_headers = vec![
            ("x-amz-copy-source".to_string(), copy_source),
            (
                "x-amz-storage-class".to_string(),
                storage_class.as_str().to_string(),
            ),
            ("x-amz-metadata-directive".to_string(), "COPY".to_string()),
        ];

        let response = self.send_request("PUT", dst_key, &[], &extra_headers)?;

        if let Some(err) = S3Error::from_status(response.status) {
            return Err(err);
        }

        Ok(())
    }

    /// Transition an object to a different storage class.
    ///
    /// This performs a copy-in-place operation with a new storage class.
    ///
    /// # Arguments
    /// * `key` - Object key
    /// * `new_class` - Target storage class
    pub fn transition_storage_class(
        &self,
        key: &str,
        new_class: S3StorageClass,
    ) -> Result<(), S3Error> {
        // Copy object to itself with new storage class
        self.copy_object(key, key, new_class)
    }
}

// ═══════════════════════════════════════════════════════════════════════════════
// HELPER FUNCTIONS
// ═══════════════════════════════════════════════════════════════════════════════

/// SHA-256 hash function returning hex string.
fn sha256_hex(data: &[u8]) -> String {
    use sha2::{Digest, Sha256};
    let mut hasher = Sha256::new();
    hasher.update(data);
    let result = hasher.finalize();
    hex_encode(&result)
}

/// HMAC-SHA256.
fn hmac_sha256(key: &[u8], data: &[u8]) -> [u8; 32] {
    use hmac::{Hmac, Mac};
    use sha2::Sha256;

    type HmacSha256 = Hmac<Sha256>;

    let mut mac = HmacSha256::new_from_slice(key).expect("HMAC key length");
    mac.update(data);
    let result = mac.finalize();
    let mut output = [0u8; 32];
    output.copy_from_slice(&result.into_bytes());
    output
}

/// Encode bytes as hex string.
fn hex_encode(data: &[u8]) -> String {
    hex_encode_crate(data)
}

/// URI-encode a path component (preserving slashes).
fn uri_encode_path(path: &str) -> String {
    utf8_percent_encode(path, URI_PATH_SET).to_string()
}

/// Canonicalize query string (sort and encode).
fn canonicalize_query(query: &str) -> String {
    if query.is_empty() {
        return String::new();
    }

    let mut pairs: Vec<(&str, &str)> = query
        .split('&')
        .filter_map(|param| {
            let mut parts = param.splitn(2, '=');
            Some((parts.next()?, parts.next().unwrap_or("")))
        })
        .collect();

    pairs.sort_by(|a, b| a.0.cmp(b.0).then(a.1.cmp(b.1)));

    pairs
        .iter()
        .map(|(k, v)| format!("{}={}", uri_encode_component(k), uri_encode_component(v)))
        .collect::<Vec<_>>()
        .join("&")
}

/// URI-encode a component (encodes everything except unreserved chars).
fn uri_encode_component(s: &str) -> String {
    let mut result = String::with_capacity(s.len() * 3);
    for c in s.chars() {
        match c {
            'A'..='Z' | 'a'..='z' | '0'..='9' | '-' | '_' | '.' | '~' => {
                result.push(c);
            }
            _ => {
                for byte in c.to_string().as_bytes() {
                    result.push_str(&format!("%{:02X}", byte));
                }
            }
        }
    }
    result
}

/// Find the end of HTTP headers (double CRLF).
fn find_header_end(data: &[u8]) -> Option<usize> {
    for i in 0..data.len().saturating_sub(3) {
        if data[i..i + 4] == *b"\r\n\r\n" {
            return Some(i + 4);
        }
    }
    None
}

/// Parse Content-Length from HTTP headers.
fn parse_content_length(headers: &[u8]) -> Option<usize> {
    let headers_str = core::str::from_utf8(headers).ok()?;
    for line in headers_str.lines() {
        let lower = line.to_lowercase();
        if lower.starts_with("content-length:") {
            return line.split(':').nth(1)?.trim().parse().ok();
        }
    }
    None
}

/// Parse an HTTP response.
fn parse_http_response(data: &[u8]) -> Result<HttpResponse, S3Error> {
    let header_end = find_header_end(data).ok_or(S3Error::NetworkError("Invalid HTTP response"))?;

    let header_bytes = &data[..header_end];
    let body = data[header_end..].to_vec();

    let header_str = core::str::from_utf8(header_bytes)
        .map_err(|_| S3Error::NetworkError("Invalid UTF-8 in headers"))?;

    let mut lines = header_str.lines();
    let status_line = lines
        .next()
        .ok_or(S3Error::NetworkError("Missing status line"))?;

    // Parse status code from "HTTP/1.1 200 OK"
    let status: u16 = status_line
        .split_whitespace()
        .nth(1)
        .and_then(|s| s.parse().ok())
        .ok_or(S3Error::NetworkError("Invalid status code"))?;

    let mut headers = Vec::new();
    for line in lines {
        if line.is_empty() {
            break;
        }
        if let Some((name, value)) = line.split_once(':') {
            headers.push((name.trim().to_string(), value.trim().to_string()));
        }
    }

    Ok(HttpResponse {
        status,
        headers,
        body,
    })
}

// ═══════════════════════════════════════════════════════════════════════════════
// INTEGRATION WITH CLOUD TIER MANAGER
// ═══════════════════════════════════════════════════════════════════════════════

/// S3-backed cloud tier manager.
pub struct S3CloudTierManager<T: NetworkTransport> {
    /// S3 client.
    client: S3Client<T>,
    /// Base key prefix for all objects.
    key_prefix: String,
}

impl<T: NetworkTransport> S3CloudTierManager<T> {
    /// Create a new S3 cloud tier manager.
    pub fn new(config: S3Config, transport: T, key_prefix: &str) -> Self {
        Self {
            client: S3Client::new(config, transport),
            key_prefix: key_prefix.to_string(),
        }
    }

    /// Build object key from dataset and block offset.
    fn build_key(&self, dataset_id: u64, block_offset: u64) -> String {
        format!(
            "{}/dataset_{:016x}/block_{:016x}",
            self.key_prefix, dataset_id, block_offset
        )
    }

    /// Upload a block to S3.
    pub fn upload_block(
        &self,
        dataset_id: u64,
        block_offset: u64,
        data: &[u8],
        storage_class: CloudStorageClass,
    ) -> Result<String, S3Error> {
        let key = self.build_key(dataset_id, block_offset);
        let s3_class = S3StorageClass::from_cloud_class(storage_class);
        self.client.put_object(&key, data, s3_class)
    }

    /// Download a block from S3.
    pub fn download_block(&self, dataset_id: u64, block_offset: u64) -> Result<Vec<u8>, S3Error> {
        let key = self.build_key(dataset_id, block_offset);
        self.client.get_object(&key)
    }

    /// Delete a block from S3.
    pub fn delete_block(&self, dataset_id: u64, block_offset: u64) -> Result<(), S3Error> {
        let key = self.build_key(dataset_id, block_offset);
        self.client.delete_object(&key)
    }

    /// Transition a block to a different storage class.
    pub fn transition_block(
        &self,
        dataset_id: u64,
        block_offset: u64,
        new_class: CloudStorageClass,
    ) -> Result<(), S3Error> {
        let key = self.build_key(dataset_id, block_offset);
        let s3_class = S3StorageClass::from_cloud_class(new_class);
        self.client.transition_storage_class(&key, s3_class)
    }

    /// Get block metadata.
    pub fn get_block_metadata(
        &self,
        dataset_id: u64,
        block_offset: u64,
    ) -> Result<ObjectMetadata, S3Error> {
        let key = self.build_key(dataset_id, block_offset);
        self.client.head_object(&key)
    }
}

// ═══════════════════════════════════════════════════════════════════════════════
// TESTS
// ═══════════════════════════════════════════════════════════════════════════════

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_s3_error_from_status() {
        assert!(S3Error::from_status(200).is_none());
        assert!(S3Error::from_status(204).is_none());
        assert_eq!(
            S3Error::from_status(401),
            Some(S3Error::AuthenticationFailed)
        );
        assert_eq!(S3Error::from_status(403), Some(S3Error::AccessDenied));
        assert_eq!(S3Error::from_status(404), Some(S3Error::ObjectNotFound));
    }

    #[test]
    fn test_s3_storage_class_conversion() {
        assert_eq!(S3StorageClass::Standard.as_str(), "STANDARD");
        assert_eq!(
            S3StorageClass::from_str("GLACIER"),
            Ok(S3StorageClass::Glacier)
        );
        assert!(S3StorageClass::from_str("INVALID").is_err());
    }

    #[test]
    fn test_cloud_storage_class_mapping() {
        assert_eq!(
            S3StorageClass::from_cloud_class(CloudStorageClass::Hot),
            S3StorageClass::Standard
        );
        assert_eq!(
            S3StorageClass::from_cloud_class(CloudStorageClass::Archive),
            S3StorageClass::DeepArchive
        );
        assert_eq!(
            S3StorageClass::Standard.to_cloud_class(),
            CloudStorageClass::Hot
        );
    }

    #[test]
    fn test_s3_config_aws() {
        // Unix timestamp for 2025-01-01 00:00:00 UTC
        let ts = 1735689600u64;
        let config = S3Config::aws("my-bucket", "us-east-1", "AKID", "SECRET", ts);
        assert_eq!(config.bucket, "my-bucket");
        assert_eq!(config.region, "us-east-1");
        assert!(config.is_https());
        assert!(!config.path_style);
        assert_eq!(config.timestamp, ts);
    }

    #[test]
    fn test_s3_config_minio() {
        // Unix timestamp for 2025-06-15 12:30:45 UTC
        let ts = 1750073445u64;
        let config = S3Config::minio(
            "http://localhost:9000",
            "my-bucket",
            "minioadmin",
            "minioadmin",
            ts,
        );
        assert_eq!(config.bucket, "my-bucket");
        assert!(!config.is_https());
        assert!(config.path_style);
        assert_eq!(config.timestamp, ts);
    }

    #[test]
    fn test_hex_encode() {
        assert_eq!(hex_encode(&[0x00, 0x01, 0xff]), "0001ff");
        assert_eq!(hex_encode(&[0xde, 0xad, 0xbe, 0xef]), "deadbeef");
    }

    #[test]
    fn test_uri_encode_path() {
        assert_eq!(uri_encode_path("/foo/bar"), "/foo/bar");
        assert_eq!(uri_encode_path("/foo bar/baz"), "/foo%20bar/baz");
        assert_eq!(uri_encode_path("/a-b_c.d~e"), "/a-b_c.d~e");
    }

    #[test]
    fn test_uri_encode_component() {
        assert_eq!(uri_encode_component("foo"), "foo");
        assert_eq!(uri_encode_component("foo bar"), "foo%20bar");
        assert_eq!(uri_encode_component("a=b&c=d"), "a%3Db%26c%3Dd");
    }

    #[test]
    fn test_canonicalize_query() {
        assert_eq!(canonicalize_query(""), "");
        assert_eq!(canonicalize_query("b=2&a=1"), "a=1&b=2");
        assert_eq!(canonicalize_query("a=1&a=2"), "a=1&a=2");
    }

    #[test]
    fn test_sha256_hex() {
        // SHA256 of empty string
        let hash = sha256_hex(b"");
        assert_eq!(
            hash,
            "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
        );
    }

    #[test]
    fn test_simulation_transport_put_get() {
        let transport = SimulationTransport::new();
        transport.set_object("test/key", b"hello world".to_vec());

        let config = S3Config::minio(
            "http://localhost:9000",
            "test-bucket",
            "admin",
            "admin",
            1735689600,
        );
        let client = S3Client::new(config, transport);

        // GET should work for pre-set object
        // Note: In simulation, PUT always succeeds
    }

    #[test]
    fn test_http_response_get_header() {
        let response = HttpResponse {
            status: 200,
            headers: vec![
                ("Content-Type".to_string(), "application/json".to_string()),
                ("ETag".to_string(), "\"abc123\"".to_string()),
            ],
            body: Vec::new(),
        };

        assert_eq!(
            response.get_header("content-type"),
            Some("application/json")
        );
        assert_eq!(
            response.get_header("Content-Type"),
            Some("application/json")
        );
        assert_eq!(response.get_header("etag"), Some("\"abc123\""));
        assert_eq!(response.get_header("missing"), None);
    }

    #[test]
    fn test_object_metadata_default() {
        let meta = ObjectMetadata::default();
        assert_eq!(meta.content_length, 0);
        assert_eq!(meta.storage_class, S3StorageClass::Standard);
        assert!(meta.version_id.is_none());
    }

    #[test]
    fn test_cloud_credentials() {
        let creds = CloudCredentials::aws("AKID", "SECRET", "us-west-2");
        assert_eq!(creds.provider, crate::cloud::tier::CloudProvider::AwsS3);
        assert_eq!(creds.region, "us-west-2");
        assert!(creds.endpoint.is_none());

        let minio_creds = CloudCredentials::minio("http://minio:9000", "admin", "password");
        assert_eq!(
            minio_creds.provider,
            crate::cloud::tier::CloudProvider::Minio
        );
        assert!(minio_creds.endpoint.is_some());
    }

    #[test]
    fn test_build_key() {
        let transport = SimulationTransport::new();
        let config = S3Config::minio(
            "http://localhost:9000",
            "bucket",
            "admin",
            "admin",
            1735689600,
        );
        let manager = S3CloudTierManager::new(config, transport, "lcpfs");

        let key = manager.build_key(1, 0x1000);
        assert!(key.starts_with("lcpfs/"));
        assert!(key.contains("dataset_"));
        assert!(key.contains("block_"));
    }

    #[test]
    fn test_sigv4_signer_creates_auth_header() {
        let signer =
            AwsSigV4Signer::new("AKIAIOSFODNN7EXAMPLE", "wJalrXUtnFEMI", "us-east-1", "s3");

        let headers = vec![
            (
                "Host".to_string(),
                "examplebucket.s3.amazonaws.com".to_string(),
            ),
            ("x-amz-date".to_string(), "20130524T000000Z".to_string()),
            (
                "x-amz-content-sha256".to_string(),
                "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855".to_string(),
            ),
        ];

        let auth = signer.sign_request(
            "GET",
            "/test.txt",
            "",
            &headers,
            "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855",
            "20130524T000000Z",
        );

        assert!(auth.starts_with("AWS4-HMAC-SHA256 Credential="));
        assert!(auth.contains("SignedHeaders="));
        assert!(auth.contains("Signature="));
    }

    #[test]
    fn test_timestamp_conversion() {
        // Test 2025-01-01 00:00:00 UTC (Unix timestamp 1735689600)
        let config = S3Config::minio(
            "http://localhost:9000",
            "bucket",
            "admin",
            "admin",
            1735689600,
        );
        let client = S3Client::new(config, SimulationTransport::new());
        assert_eq!(client.get_timestamp(), "20250101T000000Z");

        // Test 2025-06-16 11:30:45 UTC
        let config2 = S3Config::minio(
            "http://localhost:9000",
            "bucket",
            "admin",
            "admin",
            1750073445,
        );
        let client2 = S3Client::new(config2, SimulationTransport::new());
        assert_eq!(client2.get_timestamp(), "20250616T113045Z");

        // Test leap year handling: 2024-02-29 23:59:59 UTC
        let config3 = S3Config::minio(
            "http://localhost:9000",
            "bucket",
            "admin",
            "admin",
            1709251199,
        );
        let client3 = S3Client::new(config3, SimulationTransport::new());
        assert_eq!(client3.get_timestamp(), "20240229T235959Z");
    }

    #[test]
    fn test_timestamp_provider_trait() {
        let fixed = FixedTimestamp(1735689600);
        assert_eq!(fixed.unix_timestamp(), 1735689600);
    }
}