//! S3-compatible blob store (cloud target). The same `BlobStore` contract
//! as the local filesystem implementation; MinIO serves it in tests and
//! self-hosted installs, AWS S3 in managed clouds.

use object_store::aws::AmazonS3;
use object_store::path::Path as ObjectPath;
use object_store::{ObjectStore, PutPayload};

use crate::{BlobError, BlobStore};

pub struct S3BlobStore {
    store: AmazonS3,
}

impl S3BlobStore {
    /// `endpoint` is the S3 API base (http allowed for in-cluster MinIO);
    /// credentials come from deployment secrets.
    pub fn connect(
        endpoint: &str,
        bucket: &str,
        access_key: &str,
        secret_key: &str,
    ) -> Result<S3BlobStore, BlobError> {
        let store = object_store::aws::AmazonS3Builder::new()
            .with_endpoint(endpoint)
            .with_allow_http(true)
            .with_bucket_name(bucket)
            .with_access_key_id(access_key)
            .with_secret_access_key(secret_key)
            .with_region("us-east-1")
            // MinIO and most self-hosted S3s require path-style addressing.
            .with_virtual_hosted_style_request(false)
            .build()
            .map_err(|e| BlobError::Io(e.to_string()))?;
        Ok(S3BlobStore { store })
    }

    fn key(key: &str) -> Result<ObjectPath, BlobError> {
        if key.is_empty() || key.split('/').any(|part| part.is_empty() || part == "..") {
            return Err(BlobError::InvalidKey(key.to_owned()));
        }
        ObjectPath::parse(key).map_err(|e| BlobError::InvalidKey(e.to_string()))
    }
}

#[async_trait::async_trait]
impl BlobStore for S3BlobStore {
    async fn put(&self, key: &str, bytes: &[u8]) -> Result<(), BlobError> {
        let path = Self::key(key)?;
        self.store
            .put(&path, PutPayload::from_bytes(bytes.to_vec().into()))
            .await
            .map_err(|e| BlobError::Io(e.to_string()))?;
        Ok(())
    }

    async fn get(&self, key: &str) -> Result<Vec<u8>, BlobError> {
        let path = Self::key(key)?;
        match self.store.get(&path).await {
            Ok(result) => Ok(result
                .bytes()
                .await
                .map_err(|e| BlobError::Io(e.to_string()))?
                .to_vec()),
            Err(object_store::Error::NotFound { .. }) => Err(BlobError::NotFound),
            Err(e) => Err(BlobError::Io(e.to_string())),
        }
    }

    async fn delete(&self, key: &str) -> Result<(), BlobError> {
        let path = Self::key(key)?;
        // object_store's S3 delete is idempotent; the drive UX wants an
        // honest NotFound, so probe first (head).
        match self.store.head(&path).await {
            Ok(_) => {}
            Err(object_store::Error::NotFound { .. }) => return Err(BlobError::NotFound),
            Err(e) => return Err(BlobError::Io(e.to_string())),
        }
        self.store
            .delete(&path)
            .await
            .map_err(|e| BlobError::Io(e.to_string()))
    }
}