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

Performance improvements

Browse Source
This commit is contained in:
lewis
2026-01-03 00:58:28 +02:00
parent 4375d795a0
commit 0f32cc3faa
86 changed files with 3271 additions and 608 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/api/admin/account/info.rs
+125 -31
View File
@@ -217,7 +217,10 @@ pub async fn get_account_infos(
_auth: BearerAuthAdmin,
RawQuery(raw_query): RawQuery,
) -> Response {
let dids = crate::util::parse_repeated_query_param(raw_query.as_deref(), "dids");
let dids: Vec<String> = crate::util::parse_repeated_query_param(raw_query.as_deref(), "dids")
.into_iter()
.filter(|d| !d.is_empty())
.collect();
if dids.is_empty() {
return (
StatusCode::BAD_REQUEST,
@@ -225,41 +228,132 @@ pub async fn get_account_infos(
)
.into_response();
}
let mut infos = Vec::new();
for did in &dids {
if did.is_empty() {
continue;
let users = match sqlx::query!(
r#"
SELECT id, did, handle, email, created_at, invites_disabled, email_verified, deactivated_at
FROM users
WHERE did = ANY($1)
"#,
&dids
)
.fetch_all(&state.db)
.await
{
Ok(rows) => rows,
Err(e) => {
error!("Failed to fetch account infos: {:?}", e);
return (
StatusCode::INTERNAL_SERVER_ERROR,
Json(json!({"error": "InternalError"})),
)
.into_response();
}
let result = sqlx::query!(
};
let user_ids: Vec<uuid::Uuid> = users.iter().map(|u| u.id).collect();
let all_invite_codes = sqlx::query!(
r#"
SELECT ic.code, ic.available_uses, ic.disabled, ic.for_account, ic.created_at,
ic.created_by_user, u.did as created_by
FROM invite_codes ic
JOIN users u ON ic.created_by_user = u.id
WHERE ic.created_by_user = ANY($1)
"#,
&user_ids
)
.fetch_all(&state.db)
.await
.unwrap_or_default();
let all_codes: Vec<String> = all_invite_codes.iter().map(|c| c.code.clone()).collect();
let all_invite_uses = if !all_codes.is_empty() {
sqlx::query!(
r#"
SELECT id, did, handle, email, created_at, invites_disabled, email_verified, deactivated_at
FROM users
WHERE did = $1
SELECT icu.code, u.did as used_by, icu.used_at
FROM invite_code_uses icu
JOIN users u ON icu.used_by_user = u.id
WHERE icu.code = ANY($1)
"#,
did
&all_codes
)
.fetch_optional(&state.db)
.await;
if let Ok(Some(row)) = result {
let invited_by = get_invited_by(&state.db, row.id).await;
let invites = get_invites_for_user(&state.db, row.id).await;
infos.push(AccountInfo {
did: row.did,
handle: row.handle,
email: row.email,
indexed_at: row.created_at.to_rfc3339(),
invite_note: None,
invites_disabled: row.invites_disabled.unwrap_or(false),
email_confirmed_at: if row.email_verified {
Some(row.created_at.to_rfc3339())
} else {
None
},
deactivated_at: row.deactivated_at.map(|dt| dt.to_rfc3339()),
invited_by,
invites,
.fetch_all(&state.db)
.await
.unwrap_or_default()
} else {
Vec::new()
};
let invited_by_map: std::collections::HashMap<uuid::Uuid, String> = sqlx::query!(
r#"
SELECT icu.used_by_user, icu.code
FROM invite_code_uses icu
WHERE icu.used_by_user = ANY($1)
"#,
&user_ids
)
.fetch_all(&state.db)
.await
.unwrap_or_default()
.into_iter()
.map(|r| (r.used_by_user, r.code))
.collect();
let mut uses_by_code: std::collections::HashMap<String, Vec<InviteCodeUseInfo>> =
std::collections::HashMap::new();
for u in all_invite_uses {
uses_by_code
.entry(u.code.clone())
.or_default()
.push(InviteCodeUseInfo {
used_by: u.used_by,
used_at: u.used_at.to_rfc3339(),
});
}
}
let mut codes_by_user: std::collections::HashMap<uuid::Uuid, Vec<InviteCodeInfo>> =
std::collections::HashMap::new();
let mut code_info_map: std::collections::HashMap<String, InviteCodeInfo> =
std::collections::HashMap::new();
for ic in all_invite_codes {
let info = InviteCodeInfo {
code: ic.code.clone(),
available: ic.available_uses,
disabled: ic.disabled.unwrap_or(false),
for_account: ic.for_account,
created_by: ic.created_by,
created_at: ic.created_at.to_rfc3339(),
uses: uses_by_code.get(&ic.code).cloned().unwrap_or_default(),
};
code_info_map.insert(ic.code.clone(), info.clone());
codes_by_user
.entry(ic.created_by_user)
.or_default()
.push(info);
}
let mut infos = Vec::with_capacity(users.len());
for row in users {
let invited_by = invited_by_map
.get(&row.id)
.and_then(|code| code_info_map.get(code).cloned());
let invites = codes_by_user.get(&row.id).cloned();
infos.push(AccountInfo {
did: row.did,
handle: row.handle,
email: row.email,
indexed_at: row.created_at.to_rfc3339(),
invite_note: None,
invites_disabled: row.invites_disabled.unwrap_or(false),
email_confirmed_at: if row.email_verified {
Some(row.created_at.to_rfc3339())
} else {
None
},
deactivated_at: row.deactivated_at.map(|dt| dt.to_rfc3339()),
invited_by,
invites,
});
}
(StatusCode::OK, Json(GetAccountInfosOutput { infos })).into_response()
}
src/api/delegation.rs
+1 -1
View File
@@ -726,7 +726,7 @@ pub async fn create_delegated_account(
}
};
let plc_client = crate::plc::PlcClient::new(None);
let plc_client = crate::plc::PlcClient::with_cache(None, Some(state.cache.clone()));
if let Err(e) = plc_client
.send_operation(&genesis_result.did, &genesis_result.signed_operation)
.await
src/api/identity/account.rs
+23 -13
View File
@@ -451,7 +451,7 @@ pub async fn create_account(
.into_response();
}
};
let plc_client = PlcClient::new(None);
let plc_client = PlcClient::with_cache(None, Some(state.cache.clone()));
if let Err(e) = plc_client
.send_operation(&genesis_result.did, &genesis_result.signed_operation)
.await
@@ -488,7 +488,7 @@ pub async fn create_account(
.into_response();
}
};
let plc_client = PlcClient::new(None);
let plc_client = PlcClient::with_cache(None, Some(state.cache.clone()));
if let Err(e) = plc_client
.send_operation(&genesis_result.did, &genesis_result.signed_operation)
.await
@@ -745,17 +745,27 @@ pub async fn create_account(
.into_response();
}
let password_hash = match hash(&input.password, DEFAULT_COST) {
Ok(h) => h,
Err(e) => {
error!("Error hashing password: {:?}", e);
return (
StatusCode::INTERNAL_SERVER_ERROR,
Json(json!({"error": "InternalError"})),
)
.into_response();
}
};
let password_clone = input.password.clone();
let password_hash =
match tokio::task::spawn_blocking(move || hash(&password_clone, DEFAULT_COST)).await {
Ok(Ok(h)) => h,
Ok(Err(e)) => {
error!("Error hashing password: {:?}", e);
return (
StatusCode::INTERNAL_SERVER_ERROR,
Json(json!({"error": "InternalError"})),
)
.into_response();
}
Err(e) => {
error!("Failed to spawn blocking task: {:?}", e);
return (
StatusCode::INTERNAL_SERVER_ERROR,
Json(json!({"error": "InternalError"})),
)
.into_response();
}
};
let is_first_user = sqlx::query_scalar!("SELECT COUNT(*) as count FROM users")
.fetch_one(&mut *tx)
.await
src/api/identity/did.rs
+2 -6
View File
@@ -10,7 +10,6 @@ use axum::{
use base64::Engine;
use k256::SecretKey;
use k256::elliptic_curve::sec1::ToEncodedPoint;
use reqwest;
use serde::{Deserialize, Serialize};
use serde_json::json;
use tracing::{error, warn};
@@ -504,10 +503,7 @@ pub async fn verify_did_web(
let path = parts[3..].join("/");
format!("{}://{}/{}/did.json", scheme, domain, path)
};
let client = reqwest::Client::builder()
.timeout(std::time::Duration::from_secs(5))
.build()
.map_err(|e| format!("Failed to create client: {}", e))?;
let client = crate::api::proxy_client::did_resolution_client();
let resp = client
.get(&url)
.send()
@@ -926,7 +922,7 @@ pub async fn update_plc_handle(
};
let key_bytes = crate::config::decrypt_key(&user_row.key_bytes, user_row.encryption_version)?;
let signing_key = k256::ecdsa::SigningKey::from_slice(&key_bytes)?;
let plc_client = crate::plc::PlcClient::new(None);
let plc_client = crate::plc::PlcClient::with_cache(None, Some(state.cache.clone()));
let last_op = plc_client.get_last_op(did).await?;
let new_also_known_as = vec![format!("at://{}", new_handle)];
let update_op =
src/api/identity/plc/sign.rs
+1 -1
View File
@@ -174,7 +174,7 @@ pub async fn sign_plc_operation(
.into_response();
}
};
let plc_client = PlcClient::new(None);
let plc_client = PlcClient::with_cache(None, Some(state.cache.clone()));
let did_clone = did.clone();
let result: Result<PlcOpOrTombstone, CircuitBreakerError<PlcError>> =
with_circuit_breaker(&state.circuit_breakers.plc_directory, || async {
src/api/identity/plc/submit.rs
+1 -1
View File
@@ -184,7 +184,7 @@ pub async fn submit_plc_operation(
.into_response();
}
}
let plc_client = PlcClient::new(None);
let plc_client = PlcClient::with_cache(None, Some(state.cache.clone()));
let operation_clone = input.operation.clone();
let did_clone = did.clone();
let result: Result<(), CircuitBreakerError<PlcError>> =
src/api/proxy_client.rs
+31
View File
@@ -10,6 +10,8 @@ pub const DEFAULT_CONNECT_TIMEOUT: Duration = Duration::from_secs(5);
pub const MAX_RESPONSE_SIZE: u64 = 10 * 1024 * 1024;
static PROXY_CLIENT: OnceLock<Client> = OnceLock::new();
static DID_RESOLUTION_CLIENT: OnceLock<Client> = OnceLock::new();
static HANDLE_RESOLUTION_CLIENT: OnceLock<Client> = OnceLock::new();
pub fn proxy_client() -> &'static Client {
PROXY_CLIENT.get_or_init(|| {
@@ -26,6 +28,35 @@ pub fn proxy_client() -> &'static Client {
})
}
pub fn did_resolution_client() -> &'static Client {
DID_RESOLUTION_CLIENT.get_or_init(|| {
ClientBuilder::new()
.timeout(Duration::from_secs(5))
.connect_timeout(DEFAULT_CONNECT_TIMEOUT)
.pool_max_idle_per_host(10)
.pool_idle_timeout(Duration::from_secs(90))
.build()
.expect(
"Failed to build DID resolution client - this indicates a TLS or system configuration issue",
)
})
}
pub fn handle_resolution_client() -> &'static Client {
HANDLE_RESOLUTION_CLIENT.get_or_init(|| {
ClientBuilder::new()
.timeout(Duration::from_secs(10))
.connect_timeout(DEFAULT_CONNECT_TIMEOUT)
.pool_max_idle_per_host(10)
.pool_idle_timeout(Duration::from_secs(90))
.redirect(reqwest::redirect::Policy::limited(5))
.build()
.expect(
"Failed to build handle resolution client - this indicates a TLS or system configuration issue",
)
})
}
pub fn is_ssrf_safe(url: &str) -> Result<(), SsrfError> {
let parsed = Url::parse(url).map_err(|_| SsrfError::InvalidUrl)?;
let scheme = parsed.scheme();
src/api/repo/blob.rs
+74 -40
View File
@@ -2,24 +2,26 @@ use crate::auth::{ServiceTokenVerifier, is_service_token};
use crate::delegation::{self, DelegationActionType};
use crate::state::AppState;
use crate::util::get_max_blob_size;
use axum::body::Bytes;
use axum::body::Body;
use axum::{
Json,
extract::{Query, State},
http::StatusCode,
response::{IntoResponse, Response},
};
use bytes::Bytes;
use cid::Cid;
use futures::StreamExt;
use multihash::Multihash;
use serde::{Deserialize, Serialize};
use serde_json::json;
use sha2::{Digest, Sha256};
use tracing::{debug, error};
use std::pin::Pin;
use tracing::{debug, error, info};
pub async fn upload_blob(
State(state): State<AppState>,
headers: axum::http::HeaderMap,
body: Bytes,
body: Body,
) -> Response {
let token = match crate::auth::extract_bearer_token_from_header(
headers.get("Authorization").and_then(|h| h.to_str().ok()),
@@ -106,39 +108,12 @@ pub async fn upload_blob(
.into_response();
}
let max_size = get_max_blob_size();
if body.len() > max_size {
return (
StatusCode::PAYLOAD_TOO_LARGE,
Json(json!({"error": "BlobTooLarge", "message": format!("Blob size {} exceeds maximum of {} bytes", body.len(), max_size)})),
)
.into_response();
}
let mime_type = headers
.get("content-type")
.and_then(|h| h.to_str().ok())
.unwrap_or("application/octet-stream")
.to_string();
let size = body.len() as i64;
let data = body.to_vec();
let mut hasher = Sha256::new();
hasher.update(&data);
let hash = hasher.finalize();
let multihash = match Multihash::wrap(0x12, &hash) {
Ok(mh) => mh,
Err(e) => {
error!("Failed to create multihash for blob: {:?}", e);
return (
StatusCode::INTERNAL_SERVER_ERROR,
Json(json!({"error": "InternalError", "message": "Failed to hash blob"})),
)
.into_response();
}
};
let cid = Cid::new_v1(0x55, multihash);
let cid_str = cid.to_string();
let storage_key = format!("blobs/{}", cid_str);
let user_query = sqlx::query!("SELECT id FROM users WHERE did = $1", did)
.fetch_optional(&state.db)
.await;
@@ -152,9 +127,65 @@ pub async fn upload_blob(
.into_response();
}
};
let temp_key = format!("temp/{}", uuid::Uuid::new_v4());
let max_size = get_max_blob_size() as u64;
let body_stream = body.into_data_stream();
let mapped_stream =
body_stream.map(|result| result.map_err(|e| std::io::Error::other(e.to_string())));
let pinned_stream: Pin<Box<dyn futures::Stream<Item = Result<Bytes, std::io::Error>> + Send>> =
Box::pin(mapped_stream);
info!("Starting streaming blob upload to temp key: {}", temp_key);
let upload_result = match state.blob_store.put_stream(&temp_key, pinned_stream).await {
Ok(result) => result,
Err(e) => {
error!("Failed to stream blob to storage: {:?}", e);
return (
StatusCode::INTERNAL_SERVER_ERROR,
Json(json!({"error": "InternalError", "message": "Failed to store blob"})),
)
.into_response();
}
};
let size = upload_result.size;
if size > max_size {
let _ = state.blob_store.delete(&temp_key).await;
return (
StatusCode::PAYLOAD_TOO_LARGE,
Json(json!({"error": "BlobTooLarge", "message": format!("Blob size {} exceeds maximum of {} bytes", size, max_size)})),
)
.into_response();
}
let multihash = match Multihash::wrap(0x12, &upload_result.sha256_hash) {
Ok(mh) => mh,
Err(e) => {
let _ = state.blob_store.delete(&temp_key).await;
error!("Failed to create multihash for blob: {:?}", e);
return (
StatusCode::INTERNAL_SERVER_ERROR,
Json(json!({"error": "InternalError", "message": "Failed to hash blob"})),
)
.into_response();
}
};
let cid = Cid::new_v1(0x55, multihash);
let cid_str = cid.to_string();
let storage_key = format!("blobs/{}", cid_str);
info!(
"Blob upload complete: size={}, cid={}, copying to final location",
size, cid_str
);
let mut tx = match state.db.begin().await {
Ok(tx) => tx,
Err(e) => {
let _ = state.blob_store.delete(&temp_key).await;
error!("Failed to begin transaction: {:?}", e);
return (
StatusCode::INTERNAL_SERVER_ERROR,
@@ -163,20 +194,23 @@ pub async fn upload_blob(
.into_response();
}
};
let insert = sqlx::query!(
"INSERT INTO blobs (cid, mime_type, size_bytes, created_by_user, storage_key) VALUES ($1, $2, $3, $4, $5) ON CONFLICT (cid) DO NOTHING RETURNING cid",
cid_str,
mime_type,
size,
size as i64,
user_id,
storage_key
)
.fetch_optional(&mut *tx)
.await;
let was_inserted = match insert {
Ok(Some(_)) => true,
Ok(None) => false,
Err(e) => {
let _ = state.blob_store.delete(&temp_key).await;
error!("Failed to insert blob record: {:?}", e);
return (
StatusCode::INTERNAL_SERVER_ERROR,
@@ -185,19 +219,19 @@ pub async fn upload_blob(
.into_response();
}
};
if was_inserted
&& let Err(e) = state
.blob_store
.put_bytes(&storage_key, bytes::Bytes::from(data))
.await
{
error!("Failed to upload blob to storage: {:?}", e);
if was_inserted && let Err(e) = state.blob_store.copy(&temp_key, &storage_key).await {
let _ = state.blob_store.delete(&temp_key).await;
error!("Failed to copy blob to final location: {:?}", e);
return (
StatusCode::INTERNAL_SERVER_ERROR,
Json(json!({"error": "InternalError", "message": "Failed to store blob"})),
)
.into_response();
}
let _ = state.blob_store.delete(&temp_key).await;
if let Err(e) = tx.commit().await {
error!("Failed to commit blob transaction: {:?}", e);
if was_inserted && let Err(cleanup_err) = state.blob_store.delete(&storage_key).await {
src/api/repo/import.rs
+2 -2
View File
@@ -16,8 +16,8 @@ use k256::ecdsa::SigningKey;
use serde_json::json;
use tracing::{debug, error, info, warn};
const DEFAULT_MAX_IMPORT_SIZE: usize = 100 * 1024 * 1024;
const DEFAULT_MAX_BLOCKS: usize = 50000;
const DEFAULT_MAX_IMPORT_SIZE: usize = 1024 * 1024 * 1024;
const DEFAULT_MAX_BLOCKS: usize = 500000;
pub async fn import_repo(
State(state): State<AppState>,
src/api/server/account_status.rs
+10 -6
View File
@@ -1,4 +1,5 @@
use crate::api::ApiError;
use crate::cache::Cache;
use crate::plc::PlcClient;
use crate::state::AppState;
use axum::{
@@ -16,6 +17,7 @@ use k256::ecdsa::SigningKey;
use serde::{Deserialize, Serialize};
use serde_json::json;
use std::str::FromStr;
use std::sync::Arc;
use tracing::{error, info, warn};
use uuid::Uuid;
@@ -140,7 +142,7 @@ pub async fn check_account_status(
.await
.unwrap_or(Some(0))
.unwrap_or(0);
let valid_did = is_valid_did_for_service(&state.db, &did).await;
let valid_did = is_valid_did_for_service(&state.db, &state.cache, &did).await;
(
StatusCode::OK,
Json(CheckAccountStatusOutput {
@@ -158,14 +160,15 @@ pub async fn check_account_status(
.into_response()
}
async fn is_valid_did_for_service(db: &sqlx::PgPool, did: &str) -> bool {
assert_valid_did_document_for_service(db, did, false)
async fn is_valid_did_for_service(db: &sqlx::PgPool, cache: &Arc<dyn Cache>, did: &str) -> bool {
assert_valid_did_document_for_service(db, cache, did, false)
.await
.is_ok()
}
async fn assert_valid_did_document_for_service(
db: &sqlx::PgPool,
cache: &Arc<dyn Cache>,
did: &str,
with_retry: bool,
) -> Result<(), (StatusCode, Json<serde_json::Value>)> {
@@ -173,7 +176,7 @@ async fn assert_valid_did_document_for_service(
let expected_endpoint = format!("https://{}", hostname);
if did.starts_with("did:plc:") {
let plc_client = PlcClient::new(None);
let plc_client = PlcClient::with_cache(None, Some(cache.clone()));
let max_attempts = if with_retry { 5 } else { 1 };
let mut last_error = None;
@@ -308,7 +311,7 @@ async fn assert_valid_did_document_for_service(
}
}
} else if let Some(host_and_path) = did.strip_prefix("did:web:") {
let client = reqwest::Client::new();
let client = crate::api::proxy_client::did_resolution_client();
let decoded = host_and_path.replace("%3A", ":");
let parts: Vec<&str> = decoded.split(':').collect();
let (host, path_parts) = if parts.len() > 1 && parts[1].chars().all(|c| c.is_ascii_digit())
@@ -438,7 +441,8 @@ pub async fn activate_account(
did
);
let did_validation_start = std::time::Instant::now();
if let Err((status, json)) = assert_valid_did_document_for_service(&state.db, &did, true).await
if let Err((status, json)) =
assert_valid_did_document_for_service(&state.db, &state.cache, &did, true).await
{
info!(
"[MIGRATION] activateAccount: DID document validation FAILED for {} (took {:?})",
src/api/server/app_password.rs
+12 -3
View File
@@ -158,12 +158,21 @@ pub async fn create_app_password(
})
.collect::<Vec<String>>()
.join("-");
let password_hash = match bcrypt::hash(&password, bcrypt::DEFAULT_COST) {
Ok(h) => h,
Err(e) => {
let password_clone = password.clone();
let password_hash = match tokio::task::spawn_blocking(move || {
bcrypt::hash(&password_clone, bcrypt::DEFAULT_COST)
})
.await
{
Ok(Ok(h)) => h,
Ok(Err(e)) => {
error!("Failed to hash password: {:?}", e);
return ApiError::InternalError.into_response();
}
Err(e) => {
error!("Failed to spawn blocking task: {:?}", e);
return ApiError::InternalError.into_response();
}
};
let privileged = input.privileged.unwrap_or(false);
let created_at = chrono::Utc::now();
src/api/server/passkey_account.rs
+1 -1
View File
@@ -436,7 +436,7 @@ pub async fn create_passkey_account(
}
};
let plc_client = crate::plc::PlcClient::new(None);
let plc_client = crate::plc::PlcClient::with_cache(None, Some(state.cache.clone()));
if let Err(e) = plc_client
.send_operation(&genesis_result.did, &genesis_result.signed_operation)
.await
src/api/server/password.rs
+42 -22
View File
@@ -226,17 +226,27 @@ pub async fn reset_password(
)
.into_response();
}
let password_hash = match hash(password, DEFAULT_COST) {
Ok(h) => h,
Err(e) => {
error!("Failed to hash password: {:?}", e);
return (
StatusCode::INTERNAL_SERVER_ERROR,
Json(json!({"error": "InternalError"})),
)
.into_response();
}
};
let password_clone = password.to_string();
let password_hash =
match tokio::task::spawn_blocking(move || hash(password_clone, DEFAULT_COST)).await {
Ok(Ok(h)) => h,
Ok(Err(e)) => {
error!("Failed to hash password: {:?}", e);
return (
StatusCode::INTERNAL_SERVER_ERROR,
Json(json!({"error": "InternalError"})),
)
.into_response();
}
Err(e) => {
error!("Failed to spawn blocking task: {:?}", e);
return (
StatusCode::INTERNAL_SERVER_ERROR,
Json(json!({"error": "InternalError"})),
)
.into_response();
}
};
let mut tx = match state.db.begin().await {
Ok(tx) => tx,
Err(e) => {
@@ -409,17 +419,27 @@ pub async fn change_password(
)
.into_response();
}
let new_hash = match hash(new_password, DEFAULT_COST) {
Ok(h) => h,
Err(e) => {
error!("Failed to hash password: {:?}", e);
return (
StatusCode::INTERNAL_SERVER_ERROR,
Json(json!({"error": "InternalError"})),
)
.into_response();
}
};
let new_password_clone = new_password.to_string();
let new_hash =
match tokio::task::spawn_blocking(move || hash(new_password_clone, DEFAULT_COST)).await {
Ok(Ok(h)) => h,
Ok(Err(e)) => {
error!("Failed to hash password: {:?}", e);
return (
StatusCode::INTERNAL_SERVER_ERROR,
Json(json!({"error": "InternalError"})),
)
.into_response();
}
Err(e) => {
error!("Failed to spawn blocking task: {:?}", e);
return (
StatusCode::INTERNAL_SERVER_ERROR,
Json(json!({"error": "InternalError"})),
)
.into_response();
}
};
if let Err(e) = sqlx::query("UPDATE users SET password_hash = $1 WHERE id = $2")
.bind(&new_hash)
.bind(user_id)
src/auth/mod.rs
+84 -14
View File
@@ -35,6 +35,14 @@ pub use verify::{
const KEY_CACHE_TTL_SECS: u64 = 300;
const SESSION_CACHE_TTL_SECS: u64 = 60;
const USER_STATUS_CACHE_TTL_SECS: u64 = 60;
#[derive(Serialize, Deserialize)]
struct CachedUserStatus {
deactivated: bool,
takendown: bool,
is_admin: bool,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TokenValidationError {
@@ -149,23 +157,67 @@ async fn validate_bearer_token_with_options_internal(
let (decrypted_key, deactivated_at, takedown_ref, is_admin) = if let Some(key) = cached_key
{
let user_status = sqlx::query!(
"SELECT deactivated_at, takedown_ref, is_admin FROM users WHERE did = $1",
did
)
.fetch_optional(db)
.await
.ok()
.flatten();
let status_cache_key = format!("auth:status:{}", did);
let cached_status: Option<CachedUserStatus> = if let Some(c) = cache {
c.get(&status_cache_key)
.await
.and_then(|s| serde_json::from_str(&s).ok())
} else {
None
};
match user_status {
Some(status) => (
if let Some(status) = cached_status {
(
Some(key),
status.deactivated_at,
status.takedown_ref,
if status.deactivated {
Some(chrono::Utc::now())
} else {
None
},
if status.takendown {
Some("takendown".to_string())
} else {
None
},
status.is_admin,
),
None => (None, None, None, false),
)
} else {
let user_status = sqlx::query!(
"SELECT deactivated_at, takedown_ref, is_admin FROM users WHERE did = $1",
did
)
.fetch_optional(db)
.await
.ok()
.flatten();
match user_status {
Some(status) => {
if let Some(c) = cache {
let cached = CachedUserStatus {
deactivated: status.deactivated_at.is_some(),
takendown: status.takedown_ref.is_some(),
is_admin: status.is_admin,
};
if let Ok(json) = serde_json::to_string(&cached) {
let _ = c
.set(
&status_cache_key,
&json,
Duration::from_secs(USER_STATUS_CACHE_TTL_SECS),
)
.await;
}
}
(
Some(key),
status.deactivated_at,
status.takedown_ref,
status.is_admin,
)
}
None => (None, None, None, false),
}
}
} else if let Some(user) = sqlx::query!(
"SELECT k.key_bytes, k.encryption_version, u.deactivated_at, u.takedown_ref, u.is_admin
@@ -190,6 +242,22 @@ async fn validate_bearer_token_with_options_internal(
Duration::from_secs(KEY_CACHE_TTL_SECS),
)
.await;
let status_cache_key = format!("auth:status:{}", did);
let cached = CachedUserStatus {
deactivated: user.deactivated_at.is_some(),
takendown: user.takedown_ref.is_some(),
is_admin: user.is_admin,
};
if let Ok(json) = serde_json::to_string(&cached) {
let _ = c
.set(
&status_cache_key,
&json,
Duration::from_secs(USER_STATUS_CACHE_TTL_SECS),
)
.await;
}
}
(
@@ -328,7 +396,9 @@ async fn validate_bearer_token_with_options_internal(
pub async fn invalidate_auth_cache(cache: &Arc<dyn Cache>, did: &str) {
let key_cache_key = format!("auth:key:{}", did);
let status_cache_key = format!("auth:status:{}", did);
let _ = cache.delete(&key_cache_key).await;
let _ = cache.delete(&status_cache_key).await;
}
pub async fn validate_token_with_dpop(
src/auth/service.rs
+2
View File
@@ -85,6 +85,8 @@ impl ServiceTokenVerifier {
let client = Client::builder()
.timeout(Duration::from_secs(10))
.connect_timeout(Duration::from_secs(5))
.pool_max_idle_per_host(10)
.pool_idle_timeout(Duration::from_secs(90))
.build()
.unwrap_or_else(|_| Client::new());
src/crawlers.rs
+3
View File
@@ -24,6 +24,9 @@ impl Crawlers {
crawler_urls,
http_client: Client::builder()
.timeout(Duration::from_secs(30))
.connect_timeout(Duration::from_secs(5))
.pool_max_idle_per_host(5)
.pool_idle_timeout(Duration::from_secs(90))
.build()
.unwrap_or_default(),
last_notified: AtomicU64::new(0),
src/handle/mod.rs
+1 -7
View File
@@ -2,8 +2,6 @@ pub mod reserved;
use hickory_resolver::TokioAsyncResolver;
use hickory_resolver::config::{ResolverConfig, ResolverOpts};
use reqwest::Client;
use std::time::Duration;
use thiserror::Error;
#[derive(Error, Debug)]
@@ -43,11 +41,7 @@ pub async fn resolve_handle_dns(handle: &str) -> Result<String, HandleResolution
pub async fn resolve_handle_http(handle: &str) -> Result<String, HandleResolutionError> {
let url = format!("https://{}/.well-known/atproto-did", handle);
let client = Client::builder()
.timeout(Duration::from_secs(10))
.redirect(reqwest::redirect::Policy::limited(5))
.build()
.map_err(|e| HandleResolutionError::HttpError(e.to_string()))?;
let client = crate::api::proxy_client::handle_resolution_client();
let response = client
.get(&url)
.header("Accept", "text/plain")
src/oauth/client.rs
+2
View File
@@ -80,6 +80,8 @@ impl ClientMetadataCache {
http_client: Client::builder()
.timeout(std::time::Duration::from_secs(30))
.connect_timeout(std::time::Duration::from_secs(10))
.pool_max_idle_per_host(10)
.pool_idle_timeout(std::time::Duration::from_secs(90))
.build()
.unwrap_or_else(|_| Client::new()),
cache_ttl_secs,
src/plc/mod.rs
+57 -5
View File
@@ -1,3 +1,4 @@
use crate::cache::Cache;
use base32::Alphabet;
use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD};
use k256::ecdsa::{Signature, SigningKey, signature::Signer};
@@ -6,6 +7,7 @@ use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use sha2::{Digest, Sha256};
use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;
use thiserror::Error;
@@ -77,13 +79,20 @@ impl PlcOpOrTombstone {
}
}
const PLC_CACHE_TTL_SECS: u64 = 300;
pub struct PlcClient {
base_url: String,
client: Client,
cache: Option<Arc<dyn Cache>>,
}
impl PlcClient {
pub fn new(base_url: Option<String>) -> Self {
Self::with_cache(base_url, None)
}
pub fn with_cache(base_url: Option<String>, cache: Option<Arc<dyn Cache>>) -> Self {
let base_url = base_url.unwrap_or_else(|| {
std::env::var("PLC_DIRECTORY_URL")
.unwrap_or_else(|_| "https://plc.directory".to_string())
@@ -100,9 +109,14 @@ impl PlcClient {
.timeout(Duration::from_secs(timeout_secs))
.connect_timeout(Duration::from_secs(connect_timeout_secs))
.pool_max_idle_per_host(5)
.pool_idle_timeout(Duration::from_secs(90))
.build()
.unwrap_or_else(|_| Client::new());
Self { base_url, client }
Self {
base_url,
client,
cache,
}
}
fn encode_did(did: &str) -> String {
@@ -110,6 +124,13 @@ impl PlcClient {
}
pub async fn get_document(&self, did: &str) -> Result<Value, PlcError> {
let cache_key = format!("plc:doc:{}", did);
if let Some(ref cache) = self.cache
&& let Some(cached) = cache.get(&cache_key).await
&& let Ok(value) = serde_json::from_str(&cached)
{
return Ok(value);
}
let url = format!("{}/{}", self.base_url, Self::encode_did(did));
let response = self.client.get(&url).send().await?;
if response.status() == reqwest::StatusCode::NOT_FOUND {
@@ -123,13 +144,32 @@ impl PlcClient {
status, body
)));
}
response
let value: Value = response
.json()
.await
.map_err(|e| PlcError::InvalidResponse(e.to_string()))
.map_err(|e| PlcError::InvalidResponse(e.to_string()))?;
if let Some(ref cache) = self.cache
&& let Ok(json_str) = serde_json::to_string(&value)
{
let _ = cache
.set(
&cache_key,
&json_str,
Duration::from_secs(PLC_CACHE_TTL_SECS),
)
.await;
}
Ok(value)
}
pub async fn get_document_data(&self, did: &str) -> Result<Value, PlcError> {
let cache_key = format!("plc:data:{}", did);
if let Some(ref cache) = self.cache
&& let Some(cached) = cache.get(&cache_key).await
&& let Ok(value) = serde_json::from_str(&cached)
{
return Ok(value);
}
let url = format!("{}/{}/data", self.base_url, Self::encode_did(did));
let response = self.client.get(&url).send().await?;
if response.status() == reqwest::StatusCode::NOT_FOUND {
@@ -143,10 +183,22 @@ impl PlcClient {
status, body
)));
}
response
let value: Value = response
.json()
.await
.map_err(|e| PlcError::InvalidResponse(e.to_string()))
.map_err(|e| PlcError::InvalidResponse(e.to_string()))?;
if let Some(ref cache) = self.cache
&& let Ok(json_str) = serde_json::to_string(&value)
{
let _ = cache
.set(
&cache_key,
&json_str,
Duration::from_secs(PLC_CACHE_TTL_SECS),
)
.await;
}
Ok(value)
}
pub async fn get_last_op(&self, did: &str) -> Result<PlcOpOrTombstone, PlcError> {
src/scheduled.rs
+30 -25
View File
@@ -343,7 +343,9 @@ pub async fn backfill_record_blobs(db: &PgPool, block_store: PostgresBlockStore)
}
};
let mut blob_refs_found = 0;
let mut batch_record_uris: Vec<String> = Vec::new();
let mut batch_blob_cids: Vec<String> = Vec::new();
for record in records {
let record_cid = match Cid::from_str(&record.record_cid) {
Ok(c) => c,
@@ -363,33 +365,36 @@ pub async fn backfill_record_blobs(db: &PgPool, block_store: PostgresBlockStore)
let blob_refs = crate::sync::import::find_blob_refs_ipld(&record_ipld, 0);
for blob_ref in blob_refs {
let record_uri = format!("at://{}/{}/{}", user.did, record.collection, record.rkey);
if let Err(e) = sqlx::query!(
r#"
INSERT INTO record_blobs (repo_id, record_uri, blob_cid)
VALUES ($1, $2, $3)
ON CONFLICT (repo_id, record_uri, blob_cid) DO NOTHING
"#,
user.user_id,
record_uri,
blob_ref.cid
)
.execute(db)
.await
{
warn!(error = %e, "Failed to insert record_blob during backfill");
} else {
blob_refs_found += 1;
}
batch_record_uris.push(record_uri);
batch_blob_cids.push(blob_ref.cid);
}
}
if blob_refs_found > 0 {
info!(
user_id = %user.user_id,
did = %user.did,
blob_refs = blob_refs_found,
"Backfilled record_blobs"
);
let blob_refs_found = batch_record_uris.len();
if !batch_record_uris.is_empty() {
if let Err(e) = sqlx::query!(
r#"
INSERT INTO record_blobs (repo_id, record_uri, blob_cid)
SELECT $1, record_uri, blob_cid
FROM UNNEST($2::text[], $3::text[]) AS t(record_uri, blob_cid)
ON CONFLICT (repo_id, record_uri, blob_cid) DO NOTHING
"#,
user.user_id,
&batch_record_uris,
&batch_blob_cids
)
.execute(db)
.await
{
warn!(error = %e, "Failed to batch insert record_blobs during backfill");
} else {
info!(
user_id = %user.user_id,
did = %user.did,
blob_refs = blob_refs_found,
"Backfilled record_blobs"
);
}
}
success += 1;
}
src/storage/mod.rs
+184
View File
@@ -3,9 +3,16 @@ use aws_config::BehaviorVersion;
use aws_config::meta::region::RegionProviderChain;
use aws_sdk_s3::Client;
use aws_sdk_s3::primitives::ByteStream;
use aws_sdk_s3::types::CompletedMultipartUpload;
use aws_sdk_s3::types::CompletedPart;
use bytes::Bytes;
use futures::Stream;
use sha2::{Digest, Sha256};
use std::pin::Pin;
use thiserror::Error;
const MIN_PART_SIZE: usize = 5 * 1024 * 1024;
#[derive(Error, Debug)]
pub enum StorageError {
#[error("IO error: {0}")]
@@ -16,6 +23,11 @@ pub enum StorageError {
Other(String),
}
pub struct StreamUploadResult {
pub sha256_hash: [u8; 32],
pub size: u64,
}
#[async_trait]
pub trait BlobStorage: Send + Sync {
async fn put(&self, key: &str, data: &[u8]) -> Result<(), StorageError>;
@@ -23,6 +35,12 @@ pub trait BlobStorage: Send + Sync {
async fn get(&self, key: &str) -> Result<Vec<u8>, StorageError>;
async fn get_bytes(&self, key: &str) -> Result<Bytes, StorageError>;
async fn delete(&self, key: &str) -> Result<(), StorageError>;
async fn put_stream(
&self,
key: &str,
stream: Pin<Box<dyn Stream<Item = Result<Bytes, std::io::Error>> + Send>>,
) -> Result<StreamUploadResult, StorageError>;
async fn copy(&self, src_key: &str, dst_key: &str) -> Result<(), StorageError>;
}
pub struct S3BlobStorage {
@@ -233,4 +251,170 @@ impl BlobStorage for S3BlobStorage {
result?;
Ok(())
}
async fn put_stream(
&self,
key: &str,
mut stream: Pin<Box<dyn Stream<Item = Result<Bytes, std::io::Error>> + Send>>,
) -> Result<StreamUploadResult, StorageError> {
use futures::StreamExt;
let create_resp = self
.client
.create_multipart_upload()
.bucket(&self.bucket)
.key(key)
.send()
.await
.map_err(|e| StorageError::S3(format!("Failed to create multipart upload: {}", e)))?;
let upload_id = create_resp
.upload_id()
.ok_or_else(|| StorageError::S3("No upload ID returned".to_string()))?
.to_string();
let mut hasher = Sha256::new();
let mut total_size: u64 = 0;
let mut part_number = 1;
let mut completed_parts: Vec<CompletedPart> = Vec::new();
let mut buffer = Vec::with_capacity(MIN_PART_SIZE);
let upload_part = |client: &Client,
bucket: &str,
key: &str,
upload_id: &str,
part_num: i32,
data: Vec<u8>|
-> std::pin::Pin<
Box<dyn std::future::Future<Output = Result<CompletedPart, StorageError>> + Send>,
> {
let client = client.clone();
let bucket = bucket.to_string();
let key = key.to_string();
let upload_id = upload_id.to_string();
Box::pin(async move {
let resp = client
.upload_part()
.bucket(&bucket)
.key(&key)
.upload_id(&upload_id)
.part_number(part_num)
.body(ByteStream::from(data))
.send()
.await
.map_err(|e| StorageError::S3(format!("Failed to upload part: {}", e)))?;
let etag = resp
.e_tag()
.ok_or_else(|| StorageError::S3("No ETag returned for part".to_string()))?
.to_string();
Ok(CompletedPart::builder()
.part_number(part_num)
.e_tag(etag)
.build())
})
};
loop {
match stream.next().await {
Some(Ok(chunk)) => {
hasher.update(&chunk);
total_size += chunk.len() as u64;
buffer.extend_from_slice(&chunk);
if buffer.len() >= MIN_PART_SIZE {
let part_data =
std::mem::replace(&mut buffer, Vec::with_capacity(MIN_PART_SIZE));
let part = upload_part(
&self.client,
&self.bucket,
key,
&upload_id,
part_number,
part_data,
)
.await?;
completed_parts.push(part);
part_number += 1;
}
}
Some(Err(e)) => {
let _ = self
.client
.abort_multipart_upload()
.bucket(&self.bucket)
.key(key)
.upload_id(&upload_id)
.send()
.await;
return Err(StorageError::Io(e));
}
None => break,
}
}
if !buffer.is_empty() {
let part = upload_part(
&self.client,
&self.bucket,
key,
&upload_id,
part_number,
buffer,
)
.await?;
completed_parts.push(part);
}
if completed_parts.is_empty() {
let _ = self
.client
.abort_multipart_upload()
.bucket(&self.bucket)
.key(key)
.upload_id(&upload_id)
.send()
.await;
return Err(StorageError::Other("Empty upload".to_string()));
}
let completed_upload = CompletedMultipartUpload::builder()
.set_parts(Some(completed_parts))
.build();
self.client
.complete_multipart_upload()
.bucket(&self.bucket)
.key(key)
.upload_id(&upload_id)
.multipart_upload(completed_upload)
.send()
.await
.map_err(|e| StorageError::S3(format!("Failed to complete multipart upload: {}", e)))?;
crate::metrics::record_s3_operation("put_stream", "success");
let hash: [u8; 32] = hasher.finalize().into();
Ok(StreamUploadResult {
sha256_hash: hash,
size: total_size,
})
}
async fn copy(&self, src_key: &str, dst_key: &str) -> Result<(), StorageError> {
let copy_source = format!("{}/{}", self.bucket, src_key);
self.client
.copy_object()
.bucket(&self.bucket)
.copy_source(&copy_source)
.key(dst_key)
.send()
.await
.map_err(|e| StorageError::S3(format!("Failed to copy object: {}", e)))?;
crate::metrics::record_s3_operation("copy", "success");
Ok(())
}
}
src/sync/util.rs
+8 -8
View File
@@ -216,7 +216,7 @@ fn format_identity_event(event: &SequencedEvent) -> Result<Vec<u8>, anyhow::Erro
op: 1,
t: "#identity".to_string(),
};
let mut bytes = Vec::new();
let mut bytes = Vec::with_capacity(256);
serde_ipld_dagcbor::to_writer(&mut bytes, &header)?;
serde_ipld_dagcbor::to_writer(&mut bytes, &frame)?;
Ok(bytes)
@@ -234,7 +234,7 @@ fn format_account_event(event: &SequencedEvent) -> Result<Vec<u8>, anyhow::Error
op: 1,
t: "#account".to_string(),
};
let mut bytes = Vec::new();
let mut bytes = Vec::with_capacity(256);
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();
@@ -281,7 +281,7 @@ async fn format_sync_event(
op: 1,
t: "#sync".to_string(),
};
let mut bytes = Vec::new();
let mut bytes = Vec::with_capacity(512);
serde_ipld_dagcbor::to_writer(&mut bytes, &header)?;
serde_ipld_dagcbor::to_writer(&mut bytes, &frame)?;
Ok(bytes)
@@ -349,7 +349,7 @@ pub async fn format_event_for_sending(
op: 1,
t: "#commit".to_string(),
};
let mut bytes = Vec::new();
let mut bytes = Vec::with_capacity(frame.blocks.len() + 512);
serde_ipld_dagcbor::to_writer(&mut bytes, &header)?;
serde_ipld_dagcbor::to_writer(&mut bytes, &frame)?;
Ok(bytes)
@@ -385,7 +385,7 @@ pub async fn prefetch_blocks_for_events(
return Ok(HashMap::new());
}
let fetched = state.block_store.get_many(&all_cids).await?;
let mut blocks_map = HashMap::new();
let mut blocks_map = HashMap::with_capacity(all_cids.len());
for (cid, data_opt) in all_cids.into_iter().zip(fetched.into_iter()) {
if let Some(data) = data_opt {
blocks_map.insert(cid, data);
@@ -497,7 +497,7 @@ pub async fn format_event_with_prefetched_blocks(
op: 1,
t: "#commit".to_string(),
};
let mut bytes = Vec::new();
let mut bytes = Vec::with_capacity(frame.blocks.len() + 512);
serde_ipld_dagcbor::to_writer(&mut bytes, &header)?;
serde_ipld_dagcbor::to_writer(&mut bytes, &frame)?;
Ok(bytes)
@@ -512,7 +512,7 @@ pub fn format_info_frame(name: &str, message: Option<&str>) -> Result<Vec<u8>, a
name: name.to_string(),
message: message.map(String::from),
};
let mut bytes = Vec::new();
let mut bytes = Vec::with_capacity(128);
serde_ipld_dagcbor::to_writer(&mut bytes, &header)?;
serde_ipld_dagcbor::to_writer(&mut bytes, &frame)?;
Ok(bytes)
@@ -524,7 +524,7 @@ pub fn format_error_frame(error: &str, message: Option<&str>) -> Result<Vec<u8>,
error: error.to_string(),
message: message.map(String::from),
};
let mut bytes = Vec::new();
let mut bytes = Vec::with_capacity(128);
serde_ipld_dagcbor::to_writer(&mut bytes, &header)?;
serde_ipld_dagcbor::to_writer(&mut bytes, &frame)?;
Ok(bytes)
src/sync/verify.rs
+3
View File
@@ -47,6 +47,9 @@ impl CarVerifier {
Self {
http_client: Client::builder()
.timeout(std::time::Duration::from_secs(10))
.connect_timeout(std::time::Duration::from_secs(5))
.pool_max_idle_per_host(10)
.pool_idle_timeout(std::time::Duration::from_secs(90))
.build()
.unwrap_or_default(),
}
src/sync/verify_tests.rs
+285 -288
View File
@@ -1,311 +1,308 @@
#[cfg(test)]
mod tests {
use crate::sync::verify::{CarVerifier, VerifyError};
use bytes::Bytes;
use cid::Cid;
use sha2::{Digest, Sha256};
use std::collections::HashMap;
use crate::sync::verify::{CarVerifier, VerifyError};
use bytes::Bytes;
use cid::Cid;
use sha2::{Digest, Sha256};
use std::collections::HashMap;
fn make_cid(data: &[u8]) -> Cid {
let mut hasher = Sha256::new();
hasher.update(data);
let hash = hasher.finalize();
let multihash = multihash::Multihash::wrap(0x12, &hash).unwrap();
Cid::new_v1(0x71, multihash)
}
fn make_cid(data: &[u8]) -> Cid {
let mut hasher = Sha256::new();
hasher.update(data);
let hash = hasher.finalize();
let multihash = multihash::Multihash::wrap(0x12, &hash).unwrap();
Cid::new_v1(0x71, multihash)
}
#[test]
fn test_verifier_creation() {
let _verifier = CarVerifier::new();
}
#[test]
fn test_verifier_creation() {
let _verifier = CarVerifier::new();
}
#[test]
fn test_verify_error_display() {
let err = VerifyError::DidMismatch {
commit_did: "did:plc:abc".to_string(),
expected_did: "did:plc:xyz".to_string(),
};
assert!(err.to_string().contains("did:plc:abc"));
assert!(err.to_string().contains("did:plc:xyz"));
let err = VerifyError::InvalidSignature;
assert!(err.to_string().contains("signature"));
let err = VerifyError::NoSigningKey;
assert!(err.to_string().contains("signing key"));
let err = VerifyError::MstValidationFailed("test error".to_string());
assert!(err.to_string().contains("test error"));
}
#[test]
fn test_verify_error_display() {
let err = VerifyError::DidMismatch {
commit_did: "did:plc:abc".to_string(),
expected_did: "did:plc:xyz".to_string(),
};
assert!(err.to_string().contains("did:plc:abc"));
assert!(err.to_string().contains("did:plc:xyz"));
let err = VerifyError::InvalidSignature;
assert!(err.to_string().contains("signature"));
let err = VerifyError::NoSigningKey;
assert!(err.to_string().contains("signing key"));
let err = VerifyError::MstValidationFailed("test error".to_string());
assert!(err.to_string().contains("test error"));
}
#[test]
fn test_mst_validation_missing_root_block() {
let verifier = CarVerifier::new();
let blocks: HashMap<Cid, Bytes> = HashMap::new();
let fake_cid = make_cid(b"fake data");
let result = verifier.verify_mst_structure(&fake_cid, &blocks);
assert!(result.is_err());
let err = result.unwrap_err();
assert!(matches!(err, VerifyError::BlockNotFound(_)));
}
#[test]
fn test_mst_validation_missing_root_block() {
let verifier = CarVerifier::new();
let blocks: HashMap<Cid, Bytes> = HashMap::new();
let fake_cid = make_cid(b"fake data");
let result = verifier.verify_mst_structure(&fake_cid, &blocks);
assert!(result.is_err());
let err = result.unwrap_err();
assert!(matches!(err, VerifyError::BlockNotFound(_)));
}
#[test]
fn test_mst_validation_invalid_cbor() {
let verifier = CarVerifier::new();
let bad_cbor = Bytes::from(vec![0xFF, 0xFF, 0xFF]);
let cid = make_cid(&bad_cbor);
let mut blocks = HashMap::new();
blocks.insert(cid, bad_cbor);
let result = verifier.verify_mst_structure(&cid, &blocks);
assert!(result.is_err());
let err = result.unwrap_err();
assert!(matches!(err, VerifyError::InvalidCbor(_)));
}
#[test]
fn test_mst_validation_invalid_cbor() {
let verifier = CarVerifier::new();
let bad_cbor = Bytes::from(vec![0xFF, 0xFF, 0xFF]);
let cid = make_cid(&bad_cbor);
let mut blocks = HashMap::new();
blocks.insert(cid, bad_cbor);
let result = verifier.verify_mst_structure(&cid, &blocks);
assert!(result.is_err());
let err = result.unwrap_err();
assert!(matches!(err, VerifyError::InvalidCbor(_)));
}
#[test]
fn test_mst_validation_empty_node() {
let verifier = CarVerifier::new();
let empty_node = serde_ipld_dagcbor::to_vec(&serde_json::json!({
"e": []
}))
.unwrap();
let cid = make_cid(&empty_node);
let mut blocks = HashMap::new();
blocks.insert(cid, Bytes::from(empty_node));
let result = verifier.verify_mst_structure(&cid, &blocks);
assert!(result.is_ok());
}
#[test]
fn test_mst_validation_empty_node() {
let verifier = CarVerifier::new();
let empty_node = serde_ipld_dagcbor::to_vec(&serde_json::json!({
"e": []
}))
.unwrap();
let cid = make_cid(&empty_node);
let mut blocks = HashMap::new();
blocks.insert(cid, Bytes::from(empty_node));
let result = verifier.verify_mst_structure(&cid, &blocks);
assert!(result.is_ok());
}
#[test]
fn test_mst_validation_missing_left_pointer() {
use ipld_core::ipld::Ipld;
#[test]
fn test_mst_validation_missing_left_pointer() {
use ipld_core::ipld::Ipld;
let verifier = CarVerifier::new();
let missing_left_cid = make_cid(b"missing left");
let node = Ipld::Map(std::collections::BTreeMap::from([
("l".to_string(), Ipld::Link(missing_left_cid)),
("e".to_string(), Ipld::List(vec![])),
]));
let node_bytes = serde_ipld_dagcbor::to_vec(&node).unwrap();
let cid = make_cid(&node_bytes);
let mut blocks = HashMap::new();
blocks.insert(cid, Bytes::from(node_bytes));
let result = verifier.verify_mst_structure(&cid, &blocks);
assert!(result.is_err());
let err = result.unwrap_err();
assert!(matches!(err, VerifyError::BlockNotFound(_)));
assert!(err.to_string().contains("left pointer"));
}
let verifier = CarVerifier::new();
let missing_left_cid = make_cid(b"missing left");
let node = Ipld::Map(std::collections::BTreeMap::from([
("l".to_string(), Ipld::Link(missing_left_cid)),
("e".to_string(), Ipld::List(vec![])),
]));
let node_bytes = serde_ipld_dagcbor::to_vec(&node).unwrap();
let cid = make_cid(&node_bytes);
let mut blocks = HashMap::new();
blocks.insert(cid, Bytes::from(node_bytes));
let result = verifier.verify_mst_structure(&cid, &blocks);
assert!(result.is_err());
let err = result.unwrap_err();
assert!(matches!(err, VerifyError::BlockNotFound(_)));
assert!(err.to_string().contains("left pointer"));
}
#[test]
fn test_mst_validation_missing_subtree() {
use ipld_core::ipld::Ipld;
#[test]
fn test_mst_validation_missing_subtree() {
use ipld_core::ipld::Ipld;
let verifier = CarVerifier::new();
let missing_subtree_cid = make_cid(b"missing subtree");
let record_cid = make_cid(b"record");
let entry = Ipld::Map(std::collections::BTreeMap::from([
("k".to_string(), Ipld::Bytes(b"key1".to_vec())),
("v".to_string(), Ipld::Link(record_cid)),
("p".to_string(), Ipld::Integer(0)),
("t".to_string(), Ipld::Link(missing_subtree_cid)),
]));
let node = Ipld::Map(std::collections::BTreeMap::from([(
"e".to_string(),
Ipld::List(vec![entry]),
)]));
let node_bytes = serde_ipld_dagcbor::to_vec(&node).unwrap();
let cid = make_cid(&node_bytes);
let mut blocks = HashMap::new();
blocks.insert(cid, Bytes::from(node_bytes));
let result = verifier.verify_mst_structure(&cid, &blocks);
assert!(result.is_err());
let err = result.unwrap_err();
assert!(matches!(err, VerifyError::BlockNotFound(_)));
assert!(err.to_string().contains("subtree"));
}
let verifier = CarVerifier::new();
let missing_subtree_cid = make_cid(b"missing subtree");
let record_cid = make_cid(b"record");
let entry = Ipld::Map(std::collections::BTreeMap::from([
("k".to_string(), Ipld::Bytes(b"key1".to_vec())),
("v".to_string(), Ipld::Link(record_cid)),
("p".to_string(), Ipld::Integer(0)),
("t".to_string(), Ipld::Link(missing_subtree_cid)),
]));
let node = Ipld::Map(std::collections::BTreeMap::from([(
"e".to_string(),
Ipld::List(vec![entry]),
)]));
let node_bytes = serde_ipld_dagcbor::to_vec(&node).unwrap();
let cid = make_cid(&node_bytes);
let mut blocks = HashMap::new();
blocks.insert(cid, Bytes::from(node_bytes));
let result = verifier.verify_mst_structure(&cid, &blocks);
assert!(result.is_err());
let err = result.unwrap_err();
assert!(matches!(err, VerifyError::BlockNotFound(_)));
assert!(err.to_string().contains("subtree"));
}
#[test]
fn test_mst_validation_unsorted_keys() {
use ipld_core::ipld::Ipld;
#[test]
fn test_mst_validation_unsorted_keys() {
use ipld_core::ipld::Ipld;
let verifier = CarVerifier::new();
let record_cid = make_cid(b"record");
let entry1 = Ipld::Map(std::collections::BTreeMap::from([
("k".to_string(), Ipld::Bytes(b"zzz".to_vec())),
("v".to_string(), Ipld::Link(record_cid)),
("p".to_string(), Ipld::Integer(0)),
]));
let entry2 = Ipld::Map(std::collections::BTreeMap::from([
("k".to_string(), Ipld::Bytes(b"aaa".to_vec())),
("v".to_string(), Ipld::Link(record_cid)),
("p".to_string(), Ipld::Integer(0)),
]));
let node = Ipld::Map(std::collections::BTreeMap::from([(
"e".to_string(),
Ipld::List(vec![entry1, entry2]),
)]));
let node_bytes = serde_ipld_dagcbor::to_vec(&node).unwrap();
let cid = make_cid(&node_bytes);
let mut blocks = HashMap::new();
blocks.insert(cid, Bytes::from(node_bytes));
let result = verifier.verify_mst_structure(&cid, &blocks);
assert!(result.is_err());
let err = result.unwrap_err();
assert!(matches!(err, VerifyError::MstValidationFailed(_)));
assert!(err.to_string().contains("sorted"));
}
let verifier = CarVerifier::new();
let record_cid = make_cid(b"record");
let entry1 = Ipld::Map(std::collections::BTreeMap::from([
("k".to_string(), Ipld::Bytes(b"zzz".to_vec())),
("v".to_string(), Ipld::Link(record_cid)),
("p".to_string(), Ipld::Integer(0)),
]));
let entry2 = Ipld::Map(std::collections::BTreeMap::from([
("k".to_string(), Ipld::Bytes(b"aaa".to_vec())),
("v".to_string(), Ipld::Link(record_cid)),
("p".to_string(), Ipld::Integer(0)),
]));
let node = Ipld::Map(std::collections::BTreeMap::from([(
"e".to_string(),
Ipld::List(vec![entry1, entry2]),
)]));
let node_bytes = serde_ipld_dagcbor::to_vec(&node).unwrap();
let cid = make_cid(&node_bytes);
let mut blocks = HashMap::new();
blocks.insert(cid, Bytes::from(node_bytes));
let result = verifier.verify_mst_structure(&cid, &blocks);
assert!(result.is_err());
let err = result.unwrap_err();
assert!(matches!(err, VerifyError::MstValidationFailed(_)));
assert!(err.to_string().contains("sorted"));
}
#[test]
fn test_mst_validation_sorted_keys_ok() {
use ipld_core::ipld::Ipld;
#[test]
fn test_mst_validation_sorted_keys_ok() {
use ipld_core::ipld::Ipld;
let verifier = CarVerifier::new();
let record_cid = make_cid(b"record");
let entry1 = Ipld::Map(std::collections::BTreeMap::from([
("k".to_string(), Ipld::Bytes(b"aaa".to_vec())),
("v".to_string(), Ipld::Link(record_cid)),
("p".to_string(), Ipld::Integer(0)),
]));
let entry2 = Ipld::Map(std::collections::BTreeMap::from([
("k".to_string(), Ipld::Bytes(b"bbb".to_vec())),
("v".to_string(), Ipld::Link(record_cid)),
("p".to_string(), Ipld::Integer(0)),
]));
let entry3 = Ipld::Map(std::collections::BTreeMap::from([
("k".to_string(), Ipld::Bytes(b"zzz".to_vec())),
("v".to_string(), Ipld::Link(record_cid)),
("p".to_string(), Ipld::Integer(0)),
]));
let node = Ipld::Map(std::collections::BTreeMap::from([(
"e".to_string(),
Ipld::List(vec![entry1, entry2, entry3]),
)]));
let node_bytes = serde_ipld_dagcbor::to_vec(&node).unwrap();
let cid = make_cid(&node_bytes);
let mut blocks = HashMap::new();
blocks.insert(cid, Bytes::from(node_bytes));
let result = verifier.verify_mst_structure(&cid, &blocks);
assert!(result.is_ok());
}
let verifier = CarVerifier::new();
let record_cid = make_cid(b"record");
let entry1 = Ipld::Map(std::collections::BTreeMap::from([
("k".to_string(), Ipld::Bytes(b"aaa".to_vec())),
("v".to_string(), Ipld::Link(record_cid)),
("p".to_string(), Ipld::Integer(0)),
]));
let entry2 = Ipld::Map(std::collections::BTreeMap::from([
("k".to_string(), Ipld::Bytes(b"bbb".to_vec())),
("v".to_string(), Ipld::Link(record_cid)),
("p".to_string(), Ipld::Integer(0)),
]));
let entry3 = Ipld::Map(std::collections::BTreeMap::from([
("k".to_string(), Ipld::Bytes(b"zzz".to_vec())),
("v".to_string(), Ipld::Link(record_cid)),
("p".to_string(), Ipld::Integer(0)),
]));
let node = Ipld::Map(std::collections::BTreeMap::from([(
"e".to_string(),
Ipld::List(vec![entry1, entry2, entry3]),
)]));
let node_bytes = serde_ipld_dagcbor::to_vec(&node).unwrap();
let cid = make_cid(&node_bytes);
let mut blocks = HashMap::new();
blocks.insert(cid, Bytes::from(node_bytes));
let result = verifier.verify_mst_structure(&cid, &blocks);
assert!(result.is_ok());
}
#[test]
fn test_mst_validation_with_valid_left_pointer() {
use ipld_core::ipld::Ipld;
#[test]
fn test_mst_validation_with_valid_left_pointer() {
use ipld_core::ipld::Ipld;
let verifier = CarVerifier::new();
let left_node = Ipld::Map(std::collections::BTreeMap::from([(
"e".to_string(),
Ipld::List(vec![]),
)]));
let left_node_bytes = serde_ipld_dagcbor::to_vec(&left_node).unwrap();
let left_cid = make_cid(&left_node_bytes);
let root_node = Ipld::Map(std::collections::BTreeMap::from([
("l".to_string(), Ipld::Link(left_cid)),
("e".to_string(), Ipld::List(vec![])),
]));
let root_node_bytes = serde_ipld_dagcbor::to_vec(&root_node).unwrap();
let root_cid = make_cid(&root_node_bytes);
let mut blocks = HashMap::new();
blocks.insert(root_cid, Bytes::from(root_node_bytes));
blocks.insert(left_cid, Bytes::from(left_node_bytes));
let result = verifier.verify_mst_structure(&root_cid, &blocks);
assert!(result.is_ok());
}
let verifier = CarVerifier::new();
let left_node = Ipld::Map(std::collections::BTreeMap::from([(
"e".to_string(),
Ipld::List(vec![]),
)]));
let left_node_bytes = serde_ipld_dagcbor::to_vec(&left_node).unwrap();
let left_cid = make_cid(&left_node_bytes);
let root_node = Ipld::Map(std::collections::BTreeMap::from([
("l".to_string(), Ipld::Link(left_cid)),
("e".to_string(), Ipld::List(vec![])),
]));
let root_node_bytes = serde_ipld_dagcbor::to_vec(&root_node).unwrap();
let root_cid = make_cid(&root_node_bytes);
let mut blocks = HashMap::new();
blocks.insert(root_cid, Bytes::from(root_node_bytes));
blocks.insert(left_cid, Bytes::from(left_node_bytes));
let result = verifier.verify_mst_structure(&root_cid, &blocks);
assert!(result.is_ok());
}
#[test]
fn test_mst_validation_cycle_detection() {
let verifier = CarVerifier::new();
let node = serde_ipld_dagcbor::to_vec(&serde_json::json!({
"e": []
}))
.unwrap();
let cid = make_cid(&node);
let mut blocks = HashMap::new();
blocks.insert(cid, Bytes::from(node));
let result = verifier.verify_mst_structure(&cid, &blocks);
assert!(result.is_ok());
}
#[test]
fn test_mst_validation_cycle_detection() {
let verifier = CarVerifier::new();
let node = serde_ipld_dagcbor::to_vec(&serde_json::json!({
"e": []
}))
.unwrap();
let cid = make_cid(&node);
let mut blocks = HashMap::new();
blocks.insert(cid, Bytes::from(node));
let result = verifier.verify_mst_structure(&cid, &blocks);
assert!(result.is_ok());
}
#[tokio::test]
async fn test_unsupported_did_method() {
let verifier = CarVerifier::new();
let result = verifier.resolve_did_document("did:unknown:test").await;
assert!(result.is_err());
let err = result.unwrap_err();
assert!(matches!(err, VerifyError::DidResolutionFailed(_)));
assert!(err.to_string().contains("Unsupported"));
}
#[tokio::test]
async fn test_unsupported_did_method() {
let verifier = CarVerifier::new();
let result = verifier.resolve_did_document("did:unknown:test").await;
assert!(result.is_err());
let err = result.unwrap_err();
assert!(matches!(err, VerifyError::DidResolutionFailed(_)));
assert!(err.to_string().contains("Unsupported"));
}
#[test]
fn test_mst_validation_with_prefix_compression() {
use ipld_core::ipld::Ipld;
#[test]
fn test_mst_validation_with_prefix_compression() {
use ipld_core::ipld::Ipld;
let verifier = CarVerifier::new();
let record_cid = make_cid(b"record");
let entry1 = Ipld::Map(std::collections::BTreeMap::from([
(
"k".to_string(),
Ipld::Bytes(b"app.bsky.feed.post/abc".to_vec()),
),
("v".to_string(), Ipld::Link(record_cid)),
("p".to_string(), Ipld::Integer(0)),
]));
let entry2 = Ipld::Map(std::collections::BTreeMap::from([
("k".to_string(), Ipld::Bytes(b"def".to_vec())),
("v".to_string(), Ipld::Link(record_cid)),
("p".to_string(), Ipld::Integer(19)),
]));
let entry3 = Ipld::Map(std::collections::BTreeMap::from([
("k".to_string(), Ipld::Bytes(b"xyz".to_vec())),
("v".to_string(), Ipld::Link(record_cid)),
("p".to_string(), Ipld::Integer(19)),
]));
let node = Ipld::Map(std::collections::BTreeMap::from([(
"e".to_string(),
Ipld::List(vec![entry1, entry2, entry3]),
)]));
let node_bytes = serde_ipld_dagcbor::to_vec(&node).unwrap();
let cid = make_cid(&node_bytes);
let mut blocks = HashMap::new();
blocks.insert(cid, Bytes::from(node_bytes));
let result = verifier.verify_mst_structure(&cid, &blocks);
assert!(
result.is_ok(),
"Prefix-compressed keys should be validated correctly"
);
}
let verifier = CarVerifier::new();
let record_cid = make_cid(b"record");
let entry1 = Ipld::Map(std::collections::BTreeMap::from([
(
"k".to_string(),
Ipld::Bytes(b"app.bsky.feed.post/abc".to_vec()),
),
("v".to_string(), Ipld::Link(record_cid)),
("p".to_string(), Ipld::Integer(0)),
]));
let entry2 = Ipld::Map(std::collections::BTreeMap::from([
("k".to_string(), Ipld::Bytes(b"def".to_vec())),
("v".to_string(), Ipld::Link(record_cid)),
("p".to_string(), Ipld::Integer(19)),
]));
let entry3 = Ipld::Map(std::collections::BTreeMap::from([
("k".to_string(), Ipld::Bytes(b"xyz".to_vec())),
("v".to_string(), Ipld::Link(record_cid)),
("p".to_string(), Ipld::Integer(19)),
]));
let node = Ipld::Map(std::collections::BTreeMap::from([(
"e".to_string(),
Ipld::List(vec![entry1, entry2, entry3]),
)]));
let node_bytes = serde_ipld_dagcbor::to_vec(&node).unwrap();
let cid = make_cid(&node_bytes);
let mut blocks = HashMap::new();
blocks.insert(cid, Bytes::from(node_bytes));
let result = verifier.verify_mst_structure(&cid, &blocks);
assert!(
result.is_ok(),
"Prefix-compressed keys should be validated correctly"
);
}
#[test]
fn test_mst_validation_prefix_compression_unsorted() {
use ipld_core::ipld::Ipld;
#[test]
fn test_mst_validation_prefix_compression_unsorted() {
use ipld_core::ipld::Ipld;
let verifier = CarVerifier::new();
let record_cid = make_cid(b"record");
let entry1 = Ipld::Map(std::collections::BTreeMap::from([
(
"k".to_string(),
Ipld::Bytes(b"app.bsky.feed.post/xyz".to_vec()),
),
("v".to_string(), Ipld::Link(record_cid)),
("p".to_string(), Ipld::Integer(0)),
]));
let entry2 = Ipld::Map(std::collections::BTreeMap::from([
("k".to_string(), Ipld::Bytes(b"abc".to_vec())),
("v".to_string(), Ipld::Link(record_cid)),
("p".to_string(), Ipld::Integer(19)),
]));
let node = Ipld::Map(std::collections::BTreeMap::from([(
"e".to_string(),
Ipld::List(vec![entry1, entry2]),
)]));
let node_bytes = serde_ipld_dagcbor::to_vec(&node).unwrap();
let cid = make_cid(&node_bytes);
let mut blocks = HashMap::new();
blocks.insert(cid, Bytes::from(node_bytes));
let result = verifier.verify_mst_structure(&cid, &blocks);
assert!(
result.is_err(),
"Unsorted prefix-compressed keys should fail validation"
);
let err = result.unwrap_err();
assert!(matches!(err, VerifyError::MstValidationFailed(_)));
}
let verifier = CarVerifier::new();
let record_cid = make_cid(b"record");
let entry1 = Ipld::Map(std::collections::BTreeMap::from([
(
"k".to_string(),
Ipld::Bytes(b"app.bsky.feed.post/xyz".to_vec()),
),
("v".to_string(), Ipld::Link(record_cid)),
("p".to_string(), Ipld::Integer(0)),
]));
let entry2 = Ipld::Map(std::collections::BTreeMap::from([
("k".to_string(), Ipld::Bytes(b"abc".to_vec())),
("v".to_string(), Ipld::Link(record_cid)),
("p".to_string(), Ipld::Integer(19)),
]));
let node = Ipld::Map(std::collections::BTreeMap::from([(
"e".to_string(),
Ipld::List(vec![entry1, entry2]),
)]));
let node_bytes = serde_ipld_dagcbor::to_vec(&node).unwrap();
let cid = make_cid(&node_bytes);
let mut blocks = HashMap::new();
blocks.insert(cid, Bytes::from(node_bytes));
let result = verifier.verify_mst_structure(&cid, &blocks);
assert!(
result.is_err(),
"Unsorted prefix-compressed keys should fail validation"
);
let err = result.unwrap_err();
assert!(matches!(err, VerifyError::MstValidationFailed(_)));
}