cratefield_module_sealed/
store.rs

1//! The sealed store's tables, read and written in the portable SQL subset
2//! (ADR 0004). Every query is bounded by the subject the caller's verified
3//! credential resolved to — a blob id alone names nothing.
4//!
5//! The layout is the crypto-shred: the body exists in exactly one place
6//! (inline in `sealed_blobs.outer_ct`, or the blob-store object `object_key`
7//! names) and the only key that opens it is the row in `sealed_deks`.
8
9use sea_query::Value as SeaValue;
10use serde_json::json;
11
12use cratefield_core::{Database, DbError, Row, Statement};
13use cratefield_kms::{Dek, Kms};
14
15use crate::SealedError;
16use crate::outer;
17use crate::record::{CreateBlob, Wrap, b64url_decode, b64url_encode};
18
19/// Bodies at or below this many (outer-wrapped) bytes live inline in
20/// `sealed_blobs.outer_ct`; larger ones go to the blob store. Below the
21/// harness's 64 KiB request ceiling so the inline path stays exercisable.
22pub const INLINE_BODY_MAX_BYTES: usize = 32 * 1024;
23
24/// Why a store call refused.
25#[derive(Debug)]
26#[non_exhaustive]
27pub enum StoreError {
28    /// The blob id is already this subject's.
29    Taken,
30    /// No such blob for this subject.
31    NotFound,
32    /// The record changed since the caller last read it: its revision no
33    /// longer matches the `ETag` the request carried.
34    Stale,
35    /// The database refused the statement.
36    Db(DbError),
37    /// The KMS could not open a wrapped DEK.
38    Kms(cratefield_kms::KmsError),
39    /// An outer ciphertext did not authenticate.
40    Tampered(String),
41    /// Something that should be unreachable: a row is malformed, the RNG
42    /// failed. Logged, never described to a caller.
43    Internal(String),
44}
45
46impl std::fmt::Display for StoreError {
47    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
48        match self {
49            Self::Taken => f.write_str("that blob_id is already stored"),
50            Self::NotFound => f.write_str("no such blob"),
51            Self::Stale => f.write_str("the record changed since the caller last read it"),
52            Self::Db(err) => write!(f, "{err}"),
53            Self::Kms(err) => write!(f, "{err}"),
54            Self::Tampered(what) => write!(f, "the stored body failed authentication: {what}"),
55            Self::Internal(what) => write!(f, "{what}"),
56        }
57    }
58}
59
60impl std::error::Error for StoreError {}
61
62impl From<DbError> for StoreError {
63    fn from(err: DbError) -> Self {
64        Self::Db(err)
65    }
66}
67
68impl From<cratefield_kms::KmsError> for StoreError {
69    fn from(err: cratefield_kms::KmsError) -> Self {
70        Self::Kms(err)
71    }
72}
73
74impl From<SealedError> for StoreError {
75    fn from(err: SealedError) -> Self {
76        match err {
77            SealedError::Tampered(what) => Self::Tampered(what),
78            other => Self::Internal(other.to_string()),
79        }
80    }
81}
82
83impl From<cratefield_core::BlobError> for StoreError {
84    fn from(err: cratefield_core::BlobError) -> Self {
85        Self::Internal(format!("the blob store refused: {err}"))
86    }
87}
88
89/// One stored blob: the metadata the wire record carries, plus where the
90/// body lives.
91#[derive(Debug, Clone, PartialEq, Eq)]
92pub struct BlobRow {
93    pub blob_id: String,
94    pub version: u32,
95    pub purpose: String,
96    pub alg: String,
97    /// The wraps, as the canonical JSON array the client sent.
98    pub wraps: String,
99    pub created_by_credential: String,
100    /// The decoded client ciphertext's size in bytes.
101    pub ciphertext_len: i64,
102    /// The outer-wrapped body when inline; `None` when it is in the blob
103    /// store under `object_key`.
104    pub outer_ct: Option<Vec<u8>>,
105    pub object_key: Option<String>,
106    /// The server-side validator behind the `ETag`: replaced with a fresh
107    /// value by every mutation, and the value a guarded mutation's `WHERE`
108    /// clause must match. `version` cannot serve this: it is the client's,
109    /// and the client payload's AAD binds it, so a wrap edit must leave it
110    /// alone.
111    pub revision: i64,
112    pub created_at: String,
113    pub updated_at: String,
114}
115
116fn text(value: &str) -> SeaValue {
117    SeaValue::String(Some(Box::new(value.to_owned())))
118}
119
120fn bytes(value: &[u8]) -> SeaValue {
121    SeaValue::Bytes(Some(Box::new(value.to_vec())))
122}
123
124fn row_text(row: &Row, column: &str) -> Option<String> {
125    row.get::<Option<String>>(column).flatten()
126}
127
128const ROW_COLUMNS: &str = "blob_id, version, purpose, alg, wraps, created_by_credential, \
129     ciphertext_len, outer_ct, object_key, revision, created_at, updated_at";
130
131fn from_row(row: &Row) -> BlobRow {
132    BlobRow {
133        blob_id: row.get::<String>("blob_id").unwrap_or_default(),
134        version: u32::try_from(row.get::<i64>("version").unwrap_or_default()).unwrap_or_default(),
135        purpose: row_text(row, "purpose").unwrap_or_default(),
136        alg: row_text(row, "alg").unwrap_or_default(),
137        wraps: row_text(row, "wraps").unwrap_or_default(),
138        created_by_credential: row_text(row, "created_by_credential").unwrap_or_default(),
139        ciphertext_len: row.get::<i64>("ciphertext_len").unwrap_or_default(),
140        outer_ct: match row.get::<SeaValue>("outer_ct") {
141            Some(SeaValue::Bytes(Some(bytes))) => Some(bytes.as_ref().clone()),
142            _ => None,
143        },
144        object_key: row_text(row, "object_key"),
145        revision: row.get::<i64>("revision").unwrap_or_default(),
146        created_at: row_text(row, "created_at").unwrap_or_default(),
147        updated_at: row_text(row, "updated_at").unwrap_or_default(),
148    }
149}
150
151/// The blob-store key for one parked body: the subject is only ever a hashed
152/// slug (the key never embeds a subject id or a guessable name), but the
153/// suffix is **fresh per write** — version plus random bytes. Uniqueness is
154/// what makes rotation safe: a new body never lands on the key the live body
155/// occupies, so superseding one key after the commit cannot destroy the
156/// record, and unparking a refused request's key cannot touch anyone else's.
157fn object_key(subject: &str, blob_id: &str, version: u32) -> Result<String, StoreError> {
158    use sha2::{Digest, Sha256};
159    use std::fmt::Write as _;
160    let digest = Sha256::digest(format!("{subject}/{blob_id}").as_bytes());
161    let mut slug = String::with_capacity(48);
162    for byte in digest.iter().take(16) {
163        let _ = write!(slug, "{byte:02x}");
164    }
165    let mut suffix = [0u8; 8];
166    getrandom::fill(&mut suffix).map_err(|err| StoreError::Internal(err.to_string()))?;
167    let _ = write!(slug, "-v{version}-");
168    for byte in suffix {
169        let _ = write!(slug, "{byte:02x}");
170    }
171    // Module-relative: the Blob port arrives scoped to this module's
172    // `sealed/` prefix (`Ports::view_for`), so the object lands at
173    // `sealed/blobs/{slug}.outer` in the store.
174    Ok(format!("blobs/{slug}.outer"))
175}
176
177/// Whether the outer-wrapped body at this size goes inline.
178pub(crate) fn is_inline(outer_ct_len: usize) -> bool {
179    outer_ct_len <= INLINE_BODY_MAX_BYTES
180}
181
182/// A fresh validator value for one guarded write. The mutation sets
183/// `revision` to *this* value under a `WHERE revision = expected` guard, so
184/// the read-back after the batch can tell "the row now carries my write"
185/// from "somebody else had already moved it" — a mere bump by one cannot
186/// make that distinction, because the winner moves it by one too.
187fn fresh_revision() -> Result<i64, StoreError> {
188    let mut bytes = [0_u8; 8];
189    getrandom::fill(&mut bytes).map_err(|err| StoreError::Internal(err.to_string()))?;
190    Ok(i64::from_be_bytes(bytes) & i64::MAX)
191}
192
193/// Seals `body` (the client's decoded ciphertext bytes) under a fresh DEK
194/// and returns everything a row needs: the wrapped DEK, the outer body and —
195/// when the body is too big for the row — the blob-store key to park it at.
196pub struct SealedBody {
197    pub wrapped_dek: Vec<u8>,
198    pub provider: &'static str,
199    pub key_ref: String,
200    pub outer_ct: Vec<u8>,
201    /// The client ciphertext's own size in bytes (before the outer wrap).
202    pub ciphertext_len: usize,
203    /// `Some` when the body goes to the blob store; `None` when it is inline.
204    pub object_key: Option<String>,
205}
206
207/// Wraps a fresh per-blob DEK through `kms` and seals `body` with it.
208///
209/// # Errors
210///
211/// [`StoreError::Kms`] when the KMS refuses, [`StoreError::Internal`] when
212/// the RNG or the cipher fails.
213pub async fn seal_body(
214    kms: &dyn Kms,
215    body: &[u8],
216    subject: &str,
217    blob_id: &str,
218    version: u32,
219) -> Result<SealedBody, StoreError> {
220    let dek = Dek::generate().map_err(|err| StoreError::Internal(err.to_string()))?;
221    let wrapped_dek = kms.wrap(&dek).await?;
222    let outer_ct = outer::seal(&dek, body, subject, blob_id, version)?;
223    let inline = is_inline(outer_ct.len());
224    let object_key = match (!inline).then(|| object_key(subject, blob_id, version)) {
225        Some(key) => Some(key?),
226        None => None,
227    };
228    Ok(SealedBody {
229        wrapped_dek,
230        provider: kms.provider(),
231        key_ref: kms.key_ref().to_owned(),
232        ciphertext_len: body.len(),
233        object_key,
234        outer_ct,
235    })
236}
237
238/// Opens a row's body: unwraps the DEK through `kms`, fetches the outer
239/// ciphertext from the row or the blob store, and returns the client's
240/// ciphertext bytes (not the client plaintext — no code path reaches that).
241///
242/// # Errors
243///
244/// [`StoreError::NotFound`] without a row, [`StoreError::Tampered`] when the
245/// outer layer does not authenticate, [`StoreError::Kms`] when the KMS
246/// refuses, and blob-store failures as [`StoreError::Db`]-shaped internals.
247pub async fn open_body(
248    kms: &dyn Kms,
249    blob: Option<&dyn cratefield_core::Blob>,
250    db: &dyn Database,
251    subject: &str,
252    row: &BlobRow,
253) -> Result<Vec<u8>, StoreError> {
254    let rows = db
255        .query(&Statement::with_values(
256            "SELECT wrapped_dek FROM sealed_deks WHERE subject = ? AND blob_id = ?",
257            vec![text(subject), text(&row.blob_id)],
258        ))
259        .await?;
260    let wrapped_dek = rows
261        .first()
262        .and_then(|r| r.get::<Vec<u8>>("wrapped_dek"))
263        .ok_or(StoreError::NotFound)?;
264    let dek = kms.unwrap(&wrapped_dek).await?;
265    let outer_ct = match (&row.outer_ct, &row.object_key) {
266        (Some(inline), _) => inline.clone(),
267        (None, Some(key)) => {
268            let store = blob.ok_or_else(|| {
269                StoreError::Internal("no Blob port is mounted, but the body lives in one".into())
270            })?;
271            store
272                .get(key)
273                .await?
274                .map(|object| object.bytes)
275                .ok_or(StoreError::NotFound)?
276        }
277        (None, None) => {
278            return Err(StoreError::Internal(
279                "a blob row names neither an inline body nor an object key".into(),
280            ));
281        }
282    };
283    Ok(outer::open(
284        &dek,
285        &outer_ct,
286        subject,
287        &row.blob_id,
288        row.version,
289    )?)
290}
291
292/// The wire record a row serves, rebuilt from the stored metadata and the
293/// client ciphertext `open_body` recovered. The base64url fields round-trip
294/// byte-identically because the validators only ever accepted the canonical
295/// encoding.
296///
297/// # Errors
298///
299/// [`StoreError::Internal`] when the stored ciphertext does not decode or
300/// the stored wraps do not parse — a row only the routes write cannot be
301/// either, so this is the "someone else's writer" arm.
302pub fn wire_record(
303    row: &BlobRow,
304    subject: &str,
305    ciphertext: &[u8],
306) -> Result<serde_json::Value, StoreError> {
307    let wraps: Vec<Wrap> = serde_json::from_str(&row.wraps).map_err(|err| {
308        StoreError::Internal(format!("a stored wraps column does not parse: {err}"))
309    })?;
310    Ok(json!({
311        "blob_id": row.blob_id,
312        "version": row.version,
313        "purpose": row.purpose,
314        "alg": row.alg,
315        "ciphertext": b64url_encode(ciphertext),
316        "wraps": wraps,
317        "created_by_credential": row.created_by_credential,
318        "subject": subject,
319        "created_at": row.created_at,
320        "updated_at": row.updated_at,
321    }))
322}
323
324/// The decoded client ciphertext a create or rotation carries.
325///
326/// # Errors
327///
328/// [`StoreError::Internal`] — unreachable for a body `validate_create`
329/// accepted; the arm exists so a caller that skips validation cannot write
330/// garbage.
331pub fn decode_body(blob: &CreateBlob) -> Result<Vec<u8>, StoreError> {
332    b64url_decode(&blob.ciphertext)
333        .ok_or_else(|| StoreError::Internal("a validated ciphertext failed to decode".into()))
334}
335
336/// Inserts the blob row and its DEK row together, refusing a blob id this
337/// subject already holds. The two statements are one atomic batch: a row
338/// whose key never landed would be unopenable, and a key without a row would
339/// be an unshreddable orphan.
340///
341/// # Errors
342///
343/// [`StoreError::Taken`] when `(subject, blob_id)` exists;
344/// [`StoreError::Db`] otherwise.
345pub async fn insert(
346    db: &dyn Database,
347    subject: &str,
348    blob: &CreateBlob,
349    sealed: &SealedBody,
350    ts: &str,
351) -> Result<(), StoreError> {
352    let existing = db
353        .query(&Statement::with_values(
354            "SELECT blob_id FROM sealed_blobs WHERE subject = ? AND blob_id = ?",
355            vec![text(subject), text(&blob.blob_id)],
356        ))
357        .await?;
358    if !existing.is_empty() {
359        return Err(StoreError::Taken);
360    }
361    let (outer_ct, object_key): (SeaValue, SeaValue) = match &sealed.object_key {
362        None => (bytes(&sealed.outer_ct), SeaValue::String(None)),
363        Some(key) => (SeaValue::Bytes(None), text(key)),
364    };
365    db.batch_atomic(&[
366        Statement::with_values(
367            "INSERT INTO sealed_blobs (subject, blob_id, version, purpose, alg, wraps, \
368             created_by_credential, ciphertext_len, outer_ct, object_key, revision, \
369             created_at, updated_at) \
370             VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 1, ?, ?)",
371            vec![
372                text(subject),
373                text(&blob.blob_id),
374                SeaValue::BigInt(Some(i64::from(blob.version))),
375                text(&blob.purpose),
376                text(&blob.alg),
377                text(&wraps_json(&blob.wraps)?),
378                text(&blob.created_by_credential),
379                big(sealed.ciphertext_len),
380                outer_ct,
381                object_key,
382                text(ts),
383                text(ts),
384            ],
385        ),
386        Statement::with_values(
387            "INSERT INTO sealed_deks (subject, blob_id, kms_provider, kms_key_ref, wrapped_dek, \
388             revision) VALUES (?, ?, ?, ?, ?, 1)",
389            vec![
390                text(subject),
391                text(&blob.blob_id),
392                text(sealed.provider),
393                text(&sealed.key_ref),
394                bytes(&sealed.wrapped_dek),
395            ],
396        ),
397    ])
398    .await?;
399    Ok(())
400}
401
402/// The wraps, as the canonical JSON string the row stores and the wire
403/// record re-parses.
404fn wraps_json(wraps: &[Wrap]) -> Result<String, StoreError> {
405    serde_json::to_string(wraps).map_err(|err| {
406        StoreError::Internal(format!("a validated wrap set does not serialise: {err}"))
407    })
408}
409
410/// A count that started life as a `usize` column value.
411fn big(len: usize) -> SeaValue {
412    SeaValue::BigInt(Some(i64::try_from(len).unwrap_or(i64::MAX)))
413}
414
415/// The row for `(subject, blob_id)`, or `None`.
416///
417/// # Errors
418///
419/// [`StoreError::Db`].
420pub async fn load(
421    db: &dyn Database,
422    subject: &str,
423    blob_id: &str,
424) -> Result<Option<BlobRow>, StoreError> {
425    let rows = db
426        .query(&Statement::with_values(
427            format!("SELECT {ROW_COLUMNS} FROM sealed_blobs WHERE subject = ? AND blob_id = ?"),
428            vec![text(subject), text(blob_id)],
429        ))
430        .await?;
431    Ok(rows.first().map(from_row))
432}
433
434/// The subject's rows, newest first: metadata only, no bodies.
435///
436/// # Errors
437///
438/// [`StoreError::Db`].
439pub async fn list(db: &dyn Database, subject: &str) -> Result<Vec<BlobRow>, StoreError> {
440    let rows = db
441        .query(&Statement::with_values(
442            format!(
443                "SELECT {ROW_COLUMNS} FROM sealed_blobs WHERE subject = ? ORDER BY created_at DESC"
444            ),
445            vec![text(subject)],
446        ))
447        .await?;
448    Ok(rows.rows.iter().map(from_row).collect())
449}
450
451/// Replaces the wrap set under an optimistic-concurrency check on the row's
452/// [`BlobRow::revision`] (the `ETag`): the batch writes only where the row
453/// still carries `expected_revision`, and writes a **fresh** validator value
454/// there — so the read-back after the batch can attribute the row's state to
455/// this request alone, and a lost race surfaces as [`StoreError::Stale`]
456/// instead of a silent overwrite. Two edits sent from the same `ETag` cannot
457/// both land. The payload — the outer ciphertext, the DEK material, the
458/// version itself — is untouched; the DEK row's validator moves with the
459/// blob row's, so the next rotation can guard both with the one `ETag` the
460/// caller holds.
461///
462/// # Errors
463///
464/// [`StoreError::NotFound`], [`StoreError::Stale`] when the row's revision is
465/// not `expected_revision`, [`StoreError::Db`], [`StoreError::Internal`]
466/// when the RNG fails.
467pub async fn replace_wraps(
468    db: &dyn Database,
469    subject: &str,
470    blob_id: &str,
471    expected_revision: i64,
472    wraps: &str,
473    ts: &str,
474) -> Result<(), StoreError> {
475    let mut fresh = fresh_revision()?;
476    if fresh == expected_revision {
477        fresh ^= 1;
478    }
479    let expected = SeaValue::BigInt(Some(expected_revision));
480    let written = SeaValue::BigInt(Some(fresh));
481    db.batch_atomic(&[
482        Statement::with_values(
483            "UPDATE sealed_blobs SET wraps = ?, updated_at = ?, revision = ? \
484             WHERE subject = ? AND blob_id = ? AND revision = ?",
485            vec![
486                text(wraps),
487                text(ts),
488                written.clone(),
489                text(subject),
490                text(blob_id),
491                expected,
492            ],
493        ),
494        Statement::with_values(
495            "UPDATE sealed_deks SET revision = ? \
496             WHERE subject = ? AND blob_id = ? AND revision = ?",
497            vec![
498                written,
499                text(subject),
500                text(blob_id),
501                SeaValue::BigInt(Some(expected_revision)),
502            ],
503        ),
504    ])
505    .await?;
506    // `batch_atomic` reports no per-statement row counts, so the read-back is
507    // the verdict: the row carries `fresh` only if this request wrote it.
508    match load(db, subject, blob_id).await? {
509        Some(row) if row.revision == fresh => Ok(()),
510        Some(_) => Err(StoreError::Stale),
511        None => Err(StoreError::NotFound),
512    }
513}
514
515/// A rotation: the new payload, wraps, version and a **fresh** DEK land
516/// together, writing a **fresh** validator (see `replace_wraps`). Both
517/// statements are guarded by `expected_revision` (the `ETag`), so of two
518/// concurrent rotations at most one matches; the loser writes nothing — in
519/// particular it can never pair its new ciphertext with the winner's DEK —
520/// and the read-back reports [`StoreError::Stale`]. Only the winner goes on
521/// to free the object key the previous body occupied; keys name exactly one
522/// parked body, so that delete cannot touch the new one.
523///
524/// # Errors
525///
526/// [`StoreError::NotFound`], [`StoreError::Stale`] when the row's revision is
527/// not `expected_revision` or its version is not the body's `version - 1`,
528/// [`StoreError::Db`], [`StoreError::Internal`] when the RNG fails.
529pub async fn rotate(
530    db: &dyn Database,
531    subject: &str,
532    blob: &CreateBlob,
533    sealed: &SealedBody,
534    ts: &str,
535    expected_revision: i64,
536    store: Option<&dyn cratefield_core::Blob>,
537) -> Result<(), StoreError> {
538    let old_key = load(db, subject, &blob.blob_id)
539        .await?
540        .ok_or(StoreError::NotFound)?
541        .object_key;
542    let previous_version = blob.version.saturating_sub(1);
543    let mut fresh = fresh_revision()?;
544    if fresh == expected_revision {
545        fresh ^= 1;
546    }
547    let expected = SeaValue::BigInt(Some(expected_revision));
548    let written = SeaValue::BigInt(Some(fresh));
549    let (outer_ct, new_key): (SeaValue, SeaValue) = match &sealed.object_key {
550        None => (bytes(&sealed.outer_ct), SeaValue::String(None)),
551        Some(key) => (SeaValue::Bytes(None), text(key)),
552    };
553    db.batch_atomic(&[
554        Statement::with_values(
555            "UPDATE sealed_blobs SET version = ?, wraps = ?, ciphertext_len = ?, outer_ct = ?, \
556             object_key = ?, updated_at = ?, revision = ? \
557             WHERE subject = ? AND blob_id = ? AND version = ? AND revision = ?",
558            vec![
559                SeaValue::BigInt(Some(i64::from(blob.version))),
560                text(&wraps_json(&blob.wraps)?),
561                big(sealed.ciphertext_len),
562                outer_ct,
563                new_key,
564                text(ts),
565                written.clone(),
566                text(subject),
567                text(&blob.blob_id),
568                SeaValue::BigInt(Some(i64::from(previous_version))),
569                expected.clone(),
570            ],
571        ),
572        Statement::with_values(
573            "UPDATE sealed_deks SET kms_provider = ?, kms_key_ref = ?, wrapped_dek = ?, \
574             revision = ? WHERE subject = ? AND blob_id = ? AND revision = ?",
575            vec![
576                text(sealed.provider),
577                text(&sealed.key_ref),
578                bytes(&sealed.wrapped_dek),
579                written,
580                text(subject),
581                text(&blob.blob_id),
582                expected,
583            ],
584        ),
585    ])
586    .await?;
587    // The read-back is the verdict (see `replace_wraps`): the row carries
588    // this request's validator only if this request won.
589    match load(db, subject, &blob.blob_id).await? {
590        Some(row) if row.revision == fresh => {}
591        Some(_) => return Err(StoreError::Stale),
592        None => return Err(StoreError::NotFound),
593    }
594    // An object the previous version parked in the blob store is superseded
595    // by this one; removing it here (after the commit names the new key)
596    // keeps exactly one live body per row. Best effort: a failure leaves an
597    // unopenable orphan, never a live one.
598    if let (Some(old_key), Some(store)) = (old_key, store)
599        && let Err(err) = store.delete(&old_key).await
600    {
601        tracing::warn!(error = %err, "sealed: the previous body object could not be removed");
602    }
603    Ok(())
604}
605
606/// The crypto-shred: the DEK row goes first, then the blob row, in one
607/// atomic batch. Returns the blob-store object key to remove afterwards —
608/// from the moment this returns, every remaining copy of the body anywhere
609/// is inert.
610///
611/// # Errors
612///
613/// [`StoreError::NotFound`], [`StoreError::Db`].
614pub async fn erase(
615    db: &dyn Database,
616    subject: &str,
617    blob_id: &str,
618) -> Result<Option<String>, StoreError> {
619    let row = load(db, subject, blob_id)
620        .await?
621        .ok_or(StoreError::NotFound)?;
622    db.batch_atomic(&[
623        Statement::with_values(
624            "DELETE FROM sealed_deks WHERE subject = ? AND blob_id = ?",
625            vec![text(subject), text(blob_id)],
626        ),
627        Statement::with_values(
628            "DELETE FROM sealed_blobs WHERE subject = ? AND blob_id = ?",
629            vec![text(subject), text(blob_id)],
630        ),
631    ])
632    .await?;
633    Ok(row.object_key)
634}
cratefield_module_sealed/store.rs

cratefield_module_sealed/
store.rs
