lcpfs 2026.1.102

// Copyright 2025 LunaOS Contributors
// SPDX-License-Identifier: Apache-2.0

//! S3 operations implementation.
//!
//! Maps S3 API operations to LCPFS filesystem operations.

use alloc::collections::BTreeMap;
use alloc::string::{String, ToString};
use alloc::sync::Arc;
use alloc::vec;
use alloc::vec::Vec;

use super::types::{
    BucketInfo, HttpResponse, ListObjectsParams, ListObjectsResult, MultipartUpload, S3Error,
    S3GatewayConfig, S3ObjectMeta, S3ObjectVersion, S3Request, UploadPart,
};
use super::xml;

// ═══════════════════════════════════════════════════════════════════════════════
// STORAGE PROVIDER TRAIT
// ═══════════════════════════════════════════════════════════════════════════════

/// Trait for filesystem operations required by S3 gateway.
pub trait StorageProvider: Send + Sync {
    // Dataset (bucket) operations
    /// List all datasets.
    fn list_datasets(&self) -> Result<Vec<DatasetInfo>, String>;
    /// Create a dataset.
    fn create_dataset(&mut self, name: &str) -> Result<(), String>;
    /// Delete a dataset.
    fn delete_dataset(&mut self, name: &str) -> Result<(), String>;
    /// Check if dataset exists.
    fn dataset_exists(&self, name: &str) -> Result<bool, String>;

    // File (object) operations
    /// List files in a dataset with optional prefix.
    fn list_files(&self, dataset: &str, prefix: Option<&str>) -> Result<Vec<FileInfo>, String>;
    /// Read a file.
    fn read_file(&self, dataset: &str, path: &str) -> Result<Vec<u8>, String>;
    /// Read a file range.
    fn read_file_range(
        &self,
        dataset: &str,
        path: &str,
        start: u64,
        end: Option<u64>,
    ) -> Result<Vec<u8>, String>;
    /// Write a file.
    fn write_file(&mut self, dataset: &str, path: &str, data: &[u8]) -> Result<(), String>;
    /// Delete a file.
    fn delete_file(&mut self, dataset: &str, path: &str) -> Result<(), String>;
    /// Check if file exists.
    fn file_exists(&self, dataset: &str, path: &str) -> Result<bool, String>;
    /// Get file metadata.
    fn file_info(&self, dataset: &str, path: &str) -> Result<Option<FileInfo>, String>;
    /// Copy a file.
    fn copy_file(
        &mut self,
        src_dataset: &str,
        src_path: &str,
        dst_dataset: &str,
        dst_path: &str,
    ) -> Result<(), String>;

    // Versioning operations (optional - maps to snapshots)
    /// Get current version ID for a file.
    fn get_version_id(&self, dataset: &str, path: &str) -> Result<String, String> {
        // Default: use TXG as version
        Ok("1".into())
    }

    /// List versions of a file.
    fn list_versions(
        &self,
        dataset: &str,
        prefix: Option<&str>,
    ) -> Result<Vec<VersionInfo>, String> {
        Ok(Vec::new())
    }

    // Temp storage for multipart
    /// Write a temp file (for multipart parts).
    fn write_temp(&mut self, key: &str, data: &[u8]) -> Result<(), String>;
    /// Read a temp file.
    fn read_temp(&self, key: &str) -> Result<Vec<u8>, String>;
    /// Delete a temp file.
    fn delete_temp(&mut self, key: &str) -> Result<(), String>;
    /// List temp files with prefix.
    fn list_temp(&self, prefix: &str) -> Result<Vec<String>, String>;

    /// Get current timestamp.
    fn current_timestamp(&self) -> u64;
}

/// Dataset info from storage.
#[derive(Debug, Clone)]
pub struct DatasetInfo {
    /// Dataset name.
    pub name: String,
    /// Creation time.
    pub created: u64,
}

