Something went wrong. Try again.
Chris’s PDS implementation
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509//! Blob metadata persistence; opaque bytes remain in `BlobStore`.
use cid::Cid;use rusqlite::{Transaction, params};
use crate::{ domain::blob::{Blob, referenced_blob_cids, writable_blob_refs}, error::{Error, ErrorContext as _, Result},};
use super::{Database, now, query, require_account_active};
impl Database { pub(crate) fn is_public_blob(&self, did: &str, cid: &Cid) -> Result<bool> { self.query_one( "SELECT EXISTS(SELECT 1 FROM record_blobs WHERE did = ?1 AND cid = ?2)", params![did, cid.to_string()], |row| Ok(row.get(0)?), ) }
/// Reserve metadata before publishing blob bytes. /// /// # Errors /// /// Returns an error when metadata cannot be written, conflicts with an /// earlier upload, or deletion already owns this content-addressed key. pub fn begin_blob_upload(&self, did: &str, blob: &Blob) -> Result<()> { let connection = self.connection()?; let changed = connection.execute( "INSERT INTO blobs (did, cid, media_type, size, created_at) SELECT ?1, ?2, ?3, ?4, ?5 WHERE (SELECT status FROM accounts WHERE did = ?1) = 'active' ON CONFLICT(did, cid) DO UPDATE SET created_at = excluded.created_at WHERE blobs.delete_started_at IS NULL AND blobs.media_type = excluded.media_type AND blobs.size = excluded.size", params![ did, blob.cid.to_string(), blob.media_type, i64::try_from(blob.size).unwrap_or(i64::MAX), now(), ], )?; if changed == 0 { require_account_active(&connection, did)?; return Err(Error::Conflict( "blob metadata differs from its earlier upload or deletion is in progress" .to_owned(), )); } Ok(()) }
/// Mark a reserved blob as visible after its bytes are durable. /// /// # Errors /// /// Returns an error when the reservation disappeared or entered deletion. pub fn finish_blob_upload(&self, did: &str, cid: &Cid) -> Result<()> { let connection = self.connection()?; let changed = connection.execute( "UPDATE blobs SET available_at = COALESCE(available_at, ?3) WHERE did = ?1 AND cid = ?2 AND delete_started_at IS NULL AND (SELECT status FROM accounts WHERE did = ?1) = 'active'", params![did, cid.to_string(), now()], )?; if changed == 0 { require_account_active(&connection, did)?; return Err(Error::Internal(format!( "blob upload reservation disappeared for {did}/{cid}" ))); } Ok(()) }
/// Find available metadata for a content-addressed blob, or `None` when absent. /// /// # Errors /// /// Returns an error for failed lookup or invalid stored metadata. pub fn find_blob(&self, did: &str, cid: &Cid) -> Result<Option<Blob>> { self.query_optional( "SELECT media_type, size FROM blobs WHERE did = ?1 AND cid = ?2 AND delete_started_at IS NULL AND available_at IS NOT NULL", params![did, cid.to_string()], |row| decode_blob(*cid, row.get(0)?, row.get(1)?), ) }
pub(crate) fn validate_blob_refs(&self, did: &str, value: &serde_json::Value) -> Result<()> { for referenced in writable_blob_refs(value).map_err(Error::InvalidRequest)? { let stored = self.find_blob(did, &referenced.cid)?.ok_or_else(|| { Error::InvalidRequest(format!( "record references blob {} before it is uploaded", referenced.cid )) })?; if referenced.media_type != stored.media_type || referenced.size != stored.size { return Err(Error::InvalidRequest(format!( "record blob {} metadata does not match the uploaded blob", referenced.cid ))); } } Ok(()) }
pub(crate) fn all_blobs(&self) -> Result<Vec<(String, Blob)>> { self.query( "SELECT did, cid, media_type, size FROM blobs WHERE delete_started_at IS NULL AND available_at IS NOT NULL ORDER BY did, cid", [], |row| { let (did, cid, media_type, size) = <(_, String, _, _)>::try_from(row)?; Ok((did, decode_blob(parse_stored_cid(&cid)?, media_type, size)?)) }, ) }
pub(crate) fn claim_orphan_blobs( &self, created_before: i64, limit: usize, ) -> Result<Vec<(String, Cid)>> { let candidates = self.transaction(|transaction| { let candidates = query( transaction, "SELECT blobs.did, blobs.cid FROM blobs WHERE blobs.delete_started_at IS NOT NULL OR (blobs.created_at <= ?1 AND NOT EXISTS ( SELECT 1 FROM record_blobs WHERE record_blobs.did = blobs.did AND record_blobs.cid = blobs.cid ) AND NOT EXISTS ( SELECT 1 FROM space_record_blobs WHERE space_record_blobs.did = blobs.did AND space_record_blobs.cid = blobs.cid )) ORDER BY blobs.delete_started_at IS NULL, blobs.created_at, blobs.did, blobs.cid LIMIT ?2", params![created_before, i64::try_from(limit).unwrap_or(i64::MAX)], |row| Ok(<(String, String)>::try_from(row)?), )?; for (did, cid) in &candidates { transaction.execute( "UPDATE blobs SET delete_started_at = COALESCE(delete_started_at, ?3) WHERE did = ?1 AND cid = ?2", params![did, cid, now()], )?; } Ok(candidates) })?; candidates .into_iter() .map(|(did, cid)| Ok((did, parse_stored_cid(&cid)?))) .collect() }
pub(crate) fn finish_blob_deletion(&self, did: &str, cid: &Cid) -> Result<()> { let changed = self.connection()?.execute( "DELETE FROM blobs WHERE did = ?1 AND cid = ?2 AND delete_started_at IS NOT NULL", params![did, cid.to_string()], )?; if changed == 0 { return Err(Error::Internal(format!( "blob deletion claim disappeared for {did}/{cid}" ))); } Ok(()) }
pub(crate) fn referenced_blob_cids( &self, did: &str, since: Option<&str>, cursor: Option<&str>, limit: usize, ) -> Result<Vec<Cid>> { self.query( "SELECT DISTINCT record_blobs.cid FROM record_blobs WHERE record_blobs.did = ?1 AND (?2 IS NULL OR record_blobs.cid > ?2) AND ( ?3 IS NULL OR EXISTS ( SELECT 1 FROM repo_ops WHERE repo_ops.did = record_blobs.did AND repo_ops.collection = record_blobs.collection AND repo_ops.rkey = record_blobs.rkey AND repo_ops.rev > ?3 ) ) ORDER BY record_blobs.cid LIMIT ?4", params![did, cursor, since, i64::try_from(limit).unwrap_or(i64::MAX)], |row| parse_stored_cid(&row.get::<_, String>(0)?), ) }
pub(crate) fn missing_blob_references( &self, did: &str, cursor: Option<&str>, limit: usize, ) -> Result<Vec<(Cid, String)>> { self.query( "SELECT record_blobs.cid, MIN(record_blobs.collection || '/' || record_blobs.rkey) FROM record_blobs LEFT JOIN blobs ON blobs.did = record_blobs.did AND blobs.cid = record_blobs.cid AND blobs.delete_started_at IS NULL AND blobs.available_at IS NOT NULL WHERE record_blobs.did = ?1 AND blobs.cid IS NULL AND (?2 IS NULL OR record_blobs.cid > ?2) GROUP BY record_blobs.cid ORDER BY record_blobs.cid LIMIT ?3", params![did, cursor, i64::try_from(limit).unwrap_or(i64::MAX)], |row| { let (cid, path) = <(String, String)>::try_from(row)?; Ok((parse_stored_cid(&cid)?, format!("at://{did}/{path}"))) }, ) }}
fn parse_stored_cid(value: &str) -> Result<Cid> { value .parse() .context(Error::Internal, "stored blob CID is invalid")}
pub(crate) fn active_blob_ref_cids( transaction: &Transaction<'_>, did: &str, value: &serde_json::Value,) -> Result<Vec<String>> { let cids = referenced_blob_cids(value).into_iter().collect::<Vec<_>>(); for cid in &cids { let exists = transaction.query_row( "SELECT EXISTS( SELECT 1 FROM blobs WHERE did = ?1 AND cid = ?2 AND delete_started_at IS NULL AND available_at IS NOT NULL )", params![did, cid], |row| row.get::<_, bool>(0), )?; if !exists { return Err(Error::InvalidRequest(format!( "record references blob {cid} before it is uploaded" ))); } } Ok(cids)}
fn decode_blob(cid: Cid, media_type: String, size: i64) -> Result<Blob> { Ok(Blob { cid, media_type, size: u64::try_from(size) .map_err(|_| Error::Internal("stored blob size is negative".to_owned()))?, })}
#[cfg(test)]mod tests { use std::str::FromStr as _;
use super::*;
const DID: &str = "did:web:alice.test";
fn previous_blob_lookup(database: &Database, cid: &Cid) -> Result<Blob> { database .query_optional( "SELECT media_type, size FROM blobs WHERE did = ?1 AND cid = ?2 AND delete_started_at IS NULL AND available_at IS NOT NULL", params![DID, cid.to_string()], |row| decode_blob(*cid, row.get(0)?, row.get(1)?), )? .ok_or(Error::NotFound) }
fn assert_optional_blob_parity(database: &Database, reference: &Blob) { let previous = previous_blob_lookup(database, &reference.cid); let actual = database .find_blob(DID, &reference.cid) .and_then(|blob| blob.ok_or(Error::NotFound)); assert_eq!(format!("{actual:?}"), format!("{previous:?}")); let expected = previous .map_err(|error| { if matches!(error, Error::NotFound) { Error::InvalidRequest(format!( "record references blob {} before it is uploaded", reference.cid )) } else { error } }) .and_then(|stored| { if reference.media_type != stored.media_type || reference.size != stored.size { return Err(Error::InvalidRequest(format!( "record blob {} metadata does not match the uploaded blob", reference.cid ))); } Ok(()) }); let value = serde_json::json!({ "$type": "app.test.record", "image": {"$type": "blob", "ref": {"$link": reference.cid.to_string()}, "mimeType": reference.media_type, "size": reference.size}, }); let actual = database.validate_blob_refs(DID, &value); assert_eq!(format!("{actual:?}"), format!("{expected:?}")); }
#[test] fn optional_blob_lookup_preserves_visibility_and_reference_validation_errors() { use rusqlite::types::Value as SqlValue;
let database = Database::open_in_memory().unwrap(); let reference = Blob { cid: cpds_protocol::sha256_cid(0x55, b"blob"), media_type: "image/png".to_owned(), size: 3, }; assert!(database.find_blob(DID, &reference.cid).unwrap().is_none()); assert_optional_blob_parity(&database, &reference); database.connection().unwrap().execute_batch( "CREATE TEMP TABLE blobs (did, cid, media_type, size, available_at, delete_started_at)", ).unwrap(); for media_type in [ SqlValue::Text("image/png".to_owned()), SqlValue::Text("text/plain".to_owned()), SqlValue::Text("not a MIME type".to_owned()), SqlValue::Null, SqlValue::Integer(1), SqlValue::Blob(vec![0]), ] { for size in [ SqlValue::Integer(3), SqlValue::Integer(0), SqlValue::Integer(-1), SqlValue::Integer(i64::MAX), SqlValue::Real(3.5), SqlValue::Text("3".to_owned()), SqlValue::Null, SqlValue::Blob(vec![0]), ] { for available_at in [None, Some(1)] { for delete_started_at in [None, Some(1)] { { let connection = database.connection().unwrap(); connection.execute("DELETE FROM temp.blobs", []).unwrap(); connection .execute( "INSERT INTO temp.blobs VALUES (?1, ?2, ?3, ?4, ?5, ?6)", params![ DID, reference.cid.to_string(), media_type, size, available_at, delete_started_at ], ) .unwrap(); } assert_optional_blob_parity(&database, &reference); } } } } database .connection() .unwrap() .execute_batch("DROP TABLE temp.blobs; DROP TABLE main.blobs") .unwrap(); assert_optional_blob_parity(&database, &reference); assert!(matches!( database.find_blob(DID, &reference.cid), Err(Error::Database(_)) )); }
#[test] fn stored_blob_cid_context_preserves_values_and_exact_errors() { let cid = cpds_protocol::sha256_cid(0x55, b"blob"); let v0 = Cid::new_v0(*cid.hash()).unwrap().to_string(); let valid = cid.to_string(); for value in [ valid.as_str(), v0.as_str(), "", "invalid-cid", "b", "Qm", "é", " bafkreia", ] { let old = Cid::from_str(value) .map_err(|error| Error::Internal(format!("stored blob CID is invalid: {error}"))); let current = parse_stored_cid(value); assert_eq!(format!("{current:?}"), format!("{old:?}"), "{value:?}"); } }
#[test] fn repeated_uploads_cannot_change_content_addressed_metadata() { let database = Database::open_in_memory().unwrap(); database .ensure_account(DID, "alice.test", "password") .unwrap(); let blob = Blob { cid: Cid::from_str("bafkreibme22gw2h7y2h7tg2fhqotaqjucnbc24deqo72b6mkl2egezxhvy") .unwrap(), media_type: "image/png".to_owned(), size: 3, }; database.begin_blob_upload(DID, &blob).unwrap(); for available in [false, true] { if available { database.finish_blob_upload(DID, &blob.cid).unwrap(); } database.begin_blob_upload(DID, &blob).unwrap(); for inconsistent in [ Blob { media_type: "text/plain".to_owned(), ..blob.clone() }, Blob { size: 4, ..blob.clone() }, ] { assert!(matches!( database.begin_blob_upload(DID, &inconsistent), Err(Error::Conflict(_)) )); } } assert_eq!(database.find_blob(DID, &blob.cid).unwrap().unwrap(), blob); }
#[test] fn deactivation_prevents_blob_reservation_and_publication() { let database = Database::open_in_memory().unwrap(); database .ensure_account(DID, "alice.test", "password") .unwrap(); let blob = Blob { cid: Cid::from_str("bafkreibme22gw2h7y2h7tg2fhqotaqjucnbc24deqo72b6mkl2egezxhvy") .unwrap(), media_type: "image/png".to_owned(), size: 3, };
database.begin_blob_upload(DID, &blob).unwrap(); database .connection() .unwrap() .execute( "UPDATE accounts SET status = 'deactivated' WHERE did = ?1", [DID], ) .unwrap(); assert!(matches!( database.finish_blob_upload(DID, &blob.cid), Err(Error::Xrpc { name: "AccountDeactivated", .. }) )); assert!(matches!( database .find_blob(DID, &blob.cid) .and_then(|blob| blob.ok_or(Error::NotFound)), Err(Error::NotFound) ));
let other = Blob { cid: Cid::from_str("bafkreid7zfv2egb7qol7vr43m6xju2wwhf4l75uz2e6fmh5f3av7p7ddca") .unwrap(), ..blob }; assert!(matches!( database.begin_blob_upload(DID, &other), Err(Error::Xrpc { name: "AccountDeactivated", .. }) )); }}