prototype account_manager

src/account_manager/helpers/account.rs

···
 //! blacksky-algorithms/rsky is licensed under the Apache License 2.0
 //!
 //! Modified for SQLite backend
-use crate::db::DbConn;
-use crate::schema::pds::account::dsl as AccountSchema;
-use crate::schema::pds::account::table as AccountTable;
-use crate::schema::pds::actor::dsl as ActorSchema;
-use crate::schema::pds::actor::table as ActorTable;
 use anyhow::Result;
 use chrono::DateTime;
 use chrono::offset::Utc as UtcOffset;
-use diesel::dsl::{LeftJoinOn, exists, not};
-use diesel::helper_types::{Eq, IntoBoxed};
-use diesel::pg::Pg;
+use diesel::dsl::{exists, not};
 use diesel::result::{DatabaseErrorKind, Error as DieselError};
 use diesel::*;
-use rsky_common;
 use rsky_common::RFC3339_VARIANT;
 use rsky_lexicon::com::atproto::admin::StatusAttr;
+#[expect(unused_imports)]
+pub(crate) use rsky_pds::account_manager::helpers::account::{
+    AccountStatus, ActorAccount, ActorJoinAccount, AvailabilityFlags, BoxedQuery,
+    FormattedAccountStatus, GetAccountAdminStatusOutput, format_account_status, select_account_qb,
+};
+use rsky_pds::schema::pds::account::dsl as AccountSchema;
+use rsky_pds::schema::pds::actor::dsl as ActorSchema;
 use std::ops::Add;
 use std::time::SystemTime;
 use thiserror::Error;
···
     DieselError(String),
 }
 