/// File info from storage.
#[derive(Debug, Clone)]
pub struct FileInfo {
    /// File path.
    pub path: String,
    /// File size.
    pub size: u64,
    /// Modification time.
    pub mtime: u64,
    /// Is directory.
    pub is_dir: bool,
    /// Checksum (for ETag).
    pub checksum: Option<String>,
}

/// Version info.
#[derive(Debug, Clone)]
pub struct VersionInfo {
    /// File path.
    pub path: String,
    /// Version ID.
    pub version_id: String,
    /// Is latest.
    pub is_latest: bool,
    /// Modification time.
    pub mtime: u64,
    /// Size.
    pub size: u64,
    /// Checksum.
    pub checksum: String,
}

// ═══════════════════════════════════════════════════════════════════════════════
// S3 OPERATIONS HANDLER
// ═══════════════════════════════════════════════════════════════════════════════

/// S3 operations handler.
pub struct S3Ops<P: StorageProvider> {
    /// Storage provider.
    provider: P,
    /// Configuration.
    config: S3GatewayConfig,
    /// Active multipart uploads.
    multipart_uploads: BTreeMap<String, MultipartUpload>,
    /// Next upload ID.
    next_upload_id: u64,
}

impl<P: StorageProvider> S3Ops<P> {
    /// Create a new S3 operations handler.
    pub fn new(provider: P, config: S3GatewayConfig) -> Self {
        Self {
            provider,
            config,
            multipart_uploads: BTreeMap::new(),
            next_upload_id: 1,
        }
    }

    /// Get the storage provider.
    pub fn provider(&self) -> &P {
        &self.provider
    }

    /// Get the config.
    pub fn config(&self) -> &S3GatewayConfig {
        &self.config
    }

    /// Map bucket name to dataset.
    fn bucket_to_dataset(&self, bucket: &str) -> Result<String, S3Error> {
        self.config
            .dataset_for_bucket(bucket)
            .cloned()
            .ok_or(S3Error::NoSuchBucket)
    }

    // ═══════════════════════════════════════════════════════════════════════════
    // BUCKET OPERATIONS
    // ═══════════════════════════════════════════════════════════════════════════

    /// List all buckets.
    pub fn list_buckets(&self) -> Result<HttpResponse, S3Error> {
        let datasets = self
            .provider
            .list_datasets()
            .map_err(S3Error::InternalError)?;

        let buckets: Vec<BucketInfo> = datasets
            .into_iter()
            .map(|d| BucketInfo {
                name: d.name,
                creation_date: xml::format_timestamp(d.created),
            })
            .collect();

        let xml = xml::list_buckets_xml(&self.config.access_key, "S3 Gateway", &buckets);
        Ok(HttpResponse::ok().with_xml(xml))
    }

    /// Create a bucket.
    pub fn create_bucket(&mut self, bucket: &str) -> Result<HttpResponse, S3Error> {
        // Validate bucket name
        validate_bucket_name(bucket)?;

        // Check if it already exists
        let exists = self
            .provider
            .dataset_exists(bucket)
            .map_err(S3Error::InternalError)?;

        if exists {
            return Err(S3Error::BucketAlreadyExists);
        }

        // Create dataset
        self.provider
            .create_dataset(bucket)
            .map_err(S3Error::InternalError)?;

        // Add to bucket map
        self.config.bucket_map.insert(bucket.into(), bucket.into());

        Ok(HttpResponse::ok().with_header("Location", alloc::format!("/{}", bucket)))
    }

    /// Delete a bucket.
    pub fn delete_bucket(&mut self, bucket: &str) -> Result<HttpResponse, S3Error> {
        let dataset = self.bucket_to_dataset(bucket)?;

        // Check if empty
        let files = self
            .provider
            .list_files(&dataset, None)
            .map_err(S3Error::InternalError)?;

        if !files.is_empty() {
            return Err(S3Error::BucketNotEmpty);
        }

        // Delete dataset
        self.provider
            .delete_dataset(&dataset)
            .map_err(S3Error::InternalError)?;

        // Remove from bucket map
        self.config.bucket_map.remove(bucket);

        Ok(HttpResponse::no_content())
    }

