prototype account_manager; reorganize

src/account_manager/mod.rs

···
+use anyhow::Result;
+use chrono::DateTime;
+use chrono::offset::Utc as UtcOffset;
+use cidv10::Cid;
+use futures::try_join;
+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::CreateAccountOpts;
+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::helpers::{account, auth, email_token, invite, password};
+use rsky_pds::auth_verifier::AuthScope;
+use rsky_pds::models::models::EmailTokenPurpose;
+use secp256k1::{Keypair, Secp256k1, SecretKey};
+use std::collections::BTreeMap;
+use std::env;
+use std::sync::Arc;
+use std::time::SystemTime;
+
+use crate::db::DbConn;
+
+#[derive(Clone, Debug)]
+pub struct AccountManager {
+    pub db: Arc<DbConn>,
+}
+
+pub type AccountManagerCreator = Box<dyn Fn(Arc<DbConn>) -> AccountManager + Send + Sync>;
+
+impl AccountManager {
+    pub fn new(db: Arc<DbConn>) -> Self {
+        Self { db }
+    }
+
+    pub fn creator() -> AccountManagerCreator {
+        Box::new(move |db: Arc<DbConn>| -> AccountManager { AccountManager::new(db) })
+    }
+
+    pub async fn get_account(
+        &self,
+        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
+    }
+
+    pub async fn get_account_by_email(
+        &self,
+        email: &str,
+        flags: Option<AvailabilityFlags>,
+    ) -> Result<Option<ActorAccount>> {
+        let db = self.db.clone();
+        account::get_account_by_email(email, flags, db.as_ref()).await
+    }
+
+    pub async fn is_account_activated(&self, did: &str) -> Result<bool> {
+        let account = self
+            .get_account(
+                did,
+                Some(AvailabilityFlags {
+                    include_taken_down: None,
+                    include_deactivated: Some(true),
+                }),
+            )
+            .await?;
+        if let Some(account) = account {
+            Ok(account.deactivated_at.is_none())
+        } else {
+            Ok(false)
+        }
+    }
+
+    pub async fn get_did_for_actor(
+        &self,
+        handle_or_did: &str,
+        flags: Option<AvailabilityFlags>,
+    ) -> Result<Option<String>> {
+        match self.get_account(handle_or_did, flags).await {
+            Ok(Some(got)) => Ok(Some(got.did)),
+            _ => Ok(None),
+        }
+    }
+
+    pub async fn create_account(&self, opts: CreateAccountOpts) -> Result<(String, String)> {
+        let db = self.db.clone();
+        let CreateAccountOpts {
+            did,
+            handle,
+            email,
+            password,
+            repo_cid,
+            repo_rev,
+            invite_code,
+            deactivated,
+        } = opts;
+        let password_encrypted: Option<String> = match password {
+            Some(password) => Some(password::gen_salt_and_hash(password)?),
+            None => None,
+        };
+        // Should be a global var so this only happens once
+        let secp = Secp256k1::new();
+        let private_key = env::var("PDS_JWT_KEY_K256_PRIVATE_KEY_HEX")?;
+        let secret_key =
+            SecretKey::from_slice(&Result::unwrap(hex::decode(private_key.as_bytes())))?;
+        let jwt_key = Keypair::from_secret_key(&secp, &secret_key);
+        let (access_jwt, refresh_jwt) = auth::create_tokens(CreateTokensOpts {
+            did: did.clone(),
+            jwt_key,
+            service_did: env::var("PDS_SERVICE_DID").unwrap(),
+            scope: Some(AuthScope::Access),
+            jti: None,
+            expires_in: None,
+        })?;
+        let refresh_payload = auth::decode_refresh_token(refresh_jwt.clone(), jwt_key)?;
+        let now = rsky_common::now();
+
+        if let Some(invite_code) = invite_code.clone() {
+            invite::ensure_invite_is_available(invite_code, db.as_ref()).await?;
+        }
+        account::register_actor(did.clone(), handle, deactivated, db.as_ref()).await?;
+        if let (Some(email), Some(password_encrypted)) = (email, password_encrypted) {
+            account::register_account(did.clone(), email, password_encrypted, db.as_ref()).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?;
+        Ok((access_jwt, refresh_jwt))
+    }
+
+    pub async fn get_account_admin_status(
+        &self,
+        did: &str,
+    ) -> Result<Option<GetAccountAdminStatusOutput>> {
+        let db = self.db.clone();
+        account::get_account_admin_status(did, db.as_ref()).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
+    }
+
+    pub async fn delete_account(&self, did: &str) -> Result<()> {
+        let db = self.db.clone();
+        account::delete_account(did, db.as_ref()).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())
+        )?;
+        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
+    }
+
+    pub async fn deactivate_account(&self, did: &str, delete_after: Option<String>) -> Result<()> {
+        account::deactivate_account(did, delete_after, self.db.as_ref()).await
+    }
+
+    pub async fn activate_account(&self, did: &str) -> Result<()> {
+        let db = self.db.clone();
+        account::activate_account(did, db.as_ref()).await
+    }
+
+    pub async fn get_account_status(&self, handle_or_did: &str) -> Result<AccountStatus> {
+        let got = account::get_account(
+            handle_or_did,
+            Some(AvailabilityFlags {
+                include_deactivated: Some(true),
+                include_taken_down: Some(true),
+            }),
+            self.db.as_ref(),
+        )
+        .await?;
+        let res = account::format_account_status(got);
+        match res.active {
+            true => Ok(AccountStatus::Active),
+            false => Ok(res.status.expect("Account status not properly formatted.")),
+        }
+    }
+
+    // Auth
+    // ----------
+    pub async fn create_session(
+        &self,
+        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())?)?;
+        let jwt_key = Keypair::from_secret_key(&secp, &secret_key);
+        let scope = if app_password_name.is_none() {
+            AuthScope::Access
+        } else {
+            AuthScope::AppPass
+        };
+        let (access_jwt, refresh_jwt) = auth::create_tokens(CreateTokensOpts {
+            did,
+            jwt_key,
+            service_did: env::var("PDS_SERVICE_DID").unwrap(),
+            scope: Some(scope),
+            jti: None,
+            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?;
+        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?;
+        if let Some(token) = token {
+            let system_time = SystemTime::now();
+            let dt: DateTime<UtcOffset> = system_time.into();
+            let now = format!("{}", dt.format(RFC3339_VARIANT));
+
+            // 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?;
+
+            // Shorten the refresh token lifespan down from its
+            // original expiration time to its revocation grace period.
+            let prev_expires_at = from_str_to_micros(&token.expires_at);
+
+            const REFRESH_GRACE_MS: i32 = 2 * HOUR;
+            let grace_expires_at = dt.timestamp_micros() + REFRESH_GRACE_MS as i64;
+
+            let expires_at = if grace_expires_at < prev_expires_at {
+                grace_expires_at
+            } else {
+                prev_expires_at
+            };
+
+            if expires_at <= dt.timestamp_micros() {
+                return Ok(None);
+            }
+
+            // Determine the next refresh token id: upon refresh token
+            // reuse you always receive a refresh token with the same id.
+            let next_id = token.next_id.unwrap_or_else(auth::get_refresh_token_id);
+
+            let secp = Secp256k1::new();
+            let private_key = env::var("PDS_JWT_KEY_K256_PRIVATE_KEY_HEX").unwrap();
+            let secret_key =
+                SecretKey::from_slice(&hex::decode(private_key.as_bytes()).unwrap()).unwrap();
+            let jwt_key = Keypair::from_secret_key(&secp, &secret_key);
+
+            let (access_jwt, refresh_jwt) = auth::create_tokens(CreateTokensOpts {
+                did: token.did,
+                jwt_key,
+                service_did: env::var("PDS_SERVICE_DID").unwrap(),
+                scope: Some(if token.app_password_name.is_none() {
+                    AuthScope::Access
+                } else {
+                    AuthScope::AppPass
+                }),
+                jti: Some(next_id.clone()),
+                expires_in: None,
+            })?;
+            let refresh_payload = auth::decode_refresh_token(refresh_jwt.clone(), jwt_key)?;
+            match try_join!(
+                auth::add_refresh_grace_period(
+                    RefreshGracePeriodOpts {
+                        id: id.clone(),
+                        expires_at: from_micros_to_str(expires_at),
+                        next_id
+                    },
+                    self.db.as_ref()
+                ),
+                auth::store_refresh_token(
+                    refresh_payload,
+                    token.app_password_name,
+                    self.db.as_ref()
+                )
+            ) {
+                Ok(_) => Ok(Some((access_jwt, refresh_jwt))),
+                Err(e) => match e.downcast_ref() {
+                    Some(AuthHelperError::ConcurrentRefresh) => {
+                        Box::pin(self.rotate_refresh_token(id)).await
+                    }
+                    _ => Err(e),
+                },
+            }
+        } else {
+            Ok(None)
+        }
+    }
+
+    pub async fn revoke_refresh_token(&self, id: String) -> Result<bool> {
+        auth::revoke_refresh_token(id, self.db.as_ref()).await
+    }
+
+    // Invites
+    // ----------
+
+    pub async fn create_invite_codes(
+        &self,
+        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
+    }
+
+    pub async fn create_account_invite_codes(
+        &self,
+        for_account: &str,
+        codes: Vec<String>,
+        expected_total: usize,
+        disabled: bool,
+    ) -> Result<Vec<CodeDetail>> {
+        invite::create_account_invite_codes(
+            for_account,
+            codes,
+            expected_total,
+            disabled,
+            self.db.as_ref(),
+        )
+        .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
+    }
+
+    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
+    }
+
+    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
+    }
+
+    pub async fn disable_invite_codes(&self, opts: DisableInviteCodesOpts) -> Result<()> {
+        invite::disable_invite_codes(opts, self.db.as_ref()).await
+    }
+
+    // Passwords
+    // ----------
+
+    pub async fn create_app_password(
+        &self,
+        did: String,
+        name: String,
+    ) -> Result<CreateAppPasswordOutput> {
+        password::create_app_password(did, name, self.db.as_ref()).await
+    }
+
+    pub async fn list_app_passwords(&self, did: &str) -> Result<Vec<(String, String)>> {
+        password::list_app_passwords(did, self.db.as_ref()).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
+    }
+
+    pub async fn verify_app_password(
+        &self,
+        did: &str,
+        password_str: &str,
+    ) -> Result<Option<String>> {
+        let db = self.db.clone();
+        password::verify_app_password(did, password_str, db.as_ref()).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(),
+        )
+        .await?;
+        self.update_account_password(UpdateAccountPasswordOpts {
+            did,
+            password: opts.password,
+        })
+        .await
+    }
+
+    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!(
+            password::update_user_password(
+                UpdateUserPasswordOpts {
+                    did: did.clone(),
+                    password_encrypted
+                },
+                self.db.as_ref()
+            ),
+            email_token::delete_email_token(&did, EmailTokenPurpose::ResetPassword, db.as_ref()),
+            auth::revoke_refresh_tokens_by_did(&did, self.db.as_ref())
+        )?;
+        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())
+        )?;
+        Ok(())
+    }
+
+    // Email Tokens
+    // ----------
+    pub async fn confirm_email<'em>(&self, opts: ConfirmEmailOpts<'em>) -> Result<()> {
+        let db = self.db.clone();
+        let ConfirmEmailOpts { did, token } = opts;
+        email_token::assert_valid_token(
+            did,
+            EmailTokenPurpose::ConfirmEmail,
+            token,
+            None,
+            db.as_ref(),
+        )
+        .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())
+        )?;
+        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())
+        )?;
+        Ok(())
+    }
+
+    pub async fn assert_valid_email_token(
+        &self,
+        did: &str,
+        purpose: EmailTokenPurpose,
+        token: &str,
+    ) -> Result<()> {
+        let db = self.db.clone();
+        email_token::assert_valid_token(did, purpose, token, None, db.as_ref()).await
+    }
+
+    pub async fn assert_valid_email_token_and_cleanup(
+        &self,
+        did: &str,
+        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
+    }
+
+    pub async fn create_email_token(
+        &self,
+        did: &str,
+        purpose: EmailTokenPurpose,
+    ) -> Result<String> {
+        let db = self.db.clone();
+        email_token::create_email_token(did, purpose, db.as_ref()).await
+    }
+}

