alien_bindings/providers/storage/

gcp_gcs.rs

use crate::providers::storage::credential_bridge::GcpCredentialBridge;
use crate::providers::utils::{prefixed_path, relativize_path};
use crate::{
    error::{Error, ErrorData},
    presigned::{PresignedOperation, PresignedRequest, PresignedRequestBackend},
    traits::{Binding, Storage},
};
use alien_error::{AlienError, Context, IntoAlienError};
use async_trait::async_trait;
use bytes::Bytes;
use chrono::Utc;
use futures::stream::BoxStream;
use futures::TryStreamExt as _;
use object_store::signer::Signer;
use object_store::{
    gcp::GoogleCloudStorage, path::Path, Attribute, Attributes, GetOptions, GetResult, ListResult,
    ObjectMeta, ObjectStore, PutMultipartOptions, PutOptions, PutPayload, PutResult,
    Result as ObjectStoreResult,
};
use reqwest::Method;
use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;
use url::Url;

/// Google Cloud Storage implementation.
#[derive(Debug)]
pub struct GcsStorage {
    url: Url,
    base_dir: Path,
    inner: GoogleCloudStorage,
}

impl GcsStorage {
    /// Creates a new `GcsStorage` instance from bucket configuration.
    ///
    /// Uses GCP config for credentials.
    pub fn new(
        bucket_name: String,
        gcp_config: &alien_core::GcpClientConfig,
    ) -> Result<Self, Error> {
        let gcs_url = format!("gs://{}", bucket_name);
        let url = Url::parse(&gcs_url).into_alien_error().context(
            ErrorData::InvalidConfigurationUrl {
                url: gcs_url.clone(),
                reason: "Invalid GCS URL format".to_string(),
            },
        )?;

        // Build the store with credentials bridged from GcpClientConfig.
        // For presigned URLs, object_store needs signing credentials (private key), so
        // provide the service account key directly when that credential mode is used.
        let credentials = GcpCredentialBridge::new(gcp_config.clone());
        let mut builder = object_store::gcp::GoogleCloudStorageBuilder::new()
            .with_bucket_name(&bucket_name)
            .with_credentials(Arc::new(credentials));

        if let alien_core::GcpCredentials::ServiceAccountKey { json } = &gcp_config.credentials {
            builder = builder.with_service_account_key(json);
        }

        let store = builder
            .build()
            .into_alien_error()
            .context(ErrorData::BindingSetupFailed {
                binding_type: "GCP GCS storage".to_string(),
                reason: format!("Failed to build GCS client for bucket: {}", bucket_name),
            })?;

        // Extract the base path from the URL path segments, handling the None case.
        let base_dir = match url.path_segments() {
            Some(segments) => Path::from_iter(segments.filter(|s| !s.is_empty())),
            None => Path::default(), // Use an empty path if there are no segments
        };

        Ok(Self {
            url,
            base_dir,
            inner: store,
        })
    }
}

impl Binding for GcsStorage {}

fn validate_put_attributes(attributes: &Attributes) -> ObjectStoreResult<()> {
    let gzip_encoding = attributes
        .get(&Attribute::ContentEncoding)
        .is_some_and(|value| value.as_ref().eq_ignore_ascii_case("gzip"));
    if gzip_encoding {
        return Err(object_store::Error::Generic {
            store: "GcsStorage",
            source: Box::new(AlienError::new(ErrorData::OperationNotSupported {
                operation: "storage.put contentEncoding=gzip".to_string(),
                reason: "GCS decompressive transcoding is incompatible with object reads"
                    .to_string(),
            })),
        });
    }
    Ok(())
}

#[async_trait]
impl Storage for GcsStorage {
    fn get_base_dir(&self) -> Path {
        self.base_dir.clone()
    }

    fn get_url(&self) -> Url {
        self.url.clone()
    }