    /// Head bucket (check existence).
    pub fn head_bucket(&self, bucket: &str) -> Result<HttpResponse, S3Error> {
        self.bucket_to_dataset(bucket)?;
        Ok(HttpResponse::ok())
    }

    /// Get bucket location.
    pub fn get_bucket_location(&self, bucket: &str) -> Result<HttpResponse, S3Error> {
        self.bucket_to_dataset(bucket)?;
        let xml = xml::bucket_location_xml(&self.config.region);
        Ok(HttpResponse::ok().with_xml(xml))
    }

    /// Get bucket versioning.
    pub fn get_bucket_versioning(&self, bucket: &str) -> Result<HttpResponse, S3Error> {
        self.bucket_to_dataset(bucket)?;
        let status = if self.config.enable_versioning {
            "Enabled"
        } else {
            ""
        };
        let xml = xml::bucket_versioning_xml(status);
        Ok(HttpResponse::ok().with_xml(xml))
    }

    // ═══════════════════════════════════════════════════════════════════════════
    // OBJECT LISTING
    // ═══════════════════════════════════════════════════════════════════════════

    /// List objects v2.
    pub fn list_objects_v2(
        &self,
        bucket: &str,
        params: &ListObjectsParams,
    ) -> Result<HttpResponse, S3Error> {
        let dataset = self.bucket_to_dataset(bucket)?;

        let files = self
            .provider
            .list_files(&dataset, params.prefix.as_deref())
            .map_err(S3Error::InternalError)?;

        let mut contents = Vec::new();
        let mut common_prefixes = Vec::new();

        for file in files {
            if file.is_dir {
                continue;
            }

            // Apply start_after filter
            if let Some(ref start_after) = params.start_after {
                if file.path <= *start_after {
                    continue;
                }
            }

            // Handle delimiter (common prefixes)
            if let Some(ref delimiter) = params.delimiter {
                if let Some(ref prefix) = params.prefix {
                    let relative = &file.path[prefix.len()..];
                    if let Some(idx) = relative.find(delimiter) {
                        let common_prefix =
                            alloc::format!("{}{}{}", prefix, &relative[..idx], delimiter);
                        if !common_prefixes.contains(&common_prefix) {
                            common_prefixes.push(common_prefix);
                        }
                        continue;
                    }
                } else if let Some(idx) = file.path.find(delimiter) {
                    let common_prefix = alloc::format!("{}{}", &file.path[..idx], delimiter);
                    if !common_prefixes.contains(&common_prefix) {
                        common_prefixes.push(common_prefix);
                    }
                    continue;
                }
            }

            let etag = file
                .checksum
                .unwrap_or_else(|| alloc::format!("\"{}\"", file.mtime));

            contents.push(S3ObjectMeta::new(
                file.path,
                file.size,
                etag,
                xml::format_timestamp(file.mtime),
            ));
        }

        // Sort by key
        contents.sort_by(|a, b| a.key.cmp(&b.key));
        common_prefixes.sort();

        // Apply max_keys limit
        let max_keys = params.max_keys as usize;
        let is_truncated = contents.len() > max_keys;
        let next_token = if is_truncated {
            contents.get(max_keys).map(|o| o.key.clone())
        } else {
            None
        };
        contents.truncate(max_keys);

        let result = ListObjectsResult {
            key_count: contents.len(),
            contents,
            common_prefixes,
            is_truncated,
            next_continuation_token: next_token,
        };

        let xml = xml::list_objects_v2_xml(bucket, params, &result);
        Ok(HttpResponse::ok().with_xml(xml))
    }

    // ═══════════════════════════════════════════════════════════════════════════
    // OBJECT OPERATIONS
    // ═══════════════════════════════════════════════════════════════════════════

    /// Put object.
    pub fn put_object(
        &mut self,
        bucket: &str,
        key: &str,
        data: &[u8],
        content_type: Option<&str>,
    ) -> Result<HttpResponse, S3Error> {
        let dataset = self.bucket_to_dataset(bucket)?;

        self.provider
            .write_file(&dataset, key, data)
            .map_err(S3Error::InternalError)?;

        let etag = xml::compute_etag(data);

        Ok(HttpResponse::ok().with_header("ETag", etag))
    }

