Introduce job to repair missing blobs

My last PR introduced a migration to blobs, changing them from being unique per cid to being unique per cid+user. This is because two users can upload the same blob, get the same cid, and then previously in tranquil the second user would never actually get their ownership recorded, meaning listBlobs was missing the blob, and a migration _off_ of tranquil would leave them sans blob https://tangled.org/tranquil.farm/tranquil-pds/pulls/261/round/3

This PR attempts to add a job to repair existing instances where blobs have been deduplicated and not attributed to each owning user. It does this by pulling all records for each user, and then in batches crawling the records to find blob references. For each blob reference, it checks whether that user has its ownership recorded, and if not, repairs it by adding the entry.

To avoid re-running this potentially heavy job on every startup, I've used the config table to store a little record of whether it has been run.

.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"
+}

.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"
+}

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<Vec<String>, DbError>;
 
+    async fn ensure_blob_ownership(&self, user_id: Uuid, cid: &CidLink) -> Result<bool, DbError>;
+
     async fn insert_record_blobs(
         &self,
         repo_id: Uuid,

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};

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<Vec<UserNeedingRecordBlobsBackfill>, DbError>;
 
+    async fn get_all_repo_identities(&self) -> Result<Vec<RepoIdentity>, DbError>;
+
     async fn insert_record_blobs(
         &self,
         repo_id: Uuid,

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<bool, DbError> {
+        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,

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<Vec<RepoIdentity>, 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,

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<CidLink> = 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::<Vec<_>>(),
+            )
+        }))
+        .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<dyn RepoRepository>, 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<dyn RepoRepository>, 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<dyn InfraRepository>,
+    repo_repo: Arc<dyn RepoRepository>,
+    blob_repo: Arc<dyn BlobRepository>,
+    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<dyn UserRepository>,

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;

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<dyn std::error::Error>> {
 
     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
+            )
         );
     });
 

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<bool, MetastoreError> {
+        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<Option<BlobContentValue>, 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);
+    }
 }

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<S: StorageIO + 'static> tranquil_db_traits::RepoRepository for MetastoreCli
         recv(rx).await
     }
 
+    async fn get_all_repo_identities(&self) -> Result<Vec<RepoIdentity>, 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<S: StorageIO + 'static> tranquil_db_traits::BlobRepository for MetastoreCli
         recv(rx).await
     }
 
+    async fn ensure_blob_ownership(&self, user_id: Uuid, cid: &CidLink) -> Result<bool, DbError> {
+        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,

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<S: StorageIO + 'static> CommitOps<S> {
         )
     }
 
+    pub fn get_all_repo_identities(&self) -> Result<Vec<RepoIdentity>, 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<T, F, P>(
         &self,
         make_prefix: P,
@@ -410,6 +425,33 @@ impl<S: StorageIO + 'static> CommitOps<S> {
     where
         F: Fn(RepoMetaValue, Uuid) -> Result<T, MetastoreError>,
         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<T, F, P>(
+        &self,
+        include: P,
+        build_result: F,
+        limit: usize,
+    ) -> Result<Vec<T>, MetastoreError>
+    where
+        F: Fn(RepoMetaValue, Uuid) -> Result<T, MetastoreError>,
+        P: Fn(&Self, UserHash) -> Result<bool, MetastoreError>,
     {
         let prefix = repo_meta_prefix();
 
@@ -429,18 +471,10 @@ impl<S: StorageIO + 'static> CommitOps<S> {
                     }
                 };
 
-                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 => {

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<Vec<UserNeedingRecordBlobsBackfill>>,
     },
+    GetAllRepoIdentities {
+        tx: Tx<Vec<RepoIdentity>>,
+    },
     InsertRecordBlobs {
         repo_id: Uuid,
         record_uris: Vec<AtUri>,
@@ -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<Option<CidLink>>,
     },
+    EnsureBlobOwnership {
+        user_id: Uuid,
+        cid: CidLink,
+        tx: Tx<bool>,
+    },
     GetBlobMetadata {
         cid: CidLink,
         tx: Tx<Option<tranquil_db_traits::BlobMetadata>>,
@@ -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<S: StorageIO + 'static>(state: &HandlerState<S>, 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<S: StorageIO + 'static>(state: &HandlerState<S>, 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