From cae667d219678cff5aa439b98b169073305225fe Mon Sep 17 00:00:00 2001 From: lewis Date: Sat, 10 Jan 2026 18:38:03 +0200 Subject: [PATCH] Better repo action code quality --- src/api/admin/account/delete.rs | 2 +- src/api/admin/account/update.rs | 7 +- src/api/admin/status.rs | 18 +- src/api/delegation.rs | 21 +- src/api/identity/account.rs | 23 ++- src/api/identity/did.rs | 10 +- src/api/repo/record/batch.rs | 329 +++++++++++++++++------------- src/api/repo/record/delete.rs | 11 +- src/api/repo/record/read.rs | 47 ++--- src/api/repo/record/utils.rs | 167 +++++++++------ src/api/repo/record/validation.rs | 19 +- src/api/repo/record/write.rs | 24 +-- src/api/server/account_status.rs | 13 +- src/api/server/passkey_account.rs | 18 +- tests/commit_signing.rs | 5 +- 15 files changed, 413 insertions(+), 301 deletions(-) diff --git a/src/api/admin/account/delete.rs b/src/api/admin/account/delete.rs index f0dc447..b305fac 100644 --- a/src/api/admin/account/delete.rs +++ b/src/api/admin/account/delete.rs @@ -132,7 +132,7 @@ pub async fn delete_account( } if let Err(e) = crate::api::repo::record::sequence_account_event( &state, - did.as_str(), + did, false, Some("deleted"), ) diff --git a/src/api/admin/account/update.rs b/src/api/admin/account/update.rs index 9cd4d7e..0c0ad1e 100644 --- a/src/api/admin/account/update.rs +++ b/src/api/admin/account/update.rs @@ -2,7 +2,7 @@ use crate::api::EmptyResponse; use crate::api::error::ApiError; use crate::auth::BearerAuthAdmin; use crate::state::AppState; -use crate::types::{Did, PlainPassword}; +use crate::types::{Did, Handle, PlainPassword}; use axum::{ Json, extract::State, @@ -103,10 +103,11 @@ pub async fn update_account_handle( let _ = state.cache.delete(&format!("handle:{}", old)).await; } let _ = state.cache.delete(&format!("handle:{}", handle)).await; + let handle_typed = Handle::new_unchecked(&handle); if let Err(e) = crate::api::repo::record::sequence_identity_event( &state, - did.as_str(), - Some(&handle), + did, + Some(&handle_typed), ) .await { diff --git a/src/api/admin/status.rs b/src/api/admin/status.rs index 99b6ca1..b298dd1 100644 --- a/src/api/admin/status.rs +++ b/src/api/admin/status.rs @@ -1,6 +1,7 @@ use crate::api::error::ApiError; use crate::auth::BearerAuthAdmin; use crate::state::AppState; +use crate::types::Did; use axum::{ Json, extract::{Query, State}, @@ -183,8 +184,9 @@ pub async fn update_subject_status( let subject_type = input.subject.get("$type").and_then(|t| t.as_str()); match subject_type { Some("com.atproto.admin.defs#repoRef") => { - let did = input.subject.get("did").and_then(|d| d.as_str()); - if let Some(did) = did { + let did_str = input.subject.get("did").and_then(|d| d.as_str()); + if let Some(did_str) = did_str { + let did = Did::new_unchecked(did_str); let mut tx = match state.db.begin().await { Ok(tx) => tx, Err(e) => { @@ -201,7 +203,7 @@ pub async fn update_subject_status( if let Err(e) = sqlx::query!( "UPDATE users SET takedown_ref = $1 WHERE did = $2", takedown_ref, - did + did.as_str() ) .execute(&mut *tx) .await @@ -217,12 +219,12 @@ pub async fn update_subject_status( let result = if deactivated.applied { sqlx::query!( "UPDATE users SET deactivated_at = NOW() WHERE did = $1", - did + did.as_str() ) .execute(&mut *tx) .await } else { - sqlx::query!("UPDATE users SET deactivated_at = NULL WHERE did = $1", did) + sqlx::query!("UPDATE users SET deactivated_at = NULL WHERE did = $1", did.as_str()) .execute(&mut *tx) .await }; @@ -249,7 +251,7 @@ pub async fn update_subject_status( }; if let Err(e) = crate::api::repo::record::sequence_account_event( &state, - did, + &did, !takedown.applied, status, ) @@ -266,7 +268,7 @@ pub async fn update_subject_status( }; if let Err(e) = crate::api::repo::record::sequence_account_event( &state, - did, + &did, !deactivated.applied, status, ) @@ -276,7 +278,7 @@ pub async fn update_subject_status( } } if let Ok(Some(handle)) = - sqlx::query_scalar!("SELECT handle FROM users WHERE did = $1", did) + sqlx::query_scalar!("SELECT handle FROM users WHERE did = $1", did.as_str()) .fetch_optional(&state.db) .await { diff --git a/src/api/delegation.rs b/src/api/delegation.rs index c64857c..734cb95 100644 --- a/src/api/delegation.rs +++ b/src/api/delegation.rs @@ -4,7 +4,7 @@ use crate::auth::BearerAuth; use crate::delegation::{self, DelegationActionType}; use crate::oauth::db as oauth_db; use crate::state::{AppState, RateLimitKind}; -use crate::types::{Did, Handle}; +use crate::types::{Did, Handle, Nsid, Rkey}; use crate::util::extract_client_ip; use axum::{ Json, @@ -568,7 +568,8 @@ pub async fn create_delegated_account( .into_response(); } - let did = genesis_result.did; + let did = Did::new_unchecked(&genesis_result.did); + let handle = Handle::new_unchecked(&handle); info!(did = %did, handle = %handle, controller = %&auth.0.did, "Created DID for delegated account"); let mut tx = match state.db.begin().await { @@ -585,9 +586,9 @@ pub async fn create_delegated_account( account_type, preferred_comms_channel ) VALUES ($1, $2, $3, NULL, FALSE, 'delegated'::account_type, 'email'::comms_channel) RETURNING id"#, ) - .bind(&handle) + .bind(handle.as_str()) .bind(&email) - .bind(&did) + .bind(did.as_str()) .fetch_one(&mut *tx) .await; @@ -633,10 +634,10 @@ pub async fn create_delegated_account( if let Err(e) = sqlx::query!( r#"INSERT INTO account_delegations (delegated_did, controller_did, granted_scopes, granted_by) VALUES ($1, $2, $3, $4)"#, - did, - &auth.0.did, + did.as_str(), + auth.0.did.as_str(), input.controller_scopes, - &auth.0.did + auth.0.did.as_str() ) .execute(&mut *tx) .await @@ -736,11 +737,13 @@ pub async fn create_delegated_account( "$type": "app.bsky.actor.profile", "displayName": handle }); + let profile_collection = Nsid::new_unchecked("app.bsky.actor.profile"); + let profile_rkey = Rkey::new_unchecked("self"); if let Err(e) = crate::api::repo::record::create_record_internal( &state, &did, - "app.bsky.actor.profile", - "self", + &profile_collection, + &profile_rkey, &profile_record, ) .await diff --git a/src/api/identity/account.rs b/src/api/identity/account.rs index c7aee84..aebabeb 100644 --- a/src/api/identity/account.rs +++ b/src/api/identity/account.rs @@ -4,7 +4,7 @@ use crate::api::repo::record::utils::create_signed_commit; use crate::auth::{ServiceTokenVerifier, is_service_token}; use crate::plc::{PlcClient, create_genesis_operation, signing_key_to_did_key}; use crate::state::{AppState, RateLimitKind}; -use crate::types::{Did, Handle, PlainPassword}; +use crate::types::{Did, Handle, Nsid, PlainPassword, Rkey}; use crate::validation::validate_password; use axum::{ Json, @@ -710,8 +710,9 @@ pub async fn create_account( } }; let rev = Tid::now(LimitedU32::MIN); + let did_for_commit = Did::new_unchecked(&did); let (commit_bytes, _sig) = - match create_signed_commit(&did, mst_root, rev.as_ref(), None, &signing_key) { + match create_signed_commit(&did_for_commit, mst_root, rev.as_ref(), None, &signing_key) { Ok(result) => result, Err(e) => { error!("Error creating genesis commit: {:?}", e); @@ -793,19 +794,21 @@ pub async fn create_account( return ApiError::InternalError(None).into_response(); } if !is_migration && !is_did_web_byod { + let did_typed = Did::new_unchecked(&did); + let handle_typed = Handle::new_unchecked(&handle); if let Err(e) = - crate::api::repo::record::sequence_identity_event(&state, &did, Some(&handle)).await + crate::api::repo::record::sequence_identity_event(&state, &did_typed, Some(&handle_typed)).await { warn!("Failed to sequence identity event for {}: {}", did, e); } if let Err(e) = - crate::api::repo::record::sequence_account_event(&state, &did, true, None).await + crate::api::repo::record::sequence_account_event(&state, &did_typed, true, None).await { warn!("Failed to sequence account event for {}: {}", did, e); } if let Err(e) = crate::api::repo::record::sequence_genesis_commit( &state, - &did, + &did_typed, &commit_cid, &mst_root, &rev_str, @@ -816,7 +819,7 @@ pub async fn create_account( } if let Err(e) = crate::api::repo::record::sequence_sync_event( &state, - &did, + &did_typed, &commit_cid_str, Some(rev.as_ref()), ) @@ -828,11 +831,13 @@ pub async fn create_account( "$type": "app.bsky.actor.profile", "displayName": input.handle }); + let profile_collection = Nsid::new_unchecked("app.bsky.actor.profile"); + let profile_rkey = Rkey::new_unchecked("self"); if let Err(e) = crate::api::repo::record::create_record_internal( &state, - &did, - "app.bsky.actor.profile", - "self", + &did_typed, + &profile_collection, + &profile_rkey, &profile_record, ) .await diff --git a/src/api/identity/did.rs b/src/api/identity/did.rs index 83077fd..ed3b608 100644 --- a/src/api/identity/did.rs +++ b/src/api/identity/did.rs @@ -2,6 +2,7 @@ use crate::api::{ApiError, DidResponse, EmptyResponse}; use crate::auth::BearerAuthAllowDeactivated; use crate::plc::signing_key_to_did_key; use crate::state::AppState; +use crate::types::Handle; use axum::{ Json, extract::{Path, Query, State}, @@ -669,8 +670,9 @@ pub async fn update_handle( format!("{}.{}", new_handle, hostname) }; if full_handle == current_handle { + let handle_typed = Handle::new_unchecked(&full_handle); if let Err(e) = - crate::api::repo::record::sequence_identity_event(&state, &did, Some(&full_handle)) + crate::api::repo::record::sequence_identity_event(&state, &did, Some(&handle_typed)) .await { warn!("Failed to sequence identity event for handle update: {}", e); @@ -692,8 +694,9 @@ pub async fn update_handle( full_handle } else { if new_handle == current_handle { + let handle_typed = Handle::new_unchecked(&new_handle); if let Err(e) = - crate::api::repo::record::sequence_identity_event(&state, &did, Some(&new_handle)) + crate::api::repo::record::sequence_identity_event(&state, &did, Some(&handle_typed)) .await { warn!("Failed to sequence identity event for handle update: {}", e); @@ -749,8 +752,9 @@ pub async fn update_handle( .await; } let _ = state.cache.delete(&format!("handle:{}", handle)).await; + let handle_typed = Handle::new_unchecked(&handle); if let Err(e) = - crate::api::repo::record::sequence_identity_event(&state, &did, Some(&handle)).await + crate::api::repo::record::sequence_identity_event(&state, &did, Some(&handle_typed)).await { warn!("Failed to sequence identity event for handle update: {}", e); } diff --git a/src/api/repo/record/batch.rs b/src/api/repo/record/batch.rs index 5ef8bd8..d752749 100644 --- a/src/api/repo/record/batch.rs +++ b/src/api/repo/record/batch.rs @@ -6,7 +6,7 @@ use crate::auth::BearerAuth; use crate::delegation::{self, DelegationActionType}; use crate::repo::tracking::TrackingBlockStore; use crate::state::AppState; -use crate::types::{AtIdentifier, AtUri, Nsid, Rkey}; +use crate::types::{AtIdentifier, AtUri, Did, Nsid, Rkey}; use axum::{ Json, extract::State, @@ -23,6 +23,183 @@ use tracing::{error, info}; const MAX_BATCH_WRITES: usize = 200; +struct WriteAccumulator { + mst: Mst, + results: Vec, + ops: Vec, + modified_keys: Vec, + all_blob_cids: Vec, +} + +async fn process_single_write( + write: &WriteOp, + acc: WriteAccumulator, + did: &Did, + validate: Option, + tracking_store: &TrackingBlockStore, +) -> Result { + let WriteAccumulator { + mst, + mut results, + mut ops, + mut modified_keys, + mut all_blob_cids, + } = acc; + + match write { + WriteOp::Create { + collection, + rkey, + value, + } => { + let validation_status = match validate { + Some(false) => None, + _ => { + let require_lexicon = validate == Some(true); + match validate_record_with_status( + value, + collection, + rkey.as_ref(), + require_lexicon, + ) { + Ok(status) => Some(status), + Err(err_response) => return Err(*err_response), + } + } + }; + all_blob_cids.extend(extract_blob_cids(value)); + let rkey = rkey.clone().unwrap_or_else(Rkey::generate); + let record_ipld = crate::util::json_to_ipld(value); + let record_bytes = serde_ipld_dagcbor::to_vec(&record_ipld).map_err(|_| { + ApiError::InvalidRecord("Failed to serialize record".into()).into_response() + })?; + let record_cid = tracking_store.put(&record_bytes).await.map_err(|_| { + ApiError::InternalError(Some("Failed to store record".into())).into_response() + })?; + let key = format!("{}/{}", collection, rkey); + modified_keys.push(key.clone()); + let new_mst = mst.add(&key, record_cid).await.map_err(|_| { + ApiError::InternalError(Some("Failed to add to MST".into())).into_response() + })?; + let uri = AtUri::from_parts(did, collection, &rkey); + results.push(WriteResult::CreateResult { + uri, + cid: record_cid.to_string(), + validation_status: validation_status.map(|s| s.to_string()), + }); + ops.push(RecordOp::Create { + collection: collection.clone(), + rkey: rkey.clone(), + cid: record_cid, + }); + Ok(WriteAccumulator { + mst: new_mst, + results, + ops, + modified_keys, + all_blob_cids, + }) + } + WriteOp::Update { + collection, + rkey, + value, + } => { + let validation_status = match validate { + Some(false) => None, + _ => { + let require_lexicon = validate == Some(true); + match validate_record_with_status( + value, + collection, + Some(rkey), + require_lexicon, + ) { + Ok(status) => Some(status), + Err(err_response) => return Err(*err_response), + } + } + }; + all_blob_cids.extend(extract_blob_cids(value)); + let record_ipld = crate::util::json_to_ipld(value); + let record_bytes = serde_ipld_dagcbor::to_vec(&record_ipld).map_err(|_| { + ApiError::InvalidRecord("Failed to serialize record".into()).into_response() + })?; + let record_cid = tracking_store.put(&record_bytes).await.map_err(|_| { + ApiError::InternalError(Some("Failed to store record".into())).into_response() + })?; + let key = format!("{}/{}", collection, rkey); + modified_keys.push(key.clone()); + let prev_record_cid = mst.get(&key).await.ok().flatten(); + let new_mst = mst.update(&key, record_cid).await.map_err(|_| { + ApiError::InternalError(Some("Failed to update MST".into())).into_response() + })?; + let uri = AtUri::from_parts(did, collection, rkey); + results.push(WriteResult::UpdateResult { + uri, + cid: record_cid.to_string(), + validation_status: validation_status.map(|s| s.to_string()), + }); + ops.push(RecordOp::Update { + collection: collection.clone(), + rkey: rkey.clone(), + cid: record_cid, + prev: prev_record_cid, + }); + Ok(WriteAccumulator { + mst: new_mst, + results, + ops, + modified_keys, + all_blob_cids, + }) + } + WriteOp::Delete { collection, rkey } => { + let key = format!("{}/{}", collection, rkey); + modified_keys.push(key.clone()); + let prev_record_cid = mst.get(&key).await.ok().flatten(); + let new_mst = mst.delete(&key).await.map_err(|_| { + ApiError::InternalError(Some("Failed to delete from MST".into())).into_response() + })?; + results.push(WriteResult::DeleteResult {}); + ops.push(RecordOp::Delete { + collection: collection.clone(), + rkey: rkey.clone(), + prev: prev_record_cid, + }); + Ok(WriteAccumulator { + mst: new_mst, + results, + ops, + modified_keys, + all_blob_cids, + }) + } + } +} + +async fn process_writes( + writes: &[WriteOp], + initial_mst: Mst, + did: &Did, + validate: Option, + tracking_store: &TrackingBlockStore, +) -> Result { + use futures::stream::{self, TryStreamExt}; + let initial_acc = WriteAccumulator { + mst: initial_mst, + results: Vec::new(), + ops: Vec::new(), + modified_keys: Vec::new(), + all_blob_cids: Vec::new(), + }; + stream::iter(writes.iter().map(Ok::<_, Response>)) + .try_fold(initial_acc, |acc, write| async move { + process_single_write(write, acc, did, validate, tracking_store).await + }) + .await +} + #[derive(Deserialize)] #[serde(tag = "$type")] pub enum WriteOp { @@ -237,144 +414,18 @@ pub async fn apply_writes( _ => return ApiError::InternalError(Some("Failed to parse commit".into())).into_response(), }; let original_mst = Mst::load(Arc::new(tracking_store.clone()), commit.data, None); - let mut mst = Mst::load(Arc::new(tracking_store.clone()), commit.data, None); - let mut results: Vec = Vec::new(); - let mut ops: Vec = Vec::new(); - let mut modified_keys: Vec = Vec::new(); - let mut all_blob_cids: Vec = Vec::new(); - for write in &input.writes { - match write { - WriteOp::Create { - collection, - rkey, - value, - } => { - let validation_status = if input.validate == Some(false) { - None - } else { - let require_lexicon = input.validate == Some(true); - match validate_record_with_status( - value, - collection, - rkey.as_ref().map(|r| r.as_str()), - require_lexicon, - ) { - Ok(status) => Some(status), - Err(err_response) => return *err_response, - } - }; - all_blob_cids.extend(extract_blob_cids(value)); - let rkey = rkey.clone().unwrap_or_else(Rkey::generate); - let record_ipld = crate::util::json_to_ipld(value); - let mut record_bytes = Vec::new(); - if serde_ipld_dagcbor::to_writer(&mut record_bytes, &record_ipld).is_err() { - return ApiError::InvalidRecord("Failed to serialize record".into()) - .into_response(); - } - let record_cid = match tracking_store.put(&record_bytes).await { - Ok(c) => c, - Err(_) => { - return ApiError::InternalError(Some("Failed to store record".into())) - .into_response(); - } - }; - let key = format!("{}/{}", collection, rkey); - modified_keys.push(key.clone()); - mst = match mst.add(&key, record_cid).await { - Ok(m) => m, - Err(_) => { - return ApiError::InternalError(Some("Failed to add to MST".into())) - .into_response(); - } - }; - let uri = AtUri::from_parts(&did, collection, &rkey); - results.push(WriteResult::CreateResult { - uri, - cid: record_cid.to_string(), - validation_status: validation_status.map(|s| s.to_string()), - }); - ops.push(RecordOp::Create { - collection: collection.to_string(), - rkey: rkey.to_string(), - cid: record_cid, - }); - } - WriteOp::Update { - collection, - rkey, - value, - } => { - let validation_status = if input.validate == Some(false) { - None - } else { - let require_lexicon = input.validate == Some(true); - match validate_record_with_status( - value, - collection, - Some(rkey.as_str()), - require_lexicon, - ) { - Ok(status) => Some(status), - Err(err_response) => return *err_response, - } - }; - all_blob_cids.extend(extract_blob_cids(value)); - let record_ipld = crate::util::json_to_ipld(value); - let mut record_bytes = Vec::new(); - if serde_ipld_dagcbor::to_writer(&mut record_bytes, &record_ipld).is_err() { - return ApiError::InvalidRecord("Failed to serialize record".into()) - .into_response(); - } - let record_cid = match tracking_store.put(&record_bytes).await { - Ok(c) => c, - Err(_) => { - return ApiError::InternalError(Some("Failed to store record".into())) - .into_response(); - } - }; - let key = format!("{}/{}", collection, rkey); - modified_keys.push(key.clone()); - let prev_record_cid = mst.get(&key).await.ok().flatten(); - mst = match mst.update(&key, record_cid).await { - Ok(m) => m, - Err(_) => { - return ApiError::InternalError(Some("Failed to update MST".into())) - .into_response(); - } - }; - let uri = AtUri::from_parts(&did, collection, rkey); - results.push(WriteResult::UpdateResult { - uri, - cid: record_cid.to_string(), - validation_status: validation_status.map(|s| s.to_string()), - }); - ops.push(RecordOp::Update { - collection: collection.to_string(), - rkey: rkey.to_string(), - cid: record_cid, - prev: prev_record_cid, - }); - } - WriteOp::Delete { collection, rkey } => { - let key = format!("{}/{}", collection, rkey); - modified_keys.push(key.clone()); - let prev_record_cid = mst.get(&key).await.ok().flatten(); - mst = match mst.delete(&key).await { - Ok(m) => m, - Err(_) => { - return ApiError::InternalError(Some("Failed to delete from MST".into())) - .into_response(); - } - }; - results.push(WriteResult::DeleteResult {}); - ops.push(RecordOp::Delete { - collection: collection.to_string(), - rkey: rkey.to_string(), - prev: prev_record_cid, - }); - } - } - } + let initial_mst = Mst::load(Arc::new(tracking_store.clone()), commit.data, None); + let WriteAccumulator { + mst, + results, + ops, + modified_keys, + all_blob_cids, + } = match process_writes(&input.writes, initial_mst, &did, input.validate, &tracking_store).await + { + Ok(acc) => acc, + Err(response) => return response, + }; let new_mst_root = match mst.persist().await { Ok(c) => c, Err(_) => { diff --git a/src/api/repo/record/delete.rs b/src/api/repo/record/delete.rs index 06589a6..e65019c 100644 --- a/src/api/repo/record/delete.rs +++ b/src/api/repo/record/delete.rs @@ -65,13 +65,6 @@ pub async fn delete_record( return e; } - if crate::util::is_account_migrated(&state.db, &auth.did) - .await - .unwrap_or(false) - { - return ApiError::AccountMigrated.into_response(); - } - let did = auth.did; let user_id = auth.user_id; let current_root_cid = auth.current_root_cid; @@ -125,8 +118,8 @@ pub async fn delete_record( let collection_for_audit = input.collection.to_string(); let rkey_for_audit = input.rkey.to_string(); let op = RecordOp::Delete { - collection: input.collection.to_string(), - rkey: rkey_for_audit.clone(), + collection: input.collection.clone(), + rkey: input.rkey.clone(), prev: prev_record_cid, }; let mut new_mst_blocks = std::collections::BTreeMap::new(); diff --git a/src/api/repo/record/read.rs b/src/api/repo/record/read.rs index bab93d7..ae9a40d 100644 --- a/src/api/repo/record/read.rs +++ b/src/api/repo/record/read.rs @@ -13,7 +13,6 @@ use ipld_core::ipld::Ipld; use jacquard_repo::storage::BlockStore; use serde::{Deserialize, Serialize}; use serde_json::{Map, Value, json}; -use std::collections::HashMap; use std::str::FromStr; use tracing::error; @@ -237,14 +236,15 @@ pub async fn list_records( } }; let last_rkey = rows.last().map(|(rkey, _)| rkey.clone()); - let mut cid_to_rkey: HashMap = HashMap::new(); - let mut cids: Vec = Vec::with_capacity(rows.len()); - for (rkey, cid_str) in &rows { - if let Ok(cid) = Cid::from_str(cid_str) { - cid_to_rkey.insert(cid, (rkey.clone(), cid_str.clone())); - cids.push(cid); - } - } + let parsed_rows: Vec<(Cid, String, String)> = rows + .iter() + .filter_map(|(rkey, cid_str)| { + Cid::from_str(cid_str) + .ok() + .map(|cid| (cid, rkey.clone(), cid_str.clone())) + }) + .collect(); + let cids: Vec = parsed_rows.iter().map(|(cid, _, _)| *cid).collect(); let blocks = match state.block_store.get_many(&cids).await { Ok(b) => b, Err(e) => { @@ -252,20 +252,21 @@ pub async fn list_records( return ApiError::InternalError(None).into_response(); } }; - let mut records = Vec::new(); - for (cid, block_opt) in cids.iter().zip(blocks.into_iter()) { - if let Some(block) = block_opt - && let Some((rkey, cid_str)) = cid_to_rkey.get(cid) - && let Ok(ipld) = serde_ipld_dagcbor::from_slice::(&block) - { - let value = ipld_to_json(ipld); - records.push(json!({ - "uri": format!("at://{}/{}/{}", input.repo, input.collection, rkey), - "cid": cid_str, - "value": value - })); - } - } + let records: Vec = parsed_rows + .iter() + .zip(blocks.into_iter()) + .filter_map(|((_, rkey, cid_str), block_opt)| { + block_opt.and_then(|block| { + serde_ipld_dagcbor::from_slice::(&block).ok().map(|ipld| { + json!({ + "uri": format!("at://{}/{}/{}", input.repo, input.collection, rkey), + "cid": cid_str, + "value": ipld_to_json(ipld) + }) + }) + }) + }) + .collect(); Json(ListRecordsOutput { cursor: last_rkey, records, diff --git a/src/api/repo/record/utils.rs b/src/api/repo/record/utils.rs index e9d6f80..dd46c58 100644 --- a/src/api/repo/record/utils.rs +++ b/src/api/repo/record/utils.rs @@ -1,4 +1,5 @@ use crate::state::AppState; +use crate::types::{Did, Handle, Nsid, Rkey}; use bytes::Bytes; use cid::Cid; use jacquard::types::{integer::LimitedU32, string::Tid}; @@ -38,14 +39,14 @@ fn extract_blob_cids_recursive(value: &Value, blobs: &mut Vec) { } pub fn create_signed_commit( - did: &str, + did: &Did, data: Cid, rev: &str, prev: Option, signing_key: &SigningKey, ) -> Result<(Vec, Bytes), String> { - let did = - jacquard::types::string::Did::new(did).map_err(|e| format!("Invalid DID: {:?}", e))?; + let did = jacquard::types::string::Did::new(did.as_str()) + .map_err(|e| format!("Invalid DID: {:?}", e))?; let rev = jacquard::types::string::Tid::from_str(rev).map_err(|e| format!("Invalid TID: {:?}", e))?; let unsigned = Commit::new_unsigned(did, data, rev, prev); @@ -61,19 +62,19 @@ pub fn create_signed_commit( pub enum RecordOp { Create { - collection: String, - rkey: String, + collection: Nsid, + rkey: Rkey, cid: Cid, }, Update { - collection: String, - rkey: String, + collection: Nsid, + rkey: Rkey, cid: Cid, prev: Option, }, Delete { - collection: String, - rkey: String, + collection: Nsid, + rkey: Rkey, prev: Option, }, } @@ -84,7 +85,7 @@ pub struct CommitResult { } pub struct CommitParams<'a> { - pub did: &'a str, + pub did: &'a Did, pub user_id: Uuid, pub current_root_cid: Option, pub prev_data_cid: Option, @@ -218,36 +219,44 @@ pub async fn commit_and_log( .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(); - let mut delete_collections: Vec = Vec::new(); - let mut delete_rkeys: Vec = Vec::new(); - for op in &ops { - match op { - RecordOp::Create { - collection, - rkey, - cid, - } - | RecordOp::Update { - collection, - rkey, - cid, - .. - } => { - upsert_collections.push(collection.clone()); - upsert_rkeys.push(rkey.clone()); - upsert_cids.push(cid.to_string()); - } + let (upserts, deletes): (Vec<_>, Vec<_>) = ops.iter().partition(|op| { + matches!(op, RecordOp::Create { .. } | RecordOp::Update { .. }) + }); + let (upsert_collections, upsert_rkeys, upsert_cids): (Vec, Vec, Vec) = + upserts + .into_iter() + .filter_map(|op| match op { + RecordOp::Create { + collection, + rkey, + cid, + } + | RecordOp::Update { + collection, + rkey, + cid, + .. + } => Some((collection.to_string(), rkey.to_string(), cid.to_string())), + _ => None, + }) + .fold( + (Vec::new(), Vec::new(), Vec::new()), + |(mut cols, mut rkeys, mut cids), (c, r, ci)| { + cols.push(c); + rkeys.push(r); + cids.push(ci); + (cols, rkeys, cids) + }, + ); + let (delete_collections, delete_rkeys): (Vec, Vec) = deletes + .into_iter() + .filter_map(|op| match op { RecordOp::Delete { collection, rkey, .. - } => { - delete_collections.push(collection.clone()); - delete_rkeys.push(rkey.clone()); - } - } - } + } => Some((collection.to_string(), rkey.to_string())), + _ => None, + }) + .unzip(); if !upsert_collections.is_empty() { sqlx::query!( r#" @@ -337,7 +346,7 @@ pub async fn commit_and_log( VALUES ($1, $2, $3, $4, $5, $6, $7, $8) RETURNING seq "#, - did, + did.as_str(), event_type, new_root_cid.to_string(), prev_cid_str, @@ -367,15 +376,15 @@ pub async fn commit_and_log( } pub async fn create_record_internal( state: &AppState, - did: &str, - collection: &str, - rkey: &str, + did: &Did, + collection: &Nsid, + rkey: &Rkey, record: &serde_json::Value, ) -> Result<(String, Cid), String> { use crate::repo::tracking::TrackingBlockStore; use jacquard_repo::mst::Mst; use std::sync::Arc; - let user_id: Uuid = sqlx::query_scalar!("SELECT id FROM users WHERE did = $1", did) + let user_id: Uuid = sqlx::query_scalar!("SELECT id FROM users WHERE did = $1", did.as_str()) .fetch_optional(&state.db) .await .map_err(|e| format!("DB error: {}", e))? @@ -417,8 +426,8 @@ pub async fn create_record_internal( .await .map_err(|e| format!("Failed to persist MST: {:?}", e))?; let op = RecordOp::Create { - collection: collection.to_string(), - rkey: rkey.to_string(), + collection: collection.clone(), + rkey: rkey.clone(), cid: record_cid, }; let mut new_mst_blocks = std::collections::BTreeMap::new(); @@ -471,81 +480,105 @@ pub async fn create_record_internal( pub async fn sequence_identity_event( state: &AppState, - did: &str, - handle: Option<&str>, + did: &Did, + handle: Option<&Handle>, ) -> Result { + let mut tx = state + .db + .begin() + .await + .map_err(|e| format!("Failed to begin transaction: {}", e))?; let seq_row = sqlx::query!( r#" INSERT INTO repo_seq (did, event_type, handle) VALUES ($1, 'identity', $2) RETURNING seq "#, - did, - handle, + did.as_str(), + handle.map(|h| h.as_str()), ) - .fetch_one(&state.db) + .fetch_one(&mut *tx) .await .map_err(|e| format!("DB Error (repo_seq identity): {}", e))?; sqlx::query(&format!("NOTIFY repo_updates, '{}'", seq_row.seq)) - .execute(&state.db) + .execute(&mut *tx) .await .map_err(|e| format!("DB Error (notify): {}", e))?; + tx.commit() + .await + .map_err(|e| format!("Failed to commit transaction: {}", e))?; Ok(seq_row.seq) } pub async fn sequence_account_event( state: &AppState, - did: &str, + did: &Did, active: bool, status: Option<&str>, ) -> Result { + let mut tx = state + .db + .begin() + .await + .map_err(|e| format!("Failed to begin transaction: {}", e))?; let seq_row = sqlx::query!( r#" INSERT INTO repo_seq (did, event_type, active, status) VALUES ($1, 'account', $2, $3) RETURNING seq "#, - did, + did.as_str(), active, status, ) - .fetch_one(&state.db) + .fetch_one(&mut *tx) .await .map_err(|e| format!("DB Error (repo_seq account): {}", e))?; sqlx::query(&format!("NOTIFY repo_updates, '{}'", seq_row.seq)) - .execute(&state.db) + .execute(&mut *tx) .await .map_err(|e| format!("DB Error (notify): {}", e))?; + tx.commit() + .await + .map_err(|e| format!("Failed to commit transaction: {}", e))?; Ok(seq_row.seq) } pub async fn sequence_sync_event( state: &AppState, - did: &str, + did: &Did, commit_cid: &str, rev: Option<&str>, ) -> Result { + let mut tx = state + .db + .begin() + .await + .map_err(|e| format!("Failed to begin transaction: {}", e))?; let seq_row = sqlx::query!( r#" INSERT INTO repo_seq (did, event_type, commit_cid, rev) VALUES ($1, 'sync', $2, $3) RETURNING seq "#, - did, + did.as_str(), commit_cid, rev, ) - .fetch_one(&state.db) + .fetch_one(&mut *tx) .await .map_err(|e| format!("DB Error (repo_seq sync): {}", e))?; sqlx::query(&format!("NOTIFY repo_updates, '{}'", seq_row.seq)) - .execute(&state.db) + .execute(&mut *tx) .await .map_err(|e| format!("DB Error (notify): {}", e))?; + tx.commit() + .await + .map_err(|e| format!("Failed to commit transaction: {}", e))?; Ok(seq_row.seq) } pub async fn sequence_genesis_commit( state: &AppState, - did: &str, + did: &Did, commit_cid: &Cid, mst_root_cid: &Cid, rev: &str, @@ -555,13 +588,18 @@ pub async fn sequence_genesis_commit( 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 mut tx = state + .db + .begin() + .await + .map_err(|e| format!("Failed to begin transaction: {}", e))?; let seq_row = sqlx::query!( r#" INSERT INTO repo_seq (did, event_type, commit_cid, prev_cid, ops, blobs, blocks_cids, rev) VALUES ($1, 'commit', $2, $3::TEXT, $4, $5, $6, $7) RETURNING seq "#, - did, + did.as_str(), commit_cid_str, prev_cid, ops, @@ -569,12 +607,15 @@ pub async fn sequence_genesis_commit( &blocks_cids, rev ) - .fetch_one(&state.db) + .fetch_one(&mut *tx) .await .map_err(|e| format!("DB Error (repo_seq genesis commit): {}", e))?; sqlx::query(&format!("NOTIFY repo_updates, '{}'", seq_row.seq)) - .execute(&state.db) + .execute(&mut *tx) .await .map_err(|e| format!("DB Error (notify): {}", e))?; + tx.commit() + .await + .map_err(|e| format!("Failed to commit transaction: {}", e))?; Ok(seq_row.seq) } diff --git a/src/api/repo/record/validation.rs b/src/api/repo/record/validation.rs index fe10020..56c20fe 100644 --- a/src/api/repo/record/validation.rs +++ b/src/api/repo/record/validation.rs @@ -1,28 +1,33 @@ use crate::api::error::ApiError; +use crate::types::{Nsid, Rkey}; use crate::validation::{RecordValidator, ValidationError, ValidationStatus}; use axum::response::Response; -pub fn validate_record(record: &serde_json::Value, collection: &str) -> Result<(), Box> { +pub fn validate_record(record: &serde_json::Value, collection: &Nsid) -> Result<(), Box> { validate_record_with_rkey(record, collection, None) } pub fn validate_record_with_rkey( record: &serde_json::Value, - collection: &str, - rkey: Option<&str>, + collection: &Nsid, + rkey: Option<&Rkey>, ) -> Result<(), Box> { let validator = RecordValidator::new(); - validation_error_to_response(validator.validate_with_rkey(record, collection, rkey)) + validation_error_to_response(validator.validate_with_rkey( + record, + collection.as_str(), + rkey.map(|r| r.as_str()), + )) } pub fn validate_record_with_status( record: &serde_json::Value, - collection: &str, - rkey: Option<&str>, + collection: &Nsid, + rkey: Option<&Rkey>, require_lexicon: bool, ) -> Result> { let validator = RecordValidator::new().require_lexicon(require_lexicon); - match validator.validate_with_rkey(record, collection, rkey) { + match validator.validate_with_rkey(record, collection.as_str(), rkey.map(|r| r.as_str())) { Ok(status) => Ok(status), Err(e) => Err(validation_error_to_box_response(e)), } diff --git a/src/api/repo/record/write.rs b/src/api/repo/record/write.rs index da25ba4..b5afb53 100644 --- a/src/api/repo/record/write.rs +++ b/src/api/repo/record/write.rs @@ -21,7 +21,7 @@ use std::sync::Arc; use tracing::error; use uuid::Uuid; -pub async fn has_verified_comms_channel(db: &PgPool, did: &str) -> Result { +pub async fn has_verified_comms_channel(db: &PgPool, did: &Did) -> Result { let row = sqlx::query( r#" SELECT @@ -33,7 +33,7 @@ pub async fn has_verified_comms_channel(db: &PgPool, did: &str) -> Result Result { @@ -96,7 +96,7 @@ pub async fn prepare_repo_write( } response })?; - if repo_did != auth_user.did { + if repo.as_str() != auth_user.did.as_str() { return Err( ApiError::InvalidRepo("Repo does not match authenticated user".into()).into_response(), ); @@ -229,7 +229,7 @@ pub async fn create_record( match validate_record_with_status( &input.record, &input.collection, - input.rkey.as_ref().map(|r| r.as_str()), + input.rkey.as_ref(), require_lexicon, ) { Ok(status) => Some(status), @@ -259,8 +259,8 @@ pub async fn create_record( _ => return ApiError::InternalError(Some("Failed to persist MST".into())).into_response(), }; let op = RecordOp::Create { - collection: input.collection.to_string(), - rkey: rkey.to_string(), + collection: input.collection.clone(), + rkey: rkey.clone(), cid: record_cid, }; let mut new_mst_blocks = std::collections::BTreeMap::new(); @@ -443,7 +443,7 @@ pub async fn put_record( match validate_record_with_status( &input.record, &input.collection, - Some(input.rkey.as_str()), + Some(&input.rkey), require_lexicon, ) { Ok(status) => Some(status), @@ -510,15 +510,15 @@ pub async fn put_record( }; let op = if existing_cid.is_some() { RecordOp::Update { - collection: input.collection.to_string(), - rkey: input.rkey.to_string(), + collection: input.collection.clone(), + rkey: input.rkey.clone(), cid: record_cid, prev: existing_cid, } } else { RecordOp::Create { - collection: input.collection.to_string(), - rkey: input.rkey.to_string(), + collection: input.collection.clone(), + rkey: input.rkey.clone(), cid: record_cid, } }; diff --git a/src/api/server/account_status.rs b/src/api/server/account_status.rs index 41aa487..5c628e8 100644 --- a/src/api/server/account_status.rs +++ b/src/api/server/account_status.rs @@ -3,7 +3,7 @@ use crate::api::error::ApiError; use crate::cache::Cache; use crate::plc::PlcClient; use crate::state::AppState; -use crate::types::PlainPassword; +use crate::types::{Handle, PlainPassword}; use axum::{ Json, extract::State, @@ -449,7 +449,7 @@ pub async fn activate_account( did ); if let Err(e) = - crate::api::repo::record::sequence_account_event(&state, did.as_str(), true, None) + crate::api::repo::record::sequence_account_event(&state, &did, true, None) .await { warn!( @@ -463,10 +463,11 @@ pub async fn activate_account( "[MIGRATION] activateAccount: Sequencing identity event for did={} handle={:?}", did, handle ); + let handle_typed = handle.as_ref().map(|h| Handle::new_unchecked(h)); if let Err(e) = crate::api::repo::record::sequence_identity_event( &state, - did.as_str(), - handle.as_deref(), + &did, + handle_typed.as_ref(), ) .await { @@ -501,7 +502,7 @@ pub async fn activate_account( }; if let Err(e) = crate::api::repo::record::sequence_sync_event( &state, - did.as_str(), + &did, &root_cid, rev.as_deref(), ) @@ -609,7 +610,7 @@ pub async fn deactivate_account( } if let Err(e) = crate::api::repo::record::sequence_account_event( &state, - did.as_str(), + &did, false, Some("deactivated"), ) diff --git a/src/api/server/passkey_account.rs b/src/api/server/passkey_account.rs index d3da564..cf7b2d1 100644 --- a/src/api/server/passkey_account.rs +++ b/src/api/server/passkey_account.rs @@ -20,7 +20,7 @@ use uuid::Uuid; use crate::api::repo::record::utils::create_signed_commit; use crate::auth::{ServiceTokenVerifier, is_service_token}; use crate::state::{AppState, RateLimitKind}; -use crate::types::{Did, Handle, PlainPassword}; +use crate::types::{Did, Handle, Nsid, PlainPassword, Rkey}; use crate::validation::validate_password; fn extract_client_ip(headers: &HeaderMap) -> String { @@ -512,8 +512,9 @@ pub async fn create_passkey_account( } }; let rev = Tid::now(LimitedU32::MIN); + let did_typed = Did::new_unchecked(&did); let (commit_bytes, _sig) = - match create_signed_commit(&did, mst_root, rev.as_ref(), None, &secret_key) { + match create_signed_commit(&did_typed, mst_root, rev.as_ref(), None, &secret_key) { Ok(result) => result, Err(e) => { error!("Error creating genesis commit: {:?}", e); @@ -600,13 +601,14 @@ pub async fn create_passkey_account( } if !is_byod_did_web { + let handle_typed = Handle::new_unchecked(&handle); if let Err(e) = - crate::api::repo::record::sequence_identity_event(&state, &did, Some(&handle)).await + crate::api::repo::record::sequence_identity_event(&state, &did_typed, Some(&handle_typed)).await { warn!("Failed to sequence identity event for {}: {}", did, e); } if let Err(e) = - crate::api::repo::record::sequence_account_event(&state, &did, true, None).await + crate::api::repo::record::sequence_account_event(&state, &did_typed, true, None).await { warn!("Failed to sequence account event for {}: {}", did, e); } @@ -614,11 +616,13 @@ pub async fn create_passkey_account( "$type": "app.bsky.actor.profile", "displayName": handle }); + let profile_collection = Nsid::new_unchecked("app.bsky.actor.profile"); + let profile_rkey = Rkey::new_unchecked("self"); if let Err(e) = crate::api::repo::record::create_record_internal( &state, - &did, - "app.bsky.actor.profile", - "self", + &did_typed, + &profile_collection, + &profile_rkey, &profile_record, ) .await diff --git a/tests/commit_signing.rs b/tests/commit_signing.rs index a10e0d4..7d1c765 100644 --- a/tests/commit_signing.rs +++ b/tests/commit_signing.rs @@ -3,6 +3,7 @@ use jacquard::types::{integer::LimitedU32, string::Tid}; use jacquard_repo::commit::Commit; use k256::ecdsa::SigningKey; use std::str::FromStr; +use tranquil_pds::Did; #[test] fn test_commit_signing_produces_valid_signature() { @@ -98,12 +99,12 @@ fn test_create_signed_commit_helper() { use tranquil_pds::api::repo::record::utils::create_signed_commit; let signing_key = SigningKey::random(&mut rand::thread_rng()); - let did = "did:plc:testuser123456789abcdef"; + let did = Did::new_unchecked("did:plc:testuser123456789abcdef"); let data_cid = Cid::from_str("bafyreib2rxk3ryblouj3fxza5jvx6psmwewwessc4m6g6e7pqhhkwqomfi").unwrap(); let rev = Tid::now(LimitedU32::MIN).to_string(); - let (signed_bytes, sig) = create_signed_commit(did, data_cid, &rev, None, &signing_key) + let (signed_bytes, sig) = create_signed_commit(&did, data_cid, &rev, None, &signing_key) .expect("signing should succeed"); assert!(!signed_bytes.is_empty());