Merge branch 'main' into createAccountEnhancement

Author: rudyfraser

Cargo.toml

@@ -6,7 +6,7 @@ resolver = "2"
 cargo = { version = "0.84.0",features = ["vendored-openssl"] }
 rsky-lexicon = {path = "rsky-lexicon", version = "0.2.3"}
 rsky-identity = {path = "rsky-identity", version = "0.1.0"}
-rsky-crypto = {path = "rsky-crypto", version = "0.1.0"}
+rsky-crypto = {path = "rsky-crypto", version = "0.1.1"}
 rsky-syntax = {path = "rsky-syntax", version = "0.1.0"}
 rsky-common = {path = "rsky-common", version = "0.1.0"}
 

rsky-crypto/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "rsky-crypto"
-version = "0.1.0"
+version = "0.1.1"
 authors = ["Rudy Fraser <him@rudyfraser.com>"]
 description = "Rust library providing basic cryptographic helpers as needed in atproto"
 license = "Apache-2.0"
@@ -12,6 +12,6 @@ documentation = "https://docs.rs/rsky-crypto"
 
 [dependencies]
 multibase = "0.9.1"
-secp256k1 = { version = "0.28.2", features = ["global-context", "serde", "rand", "hashes"] }
+secp256k1 = { version = "0.28.2", features = ["global-context", "serde", "rand", "hashes","rand-std"] }
 anyhow = "1.0.79"
 p256 = { version = "0.13.2", features = ["ecdsa","arithmetic","alloc"] }

rsky-crypto/src/utils.rs

@@ -1,6 +1,8 @@
 use crate::constants::{BASE58_MULTIBASE_PREFIX, DID_KEY_PREFIX};
 use anyhow::{bail, Result};
