use std::sync::Arc;
use std::time::Duration;

use async_trait::async_trait;
use bytes::Bytes;
use futures::TryStreamExt;
use http::Method;
use object_store::ObjectStoreExt;
use object_store::aws::{AmazonS3, AmazonS3Builder};
use object_store::path::Path;
use object_store::prefix::PrefixStore;
use object_store::signer::Signer;
use object_store::{GetResult, ObjectMeta, ObjectStore, PutResult};
use url::Url;

use crate::config::StorageSettings;
use crate::repo::{StorageError, StorageRepository};

#[derive(Debug)]
pub struct S3Storage {
    store: Arc<PrefixStore<AmazonS3>>,
}

impl S3Storage {
    pub fn from_config(settings: &StorageSettings) -> Result<Self, StorageError> {
        let s3 = settings.get_s3();
        let bucket = s3.get_bucket().clone();
        let region = s3.get_region().clone();
        if bucket.trim().is_empty() {
            return Err(StorageError::Config(
                "s3.bucket must not be empty".to_string(),
            ));
        }
        if region.trim().is_empty() {
            return Err(StorageError::Config(
                "s3.region must not be empty".to_string(),
            ));
        }

        let mut builder = AmazonS3Builder::new()
            .with_bucket_name(bucket)
            .with_region(region);
        if let Some(endpoint) = s3.get_endpoint() {
            builder = builder.with_endpoint(endpoint.clone());
        }
        if let Some(access_key_id) = s3.get_access_key_id() {
            builder = builder.with_access_key_id(access_key_id.clone());
        }
        if let Some(secret_access_key) = s3.get_secret_access_key() {
            builder = builder.with_secret_access_key(secret_access_key.clone());
        }
        if *s3.get_allow_http() {
            builder = builder.with_allow_http(true);
        }

        let store = builder
            .build()
            .map_err(|e| StorageError::Config(format!("failed to initialize s3 storage: {e}")))?;
        let store = match s3.get_prefix() {
            Some(prefix) => PrefixStore::new(store, prefix.clone()),
            None => PrefixStore::new(store, ""),
        };
        Ok(Self {
            store: Arc::new(store),
        })
    }
}

#[async_trait]
impl StorageRepository for S3Storage {
    async fn put(&self, path: &Path, data: Bytes) -> Result<PutResult, StorageError> {
        Ok(self.store.put(path, data.into()).await?)
    }

    async fn get(&self, path: &Path) -> Result<GetResult, StorageError> {
        Ok(self.store.get(path).await?)
    }

    async fn delete(&self, path: &Path) -> Result<(), StorageError> {
        Ok(self.store.delete(path).await?)
    }

    async fn head(&self, path: &Path) -> Result<ObjectMeta, StorageError> {
        Ok(self.store.head(path).await?)
    }

    async fn exists(&self, path: &Path) -> Result<bool, StorageError> {
        Ok(self.store.head(path).await.is_ok())
    }

    async fn list(&self, prefix: Option<&Path>) -> Result<Vec<ObjectMeta>, StorageError> {
        let stream = self.store.list(prefix);
        Ok(stream.try_collect::<Vec<_>>().await?)
    }

    async fn copy(&self, from: &Path, to: &Path) -> Result<(), StorageError> {
        Ok(self.store.copy(from, to).await?)
    }

    async fn presigned_upload(
        &self,
        path: &Path,
        expires_in: Duration,
    ) -> Result<Url, StorageError> {
        Ok(self.store.signed_url(Method::PUT, path, expires_in).await?)
    }

    async fn presigned_download(
        &self,
        path: &Path,
        expires_in: Duration,
    ) -> Result<Url, StorageError> {
        Ok(self.store.signed_url(Method::GET, path, expires_in).await?)
    }
}