post3 0.1.0

use std::collections::HashMap;
use std::path::{Path, PathBuf};

use bytes::Bytes;
use chrono::{DateTime, Utc};
use md5::{Digest, Md5};
use serde::{Deserialize, Serialize};
use tokio::fs;

use crate::backend::StorageBackend;
use crate::error::Post3Error;
use crate::models::{
    BucketInfo, CompleteMultipartUploadResult, CreateMultipartUploadResult, GetObjectResult,
    HeadObjectResult, ListMultipartUploadsResult, ListObjectsResult, ListPartsResult,
    MultipartUploadInfo, ObjectInfo, ObjectMeta, PutObjectResult, UploadPartInfo,
    UploadPartResult,
};

// --- On-disk metadata types ---

#[derive(Serialize, Deserialize)]
struct BucketMeta {
    created_at: DateTime<Utc>,
}

#[derive(Serialize, Deserialize)]
struct ObjectFileMeta {
    size: i64,
    etag: String,
    content_type: String,
    last_modified: DateTime<Utc>,
    user_metadata: HashMap<String, String>,
}

#[derive(Serialize, Deserialize)]
struct UploadFileMeta {
    key: String,
    content_type: String,
    created_at: DateTime<Utc>,
    user_metadata: HashMap<String, String>,
}

#[derive(Serialize, Deserialize)]
struct PartFileMeta {
    size: i64,
    etag: String,
    created_at: DateTime<Utc>,
}

// --- Key encoding ---

/// Characters that are safe in filenames (not encoded).
const SAFE_CHARS: &percent_encoding::AsciiSet = &percent_encoding::NON_ALPHANUMERIC
    .remove(b'-')
    .remove(b'_')
    .remove(b'.')
    .remove(b'~');

fn encode_key(key: &str) -> String {
    percent_encoding::utf8_percent_encode(key, SAFE_CHARS).to_string()
}

fn decode_key(encoded: &str) -> String {
    percent_encoding::percent_decode_str(encoded)
        .decode_utf8_lossy()
        .into_owned()
}

// --- Atomic write helper ---

async fn atomic_write(path: &Path, data: &[u8]) -> Result<(), Post3Error> {
    let tmp = path.with_extension("tmp");
    fs::write(&tmp, data).await?;
    fs::rename(&tmp, path).await?;
    Ok(())
}

// --- FilesystemBackend ---

/// Local filesystem storage backend. Useful for testing without PostgreSQL.
#[derive(Clone, Debug)]
pub struct FilesystemBackend {
    root: PathBuf,
}

impl FilesystemBackend {
    pub fn new(root: impl Into<PathBuf>) -> Self {
        Self { root: root.into() }
    }

    fn bucket_dir(&self, bucket: &str) -> PathBuf {
        self.root.join("buckets").join(bucket)
    }

    fn bucket_meta_path(&self, bucket: &str) -> PathBuf {
        self.bucket_dir(bucket).join(".bucket.json")
    }

    fn objects_dir(&self, bucket: &str) -> PathBuf {
        self.bucket_dir(bucket).join("objects")
    }

    fn object_dir(&self, bucket: &str, key: &str) -> PathBuf {
        self.objects_dir(bucket).join(encode_key(key))
    }

    fn multipart_base_dir(&self, bucket: &str) -> PathBuf {
        self.bucket_dir(bucket).join("multipart")
    }

    fn multipart_dir(&self, bucket: &str, upload_id: &str) -> PathBuf {
        self.multipart_base_dir(bucket).join(upload_id)
    }

    async fn require_bucket(&self, bucket: &str) -> Result<BucketMeta, Post3Error> {
        let meta_path = self.bucket_meta_path(bucket);
        if !meta_path.exists() {
            return Err(Post3Error::BucketNotFound(bucket.to_string()));
        }
        let data = fs::read(&meta_path).await?;
        serde_json::from_slice(&data).map_err(|e| Post3Error::Serialization(e.to_string()))
    }

    async fn require_upload(
        &self,
        bucket: &str,
        key: &str,
        upload_id: &str,
    ) -> Result<UploadFileMeta, Post3Error> {
        let upload_dir = self.multipart_dir(bucket, upload_id);
        let meta_path = upload_dir.join("upload.json");
        if !meta_path.exists() {
            return Err(Post3Error::UploadNotFound(upload_id.to_string()));
        }
        let data = fs::read(&meta_path).await?;
        let meta: UploadFileMeta =
            serde_json::from_slice(&data).map_err(|e| Post3Error::Serialization(e.to_string()))?;
        if meta.key != key {
            return Err(Post3Error::UploadNotFound(upload_id.to_string()));
        }
        Ok(meta)
    }
}

impl StorageBackend for FilesystemBackend {
    // --- Bucket operations ---

    async fn create_bucket(&self, name: &str) -> Result<BucketInfo, Post3Error> {
        let bucket_dir = self.bucket_dir(name);
        let meta_path = self.bucket_meta_path(name);

        if meta_path.exists() {
            return Err(Post3Error::BucketAlreadyExists(name.to_string()));
        }

        fs::create_dir_all(&bucket_dir).await?;
        fs::create_dir_all(self.objects_dir(name)).await?;

        let now = Utc::now();
        let meta = BucketMeta { created_at: now };
        let json = serde_json::to_vec(&meta).map_err(|e| Post3Error::Serialization(e.to_string()))?;
        atomic_write(&meta_path, &json).await?;

        Ok(BucketInfo {
            name: name.to_string(),
            created_at: now,
        })
    }

    async fn head_bucket(&self, name: &str) -> Result<Option<BucketInfo>, Post3Error> {
        let meta_path = self.bucket_meta_path(name);
        if !meta_path.exists() {
            return Ok(None);
        }
        let data = fs::read(&meta_path).await?;
        let meta: BucketMeta =
            serde_json::from_slice(&data).map_err(|e| Post3Error::Serialization(e.to_string()))?;
        Ok(Some(BucketInfo {
            name: name.to_string(),
            created_at: meta.created_at,
        }))
    }

    async fn delete_bucket(&self, name: &str) -> Result<(), Post3Error> {
        self.require_bucket(name).await?;

        // Check if bucket has objects
        let objects_dir = self.objects_dir(name);
        if objects_dir.exists() {
            let mut entries = fs::read_dir(&objects_dir).await?;
            if entries.next_entry().await?.is_some() {
                return Err(Post3Error::BucketNotEmpty(name.to_string()));
            }
        }

        fs::remove_dir_all(self.bucket_dir(name)).await?;
        Ok(())
    }

    async fn list_buckets(&self) -> Result<Vec<BucketInfo>, Post3Error> {
        let buckets_dir = self.root.join("buckets");
        if !buckets_dir.exists() {
            return Ok(Vec::new());
        }

        let mut buckets = Vec::new();
        let mut entries = fs::read_dir(&buckets_dir).await?;
        while let Some(entry) = entries.next_entry().await? {
            if entry.file_type().await?.is_dir() {
                let name = entry.file_name().to_string_lossy().to_string();
                let meta_path = entry.path().join(".bucket.json");
                if meta_path.exists() {
                    let data = fs::read(&meta_path).await?;
                    if let Ok(meta) = serde_json::from_slice::<BucketMeta>(&data) {
                        buckets.push(BucketInfo {
                            name,
                            created_at: meta.created_at,
                        });
                    }
                }
            }
        }
        buckets.sort_by(|a, b| a.name.cmp(&b.name));
        Ok(buckets)
    }

    // --- Object operations ---

    async fn put_object(
        &self,
        bucket: &str,
        key: &str,
        content_type: Option<&str>,
        metadata: HashMap<String, String>,
        body: Bytes,
    ) -> Result<PutObjectResult, Post3Error> {
        self.require_bucket(bucket).await?;
        let content_type = content_type.unwrap_or("application/octet-stream");

        let mut hasher = Md5::new();
        hasher.update(&body);
        let etag = format!("\"{}\"", hex::encode(hasher.finalize()));
        let size = body.len() as i64;

        let obj_dir = self.object_dir(bucket, key);
        fs::create_dir_all(&obj_dir).await?;

        // Write data
        atomic_write(&obj_dir.join("data"), &body).await?;

        // Write metadata
        let meta = ObjectFileMeta {
            size,
            etag: etag.clone(),
            content_type: content_type.to_string(),
            last_modified: Utc::now(),
            user_metadata: metadata,
        };
        let json =
            serde_json::to_vec(&meta).map_err(|e| Post3Error::Serialization(e.to_string()))?;
        atomic_write(&obj_dir.join("meta.json"), &json).await?;

        Ok(PutObjectResult { etag, size })
    }

