diff --git a/.sqlx/query-3b791fdb8e29043c980963d4d18e1e492c73c39818a8648a7af70555418fb5d1.json b/.sqlx/query-3b791fdb8e29043c980963d4d18e1e492c73c39818a8648a7af70555418fb5d1.json new file mode 100644 index 0000000..973efac --- /dev/null +++ b/.sqlx/query-3b791fdb8e29043c980963d4d18e1e492c73c39818a8648a7af70555418fb5d1.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE repo_seq SET blocks_cids = $1 WHERE seq = $2", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "TextArray", + "Int8" + ] + }, + "nullable": [] + }, + "hash": "3b791fdb8e29043c980963d4d18e1e492c73c39818a8648a7af70555418fb5d1" +} diff --git a/.sqlx/query-3fae97c8a2551c1ef8db06c4cde5480e44c5f771397e01574d0026e5bac6af55.json b/.sqlx/query-3fae97c8a2551c1ef8db06c4cde5480e44c5f771397e01574d0026e5bac6af55.json deleted file mode 100644 index f7acefd..0000000 --- a/.sqlx/query-3fae97c8a2551c1ef8db06c4cde5480e44c5f771397e01574d0026e5bac6af55.json +++ /dev/null @@ -1,22 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n SELECT r.repo_root_cid\n FROM repos r\n JOIN users u ON u.id = r.user_id\n WHERE u.did = $1\n ", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "repo_root_cid", - "type_info": "Text" - } - ], - "parameters": { - "Left": [ - "Text" - ] - }, - "nullable": [ - false - ] - }, - "hash": "3fae97c8a2551c1ef8db06c4cde5480e44c5f771397e01574d0026e5bac6af55" -} diff --git a/.sqlx/query-5b692e8f6d32dcbdcb45a3fff152a2be5672aadd807a4abab6914f80d57cba02.json b/.sqlx/query-5b692e8f6d32dcbdcb45a3fff152a2be5672aadd807a4abab6914f80d57cba02.json deleted file mode 100644 index 1bd55fb..0000000 --- a/.sqlx/query-5b692e8f6d32dcbdcb45a3fff152a2be5672aadd807a4abab6914f80d57cba02.json +++ /dev/null @@ -1,22 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n SELECT r.repo_root_cid\n FROM repos r\n JOIN users u ON r.user_id = u.id\n WHERE u.did = $1\n ", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "repo_root_cid", - "type_info": "Text" - } - ], - "parameters": { - "Left": [ - "Text" - ] - }, - "nullable": [ - false - ] - }, - "hash": "5b692e8f6d32dcbdcb45a3fff152a2be5672aadd807a4abab6914f80d57cba02" -} diff --git a/.sqlx/query-933f6585efdafedc82a8b6ac3c1513f25459bd9ab08e385ebc929469666d7747.json b/.sqlx/query-933f6585efdafedc82a8b6ac3c1513f25459bd9ab08e385ebc929469666d7747.json deleted file mode 100644 index 97f7542..0000000 --- a/.sqlx/query-933f6585efdafedc82a8b6ac3c1513f25459bd9ab08e385ebc929469666d7747.json +++ /dev/null @@ -1,28 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT id, deactivated_at FROM users WHERE did = $1", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "id", - "type_info": "Uuid" - }, - { - "ordinal": 1, - "name": "deactivated_at", - "type_info": "Timestamptz" - } - ], - "parameters": { - "Left": [ - "Text" - ] - }, - "nullable": [ - false, - true - ] - }, - "hash": "933f6585efdafedc82a8b6ac3c1513f25459bd9ab08e385ebc929469666d7747" -} diff --git a/.sqlx/query-d65ebbc09a5756438063cb6eaf8284f17beeedde25d4f41dd6788d9c60d162f7.json b/.sqlx/query-d65ebbc09a5756438063cb6eaf8284f17beeedde25d4f41dd6788d9c60d162f7.json new file mode 100644 index 0000000..e8d66dc --- /dev/null +++ b/.sqlx/query-d65ebbc09a5756438063cb6eaf8284f17beeedde25d4f41dd6788d9c60d162f7.json @@ -0,0 +1,32 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT seq, did, commit_cid\n FROM repo_seq\n WHERE event_type = 'commit'\n AND prev_cid IS NULL\n AND (blocks_cids IS NULL OR array_length(blocks_cids, 1) IS NULL OR array_length(blocks_cids, 1) = 0)\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "seq", + "type_info": "Int8" + }, + { + "ordinal": 1, + "name": "did", + "type_info": "Text" + }, + { + "ordinal": 2, + "name": "commit_cid", + "type_info": "Text" + } + ], + "parameters": { + "Left": [] + }, + "nullable": [ + false, + false, + true + ] + }, + "hash": "d65ebbc09a5756438063cb6eaf8284f17beeedde25d4f41dd6788d9c60d162f7" +} diff --git a/src/api/identity/account.rs b/src/api/identity/account.rs index 805be63..cb3bcd5 100644 --- a/src/api/identity/account.rs +++ b/src/api/identity/account.rs @@ -996,7 +996,7 @@ pub async fn create_account( warn!("Failed to sequence account event for {}: {}", did, e); } if let Err(e) = - crate::api::repo::record::sequence_empty_commit_event(&state, &did).await + crate::api::repo::record::sequence_genesis_commit(&state, &did, &commit_cid, &mst_root, &rev_str).await { warn!("Failed to sequence commit event for {}: {}", did, e); } diff --git a/src/api/repo/record/utils.rs b/src/api/repo/record/utils.rs index 6ea56f1..23c0810 100644 --- a/src/api/repo/record/utils.rs +++ b/src/api/repo/record/utils.rs @@ -512,19 +512,18 @@ pub async fn sequence_sync_event( Ok(seq_row.seq) } -pub async fn sequence_empty_commit_event(state: &AppState, did: &str) -> Result { - let repo_info = sqlx::query!( - "SELECT r.repo_root_cid, r.repo_rev FROM repos r JOIN users u ON r.user_id = u.id WHERE u.did = $1", - did - ) - .fetch_optional(&state.db) - .await - .map_err(|e| format!("DB Error fetching repo root: {}", e))? - .ok_or_else(|| "Repo not found".to_string())?; +pub async fn sequence_genesis_commit( + state: &AppState, + did: &str, + commit_cid: &Cid, + mst_root_cid: &Cid, + rev: &str, +) -> Result { let ops = serde_json::json!([]); let blobs: Vec = vec![]; - let blocks_cids: Vec = vec![]; + let blocks_cids: Vec = vec![mst_root_cid.to_string(), commit_cid.to_string()]; let prev_cid: Option<&str> = None; + let commit_cid_str = commit_cid.to_string(); let seq_row = sqlx::query!( r#" INSERT INTO repo_seq (did, event_type, commit_cid, prev_cid, ops, blobs, blocks_cids, rev) @@ -532,16 +531,16 @@ pub async fn sequence_empty_commit_event(state: &AppState, did: &str) -> Result< RETURNING seq "#, did, - repo_info.repo_root_cid, + commit_cid_str, prev_cid, ops, &blobs, &blocks_cids, - repo_info.repo_rev + rev ) .fetch_one(&state.db) .await - .map_err(|e| format!("DB Error (repo_seq empty commit): {}", e))?; + .map_err(|e| format!("DB Error (repo_seq genesis commit): {}", e))?; sqlx::query(&format!("NOTIFY repo_updates, '{}'", seq_row.seq)) .execute(&state.db) .await diff --git a/src/main.rs b/src/main.rs index 7c54349..94cab14 100644 --- a/src/main.rs +++ b/src/main.rs @@ -5,7 +5,7 @@ use tokio::sync::watch; use tracing::{error, info, warn}; use tranquil_pds::comms::{CommsService, DiscordSender, EmailSender, SignalSender, TelegramSender}; use tranquil_pds::crawlers::{Crawlers, start_crawlers_service}; -use tranquil_pds::scheduled::{backfill_repo_rev, backfill_user_blocks, start_scheduled_tasks}; +use tranquil_pds::scheduled::{backfill_genesis_commit_blocks, backfill_repo_rev, backfill_user_blocks, start_scheduled_tasks}; use tranquil_pds::state::AppState; #[tokio::main] @@ -32,6 +32,7 @@ async fn run() -> Result<(), Box> { let backfill_db = state.db.clone(); let backfill_block_store = state.block_store.clone(); tokio::spawn(async move { + backfill_genesis_commit_blocks(&backfill_db, backfill_block_store.clone()).await; backfill_repo_rev(&backfill_db, backfill_block_store.clone()).await; backfill_user_blocks(&backfill_db, backfill_block_store).await; }); diff --git a/src/scheduled.rs b/src/scheduled.rs index 0f39bd0..89f3ab8 100644 --- a/src/scheduled.rs +++ b/src/scheduled.rs @@ -13,6 +13,103 @@ use tracing::{debug, error, info, warn}; use crate::repo::PostgresBlockStore; use crate::storage::BlobStorage; +pub async fn backfill_genesis_commit_blocks(db: &PgPool, block_store: PostgresBlockStore) { + let broken_genesis_commits = match sqlx::query!( + r#" + SELECT seq, did, commit_cid + FROM repo_seq + WHERE event_type = 'commit' + AND prev_cid IS NULL + AND (blocks_cids IS NULL OR array_length(blocks_cids, 1) IS NULL OR array_length(blocks_cids, 1) = 0) + "# + ) + .fetch_all(db) + .await + { + Ok(rows) => rows, + Err(e) => { + error!("Failed to query repo_seq for genesis commit backfill: {}", e); + return; + } + }; + + if broken_genesis_commits.is_empty() { + debug!("No genesis commits need blocks_cids backfill"); + return; + } + + info!( + count = broken_genesis_commits.len(), + "Backfilling blocks_cids for genesis commits" + ); + + let mut success = 0; + let mut failed = 0; + + for commit_row in broken_genesis_commits { + let commit_cid_str = match &commit_row.commit_cid { + Some(c) => c.clone(), + None => { + warn!(seq = commit_row.seq, "Genesis commit missing commit_cid"); + failed += 1; + continue; + } + }; + + let commit_cid = match Cid::from_str(&commit_cid_str) { + Ok(c) => c, + Err(_) => { + warn!(seq = commit_row.seq, "Invalid commit CID"); + failed += 1; + continue; + } + }; + + let block = match block_store.get(&commit_cid).await { + Ok(Some(b)) => b, + Ok(None) => { + warn!(seq = commit_row.seq, cid = %commit_cid_str, "Commit block not found in store"); + failed += 1; + continue; + } + Err(e) => { + warn!(seq = commit_row.seq, error = %e, "Failed to fetch commit block"); + failed += 1; + continue; + } + }; + + let commit = match Commit::from_cbor(&block) { + Ok(c) => c, + Err(e) => { + warn!(seq = commit_row.seq, error = %e, "Failed to parse commit"); + failed += 1; + continue; + } + }; + + let mst_root_cid = commit.data; + let blocks_cids: Vec = vec![mst_root_cid.to_string(), commit_cid.to_string()]; + + if let Err(e) = sqlx::query!( + "UPDATE repo_seq SET blocks_cids = $1 WHERE seq = $2", + &blocks_cids, + commit_row.seq + ) + .execute(db) + .await + { + warn!(seq = commit_row.seq, error = %e, "Failed to update blocks_cids"); + failed += 1; + } else { + info!(seq = commit_row.seq, did = %commit_row.did, "Fixed genesis commit blocks_cids"); + success += 1; + } + } + + info!(success, failed, "Completed genesis commit blocks_cids backfill"); +} + pub async fn backfill_repo_rev(db: &PgPool, block_store: PostgresBlockStore) { let repos_missing_rev = match sqlx::query!( "SELECT user_id, repo_root_cid FROM repos WHERE repo_rev IS NULL"