    /// Get object.
    pub fn get_object(
        &self,
        bucket: &str,
        key: &str,
        range: Option<(u64, Option<u64>)>,
    ) -> Result<HttpResponse, S3Error> {
        let dataset = self.bucket_to_dataset(bucket)?;

        // Check if exists
        let info = self
            .provider
            .file_info(&dataset, key)
            .map_err(S3Error::InternalError)?
            .ok_or(S3Error::NoSuchKey)?;

        let (data, status) = if let Some((start, end)) = range {
            let data = self
                .provider
                .read_file_range(&dataset, key, start, end)
                .map_err(S3Error::InternalError)?;
            (data, 206)
        } else {
            let data = self
                .provider
                .read_file(&dataset, key)
                .map_err(S3Error::InternalError)?;
            (data, 200)
        };

        let etag = info
            .checksum
            .unwrap_or_else(|| alloc::format!("\"{}\"", info.mtime));

        let mut response = HttpResponse::new(status)
            .with_body(data)
            .with_header("ETag", etag)
            .with_header("Content-Length", info.size.to_string())
            .with_header("Last-Modified", xml::format_timestamp(info.mtime));

        Ok(response)
    }

    /// Delete object.
    pub fn delete_object(&mut self, bucket: &str, key: &str) -> Result<HttpResponse, S3Error> {
        let dataset = self.bucket_to_dataset(bucket)?;

        self.provider
            .delete_file(&dataset, key)
            .map_err(S3Error::InternalError)?;

        Ok(HttpResponse::no_content())
    }

    /// Head object.
    pub fn head_object(&self, bucket: &str, key: &str) -> Result<HttpResponse, S3Error> {
        let dataset = self.bucket_to_dataset(bucket)?;

        let info = self
            .provider
            .file_info(&dataset, key)
            .map_err(S3Error::InternalError)?
            .ok_or(S3Error::NoSuchKey)?;

        let etag = info
            .checksum
            .unwrap_or_else(|| alloc::format!("\"{}\"", info.mtime));

        Ok(HttpResponse::ok()
            .with_header("ETag", etag)
            .with_header("Content-Length", info.size.to_string())
            .with_header("Last-Modified", xml::format_timestamp(info.mtime)))
    }

    /// Copy object.
    pub fn copy_object(
        &mut self,
        bucket: &str,
        key: &str,
        source: &str,
    ) -> Result<HttpResponse, S3Error> {
        // Parse source: /bucket/key or bucket/key
        let source = source.trim_start_matches('/');
        let (src_bucket, src_key) = source
            .split_once('/')
            .ok_or_else(|| S3Error::InvalidArgument("Invalid copy source".into()))?;

        let src_dataset = self.bucket_to_dataset(src_bucket)?;
        let dst_dataset = self.bucket_to_dataset(bucket)?;

        self.provider
            .copy_file(&src_dataset, src_key, &dst_dataset, key)
            .map_err(S3Error::InternalError)?;

        let timestamp = xml::format_timestamp(self.provider.current_timestamp());
        let etag = "\"copyetag\""; // Would compute from copied file

        let xml = xml::copy_object_xml(etag, &timestamp);
        Ok(HttpResponse::ok().with_xml(xml))
    }

    /// Delete multiple objects.
    pub fn delete_objects(
        &mut self,
        bucket: &str,
        keys: &[String],
    ) -> Result<HttpResponse, S3Error> {
        let dataset = self.bucket_to_dataset(bucket)?;

        let mut deleted = Vec::new();
        let mut errors = Vec::new();

        for key in keys {
            match self.provider.delete_file(&dataset, key) {
                Ok(_) => deleted.push(key.clone()),
                Err(e) => errors.push((key.clone(), "InternalError".into(), e)),
            }
        }

        let xml = xml::delete_objects_xml(&deleted, &errors);
        Ok(HttpResponse::ok().with_xml(xml))
    }

