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

refactor(api): migrate all endpoints to repos accessor pattern

Browse Source
This commit is contained in:
Lewis
2026-03-22 12:51:44 +00:00
committed by Tangled
parent 19b0ea19c6
commit e454e99b8a
60 changed files with 520 additions and 615 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/tranquil-api/src/actor/preferences.rs
+4 -4
View File
@@ -34,13 +34,13 @@ pub struct GetPreferencesOutput {
}
pub async fn get_preferences(State(state): State<AppState>, auth: Auth<Permissive>) -> 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<PutPreferencesInput>,
) -> 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()
crates/tranquil-api/src/admin/account/delete.rs
+2 -2
View File
@@ -19,7 +19,7 @@ pub async fn delete_account(
) -> Result<Json<EmptyResponse>, 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")?;
crates/tranquil-api/src/admin/account/email.rs
+2 -2
View File
@@ -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,
crates/tranquil-api/src/admin/account/info.rs
+10 -10
View File
@@ -69,7 +69,7 @@ pub async fn get_account_info(
Query(params): Query<GetAccountInfoParams>,
) -> Result<Json<AccountInfo>, ApiError> {
let account = state
.infra_repo
.repos.infra
.get_admin_account_info_by_did(&params.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<InviteCodeInfo> {
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<Vec<InviteCodeInfo>> {
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<String> = 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<InviteCodeInfo> {
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<Did> = 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<uuid::Uuid> = 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<uuid::Uuid, String> = state
.infra_repo
.repos.infra
.get_invite_code_uses_by_users(&user_ids)
.await
.unwrap_or_default()
crates/tranquil-api/src/admin/account/search.rs
+1 -1
View File
@@ -55,7 +55,7 @@ pub async fn search_accounts(
let handle_filter = params.handle.as_deref().map(|h| format!("%{}%", h));
let cursor_did: Option<Did> = 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(),
crates/tranquil-api/src/admin/account/update.rs
+7 -10
View File
@@ -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
{
crates/tranquil-api/src/admin/config.rs
+15 -15
View File
@@ -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")?;
crates/tranquil-api/src/admin/invite.rs
+7 -7
View File
@@ -24,7 +24,7 @@ pub async fn disable_invite_codes(
Json(input): Json<DisableInviteCodesInput>,
) -> Result<Json<EmptyResponse>, 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<tranquil_types::Did> =
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<String> = codes_rows.iter().map(|r| r.code.clone()).collect();
let creator_dids: std::collections::HashMap<uuid::Uuid, tranquil_types::Did> = 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
{
crates/tranquil-api/src/admin/server_stats.rs
+4 -4
View File
@@ -17,10 +17,10 @@ pub async fn get_server_stats(
State(state): State<AppState>,
_auth: Auth<Admin>,
) -> Result<Json<ServerStatsOutput>, 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,
crates/tranquil-api/src/admin/status.rs
+9 -9
View File
@@ -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| {
crates/tranquil-api/src/age_assurance.rs
+3 -3
View File
@@ -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())
crates/tranquil-api/src/common.rs
+15 -1
View File
@@ -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<String, ApiError> {
})
}
pub async fn hash_password_async(password: &str) -> Result<String, ApiError> {
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<DateTime<Utc>>,
stored_hash: &str,
crates/tranquil-api/src/delegation.rs
+18 -18
View File
@@ -24,7 +24,7 @@ pub async fn list_controllers(
auth: Auth<Active>,
) -> Result<Json<ControllersOutput<Vec<tranquil_db_traits::ControllerInfo>>>, 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<RemoveControllerInput>,
) -> Result<Json<SuccessResponse>, 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<UpdateControllerScopesInput>,
) -> Result<Json<SuccessResponse>, 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<Active>,
) -> Result<Json<AccountsOutput<Vec<tranquil_db_traits::DelegatedAccountInfo>>>, 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<Handle> = 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 {
crates/tranquil-api/src/discord_webhook.rs
+3 -3
View File
@@ -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,
crates/tranquil-api/src/identity/account.rs
+11 -21
View File
@@ -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<chrono::DateTime<chrono::Utc>> = 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();
crates/tranquil-api/src/identity/did.rs
+12 -12
View File
@@ -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<AppState>, Path(handle): Path<Stri
}
};
let user = match state
.user_repo
.repos.user
.get_did_web_info_by_handle(&current_handle_typed)
.await
{
@@ -246,7 +246,7 @@ pub async fn user_did_doc(State(state): State<AppState>, Path(handle): Path<Stri
}
let overrides = state
.user_repo
.repos.user
.get_did_web_overrides(user_id)
.await
.ok()
@@ -295,7 +295,7 @@ async fn build_override_or_key_verification_methods(
.collect());
}
let key_info = match state.user_repo.get_user_key_by_id(user_id).await {
let key_info = match state.repos.user.get_user_key_by_id(user_id).await {
Ok(Some(k)) => k,
_ => return Err(ApiError::InternalError(None).into_response()),
};
@@ -468,7 +468,7 @@ pub async fn get_recommended_did_credentials(
auth: Auth<NotTakendown>,
) -> Result<Json<GetRecommendedDidCredentialsOutput>, 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<AppState>, 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(),
crates/tranquil-api/src/identity/plc/request.rs
+5 -5
View File
@@ -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,
crates/tranquil-api/src/identity/plc/sign.rs
+5 -5
View File
@@ -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,
crates/tranquil-api/src/identity/plc/submit.rs
+4 -4
View File
@@ -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);
}
}
crates/tranquil-api/src/identity/provision.rs
+6 -6
View File
@@ -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,
crates/tranquil-api/src/moderation/mod.rs
+2 -2
View File
@@ -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(),
crates/tranquil-api/src/notification_prefs.rs
+16 -16
View File
@@ -26,7 +26,7 @@ pub async fn get_notification_prefs(
auth: Auth<Active>,
) -> Result<Json<NotificationPrefsOutput>, 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<Active>,
) -> Result<Json<GetNotificationHistoryOutput>, 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<UpdateNotificationPrefsInput>,
) -> Result<Json<UpdateNotificationPrefsOutput>, 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")?;
crates/tranquil-api/src/repo/blob.rs
+7 -7
View File
@@ -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<Json<ListMissingBlobsOutput>, 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")?;
crates/tranquil-api/src/repo/import.rs
+10 -10
View File
@@ -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<Vec<u8>> = 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(())
}
crates/tranquil-api/src/repo/meta.rs
+2 -2
View File
@@ -18,12 +18,12 @@ pub async fn describe_repo(
State(state): State<AppState>,
Query(input): Query<DescribeRepoInput>,
) -> 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();
crates/tranquil-api/src/repo/record/batch.rs
+2 -2
View File
@@ -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")?
crates/tranquil-api/src/repo/record/delete.rs
+3 -3
View File
@@ -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()))?
crates/tranquil-api/src/repo/record/read.rs
+4 -4
View File
@@ -60,12 +60,12 @@ pub async fn get_record(
_headers: HeaderMap,
Query(input): Query<GetRecordInput>,
) -> 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<AppState>,
Query(input): Query<ListRecordsInput>,
) -> 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::<tranquil_pds::types::Rkey>().ok());
let rows = match state
.repo_repo
.repos.repo
.list_records(
user_id,
&input.collection,
crates/tranquil-api/src/repo/record/write.rs
+4 -4
View File
@@ -46,7 +46,7 @@ pub async fn prepare_repo_write<A: RepoScopeAction>(
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);
}
crates/tranquil-api/src/server/account_status.rs
+24 -24
View File
@@ -41,24 +41,24 @@ pub async fn check_account_status(
) -> Result<Json<CheckAccountStatusOutput>, 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
crates/tranquil-api/src/server/app_password.rs
+12 -24
View File
@@ -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<Permissive>,
) -> Result<Json<ListAppPasswordsOutput>, 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<CreateAppPasswordInput>,
) -> Result<Json<CreateAppPasswordOutput>, 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<RevokeAppPasswordInput>,
) -> Result<Json<EmptyResponse>, 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")?;
crates/tranquil-api/src/server/email.rs
+32 -48
View File
@@ -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<PendingEmailUpdate> {
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::<PendingEmailUpdate>(&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<CheckEmailVerifiedInput>,
) -> Result<Json<VerifiedResponse>, 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<CheckChannelVerifiedInput>,
) -> Result<Json<VerifiedResponse>, 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<Json<EmailUpdateStatusOutput>, 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| {
crates/tranquil-api/src/server/invite.rs
+6 -24
View File
@@ -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<String> = (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)
crates/tranquil-api/src/server/logo.rs
+2 -2
View File
@@ -9,7 +9,7 @@ use tracing::error;
use tranquil_pds::state::AppState;
pub async fn get_logo(State(state): State<AppState>) -> 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<AppState>) -> 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) => {
crates/tranquil-api/src/server/meta.rs
+1 -1
View File
@@ -98,7 +98,7 @@ pub struct HealthOutput {
}
pub async fn health(State(state): State<AppState>) -> impl IntoResponse {
match state.infra_repo.health_check().await {
match state.repos.infra.health_check().await {
Ok(true) => (
StatusCode::OK,
Json(HealthOutput {
crates/tranquil-api/src/server/migration.rs
+6 -6
View File
@@ -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<Vec<String>> = 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()
crates/tranquil-api/src/server/passkey_account.rs
+20 -20
View File
@@ -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<AppState>,
Json(input): Json<CompletePasskeySetupInput>,
) -> Result<Json<CompletePasskeySetupOutput>, 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<AppState>,
Json(input): Json<StartPasskeyRegistrationInput>,
) -> Result<Json<OptionsResponse<serde_json::Value>>, 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
{
crates/tranquil-api/src/server/passkeys.rs
+10 -10
View File
@@ -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<Active>,
) -> Result<Json<ListPasskeysOutput>, 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)
}
crates/tranquil-api/src/server/password.rs
+21 -54
View File
@@ -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<Active>,
) -> Result<Json<HasPasswordResponse>, 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")?;
crates/tranquil-api/src/server/reauth.rs
+13 -13
View File
@@ -33,13 +33,13 @@ pub async fn get_reauth_status(
auth: Auth<Active>,
) -> Result<Json<ReauthStatusOutput>, 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<PasswordReauthInput>,
) -> Result<Json<ReauthOutput>, 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<Active>,
) -> Result<Json<PasskeyReauthStartOutput>, 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<PasskeyReauthFinishInput>,
) -> Result<Json<ReauthOutput>, 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")?;
crates/tranquil-api/src/server/service_auth.rs
+1 -1
View File
@@ -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(
crates/tranquil-api/src/server/session.rs
+32 -32
View File
@@ -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<Json<EmptyResponse>, 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<ConfirmSignupInput>,
) -> Result<Json<ConfirmSignupOutput>, 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<AutoResendResult> {
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<Json<SuccessResponse>, 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<Active>,
) -> Result<Json<LegacyLoginPreferenceOutput>, 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")?;
crates/tranquil-api/src/server/signing_key.rs
+1 -1
View File
@@ -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,
crates/tranquil-api/src/server/totp.rs
+14 -14
View File
@@ -29,7 +29,7 @@ pub async fn create_totp_secret(
) -> Result<Json<CreateTotpSecretOutput>, 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<Json<GetTotpStatusOutput>, 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)
}
crates/tranquil-api/src/server/trusted_devices.rs
+5 -5
View File
@@ -72,7 +72,7 @@ pub async fn list_trusted_devices(
auth: Auth<Active>,
) -> Result<Json<ListTrustedDevicesOutput>, 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<RevokeTrustedDeviceInput>,
) -> Result<Json<SuccessResponse>, 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<UpdateTrustedDeviceInput>,
) -> Result<Json<SuccessResponse>, 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")?;
crates/tranquil-api/src/server/verify_email.rs
+1 -1
View File
@@ -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 }));
crates/tranquil-api/src/server/verify_token.rs
+30 -42
View File
@@ -79,7 +79,7 @@ async fn handle_migration_verification(
identifier: &str,
) -> Result<Json<VerifyTokenOutput>, 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<Json<VerifyTokenOutput>, 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<Json<VerifyTokenOutput>, 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,
crates/tranquil-api/src/telegram_webhook.rs
+3 -3
View File
@@ -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(),
crates/tranquil-oauth-server/src/endpoints/delegation.rs
+7 -7
View File
@@ -25,7 +25,7 @@ fn parse_did(s: &str, label: &str) -> Result<Did, Response> {
async fn get_auth_request(state: &AppState, request_uri: &str) -> Result<RequestData, Response> {
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<tranquil_db_traits::DelegationGrant, Response> {
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,
crates/tranquil-oauth-server/src/endpoints/par.rs
+2 -2
View File
@@ -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);
crates/tranquil-oauth-server/src/endpoints/token/grants.rs
+10 -10
View File
@@ -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::<tranquil_types::Did>()
@@ -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)?;
crates/tranquil-oauth-server/src/endpoints/token/introspect.rs
+4 -4
View File
@@ -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)),
};
crates/tranquil-oauth-server/src/sso_endpoints.rs
+37 -42
View File
@@ -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<tranquil_pds::auth::Active>,
) -> Result<Json<LinkedAccountsResponse>, 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,
crates/tranquil-server/src/main.rs
+5 -5
View File
@@ -127,7 +127,7 @@ async fn run() -> Result<(), Box<dyn std::error::Error>> {
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<dyn std::error::Error>> {
);
});
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<dyn std::error::Error>> {
};
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(),
));
crates/tranquil-sync/src/blob.rs
+5 -5
View File
@@ -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<Vec<String>, _> = if let Some(since) = &params.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())
crates/tranquil-sync/src/commit.rs
+3 -3
View File
@@ -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<Did> = 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)))
crates/tranquil-sync/src/deprecated.rs
+4 -4
View File
@@ -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
crates/tranquil-sync/src/listener.rs
+5 -5
View File
@@ -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()
crates/tranquil-sync/src/repo.rs
+6 -6
View File
@@ -46,7 +46,7 @@ pub async fn get_blocks(State(state): State<AppState>, 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,
crates/tranquil-sync/src/subscribe_repos.rs
+5 -5
View File
@@ -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()