diff --git a/.sqlx/query-8951136879711bca5b562c34f88e691a8ee16f370f6ef9b88ddb3873ddf2b45f.json b/.sqlx/query-8951136879711bca5b562c34f88e691a8ee16f370f6ef9b88ddb3873ddf2b45f.json new file mode 100644 index 0000000..91496ba --- /dev/null +++ b/.sqlx/query-8951136879711bca5b562c34f88e691a8ee16f370f6ef9b88ddb3873ddf2b45f.json @@ -0,0 +1,26 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT u.id as user_id, u.did\n FROM users u\n JOIN repos r ON r.user_id = u.id\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "user_id", + "type_info": "Uuid" + }, + { + "ordinal": 1, + "name": "did", + "type_info": "Text" + } + ], + "parameters": { + "Left": [] + }, + "nullable": [ + false, + false + ] + }, + "hash": "8951136879711bca5b562c34f88e691a8ee16f370f6ef9b88ddb3873ddf2b45f" +} diff --git a/.sqlx/query-e248d71f595abf0207b01bc2f4e1f312d0c96b0f2f5131dfc13bfbb42a79d886.json b/.sqlx/query-e248d71f595abf0207b01bc2f4e1f312d0c96b0f2f5131dfc13bfbb42a79d886.json new file mode 100644 index 0000000..fb7e990 --- /dev/null +++ b/.sqlx/query-e248d71f595abf0207b01bc2f4e1f312d0c96b0f2f5131dfc13bfbb42a79d886.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO blobs (cid, mime_type, size_bytes, created_by_user, storage_key)\n SELECT DISTINCT b.cid, b.mime_type, b.size_bytes, $1::uuid, b.storage_key\n FROM blobs b WHERE b.cid = $2\n ON CONFLICT (cid, created_by_user) DO NOTHING", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Uuid", + "Text" + ] + }, + "nullable": [] + }, + "hash": "e248d71f595abf0207b01bc2f4e1f312d0c96b0f2f5131dfc13bfbb42a79d886" +} diff --git a/crates/tranquil-db-traits/src/blob.rs b/crates/tranquil-db-traits/src/blob.rs index b40e551..de7600d 100644 --- a/crates/tranquil-db-traits/src/blob.rs +++ b/crates/tranquil-db-traits/src/blob.rs @@ -74,6 +74,8 @@ pub trait BlobRepository: Send + Sync { async fn get_blob_storage_keys_by_user(&self, user_id: Uuid) -> Result, DbError>; + async fn ensure_blob_ownership(&self, user_id: Uuid, cid: &CidLink) -> Result; + async fn insert_record_blobs( &self, repo_id: Uuid, diff --git a/crates/tranquil-db-traits/src/lib.rs b/crates/tranquil-db-traits/src/lib.rs index b71a3eb..fb42511 100644 --- a/crates/tranquil-db-traits/src/lib.rs +++ b/crates/tranquil-db-traits/src/lib.rs @@ -36,8 +36,8 @@ pub use repo::{ AccountStatus, ApplyCommitError, ApplyCommitInput, ApplyCommitResult, CommitEventData, EventBlockInline, EventBlocks, FullRecordInfo, ImportBlock, ImportRecord, ImportRepoError, PruneCount, RecordDelete, RecordInfo, RecordUpsert, RecordWithTakedown, RepoAccountInfo, - RepoEventNotifier, RepoEventReceiver, RepoEventType, RepoInfo, RepoListItem, RepoRepository, - RepoSeqEvent, RepoWithoutRev, SequencedEvent, UserNeedingRecordBlobsBackfill, + RepoEventNotifier, RepoEventReceiver, RepoEventType, RepoIdentity, RepoInfo, RepoListItem, + RepoRepository, RepoSeqEvent, RepoWithoutRev, SequencedEvent, UserNeedingRecordBlobsBackfill, UserWithoutBlocks, }; pub use scope::{DbScope, InvalidScopeError}; diff --git a/crates/tranquil-db-traits/src/repo.rs b/crates/tranquil-db-traits/src/repo.rs index 3520753..890ebf7 100644 --- a/crates/tranquil-db-traits/src/repo.rs +++ b/crates/tranquil-db-traits/src/repo.rs @@ -171,6 +171,12 @@ pub struct UserNeedingRecordBlobsBackfill { pub did: Did, } +#[derive(Debug, Clone)] +pub struct RepoIdentity { + pub user_id: Uuid, + pub did: Did, +} + #[derive(Debug, Clone, Serialize, Deserialize)] pub struct RepoSeqEvent { pub seq: SequenceNumber, @@ -545,6 +551,8 @@ pub trait RepoRepository: Send + Sync { limit: i64, ) -> Result, DbError>; + async fn get_all_repo_identities(&self) -> Result, DbError>; + async fn insert_record_blobs( &self, repo_id: Uuid, diff --git a/crates/tranquil-db/src/postgres/blob.rs b/crates/tranquil-db/src/postgres/blob.rs index 546acfe..3c6a185 100644 --- a/crates/tranquil-db/src/postgres/blob.rs +++ b/crates/tranquil-db/src/postgres/blob.rs @@ -202,6 +202,22 @@ impl BlobRepository for PostgresBlobRepository { Ok(results) } + async fn ensure_blob_ownership(&self, user_id: Uuid, cid: &CidLink) -> Result { + let result = sqlx::query!( + r#"INSERT INTO blobs (cid, mime_type, size_bytes, created_by_user, storage_key) + SELECT DISTINCT b.cid, b.mime_type, b.size_bytes, $1::uuid, b.storage_key + FROM blobs b WHERE b.cid = $2 + ON CONFLICT (cid, created_by_user) DO NOTHING"#, + user_id, + cid.as_str() + ) + .execute(&self.pool) + .await + .map_err(map_sqlx_error)?; + + Ok(result.rows_affected() > 0) + } + async fn insert_record_blobs( &self, repo_id: Uuid, diff --git a/crates/tranquil-db/src/postgres/repo.rs b/crates/tranquil-db/src/postgres/repo.rs index 6b10f6e..37440a7 100644 --- a/crates/tranquil-db/src/postgres/repo.rs +++ b/crates/tranquil-db/src/postgres/repo.rs @@ -4,8 +4,9 @@ use sqlx::PgPool; use tranquil_db_traits::{ AccountStatus, CommitEventData, DbError, EventBlockInline, EventBlocks, FullRecordInfo, ImportBlock, ImportRecord, ImportRepoError, PruneCount, RecordInfo, RecordWithTakedown, - RepoAccountInfo, RepoEventType, RepoInfo, RepoListItem, RepoRepository, RepoWithoutRev, - SequenceNumber, SequencedEvent, UserNeedingRecordBlobsBackfill, UserWithoutBlocks, + RepoAccountInfo, RepoEventType, RepoIdentity, RepoInfo, RepoListItem, RepoRepository, + RepoWithoutRev, SequenceNumber, SequencedEvent, UserNeedingRecordBlobsBackfill, + UserWithoutBlocks, }; use tranquil_types::{AtUri, CidLink, Did, Handle, Nsid, Rkey, Tid}; use uuid::Uuid; @@ -1650,6 +1651,28 @@ impl RepoRepository for PostgresRepoRepository { .collect() } + async fn get_all_repo_identities(&self) -> Result, DbError> { + let rows = sqlx::query!( + r#" + SELECT u.id as user_id, u.did + FROM users u + JOIN repos r ON r.user_id = u.id + "# + ) + .fetch_all(&self.pool) + .await + .map_err(map_sqlx_error)?; + + rows.into_iter() + .map(|r| { + Ok(RepoIdentity { + user_id: r.user_id, + did: column(r.did, col::USERS_DID)?, + }) + }) + .collect() + } + async fn insert_record_blobs( &self, repo_id: Uuid, diff --git a/crates/tranquil-pds/src/scheduled.rs b/crates/tranquil-pds/src/scheduled.rs index 438e3c6..f55a218 100644 --- a/crates/tranquil-pds/src/scheduled.rs +++ b/crates/tranquil-pds/src/scheduled.rs @@ -3,13 +3,16 @@ use cid::Cid; use ipld_core::ipld::Ipld; use jacquard_repo::commit::Commit; use jacquard_repo::storage::BlockStore; +use std::collections::BTreeSet; use std::str::FromStr; use std::sync::Arc; use std::time::Duration; use tokio::time::interval; use tokio_util::sync::CancellationToken; use tracing::{debug, error, info, warn}; -use tranquil_db_traits::{BlobRepository, RepoRepository, SsoRepository, UserRepository}; +use tranquil_db_traits::{ + BlobRepository, InfraRepository, RepoRepository, SsoRepository, UserRepository, +}; use tranquil_store::blockstore::CidBytes; use tranquil_store::bloom::BloomFilter; use tranquil_types::{AtUri, CidLink, Did}; @@ -307,6 +310,58 @@ async fn process_record_blobs( Ok((user_id, did, blob_refs_found)) } +const OWNERSHIP_CHUNK_SIZE: usize = 500; + +async fn process_blob_ownership( + repo_repo: &dyn RepoRepository, + blob_repo: &dyn BlobRepository, + block_store: &AnyBlockStore, + user_id: uuid::Uuid, + did: Did, +) -> Result<(uuid::Uuid, Did, u64), (uuid::Uuid, &'static str)> { + let records = repo_repo + .get_all_records(user_id) + .await + .map_err(|_| (user_id, "failed to fetch records"))?; + + let mut cids: BTreeSet = BTreeSet::new(); + + for chunk in records.chunks(OWNERSHIP_CHUNK_SIZE) { + futures::future::join_all(chunk.iter().map(|record| async move { + let cid = Cid::from_str(record.record_cid.as_str()).ok()?; + let block_bytes = block_store.get(&cid).await.ok()??; + let record_ipld: Ipld = serde_ipld_dagcbor::from_slice(&block_bytes).ok()?; + + Some( + crate::sync::import::find_blob_refs_ipld(&record_ipld, 0) + .into_iter() + .map(|blob_ref| blob_ref.cid) + .collect::>(), + ) + })) + .await + .into_iter() + .flatten() + .flatten() + .for_each(|cid| { + cids.insert(cid); + }); + } + + let mut granted = 0u64; + for cid in &cids { + if blob_repo + .ensure_blob_ownership(user_id, cid) + .await + .map_err(|_| (user_id, "failed to grant ownership"))? + { + granted += 1; + } + } + + Ok((user_id, did, granted)) +} + pub async fn backfill_record_blobs(repo_repo: Arc, block_store: AnyBlockStore) { let users_needing_backfill = match repo_repo.get_users_needing_record_blobs_backfill(100).await { @@ -352,6 +407,90 @@ pub async fn backfill_record_blobs(repo_repo: Arc, block_sto info!(success, failed, "Completed record_blobs backfill"); } +const BLOB_OWNERSHIP_BACKFILL_KEY: &str = "blob_ownership_backfilled"; + +pub async fn backfill_blob_ownership( + infra_repo: Arc, + repo_repo: Arc, + blob_repo: Arc, + block_store: AnyBlockStore, +) { + match infra_repo + .get_server_config(BLOB_OWNERSHIP_BACKFILL_KEY) + .await + { + Ok(Some(_)) => return, + Ok(None) => {} + Err(e) => { + error!("Failed to read blob ownership backfill marker: {:?}", e); + return; + } + } + + let repos = match repo_repo.get_all_repo_identities().await { + Ok(rows) => rows, + Err(e) => { + error!("Failed to query repos for blob ownership backfill: {:?}", e); + return; + } + }; + + if repos.is_empty() { + debug!("No repos need blob ownership backfill",); + return; + } + + info!( + count = repos.len(), + "Backfilling blob ownership for existing repos" + ); + + let mut success = 0; + let mut failed = 0; + + for chunk in repos.chunks(OWNERSHIP_CHUNK_SIZE) { + let results = futures::future::join_all(chunk.iter().map(|repo| { + let repo_repo = repo_repo.clone(); + let blob_repo = blob_repo.clone(); + let block_store = block_store.clone(); + + async move { + process_blob_ownership( + repo_repo.as_ref(), + blob_repo.as_ref(), + &block_store, + repo.user_id, + repo.did.clone(), + ) + .await + } + })) + .await; + + results.iter().for_each(|r| match r { + Ok((user_id, did, granted)) => { + if *granted > 0 { + info!(user_id = %user_id, did = %did, granted = granted, "Granted blob ownership"); + } + success += 1; + } + Err((user_id, reason)) => { + warn!(user_id = %user_id, reason = reason, "Failed to backfill blob ownership"); + failed += 1; + } + }); + } + + if let Err(e) = infra_repo + .upsert_server_config(BLOB_OWNERSHIP_BACKFILL_KEY, "1") + .await + { + error!("Failed to set blob ownership backfill marker: {:?}", e); + } + + info!(success, failed, "Completed blob ownership backfill"); +} + #[allow(clippy::too_many_arguments)] pub async fn start_scheduled_tasks( user_repo: Arc, diff --git a/crates/tranquil-pds/tests/store_parity.rs b/crates/tranquil-pds/tests/store_parity.rs index b696960..c93acb0 100644 --- a/crates/tranquil-pds/tests/store_parity.rs +++ b/crates/tranquil-pds/tests/store_parity.rs @@ -1063,6 +1063,85 @@ async fn parity_blob_shared_between_repos() { } } +#[tokio::test(flavor = "multi_thread")] +async fn parity_ensure_blob_ownership() { + let f = ParityFixture::new().await; + + let did_a = test_did("ensurea"); + let did_b = test_did("ensureb"); + let (pg_a, store_a) = seed_repos(&f, &did_a, &test_handle("ensurea")).await; + let (pg_b, store_b) = seed_repos(&f, &did_b, &test_handle("ensureb")).await; + + let cid = test_cid(211); + + f.pg.blob + .insert_blob(&cid, "image/png", 100, pg_a, "blobs/ensure.png") + .await + .unwrap(); + f.store + .blob + .insert_blob(&cid, "image/png", 100, store_a, "blobs/ensure.png") + .await + .unwrap(); + + assert!(f.pg.blob.ensure_blob_ownership(pg_b, &cid).await.unwrap()); + assert!( + f.store + .blob + .ensure_blob_ownership(store_b, &cid) + .await + .unwrap() + ); + + assert!(!f.pg.blob.ensure_blob_ownership(pg_b, &cid).await.unwrap()); + assert!( + !f.store + .blob + .ensure_blob_ownership(store_b, &cid) + .await + .unwrap() + ); + + let absent = test_cid(212); + assert!( + !f.pg + .blob + .ensure_blob_ownership(pg_b, &absent) + .await + .unwrap() + ); + assert!( + !f.store + .blob + .ensure_blob_ownership(store_b, &absent) + .await + .unwrap() + ); + + 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!( + 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() + ); + } +} + #[tokio::test] async fn parity_get_all_records() { let f = ParityFixture::new().await; diff --git a/crates/tranquil-server/src/main.rs b/crates/tranquil-server/src/main.rs index ab23437..12c028b 100644 --- a/crates/tranquil-server/src/main.rs +++ b/crates/tranquil-server/src/main.rs @@ -10,7 +10,8 @@ use tranquil_pds::comms::{CommsService, DiscordSender, EmailSender, SignalSender use tranquil_pds::crawlers::{Crawlers, start_crawlers_service}; use tranquil_pds::scheduled::{ - backfill_record_blobs, backfill_repo_rev, backfill_user_blocks, start_scheduled_tasks, + backfill_blob_ownership, backfill_record_blobs, backfill_repo_rev, backfill_user_blocks, + start_scheduled_tasks, }; use tranquil_pds::state::AppState; @@ -195,11 +196,21 @@ async fn run() -> Result<(), Box> { let backfill_repo_repo = state.repos.repo.clone(); let backfill_block_store = state.block_store.clone(); + let ownership_repo_repo = state.repos.repo.clone(); + let ownership_infra_repo = state.repos.infra.clone(); + let ownership_blob_repo = state.repos.blob.clone(); + let ownership_block_store = state.block_store.clone(); tokio::spawn(async move { tokio::join!( backfill_repo_rev(backfill_repo_repo.clone(), backfill_block_store.clone()), backfill_user_blocks(backfill_repo_repo.clone(), backfill_block_store.clone()), backfill_record_blobs(backfill_repo_repo, backfill_block_store), + backfill_blob_ownership( + ownership_infra_repo, + ownership_repo_repo, + ownership_blob_repo, + ownership_block_store + ) ); }); diff --git a/crates/tranquil-store/src/metastore/blob_ops.rs b/crates/tranquil-store/src/metastore/blob_ops.rs index c1a67b5..2cd9f34 100644 --- a/crates/tranquil-store/src/metastore/blob_ops.rs +++ b/crates/tranquil-store/src/metastore/blob_ops.rs @@ -111,6 +111,43 @@ impl BlobOps { Ok(Some(cid.clone())) } + pub fn ensure_blob_ownership( + &self, + user_id: Uuid, + cid: &CidLink, + ) -> Result { + let _guard = self.counter_lock.lock(); + let user_hash = self.resolve_user_hash(user_id)?; + let cid_str = cid.as_str(); + let marker_key = blob_meta_key(user_hash, cid_str); + + if self + .repo_data + .get(marker_key.as_slice()) + .map_err(MetastoreError::Fjall)? + .is_some() + { + return Ok(false); + } + + let cid_index_key = blob_by_cid_key(cid_str); + let Some(mut content) = self.get_blob_content(cid)? else { + return Ok(false); + }; + content.ref_count = content.ref_count.saturating_add(1); + + let mut batch = self.db.batch(); + batch.insert(&self.repo_data, marker_key.as_slice(), &[] as &[u8]); + batch.insert( + &self.repo_data, + cid_index_key.as_slice(), + content.serialize(), + ); + batch.commit().map_err(MetastoreError::Fjall)?; + + Ok(true) + } + fn get_blob_content(&self, cid: &CidLink) -> Result, MetastoreError> { point_lookup( &self.repo_data, @@ -762,4 +799,39 @@ mod tests { vec!["k".to_string()] ); } + + #[test] + fn ensure_blob_ownership_grants_to_second_user() { + 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(81); + ops.insert_blob(&cid, "a/b", 10, user_a, "k").unwrap(); + + assert!(ops.ensure_blob_ownership(user_b, &cid).unwrap()); + assert!(!ops.ensure_blob_ownership(user_b, &cid).unwrap()); + + assert_eq!(ops.count_blobs_by_user(user_b).unwrap(), 1); + assert_eq!(ops.sum_blob_storage().unwrap(), 10); + assert!( + ops.get_blob_storage_keys_by_user(user_a) + .unwrap() + .is_empty() + ); + } + + #[test] + fn ensure_blob_ownership_ignores_absent_blob() { + let (_dir, ms) = open_fresh(); + let (user_id, _) = setup_user(&ms); + let ops = ms.blob_ops(); + + assert!( + !ops.ensure_blob_ownership(user_id, &test_cid_link(82)) + .unwrap() + ); + assert_eq!(ops.count_blobs_by_user(user_id).unwrap(), 0); + } } diff --git a/crates/tranquil-store/src/metastore/client.rs b/crates/tranquil-store/src/metastore/client.rs index 2170b82..e7172ef 100644 --- a/crates/tranquil-store/src/metastore/client.rs +++ b/crates/tranquil-store/src/metastore/client.rs @@ -14,17 +14,18 @@ use tranquil_db_traits::{ InviteCodeSortOrder, InviteCodeUse, MigrationReactivationError, MigrationReactivationInput, NotificationHistoryRow, NotificationPrefs, OAuthTokenWithUser, PasswordResetResult, PlcTokenInfo, PruneCount, QueuedComms, ReactivatedAccountInfo, RecoverPasskeyAccountInput, - RecoverPasskeyAccountResult, RepoAccountInfo, RepoInfo, RepoListItem, RepoWithoutRev, - ReservedSigningKey, ReservedSigningKeyFull, ScheduledDeletionAccount, ScopePreference, - SequenceNumber, SequencedEvent, StoredBackupCode, StoredPasskey, TokenFamilyId, TotpRecord, - TotpRecordState, User2faStatus, UserAuthInfo, UserCommsPrefs, UserConfirmSignup, - UserDidWebInfo, UserEmailInfo, UserForDeletion, UserForDidDoc, UserForDidDocBuild, - UserForPasskeyRecovery, UserForPasskeySetup, UserForRecovery, UserForVerification, - UserIdAndHandle, UserIdAndPasswordHash, UserIdHandleEmail, UserInfoForAuth, UserKeyInfo, - UserKeyWithId, UserLegacyLoginPref, UserLoginCheck, UserLoginFull, UserLoginInfo, - UserNeedingRecordBlobsBackfill, UserPasswordInfo, UserResendVerification, UserResetCodeInfo, - UserRow, UserSessionInfo, UserStatus, UserVerificationInfo, UserWithKey, UserWithoutBlocks, - ValidatedInviteCode, WebauthnChallengeType, + RecoverPasskeyAccountResult, RepoAccountInfo, RepoIdentity, RepoInfo, RepoListItem, + RepoWithoutRev, ReservedSigningKey, ReservedSigningKeyFull, ScheduledDeletionAccount, + ScopePreference, SequenceNumber, SequencedEvent, StoredBackupCode, StoredPasskey, + TokenFamilyId, TotpRecord, TotpRecordState, User2faStatus, UserAuthInfo, UserCommsPrefs, + UserConfirmSignup, UserDidWebInfo, UserEmailInfo, UserForDeletion, UserForDidDoc, + UserForDidDocBuild, UserForPasskeyRecovery, UserForPasskeySetup, UserForRecovery, + UserForVerification, UserIdAndHandle, UserIdAndPasswordHash, UserIdHandleEmail, + UserInfoForAuth, UserKeyInfo, UserKeyWithId, UserLegacyLoginPref, UserLoginCheck, + UserLoginFull, UserLoginInfo, UserNeedingRecordBlobsBackfill, UserPasswordInfo, + UserResendVerification, UserResetCodeInfo, UserRow, UserSessionInfo, UserStatus, + UserVerificationInfo, UserWithKey, UserWithoutBlocks, ValidatedInviteCode, + WebauthnChallengeType, }; use tranquil_oauth::{AuthorizedClientData, DeviceData, RequestData, TokenData}; use tranquil_types::{ @@ -782,6 +783,14 @@ impl tranquil_db_traits::RepoRepository for MetastoreCli recv(rx).await } + async fn get_all_repo_identities(&self) -> Result, DbError> { + let (tx, rx) = oneshot::channel(); + self.pool.send(MetastoreRequest::Commit(Box::new( + CommitRequest::GetAllRepoIdentities { tx }, + )))?; + recv(rx).await + } + async fn insert_record_blobs( &self, repo_id: Uuid, @@ -875,6 +884,17 @@ impl tranquil_db_traits::BlobRepository for MetastoreCli recv(rx).await } + async fn ensure_blob_ownership(&self, user_id: Uuid, cid: &CidLink) -> Result { + let (tx, rx) = oneshot::channel(); + self.pool + .send(MetastoreRequest::Blob(BlobRequest::EnsureBlobOwnership { + user_id, + cid: cid.clone(), + tx, + }))?; + recv(rx).await + } + async fn get_blob_metadata( &self, cid: &CidLink, diff --git a/crates/tranquil-store/src/metastore/commit_ops.rs b/crates/tranquil-store/src/metastore/commit_ops.rs index 0e9003a..1c7bad7 100644 --- a/crates/tranquil-store/src/metastore/commit_ops.rs +++ b/crates/tranquil-store/src/metastore/commit_ops.rs @@ -26,7 +26,7 @@ use crate::io::{RealIO, StorageIO}; use tranquil_db_traits::{ ApplyCommitError, ApplyCommitInput, ApplyCommitResult, ImportBlock, ImportRecord, - ImportRepoError, UserNeedingRecordBlobsBackfill, UserWithoutBlocks, + ImportRepoError, RepoIdentity, UserNeedingRecordBlobsBackfill, UserWithoutBlocks, }; use tranquil_types::{AtUri, CidLink, Did, Tid}; @@ -401,6 +401,21 @@ impl CommitOps { ) } + pub fn get_all_repo_identities(&self) -> Result, MetastoreError> { + self.scan_users( + |_, _| Ok(true), + |meta, user_id| { + let did = match meta.did { + None => Err(MetastoreError::CorruptData("repo_meta missing DID field")), + Some(d) => Did::new(d) + .map_err(|_| MetastoreError::CorruptData("corrupt repo_meta did")), + }?; + Ok(RepoIdentity { user_id, did }) + }, + usize::MAX, + ) + } + fn scan_users_missing_prefix( &self, make_prefix: P, @@ -410,6 +425,33 @@ impl CommitOps { where F: Fn(RepoMetaValue, Uuid) -> Result, P: Fn(UserHash) -> SmallVec<[u8; 128]>, + { + self.scan_users( + |ops, user_hash| match ops + .repo_data + .prefix(make_prefix(user_hash).as_slice()) + .next() + { + Some(guard) => guard + .into_inner() + .map(|_| false) + .map_err(MetastoreError::Fjall), + None => Ok(true), + }, + build_result, + limit, + ) + } + + fn scan_users( + &self, + include: P, + build_result: F, + limit: usize, + ) -> Result, MetastoreError> + where + F: Fn(RepoMetaValue, Uuid) -> Result, + P: Fn(&Self, UserHash) -> Result, { let prefix = repo_meta_prefix(); @@ -429,18 +471,10 @@ impl CommitOps { } }; - let check_prefix = make_prefix(user_hash); - let has_entries = match self.repo_data.prefix(check_prefix.as_slice()).next() { - Some(guard) => match guard.into_inner() { - Ok(_) => true, - Err(e) => return Some(Err(MetastoreError::Fjall(e))), - }, - None => false, - }; - - match has_entries { - true => None, - false => { + match include(self, user_hash) { + Err(e) => return Some(Err(e)), + Ok(false) => None, + Ok(true) => { let meta = match RepoMetaValue::deserialize(&val_bytes) { Some(v) => v, None => { diff --git a/crates/tranquil-store/src/metastore/handler.rs b/crates/tranquil-store/src/metastore/handler.rs index f5c6aec..82d0945 100644 --- a/crates/tranquil-store/src/metastore/handler.rs +++ b/crates/tranquil-store/src/metastore/handler.rs @@ -16,17 +16,17 @@ use tranquil_db_traits::{ MigrationReactivationError, MigrationReactivationInput, NotificationHistoryRow, NotificationPrefs, OAuthTokenWithUser, PasswordResetResult, PlcTokenInfo, QueuedComms, ReactivatedAccountInfo, RecoverPasskeyAccountInput, RecoverPasskeyAccountResult, - RefreshSessionResult, ReservedSigningKey, ReservedSigningKeyFull, ScheduledDeletionAccount, - ScopePreference, SequenceNumber, SequencedEvent, SessionId, StoredBackupCode, StoredPasskey, - TokenFamilyId, TotpRecord, TotpRecordState, User2faStatus, UserAuthInfo, UserCommsPrefs, - UserConfirmSignup, UserDidWebInfo, UserEmailInfo, UserForDeletion, UserForDidDoc, - UserForDidDocBuild, UserForPasskeyRecovery, UserForPasskeySetup, UserForRecovery, - UserForVerification, UserIdAndHandle, UserIdAndPasswordHash, UserIdHandleEmail, - UserInfoForAuth, UserKeyInfo, UserKeyWithId, UserLegacyLoginPref, UserLoginCheck, - UserLoginFull, UserLoginInfo, UserNeedingRecordBlobsBackfill, UserPasswordInfo, - UserResendVerification, UserResetCodeInfo, UserRow, UserSessionInfo, UserStatus, - UserVerificationInfo, UserWithKey, UserWithoutBlocks, ValidatedInviteCode, - WebauthnChallengeType, + RefreshSessionResult, RepoIdentity, ReservedSigningKey, ReservedSigningKeyFull, + ScheduledDeletionAccount, ScopePreference, SequenceNumber, SequencedEvent, SessionId, + StoredBackupCode, StoredPasskey, TokenFamilyId, TotpRecord, TotpRecordState, User2faStatus, + UserAuthInfo, UserCommsPrefs, UserConfirmSignup, UserDidWebInfo, UserEmailInfo, + UserForDeletion, UserForDidDoc, UserForDidDocBuild, UserForPasskeyRecovery, + UserForPasskeySetup, UserForRecovery, UserForVerification, UserIdAndHandle, + UserIdAndPasswordHash, UserIdHandleEmail, UserInfoForAuth, UserKeyInfo, UserKeyWithId, + UserLegacyLoginPref, UserLoginCheck, UserLoginFull, UserLoginInfo, + UserNeedingRecordBlobsBackfill, UserPasswordInfo, UserResendVerification, UserResetCodeInfo, + UserRow, UserSessionInfo, UserStatus, UserVerificationInfo, UserWithKey, UserWithoutBlocks, + ValidatedInviteCode, WebauthnChallengeType, }; use tranquil_oauth::{AuthorizedClientData, DeviceData, RequestData, TokenData}; use tranquil_types::{ @@ -499,6 +499,9 @@ pub enum CommitRequest { limit: i64, tx: Tx>, }, + GetAllRepoIdentities { + tx: Tx>, + }, InsertRecordBlobs { repo_id: Uuid, record_uris: Vec, @@ -516,7 +519,8 @@ impl CommitRequest { repo_id: user_id, .. } => uuid_to_routing(user_hashes, user_id), Self::GetUsersWithoutBlocks { .. } - | Self::GetUsersNeedingRecordBlobsBackfill { .. } => Routing::Global, + | Self::GetUsersNeedingRecordBlobsBackfill { .. } + | Self::GetAllRepoIdentities { .. } => Routing::Global, } } } @@ -566,6 +570,11 @@ pub enum BlobRequest { storage_key: String, tx: Tx>, }, + EnsureBlobOwnership { + user_id: Uuid, + cid: CidLink, + tx: Tx, + }, GetBlobMetadata { cid: CidLink, tx: Tx>, @@ -628,9 +637,9 @@ pub enum BlobRequest { impl BlobRequest { fn routing(&self, user_hashes: &UserHashMap) -> Routing { match self { - Self::InsertBlob { cid, .. } | Self::UpdateBlobTakedown { cid, .. } => { - cid_to_routing(cid) - } + Self::InsertBlob { cid, .. } + | Self::EnsureBlobOwnership { cid, .. } + | Self::UpdateBlobTakedown { cid, .. } => cid_to_routing(cid), Self::DeleteBlobsByUser { user_id, .. } => uuid_to_routing(user_hashes, user_id), @@ -3074,6 +3083,14 @@ fn dispatch_commit(state: &HandlerState, req: CommitR .map_err(metastore_to_db), ); } + CommitRequest::GetAllRepoIdentities { tx } => { + let _ = tx.send( + state + .commit_ops + .get_all_repo_identities() + .map_err(metastore_to_db), + ); + } CommitRequest::InsertRecordBlobs { repo_id, record_uris, @@ -3179,6 +3196,14 @@ fn dispatch_blob(state: &HandlerState, req: BlobReque .map_err(metastore_to_db); let _ = tx.send(result); } + BlobRequest::EnsureBlobOwnership { user_id, cid, tx } => { + let result = state + .metastore + .blob_ops() + .ensure_blob_ownership(user_id, &cid) + .map_err(metastore_to_db); + let _ = tx.send(result); + } BlobRequest::GetBlobMetadata { cid, tx } => { let result = state .metastore