tranquil-db crates, repository pattern for db access

Cargo.lock

@@ -6375,6 +6375,7 @@ dependencies = [
  "sqlx",
  "thiserror 2.0.17",
  "tokio",
+ "tranquil-db-traits",
  "urlencoding",
  "uuid",
 ]
@@ -6396,6 +6397,40 @@ dependencies = [
  "thiserror 2.0.17",
 ]
 
+[[package]]
+name = "tranquil-db"
+version = "0.1.0"
+dependencies = [
+ "async-trait",
+ "chrono",
+ "rand 0.8.5",
+ "serde",
+ "serde_json",
+ "sqlx",
+ "thiserror 2.0.17",
+ "tracing",
+ "tranquil-db-traits",
+ "tranquil-oauth",
+ "tranquil-types",
+ "uuid",
+]
+
+[[package]]
+name = "tranquil-db-traits"
+version = "0.1.0"
+dependencies = [
+ "async-trait",
+ "base64 0.22.1",
+ "chrono",
+ "serde",
+ "serde_json",
+ "sqlx",
+ "thiserror 2.0.17",
+ "tranquil-oauth",
+ "tranquil-types",
+ "uuid",
+]
+
 [[package]]
 name = "tranquil-infra"
 version = "0.1.0"
@@ -6503,6 +6538,8 @@ dependencies = [
  "tranquil-cache",
  "tranquil-comms",
  "tranquil-crypto",
+ "tranquil-db",
+ "tranquil-db-traits",
  "tranquil-infra",
  "tranquil-oauth",
  "tranquil-repo",

Cargo.toml

@@ -11,6 +11,8 @@ members = [
     "crates/tranquil-auth",
     "crates/tranquil-oauth",
     "crates/tranquil-comms",
+    "crates/tranquil-db-traits",
+    "crates/tranquil-db",
     "crates/tranquil-pds",
 ]
 
@@ -30,6 +32,8 @@ tranquil-scopes = { path = "crates/tranquil-scopes" }
 tranquil-auth = { path = "crates/tranquil-auth" }
 tranquil-oauth = { path = "crates/tranquil-oauth" }
 tranquil-comms = { path = "crates/tranquil-comms" }
+tranquil-db-traits = { path = "crates/tranquil-db-traits" }
+tranquil-db = { path = "crates/tranquil-db" }
 
 aes-gcm = "0.10"
 backon = "1"
@@ -92,7 +96,7 @@ tower-layer = "0.3"
 tracing = "0.1"
 tracing-subscriber = "0.3"
 urlencoding = "2.1"
-uuid = { version = "1.19", features = ["v4", "v5", "v7", "fast-rng"] }
+uuid = { version = "1.19", features = ["v4", "v5", "v7", "fast-rng", "serde"] }
 webauthn-rs = { version = "0.5", features = ["danger-allow-state-serialisation", "danger-user-presence-only-security-keys"] }
 webauthn-rs-proto = "0.5"
 zip = { version = "7.0", default-features = false, features = ["deflate"] }

crates/tranquil-db-traits/Cargo.toml

@@ -0,0 +1,17 @@
+[package]
+name = "tranquil-db-traits"
+version.workspace = true
+edition.workspace = true
+license.workspace = true
+
+[dependencies]
+async-trait = { workspace = true }
+base64 = { workspace = true }
+chrono = { workspace = true }
+serde = { workspace = true }
+serde_json = { workspace = true }
+sqlx = { workspace = true }
+thiserror = { workspace = true }
+uuid = { workspace = true }
+tranquil-oauth = { workspace = true }
+tranquil-types = { workspace = true }

crates/tranquil-db-traits/src/backlink.rs

@@ -0,0 +1,28 @@
+use async_trait::async_trait;
+use tranquil_types::{AtUri, Nsid};
+use uuid::Uuid;
+
+use crate::DbError;
+
+#[derive(Debug, Clone)]
+pub struct Backlink {
+    pub uri: AtUri,
+    pub path: String,
+    pub link_to: String,
+}
+
+#[async_trait]
+pub trait BacklinkRepository: Send + Sync {
+    async fn get_backlink_conflicts(
+        &self,
+        repo_id: Uuid,
+        collection: &Nsid,
+        backlinks: &[Backlink],
+    ) -> Result<Vec<AtUri>, DbError>;
+
+    async fn add_backlinks(&self, repo_id: Uuid, backlinks: &[Backlink]) -> Result<(), DbError>;
+
+    async fn remove_backlinks_by_uri(&self, uri: &AtUri) -> Result<(), DbError>;
+
+    async fn remove_backlinks_by_repo(&self, repo_id: Uuid) -> Result<(), DbError>;
+}

crates/tranquil-db-traits/src/backup.rs

@@ -0,0 +1,109 @@
+use async_trait::async_trait;
+use chrono::{DateTime, Utc};
+use tranquil_types::Did;
+use uuid::Uuid;
+
+use crate::DbError;
+
+#[derive(Debug, Clone)]
+pub struct BackupRow {
+    pub id: Uuid,
+    pub repo_rev: String,
+    pub repo_root_cid: String,
+    pub block_count: i32,
+    pub size_bytes: i64,
+    pub created_at: DateTime<Utc>,
+}
+
+#[derive(Debug, Clone)]
+pub struct BackupStorageInfo {
+    pub storage_key: String,
+    pub repo_rev: String,
+}
+
+#[derive(Debug, Clone)]
+pub struct BackupForDeletion {
+    pub id: Uuid,
+    pub storage_key: String,
+    pub deactivated_at: Option<DateTime<Utc>>,
+}
+
+#[derive(Debug, Clone)]
+pub struct OldBackupInfo {
+    pub id: Uuid,
+    pub storage_key: String,
+}
+
+#[derive(Debug, Clone)]
+pub struct UserBackupInfo {
+    pub id: Uuid,
+    pub did: Did,
+    pub backup_enabled: bool,
+    pub deactivated_at: Option<DateTime<Utc>>,
+    pub repo_root_cid: String,
+    pub repo_rev: Option<String>,
+}
+
+#[derive(Debug, Clone)]
+pub struct BlobExportInfo {
+    pub cid: String,
+    pub storage_key: String,
+    pub mime_type: String,
+}
+
+#[async_trait]
+pub trait BackupRepository: Send + Sync {
+    async fn get_user_backup_status(
+        &self,
+        did: &Did,
+    ) -> Result<Option<(Uuid, bool)>, DbError>;
+
+    async fn list_backups_for_user(&self, user_id: Uuid) -> Result<Vec<BackupRow>, DbError>;
+
+    async fn get_backup_storage_info(
+        &self,
+        backup_id: Uuid,
+        did: &Did,
+    ) -> Result<Option<BackupStorageInfo>, DbError>;
+
+    async fn get_user_for_backup(&self, did: &Did) -> Result<Option<UserBackupInfo>, DbError>;
+
+    async fn insert_backup(
+        &self,
+        user_id: Uuid,
+        storage_key: &str,
+        repo_root_cid: &str,
+        repo_rev: &str,
+        block_count: i32,
+        size_bytes: i64,
+    ) -> Result<Uuid, DbError>;
+
+    async fn get_old_backups(
+        &self,
+        user_id: Uuid,
+        retention_offset: i64,
+    ) -> Result<Vec<OldBackupInfo>, DbError>;
+
+    async fn delete_backup(&self, backup_id: Uuid) -> Result<(), DbError>;
+
+    async fn get_backup_for_deletion(
+        &self,
+        backup_id: Uuid,
+        did: &Did,
+    ) -> Result<Option<BackupForDeletion>, DbError>;
+
+    async fn get_user_deactivated_status(&self, did: &Did)
+        -> Result<Option<Option<DateTime<Utc>>>, DbError>;
+
+    async fn update_backup_enabled(&self, did: &Did, enabled: bool) -> Result<(), DbError>;
+
+    async fn get_user_id_by_did(&self, did: &Did) -> Result<Option<Uuid>, DbError>;
+
+    async fn get_blobs_for_export(&self, user_id: Uuid) -> Result<Vec<BlobExportInfo>, DbError>;
+
+    async fn get_users_needing_backup(
+        &self,
+        backup_interval_secs: i64,
+        limit: i64,
+    ) -> Result<Vec<UserBackupInfo>, DbError>;
+}

crates/tranquil-db-traits/src/blob.rs

@@ -0,0 +1,100 @@
+use async_trait::async_trait;
+use serde::{Deserialize, Serialize};
+use tranquil_types::{AtUri, CidLink, Did};
+use uuid::Uuid;
+
+use crate::DbError;
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct BlobMetadata {
+    pub storage_key: String,
+    pub mime_type: String,
+    pub size_bytes: i64,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct BlobWithTakedown {
+    pub cid: CidLink,
+    pub takedown_ref: Option<String>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct BlobForExport {
+    pub cid: CidLink,
+    pub storage_key: String,
+    pub mime_type: String,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct MissingBlobInfo {
+    pub blob_cid: CidLink,
+    pub record_uri: AtUri,
+}
+
+#[async_trait]
+pub trait BlobRepository: Send + Sync {
+    async fn insert_blob(
+        &self,
+        cid: &CidLink,
+        mime_type: &str,
+        size_bytes: i64,
+        created_by_user: Uuid,
+        storage_key: &str,
+    ) -> Result<Option<CidLink>, DbError>;
+
+    async fn get_blob_metadata(&self, cid: &CidLink) -> Result<Option<BlobMetadata>, DbError>;
+
+    async fn get_blob_with_takedown(
+        &self,
+        cid: &CidLink,
+    ) -> Result<Option<BlobWithTakedown>, DbError>;
+
+    async fn get_blob_storage_key(&self, cid: &CidLink) -> Result<Option<String>, DbError>;
+
+    async fn list_blobs_by_user(
+        &self,
+        user_id: Uuid,
+        cursor: Option<&str>,
+        limit: i64,
+    ) -> Result<Vec<CidLink>, DbError>;
+
+    async fn list_blobs_since_rev(
+        &self,
+        did: &Did,
+        since: &str,
+    ) -> Result<Vec<CidLink>, DbError>;
+
+    async fn count_blobs_by_user(&self, user_id: Uuid) -> Result<i64, DbError>;
+
+    async fn sum_blob_storage(&self) -> Result<i64, DbError>;
+
+    async fn update_blob_takedown(
+        &self,
+        cid: &CidLink,
+        takedown_ref: Option<&str>,
+    ) -> Result<bool, DbError>;
+
+    async fn delete_blob_by_cid(&self, cid: &CidLink) -> Result<bool, DbError>;
+
+    async fn delete_blobs_by_user(&self, user_id: Uuid) -> Result<u64, DbError>;
+
+    async fn get_blob_storage_keys_by_user(&self, user_id: Uuid) -> Result<Vec<String>, DbError>;
+
+    async fn insert_record_blobs(
+        &self,
+        repo_id: Uuid,
+        record_uris: &[AtUri],
+        blob_cids: &[CidLink],
+    ) -> Result<(), DbError>;
+
+    async fn list_missing_blobs(
+        &self,
+        repo_id: Uuid,
+        cursor: Option<&str>,
+        limit: i64,
+    ) -> Result<Vec<MissingBlobInfo>, DbError>;
+
+    async fn count_distinct_record_blobs(&self, repo_id: Uuid) -> Result<i64, DbError>;
+
+    async fn get_blobs_for_export(&self, repo_id: Uuid) -> Result<Vec<BlobForExport>, DbError>;
+}

crates/tranquil-db-traits/src/delegation.rs

@@ -0,0 +1,141 @@
+use async_trait::async_trait;
+use chrono::{DateTime, Utc};
+use serde::{Deserialize, Serialize};
+use tranquil_types::{Did, Handle};
+use uuid::Uuid;
+
+use crate::DbError;
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct DelegationGrant {
+    pub id: Uuid,
+    pub delegated_did: Did,
+    pub controller_did: Did,
+    pub granted_scopes: String,
+    pub granted_at: DateTime<Utc>,
+    pub granted_by: Did,
+    pub revoked_at: Option<DateTime<Utc>>,
+    pub revoked_by: Option<Did>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct DelegatedAccountInfo {
+    pub did: Did,
+    pub handle: Handle,
+    pub granted_scopes: String,
+    pub granted_at: DateTime<Utc>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ControllerInfo {
+    pub did: Did,
+    pub handle: Handle,
+    pub granted_scopes: String,
+    pub granted_at: DateTime<Utc>,
+    pub is_active: bool,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+pub enum DelegationActionType {
+    GrantCreated,
+    GrantRevoked,
+    ScopesModified,
+    TokenIssued,
+    RepoWrite,
+    BlobUpload,
+    AccountAction,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct AuditLogEntry {
+    pub id: Uuid,
+    pub delegated_did: Did,
+    pub actor_did: Did,
+    pub controller_did: Option<Did>,
+    pub action_type: DelegationActionType,
+    pub action_details: Option<serde_json::Value>,
+    pub ip_address: Option<String>,
+    pub user_agent: Option<String>,
+    pub created_at: DateTime<Utc>,
+}
+
+#[async_trait]
+pub trait DelegationRepository: Send + Sync {
+    async fn is_delegated_account(&self, did: &Did) -> Result<bool, DbError>;
+
+    async fn create_delegation(
+        &self,
+        delegated_did: &Did,
+        controller_did: &Did,
+        granted_scopes: &str,
+        granted_by: &Did,
+    ) -> Result<Uuid, DbError>;
+
+    async fn revoke_delegation(
+        &self,
+        delegated_did: &Did,
+        controller_did: &Did,
+        revoked_by: &Did,
+    ) -> Result<bool, DbError>;
+
+    async fn update_delegation_scopes(
+        &self,
+        delegated_did: &Did,
+        controller_did: &Did,
+        new_scopes: &str,
+    ) -> Result<bool, DbError>;
+
+    async fn get_delegation(
+        &self,
+        delegated_did: &Did,
+        controller_did: &Did,
+    ) -> Result<Option<DelegationGrant>, DbError>;
+
+    async fn get_delegations_for_account(
+        &self,
+        delegated_did: &Did,
+    ) -> Result<Vec<ControllerInfo>, DbError>;
+
+    async fn get_accounts_controlled_by(
+        &self,
+        controller_did: &Did,
+    ) -> Result<Vec<DelegatedAccountInfo>, DbError>;
+
+    async fn get_active_controllers_for_account(
+        &self,
+        delegated_did: &Did,
+    ) -> Result<Vec<ControllerInfo>, DbError>;
+
+    async fn count_active_controllers(&self, delegated_did: &Did) -> Result<i64, DbError>;
+
+    async fn has_any_controllers(&self, did: &Did) -> Result<bool, DbError>;
+
+    async fn controls_any_accounts(&self, did: &Did) -> Result<bool, DbError>;
+
+    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>;
+
+    async fn get_audit_log_for_account(
+        &self,
+        delegated_did: &Did,
+        limit: i64,
+        offset: i64,
+    ) -> Result<Vec<AuditLogEntry>, DbError>;
+
+    async fn get_audit_log_by_controller(
+        &self,
+        controller_did: &Did,
+        limit: i64,
+        offset: i64,
+    ) -> Result<Vec<AuditLogEntry>, DbError>;
+
+    async fn count_audit_log_entries(&self, delegated_did: &Did) -> Result<i64, DbError>;
+}

crates/tranquil-db-traits/src/error.rs

@@ -0,0 +1,39 @@
+use thiserror::Error;
+
+#[derive(Debug, Error)]
+pub enum DbError {
+    #[error("Database query error: {0}")]
+    Query(String),
+
+    #[error("Record not found")]
+    NotFound,
+
+    #[error("Constraint violation: {0}")]
+    Constraint(String),
+
+    #[error("Connection error: {0}")]
+    Connection(String),
+
+    #[error("Transaction error: {0}")]
+    Transaction(String),
+
+    #[error("Serialization error: {0}")]
+    Serialization(String),
+
+    #[error("Other database error: {0}")]
+    Other(String),
+}
+
+impl DbError {
+    pub fn from_query_error(msg: impl Into<String>) -> Self {
+        DbError::Query(msg.into())
+    }
+
+    pub fn from_constraint_error(msg: impl Into<String>) -> Self {
+        DbError::Constraint(msg.into())
+    }
+
+    pub fn from_connection_error(msg: impl Into<String>) -> Self {
+        DbError::Connection(msg.into())
+    }
+}

crates/tranquil-db-traits/src/infra.rs

@@ -0,0 +1,339 @@
+use async_trait::async_trait;
+use chrono::{DateTime, Utc};
+use serde::{Deserialize, Serialize};
+use tranquil_types::{CidLink, Did, Handle};
+use uuid::Uuid;
+
+use crate::DbError;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
+pub enum InviteCodeSortOrder {
+    #[default]
+    Recent,
+    Usage,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize, sqlx::Type)]
+#[sqlx(type_name = "comms_channel", rename_all = "snake_case")]
+pub enum CommsChannel {
+    Email,
+    Discord,
+    Telegram,
+    Signal,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, sqlx::Type)]
+#[sqlx(type_name = "comms_type", rename_all = "snake_case")]
+pub enum CommsType {
+    Welcome,
+    EmailVerification,
+    PasswordReset,
+    EmailUpdate,
+    AccountDeletion,
+    AdminEmail,
+    PlcOperation,
+    TwoFactorCode,
+    PasskeyRecovery,
+    LegacyLoginAlert,
+    MigrationVerification,
+    ChannelVerification,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, sqlx::Type)]
+#[sqlx(type_name = "comms_status", rename_all = "snake_case")]
+pub enum CommsStatus {
+    Pending,
+    Processing,
+    Sent,
+    Failed,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct QueuedComms {
+    pub id: Uuid,
+    pub user_id: Option<Uuid>,
+    pub channel: CommsChannel,
+    pub comms_type: CommsType,
+    pub status: CommsStatus,
+    pub recipient: String,
+    pub subject: Option<String>,
+    pub body: String,
+    pub metadata: Option<serde_json::Value>,
+    pub attempts: i32,
+    pub max_attempts: i32,
+    pub last_error: Option<String>,
+    pub created_at: DateTime<Utc>,
+    pub updated_at: DateTime<Utc>,
+    pub scheduled_for: DateTime<Utc>,
+    pub processed_at: Option<DateTime<Utc>>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct InviteCodeInfo {
+    pub code: String,
+    pub available_uses: i32,
+    pub disabled: bool,
+    pub for_account: Option<Did>,
+    pub created_at: DateTime<Utc>,
+    pub created_by: Option<Did>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct InviteCodeUse {
+    pub code: String,
+    pub used_by_did: Did,
+    pub used_by_handle: Option<Handle>,
+    pub used_at: DateTime<Utc>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct InviteCodeRow {
+    pub code: String,
+    pub available_uses: i32,
+    pub disabled: Option<bool>,
+    pub created_by_user: Uuid,
+    pub created_at: DateTime<Utc>,
+}
+
+#[derive(Debug, Clone)]
+pub struct ReservedSigningKey {
+    pub id: Uuid,
+    pub private_key_bytes: Vec<u8>,
+}
+
+#[derive(Debug, Clone)]
+pub struct DeletionRequest {
+    pub did: Did,
+    pub expires_at: DateTime<Utc>,
+}
+
+#[async_trait]
+pub trait InfraRepository: Send + Sync {
+    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>;
+
+    async fn fetch_pending_comms(
+        &self,
+        now: DateTime<Utc>,
+        batch_size: i64,
+    ) -> Result<Vec<QueuedComms>, DbError>;
+
+    async fn mark_comms_sent(&self, id: Uuid) -> Result<(), DbError>;
+
+    async fn mark_comms_failed(&self, id: Uuid, error: &str) -> Result<(), DbError>;
+
+    async fn create_invite_code(
+        &self,
+        code: &str,
+        use_count: i32,
+        for_account: Option<&Did>,
+    ) -> Result<bool, DbError>;
+
+    async fn create_invite_codes_batch(
+        &self,
+        codes: &[String],
+        use_count: i32,
+        created_by_user: Uuid,
+        for_account: Option<&Did>,
+    ) -> Result<(), DbError>;
+
+    async fn get_invite_code_available_uses(&self, code: &str) -> Result<Option<i32>, DbError>;
+
+    async fn is_invite_code_valid(&self, code: &str) -> Result<bool, DbError>;
+
+    async fn decrement_invite_code_uses(&self, code: &str) -> Result<(), DbError>;
+
+    async fn record_invite_code_use(&self, code: &str, used_by_user: Uuid) -> Result<(), DbError>;
+
+    async fn get_invite_codes_for_account(
+        &self,
+        for_account: &Did,
+    ) -> Result<Vec<InviteCodeInfo>, DbError>;
+
+    async fn get_invite_code_uses(&self, code: &str) -> Result<Vec<InviteCodeUse>, DbError>;
+
+    async fn disable_invite_codes_by_code(&self, codes: &[String]) -> Result<(), DbError>;
+
+    async fn disable_invite_codes_by_account(&self, accounts: &[Did]) -> Result<(), DbError>;
+
+    async fn list_invite_codes(
+        &self,
+        cursor: Option<&str>,
+        limit: i64,
+        sort: InviteCodeSortOrder,
+    ) -> Result<Vec<InviteCodeRow>, DbError>;
+
+    async fn get_user_dids_by_ids(&self, user_ids: &[Uuid]) -> Result<Vec<(Uuid, Did)>, DbError>;
+
+    async fn get_invite_code_uses_batch(
+        &self,
+        codes: &[String],
+    ) -> Result<Vec<InviteCodeUse>, DbError>;
+
+    async fn get_invites_created_by_user(
+        &self,
+        user_id: Uuid,
+    ) -> Result<Vec<InviteCodeInfo>, DbError>;
+
+    async fn get_invite_code_info(&self, code: &str) -> Result<Option<InviteCodeInfo>, DbError>;
+
+    async fn get_invite_codes_by_users(
+        &self,
+        user_ids: &[Uuid],
+    ) -> Result<Vec<(Uuid, InviteCodeInfo)>, DbError>;
+
+    async fn get_invite_code_used_by_user(&self, user_id: Uuid) -> Result<Option<String>, DbError>;
+
+    async fn delete_invite_code_uses_by_user(&self, user_id: Uuid) -> Result<(), DbError>;
+
+    async fn delete_invite_codes_by_user(&self, user_id: Uuid) -> Result<(), DbError>;
+
+    async fn reserve_signing_key(
+        &self,
+        did: Option<&Did>,
+        public_key_did_key: &str,
+        private_key_bytes: &[u8],
+        expires_at: DateTime<Utc>,
+    ) -> Result<Uuid, DbError>;
+
+    async fn get_reserved_signing_key(
+        &self,
+        public_key_did_key: &str,
+    ) -> Result<Option<ReservedSigningKey>, DbError>;
+
+    async fn mark_signing_key_used(&self, key_id: Uuid) -> Result<(), DbError>;
+
+    async fn create_deletion_request(
+        &self,
+        token: &str,
+        did: &Did,
+        expires_at: DateTime<Utc>,
+    ) -> Result<(), DbError>;
+
+    async fn get_deletion_request(&self, token: &str) -> Result<Option<DeletionRequest>, DbError>;
+
+    async fn delete_deletion_request(&self, token: &str) -> Result<(), DbError>;
+
+    async fn delete_deletion_requests_by_did(&self, did: &Did) -> Result<(), DbError>;
+
+    async fn upsert_account_preference(
+        &self,
+        user_id: Uuid,
+        name: &str,
+        value_json: serde_json::Value,
+    ) -> Result<(), DbError>;
+
+    async fn insert_account_preference_if_not_exists(
+        &self,
+        user_id: Uuid,
+        name: &str,
+        value_json: serde_json::Value,
+    ) -> Result<(), DbError>;
+
+    async fn get_server_config(&self, key: &str) -> Result<Option<String>, DbError>;
+
+    async fn health_check(&self) -> Result<bool, DbError>;
+
+    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>;
+
+    async fn delete_plc_tokens_for_user(&self, user_id: Uuid) -> Result<(), DbError>;
+
+    async fn insert_plc_token(
+        &self,
+        user_id: Uuid,
+        token: &str,
+        expires_at: DateTime<Utc>,
+    ) -> Result<(), DbError>;
+
+    async fn get_plc_token_expiry(
+        &self,
+        user_id: Uuid,
+        token: &str,
+    ) -> Result<Option<DateTime<Utc>>, DbError>;
+
+    async fn delete_plc_token(&self, user_id: Uuid, token: &str) -> Result<(), DbError>;
+
+    async fn get_account_preferences(
+        &self,
+        user_id: Uuid,
+    ) -> Result<Vec<(String, serde_json::Value)>, DbError>;
+
+    async fn replace_namespace_preferences(
+        &self,
+        user_id: Uuid,
+        namespace: &str,
+        preferences: Vec<(String, serde_json::Value)>,
+    ) -> Result<(), DbError>;
+
+    async fn get_notification_history(
+        &self,
+        user_id: Uuid,
+        limit: i64,
+    ) -> Result<Vec<NotificationHistoryRow>, DbError>;
+
+    async fn get_server_configs(
+        &self,
+        keys: &[&str],
+    ) -> Result<Vec<(String, String)>, DbError>;
+
+    async fn upsert_server_config(&self, key: &str, value: &str) -> Result<(), DbError>;
+
+    async fn delete_server_config(&self, key: &str) -> Result<(), DbError>;
+
+    async fn get_blob_storage_key_by_cid(&self, cid: &CidLink) -> Result<Option<String>, DbError>;
+
+    async fn delete_blob_by_cid(&self, cid: &CidLink) -> Result<(), DbError>;
+
+    async fn get_admin_account_info_by_did(
+        &self,
+        did: &Did,
+    ) -> Result<Option<AdminAccountInfo>, DbError>;
+
+    async fn get_admin_account_infos_by_dids(
+        &self,
+        dids: &[Did],
+    ) -> Result<Vec<AdminAccountInfo>, DbError>;
+
+    async fn get_invite_code_uses_by_users(
+        &self,
+        user_ids: &[Uuid],
+    ) -> Result<Vec<(Uuid, String)>, DbError>;
+}
+
+#[derive(Debug, Clone)]
+pub struct NotificationHistoryRow {
+    pub created_at: DateTime<Utc>,
+    pub channel: String,
+    pub comms_type: String,
+    pub status: String,
+    pub subject: Option<String>,
+    pub body: String,
+}
+
+#[derive(Debug, Clone)]
+pub struct AdminAccountInfo {
+    pub id: Uuid,
+    pub did: Did,
+    pub handle: Handle,
+    pub email: Option<String>,
+    pub created_at: DateTime<Utc>,
+    pub invites_disabled: bool,
+    pub email_verified: bool,
+    pub deactivated_at: Option<DateTime<Utc>>,
+}

crates/tranquil-db-traits/src/lib.rs

@@ -0,0 +1,58 @@
+mod backlink;
+mod backup;
+mod blob;
+mod delegation;
+mod error;
+mod infra;
+mod oauth;
+mod repo;
+mod session;
+mod user;
+
+pub use backlink::{Backlink, BacklinkRepository};
+pub use backup::{
+    BackupForDeletion, BackupRepository, BackupRow, BackupStorageInfo, BlobExportInfo,
+    OldBackupInfo, UserBackupInfo,
+};
+pub use blob::{
+    BlobForExport, BlobMetadata, BlobRepository, BlobWithTakedown, MissingBlobInfo,
+};
+pub use delegation::{
+    AuditLogEntry, ControllerInfo, DelegatedAccountInfo, DelegationActionType, DelegationGrant,
+    DelegationRepository,
+};
+pub use error::DbError;
+pub use infra::{
+    AdminAccountInfo, CommsChannel, CommsStatus, CommsType, DeletionRequest, InfraRepository,
+    InviteCodeInfo, InviteCodeRow, InviteCodeSortOrder, InviteCodeUse, NotificationHistoryRow,
+    QueuedComms, ReservedSigningKey,
+};
+pub use oauth::{
+    DeviceAccountRow, DeviceTrustInfo, OAuthRepository, OAuthSessionListItem, RefreshTokenLookup,
+    ScopePreference, TrustedDeviceRow, TwoFactorChallenge,
+};
+pub use repo::{
+    ApplyCommitError, ApplyCommitInput, ApplyCommitResult, BrokenGenesisCommit, CommitEventData,
+    EventBlocksCids, FullRecordInfo, ImportBlock, ImportRecord, ImportRepoError, RecordDelete,
+    RecordInfo, RecordUpsert, RecordWithTakedown, RepoAccountInfo, RepoEventNotifier,
+    RepoEventReceiver, RepoInfo, RepoListItem, RepoRepository, RepoSeqEvent, RepoWithoutRev,
+    SequencedEvent, UserNeedingRecordBlobsBackfill, UserWithoutBlocks,
+};
+pub use session::{
+    AppPasswordCreate, AppPasswordRecord, RefreshSessionResult, SessionForRefresh, SessionListItem,
+    SessionMfaStatus, SessionRefreshData, SessionRepository, SessionToken, SessionTokenCreate,
+};
+pub use user::{
+    AccountSearchResult, CompletePasskeySetupInput, CreateAccountError, CreateDelegatedAccountInput,
+    CreatePasskeyAccountInput, CreatePasswordAccountInput, CreatePasswordAccountResult,
+    DidWebOverrides, MigrationReactivationError, MigrationReactivationInput, NotificationPrefs,
+    OAuthTokenWithUser, PasswordResetResult, ReactivatedAccountInfo, RecoverPasskeyAccountInput,
+    RecoverPasskeyAccountResult, ScheduledDeletionAccount, StoredBackupCode, StoredPasskey,
+    TotpRecord, 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,
+};

crates/tranquil-db-traits/src/oauth.rs

@@ -0,0 +1,245 @@
+use async_trait::async_trait;
+use chrono::{DateTime, Utc};
+use serde::{Deserialize, Serialize};
+use tranquil_oauth::{AuthorizedClientData, DeviceData, RequestData, TokenData};
+use tranquil_types::{AuthorizationCode, ClientId, DPoPProofId, DeviceId, Did, Handle, RefreshToken, RequestId, TokenId};
+use uuid::Uuid;
+
+use crate::DbError;
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ScopePreference {
+    pub scope: String,
+    pub granted: bool,
+}
+
+#[derive(Debug, Clone)]
+pub struct DeviceAccountRow {
+    pub did: Did,
+    pub handle: Handle,
+    pub email: Option<String>,
+    pub last_used_at: DateTime<Utc>,
+}
+
+#[derive(Debug, Clone)]
+pub struct TwoFactorChallenge {
+    pub id: Uuid,
+    pub did: Did,
+    pub request_uri: String,
+    pub code: String,
+    pub attempts: i32,
+    pub created_at: DateTime<Utc>,
+    pub expires_at: DateTime<Utc>,
+}
+
+#[derive(Debug, Clone)]
+pub struct TrustedDeviceRow {
+    pub id: String,
+    pub user_agent: Option<String>,
+    pub friendly_name: Option<String>,
+    pub trusted_at: Option<DateTime<Utc>>,
+    pub trusted_until: Option<DateTime<Utc>>,
+    pub last_seen_at: DateTime<Utc>,
+}
+
+#[derive(Debug, Clone)]
+pub struct DeviceTrustInfo {
+    pub trusted_at: Option<DateTime<Utc>>,
+    pub trusted_until: Option<DateTime<Utc>>,
+}
+
+#[derive(Debug, Clone)]
+pub struct OAuthSessionListItem {
+    pub id: i32,
+    pub token_id: TokenId,
+    pub created_at: DateTime<Utc>,
+    pub expires_at: DateTime<Utc>,
+    pub client_id: ClientId,
+}
+
+pub enum RefreshTokenLookup {
+    Valid {
+        db_id: i32,
+        token_data: TokenData,
+    },
+    InGracePeriod {
+        db_id: i32,
+        token_data: TokenData,
+        rotated_at: DateTime<Utc>,
+    },
+    Used {
+        original_token_id: i32,
+    },
+    Expired {
+        db_id: i32,
+    },
+    NotFound,
+}
+
+impl RefreshTokenLookup {
+    pub fn state(&self) -> &'static str {
+        match self {
+            Self::Valid { .. } => "valid",
+            Self::InGracePeriod { .. } => "grace_period",
+            Self::Used { .. } => "used",
+            Self::Expired { .. } => "expired",
+            Self::NotFound => "not_found",
+        }
+    }
+}
+
+#[async_trait]
+pub trait OAuthRepository: Send + Sync {
+    async fn create_token(&self, data: &TokenData) -> Result<i32, DbError>;
+    async fn get_token_by_id(&self, token_id: &TokenId) -> Result<Option<TokenData>, DbError>;
+    async fn get_token_by_refresh_token(
+        &self,
+        refresh_token: &RefreshToken,
+    ) -> Result<Option<(i32, TokenData)>, DbError>;
+    async fn get_token_by_previous_refresh_token(
+        &self,
+        refresh_token: &RefreshToken,
+    ) -> Result<Option<(i32, TokenData)>, DbError>;
+    async fn rotate_token(
+        &self,
+        old_db_id: i32,
+        new_refresh_token: &RefreshToken,
+        new_expires_at: DateTime<Utc>,
+    ) -> Result<(), DbError>;
+    async fn check_refresh_token_used(&self, refresh_token: &RefreshToken) -> Result<Option<i32>, DbError>;
+    async fn delete_token(&self, token_id: &TokenId) -> Result<(), DbError>;
+    async fn delete_token_family(&self, db_id: i32) -> Result<(), DbError>;
+    async fn list_tokens_for_user(&self, did: &Did) -> Result<Vec<TokenData>, DbError>;
+    async fn count_tokens_for_user(&self, did: &Did) -> Result<i64, DbError>;
+    async fn delete_oldest_tokens_for_user(
+        &self,
+        did: &Did,
+        keep_count: i64,
+    ) -> Result<u64, DbError>;
+    async fn revoke_tokens_for_client(&self, did: &Did, client_id: &ClientId) -> Result<u64, DbError>;
+    async fn revoke_tokens_for_controller(
+        &self,
+        delegated_did: &Did,
+        controller_did: &Did,
+    ) -> Result<u64, DbError>;
+
+    async fn create_authorization_request(
+        &self,
+        request_id: &RequestId,
+        data: &RequestData,
+    ) -> Result<(), DbError>;
+    async fn get_authorization_request(
+        &self,
+        request_id: &RequestId,
+    ) -> Result<Option<RequestData>, DbError>;
+    async fn set_authorization_did(
+        &self,
+        request_id: &RequestId,
+        did: &Did,
+        device_id: Option<&DeviceId>,
+    ) -> Result<(), DbError>;
+    async fn update_authorization_request(
+        &self,
+        request_id: &RequestId,
+        did: &Did,
+        device_id: Option<&DeviceId>,
+        code: &AuthorizationCode,
+    ) -> Result<(), DbError>;
+    async fn consume_authorization_request_by_code(
+        &self,
+        code: &AuthorizationCode,
+    ) -> Result<Option<RequestData>, DbError>;
+    async fn delete_authorization_request(&self, request_id: &RequestId) -> Result<(), DbError>;
+    async fn delete_expired_authorization_requests(&self) -> Result<u64, DbError>;
+    async fn mark_request_authenticated(
+        &self,
+        request_id: &RequestId,
+        did: &Did,
+        device_id: Option<&DeviceId>,
+    ) -> Result<(), DbError>;
+    async fn update_request_scope(&self, request_id: &RequestId, scope: &str) -> Result<(), DbError>;
+    async fn set_controller_did(&self, request_id: &RequestId, controller_did: &Did)
+        -> Result<(), DbError>;
+    async fn set_request_did(&self, request_id: &RequestId, did: &Did) -> Result<(), DbError>;
+
+    async fn create_device(&self, device_id: &DeviceId, data: &DeviceData) -> Result<(), DbError>;
+    async fn get_device(&self, device_id: &DeviceId) -> Result<Option<DeviceData>, DbError>;
+    async fn update_device_last_seen(&self, device_id: &DeviceId) -> Result<(), DbError>;
+    async fn delete_device(&self, device_id: &DeviceId) -> Result<(), DbError>;
+    async fn upsert_account_device(&self, did: &Did, device_id: &DeviceId) -> Result<(), DbError>;
+    async fn get_device_accounts(&self, device_id: &DeviceId) -> Result<Vec<DeviceAccountRow>, DbError>;
+    async fn verify_account_on_device(&self, device_id: &DeviceId, did: &Did) -> Result<bool, DbError>;
+
+    async fn check_and_record_dpop_jti(&self, jti: &DPoPProofId) -> Result<bool, DbError>;
+    async fn cleanup_expired_dpop_jtis(&self, max_age_secs: i64) -> Result<u64, DbError>;
+
+    async fn create_2fa_challenge(
+        &self,
+        did: &Did,
+        request_uri: &RequestId,
+    ) -> Result<TwoFactorChallenge, DbError>;
+    async fn get_2fa_challenge(
+        &self,
+        request_uri: &RequestId,
+    ) -> Result<Option<TwoFactorChallenge>, DbError>;
+    async fn increment_2fa_attempts(&self, id: Uuid) -> Result<i32, DbError>;
+    async fn delete_2fa_challenge(&self, id: Uuid) -> Result<(), DbError>;
+    async fn delete_2fa_challenge_by_request_uri(&self, request_uri: &RequestId) -> Result<(), DbError>;
+    async fn cleanup_expired_2fa_challenges(&self) -> Result<u64, DbError>;
+    async fn check_user_2fa_enabled(&self, did: &Did) -> Result<bool, DbError>;
+
+    async fn get_scope_preferences(
+        &self,
+        did: &Did,
+        client_id: &ClientId,
+    ) -> Result<Vec<ScopePreference>, DbError>;
+    async fn upsert_scope_preferences(
+        &self,
+        did: &Did,
+        client_id: &ClientId,
+        prefs: &[ScopePreference],
+    ) -> Result<(), DbError>;
+    async fn delete_scope_preferences(&self, did: &Did, client_id: &ClientId) -> Result<(), DbError>;
+
+    async fn upsert_authorized_client(
+        &self,
+        did: &Did,
+        client_id: &ClientId,
+        data: &AuthorizedClientData,
+    ) -> Result<(), DbError>;
+    async fn get_authorized_client(
+        &self,
+        did: &Did,
+        client_id: &ClientId,
+    ) -> Result<Option<AuthorizedClientData>, DbError>;
+
+    async fn list_trusted_devices(&self, did: &Did) -> Result<Vec<TrustedDeviceRow>, DbError>;
+    async fn get_device_trust_info(
+        &self,
+        device_id: &DeviceId,
+        did: &Did,
+    ) -> Result<Option<DeviceTrustInfo>, DbError>;
+    async fn device_belongs_to_user(&self, device_id: &DeviceId, did: &Did) -> Result<bool, DbError>;
+    async fn revoke_device_trust(&self, device_id: &DeviceId) -> Result<(), DbError>;
+    async fn update_device_friendly_name(
+        &self,
+        device_id: &DeviceId,
+        friendly_name: Option<&str>,
+    ) -> Result<(), DbError>;
+    async fn trust_device(
+        &self,
+        device_id: &DeviceId,
+        trusted_at: DateTime<Utc>,
+        trusted_until: DateTime<Utc>,
+    ) -> Result<(), DbError>;
+    async fn extend_device_trust(
+        &self,
+        device_id: &DeviceId,
+        trusted_until: DateTime<Utc>,
+    ) -> Result<(), DbError>;
+
+    async fn list_sessions_by_did(&self, did: &Did) -> Result<Vec<OAuthSessionListItem>, DbError>;
+    async fn delete_session_by_id(&self, session_id: i32, did: &Did) -> Result<u64, DbError>;
+    async fn delete_sessions_by_did(&self, did: &Did) -> Result<u64, DbError>;
+    async fn delete_sessions_by_did_except(&self, did: &Did, except_token_id: &TokenId) -> Result<u64, DbError>;
+}

crates/tranquil-db-traits/src/repo.rs

@@ -0,0 +1,388 @@
+use async_trait::async_trait;
+use chrono::{DateTime, Utc};
+use serde::{Deserialize, Serialize};
+use tranquil_types::{AtUri, CidLink, Did, Handle, Nsid, Rkey};
+use uuid::Uuid;
+
+use crate::DbError;
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct RepoAccountInfo {
+    pub user_id: Uuid,
+    pub did: Did,
+    pub deactivated_at: Option<DateTime<Utc>>,
+    pub takedown_ref: Option<String>,
+    pub repo_root_cid: Option<CidLink>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct RepoInfo {
+    pub user_id: Uuid,
+    pub repo_root_cid: CidLink,
+    pub repo_rev: Option<String>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct RecordInfo {
+    pub rkey: Rkey,
+    pub record_cid: CidLink,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct FullRecordInfo {
+    pub collection: Nsid,
+    pub rkey: Rkey,
+    pub record_cid: CidLink,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct RecordWithTakedown {
+    pub id: Uuid,
+    pub takedown_ref: Option<String>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct RepoWithoutRev {
+    pub user_id: Uuid,
+    pub repo_root_cid: CidLink,
+}
+
+#[derive(Debug, Clone)]
+pub struct BrokenGenesisCommit {
+    pub seq: i64,
+    pub did: Did,
+    pub commit_cid: Option<CidLink>,
+}
+
+#[derive(Debug, Clone)]
+pub struct UserWithoutBlocks {
+    pub user_id: Uuid,
+    pub repo_root_cid: CidLink,
+    pub repo_rev: Option<String>,
+}
+
+#[derive(Debug, Clone)]
+pub struct UserNeedingRecordBlobsBackfill {
+    pub user_id: Uuid,
+    pub did: Did,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct RepoSeqEvent {
+    pub seq: i64,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct SequencedEvent {
+    pub seq: i64,
+    pub did: Did,
+    pub created_at: DateTime<Utc>,
+    pub event_type: String,
+    pub commit_cid: Option<CidLink>,
+    pub prev_cid: Option<CidLink>,
+    pub prev_data_cid: Option<CidLink>,
+    pub ops: Option<serde_json::Value>,
+    pub blobs: Option<Vec<String>>,
+    pub blocks_cids: Option<Vec<String>>,
+    pub handle: Option<Handle>,
+    pub active: Option<bool>,
+    pub status: Option<String>,
+    pub rev: Option<String>,
+}
+
+#[derive(Debug, Clone)]
+pub struct CommitEventData {
+    pub did: Did,
+    pub event_type: String,
+    pub commit_cid: Option<CidLink>,
+    pub prev_cid: Option<CidLink>,
+    pub ops: Option<serde_json::Value>,
+    pub blobs: Option<Vec<String>>,
+    pub blocks_cids: Option<Vec<String>>,
+    pub prev_data_cid: Option<CidLink>,
+    pub rev: Option<String>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct EventBlocksCids {
+    pub blocks_cids: Option<Vec<String>>,
+    pub commit_cid: Option<CidLink>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct RepoListItem {
+    pub did: Did,
+    pub deactivated_at: Option<DateTime<Utc>>,
+    pub takedown_ref: Option<String>,
+    pub repo_root_cid: CidLink,
+    pub repo_rev: Option<String>,
+}
+
+#[derive(Debug, Clone)]
+pub struct ImportBlock {
+    pub cid_bytes: Vec<u8>,
+    pub data: Vec<u8>,
+}
+
+#[derive(Debug, Clone)]
+pub struct ImportRecord {
+    pub collection: Nsid,
+    pub rkey: Rkey,
+    pub record_cid: CidLink,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum ImportRepoError {
+    RepoNotFound,
+    ConcurrentModification,
+    Database(String),
+}
+
+#[derive(Debug, Clone)]
+pub struct RecordUpsert {
+    pub collection: Nsid,
+    pub rkey: Rkey,
+    pub cid: CidLink,
+}
+
+#[derive(Debug, Clone)]
+pub struct RecordDelete {
+    pub collection: Nsid,
+    pub rkey: Rkey,
+}
+
+#[derive(Debug, Clone)]
+pub struct ApplyCommitInput {
+    pub user_id: Uuid,
+    pub did: Did,
+    pub expected_root_cid: Option<CidLink>,
+    pub new_root_cid: CidLink,
+    pub new_rev: String,
+    pub new_block_cids: Vec<Vec<u8>>,
+    pub obsolete_block_cids: Vec<Vec<u8>>,
+    pub record_upserts: Vec<RecordUpsert>,
+    pub record_deletes: Vec<RecordDelete>,
+    pub commit_event: CommitEventData,
+}
+
+#[derive(Debug, Clone)]
+pub struct ApplyCommitResult {
+    pub seq: i64,
+    pub is_account_active: bool,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum ApplyCommitError {
+    RepoNotFound,
+    ConcurrentModification,
+    Database(String),
+}
+
+#[async_trait]
+pub trait RepoRepository: Send + Sync {
+    async fn create_repo(
+        &self,
+        user_id: Uuid,
+        repo_root_cid: &CidLink,
+        repo_rev: &str,
+    ) -> Result<(), DbError>;
+
+    async fn update_repo_root(
+        &self,
+        user_id: Uuid,
+        repo_root_cid: &CidLink,
+        repo_rev: &str,
+    ) -> Result<(), DbError>;
+
+    async fn update_repo_rev(&self, user_id: Uuid, repo_rev: &str) -> Result<(), DbError>;
+
+    async fn delete_repo(&self, user_id: Uuid) -> Result<(), DbError>;
+
+    async fn get_repo_root_for_update(&self, user_id: Uuid) -> Result<Option<CidLink>, DbError>;
+
+    async fn get_repo(&self, user_id: Uuid) -> Result<Option<RepoInfo>, DbError>;
+
+    async fn get_repo_root_by_did(&self, did: &Did) -> Result<Option<CidLink>, DbError>;
+
+    async fn count_repos(&self) -> Result<i64, DbError>;
+
+    async fn get_repos_without_rev(&self) -> Result<Vec<RepoWithoutRev>, DbError>;
+
+    async fn upsert_records(
+        &self,
+        repo_id: Uuid,
+        collections: &[Nsid],
+        rkeys: &[Rkey],
+        record_cids: &[CidLink],
+        repo_rev: &str,
+    ) -> Result<(), DbError>;
+
+    async fn delete_records(
+        &self,
+        repo_id: Uuid,
+        collections: &[Nsid],
+        rkeys: &[Rkey],
+    ) -> Result<(), DbError>;
+
+    async fn delete_all_records(&self, repo_id: Uuid) -> Result<(), DbError>;
+
+    async fn get_record_cid(
+        &self,
+        repo_id: Uuid,
+        collection: &Nsid,
+        rkey: &Rkey,
+    ) -> Result<Option<CidLink>, DbError>;
+
+    async fn list_records(
+        &self,
+        repo_id: Uuid,
+        collection: &Nsid,
+        cursor: Option<&Rkey>,
+        limit: i64,
+        reverse: bool,
+        rkey_start: Option<&Rkey>,
+        rkey_end: Option<&Rkey>,
+    ) -> Result<Vec<RecordInfo>, DbError>;
+
+    async fn get_all_records(&self, repo_id: Uuid) -> Result<Vec<FullRecordInfo>, DbError>;
+
+    async fn list_collections(&self, repo_id: Uuid) -> Result<Vec<Nsid>, DbError>;
+
+    async fn count_records(&self, repo_id: Uuid) -> Result<i64, DbError>;
+
+    async fn count_all_records(&self) -> Result<i64, DbError>;
+
+    async fn get_record_by_cid(&self, cid: &CidLink) -> Result<Option<RecordWithTakedown>, DbError>;
+
+    async fn set_record_takedown(&self, cid: &CidLink, takedown_ref: Option<&str>)
+        -> Result<(), DbError>;
+
+    async fn insert_user_blocks(
+        &self,
+        user_id: Uuid,
+        block_cids: &[Vec<u8>],
+        repo_rev: &str,
+    ) -> Result<(), DbError>;
+
+    async fn delete_user_blocks(&self, user_id: Uuid, block_cids: &[Vec<u8>])
+        -> Result<(), DbError>;
+
+    async fn get_user_block_cids_since_rev(
+        &self,
+        user_id: Uuid,
+        since_rev: &str,
+    ) -> Result<Vec<Vec<u8>>, DbError>;
+
+    async fn count_user_blocks(&self, user_id: Uuid) -> Result<i64, DbError>;
+
+    async fn insert_commit_event(&self, data: &CommitEventData) -> Result<i64, DbError>;
+
+    async fn insert_identity_event(&self, did: &Did, handle: Option<&Handle>) -> Result<i64, DbError>;
+
+    async fn insert_account_event(
+        &self,
+        did: &Did,
+        active: bool,
+        status: Option<&str>,
+    ) -> Result<i64, DbError>;
+
+    async fn insert_sync_event(
+        &self,
+        did: &Did,
+        commit_cid: &CidLink,
+        rev: Option<&str>,
+    ) -> Result<i64, DbError>;
+
+    async fn insert_genesis_commit_event(
+        &self,
+        did: &Did,
+        commit_cid: &CidLink,
+        mst_root_cid: &CidLink,
+        rev: &str,
+    ) -> Result<i64, DbError>;
+
+    async fn update_seq_blocks_cids(&self, seq: i64, blocks_cids: &[String])
+        -> Result<(), DbError>;
+
+    async fn delete_sequences_except(&self, did: &Did, keep_seq: i64) -> Result<(), DbError>;
+
+    async fn get_max_seq(&self) -> Result<i64, DbError>;
+
+    async fn get_min_seq_since(&self, since: DateTime<Utc>) -> Result<Option<i64>, DbError>;
+
+    async fn get_account_with_repo(&self, did: &Did) -> Result<Option<RepoAccountInfo>, DbError>;
+
+    async fn get_events_since_seq(
+        &self,
+        since_seq: i64,
+        limit: Option<i64>,
+    ) -> Result<Vec<SequencedEvent>, DbError>;
+
+    async fn get_events_in_seq_range(
+        &self,
+        start_seq: i64,
+        end_seq: i64,
+    ) -> Result<Vec<SequencedEvent>, DbError>;
+
+    async fn get_event_by_seq(&self, seq: i64) -> Result<Option<SequencedEvent>, DbError>;
+
+    async fn get_events_since_cursor(
+        &self,
+        cursor: i64,
+        limit: i64,
+    ) -> Result<Vec<SequencedEvent>, DbError>;
+
+    async fn get_events_since_rev(
+        &self,
+        did: &Did,
+        since_rev: &str,
+    ) -> Result<Vec<EventBlocksCids>, DbError>;
+
+    async fn list_repos_paginated(
+        &self,
+        cursor_did: Option<&Did>,
+        limit: i64,
+    ) -> Result<Vec<RepoListItem>, DbError>;
+
+    async fn get_repo_root_cid_by_user_id(&self, user_id: Uuid) -> Result<Option<CidLink>, DbError>;
+
+    async fn notify_update(&self, seq: i64) -> Result<(), DbError>;
+
+    async fn import_repo_data(
+        &self,
+        user_id: Uuid,
+        blocks: &[ImportBlock],
+        records: &[ImportRecord],
+    ) -> Result<(), ImportRepoError>;
+
+    async fn apply_commit(
+        &self,
+        input: ApplyCommitInput,
+    ) -> Result<ApplyCommitResult, ApplyCommitError>;
+
+    async fn get_broken_genesis_commits(&self) -> Result<Vec<BrokenGenesisCommit>, DbError>;
+
+    async fn get_users_without_blocks(&self) -> Result<Vec<UserWithoutBlocks>, DbError>;
+
+    async fn get_users_needing_record_blobs_backfill(
+        &self,
+        limit: i64,
+    ) -> Result<Vec<UserNeedingRecordBlobsBackfill>, DbError>;
+
+    async fn insert_record_blobs(
+        &self,
+        repo_id: Uuid,
+        record_uris: &[AtUri],
+        blob_cids: &[CidLink],
+    ) -> Result<(), DbError>;
+}
+
+#[async_trait]
+pub trait RepoEventNotifier: Send + Sync {
+    async fn subscribe(&self) -> Result<Box<dyn RepoEventReceiver>, DbError>;
+}
+
+#[async_trait]
+pub trait RepoEventReceiver: Send {
+    async fn recv(&mut self) -> Option<i64>;
+}

crates/tranquil-db-traits/src/session.rs

@@ -0,0 +1,203 @@
+use async_trait::async_trait;
+use chrono::{DateTime, Utc};
+use tranquil_types::Did;
+use uuid::Uuid;
+
+use crate::DbError;
+
+#[derive(Debug, Clone)]
+pub struct SessionToken {
+    pub id: i32,
+    pub did: Did,
+    pub access_jti: String,
+    pub refresh_jti: String,
+    pub access_expires_at: DateTime<Utc>,
+    pub refresh_expires_at: DateTime<Utc>,
+    pub legacy_login: bool,
+    pub mfa_verified: bool,
+    pub scope: Option<String>,
+    pub controller_did: Option<Did>,
+    pub app_password_name: Option<String>,
+    pub created_at: DateTime<Utc>,
+    pub updated_at: DateTime<Utc>,
+}
+
+#[derive(Debug, Clone)]
+pub struct SessionTokenCreate {
+    pub did: Did,
+    pub access_jti: String,
+    pub refresh_jti: String,
+    pub access_expires_at: DateTime<Utc>,
+    pub refresh_expires_at: DateTime<Utc>,
+    pub legacy_login: bool,
+    pub mfa_verified: bool,
+    pub scope: Option<String>,
+    pub controller_did: Option<Did>,
+    pub app_password_name: Option<String>,
+}
+
+#[derive(Debug, Clone)]
+pub struct SessionForRefresh {
+    pub id: i32,
+    pub did: Did,
+    pub scope: Option<String>,
+    pub controller_did: Option<Did>,
+    pub key_bytes: Vec<u8>,
+    pub encryption_version: i32,
+}
+
+#[derive(Debug, Clone)]
+pub struct SessionListItem {
+    pub id: i32,
+    pub access_jti: String,
+    pub created_at: DateTime<Utc>,
+    pub refresh_expires_at: DateTime<Utc>,
+}
+
+#[derive(Debug, Clone)]
+pub struct AppPasswordRecord {
+    pub id: Uuid,
+    pub user_id: Uuid,
+    pub name: String,
+    pub password_hash: String,
+    pub created_at: DateTime<Utc>,
+    pub privileged: bool,
+    pub scopes: Option<String>,
+    pub created_by_controller_did: Option<Did>,
+}
+
+#[derive(Debug, Clone)]
+pub struct AppPasswordCreate {
+    pub user_id: Uuid,
+    pub name: String,
+    pub password_hash: String,
+    pub privileged: bool,
+    pub scopes: Option<String>,
+    pub created_by_controller_did: Option<Did>,
+}
+
+#[derive(Debug, Clone)]
+pub struct SessionMfaStatus {
+    pub legacy_login: bool,
+    pub mfa_verified: bool,
+    pub last_reauth_at: Option<DateTime<Utc>>,
+}
+
+#[derive(Debug, Clone)]
+pub enum RefreshSessionResult {
+    Success,
+    TokenAlreadyUsed,
+    ConcurrentRefresh,
+}
+
+#[derive(Debug, Clone)]
+pub struct SessionRefreshData {
+    pub old_refresh_jti: String,
+    pub session_id: i32,
+    pub new_access_jti: String,
+    pub new_refresh_jti: String,
+    pub new_access_expires_at: DateTime<Utc>,
+    pub new_refresh_expires_at: DateTime<Utc>,
+}
+
+#[async_trait]
+pub trait SessionRepository: Send + Sync {
+    async fn create_session(&self, data: &SessionTokenCreate) -> Result<i32, DbError>;
+
+    async fn get_session_by_access_jti(
+        &self,
+        access_jti: &str,
+    ) -> Result<Option<SessionToken>, DbError>;
+
+    async fn get_session_for_refresh(
+        &self,
+        refresh_jti: &str,
+    ) -> Result<Option<SessionForRefresh>, DbError>;
+
+    async fn update_session_tokens(
+        &self,
+        session_id: i32,
+        new_access_jti: &str,
+        new_refresh_jti: &str,
+        new_access_expires_at: DateTime<Utc>,
+        new_refresh_expires_at: DateTime<Utc>,
+    ) -> Result<(), DbError>;
+
+    async fn delete_session_by_access_jti(&self, access_jti: &str) -> Result<u64, DbError>;
+
+    async fn delete_session_by_id(&self, session_id: i32) -> Result<u64, DbError>;
+
+    async fn delete_sessions_by_did(&self, did: &Did) -> Result<u64, DbError>;
+
+    async fn delete_sessions_by_did_except_jti(
+        &self,
+        did: &Did,
+        except_jti: &str,
+    ) -> Result<u64, DbError>;
+
+    async fn list_sessions_by_did(&self, did: &Did) -> Result<Vec<SessionListItem>, DbError>;
+
+    async fn get_session_access_jti_by_id(
+        &self,
+        session_id: i32,
+        did: &Did,
+    ) -> Result<Option<String>, DbError>;
+
+    async fn delete_sessions_by_app_password(
+        &self,
+        did: &Did,
+        app_password_name: &str,
+    ) -> Result<u64, DbError>;
+
+    async fn get_session_jtis_by_app_password(
+        &self,
+        did: &Did,
+        app_password_name: &str,
+    ) -> Result<Vec<String>, DbError>;
+
+    async fn check_refresh_token_used(&self, refresh_jti: &str) -> Result<Option<i32>, DbError>;
+
+    async fn mark_refresh_token_used(
+        &self,
+        refresh_jti: &str,
+        session_id: i32,
+    ) -> Result<bool, DbError>;
+
+    async fn list_app_passwords(&self, user_id: Uuid) -> Result<Vec<AppPasswordRecord>, DbError>;
+
+    async fn get_app_passwords_for_login(
+        &self,
+        user_id: Uuid,
+    ) -> Result<Vec<AppPasswordRecord>, DbError>;
+
+    async fn get_app_password_by_name(
+        &self,
+        user_id: Uuid,
+        name: &str,
+    ) -> Result<Option<AppPasswordRecord>, DbError>;
+
+    async fn create_app_password(&self, data: &AppPasswordCreate) -> Result<Uuid, DbError>;
+
+    async fn delete_app_password(&self, user_id: Uuid, name: &str) -> Result<u64, DbError>;
+
+    async fn delete_app_passwords_by_controller(
+        &self,
+        did: &Did,
+        controller_did: &Did,
+    ) -> Result<u64, DbError>;
+
+    async fn get_last_reauth_at(&self, did: &Did) -> Result<Option<DateTime<Utc>>, DbError>;
+
+    async fn update_last_reauth(&self, did: &Did) -> Result<DateTime<Utc>, DbError>;
+
+    async fn get_session_mfa_status(&self, did: &Did) -> Result<Option<SessionMfaStatus>, DbError>;
+
+    async fn update_mfa_verified(&self, did: &Did) -> Result<(), DbError>;
+
+    async fn get_app_password_hashes_by_did(&self, did: &Did) -> Result<Vec<String>, DbError>;
+
+    async fn refresh_session_atomic(
+        &self,
+        data: &SessionRefreshData,
+    ) -> Result<RefreshSessionResult, DbError>;
+}

crates/tranquil-db-traits/src/user.rs

@@ -0,0 +1,902 @@
+use async_trait::async_trait;
+use chrono::{DateTime, Utc};
+use tranquil_types::{Did, Handle};
+use uuid::Uuid;
+
+use crate::{CommsChannel, DbError};
+
+#[derive(Debug, Clone)]
+pub struct UserRow {
+    pub id: Uuid,
+    pub did: Did,
+    pub handle: Handle,
+    pub email: Option<String>,
+    pub created_at: DateTime<Utc>,
+    pub deactivated_at: Option<DateTime<Utc>>,
+    pub takedown_ref: Option<String>,
+    pub is_admin: bool,
+}
+
+#[derive(Debug, Clone)]
+pub struct UserWithKey {
+    pub id: Uuid,
+    pub did: Did,
+    pub handle: Handle,
+    pub email: Option<String>,
+    pub deactivated_at: Option<DateTime<Utc>>,
+    pub takedown_ref: Option<String>,
+    pub is_admin: bool,
+    pub key_bytes: Vec<u8>,
+    pub encryption_version: Option<i32>,
+}
+
+#[derive(Debug, Clone)]
+pub struct UserStatus {
+    pub deactivated_at: Option<DateTime<Utc>>,
+    pub takedown_ref: Option<String>,
+    pub is_admin: bool,
+}
+
+#[derive(Debug, Clone)]
+pub struct UserEmailInfo {
+    pub id: Uuid,
+    pub handle: Handle,
+    pub email: Option<String>,
+    pub email_verified: bool,
+}
+
+#[derive(Debug, Clone)]
+pub struct UserLoginCheck {
+    pub did: Did,
+    pub password_hash: Option<String>,
+}
+
+#[derive(Debug, Clone)]
+pub struct UserLoginInfo {
+    pub id: Uuid,
+    pub did: Did,
+    pub email: Option<String>,
+    pub password_hash: Option<String>,
+    pub password_required: bool,
+    pub two_factor_enabled: bool,
+    pub preferred_comms_channel: CommsChannel,
+    pub deactivated_at: Option<DateTime<Utc>>,
+    pub takedown_ref: Option<String>,
+    pub email_verified: bool,
+    pub discord_verified: bool,
+    pub telegram_verified: bool,
+    pub signal_verified: bool,
+    pub account_type: String,
+}
+
+#[derive(Debug, Clone)]
+pub struct User2faStatus {
+    pub id: Uuid,
+    pub two_factor_enabled: bool,
+    pub preferred_comms_channel: CommsChannel,
+    pub email_verified: bool,
+    pub discord_verified: bool,
+    pub telegram_verified: bool,
+    pub signal_verified: bool,
+}
+
+#[async_trait]
+pub trait UserRepository: Send + Sync {
+    async fn get_by_did(&self, did: &Did) -> Result<Option<UserRow>, DbError>;
+
+    async fn get_by_handle(&self, handle: &Handle) -> Result<Option<UserRow>, DbError>;
+
+    async fn get_with_key_by_did(&self, did: &Did) -> Result<Option<UserWithKey>, DbError>;
+
+    async fn get_status_by_did(&self, did: &Did) -> Result<Option<UserStatus>, DbError>;
+
+    async fn count_users(&self) -> Result<i64, DbError>;
+
+    async fn get_session_access_expiry(
+        &self,
+        did: &Did,
+        access_jti: &str,
+    ) -> Result<Option<DateTime<Utc>>, DbError>;
+
+    async fn get_oauth_token_with_user(
+        &self,
+        token_id: &str,
+    ) -> Result<Option<OAuthTokenWithUser>, DbError>;
+
+    async fn get_user_info_by_did(&self, did: &Did) -> Result<Option<UserInfoForAuth>, DbError>;
+
+    async fn get_any_admin_user_id(&self) -> Result<Option<Uuid>, DbError>;
+
+    async fn set_invites_disabled(&self, did: &Did, disabled: bool) -> Result<bool, DbError>;
+
+    async fn search_accounts(
+        &self,
+        cursor_did: Option<&Did>,
+        email_filter: Option<&str>,
+        handle_filter: Option<&str>,
+        limit: i64,
+    ) -> Result<Vec<AccountSearchResult>, DbError>;
+
+    async fn get_auth_info_by_did(&self, did: &Did) -> Result<Option<UserAuthInfo>, DbError>;
+
+    async fn get_by_email(&self, email: &str) -> Result<Option<UserForVerification>, DbError>;
+
+    async fn get_login_check_by_handle_or_email(
+        &self,
+        identifier: &str,
+    ) -> Result<Option<UserLoginCheck>, DbError>;
+
+    async fn get_login_info_by_handle_or_email(
+        &self,
+        identifier: &str,
+    ) -> Result<Option<UserLoginInfo>, DbError>;
+
+    async fn get_2fa_status_by_did(&self, did: &Did) -> Result<Option<User2faStatus>, DbError>;
+
+    async fn get_comms_prefs(&self, user_id: Uuid) -> Result<Option<UserCommsPrefs>, DbError>;
+
+    async fn get_id_by_did(&self, did: &Did) -> Result<Option<Uuid>, DbError>;
+
+    async fn get_user_key_by_id(&self, user_id: Uuid) -> Result<Option<UserKeyInfo>, DbError>;
+
+    async fn get_id_and_handle_by_did(&self, did: &Did) -> Result<Option<UserIdAndHandle>, DbError>;
+
+    async fn get_did_web_info_by_handle(
+        &self,
+        handle: &Handle,
+    ) -> Result<Option<UserDidWebInfo>, DbError>;
+
+    async fn get_did_web_overrides(&self, user_id: Uuid) -> Result<Option<DidWebOverrides>, DbError>;
+
+    async fn get_handle_by_did(&self, did: &Did) -> Result<Option<Handle>, DbError>;
+
+    async fn is_account_active_by_did(&self, did: &Did) -> Result<Option<bool>, DbError>;
+
+    async fn get_user_for_deletion(
+        &self,
+        did: &Did,
+    ) -> Result<Option<UserForDeletion>, DbError>;
+
+    async fn check_handle_exists(&self, handle: &Handle, exclude_user_id: Uuid) -> Result<bool, DbError>;
+
+    async fn update_handle(&self, user_id: Uuid, handle: &Handle) -> Result<(), DbError>;
+
+    async fn get_user_with_key_by_did(
+        &self,
+        did: &Did,
+    ) -> Result<Option<UserKeyWithId>, DbError>;
+
+    async fn is_account_migrated(&self, did: &Did) -> Result<bool, DbError>;
+
+    async fn has_verified_comms_channel(&self, did: &Did) -> Result<bool, DbError>;
+
+    async fn get_id_by_handle(&self, handle: &Handle) -> Result<Option<Uuid>, DbError>;
+
+    async fn get_email_info_by_did(&self, did: &Did) -> Result<Option<UserEmailInfo>, DbError>;
+
+    async fn check_email_exists(&self, email: &str, exclude_user_id: Uuid) -> Result<bool, DbError>;
+
+    async fn update_email(&self, user_id: Uuid, email: &str) -> Result<(), DbError>;
+
+    async fn set_email_verified(&self, user_id: Uuid, verified: bool) -> Result<(), DbError>;
+
+    async fn check_email_verified_by_identifier(
+        &self,
+        identifier: &str,
+    ) -> Result<Option<bool>, DbError>;
+
+    async fn admin_update_email(&self, did: &Did, email: &str) -> Result<u64, DbError>;
+
+    async fn admin_update_handle(&self, did: &Did, handle: &Handle) -> Result<u64, DbError>;
+
+    async fn admin_update_password(&self, did: &Did, password_hash: &str) -> Result<u64, DbError>;
+
+    async fn get_notification_prefs(&self, did: &Did) -> Result<Option<NotificationPrefs>, DbError>;
+
+    async fn get_id_handle_email_by_did(
+        &self,
+        did: &Did,
+    ) -> Result<Option<UserIdHandleEmail>, DbError>;
+
+    async fn update_preferred_comms_channel(&self, did: &Did, channel: &str) -> Result<(), DbError>;
+
+    async fn clear_discord(&self, user_id: Uuid) -> Result<(), DbError>;
+
+    async fn clear_telegram(&self, user_id: Uuid) -> Result<(), DbError>;
+
+    async fn clear_signal(&self, user_id: Uuid) -> Result<(), DbError>;
+
+    async fn get_verification_info(
+        &self,
+        did: &Did,
+    ) -> Result<Option<UserVerificationInfo>, DbError>;
+
+    async fn verify_email_channel(&self, user_id: Uuid, email: &str) -> Result<bool, DbError>;
+
+    async fn verify_discord_channel(&self, user_id: Uuid, discord_id: &str) -> Result<(), DbError>;
+
+    async fn verify_telegram_channel(
+        &self,
+        user_id: Uuid,
+        telegram_username: &str,
+    ) -> Result<(), DbError>;
+
+    async fn verify_signal_channel(&self, user_id: Uuid, signal_number: &str)
+        -> Result<(), DbError>;
+
+    async fn set_email_verified_flag(&self, user_id: Uuid) -> Result<(), DbError>;
+
+    async fn set_discord_verified_flag(&self, user_id: Uuid) -> Result<(), DbError>;
+
+    async fn set_telegram_verified_flag(&self, user_id: Uuid) -> Result<(), DbError>;
+
+    async fn set_signal_verified_flag(&self, user_id: Uuid) -> Result<(), DbError>;
+
+    async fn has_totp_enabled(&self, did: &Did) -> Result<bool, DbError>;
+
+    async fn has_passkeys(&self, did: &Did) -> Result<bool, DbError>;
+
+    async fn get_password_hash_by_did(&self, did: &Did) -> Result<Option<String>, DbError>;
+
+    async fn get_passkeys_for_user(&self, did: &Did) -> Result<Vec<StoredPasskey>, DbError>;
+
+    async fn get_passkey_by_credential_id(
+        &self,
+        credential_id: &[u8],
+    ) -> Result<Option<StoredPasskey>, DbError>;
+
+    async fn save_passkey(
+        &self,
+        did: &Did,
+        credential_id: &[u8],
+        public_key: &[u8],
+        friendly_name: Option<&str>,
+    ) -> Result<Uuid, DbError>;
+
+    async fn update_passkey_counter(
+        &self,
+        credential_id: &[u8],
+        new_counter: i32,
+    ) -> Result<bool, DbError>;
+
+    async fn delete_passkey(&self, id: Uuid, did: &Did) -> Result<bool, DbError>;
+
+    async fn update_passkey_name(&self, id: Uuid, did: &Did, name: &str) -> Result<bool, DbError>;
+
+    async fn save_webauthn_challenge(
+        &self,
+        did: &Did,
+        challenge_type: &str,
+        state_json: &str,
+    ) -> Result<Uuid, DbError>;
+
+    async fn load_webauthn_challenge(
+        &self,
+        did: &Did,
+        challenge_type: &str,
+    ) -> Result<Option<String>, DbError>;
+
+    async fn delete_webauthn_challenge(&self, did: &Did, challenge_type: &str)
+        -> Result<(), DbError>;
+
+    async fn get_totp_record(&self, did: &Did) -> Result<Option<TotpRecord>, DbError>;
+
+    async fn upsert_totp_secret(
+        &self,
+        did: &Did,
+        secret_encrypted: &[u8],
+        encryption_version: i32,
+    ) -> Result<(), DbError>;
+
+    async fn set_totp_verified(&self, did: &Did) -> Result<(), DbError>;
+
+    async fn update_totp_last_used(&self, did: &Did) -> Result<(), DbError>;
+
+    async fn delete_totp(&self, did: &Did) -> Result<(), DbError>;
+
+    async fn get_unused_backup_codes(&self, did: &Did) -> Result<Vec<StoredBackupCode>, DbError>;
+
+    async fn mark_backup_code_used(&self, code_id: Uuid) -> Result<bool, DbError>;
+
+    async fn count_unused_backup_codes(&self, did: &Did) -> Result<i64, DbError>;
+
+    async fn delete_backup_codes(&self, did: &Did) -> Result<u64, DbError>;
+
+    async fn insert_backup_codes(&self, did: &Did, code_hashes: &[String]) -> Result<(), DbError>;
+
+    async fn enable_totp_with_backup_codes(
+        &self,
+        did: &Did,
+        code_hashes: &[String],
+    ) -> Result<(), DbError>;
+
+    async fn delete_totp_and_backup_codes(&self, did: &Did) -> Result<(), DbError>;
+
+    async fn replace_backup_codes(&self, did: &Did, code_hashes: &[String]) -> Result<(), DbError>;
+
+    async fn get_session_info_by_did(&self, did: &Did) -> Result<Option<UserSessionInfo>, DbError>;
+
+    async fn get_legacy_login_pref(&self, did: &Did) -> Result<Option<UserLegacyLoginPref>, DbError>;
+
+    async fn update_legacy_login(&self, did: &Did, allow: bool) -> Result<bool, DbError>;
+
+    async fn update_locale(&self, did: &Did, locale: &str) -> Result<bool, DbError>;
+
+    async fn get_login_full_by_identifier(
+        &self,
+        identifier: &str,
+    ) -> Result<Option<UserLoginFull>, DbError>;
+
+    async fn get_confirm_signup_by_did(
+        &self,
+        did: &Did,
+    ) -> Result<Option<UserConfirmSignup>, DbError>;
+
+    async fn get_resend_verification_by_did(
+        &self,
+        did: &Did,
+    ) -> Result<Option<UserResendVerification>, DbError>;
+
+    async fn set_channel_verified(&self, did: &Did, channel: CommsChannel) -> Result<(), DbError>;
+
+    async fn get_id_by_email_or_handle(
+        &self,
+        email: &str,
+        handle: &str,
+    ) -> Result<Option<Uuid>, DbError>;
+
+    async fn set_password_reset_code(
+        &self,
+        user_id: Uuid,
+        code: &str,
+        expires_at: DateTime<Utc>,
+    ) -> Result<(), DbError>;
+
+    async fn get_user_by_reset_code(
+        &self,
+        code: &str,
+    ) -> Result<Option<UserResetCodeInfo>, DbError>;
+
+    async fn clear_password_reset_code(&self, user_id: Uuid) -> Result<(), DbError>;
+
+    async fn get_id_and_password_hash_by_did(
+        &self,
+        did: &Did,
+    ) -> Result<Option<UserIdAndPasswordHash>, DbError>;
+
+    async fn update_password_hash(&self, user_id: Uuid, password_hash: &str) -> Result<(), DbError>;
+
+    async fn reset_password_with_sessions(
+        &self,
+        user_id: Uuid,
+        password_hash: &str,
+    ) -> Result<PasswordResetResult, DbError>;
+
+    async fn activate_account(&self, did: &Did) -> Result<bool, DbError>;
+
+    async fn deactivate_account(
+        &self,
+        did: &Did,
+        delete_after: Option<DateTime<Utc>>,
+    ) -> Result<bool, DbError>;
+
+    async fn has_password_by_did(&self, did: &Did) -> Result<Option<bool>, DbError>;
+
+    async fn get_password_info_by_did(
+        &self,
+        did: &Did,
+    ) -> Result<Option<UserPasswordInfo>, DbError>;
+
+    async fn remove_user_password(&self, user_id: Uuid) -> Result<(), DbError>;
+
+    async fn set_new_user_password(&self, user_id: Uuid, password_hash: &str) -> Result<(), DbError>;
+
+    async fn get_user_key_by_did(&self, did: &Did) -> Result<Option<UserKeyInfo>, DbError>;
+
+    async fn delete_account_complete(
+        &self,
+        user_id: Uuid,
+        did: &Did,
+    ) -> Result<(), DbError>;
+
+    async fn set_user_takedown(&self, did: &Did, takedown_ref: Option<&str>) -> Result<bool, DbError>;
+
+    async fn admin_delete_account_complete(&self, user_id: Uuid, did: &Did) -> Result<(), DbError>;
+
+    async fn get_user_for_did_doc(&self, did: &Did) -> Result<Option<UserForDidDoc>, DbError>;
+
+    async fn get_user_for_did_doc_build(&self, did: &Did) -> Result<Option<UserForDidDocBuild>, DbError>;
+
+    async fn upsert_did_web_overrides(
+        &self,
+        user_id: Uuid,
+        verification_methods: Option<serde_json::Value>,
+        also_known_as: Option<Vec<String>>,
+    ) -> Result<(), DbError>;
+
+    async fn update_migrated_to_pds(
+        &self,
+        did: &Did,
+        endpoint: &str,
+    ) -> Result<(), DbError>;
+
+    async fn get_user_for_passkey_setup(&self, did: &Did) -> Result<Option<UserForPasskeySetup>, DbError>;
+
+    async fn get_user_for_passkey_recovery(
+        &self,
+        identifier: &str,
+        normalized_handle: &str,
+    ) -> Result<Option<UserForPasskeyRecovery>, DbError>;
+
+    async fn set_recovery_token(
+        &self,
+        did: &Did,
+        token_hash: &str,
+        expires_at: DateTime<Utc>,
+    ) -> Result<(), DbError>;
+
+    async fn get_user_for_recovery(&self, did: &Did) -> Result<Option<UserForRecovery>, DbError>;
+
+    async fn get_accounts_scheduled_for_deletion(
+        &self,
+        limit: i64,
+    ) -> Result<Vec<ScheduledDeletionAccount>, DbError>;
+
+    async fn delete_account_with_firehose(
+        &self,
+        user_id: Uuid,
+        did: &Did,
+    ) -> Result<i64, DbError>;
+
+    async fn create_password_account(
+        &self,
+        input: &CreatePasswordAccountInput,
+    ) -> Result<CreatePasswordAccountResult, CreateAccountError>;
+
+    async fn create_delegated_account(
+        &self,
+        input: &CreateDelegatedAccountInput,
+    ) -> Result<Uuid, CreateAccountError>;
+
+    async fn create_passkey_account(
+        &self,
+        input: &CreatePasskeyAccountInput,
+    ) -> Result<CreatePasswordAccountResult, CreateAccountError>;
+
+    async fn reactivate_migration_account(
+        &self,
+        input: &MigrationReactivationInput,
+    ) -> Result<ReactivatedAccountInfo, MigrationReactivationError>;
+
+    async fn check_handle_available_for_new_account(&self, handle: &Handle) -> Result<bool, DbError>;
+
+    async fn check_and_consume_invite_code(&self, code: &str) -> Result<bool, DbError>;
+
+    async fn complete_passkey_setup(
+        &self,
+        input: &CompletePasskeySetupInput,
+    ) -> Result<(), DbError>;
+
+    async fn recover_passkey_account(
+        &self,
+        input: &RecoverPasskeyAccountInput,
+    ) -> Result<RecoverPasskeyAccountResult, DbError>;
+}
+
+#[derive(Debug, Clone)]
+pub struct UserKeyWithId {
+    pub id: Uuid,
+    pub key_bytes: Vec<u8>,
+    pub encryption_version: Option<i32>,
+}
+
+#[derive(Debug, Clone)]
+pub struct UserKeyInfo {
+    pub key_bytes: Vec<u8>,
+    pub encryption_version: Option<i32>,
+}
+
+#[derive(Debug, Clone)]
+pub struct UserIdAndHandle {
+    pub id: Uuid,
+    pub handle: Handle,
+}
+
+#[derive(Debug, Clone)]
+pub struct UserDidWebInfo {
+    pub id: Uuid,
+    pub did: Did,
+    pub migrated_to_pds: Option<String>,
+}
+
+#[derive(Debug, Clone)]
+pub struct DidWebOverrides {
+    pub verification_methods: serde_json::Value,
+    pub also_known_as: Vec<String>,
+}
+
+#[derive(Debug, Clone)]
+pub struct UserCommsPrefs {
+    pub email: Option<String>,
+    pub handle: Handle,
+    pub preferred_channel: String,
+    pub preferred_locale: Option<String>,
+}
+
+#[derive(Debug, Clone)]
+pub struct UserForVerification {
+    pub id: Uuid,
+    pub did: Did,
+    pub email: Option<String>,
+    pub email_verified: bool,
+    pub handle: Handle,
+}
+
+#[derive(Debug, Clone)]
+pub struct OAuthTokenWithUser {
+    pub did: Did,
+    pub expires_at: DateTime<Utc>,
+    pub deactivated_at: Option<DateTime<Utc>>,
+    pub takedown_ref: Option<String>,
+    pub is_admin: bool,
+    pub key_bytes: Option<Vec<u8>>,
+    pub encryption_version: Option<i32>,
+}
+
+#[derive(Debug, Clone)]
+pub struct UserInfoForAuth {
+    pub deactivated_at: Option<DateTime<Utc>>,
+    pub takedown_ref: Option<String>,
+    pub is_admin: bool,
+    pub key_bytes: Option<Vec<u8>>,
+    pub encryption_version: Option<i32>,
+}
+
+#[derive(Debug, Clone)]
+pub struct AccountSearchResult {
+    pub did: Did,
+    pub handle: Handle,
+    pub email: Option<String>,
+    pub created_at: DateTime<Utc>,
+    pub email_verified: bool,
+    pub deactivated_at: Option<DateTime<Utc>>,
+    pub invites_disabled: Option<bool>,
+}
+
+#[derive(Debug, Clone)]
+pub struct UserAuthInfo {
+    pub id: Uuid,
+    pub did: Did,
+    pub password_hash: Option<String>,
+    pub deactivated_at: Option<DateTime<Utc>>,
+    pub takedown_ref: Option<String>,
+    pub email_verified: bool,
+    pub discord_verified: bool,
+    pub telegram_verified: bool,
+    pub signal_verified: bool,
+}
+
+#[derive(Debug, Clone)]
+pub struct NotificationPrefs {
+    pub email: String,
+    pub preferred_channel: String,
+    pub discord_id: Option<String>,
+    pub discord_verified: bool,
+    pub telegram_username: Option<String>,
+    pub telegram_verified: bool,
+    pub signal_number: Option<String>,
+    pub signal_verified: bool,
+}
+
+#[derive(Debug, Clone)]
+pub struct UserIdHandleEmail {
+    pub id: Uuid,
+    pub handle: Handle,
+    pub email: Option<String>,
+}
+
+#[derive(Debug, Clone)]
+pub struct UserVerificationInfo {
+    pub id: Uuid,
+    pub handle: Handle,
+    pub email: Option<String>,
+    pub email_verified: bool,
+    pub discord_verified: bool,
+    pub telegram_verified: bool,
+    pub signal_verified: bool,
+}
+
+#[derive(Debug, Clone)]
+pub struct StoredPasskey {
+    pub id: Uuid,
+    pub did: Did,
+    pub credential_id: Vec<u8>,
+    pub public_key: Vec<u8>,
+    pub sign_count: i32,
+    pub created_at: DateTime<Utc>,
+    pub last_used: Option<DateTime<Utc>>,
+    pub friendly_name: Option<String>,
+    pub aaguid: Option<Vec<u8>>,
+    pub transports: Option<Vec<String>>,
+}
+
+impl StoredPasskey {
+    pub fn credential_id_base64(&self) -> String {
+        use base64::{Engine, engine::general_purpose::URL_SAFE_NO_PAD};
+        URL_SAFE_NO_PAD.encode(&self.credential_id)
+    }
+}
+
+#[derive(Debug, Clone)]
+pub struct TotpRecord {
+    pub secret_encrypted: Vec<u8>,
+    pub encryption_version: i32,
+    pub verified: bool,
+}
+
+#[derive(Debug, Clone)]
+pub struct StoredBackupCode {
+    pub id: Uuid,
+    pub code_hash: String,
+}
+
+#[derive(Debug, Clone)]
+pub struct UserSessionInfo {
+    pub handle: Handle,
+    pub email: Option<String>,
+    pub email_verified: bool,
+    pub is_admin: bool,
+    pub deactivated_at: Option<DateTime<Utc>>,
+    pub takedown_ref: Option<String>,
+    pub preferred_locale: Option<String>,
+    pub preferred_comms_channel: CommsChannel,
+    pub discord_verified: bool,
+    pub telegram_verified: bool,
+    pub signal_verified: bool,
+    pub migrated_to_pds: Option<String>,
+    pub migrated_at: Option<DateTime<Utc>>,
+}
+
+#[derive(Debug, Clone)]
+pub struct UserLegacyLoginPref {
+    pub allow_legacy_login: bool,
+    pub has_mfa: bool,
+}
+
+#[derive(Debug, Clone)]
+pub struct UserLoginFull {
+    pub id: Uuid,
+    pub did: Did,
+    pub handle: Handle,
+    pub password_hash: Option<String>,
+    pub email: Option<String>,
+    pub deactivated_at: Option<DateTime<Utc>>,
+    pub takedown_ref: Option<String>,
+    pub email_verified: bool,
+    pub discord_verified: bool,
+    pub telegram_verified: bool,
+    pub signal_verified: bool,
+    pub allow_legacy_login: bool,
+    pub migrated_to_pds: Option<String>,
+    pub preferred_comms_channel: CommsChannel,
+    pub key_bytes: Vec<u8>,
+    pub encryption_version: Option<i32>,
+    pub totp_enabled: bool,
+}
+
+#[derive(Debug, Clone)]
+pub struct UserConfirmSignup {
+    pub id: Uuid,
+    pub did: Did,
+    pub handle: Handle,
+    pub email: Option<String>,
+    pub channel: CommsChannel,
+    pub discord_id: Option<String>,
+    pub telegram_username: Option<String>,
+    pub signal_number: Option<String>,
+    pub key_bytes: Vec<u8>,
+    pub encryption_version: Option<i32>,
+}
+
+#[derive(Debug, Clone)]
+pub struct UserResendVerification {
+    pub id: Uuid,
+    pub handle: Handle,
+    pub email: Option<String>,
+    pub channel: CommsChannel,
+    pub discord_id: Option<String>,
+    pub telegram_username: Option<String>,
+    pub signal_number: Option<String>,
+    pub email_verified: bool,
+    pub discord_verified: bool,
+    pub telegram_verified: bool,
+    pub signal_verified: bool,
+}
+
+#[derive(Debug, Clone)]
+pub struct UserResetCodeInfo {
+    pub id: Uuid,
+    pub expires_at: Option<DateTime<Utc>>,
+}
+
+#[derive(Debug, Clone)]
+pub struct UserPasswordInfo {
+    pub id: Uuid,
+    pub password_hash: Option<String>,
+}
+
+#[derive(Debug, Clone)]
+pub struct UserIdAndPasswordHash {
+    pub id: Uuid,
+    pub password_hash: String,
+}
+
+#[derive(Debug, Clone)]
+pub struct PasswordResetResult {
+    pub did: Did,
+    pub session_jtis: Vec<String>,
+}
+
+#[derive(Debug, Clone)]
+pub struct UserForDeletion {
+    pub id: Uuid,
+    pub password_hash: Option<String>,
+    pub handle: Handle,
+}
+
+#[derive(Debug, Clone)]
+pub struct ScheduledDeletionAccount {
+    pub id: Uuid,
+    pub did: Did,
+    pub handle: Handle,
+}
+
+#[derive(Debug, Clone)]
+pub struct UserForDidDoc {
+    pub id: Uuid,
+    pub handle: Handle,
+    pub deactivated_at: Option<DateTime<Utc>>,
+}
+
+#[derive(Debug, Clone)]
+pub struct UserForDidDocBuild {
+    pub id: Uuid,
+    pub handle: Handle,
+    pub migrated_to_pds: Option<String>,
+}
+
+#[derive(Debug, Clone)]
+pub struct UserForPasskeySetup {
+    pub id: Uuid,
+    pub handle: Handle,
+    pub recovery_token: Option<String>,
+    pub recovery_token_expires_at: Option<DateTime<Utc>>,
+    pub password_required: bool,
+}
+
+#[derive(Debug, Clone)]
+pub struct UserForPasskeyRecovery {
+    pub id: Uuid,
+    pub did: Did,
+    pub handle: Handle,
+    pub password_required: bool,
+}
+
+#[derive(Debug, Clone)]
+pub struct UserForRecovery {
+    pub id: Uuid,
+    pub did: Did,
+    pub recovery_token: Option<String>,
+    pub recovery_token_expires_at: Option<DateTime<Utc>>,
+}
+
+#[derive(Debug, Clone)]
+pub struct CreatePasswordAccountInput {
+    pub handle: Handle,
+    pub email: Option<String>,
+    pub did: Did,
+    pub password_hash: String,
+    pub preferred_comms_channel: CommsChannel,
+    pub discord_id: Option<String>,
+    pub telegram_username: Option<String>,
+    pub signal_number: Option<String>,
+    pub deactivated_at: Option<DateTime<Utc>>,
+    pub encrypted_key_bytes: Vec<u8>,
+    pub encryption_version: i32,
+    pub reserved_key_id: Option<Uuid>,
+    pub commit_cid: String,
+    pub repo_rev: String,
+    pub genesis_block_cids: Vec<Vec<u8>>,
+    pub invite_code: Option<String>,
+    pub birthdate_pref: Option<serde_json::Value>,
+}
+
+#[derive(Debug, Clone, Default)]
+pub struct CreatePasswordAccountResult {
+    pub user_id: Uuid,
+    pub is_admin: bool,
+}
+
+#[derive(Debug, Clone)]
+pub enum CreateAccountError {
+    HandleTaken,
+    EmailTaken,
+    DidExists,
+    Database(String),
+}
+
+#[derive(Debug, Clone)]
+pub struct CreateDelegatedAccountInput {
+    pub handle: Handle,
+    pub email: Option<String>,
+    pub did: Did,
+    pub controller_did: Did,
+    pub controller_scopes: String,
+    pub encrypted_key_bytes: Vec<u8>,
+    pub encryption_version: i32,
+    pub commit_cid: String,
+    pub repo_rev: String,
+    pub genesis_block_cids: Vec<Vec<u8>>,
+    pub invite_code: Option<String>,
+}
+
+#[derive(Debug, Clone)]
+pub struct CreatePasskeyAccountInput {
+    pub handle: Handle,
+    pub email: String,
+    pub did: Did,
+    pub preferred_comms_channel: CommsChannel,
+    pub discord_id: Option<String>,
+    pub telegram_username: Option<String>,
+    pub signal_number: Option<String>,
+    pub setup_token_hash: String,
+    pub setup_expires_at: DateTime<Utc>,
+    pub deactivated_at: Option<DateTime<Utc>>,
+    pub encrypted_key_bytes: Vec<u8>,
+    pub encryption_version: i32,
+    pub reserved_key_id: Option<Uuid>,
+    pub commit_cid: String,
+    pub repo_rev: String,
+    pub genesis_block_cids: Vec<Vec<u8>>,
+    pub invite_code: Option<String>,
+    pub birthdate_pref: Option<serde_json::Value>,
+}
+
+#[derive(Debug, Clone)]
+pub struct CompletePasskeySetupInput {
+    pub user_id: Uuid,
+    pub did: Did,
+    pub app_password_name: String,
+    pub app_password_hash: String,
+}
+
+#[derive(Debug, Clone)]
+pub struct RecoverPasskeyAccountInput {
+    pub did: Did,
+    pub password_hash: String,
+}
+
+#[derive(Debug, Clone)]
+pub struct RecoverPasskeyAccountResult {
+    pub passkeys_deleted: u64,
+}
+
+#[derive(Debug, Clone)]
+pub struct MigrationReactivationInput {
+    pub did: Did,
+    pub new_handle: Handle,
+}
+
+#[derive(Debug, Clone)]
+pub struct ReactivatedAccountInfo {
+    pub user_id: Uuid,
+    pub old_handle: Handle,
+}
+
+#[derive(Debug, Clone)]
+pub enum MigrationReactivationError {
+    NotFound,
+    NotDeactivated,
+    HandleTaken,
+    Database(String),
+}

crates/tranquil-db/Cargo.toml

@@ -0,0 +1,25 @@
+[package]
+name = "tranquil-db"
+version.workspace = true
+edition.workspace = true
+license.workspace = true
+
+[features]
+default = ["postgres"]
+postgres = []
+sqlite = []
+
+[dependencies]
+tranquil-db-traits = { workspace = true }
+tranquil-oauth = { workspace = true }
+tranquil-types = { workspace = true }
+async-trait = { workspace = true }
+chrono = { workspace = true }
+rand = { workspace = true }
+serde = { workspace = true }
+serde_json = { workspace = true }
+thiserror = { workspace = true }
+tracing = { workspace = true }
+uuid = { workspace = true }
+
+sqlx = { workspace = true }

crates/tranquil-db/src/lib.rs

@@ -0,0 +1,7 @@
+#[cfg(feature = "postgres")]
+pub mod postgres;
+
+pub use tranquil_db_traits::*;
+
+#[cfg(feature = "postgres")]
+pub use postgres::PostgresRepositories;

crates/tranquil-db/src/postgres/backlink.rs

@@ -0,0 +1,99 @@
+use async_trait::async_trait;
+use sqlx::PgPool;
+use tranquil_db_traits::{Backlink, BacklinkRepository, DbError};
+use tranquil_types::{AtUri, Nsid};
+use uuid::Uuid;
+
+use super::user::map_sqlx_error;
+
+pub struct PostgresBacklinkRepository {
+    pool: PgPool,
+}
+
+impl PostgresBacklinkRepository {
+    pub fn new(pool: PgPool) -> Self {
+        Self { pool }
+    }
+}
+
+#[async_trait]
+impl BacklinkRepository for PostgresBacklinkRepository {
+    async fn get_backlink_conflicts(
+        &self,
+        repo_id: Uuid,
+        collection: &Nsid,
+        backlinks: &[Backlink],
+    ) -> Result<Vec<AtUri>, DbError> {
+        if backlinks.is_empty() {
+            return Ok(Vec::new());
+        }
+
+        let paths: Vec<&str> = backlinks.iter().map(|b| b.path.as_str()).collect();
+        let link_tos: Vec<&str> = backlinks.iter().map(|b| b.link_to.as_str()).collect();
+        let collection_pattern = format!("%/{}/%", collection.as_str());
+
+        let results = sqlx::query_scalar!(
+            r#"
+            SELECT DISTINCT uri
+            FROM backlinks
+            WHERE repo_id = $1
+              AND uri LIKE $4
+              AND (path, link_to) IN (SELECT unnest($2::text[]), unnest($3::text[]))
+            "#,
+            repo_id,
+            &paths as &[&str],
+            &link_tos as &[&str],
+            collection_pattern
+        )
+        .fetch_all(&self.pool)
+        .await
+        .map_err(map_sqlx_error)?;
+
+        Ok(results.into_iter().map(Into::into).collect())
+    }
+
+    async fn add_backlinks(&self, repo_id: Uuid, backlinks: &[Backlink]) -> Result<(), DbError> {
+        if backlinks.is_empty() {
+            return Ok(());
+        }
+
+        let uris: Vec<&str> = backlinks.iter().map(|b| b.uri.as_str()).collect();
+        let paths: Vec<&str> = backlinks.iter().map(|b| b.path.as_str()).collect();
+        let link_tos: Vec<&str> = backlinks.iter().map(|b| b.link_to.as_str()).collect();
+
+        sqlx::query!(
+            r#"
+            INSERT INTO backlinks (uri, path, link_to, repo_id)
+            SELECT unnest($1::text[]), unnest($2::text[]), unnest($3::text[]), $4
+            ON CONFLICT (uri, path) DO NOTHING
+            "#,
+            &uris as &[&str],
+            &paths as &[&str],
+            &link_tos as &[&str],
+            repo_id
+        )
+        .execute(&self.pool)
+        .await
+        .map_err(map_sqlx_error)?;
+
+        Ok(())
+    }
+
+    async fn remove_backlinks_by_uri(&self, uri: &AtUri) -> Result<(), DbError> {
+        sqlx::query!("DELETE FROM backlinks WHERE uri = $1", uri.as_str())
+            .execute(&self.pool)
+            .await
+            .map_err(map_sqlx_error)?;
+
+        Ok(())
+    }
+
+    async fn remove_backlinks_by_repo(&self, repo_id: Uuid) -> Result<(), DbError> {
+        sqlx::query!("DELETE FROM backlinks WHERE repo_id = $1", repo_id)
+            .execute(&self.pool)
+            .await
+            .map_err(map_sqlx_error)?;
+
+        Ok(())
+    }
+}

crates/tranquil-db/src/postgres/backup.rs

@@ -0,0 +1,299 @@
+use async_trait::async_trait;
+use chrono::{DateTime, Utc};
+use sqlx::PgPool;
+use tranquil_db_traits::{
+    BackupForDeletion, BackupRepository, BackupRow, BackupStorageInfo, BlobExportInfo, DbError,
+    OldBackupInfo, UserBackupInfo,
+};
+use tranquil_types::Did;
+use uuid::Uuid;
+
+use super::user::map_sqlx_error;
+
+pub struct PostgresBackupRepository {
+    pool: PgPool,
+}
+
+impl PostgresBackupRepository {
+    pub fn new(pool: PgPool) -> Self {
+        Self { pool }
+    }
+}
+
+#[async_trait]
+impl BackupRepository for PostgresBackupRepository {
+    async fn get_user_backup_status(&self, did: &Did) -> Result<Option<(Uuid, bool)>, DbError> {
+        let result = sqlx::query!(
+            "SELECT id, backup_enabled FROM users WHERE did = $1",
+            did.as_str()
+        )
+        .fetch_optional(&self.pool)
+        .await
+        .map_err(map_sqlx_error)?;
+
+        Ok(result.map(|r| (r.id, r.backup_enabled)))
+    }
+
+    async fn list_backups_for_user(&self, user_id: Uuid) -> Result<Vec<BackupRow>, DbError> {
+        let results = sqlx::query_as!(
+            BackupRow,
+            r#"
+            SELECT id, repo_rev, repo_root_cid, block_count, size_bytes, created_at
+            FROM account_backups
+            WHERE user_id = $1
+            ORDER BY created_at DESC
+            "#,
+            user_id
+        )
+        .fetch_all(&self.pool)
+        .await
+        .map_err(map_sqlx_error)?;
+
+        Ok(results)
+    }
+
+    async fn get_backup_storage_info(
+        &self,
+        backup_id: Uuid,
+        did: &Did,
+    ) -> Result<Option<BackupStorageInfo>, DbError> {
+        let result = sqlx::query!(
+            r#"
+            SELECT ab.storage_key, ab.repo_rev
+            FROM account_backups ab
+            JOIN users u ON u.id = ab.user_id
+            WHERE ab.id = $1 AND u.did = $2
+            "#,
+            backup_id,
+            did.as_str()
+        )
+        .fetch_optional(&self.pool)
+        .await
+        .map_err(map_sqlx_error)?;
+
+        Ok(result.map(|r| BackupStorageInfo {
+            storage_key: r.storage_key,
+            repo_rev: r.repo_rev,
+        }))
+    }
+
+    async fn get_user_for_backup(&self, did: &Did) -> Result<Option<UserBackupInfo>, DbError> {
+        let result = sqlx::query!(
+            r#"
+            SELECT u.id, u.did, u.backup_enabled, u.deactivated_at, r.repo_root_cid, r.repo_rev
+            FROM users u
+            JOIN repos r ON r.user_id = u.id
+            WHERE u.did = $1
+            "#,
+            did.as_str()
+        )
+        .fetch_optional(&self.pool)
+        .await
+        .map_err(map_sqlx_error)?;
+
+        Ok(result.map(|r| UserBackupInfo {
+            id: r.id,
+            did: r.did.into(),
+            backup_enabled: r.backup_enabled,
+            deactivated_at: r.deactivated_at,
+            repo_root_cid: r.repo_root_cid,
+            repo_rev: r.repo_rev,
+        }))
+    }
+
+    async fn insert_backup(
+        &self,
+        user_id: Uuid,
+        storage_key: &str,
+        repo_root_cid: &str,
+        repo_rev: &str,
+        block_count: i32,
+        size_bytes: i64,
+    ) -> Result<Uuid, DbError> {
+        let id = sqlx::query_scalar!(
+            r#"
+            INSERT INTO account_backups (user_id, storage_key, repo_root_cid, repo_rev, block_count, size_bytes)
+            VALUES ($1, $2, $3, $4, $5, $6)
+            RETURNING id
+            "#,
+            user_id,
+            storage_key,
+            repo_root_cid,
+            repo_rev,
+            block_count,
+            size_bytes
+        )
+        .fetch_one(&self.pool)
+        .await
+        .map_err(map_sqlx_error)?;
+
+        Ok(id)
+    }
+
+    async fn get_old_backups(
+        &self,
+        user_id: Uuid,
+        retention_offset: i64,
+    ) -> Result<Vec<OldBackupInfo>, DbError> {
+        let results = sqlx::query!(
+            r#"
+            SELECT id, storage_key
+            FROM account_backups
+            WHERE user_id = $1
+            ORDER BY created_at DESC
+            OFFSET $2
+            "#,
+            user_id,
+            retention_offset
+        )
+        .fetch_all(&self.pool)
+        .await
+        .map_err(map_sqlx_error)?;
+
+        Ok(results
+            .into_iter()
+            .map(|r| OldBackupInfo {
+                id: r.id,
+                storage_key: r.storage_key,
+            })
+            .collect())
+    }
+
+    async fn delete_backup(&self, backup_id: Uuid) -> Result<(), DbError> {
+        sqlx::query!("DELETE FROM account_backups WHERE id = $1", backup_id)
+            .execute(&self.pool)
+            .await
+            .map_err(map_sqlx_error)?;
+
+        Ok(())
+    }
+
+    async fn get_backup_for_deletion(
+        &self,
+        backup_id: Uuid,
+        did: &Did,
+    ) -> Result<Option<BackupForDeletion>, DbError> {
+        let result = sqlx::query!(
+            r#"
+            SELECT ab.id, ab.storage_key, u.deactivated_at
+            FROM account_backups ab
+            JOIN users u ON u.id = ab.user_id
+            WHERE ab.id = $1 AND u.did = $2
+            "#,
+            backup_id,
+            did.as_str()
+        )
+        .fetch_optional(&self.pool)
+        .await
+        .map_err(map_sqlx_error)?;
+
+        Ok(result.map(|r| BackupForDeletion {
+            id: r.id,
+            storage_key: r.storage_key,
+            deactivated_at: r.deactivated_at,
+        }))
+    }
+
+    async fn get_user_deactivated_status(
+        &self,
+        did: &Did,
+    ) -> Result<Option<Option<DateTime<Utc>>>, DbError> {
+        let result = sqlx::query!(
+            "SELECT deactivated_at FROM users WHERE did = $1",
+            did.as_str()
+        )
+        .fetch_optional(&self.pool)
+        .await
+        .map_err(map_sqlx_error)?;
+
+        Ok(result.map(|r| r.deactivated_at))
+    }
+
+    async fn update_backup_enabled(&self, did: &Did, enabled: bool) -> Result<(), DbError> {
+        sqlx::query!(
+            "UPDATE users SET backup_enabled = $1 WHERE did = $2",
+            enabled,
+            did.as_str()
+        )
+        .execute(&self.pool)
+        .await
+        .map_err(map_sqlx_error)?;
+
+        Ok(())
+    }
+
+    async fn get_user_id_by_did(&self, did: &Did) -> Result<Option<Uuid>, DbError> {
+        let result = sqlx::query_scalar!("SELECT id FROM users WHERE did = $1", did.as_str())
+            .fetch_optional(&self.pool)
+            .await
+            .map_err(map_sqlx_error)?;
+
+        Ok(result)
+    }
+
+    async fn get_blobs_for_export(&self, user_id: Uuid) -> Result<Vec<BlobExportInfo>, DbError> {
+        let results = sqlx::query!(
+            r#"
+            SELECT DISTINCT b.cid, b.storage_key, b.mime_type
+            FROM blobs b
+            JOIN record_blobs rb ON rb.blob_cid = b.cid
+            WHERE rb.repo_id = $1
+            "#,
+            user_id
+        )
+        .fetch_all(&self.pool)
+        .await
+        .map_err(map_sqlx_error)?;
+
+        Ok(results
+            .into_iter()
+            .map(|r| BlobExportInfo {
+                cid: r.cid,
+                storage_key: r.storage_key,
+                mime_type: r.mime_type,
+            })
+            .collect())
+    }
+
+    async fn get_users_needing_backup(
+        &self,
+        backup_interval_secs: i64,
+        limit: i64,
+    ) -> Result<Vec<UserBackupInfo>, DbError> {
+        let results = sqlx::query!(
+            r#"
+            SELECT u.id, u.did, u.backup_enabled, u.deactivated_at, r.repo_root_cid, r.repo_rev
+            FROM users u
+            JOIN repos r ON r.user_id = u.id
+            WHERE u.backup_enabled = true
+              AND u.deactivated_at IS NULL
+              AND (
+                NOT EXISTS (
+                  SELECT 1 FROM account_backups ab WHERE ab.user_id = u.id
+                )
+                OR (
+                  SELECT MAX(ab.created_at) FROM account_backups ab WHERE ab.user_id = u.id
+                ) < NOW() - make_interval(secs => $1)
+              )
+            LIMIT $2
+            "#,
+            backup_interval_secs as f64,
+            limit
+        )
+        .fetch_all(&self.pool)
+        .await
+        .map_err(map_sqlx_error)?;
+
+        Ok(results
+            .into_iter()
+            .map(|r| UserBackupInfo {
+                id: r.id,
+                did: r.did.into(),
+                backup_enabled: r.backup_enabled,
+                deactivated_at: r.deactivated_at,
+                repo_root_cid: r.repo_root_cid,
+                repo_rev: r.repo_rev,
+            })
+            .collect())
+    }
+}

crates/tranquil-db/src/postgres/blob.rs

@@ -0,0 +1,295 @@
+use async_trait::async_trait;
+use sqlx::PgPool;
+use tranquil_db_traits::{
+    BlobForExport, BlobMetadata, BlobRepository, BlobWithTakedown, DbError, MissingBlobInfo,
+};
+use tranquil_types::{AtUri, CidLink, Did};
+use uuid::Uuid;
+
+use super::user::map_sqlx_error;
+
+pub struct PostgresBlobRepository {
+    pool: PgPool,
+}
+
+impl PostgresBlobRepository {
+    pub fn new(pool: PgPool) -> Self {
+        Self { pool }
+    }
+}
+
+#[async_trait]
+impl BlobRepository for PostgresBlobRepository {
+    async fn insert_blob(
+        &self,
+        cid: &CidLink,
+        mime_type: &str,
+        size_bytes: i64,
+        created_by_user: Uuid,
+        storage_key: &str,
+    ) -> Result<Option<CidLink>, DbError> {
+        let result = sqlx::query_scalar!(
+            r#"INSERT INTO blobs (cid, mime_type, size_bytes, created_by_user, storage_key)
+               VALUES ($1, $2, $3, $4, $5)
+               ON CONFLICT (cid) DO NOTHING RETURNING cid"#,
+            cid.as_str(),
+            mime_type,
+            size_bytes,
+            created_by_user,
+            storage_key
+        )
+        .fetch_optional(&self.pool)
+        .await
+        .map_err(map_sqlx_error)?;
+
+        Ok(result.map(CidLink::from))
+    }
+
+    async fn get_blob_metadata(&self, cid: &CidLink) -> Result<Option<BlobMetadata>, DbError> {
+        let result = sqlx::query!(
+            "SELECT storage_key, mime_type, size_bytes FROM blobs WHERE cid = $1",
+            cid.as_str()
+        )
+        .fetch_optional(&self.pool)
+        .await
+        .map_err(map_sqlx_error)?;
+
+        Ok(result.map(|r| BlobMetadata {
+            storage_key: r.storage_key,
+            mime_type: r.mime_type,
+            size_bytes: r.size_bytes,
+        }))
+    }
+
+    async fn get_blob_with_takedown(
+        &self,
+        cid: &CidLink,
+    ) -> Result<Option<BlobWithTakedown>, DbError> {
+        let result = sqlx::query!(
+            "SELECT cid, takedown_ref FROM blobs WHERE cid = $1",
+            cid.as_str()
+        )
+        .fetch_optional(&self.pool)
+        .await
+        .map_err(map_sqlx_error)?;
+
+        Ok(result.map(|r| BlobWithTakedown {
+            cid: CidLink::from(r.cid),
+            takedown_ref: r.takedown_ref,
+        }))
+    }
+
+    async fn get_blob_storage_key(&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 list_blobs_by_user(
+        &self,
+        user_id: Uuid,
+        cursor: Option<&str>,
+        limit: i64,
+    ) -> Result<Vec<CidLink>, DbError> {
+        let cursor_val = cursor.unwrap_or("");
+        let results = sqlx::query_scalar!(
+            r#"SELECT cid FROM blobs
+               WHERE created_by_user = $1 AND cid > $2
+               ORDER BY cid ASC
+               LIMIT $3"#,
+            user_id,
+            cursor_val,
+            limit
+        )
+        .fetch_all(&self.pool)
+        .await
+        .map_err(map_sqlx_error)?;
+
+        Ok(results.into_iter().map(CidLink::from).collect())
+    }
+
+    async fn list_blobs_since_rev(
+        &self,
+        did: &Did,
+        since: &str,
+    ) -> Result<Vec<CidLink>, DbError> {
+        let results = sqlx::query_scalar!(
+            r#"SELECT DISTINCT unnest(blobs) as "cid!"
+               FROM repo_seq
+               WHERE did = $1 AND rev > $2 AND blobs IS NOT NULL"#,
+            did.as_str(),
+            since
+        )
+        .fetch_all(&self.pool)
+        .await
+        .map_err(map_sqlx_error)?;
+
+        Ok(results.into_iter().map(CidLink::from).collect())
+    }
+
+    async fn count_blobs_by_user(&self, user_id: Uuid) -> Result<i64, DbError> {
+        let result = sqlx::query_scalar!(
+            r#"SELECT COUNT(*) as "count!" FROM blobs WHERE created_by_user = $1"#,
+            user_id
+        )
+        .fetch_one(&self.pool)
+        .await
+        .map_err(map_sqlx_error)?;
+
+        Ok(result)
+    }
+
+    async fn sum_blob_storage(&self) -> Result<i64, DbError> {
+        let result = sqlx::query_scalar!(
+            r#"SELECT COALESCE(SUM(size_bytes), 0)::BIGINT as "total!" FROM blobs"#
+        )
+        .fetch_one(&self.pool)
+        .await
+        .map_err(map_sqlx_error)?;
+
+        Ok(result)
+    }
+
+    async fn update_blob_takedown(
+        &self,
+        cid: &CidLink,
+        takedown_ref: Option<&str>,
+    ) -> Result<bool, DbError> {
+        let result = sqlx::query!(
+            "UPDATE blobs SET takedown_ref = $1 WHERE cid = $2",
+            takedown_ref,
+            cid.as_str()
+        )
+        .execute(&self.pool)
+        .await
+        .map_err(map_sqlx_error)?;
+
+        Ok(result.rows_affected() > 0)
+    }
+
+    async fn delete_blob_by_cid(&self, cid: &CidLink) -> Result<bool, DbError> {
+        let result = sqlx::query!("DELETE FROM blobs WHERE cid = $1", cid.as_str())
+            .execute(&self.pool)
+            .await
+            .map_err(map_sqlx_error)?;
+
+        Ok(result.rows_affected() > 0)
+    }
+
+    async fn delete_blobs_by_user(&self, user_id: Uuid) -> Result<u64, DbError> {
+        let result = sqlx::query!("DELETE FROM blobs WHERE created_by_user = $1", user_id)
+            .execute(&self.pool)
+            .await
+            .map_err(map_sqlx_error)?;
+
+        Ok(result.rows_affected())
+    }
+
+    async fn get_blob_storage_keys_by_user(&self, user_id: Uuid) -> Result<Vec<String>, DbError> {
+        let results = sqlx::query_scalar!(
+            r#"SELECT storage_key as "storage_key!" FROM blobs WHERE created_by_user = $1"#,
+            user_id
+        )
+        .fetch_all(&self.pool)
+        .await
+        .map_err(map_sqlx_error)?;
+
+        Ok(results)
+    }
+
+    async fn insert_record_blobs(
+        &self,
+        repo_id: Uuid,
+        record_uris: &[AtUri],
+        blob_cids: &[CidLink],
+    ) -> Result<(), DbError> {
+        let uris_str: Vec<&str> = record_uris.iter().map(|u| u.as_str()).collect();
+        let cids_str: Vec<&str> = blob_cids.iter().map(|c| c.as_str()).collect();
+
+        sqlx::query!(
+            r#"INSERT INTO record_blobs (repo_id, record_uri, blob_cid)
+               SELECT $1, record_uri, blob_cid
+               FROM UNNEST($2::text[], $3::text[]) AS t(record_uri, blob_cid)
+               ON CONFLICT (repo_id, record_uri, blob_cid) DO NOTHING"#,
+            repo_id,
+            &uris_str as &[&str],
+            &cids_str as &[&str]
+        )
+        .execute(&self.pool)
+        .await
+        .map_err(map_sqlx_error)?;
+
+        Ok(())
+    }
+
+    async fn list_missing_blobs(
+        &self,
+        repo_id: Uuid,
+        cursor: Option<&str>,
+        limit: i64,
+    ) -> Result<Vec<MissingBlobInfo>, DbError> {
+        let cursor_val = cursor.unwrap_or("");
+        let results = sqlx::query!(
+            r#"SELECT rb.blob_cid, rb.record_uri
+               FROM record_blobs rb
+               LEFT JOIN blobs b ON rb.blob_cid = b.cid
+               WHERE rb.repo_id = $1 AND b.cid IS NULL AND rb.blob_cid > $2
+               ORDER BY rb.blob_cid
+               LIMIT $3"#,
+            repo_id,
+            cursor_val,
+            limit
+        )
+        .fetch_all(&self.pool)
+        .await
+        .map_err(map_sqlx_error)?;
+
+        Ok(results
+            .into_iter()
+            .map(|r| MissingBlobInfo {
+                blob_cid: CidLink::from(r.blob_cid),
+                record_uri: AtUri::from(r.record_uri),
+            })
+            .collect())
+    }
+
+    async fn count_distinct_record_blobs(&self, repo_id: Uuid) -> Result<i64, DbError> {
+        let result = sqlx::query_scalar!(
+            r#"SELECT COUNT(DISTINCT blob_cid) as "count!" FROM record_blobs WHERE repo_id = $1"#,
+            repo_id
+        )
+        .fetch_one(&self.pool)
+        .await
+        .map_err(map_sqlx_error)?;
+
+        Ok(result)
+    }
+
+    async fn get_blobs_for_export(&self, repo_id: Uuid) -> Result<Vec<BlobForExport>, DbError> {
+        let results = sqlx::query!(
+            r#"SELECT DISTINCT b.cid, b.storage_key, b.mime_type
+               FROM blobs b
+               JOIN record_blobs rb ON rb.blob_cid = b.cid
+               WHERE rb.repo_id = $1"#,
+            repo_id
+        )
+        .fetch_all(&self.pool)
+        .await
+        .map_err(map_sqlx_error)?;
+
+        Ok(results
+            .into_iter()
+            .map(|r| BlobForExport {
+                cid: CidLink::from(r.cid),
+                storage_key: r.storage_key,
+                mime_type: r.mime_type,
+            })
+            .collect())
+    }
+}

crates/tranquil-db/src/postgres/delegation.rs

@@ -0,0 +1,476 @@
+use async_trait::async_trait;
+use sqlx::PgPool;
+use tranquil_db_traits::{
+    AuditLogEntry, ControllerInfo, DbError, DelegatedAccountInfo, DelegationActionType,
+    DelegationGrant, DelegationRepository,
+};
+use tranquil_types::Did;
+use uuid::Uuid;
+
+use super::user::map_sqlx_error;
+
+#[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 result = sqlx::query_scalar!(
+            r#"SELECT account_type::text = 'delegated' as "is_delegated!" FROM users WHERE did = $1"#,
+            did.as_str()
+        )
+        .fetch_optional(&self.pool)
+        .await
+        .map_err(map_sqlx_error)?;
+
+        Ok(result.unwrap_or(false))
+    }
+
+    async fn create_delegation(
+        &self,
+        delegated_did: &Did,
+        controller_did: &Did,
+        granted_scopes: &str,
+        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,
+            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: &str,
+    ) -> 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,
+            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)?;
+
+        Ok(row.map(|r| DelegationGrant {
+            id: r.id,
+            delegated_did: r.delegated_did.into(),
+            controller_did: r.controller_did.into(),
+            granted_scopes: r.granted_scopes,
+            granted_at: r.granted_at,
+            granted_by: r.granted_by.into(),
+            revoked_at: r.revoked_at,
+            revoked_by: r.revoked_by.map(Into::into),
+        }))
+    }
+
+    async fn get_delegations_for_account(
+        &self,
+        delegated_did: &Did,
+    ) -> Result<Vec<ControllerInfo>, DbError> {
+        let rows = sqlx::query!(
+            r#"
+            SELECT
+                u.did,
+                u.handle,
+                d.granted_scopes,
+                d.granted_at,
+                (u.deactivated_at IS NULL AND u.takedown_ref IS NULL) as "is_active!"
+            FROM account_delegations d
+            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)?;
+
+        Ok(rows
+            .into_iter()
+            .map(|r| ControllerInfo {
+                did: r.did.into(),
+                handle: r.handle.into(),
+                granted_scopes: r.granted_scopes,
+                granted_at: r.granted_at,
+                is_active: r.is_active,
+            })
+            .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)?;
+
+        Ok(rows
+            .into_iter()
+            .map(|r| DelegatedAccountInfo {
+                did: r.did.into(),
+                handle: r.handle.into(),
+                granted_scopes: r.granted_scopes,
+                granted_at: r.granted_at,
+            })
+            .collect())
+    }
+
+    async fn get_active_controllers_for_account(
+        &self,
+        delegated_did: &Did,
+    ) -> Result<Vec<ControllerInfo>, DbError> {
+        let rows = sqlx::query!(
+            r#"
+            SELECT
+                u.did,
+                u.handle,
+                d.granted_scopes,
+                d.granted_at,
+                true as "is_active!"
+            FROM account_delegations d
+            JOIN users u ON u.did = d.controller_did
+            WHERE d.delegated_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
+            "#,
+            delegated_did.as_str()
+        )
+        .fetch_all(&self.pool)
+        .await
+        .map_err(map_sqlx_error)?;
+
+        Ok(rows
+            .into_iter()
+            .map(|r| ControllerInfo {
+                did: r.did.into(),
+                handle: r.handle.into(),
+                granted_scopes: r.granted_scopes,
+                granted_at: r.granted_at,
+                is_active: r.is_active,
+            })
+            .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
+            JOIN users u ON u.did = d.controller_did
+            WHERE d.delegated_did = $1
+              AND d.revoked_at IS NULL
+              AND 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 has_any_controllers(&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 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)?;
+
+        Ok(rows
+            .into_iter()
+            .map(|r| AuditLogEntry {
+                id: r.id,
+                delegated_did: r.delegated_did.into(),
+                actor_did: r.actor_did.into(),
+                controller_did: r.controller_did.map(Into::into),
+                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 get_audit_log_by_controller(
+        &self,
+        controller_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 controller_did = $1
+            ORDER BY created_at DESC
+            LIMIT $2 OFFSET $3
+            "#,
+            controller_did.as_str(),
+            limit,
+            offset
+        )
+        .fetch_all(&self.pool)
+        .await
+        .map_err(map_sqlx_error)?;
+
+        Ok(rows
+            .into_iter()
+            .map(|r| AuditLogEntry {
+                id: r.id,
+                delegated_did: r.delegated_did.into(),
+                actor_did: r.actor_did.into(),
+                controller_did: r.controller_did.map(Into::into),
+                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)
+    }
+}

crates/tranquil-db/src/postgres/event_notifier.rs

@@ -0,0 +1,41 @@
+use async_trait::async_trait;
+use sqlx::postgres::PgListener;
+use sqlx::PgPool;
+use tranquil_db_traits::{DbError, RepoEventNotifier, RepoEventReceiver};
+
+use super::user::map_sqlx_error;
+
+pub struct PostgresRepoEventNotifier {
+    pool: PgPool,
+}
+
+impl PostgresRepoEventNotifier {
+    pub fn new(pool: PgPool) -> Self {
+        Self { pool }
+    }
+}
+
+#[async_trait]
+impl RepoEventNotifier for PostgresRepoEventNotifier {
+    async fn subscribe(&self) -> Result<Box<dyn RepoEventReceiver>, DbError> {
+        let mut listener = PgListener::connect_with(&self.pool)
+            .await
+            .map_err(map_sqlx_error)?;
+        listener.listen("repo_updates").await.map_err(map_sqlx_error)?;
+        Ok(Box::new(PostgresRepoEventReceiver { listener }))
+    }
+}
+
+pub struct PostgresRepoEventReceiver {
+    listener: PgListener,
+}
+
+#[async_trait]
+impl RepoEventReceiver for PostgresRepoEventReceiver {
+    async fn recv(&mut self) -> Option<i64> {
+        match self.listener.recv().await {
+            Ok(notification) => notification.payload().parse().ok(),
+            Err(_) => None,
+        }
+    }
+}

crates/tranquil-db/src/postgres/mod.rs

@@ -0,0 +1,60 @@
+mod backlink;
+mod backup;
+mod blob;
+mod delegation;
+mod event_notifier;
+mod infra;
+mod oauth;
+mod repo;
+mod session;
+mod user;
+
+use sqlx::PgPool;
+use std::sync::Arc;
+
+pub use backlink::PostgresBacklinkRepository;
+pub use backup::PostgresBackupRepository;
+pub use blob::PostgresBlobRepository;
+pub use delegation::PostgresDelegationRepository;
+pub use event_notifier::PostgresRepoEventNotifier;
+pub use infra::PostgresInfraRepository;
+pub use oauth::PostgresOAuthRepository;
+pub use repo::PostgresRepoRepository;
+pub use session::PostgresSessionRepository;
+pub use user::PostgresUserRepository;
+use tranquil_db_traits::{
+    BacklinkRepository, BackupRepository, BlobRepository, DelegationRepository, InfraRepository,
+    OAuthRepository, RepoEventNotifier, RepoRepository, SessionRepository, UserRepository,
+};
+
+pub struct PostgresRepositories {
+    pub pool: PgPool,
+    pub user: Arc<dyn UserRepository>,
+    pub oauth: Arc<dyn OAuthRepository>,
+    pub session: Arc<dyn SessionRepository>,
+    pub delegation: Arc<dyn DelegationRepository>,
+    pub repo: Arc<dyn RepoRepository>,
+    pub blob: Arc<dyn BlobRepository>,
+    pub infra: Arc<dyn InfraRepository>,
+    pub backup: Arc<dyn BackupRepository>,
+    pub backlink: Arc<dyn BacklinkRepository>,
+    pub event_notifier: Arc<dyn RepoEventNotifier>,
+}
+
+impl PostgresRepositories {
+    pub fn new(pool: PgPool) -> Self {
+        Self {
+            pool: pool.clone(),
+            user: Arc::new(PostgresUserRepository::new(pool.clone())),
+            oauth: Arc::new(PostgresOAuthRepository::new(pool.clone())),
+            session: Arc::new(PostgresSessionRepository::new(pool.clone())),
+            delegation: Arc::new(PostgresDelegationRepository::new(pool.clone())),
+            repo: Arc::new(PostgresRepoRepository::new(pool.clone())),
+            blob: Arc::new(PostgresBlobRepository::new(pool.clone())),
+            infra: Arc::new(PostgresInfraRepository::new(pool.clone())),
+            backup: Arc::new(PostgresBackupRepository::new(pool.clone())),
+            backlink: Arc::new(PostgresBacklinkRepository::new(pool.clone())),
+            event_notifier: Arc::new(PostgresRepoEventNotifier::new(pool)),
+        }
+    }
+}

crates/tranquil-db/src/postgres/session.rs

@@ -0,0 +1,567 @@
+use async_trait::async_trait;
+use chrono::{DateTime, Utc};
+use sqlx::PgPool;
+use tranquil_db_traits::{
+    AppPasswordCreate, AppPasswordRecord, DbError, RefreshSessionResult, SessionForRefresh,
+    SessionListItem, SessionMfaStatus, SessionRefreshData, SessionRepository, SessionToken,
+    SessionTokenCreate,
+};
+use tranquil_types::Did;
+use uuid::Uuid;
+
+use super::user::map_sqlx_error;
+
+pub struct PostgresSessionRepository {
+    pool: PgPool,
+}
+
+impl PostgresSessionRepository {
+    pub fn new(pool: PgPool) -> Self {
+        Self { pool }
+    }
+}
+
+#[async_trait]
+impl SessionRepository for PostgresSessionRepository {
+    async fn create_session(&self, data: &SessionTokenCreate) -> Result<i32, DbError> {
+        let row = sqlx::query!(
+            r#"
+            INSERT INTO session_tokens
+                (did, access_jti, refresh_jti, access_expires_at, refresh_expires_at,
+                 legacy_login, mfa_verified, scope, controller_did, app_password_name)
+            VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
+            RETURNING id
+            "#,
+            data.did.as_str(),
+            data.access_jti,
+            data.refresh_jti,
+            data.access_expires_at,
+            data.refresh_expires_at,
+            data.legacy_login,
+            data.mfa_verified,
+            data.scope,
+            data.controller_did.as_ref().map(|d| d.as_str()),
+            data.app_password_name
+        )
+        .fetch_one(&self.pool)
+        .await
+        .map_err(map_sqlx_error)?;
+
+        Ok(row.id)
+    }
+
+    async fn get_session_by_access_jti(
+        &self,
+        access_jti: &str,
+    ) -> Result<Option<SessionToken>, DbError> {
+        let row = sqlx::query!(
+            r#"
+            SELECT id, did, access_jti, refresh_jti, access_expires_at, refresh_expires_at,
+                   legacy_login, mfa_verified, scope, controller_did, app_password_name,
+                   created_at, updated_at
+            FROM session_tokens
+            WHERE access_jti = $1
+            "#,
+            access_jti
+        )
+        .fetch_optional(&self.pool)
+        .await
+        .map_err(map_sqlx_error)?;
+
+        Ok(row.map(|r| SessionToken {
+            id: r.id,
+            did: Did::from(r.did),
+            access_jti: r.access_jti,
+            refresh_jti: r.refresh_jti,
+            access_expires_at: r.access_expires_at,
+            refresh_expires_at: r.refresh_expires_at,
+            legacy_login: r.legacy_login,
+            mfa_verified: r.mfa_verified,
+            scope: r.scope,
+            controller_did: r.controller_did.map(Did::from),
+            app_password_name: r.app_password_name,
+            created_at: r.created_at,
+            updated_at: r.updated_at,
+        }))
+    }
+
+    async fn get_session_for_refresh(
+        &self,
+        refresh_jti: &str,
+    ) -> Result<Option<SessionForRefresh>, DbError> {
+        let row = sqlx::query!(
+            r#"
+            SELECT st.id, st.did, st.scope, st.controller_did, k.key_bytes, k.encryption_version
+            FROM session_tokens st
+            JOIN users u ON st.did = u.did
+            JOIN user_keys k ON u.id = k.user_id
+            WHERE st.refresh_jti = $1 AND st.refresh_expires_at > NOW()
+            "#,
+            refresh_jti
+        )
+        .fetch_optional(&self.pool)
+        .await
+        .map_err(map_sqlx_error)?;
+
+        Ok(row.map(|r| SessionForRefresh {
+            id: r.id,
+            did: Did::from(r.did),
+            scope: r.scope,
+            controller_did: r.controller_did.map(Did::from),
+            key_bytes: r.key_bytes,
+            encryption_version: r.encryption_version.unwrap_or(0),
+        }))
+    }
+
+    async fn update_session_tokens(
+        &self,
+        session_id: i32,
+        new_access_jti: &str,
+        new_refresh_jti: &str,
+        new_access_expires_at: DateTime<Utc>,
+        new_refresh_expires_at: DateTime<Utc>,
+    ) -> Result<(), DbError> {
+        sqlx::query!(
+            r#"
+            UPDATE session_tokens
+            SET access_jti = $1, refresh_jti = $2, access_expires_at = $3,
+                refresh_expires_at = $4, updated_at = NOW()
+            WHERE id = $5
+            "#,
+            new_access_jti,
+            new_refresh_jti,
+            new_access_expires_at,
+            new_refresh_expires_at,
+            session_id
+        )
+        .execute(&self.pool)
+        .await
+        .map_err(map_sqlx_error)?;
+
+        Ok(())
+    }
+
+    async fn delete_session_by_access_jti(&self, access_jti: &str) -> Result<u64, DbError> {
+        let result = sqlx::query!(
+            "DELETE FROM session_tokens WHERE access_jti = $1",
+            access_jti
+        )
+        .execute(&self.pool)
+        .await
+        .map_err(map_sqlx_error)?;
+
+        Ok(result.rows_affected())
+    }
+
+    async fn delete_session_by_id(&self, session_id: i32) -> Result<u64, DbError> {
+        let result = sqlx::query!("DELETE FROM session_tokens WHERE id = $1", session_id)
+            .execute(&self.pool)
+            .await
+            .map_err(map_sqlx_error)?;
+
+        Ok(result.rows_affected())
+    }
+
+    async fn delete_sessions_by_did(&self, did: &Did) -> Result<u64, DbError> {
+        let result = sqlx::query!("DELETE FROM session_tokens WHERE did = $1", did.as_str())
+            .execute(&self.pool)
+            .await
+            .map_err(map_sqlx_error)?;
+
+        Ok(result.rows_affected())
+    }
+
+    async fn delete_sessions_by_did_except_jti(
+        &self,
+        did: &Did,
+        except_jti: &str,
+    ) -> Result<u64, DbError> {
+        let result = sqlx::query!(
+            "DELETE FROM session_tokens WHERE did = $1 AND access_jti != $2",
+            did.as_str(),
+            except_jti
+        )
+        .execute(&self.pool)
+        .await
+        .map_err(map_sqlx_error)?;
+
+        Ok(result.rows_affected())
+    }
+
+    async fn list_sessions_by_did(&self, did: &Did) -> Result<Vec<SessionListItem>, DbError> {
+        let rows = sqlx::query!(
+            r#"
+            SELECT id, access_jti, created_at, refresh_expires_at
+            FROM session_tokens
+            WHERE did = $1 AND refresh_expires_at > NOW()
+            ORDER BY created_at DESC
+            "#,
+            did.as_str()
+        )
+        .fetch_all(&self.pool)
+        .await
+        .map_err(map_sqlx_error)?;
+
+        Ok(rows
+            .into_iter()
+            .map(|r| SessionListItem {
+                id: r.id,
+                access_jti: r.access_jti,
+                created_at: r.created_at,
+                refresh_expires_at: r.refresh_expires_at,
+            })
+            .collect())
+    }
+
+    async fn get_session_access_jti_by_id(
+        &self,
+        session_id: i32,
+        did: &Did,
+    ) -> Result<Option<String>, DbError> {
+        let row = sqlx::query_scalar!(
+            "SELECT access_jti FROM session_tokens WHERE id = $1 AND did = $2",
+            session_id,
+            did.as_str()
+        )
+        .fetch_optional(&self.pool)
+        .await
+        .map_err(map_sqlx_error)?;
+
+        Ok(row)
+    }
+
+    async fn delete_sessions_by_app_password(
+        &self,
+        did: &Did,
+        app_password_name: &str,
+    ) -> Result<u64, DbError> {
+        let result = sqlx::query!(
+            "DELETE FROM session_tokens WHERE did = $1 AND app_password_name = $2",
+            did.as_str(),
+            app_password_name
+        )
+        .execute(&self.pool)
+        .await
+        .map_err(map_sqlx_error)?;
+
+        Ok(result.rows_affected())
+    }
+
+    async fn get_session_jtis_by_app_password(
+        &self,
+        did: &Did,
+        app_password_name: &str,
+    ) -> Result<Vec<String>, DbError> {
+        let rows = sqlx::query_scalar!(
+            "SELECT access_jti FROM session_tokens WHERE did = $1 AND app_password_name = $2",
+            did.as_str(),
+            app_password_name
+        )
+        .fetch_all(&self.pool)
+        .await
+        .map_err(map_sqlx_error)?;
+
+        Ok(rows)
+    }
+
+    async fn check_refresh_token_used(&self, refresh_jti: &str) -> Result<Option<i32>, DbError> {
+        let row = sqlx::query_scalar!(
+            "SELECT session_id FROM used_refresh_tokens WHERE refresh_jti = $1",
+            refresh_jti
+        )
+        .fetch_optional(&self.pool)
+        .await
+        .map_err(map_sqlx_error)?;
+
+        Ok(row)
+    }
+
+    async fn mark_refresh_token_used(
+        &self,
+        refresh_jti: &str,
+        session_id: i32,
+    ) -> Result<bool, DbError> {
+        let result = sqlx::query!(
+            r#"
+            INSERT INTO used_refresh_tokens (refresh_jti, session_id)
+            VALUES ($1, $2)
+            ON CONFLICT (refresh_jti) DO NOTHING
+            "#,
+            refresh_jti,
+            session_id
+        )
+        .execute(&self.pool)
+        .await
+        .map_err(map_sqlx_error)?;
+
+        Ok(result.rows_affected() > 0)
+    }
+
+    async fn list_app_passwords(&self, user_id: Uuid) -> Result<Vec<AppPasswordRecord>, DbError> {
+        let rows = sqlx::query!(
+            r#"
+            SELECT id, user_id, name, password_hash, created_at, privileged, scopes, created_by_controller_did
+            FROM app_passwords
+            WHERE user_id = $1
+            ORDER BY created_at DESC
+            "#,
+            user_id
+        )
+        .fetch_all(&self.pool)
+        .await
+        .map_err(map_sqlx_error)?;
+
+        Ok(rows
+            .into_iter()
+            .map(|r| AppPasswordRecord {
+                id: r.id,
+                user_id: r.user_id,
+                name: r.name,
+                password_hash: r.password_hash,
+                created_at: r.created_at,
+                privileged: r.privileged,
+                scopes: r.scopes,
+                created_by_controller_did: r.created_by_controller_did.map(Did::from),
+            })
+            .collect())
+    }
+
+    async fn get_app_passwords_for_login(
+        &self,
+        user_id: Uuid,
+    ) -> Result<Vec<AppPasswordRecord>, DbError> {
+        let rows = sqlx::query!(
+            r#"
+            SELECT id, user_id, name, password_hash, created_at, privileged, scopes, created_by_controller_did
+            FROM app_passwords
+            WHERE user_id = $1
+            ORDER BY created_at DESC
+            LIMIT 20
+            "#,
+            user_id
+        )
+        .fetch_all(&self.pool)
+        .await
+        .map_err(map_sqlx_error)?;
+
+        Ok(rows
+            .into_iter()
+            .map(|r| AppPasswordRecord {
+                id: r.id,
+                user_id: r.user_id,
+                name: r.name,
+                password_hash: r.password_hash,
+                created_at: r.created_at,
+                privileged: r.privileged,
+                scopes: r.scopes,
+                created_by_controller_did: r.created_by_controller_did.map(Did::from),
+            })
+            .collect())
+    }
+
+    async fn get_app_password_by_name(
+        &self,
+        user_id: Uuid,
+        name: &str,
+    ) -> Result<Option<AppPasswordRecord>, DbError> {
+        let row = sqlx::query!(
+            r#"
+            SELECT id, user_id, name, password_hash, created_at, privileged, scopes, created_by_controller_did
+            FROM app_passwords
+            WHERE user_id = $1 AND name = $2
+            "#,
+            user_id,
+            name
+        )
+        .fetch_optional(&self.pool)
+        .await
+        .map_err(map_sqlx_error)?;
+
+        Ok(row.map(|r| AppPasswordRecord {
+            id: r.id,
+            user_id: r.user_id,
+            name: r.name,
+            password_hash: r.password_hash,
+            created_at: r.created_at,
+            privileged: r.privileged,
+            scopes: r.scopes,
+            created_by_controller_did: r.created_by_controller_did.map(Did::from),
+        }))
+    }
+
+    async fn create_app_password(&self, data: &AppPasswordCreate) -> Result<Uuid, DbError> {
+        let row = sqlx::query!(
+            r#"
+            INSERT INTO app_passwords (user_id, name, password_hash, privileged, scopes, created_by_controller_did)
+            VALUES ($1, $2, $3, $4, $5, $6)
+            RETURNING id
+            "#,
+            data.user_id,
+            data.name,
+            data.password_hash,
+            data.privileged,
+            data.scopes,
+            data.created_by_controller_did.as_ref().map(|d| d.as_str())
+        )
+        .fetch_one(&self.pool)
+        .await
+        .map_err(map_sqlx_error)?;
+
+        Ok(row.id)
+    }
+
+    async fn delete_app_password(&self, user_id: Uuid, name: &str) -> Result<u64, DbError> {
+        let result = sqlx::query!(
+            "DELETE FROM app_passwords WHERE user_id = $1 AND name = $2",
+            user_id,
+            name
+        )
+        .execute(&self.pool)
+        .await
+        .map_err(map_sqlx_error)?;
+
+        Ok(result.rows_affected())
+    }
+
+    async fn delete_app_passwords_by_controller(
+        &self,
+        did: &Did,
+        controller_did: &Did,
+    ) -> Result<u64, DbError> {
+        let result = sqlx::query!(
+            r#"DELETE FROM app_passwords
+               WHERE user_id = (SELECT id FROM users WHERE did = $1)
+               AND created_by_controller_did = $2"#,
+            did.as_str(),
+            controller_did.as_str()
+        )
+        .execute(&self.pool)
+        .await
+        .map_err(map_sqlx_error)?;
+
+        Ok(result.rows_affected())
+    }
+
+    async fn get_last_reauth_at(&self, did: &Did) -> Result<Option<DateTime<Utc>>, DbError> {
+        let row = sqlx::query_scalar!(
+            r#"SELECT last_reauth_at FROM session_tokens
+               WHERE did = $1 ORDER BY created_at DESC LIMIT 1"#,
+            did.as_str()
+        )
+        .fetch_optional(&self.pool)
+        .await
+        .map_err(map_sqlx_error)?;
+
+        Ok(row.flatten())
+    }
+
+    async fn update_last_reauth(&self, did: &Did) -> Result<DateTime<Utc>, DbError> {
+        let now = Utc::now();
+        sqlx::query!(
+            "UPDATE session_tokens SET last_reauth_at = $1, mfa_verified = TRUE WHERE did = $2",
+            now,
+            did.as_str()
+        )
+        .execute(&self.pool)
+        .await
+        .map_err(map_sqlx_error)?;
+
+        Ok(now)
+    }
+
+    async fn get_session_mfa_status(&self, did: &Did) -> Result<Option<SessionMfaStatus>, DbError> {
+        let row = sqlx::query!(
+            r#"SELECT legacy_login, mfa_verified, last_reauth_at FROM session_tokens
+               WHERE did = $1 ORDER BY created_at DESC LIMIT 1"#,
+            did.as_str()
+        )
+        .fetch_optional(&self.pool)
+        .await
+        .map_err(map_sqlx_error)?;
+
+        Ok(row.map(|r| SessionMfaStatus {
+            legacy_login: r.legacy_login,
+            mfa_verified: r.mfa_verified,
+            last_reauth_at: r.last_reauth_at,
+        }))
+    }
+
+    async fn update_mfa_verified(&self, did: &Did) -> Result<(), DbError> {
+        sqlx::query!(
+            "UPDATE session_tokens SET mfa_verified = TRUE, last_reauth_at = NOW() WHERE did = $1",
+            did.as_str()
+        )
+        .execute(&self.pool)
+        .await
+        .map_err(map_sqlx_error)?;
+
+        Ok(())
+    }
+
+    async fn get_app_password_hashes_by_did(&self, did: &Did) -> Result<Vec<String>, DbError> {
+        let rows = sqlx::query_scalar!(
+            r#"SELECT ap.password_hash FROM app_passwords ap
+               JOIN users u ON ap.user_id = u.id
+               WHERE u.did = $1"#,
+            did.as_str()
+        )
+        .fetch_all(&self.pool)
+        .await
+        .map_err(map_sqlx_error)?;
+
+        Ok(rows)
+    }
+
+    async fn refresh_session_atomic(
+        &self,
+        data: &SessionRefreshData,
+    ) -> Result<RefreshSessionResult, DbError> {
+        let mut tx = self.pool.begin().await.map_err(map_sqlx_error)?;
+
+        if let Ok(Some(session_id)) = sqlx::query_scalar!(
+            "SELECT session_id FROM used_refresh_tokens WHERE refresh_jti = $1 FOR UPDATE",
+            data.old_refresh_jti
+        )
+        .fetch_optional(&mut *tx)
+        .await
+        {
+            let _ = sqlx::query!("DELETE FROM session_tokens WHERE id = $1", session_id)
+                .execute(&mut *tx)
+                .await;
+            tx.commit().await.map_err(map_sqlx_error)?;
+            return Ok(RefreshSessionResult::TokenAlreadyUsed);
+        }
+
+        let result = sqlx::query!(
+            "INSERT INTO used_refresh_tokens (refresh_jti, session_id) VALUES ($1, $2) ON CONFLICT (refresh_jti) DO NOTHING",
+            data.old_refresh_jti,
+            data.session_id
+        )
+        .execute(&mut *tx)
+        .await
+        .map_err(map_sqlx_error)?;
+
+        if result.rows_affected() == 0 {
+            let _ = sqlx::query!("DELETE FROM session_tokens WHERE id = $1", data.session_id)
+                .execute(&mut *tx)
+                .await;
+            tx.commit().await.map_err(map_sqlx_error)?;
+            return Ok(RefreshSessionResult::ConcurrentRefresh);
+        }
+
+        sqlx::query!(
+            "UPDATE session_tokens SET access_jti = $1, refresh_jti = $2, access_expires_at = $3, refresh_expires_at = $4, updated_at = NOW() WHERE id = $5",
+            data.new_access_jti,
+            data.new_refresh_jti,
+            data.new_access_expires_at,
+            data.new_refresh_expires_at,
+            data.session_id
+        )
+        .execute(&mut *tx)
+        .await
+        .map_err(map_sqlx_error)?;
+
+        tx.commit().await.map_err(map_sqlx_error)?;
+        Ok(RefreshSessionResult::Success)
+    }
+}