kellnr_storage/
gcs_storage.rs

1use std::time::Duration;
2
3use async_trait::async_trait;
4use bytes::Bytes;
5use futures_util::StreamExt;
6use kellnr_settings::Settings;
7use object_store::gcp::{GoogleCloudStorage, GoogleCloudStorageBuilder};
8use object_store::path::Path;
9use object_store::{ClientOptions, ObjectStore, ObjectStoreExt, PutMode};
10
11use crate::storage::{ObjectWithMeta, Storage};
12use crate::storage_error::StorageError;
13
14pub struct GCSStorage(GoogleCloudStorage);
15
16#[async_trait]
17impl Storage for GCSStorage {
18    async fn get(&self, key: &str) -> Result<Bytes, StorageError> {
19        self.storage()
20            .get(&Self::try_path_from(key)?)
21            .await?
22            .bytes()
23            .await
24            .map_err(StorageError::from)
25    }
26
27    async fn get_with_meta(&self, key: &str) -> Result<ObjectWithMeta, StorageError> {
28        let result = self.storage().get(&Self::try_path_from(key)?).await?;
29        let e_tag = result.meta.e_tag.clone();
30        let last_modified = result.meta.last_modified;
31        let bytes = result.bytes().await?;
32        Ok(ObjectWithMeta {
33            bytes,
34            e_tag,
35            last_modified,
36        })
37    }
38
39    async fn put(&self, key: &str, object: Bytes) -> Result<(), StorageError> {
40        self.storage()
41            .put_opts(
42                &Self::try_path_from(key)?,
43                object.into(),
44                PutMode::Create.into(),
45            )
46            .await?;
47        Ok(())
48    }
49
50    async fn put_overwrite(&self, key: &str, object: Bytes) -> Result<(), StorageError> {
51        self.storage()
52            .put_opts(
53                &Self::try_path_from(key)?,
54                object.into(),
55                PutMode::Overwrite.into(),
56            )
57            .await?;
58        Ok(())
59    }
60
61    async fn delete(&self, key: &str) -> Result<(), StorageError> {
62        let path = Self::try_path_from(key)?;
63        self.storage().delete(&path).await?;
64        Ok(())
65    }
66
67    async fn exists(&self, key: &str) -> Result<bool, StorageError> {
68        let path = Self::try_path_from(key)?;
69        self.storage()
70            .head(&path)
71            .await
72            .map(|_| true)
73            .or_else(|e| match e {
74                object_store::Error::NotFound { .. } => Ok(false),
75                _ => Err(StorageError::from(e)),
76            })
77    }
78
79    async fn size(&self, key: &str) -> Result<u64, StorageError> {
80        let path = Self::try_path_from(key)?;
81        Ok(self.storage().head(&path).await?.size)
82    }
83
84    async fn list(&self, prefix: &str) -> Result<Vec<String>, StorageError> {
85        let prefix = if prefix.is_empty() {
86            None
87        } else {
88            Some(Self::try_path_from(prefix)?)
89        };
90        let mut stream = self.storage().list(prefix.as_ref());
91        let mut keys = Vec::new();
92        while let Some(meta) = stream.next().await {
93            keys.push(meta?.location.to_string());
94        }
95        Ok(keys)
96    }
97}
98
99impl GCSStorage {
100    fn try_path_from(key: &str) -> Result<Path, object_store::path::Error> {
101        Path::from_url_path(key)
102    }
103
104    fn storage(&self) -> &GoogleCloudStorage {
105        &self.0
106    }
107}
108
109impl TryFrom<(&str, &Settings)> for GCSStorage {
110    type Error = StorageError;
111
112    fn try_from((bucket, settings): (&str, &Settings)) -> Result<Self, Self::Error> {
113        // NOTE: `with_client_options` replaces the builder's internal ClientOptions entirely
114        // (see the same note in `s3_storage.rs`). `GoogleCloudStorageBuilder::default()`
115        // sets `allow_http(true)`, so overriding it here without carrying the setting over
116        // would make any `http://` endpoint override (e.g. fake-gcs-server in tests) fail
117        // instantly with a client-side "builder error" (reqwest's scheme check), before any
118        // request is even sent.
119        let client_options = ClientOptions::new()
120            .with_connect_timeout(Duration::from_secs(settings.gcs.connect_timeout_seconds))
121            .with_timeout(Duration::from_secs(settings.gcs.request_timeout_seconds))
122            .with_allow_http(settings.gcs.allow_http);
123
124        let mut gcs = if settings.gcs.skip_signature {
125            // Unauthenticated target (e.g. fake-gcs-server in tests). `from_env()` would
126            // still perform real Application Default Credentials discovery (gcloud ADC file
127            // / metadata server) inside `build()`, and `with_skip_signature` alone doesn't
128            // skip the OAuth2 token fetch on the PUT path (object_store 0.14's PUT request
129            // path doesn't consult `skip_signature`, only GET/DELETE/HEAD do). A
130            // `disable_oauth` service account key makes every credential provider return a
131            // static empty bearer token with no network calls at all, which is what
132            // object_store's own test suite uses against fake-gcs-server.
133            const FAKE_SERVICE_ACCOUNT_KEY: &str = r#"{"private_key":"unused","private_key_id":"unused","client_email":"unused@example.com","disable_oauth":true}"#;
134            GoogleCloudStorageBuilder::new()
135                .with_service_account_key(FAKE_SERVICE_ACCOUNT_KEY)
136                .with_skip_signature(true)
137        } else {
138            GoogleCloudStorageBuilder::from_env()
139        };
140        if let Some(endpoint) = &settings.gcs.endpoint {
141            gcs = gcs.with_base_url(endpoint);
142        }
143        let gcs = gcs
144            .with_bucket_name(bucket)
145            .with_client_options(client_options);
146        Ok(Self(gcs.build()?))
147    }
148}
kellnr_storage/gcs_storage.rs

kellnr_storage/
gcs_storage.rs