src/actor_endpoints.rs

···
+use atrium_api::app::bsky::actor;
+use axum::{Json, routing::post};
+use constcat::concat;
+use diesel::prelude::*;
+
+use super::*;
+
+async fn put_preferences(
+    user: AuthenticatedUser,
+    State(db): State<Db>,
+    Json(input): Json<actor::put_preferences::Input>,
+) -> Result<()> {
+    let did = user.did();
+    let json_string =
+        serde_json::to_string(&input.preferences).context("failed to serialize preferences")?;
+
+    // Use the db connection pool to execute the update
+    let conn = &mut db.get().context("failed to get database connection")?;
+    diesel::sql_query("UPDATE accounts SET private_prefs = ? WHERE did = ?")
+        .bind::<diesel::sql_types::Text, _>(json_string)
+        .bind::<diesel::sql_types::Text, _>(did)
+        .execute(conn)
+        .context("failed to update user preferences")?;
+
+    Ok(())
+}
+
+async fn get_preferences(
+    user: AuthenticatedUser,
+    State(db): State<Db>,
+) -> Result<Json<actor::get_preferences::Output>> {
+    let did = user.did();
+    let conn = &mut db.get().context("failed to get database connection")?;
+
+    #[derive(QueryableByName)]
+    struct Prefs {
+        #[diesel(sql_type = diesel::sql_types::Text)]
+        private_prefs: Option<String>,
+    }
+
+    let result = diesel::sql_query("SELECT private_prefs FROM accounts WHERE did = ?")
+        .bind::<diesel::sql_types::Text, _>(did)
+        .get_result::<Prefs>(conn)
+        .context("failed to fetch preferences")?;
+
+    if let Some(prefs_json) = result.private_prefs {
+        let prefs: actor::defs::Preferences =
+            serde_json::from_str(&prefs_json).context("failed to deserialize preferences")?;
+
+        Ok(Json(
+            actor::get_preferences::OutputData { preferences: prefs }.into(),
+        ))
+    } else {
+        Ok(Json(
+            actor::get_preferences::OutputData {
+                preferences: Vec::new(),
+            }
+            .into(),
+        ))
+    }
+}
+
+/// Register all actor endpoints.
+pub(crate) fn routes() -> Router<AppState> {
+    // AP /xrpc/app.bsky.actor.putPreferences
+    // AG /xrpc/app.bsky.actor.getPreferences
+    Router::new()
+        .route(
+            concat!("/", actor::put_preferences::NSID),
+            post(put_preferences),
+        )
+        .route(
+            concat!("/", actor::get_preferences::NSID),
+            get(get_preferences),
+        )
+}

