Skip to main content

max / makenotwork

15.2 KB · 483 lines History Blame Raw
1 //! The S3 client itself: its handle, the operations it delegates, and its
2 //! implementation of [`super::StorageBackend`].
3 //!
4 //! The trait impl and the inherent methods it delegates to are two halves of
5 //! one contract, which is why they stay in one file.
6
7 use super::backend::StorageBackend;
8 use super::bucket::PRESIGN_EXPIRY_SECS;
9 use super::bucket::S3DeleteAuthority;
10 use super::key::S3Key;
11 use crate::config::StorageConfig;
12 use crate::error::{AppError, Result};
13
14 /// S3 client wrapper for presigned URL operations.
15 /// Delegates S3 operations to `s3_storage::S3Client`.
16 #[derive(Clone)]
17 pub struct S3Client {
18 inner: s3_storage::S3Client,
19 }
20
21 impl S3Client {
22 /// Create a new S3 client from storage configuration.
23 ///
24 /// Configures CORS on the bucket at startup so browser PUT uploads to
25 /// presigned URLs work without manual bucket configuration.
26 pub async fn new(config: &StorageConfig, host_url: &str) -> Result<Self> {
27 let s3_config = s3_storage::S3Config {
28 endpoint: config.endpoint.clone(),
29 bucket: config.bucket.clone(),
30 access_key: config.access_key.clone(),
31 secret_key: config.secret_key.clone(),
32 region: config.region.clone(),
33 };
34
35 let inner = s3_storage::S3Client::new(&s3_config)
36 .await
37 .map_err(AppError::Storage)?;
38
39 inner.configure_cors(host_url).await;
40
41 Ok(S3Client { inner })
42 }
43
44 /// Generate a presigned URL for uploading a file. `max_bytes`, when set,
45 /// binds `Content-Length` into the signature, S3 will reject any PUT
46 /// whose actual body length differs from `max_bytes`.
47 pub async fn presign_upload(
48 &self,
49 s3_key: &S3Key,
50 content_type: &str,
51 expiry_secs: Option<u64>,
52 cache_control: Option<&str>,
53 max_bytes: Option<i64>,
54 ) -> Result<String> {
55 self.inner
56 .presign_upload(
57 s3_key.as_str(),
58 content_type,
59 expiry_secs.unwrap_or(PRESIGN_EXPIRY_SECS),
60 cache_control,
61 max_bytes,
62 )
63 .await
64 .map_err(AppError::Storage)
65 }
66
67 /// Generate a presigned URL for downloading/streaming a file
68 pub async fn presign_download(
69 &self,
70 s3_key: &S3Key,
71 expiry_secs: Option<u64>,
72 ) -> Result<String> {
73 self.inner
74 .presign_download(s3_key.as_str(), expiry_secs.unwrap_or(PRESIGN_EXPIRY_SECS))
75 .await
76 .map_err(AppError::Storage)
77 }
78
79 /// Check if an object exists in S3
80 pub async fn object_exists(&self, s3_key: &str) -> Result<bool> {
81 self.inner
82 .object_exists(s3_key)
83 .await
84 .map_err(AppError::Storage)
85 }
86
87 /// Get the size of an object in S3 (bytes), or None if not found.
88 pub async fn object_size(&self, s3_key: &str) -> Result<Option<i64>> {
89 self.inner
90 .object_size(s3_key)
91 .await
92 .map_err(AppError::Storage)
93 }
94
95 /// Download an object's bytes from S3
96 pub async fn download_object(&self, s3_key: &str) -> Result<Vec<u8>> {
97 self.inner
98 .download(s3_key)
99 .await
100 .map(|(bytes, _content_type)| bytes)
101 .map_err(AppError::Storage)
102 }
103
104 /// Download an object as `bytes::Bytes` without the `to_vec` copy. See trait docs.
105 pub async fn download_object_buf(&self, s3_key: &str) -> Result<bytes::Bytes> {
106 self.inner
107 .download_buf(s3_key)
108 .await
109 .map(|(bytes, _content_type)| bytes)
110 .map_err(AppError::Storage)
111 }
112
113 /// Stream an object's body from S3 without buffering. See trait docs.
114 pub async fn download_stream(&self, s3_key: &str) -> Result<s3_storage::ByteStream> {
115 self.inner
116 .download_stream(s3_key)
117 .await
118 .map_err(AppError::Storage)
119 }
120
121 /// Read the first `len` bytes via a ranged S3 GET (for content sniffing).
122 pub async fn download_head(&self, s3_key: &str, len: usize) -> Result<Vec<u8>> {
123 self.inner
124 .download_head(s3_key, len)
125 .await
126 .map_err(AppError::Storage)
127 }
128
129 /// Upload an object to S3 from bytes
130 pub async fn upload_object(
131 &self,
132 s3_key: &S3Key,
133 content_type: &str,
134 data: Vec<u8>,
135 cache_control: Option<&str>,
136 ) -> Result<()> {
137 self.inner
138 .upload(s3_key.as_str(), content_type, data, cache_control)
139 .await
140 .map_err(AppError::Storage)
141 }
142
143 /// Delete an object from S3
144 pub async fn delete_object(&self, s3_key: &S3Key) -> Result<()> {
145 self.inner
146 .delete(s3_key.as_str())
147 .await
148 .map_err(AppError::Storage)
149 }
150
151 /// Server-side copy within the bucket. See the [`StorageBackend::copy_object`]
152 /// trait method for the scan-then-promote rationale.
153 pub async fn copy_object(&self, src_key: &S3Key, dst_key: &S3Key) -> Result<()> {
154 self.inner
155 .copy_object(src_key.as_str(), dst_key.as_str())
156 .await
157 .map_err(AppError::Storage)
158 }
159
160 /// Server-side copy from `src_bucket` into this client's bucket. See the
161 /// [`StorageBackend::copy_object_from`] trait method for the cross-bucket
162 /// promote rationale.
163 pub async fn copy_object_from(
164 &self,
165 src_bucket: &str,
166 src_key: &S3Key,
167 dst_key: &S3Key,
168 ) -> Result<()> {
169 self.inner
170 .copy_object_from(src_bucket, src_key.as_str(), dst_key.as_str())
171 .await
172 .map_err(AppError::Storage)
173 }
174
175 /// Batched S3 delete (`DeleteObjects`, up to 1000 keys per call).
176 /// Chunks larger slices into 1000-key batches and logs per-key failures
177 /// without bubbling, the pending_s3_deletions queue is the safety net.
178 pub async fn delete_objects(&self, keys: &[S3Key]) -> Result<()> {
179 if keys.is_empty() {
180 return Ok(());
181 }
182 for chunk in keys.chunks(1000) {
183 let chunk: Vec<String> = chunk.iter().map(|k| k.as_str().to_string()).collect();
184 match self.inner.delete_objects(&chunk).await {
185 Ok(failures) => {
186 for (k, msg) in &failures {
187 tracing::warn!(key = %k, error = %msg, "S3 delete_objects: key-level failure");
188 }
189 // A whole-batch failure must not read as success (Run #2
190 // Storage MINOR); partial failures stay logged and the
191 // pending_s3_deletions queue is the retry net.
192 if !chunk.is_empty() && failures.len() == chunk.len() {
193 return Err(AppError::Storage(format!(
194 "S3 delete_objects: all {} keys in batch failed",
195 chunk.len()
196 )));
197 }
198 }
199 Err(e) => return Err(AppError::Storage(e)),
200 }
201 }
202 Ok(())
203 }
204
205 /// Upload a file to S3 using multipart upload (10 MB parts).
206 pub async fn upload_multipart(
207 &self,
208 s3_key: &S3Key,
209 content_type: &str,
210 file_path: &std::path::Path,
211 ) -> Result<()> {
212 self.inner
213 .upload_multipart(s3_key.as_str(), content_type, file_path, None)
214 .await
215 .map_err(AppError::Storage)
216 }
217
218 /// Server-side multipart copy for sources over the 5 GiB single-part
219 /// `CopyObject` limit. See the [`StorageBackend::copy_object_multipart`]
220 /// trait method for the promote rationale. `part_size` of `None` lets the
221 /// storage layer auto-size parts for `src_size`.
222 pub async fn copy_object_multipart(
223 &self,
224 src_bucket: &str,
225 src_key: &S3Key,
226 dst_key: &S3Key,
227 content_type: &str,
228 src_size: u64,
229 part_size: Option<usize>,
230 ) -> Result<()> {
231 self.inner
232 .copy_object_multipart(
233 src_bucket,
234 src_key.as_str(),
235 dst_key.as_str(),
236 content_type,
237 src_size,
238 part_size,
239 )
240 .await
241 .map_err(AppError::Storage)
242 }
243
244 /// Begin a client-direct multipart upload. See the
245 /// [`StorageBackend::create_multipart_upload`] trait method.
246 pub async fn create_multipart_upload(
247 &self,
248 s3_key: &S3Key,
249 content_type: &str,
250 ) -> Result<String> {
251 self.inner
252 .create_multipart_upload(s3_key.as_str(), content_type)
253 .await
254 .map_err(AppError::Storage)
255 }
256
257 /// Presign one `UploadPart` request. See the
258 /// [`StorageBackend::presign_upload_part`] trait method.
259 pub async fn presign_upload_part(
260 &self,
261 s3_key: &S3Key,
262 upload_id: &str,
263 part_number: i32,
264 expiry_secs: Option<u64>,
265 max_bytes: Option<i64>,
266 checksum_sha256: Option<&str>,
267 ) -> Result<String> {
268 self.inner
269 .presign_upload_part(
270 s3_key.as_str(),
271 upload_id,
272 part_number,
273 expiry_secs.unwrap_or(PRESIGN_EXPIRY_SECS),
274 max_bytes,
275 checksum_sha256,
276 )
277 .await
278 .map_err(AppError::Storage)
279 }
280
281 /// Complete a multipart upload. See the
282 /// [`StorageBackend::complete_multipart_upload`] trait method.
283 pub async fn complete_multipart_upload(
284 &self,
285 s3_key: &S3Key,
286 upload_id: &str,
287 parts: &[(i32, String)],
288 ) -> Result<()> {
289 self.inner
290 .complete_multipart_upload(s3_key.as_str(), upload_id, parts)
291 .await
292 .map_err(AppError::Storage)
293 }
294
295 /// Abort a multipart upload. See the
296 /// [`StorageBackend::abort_multipart_upload`] trait method.
297 pub async fn abort_multipart_upload(&self, s3_key: &S3Key, upload_id: &str) -> Result<()> {
298 self.inner
299 .abort_multipart_upload(s3_key.as_str(), upload_id)
300 .await
301 .map_err(AppError::Storage)
302 }
303
304 /// In-progress multipart sessions for a key. See the
305 /// [`StorageBackend::list_multipart_uploads_for_key`] trait method.
306 pub async fn list_multipart_uploads_for_key(&self, s3_key: &str) -> Result<Vec<String>> {
307 self.inner
308 .list_multipart_uploads_for_key(s3_key)
309 .await
310 .map_err(AppError::Storage)
311 }
312
313 /// Lightweight connectivity check, issues a list with max_keys(0).
314 pub async fn check_connectivity(&self) -> std::result::Result<(), String> {
315 self.inner.check_connectivity().await
316 }
317 }
318
319 #[async_trait::async_trait]
320 impl StorageBackend for S3Client {
321 async fn presign_upload(
322 &self,
323 s3_key: &S3Key,
324 content_type: &str,
325 expiry_secs: Option<u64>,
326 cache_control: Option<&str>,
327 max_bytes: Option<i64>,
328 ) -> Result<String> {
329 self.presign_upload(s3_key, content_type, expiry_secs, cache_control, max_bytes)
330 .await
331 }
332
333 async fn presign_download(&self, s3_key: &S3Key, expiry_secs: Option<u64>) -> Result<String> {
334 self.presign_download(s3_key, expiry_secs).await
335 }
336
337 async fn object_exists(&self, s3_key: &str) -> Result<bool> {
338 self.object_exists(s3_key).await
339 }
340
341 async fn object_size(&self, s3_key: &str) -> Result<Option<i64>> {
342 self.object_size(s3_key).await
343 }
344
345 async fn download_object(&self, s3_key: &str) -> Result<Vec<u8>> {
346 self.download_object(s3_key).await
347 }
348
349 async fn download_object_buf(&self, s3_key: &str) -> Result<bytes::Bytes> {
350 self.download_object_buf(s3_key).await
351 }
352
353 async fn download_stream(&self, s3_key: &str) -> Result<s3_storage::ByteStream> {
354 self.download_stream(s3_key).await
355 }
356
357 async fn download_head(&self, s3_key: &str, len: usize) -> Result<Vec<u8>> {
358 self.download_head(s3_key, len).await
359 }
360
361 async fn upload_object(
362 &self,
363 s3_key: &S3Key,
364 content_type: &str,
365 data: Vec<u8>,
366 cache_control: Option<&str>,
367 ) -> Result<()> {
368 self.upload_object(s3_key, content_type, data, cache_control)
369 .await
370 }
371
372 async fn delete_object(&self, _auth: &S3DeleteAuthority, s3_key: &S3Key) -> Result<()> {
373 // Authority proven by the caller; delegate to the inherent impl.
374 self.delete_object(s3_key).await
375 }
376
377 async fn delete_objects(&self, _auth: &S3DeleteAuthority, keys: &[S3Key]) -> Result<()> {
378 self.delete_objects(keys).await
379 }
380
381 async fn copy_object_from(
382 &self,
383 src_bucket: &str,
384 src_key: &S3Key,
385 dst_key: &S3Key,
386 ) -> Result<()> {
387 // Inherent method; delegate.
388 self.copy_object_from(src_bucket, src_key, dst_key).await
389 }
390
391 async fn copy_object(&self, src_key: &S3Key, dst_key: &S3Key) -> Result<()> {
392 // Inherent method; delegate.
393 self.copy_object(src_key, dst_key).await
394 }
395
396 async fn delete_prefix(&self, _auth: &S3DeleteAuthority, prefix: &str) -> Result<()> {
397 self.inner
398 .delete_prefix(prefix)
399 .await
400 .map_err(AppError::Storage)
401 }
402
403 async fn upload_multipart(
404 &self,
405 s3_key: &S3Key,
406 content_type: &str,
407 file_path: &std::path::Path,
408 ) -> Result<()> {
409 self.upload_multipart(s3_key, content_type, file_path).await
410 }
411
412 async fn copy_object_multipart(
413 &self,
414 src_bucket: &str,
415 src_key: &S3Key,
416 dst_key: &S3Key,
417 content_type: &str,
418 src_size: u64,
419 part_size: Option<usize>,
420 ) -> Result<()> {
421 // Inherent method; delegate.
422 self.copy_object_multipart(
423 src_bucket,
424 src_key,
425 dst_key,
426 content_type,
427 src_size,
428 part_size,
429 )
430 .await
431 }
432
433 async fn create_multipart_upload(&self, s3_key: &S3Key, content_type: &str) -> Result<String> {
434 self.create_multipart_upload(s3_key, content_type).await
435 }
436
437 async fn presign_upload_part(
438 &self,
439 s3_key: &S3Key,
440 upload_id: &str,
441 part_number: i32,
442 expiry_secs: Option<u64>,
443 max_bytes: Option<i64>,
444 checksum_sha256: Option<&str>,
445 ) -> Result<String> {
446 self.presign_upload_part(
447 s3_key,
448 upload_id,
449 part_number,
450 expiry_secs,
451 max_bytes,
452 checksum_sha256,
453 )
454 .await
455 }
456
457 async fn complete_multipart_upload(
458 &self,
459 s3_key: &S3Key,
460 upload_id: &str,
461 parts: &[(i32, String)],
462 ) -> Result<()> {
463 self.complete_multipart_upload(s3_key, upload_id, parts)
464 .await
465 }
466
467 async fn abort_multipart_upload(&self, s3_key: &S3Key, upload_id: &str) -> Result<()> {
468 self.abort_multipart_upload(s3_key, upload_id).await
469 }
470
471 async fn list_multipart_uploads_for_key(&self, s3_key: &str) -> Result<Vec<String>> {
472 self.list_multipart_uploads_for_key(s3_key).await
473 }
474
475 async fn check_connectivity(&self) -> std::result::Result<(), String> {
476 self.check_connectivity().await
477 }
478
479 fn bucket(&self) -> &str {
480 self.inner.bucket()
481 }
482 }
483