Something went wrong. Try again.
Our Personal Data Server from scratch!
Something went wrong. Try again.
116 kB · 3465 lines
Rust
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864186518661867186818691870187118721873187418751876187718781879188018811882188318841885188618871888188918901891189218931894189518961897189818991900190119021903190419051906190719081909191019111912191319141915191619171918191919201921192219231924192519261927192819291930193119321933193419351936193719381939194019411942194319441945194619471948194919501951195219531954195519561957195819591960196119621963196419651966196719681969197019711972197319741975197619771978197919801981198219831984198519861987198819891990199119921993199419951996199719981999200020012002200320042005200620072008200920102011201220132014201520162017201820192020202120222023202420252026202720282029203020312032203320342035203620372038203920402041204220432044204520462047204820492050205120522053205420552056205720582059206020612062206320642065206620672068206920702071207220732074207520762077207820792080208120822083208420852086208720882089209020912092209320942095209620972098209921002101210221032104210521062107210821092110211121122113211421152116211721182119212021212122212321242125212621272128212921302131213221332134213521362137213821392140214121422143214421452146214721482149215021512152215321542155215621572158215921602161216221632164216521662167216821692170217121722173217421752176217721782179218021812182218321842185218621872188218921902191219221932194219521962197219821992200220122022203220422052206220722082209221022112212221322142215221622172218221922202221222222232224222522262227222822292230223122322233223422352236223722382239224022412242224322442245224622472248224922502251225222532254225522562257225822592260226122622263226422652266226722682269227022712272227322742275227622772278227922802281228222832284228522862287228822892290229122922293229422952296229722982299230023012302230323042305230623072308230923102311231223132314231523162317231823192320232123222323232423252326232723282329233023312332233323342335233623372338233923402341234223432344234523462347234823492350235123522353235423552356235723582359236023612362236323642365236623672368236923702371237223732374237523762377237823792380238123822383238423852386238723882389239023912392239323942395239623972398239924002401240224032404240524062407240824092410241124122413241424152416241724182419242024212422242324242425242624272428242924302431243224332434243524362437243824392440244124422443244424452446244724482449245024512452245324542455245624572458245924602461246224632464246524662467246824692470247124722473247424752476247724782479248024812482248324842485248624872488248924902491249224932494249524962497249824992500250125022503250425052506250725082509251025112512251325142515251625172518251925202521252225232524252525262527252825292530253125322533253425352536253725382539254025412542254325442545254625472548254925502551255225532554255525562557255825592560256125622563256425652566256725682569257025712572257325742575257625772578257925802581258225832584258525862587258825892590259125922593259425952596259725982599260026012602260326042605260626072608260926102611261226132614261526162617261826192620262126222623262426252626262726282629263026312632263326342635263626372638263926402641264226432644264526462647264826492650265126522653265426552656265726582659266026612662266326642665266626672668266926702671267226732674267526762677267826792680268126822683268426852686268726882689269026912692269326942695269626972698269927002701270227032704270527062707270827092710271127122713271427152716271727182719272027212722272327242725272627272728272927302731273227332734273527362737273827392740274127422743274427452746274727482749275027512752275327542755275627572758275927602761276227632764276527662767276827692770277127722773277427752776277727782779278027812782278327842785278627872788278927902791279227932794279527962797279827992800280128022803280428052806280728082809281028112812281328142815281628172818281928202821282228232824282528262827282828292830283128322833283428352836283728382839284028412842284328442845284628472848284928502851285228532854285528562857285828592860286128622863286428652866286728682869287028712872287328742875287628772878287928802881288228832884288528862887288828892890289128922893289428952896289728982899290029012902290329042905290629072908290929102911291229132914291529162917291829192920292129222923292429252926292729282929293029312932293329342935293629372938293929402941294229432944294529462947294829492950295129522953295429552956295729582959296029612962296329642965296629672968296929702971297229732974297529762977297829792980298129822983298429852986298729882989299029912992299329942995299629972998299930003001300230033004300530063007300830093010301130123013301430153016301730183019302030213022302330243025302630273028302930303031303230333034303530363037303830393040304130423043304430453046304730483049305030513052305330543055305630573058305930603061306230633064306530663067306830693070307130723073307430753076307730783079308030813082308330843085308630873088308930903091309230933094309530963097309830993100310131023103310431053106310731083109311031113112311331143115311631173118311931203121312231233124312531263127312831293130313131323133313431353136313731383139314031413142314331443145314631473148314931503151315231533154315531563157315831593160316131623163316431653166316731683169317031713172317331743175317631773178317931803181318231833184318531863187318831893190319131923193319431953196319731983199320032013202320332043205320632073208320932103211321232133214321532163217321832193220322132223223322432253226322732283229323032313232323332343235323632373238323932403241324232433244324532463247324832493250325132523253325432553256325732583259326032613262326332643265326632673268326932703271327232733274327532763277327832793280328132823283328432853286328732883289329032913292329332943295329632973298329933003301330233033304330533063307330833093310331133123313331433153316331733183319332033213322332333243325332633273328332933303331333233333334333533363337333833393340334133423343334433453346334733483349335033513352335333543355335633573358335933603361336233633364336533663367336833693370337133723373337433753376337733783379338033813382338333843385338633873388338933903391339233933394339533963397339833993400340134023403340434053406340734083409341034113412341334143415341634173418341934203421342234233424342534263427342834293430343134323433343434353436343734383439344034413442344334443445344634473448344934503451345234533454345534563457345834593460346134623463346434653466use async_trait::async_trait;use chrono::{DateTime, Utc};use sqlx::PgPool;use tranquil_types::{AtIdentifier, Did, Handle, Jti, PasswordHash, TokenId};use uuid::Uuid;
use super::col;use super::{column, legacy_column, opt_column};use tranquil_db_traits::{ AccountSearchResult, AccountType, ChannelVerificationStatus, CommsChannel, DbError, DidWebOverrides, NotificationPrefs, OAuthTokenWithUser, PasswordResetResult, SsoProviderType, StoredBackupCode, StoredPasskey, TotpRecord, TotpRecordState, User2faStatus, UserAuthInfo, UserCommsPrefs, UserConfirmSignup, UserDidWebInfo, UserEmailInfo, UserForDeletion, UserForDidDoc, UserForDidDocBuild, UserForPasskeyRecovery, UserForPasskeySetup, UserForRecovery, UserForVerification, UserIdAndHandle, UserIdAndPasswordHash, UserIdHandleEmail, UserInfoForAuth, UserKeyInfo, UserKeyWithId, UserLegacyLoginPref, UserLoginCheck, UserLoginFull, UserLoginInfo, UserPasswordInfo, UserRepository, UserResendVerification, UserResetCodeInfo, UserRow, UserSessionInfo, UserStatus, UserVerificationInfo, UserWithKey, WebauthnChallengeType,};
pub struct PostgresUserRepository { pool: PgPool,}
impl PostgresUserRepository { pub fn new(pool: PgPool) -> Self { Self { pool } }}
pub(crate) fn map_sqlx_error(e: sqlx::Error) -> DbError { match e { sqlx::Error::RowNotFound => DbError::NotFound, sqlx::Error::Database(db_err) => { let msg = db_err.message().to_string(); if db_err.is_unique_violation() || db_err.is_foreign_key_violation() { DbError::Constraint(msg) } else { DbError::Query(msg) } } sqlx::Error::PoolTimedOut => DbError::Connection("Pool timed out".into()), _ => DbError::Other(e.to_string()), }}
async fn consume_invite_code( conn: &mut sqlx::PgConnection, code: &str, user_id: Uuid,) -> Result<(), tranquil_db_traits::CreateAccountError> { let map_err = |e: sqlx::Error| tranquil_db_traits::CreateAccountError::Database(e.to_string());
let decremented = sqlx::query!( "UPDATE invite_codes SET available_uses = available_uses - 1 WHERE code = $1 AND available_uses > 0 AND COALESCE(disabled, false) = false", code ) .execute(&mut *conn) .await .map_err(map_err)? .rows_affected();
if decremented == 0 { return Err(tranquil_db_traits::CreateAccountError::InviteCodeUnavailable); }
sqlx::query!( "INSERT INTO invite_code_uses (code, used_by_user) VALUES ($1, $2)", code, user_id ) .execute(&mut *conn) .await .map_err(map_err)?;
Ok(())}
#[async_trait]impl UserRepository for PostgresUserRepository { async fn get_by_did(&self, did: &Did) -> Result<Option<UserRow>, DbError> { let row = sqlx::query!( r#"SELECT id, did, handle, email, created_at, deactivated_at, takedown_ref, is_admin, inbound_migration FROM users WHERE did = $1"#, did.as_str() ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?;
row.map(|r| { Ok(UserRow { id: r.id, did: column(r.did, col::USERS_DID)?, handle: column(r.handle, col::USERS_HANDLE)?, email: r.email, created_at: r.created_at, deactivated_at: r.deactivated_at, takedown_ref: r.takedown_ref, is_admin: r.is_admin, inbound_migration: r.inbound_migration, }) }) .transpose() }
async fn get_by_handle(&self, handle: &Handle) -> Result<Option<UserRow>, DbError> { let row = sqlx::query!( r#"SELECT id, did, handle, email, created_at, deactivated_at, takedown_ref, is_admin, inbound_migration FROM users WHERE handle = $1"#, handle.as_str() ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?;
row.map(|r| { Ok(UserRow { id: r.id, did: column(r.did, col::USERS_DID)?, handle: column(r.handle, col::USERS_HANDLE)?, email: r.email, created_at: r.created_at, deactivated_at: r.deactivated_at, takedown_ref: r.takedown_ref, is_admin: r.is_admin, inbound_migration: r.inbound_migration, }) }) .transpose() }
async fn get_with_key_by_did(&self, did: &Did) -> Result<Option<UserWithKey>, DbError> { let row = sqlx::query!( r#"SELECT u.id, u.did, u.handle, u.email, u.deactivated_at, u.takedown_ref, u.is_admin, k.key_bytes, k.encryption_version FROM users u JOIN user_keys k ON u.id = k.user_id WHERE u.did = $1"#, did.as_str() ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?;
row.map(|r| { Ok(UserWithKey { id: r.id, did: column(r.did, col::USERS_DID)?, handle: column(r.handle, col::USERS_HANDLE)?, email: r.email, deactivated_at: r.deactivated_at, takedown_ref: r.takedown_ref, is_admin: r.is_admin, key_bytes: r.key_bytes, encryption_version: r.encryption_version, }) }) .transpose() }
async fn get_status_by_did(&self, did: &Did) -> Result<Option<UserStatus>, DbError> { let row = sqlx::query!( "SELECT deactivated_at, takedown_ref, is_admin FROM users WHERE did = $1", did.as_str() ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(row.map(|r| UserStatus { deactivated_at: r.deactivated_at, takedown_ref: r.takedown_ref, is_admin: r.is_admin, })) }
async fn count_users(&self) -> Result<i64, DbError> { let row = sqlx::query_scalar!("SELECT COUNT(*) FROM users") .fetch_one(&self.pool) .await .map_err(map_sqlx_error)?; Ok(row.unwrap_or(0)) }
async fn get_session_access_expiry( &self, did: &Did, access_jti: &Jti, ) -> Result<Option<DateTime<Utc>>, DbError> { let row = sqlx::query!( "SELECT access_expires_at FROM session_tokens WHERE did = $1 AND access_jti = $2", did.as_str(), access_jti.as_str() ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(row.map(|r| r.access_expires_at)) }
async fn get_oauth_token_with_user( &self, token_id: &TokenId, ) -> Result<Option<OAuthTokenWithUser>, DbError> { let row = sqlx::query!( r#"SELECT t.did, t.expires_at, u.deactivated_at, u.takedown_ref, u.is_admin, k.key_bytes as "key_bytes?", k.encryption_version as "encryption_version?" FROM oauth_token t JOIN users u ON t.did = u.did LEFT JOIN user_keys k ON u.id = k.user_id WHERE t.token_id = $1"#, token_id.as_str() ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?;
row.map(|r| { Ok(OAuthTokenWithUser { did: column(r.did, col::OAUTH_TOKEN_DID)?, expires_at: r.expires_at, deactivated_at: r.deactivated_at, takedown_ref: r.takedown_ref, is_admin: r.is_admin, key_bytes: r.key_bytes, encryption_version: r.encryption_version, }) }) .transpose() }
async fn get_user_info_by_did(&self, did: &Did) -> Result<Option<UserInfoForAuth>, DbError> { let row = sqlx::query!( r#"SELECT u.deactivated_at, u.takedown_ref, u.is_admin, k.key_bytes as "key_bytes?", k.encryption_version as "encryption_version?" FROM users u LEFT JOIN user_keys k ON u.id = k.user_id WHERE u.did = $1"#, did.as_str() ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(row.map(|r| UserInfoForAuth { deactivated_at: r.deactivated_at, takedown_ref: r.takedown_ref, is_admin: r.is_admin, key_bytes: r.key_bytes, encryption_version: r.encryption_version, })) }
async fn get_any_admin_user_id(&self) -> Result<Option<Uuid>, DbError> { let row = sqlx::query_scalar!("SELECT id FROM users WHERE is_admin = true LIMIT 1") .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?; Ok(row) }
async fn set_invites_disabled(&self, did: &Did, disabled: bool) -> Result<bool, DbError> { let result = sqlx::query!( "UPDATE users SET invites_disabled = $2 WHERE did = $1", did.as_str(), disabled ) .execute(&self.pool) .await .map_err(map_sqlx_error)?; Ok(result.rows_affected() > 0) }
async fn search_accounts( &self, cursor_did: Option<&Did>, email_filter: Option<&str>, handle_filter: Option<&str>, limit: i64, ) -> Result<Vec<AccountSearchResult>, DbError> { let cursor_str = cursor_did.map(|d| d.as_str()); let email_like = email_filter.map(|e| format!("%{e}%")); let handle_like = handle_filter.map(|h| format!("%{h}%")); let rows = sqlx::query!( r#"SELECT did, handle, email, created_at, email_verified, deactivated_at, invites_disabled FROM users WHERE ($1::text IS NULL OR did > $1) AND ($2::text IS NULL OR email ILIKE $2) AND ($3::text IS NULL OR handle ILIKE $3) ORDER BY did ASC LIMIT $4"#, cursor_str, email_like.as_deref(), handle_like.as_deref(), limit ) .fetch_all(&self.pool) .await .map_err(map_sqlx_error)?; Ok(rows .into_iter() .filter_map(|r| { Some(AccountSearchResult { did: legacy_column(r.did, col::USERS_DID)?, handle: legacy_column(r.handle, col::USERS_HANDLE)?, email: r.email, created_at: r.created_at, email_verified: r.email_verified, deactivated_at: r.deactivated_at, invites_disabled: r.invites_disabled, }) }) .collect()) }
async fn get_auth_info_by_did(&self, did: &Did) -> Result<Option<UserAuthInfo>, DbError> { let row = sqlx::query!( r#"SELECT id, did, password_hash, deactivated_at, takedown_ref, email_verified, discord_verified, telegram_verified, signal_verified FROM users WHERE did = $1"#, did.as_str() ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?; row.map(|r| { Ok(UserAuthInfo { id: r.id, did: column(r.did, col::USERS_DID)?, password_hash: r.password_hash.map(PasswordHash::new), deactivated_at: r.deactivated_at, takedown_ref: r.takedown_ref, channel_verification: ChannelVerificationStatus::from_db_row( r.email_verified, r.discord_verified, r.telegram_verified, r.signal_verified, ), }) }) .transpose() }
async fn get_by_email(&self, email: &str) -> Result<Option<UserForVerification>, DbError> { let row = sqlx::query!( r#"SELECT id, did, email, email_verified FROM users WHERE LOWER(email) = $1"#, email ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?; row.map(|r| { Ok(UserForVerification { id: r.id, did: column(r.did, col::USERS_DID)?, email: r.email, email_verified: r.email_verified, }) }) .transpose() }
async fn get_comms_prefs(&self, user_id: Uuid) -> Result<Option<UserCommsPrefs>, DbError> { let row = sqlx::query!( r#"SELECT email, handle, preferred_comms_channel as "preferred_channel!: CommsChannel", preferred_locale, telegram_chat_id, discord_id, signal_username FROM users WHERE id = $1"#, user_id ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?; row.map(|r| { Ok(UserCommsPrefs { email: r.email, handle: column(r.handle, col::USERS_HANDLE)?, preferred_channel: r.preferred_channel, preferred_locale: r.preferred_locale, telegram_chat_id: r.telegram_chat_id, discord_id: r.discord_id, signal_username: r.signal_username, }) }) .transpose() }
async fn get_id_by_did(&self, did: &Did) -> Result<Option<Uuid>, DbError> { let id = sqlx::query_scalar!("SELECT id FROM users WHERE did = $1", did.as_str()) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?; Ok(id) }
async fn get_user_key_by_id(&self, user_id: Uuid) -> Result<Option<UserKeyInfo>, DbError> { let row = sqlx::query!( "SELECT key_bytes, encryption_version FROM user_keys WHERE user_id = $1", user_id ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?; Ok(row.map(|r| UserKeyInfo { key_bytes: r.key_bytes, encryption_version: r.encryption_version, })) }
async fn get_id_and_handle_by_did( &self, did: &Did, ) -> Result<Option<UserIdAndHandle>, DbError> { let row = sqlx::query!("SELECT id, handle FROM users WHERE did = $1", did.as_str()) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?; row.map(|r| { Ok(UserIdAndHandle { id: r.id, handle: column(r.handle, col::USERS_HANDLE)?, }) }) .transpose() }
async fn get_did_web_info_by_handle( &self, handle: &Handle, ) -> Result<Option<UserDidWebInfo>, DbError> { let row = sqlx::query!( "SELECT id, did, migrated_to_pds FROM users WHERE handle = $1", handle.as_str() ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?; row.map(|r| { Ok(UserDidWebInfo { id: r.id, did: column(r.did, col::USERS_DID)?, migrated_to_pds: r.migrated_to_pds, }) }) .transpose() }
async fn get_did_web_overrides( &self, user_id: Uuid, ) -> Result<Option<DidWebOverrides>, DbError> { let row = sqlx::query!( "SELECT verification_methods, also_known_as FROM did_web_overrides WHERE user_id = $1", user_id ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?; Ok(row.map(|r| DidWebOverrides { verification_methods: r.verification_methods, also_known_as: r.also_known_as, })) }
async fn get_handle_by_did(&self, did: &Did) -> Result<Option<Handle>, DbError> { let handle = sqlx::query_scalar!("SELECT handle FROM users WHERE did = $1", did.as_str()) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?; opt_column(handle, col::USERS_HANDLE) }
async fn check_handle_exists( &self, handle: &Handle, exclude_user_id: Uuid, ) -> Result<bool, DbError> { let exists = sqlx::query_scalar!( "SELECT EXISTS(SELECT 1 FROM users WHERE handle = $1 AND id != $2) as \"exists!\"", handle.as_str(), exclude_user_id ) .fetch_one(&self.pool) .await .map_err(map_sqlx_error)?; Ok(exists) }
async fn update_handle(&self, user_id: Uuid, handle: &Handle) -> Result<(), DbError> { sqlx::query!( "UPDATE users SET handle = $1 WHERE id = $2", handle.as_str(), user_id ) .execute(&self.pool) .await .map_err(map_sqlx_error)?; Ok(()) }
async fn get_user_with_key_by_did(&self, did: &Did) -> Result<Option<UserKeyWithId>, DbError> { let row = sqlx::query!( r#"SELECT u.id, uk.key_bytes, uk.encryption_version FROM users u JOIN user_keys uk ON u.id = uk.user_id WHERE u.did = $1"#, did.as_str() ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?; Ok(row.map(|r| UserKeyWithId { id: r.id, key_bytes: r.key_bytes, encryption_version: r.encryption_version, })) }
async fn is_account_migrated(&self, did: &Did) -> Result<bool, DbError> { let row = sqlx::query!( r#"SELECT (migrated_to_pds IS NOT NULL AND deactivated_at IS NOT NULL) as "migrated!: bool" FROM users WHERE did = $1"#, did.as_str() ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?; Ok(row.map(|r| r.migrated).unwrap_or(false)) }
async fn has_verified_comms_channel(&self, did: &Did) -> Result<bool, DbError> { let row = sqlx::query!( r#"SELECT email_verified, discord_verified, telegram_verified, signal_verified FROM users WHERE did = $1"#, did.as_str() ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?; Ok(row .map(|r| { r.email_verified || r.discord_verified || r.telegram_verified || r.signal_verified }) .unwrap_or(false)) }
async fn get_id_by_handle(&self, handle: &Handle) -> Result<Option<Uuid>, DbError> { let id = sqlx::query_scalar!("SELECT id FROM users WHERE handle = $1", handle.as_str()) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?; Ok(id) }
async fn get_email_info_by_did(&self, did: &Did) -> Result<Option<UserEmailInfo>, DbError> { let row = sqlx::query!( "SELECT id, handle, email, email_verified FROM users WHERE did = $1", did.as_str() ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?; row.map(|r| { Ok(UserEmailInfo { id: r.id, handle: column(r.handle, col::USERS_HANDLE)?, email: r.email, email_verified: r.email_verified, }) }) .transpose() }
async fn check_email_exists( &self, email: &str, exclude_user_id: Uuid, ) -> Result<bool, DbError> { let row = sqlx::query!( "SELECT 1 as one FROM users WHERE LOWER(email) = $1 AND id != $2", email.to_lowercase(), exclude_user_id ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?; Ok(row.is_some()) }
async fn update_email(&self, user_id: Uuid, email: &str) -> Result<(), DbError> { sqlx::query!( "UPDATE users SET email = $1, email_verified = FALSE, updated_at = NOW() WHERE id = $2", email, user_id ) .execute(&self.pool) .await .map_err(map_sqlx_error)?; Ok(()) }
async fn set_email_verified(&self, user_id: Uuid, verified: bool) -> Result<(), DbError> { sqlx::query!( "UPDATE users SET email_verified = $1, updated_at = NOW() WHERE id = $2", verified, user_id ) .execute(&self.pool) .await .map_err(map_sqlx_error)?; Ok(()) }
async fn check_email_verified_by_identifier( &self, identifier: &AtIdentifier, ) -> Result<Option<bool>, DbError> { let row = sqlx::query_scalar!( "SELECT email_verified FROM users WHERE did = $1 OR email = $1 OR handle = $1", identifier.as_str() ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?; Ok(row) }
async fn check_channel_verified_by_did( &self, did: &Did, channel: CommsChannel, ) -> Result<Option<bool>, DbError> { let row = sqlx::query!( r#"SELECT email_verified, discord_verified, telegram_verified, signal_verified FROM users WHERE did = $1"#, did.as_str() ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(row.map(|r| match channel { CommsChannel::Email => r.email_verified, CommsChannel::Discord => r.discord_verified, CommsChannel::Telegram => r.telegram_verified, CommsChannel::Signal => r.signal_verified, })) }
async fn admin_update_email(&self, did: &Did, email: &str) -> Result<u64, DbError> { let result = sqlx::query!( "UPDATE users SET email = $1 WHERE did = $2", email, did.as_str() ) .execute(&self.pool) .await .map_err(map_sqlx_error)?; Ok(result.rows_affected()) }
async fn admin_update_handle(&self, did: &Did, handle: &Handle) -> Result<u64, DbError> { let result = sqlx::query!( "UPDATE users SET handle = $1 WHERE did = $2", handle.as_str(), did.as_str() ) .execute(&self.pool) .await .map_err(map_sqlx_error)?; Ok(result.rows_affected()) }
async fn admin_update_password( &self, did: &Did, password_hash: &PasswordHash, ) -> Result<u64, DbError> { let result = sqlx::query!( "UPDATE users SET password_hash = $1 WHERE did = $2", password_hash.as_str(), did.as_str() ) .execute(&self.pool) .await .map_err(map_sqlx_error)?; Ok(result.rows_affected()) }
async fn set_admin_status(&self, did: &Did, is_admin: bool) -> Result<(), DbError> { sqlx::query!( "UPDATE users SET is_admin = $1 WHERE did = $2", is_admin, did.as_str() ) .execute(&self.pool) .await .map_err(map_sqlx_error)?; Ok(()) }
async fn get_notification_prefs( &self, did: &Did, ) -> Result<Option<NotificationPrefs>, DbError> { let row = sqlx::query!( r#"SELECT email, preferred_comms_channel as "preferred_channel!: CommsChannel", discord_id, discord_username, discord_verified, telegram_username, telegram_verified, telegram_chat_id, signal_username, signal_verified FROM users WHERE did = $1"#, did.as_str() ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?; Ok(row.map(|r| NotificationPrefs { email: r.email.unwrap_or_default(), preferred_channel: r.preferred_channel, discord_id: r.discord_id, discord_username: r.discord_username, discord_verified: r.discord_verified, telegram_username: r.telegram_username, telegram_verified: r.telegram_verified, telegram_chat_id: r.telegram_chat_id, signal_username: r.signal_username, signal_verified: r.signal_verified, })) }
async fn get_id_handle_email_by_did( &self, did: &Did, ) -> Result<Option<UserIdHandleEmail>, DbError> { let row = sqlx::query!( "SELECT id, handle, email FROM users WHERE did = $1", did.as_str() ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?; row.map(|r| { Ok(UserIdHandleEmail { id: r.id, handle: column(r.handle, col::USERS_HANDLE)?, email: r.email, }) }) .transpose() }
async fn update_preferred_comms_channel( &self, did: &Did, channel: CommsChannel, ) -> Result<(), DbError> { sqlx::query!( "UPDATE users SET preferred_comms_channel = $1, updated_at = NOW() WHERE did = $2", channel as CommsChannel, did.as_str() ) .execute(&self.pool) .await .map_err(map_sqlx_error)?; Ok(()) }
async fn clear_discord(&self, user_id: Uuid) -> Result<(), DbError> { sqlx::query!( "UPDATE users SET discord_id = NULL, discord_username = NULL, discord_verified = FALSE, updated_at = NOW() WHERE id = $1", user_id ) .execute(&self.pool) .await .map_err(map_sqlx_error)?; Ok(()) }
async fn clear_telegram(&self, user_id: Uuid) -> Result<(), DbError> { sqlx::query!( "UPDATE users SET telegram_username = NULL, telegram_verified = FALSE, telegram_chat_id = NULL, updated_at = NOW() WHERE id = $1", user_id ) .execute(&self.pool) .await .map_err(map_sqlx_error)?; Ok(()) }
async fn clear_signal(&self, user_id: Uuid) -> Result<(), DbError> { sqlx::query!( "UPDATE users SET signal_username = NULL, signal_verified = FALSE, updated_at = NOW() WHERE id = $1", user_id ) .execute(&self.pool) .await .map_err(map_sqlx_error)?; Ok(()) }
async fn get_verification_info( &self, did: &Did, ) -> Result<Option<UserVerificationInfo>, DbError> { let row = sqlx::query!( r#"SELECT id, handle, email, email_verified, discord_verified, telegram_verified, signal_verified FROM users WHERE did = $1"#, did.as_str() ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?; row.map(|r| { Ok(UserVerificationInfo { id: r.id, handle: column(r.handle, col::USERS_HANDLE)?, email: r.email, channel_verification: ChannelVerificationStatus::from_db_row( r.email_verified, r.discord_verified, r.telegram_verified, r.signal_verified, ), }) }) .transpose() }
async fn verify_email_channel(&self, user_id: Uuid, email: &str) -> Result<bool, DbError> { sqlx::query!( "UPDATE users SET email = $1, email_verified = TRUE, updated_at = NOW() WHERE id = $2", email, user_id ) .execute(&self.pool) .await .map_err(map_sqlx_error)?; Ok(true) }
async fn verify_discord_channel(&self, user_id: Uuid, discord_id: &str) -> Result<(), DbError> { sqlx::query!( "UPDATE users SET discord_id = $1, discord_verified = TRUE, updated_at = NOW() WHERE id = $2", discord_id, user_id ) .execute(&self.pool) .await .map_err(map_sqlx_error)?; Ok(()) }
async fn verify_telegram_channel( &self, user_id: Uuid, telegram_username: &str, ) -> Result<(), DbError> { sqlx::query!( "UPDATE users SET telegram_username = $1, telegram_verified = TRUE, updated_at = NOW() WHERE id = $2", telegram_username, user_id ) .execute(&self.pool) .await .map_err(map_sqlx_error)?; Ok(()) }
async fn verify_signal_channel( &self, user_id: Uuid, signal_username: &str, ) -> Result<(), DbError> { sqlx::query!( "UPDATE users SET signal_username = $1, signal_verified = TRUE, updated_at = NOW() WHERE id = $2", signal_username, user_id ) .execute(&self.pool) .await .map_err(map_sqlx_error)?; Ok(()) }
async fn set_email_verified_flag(&self, user_id: Uuid) -> Result<(), DbError> { sqlx::query!( "UPDATE users SET email_verified = TRUE WHERE id = $1", user_id ) .execute(&self.pool) .await .map_err(map_sqlx_error)?; Ok(()) }
async fn set_discord_verified_flag(&self, user_id: Uuid) -> Result<(), DbError> { sqlx::query!( "UPDATE users SET discord_verified = TRUE WHERE id = $1", user_id ) .execute(&self.pool) .await .map_err(map_sqlx_error)?; Ok(()) }
async fn set_telegram_verified_flag(&self, user_id: Uuid) -> Result<(), DbError> { sqlx::query!( "UPDATE users SET telegram_verified = TRUE WHERE id = $1", user_id ) .execute(&self.pool) .await .map_err(map_sqlx_error)?; Ok(()) }
async fn set_signal_verified_flag(&self, user_id: Uuid) -> Result<(), DbError> { sqlx::query!( "UPDATE users SET signal_verified = TRUE WHERE id = $1", user_id ) .execute(&self.pool) .await .map_err(map_sqlx_error)?; Ok(()) }
async fn has_totp_enabled(&self, did: &Did) -> Result<bool, DbError> { let row = sqlx::query_scalar!( "SELECT verified FROM user_totp WHERE did = $1", did.as_str() ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(matches!(row, Some(true))) }
async fn has_passkeys(&self, did: &Did) -> Result<bool, DbError> { let count = sqlx::query_scalar!( "SELECT COUNT(*) as count FROM passkeys WHERE did = $1", did.as_str() ) .fetch_one(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(count.unwrap_or(0) > 0) }
async fn get_password_hash_by_did(&self, did: &Did) -> Result<Option<PasswordHash>, DbError> { let row = sqlx::query_scalar!( "SELECT password_hash FROM users WHERE did = $1", did.as_str() ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(row.flatten().map(PasswordHash::new)) }
async fn get_passkeys_for_user(&self, did: &Did) -> Result<Vec<StoredPasskey>, DbError> { let rows = sqlx::query!( r#"SELECT id, did, credential_id, public_key, sign_count, created_at, last_used, friendly_name, aaguid, transports FROM passkeys WHERE did = $1 ORDER BY created_at DESC"#, did.as_str() ) .fetch_all(&self.pool) .await .map_err(map_sqlx_error)?;
rows.into_iter() .map(|r| { Ok(StoredPasskey { id: r.id, did: column(r.did, col::PASSKEYS_DID)?, credential_id: r.credential_id, public_key: r.public_key, sign_count: r.sign_count, created_at: r.created_at, last_used: r.last_used, friendly_name: r.friendly_name, aaguid: r.aaguid, transports: r.transports, }) }) .collect() }
async fn get_passkey_by_credential_id( &self, credential_id: &[u8], ) -> Result<Option<StoredPasskey>, DbError> { let row = sqlx::query!( r#"SELECT id, did, credential_id, public_key, sign_count, created_at, last_used, friendly_name, aaguid, transports FROM passkeys WHERE credential_id = $1"#, credential_id ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?;
row.map(|r| { Ok(StoredPasskey { id: r.id, did: column(r.did, col::PASSKEYS_DID)?, credential_id: r.credential_id, public_key: r.public_key, sign_count: r.sign_count, created_at: r.created_at, last_used: r.last_used, friendly_name: r.friendly_name, aaguid: r.aaguid, transports: r.transports, }) }) .transpose() }
async fn save_passkey( &self, did: &Did, credential_id: &[u8], public_key: &[u8], friendly_name: Option<&str>, ) -> Result<Uuid, DbError> { let id = Uuid::new_v4(); let aaguid: Option<Vec<u8>> = None; sqlx::query!( r#"INSERT INTO passkeys (id, did, credential_id, public_key, sign_count, friendly_name, aaguid) VALUES ($1, $2, $3, $4, 0, $5, $6)"#, id, did.as_str(), credential_id, public_key, friendly_name, aaguid, ) .execute(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(id) }
async fn update_passkey_counter( &self, credential_id: &[u8], new_counter: i32, ) -> Result<bool, DbError> { let stored = self.get_passkey_by_credential_id(credential_id).await?; let Some(stored) = stored else { return Err(DbError::NotFound); };
if new_counter > 0 && new_counter <= stored.sign_count { return Ok(false); }
sqlx::query!( "UPDATE passkeys SET sign_count = $1, last_used = NOW() WHERE credential_id = $2", new_counter, credential_id, ) .execute(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(true) }
async fn delete_passkey(&self, id: Uuid, did: &Did) -> Result<bool, DbError> { let result = sqlx::query!( "DELETE FROM passkeys WHERE id = $1 AND did = $2", id, did.as_str() ) .execute(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(result.rows_affected() > 0) }
async fn update_passkey_name(&self, id: Uuid, did: &Did, name: &str) -> Result<bool, DbError> { let result = sqlx::query!( "UPDATE passkeys SET friendly_name = $1 WHERE id = $2 AND did = $3", name, id, did.as_str() ) .execute(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(result.rows_affected() > 0) }
async fn save_webauthn_challenge( &self, did: &Did, challenge_type: WebauthnChallengeType, state_json: &str, ) -> Result<Uuid, DbError> { let id = Uuid::new_v4(); let challenge = id.as_bytes().to_vec(); let expires_at = chrono::Utc::now() + chrono::Duration::minutes(5); sqlx::query!( r#"INSERT INTO webauthn_challenges (id, did, challenge, challenge_type, state_json, expires_at) VALUES ($1, $2, $3, $4, $5, $6)"#, id, did.as_str(), challenge, challenge_type.as_str(), state_json, expires_at, ) .execute(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(id) }
async fn load_webauthn_challenge( &self, did: &Did, challenge_type: WebauthnChallengeType, ) -> Result<Option<String>, DbError> { let row = sqlx::query_scalar!( r#"SELECT state_json FROM webauthn_challenges WHERE did = $1 AND challenge_type = $2 AND expires_at > NOW() ORDER BY created_at DESC LIMIT 1"#, did.as_str(), challenge_type.as_str() ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(row) }
async fn delete_webauthn_challenge( &self, did: &Did, challenge_type: WebauthnChallengeType, ) -> Result<(), DbError> { sqlx::query!( "DELETE FROM webauthn_challenges WHERE did = $1 AND challenge_type = $2", did.as_str(), challenge_type.as_str() ) .execute(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(()) }
async fn save_discoverable_challenge( &self, request_key: &str, state_json: &str, ) -> Result<Uuid, DbError> { let id = Uuid::new_v4(); let challenge = id.as_bytes().to_vec(); let expires_at = chrono::Utc::now() + chrono::Duration::minutes(5); sqlx::query!( r#"INSERT INTO webauthn_challenges (id, did, challenge, challenge_type, state_json, expires_at) VALUES ($1, $2, $3, 'discoverable', $4, $5)"#, id, request_key, challenge, state_json, expires_at, ) .execute(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(id) }
async fn load_discoverable_challenge( &self, request_key: &str, ) -> Result<Option<String>, DbError> { let row = sqlx::query_scalar!( r#"SELECT state_json FROM webauthn_challenges WHERE did = $1 AND challenge_type = 'discoverable' AND expires_at > NOW() ORDER BY created_at DESC LIMIT 1"#, request_key, ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(row) }
async fn delete_discoverable_challenge(&self, request_key: &str) -> Result<(), DbError> { sqlx::query!( "DELETE FROM webauthn_challenges WHERE did = $1 AND challenge_type = 'discoverable'", request_key, ) .execute(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(()) }
async fn get_totp_record(&self, did: &Did) -> Result<Option<TotpRecord>, DbError> { let row = sqlx::query!( "SELECT secret_encrypted, encryption_version, verified FROM user_totp WHERE did = $1", did.as_str() ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(row.map(|r| TotpRecord { secret_encrypted: r.secret_encrypted, encryption_version: r.encryption_version, verified: r.verified, })) }
async fn get_totp_record_state(&self, did: &Did) -> Result<Option<TotpRecordState>, DbError> { self.get_totp_record(did) .await .map(|opt| opt.map(TotpRecordState::from)) }
async fn upsert_totp_secret( &self, did: &Did, secret_encrypted: &[u8], encryption_version: i32, ) -> Result<(), DbError> { sqlx::query!( r#"INSERT INTO user_totp (did, secret_encrypted, encryption_version, verified, created_at) VALUES ($1, $2, $3, false, NOW()) ON CONFLICT (did) DO UPDATE SET secret_encrypted = $2, encryption_version = $3, verified = false, created_at = NOW(), last_used = NULL"#, did.as_str(), secret_encrypted, encryption_version ) .execute(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(()) }
async fn set_totp_verified(&self, did: &Did) -> Result<(), DbError> { sqlx::query!( "UPDATE user_totp SET verified = true, last_used = NOW() WHERE did = $1", did.as_str() ) .execute(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(()) }
async fn update_totp_last_used(&self, did: &Did) -> Result<(), DbError> { sqlx::query!( "UPDATE user_totp SET last_used = NOW() WHERE did = $1", did.as_str() ) .execute(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(()) }
async fn delete_totp(&self, did: &Did) -> Result<(), DbError> { sqlx::query!("DELETE FROM user_totp WHERE did = $1", did.as_str()) .execute(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(()) }
async fn get_unused_backup_codes(&self, did: &Did) -> Result<Vec<StoredBackupCode>, DbError> { let rows = sqlx::query!( "SELECT id, code_hash FROM backup_codes WHERE did = $1 AND used_at IS NULL", did.as_str() ) .fetch_all(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(rows .into_iter() .map(|r| StoredBackupCode { id: r.id, code_hash: r.code_hash, }) .collect()) }
async fn mark_backup_code_used(&self, code_id: Uuid) -> Result<bool, DbError> { let result = sqlx::query!( "UPDATE backup_codes SET used_at = NOW() WHERE id = $1", code_id ) .execute(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(result.rows_affected() > 0) }
async fn count_unused_backup_codes(&self, did: &Did) -> Result<i64, DbError> { let row = sqlx::query!( "SELECT COUNT(*) as count FROM backup_codes WHERE did = $1 AND used_at IS NULL", did.as_str() ) .fetch_one(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(row.count.unwrap_or(0)) }
async fn delete_backup_codes(&self, did: &Did) -> Result<u64, DbError> { let result = sqlx::query!("DELETE FROM backup_codes WHERE did = $1", did.as_str()) .execute(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(result.rows_affected()) }
async fn insert_backup_codes(&self, did: &Did, code_hashes: &[String]) -> Result<(), DbError> { sqlx::query!( r#" INSERT INTO backup_codes (did, code_hash, created_at) SELECT $1, hash, NOW() FROM UNNEST($2::text[]) AS t(hash) "#, did.as_str(), code_hashes ) .execute(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(()) }
async fn enable_totp_with_backup_codes( &self, did: &Did, code_hashes: &[String], ) -> Result<(), DbError> { let mut tx = self.pool.begin().await.map_err(map_sqlx_error)?;
sqlx::query!( "UPDATE user_totp SET verified = true, last_used = NOW() WHERE did = $1", did.as_str() ) .execute(&mut *tx) .await .map_err(map_sqlx_error)?;
sqlx::query!("DELETE FROM backup_codes WHERE did = $1", did.as_str()) .execute(&mut *tx) .await .map_err(map_sqlx_error)?;
sqlx::query!( r#" INSERT INTO backup_codes (did, code_hash, created_at) SELECT $1, hash, NOW() FROM UNNEST($2::text[]) AS t(hash) "#, did.as_str(), code_hashes ) .execute(&mut *tx) .await .map_err(map_sqlx_error)?;
tx.commit().await.map_err(map_sqlx_error)?;
Ok(()) }
async fn delete_totp_and_backup_codes(&self, did: &Did) -> Result<(), DbError> { let mut tx = self.pool.begin().await.map_err(map_sqlx_error)?;
sqlx::query!("DELETE FROM user_totp WHERE did = $1", did.as_str()) .execute(&mut *tx) .await .map_err(map_sqlx_error)?;
sqlx::query!("DELETE FROM backup_codes WHERE did = $1", did.as_str()) .execute(&mut *tx) .await .map_err(map_sqlx_error)?;
tx.commit().await.map_err(map_sqlx_error)?;
Ok(()) }
async fn replace_backup_codes(&self, did: &Did, code_hashes: &[String]) -> Result<(), DbError> { let mut tx = self.pool.begin().await.map_err(map_sqlx_error)?;
sqlx::query!("DELETE FROM backup_codes WHERE did = $1", did.as_str()) .execute(&mut *tx) .await .map_err(map_sqlx_error)?;
sqlx::query!( r#" INSERT INTO backup_codes (did, code_hash, created_at) SELECT $1, hash, NOW() FROM UNNEST($2::text[]) AS t(hash) "#, did.as_str(), code_hashes ) .execute(&mut *tx) .await .map_err(map_sqlx_error)?;
tx.commit().await.map_err(map_sqlx_error)?;
Ok(()) }
async fn get_login_check_by_identifier( &self, identifier: &AtIdentifier, ) -> Result<Option<UserLoginCheck>, DbError> { sqlx::query!( "SELECT did, password_hash FROM users WHERE handle = $1 OR did = $1", identifier.as_str() ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)? .map(|r| { Ok(UserLoginCheck { did: column(r.did, col::USERS_DID)?, password_hash: r.password_hash.map(PasswordHash::new), }) }) .transpose() }
async fn get_login_info_by_identifier( &self, identifier: &AtIdentifier, ) -> Result<Option<UserLoginInfo>, DbError> { sqlx::query!( r#" SELECT id, did, email, password_hash, password_required, two_factor_enabled, preferred_comms_channel as "preferred_comms_channel!: CommsChannel", deactivated_at, takedown_ref, email_verified, discord_verified, telegram_verified, signal_verified, account_type as "account_type!: AccountType" FROM users WHERE handle = $1 OR did = $1 "#, identifier.as_str() ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)? .map(|row| { Ok(UserLoginInfo { id: row.id, did: column(row.did, col::USERS_DID)?, email: row.email, password_hash: row.password_hash.map(PasswordHash::new), password_required: row.password_required, two_factor_enabled: row.two_factor_enabled, preferred_comms_channel: row.preferred_comms_channel, deactivated_at: row.deactivated_at, takedown_ref: row.takedown_ref, channel_verification: ChannelVerificationStatus::from_db_row( row.email_verified, row.discord_verified, row.telegram_verified, row.signal_verified, ), account_type: row.account_type, }) }) .transpose() }
async fn get_2fa_status_by_did(&self, did: &Did) -> Result<Option<User2faStatus>, DbError> { sqlx::query!( r#" SELECT id, two_factor_enabled, preferred_comms_channel as "preferred_comms_channel!: CommsChannel", email_verified, discord_verified, telegram_verified, signal_verified FROM users WHERE did = $1 "#, did.as_str() ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error) .map(|opt| { opt.map(|row| User2faStatus { id: row.id, two_factor_enabled: row.two_factor_enabled, preferred_comms_channel: row.preferred_comms_channel, channel_verification: ChannelVerificationStatus::from_db_row( row.email_verified, row.discord_verified, row.telegram_verified, row.signal_verified, ), }) }) }
async fn get_session_info_by_did(&self, did: &Did) -> Result<Option<UserSessionInfo>, DbError> { sqlx::query!( r#" SELECT u.handle, u.email, u.email_verified, u.is_admin, u.deactivated_at, u.takedown_ref, u.preferred_locale, u.preferred_comms_channel as "preferred_comms_channel!: CommsChannel", u.discord_verified, u.telegram_verified, u.signal_verified, u.migrated_to_pds, u.migrated_at, (SELECT verified FROM user_totp WHERE did = u.did) as totp_enabled, COALESCE((SELECT (value_json)::boolean FROM account_preferences WHERE user_id = u.id AND name = 'email_auth_factor' ORDER BY created_at DESC LIMIT 1), false) as "email_2fa_enabled!" FROM users u WHERE u.did = $1 "#, did.as_str() ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)? .map(|row| { Ok(UserSessionInfo { handle: column(row.handle, col::USERS_HANDLE)?, email: row.email, is_admin: row.is_admin, deactivated_at: row.deactivated_at, takedown_ref: row.takedown_ref, preferred_locale: row.preferred_locale, preferred_comms_channel: row.preferred_comms_channel, channel_verification: ChannelVerificationStatus::from_db_row( row.email_verified, row.discord_verified, row.telegram_verified, row.signal_verified, ), migrated_to_pds: row.migrated_to_pds, migrated_at: row.migrated_at, totp_enabled: row.totp_enabled.unwrap_or(false), email_2fa_enabled: row.email_2fa_enabled, }) }) .transpose() }
async fn get_legacy_login_pref( &self, did: &Did, ) -> Result<Option<UserLegacyLoginPref>, DbError> { sqlx::query!( r#" SELECT u.allow_legacy_login, (EXISTS(SELECT 1 FROM user_totp t WHERE t.did = u.did AND t.verified = TRUE) OR EXISTS(SELECT 1 FROM passkeys p WHERE p.did = u.did)) as "has_mfa!" FROM users u WHERE u.did = $1 "#, did.as_str() ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error) .map(|opt| { opt.map(|row| UserLegacyLoginPref { allow_legacy_login: row.allow_legacy_login, has_mfa: row.has_mfa, }) }) }
async fn update_legacy_login(&self, did: &Did, allow: bool) -> Result<bool, DbError> { let result = sqlx::query!( "UPDATE users SET allow_legacy_login = $1 WHERE did = $2 RETURNING did", allow, did.as_str() ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?; Ok(result.is_some()) }
async fn update_locale(&self, did: &Did, locale: &str) -> Result<bool, DbError> { let result = sqlx::query!( "UPDATE users SET preferred_locale = $1 WHERE did = $2 RETURNING did", locale, did.as_str() ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?; Ok(result.is_some()) }
async fn get_login_full_by_identifier( &self, identifier: &AtIdentifier, ) -> Result<Option<UserLoginFull>, DbError> { sqlx::query!( r#"SELECT u.id, u.did, u.handle, u.password_hash, u.email, u.deactivated_at, u.takedown_ref, u.email_verified, u.discord_verified, u.telegram_verified, u.signal_verified, u.allow_legacy_login, u.migrated_to_pds, u.preferred_comms_channel as "preferred_comms_channel: CommsChannel", k.key_bytes, k.encryption_version, (SELECT verified FROM user_totp WHERE did = u.did) as totp_enabled, COALESCE((SELECT (value_json)::boolean FROM account_preferences WHERE user_id = u.id AND name = 'email_auth_factor' ORDER BY created_at DESC LIMIT 1), false) as "email_2fa_enabled!" FROM users u JOIN user_keys k ON u.id = k.user_id WHERE u.handle = $1 OR u.did = $1"#, identifier.as_str() ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)? .map(|row| { Ok(UserLoginFull { id: row.id, did: column(row.did, col::USERS_DID)?, handle: column(row.handle, col::USERS_HANDLE)?, password_hash: row.password_hash.map(PasswordHash::new), email: row.email, deactivated_at: row.deactivated_at, takedown_ref: row.takedown_ref, channel_verification: ChannelVerificationStatus::from_db_row( row.email_verified, row.discord_verified, row.telegram_verified, row.signal_verified, ), allow_legacy_login: row.allow_legacy_login, migrated_to_pds: row.migrated_to_pds, preferred_comms_channel: row.preferred_comms_channel, key_bytes: row.key_bytes, encryption_version: row.encryption_version, totp_enabled: row.totp_enabled.unwrap_or(false), email_2fa_enabled: row.email_2fa_enabled, }) }) .transpose() }
async fn get_confirm_signup_by_did( &self, did: &Did, ) -> Result<Option<UserConfirmSignup>, DbError> { sqlx::query!( r#"SELECT u.id, u.did, u.handle, u.email, u.preferred_comms_channel as "channel: CommsChannel", u.discord_username, u.telegram_username, u.signal_username, k.key_bytes, k.encryption_version FROM users u JOIN user_keys k ON u.id = k.user_id WHERE u.did = $1"#, did.as_str() ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)? .map(|row| { Ok(UserConfirmSignup { id: row.id, did: column(row.did, col::USERS_DID)?, handle: column(row.handle, col::USERS_HANDLE)?, email: row.email, channel: row.channel, discord_username: row.discord_username, telegram_username: row.telegram_username, signal_username: row.signal_username, key_bytes: row.key_bytes, encryption_version: row.encryption_version, }) }) .transpose() }
async fn get_resend_verification_by_did( &self, did: &Did, ) -> Result<Option<UserResendVerification>, DbError> { sqlx::query!( r#"SELECT id, handle, email, preferred_comms_channel as "channel: CommsChannel", discord_username, telegram_username, signal_username, email_verified, discord_verified, telegram_verified, signal_verified FROM users WHERE did = $1"#, did.as_str() ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)? .map(|row| { Ok(UserResendVerification { id: row.id, handle: column(row.handle, col::USERS_HANDLE)?, email: row.email, channel: row.channel, discord_username: row.discord_username, telegram_username: row.telegram_username, signal_username: row.signal_username, channel_verification: ChannelVerificationStatus::from_db_row( row.email_verified, row.discord_verified, row.telegram_verified, row.signal_verified, ), }) }) .transpose() }
async fn set_channel_verified(&self, did: &Did, channel: CommsChannel) -> Result<(), DbError> { let column = match channel { CommsChannel::Email => "email_verified", CommsChannel::Discord => "discord_verified", CommsChannel::Telegram => "telegram_verified", CommsChannel::Signal => "signal_verified", }; let query = format!("UPDATE users SET {} = TRUE WHERE did = $1", column); sqlx::query(&query) .bind(did.as_str()) .execute(&self.pool) .await .map_err(map_sqlx_error)?; Ok(()) }
async fn get_id_by_email_or_handle( &self, email: &str, handle: &Handle, ) -> Result<Option<Uuid>, DbError> { sqlx::query_scalar!( "SELECT id FROM users WHERE LOWER(email) = $1 OR handle = $2", email, handle.as_str() ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error) }
async fn count_accounts_by_email(&self, email: &str) -> Result<i64, DbError> { sqlx::query_scalar!( "SELECT COUNT(*) FROM users WHERE LOWER(email) = LOWER($1) AND deactivated_at IS NULL", email ) .fetch_one(&self.pool) .await .map(|c| c.unwrap_or(0)) .map_err(map_sqlx_error) }
async fn get_handles_by_email(&self, email: &str) -> Result<Vec<Handle>, DbError> { let handles = sqlx::query_scalar!( "SELECT handle FROM users WHERE LOWER(email) = LOWER($1) AND deactivated_at IS NULL ORDER BY created_at DESC", email ) .fetch_all(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(handles .into_iter() .filter_map(|h| legacy_column(h, col::USERS_HANDLE)) .collect()) }
async fn set_password_reset_code( &self, user_id: Uuid, code: &str, expires_at: DateTime<Utc>, ) -> Result<(), DbError> { sqlx::query!( "UPDATE users SET password_reset_code = $1, password_reset_code_expires_at = $2 WHERE id = $3", code, expires_at, user_id ) .execute(&self.pool) .await .map_err(map_sqlx_error)?; Ok(()) }
async fn get_user_by_reset_code( &self, code: &str, ) -> Result<Option<UserResetCodeInfo>, DbError> { sqlx::query!( "SELECT id, did, preferred_comms_channel as \"preferred_comms_channel: CommsChannel\", password_reset_code_expires_at FROM users WHERE password_reset_code = $1", code ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)? .map(|row| { Ok(UserResetCodeInfo { id: row.id, did: column(row.did, col::USERS_DID)?, preferred_comms_channel: row.preferred_comms_channel, expires_at: row.password_reset_code_expires_at, }) }) .transpose() }
async fn clear_password_reset_code(&self, user_id: Uuid) -> Result<(), DbError> { sqlx::query!( "UPDATE users SET password_reset_code = NULL, password_reset_code_expires_at = NULL WHERE id = $1", user_id ) .execute(&self.pool) .await .map_err(map_sqlx_error)?; Ok(()) }
async fn get_id_and_password_hash_by_did( &self, did: &Did, ) -> Result<Option<UserIdAndPasswordHash>, DbError> { sqlx::query!( "SELECT id, password_hash FROM users WHERE did = $1", did.as_str() ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error) .map(|opt| { opt.and_then(|row| { row.password_hash.map(|hash| UserIdAndPasswordHash { id: row.id, password_hash: PasswordHash::new(hash), }) }) }) }
async fn update_password_hash( &self, user_id: Uuid, password_hash: &PasswordHash, ) -> Result<(), DbError> { sqlx::query!( "UPDATE users SET password_hash = $1 WHERE id = $2", password_hash.as_str(), user_id ) .execute(&self.pool) .await .map_err(map_sqlx_error)?; Ok(()) }
async fn reset_password_with_sessions( &self, user_id: Uuid, password_hash: &PasswordHash, ) -> Result<PasswordResetResult, DbError> { let mut tx = self.pool.begin().await.map_err(map_sqlx_error)?;
sqlx::query!( "UPDATE users SET password_hash = $1, password_reset_code = NULL, password_reset_code_expires_at = NULL, password_required = TRUE WHERE id = $2", password_hash.as_str(), user_id ) .execute(&mut *tx) .await .map_err(map_sqlx_error)?;
let user_did = sqlx::query_scalar!("SELECT did FROM users WHERE id = $1", user_id) .fetch_one(&mut *tx) .await .map_err(map_sqlx_error)?;
let session_jtis: Vec<Jti> = sqlx::query_scalar!( "SELECT access_jti FROM session_tokens WHERE did = $1", user_did ) .fetch_all(&mut *tx) .await .map_err(map_sqlx_error)? .into_iter() .map(Jti::from) .collect();
sqlx::query!("DELETE FROM session_tokens WHERE did = $1", user_did) .execute(&mut *tx) .await .map_err(map_sqlx_error)?;
let did = column(user_did, col::USERS_DID)?;
tx.commit().await.map_err(map_sqlx_error)?;
Ok(PasswordResetResult { did, session_jtis }) }
async fn activate_account(&self, did: &Did) -> Result<bool, DbError> { let result = sqlx::query!( "UPDATE users SET deactivated_at = NULL, inbound_migration = FALSE WHERE did = $1", did.as_str() ) .execute(&self.pool) .await .map_err(map_sqlx_error)?; Ok(result.rows_affected() > 0) }
async fn deactivate_account( &self, did: &Did, delete_after: Option<DateTime<Utc>>, ) -> Result<bool, DbError> { let result = sqlx::query!( "UPDATE users SET deactivated_at = NOW(), delete_after = $2 WHERE did = $1", did.as_str(), delete_after ) .execute(&self.pool) .await .map_err(map_sqlx_error)?; Ok(result.rows_affected() > 0) }
async fn has_password_by_did(&self, did: &Did) -> Result<Option<bool>, DbError> { sqlx::query_scalar!( "SELECT password_hash IS NOT NULL as has_password FROM users WHERE did = $1", did.as_str() ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error) .map(|opt| opt.flatten()) }
async fn get_password_info_by_did( &self, did: &Did, ) -> Result<Option<UserPasswordInfo>, DbError> { sqlx::query!( "SELECT id, password_hash FROM users WHERE did = $1", did.as_str() ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error) .map(|opt| { opt.map(|row| UserPasswordInfo { id: row.id, password_hash: row.password_hash.map(PasswordHash::new), }) }) }
async fn remove_user_password(&self, user_id: Uuid) -> Result<(), DbError> { sqlx::query!( "UPDATE users SET password_hash = NULL, password_required = FALSE WHERE id = $1", user_id ) .execute(&self.pool) .await .map_err(map_sqlx_error)?; Ok(()) }
async fn set_new_user_password( &self, user_id: Uuid, password_hash: &PasswordHash, ) -> Result<(), DbError> { sqlx::query!( "UPDATE users SET password_hash = $1, password_required = TRUE WHERE id = $2", password_hash.as_str(), user_id ) .execute(&self.pool) .await .map_err(map_sqlx_error)?; Ok(()) }
async fn is_account_active_by_did(&self, did: &Did) -> Result<Option<bool>, DbError> { sqlx::query_scalar!( "SELECT deactivated_at IS NULL as is_active FROM users WHERE did = $1", did.as_str() ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error) .map(|opt| opt.flatten()) }
async fn get_user_for_deletion(&self, did: &Did) -> Result<Option<UserForDeletion>, DbError> { sqlx::query!( "SELECT id, password_hash, handle FROM users WHERE did = $1", did.as_str() ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)? .map(|row| { Ok(UserForDeletion { id: row.id, password_hash: row.password_hash.map(PasswordHash::new), handle: column(row.handle, col::USERS_HANDLE)?, }) }) .transpose() }
async fn get_user_key_by_did(&self, did: &Did) -> Result<Option<UserKeyInfo>, DbError> { sqlx::query!( r#"SELECT uk.key_bytes, uk.encryption_version FROM user_keys uk JOIN users u ON uk.user_id = u.id WHERE u.did = $1"#, did.as_str() ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error) .map(|opt| { opt.map(|row| UserKeyInfo { key_bytes: row.key_bytes, encryption_version: row.encryption_version, }) }) }
async fn delete_account_complete(&self, user_id: Uuid, did: &Did) -> Result<(), DbError> { let mut tx = self.pool.begin().await.map_err(map_sqlx_error)?; sqlx::query!("DELETE FROM session_tokens WHERE did = $1", did.as_str()) .execute(&mut *tx) .await .map_err(map_sqlx_error)?; sqlx::query!("DELETE FROM records WHERE repo_id = $1", user_id) .execute(&mut *tx) .await .map_err(map_sqlx_error)?; sqlx::query!("DELETE FROM repos WHERE user_id = $1", user_id) .execute(&mut *tx) .await .map_err(map_sqlx_error)?; sqlx::query!("DELETE FROM blobs WHERE created_by_user = $1", user_id) .execute(&mut *tx) .await .map_err(map_sqlx_error)?; sqlx::query!("DELETE FROM user_keys WHERE user_id = $1", user_id) .execute(&mut *tx) .await .map_err(map_sqlx_error)?; sqlx::query!("DELETE FROM app_passwords WHERE user_id = $1", user_id) .execute(&mut *tx) .await .map_err(map_sqlx_error)?; sqlx::query!( "DELETE FROM account_deletion_requests WHERE did = $1", did.as_str() ) .execute(&mut *tx) .await .map_err(map_sqlx_error)?; sqlx::query!("DELETE FROM users WHERE id = $1", user_id) .execute(&mut *tx) .await .map_err(map_sqlx_error)?; tx.commit().await.map_err(map_sqlx_error)?; Ok(()) }
async fn set_user_takedown( &self, did: &Did, takedown_ref: Option<&str>, ) -> Result<bool, DbError> { let result = sqlx::query!( "UPDATE users SET takedown_ref = $1 WHERE did = $2", takedown_ref, did.as_str() ) .execute(&self.pool) .await .map_err(map_sqlx_error)?; Ok(result.rows_affected() > 0) }
async fn admin_delete_account_complete(&self, user_id: Uuid, did: &Did) -> Result<(), DbError> { let mut tx = self.pool.begin().await.map_err(map_sqlx_error)?; sqlx::query!("DELETE FROM session_tokens WHERE did = $1", did.as_str()) .execute(&mut *tx) .await .map_err(map_sqlx_error)?; sqlx::query!( "DELETE FROM used_refresh_tokens WHERE session_id IN (SELECT id FROM session_tokens WHERE did = $1)", did.as_str() ) .execute(&mut *tx) .await .ok(); sqlx::query!("DELETE FROM records WHERE repo_id = $1", user_id) .execute(&mut *tx) .await .map_err(map_sqlx_error)?; sqlx::query!("DELETE FROM repos WHERE user_id = $1", user_id) .execute(&mut *tx) .await .map_err(map_sqlx_error)?; sqlx::query!("DELETE FROM blobs WHERE created_by_user = $1", user_id) .execute(&mut *tx) .await .map_err(map_sqlx_error)?; sqlx::query!("DELETE FROM app_passwords WHERE user_id = $1", user_id) .execute(&mut *tx) .await .map_err(map_sqlx_error)?; sqlx::query!( "DELETE FROM invite_code_uses WHERE used_by_user = $1", user_id ) .execute(&mut *tx) .await .ok(); sqlx::query!( "DELETE FROM invite_codes WHERE created_by_user = $1", user_id ) .execute(&mut *tx) .await .ok(); sqlx::query!("DELETE FROM user_keys WHERE user_id = $1", user_id) .execute(&mut *tx) .await .map_err(map_sqlx_error)?; sqlx::query!("DELETE FROM users WHERE id = $1", user_id) .execute(&mut *tx) .await .map_err(map_sqlx_error)?; tx.commit().await.map_err(map_sqlx_error)?; Ok(()) }
async fn get_user_for_did_doc(&self, did: &Did) -> Result<Option<UserForDidDoc>, DbError> { let row = sqlx::query!( "SELECT id, handle, deactivated_at FROM users WHERE did = $1", did.as_str() ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?;
row.map(|r| { Ok(UserForDidDoc { id: r.id, handle: column(r.handle, col::USERS_HANDLE)?, deactivated_at: r.deactivated_at, }) }) .transpose() }
async fn get_user_for_did_doc_build( &self, did: &Did, ) -> Result<Option<UserForDidDocBuild>, DbError> { let row = sqlx::query!( "SELECT id, handle, migrated_to_pds FROM users WHERE did = $1", did.as_str() ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?;
row.map(|r| { Ok(UserForDidDocBuild { id: r.id, handle: column(r.handle, col::USERS_HANDLE)?, migrated_to_pds: r.migrated_to_pds, }) }) .transpose() }
async fn upsert_did_web_overrides( &self, user_id: Uuid, verification_methods: Option<serde_json::Value>, also_known_as: Option<Vec<String>>, ) -> Result<(), DbError> { let now = chrono::Utc::now(); sqlx::query!( r#" INSERT INTO did_web_overrides (user_id, verification_methods, also_known_as, updated_at) VALUES ($1, COALESCE($2, '[]'::jsonb), COALESCE($3, '{}'::text[]), $4) ON CONFLICT (user_id) DO UPDATE SET verification_methods = CASE WHEN $2 IS NOT NULL THEN $2 ELSE did_web_overrides.verification_methods END, also_known_as = CASE WHEN $3 IS NOT NULL THEN $3 ELSE did_web_overrides.also_known_as END, updated_at = $4 "#, user_id, verification_methods, also_known_as.as_deref(), now ) .execute(&self.pool) .await .map_err(map_sqlx_error)?; Ok(()) }
async fn update_migrated_to_pds(&self, did: &Did, endpoint: &str) -> Result<(), DbError> { let now = chrono::Utc::now(); sqlx::query!( "UPDATE users SET migrated_to_pds = $1, migrated_at = $2 WHERE did = $3", endpoint, now, did.as_str() ) .execute(&self.pool) .await .map_err(map_sqlx_error)?; Ok(()) }
async fn get_user_for_passkey_setup( &self, did: &Did, ) -> Result<Option<UserForPasskeySetup>, DbError> { let row = sqlx::query!( r#"SELECT id, handle, recovery_token, recovery_token_expires_at, password_required FROM users WHERE did = $1"#, did.as_str() ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?;
row.map(|r| { Ok(UserForPasskeySetup { id: r.id, handle: column(r.handle, col::USERS_HANDLE)?, recovery_token: r.recovery_token, recovery_token_expires_at: r.recovery_token_expires_at, password_required: r.password_required, }) }) .transpose() }
async fn get_user_for_passkey_recovery( &self, identifier: &str, normalized_handle: &str, ) -> Result<Option<UserForPasskeyRecovery>, DbError> { let row = sqlx::query!( "SELECT id, did, handle, password_required FROM users WHERE LOWER(email) = $1 OR handle = $2", identifier, normalized_handle ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?;
row.map(|r| { Ok(UserForPasskeyRecovery { id: r.id, did: column(r.did, col::USERS_DID)?, handle: column(r.handle, col::USERS_HANDLE)?, password_required: r.password_required, }) }) .transpose() }
async fn set_recovery_token( &self, did: &Did, token_hash: &str, expires_at: DateTime<Utc>, ) -> Result<(), DbError> { sqlx::query!( "UPDATE users SET recovery_token = $1, recovery_token_expires_at = $2 WHERE did = $3", token_hash, expires_at, did.as_str() ) .execute(&self.pool) .await .map_err(map_sqlx_error)?; Ok(()) }
async fn get_user_for_recovery(&self, did: &Did) -> Result<Option<UserForRecovery>, DbError> { let row = sqlx::query!( "SELECT id, did, preferred_comms_channel as \"preferred_comms_channel: CommsChannel\", recovery_token, recovery_token_expires_at FROM users WHERE did = $1", did.as_str() ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?;
row.map(|r| { Ok(UserForRecovery { id: r.id, did: column(r.did, col::USERS_DID)?, preferred_comms_channel: r.preferred_comms_channel, recovery_token: r.recovery_token, recovery_token_expires_at: r.recovery_token_expires_at, }) }) .transpose() }
async fn get_accounts_scheduled_for_deletion( &self, limit: i64, ) -> Result<Vec<tranquil_db_traits::ScheduledDeletionAccount>, DbError> { let rows = sqlx::query!( r#" SELECT id, did, handle FROM users WHERE delete_after IS NOT NULL AND delete_after < NOW() AND deactivated_at IS NOT NULL LIMIT $1 "#, limit ) .fetch_all(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(rows .into_iter() .filter_map(|r| { Some(tranquil_db_traits::ScheduledDeletionAccount { id: r.id, did: legacy_column(r.did, col::USERS_DID)?, handle: legacy_column(r.handle, col::USERS_HANDLE)?, }) }) .collect()) }
async fn delete_account_with_firehose(&self, user_id: Uuid, did: &Did) -> Result<(), DbError> { let mut tx = self.pool.begin().await.map_err(map_sqlx_error)?;
sqlx::query!("DELETE FROM blobs WHERE created_by_user = $1", user_id) .execute(&mut *tx) .await .map_err(map_sqlx_error)?;
sqlx::query!("DELETE FROM record_blobs WHERE repo_id = $1", user_id) .execute(&mut *tx) .await .map_err(map_sqlx_error)?;
sqlx::query!("DELETE FROM records WHERE repo_id = $1", user_id) .execute(&mut *tx) .await .map_err(map_sqlx_error)?;
sqlx::query!("DELETE FROM repos WHERE user_id = $1", user_id) .execute(&mut *tx) .await .map_err(map_sqlx_error)?;
sqlx::query!("DELETE FROM user_blocks WHERE user_id = $1", user_id) .execute(&mut *tx) .await .map_err(map_sqlx_error)?;
sqlx::query!("DELETE FROM user_keys WHERE user_id = $1", user_id) .execute(&mut *tx) .await .map_err(map_sqlx_error)?;
sqlx::query!("DELETE FROM session_tokens WHERE did = $1", did.as_str()) .execute(&mut *tx) .await .map_err(map_sqlx_error)?;
sqlx::query!("DELETE FROM app_passwords WHERE user_id = $1", user_id) .execute(&mut *tx) .await .map_err(map_sqlx_error)?;
sqlx::query!("DELETE FROM passkeys WHERE did = $1", did.as_str()) .execute(&mut *tx) .await .map_err(map_sqlx_error)?;
sqlx::query!("DELETE FROM user_totp WHERE did = $1", did.as_str()) .execute(&mut *tx) .await .map_err(map_sqlx_error)?;
sqlx::query!("DELETE FROM backup_codes WHERE did = $1", did.as_str()) .execute(&mut *tx) .await .map_err(map_sqlx_error)?;
sqlx::query!( "DELETE FROM webauthn_challenges WHERE did = $1", did.as_str() ) .execute(&mut *tx) .await .map_err(map_sqlx_error)?;
sqlx::query!( "DELETE FROM account_deletion_requests WHERE did = $1", did.as_str() ) .execute(&mut *tx) .await .map_err(map_sqlx_error)?;
sqlx::query!("DELETE FROM users WHERE id = $1", user_id) .execute(&mut *tx) .await .map_err(map_sqlx_error)?;
let event_id: i64 = sqlx::query_scalar!( r#" INSERT INTO repo_seq (did, event_type, active, status) VALUES ($1, 'account', false, 'deleted') RETURNING id "#, did.as_str() ) .fetch_one(&mut *tx) .await .map_err(map_sqlx_error)?;
sqlx::query!( "DELETE FROM repo_seq WHERE did = $1 AND id <> $2", did.as_str(), event_id ) .execute(&mut *tx) .await .map_err(map_sqlx_error)?;
tx.commit().await.map_err(map_sqlx_error)?;
sqlx::query!("NOTIFY repo_updates") .execute(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(()) }
async fn create_password_account( &self, input: &tranquil_db_traits::CreatePasswordAccountInput, ) -> Result< tranquil_db_traits::CreatePasswordAccountResult, tranquil_db_traits::CreateAccountError, > { tracing::info!(did = %input.did, handle = %input.handle, "create_password_account: starting transaction"); let mut tx = self.pool.begin().await.map_err(|e: sqlx::Error| { tracing::error!( "create_password_account: failed to begin transaction: {}", e ); tranquil_db_traits::CreateAccountError::Database(e.to_string()) })?;
let is_first_user: bool = sqlx::query_scalar!("SELECT COUNT(*) as count FROM users") .fetch_one(&mut *tx) .await .map(|c| c.unwrap_or(0) == 0) .unwrap_or(false);
let user_insert: Result<(uuid::Uuid,), _> = sqlx::query_as( r#"INSERT INTO users ( handle, email, did, password_hash, preferred_comms_channel, discord_username, telegram_username, signal_username, is_admin, deactivated_at, inbound_migration, email_verified ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, FALSE) RETURNING id"#, ) .bind(input.handle.as_str()) .bind(&input.email) .bind(input.did.as_str()) .bind(&input.password_hash) .bind(input.preferred_comms_channel) .bind(&input.discord_username) .bind(&input.telegram_username) .bind(&input.signal_username) .bind(is_first_user) .bind(input.deactivated_at) .bind(input.inbound_migration) .fetch_one(&mut *tx) .await;
let user_id = match user_insert { Ok((id,)) => { tracing::info!(did = %input.did, user_id = %id, "create_password_account: user row inserted"); id } Err(e) => { tracing::error!(did = %input.did, error = %e, "create_password_account: user insert failed"); if let Some(db_err) = e.as_database_error() && db_err.code().as_deref() == Some("23505") { let constraint = db_err.constraint().unwrap_or(""); if constraint.contains("handle") { return Err(tranquil_db_traits::CreateAccountError::HandleTaken); } else if constraint.contains("email") { return Err(tranquil_db_traits::CreateAccountError::EmailTaken); } else if constraint.contains("did") { return Err(tranquil_db_traits::CreateAccountError::DidExists); } } return Err(tranquil_db_traits::CreateAccountError::Database( e.to_string(), )); } };
sqlx::query!( "INSERT INTO user_keys (user_id, key_bytes, encryption_version, encrypted_at) VALUES ($1, $2, $3, NOW())", user_id, &input.encrypted_key_bytes[..], input.encryption_version ) .execute(&mut *tx) .await .map_err(|e: sqlx::Error| tranquil_db_traits::CreateAccountError::Database(e.to_string()))?;
if let Some(key_id) = input.reserved_key_id { sqlx::query!( "UPDATE reserved_signing_keys SET used_at = NOW() WHERE id = $1", key_id ) .execute(&mut *tx) .await .map_err(|e: sqlx::Error| { tranquil_db_traits::CreateAccountError::Database(e.to_string()) })?; }
sqlx::query!( "INSERT INTO repos (user_id, repo_root_cid, repo_rev) VALUES ($1, $2, $3)", user_id, input.commit_cid.as_str(), input.repo_rev.as_str() ) .execute(&mut *tx) .await .map_err(|e: sqlx::Error| { tranquil_db_traits::CreateAccountError::Database(e.to_string()) })?;
sqlx::query( r#" INSERT INTO user_blocks (user_id, block_cid, repo_rev) SELECT $1, block_cid, $3 FROM UNNEST($2::bytea[]) AS t(block_cid) ON CONFLICT (user_id, block_cid) DO NOTHING "#, ) .bind(user_id) .bind(&input.genesis_block_cids) .bind(&input.repo_rev) .execute(&mut *tx) .await .map_err(|e: sqlx::Error| { tranquil_db_traits::CreateAccountError::Database(e.to_string()) })?;
if let Some(code) = &input.invite_code { consume_invite_code(&mut tx, code, user_id).await?; }
if let Some(birthdate_pref) = &input.birthdate_pref { let _ = sqlx::query!( "INSERT INTO account_preferences (user_id, name, value_json) VALUES ($1, $2, $3)", user_id, "app.bsky.actor.defs#personalDetailsPref", birthdate_pref ) .execute(&mut *tx) .await; }
tracing::info!(did = %input.did, user_id = %user_id, "create_password_account: committing transaction"); tx.commit().await.map_err(|e: sqlx::Error| { tracing::error!(did = %input.did, user_id = %user_id, error = %e, "create_password_account: commit failed"); tranquil_db_traits::CreateAccountError::Database(e.to_string()) })?; tracing::info!(did = %input.did, user_id = %user_id, "create_password_account: transaction committed successfully");
Ok(tranquil_db_traits::CreatePasswordAccountResult { user_id, is_admin: is_first_user, }) }
async fn create_delegated_account( &self, input: &tranquil_db_traits::CreateDelegatedAccountInput, ) -> Result<uuid::Uuid, tranquil_db_traits::CreateAccountError> { let mut tx = self.pool.begin().await.map_err(|e: sqlx::Error| { tranquil_db_traits::CreateAccountError::Database(e.to_string()) })?;
let user_insert: Result<(uuid::Uuid,), _> = sqlx::query_as( r#"INSERT INTO users ( handle, email, did, password_hash, password_required, account_type, preferred_comms_channel ) VALUES ($1, $2, $3, NULL, FALSE, 'delegated'::account_type, 'email'::comms_channel) RETURNING id"#, ) .bind(input.handle.as_str()) .bind(&input.email) .bind(input.did.as_str()) .fetch_one(&mut *tx) .await;
let user_id = match user_insert { Ok((id,)) => id, Err(e) => { if let Some(db_err) = e.as_database_error() && db_err.code().as_deref() == Some("23505") { let constraint = db_err.constraint().unwrap_or(""); if constraint.contains("handle") { return Err(tranquil_db_traits::CreateAccountError::HandleTaken); } else if constraint.contains("email") { return Err(tranquil_db_traits::CreateAccountError::EmailTaken); } } return Err(tranquil_db_traits::CreateAccountError::Database( e.to_string(), )); } };
sqlx::query!( "INSERT INTO user_keys (user_id, key_bytes, encryption_version, encrypted_at) VALUES ($1, $2, $3, NOW())", user_id, &input.encrypted_key_bytes[..], input.encryption_version ) .execute(&mut *tx) .await .map_err(|e: sqlx::Error| tranquil_db_traits::CreateAccountError::Database(e.to_string()))?;
sqlx::query!( r#"INSERT INTO account_delegations (delegated_did, controller_did, granted_scopes, granted_by) VALUES ($1, $2, $3, $4)"#, input.did.as_str(), input.controller_did.as_str(), &input.controller_scopes, input.controller_did.as_str() ) .execute(&mut *tx) .await .map_err(|e: sqlx::Error| tranquil_db_traits::CreateAccountError::Database(e.to_string()))?;
sqlx::query!( "INSERT INTO repos (user_id, repo_root_cid, repo_rev) VALUES ($1, $2, $3)", user_id, input.commit_cid.as_str(), input.repo_rev.as_str() ) .execute(&mut *tx) .await .map_err(|e: sqlx::Error| { tranquil_db_traits::CreateAccountError::Database(e.to_string()) })?;
sqlx::query( r#" INSERT INTO user_blocks (user_id, block_cid, repo_rev) SELECT $1, block_cid, $3 FROM UNNEST($2::bytea[]) AS t(block_cid) ON CONFLICT (user_id, block_cid) DO NOTHING "#, ) .bind(user_id) .bind(&input.genesis_block_cids) .bind(&input.repo_rev) .execute(&mut *tx) .await .map_err(|e: sqlx::Error| { tranquil_db_traits::CreateAccountError::Database(e.to_string()) })?;
tx.commit().await.map_err(|e: sqlx::Error| { tranquil_db_traits::CreateAccountError::Database(e.to_string()) })?;
Ok(user_id) }
async fn create_passkey_account( &self, input: &tranquil_db_traits::CreatePasskeyAccountInput, ) -> Result< tranquil_db_traits::CreatePasswordAccountResult, tranquil_db_traits::CreateAccountError, > { let mut tx = self.pool.begin().await.map_err(|e: sqlx::Error| { tranquil_db_traits::CreateAccountError::Database(e.to_string()) })?;
let is_first_user: bool = sqlx::query_scalar!("SELECT COUNT(*) as count FROM users") .fetch_one(&mut *tx) .await .map(|c| c.unwrap_or(0) == 0) .unwrap_or(false);
let user_insert: Result<(uuid::Uuid,), _> = sqlx::query_as( r#"INSERT INTO users ( handle, email, did, password_hash, password_required, preferred_comms_channel, discord_username, telegram_username, signal_username, recovery_token, recovery_token_expires_at, is_admin, deactivated_at ) VALUES ($1, $2, $3, NULL, FALSE, $4, $5, $6, $7, $8, $9, $10, $11) RETURNING id"#, ) .bind(input.handle.as_str()) .bind(&input.email) .bind(input.did.as_str()) .bind(input.preferred_comms_channel) .bind(&input.discord_username) .bind(&input.telegram_username) .bind(&input.signal_username) .bind(&input.setup_token_hash) .bind(input.setup_expires_at) .bind(is_first_user) .bind(input.deactivated_at) .fetch_one(&mut *tx) .await;
let user_id = match user_insert { Ok((id,)) => id, Err(e) => { if let Some(db_err) = e.as_database_error() && db_err.code().as_deref() == Some("23505") { let constraint = db_err.constraint().unwrap_or(""); if constraint.contains("handle") { return Err(tranquil_db_traits::CreateAccountError::HandleTaken); } else if constraint.contains("email") { return Err(tranquil_db_traits::CreateAccountError::EmailTaken); } } return Err(tranquil_db_traits::CreateAccountError::Database( e.to_string(), )); } };
sqlx::query!( "INSERT INTO user_keys (user_id, key_bytes, encryption_version, encrypted_at) VALUES ($1, $2, $3, NOW())", user_id, &input.encrypted_key_bytes[..], input.encryption_version ) .execute(&mut *tx) .await .map_err(|e: sqlx::Error| tranquil_db_traits::CreateAccountError::Database(e.to_string()))?;
if let Some(key_id) = input.reserved_key_id { sqlx::query!( "UPDATE reserved_signing_keys SET used_at = NOW() WHERE id = $1", key_id ) .execute(&mut *tx) .await .map_err(|e: sqlx::Error| { tranquil_db_traits::CreateAccountError::Database(e.to_string()) })?; }
sqlx::query!( "INSERT INTO repos (user_id, repo_root_cid, repo_rev) VALUES ($1, $2, $3)", user_id, input.commit_cid.as_str(), input.repo_rev.as_str() ) .execute(&mut *tx) .await .map_err(|e: sqlx::Error| { tranquil_db_traits::CreateAccountError::Database(e.to_string()) })?;
sqlx::query( r#" INSERT INTO user_blocks (user_id, block_cid, repo_rev) SELECT $1, block_cid, $3 FROM UNNEST($2::bytea[]) AS t(block_cid) ON CONFLICT (user_id, block_cid) DO NOTHING "#, ) .bind(user_id) .bind(&input.genesis_block_cids) .bind(&input.repo_rev) .execute(&mut *tx) .await .map_err(|e: sqlx::Error| { tranquil_db_traits::CreateAccountError::Database(e.to_string()) })?;
if let Some(code) = &input.invite_code { consume_invite_code(&mut tx, code, user_id).await?; }
if let Some(birthdate_pref) = &input.birthdate_pref { let _ = sqlx::query!( "INSERT INTO account_preferences (user_id, name, value_json) VALUES ($1, $2, $3)", user_id, "app.bsky.actor.defs#personalDetailsPref", birthdate_pref ) .execute(&mut *tx) .await; }
tx.commit().await.map_err(|e: sqlx::Error| { tranquil_db_traits::CreateAccountError::Database(e.to_string()) })?;
Ok(tranquil_db_traits::CreatePasswordAccountResult { user_id, is_admin: is_first_user, }) }
async fn create_sso_account( &self, input: &tranquil_db_traits::CreateSsoAccountInput, ) -> Result< tranquil_db_traits::CreatePasswordAccountResult, tranquil_db_traits::CreateAccountError, > { let mut tx = self.pool.begin().await.map_err(|e: sqlx::Error| { tranquil_db_traits::CreateAccountError::Database(e.to_string()) })?;
let token_consumed: Option<(String,)> = sqlx::query_as( r#" DELETE FROM sso_pending_registration WHERE token = $1 AND expires_at > NOW() RETURNING token "#, ) .bind(&input.pending_registration_token) .fetch_optional(&mut *tx) .await .map_err(|e: sqlx::Error| { tranquil_db_traits::CreateAccountError::Database(e.to_string()) })?;
if token_consumed.is_none() { return Err(tranquil_db_traits::CreateAccountError::InvalidToken); }
let is_first_user: bool = sqlx::query_scalar!("SELECT COUNT(*) as count FROM users") .fetch_one(&mut *tx) .await .map(|c| c.unwrap_or(0) == 0) .unwrap_or(false);
let user_insert: Result<(uuid::Uuid,), _> = sqlx::query_as( r#"INSERT INTO users ( handle, email, did, password_hash, password_required, preferred_comms_channel, discord_username, telegram_username, signal_username, is_admin ) VALUES ($1, $2, $3, NULL, FALSE, $4, $5, $6, $7, $8) RETURNING id"#, ) .bind(input.handle.as_str()) .bind(&input.email) .bind(input.did.as_str()) .bind(input.preferred_comms_channel) .bind(&input.discord_username) .bind(&input.telegram_username) .bind(&input.signal_username) .bind(is_first_user) .fetch_one(&mut *tx) .await;
let user_id = match user_insert { Ok((id,)) => id, Err(e) => { if let Some(db_err) = e.as_database_error() && db_err.code().as_deref() == Some("23505") { let constraint = db_err.constraint().unwrap_or(""); if constraint.contains("handle") { return Err(tranquil_db_traits::CreateAccountError::HandleTaken); } else if constraint.contains("email") { return Err(tranquil_db_traits::CreateAccountError::EmailTaken); } } return Err(tranquil_db_traits::CreateAccountError::Database( e.to_string(), )); } };
sqlx::query!( "INSERT INTO user_keys (user_id, key_bytes, encryption_version, encrypted_at) VALUES ($1, $2, $3, NOW())", user_id, &input.encrypted_key_bytes[..], input.encryption_version ) .execute(&mut *tx) .await .map_err(|e: sqlx::Error| tranquil_db_traits::CreateAccountError::Database(e.to_string()))?;
sqlx::query!( "INSERT INTO repos (user_id, repo_root_cid, repo_rev) VALUES ($1, $2, $3)", user_id, input.commit_cid.as_str(), input.repo_rev.as_str() ) .execute(&mut *tx) .await .map_err(|e: sqlx::Error| { tranquil_db_traits::CreateAccountError::Database(e.to_string()) })?;
sqlx::query( r#" INSERT INTO user_blocks (user_id, block_cid, repo_rev) SELECT $1, block_cid, $3 FROM UNNEST($2::bytea[]) AS t(block_cid) ON CONFLICT (user_id, block_cid) DO NOTHING "#, ) .bind(user_id) .bind(&input.genesis_block_cids) .bind(&input.repo_rev) .execute(&mut *tx) .await .map_err(|e: sqlx::Error| { tranquil_db_traits::CreateAccountError::Database(e.to_string()) })?;
if let Some(code) = &input.invite_code { consume_invite_code(&mut tx, code, user_id).await?; }
if let Some(birthdate_pref) = &input.birthdate_pref { let _ = sqlx::query!( "INSERT INTO account_preferences (user_id, name, value_json) VALUES ($1, $2, $3)", user_id, "app.bsky.actor.defs#personalDetailsPref", birthdate_pref ) .execute(&mut *tx) .await; }
sqlx::query!( r#" INSERT INTO external_identities (did, provider, provider_user_id, provider_username, provider_email, provider_email_verified) VALUES ($1, $2, $3, $4, $5, $6) "#, input.did.as_str(), input.sso_provider as SsoProviderType, &input.sso_provider_user_id, input.sso_provider_username.as_deref(), input.sso_provider_email.as_deref(), input.sso_provider_email_verified, ) .execute(&mut *tx) .await .map_err(|e: sqlx::Error| { tranquil_db_traits::CreateAccountError::Database(e.to_string()) })?;
tx.commit().await.map_err(|e: sqlx::Error| { tranquil_db_traits::CreateAccountError::Database(e.to_string()) })?;
Ok(tranquil_db_traits::CreatePasswordAccountResult { user_id, is_admin: is_first_user, }) }
async fn reactivate_migration_account( &self, input: &tranquil_db_traits::MigrationReactivationInput, ) -> Result< tranquil_db_traits::ReactivatedAccountInfo, tranquil_db_traits::MigrationReactivationError, > { let mut tx = self.pool.begin().await.map_err(|e| { tranquil_db_traits::MigrationReactivationError::Database(e.to_string()) })?;
let existing: Option<(uuid::Uuid, String, Option<chrono::DateTime<chrono::Utc>>)> = sqlx::query_as( "SELECT id, handle, deactivated_at FROM users WHERE did = $1 FOR UPDATE", ) .bind(input.did.as_str()) .fetch_optional(&mut *tx) .await .map_err(|e| tranquil_db_traits::MigrationReactivationError::Database(e.to_string()))?;
let (account_id, old_handle, deactivated_at) = existing.ok_or(tranquil_db_traits::MigrationReactivationError::NotFound)?;
if deactivated_at.is_none() { return Err(tranquil_db_traits::MigrationReactivationError::NotDeactivated); }
let update_result: Result<_, sqlx::Error> = if let Some(ref new_email) = input.new_email { sqlx::query( "UPDATE users SET handle = $1, email = $2, email_verified = false WHERE id = $3", ) .bind(input.new_handle.as_str()) .bind(new_email) .bind(account_id) .execute(&mut *tx) .await } else { sqlx::query("UPDATE users SET handle = $1 WHERE id = $2") .bind(input.new_handle.as_str()) .bind(account_id) .execute(&mut *tx) .await };
if let Err(e) = update_result { if let Some(db_err) = e.as_database_error() && db_err .constraint() .map(|c| c.contains("handle")) .unwrap_or(false) { return Err(tranquil_db_traits::MigrationReactivationError::HandleTaken); } return Err(tranquil_db_traits::MigrationReactivationError::Database( e.to_string(), )); }
let old_handle = legacy_column(old_handle, col::USERS_HANDLE);
tx.commit() .await .map_err(|e| tranquil_db_traits::MigrationReactivationError::Database(e.to_string()))?;
Ok(tranquil_db_traits::ReactivatedAccountInfo { user_id: account_id, old_handle, }) }
async fn check_handle_available_for_new_account( &self, handle: &Handle, ) -> Result<bool, DbError> { let exists: Option<(i32,)> = sqlx::query_as( r#" SELECT 1 FROM users WHERE handle = $1 AND deactivated_at IS NULL UNION ALL SELECT 1 FROM handle_reservations WHERE handle = $1 AND expires_at > NOW() LIMIT 1 "#, ) .bind(handle.as_str()) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(exists.is_none()) }
async fn reserve_handle(&self, handle: &Handle, reserved_by: &str) -> Result<bool, DbError> { sqlx::query!("DELETE FROM handle_reservations WHERE expires_at <= NOW()") .execute(&self.pool) .await .map_err(map_sqlx_error)?;
let result = sqlx::query!( r#" INSERT INTO handle_reservations (handle, reserved_by) SELECT $1, $2 WHERE NOT EXISTS ( SELECT 1 FROM users WHERE handle = $1 AND deactivated_at IS NULL ) AND NOT EXISTS ( SELECT 1 FROM handle_reservations WHERE handle = $1 AND expires_at > NOW() ) "#, handle.as_str(), reserved_by, ) .execute(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(result.rows_affected() > 0) }
async fn release_handle_reservation(&self, handle: &Handle) -> Result<(), DbError> { sqlx::query!( "DELETE FROM handle_reservations WHERE handle = $1", handle.as_str() ) .execute(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(()) }
async fn cleanup_expired_handle_reservations(&self) -> Result<u64, DbError> { let result = sqlx::query!("DELETE FROM handle_reservations WHERE expires_at <= NOW()") .execute(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(result.rows_affected()) }
async fn complete_passkey_setup( &self, input: &tranquil_db_traits::CompletePasskeySetupInput, ) -> Result<(), DbError> { let mut tx = self.pool.begin().await.map_err(map_sqlx_error)?;
sqlx::query!( "INSERT INTO app_passwords (user_id, name, password_hash, privileged) VALUES ($1, $2, $3, FALSE)", input.user_id, input.app_password_name, input.app_password_hash.as_str() ) .execute(&mut *tx) .await .map_err(map_sqlx_error)?;
sqlx::query!( "UPDATE users SET recovery_token = NULL, recovery_token_expires_at = NULL WHERE did = $1", input.did.as_str() ) .execute(&mut *tx) .await .map_err(map_sqlx_error)?;
tx.commit().await.map_err(map_sqlx_error)?;
Ok(()) }
async fn recover_passkey_account( &self, input: &tranquil_db_traits::RecoverPasskeyAccountInput, ) -> Result<tranquil_db_traits::RecoverPasskeyAccountResult, DbError> { let mut tx = self.pool.begin().await.map_err(map_sqlx_error)?;
sqlx::query!( "UPDATE users SET password_hash = $1, password_required = TRUE, recovery_token = NULL, recovery_token_expires_at = NULL WHERE did = $2", input.password_hash.as_str(), input.did.as_str() ) .execute(&mut *tx) .await .map_err(map_sqlx_error)?;
let deleted = sqlx::query!("DELETE FROM passkeys WHERE did = $1", input.did.as_str()) .execute(&mut *tx) .await .map_err(map_sqlx_error)?;
tx.commit().await.map_err(map_sqlx_error)?;
Ok(tranquil_db_traits::RecoverPasskeyAccountResult { passkeys_deleted: deleted.rows_affected(), }) }
async fn set_unverified_telegram( &self, user_id: Uuid, telegram_username: &str, ) -> Result<(), DbError> { sqlx::query!( r#"UPDATE users SET telegram_username = $1, telegram_verified = CASE WHEN LOWER(telegram_username) = LOWER($1) THEN telegram_verified ELSE FALSE END, telegram_chat_id = CASE WHEN LOWER(telegram_username) = LOWER($1) THEN telegram_chat_id ELSE NULL END, updated_at = NOW() WHERE id = $2"#, telegram_username, user_id ) .execute(&self.pool) .await .map_err(map_sqlx_error)?; Ok(()) }
async fn set_unverified_signal( &self, user_id: Uuid, signal_username: &str, ) -> Result<(), DbError> { sqlx::query!( r#"UPDATE users SET signal_username = $1, signal_verified = CASE WHEN LOWER(signal_username) = LOWER($1) THEN signal_verified ELSE FALSE END, updated_at = NOW() WHERE id = $2"#, signal_username, user_id ) .execute(&self.pool) .await .map_err(map_sqlx_error)?; Ok(()) }
async fn set_unverified_discord( &self, user_id: Uuid, discord_username: &str, ) -> Result<(), DbError> { sqlx::query!( r#"UPDATE users SET discord_username = $1, discord_verified = CASE WHEN LOWER(discord_username) = LOWER($1) THEN discord_verified ELSE FALSE END, discord_id = CASE WHEN LOWER(discord_username) = LOWER($1) THEN discord_id ELSE NULL END, updated_at = NOW() WHERE id = $2"#, discord_username, user_id ) .execute(&self.pool) .await .map_err(map_sqlx_error)?; Ok(()) }
async fn store_discord_user_id( &self, discord_username: &str, discord_id: &str, handle: Option<&Handle>, ) -> Result<Option<Uuid>, DbError> { let result = match handle { Some(h) => sqlx::query_scalar!( "UPDATE users SET discord_id = $2, discord_verified = TRUE, updated_at = NOW() WHERE LOWER(discord_username) = LOWER($1) AND discord_username IS NOT NULL AND handle = $3 RETURNING id", discord_username, discord_id, h.as_str() ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?, None => { let mut tx = self.pool.begin().await.map_err(map_sqlx_error)?;
let matching: Vec<uuid::Uuid> = match sqlx::query_scalar!( "SELECT id FROM users WHERE LOWER(discord_username) = LOWER($1) AND discord_username IS NOT NULL AND deactivated_at IS NULL FOR UPDATE NOWAIT", discord_username ) .fetch_all(&mut *tx) .await { Ok(ids) => ids, Err(sqlx::Error::Database(ref db_err)) if db_err.code().as_deref() == Some("55P03") => { return Err(DbError::LockContention); } Err(e) => return Err(map_sqlx_error(e)), };
let result = match matching.len() { 0 => None, 1 => { sqlx::query_scalar!( "UPDATE users SET discord_id = $2, discord_verified = TRUE, updated_at = NOW() WHERE id = $1 RETURNING id", matching[0], discord_id ) .fetch_optional(&mut *tx) .await .map_err(map_sqlx_error)? } _ => { tx.rollback().await.ok(); return Err(DbError::Ambiguous( "Multiple accounts use this Discord username. Type: /start your-handle.example.com".to_string(), )); } };
tx.commit().await.map_err(map_sqlx_error)?; result } }; Ok(result) }
async fn store_telegram_chat_id( &self, telegram_username: &str, chat_id: i64, handle: Option<&Handle>, ) -> Result<Option<Uuid>, DbError> { let result = match handle { Some(h) => sqlx::query_scalar!( "UPDATE users SET telegram_chat_id = $2, telegram_verified = TRUE, updated_at = NOW() WHERE LOWER(telegram_username) = LOWER($1) AND telegram_username IS NOT NULL AND handle = $3 RETURNING id", telegram_username, chat_id, h.as_str() ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?, None => sqlx::query_scalar!( r#"UPDATE users SET telegram_chat_id = $2, telegram_verified = TRUE, updated_at = NOW() WHERE id = ( SELECT id FROM users WHERE LOWER(telegram_username) = LOWER($1) AND telegram_username IS NOT NULL AND deactivated_at IS NULL LIMIT 1 ) RETURNING id"#, telegram_username, chat_id ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?, }; Ok(result) }
async fn get_telegram_chat_id(&self, user_id: Uuid) -> Result<Option<i64>, DbError> { let row = sqlx::query_scalar!("SELECT telegram_chat_id FROM users WHERE id = $1", user_id) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?; Ok(row.flatten()) }
async fn get_password_reset_info( &self, email: &str, ) -> Result<Option<tranquil_db_traits::PasswordResetInfo>, DbError> { let row = sqlx::query!( "SELECT password_reset_code, password_reset_code_expires_at FROM users WHERE email = $1", email ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(row.map(|r| tranquil_db_traits::PasswordResetInfo { code: r.password_reset_code, expires_at: r.password_reset_code_expires_at, })) }
async fn enable_totp_verified( &self, did: &Did, encrypted_secret: &[u8], ) -> Result<(), DbError> { sqlx::query!( r#"INSERT INTO user_totp (did, secret_encrypted, encryption_version, verified, created_at) VALUES ($1, $2, 1, TRUE, NOW()) ON CONFLICT (did) DO UPDATE SET secret_encrypted = $2, verified = TRUE"#, did.as_str(), encrypted_secret ) .execute(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(()) }
async fn set_two_factor_enabled(&self, did: &Did, enabled: bool) -> Result<(), DbError> { sqlx::query!( "UPDATE users SET two_factor_enabled = $1 WHERE did = $2", enabled, did.as_str() ) .execute(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(()) }
async fn expire_password_reset_code(&self, email: &str) -> Result<(), DbError> { sqlx::query!( "UPDATE users SET password_reset_code_expires_at = NOW() - INTERVAL '1 hour' WHERE email = $1", email ) .execute(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(()) }}