-pub struct AvailabilityFlags {
-    pub include_taken_down: Option<bool>,
-    pub include_deactivated: Option<bool>,
-}
-
-#[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
-pub enum AccountStatus {
-    Active,
-    Takendown,
-    Suspended,
-    Deleted,
-    Deactivated,
-    Desynchronized,
-    Throttled,
-}
-
-#[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
-pub struct FormattedAccountStatus {
-    pub active: bool,
-    pub status: Option<AccountStatus>,
-}
-
-#[derive(Debug)]
-pub struct GetAccountAdminStatusOutput {
-    pub takedown: StatusAttr,
-    pub deactivated: StatusAttr,
-}
-
-pub type ActorJoinAccount =
-    LeftJoinOn<ActorTable, AccountTable, Eq<ActorSchema::did, AccountSchema::did>>;
-pub type BoxedQuery<'a> = IntoBoxed<'a, ActorJoinAccount, Pg>;
-
-#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
-pub struct ActorAccount {
-    pub did: String,
-    pub handle: Option<String>,
-    #[serde(rename = "createdAt")]
-    pub created_at: String,
-    #[serde(rename = "takedownRef")]
-    pub takedown_ref: Option<String>,
-    #[serde(rename = "deactivatedAt")]
-    pub deactivated_at: Option<String>,
-    #[serde(rename = "deleteAfter")]
-    pub delete_after: Option<String>,
-    pub email: Option<String>,
-    #[serde(rename = "invitesDisabled")]
-    pub invites_disabled: Option<i16>,
-    #[serde(rename = "emailConfirmedAt")]
-    pub email_confirmed_at: Option<String>,
-}
-
-pub fn select_account_qb(flags: Option<AvailabilityFlags>) -> BoxedQuery<'static> {
-    let AvailabilityFlags {
-        include_taken_down,
-        include_deactivated,
-    } = flags.unwrap_or_else(|| AvailabilityFlags {
-        include_taken_down: Some(false),
-        include_deactivated: Some(false),
-    });
-    let include_taken_down = include_taken_down.unwrap_or(false);
-    let include_deactivated = include_deactivated.unwrap_or(false);
-
-    let mut builder = ActorSchema::actor
-        .left_join(AccountSchema::account.on(ActorSchema::did.eq(AccountSchema::did)))
-        .into_boxed();
-    if !include_taken_down {
-        builder = builder.filter(ActorSchema::takedownRef.is_null());
-    }
-    if !include_deactivated {
-        builder = builder.filter(ActorSchema::deactivatedAt.is_null());
-    }
-    builder
-}
-
 pub async fn get_account(
     _handle_or_did: &str,
     flags: Option<AvailabilityFlags>,
-    db: &DbConn,
+    db: &deadpool_diesel::Pool<
+        deadpool_diesel::Manager<SqliteConnection>,
+        deadpool_diesel::sqlite::Object,
+    >,
 ) -> Result<Option<ActorAccount>> {
     let handle_or_did = _handle_or_did.to_owned();
     let found = db
-        .run(move |conn| {
+        .get()
+        .await?
+        .interact(move |conn| {
             let mut builder = select_account_qb(flags);
             if handle_or_did.starts_with("did:") {
                 builder = builder.filter(ActorSchema::did.eq(handle_or_did));
···
                 })
                 .optional()
         })
-        .await?;
+        .await
+        .expect("Failed to get account")?;
     Ok(found)
 }
 
 pub async fn get_account_by_email(
     _email: &str,
     flags: Option<AvailabilityFlags>,
-    db: &DbConn,
+    db: &deadpool_diesel::Pool<
+        deadpool_diesel::Manager<SqliteConnection>,
+        deadpool_diesel::sqlite::Object,
+    >,
 ) -> Result<Option<ActorAccount>> {
     let email = _email.to_owned();
     let found = db
-        .run(move |conn| {
+        .get()
+        .await?
+        .interact(move |conn| {
             select_account_qb(flags)
                 .select((
                     ActorSchema::did,
···
                 })
                 .optional()
         })
-        .await?;
+        .await
+        .expect("Failed to get account")?;
     Ok(found)
 }
 
···
     did: String,
     handle: String,
     deactivated: Option<bool>,
-    db: &DbConn,
+    db: &deadpool_diesel::Pool<
+        deadpool_diesel::Manager<SqliteConnection>,
+        deadpool_diesel::sqlite::Object,
+    >,
 ) -> Result<()> {
     let system_time = SystemTime::now();
     let dt: DateTime<UtcOffset> = system_time.into();
···
     };
 
     let _: String = db
-        .run(move |conn| {
+        .get()
+        .await?
+        .interact(move |conn| {
             insert_into(ActorSchema::actor)
                 .values((
                     ActorSchema::did.eq(did),
···
                 .returning(ActorSchema::did)
                 .get_result(conn)
         })
-        .await?;
+        .await
+        .expect("Failed to register actor")?;
     Ok(())
 }
 
···
     did: String,
     email: String,
     password: String,
-    db: &DbConn,
+    db: &deadpool_diesel::Pool<
+        deadpool_diesel::Manager<SqliteConnection>,
+        deadpool_diesel::sqlite::Object,
+    >,
 ) -> Result<()> {
     let created_at = rsky_common::now();
 
     // @TODO record recovery key for bring your own recovery key
     let _: String = db
-        .run(move |conn| {
+        .get()
+        .await?
+        .interact(move |conn| {
             insert_into(AccountSchema::account)
                 .values((
                     AccountSchema::did.eq(did),
···
                 .returning(AccountSchema::did)
                 .get_result(conn)
         })
-        .await?;
+        .await
+        .expect("Failed to register account")?;
     Ok(())
 }
 
-pub async fn delete_account(did: &str, db: &DbConn) -> Result<()> {
-    use crate::schema::pds::email_token::dsl as EmailTokenSchema;
-    use crate::schema::pds::refresh_token::dsl as RefreshTokenSchema;
-    use crate::schema::pds::repo_root::dsl as RepoRootSchema;
+pub async fn delete_account(
+    did: &str,
+    db: &deadpool_diesel::Pool<
+        deadpool_diesel::Manager<SqliteConnection>,
+        deadpool_diesel::sqlite::Object,
+    >,
+) -> Result<()> {
+    use rsky_pds::schema::pds::email_token::dsl as EmailTokenSchema;
+    use rsky_pds::schema::pds::refresh_token::dsl as RefreshTokenSchema;
+    use rsky_pds::schema::pds::repo_root::dsl as RepoRootSchema;
 
     let did = did.to_owned();
-    db.run(move |conn| {
-        delete(RepoRootSchema::repo_root)
-            .filter(RepoRootSchema::did.eq(&did))
-            .execute(conn)?;
-        delete(EmailTokenSchema::email_token)
-            .filter(EmailTokenSchema::did.eq(&did))
-            .execute(conn)?;
-        delete(RefreshTokenSchema::refresh_token)
-            .filter(RefreshTokenSchema::did.eq(&did))
-            .execute(conn)?;
-        delete(AccountSchema::account)
-            .filter(AccountSchema::did.eq(&did))
-            .execute(conn)?;
-        delete(ActorSchema::actor)
-            .filter(ActorSchema::did.eq(&did))
-            .execute(conn)
-    })
-    .await?;
+    db.get()
+        .await?
+        .interact(move |conn| {
+            delete(RepoRootSchema::repo_root)
+                .filter(RepoRootSchema::did.eq(&did))
+                .execute(conn)?;
+            delete(EmailTokenSchema::email_token)
+                .filter(EmailTokenSchema::did.eq(&did))
+                .execute(conn)?;
+            delete(RefreshTokenSchema::refresh_token)
+                .filter(RefreshTokenSchema::did.eq(&did))
+                .execute(conn)?;
+            delete(AccountSchema::account)
+                .filter(AccountSchema::did.eq(&did))
+                .execute(conn)?;
+            delete(ActorSchema::actor)
+                .filter(ActorSchema::did.eq(&did))
+                .execute(conn)
+        })
+        .await
+        .expect("Failed to delete account")?;
     Ok(())
 }
 
 pub async fn update_account_takedown_status(
     did: &str,
     takedown: StatusAttr,
-    db: &DbConn,
+    db: &deadpool_diesel::Pool<
+        deadpool_diesel::Manager<SqliteConnection>,
+        deadpool_diesel::sqlite::Object,
+    >,
 ) -> Result<()> {
     let takedown_ref: Option<String> = match takedown.applied {
         true => match takedown.r#ref {
···
         false => None,
     };
     let did = did.to_owned();
-    db.run(move |conn| {
-        update(ActorSchema::actor)
-            .filter(ActorSchema::did.eq(did))
-            .set((ActorSchema::takedownRef.eq(takedown_ref),))
-            .execute(conn)
-    })
-    .await?;
+    db.get()
+        .await?
+        .interact(move |conn| {
+            update(ActorSchema::actor)
+                .filter(ActorSchema::did.eq(did))
+                .set((ActorSchema::takedownRef.eq(takedown_ref),))
+                .execute(conn)
+        })
+        .await
+        .expect("Failed to update account takedown status")?;
     Ok(())
 }
 
 pub async fn deactivate_account(
     did: &str,
     delete_after: Option<String>,
-    db: &DbConn,
+    db: &deadpool_diesel::Pool<
+        deadpool_diesel::Manager<SqliteConnection>,
+        deadpool_diesel::sqlite::Object,
+    >,
 ) -> Result<()> {
     let did = did.to_owned();
-    db.run(move |conn| {
-        update(ActorSchema::actor)
-            .filter(ActorSchema::did.eq(did))
-            .set((
-                ActorSchema::deactivatedAt.eq(rsky_common::now()),
-                ActorSchema::deleteAfter.eq(delete_after),
-            ))
-            .execute(conn)
-    })
-    .await?;
+    db.get()
+        .await?
+        .interact(move |conn| {
+            update(ActorSchema::actor)
+                .filter(ActorSchema::did.eq(did))
+                .set((
+                    ActorSchema::deactivatedAt.eq(rsky_common::now()),
+                    ActorSchema::deleteAfter.eq(delete_after),
+                ))
+                .execute(conn)
+        })
+        .await
+        .expect("Failed to deactivate account")?;
     Ok(())
 }
 
-pub async fn activate_account(did: &str, db: &DbConn) -> Result<()> {
+pub async fn activate_account(
+    did: &str,
+    db: &deadpool_diesel::Pool<
+        deadpool_diesel::Manager<SqliteConnection>,
+        deadpool_diesel::sqlite::Object,
+    >,
+) -> Result<()> {
     let did = did.to_owned();
-    db.run(move |conn| {
-        update(ActorSchema::actor)
-            .filter(ActorSchema::did.eq(did))
-            .set((
-                ActorSchema::deactivatedAt.eq::<Option<String>>(None),
-                ActorSchema::deleteAfter.eq::<Option<String>>(None),
-            ))
-            .execute(conn)
-    })
-    .await?;
+    db.get()
+        .await?
+        .interact(move |conn| {
+            update(ActorSchema::actor)
+                .filter(ActorSchema::did.eq(did))
+                .set((
+                    ActorSchema::deactivatedAt.eq::<Option<String>>(None),
+                    ActorSchema::deleteAfter.eq::<Option<String>>(None),
+                ))
+                .execute(conn)
+        })
+        .await
+        .expect("Failed to activate account")?;
     Ok(())
 }
 
-pub async fn update_email(did: &str, email: &str, db: &DbConn) -> Result<()> {
+pub async fn update_email(
+    did: &str,
+    email: &str,
+    db: &deadpool_diesel::Pool<
+        deadpool_diesel::Manager<SqliteConnection>,
+        deadpool_diesel::sqlite::Object,
+    >,
+) -> Result<()> {
     let did = did.to_owned();
     let email = email.to_owned();
     let res = db
-        .run(move |conn| {
+        .get()
+        .await?
+        .interact(move |conn| {
             update(AccountSchema::account)
                 .filter(AccountSchema::did.eq(did))
                 .set((
···
     }
 }
 
-pub async fn update_handle(did: &str, handle: &str, db: &DbConn) -> Result<()> {
-    use crate::schema::pds::actor;
+pub async fn update_handle(
+    did: &str,
+    handle: &str,
+    db: &deadpool_diesel::Pool<
+        deadpool_diesel::Manager<SqliteConnection>,
+        deadpool_diesel::sqlite::Object,
+    >,
+) -> Result<()> {
+    use rsky_pds::schema::pds::actor;
 
     let actor2 = diesel::alias!(actor as actor2);
 
     let did = did.to_owned();
     let handle = handle.to_owned();
     let res = db
-        .run(move |conn| {
+        .get()
+        .await?
+        .interact(move |conn| {
             update(ActorSchema::actor)
                 .filter(ActorSchema::did.eq(did))
                 .filter(not(exists(actor2.filter(ActorSchema::handle.eq(&handle)))))
                 .set((ActorSchema::handle.eq(&handle),))
                 .execute(conn)
         })
-        .await?;
+        .await
+        .expect("Failed to update handle")?;
 
     if res < 1 {
         return Err(anyhow::Error::new(
···
 pub async fn set_email_confirmed_at(
     did: &str,
     email_confirmed_at: String,
-    db: &DbConn,
+    db: &deadpool_diesel::Pool<
+        deadpool_diesel::Manager<SqliteConnection>,
+        deadpool_diesel::sqlite::Object,
+    >,
 ) -> Result<()> {
     let did = did.to_owned();
-    db.run(move |conn| {
-        update(AccountSchema::account)
-            .filter(AccountSchema::did.eq(did))
-            .set(AccountSchema::emailConfirmedAt.eq(email_confirmed_at))
-            .execute(conn)
-    })
-    .await?;
+    db.get()
+        .await?
+        .interact(move |conn| {
+            update(AccountSchema::account)
+                .filter(AccountSchema::did.eq(did))
+                .set(AccountSchema::emailConfirmedAt.eq(email_confirmed_at))
+                .execute(conn)
+        })
+        .await
+        .expect("Failed to set email confirmed at")?;
     Ok(())
 }
 
 pub async fn get_account_admin_status(
     did: &str,
-    db: &DbConn,
+    db: &deadpool_diesel::Pool<
+        deadpool_diesel::Manager<SqliteConnection>,
+        deadpool_diesel::sqlite::Object,
+    >,
 ) -> Result<Option<GetAccountAdminStatusOutput>> {
     let did = did.to_owned();
     let res: Option<(Option<String>, Option<String>)> = db
-        .run(move |conn| {
+        .get()
+        .await?
+        .interact(move |conn| {
             ActorSchema::actor
                 .filter(ActorSchema::did.eq(did))
                 .select((ActorSchema::takedownRef, ActorSchema::deactivatedAt))
                 .first(conn)
                 .optional()
         })
-        .await?;
+        .await
+        .expect("Failed to get account admin status")?;
     match res {
         None => Ok(None),
         Some(res) => {
···
         }
     }
 }
-
-pub fn format_account_status(account: Option<ActorAccount>) -> FormattedAccountStatus {
-    match account {
-        None => FormattedAccountStatus {
-            active: false,
-            status: Some(AccountStatus::Deleted),
-        },
-        Some(got) if got.takedown_ref.is_some() => FormattedAccountStatus {
-            active: false,
-            status: Some(AccountStatus::Takendown),
-        },
-        Some(got) if got.deactivated_at.is_some() => FormattedAccountStatus {
-            active: false,
-            status: Some(AccountStatus::Deactivated),
-        },
-        _ => FormattedAccountStatus {
-            active: true,
-            status: None,
-        },
-    }
-}

src/account_manager/helpers/auth.rs

···
 //! blacksky-algorithms/rsky is licensed under the Apache License 2.0
 //!
 //! Modified for SQLite backend
-use crate::auth_verifier::AuthScope;
-use crate::db::DbConn;
-use crate::models;
 use anyhow::Result;
 use diesel::*;
-use jwt_simple::prelude::*;
-use rsky_common::time::{MINUTE, from_micros_to_utc};
-use rsky_common::{RFC3339_VARIANT, get_random_str, json_to_b64url};
-use secp256k1::{Keypair, Message, SecretKey};
-use sha2::{Digest, Sha256};
-use std::time::SystemTime;
-use thiserror::Error;
-
-pub struct CreateTokensOpts {
-    pub did: String,
-    pub jwt_key: Keypair,
-    pub service_did: String,
-    pub scope: Option<AuthScope>,
-    pub jti: Option<String>,
-    pub expires_in: Option<Duration>,
-}
-
-pub struct RefreshGracePeriodOpts {
-    pub id: String,
-    pub expires_at: String,
-    pub next_id: String,
-}
-
-pub struct AuthToken {
-    pub scope: AuthScope,
-    pub sub: String,
-    pub exp: Duration,
-}
-
-pub struct RefreshToken {
-    pub scope: AuthScope, // AuthScope::Refresh
-    pub sub: String,
-    pub exp: Duration,
-    pub jti: String,
-}
-
-#[derive(Debug, Deserialize, Serialize, Clone)]
-pub struct ServiceJwtPayload {
-    pub iss: String,
-    pub aud: String,
-    pub exp: Option<u64>,
-    pub lxm: Option<String>,
-    pub jti: Option<String>,
-}
-
-#[derive(Debug, Deserialize, Serialize, Clone)]
-pub struct ServiceJwtHeader {
-    pub typ: String,
-    pub alg: String,
-}
-
-pub struct ServiceJwtParams {
-    pub iss: String,
-    pub aud: String,
-    pub exp: Option<u64>,
-    pub lxm: Option<String>,
-    pub jti: Option<String>,
-    pub keypair: SecretKey,
-}
-
-#[derive(Serialize, Deserialize)]
-pub struct CustomClaimObj {
-    pub scope: String,
-}
-
-#[derive(Error, Debug)]
-pub enum AuthHelperError {
-    #[error("ConcurrentRefreshError")]
-    ConcurrentRefresh,
-}
-
-pub fn create_tokens(opts: CreateTokensOpts) -> Result<(String, String)> {
-    let CreateTokensOpts {
-        did,
-        jwt_key,
-        service_did,
-        scope,
-        jti,
-        expires_in,
-    } = opts;
-    let access_jwt = create_access_token(CreateTokensOpts {
-        did: did.clone(),
-        jwt_key,
-        service_did: service_did.clone(),
-        scope,
-        expires_in,
-        jti: None,
-    })?;
-    let refresh_jwt = create_refresh_token(CreateTokensOpts {
-        did,
-        jwt_key,
-        service_did,
-        jti,
-        expires_in,
-        scope: None,
-    })?;
-    Ok((access_jwt, refresh_jwt))
-}
-
-pub fn create_access_token(opts: CreateTokensOpts) -> Result<String> {
-    let CreateTokensOpts {
-        did,
-        jwt_key,
-        service_did,
-        scope,
-        expires_in,
-        ..
-    } = opts;
-    let scope = scope.unwrap_or(AuthScope::Access);
-    let expires_in = expires_in.unwrap_or_else(|| Duration::from_hours(2));
-    let claims = Claims::with_custom_claims(
-        CustomClaimObj {
-            scope: scope.as_str().to_owned(),
-        },
-        expires_in,
-    )
-    .with_audience(service_did)
-    .with_subject(did);
-    // alg ES256K
-    let key = ES256kKeyPair::from_bytes(jwt_key.secret_bytes().as_slice())?;
-    let token = key.sign(claims)?;
-    Ok(token)
-}
-
-pub fn create_refresh_token(opts: CreateTokensOpts) -> Result<String> {
-    let CreateTokensOpts {
-        did,
-        jwt_key,
-        service_did,
-        jti,
-        expires_in,
-        ..
-    } = opts;
-    let jti = jti.unwrap_or_else(get_random_str);
-    let expires_in = expires_in.unwrap_or_else(|| Duration::from_days(90));
-    let claims = Claims::with_custom_claims(
-        CustomClaimObj {
-            scope: AuthScope::Refresh.as_str().to_owned(),
-        },
-        expires_in,
-    )
-    .with_audience(service_did)
-    .with_subject(did)
-    .with_jwt_id(jti);
-    // alg ES256K
-    let key = ES256kKeyPair::from_bytes(jwt_key.secret_bytes().as_slice())?;
-    let token = key.sign(claims)?;
-    Ok(token)
-}
-
-pub async fn create_service_jwt(params: ServiceJwtParams) -> Result<String> {
-    let ServiceJwtParams {
-        iss, aud, keypair, ..
-    } = params;
-    let now = SystemTime::now()
-        .duration_since(SystemTime::UNIX_EPOCH)
-        .expect("timestamp in micros since UNIX epoch")
-        .as_micros() as usize;
-    let exp = params
-        .exp
-        .unwrap_or(((now + MINUTE as usize) / 1000) as u64);
-    let lxm = params.lxm;
-    let jti = get_random_str();
-    let header = ServiceJwtHeader {
-        typ: "JWT".to_string(),
-        alg: "ES256K".to_string(),
-    };
-    let payload = ServiceJwtPayload {
-        iss,
-        aud,
-        exp: Some(exp),
-        lxm,
-        jti: Some(jti),
-    };
-    let to_sign_str = format!(
-        "{0}.{1}",
-        json_to_b64url(&header)?,
-        json_to_b64url(&payload)?
-    );
-    let hash = Sha256::digest(to_sign_str.clone());
-    let message = Message::from_digest_slice(hash.as_ref())?;
-    let mut sig = keypair.sign_ecdsa(message);
-    // Convert to low-s
-    sig.normalize_s();
-    // ASN.1 encoded per decode_dss_signature
-    let compact_sig = sig.serialize_compact();
-    Ok(format!(
-        "{0}.{1}",
-        to_sign_str,
-        base64_url::encode(&compact_sig).replace("=", "") // Base 64 encode signature bytes
-    ))
-}
-
-// @NOTE unsafe for verification, should only be used w/ direct output from createRefreshToken() or createTokens()
-pub fn decode_refresh_token(jwt: String, jwt_key: Keypair) -> Result<RefreshToken> {
-    let key = ES256kKeyPair::from_bytes(jwt_key.secret_bytes().as_slice())?;
-    let public_key = key.public_key();
-    let claims = public_key.verify_token::<CustomClaimObj>(&jwt, None)?;
-    assert_eq!(
-        claims.custom.scope,
-        AuthScope::Refresh.as_str().to_owned(),
-        "not a refresh token"
-    );
-    Ok(RefreshToken {
-        scope: AuthScope::from_str(&claims.custom.scope)?,
-        sub: claims.subject.unwrap(),
-        exp: claims.expires_at.unwrap(),
-        jti: claims.jwt_id.unwrap(),
-    })
-}
+use rsky_common::time::from_micros_to_utc;
+use rsky_common::{RFC3339_VARIANT, get_random_str};
+#[expect(unused_imports)]
+pub(crate) use rsky_pds::account_manager::helpers::auth::{
+    AuthHelperError, AuthToken, CreateTokensOpts, CustomClaimObj, RefreshGracePeriodOpts,
+    RefreshToken, ServiceJwtHeader, ServiceJwtParams, ServiceJwtPayload, create_access_token,
+    create_refresh_token, create_service_jwt, create_tokens, decode_refresh_token,
+};
+use rsky_pds::models;
 
 pub async fn store_refresh_token(
     payload: RefreshToken,
     app_password_name: Option<String>,
-    db: &DbConn,
+    db: &deadpool_diesel::Pool<
+        deadpool_diesel::Manager<SqliteConnection>,
+        deadpool_diesel::sqlite::Object,
+    >,
 ) -> Result<()> {
-    use crate::schema::pds::refresh_token::dsl as RefreshTokenSchema;
+    use rsky_pds::schema::pds::refresh_token::dsl as RefreshTokenSchema;
 
     let exp = from_micros_to_utc((payload.exp.as_millis() / 1000) as i64);
 
-    db.run(move |conn| {
-        insert_into(RefreshTokenSchema::refresh_token)
-            .values((
-                RefreshTokenSchema::id.eq(payload.jti),
-                RefreshTokenSchema::did.eq(payload.sub),
-                RefreshTokenSchema::appPasswordName.eq(app_password_name),
-                RefreshTokenSchema::expiresAt.eq(format!("{}", exp.format(RFC3339_VARIANT))),
-            ))
-            .on_conflict_do_nothing() // E.g. when re-granting during a refresh grace period
-            .execute(conn)
-    })
-    .await?;
+    db.get()
+        .await?
+        .interact(move |conn| {
+            insert_into(RefreshTokenSchema::refresh_token)
+                .values((
+                    RefreshTokenSchema::id.eq(payload.jti),
+                    RefreshTokenSchema::did.eq(payload.sub),
+                    RefreshTokenSchema::appPasswordName.eq(app_password_name),
+                    RefreshTokenSchema::expiresAt.eq(format!("{}", exp.format(RFC3339_VARIANT))),
+                ))
+                .on_conflict_do_nothing() // E.g. when re-granting during a refresh grace period
+                .execute(conn)
+        })
+        .await
+        .expect("Failed to store refresh token")?;
 
     Ok(())
 }
 
-pub async fn revoke_refresh_token(id: String, db: &DbConn) -> Result<bool> {
-    use crate::schema::pds::refresh_token::dsl as RefreshTokenSchema;
-    db.run(move |conn| {
-        let deleted_rows = delete(RefreshTokenSchema::refresh_token)
-            .filter(RefreshTokenSchema::id.eq(id))
-            .get_results::<models::RefreshToken>(conn)?;
+pub async fn revoke_refresh_token(
+    id: String,
+    db: &deadpool_diesel::Pool<
+        deadpool_diesel::Manager<SqliteConnection>,
+        deadpool_diesel::sqlite::Object,
+    >,
+) -> Result<bool> {
+    use rsky_pds::schema::pds::refresh_token::dsl as RefreshTokenSchema;
+    db.get()
+        .await?
+        .interact(move |conn| {
+            let deleted_rows = delete(RefreshTokenSchema::refresh_token)
+                .filter(RefreshTokenSchema::id.eq(id))
+                .get_results::<models::RefreshToken>(conn)?;
 
-        Ok(!deleted_rows.is_empty())
-    })
-    .await
+            Ok(!deleted_rows.is_empty())
+        })
+        .await
+        .expect("Failed to revoke refresh token")
 }
 
-pub async fn revoke_refresh_tokens_by_did(did: &str, db: &DbConn) -> Result<bool> {
-    use crate::schema::pds::refresh_token::dsl as RefreshTokenSchema;
+pub async fn revoke_refresh_tokens_by_did(
+    did: &str,
+    db: &deadpool_diesel::Pool<
+        deadpool_diesel::Manager<SqliteConnection>,
+        deadpool_diesel::sqlite::Object,
+    >,
+) -> Result<bool> {
+    use rsky_pds::schema::pds::refresh_token::dsl as RefreshTokenSchema;
     let did = did.to_owned();
-    db.run(move |conn| {
-        let deleted_rows = delete(RefreshTokenSchema::refresh_token)
-            .filter(RefreshTokenSchema::did.eq(did))
-            .get_results::<models::RefreshToken>(conn)?;
+    db.get()
+        .await?
+        .interact(move |conn| {
+            let deleted_rows = delete(RefreshTokenSchema::refresh_token)
+                .filter(RefreshTokenSchema::did.eq(did))
+                .get_results::<models::RefreshToken>(conn)?;
 
-        Ok(!deleted_rows.is_empty())
-    })
-    .await
+            Ok(!deleted_rows.is_empty())
+        })
+        .await
+        .expect("Failed to revoke refresh tokens by DID")
 }
 
 pub async fn revoke_app_password_refresh_token(
     did: &str,
     app_pass_name: &str,
-    db: &DbConn,
+    db: &deadpool_diesel::Pool<
+        deadpool_diesel::Manager<SqliteConnection>,
+        deadpool_diesel::sqlite::Object,
+    >,
 ) -> Result<bool> {
-    use crate::schema::pds::refresh_token::dsl as RefreshTokenSchema;
+    use rsky_pds::schema::pds::refresh_token::dsl as RefreshTokenSchema;
 
     let did = did.to_owned();
     let app_pass_name = app_pass_name.to_owned();
-    db.run(move |conn| {
-        let deleted_rows = delete(RefreshTokenSchema::refresh_token)
-            .filter(RefreshTokenSchema::did.eq(did))
-            .filter(RefreshTokenSchema::appPasswordName.eq(app_pass_name))
-            .get_results::<models::RefreshToken>(conn)?;
+    db.get()
+        .await?
+        .interact(move |conn| {
+            let deleted_rows = delete(RefreshTokenSchema::refresh_token)
+                .filter(RefreshTokenSchema::did.eq(did))
+                .filter(RefreshTokenSchema::appPasswordName.eq(app_pass_name))
+                .get_results::<models::RefreshToken>(conn)?;
 
-        Ok(!deleted_rows.is_empty())
-    })
-    .await
+            Ok(!deleted_rows.is_empty())
+        })
+        .await
+        .expect("Failed to revoke app password refresh token")
 }
 
-pub async fn get_refresh_token(id: &str, db: &DbConn) -> Result<Option<models::RefreshToken>> {
-    use crate::schema::pds::refresh_token::dsl as RefreshTokenSchema;
+pub async fn get_refresh_token(
+    id: &str,
+    db: &deadpool_diesel::Pool<
+        deadpool_diesel::Manager<SqliteConnection>,
+        deadpool_diesel::sqlite::Object,
+    >,
+) -> Result<Option<models::RefreshToken>> {
+    use rsky_pds::schema::pds::refresh_token::dsl as RefreshTokenSchema;
     let id = id.to_owned();
-    db.run(move |conn| {
-        Ok(RefreshTokenSchema::refresh_token
-            .find(id)
-            .first(conn)
-            .optional()?)
-    })
-    .await
+    db.get()
+        .await?
+        .interact(move |conn| {
+            Ok(RefreshTokenSchema::refresh_token
+                .find(id)
+                .first(conn)
+                .optional()?)
+        })
+        .await
+        .expect("Failed to get refresh token")
 }
 
-pub async fn delete_expired_refresh_tokens(did: &str, now: String, db: &DbConn) -> Result<()> {
-    use crate::schema::pds::refresh_token::dsl as RefreshTokenSchema;
+pub async fn delete_expired_refresh_tokens(
+    did: &str,
+    now: String,
+    db: &deadpool_diesel::Pool<
+        deadpool_diesel::Manager<SqliteConnection>,
+        deadpool_diesel::sqlite::Object,
+    >,
+) -> Result<()> {
+    use rsky_pds::schema::pds::refresh_token::dsl as RefreshTokenSchema;
     let did = did.to_owned();
 
-    db.run(move |conn| {
-        delete(RefreshTokenSchema::refresh_token)
-            .filter(RefreshTokenSchema::did.eq(did))
-            .filter(RefreshTokenSchema::expiresAt.le(now))
-            .execute(conn)?;
-        Ok(())
-    })
-    .await
+    db.get()
+        .await?
+        .interact(move |conn| {
+            delete(RefreshTokenSchema::refresh_token)
+                .filter(RefreshTokenSchema::did.eq(did))
+                .filter(RefreshTokenSchema::expiresAt.le(now))
+                .execute(conn)?;
+            Ok(())
+        })
+        .await
+        .expect("Failed to delete expired refresh tokens")
 }
 
-pub async fn add_refresh_grace_period(opts: RefreshGracePeriodOpts, db: &DbConn) -> Result<()> {
-    db.run(move |conn| {
-        let RefreshGracePeriodOpts {
-            id,
-            expires_at,
-            next_id,
-        } = opts;
-        use crate::schema::pds::refresh_token::dsl as RefreshTokenSchema;
+pub async fn add_refresh_grace_period(
+    opts: RefreshGracePeriodOpts,
+    db: &deadpool_diesel::Pool<
+        deadpool_diesel::Manager<SqliteConnection>,
+        deadpool_diesel::sqlite::Object,
+    >,
+) -> Result<()> {
+    db.get()
+        .await?
+        .interact(move |conn| {
+            let RefreshGracePeriodOpts {
+                id,
+                expires_at,
+                next_id,
+            } = opts;
+            use rsky_pds::schema::pds::refresh_token::dsl as RefreshTokenSchema;
 
-        update(RefreshTokenSchema::refresh_token)
-            .filter(RefreshTokenSchema::id.eq(id))
-            .filter(
-                RefreshTokenSchema::nextId
-                    .is_null()
-                    .or(RefreshTokenSchema::nextId.eq(&next_id)),
-            )
-            .set((
-                RefreshTokenSchema::expiresAt.eq(expires_at),
-                RefreshTokenSchema::nextId.eq(&next_id),
-            ))
-            .returning(models::RefreshToken::as_select())
-            .get_results(conn)
-            .map_err(|error| {
-                anyhow::Error::new(AuthHelperError::ConcurrentRefresh).context(error)
-            })?;
-        Ok(())
-    })
-    .await
+            update(RefreshTokenSchema::refresh_token)
+                .filter(RefreshTokenSchema::id.eq(id))
+                .filter(
+                    RefreshTokenSchema::nextId
+                        .is_null()
+                        .or(RefreshTokenSchema::nextId.eq(&next_id)),
+                )
+                .set((
+                    RefreshTokenSchema::expiresAt.eq(expires_at),
+                    RefreshTokenSchema::nextId.eq(&next_id),
+                ))
+                .returning(models::RefreshToken::as_select())
+                .get_results(conn)
+                .map_err(|error| {
+                    anyhow::Error::new(AuthHelperError::ConcurrentRefresh).context(error)
+                })?;
+            Ok(())
+        })
+        .await
+        .expect("Failed to add refresh grace period")
 }
 
 pub fn get_refresh_token_id() -> String {

src/account_manager/helpers/email_token.rs

···
 //! blacksky-algorithms/rsky is licensed under the Apache License 2.0
 //!
 //! Modified for SQLite backend
-use crate::apis::com::atproto::server::get_random_token;
-use crate::db::DbConn;
-use crate::models::EmailToken;
-use crate::models::models::EmailTokenPurpose;
 use anyhow::{Result, bail};
 use diesel::*;
-use rsky_common;
 use rsky_common::time::{MINUTE, from_str_to_utc, less_than_ago_s};
+use rsky_pds::apis::com::atproto::server::get_random_token;
+use rsky_pds::models::EmailToken;
+use rsky_pds::models::models::EmailTokenPurpose;
 
 pub async fn create_email_token(
     did: &str,
     purpose: EmailTokenPurpose,
-    db: &DbConn,
+    db: &deadpool_diesel::Pool<
+        deadpool_diesel::Manager<SqliteConnection>,
+        deadpool_diesel::sqlite::Object,
+    >,
 ) -> Result<String> {
-    use crate::schema::pds::email_token::dsl as EmailTokenSchema;
+    use rsky_pds::schema::pds::email_token::dsl as EmailTokenSchema;
     let token = get_random_token().to_uppercase();
     let now = rsky_common::now();
 
     let did = did.to_owned();
-    db.run(move |conn| {
-        insert_into(EmailTokenSchema::email_token)
-            .values((
-                EmailTokenSchema::purpose.eq(purpose),
-                EmailTokenSchema::did.eq(did),
-                EmailTokenSchema::token.eq(&token),
-                EmailTokenSchema::requestedAt.eq(&now),
-            ))
-            .on_conflict((EmailTokenSchema::purpose, EmailTokenSchema::did))
-            .do_update()
-            .set((
-                EmailTokenSchema::token.eq(&token),
-                EmailTokenSchema::requestedAt.eq(&now),
-            ))
-            .execute(conn)?;
-        Ok(token)
-    })
-    .await
+    db.get()
+        .await?
+        .interact(move |conn| {
+            insert_into(EmailTokenSchema::email_token)
+                .values((
+                    EmailTokenSchema::purpose.eq(purpose),
+                    EmailTokenSchema::did.eq(did),
+                    EmailTokenSchema::token.eq(&token),
+                    EmailTokenSchema::requestedAt.eq(&now),
+                ))
+                .on_conflict((EmailTokenSchema::purpose, EmailTokenSchema::did))
+                .do_update()
+                .set((
+                    EmailTokenSchema::token.eq(&token),
+                    EmailTokenSchema::requestedAt.eq(&now),
+                ))
+                .execute(conn)?;
+            Ok(token)
+        })
+        .await
+        .expect("Failed to create email token")
 }
 
 pub async fn assert_valid_token(
···
     purpose: EmailTokenPurpose,
     token: &str,
     expiration_len: Option<i32>,
-    db: &DbConn,
+    db: &deadpool_diesel::Pool<
+        deadpool_diesel::Manager<SqliteConnection>,
+        deadpool_diesel::sqlite::Object,
+    >,
 ) -> Result<()> {
     let expiration_len = expiration_len.unwrap_or(MINUTE * 15);
-    use crate::schema::pds::email_token::dsl as EmailTokenSchema;
+    use rsky_pds::schema::pds::email_token::dsl as EmailTokenSchema;
 
     let did = did.to_owned();
     let token = token.to_owned();
     let res = db
-        .run(move |conn| {
+        .get()
+        .await?
+        .interact(move |conn| {
             EmailTokenSchema::email_token
                 .filter(EmailTokenSchema::purpose.eq(purpose))
                 .filter(EmailTokenSchema::did.eq(did))
···
                 .first(conn)
                 .optional()
         })
-        .await?;
+        .await
+        .expect("Failed to assert token")?;
     if let Some(res) = res {
         let requested_at = from_str_to_utc(&res.requested_at);
         let expired = !less_than_ago_s(requested_at, expiration_len);
···
     purpose: EmailTokenPurpose,
     token: &str,
     expiration_len: Option<i32>,
-    db: &DbConn,
+    db: &deadpool_diesel::Pool<
+        deadpool_diesel::Manager<SqliteConnection>,
+        deadpool_diesel::sqlite::Object,
+    >,
 ) -> Result<String> {
     let expiration_len = expiration_len.unwrap_or(MINUTE * 15);
-    use crate::schema::pds::email_token::dsl as EmailTokenSchema;
+    use rsky_pds::schema::pds::email_token::dsl as EmailTokenSchema;
 
     let token = token.to_owned();
     let res = db
-        .run(move |conn| {
+        .get()
+        .await?
+        .interact(move |conn| {
             EmailTokenSchema::email_token
                 .filter(EmailTokenSchema::purpose.eq(purpose))
                 .filter(EmailTokenSchema::token.eq(token.to_uppercase()))
···
                 .first(conn)
                 .optional()
         })
-        .await?;
+        .await
+        .expect("Failed to assert token")?;
     if let Some(res) = res {
         let requested_at = from_str_to_utc(&res.requested_at);
         let expired = !less_than_ago_s(requested_at, expiration_len);
···
     }
 }
 
-pub async fn delete_email_token(did: &str, purpose: EmailTokenPurpose, db: &DbConn) -> Result<()> {
-    use crate::schema::pds::email_token::dsl as EmailTokenSchema;
+pub async fn delete_email_token(
+    did: &str,
+    purpose: EmailTokenPurpose,
+    db: &deadpool_diesel::Pool<
+        deadpool_diesel::Manager<SqliteConnection>,
+        deadpool_diesel::sqlite::Object,
+    >,
+) -> Result<()> {
+    use rsky_pds::schema::pds::email_token::dsl as EmailTokenSchema;
     let did = did.to_owned();
-    db.run(move |conn| {
-        delete(EmailTokenSchema::email_token)
-            .filter(EmailTokenSchema::did.eq(did))
-            .filter(EmailTokenSchema::purpose.eq(purpose))
-            .execute(conn)
-    })
-    .await?;
+    db.get()
+        .await?
+        .interact(move |conn| {
+            delete(EmailTokenSchema::email_token)
+                .filter(EmailTokenSchema::did.eq(did))
+                .filter(EmailTokenSchema::purpose.eq(purpose))
+                .execute(conn)
+        })
+        .await
+        .expect("Failed to delete token")?;
     Ok(())
 }
 
-pub async fn delete_all_email_tokens(did: &str, db: &DbConn) -> Result<()> {
-    use crate::schema::pds::email_token::dsl as EmailTokenSchema;
+pub async fn delete_all_email_tokens(
+    did: &str,
+    db: &deadpool_diesel::Pool<
+        deadpool_diesel::Manager<SqliteConnection>,
+        deadpool_diesel::sqlite::Object,
+    >,
+) -> Result<()> {
+    use rsky_pds::schema::pds::email_token::dsl as EmailTokenSchema;
 
     let did = did.to_owned();
-    db.run(move |conn| {
-        delete(EmailTokenSchema::email_token)
-            .filter(EmailTokenSchema::did.eq(did))
-            .execute(conn)
-    })
-    .await?;
+    db.get()
+        .await?
+        .interact(move |conn| {
+            delete(EmailTokenSchema::email_token)
+                .filter(EmailTokenSchema::did.eq(did))
+                .execute(conn)
+        })
+        .await
+        .expect("Failed to delete all tokens")?;
 
     Ok(())
 }

src/account_manager/helpers/invite.rs

···
 //! blacksky-algorithms/rsky is licensed under the Apache License 2.0
 //!
 //! Modified for SQLite backend
-use crate::account_manager::DisableInviteCodesOpts;
-use crate::db::DbConn;
-use crate::models::models;
 use anyhow::{Result, bail};
 use diesel::*;
-use rsky_common;
 use rsky_lexicon::com::atproto::server::AccountCodes;
 use rsky_lexicon::com::atproto::server::{
     InviteCode as LexiconInviteCode, InviteCodeUse as LexiconInviteCodeUse,
 };
+use rsky_pds::account_manager::DisableInviteCodesOpts;
+use rsky_pds::models::models;
 use std::collections::BTreeMap;
 use std::mem;
 
 pub type CodeUse = LexiconInviteCodeUse;
 pub type CodeDetail = LexiconInviteCode;
 
-pub async fn ensure_invite_is_available(invite_code: String, db: &DbConn) -> Result<()> {
-    use crate::schema::pds::actor::dsl as ActorSchema;
-    use crate::schema::pds::invite_code::dsl as InviteCodeSchema;
-    use crate::schema::pds::invite_code_use::dsl as InviteCodeUseSchema;
+pub async fn ensure_invite_is_available(
+    invite_code: String,
+    db: &deadpool_diesel::Pool<
+        deadpool_diesel::Manager<SqliteConnection>,
+        deadpool_diesel::sqlite::Object,
+    >,
+) -> Result<()> {
+    use rsky_pds::schema::pds::actor::dsl as ActorSchema;
+    use rsky_pds::schema::pds::invite_code::dsl as InviteCodeSchema;
+    use rsky_pds::schema::pds::invite_code_use::dsl as InviteCodeUseSchema;
 
-    db.run(move |conn| {
+    db.get().await?.interact(move |conn| {
         let invite: Option<models::InviteCode> = InviteCodeSchema::invite_code
             .left_join(
                 ActorSchema::actor.on(InviteCodeSchema::forAccount
···
             bail!("InvalidInviteCode: Not enough uses. Provided invite code not available `{invite_code:?}`")
         }
         Ok(())
-    }).await?;
+    }).await.expect("Failed to check invite code availability")?;
 
     Ok(())
 }
···
     did: String,
     invite_code: Option<String>,
     now: String,
-    db: &DbConn,
+    db: &deadpool_diesel::Pool<
+        deadpool_diesel::Manager<SqliteConnection>,
+        deadpool_diesel::sqlite::Object,
+    >,
 ) -> Result<()> {
     if let Some(invite_code) = invite_code {
-        use crate::schema::pds::invite_code_use::dsl as InviteCodeUseSchema;
+        use rsky_pds::schema::pds::invite_code_use::dsl as InviteCodeUseSchema;
 
-        db.run(move |conn| {
-            insert_into(InviteCodeUseSchema::invite_code_use)
-                .values((
-                    InviteCodeUseSchema::code.eq(invite_code),
-                    InviteCodeUseSchema::usedBy.eq(did),
-                    InviteCodeUseSchema::usedAt.eq(now),
-                ))
-                .execute(conn)
-        })
-        .await?;
+        db.get()
+            .await?
+            .interact(move |conn| {
+                insert_into(InviteCodeUseSchema::invite_code_use)
+                    .values((
+                        InviteCodeUseSchema::code.eq(invite_code),
+                        InviteCodeUseSchema::usedBy.eq(did),
+                        InviteCodeUseSchema::usedAt.eq(now),
+                    ))
+                    .execute(conn)
+            })
+            .await
+            .expect("Failed to record invite code use")?;
     }
     Ok(())
 }
···
 pub async fn create_invite_codes(
     to_create: Vec<AccountCodes>,
     use_count: i32,
-    db: &DbConn,
+    db: &deadpool_diesel::Pool<
+        deadpool_diesel::Manager<SqliteConnection>,
+        deadpool_diesel::sqlite::Object,
+    >,
 ) -> Result<()> {
-    use crate::schema::pds::invite_code::dsl as InviteCodeSchema;
+    use rsky_pds::schema::pds::invite_code::dsl as InviteCodeSchema;
     let created_at = rsky_common::now();
 
-    db.run(move |conn| {
-        let rows: Vec<models::InviteCode> = to_create
-            .into_iter()
-            .flat_map(|account| {
-                let for_account = account.account;
-                account
-                    .codes
-                    .iter()
-                    .map(|code| models::InviteCode {
-                        code: code.clone(),
-                        available_uses: use_count,
-                        disabled: 0,
-                        for_account: for_account.clone(),
-                        created_by: "admin".to_owned(),
-                        created_at: created_at.clone(),
-                    })
-                    .collect::<Vec<models::InviteCode>>()
-            })
-            .collect();
-        insert_into(InviteCodeSchema::invite_code)
-            .values(&rows)
-            .execute(conn)
-    })
-    .await?;
+    db.get()
+        .await?
+        .interact(move |conn| {
+            let rows: Vec<models::InviteCode> = to_create
+                .into_iter()
+                .flat_map(|account| {
+                    let for_account = account.account;
+                    account
+                        .codes
+                        .iter()
+                        .map(|code| models::InviteCode {
+                            code: code.clone(),
+                            available_uses: use_count,
+                            disabled: 0,
+                            for_account: for_account.clone(),
+                            created_by: "admin".to_owned(),
+                            created_at: created_at.clone(),
+                        })
+                        .collect::<Vec<models::InviteCode>>()
+                })
+                .collect();
+            insert_into(InviteCodeSchema::invite_code)
+                .values(&rows)
+                .execute(conn)
+        })
+        .await
+        .expect("Failed to create invite codes")?;
     Ok(())
 }
 
···
     codes: Vec<String>,
     expected_total: usize,
     disabled: bool,
-    db: &DbConn,
+    db: &deadpool_diesel::Pool<
+        deadpool_diesel::Manager<SqliteConnection>,
+        deadpool_diesel::sqlite::Object,
+    >,
 ) -> Result<Vec<CodeDetail>> {
-    use crate::schema::pds::invite_code::dsl as InviteCodeSchema;
+    use rsky_pds::schema::pds::invite_code::dsl as InviteCodeSchema;
 
     let for_account = for_account.to_owned();
     let rows = db
-        .run(move |conn| {
+        .get()
+        .await?
+        .interact(move |conn| {
             let now = rsky_common::now();
 
             let rows: Vec<models::InviteCode> = codes
···
                 uses: Vec::new(),
             }))
         })
-        .await?;
+        .await
+        .expect("Failed to create account invite codes")?;
     Ok(rows.collect())
 }
 
-pub async fn get_account_invite_codes(did: &str, db: &DbConn) -> Result<Vec<CodeDetail>> {
-    use crate::schema::pds::invite_code::dsl as InviteCodeSchema;
+pub async fn get_account_invite_codes(
+    did: &str,
+    db: &deadpool_diesel::Pool<
+        deadpool_diesel::Manager<SqliteConnection>,
+        deadpool_diesel::sqlite::Object,
+    >,
+) -> Result<Vec<CodeDetail>> {
+    use rsky_pds::schema::pds::invite_code::dsl as InviteCodeSchema;
 
     let did = did.to_owned();
     let res: Vec<models::InviteCode> = db
-        .run(move |conn| {
+        .get()
+        .await?
+        .interact(move |conn| {
             InviteCodeSchema::invite_code
                 .filter(InviteCodeSchema::forAccount.eq(did))
                 .select(models::InviteCode::as_select())
                 .get_results(conn)
         })
-        .await?;
+        .await
+        .expect("Failed to get account invite codes")?;
 
     let codes: Vec<String> = res.iter().map(|row| row.code.clone()).collect();
     let mut uses = get_invite_codes_uses_v2(codes, db).await?;
···
 
 pub async fn get_invite_codes_uses_v2(
     codes: Vec<String>,
-    db: &DbConn,
+    db: &deadpool_diesel::Pool<
+        deadpool_diesel::Manager<SqliteConnection>,
+        deadpool_diesel::sqlite::Object,
+    >,
 ) -> Result<BTreeMap<String, Vec<CodeUse>>> {
-    use crate::schema::pds::invite_code_use::dsl as InviteCodeUseSchema;
+    use rsky_pds::schema::pds::invite_code_use::dsl as InviteCodeUseSchema;
 
     let mut uses: BTreeMap<String, Vec<CodeUse>> = BTreeMap::new();
     if !codes.is_empty() {
         let uses_res: Vec<models::InviteCodeUse> = db
-            .run(|conn| {
+            .get()
+            .await?
+            .interact(|conn| {
                 InviteCodeUseSchema::invite_code_use
                     .filter(InviteCodeUseSchema::code.eq_any(codes))
                     .order_by(InviteCodeUseSchema::usedAt.desc())
                     .select(models::InviteCodeUse::as_select())
                     .get_results(conn)
             })
-            .await?;
+            .await
+            .expect("Failed to get invite code uses")?;
         for invite_code_use in uses_res {
             let models::InviteCodeUse {
                 code,
···
 
 pub async fn get_invited_by_for_accounts(
     dids: Vec<String>,
-    db: &DbConn,
+    db: &deadpool_diesel::Pool<
+        deadpool_diesel::Manager<SqliteConnection>,
+        deadpool_diesel::sqlite::Object,
+    >,
 ) -> Result<BTreeMap<String, CodeDetail>> {
     if dids.is_empty() {
         return Ok(BTreeMap::new());
     }
-    use crate::schema::pds::invite_code::dsl as InviteCodeSchema;
-    use crate::schema::pds::invite_code_use::dsl as InviteCodeUseSchema;
+    use rsky_pds::schema::pds::invite_code::dsl as InviteCodeSchema;
+    use rsky_pds::schema::pds::invite_code_use::dsl as InviteCodeUseSchema;
 
     let dids = dids.clone();
     let res: Vec<models::InviteCode> = db
-        .run(|conn| {
+        .get()
+        .await?
+        .interact(|conn| {
             InviteCodeSchema::invite_code
                 .filter(
                     InviteCodeSchema::forAccount.eq_any(
···
                 .select(models::InviteCode::as_select())
                 .get_results(conn)
         })
-        .await?;
+        .await
+        .expect("Failed to get account invite codes")?;
     let codes: Vec<String> = res.iter().map(|row| row.code.clone()).collect();
     let mut uses = get_invite_codes_uses_v2(codes, db).await?;
 
···
     ))
 }
 
-pub async fn set_account_invites_disabled(did: &str, disabled: bool, db: &DbConn) -> Result<()> {
-    use crate::schema::pds::account::dsl as AccountSchema;
+pub async fn set_account_invites_disabled(
+    did: &str,
+    disabled: bool,
+    db: &deadpool_diesel::Pool<
+        deadpool_diesel::Manager<SqliteConnection>,
+        deadpool_diesel::sqlite::Object,
+    >,
+) -> Result<()> {
+    use rsky_pds::schema::pds::account::dsl as AccountSchema;
 
     let disabled: i16 = if disabled { 1 } else { 0 };
     let did = did.to_owned();
-    db.run(move |conn| {
-        update(AccountSchema::account)
-            .filter(AccountSchema::did.eq(did))
-            .set((AccountSchema::invitesDisabled.eq(disabled),))
-            .execute(conn)
-    })
-    .await?;
+    db.get()
+        .await?
+        .interact(move |conn| {
+            update(AccountSchema::account)
+                .filter(AccountSchema::did.eq(did))
+                .set((AccountSchema::invitesDisabled.eq(disabled),))
+                .execute(conn)
+        })
+        .await
+        .expect("Failed to set account invites disabled")?;
     Ok(())
 }
 
-pub async fn disable_invite_codes(opts: DisableInviteCodesOpts, db: &DbConn) -> Result<()> {
-    use crate::schema::pds::invite_code::dsl as InviteCodeSchema;
+pub async fn disable_invite_codes(
+    opts: DisableInviteCodesOpts,
+    db: &deadpool_diesel::Pool<
+        deadpool_diesel::Manager<SqliteConnection>,
+        deadpool_diesel::sqlite::Object,
+    >,
+) -> Result<()> {
+    use rsky_pds::schema::pds::invite_code::dsl as InviteCodeSchema;
 
     let DisableInviteCodesOpts { codes, accounts } = opts;
     if !codes.is_empty() {
-        db.run(move |conn| {
-            update(InviteCodeSchema::invite_code)
-                .filter(InviteCodeSchema::code.eq_any(&codes))
-                .set((InviteCodeSchema::disabled.eq(1),))
-                .execute(conn)
-        })
-        .await?;
+        db.get()
+            .await?
+            .interact(move |conn| {
+                update(InviteCodeSchema::invite_code)
+                    .filter(InviteCodeSchema::code.eq_any(&codes))
+                    .set((InviteCodeSchema::disabled.eq(1),))
+                    .execute(conn)
+            })
+            .await
+            .expect("Failed to disable invite codes")?;
     }
     if !accounts.is_empty() {
-        db.run(move |conn| {
-            update(InviteCodeSchema::invite_code)
-                .filter(InviteCodeSchema::forAccount.eq_any(&accounts))
-                .set((InviteCodeSchema::disabled.eq(1),))
-                .execute(conn)
-        })
-        .await?;
+        db.get()
+            .await?
+            .interact(move |conn| {
+                update(InviteCodeSchema::invite_code)
+                    .filter(InviteCodeSchema::forAccount.eq_any(&accounts))
+                    .set((InviteCodeSchema::disabled.eq(1),))
+                    .execute(conn)
+            })
+            .await
+            .expect("Failed to disable invite codes")?;
     }
     Ok(())
 }

src/account_manager/helpers/password.rs

···
 //! blacksky-algorithms/rsky is licensed under the Apache License 2.0
 //!
 //! Modified for SQLite backend
-use crate::db::DbConn;
-use crate::models;
-use crate::models::AppPassword;
 use anyhow::{Result, anyhow, bail};
 use argon2::{
     Argon2,
     password_hash::{PasswordHash, PasswordHasher, PasswordVerifier, SaltString, rand_core::OsRng},
 };
-use base64ct::{Base64, Encoding};
+// use base64ct::{Base64, Encoding};
 use diesel::*;
 use rsky_common::{get_random_str, now};
 use rsky_lexicon::com::atproto::server::CreateAppPasswordOutput;
-use sha2::{Digest, Sha256};
+#[expect(unused_imports)]
+pub(crate) use rsky_pds::account_manager::helpers::password::{
+    UpdateUserPasswordOpts, gen_salt_and_hash, hash_app_password, hash_with_salt, verify,
+};
+use rsky_pds::models;
+use rsky_pds::models::AppPassword;
 
-pub struct UpdateUserPasswordOpts {
-    pub did: String,
-    pub password_encrypted: String,
-}
-
-pub async fn verify_account_password(did: &str, password: &String, db: &DbConn) -> Result<bool> {
-    use crate::schema::pds::account::dsl as AccountSchema;
+pub async fn verify_account_password(
+    did: &str,
+    password: &String,
+    db: &deadpool_diesel::Pool<
+        deadpool_diesel::Manager<SqliteConnection>,
+        deadpool_diesel::sqlite::Object,
+    >,
+) -> Result<bool> {
+    use rsky_pds::schema::pds::account::dsl as AccountSchema;
 
     let did = did.to_owned();
     let found = db
-        .run(move |conn| {
+        .get()
+        .await?
+        .interact(move |conn| {
             AccountSchema::account
                 .filter(AccountSchema::did.eq(did))
                 .select(models::Account::as_select())
                 .first(conn)
                 .optional()
         })
-        .await?;
+        .await
+        .expect("Failed to get account")?;
     if let Some(found) = found {
         verify(password, &found.password)
     } else {
···
     }
 }
 
-pub async fn verify_app_password(did: &str, password: &str, db: &DbConn) -> Result<Option<String>> {
-    use crate::schema::pds::app_password::dsl as AppPasswordSchema;
+pub async fn verify_app_password(
+    did: &str,
+    password: &str,
+    db: &deadpool_diesel::Pool<
+        deadpool_diesel::Manager<SqliteConnection>,
+        deadpool_diesel::sqlite::Object,
+    >,
+) -> Result<Option<String>> {
+    use rsky_pds::schema::pds::app_password::dsl as AppPasswordSchema;
 
     let did = did.to_owned();
     let password = password.to_owned();
     let password_encrypted = hash_app_password(&did, &password).await?;
     let found = db
-        .run(move |conn| {
+        .get()
+        .await?
+        .interact(move |conn| {
             AppPasswordSchema::app_password
                 .filter(AppPasswordSchema::did.eq(did))
                 .filter(AppPasswordSchema::password.eq(password_encrypted))
···
                 .first(conn)
                 .optional()
         })
-        .await?;
+        .await
+        .expect("Failed to get app password")?;
     if let Some(found) = found {
         Ok(Some(found.name))
     } else {
···
     }
 }
 
-// We use Argon because it's 3x faster than scrypt.
-pub fn gen_salt_and_hash(password: String) -> Result<String> {
-    let salt = SaltString::generate(&mut OsRng);
-    // Hash password to PHC string
-    let argon2 = Argon2::default();
-    let password_hash = argon2
-        .hash_password(password.as_ref(), &salt)
-        .map_err(|error| anyhow!(error.to_string()))?
-        .to_string();
-    Ok(password_hash)
-}
-
-pub fn hash_with_salt(password: &String, salt: &str) -> Result<String> {
-    let salt = SaltString::from_b64(salt).map_err(|error| anyhow!(error.to_string()))?;
-    let argon2 = Argon2::default();
-    let password_hash = argon2
-        .hash_password(password.as_ref(), &salt)
-        .map_err(|error| anyhow!(error.to_string()))?
-        .to_string();
-    Ok(password_hash)
-}
-
-pub fn verify(password: &String, stored_hash: &str) -> Result<bool> {
-    let parsed_hash = PasswordHash::new(stored_hash).map_err(|error| anyhow!(error.to_string()))?;
-    Ok(Argon2::default()
-        .verify_password(password.as_ref(), &parsed_hash)
-        .is_ok())
-}
-
-pub async fn hash_app_password(did: &String, password: &String) -> Result<String> {
-    let hash = Sha256::digest(did);
-    let salt = Base64::encode_string(&hash).replace("=", "");
-    hash_with_salt(password, &salt)
-}
-
 /// create an app password with format:
 /// 1234-abcd-5678-efgh
 pub async fn create_app_password(
     did: String,
     name: String,
-    db: &DbConn,
+    db: &deadpool_diesel::Pool<
+        deadpool_diesel::Manager<SqliteConnection>,
+        deadpool_diesel::sqlite::Object,
+    >,
 ) -> Result<CreateAppPasswordOutput> {
     let str = &get_random_str()[0..16].to_lowercase();
     let chunks = [&str[0..4], &str[4..8], &str[8..12], &str[12..16]];
     let password = chunks.join("-");
     let password_encrypted = hash_app_password(&did, &password).await?;
 
-    use crate::schema::pds::app_password::dsl as AppPasswordSchema;
+    use rsky_pds::schema::pds::app_password::dsl as AppPasswordSchema;
 
     let created_at = now();
 
-    db.run(move |conn| {
-        let got: Option<AppPassword> = insert_into(AppPasswordSchema::app_password)
-            .values((
-                AppPasswordSchema::did.eq(did),
-                AppPasswordSchema::name.eq(&name),
-                AppPasswordSchema::password.eq(password_encrypted),
-                AppPasswordSchema::createdAt.eq(&created_at),
-            ))
-            .returning(AppPassword::as_select())
-            .get_result(conn)
-            .optional()?;
-        if got.is_some() {
-            Ok(CreateAppPasswordOutput {
-                name,
-                password,
-                created_at,
-            })
-        } else {
-            bail!("could not create app-specific password")
-        }
-    })
-    .await
+    db.get()
+        .await?
+        .interact(move |conn| {
+            let got: Option<AppPassword> = insert_into(AppPasswordSchema::app_password)
+                .values((
+                    AppPasswordSchema::did.eq(did),
+                    AppPasswordSchema::name.eq(&name),
+                    AppPasswordSchema::password.eq(password_encrypted),
+                    AppPasswordSchema::createdAt.eq(&created_at),
+                ))
+                .returning(AppPassword::as_select())
+                .get_result(conn)
+                .optional()?;
+            if got.is_some() {
+                Ok(CreateAppPasswordOutput {
+                    name,
+                    password,
+                    created_at,
+                })
+            } else {
+                bail!("could not create app-specific password")
+            }
+        })
+        .await
+        .expect("Failed to create app password")
 }
 
-pub async fn list_app_passwords(did: &str, db: &DbConn) -> Result<Vec<(String, String)>> {
-    use crate::schema::pds::app_password::dsl as AppPasswordSchema;
+pub async fn list_app_passwords(
+    did: &str,
+    db: &deadpool_diesel::Pool<
+        deadpool_diesel::Manager<SqliteConnection>,
+        deadpool_diesel::sqlite::Object,
+    >,
+) -> Result<Vec<(String, String)>> {
+    use rsky_pds::schema::pds::app_password::dsl as AppPasswordSchema;
 
     let did = did.to_owned();
-    db.run(move |conn| {
-        Ok(AppPasswordSchema::app_password
-            .filter(AppPasswordSchema::did.eq(did))
-            .select((AppPasswordSchema::name, AppPasswordSchema::createdAt))
-            .get_results(conn)?)
-    })
-    .await
+    db.get()
+        .await?
+        .interact(move |conn| {
+            Ok(AppPasswordSchema::app_password
+                .filter(AppPasswordSchema::did.eq(did))
+                .select((AppPasswordSchema::name, AppPasswordSchema::createdAt))
+                .get_results(conn)?)
+        })
+        .await
+        .expect("Failed to list app passwords")
 }
 
-pub async fn update_user_password(opts: UpdateUserPasswordOpts, db: &DbConn) -> Result<()> {
-    use crate::schema::pds::account::dsl as AccountSchema;
+pub async fn update_user_password(
+    opts: UpdateUserPasswordOpts,
+    db: &deadpool_diesel::Pool<
+        deadpool_diesel::Manager<SqliteConnection>,
+        deadpool_diesel::sqlite::Object,
+    >,
+) -> Result<()> {
+    use rsky_pds::schema::pds::account::dsl as AccountSchema;
 
-    db.run(move |conn| {
-        update(AccountSchema::account)
-            .filter(AccountSchema::did.eq(opts.did))
-            .set(AccountSchema::password.eq(opts.password_encrypted))
-            .execute(conn)?;
-        Ok(())
-    })
-    .await
+    db.get()
+        .await?
+        .interact(move |conn| {
+            update(AccountSchema::account)
+                .filter(AccountSchema::did.eq(opts.did))
+                .set(AccountSchema::password.eq(opts.password_encrypted))
+                .execute(conn)?;
+            Ok(())
+        })
+        .await
+        .expect("Failed to update user password")
 }
 
-pub async fn delete_app_password(did: &str, name: &str, db: &DbConn) -> Result<()> {
-    use crate::schema::pds::app_password::dsl as AppPasswordSchema;
+pub async fn delete_app_password(
+    did: &str,
+    name: &str,
+    db: &deadpool_diesel::Pool<
+        deadpool_diesel::Manager<SqliteConnection>,
+        deadpool_diesel::sqlite::Object,
+    >,
+) -> Result<()> {
+    use rsky_pds::schema::pds::app_password::dsl as AppPasswordSchema;
 
     let did = did.to_owned();
     let name = name.to_owned();
-    db.run(move |conn| {
-        delete(AppPasswordSchema::app_password)
-            .filter(AppPasswordSchema::did.eq(did))
-            .filter(AppPasswordSchema::name.eq(name))
-            .execute(conn)?;
-        Ok(())
-    })
-    .await
+    db.get()
+        .await?
+        .interact(move |conn| {
+            delete(AppPasswordSchema::app_password)
+                .filter(AppPasswordSchema::did.eq(did))
+                .filter(AppPasswordSchema::name.eq(name))
+                .execute(conn)?;
+            Ok(())
+        })
+        .await
+        .expect("Failed to delete app password")
 }

src/account_manager/helpers/repo.rs

···
 //! blacksky-algorithms/rsky is licensed under the Apache License 2.0
 //!
 //! Modified for SQLite backend
-use crate::db::DbConn;
 use anyhow::Result;
+use cidv10::Cid;
 use diesel::*;
-use libipld::Cid;
-use rsky_common;
 
-pub async fn update_root(did: String, cid: Cid, rev: String, db: &DbConn) -> Result<()> {
+pub async fn update_root(
+    did: String,
+    cid: Cid,
+    rev: String,
+    db: &deadpool_diesel::Pool<
+        deadpool_diesel::Manager<SqliteConnection>,
+        deadpool_diesel::sqlite::Object,
+    >,
+) -> Result<()> {
     // @TODO balance risk of a race in the case of a long retry
-    use crate::schema::pds::repo_root::dsl as RepoRootSchema;
+    use rsky_pds::schema::pds::repo_root::dsl as RepoRootSchema;
 
     let now = rsky_common::now();
 
-    db.run(move |conn| {
-        insert_into(RepoRootSchema::repo_root)
-            .values((
-                RepoRootSchema::did.eq(did),
-                RepoRootSchema::cid.eq(cid.to_string()),
-                RepoRootSchema::rev.eq(rev.clone()),
-                RepoRootSchema::indexedAt.eq(now),
-            ))
-            .on_conflict(RepoRootSchema::did)
-            .do_update()
-            .set((
-                RepoRootSchema::cid.eq(cid.to_string()),
-                RepoRootSchema::rev.eq(rev),
-            ))
-            .execute(conn)
-    })
-    .await?;
+    db.get()
+        .await?
+        .interact(move |conn| {
+            insert_into(RepoRootSchema::repo_root)
+                .values((
+                    RepoRootSchema::did.eq(did),
+                    RepoRootSchema::cid.eq(cid.to_string()),
+                    RepoRootSchema::rev.eq(rev.clone()),
+                    RepoRootSchema::indexedAt.eq(now),
+                ))
+                .on_conflict(RepoRootSchema::did)
+                .do_update()
+                .set((
+                    RepoRootSchema::cid.eq(cid.to_string()),
+                    RepoRootSchema::rev.eq(rev),
+                ))
+                .execute(conn)
+        })
+        .await
+        .expect("Failed to update repo root")?;
 
     Ok(())
 }

src/account_manager/mod.rs

···
 //! blacksky-algorithms/rsky is licensed under the Apache License 2.0
 //!
 //! Modified for SQLite backend
+use crate::account_manager::helpers::account::{
+    AccountStatus, ActorAccount, AvailabilityFlags, GetAccountAdminStatusOutput,
+};
+use crate::account_manager::helpers::auth::{
+    AuthHelperError, CreateTokensOpts, RefreshGracePeriodOpts,
+};
+use crate::account_manager::helpers::invite::CodeDetail;
+use crate::account_manager::helpers::password::UpdateUserPasswordOpts;
 use anyhow::Result;
 use chrono::DateTime;
 use chrono::offset::Utc as UtcOffset;
 use cidv10::Cid;
+use diesel::*;
 use futures::try_join;
-use helpers::{account, auth, email_token, invite, password};
+use helpers::{account, auth, email_token, invite, password, repo};
 use rsky_common::RFC3339_VARIANT;
 use rsky_common::time::{HOUR, from_micros_to_str, from_str_to_micros};
 use rsky_lexicon::com::atproto::admin::StatusAttr;
 use rsky_lexicon::com::atproto::server::{AccountCodes, CreateAppPasswordOutput};
-use rsky_pds::account_manager::helpers::account::{
-    AccountStatus, ActorAccount, AvailabilityFlags, GetAccountAdminStatusOutput,
-};
-use rsky_pds::account_manager::helpers::auth::{
-    AuthHelperError, CreateTokensOpts, RefreshGracePeriodOpts,
-};
-use rsky_pds::account_manager::helpers::invite::CodeDetail;
-use rsky_pds::account_manager::helpers::password::UpdateUserPasswordOpts;
-use rsky_pds::account_manager::helpers::repo;
 use rsky_pds::account_manager::{
     ConfirmEmailOpts, CreateAccountOpts, DisableInviteCodesOpts, ResetPasswordOpts,
     UpdateAccountPasswordOpts, UpdateEmailOpts,
···
 #[derive(Clone)]
 pub struct AccountManager {
     pub db: deadpool_diesel::Pool<
-        deadpool_diesel::Manager<diesel::SqliteConnection>,
+        deadpool_diesel::Manager<SqliteConnection>,
         deadpool_diesel::sqlite::Object,
     >,
 }
···
 pub type AccountManagerCreator = Box<
     dyn Fn(
             deadpool_diesel::Pool<
-                deadpool_diesel::Manager<diesel::SqliteConnection>,
+                deadpool_diesel::Manager<SqliteConnection>,
                 deadpool_diesel::sqlite::Object,
             >,
         ) -> AccountManager
···
 impl AccountManager {
     pub fn new(
         db: deadpool_diesel::Pool<
-            deadpool_diesel::Manager<diesel::SqliteConnection>,
+            deadpool_diesel::Manager<SqliteConnection>,
             deadpool_diesel::sqlite::Object,
         >,
     ) -> Self {
···
     pub fn creator() -> AccountManagerCreator {
         Box::new(
             move |db: deadpool_diesel::Pool<
-                deadpool_diesel::Manager<diesel::SqliteConnection>,
+                deadpool_diesel::Manager<SqliteConnection>,
                 deadpool_diesel::sqlite::Object,
             >|
                   -> AccountManager { AccountManager::new(db) },
···
         handle_or_did: &str,
         flags: Option<AvailabilityFlags>,
     ) -> Result<Option<ActorAccount>> {
-        let db = self.db.clone();
-        account::get_account(handle_or_did, flags, db.as_ref()).await
+        account::get_account(handle_or_did, flags, &self.db).await
     }
 
     pub async fn get_account_by_email(
···
         email: &str,
         flags: Option<AvailabilityFlags>,
     ) -> Result<Option<ActorAccount>> {
-        let db = self.db.clone();
-        account::get_account_by_email(email, flags, db.as_ref()).await
+        account::get_account_by_email(email, flags, &self.db).await
     }
 
     pub async fn is_account_activated(&self, did: &str) -> Result<bool> {
···
     }
 
     pub async fn create_account(&self, opts: CreateAccountOpts) -> Result<(String, String)> {
-        let db = self.db.clone();
         let CreateAccountOpts {
             did,
             handle,
···
         let now = rsky_common::now();
 
         if let Some(invite_code) = invite_code.clone() {
-            invite::ensure_invite_is_available(invite_code, db.as_ref()).await?;
+            invite::ensure_invite_is_available(invite_code, &self.db).await?;
         }
-        account::register_actor(did.clone(), handle, deactivated, db.as_ref()).await?;
+        account::register_actor(did.clone(), handle, deactivated, &self.db).await?;
         if let (Some(email), Some(password_encrypted)) = (email, password_encrypted) {
-            account::register_account(did.clone(), email, password_encrypted, db.as_ref()).await?;
+            account::register_account(did.clone(), email, password_encrypted, &self.db).await?;
         }
-        invite::record_invite_use(did.clone(), invite_code, now, db.as_ref()).await?;
-        auth::store_refresh_token(refresh_payload, None, db.as_ref()).await?;
-        repo::update_root(did, repo_cid, repo_rev, db.as_ref()).await?;
+        invite::record_invite_use(did.clone(), invite_code, now, &self.db).await?;
+        auth::store_refresh_token(refresh_payload, None, &self.db).await?;
+        repo::update_root(did, repo_cid, repo_rev, &self.db).await?;
         Ok((access_jwt, refresh_jwt))
     }
 
···
         &self,
         did: &str,
     ) -> Result<Option<GetAccountAdminStatusOutput>> {
-        let db = self.db.clone();
-        account::get_account_admin_status(did, db.as_ref()).await
+        account::get_account_admin_status(did, &self.db).await
     }
 
     pub async fn update_repo_root(&self, did: String, cid: Cid, rev: String) -> Result<()> {
-        let db = self.db.clone();
-        repo::update_root(did, cid, rev, db.as_ref()).await
+        repo::update_root(did, cid, rev, &self.db).await
     }
 
     pub async fn delete_account(&self, did: &str) -> Result<()> {
-        let db = self.db.clone();
-        account::delete_account(did, db.as_ref()).await
+        account::delete_account(did, &self.db).await
     }
 
     pub async fn takedown_account(&self, did: &str, takedown: StatusAttr) -> Result<()> {
         (_, _) = try_join!(
-            account::update_account_takedown_status(did, takedown, self.db.as_ref()),
-            auth::revoke_refresh_tokens_by_did(did, self.db.as_ref())
+            account::update_account_takedown_status(did, takedown, &self.db),
+            auth::revoke_refresh_tokens_by_did(did, &self.db)
         )?;
         Ok(())
     }
 
     // @NOTE should always be paired with a sequenceHandle().
     pub async fn update_handle(&self, did: &str, handle: &str) -> Result<()> {
-        let db = self.db.clone();
-        account::update_handle(did, handle, db.as_ref()).await
+        account::update_handle(did, handle, &self.db).await
     }
 
     pub async fn deactivate_account(&self, did: &str, delete_after: Option<String>) -> Result<()> {
-        account::deactivate_account(did, delete_after, self.db.as_ref()).await
+        account::deactivate_account(did, delete_after, &self.db).await
     }
 
     pub async fn activate_account(&self, did: &str) -> Result<()> {
-        let db = self.db.clone();
-        account::activate_account(did, db.as_ref()).await
+        account::activate_account(did, &self.db).await
     }
 
     pub async fn get_account_status(&self, handle_or_did: &str) -> Result<AccountStatus> {
···
                 include_deactivated: Some(true),
                 include_taken_down: Some(true),
             }),
-            self.db.as_ref(),
+            &self.db,
         )
         .await?;
         let res = account::format_account_status(got);
···
         did: String,
         app_password_name: Option<String>,
     ) -> Result<(String, String)> {
-        let db = self.db.clone();
         let secp = Secp256k1::new();
         let private_key = env::var("PDS_JWT_KEY_K256_PRIVATE_KEY_HEX")?;
         let secret_key = SecretKey::from_slice(&hex::decode(private_key.as_bytes())?)?;
···
             expires_in: None,
         })?;
         let refresh_payload = auth::decode_refresh_token(refresh_jwt.clone(), jwt_key)?;
-        auth::store_refresh_token(refresh_payload, app_password_name, db.as_ref()).await?;
+        auth::store_refresh_token(refresh_payload, app_password_name, &self.db).await?;
         Ok((access_jwt, refresh_jwt))
     }
 
     pub async fn rotate_refresh_token(&self, id: &String) -> Result<Option<(String, String)>> {
-        let token = auth::get_refresh_token(id, self.db.as_ref()).await?;
+        let token = auth::get_refresh_token(id, &self.db).await?;
         if let Some(token) = token {
             let system_time = SystemTime::now();
             let dt: DateTime<UtcOffset> = system_time.into();
···
 
             // take the chance to tidy all of a user's expired tokens
             // does not need to be transactional since this is just best-effort
-            auth::delete_expired_refresh_tokens(&token.did, now, self.db.as_ref()).await?;
+            auth::delete_expired_refresh_tokens(&token.did, now, &self.db).await?;
 
             // Shorten the refresh token lifespan down from its
             // original expiration time to its revocation grace period.
···
                         expires_at: from_micros_to_str(expires_at),
                         next_id
                     },
-                    self.db.as_ref()
+                    &self.db
                 ),
-                auth::store_refresh_token(
-                    refresh_payload,
-                    token.app_password_name,
-                    self.db.as_ref()
-                )
+                auth::store_refresh_token(refresh_payload, token.app_password_name, &self.db)
             ) {
                 Ok(_) => Ok(Some((access_jwt, refresh_jwt))),
                 Err(e) => match e.downcast_ref() {
···
     }
 
     pub async fn revoke_refresh_token(&self, id: String) -> Result<bool> {
-        auth::revoke_refresh_token(id, self.db.as_ref()).await
+        auth::revoke_refresh_token(id, &self.db).await
     }
 
     // Invites
···
         to_create: Vec<AccountCodes>,
         use_count: i32,
     ) -> Result<()> {
-        let db = self.db.clone();
-        invite::create_invite_codes(to_create, use_count, db.as_ref()).await
+        invite::create_invite_codes(to_create, use_count, &self.db).await
     }
 
     pub async fn create_account_invite_codes(
···
         expected_total: usize,
         disabled: bool,
     ) -> Result<Vec<CodeDetail>> {
-        invite::create_account_invite_codes(
-            for_account,
-            codes,
-            expected_total,
-            disabled,
-            self.db.as_ref(),
-        )
-        .await
+        invite::create_account_invite_codes(for_account, codes, expected_total, disabled, &self.db)
+            .await
     }
 
     pub async fn get_account_invite_codes(&self, did: &str) -> Result<Vec<CodeDetail>> {
-        let db = self.db.clone();
-        invite::get_account_invite_codes(did, db.as_ref()).await
+        invite::get_account_invite_codes(did, &self.db).await
     }
 
     pub async fn get_invited_by_for_accounts(
         &self,
         dids: Vec<String>,
     ) -> Result<BTreeMap<String, CodeDetail>> {
-        let db = self.db.clone();
-        invite::get_invited_by_for_accounts(dids, db.as_ref()).await
+        invite::get_invited_by_for_accounts(dids, &self.db).await
     }
 
     pub async fn set_account_invites_disabled(&self, did: &str, disabled: bool) -> Result<()> {
-        invite::set_account_invites_disabled(did, disabled, self.db.as_ref()).await
+        invite::set_account_invites_disabled(did, disabled, &self.db).await
     }
 
     pub async fn disable_invite_codes(&self, opts: DisableInviteCodesOpts) -> Result<()> {
-        invite::disable_invite_codes(opts, self.db.as_ref()).await
+        invite::disable_invite_codes(opts, &self.db).await
     }
 
     // Passwords
···
         did: String,
         name: String,
     ) -> Result<CreateAppPasswordOutput> {
-        password::create_app_password(did, name, self.db.as_ref()).await
+        password::create_app_password(did, name, &self.db).await
     }
 
     pub async fn list_app_passwords(&self, did: &str) -> Result<Vec<(String, String)>> {
-        password::list_app_passwords(did, self.db.as_ref()).await
+        password::list_app_passwords(did, &self.db).await
     }
 
     pub async fn verify_account_password(&self, did: &str, password_str: &String) -> Result<bool> {
-        let db = self.db.clone();
-        password::verify_account_password(did, password_str, db.as_ref()).await
+        password::verify_account_password(did, password_str, &self.db).await
     }
 
     pub async fn verify_app_password(
···
         did: &str,
         password_str: &str,
     ) -> Result<Option<String>> {
-        let db = self.db.clone();
-        password::verify_app_password(did, password_str, db.as_ref()).await
+        password::verify_app_password(did, password_str, &self.db).await
     }
 
     pub async fn reset_password(&self, opts: ResetPasswordOpts) -> Result<()> {
-        let db = self.db.clone();
         let did = email_token::assert_valid_token_and_find_did(
             EmailTokenPurpose::ResetPassword,
             &opts.token,
             None,
-            db.as_ref(),
+            &self.db,
         )
         .await?;
         self.update_account_password(UpdateAccountPasswordOpts {
···
     }
 
     pub async fn update_account_password(&self, opts: UpdateAccountPasswordOpts) -> Result<()> {
-        let db = self.db.clone();
         let UpdateAccountPasswordOpts { did, .. } = opts;
         let password_encrypted = password::gen_salt_and_hash(opts.password)?;
         try_join!(
···
                     did: did.clone(),
                     password_encrypted
                 },
-                self.db.as_ref()
+                &self.db
             ),
-            email_token::delete_email_token(&did, EmailTokenPurpose::ResetPassword, db.as_ref()),
-            auth::revoke_refresh_tokens_by_did(&did, self.db.as_ref())
+            email_token::delete_email_token(&did, EmailTokenPurpose::ResetPassword, &self.db),
+            auth::revoke_refresh_tokens_by_did(&did, &self.db)
         )?;
         Ok(())
     }
 
     pub async fn revoke_app_password(&self, did: String, name: String) -> Result<()> {
         try_join!(
-            password::delete_app_password(&did, &name, self.db.as_ref()),
-            auth::revoke_app_password_refresh_token(&did, &name, self.db.as_ref())
+            password::delete_app_password(&did, &name, &self.db),
+            auth::revoke_app_password_refresh_token(&did, &name, &self.db)
         )?;
         Ok(())
     }
 
     // Email Tokens
     // ----------
-    pub async fn confirm_email<'em>(&self, opts: ConfirmEmailOpts<'em>) -> Result<()> {
-        let db = self.db.clone();
+    pub async fn confirm_email(&self, opts: ConfirmEmailOpts<'_>) -> Result<()> {
         let ConfirmEmailOpts { did, token } = opts;
         email_token::assert_valid_token(
             did,
             EmailTokenPurpose::ConfirmEmail,
             token,
             None,
-            db.as_ref(),
+            &self.db,
         )
         .await?;
         let now = rsky_common::now();
         try_join!(
-            email_token::delete_email_token(did, EmailTokenPurpose::ConfirmEmail, db.as_ref()),
-            account::set_email_confirmed_at(did, now, self.db.as_ref())
+            email_token::delete_email_token(did, EmailTokenPurpose::ConfirmEmail, &self.db),
+            account::set_email_confirmed_at(did, now, &self.db)
         )?;
         Ok(())
     }
 
     pub async fn update_email(&self, opts: UpdateEmailOpts) -> Result<()> {
-        let db = self.db.clone();
         let UpdateEmailOpts { did, email } = opts;
         try_join!(
-            account::update_email(&did, &email, db.as_ref()),
-            email_token::delete_all_email_tokens(&did, db.as_ref())
+            account::update_email(&did, &email, &self.db),
+            email_token::delete_all_email_tokens(&did, &self.db)
         )?;
         Ok(())
     }
···
         purpose: EmailTokenPurpose,
         token: &str,
     ) -> Result<()> {
-        let db = self.db.clone();
-        email_token::assert_valid_token(did, purpose, token, None, db.as_ref()).await
+        email_token::assert_valid_token(did, purpose, token, None, &self.db).await
     }
 
     pub async fn assert_valid_email_token_and_cleanup(
···
         purpose: EmailTokenPurpose,
         token: &str,
     ) -> Result<()> {
-        let db = self.db.clone();
-        email_token::assert_valid_token(did, purpose, token, None, db.as_ref()).await?;
-        email_token::delete_email_token(did, purpose, db.as_ref()).await
+        email_token::assert_valid_token(did, purpose, token, None, &self.db).await?;
+        email_token::delete_email_token(did, purpose, &self.db).await
     }
 
     pub async fn create_email_token(
···
         did: &str,
         purpose: EmailTokenPurpose,
     ) -> Result<String> {
-        let db = self.db.clone();
-        email_token::create_email_token(did, purpose, db.as_ref()).await
+        email_token::create_email_token(did, purpose, &self.db).await
     }
 }

src/endpoints/mod.rs

···
 //! Root module for all endpoints.
-mod identity;
-mod repo;
-mod server;
-mod sync;
+// mod identity;
+// mod repo;
+// mod server;
+// mod sync;
 
 use axum::{Json, Router, routing::get};
 use serde_json::json;
···
 
 /// Register all root routes.
 pub(crate) fn routes() -> Router<AppState> {
-    Router::new()
-        .route("/_health", get(health))
-        .merge(identity::routes()) // com.atproto.identity
-        .merge(repo::routes()) // com.atproto.repo
-        .merge(server::routes()) // com.atproto.server
-        .merge(sync::routes()) // com.atproto.sync
+    Router::new().route("/_health", get(health))
+    // .merge(identity::routes()) // com.atproto.identity
+    // .merge(repo::routes()) // com.atproto.repo
+    // .merge(server::routes()) // com.atproto.server
+    // .merge(sync::routes()) // com.atproto.sync
 }