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