reorganize; apis/com/atproto/repo/apply_writes prototype

src/account_manager/mod.rs

···
 use crate::account_manager::helpers::password::UpdateUserPasswordOpts;
 use crate::models::pds::EmailTokenPurpose;
 use anyhow::Result;
+use axum::extract::FromRef;
 use chrono::DateTime;
 use chrono::offset::Utc as UtcOffset;
 use cidv10::Cid;
···
 }
 
 pub struct SharedAccountManager {
-    pub account_manager: RwLock<AccountManagerCreator>,
+    pub account_manager: RwLock<AccountManager>,
 }

src/actor_store/mod.rs

···
 use sql_blob::BlobStoreSql;
 use sql_repo::SqlRepoReader;
 
+use crate::ActorPools;
+
 #[derive(Debug)]
 enum FormatCommitError {
     BadRecordSwap(String),
···
             did,
             blob: BlobReader::new(blobstore, db),
         }
+    }
+
+    /// Create a new ActorStore taking ActorPools HashMap as input
+    pub async fn from_actor_pools(
+        did: &String,
+        hashmap_actor_pools: &std::collections::HashMap<String, ActorPools>,
+    ) -> Self {
+        let actor_pool = hashmap_actor_pools
+            .get(did)
+            .expect("Actor pool not found")
+            .clone();
+        let blobstore = BlobStoreSql::new(did.clone(), actor_pool.blob);
+        let conn = actor_pool
+            .repo
+            .clone()
+            .get()
+            .await
+            .expect("Failed to get connection");
+        Self::new(did.clone(), blobstore, actor_pool.repo, conn)
     }
 
     pub async fn get_repo_root(&self) -> Option<Cid> {

src/actor_store/sql_blob.rs

···
     }
 
     // /// Create a factory function for blob stores