    async fn get_object(
        &self,
        bucket: &str,
        key: &str,
    ) -> Result<GetObjectResult, Post3Error> {
        self.require_bucket(bucket).await?;
        let obj_dir = self.object_dir(bucket, key);
        let meta_path = obj_dir.join("meta.json");

        if !meta_path.exists() {
            return Err(Post3Error::ObjectNotFound {
                bucket: bucket.to_string(),
                key: key.to_string(),
            });
        }

        let meta_data = fs::read(&meta_path).await?;
        let meta: ObjectFileMeta = serde_json::from_slice(&meta_data)
            .map_err(|e| Post3Error::Serialization(e.to_string()))?;

        let body = fs::read(obj_dir.join("data")).await?;

        Ok(GetObjectResult {
            metadata: ObjectMeta {
                key: key.to_string(),
                size: meta.size,
                etag: meta.etag,
                content_type: meta.content_type,
                last_modified: meta.last_modified,
            },
            user_metadata: meta.user_metadata,
            body: Bytes::from(body),
        })
    }

    async fn head_object(
        &self,
        bucket: &str,
        key: &str,
    ) -> Result<Option<HeadObjectResult>, Post3Error> {
        self.require_bucket(bucket).await?;
        let obj_dir = self.object_dir(bucket, key);
        let meta_path = obj_dir.join("meta.json");

        if !meta_path.exists() {
            return Ok(None);
        }

        let meta_data = fs::read(&meta_path).await?;
        let meta: ObjectFileMeta = serde_json::from_slice(&meta_data)
            .map_err(|e| Post3Error::Serialization(e.to_string()))?;

        Ok(Some(HeadObjectResult {
            object: ObjectMeta {
                key: key.to_string(),
                size: meta.size,
                etag: meta.etag,
                content_type: meta.content_type,
                last_modified: meta.last_modified,
            },
            user_metadata: meta.user_metadata,
        }))
    }

    async fn delete_object(
        &self,
        bucket: &str,
        key: &str,
    ) -> Result<(), Post3Error> {
        self.require_bucket(bucket).await?;
        let obj_dir = self.object_dir(bucket, key);
        if obj_dir.exists() {
            fs::remove_dir_all(&obj_dir).await?;
        }
        Ok(())
    }

    async fn list_objects_v2(
        &self,
        bucket: &str,
        prefix: Option<&str>,
        continuation_token: Option<&str>,
        max_keys: Option<i64>,
        delimiter: Option<&str>,
    ) -> Result<ListObjectsResult, Post3Error> {
        self.require_bucket(bucket).await?;
        let objects_dir = self.objects_dir(bucket);
        let max_keys = max_keys.unwrap_or(1000);

        // MaxKeys=0 is valid: return empty result
        if max_keys == 0 {
            return Ok(ListObjectsResult {
                objects: Vec::new(),
                is_truncated: false,
                next_continuation_token: None,
                prefix: prefix.map(|s| s.to_string()),
                delimiter: delimiter.map(|s| s.to_string()),
                common_prefixes: Vec::new(),
                key_count: 0,
            });
        }

        let mut all_objects = Vec::new();
        if objects_dir.exists() {
            let mut entries = fs::read_dir(&objects_dir).await?;
            while let Some(entry) = entries.next_entry().await? {
                if !entry.file_type().await?.is_dir() {
                    continue;
                }
                let encoded = entry.file_name().to_string_lossy().to_string();
                let key = decode_key(&encoded);

                // Filter by prefix
                if let Some(pfx) = prefix {
                    if !key.starts_with(pfx) {
                        continue;
                    }
                }

                // Filter by continuation token (key must be > token)
                if let Some(token) = continuation_token {
                    if key.as_str() <= token {
                        continue;
                    }
                }

                let meta_path = entry.path().join("meta.json");
                if !meta_path.exists() {
                    continue;
                }

                let meta_data = fs::read(&meta_path).await?;
                if let Ok(meta) = serde_json::from_slice::<ObjectFileMeta>(&meta_data) {
                    all_objects.push(ObjectInfo {
                        key,
                        size: meta.size,
                        etag: meta.etag,
                        last_modified: meta.last_modified,
                    });
                }
            }
        }

        // Sort by key
        all_objects.sort_by(|a, b| a.key.cmp(&b.key));

        // Apply delimiter grouping and MaxKeys-aware interleaving.
        // In S3, MaxKeys limits the TOTAL count of objects + common_prefixes.
        // They are interleaved in sorted order (objects by key, prefixes by prefix string).
        let prefix_str = prefix.unwrap_or("");
        if let Some(delim) = delimiter {
            // Separate into direct objects and rolled-up common prefixes
            let mut seen_prefixes = std::collections::BTreeSet::new();
            let mut direct_objects = Vec::new();
            for obj in &all_objects {
                let after_prefix = &obj.key[prefix_str.len()..];
                if let Some(pos) = after_prefix.find(delim) {
                    let cp = format!("{}{}", prefix_str, &after_prefix[..pos + delim.len()]);
                    seen_prefixes.insert(cp);
                } else {
                    direct_objects.push(obj.clone());
                }
            }
            // Filter out common prefixes that are <= continuation token
            let all_prefixes: Vec<String> = if let Some(token) = continuation_token {
                seen_prefixes
                    .into_iter()
                    .filter(|cp| cp.as_str() > token)
                    .collect()
            } else {
                seen_prefixes.into_iter().collect()
            };

            // Merge objects and common_prefixes in sorted order, limited to max_keys
            let mut result_objects = Vec::new();
            let mut result_prefixes = Vec::new();
            let mut oi = 0usize;
            let mut pi = 0usize;
            let mut count = 0i64;
            let mut last_key: Option<String> = None;

            while count < max_keys && (oi < direct_objects.len() || pi < all_prefixes.len()) {
                let take_object = match (direct_objects.get(oi), all_prefixes.get(pi)) {
                    (Some(obj), Some(pfx)) => obj.key.as_str() < pfx.as_str(),
                    (Some(_), None) => true,
                    (None, Some(_)) => false,
                    (None, None) => break,
                };

                if take_object {
                    last_key = Some(direct_objects[oi].key.clone());
                    result_objects.push(direct_objects[oi].clone());
                    oi += 1;
                } else {
                    last_key = Some(all_prefixes[pi].clone());
                    result_prefixes.push(all_prefixes[pi].clone());
                    pi += 1;
                }
                count += 1;
            }

            let is_truncated = oi < direct_objects.len() || pi < all_prefixes.len();
            let next_token = if is_truncated { last_key } else { None };
            let key_count = result_objects.len() + result_prefixes.len();

            Ok(ListObjectsResult {
                objects: result_objects,
                is_truncated,
                next_continuation_token: next_token,
                prefix: prefix.map(|s| s.to_string()),
                delimiter: Some(delim.to_string()),
                common_prefixes: result_prefixes,
                key_count,
            })
        } else {
            // No delimiter: simple pagination by objects only
            let is_truncated = all_objects.len() as i64 > max_keys;
            let items: Vec<_> = all_objects.into_iter().take(max_keys as usize).collect();
            let next_token = if is_truncated {
                items.last().map(|o| o.key.clone())
            } else {
                None
            };
            let key_count = items.len();

            Ok(ListObjectsResult {
                objects: items,
                is_truncated,
                next_continuation_token: next_token,
                prefix: prefix.map(|s| s.to_string()),
                delimiter: None,
                common_prefixes: Vec::new(),
                key_count,
            })
        }
    }

    // --- Multipart upload operations ---