src/actor_store/mod.rs

···
 mod blob;
 mod preference;
 mod record;
-mod sql_blob;
+pub(crate) mod sql_blob;
 mod sql_repo;
 
 use anyhow::Result;

src/endpoints/repo/apply_writes.rs

···
 //! Apply a batch transaction of repository creates, updates, and deletes. Requires auth, implemented by PDS.
-use std::{collections::HashSet, str::FromStr};
-
+use crate::{
+    AppState, Db, Error, Result, SigningKey,
+    actor_store::ActorStore,
+    actor_store::sql_blob::BlobStoreSql,
+    auth::AuthenticatedUser,
+    config::AppConfig,
+    error::ErrorMessage,
+    firehose::{self, FirehoseProducer, RepoOp},
+    metrics::{REPO_COMMITS, REPO_OP_CREATE, REPO_OP_DELETE, REPO_OP_UPDATE},
+    storage,
+};
+use anyhow::bail;
 use anyhow::{Context as _, anyhow};
 use atrium_api::com::atproto::repo::apply_writes::{self, InputWritesItem, OutputResultsItem};
 use atrium_api::{
···
         string::{AtIdentifier, Nsid, Tid},
     },
 };
-use atrium_repo::{Cid, blockstore::CarStore};
+use atrium_repo::blockstore::CarStore;
 use axum::{
     Json, Router,
     body::Body,
···
     http::{self, StatusCode},
     routing::{get, post},
 };
+use cidv10::Cid;
 use constcat::concat;
 use futures::TryStreamExt as _;
+use futures::stream::{self, StreamExt};
 use metrics::counter;
+use rsky_lexicon::com::atproto::repo::{ApplyWritesInput, ApplyWritesInputRefWrite};
+use rsky_pds::SharedSequencer;
+use rsky_pds::account_manager::AccountManager;
+use rsky_pds::account_manager::helpers::account::AvailabilityFlags;
+use rsky_pds::apis::ApiError;
+use rsky_pds::auth_verifier::AccessStandardIncludeChecks;
+use rsky_pds::repo::prepare::{
+    PrepareCreateOpts, PrepareDeleteOpts, PrepareUpdateOpts, prepare_create, prepare_delete,
+    prepare_update,
+};
+use rsky_repo::types::PreparedWrite;
 use rsky_syntax::aturi::AtUri;
 use serde::Deserialize;
+use std::{collections::HashSet, str::FromStr};
 use tokio::io::AsyncWriteExt as _;
 
-use crate::repo::block_map::cid_for_cbor;
-use crate::repo::types::PreparedCreateOrUpdate;
-use crate::{
-    AppState, Db, Error, Result, SigningKey,
-    actor_store::{ActorStoreTransactor, ActorStoreWriter},
-    auth::AuthenticatedUser,
-    config::AppConfig,
-    error::ErrorMessage,
-    firehose::{self, FirehoseProducer, RepoOp},
-    metrics::{REPO_COMMITS, REPO_OP_CREATE, REPO_OP_DELETE, REPO_OP_UPDATE},
-    repo::types::{PreparedWrite, WriteOpAction},
-    storage,
-};
-
 use super::resolve_did;
 
 /// Apply a batch transaction of repository creates, updates, and deletes. Requires auth, implemented by PDS.
···
     State(config): State<AppConfig>,
     State(db): State<Db>,
     State(fhp): State<FirehoseProducer>,
