diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..027d7b4 --- /dev/null +++ b/.env.example @@ -0,0 +1,2 @@ +BOOTSTRAP_PVT_KEY=308204bd020100300d06092a864886f70d0101010500048204a7308204a30201000282010100d31f4fc9f46fa664ce46637ecda1e415d45dd471f0c8c7f6f921c9375a1cb97d121c0322e376d1dca942fafd284c25d71b917ce68b3379057e2a270ae38395624306ec2149eb58fd6eef50bc99819d59fbcfafc337d89109d75be8715a6c0857a5efc7ec51d6d248ab08a9da9568ff953df3248f7d88788e572eb78b3391078837fa3d1513523d85c8534ca9c3887ba057bb730511028866c6d9eab8be11503280f9239a718c39725e9c6efac75fcc99924ef32ef613178403054517628616604eeb6d949b5e45b8d367193b3266fb4a747386bc9e7062bfbbd1ce34fa3459c702dad13e81a8717c74214fdc2a8926404aefeb4e05cf9eb1e3f40a878f1d4c9702030100010282010059932969485aaec08a7c76ee7014c60f35017287bd5f92e8089ffb1d8f102ceb11e9ecb2bacbc912d485e4b3d08b2e6aebf97882e080c052424b64314c82a1b9053344b0830a65c17175fc4d68dc0385103ff814ddf6f6e2b40afcbe4b5d16d5094489108d5095fbdb055bfbaff4d2abe192edea2b9ae17b9325f11e108cf222660dd7e57732524151df3a56f214c83bcbb445b9f198b0eb73390252c92ab486d925977122412361d4f6750c1dab32dacbbdb03a39986a672f814890bd9c7d962a647c0bc8248ab166e36ca262d8c715195df7b2a861d00707b50509497130143fb48ebf9c8f274c205f0b04760b2d82e77e60ccb128372c3e24b1a22879e71102818100ddb06ab74e6c46ba35f6f2a0fea07fe3b31118dcf31c62e90d20c17f9369e5dd78331098746c2d5b2e3f3a3f4731417e11b2057a0c20a13e16ed69d60c988fcf978a420983308c5334bcfb7d412600649a55047cee9b86e0597cea69ec47494e5c1330026faa15b060867956db955d095bc6ed7affedc20377f425467887ede302818100f3cc39a267610fdbe73443963c0fa486b61572f9d047f8b56ca419b3ea10984bbd12982f36f0bf300fe6171e5ae6b55443dcf3ac63dfc10d9d803da32084987e69e8ab62663ac5836f4c449d3a0c9d02764f2f0889cdbf61252017918b97a644c3caaea808a6a84ae90f327b35499895967394bf8ac7f3954ffbfe7446b264bd02818100d10a97b67e26af92cb6c2f620c787fdfd06ca0cb481b4e517fe59150a14acc4c4819aede7379c653228ad07bf6bfc588015ba2ba71511b7a96d80c62a0c20369b6f5d24b71ff8772b65a35b2e889d2f90a667cb2e9b4923c5f65e10d3c9603d3f612c9c4a252f9dcb5485886c4cc554c8bd44aff1ef1ad6ee3dfbe7e730d8a0d0281804d29fba7fa7e8f2d73fd67d2fa499d32821941ddeddd89cbaa8a3a1466e63fbde1272c81ed2abfe8af6f8d2baf6bcc42d283f1a043b0438352b18f50bd6f1a71108673105ef90520f7ae351d6a8d2978e90761049bd1425cb60a754b3cda8fe6bdedbde03374c4598970a4f2b84806027aaf348a779b97ee4f759bf1239b3d250281801a21da15b58bff6a52f660bf7ecb3d93cefd49f247caf333191ba56f1ecc0493cd9d73d58f142d91bef3d60efd58a7fe1e70f231407777bd1bd6f44e8233c9b4ce37bcb757b107a18d0d24b590f7272dd011c601edd0d92823d5968a397f4d66f1a905572a39910481e7a963b4bc65b323bf8c74fcabad7a25a8e48d54c2aa09 +BOOTSTRAP_NODE=/ip4/127.0.0.1/tcp/8888/p2p/6wrGkCFmUe2H8c23pCfPcXktZztSu2eckvqaWP6y4E2w \ No newline at end of file diff --git a/.gitignore b/.gitignore index 869df07..3549fae 100644 --- a/.gitignore +++ b/.gitignore @@ -1,2 +1,3 @@ /target -Cargo.lock \ No newline at end of file +Cargo.lock +.env \ No newline at end of file diff --git a/Cargo.toml b/Cargo.toml index f069e05..36b6a99 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,5 +1,5 @@ [workspace] -members = ["examples/inference", "core"] +members = ["examples/raw", "examples/agentic", "inode"] resolver = "2" @@ -9,6 +9,7 @@ rust-version = "1.90.0" edition = "2021" [workspace.dependencies] +rnet-p2p = { git = "https://github.com/rnet-stack/rnet-p2p.git" } tokio = { version = "1", features = ["full"] } anyhow = "1" async-trait = "0.1" @@ -23,6 +24,8 @@ unsigned-varint = "0.7" prost = "0.13" x25519-dalek = "2" chacha20poly1305 = { version = "0.10.1" } +dotenvy = "0.15.7" +rand = "0.10.1" [workspace.lints] clippy.pedantic = { level = "allow", priority = -1 } diff --git a/core/src/agent/mod.rs b/core/src/agent/mod.rs deleted file mode 100644 index e69de29..0000000 diff --git a/core/src/p2p/mod.rs b/core/src/p2p/mod.rs deleted file mode 100644 index e69de29..0000000 diff --git a/core/Cargo.toml b/examples/agentic/Cargo.toml similarity index 95% rename from core/Cargo.toml rename to examples/agentic/Cargo.toml index 99d92d1..21c1cbf 100644 --- a/core/Cargo.toml +++ b/examples/agentic/Cargo.toml @@ -1,5 +1,5 @@ [package] -name = "core" +name = "agentic" version = "0.1.0" edition = "2021" diff --git a/examples/inference/src/main.rs b/examples/agentic/src/main.rs similarity index 100% rename from examples/inference/src/main.rs rename to examples/agentic/src/main.rs diff --git a/examples/raw/Cargo.toml b/examples/raw/Cargo.toml new file mode 100644 index 0000000..88f3882 --- /dev/null +++ b/examples/raw/Cargo.toml @@ -0,0 +1,24 @@ +[package] +name = "raw" +version = "0.1.0" +edition = "2021" + +[[bin]] +name = "raw" +path = "src/main.rs" + +[dependencies] +tokio = { workspace = true } +anyhow = { workspace = true } +async-trait = { workspace = true } +tracing = { workspace = true } +tracing-subscriber = { workspace = true } +rsa = { workspace = true } +sha2 = { workspace = true } +serde = { workspace = true } +bincode = { workspace = true } +x25519-dalek = { workspace = true } +chacha20poly1305 = { workspace = true } +dotenvy = { workspace = true } + +inode = { path = "../../inode" } diff --git a/examples/raw/src/main.rs b/examples/raw/src/main.rs new file mode 100644 index 0000000..2d764ee --- /dev/null +++ b/examples/raw/src/main.rs @@ -0,0 +1,33 @@ +use std::io::{self, Write}; + +use anyhow::Result; +use inode::p2p::{cmds::io_loop, node::InferenceNode, types::Mode}; +use tracing_subscriber::EnvFilter; + +#[tokio::main] +async fn main() -> Result<()> { + tracing_subscriber::fmt() + .with_env_filter(EnvFilter::new("trace")) + .without_time() + .with_target(false) + .compact() + .init(); + + dotenvy::dotenv().ok(); + + print!("Intiate at BOOTSTRAP node (Y/n): "); + io::stdout().flush().unwrap(); + + let mut io = String::new(); + io::stdin().read_line(&mut io).unwrap(); + + let mut mode = Mode::Bootstrap; + if io.trim().to_lowercase() != "" { + mode = Mode::Provider; + } + + let inode = InferenceNode::new(mode).await; + io_loop(inode).await.unwrap(); + + Ok(()) +} diff --git a/examples/inference/Cargo.toml b/inode/Cargo.toml similarity index 73% rename from examples/inference/Cargo.toml rename to inode/Cargo.toml index 7378521..b464a4d 100644 --- a/examples/inference/Cargo.toml +++ b/inode/Cargo.toml @@ -1,9 +1,11 @@ [package] -name = "inference" +name = "inode" version = "0.1.0" edition = "2021" [dependencies] +rnet-p2p = { workspace = true } + tokio = { workspace = true } anyhow = { workspace = true } async-trait = { workspace = true } @@ -14,3 +16,6 @@ serde = { workspace = true } bincode = { workspace = true } x25519-dalek = { workspace = true } chacha20poly1305 = { workspace = true } +dotenvy = { workspace = true } +rand = { workspace = true } +hex = { workspace = true } diff --git a/inode/src/agent/mod.rs b/inode/src/agent/mod.rs new file mode 100644 index 0000000..fd2e6d0 --- /dev/null +++ b/inode/src/agent/mod.rs @@ -0,0 +1 @@ +pub mod tools; diff --git a/inode/src/agent/tools.rs b/inode/src/agent/tools.rs new file mode 100644 index 0000000..e620fd5 --- /dev/null +++ b/inode/src/agent/tools.rs @@ -0,0 +1,3 @@ +pub struct Tools {} + +impl Tools {} diff --git a/inode/src/common/mod.rs b/inode/src/common/mod.rs new file mode 100644 index 0000000..5006a36 --- /dev/null +++ b/inode/src/common/mod.rs @@ -0,0 +1,24 @@ +use rand::{distr::Alphanumeric, rng, RngExt}; +use sha2::{Digest, Sha256}; +use std::time::{SystemTime, UNIX_EPOCH}; + +pub fn unix_epoch() -> u64 { + let now = SystemTime::now(); + let duration = now.duration_since(UNIX_EPOCH).unwrap(); + + duration.as_secs() +} + +pub fn generate_entropy() -> (String, String) { + let secret: String = rng() + .sample_iter(&Alphanumeric) + .take(16) + .map(char::from) + .collect(); + + let mut hasher = Sha256::new(); + hasher.update(secret.as_bytes()); + let hash = hex::encode(hasher.finalize()); + + (secret, hash) +} diff --git a/core/src/lib.rs b/inode/src/lib.rs similarity index 63% rename from core/src/lib.rs rename to inode/src/lib.rs index 69e04ad..614154d 100644 --- a/core/src/lib.rs +++ b/inode/src/lib.rs @@ -1,3 +1,3 @@ - pub mod agent; +pub mod common; pub mod p2p; diff --git a/inode/src/p2p/cmds.rs b/inode/src/p2p/cmds.rs new file mode 100644 index 0000000..bfbb05e --- /dev/null +++ b/inode/src/p2p/cmds.rs @@ -0,0 +1,163 @@ +use anyhow::Result; +use rnet_p2p::identity::multiaddr::Multiaddr; +use rnet_p2p::identity::traits::{ + core::INode, + protocols::{INodeFloodsubAPI, INodePingAPI}, +}; +use std::collections::HashMap; +use std::{io::Write, sync::Arc, time::Duration}; +use tokio::io::{self, AsyncBufReadExt}; + +use crate::p2p::node::InferenceNode; +use crate::p2p::types::IMsgType; + +const CLI_DELAY: Duration = Duration::from_nanos(1000); +const FLOODSUB: &str = "rnet/floodsub/0.0.1"; +const COMMANDS: &[&str] = &[ + "help => print all the commands", + "local => get local peer-info", + "connect => connect with a new peer", + "ping => exchange ping with a peer", + "peers => list the connected peers", + "\n", + "fsub => open a new floodsub stream with the peer", + "join => subscribe to a new-topic", + "leave => unsubscribe to a new-topic", + "publish => publish a msg to a topic", + "topics => list the subscribed topics", + "fpeers => list the connected Floodsub peers", + "bootmesh => map of topics -> peer (BOOTSTRAP)", + "mesh => map of topics -> peer", + "\n", +]; + +pub fn print_commands() { + for cmd in COMMANDS { + println!(" {}", cmd); + } +} + +pub async fn handle_cmd(line: &str, inode: &Arc) -> Result<()> { + let mut parts = line.split_whitespace(); + let cmd = parts.next().unwrap(); + + match cmd { + "help" => { + print_commands(); + } + + "local" => { + let peer_info = inode.local.clone(); + println!("{}", peer_info); + } + + "connect" => { + let maddr = Multiaddr::new(parts.next().unwrap()).unwrap(); + inode.p2p.connect(&maddr).await?; + tokio::time::sleep(Duration::from_millis(300)).await; + } + + "ping" => { + let maddr = parts.next().unwrap(); + let count: u32 = parts.next().unwrap_or("0").parse().unwrap_or(0); + inode.p2p.ping(Some(count), maddr).await.unwrap(); + } + + "fsub" => { + let maddr = parts.next().unwrap(); + inode + .p2p + .new_stream(maddr, vec![FLOODSUB.to_string()]) + .await + .unwrap(); + tokio::time::sleep(Duration::from_millis(300)).await; + } + + "join" => { + let topic = parts.next().unwrap().to_string(); + inode.p2p.floodsub_subscribe(topic).await.unwrap(); + } + + "leave" => { + let topic = parts.next().unwrap().to_string(); + inode.p2p.floodsub_unsubscribe(vec![topic]).await.unwrap(); + } + + "publish" => { + let topic = parts.next().unwrap().to_string(); + let msg = parts.collect::>().join(" "); + + // Wrap this into IMsgType::General + let mpc_general = IMsgType::General(msg); + let payload = bincode::serialize(&mpc_general).unwrap(); + + inode.p2p.floodsub_publish(topic, payload).await.unwrap(); + } + + "topics" => { + let topics = inode.p2p.floodsub_topics().await.unwrap_or(vec![]); + println!("{:?}", topics); + } + + "fpeers" => { + let fpeers = inode.p2p.floodsub_peers().await.unwrap_or(vec![]); + fpeers.iter().for_each(|x| { + println!("{}", x); + }); + } + + "bootmesh" => { + let bootmesh = inode.bootmesh.lock().await.clone(); + + bootmesh.iter().for_each(|(topic, peers)| { + println!("- {}", topic); + peers.iter().for_each(|peer| println!(" - {}", peer)); + }); + } + + "peers" => { + let gpeers = inode.p2p.get_peers().await; + gpeers.iter().for_each(|peer| println!("{}", peer)); + } + + "mesh" => { + let mesh = inode.p2p.floodsub_mesh().await.unwrap_or(HashMap::new()); + + mesh.iter().for_each(|(topic, peers)| { + println!("- {}", topic); + peers.iter().for_each(|peer| println!(" - {}", peer)); + }); + } + + _ => println!("Unknown command"), + } + Ok(()) +} + +pub async fn io_loop(inode: Arc) -> Result<()> { + let stdin = io::BufReader::new(io::stdin()); + let mut lines = stdin.lines(); + + println!("\n inode CLI ready. Commands:"); + print_commands(); + + loop { + print!("\nCommand => "); + std::io::stdout().flush().unwrap(); + + let line = match lines.next_line().await? { + Some(line) => line, + None => break, + }; + + let line = line.trim(); + if line.is_empty() { + continue; + } + + handle_cmd(line, &inode).await?; + tokio::time::sleep(CLI_DELAY).await; + } + + Ok(()) +} diff --git a/inode/src/p2p/mod.rs b/inode/src/p2p/mod.rs new file mode 100644 index 0000000..49ab53d --- /dev/null +++ b/inode/src/p2p/mod.rs @@ -0,0 +1,4 @@ +pub mod cmds; +pub mod node; +pub mod service; +pub mod types; diff --git a/inode/src/p2p/node.rs b/inode/src/p2p/node.rs new file mode 100644 index 0000000..f73d01b --- /dev/null +++ b/inode/src/p2p/node.rs @@ -0,0 +1,184 @@ +use anyhow::Result; +use rnet_p2p::{ + identity::{ + events::{FloodsubMsgType, GlobalEvent}, + multiaddr::Multiaddr, + traits::{core::INode, protocols::INodeFloodsubAPI}, + }, + node::{inner::NodeInner, node::Node, protocol::InnerProtocolOpt}, + protocols::FLOODSUB, +}; +use std::{collections::HashMap, env, sync::Arc, time::Duration}; +use tokio::sync::{mpsc::Receiver, Mutex}; +use tracing::{debug, info}; + +use crate::p2p::types::{IMsgType, Mode}; + +pub type BOOTMESH = Arc>>>; +pub const PROVIDER_MESH: &str = "swarm/mesh"; + +pub struct InferenceNode { + pub p2p: Arc, + + pub mode: Mode, + pub local: Multiaddr, + + pub bootmesh: BOOTMESH, +} + +impl InferenceNode { + pub async fn new(mode: Mode) -> Arc { + debug!("Running as {:?} node", mode); + + let (p2p_tx, event_rx, listen_addr) = match mode { + Mode::Bootstrap => { + let mut listen_addr = Multiaddr::new("ip4/127.0.0.1/tcp/8888").unwrap(); + let key_hex = env::var("BOOTSTRAP_PVT_KEY").unwrap(); + let (host_mpsc_tx, global_rx) = NodeInner::new( + &mut listen_addr, + vec![InnerProtocolOpt::Floodsub, InnerProtocolOpt::Ping], + Some(key_hex), + ) + .await + .unwrap(); + + (host_mpsc_tx, global_rx, listen_addr) + } + _ => { + let mut listen_addr = Multiaddr::new("ip4/127.0.0.1/tcp/0").unwrap(); + let (host_mpsc_tx, global_rx) = NodeInner::new( + &mut listen_addr, + vec![InnerProtocolOpt::Floodsub, InnerProtocolOpt::Ping], + None, + ) + .await + .unwrap(); + + (host_mpsc_tx, global_rx, listen_addr) + } + }; + + let inode = Arc::new(InferenceNode { + p2p: p2p_tx.clone(), + mode, + local: listen_addr, + bootmesh: Arc::new(Mutex::new(HashMap::new())), + }); + + let events = inode.clone(); + tokio::spawn(async move { + events.event_handler(event_rx).await.unwrap(); + }); + + tokio::time::sleep(Duration::from_millis(500)).await; + + let helper = inode.clone(); + inode.initiate(helper).await.unwrap(); + + inode + } + + pub async fn initiate(&self, mesh_mpc_node: Arc) -> Result<()> { + self.p2p + .floodsub_subscribe(PROVIDER_MESH.to_string()) + .await + .unwrap(); + tokio::time::sleep(Duration::from_millis(2000)).await; + + match self.mode { + Mode::Bootstrap => { + tokio::spawn(async move { + mesh_mpc_node.periodic_mesh_update().await.unwrap(); + }); + } + Mode::Provider => { + // CONNECT TO BOOTSTRAP NODE + let bootstrap_node = Multiaddr::new(&env::var("BOOTSTRAP_NODE").unwrap()).unwrap(); + info!("BOOTSTRAP node found, CONNECTING...\n"); + self.p2p + .new_stream(&bootstrap_node.to_string(), vec![FLOODSUB.to_string()]) + .await + .unwrap(); + + tokio::time::sleep(Duration::from_millis(2000)).await; + } + } + + Ok(()) + } + + pub async fn event_handler(&self, mut event_rx: Receiver>) -> Result<()> { + loop { + let notification = event_rx.recv().await.unwrap(); + let decoded = bincode::deserialize::(¬ification).unwrap(); + + match decoded { + GlobalEvent::Floodsub(event) => match event.msg_type { + FloodsubMsgType::Publish => { + let topic = event.topic.clone(); + let source = event.source.unwrap(); + let decoded_msg = + bincode::deserialize::(&event.msg.unwrap()).unwrap(); + + match decoded_msg { + IMsgType::General(msg) => { + debug!("FloodsubEvent: {topic} - {source}: {msg}") + } + IMsgType::Service(_payload) => {} + IMsgType::Bootmesh(mesh) => { + let mut bootmesh = self.bootmesh.lock().await; + *bootmesh = mesh; + + debug!("BOOTMESH"); + } + } + } + + FloodsubMsgType::Subscribe => {} + _ => {} + }, + GlobalEvent::Ping(event) => debug!("{:?}", event), + } + } + } + + pub async fn periodic_mesh_update(&self) -> Result<()> { + loop { + let bootmesh = { + let bootmesh = self.bootmesh.lock().await; + bootmesh.clone() + }; + + let latest_mesh = self.p2p.floodsub_mesh().await.unwrap_or(HashMap::new()); + + if bootmesh == latest_mesh { + tokio::time::sleep(Duration::from_secs(1)).await; + continue; + } + + { + let mut bootmesh = self.bootmesh.lock().await; + *bootmesh = latest_mesh.clone(); + } + + self.broadcast_bootmesh(latest_mesh).await.unwrap(); + } + } + + pub async fn broadcast_bootmesh(&self, mesh: HashMap>) -> Result<()> { + let fsub_msg = IMsgType::Bootmesh(mesh); + let payload = bincode::serialize(&fsub_msg).unwrap(); + + // Wait 2 seconds for the new node to settle down + tokio::time::sleep(Duration::from_secs(2)).await; + + debug!("BOOTMESH"); + + self.p2p + .floodsub_publish(PROVIDER_MESH.to_string(), payload) + .await + .unwrap(); + + Ok(()) + } +} diff --git a/inode/src/p2p/service.rs b/inode/src/p2p/service.rs new file mode 100644 index 0000000..725d393 --- /dev/null +++ b/inode/src/p2p/service.rs @@ -0,0 +1,4 @@ +use serde::{Deserialize, Serialize}; + +#[derive(Debug, Serialize, Deserialize)] +pub struct IPayload {} diff --git a/inode/src/p2p/types.rs b/inode/src/p2p/types.rs new file mode 100644 index 0000000..ec72272 --- /dev/null +++ b/inode/src/p2p/types.rs @@ -0,0 +1,18 @@ +use std::collections::HashMap; + +use serde::{Deserialize, Serialize}; + +use crate::p2p::service::IPayload; + +#[derive(Debug)] +pub enum Mode { + Provider, + Bootstrap, +} + +#[derive(Debug, Serialize, Deserialize)] +pub enum IMsgType { + General(String), + Service(IPayload), + Bootmesh(HashMap>), +}