WIP: node: Update p2panda to new API

Cargo.toml

@@ -1,3 +1,6 @@
 [workspace]
 resolver = "2"
 members = ["reflection-app", "reflection-doc", "reflection-node"]
+
+[patch.crates-io]
+iroh-gossip = { git = "https://github.com/p2panda/iroh-gossip", rev = "533c34a2758518ece19c1de9f21bc40d61f9b5a5" }

reflection-node/Cargo.toml

@@ -12,12 +12,12 @@ authors = [
 thiserror = "2.0.17"
 chrono = "0.4.42"
 ciborium = "0.2.2"
-p2panda-core = { git = "https://github.com/p2panda/p2panda", rev = "a83a80b2ce0733c9437ceeaae33503d3b3742436" }
-p2panda-discovery = { git = "https://github.com/p2panda/p2panda", rev = "a83a80b2ce0733c9437ceeaae33503d3b3742436", package = "p2panda-discovery-next" }
-p2panda-net = { git = "https://github.com/p2panda/p2panda", rev = "a83a80b2ce0733c9437ceeaae33503d3b3742436", package = "p2panda-net-next" }
-p2panda-store = { git = "https://github.com/p2panda/p2panda", rev = "a83a80b2ce0733c9437ceeaae33503d3b3742436", features = ["sqlite"], default-features = false }
-p2panda-stream = { git = "https://github.com/p2panda/p2panda", rev = "a83a80b2ce0733c9437ceeaae33503d3b3742436" }
-p2panda-sync = { git = "https://github.com/p2panda/p2panda", rev = "a83a80b2ce0733c9437ceeaae33503d3b3742436", package = "p2panda-sync-next" }
+p2panda-core = { version = "0.5.0" }
+p2panda-discovery = { version = "0.5.0" }
+p2panda-net = { version = "0.5.0" }
+p2panda-store = { version = "0.5.0", features = ["sqlite"], default-features = false }
+p2panda-stream = { version = "0.5.0" }
+p2panda-sync = { version = "0.5.0" }
 serde = { version = "1.0.228", features = ["derive"] }
 serde_bytes = "0.11.19"
 sqlx = { version = "0.8.6", features = ["runtime-tokio", "sqlite", "chrono"], default-features = false }
@@ -27,4 +27,4 @@ tracing = "0.1"
 test-log = { version = "0.2.18", default-features = false, features = ["trace", "color"] }
 hex = "0.4.3"
 rand_chacha = { version = "0.9.0", features = ["os_rng"] }
-iroh = "0.95.1"
+iroh = "0.95.1"

reflection-node/src/author_tracker.rs

@@ -10,7 +10,7 @@ use crate::topic::SubscribableTopic;
 use chrono::Utc;
 use p2panda_core::cbor::{DecodeError, decode_cbor, encode_cbor};
 use p2panda_core::{PrivateKey, PublicKey};
-use p2panda_net::streams::EphemeralStream;
+use p2panda_net::gossip::GossipHandle;
 use tokio::sync::{Mutex, RwLock};
 use tracing::error;
 
@@ -47,7 +47,7 @@ pub struct AuthorTracker<T> {
     last_ping: Mutex<HashMap<PublicKey, Instant>>,
     subscribable_topic: Arc<T>,
     node: Arc<NodeInner>,
-    tx: RwLock<Option<EphemeralStream>>,
+    tx: RwLock<Option<GossipHandle>>,
 }
 
 impl<T: SubscribableTopic> AuthorTracker<T> {
@@ -60,7 +60,7 @@ impl<T: SubscribableTopic> AuthorTracker<T> {
         })
     }
 
