Something went wrong. Try again.
A downstream fork of Tranquil PDS for use in margin.cafe
Something went wrong. Try again.
13 kB · 397 lines
Rust
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398use async_trait::async_trait;use sqlx::PgPool;use tranquil_db_traits::{ AuditLogEntry, ControllerInfo, DbError, DbScope, DelegatedAccountInfo, DelegationActionType, DelegationGrant, DelegationRepository,};use tranquil_types::Did;use uuid::Uuid;
use super::col;use super::user::map_sqlx_error;use super::{column, legacy_column, opt_column};
#[derive(Debug, Clone, Copy, PartialEq, Eq, sqlx::Type)]#[sqlx(type_name = "delegation_action_type", rename_all = "snake_case")]pub enum PgDelegationActionType { GrantCreated, GrantRevoked, ScopesModified, TokenIssued, RepoWrite, BlobUpload, AccountAction,}
impl From<DelegationActionType> for PgDelegationActionType { fn from(t: DelegationActionType) -> Self { match t { DelegationActionType::GrantCreated => Self::GrantCreated, DelegationActionType::GrantRevoked => Self::GrantRevoked, DelegationActionType::ScopesModified => Self::ScopesModified, DelegationActionType::TokenIssued => Self::TokenIssued, DelegationActionType::RepoWrite => Self::RepoWrite, DelegationActionType::BlobUpload => Self::BlobUpload, DelegationActionType::AccountAction => Self::AccountAction, } }}
impl From<PgDelegationActionType> for DelegationActionType { fn from(t: PgDelegationActionType) -> Self { match t { PgDelegationActionType::GrantCreated => Self::GrantCreated, PgDelegationActionType::GrantRevoked => Self::GrantRevoked, PgDelegationActionType::ScopesModified => Self::ScopesModified, PgDelegationActionType::TokenIssued => Self::TokenIssued, PgDelegationActionType::RepoWrite => Self::RepoWrite, PgDelegationActionType::BlobUpload => Self::BlobUpload, PgDelegationActionType::AccountAction => Self::AccountAction, } }}
pub struct PostgresDelegationRepository { pool: PgPool,}
impl PostgresDelegationRepository { pub fn new(pool: PgPool) -> Self { Self { pool } }}
#[async_trait]impl DelegationRepository for PostgresDelegationRepository { async fn is_delegated_account(&self, did: &Did) -> Result<bool, DbError> { let exists = sqlx::query_scalar!( r#"SELECT EXISTS( SELECT 1 FROM account_delegations WHERE delegated_did = $1 AND revoked_at IS NULL ) as "exists!""#, did.as_str() ) .fetch_one(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(exists) }
async fn create_delegation( &self, delegated_did: &Did, controller_did: &Did, granted_scopes: &DbScope, granted_by: &Did, ) -> Result<Uuid, DbError> { let id = sqlx::query_scalar!( r#" INSERT INTO account_delegations (delegated_did, controller_did, granted_scopes, granted_by) VALUES ($1, $2, $3, $4) RETURNING id "#, delegated_did.as_str(), controller_did.as_str(), granted_scopes.as_str(), granted_by.as_str() ) .fetch_one(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(id) }
async fn revoke_delegation( &self, delegated_did: &Did, controller_did: &Did, revoked_by: &Did, ) -> Result<bool, DbError> { let result = sqlx::query!( r#" UPDATE account_delegations SET revoked_at = NOW(), revoked_by = $1 WHERE delegated_did = $2 AND controller_did = $3 AND revoked_at IS NULL "#, revoked_by.as_str(), delegated_did.as_str(), controller_did.as_str() ) .execute(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(result.rows_affected() > 0) }
async fn update_delegation_scopes( &self, delegated_did: &Did, controller_did: &Did, new_scopes: &DbScope, ) -> Result<bool, DbError> { let result = sqlx::query!( r#" UPDATE account_delegations SET granted_scopes = $1 WHERE delegated_did = $2 AND controller_did = $3 AND revoked_at IS NULL "#, new_scopes.as_str(), delegated_did.as_str(), controller_did.as_str() ) .execute(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(result.rows_affected() > 0) }
async fn get_delegation( &self, delegated_did: &Did, controller_did: &Did, ) -> Result<Option<DelegationGrant>, DbError> { let row = sqlx::query!( r#" SELECT id, delegated_did, controller_did, granted_scopes, granted_at, granted_by, revoked_at, revoked_by FROM account_delegations WHERE delegated_did = $1 AND controller_did = $2 AND revoked_at IS NULL "#, delegated_did.as_str(), controller_did.as_str() ) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?;
row.map(|r| { Ok(DelegationGrant { id: r.id, delegated_did: column(r.delegated_did, col::ACCOUNT_DELEGATIONS_DELEGATED_DID)?, controller_did: column(r.controller_did, col::ACCOUNT_DELEGATIONS_CONTROLLER_DID)?, granted_scopes: DbScope::from_db(r.granted_scopes), granted_at: r.granted_at, granted_by: column(r.granted_by, col::ACCOUNT_DELEGATIONS_GRANTED_BY)?, revoked_at: r.revoked_at, revoked_by: opt_column(r.revoked_by, col::ACCOUNT_DELEGATIONS_REVOKED_BY)?, }) }) .transpose() }
async fn get_delegations_for_account( &self, delegated_did: &Did, ) -> Result<Vec<ControllerInfo>, DbError> { let rows = sqlx::query!( r#" SELECT d.controller_did, u.handle as "handle?", d.granted_scopes, d.granted_at, CASE WHEN u.did IS NOT NULL THEN u.deactivated_at IS NULL AND u.takedown_ref IS NULL ELSE true END as "is_active!", u.did IS NOT NULL as "is_local!" FROM account_delegations d LEFT JOIN users u ON u.did = d.controller_did WHERE d.delegated_did = $1 AND d.revoked_at IS NULL ORDER BY d.granted_at DESC "#, delegated_did.as_str() ) .fetch_all(&self.pool) .await .map_err(map_sqlx_error)?;
rows.into_iter() .map(|r| { Ok(ControllerInfo { did: column(r.controller_did, col::ACCOUNT_DELEGATIONS_CONTROLLER_DID)?, handle: r.handle.and_then(|h| legacy_column(h, col::USERS_HANDLE)), granted_scopes: DbScope::from_db(r.granted_scopes), granted_at: r.granted_at, is_active: r.is_active, is_local: r.is_local, }) }) .collect() }
async fn get_accounts_controlled_by( &self, controller_did: &Did, ) -> Result<Vec<DelegatedAccountInfo>, DbError> { let rows = sqlx::query!( r#" SELECT u.did, u.handle, d.granted_scopes, d.granted_at FROM account_delegations d JOIN users u ON u.did = d.delegated_did WHERE d.controller_did = $1 AND d.revoked_at IS NULL AND u.deactivated_at IS NULL AND u.takedown_ref IS NULL ORDER BY d.granted_at DESC "#, controller_did.as_str() ) .fetch_all(&self.pool) .await .map_err(map_sqlx_error)?;
rows.into_iter() .map(|r| { Ok(DelegatedAccountInfo { did: column(r.did, col::USERS_DID)?, handle: legacy_column(r.handle, col::USERS_HANDLE), granted_scopes: DbScope::from_db(r.granted_scopes), granted_at: r.granted_at, }) }) .collect() }
async fn count_active_controllers(&self, delegated_did: &Did) -> Result<i64, DbError> { let count = sqlx::query_scalar!( r#" SELECT COUNT(*) as "count!" FROM account_delegations d LEFT JOIN users u ON u.did = d.controller_did WHERE d.delegated_did = $1 AND d.revoked_at IS NULL AND (u.did IS NULL OR (u.deactivated_at IS NULL AND u.takedown_ref IS NULL)) "#, delegated_did.as_str() ) .fetch_one(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(count) }
async fn controls_any_accounts(&self, did: &Did) -> Result<bool, DbError> { let exists = sqlx::query_scalar!( r#"SELECT EXISTS( SELECT 1 FROM account_delegations WHERE controller_did = $1 AND revoked_at IS NULL ) as "exists!""#, did.as_str() ) .fetch_one(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(exists) }
async fn log_delegation_action( &self, delegated_did: &Did, actor_did: &Did, controller_did: Option<&Did>, action_type: DelegationActionType, action_details: Option<serde_json::Value>, ip_address: Option<&str>, user_agent: Option<&str>, ) -> Result<Uuid, DbError> { let pg_action_type: PgDelegationActionType = action_type.into(); let controller_did_str = controller_did.map(|d| d.as_str()); let id = sqlx::query_scalar!( r#" INSERT INTO delegation_audit_log (delegated_did, actor_did, controller_did, action_type, action_details, ip_address, user_agent) VALUES ($1, $2, $3, $4, $5, $6, $7) RETURNING id "#, delegated_did.as_str(), actor_did.as_str(), controller_did_str, pg_action_type as PgDelegationActionType, action_details, ip_address, user_agent ) .fetch_one(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(id) }
async fn get_audit_log_for_account( &self, delegated_did: &Did, limit: i64, offset: i64, ) -> Result<Vec<AuditLogEntry>, DbError> { let rows = sqlx::query!( r#" SELECT id, delegated_did, actor_did, controller_did, action_type as "action_type: PgDelegationActionType", action_details, ip_address, user_agent, created_at FROM delegation_audit_log WHERE delegated_did = $1 ORDER BY created_at DESC LIMIT $2 OFFSET $3 "#, delegated_did.as_str(), limit, offset ) .fetch_all(&self.pool) .await .map_err(map_sqlx_error)?;
rows.into_iter() .map(|r| { Ok(AuditLogEntry { id: r.id, delegated_did: column( r.delegated_did, col::DELEGATION_AUDIT_LOG_DELEGATED_DID, )?, actor_did: column(r.actor_did, col::DELEGATION_AUDIT_LOG_ACTOR_DID)?, controller_did: opt_column( r.controller_did, col::DELEGATION_AUDIT_LOG_CONTROLLER_DID, )?, action_type: r.action_type.into(), action_details: r.action_details, ip_address: r.ip_address, user_agent: r.user_agent, created_at: r.created_at, }) }) .collect() }
async fn count_audit_log_entries(&self, delegated_did: &Did) -> Result<i64, DbError> { let count = sqlx::query_scalar!( r#"SELECT COUNT(*) as "count!" FROM delegation_audit_log WHERE delegated_did = $1"#, delegated_did.as_str() ) .fetch_one(&self.pool) .await .map_err(map_sqlx_error)?;
Ok(count) }}