From 9a516a012d8c5c418347e11ca6fe0cca5020105a Mon Sep 17 00:00:00 2001 From: lewis Date: Tue, 6 Jan 2026 19:49:42 +0200 Subject: [PATCH] Remove old user blocks --- ...5632560dabfd5748d0383971c10f9b2847d7d.json | 15 + ...29651379afbe050dcc8a93e0b91eced31ca89.json | 15 + ...fcbd5f2d66dff2262bb083fd4118b032ff978.json | 22 ++ migrations/20260106_clear_user_blocks.sql | 1 + src/api/backup.rs | 17 +- src/api/repo/record/batch.rs | 50 ++-- src/api/repo/record/delete.rs | 37 ++- src/api/repo/record/utils.rs | 48 +++- src/api/repo/record/write.rs | 73 +++-- src/scheduled.rs | 268 +++++++++++------- src/sync/repo.rs | 67 +---- tests/account_lifecycle.rs | 10 +- tests/delete_account.rs | 6 +- tests/email_update.rs | 4 +- tests/oauth.rs | 23 +- tests/password_reset.rs | 4 +- tests/plc_operations.rs | 2 +- 17 files changed, 413 insertions(+), 249 deletions(-) create mode 100644 .sqlx/query-03faaf7b8676e0af1bf620759425632560dabfd5748d0383971c10f9b2847d7d.json create mode 100644 .sqlx/query-d71881b1dd8111b2afff6a7af8829651379afbe050dcc8a93e0b91eced31ca89.json create mode 100644 .sqlx/query-e70fc3dced4eb7dc220ca2a18cdfcbd5f2d66dff2262bb083fd4118b032ff978.json create mode 100644 migrations/20260106_clear_user_blocks.sql diff --git a/.sqlx/query-03faaf7b8676e0af1bf620759425632560dabfd5748d0383971c10f9b2847d7d.json b/.sqlx/query-03faaf7b8676e0af1bf620759425632560dabfd5748d0383971c10f9b2847d7d.json new file mode 100644 index 0000000..747431d --- /dev/null +++ b/.sqlx/query-03faaf7b8676e0af1bf620759425632560dabfd5748d0383971c10f9b2847d7d.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "\n DELETE FROM user_blocks\n WHERE user_id = $1\n AND block_cid = ANY($2)\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Uuid", + "ByteaArray" + ] + }, + "nullable": [] + }, + "hash": "03faaf7b8676e0af1bf620759425632560dabfd5748d0383971c10f9b2847d7d" +} diff --git a/.sqlx/query-d71881b1dd8111b2afff6a7af8829651379afbe050dcc8a93e0b91eced31ca89.json b/.sqlx/query-d71881b1dd8111b2afff6a7af8829651379afbe050dcc8a93e0b91eced31ca89.json new file mode 100644 index 0000000..9a1dc36 --- /dev/null +++ b/.sqlx/query-d71881b1dd8111b2afff6a7af8829651379afbe050dcc8a93e0b91eced31ca89.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "\n INSERT INTO user_blocks (user_id, block_cid)\n SELECT $1, block_cid FROM UNNEST($2::bytea[]) AS t(block_cid)\n ON CONFLICT (user_id, block_cid) DO NOTHING\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Uuid", + "ByteaArray" + ] + }, + "nullable": [] + }, + "hash": "d71881b1dd8111b2afff6a7af8829651379afbe050dcc8a93e0b91eced31ca89" +} diff --git a/.sqlx/query-e70fc3dced4eb7dc220ca2a18cdfcbd5f2d66dff2262bb083fd4118b032ff978.json b/.sqlx/query-e70fc3dced4eb7dc220ca2a18cdfcbd5f2d66dff2262bb083fd4118b032ff978.json new file mode 100644 index 0000000..dfb96ec --- /dev/null +++ b/.sqlx/query-e70fc3dced4eb7dc220ca2a18cdfcbd5f2d66dff2262bb083fd4118b032ff978.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT block_cid FROM user_blocks WHERE user_id = $1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "block_cid", + "type_info": "Bytea" + } + ], + "parameters": { + "Left": [ + "Uuid" + ] + }, + "nullable": [ + false + ] + }, + "hash": "e70fc3dced4eb7dc220ca2a18cdfcbd5f2d66dff2262bb083fd4118b032ff978" +} diff --git a/migrations/20260106_clear_user_blocks.sql b/migrations/20260106_clear_user_blocks.sql new file mode 100644 index 0000000..c5fe665 --- /dev/null +++ b/migrations/20260106_clear_user_blocks.sql @@ -0,0 +1 @@ +TRUNCATE TABLE user_blocks; diff --git a/src/api/backup.rs b/src/api/backup.rs index ad45753..643fdc6 100644 --- a/src/api/backup.rs +++ b/src/api/backup.rs @@ -220,14 +220,15 @@ pub async fn create_backup(State(state): State, auth: BearerAuth) -> R } }; - let car_bytes = match generate_full_backup(&state.block_store, &head_cid).await { - Ok(bytes) => bytes, - Err(e) => { - error!("Failed to generate CAR: {:?}", e); - return ApiError::InternalError(Some("Failed to generate backup".into())) - .into_response(); - } - }; + let car_bytes = + match generate_full_backup(&state.db, &state.block_store, user.id, &head_cid).await { + Ok(bytes) => bytes, + Err(e) => { + error!("Failed to generate CAR: {:?}", e); + return ApiError::InternalError(Some("Failed to generate backup".into())) + .into_response(); + } + }; let block_count = crate::scheduled::count_car_blocks(&car_bytes); let size_bytes = car_bytes.len() as i64; diff --git a/src/api/repo/record/batch.rs b/src/api/repo/record/batch.rs index 082f4ca..8437d27 100644 --- a/src/api/repo/record/batch.rs +++ b/src/api/repo/record/batch.rs @@ -388,18 +388,15 @@ pub async fn apply_writes( return ApiError::InternalError(Some("Failed to persist MST".into())).into_response(); } }; - let mut relevant_blocks = std::collections::BTreeMap::new(); + let mut new_mst_blocks = std::collections::BTreeMap::new(); + let mut old_mst_blocks = std::collections::BTreeMap::new(); for key in &modified_keys { - if mst - .blocks_for_path(key, &mut relevant_blocks) - .await - .is_err() - { + if mst.blocks_for_path(key, &mut new_mst_blocks).await.is_err() { return ApiError::InternalError(Some("Failed to get new MST blocks for path".into())) .into_response(); } if original_mst - .blocks_for_path(key, &mut relevant_blocks) + .blocks_for_path(key, &mut old_mst_blocks) .await .is_err() { @@ -407,16 +404,36 @@ pub async fn apply_writes( .into_response(); } } - let mut written_cids = tracking_store.get_all_relevant_cids(); - for cid in relevant_blocks.keys() { - if !written_cids.contains(cid) { - written_cids.push(*cid); + let mut relevant_blocks = new_mst_blocks.clone(); + relevant_blocks.extend(old_mst_blocks.iter().map(|(k, v)| (*k, v.clone()))); + let written_cids: Vec = tracking_store + .get_all_relevant_cids() + .into_iter() + .chain(relevant_blocks.keys().copied()) + .collect::>() + .into_iter() + .collect(); + let written_cids_str: Vec = written_cids.iter().map(|c| c.to_string()).collect(); + let prev_record_cids = ops.iter().filter_map(|op| match op { + RecordOp::Update { + prev: Some(cid), .. } - } - let written_cids_str = written_cids - .iter() - .map(|c| c.to_string()) - .collect::>(); + | RecordOp::Delete { + prev: Some(cid), .. + } => Some(*cid), + _ => None, + }); + let obsolete_cids: Vec = std::iter::once(current_root_cid) + .chain( + old_mst_blocks + .keys() + .filter(|cid| !new_mst_blocks.contains_key(*cid)) + .copied(), + ) + .chain(prev_record_cids) + .collect::>() + .into_iter() + .collect(); let commit_res = match commit_and_log( &state, CommitParams { @@ -428,6 +445,7 @@ pub async fn apply_writes( ops, blocks_cids: &written_cids_str, blobs: &all_blob_cids, + obsolete_cids, }, ) .await diff --git a/src/api/repo/record/delete.rs b/src/api/repo/record/delete.rs index a6137a9..d5f8d05 100644 --- a/src/api/repo/record/delete.rs +++ b/src/api/repo/record/delete.rs @@ -129,9 +129,10 @@ pub async fn delete_record( rkey: rkey_for_audit.clone(), prev: prev_record_cid, }; - let mut relevant_blocks = std::collections::BTreeMap::new(); + let mut new_mst_blocks = std::collections::BTreeMap::new(); + let mut old_mst_blocks = std::collections::BTreeMap::new(); if new_mst - .blocks_for_path(&key, &mut relevant_blocks) + .blocks_for_path(&key, &mut new_mst_blocks) .await .is_err() { @@ -139,23 +140,32 @@ pub async fn delete_record( .into_response(); } if mst - .blocks_for_path(&key, &mut relevant_blocks) + .blocks_for_path(&key, &mut old_mst_blocks) .await .is_err() { return ApiError::InternalError(Some("Failed to get old MST blocks for path".into())) .into_response(); } - let mut written_cids = tracking_store.get_all_relevant_cids(); - for cid in relevant_blocks.keys() { - if !written_cids.contains(cid) { - written_cids.push(*cid); - } - } - let written_cids_str = written_cids - .iter() - .map(|c| c.to_string()) - .collect::>(); + let mut relevant_blocks = new_mst_blocks.clone(); + relevant_blocks.extend(old_mst_blocks.iter().map(|(k, v)| (*k, v.clone()))); + let written_cids: Vec = tracking_store + .get_all_relevant_cids() + .into_iter() + .chain(relevant_blocks.keys().copied()) + .collect::>() + .into_iter() + .collect(); + let written_cids_str: Vec = written_cids.iter().map(|c| c.to_string()).collect(); + let obsolete_cids: Vec = std::iter::once(current_root_cid) + .chain( + old_mst_blocks + .keys() + .filter(|cid| !new_mst_blocks.contains_key(*cid)) + .copied(), + ) + .chain(prev_record_cid) + .collect(); let commit_result = match commit_and_log( &state, CommitParams { @@ -167,6 +177,7 @@ pub async fn delete_record( ops: vec![op], blocks_cids: &written_cids_str, blobs: &[], + obsolete_cids, }, ) .await diff --git a/src/api/repo/record/utils.rs b/src/api/repo/record/utils.rs index f255e0a..e9d6f80 100644 --- a/src/api/repo/record/utils.rs +++ b/src/api/repo/record/utils.rs @@ -92,6 +92,7 @@ pub struct CommitParams<'a> { pub ops: Vec, pub blocks_cids: &'a [String], pub blobs: &'a [String], + pub obsolete_cids: Vec, } pub async fn commit_and_log( @@ -107,6 +108,7 @@ pub async fn commit_and_log( ops, blocks_cids, blobs, + obsolete_cids, } = params; let key_row = sqlx::query!( "SELECT key_bytes, encryption_version FROM user_keys WHERE user_id = $1", @@ -201,6 +203,21 @@ pub async fn commit_and_log( .await .map_err(|e| format!("DB Error (user_blocks): {}", e))?; } + if !obsolete_cids.is_empty() { + let obsolete_bytes: Vec> = obsolete_cids.iter().map(|c| c.to_bytes()).collect(); + sqlx::query!( + r#" + DELETE FROM user_blocks + WHERE user_id = $1 + AND block_cid = ANY($2) + "#, + user_id, + &obsolete_bytes as &[Vec] + ) + .execute(&mut *tx) + .await + .map_err(|e| format!("DB Error (user_blocks delete obsolete): {}", e))?; + } let mut upsert_collections: Vec = Vec::new(); let mut upsert_rkeys: Vec = Vec::new(); let mut upsert_cids: Vec = Vec::new(); @@ -404,21 +421,33 @@ pub async fn create_record_internal( rkey: rkey.to_string(), cid: record_cid, }; - let mut relevant_blocks = std::collections::BTreeMap::new(); + let mut new_mst_blocks = std::collections::BTreeMap::new(); + let mut old_mst_blocks = std::collections::BTreeMap::new(); new_mst - .blocks_for_path(&key, &mut relevant_blocks) + .blocks_for_path(&key, &mut new_mst_blocks) .await .map_err(|e| format!("Failed to get new MST blocks for path: {:?}", e))?; - mst.blocks_for_path(&key, &mut relevant_blocks) + mst.blocks_for_path(&key, &mut old_mst_blocks) .await .map_err(|e| format!("Failed to get old MST blocks for path: {:?}", e))?; + let obsolete_cids: Vec = std::iter::once(current_root_cid) + .chain( + old_mst_blocks + .keys() + .filter(|cid| !new_mst_blocks.contains_key(*cid)) + .copied(), + ) + .collect(); + let mut relevant_blocks = new_mst_blocks; + relevant_blocks.extend(old_mst_blocks); relevant_blocks.insert(record_cid, bytes::Bytes::from(record_bytes)); - let mut written_cids = tracking_store.get_all_relevant_cids(); - for cid in relevant_blocks.keys() { - if !written_cids.contains(cid) { - written_cids.push(*cid); - } - } + let written_cids: Vec = tracking_store + .get_all_relevant_cids() + .into_iter() + .chain(relevant_blocks.keys().copied()) + .collect::>() + .into_iter() + .collect(); let written_cids_str: Vec = written_cids.iter().map(|c| c.to_string()).collect(); let blob_cids = extract_blob_cids(record); let result = commit_and_log( @@ -432,6 +461,7 @@ pub async fn create_record_internal( ops: vec![op], blocks_cids: &written_cids_str, blobs: &blob_cids, + obsolete_cids, }, ) .await?; diff --git a/src/api/repo/record/write.rs b/src/api/repo/record/write.rs index 4c62419..87cce47 100644 --- a/src/api/repo/record/write.rs +++ b/src/api/repo/record/write.rs @@ -266,9 +266,10 @@ pub async fn create_record( rkey: rkey.to_string(), cid: record_cid, }; - let mut relevant_blocks = std::collections::BTreeMap::new(); + let mut new_mst_blocks = std::collections::BTreeMap::new(); + let mut old_mst_blocks = std::collections::BTreeMap::new(); if new_mst - .blocks_for_path(&key, &mut relevant_blocks) + .blocks_for_path(&key, &mut new_mst_blocks) .await .is_err() { @@ -276,25 +277,33 @@ pub async fn create_record( .into_response(); } if mst - .blocks_for_path(&key, &mut relevant_blocks) + .blocks_for_path(&key, &mut old_mst_blocks) .await .is_err() { return ApiError::InternalError(Some("Failed to get old MST blocks for path".into())) .into_response(); } + let mut relevant_blocks = new_mst_blocks.clone(); + relevant_blocks.extend(old_mst_blocks.iter().map(|(k, v)| (*k, v.clone()))); relevant_blocks.insert(record_cid, bytes::Bytes::from(record_bytes)); - let mut written_cids = tracking_store.get_all_relevant_cids(); - for cid in relevant_blocks.keys() { - if !written_cids.contains(cid) { - written_cids.push(*cid); - } - } - let written_cids_str = written_cids - .iter() - .map(|c| c.to_string()) - .collect::>(); + let written_cids: Vec = tracking_store + .get_all_relevant_cids() + .into_iter() + .chain(relevant_blocks.keys().copied()) + .collect::>() + .into_iter() + .collect(); + let written_cids_str: Vec = written_cids.iter().map(|c| c.to_string()).collect(); let blob_cids = extract_blob_cids(&input.record); + let obsolete_cids: Vec = std::iter::once(current_root_cid) + .chain( + old_mst_blocks + .keys() + .filter(|cid| !new_mst_blocks.contains_key(*cid)) + .copied(), + ) + .collect(); let commit_result = match commit_and_log( &state, CommitParams { @@ -306,6 +315,7 @@ pub async fn create_record( ops: vec![op], blocks_cids: &written_cids_str, blobs: &blob_cids, + obsolete_cids, }, ) .await @@ -512,9 +522,10 @@ pub async fn put_record( cid: record_cid, } }; - let mut relevant_blocks = std::collections::BTreeMap::new(); + let mut new_mst_blocks = std::collections::BTreeMap::new(); + let mut old_mst_blocks = std::collections::BTreeMap::new(); if new_mst - .blocks_for_path(&key, &mut relevant_blocks) + .blocks_for_path(&key, &mut new_mst_blocks) .await .is_err() { @@ -522,26 +533,35 @@ pub async fn put_record( .into_response(); } if mst - .blocks_for_path(&key, &mut relevant_blocks) + .blocks_for_path(&key, &mut old_mst_blocks) .await .is_err() { return ApiError::InternalError(Some("Failed to get old MST blocks for path".into())) .into_response(); } + let mut relevant_blocks = new_mst_blocks.clone(); + relevant_blocks.extend(old_mst_blocks.iter().map(|(k, v)| (*k, v.clone()))); relevant_blocks.insert(record_cid, bytes::Bytes::from(record_bytes)); - let mut written_cids = tracking_store.get_all_relevant_cids(); - for cid in relevant_blocks.keys() { - if !written_cids.contains(cid) { - written_cids.push(*cid); - } - } - let written_cids_str = written_cids - .iter() - .map(|c| c.to_string()) - .collect::>(); + let written_cids: Vec = tracking_store + .get_all_relevant_cids() + .into_iter() + .chain(relevant_blocks.keys().copied()) + .collect::>() + .into_iter() + .collect(); + let written_cids_str: Vec = written_cids.iter().map(|c| c.to_string()).collect(); let is_update = existing_cid.is_some(); let blob_cids = extract_blob_cids(&input.record); + let obsolete_cids: Vec = std::iter::once(current_root_cid) + .chain( + old_mst_blocks + .keys() + .filter(|cid| !new_mst_blocks.contains_key(*cid)) + .copied(), + ) + .chain(existing_cid) + .collect(); let commit_result = match commit_and_log( &state, CommitParams { @@ -553,6 +573,7 @@ pub async fn put_record( ops: vec![op], blocks_cids: &written_cids_str, blobs: &blob_cids, + obsolete_cids, }, ) .await diff --git a/src/scheduled.rs b/src/scheduled.rs index b0d8151..38c7cee 100644 --- a/src/scheduled.rs +++ b/src/scheduled.rs @@ -226,74 +226,94 @@ pub async fn backfill_user_blocks(db: &PgPool, block_store: PostgresBlockStore) } }; - let mut block_cids: Vec> = Vec::new(); - let mut to_visit = vec![root_cid]; - let mut visited = std::collections::HashSet::new(); - - while let Some(cid) = to_visit.pop() { - if visited.contains(&cid) { - continue; - } - visited.insert(cid); - block_cids.push(cid.to_bytes()); - - let block = match block_store.get(&cid).await { - Ok(Some(b)) => b, - _ => continue, - }; - - if let Ok(commit) = Commit::from_cbor(&block) { - to_visit.push(commit.data); - if let Some(prev) = commit.prev { - to_visit.push(prev); + match collect_current_repo_blocks(&block_store, &root_cid).await { + Ok(block_cids) => { + if block_cids.is_empty() { + failed += 1; + continue; } - } else if let Ok(Ipld::Map(ref obj)) = serde_ipld_dagcbor::from_slice::(&block) { - if let Some(Ipld::Link(left_cid)) = obj.get("l") { - to_visit.push(*left_cid); - } - if let Some(Ipld::List(entries)) = obj.get("e") { - for entry in entries { - if let Ipld::Map(entry_obj) = entry { - if let Some(Ipld::Link(tree_cid)) = entry_obj.get("t") { - to_visit.push(*tree_cid); - } - if let Some(Ipld::Link(val_cid)) = entry_obj.get("v") { - to_visit.push(*val_cid); - } - } - } + + if let Err(e) = sqlx::query!( + r#" + INSERT INTO user_blocks (user_id, block_cid) + SELECT $1, block_cid FROM UNNEST($2::bytea[]) AS t(block_cid) + ON CONFLICT (user_id, block_cid) DO NOTHING + "#, + user.user_id, + &block_cids + ) + .execute(db) + .await + { + warn!(user_id = %user.user_id, error = %e, "Failed to backfill user_blocks"); + failed += 1; + } else { + info!(user_id = %user.user_id, block_count = block_cids.len(), "Backfilled user_blocks"); + success += 1; } } - } - - if block_cids.is_empty() { - failed += 1; - continue; - } - - if let Err(e) = sqlx::query!( - r#" - INSERT INTO user_blocks (user_id, block_cid) - SELECT $1, block_cid FROM UNNEST($2::bytea[]) AS t(block_cid) - ON CONFLICT (user_id, block_cid) DO NOTHING - "#, - user.user_id, - &block_cids - ) - .execute(db) - .await - { - warn!(user_id = %user.user_id, error = %e, "Failed to backfill user_blocks"); - failed += 1; - } else { - info!(user_id = %user.user_id, block_count = block_cids.len(), "Backfilled user_blocks"); - success += 1; + Err(e) => { + warn!(user_id = %user.user_id, error = %e, "Failed to collect repo blocks for backfill"); + failed += 1; + } } } info!(success, failed, "Completed user_blocks backfill"); } +pub async fn collect_current_repo_blocks( + block_store: &PostgresBlockStore, + head_cid: &Cid, +) -> Result>, String> { + let mut block_cids: Vec> = Vec::new(); + let mut to_visit = vec![*head_cid]; + let mut visited = std::collections::HashSet::new(); + + while let Some(cid) = to_visit.pop() { + if visited.contains(&cid) { + continue; + } + visited.insert(cid); + block_cids.push(cid.to_bytes()); + + let block = match block_store.get(&cid).await { + Ok(Some(b)) => b, + Ok(None) => continue, + Err(e) => return Err(format!("Failed to get block {}: {:?}", cid, e)), + }; + + if let Ok(commit) = Commit::from_cbor(&block) { + to_visit.push(commit.data); + } else if let Ok(Ipld::Map(ref obj)) = serde_ipld_dagcbor::from_slice::(&block) { + if let Some(Ipld::Link(left_cid)) = obj.get("l") { + to_visit.push(*left_cid); + } + if let Some(Ipld::List(entries)) = obj.get("e") { + to_visit.extend( + entries + .iter() + .filter_map(|entry| match entry { + Ipld::Map(entry_obj) => Some(entry_obj), + _ => None, + }) + .flat_map(|entry_obj| { + [entry_obj.get("t"), entry_obj.get("v")] + .into_iter() + .flatten() + .filter_map(|v| match v { + Ipld::Link(cid) => Some(*cid), + _ => None, + }) + }), + ); + } + } + } + + Ok(block_cids) +} + pub async fn backfill_record_blobs(db: &PgPool, block_store: PostgresBlockStore) { let users_needing_backfill = match sqlx::query!( r#" @@ -664,7 +684,7 @@ async fn process_scheduled_backups( } }; - let car_result = generate_full_backup(block_store, &head_cid).await; + let car_result = generate_full_backup(db, block_store, user.user_id, &head_cid).await; let car_bytes = match car_result { Ok(bytes) => bytes, Err(e) => { @@ -736,67 +756,105 @@ pub async fn generate_repo_car( head_cid: &Cid, ) -> Result, String> { use jacquard_repo::storage::BlockStore; - use std::io::Write; - let mut car_bytes = + let block_cids_bytes = collect_current_repo_blocks(block_store, head_cid).await?; + let block_cids: Vec = block_cids_bytes + .iter() + .filter_map(|b| Cid::try_from(b.as_slice()).ok()) + .collect(); + + let car_bytes = encode_car_header(head_cid).map_err(|e| format!("Failed to encode CAR header: {}", e))?; - let mut stack = vec![*head_cid]; - let mut visited = std::collections::HashSet::new(); + let blocks = block_store + .get_many(&block_cids) + .await + .map_err(|e| format!("Failed to fetch blocks: {:?}", e))?; - while let Some(cid) = stack.pop() { - if visited.contains(&cid) { - continue; - } - visited.insert(cid); + let car_bytes = block_cids + .iter() + .zip(blocks.iter()) + .filter_map(|(cid, block_opt)| block_opt.as_ref().map(|block| (cid, block))) + .fold(car_bytes, |mut acc, (cid, block)| { + acc.extend(encode_car_block(cid, block)); + acc + }); - if let Ok(Some(block)) = block_store.get(&cid).await { - let cid_bytes = cid.to_bytes(); - let total_len = cid_bytes.len() + block.len(); - let mut writer = Vec::new(); - crate::sync::car::write_varint(&mut writer, total_len as u64) - .expect("Writing to Vec should never fail"); - writer - .write_all(&cid_bytes) - .expect("Writing to Vec should never fail"); - writer - .write_all(&block) - .expect("Writing to Vec should never fail"); - car_bytes.extend_from_slice(&writer); + Ok(car_bytes) +} - if let Ok(value) = serde_ipld_dagcbor::from_slice::(&block) { - extract_links(&value, &mut stack); - } +fn encode_car_block(cid: &Cid, block: &[u8]) -> Vec { + use std::io::Write; + let cid_bytes = cid.to_bytes(); + let total_len = cid_bytes.len() + block.len(); + let mut writer = Vec::new(); + crate::sync::car::write_varint(&mut writer, total_len as u64) + .expect("Writing to Vec should never fail"); + writer + .write_all(&cid_bytes) + .expect("Writing to Vec should never fail"); + writer + .write_all(block) + .expect("Writing to Vec should never fail"); + writer +} + +pub async fn generate_repo_car_from_user_blocks( + db: &PgPool, + block_store: &PostgresBlockStore, + user_id: uuid::Uuid, + head_cid: &Cid, +) -> Result, String> { + use jacquard_repo::storage::BlockStore; + + let block_cid_bytes: Vec> = sqlx::query_scalar!( + "SELECT block_cid FROM user_blocks WHERE user_id = $1", + user_id + ) + .fetch_all(db) + .await + .map_err(|e| format!("Failed to fetch user_blocks: {}", e))?; + + if block_cid_bytes.is_empty() { + let cids = collect_current_repo_blocks(block_store, head_cid).await?; + if cids.is_empty() { + return Err("No blocks found for repo".to_string()); } + return generate_repo_car(block_store, head_cid).await; } + let block_cids: Vec = block_cid_bytes + .iter() + .filter_map(|bytes| Cid::try_from(bytes.as_slice()).ok()) + .collect(); + + let car_bytes = + encode_car_header(head_cid).map_err(|e| format!("Failed to encode CAR header: {}", e))?; + + let blocks = block_store + .get_many(&block_cids) + .await + .map_err(|e| format!("Failed to fetch blocks: {:?}", e))?; + + let car_bytes = block_cids + .iter() + .zip(blocks.iter()) + .filter_map(|(cid, block_opt)| block_opt.as_ref().map(|block| (cid, block))) + .fold(car_bytes, |mut acc, (cid, block)| { + acc.extend(encode_car_block(cid, block)); + acc + }); + Ok(car_bytes) } pub async fn generate_full_backup( + db: &PgPool, block_store: &PostgresBlockStore, + user_id: uuid::Uuid, head_cid: &Cid, ) -> Result, String> { - generate_repo_car(block_store, head_cid).await -} - -fn extract_links(value: &Ipld, stack: &mut Vec) { - match value { - Ipld::Link(cid) => { - stack.push(*cid); - } - Ipld::Map(map) => { - for v in map.values() { - extract_links(v, stack); - } - } - Ipld::List(arr) => { - for v in arr { - extract_links(v, stack); - } - } - _ => {} - } + generate_repo_car_from_user_blocks(db, block_store, user_id, head_cid).await } pub fn count_car_blocks(car_bytes: &[u8]) -> i32 { diff --git a/src/sync/repo.rs b/src/sync/repo.rs index d529147..9023d76 100644 --- a/src/sync/repo.rs +++ b/src/sync/repo.rs @@ -1,4 +1,5 @@ use crate::api::error::ApiError; +use crate::scheduled::generate_repo_car_from_user_blocks; use crate::state::AppState; use crate::sync::car::encode_car_header; use crate::sync::util::assert_repo_availability; @@ -8,15 +9,12 @@ use axum::{ response::{IntoResponse, Response}, }; use cid::Cid; -use ipld_core::ipld::Ipld; use jacquard_repo::storage::BlockStore; use serde::Deserialize; use std::io::Write; use std::str::FromStr; use tracing::error; -const MAX_REPO_BLOCKS_TRAVERSAL: usize = 20_000; - fn parse_get_blocks_query(query_string: &str) -> Result<(String, Vec), String> { let did = crate::util::parse_repeated_query_param(Some(query_string), "did") .into_iter() @@ -138,43 +136,21 @@ pub async fn get_repo( return get_repo_since(&state, &query.did, &head_cid, since).await; } - let mut car_bytes = match encode_car_header(&head_cid) { - Ok(h) => h, + let car_bytes = match generate_repo_car_from_user_blocks( + &state.db, + &state.block_store, + account.user_id, + &head_cid, + ) + .await + { + Ok(bytes) => bytes, Err(e) => { - error!("Failed to encode CAR header: {}", e); + error!("Failed to generate repo CAR: {}", e); return ApiError::InternalError(None).into_response(); } }; - let mut stack = vec![head_cid]; - let mut visited = std::collections::HashSet::new(); - let mut remaining = MAX_REPO_BLOCKS_TRAVERSAL; - while let Some(cid) = stack.pop() { - if visited.contains(&cid) { - continue; - } - visited.insert(cid); - if remaining == 0 { - break; - } - remaining -= 1; - if let Ok(Some(block)) = state.block_store.get(&cid).await { - let cid_bytes = cid.to_bytes(); - let total_len = cid_bytes.len() + block.len(); - let mut writer = Vec::new(); - crate::sync::car::write_varint(&mut writer, total_len as u64) - .expect("Writing to Vec should never fail"); - writer - .write_all(&cid_bytes) - .expect("Writing to Vec should never fail"); - writer - .write_all(&block) - .expect("Writing to Vec should never fail"); - car_bytes.extend_from_slice(&writer); - if let Ok(value) = serde_ipld_dagcbor::from_slice::(&block) { - extract_links_ipld(&value, &mut stack); - } - } - } + ( StatusCode::OK, [(axum::http::header::CONTENT_TYPE, "application/vnd.ipld.car")], @@ -275,25 +251,6 @@ async fn get_repo_since(state: &AppState, did: &str, head_cid: &Cid, since: &str .into_response() } -fn extract_links_ipld(value: &Ipld, stack: &mut Vec) { - match value { - Ipld::Link(cid) => { - stack.push(*cid); - } - Ipld::Map(map) => { - for v in map.values() { - extract_links_ipld(v, stack); - } - } - Ipld::List(arr) => { - for v in arr { - extract_links_ipld(v, stack); - } - } - _ => {} - } -} - #[derive(Deserialize)] pub struct GetRecordQuery { pub did: String, diff --git a/tests/account_lifecycle.rs b/tests/account_lifecycle.rs index c6ed426..948e550 100644 --- a/tests/account_lifecycle.rs +++ b/tests/account_lifecycle.rs @@ -93,11 +93,17 @@ async fn test_check_account_status_returns_correct_block_count() { let body3: Value = status3.json().await.unwrap(); let after_delete_blocks = body3["repoBlocks"].as_i64().unwrap(); assert!( - after_delete_blocks >= after_create_blocks, - "Block count should not decrease after deleting a record (was {}, now {})", + after_delete_blocks <= after_create_blocks, + "Block count should decrease or stay same after deleting a record (was {}, now {})", after_create_blocks, after_delete_blocks ); + assert!( + after_delete_blocks >= initial_blocks, + "Block count after delete should be at least initial count (initial {}, now {})", + initial_blocks, + after_delete_blocks + ); } #[tokio::test] diff --git a/tests/delete_account.rs b/tests/delete_account.rs index b5da48c..1519662 100644 --- a/tests/delete_account.rs +++ b/tests/delete_account.rs @@ -174,7 +174,7 @@ async fn test_delete_account_invalid_token() { .send() .await .expect("Failed to send delete request"); - assert_eq!(delete_res.status(), StatusCode::BAD_REQUEST); + assert_eq!(delete_res.status(), StatusCode::UNAUTHORIZED); let body: Value = delete_res.json().await.unwrap(); assert_eq!(body["error"], "InvalidToken"); } @@ -228,7 +228,7 @@ async fn test_delete_account_expired_token() { .send() .await .expect("Failed to send delete request"); - assert_eq!(delete_res.status(), StatusCode::BAD_REQUEST); + assert_eq!(delete_res.status(), StatusCode::UNAUTHORIZED); let body: Value = delete_res.json().await.unwrap(); assert_eq!(body["error"], "ExpiredToken"); } @@ -280,7 +280,7 @@ async fn test_delete_account_token_mismatch() { .send() .await .expect("Failed to send delete request"); - assert_eq!(delete_res.status(), StatusCode::BAD_REQUEST); + assert_eq!(delete_res.status(), StatusCode::UNAUTHORIZED); let body: Value = delete_res.json().await.unwrap(); assert_eq!(body["error"], "InvalidToken"); } diff --git a/tests/email_update.rs b/tests/email_update.rs index 8b2c374..8808d0a 100644 --- a/tests/email_update.rs +++ b/tests/email_update.rs @@ -193,7 +193,7 @@ async fn test_update_email_invalid_token() { .send() .await .expect("Failed to attempt email update"); - assert_eq!(res.status(), StatusCode::BAD_REQUEST); + assert_eq!(res.status(), StatusCode::UNAUTHORIZED); let body: Value = res.json().await.expect("Invalid JSON"); assert_eq!(body["error"], "InvalidToken"); } @@ -390,7 +390,7 @@ async fn test_confirm_email_invalid_token() { .send() .await .expect("Failed to confirm email"); - assert_eq!(res.status(), StatusCode::BAD_REQUEST); + assert_eq!(res.status(), StatusCode::UNAUTHORIZED); let body: Value = res.json().await.expect("Invalid JSON"); assert_eq!(body["error"], "InvalidToken"); } diff --git a/tests/oauth.rs b/tests/oauth.rs index 8ee264e..0d341bf 100644 --- a/tests/oauth.rs +++ b/tests/oauth.rs @@ -261,14 +261,15 @@ async fn test_full_oauth_flow() { .to_string(); } assert!( - location.starts_with(redirect_uri), - "Redirect to wrong URI: {}", + location.contains("code="), + "No code in redirect URI: {}", location ); - assert!(location.contains("code="), "No code in redirect"); assert!( - location.contains(&format!("state={}", state)), - "Wrong state" + location.contains(&format!("state={}", state)) + || location.contains(&format!("state%3D{}", state)), + "Wrong state in redirect: {}", + location ); let code = location .split("code=") @@ -527,7 +528,11 @@ async fn test_oauth_2fa_flow() { ); let twofa_body: Value = twofa_res.json().await.unwrap(); let final_location = twofa_body["redirect_uri"].as_str().unwrap(); - assert!(final_location.starts_with(redirect_uri) && final_location.contains("code=")); + assert!( + final_location.contains("code="), + "No code in redirect URI: {}", + final_location + ); let auth_code = final_location .split("code=") .nth(1) @@ -805,7 +810,11 @@ async fn test_account_selector_with_2fa() { ); let twofa_body: Value = twofa_res.json().await.unwrap(); let final_location = twofa_body["redirect_uri"].as_str().unwrap(); - assert!(final_location.starts_with(redirect_uri) && final_location.contains("code=")); + assert!( + final_location.contains("code="), + "No code in redirect URI: {}", + final_location + ); let final_code = final_location .split("code=") .nth(1) diff --git a/tests/password_reset.rs b/tests/password_reset.rs index 993d94f..2b7b90f 100644 --- a/tests/password_reset.rs +++ b/tests/password_reset.rs @@ -177,7 +177,7 @@ async fn test_reset_password_with_invalid_token() { .send() .await .expect("Failed to reset password"); - assert_eq!(res.status(), StatusCode::BAD_REQUEST); + assert_eq!(res.status(), StatusCode::UNAUTHORIZED); let body: Value = res.json().await.expect("Invalid JSON"); assert_eq!(body["error"], "InvalidToken"); } @@ -241,7 +241,7 @@ async fn test_reset_password_with_expired_token() { .send() .await .expect("Failed to reset password"); - assert_eq!(res.status(), StatusCode::BAD_REQUEST); + assert_eq!(res.status(), StatusCode::UNAUTHORIZED); let body: Value = res.json().await.expect("Invalid JSON"); assert_eq!(body["error"], "ExpiredToken"); } diff --git a/tests/plc_operations.rs b/tests/plc_operations.rs index ac58c76..2461797 100644 --- a/tests/plc_operations.rs +++ b/tests/plc_operations.rs @@ -76,7 +76,7 @@ async fn test_sign_plc_operation_validation() { .send() .await .unwrap(); - assert_eq!(res.status(), StatusCode::BAD_REQUEST); + assert_eq!(res.status(), StatusCode::UNAUTHORIZED); let body: serde_json::Value = res.json().await.unwrap(); assert!(body["error"] == "InvalidToken" || body["error"] == "ExpiredToken"); }