    // ═══════════════════════════════════════════════════════════════════════════
    // MULTIPART UPLOAD
    // ═══════════════════════════════════════════════════════════════════════════

    /// Create multipart upload.
    pub fn create_multipart_upload(
        &mut self,
        bucket: &str,
        key: &str,
    ) -> Result<HttpResponse, S3Error> {
        self.bucket_to_dataset(bucket)?;

        let upload_id = alloc::format!("upload-{:016x}", self.next_upload_id);
        self.next_upload_id += 1;

        let upload = MultipartUpload {
            upload_id: upload_id.clone(),
            bucket: bucket.into(),
            key: key.into(),
            parts: BTreeMap::new(),
            initiated: self.provider.current_timestamp(),
        };

        self.multipart_uploads.insert(upload_id.clone(), upload);

        let xml = xml::initiate_multipart_xml(bucket, key, &upload_id);
        Ok(HttpResponse::ok().with_xml(xml))
    }

    /// Upload part.
    pub fn upload_part(
        &mut self,
        bucket: &str,
        key: &str,
        upload_id: &str,
        part_number: u32,
        data: &[u8],
    ) -> Result<HttpResponse, S3Error> {
        // Verify upload exists
        let upload = self
            .multipart_uploads
            .get_mut(upload_id)
            .ok_or(S3Error::NoSuchUpload)?;

        if upload.bucket != bucket || upload.key != key {
            return Err(S3Error::NoSuchUpload);
        }

        // Store part in temp storage
        let temp_key = alloc::format!("{}/{}", upload_id, part_number);
        self.provider
            .write_temp(&temp_key, data)
            .map_err(S3Error::InternalError)?;

        let etag = xml::compute_etag(data);

        // Record part
        upload.parts.insert(
            part_number,
            UploadPart {
                part_number,
                etag: etag.clone(),
                size: data.len() as u64,
                last_modified: self.provider.current_timestamp(),
            },
        );

        Ok(HttpResponse::ok().with_header("ETag", etag))
    }

    /// Complete multipart upload.
    pub fn complete_multipart_upload(
        &mut self,
        bucket: &str,
        key: &str,
        upload_id: &str,
        parts: &[(u32, String)],
    ) -> Result<HttpResponse, S3Error> {
        let dataset = self.bucket_to_dataset(bucket)?;

        // Get and remove upload
        let upload = self
            .multipart_uploads
            .remove(upload_id)
            .ok_or(S3Error::NoSuchUpload)?;

        if upload.bucket != bucket || upload.key != key {
            // Put it back
            self.multipart_uploads.insert(upload_id.into(), upload);
            return Err(S3Error::NoSuchUpload);
        }

        // Verify all parts exist and are in order
        let mut prev_num = 0;
        for (part_num, _etag) in parts {
            if *part_num <= prev_num {
                return Err(S3Error::InvalidPartOrder);
            }
            if !upload.parts.contains_key(part_num) {
                return Err(S3Error::InvalidPart);
            }
            prev_num = *part_num;
        }

        // Concatenate parts
        let mut combined = Vec::new();
        for (part_num, _) in parts {
            let temp_key = alloc::format!("{}/{}", upload_id, part_num);
            let part_data = self
                .provider
                .read_temp(&temp_key)
                .map_err(S3Error::InternalError)?;
            combined.extend_from_slice(&part_data);
        }

        // Write final object
        self.provider
            .write_file(&dataset, key, &combined)
            .map_err(S3Error::InternalError)?;

        // Clean up temp files
        for (part_num, _) in parts {
            let temp_key = alloc::format!("{}/{}", upload_id, part_num);
            let _ = self.provider.delete_temp(&temp_key);
        }

        let etag = xml::compute_etag(&combined);
        let location = alloc::format!("/{}/{}", bucket, key);
        let xml = xml::complete_multipart_xml(&location, bucket, key, &etag);

        Ok(HttpResponse::ok().with_xml(xml))
    }

