// nb: for paths inside the root, we use String, since s3 only allows utf8 object keys

use aws_sdk_s3::primitives::ByteStream;
use url::Url;

pub enum Bucket {
    #[allow(unused)]
    Local {
        root: std::path::PathBuf,
    },
    S3 {
        client: aws_sdk_s3::Client,
        bucket_name: String,
        prefix: String,
    },
}

fn get_content_type(key: &str) -> &'static str {
    if key.ends_with(".html") {
        "text/html"
    } else if key.ends_with(".css") {
        "text/css"
    } else if key.ends_with(".js") {
        "application/javascript"
    } else if key.ends_with(".txt") {
        "text/plain"
    } else if key.ends_with(".json") {
        "application/json"
    } else {
        "application/octet-stream"
    }
}

impl Bucket {
    #[allow(unused)]
    pub fn new_local(root: std::path::PathBuf) -> Self {
        Bucket::Local { root }
    }

    #[allow(unused)]
    pub fn new_tmp() -> Self {
        let path = std::env::temp_dir().join(format!(
            "j3-{:x}",
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_nanos()
        ));
        std::fs::create_dir_all(&path).expect("failed to create temp dir");
        Bucket::Local { root: path }
    }

    pub async fn new_s3(url_str: &str) -> Self {
        let url_str = if url_str.starts_with("http://") || url_str.starts_with("https://") {
            url_str.to_string()
        } else {
            format!("https://{}", url_str)
        };
        let parsed = Url::parse(&url_str).expect("Invalid URL format");
        let host = parsed.host_str().expect("URL missing endpoint/host");
        let port = parsed.port();
        let scheme = parsed.scheme();

        let endpoint_url = if let Some(p) = port {
            format!("{}://{}:{}", scheme, host, p)
        } else {
            format!("{}://{}", scheme, host)
        };

        let mut segments = parsed.path_segments().expect("URL cannot be a base");
        let bucket_name = segments
            .next()
            .filter(|s| !s.is_empty())
            .expect("URL missing bucket name (format: j3://endpoint/bucket[/prefix])")
            .to_string();

        // The rest of the segments form the prefix.
        // If there are remaining segments, join them with '/'.
        // We also append a trailing slash if the prefix is not empty, to make joining keys easier later.
        let mut prefix = segments.collect::<Vec<_>>().join("/");
        if !prefix.is_empty() && !prefix.ends_with('/') {
            prefix.push('/');
        }

        let config = aws_config::load_defaults(aws_config::BehaviorVersion::latest()).await;

        let mut s3_config_builder = aws_sdk_s3::config::Builder::from(&config)
            .endpoint_url(endpoint_url)
            .force_path_style(true);

        if config.region().is_none() {
            s3_config_builder = s3_config_builder.region(aws_sdk_s3::config::Region::new("us-east-1"));
        }

        if !parsed.username().is_empty() {
            let creds = aws_sdk_s3::config::Credentials::new(
                parsed.username(),
                parsed.password().unwrap_or(""),
                None,
                None,
                "url",
            );
            s3_config_builder = s3_config_builder.credentials_provider(creds);
        }

        let client = aws_sdk_s3::Client::from_conf(s3_config_builder.build());

        Bucket::S3 {
            client,
            bucket_name,
            prefix,
        }
    }

    /// lists up to 1000 files with 'prefix' as their key prefix. prefix must
    /// end in a '/'. returns the full key of each of those files, including `prefix` in each key.
    pub async fn list_files(&mut self, list_prefix: String) -> Vec<String> {
        match self {
            Bucket::Local { root } => {
                let mut results = Vec::new();
                let mut stack = vec![root.join(&list_prefix)];

                while let Some(dir) = stack.pop() {
                    if !dir.exists() {
                        continue;
                    }

                    let mut entries = match tokio::fs::read_dir(dir).await {
                        Ok(e) => e,
                        Err(_) => continue,
                    };

                    while let Ok(Some(entry)) = entries.next_entry().await {
                        let file_type = match entry.file_type().await {
                            Ok(ft) => ft,
                            Err(_) => continue,
                        };
                        let path = entry.path();
                        if file_type.is_dir() {
                            stack.push(path);
                        } else {
                            if let Ok(rel_path) = path.strip_prefix(root.as_path()) {
                                if let Some(key) = rel_path.to_str() {
                                    if key.starts_with(&list_prefix) {
                                        results.push(key.to_string());
                                    }
                                }
                            }
                        }
                    }
                }
                results.sort();
                results.truncate(1000);
                results
            }
            Bucket::S3 {
                client,
                bucket_name,
                prefix: bucket_prefix,
            } => {
                // effective_prefix = bucket_prefix + list_prefix
                let effective_prefix = format!("{}{}", bucket_prefix, list_prefix);

                let resp = client
                    .list_objects_v2()
                    .bucket(bucket_name.as_str())
                    .prefix(&effective_prefix)
                    .max_keys(1000)
                    .send()
                    .await
                    .expect("failed to list objects");

                let mut results = Vec::new();
                for object in resp.contents.unwrap_or_default() {
                    if let Some(full_key) = object.key {
                        // We need to return the key relative to the bucket root (which implies removing bucket_prefix?)
                        // Wait, the interface doc says: "returns the full key of each of those files, including `prefix` in each key."
                        // In local implementation: `key.starts_with(&prefix)`. The `key` is relative to `root`.
                        // `root` is equivalent to `bucket_prefix`.
                        // So if we have `bucket_prefix="repo1/"` and we list "refs/", we look for "repo1/refs/...".
                        // S3 returns "repo1/refs/heads/main".
                        // We should strip "repo1/" to return "refs/heads/main" to the caller?
                        // Let's check Local implementation.
                        // `root` is the base path. `path.strip_prefix(root)` gives the key.
                        // So if `root` = `/tmp/j3-123/`, file is `/tmp/j3-123/refs/heads/main`. Key is `refs/heads/main`.
                        // Caller asks for list_prefix="refs/". Result includes "refs/heads/main".
                        // So yes, we must strip `bucket_prefix` from the S3 key.

                        if let Some(stripped) = full_key.strip_prefix(bucket_prefix.as_str()) {
                            results.push(stripped.to_string());
                        }
                    }
                }
                results.sort();
                results
            }
        }
    }