-    Json(input): Json<repo::apply_writes::Input>,
+    Json(input): Json<ApplyWritesInput>,
 ) -> Result<Json<repo::apply_writes::Output>> {
-    todo!();
-    // // TODO: `input.validate`
-
-    // // Resolve DID from identifier
-    // let (target_did, _) = resolve_did(&db, &input.repo)
-    //     .await
-    //     .context("failed to resolve did")?;
-
-    // // Ensure that we are updating the correct repository
-    // if target_did.as_str() != user.did() {
-    //     return Err(Error::with_status(
-    //         StatusCode::BAD_REQUEST,
-    //         anyhow!("repo did not match the authenticated user"),
-    //     ));
-    // }
-
-    // // Validate writes count
-    // if input.writes.len() > 200 {
-    //     return Err(Error::with_status(
-    //         StatusCode::BAD_REQUEST,
-    //         anyhow!("Too many writes. Max: 200"),
-    //     ));
-    // }
-
-    // // Convert input writes to prepared format
-    // let mut prepared_writes = Vec::with_capacity(input.writes.len());
-    // for write in input.writes.iter() {
-    //     match write {
-    //         InputWritesItem::Create(create) => {
-    //             let uri = AtUri::make(
-    //                 user.did(),
-    //                 &create.collection.as_str(),
-    //                 create
-    //                     .rkey
-    //                     .as_deref()
-    //                     .unwrap_or(&Tid::now(LimitedU32::MIN).to_string()),
-    //             );
-
-    //             let cid = match cid_for_cbor(&create.value) {
-    //                 Ok(cid) => cid,
-    //                 Err(e) => {
-    //                     return Err(Error::with_status(
-    //                         StatusCode::BAD_REQUEST,
-    //                         anyhow!("Failed to encode record: {}", e),
-    //                     ));
-    //                 }
-    //             };
-
-    //             let blobs = scan_blobs(&create.value)
-    //                 .unwrap_or_default()
-    //                 .into_iter()
-    //                 .map(|cid| {
-    //                     // TODO: Create BlobRef from cid with proper metadata
-    //                     BlobRef {
-    //                         cid,
-    //                         mime_type: "application/octet-stream".to_string(), // Default
-    //                         size: 0, // Unknown at this point
-    //                     }
-    //                 })
-    //                 .collect();
-
-    //             prepared_writes.push(PreparedCreateOrUpdate {
-    //                 action: WriteOpAction::Create,
-    //                 uri: uri?.to_string(),
-    //                 cid,
-    //                 record: create.value.clone(),
-    //                 blobs,
-    //                 swap_cid: None,
-    //             });
-    //         }
-    //         InputWritesItem::Update(update) => {
-    //             let uri = AtUri::make(
-    //                 user.did(),
-    //                 Some(update.collection.to_string()),
-    //                 Some(update.rkey.to_string()),
-    //             );
-
-    //             let cid = match cid_for_cbor(&update.value) {
-    //                 Ok(cid) => cid,
-    //                 Err(e) => {
-    //                     return Err(Error::with_status(
-    //                         StatusCode::BAD_REQUEST,
-    //                         anyhow!("Failed to encode record: {}", e),
-    //                     ));
-    //                 }
-    //             };
-
-    //             let blobs = scan_blobs(&update.value)
-    //                 .unwrap_or_default()
-    //                 .into_iter()
-    //                 .map(|cid| {
-    //                     // TODO: Create BlobRef from cid with proper metadata
-    //                     BlobRef {
-    //                         cid,
-    //                         mime_type: "application/octet-stream".to_string(),
-    //                         size: 0,
-    //                     }
-    //                 })
-    //                 .collect();
-
-    //             prepared_writes.push(PreparedCreateOrUpdate {
-    //                 action: WriteOpAction::Update,
-    //                 uri: uri?.to_string(),
-    //                 cid,
-    //                 record: update.value.clone(),
-    //                 blobs,
-    //                 swap_cid: None,
-    //             });
-    //         }
-    //         InputWritesItem::Delete(delete) => {
-    //             let uri = AtUri::make(user.did(), &delete.collection.as_str(), &delete.rkey);
-
-    //             prepared_writes.push(PreparedCreateOrUpdate {
-    //                 action: WriteOpAction::Delete,
-    //                 uri: uri?.to_string(),
-    //                 cid: Cid::default(), // Not needed for delete
-    //                 record: serde_json::Value::Null,
-    //                 blobs: vec![],
-    //                 swap_cid: None,
-    //             });
-    //         }
-    //     }
-    // }
-
-    // // Get swap commit CID if provided
-    // let swap_commit_cid = input.swap_commit.as_ref().map(|cid| *cid.as_ref());
-
-    // let did_str = user.did();
-    // let mut repo = storage::open_repo_db(&config.repo, &db, did_str)
-    //     .await
-    //     .context("failed to open user repo")?;
-    // let orig_cid = repo.root();
-    // let orig_rev = repo.commit().rev();
-
-    // let mut blobs = vec![];
-    // let mut res = vec![];
-    // let mut ops = vec![];
-
-    // for write in &prepared_writes {
-    //     let (builder, key) = match write.action {
-    //         WriteOpAction::Create => {
-    //             let key = format!("{}/{}", write.uri.collection, write.uri.rkey);
-    //             let uri = format!("at://{}/{}", user.did(), key);
-
-    //             let (builder, cid) = repo
-    //                 .add_raw(&key, &write.record)
-    //                 .await
-    //                 .context("failed to add record")?;
-
-    //             // Extract and track blobs
-    //             if let Ok(new_blobs) = scan_blobs(&write.record) {
-    //                 blobs.extend(
-    //                     new_blobs
-    //                         .into_iter()
-    //                         .map(|blob_cid| (key.clone(), blob_cid)),
-    //                 );
-    //             }
-
-    //             ops.push(RepoOp::Create {
-    //                 cid,
-    //                 path: key.clone(),
-    //             });
-
-    //             res.push(OutputResultsItem::CreateResult(Box::new(
-    //                 apply_writes::CreateResultData {
-    //                     cid: atrium_api::types::string::Cid::new(cid),
-    //                     uri,
-    //                     validation_status: None,
-    //                 }
-    //                 .into(),
-    //             )));
-
-    //             (builder, key)
-    //         }
-    //         WriteOpAction::Update => {
-    //             let key = format!("{}/{}", write.uri.collection, write.uri.rkey);
-    //             let uri = format!("at://{}/{}", user.did(), key);
-
-    //             let prev = repo
-    //                 .tree()
-    //                 .get(&key)
-    //                 .await
-    //                 .context("failed to search MST")?;
-
-    //             if prev.is_none() {
-    //                 // No existing record, treat as create
-    //                 let (create_builder, cid) = repo
-    //                     .add_raw(&key, &write.record)
-    //                     .await
-    //                     .context("failed to add record")?;
-
-    //                 if let Ok(new_blobs) = scan_blobs(&write.record) {
-    //                     blobs.extend(
-    //                         new_blobs
-    //                             .into_iter()
-    //                             .map(|blob_cid| (key.clone(), blob_cid)),
-    //                     );
-    //                 }
-
-    //                 ops.push(RepoOp::Create {
-    //                     cid,
-    //                     path: key.clone(),
-    //                 });
-
-    //                 res.push(OutputResultsItem::CreateResult(Box::new(
-    //                     apply_writes::CreateResultData {
-    //                         cid: atrium_api::types::string::Cid::new(cid),
-    //                         uri,
-    //                         validation_status: None,
-    //                     }
-    //                     .into(),
-    //                 )));
-
-    //                 (create_builder, key)
-    //             } else {
-    //                 // Update existing record
-    //                 let prev = prev.context("should be able to find previous record")?;
-    //                 let (update_builder, cid) = repo
-    //                     .update_raw(&key, &write.record)
-    //                     .await
-    //                     .context("failed to add record")?;
-
-    //                 if let Ok(new_blobs) = scan_blobs(&write.record) {
-    //                     blobs.extend(
-    //                         new_blobs
-    //                             .into_iter()
-    //                             .map(|blob_cid| (key.clone(), blob_cid)),
-    //                     );
-    //                 }
-
-    //                 ops.push(RepoOp::Update {
-    //                     cid,
-    //                     path: key.clone(),
-    //                     prev,
-    //                 });
-
-    //                 res.push(OutputResultsItem::UpdateResult(Box::new(
-    //                     apply_writes::UpdateResultData {
-    //                         cid: atrium_api::types::string::Cid::new(cid),
-    //                         uri,
-    //                         validation_status: None,
-    //                     }
-    //                     .into(),
-    //                 )));
+    let tx: ApplyWritesInput = input;
+    let ApplyWritesInput {
+        repo,
+        validate,
+        swap_commit,
+        ..
+    } = tx;
+    let account = account_manager
+        .get_account(
+            &repo,
+            Some(AvailabilityFlags {
+                include_deactivated: Some(true),
+                include_taken_down: None,
+            }),
+        )
+        .await?;
 
-    //                 (update_builder, key)
-    //             }
-    //         }
-    //         WriteOpAction::Delete => {
-    //             let key = format!("{}/{}", write.uri.collection, write.uri.rkey);
+    if let Some(account) = account {
+        if account.deactivated_at.is_some() {
+            return Err(Error::with_message(
+                StatusCode::FORBIDDEN,
+                anyhow!("Account is deactivated"),
+                ErrorMessage::new("AccountDeactivated", "Account is deactivated"),
+            ));
+        }
+        let did = account.did;
+        if did != user.did() {
+            return Err(Error::with_message(
+                StatusCode::FORBIDDEN,
+                anyhow!("AuthRequiredError"),
+                ErrorMessage::new("AuthRequiredError", "Auth required"),
+            ));
+        }
+        let did: &String = &did;
+        if tx.writes.len() > 200 {
+            return Err(Error::with_message(
+                StatusCode::BAD_REQUEST,
+                anyhow!("Too many writes. Max: 200"),
+                ErrorMessage::new("TooManyWrites", "Too many writes. Max: 200"),
+            ));
+        }
 
-    //             let prev = repo
-    //                 .tree()
-    //                 .get(&key)
-    //                 .await
-    //                 .context("failed to search MST")?
-    //                 .context("previous record does not exist")?;
-
-    //             ops.push(RepoOp::Delete {
-    //                 path: key.clone(),
-    //                 prev,
-    //             });
-
-    //             res.push(OutputResultsItem::DeleteResult(Box::new(
-    //                 apply_writes::DeleteResultData {}.into(),
-    //             )));
+        let writes: Vec<PreparedWrite> = stream::iter(tx.writes)
+            .then(|write| async move {
+                Ok::<PreparedWrite, anyhow::Error>(match write {
+                    ApplyWritesInputRefWrite::Create(write) => PreparedWrite::Create(
+                        prepare_create(PrepareCreateOpts {
+                            did: did.clone(),
+                            collection: write.collection,
+                            rkey: write.rkey,
+                            swap_cid: None,
+                            record: serde_json::from_value(write.value)?,
+                            validate,
+                        })
+                        .await?,
+                    ),
+                    ApplyWritesInputRefWrite::Update(write) => PreparedWrite::Update(
+                        prepare_update(PrepareUpdateOpts {
+                            did: did.clone(),
+                            collection: write.collection,
+                            rkey: write.rkey,
+                            swap_cid: None,
+                            record: serde_json::from_value(write.value)?,
+                            validate,
+                        })
+                        .await?,
+                    ),
+                    ApplyWritesInputRefWrite::Delete(write) => {
+                        PreparedWrite::Delete(prepare_delete(PrepareDeleteOpts {
+                            did: did.clone(),
+                            collection: write.collection,
+                            rkey: write.rkey,
+                            swap_cid: None,
+                        })?)
+                    }
+                })
+            })
+            .collect::<Vec<_>>()
+            .await
+            .into_iter()
+            .collect::<Result<Vec<PreparedWrite>, _>>()?;
 
-    //             let builder = repo
-    //                 .delete_raw(&key)
-    //                 .await
-    //                 .context("failed to add record")?;
+        let swap_commit_cid = match swap_commit {
+            Some(swap_commit) => Some(Cid::from_str(&swap_commit)?),
+            None => None,
+        };
 
-    //             (builder, key)
-    //         }
-    //     };
+        let mut actor_store = ActorStore::new(did.clone(), BlobStoreSql::new(did.clone(), db), db);
 
-    //     let sig = skey
-    //         .sign(&builder.bytes())
-    //         .context("failed to sign commit")?;
+        let commit = actor_store
+            .process_writes(writes.clone(), swap_commit_cid)
+            .await?;
 
-    //     _ = builder
-    //         .finalize(sig)
-    //         .await
-    //         .context("failed to write signed commit")?;
-    // }
-
-    // // Construct a firehose record
-    // let mut mem = Vec::new();
-    // let mut store = CarStore::create_with_roots(std::io::Cursor::new(&mut mem), [repo.root()])
-    //     .await
-    //     .context("failed to create temp store")?;
-
-    // // Extract the records out of the user's repository
-    // for write in &prepared_writes {
-    //     let key = format!("{}/{}", write.uri.collection, write.uri.rkey);
-    //     repo.extract_raw_into(&key, &mut store)
-    //         .await
-    //         .context("failed to extract key")?;
-    // }
-
-    // let mut tx = db.begin().await.context("failed to begin transaction")?;
-
-    // if !swap_commit(
-    //     &mut *tx,
-    //     repo.root(),
-    //     repo.commit().rev(),
-    //     input.swap_commit.as_ref().map(|cid| *cid.as_ref()),
-    //     &user.did(),
-    // )
-    // .await
-    // .context("failed to swap commit")?
-    // {
-    //     // This should always succeed.
-    //     let old = input
-    //         .swap_commit
-    //         .clone()
-    //         .context("swap_commit should always be Some")?;
-
-    //     // The swap failed. Return the old commit and do not update the repository.
-    //     return Ok(Json(
-    //         apply_writes::OutputData {
-    //             results: None,
-    //             commit: Some(
-    //                 CommitMetaData {
-    //                     cid: old,
-    //                     rev: orig_rev,
-    //                 }
-    //                 .into(),
-    //             ),
-    //         }
-    //         .into(),
-    //     ));
-    // }
-
-    // // For updates and removals, unlink the old/deleted record from the blob_ref table
-    // for op in &ops {
-    //     match op {
-    //         &RepoOp::Update { ref path, .. } | &RepoOp::Delete { ref path, .. } => {
-    //             // FIXME: This may cause issues if a user deletes more than one record referencing the same blob.
-    //             _ = &sqlx::query!(
-    //                 r#"UPDATE blob_ref SET record = NULL WHERE did = ? AND record = ?"#,
-    //                 did_str,
-    //                 path
-    //             )
-    //             .execute(&mut *tx)
-    //             .await
-    //             .context("failed to remove blob_ref")?;
-    //         }
-    //         &RepoOp::Create { .. } => {}
-    //     }
-    // }
-
-    // // Process blobs
-    // for (key, cid) in &blobs {
-    //     let cid_str = cid.to_string();
-
-    //     // Handle the case where a new record references an existing blob
-    //     if sqlx::query!(
-    //         r#"UPDATE blob_ref SET record = ? WHERE cid = ? AND did = ? AND record IS NULL"#,
-    //         key,
-    //         cid_str,
-    //         did_str,
-    //     )
-    //     .execute(&mut *tx)
-    //     .await
-    //     .context("failed to update blob_ref")?
-    //     .rows_affected()
-    //         == 0
-    //     {
-    //         _ = sqlx::query!(
-    //             r#"INSERT INTO blob_ref (record, cid, did) VALUES (?, ?, ?)"#,
-    //             key,
-    //             cid_str,
-    //             did_str,
-    //         )
-    //         .execute(&mut *tx)
-    //         .await
-    //         .context("failed to update blob_ref")?;
-    //     }
-    // }
-
-    // tx.commit()
-    //     .await
-    //     .context("failed to commit blob ref to database")?;
-
-    // // Update counters
-    // counter!(REPO_COMMITS).increment(1);
-    // for op in &ops {
-    //     match *op {
-    //         RepoOp::Create { .. } => counter!(REPO_OP_CREATE).increment(1),
-    //         RepoOp::Update { .. } => counter!(REPO_OP_UPDATE).increment(1),
-    //         RepoOp::Delete { .. } => counter!(REPO_OP_DELETE).increment(1),
-    //     }
-    // }
-
-    // // We've committed the transaction to the database, and the commit is now stored in the user's
-    // // canonical repository.
-    // // We can now broadcast this on the firehose.
-    // fhp.commit(firehose::Commit {
-    //     car: mem,
-    //     ops,
-    //     cid: repo.root(),
-    //     rev: repo.commit().rev().to_string(),
-    //     did: atrium_api::types::string::Did::new(user.did()).expect("should be valid DID"),
-    //     pcid: Some(orig_cid),
-    //     blobs: blobs.into_iter().map(|(_, cid)| cid).collect::<Vec<_>>(),
-    // })
-    // .await;
-
-    // Ok(Json(
-    //     apply_writes::OutputData {
-    //         results: Some(res),
-    //         commit: Some(
-    //             CommitMetaData {
-    //                 cid: atrium_api::types::string::Cid::new(repo.root()),
-    //                 rev: repo.commit().rev(),
-    //             }
-    //             .into(),
-    //         ),
-    //     }
-    //     .into(),
-    // ))
+        let mut lock = sequencer.sequencer.write().await;
+        lock.sequence_commit(did.clone(), commit.clone()).await?;
+        account_manager
+            .update_repo_root(
+                did.to_string(),
+                commit.commit_data.cid,
+                commit.commit_data.rev,
+            )
+            .await?;
+        Ok(())
+    } else {
+        Err(Error::with_message(
+            StatusCode::NOT_FOUND,
+            anyhow!("Could not find repo: `{repo}`"),
+            ErrorMessage::new("RepoNotFound", "Could not find repo"),
+        ))
+    }
 }

