Feature/iroh net

Cargo.toml

@@ -98,6 +98,7 @@ tokio = { version = "1", features = ["net"] }
 tokio-stream = { version = "0.1.8", features = ["io-util"] }
 tokio-util = { version = "0.7", features = ["codec"] }
 url = "2.1.1"
+typed-builder = "0.20.0"
 
 [dev-dependencies]
 erc20_processor = { workspace = true }
@@ -242,7 +243,7 @@ members = [
   "test-utils/test-framework/framework-macro",
   "test-utils/test-framework/framework-basic",
   "test-utils/test-framework/framework-mocks",
-  "core/healthcheck",
+  "core/healthcheck", "core/net-iroh",
 ]
 
 [workspace.dependencies]

core/model/src/net.rs

@@ -465,13 +465,34 @@ impl RemoteEndpoint for NetDst {
 
 #[cfg(test)]
 mod tests {
+    use ya_client_model::NodeId;
     use super::*;
 
     #[test]
     fn ok_try_service_on_public() {
-        "0xbabe000000000000000000000000000000000000"
+        let ep = "0xbabe000000000000000000000000000000000000"
             .try_service("/public/x")
             .unwrap();
+        eprintln!("addr={}", ep.addr());
+
+        let n1 : NodeId ="0xbabe000000000000000000000000000000000000".parse().unwrap();
+        let n2 : NodeId ="0xcafe000000000000000000000000000000000000".parse().unwrap();
+
+        let ep = super::from(n1)
+            .to(n2)
+            .service("/public/x");
+        eprintln!("service from: {n1} to {n2} addr={}", ep.addr());
+
+        let ep = super::from(n1)
+            .to(n2)
+            .service_udp("/public/x");
+        eprintln!("service_udp from: {n1} to {n2} addr={}", ep.addr());
+
+        let ep = super::from(n1)
+            .to(n2)
+            .service_transfer("/public/x");
+        eprintln!("service_transfer from: {n1} to {n2} addr={}", ep.addr());
+
     }
 
     #[test]

core/net-iroh/Cargo.toml

@@ -0,0 +1,23 @@
+[package]
+name = "net-iroh"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+actix-rt.workspace = true
+anyhow = "1.0.93"
+bincode = "1.3.3"
+futures = "0.3.31"
+iroh-net = {  version = "0.28.1", features = ['discovery-local-network', "iroh-relay"] }
+rustyline = "14.0.0"
+serde = { version = "1.0.214", features = ["derive"] }
+ya-core-model = { workspace = true, features = ["net", "identity"] }
+bytes = { version = "1" }
+ethsign = { version = "0.8" }
+url.workspace = true
+
+
+[lints]
+workspace = true

core/net-iroh/examples/fake.rs

@@ -0,0 +1,43 @@
+use futures::prelude::*;
+use iroh_net::discovery::dns::DnsDiscovery;
+use iroh_net::discovery::local_swarm_discovery::LocalSwarmDiscovery;
+use iroh_net::discovery::pkarr::dht::DhtDiscovery;
+use iroh_net::discovery::pkarr::PkarrPublisher;
+use iroh_net::discovery::ConcurrentDiscovery;
+use iroh_net::key::SecretKey;
+use iroh_net::Endpoint;
+
+#[actix_rt::main]
+async fn main() -> anyhow::Result<()> {
+    let secret_key = SecretKey::generate();
+    let id = secret_key.public();
+    let dht = DhtDiscovery::builder()
+        .dht(true)
+        .secret_key(secret_key.clone())
+        .build()?;
+
+    let discovery = ConcurrentDiscovery::from_services(vec![
+        Box::new(LocalSwarmDiscovery::new(id)?),
+        Box::new(dht),
+    ]);
+    let ep = Endpoint::builder()
+        .secret_key(secret_key)
+        .discovery(Box::new(discovery))
+        .bind()
+        .await?;
+
+    ep.discovery()
+        .unwrap()
+        .subscribe()
+        .unwrap()
+        .for_each(|di| {
+            eprintln!("di: {:?}", di);
+            async { () }
+        })
+        .await;
+
+    let addr = ep.node_addr().await?;
+    eprintln!("node addr = {:?}", addr);
+
+    Ok(())
+}

core/net-iroh/src/lib.rs

@@ -0,0 +1,109 @@
+// YA_NET_BIND_URL=udp://0.0.0.0:11500
+//
+
+use anyhow::Result;
+use bytes::Bytes;
+use ethsign::{PublicKey, Signature};
+use futures::future::LocalBoxFuture;
+use futures::Stream;
+use serde::{Deserialize, Serialize};
+use std::rc::Rc;
+use url::Url;
+use ya_core_model::NodeId;
+
+///
+/// let client = NetClient::builder()
+///     .bind_url("udp://0.0.0.0:11500")
+///     .crypto_provider(provider)
+///     .start().await?;
+///
+
+pub struct NetClientBuilder {
+    _inner: (),
+}
+
+impl NetClientBuilder {
+
+    pub fn bind_url(self, _url : Url) -> Self {
+        self
+    }
+
+    pub fn crypto_provider(self, _cypher : impl CryptoProvider) -> Self {
+        self
+    }
+
+    pub async fn start(self) -> Result<NetClient> {
+        todo!()
+    }
+
+}
+
+#[derive(Clone)]
+pub struct NetClient {
+    _inner: (),
+}
+
+impl NetClient {
+
+    pub fn builder() -> NetClientBuilder {
+        NetClientBuilder {
+            _inner: (),
+        }
+    }
+
+    pub async fn send_msg(&self, from: NodeId, to: NodeId, msg: Bytes) -> Result<()> {
+        todo!()
+    }
+
+    pub async fn recv_msg(&self) -> Result<(NodeId, NodeId, Bytes)> {
+        todo!()
+    }
+
+    pub async fn send_unreliable_msg(&self, from: NodeId, to: NodeId, msg: Bytes) -> Result<()> {
+        todo!()
+    }
+
+    pub async fn recv_unreliable_msg(&self) -> Result<(NodeId, NodeId, Bytes)> {
+        todo!()
+    }
+
+    pub async fn find_mtu(&self, to: NodeId) -> Result<usize> {
+        todo!()
+    }
+
+    pub async fn send_broadcast(
+        &self,
+        from: NodeId,
+        topic: String,
+        size: usize,
+    ) -> Result<(NodeId, NodeId, Bytes)> {
+        todo!()
+    }
+
+    pub async fn recv_broadcast(&self, topic: String) -> impl Stream<Item = (NodeId, Bytes)> {
+        futures::stream::empty()
+    }
+
+    pub async fn status(&self) -> Result<NetStatus> {
+        todo!()
+    }
+}
+
+#[derive(Serialize, Deserialize)]
+pub struct NetStatus {}
+
+pub trait CryptoProvider {
+    fn default_id<'a>(&self) -> LocalBoxFuture<'a, anyhow::Result<NodeId>>;
+    fn aliases<'a>(&self) -> LocalBoxFuture<'a, anyhow::Result<Vec<NodeId>>>;
+    fn get<'a>(&self, node_id: NodeId) -> LocalBoxFuture<'a, anyhow::Result<Rc<dyn Crypto>>>;
+}
+
+pub trait Crypto {
+    fn public_key<'a>(&self) -> LocalBoxFuture<'a, anyhow::Result<PublicKey>>;
+    fn sign<'a>(&self, message: &'a [u8]) -> LocalBoxFuture<'a, anyhow::Result<Signature>>;
+    fn encrypt<'a>(
+        &self,
+        message: &'a [u8],
+        remote_key: &'a PublicKey,
+    ) -> LocalBoxFuture<'a, anyhow::Result<Vec<u8>>>;
+}