-    // pub fn creator(
-    //     db: deadpool_diesel::Pool<
-    //     deadpool_diesel::Manager<SqliteConnection>,
-    //     deadpool_diesel::sqlite::Object,
-    // >,
-    // ) -> Box<dyn Fn(String) -> BlobStoreSql> {
-    //     let db_clone = db.clone();
-    //     Box::new(move |did: String| BlobStoreSql::new(did, db_clone.clone()))
-    // }
+    pub fn creator(
+        db: deadpool_diesel::Pool<
+            deadpool_diesel::Manager<SqliteConnection>,
+            deadpool_diesel::sqlite::Object,
+        >,
+    ) -> Box<dyn Fn(String) -> BlobStoreSql> {
+        let db_clone = db.clone();
+        Box::new(move |did: String| BlobStoreSql::new(did, db_clone.clone()))
+    }
 
     /// Store a blob temporarily - now just stores permanently with a key returned for API compatibility
     pub async fn put_temp(&self, bytes: Vec<u8>) -> Result<String> {
         // Generate a unique key as a CID based on the data
-        use sha2::{Digest, Sha256};
-        let digest = Sha256::digest(&bytes);
-        let key = hex::encode(digest);
+        // use sha2::{Digest, Sha256};
+        // let digest = Sha256::digest(&bytes);
+        // let key = hex::encode(digest);
+        let key = rsky_common::get_random_str();
 
         // Just store the blob directly
         self.put_permanent_with_mime(

src/apis/com/atproto/mod.rs

···
+// pub mod admin;
+// pub mod identity;
+pub mod repo;
+// pub mod server;
+// pub mod sync;

src/apis/com/atproto/repo/mod.rs

···
+use atrium_api::com::atproto::repo;
+use axum::{Router, routing::post};
+use constcat::concat;
+
+use crate::AppState;
+
+pub mod apply_writes;
+// pub mod create_record;
+// pub mod delete_record;
+// pub mod describe_repo;
+// pub mod get_record;
+// pub mod import_repo;
+// pub mod list_missing_blobs;
+// pub mod list_records;
+// pub mod put_record;
+// pub mod upload_blob;
+
+/// These endpoints are part of the atproto PDS repository management APIs. \
+/// Requests usually require authentication (unlike the com.atproto.sync.* endpoints), and are made directly to the user's own PDS instance.
+/// ### Routes
+/// - AP /xrpc/com.atproto.repo.applyWrites     -> [`apply_writes`]
+/// - AP /xrpc/com.atproto.repo.createRecord    -> [`create_record`]
+/// - AP /xrpc/com.atproto.repo.putRecord       -> [`put_record`]
+/// - AP /xrpc/com.atproto.repo.deleteRecord    -> [`delete_record`]
+/// - AP /xrpc/com.atproto.repo.uploadBlob      -> [`upload_blob`]
+/// - UG /xrpc/com.atproto.repo.describeRepo    -> [`describe_repo`]
+/// - UG /xrpc/com.atproto.repo.getRecord       -> [`get_record`]
+/// - UG /xrpc/com.atproto.repo.listRecords     -> [`list_records`]
+///     - [ ] xx /xrpc/com.atproto.repo.importRepo
+// - [ ] xx /xrpc/com.atproto.repo.listMissingBlobs
+pub(crate) fn routes() -> Router<AppState> {
+    Router::new().route(
+        concat!("/", repo::apply_writes::NSID),
+        post(apply_writes::apply_writes),
+    )
+    // .route(concat!("/", repo::create_record::NSID), post(create_record))
+    // .route(concat!("/", repo::put_record::NSID), post(put_record))
+    // .route(concat!("/", repo::delete_record::NSID), post(delete_record))
+    // .route(concat!("/", repo::upload_blob::NSID), post(upload_blob))
+    // .route(concat!("/", repo::describe_repo::NSID), get(describe_repo))
+    // .route(concat!("/", repo::get_record::NSID), get(get_record))
+    // .route(concat!("/", repo::import_repo::NSID), post(todo))
+    // .route(concat!("/", repo::list_missing_blobs::NSID), get(todo))
+    // .route(concat!("/", repo::list_records::NSID), get(list_records))
+}

src/apis/com/mod.rs

···
+pub mod atproto;

src/endpoints/mod.rs src/apis/mod.rs

···
 //! Root module for all endpoints.
 // mod identity;
-// mod repo;
+mod com;
 // mod server;
 // mod sync;
 
···
 
 /// 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
+    Router::new()
+        .route("/_health", get(health))
+        // .merge(identity::routes()) // com.atproto.identity
+        .merge(com::atproto::repo::routes()) // com.atproto.repo
     // .merge(server::routes()) // com.atproto.server
     // .merge(sync::routes()) // com.atproto.sync
 }

src/endpoints/repo/apply_writes.rs src/apis/com/atproto/repo/apply_writes.rs

···
 //! Apply a batch transaction of repository creates, updates, and deletes. Requires auth, implemented by PDS.
+use crate::SharedSequencer;
+use crate::account_manager::helpers::account::AvailabilityFlags;
+use crate::account_manager::{AccountManager, AccountManagerCreator, SharedAccountManager};
 use crate::{
-    ActorPools, AppState, Db, Error, Result, SigningKey,
+    ActorPools, AppState, SigningKey,
     actor_store::{ActorStore, 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,
+    error::{ApiError, ErrorMessage},
 };
-use anyhow::bail;
-use anyhow::{Context as _, anyhow};
-use atrium_api::com::atproto::repo::apply_writes::{self, InputWritesItem, OutputResultsItem};
-use atrium_api::{
-    com::atproto::repo::{self, defs::CommitMetaData},
-    types::{
-        LimitedU32, Object, TryFromUnknown as _, TryIntoUnknown as _, Unknown,
-        string::{AtIdentifier, Nsid, Tid},
-    },
-};
-use atrium_repo::blockstore::CarStore;
+use anyhow::{Result, bail};
 use axum::{
     Json, Router,
     body::Body,
···
     routing::{get, post},
 };
 use cidv10::Cid;
-use constcat::concat;
-use futures::TryStreamExt as _;
+use deadpool_diesel::sqlite::Pool;
 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_pds::sequencer::Sequencer;
 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 std::str::FromStr;
+use std::sync::Arc;
+use tokio::sync::RwLock;
 
-use super::resolve_did;
-
-/// Apply a batch transaction of repository creates, updates, and deletes. Requires auth, implemented by PDS.
-/// - POST /xrpc/com.atproto.repo.applyWrites
-/// ### Request Body
-/// - `repo`: `at-identifier` // The handle or DID of the repo (aka, current account).
-/// - `validate`: `boolean` // Can be set to 'false' to skip Lexicon schema validation of record data across all operations, 'true' to require it, or leave unset to validate only for known Lexicons.
-/// - `writes`: `object[]` // One of:
-/// - - com.atproto.repo.applyWrites.create
-/// - - com.atproto.repo.applyWrites.update
-/// - - com.atproto.repo.applyWrites.delete
-/// - `swap_commit`: `cid` // If provided, the entire operation will fail if the current repo commit CID does not match this value. Used to prevent conflicting repo mutations.
-pub(crate) async fn apply_writes(
+async fn inner_apply_writes(
+    body: ApplyWritesInput,
     user: AuthenticatedUser,
-    State(skey): State<SigningKey>,
-    State(config): State<AppConfig>,
-    State(db): State<Db>,
-    State(db_actors): State<std::collections::HashMap<String, ActorPools>>,
-    State(fhp): State<FirehoseProducer>,
-    Json(input): Json<ApplyWritesInput>,
-) -> Result<Json<repo::apply_writes::Output>> {
-    let tx: ApplyWritesInput = input;
+    sequencer: &RwLock<Sequencer>,
+    actor_pools: std::collections::HashMap<String, ActorPools>,
+    account_manager: &RwLock<AccountManager>,
+) -> Result<()> {
+    let tx: ApplyWritesInput = body;
     let ApplyWritesInput {
         repo,
         validate,
···
         ..
     } = tx;
     let account = account_manager
+        .read()
+        .await
         .get_account(
             &repo,
             Some(AvailabilityFlags {
···
 
     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"),
-            ));
+            bail!("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"),
-            ));
+            bail!("AuthRequiredError")
         }
         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"),
-            ));
+            bail!("Too many writes. Max: 200")
         }
 
         let writes: Vec<PreparedWrite> = stream::iter(tx.writes)