    async fn presigned_put(
        &self,
        path: &Path,
        expires_in: Duration,
    ) -> crate::error::Result<PresignedRequest> {
        // Note: Presigned URLs require signing credentials (private key).
        // This works with ServiceAccountKey but may fail with AccessToken,
        // ServiceMetadata, or ProjectedServiceAccount credentials.
        let dst = prefixed_path(&self.base_dir, path);
        let signed_url = self
            .inner
            .signed_url(Method::PUT, &dst, expires_in)
            .await
            .into_alien_error()
            .context(ErrorData::StorageOperationFailed {
                binding_name: "gcp-gcs".to_string(),
                operation: format!("generate presigned PUT URL for {}", path),
            })?;

        let headers = HashMap::new();

        Ok(PresignedRequest {
            backend: PresignedRequestBackend::Http {
                url: signed_url.to_string(),
                method: "PUT".to_string(),
                headers,
            },
            expiration: Utc::now()
                + chrono::Duration::from_std(expires_in).map_err(|e| {
                    AlienError::new(ErrorData::Other {
                        message: format!("Invalid duration: {}", e),
                    })
                })?,
            operation: PresignedOperation::Put,
            path: path.to_string(),
        })
    }

    async fn presigned_get(
        &self,
        path: &Path,
        expires_in: Duration,
    ) -> crate::error::Result<PresignedRequest> {
        let dst = prefixed_path(&self.base_dir, path);
        let signed_url = self
            .inner
            .signed_url(Method::GET, &dst, expires_in)
            .await
            .into_alien_error()
            .context(ErrorData::StorageOperationFailed {
                binding_name: "gcp-gcs".to_string(),
                operation: format!("generate presigned GET URL for {}", path),
            })?;

        let headers = HashMap::new();

        Ok(PresignedRequest {
            backend: PresignedRequestBackend::Http {
                url: signed_url.to_string(),
                method: "GET".to_string(),
                headers,
            },
            expiration: Utc::now()
                + chrono::Duration::from_std(expires_in).map_err(|e| {
                    AlienError::new(ErrorData::Other {
                        message: format!("Invalid duration: {}", e),
                    })
                })?,
            operation: PresignedOperation::Get,
            path: path.to_string(),
        })
    }

    async fn presigned_delete(
        &self,
        path: &Path,
        expires_in: Duration,
    ) -> crate::error::Result<PresignedRequest> {
        let dst = prefixed_path(&self.base_dir, path);
        let signed_url = self
            .inner
            .signed_url(Method::DELETE, &dst, expires_in)
            .await
            .into_alien_error()
            .context(ErrorData::StorageOperationFailed {
                binding_name: "gcp-gcs".to_string(),
                operation: format!("generate presigned DELETE URL for {}", path),
            })?;

        let headers = HashMap::new();

        Ok(PresignedRequest {
            backend: PresignedRequestBackend::Http {
                url: signed_url.to_string(),
                method: "DELETE".to_string(),
                headers,
            },
            expiration: Utc::now()
                + chrono::Duration::from_std(expires_in).map_err(|e| {
                    AlienError::new(ErrorData::Other {
                        message: format!("Invalid duration: {}", e),
                    })
                })?,
            operation: PresignedOperation::Delete,
            path: path.to_string(),
        })
    }
}

// Delegate ObjectStore trait implementation to the inner store,
// prefixing paths with the base_dir.
#[async_trait]
impl ObjectStore for GcsStorage {
    async fn put(&self, location: &Path, payload: PutPayload) -> ObjectStoreResult<PutResult> {
        let dst = prefixed_path(&self.base_dir, location);
        self.inner.put(&dst, payload).await
    }

    async fn put_opts(
        &self,
        location: &Path,
        payload: PutPayload,
        opts: PutOptions,
    ) -> ObjectStoreResult<PutResult> {
        let dst = prefixed_path(&self.base_dir, location);
        validate_put_attributes(&opts.attributes)?;
        self.inner.put_opts(&dst, payload, opts).await
    }

    async fn put_multipart(
        &self,
        location: &Path,
    ) -> ObjectStoreResult<Box<dyn object_store::MultipartUpload>> {
        let dst = prefixed_path(&self.base_dir, location);
        self.inner.put_multipart(&dst).await
    }

    async fn put_multipart_opts(
        &self,
        location: &Path,
        opts: PutMultipartOptions,
    ) -> ObjectStoreResult<Box<dyn object_store::MultipartUpload>> {
        let dst = prefixed_path(&self.base_dir, location);
        validate_put_attributes(&opts.attributes)?;
        self.inner.put_multipart_opts(&dst, opts).await
    }

