use aws_credential_types::Credentials;
use aws_sdk_s3::types::{CompletedMultipartUpload, CompletedPart};
use aws_sdk_s3::Client;
use bytes::Bytes;

pub use aws_sdk_s3;
pub use bytes;

/// Error type for post3-sdk operations.
#[derive(Debug, thiserror::Error)]
pub enum Error {
    #[error("bucket not found: {0}")]
    BucketNotFound(String),

    #[error("object not found: {bucket}/{key}")]
    ObjectNotFound { bucket: String, key: String },

    #[error("s3 error: {0}")]
    S3(String),
}

impl<E: std::fmt::Display> From<aws_sdk_s3::error::SdkError<E>> for Error {
    fn from(err: aws_sdk_s3::error::SdkError<E>) -> Self {
        Error::S3(err.to_string())
    }
}

pub type Result<T> = std::result::Result<T, Error>;

/// Summary of an object returned by list operations.
#[derive(Debug, Clone)]
pub struct ObjectInfo {
    pub key: String,
    pub size: i64,
    pub etag: Option<String>,
    pub last_modified: Option<chrono::DateTime<chrono::Utc>>,
}

/// A client for post3 that wraps `aws-sdk-s3` with ergonomic defaults.
///
/// # Example
///
/// ```no_run
/// # async fn example() -> post3_sdk::Result<()> {
/// let client = post3_sdk::Post3Client::new("http://localhost:9000");
///
/// client.create_bucket("my-bucket").await?;
/// client.put_object("my-bucket", "hello.txt", b"hello world").await?;
///
/// let data = client.get_object("my-bucket", "hello.txt").await?;
/// assert_eq!(data.as_ref(), b"hello world");
/// # Ok(())
/// # }
/// ```
pub struct Post3Client {
    inner: Client,
}

impl Post3Client {
    /// Create a client with default configuration (dummy credentials, us-east-1, path-style).
    pub fn new(endpoint_url: impl Into<String>) -> Self {
        Self::builder().endpoint_url(endpoint_url).build()
    }

    /// Access the underlying `aws_sdk_s3::Client` for advanced operations.
    pub fn inner(&self) -> &Client {
        &self.inner
    }

    /// Start building a client with custom configuration.
    pub fn builder() -> Post3ClientBuilder {
        Post3ClientBuilder::default()
    }

    // -- Bucket operations --

    pub async fn create_bucket(&self, name: &str) -> Result<()> {
        self.inner
            .create_bucket()
            .bucket(name)
            .send()
            .await?;
        Ok(())
    }

    pub async fn head_bucket(&self, name: &str) -> Result<bool> {
        match self.inner.head_bucket().bucket(name).send().await {
            Ok(_) => Ok(true),
            Err(err) => {
                if err
                    .as_service_error()
                    .map_or(false, |e| e.is_not_found())
                {
                    Ok(false)
                } else {
                    Err(Error::S3(err.to_string()))
                }
            }
        }
    }

    pub async fn delete_bucket(&self, name: &str) -> Result<()> {
        self.inner
            .delete_bucket()
            .bucket(name)
            .send()
            .await?;
        Ok(())
    }

    pub async fn list_buckets(&self) -> Result<Vec<String>> {
        let resp = self.inner.list_buckets().send().await?;
        Ok(resp
            .buckets()
            .iter()
            .filter_map(|b| b.name().map(|s| s.to_string()))
            .collect())
    }

    // -- Object operations --

    pub async fn put_object(
        &self,
        bucket: &str,
        key: &str,
        body: impl AsRef<[u8]>,
    ) -> Result<()> {
        let body = Bytes::copy_from_slice(body.as_ref());
        self.inner
            .put_object()
            .bucket(bucket)
            .key(key)
            .body(body.into())
            .send()
            .await?;
        Ok(())
    }

    pub async fn get_object(&self, bucket: &str, key: &str) -> Result<Bytes> {
        let resp = self
            .inner
            .get_object()
            .bucket(bucket)
            .key(key)
            .send()
            .await
            .map_err(|e| {
                if e.as_service_error()
                    .map_or(false, |se| se.is_no_such_key())
                {
                    Error::ObjectNotFound {
                        bucket: bucket.to_string(),
                        key: key.to_string(),
                    }
                } else {
                    Error::S3(e.to_string())
                }
            })?;

        let data = resp
            .body
            .collect()
            .await
            .map_err(|e| Error::S3(e.to_string()))?;
        Ok(data.into_bytes())
    }

    pub async fn head_object(
        &self,
        bucket: &str,
        key: &str,
    ) -> Result<Option<ObjectInfo>> {
        match self
            .inner
            .head_object()
            .bucket(bucket)
            .key(key)
            .send()
            .await
        {
            Ok(resp) => Ok(Some(ObjectInfo {
                key: key.to_string(),
                size: resp.content_length().unwrap_or(0),
                etag: resp.e_tag().map(|s| s.to_string()),
                last_modified: resp
                    .last_modified()
                    .and_then(|t| {
                        chrono::DateTime::from_timestamp(t.secs(), t.subsec_nanos())
                    }),
            })),
            Err(err) => {
                if err
                    .as_service_error()
                    .map_or(false, |e| e.is_not_found())
                {
                    Ok(None)
                } else {
                    Err(Error::S3(err.to_string()))
                }
            }
        }
    }