···
             None => None,
         };
 
-        let actor_db = db_actors
-            .get(did)
-            .ok_or_else(|| anyhow!("Actor DB not found"))?;
-        let conn = actor_db
-            .repo
-            .get()
-            .await
-            .context("Failed to get actor db connection")?;
-        let mut actor_store = ActorStore::new(
-            did.clone(),
-            BlobStoreSql::new(did.clone(), actor_db.blob),
-            actor_db.repo,
-            conn,
-        );
+        let mut actor_store = ActorStore::from_actor_pools(did, &actor_pools).await;
 
         let commit = actor_store
             .process_writes(writes.clone(), swap_commit_cid)
             .await?;
 
-        let mut lock = sequencer.sequencer.write().await;
-        lock.sequence_commit(did.clone(), commit.clone()).await?;
+        sequencer
+            .write()
+            .await
+            .sequence_commit(did.clone(), commit.clone())
+            .await?;
         account_manager
+            .write()
+            .await
             .update_repo_root(
                 did.to_string(),
                 commit.commit_data.cid,
···
             .await?;
         Ok(())
     } else {
-        Err(Error::with_message(
-            StatusCode::NOT_FOUND,
-            anyhow!("Could not find repo: `{repo}`"),
-            ErrorMessage::new("RepoNotFound", "Could not find repo"),
-        ))
+        bail!("Could not find repo: `{repo}`")
+    }
+}
+
+/// Apply a batch transaction of repository creates, updates, and deletes. Requires auth, implemented by PDS.
+/// - POST /xrpc/com.atproto.repo.applyWrites
+/// ### Request Body
+/// - `repo`: `at-identifier` // The handle or DID of the repo (aka, current account).
+/// - `validate`: `boolean` // Can be set to 'false' to skip Lexicon schema validation of record data across all operations, 'true' to require it, or leave unset to validate only for known Lexicons.
+/// - `writes`: `object[]` // One of:
+/// - - com.atproto.repo.applyWrites.create
+/// - - com.atproto.repo.applyWrites.update
+/// - - com.atproto.repo.applyWrites.delete
+/// - `swap_commit`: `cid` // If provided, the entire operation will fail if the current repo commit CID does not match this value. Used to prevent conflicting repo mutations.
+#[axum::debug_handler(state = AppState)]
+pub(crate) async fn apply_writes(
+    user: AuthenticatedUser,
+    State(state): State<AppState>,
+    Json(body): Json<ApplyWritesInput>,
+) -> Result<(), ApiError> {
+    tracing::debug!("@LOG: debug apply_writes {body:#?}");
+    let db_actors = state.db_actors.clone();
+    let sequencer = &state.sequencer.sequencer;
+    let account_manager = &state.account_manager.account_manager;
+    match inner_apply_writes(
+        body,
+        user,
+        sequencer.clone(),
+        db_actors,
+        account_manager.clone(),
+    )
+    .await
+    {
+        Ok(()) => Ok(()),
+        Err(error) => {
+            tracing::error!("@LOG: ERROR: {error}");
+            Err(ApiError::RuntimeError)
+        }
     }
 }

