prototype spis com atproto repo

src/actor_store/blob.rs

···
 
 use crate::models::actor_store as models;
 use anyhow::{Result, bail};
+use axum::body::Bytes;
 use cidv10::Cid;
 use diesel::dsl::{count_distinct, exists, not};
 use diesel::sql_types::{Integer, Nullable, Text};
···
     pub async fn upload_blob_and_get_metadata(
         &self,
         user_suggested_mime: String,
-        blob: Vec<u8>,
+        blob: Bytes,
     ) -> Result<BlobMetadata> {
         let bytes = blob;
         let size = bytes.len() as i64;
 
         let (temp_key, sha256, img_info, sniffed_mime) = try_join!(
             self.blobstore.put_temp(bytes.clone()),
-            sha256_stream(bytes.clone()),
-            image::maybe_get_info(bytes.clone()),
-            image::mime_type_from_bytes(bytes.clone())
+            // TODO: reimpl funcs to use Bytes instead of Vec<u8>
+            sha256_stream(bytes.to_vec()),
+            image::maybe_get_info(bytes.to_vec()),
+            image::mime_type_from_bytes(bytes.to_vec())
         )?;
 
         let cid = sha256_raw_to_cid(sha256);

src/actor_store/blob_fs.rs

···
 //! File system implementation of blob storage
 //! Based on the S3 implementation but using local file system instead
 use anyhow::Result;
+use axum::body::Bytes;
 use cidv10::Cid;
 use rsky_common::get_random_str;
 use rsky_repo::error::BlobError;
···
 
 /// ByteStream implementation for blob data
 pub struct ByteStream {
-    pub bytes: Vec<u8>,
+    pub bytes: Bytes,
 }
 
 impl ByteStream {
     /// Create a new ByteStream with the given bytes
-    pub const fn new(bytes: Vec<u8>) -> Self {
+    pub const fn new(bytes: Bytes) -> Self {
         Self { bytes }
     }
 
     /// Collect the bytes from the stream
-    pub async fn collect(self) -> Result<Vec<u8>> {
+    pub async fn collect(self) -> Result<Bytes> {
         Ok(self.bytes)
     }
 }
···
     }
 
     /// Store a blob temporarily
-    pub async fn put_temp(&self, bytes: Vec<u8>) -> Result<String> {
+    pub async fn put_temp(&self, bytes: Bytes) -> Result<String> {
         let key = self.gen_key();
         let temp_path = self.get_tmp_path(&key);
 
···
     }
 
     /// Store a blob directly as permanent
-    pub async fn put_permanent(&self, cid: Cid, bytes: Vec<u8>) -> Result<()> {
+    pub async fn put_permanent(&self, cid: Cid, bytes: Bytes) -> Result<()> {
         let target_path = self.get_stored_path(cid);
 
         // Ensure the directory exists
···
         let blob_path = self.get_stored_path(cid);
 
         match async_fs::read(&blob_path).await {
-            Ok(bytes) => Ok(ByteStream::new(bytes)),
+            Ok(bytes) => Ok(ByteStream::new(Bytes::from(bytes))),
             Err(e) => {
                 error!("Failed to read blob at path {:?}: {}", blob_path, e);
                 Err(anyhow::Error::new(BlobError::BlobNotFoundError))
···
     }
 
     /// Get blob bytes
-    pub async fn get_bytes(&self, cid: Cid) -> Result<Vec<u8>> {
+    pub async fn get_bytes(&self, cid: Cid) -> Result<Bytes> {
         let stream = self.get_object(cid).await?;
         stream.collect().await
     }

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::account_manager::AccountManager;
-use crate::account_manager::helpers::account::AvailabilityFlags;
-use crate::{
-    actor_store::ActorStore,
-    auth::AuthenticatedUser,
-    error::ApiError,
-    serve::{ActorStorage, AppState},
-};
-use anyhow::{Result, bail};
-use axum::{Json, extract::State};
-use cidv10::Cid;
-use futures::stream::{self, StreamExt};
-use rsky_lexicon::com::atproto::repo::{ApplyWritesInput, ApplyWritesInputRefWrite};
-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 std::collections::HashMap;
-use std::hash::RandomState;
-use std::str::FromStr;
-use std::sync::Arc;
-use tokio::sync::RwLock;
+
+use super::*;
 
 async fn inner_apply_writes(
     body: ApplyWritesInput,
-    user: AuthenticatedUser,
+    auth: AuthenticatedUser,
     sequencer: Arc<RwLock<Sequencer>>,
     actor_pools: HashMap<String, ActorStorage>,
     account_manager: Arc<RwLock<AccountManager>>,
···
             bail!("Account is deactivated")
         }
         let did = account.did;
-        if did != user.did() {
+        if did != auth.did() {
             bail!("AuthRequiredError")
         }
         let did: &String = &did;
···
         }
 
         let writes: Vec<PreparedWrite> = stream::iter(tx.writes)
-            .then(|write| async move {
+            .then(async |write| {
                 Ok::<PreparedWrite, anyhow::Error>(match write {
                     ApplyWritesInputRefWrite::Create(write) => PreparedWrite::Create(
                         prepare_create(PrepareCreateOpts {
···
 /// - `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(db_actors): State<HashMap<String, ActorStorage, RandomState>>,
+    auth: AuthenticatedUser,
+    State(actor_pools): State<HashMap<String, ActorStorage, RandomState>>,
     State(account_manager): State<Arc<RwLock<AccountManager>>>,
     State(sequencer): State<Arc<RwLock<Sequencer>>>,
     Json(body): Json<ApplyWritesInput>,
 ) -> Result<(), ApiError> {
     tracing::debug!("@LOG: debug apply_writes {body:#?}");
-    match inner_apply_writes(body, user, sequencer, db_actors, account_manager).await {
+    match inner_apply_writes(body, auth, sequencer, actor_pools, account_manager).await {
         Ok(()) => Ok(()),
         Err(error) => {
             tracing::error!("@LOG: ERROR: {error}");

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

···
 //! Create a single new repository record. Requires auth, implemented by PDS.
-use crate::account_manager::AccountManager;
-use crate::account_manager::helpers::account::AvailabilityFlags;
-use crate::{
-    actor_store::ActorStore,
-    auth::AuthenticatedUser,
-    error::ApiError,
-    serve::{ActorStorage, AppState},
-};
-use anyhow::{Result, bail};
-use axum::{Json, extract::State};
-use cidv10::Cid;
-use rsky_lexicon::com::atproto::repo::{CreateRecordInput, CreateRecordOutput};
-use rsky_pds::SharedIdResolver;
-use rsky_pds::repo::prepare::{
-    PrepareCreateOpts, PrepareDeleteOpts, prepare_create, prepare_delete,
-};
-use rsky_pds::sequencer::Sequencer;
-use rsky_repo::types::{PreparedDelete, PreparedWrite};
-use rsky_syntax::aturi::AtUri;
-use std::collections::HashMap;
-use std::hash::RandomState;
-use std::str::FromStr;
-use std::sync::Arc;
-use tokio::sync::RwLock;
+
+use super::*;
 
 async fn inner_create_record(
     body: CreateRecordInput,
     user: AuthenticatedUser,
     sequencer: Arc<RwLock<Sequencer>>,
-    actor_pools: std::collections::HashMap<String, ActorStorage>,
+    actor_pools: HashMap<String, ActorStorage>,
     account_manager: Arc<RwLock<AccountManager>>,
 ) -> Result<CreateRecordOutput> {
     let CreateRecordInput {

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

···
-/// Delete a repository record, or ensure it doesn't exist. Requires auth, implemented by PDS.
-use crate::account_manager::AccountManager;
-use crate::account_manager::helpers::account::AvailabilityFlags;
-use crate::{
-    actor_store::ActorStore,
-    auth::AuthenticatedUser,
-    error::ApiError,
-    serve::{ActorStorage, AppState},
-};
-use anyhow::{Result, bail};
-use axum::{Json, extract::State};
-use cidv10::Cid;
-use rsky_lexicon::com::atproto::repo::DeleteRecordInput;
-use rsky_pds::repo::prepare::{PrepareDeleteOpts, prepare_delete};
-use rsky_pds::sequencer::Sequencer;
-use rsky_repo::types::PreparedWrite;
-use rsky_syntax::aturi::AtUri;
-use std::collections::HashMap;
-use std::hash::RandomState;
-use std::str::FromStr;
-use std::sync::Arc;
-use tokio::sync::RwLock;
+//! Delete a repository record, or ensure it doesn't exist. Requires auth, implemented by PDS.
+use super::*;
 
 async fn inner_delete_record(
     body: DeleteRecordInput,

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

···
 //! Get information about an account and repository, including the list of collections. Does not require auth.
-use crate::account_manager::AccountManager;
-use crate::serve::ActorStorage;
-use crate::{actor_store::ActorStore, error::ApiError, serve::AppState};
-use anyhow::{Result, bail};
-use axum::extract::Query;
-use axum::{Json, extract::State};
-use rsky_identity::IdResolver;
-use rsky_identity::types::DidDocument;
-use rsky_lexicon::com::atproto::repo::DescribeRepoOutput;
-use rsky_syntax::handle::INVALID_HANDLE;
-use std::collections::HashMap;
-use std::hash::RandomState;
-use std::sync::Arc;
-use tokio::sync::RwLock;
+use super::*;
 
 async fn inner_describe_repo(
     repo: String,
···
     match account {
         None => bail!("Cound not find user: `{repo}`"),
         Some(account) => {
-            let mut lock = id_resolver.write().await;
-            let did_doc: DidDocument = match lock.did.ensure_resolve(&account.did, None).await {
+            let did_doc: DidDocument = match id_resolver
+                .write()
+                .await
+                .did
+                .ensure_resolve(&account.did, None)
+                .await
+            {
                 Err(err) => bail!("Could not resolve DID: `{err}`"),
                 Ok(res) => res,
             };
···
             let collections = actor_store.record.list_collections().await?;
 
             Ok(DescribeRepoOutput {
-                handle: account.handle.unwrap_or(INVALID_HANDLE.to_string()),
+                handle: account.handle.unwrap_or_else(|| INVALID_HANDLE.to_owned()),
                 did: account.did,
                 did_doc: serde_json::to_value(did_doc)?,
                 collections,
···
 #[tracing::instrument(skip_all)]
 #[axum::debug_handler(state = AppState)]
 pub async fn describe_repo(
-    Query(input): Query<atrium_api::com::atproto::repo::describe_repo::ParametersData>,
+    Query(input): Query<atrium_repo::describe_repo::ParametersData>,
     State(db_actors): State<HashMap<String, ActorStorage, RandomState>>,
     State(account_manager): State<Arc<RwLock<AccountManager>>>,
     State(id_resolver): State<Arc<RwLock<IdResolver>>>,

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

···
 #[tracing::instrument(skip_all)]
 #[axum::debug_handler(state = AppState)]
 pub async fn fun(
+    auth: AuthenticatedUser,
     Query(input): Query<atrium_api::com::atproto::repo::describe_repo::ParametersData>,
-    State(db_actors): State<HashMap<String, ActorStorage, RandomState>>,
+    State(actor_pools): State<HashMap<String, ActorStorage, RandomState>>,
     State(account_manager): State<Arc<RwLock<AccountManager>>>,
     State(id_resolver): State<Arc<RwLock<IdResolver>>>,
     State(sequencer): State<Arc<RwLock<Sequencer>>>,
+    Json(body): Json<ApplyWritesInput>,
 ) -> Result<Json<_>, ApiError> {
     todo!();
 }

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

···
 //! Get a single record from a repository. Does not require auth.
-use crate::account_manager::AccountManager;
-use crate::serve::ActorStorage;
-use crate::{actor_store::ActorStore, error::ApiError, serve::AppState};
-use anyhow::{Result, bail};
-use axum::extract::Query;
-use axum::{Json, extract::State};
-use rsky_identity::IdResolver;
-use rsky_lexicon::com::atproto::repo::GetRecordOutput;
-use rsky_pds::pipethrough::{OverrideOpts, ProxyRequest, pipethrough};
-use rsky_pds::sequencer::Sequencer;
-use rsky_syntax::aturi::AtUri;
-use std::collections::HashMap;
-use std::hash::RandomState;
-use std::sync::Arc;
-use tokio::sync::RwLock;
+use super::*;
 
 async fn inner_get_record(
     repo: String,

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

···
-use crate::account_manager::AccountManager;
-use crate::auth::AuthenticatedUser;
-use crate::serve::ActorStorage;
-use crate::{actor_store::ActorStore, error::ApiError, serve::AppState};
-use anyhow::{Result, bail};
-use axum::extract::Query;
-use axum::{Json, extract::State};
-use cidv10::Cid;
-use futures::{StreamExt, stream};
 use reqwest::header;
 use rsky_common::env::env_int;
-use rsky_identity::IdResolver;
-use rsky_pds::repo::prepare::{
-    PrepareCreateOpts, PrepareDeleteOpts, PrepareUpdateOpts, prepare_create, prepare_delete,
-    prepare_update,
-};
-use rsky_pds::sequencer::Sequencer;
 use rsky_repo::block_map::BlockMap;
 use rsky_repo::car::{CarWithRoot, read_stream_car_with_root};
 use rsky_repo::parse::get_and_parse_record;
 use rsky_repo::repo::Repo;
 use rsky_repo::sync::consumer::{VerifyRepoInput, verify_diff};
-use rsky_repo::types::{PreparedWrite, RecordWriteDescript, VerifiedDiff};
+use rsky_repo::types::{RecordWriteDescript, VerifiedDiff};
 use serde::Deserialize;
-use std::collections::HashMap;
-use std::hash::RandomState;
 use std::num::NonZeroU64;
-use std::sync::Arc;
-use tokio::sync::RwLock;
+
+use super::*;
 
 struct ImportRepoInput {
     car_with_root: CarWithRoot,
 }
 
-impl<'de> Deserialize<'de> for ImportRepoInput {
-    fn deserialize<D>(deserializer: D) -> core::result::Result<Self, D::Error>
-    where
-        D: serde::Deserializer<'de>,
-    {
-        async fn from_data(req: &Request<'_>, data: Data<'_>) -> Result<CarWithRoot, ApiError> {
-            let max_import_size = env_int("IMPORT_REPO_LIMIT").unwrap_or(100).megabytes();
-            match req.headers().get_one(header::CONTENT_LENGTH.as_ref()) {
-                None => {
-                    let error =
-                        ApiError::InvalidRequest("Missing content-length header".to_string());
-                    req.local_cache(|| Some(error.clone()));
-                    return Err(error);
-                }
-                Some(res) => match res.parse::<NonZeroU64>() {
-                    Ok(content_length) => {
-                        if content_length.get().bytes() > max_import_size {
-                            let error = ApiError::InvalidRequest(format!(
-                                "Content-Length is greater than maximum of {max_import_size}"
-                            ));
-                            req.local_cache(|| Some(error.clone()));
-                            return Err(error);
-                        }
-
-                        let import_datastream = data.open(content_length.get().bytes());
-                        read_stream_car_with_root(import_datastream).await
-                    }
-                    Err(_error) => {
-                        tracing::error!("{}", format!("Error parsing content-length\n{_error}"));
-                        let error =
-                            ApiError::InvalidRequest("Error parsing content-length".to_string());
-                        req.local_cache(|| Some(error.clone()));
-                        Err(error)
-                    }
-                },
-            }
-        }
-    }
-}
-
 // #[rocket::async_trait]
 // impl<'r> FromData<'r> for ImportRepoInput {
 //     type Error = ApiError;
···
 //     }
 // }
 
+// TODO: lookup axum docs to impl deserialize
+
 #[tracing::instrument(skip_all)]
 #[axum::debug_handler(state = AppState)]
 pub async fn import_repo(
     // auth: AccessFullImport,
-    user: AuthenticatedUser,
+    auth: AuthenticatedUser,
     Query(import_repo_input): Query<ImportRepoInput>,
     State(actor_pools): State<HashMap<String, ActorStorage, RandomState>>,
 ) -> Result<(), ApiError> {
     // let requester = auth.access.credentials.unwrap().did.unwrap();
-    let requester = user.did();
+    let requester = auth.did();
     let mut actor_store = ActorStore::from_actor_pools(&requester, &actor_pools).await;
 
     // Get current repo if it exists
···
         .collect::<Vec<_>>()
         .await
         .into_iter()
-        .collect::<anyhow::Result<Vec<PreparedWrite>, _>>()
+        .collect::<Result<Vec<PreparedWrite>, _>>()
     {
         Ok(res) => Ok(res),
         Err(error) => {

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

···
+//!
+use rsky_lexicon::com::atproto::repo::ListMissingBlobsOutput;
+use rsky_pds::actor_store::blob::ListMissingBlobsOpts;
+
+use super::*;
+
+///
+#[tracing::instrument(skip_all)]
+#[axum::debug_handler(state = AppState)]
+pub async fn list_missing_blobs(
+    user: AuthenticatedUser,
+    Query(input): Query<atrium_repo::list_missing_blobs::ParametersData>,
+    State(actor_pools): State<HashMap<String, ActorStorage, RandomState>>,
+) -> Result<Json<ListMissingBlobsOutput>, ApiError> {
+    let cursor = input.cursor;
+    let limit = input.limit;
+    let limit: Option<u16> = Some(limit.unwrap().into());
+    // let did = auth.access.credentials.unwrap().did.unwrap();
+    let did = user.did();
+    let limit: u16 = limit.unwrap_or(500);
+
+    let actor_store = ActorStore::from_actor_pools(&did, &actor_pools).await;
+
+    match actor_store
+        .blob
+        .list_missing_blobs(ListMissingBlobsOpts { cursor, limit })
+        .await
+    {
+        Ok(blobs) => {
+            let cursor = match blobs.last() {
+                Some(last_blob) => Some(last_blob.cid.clone()),
+                None => None,
+            };
+            Ok(Json(ListMissingBlobsOutput { cursor, blobs }))
+        }
+        Err(error) => {
+            tracing::error!("{error:?}");
+            Err(ApiError::RuntimeError)
+        }
+    }
+}

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

···
+//! List a range of records in a repository, matching a specific collection. Does not require auth.
+use super::*;
+
+// #[derive(serde::Serialize, serde::Deserialize, Debug, Clone, PartialEq, Eq)]
+// #[serde(rename_all = "camelCase")]
+// /// Parameters for [`list_records`].
+// pub(super) struct ListRecordsParameters {
+//     ///The NSID of the record type.
+//     pub collection: Nsid,
+//     /// The cursor to start from.
+//     #[serde(skip_serializing_if = "core::option::Option::is_none")]
+//     pub cursor: Option<String>,
+//     ///The number of records to return.
+//     #[serde(skip_serializing_if = "core::option::Option::is_none")]
+//     pub limit: Option<String>,
+//     ///The handle or DID of the repo.
+//     pub repo: AtIdentifier,
+//     ///Flag to reverse the order of the returned records.
+//     #[serde(skip_serializing_if = "core::option::Option::is_none")]
+//     pub reverse: Option<bool>,
+//     ///DEPRECATED: The highest sort-ordered rkey to stop at (exclusive)
+//     #[serde(skip_serializing_if = "core::option::Option::is_none")]
+//     pub rkey_end: Option<String>,
+//     ///DEPRECATED: The lowest sort-ordered rkey to start from (exclusive)
+//     #[serde(skip_serializing_if = "core::option::Option::is_none")]
+//     pub rkey_start: Option<String>,
+// }
+
+#[expect(non_snake_case, clippy::too_many_arguments)]
+async fn inner_list_records(
+    // The handle or DID of the repo.
+    repo: String,
+    // The NSID of the record type.
+    collection: String,
+    // The number of records to return.
+    limit: u16,
+    cursor: Option<String>,
+    // DEPRECATED: The lowest sort-ordered rkey to start from (exclusive)
+    rkeyStart: Option<String>,
+    // DEPRECATED: The highest sort-ordered rkey to stop at (exclusive)
+    rkeyEnd: Option<String>,
+    // Flag to reverse the order of the returned records.
+    reverse: bool,
+    // The actor pools
+    actor_pools: HashMap<String, ActorStorage>,
+    account_manager: Arc<RwLock<AccountManager>>,
+) -> Result<ListRecordsOutput> {
+    if limit > 100 {
+        bail!("Error: limit can not be greater than 100")
+    }
+    let did = account_manager
+        .read()
+        .await
+        .get_did_for_actor(&repo, None)
+        .await?;
+    if let Some(did) = did {
+        let mut actor_store = ActorStore::from_actor_pools(&did, &actor_pools).await;
+
+        let records: Vec<Record> = actor_store
+            .record
+            .list_records_for_collection(
+                collection,
+                limit as i64,
+                reverse,
+                cursor,
+                rkeyStart,
+                rkeyEnd,
+                None,
+            )
+            .await?
+            .into_iter()
+            .map(|record| {
+                Ok(Record {
+                    uri: record.uri.clone(),
+                    cid: record.cid.clone(),
+                    value: serde_json::to_value(record)?,
+                })
+            })
+            .collect::<Result<Vec<Record>>>()?;
+
+        let last_record = records.last();
+        let cursor: Option<String>;
+        if let Some(last_record) = last_record {
+            let last_at_uri: AtUri = last_record.uri.clone().try_into()?;
+            cursor = Some(last_at_uri.get_rkey());
+        } else {
+            cursor = None;
+        }
+        Ok(ListRecordsOutput { records, cursor })
+    } else {
+        bail!("Could not find repo: {repo}")
+    }
+}
+
+/// List a range of records in a repository, matching a specific collection. Does not require auth.
+/// - GET /xrpc/com.atproto.repo.listRecords
+/// ### Query Parameters
+/// - `repo`: `at-identifier` // The handle or DID of the repo.
+/// - `collection`: `nsid` // The NSID of the record type.
+/// - `limit`: `integer` // The maximum number of records to return. Default 50, >=1 and <=100.
+/// - `cursor`: `string`
+/// - `reverse`: `boolean` // Flag to reverse the order of the returned records.
+/// ### Responses
+/// - 200 OK: {"cursor": "string","records": [{"uri": "string","cid": "string","value": {}}]}
+/// - 400 Bad Request: {error:[`InvalidRequest`, `ExpiredToken`, `InvalidToken`]}
+/// - 401 Unauthorized
+#[tracing::instrument(skip_all)]
+#[allow(non_snake_case)]
+#[axum::debug_handler(state = AppState)]
+pub async fn list_records(
+    Query(input): Query<atrium_repo::list_records::ParametersData>,
+    State(actor_pools): State<HashMap<String, ActorStorage, RandomState>>,
+    State(account_manager): State<Arc<RwLock<AccountManager>>>,
+) -> Result<Json<ListRecordsOutput>, ApiError> {
+    let repo = input.repo;
+    let collection = input.collection;
+    let limit: Option<u8> = input.limit.map(u8::from);
+    let limit: Option<u16> = limit.map(|x| x.into());
+    let cursor = input.cursor;
+    let reverse = input.reverse;
+    let rkeyStart = None;
+    let rkeyEnd = None;
+
+    let limit = limit.unwrap_or(50);
+    let reverse = reverse.unwrap_or(false);
+
+    match inner_list_records(
+        repo.into(),
+        collection.into(),
+        limit,
+        cursor,
+        rkeyStart,
+        rkeyEnd,
+        reverse,
+        actor_pools,
+        account_manager,
+    )
+    .await
+    {
+        Ok(res) => Ok(Json(res)),
+        Err(error) => {
+            tracing::error!("@LOG: ERROR: {error}");
+            Err(ApiError::RuntimeError)
+        }
+    }
+}

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

···
-use atrium_api::com::atproto::repo;
+use atrium_api::com::atproto::repo as atrium_repo;
 use axum::{
     Router,
     routing::{get, post},
 };
 use constcat::concat;
 
-use crate::serve::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;
+pub mod import_repo;
+pub mod list_missing_blobs;
+pub mod list_records;
+pub mod put_record;
+pub mod upload_blob;
+
+use crate::account_manager::AccountManager;
+use crate::account_manager::helpers::account::AvailabilityFlags;
+use crate::{
+    actor_store::ActorStore,
+    auth::AuthenticatedUser,
+    error::ApiError,
+    serve::{ActorStorage, AppState},
+};
+use anyhow::{Result, bail};
+use axum::extract::Query;
+use axum::{Json, extract::State};
+use cidv10::Cid;
+use futures::stream::{self, StreamExt};
+use rsky_identity::IdResolver;
+use rsky_identity::types::DidDocument;
+use rsky_lexicon::com::atproto::repo::DeleteRecordInput;
+use rsky_lexicon::com::atproto::repo::DescribeRepoOutput;
+use rsky_lexicon::com::atproto::repo::GetRecordOutput;
+use rsky_lexicon::com::atproto::repo::{ApplyWritesInput, ApplyWritesInputRefWrite};
+use rsky_lexicon::com::atproto::repo::{CreateRecordInput, CreateRecordOutput};
+use rsky_lexicon::com::atproto::repo::{ListRecordsOutput, Record};
+// use rsky_pds::pipethrough::{OverrideOpts, ProxyRequest, pipethrough};
+use rsky_pds::repo::prepare::{
+    PrepareCreateOpts, PrepareDeleteOpts, PrepareUpdateOpts, prepare_create, prepare_delete,
+    prepare_update,
+};
+use rsky_pds::sequencer::Sequencer;
+use rsky_repo::types::PreparedDelete;
+use rsky_repo::types::PreparedWrite;
+use rsky_syntax::aturi::AtUri;
+use rsky_syntax::handle::INVALID_HANDLE;
+use std::collections::HashMap;
+use std::hash::RandomState;
+use std::str::FromStr;
+use std::sync::Arc;
+use tokio::sync::RwLock;
 
 /// 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.
···
 pub(crate) fn routes() -> Router<AppState> {
     Router::new()
         .route(
-            concat!("/", repo::apply_writes::NSID),
+            concat!("/", atrium_repo::apply_writes::NSID),
             post(apply_writes::apply_writes),
         )
         .route(
-            concat!("/", repo::create_record::NSID),
+            concat!("/", atrium_repo::create_record::NSID),
             post(create_record::create_record),
         )
-        // .route(concat!("/", repo::put_record::NSID), post(put_record))
+        .route(
+            concat!("/", atrium_repo::put_record::NSID),
+            post(put_record::put_record),
+        )
         .route(
-            concat!("/", repo::delete_record::NSID),
+            concat!("/", atrium_repo::delete_record::NSID),
             post(delete_record::delete_record),
         )
-        // .route(concat!("/", repo::upload_blob::NSID), post(upload_blob))
+        .route(
+            concat!("/", atrium_repo::upload_blob::NSID),
+            post(upload_blob::upload_blob),
+        )
         .route(
-            concat!("/", repo::describe_repo::NSID),
+            concat!("/", atrium_repo::describe_repo::NSID),
             get(describe_repo::describe_repo),
         )
         .route(
-            concat!("/", repo::get_record::NSID),
+            concat!("/", atrium_repo::get_record::NSID),
             get(get_record::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))
+        .route(
+            concat!("/", atrium_repo::import_repo::NSID),
+            post(import_repo::import_repo),
+        )
+        .route(
+            concat!("/", atrium_repo::list_missing_blobs::NSID),
+            get(list_missing_blobs::list_missing_blobs),
+        )
+        .route(
+            concat!("/", atrium_repo::list_records::NSID),
+            get(list_records::list_records),
+        )
 }

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

···
+//! Write a repository record, creating or updating it as needed. Requires auth, implemented by PDS.
+use anyhow::bail;
+use rsky_lexicon::com::atproto::repo::{PutRecordInput, PutRecordOutput};
+use rsky_repo::types::CommitDataWithOps;
+
+use super::*;
+
+#[tracing::instrument(skip_all)]
+async fn inner_put_record(
+    body: PutRecordInput,
+    auth: AuthenticatedUser,
+    sequencer: Arc<RwLock<Sequencer>>,
+    actor_pools: HashMap<String, ActorStorage>,
+    account_manager: Arc<RwLock<AccountManager>>,
+) -> Result<PutRecordOutput> {
+    let PutRecordInput {
+        repo,
+        collection,
+        rkey,
+        validate,
+        record,
+        swap_record,
+        swap_commit,
+    } = body;
+    let account = account_manager
+        .read()
+        .await
+        .get_account(
+            &repo,
+            Some(AvailabilityFlags {
+                include_deactivated: Some(true),
+                include_taken_down: None,
+            }),
+        )
+        .await?;
+    if let Some(account) = account {
+        if account.deactivated_at.is_some() {
+            bail!("Account is deactivated")
+        }
+        let did = account.did;
+        // if did != auth.access.credentials.unwrap().did.unwrap() {
+        if did != auth.did() {
+            bail!("AuthRequiredError")
+        }
+        let uri = AtUri::make(did.clone(), Some(collection.clone()), Some(rkey.clone()))?;
+        let swap_commit_cid = match swap_commit {
+            Some(swap_commit) => Some(Cid::from_str(&swap_commit)?),
+            None => None,
+        };
+        let swap_record_cid = match swap_record {
+            Some(swap_record) => Some(Cid::from_str(&swap_record)?),
+            None => None,
+        };
+        let (commit, write): (Option<CommitDataWithOps>, PreparedWrite) = {
+            let mut actor_store = ActorStore::from_actor_pools(&did, &actor_pools).await;
+
+            let current = actor_store
+                .record
+                .get_record(&uri, None, Some(true))
+                .await?;
+            tracing::debug!("@LOG: debug inner_put_record, current: {current:?}");
+            let write: PreparedWrite = if current.is_some() {
+                PreparedWrite::Update(
+                    prepare_update(PrepareUpdateOpts {
+                        did: did.clone(),
+                        collection,
+                        rkey,
+                        swap_cid: swap_record_cid,
+                        record: serde_json::from_value(record)?,
+                        validate,
+                    })
+                    .await?,
+                )
+            } else {
+                PreparedWrite::Create(
+                    prepare_create(PrepareCreateOpts {
+                        did: did.clone(),
+                        collection,
+                        rkey: Some(rkey),
+                        swap_cid: swap_record_cid,
+                        record: serde_json::from_value(record)?,
+                        validate,
+                    })
+                    .await?,
+                )
+            };
+
+            match current {
+                Some(current) if current.cid == write.cid().unwrap().to_string() => (None, write),
+                _ => {
+                    let commit = actor_store
+                        .process_writes(vec![write.clone()], swap_commit_cid)
+                        .await?;
+                    (Some(commit), write)
+                }
+            }
+        };
+
+        if let Some(commit) = commit {
+            sequencer
+                .write()
+                .await
+                .sequence_commit(did.clone(), commit.clone())
+                .await?;
+            account_manager
+                .write()
+                .await
+                .update_repo_root(
+                    did,
+                    commit.commit_data.cid,
+                    commit.commit_data.rev,
+                    &actor_pools,
+                )
+                .await?;
+        }
+        Ok(PutRecordOutput {
+            uri: write.uri().to_string(),
+            cid: write.cid().unwrap().to_string(),
+        })
+    } else {
+        bail!("Could not find repo: `{repo}`")
+    }
+}
+
+/// Write a repository record, creating or updating it as needed. Requires auth, implemented by PDS.
+/// - POST /xrpc/com.atproto.repo.putRecord
+/// ### Request Body
+/// - `repo`: `at-identifier` // The handle or DID of the repo (aka, current account).
+/// - `collection`: `nsid` // The NSID of the record collection.
+/// - `rkey`: `string` // The record key. <= 512 characters.
+/// - `validate`: `boolean` // Can be set to 'false' to skip Lexicon schema validation of record data, 'true' to require it, or leave unset to validate only for known Lexicons.
+/// - `record`
+/// - `swap_record`: `boolean` // Compare and swap with the previous record by CID. WARNING: nullable and optional field; may cause problems with golang implementation
+/// - `swap_commit`: `cid` // Compare and swap with the previous commit by CID.
+/// ### Responses
+/// - 200 OK: {"uri": "string","cid": "string","commit": {"cid": "string","rev": "string"},"validationStatus": "valid | unknown"}
+/// - 400 Bad Request: {error:"`InvalidRequest` | `ExpiredToken` | `InvalidToken` | `InvalidSwap`"}
+/// - 401 Unauthorized
+#[tracing::instrument(skip_all)]
+pub async fn put_record(
+    auth: AuthenticatedUser,
+    State(sequencer): State<Arc<RwLock<Sequencer>>>,
+    State(actor_pools): State<HashMap<String, ActorStorage, RandomState>>,
+    State(account_manager): State<Arc<RwLock<AccountManager>>>,
+    Json(body): Json<PutRecordInput>,
+) -> Result<Json<PutRecordOutput>, ApiError> {
+    tracing::debug!("@LOG: debug put_record {body:#?}");
+    match inner_put_record(body, auth, sequencer, actor_pools, account_manager).await {
+        Ok(res) => Ok(Json(res)),
+        Err(error) => {
+            tracing::error!("@LOG: ERROR: {error}");
+            Err(ApiError::RuntimeError)
+        }
+    }
+}

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

···
-//! PDS repository endpoints /xrpc/com.atproto.repo.*)
-mod apply_writes;
-pub(crate) use apply_writes::apply_writes;
-
-use std::{collections::HashSet, str::FromStr};
-
-use anyhow::{Context as _, anyhow};
-use atrium_api::com::atproto::repo::apply_writes::{
-    self as atrium_apply_writes, 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::{Cid, blockstore::CarStore};
-use axum::{
-    Json, Router,
-    body::Body,
-    extract::{Query, Request, State},
-    http::{self, StatusCode},
-    routing::{get, post},
-};
-use constcat::concat;
-use futures::TryStreamExt as _;
-use metrics::counter;
-use rsky_syntax::aturi::AtUri;
-use serde::Deserialize;
-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,
-};
-
-#[derive(serde::Serialize, serde::Deserialize, Debug, Clone, PartialEq, Eq)]
-#[serde(rename_all = "camelCase")]
-/// Parameters for [`list_records`].
-pub(super) struct ListRecordsParameters {
-    ///The NSID of the record type.
-    pub collection: Nsid,
-    /// The cursor to start from.
-    #[serde(skip_serializing_if = "core::option::Option::is_none")]
-    pub cursor: Option<String>,
-    ///The number of records to return.
-    #[serde(skip_serializing_if = "core::option::Option::is_none")]
-    pub limit: Option<String>,
-    ///The handle or DID of the repo.
-    pub repo: AtIdentifier,
-    ///Flag to reverse the order of the returned records.
-    #[serde(skip_serializing_if = "core::option::Option::is_none")]
-    pub reverse: Option<bool>,
-    ///DEPRECATED: The highest sort-ordered rkey to stop at (exclusive)
-    #[serde(skip_serializing_if = "core::option::Option::is_none")]
-    pub rkey_end: Option<String>,
-    ///DEPRECATED: The lowest sort-ordered rkey to start from (exclusive)
-    #[serde(skip_serializing_if = "core::option::Option::is_none")]
-    pub rkey_start: Option<String>,
-}
-
-/// Resolve DID to DID document. Does not bi-directionally verify handle.
-/// - GET /xrpc/com.atproto.repo.resolveDid
-/// ### Query Parameters
-/// - `did`: DID to resolve.
-/// ### Responses
-/// - 200 OK: {`did_doc`: `did_doc`}
-/// - 400 Bad Request: {error:[`InvalidRequest`, `ExpiredToken`, `InvalidToken`, `DidNotFound`, `DidDeactivated`]}
-async fn resolve_did(
-    db: &Db,
-    identifier: &AtIdentifier,
-) -> anyhow::Result<(
-    atrium_api::types::string::Did,
-    atrium_api::types::string::Handle,
-)> {
-    let (handle, did) = match *identifier {
-        AtIdentifier::Handle(ref handle) => {
-            let handle_as_str = &handle.as_str();
-            (
-                &handle.to_owned(),
-                &atrium_api::types::string::Did::new(
-                    sqlx::query_scalar!(
-                        r#"SELECT did FROM handles WHERE handle = ?"#,
-                        handle_as_str
-                    )
-                    .fetch_one(db)
-                    .await
-                    .context("failed to query did")?,
-                )
-                .expect("should be valid DID"),
-            )
-        }
-        AtIdentifier::Did(ref did) => {
-            let did_as_str = &did.as_str();
-            (
-                &atrium_api::types::string::Handle::new(
-                    sqlx::query_scalar!(r#"SELECT handle FROM handles WHERE did = ?"#, did_as_str)
-                        .fetch_one(db)
-                        .await
-                        .context("failed to query did")?,
-                )
-                .expect("should be valid handle"),
-                &did.to_owned(),
-            )
-        }
-    };
-
-    Ok((did.to_owned(), handle.to_owned()))
-}
-
-/// Create a single new repository record. Requires auth, implemented by PDS.
-/// - POST /xrpc/com.atproto.repo.createRecord
-/// ### Request Body
-/// - `repo`: `at-identifier` // The handle or DID of the repo (aka, current account).
-/// - `collection`: `nsid` // The NSID of the record collection.
-/// - `rkey`: `string` // The record key. <= 512 characters.
-/// - `validate`: `boolean` // Can be set to 'false' to skip Lexicon schema validation of record data, 'true' to require it, or leave unset to validate only for known Lexicons.
-/// - `record`
-/// - `swap_commit`: `cid` // Compare and swap with the previous commit by CID.
-/// ### Responses
-/// - 200 OK: {`cid`: `cid`, `uri`: `at-uri`, `commit`: {`cid`: `cid`, `rev`: `tid`}, `validation_status`: [`valid`, `unknown`]}
-/// - 400 Bad Request: {error:[`InvalidRequest`, `ExpiredToken`, `InvalidToken`, `InvalidSwap`]}
-/// - 401 Unauthorized
-async fn create_record(
-    user: AuthenticatedUser,
-    State(actor_store): State<ActorStore>,
-    State(skey): State<SigningKey>,
-    State(config): State<AppConfig>,
-    State(db): State<Db>,
-    State(fhp): State<FirehoseProducer>,
-    Json(input): Json<repo::create_record::Input>,
-) -> Result<Json<repo::create_record::Output>> {
-    todo!();
-    // let write_result = apply_writes::apply_writes(
-    //     user,
-    //     State(actor_store),
-    //     State(skey),
-    //     State(config),
-    //     State(db),
-    //     State(fhp),
-    //     Json(
-    //         repo::apply_writes::InputData {
-    //             repo: input.repo.clone(),
-    //             validate: input.validate,
-    //             swap_commit: input.swap_commit.clone(),
-    //             writes: vec![repo::apply_writes::InputWritesItem::Create(Box::new(
-    //                 repo::apply_writes::CreateData {
-    //                     collection: input.collection.clone(),
-    //                     rkey: input.rkey.clone(),
-    //                     value: input.record.clone(),
-    //                 }
-    //                 .into(),
-    //             ))],
-    //         }
-    //         .into(),
-    //     ),
-    // )
-    // .await
-    // .context("failed to apply writes")?;
-
-    // let create_result = if let repo::apply_writes::OutputResultsItem::CreateResult(create_result) =
-    //     write_result
-    //         .results
-    //         .clone()
-    //         .and_then(|result| result.first().cloned())
-    //         .context("unexpected output from apply_writes")?
-    // {
-    //     Some(create_result)
-    // } else {
-    //     None
-    // }
-    // .context("unexpected result from apply_writes")?;
-
-    // Ok(Json(
-    //     repo::create_record::OutputData {
-    //         cid: create_result.cid.clone(),
-    //         commit: write_result.commit.clone(),
-    //         uri: create_result.uri.clone(),
-    //         validation_status: Some("unknown".to_owned()),
-    //     }
-    //     .into(),
-    // ))
-}
-
-/// Write a repository record, creating or updating it as needed. Requires auth, implemented by PDS.
-/// - POST /xrpc/com.atproto.repo.putRecord
-/// ### Request Body
-/// - `repo`: `at-identifier` // The handle or DID of the repo (aka, current account).
-/// - `collection`: `nsid` // The NSID of the record collection.
-/// - `rkey`: `string` // The record key. <= 512 characters.
-/// - `validate`: `boolean` // Can be set to 'false' to skip Lexicon schema validation of record data, 'true' to require it, or leave unset to validate only for known Lexicons.
-/// - `record`
-/// - `swap_record`: `boolean` // Compare and swap with the previous record by CID. WARNING: nullable and optional field; may cause problems with golang implementation
-/// - `swap_commit`: `cid` // Compare and swap with the previous commit by CID.
-/// ### Responses
-/// - 200 OK: {"uri": "string","cid": "string","commit": {"cid": "string","rev": "string"},"validationStatus": "valid | unknown"}
-/// - 400 Bad Request: {error:"`InvalidRequest` | `ExpiredToken` | `InvalidToken` | `InvalidSwap`"}
-/// - 401 Unauthorized
-async fn put_record(
-    user: AuthenticatedUser,
-    State(actor_store): State<ActorStore>,
-    State(skey): State<SigningKey>,
-    State(config): State<AppConfig>,
-    State(db): State<Db>,
-    State(fhp): State<FirehoseProducer>,
-    Json(input): Json<repo::put_record::Input>,
-) -> Result<Json<repo::put_record::Output>> {
-    todo!();
-    // // TODO: `input.swap_record`
-    // // FIXME: "put" implies that we will create the record if it does not exist.
-    // // We currently only update existing records and/or throw an error if one doesn't exist.
-    // let input = (*input).clone();
-    // let input = repo::apply_writes::InputData {
-    //     repo: input.repo,
-    //     validate: input.validate,
-    //     swap_commit: input.swap_commit,
-    //     writes: vec![repo::apply_writes::InputWritesItem::Update(Box::new(
-    //         repo::apply_writes::UpdateData {
-    //             collection: input.collection,
-    //             rkey: input.rkey,
-    //             value: input.record,
-    //         }
-    //         .into(),
-    //     ))],
-    // }
-    // .into();
-
-    // let write_result = apply_writes::apply_writes(
-    //     user,
-    //     State(actor_store),
-    //     State(skey),
-    //     State(config),
-    //     State(db),
-    //     State(fhp),
-    //     Json(input),
-    // )
-    // .await
-    // .context("failed to apply writes")?;
-
-    // let update_result = write_result
-    //     .results
-    //     .clone()
-    //     .and_then(|result| result.first().cloned())
-    //     .context("unexpected output from apply_writes")?;
-    // let (cid, uri) = match update_result {
-    //     repo::apply_writes::OutputResultsItem::CreateResult(create_result) => (
-    //         Some(create_result.cid.clone()),
-    //         Some(create_result.uri.clone()),
-    //     ),
-    //     repo::apply_writes::OutputResultsItem::UpdateResult(update_result) => (
-    //         Some(update_result.cid.clone()),
-    //         Some(update_result.uri.clone()),
-    //     ),
-    //     repo::apply_writes::OutputResultsItem::DeleteResult(_) => (None, None),
-    // };
-    // Ok(Json(
-    //     repo::put_record::OutputData {
-    //         cid: cid.context("missing cid")?,
-    //         commit: write_result.commit.clone(),
-    //         uri: uri.context("missing uri")?,
-    //         validation_status: Some("unknown".to_owned()),
-    //     }
-    //     .into(),
-    // ))
-}
-
-/// Delete a repository record, or ensure it doesn't exist. Requires auth, implemented by PDS.
-/// - POST /xrpc/com.atproto.repo.deleteRecord
-/// ### Request Body
-/// - `repo`: `at-identifier` // The handle or DID of the repo (aka, current account).
-/// - `collection`: `nsid` // The NSID of the record collection.
-/// - `rkey`: `string` // The record key. <= 512 characters.
-/// - `swap_record`: `boolean` // Compare and swap with the previous record by CID.
-/// - `swap_commit`: `cid` // Compare and swap with the previous commit by CID.
-/// ### Responses
-/// - 200 OK: {"commit": {"cid": "string","rev": "string"}}
-/// - 400 Bad Request: {error:[`InvalidRequest`, `ExpiredToken`, `InvalidToken`, `InvalidSwap`]}
-/// - 401 Unauthorized
-async fn delete_record(
-    user: AuthenticatedUser,
-    State(actor_store): State<ActorStore>,
-    State(skey): State<SigningKey>,
-    State(config): State<AppConfig>,
-    State(db): State<Db>,
-    State(fhp): State<FirehoseProducer>,
-    Json(input): Json<repo::delete_record::Input>,
-) -> Result<Json<repo::delete_record::Output>> {
-    todo!();
-    // // TODO: `input.swap_record`
-
-    // Ok(Json(
-    //     repo::delete_record::OutputData {
-    //         commit: apply_writes::apply_writes(
-    //             user,
-    //             State(actor_store),
-    //             State(skey),
-    //             State(config),
-    //             State(db),
-    //             State(fhp),
-    //             Json(
-    //                 repo::apply_writes::InputData {
-    //                     repo: input.repo.clone(),
-    //                     swap_commit: input.swap_commit.clone(),
-    //                     validate: None,
-    //                     writes: vec![repo::apply_writes::InputWritesItem::Delete(Box::new(
-    //                         repo::apply_writes::DeleteData {
-    //                             collection: input.collection.clone(),
-    //                             rkey: input.rkey.clone(),
-    //                         }
-    //                         .into(),
-    //                     ))],
-    //                 }
-    //                 .into(),
-    //             ),
-    //         )
-    //         .await
-    //         .context("failed to apply writes")?
-    //         .commit
-    //         .clone(),
-    //     }
-    //     .into(),
-    // ))
-}
-
-/// Get information about an account and repository, including the list of collections. Does not require auth.
-/// - GET /xrpc/com.atproto.repo.describeRepo
-/// ### Query Parameters
-/// - `repo`: `at-identifier` // The handle or DID of the repo.
-/// ### Responses
-/// - 200 OK: {"handle": "string","did": "string","didDoc": {},"collections": [string],"handleIsCorrect": true} \
-///   handeIsCorrect - boolean - Indicates if handle is currently valid (resolves bi-directionally)
-/// - 400 Bad Request: {error:[`InvalidRequest`, `ExpiredToken`, `InvalidToken`]}
-/// - 401 Unauthorized
-async fn describe_repo(
-    State(actor_store): State<ActorStore>,
-    State(config): State<AppConfig>,
-    State(db): State<Db>,
-    Query(input): Query<repo::describe_repo::ParametersData>,
-) -> Result<Json<repo::describe_repo::Output>> {
-    // Lookup the DID by the provided handle.
-    let (did, handle) = resolve_did(&db, &input.repo)
-        .await
-        .context("failed to resolve handle")?;
-
-    // Use Actor Store to get the collections
-    todo!();
-}
-
-/// Get a single record from a repository. Does not require auth.
-/// - GET /xrpc/com.atproto.repo.getRecord
-/// ### Query Parameters
-/// - `repo`: `at-identifier` // The handle or DID of the repo.
-/// - `collection`: `nsid` // The NSID of the record collection.
-/// - `rkey`: `string` // The record key. <= 512 characters.
-/// - `cid`: `cid` // The CID of the version of the record. If not specified, then return the most recent version.
-/// ### Responses
-/// - 200 OK: {"uri": "string","cid": "string","value": {}}
-/// - 400 Bad Request: {error:[`InvalidRequest`, `ExpiredToken`, `InvalidToken`, `RecordNotFound`]}
-/// - 401 Unauthorized
-async fn get_record(
-    State(actor_store): State<ActorStore>,
-    State(config): State<AppConfig>,
-    State(db): State<Db>,
-    Query(input): Query<repo::get_record::ParametersData>,
-) -> Result<Json<repo::get_record::Output>> {
-    if input.cid.is_some() {
-        return Err(Error::unimplemented(anyhow!(
-            "looking up old records is unsupported"
-        )));
-    }
-
-    // Lookup the DID by the provided handle.
-    let (did, _handle) = resolve_did(&db, &input.repo)
-        .await
-        .context("failed to resolve handle")?;
-
-    // Create a URI from the parameters
-    let uri = format!(
-        "at://{}/{}/{}",
-        did.as_str(),
-        input.collection.as_str(),
-        input.rkey.as_str()
-    );
-
-    // Use Actor Store to get the record
-    todo!();
-}
-
-/// List a range of records in a repository, matching a specific collection. Does not require auth.
-/// - GET /xrpc/com.atproto.repo.listRecords
-/// ### Query Parameters
-/// - `repo`: `at-identifier` // The handle or DID of the repo.
-/// - `collection`: `nsid` // The NSID of the record type.
-/// - `limit`: `integer` // The maximum number of records to return. Default 50, >=1 and <=100.
-/// - `cursor`: `string`
-/// - `reverse`: `boolean` // Flag to reverse the order of the returned records.
-/// ### Responses
-/// - 200 OK: {"cursor": "string","records": [{"uri": "string","cid": "string","value": {}}]}
-/// - 400 Bad Request: {error:[`InvalidRequest`, `ExpiredToken`, `InvalidToken`]}
-/// - 401 Unauthorized
-async fn list_records(
-    State(actor_store): State<ActorStore>,
-    State(config): State<AppConfig>,
-    State(db): State<Db>,
-    Query(input): Query<Object<ListRecordsParameters>>,
-) -> Result<Json<repo::list_records::Output>> {
-    // Lookup the DID by the provided handle.
-    let (did, _handle) = resolve_did(&db, &input.repo)
-        .await
-        .context("failed to resolve handle")?;
-
-    // Use Actor Store to list records for the collection
-    todo!();
-}
-
-/// Upload a new blob, to be referenced from a repository record. \
-/// The blob will be deleted if it is not referenced within a time window (eg, minutes). \
-/// Blob restrictions (mimetype, size, etc) are enforced when the reference is created. \
-/// Requires auth, implemented by PDS.
-/// - POST /xrpc/com.atproto.repo.uploadBlob
-/// ### Request Body
-/// ### Responses
-/// - 200 OK: {"blob": "binary"}
-/// - 400 Bad Request: {error:[`InvalidRequest`, `ExpiredToken`, `InvalidToken`]}
-/// - 401 Unauthorized
-async fn upload_blob(
-    user: AuthenticatedUser,
-    State(actor_store): State<ActorStore>,
-    State(config): State<AppConfig>,
-    State(db): State<Db>,
-    request: Request<Body>,
-) -> Result<Json<repo::upload_blob::Output>> {
-    let length = request
-        .headers()
-        .get(http::header::CONTENT_LENGTH)
-        .context("no content length provided")?
-        .to_str()
-        .map_err(anyhow::Error::from)
-        .and_then(|content_length| content_length.parse::<u64>().map_err(anyhow::Error::from))
-        .context("invalid content-length header")?;
-    let mime = request
-        .headers()
-        .get(http::header::CONTENT_TYPE)
-        .context("no content-type provided")?
-        .to_str()
-        .context("invalid content-type provided")?
-        .to_owned();
-
-    if length > config.blob.limit {
-        return Err(Error::with_status(
-            StatusCode::PAYLOAD_TOO_LARGE,
-            anyhow!("size {} above limit {}", length, config.blob.limit),
-        ));
-    }
-
-    // Read the blob data
-    let mut body_data = Vec::new();
-    let mut stream = request.into_body().into_data_stream();
-    while let Some(bytes) = stream.try_next().await.context("failed to receive file")? {
-        body_data.extend_from_slice(&bytes);
-
-        // Check size limit incrementally
-        if body_data.len() as u64 > config.blob.limit {
-            return Err(Error::with_status(
-                StatusCode::PAYLOAD_TOO_LARGE,
-                anyhow!("size above limit and content-length header was wrong"),
-            ));
-        }
-    }
-
-    // Use Actor Store to upload the blob
-    todo!();
-}
-
-async fn todo() -> Result<()> {
-    Err(Error::unimplemented(anyhow!("not implemented")))
-}
-
-/// 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(super) fn routes() -> Router<AppState> {
-    Router::new()
-        .route(concat!("/", repo::apply_writes::NSID), post(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/atproto/repo/upload_blob.rs

···
+//! Upload a new blob, to be referenced from a repository record.
+use crate::config::AppConfig;
+use anyhow::Context as _;
+use axum::{
+    body::Bytes,
+    http::{self, HeaderMap},
+};
+use rsky_lexicon::com::atproto::repo::{Blob, BlobOutput};
+use rsky_repo::types::{BlobConstraint, PreparedBlobRef};
+// use rsky_common::BadContentTypeError;
+
+use super::*;
+
+async fn inner_upload_blob(
+    auth: AuthenticatedUser,
+    blob: Bytes,
+    content_type: String,
+    actor_pools: HashMap<String, ActorStorage>,
+) -> Result<BlobOutput> {
+    // let requester = auth.access.credentials.unwrap().did.unwrap();
+    let requester = auth.did();
+
+    let actor_store = ActorStore::from_actor_pools(&requester, &actor_pools).await;
+
+    let metadata = actor_store
+        .blob
+        .upload_blob_and_get_metadata(content_type, blob)
+        .await?;
+    let blobref = actor_store.blob.track_untethered_blob(metadata).await?;
+
+    // make the blob permanent if an associated record is already indexed
+    let records_for_blob = actor_store
+        .blob
+        .get_records_for_blob(blobref.get_cid()?)
+        .await?;
+
+    if !records_for_blob.is_empty() {
+        actor_store
+            .blob
+            .verify_blob_and_make_permanent(PreparedBlobRef {
+                cid: blobref.get_cid()?,
+                mime_type: blobref.get_mime_type().to_string(),
+                constraints: BlobConstraint {
+                    max_size: None,
+                    accept: None,
+                },
+            })
+            .await?;
+    }
+
+    Ok(BlobOutput {
+        blob: Blob {
+            r#type: Some("blob".to_string()),
+            r#ref: Some(blobref.get_cid()?),
+            cid: None,
+            mime_type: blobref.get_mime_type().to_string(),
+            size: blobref.get_size(),
+            original: None,
+        },
+    })
+}
+
+/// Upload a new blob, to be referenced from a repository record. \
+/// The blob will be deleted if it is not referenced within a time window (eg, minutes). \
+/// Blob restrictions (mimetype, size, etc) are enforced when the reference is created. \
+/// Requires auth, implemented by PDS.
+/// - POST /xrpc/com.atproto.repo.uploadBlob
+/// ### Request Body
+/// ### Responses
+/// - 200 OK: {"blob": "binary"}
+/// - 400 Bad Request: {error:[`InvalidRequest`, `ExpiredToken`, `InvalidToken`]}
+/// - 401 Unauthorized
+#[tracing::instrument(skip_all)]
+#[axum::debug_handler(state = AppState)]
+pub async fn upload_blob(
+    auth: AuthenticatedUser,
+    headers: HeaderMap,
+    State(config): State<AppConfig>,
+    State(actor_pools): State<HashMap<String, ActorStorage, RandomState>>,
+    blob: Bytes,
+) -> Result<Json<BlobOutput>, ApiError> {
+    let content_length = headers
+        .get(http::header::CONTENT_LENGTH)
+        .context("no content length provided")?
+        .to_str()
+        .map_err(anyhow::Error::from)
+        .and_then(|content_length| content_length.parse::<u64>().map_err(anyhow::Error::from))
+        .context("invalid content-length header")?;
+    let content_type = headers
+        .get(http::header::CONTENT_TYPE)
+        .context("no content-type provided")?
+        .to_str()
+        // .map_err(BadContentTypeError::MissingType)
+        .context("invalid content-type provided")?
+        .to_owned();
+
+    if content_length > config.blob.limit {
+        return Err(ApiError::InvalidRequest(format!(
+            "Content-Length is greater than maximum of {}",
+            config.blob.limit
+        )));
+    };
+    if blob.len() as u64 > config.blob.limit {
+        return Err(ApiError::InvalidRequest(format!(
+            "Blob size is greater than maximum of {} despite content-length header",
+            config.blob.limit
+        )));
+    };
+
+    match inner_upload_blob(auth, blob, content_type, actor_pools).await {
+        Ok(res) => Ok(Json(res)),
+        Err(error) => {
+            tracing::error!("{error:?}");
+            Err(ApiError::RuntimeError)
+        }
+    }
+}