-    pub async fn set_topic_tx(&self, tx: Option<EphemeralStream>) {
+    pub async fn set_topic_tx(&self, tx: Option<GossipHandle>) {
         let mut tx_guard = self.tx.write().await;
         // Send good bye message to the network
         if let Some(tx) = tx_guard.as_ref() {
@@ -174,7 +174,7 @@ impl<T: SubscribableTopic> AuthorTracker<T> {
     }
 }
 
-async fn send_message(private_key: &PrivateKey, tx: &EphemeralStream, message: AuthorMessage) {
+async fn send_message(private_key: &PrivateKey, tx: &GossipHandle, message: AuthorMessage) {
     // FIXME: We need to add the current time to the message,
     // because iroh doesn't broadcast twice the same message message.
     let author_message = match encode_cbor(&(&message, SystemTime::now())) {

reflection-node/src/lib.rs

@@ -1,5 +1,6 @@
 mod author_tracker;
 mod ephemerial_operation;
+mod network;
 pub mod node;
 mod node_inner;
 mod operation;

reflection-node/src/network.rs

@@ -0,0 +1,126 @@
+use std::sync::LazyLock;
+
+use p2panda_net::Discovery;
+use p2panda_net::discovery::DiscoveryError;
+use thiserror::Error;
+use tracing::error;
+
+use p2panda_core::Hash;
+use p2panda_core::PrivateKey;
+use p2panda_net::address_book::{AddressBook, AddressBookError};
+use p2panda_net::addrs::NodeInfo;
+use p2panda_net::gossip::{Gossip, GossipError};
+use p2panda_net::iroh_endpoint::{Endpoint, EndpointError};
+use p2panda_net::iroh_mdns::{MdnsDiscovery, MdnsDiscoveryError, MdnsDiscoveryMode};
+
+use crate::operation::ReflectionExtensions;
+use crate::operation_store::OperationStore;
+use crate::topic_store::{LogId, TopicStore};
+
+static RELAY_URL: LazyLock<iroh::RelayUrl> = LazyLock::new(|| {
+    "https://euc1-1.relay.n0.iroh-canary.iroh.link"
+        .parse()
+        .expect("valid relay URL")
+});
+
+static BOOTSTRAP_NODE: LazyLock<NodeInfo> = LazyLock::new(|| {
+    let endpoint_addr = iroh::EndpointAddr::new(
+        "9f63a15ab95959a992af96bf72fbc3e7dc98eeb4799f788bb07b20125053e795"
+            .parse()
+            .expect("valid bootstrap node id"),
+    )
+    .with_relay_url(RELAY_URL.clone());
+    NodeInfo::from(endpoint_addr).bootstrap()
+});
+
+pub type LogSync = p2panda_net::sync::LogSync<
+    p2panda_store::SqliteStore<LogId, ReflectionExtensions>,
+    LogId,
+    ReflectionExtensions,
+    TopicStore,
+>;
+pub type LogSyncError = p2panda_net::sync::LogSyncError<ReflectionExtensions>;
+
+#[derive(Error, Debug)]
+pub enum NetworkError {
+    #[error(transparent)]
+    Gossip(#[from] GossipError),
+    #[error(transparent)]
+    LogSync(#[from] LogSyncError),
+    #[error(transparent)]
+    AddressBook(#[from] AddressBookError),
+    #[error(transparent)]
+    MdnsDiscovery(#[from] MdnsDiscoveryError),
+    #[error(transparent)]
+    Discovery(#[from] DiscoveryError),
+    #[error(transparent)]
+    Endpoint(#[from] EndpointError),
+}
+
+#[allow(dead_code)]
+pub struct Network {
+    pub(crate) mdns_discovery: MdnsDiscovery,
+    pub(crate) discovery: Discovery,
+    pub(crate) gossip: Gossip,
+    pub(crate) log_sync: LogSync,
+    pub(crate) endpoint: Endpoint,
+}
+
+// FIXME: Endpoint, LogSync, MdnsDiscovery, and Gossip should implement debug
+impl std::fmt::Debug for Network {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("Network").finish()
+    }
+}
+
+impl Network {
+    pub async fn new(
+        private_key: &PrivateKey,
+        network_id: &Hash,
+        topic_store: &TopicStore,
+        operation_store: &OperationStore,
+    ) -> Result<Self, NetworkError> {
+        let address_book = AddressBook::builder().spawn().await?;
+
+        if let Err(error) = address_book.insert_node_info(BOOTSTRAP_NODE.clone()).await {
+            error!("Failed to add bootstrap node to the address book: {error}");
+        }
+
+        let endpoint = Endpoint::builder(address_book.clone())
+            .network_id(network_id.into())
+            .private_key(private_key.clone())
+            .relay_url(RELAY_URL.clone())
+            .spawn()
+            .await?;
+
+        let mdns_discovery = MdnsDiscovery::builder(address_book.clone(), endpoint.clone())
+            .mode(MdnsDiscoveryMode::Active)
+            .spawn()
+            .await?;
+
+        let discovery = Discovery::builder(address_book.clone(), endpoint.clone())
+            .spawn()
+            .await?;
+
+        let gossip = Gossip::builder(address_book.clone(), endpoint.clone())
+            .spawn()
+            .await?;
+
+        let log_sync = LogSync::builder(
+            operation_store.clone_inner(),
+            topic_store.clone(),
+            endpoint.clone(),
+            gossip.clone(),
+        )
+        .spawn()
+        .await?;
+
+        Ok(Network {
+            mdns_discovery,
+            discovery,
+            gossip,
+            log_sync,
+            endpoint,
+        })
+    }
+}