prototype apis com atproto repo

Cargo.lock

···
  "tower-http",
  "tracing",
  "tracing-subscriber",
+ "ubyte",
  "url",
  "urlencoding",
  "uuid 1.16.0",

Cargo.toml

···
     "sqlite",
     "tracing",
 ] }
+ubyte = "0.10.4"

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

···
+use axum::{body::Bytes, http::HeaderMap};
 use reqwest::header;
 use rsky_common::env::env_int;
 use rsky_repo::block_map::BlockMap;
···
 use rsky_repo::repo::Repo;
 use rsky_repo::sync::consumer::{VerifyRepoInput, verify_diff};
 use rsky_repo::types::{RecordWriteDescript, VerifiedDiff};
-use serde::Deserialize;
-use std::num::NonZeroU64;
+use ubyte::ToByteUnit;
 
 use super::*;
 
-struct ImportRepoInput {
-    car_with_root: CarWithRoot,
-}
-
-// #[rocket::async_trait]
-// impl<'r> FromData<'r> for ImportRepoInput {
-//     type Error = ApiError;
-
-//     #[tracing::instrument(skip_all)]
-//     async fn from_data(req: &'r Request<'_>, data: Data<'r>) -> Outcome<'r, Self, Self::Error> {
-//         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()));
-//                 Outcome::Error((Status::BadRequest, 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 Outcome::Error((Status::BadRequest, error));
-//                     }
-
-//                     let import_datastream = data.open(content_length.get().bytes());
-//                     match read_stream_car_with_root(import_datastream).await {
-//                         Ok(car_with_root) => Outcome::Success(ImportRepoInput { car_with_root }),
-//                         Err(error) => {
-//                             let error = ApiError::InvalidRequest(error.to_string());
-//                             req.local_cache(|| Some(error.clone()));
-//                             Outcome::Error((Status::BadRequest, error))
-//                         }
-//                     }
-//                 }
-//                 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()));
-//                     Outcome::Error((Status::BadRequest, error))
-//                 }
-//             },
-//         }
-//     }
-// }
+async fn from_data(bytes: Bytes) -> Result<CarWithRoot, ApiError> {
+    let max_import_size = env_int("IMPORT_REPO_LIMIT").unwrap_or(100).megabytes();
+    if bytes.len() > max_import_size {
+        return Err(ApiError::InvalidRequest(format!(
+            "Content-Length is greater than maximum of {max_import_size}"
+        )));
+    }
 
-// TODO: lookup axum docs to impl deserialize
+    let mut cursor = std::io::Cursor::new(bytes);
+    match read_stream_car_with_root(&mut cursor).await {
+        Ok(car_with_root) => Ok(car_with_root),
+        Err(error) => {
+            tracing::error!("Error reading stream car with root\n{error}");
+            Err(ApiError::InvalidRequest("Invalid CAR file".to_owned()))
+        }
+    }
+}
 
 #[tracing::instrument(skip_all)]
 #[axum::debug_handler(state = AppState)]