src/error.rs

···
     http::StatusCode,
     response::{IntoResponse, Response},
 };
+use rsky_pds::handle::{self, errors::ErrorKind};
 use thiserror::Error;
 use tracing::error;
 
···
         }
     }
 }
+
+/// API error types that can be returned to clients
+#[derive(Clone, Debug)]
+pub enum ApiError {
+    RuntimeError,
+    InvalidLogin,
+    AccountTakendown,
+    InvalidRequest(String),
+    ExpiredToken,
+    InvalidToken,
+    RecordNotFound,
+    InvalidHandle,
+    InvalidEmail,
+    InvalidPassword,
+    InvalidInviteCode,
+    HandleNotAvailable,
+    EmailNotAvailable,
+    UnsupportedDomain,
+    UnresolvableDid,
+    IncompatibleDidDoc,
+    WellKnownNotFound,
+    AccountNotFound,
+    BlobNotFound,
+    BadRequest(String, String),
+    AuthRequiredError(String),
+}
+
+impl ApiError {
+    /// Get the appropriate HTTP status code for this error
+    fn status_code(&self) -> StatusCode {
+        match self {
+            Self::RuntimeError => StatusCode::INTERNAL_SERVER_ERROR,
+            Self::InvalidLogin
+            | Self::ExpiredToken
+            | Self::InvalidToken
+            | Self::AuthRequiredError(_) => StatusCode::UNAUTHORIZED,
+            Self::AccountTakendown => StatusCode::FORBIDDEN,
+            Self::RecordNotFound
+            | Self::WellKnownNotFound
+            | Self::AccountNotFound
+            | Self::BlobNotFound => StatusCode::NOT_FOUND,
+            // All bad requests grouped together
+            _ => StatusCode::BAD_REQUEST,
+        }
+    }
+
+    /// Get the error type string for API responses
+    fn error_type(&self) -> String {
+        match self {
+            Self::RuntimeError => "InternalServerError",
+            Self::InvalidLogin => "InvalidLogin",
+            Self::AccountTakendown => "AccountTakendown",
+            Self::InvalidRequest(_) => "InvalidRequest",
+            Self::ExpiredToken => "ExpiredToken",
+            Self::InvalidToken => "InvalidToken",
+            Self::RecordNotFound => "RecordNotFound",
+            Self::InvalidHandle => "InvalidHandle",
+            Self::InvalidEmail => "InvalidEmail",
+            Self::InvalidPassword => "InvalidPassword",
+            Self::InvalidInviteCode => "InvalidInviteCode",
+            Self::HandleNotAvailable => "HandleNotAvailable",
+            Self::EmailNotAvailable => "EmailNotAvailable",
+            Self::UnsupportedDomain => "UnsupportedDomain",
+            Self::UnresolvableDid => "UnresolvableDid",
+            Self::IncompatibleDidDoc => "IncompatibleDidDoc",
+            Self::WellKnownNotFound => "WellKnownNotFound",
+            Self::AccountNotFound => "AccountNotFound",
+            Self::BlobNotFound => "BlobNotFound",
+            Self::BadRequest(error, _) => error,
+            Self::AuthRequiredError(_) => "AuthRequiredError",
+        }
+        .to_string()
+    }
+
+    /// Get the user-facing error message
+    fn message(&self) -> String {
+        match self {
+            Self::RuntimeError => "Something went wrong",
+            Self::InvalidLogin => "Invalid identifier or password",
+            Self::AccountTakendown => "Account has been taken down",
+            Self::InvalidRequest(msg) => msg,
+            Self::ExpiredToken => "Token is expired",
+            Self::InvalidToken => "Token is invalid",
+            Self::RecordNotFound => "Record could not be found",
+            Self::InvalidHandle => "Handle is invalid",
+            Self::InvalidEmail => "Invalid email",
+            Self::InvalidPassword => "Invalid Password",
+            Self::InvalidInviteCode => "Invalid invite code",
+            Self::HandleNotAvailable => "Handle not available",
+            Self::EmailNotAvailable => "Email not available",
+            Self::UnsupportedDomain => "Unsupported domain",
+            Self::UnresolvableDid => "Unresolved Did",
+            Self::IncompatibleDidDoc => "IncompatibleDidDoc",
+            Self::WellKnownNotFound => "User not found",
+            Self::AccountNotFound => "Account could not be found",
+            Self::BlobNotFound => "Blob could not be found",
+            Self::BadRequest(_, msg) => msg,
+            Self::AuthRequiredError(msg) => msg,
+        }
+        .to_string()
+    }
+}
+
+impl From<Error> for ApiError {
+    fn from(_value: Error) -> Self {
+        ApiError::RuntimeError
+    }
+}
+
+impl From<handle::errors::Error> for ApiError {
+    fn from(value: handle::errors::Error) -> Self {
+        match value.kind {
+            ErrorKind::InvalidHandle => ApiError::InvalidHandle,
+            ErrorKind::HandleNotAvailable => ApiError::HandleNotAvailable,
+            ErrorKind::UnsupportedDomain => ApiError::UnsupportedDomain,
+            ErrorKind::InternalError => ApiError::RuntimeError,
+        }
+    }
+}
+
+impl IntoResponse for ApiError {
+    fn into_response(self) -> Response {
+        let status = self.status_code();
+        let error_type = self.error_type();
+        let message = self.message();
+
+        // Log the error for debugging
+        error!("API Error: {}: {}", error_type, message);
+
+        // Create the error message and serialize to JSON
+        let error_message = ErrorMessage::new(error_type, message);
+        let body = serde_json::to_string(&error_message).unwrap_or_else(|_| {
+            r#"{"error":"InternalServerError","message":"Error serializing response"}"#.to_string()
+        });
+
+        // Build the response
+        Response::builder()
+            .status(status)
+            .header("Content-Type", "application/json")
+            .body(Body::new(body))
+            .expect("should be a valid response")
+    }
+}

