Fix double-encoding mess

Cargo.toml

@@ -40,7 +40,7 @@ base32 = "0.5.1"
 blake3 = "1.5.5"
 url = "2.5.4"
 async-trait = "0.1.84"
-reqwest = { version = "0.12.9", features = ["json"] }
+reqwest = "0.12.9"
 
 [dev-dependencies]
 anyhow = "1.0.95"

examples/new_homeserver_user.rs

@@ -39,17 +39,14 @@ async fn main() -> Result<()> {
         Some("Running Bitcoin".to_string()),
     );
 
-    // Serialize the profile to JSON
-    let profile_json = serde_json::to_vec(&user)?;
-
     // Put some content into the Pubky homeserver
     let url = format!(
         "{protocol}{pk}{path}",
         protocol = PROTOCOL,
         pk = pk,
         path = user.create_path()
     );
-    client.put(url.as_str()).json(&profile_json).send().await?;
+    client.put(url.as_str()).json(&user).send().await?;
 
     Ok(())
 }

src/db/connectors/pubky.rs

@@ -56,7 +56,6 @@ impl PubkyConnector {
             },
         };
 
-        println!("CLIEEENNNNNNNNNNNNT, {:?}", pubky_client);
         let manager = Self {
             pubky_client: Arc::new(pubky_client),
         };

src/events/handlers/file.rs

@@ -11,7 +11,6 @@ use crate::{
     },
     Config,
 };
-use axum::body::Bytes;
 use log::{debug, error};
 use pubky_app_specs::{traits::Validatable, PubkyAppFile};
 use tokio::{
@@ -71,6 +70,8 @@ async fn ingest(
         }
     };
 
