mirrors/tranquil-pds
1
0
Fork 0
mirror of https://tangled.org/tranquil.farm/tranquil-pds synced 2026-09-11 04:36:05 +00:00
Code Issues Packages Projects Releases Wiki Activity

Three-quarter-done in-house migration flow

Browse Source
This commit is contained in:
lewis
2025-12-27 22:52:53 +02:00
parent 77a67f3554
commit f6d7d65ef8
36 changed files with 5940 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/api/identity/did.rs
+1
View File
@@ -449,6 +449,7 @@ pub struct VerificationMethods {
#[derive(serde::Serialize)]
#[serde(rename_all = "camelCase")]
pub struct Services {
#[serde(rename = "atproto_pds")]
pub atproto_pds: AtprotoPds,
}
src/api/repo/blob.rs
+1 -1
View File
@@ -295,7 +295,7 @@ pub async fn list_missing_blobs(
.into_response();
}
};
let auth_user = match crate::auth::validate_bearer_token(&state.db, &token).await {
let auth_user = match crate::auth::validate_bearer_token_allow_deactivated(&state.db, &token).await {
Ok(user) => user,
Err(_) => {
return (
src/api/repo/import.rs
+4 -2
View File
@@ -318,8 +318,10 @@ pub async fn import_repo(
records.len(),
did
);
if let Err(e) = sequence_import_event(&state, did, &root.to_string()).await {
warn!("Failed to sequence import event: {:?}", e);
if !is_migration {
if let Err(e) = sequence_import_event(&state, did, &root.to_string()).await {
warn!("Failed to sequence import event: {:?}", e);
}
}
(StatusCode::OK, Json(json!({}))).into_response()
}
src/api/repo/record/batch.rs
+5 -1
View File
@@ -1,6 +1,6 @@
use super::validation::validate_record;
use super::write::has_verified_comms_channel;
use crate::api::repo::record::utils::{CommitParams, RecordOp, commit_and_log};
use crate::api::repo::record::utils::{CommitParams, RecordOp, commit_and_log, extract_blob_cids};
use crate::delegation::{self, DelegationActionType};
use crate::repo::tracking::TrackingBlockStore;
use crate::state::AppState;
@@ -295,6 +295,7 @@ pub async fn apply_writes(
let mut results: Vec<WriteResult> = Vec::new();
let mut ops: Vec<RecordOp> = Vec::new();
let mut modified_keys: Vec<String> = Vec::new();
let mut all_blob_cids: Vec<String> = Vec::new();
for write in &input.writes {
match write {
WriteOp::Create {
@@ -307,6 +308,7 @@ pub async fn apply_writes(
{
return *err_response;
}
all_blob_cids.extend(extract_blob_cids(value));
let rkey = rkey
.clone()
.unwrap_or_else(|| Tid::now(LimitedU32::MIN).to_string());
@@ -359,6 +361,7 @@ pub async fn apply_writes(
{
return *err_response;
}
all_blob_cids.extend(extract_blob_cids(value));
let mut record_bytes = Vec::new();
if serde_ipld_dagcbor::to_writer(&mut record_bytes, value).is_err() {
return (StatusCode::BAD_REQUEST, Json(json!({"error": "InvalidRecord", "message": "Failed to serialize record"}))).into_response();
@@ -468,6 +471,7 @@ pub async fn apply_writes(
new_mst_root,
ops,
blocks_cids: &written_cids_str,
blobs: &all_blob_cids,
},
)
.await
src/api/repo/record/delete.rs
+1
View File
@@ -168,6 +168,7 @@ pub async fn delete_record(
new_mst_root,
ops: vec![op],
blocks_cids: &written_cids_str,
blobs: &[],
},
)
.await
src/api/repo/record/read.rs
+38 -3
View File
@@ -6,14 +6,47 @@ use axum::{
http::{HeaderMap, StatusCode},
response::{IntoResponse, Response},
};
use base64::Engine;
use cid::Cid;
use ipld_core::ipld::Ipld;
use jacquard_repo::storage::BlockStore;
use serde::{Deserialize, Serialize};
use serde_json::json;
use serde_json::{json, Map, Value};
use std::collections::HashMap;
use std::str::FromStr;
use tracing::{error, info};
fn ipld_to_json(ipld: Ipld) -> Value {
match ipld {
Ipld::Null => Value::Null,
Ipld::Bool(b) => Value::Bool(b),
Ipld::Integer(i) => {
if let Ok(n) = i64::try_from(i) {
Value::Number(n.into())
} else {
Value::String(i.to_string())
}
}
Ipld::Float(f) => serde_json::Number::from_f64(f)
.map(Value::Number)
.unwrap_or(Value::Null),
Ipld::String(s) => Value::String(s),
Ipld::Bytes(b) => {
let encoded = base64::engine::general_purpose::STANDARD.encode(&b);
json!({ "$bytes": encoded })
}
Ipld::List(arr) => Value::Array(arr.into_iter().map(ipld_to_json).collect()),
Ipld::Map(map) => {
let obj: Map<String, Value> = map
.into_iter()
.map(|(k, v)| (k, ipld_to_json(v)))
.collect();
Value::Object(obj)
}
Ipld::Link(cid) => json!({ "$link": cid.to_string() }),
}
}
#[derive(Deserialize)]
pub struct GetRecordInput {
pub repo: String,
@@ -163,7 +196,7 @@ pub async fn get_record(
.into_response();
}
};
let value: serde_json::Value = match serde_ipld_dagcbor::from_slice(&block) {
let ipld: Ipld = match serde_ipld_dagcbor::from_slice(&block) {
Ok(v) => v,
Err(e) => {
error!("Failed to deserialize record: {:?}", e);
@@ -174,6 +207,7 @@ pub async fn get_record(
.into_response();
}
};
let value = ipld_to_json(ipld);
Json(json!({
"uri": format!("at://{}/{}/{}", input.repo, input.collection, input.rkey),
"cid": record_cid_str,
@@ -323,8 +357,9 @@ pub async fn list_records(
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(value) = serde_ipld_dagcbor::from_slice::<serde_json::Value>(&block)
&& let Ok(ipld) = serde_ipld_dagcbor::from_slice::<Ipld>(&block)
{
let value = ipld_to_json(ipld);
records.push(json!({
"uri": format!("at://{}/{}/{}", input.repo, input.collection, rkey),
"cid": cid_str,
src/api/repo/record/utils.rs
+35 -2
View File
@@ -5,10 +5,39 @@ use jacquard::types::{integer::LimitedU32, string::Tid};
use jacquard_repo::commit::Commit;
use jacquard_repo::storage::BlockStore;
use k256::ecdsa::SigningKey;
use serde_json::json;
use serde_json::{json, Value};
use std::str::FromStr;
use uuid::Uuid;
pub fn extract_blob_cids(record: &Value) -> Vec<String> {
let mut blobs = Vec::new();
extract_blob_cids_recursive(record, &mut blobs);
blobs
}
fn extract_blob_cids_recursive(value: &Value, blobs: &mut Vec<String>) {
match value {
Value::Object(map) => {
if map.get("$type").and_then(|v| v.as_str()) == Some("blob") {
if let Some(ref_obj) = map.get("ref") {
if let Some(link) = ref_obj.get("$link").and_then(|v| v.as_str()) {
blobs.push(link.to_string());
}
}
}
for v in map.values() {
extract_blob_cids_recursive(v, blobs);
}
}
Value::Array(arr) => {
for v in arr {
extract_blob_cids_recursive(v, blobs);
}
}
_ => {}
}
}
pub fn create_signed_commit(
did: &str,
data: Cid,
@@ -63,6 +92,7 @@ pub struct CommitParams<'a> {
pub new_mst_root: Cid,
pub ops: Vec<RecordOp>,
pub blocks_cids: &'a [String],
pub blobs: &'a [String],
}
pub async fn commit_and_log(
@@ -77,6 +107,7 @@ pub async fn commit_and_log(
new_mst_root,
ops,
blocks_cids,
blobs,
} = params;
let key_row = sqlx::query!(
"SELECT key_bytes, encryption_version FROM user_keys WHERE user_id = $1",
@@ -274,7 +305,7 @@ pub async fn commit_and_log(
new_root_cid.to_string(),
prev_cid_str,
json!(ops_json),
&[] as &[String],
blobs,
blocks_cids,
prev_data_cid_str,
)
@@ -368,6 +399,7 @@ pub async fn create_record_internal(
}
}
let written_cids_str: Vec<String> = written_cids.iter().map(|c| c.to_string()).collect();
let blob_cids = extract_blob_cids(record);
let result = commit_and_log(
state,
CommitParams {
@@ -378,6 +410,7 @@ pub async fn create_record_internal(
new_mst_root,
ops: vec![op],
blocks_cids: &written_cids_str,
blobs: &blob_cids,
},
)
.await?;
src/api/repo/record/write.rs
+5 -1
View File
@@ -1,5 +1,5 @@
use super::validation::validate_record;
use crate::api::repo::record::utils::{CommitParams, RecordOp, commit_and_log};
use crate::api::repo::record::utils::{CommitParams, RecordOp, commit_and_log, extract_blob_cids};
use crate::delegation::{self, DelegationActionType};
use crate::repo::tracking::TrackingBlockStore;
use crate::state::AppState;
@@ -334,6 +334,7 @@ pub async fn create_record(
.iter()
.map(|c| c.to_string())
.collect::<Vec<_>>();
let blob_cids = extract_blob_cids(&input.record);
if let Err(e) = commit_and_log(
&state,
CommitParams {
@@ -344,6 +345,7 @@ pub async fn create_record(
new_mst_root,
ops: vec![op],
blocks_cids: &written_cids_str,
blobs: &blob_cids,
},
)
.await
@@ -582,6 +584,7 @@ pub async fn put_record(
.map(|c| c.to_string())
.collect::<Vec<_>>();
let is_update = existing_cid.is_some();
let blob_cids = extract_blob_cids(&input.record);
if let Err(e) = commit_and_log(
&state,
CommitParams {
@@ -592,6 +595,7 @@ pub async fn put_record(
new_mst_root,
ops: vec![op],
blocks_cids: &written_cids_str,
blobs: &blob_cids,
},
)
.await
src/api/server/account_status.rs
+204 -7
View File
@@ -1,4 +1,5 @@
use crate::api::ApiError;
use crate::plc::PlcClient;
use crate::state::AppState;
use axum::{
Json,
@@ -8,6 +9,7 @@ use axum::{
};
use bcrypt::verify;
use chrono::{Duration, Utc};
use k256::ecdsa::SigningKey;
use serde::{Deserialize, Serialize};
use serde_json::json;
use tracing::{error, info, warn};
@@ -118,6 +120,185 @@ pub async fn check_account_status(
.into_response()
}
async fn assert_valid_did_document_for_service(
db: &sqlx::PgPool,
did: &str,
) -> Result<(), (StatusCode, Json<serde_json::Value>)> {
let hostname = std::env::var("PDS_HOSTNAME").unwrap_or_else(|_| "localhost".to_string());
let expected_endpoint = format!("https://{}", hostname);
if did.starts_with("did:plc:") {
let plc_client = PlcClient::new(None);
let mut last_error = None;
let mut doc_data = None;
for attempt in 0..5 {
if attempt > 0 {
let delay_ms = 500 * (1 << (attempt - 1));
info!(
"Waiting {}ms before retry {} for DID document validation ({})",
delay_ms, attempt, did
);
tokio::time::sleep(tokio::time::Duration::from_millis(delay_ms)).await;
}
match plc_client.get_document_data(did).await {
Ok(data) => {
let pds_endpoint = data
.get("services")
.and_then(|s| s.get("atproto_pds").or_else(|| s.get("atprotoPds")))
.and_then(|p| p.get("endpoint"))
.and_then(|e| e.as_str());
if pds_endpoint == Some(&expected_endpoint) {
doc_data = Some(data);
break;
} else {
info!(
"Attempt {}: DID {} has endpoint {:?}, expected {} - retrying",
attempt + 1,
did,
pds_endpoint,
expected_endpoint
);
last_error = Some(format!(
"DID document endpoint {:?} does not match expected {}",
pds_endpoint, expected_endpoint
));
}
}
Err(e) => {
warn!(
"Attempt {}: Failed to fetch PLC document for {}: {:?}",
attempt + 1,
did,
e
);
last_error = Some(format!("Could not resolve DID document: {}", e));
}
}
}
let doc_data = match doc_data {
Some(d) => d,
None => {
return Err((
StatusCode::BAD_REQUEST,
Json(json!({
"error": "InvalidRequest",
"message": last_error.unwrap_or_else(|| "DID document validation failed".to_string())
})),
));
}
};
let doc_signing_key = doc_data
.get("verificationMethods")
.and_then(|v| v.get("atproto"))
.and_then(|k| k.as_str());
let user_row = sqlx::query!(
"SELECT uk.key_bytes, uk.encryption_version FROM user_keys uk JOIN users u ON uk.user_id = u.id WHERE u.did = $1",
did
)
.fetch_optional(db)
.await
.map_err(|e| {
error!("Failed to fetch user key: {:?}", e);
(
StatusCode::INTERNAL_SERVER_ERROR,
Json(json!({"error": "InternalError"})),
)
})?;
if let Some(row) = user_row {
let key_bytes =
crate::config::decrypt_key(&row.key_bytes, row.encryption_version).map_err(|e| {
error!("Failed to decrypt user key: {}", e);
(
StatusCode::INTERNAL_SERVER_ERROR,
Json(json!({"error": "InternalError"})),
)
})?;
let signing_key = SigningKey::from_slice(&key_bytes).map_err(|e| {
error!("Failed to create signing key: {:?}", e);
(
StatusCode::INTERNAL_SERVER_ERROR,
Json(json!({"error": "InternalError"})),
)
})?;
let expected_did_key = crate::plc::signing_key_to_did_key(&signing_key);
if doc_signing_key != Some(&expected_did_key) {
warn!(
"DID {} has signing key {:?}, expected {}",
did, doc_signing_key, expected_did_key
);
return Err((
StatusCode::BAD_REQUEST,
Json(json!({
"error": "InvalidRequest",
"message": "DID document verification method does not match expected signing key"
})),
));
}
}
} else if did.starts_with("did:web:") {
let client = reqwest::Client::new();
let did_path = &did[8..];
let url = format!("https://{}/.well-known/did.json", did_path.replace(':', "/"));
let resp = client.get(&url).send().await.map_err(|e| {
warn!("Failed to fetch did:web document for {}: {:?}", did, e);
(
StatusCode::BAD_REQUEST,
Json(json!({
"error": "InvalidRequest",
"message": format!("Could not resolve DID document: {}", e)
})),
)
})?;
let doc: serde_json::Value = resp.json().await.map_err(|e| {
warn!("Failed to parse did:web document for {}: {:?}", did, e);
(
StatusCode::BAD_REQUEST,
Json(json!({
"error": "InvalidRequest",
"message": format!("Could not parse DID document: {}", e)
})),
)
})?;
let pds_endpoint = doc
.get("service")
.and_then(|s| s.as_array())
.and_then(|arr| {
arr.iter().find(|svc| {
svc.get("id").and_then(|id| id.as_str()) == Some("#atproto_pds")
|| svc.get("type").and_then(|t| t.as_str())
== Some("AtprotoPersonalDataServer")
})
})
.and_then(|svc| svc.get("serviceEndpoint"))
.and_then(|e| e.as_str());
if pds_endpoint != Some(&expected_endpoint) {
warn!(
"DID {} has endpoint {:?}, expected {}",
did, pds_endpoint, expected_endpoint
);
return Err((
StatusCode::BAD_REQUEST,
Json(json!({
"error": "InvalidRequest",
"message": "DID document atproto_pds service endpoint does not match PDS public url"
})),
));
}
}
Ok(())
}
pub async fn activate_account(
State(state): State<AppState>,
headers: axum::http::HeaderMap,
@@ -158,6 +339,15 @@ pub async fn activate_account(
}
let did = auth_user.did;
if let Err((status, json)) = assert_valid_did_document_for_service(&state.db, &did).await {
info!(
"activateAccount rejected for {}: DID document validation failed",
did
);
return (status, json).into_response();
}
let handle = sqlx::query_scalar!("SELECT handle FROM users WHERE did = $1", did)
.fetch_optional(&state.db)
.await
@@ -182,13 +372,20 @@ pub async fn activate_account(
{
warn!("Failed to sequence identity event for activation: {}", e);
}
if let Err(e) =
crate::api::repo::record::sequence_empty_commit_event(&state, &did).await
{
warn!(
"Failed to sequence empty commit event for activation: {}",
e
);
let repo_root = sqlx::query_scalar!(
"SELECT r.repo_root_cid FROM repos r JOIN users u ON r.user_id = u.id WHERE u.did = $1",
did
)
.fetch_optional(&state.db)
.await
.ok()
.flatten();
if let Some(root_cid) = repo_root {
if let Err(e) =
crate::api::repo::record::sequence_sync_event(&state, &did, &root_cid).await
{
warn!("Failed to sequence sync event for activation: {}", e);
}
}
(StatusCode::OK, Json(json!({}))).into_response()
}
src/api/server/migration.rs
+239
View File
@@ -0,0 +1,239 @@
use crate::api::ApiError;
use crate::state::AppState;
use axum::{
Json,
extract::State,
http::StatusCode,
response::{IntoResponse, Response},
};
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use serde_json::json;
#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
pub struct GetMigrationStatusOutput {
pub did: String,
pub did_type: String,
pub migrated: bool,
#[serde(skip_serializing_if = "Option::is_none")]
pub migrated_to_pds: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub migrated_at: Option<DateTime<Utc>>,
}
pub async fn get_migration_status(
State(state): State<AppState>,
headers: axum::http::HeaderMap,
) -> Response {
let extracted = match crate::auth::extract_auth_token_from_header(
headers.get("Authorization").and_then(|h| h.to_str().ok()),
) {
Some(t) => t,
None => return ApiError::AuthenticationRequired.into_response(),
};
let dpop_proof = headers.get("DPoP").and_then(|h| h.to_str().ok());
let http_uri = format!(
"https://{}/xrpc/com.tranquil.account.getMigrationStatus",
std::env::var("PDS_HOSTNAME").unwrap_or_else(|_| "localhost".to_string())
);
let auth_user = match crate::auth::validate_token_with_dpop(
&state.db,
&extracted.token,
extracted.is_dpop,
dpop_proof,
"GET",
&http_uri,
true,
)
.await
{
Ok(user) => user,
Err(e) => return ApiError::from(e).into_response(),
};
let user = match sqlx::query!(
"SELECT did, migrated_to_pds, migrated_at FROM users WHERE did = $1",
auth_user.did
)
.fetch_optional(&state.db)
.await
{
Ok(Some(row)) => row,
Ok(None) => return ApiError::AccountNotFound.into_response(),
Err(e) => {
tracing::error!("DB error getting migration status: {:?}", e);
return ApiError::InternalError.into_response();
}
};
let did_type = if user.did.starts_with("did:plc:") {
"plc"
} else if user.did.starts_with("did:web:") {
"web"
} else {
"unknown"
};
let migrated = user.migrated_to_pds.is_some();
(
StatusCode::OK,
Json(GetMigrationStatusOutput {
did: user.did,
did_type: did_type.to_string(),
migrated,
migrated_to_pds: user.migrated_to_pds,
migrated_at: user.migrated_at,
}),
)
.into_response()
}
#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct UpdateMigrationForwardingInput {
pub pds_url: String,
}
#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
pub struct UpdateMigrationForwardingOutput {
pub success: bool,
pub migrated_to_pds: String,
pub migrated_at: DateTime<Utc>,
}
pub async fn update_migration_forwarding(
State(state): State<AppState>,
headers: axum::http::HeaderMap,
Json(input): Json<UpdateMigrationForwardingInput>,
) -> Response {
let extracted = match crate::auth::extract_auth_token_from_header(
headers.get("Authorization").and_then(|h| h.to_str().ok()),
) {
Some(t) => t,
None => return ApiError::AuthenticationRequired.into_response(),
};
let dpop_proof = headers.get("DPoP").and_then(|h| h.to_str().ok());
let http_uri = format!(
"https://{}/xrpc/com.tranquil.account.updateMigrationForwarding",
std::env::var("PDS_HOSTNAME").unwrap_or_else(|_| "localhost".to_string())
);
let auth_user = match crate::auth::validate_token_with_dpop(
&state.db,
&extracted.token,
extracted.is_dpop,
dpop_proof,
"POST",
&http_uri,
true,
)
.await
{
Ok(user) => user,
Err(e) => return ApiError::from(e).into_response(),
};
if !auth_user.did.starts_with("did:web:") {
return (
StatusCode::BAD_REQUEST,
Json(json!({
"error": "InvalidRequest",
"message": "Migration forwarding is only available for did:web accounts. did:plc accounts use PLC directory for identity updates."
})),
)
.into_response();
}
let pds_url = input.pds_url.trim();
if pds_url.is_empty() {
return ApiError::InvalidRequest("pds_url is required".into()).into_response();
}
if !pds_url.starts_with("https://") {
return ApiError::InvalidRequest("pds_url must start with https://".into()).into_response();
}
let pds_url_clean = pds_url.trim_end_matches('/');
let now = Utc::now();
let result = sqlx::query!(
"UPDATE users SET migrated_to_pds = $1, migrated_at = $2 WHERE did = $3",
pds_url_clean,
now,
auth_user.did
)
.execute(&state.db)
.await;
match result {
Ok(_) => {
tracing::info!(
"Updated migration forwarding for {} to {}",
auth_user.did,
pds_url_clean
);
(
StatusCode::OK,
Json(UpdateMigrationForwardingOutput {
success: true,
migrated_to_pds: pds_url_clean.to_string(),
migrated_at: now,
}),
)
.into_response()
}
Err(e) => {
tracing::error!("DB error updating migration forwarding: {:?}", e);
ApiError::InternalError.into_response()
}
}
}
pub async fn clear_migration_forwarding(
State(state): State<AppState>,
headers: axum::http::HeaderMap,
) -> Response {
let extracted = match crate::auth::extract_auth_token_from_header(
headers.get("Authorization").and_then(|h| h.to_str().ok()),
) {
Some(t) => t,
None => return ApiError::AuthenticationRequired.into_response(),
};
let dpop_proof = headers.get("DPoP").and_then(|h| h.to_str().ok());
let http_uri = format!(
"https://{}/xrpc/com.tranquil.account.clearMigrationForwarding",
std::env::var("PDS_HOSTNAME").unwrap_or_else(|_| "localhost".to_string())
);
let auth_user = match crate::auth::validate_token_with_dpop(
&state.db,
&extracted.token,
extracted.is_dpop,
dpop_proof,
"POST",
&http_uri,
true,
)
.await
{
Ok(user) => user,
Err(e) => return ApiError::from(e).into_response(),
};
if !auth_user.did.starts_with("did:web:") {
return (
StatusCode::BAD_REQUEST,
Json(json!({
"error": "InvalidRequest",
"message": "Migration forwarding is only available for did:web accounts"
})),
)
.into_response();
}
let result = sqlx::query!(
"UPDATE users SET migrated_to_pds = NULL, migrated_at = NULL WHERE did = $1",
auth_user.did
)
.execute(&state.db)
.await;
match result {
Ok(_) => {
tracing::info!("Cleared migration forwarding for {}", auth_user.did);
(StatusCode::OK, Json(json!({ "success": true }))).into_response()
}
Err(e) => {
tracing::error!("DB error clearing migration forwarding: {:?}", e);
ApiError::InternalError.into_response()
}
}
}
src/api/server/mod.rs
+4
View File
@@ -4,6 +4,7 @@ pub mod email;
pub mod invite;
pub mod logo;
pub mod meta;
pub mod migration;
pub mod passkey_account;
pub mod passkeys;
pub mod password;
@@ -56,5 +57,8 @@ pub use trusted_devices::{
extend_device_trust, is_device_trusted, list_trusted_devices, revoke_trusted_device,
trust_device, update_trusted_device,
};
pub use migration::{
clear_migration_forwarding, get_migration_status, update_migration_forwarding,
};
pub use verify_email::{resend_migration_verification, verify_migration_email};
pub use verify_token::{VerifyTokenInput, VerifyTokenOutput, verify_token, verify_token_internal};
src/lib.rs
+12
View File
@@ -280,6 +280,18 @@ pub fn app(state: AppState) -> Router {
"/xrpc/com.tranquil.account.recoverPasskeyAccount",
post(api::server::recover_passkey_account),
)
.route(
"/xrpc/com.tranquil.account.getMigrationStatus",
get(api::server::get_migration_status),
)
.route(
"/xrpc/com.tranquil.account.updateMigrationForwarding",
post(api::server::update_migration_forwarding),
)
.route(
"/xrpc/com.tranquil.account.clearMigrationForwarding",
post(api::server::clear_migration_forwarding),
)
.route(
"/xrpc/com.atproto.server.requestEmailUpdate",
post(api::server::request_email_update),
src/sync/import.rs
+77 -54
View File
@@ -163,68 +163,91 @@ pub fn walk_mst(
root_cid: &Cid,
) -> Result<Vec<ImportedRecord>, ImportError> {
let mut records = Vec::new();
let mut stack = vec![*root_cid];
let mut visited = std::collections::HashSet::new();
while let Some(cid) = stack.pop() {
if visited.contains(&cid) {
continue;
walk_mst_node(blocks, root_cid, &[], &mut records)?;
Ok(records)
}
fn walk_mst_node(
blocks: &HashMap<Cid, Bytes>,
cid: &Cid,
prev_key: &[u8],
records: &mut Vec<ImportedRecord>,
) -> Result<(), ImportError> {
let block = blocks
.get(cid)
.ok_or_else(|| ImportError::BlockNotFound(cid.to_string()))?;
let value: Ipld = serde_ipld_dagcbor::from_slice(block)
.map_err(|e| ImportError::InvalidCbor(e.to_string()))?;
if let Ipld::Map(ref obj) = value {
if let Some(Ipld::Link(left_cid)) = obj.get("l") {
walk_mst_node(blocks, left_cid, prev_key, records)?;
}
visited.insert(cid);
let block = blocks
.get(&cid)
.ok_or_else(|| ImportError::BlockNotFound(cid.to_string()))?;
let value: Ipld = serde_ipld_dagcbor::from_slice(block)
.map_err(|e| ImportError::InvalidCbor(e.to_string()))?;
if let Ipld::Map(ref obj) = value {
if let Some(Ipld::List(entries)) = obj.get("e") {
for entry in entries {
if let Ipld::Map(entry_obj) = entry {
let key = entry_obj.get("k").and_then(|k| {
if let Ipld::Bytes(b) = k {
String::from_utf8(b.clone()).ok()
} else if let Ipld::String(s) = k {
Some(s.clone())
} else {
None
}
});
let record_cid = entry_obj.get("v").and_then(|v| {
if let Ipld::Link(cid) = v {
Some(*cid)
} else {
None
}
});
if let (Some(key), Some(record_cid)) = (key, record_cid)
&& let Some(record_block) = blocks.get(&record_cid)
&& let Ok(record_value) =
serde_ipld_dagcbor::from_slice::<Ipld>(record_block)
{
let blob_refs = find_blob_refs_ipld(&record_value, 0);
let parts: Vec<&str> = key.split('/').collect();
if parts.len() >= 2 {
let collection = parts[..parts.len() - 1].join("/");
let rkey = parts[parts.len() - 1].to_string();
records.push(ImportedRecord {
collection,
rkey,
cid: record_cid,
blob_refs,
});
}
let mut current_key = prev_key.to_vec();
if let Some(Ipld::List(entries)) = obj.get("e") {
for entry in entries {
if let Ipld::Map(entry_obj) = entry {
let prefix_len = entry_obj.get("p").and_then(|p| {
if let Ipld::Integer(n) = p {
Some(*n as usize)
} else {
None
}
if let Some(Ipld::Link(tree_cid)) = entry_obj.get("t") {
stack.push(*tree_cid);
}).unwrap_or(0);
let key_suffix = entry_obj.get("k").and_then(|k| {
if let Ipld::Bytes(b) = k {
Some(b.clone())
} else {
None
}
});
if let Some(suffix) = key_suffix {
current_key.truncate(prefix_len);
current_key.extend_from_slice(&suffix);
}
if let Some(Ipld::Link(tree_cid)) = entry_obj.get("t") {
walk_mst_node(blocks, tree_cid, &current_key, records)?;
}
let record_cid = entry_obj.get("v").and_then(|v| {
if let Ipld::Link(cid) = v {
Some(*cid)
} else {
None
}
});
if let Some(record_cid) = record_cid {
if let Ok(full_key) = String::from_utf8(current_key.clone()) {
if let Some(record_block) = blocks.get(&record_cid)
&& let Ok(record_value) =
serde_ipld_dagcbor::from_slice::<Ipld>(record_block)
{
let blob_refs = find_blob_refs_ipld(&record_value, 0);
let parts: Vec<&str> = full_key.split('/').collect();
if parts.len() >= 2 {
let collection = parts[..parts.len() - 1].join("/");
let rkey = parts[parts.len() - 1].to_string();
records.push(ImportedRecord {
collection,
rkey,
cid: record_cid,
blob_refs,
});
}
}
}
}
}
}
if let Some(Ipld::Link(left_cid)) = obj.get("l") {
stack.push(*left_cid);
}
}
}
Ok(records)
Ok(())
}
pub struct CommitInfo {
src/sync/util.rs
+9
View File
@@ -86,6 +86,15 @@ fn format_account_event(event: &SequencedEvent) -> Result<Vec<u8>, anyhow::Error
let mut bytes = Vec::new();
serde_ipld_dagcbor::to_writer(&mut bytes, &header)?;
serde_ipld_dagcbor::to_writer(&mut bytes, &frame)?;
let hex_str: String = bytes.iter().map(|b| format!("{:02x}", b)).collect();
tracing::info!(
did = %frame.did,
active = frame.active,
status = ?frame.status,
cbor_len = bytes.len(),
cbor_hex = %hex_str,
"Sending account event to firehose"
);
Ok(bytes)
}