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

minco_plugin_object_storage/
base.rs