core/net/Cargo.toml

@@ -9,6 +9,7 @@ default = []
 service = []
 # Temporary to make goth integration tests work
 central-net = []
+iroh-net = []
 packet-trace-enable = [
     "ya-packet-trace/enable",
     "ya-relay-client/packet-trace-enable",
@@ -25,8 +26,10 @@ ya-sb-util = { workspace = true }
 ya-service-api.workspace = true
 ya-service-api-interfaces.workspace = true
 ya-service-bus = { workspace = true, features = ["tls"] }
-ya-utils-networking.workspace = true
+ya-utils-networking = { workspace = true, features = ["dns"] }
 ya-packet-trace = { git = "https://github.com/golemfactory/ya-packet-trace" }
+net-iroh = { path = "../net-iroh" }
+postcard = { version = "1.0.10", features = ["alloc"]}
 
 actix.workspace = true
 actix-web.workspace = true
@@ -51,6 +54,7 @@ url = { version = "2.2" }
 prost = { version = "0.10" }
 rand = { version = "0.7" }
 regex = { workspace = true }
+serde = { version = "1.0", features = ["derive"] }
 
 [dev-dependencies]
 ya-sb-proto = { workspace = true }

core/net/src/config.rs

@@ -19,14 +19,11 @@ use url::Url;
 )]
 #[strum(serialize_all = "kebab-case")]
 pub enum NetType {
-    /// TODO: Remove compilation flag.
-    ///  This conditional compilation is hack to make Goth integration tests work.
-    ///  Current solution in Goth is to build separate binary with compilation flag.
-    ///  This is only temporary for transition period, to make this PR as small as possible.
     #[cfg_attr(feature = "central-net", default)]
     Central,
-    #[cfg_attr(not(feature = "central-net"), default)]
     Hybrid,
+    #[cfg_attr(not(feature = "central-net"), default)]
+    IROH,
 }
 
 #[derive(StructOpt, Clone)]