    async fn create_multipart_upload(
        &self,
        bucket: &str,
        key: &str,
        content_type: Option<&str>,
        metadata: HashMap<String, String>,
    ) -> Result<CreateMultipartUploadResult, Post3Error> {
        self.require_bucket(bucket).await?;
        let content_type = content_type.unwrap_or("application/octet-stream");
        let upload_id = uuid::Uuid::new_v4().to_string();

        let upload_dir = self.multipart_dir(bucket, &upload_id);
        fs::create_dir_all(upload_dir.join("parts")).await?;

        let meta = UploadFileMeta {
            key: key.to_string(),
            content_type: content_type.to_string(),
            created_at: Utc::now(),
            user_metadata: metadata,
        };
        let json =
            serde_json::to_vec(&meta).map_err(|e| Post3Error::Serialization(e.to_string()))?;
        atomic_write(&upload_dir.join("upload.json"), &json).await?;

        Ok(CreateMultipartUploadResult {
            bucket: bucket.to_string(),
            key: key.to_string(),
            upload_id,
        })
    }

    async fn upload_part(
        &self,
        bucket: &str,
        key: &str,
        upload_id: &str,
        part_number: i32,
        body: Bytes,
    ) -> Result<UploadPartResult, Post3Error> {
        self.require_bucket(bucket).await?;
        self.require_upload(bucket, key, upload_id).await?;

        let mut hasher = Md5::new();
        hasher.update(&body);
        let etag = format!("\"{}\"", hex::encode(hasher.finalize()));
        let size = body.len() as i64;

        let parts_dir = self.multipart_dir(bucket, upload_id).join("parts");

        // Write part data
        atomic_write(&parts_dir.join(part_number.to_string()), &body).await?;

        // Write part metadata
        let part_meta = PartFileMeta {
            size,
            etag: etag.clone(),
            created_at: Utc::now(),
        };
        let json = serde_json::to_vec(&part_meta)
            .map_err(|e| Post3Error::Serialization(e.to_string()))?;
        atomic_write(
            &parts_dir.join(format!("{}.meta", part_number)),
            &json,
        )
        .await?;

        Ok(UploadPartResult { etag })
    }

    async fn complete_multipart_upload(
        &self,
        bucket: &str,
        key: &str,
        upload_id: &str,
        part_etags: Vec<(i32, String)>,
    ) -> Result<CompleteMultipartUploadResult, Post3Error> {
        self.require_bucket(bucket).await?;
        let upload_meta = self.require_upload(bucket, key, upload_id).await?;

        // Validate part numbers are in ascending order
        for window in part_etags.windows(2) {
            if window[0].0 >= window[1].0 {
                return Err(Post3Error::InvalidPartOrder);
            }
        }

        let parts_dir = self.multipart_dir(bucket, upload_id).join("parts");

        // Load and validate parts
        let mut assembled = Vec::new();
        let mut etag_hasher = Md5::new();
        let part_count = part_etags.len();

        for (expected_num, expected_etag) in &part_etags {
            // Read part metadata
            let meta_path = parts_dir.join(format!("{}.meta", expected_num));
            if !meta_path.exists() {
                return Err(Post3Error::InvalidPart {
                    upload_id: upload_id.to_string(),
                    part_number: *expected_num,
                });
            }

            let meta_data = fs::read(&meta_path).await?;
            let part_meta: PartFileMeta = serde_json::from_slice(&meta_data)
                .map_err(|e| Post3Error::Serialization(e.to_string()))?;

            // Normalize ETags by stripping quotes for comparison
            // Stored ETags have quotes ("abc"), clients may send with or without
            let stored = part_meta.etag.trim_matches('"');
            let expected = expected_etag.trim_matches('"');
            if stored != expected {
                return Err(Post3Error::ETagMismatch {
                    part_number: *expected_num,
                    expected: expected_etag.clone(),
                    got: part_meta.etag,
                });
            }

            // Read part data
            let data_path = parts_dir.join(expected_num.to_string());
            let data = fs::read(&data_path).await?;

            // Validate minimum part size (5 MB) for all parts except the last
            const MIN_PART_SIZE: i64 = 5 * 1024 * 1024;
            let is_last = *expected_num == part_etags.last().unwrap().0;
            if !is_last && (data.len() as i64) < MIN_PART_SIZE {
                return Err(Post3Error::EntityTooSmall {
                    part_number: *expected_num,
                    size: data.len() as i64,
                });
            }

            assembled.extend_from_slice(&data);

            // Feed into compound ETag
            let hex_str = part_meta.etag.trim_matches('"');
            if let Ok(raw_md5) = hex::decode(hex_str) {
                etag_hasher.update(&raw_md5);
            }
        }

        let compound_etag = format!(
            "\"{}-{}\"",
            hex::encode(etag_hasher.finalize()),
            part_count
        );
        let total_size = assembled.len() as i64;

        // Write final object
        let obj_dir = self.object_dir(bucket, key);
        fs::create_dir_all(&obj_dir).await?;
        atomic_write(&obj_dir.join("data"), &assembled).await?;

        let meta = ObjectFileMeta {
            size: total_size,
            etag: compound_etag.clone(),
            content_type: upload_meta.content_type,
            last_modified: Utc::now(),
            user_metadata: upload_meta.user_metadata,
        };
        let json =
            serde_json::to_vec(&meta).map_err(|e| Post3Error::Serialization(e.to_string()))?;
        atomic_write(&obj_dir.join("meta.json"), &json).await?;

        // Clean up multipart upload
        fs::remove_dir_all(self.multipart_dir(bucket, upload_id)).await?;

        Ok(CompleteMultipartUploadResult {
            bucket: bucket.to_string(),
            key: key.to_string(),
            etag: compound_etag,
            size: total_size,
        })
    }

    async fn abort_multipart_upload(
        &self,
        bucket: &str,
        key: &str,
        upload_id: &str,
    ) -> Result<(), Post3Error> {
        self.require_bucket(bucket).await?;
        self.require_upload(bucket, key, upload_id).await?;

        let upload_dir = self.multipart_dir(bucket, upload_id);
        if upload_dir.exists() {
            fs::remove_dir_all(&upload_dir).await?;
        }
        Ok(())
    }

    async fn list_parts(
        &self,
        bucket: &str,
        key: &str,
        upload_id: &str,
        max_parts: Option<i32>,
        part_number_marker: Option<i32>,
    ) -> Result<ListPartsResult, Post3Error> {
        self.require_bucket(bucket).await?;
        self.require_upload(bucket, key, upload_id).await?;

        let parts_dir = self.multipart_dir(bucket, upload_id).join("parts");
        let max_parts = max_parts.unwrap_or(1000) as usize;

        let mut all_parts = Vec::new();
        if parts_dir.exists() {
            let mut entries = fs::read_dir(&parts_dir).await?;
            while let Some(entry) = entries.next_entry().await? {
                let name = entry.file_name().to_string_lossy().to_string();
                // Only read .meta files
                if !name.ends_with(".meta") {
                    continue;
                }
                let part_num_str = name.trim_end_matches(".meta");
                let part_number: i32 = match part_num_str.parse() {
                    Ok(n) => n,
                    Err(_) => continue,
                };

                // Filter by marker
                if let Some(marker) = part_number_marker {
                    if part_number <= marker {
                        continue;
                    }
                }

                let data = fs::read(entry.path()).await?;
                if let Ok(meta) = serde_json::from_slice::<PartFileMeta>(&data) {
                    all_parts.push(UploadPartInfo {
                        part_number,
                        size: meta.size,
                        etag: meta.etag,
                        created_at: meta.created_at,
                    });
                }
            }
        }

        all_parts.sort_by_key(|p| p.part_number);

        let is_truncated = all_parts.len() > max_parts;
        let items: Vec<_> = all_parts.into_iter().take(max_parts).collect();
        let next_marker = if is_truncated {
            items.last().map(|p| p.part_number)
        } else {
            None
        };

        Ok(ListPartsResult {
            bucket: bucket.to_string(),
            key: key.to_string(),
            upload_id: upload_id.to_string(),
            parts: items,
            is_truncated,
            next_part_number_marker: next_marker,
        })
    }

