rskit_storage_s3/
store.rs

1//! S3 / MinIO storage backend implementing [`rskit_storage::FileStore`].
2
3use std::collections::HashMap;
4use std::fmt;
5use std::future::Future;
6use std::pin::Pin;
7use std::sync::Arc;
8use std::time::Duration;
9
10use rskit_errors::{AppError, AppResult, ErrorCode};
11use rskit_storage::FileSource;
12use rskit_storage::store::{
13    FileStore, StorageConfig, StorageFactory, StorageRegistry, StoredFile, UploadOptions,
14    content_type_or_default, prefixed_key,
15};
16use rskit_util::env;
17use serde::{Deserialize, Serialize};
18use tokio::sync::OnceCell;
19
20/// Configuration for the S3 store.
21///
22/// Supports AWS S3 and S3-compatible services (MinIO, LocalStack, etc.) via `endpoint`,
23/// `force_path_style`, and explicit credentials.
24#[derive(Clone, Deserialize, Serialize)]
25pub struct Config {
26    /// S3 bucket name.
27    pub bucket: String,
28    /// AWS region (e.g., `"us-east-1"`). Falls back to `AWS_REGION`
29    /// or `AWS_DEFAULT_REGION` env vars if unset.
30    pub region: Option<String>,
31    /// Custom endpoint URL for S3-compatible services (e.g., `"http://localhost:9000"` for MinIO).
32    pub endpoint: Option<String>,
33    /// Key prefix applied to all objects (e.g., `"uploads"`).
34    pub prefix: Option<String>,
35    /// Use path-style access (`http://host/bucket/key`) instead of virtual-hosted-style (`http://bucket.host/key`).
36    /// Required for MinIO and most S3-compatible services.
37    #[serde(default)]
38    pub force_path_style: bool,
39    /// Explicit AWS access key ID. Falls back to `AWS_ACCESS_KEY_ID` env var.
40    pub access_key_id: Option<String>,
41    /// Explicit AWS secret access key. Falls back to `AWS_SECRET_ACCESS_KEY` env var.
42    pub secret_access_key: Option<String>,
43}
44
45impl fmt::Debug for Config {
46    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
47        f.debug_struct("Config")
48            .field("bucket", &self.bucket)
49            .field("region", &self.region)
50            .field("endpoint", &self.endpoint)
51            .field("prefix", &self.prefix)
52            .field("force_path_style", &self.force_path_style)
53            .field(
54                "access_key_id",
55                &self.access_key_id.as_ref().map(|_| "<redacted>"),
56            )
57            .field(
58                "secret_access_key",
59                &self.secret_access_key.as_ref().map(|_| "<redacted>"),
60            )
61            .finish()
62    }
63}
64
65/// Amazon S3 / MinIO storage backend.
66///
67/// Created by registering [`Config`] with the storage registry.
68/// Implements [`FileStore`] for use with any rskit-storage consumer.
69struct S3Store {
70    client: OnceCell<aws_sdk_s3::Client>,
71    builder: S3ClientBuilder,
72    config: Config,
73}
74
75type S3ClientFuture = Pin<Box<dyn Future<Output = AppResult<aws_sdk_s3::Client>> + Send>>;
76type S3ClientBuilder = Box<dyn Fn() -> S3ClientFuture + Send + Sync>;
77
78impl S3Store {
79    /// Create a new S3 store from config.
80    ///
81    /// Resolves credentials from config fields, then env vars. The client is lazily constructed
82    /// and reused for all operations.
83    fn new(config: Config) -> AppResult<Self> {
84        let client_config = client_config_builder(&config)?.build();
85        let builder = Box::new(move || {
86            let client_config = client_config.clone();
87            Box::pin(async move { Ok(aws_sdk_s3::Client::from_conf(client_config)) })
88                as S3ClientFuture
89        });
90        Ok(Self {
91            client: OnceCell::new(),
92            builder,
93            config,
94        })
95    }
96
97    fn full_key(&self, key: &str) -> String {
98        prefixed_key(self.config.prefix.as_deref(), key)
99    }
100
101    async fn client(&self) -> AppResult<&aws_sdk_s3::Client> {
102        self.client.get_or_try_init(|| (self.builder)()).await
103    }
104}
105
106/// Build the AWS SDK configuration for a store from validated [`Config`].
107///
108/// Kept separate from [`S3Store::new`]
109/// so tests can attach a wire-level mock HTTP client to the same builder the production path uses.
110fn client_config_builder(config: &Config) -> AppResult<aws_sdk_s3::config::Builder> {
111    let (access_key, secret_key) = resolve_credentials(config)?;
112
113    let creds = aws_sdk_s3::config::Credentials::new(
114        &access_key,
115        &secret_key,
116        None,
117        None,
118        "rskit-storage-s3",
119    );
120
121    let mut builder = aws_sdk_s3::Config::builder()
122        .credentials_provider(creds)
123        .behavior_version_latest();
124
125    // Region: config → AWS_REGION → AWS_DEFAULT_REGION
126    if let Some(region) = &config.region {
127        builder = builder.region(aws_sdk_s3::config::Region::new(region.clone()));
128    } else if let Some(region) = env::get_non_empty("AWS_REGION") {
129        builder = builder.region(aws_sdk_s3::config::Region::new(region));
130    } else if let Some(region) = env::get_non_empty("AWS_DEFAULT_REGION") {
131        builder = builder.region(aws_sdk_s3::config::Region::new(region));
132    }
133
134    if let Some(endpoint) = &config.endpoint {
135        builder = builder.endpoint_url(endpoint);
136    }
137
138    if config.force_path_style {
139        builder = builder.force_path_style(true);
140    }
141
142    Ok(builder)
143}
144
145#[async_trait::async_trait]
146impl FileStore for S3Store {
147    async fn upload(
148        &self,
149        source: &FileSource,
150        key: &str,
151        options: UploadOptions,
152    ) -> AppResult<StoredFile> {
153        let data = source.read_all().await?;
154        let size = data.len() as u64;
155        let full_key = self.full_key(key);
156        let client = self.client().await?;
157
158        let mut req = client
159            .put_object()
160            .bucket(&self.config.bucket)
161            .key(&full_key)
162            .body(data.to_vec().into());
163
164        req = req.content_type(content_type_or_default(options.content_type()));
165        for (k, v) in &options.metadata {
166            req = req.metadata(k, v);
167        }
168
169        req.send()
170            .await
171            .map_err(|e| AppError::new(ErrorCode::Internal, format!("S3 upload failed: {e}")))?;
172
173        let content_type = options.content_type().map(str::to_owned);
174        Ok(uploaded_file(
175            key,
176            size,
177            content_type.as_deref(),
178            options.metadata,
179        ))
180    }
181
182    async fn download(&self, key: &str) -> AppResult<FileSource> {
183        let full_key = self.full_key(key);
184        let client = self.client().await?;
185        let resp = client
186            .get_object()
187            .bucket(&self.config.bucket)
188            .key(&full_key)
189            .send()
190            .await
191            .map_err(|e| AppError::new(ErrorCode::NotFound, format!("S3 download failed: {e}")))?;
192
193        let data =
194            resp.body.collect().await.map_err(|e| {
195                AppError::new(ErrorCode::Internal, format!("S3 read body failed: {e}"))
196            })?;
197
198        Ok(FileSource::Bytes(bytes::Bytes::from(data.to_vec())))
199    }
200
201    async fn delete(&self, key: &str) -> AppResult<()> {
202        let full_key = self.full_key(key);
203        let client = self.client().await?;
204        client
205            .delete_object()
206            .bucket(&self.config.bucket)
207            .key(&full_key)
208            .send()
209            .await
210            .map_err(|e| AppError::new(ErrorCode::Internal, format!("S3 delete failed: {e}")))?;
211        Ok(())
212    }
213
214    async fn exists(&self, key: &str) -> AppResult<bool> {
215        let full_key = self.full_key(key);
216        let client = self.client().await?;
217        match client
218            .head_object()
219            .bucket(&self.config.bucket)
220            .key(&full_key)
221            .send()
222            .await
223        {
224            Ok(_) => Ok(true),
225            Err(_) => Ok(false),
226        }
227    }
228
229    async fn head(&self, key: &str) -> AppResult<StoredFile> {
230        let full_key = self.full_key(key);
231        let client = self.client().await?;
232        let resp = client
233            .head_object()
234            .bucket(&self.config.bucket)
235            .key(&full_key)
236            .send()
237            .await
238            .map_err(|e| AppError::new(ErrorCode::NotFound, format!("S3 head failed: {e}")))?;
239
240        Ok(stored_file_from_head(key, &resp))
241    }
242
243    async fn list(&self, prefix: &str, limit: Option<usize>) -> AppResult<Vec<StoredFile>> {
244        let full_prefix = self.full_key(prefix);
245        let client = self.client().await?;
246        let mut req = client
247            .list_objects_v2()
248            .bucket(&self.config.bucket)
249            .prefix(&full_prefix);
250
251        if let Some(max) = limit {
252            req = req.max_keys(max as i32);
253        }
254
255        let resp = req
256            .send()
257            .await
258            .map_err(|e| AppError::new(ErrorCode::Internal, format!("S3 list failed: {e}")))?;
259
260        let items = resp
261            .contents()
262            .iter()
263            .map(stored_file_from_object)
264            .collect();
265
266        Ok(items)
267    }
268
269    async fn presigned_url(&self, key: &str, expires_in: Duration) -> AppResult<String> {
270        let full_key = self.full_key(key);
271        let presigning_config = aws_sdk_s3::presigning::PresigningConfig::expires_in(expires_in)
272            .map_err(|e| {
273                AppError::new(
274                    ErrorCode::InvalidInput,
275                    format!("Invalid presigning duration: {e}"),
276                )
277            })?;
278
279        let presigned = self
280            .client()
281            .await?
282            .get_object()
283            .bucket(&self.config.bucket)
284            .key(&full_key)
285            .presigned(presigning_config)
286            .await
287            .map_err(|e| {
288                AppError::new(
289                    ErrorCode::Internal,
290                    format!("S3 presigned URL generation failed: {e}"),
291                )
292            })?;
293
294        Ok(presigned.uri().to_string())
295    }
296
297    async fn copy(&self, from_key: &str, to_key: &str) -> AppResult<StoredFile> {
298        let full_from = self.full_key(from_key);
299        let full_to = self.full_key(to_key);
300        let client = self.client().await?;
301
302        client
303            .copy_object()
304            .bucket(&self.config.bucket)
305            .copy_source(format!("{}/{}", self.config.bucket, full_from))
306            .key(&full_to)
307            .send()
308            .await
309            .map_err(|e| AppError::new(ErrorCode::Internal, format!("S3 copy failed: {e}")))?;
310
311        self.head(to_key).await
312    }
313
314    async fn rename(&self, from_key: &str, to_key: &str) -> AppResult<StoredFile> {
315        let result = self.copy(from_key, to_key).await?;
316        self.delete(from_key).await?;
317        Ok(result)
318    }
319}
320
321fn uploaded_file(
322    key: &str,
323    size: u64,
324    content_type: Option<&str>,
325    metadata: HashMap<String, String>,
326) -> StoredFile {
327    StoredFile::new(prefixed_key(None, key), size, content_type).with_metadata(metadata)
328}
329
330fn stored_file_from_head(
331    key: &str,
332    resp: &aws_sdk_s3::operation::head_object::HeadObjectOutput,
333) -> StoredFile {
334    StoredFile::new(
335        prefixed_key(None, key),
336        resp.content_length().unwrap_or(0) as u64,
337        resp.content_type(),
338    )
339    .with_metadata(
340        resp.metadata()
341            .map(|m| m.iter().map(|(k, v)| (k.clone(), v.clone())).collect())
342            .unwrap_or_default(),
343    )
344}
345
346fn stored_file_from_object(obj: &aws_sdk_s3::types::Object) -> StoredFile {
347    StoredFile::new(
348        obj.key().unwrap_or(""),
349        obj.size().unwrap_or(0) as u64,
350        None,
351    )
352}
353
354/// Resolve AWS credentials from config fields or environment variables.
355fn resolve_credentials(config: &Config) -> AppResult<(String, String)> {
356    resolve_credentials_with(config, env::get_non_empty)
357}
358
359fn resolve_credentials_with(
360    config: &Config,
361    get_env: impl Fn(&str) -> Option<String>,
362) -> AppResult<(String, String)> {
363    if let (Some(key), Some(secret)) = (&config.access_key_id, &config.secret_access_key)
364        && !key.is_empty()
365        && !secret.is_empty()
366    {
367        return Ok((key.clone(), secret.clone()));
368    }
369
370    let key = get_env("AWS_ACCESS_KEY_ID");
371    let secret = get_env("AWS_SECRET_ACCESS_KEY");
372
373    let (Some(key), Some(secret)) = (key, secret) else {
374        return Err(AppError::new(
375            ErrorCode::MissingField,
376            "S3 credentials not found. Set access_key_id/secret_access_key in config \
377             or AWS_ACCESS_KEY_ID/AWS_SECRET_ACCESS_KEY environment variables.",
378        ));
379    };
380
381    Ok((key, secret))
382}
383
384struct S3Factory {
385    config: Config,
386}
387
388#[async_trait::async_trait]
389impl StorageFactory for S3Factory {
390    async fn create(&self, _config: &StorageConfig) -> AppResult<Arc<dyn FileStore>> {
391        Ok(Arc::new(S3Store::new(self.config.clone())?))
392    }
393}
394
395/// Explicitly register the S3 backend in an injected storage registry.
396pub fn register(registry: &mut StorageRegistry, config: Config) -> AppResult<()> {
397    registry.register("s3", Arc::new(S3Factory { config }))
398}
399
400#[cfg(test)]
401mod tests {
402    use super::*;
403
404    #[test]
405    fn config_deserializes_with_defaults() {
406        let json = r#"{"bucket": "test", "endpoint": "http://localhost:9000"}"#;
407        let cfg: Config = serde_json::from_str(json).unwrap();
408        assert_eq!(cfg.bucket, "test");
409        assert!(!cfg.force_path_style);
410        assert!(cfg.access_key_id.is_none());
411        assert!(cfg.region.is_none());
412    }
413
414    #[test]
415    fn config_deserializes_full() {
416        let json = r#"{
417            "bucket": "assets",
418            "region": "us-east-1",
419            "endpoint": "http://minio:9000",
420            "prefix": "uploads",
421            "force_path_style": true,
422            "access_key_id": "minio",
423            "secret_access_key": "minio123"
424        }"#;
425        let cfg: Config = serde_json::from_str(json).unwrap();
426        assert_eq!(cfg.bucket, "assets");
427        assert!(cfg.force_path_style);
428        assert_eq!(cfg.access_key_id.as_deref(), Some("minio"));
429    }
430
431    #[test]
432    fn config_debug_redacts_credentials() {
433        let cfg = Config {
434            bucket: "assets".into(),
435            region: Some("us-east-1".into()),
436            endpoint: Some("https://s3.example.test".into()),
437            prefix: Some("uploads".into()),
438            force_path_style: true,
439            access_key_id: Some("access-key".into()),
440            secret_access_key: Some("secret-key".into()),
441        };
442
443        let debug = format!("{cfg:?}");
444
445        assert!(debug.contains("<redacted>"));
446        assert!(!debug.contains("access-key"));
447        assert!(!debug.contains("secret-key"));
448        assert!(debug.contains("assets"));
449    }
450
451    #[test]
452    fn config_debug_omits_redaction_marker_without_credentials() {
453        let cfg = Config {
454            bucket: "assets".into(),
455            region: None,
456            endpoint: None,
457            prefix: None,
458            force_path_style: false,
459            access_key_id: None,
460            secret_access_key: None,
461        };
462
463        let debug = format!("{cfg:?}");
464
465        assert!(debug.contains("access_key_id: None"));
466        assert!(debug.contains("secret_access_key: None"));
467        assert!(!debug.contains("<redacted>"));
468    }
469
470    #[test]
471    fn resolve_explicit_credentials() {
472        let cfg = Config {
473            bucket: "test".into(),
474            region: None,
475            endpoint: None,
476            prefix: None,
477            force_path_style: false,
478            access_key_id: Some("key123".into()),
479            secret_access_key: Some("secret456".into()),
480        };
481        let (key, secret) = resolve_credentials(&cfg).unwrap();
482        assert_eq!(key, "key123");
483        assert_eq!(secret, "secret456");
484    }
485
486    #[test]
487    fn resolve_blank_explicit_credentials_falls_back_to_error_without_env() {
488        let cfg = Config {
489            bucket: "test".into(),
490            region: None,
491            endpoint: None,
492            prefix: None,
493            force_path_style: false,
494            access_key_id: Some(String::new()),
495            secret_access_key: Some(String::new()),
496        };
497
498        let err = resolve_credentials_with(&cfg, |_| None).unwrap_err();
499
500        assert_eq!(err.code(), ErrorCode::MissingField);
501    }
502
503    #[test]
504    fn resolve_empty_credentials_errors() {
505        let cfg = Config {
506            bucket: "test".into(),
507            region: None,
508            endpoint: None,
509            prefix: None,
510            force_path_style: false,
511            access_key_id: None,
512            secret_access_key: None,
513        };
514        let err = resolve_credentials_with(&cfg, |_| None).unwrap_err();
515
516        assert_eq!(err.code(), ErrorCode::MissingField);
517    }
518
519    #[test]
520    fn resolve_credentials_uses_environment_when_explicit_credentials_absent() {
521        let cfg = Config {
522            bucket: "test".into(),
523            region: None,
524            endpoint: None,
525            prefix: None,
526            force_path_style: false,
527            access_key_id: None,
528            secret_access_key: None,
529        };
530
531        let (key, secret) = resolve_credentials_with(&cfg, |name| match name {
532            "AWS_ACCESS_KEY_ID" => Some("env-key".to_string()),
533            "AWS_SECRET_ACCESS_KEY" => Some("env-secret".to_string()),
534            _ => None,
535        })
536        .unwrap();
537
538        assert_eq!(key, "env-key");
539        assert_eq!(secret, "env-secret");
540    }
541
542    #[test]
543    fn full_key_with_prefix() {
544        let store = S3Store::new(Config {
545            bucket: "b".into(),
546            region: None,
547            endpoint: None,
548            prefix: Some("pfx".into()),
549            force_path_style: false,
550            access_key_id: Some("k".into()),
551            secret_access_key: Some("s".into()),
552        })
553        .unwrap();
554        assert_eq!(store.full_key("file.txt"), "pfx/file.txt");
555    }
556
557    #[test]
558    fn full_key_with_slash_only_prefix() {
559        let store = S3Store::new(Config {
560            bucket: "b".into(),
561            region: None,
562            endpoint: None,
563            prefix: Some("///".into()),
564            force_path_style: false,
565            access_key_id: Some("k".into()),
566            secret_access_key: Some("s".into()),
567        })
568        .unwrap();
569
570        assert_eq!(store.full_key("file.txt"), "file.txt");
571    }
572
573    #[test]
574    fn full_key_without_prefix() {
575        let store = S3Store::new(Config {
576            bucket: "b".into(),
577            region: None,
578            endpoint: None,
579            prefix: None,
580            force_path_style: false,
581            access_key_id: Some("k".into()),
582            secret_access_key: Some("s".into()),
583        })
584        .unwrap();
585        assert_eq!(store.full_key("file.txt"), "file.txt");
586    }
587
588    #[test]
589    fn register_rejects_duplicate_s3_backend() {
590        let mut registry = StorageRegistry::new();
591        let config = Config {
592            bucket: "b".into(),
593            region: None,
594            endpoint: None,
595            prefix: None,
596            force_path_style: false,
597            access_key_id: Some("k".into()),
598            secret_access_key: Some("s".into()),
599        };
600
601        register(&mut registry, config.clone()).unwrap();
602        let err = register(&mut registry, config).unwrap_err();
603
604        assert_eq!(err.code(), ErrorCode::AlreadyExists);
605    }
606
607    #[test]
608    fn store_construction_applies_region_endpoint_and_path_style() {
609        let store = S3Store::new(Config {
610            bucket: "b".into(),
611            region: Some("us-east-1".into()),
612            endpoint: Some("http://127.0.0.1:9000".into()),
613            prefix: Some("uploads".into()),
614            force_path_style: true,
615            access_key_id: Some("k".into()),
616            secret_access_key: Some("s".into()),
617        })
618        .unwrap();
619
620        assert_eq!(store.full_key("file.txt"), "uploads/file.txt");
621    }
622
623    #[test]
624    fn constructs_offline_without_building_client() {
625        let store = S3Store::new(Config {
626            bucket: "b".into(),
627            region: Some("us-east-1".into()),
628            endpoint: Some("http://127.0.0.1:9000".into()),
629            prefix: Some("uploads".into()),
630            force_path_style: true,
631            access_key_id: Some("k".into()),
632            secret_access_key: Some("s".into()),
633        })
634        .unwrap();
635
636        assert_eq!(store.full_key("file.txt"), "uploads/file.txt");
637    }
638
639    #[tokio::test]
640    async fn factory_creates_store_from_explicit_credentials() {
641        let factory = S3Factory {
642            config: Config {
643                bucket: "b".into(),
644                region: Some("us-east-1".into()),
645                endpoint: Some("http://127.0.0.1:9000".into()),
646                prefix: None,
647                force_path_style: true,
648                access_key_id: Some("k".into()),
649                secret_access_key: Some("s".into()),
650            },
651        };
652
653        factory.create(&StorageConfig::default()).await.unwrap();
654    }
655
656    #[tokio::test]
657    async fn presigned_url_validates_duration_and_uses_configured_key_prefix() {
658        let store = S3Store::new(Config {
659            bucket: "bucket".into(),
660            region: Some("us-east-1".into()),
661            endpoint: Some("http://127.0.0.1:9000".into()),
662            prefix: Some("uploads".into()),
663            force_path_style: true,
664            access_key_id: Some("access".into()),
665            secret_access_key: Some("secret".into()),
666        })
667        .unwrap();
668
669        let too_long = store
670            .presigned_url("file.txt", Duration::from_secs(60 * 60 * 24 * 8))
671            .await
672            .unwrap_err();
673        assert_eq!(too_long.code(), ErrorCode::InvalidInput);
674
675        let url = store
676            .presigned_url("file.txt", Duration::from_secs(60))
677            .await
678            .unwrap();
679        assert!(url.contains("uploads/file.txt"));
680    }
681
682    #[test]
683    fn stored_file_mappers_preserve_keys_sizes_content_types_and_metadata() {
684        let mut metadata = HashMap::new();
685        metadata.insert("owner".to_string(), "media".to_string());
686        let uploaded = uploaded_file("file.txt", 4, Some("text/plain"), metadata.clone());
687        assert_eq!(uploaded.key, "file.txt");
688        assert_eq!(uploaded.size, 4);
689        assert_eq!(uploaded.content_type, "text/plain");
690        assert_eq!(uploaded.metadata, metadata);
691
692        let head = aws_sdk_s3::operation::head_object::HeadObjectOutput::builder()
693            .content_length(9)
694            .content_type("application/json")
695            .metadata("trace", "abc")
696            .build();
697        let from_head = stored_file_from_head("meta.json", &head);
698        assert_eq!(from_head.key, "meta.json");
699        assert_eq!(from_head.size, 9);
700        assert_eq!(from_head.content_type, "application/json");
701        assert_eq!(
702            from_head.metadata.get("trace").map(String::as_str),
703            Some("abc")
704        );
705
706        let object = aws_sdk_s3::types::Object::builder()
707            .key("uploads/a.txt")
708            .size(12)
709            .build();
710        let from_object = stored_file_from_object(&object);
711        assert_eq!(from_object.key, "uploads/a.txt");
712        assert_eq!(from_object.size, 12);
713
714        let empty_object = aws_sdk_s3::types::Object::builder().build();
715        let from_empty_object = stored_file_from_object(&empty_object);
716        assert_eq!(from_empty_object.key, "");
717        assert_eq!(from_empty_object.size, 0);
718    }
719
720    // --- Wire-level operation tests -----------------------------------------
721    //
722    // These exercise request construction
723    // and response/error mapping for every `FileStore` operation against an in-process mock HTTP client.
724    // No network, credentials, or live S3 service are involved.
725
726    use aws_smithy_http_client::test_util::{ReplayEvent, StaticReplayClient};
727    use aws_smithy_types::body::SdkBody;
728
729    fn wire_config(prefix: Option<&str>) -> Config {
730        Config {
731            bucket: "test-bucket".into(),
732            region: Some("us-east-1".into()),
733            endpoint: Some("http://s3.local".into()),
734            prefix: prefix.map(str::to_owned),
735            force_path_style: true,
736            access_key_id: Some("test-key".into()),
737            secret_access_key: Some("test-secret".into()),
738        }
739    }
740
741    /// Build a store whose AWS client replays the given responses in order.
742    fn wire_store(config: Config, events: Vec<ReplayEvent>) -> (S3Store, StaticReplayClient) {
743        let http = StaticReplayClient::new(events);
744        let conf = client_config_builder(&config)
745            .unwrap()
746            .http_client(http.clone())
747            .build();
748        (
749            S3Store {
750                client: OnceCell::new_with(Some(aws_sdk_s3::Client::from_conf(conf))),
751                builder: Box::new(|| {
752                    Box::pin(async {
753                        Err(AppError::new(
754                            ErrorCode::Internal,
755                            "test S3 client builder should not be called",
756                        ))
757                    })
758                }),
759                config,
760            },
761            http,
762        )
763    }
764
765    fn ok_response(status: u16, body: impl Into<SdkBody>) -> ReplayEvent {
766        ReplayEvent::new(
767            http::Request::builder().body(SdkBody::empty()).unwrap(),
768            http::Response::builder()
769                .status(status)
770                .body(body.into())
771                .unwrap(),
772        )
773    }
774
775    fn error_response(status: u16, code: &str) -> ReplayEvent {
776        let body = format!(
777            "<?xml version=\"1.0\" encoding=\"UTF-8\"?>\
778             <Error><Code>{code}</Code><Message>mock failure</Message>\
779             <RequestId>req-1</RequestId></Error>"
780        );
781        ok_response(status, SdkBody::from(body))
782    }
783
784    #[tokio::test]
785    async fn upload_puts_prefixed_object_and_returns_logical_key() {
786        let (store, http) = wire_store(wire_config(Some("uploads")), vec![ok_response(200, "")]);
787        let mut metadata = HashMap::new();
788        metadata.insert("owner".to_string(), "media".to_string());
789
790        let stored = store
791            .upload(
792                &FileSource::Bytes(bytes::Bytes::from_static(b"payload")),
793                "file.txt",
794                UploadOptions::new()
795                    .with_content_type("text/plain")
796                    .with_metadata(metadata),
797            )
798            .await
799            .unwrap();
800
801        assert_eq!(stored.key, "file.txt");
802        assert_eq!(stored.size, 7);
803        assert_eq!(stored.content_type, "text/plain");
804        assert_eq!(
805            stored.metadata.get("owner").map(String::as_str),
806            Some("media")
807        );
808
809        let request = http.actual_requests().next().expect("a request was sent");
810        assert_eq!(request.method(), "PUT");
811        assert!(
812            request.uri().contains("/test-bucket/uploads/file.txt"),
813            "unexpected upload uri: {}",
814            request.uri()
815        );
816    }
817
818    #[tokio::test]
819    async fn upload_maps_remote_failure_to_internal_error() {
820        let (store, _http) = wire_store(
821            wire_config(None),
822            vec![error_response(500, "InternalError")],
823        );
824
825        let err = store
826            .upload(
827                &FileSource::Bytes(bytes::Bytes::from_static(b"x")),
828                "file.txt",
829                UploadOptions::new(),
830            )
831            .await
832            .unwrap_err();
833
834        assert_eq!(err.code(), ErrorCode::Internal);
835        assert!(err.message().contains("S3 upload failed"));
836    }
837
838    #[tokio::test]
839    async fn download_collects_object_body_into_bytes() {
840        let (store, http) = wire_store(wire_config(None), vec![ok_response(200, "downloaded")]);
841
842        let source = store.download("file.txt").await.unwrap();
843        let data = source.read_all().await.unwrap();
844        assert_eq!(data.as_ref(), b"downloaded");
845
846        let request = http.actual_requests().next().unwrap();
847        assert_eq!(request.method(), "GET");
848        assert!(request.uri().contains("/test-bucket/file.txt"));
849    }
850
851    #[tokio::test]
852    async fn download_maps_missing_object_to_not_found() {
853        let (store, _http) = wire_store(wire_config(None), vec![error_response(404, "NoSuchKey")]);
854
855        let err = store.download("missing.txt").await.unwrap_err();
856
857        assert_eq!(err.code(), ErrorCode::NotFound);
858        assert!(err.message().contains("S3 download failed"));
859    }
860
861    #[tokio::test]
862    async fn delete_sends_delete_object_request() {
863        let (store, http) = wire_store(wire_config(None), vec![ok_response(204, "")]);
864
865        store.delete("file.txt").await.unwrap();
866
867        let request = http.actual_requests().next().unwrap();
868        assert_eq!(request.method(), "DELETE");
869        assert!(request.uri().contains("/test-bucket/file.txt"));
870    }
871
872    #[tokio::test]
873    async fn delete_maps_remote_failure_to_internal_error() {
874        let (store, _http) =
875            wire_store(wire_config(None), vec![error_response(403, "AccessDenied")]);
876
877        let err = store.delete("file.txt").await.unwrap_err();
878
879        assert_eq!(err.code(), ErrorCode::Internal);
880        assert!(err.message().contains("S3 delete failed"));
881    }
882
883    #[tokio::test]
884    async fn exists_reports_presence_from_head_outcome() {
885        let (present, _) = wire_store(wire_config(None), vec![ok_response(200, "")]);
886        assert!(present.exists("file.txt").await.unwrap());
887
888        let (absent, _) = wire_store(wire_config(None), vec![error_response(404, "NotFound")]);
889        assert!(!absent.exists("file.txt").await.unwrap());
890    }
891
892    #[tokio::test]
893    async fn head_maps_headers_to_stored_file_metadata() {
894        let response = ReplayEvent::new(
895            http::Request::builder().body(SdkBody::empty()).unwrap(),
896            http::Response::builder()
897                .status(200)
898                .header("content-length", "9")
899                .header("content-type", "application/json")
900                .header("x-amz-meta-trace", "abc")
901                .body(SdkBody::empty())
902                .unwrap(),
903        );
904        let (store, _http) = wire_store(wire_config(Some("uploads")), vec![response]);
905
906        let stored = store.head("meta.json").await.unwrap();
907
908        assert_eq!(stored.key, "meta.json");
909        assert_eq!(stored.size, 9);
910        assert_eq!(stored.content_type, "application/json");
911        assert_eq!(
912            stored.metadata.get("trace").map(String::as_str),
913            Some("abc")
914        );
915    }
916
917    #[tokio::test]
918    async fn head_maps_missing_object_to_not_found() {
919        let (store, _http) = wire_store(wire_config(None), vec![error_response(404, "NoSuchKey")]);
920
921        let err = store.head("missing.txt").await.unwrap_err();
922
923        assert_eq!(err.code(), ErrorCode::NotFound);
924        assert!(err.message().contains("S3 head failed"));
925    }
926
927    #[tokio::test]
928    async fn list_parses_contents_and_sends_prefix() {
929        let body = "<?xml version=\"1.0\" encoding=\"UTF-8\"?>\
930            <ListBucketResult xmlns=\"http://s3.amazonaws.com/doc/2006-03-01/\">\
931            <Name>test-bucket</Name><Prefix>uploads/logs</Prefix>\
932            <KeyCount>2</KeyCount><MaxKeys>2</MaxKeys><IsTruncated>false</IsTruncated>\
933            <Contents><Key>uploads/logs/a.txt</Key><Size>12</Size>\
934            <LastModified>2024-01-01T00:00:00.000Z</LastModified></Contents>\
935            <Contents><Key>uploads/logs/b.txt</Key><Size>34</Size>\
936            <LastModified>2024-01-01T00:00:00.000Z</LastModified></Contents>\
937            </ListBucketResult>";
938        let (store, http) = wire_store(
939            wire_config(Some("uploads")),
940            vec![ok_response(200, SdkBody::from(body))],
941        );
942
943        let items = store.list("logs", Some(2)).await.unwrap();
944
945        assert_eq!(items.len(), 2);
946        assert_eq!(items[0].key, "uploads/logs/a.txt");
947        assert_eq!(items[0].size, 12);
948        assert_eq!(items[1].size, 34);
949
950        let request = http.actual_requests().next().unwrap();
951        assert!(request.uri().contains("prefix=uploads%2Flogs"));
952        assert!(request.uri().contains("max-keys=2"));
953    }
954
955    #[tokio::test]
956    async fn list_maps_remote_failure_to_internal_error() {
957        let (store, _http) = wire_store(
958            wire_config(None),
959            vec![error_response(500, "InternalError")],
960        );
961
962        let err = store.list("logs", None).await.unwrap_err();
963
964        assert_eq!(err.code(), ErrorCode::Internal);
965        assert!(err.message().contains("S3 list failed"));
966    }
967
968    #[tokio::test]
969    async fn copy_copies_source_then_heads_destination() {
970        let copy_body = "<?xml version=\"1.0\" encoding=\"UTF-8\"?>\
971            <CopyObjectResult xmlns=\"http://s3.amazonaws.com/doc/2006-03-01/\">\
972            <ETag>\"abc\"</ETag><LastModified>2024-01-01T00:00:00.000Z</LastModified>\
973            </CopyObjectResult>";
974        let head = ReplayEvent::new(
975            http::Request::builder().body(SdkBody::empty()).unwrap(),
976            http::Response::builder()
977                .status(200)
978                .header("content-length", "5")
979                .header("content-type", "text/plain")
980                .body(SdkBody::empty())
981                .unwrap(),
982        );
983        let (store, http) = wire_store(
984            wire_config(Some("uploads")),
985            vec![ok_response(200, SdkBody::from(copy_body)), head],
986        );
987
988        let stored = store.copy("a.txt", "b.txt").await.unwrap();
989
990        assert_eq!(stored.key, "b.txt");
991        assert_eq!(stored.size, 5);
992
993        let mut requests = http.actual_requests();
994        let copy = requests.next().unwrap();
995        assert_eq!(copy.method(), "PUT");
996        assert!(copy.uri().contains("/test-bucket/uploads/b.txt"));
997        assert_eq!(
998            copy.headers().get("x-amz-copy-source"),
999            Some("test-bucket/uploads/a.txt")
1000        );
1001        assert_eq!(requests.next().unwrap().method(), "HEAD");
1002    }
1003
1004    #[tokio::test]
1005    async fn copy_maps_remote_failure_to_internal_error() {
1006        let (store, _http) = wire_store(wire_config(None), vec![error_response(404, "NoSuchKey")]);
1007
1008        let err = store.copy("a.txt", "b.txt").await.unwrap_err();
1009
1010        assert_eq!(err.code(), ErrorCode::Internal);
1011        assert!(err.message().contains("S3 copy failed"));
1012    }
1013
1014    #[tokio::test]
1015    async fn rename_copies_then_deletes_source() {
1016        let copy_body = "<?xml version=\"1.0\" encoding=\"UTF-8\"?>\
1017            <CopyObjectResult xmlns=\"http://s3.amazonaws.com/doc/2006-03-01/\">\
1018            <ETag>\"abc\"</ETag><LastModified>2024-01-01T00:00:00.000Z</LastModified>\
1019            </CopyObjectResult>";
1020        let head = ReplayEvent::new(
1021            http::Request::builder().body(SdkBody::empty()).unwrap(),
1022            http::Response::builder()
1023                .status(200)
1024                .header("content-length", "5")
1025                .body(SdkBody::empty())
1026                .unwrap(),
1027        );
1028        let (store, http) = wire_store(
1029            wire_config(None),
1030            vec![
1031                ok_response(200, SdkBody::from(copy_body)),
1032                head,
1033                ok_response(204, ""),
1034            ],
1035        );
1036
1037        let stored = store.rename("a.txt", "b.txt").await.unwrap();
1038
1039        assert_eq!(stored.key, "b.txt");
1040
1041        let methods: Vec<String> = http
1042            .actual_requests()
1043            .map(|r| r.method().to_owned())
1044            .collect();
1045        assert_eq!(
1046            methods.iter().map(String::as_str).collect::<Vec<_>>(),
1047            ["PUT", "HEAD", "DELETE"]
1048        );
1049    }
1050}
rskit_storage_s3/store.rs

rskit_storage_s3/
store.rs
