use super::Options;
use crate::storage::Error;
use async_trait::async_trait;
use opendal::Operator;
use opendal::layers::RetryLayer;
use opendal::services::S3;
use std::io::{Read, Write};
use std::{env, path::Path};
use tempfile::NamedTempFile;
use url::Url;

const AWS_ACCESS_KEY_ID: &str = "AWS_ACCESS_KEY_ID";
const AWS_SECRET_ACCESS_KEY_ID: &str = "AWS_SECRET_ACCESS_KEY_ID";

pub struct S3ObjectStore {
    op: Operator,
}

impl S3ObjectStore {
    pub async fn new(url: &Url, _options: Options) -> Result<Self, Error> {
        let url_str = url.to_string();
        let op = {
            // Digital Ocean support
            if url_str.contains(".digitaloceanspaces.com") {
                let host = match url.host() {
                    Some(host) => host.to_string(),
                    None => return Err(Error::invalid_argument("Expected a host part of the URL".to_string())),
                };
                let mut parts: Vec<_> = host.split(".").map(str::to_string).collect();
                let bucket = parts.remove(0);
                let endpoint = format!("{}://{host}", url.scheme());
                let mut builder = S3::default()
                    .region("US")
                    .endpoint(&endpoint)
                    .bucket(&bucket)
                    .root(url.path());
                let (access_key, secret_key) = {
                    if !url.username().is_empty() && url.password().is_some() {
                        let access_key = url.username().to_string();
                        let secret_key = url.password().unwrap_or_default().to_string();
                        (Some(access_key), Some(secret_key))
                    } else if let (Ok(access_key), Ok(secret_key)) =
                        (env::var(AWS_ACCESS_KEY_ID), env::var(AWS_SECRET_ACCESS_KEY_ID))
                    {
                        (Some(access_key), Some(secret_key))
                    } else {
                        (None, None)
                    }
                };
                if let (Some(access_key), Some(secret_key)) = (access_key, secret_key) {
                    builder = builder.access_key_id(&access_key);
                    builder = builder.secret_access_key(&secret_key);
                }
                Operator::new(builder)?.layer(RetryLayer::new()).finish()
            } else {
                return Err(Error::invalid_argument("Unsupported URL"));
            }
        };
        Ok(Self { op })
    }
}

#[async_trait]
impl crate::storage::object_store::ObjectStore for S3ObjectStore {
    async fn del(&self, key: &str) -> Result<(), Error> {
        self.op.delete(key).await?;
        Ok(())
    }

    async fn get(&self, key: &str) -> Result<NamedTempFile, Error> {
        let buf = self.op.read(key).await?;
        let mut file = NamedTempFile::new().map_err(|e| Error::internal(e.to_string()))?;
        file.write_all(buf.to_bytes().as_ref())
            .map_err(|e| Error::internal(e.to_string()))?;
        Ok(file)
    }

    async fn put(&self, key: &str, path: &Path) -> Result<(), Error> {
        let mut file = std::fs::File::open(path).map_err(|e| Error::internal(e.to_string()))?;
        let mut buf = vec![];
        file.read_to_end(&mut buf).map_err(|e| Error::internal(e.to_string()))?;
        self.op.write(key, buf).await?;
        Ok(())
    }
}

impl From<opendal::Error> for Error {
    fn from(e: opendal::Error) -> Self {
        match e.kind() {
            opendal::ErrorKind::NotFound => Error::not_found(e.to_string()),
            opendal::ErrorKind::Unsupported => Error::unimplemented(e.to_string()),
            _ => Error::internal(e.to_string()),
        }
    }
}