    async fn get(&self, location: &Path) -> ObjectStoreResult<GetResult> {
        let src = prefixed_path(&self.base_dir, location);
        self.inner.get(&src).await
    }

    async fn get_opts(&self, location: &Path, options: GetOptions) -> ObjectStoreResult<GetResult> {
        let src = prefixed_path(&self.base_dir, location);
        self.inner.get_opts(&src, options).await
    }

    async fn get_range(
        &self,
        location: &Path,
        range: std::ops::Range<u64>,
    ) -> ObjectStoreResult<Bytes> {
        let src = prefixed_path(&self.base_dir, location);
        self.inner.get_range(&src, range).await
    }

    async fn head(&self, location: &Path) -> ObjectStoreResult<ObjectMeta> {
        let src = prefixed_path(&self.base_dir, location);
        let mut meta = self.inner.head(&src).await?;
        meta.location = relativize_path(&self.base_dir, meta.location, "GcpStorage")?;
        Ok(meta)
    }

    async fn delete(&self, location: &Path) -> ObjectStoreResult<()> {
        let src = prefixed_path(&self.base_dir, location);
        self.inner.delete(&src).await
    }

    fn list(&self, prefix: Option<&Path>) -> BoxStream<'static, ObjectStoreResult<ObjectMeta>> {
        let list_prefix_for_inner = prefix
            .map(|p| prefixed_path(&self.base_dir, p))
            .unwrap_or_else(|| self.base_dir.clone());

        let base_dir_for_stream = self.base_dir.clone();

        Box::pin(
            self.inner
                .list(Some(&list_prefix_for_inner))
                .and_then(move |mut meta| {
                    let captured_base_dir = base_dir_for_stream.clone();
                    async move {
                        meta.location =
                            relativize_path(&captured_base_dir, meta.location, "GcpStorage")?;
                        Ok(meta)
                    }
                }),
        )
    }

    async fn list_with_delimiter(&self, prefix: Option<&Path>) -> ObjectStoreResult<ListResult> {
        let list_prefix_for_inner = prefix
            .map(|p| prefixed_path(&self.base_dir, p))
            .unwrap_or_else(|| self.base_dir.clone());
        let mut result = self
            .inner
            .list_with_delimiter(Some(&list_prefix_for_inner))
            .await?;

        for meta_obj in &mut result.objects {
            let original_location = std::mem::take(&mut meta_obj.location);
            meta_obj.location = relativize_path(&self.base_dir, original_location, "GcpStorage")?;
        }

        let mut new_common_prefixes = Vec::with_capacity(result.common_prefixes.len());
        for cp in result.common_prefixes {
            new_common_prefixes.push(relativize_path(&self.base_dir, cp, "GcpStorage")?);
        }
        result.common_prefixes = new_common_prefixes;

        Ok(result)
    }

    async fn copy(&self, from: &Path, to: &Path) -> ObjectStoreResult<()> {
        let src = prefixed_path(&self.base_dir, from);
        let dst = prefixed_path(&self.base_dir, to);
        self.inner.copy(&src, &dst).await
    }

    async fn copy_if_not_exists(&self, from: &Path, to: &Path) -> ObjectStoreResult<()> {
        let src = prefixed_path(&self.base_dir, from);
        let dst = prefixed_path(&self.base_dir, to);
        self.inner.copy_if_not_exists(&src, &dst).await
    }
}

impl std::fmt::Display for GcsStorage {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "GcpStorage(url={})", self.url)
    }
}

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

    #[test]
    fn rejects_gzip_content_encoding_before_upload() {
        for encoding in ["gzip", "GZip"] {
            let attributes = Attributes::from_iter([(
                Attribute::ContentEncoding,
                AttributeValue::from(encoding),
            )]);

            let error = validate_put_attributes(&attributes)
                .expect_err("gzip would make the object unreadable through this backend");

            assert!(matches!(error, object_store::Error::Generic { .. }));
            assert!(error.to_string().contains("decompressive transcoding"));
        }
    }

    #[test]
    fn permits_non_transcoded_content_encoding() {
        let attributes =
            Attributes::from_iter([(Attribute::ContentEncoding, AttributeValue::from("br"))]);

        validate_put_attributes(&attributes)
            .expect("Brotli objects are served without transcoding");
    }
}