+    let blob = response.bytes().await?;
+
     store_blob(file_id.to_string(), user_id.to_string(), &blob).await?;
 
     Ok(FileMeta {

src/events/mod.rs

@@ -1,14 +1,12 @@
 use crate::{db::connectors::pubky::PubkyConnector, types::PubkyId};
 use log::{debug, error};
+use reqwest;
 use uri::ParsedUri;
 
 pub mod handlers;
 pub mod processor;
 pub mod uri;
 
-use axum::body::Bytes;
-use reqwest;
-
 #[derive(Debug, Clone)]
 enum ResourceType {
     User {

src/events/processor.rs

@@ -59,11 +59,10 @@ impl EventProcessor {
     async fn poll_events(&mut self) -> Result<Option<Vec<String>>, Box<dyn std::error::Error>> {
         debug!("Polling new events from homeserver");
 
-        let res: String;
+        let response: String;
         {
             let pubky_client = PubkyConnector::get_pubky_client()?;
-            debug!("BEFORE PUBKY CLIENT");
-            res = pubky_client
+            response = pubky_client
                 .get(format!(
                     "https://{}/events/?cursor={}&limit={}",
                     self.homeserver.id, self.homeserver.cursor, self.limit
@@ -72,10 +71,9 @@ impl EventProcessor {
                 .await?
                 .text()
                 .await?;
-            debug!("AFTER PUBKY CLIENT");
         }
 
-        let lines: Vec<String> = res.trim().split('\n').map(|s| s.to_string()).collect();
+        let lines: Vec<String> = response.trim().split('\n').map(|s| s.to_string()).collect();
         debug!("Homeserver response lines {:?}", lines);
 
         if lines.len() == 1 && lines[0].is_empty() {

tests/watcher/bookmarks/del.rs

@@ -46,15 +46,14 @@ async fn test_homeserver_unbookmark() -> Result<()> {
         uri: format!("pubky://{}/pub/pubky.app/posts/{}", author_id, post_id),
         created_at: chrono::Utc::now().timestamp_millis(),
     };
-    let bookmark_blob = serde_json::to_vec(&bookmark)?;
     let bookmark_id = bookmark.create_id();
     let bookmark_url = format!(
         "pubky://{}/pub/pubky.app/bookmarks/{}",
         bookmarker_id, bookmark_id
     );
 
     // Put bookmark
-    test.put(&bookmark_url, bookmark_blob).await.unwrap();
+    test.put(&bookmark_url, bookmark).await.unwrap();
 
     // Step 4: Delete bookmark
     test.del(&bookmark_url).await?;

tests/watcher/bookmarks/raw.rs

@@ -36,15 +36,14 @@ async fn test_homeserver_bookmark() -> Result<()> {
         uri: format!("pubky://{}/pub/pubky.app/posts/{}", user_id, post_id),
         created_at: chrono::Utc::now().timestamp_millis(),
     };
-    let bookmark_blob = serde_json::to_vec(&bookmark)?;
     let bookmark_id = bookmark.create_id();
     let bookmark_url = format!(
         "pubky://{}/pub/pubky.app/bookmarks/{}",
         user_id, bookmark_id
     );
 
     // Put bookmark
-    test.put(&bookmark_url, bookmark_blob).await.unwrap();
+    test.put(&bookmark_url, bookmark).await.unwrap();
 
     // Step 4: Verify the bookmark exists in Nexus
     // GRAPH_OP: Assert if the event writes the graph

tests/watcher/bookmarks/viewer.rs

@@ -46,15 +46,14 @@ async fn test_homeserver_viewer_bookmark() -> Result<()> {
         uri: format!("pubky://{}/pub/pubky.app/posts/{}", user_id, post_id),
         created_at: chrono::Utc::now().timestamp_millis(),
     };
-    let bookmark_blob = serde_json::to_vec(&bookmark)?;
     let bookmark_id = bookmark.create_id();
     let bookmark_url = format!(
         "pubky://{}/pub/pubky.app/bookmarks/{}",
         viewer_id, bookmark_id
     );
 
     // Put bookmark
-    test.put(&bookmark_url, bookmark_blob).await.unwrap();
+    test.put(&bookmark_url, bookmark).await.unwrap();
 
     // Step 4: Verify the bookmark exists in Nexus
     // GRAPH_OP: Assert if the event writes the graph

tests/watcher/files/create.rs

@@ -33,7 +33,7 @@ async fn test_put_pubkyapp_file() -> Result<()> {
     let pubky_client = PubkyConnector::get_pubky_client()?;
     pubky_client
         .put(blob_url.as_str())
-        .json(&json_data)
+        .json(&blob)
         .send()
         .await?;
 

tests/watcher/files/delete.rs

@@ -8,7 +8,6 @@ use pubky_nexus::{
     models::{file::FileDetails, traits::Collection},
     PubkyConnector,
 };
-use serde_json::to_vec;
 
 #[tokio_shared_rt::test(shared)]
 async fn test_delete_pubkyapp_file() -> Result<()> {
@@ -29,11 +28,10 @@ async fn test_delete_pubkyapp_file() -> Result<()> {
     let blob = "Hello World!";
     let blob_id = Timestamp::now().to_string();
     let blob_url = format!("pubky://{}/pub/pubky.app/blobs/{}", user_id, blob_id);
-    let json_data = to_vec(blob)?;
     let pubky_client = PubkyConnector::get_pubky_client()?;
     pubky_client
         .put(blob_url.as_str())
-        .json(&json_data)
+        .json(&blob)
         .send()
         .await?;
 

tests/watcher/network/counts.rs

@@ -138,7 +138,7 @@ async fn test_large_network_scenario_counts() -> Result<()> {
                     let pubky_client = PubkyConnector::get_pubky_client()?;
                     pubky_client
                         .put(mute_url.as_str())
-                        .json(&serde_json::to_vec(&mute)?)
+                        .json(&mute)
                         .send()
                         .await?;
                     _total_mutes += 1;
@@ -171,8 +171,7 @@ async fn test_large_network_scenario_counts() -> Result<()> {
                     bookmark.create_id()
                 );
 
-                test.put(&bookmark_url, serde_json::to_vec(&bookmark)?)
-                    .await?;
+                test.put(&bookmark_url, &bookmark).await?;
                 total_bookmarks += 1;
             }
         }
@@ -200,7 +199,7 @@ async fn test_large_network_scenario_counts() -> Result<()> {
 
                 let tag_url = format!("pubky://{}/pub/pubky.app/tags/{}", user_id, tag.create_id());
 
-                test.put(&tag_url, serde_json::to_vec(&tag)?).await?;
+                test.put(&tag_url, &tag).await?;
                 total_tags += 1;
 
                 // FAILS: possibly deletes a tag twice and decrements twice in index.

tests/watcher/posts/attachments.rs

@@ -30,7 +30,7 @@ async fn test_homeserver_post_attachments() -> Result<()> {
     let pubky_client = PubkyConnector::get_pubky_client()?;
     pubky_client
         .put(blob_url.as_str())
-        .json(&json_data)
+        .json(&blob)
         .send()
         .await?;
 

tests/watcher/posts/del_bookmarked_notification.rs

@@ -50,13 +50,12 @@ async fn test_delete_bookmarked_post_notification() -> Result<()> {
         uri: format!("pubky://{}/pub/pubky.app/posts/{}", user_a_id, post_id),
         created_at: 0,
     };
-    let bookmark_blob = serde_json::to_vec(&bookmark)?;
     let bookmark_url = format!(
         "pubky://{}/pub/pubky.app/bookmarks/{}",
         user_b_id,
         bookmark.create_id()
     );
-    test.put(&bookmark_url, bookmark_blob).await?;
+    test.put(&bookmark_url, bookmark).await?;
 
     // User A deletes their post
     test.cleanup_post(&user_a_id, &post_id).await?;

tests/watcher/posts/del_tagged_notification.rs

@@ -51,12 +51,11 @@ async fn test_delged_post_notification() -> Result<()> {
         label: label.to_string(),
         created_at: Utc::now().timestamp_millis(),
     };
-    let tag_blob = serde_json::to_vec(&tag)?;
     let tag_id = tag.create_id();
     let tag_url = format!("pubky://{}/pub/pubky.app/tags/{}", user_b_id, tag_id);
 
     // Put tag
-    test.put(&tag_url, tag_blob).await?;
+    test.put(&tag_url, tag).await?;
 
     // User A deletes their post
     test.cleanup_post(&user_a_id, &post_id).await?;

tests/watcher/posts/del_with_relations.rs

@@ -36,11 +36,10 @@ async fn test_delete_post_with_relationships() -> Result<()> {
         label: "funny".to_string(),
         created_at: Utc::now().timestamp_millis(),
     };
-    let tag_blob = serde_json::to_vec(&tag)?;
     let tag_url = format!("pubky://{}/pub/pubky.app/tags/{}", user_id, tag.create_id());
 
     // Put tag
-    test.put(&tag_url, tag_blob).await?;
+    test.put(&tag_url, tag).await?;
 
     // Delete the post using the event handler
     test.cleanup_post(&user_id, &post_id).await?;

tests/watcher/posts/edit_bookmarked_notification.rs

@@ -51,24 +51,22 @@ async fn test_edit_bookmarked_post_notification() -> Result<()> {
         uri: format!("pubky://{}/pub/pubky.app/posts/{}", user_a_id, post_id),
         created_at: 0,
     };
-    let bookmark_blob = serde_json::to_vec(&bookmark)?;
     let bookmark_url = format!(
         "pubky://{}/pub/pubky.app/bookmarks/{}",
         user_b_id,
         bookmark.create_id()
     );
-    test.put(&bookmark_url, bookmark_blob).await?;
+    test.put(&bookmark_url, bookmark).await?;
 
     // User A edits their post
     post.content = "Edited post by User A".to_string();
-    let edited_post_blob = serde_json::to_vec(&post)?;
     let edited_url = format!("pubky://{}/pub/pubky.app/posts/{}", user_a_id, post_id);
 
     // Overwrite existing post in the homeserver for the edited one
     let pubky_client = PubkyConnector::get_pubky_client()?;
     pubky_client
         .put(edited_url.as_str())
-        .json(&edited_post_blob)
+        .json(&post)
         .send()
         .await?;
     test.ensure_event_processing_complete().await?;

tests/watcher/posts/edit_reply_parent_notification.rs

@@ -59,14 +59,13 @@ async fn test_edit_parent_post_notification() -> Result<()> {
 
     // User A edits their original post
     post.content = "Edited post by User A".to_string();
-    let edited_post_blob = serde_json::to_vec(&post)?;
     let edited_url = format!("pubky://{}/pub/pubky.app/posts/{}", user_a_id, post_id);
 
     // Overwrite existing post in the homeserver with the edited one
     let pubky_client = PubkyConnector::get_pubky_client()?;
     pubky_client
         .put(edited_url.as_str())
-        .json(&edited_post_blob)
+        .json(&post)
         .send()
         .await?;
     test.ensure_event_processing_complete().await?;

tests/watcher/posts/edit_reposted_notification.rs

@@ -59,14 +59,13 @@ async fn test_edit_reposted_post_notification() -> Result<()> {
 
     // User A edits their post
     post.content = "Edited post by User A".to_string();
-    let edited_post_blob = serde_json::to_vec(&post)?;
     let edited_url = format!("pubky://{}/pub/pubky.app/posts/{}", user_a_id, post_id);
 
     // Overwrite existing post in the homeserver for the edited one
     let pubky_client = PubkyConnector::get_pubky_client()?;
     pubky_client
         .put(edited_url.as_str())
-        .json(&edited_post_blob)
+        .json(&post)
         .send()
         .await?;
     test.ensure_event_processing_complete().await?;

tests/watcher/posts/edit_tagged_notification.rs

@@ -52,23 +52,21 @@ async fn test_edit_tagged_post_notification() -> Result<()> {
         label: label.to_string(),
         created_at: Utc::now().timestamp_millis(),
     };
-    let tag_blob = serde_json::to_vec(&tag)?;
     let tag_id = tag.create_id();
     let tag_url = format!("pubky://{}/pub/pubky.app/tags/{}", user_b_id, tag_id);
 
     // Put tag
-    test.put(&tag_url, tag_blob).await?;
+    test.put(&tag_url, tag).await?;
 
     // User A edits their post
     post.content = "Edited post by User A".to_string();
-    let edited_post_blob = serde_json::to_vec(&post)?;
     let edited_url = format!("pubky://{}/pub/pubky.app/posts/{}", user_a_id, post_id);
 
     // Overwrite existing post in the homeserver with the edited one
     let pubky_client = PubkyConnector::get_pubky_client()?;
     pubky_client
         .put(edited_url.as_str())
-        .json(&edited_post_blob)
+        .json(&post)
         .send()
         .await?;
     test.ensure_event_processing_complete().await?;

tests/watcher/posts/reply_engagement.rs

@@ -139,15 +139,14 @@ async fn test_homeserver_reply_engagement_control() -> Result<()> {
         label: label.to_string(),
         created_at: Utc::now().timestamp_millis(),
     };
-    let tag_blob = serde_json::to_vec(&tag)?;
 
     let tag_url = format!(
         "pubky://{}/pub/pubky.app/tags/{}",
         tagger_user_id,
         tag.create_id()
     );
 
-    test.put(&tag_url, tag_blob).await?;
+    test.put(&tag_url, tag).await?;
 
     // Check if reply post is not in total engagement index: Sorted:Posts:Global:TotalEngagement:user_id:post_id
     let total_engagement = check_member_total_engagement_user_posts(&[&author_id, &reply_id])