minco_plugin_object_storage/
base.rs

1//! Provider-neutral object storage and a deterministic in-memory reference adapter.
2#![forbid(unsafe_code)]
3
4use async_trait::async_trait;
5use chrono::{DateTime, TimeDelta, Utc};
6use minco_core::{
7    CapabilityProvision, DataClass, Plugin, PluginContext, PluginDescriptor, PluginError, PluginId,
8    PluginStability,
9};
10use semver::{Version, VersionReq};
11use serde::{Deserialize, Serialize};
12use sha2::{Digest, Sha256};
13use std::{
14    collections::{BTreeMap, VecDeque},
15    fmt,
16    sync::Arc,
17};
18use tokio::sync::{Mutex, RwLock};
19
20#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
21#[serde(transparent)]
22pub struct ObjectKey(String);
23
24impl ObjectKey {
25    pub fn parse(value: impl Into<String>) -> Result<Self, ObjectStoreError> {
26        let value = value.into();
27        if value.is_empty()
28            || value.len() > 1024
29            || value.starts_with('/')
30            || value.ends_with('/')
31            || value.split('/').any(|part| {
32                part.is_empty() || part == "." || part == ".." || part.chars().any(char::is_control)
33            })
34        {
35            return Err(ObjectStoreError::InvalidKey(value));
36        }
37        Ok(Self(value))
38    }
39
40    pub fn as_str(&self) -> &str {
41        &self.0
42    }
43}
44
45#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
46pub struct ObjectMetadata {
47    pub content_type: String,
48    pub size_bytes: u64,
49    pub sha256: String,
50    pub created_at: DateTime<Utc>,
51    #[serde(default)]
52    pub attributes: BTreeMap<String, String>,
53}
54
55#[derive(Debug, Clone, PartialEq, Eq)]
56pub struct StoredObject {
57    pub key: ObjectKey,
58    pub bytes: Vec<u8>,
59    pub metadata: ObjectMetadata,
60}
61
62#[derive(Debug, Clone, PartialEq, Eq)]
63pub struct PutObject {
64    pub key: ObjectKey,
65    pub bytes: Vec<u8>,
66    pub content_type: String,
67    pub attributes: BTreeMap<String, String>,
68}
69
70#[async_trait]
71pub trait ObjectStore: Send + Sync + std::fmt::Debug {
72    async fn put(&self, object: PutObject) -> Result<ObjectMetadata, ObjectStoreError>;
73    async fn get(&self, key: &ObjectKey) -> Result<Option<StoredObject>, ObjectStoreError>;
74    async fn delete(&self, key: &ObjectKey) -> Result<bool, ObjectStoreError>;
75}
76
77#[derive(Clone)]
78pub struct ObjectStoreService(pub Arc<dyn ObjectStore>);
79
80impl std::fmt::Debug for ObjectStoreService {
81    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
82        formatter.debug_tuple("ObjectStoreService").finish()
83    }
84}
85
86impl ObjectStoreService {
87    pub fn new(store: Arc<dyn ObjectStore>) -> Self {
88        Self(store)
89    }
90
91    pub async fn put(&self, object: PutObject) -> Result<ObjectMetadata, ObjectStoreError> {
92        self.0.put(object).await
93    }
94
95    pub async fn get(&self, key: &ObjectKey) -> Result<Option<StoredObject>, ObjectStoreError> {
96        self.0.get(key).await
97    }
98
99    pub async fn delete(&self, key: &ObjectKey) -> Result<bool, ObjectStoreError> {
100        self.0.delete(key).await
101    }
102}
103
104/// HTTP method required by a signed direct-object request.
105#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
106#[serde(rename_all = "UPPERCASE")]
107pub enum PresignedMethod {
108    Get,
109    Put,
110    Post,
111}
112
113/// Browser- or client-usable request produced by a provider adapter such as S3.
114///
115/// `form_fields` is populated for multipart POST uploads. This is required for
116/// providers such as S3 where the signed POST policy, rather than a presigned
117/// PUT URL, enforces an upload-size range.
118#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
119pub struct PresignedObjectRequest {
120    pub method: PresignedMethod,
121    pub url: String,
122    #[serde(default)]
123    pub headers: BTreeMap<String, String>,
124    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
125    pub form_fields: BTreeMap<String, String>,
126    pub expires_at: DateTime<Utc>,
127}
128
129impl std::fmt::Debug for PresignedObjectRequest {
130    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
131        formatter
132            .debug_struct("PresignedObjectRequest")
133            .field("method", &self.method)
134            .field("url", &"[REDACTED PRESIGNED URL]")
135            .field("header_names", &self.headers.keys().collect::<Vec<_>>())
136            .field(
137                "form_field_names",
138                &self.form_fields.keys().collect::<Vec<_>>(),
139            )
140            .field("expires_at", &self.expires_at)
141            .finish()
142    }
143}
144
145#[derive(Debug, Clone, PartialEq, Eq)]
146pub struct PresignPutObject {
147    pub key: ObjectKey,
148    pub content_type: String,
149    pub maximum_size_bytes: u64,
150    pub expires_in: TimeDelta,
151    pub attributes: BTreeMap<String, String>,
152}
153
154#[derive(Debug, Clone, PartialEq, Eq)]
155pub struct PresignGetObject {
156    pub key: ObjectKey,
157    pub expires_in: TimeDelta,
158    pub download_file_name: Option<String>,
159}
160
161/// Provider adapter for direct upload and download URLs.
162///
163/// Keeping signing separate from [`ObjectStore`] lets applications use server-side storage without
164/// exposing direct browser access. AWS implementations can map this port to S3 presigning while
165/// local/test implementations remain deterministic.
166#[async_trait]
167pub trait ObjectAccessSigner: Send + Sync + std::fmt::Debug {
168    async fn sign_put(
169        &self,
170        request: PresignPutObject,
171    ) -> Result<PresignedObjectRequest, ObjectStoreError>;
172
173    async fn sign_get(
174        &self,
175        request: PresignGetObject,
176    ) -> Result<PresignedObjectRequest, ObjectStoreError>;
177}
178
179#[derive(Clone)]
180pub struct ObjectAccessService(pub Arc<dyn ObjectAccessSigner>);
181
182impl std::fmt::Debug for ObjectAccessService {
183    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
184        formatter.debug_tuple("ObjectAccessService").finish()
185    }
186}
187
188impl ObjectAccessService {
189    pub fn new(signer: Arc<dyn ObjectAccessSigner>) -> Self {
190        Self(signer)
191    }
192
193    pub async fn sign_put(
194        &self,
195        request: PresignPutObject,
196    ) -> Result<PresignedObjectRequest, ObjectStoreError> {
197        validate_expiry(request.expires_in)?;
198        if request.content_type.trim().is_empty() {
199            return Err(ObjectStoreError::InvalidContentType);
200        }
201        if request.maximum_size_bytes == 0 {
202            return Err(ObjectStoreError::InvalidMaximumSize);
203        }
204        self.0.sign_put(request).await
205    }
206
207    pub async fn sign_get(
208        &self,
209        request: PresignGetObject,
210    ) -> Result<PresignedObjectRequest, ObjectStoreError> {
211        validate_expiry(request.expires_in)?;
212        self.0.sign_get(request).await
213    }
214}
215
216fn validate_expiry(expires_in: TimeDelta) -> Result<(), ObjectStoreError> {
217    if expires_in <= TimeDelta::zero() || expires_in > TimeDelta::hours(24) {
218        return Err(ObjectStoreError::InvalidExpiry);
219    }
220    Ok(())
221}
222
223#[derive(Debug, Default)]
224pub struct MemoryObjectStore {
225    objects: RwLock<BTreeMap<ObjectKey, StoredObject>>,
226}
227
228impl MemoryObjectStore {
229    /// Number of objects currently retained by the deterministic memory adapter.
230    ///
231    /// This is primarily useful for conformance tests and local diagnostics.
232    pub async fn len(&self) -> usize {
233        self.objects.read().await.len()
234    }
235
236    pub async fn is_empty(&self) -> bool {
237        self.len().await == 0
238    }
239}
240
241#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
242pub enum ObjectStoreOperation {
243    Put,
244    Get,
245    Delete,
246}
247
248#[derive(Clone, PartialEq, Eq)]
249pub enum ObjectStoreAttempt {
250    Put(PutObject),
251    Get(ObjectKey),
252    Delete(ObjectKey),
253}
254
255impl fmt::Debug for ObjectStoreAttempt {
256    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
257        match self {
258            Self::Put(object) => formatter
259                .debug_struct("Put")
260                .field("key", &object.key)
261                .field("byte_count", &object.bytes.len())
262                .field("content_type", &object.content_type)
263                .field(
264                    "attribute_names",
265                    &object.attributes.keys().collect::<Vec<_>>(),
266                )
267                .finish(),
268            Self::Get(key) => formatter.debug_tuple("Get").field(key).finish(),
269            Self::Delete(key) => formatter.debug_tuple("Delete").field(key).finish(),
270        }
271    }
272}
273
274/// Deterministic object-store fake with exact attempt capture and one-shot failures.
275#[derive(Default)]
276pub struct FakeObjectStore {
277    inner: MemoryObjectStore,
278    attempts: RwLock<Vec<ObjectStoreAttempt>>,
279    failures: Mutex<BTreeMap<ObjectStoreOperation, VecDeque<String>>>,
280}
281
282impl FakeObjectStore {
283    pub async fn fail_next(&self, operation: ObjectStoreOperation, message: impl Into<String>) {
284        self.failures
285            .lock()
286            .await
287            .entry(operation)
288            .or_default()
289            .push_back(message.into());
290    }
291
292    pub async fn attempts(&self) -> Vec<ObjectStoreAttempt> {
293        self.attempts.read().await.clone()
294    }
295
296    async fn take_failure(&self, operation: ObjectStoreOperation) -> Option<String> {
297        let mut failures = self.failures.lock().await;
298        let failure = failures.get_mut(&operation).and_then(VecDeque::pop_front);
299        if failures.get(&operation).is_some_and(VecDeque::is_empty) {
300            failures.remove(&operation);
301        }
302        drop(failures);
303        failure
304    }
305}
306
307impl fmt::Debug for FakeObjectStore {
308    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
309        formatter
310            .debug_struct("FakeObjectStore")
311            .finish_non_exhaustive()
312    }
313}
314
315#[async_trait]
316impl ObjectStore for FakeObjectStore {
317    async fn put(&self, object: PutObject) -> Result<ObjectMetadata, ObjectStoreError> {
318        validate_put_object(&object)?;
319        self.attempts
320            .write()
321            .await
322            .push(ObjectStoreAttempt::Put(object.clone()));
323        if let Some(message) = self.take_failure(ObjectStoreOperation::Put).await {
324            return Err(ObjectStoreError::Store(message));
325        }
326        self.inner.put(object).await
327    }
328
329    async fn get(&self, key: &ObjectKey) -> Result<Option<StoredObject>, ObjectStoreError> {
330        self.attempts
331            .write()
332            .await
333            .push(ObjectStoreAttempt::Get(key.clone()));
334        if let Some(message) = self.take_failure(ObjectStoreOperation::Get).await {
335            return Err(ObjectStoreError::Store(message));
336        }
337        self.inner.get(key).await
338    }
339
340    async fn delete(&self, key: &ObjectKey) -> Result<bool, ObjectStoreError> {
341        self.attempts
342            .write()
343            .await
344            .push(ObjectStoreAttempt::Delete(key.clone()));
345        if let Some(message) = self.take_failure(ObjectStoreOperation::Delete).await {
346            return Err(ObjectStoreError::Store(message));
347        }
348        self.inner.delete(key).await
349    }
350}
351
352#[async_trait]
353impl ObjectStore for MemoryObjectStore {
354    async fn put(&self, object: PutObject) -> Result<ObjectMetadata, ObjectStoreError> {
355        validate_put_object(&object)?;
356        let metadata = ObjectMetadata {
357            content_type: object.content_type,
358            size_bytes: u64::try_from(object.bytes.len())
359                .map_err(|_| ObjectStoreError::ObjectTooLarge)?,
360            sha256: hex::encode(Sha256::digest(&object.bytes)),
361            created_at: Utc::now(),
362            attributes: object.attributes,
363        };
364        self.objects.write().await.insert(
365            object.key.clone(),
366            StoredObject {
367                key: object.key,
368                bytes: object.bytes,
369                metadata: metadata.clone(),
370            },
371        );
372        Ok(metadata)
373    }
374
375    async fn get(&self, key: &ObjectKey) -> Result<Option<StoredObject>, ObjectStoreError> {
376        Ok(self.objects.read().await.get(key).cloned())
377    }
378
379    async fn delete(&self, key: &ObjectKey) -> Result<bool, ObjectStoreError> {
380        Ok(self.objects.write().await.remove(key).is_some())
381    }
382}
383
384fn validate_put_object(object: &PutObject) -> Result<(), ObjectStoreError> {
385    if object.content_type.trim().is_empty() {
386        Err(ObjectStoreError::InvalidContentType)
387    } else {
388        Ok(())
389    }
390}
391
392#[derive(Debug, Clone)]
393pub struct ObjectStoragePlugin {
394    store: ObjectStoreService,
395    access: Option<ObjectAccessService>,
396}
397
398impl ObjectStoragePlugin {
399    pub fn new(store: Arc<dyn ObjectStore>) -> Self {
400        Self {
401            store: ObjectStoreService::new(store),
402            access: None,
403        }
404    }
405
406    pub fn memory() -> Self {
407        Self::new(Arc::new(MemoryObjectStore::default()))
408    }
409
410    #[must_use]
411    pub fn with_access_signer(mut self, signer: Arc<dyn ObjectAccessSigner>) -> Self {
412        self.access = Some(ObjectAccessService::new(signer));
413        self
414    }
415}
416
417impl Plugin for ObjectStoragePlugin {
418    fn descriptor(&self) -> PluginDescriptor {
419        let mut descriptor = PluginDescriptor::new(
420            PluginId::new("object-storage").expect("static plugin ID"),
421            Version::new(1, 0, 0),
422            "Provider-neutral object storage used by uploads, exports, and feedback attachments",
423        );
424        descriptor.documentation = Some("https://docs.rs/minco-plugin-object-storage".into());
425        descriptor.core_compatibility =
426            VersionReq::parse(concat!("^", env!("CARGO_PKG_VERSION"))).expect("package version");
427        descriptor.stability = PluginStability::Beta;
428        descriptor
429            .data_classes
430            .extend([DataClass::CustomerProvided, DataClass::Confidential]);
431        descriptor.provides.push(CapabilityProvision {
432            name: "storage.object".into(),
433            version: Version::new(1, 0, 0),
434        });
435        if self.access.is_some() {
436            descriptor.provides.push(CapabilityProvision {
437                name: "storage.object.presign".into(),
438                version: Version::new(1, 0, 0),
439            });
440        }
441        descriptor
442    }
443
444    fn install(&self, context: &mut PluginContext<'_>) -> Result<(), PluginError> {
445        context.services().insert(Arc::new(self.store.clone()))?;
446        if let Some(access) = &self.access {
447            context.services().insert(Arc::new(access.clone()))?;
448        }
449        Ok(())
450    }
451}
452
453#[derive(Debug, thiserror::Error)]
454pub enum ObjectStoreError {
455    #[error("invalid object key: {0}")]
456    InvalidKey(String),
457    #[error("content type must not be empty")]
458    InvalidContentType,
459    #[error("maximum object size must be greater than zero")]
460    InvalidMaximumSize,
461    #[error("presigned request expiry must be greater than zero and no more than 24 hours")]
462    InvalidExpiry,
463    #[error("object is too large for this platform")]
464    ObjectTooLarge,
465    #[error("object store failed: {0}")]
466    Store(String),
467}
468
469#[cfg(test)]
470mod tests {
471    use super::*;
472    use minco_core::{PluginManager, PluginSelection};
473
474    #[tokio::test]
475    async fn memory_store_round_trips_bytes_and_metadata() {
476        let store = MemoryObjectStore::default();
477        let key = ObjectKey::parse("feedback/one/screenshot.png").unwrap();
478        let metadata = store
479            .put(PutObject {
480                key: key.clone(),
481                bytes: b"png".to_vec(),
482                content_type: "image/png".into(),
483                attributes: BTreeMap::new(),
484            })
485            .await
486            .unwrap();
487        assert_eq!(metadata.size_bytes, 3);
488        assert_eq!(store.get(&key).await.unwrap().unwrap().bytes, b"png");
489        assert!(store.delete(&key).await.unwrap());
490        assert!(store.get(&key).await.unwrap().is_none());
491    }
492
493    #[test]
494    fn unsafe_or_ambiguous_keys_are_rejected() {
495        for key in ["", "/absolute", "folder/", "a//b", "a/../b"] {
496            assert!(ObjectKey::parse(key).is_err(), "{key}");
497        }
498    }
499
500    #[test]
501    fn presigned_request_debug_redacts_capability_values() {
502        let request = PresignedObjectRequest {
503            method: PresignedMethod::Post,
504            url: "https://objects.example/key?X-Amz-Signature=secret-signature".into(),
505            headers: BTreeMap::from([("authorization".into(), "secret-header".into())]),
506            form_fields: BTreeMap::from([
507                ("x-amz-security-token".into(), "secret-token".into()),
508                ("x-amz-signature".into(), "secret-signature".into()),
509            ]),
510            expires_at: Utc::now() + TimeDelta::minutes(5),
511        };
512        let debug = format!("{request:?}");
513        assert!(!debug.contains("secret-token"));
514        assert!(!debug.contains("secret-signature"));
515        assert!(!debug.contains("secret-header"));
516        assert!(debug.contains("x-amz-security-token"));
517    }
518
519    #[derive(Debug)]
520    struct TestSigner;
521
522    #[async_trait]
523    impl ObjectAccessSigner for TestSigner {
524        async fn sign_put(
525            &self,
526            request: PresignPutObject,
527        ) -> Result<PresignedObjectRequest, ObjectStoreError> {
528            Ok(PresignedObjectRequest {
529                method: PresignedMethod::Put,
530                url: format!("https://objects.example/{}", request.key.as_str()),
531                headers: BTreeMap::from([("content-type".into(), request.content_type)]),
532                form_fields: BTreeMap::new(),
533                expires_at: Utc::now() + request.expires_in,
534            })
535        }
536
537        async fn sign_get(
538            &self,
539            request: PresignGetObject,
540        ) -> Result<PresignedObjectRequest, ObjectStoreError> {
541            Ok(PresignedObjectRequest {
542                method: PresignedMethod::Get,
543                url: format!("https://objects.example/{}", request.key.as_str()),
544                headers: BTreeMap::new(),
545                form_fields: BTreeMap::new(),
546                expires_at: Utc::now() + request.expires_in,
547            })
548        }
549    }
550
551    #[tokio::test]
552    async fn optional_presigning_is_typed_and_advertised_only_when_configured() {
553        let mut manager = PluginManager::default();
554        manager
555            .register(ObjectStoragePlugin::memory().with_access_signer(Arc::new(TestSigner)))
556            .unwrap();
557        let id = PluginId::new("object-storage").unwrap();
558        let mut selection = PluginSelection::default();
559        selection.enabled.insert(id);
560        let application = manager.compose(&selection).unwrap();
561        assert!(
562            application
563                .graph
564                .capabilities
565                .contains_key("storage.object.presign")
566        );
567
568        let access = application.services.get::<ObjectAccessService>().unwrap();
569        let signed = access
570            .sign_get(PresignGetObject {
571                key: ObjectKey::parse("documents/report.pdf").unwrap(),
572                expires_in: TimeDelta::minutes(5),
573                download_file_name: Some("report.pdf".into()),
574            })
575            .await
576            .unwrap();
577        assert_eq!(signed.method, PresignedMethod::Get);
578    }
579}
minco_plugin_object_storage/base.rs

minco_plugin_object_storage/
base.rs
