diff --git a/.sqlx/query-dd1b61d6ec81fd891d4effd3b51e6c22308b878acdc5355dfcb04c5664c9463b.json b/.sqlx/query-03f129e4984e1bed9e87294adc9caf1730906d889101b9039113ec8aa234618d.json similarity index 84% rename from .sqlx/query-dd1b61d6ec81fd891d4effd3b51e6c22308b878acdc5355dfcb04c5664c9463b.json rename to .sqlx/query-03f129e4984e1bed9e87294adc9caf1730906d889101b9039113ec8aa234618d.json index 236230f..910d998 100644 --- a/.sqlx/query-dd1b61d6ec81fd891d4effd3b51e6c22308b878acdc5355dfcb04c5664c9463b.json +++ b/.sqlx/query-03f129e4984e1bed9e87294adc9caf1730906d889101b9039113ec8aa234618d.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "SELECT storage_key, mime_type, size_bytes FROM blobs WHERE cid = $1", + "query": "SELECT storage_key, mime_type, size_bytes FROM blobs WHERE cid = $1 LIMIT 1", "describe": { "columns": [ { @@ -30,5 +30,5 @@ false ] }, - "hash": "dd1b61d6ec81fd891d4effd3b51e6c22308b878acdc5355dfcb04c5664c9463b" + "hash": "03f129e4984e1bed9e87294adc9caf1730906d889101b9039113ec8aa234618d" } diff --git a/.sqlx/query-0890b2c7c921005f58ed0e57b6e062b2085ce804a4cccb27b4ae2ba6711f24c4.json b/.sqlx/query-155efbae4cd55f73ec0709dda7b18a76e92065e6ae4a6081bd38a19821fbfcc3.json similarity index 67% rename from .sqlx/query-0890b2c7c921005f58ed0e57b6e062b2085ce804a4cccb27b4ae2ba6711f24c4.json rename to .sqlx/query-155efbae4cd55f73ec0709dda7b18a76e92065e6ae4a6081bd38a19821fbfcc3.json index 14db6c2..3f02003 100644 --- a/.sqlx/query-0890b2c7c921005f58ed0e57b6e062b2085ce804a4cccb27b4ae2ba6711f24c4.json +++ b/.sqlx/query-155efbae4cd55f73ec0709dda7b18a76e92065e6ae4a6081bd38a19821fbfcc3.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "SELECT COALESCE(SUM(size_bytes), 0)::BIGINT as \"total!\" FROM blobs", + "query": "SELECT COALESCE(SUM(size_bytes), 0)::BIGINT as \"total!\"\n FROM (SELECT DISTINCT cid, size_bytes FROM blobs) t", "describe": { "columns": [ { @@ -16,5 +16,5 @@ null ] }, - "hash": "0890b2c7c921005f58ed0e57b6e062b2085ce804a4cccb27b4ae2ba6711f24c4" + "hash": "155efbae4cd55f73ec0709dda7b18a76e92065e6ae4a6081bd38a19821fbfcc3" } diff --git a/.sqlx/query-62942bd21d545eb15bfea4f46378b6c2ebfe12b8bc9e27c63a6c0f77a9105303.json b/.sqlx/query-5996484ff0f8dbc3b278cfd01b8375dbf7bf6da8d903145b12871dda6e1fd5d9.json similarity index 67% rename from .sqlx/query-62942bd21d545eb15bfea4f46378b6c2ebfe12b8bc9e27c63a6c0f77a9105303.json rename to .sqlx/query-5996484ff0f8dbc3b278cfd01b8375dbf7bf6da8d903145b12871dda6e1fd5d9.json index f3ae0be..df39367 100644 --- a/.sqlx/query-62942bd21d545eb15bfea4f46378b6c2ebfe12b8bc9e27c63a6c0f77a9105303.json +++ b/.sqlx/query-5996484ff0f8dbc3b278cfd01b8375dbf7bf6da8d903145b12871dda6e1fd5d9.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "SELECT cid, takedown_ref FROM blobs WHERE cid = $1", + "query": "SELECT cid, takedown_ref FROM blobs WHERE cid = $1 ORDER BY takedown_ref NULLS LAST LIMIT 1", "describe": { "columns": [ { @@ -24,5 +24,5 @@ true ] }, - "hash": "62942bd21d545eb15bfea4f46378b6c2ebfe12b8bc9e27c63a6c0f77a9105303" + "hash": "5996484ff0f8dbc3b278cfd01b8375dbf7bf6da8d903145b12871dda6e1fd5d9" } diff --git a/.sqlx/query-8844d942ef2810afc386e5a9838624ee07a43c380d2df31efdba5cf299aab571.json b/.sqlx/query-8844d942ef2810afc386e5a9838624ee07a43c380d2df31efdba5cf299aab571.json new file mode 100644 index 0000000..dfc4e45 --- /dev/null +++ b/.sqlx/query-8844d942ef2810afc386e5a9838624ee07a43c380d2df31efdba5cf299aab571.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT storage_key as \"storage_key!\" FROM blobs b\n WHERE created_by_user = $1\n AND NOT EXISTS (\n SELECT 1 FROM blobs o\n WHERE o.cid = b.cid AND o.created_by_user <> $1\n )", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "storage_key!", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "Uuid" + ] + }, + "nullable": [ + false + ] + }, + "hash": "8844d942ef2810afc386e5a9838624ee07a43c380d2df31efdba5cf299aab571" +} diff --git a/.sqlx/query-8afea2b745385348f4c78b51f74145d6718bfcf9a3a0c218109ec691aeb930ba.json b/.sqlx/query-996e5513fb55670fe3304a6046381e377da6a187dfa3347bd285078a7b4410f2.json similarity index 74% rename from .sqlx/query-8afea2b745385348f4c78b51f74145d6718bfcf9a3a0c218109ec691aeb930ba.json rename to .sqlx/query-996e5513fb55670fe3304a6046381e377da6a187dfa3347bd285078a7b4410f2.json index 026d3ff..7857cc0 100644 --- a/.sqlx/query-8afea2b745385348f4c78b51f74145d6718bfcf9a3a0c218109ec691aeb930ba.json +++ b/.sqlx/query-996e5513fb55670fe3304a6046381e377da6a187dfa3347bd285078a7b4410f2.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "INSERT INTO blobs (cid, mime_type, size_bytes, created_by_user, storage_key)\n VALUES ($1, $2, $3, $4, $5)\n ON CONFLICT (cid) DO NOTHING RETURNING cid", + "query": "INSERT INTO blobs (cid, mime_type, size_bytes, created_by_user, storage_key)\n VALUES ($1, $2, $3, $4, $5)\n ON CONFLICT (cid, created_by_user) DO NOTHING RETURNING cid", "describe": { "columns": [ { @@ -22,5 +22,5 @@ false ] }, - "hash": "8afea2b745385348f4c78b51f74145d6718bfcf9a3a0c218109ec691aeb930ba" + "hash": "996e5513fb55670fe3304a6046381e377da6a187dfa3347bd285078a7b4410f2" } diff --git a/.sqlx/query-6131bb5b39ca81bdbb193c0a9867bead8d9f3d793ad4eca97a79d166467a5052.json b/.sqlx/query-9fb9e128076b20ff067d01955221488ce7e5b886dba0529fb073c3e0461fe030.json similarity index 65% rename from .sqlx/query-6131bb5b39ca81bdbb193c0a9867bead8d9f3d793ad4eca97a79d166467a5052.json rename to .sqlx/query-9fb9e128076b20ff067d01955221488ce7e5b886dba0529fb073c3e0461fe030.json index 5c07cbf..d05f896 100644 --- a/.sqlx/query-6131bb5b39ca81bdbb193c0a9867bead8d9f3d793ad4eca97a79d166467a5052.json +++ b/.sqlx/query-9fb9e128076b20ff067d01955221488ce7e5b886dba0529fb073c3e0461fe030.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "SELECT storage_key FROM blobs WHERE cid = $1", + "query": "SELECT storage_key FROM blobs WHERE cid = $1 LIMIT 1", "describe": { "columns": [ { @@ -18,5 +18,5 @@ false ] }, - "hash": "6131bb5b39ca81bdbb193c0a9867bead8d9f3d793ad4eca97a79d166467a5052" + "hash": "9fb9e128076b20ff067d01955221488ce7e5b886dba0529fb073c3e0461fe030" } diff --git a/.sqlx/query-f59010ecdd7f782489e0e03288a06dacd72b33d04c1e2b98475018ad25485852.json b/.sqlx/query-f59010ecdd7f782489e0e03288a06dacd72b33d04c1e2b98475018ad25485852.json deleted file mode 100644 index 218f7cd..0000000 --- a/.sqlx/query-f59010ecdd7f782489e0e03288a06dacd72b33d04c1e2b98475018ad25485852.json +++ /dev/null @@ -1,22 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT storage_key as \"storage_key!\" FROM blobs WHERE created_by_user = $1", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "storage_key!", - "type_info": "Text" - } - ], - "parameters": { - "Left": [ - "Uuid" - ] - }, - "nullable": [ - false - ] - }, - "hash": "f59010ecdd7f782489e0e03288a06dacd72b33d04c1e2b98475018ad25485852" -} diff --git a/crates/tranquil-api/src/repo/blob.rs b/crates/tranquil-api/src/repo/blob.rs index aa0f7b1..1e2df0f 100644 --- a/crates/tranquil-api/src/repo/blob.rs +++ b/crates/tranquil-api/src/repo/blob.rs @@ -148,7 +148,13 @@ pub async fn upload_blob( size, cid_str ); - match state + if let Err(e) = state.blob_store.copy(&temp_key, &storage_key).await { + let _ = state.blob_store.delete(&temp_key).await; + error!("Failed to copy blob to final location: {:?}", e); + return Err(ApiError::InternalError(Some("Failed to store blob".into()))); + } + + if let Err(e) = state .repos .blob .insert_blob( @@ -160,24 +166,9 @@ pub async fn upload_blob( ) .await { - Ok(_) => {} - Err(e) => { - let _ = state.blob_store.delete(&temp_key).await; - error!("Failed to insert blob record: {:?}", e); - return Err(ApiError::InternalError(None)); - } - }; - - if let Err(e) = state.blob_store.copy(&temp_key, &storage_key).await { let _ = state.blob_store.delete(&temp_key).await; - if let Err(db_err) = state.repos.blob.delete_blob_by_cid(&cid_link).await { - error!( - "Failed to clean up orphaned blob record after copy failure: {:?}", - db_err - ); - } - error!("Failed to copy blob to final location: {:?}", e); - return Err(ApiError::InternalError(Some("Failed to store blob".into()))); + error!("Failed to insert blob record: {:?}", e); + return Err(ApiError::InternalError(None)); } let _ = state.blob_store.delete(&temp_key).await; diff --git a/crates/tranquil-db/src/postgres/blob.rs b/crates/tranquil-db/src/postgres/blob.rs index 6e69a83..f0a1f48 100644 --- a/crates/tranquil-db/src/postgres/blob.rs +++ b/crates/tranquil-db/src/postgres/blob.rs @@ -33,7 +33,7 @@ impl BlobRepository for PostgresBlobRepository { let result = sqlx::query_scalar!( r#"INSERT INTO blobs (cid, mime_type, size_bytes, created_by_user, storage_key) VALUES ($1, $2, $3, $4, $5) - ON CONFLICT (cid) DO NOTHING RETURNING cid"#, + ON CONFLICT (cid, created_by_user) DO NOTHING RETURNING cid"#, cid.as_str(), mime_type, size_bytes, @@ -49,7 +49,7 @@ impl BlobRepository for PostgresBlobRepository { async fn get_blob_metadata(&self, cid: &CidLink) -> Result, DbError> { let result = sqlx::query!( - "SELECT storage_key, mime_type, size_bytes FROM blobs WHERE cid = $1", + "SELECT storage_key, mime_type, size_bytes FROM blobs WHERE cid = $1 LIMIT 1", cid.as_str() ) .fetch_optional(&self.pool) @@ -68,7 +68,7 @@ impl BlobRepository for PostgresBlobRepository { cid: &CidLink, ) -> Result, DbError> { let result = sqlx::query!( - "SELECT cid, takedown_ref FROM blobs WHERE cid = $1", + "SELECT cid, takedown_ref FROM blobs WHERE cid = $1 ORDER BY takedown_ref NULLS LAST LIMIT 1", cid.as_str() ) .fetch_optional(&self.pool) @@ -86,11 +86,13 @@ impl BlobRepository for PostgresBlobRepository { } async fn get_blob_storage_key(&self, cid: &CidLink) -> Result, DbError> { - let result = - sqlx::query_scalar!("SELECT storage_key FROM blobs WHERE cid = $1", cid.as_str()) - .fetch_optional(&self.pool) - .await - .map_err(map_sqlx_error)?; + let result = sqlx::query_scalar!( + "SELECT storage_key FROM blobs WHERE cid = $1 LIMIT 1", + cid.as_str() + ) + .fetch_optional(&self.pool) + .await + .map_err(map_sqlx_error)?; Ok(result) } @@ -147,7 +149,8 @@ impl BlobRepository for PostgresBlobRepository { async fn sum_blob_storage(&self) -> Result { let result = sqlx::query_scalar!( - r#"SELECT COALESCE(SUM(size_bytes), 0)::BIGINT as "total!" FROM blobs"# + r#"SELECT COALESCE(SUM(size_bytes), 0)::BIGINT as "total!" + FROM (SELECT DISTINCT cid, size_bytes FROM blobs) t"# ) .fetch_one(&self.pool) .await @@ -193,7 +196,12 @@ impl BlobRepository for PostgresBlobRepository { async fn get_blob_storage_keys_by_user(&self, user_id: Uuid) -> Result, DbError> { let results = sqlx::query_scalar!( - r#"SELECT storage_key as "storage_key!" FROM blobs WHERE created_by_user = $1"#, + r#"SELECT storage_key as "storage_key!" FROM blobs b + WHERE created_by_user = $1 + AND NOT EXISTS ( + SELECT 1 FROM blobs o + WHERE o.cid = b.cid AND o.created_by_user <> $1 + )"#, user_id ) .fetch_all(&self.pool) diff --git a/crates/tranquil-db/src/postgres/infra.rs b/crates/tranquil-db/src/postgres/infra.rs index 2ee3ec1..445cbae 100644 --- a/crates/tranquil-db/src/postgres/infra.rs +++ b/crates/tranquil-db/src/postgres/infra.rs @@ -1011,11 +1011,13 @@ impl InfraRepository for PostgresInfraRepository { } async fn get_blob_storage_key_by_cid(&self, cid: &CidLink) -> Result, DbError> { - let result = - sqlx::query_scalar!("SELECT storage_key FROM blobs WHERE cid = $1", cid.as_str()) - .fetch_optional(&self.pool) - .await - .map_err(map_sqlx_error)?; + let result = sqlx::query_scalar!( + "SELECT storage_key FROM blobs WHERE cid = $1 LIMIT 1", + cid.as_str() + ) + .fetch_optional(&self.pool) + .await + .map_err(map_sqlx_error)?; Ok(result) } diff --git a/crates/tranquil-pds/tests/store_parity.rs b/crates/tranquil-pds/tests/store_parity.rs index 42334fc..fc1f656 100644 --- a/crates/tranquil-pds/tests/store_parity.rs +++ b/crates/tranquil-pds/tests/store_parity.rs @@ -986,6 +986,86 @@ async fn parity_blob_duplicate_insert() { assert_eq!(pg_dup, store_dup); } +#[tokio::test(flavor = "multi_thread")] +async fn parity_blob_shared_between_repos() { + let f = ParityFixture::new().await; + + let did_a = test_did("shareda"); + let did_b = test_did("sharedb"); + let (pg_a, store_a) = seed_repos(&f, &did_a, &test_handle("shareda")).await; + let (pg_b, store_b) = seed_repos(&f, &did_b, &test_handle("sharedb")).await; + + let cid = test_cid(210); + + let pg_first = + f.pg.blob + .insert_blob(&cid, "image/png", 100, pg_a, "blobs/shared.png") + .await + .unwrap(); + let store_first = f + .store + .blob + .insert_blob(&cid, "image/png", 100, store_a, "blobs/shared.png") + .await + .unwrap(); + assert_eq!(pg_first, store_first); + + let pg_second = + f.pg.blob + .insert_blob(&cid, "image/png", 100, pg_b, "blobs/shared.png") + .await + .unwrap(); + let store_second = f + .store + .blob + .insert_blob(&cid, "image/png", 100, store_b, "blobs/shared.png") + .await + .unwrap(); + assert_eq!(pg_second, store_second); + assert!(pg_second.is_some()); + + for (pg_uid, store_uid) in [(pg_a, store_a), (pg_b, store_b)] { + assert_eq!(f.pg.blob.count_blobs_by_user(pg_uid).await.unwrap(), 1); + assert_eq!( + f.store.blob.count_blobs_by_user(store_uid).await.unwrap(), + 1 + ); + assert_eq!( + f.pg.blob + .list_blobs_by_user(pg_uid, None, 100) + .await + .unwrap(), + vec![cid.clone()] + ); + assert_eq!( + f.store + .blob + .list_blobs_by_user(store_uid, None, 100) + .await + .unwrap(), + vec![cid.clone()] + ); + assert!( + f.pg.blob + .get_blob_storage_keys_by_user(pg_uid) + .await + .unwrap() + .is_empty() + ); + assert!( + f.store + .blob + .get_blob_storage_keys_by_user(store_uid) + .await + .unwrap() + .is_empty() + ); + } + + assert_eq!(f.pg.blob.sum_blob_storage().await.unwrap(), 100); + assert_eq!(f.store.blob.sum_blob_storage().await.unwrap(), 100); +} + #[tokio::test] async fn parity_get_all_records() { let f = ParityFixture::new().await; diff --git a/crates/tranquil-store/src/metastore/blob_ops.rs b/crates/tranquil-store/src/metastore/blob_ops.rs index 50c496f..69f9165 100644 --- a/crates/tranquil-store/src/metastore/blob_ops.rs +++ b/crates/tranquil-store/src/metastore/blob_ops.rs @@ -7,7 +7,10 @@ use smallvec::SmallVec; use uuid::Uuid; use super::MetastoreError; -use super::blobs::{BlobMetaValue, blob_by_cid_key, blob_meta_key, blob_user_prefix, blobs_prefix}; +use super::blobs::{ + BlobContentValue, BlobMetaValue, blob_by_cid_key, blob_by_cid_prefix, blob_meta_key, + blob_user_prefix, +}; use super::commit_ops::{RecordBlobsValue, record_blobs_user_prefix}; use super::encoding::{KeyReader, exclusive_upper_bound}; use super::keys::{KeyTag, UserHash}; @@ -55,69 +58,62 @@ impl BlobOps { let user_hash = self.resolve_user_hash(created_by_user)?; let cid_str = cid.as_str(); - - let cid_index_key = blob_by_cid_key(cid_str); - let existing = self + let marker_key = blob_meta_key(user_hash, cid_str); + if self .repo_data - .get(cid_index_key.as_slice()) - .map_err(MetastoreError::Fjall)?; - if existing.is_some() { + .get(marker_key.as_slice()) + .map_err(MetastoreError::Fjall)? + .is_some() + { return Ok(None); } - let value = BlobMetaValue { - size_bytes, - mime_type: mime_type.to_owned(), - storage_key: storage_key.to_owned(), - takedown_ref: None, - created_at_ms: chrono::Utc::now().timestamp_millis(), + let cid_index_key = blob_by_cid_key(cid_str); + let content = match point_lookup( + &self.repo_data, + cid_index_key.as_slice(), + BlobContentValue::deserialize, + "corrupt blob_content value", + )? { + Some(mut existing) => { + existing.ref_count = existing.ref_count.saturating_add(1); + existing + } + None => BlobContentValue { + meta: BlobMetaValue { + size_bytes, + mime_type: mime_type.to_owned(), + storage_key: storage_key.to_owned(), + takedown_ref: None, + created_at_ms: chrono::Utc::now().timestamp_millis(), + }, + ref_count: 1, + }, }; - let primary_key = blob_meta_key(user_hash, cid_str); - let mut batch = self.db.batch(); - batch.insert(&self.repo_data, primary_key.as_slice(), value.serialize()); + batch.insert(&self.repo_data, marker_key.as_slice(), &[] as &[u8]); batch.insert( &self.repo_data, cid_index_key.as_slice(), - user_hash.raw().to_be_bytes(), + content.serialize(), ); batch.commit().map_err(MetastoreError::Fjall)?; Ok(Some(cid.clone())) } - fn lookup_user_hash_by_cid(&self, cid_str: &str) -> Result, MetastoreError> { - let key = blob_by_cid_key(cid_str); - match self - .repo_data - .get(key.as_slice()) - .map_err(MetastoreError::Fjall)? - { - Some(raw) => { - let arr: [u8; 8] = raw - .as_ref() - .try_into() - .map_err(|_| MetastoreError::CorruptData("blob_by_cid value not 8 bytes"))?; - Ok(Some(UserHash::from_raw(u64::from_be_bytes(arr)))) - } - None => Ok(None), - } + fn get_blob_content(&self, cid: &CidLink) -> Result, MetastoreError> { + point_lookup( + &self.repo_data, + blob_by_cid_key(cid.as_str()).as_slice(), + BlobContentValue::deserialize, + "corrupt blob_content value", + ) } fn get_blob_value(&self, cid: &CidLink) -> Result, MetastoreError> { - let cid_str = cid.as_str(); - let user_hash = match self.lookup_user_hash_by_cid(cid_str)? { - Some(h) => h, - None => return Ok(None), - }; - let key = blob_meta_key(user_hash, cid_str); - point_lookup( - &self.repo_data, - key.as_slice(), - BlobMetaValue::deserialize, - "corrupt blob_meta value", - ) + Ok(self.get_blob_content(cid)?.map(|c| c.meta)) } pub fn get_blob_metadata( @@ -186,14 +182,14 @@ impl BlobOps { } pub fn sum_blob_storage(&self) -> Result { - let prefix = blobs_prefix(); + let prefix = blob_by_cid_prefix(); self.repo_data .prefix(prefix.as_slice()) .try_fold(0i64, |acc, guard| { let (_, val_bytes) = guard.into_inner().map_err(MetastoreError::Fjall)?; - let value = BlobMetaValue::deserialize(&val_bytes) - .ok_or(MetastoreError::CorruptData("corrupt blob_meta in sum"))?; - Ok::<_, MetastoreError>(acc.saturating_add(value.size_bytes)) + let content = BlobContentValue::deserialize(&val_bytes) + .ok_or(MetastoreError::CorruptData("corrupt blob_content in sum"))?; + Ok::<_, MetastoreError>(acc.saturating_add(content.meta.size_bytes)) }) } @@ -202,50 +198,34 @@ impl BlobOps { cid: &CidLink, takedown_ref: Option<&str>, ) -> Result { - let cid_str = cid.as_str(); - let user_hash = match self.lookup_user_hash_by_cid(cid_str)? { - Some(h) => h, - None => return Ok(false), - }; - let key = blob_meta_key(user_hash, cid_str); - let mut value = match point_lookup( - &self.repo_data, - key.as_slice(), - BlobMetaValue::deserialize, - "corrupt blob_meta value", - )? { - Some(v) => v, + let mut content = match self.get_blob_content(cid)? { + Some(c) => c, None => return Ok(false), }; - value.takedown_ref = takedown_ref.map(str::to_owned); + content.meta.takedown_ref = takedown_ref.map(str::to_owned); let mut batch = self.db.batch(); - batch.insert(&self.repo_data, key.as_slice(), value.serialize()); + batch.insert( + &self.repo_data, + blob_by_cid_key(cid.as_str()).as_slice(), + content.serialize(), + ); batch.commit().map_err(MetastoreError::Fjall)?; Ok(true) } pub fn delete_blob_by_cid(&self, cid: &CidLink) -> Result { - let cid_str = cid.as_str(); - let user_hash = match self.lookup_user_hash_by_cid(cid_str)? { - Some(h) => h, - None => return Ok(false), - }; - - let primary_key = blob_meta_key(user_hash, cid_str); - let exists = self + let cid_index_key = blob_by_cid_key(cid.as_str()); + if self .repo_data - .get(primary_key.as_slice()) + .get(cid_index_key.as_slice()) .map_err(MetastoreError::Fjall)? - .is_some(); - if !exists { + .is_none() + { return Ok(false); } - let cid_index_key = blob_by_cid_key(cid_str); - let mut batch = self.db.batch(); - batch.remove(&self.repo_data, primary_key.as_slice()); batch.remove(&self.repo_data, cid_index_key.as_slice()); batch.commit().map_err(MetastoreError::Fjall)?; @@ -255,7 +235,6 @@ impl BlobOps { pub fn delete_blobs_by_user(&self, user_id: Uuid) -> Result { let user_hash = self.resolve_user_hash(user_id)?; let prefix = blob_user_prefix(user_hash); - let user_hash_bytes = user_hash.raw().to_be_bytes(); let (final_batch, remaining, total) = self .repo_data @@ -273,14 +252,25 @@ impl BlobOps { blob_meta_key(user_hash, &cid_str).as_slice(), ); let cid_index_key = blob_by_cid_key(&cid_str); - let owns_cid = self - .repo_data - .get(cid_index_key.as_slice()) - .map_err(MetastoreError::Fjall)? - .is_some_and(|raw| raw.as_ref() == user_hash_bytes); - if owns_cid { - batch.remove(&self.repo_data, cid_index_key.as_slice()); + + if let Some(mut content) = point_lookup( + &self.repo_data, + cid_index_key.as_slice(), + BlobContentValue::deserialize, + "corrupt blob_content value", + )? { + content.ref_count = content.ref_count.saturating_sub(1); + if content.ref_count == 0 { + batch.remove(&self.repo_data, cid_index_key.as_slice()); + } else { + batch.insert( + &self.repo_data, + cid_index_key.as_slice(), + content.serialize(), + ); + } } + let new_count = count + 1; if new_count >= DELETE_BATCH_SIZE { batch.commit().map_err(MetastoreError::Fjall)?; @@ -311,11 +301,14 @@ impl BlobOps { self.repo_data .prefix(prefix.as_slice()) .map(|guard| { - let (_, val_bytes) = guard.into_inner().map_err(MetastoreError::Fjall)?; - let value = BlobMetaValue::deserialize(&val_bytes) - .ok_or(MetastoreError::CorruptData("corrupt blob_meta value"))?; - Ok(value.storage_key) + let (key_bytes, _) = guard.into_inner().map_err(MetastoreError::Fjall)?; + let cid = parse_blob_cid_from_key(key_bytes.as_ref())?; + Ok(self + .get_blob_content(&cid)? + .filter(|c| c.ref_count == 1) + .map(|c| c.meta.storage_key)) }) + .filter_map(Result::transpose) .collect() } @@ -346,12 +339,7 @@ impl BlobOps { if acc.contains_key(&cid_str) { return Ok(()); } - let key = blob_meta_key(user_hash, &cid_str); - let exists = self - .repo_data - .get(key.as_slice()) - .map_err(MetastoreError::Fjall)? - .is_some(); + let exists = self.get_blob_content(&cid_link)?.is_some(); if !exists { acc.insert(cid_str, record_uri.clone()); } @@ -415,17 +403,11 @@ impl BlobOps { Ok(c) => c, Err(e) => return Some(Err(e)), }; - let key = blob_meta_key(user_hash, cid_link.as_str()); - match point_lookup( - &self.repo_data, - key.as_slice(), - BlobMetaValue::deserialize, - "corrupt blob_meta value", - ) { - Ok(Some(v)) => Some(Ok(tranquil_db_traits::BlobForExport { + match self.get_blob_content(&cid_link) { + Ok(Some(c)) => Some(Ok(tranquil_db_traits::BlobForExport { cid: cid_link, - storage_key: v.storage_key, - mime_type: v.mime_type, + storage_key: c.meta.storage_key, + mime_type: c.meta.mime_type, })), Ok(None) => None, Err(e) => Some(Err(e)), @@ -537,26 +519,6 @@ mod tests { ); } - #[test] - fn insert_same_cid_different_user_returns_none() { - let (_dir, ms) = open_fresh(); - let (user_a, _) = setup_user(&ms); - let (user_b, _) = setup_user(&ms); - let ops = ms.blob_ops(); - - let cid = test_cid_link(80); - assert!( - ops.insert_blob(&cid, "image/png", 100, user_a, "ka") - .unwrap() - .is_some() - ); - assert!( - ops.insert_blob(&cid, "image/png", 100, user_b, "kb") - .unwrap() - .is_none() - ); - } - #[test] fn get_blob_with_takedown_no_takedown() { let (_dir, ms) = open_fresh(); @@ -704,7 +666,7 @@ mod tests { ops.delete_blob_by_cid(&cid).unwrap(); - assert!(ops.lookup_user_hash_by_cid(cid.as_str()).unwrap().is_none()); + assert!(ops.get_blob_metadata(&cid).unwrap().is_none()); } #[test] @@ -735,7 +697,7 @@ mod tests { ops.delete_blobs_by_user(user_id).unwrap(); - assert!(ops.lookup_user_hash_by_cid(cid.as_str()).unwrap().is_none()); + assert!(ops.get_blob_metadata(&cid).unwrap().is_none()); } #[test] @@ -787,4 +749,52 @@ mod tests { assert_eq!(ops.count_blobs_by_user(user_b).unwrap(), 1); assert_eq!(ops.sum_blob_storage().unwrap(), 30); } + + #[test] + fn blob_shared_between_users() { + let (_dir, ms) = open_fresh(); + let (user_a, _) = setup_user(&ms); + let (user_b, _) = setup_user(&ms); + let ops = ms.blob_ops(); + + let cid = test_cid_link(80); + + assert!( + ops.insert_blob(&cid, "a/b", 10, user_a, "k") + .unwrap() + .is_some() + ); + assert!( + ops.insert_blob(&cid, "a/b", 10, user_b, "k") + .unwrap() + .is_some() + ); + assert!( + ops.insert_blob(&cid, "a/b", 10, user_b, "k") + .unwrap() + .is_none() + ); + + assert_eq!(ops.count_blobs_by_user(user_a).unwrap(), 1); + assert_eq!(ops.count_blobs_by_user(user_b).unwrap(), 1); + assert_eq!( + ops.list_blobs_by_user(user_b, None, 100).unwrap(), + vec![cid.clone()] + ); + + assert_eq!(ops.sum_blob_storage().unwrap(), 10); + assert!( + ops.get_blob_storage_keys_by_user(user_a) + .unwrap() + .is_empty() + ); + + ops.delete_blobs_by_user(user_a).unwrap(); + assert_eq!(ops.count_blobs_by_user(user_b).unwrap(), 1); + assert!(ops.get_blob_metadata(&cid).unwrap().is_some()); + assert_eq!( + ops.get_blob_storage_keys_by_user(user_b).unwrap(), + vec!["k".to_string()] + ); + } } diff --git a/crates/tranquil-store/src/metastore/blobs.rs b/crates/tranquil-store/src/metastore/blobs.rs index 94733ee..798a62c 100644 --- a/crates/tranquil-store/src/metastore/blobs.rs +++ b/crates/tranquil-store/src/metastore/blobs.rs @@ -33,6 +33,33 @@ impl BlobMetaValue { } } +const BLOB_CONTENT_SCHEMA_VERSION: u8 = 1; + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct BlobContentValue { + pub meta: BlobMetaValue, + pub ref_count: u32, +} + +impl BlobContentValue { + pub fn serialize(&self) -> Vec { + let payload = + postcard::to_allocvec(self).expect("BlobContentValue serialization cannot fail"); + let mut buf = Vec::with_capacity(1 + payload.len()); + buf.push(BLOB_CONTENT_SCHEMA_VERSION); + buf.extend_from_slice(&payload); + buf + } + + pub fn deserialize(bytes: &[u8]) -> Option { + let (&version, payload) = bytes.split_first()?; + match version { + BLOB_CONTENT_SCHEMA_VERSION => postcard::from_bytes(payload).ok(), + _ => None, + } + } +} + pub fn blob_meta_key(user_hash: UserHash, cid_str: &str) -> SmallVec<[u8; 128]> { KeyBuilder::new() .tag(KeyTag::BLOBS) @@ -59,6 +86,10 @@ pub fn blob_by_cid_key(cid_str: &str) -> SmallVec<[u8; 128]> { .build() } +pub fn blob_by_cid_prefix() -> SmallVec<[u8; 128]> { + KeyBuilder::new().tag(KeyTag::BLOB_BY_CID).build() +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/tranquil-store/src/metastore/mod.rs b/crates/tranquil-store/src/metastore/mod.rs index 50af461..15f382e 100644 --- a/crates/tranquil-store/src/metastore/mod.rs +++ b/crates/tranquil-store/src/metastore/mod.rs @@ -35,11 +35,12 @@ use std::sync::Arc; use fjall::{Database, Keyspace}; +use self::encoding::KeyReader; use self::keys::KeyTag; use self::partitions::Partition; use self::user_hash::UserHashMap; -const CURRENT_FORMAT_VERSION: u64 = 2; +const CURRENT_FORMAT_VERSION: u64 = 3; #[derive(Debug, Clone)] pub struct MetastoreConfig { @@ -240,6 +241,7 @@ impl Metastore { "upgrading metastore format and rebuilding derived indexes" ); repo_data.remove(records::record_by_cid_built_key().as_slice())?; + Self::migrate_blob_ownership(db, repo_data)?; repo_data.insert(version_key, version_bytes)?; db.persist(fjall::PersistMode::SyncData)?; Ok(()) @@ -254,6 +256,47 @@ impl Metastore { } } + fn migrate_blob_ownership(db: &Database, repo_data: &Keyspace) -> Result<(), MetastoreError> { + let entries: Vec<(Vec, Vec)> = repo_data + .prefix(blobs::blobs_prefix().as_slice()) + .map(|guard| { + let (k, v) = guard.into_inner()?; + Ok((k.as_ref().to_vec(), v.as_ref().to_vec())) + }) + .collect::>()?; + + for (key_bytes, val_bytes) in entries { + let Some(meta) = blobs::BlobMetaValue::deserialize(&val_bytes) else { + continue; + }; + let mut reader = KeyReader::new(&key_bytes); + reader.tag(); + reader.u64(); + let Some(cid_str) = reader.string() else { + continue; + }; + + let cid_index_key = blobs::blob_by_cid_key(cid_str.as_str()); + let content = match repo_data + .get(cid_index_key.as_slice())? + .and_then(|raw| blobs::BlobContentValue::deserialize(raw.as_ref())) + { + Some(mut existing) => { + existing.ref_count = existing.ref_count.saturating_add(1); + existing + } + None => blobs::BlobContentValue { meta, ref_count: 1 }, + }; + + let mut batch = db.batch(); + batch.insert(repo_data, cid_index_key.as_slice(), content.serialize()); + batch.insert(repo_data, key_bytes.as_slice(), &[] as &[u8]); + batch.commit()?; + } + + Ok(()) + } + pub fn path(&self) -> &Path { &self.path } diff --git a/migrations/20260906_blob_ownership_per_user.sql b/migrations/20260906_blob_ownership_per_user.sql new file mode 100644 index 0000000..46bcee7 --- /dev/null +++ b/migrations/20260906_blob_ownership_per_user.sql @@ -0,0 +1,2 @@ +ALTER TABLE blobs DROP CONSTRAINT IF EXISTS blobs_pkey; +ALTER TABLE blobs ADD PRIMARY KEY (cid, created_by_user);