chore(deps): bump pubky v0.4

Cargo.toml

@@ -11,10 +11,10 @@ build = "build.rs"
 
 [dependencies]
 bytes = "1.9.0" # Enforce this version as Pubky and Axum conflict v1.7.1 vs v1.9.0
-pkarr = { git = "https://github.com/pubky/pkarr", branch = "v3-rc1", package = "pkarr" }
-mainline = { git = "https://github.com/pubky/mainline", branch = "v5-rc1", default-features = false }
-pubky = { git = "https://github.com/pubky/pubky", branch = "v0.4.0-rc1" }
-pubky-app-specs = { git = "https://github.com/pubky/pubky-app-specs", features = [
+pkarr = { git = "https://github.com/pubky/pkarr", rev = "4269e68" } # a few commits after v3.3.1
+pubky = { git = "https://github.com/pubky/pubky", branch = "main" }
+pubky-testnet = { git = "https://github.com/pubky/pubky", branch = "main" }
+pubky-app-specs = { git = "https://github.com/pubky/pubky-app-specs", branch = "main", features = [
     "openapi",
 ] }
 tokio = { version = "1.43.0", features = ["full"] }
@@ -43,7 +43,7 @@ reqwest = "0.12.9"
 anyhow = "1.0.95"
 httpc-test = "0.1.10"
 criterion = { version = "0.5.1", features = ["async_tokio"] }
-pubky-homeserver = { git = "https://github.com/pubky/pubky", branch = "v0.4.0-rc1" }
+pubky-homeserver = { git = "https://github.com/pubky/pubky", branch = "main" }
 rand = "0.8.5"
 rand_distr = "0.4.3"
 tokio-shared-rt = "0.1"

benches/watcher.rs

@@ -1,10 +1,8 @@
 use criterion::{criterion_group, criterion_main, Criterion};
-use mainline::Testnet;
 use pkarr::Keypair;
-use pubky::Client;
 use pubky_app_specs::{PubkyAppUser, PubkyAppUserLink};
-use pubky_homeserver::Homeserver;
 use pubky_nexus::EventProcessor;
+use pubky_testnet::Testnet;
 use setup::run_setup;
 use std::time::Duration;
 use tokio::runtime::Runtime;
@@ -18,9 +16,9 @@ mod setup;
 /// 4. Delete the profile.json
 async fn create_homeserver_with_events() -> (Testnet, String) {
     // Create the test environment
-    let testnet = Testnet::new(3).unwrap();
-    let homeserver = Homeserver::start_test(&testnet).await.unwrap();
-    let client = Client::builder().testnet(&testnet).build().unwrap();
+    let testnet = Testnet::run().await.unwrap();
+    let homeserver = testnet.run_homeserver().await.unwrap();
+    let client = testnet.client_builder().build().unwrap();
     let homeserver_url = homeserver.url().to_string();
 
     // Generate user data

examples/from_file.rs

@@ -15,7 +15,7 @@ async fn main() -> Result<(), DynError> {
     let config = Config::from_env();
     StackManager::setup(&config).await;
 
-    PubkyConnector::initialise(&config, None).await?;
+    PubkyConnector::initialise(&config).await?;
 
     let mut event_processor = EventProcessor::from_config(&config).await?;
 

examples/stress_network_sim.rs

@@ -39,10 +39,13 @@ async fn main() -> Result<()> {
     let config = Config::from_env();
 
     // Initialize the Client based on configuration
-    let client = match config.testnet {
-        true => Client::testnet()?,
-        false => Client::new()?,
-    };
+    // let client = match config.testnet {
+    //     true => Client::testnet()?,
+    //     false => Client::new()?,
+    // };
+
+    // Only mainnet
+    let client = Client::builder().build()?;
 
     // Convert the homeserver from the config into a PublicKey
     let homeserver = PublicKey::try_from(config.homeserver.as_str())?;

examples/test_testnet.rs

@@ -0,0 +1,38 @@
+use pkarr::Keypair;
+use pubky_testnet::Testnet;
+
+#[tokio::main]
+async fn main() {
+    // Run a new testnet.
+    let testnet = Testnet::run().await.unwrap();
+
+    // Optionally create and run a Homeserver.
+    let server = testnet.run_homeserver().await.unwrap();
+
+    // Create a Pubky Client from the testnet.
+    let client = testnet.client_builder().build().unwrap();
+
+    let keypair = Keypair::random();
+
+    client.signup(&keypair, &server.public_key()).await.unwrap();
+    client
+        .put(format!(
+            "pubky://{}/pub/pubky.app/object",
+            keypair.public_key(),
+        ))
+        .body([0, 1, 2, 3].to_vec())
+        .send()
+        .await
+        .unwrap();
+
+    let response = client
+        .get(format!(
+            "https://{}/events/?cursor=0000000000000&limit=1000",
+            server.public_key()
+        ))
+        .send()
+        .await
+        .unwrap();
+
+    println!("Event Lines: {}", response.text().await.unwrap())
+}

src/db/connectors/pubky.rs

@@ -1,10 +1,10 @@
-use crate::Config;
-use mainline::Testnet;
 use pubky::Client;
 use std::sync::Arc;
 use thiserror::Error;
 use tokio::sync::OnceCell;
 
+use crate::Config;
+
 static PUBKY_CONNECTOR_SINGLETON: OnceCell<Arc<Client>> = OnceCell::const_new();
 
 #[derive(Debug, Error)]
@@ -26,26 +26,20 @@ pub struct PubkyConnector;
 
 impl PubkyConnector {
     /// Initializes the PubkyConnector singleton with the given configuration
-    pub async fn initialise(
-        config: &Config,
-        testnet: Option<&Testnet>,
-    ) -> Result<(), PubkyConnectorError> {
+    pub async fn initialise(config: &Config) -> Result<(), PubkyConnectorError> {
         PUBKY_CONNECTOR_SINGLETON
             .get_or_try_init(|| async {
-                let pubky_client = match testnet {
-                    Some(testnet) => Client::builder()
-                        .testnet(testnet)
+                let client = match config.testnet {
+                    true => Client::builder()
+                        .testnet()
+                        .build()
+                        .map_err(|e| PubkyConnectorError::ClientError(e.to_string()))?,
+                    false => Client::builder()
                         .build()
                         .map_err(|e| PubkyConnectorError::ClientError(e.to_string()))?,
-                    None => match config.testnet {
-                        true => Client::testnet()
-                            .map_err(|e| PubkyConnectorError::ClientError(e.to_string()))?,
-                        false => Client::new()
-                            .map_err(|e| PubkyConnectorError::ClientError(e.to_string()))?,
-                    },
                 };
 
-                Ok(Arc::new(pubky_client))
+                Ok(Arc::new(client))
             })
             .await
             .map(|_| ())
@@ -58,4 +52,11 @@ impl PubkyConnector {
             .cloned()
             .ok_or(PubkyConnectorError::NotInitialized)
     }
+
+    pub async fn init_from_client(client: Client) -> Result<(), PubkyConnectorError> {
+        PUBKY_CONNECTOR_SINGLETON
+            .get_or_try_init(|| async { Ok(Arc::new(client)) })
+            .await
+            .map(|_| ())
+    }
 }

src/watcher.rs

@@ -11,7 +11,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Sync + Send>> {
 
     StackManager::setup(&config).await;
 
-    PubkyConnector::initialise(&config, None).await?;
+    PubkyConnector::initialise(&config).await?;
 
     let mut event_processor = EventProcessor::from_config(&config).await?;
 

tests/watcher/users/del_with_relations.rs

@@ -13,6 +13,7 @@ use pubky_nexus::{
     models::user::{UserCounts, UserView},
     PubkyConnector,
 };
+
 #[tokio_shared_rt::test(shared)]
 async fn test_delete_user_with_relationships() -> Result<()> {
     let mut test = WatcherTest::setup().await?;
@@ -83,7 +84,7 @@ async fn test_delete_user_with_relationships() -> Result<()> {
     test.cleanup_post(&user_id, &post_id).await?;
 
     // Write and delete the user again; this time it should be fully removed
-    test.create_user(&keypair, &user).await?;
+    test.create_profile(&user_id, &user).await?;
     test.cleanup_user(&user_id).await?;
 
     // Attempt to find user details; should not exist
@@ -151,7 +152,7 @@ async fn test_delete_user_with_relationships() -> Result<()> {
         name: "Watcher:UserDeleteWith:UserWith".to_string(),
         status: Some("Zombie soon".to_string()),
     };
-    let _ = test.create_user(&keypair, &user_with).await?;
+    let _ = test.create_profile(&user_with_id, &user_with).await?;
 
     // Create a post to establish a relationship
     let post_b = PubkyAppPost {
@@ -210,7 +211,7 @@ async fn test_delete_user_with_relationships() -> Result<()> {
     test.cleanup_post(&user_with_id, &post_b_id).await?;
 
     // Write and delete the user again; this time it should be fully removed
-    test.create_user(&keypair, &user_with).await?;
+    test.create_profile(&user_with_id, &user_with).await?;
     test.cleanup_user(&user_with_id).await?;
     // Delete the file
     test.cleanup_file(&user_with_id, &file_id).await?;

tests/watcher/utils/mod.rs

@@ -1,2 +1,2 @@
-mod dht;
+pub mod testnet;
 pub mod watcher;

tests/watcher/utils/testnet.rs

@@ -0,0 +1,29 @@
+use anyhow::Error;
+use pubky_testnet::Testnet;
+use std::sync::Arc;
+use tokio::sync::OnceCell;
+
+static TESTNET_NETWORK: OnceCell<TestnetNetwork> = OnceCell::const_new();
+
+/// Represents a test network for the Distributed Hash Table (DHT).
+pub struct TestnetNetwork {
+    nodes: Arc<Testnet>,
+}
+
+impl TestnetNetwork {
+    /// Returns the testnet. This method will initialize the testnet if it hasn't been
+    /// already, and will wait for the initialization to complete if another task is doing it.
+    pub async fn get() -> Result<Arc<Testnet>, Error> {
+        let network = TESTNET_NETWORK
+            .get_or_init(|| async {
+                let testnet = Testnet::run()
+                    .await
+                    .expect("Failed to run testnet with hardcoded configurations");
+                TestnetNetwork {
+                    nodes: Arc::new(testnet),
+                }
+            })
+            .await;
+        Ok(network.nodes.clone())
+    }
+}

tests/watcher/utils/watcher.rs

@@ -1,6 +1,4 @@
-use std::time::Duration;
-
-use super::dht::TestnetDHTNetwork;
+use super::testnet::TestnetNetwork;
 use anyhow::{anyhow, Result};
 use chrono::Utc;
 use log::debug;
@@ -13,6 +11,7 @@ use pubky_nexus::events::retry::event::RetryEvent;
 use pubky_nexus::events::Event;
 use pubky_nexus::types::DynError;
 use pubky_nexus::{Config, EventProcessor, PubkyConnector, StackManager};
+use std::time::Duration;
 
 /// Struct to hold the setup environment for tests
 pub struct WatcherTest {
@@ -40,13 +39,15 @@ impl WatcherTest {
         let config = Config::from_env();
         StackManager::setup(&config).await;
 
-        TestnetDHTNetwork::initialise(10)?;
-        let testnet = TestnetDHTNetwork::get_testnet_dht_nodes()?;
+        // testnet initialization is time expensive, we only init one per process
+        let testnet = TestnetNetwork::get().await?;
 
-        let homeserver = Homeserver::start_test(&testnet).await?;
+        let homeserver = testnet.run_homeserver().await.unwrap();
         let homeserver_id = homeserver.public_key().to_string();
 
-        match PubkyConnector::initialise(&config, Some(&testnet)).await {
+        let client = testnet.client_builder().build().unwrap();
+
+        match PubkyConnector::init_from_client(client).await {
             Ok(_) => debug!("WatcherTest: PubkyConnector initialised"),
             Err(e) => debug!("WatcherTest: {}", e),
         }
@@ -145,6 +146,18 @@ impl WatcherTest {
         Ok(user_id)
     }
 
+    pub async fn create_profile(&mut self, user_id: &str, user: &PubkyAppUser) -> Result<String> {
+        let pubky_client = PubkyConnector::get_pubky_client()?;
+        let url = format!("pubky://{}/pub/pubky.app/profile.json", user_id);
+
+        // Write the user profile in the pubky.app repository
+        pubky_client.put(url.as_str()).json(&user).send().await?;
+
+        // Index to Nexus from Homeserver using the events processor
+        self.ensure_event_processing_complete().await?;
+        Ok(user_id.to_string())
+    }
+
     pub async fn create_post(&mut self, user_id: &str, post: &PubkyAppPost) -> Result<String> {
         let post_id = post.create_id();
         let url = format!("pubky://{}/pub/pubky.app/posts/{}", user_id, post_id);