haste_document_storage/providers/gcs/
mod.rs

1//! Google Cloud Storage backend, speaking the JSON API over HTTPS.
2
3mod api;
4mod auth;
5mod http;
6
7use std::time::{Duration, SystemTime, UNIX_EPOCH};
8
9use bytes::{Bytes, BytesMut};
10use futures_util::TryStreamExt as _;
11use percent_encoding::{AsciiSet, NON_ALPHANUMERIC, utf8_percent_encode};
12use reqwest::{Client, Method, RequestBuilder, Response, header::CONTENT_TYPE};
13use time::{OffsetDateTime, format_description::well_known::Rfc3339};
14
15pub use auth::GcsCredentials;
16
17use self::{
18    api::{Object, ObjectList, UploadMetadata},
19    auth::Authenticator,
20    http::{check, parse_json, transport},
21};
22use crate::{
23    error::StorageError,
24    retry::with_retry,
25    traits::{DocumentStorage, StorageFuture},
26    types::{
27        Document, DocumentBody, DocumentKey, DocumentMetadata, ListPage, ListRequest,
28        StoredDocument,
29    },
30};
31
32const DEFAULT_ENDPOINT: &str = "https://storage.googleapis.com";
33const DEFAULT_CONTENT_TYPE: &str = "application/octet-stream";
34const CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
35/// Longest pause tolerated while waiting for response bytes.
36const READ_TIMEOUT: Duration = Duration::from_secs(60);
37
38/// Everything but RFC 3986 unreserved characters is percent-encoded in object
39/// names and query values, so `/` becomes `%2F` as the API requires.
40const ENCODE_SET: &AsciiSet = &NON_ALPHANUMERIC
41    .remove(b'-')
42    .remove(b'.')
43    .remove(b'_')
44    .remove(b'~');
45
46/// Settings for [`GcsDocumentStorage`].
47#[derive(Debug, Clone)]
48pub struct GcsConfig {
49    pub bucket: String,
50    /// Folder-like prefix placed before every key, so one bucket can serve
51    /// several deployments.
52    pub prefix: Option<String>,
53    pub credentials: GcsCredentials,
54    /// API root, `https://storage.googleapis.com` unless set. Point it at an
55    /// emulator for local development.
56    pub endpoint: Option<String>,
57}
58
59/// [`DocumentStorage`] over one Google Cloud Storage bucket.
60pub struct GcsDocumentStorage {
61    client: Client,
62    auth: Authenticator,
63    bucket: String,
64    /// Prefix without surrounding slashes; empty when unset.
65    prefix: String,
66    /// `{endpoint}/storage/v1/b/{bucket}/o`
67    objects_url: String,
68    /// `{endpoint}/upload/storage/v1/b/{bucket}/o?uploadType=multipart`
69    upload_url: String,
70}
71
72impl GcsDocumentStorage {
73    /// Builds the client and obtains a first access token, so rejected
74    /// credentials fail here rather than on the first document.
75    ///
76    /// # Errors
77    ///
78    /// Returns [`StorageError::Configuration`] for an unusable configuration or
79    /// credentials file, and [`StorageError::Unauthorized`] when the token
80    /// endpoint rejects the credentials.
81    pub async fn connect(config: GcsConfig) -> Result<Self, StorageError> {
82        if config.bucket.is_empty() {
83            return Err(StorageError::Configuration(
84                "bucket name is empty".to_string(),
85            ));
86        }
87        let client = Client::builder()
88            .connect_timeout(CONNECT_TIMEOUT)
89            .read_timeout(READ_TIMEOUT)
90            .build()
91            .map_err(|error| {
92                StorageError::Configuration(format!("could not build HTTP client: {error}"))
93            })?;
94        let auth = Authenticator::new(&config.credentials).await?;
95        let endpoint = config
96            .endpoint
97            .as_deref()
98            .unwrap_or(DEFAULT_ENDPOINT)
99            .trim_end_matches('/');
100        let bucket = config.bucket;
101
102        let storage = Self {
103            client,
104            auth,
105            prefix: config
106                .prefix
107                .as_deref()
108                .map(|prefix| prefix.trim_matches('/'))
109                .unwrap_or_default()
110                .to_string(),
111            objects_url: format!("{endpoint}/storage/v1/b/{bucket}/o"),
112            upload_url: format!("{endpoint}/upload/storage/v1/b/{bucket}/o?uploadType=multipart"),
113            bucket,
114        };
115        storage.auth.bearer(&storage.client).await?;
116        Ok(storage)
117    }
118
119    /// The object name for a key, with the deployment prefix applied.
120    fn object_name(&self, key: &str) -> String {
121        if self.prefix.is_empty() {
122            key.to_string()
123        } else {
124            format!("{}/{key}", self.prefix)
125        }
126    }
127
128    fn object_url(&self, name: &str) -> String {
129        format!("{}/{}", self.objects_url, encode(name))
130    }
131
132    async fn request(&self, method: Method, url: &str) -> Result<RequestBuilder, StorageError> {
133        let request = self.client.request(method, url);
134        Ok(match self.auth.bearer(&self.client).await? {
135            Some(token) => request.bearer_auth(token),
136            None => request,
137        })
138    }
139
140    async fn send(&self, request: RequestBuilder, subject: &str) -> Result<Response, StorageError> {
141        check(request.send().await.map_err(transport)?, subject).await
142    }
143
144    async fn fetch_object(&self, key: &DocumentKey) -> Result<Object, StorageError> {
145        let url = self.object_url(&self.object_name(key.as_str()));
146        // Shared references are `Copy`, so every retry attempt can move them
147        // into its own future. The other operations follow the same pattern.
148        let (url, key) = (&url, key.as_str());
149        with_retry("head", || async move {
150            parse_json(
151                self.send(self.request(Method::GET, url).await?, key)
152                    .await?,
153            )
154            .await
155        })
156        .await
157    }
158
159    fn metadata_of(&self, object: Object) -> DocumentMetadata {
160        let key = if self.prefix.is_empty() {
161            object.name.as_str()
162        } else {
163            object
164                .name
165                .strip_prefix(&self.prefix)
166                .and_then(|rest| rest.strip_prefix('/'))
167                .unwrap_or(&object.name)
168        };
169        DocumentMetadata {
170            key: DocumentKey::from_backend(key.to_string()),
171            size: object
172                .size
173                .as_deref()
174                .and_then(|size| size.parse().ok())
175                .unwrap_or(0),
176            content_type: object.content_type,
177            etag: object.etag,
178            last_modified: object
179                .updated
180                .as_deref()
181                .and_then(|updated| OffsetDateTime::parse(updated, &Rfc3339).ok()),
182            metadata: object.metadata,
183        }
184    }
185}
186
187impl DocumentStorage for GcsDocumentStorage {
188    fn put<'a>(
189        &'a self,
190        key: &'a DocumentKey,
191        document: Document,
192    ) -> StorageFuture<'a, DocumentMetadata> {
193        Box::pin(async move {
194            let content_type = document
195                .content_type
196                .as_deref()
197                .unwrap_or(DEFAULT_CONTENT_TYPE);
198            if !content_type
199                .bytes()
200                .all(|byte| byte.is_ascii_graphic() || byte == b' ')
201            {
202                return Err(StorageError::InvalidDocument(
203                    "content type must be printable ASCII".to_string(),
204                ));
205            }
206
207            let name = self.object_name(key.as_str());
208            let metadata = serde_json::to_vec(&UploadMetadata {
209                name: &name,
210                content_type: document.content_type.as_deref(),
211                metadata: &document.metadata,
212            })
213            .map_err(|error| {
214                StorageError::InvalidDocument(format!("metadata is not serializable: {error}"))
215            })?;
216            let boundary = boundary();
217            let body = multipart_related(&boundary, &metadata, content_type, &document.body);
218
219            let (body, boundary, key) = (&body, &boundary, key.as_str());
220            let object: Object = with_retry("put", || async move {
221                let request = self
222                    .request(Method::POST, &self.upload_url)
223                    .await?
224                    .header(
225                        CONTENT_TYPE,
226                        format!("multipart/related; boundary={boundary}"),
227                    )
228                    .body(body.clone());
229                parse_json(self.send(request, key).await?).await
230            })
231            .await?;
232            Ok(self.metadata_of(object))
233        })
234    }
235
236    fn head<'a>(&'a self, key: &'a DocumentKey) -> StorageFuture<'a, DocumentMetadata> {
237        Box::pin(async move { Ok(self.metadata_of(self.fetch_object(key).await?)) })
238    }
239
240    fn get<'a>(&'a self, key: &'a DocumentKey) -> StorageFuture<'a, StoredDocument> {
241        Box::pin(async move {
242            let object = self.fetch_object(key).await?;
243            let url = match &object.generation {
244                Some(generation) => format!(
245                    "{}?alt=media&generation={generation}",
246                    self.object_url(&object.name)
247                ),
248                None => format!("{}?alt=media", self.object_url(&object.name)),
249            };
250
251            let (url, key) = (&url, key.as_str());
252            let response = with_retry("get", || async move {
253                self.send(self.request(Method::GET, url).await?, key).await
254            })
255            .await?;
256
257            Ok(StoredDocument {
258                metadata: self.metadata_of(object),
259                body: DocumentBody::from_stream(response.bytes_stream().map_err(transport)),
260            })
261        })
262    }
263
264    fn delete<'a>(&'a self, key: &'a DocumentKey) -> StorageFuture<'a, ()> {
265        Box::pin(async move {
266            let url = self.object_url(&self.object_name(key.as_str()));
267            let (url, key) = (&url, key.as_str());
268            let result = with_retry("delete", || async move {
269                self.send(self.request(Method::DELETE, url).await?, key)
270                    .await
271            })
272            .await;
273            match result {
274                Ok(_) | Err(StorageError::NotFound(_)) => Ok(()),
275                Err(error) => Err(error),
276            }
277        })
278    }
279
280    fn list(&self, request: ListRequest) -> StorageFuture<'_, ListPage> {
281        Box::pin(async move {
282            let prefix = self.object_name(request.prefix.as_deref().unwrap_or_default());
283            let mut query = vec![format!("prefix={}", encode(&prefix))];
284            if let Some(token) = &request.page_token {
285                query.push(format!("pageToken={}", encode(token)));
286            }
287            if let Some(size) = request.page_size {
288                query.push(format!("maxResults={size}"));
289            }
290            let url = format!("{}?{}", self.objects_url, query.join("&"));
291
292            let url = &url;
293            let page: ObjectList = with_retry("list", || async move {
294                parse_json(
295                    self.send(self.request(Method::GET, url).await?, &self.bucket)
296                        .await?,
297                )
298                .await
299            })
300            .await?;
301
302            Ok(ListPage {
303                documents: page
304                    .items
305                    .into_iter()
306                    .map(|object| self.metadata_of(object))
307                    .collect(),
308                next_page_token: page.next_page_token,
309            })
310        })
311    }
312}
313
314fn encode(value: &str) -> String {
315    utf8_percent_encode(value, ENCODE_SET).to_string()
316}
317
318/// A `multipart/related` upload body: the JSON metadata part, then the bytes.
319fn multipart_related(boundary: &str, metadata: &[u8], content_type: &str, body: &[u8]) -> Bytes {
320    let mut out = BytesMut::with_capacity(metadata.len() + body.len() + 256);
321    out.extend_from_slice(
322        format!("--{boundary}\r\nContent-Type: application/json; charset=UTF-8\r\n\r\n").as_bytes(),
323    );
324    out.extend_from_slice(metadata);
325    out.extend_from_slice(
326        format!("\r\n--{boundary}\r\nContent-Type: {content_type}\r\n\r\n").as_bytes(),
327    );
328    out.extend_from_slice(body);
329    out.extend_from_slice(format!("\r\n--{boundary}--\r\n").as_bytes());
330    out.freeze()
331}
332
333/// A part boundary whoever produced the document could not have predicted,
334/// so it cannot occur inside the body.
335fn boundary() -> String {
336    let nanos = SystemTime::now()
337        .duration_since(UNIX_EPOCH)
338        .map(|elapsed| elapsed.as_nanos())
339        .unwrap_or_default();
340    format!("haste-document-{}-{nanos:x}", std::process::id())
341}
haste_document_storage/providers/gcs/mod.rs

haste_document_storage/providers/gcs/
mod.rs