core/net/src/iroh.rs

@@ -0,0 +1,93 @@
+use ya_core_model::net::local::{BindBroadcastError, BroadcastMessage, SendBroadcastMessage, ToEndpoint};
+use std::sync::Arc;
+use actix_web::Scope;
+use futures::{future, stream};
+use ya_service_bus::{Error, RpcEndpoint, RpcMessage};
+use crate::Config;
+use net_iroh::NetClient;
+use ya_service_bus::untyped::{Fn4HandlerExt, Fn4StreamHandlerExt};
+use ya_service_bus::typed as bus;
+use ya_core_model::net;
+
+mod local_service;
+mod cli;
+mod bridge;
+mod crypto;
+mod rpc;
+
+
+pub struct IRohNet;
+
+pub async fn gsb<Context>(_: Context, config: Config) -> anyhow::Result<()> {
+    use ya_service_bus::{untyped as gsb, error::Error as GsbError, ResponseChunk};
+
+    let (default_id, ids) = crate::service::identities().await?;
+    let crypto = self::crypto::IdentityCryptoProvider::new(default_id);
+    let client = NetClient::builder().bind_url(config.bind_url).crypto_provider(crypto).start().await?;
+
+    // /net/{id}/{service} -> /public/{service}
+    // /transfer/net/{dst}/service
+    // /udp/net/{dst}/service
+
+    // /from/{src}/to/{dst}/{service} -> /public/{service}
+    // /udp/from/{src}/to/{dst}/{service} -> /public/{service}
+    // /transfer/from/{src}/to/{dst}/{service} -> /public/{service}
+
+
+    // -> Future<Result<Vec<u8>, GsbError>
+    let rpc = move |caller: &str, addr: &str, msg: &[u8], no_reply: bool| {
+
+        future::ok(Vec::new())
+    };
+
+    let rpc_stream = move |caller: &str, addr: &str, msg: &[u8], no_reply: bool| {
+
+        stream::once(future::ok(ResponseChunk::Full(Vec::new())))
+    };
+
+    let _ = gsb::subscribe("/net", rpc.into_handler(), rpc_stream.into_stream_handler());
+
+    Ok(())
+}
+
+pub fn scope() -> Scope {
+    todo!()
+}
+
+
+pub async fn bind_broadcast_with_caller<M, T, F>(
+    broadcast_address: &str,
+    handler: F,
+) -> Result<(), BindBroadcastError>
+where
+    M: BroadcastMessage + Send + Sync + 'static,
+    T: std::future::Future<
+        Output = Result<
+            <SendBroadcastMessage<M> as RpcMessage>::Item,
+            <SendBroadcastMessage<M> as RpcMessage>::Error,
+        >,
+    > + 'static,
+    F: FnMut(String, SendBroadcastMessage<M>) -> T + 'static,
+{
+    let address = broadcast_address.to_string();
+    let subscription = M::into_subscribe_msg(address.clone());
+
+    log::debug!(
+        "Binding broadcast handler for topic: {}",
+        subscription.topic()
+    );
+
+    bus::service(net::local::BUS_ID)
+        .send(subscription)
+        .await??;
+
+    log::debug!(
+        "Binding handler '{broadcast_address}' for broadcast topic {}.",
+        M::TOPIC
+    );
+
+    // We created endpoint address above. Now we must add handler, which will
+    // handle broadcasts forwarded to this address.
+    bus::bind_with_caller(broadcast_address, handler);
+    Ok(())
+}