src/main.rs

···
 //! PDS implementation.
+mod account_manager;
 mod actor_store;
 mod auth;
 mod config;
···
 mod mmap;
 mod oauth;
 mod plc;
-mod storage;
 #[cfg(test)]
 mod tests;
 
···
 ///
 /// We shouldn't have to know about any bsky endpoints to store private user data.
 /// This will _very likely_ be changed in the future.
-mod actor_endpoints {
-    use atrium_api::app::bsky::actor;
-    use axum::{Json, routing::post};
-    use constcat::concat;
-
-    use super::*;
-
-    async fn put_preferences(
-        user: AuthenticatedUser,
-        State(db): State<Db>,
-        Json(input): Json<actor::put_preferences::Input>,
-    ) -> Result<()> {
-        let did = user.did();
-        let prefs = sqlx::types::Json(input.preferences.clone());
-        _ = sqlx::query!(
-            r#"UPDATE accounts SET private_prefs = ? WHERE did = ?"#,
-            prefs,
-            did
-        )
-        .execute(&db)
-        .await
-        .context("failed to update user preferences")?;
-
-        Ok(())
-    }
-
-    async fn get_preferences(
-        user: AuthenticatedUser,
-        State(db): State<Db>,
-    ) -> Result<Json<actor::get_preferences::Output>> {
-        let did = user.did();
-        let json: Option<sqlx::types::Json<actor::defs::Preferences>> =
-            sqlx::query_scalar("SELECT private_prefs FROM accounts WHERE did = ?")
-                .bind(did)
-                .fetch_one(&db)
-                .await
-                .context("failed to fetch preferences")?;
-
-        if let Some(prefs) = json {
-            Ok(Json(
-                actor::get_preferences::OutputData {
-                    preferences: prefs.0,
-                }
-                .into(),
-            ))
-        } else {
-            Ok(Json(
-                actor::get_preferences::OutputData {
-                    preferences: Vec::new(),
-                }
-                .into(),
-            ))
-        }
-    }
-
-    /// Register all actor endpoints.
-    pub(crate) fn routes() -> Router<AppState> {
-        // AP /xrpc/app.bsky.actor.putPreferences
-        // AG /xrpc/app.bsky.actor.getPreferences
-        Router::new()
-            .route(
-                concat!("/", actor::put_preferences::NSID),
-                post(put_preferences),
-            )
-            .route(
-                concat!("/", actor::get_preferences::NSID),
-                get(get_preferences),
-            )
-    }
-}
+mod actor_endpoints;
 
 use anyhow::{Context as _, anyhow};
 use atrium_api::types::string::Did;