    async fn list_multipart_uploads(
        &self,
        bucket: &str,
        prefix: Option<&str>,
        key_marker: Option<&str>,
        upload_id_marker: Option<&str>,
        max_uploads: Option<i32>,
    ) -> Result<ListMultipartUploadsResult, Post3Error> {
        self.require_bucket(bucket).await?;

        let mp_dir = self.multipart_base_dir(bucket);
        let max_uploads = max_uploads.unwrap_or(1000) as usize;

        let mut all_uploads = Vec::new();
        if mp_dir.exists() {
            let mut entries = fs::read_dir(&mp_dir).await?;
            while let Some(entry) = entries.next_entry().await? {
                if !entry.file_type().await?.is_dir() {
                    continue;
                }
                let upload_id = entry.file_name().to_string_lossy().to_string();
                let meta_path = entry.path().join("upload.json");
                if !meta_path.exists() {
                    continue;
                }
                let data = fs::read(&meta_path).await?;
                let meta: UploadFileMeta = match serde_json::from_slice(&data) {
                    Ok(m) => m,
                    Err(_) => continue,
                };

                // Filter by prefix
                if let Some(pfx) = prefix {
                    if !meta.key.starts_with(pfx) {
                        continue;
                    }
                }

                // Filter by markers
                if let Some(km) = key_marker {
                    if meta.key.as_str() < km {
                        continue;
                    }
                    if meta.key.as_str() == km {
                        if let Some(um) = upload_id_marker {
                            if upload_id.as_str() <= um {
                                continue;
                            }
                        }
                    }
                }

                all_uploads.push(MultipartUploadInfo {
                    key: meta.key,
                    upload_id,
                    initiated: meta.created_at,
                });
            }
        }

        all_uploads.sort_by(|a, b| (&a.key, &a.upload_id).cmp(&(&b.key, &b.upload_id)));

        let is_truncated = all_uploads.len() > max_uploads;
        let items: Vec<_> = all_uploads.into_iter().take(max_uploads).collect();
        let (next_key_marker, next_upload_id_marker) = if is_truncated {
            items
                .last()
                .map(|u| (Some(u.key.clone()), Some(u.upload_id.clone())))
                .unwrap_or((None, None))
        } else {
            (None, None)
        };

        Ok(ListMultipartUploadsResult {
            bucket: bucket.to_string(),
            uploads: items,
            is_truncated,
            next_key_marker,
            next_upload_id_marker,
            prefix: prefix.map(|s| s.to_string()),
        })
    }
}

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

    async fn temp_backend() -> (FilesystemBackend, tempfile::TempDir) {
        let dir = tempfile::tempdir().unwrap();
        let backend = FilesystemBackend::new(dir.path());
        (backend, dir)
    }

    // =========================================================================
    // Bucket operations
    // =========================================================================

    #[tokio::test]
    async fn test_bucket_crud() {
        let (backend, _dir) = temp_backend().await;

        // Create
        let info = backend.create_bucket("test-bucket").await.unwrap();
        assert_eq!(info.name, "test-bucket");

        // Head
        let head = backend.head_bucket("test-bucket").await.unwrap();
        assert!(head.is_some());
        assert_eq!(head.unwrap().name, "test-bucket");

        // List
        let list = backend.list_buckets().await.unwrap();
        assert_eq!(list.len(), 1);
        assert_eq!(list[0].name, "test-bucket");

        // Delete
        backend.delete_bucket("test-bucket").await.unwrap();
        let head = backend.head_bucket("test-bucket").await.unwrap();
        assert!(head.is_none());
    }

    #[tokio::test]
    async fn test_bucket_duplicate_create() {
        let (backend, _dir) = temp_backend().await;
        backend.create_bucket("dup").await.unwrap();

        let err = backend.create_bucket("dup").await.unwrap_err();
        assert!(
            matches!(err, Post3Error::BucketAlreadyExists(ref n) if n == "dup"),
            "Expected BucketAlreadyExists, got: {err:?}"
        );
    }

    #[tokio::test]
    async fn test_head_nonexistent_bucket() {
        let (backend, _dir) = temp_backend().await;
        let head = backend.head_bucket("no-such-bucket").await.unwrap();
        assert!(head.is_none());
    }

    #[tokio::test]
    async fn test_delete_nonexistent_bucket() {
        let (backend, _dir) = temp_backend().await;
        let err = backend.delete_bucket("no-such-bucket").await.unwrap_err();
        assert!(matches!(err, Post3Error::BucketNotFound(_)));
    }

    #[tokio::test]
    async fn test_delete_non_empty_bucket() {
        let (backend, _dir) = temp_backend().await;
        backend.create_bucket("test").await.unwrap();
        backend
            .put_object("test", "file.txt", None, HashMap::new(), Bytes::from("x"))
            .await
            .unwrap();

        let err = backend.delete_bucket("test").await.unwrap_err();
        assert!(
            matches!(err, Post3Error::BucketNotEmpty(_)),
            "Expected BucketNotEmpty, got: {err:?}"
        );
    }

    #[tokio::test]
    async fn test_list_buckets_empty() {
        let (backend, _dir) = temp_backend().await;
        let list = backend.list_buckets().await.unwrap();
        assert!(list.is_empty());
    }

    #[tokio::test]
    async fn test_list_buckets_sorted() {
        let (backend, _dir) = temp_backend().await;
        backend.create_bucket("charlie").await.unwrap();
        backend.create_bucket("alpha").await.unwrap();
        backend.create_bucket("bravo").await.unwrap();

        let list = backend.list_buckets().await.unwrap();
        let names: Vec<_> = list.iter().map(|b| b.name.as_str()).collect();
        assert_eq!(names, vec!["alpha", "bravo", "charlie"]);
    }

    #[tokio::test]
    async fn test_bucket_created_at_preserved() {
        let (backend, _dir) = temp_backend().await;
        let info = backend.create_bucket("ts-test").await.unwrap();
        let head = backend.head_bucket("ts-test").await.unwrap().unwrap();
        assert_eq!(info.created_at, head.created_at);

        let list = backend.list_buckets().await.unwrap();
        assert_eq!(list[0].created_at, info.created_at);
    }

    // =========================================================================
    // Object operations — basic
    // =========================================================================

    #[tokio::test]
    async fn test_object_crud() {
        let (backend, _dir) = temp_backend().await;
        backend.create_bucket("test").await.unwrap();

        // Put
        let put = backend
            .put_object(
                "test",
                "hello.txt",
                Some("text/plain"),
                HashMap::new(),
                Bytes::from("hello world"),
            )
            .await
            .unwrap();
        assert!(put.etag.starts_with('"'));
        assert!(put.etag.ends_with('"'));
        assert_eq!(put.size, 11);

        // Get
        let get = backend.get_object("test", "hello.txt").await.unwrap();
        assert_eq!(get.body.as_ref(), b"hello world");
        assert_eq!(get.metadata.content_type, "text/plain");
        assert_eq!(get.metadata.size, 11);
        assert_eq!(get.metadata.key, "hello.txt");
        assert_eq!(get.metadata.etag, put.etag);

        // Head
        let head = backend.head_object("test", "hello.txt").await.unwrap();
        assert!(head.is_some());
        let h = head.unwrap();
        assert_eq!(h.object.size, 11);
        assert_eq!(h.object.content_type, "text/plain");
        assert_eq!(h.object.etag, put.etag);
        assert_eq!(h.object.key, "hello.txt");

        // Delete
        backend.delete_object("test", "hello.txt").await.unwrap();
        let head = backend.head_object("test", "hello.txt").await.unwrap();
        assert!(head.is_none());
    }

    #[tokio::test]
    async fn test_put_empty_body() {
        let (backend, _dir) = temp_backend().await;
        backend.create_bucket("test").await.unwrap();

        let put = backend
            .put_object("test", "empty", None, HashMap::new(), Bytes::new())
            .await
            .unwrap();
        assert_eq!(put.size, 0);

        let get = backend.get_object("test", "empty").await.unwrap();
        assert!(get.body.is_empty());
        assert_eq!(get.metadata.size, 0);
    }

    #[tokio::test]
    async fn test_default_content_type() {
        let (backend, _dir) = temp_backend().await;
        backend.create_bucket("test").await.unwrap();

        backend
            .put_object("test", "blob", None, HashMap::new(), Bytes::from("data"))
            .await
            .unwrap();

        let get = backend.get_object("test", "blob").await.unwrap();
        assert_eq!(get.metadata.content_type, "application/octet-stream");
    }

    #[tokio::test]
    async fn test_overwrite_object() {
        let (backend, _dir) = temp_backend().await;
        backend.create_bucket("test").await.unwrap();

        backend
            .put_object(
                "test",
                "key",
                Some("text/plain"),
                HashMap::new(),
                Bytes::from("version1"),
            )
            .await
            .unwrap();

        let put2 = backend
            .put_object(
                "test",
                "key",
                Some("application/json"),
                HashMap::new(),
                Bytes::from("version2"),
            )
            .await
            .unwrap();

        let get = backend.get_object("test", "key").await.unwrap();
        assert_eq!(get.body.as_ref(), b"version2");
        assert_eq!(get.metadata.content_type, "application/json");
        assert_eq!(get.metadata.etag, put2.etag);
        assert_eq!(get.metadata.size, 8);
    }

    #[tokio::test]
    async fn test_overwrite_clears_old_metadata() {
        let (backend, _dir) = temp_backend().await;
        backend.create_bucket("test").await.unwrap();

        let mut meta1 = HashMap::new();
        meta1.insert("version".to_string(), "1".to_string());
        meta1.insert("author".to_string(), "alice".to_string());
        backend
            .put_object("test", "key", None, meta1, Bytes::from("v1"))
            .await
            .unwrap();

        // Overwrite with different metadata
        let mut meta2 = HashMap::new();
        meta2.insert("version".to_string(), "2".to_string());
        backend
            .put_object("test", "key", None, meta2, Bytes::from("v2"))
            .await
            .unwrap();

        let get = backend.get_object("test", "key").await.unwrap();
        assert_eq!(get.user_metadata.get("version").unwrap(), "2");
        // "author" should be gone
        assert!(get.user_metadata.get("author").is_none());
    }

    #[tokio::test]
    async fn test_get_nonexistent_object() {
        let (backend, _dir) = temp_backend().await;
        backend.create_bucket("test").await.unwrap();

        let err = backend.get_object("test", "missing").await.unwrap_err();
        assert!(matches!(err, Post3Error::ObjectNotFound { .. }));
    }

    #[tokio::test]
    async fn test_head_nonexistent_object() {
        let (backend, _dir) = temp_backend().await;
        backend.create_bucket("test").await.unwrap();

        let head = backend.head_object("test", "missing").await.unwrap();
        assert!(head.is_none());
    }

    #[tokio::test]
    async fn test_delete_nonexistent_object_is_idempotent() {
        let (backend, _dir) = temp_backend().await;
        backend.create_bucket("test").await.unwrap();

        // S3 spec: DELETE on nonexistent key returns 204, no error
        backend.delete_object("test", "nope").await.unwrap();
    }

    #[tokio::test]
    async fn test_operations_on_nonexistent_bucket() {
        let (backend, _dir) = temp_backend().await;

        let err = backend
            .put_object("nope", "key", None, HashMap::new(), Bytes::from("x"))
            .await
            .unwrap_err();
        assert!(matches!(err, Post3Error::BucketNotFound(_)));

        let err = backend.get_object("nope", "key").await.unwrap_err();
        assert!(matches!(err, Post3Error::BucketNotFound(_)));

        let err = backend.head_object("nope", "key").await.unwrap_err();
        assert!(matches!(err, Post3Error::BucketNotFound(_)));

        let err = backend.delete_object("nope", "key").await.unwrap_err();
        assert!(matches!(err, Post3Error::BucketNotFound(_)));

        let err = backend
            .list_objects_v2("nope", None, None, None, None)
            .await
            .unwrap_err();
        assert!(matches!(err, Post3Error::BucketNotFound(_)));
    }

    // =========================================================================
    // Object keys — special characters and paths
    // =========================================================================

    #[tokio::test]
    async fn test_key_with_slashes() {
        let (backend, _dir) = temp_backend().await;
        backend.create_bucket("test").await.unwrap();

        let key = "path/to/nested/file.txt";
        backend
            .put_object("test", key, None, HashMap::new(), Bytes::from("nested"))
            .await
            .unwrap();

        let get = backend.get_object("test", key).await.unwrap();
        assert_eq!(get.body.as_ref(), b"nested");
        assert_eq!(get.metadata.key, key);
    }

    #[tokio::test]
    async fn test_key_with_special_characters() {
        let (backend, _dir) = temp_backend().await;
        backend.create_bucket("test").await.unwrap();

        let keys = [
            "hello world.txt",       // space
            "file (1).txt",          // parentheses
            "data=value&other=1",    // query-like
            "日本語/テスト.txt",     // unicode
            "100%.txt",              // percent sign
            "a+b.txt",              // plus sign
            "file#anchor",           // hash
            "path/with spaces/f.txt", // spaces in path
        ];

        for key in &keys {
            backend
                .put_object("test", key, None, HashMap::new(), Bytes::from(*key))
                .await
                .unwrap();

            let get = backend.get_object("test", key).await.unwrap();
            assert_eq!(get.body.as_ref(), key.as_bytes(), "Round-trip failed for key: {key}");
            assert_eq!(&get.metadata.key, key);
        }

        // All keys should be listable
        let list = backend
            .list_objects_v2("test", None, None, None, None)
            .await
            .unwrap();
        assert_eq!(list.objects.len(), keys.len());
    }

    #[tokio::test]
    async fn test_key_with_leading_slash() {
        let (backend, _dir) = temp_backend().await;
        backend.create_bucket("test").await.unwrap();

        // S3 allows keys starting with /
        backend
            .put_object("test", "/leading", None, HashMap::new(), Bytes::from("data"))
            .await
            .unwrap();

        let get = backend.get_object("test", "/leading").await.unwrap();
        assert_eq!(get.metadata.key, "/leading");
    }

    #[tokio::test]
    async fn test_key_with_dots() {
        let (backend, _dir) = temp_backend().await;
        backend.create_bucket("test").await.unwrap();

        // Ensure . and .. in keys don't cause path traversal
        let keys = ["./relative", "../parent", "a/../b", "..."];
        for key in &keys {
            backend
                .put_object("test", key, None, HashMap::new(), Bytes::from("safe"))
                .await
                .unwrap();

            let get = backend.get_object("test", key).await.unwrap();
            assert_eq!(get.body.as_ref(), b"safe");
            assert_eq!(&get.metadata.key, key);
        }
    }

    // =========================================================================
    // Object metadata
    // =========================================================================

    #[tokio::test]
    async fn test_object_metadata_roundtrip() {
        let (backend, _dir) = temp_backend().await;
        backend.create_bucket("test").await.unwrap();

        let mut meta = HashMap::new();
        meta.insert("author".to_string(), "test-user".to_string());
        meta.insert("version".to_string(), "42".to_string());
        meta.insert("empty-value".to_string(), String::new());

        backend
            .put_object("test", "doc.txt", None, meta.clone(), Bytes::from("content"))
            .await
            .unwrap();

        // Via get_object
        let get = backend.get_object("test", "doc.txt").await.unwrap();
        assert_eq!(get.user_metadata.len(), 3);
        assert_eq!(get.user_metadata.get("author").unwrap(), "test-user");
        assert_eq!(get.user_metadata.get("version").unwrap(), "42");
        assert_eq!(get.user_metadata.get("empty-value").unwrap(), "");

        // Via head_object
        let head = backend.head_object("test", "doc.txt").await.unwrap().unwrap();
        assert_eq!(head.user_metadata, get.user_metadata);
    }

    #[tokio::test]
    async fn test_object_no_metadata() {
        let (backend, _dir) = temp_backend().await;
        backend.create_bucket("test").await.unwrap();

        backend
            .put_object("test", "bare", None, HashMap::new(), Bytes::from("data"))
            .await
            .unwrap();

        let get = backend.get_object("test", "bare").await.unwrap();
        assert!(get.user_metadata.is_empty());
    }

    // =========================================================================
    // ETag correctness
    // =========================================================================

    #[tokio::test]
    async fn test_etag_is_md5() {
        let (backend, _dir) = temp_backend().await;
        backend.create_bucket("test").await.unwrap();

        let body = b"hello world";
        let expected_md5 = format!(
            "\"{}\"",
            hex::encode(md5::Md5::digest(body))
        );

        let put = backend
            .put_object("test", "k", None, HashMap::new(), Bytes::from_static(body))
            .await
            .unwrap();
        assert_eq!(put.etag, expected_md5);

        let get = backend.get_object("test", "k").await.unwrap();
        assert_eq!(get.metadata.etag, expected_md5);
    }

    #[tokio::test]
    async fn test_etag_changes_on_overwrite() {
        let (backend, _dir) = temp_backend().await;
        backend.create_bucket("test").await.unwrap();

        let put1 = backend
            .put_object("test", "k", None, HashMap::new(), Bytes::from("aaa"))
            .await
            .unwrap();
        let put2 = backend
            .put_object("test", "k", None, HashMap::new(), Bytes::from("bbb"))
            .await
            .unwrap();

        assert_ne!(put1.etag, put2.etag);
    }

    #[tokio::test]
    async fn test_same_content_same_etag() {
        let (backend, _dir) = temp_backend().await;
        backend.create_bucket("test").await.unwrap();

        let put1 = backend
            .put_object("test", "a", None, HashMap::new(), Bytes::from("same"))
            .await
            .unwrap();
        let put2 = backend
            .put_object("test", "b", None, HashMap::new(), Bytes::from("same"))
            .await
            .unwrap();

        assert_eq!(put1.etag, put2.etag);
    }

    // =========================================================================
    // ListObjectsV2 — pagination, prefix, continuation
    // =========================================================================

    #[tokio::test]
    async fn test_list_objects_empty_bucket() {
        let (backend, _dir) = temp_backend().await;
        backend.create_bucket("test").await.unwrap();

        let list = backend
            .list_objects_v2("test", None, None, None, None)
            .await
            .unwrap();
        assert_eq!(list.objects.len(), 0);
        assert_eq!(list.key_count, 0);
        assert!(!list.is_truncated);
        assert!(list.next_continuation_token.is_none());
    }

    #[tokio::test]
    async fn test_list_objects_sorted_by_key() {
        let (backend, _dir) = temp_backend().await;
        backend.create_bucket("test").await.unwrap();

        // Insert out of order
        for key in ["zebra", "apple", "mango", "banana"] {
            backend
                .put_object("test", key, None, HashMap::new(), Bytes::from("x"))
                .await
                .unwrap();
        }

        let list = backend
            .list_objects_v2("test", None, None, None, None)
            .await
            .unwrap();
        let keys: Vec<_> = list.objects.iter().map(|o| o.key.as_str()).collect();
        assert_eq!(keys, vec!["apple", "banana", "mango", "zebra"]);
    }

    #[tokio::test]
    async fn test_list_objects_with_prefix() {
        let (backend, _dir) = temp_backend().await;
        backend.create_bucket("test").await.unwrap();

        for key in ["photos/2024/a.jpg", "photos/2024/b.jpg", "photos/2025/c.jpg", "docs/readme.md"] {
            backend
                .put_object("test", key, None, HashMap::new(), Bytes::from("x"))
                .await
                .unwrap();
        }

        let list = backend
            .list_objects_v2("test", Some("photos/2024/"), None, None, None)
            .await
            .unwrap();
        assert_eq!(list.key_count, 2);
        assert_eq!(list.prefix.as_deref(), Some("photos/2024/"));

        let list = backend
            .list_objects_v2("test", Some("photos/"), None, None, None)
            .await
            .unwrap();
        assert_eq!(list.key_count, 3);

        let list = backend
            .list_objects_v2("test", Some("nonexistent/"), None, None, None)
            .await
            .unwrap();
        assert_eq!(list.key_count, 0);
    }

    #[tokio::test]
    async fn test_list_objects_pagination() {
        let (backend, _dir) = temp_backend().await;
        backend.create_bucket("test").await.unwrap();

        for i in 0..5 {
            backend
                .put_object(
                    "test",
                    &format!("item-{i:02}"),
                    None,
                    HashMap::new(),
                    Bytes::from("data"),
                )
                .await
                .unwrap();
        }

        // Page 1
        let page1 = backend
            .list_objects_v2("test", None, None, Some(2), None)
            .await
            .unwrap();
        assert_eq!(page1.objects.len(), 2);
        assert!(page1.is_truncated);
        assert!(page1.next_continuation_token.is_some());
        assert_eq!(page1.objects[0].key, "item-00");
        assert_eq!(page1.objects[1].key, "item-01");

        // Page 2
        let page2 = backend
            .list_objects_v2(
                "test",
                None,
                page1.next_continuation_token.as_deref(),
                Some(2),
                None,
            )
            .await
            .unwrap();
        assert_eq!(page2.objects.len(), 2);
        assert!(page2.is_truncated);
        assert_eq!(page2.objects[0].key, "item-02");
        assert_eq!(page2.objects[1].key, "item-03");

        // Page 3 (last)
        let page3 = backend
            .list_objects_v2(
                "test",
                None,
                page2.next_continuation_token.as_deref(),
                Some(2),
                None,
            )
            .await
            .unwrap();
        assert_eq!(page3.objects.len(), 1);
        assert!(!page3.is_truncated);
        assert!(page3.next_continuation_token.is_none());
        assert_eq!(page3.objects[0].key, "item-04");
    }

    #[tokio::test]
    async fn test_list_objects_metadata_fields() {
        let (backend, _dir) = temp_backend().await;
        backend.create_bucket("test").await.unwrap();

        let put = backend
            .put_object("test", "file.txt", None, HashMap::new(), Bytes::from("hello"))
            .await
            .unwrap();

        let list = backend
            .list_objects_v2("test", None, None, None, None)
            .await
            .unwrap();
        assert_eq!(list.objects.len(), 1);
        let obj = &list.objects[0];
        assert_eq!(obj.key, "file.txt");
        assert_eq!(obj.size, 5);
        assert_eq!(obj.etag, put.etag);
    }

    // =========================================================================
    // Multipart upload — basic
    // =========================================================================

    #[tokio::test]
    async fn test_multipart_upload() {
        let (backend, _dir) = temp_backend().await;
        backend.create_bucket("test").await.unwrap();

        let create = backend
            .create_multipart_upload("test", "big.bin", None, HashMap::new())
            .await
            .unwrap();
        assert_eq!(create.bucket, "test");
        assert_eq!(create.key, "big.bin");
        assert!(!create.upload_id.is_empty());
        let uid = &create.upload_id;

        // Parts must be >= 5 MB (except the last) per S3 spec
        let min_part = 5 * 1024 * 1024;
        let part1 = backend
            .upload_part("test", "big.bin", uid, 1, Bytes::from(vec![0xAAu8; min_part]))
            .await
            .unwrap();
        let part2 = backend
            .upload_part("test", "big.bin", uid, 2, Bytes::from(vec![0xBBu8; 100]))
            .await
            .unwrap();

        let parts = backend
            .list_parts("test", "big.bin", uid, None, None)
            .await
            .unwrap();
        assert_eq!(parts.parts.len(), 2);
        assert_eq!(parts.parts[0].part_number, 1);
        assert_eq!(parts.parts[1].part_number, 2);

        let complete = backend
            .complete_multipart_upload(
                "test",
                "big.bin",
                uid,
                vec![
                    (1, part1.etag.clone()),
                    (2, part2.etag.clone()),
                ],
            )
            .await
            .unwrap();
        assert!(complete.etag.contains("-2"), "Expected compound ETag with -2 suffix");
        assert_eq!(complete.size as usize, min_part + 100);
        assert_eq!(complete.bucket, "test");
        assert_eq!(complete.key, "big.bin");

        let get = backend.get_object("test", "big.bin").await.unwrap();
        assert_eq!(get.body.len(), min_part + 100);
        assert!(get.body[..min_part].iter().all(|b| *b == 0xAA));
        assert!(get.body[min_part..].iter().all(|b| *b == 0xBB));
    }

    #[tokio::test]
    async fn test_abort_multipart() {
        let (backend, _dir) = temp_backend().await;
        backend.create_bucket("test").await.unwrap();

        let create = backend
            .create_multipart_upload("test", "aborted.bin", None, HashMap::new())
            .await
            .unwrap();

        backend
            .upload_part(
                "test",
                "aborted.bin",
                &create.upload_id,
                1,
                Bytes::from(vec![0u8; 50]),
            )
            .await
            .unwrap();

        backend
            .abort_multipart_upload("test", "aborted.bin", &create.upload_id)
            .await
            .unwrap();

        // Upload should be gone
        let result = backend
            .list_parts("test", "aborted.bin", &create.upload_id, None, None)
            .await;
        assert!(result.is_err());

        // No object should exist
        let head = backend.head_object("test", "aborted.bin").await.unwrap();
        assert!(head.is_none());
    }

    // =========================================================================
    // Multipart upload — metadata preserved
    // =========================================================================

    #[tokio::test]
    async fn test_multipart_preserves_content_type() {
        let (backend, _dir) = temp_backend().await;
        backend.create_bucket("test").await.unwrap();

        let create = backend
            .create_multipart_upload("test", "image.png", Some("image/png"), HashMap::new())
            .await
            .unwrap();

        let part = backend
            .upload_part("test", "image.png", &create.upload_id, 1, Bytes::from(vec![0u8; 10]))
            .await
            .unwrap();

        backend
            .complete_multipart_upload(
                "test",
                "image.png",
                &create.upload_id,
                vec![(1, part.etag)],
            )
            .await
            .unwrap();

        let get = backend.get_object("test", "image.png").await.unwrap();
        assert_eq!(get.metadata.content_type, "image/png");
    }

    #[tokio::test]
    async fn test_multipart_preserves_user_metadata() {
        let (backend, _dir) = temp_backend().await;
        backend.create_bucket("test").await.unwrap();

        let mut meta = HashMap::new();
        meta.insert("project".to_string(), "post3".to_string());
        meta.insert("author".to_string(), "test".to_string());

        let create = backend
            .create_multipart_upload("test", "doc.bin", None, meta)
            .await
            .unwrap();

        let part = backend
            .upload_part("test", "doc.bin", &create.upload_id, 1, Bytes::from("data"))
            .await
            .unwrap();

        backend
            .complete_multipart_upload(
                "test",
                "doc.bin",
                &create.upload_id,
                vec![(1, part.etag)],
            )
            .await
            .unwrap();

        let get = backend.get_object("test", "doc.bin").await.unwrap();
        assert_eq!(get.user_metadata.get("project").unwrap(), "post3");
        assert_eq!(get.user_metadata.get("author").unwrap(), "test");
    }

    // =========================================================================
    // Multipart upload — error cases
    // =========================================================================

    #[tokio::test]
    async fn test_multipart_invalid_part_order() {
        let (backend, _dir) = temp_backend().await;
        backend.create_bucket("test").await.unwrap();

        let create = backend
            .create_multipart_upload("test", "k", None, HashMap::new())
            .await
            .unwrap();
        let uid = &create.upload_id;

        let part1 = backend
            .upload_part("test", "k", uid, 1, Bytes::from("a"))
            .await
            .unwrap();
        let part2 = backend
            .upload_part("test", "k", uid, 2, Bytes::from("b"))
            .await
            .unwrap();

        // Wrong order in complete request
        let err = backend
            .complete_multipart_upload(
                "test",
                "k",
                uid,
                vec![(2, part2.etag), (1, part1.etag)],
            )
            .await
            .unwrap_err();
        assert!(matches!(err, Post3Error::InvalidPartOrder));
    }

    #[tokio::test]
    async fn test_multipart_wrong_etag() {
        let (backend, _dir) = temp_backend().await;
        backend.create_bucket("test").await.unwrap();

        let create = backend
            .create_multipart_upload("test", "k", None, HashMap::new())
            .await
            .unwrap();
        let uid = &create.upload_id;

        backend
            .upload_part("test", "k", uid, 1, Bytes::from("data"))
            .await
            .unwrap();

        let err = backend
            .complete_multipart_upload(
                "test",
                "k",
                uid,
                vec![(1, "\"wrong-etag\"".to_string())],
            )
            .await
            .unwrap_err();
        assert!(matches!(err, Post3Error::ETagMismatch { .. }));
    }

    #[tokio::test]
    async fn test_multipart_missing_part() {
        let (backend, _dir) = temp_backend().await;
        backend.create_bucket("test").await.unwrap();

        let create = backend
            .create_multipart_upload("test", "k", None, HashMap::new())
            .await
            .unwrap();
        let uid = &create.upload_id;

        // Part 1 must be >= 5 MB since it's not the last part
        let min_part = 5 * 1024 * 1024;
        let part1 = backend
            .upload_part("test", "k", uid, 1, Bytes::from(vec![0u8; min_part]))
            .await
            .unwrap();

        // Request part 1 and 3, but 3 was never uploaded
        let err = backend
            .complete_multipart_upload(
                "test",
                "k",
                uid,
                vec![(1, part1.etag), (3, "\"fake\"".to_string())],
            )
            .await
            .unwrap_err();
        assert!(matches!(err, Post3Error::InvalidPart { .. }));
    }

    #[tokio::test]
    async fn test_multipart_nonexistent_upload() {
        let (backend, _dir) = temp_backend().await;
        backend.create_bucket("test").await.unwrap();

        let err = backend
            .upload_part(
                "test",
                "key",
                "nonexistent-upload-id",
                1,
                Bytes::from("data"),
            )
            .await
            .unwrap_err();
        assert!(matches!(err, Post3Error::UploadNotFound(_)));
    }

    #[tokio::test]
    async fn test_abort_nonexistent_upload() {
        let (backend, _dir) = temp_backend().await;
        backend.create_bucket("test").await.unwrap();

        let err = backend
            .abort_multipart_upload("test", "key", "no-such-upload")
            .await
            .unwrap_err();
        assert!(matches!(err, Post3Error::UploadNotFound(_)));
    }

    #[tokio::test]
    async fn test_multipart_on_nonexistent_bucket() {
        let (backend, _dir) = temp_backend().await;

        let err = backend
            .create_multipart_upload("nope", "key", None, HashMap::new())
            .await
            .unwrap_err();
        assert!(matches!(err, Post3Error::BucketNotFound(_)));
    }

    // =========================================================================
    // Multipart upload — part overwrite
    // =========================================================================

    #[tokio::test]
    async fn test_upload_part_overwrites_previous() {
        let (backend, _dir) = temp_backend().await;
        backend.create_bucket("test").await.unwrap();

        let create = backend
            .create_multipart_upload("test", "k", None, HashMap::new())
            .await
            .unwrap();
        let uid = &create.upload_id;

        // Upload part 1 with initial data
        backend
            .upload_part("test", "k", uid, 1, Bytes::from("old_data"))
            .await
            .unwrap();

        // Re-upload part 1 with new data
        let part1_new = backend
            .upload_part("test", "k", uid, 1, Bytes::from("new_data"))
            .await
            .unwrap();

        // Complete with the new etag
        let complete = backend
            .complete_multipart_upload("test", "k", uid, vec![(1, part1_new.etag)])
            .await
            .unwrap();
        assert_eq!(complete.size, 8);

        let get = backend.get_object("test", "k").await.unwrap();
        assert_eq!(get.body.as_ref(), b"new_data");
    }

    // =========================================================================
    // Multipart upload — single part
    // =========================================================================

    #[tokio::test]
    async fn test_multipart_single_part() {
        let (backend, _dir) = temp_backend().await;
        backend.create_bucket("test").await.unwrap();

        let create = backend
            .create_multipart_upload("test", "single.bin", None, HashMap::new())
            .await
            .unwrap();

        let part = backend
            .upload_part("test", "single.bin", &create.upload_id, 1, Bytes::from("only-part"))
            .await
            .unwrap();

        let complete = backend
            .complete_multipart_upload(
                "test",
                "single.bin",
                &create.upload_id,
                vec![(1, part.etag)],
            )
            .await
            .unwrap();
        assert!(complete.etag.contains("-1"));
        assert_eq!(complete.size, 9);
    }

    // =========================================================================
    // Multipart upload — non-contiguous part numbers
    // =========================================================================

    #[tokio::test]
    async fn test_multipart_non_contiguous_parts() {
        let (backend, _dir) = temp_backend().await;
        backend.create_bucket("test").await.unwrap();

        let create = backend
            .create_multipart_upload("test", "sparse.bin", None, HashMap::new())
            .await
            .unwrap();
        let uid = &create.upload_id;

        // Upload parts 1, 5, 10 (skipping numbers is valid in S3)
        // Non-last parts must be >= 5 MB per S3 spec
        let min_part = 5 * 1024 * 1024;
        let p1 = backend
            .upload_part(
                "test",
                "sparse.bin",
                uid,
                1,
                Bytes::from(vec![0xAAu8; min_part]),
            )
            .await
            .unwrap();
        let p5 = backend
            .upload_part(
                "test",
                "sparse.bin",
                uid,
                5,
                Bytes::from(vec![0xBBu8; min_part]),
            )
            .await
            .unwrap();
        let p10 = backend
            .upload_part("test", "sparse.bin", uid, 10, Bytes::from("ccc"))
            .await
            .unwrap();

        let complete = backend
            .complete_multipart_upload(
                "test",
                "sparse.bin",
                uid,
                vec![(1, p1.etag), (5, p5.etag), (10, p10.etag)],
            )
            .await
            .unwrap();
        assert!(complete.etag.contains("-3"));
        assert_eq!(complete.size as usize, min_part * 2 + 3);

        let get = backend.get_object("test", "sparse.bin").await.unwrap();
        assert_eq!(get.body.len(), min_part * 2 + 3);
        assert!(get.body[..min_part].iter().all(|b| *b == 0xAA));
        assert!(get.body[min_part..min_part * 2].iter().all(|b| *b == 0xBB));
        assert_eq!(&get.body[min_part * 2..], b"ccc");
    }

    // =========================================================================
    // ListParts — pagination
    // =========================================================================

    #[tokio::test]
    async fn test_list_parts_pagination() {
        let (backend, _dir) = temp_backend().await;
        backend.create_bucket("test").await.unwrap();

        let create = backend
            .create_multipart_upload("test", "k", None, HashMap::new())
            .await
            .unwrap();
        let uid = &create.upload_id;

        for i in 1..=5 {
            backend
                .upload_part("test", "k", uid, i, Bytes::from(format!("part{i}")))
                .await
                .unwrap();
        }

        // Page 1: max_parts=2
        let page1 = backend
            .list_parts("test", "k", uid, Some(2), None)
            .await
            .unwrap();
        assert_eq!(page1.parts.len(), 2);
        assert!(page1.is_truncated);
        assert_eq!(page1.parts[0].part_number, 1);
        assert_eq!(page1.parts[1].part_number, 2);

        // Page 2: marker=2
        let page2 = backend
            .list_parts("test", "k", uid, Some(2), page1.next_part_number_marker)
            .await
            .unwrap();
        assert_eq!(page2.parts.len(), 2);
        assert!(page2.is_truncated);
        assert_eq!(page2.parts[0].part_number, 3);
        assert_eq!(page2.parts[1].part_number, 4);

        // Page 3: last page
        let page3 = backend
            .list_parts("test", "k", uid, Some(2), page2.next_part_number_marker)
            .await
            .unwrap();
        assert_eq!(page3.parts.len(), 1);
        assert!(!page3.is_truncated);
        assert_eq!(page3.parts[0].part_number, 5);
    }

    // =========================================================================
    // ListMultipartUploads
    // =========================================================================

    #[tokio::test]
    async fn test_list_multipart_uploads() {
        let (backend, _dir) = temp_backend().await;
        backend.create_bucket("test").await.unwrap();

        let u1 = backend
            .create_multipart_upload("test", "alpha.bin", None, HashMap::new())
            .await
            .unwrap();
        let u2 = backend
            .create_multipart_upload("test", "beta.bin", None, HashMap::new())
            .await
            .unwrap();

        let list = backend
            .list_multipart_uploads("test", None, None, None, None)
            .await
            .unwrap();
        assert_eq!(list.uploads.len(), 2);
        assert_eq!(list.bucket, "test");
        // Should be sorted by key
        assert_eq!(list.uploads[0].key, "alpha.bin");
        assert_eq!(list.uploads[1].key, "beta.bin");

        // After aborting one, only the other should remain
        backend
            .abort_multipart_upload("test", "alpha.bin", &u1.upload_id)
            .await
            .unwrap();

        let list = backend
            .list_multipart_uploads("test", None, None, None, None)
            .await
            .unwrap();
        assert_eq!(list.uploads.len(), 1);
        assert_eq!(list.uploads[0].key, "beta.bin");
        assert_eq!(list.uploads[0].upload_id, u2.upload_id);
    }

    #[tokio::test]
    async fn test_list_multipart_uploads_with_prefix() {
        let (backend, _dir) = temp_backend().await;
        backend.create_bucket("test").await.unwrap();

        backend
            .create_multipart_upload("test", "photos/a.jpg", None, HashMap::new())
            .await
            .unwrap();
        backend
            .create_multipart_upload("test", "photos/b.jpg", None, HashMap::new())
            .await
            .unwrap();
        backend
            .create_multipart_upload("test", "docs/readme.md", None, HashMap::new())
            .await
            .unwrap();

        let list = backend
            .list_multipart_uploads("test", Some("photos/"), None, None, None)
            .await
            .unwrap();
        assert_eq!(list.uploads.len(), 2);
        assert_eq!(list.prefix.as_deref(), Some("photos/"));
    }

    #[tokio::test]
    async fn test_list_multipart_uploads_empty() {
        let (backend, _dir) = temp_backend().await;
        backend.create_bucket("test").await.unwrap();

        let list = backend
            .list_multipart_uploads("test", None, None, None, None)
            .await
            .unwrap();
        assert!(list.uploads.is_empty());
        assert!(!list.is_truncated);
    }

    // =========================================================================
    // Multipart upload cleans up after completion
    // =========================================================================

    #[tokio::test]
    async fn test_multipart_cleanup_after_complete() {
        let (backend, _dir) = temp_backend().await;
        backend.create_bucket("test").await.unwrap();

        let create = backend
            .create_multipart_upload("test", "k", None, HashMap::new())
            .await
            .unwrap();
        let uid = &create.upload_id;

        let part = backend
            .upload_part("test", "k", uid, 1, Bytes::from("data"))
            .await
            .unwrap();

        backend
            .complete_multipart_upload("test", "k", uid, vec![(1, part.etag)])
            .await
            .unwrap();

        // The upload should no longer be listed
        let list = backend
            .list_multipart_uploads("test", None, None, None, None)
            .await
            .unwrap();
        assert!(list.uploads.is_empty());

        // list_parts should fail
        let err = backend
            .list_parts("test", "k", uid, None, None)
            .await
            .unwrap_err();
        assert!(matches!(err, Post3Error::UploadNotFound(_)));
    }

    // =========================================================================
    // Multiple concurrent uploads for same key
    // =========================================================================

    #[tokio::test]
    async fn test_multiple_uploads_same_key() {
        let (backend, _dir) = temp_backend().await;
        backend.create_bucket("test").await.unwrap();

        // Start two uploads for the same key
        let u1 = backend
            .create_multipart_upload("test", "same-key", None, HashMap::new())
            .await
            .unwrap();
        let u2 = backend
            .create_multipart_upload("test", "same-key", None, HashMap::new())
            .await
            .unwrap();
        assert_ne!(u1.upload_id, u2.upload_id);

        // Both should appear in listings
        let list = backend
            .list_multipart_uploads("test", None, None, None, None)
            .await
            .unwrap();
        assert_eq!(list.uploads.len(), 2);

        // Complete one, abort the other
        let part = backend
            .upload_part("test", "same-key", &u1.upload_id, 1, Bytes::from("from-u1"))
            .await
            .unwrap();
        backend
            .complete_multipart_upload("test", "same-key", &u1.upload_id, vec![(1, part.etag)])
            .await
            .unwrap();
        backend
            .abort_multipart_upload("test", "same-key", &u2.upload_id)
            .await
            .unwrap();

        let get = backend.get_object("test", "same-key").await.unwrap();
        assert_eq!(get.body.as_ref(), b"from-u1");
    }

    // =========================================================================
    // Key encoding round-trip
    // =========================================================================

    #[tokio::test]
    async fn test_key_encoding_roundtrip() {
        // Verify the encode/decode functions are inverse
        let test_keys = [
            "simple",
            "with spaces",
            "path/to/file",
            "special!@#$%^&*()",
            "unicode/日本語",
            "dots...and..more",
            "",
        ];

        for key in &test_keys {
            let encoded = encode_key(key);
            let decoded = decode_key(&encoded);
            assert_eq!(&decoded, key, "Round-trip failed for: {key:?}");
        }
    }
}