diff --git a/crates/tranquil-api/src/actor/preferences.rs b/crates/tranquil-api/src/actor/preferences.rs index 4c63e66..bff877c 100644 --- a/crates/tranquil-api/src/actor/preferences.rs +++ b/crates/tranquil-api/src/actor/preferences.rs @@ -34,13 +34,13 @@ pub struct GetPreferencesOutput { } pub async fn get_preferences(State(state): State, auth: Auth) -> Response { let has_full_access = auth.permissions().has_full_access(); - let user_id: uuid::Uuid = match state.user_repo.get_id_by_did(&auth.did).await { + let user_id: uuid::Uuid = match state.repos.user.get_id_by_did(&auth.did).await { Ok(Some(id)) => id, _ => { return ApiError::InternalError(Some("User not found".into())).into_response(); } }; - let prefs = match state.infra_repo.get_account_preferences(user_id).await { + let prefs = match state.repos.infra.get_account_preferences(user_id).await { Ok(rows) => rows, Err(_) => { return ApiError::InternalError(Some("Failed to fetch preferences".into())) @@ -93,7 +93,7 @@ pub async fn put_preferences( Json(input): Json, ) -> Response { let has_full_access = auth.permissions().has_full_access(); - let user_id: uuid::Uuid = match state.user_repo.get_id_by_did(&auth.did).await { + let user_id: uuid::Uuid = match state.repos.user.get_id_by_did(&auth.did).await { Ok(Some(id)) => id, _ => { return ApiError::InternalError(Some("User not found".into())).into_response(); @@ -188,7 +188,7 @@ pub async fn put_preferences( .collect(); if state - .infra_repo + .repos.infra .replace_namespace_preferences(user_id, APP_BSKY_NAMESPACE, prefs_to_save) .await .is_err() diff --git a/crates/tranquil-api/src/admin/account/delete.rs b/crates/tranquil-api/src/admin/account/delete.rs index fb4e7b0..e3256ed 100644 --- a/crates/tranquil-api/src/admin/account/delete.rs +++ b/crates/tranquil-api/src/admin/account/delete.rs @@ -19,7 +19,7 @@ pub async fn delete_account( ) -> Result, ApiError> { let did = &input.did; let (user_id, handle) = state - .user_repo + .repos.user .get_id_and_handle_by_did(did) .await .log_db_err("in delete_account")? @@ -27,7 +27,7 @@ pub async fn delete_account( .map(|row| (row.id, row.handle))?; state - .user_repo + .repos.user .admin_delete_account_complete(user_id, did) .await .log_db_err("deleting account")?; diff --git a/crates/tranquil-api/src/admin/account/email.rs b/crates/tranquil-api/src/admin/account/email.rs index dbf0e17..8dc6929 100644 --- a/crates/tranquil-api/src/admin/account/email.rs +++ b/crates/tranquil-api/src/admin/account/email.rs @@ -31,7 +31,7 @@ pub async fn send_email( return Err(ApiError::InvalidRequest("content is required".into())); } let user = state - .user_repo + .repos.user .get_by_did(&input.recipient_did) .await .log_db_err("in send_email")? @@ -45,7 +45,7 @@ pub async fn send_email( .clone() .unwrap_or_else(|| format!("Message from {}", hostname)); let result = state - .infra_repo + .repos.infra .enqueue_comms( Some(user_id), tranquil_db_traits::CommsChannel::Email, diff --git a/crates/tranquil-api/src/admin/account/info.rs b/crates/tranquil-api/src/admin/account/info.rs index 1b75f21..2209b03 100644 --- a/crates/tranquil-api/src/admin/account/info.rs +++ b/crates/tranquil-api/src/admin/account/info.rs @@ -69,7 +69,7 @@ pub async fn get_account_info( Query(params): Query, ) -> Result, ApiError> { let account = state - .infra_repo + .repos.infra .get_admin_account_info_by_did(¶ms.did) .await .log_db_err("in get_account_info")? @@ -98,7 +98,7 @@ pub async fn get_account_info( async fn get_invited_by(state: &AppState, user_id: uuid::Uuid) -> Option { let code = state - .infra_repo + .repos.infra .get_invite_code_used_by_user(user_id) .await .ok()??; @@ -111,7 +111,7 @@ async fn get_invites_for_user( user_id: uuid::Uuid, ) -> Option> { let invite_codes = state - .infra_repo + .repos.infra .get_invites_created_by_user(user_id) .await .ok()?; @@ -123,7 +123,7 @@ async fn get_invites_for_user( let code_strings: Vec = invite_codes.iter().map(|ic| ic.code.clone()).collect(); let uses = state - .infra_repo + .repos.infra .get_invite_code_uses_batch(&code_strings) .await .ok()?; @@ -154,10 +154,10 @@ async fn get_invites_for_user( } async fn get_invite_code_info(state: &AppState, code: &str) -> Option { - let info = state.infra_repo.get_invite_code_info(code).await.ok()??; + let info = state.repos.infra.get_invite_code_info(code).await.ok()??; let uses = state - .infra_repo + .repos.infra .get_invite_code_uses(code) .await .ok() @@ -197,7 +197,7 @@ pub async fn get_account_infos( let dids_typed: Vec = dids.iter().filter_map(|d| d.parse().ok()).collect(); let accounts = state - .infra_repo + .repos.infra .get_admin_account_infos_by_dids(&dids_typed) .await .log_db_err("fetching account infos")?; @@ -205,7 +205,7 @@ pub async fn get_account_infos( let user_ids: Vec = accounts.iter().map(|u| u.id).collect(); let all_invite_codes = state - .infra_repo + .repos.infra .get_invite_codes_by_users(&user_ids) .await .unwrap_or_default(); @@ -217,7 +217,7 @@ pub async fn get_account_infos( let all_invite_uses = if !all_codes.is_empty() { state - .infra_repo + .repos.infra .get_invite_code_uses_batch(&all_codes) .await .unwrap_or_default() @@ -226,7 +226,7 @@ pub async fn get_account_infos( }; let invited_by_map: HashMap = state - .infra_repo + .repos.infra .get_invite_code_uses_by_users(&user_ids) .await .unwrap_or_default() diff --git a/crates/tranquil-api/src/admin/account/search.rs b/crates/tranquil-api/src/admin/account/search.rs index f96e2f8..1128988 100644 --- a/crates/tranquil-api/src/admin/account/search.rs +++ b/crates/tranquil-api/src/admin/account/search.rs @@ -55,7 +55,7 @@ pub async fn search_accounts( let handle_filter = params.handle.as_deref().map(|h| format!("%{}%", h)); let cursor_did: Option = params.cursor.as_ref().and_then(|c| c.parse().ok()); let rows = state - .user_repo + .repos.user .search_accounts( cursor_did.as_ref(), email_filter.as_deref(), diff --git a/crates/tranquil-api/src/admin/account/update.rs b/crates/tranquil-api/src/admin/account/update.rs index 791a9af..4b8624d 100644 --- a/crates/tranquil-api/src/admin/account/update.rs +++ b/crates/tranquil-api/src/admin/account/update.rs @@ -30,7 +30,7 @@ pub async fn update_account_email( .map_err(|_| ApiError::InvalidDid("Invalid DID format".into()))?; match state - .user_repo + .repos.user .admin_update_email(&account_did, email) .await { @@ -71,9 +71,9 @@ pub async fn update_account_handle( } else { input_handle.to_string() }; - let old_handle = state.user_repo.get_handle_by_did(did).await.ok().flatten(); + let old_handle = state.repos.user.get_handle_by_did(did).await.ok().flatten(); let user_id = state - .user_repo + .repos.user .get_id_by_did(did) .await .ok() @@ -81,14 +81,14 @@ pub async fn update_account_handle( .ok_or(ApiError::AccountNotFound)?; let handle_for_check: Handle = handle.parse().map_err(|_| ApiError::InvalidHandle(None))?; if let Ok(true) = state - .user_repo + .repos.user .check_handle_exists(&handle_for_check, user_id) .await { return Err(ApiError::HandleTaken); } match state - .user_repo + .repos.user .admin_update_handle(did, &handle_for_check) .await { @@ -146,13 +146,10 @@ pub async fn update_account_password( if password.is_empty() { return Err(ApiError::InvalidRequest("password is required".into())); } - let password_hash = bcrypt::hash(password, bcrypt::DEFAULT_COST).map_err(|e| { - error!("Failed to hash password: {:?}", e); - ApiError::InternalError(None) - })?; + let password_hash = crate::common::hash_or_internal_error(password)?; match state - .user_repo + .repos.user .admin_update_password(did, &password_hash) .await { diff --git a/crates/tranquil-api/src/admin/config.rs b/crates/tranquil-api/src/admin/config.rs index e43d61a..4314046 100644 --- a/crates/tranquil-api/src/admin/config.rs +++ b/crates/tranquil-api/src/admin/config.rs @@ -53,7 +53,7 @@ pub async fn get_server_config( ]; let rows = state - .infra_repo + .repos.infra .get_server_configs(keys) .await .log_db_err("fetching server config")?; @@ -86,7 +86,7 @@ pub async fn update_server_config( )); } state - .infra_repo + .repos.infra .upsert_server_config("server_name", trimmed) .await .log_db_err("upserting server_name")?; @@ -95,13 +95,13 @@ pub async fn update_server_config( if let Some(ref color) = req.primary_color { if color.is_empty() { state - .infra_repo + .repos.infra .delete_server_config("primary_color") .await .log_db_err("deleting primary_color")?; } else if is_valid_hex_color(color) { state - .infra_repo + .repos.infra .upsert_server_config("primary_color", color) .await .log_db_err("upserting primary_color")?; @@ -115,13 +115,13 @@ pub async fn update_server_config( if let Some(ref color) = req.primary_color_dark { if color.is_empty() { state - .infra_repo + .repos.infra .delete_server_config("primary_color_dark") .await .log_db_err("deleting primary_color_dark")?; } else if is_valid_hex_color(color) { state - .infra_repo + .repos.infra .upsert_server_config("primary_color_dark", color) .await .log_db_err("upserting primary_color_dark")?; @@ -135,13 +135,13 @@ pub async fn update_server_config( if let Some(ref color) = req.secondary_color { if color.is_empty() { state - .infra_repo + .repos.infra .delete_server_config("secondary_color") .await .log_db_err("deleting secondary_color")?; } else if is_valid_hex_color(color) { state - .infra_repo + .repos.infra .upsert_server_config("secondary_color", color) .await .log_db_err("upserting secondary_color")?; @@ -155,13 +155,13 @@ pub async fn update_server_config( if let Some(ref color) = req.secondary_color_dark { if color.is_empty() { state - .infra_repo + .repos.infra .delete_server_config("secondary_color_dark") .await .log_db_err("deleting secondary_color_dark")?; } else if is_valid_hex_color(color) { state - .infra_repo + .repos.infra .upsert_server_config("secondary_color_dark", color) .await .log_db_err("upserting secondary_color_dark")?; @@ -174,7 +174,7 @@ pub async fn update_server_config( if let Some(ref logo_cid) = req.logo_cid { let old_logo_cid = state - .infra_repo + .repos.infra .get_server_config("logo_cid") .await .ok() @@ -190,12 +190,12 @@ pub async fn update_server_config( match CidLink::new(old_cid_str) { Ok(old_cid) => { if let Ok(Some(storage_key)) = - state.infra_repo.get_blob_storage_key_by_cid(&old_cid).await + state.repos.infra.get_blob_storage_key_by_cid(&old_cid).await { if let Err(e) = state.blob_store.delete(&storage_key).await { error!("Failed to delete old logo blob from storage: {:?}", e); } - if let Err(e) = state.infra_repo.delete_blob_by_cid(&old_cid).await { + if let Err(e) = state.repos.infra.delete_blob_by_cid(&old_cid).await { error!("Failed to delete old logo blob record: {:?}", e); } } @@ -211,13 +211,13 @@ pub async fn update_server_config( if logo_cid.is_empty() { state - .infra_repo + .repos.infra .delete_server_config("logo_cid") .await .log_db_err("deleting logo_cid")?; } else { state - .infra_repo + .repos.infra .upsert_server_config("logo_cid", logo_cid) .await .log_db_err("upserting logo_cid")?; diff --git a/crates/tranquil-api/src/admin/invite.rs b/crates/tranquil-api/src/admin/invite.rs index 26d02af..07b6cfc 100644 --- a/crates/tranquil-api/src/admin/invite.rs +++ b/crates/tranquil-api/src/admin/invite.rs @@ -24,7 +24,7 @@ pub async fn disable_invite_codes( Json(input): Json, ) -> Result, ApiError> { if let Some(codes) = &input.codes - && let Err(e) = state.infra_repo.disable_invite_codes_by_code(codes).await + && let Err(e) = state.repos.infra.disable_invite_codes_by_code(codes).await { error!("DB error disabling invite codes: {:?}", e); } @@ -32,7 +32,7 @@ pub async fn disable_invite_codes( let accounts_typed: Vec = accounts.iter().filter_map(|a| a.parse().ok()).collect(); if let Err(e) = state - .infra_repo + .repos.infra .disable_invite_codes_by_account(&accounts_typed) .await { @@ -87,7 +87,7 @@ pub async fn get_invite_codes( }; let codes_rows = state - .infra_repo + .repos.infra .list_invite_codes(params.cursor.as_deref(), limit, sort_order) .await .log_db_err("fetching invite codes")?; @@ -96,7 +96,7 @@ pub async fn get_invite_codes( let code_strings: Vec = codes_rows.iter().map(|r| r.code.clone()).collect(); let creator_dids: std::collections::HashMap = state - .infra_repo + .repos.infra .get_user_dids_by_ids(&user_ids) .await .unwrap_or_default() @@ -108,7 +108,7 @@ pub async fn get_invite_codes( } else { common::group_invite_uses_by_code( state - .infra_repo + .repos.infra .get_invite_code_uses_batch(&code_strings) .await .unwrap_or_default(), @@ -168,7 +168,7 @@ pub async fn disable_account_invites( .map_err(|_| ApiError::InvalidDid("Invalid DID format".into()))?; match state - .user_repo + .repos.user .set_invites_disabled(&account_did, true) .await { @@ -200,7 +200,7 @@ pub async fn enable_account_invites( .map_err(|_| ApiError::InvalidDid("Invalid DID format".into()))?; match state - .user_repo + .repos.user .set_invites_disabled(&account_did, false) .await { diff --git a/crates/tranquil-api/src/admin/server_stats.rs b/crates/tranquil-api/src/admin/server_stats.rs index f5ead06..28e90cb 100644 --- a/crates/tranquil-api/src/admin/server_stats.rs +++ b/crates/tranquil-api/src/admin/server_stats.rs @@ -17,10 +17,10 @@ pub async fn get_server_stats( State(state): State, _auth: Auth, ) -> Result, ApiError> { - let user_count = state.user_repo.count_users().await.unwrap_or(0); - let repo_count = state.repo_repo.count_repos().await.unwrap_or(0); - let record_count = state.repo_repo.count_all_records().await.unwrap_or(0); - let blob_storage_bytes = state.blob_repo.sum_blob_storage().await.unwrap_or(0); + let user_count = state.repos.user.count_users().await.unwrap_or(0); + let repo_count = state.repos.repo.count_repos().await.unwrap_or(0); + let record_count = state.repos.repo.count_all_records().await.unwrap_or(0); + let blob_storage_bytes = state.repos.blob.sum_blob_storage().await.unwrap_or(0); Ok(Json(ServerStatsOutput { user_count, diff --git a/crates/tranquil-api/src/admin/status.rs b/crates/tranquil-api/src/admin/status.rs index 227de4a..bdf086f 100644 --- a/crates/tranquil-api/src/admin/status.rs +++ b/crates/tranquil-api/src/admin/status.rs @@ -45,7 +45,7 @@ pub async fn get_subject_status( let did: Did = did_str .parse() .map_err(|_| ApiError::InvalidDid("Invalid DID format".into()))?; - match state.user_repo.get_status_by_did(&did).await { + match state.repos.user.get_status_by_did(&did).await { Ok(Some(status)) => { let deactivated = status.deactivated_at.map(|_| StatusAttr { applied: true, @@ -77,7 +77,7 @@ pub async fn get_subject_status( let cid: CidLink = uri_str .parse() .map_err(|_| ApiError::InvalidRequest("Invalid CID format".into()))?; - match state.repo_repo.get_record_by_cid(&cid).await { + match state.repos.repo.get_record_by_cid(&cid).await { Ok(Some(record)) => { let takedown = record.takedown_ref.as_ref().map(|r| StatusAttr { applied: true, @@ -109,7 +109,7 @@ pub async fn get_subject_status( let did = params.did.as_ref().ok_or_else(|| { ApiError::InvalidRequest("Must provide a did to request blob state".into()) })?; - match state.blob_repo.get_blob_with_takedown(&blob_cid).await { + match state.repos.blob.get_blob_with_takedown(&blob_cid).await { Ok(Some(blob)) => { let takedown = blob.takedown_ref.as_ref().map(|r| StatusAttr { applied: true, @@ -172,7 +172,7 @@ pub async fn update_subject_status( None }; state - .user_repo + .repos.user .set_user_takedown(&did, takedown_ref) .await .map_err(|e| { @@ -182,9 +182,9 @@ pub async fn update_subject_status( } if let Some(deactivated) = &input.deactivated { let result = if deactivated.applied { - state.user_repo.deactivate_account(&did, None).await + state.repos.user.deactivate_account(&did, None).await } else { - state.user_repo.activate_account(&did).await + state.repos.user.activate_account(&did).await }; result.map_err(|e| { error!( @@ -218,7 +218,7 @@ pub async fn update_subject_status( warn!("Failed to sequence account event for deactivation: {}", e); } } - if let Ok(Some(handle)) = state.user_repo.get_handle_by_did(&did).await { + if let Ok(Some(handle)) = state.repos.user.get_handle_by_did(&did).await { let _ = state .cache .delete(&tranquil_pds::cache_keys::handle_key(&handle)) @@ -249,7 +249,7 @@ pub async fn update_subject_status( None }; state - .repo_repo + .repos.repo .set_record_takedown(&cid, takedown_ref) .await .map_err(|e| { @@ -282,7 +282,7 @@ pub async fn update_subject_status( None }; state - .blob_repo + .repos.blob .update_blob_takedown(&cid, takedown_ref) .await .map_err(|e| { diff --git a/crates/tranquil-api/src/age_assurance.rs b/crates/tranquil-api/src/age_assurance.rs index 6a528f3..f92dcc7 100644 --- a/crates/tranquil-api/src/age_assurance.rs +++ b/crates/tranquil-api/src/age_assurance.rs @@ -68,8 +68,8 @@ async fn get_account_created_at(state: &AppState, headers: &HeaderMap) -> Option let http_uri = "/"; let auth_user = match validate_token_with_dpop( - state.user_repo.as_ref(), - state.oauth_repo.as_ref(), + state.repos.user.as_ref(), + state.repos.oauth.as_ref(), &extracted.token, extracted.scheme, dpop_proof, @@ -89,7 +89,7 @@ async fn get_account_created_at(state: &AppState, headers: &HeaderMap) -> Option } }; - match state.user_repo.get_by_did(&auth_user.did).await { + match state.repos.user.get_by_did(&auth_user.did).await { Ok(Some(user)) => { tracing::debug!(created_at = ?user.created_at, "age assurance: got user"); Some(user.created_at.to_rfc3339()) diff --git a/crates/tranquil-api/src/common.rs b/crates/tranquil-api/src/common.rs index 8666307..fd9a6ea 100644 --- a/crates/tranquil-api/src/common.rs +++ b/crates/tranquil-api/src/common.rs @@ -1,4 +1,4 @@ -use bcrypt::DEFAULT_COST; +use bcrypt::{DEFAULT_COST, hash}; use chrono::{DateTime, Utc}; use std::collections::HashMap; use tracing::error; @@ -245,6 +245,20 @@ pub fn hash_or_internal_error(value: &str) -> Result { }) } +pub async fn hash_password_async(password: &str) -> Result { + let password = password.to_string(); + tokio::task::spawn_blocking(move || hash(password, DEFAULT_COST)) + .await + .map_err(|e| { + error!("Failed to spawn blocking task: {:?}", e); + ApiError::InternalError(None) + })? + .map_err(|e| { + error!("Failed to hash password: {:?}", e); + ApiError::InternalError(None) + }) +} + pub fn validate_token_hash( expires_at: Option>, stored_hash: &str, diff --git a/crates/tranquil-api/src/delegation.rs b/crates/tranquil-api/src/delegation.rs index 26cd40e..9597067 100644 --- a/crates/tranquil-api/src/delegation.rs +++ b/crates/tranquil-api/src/delegation.rs @@ -24,7 +24,7 @@ pub async fn list_controllers( auth: Auth, ) -> Result>>, ApiError> { let controllers = state - .delegation_repo + .repos.delegation .get_delegations_for_account(&auth.did) .await .map_err(|e| { @@ -99,7 +99,7 @@ pub async fn add_controller( if resolved.is_local && state - .delegation_repo + .repos.delegation .is_delegated_account(&input.controller_did) .await .unwrap_or(false) @@ -110,7 +110,7 @@ pub async fn add_controller( } match state - .delegation_repo + .repos.delegation .create_delegation( can_add.did(), &input.controller_did, @@ -121,7 +121,7 @@ pub async fn add_controller( { Ok(_) => { let _ = state - .delegation_repo + .repos.delegation .log_delegation_action( can_add.did(), can_add.did(), @@ -158,13 +158,13 @@ pub async fn remove_controller( Json(input): Json, ) -> Result, ApiError> { match state - .delegation_repo + .repos.delegation .revoke_delegation(&auth.did, &input.controller_did, &auth.did) .await { Ok(true) => { let revoked_app_passwords = state - .session_repo + .repos.session .delete_app_passwords_by_controller(&auth.did, &input.controller_did) .await .unwrap_or(0) @@ -172,13 +172,13 @@ pub async fn remove_controller( .unwrap_or(0usize); let revoked_oauth_tokens = state - .oauth_repo + .repos.oauth .revoke_tokens_for_controller(&auth.did, &input.controller_did) .await .unwrap_or(0); let _ = state - .delegation_repo + .repos.delegation .log_delegation_action( &auth.did, &auth.did, @@ -217,13 +217,13 @@ pub async fn update_controller_scopes( Json(input): Json, ) -> Result, ApiError> { match state - .delegation_repo + .repos.delegation .update_delegation_scopes(&auth.did, &input.controller_did, &input.granted_scopes) .await { Ok(true) => { let _ = state - .delegation_repo + .repos.delegation .log_delegation_action( &auth.did, &auth.did, @@ -254,7 +254,7 @@ pub async fn list_controlled_accounts( auth: Auth, ) -> Result>>, ApiError> { let accounts = state - .delegation_repo + .repos.delegation .get_accounts_controlled_by(&auth.did) .await .map_err(|e| { @@ -286,7 +286,7 @@ pub async fn get_audit_log( let offset = params.offset.max(0); let entries = state - .delegation_repo + .repos.delegation .get_audit_log_for_account(&auth.did, limit, offset) .await .map_err(|e| { @@ -295,7 +295,7 @@ pub async fn get_audit_log( })?; let total = state - .delegation_repo + .repos.delegation .count_audit_log_entries(&auth.did) .await .unwrap_or_default(); @@ -349,7 +349,7 @@ pub async fn create_delegated_account( } let validated_invite_code = if let Some(ref code) = input.invite_code { - match state.infra_repo.validate_invite_code(code).await { + match state.repos.infra.validate_invite_code(code).await { Ok(validated) => Some(validated), Err(_) => return Err(ApiError::InvalidInviteCode), } @@ -387,7 +387,7 @@ pub async fn create_delegated_account( }; let user_id = match state - .user_repo + .repos.user .create_delegated_account(&create_input) .await { @@ -406,7 +406,7 @@ pub async fn create_delegated_account( if let Some(validated) = validated_invite_code && let Err(e) = state - .infra_repo + .repos.infra .record_invite_code_use(&validated, user_id) .await { @@ -423,7 +423,7 @@ pub async fn create_delegated_account( .await; let _ = state - .delegation_repo + .repos.delegation .log_delegation_action( &did, &auth.did, @@ -461,7 +461,7 @@ pub async fn resolve_controller( } else { let local_handle: Option = identifier.parse().ok(); let local_user = match local_handle { - Some(ref h) => state.user_repo.get_by_handle(h).await.ok().flatten(), + Some(ref h) => state.repos.user.get_by_handle(h).await.ok().flatten(), None => None, }; match local_user { diff --git a/crates/tranquil-api/src/discord_webhook.rs b/crates/tranquil-api/src/discord_webhook.rs index 63b08e2..ccd0e33 100644 --- a/crates/tranquil-api/src/discord_webhook.rs +++ b/crates/tranquil-api/src/discord_webhook.rs @@ -169,7 +169,7 @@ async fn handle_command(state: AppState, interaction: Interaction) -> Response { ); match state - .user_repo + .repos.user .store_discord_user_id(&discord_username, &discord_user_id, handle.as_deref()) .await { @@ -180,8 +180,8 @@ async fn handle_command(state: AppState, interaction: Interaction) -> Response { "Verified Discord user and stored user ID" ); if let Err(e) = comms_repo::enqueue_channel_verified( - state.user_repo.as_ref(), - state.infra_repo.as_ref(), + state.repos.user.as_ref(), + state.repos.infra.as_ref(), user_id, tranquil_db_traits::CommsChannel::Discord, &discord_user_id, diff --git a/crates/tranquil-api/src/identity/account.rs b/crates/tranquil-api/src/identity/account.rs index e02bdc0..f07d509 100644 --- a/crates/tranquil-api/src/identity/account.rs +++ b/crates/tranquil-api/src/identity/account.rs @@ -6,7 +6,6 @@ use axum::{ http::{HeaderMap, StatusCode}, response::{IntoResponse, Response}, }; -use bcrypt::{DEFAULT_COST, hash}; use serde::{Deserialize, Serialize}; use serde_json::json; use tracing::{debug, error, info}; @@ -68,14 +67,14 @@ async fn try_reactivate_migration( new_email: email.clone(), }; match state - .user_repo + .repos.user .reactivate_migration_account(&reactivate_input) .await { Ok(reactivated) => { info!(did = %did, old_handle = %reactivated.old_handle, new_handle = %handle, "Preparing existing account for inbound migration"); let secret_key_bytes = match state - .user_repo + .repos.user .get_user_key_by_id(reactivated.user_id) .await { @@ -130,7 +129,7 @@ async fn try_reactivate_migration( controller_did: None, app_password_name: None, }; - if let Err(e) = state.session_repo.create_session(&session_data).await { + if let Err(e) = state.repos.session.create_session(&session_data).await { error!("Error creating session: {:?}", e); return Some(ApiError::InternalError(None).into_response()); } @@ -395,7 +394,7 @@ pub async fn create_account( Err(_) => return ApiError::InvalidHandle(None).into_response(), }; let handle_available = match state - .user_repo + .repos.user .check_handle_available_for_new_account(&handle_typed) .await { @@ -410,7 +409,7 @@ pub async fn create_account( } let is_bootstrap = state.bootstrap_invite_code.is_some() - && state.user_repo.count_users().await.unwrap_or(1) == 0; + && state.repos.user.count_users().await.unwrap_or(1) == 0; if is_bootstrap { match input.invite_code.as_deref() { @@ -431,7 +430,7 @@ pub async fn create_account( if let Some(code) = &input.invite_code && !code.trim().is_empty() { - let valid = match state.user_repo.check_and_consume_invite_code(code).await { + let valid = match state.repos.user.check_and_consume_invite_code(code).await { Ok(v) => v, Err(e) => { error!("Error checking invite code: {:?}", e); @@ -448,19 +447,10 @@ pub async fn create_account( return ApiError::InvalidRequest(e.to_string()).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 ApiError::InternalError(None).into_response(); - } - Err(e) => { - error!("Failed to spawn blocking task: {:?}", e); - return ApiError::InternalError(None).into_response(); - } - }; + let password_hash = match crate::common::hash_password_async(&input.password).await { + Ok(h) => h, + Err(e) => return e.into_response(), + }; let deactivated_at: Option> = if is_migration || is_did_web_byod { Some(chrono::Utc::now()) @@ -527,7 +517,7 @@ pub async fn create_account( birthdate_pref, }; - let create_result = match state.user_repo.create_password_account(&create_input).await { + let create_result = match state.repos.user.create_password_account(&create_input).await { Ok(r) => r, Err(tranquil_db_traits::CreateAccountError::HandleTaken) => { return ApiError::HandleNotAvailable(None).into_response(); diff --git a/crates/tranquil-api/src/identity/did.rs b/crates/tranquil-api/src/identity/did.rs index 361fde4..78465af 100644 --- a/crates/tranquil-api/src/identity/did.rs +++ b/crates/tranquil-api/src/identity/did.rs @@ -54,7 +54,7 @@ pub async fn resolve_handle( return ApiError::InvalidHandle(Some("Invalid handle format".into())).into_response(); } }; - let user = state.user_repo.get_by_handle(&handle).await; + let user = state.repos.user.get_by_handle(&handle).await; match user { Ok(Some(row)) => { let _ = state @@ -165,7 +165,7 @@ async fn serve_handle_did_doc(state: &AppState, handle: &str, hostname: &str) -> Err(_) => return ApiError::InvalidRequest("Invalid DID format".into()).into_response(), }; let user = match state - .user_repo + .repos.user .get_user_for_did_doc_build(&expected_did_typed) .await { @@ -182,7 +182,7 @@ async fn serve_handle_did_doc(state: &AppState, handle: &str, hostname: &str) -> let did = expected_did; let overrides = state - .user_repo + .repos.user .get_did_web_overrides(user_id) .await .ok() @@ -218,7 +218,7 @@ pub async fn user_did_doc(State(state): State, Path(handle): Path, Path(handle): Path k, _ => return Err(ApiError::InternalError(None).into_response()), }; @@ -468,7 +468,7 @@ pub async fn get_recommended_did_credentials( auth: Auth, ) -> Result, ApiError> { let handle = state - .user_repo + .repos.user .get_handle_by_did(&auth.did) .await .log_db_err("fetching handle for DID credentials")? @@ -539,7 +539,7 @@ pub async fn update_handle( ) .await?; let user_row = state - .user_repo + .repos.user .get_id_and_handle_by_did(&did) .await .log_db_err("fetching user for handle update")? @@ -661,7 +661,7 @@ pub async fn update_handle( .parse() .map_err(|_| ApiError::InvalidHandle(Some("Invalid handle format".into())))?; let handle_exists = state - .user_repo + .repos.user .check_handle_exists(&handle_typed, user_id) .await .log_db_err("checking handle existence")?; @@ -669,7 +669,7 @@ pub async fn update_handle( return Err(ApiError::HandleTaken); } state - .user_repo + .repos.user .update_handle(user_id, &handle_typed) .await .map_err(|e| { @@ -706,7 +706,7 @@ pub async fn update_plc_handle( if !did.as_str().starts_with("did:plc:") { return Ok(()); } - let user_row = match state.user_repo.get_user_with_key_by_did(did).await? { + let user_row = match state.repos.user.get_user_with_key_by_did(did).await? { Some(r) => r, None => return Ok(()), }; @@ -733,7 +733,7 @@ pub async fn well_known_atproto_did(State(state): State, headers: Head Ok(h) => h, Err(_) => return (StatusCode::BAD_REQUEST, "Invalid handle format").into_response(), }; - let user = state.user_repo.get_by_handle(&handle).await; + let user = state.repos.user.get_by_handle(&handle).await; match user { Ok(Some(row)) => row.did.to_string().into_response(), Ok(None) => (StatusCode::NOT_FOUND, "Handle not found").into_response(), diff --git a/crates/tranquil-api/src/identity/plc/request.rs b/crates/tranquil-api/src/identity/plc/request.rs index 8883600..737eaba 100644 --- a/crates/tranquil-api/src/identity/plc/request.rs +++ b/crates/tranquil-api/src/identity/plc/request.rs @@ -20,25 +20,25 @@ pub async fn request_plc_operation_signature( tranquil_pds::oauth::scopes::IdentityAttr::Wildcard, )?; let user_id = state - .user_repo + .repos.user .get_id_by_did(&auth.did) .await .log_db_err("fetching user id")? .ok_or(ApiError::AccountNotFound)?; - let _ = state.infra_repo.delete_plc_tokens_for_user(user_id).await; + let _ = state.repos.infra.delete_plc_tokens_for_user(user_id).await; let plc_token = generate_plc_token(); let expires_at = Utc::now() + Duration::minutes(10); state - .infra_repo + .repos.infra .insert_plc_token(user_id, &plc_token, expires_at) .await .log_db_err("creating PLC token")?; let hostname = &tranquil_config::get().server.hostname; if let Err(e) = tranquil_pds::comms::comms_repo::enqueue_plc_operation( - state.user_repo.as_ref(), - state.infra_repo.as_ref(), + state.repos.user.as_ref(), + state.repos.infra.as_ref(), user_id, &plc_token, hostname, diff --git a/crates/tranquil-api/src/identity/plc/sign.rs b/crates/tranquil-api/src/identity/plc/sign.rs index 81ff781..626cdb6 100644 --- a/crates/tranquil-api/src/identity/plc/sign.rs +++ b/crates/tranquil-api/src/identity/plc/sign.rs @@ -55,25 +55,25 @@ pub async fn sign_plc_operation( })?; let user_id = state - .user_repo + .repos.user .get_id_by_did(did) .await .log_db_err("fetching user id")? .ok_or(ApiError::AccountNotFound)?; let token_expiry = state - .infra_repo + .repos.infra .get_plc_token_expiry(user_id, token) .await .log_db_err("fetching PLC token expiry")? .ok_or_else(|| ApiError::InvalidToken(Some("Invalid or expired token".into())))?; if Utc::now() > token_expiry { - let _ = state.infra_repo.delete_plc_token(user_id, token).await; + let _ = state.repos.infra.delete_plc_token(user_id, token).await; return Err(ApiError::ExpiredToken(Some("Token has expired".into()))); } let key_row = state - .user_repo + .repos.user .get_user_key_by_id(user_id) .await .log_db_err("fetching user key")? @@ -136,7 +136,7 @@ pub async fn sign_plc_operation( ApiError::InternalError(None) })?; - let _ = state.infra_repo.delete_plc_token(user_id, token).await; + let _ = state.repos.infra.delete_plc_token(user_id, token).await; info!("Signed PLC operation for user {}", did); Ok(Json(SignPlcOperationOutput { operation: signed_op, diff --git a/crates/tranquil-api/src/identity/plc/submit.rs b/crates/tranquil-api/src/identity/plc/submit.rs index 4a2bb21..e1cedcf 100644 --- a/crates/tranquil-api/src/identity/plc/submit.rs +++ b/crates/tranquil-api/src/identity/plc/submit.rs @@ -38,14 +38,14 @@ pub async fn submit_plc_operation( let hostname = &tranquil_config::get().server.hostname; let public_url = format!("https://{}", hostname); let user = state - .user_repo + .repos.user .get_id_and_handle_by_did(did) .await .log_db_err("fetching user")? .ok_or(ApiError::AccountNotFound)?; let key_row = state - .user_repo + .repos.user .get_user_key_by_id(user.id) .await .log_db_err("fetching user key")? @@ -128,12 +128,12 @@ pub async fn submit_plc_operation( .map_err(ApiError::from)?; match state - .repo_repo + .repos.repo .insert_identity_event(did, Some(&user.handle)) .await { Ok(seq) => { - if let Err(e) = state.repo_repo.notify_update(seq).await { + if let Err(e) = state.repos.repo.notify_update(seq).await { warn!("Failed to notify identity event: {:?}", e); } } diff --git a/crates/tranquil-api/src/identity/provision.rs b/crates/tranquil-api/src/identity/provision.rs index b050ad4..f665a1c 100644 --- a/crates/tranquil-api/src/identity/provision.rs +++ b/crates/tranquil-api/src/identity/provision.rs @@ -136,7 +136,7 @@ pub async fn resolve_signing_key( match signing_key_did { Some(key_did) => { let key = state - .infra_repo + .repos.infra .get_reserved_signing_key(key_did) .await .map_err(|e| { @@ -295,7 +295,7 @@ pub async fn create_and_store_session( app_password_name: None, }; state - .session_repo + .repos.session .create_session(&session_data) .await .map_err(|e| { @@ -320,8 +320,8 @@ pub async fn enqueue_signup_verification( let formatted = tranquil_pds::auth::verification_token::format_token_for_display(&token); let hostname = &tranquil_config::get().server.hostname; if let Err(e) = tranquil_pds::comms::comms_repo::enqueue_signup_verification( - state.user_repo.as_ref(), - state.infra_repo.as_ref(), + state.repos.user.as_ref(), + state.repos.infra.as_ref(), user_id, channel, recipient, @@ -346,8 +346,8 @@ pub async fn enqueue_migration_verification( let formatted = tranquil_pds::auth::verification_token::format_token_for_display(&token); let hostname = &tranquil_config::get().server.hostname; if let Err(e) = tranquil_pds::comms::comms_repo::enqueue_migration_verification( - state.user_repo.as_ref(), - state.infra_repo.as_ref(), + state.repos.user.as_ref(), + state.repos.infra.as_ref(), user_id, channel, recipient, diff --git a/crates/tranquil-api/src/moderation/mod.rs b/crates/tranquil-api/src/moderation/mod.rs index 7fba74a..3df4639 100644 --- a/crates/tranquil-api/src/moderation/mod.rs +++ b/crates/tranquil-api/src/moderation/mod.rs @@ -107,7 +107,7 @@ async fn proxy_to_report_service( let key_bytes = match &auth_user.key_bytes { Some(kb) => kb.clone(), - None => match state.user_repo.get_with_key_by_did(&auth_user.did).await { + None => match state.repos.user.get_with_key_by_did(&auth_user.did).await { Ok(Some(user_with_key)) => { match tranquil_pds::config::decrypt_key( &user_with_key.key_bytes, @@ -226,7 +226,7 @@ async fn create_report_locally( let subject_json = json!(input.subject); if let Err(e) = state - .infra_repo + .repos.infra .insert_report( report_id, input.reason_type.as_str(), diff --git a/crates/tranquil-api/src/notification_prefs.rs b/crates/tranquil-api/src/notification_prefs.rs index a00dcc7..ab94d2b 100644 --- a/crates/tranquil-api/src/notification_prefs.rs +++ b/crates/tranquil-api/src/notification_prefs.rs @@ -26,7 +26,7 @@ pub async fn get_notification_prefs( auth: Auth, ) -> Result, ApiError> { let prefs = state - .user_repo + .repos.user .get_notification_prefs(&auth.did) .await .log_db_err("get notification prefs")? @@ -65,14 +65,14 @@ pub async fn get_notification_history( auth: Auth, ) -> Result, ApiError> { let user_id = state - .user_repo + .repos.user .get_id_by_did(&auth.did) .await .log_db_err("get user id by did")? .ok_or(ApiError::AccountNotFound)?; let rows = state - .infra_repo + .repos.infra .get_notification_history(user_id, 50) .await .log_db_err("get notification history")?; @@ -146,7 +146,7 @@ pub async fn request_channel_verification( let hostname = &tranquil_config::get().server.hostname; let handle_str = handle.unwrap_or("user"); tranquil_pds::comms::comms_repo::enqueue_email_update( - state.infra_repo.as_ref(), + state.repos.infra.as_ref(), user_id, identifier, handle_str, @@ -165,7 +165,7 @@ pub async fn request_channel_verification( hostname, encoded_token, encoded_identifier ); let prefs = state - .user_repo + .repos.user .get_comms_prefs(user_id) .await .ok() @@ -185,7 +185,7 @@ pub async fn request_channel_verification( ); let recipient = match channel { CommsChannel::Telegram => state - .user_repo + .repos.user .get_telegram_chat_id(user_id) .await .ok() @@ -195,7 +195,7 @@ pub async fn request_channel_verification( _ => identifier.to_string(), }; state - .infra_repo + .repos.infra .enqueue_comms( Some(user_id), channel, @@ -238,17 +238,17 @@ async fn process_messaging_channel_update( } match channel { CommsChannel::Discord => state - .user_repo + .repos.user .clear_discord(user_id) .await .log_db_err("clear discord")?, CommsChannel::Telegram => state - .user_repo + .repos.user .clear_telegram(user_id) .await .log_db_err("clear telegram")?, CommsChannel::Signal => state - .user_repo + .repos.user .clear_signal(user_id) .await .log_db_err("clear signal")?, @@ -281,17 +281,17 @@ async fn process_messaging_channel_update( match channel { CommsChannel::Discord => state - .user_repo + .repos.user .set_unverified_discord(user_id, &clean) .await .log_db_err("set unverified discord")?, CommsChannel::Telegram => state - .user_repo + .repos.user .set_unverified_telegram(user_id, &clean) .await .log_db_err("set unverified telegram")?, CommsChannel::Signal => state - .user_repo + .repos.user .set_unverified_signal(user_id, &clean) .await .log_db_err("set unverified signal")?, @@ -313,7 +313,7 @@ pub async fn update_notification_prefs( Json(input): Json, ) -> Result, ApiError> { let user_row = state - .user_repo + .repos.user .get_id_handle_email_by_did(&auth.did) .await .log_db_err("get user by did")? @@ -324,7 +324,7 @@ pub async fn update_notification_prefs( let current_email = user_row.email; let current_prefs = state - .user_repo + .repos.user .get_notification_prefs(&auth.did) .await .log_db_err("get notification prefs for update")? @@ -347,7 +347,7 @@ pub async fn update_notification_prefs( if input.preferred_channel.is_some() { state - .user_repo + .repos.user .update_preferred_comms_channel(&auth.did, effective_channel) .await .log_db_err("update preferred channel")?; diff --git a/crates/tranquil-api/src/repo/blob.rs b/crates/tranquil-api/src/repo/blob.rs index ef2270d..160ba44 100644 --- a/crates/tranquil-api/src/repo/blob.rs +++ b/crates/tranquil-api/src/repo/blob.rs @@ -66,7 +66,7 @@ pub async fn upload_blob( }; if state - .user_repo + .repos.user .is_account_migrated(&did) .await .unwrap_or(false) @@ -78,7 +78,7 @@ pub async fn upload_blob( get_header_str(&headers, http::header::CONTENT_TYPE).unwrap_or("application/octet-stream"); let user_id = state - .user_repo + .repos.user .get_id_by_did(&did) .await .log_db_err("fetching user id for blob upload")? @@ -143,7 +143,7 @@ pub async fn upload_blob( ); match state - .blob_repo + .repos.blob .insert_blob( &cid_link, &mime_type, @@ -163,7 +163,7 @@ pub async fn upload_blob( if let Err(e) = state.blob_store.copy(&temp_key, &storage_key).await { let _ = state.blob_store.delete(&temp_key).await; - if let Err(db_err) = state.blob_repo.delete_blob_by_cid(&cid_link).await { + if let Err(db_err) = state.repos.blob.delete_blob_by_cid(&cid_link).await { error!( "Failed to clean up orphaned blob record after copy failure: {:?}", db_err @@ -177,7 +177,7 @@ pub async fn upload_blob( if let Some(ref controller) = controller_did && let Err(e) = state - .delegation_repo + .repos.delegation .log_delegation_action( &did, controller, @@ -236,7 +236,7 @@ pub async fn list_missing_blobs( ) -> Result, ApiError> { let did = &auth.did; let user = state - .user_repo + .repos.user .get_by_did(did) .await .log_db_err("fetching user")? @@ -245,7 +245,7 @@ pub async fn list_missing_blobs( let limit = params.limit.unwrap_or(500).clamp(1, 1000); let cursor = params.cursor.as_deref(); let missing = state - .blob_repo + .repos.blob .list_missing_blobs(user.id, cursor, limit + 1) .await .log_db_err("fetching missing blobs")?; diff --git a/crates/tranquil-api/src/repo/import.rs b/crates/tranquil-api/src/repo/import.rs index 5122416..4a25753 100644 --- a/crates/tranquil-api/src/repo/import.rs +++ b/crates/tranquil-api/src/repo/import.rs @@ -34,7 +34,7 @@ pub async fn import_repo( } let did = &auth.did; let user = state - .user_repo + .repos.user .get_by_did(did) .await .log_db_err("fetching user")? @@ -44,7 +44,7 @@ pub async fn import_repo( } let user_id = user.id; let expected_root_cid = state - .repo_repo + .repos.repo .get_repo_root_cid_by_user_id(user_id) .await .map_err(|e| { @@ -194,7 +194,7 @@ pub async fn import_repo( let max_blocks = tranquil_config::get().import.max_blocks as usize; let _write_lock = state.repo_write_locks.lock(user_id).await; match apply_import( - &state.repo_repo, + &state.repos.repo, user_id, root, blocks.clone(), @@ -232,7 +232,7 @@ pub async fn import_repo( blob_refs.into_iter().unzip(); match state - .blob_repo + .repos.blob .insert_record_blobs(user_id, &record_uris, &blob_cids) .await { @@ -248,7 +248,7 @@ pub async fn import_repo( } } let key_row = state - .user_repo + .repos.user .get_user_with_key_by_did(did) .await .map_err(|e| { @@ -289,7 +289,7 @@ pub async fn import_repo( })?; let new_root_cid_link = CidLink::from(&new_root_cid); state - .repo_repo + .repos.repo .update_repo_root(user_id, &new_root_cid_link, &new_rev_str) .await .map_err(|e| { @@ -299,7 +299,7 @@ pub async fn import_repo( let mut all_block_cids: Vec> = blocks.keys().map(|c| c.to_bytes()).collect(); all_block_cids.push(new_root_cid.to_bytes()); state - .repo_repo + .repos.repo .insert_user_blocks(user_id, &all_block_cids, &new_rev_str) .await .map_err(|e| { @@ -322,7 +322,7 @@ pub async fn import_repo( "birthDate": "1998-05-06T00:00:00.000Z" }); if let Err(e) = state - .infra_repo + .repos.infra .insert_account_preference_if_not_exists( user_id, "app.bsky.actor.defs#personalDetailsPref", @@ -391,7 +391,7 @@ async fn sequence_import_event( rev: None, }; - let seq = state.repo_repo.insert_commit_event(&data).await?; - state.repo_repo.notify_update(seq).await?; + let seq = state.repos.repo.insert_commit_event(&data).await?; + state.repos.repo.notify_update(seq).await?; Ok(()) } diff --git a/crates/tranquil-api/src/repo/meta.rs b/crates/tranquil-api/src/repo/meta.rs index 1192fd2..87df549 100644 --- a/crates/tranquil-api/src/repo/meta.rs +++ b/crates/tranquil-api/src/repo/meta.rs @@ -18,12 +18,12 @@ pub async fn describe_repo( State(state): State, Query(input): Query, ) -> Response { - let resolved = match common::resolve_repo(state.user_repo.as_ref(), &input.repo).await { + let resolved = match common::resolve_repo(state.repos.user.as_ref(), &input.repo).await { Ok(r) => r, Err(e) => return e.into_response(), }; let collections = state - .repo_repo + .repos.repo .list_collections(resolved.user_id) .await .unwrap_or_default(); diff --git a/crates/tranquil-api/src/repo/record/batch.rs b/crates/tranquil-api/src/repo/record/batch.rs index 7d98ef4..f861d38 100644 --- a/crates/tranquil-api/src/repo/record/batch.rs +++ b/crates/tranquil-api/src/repo/record/batch.rs @@ -11,7 +11,7 @@ use tranquil_pds::auth::{ Active, Auth, WriteOpKind, require_not_migrated, require_verified_or_delegated, verify_batch_write_scopes, }; -use tranquil_pds::repo::tracking::TrackingBlockStore; +use tranquil_pds::repo::TrackingBlockStore; use tranquil_pds::repo_ops::{ FinalizeParams, RecordOp, begin_repo_write, extract_blob_cids, finalize_repo_write, }; @@ -304,7 +304,7 @@ pub async fn apply_writes( require_verified_or_delegated(&state, batch_proof.user()).await?; let user_id: uuid::Uuid = state - .user_repo + .repos.user .get_id_by_did(&did) .await .log_db_err("fetching user for batch write")? diff --git a/crates/tranquil-api/src/repo/record/delete.rs b/crates/tranquil-api/src/repo/record/delete.rs index 0537fa6..14b01e8 100644 --- a/crates/tranquil-api/src/repo/record/delete.rs +++ b/crates/tranquil-api/src/repo/record/delete.rs @@ -9,7 +9,7 @@ use std::sync::Arc; use tracing::error; use tranquil_pds::api::error::ApiError; use tranquil_pds::auth::{Active, Auth, VerifyScope}; -use tranquil_pds::repo::tracking::TrackingBlockStore; +use tranquil_pds::repo::TrackingBlockStore; use tranquil_pds::repo_ops::{ CommitError, FinalizeParams, RecordOp, begin_repo_write, finalize_repo_write, }; @@ -101,7 +101,7 @@ pub async fn delete_record( let deleted_uri = AtUri::from_parts(&did, &input.collection, &input.rkey); if let Err(e) = state - .backlink_repo + .repos.backlink .remove_backlinks_by_uri(&deleted_uri) .await { @@ -130,7 +130,7 @@ pub async fn delete_record_internal( let _write_lock = state.repo_write_locks.lock(user_id).await; let root_cid_str = state - .repo_repo + .repos.repo .get_repo_root_cid_by_user_id(user_id) .await .map_err(|e| CommitError::DatabaseError(e.to_string()))? diff --git a/crates/tranquil-api/src/repo/record/read.rs b/crates/tranquil-api/src/repo/record/read.rs index 8ba42ea..68ae99e 100644 --- a/crates/tranquil-api/src/repo/record/read.rs +++ b/crates/tranquil-api/src/repo/record/read.rs @@ -60,12 +60,12 @@ pub async fn get_record( _headers: HeaderMap, Query(input): Query, ) -> Response { - let user_id = match common::resolve_repo_user_id(state.user_repo.as_ref(), &input.repo).await { + let user_id = match common::resolve_repo_user_id(state.repos.user.as_ref(), &input.repo).await { Ok(id) => id, Err(e) => return e.into_response(), }; let record_row = state - .repo_repo + .repos.repo .get_record_cid(user_id, &input.collection, &input.rkey) .await; let record_cid_link = match record_row { @@ -128,7 +128,7 @@ pub async fn list_records( State(state): State, Query(input): Query, ) -> Response { - let user_id = match common::resolve_repo_user_id(state.user_repo.as_ref(), &input.repo).await { + let user_id = match common::resolve_repo_user_id(state.repos.user.as_ref(), &input.repo).await { Ok(id) => id, Err(e) => return e.into_response(), }; @@ -139,7 +139,7 @@ pub async fn list_records( .as_ref() .and_then(|c| c.parse::().ok()); let rows = match state - .repo_repo + .repos.repo .list_records( user_id, &input.collection, diff --git a/crates/tranquil-api/src/repo/record/write.rs b/crates/tranquil-api/src/repo/record/write.rs index e1b65b2..93f40de 100644 --- a/crates/tranquil-api/src/repo/record/write.rs +++ b/crates/tranquil-api/src/repo/record/write.rs @@ -46,7 +46,7 @@ pub async fn prepare_repo_write( let _account_verified = require_verified_or_delegated(state, user).await?; let user_id = state - .user_repo + .repos.user .get_id_by_did(principal_did.as_did()) .await .log_db_err("fetching user for repo write")? @@ -128,7 +128,7 @@ pub async fn create_record( if !backlinks.is_empty() { let conflicts = state - .backlink_repo + .repos.backlink .get_backlink_conflicts(user_id, &input.collection, &backlinks) .await .log_db_err("checking backlink conflicts")?; @@ -229,7 +229,7 @@ pub async fn create_record( .await?; { - let backlink_repo = state.backlink_repo.clone(); + let backlink_repo = state.repos.backlink.clone(); futures::future::join_all(conflict_uris_to_cleanup.iter().map(|uri| { let backlink_repo = backlink_repo.clone(); async move { @@ -244,7 +244,7 @@ pub async fn create_record( let created_uri = AtUri::from_parts(&did, &input.collection, &rkey); let backlinks = extract_backlinks(&created_uri, &input.record); if !backlinks.is_empty() - && let Err(e) = state.backlink_repo.add_backlinks(user_id, &backlinks).await + && let Err(e) = state.repos.backlink.add_backlinks(user_id, &backlinks).await { error!("Failed to add backlinks for {}: {}", created_uri, e); } diff --git a/crates/tranquil-api/src/server/account_status.rs b/crates/tranquil-api/src/server/account_status.rs index 562bfdb..dd68a94 100644 --- a/crates/tranquil-api/src/server/account_status.rs +++ b/crates/tranquil-api/src/server/account_status.rs @@ -41,24 +41,24 @@ pub async fn check_account_status( ) -> Result, ApiError> { let did = &auth.did; let user_id = state - .user_repo + .repos.user .get_id_by_did(did) .await .log_db_err("fetching user ID for account status")? .ok_or(ApiError::InternalError(None))?; let is_active = state - .user_repo + .repos.user .is_account_active_by_did(did) .await .ok() .flatten() .unwrap_or(false); - let repo_info = state.repo_repo.get_repo(user_id).await.ok().flatten(); + let repo_info = state.repos.repo.get_repo(user_id).await.ok().flatten(); let (repo_commit, repo_rev_from_db) = repo_info .map(|r| (r.repo_root_cid.to_string(), r.repo_rev)) .unwrap_or_else(|| (String::new(), None)); let block_count: i64 = state - .repo_repo + .repos.repo .count_user_blocks(user_id) .await .unwrap_or(0); @@ -80,19 +80,19 @@ pub async fn check_account_status( } else { String::new() }; - let record_count: i64 = state.repo_repo.count_records(user_id).await.unwrap_or(0); + let record_count: i64 = state.repos.repo.count_records(user_id).await.unwrap_or(0); let imported_blobs: i64 = state - .blob_repo + .repos.blob .count_blobs_by_user(user_id) .await .unwrap_or(0); let expected_blobs: i64 = state - .blob_repo + .repos.blob .count_distinct_record_blobs(user_id) .await .unwrap_or(0); let valid_did = - is_valid_did_for_service(state.user_repo.as_ref(), state.cache.clone(), did).await; + is_valid_did_for_service(state.repos.user.as_ref(), state.cache.clone(), did).await; Ok(Json(CheckAccountStatusOutput { activated: is_active, valid_did, @@ -319,7 +319,7 @@ pub async fn activate_account( ); let did_validation_start = std::time::Instant::now(); if let Err(e) = assert_valid_did_document_for_service( - state.user_repo.as_ref(), + state.repos.user.as_ref(), state.cache.clone(), &did, true, @@ -339,12 +339,12 @@ pub async fn activate_account( did_validation_start.elapsed() ); - let handle = state.user_repo.get_handle_by_did(&did).await.ok().flatten(); + let handle = state.repos.user.get_handle_by_did(&did).await.ok().flatten(); info!( "[MIGRATION] activateAccount: Activating account did={} handle={:?}", did, handle ); - let result = state.user_repo.activate_account(&did).await; + let result = state.repos.user.activate_account(&did).await; match result { Ok(_) => { info!( @@ -406,7 +406,7 @@ pub async fn activate_account( info!("[MIGRATION] activateAccount: Identity event sequenced successfully"); } let repo_root = state - .repo_repo + .repos.repo .get_repo_root_by_did(&did) .await .ok() @@ -480,9 +480,9 @@ pub async fn deactivate_account( let did = auth.did.clone(); - let handle = state.user_repo.get_handle_by_did(&did).await.ok().flatten(); + let handle = state.repos.user.get_handle_by_did(&did).await.ok().flatten(); - let result = state.user_repo.deactivate_account(&did, delete_after).await; + let result = state.repos.user.deactivate_account(&did, delete_after).await; match result { Ok(true) => { @@ -518,7 +518,7 @@ pub async fn request_account_delete( let session_mfa = require_legacy_session_mfa(&state, &auth).await?; let user_id = state - .user_repo + .repos.user .get_id_by_did(session_mfa.did()) .await .ok() @@ -527,14 +527,14 @@ pub async fn request_account_delete( let confirmation_token = Uuid::new_v4().to_string(); let expires_at = Utc::now() + Duration::minutes(15); state - .infra_repo + .repos.infra .create_deletion_request(&confirmation_token, session_mfa.did(), expires_at) .await .log_db_err("creating deletion token")?; let hostname = &tranquil_config::get().server.hostname; if let Err(e) = tranquil_pds::comms::comms_repo::enqueue_account_deletion( - state.user_repo.as_ref(), - state.infra_repo.as_ref(), + state.repos.user.as_ref(), + state.repos.infra.as_ref(), user_id, &confirmation_token, hostname, @@ -572,7 +572,7 @@ pub async fn delete_account( return Err(ApiError::InvalidToken(Some("token is required".into()))); } let user = state - .user_repo + .repos.user .get_user_for_deletion(did) .await .map_err(|e| { @@ -582,7 +582,7 @@ pub async fn delete_account( .ok_or(ApiError::InvalidRequest("account not found".into()))?; let (user_id, password_hash, handle) = (user.id, user.password_hash, user.handle); if crate::common::verify_credential( - state.session_repo.as_ref(), + state.repos.session.as_ref(), user_id, password, password_hash.as_deref(), @@ -595,7 +595,7 @@ pub async fn delete_account( ))); } let deletion_request = state - .infra_repo + .repos.infra .get_deletion_request(token) .await .map_err(|e| { @@ -611,11 +611,11 @@ pub async fn delete_account( ))); } if Utc::now() > deletion_request.expires_at { - let _ = state.infra_repo.delete_deletion_request(token).await; + let _ = state.repos.infra.delete_deletion_request(token).await; return Err(ApiError::ExpiredToken(None)); } state - .user_repo + .repos.user .delete_account_complete(user_id, did) .await .map_err(|e| { @@ -630,7 +630,7 @@ pub async fn delete_account( .await; match account_seq { Ok(seq) => { - if let Err(e) = state.repo_repo.delete_sequences_except(did, seq).await { + if let Err(e) = state.repos.repo.delete_sequences_except(did, seq).await { warn!( "Failed to cleanup sequences for deleted account {}: {}", did, e diff --git a/crates/tranquil-api/src/server/app_password.rs b/crates/tranquil-api/src/server/app_password.rs index 675a4d2..5591029 100644 --- a/crates/tranquil-api/src/server/app_password.rs +++ b/crates/tranquil-api/src/server/app_password.rs @@ -1,7 +1,6 @@ use axum::{Json, extract::State}; use serde::{Deserialize, Serialize}; use serde_json::json; -use tracing::error; use tranquil_db_traits::AppPasswordCreate; use tranquil_pds::api::EmptyResponse; use tranquil_pds::api::error::{ApiError, DbResultExt}; @@ -32,14 +31,14 @@ pub async fn list_app_passwords( auth: Auth, ) -> Result, ApiError> { let user = state - .user_repo + .repos.user .get_by_did(&auth.did) .await .log_db_err("getting user")? .ok_or(ApiError::AccountNotFound)?; let rows = state - .session_repo + .repos.session .list_app_passwords(user.id) .await .log_db_err("listing app passwords")?; @@ -84,7 +83,7 @@ pub async fn create_app_password( Json(input): Json, ) -> Result, ApiError> { let user = state - .user_repo + .repos.user .get_by_did(&auth.did) .await .log_db_err("getting user")? @@ -96,7 +95,7 @@ pub async fn create_app_password( } if state - .session_repo + .repos.session .get_app_password_by_name(user.id, name) .await .log_db_err("checking app password")? @@ -107,7 +106,7 @@ pub async fn create_app_password( let (final_scopes, controller_did) = if let Some(ref controller) = auth.controller_did { let grant = state - .delegation_repo + .repos.delegation .get_delegation(&auth.did, controller) .await .ok() @@ -133,18 +132,7 @@ pub async fn create_app_password( let password = generate_app_password(); - let password_clone = password.clone(); - let password_hash = - tokio::task::spawn_blocking(move || bcrypt::hash(&password_clone, bcrypt::DEFAULT_COST)) - .await - .map_err(|e| { - error!("Failed to spawn blocking task: {:?}", e); - ApiError::InternalError(None) - })? - .map_err(|e| { - error!("Failed to hash password: {:?}", e); - ApiError::InternalError(None) - })?; + let password_hash = crate::common::hash_password_async(&password).await?; let privilege = tranquil_db_traits::AppPasswordPrivilege::from_privileged_flag( input.privileged.unwrap_or(false), @@ -161,14 +149,14 @@ pub async fn create_app_password( }; state - .session_repo + .repos.session .create_app_password(&create_data) .await .log_db_err("creating app password")?; if let Some(ref controller) = controller_did { let _ = state - .delegation_repo + .repos.delegation .log_delegation_action( &auth.did, controller, @@ -204,7 +192,7 @@ pub async fn revoke_app_password( Json(input): Json, ) -> Result, ApiError> { let user = state - .user_repo + .repos.user .get_by_did(&auth.did) .await .log_db_err("getting user")? @@ -216,13 +204,13 @@ pub async fn revoke_app_password( } let sessions_to_invalidate = state - .session_repo + .repos.session .get_session_jtis_by_app_password(&auth.did, name) .await .unwrap_or_default(); state - .session_repo + .repos.session .delete_sessions_by_app_password(&auth.did, name) .await .log_db_err("revoking sessions for app password")?; @@ -237,7 +225,7 @@ pub async fn revoke_app_password( .await; state - .session_repo + .repos.session .delete_app_password(user.id, name) .await .log_db_err("revoking app password")?; diff --git a/crates/tranquil-api/src/server/email.rs b/crates/tranquil-api/src/server/email.rs index 68bc868..0d033b9 100644 --- a/crates/tranquil-api/src/server/email.rs +++ b/crates/tranquil-api/src/server/email.rs @@ -22,10 +22,6 @@ use tranquil_pds::state::AppState; const EMAIL_UPDATE_TTL: Duration = Duration::from_secs(30 * 60); -fn email_update_cache_key(did: &str) -> String { - tranquil_pds::cache_keys::email_update_key(did) -} - fn hash_token(token: &str) -> String { let mut hasher = Sha256::new(); hasher.update(token.as_bytes()); @@ -39,6 +35,16 @@ struct PendingEmailUpdate { authorized: bool, } +async fn get_pending_email_update( + cache: &dyn tranquil_pds::cache::Cache, + did: &str, +) -> Option { + cache + .get(&tranquil_pds::cache_keys::email_update_key(did)) + .await + .and_then(|json| serde_json::from_str(&json).ok()) +} + #[derive(Deserialize)] #[serde(rename_all = "camelCase")] pub struct RequestEmailUpdateInput { @@ -55,7 +61,7 @@ pub async fn request_email_update( auth.check_account_scope(AccountAttr::Email, AccountAction::Manage)?; let user = state - .user_repo + .repos.user .get_email_info_by_did(&auth.did) .await .log_db_err("getting email info")? @@ -92,7 +98,7 @@ pub async fn request_email_update( authorized: false, }; if let Ok(json) = serde_json::to_string(&pending) { - let cache_key = email_update_cache_key(&auth.did); + let cache_key = tranquil_pds::cache_keys::email_update_key(&auth.did); if let Err(e) = state.cache.set(&cache_key, &json, EMAIL_UPDATE_TTL).await { warn!("Failed to cache pending email update: {:?}", e); } @@ -102,8 +108,8 @@ pub async fn request_email_update( let hostname = &tranquil_config::get().server.hostname; if let Err(e) = tranquil_pds::comms::comms_repo::enqueue_short_token_email( - state.user_repo.as_ref(), - state.infra_repo.as_ref(), + state.repos.user.as_ref(), + state.repos.infra.as_ref(), user.id, &token, hostname, @@ -135,7 +141,7 @@ pub async fn confirm_email( let did = &auth.did; let user = state - .user_repo + .repos.user .get_email_info_by_did(did) .await .log_db_err("getting email info")? @@ -179,7 +185,7 @@ pub async fn confirm_email( } state - .user_repo + .repos.user .set_email_verified(user.id, true) .await .log_db_err("confirming email")?; @@ -206,7 +212,7 @@ pub async fn update_email( let did = &auth.did; let user = state - .user_repo + .repos.user .get_email_info_by_did(did) .await .log_db_err("getting email info")? @@ -253,7 +259,7 @@ pub async fn update_email( } state - .infra_repo + .repos.infra .upsert_account_preference(user_id, "email_auth_factor", json!(email_auth_factor)) .await .map_err(|e| { @@ -267,7 +273,7 @@ pub async fn update_email( if email_verified { let mut authorized_via_link = false; - let cache_key = email_update_cache_key(did); + let cache_key = tranquil_pds::cache_keys::email_update_key(did); if let Some(pending_json) = state.cache.get(&cache_key).await && let Ok(pending) = serde_json::from_str::(&pending_json) && pending.authorized @@ -336,7 +342,7 @@ pub async fn update_email( } state - .user_repo + .repos.user .update_email(user_id, &new_email) .await .log_db_err("updating email")?; @@ -350,8 +356,8 @@ pub async fn update_email( tranquil_pds::auth::verification_token::format_token_for_display(&verification_token); let hostname = &tranquil_config::get().server.hostname; if let Err(e) = tranquil_pds::comms::comms_repo::enqueue_signup_verification( - state.user_repo.as_ref(), - state.infra_repo.as_ref(), + state.repos.user.as_ref(), + state.repos.infra.as_ref(), user_id, tranquil_db_traits::CommsChannel::Email, &new_email, @@ -364,7 +370,7 @@ pub async fn update_email( } if let Err(e) = state - .infra_repo + .repos.infra .upsert_account_preference( user_id, "email_auth_factor", @@ -390,7 +396,7 @@ pub async fn check_email_verified( Json(input): Json, ) -> Result, ApiError> { let verified = state - .user_repo + .repos.user .check_email_verified_by_identifier(&input.identifier) .await .map_err(|e| { @@ -414,7 +420,7 @@ pub async fn check_channel_verified( Json(input): Json, ) -> Result, ApiError> { let verified = state - .user_repo + .repos.user .check_channel_verified_by_did(&input.did, input.channel) .await .map_err(|e| { @@ -470,26 +476,16 @@ pub async fn authorize_email_update( let did = token_data.did; info!("authorize_email_update: token valid for did={}", did); - let cache_key = email_update_cache_key(&did); - let pending_json = match state.cache.get(&cache_key).await { - Some(json) => json, + let cache_key = tranquil_pds::cache_keys::email_update_key(&did); + let mut pending = match get_pending_email_update(state.cache.as_ref(), &did).await { + Some(p) => p, None => { - warn!( - "authorize_email_update: no pending email update in cache for did={}", - did - ); + warn!("authorize_email_update: no pending email update in cache for did={}", did); return ApiError::InvalidRequest("No pending email update found".into()) .into_response(); } }; - let mut pending: PendingEmailUpdate = match serde_json::from_str(&pending_json) { - Ok(p) => p, - Err(_) => { - return ApiError::InternalError(None).into_response(); - } - }; - let token_hash = hash_token(&query.token); if pending .token_hash @@ -528,9 +524,8 @@ pub async fn check_email_update_status( ) -> Result, ApiError> { auth.check_account_scope(AccountAttr::Email, AccountAction::Read)?; - let cache_key = email_update_cache_key(&auth.did); - let pending_json = match state.cache.get(&cache_key).await { - Some(json) => json, + let pending = match get_pending_email_update(state.cache.as_ref(), &auth.did).await { + Some(p) => p, None => { return Ok(Json(EmailUpdateStatusOutput { pending: false, @@ -540,17 +535,6 @@ pub async fn check_email_update_status( } }; - let pending: PendingEmailUpdate = match serde_json::from_str(&pending_json) { - Ok(p) => p, - Err(_) => { - return Ok(Json(EmailUpdateStatusOutput { - pending: false, - authorized: false, - new_email: None, - })); - } - }; - Ok(Json(EmailUpdateStatusOutput { pending: true, authorized: pending.authorized, @@ -574,7 +558,7 @@ pub async fn check_email_in_use( } let count = state - .user_repo + .repos.user .count_accounts_by_email(&email) .await .map_err(|e| { diff --git a/crates/tranquil-api/src/server/invite.rs b/crates/tranquil-api/src/server/invite.rs index 68d09c9..efff1c7 100644 --- a/crates/tranquil-api/src/server/invite.rs +++ b/crates/tranquil-api/src/server/invite.rs @@ -1,5 +1,4 @@ use axum::{Json, extract::State}; -use rand::Rng; use serde::{Deserialize, Serialize}; use tracing::error; use tranquil_pds::api::ApiError; @@ -7,24 +6,7 @@ use tranquil_pds::api::error::DbResultExt; use tranquil_pds::auth::{Admin, Auth, NotTakendown}; use tranquil_pds::state::AppState; use tranquil_pds::types::Did; - -const BASE32_ALPHABET: &[u8] = b"abcdefghijklmnopqrstuvwxyz234567"; - -pub(crate) fn gen_random_token() -> String { - let mut rng = rand::thread_rng(); - let gen_segment = |rng: &mut rand::rngs::ThreadRng, len: usize| -> String { - (0..len) - .map(|_| BASE32_ALPHABET[rng.gen_range(0..32)] as char) - .collect() - }; - format!("{}-{}", gen_segment(&mut rng, 5), gen_segment(&mut rng, 5)) -} - -pub fn gen_invite_code() -> String { - let hostname = &tranquil_config::get().server.hostname; - let hostname_prefix = hostname.replace('.', "-"); - format!("{}-{}", hostname_prefix, gen_random_token()) -} +use tranquil_pds::util::gen_invite_code; #[derive(Deserialize)] #[serde(rename_all = "camelCase")] @@ -58,7 +40,7 @@ pub async fn create_invite_code( let code = gen_invite_code(); match state - .infra_repo + .repos.infra .create_invite_code(&code, input.use_count, Some(&for_account)) .await { @@ -115,7 +97,7 @@ pub async fn create_invite_codes( }; let admin_user_id = state - .user_repo + .repos.user .get_any_admin_user_id() .await .log_db_err("looking up admin user")? @@ -125,7 +107,7 @@ pub async fn create_invite_codes( })?; let result = futures::future::try_join_all(for_accounts.into_iter().map(|account| { - let infra_repo = state.infra_repo.clone(); + let infra_repo = state.repos.infra.clone(); let use_count = input.use_count; async move { let codes: Vec = (0..code_count).map(|_| gen_invite_code()).collect(); @@ -192,7 +174,7 @@ pub async fn get_account_invite_codes( let include_used = params.include_used.unwrap_or(true); let codes_info = state - .infra_repo + .repos.infra .get_invite_codes_for_account(&auth.did) .await .log_db_err("fetching invite codes")?; @@ -203,7 +185,7 @@ pub async fn get_account_invite_codes( .collect(); let codes = futures::future::join_all(filtered_codes.into_iter().map(|info| { - let infra_repo = state.infra_repo.clone(); + let infra_repo = state.repos.infra.clone(); async move { let uses = infra_repo .get_invite_code_uses(&info.code) diff --git a/crates/tranquil-api/src/server/logo.rs b/crates/tranquil-api/src/server/logo.rs index 4cf8d17..02817e9 100644 --- a/crates/tranquil-api/src/server/logo.rs +++ b/crates/tranquil-api/src/server/logo.rs @@ -9,7 +9,7 @@ use tracing::error; use tranquil_pds::state::AppState; pub async fn get_logo(State(state): State) -> Response { - let logo_cid = match state.infra_repo.get_server_config("logo_cid").await { + let logo_cid = match state.repos.infra.get_server_config("logo_cid").await { Ok(cid) => cid, Err(e) => { error!("DB error fetching logo_cid: {:?}", e); @@ -26,7 +26,7 @@ pub async fn get_logo(State(state): State) -> Response { Err(_) => return StatusCode::NOT_FOUND.into_response(), }; - let metadata = match state.blob_repo.get_blob_metadata(&cid).await { + let metadata = match state.repos.blob.get_blob_metadata(&cid).await { Ok(Some(m)) => m, Ok(None) => return StatusCode::NOT_FOUND.into_response(), Err(e) => { diff --git a/crates/tranquil-api/src/server/meta.rs b/crates/tranquil-api/src/server/meta.rs index 5067c9f..d07271f 100644 --- a/crates/tranquil-api/src/server/meta.rs +++ b/crates/tranquil-api/src/server/meta.rs @@ -98,7 +98,7 @@ pub struct HealthOutput { } pub async fn health(State(state): State) -> impl IntoResponse { - match state.infra_repo.health_check().await { + match state.repos.infra.health_check().await { Ok(true) => ( StatusCode::OK, Json(HealthOutput { diff --git a/crates/tranquil-api/src/server/migration.rs b/crates/tranquil-api/src/server/migration.rs index 6e54df2..a5fe2ba 100644 --- a/crates/tranquil-api/src/server/migration.rs +++ b/crates/tranquil-api/src/server/migration.rs @@ -42,7 +42,7 @@ pub async fn update_did_document( } let user = state - .user_repo + .repos.user .get_user_for_did_doc(&auth.did) .await .log_db_err("getting user")? @@ -97,7 +97,7 @@ pub async fn update_did_document( let also_known_as: Option> = input.also_known_as.clone(); state - .user_repo + .repos.user .upsert_did_web_overrides(user.id, verification_methods_json, also_known_as) .await .log_db_err("upserting did_web_overrides")?; @@ -105,7 +105,7 @@ pub async fn update_did_document( if let Some(ref endpoint) = input.service_endpoint { let endpoint_clean = endpoint.trim().trim_end_matches('/'); state - .user_repo + .repos.user .update_migrated_to_pds(&auth.did, endpoint_clean) .await .log_db_err("updating service endpoint")?; @@ -139,7 +139,7 @@ pub async fn get_did_document( async fn build_did_document(state: &AppState, did: &tranquil_pds::types::Did) -> serde_json::Value { let hostname = &tranquil_config::get().server.hostname; - let user = match state.user_repo.get_user_for_did_doc_build(did).await { + let user = match state.repos.user.get_user_for_did_doc_build(did).await { Ok(Some(row)) => row, _ => { return json!({ @@ -149,7 +149,7 @@ async fn build_did_document(state: &AppState, did: &tranquil_pds::types::Did) -> }; let overrides = state - .user_repo + .repos.user .get_did_web_overrides(user.id) .await .ok() @@ -193,7 +193,7 @@ async fn build_did_document(state: &AppState, did: &tranquil_pds::types::Did) -> } let key_info = state - .user_repo + .repos.user .get_user_key_by_id(user.id) .await .ok() diff --git a/crates/tranquil-api/src/server/passkey_account.rs b/crates/tranquil-api/src/server/passkey_account.rs index 1766d5e..6c2f31c 100644 --- a/crates/tranquil-api/src/server/passkey_account.rs +++ b/crates/tranquil-api/src/server/passkey_account.rs @@ -120,7 +120,7 @@ pub async fn create_passkey_account( } let is_bootstrap = state.bootstrap_invite_code.is_some() - && state.user_repo.count_users().await.unwrap_or(1) == 0; + && state.repos.user.count_users().await.unwrap_or(1) == 0; let _validated_invite_code = if is_bootstrap { match input.invite_code.as_deref() { @@ -128,7 +128,7 @@ pub async fn create_passkey_account( _ => return Err(ApiError::InvalidInviteCode), } } else if let Some(ref code) = input.invite_code { - match state.infra_repo.validate_invite_code(code).await { + match state.repos.infra.validate_invite_code(code).await { Ok(validated) => Some(validated), Err(_) => return Err(ApiError::InvalidInviteCode), } @@ -351,7 +351,7 @@ pub async fn create_passkey_account( birthdate_pref, }; - let create_result = match state.user_repo.create_passkey_account(&create_input).await { + let create_result = match state.repos.user.create_passkey_account(&create_input).await { Ok(r) => r, Err(tranquil_db_traits::CreateAccountError::HandleTaken) => { return Err(ApiError::HandleNotAvailable(None)); @@ -405,7 +405,7 @@ pub async fn create_passkey_account( controller_did: None, app_password_name: None, }; - if let Err(e) = state.session_repo.create_session(&session_data).await { + if let Err(e) = state.repos.session.create_session(&session_data).await { warn!(did = %did, "Failed to insert migration session: {:?}", e); } info!(did = %did, "Generated migration access token for BYOD passkey account"); @@ -451,7 +451,7 @@ pub async fn complete_passkey_setup( State(state): State, Json(input): Json, ) -> Result, ApiError> { - let user = match state.user_repo.get_user_for_passkey_setup(&input.did).await { + let user = match state.repos.user.get_user_for_passkey_setup(&input.did).await { Ok(Some(u)) => u, Ok(None) => { return Err(ApiError::AccountNotFound); @@ -484,7 +484,7 @@ pub async fn complete_passkey_setup( let webauthn = &state.webauthn_config; let reg_state = match state - .user_repo + .repos.user .load_webauthn_challenge(&input.did, WebauthnChallengeType::Registration) .await { @@ -530,7 +530,7 @@ pub async fn complete_passkey_setup( } }; if let Err(e) = state - .user_repo + .repos.user .save_passkey( &input.did, &credential_id, @@ -553,13 +553,13 @@ pub async fn complete_passkey_setup( app_password_name: app_password_name.clone(), app_password_hash: password_hash, }; - if let Err(e) = state.user_repo.complete_passkey_setup(&setup_input).await { + if let Err(e) = state.repos.user.complete_passkey_setup(&setup_input).await { error!("Error completing passkey setup: {:?}", e); return Err(ApiError::InternalError(None)); } let _ = state - .user_repo + .repos.user .delete_webauthn_challenge(&input.did, WebauthnChallengeType::Registration) .await; @@ -577,7 +577,7 @@ pub async fn start_passkey_registration_for_setup( State(state): State, Json(input): Json, ) -> Result>, ApiError> { - let user = match state.user_repo.get_user_for_passkey_setup(&input.did).await { + let user = match state.repos.user.get_user_for_passkey_setup(&input.did).await { Ok(Some(u)) => u, Ok(None) => { return Err(ApiError::AccountNotFound); @@ -610,7 +610,7 @@ pub async fn start_passkey_registration_for_setup( let webauthn = &state.webauthn_config; let existing_passkeys = state - .user_repo + .repos.user .get_passkeys_for_user(&input.did) .await .unwrap_or_default(); @@ -643,7 +643,7 @@ pub async fn start_passkey_registration_for_setup( } }; if let Err(e) = state - .user_repo + .repos.user .save_webauthn_challenge(&input.did, WebauthnChallengeType::Registration, &state_json) .await { @@ -682,7 +682,7 @@ pub async fn request_passkey_recovery( NormalizedLoginIdentifier::normalize(&input.email, hostname_for_handles); let user = match state - .user_repo + .repos.user .get_user_for_passkey_recovery(identifier, normalized_handle.as_str()) .await { @@ -697,7 +697,7 @@ pub async fn request_passkey_recovery( let expires_at = Utc::now() + Duration::hours(1); if let Err(e) = state - .user_repo + .repos.user .set_recovery_token(&user.did, &recovery_token_hash, expires_at) .await { @@ -714,8 +714,8 @@ pub async fn request_passkey_recovery( ); let _ = tranquil_pds::comms::comms_repo::enqueue_passkey_recovery( - state.user_repo.as_ref(), - state.infra_repo.as_ref(), + state.repos.user.as_ref(), + state.repos.infra.as_ref(), user.id, &recovery_url, hostname, @@ -742,7 +742,7 @@ pub async fn recover_passkey_account( return Err(ApiError::InvalidRequest(e.to_string())); } - let user = match state.user_repo.get_user_for_recovery(&input.did).await { + let user = match state.repos.user.get_user_for_recovery(&input.did).await { Ok(Some(u)) => u, _ => { return Err(ApiError::InvalidRecoveryLink); @@ -771,7 +771,7 @@ pub async fn recover_passkey_account( password_hash, }; let result = match state - .user_repo + .repos.user .recover_passkey_account(&recover_input) .await { @@ -785,11 +785,11 @@ pub async fn recover_passkey_account( if result.passkeys_deleted > 0 { info!(did = %input.did, count = result.passkeys_deleted, "Deleted lost passkeys during account recovery"); } - if let Ok(Some(prefs)) = state.user_repo.get_comms_prefs(user.id).await { + if let Ok(Some(prefs)) = state.repos.user.get_comms_prefs(user.id).await { let actual_channel = tranquil_pds::comms::resolve_delivery_channel(&prefs, user.preferred_comms_channel); if let Err(e) = state - .user_repo + .repos.user .set_channel_verified(&input.did, actual_channel) .await { diff --git a/crates/tranquil-api/src/server/passkeys.rs b/crates/tranquil-api/src/server/passkeys.rs index 32c1e9b..93286fa 100644 --- a/crates/tranquil-api/src/server/passkeys.rs +++ b/crates/tranquil-api/src/server/passkeys.rs @@ -28,14 +28,14 @@ pub async fn start_passkey_registration( let webauthn = &state.webauthn_config; let handle = state - .user_repo + .repos.user .get_handle_by_did(&auth.did) .await .log_db_err("fetching user")? .ok_or(ApiError::AccountNotFound)?; let existing_passkeys = state - .user_repo + .repos.user .get_passkeys_for_user(&auth.did) .await .log_db_err("fetching existing passkeys")?; @@ -60,7 +60,7 @@ pub async fn start_passkey_registration( })?; state - .user_repo + .repos.user .save_webauthn_challenge(&auth.did, WebauthnChallengeType::Registration, &state_json) .await .log_db_err("saving registration state")?; @@ -94,7 +94,7 @@ pub async fn finish_passkey_registration( let webauthn = &state.webauthn_config; let reg_state_json = state - .user_repo + .repos.user .load_webauthn_challenge(&auth.did, WebauthnChallengeType::Registration) .await .log_db_err("loading registration state")? @@ -125,7 +125,7 @@ pub async fn finish_passkey_registration( })?; let passkey_id = state - .user_repo + .repos.user .save_passkey( &auth.did, passkey.cred_id(), @@ -136,7 +136,7 @@ pub async fn finish_passkey_registration( .log_db_err("saving passkey")?; if let Err(e) = state - .user_repo + .repos.user .delete_webauthn_challenge(&auth.did, WebauthnChallengeType::Registration) .await { @@ -177,7 +177,7 @@ pub async fn list_passkeys( auth: Auth, ) -> Result, ApiError> { let passkeys = state - .user_repo + .repos.user .get_passkeys_for_user(&auth.did) .await .log_db_err("fetching passkeys")?; @@ -215,7 +215,7 @@ pub async fn delete_passkey( let id: uuid::Uuid = input.id.parse().map_err(|_| ApiError::InvalidId)?; - match state.user_repo.delete_passkey(id, reauth_mfa.did()).await { + match state.repos.user.delete_passkey(id, reauth_mfa.did()).await { Ok(true) => { info!(did = %session_mfa.did(), passkey_id = %id, "Passkey deleted"); Ok(Json(EmptyResponse {})) @@ -243,7 +243,7 @@ pub async fn update_passkey( let id: uuid::Uuid = input.id.parse().map_err(|_| ApiError::InvalidId)?; match state - .user_repo + .repos.user .update_passkey_name(id, &auth.did, &input.friendly_name) .await { @@ -260,5 +260,5 @@ pub async fn update_passkey( } pub async fn has_passkeys_for_user(state: &AppState, did: &tranquil_pds::types::Did) -> bool { - state.user_repo.has_passkeys(did).await.unwrap_or(false) + state.repos.user.has_passkeys(did).await.unwrap_or(false) } diff --git a/crates/tranquil-api/src/server/password.rs b/crates/tranquil-api/src/server/password.rs index e0e8110..40b6f21 100644 --- a/crates/tranquil-api/src/server/password.rs +++ b/crates/tranquil-api/src/server/password.rs @@ -1,5 +1,4 @@ use axum::{Json, extract::State}; -use bcrypt::{DEFAULT_COST, hash}; use chrono::{Duration, Utc}; use serde::Deserialize; use tracing::{error, info, warn}; @@ -42,7 +41,7 @@ pub async fn request_password_reset( let normalized_handle = NormalizedLoginIdentifier::normalize(identifier, hostname_for_handles); let multiple_accounts_warning = if is_email_lookup { - match state.user_repo.count_accounts_by_email(normalized).await { + match state.repos.user.count_accounts_by_email(normalized).await { Ok(count) if count > 1 => Some(count), _ => None, } @@ -51,7 +50,7 @@ pub async fn request_password_reset( }; let user_id = match state - .user_repo + .repos.user .get_id_by_email_or_handle(normalized, normalized_handle.as_str()) .await { @@ -73,7 +72,7 @@ pub async fn request_password_reset( let code = generate_reset_code(); let expires_at = Utc::now() + Duration::minutes(10); if let Err(e) = state - .user_repo + .repos.user .set_password_reset_code(user_id, &code, expires_at) .await { @@ -82,8 +81,8 @@ pub async fn request_password_reset( } let hostname = &tranquil_config::get().server.hostname; if let Err(e) = tranquil_pds::comms::comms_repo::enqueue_password_reset( - state.user_repo.as_ref(), - state.infra_repo.as_ref(), + state.repos.user.as_ref(), + state.repos.infra.as_ref(), user_id, &code, hostname, @@ -132,7 +131,7 @@ pub async fn reset_password( if let Err(e) = validate_password(password) { return Err(ApiError::InvalidRequest(e.to_string())); } - let user = match state.user_repo.get_user_by_reset_code(token).await { + let user = match state.repos.user.get_user_by_reset_code(token).await { Ok(Some(u)) => u, Ok(None) => { return Err(ApiError::InvalidToken(None)); @@ -147,26 +146,14 @@ pub async fn reset_password( return Err(ApiError::InvalidToken(None)); }; if Utc::now() > exp { - if let Err(e) = state.user_repo.clear_password_reset_code(user_id).await { + if let Err(e) = state.repos.user.clear_password_reset_code(user_id).await { error!("Failed to clear expired reset code: {:?}", e); } return Err(ApiError::ExpiredToken(None)); } - 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 Err(ApiError::InternalError(None)); - } - Err(e) => { - error!("Failed to spawn blocking task: {:?}", e); - return Err(ApiError::InternalError(None)); - } - }; + let password_hash = crate::common::hash_password_async(&password).await?; let result = match state - .user_repo + .repos.user .reset_password_with_sessions(user_id, &password_hash) .await { @@ -189,11 +176,11 @@ pub async fn reset_password( } })) .await; - if let Ok(Some(prefs)) = state.user_repo.get_comms_prefs(user_id).await { + if let Ok(Some(prefs)) = state.repos.user.get_comms_prefs(user_id).await { let actual_channel = tranquil_pds::comms::resolve_delivery_channel(&prefs, user.preferred_comms_channel); if let Err(e) = state - .user_repo + .repos.user .set_channel_verified(&user.did, actual_channel) .await { @@ -238,26 +225,16 @@ pub async fn change_password( let password_mfa = verify_password_mfa(&state, &auth, &input.current_password).await?; let user = state - .user_repo + .repos.user .get_id_and_password_hash_by_did(password_mfa.did()) .await .log_db_err("in change_password")? .ok_or(ApiError::AccountNotFound)?; - let new_password_clone = input.new_password.to_string(); - let new_hash = tokio::task::spawn_blocking(move || hash(new_password_clone, DEFAULT_COST)) - .await - .map_err(|e| { - error!("Failed to spawn blocking task: {:?}", e); - ApiError::InternalError(None) - })? - .map_err(|e| { - error!("Failed to hash password: {:?}", e); - ApiError::InternalError(None) - })?; + let new_hash = crate::common::hash_password_async(&input.new_password).await?; state - .user_repo + .repos.user .update_password_hash(user.id, &new_hash) .await .log_db_err("updating password")?; @@ -271,7 +248,7 @@ pub async fn get_password_status( auth: Auth, ) -> Result, ApiError> { let has = state - .user_repo + .repos.user .has_password_by_did(&auth.did) .await .log_db_err("checking password status")? @@ -288,7 +265,7 @@ pub async fn remove_password( let reauth_mfa = require_reauth_window(&state, &auth).await?; let has_passkeys = state - .user_repo + .repos.user .has_passkeys(reauth_mfa.did()) .await .unwrap_or(false); @@ -299,7 +276,7 @@ pub async fn remove_password( } let user = state - .user_repo + .repos.user .get_password_info_by_did(reauth_mfa.did()) .await .log_db_err("getting password info")? @@ -312,7 +289,7 @@ pub async fn remove_password( } state - .user_repo + .repos.user .remove_user_password(user.id) .await .log_db_err("removing password")?; @@ -345,7 +322,7 @@ pub async fn set_password( let did = reauth_mfa.as_ref().map(|m| m.did()).unwrap_or(&auth.did); let user = state - .user_repo + .repos.user .get_password_info_by_did(did) .await .log_db_err("getting password info")? @@ -357,20 +334,10 @@ pub async fn set_password( )); } - let new_password_clone = new_password.to_string(); - let new_hash = tokio::task::spawn_blocking(move || hash(new_password_clone, DEFAULT_COST)) - .await - .map_err(|e| { - error!("Failed to spawn blocking task: {:?}", e); - ApiError::InternalError(None) - })? - .map_err(|e| { - error!("Failed to hash password: {:?}", e); - ApiError::InternalError(None) - })?; + let new_hash = crate::common::hash_password_async(&new_password).await?; state - .user_repo + .repos.user .set_new_user_password(user.id, &new_hash) .await .log_db_err("setting password")?; diff --git a/crates/tranquil-api/src/server/reauth.rs b/crates/tranquil-api/src/server/reauth.rs index ab614f4..10b93bc 100644 --- a/crates/tranquil-api/src/server/reauth.rs +++ b/crates/tranquil-api/src/server/reauth.rs @@ -33,13 +33,13 @@ pub async fn get_reauth_status( auth: Auth, ) -> Result, ApiError> { let last_reauth_at = state - .session_repo + .repos.session .get_last_reauth_at(&auth.did) .await .log_db_err("getting last reauth")?; let reauth_required = is_reauth_required(last_reauth_at); - let available_methods = get_available_reauth_methods(&*state.user_repo, &auth.did).await; + let available_methods = get_available_reauth_methods(&*state.repos.user, &auth.did).await; Ok(Json(ReauthStatusOutput { last_reauth_at, @@ -66,7 +66,7 @@ pub async fn reauth_password( Json(input): Json, ) -> Result, ApiError> { let password_hash = state - .user_repo + .repos.user .get_password_hash_by_did(&auth.did) .await .log_db_err("fetching password hash")? @@ -76,7 +76,7 @@ pub async fn reauth_password( if !password_valid { let app_password_hashes = state - .session_repo + .repos.session .get_app_password_hashes_by_did(&auth.did) .await .unwrap_or_default(); @@ -91,7 +91,7 @@ pub async fn reauth_password( } } - let reauthed_at = update_last_reauth_cached(&*state.session_repo, &state.cache, &auth.did) + let reauthed_at = update_last_reauth_cached(&*state.repos.session, &state.cache, &auth.did) .await .log_db_err("updating reauth")?; @@ -127,7 +127,7 @@ pub async fn reauth_totp( ))); } - let reauthed_at = update_last_reauth_cached(&*state.session_repo, &state.cache, &auth.did) + let reauthed_at = update_last_reauth_cached(&*state.repos.session, &state.cache, &auth.did) .await .log_db_err("updating reauth")?; @@ -146,7 +146,7 @@ pub async fn reauth_passkey_start( auth: Auth, ) -> Result, ApiError> { let stored_passkeys = state - .user_repo + .repos.user .get_passkeys_for_user(&auth.did) .await .log_db_err("getting passkeys")?; @@ -179,7 +179,7 @@ pub async fn reauth_passkey_start( })?; state - .user_repo + .repos.user .save_webauthn_challenge( &auth.did, WebauthnChallengeType::Authentication, @@ -204,7 +204,7 @@ pub async fn reauth_passkey_finish( Json(input): Json, ) -> Result, ApiError> { let auth_state_json = state - .user_repo + .repos.user .load_webauthn_challenge(&auth.did, WebauthnChallengeType::Authentication) .await .log_db_err("loading authentication state")? @@ -232,7 +232,7 @@ pub async fn reauth_passkey_finish( let cred_id_bytes = auth_result.cred_id().as_ref(); match state - .user_repo + .repos.user .update_passkey_counter( cred_id_bytes, i32::try_from(auth_result.counter()).unwrap_or(i32::MAX), @@ -242,7 +242,7 @@ pub async fn reauth_passkey_finish( Ok(false) => { warn!(did = %&auth.did, "Passkey counter anomaly detected - possible cloned key"); let _ = state - .user_repo + .repos.user .delete_webauthn_challenge(&auth.did, WebauthnChallengeType::Authentication) .await; return Err(ApiError::PasskeyCounterAnomaly); @@ -254,11 +254,11 @@ pub async fn reauth_passkey_finish( } let _ = state - .user_repo + .repos.user .delete_webauthn_challenge(&auth.did, WebauthnChallengeType::Authentication) .await; - let reauthed_at = update_last_reauth_cached(&*state.session_repo, &state.cache, &auth.did) + let reauthed_at = update_last_reauth_cached(&*state.repos.session, &state.cache, &auth.did) .await .log_db_err("updating reauth")?; diff --git a/crates/tranquil-api/src/server/service_auth.rs b/crates/tranquil-api/src/server/service_auth.rs index 2d3c952..aae6972 100644 --- a/crates/tranquil-api/src/server/service_auth.rs +++ b/crates/tranquil-api/src/server/service_auth.rs @@ -72,7 +72,7 @@ pub async fn get_service_auth( Some(kb) => kb.clone(), None => { warn!(did = %&auth.did, "getServiceAuth: no key_bytes in auth, fetching from DB"); - match state.user_repo.get_user_info_by_did(&auth.did).await { + match state.repos.user.get_user_info_by_did(&auth.did).await { Ok(Some(info)) => match info.key_bytes { Some(key_bytes_enc) => { match tranquil_pds::config::decrypt_key( diff --git a/crates/tranquil-api/src/server/session.rs b/crates/tranquil-api/src/server/session.rs index ef38322..9682375 100644 --- a/crates/tranquil-api/src/server/session.rs +++ b/crates/tranquil-api/src/server/session.rs @@ -69,7 +69,7 @@ pub async fn create_session( input.identifier, normalized_identifier ); let row = match state - .user_repo + .repos.user .get_login_full_by_identifier(normalized_identifier.as_str()) .await { @@ -98,7 +98,7 @@ pub async fn create_session( } }; let credential = crate::common::verify_credential( - state.session_repo.as_ref(), + state.repos.session.as_ref(), row.id, &input.password, row.password_hash.as_deref(), @@ -130,7 +130,7 @@ pub async fn create_session( } let is_verified = row.channel_verification.has_any_verified(); let is_delegated = state - .delegation_repo + .repos.delegation .is_delegated_account(&row.did) .await .unwrap_or(false); @@ -181,8 +181,8 @@ pub async fn create_session( Ok(tranquil_pds::auth::legacy_2fa::Legacy2faOutcome::ChallengeSent(code)) => { let hostname = &tranquil_config::get().server.hostname; if let Err(e) = tranquil_pds::comms::comms_repo::enqueue_2fa_code( - state.user_repo.as_ref(), - state.infra_repo.as_ref(), + state.repos.user.as_ref(), + state.repos.infra.as_ref(), row.id, code.as_str(), hostname, @@ -269,7 +269,7 @@ pub async fn create_session( app_password_name: app_password_name.clone(), }; let (insert_result, did_doc) = tokio::join!( - state.session_repo.create_session(&session_data), + state.repos.session.create_session(&session_data), did_resolver.resolve_did_document(&did_for_doc) ); if let Err(e) = insert_result { @@ -284,8 +284,8 @@ pub async fn create_session( ); let hostname = &tranquil_config::get().server.hostname; if let Err(e) = tranquil_pds::comms::comms_repo::enqueue_legacy_login( - state.user_repo.as_ref(), - state.infra_repo.as_ref(), + state.repos.user.as_ref(), + state.repos.infra.as_ref(), row.id, hostname, client_ip, @@ -359,7 +359,7 @@ pub async fn get_session( let did_for_doc = auth.did.clone(); let did_resolver = state.did_resolver.clone(); let (db_result, did_doc) = tokio::join!( - state.user_repo.get_session_info_by_did(&auth.did), + state.repos.user.get_session_info_by_did(&auth.did), did_resolver.resolve_did_document(&did_for_doc) ); match db_result { @@ -422,7 +422,7 @@ pub async fn delete_session( ) -> Result, ApiError> { let jti = tranquil_pds::auth::extract_jti_from_headers(&headers) .ok_or(ApiError::AuthenticationRequired)?; - match state.session_repo.delete_session_by_access_jti(&jti).await { + match state.repos.session.delete_session_by_access_jti(&jti).await { Ok(rows) if rows > 0 => { let session_cache_key = tranquil_pds::cache_keys::session_key(&auth.did, &jti); let _ = state.cache.delete(&session_cache_key).await; @@ -476,7 +476,7 @@ pub async fn refresh_session( } }; if let Ok(Some(_)) = state - .session_repo + .repos.session .check_refresh_token_used(&refresh_jti) .await { @@ -486,7 +486,7 @@ pub async fn refresh_session( ))); } let session_row = match state - .session_repo + .repos.session .get_session_for_refresh(&refresh_jti) .await { @@ -548,7 +548,7 @@ pub async fn refresh_session( new_refresh_expires_at: new_refresh_meta.expires_at, }; match state - .session_repo + .repos.session .refresh_session_atomic(&refresh_data) .await { @@ -576,7 +576,7 @@ pub async fn refresh_session( let did_for_doc = session_row.did.clone(); let did_resolver = state.did_resolver.clone(); let (db_result, did_doc) = tokio::join!( - state.user_repo.get_session_info_by_did(&session_row.did), + state.repos.user.get_session_info_by_did(&session_row.did), did_resolver.resolve_did_document(&did_for_doc) ); match db_result { @@ -639,7 +639,7 @@ pub async fn confirm_signup( Json(input): Json, ) -> Result, ApiError> { info!("confirm_signup called for DID: {}", input.did); - let row = match state.user_repo.get_confirm_signup_by_did(&input.did).await { + let row = match state.repos.user.get_confirm_signup_by_did(&input.did).await { Ok(Some(row)) => row, Ok(None) => { warn!("User not found for confirm_signup: {}", input.did); @@ -702,7 +702,7 @@ pub async fn confirm_signup( }; if let Err(e) = state - .user_repo + .repos.user .set_channel_verified(&input.did, row.channel) .await { @@ -726,8 +726,8 @@ pub async fn confirm_signup( let hostname = &tranquil_config::get().server.hostname; if let Err(e) = tranquil_pds::comms::comms_repo::enqueue_welcome( - state.user_repo.as_ref(), - state.infra_repo.as_ref(), + state.repos.user.as_ref(), + state.repos.infra.as_ref(), row.id, hostname, ) @@ -757,7 +757,7 @@ pub struct AutoResendResult { pub async fn auto_resend_verification(state: &AppState, did: &Did) -> Option { let debounce_key = tranquil_pds::cache_keys::auto_verify_sent_key(did.as_str()); let debounced = state.cache.get(&debounce_key).await.is_some(); - let row = match state.user_repo.get_resend_verification_by_did(did).await { + let row = match state.repos.user.get_resend_verification_by_did(did).await { Ok(Some(row)) => row, Ok(None) => return None, Err(e) => { @@ -821,7 +821,7 @@ pub async fn resend_verification( ) -> Result, ApiError> { info!("resend_verification called for DID: {}", input.did); let row = match state - .user_repo + .repos.user .get_resend_verification_by_did(&input.did) .await { @@ -895,13 +895,13 @@ pub async fn list_sessions( let current_jti = tranquil_pds::auth::extract_jti_from_headers(&headers); let jwt_rows = state - .session_repo + .repos.session .list_sessions_by_did(&auth.did) .await .log_db_err("fetching JWT sessions")?; let oauth_rows = state - .oauth_repo + .repos.oauth .list_sessions_by_did(&auth.did) .await .log_db_err("fetching OAuth sessions")?; @@ -962,13 +962,13 @@ pub async fn revoke_session( .map(SessionId::new) .map_err(|_| ApiError::InvalidRequest("Invalid session ID".into()))?; let access_jti = state - .session_repo + .repos.session .get_session_access_jti_by_id(session_id, &auth.did) .await .log_db_err("in revoke_session")? .ok_or(ApiError::SessionNotFound)?; state - .session_repo + .repos.session .delete_session_by_id(session_id) .await .log_db_err("deleting session")?; @@ -983,7 +983,7 @@ pub async fn revoke_session( .map(TokenFamilyId::new) .map_err(|_| ApiError::InvalidRequest("Invalid session ID".into()))?; let deleted = state - .oauth_repo + .repos.oauth .delete_session_by_id(session_id, &auth.did) .await .log_db_err("deleting OAuth session")?; @@ -1007,24 +1007,24 @@ pub async fn revoke_all_sessions( if auth.is_oauth() { state - .session_repo + .repos.session .delete_sessions_by_did(&auth.did) .await .log_db_err("revoking JWT sessions")?; let jti_typed = TokenId::from(jti.clone()); state - .oauth_repo + .repos.oauth .delete_sessions_by_did_except(&auth.did, &jti_typed) .await .log_db_err("revoking OAuth sessions")?; } else { state - .session_repo + .repos.session .delete_sessions_by_did_except_jti(&auth.did, &jti) .await .log_db_err("revoking JWT sessions")?; state - .oauth_repo + .repos.oauth .delete_sessions_by_did(&auth.did) .await .log_db_err("revoking OAuth sessions")?; @@ -1046,7 +1046,7 @@ pub async fn get_legacy_login_preference( auth: Auth, ) -> Result, ApiError> { let pref = state - .user_repo + .repos.user .get_legacy_login_pref(&auth.did) .await .log_db_err("getting legacy login pref")? @@ -1079,7 +1079,7 @@ pub async fn update_legacy_login_preference( let reauth_mfa = require_reauth_window(&state, &auth).await?; let updated = state - .user_repo + .repos.user .update_legacy_login(reauth_mfa.did(), input.allow_legacy_login) .await .log_db_err("updating legacy login")?; @@ -1117,7 +1117,7 @@ pub async fn update_locale( } let updated = state - .user_repo + .repos.user .update_locale(&auth.did, &input.preferred_locale) .await .log_db_err("updating locale")?; diff --git a/crates/tranquil-api/src/server/signing_key.rs b/crates/tranquil-api/src/server/signing_key.rs index 3396abc..fe5f89a 100644 --- a/crates/tranquil-api/src/server/signing_key.rs +++ b/crates/tranquil-api/src/server/signing_key.rs @@ -44,7 +44,7 @@ pub async fn reserve_signing_key( let expires_at = Utc::now() + Duration::hours(24); let private_bytes: &[u8] = &private_key_bytes; match state - .infra_repo + .repos.infra .reserve_signing_key( input.did.as_ref(), &public_key_did_key, diff --git a/crates/tranquil-api/src/server/totp.rs b/crates/tranquil-api/src/server/totp.rs index fa14271..80311cf 100644 --- a/crates/tranquil-api/src/server/totp.rs +++ b/crates/tranquil-api/src/server/totp.rs @@ -29,7 +29,7 @@ pub async fn create_totp_secret( ) -> Result, ApiError> { use tranquil_db_traits::TotpRecordState; - match state.user_repo.get_totp_record_state(&auth.did).await { + match state.repos.user.get_totp_record_state(&auth.did).await { Ok(Some(TotpRecordState::Verified(_))) => return Err(ApiError::TotpAlreadyEnabled), Ok(Some(TotpRecordState::Unverified(_))) | Ok(None) => {} Err(e) => { @@ -41,7 +41,7 @@ pub async fn create_totp_secret( let secret = generate_totp_secret(); let handle = state - .user_repo + .repos.user .get_handle_by_did(&auth.did) .await .log_db_err("fetching handle")? @@ -61,7 +61,7 @@ pub async fn create_totp_secret( })?; state - .user_repo + .repos.user .upsert_totp_secret(&auth.did, &encrypted_secret, ENCRYPTION_VERSION) .await .log_db_err("storing TOTP secret")?; @@ -102,7 +102,7 @@ pub async fn enable_totp( ) .await?; - let unverified_record = match state.user_repo.get_totp_record_state(&auth.did).await { + let unverified_record = match state.repos.user.get_totp_record_state(&auth.did).await { Ok(Some(TotpRecordState::Unverified(record))) => record, Ok(Some(TotpRecordState::Verified(_))) => return Err(ApiError::TotpAlreadyEnabled), Ok(None) => return Err(ApiError::TotpNotEnabled), @@ -139,7 +139,7 @@ pub async fn enable_totp( })?; state - .user_repo + .repos.user .enable_totp_with_backup_codes(&auth.did, &backup_hashes) .await .log_db_err("enabling TOTP")?; @@ -173,7 +173,7 @@ pub async fn disable_totp( let totp_mfa = verify_totp_mfa(&state, &auth, &input.code).await?; state - .user_repo + .repos.user .delete_totp_and_backup_codes(totp_mfa.did()) .await .log_db_err("deleting TOTP")?; @@ -199,7 +199,7 @@ pub async fn get_totp_status( ) -> Result, ApiError> { use tranquil_db_traits::TotpRecordState; - let enabled = match state.user_repo.get_totp_record_state(&auth.did).await { + let enabled = match state.repos.user.get_totp_record_state(&auth.did).await { Ok(Some(TotpRecordState::Verified(_))) => true, Ok(Some(TotpRecordState::Unverified(_))) | Ok(None) => false, Err(e) => { @@ -209,7 +209,7 @@ pub async fn get_totp_status( }; let backup_count = state - .user_repo + .repos.user .count_unused_backup_codes(&auth.did) .await .log_db_err("counting backup codes")?; @@ -259,7 +259,7 @@ pub async fn regenerate_backup_codes( })?; state - .user_repo + .repos.user .replace_backup_codes(totp_mfa.did(), &backup_hashes) .await .log_db_err("replacing backup codes")?; @@ -276,7 +276,7 @@ async fn verify_backup_code_for_user( ) -> bool { let code = code.trim().to_uppercase(); - let backup_codes = match state.user_repo.get_unused_backup_codes(did).await { + let backup_codes = match state.repos.user.get_unused_backup_codes(did).await { Ok(codes) => codes, Err(e) => { warn!("Failed to fetch backup codes: {:?}", e); @@ -290,7 +290,7 @@ async fn verify_backup_code_for_user( match matched { Some(row) => { - let _ = state.user_repo.mark_backup_code_used(row.id).await; + let _ = state.repos.user.mark_backup_code_used(row.id).await; true } None => false, @@ -310,7 +310,7 @@ pub async fn verify_totp_or_backup_for_user( return verify_backup_code_for_user(state, did, code).await; } - let verified_record = match state.user_repo.get_totp_record_state(did).await { + let verified_record = match state.repos.user.get_totp_record_state(did).await { Ok(Some(TotpRecordState::Verified(record))) => record, _ => return false, }; @@ -324,7 +324,7 @@ pub async fn verify_totp_or_backup_for_user( }; if verify_totp_code(&secret, code) { - let _ = state.user_repo.update_totp_last_used(did).await; + let _ = state.repos.user.update_totp_last_used(did).await; return true; } @@ -332,5 +332,5 @@ pub async fn verify_totp_or_backup_for_user( } pub async fn has_totp_enabled(state: &AppState, did: &tranquil_pds::types::Did) -> bool { - state.user_repo.has_totp_enabled(did).await.unwrap_or(false) + state.repos.user.has_totp_enabled(did).await.unwrap_or(false) } diff --git a/crates/tranquil-api/src/server/trusted_devices.rs b/crates/tranquil-api/src/server/trusted_devices.rs index 7528404..afb52ac 100644 --- a/crates/tranquil-api/src/server/trusted_devices.rs +++ b/crates/tranquil-api/src/server/trusted_devices.rs @@ -72,7 +72,7 @@ pub async fn list_trusted_devices( auth: Auth, ) -> Result, ApiError> { let rows = state - .oauth_repo + .repos.oauth .list_trusted_devices(&auth.did) .await .log_db_err("listing trusted devices")?; @@ -108,7 +108,7 @@ pub async fn revoke_trusted_device( Json(input): Json, ) -> Result, ApiError> { match state - .oauth_repo + .repos.oauth .device_belongs_to_user(&input.device_id, &auth.did) .await { @@ -123,7 +123,7 @@ pub async fn revoke_trusted_device( } state - .oauth_repo + .repos.oauth .revoke_device_trust(&input.device_id) .await .log_db_err("revoking device trust")?; @@ -145,7 +145,7 @@ pub async fn update_trusted_device( Json(input): Json, ) -> Result, ApiError> { match state - .oauth_repo + .repos.oauth .device_belongs_to_user(&input.device_id, &auth.did) .await { @@ -160,7 +160,7 @@ pub async fn update_trusted_device( } state - .oauth_repo + .repos.oauth .update_device_friendly_name(&input.device_id, input.friendly_name.as_deref()) .await .log_db_err("updating device friendly name")?; diff --git a/crates/tranquil-api/src/server/verify_email.rs b/crates/tranquil-api/src/server/verify_email.rs index 105fa00..d905c30 100644 --- a/crates/tranquil-api/src/server/verify_email.rs +++ b/crates/tranquil-api/src/server/verify_email.rs @@ -59,7 +59,7 @@ pub async fn resend_migration_verification( .unwrap_or(tranquil_db_traits::CommsChannel::Email); let identifier = input.identifier.trim().to_lowercase(); - let user = match state.user_repo.get_by_email(&identifier).await { + let user = match state.repos.user.get_by_email(&identifier).await { Ok(Some(u)) => u, Ok(None) => { return Ok(Json(ResendMigrationVerificationOutput { sent: true })); diff --git a/crates/tranquil-api/src/server/verify_token.rs b/crates/tranquil-api/src/server/verify_token.rs index 9baa18f..bc280e6 100644 --- a/crates/tranquil-api/src/server/verify_token.rs +++ b/crates/tranquil-api/src/server/verify_token.rs @@ -79,7 +79,7 @@ async fn handle_migration_verification( identifier: &str, ) -> Result, ApiError> { let user = state - .user_repo + .repos.user .get_verification_info(did) .await .log_db_err("during migration verification")? @@ -92,13 +92,13 @@ async fn handle_migration_verification( } if !user.channel_verification.email { state - .user_repo + .repos.user .set_email_verified_flag(user.id) .await .log_db_err("updating email_verified status")?; } } - _ => common::set_channel_verified_flag(state.user_repo.as_ref(), user.id, channel).await?, + _ => common::set_channel_verified_flag(state.repos.user.as_ref(), user.id, channel).await?, }; info!(did = %did, channel = ?channel, "Migration verification completed successfully"); @@ -118,7 +118,7 @@ async fn handle_channel_update( identifier: &str, ) -> Result, ApiError> { let user_id = state - .user_repo + .repos.user .get_id_by_did(did) .await .log_db_err("fetching user id")? @@ -127,7 +127,7 @@ async fn handle_channel_update( match channel { CommsChannel::Email => { let success = state - .user_repo + .repos.user .verify_email_channel(user_id, identifier) .await .log_db_err("updating email channel")?; @@ -137,21 +137,21 @@ async fn handle_channel_update( } CommsChannel::Discord => { state - .user_repo + .repos.user .verify_discord_channel(user_id, identifier) .await .log_db_err("updating discord channel")?; } CommsChannel::Telegram => { state - .user_repo + .repos.user .verify_telegram_channel(user_id, identifier) .await .log_db_err("updating telegram channel")?; } CommsChannel::Signal => { state - .user_repo + .repos.user .verify_signal_channel(user_id, identifier) .await .log_db_err("updating signal channel")?; @@ -160,19 +160,7 @@ async fn handle_channel_update( info!(did = %did, channel = ?channel, "Channel verified successfully"); - let recipient = resolve_verified_recipient(state, user_id, channel, identifier).await; - if let Err(e) = comms_repo::enqueue_channel_verified( - state.user_repo.as_ref(), - state.infra_repo.as_ref(), - user_id, - channel, - &recipient, - &tranquil_config::get().server.hostname, - ) - .await - { - warn!(error = %e, "Failed to enqueue channel verified notification"); - } + notify_channel_verified(state, user_id, channel, identifier).await; Ok(Json(VerifyTokenOutput { success: true, @@ -182,15 +170,15 @@ async fn handle_channel_update( })) } -async fn resolve_verified_recipient( +async fn notify_channel_verified( state: &AppState, user_id: uuid::Uuid, - channel: tranquil_db_traits::CommsChannel, + channel: CommsChannel, identifier: &str, -) -> String { - match channel { - tranquil_db_traits::CommsChannel::Telegram => state - .user_repo +) { + let recipient = match channel { + CommsChannel::Telegram => state + .repos.user .get_telegram_chat_id(user_id) .await .ok() @@ -198,6 +186,18 @@ async fn resolve_verified_recipient( .map(|id| id.to_string()) .unwrap_or_else(|| identifier.to_string()), _ => identifier.to_string(), + }; + if let Err(e) = comms_repo::enqueue_channel_verified( + state.repos.user.as_ref(), + state.repos.infra.as_ref(), + user_id, + channel, + &recipient, + &tranquil_config::get().server.hostname, + ) + .await + { + warn!(error = %e, "Failed to enqueue channel verified notification"); } } @@ -208,7 +208,7 @@ async fn handle_signup_verification( identifier: &str, ) -> Result, ApiError> { let user = state - .user_repo + .repos.user .get_verification_info(did) .await .log_db_err("during signup verification")? @@ -225,23 +225,11 @@ async fn handle_signup_verification( })); } - common::set_channel_verified_flag(state.user_repo.as_ref(), user.id, channel).await?; + common::set_channel_verified_flag(state.repos.user.as_ref(), user.id, channel).await?; info!(did = %did, channel = ?channel, "Signup verified successfully"); - let recipient = resolve_verified_recipient(state, user.id, channel, identifier).await; - if let Err(e) = comms_repo::enqueue_channel_verified( - state.user_repo.as_ref(), - state.infra_repo.as_ref(), - user.id, - channel, - &recipient, - &tranquil_config::get().server.hostname, - ) - .await - { - warn!(error = %e, "Failed to enqueue channel verified notification"); - } + notify_channel_verified(state, user.id, channel, identifier).await; Ok(Json(VerifyTokenOutput { success: true, diff --git a/crates/tranquil-api/src/telegram_webhook.rs b/crates/tranquil-api/src/telegram_webhook.rs index 3c09421..dbc7659 100644 --- a/crates/tranquil-api/src/telegram_webhook.rs +++ b/crates/tranquil-api/src/telegram_webhook.rs @@ -71,7 +71,7 @@ pub async fn handle_telegram_webhook( "Received /start from Telegram user" ); match state - .user_repo + .repos.user .store_telegram_chat_id(&username, from.id, handle.as_deref()) .await { @@ -82,8 +82,8 @@ pub async fn handle_telegram_webhook( "Verified Telegram user and stored chat_id" ); if let Err(e) = comms_repo::enqueue_channel_verified( - state.user_repo.as_ref(), - state.infra_repo.as_ref(), + state.repos.user.as_ref(), + state.repos.infra.as_ref(), user_id, tranquil_db_traits::CommsChannel::Telegram, &from.id.to_string(), diff --git a/crates/tranquil-oauth-server/src/endpoints/delegation.rs b/crates/tranquil-oauth-server/src/endpoints/delegation.rs index de0f56a..4c025a9 100644 --- a/crates/tranquil-oauth-server/src/endpoints/delegation.rs +++ b/crates/tranquil-oauth-server/src/endpoints/delegation.rs @@ -25,7 +25,7 @@ fn parse_did(s: &str, label: &str) -> Result { async fn get_auth_request(state: &AppState, request_uri: &str) -> Result { let request_id = RequestId::from(request_uri.to_string()); match state - .oauth_repo + .repos.oauth .get_authorization_request(&request_id) .await { @@ -43,7 +43,7 @@ async fn get_delegation_grant( controller_did: &Did, ) -> Result { match state - .delegation_repo + .repos.delegation .get_delegation(delegated_did, controller_did) .await { @@ -65,7 +65,7 @@ async fn finalize_delegation_auth( user_agent: Option<&str>, ) -> Response { let _ = state - .delegation_repo + .repos.delegation .log_delegation_action( delegated_did, controller_did, @@ -87,12 +87,12 @@ async fn bind_delegation_to_request( ) -> Result<(), Response> { let request_id = RequestId::from(request_uri.to_string()); state - .oauth_repo + .repos.oauth .set_request_did(&request_id, delegated_did) .await .map_err(|_| DelegationAuthResponse::err("Failed to update authorization request"))?; state - .oauth_repo + .repos.oauth .set_controller_did(&request_id, controller_did) .await .map_err(|_| DelegationAuthResponse::err("Failed to update authorization request"))?; @@ -211,7 +211,7 @@ pub async fn delegation_auth( let is_cross_pds = form.auth_method.as_deref() == Some("cross_pds"); let controller_local = state - .user_repo + .repos.user .get_auth_info_by_did(&controller_did) .await .ok() @@ -562,7 +562,7 @@ pub async fn delegation_callback( } let _ = state - .delegation_repo + .repos.delegation .log_delegation_action( delegated_did, controller_did, diff --git a/crates/tranquil-oauth-server/src/endpoints/par.rs b/crates/tranquil-oauth-server/src/endpoints/par.rs index 93be937..a4dcb27 100644 --- a/crates/tranquil-oauth-server/src/endpoints/par.rs +++ b/crates/tranquil-oauth-server/src/endpoints/par.rs @@ -115,12 +115,12 @@ pub async fn pushed_authorization_request( }; let request_id_typed = RequestIdType::from(request_id.0.clone()); state - .oauth_repo + .repos.oauth .create_authorization_request(&request_id_typed, &request_data) .await .map_err(tranquil_pds::oauth::db_err_to_oauth)?; tokio::spawn({ - let oauth_repo = state.oauth_repo.clone(); + let oauth_repo = state.repos.oauth.clone(); async move { if let Err(e) = oauth_repo.delete_expired_authorization_requests().await { tracing::warn!("Failed to cleanup expired authorization requests: {:?}", e); diff --git a/crates/tranquil-oauth-server/src/endpoints/token/grants.rs b/crates/tranquil-oauth-server/src/endpoints/token/grants.rs index 1b3d693..195d52f 100644 --- a/crates/tranquil-oauth-server/src/endpoints/token/grants.rs +++ b/crates/tranquil-oauth-server/src/endpoints/token/grants.rs @@ -47,7 +47,7 @@ pub async fn handle_authorization_code_grant( }; let auth_code = AuthorizationCode::from(code); let auth_request = state - .oauth_repo + .repos.oauth .consume_authorization_request_by_code(&auth_code) .await .map_err(tranquil_pds::oauth::db_err_to_oauth)? @@ -104,7 +104,7 @@ pub async fn handle_authorization_code_grant( let token_endpoint = format!("https://{}/oauth/token", pds_hostname); let result = verifier.verify_proof(proof, Method::POST.as_str(), &token_endpoint, None)?; if !state - .oauth_repo + .repos.oauth .check_and_record_dpop_jti(&result.jti) .await .map_err(tranquil_pds::oauth::db_err_to_oauth)? @@ -140,7 +140,7 @@ pub async fn handle_authorization_code_grant( .parse() .map_err(|_| OAuthError::InvalidRequest("Invalid controller DID format".to_string()))?; let grant = state - .delegation_repo + .repos.delegation .get_delegation(&did_parsed, &controller_parsed) .await .ok() @@ -200,7 +200,7 @@ pub async fn handle_authorization_code_grant( controller_did: controller_did.clone(), }; state - .oauth_repo + .repos.oauth .create_token(&token_data) .await .map_err(tranquil_pds::oauth::db_err_to_oauth)?; @@ -211,7 +211,7 @@ pub async fn handle_authorization_code_grant( "Authorization code grant completed, token created" ); tokio::spawn({ - let oauth_repo = state.oauth_repo.clone(); + let oauth_repo = state.repos.oauth.clone(); let did_clone = did.clone(); async move { if let Ok(did_typed) = did_clone.parse::() @@ -267,7 +267,7 @@ pub async fn handle_refresh_token_grant( ); let refresh_token_typed = RefreshTokenType::from(refresh_token_str.clone()); - let lookup = lookup_refresh_token(state.oauth_repo.as_ref(), &refresh_token_typed).await?; + let lookup = lookup_refresh_token(state.repos.oauth.as_ref(), &refresh_token_typed).await?; let token_state = lookup.state(); tracing::debug!(state = %token_state, "Refresh token state"); @@ -320,7 +320,7 @@ pub async fn handle_refresh_token_grant( "Refresh token reuse detected, revoking token family" ); state - .oauth_repo + .repos.oauth .delete_token_family(original_token_id) .await .map_err(tranquil_pds::oauth::db_err_to_oauth)?; @@ -331,7 +331,7 @@ pub async fn handle_refresh_token_grant( RefreshTokenLookup::Expired { db_id } => { tracing::warn!(refresh_token_prefix = %token_prefix, "Refresh token has expired"); state - .oauth_repo + .repos.oauth .delete_token_family(db_id) .await .map_err(tranquil_pds::oauth::db_err_to_oauth)?; @@ -353,7 +353,7 @@ pub async fn handle_refresh_token_grant( let token_endpoint = format!("https://{}/oauth/token", pds_hostname); let result = verifier.verify_proof(proof, Method::POST.as_str(), &token_endpoint, None)?; if !state - .oauth_repo + .repos.oauth .check_and_record_dpop_jti(&result.jti) .await .map_err(tranquil_pds::oauth::db_err_to_oauth)? @@ -386,7 +386,7 @@ pub async fn handle_refresh_token_grant( let new_expires_at = Utc::now() + Duration::days(refresh_expiry_days); let new_refresh_typed = RefreshTokenType::from(new_refresh_token.0.clone()); state - .oauth_repo + .repos.oauth .rotate_token(db_id, &new_refresh_typed, new_expires_at) .await .map_err(tranquil_pds::oauth::db_err_to_oauth)?; diff --git a/crates/tranquil-oauth-server/src/endpoints/token/introspect.rs b/crates/tranquil-oauth-server/src/endpoints/token/introspect.rs index 45680f8..279f5d2 100644 --- a/crates/tranquil-oauth-server/src/endpoints/token/introspect.rs +++ b/crates/tranquil-oauth-server/src/endpoints/token/introspect.rs @@ -24,20 +24,20 @@ pub async fn revoke_token( if let Some(token) = &request.token { let refresh_token = RefreshToken::from(token.clone()); if let Some((db_id, _)) = state - .oauth_repo + .repos.oauth .get_token_by_refresh_token(&refresh_token) .await .map_err(tranquil_pds::oauth::db_err_to_oauth)? { state - .oauth_repo + .repos.oauth .delete_token_family(db_id) .await .map_err(tranquil_pds::oauth::db_err_to_oauth)?; } else { let token_id = TokenId::from(token.clone()); state - .oauth_repo + .repos.oauth .delete_token(&token_id) .await .map_err(tranquil_pds::oauth::db_err_to_oauth)?; @@ -104,7 +104,7 @@ pub async fn introspect_token( Err(_) => return Ok(Json(inactive_response)), }; let token_id = TokenId::from(token_info.sid.clone()); - let token_data = match state.oauth_repo.get_token_by_id(&token_id).await { + let token_data = match state.repos.oauth.get_token_by_id(&token_id).await { Ok(Some(data)) => data, _ => return Ok(Json(inactive_response)), }; diff --git a/crates/tranquil-oauth-server/src/sso_endpoints.rs b/crates/tranquil-oauth-server/src/sso_endpoints.rs index 136ed73..a72f30c 100644 --- a/crates/tranquil-oauth-server/src/sso_endpoints.rs +++ b/crates/tranquil-oauth-server/src/sso_endpoints.rs @@ -100,7 +100,7 @@ pub async fn sso_initiate( let extracted = extract_auth_token_from_header(auth_header).ok_or(ApiError::SsoNotAuthenticated)?; let auth_user = validate_bearer_token_cached( - state.user_repo.as_ref(), + state.repos.user.as_ref(), state.cache.as_ref(), &extracted.token, ) @@ -112,7 +112,7 @@ pub async fn sso_initiate( _ => { let request_id = RequestId::new(request_uri.clone()); let _request_data = state - .oauth_repo + .repos.oauth .get_authorization_request(&request_id) .await? .ok_or(ApiError::InvalidRequest( @@ -135,7 +135,7 @@ pub async fn sso_initiate( })?; state - .sso_repo + .repos.sso .create_sso_auth_state( &sso_state, &request_uri, @@ -230,7 +230,7 @@ async fn sso_callback_internal(state: &AppState, query: SsoCallbackQuery) -> Res _ => return redirect_to_error("Missing code or state parameter"), }; - let auth_state = match state.sso_repo.consume_sso_auth_state(&sso_state).await { + let auth_state = match state.repos.sso.consume_sso_auth_state(&sso_state).await { Ok(Some(s)) => s, Ok(None) => return redirect_to_error("SSO session expired or invalid"), Err(e) => { @@ -363,7 +363,7 @@ async fn handle_sso_login( user_info: &tranquil_pds::sso::providers::SsoUserInfo, ) -> Response { let identity = match state - .sso_repo + .repos.sso .get_external_identity_by_provider(provider, &user_info.provider_user_id) .await { @@ -371,7 +371,7 @@ async fn handle_sso_login( Ok(None) => { let token = generate_registration_token(); if let Err(e) = state - .sso_repo + .repos.sso .create_pending_registration( &token, request_uri, @@ -398,7 +398,7 @@ async fn handle_sso_login( } }; - let is_verified = match state.user_repo.get_session_info_by_did(&identity.did).await { + let is_verified = match state.repos.user.get_session_info_by_did(&identity.did).await { Ok(Some(info)) => info.channel_verification.has_any_verified(), Ok(None) => { tracing::error!("User not found for SSO login: {}", identity.did); @@ -423,7 +423,7 @@ async fn handle_sso_login( } if let Err(e) = state - .sso_repo + .repos.sso .update_external_identity_login( identity.id, user_info.username.as_deref(), @@ -436,7 +436,7 @@ async fn handle_sso_login( let request_id = RequestId::new(request_uri.to_string()); if let Err(e) = state - .oauth_repo + .repos.oauth .set_authorization_did(&request_id, &identity.did, None) .await { @@ -452,7 +452,7 @@ async fn handle_sso_login( ); let has_totp = matches!( - state.user_repo.get_totp_record_state(&identity.did).await, + state.repos.user.get_totp_record_state(&identity.did).await, Ok(Some(tranquil_db_traits::TotpRecordState::Verified(_))) ); @@ -478,7 +478,7 @@ async fn handle_sso_link( user_info: &tranquil_pds::sso::providers::SsoUserInfo, ) -> Response { let existing = state - .sso_repo + .repos.sso .get_external_identity_by_provider(provider, &user_info.provider_user_id) .await; @@ -517,7 +517,7 @@ async fn handle_sso_link( } if let Err(e) = state - .sso_repo + .repos.sso .create_external_identity( &did, provider, @@ -551,7 +551,7 @@ async fn handle_sso_register( user_info: &tranquil_pds::sso::providers::SsoUserInfo, ) -> Response { match state - .sso_repo + .repos.sso .get_external_identity_by_provider(provider, &user_info.provider_user_id) .await { @@ -569,7 +569,7 @@ async fn handle_sso_register( let token = generate_registration_token(); if let Err(e) = state - .sso_repo + .repos.sso .create_pending_registration( &token, request_uri, @@ -612,7 +612,7 @@ pub async fn get_linked_accounts( auth: tranquil_pds::auth::Auth, ) -> Result, ApiError> { let identities = state - .sso_repo + .repos.sso .get_external_identities_by_did(&auth.did) .await?; @@ -657,17 +657,17 @@ pub async fn unlink_account( let id = uuid::Uuid::parse_str(&input.id).map_err(|_| ApiError::InvalidId)?; let has_password = state - .user_repo + .repos.user .has_password_by_did(&auth.did) .await? .unwrap_or(false); - let passkeys = state.user_repo.get_passkeys_for_user(&auth.did).await?; + let passkeys = state.repos.user.get_passkeys_for_user(&auth.did).await?; let has_passkeys = !passkeys.is_empty(); if !has_password && !has_passkeys { let identities = state - .sso_repo + .repos.sso .get_external_identities_by_did(&auth.did) .await?; @@ -680,7 +680,7 @@ pub async fn unlink_account( } let deleted = state - .sso_repo + .repos.sso .delete_external_identity(id, &auth.did) .await?; @@ -718,7 +718,7 @@ pub async fn get_pending_registration( } let pending = state - .sso_repo + .repos.sso .get_pending_registration(&query.token) .await? .ok_or(ApiError::SsoSessionExpired)?; @@ -780,7 +780,7 @@ pub async fn check_handle_available( }; let db_available = state - .user_repo + .repos.user .check_handle_available_for_new_account(&handle_typed) .await .unwrap_or(false); @@ -850,7 +850,7 @@ pub async fn complete_registration( } let pending_preview = state - .sso_repo + .repos.sso .get_pending_registration(&input.token) .await? .ok_or(ApiError::SsoSessionExpired)?; @@ -962,7 +962,7 @@ pub async fn complete_registration( }; let _validated_invite_code = if let Some(ref code) = input.invite_code { - match state.infra_repo.validate_invite_code(code).await { + match state.repos.infra.validate_invite_code(code).await { Ok(validated) => Some(validated), Err(_) => return Err(ApiError::InvalidInviteCode), } @@ -977,7 +977,7 @@ pub async fn complete_registration( let handle_typed: tranquil_pds::types::Handle = handle.parse().map_err(|_| ApiError::InvalidHandle(None))?; let reserved = state - .user_repo + .repos.user .reserve_handle(&handle_typed, client_ip) .await .unwrap_or(false); @@ -1160,7 +1160,7 @@ pub async fn complete_registration( pending_registration_token: input.token.clone(), }; - let create_result = match state.user_repo.create_sso_account(&create_input).await { + let create_result = match state.repos.user.create_sso_account(&create_input).await { Ok(r) => r, Err(tranquil_db_traits::CreateAccountError::HandleTaken) => { return Err(ApiError::HandleNotAvailable(None)); @@ -1178,7 +1178,7 @@ pub async fn complete_registration( }; let _ = state - .user_repo + .repos.user .release_handle_reservation(&handle_typed) .await; @@ -1216,13 +1216,8 @@ pub async fn complete_registration( let app_password = generate_app_password(); let app_password_name = "bsky.app".to_string(); - let app_password_hash = match bcrypt::hash(&app_password, bcrypt::DEFAULT_COST) { - Ok(h) => h, - Err(e) => { - tracing::error!("Failed to hash app password: {:?}", e); - return Err(ApiError::InternalError(None)); - } - }; + let app_password_hash = + tranquil_api::common::hash_or_internal_error(&app_password)?; let app_password_data = tranquil_db_traits::AppPasswordCreate { user_id: create_result.user_id, @@ -1233,7 +1228,7 @@ pub async fn complete_registration( created_by_controller_did: None, }; if let Err(e) = state - .session_repo + .repos.session .create_app_password(&app_password_data) .await { @@ -1245,7 +1240,7 @@ pub async fn complete_registration( if !is_standalone { let request_id = RequestId::new(pending_preview.request_uri.clone()); if let Err(e) = state - .oauth_repo + .repos.oauth .set_authorization_did(&request_id, &did_typed, None) .await { @@ -1264,7 +1259,7 @@ pub async fn complete_registration( ); let user_id = state - .user_repo + .repos.user .get_id_by_did(&did_typed) .await .unwrap_or(None); @@ -1275,7 +1270,7 @@ pub async fn complete_registration( if channel_auto_verified { let _ = state - .user_repo + .repos.user .set_channel_verified(&did_typed, tranquil_db_traits::CommsChannel::Email) .await; tracing::info!(did = %did, "Auto-verified email from SSO provider"); @@ -1321,15 +1316,15 @@ pub async fn complete_registration( controller_did: None, app_password_name: None, }; - if let Err(e) = state.session_repo.create_session(&session_data).await { + if let Err(e) = state.repos.session.create_session(&session_data).await { tracing::error!("Failed to insert session: {:?}", e); return Err(ApiError::InternalError(None)); } let hostname = &tranquil_config::get().server.hostname; if let Err(e) = tranquil_pds::comms::comms_repo::enqueue_welcome( - state.user_repo.as_ref(), - state.infra_repo.as_ref(), + state.repos.user.as_ref(), + state.repos.infra.as_ref(), user_id.unwrap_or(uuid::Uuid::nil()), hostname, ) @@ -1372,8 +1367,8 @@ pub async fn complete_registration( let formatted_token = tranquil_pds::auth::verification_token::format_token_for_display(&verification_token); if let Err(e) = tranquil_pds::comms::comms_repo::enqueue_signup_verification( - state.user_repo.as_ref(), - state.infra_repo.as_ref(), + state.repos.user.as_ref(), + state.repos.infra.as_ref(), uid, verification_channel, &verification_recipient, diff --git a/crates/tranquil-server/src/main.rs b/crates/tranquil-server/src/main.rs index 03250a7..c7f2ce4 100644 --- a/crates/tranquil-server/src/main.rs +++ b/crates/tranquil-server/src/main.rs @@ -127,7 +127,7 @@ async fn run() -> Result<(), Box> { tranquil_sync::listener::start_sequencer_listener(state.clone()).await; - let backfill_repo_repo = state.repo_repo.clone(); + let backfill_repo_repo = state.repos.repo.clone(); let backfill_block_store = state.block_store.clone(); tokio::spawn(async move { tokio::join!( @@ -141,7 +141,7 @@ async fn run() -> Result<(), Box> { ); }); - let mut comms_service = CommsService::new(state.infra_repo.clone()); + let mut comms_service = CommsService::new(state.repos.infra.clone()); let mut deferred_discord_endpoint: Option<(DiscordSender, String, String)> = None; let cfg = tranquil_config::get(); @@ -249,10 +249,10 @@ async fn run() -> Result<(), Box> { }; let scheduled_handle = tokio::spawn(start_scheduled_tasks( - state.user_repo.clone(), - state.blob_repo.clone(), + state.repos.user.clone(), + state.repos.blob.clone(), state.blob_store.clone(), - state.sso_repo.clone(), + state.repos.sso.clone(), shutdown.clone(), )); diff --git a/crates/tranquil-sync/src/blob.rs b/crates/tranquil-sync/src/blob.rs index 09c8a33..18f8bc4 100644 --- a/crates/tranquil-sync/src/blob.rs +++ b/crates/tranquil-sync/src/blob.rs @@ -27,14 +27,14 @@ pub async fn get_blob( let cid = params.cid; let _account = - match assert_repo_availability(state.repo_repo.as_ref(), &did, RepoAccessLevel::Public) + match assert_repo_availability(state.repos.repo.as_ref(), &did, RepoAccessLevel::Public) .await { Ok(a) => a, Err(e) => return e.into_response(), }; - let blob_result = state.blob_repo.get_blob_metadata(&cid).await; + let blob_result = state.repos.blob.get_blob_metadata(&cid).await; match blob_result { Ok(Some(metadata)) => match state.blob_store.get(&metadata.storage_key).await { Ok(data) => Response::builder() @@ -80,7 +80,7 @@ pub async fn list_blobs( let did = params.did; let account = - match assert_repo_availability(state.repo_repo.as_ref(), &did, RepoAccessLevel::Public) + match assert_repo_availability(state.repos.repo.as_ref(), &did, RepoAccessLevel::Public) .await { Ok(a) => a, @@ -93,7 +93,7 @@ pub async fn list_blobs( let cids_result: Result, _> = if let Some(since) = ¶ms.since { state - .blob_repo + .repos.blob .list_blobs_since_rev(&did, since) .await .map(|cids| { @@ -107,7 +107,7 @@ pub async fn list_blobs( }) } else { state - .blob_repo + .repos.blob .list_blobs_by_user(user_id, Some(cursor_cid), limit + 1) .await .map(|cids| cids.into_iter().map(|c| c.to_string()).collect()) diff --git a/crates/tranquil-sync/src/commit.rs b/crates/tranquil-sync/src/commit.rs index d9c6a48..0123ccf 100644 --- a/crates/tranquil-sync/src/commit.rs +++ b/crates/tranquil-sync/src/commit.rs @@ -43,7 +43,7 @@ pub async fn get_latest_commit( let did = params.did; let account = - match assert_repo_availability(state.repo_repo.as_ref(), &did, RepoAccessLevel::Public) + match assert_repo_availability(state.repos.repo.as_ref(), &did, RepoAccessLevel::Public) .await { Ok(a) => a, @@ -104,7 +104,7 @@ pub async fn list_repos( let cursor_did: Option = params.cursor.as_ref().and_then(|s| s.parse().ok()); let cursor_ref = cursor_did.as_ref(); let result = state - .repo_repo + .repos.repo .list_repos_paginated(cursor_ref, limit + 1) .await; match result { @@ -175,7 +175,7 @@ pub async fn get_repo_status( ) -> Response { let did = params.did; - let account = match get_account_with_status(state.repo_repo.as_ref(), &did).await { + let account = match get_account_with_status(state.repos.repo.as_ref(), &did).await { Ok(Some(a)) => a, Ok(None) => { return ApiError::RepoNotFound(Some(format!("Could not find repo for DID: {}", did))) diff --git a/crates/tranquil-sync/src/deprecated.rs b/crates/tranquil-sync/src/deprecated.rs index 1330e21..c17a010 100644 --- a/crates/tranquil-sync/src/deprecated.rs +++ b/crates/tranquil-sync/src/deprecated.rs @@ -27,8 +27,8 @@ async fn check_admin_or_self(state: &AppState, headers: &HeaderMap, did: &Did) - let dpop_proof = tranquil_pds::util::get_header_str(headers, tranquil_pds::util::HEADER_DPOP); let http_uri = "/"; match tranquil_pds::auth::validate_token_with_dpop( - state.user_repo.as_ref(), - state.oauth_repo.as_ref(), + state.repos.user.as_ref(), + state.repos.oauth.as_ref(), &extracted.token, extracted.scheme, dpop_proof, @@ -68,7 +68,7 @@ pub async fn get_head( }; let is_admin_or_self = check_admin_or_self(&state, &headers, &did).await; let account = match assert_repo_availability( - state.repo_repo.as_ref(), + state.repos.repo.as_ref(), &did, if is_admin_or_self { RepoAccessLevel::Privileged @@ -108,7 +108,7 @@ pub async fn get_checkout( }; let is_admin_or_self = check_admin_or_self(&state, &headers, &did).await; let account = match assert_repo_availability( - state.repo_repo.as_ref(), + state.repos.repo.as_ref(), &did, if is_admin_or_self { RepoAccessLevel::Privileged diff --git a/crates/tranquil-sync/src/listener.rs b/crates/tranquil-sync/src/listener.rs index 5749cb3..95cdf7f 100644 --- a/crates/tranquil-sync/src/listener.rs +++ b/crates/tranquil-sync/src/listener.rs @@ -8,7 +8,7 @@ static LAST_BROADCAST_SEQ: AtomicI64 = AtomicI64::new(0); pub async fn start_sequencer_listener(state: AppState) { let initial_seq = state - .repo_repo + .repos.repo .get_max_seq() .await .unwrap_or(SequenceNumber::ZERO); @@ -30,14 +30,14 @@ pub async fn start_sequencer_listener(state: AppState) { async fn listen_loop(state: AppState) -> anyhow::Result<()> { let mut receiver = state - .event_notifier + .repos.event_notifier .subscribe() .await .map_err(|e| anyhow::anyhow!("Failed to subscribe to events: {:?}", e))?; info!("Connected to database and listening for repo updates"); let catchup_start = SequenceNumber::from_raw(LAST_BROADCAST_SEQ.load(Ordering::SeqCst)); let events = state - .repo_repo + .repos.repo .get_events_since_seq(catchup_start, None) .await .map_err(|e| anyhow::anyhow!("Failed to fetch catchup events: {:?}", e))?; @@ -70,7 +70,7 @@ async fn listen_loop(state: AppState) -> anyhow::Result<()> { } if seq_id > last_seq + 1 { let gap_events = state - .repo_repo + .repos.repo .get_events_in_seq_range( SequenceNumber::from_raw(last_seq), SequenceNumber::from_raw(seq_id), @@ -88,7 +88,7 @@ async fn listen_loop(state: AppState) -> anyhow::Result<()> { } } let event = state - .repo_repo + .repos.repo .get_event_by_seq(SequenceNumber::from_raw(seq_id)) .await .ok() diff --git a/crates/tranquil-sync/src/repo.rs b/crates/tranquil-sync/src/repo.rs index efebea4..84b1194 100644 --- a/crates/tranquil-sync/src/repo.rs +++ b/crates/tranquil-sync/src/repo.rs @@ -46,7 +46,7 @@ pub async fn get_blocks(State(state): State, RawQuery(query): RawQuery }; let _account = - match assert_repo_availability(state.repo_repo.as_ref(), &did, RepoAccessLevel::Public) + match assert_repo_availability(state.repos.repo.as_ref(), &did, RepoAccessLevel::Public) .await { Ok(a) => a, @@ -123,7 +123,7 @@ pub async fn get_repo( ) -> Response { let did = query.did; let account = - match assert_repo_availability(state.repo_repo.as_ref(), &did, RepoAccessLevel::Public) + match assert_repo_availability(state.repos.repo.as_ref(), &did, RepoAccessLevel::Public) .await { Ok(a) => a, @@ -143,7 +143,7 @@ pub async fn get_repo( } let car_bytes = match generate_repo_car_from_user_blocks( - state.repo_repo.as_ref(), + state.repos.repo.as_ref(), &state.block_store, account.user_id, &head_cid, @@ -166,7 +166,7 @@ pub async fn get_repo( } async fn get_repo_since(state: &AppState, did: &Did, head_cid: &Cid, since: &str) -> Response { - let user_id = match state.user_repo.get_id_by_did(did).await { + let user_id = match state.repos.user.get_id_by_did(did).await { Ok(Some(id)) => id, Ok(None) => { return ApiError::RepoNotFound(Some(format!("Could not find repo for DID: {}", did))) @@ -179,7 +179,7 @@ async fn get_repo_since(state: &AppState, did: &Did, head_cid: &Cid, since: &str }; let block_cid_bytes = match state - .repo_repo + .repos.repo .get_user_block_cids_since_rev(user_id, since) .await { @@ -252,7 +252,7 @@ pub async fn get_record( let did = query.did; let account = - match assert_repo_availability(state.repo_repo.as_ref(), &did, RepoAccessLevel::Public) + match assert_repo_availability(state.repos.repo.as_ref(), &did, RepoAccessLevel::Public) .await { Ok(a) => a, diff --git a/crates/tranquil-sync/src/subscribe_repos.rs b/crates/tranquil-sync/src/subscribe_repos.rs index be53718..5d62ae1 100644 --- a/crates/tranquil-sync/src/subscribe_repos.rs +++ b/crates/tranquil-sync/src/subscribe_repos.rs @@ -73,7 +73,7 @@ async fn handle_socket_inner( if let Some(cursor) = params.cursor { let cursor_seq = SequenceNumber::from_raw(cursor); let current_seq = state - .repo_repo + .repos.repo .get_max_seq() .await .unwrap_or(SequenceNumber::ZERO); @@ -91,7 +91,7 @@ async fn handle_socket_inner( let backfill_time = chrono::Utc::now() - chrono::Duration::hours(get_backfill_hours()); let first_event = state - .repo_repo + .repos.repo .get_events_since_cursor(cursor_seq, 1) .await .ok() @@ -110,7 +110,7 @@ async fn handle_socket_inner( } let earliest = state - .repo_repo + .repos.repo .get_min_seq_since(backfill_time) .await .ok() @@ -125,7 +125,7 @@ async fn handle_socket_inner( loop { let events = state - .repo_repo + .repos.repo .get_events_since_cursor(current_cursor, BACKFILL_BATCH_SIZE) .await; match events { @@ -171,7 +171,7 @@ async fn handle_socket_inner( } } - let cutover_events = state.repo_repo.get_events_since_seq(last_seen, None).await; + let cutover_events = state.repos.repo.get_events_since_seq(last_seen, None).await; if let Ok(events) = cutover_events && !events.is_empty()