    pub async fn delete_object(&self, bucket: &str, key: &str) -> Result<()> {
        self.inner
            .delete_object()
            .bucket(bucket)
            .key(key)
            .send()
            .await?;
        Ok(())
    }

    /// Upload an object using multipart upload, splitting into parts of the given size.
    ///
    /// This is useful for large files where multipart upload provides better performance
    /// through parallelism and resumability.
    pub async fn multipart_upload(
        &self,
        bucket: &str,
        key: &str,
        data: impl AsRef<[u8]>,
        part_size: usize,
    ) -> Result<()> {
        let data = data.as_ref();

        // Create multipart upload
        let create_resp = self
            .inner
            .create_multipart_upload()
            .bucket(bucket)
            .key(key)
            .send()
            .await?;

        let upload_id = create_resp
            .upload_id()
            .ok_or_else(|| Error::S3("missing upload_id in response".to_string()))?
            .to_string();

        // Upload parts
        let mut completed_parts = Vec::new();
        let mut part_number = 1i32;

        for chunk in data.chunks(part_size) {
            let body = Bytes::copy_from_slice(chunk);
            let upload_resp = self
                .inner
                .upload_part()
                .bucket(bucket)
                .key(key)
                .upload_id(&upload_id)
                .part_number(part_number)
                .body(body.into())
                .send()
                .await
                .map_err(|e| {
                    // Try to abort on failure
                    Error::S3(e.to_string())
                })?;

            let etag = upload_resp
                .e_tag()
                .ok_or_else(|| Error::S3("missing ETag in upload_part response".to_string()))?
                .to_string();

            completed_parts.push(
                CompletedPart::builder()
                    .part_number(part_number)
                    .e_tag(etag)
                    .build(),
            );

            part_number += 1;
        }

        // Complete multipart upload
        let mut builder = CompletedMultipartUpload::builder();
        for part in completed_parts {
            builder = builder.parts(part);
        }

        self.inner
            .complete_multipart_upload()
            .bucket(bucket)
            .key(key)
            .upload_id(&upload_id)
            .multipart_upload(builder.build())
            .send()
            .await?;

        Ok(())
    }

    pub async fn list_objects(
        &self,
        bucket: &str,
        prefix: Option<&str>,
    ) -> Result<Vec<ObjectInfo>> {
        let mut req = self
            .inner
            .list_objects_v2()
            .bucket(bucket);

        if let Some(p) = prefix {
            req = req.prefix(p);
        }

        let resp = req.send().await?;
        Ok(resp
            .contents()
            .iter()
            .map(|obj| ObjectInfo {
                key: obj.key().unwrap_or_default().to_string(),
                size: obj.size().unwrap_or(0),
                etag: obj.e_tag().map(|s| s.to_string()),
                last_modified: obj
                    .last_modified()
                    .and_then(|t| {
                        chrono::DateTime::from_timestamp(t.secs(), t.subsec_nanos())
                    }),
            })
            .collect())
    }
}

/// Builder for `Post3Client` with custom configuration.
pub struct Post3ClientBuilder {
    endpoint_url: Option<String>,
    access_key: String,
    secret_key: String,
    region: String,
}

impl Default for Post3ClientBuilder {
    fn default() -> Self {
        Self {
            endpoint_url: None,
            access_key: "test".to_string(),
            secret_key: "test".to_string(),
            region: "us-east-1".to_string(),
        }
    }
}

impl Post3ClientBuilder {
    pub fn endpoint_url(mut self, url: impl Into<String>) -> Self {
        self.endpoint_url = Some(url.into());
        self
    }

    pub fn credentials(mut self, access_key: impl Into<String>, secret_key: impl Into<String>) -> Self {
        self.access_key = access_key.into();
        self.secret_key = secret_key.into();
        self
    }

    pub fn region(mut self, region: impl Into<String>) -> Self {
        self.region = region.into();
        self
    }

    pub fn build(self) -> Post3Client {
        let creds = Credentials::new(
            &self.access_key,
            &self.secret_key,
            None,
            None,
            "post3-sdk",
        );

        let mut config = aws_sdk_s3::Config::builder()
            .behavior_version_latest()
            .region(aws_types::region::Region::new(self.region))
            .credentials_provider(creds)
            .force_path_style(true);

        if let Some(url) = self.endpoint_url {
            config = config.endpoint_url(url);
        }

        Post3Client {
            inner: Client::from_conf(config.build()),
        }
    }
}

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

    #[test]
    fn test_construct_client() {
        let client = Post3Client::new("http://localhost:9000");
        // Verify we can access the inner client
        let _inner = client.inner();
    }

    #[test]
    fn test_builder_custom_creds() {
        let client = Post3Client::builder()
            .endpoint_url("http://localhost:9000")
            .credentials("my-access-key", "my-secret-key")
            .region("eu-west-1")
            .build();
        let _inner = client.inner();
    }
}