Something went wrong. Try again.
A downstream fork of Tranquil PDS for use in margin.cafe
Something went wrong. Try again.
116 kB · 3476 lines
Rust
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821182218231824182518261827182818291830183118321833183418351836183718381839184018411842184318441845184618471848184918501851185218531854185518561857185818591860186118621863186418651866186718681869187018711872187318741875187618771878187918801881188218831884188518861887188818891890189118921893189418951896189718981899190019011902190319041905190619071908190919101911191219131914191519161917191819191920192119221923192419251926192719281929193019311932193319341935193619371938193919401941194219431944194519461947194819491950195119521953195419551956195719581959196019611962196319641965196619671968196919701971197219731974197519761977197819791980198119821983198419851986198719881989199019911992199319941995199619971998199920002001200220032004200520062007200820092010201120122013201420152016201720182019202020212022202320242025202620272028202920302031203220332034203520362037203820392040204120422043204420452046204720482049205020512052205320542055205620572058205920602061206220632064206520662067206820692070207120722073207420752076207720782079208020812082208320842085208620872088208920902091209220932094209520962097209820992100210121022103210421052106210721082109211021112112211321142115211621172118211921202121212221232124212521262127212821292130213121322133213421352136213721382139214021412142214321442145214621472148214921502151215221532154215521562157215821592160216121622163216421652166216721682169217021712172217321742175217621772178217921802181218221832184218521862187218821892190219121922193219421952196219721982199220022012202220322042205220622072208220922102211221222132214221522162217221822192220222122222223222422252226222722282229223022312232223322342235223622372238223922402241224222432244224522462247224822492250225122522253225422552256225722582259226022612262226322642265226622672268226922702271227222732274227522762277227822792280228122822283228422852286228722882289229022912292229322942295229622972298229923002301230223032304230523062307230823092310231123122313231423152316231723182319232023212322232323242325232623272328232923302331233223332334233523362337233823392340234123422343234423452346234723482349235023512352235323542355235623572358235923602361236223632364236523662367236823692370237123722373237423752376237723782379238023812382238323842385238623872388238923902391239223932394239523962397239823992400240124022403240424052406240724082409241024112412241324142415241624172418241924202421242224232424242524262427242824292430243124322433243424352436243724382439244024412442244324442445244624472448244924502451245224532454245524562457245824592460246124622463246424652466246724682469247024712472247324742475247624772478247924802481248224832484248524862487248824892490249124922493249424952496249724982499250025012502250325042505250625072508250925102511251225132514251525162517251825192520252125222523252425252526252725282529253025312532253325342535253625372538253925402541254225432544254525462547254825492550255125522553255425552556255725582559256025612562256325642565256625672568256925702571257225732574257525762577257825792580258125822583258425852586258725882589259025912592259325942595259625972598259926002601260226032604260526062607260826092610261126122613261426152616261726182619262026212622262326242625262626272628262926302631263226332634263526362637263826392640264126422643264426452646264726482649265026512652265326542655265626572658265926602661266226632664266526662667266826692670267126722673267426752676267726782679268026812682268326842685268626872688268926902691269226932694269526962697269826992700270127022703270427052706270727082709271027112712271327142715271627172718271927202721272227232724272527262727272827292730273127322733273427352736273727382739274027412742274327442745274627472748274927502751275227532754275527562757275827592760276127622763276427652766276727682769277027712772277327742775277627772778277927802781278227832784278527862787278827892790279127922793279427952796279727982799280028012802280328042805280628072808280928102811281228132814281528162817281828192820282128222823282428252826282728282829283028312832283328342835283628372838283928402841284228432844284528462847284828492850285128522853285428552856285728582859286028612862286328642865286628672868286928702871287228732874287528762877287828792880288128822883288428852886288728882889289028912892289328942895289628972898289929002901290229032904290529062907290829092910291129122913291429152916291729182919292029212922292329242925292629272928292929302931293229332934293529362937293829392940294129422943294429452946294729482949295029512952295329542955295629572958295929602961296229632964296529662967296829692970297129722973297429752976297729782979298029812982298329842985298629872988298929902991299229932994299529962997299829993000300130023003300430053006300730083009301030113012301330143015301630173018301930203021302230233024302530263027302830293030303130323033303430353036303730383039304030413042304330443045304630473048304930503051305230533054305530563057305830593060306130623063306430653066306730683069307030713072307330743075307630773078307930803081308230833084308530863087308830893090309130923093309430953096309730983099310031013102310331043105310631073108310931103111311231133114311531163117311831193120312131223123312431253126312731283129313031313132313331343135313631373138313931403141314231433144314531463147314831493150315131523153315431553156315731583159316031613162316331643165316631673168316931703171317231733174317531763177317831793180318131823183318431853186318731883189319031913192319331943195319631973198319932003201320232033204320532063207320832093210321132123213321432153216321732183219322032213222322332243225322632273228322932303231323232333234323532363237323832393240324132423243324432453246324732483249325032513252325332543255325632573258325932603261326232633264326532663267326832693270327132723273327432753276327732783279328032813282328332843285328632873288328932903291329232933294329532963297329832993300330133023303330433053306330733083309331033113312331333143315331633173318331933203321332233233324332533263327332833293330333133323333333433353336333733383339334033413342334333443345334633473348334933503351335233533354335533563357335833593360336133623363336433653366336733683369337033713372337333743375337633773378337933803381338233833384338533863387338833893390339133923393339433953396339733983399340034013402340334043405340634073408340934103411341234133414341534163417341834193420342134223423342434253426342734283429343034313432343334343435343634373438343934403441344234433444344534463447344834493450345134523453345434553456345734583459346034613462346334643465346634673468346934703471347234733474347534763477use 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,};
const FIRST_ADMIN_ADVISORY_LOCK_KEY: i64 = 0x7472_616e_7175_696c;
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 fn determine_first_admin( conn: &mut sqlx::PgConnection,) -> Result<bool, tranquil_db_traits::CreateAccountError> { let map_err = |e: sqlx::Error| tranquil_db_traits::CreateAccountError::Database(e.to_string());
sqlx::query!( "SELECT pg_advisory_xact_lock($1)", FIRST_ADMIN_ADVISORY_LOCK_KEY ) .execute(&mut *conn) .await .map_err(map_err)?;
let count = sqlx::query_scalar!("SELECT COUNT(*) FROM users") .fetch_one(&mut *conn) .await .map_err(map_err)?;
Ok(count.unwrap_or(0) == 0)}
#[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 used_refresh_tokens WHERE session_id IN (SELECT id FROM session_tokens WHERE did = $1)", did.as_str() ) .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 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 .map_err(map_sqlx_error)?; sqlx::query!( "DELETE FROM invite_codes 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 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 = determine_first_admin(&mut tx).await?;
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 = determine_first_admin(&mut tx).await?;
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 = determine_first_admin(&mut tx).await?;
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(()) }}