Something went wrong. Try again.
A downstream fork of Tranquil PDS for use in margin.cafe
Something went wrong. Try again.
39 kB · 1283 lines
Rust
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284use async_trait::async_trait;use chrono::{DateTime, Utc};use sqlx::PgPool;use tranquil_db_traits::{ AdminAccountInfo, CommsChannel, CommsStatus, CommsType, DbError, DeletionRequest, DeletionRequestWithToken, InfraRepository, InviteCodeError, InviteCodeInfo, InviteCodeRow, InviteCodeSortOrder, InviteCodeState, InviteCodeUse, NotificationHistoryRow, PlcTokenInfo, QueuedComms, ReservedSigningKey, ReservedSigningKeyFull, ValidatedInviteCode,};use tranquil_types::{CidLink, Did, InviteCode};use uuid::Uuid;
use super::col;use super::user::map_sqlx_error;use super::{column, legacy_column, opt_column};
pub struct PostgresInfraRepository { pool: PgPool,}
impl PostgresInfraRepository { pub fn new(pool: PgPool) -> Self { Self { pool } }}
#[async_trait]impl InfraRepository for PostgresInfraRepository { async fn enqueue_comms( &self, user_id: Option<Uuid>, channel: CommsChannel, comms_type: CommsType, recipient: &str, subject: Option<&str>, body: &str, metadata: Option<serde_json::Value>, ) -> Result<Uuid, DbError> { let id = sqlx::query_scalar!( r#"INSERT INTO comms_queue (user_id, channel, comms_type, recipient, subject, body, metadata) VALUES ($1, $2, $3, $4, $5, $6, $7) RETURNING id"#, user_id, channel as CommsChannel, comms_type as CommsType, recipient, subject, body, metadata ) .fetch_one(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(id) }
async fn fetch_pending_comms( &self, now: DateTime<Utc>, batch_size: i64, ) -> Result<Vec<QueuedComms>, DbError> { let results = sqlx::query_as!( QueuedComms, r#"UPDATE comms_queue SET status = 'processing', updated_at = NOW() WHERE id IN ( SELECT id FROM comms_queue WHERE attempts < max_attempts AND scheduled_for <= $1 AND ( status = 'pending' OR (status = 'processing' AND updated_at < $1 - INTERVAL '10 minutes') ) ORDER BY scheduled_for ASC LIMIT $2 FOR UPDATE SKIP LOCKED ) RETURNING id, user_id, channel as "channel: CommsChannel", comms_type as "comms_type: CommsType", status as "status: CommsStatus", recipient, subject, body, metadata, attempts, max_attempts, last_error, created_at, updated_at, scheduled_for, processed_at"#, now, batch_size ) .fetch_all(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(results) }
async fn mark_comms_sent(&self, id: Uuid) -> Result<(), DbError> { sqlx::query!( r#"UPDATE comms_queue SET status = 'sent', processed_at = NOW(), updated_at = NOW() WHERE id = $1"#, id ) .execute(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(()) }
async fn mark_comms_failed(&self, id: Uuid, error: &str) -> Result<(), DbError> { sqlx::query!( r#"UPDATE comms_queue SET status = CASE WHEN attempts + 1 >= max_attempts THEN 'failed'::comms_status ELSE 'pending'::comms_status END, attempts = attempts + 1, last_error = $2, updated_at = NOW(), scheduled_for = NOW() + (INTERVAL '1 minute' * (attempts + 1)) WHERE id = $1"#, id, error ) .execute(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(()) }
async fn mark_comms_failed_permanent(&self, id: Uuid, error: &str) -> Result<(), DbError> { sqlx::query!( r#"UPDATE comms_queue SET status = 'failed'::comms_status, attempts = max_attempts, last_error = $2, updated_at = NOW() WHERE id = $1"#, id, error ) .execute(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(()) }
async fn create_invite_code( &self, code: &InviteCode, use_count: i32, for_account: &Did, ) -> Result<bool, DbError> { let for_account_str = for_account.as_str(); let result = sqlx::query!( r#"INSERT INTO invite_codes (code, available_uses, created_by_user, for_account) SELECT $1, $2, id, $3 FROM users WHERE is_admin = true LIMIT 1"#, code.as_str(), use_count, for_account_str ) .execute(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(result.rows_affected() > 0) }
async fn create_invite_codes_batch( &self, codes: &[InviteCode], use_count: i32, created_by_user: Uuid, for_account: &Did, ) -> Result<(), DbError> { let for_account_str = for_account.as_str(); let code_strs: Vec<String> = codes.iter().map(|c| c.to_string()).collect(); sqlx::query!( r#"INSERT INTO invite_codes (code, available_uses, created_by_user, for_account) SELECT code, $2, $3, $4 FROM UNNEST($1::text[]) AS t(code)"#, &code_strs, use_count, created_by_user, for_account_str ) .execute(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(()) }
async fn get_invite_code_available_uses( &self, code: &InviteCode, ) -> Result<Option<i32>, DbError> { let result = sqlx::query_scalar!( "SELECT available_uses FROM invite_codes WHERE code = $1 FOR UPDATE", code.as_str() ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(result) }
async fn validate_invite_code<'a>( &self, code: &'a InviteCode, ) -> Result<ValidatedInviteCode<'a>, InviteCodeError> { let result = sqlx::query!( r#"SELECT available_uses, COALESCE(disabled, false) as "disabled!" FROM invite_codes WHERE code = $1"#, code.as_str() ) .fetch_optional(&self.pool) .await .map_err(|e| InviteCodeError::DatabaseError(map_sqlx_error(e)))?;
match result { None => Err(InviteCodeError::NotFound), Some(row) if row.disabled => Err(InviteCodeError::Disabled), Some(row) if row.available_uses <= 0 => Err(InviteCodeError::ExhaustedUses), Some(_) => Ok(ValidatedInviteCode::new_validated(code)), } }
async fn get_invite_codes_for_account( &self, for_account: &Did, ) -> Result<Vec<InviteCodeInfo>, DbError> { let results = sqlx::query!( r#"SELECT ic.code, ic.available_uses, ic.created_at, ic.disabled, ic.for_account, (SELECT COUNT(*) FROM invite_code_uses icu WHERE icu.code = ic.code)::int as "use_count!" FROM invite_codes ic WHERE ic.for_account = $1 ORDER BY ic.created_at DESC"#, for_account.as_str() ) .fetch_all(&self.pool) .await .map_err(map_sqlx_error)?;
results .into_iter() .map(|r| { Ok(InviteCodeInfo { code: InviteCode::from(r.code), available_uses: r.available_uses, state: InviteCodeState::from_optional_disabled_flag(r.disabled), for_account: legacy_column(r.for_account, col::INVITE_CODES_FOR_ACCOUNT), created_at: r.created_at, created_by: None, }) }) .collect() }
async fn get_invite_code_uses(&self, code: &InviteCode) -> Result<Vec<InviteCodeUse>, DbError> { let results = sqlx::query!( r#"SELECT u.did, u.handle, icu.used_at FROM invite_code_uses icu JOIN users u ON icu.used_by_user = u.id WHERE icu.code = $1 ORDER BY icu.used_at DESC"#, code.as_str() ) .fetch_all(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(results .into_iter() .filter_map(|r| { Some(InviteCodeUse { code: code.clone(), used_by_did: legacy_column(r.did, col::USERS_DID)?, used_by_handle: legacy_column(r.handle, col::USERS_HANDLE), used_at: r.used_at, }) }) .collect()) }
async fn disable_invite_codes_by_code(&self, codes: &[InviteCode]) -> Result<(), DbError> { let code_strs: Vec<String> = codes.iter().map(|c| c.to_string()).collect(); sqlx::query!( "UPDATE invite_codes SET disabled = TRUE WHERE code = ANY($1)", &code_strs ) .execute(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(()) }
async fn disable_invite_codes_by_account(&self, accounts: &[Did]) -> Result<(), DbError> { let accounts_str: Vec<&str> = accounts.iter().map(|d| d.as_str()).collect(); sqlx::query!( r#"UPDATE invite_codes SET disabled = TRUE WHERE created_by_user IN (SELECT id FROM users WHERE did = ANY($1))"#, &accounts_str as &[&str] ) .execute(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(()) }
async fn list_invite_codes( &self, cursor: Option<&str>, limit: i64, sort: InviteCodeSortOrder, ) -> Result<Vec<InviteCodeRow>, DbError> { fn to_row( code: String, available_uses: i32, disabled: Option<bool>, created_by_user: Uuid, created_at: DateTime<Utc>, ) -> InviteCodeRow { InviteCodeRow { code: InviteCode::from(code), available_uses, disabled, created_by_user, created_at, } }
let results = match (cursor, sort) { (Some(cursor_code), InviteCodeSortOrder::Recent) => sqlx::query!( r#"SELECT ic.code, ic.available_uses, ic.disabled, ic.created_by_user, ic.created_at FROM invite_codes ic WHERE ic.created_at < (SELECT created_at FROM invite_codes WHERE code = $1) ORDER BY created_at DESC LIMIT $2"#, cursor_code, limit ) .fetch_all(&self.pool) .await .map_err(map_sqlx_error)? .into_iter() .map(|r| { to_row( r.code, r.available_uses, r.disabled, r.created_by_user, r.created_at, ) }) .collect(), (None, InviteCodeSortOrder::Recent) => sqlx::query!( r#"SELECT ic.code, ic.available_uses, ic.disabled, ic.created_by_user, ic.created_at FROM invite_codes ic ORDER BY created_at DESC LIMIT $1"#, limit ) .fetch_all(&self.pool) .await .map_err(map_sqlx_error)? .into_iter() .map(|r| { to_row( r.code, r.available_uses, r.disabled, r.created_by_user, r.created_at, ) }) .collect(), (Some(cursor_code), InviteCodeSortOrder::Usage) => sqlx::query!( r#"SELECT ic.code, ic.available_uses, ic.disabled, ic.created_by_user, ic.created_at FROM invite_codes ic WHERE ic.created_at < (SELECT created_at FROM invite_codes WHERE code = $1) ORDER BY available_uses DESC LIMIT $2"#, cursor_code, limit ) .fetch_all(&self.pool) .await .map_err(map_sqlx_error)? .into_iter() .map(|r| { to_row( r.code, r.available_uses, r.disabled, r.created_by_user, r.created_at, ) }) .collect(), (None, InviteCodeSortOrder::Usage) => sqlx::query!( r#"SELECT ic.code, ic.available_uses, ic.disabled, ic.created_by_user, ic.created_at FROM invite_codes ic ORDER BY available_uses DESC LIMIT $1"#, limit ) .fetch_all(&self.pool) .await .map_err(map_sqlx_error)? .into_iter() .map(|r| { to_row( r.code, r.available_uses, r.disabled, r.created_by_user, r.created_at, ) }) .collect(), };
Ok(results) }
async fn get_user_dids_by_ids(&self, user_ids: &[Uuid]) -> Result<Vec<(Uuid, Did)>, DbError> { let results = sqlx::query!("SELECT id, did FROM users WHERE id = ANY($1)", user_ids) .fetch_all(&self.pool) .await .map_err(map_sqlx_error)?;
results .into_iter() .map(|r| Ok((r.id, column(r.did, col::USERS_DID)?))) .collect() }
async fn get_invite_code_uses_batch( &self, codes: &[InviteCode], ) -> Result<Vec<InviteCodeUse>, DbError> { let code_strs: Vec<String> = codes.iter().map(|c| c.to_string()).collect(); let results = sqlx::query!( r#"SELECT icu.code, u.did, icu.used_at FROM invite_code_uses icu JOIN users u ON icu.used_by_user = u.id WHERE icu.code = ANY($1) ORDER BY icu.used_at DESC"#, &code_strs ) .fetch_all(&self.pool) .await .map_err(map_sqlx_error)?;
results .into_iter() .map(|r| { Ok(InviteCodeUse { code: InviteCode::from(r.code), used_by_did: column(r.did, col::USERS_DID)?, used_by_handle: None, used_at: r.used_at, }) }) .collect() }
async fn get_invites_created_by_user( &self, user_id: Uuid, ) -> Result<Vec<InviteCodeInfo>, DbError> { let results = sqlx::query!( r#"SELECT ic.code, ic.available_uses, ic.disabled, ic.for_account, ic.created_at, u.did as created_by FROM invite_codes ic JOIN users u ON ic.created_by_user = u.id WHERE ic.created_by_user = $1"#, user_id ) .fetch_all(&self.pool) .await .map_err(map_sqlx_error)?;
results .into_iter() .map(|r| { Ok(InviteCodeInfo { code: InviteCode::from(r.code), available_uses: r.available_uses, state: InviteCodeState::from_optional_disabled_flag(r.disabled), for_account: legacy_column(r.for_account, col::INVITE_CODES_FOR_ACCOUNT), created_at: r.created_at, created_by: Some(column(r.created_by, col::USERS_DID)?), }) }) .collect() }
async fn get_invite_code_info( &self, code: &InviteCode, ) -> Result<Option<InviteCodeInfo>, DbError> { let result = sqlx::query!( r#"SELECT ic.code, ic.available_uses, ic.disabled, ic.for_account, ic.created_at, u.did as created_by FROM invite_codes ic JOIN users u ON ic.created_by_user = u.id WHERE ic.code = $1"#, code.as_str() ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?;
result .map(|r| { Ok(InviteCodeInfo { code: InviteCode::from(r.code), available_uses: r.available_uses, state: InviteCodeState::from_optional_disabled_flag(r.disabled), for_account: legacy_column(r.for_account, col::INVITE_CODES_FOR_ACCOUNT), created_at: r.created_at, created_by: Some(column(r.created_by, col::USERS_DID)?), }) }) .transpose() }
async fn get_invite_codes_by_users( &self, user_ids: &[Uuid], ) -> Result<Vec<(Uuid, InviteCodeInfo)>, DbError> { let results = sqlx::query!( r#"SELECT ic.code, ic.available_uses, ic.disabled, ic.for_account, ic.created_at, ic.created_by_user, u.did as created_by FROM invite_codes ic JOIN users u ON ic.created_by_user = u.id WHERE ic.created_by_user = ANY($1)"#, user_ids ) .fetch_all(&self.pool) .await .map_err(map_sqlx_error)?;
results .into_iter() .map(|r| { Ok(( r.created_by_user, InviteCodeInfo { code: InviteCode::from(r.code), available_uses: r.available_uses, state: InviteCodeState::from_optional_disabled_flag(r.disabled), for_account: legacy_column(r.for_account, col::INVITE_CODES_FOR_ACCOUNT), created_at: r.created_at, created_by: Some(column(r.created_by, col::USERS_DID)?), }, )) }) .collect() }
async fn get_invite_code_used_by_user( &self, user_id: Uuid, ) -> Result<Option<InviteCode>, DbError> { let result = sqlx::query_scalar!( "SELECT code FROM invite_code_uses WHERE used_by_user = $1", user_id ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(result.map(InviteCode::from)) }
async fn delete_invite_code_uses_by_user(&self, user_id: Uuid) -> Result<(), DbError> { sqlx::query!( "DELETE FROM invite_code_uses WHERE used_by_user = $1", user_id ) .execute(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(()) }
async fn delete_invite_codes_by_user(&self, user_id: Uuid) -> Result<(), DbError> { sqlx::query!( "DELETE FROM invite_codes WHERE created_by_user = $1", user_id ) .execute(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(()) }
async fn reserve_signing_key( &self, did: Option<&Did>, public_key_did_key: &Did, private_key_bytes: &[u8], expires_at: DateTime<Utc>, ) -> Result<Uuid, DbError> { let did_str = did.map(|d| d.as_str()); let id = sqlx::query_scalar!( r#"INSERT INTO reserved_signing_keys (did, public_key_did_key, private_key_bytes, expires_at) VALUES ($1, $2, $3, $4) RETURNING id"#, did_str, public_key_did_key.as_str(), private_key_bytes, expires_at ) .fetch_one(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(id) }
async fn get_reserved_signing_key( &self, public_key_did_key: &Did, ) -> Result<Option<ReservedSigningKey>, DbError> { let result = sqlx::query!( r#"SELECT id, private_key_bytes FROM reserved_signing_keys WHERE public_key_did_key = $1 AND used_at IS NULL AND expires_at > NOW() FOR UPDATE"#, public_key_did_key.as_str() ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(result.map(|r| ReservedSigningKey { id: r.id, private_key_bytes: r.private_key_bytes, })) }
async fn mark_signing_key_used(&self, key_id: Uuid) -> Result<(), DbError> { sqlx::query!( "UPDATE reserved_signing_keys SET used_at = NOW() WHERE id = $1", key_id ) .execute(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(()) }
async fn create_deletion_request( &self, token: &str, did: &Did, expires_at: DateTime<Utc>, ) -> Result<(), DbError> { sqlx::query!( "INSERT INTO account_deletion_requests (token, did, expires_at) VALUES ($1, $2, $3)", token, did.as_str(), expires_at ) .execute(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(()) }
async fn get_deletion_request(&self, token: &str) -> Result<Option<DeletionRequest>, DbError> { let result = sqlx::query!( "SELECT did, expires_at FROM account_deletion_requests WHERE token = $1", token ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?;
result .map(|r| { Ok(DeletionRequest { did: column(r.did, col::ACCOUNT_DELETION_REQUESTS_DID)?, expires_at: r.expires_at, }) }) .transpose() }
async fn delete_deletion_request(&self, token: &str) -> Result<(), DbError> { sqlx::query!( "DELETE FROM account_deletion_requests WHERE token = $1", token ) .execute(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(()) }
async fn delete_deletion_requests_by_did(&self, did: &Did) -> Result<(), DbError> { sqlx::query!( "DELETE FROM account_deletion_requests WHERE did = $1", did.as_str() ) .execute(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(()) }
async fn upsert_account_preference( &self, user_id: Uuid, name: &str, value_json: serde_json::Value, ) -> Result<(), DbError> { let mut tx = self.pool.begin().await.map_err(map_sqlx_error)?;
sqlx::query!( r#"DELETE FROM account_preferences WHERE user_id = $1 AND name = $2"#, user_id, name ) .execute(&mut *tx) .await .map_err(map_sqlx_error)?;
sqlx::query!( r#"INSERT INTO account_preferences (user_id, name, value_json) VALUES ($1, $2, $3)"#, user_id, name, value_json ) .execute(&mut *tx) .await .map_err(map_sqlx_error)?;
tx.commit().await.map_err(map_sqlx_error)?;
Ok(()) }
async fn insert_account_preference_if_not_exists( &self, user_id: Uuid, name: &str, value_json: serde_json::Value, ) -> Result<(), DbError> { sqlx::query!( r#"INSERT INTO account_preferences (user_id, name, value_json) VALUES ($1, $2, $3) ON CONFLICT (user_id, name) DO NOTHING"#, user_id, name, value_json ) .execute(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(()) }
async fn get_server_config(&self, key: &str) -> Result<Option<String>, DbError> { let row = sqlx::query_scalar!("SELECT value FROM server_config WHERE key = $1", key) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?; Ok(row) }
async fn health_check(&self) -> Result<bool, DbError> { sqlx::query_scalar!("SELECT 1 as one") .fetch_one(&self.pool) .await .map_err(map_sqlx_error)?; Ok(true) }
async fn insert_report( &self, id: i64, reason_type: &str, reason: Option<&str>, subject_json: serde_json::Value, reported_by_did: &Did, created_at: DateTime<Utc>, ) -> Result<(), DbError> { sqlx::query!( "INSERT INTO reports (id, reason_type, reason, subject_json, reported_by_did, created_at) VALUES ($1, $2, $3, $4, $5, $6)", id, reason_type, reason, subject_json, reported_by_did.as_str(), created_at ) .execute(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(()) }
async fn delete_plc_tokens_for_user(&self, user_id: Uuid) -> Result<(), DbError> { sqlx::query!( "DELETE FROM plc_operation_tokens WHERE user_id = $1 OR expires_at < NOW()", user_id ) .execute(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(()) }
async fn insert_plc_token( &self, user_id: Uuid, token: &str, expires_at: DateTime<Utc>, ) -> Result<(), DbError> { sqlx::query!( "INSERT INTO plc_operation_tokens (user_id, token, expires_at) VALUES ($1, $2, $3)", user_id, token, expires_at ) .execute(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(()) }
async fn get_plc_token_expiry( &self, user_id: Uuid, token: &str, ) -> Result<Option<DateTime<Utc>>, DbError> { let expiry = sqlx::query_scalar!( "SELECT expires_at FROM plc_operation_tokens WHERE user_id = $1 AND token = $2", user_id, token ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(expiry) }
async fn delete_plc_token(&self, user_id: Uuid, token: &str) -> Result<(), DbError> { sqlx::query!( "DELETE FROM plc_operation_tokens WHERE user_id = $1 AND token = $2", user_id, token ) .execute(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(()) }
async fn get_account_preferences( &self, user_id: Uuid, ) -> Result<Vec<(String, serde_json::Value)>, DbError> { let rows = sqlx::query!( "SELECT name, value_json FROM account_preferences WHERE user_id = $1", user_id ) .fetch_all(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(rows.into_iter().map(|r| (r.name, r.value_json)).collect()) }
async fn replace_namespace_preferences( &self, user_id: Uuid, namespace: &str, preferences: Vec<(String, serde_json::Value)>, ) -> Result<(), DbError> { let mut tx = self.pool.begin().await.map_err(map_sqlx_error)?;
let like_pattern = format!("{}.%", namespace); sqlx::query!( "DELETE FROM account_preferences WHERE user_id = $1 AND (name = $2 OR name LIKE $3)", user_id, namespace, like_pattern ) .execute(&mut *tx) .await .map_err(map_sqlx_error)?;
for (name, value_json) in preferences { sqlx::query!( "INSERT INTO account_preferences (user_id, name, value_json) VALUES ($1, $2, $3)", user_id, name, value_json ) .execute(&mut *tx) .await .map_err(map_sqlx_error)?; }
tx.commit().await.map_err(map_sqlx_error)?;
Ok(()) }
async fn get_notification_history( &self, user_id: Uuid, limit: i64, ) -> Result<Vec<NotificationHistoryRow>, DbError> { let rows = sqlx::query!( r#" SELECT created_at, channel as "channel: CommsChannel", comms_type as "comms_type: CommsType", status as "status: CommsStatus", subject, body FROM comms_queue WHERE user_id = $1 ORDER BY created_at DESC LIMIT $2 "#, user_id, limit ) .fetch_all(&self.pool) .await .map_err(map_sqlx_error)?; Ok(rows .into_iter() .map(|r| NotificationHistoryRow { created_at: r.created_at, channel: r.channel, comms_type: r.comms_type, status: r.status, subject: r.subject, body: r.body, }) .collect()) }
async fn get_server_configs(&self, keys: &[&str]) -> Result<Vec<(String, String)>, DbError> { let keys_vec: Vec<String> = keys.iter().map(|s| s.to_string()).collect(); let rows: Vec<(String, String)> = sqlx::query_as("SELECT key, value FROM server_config WHERE key = ANY($1)") .bind(&keys_vec) .fetch_all(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(rows) }
async fn upsert_server_config(&self, key: &str, value: &str) -> Result<(), DbError> { sqlx::query( "INSERT INTO server_config (key, value, updated_at) VALUES ($1, $2, NOW()) ON CONFLICT (key) DO UPDATE SET value = $2, updated_at = NOW()", ) .bind(key) .bind(value) .execute(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(()) }
async fn delete_server_config(&self, key: &str) -> Result<(), DbError> { sqlx::query("DELETE FROM server_config WHERE key = $1") .bind(key) .execute(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(()) }
async fn get_blob_storage_key_by_cid(&self, cid: &CidLink) -> Result<Option<String>, DbError> { let result = sqlx::query_scalar!("SELECT storage_key FROM blobs WHERE cid = $1", cid.as_str()) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(result) }
async fn delete_blob_by_cid(&self, cid: &CidLink) -> Result<(), DbError> { sqlx::query!("DELETE FROM blobs WHERE cid = $1", cid.as_str()) .execute(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(()) }
async fn get_admin_account_info_by_did( &self, did: &Did, ) -> Result<Option<AdminAccountInfo>, DbError> { let result = sqlx::query!( r#" SELECT id, did, handle, email, created_at, invites_disabled, email_verified, deactivated_at FROM users WHERE did = $1 "#, did.as_str() ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?;
result .map(|r| { Ok(AdminAccountInfo { 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, invites_disabled: r.invites_disabled.unwrap_or(false), email_verified: r.email_verified, deactivated_at: r.deactivated_at, }) }) .transpose() }
async fn get_admin_account_infos_by_dids( &self, dids: &[Did], ) -> Result<Vec<AdminAccountInfo>, DbError> { let dids_str: Vec<&str> = dids.iter().map(|d| d.as_str()).collect(); let results = sqlx::query!( r#" SELECT id, did, handle, email, created_at, invites_disabled, email_verified, deactivated_at FROM users WHERE did = ANY($1) "#, &dids_str as &[&str] ) .fetch_all(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(results .into_iter() .filter_map(|r| { Some(AdminAccountInfo { id: r.id, did: legacy_column(r.did, col::USERS_DID)?, handle: legacy_column(r.handle, col::USERS_HANDLE)?, email: r.email, created_at: r.created_at, invites_disabled: r.invites_disabled.unwrap_or(false), email_verified: r.email_verified, deactivated_at: r.deactivated_at, }) }) .collect()) }
async fn get_invite_code_uses_by_users( &self, user_ids: &[Uuid], ) -> Result<Vec<(Uuid, InviteCode)>, DbError> { let results = sqlx::query!( r#" SELECT used_by_user, code FROM invite_code_uses WHERE used_by_user = ANY($1) "#, user_ids ) .fetch_all(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(results .into_iter() .map(|r| (r.used_by_user, InviteCode::from(r.code))) .collect()) }
async fn get_deletion_request_by_did( &self, did: &Did, ) -> Result<Option<DeletionRequestWithToken>, DbError> { let row = sqlx::query!( r#"SELECT token, did, expires_at FROM account_deletion_requests WHERE did = $1"#, did.as_str() ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?;
row.map(|r| { Ok(DeletionRequestWithToken { token: r.token, did: column(r.did, col::ACCOUNT_DELETION_REQUESTS_DID)?, expires_at: r.expires_at, }) }) .transpose() }
async fn get_latest_comms_for_user( &self, user_id: Uuid, comms_type: CommsType, limit: i64, ) -> Result<Vec<QueuedComms>, DbError> { let results = sqlx::query_as!( QueuedComms, r#"SELECT id, user_id, channel as "channel: CommsChannel", comms_type as "comms_type: CommsType", status as "status: CommsStatus", recipient, subject, body, metadata, attempts, max_attempts, last_error, created_at, updated_at, scheduled_for, processed_at FROM comms_queue WHERE user_id = $1 AND comms_type = $2 ORDER BY created_at DESC LIMIT $3"#, user_id, comms_type as CommsType, limit ) .fetch_all(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(results) }
async fn count_comms_by_type( &self, user_id: Uuid, comms_type: CommsType, ) -> Result<i64, DbError> { let count = sqlx::query_scalar!( r#"SELECT COUNT(*) as "count!" FROM comms_queue WHERE user_id = $1 AND comms_type = $2"#, user_id, comms_type as CommsType ) .fetch_one(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(count) }
async fn delete_comms_by_type_for_user( &self, user_id: Uuid, comms_type: CommsType, ) -> Result<u64, DbError> { let result = sqlx::query!( "DELETE FROM comms_queue WHERE user_id = $1 AND comms_type = $2", user_id, comms_type as CommsType ) .execute(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(result.rows_affected()) }
async fn expire_deletion_request(&self, token: &str) -> Result<(), DbError> { sqlx::query!( "UPDATE account_deletion_requests SET expires_at = NOW() - INTERVAL '1 hour' WHERE token = $1", token ) .execute(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(()) }
async fn get_reserved_signing_key_full( &self, public_key_did_key: &Did, ) -> Result<Option<ReservedSigningKeyFull>, DbError> { let row = sqlx::query!( r#"SELECT id, did, public_key_did_key, private_key_bytes, expires_at, used_at FROM reserved_signing_keys WHERE public_key_did_key = $1"#, public_key_did_key.as_str() ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?;
row.map(|r| { Ok(ReservedSigningKeyFull { id: r.id, did: opt_column(r.did, col::RESERVED_SIGNING_KEYS_DID)?, public_key_did_key: column( r.public_key_did_key, col::RESERVED_SIGNING_KEYS_PUBLIC_KEY_DID_KEY, )?, private_key_bytes: r.private_key_bytes, expires_at: r.expires_at, used_at: r.used_at, }) }) .transpose() }
async fn get_plc_tokens_by_did(&self, did: &Did) -> Result<Vec<PlcTokenInfo>, DbError> { let results = sqlx::query!( r#"SELECT t.token, t.expires_at FROM plc_operation_tokens t JOIN users u ON t.user_id = u.id WHERE u.did = $1"#, did.as_str() ) .fetch_all(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(results .into_iter() .map(|r| PlcTokenInfo { token: r.token, expires_at: r.expires_at, }) .collect()) }
async fn count_plc_tokens_by_did(&self, did: &Did) -> Result<i64, DbError> { let count = sqlx::query_scalar!( r#"SELECT COUNT(*) as "count!" FROM plc_operation_tokens t JOIN users u ON t.user_id = u.id WHERE u.did = $1"#, did.as_str() ) .fetch_one(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(count) }}