From 7e3dca4026e313dcf1efe9a11d1fe0ddb7b30a7a Mon Sep 17 00:00:00 2001 From: Lewis Date: Thu, 19 Mar 2026 19:51:07 +0200 Subject: [PATCH] refactor(api): update repo batch/delete to use repo_ops, clean up remaining repo endpoints --- crates/tranquil-api/src/repo/blob.rs | 20 +- crates/tranquil-api/src/repo/import.rs | 10 +- crates/tranquil-api/src/repo/meta.rs | 53 +-- crates/tranquil-api/src/repo/record/batch.rs | 308 ++++++------------ crates/tranquil-api/src/repo/record/delete.rs | 228 ++++--------- crates/tranquil-api/src/server/mod.rs | 15 +- .../tranquil-api/src/server/service_auth.rs | 4 +- .../tranquil-api/src/server/verify_email.rs | 21 +- crates/tranquil-api/src/temp.rs | 8 +- 9 files changed, 187 insertions(+), 480 deletions(-) diff --git a/crates/tranquil-api/src/repo/blob.rs b/crates/tranquil-api/src/repo/blob.rs index 646b1f0..ef2270d 100644 --- a/crates/tranquil-api/src/repo/blob.rs +++ b/crates/tranquil-api/src/repo/blob.rs @@ -2,7 +2,6 @@ use axum::body::Body; use axum::{ Json, extract::{Query, State}, - http::StatusCode, response::{IntoResponse, Response}, }; use bytes::Bytes; @@ -58,10 +57,7 @@ pub async fn upload_blob( } let mime_type_for_check = get_header_str(&headers, http::header::CONTENT_TYPE) .unwrap_or("application/octet-stream"); - let scope_proof = match user.verify_blob_upload(mime_type_for_check) { - Ok(proof) => proof, - Err(e) => return Ok(e.into_response()), - }; + let scope_proof = user.verify_blob_upload(mime_type_for_check)?; ( scope_proof.principal_did().into_did(), scope_proof.controller_did().map(|c| c.into_did()), @@ -237,7 +233,7 @@ pub async fn list_missing_blobs( State(state): State, auth: Auth, Query(params): Query, -) -> Result { +) -> Result, ApiError> { let did = &auth.did; let user = state .user_repo @@ -269,12 +265,8 @@ pub async fn list_missing_blobs( } else { None }; - Ok(( - StatusCode::OK, - Json(ListMissingBlobsOutput { - cursor: next_cursor, - blobs, - }), - ) - .into_response()) + Ok(Json(ListMissingBlobsOutput { + cursor: next_cursor, + blobs, + })) } diff --git a/crates/tranquil-api/src/repo/import.rs b/crates/tranquil-api/src/repo/import.rs index b416bc7..5122416 100644 --- a/crates/tranquil-api/src/repo/import.rs +++ b/crates/tranquil-api/src/repo/import.rs @@ -1,8 +1,4 @@ -use axum::{ - body::Bytes, - extract::State, - response::{IntoResponse, Response}, -}; +use axum::{Json, body::Bytes, extract::State}; use jacquard_common::types::{integer::LimitedU32, string::Tid}; use jacquard_repo::storage::BlockStore; use k256::ecdsa::SigningKey; @@ -22,7 +18,7 @@ pub async fn import_repo( State(state): State, auth: Auth, body: Bytes, -) -> Result { +) -> Result, ApiError> { let accepting_imports = tranquil_config::get().import.accepting; if !accepting_imports { return Err(ApiError::InvalidRequest( @@ -340,7 +336,7 @@ pub async fn import_repo( ); } } - Ok(EmptyResponse::ok().into_response()) + Ok(Json(EmptyResponse {})) } Err(ImportError::SizeLimitExceeded) => Err(ApiError::PayloadTooLarge(format!( "Import exceeds block limit of {}", diff --git a/crates/tranquil-api/src/repo/meta.rs b/crates/tranquil-api/src/repo/meta.rs index 6ced06d..1192fd2 100644 --- a/crates/tranquil-api/src/repo/meta.rs +++ b/crates/tranquil-api/src/repo/meta.rs @@ -1,3 +1,4 @@ +use crate::common; use axum::{ Json, extract::{Query, State}, @@ -5,7 +6,6 @@ use axum::{ }; use serde::Deserialize; use serde_json::json; -use tranquil_pds::api::error::ApiError; use tranquil_pds::state::AppState; use tranquil_pds::types::AtIdentifier; @@ -18,57 +18,22 @@ pub async fn describe_repo( State(state): State, Query(input): Query, ) -> Response { - let hostname_for_handles = tranquil_config::get().server.hostname_without_port(); - let user_row = if input.repo.is_did() { - let did: tranquil_pds::types::Did = match input.repo.as_str().parse() { - Ok(d) => d, - Err(_) => return ApiError::InvalidRequest("Invalid DID format".into()).into_response(), - }; - state - .user_repo - .get_by_did(&did) - .await - .map(|opt| opt.map(|r| (r.id, r.handle, r.did))) - } else { - let repo_str = input.repo.as_str(); - let handle_str = if !repo_str.contains('.') { - format!("{}.{}", repo_str, hostname_for_handles) - } else { - repo_str.to_string() - }; - let handle: tranquil_pds::types::Handle = match handle_str.parse() { - Ok(h) => h, - Err(_) => { - return ApiError::InvalidRequest("Invalid handle format".into()).into_response(); - } - }; - state - .user_repo - .get_by_handle(&handle) - .await - .map(|opt| opt.map(|r| (r.id, r.handle, r.did))) - }; - let (user_id, handle, did) = match user_row { - Ok(Some((id, handle, did))) => (id, handle, did), - Ok(None) => { - return ApiError::RepoNotFound(Some("Repo not found".into())).into_response(); - } - Err(_) => { - return ApiError::InternalError(None).into_response(); - } + let resolved = match common::resolve_repo(state.user_repo.as_ref(), &input.repo).await { + Ok(r) => r, + Err(e) => return e.into_response(), }; let collections = state .repo_repo - .list_collections(user_id) + .list_collections(resolved.user_id) .await .unwrap_or_default(); let did_doc = json!({ - "id": did, - "alsoKnownAs": [format!("at://{}", handle)] + "id": resolved.did, + "alsoKnownAs": [format!("at://{}", resolved.handle)] }); Json(json!({ - "handle": handle, - "did": did, + "handle": resolved.handle, + "did": resolved.did, "didDoc": did_doc, "collections": collections, "handleIsCorrect": true diff --git a/crates/tranquil-api/src/repo/record/batch.rs b/crates/tranquil-api/src/repo/record/batch.rs index fecc5b2..7d98ef4 100644 --- a/crates/tranquil-api/src/repo/record/batch.rs +++ b/crates/tranquil-api/src/repo/record/batch.rs @@ -1,27 +1,20 @@ use super::validation::validate_record_with_status; use super::validation_mode::{ValidationMode, deserialize_validation_mode}; -use crate::repo::record::utils::{CommitParams, RecordOp, commit_and_log, extract_blob_cids}; -use axum::{ - Json, - extract::State, - http::StatusCode, - response::{IntoResponse, Response}, -}; -use cid::Cid; -use jacquard_repo::{commit::Commit, mst::Mst, storage::BlockStore}; +use crate::repo::record::write::CommitInfo; +use axum::{Json, extract::State}; +use jacquard_repo::{mst::Mst, storage::BlockStore}; use serde::{Deserialize, Serialize}; use serde_json::json; -use std::str::FromStr; -use std::sync::Arc; use tracing::info; -use tranquil_pds::api::error::ApiError; +use tranquil_pds::api::error::{ApiError, DbResultExt}; use tranquil_pds::auth::{ Active, Auth, WriteOpKind, require_not_migrated, require_verified_or_delegated, verify_batch_write_scopes, }; -use tranquil_pds::cid_types::CommitCid; -use tranquil_pds::delegation::DelegationActionType; use tranquil_pds::repo::tracking::TrackingBlockStore; +use tranquil_pds::repo_ops::{ + FinalizeParams, RecordOp, begin_repo_write, extract_blob_cids, finalize_repo_write, +}; use tranquil_pds::state::AppState; use tranquil_pds::types::{AtIdentifier, AtUri, Did, Nsid, Rkey}; use tranquil_pds::validation::ValidationStatus; @@ -42,7 +35,7 @@ async fn process_single_write( did: &Did, validate: ValidationMode, tracking_store: &TrackingBlockStore, -) -> Result { +) -> Result { let WriteAccumulator { mst, mut results, @@ -60,32 +53,31 @@ async fn process_single_write( let validation_status = if validate.should_skip() { None } else { - match validate_record_with_status( - value, - collection, - rkey.as_ref(), - validate.requires_lexicon(), + Some( + validate_record_with_status( + value, + collection, + rkey.as_ref(), + validate.requires_lexicon(), + ) + .await?, ) - .await - { - 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 = tranquil_pds::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 record_bytes = serde_ipld_dagcbor::to_vec(&record_ipld) + .map_err(|_| ApiError::InvalidRecord("Failed to serialize record".into()))?; + let record_cid = tracking_store + .put(&record_bytes) + .await + .map_err(|_| ApiError::InternalError(Some("Failed to store record".into())))?; 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 new_mst = mst + .add(&key, record_cid) + .await + .map_err(|_| ApiError::InternalError(Some("Failed to add to MST".into())))?; let uri = AtUri::from_parts(did, collection, &rkey); results.push(WriteResult::CreateResult { uri, @@ -113,32 +105,31 @@ async fn process_single_write( let validation_status = if validate.should_skip() { None } else { - match validate_record_with_status( - value, - collection, - Some(rkey), - validate.requires_lexicon(), + Some( + validate_record_with_status( + value, + collection, + Some(rkey), + validate.requires_lexicon(), + ) + .await?, ) - .await - { - Ok(status) => Some(status), - Err(err_response) => return Err(*err_response), - } }; all_blob_cids.extend(extract_blob_cids(value)); let record_ipld = tranquil_pds::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 record_bytes = serde_ipld_dagcbor::to_vec(&record_ipld) + .map_err(|_| ApiError::InvalidRecord("Failed to serialize record".into()))?; + let record_cid = tracking_store + .put(&record_bytes) + .await + .map_err(|_| ApiError::InternalError(Some("Failed to store record".into())))?; 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 new_mst = mst + .update(&key, record_cid) + .await + .map_err(|_| ApiError::InternalError(Some("Failed to update MST".into())))?; let uri = AtUri::from_parts(did, collection, rkey); results.push(WriteResult::UpdateResult { uri, @@ -163,9 +154,10 @@ async fn process_single_write( 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() - })?; + let new_mst = mst + .delete(&key) + .await + .map_err(|_| ApiError::InternalError(Some("Failed to delete from MST".into())))?; results.push(WriteResult::DeleteResult {}); ops.push(RecordOp::Delete { collection: collection.clone(), @@ -189,7 +181,7 @@ async fn process_writes( did: &Did, validate: ValidationMode, tracking_store: &TrackingBlockStore, -) -> Result { +) -> Result { use futures::stream::{self, TryStreamExt}; let initial_acc = WriteAccumulator { mst: initial_mst, @@ -198,7 +190,7 @@ async fn process_writes( modified_keys: Vec::new(), all_blob_cids: Vec::new(), }; - stream::iter(writes.iter().map(Ok::<_, Response>)) + stream::iter(writes.iter().map(Ok::<_, ApiError>)) .try_fold(initial_acc, |acc, write| async move { process_single_write(write, acc, did, validate, tracking_store).await }) @@ -261,17 +253,11 @@ pub struct ApplyWritesOutput { pub results: Vec, } -#[derive(Serialize)] -pub struct CommitInfo { - pub cid: String, - pub rev: String, -} - pub async fn apply_writes( State(state): State, auth: Auth, Json(input): Json, -) -> Result { +) -> Result, ApiError> { info!( "apply_writes called: repo={}, writes={}", input.repo, @@ -288,7 +274,7 @@ pub async fn apply_writes( ))); } - let batch_proof = match verify_batch_write_scopes( + let batch_proof = verify_batch_write_scopes( &auth, &auth, &input.writes, @@ -302,10 +288,7 @@ pub async fn apply_writes( WriteOp::Update { .. } => WriteOpKind::Update, WriteOp::Delete { .. } => WriteOpKind::Delete, }, - ) { - Ok(proof) => proof, - Err(e) => return Ok(e.into_response()), - }; + )?; let principal_did = batch_proof.principal_did(); let controller_did = batch_proof.controller_did().map(|c| c.into_did()); @@ -317,140 +300,35 @@ pub async fn apply_writes( } let did = principal_did.into_did(); - if let Err(e) = require_not_migrated(&state, &did).await { - return Ok(e); - } - if let Err(e) = require_verified_or_delegated(&state, batch_proof.user()).await { - return Ok(e); - } + require_not_migrated(&state, &did).await?; + require_verified_or_delegated(&state, batch_proof.user()).await?; let user_id: uuid::Uuid = state .user_repo .get_id_by_did(&did) .await - .ok() - .flatten() - .ok_or_else(|| ApiError::InternalError(Some("User not found".into())))?; + .log_db_err("fetching user for batch write")? + .ok_or(ApiError::InternalError(Some("User not found".into())))?; - let _write_lock = state.repo_write_locks.lock(user_id).await; + let (ctx, mst) = begin_repo_write(&state, user_id, input.swap_commit.as_deref()).await?; - let root_cid_str = state - .repo_repo - .get_repo_root_cid_by_user_id(user_id) - .await - .ok() - .flatten() - .ok_or_else(|| ApiError::InternalError(Some("Repo root not found".into())))?; - let current_root_cid = CommitCid::from_str(&root_cid_str) - .map_err(|_| ApiError::InternalError(Some("Invalid repo root CID".into())))?; - if let Some(swap_commit) = &input.swap_commit - && CommitCid::from_str(swap_commit).ok().as_ref() != Some(¤t_root_cid) - { - return Err(ApiError::InvalidSwap(Some("Repo has been modified".into()))); - } - let tracking_store = TrackingBlockStore::new(state.block_store.clone()); - let commit_bytes = tracking_store - .get(current_root_cid.as_cid()) - .await - .ok() - .flatten() - .ok_or_else(|| ApiError::InternalError(Some("Commit block not found".into())))?; - let commit = Commit::from_cbor(&commit_bytes) - .map_err(|_| ApiError::InternalError(Some("Failed to parse commit".into())))?; - let original_mst = Mst::load(Arc::new(tracking_store.clone()), commit.data, None); - let initial_mst = Mst::load(Arc::new(tracking_store.clone()), commit.data, None); let WriteAccumulator { - mst, + mst: final_mst, results, ops, modified_keys, all_blob_cids, - } = match process_writes( + } = process_writes( &input.writes, - initial_mst, + mst, &did, input.validate, - &tracking_store, + &ctx.tracking_store, ) - .await - { - Ok(acc) => acc, - Err(response) => return Ok(response), - }; - let new_mst_root = mst - .persist() - .await - .map_err(|_| ApiError::InternalError(Some("Failed to persist MST".into())))?; - let (new_mst_blocks, old_mst_blocks) = { - let mut new_blocks = std::collections::BTreeMap::new(); - let mut old_blocks = std::collections::BTreeMap::new(); - for key in &modified_keys { - mst.blocks_for_path(key, &mut new_blocks) - .await - .map_err(|_| { - ApiError::InternalError(Some("Failed to get new MST blocks for path".into())) - })?; - original_mst - .blocks_for_path(key, &mut old_blocks) - .await - .map_err(|_| { - ApiError::InternalError(Some("Failed to get old MST blocks for path".into())) - })?; - } - (new_blocks, old_blocks) - }; - 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), .. - } - | RecordOp::Delete { - prev: Some(cid), .. - } => Some(*cid), - _ => None, - }); - let obsolete_cids: Vec = std::iter::once(current_root_cid.into_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 { - did: &did, - user_id, - current_root_cid: Some(current_root_cid.into_cid()), - prev_data_cid: Some(commit.data), - new_mst_root, - ops, - blocks_cids: &written_cids_str, - blobs: &all_blob_cids, - obsolete_cids, - }, - ) - .await - { - Ok(res) => res, - Err(e) => return Err(ApiError::from(e)), - }; + .await?; - if let Some(ref controller) = controller_did { - let write_summary: Vec = input + let write_summary: Option = controller_did.as_ref().map(|_| { + let writes: Vec = input .writes .iter() .map(|w| match w { @@ -475,34 +353,34 @@ pub async fn apply_writes( }), }) .collect(); + json!({ + "action": "apply_writes", + "count": input.writes.len(), + "writes": writes + }) + }); - let _ = state - .delegation_repo - .log_delegation_action( - &did, - controller, - Some(controller), - DelegationActionType::RepoWrite, - Some(json!({ - "action": "apply_writes", - "count": input.writes.len(), - "writes": write_summary - })), - None, - None, - ) - .await; - } - - Ok(( - StatusCode::OK, - Json(ApplyWritesOutput { - commit: CommitInfo { - cid: commit_res.commit_cid.to_string(), - rev: commit_res.rev, - }, - results, - }), + let commit_result = finalize_repo_write( + &state, + ctx, + final_mst, + FinalizeParams { + did: &did, + user_id, + controller_did: controller_did.as_ref(), + delegation_detail: write_summary, + ops, + modified_keys: &modified_keys, + blob_cids: &all_blob_cids, + }, ) - .into_response()) + .await?; + + Ok(Json(ApplyWritesOutput { + commit: CommitInfo { + cid: commit_result.commit_cid.to_string(), + rev: commit_result.rev, + }, + results, + })) } diff --git a/crates/tranquil-api/src/repo/record/delete.rs b/crates/tranquil-api/src/repo/record/delete.rs index 4f6a366..0537fa6 100644 --- a/crates/tranquil-api/src/repo/record/delete.rs +++ b/crates/tranquil-api/src/repo/record/delete.rs @@ -1,13 +1,5 @@ -use crate::repo::record::utils::{ - CommitError, CommitParams, RecordOp, commit_and_log, get_current_root_cid, -}; use crate::repo::record::write::{CommitInfo, prepare_repo_write}; -use axum::{ - Json, - extract::State, - http::StatusCode, - response::{IntoResponse, Response}, -}; +use axum::{Json, extract::State}; use cid::Cid; use jacquard_repo::{commit::Commit, mst::Mst, storage::BlockStore}; use serde::{Deserialize, Serialize}; @@ -17,11 +9,12 @@ use std::sync::Arc; use tracing::error; use tranquil_pds::api::error::ApiError; use tranquil_pds::auth::{Active, Auth, VerifyScope}; -use tranquil_pds::cid_types::CommitCid; -use tranquil_pds::delegation::DelegationActionType; use tranquil_pds::repo::tracking::TrackingBlockStore; +use tranquil_pds::repo_ops::{ + CommitError, FinalizeParams, RecordOp, begin_repo_write, finalize_repo_write, +}; use tranquil_pds::state::AppState; -use tranquil_pds::types::{AtIdentifier, AtUri, Nsid, Rkey}; +use tranquil_pds::types::{AtIdentifier, AtUri, Did, Nsid, Rkey}; #[derive(Deserialize)] pub struct DeleteRecordInput { @@ -45,168 +38,66 @@ pub async fn delete_record( State(state): State, auth: Auth, Json(input): Json, -) -> Result { - let scope_proof = match auth.verify_repo_delete(&input.collection) { - Ok(proof) => proof, - Err(e) => return Ok(e.into_response()), - }; - - let repo_auth = match prepare_repo_write(&state, &scope_proof, &input.repo).await { - Ok(res) => res, - Err(err_res) => return Ok(err_res), - }; - +) -> Result, ApiError> { + let scope_proof = auth.verify_repo_delete(&input.collection)?; + let repo_auth = prepare_repo_write(&state, &scope_proof, &input.repo).await?; let did = repo_auth.did; let user_id = repo_auth.user_id; let controller_did = repo_auth.controller_did; - let _write_lock = state.repo_write_locks.lock(user_id).await; - let current_root_cid = get_current_root_cid(&state, user_id).await?; + let (ctx, mst) = begin_repo_write(&state, user_id, input.swap_commit.as_deref()).await?; - if let Some(swap_commit) = &input.swap_commit - && CommitCid::from_str(swap_commit).ok().as_ref() != Some(¤t_root_cid) - { - return Ok(ApiError::InvalidSwap(Some("Repo has been modified".into())).into_response()); - } - let tracking_store = TrackingBlockStore::new(state.block_store.clone()); - let commit_bytes = match tracking_store.get(current_root_cid.as_cid()).await { - Ok(Some(b)) => b, - _ => { - return Ok( - ApiError::InternalError(Some("Commit block not found".into())).into_response(), - ); - } - }; - let commit = match Commit::from_cbor(&commit_bytes) { - Ok(c) => c, - _ => { - return Ok( - ApiError::InternalError(Some("Failed to parse commit".into())).into_response(), - ); - } - }; - let mst = Mst::load(Arc::new(tracking_store.clone()), commit.data, None); let key = format!("{}/{}", input.collection, input.rkey); + if let Some(swap_record_str) = &input.swap_record { let expected_cid = Cid::from_str(swap_record_str).ok(); let actual_cid = mst.get(&key).await.ok().flatten(); if expected_cid != actual_cid { - return Ok(ApiError::InvalidSwap(Some( + return Err(ApiError::InvalidSwap(Some( "Record has been modified or does not exist".into(), - )) - .into_response()); + ))); } } + let prev_record_cid = mst.get(&key).await.ok().flatten(); if prev_record_cid.is_none() { - return Ok((StatusCode::OK, Json(DeleteRecordOutput { commit: None })).into_response()); + return Ok(Json(DeleteRecordOutput { commit: None })); } - let new_mst = match mst.delete(&key).await { - Ok(m) => m, - Err(e) => { - error!("Failed to delete from MST: {:?}", e); - return Ok(ApiError::InternalError(Some(format!( - "Failed to delete from MST: {:?}", - e - ))) - .into_response()); - } - }; - let new_mst_root = match new_mst.persist().await { - Ok(c) => c, - Err(e) => { - error!("Failed to persist MST: {:?}", e); - return Ok( - ApiError::InternalError(Some("Failed to persist MST".into())).into_response(), - ); - } - }; - let collection_for_audit = input.collection.to_string(); - let rkey_for_audit = input.rkey.to_string(); + + let new_mst = mst.delete(&key).await.map_err(|e| { + error!("Failed to delete from MST: {:?}", e); + ApiError::InternalError(Some("Failed to delete from MST".into())) + })?; + let op = RecordOp::Delete { collection: input.collection.clone(), rkey: input.rkey.clone(), prev: prev_record_cid, }; - 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 new_mst_blocks) - .await - .is_err() - { - return Ok( - ApiError::InternalError(Some("Failed to get new MST blocks for path".into())) - .into_response(), - ); - } - if mst - .blocks_for_path(&key, &mut old_mst_blocks) - .await - .is_err() - { - return Ok( - 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()))); - 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.into_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( + + let modified_keys = [key]; + + let commit_result = finalize_repo_write( &state, - CommitParams { + ctx, + new_mst, + FinalizeParams { did: &did, user_id, - current_root_cid: Some(current_root_cid.into_cid()), - prev_data_cid: Some(commit.data), - new_mst_root, + controller_did: controller_did.as_ref(), + delegation_detail: controller_did.as_ref().map(|_| { + json!({ + "action": "delete", + "collection": input.collection, + "rkey": input.rkey + }) + }), ops: vec![op], - blocks_cids: &written_cids_str, - blobs: &[], - obsolete_cids, + modified_keys: &modified_keys, + blob_cids: &[], }, ) - .await - { - Ok(res) => res, - Err(e) => return Ok(ApiError::from(e).into_response()), - }; - - if let Some(ref controller) = controller_did { - let _ = state - .delegation_repo - .log_delegation_action( - &did, - controller, - Some(controller), - DelegationActionType::RepoWrite, - Some(json!({ - "action": "delete", - "collection": collection_for_audit, - "rkey": rkey_for_audit - })), - None, - None, - ) - .await; - } + .await?; let deleted_uri = AtUri::from_parts(&did, &input.collection, &input.rkey); if let Err(e) = state @@ -217,19 +108,14 @@ pub async fn delete_record( error!("Failed to remove backlinks for {}: {}", deleted_uri, e); } - Ok(( - StatusCode::OK, - Json(DeleteRecordOutput { - commit: Some(CommitInfo { - cid: commit_result.commit_cid.to_string(), - rev: commit_result.rev, - }), + Ok(Json(DeleteRecordOutput { + commit: Some(CommitInfo { + cid: commit_result.commit_cid.to_string(), + rev: commit_result.rev, }), - ) - .into_response()) + })) } -use tranquil_pds::types::Did; use uuid::Uuid; pub async fn delete_record_internal( @@ -239,6 +125,8 @@ pub async fn delete_record_internal( collection: &Nsid, rkey: &Rkey, ) -> Result<(), CommitError> { + use tranquil_pds::repo_ops::{CommitParams, RecordOp, commit_and_log}; + let _write_lock = state.repo_write_locks.lock(user_id).await; let root_cid_str = state @@ -303,19 +191,6 @@ pub async fn delete_record_internal( .await .map_err(|e| CommitError::MstOperationFailed(format!("{:?}", e)))?; - 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 @@ -326,6 +201,19 @@ pub async fn delete_record_internal( .chain(std::iter::once(prev_cid)) .collect(); + let mut relevant_blocks = new_mst_blocks; + relevant_blocks.extend(old_mst_blocks); + + 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(ToString::to_string).collect(); + commit_and_log( state, CommitParams { diff --git a/crates/tranquil-api/src/server/mod.rs b/crates/tranquil-api/src/server/mod.rs index 7dbc6cc..015f51e 100644 --- a/crates/tranquil-api/src/server/mod.rs +++ b/crates/tranquil-api/src/server/mod.rs @@ -23,9 +23,8 @@ pub use account_status::{ }; pub use app_password::{create_app_password, list_app_passwords, revoke_app_password}; pub use email::{ - authorize_email_update, check_channel_verified, check_email_in_use, - check_email_update_status, check_email_verified, confirm_email, request_email_update, - update_email, + authorize_email_update, check_channel_verified, check_email_in_use, check_email_update_status, + check_email_verified, confirm_email, request_email_update, update_email, }; pub use invite::{create_invite_code, create_invite_codes, get_account_invite_codes}; pub use logo::get_logo; @@ -44,9 +43,8 @@ pub use password::{ set_password, }; pub use reauth::{ - check_legacy_session_mfa, check_reauth_required, get_reauth_status, - legacy_mfa_required_response, reauth_passkey_finish, reauth_passkey_start, reauth_password, - reauth_required_response, reauth_totp, update_mfa_verified, + check_legacy_session_mfa, check_reauth_required, get_reauth_status, reauth_passkey_finish, + reauth_passkey_start, reauth_password, reauth_totp, update_mfa_verified, }; pub use service_auth::get_service_auth; pub use session::{ @@ -64,4 +62,7 @@ pub use trusted_devices::{ trust_device, update_trusted_device, }; pub use verify_email::{resend_migration_verification, verify_migration_email}; -pub use verify_token::{VerifyTokenInput, VerifyTokenOutput, verify_token, verify_token_internal}; +pub use verify_token::{ + VerifyTokenInput, VerifyTokenOutput, confirm_channel_verification, verify_token, + verify_token_internal, +}; diff --git a/crates/tranquil-api/src/server/service_auth.rs b/crates/tranquil-api/src/server/service_auth.rs index 995b62c..2d3c952 100644 --- a/crates/tranquil-api/src/server/service_auth.rs +++ b/crates/tranquil-api/src/server/service_auth.rs @@ -112,7 +112,7 @@ pub async fn get_service_auth( }; let lxm = params.lxm.as_ref(); - let lxm_for_token = lxm.map_or("*", |n| n.as_str()); + let lxm_for_token = lxm.map_or("*", |v| v.as_str()); if let Some(method) = lxm { if let Err(e) = tranquil_pds::auth::scope_check::check_rpc_scope( @@ -121,7 +121,7 @@ pub async fn get_service_auth( params.aud.as_str(), method.as_str(), ) { - return e; + return e.into_response(); } } else if auth.is_oauth() { let permissions = auth.permissions(); diff --git a/crates/tranquil-api/src/server/verify_email.rs b/crates/tranquil-api/src/server/verify_email.rs index 72f8c53..105fa00 100644 --- a/crates/tranquil-api/src/server/verify_email.rs +++ b/crates/tranquil-api/src/server/verify_email.rs @@ -74,27 +74,14 @@ pub async fn resend_migration_verification( return Ok(Json(ResendMigrationVerificationOutput { sent: true })); } - let hostname = &tranquil_config::get().server.hostname; - let token = tranquil_pds::auth::verification_token::generate_migration_token( + crate::identity::provision::enqueue_migration_verification( + &state, + user.id, &user.did, channel, &identifier, - ); - let formatted_token = tranquil_pds::auth::verification_token::format_token_for_display(&token); - - if let Err(e) = tranquil_pds::comms::comms_repo::enqueue_migration_verification( - state.user_repo.as_ref(), - state.infra_repo.as_ref(), - user.id, - channel, - &identifier, - &formatted_token, - hostname, ) - .await - { - warn!(error = ?e, channel = ?channel, "Failed to enqueue migration verification"); - } + .await; info!(did = %user.did, channel = ?channel, "Resent migration verification"); diff --git a/crates/tranquil-api/src/temp.rs b/crates/tranquil-api/src/temp.rs index ecb025c..4e5fd98 100644 --- a/crates/tranquil-api/src/temp.rs +++ b/crates/tranquil-api/src/temp.rs @@ -6,6 +6,7 @@ use axum::{ use cid::Cid; use jacquard_repo::storage::BlockStore; use serde::{Deserialize, Serialize}; +use serde_json::Value; use std::str::FromStr; use tranquil_pds::api::error::ApiError; use tranquil_pds::auth::{Active, Auth, Permissive}; @@ -51,7 +52,7 @@ pub async fn dereference_scope( State(state): State, _auth: Auth, Json(input): Json, -) -> Result { +) -> Result, ApiError> { let scope_parts: Vec<&str> = input.scope.split_whitespace().collect(); let mut resolved_scopes: Vec = Vec::new(); @@ -96,7 +97,7 @@ pub async fn dereference_scope( } }; - if let Some(scope_value) = scope_record.get("scope").and_then(|v| v.as_str()) { + if let Some(scope_value) = scope_record.get("scope").and_then(Value::as_str) { let _ = state .cache .set( @@ -118,6 +119,5 @@ pub async fn dereference_scope( Ok(Json(DereferenceScopeOutput { scope: resolved_scopes.join(" "), - }) - .into_response()) + })) }