use super::Options;
use crate::storage::{Error, models};
use async_trait::async_trait;
use bytes::Bytes;
use futures_util::StreamExt;
use object_store::aws::{AmazonS3, AmazonS3Builder};
use object_store::{MultipartUpload, ObjectStore, PutPayload};
use pbbson::Model;
use pbbson::bson::spec::BinarySubtype;
use pbbson::bson::{self, Binary, Bson};
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 {
    store: AmazonS3,
}

impl S3ObjectStore {
    pub async fn new(url: &Url, _options: Options) -> Result<Self, Error> {
        let url_str = url.to_string();
        let store = {
            // 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_name = parts.remove(0);
                let url_str = format!("{}://{}/{}", url.scheme(), parts.join("."), url.path());
                let mut builder = AmazonS3Builder::new()
                    .with_region("US")
                    .with_endpoint(url_str)
                    .with_bucket_name(bucket_name);
                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.with_access_key_id(access_key);
                    builder = builder.with_secret_access_key(secret_key);
                }
                builder.build().map_err(|e| Error::invalid_argument(e.to_string()))?
            } else {
                return Err(Error::invalid_argument("Unsupported URL"));
            }
        };
        Ok(Self { store })
    }
}

#[async_trait]
impl crate::storage::object_store::ObjectStore for S3ObjectStore {
    async fn create(&self, metadata: Model, path: &Path) -> Result<Model, Error> {
        // Validate
        let mut metadata = metadata.clone();
        if !metadata.contains_field("filename") {
            return Err(Error::invalid_argument("Metadata requires a filename"));
        }

        // Get or assign id
        let id = match metadata.id() {
            Ok(id) => id,
            _ => {
                let id = xid::new().to_string();
                metadata.insert("id", id.clone());
                id
            }
        };

        // Store file
        let location = object_store::path::Path::from(id.as_str());
        let mut file = std::fs::File::open(path).map_err(|e| Error::internal(e.to_string()))?;
        let mut uploader = self
            .store
            .put_multipart(&location)
            .await
            .map_err(|e| Error::internal(e.to_string()))?;
        let mut buf = vec![0x8; 5_000_000];
        let mut total_bytes_read = 0;
        while let Ok(bytes_read) = file.read(&mut buf) {
            if bytes_read == 0 {
                break;
            }
            total_bytes_read += bytes_read;
            let data = PutPayload::from(Bytes::from(buf[..bytes_read].to_vec()));
            uploader
                .put_part(data)
                .await
                .map_err(|e| Error::internal(e.to_string()))?;
        }
        uploader.complete().await.map_err(|e| Error::internal(e.to_string()))?;
        metadata.insert("length".to_string(), Bson::Int64(total_bytes_read as i64));
        if !metadata.contains_field("md5") {
            metadata.insert(
                "md5",
                Bson::Binary(Binary {
                    subtype: BinarySubtype::Md5,
                    bytes: md5::compute(&buf).to_vec(),
                }),
            );
        }

        // Store metadata
        let location = object_store::path::Path::from(format!("{id}.json").as_str());
        let metadata_as_json = models::to_json(&metadata, false)?;
        let data = serde_json::to_vec(&metadata_as_json).map_err(|e| Error::internal(e.to_string()))?;
        let mut uploader = self
            .store
            .put_multipart(&location)
            .await
            .map_err(|e| Error::internal(e.to_string()))?;
        let data = PutPayload::from(Bytes::from(data));
        uploader
            .put_part(data)
            .await
            .map_err(|e| Error::internal(e.to_string()))?;
        uploader.complete().await.map_err(|e| Error::internal(e.to_string()))?;

        Ok(metadata)
    }

    async fn delete(&self, id: &str, _by_account_id: Option<String>) -> Result<(), Error> {
        let location = object_store::path::Path::from(id);
        self.store
            .delete(&location)
            .await
            .map_err(|e| Error::internal(e.to_string()))?;
        let location = object_store::path::Path::from(format!("{id}.json").as_str());
        self.store
            .delete(&location)
            .await
            .map_err(|e| Error::internal(e.to_string()))?;
        Ok(())
    }

    async fn download(&self, id: &str) -> Result<NamedTempFile, Error> {
        let location = object_store::path::Path::from(id);
        let res = self
            .store
            .get(&location)
            .await
            .map_err(|e| Error::internal(e.to_string()))?;
        let mut stream = res.into_stream();
        let mut file = NamedTempFile::new().map_err(|e| Error::internal(e.to_string()))?;
        while let Some(maybe_bytes) = stream.next().await {
            let bytes = maybe_bytes.map_err(|e| Error::internal(e.to_string()))?;
            let _ = file.write(&bytes).map_err(|e| Error::internal(e.to_string()))?;
        }
        Ok(file)
    }

    async fn find(&self, id: &str) -> Result<Model, Error> {
        let location = object_store::path::Path::from(format!("{id}.json").as_str());
        let res = match self.store.get(&location).await {
            Ok(res) => res,
            Err(object_store::Error::NotFound { path, source: _ }) => return Err(Error::not_found(path.to_string())),
            Err(e) => return Err(Error::internal(e.to_string())),
        };
        let data = res.bytes().await.map_err(|e| Error::internal(e.to_string()))?;
        let json_data: serde_json::Value = serde_json::from_slice(&data).map_err(|e| Error::internal(e.to_string()))?;
        let mut doc = bson::serialize_to_document(&json_data).map_err(|e| Error::internal(e.to_string()))?;
        doc.insert("id", Bson::String(id.to_string()));
        Ok(Model::from(doc))
    }

    async fn find_many(&self, _filter: Model, _ordering: Option<Model>) -> Result<Vec<Model>, Error> {
        let mut stream = self.store.list(None);
        let mut metadatas = Vec::new();
        while let Some(Ok(object_meta)) = stream.next().await {
            let filename = match object_meta.location.filename() {
                None => continue,
                Some(filename) => filename,
            };
            if !filename.ends_with(".json") {
                continue;
            }
            if let Some(id) = filename.strip_suffix(".json") {
                let model = self.find(id).await?;
                metadatas.push(model);
            }
        }
        Ok(metadatas)
    }

    async fn update(&self, _metadata: Model) -> Result<Model, Error> {
        Err(Error::unimplemented("NIY")) // TODO
    }
}
