scirs2_io/cloud/
mod.rs

1//! Cloud storage abstraction layer
2//!
3//! Provides a unified `ObjectStore` trait with local-filesystem and in-memory
4//! implementations, enabling code to be written against a storage-agnostic API
5//! and tested without network access.
6//!
7//! ## Components
8//!
9//! - `ObjectStore` trait: `put`, `get`, `delete`, `list`, `head`
10//! - `LocalObjectStore`: local filesystem as an object store
11//! - `MemoryObjectStore`: in-memory object store for unit testing
12//! - `ObjectKey`: URI-like key scheme (`<bucket>/<path>`)
13//! - `StorageConfig`: endpoint, credentials, retry policy
14//! - `MultipartUpload`: chunked upload simulation
15//!
16//! ## Example
17//!
18//! ```rust,no_run
19//! use scirs2_io::cloud::{LocalObjectStore, ObjectStore, ObjectKey, StorageConfig};
20//!
21//! let store = LocalObjectStore::new(std::env::temp_dir().join("my_bucket"));
22//! let key = ObjectKey::new("my_bucket", "data/file.bin");
23//! store.put(&key, b"hello world").unwrap();
24//! let data = store.get(&key).unwrap();
25//! assert_eq!(data, b"hello world");
26//! ```
27
28#![allow(dead_code)]
29#![allow(missing_docs)]
30
31pub mod azure_sas;
32pub mod gcs;
33
34pub use azure_sas::{
35    build_sas_url, generate_sas_token, is_sas_valid, parse_sas_token, AzureError, AzureSasParams,
36    SasPermissions, SasResource,
37};
38pub use gcs::{GcsError, GcsResumableUpload, UploadStatus};
39
40use crate::error::{IoError, Result};
41use serde::{Deserialize, Serialize};
42use std::collections::HashMap;
43use std::io::{Read, Write};
44use std::path::{Path, PathBuf};
45use std::sync::{Arc, Mutex};
46use std::time::{Duration, SystemTime};
47
48// ---------------------------------------------------------------------------
49// ObjectKey
50// ---------------------------------------------------------------------------
51
52/// A URI-like object key composed of an optional bucket and a path.
53///
54/// Examples:
55/// - `ObjectKey::new("raw-data", "2024/01/events.parquet")`
56/// - `ObjectKey::root("checkpoint.bin")` (no bucket)
57#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
58pub struct ObjectKey {
59    /// Optional bucket / container name
60    pub bucket: Option<String>,
61    /// Object path within the bucket (must not start with '/')
62    pub path: String,
63}
64
65impl ObjectKey {
66    /// Create a key with a bucket prefix.
67    pub fn new(bucket: impl Into<String>, path: impl Into<String>) -> Self {
68        Self {
69            bucket: Some(bucket.into()),
70            path: path.into(),
71        }
72    }
73
74    /// Create a key without a bucket (flat namespace).
75    pub fn root(path: impl Into<String>) -> Self {
76        Self {
77            bucket: None,
78            path: path.into(),
79        }
80    }
81
82    /// Return a canonical string representation: `"<bucket>/<path>"` or `"<path>"`.
83    pub fn as_uri(&self) -> String {
84        match &self.bucket {
85            Some(b) => format!("{b}/{}", self.path),
86            None => self.path.clone(),
87        }
88    }
89
90    /// Parse a URI string into an `ObjectKey`.
91    /// If the string contains a `/`, everything before the first `/` is the bucket.
92    pub fn parse(uri: &str) -> Self {
93        match uri.find('/') {
94            Some(idx) => Self::new(&uri[..idx], &uri[idx + 1..]),
95            None => Self::root(uri),
96        }
97    }
98}
99
100impl std::fmt::Display for ObjectKey {
101    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
102        write!(f, "{}", self.as_uri())
103    }
104}
105
106// ---------------------------------------------------------------------------
107// ObjectMetadata
108// ---------------------------------------------------------------------------
109
110/// Metadata returned by `head()` or `list()`.
111#[derive(Debug, Clone, Serialize, Deserialize)]
112pub struct ObjectMetadata {
113    /// Object key
114    pub key: ObjectKey,
115    /// Content length in bytes
116    pub size: u64,
117    /// Last-modified time
118    pub last_modified: SystemTime,
119    /// Optional content-type / MIME type
120    pub content_type: Option<String>,
121    /// Optional ETag (hex digest of content)
122    pub etag: Option<String>,
123    /// User-defined metadata key-value pairs
124    pub user_metadata: HashMap<String, String>,
125}
126
127// ---------------------------------------------------------------------------
128// StorageConfig
129// ---------------------------------------------------------------------------
130
131/// Configuration for a cloud storage backend.
132#[derive(Debug, Clone, Serialize, Deserialize)]
133pub struct StorageConfig {
134    /// Endpoint URL (e.g. `"https://s3.amazonaws.com"`)
135    pub endpoint: Option<String>,
136    /// Access key / client ID
137    pub access_key: Option<String>,
138    /// Secret key (stored in memory only; do not log)
139    #[serde(skip_serializing)]
140    pub secret_key: Option<String>,
141    /// Region name
142    pub region: Option<String>,
143    /// Maximum number of retries on transient failures
144    pub max_retries: u32,
145    /// Initial retry backoff duration
146    pub retry_backoff: Duration,
147    /// Request timeout
148    pub timeout: Duration,
149    /// Whether to use TLS/HTTPS
150    pub use_tls: bool,
151}
152
153impl Default for StorageConfig {
154    fn default() -> Self {
155        Self {
156            endpoint: None,
157            access_key: None,
158            secret_key: None,
159            region: None,
160            max_retries: 3,
161            retry_backoff: Duration::from_millis(100),
162            timeout: Duration::from_secs(30),
163            use_tls: true,
164        }
165    }
166}
167
168impl StorageConfig {
169    /// Create a default configuration.
170    pub fn new() -> Self {
171        Self::default()
172    }
173
174    /// Set endpoint.
175    pub fn with_endpoint(mut self, endpoint: impl Into<String>) -> Self {
176        self.endpoint = Some(endpoint.into());
177        self
178    }
179
180    /// Set credentials.
181    pub fn with_credentials(
182        mut self,
183        access_key: impl Into<String>,
184        secret_key: impl Into<String>,
185    ) -> Self {
186        self.access_key = Some(access_key.into());
187        self.secret_key = Some(secret_key.into());
188        self
189    }
190
191    /// Set max retries.
192    pub fn with_max_retries(mut self, n: u32) -> Self {
193        self.max_retries = n;
194        self
195    }
196
197    /// Set request timeout.
198    pub fn with_timeout(mut self, t: Duration) -> Self {
199        self.timeout = t;
200        self
201    }
202}
203
204// ---------------------------------------------------------------------------
205// ObjectStore trait
206// ---------------------------------------------------------------------------
207
208/// Trait for object storage backends.
209///
210/// Implementations must be `Send + Sync` to allow use from multiple threads.
211pub trait ObjectStore: Send + Sync {
212    /// Store `data` under `key`, overwriting any existing object.
213    fn put(&self, key: &ObjectKey, data: &[u8]) -> Result<()>;
214
215    /// Retrieve the object stored under `key`.
216    fn get(&self, key: &ObjectKey) -> Result<Vec<u8>>;
217
218    /// Delete the object at `key`.  Returns `Ok(())` if the key does not exist.
219    fn delete(&self, key: &ObjectKey) -> Result<()>;
220
221    /// List all keys whose URI starts with `prefix`.
222    fn list(&self, prefix: &str) -> Result<Vec<ObjectMetadata>>;
223
224    /// Return metadata for a single object without downloading its content.
225    fn head(&self, key: &ObjectKey) -> Result<ObjectMetadata>;
226
227    /// Check whether an object exists.
228    fn exists(&self, key: &ObjectKey) -> bool {
229        self.head(key).is_ok()
230    }
231
232    /// Copy an object from `src` to `dst` within the same store.
233    fn copy(&self, src: &ObjectKey, dst: &ObjectKey) -> Result<()> {
234        let data = self.get(src)?;
235        self.put(dst, &data)
236    }
237
238    /// Rename/move an object from `src` to `dst`.
239    fn rename(&self, src: &ObjectKey, dst: &ObjectKey) -> Result<()> {
240        self.copy(src, dst)?;
241        self.delete(src)
242    }
243}
244
245// ---------------------------------------------------------------------------
246// LocalObjectStore
247// ---------------------------------------------------------------------------
248
249/// Object store backed by the local filesystem.
250///
251/// Keys are mapped to paths as `<root>/<bucket>/<path>`.
252/// If the key has no bucket the root is used directly.
253pub struct LocalObjectStore {
254    root: PathBuf,
255}
256
257impl LocalObjectStore {
258    /// Create a store rooted at `root`.  The directory is created if it does not exist.
259    pub fn new<P: AsRef<Path>>(root: P) -> Self {
260        let root = root.as_ref().to_path_buf();
261        let _ = std::fs::create_dir_all(&root);
262        Self { root }
263    }
264
265    fn key_to_path(&self, key: &ObjectKey) -> PathBuf {
266        match &key.bucket {
267            Some(b) => self.root.join(b).join(&key.path),
268            None => self.root.join(&key.path),
269        }
270    }
271
272    fn etag_for(data: &[u8]) -> String {
273        use sha2::{Digest, Sha256};
274        let mut h = Sha256::new();
275        h.update(data);
276        crate::encoding_utils::hex_encode(h.finalize())
277    }
278}
279
280impl ObjectStore for LocalObjectStore {
281    fn put(&self, key: &ObjectKey, data: &[u8]) -> Result<()> {
282        let path = self.key_to_path(key);
283        if let Some(parent) = path.parent() {
284            std::fs::create_dir_all(parent)
285                .map_err(|e| IoError::FileError(format!("Cannot create dir: {e}")))?;
286        }
287        let mut f = std::fs::File::create(&path)
288            .map_err(|e| IoError::FileError(format!("Cannot create object file: {e}")))?;
289        f.write_all(data)
290            .map_err(|e| IoError::FileError(format!("Cannot write object: {e}")))?;
291        Ok(())
292    }
293
294    fn get(&self, key: &ObjectKey) -> Result<Vec<u8>> {
295        let path = self.key_to_path(key);
296        let mut f = std::fs::File::open(&path)
297            .map_err(|_| IoError::NotFound(format!("Object not found: {key}")))?;
298        let mut buf = Vec::new();
299        f.read_to_end(&mut buf)
300            .map_err(|e| IoError::FileError(format!("Cannot read object: {e}")))?;
301        Ok(buf)
302    }
303
304    fn delete(&self, key: &ObjectKey) -> Result<()> {
305        let path = self.key_to_path(key);
306        if path.exists() {
307            std::fs::remove_file(&path)
308                .map_err(|e| IoError::FileError(format!("Cannot delete object: {e}")))?;
309        }
310        Ok(())
311    }
312
313    fn list(&self, prefix: &str) -> Result<Vec<ObjectMetadata>> {
314        let mut results = Vec::new();
315        self.collect_entries(&self.root, prefix, &mut results)?;
316        Ok(results)
317    }
318
319    fn head(&self, key: &ObjectKey) -> Result<ObjectMetadata> {
320        let path = self.key_to_path(key);
321        let meta = std::fs::metadata(&path)
322            .map_err(|_| IoError::NotFound(format!("Object not found: {key}")))?;
323        let data = self.get(key)?;
324        Ok(ObjectMetadata {
325            key: key.clone(),
326            size: meta.len(),
327            last_modified: meta.modified().unwrap_or(SystemTime::UNIX_EPOCH),
328            content_type: None,
329            etag: Some(Self::etag_for(&data)),
330            user_metadata: HashMap::new(),
331        })
332    }
333}
334
335impl LocalObjectStore {
336    fn collect_entries(
337        &self,
338        dir: &Path,
339        prefix: &str,
340        results: &mut Vec<ObjectMetadata>,
341    ) -> Result<()> {
342        let entries = match std::fs::read_dir(dir) {
343            Ok(e) => e,
344            Err(_) => return Ok(()),
345        };
346        for entry in entries {
347            let entry =
348                entry.map_err(|e| IoError::FileError(format!("Cannot read dir entry: {e}")))?;
349            let path = entry.path();
350            if path.is_dir() {
351                self.collect_entries(&path, prefix, results)?;
352            } else {
353                // Compute relative key from root
354                let rel = path.strip_prefix(&self.root).unwrap_or(&path);
355                let rel_str = rel.to_string_lossy().replace('\\', "/");
356                if rel_str.starts_with(prefix) {
357                    // Parse bucket / path
358                    let key = ObjectKey::parse(&rel_str);
359                    let meta_fs = std::fs::metadata(&path)
360                        .map_err(|e| IoError::FileError(format!("Metadata error: {e}")))?;
361                    results.push(ObjectMetadata {
362                        key,
363                        size: meta_fs.len(),
364                        last_modified: meta_fs.modified().unwrap_or(SystemTime::UNIX_EPOCH),
365                        content_type: None,
366                        etag: None,
367                        user_metadata: HashMap::new(),
368                    });
369                }
370            }
371        }
372        Ok(())
373    }
374}
375
376// ---------------------------------------------------------------------------
377// MemoryObjectStore
378// ---------------------------------------------------------------------------
379
380/// In-memory object store, suitable for unit tests and simulations.
381///
382/// Thread-safe via an `Arc<Mutex<...>>` interior.
383#[derive(Clone)]
384pub struct MemoryObjectStore {
385    data: Arc<Mutex<HashMap<String, Vec<u8>>>>,
386}
387
388impl MemoryObjectStore {
389    /// Create an empty in-memory store.
390    pub fn new() -> Self {
391        Self {
392            data: Arc::new(Mutex::new(HashMap::new())),
393        }
394    }
395
396    /// Return the number of objects currently stored.
397    pub fn len(&self) -> usize {
398        self.data.lock().map(|g| g.len()).unwrap_or(0)
399    }
400
401    /// Return `true` if the store is empty.
402    pub fn is_empty(&self) -> bool {
403        self.len() == 0
404    }
405}
406
407impl Default for MemoryObjectStore {
408    fn default() -> Self {
409        Self::new()
410    }
411}
412
413impl ObjectStore for MemoryObjectStore {
414    fn put(&self, key: &ObjectKey, data: &[u8]) -> Result<()> {
415        let mut guard = self
416            .data
417            .lock()
418            .map_err(|_| IoError::Other("MemoryStore lock poisoned".to_string()))?;
419        guard.insert(key.as_uri(), data.to_vec());
420        Ok(())
421    }
422
423    fn get(&self, key: &ObjectKey) -> Result<Vec<u8>> {
424        let guard = self
425            .data
426            .lock()
427            .map_err(|_| IoError::Other("MemoryStore lock poisoned".to_string()))?;
428        guard
429            .get(&key.as_uri())
430            .cloned()
431            .ok_or_else(|| IoError::NotFound(format!("Object not found: {key}")))
432    }
433
434    fn delete(&self, key: &ObjectKey) -> Result<()> {
435        let mut guard = self
436            .data
437            .lock()
438            .map_err(|_| IoError::Other("MemoryStore lock poisoned".to_string()))?;
439        guard.remove(&key.as_uri());
440        Ok(())
441    }
442
443    fn list(&self, prefix: &str) -> Result<Vec<ObjectMetadata>> {
444        let guard = self
445            .data
446            .lock()
447            .map_err(|_| IoError::Other("MemoryStore lock poisoned".to_string()))?;
448        let results = guard
449            .iter()
450            .filter(|(uri, _)| uri.starts_with(prefix))
451            .map(|(uri, data)| ObjectMetadata {
452                key: ObjectKey::parse(uri),
453                size: data.len() as u64,
454                last_modified: SystemTime::UNIX_EPOCH,
455                content_type: None,
456                etag: None,
457                user_metadata: HashMap::new(),
458            })
459            .collect();
460        Ok(results)
461    }
462
463    fn head(&self, key: &ObjectKey) -> Result<ObjectMetadata> {
464        let guard = self
465            .data
466            .lock()
467            .map_err(|_| IoError::Other("MemoryStore lock poisoned".to_string()))?;
468        let data = guard
469            .get(&key.as_uri())
470            .ok_or_else(|| IoError::NotFound(format!("Object not found: {key}")))?;
471        Ok(ObjectMetadata {
472            key: key.clone(),
473            size: data.len() as u64,
474            last_modified: SystemTime::UNIX_EPOCH,
475            content_type: None,
476            etag: None,
477            user_metadata: HashMap::new(),
478        })
479    }
480}
481
482// ---------------------------------------------------------------------------
483// MultipartUpload
484// ---------------------------------------------------------------------------
485
486/// Chunked multipart upload simulation.
487///
488/// Mimics the semantics of S3/GCS multipart uploads:
489/// 1. Create an upload session via `MultipartUpload::new(store, key)`.
490/// 2. Upload parts with `upload_part(part_number, data)`.
491/// 3. Finalise with `complete()` which concatenates parts and stores the object.
492///
493/// Parts may be uploaded in any order; they are sorted by part number on `complete`.
494///
495/// Any part number ≥ 1 is valid.
496pub struct MultipartUpload<'a> {
497    store: &'a dyn ObjectStore,
498    key: ObjectKey,
499    parts: Vec<(u16, Vec<u8>)>,
500    min_part_size: usize,
501}
502
503impl<'a> MultipartUpload<'a> {
504    /// Create a new multipart upload session.
505    pub fn new(store: &'a dyn ObjectStore, key: ObjectKey) -> Self {
506        Self {
507            store,
508            key,
509            parts: Vec::new(),
510            min_part_size: 5 * 1024 * 1024, // 5 MiB (S3 minimum, advisory only)
511        }
512    }
513
514    /// Set the minimum part size advisory (default: 5 MiB).
515    /// Parts smaller than this will be accepted but a warning can be emitted.
516    pub fn with_min_part_size(mut self, size: usize) -> Self {
517        self.min_part_size = size;
518        self
519    }
520
521    /// Upload a single part.
522    ///
523    /// Part numbers must be in `[1, 10_000]`.
524    /// Parts are buffered in memory until `complete()` is called.
525    pub fn upload_part(&mut self, part_number: u16, data: Vec<u8>) -> Result<()> {
526        if part_number == 0 {
527            return Err(IoError::ValidationError(
528                "MultipartUpload: part number must be >= 1".to_string(),
529            ));
530        }
531        if part_number > 10_000 {
532            return Err(IoError::ValidationError(
533                "MultipartUpload: part number must be <= 10000".to_string(),
534            ));
535        }
536        // Replace any existing part with this number
537        self.parts.retain(|(n, _)| *n != part_number);
538        self.parts.push((part_number, data));
539        Ok(())
540    }
541
542    /// Return the number of parts currently uploaded.
543    pub fn part_count(&self) -> usize {
544        self.parts.len()
545    }
546
547    /// Return the total bytes buffered so far.
548    pub fn total_bytes(&self) -> usize {
549        self.parts.iter().map(|(_, d)| d.len()).sum()
550    }
551
552    /// Abort the upload and discard all buffered parts.
553    pub fn abort(&mut self) {
554        self.parts.clear();
555    }
556
557    /// Finalise the upload: sort parts by part number, concatenate, and store.
558    ///
559    /// Consumes `self`.
560    pub fn complete(mut self) -> Result<UploadResult> {
561        if self.parts.is_empty() {
562            return Err(IoError::ValidationError(
563                "MultipartUpload: no parts to complete".to_string(),
564            ));
565        }
566        self.parts.sort_by_key(|(n, _)| *n);
567        let total_size: usize = self.parts.iter().map(|(_, d)| d.len()).sum();
568        let mut assembled = Vec::with_capacity(total_size);
569        for (_, data) in &self.parts {
570            assembled.extend_from_slice(data);
571        }
572        let etag = {
573            use sha2::{Digest, Sha256};
574            let mut h = Sha256::new();
575            h.update(&assembled);
576            crate::encoding_utils::hex_encode(h.finalize())
577        };
578        self.store.put(&self.key, &assembled)?;
579        Ok(UploadResult {
580            key: self.key.clone(),
581            total_size,
582            part_count: self.parts.len(),
583            etag,
584        })
585    }
586}
587
588/// Result returned by `MultipartUpload::complete`.
589#[derive(Debug, Clone)]
590pub struct UploadResult {
591    /// The key the object was stored under
592    pub key: ObjectKey,
593    /// Total assembled size in bytes
594    pub total_size: usize,
595    /// Number of parts
596    pub part_count: usize,
597    /// SHA-256 ETag of the assembled object
598    pub etag: String,
599}
600
601// ---------------------------------------------------------------------------
602// StorageStats
603// ---------------------------------------------------------------------------
604
605/// Aggregated statistics for a storage backend session.
606#[derive(Debug, Clone, Default)]
607pub struct StorageStats {
608    /// Total bytes written (via `put`)
609    pub bytes_written: u64,
610    /// Total bytes read (via `get`)
611    pub bytes_read: u64,
612    /// Number of `put` operations
613    pub put_count: u64,
614    /// Number of `get` operations
615    pub get_count: u64,
616    /// Number of `delete` operations
617    pub delete_count: u64,
618    /// Number of errors encountered
619    pub error_count: u64,
620}
621
622/// A wrapper that instruments any `ObjectStore` implementation with statistics.
623pub struct InstrumentedStore<S: ObjectStore> {
624    inner: S,
625    stats: Arc<Mutex<StorageStats>>,
626}
627
628impl<S: ObjectStore> InstrumentedStore<S> {
629    /// Wrap `store` in a statistics-gathering layer.
630    pub fn new(store: S) -> Self {
631        Self {
632            inner: store,
633            stats: Arc::new(Mutex::new(StorageStats::default())),
634        }
635    }
636
637    /// Take a snapshot of the current statistics.
638    pub fn stats(&self) -> StorageStats {
639        self.stats.lock().map(|g| g.clone()).unwrap_or_default()
640    }
641
642    /// Reset all statistics to zero.
643    pub fn reset_stats(&self) {
644        if let Ok(mut g) = self.stats.lock() {
645            *g = StorageStats::default();
646        }
647    }
648}
649
650impl<S: ObjectStore> ObjectStore for InstrumentedStore<S> {
651    fn put(&self, key: &ObjectKey, data: &[u8]) -> Result<()> {
652        let result = self.inner.put(key, data);
653        if let Ok(mut s) = self.stats.lock() {
654            match &result {
655                Ok(()) => {
656                    s.bytes_written += data.len() as u64;
657                    s.put_count += 1;
658                }
659                Err(_) => s.error_count += 1,
660            }
661        }
662        result
663    }
664
665    fn get(&self, key: &ObjectKey) -> Result<Vec<u8>> {
666        let result = self.inner.get(key);
667        if let Ok(mut s) = self.stats.lock() {
668            match &result {
669                Ok(data) => {
670                    s.bytes_read += data.len() as u64;
671                    s.get_count += 1;
672                }
673                Err(_) => s.error_count += 1,
674            }
675        }
676        result
677    }
678
679    fn delete(&self, key: &ObjectKey) -> Result<()> {
680        let result = self.inner.delete(key);
681        if let Ok(mut s) = self.stats.lock() {
682            match &result {
683                Ok(()) => s.delete_count += 1,
684                Err(_) => s.error_count += 1,
685            }
686        }
687        result
688    }
689
690    fn list(&self, prefix: &str) -> Result<Vec<ObjectMetadata>> {
691        self.inner.list(prefix)
692    }
693
694    fn head(&self, key: &ObjectKey) -> Result<ObjectMetadata> {
695        self.inner.head(key)
696    }
697}
698
699// ---------------------------------------------------------------------------
700// Cloud provider stubs: S3, GCS, Azure Blob
701// ---------------------------------------------------------------------------
702
703/// Cloud storage provider type.
704#[derive(Debug, Clone, PartialEq, Eq)]
705pub enum StoreProviderType {
706    /// Local filesystem.
707    LocalFs,
708    /// AWS S3 or S3-compatible (MinIO, Ceph, etc.).
709    S3,
710    /// Google Cloud Storage.
711    Gcs,
712    /// Azure Blob Storage.
713    AzureBlob,
714    /// In-memory (testing).
715    Memory,
716}
717
718/// AWS S3-compatible object store stub.
719///
720/// Full HTTP/SigV4 implementation requires the `aws-sdk-s3` feature.
721/// Without that feature, all operations return `IoError::Other("s3 feature not enabled")`.
722pub struct S3Store {
723    /// Target bucket name.
724    pub bucket: String,
725    /// AWS region (e.g. `"us-east-1"`).
726    pub region: String,
727    /// Optional custom endpoint (for S3-compatible services).
728    pub endpoint: Option<String>,
729    /// AWS access key ID.
730    pub access_key: String,
731    /// AWS secret access key.
732    pub secret_key: String,
733}
734
735impl S3Store {
736    /// Create a new S3 store configuration.
737    pub fn new(
738        bucket: impl Into<String>,
739        region: impl Into<String>,
740        access_key: impl Into<String>,
741        secret_key: impl Into<String>,
742    ) -> Self {
743        Self {
744            bucket: bucket.into(),
745            region: region.into(),
746            endpoint: None,
747            access_key: access_key.into(),
748            secret_key: secret_key.into(),
749        }
750    }
751
752    /// Set a custom endpoint URL (e.g. for MinIO).
753    pub fn with_endpoint(mut self, endpoint: impl Into<String>) -> Self {
754        self.endpoint = Some(endpoint.into());
755        self
756    }
757}
758
759impl ObjectStore for S3Store {
760    fn put(&self, key: &ObjectKey, _data: &[u8]) -> Result<()> {
761        #[cfg(feature = "aws-sdk-s3")]
762        {
763            // Real implementation would use reqwest + SigV4 signing.
764            // For now even with the feature flag we return a stub error.
765            Err(IoError::Other(format!(
766                "S3 put '{}': real HTTP implementation not yet complete",
767                key
768            )))
769        }
770        #[cfg(not(feature = "aws-sdk-s3"))]
771        Err(IoError::Other(format!(
772            "S3 put '{}': enable the 'aws-sdk-s3' feature for AWS S3 support",
773            key
774        )))
775    }
776
777    fn get(&self, key: &ObjectKey) -> Result<Vec<u8>> {
778        #[cfg(feature = "aws-sdk-s3")]
779        {
780            Err(IoError::Other(format!(
781                "S3 get '{}': real HTTP implementation not yet complete",
782                key
783            )))
784        }
785        #[cfg(not(feature = "aws-sdk-s3"))]
786        Err(IoError::Other(format!(
787            "S3 get '{}': enable the 'aws-sdk-s3' feature for AWS S3 support",
788            key
789        )))
790    }
791
792    fn delete(&self, key: &ObjectKey) -> Result<()> {
793        #[cfg(feature = "aws-sdk-s3")]
794        {
795            Err(IoError::Other(format!(
796                "S3 delete '{}': real HTTP implementation not yet complete",
797                key
798            )))
799        }
800        #[cfg(not(feature = "aws-sdk-s3"))]
801        Err(IoError::Other(format!(
802            "S3 delete '{}': enable the 'aws-sdk-s3' feature for AWS S3 support",
803            key
804        )))
805    }
806
807    fn list(&self, prefix: &str) -> Result<Vec<ObjectMetadata>> {
808        #[cfg(feature = "aws-sdk-s3")]
809        {
810            Err(IoError::Other(format!(
811                "S3 list '{}': real HTTP implementation not yet complete",
812                prefix
813            )))
814        }
815        #[cfg(not(feature = "aws-sdk-s3"))]
816        Err(IoError::Other(format!(
817            "S3 list '{}': enable the 'aws-sdk-s3' feature for AWS S3 support",
818            prefix
819        )))
820    }
821
822    fn head(&self, key: &ObjectKey) -> Result<ObjectMetadata> {
823        #[cfg(feature = "aws-sdk-s3")]
824        {
825            Err(IoError::Other(format!(
826                "S3 head '{}': real HTTP implementation not yet complete",
827                key
828            )))
829        }
830        #[cfg(not(feature = "aws-sdk-s3"))]
831        Err(IoError::Other(format!(
832            "S3 head '{}': enable the 'aws-sdk-s3' feature for AWS S3 support",
833            key
834        )))
835    }
836
837    fn exists(&self, key: &ObjectKey) -> bool {
838        // Stubs always report "not found".
839        let _ = key;
840        false
841    }
842}
843
844/// Google Cloud Storage stub.
845///
846/// Full implementation requires the `google-cloud-storage` feature.
847pub struct GcsStore {
848    /// GCS bucket name.
849    pub bucket: String,
850    /// GCP project ID.
851    pub project_id: String,
852    /// Path to service-account JSON key file (optional).
853    pub credentials_path: Option<String>,
854}
855
856impl GcsStore {
857    /// Create a new GCS store configuration.
858    pub fn new(bucket: impl Into<String>, project_id: impl Into<String>) -> Self {
859        Self {
860            bucket: bucket.into(),
861            project_id: project_id.into(),
862            credentials_path: None,
863        }
864    }
865
866    /// Set the service-account credentials JSON path.
867    pub fn with_credentials(mut self, path: impl Into<String>) -> Self {
868        self.credentials_path = Some(path.into());
869        self
870    }
871}
872
873impl ObjectStore for GcsStore {
874    fn put(&self, key: &ObjectKey, _data: &[u8]) -> Result<()> {
875        Err(IoError::Other(format!(
876            "GCS put '{}': enable the 'google-cloud-storage' feature for GCS support",
877            key
878        )))
879    }
880
881    fn get(&self, key: &ObjectKey) -> Result<Vec<u8>> {
882        Err(IoError::Other(format!(
883            "GCS get '{}': enable the 'google-cloud-storage' feature for GCS support",
884            key
885        )))
886    }
887
888    fn delete(&self, key: &ObjectKey) -> Result<()> {
889        Err(IoError::Other(format!(
890            "GCS delete '{}': enable the 'google-cloud-storage' feature for GCS support",
891            key
892        )))
893    }
894
895    fn list(&self, prefix: &str) -> Result<Vec<ObjectMetadata>> {
896        Err(IoError::Other(format!(
897            "GCS list '{}': enable the 'google-cloud-storage' feature for GCS support",
898            prefix
899        )))
900    }
901
902    fn head(&self, key: &ObjectKey) -> Result<ObjectMetadata> {
903        Err(IoError::Other(format!(
904            "GCS head '{}': enable the 'google-cloud-storage' feature for GCS support",
905            key
906        )))
907    }
908
909    fn exists(&self, key: &ObjectKey) -> bool {
910        let _ = key;
911        false
912    }
913}
914
915/// Azure Blob Storage stub.
916///
917/// Full implementation requires the `azure-storage-blobs` feature.
918pub struct AzureBlobStore {
919    /// Storage account name.
920    pub account: String,
921    /// Container name.
922    pub container: String,
923    /// SAS token or account key for authentication.
924    pub credential: Option<String>,
925    /// Optional custom endpoint (for Azurite emulator).
926    pub endpoint: Option<String>,
927}
928
929impl AzureBlobStore {
930    /// Create a new Azure Blob store configuration.
931    pub fn new(account: impl Into<String>, container: impl Into<String>) -> Self {
932        Self {
933            account: account.into(),
934            container: container.into(),
935            credential: None,
936            endpoint: None,
937        }
938    }
939
940    /// Set an authentication credential (SAS token or account key).
941    pub fn with_credential(mut self, cred: impl Into<String>) -> Self {
942        self.credential = Some(cred.into());
943        self
944    }
945
946    /// Set a custom endpoint (e.g. Azurite emulator).
947    pub fn with_endpoint(mut self, endpoint: impl Into<String>) -> Self {
948        self.endpoint = Some(endpoint.into());
949        self
950    }
951}
952
953impl ObjectStore for AzureBlobStore {
954    fn put(&self, key: &ObjectKey, _data: &[u8]) -> Result<()> {
955        Err(IoError::Other(format!(
956            "Azure put '{}': enable the 'azure-storage-blobs' feature for Azure support",
957            key
958        )))
959    }
960
961    fn get(&self, key: &ObjectKey) -> Result<Vec<u8>> {
962        Err(IoError::Other(format!(
963            "Azure get '{}': enable the 'azure-storage-blobs' feature for Azure support",
964            key
965        )))
966    }
967
968    fn delete(&self, key: &ObjectKey) -> Result<()> {
969        Err(IoError::Other(format!(
970            "Azure delete '{}': enable the 'azure-storage-blobs' feature for Azure support",
971            key
972        )))
973    }
974
975    fn list(&self, prefix: &str) -> Result<Vec<ObjectMetadata>> {
976        Err(IoError::Other(format!(
977            "Azure list '{}': enable the 'azure-storage-blobs' feature for Azure support",
978            prefix
979        )))
980    }
981
982    fn head(&self, key: &ObjectKey) -> Result<ObjectMetadata> {
983        Err(IoError::Other(format!(
984            "Azure head '{}': enable the 'azure-storage-blobs' feature for Azure support",
985            key
986        )))
987    }
988
989    fn exists(&self, key: &ObjectKey) -> bool {
990        let _ = key;
991        false
992    }
993}
994
995// ---------------------------------------------------------------------------
996// URL parsing and factory
997// ---------------------------------------------------------------------------
998
999/// Parse a cloud storage URL into provider type, bucket, and path.
1000///
1001/// Supported URL schemes:
1002/// - `s3://bucket/key` — AWS S3
1003/// - `gs://bucket/key` — Google Cloud Storage
1004/// - `az://container/blob` — Azure Blob Storage
1005/// - Any other string — treated as a local filesystem path
1006///
1007/// Returns `(provider_type, bucket_or_container, object_path)`.
1008pub fn parse_store_url(
1009    url: &str,
1010) -> std::result::Result<(StoreProviderType, String, String), IoError> {
1011    if let Some(rest) = url.strip_prefix("s3://") {
1012        let (bucket, path) = split_bucket_path(rest)?;
1013        return Ok((StoreProviderType::S3, bucket, path));
1014    }
1015    if let Some(rest) = url.strip_prefix("gs://") {
1016        let (bucket, path) = split_bucket_path(rest)?;
1017        return Ok((StoreProviderType::Gcs, bucket, path));
1018    }
1019    if let Some(rest) = url.strip_prefix("az://") {
1020        let (bucket, path) = split_bucket_path(rest)?;
1021        return Ok((StoreProviderType::AzureBlob, bucket, path));
1022    }
1023    // Local path
1024    Ok((StoreProviderType::LocalFs, String::new(), url.to_string()))
1025}
1026
1027fn split_bucket_path(rest: &str) -> std::result::Result<(String, String), IoError> {
1028    match rest.find('/') {
1029        Some(idx) => Ok((rest[..idx].to_string(), rest[idx + 1..].to_string())),
1030        None => {
1031            if rest.is_empty() {
1032                Err(IoError::Other(
1033                    "Invalid store URL: missing bucket name".to_string(),
1034                ))
1035            } else {
1036                // Bucket-only URL (no path component) — path is empty
1037                Ok((rest.to_string(), String::new()))
1038            }
1039        }
1040    }
1041}
1042
1043/// Build a boxed `ObjectStore` from a URL string.
1044///
1045/// - `file://path` or a bare path: returns `LocalObjectStore` rooted at that path.
1046/// - `s3://...`, `gs://...`, `az://...`: returns the corresponding stub store.
1047///   The stub will fail at runtime unless credentials are provided via environment
1048///   variables and the appropriate Cargo feature is enabled.
1049/// - Credentials for cloud stores are read from standard environment variables:
1050///   - S3: `AWS_ACCESS_KEY_ID`, `AWS_SECRET_ACCESS_KEY`, `AWS_DEFAULT_REGION`
1051///   - GCS: `GCP_PROJECT_ID`
1052///   - Azure: `AZURE_STORAGE_ACCOUNT`, `AZURE_STORAGE_KEY`
1053pub fn from_url(url: &str) -> std::result::Result<Box<dyn ObjectStore>, IoError> {
1054    let (provider, bucket, path) = parse_store_url(url)?;
1055    match provider {
1056        StoreProviderType::LocalFs => {
1057            let root = if path.is_empty() { "." } else { &path };
1058            Ok(Box::new(LocalObjectStore::new(root)))
1059        }
1060        StoreProviderType::S3 => {
1061            let access_key = std::env::var("AWS_ACCESS_KEY_ID").unwrap_or_default();
1062            let secret_key = std::env::var("AWS_SECRET_ACCESS_KEY").unwrap_or_default();
1063            let region =
1064                std::env::var("AWS_DEFAULT_REGION").unwrap_or_else(|_| "us-east-1".to_string());
1065            Ok(Box::new(S3Store::new(
1066                bucket, region, access_key, secret_key,
1067            )))
1068        }
1069        StoreProviderType::Gcs => {
1070            let project_id = std::env::var("GCP_PROJECT_ID").unwrap_or_default();
1071            let mut store = GcsStore::new(bucket, project_id);
1072            if let Ok(creds) = std::env::var("GOOGLE_APPLICATION_CREDENTIALS") {
1073                store = store.with_credentials(creds);
1074            }
1075            Ok(Box::new(store))
1076        }
1077        StoreProviderType::AzureBlob => {
1078            let account = std::env::var("AZURE_STORAGE_ACCOUNT").unwrap_or_else(|_| bucket);
1079            let mut store = AzureBlobStore::new(account, path.clone());
1080            if let Ok(key) = std::env::var("AZURE_STORAGE_KEY") {
1081                store = store.with_credential(key);
1082            }
1083            Ok(Box::new(store))
1084        }
1085        StoreProviderType::Memory => Ok(Box::new(MemoryObjectStore::new())),
1086    }
1087}
1088
1089// ---------------------------------------------------------------------------
1090// Tests
1091// ---------------------------------------------------------------------------
1092
1093#[cfg(test)]
1094mod tests {
1095    use super::*;
1096    use std::env::temp_dir;
1097    use uuid::Uuid;
1098
1099    // ---- ObjectKey ----
1100
1101    #[test]
1102    fn test_object_key_uri() {
1103        let k = ObjectKey::new("my-bucket", "data/foo.bin");
1104        assert_eq!(k.as_uri(), "my-bucket/data/foo.bin");
1105    }
1106
1107    #[test]
1108    fn test_object_key_root() {
1109        let k = ObjectKey::root("bar.bin");
1110        assert_eq!(k.as_uri(), "bar.bin");
1111    }
1112
1113    #[test]
1114    fn test_object_key_parse() {
1115        let k = ObjectKey::parse("bucket/path/to/file");
1116        assert_eq!(k.bucket.as_deref(), Some("bucket"));
1117        assert_eq!(k.path, "path/to/file");
1118        let k2 = ObjectKey::parse("no-slash");
1119        assert_eq!(k2.bucket, None);
1120        assert_eq!(k2.path, "no-slash");
1121    }
1122
1123    // ---- MemoryObjectStore ----
1124
1125    #[test]
1126    fn test_memory_store_put_get() {
1127        let store = MemoryObjectStore::new();
1128        let key = ObjectKey::new("b", "hello.txt");
1129        store.put(&key, b"hello world").unwrap();
1130        assert_eq!(store.get(&key).unwrap(), b"hello world");
1131    }
1132
1133    #[test]
1134    fn test_memory_store_delete() {
1135        let store = MemoryObjectStore::new();
1136        let key = ObjectKey::root("x.bin");
1137        store.put(&key, b"data").unwrap();
1138        store.delete(&key).unwrap();
1139        assert!(!store.exists(&key));
1140    }
1141
1142    #[test]
1143    fn test_memory_store_list() {
1144        let store = MemoryObjectStore::new();
1145        store.put(&ObjectKey::new("b", "a/1.bin"), b"1").unwrap();
1146        store.put(&ObjectKey::new("b", "a/2.bin"), b"2").unwrap();
1147        store.put(&ObjectKey::new("c", "x.bin"), b"3").unwrap();
1148        let items = store.list("b/").unwrap();
1149        assert_eq!(items.len(), 2);
1150    }
1151
1152    #[test]
1153    fn test_memory_store_head() {
1154        let store = MemoryObjectStore::new();
1155        let key = ObjectKey::new("bkt", "file.bin");
1156        store.put(&key, b"1234567890").unwrap();
1157        let meta = store.head(&key).unwrap();
1158        assert_eq!(meta.size, 10);
1159    }
1160
1161    #[test]
1162    fn test_memory_store_copy_rename() {
1163        let store = MemoryObjectStore::new();
1164        let src = ObjectKey::root("src.bin");
1165        let dst = ObjectKey::root("dst.bin");
1166        store.put(&src, b"payload").unwrap();
1167        store.rename(&src, &dst).unwrap();
1168        assert!(!store.exists(&src));
1169        assert_eq!(store.get(&dst).unwrap(), b"payload");
1170    }
1171
1172    // ---- LocalObjectStore ----
1173
1174    #[test]
1175    fn test_local_store_put_get_delete() {
1176        let dir = temp_dir().join(format!("scirs2_cloud_{}", Uuid::new_v4()));
1177        let store = LocalObjectStore::new(&dir);
1178        let key = ObjectKey::new("bkt", "sub/data.bin");
1179        store.put(&key, b"binary data").unwrap();
1180        assert!(store.exists(&key));
1181        assert_eq!(store.get(&key).unwrap(), b"binary data");
1182        store.delete(&key).unwrap();
1183        assert!(!store.exists(&key));
1184        let _ = std::fs::remove_dir_all(&dir);
1185    }
1186
1187    #[test]
1188    fn test_local_store_head() {
1189        let dir = temp_dir().join(format!("scirs2_cloud_{}", Uuid::new_v4()));
1190        let store = LocalObjectStore::new(&dir);
1191        let key = ObjectKey::root("file.txt");
1192        store.put(&key, b"abcdef").unwrap();
1193        let meta = store.head(&key).unwrap();
1194        assert_eq!(meta.size, 6);
1195        assert!(meta.etag.is_some());
1196        let _ = std::fs::remove_dir_all(&dir);
1197    }
1198
1199    // ---- MultipartUpload ----
1200
1201    #[test]
1202    fn test_multipart_upload() {
1203        let store = MemoryObjectStore::new();
1204        let key = ObjectKey::new("bucket", "big.bin");
1205        let mut upload = MultipartUpload::new(&store, key.clone());
1206        upload.upload_part(1, b"part1".to_vec()).unwrap();
1207        upload.upload_part(3, b"part3".to_vec()).unwrap();
1208        upload.upload_part(2, b"part2".to_vec()).unwrap();
1209        assert_eq!(upload.part_count(), 3);
1210        assert_eq!(upload.total_bytes(), 15);
1211        let result = upload.complete().unwrap();
1212        assert_eq!(result.part_count, 3);
1213        // Parts should be concatenated in order 1,2,3
1214        assert_eq!(store.get(&key).unwrap(), b"part1part2part3");
1215    }
1216
1217    #[test]
1218    fn test_multipart_upload_abort() {
1219        let store = MemoryObjectStore::new();
1220        let key = ObjectKey::root("file.bin");
1221        let mut upload = MultipartUpload::new(&store, key.clone());
1222        upload.upload_part(1, b"data".to_vec()).unwrap();
1223        upload.abort();
1224        assert_eq!(upload.part_count(), 0);
1225    }
1226
1227    #[test]
1228    fn test_multipart_invalid_part_number() {
1229        let store = MemoryObjectStore::new();
1230        let key = ObjectKey::root("f");
1231        let mut upload = MultipartUpload::new(&store, key);
1232        assert!(upload.upload_part(0, vec![]).is_err());
1233        assert!(upload.upload_part(10_001, vec![]).is_err());
1234    }
1235
1236    // ---- InstrumentedStore ----
1237
1238    #[test]
1239    fn test_instrumented_store_stats() {
1240        let inner = MemoryObjectStore::new();
1241        let store = InstrumentedStore::new(inner);
1242        let key = ObjectKey::root("x");
1243        store.put(&key, b"hello").unwrap();
1244        store.get(&key).unwrap();
1245        store.delete(&key).unwrap();
1246        let s = store.stats();
1247        assert_eq!(s.put_count, 1);
1248        assert_eq!(s.get_count, 1);
1249        assert_eq!(s.delete_count, 1);
1250        assert_eq!(s.bytes_written, 5);
1251        assert_eq!(s.bytes_read, 5);
1252    }
1253
1254    // ---- parse_store_url ----
1255
1256    #[test]
1257    fn test_parse_store_url_s3() {
1258        let (provider, bucket, path) = parse_store_url("s3://my-bucket/path/to/key").unwrap();
1259        assert_eq!(provider, StoreProviderType::S3);
1260        assert_eq!(bucket, "my-bucket");
1261        assert_eq!(path, "path/to/key");
1262    }
1263
1264    #[test]
1265    fn test_parse_store_url_gcs() {
1266        let (provider, bucket, path) = parse_store_url("gs://my-bucket/data/file.parquet").unwrap();
1267        assert_eq!(provider, StoreProviderType::Gcs);
1268        assert_eq!(bucket, "my-bucket");
1269        assert_eq!(path, "data/file.parquet");
1270    }
1271
1272    #[test]
1273    fn test_parse_store_url_azure() {
1274        let (provider, bucket, path) = parse_store_url("az://mycontainer/blob/path").unwrap();
1275        assert_eq!(provider, StoreProviderType::AzureBlob);
1276        assert_eq!(bucket, "mycontainer");
1277        assert_eq!(path, "blob/path");
1278    }
1279
1280    #[test]
1281    fn test_parse_store_url_local() {
1282        let local_path = std::env::temp_dir()
1283            .join("local")
1284            .join("path")
1285            .to_str()
1286            .unwrap()
1287            .to_string();
1288        let (provider, bucket, path) = parse_store_url(&local_path).unwrap();
1289        assert_eq!(provider, StoreProviderType::LocalFs);
1290        assert!(bucket.is_empty());
1291        assert_eq!(path, local_path);
1292    }
1293
1294    // ---- S3Store / GcsStore / AzureBlobStore stub behaviour ----
1295
1296    #[test]
1297    fn test_s3_store_stub_returns_error() {
1298        let store = S3Store::new("bucket", "us-east-1", "key", "secret");
1299        let key = ObjectKey::new("bucket", "file.bin");
1300        assert!(store.put(&key, b"data").is_err());
1301        assert!(store.get(&key).is_err());
1302        assert!(store.list("bucket/").is_err());
1303        assert!(store.head(&key).is_err());
1304        assert!(store.delete(&key).is_err());
1305        // exists() must not panic
1306        assert!(!store.exists(&key));
1307    }
1308
1309    #[test]
1310    fn test_gcs_store_stub_returns_error() {
1311        let store = GcsStore::new("my-bucket", "my-project");
1312        let key = ObjectKey::new("my-bucket", "file.bin");
1313        assert!(store.put(&key, b"data").is_err());
1314        assert!(store.get(&key).is_err());
1315    }
1316
1317    #[test]
1318    fn test_azure_store_stub_returns_error() {
1319        let store = AzureBlobStore::new("myaccount", "mycontainer");
1320        let key = ObjectKey::new("mycontainer", "blob.bin");
1321        assert!(store.put(&key, b"data").is_err());
1322        assert!(store.get(&key).is_err());
1323    }
1324
1325    // ---- from_url ----
1326
1327    #[test]
1328    fn test_from_url_local() {
1329        let dir = temp_dir().join(format!("scirs2_from_url_{}", Uuid::new_v4()));
1330        let store = from_url(dir.to_str().unwrap()).unwrap();
1331        let key = ObjectKey::root("test.bin");
1332        store.put(&key, b"hello from_url").unwrap();
1333        assert_eq!(store.get(&key).unwrap(), b"hello from_url");
1334        let _ = std::fs::remove_dir_all(&dir);
1335    }
1336
1337    #[test]
1338    fn test_from_url_s3_stub() {
1339        let store = from_url("s3://test-bucket/prefix").unwrap();
1340        let key = ObjectKey::new("test-bucket", "file.bin");
1341        // Stub should return an error, not panic
1342        assert!(store.get(&key).is_err());
1343    }
1344}
scirs2_io/cloud/mod.rs

scirs2_io/cloud/
mod.rs