    /// Abort multipart upload.
    pub fn abort_multipart_upload(
        &mut self,
        bucket: &str,
        key: &str,
        upload_id: &str,
    ) -> Result<HttpResponse, S3Error> {
        let upload = self
            .multipart_uploads
            .remove(upload_id)
            .ok_or(S3Error::NoSuchUpload)?;

        if upload.bucket != bucket || upload.key != key {
            self.multipart_uploads.insert(upload_id.into(), upload);
            return Err(S3Error::NoSuchUpload);
        }

        // Clean up temp files
        for part_num in upload.parts.keys() {
            let temp_key = alloc::format!("{}/{}", upload_id, part_num);
            let _ = self.provider.delete_temp(&temp_key);
        }

        Ok(HttpResponse::no_content())
    }

    /// List parts.
    pub fn list_parts(
        &self,
        bucket: &str,
        key: &str,
        upload_id: &str,
    ) -> Result<HttpResponse, S3Error> {
        let upload = self
            .multipart_uploads
            .get(upload_id)
            .ok_or(S3Error::NoSuchUpload)?;

        if upload.bucket != bucket || upload.key != key {
            return Err(S3Error::NoSuchUpload);
        }

        let parts: Vec<_> = upload.parts.values().cloned().collect();
        let xml = xml::list_parts_xml(bucket, key, upload_id, &parts, false, None);

        Ok(HttpResponse::ok().with_xml(xml))
    }

    /// List multipart uploads.
    pub fn list_multipart_uploads(&self, bucket: &str) -> Result<HttpResponse, S3Error> {
        self.bucket_to_dataset(bucket)?;

        let uploads: Vec<_> = self
            .multipart_uploads
            .values()
            .filter(|u| u.bucket == bucket)
            .cloned()
            .collect();

        let xml = xml::list_multipart_uploads_xml(bucket, &uploads, false);
        Ok(HttpResponse::ok().with_xml(xml))
    }

    // ═══════════════════════════════════════════════════════════════════════════
    // VERSIONING
    // ═══════════════════════════════════════════════════════════════════════════

    /// List object versions.
    pub fn list_object_versions(
        &self,
        bucket: &str,
        prefix: Option<&str>,
    ) -> Result<HttpResponse, S3Error> {
        let dataset = self.bucket_to_dataset(bucket)?;

        let versions = self
            .provider
            .list_versions(&dataset, prefix)
            .map_err(S3Error::InternalError)?;

        let s3_versions: Vec<S3ObjectVersion> = versions
            .into_iter()
            .map(|v| S3ObjectVersion {
                key: v.path,
                version_id: v.version_id,
                is_latest: v.is_latest,
                last_modified: xml::format_timestamp(v.mtime),
                etag: v.checksum,
                size: v.size,
                storage_class: "STANDARD".into(),
            })
            .collect();

        let xml = xml::list_object_versions_xml(bucket, prefix, &s3_versions, false);
        Ok(HttpResponse::ok().with_xml(xml))
    }
}

// ═══════════════════════════════════════════════════════════════════════════════
// VALIDATION
// ═══════════════════════════════════════════════════════════════════════════════