···
 use clap::Parser;
 use clap_verbosity_flag::{InfoLevel, Verbosity, log::LevelFilter};
 use config::AppConfig;
+use diesel::prelude::*;
+use diesel::r2d2::{self, ConnectionManager};
+use diesel_migrations::{EmbeddedMigrations, MigrationHarness, embed_migrations};
 #[expect(clippy::pub_use, clippy::useless_attribute)]
 pub use error::Error;
 use figment::{Figment, providers::Format as _};
···
 use http_cache_reqwest::{CacheMode, HttpCacheOptions, MokaManager};
 use rand::Rng as _;
 use serde::{Deserialize, Serialize};
-use sqlx::{SqlitePool, sqlite::SqliteConnectOptions};
 use std::{
     net::{IpAddr, Ipv4Addr, SocketAddr},
     path::PathBuf,
···
 /// The application user agent. Concatenates the package name and version. e.g. `bluepds/0.0.0`.
 pub const APP_USER_AGENT: &str = concat!(env!("CARGO_PKG_NAME"), "/", env!("CARGO_PKG_VERSION"),);
 
+/// Embedded migrations
+pub const MIGRATIONS: EmbeddedMigrations = embed_migrations!("./migrations");
+
 /// The application-wide result type.
 pub type Result<T> = std::result::Result<T, Error>;
 /// The reqwest client type with middleware.
 pub type Client = reqwest_middleware::ClientWithMiddleware;
 /// The database connection pool.
-pub type Db = SqlitePool;
+pub type Db = r2d2::Pool<ConnectionManager<SqliteConnection>>;
 /// The Azure credential type.
 pub type Cred = Arc<dyn TokenCredential>;
 
···
 
     let cred = azure_identity::DefaultAzureCredential::new()
         .context("failed to create Azure credential")?;
-    let opts = SqliteConnectOptions::from_str(&config.db)
-        .context("failed to parse database options")?
-        .create_if_missing(true);
-    let db = SqlitePool::connect_with(opts).await?;
 
-    sqlx::migrate!()
-        .run(&db)
-        .await
-        .context("failed to apply migrations")?;
+    // Create a database connection manager and pool
+    let manager = ConnectionManager::<SqliteConnection>::new(&config.db);
+    let db = r2d2::Pool::builder()
+        .build(manager)
+        .context("failed to create database connection pool")?;
+
+    // Apply pending migrations
+    let conn = &mut db
+        .get()
+        .context("failed to get database connection for migrations")?;
+    conn.run_pending_migrations(MIGRATIONS)
+        .expect("should be able to run migrations");
 
     let (_fh, fhp) = firehose::spawn(client.clone(), config.clone());
 
···
 
     // Determine whether or not this was the first startup (i.e. no accounts exist and no invite codes were created).
     // If so, create an invite code and share it via the console.
-    let c = sqlx::query_scalar!(
-        r#"
-        SELECT
-            (SELECT COUNT(*) FROM accounts) + (SELECT COUNT(*) FROM invites)
-            AS total_count
-        "#
+    let conn = &mut db.get().context("failed to get database connection")?;
+
+    #[derive(QueryableByName)]
+    struct TotalCount {
+        #[diesel(sql_type = diesel::sql_types::Integer)]
+        total_count: i32,
+    }
+
+    let result = diesel::sql_query(
+        "SELECT (SELECT COUNT(*) FROM accounts) + (SELECT COUNT(*) FROM invites) AS total_count",
     )
-    .fetch_one(&db)
-    .await
+    .get_result::<TotalCount>(conn)
     .context("failed to query database")?;
 
+    let c = result.total_count;
+
     #[expect(clippy::print_stdout)]
     if c == 0 {
         let uuid = Uuid::new_v4().to_string();
 
-        _ = sqlx::query!(
-            r#"
-            INSERT INTO invites (id, did, count, created_at)
-                VALUES (?, NULL, 1, datetime('now'))
-            "#,
-            uuid,
+        diesel::sql_query(
+            "INSERT INTO invites (id, did, count, created_at) VALUES (?, NULL, 1, datetime('now'))",
         )
-        .execute(&db)
-        .await
+        .bind::<diesel::sql_types::Text, _>(uuid.clone())
+        .execute(conn)
         .context("failed to create new invite code")?;
 
         // N.B: This is a sensitive message, so we're bypassing `tracing` here and

src/storage/car.rs

···
-//! CAR file-based repository storage
-
-use anyhow::{Context as _, Result};
-use atrium_repo::blockstore::{AsyncBlockStoreRead, AsyncBlockStoreWrite, CarStore};
-
-use crate::{config::RepoConfig, mmap::MappedFile};
-
-/// Open a CAR block store for a given DID.
-pub(crate) async fn open_car_store(
-    config: &RepoConfig,
-    did: impl AsRef<str>,
-) -> Result<impl AsyncBlockStoreRead + AsyncBlockStoreWrite> {
-    let id = did
-        .as_ref()
-        .strip_prefix("did:plc:")
-        .context("did in unknown format")?;
-
-    let p = config.path.join(id).with_extension("car");
-
-    let f = std::fs::File::options()
-        .read(true)
-        .write(true)
-        .open(p)
-        .context("failed to open repository file")?;
-    let f = MappedFile::new(f).context("failed to map repo")?;
-
-    CarStore::open(f).await.context("failed to open car store")
-}

src/storage/mod.rs

···
-//! `ATProto` user repository datastore functionality.
-
-pub(crate) mod car;
-mod sqlite;
-
-use anyhow::{Context as _, Result};
-use atrium_repo::{
-    Cid, Repository,
-    blockstore::{AsyncBlockStoreRead, AsyncBlockStoreWrite},
-};
-use std::str::FromStr as _;
-
-use crate::{Db, config::RepoConfig};
-
-// Re-export public items
-pub(crate) use car::open_car_store;
-pub(crate) use sqlite::{SQLiteStore, open_sqlite_store};
-
-/// Open a repository for a given DID.
-pub(crate) async fn open_repo_db(
-    config: &RepoConfig,
-    db: &Db,
-    did: impl Into<String>,
-) -> Result<Repository<impl AsyncBlockStoreRead + AsyncBlockStoreWrite>> {
-    let did = did.into();
-    let cid = sqlx::query_scalar!(
-        r#"
-        SELECT root FROM accounts
-        WHERE did = ?
-        "#,
-        did
-    )
-    .fetch_one(db)
-    .await
-    .context("failed to query database")?;
-
-    open_repo(
-        config,
-        did,
-        Cid::from_str(&cid).context("should be valid CID")?,
-    )
-    .await
-}
-
-/// Open a repository for a given DID and CID.
-pub(crate) async fn open_repo(
-    config: &RepoConfig,
-    did: impl Into<String>,
-    cid: Cid,
-) -> Result<Repository<impl AsyncBlockStoreRead + AsyncBlockStoreWrite>> {
-    let store = open_car_store(config, did.into()).await?;
-    Repository::open(store, cid)
-        .await
-        .context("failed to open repo")
-}
-/// Open a repository for a given DID and CID.
-/// SQLite backend.
-pub(crate) async fn open_repo_sqlite(
-    config: &RepoConfig,
-    did: impl Into<String>,
-    cid: Cid,
-) -> Result<Repository<impl AsyncBlockStoreRead + AsyncBlockStoreWrite>> {
-    let store = open_sqlite_store(config, did.into()).await?;
-    return Repository::open(store, cid)
-        .await
-        .context("failed to open repo");
-}
-
-/// Open a block store for a given DID.
-pub(crate) async fn open_store(
-    config: &RepoConfig,
-    did: impl Into<String>,
-) -> Result<impl AsyncBlockStoreRead + AsyncBlockStoreWrite> {
-    let did = did.into();
-
-    // if config.use_sqlite {
-    return open_sqlite_store(config, did.clone()).await;
-    // }
-    // Default to CAR store
-    // open_car_store(config, &did).await
-}
-
-/// Create a storage backend for a DID
-pub(crate) async fn create_storage_for_did(
-    config: &RepoConfig,
-    did_hash: &str,
-) -> Result<impl AsyncBlockStoreRead + AsyncBlockStoreWrite> {
-    // Use standard file structure but change extension based on type
-    // if config.use_sqlite {
-    // For SQLite, create a new database file
-    let db_path = config.path.join(format!("{}.db", did_hash));
-
-    // Ensure parent directory exists
-    if let Some(parent) = db_path.parent() {
-        tokio::fs::create_dir_all(parent)
-            .await
-            .context("failed to create directory")?;
-    }
-
-    // Create SQLite store
-    let pool = sqlx::sqlite::SqlitePoolOptions::new()
-        .max_connections(5)
-        .connect_with(
-            sqlx::sqlite::SqliteConnectOptions::new()
-                .filename(&db_path)
-                .create_if_missing(true),
-        )
-        .await
-        .context("failed to connect to SQLite database")?;
-
-    // Initialize tables
-    _ = sqlx::query(
-        "
-            CREATE TABLE IF NOT EXISTS blocks (
-                cid TEXT PRIMARY KEY NOT NULL,
-                data BLOB NOT NULL,
-                multicodec INTEGER NOT NULL,
-                multihash INTEGER NOT NULL
-            );
-            CREATE TABLE IF NOT EXISTS tree_nodes (
-                repo_did TEXT NOT NULL,
-                key TEXT NOT NULL,
-                value_cid TEXT NOT NULL,
-                PRIMARY KEY (repo_did, key),
-                FOREIGN KEY (value_cid) REFERENCES blocks(cid)
-            );
-            CREATE INDEX IF NOT EXISTS idx_blocks_cid ON blocks(cid);
-            CREATE INDEX IF NOT EXISTS idx_tree_nodes_repo ON tree_nodes(repo_did);
-            PRAGMA journal_mode=WAL;
-        ",
-    )
-    .execute(&pool)
-    .await
-    .context("failed to create tables")?;
-
-    Ok(SQLiteStore {
-        pool,
-        did: format!("did:plc:{}", did_hash),
-    })
-    // } else {
-    //     // For CAR files, create a new file
-    //     let file_path = config.path.join(format!("{}.car", did_hash));
-
-    //     // Ensure parent directory exists
-    //     if let Some(parent) = file_path.parent() {
-    //         tokio::fs::create_dir_all(parent)
-    //             .await
-    //             .context("failed to create directory")?;
-    //     }
-
-    //     let file = tokio::fs::File::create_new(file_path)
-    //         .await
-    //         .context("failed to create repo file")?;
-
-    //     CarStore::create(file)
-    //         .await
-    //         .context("failed to create carstore")
-    // }
-}

src/storage/sqlite.rs

···
-//! SQLite-based repository storage implementation.
-
-use anyhow::{Context as _, Result};
-use atrium_repo::{
-    Cid, Multihash,
-    blockstore::{AsyncBlockStoreRead, AsyncBlockStoreWrite, Error as BlockstoreError},
-};
-use sha2::Digest;
-use sqlx::SqlitePool;
-
-use crate::config::RepoConfig;
-
-/// SQLite-based implementation of block storage.
-pub(crate) struct SQLiteStore {
-    pub did: String,
-    pub pool: SqlitePool,
-}
-
-impl AsyncBlockStoreRead for SQLiteStore {
-    async fn read_block(&mut self, cid: Cid) -> Result<Vec<u8>, BlockstoreError> {
-        let mut contents = Vec::new();
-        self.read_block_into(cid, &mut contents).await?;
-        Ok(contents)
-    }
-    async fn read_block_into(
-        &mut self,
-        cid: Cid,
-        contents: &mut Vec<u8>,
-    ) -> Result<(), BlockstoreError> {
-        let cid_str = cid.to_string();
-        let record = sqlx::query!(r#"SELECT data FROM blocks WHERE cid = ?"#, cid_str)
-            .fetch_optional(&self.pool)
-            .await
-            .map_err(|e| BlockstoreError::Other(Box::new(e)))?
-            .ok_or(BlockstoreError::CidNotFound)?;
-
-        contents.clear();
-        contents.extend_from_slice(&record.data);
-        Ok(())
-    }
-}
-
-impl AsyncBlockStoreWrite for SQLiteStore {
-    async fn write_block(
-        &mut self,
-        codec: u64,
-        hash: u64,
-        contents: &[u8],
-    ) -> Result<Cid, BlockstoreError> {
-        let digest = match hash {
-            atrium_repo::blockstore::SHA2_256 => sha2::Sha256::digest(&contents),
-            _ => return Err(BlockstoreError::UnsupportedHash(hash)),
-        };
-
-        let multihash = Multihash::wrap(hash, digest.as_slice())
-            .map_err(|_| BlockstoreError::UnsupportedHash(hash))?;
-
-        let cid = Cid::new_v1(codec, multihash);
-        let cid_str = cid.to_string();
-
-        // Use a transaction for atomicity
-        let mut tx = self
-            .pool
-            .begin()
-            .await
-            .map_err(|e| BlockstoreError::Other(Box::new(e)))?;
-
-        // Check if block already exists
-        let exists = sqlx::query_scalar!(r#"SELECT COUNT(*) FROM blocks WHERE cid = ?"#, cid_str)
-            .fetch_one(&mut *tx)
-            .await
-            .map_err(|e| BlockstoreError::Other(Box::new(e)))?;
-
-        // Only insert if block doesn't exist
-        let codec = codec as i64;
-        let hash = hash as i64;
-        if exists == 0 {
-            _ = sqlx::query!(
-                r#"INSERT INTO blocks (cid, data, multicodec, multihash) VALUES (?, ?, ?, ?)"#,
-                cid_str,
-                contents,
-                codec,
-                hash
-            )
-            .execute(&mut *tx)
-            .await
-            .map_err(|e| BlockstoreError::Other(Box::new(e)))?;
-        }
-
-        tx.commit()
-            .await
-            .map_err(|e| BlockstoreError::Other(Box::new(e)))?;
-
-        Ok(cid)
-    }
-}
-
-/// Open a SQLite store for the given DID.
-pub(crate) async fn open_sqlite_store(
-    config: &RepoConfig,
-    did: impl Into<String>,
-) -> Result<impl AsyncBlockStoreRead + AsyncBlockStoreWrite> {
-    tracing::info!("Opening SQLite store for DID");
-    let did_str = did.into();
-
-    // Extract the PLC ID from the DID
-    let id = did_str
-        .strip_prefix("did:plc:")
-        .context("DID in unknown format")?;
-
-    // Create database connection pool
-    let db_path = config.path.join(format!("{id}.db"));
-
-    let pool = sqlx::sqlite::SqlitePoolOptions::new()
-        .max_connections(5)
-        .connect_with(
-            sqlx::sqlite::SqliteConnectOptions::new()
-                .filename(&db_path)
-                .create_if_missing(true),
-        )
-        .await
-        .context("failed to connect to SQLite database")?;
-
-    // Ensure tables exist
-    _ = sqlx::query(
-        "
-        CREATE TABLE IF NOT EXISTS blocks (
-            cid TEXT PRIMARY KEY NOT NULL,
-            data BLOB NOT NULL,
-            multicodec INTEGER NOT NULL,
-            multihash INTEGER NOT NULL
-        );
-        CREATE TABLE IF NOT EXISTS tree_nodes (
-            repo_did TEXT NOT NULL,
-            key TEXT NOT NULL,
-            value_cid TEXT NOT NULL,
-            PRIMARY KEY (repo_did, key),
-            FOREIGN KEY (value_cid) REFERENCES blocks(cid)
-        );
-        CREATE INDEX IF NOT EXISTS idx_blocks_cid ON blocks(cid);
-        CREATE INDEX IF NOT EXISTS idx_tree_nodes_repo ON tree_nodes(repo_did);
-        ",
-    )
-    .execute(&pool)
-    .await
-    .context("failed to create tables")?;
-
-    Ok(SQLiteStore { pool, did: did_str })
-}