+/// Import a repo in the form of a CAR file. Requires Content-Length HTTP header to be set.
+/// Request
+/// mime     application/vnd.ipld.car
+/// Body - required
 pub async fn import_repo(
     // auth: AccessFullImport,
     auth: AuthenticatedUser,
-    Query(import_repo_input): Query<ImportRepoInput>,
+    headers: HeaderMap,
     State(actor_pools): State<HashMap<String, ActorStorage, RandomState>>,
+    body: Bytes,
 ) -> Result<(), ApiError> {
     // let requester = auth.access.credentials.unwrap().did.unwrap();
     let requester = auth.did();
     let mut actor_store = ActorStore::from_actor_pools(&requester, &actor_pools).await;
 
+    // Check headers
+    let content_length = headers
+        .get(header::CONTENT_LENGTH)
+        .expect("no content length provided")
+        .to_str()
+        .map_err(anyhow::Error::from)
+        .and_then(|content_length| content_length.parse::<u64>().map_err(anyhow::Error::from))
+        .expect("invalid content-length header");
+    if content_length > env_int("IMPORT_REPO_LIMIT").unwrap_or(100).megabytes() {
+        return Err(ApiError::InvalidRequest(format!(
+            "Content-Length is greater than maximum of {}",
+            env_int("IMPORT_REPO_LIMIT").unwrap_or(100).megabytes()
+        )));
+    };
+
     // Get current repo if it exists
     let curr_root: Option<Cid> = actor_store.get_repo_root().await;
     let curr_repo: Option<Repo> = match curr_root {
···
     };
 
     // Process imported car
-    let car_with_root = import_repo_input.car_with_root;
+    // let car_with_root = import_repo_input.car_with_root;
+    let car_with_root: CarWithRoot = match from_data(body).await {
+        Ok(car) => car,
+        Err(error) => {
+            tracing::error!("Error importing repo\n{error:?}");
+            return Err(ApiError::InvalidRequest("Invalid CAR file".to_owned()));
+        }
+    };
 
     // Get verified difference from current repo and imported repo
     let mut imported_blocks: BlockMap = car_with_root.blocks;
···
 
 /// Converts list of RecordWriteDescripts into a list of PreparedWrites
 async fn prepare_import_repo_writes(
-    _did: String,
+    did: String,
     writes: Vec<RecordWriteDescript>,
     blocks: &BlockMap,
 ) -> Result<Vec<PreparedWrite>, ApiError> {
     match stream::iter(writes)
         .then(|write| {
-            let did = _did.clone();
+            let did = did.clone();
             async move {
                 Ok::<PreparedWrite, anyhow::Error>(match write {
                     RecordWriteDescript::Create(write) => {

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

···
-//!
+//! Returns a list of missing blobs for the requesting account. Intended to be used in the account migration flow.
 use rsky_lexicon::com::atproto::repo::ListMissingBlobsOutput;
 use rsky_pds::actor_store::blob::ListMissingBlobsOpts;
 
 use super::*;
 
-///
+/// Returns a list of missing blobs for the requesting account. Intended to be used in the account migration flow.
+/// Request
+/// Query Parameters
+///     limit    integer
+///         Possible values: >= 1 and <= 1000
+///         Default value: 500
+///     cursor   string
+/// Responses
+/// cursor      string
+/// blobs       object[]
 #[tracing::instrument(skip_all)]
 #[axum::debug_handler(state = AppState)]
 pub async fn list_missing_blobs(
···
 ) -> Result<Json<ListMissingBlobsOutput>, ApiError> {
     let cursor = input.cursor;
     let limit = input.limit;
-    let limit: Option<u16> = Some(limit.unwrap().into());
+    let default_limit: atrium_api::types::LimitedNonZeroU16<1000> =
+        atrium_api::types::LimitedNonZeroU16::try_from(500).expect("default limit");
+    let limit: u16 = limit.unwrap_or(default_limit).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;
 
···
         .await
     {
         Ok(blobs) => {
-            let cursor = match blobs.last() {
-                Some(last_blob) => Some(last_blob.cid.clone()),
-                None => None,
-            };
+            let cursor = blobs.last().map(|last_blob| last_blob.cid.clone());
             Ok(Json(ListMissingBlobsOutput { cursor, blobs }))
         }
         Err(error) => {

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

···
             };
 
             match current {
-                Some(current) if current.cid == write.cid().unwrap().to_string() => (None, write),
+                Some(current) if current.cid == write.cid().expect("write cid").to_string() => {
+                    (None, write)
+                }
                 _ => {
                     let commit = actor_store
                         .process_writes(vec![write.clone()], swap_commit_cid)
···
         };
 
         if let Some(commit) = commit {
-            sequencer
+            _ = sequencer
                 .write()
                 .await
                 .sequence_commit(did.clone(), commit.clone())
···
         }
         Ok(PutRecordOutput {
             uri: write.uri().to_string(),
-            cid: write.cid().unwrap().to_string(),
+            cid: write.cid().expect("write cid").to_string(),
         })
     } else {
         bail!("Could not find repo: `{repo}`")

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

···
 
     Ok(BlobOutput {
         blob: Blob {
-            r#type: Some("blob".to_string()),
+            r#type: Some("blob".to_owned()),
             r#ref: Some(blobref.get_cid()?),
             cid: None,
             mime_type: blobref.get_mime_type().to_string(),