cratefield_core/ports/
blob.rs

1//! The `Blob` port (issue #105): small media stored outside the database —
2//! R2 on Workers, a directory or S3-compatible store when self-hosted.
3//!
4//! KV caps a value at 25 MB and is the wrong tool for media, and D1 `BLOB`
5//! columns are banned by the portable lint, so a coach's voice clip (the
6//! request in `yoginini-backend#4`) has nowhere to live. This port gives it
7//! one, keyed and content-typed, with the same ownership rule the tables have:
8//! a module only ever touches keys under its own `<module>/` prefix, enforced
9//! by the [`ScopedBlob`] the harness hands each module.
10
11use std::sync::Arc;
12use std::time::Duration;
13
14use async_trait::async_trait;
15use thiserror::Error;
16
17/// Hard ceiling on one stored object (issue #136). The port exists for
18/// *small* media — a coach's voice clip, an avatar — and every caller
19/// reaches it through [`ScopedBlob`] or an adapter that enforces this
20/// bound before allocating or writing. Ten MiB is well past any legit
21/// clip at conversational bitrates and far below what threatens a
22/// self-hosted process or an isolate that buffers the bytes.
23pub const MAX_BLOB_BYTES: usize = 10 * 1024 * 1024;
24
25/// Rejects a write whose payload exceeds [`MAX_BLOB_BYTES`] (issue #136).
26/// Every [`Blob::put`] implementation MUST call this before touching
27/// storage; [`ScopedBlob`] (the module-facing wrapper) already does, so
28/// an adapter's own call is the second, defence-in-depth gate for
29/// runtimes-wired stores and direct users.
30///
31/// # Errors
32///
33/// [`BlobError::TooLarge`] when `bytes.len()` exceeds the bound.
34pub fn check_blob_size(bytes: &[u8]) -> Result<(), BlobError> {
35    if bytes.len() > MAX_BLOB_BYTES {
36        return Err(BlobError::TooLarge(format!(
37            "blob of {} bytes exceeds the {MAX_BLOB_BYTES}-byte bound",
38            bytes.len()
39        )));
40    }
41    Ok(())
42}
43
44/// A stored object: its bytes and the content type to serve it with.
45#[derive(Debug, Clone, PartialEq, Eq)]
46pub struct BlobObject {
47    pub bytes: Vec<u8>,
48    pub content_type: String,
49}
50
51/// Blob store failures.
52#[derive(Debug, Clone, Error)]
53pub enum BlobError {
54    /// The store rejected or could not complete the operation.
55    #[error("blob operation failed: {0}")]
56    Operation(String),
57    /// The key is empty, absolute, or tries to escape its prefix (`..`).
58    #[error("invalid blob key: {0}")]
59    BadKey(String),
60    /// The object is bigger than [`MAX_BLOB_BYTES`] (issue #136).
61    #[error("blob rejected by size: {0}")]
62    TooLarge(String),
63    /// The adapter does not support this operation (e.g. a directory store has
64    /// no presigned URLs).
65    #[error("blob operation not supported: {0}")]
66    Unsupported(String),
67}
68
69/// A blob store. Keys are module-prefixed; the harness wraps this in a
70/// [`ScopedBlob`] per module so a module cannot name another's objects.
71#[async_trait]
72pub trait Blob: Send + Sync {
73    /// Stores `bytes` at `key` with `content_type`, replacing any existing
74    /// object. Implementations MUST reject a payload larger than
75    /// [`MAX_BLOB_BYTES`] with [`BlobError::TooLarge`] via
76    /// [`check_blob_size`], before writing (issue #136).
77    async fn put(&self, key: &str, bytes: &[u8], content_type: &str) -> Result<(), BlobError>;
78    /// Fetches the object at `key`, or `None` if there is none.
79    async fn get(&self, key: &str) -> Result<Option<BlobObject>, BlobError>;
80    /// Removes the object at `key`. Idempotent: removing a missing key is `Ok`.
81    async fn delete(&self, key: &str) -> Result<(), BlobError>;
82    /// A URL that serves the object directly for `ttl`, skipping the Worker.
83    /// Adapters without presigned URLs (a directory store) return
84    /// [`BlobError::Unsupported`]; callers then serve the bytes through
85    /// [`get`](Blob::get).
86    async fn signed_url(&self, key: &str, ttl: Duration) -> Result<String, BlobError>;
87}
88
89/// Wraps a [`Blob`] so every key is prefixed with `<module>/` and no key can
90/// escape it. The harness applies this in `Ports::view_for`, so a module sees a
91/// store scoped to itself — the blob equivalent of the table-ownership check.
92pub struct ScopedBlob {
93    inner: Arc<dyn Blob>,
94    prefix: String,
95}
96
97impl ScopedBlob {
98    /// Scopes `inner` to `<module>/`.
99    #[must_use]
100    pub fn new(inner: Arc<dyn Blob>, module: &str) -> Self {
101        Self {
102            inner,
103            prefix: format!("{module}/"),
104        }
105    }
106
107    /// Prefixes a caller key, refusing one that is empty, absolute, or walks
108    /// out of the prefix with `..`.
109    fn scope(&self, key: &str) -> Result<String, BlobError> {
110        if key.is_empty() {
111            return Err(BlobError::BadKey("a blob key cannot be empty".to_owned()));
112        }
113        if key.starts_with('/') {
114            return Err(BlobError::BadKey(format!("key `{key}` must be relative")));
115        }
116        if key
117            .split('/')
118            .any(|segment| segment == ".." || segment == ".")
119        {
120            return Err(BlobError::BadKey(format!(
121                "key `{key}` must not contain `.` or `..` segments"
122            )));
123        }
124        Ok(format!("{}{key}", self.prefix))
125    }
126}
127
128#[async_trait]
129impl Blob for ScopedBlob {
130    async fn put(&self, key: &str, bytes: &[u8], content_type: &str) -> Result<(), BlobError> {
131        check_blob_size(bytes)?;
132        self.inner.put(&self.scope(key)?, bytes, content_type).await
133    }
134    async fn get(&self, key: &str) -> Result<Option<BlobObject>, BlobError> {
135        self.inner.get(&self.scope(key)?).await
136    }
137    async fn delete(&self, key: &str) -> Result<(), BlobError> {
138        self.inner.delete(&self.scope(key)?).await
139    }
140    async fn signed_url(&self, key: &str, ttl: Duration) -> Result<String, BlobError> {
141        self.inner.signed_url(&self.scope(key)?, ttl).await
142    }
143}
144
145#[cfg(test)]
146mod tests {
147    // A recording test fixture, not request state (ADR 0007). The scoped
148    // allow follows the policy in the workspace clippy.toml, as the fakes
149    // in `cratefield-testing` and the sibling tests in this crate do.
150    #![allow(clippy::disallowed_types)]
151
152    use super::*;
153    use std::sync::Mutex;
154
155    /// An in-memory blob store for testing the scoping.
156    #[derive(Default)]
157    struct MemBlob {
158        objects: Mutex<std::collections::HashMap<String, BlobObject>>,
159    }
160
161    #[async_trait]
162    impl Blob for MemBlob {
163        async fn put(&self, key: &str, bytes: &[u8], content_type: &str) -> Result<(), BlobError> {
164            self.objects.lock().unwrap().insert(
165                key.to_owned(),
166                BlobObject {
167                    bytes: bytes.to_vec(),
168                    content_type: content_type.to_owned(),
169                },
170            );
171            Ok(())
172        }
173        async fn get(&self, key: &str) -> Result<Option<BlobObject>, BlobError> {
174            Ok(self.objects.lock().unwrap().get(key).cloned())
175        }
176        async fn delete(&self, key: &str) -> Result<(), BlobError> {
177            self.objects.lock().unwrap().remove(key);
178            Ok(())
179        }
180        async fn signed_url(&self, _key: &str, _ttl: Duration) -> Result<String, BlobError> {
181            Err(BlobError::Unsupported("memory store".to_owned()))
182        }
183    }
184
185    #[pollster::test]
186    async fn a_scoped_blob_prefixes_the_key() {
187        let mem = Arc::new(MemBlob::default());
188        let scoped = ScopedBlob::new(mem.clone(), "waitlist");
189        scoped.put("clip.mp3", b"x", "audio/mpeg").await.unwrap();
190        // The underlying store sees the prefixed key.
191        assert!(mem.get("waitlist/clip.mp3").await.unwrap().is_some());
192        assert!(mem.get("clip.mp3").await.unwrap().is_none());
193        // And the scoped view reads it back by the bare key.
194        assert!(scoped.get("clip.mp3").await.unwrap().is_some());
195    }
196
197    #[pollster::test]
198    async fn a_scoped_blob_refuses_an_escaping_key() {
199        let scoped = ScopedBlob::new(Arc::new(MemBlob::default()), "cms");
200        for bad in ["", "/etc/passwd", "../secrets/x", "a/../../b", "."] {
201            assert!(
202                matches!(scoped.get(bad).await.unwrap_err(), BlobError::BadKey(_)),
203                "key `{bad}` should be refused"
204            );
205        }
206    }
207
208    #[pollster::test]
209    async fn delete_is_idempotent() {
210        let scoped = ScopedBlob::new(Arc::new(MemBlob::default()), "cms");
211        scoped.delete("missing").await.expect("no-op delete is ok");
212    }
213
214    #[pollster::test]
215    async fn an_oversized_put_is_refused_before_the_store_is_reached() {
216        let mem = Arc::new(MemBlob::default());
217        let scoped = ScopedBlob::new(mem.clone(), "cms");
218        let oversized = vec![0_u8; MAX_BLOB_BYTES + 1];
219        let err = scoped
220            .put("big.bin", &oversized, "application/octet-stream")
221            .await
222            .expect_err("one byte past the bound must be refused");
223        assert!(matches!(err, BlobError::TooLarge(_)), "got {err}");
224        assert!(
225            mem.get("cms/big.bin").await.unwrap().is_none(),
226            "the store never saw the rejected key"
227        );
228    }
229
230    #[pollster::test]
231    async fn a_put_at_exactly_the_bound_is_accepted() {
232        let scoped = ScopedBlob::new(Arc::new(MemBlob::default()), "cms");
233        let exact = vec![0_u8; MAX_BLOB_BYTES];
234        scoped
235            .put("edge.bin", &exact, "application/octet-stream")
236            .await
237            .expect("the bound itself is allowed");
238    }
239}
cratefield_core/ports/blob.rs

cratefield_core/ports/
blob.rs