    pub async fn put(&mut self, key: String, contents: Vec<u8>) {
        match self {
            Bucket::Local { root } => {
                let path = root.join(&key);
                if let Some(parent) = path.parent() {
                    tokio::fs::create_dir_all(parent)
                        .await
                        .expect("failed to create parent directories");
                }
                tokio::fs::write(path, contents)
                    .await
                    .expect("failed to write file");
            }
            Bucket::S3 {
                client,
                bucket_name,
                prefix,
            } => {
                let full_key = format!("{}{}", prefix, key);
                let content_type = get_content_type(&key);
                client
                    .put_object()
                    .bucket(bucket_name.as_str())
                    .key(full_key)
                    .content_type(content_type)
                    .body(ByteStream::from(contents))
                    .send()
                    .await
                    .expect("failed to put object");
            }
        }
    }

    pub async fn get(&mut self, key: String) -> Vec<u8> {
        match self {
            Bucket::Local { root } => {
                let path = root.join(&key);
                tokio::fs::read(path).await.expect("failed to read file")
            }
            Bucket::S3 {
                client,
                bucket_name,
                prefix,
            } => {
                let full_key = format!("{}{}", prefix, key);
                let resp = client
                    .get_object()
                    .bucket(bucket_name.as_str())
                    .key(full_key)
                    .send()
                    .await
                    .expect("failed to get object");

                let data = resp.body.collect().await.expect("failed to read body");
                data.into_bytes().to_vec()
            }
        }
    }

    pub async fn exists(&mut self, key: String) -> bool {
        match self {
            Bucket::Local { root } => root.join(key).exists(),
            Bucket::S3 {
                client,
                bucket_name,
                prefix,
            } => {
                let full_key = format!("{}{}", prefix, key);
                match client
                    .head_object()
                    .bucket(bucket_name.as_str())
                    .key(full_key)
                    .send()
                    .await
                {
                    Ok(_) => true,
                    Err(aws_sdk_s3::error::SdkError::ServiceError(e)) => {
                        if e.raw().status().as_u16() == 404 {
                            false
                        } else {
                            panic!("failed to head object: {:?}", e)
                        }
                    }
                    Err(e) => panic!("failed to head object: {}", e),
                }
            }
        }
    }

    #[allow(unused)]
    pub async fn delete(&mut self, key: String) {
        match self {
            Bucket::Local { root } => {
                let path = root.join(key);
                if path.exists() {
                    tokio::fs::remove_file(path)
                        .await
                        .expect("failed to remove file");
                }
            }
            Bucket::S3 {
                client,
                bucket_name,
                prefix,
            } => {
                let full_key = format!("{}{}", prefix, key);
                client
                    .delete_object()
                    .bucket(bucket_name.as_str())
                    .key(full_key)
                    .send()
                    .await
                    .expect("failed to delete object");
            }
        }
    }
}

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

    #[tokio::test]
    async fn test_local_bucket() {
        let mut bucket = Bucket::new_tmp();

        // 1. Put a file
        let key = "folder/test.txt".to_string();
        let content = b"hello world".to_vec();
        bucket.put(key.clone(), content.clone()).await;

        // 2. Exists
        assert!(bucket.exists(key.clone()).await);

        // 3. Get
        let retrieved = bucket.get(key.clone()).await;
        assert_eq!(retrieved, content);

        // 4. List files (with subfolders)
        let key2 = "folder/sub/inner.txt".to_string();
        bucket.put(key2.clone(), b"inner".to_vec()).await;
        let key3 = "other/file.txt".to_string();
        bucket.put(key3.clone(), b"other".to_vec()).await;

        let files = bucket.list_files("folder/".to_string()).await;
        let mut expected = vec![key.clone(), key2.clone()];
        expected.sort();
        assert_eq!(files, expected);

        // 5. Delete
        bucket.delete(key.clone()).await;
        bucket.delete(key2.clone()).await;
        bucket.delete(key3.clone()).await;

        // 6. Check deletion
        assert!(!bucket.exists(key.clone()).await);
        let files_after = bucket.list_files("folder/".to_string()).await;
        assert!(files_after.is_empty());
    }

    #[tokio::test]
    async fn test_put_if_not_exists() {
        let mut bucket = Bucket::new_tmp();
        let key = "test.txt".to_string();
        let content = b"foo".to_vec();

        // 1. First put should succeed
        assert!(bucket
            .put_if_not_exists(key.clone(), content.clone())
            .await
            .is_ok());
        assert_eq!(bucket.get(key.clone()).await, content);

        // 2. Second put should fail with AlreadyExists
        let result = bucket.put_if_not_exists(key.clone(), b"bar".to_vec()).await;
        assert!(result.is_err());
        // Since AlreadyExists is a unit struct, we can check matching
        assert!(matches!(result.err().unwrap(), AlreadyExists));

        // 3. Content should remain "foo"
        assert_eq!(bucket.get(key.clone()).await, content);
    }
}