/// Validate bucket name according to S3 rules.
fn validate_bucket_name(name: &str) -> Result<(), S3Error> {
    if name.len() < 3 || name.len() > 63 {
        return Err(S3Error::InvalidBucketName);
    }

    // Must start with letter or number
    let first = name.chars().next().unwrap();
    if !first.is_ascii_lowercase() && !first.is_ascii_digit() {
        return Err(S3Error::InvalidBucketName);
    }

    // Must end with letter or number
    let last = name.chars().last().unwrap();
    if !last.is_ascii_lowercase() && !last.is_ascii_digit() {
        return Err(S3Error::InvalidBucketName);
    }

    // Can only contain lowercase letters, numbers, and hyphens
    for c in name.chars() {
        if !c.is_ascii_lowercase() && !c.is_ascii_digit() && c != '-' && c != '.' {
            return Err(S3Error::InvalidBucketName);
        }
    }

    // Cannot have consecutive periods
    if name.contains("..") {
        return Err(S3Error::InvalidBucketName);
    }

    // Cannot be formatted as IP address
    if name.chars().filter(|c| *c == '.').count() == 3
        && name.split('.').all(|p| p.parse::<u8>().is_ok())
    {
        return Err(S3Error::InvalidBucketName);
    }

    Ok(())
}

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

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

    #[test]
    fn test_validate_bucket_name() {
        // Valid names
        assert!(validate_bucket_name("mybucket").is_ok());
        assert!(validate_bucket_name("my-bucket").is_ok());
        assert!(validate_bucket_name("my.bucket").is_ok());
        assert!(validate_bucket_name("bucket123").is_ok());

        // Invalid names
        assert!(validate_bucket_name("ab").is_err()); // Too short
        assert!(validate_bucket_name("-bucket").is_err()); // Starts with hyphen
        assert!(validate_bucket_name("bucket-").is_err()); // Ends with hyphen
        assert!(validate_bucket_name("BUCKET").is_err()); // Uppercase
        assert!(validate_bucket_name("my..bucket").is_err()); // Consecutive periods
        assert!(validate_bucket_name("192.168.1.1").is_err()); // IP address
    }

    /// Mock storage provider for testing.
    struct MockStorage {
        datasets: BTreeMap<String, DatasetInfo>,
        files: BTreeMap<String, BTreeMap<String, (Vec<u8>, u64)>>,
        temp: BTreeMap<String, Vec<u8>>,
    }

    impl MockStorage {
        fn new() -> Self {
            Self {
                datasets: BTreeMap::new(),
                files: BTreeMap::new(),
                temp: BTreeMap::new(),
            }
        }
    }

    impl StorageProvider for MockStorage {
        fn list_datasets(&self) -> Result<Vec<DatasetInfo>, String> {
            Ok(self.datasets.values().cloned().collect())
        }

        fn create_dataset(&mut self, name: &str) -> Result<(), String> {
            self.datasets.insert(
                name.into(),
                DatasetInfo {
                    name: name.into(),
                    created: 0,
                },
            );
            self.files.insert(name.into(), BTreeMap::new());
            Ok(())
        }

        fn delete_dataset(&mut self, name: &str) -> Result<(), String> {
            self.datasets.remove(name);
            self.files.remove(name);
            Ok(())
        }

        fn dataset_exists(&self, name: &str) -> Result<bool, String> {
            Ok(self.datasets.contains_key(name))
        }

        fn list_files(&self, dataset: &str, prefix: Option<&str>) -> Result<Vec<FileInfo>, String> {
            let files = self.files.get(dataset).ok_or("no dataset")?;
            Ok(files
                .iter()
                .filter(|(k, _)| prefix.is_none() || k.starts_with(prefix.unwrap()))
                .map(|(k, (data, mtime))| FileInfo {
                    path: k.clone(),
                    size: data.len() as u64,
                    mtime: *mtime,
                    is_dir: false,
                    checksum: None,
                })
                .collect())
        }

        fn read_file(&self, dataset: &str, path: &str) -> Result<Vec<u8>, String> {
            self.files
                .get(dataset)
                .and_then(|f| f.get(path))
                .map(|(d, _)| d.clone())
                .ok_or("not found".into())
        }

        fn read_file_range(
            &self,
            dataset: &str,
            path: &str,
            start: u64,
            end: Option<u64>,
        ) -> Result<Vec<u8>, String> {
            let data = self.read_file(dataset, path)?;
            let end = end.unwrap_or(data.len() as u64) as usize;
            Ok(data[start as usize..end].to_vec())
        }

        fn write_file(&mut self, dataset: &str, path: &str, data: &[u8]) -> Result<(), String> {
            let files = self.files.get_mut(dataset).ok_or("no dataset")?;
            files.insert(path.into(), (data.to_vec(), 0));
            Ok(())
        }

        fn delete_file(&mut self, dataset: &str, path: &str) -> Result<(), String> {
            let files = self.files.get_mut(dataset).ok_or("no dataset")?;
            files.remove(path);
            Ok(())
        }

        fn file_exists(&self, dataset: &str, path: &str) -> Result<bool, String> {
            Ok(self
                .files
                .get(dataset)
                .map(|f| f.contains_key(path))
                .unwrap_or(false))
        }

        fn file_info(&self, dataset: &str, path: &str) -> Result<Option<FileInfo>, String> {
            Ok(self.files.get(dataset).and_then(|f| {
                f.get(path).map(|(data, mtime)| FileInfo {
                    path: path.into(),
                    size: data.len() as u64,
                    mtime: *mtime,
                    is_dir: false,
                    checksum: None,
                })
            }))
        }

        fn copy_file(
            &mut self,
            src_dataset: &str,
            src_path: &str,
            dst_dataset: &str,
            dst_path: &str,
        ) -> Result<(), String> {
            let data = self.read_file(src_dataset, src_path)?;
            self.write_file(dst_dataset, dst_path, &data)
        }

        fn write_temp(&mut self, key: &str, data: &[u8]) -> Result<(), String> {
            self.temp.insert(key.into(), data.to_vec());
            Ok(())
        }

        fn read_temp(&self, key: &str) -> Result<Vec<u8>, String> {
            self.temp.get(key).cloned().ok_or("not found".into())
        }

        fn delete_temp(&mut self, key: &str) -> Result<(), String> {
            self.temp.remove(key);
            Ok(())
        }

        fn list_temp(&self, prefix: &str) -> Result<Vec<String>, String> {
            Ok(self
                .temp
                .keys()
                .filter(|k| k.starts_with(prefix))
                .cloned()
                .collect())
        }

        fn current_timestamp(&self) -> u64 {
            0
        }
    }

    #[test]
    fn test_create_bucket() {
        let storage = MockStorage::new();
        let config = S3GatewayConfig::default();
        let mut ops = S3Ops::new(storage, config);

        let resp = ops.create_bucket("mybucket").unwrap();
        assert_eq!(resp.status, 200);
    }

    #[test]
    fn test_put_get_object() {
        let mut storage = MockStorage::new();
        storage.create_dataset("mybucket").unwrap();

        let mut config = S3GatewayConfig::default();
        config.map_bucket("mybucket", "mybucket");

        let mut ops = S3Ops::new(storage, config);

        // Put
        let resp = ops
            .put_object("mybucket", "test.txt", b"hello", None)
            .unwrap();
        assert_eq!(resp.status, 200);

        // Get
        let resp = ops.get_object("mybucket", "test.txt", None).unwrap();
        assert_eq!(resp.status, 200);
        assert_eq!(resp.body, b"hello");
    }

    #[test]
    fn test_multipart_upload() {
        let mut storage = MockStorage::new();
        storage.create_dataset("mybucket").unwrap();

        let mut config = S3GatewayConfig::default();
        config.map_bucket("mybucket", "mybucket");

        let mut ops = S3Ops::new(storage, config);

        // Initiate
        let resp = ops
            .create_multipart_upload("mybucket", "bigfile.bin")
            .unwrap();
        assert_eq!(resp.status, 200);
        let body = String::from_utf8(resp.body).unwrap();
        assert!(body.contains("<UploadId>"));

        // Extract upload ID (simplified)
        let upload_id = "upload-0000000000000001";

        // Upload parts
        ops.upload_part("mybucket", "bigfile.bin", upload_id, 1, b"part1")
            .unwrap();
        ops.upload_part("mybucket", "bigfile.bin", upload_id, 2, b"part2")
            .unwrap();

        // Complete
        let parts = vec![(1, "etag1".into()), (2, "etag2".into())];
        let resp = ops
            .complete_multipart_upload("mybucket", "bigfile.bin", upload_id, &parts)
            .unwrap();
        assert_eq!(resp.status, 200);

        // Verify file exists
        let resp = ops.get_object("mybucket", "bigfile.bin", None).unwrap();
        assert_eq!(resp.body, b"part1part2");
    }
}