src/lib.rs

···
 mod account_manager;
 mod actor_endpoints;
 mod actor_store;
+mod apis;
 mod auth;
 mod config;
 mod db;
 mod did;
-mod endpoints;
 pub mod error;
 mod firehose;
 mod metrics;
···
     tokio::fs::create_dir_all(&config.plc.path).await?;
     tokio::fs::create_dir_all(&config.blob.path).await?;
 
-    let cred = azure_identity::DefaultAzureCredential::new()
-        .context("failed to create Azure credential")?;
-
     // Create a database connection manager and pool for the main database.
     let pool =
         establish_pool(&config.db).context("failed to establish database connection pool")?;
···
         )),
     });
     let account_manager = SharedAccountManager {
-        account_manager: RwLock::new(AccountManager::creator()),
+        account_manager: RwLock::new(AccountManager::new(pool.clone())),
     };
 
     let addr = config
···
         .merge(oauth::routes())
         .nest(
             "/xrpc",
-            endpoints::routes()
+            apis::routes()
                 .merge(actor_endpoints::routes())
                 .fallback(service_proxy),
         )
···
         .and_then(|r| r)
         .context("failed to serve app")
 }
-
-/// Creates an app router with the provided AppState.
-pub fn create_app(state: AppState) -> Router {
-    Router::new()
-        .route("/", get(index))
-        .merge(oauth::routes())
-        .nest(
-            "/xrpc",
-            endpoints::routes()
-                .merge(actor_endpoints::routes())
-                .fallback(service_proxy),
-        )
-        .layer(CorsLayer::permissive())
-        .layer(TraceLayer::new_for_http())
-        .with_state(state)
-}