-use multibase::{encode, Base};
+use multibase::decode;
+use secp256k1::rand::rngs::OsRng;
+use secp256k1::rand::RngCore;
 
 pub fn extract_multikey(did: &String) -> Result<String> {
     if !did.starts_with(DID_KEY_PREFIX) {
@@ -13,9 +15,16 @@ pub fn extract_prefixed_bytes(multikey: String) -> Result<Vec<u8>> {
     if !multikey.starts_with(BASE58_MULTIBASE_PREFIX) {
         bail!("Incorrect prefix for multikey: {multikey}")
     }
-    Ok(encode(Base::Base58Btc, &multikey[BASE58_MULTIBASE_PREFIX.len()..]).into_bytes())
+    let (_base, bytes) = decode(&multikey)?;
+    Ok(bytes)
 }
 
 pub fn has_prefix(bytes: &Vec<u8>, prefix: &Vec<u8>) -> bool {
     *prefix == bytes[0..prefix.len()]
 }
+
+pub fn random_bytes(len: usize) -> Vec<u8> {
+    let mut buf = vec![0u8; len];
+    OsRng.fill_bytes(&mut buf);
+    buf
+}

rsky-pds/Cargo.toml

@@ -77,6 +77,7 @@ async-recursion = "1.1.1"
 once_cell = "1.19.0"
 tracing = "0.1.41"
 tracing-subscriber = "0.3.19"
+async-stream = "0.3.5"
 
 
 [dependencies.rocket_sync_db_pools]

rsky-pds/src/apis/app/bsky/actor/get_preferences.rs

@@ -1,5 +1,6 @@
 use crate::apis::ApiError;
 use crate::auth_verifier::AccessStandard;
+use crate::db::DbConn;
 use crate::repo::aws::s3::S3BlobStore;
 use crate::repo::ActorStore;
 use anyhow::Result;
@@ -11,12 +12,14 @@ use rsky_lexicon::app::bsky::actor::{GetPreferencesOutput, RefPreferences};
 async fn inner_get_preferences(
     s3_config: &State<SdkConfig>,
     auth: AccessStandard,
+    db: DbConn,
 ) -> Result<GetPreferencesOutput> {
     let auth = auth.access.credentials.unwrap();
     let requester = auth.did.unwrap().clone();
     let actor_store = ActorStore::new(
         requester.clone(),
         S3BlobStore::new(requester.clone(), s3_config),
+        db,
     );
     let preferences: Vec<RefPreferences> = actor_store
         .pref
@@ -28,15 +31,17 @@ async fn inner_get_preferences(
 
 /// Get private preferences attached to the current account. Expected use is synchronization
 /// between multiple devices, and import/export during account migration. Requires auth.
+#[tracing::instrument(skip_all)]
 #[rocket::get("/xrpc/app.bsky.actor.getPreferences")]
 pub async fn get_preferences(
     s3_config: &State<SdkConfig>,
     auth: AccessStandard,
+    db: DbConn,
 ) -> Result<Json<GetPreferencesOutput>, ApiError> {
-    match inner_get_preferences(s3_config, auth).await {
+    match inner_get_preferences(s3_config, auth, db).await {
         Ok(res) => Ok(Json(res)),
         Err(error) => {
-            eprintln!("@LOG: ERROR: {error}");
+            tracing::error!("@LOG: ERROR: {error}");
             Err(ApiError::RuntimeError)
         }
     }

rsky-pds/src/apis/app/bsky/actor/get_profile.rs

@@ -1,6 +1,7 @@
 use crate::apis::ApiError;
 use crate::auth_verifier::AccessStandard;
 use crate::config::ServerConfig;
+use crate::db::DbConn;
 use crate::read_after_write::types::LocalRecords;
 use crate::read_after_write::util::{handle_read_after_write, ReadAfterWriteResponse};
 use crate::read_after_write::viewer::LocalViewer;
@@ -20,6 +21,7 @@ pub async fn inner_get_profile(
     res: HandlerPipeThrough,
     s3_config: &State<SdkConfig>,
     state_local_viewer: &State<SharedLocalViewer>,
+    db: DbConn,
 ) -> Result<ReadAfterWriteResponse<ProfileViewDetailed>, ApiError> {
     let requester: Option<String> = match auth.access.credentials {
         None => None,
@@ -35,6 +37,7 @@ pub async fn inner_get_profile(
                 get_profile_munge,
                 s3_config,
                 state_local_viewer,
+                db,
             )
             .await?;
             Ok(read_afer_write_response)
@@ -44,6 +47,7 @@ pub async fn inner_get_profile(
 
 /// Get detailed profile view of an actor. Does not require auth,
 /// but contains relevant metadata with auth.
+#[tracing::instrument(skip_all)]
 #[rocket::get("/xrpc/app.bsky.actor.getProfile?<actor>")]
 pub async fn get_profile(
     // Handle or DID of account to fetch profile of.
@@ -53,13 +57,16 @@ pub async fn get_profile(
     s3_config: &State<SdkConfig>,
     state_local_viewer: &State<SharedLocalViewer>,
     cfg: &State<ServerConfig>,
+    db: DbConn,
 ) -> Result<ReadAfterWriteResponse<ProfileViewDetailed>, ApiError> {
     match cfg.bsky_app_view {
         None => Err(ApiError::AccountNotFound),
-        Some(_) => match inner_get_profile(actor, auth, res, s3_config, state_local_viewer).await {
-            Ok(response) => Ok(response),
-            Err(error) => Err(error),
-        },
+        Some(_) => {
+            match inner_get_profile(actor, auth, res, s3_config, state_local_viewer, db).await {
+                Ok(response) => Ok(response),
+                Err(error) => Err(error),
+            }
+        }
     }
 }
 

rsky-pds/src/apis/app/bsky/actor/get_profiles.rs

@@ -1,6 +1,7 @@
 use crate::apis::ApiError;
 use crate::auth_verifier::AccessStandard;
 use crate::config::ServerConfig;
+use crate::db::DbConn;
 use crate::read_after_write::types::LocalRecords;
 use crate::read_after_write::util::{handle_read_after_write, ReadAfterWriteResponse};
 use crate::read_after_write::viewer::LocalViewer;
@@ -19,6 +20,7 @@ pub async fn inner_get_profiles(
     res: HandlerPipeThrough,
     s3_config: &State<SdkConfig>,
     state_local_viewer: &State<SharedLocalViewer>,
+    db: DbConn,
 ) -> Result<ReadAfterWriteResponse<GetProfilesOutput>, ApiError> {
     let requester: String = match auth.access.credentials {
         None => "".to_string(),
@@ -31,12 +33,14 @@ pub async fn inner_get_profiles(
         get_profiles_munge,
         s3_config,
         state_local_viewer,
+        db,
     )
     .await?;
     Ok(read_afer_write_response)
 }
 
 /// Get detailed profile views of multiple actors.
+#[tracing::instrument(skip_all)]
 #[rocket::get("/xrpc/app.bsky.actor.getProfiles?<actors>")]
 pub async fn get_profiles(
     actors: Vec<String>,
@@ -45,14 +49,16 @@ pub async fn get_profiles(
     s3_config: &State<SdkConfig>,
     state_local_viewer: &State<SharedLocalViewer>,
     cfg: &State<ServerConfig>,
+    db: DbConn,
 ) -> Result<ReadAfterWriteResponse<GetProfilesOutput>, ApiError> {
     match cfg.bsky_app_view {
         None => Err(ApiError::AccountNotFound),
-        Some(_) => match inner_get_profiles(actors, auth, res, s3_config, state_local_viewer).await
-        {
-            Ok(response) => Ok(response),
-            Err(error) => Err(error),
-        },
+        Some(_) => {
+            match inner_get_profiles(actors, auth, res, s3_config, state_local_viewer, db).await {
+                Ok(response) => Ok(response),
+                Err(error) => Err(error),
+            }
+        }
     }
 }
 

rsky-pds/src/apis/app/bsky/actor/put_preferences.rs

@@ -1,5 +1,6 @@
 use crate::apis::ApiError;
 use crate::auth_verifier::AccessStandard;
+use crate::db::DbConn;
 use crate::repo::aws::s3::S3BlobStore;
 use crate::repo::ActorStore;
 use anyhow::Result;
@@ -12,13 +13,15 @@ async fn inner_put_preferences(
     body: Json<PutPreferencesInput>,
     s3_config: &State<SdkConfig>,
     auth: AccessStandard,
+    db: DbConn,
 ) -> Result<(), ApiError> {
     let PutPreferencesInput { preferences } = body.into_inner();
     let auth = auth.access.credentials.unwrap();
     let requester = auth.did.unwrap().clone();
     let actor_store = ActorStore::new(
         requester.clone(),
         S3BlobStore::new(requester.clone(), s3_config),
+        db,
     );
     actor_store
         .pref
@@ -27,6 +30,7 @@ async fn inner_put_preferences(
     Ok(())
 }
 
+#[tracing::instrument(skip_all)]
 #[rocket::post(
     "/xrpc/app.bsky.actor.putPreferences",
     format = "json",
@@ -36,8 +40,9 @@ pub async fn put_preferences(
     body: Json<PutPreferencesInput>,
     s3_config: &State<SdkConfig>,
     auth: AccessStandard,
+    db: DbConn,
 ) -> Result<(), ApiError> {
-    match inner_put_preferences(body, s3_config, auth).await {
+    match inner_put_preferences(body, s3_config, auth, db).await {
         Ok(_) => Ok(()),
         Err(error) => Err(error),
     }

rsky-pds/src/apis/app/bsky/feed/get_actor_likes.rs

@@ -1,6 +1,7 @@
 use crate::apis::ApiError;
 use crate::auth_verifier::AccessStandard;
 use crate::config::ServerConfig;
+use crate::db::DbConn;
 use crate::read_after_write::types::LocalRecords;
 use crate::read_after_write::util::{handle_read_after_write, ReadAfterWriteResponse};
 use crate::read_after_write::viewer::LocalViewer;
@@ -21,6 +22,7 @@ pub async fn inner_get_actor_likes(
     res: HandlerPipeThrough,
     s3_config: &State<SdkConfig>,
     state_local_viewer: &State<SharedLocalViewer>,
+    db: DbConn,
 ) -> Result<ReadAfterWriteResponse<AuthorFeed>> {
     let requester: Option<String> = match auth.access.credentials {
         None => None,
@@ -36,6 +38,7 @@ pub async fn inner_get_actor_likes(
                 get_author_munge,
                 s3_config,
                 state_local_viewer,
+                db,
             )
             .await?;
             Ok(read_afer_write_response)
@@ -44,6 +47,7 @@ pub async fn inner_get_actor_likes(
 }
 
 /// Get a list of posts liked by an actor. Does not require auth.
+#[tracing::instrument(skip_all)]
 #[rocket::get("/xrpc/app.bsky.feed.getActorLikes?<actor>&<limit>&<cursor>")]
 pub async fn get_actor_likes(
     actor: String,
@@ -54,6 +58,7 @@ pub async fn get_actor_likes(
     s3_config: &State<SdkConfig>,
     state_local_viewer: &State<SharedLocalViewer>,
     cfg: &State<ServerConfig>,
+    db: DbConn,
 ) -> Result<ReadAfterWriteResponse<AuthorFeed>, ApiError> {
     if let Some(limit) = limit {
         if limit > 100 {
@@ -70,12 +75,13 @@ pub async fn get_actor_likes(
             res,
             s3_config,
             state_local_viewer,
+            db,
         )
         .await
         {
             Ok(response) => Ok(response),
             Err(error) => {
-                eprintln!("{error}");
+                tracing::error!("{error}");
                 Err(ApiError::RuntimeError)
             }
         },

rsky-pds/src/apis/app/bsky/feed/get_author_feed.rs

@@ -1,6 +1,7 @@
 use crate::apis::ApiError;
 use crate::auth_verifier::AccessStandard;
 use crate::config::ServerConfig;
+use crate::db::DbConn;
 use crate::read_after_write::types::LocalRecords;
 use crate::read_after_write::util::{handle_read_after_write, ReadAfterWriteResponse};
 use crate::read_after_write::viewer::LocalViewer;
@@ -23,6 +24,7 @@ pub async fn inner_get_author_feed(
     res: HandlerPipeThrough,
     s3_config: &State<SdkConfig>,
     state_local_viewer: &State<SharedLocalViewer>,
+    db: DbConn,
 ) -> Result<ReadAfterWriteResponse<AuthorFeed>> {
     let requester: Option<String> = match auth.access.credentials {
         None => None,
@@ -38,6 +40,7 @@ pub async fn inner_get_author_feed(
                 get_author_munge,
                 s3_config,
                 state_local_viewer,
+                db,
             )
             .await?;
             Ok(read_afer_write_response)
@@ -46,6 +49,7 @@ pub async fn inner_get_author_feed(
 }
 
 /// Get a view of an actor's 'author feed' (post and reposts by the author). Does not require auth.
+#[tracing::instrument(skip_all)]
 #[rocket::get("/xrpc/app.bsky.feed.getAuthorFeed?<actor>&<limit>&<cursor>&<filter>")]
 pub async fn get_author_feed(
     actor: String,
@@ -57,6 +61,7 @@ pub async fn get_author_feed(
     s3_config: &State<SdkConfig>,
     state_local_viewer: &State<SharedLocalViewer>,
     cfg: &State<ServerConfig>,
+    db: DbConn,
 ) -> Result<ReadAfterWriteResponse<AuthorFeed>, ApiError> {
     if let Some(limit) = limit {
         if limit > 100 {
@@ -97,6 +102,7 @@ pub async fn get_author_feed(
             res,
             s3_config,
             state_local_viewer,
+            db,
         )
         .await
         {