diff --git a/README.md b/README.md index 385796b..3c5ccdb 100644 --- a/README.md +++ b/README.md @@ -63,20 +63,38 @@ let result = fastn_p2p_client::call( ).await?; ``` -### 4. Server Applications +### 4. Server Applications (Revolutionary!) ```rust // Add to Cargo.toml: fastn-p2p = "0.1" use fastn_p2p; -// Server uses full fastn-p2p crate with daemon utilities -let private_key = load_identity_key("alice").await?; -fastn_p2p::listen(private_key) - .handle_requests("Echo", echo_handler) - .await?; +// Modern multi-identity server with complete lifecycle +#[fastn_p2p::main] +async fn main() -> Result<(), Box> { + fastn_p2p::serve_all() + .protocol("mail.fastn.com", |p| p + .on_global_load(setup_shared_database) + .on_create(create_workspace) + .on_activate(start_mail_services) + .handle_requests("inbox.get-mails", get_mails_handler) + .handle_requests("send.send-mail", send_mail_handler) + .handle_requests("settings.add-forwarding", forwarding_handler) + .handle_streams("sync.imap-sync", imap_sync_handler) + .on_deactivate(stop_mail_services) + .on_delete(cleanup_workspace) + .on_global_unload(cleanup_shared_database) + ) + .serve() + .await +} -async fn echo_handler(req: EchoRequest) -> Result { - Ok(EchoResponse { echoed: format!("Echo: {}", req.message) }) +fn create_workspace(ctx: BindingContext) -> impl Future> { + async move { + // Create mail storage dirs in ctx.protocol_dir for ctx.identity + println!("Creating workspace for {} ({})", ctx.identity.id52(), ctx.bind_alias); + Ok(()) + } } ``` diff --git a/examples/Cargo.toml b/examples/Cargo.toml index 2c27c06..45deb52 100644 --- a/examples/Cargo.toml +++ b/examples/Cargo.toml @@ -9,6 +9,7 @@ fastn-p2p-client = { path = "../fastn-p2p-client" } fastn-p2p = { path = "../fastn-p2p" } # For server-side APIs tokio.workspace = true serde.workspace = true +serde_json.workspace = true clap.workspace = true thiserror.workspace = true reqwest.workspace = true diff --git a/examples/src/request_response.rs b/examples/src/request_response.rs index c122a17..8157c0a 100644 --- a/examples/src/request_response.rs +++ b/examples/src/request_response.rs @@ -40,7 +40,7 @@ async fn main() -> Result<(), Box> { .on_create(echo_create_handler) .on_activate(echo_activate_handler) .on_check(echo_check_handler) - .handle_requests("basic-echo", fastn_p2p::echo_request_handler) + .handle_requests("basic-echo", echo_request_handler) .on_reload(echo_reload_handler) .on_deactivate(echo_deactivate_handler) ) @@ -53,7 +53,7 @@ use std::pin::Pin; use std::future::Future; fn echo_create_handler( - ctx: fastn_p2p::server::BindingContext, + ctx: fastn_p2p::BindingContext, ) -> Pin>> + Send>> { Box::pin(async move { println!("๐Ÿ”ง Echo create: {} {} ({})", ctx.identity.id52(), ctx.bind_alias, ctx.protocol_dir.display()); @@ -63,7 +63,7 @@ fn echo_create_handler( } fn echo_activate_handler( - ctx: fastn_p2p::server::BindingContext, + ctx: fastn_p2p::BindingContext, ) -> Pin>> + Send>> { Box::pin(async move { println!("๐Ÿš€ Echo activate: {} {} ({})", ctx.identity.id52(), ctx.bind_alias, ctx.protocol_dir.display()); @@ -73,7 +73,7 @@ fn echo_activate_handler( } fn echo_check_handler( - ctx: fastn_p2p::server::BindingContext, + ctx: fastn_p2p::BindingContext, ) -> Pin>> + Send>> { Box::pin(async move { println!("๐Ÿ” Echo check: {} {} ({})", ctx.identity.id52(), ctx.bind_alias, ctx.protocol_dir.display()); @@ -83,7 +83,7 @@ fn echo_check_handler( } fn echo_reload_handler( - ctx: fastn_p2p::server::BindingContext, + ctx: fastn_p2p::BindingContext, ) -> Pin>> + Send>> { Box::pin(async move { println!("๐Ÿ”„ Echo reload: {} {} ({})", ctx.identity.id52(), ctx.bind_alias, ctx.protocol_dir.display()); @@ -93,11 +93,51 @@ fn echo_reload_handler( } fn echo_deactivate_handler( - ctx: fastn_p2p::server::BindingContext, + ctx: fastn_p2p::BindingContext, ) -> Pin>> + Send>> { Box::pin(async move { println!("๐Ÿ›‘ Echo deactivate: {} {} ({})", ctx.identity.id52(), ctx.bind_alias, ctx.protocol_dir.display()); // TODO: Stop accepting requests, but preserve data Ok(()) }) +} + +// Typed Echo request handler for this application +fn echo_request_handler( + ctx: fastn_p2p::RequestContext, + request: serde_json::Value, +) -> Pin>> + Send>> { + Box::pin(async move { + println!("๐Ÿ’ฌ Echo handler called:"); + println!(" Identity: {}", ctx.identity.id52()); + println!(" Bind alias: {}", ctx.bind_alias); + println!(" Command: {}", ctx.command); + println!(" Protocol dir: {}", ctx.protocol_dir.display()); + println!(" Args: {:?}", ctx.args); + + // Parse typed request + let echo_request: EchoRequest = serde_json::from_value(request) + .map_err(|e| format!("Invalid EchoRequest: {}", e))?; + + if echo_request.message.is_empty() { + return Err("Message cannot be empty".into()); + } + + println!(" Message: '{}'", echo_request.message); + + // Create typed response with args support + let args_info = if ctx.args.is_empty() { + String::new() + } else { + format!(" [args: {}]", ctx.args.join(", ")) + }; + + let echo_response = EchoResponse { + echoed: format!("Echo from {} ({}): {}{}", ctx.identity.id52(), ctx.command, echo_request.message, args_info) + }; + + let response = serde_json::to_value(echo_response)?; + println!("๐Ÿ“ค Echo response: {}", response); + Ok(response) + }) } \ No newline at end of file diff --git a/fastn-net/src/protocol.rs b/fastn-net/src/protocol.rs index 09912da..f8670dd 100644 --- a/fastn-net/src/protocol.rs +++ b/fastn-net/src/protocol.rs @@ -83,36 +83,19 @@ //! based on performance requirements. #[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq)] pub enum Protocol { - /// client can send this message to check if the connection is open / healthy. + /// Active built-in protocols Ping, - /// client may not be using NTP, or may only have p2p access and no other internet access, in - /// which case it can ask for the time from the peers and try to create a consensus. WhatTimeIsIt, - /// client wants to make an HTTP request to a device whose ID is specified. note that the exact - /// ip:port is not known to peers, they only the "device id" for the service. server will figure - /// out the ip:port from the device id. Http, HttpProxy, - /// if the client wants their traffic to route via this server, they can send this. for this to - /// work, the person owning the device must have created a SOCKS5 device, and allowed this peer - /// to access it. Socks5, Tcp, - // TODO: RTP/"RTCP" for audio video streaming - - // Fastn-specific protocols for entity communication - /// Messages from Device to Account (sync requests, status reports, etc.) - DeviceToAccount, - /// Messages between Accounts (email, file sharing, automerge sync, etc.) - AccountToAccount, - /// Messages from Account to Device (commands, config updates, notifications, etc.) - AccountToDevice, - /// Control messages for Rig management (bring online/offline, set current, etc.) - RigControl, - - /// Generic protocol for user-defined types - /// This allows users to define their own protocol types while maintaining - /// compatibility with the existing fastn-net infrastructure. + + /// Revolutionary per-application protocols for serve_all() architecture + /// Protocol names like "mail.fastn.com", "echo.fastn.com", etc. + Application(String), + + /// Legacy generic protocol (still used by existing fastn-p2p code) Generic(serde_json::Value), } @@ -125,10 +108,7 @@ impl std::fmt::Display for Protocol { Protocol::HttpProxy => write!(f, "HttpProxy"), Protocol::Socks5 => write!(f, "Socks5"), Protocol::Tcp => write!(f, "Tcp"), - Protocol::DeviceToAccount => write!(f, "DeviceToAccount"), - Protocol::AccountToAccount => write!(f, "AccountToAccount"), - Protocol::AccountToDevice => write!(f, "AccountToDevice"), - Protocol::RigControl => write!(f, "RigControl"), + Protocol::Application(name) => write!(f, "{}", name), Protocol::Generic(value) => write!(f, "Generic({value})"), } } @@ -165,13 +145,25 @@ mod tests { /// See module documentation for detailed rationale. pub const APNS_IDENTITY: &[u8] = b"/fastn/entity/0.1"; -/// Protocol header with optional metadata. +/// Protocol header for both built-in and revolutionary serve_all() protocols. /// /// Sent at the beginning of each bidirectional stream to identify -/// the protocol and provide any protocol-specific metadata. -#[derive(Debug)] +/// the protocol and provide routing information when needed. +#[derive(Debug, serde::Serialize, serde::Deserialize)] pub struct ProtocolHeader { + /// Protocol identifier pub protocol: Protocol, + + /// Command within the protocol (for Application protocols only) + pub command: Option, + + /// Protocol binding alias (for Application protocols only) + pub bind_alias: Option, + + /// CLI arguments support (issue #13: stdargs) + pub args: Vec, + + /// Legacy compatibility for extra protocol data pub extra: Option, } @@ -179,7 +171,47 @@ impl From for ProtocolHeader { fn from(protocol: Protocol) -> Self { Self { protocol, + command: None, // Built-in protocols don't need commands + bind_alias: None, // Built-in protocols don't need bind aliases + args: Vec::new(), extra: None, } } } + +impl ProtocolHeader { + /// Create protocol header for serve_all() Application protocols + pub fn for_application( + protocol_name: String, + command: String, + bind_alias: String, + args: Vec, + ) -> Self { + Self { + protocol: Protocol::Application(protocol_name), + command: Some(command), + bind_alias: Some(bind_alias), + args, + extra: None, + } + } + + /// Get routing information for serve_all() (returns None for built-in protocols) + pub fn serve_all_routing(&self) -> Option<(&str, &str, &str, &[String])> { + match &self.protocol { + Protocol::Application(protocol_name) => { + if let (Some(command), Some(bind_alias)) = (&self.command, &self.bind_alias) { + Some((protocol_name, command, bind_alias, &self.args)) + } else { + None + } + } + _ => None, // Built-in protocols don't have serve_all routing + } + } + + /// Check if this is a built-in protocol + pub fn is_builtin(&self) -> bool { + !matches!(self.protocol, Protocol::Application(_)) + } +} diff --git a/fastn-p2p-client/src/client.rs b/fastn-p2p-client/src/client.rs index c65fd85..ce95426 100644 --- a/fastn-p2p-client/src/client.rs +++ b/fastn-p2p-client/src/client.rs @@ -20,6 +20,16 @@ pub enum DaemonRequest { bind_alias: String, request: T, }, + #[serde(rename = "call_with_args")] + CallWithArgs { + from_identity: String, + to_peer: fastn_id52::PublicKey, + protocol: String, + command: String, + bind_alias: String, + args: Vec, + request: T, + }, #[serde(rename = "stream")] Stream { from_identity: String, diff --git a/fastn-p2p/src/cli/client.rs b/fastn-p2p/src/cli/client.rs index 0f6c30f..12911dd 100644 --- a/fastn-p2p/src/cli/client.rs +++ b/fastn-p2p/src/cli/client.rs @@ -88,6 +88,92 @@ pub async fn call( Ok(()) } +/// Make a request/response call with args support (issue #13) +pub async fn call_with_args( + fastn_home: PathBuf, + peer_id52: String, + protocol: String, + command: String, + bind_alias: String, + extra_args: Vec, + as_identity: Option, +) -> Result<(), Box> { + // Check if daemon is running + let socket_path = fastn_home.join("control.sock"); + if !socket_path.exists() { + return Err(format!("Daemon not running. Socket not found: {}. Start with: request_response run", socket_path.display()).into()); + } + + // Determine identity to send from + let from_identity = match as_identity { + Some(identity) => identity, + None => { + // TODO: Auto-detect identity if only one configured + "alice".to_string() // Hardcoded for testing + } + }; + + // Parse peer ID to PublicKey for type safety + let to_peer: fastn_id52::PublicKey = peer_id52.parse() + .map_err(|e| format!("Invalid peer ID '{}': {}", peer_id52, e))?; + + // Read JSON request from stdin + let mut stdin_input = String::new(); + io::stdin().read_to_string(&mut stdin_input)?; + let stdin_input = stdin_input.trim(); + + if stdin_input.is_empty() { + return Err("No JSON input provided on stdin".into()); + } + + // Parse JSON to validate it's valid + let request_json: serde_json::Value = serde_json::from_str(stdin_input)?; + + // Enhanced DaemonRequest with command, bind_alias, and args + let daemon_request = fastn_p2p_client::DaemonRequest::CallWithArgs { + from_identity, + to_peer, + protocol, + command, + bind_alias, + args: extra_args, + request: request_json, + }; + + println!("๐Ÿ“ค Sending enhanced P2P request with args support"); + + // Connect to daemon control socket directly + use tokio::net::UnixStream; + use tokio::io::{AsyncWriteExt, AsyncBufReadExt, BufReader}; + + let mut stream = UnixStream::connect(&socket_path).await + .map_err(|e| format!("Failed to connect to daemon: {}", e))?; + + // Send request to daemon + let request_data = serde_json::to_string(&daemon_request)?; + stream.write_all(request_data.as_bytes()).await?; + stream.write_all(b"\n").await?; + + println!("๐Ÿ“ก Enhanced request sent to daemon, reading response..."); + + // Read response from daemon + let (reader, _writer) = stream.into_split(); + let mut buf_reader = BufReader::new(reader); + let mut response_line = String::new(); + + match buf_reader.read_line(&mut response_line).await { + Ok(0) => return Err("Daemon closed connection without response".into()), + Ok(_) => { + let response: serde_json::Value = serde_json::from_str(response_line.trim())?; + println!("๐Ÿ“ฅ Response from daemon:"); + println!("{}", serde_json::to_string_pretty(&response)?); + } + Err(e) => return Err(format!("Failed to read daemon response: {}", e).into()), + } + + Ok(()) +} + /// Open a bidirectional stream to a peer via the daemon pub async fn stream( _fastn_home: PathBuf, diff --git a/fastn-p2p/src/cli/daemon/control.rs b/fastn-p2p/src/cli/daemon/control.rs deleted file mode 100644 index 4c9745e..0000000 --- a/fastn-p2p/src/cli/daemon/control.rs +++ /dev/null @@ -1,330 +0,0 @@ -//! Control socket server for handling client requests -//! -//! This module handles the Unix domain socket that clients connect to. -//! It parses JSON requests and coordinates with the P2P layer. - -use std::path::PathBuf; -use tokio::sync::broadcast; -use tokio::net::UnixListener; -use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; -use serde::{Deserialize, Serialize}; - -use super::{DaemonCommand, DaemonResponse}; - -/// Client request types - precise typing for each operation -#[derive(Debug, Deserialize)] -#[serde(tag = "type")] -pub enum ClientRequest { - #[serde(rename = "call")] - Call { - from_identity: String, - to_peer: fastn_id52::PublicKey, - protocol: String, - bind_alias: String, - request: serde_json::Value, - }, - #[serde(rename = "stream")] - Stream { - from_identity: String, - to_peer: fastn_id52::PublicKey, - protocol: String, - bind_alias: String, - initial_data: serde_json::Value, - }, - #[serde(rename = "reload-identities")] - ReloadIdentities, - #[serde(rename = "set-identity-state")] - SetIdentityState { - identity: String, - online: bool, - }, - #[serde(rename = "add-protocol")] - AddProtocol { - identity: String, - protocol: String, - bind_alias: String, - config: serde_json::Value, - }, - #[serde(rename = "remove-protocol")] - RemoveProtocol { - identity: String, - protocol: String, - bind_alias: String, - }, -} - -/// JSON response format to clients -#[derive(Debug, Serialize)] -struct ClientResponse { - /// Success status: true for ok, false for error - success: bool, - /// Response data or error message - data: serde_json::Value, -} - -/// Run the control socket server -pub async fn run( - fastn_home: PathBuf, - command_tx: broadcast::Sender, - mut response_rx: broadcast::Receiver, -) -> Result<(), Box> { - let socket_path = fastn_home.join("control.sock"); - - // Remove existing socket if it exists - if socket_path.exists() { - tokio::fs::remove_file(&socket_path).await?; - } - - let listener = UnixListener::bind(&socket_path)?; - println!("๐ŸŽง Control socket listening on: {}", socket_path.display()); - - // Start response dispatcher task to handle P2P responses - let _response_task = tokio::spawn(async move { - while let Ok(response) = response_rx.recv().await { - todo!("Route response back to appropriate client connection based on request ID"); - } - }); - - loop { - match listener.accept().await { - Ok((stream, _addr)) => { - let fastn_home_clone = fastn_home.clone(); - tokio::spawn(async move { - if let Err(e) = handle_client(stream, fastn_home_clone).await { - eprintln!("Error handling client: {}", e); - } - }); - } - Err(e) => { - eprintln!("Error accepting connection: {}", e); - } - } - } -} - -async fn handle_client( - stream: tokio::net::UnixStream, - fastn_home: PathBuf, -) -> Result<(), Box> { - println!("๐Ÿ“จ Client connected to control socket"); - - let (reader, writer) = stream.into_split(); - let mut buf_reader = BufReader::new(reader); - let mut line = String::new(); - - // Read the first line to get request header and determine routing - match buf_reader.read_line(&mut line).await { - Ok(0) => { - println!("๐Ÿ“ค Client disconnected immediately"); - return Ok(()); - } - Ok(_) => { - let request_json = line.trim(); - if request_json.is_empty() { - return Ok(()); - } - - println!("๐Ÿ“ฅ Client request: {}", request_json); - - // Parse request header to determine routing strategy - match route_client_request(&fastn_home, request_json, buf_reader, writer).await { - Ok(_) => println!("โœ… Request handled successfully"), - Err(e) => eprintln!("โŒ Request failed: {}", e), - } - } - Err(e) => { - eprintln!("Error reading client request: {}", e); - } - } - - Ok(()) -} - -/// Route client request based on type: P2P (call/stream) or control (daemon management) -async fn route_client_request( - fastn_home: &PathBuf, - request_json: &str, - unix_reader: BufReader, - unix_writer: tokio::net::unix::OwnedWriteHalf, -) -> Result<(), Box> { - // Parse the client request to determine routing - let request: ClientRequest = serde_json::from_str(request_json)?; - - match request { - ClientRequest::Call { from_identity, to_peer, protocol, bind_alias, request } => { - println!("๐Ÿ”€ Routing P2P call: {} {} from {} to {}", - protocol, bind_alias, from_identity, to_peer.id52()); - - // P2P call routing using fastn_net connection pooling - handle_p2p_call(fastn_home.clone(), from_identity, to_peer, protocol, bind_alias, request, unix_writer).await - } - ClientRequest::Stream { from_identity, to_peer, protocol, bind_alias, initial_data } => { - println!("๐Ÿ”€ Routing P2P stream: {} {} from {} to {}", - protocol, bind_alias, from_identity, to_peer.id52()); - - // P2P streaming routing with bidirectional piping - handle_p2p_stream(from_identity, to_peer, protocol, bind_alias, initial_data, unix_reader, unix_writer).await - } - // Control commands (non-P2P) - ClientRequest::ReloadIdentities => { - println!("๐Ÿ”€ Routing control: reload identities"); - handle_control_command("reload-identities", serde_json::Value::Null, unix_writer).await - } - ClientRequest::SetIdentityState { identity, online } => { - println!("๐Ÿ”€ Routing control: set {} {}", identity, if online { "online" } else { "offline" }); - let data = serde_json::json!({ "identity": identity, "online": online }); - handle_control_command("set-identity-state", data, unix_writer).await - } - ClientRequest::AddProtocol { identity, protocol, bind_alias, config } => { - println!("๐Ÿ”€ Routing control: add protocol {} {} to {}", protocol, bind_alias, identity); - let data = serde_json::json!({ "identity": identity, "protocol": protocol, "bind_alias": bind_alias, "config": config }); - handle_control_command("add-protocol", data, unix_writer).await - } - ClientRequest::RemoveProtocol { identity, protocol, bind_alias } => { - println!("๐Ÿ”€ Routing control: remove protocol {} {} from {}", protocol, bind_alias, identity); - let data = serde_json::json!({ "identity": identity, "protocol": protocol, "bind_alias": bind_alias }); - handle_control_command("remove-protocol", data, unix_writer).await - } - } -} - -/// Handle P2P call request - use fastn_net::get_stream() for connection pooling -async fn handle_p2p_call( - fastn_home: PathBuf, - from_identity: String, - to_peer: fastn_id52::PublicKey, - protocol: String, - bind_alias: String, - request: serde_json::Value, - mut unix_writer: tokio::net::unix::OwnedWriteHalf, -) -> Result<(), Box> { - println!("๐Ÿ“ž P2P call: {} {} from {} to {}", protocol, bind_alias, from_identity, to_peer.id52()); - - // Load real identity private key from daemon identity management - let from_key = match load_identity_key(&fastn_home, &from_identity).await { - Ok(key) => { - println!("๐Ÿ”‘ Loaded identity key for: {}", from_identity); - key - } - Err(e) => { - println!("โŒ Failed to load identity '{}': {}", from_identity, e); - let error_response = ClientResponse { - success: false, - data: serde_json::json!({ - "error": format!("Identity '{}' not found or offline: {}", from_identity, e) - }), - }; - let response_json = serde_json::to_string(&error_response)?; - unix_writer.write_all(response_json.as_bytes()).await?; - return Ok(()); - } - }; - - // Create endpoint for this identity - println!("๐Ÿ”Œ Creating P2P endpoint for identity: {}", from_key.public_key().id52()); - let endpoint = fastn_net::get_endpoint(from_key).await?; - - // Create protocol header - for now use a basic protocol like Ping - // TODO: Map daemon protocol strings to fastn_net Protocol enum variants - let protocol_header = fastn_net::ProtocolHeader { - protocol: fastn_net::Protocol::Ping, // Use Ping as placeholder for daemon protocols - extra: Some(format!("{}:{}", protocol, bind_alias)), // Include actual protocol info in extra - }; - - // Use global singletons for connection pooling and graceful shutdown - let pool = fastn_p2p::pool(); - let graceful = fastn_p2p::graceful(); - - println!("๐Ÿ”Œ Getting P2P stream to {} via connection pool", to_peer.id52()); - let (mut p2p_sender, mut p2p_receiver) = fastn_net::get_stream( - endpoint, - protocol_header, - &to_peer, - pool, - graceful - ).await?; - - // Send the request data to P2P - println!("๐Ÿ“ค Sending request to P2P: {}", request); - let request_bytes = serde_json::to_vec(&request)?; - use tokio::io::AsyncWriteExt; - p2p_sender.write_all(&request_bytes).await?; - p2p_sender.finish()?; // Remove .await - it's not async - - // Read response from P2P - let response_bytes = p2p_receiver.read_to_end(1024 * 1024).await?; // 1MB limit - let response_str = String::from_utf8_lossy(&response_bytes); - - println!("๐Ÿ“ฅ Received P2P response: {} bytes", response_bytes.len()); - - // Send response back to Unix socket client - let response = ClientResponse { - success: true, - data: serde_json::json!({ - "p2p_response": response_str, - "protocol": protocol, - "bind_alias": bind_alias, - "from_identity": from_identity - }), - }; - - let response_json = serde_json::to_string(&response)?; - unix_writer.write_all(response_json.as_bytes()).await?; - unix_writer.write_all(b"\n").await?; - - println!("โœ… P2P call completed and response sent to client"); - Ok(()) -} - -/// Handle P2P streaming request - bidirectional piping -async fn handle_p2p_stream( - _from_identity: String, - _to_peer: fastn_id52::PublicKey, - _protocol: String, - _bind_alias: String, - _initial_data: serde_json::Value, - _unix_reader: BufReader, - _unix_writer: tokio::net::unix::OwnedWriteHalf, -) -> Result<(), Box> { - todo!("Use fastn_net::get_stream() for P2P connection, pipe Unix socket โ†” P2P stream bidirectionally"); -} - -/// Handle control commands (daemon management, non-P2P) -async fn handle_control_command( - _command: &str, - _data: serde_json::Value, - _unix_writer: tokio::net::unix::OwnedWriteHalf, -) -> Result<(), Box> { - todo!("Handle daemon management commands: reload identities, add/remove protocols, set online/offline"); -} - -/// Load identity private key from daemon identity management -async fn load_identity_key( - fastn_home: &PathBuf, - identity_name: &str, -) -> Result> { - let identities_dir = fastn_home.join("identities"); - let identity_dir = identities_dir.join(identity_name); - - // Check if identity exists - if !identity_dir.exists() { - return Err(format!("Identity '{}' not found in {}", identity_name, identity_dir.display()).into()); - } - - // Check if identity is online - let online_marker = identity_dir.join("online"); - if !online_marker.exists() { - return Err(format!("Identity '{}' is offline", identity_name).into()); - } - - // Load the identity private key - match fastn_id52::SecretKey::load_from_dir(&identity_dir, "identity") { - Ok((_id52, secret_key)) => { - println!("๐Ÿ”‘ Loaded key for identity '{}': {}", identity_name, secret_key.public_key().id52()); - Ok(secret_key) - } - Err(e) => { - Err(format!("Failed to load key for identity '{}': {}", identity_name, e).into()) - } - } -} \ No newline at end of file diff --git a/fastn-p2p/src/cli/daemon/mod.rs b/fastn-p2p/src/cli/daemon/mod.rs deleted file mode 100644 index b144745..0000000 --- a/fastn-p2p/src/cli/daemon/mod.rs +++ /dev/null @@ -1,313 +0,0 @@ -//! Daemon functionality for fastn-p2p -//! -//! The daemon runs two main services: -//! 1. Control socket server - handles client requests via Unix domain socket -//! 2. P2P listener - handles incoming P2P connections and protocols - -use std::path::PathBuf; -use std::fs::OpenOptions; -use fs2::FileExt; -use tokio::sync::broadcast; - -/// Daemon context containing runtime state and lock -#[derive(Debug)] -pub struct DaemonContext { - pub fastn_home: PathBuf, - pub _lock_file: std::fs::File, // Keep lock file open to maintain exclusive access -} - -/// Coordination channels for daemon services -#[derive(Debug)] -pub struct CoordinationChannels { - pub command_tx: broadcast::Sender, - pub response_tx: broadcast::Sender, -} - -pub mod control; -pub mod p2p; -pub mod protocols; -pub mod test_protocols; -pub mod protocol_trait; - -/// Daemon command for coordinating between control socket and P2P -#[derive(Debug, Clone)] -pub enum DaemonCommand { - /// Make a request/response call to a peer - Call { - peer: fastn_id52::PublicKey, - protocol: String, - request_data: serde_json::Value, - }, - /// Open a stream to a peer - Stream { - peer: fastn_id52::PublicKey, - protocol: String, - initial_data: serde_json::Value, - }, - /// Reload identity configurations from disk - ReloadIdentities, - /// Set an identity online/offline - SetIdentityState { - identity: String, - online: bool, - }, - /// Add a protocol binding to an identity - AddProtocol { - identity: String, - protocol: String, - bind_alias: String, - config: serde_json::Value, - }, - /// Remove a protocol binding from an identity - RemoveProtocol { - identity: String, - protocol: String, - bind_alias: String, - }, - /// Shutdown the daemon - Shutdown, -} - -/// Daemon response back to control socket clients -#[derive(Debug, Clone)] -pub enum DaemonResponse { - /// Successful call response - CallResponse { - response_data: serde_json::Value, - }, - /// Call error - CallError { - error: String, - }, - /// Stream established - StreamReady { - stream_id: u64, - }, - /// Stream error - StreamError { - error: String, - }, - /// Identity configurations reloaded - IdentitiesReloaded { - total: usize, - online: usize, - }, - /// Identity state changed successfully - IdentityStateChanged { - identity: String, - online: bool, - }, - /// Protocol binding added successfully - ProtocolAdded { - identity: String, - protocol: String, - bind_alias: String, - }, - /// Protocol binding removed successfully - ProtocolRemoved { - identity: String, - protocol: String, - bind_alias: String, - }, - /// Operation error - OperationError { - error: String, - }, -} - -/// Run the fastn-p2p daemon with both control socket and P2P listener -pub async fn run(fastn_home: PathBuf) -> Result<(), Box> { - // Initialize daemon environment - let daemon_context = initialize_daemon(&fastn_home).await?; - - // Set up coordination channels - let coordination = setup_coordination_channels().await?; - - // Start P2P networking layer - start_p2p_service(&daemon_context, &coordination).await?; - - // Start control socket service - start_control_service(fastn_home, &coordination).await?; - - // Run main coordination loop - run_coordination_loop(coordination).await?; - - Ok(()) -} - -/// Initialize daemon environment with identity management -async fn initialize_daemon(fastn_home: &PathBuf) -> Result> { - // Use generic server utilities - fastn_p2p::server::ensure_fastn_home(fastn_home).await?; - let lock_file = fastn_p2p::server::acquire_singleton_lock(fastn_home).await?; - - // Load all available identity configurations - let all_identities = fastn_p2p::server::load_all_identities(fastn_home).await?; - - if all_identities.is_empty() { - println!("โš ๏ธ No identities found in {}/identities/", fastn_home.display()); - println!(" Daemon will start and wait for identities to be created"); - println!(" Create an identity with: fastn-p2p create-identity "); - } else { - // Show status of all identities - let online_count = all_identities.iter().filter(|id| id.online).count(); - let total_protocols: usize = all_identities.iter() - .filter(|id| id.online) - .map(|id| id.protocols.len()) - .sum(); - - println!("๐Ÿ”‘ Loaded {} identities ({} online)", all_identities.len(), online_count); - - for identity in &all_identities { - let status_icon = if identity.online { "๐ŸŸข" } else { "๐Ÿ”ด" }; - let status_text = if identity.online { "ONLINE" } else { "OFFLINE" }; - - println!(" {} {} ({}) - {} protocols", - status_icon, - identity.alias, - status_text, - identity.protocols.len()); - } - - if online_count == 0 { - println!("โš ๏ธ No online identities - no P2P services will be started"); - println!(" Enable identities with: fastn-p2p identity-online "); - } else { - println!("โœ… Will start {} P2P services for online identities", total_protocols); - } - } - - Ok(DaemonContext { - fastn_home: fastn_home.clone(), - _lock_file: lock_file, - }) -} - -/// Set up broadcast channels for coordination between services -async fn setup_coordination_channels() -> Result> { - // Create broadcast channels for communication between control socket and P2P services - let (command_tx, _command_rx) = broadcast::channel::(100); - let (response_tx, _response_rx) = broadcast::channel::(100); - - println!("๐Ÿ“ก Created coordination channels"); - println!(" Command channel: 100 message buffer"); - println!(" Response channel: 100 message buffer"); - - Ok(CoordinationChannels { - command_tx, - response_tx, - }) -} - -/// Start the P2P networking service -async fn start_p2p_service( - daemon_context: &DaemonContext, - coordination: &CoordinationChannels, -) -> Result<(), Box> { - // Load current online identities for P2P services - let online_identities: Vec<_> = fastn_p2p::server::load_all_identities(&daemon_context.fastn_home) - .await? - .into_iter() - .filter(|identity| identity.online) - .collect(); - - if online_identities.is_empty() { - println!("๐Ÿ“ก P2P service: No online identities - waiting for activation"); - // Still spawn the P2P task to handle future commands - } else { - let total_protocols: usize = online_identities.iter().map(|id| id.protocols.len()).sum(); - println!("๐Ÿ“ก P2P service: Starting {} protocols for {} online identities", - total_protocols, online_identities.len()); - - for identity in &online_identities { - println!(" ๐ŸŸข {} - {} protocols", identity.alias, identity.protocols.len()); - } - } - - // Spawn P2P service task - let command_rx = coordination.command_tx.subscribe(); - let response_tx = coordination.response_tx.clone(); - let fastn_home = daemon_context.fastn_home.clone(); - - tokio::spawn(async move { - if let Err(e) = p2p::run(fastn_home, command_rx, response_tx).await { - eprintln!("โŒ P2P service error: {}", e); - } - }); - - println!("โœ… P2P service task spawned"); - Ok(()) -} - -/// Start the control socket service -async fn start_control_service( - fastn_home: PathBuf, - coordination: &CoordinationChannels, -) -> Result<(), Box> { - // Spawn control socket server task - let command_tx = coordination.command_tx.clone(); - let response_rx = coordination.response_tx.subscribe(); - - tokio::spawn(async move { - if let Err(e) = control::run(fastn_home, command_tx, response_rx).await { - eprintln!("โŒ Control socket service error: {}", e); - } - }); - - println!("โœ… Control socket service task spawned"); - Ok(()) -} - -/// Run the main coordination loop that handles service lifecycle -async fn run_coordination_loop( - _coordination: CoordinationChannels, -) -> Result<(), Box> { - println!("๐Ÿ”„ Starting main coordination loop"); - println!(" - P2P service: Running in background"); - println!(" - Control socket: Running in background"); - println!(" - Coordination: Active via broadcast channels"); - - // Keep the daemon running - both services are now spawned - // TODO: Handle shutdown signals, coordinate service lifecycle - loop { - tokio::time::sleep(tokio::time::Duration::from_secs(1)).await; - // Services are running in background tasks - // Main loop keeps daemon alive and can handle coordination - } -} - -async fn get_or_create_daemon_key( - fastn_home: &PathBuf, -) -> Result> { - let key_file = fastn_home.join("daemon.key"); - - // Try to load existing key - if key_file.exists() { - if let Ok(key_bytes) = tokio::fs::read(&key_file).await { - if key_bytes.len() == 32 { - let mut bytes_array = [0u8; 32]; - bytes_array.copy_from_slice(&key_bytes); - let secret_key = fastn_id52::SecretKey::from_bytes(&bytes_array); - println!("๐Ÿ”‘ Loaded daemon key from: {}", key_file.display()); - return Ok(secret_key); - } - } - println!("โš ๏ธ Could not load key from {}, generating new one", key_file.display()); - } - - // Generate new key - let key = fastn_id52::SecretKey::generate(); - - // Try to save it - match tokio::fs::write(&key_file, &key.to_secret_bytes()).await { - Ok(_) => { - println!("๐Ÿ”‘ Generated and saved daemon key to: {}", key_file.display()); - } - Err(e) => { - println!("โš ๏ธ Could not save key to {} ({})", key_file.display(), e); - println!(" Using temporary key - daemon ID will change on restart"); - } - } - - Ok(key) -} \ No newline at end of file diff --git a/fastn-p2p/src/cli/daemon/p2p.rs b/fastn-p2p/src/cli/daemon/p2p.rs deleted file mode 100644 index 64f2c00..0000000 --- a/fastn-p2p/src/cli/daemon/p2p.rs +++ /dev/null @@ -1,26 +0,0 @@ -//! P2P networking layer for the daemon -//! -//! This module handles incoming P2P connections using iroh and routes -//! them to appropriate protocol handlers. - -use tokio::sync::broadcast; -use std::collections::HashMap; - -use super::{DaemonCommand, DaemonResponse}; -use super::protocols::{echo, shell}; - -/// P2P listener that handles incoming connections and protocol routing -pub async fn run( - _fastn_home: std::path::PathBuf, - _command_rx: broadcast::Receiver, - _response_tx: broadcast::Sender, -) -> Result<(), Box> { - todo!("Load online identities, initialize P2P endpoints for each identity, handle commands, manage protocol services"); -} - -async fn setup_protocol_handlers( - _daemon_key: fastn_id52::SecretKey, - _response_tx: broadcast::Sender, -) -> Result, Box> { - todo!("Initialize and register protocol handlers (Echo, Shell) with P2P listener"); -} \ No newline at end of file diff --git a/fastn-p2p/src/cli/daemon/protocol_trait.rs b/fastn-p2p/src/cli/daemon/protocol_trait.rs deleted file mode 100644 index c204694..0000000 --- a/fastn-p2p/src/cli/daemon/protocol_trait.rs +++ /dev/null @@ -1,166 +0,0 @@ -//! Protocol trait for standardizing protocol lifecycle management -//! -//! This trait defines the standard interface that all protocols must implement -//! for proper integration with the fastn-p2p daemon. - -use std::path::PathBuf; - -/// Protocol lifecycle management trait -/// -/// All protocols must implement this trait to integrate with the daemon. -/// The trait provides a standardized lifecycle for protocol configuration -/// and service management. -#[async_trait::async_trait] -pub trait Protocol { - /// Protocol name (e.g., "Mail", "Chat", "FileShare") - const NAME: &'static str; - - /// Initialize protocol configuration for first-time setup - /// - /// Creates the protocol's config directory structure and writes default - /// configuration files. This is called when a protocol is first added - /// to an identity. - /// - /// # Parameters - /// * `bind_alias` - The alias for this protocol instance (e.g., "default", "backup") - /// * `config_path` - Directory path where config files should be created - /// - /// # Directory Structure Created - /// ``` - /// config_path/ - /// โ”œโ”€โ”€ config.json # Main protocol configuration - /// โ”œโ”€โ”€ data/ # Protocol-specific data directory (optional) - /// โ””โ”€โ”€ logs/ # Protocol-specific logs (optional) - /// ``` - async fn init( - bind_alias: &str, - config_path: &PathBuf, - ) -> Result<(), Box>; - - /// Load protocol and start P2P services - /// - /// Reads configuration from config_path and starts the protocol's P2P - /// listeners and handlers. This is called when the daemon starts or - /// when an identity comes online. - /// - /// # Parameters - /// * `bind_alias` - The alias for this protocol instance - /// * `config_path` - Directory path containing config files - /// * `identity_key` - The identity's secret key for P2P operations - async fn load( - bind_alias: &str, - config_path: &PathBuf, - identity_key: &fastn_id52::SecretKey, - ) -> Result<(), Box>; - - /// Reload protocol configuration and restart services - /// - /// Re-reads configuration files and restarts P2P services with updated - /// settings. This allows configuration changes without full daemon restart. - /// - /// # Parameters - /// * `bind_alias` - The alias for this protocol instance - /// * `config_path` - Directory path containing updated config files - async fn reload( - bind_alias: &str, - config_path: &PathBuf, - ) -> Result<(), Box>; - - /// Stop protocol services cleanly - /// - /// Performs clean shutdown of all P2P listeners and handlers for this - /// protocol instance. This is called when an identity goes offline or - /// when a protocol is removed. - /// - /// # Parameters - /// * `bind_alias` - The alias for this protocol instance - async fn stop( - bind_alias: &str, - ) -> Result<(), Box>; - - /// Check protocol configuration without affecting runtime - /// - /// Validates configuration files and reports any issues without changing - /// running services. This is useful for configuration validation and - /// troubleshooting. - /// - /// # Parameters - /// * `bind_alias` - The alias for this protocol instance - /// * `config_path` - Directory path containing config files to validate - async fn check( - bind_alias: &str, - config_path: &PathBuf, - ) -> Result<(), Box>; -} - -/// Registry of available protocols -/// -/// This function returns a list of all protocols that can be loaded by the daemon. -/// Production applications would register their own protocols here. -pub fn get_available_protocols() -> Vec<&'static str> { - vec![ - super::protocols::echo::EchoProtocol::NAME, - super::protocols::shell::ShellProtocol::NAME, - ] -} - -/// Load a protocol by name using the trait interface -/// -/// This function dispatches to the appropriate protocol implementation -/// based on the protocol name. -pub async fn load_protocol( - protocol_name: &str, - bind_alias: &str, - config_path: &PathBuf, - identity_key: &fastn_id52::SecretKey, -) -> Result<(), Box> { - match protocol_name { - "Echo" => { - super::protocols::echo::EchoProtocol::load(bind_alias, config_path, identity_key).await - } - "Shell" => { - super::protocols::shell::ShellProtocol::load(bind_alias, config_path, identity_key).await - } - _ => { - Err(format!("Unknown protocol: {}", protocol_name).into()) - } - } -} - -/// Initialize a protocol by name using the trait interface -pub async fn init_protocol( - protocol_name: &str, - bind_alias: &str, - config_path: &PathBuf, -) -> Result<(), Box> { - match protocol_name { - "Echo" => { - super::protocols::echo::EchoProtocol::init(bind_alias, config_path).await - } - "Shell" => { - super::protocols::shell::ShellProtocol::init(bind_alias, config_path).await - } - _ => { - Err(format!("Unknown protocol: {}", protocol_name).into()) - } - } -} - -/// Check a protocol by name using the trait interface -pub async fn check_protocol( - protocol_name: &str, - bind_alias: &str, - config_path: &PathBuf, -) -> Result<(), Box> { - match protocol_name { - "Echo" => { - super::protocols::echo::EchoProtocol::check(bind_alias, config_path).await - } - "Shell" => { - super::protocols::shell::ShellProtocol::check(bind_alias, config_path).await - } - _ => { - Err(format!("Unknown protocol: {}", protocol_name).into()) - } - } -} \ No newline at end of file diff --git a/fastn-p2p/src/cli/daemon/protocols/echo.rs b/fastn-p2p/src/cli/daemon/protocols/echo.rs deleted file mode 100644 index 06ac0bb..0000000 --- a/fastn-p2p/src/cli/daemon/protocols/echo.rs +++ /dev/null @@ -1,96 +0,0 @@ -//! Echo protocol handler -//! -//! Simple request/response protocol that echoes back messages. - -use crate::cli::daemon::test_protocols::{EchoRequest, EchoResponse, EchoError}; -use crate::cli::daemon::protocol_trait::Protocol; - -/// Echo protocol implementation -pub struct EchoProtocol; - -#[async_trait::async_trait] -impl Protocol for EchoProtocol { - const NAME: &'static str = "Echo"; - - async fn init( - bind_alias: &str, - config_path: &std::path::PathBuf, - ) -> Result<(), Box> { - todo!("Create Echo config directory, write default echo config.json, set up Echo workspace for bind_alias: {}", bind_alias); - } - - async fn load( - bind_alias: &str, - config_path: &std::path::PathBuf, - identity_key: &fastn_id52::SecretKey, - ) -> Result<(), Box> { - todo!("Load Echo config from {}, start P2P Echo listener for identity {}, bind_alias: {}", config_path.display(), identity_key.public_key().id52(), bind_alias); - } - - async fn reload( - bind_alias: &str, - config_path: &std::path::PathBuf, - ) -> Result<(), Box> { - todo!("Reload Echo config from {}, restart Echo services for bind_alias: {}", config_path.display(), bind_alias); - } - - async fn stop( - bind_alias: &str, - ) -> Result<(), Box> { - todo!("Stop Echo protocol services for bind_alias: {}", bind_alias); - } - - async fn check( - bind_alias: &str, - config_path: &std::path::PathBuf, - ) -> Result<(), Box> { - todo!("Check Echo config at {} for bind_alias: {} - validate config.json, report issues", config_path.display(), bind_alias); - } -} - -/// Handle Echo protocol requests -pub async fn echo_handler(request: EchoRequest) -> Result { - println!("๐Ÿ“ข Echo request: {}", request.message); - - // Simple validation - if request.message.is_empty() { - return Err(EchoError::InvalidMessage("Message cannot be empty".to_string())); - } - - if request.message.len() > 1000 { - return Err(EchoError::InvalidMessage("Message too long (max 1000 chars)".to_string())); - } - - let response = EchoResponse { - echoed: format!("Echo: {}", request.message), - }; - - println!("๐Ÿ“ค Echo response: {}", response.echoed); - Ok(response) -} - -#[cfg(test)] -mod tests { - use super::*; - - #[tokio::test] - async fn test_echo_handler() { - let request = EchoRequest { - message: "Hello World".to_string(), - }; - - let response = echo_handler(request).await.unwrap(); - assert_eq!(response.echoed, "Echo: Hello World"); - } - - #[tokio::test] - async fn test_echo_handler_empty_message() { - let request = EchoRequest { - message: "".to_string(), - }; - - let result = echo_handler(request).await; - assert!(result.is_err()); - assert!(result.unwrap_err().to_string().contains("empty")); - } -} \ No newline at end of file diff --git a/fastn-p2p/src/cli/daemon/protocols/mod.rs b/fastn-p2p/src/cli/daemon/protocols/mod.rs deleted file mode 100644 index 60e02dd..0000000 --- a/fastn-p2p/src/cli/daemon/protocols/mod.rs +++ /dev/null @@ -1,6 +0,0 @@ -//! Protocol handlers for the daemon -//! -//! Each protocol gets its own module with initialization and handler functions. - -pub mod echo; -pub mod shell; \ No newline at end of file diff --git a/fastn-p2p/src/cli/daemon/protocols/shell.rs b/fastn-p2p/src/cli/daemon/protocols/shell.rs deleted file mode 100644 index 3de679e..0000000 --- a/fastn-p2p/src/cli/daemon/protocols/shell.rs +++ /dev/null @@ -1,182 +0,0 @@ -//! Shell protocol handler -//! -//! Streaming protocol for remote command execution. - -use crate::cli::daemon::protocol_trait::Protocol; - -/// Shell command structure -#[derive(serde::Serialize, serde::Deserialize, Debug)] -pub struct ShellCommand { - pub command: String, - pub args: Vec, -} - -/// Shell response structure -#[derive(serde::Serialize, serde::Deserialize, Debug)] -pub struct ShellResponse { - pub exit_code: i32, - pub stdout: String, - pub stderr: String, -} - -/// Shell error types -#[derive(Debug, thiserror::Error, serde::Serialize, serde::Deserialize)] -pub enum ShellError { - #[error("Command execution failed: {message}")] - ExecutionFailed { message: String }, - #[error("Command not allowed: {command}")] - CommandNotAllowed { command: String }, - #[error("Timeout executing command")] - Timeout, -} - -/// Shell protocol implementation -pub struct ShellProtocol; - -#[async_trait::async_trait] -impl Protocol for ShellProtocol { - const NAME: &'static str = "Shell"; - - async fn init( - bind_alias: &str, - config_path: &std::path::PathBuf, - ) -> Result<(), Box> { - todo!("Create Shell config directory, write default shell config.json with security settings for bind_alias: {}", bind_alias); - } - - async fn load( - bind_alias: &str, - config_path: &std::path::PathBuf, - identity_key: &fastn_id52::SecretKey, - ) -> Result<(), Box> { - todo!("Load Shell config from {}, start P2P Shell streaming listener for identity {}, bind_alias: {}", config_path.display(), identity_key.public_key().id52(), bind_alias); - } - - async fn reload( - bind_alias: &str, - config_path: &std::path::PathBuf, - ) -> Result<(), Box> { - todo!("Reload Shell config from {}, restart Shell services for bind_alias: {}", config_path.display(), bind_alias); - } - - async fn stop( - bind_alias: &str, - ) -> Result<(), Box> { - todo!("Stop Shell protocol services for bind_alias: {}", bind_alias); - } - - async fn check( - bind_alias: &str, - config_path: &std::path::PathBuf, - ) -> Result<(), Box> { - todo!("Check Shell config at {} for bind_alias: {} - validate security settings, allowed commands", config_path.display(), bind_alias); - } -} - -/// Handle Shell protocol streaming sessions -pub async fn shell_stream_handler( - mut _session: fastn_p2p::Session<&'static str>, - command: ShellCommand, - _state: (), -) -> Result<(), ShellError> { - println!("๐Ÿš Shell command requested: {} {:?}", command.command, command.args); - - // Basic security checks - only allow safe commands for testing - let allowed_commands = ["echo", "whoami", "pwd", "ls", "date"]; - if !allowed_commands.contains(&command.command.as_str()) { - return Err(ShellError::CommandNotAllowed { - command: command.command.clone() - }); - } - - // TODO: Execute command and stream output bidirectionally - // For now, simulate command execution - let simulated_output = match command.command.as_str() { - "whoami" => "daemon_user".to_string(), - "pwd" => "/tmp/fastn-daemon".to_string(), - "date" => chrono::Utc::now().to_rfc3339(), - "echo" => command.args.join(" "), - "ls" => "file1.txt\nfile2.txt\ndir1/".to_string(), - _ => "Command output".to_string(), - }; - - println!("๐Ÿ“ค Shell output: {}", simulated_output); - - // TODO: Stream the output back to client via session - // session.copy_from(&mut simulated_output.as_bytes()).await?; - - Ok(()) -} - -/// Execute a shell command safely (for request/response mode) -pub async fn execute_command(command: ShellCommand) -> Result { - println!("โšก Executing shell command: {} {:?}", command.command, command.args); - - // Security check - let allowed_commands = ["echo", "whoami", "pwd", "ls", "date"]; - if !allowed_commands.contains(&command.command.as_str()) { - return Err(ShellError::CommandNotAllowed { - command: command.command.clone() - }); - } - - // For testing, simulate command execution - let (exit_code, stdout) = match command.command.as_str() { - "whoami" => (0, "daemon_user\n".to_string()), - "pwd" => (0, "/tmp/fastn-daemon\n".to_string()), - "date" => (0, format!("{}\n", chrono::Utc::now().to_rfc3339())), - "echo" => (0, format!("{}\n", command.args.join(" "))), - "ls" => (0, "file1.txt\nfile2.txt\ndir1/\n".to_string()), - _ => (1, "".to_string()), - }; - - let response = ShellResponse { - exit_code, - stdout, - stderr: "".to_string(), - }; - - println!("โœ… Shell command completed with exit code: {}", exit_code); - Ok(response) -} - -#[cfg(test)] -mod tests { - use super::*; - - #[tokio::test] - async fn test_whoami_command() { - let command = ShellCommand { - command: "whoami".to_string(), - args: vec![], - }; - - let response = execute_command(command).await.unwrap(); - assert_eq!(response.exit_code, 0); - assert_eq!(response.stdout, "daemon_user\n"); - } - - #[tokio::test] - async fn test_disallowed_command() { - let command = ShellCommand { - command: "rm".to_string(), - args: vec!["-rf".to_string(), "/".to_string()], - }; - - let result = execute_command(command).await; - assert!(result.is_err()); - assert!(matches!(result.unwrap_err(), ShellError::CommandNotAllowed { .. })); - } - - #[tokio::test] - async fn test_echo_command() { - let command = ShellCommand { - command: "echo".to_string(), - args: vec!["Hello".to_string(), "World".to_string()], - }; - - let response = execute_command(command).await.unwrap(); - assert_eq!(response.exit_code, 0); - assert_eq!(response.stdout, "Hello World\n"); - } -} \ No newline at end of file diff --git a/fastn-p2p/src/cli/daemon/test_protocols.rs b/fastn-p2p/src/cli/daemon/test_protocols.rs deleted file mode 100644 index 3fecca6..0000000 --- a/fastn-p2p/src/cli/daemon/test_protocols.rs +++ /dev/null @@ -1,50 +0,0 @@ -//! Test protocols for end-to-end testing -//! -//! These protocols are only used for testing the daemon functionality. -//! Production protocols will be implemented in separate crates. - -use serde::{Deserialize, Serialize}; - -/// Protocol identifiers as const strings -pub const ECHO_PROTOCOL: &str = "Echo"; -pub const SHELL_PROTOCOL: &str = "Shell"; - -/// Echo Protocol Types (moved from request_response example) -#[derive(serde::Serialize, serde::Deserialize, Debug, Clone, PartialEq)] -pub enum EchoProtocol { - Echo, -} - -#[derive(Debug, Serialize, Deserialize)] -pub struct EchoRequest { - pub message: String, -} - -#[derive(Debug, Serialize, Deserialize)] -pub struct EchoResponse { - pub echoed: String, -} - -#[derive(Debug, Serialize, Deserialize, thiserror::Error)] -pub enum EchoError { - #[error("Invalid message: {0}")] - InvalidMessage(String), -} - -pub type EchoResult = Result; - -/// Echo request handler (moved from request_response example) -pub async fn echo_handler(req: EchoRequest) -> Result { - println!("๐Ÿ’ฌ Received: {}", req.message); - - // Basic validation - if req.message.is_empty() { - return Err(EchoError::InvalidMessage("Message cannot be empty".to_string())); - } - - Ok(EchoResponse { - echoed: format!("Echo: {}", req.message), - }) -} - -// TODO: Add Shell protocol for interactive testing \ No newline at end of file diff --git a/fastn-p2p/src/cli/identity.rs b/fastn-p2p/src/cli/identity.rs index 5b933c8..c89e119 100644 --- a/fastn-p2p/src/cli/identity.rs +++ b/fastn-p2p/src/cli/identity.rs @@ -13,6 +13,14 @@ pub async fn create_identity( let identities_dir = fastn_home.join("identities"); tokio::fs::create_dir_all(&identities_dir).await?; + // Create identity-specific directory + let identity_dir = identities_dir.join(&alias); + if identity_dir.exists() { + return Err(format!("Identity '{}' already exists at: {}", alias, identity_dir.display()).into()); + } + + tokio::fs::create_dir_all(&identity_dir).await?; + // Generate new identity let secret_key = fastn_id52::SecretKey::generate(); let public_key = secret_key.public_key(); @@ -20,17 +28,10 @@ pub async fn create_identity( println!("๐Ÿ”‘ Generated new identity: {}", alias); println!(" Peer ID: {}", public_key.id52()); - // Save to identities directory using alias name - let identity_path = identities_dir.join(format!("{}.key", alias)); - - if identity_path.exists() { - return Err(format!("Identity '{}' already exists at: {}", alias, identity_path.display()).into()); - } - - // Use save_to_dir method for proper storage - secret_key.save_to_dir(&identities_dir, &alias)?; + // Save secret key inside identity directory with standard name "identity" + secret_key.save_to_dir(&identity_dir, "identity")?; - println!("๐Ÿ’พ Saved identity to: {}", identity_path.display()); + println!("๐Ÿ’พ Saved identity to: {}", identity_dir.display()); println!("โœ… Identity '{}' created successfully", alias); Ok(()) @@ -55,7 +56,7 @@ pub async fn add_protocol( tokio::fs::create_dir_all(&protocol_config_path).await?; // Load existing identity config - let mut identity_config = fastn_p2p::server::IdentityConfig::load_from_dir(&identities_dir, &identity).await + let identity_config = fastn_p2p::server::IdentityConfig::load_from_dir(&identities_dir, &identity).await .map_err(|e| format!("Identity '{}' not found: {}", identity, e))?; // Check if binding already exists @@ -63,20 +64,16 @@ pub async fn add_protocol( return Err(format!("Protocol binding '{}' as '{}' already exists for identity '{}'", protocol, bind_alias, identity).into()); } - // Initialize the protocol handler using trait interface - if let Err(e) = crate::cli::daemon::protocol_trait::init_protocol(&protocol, &bind_alias, &protocol_config_path).await { - return Err(format!("Failed to initialize {} protocol: {}", protocol, e).into()); - } + // Initialize the protocol handler - just create the directory and config for now + // TODO: Hook into serve_all protocol handlers for proper initialization + tokio::fs::create_dir_all(&protocol_config_path).await?; // Write the initial config JSON to the protocol directory let config_file = protocol_config_path.join(format!("{}.json", protocol.to_lowercase())); tokio::fs::write(&config_file, serde_json::to_string_pretty(&config)?).await?; - // Add protocol binding with config path - identity_config = identity_config.add_protocol(protocol.clone(), bind_alias.clone(), protocol_config_path.clone()); - - // Save updated identity config - identity_config.save_to_dir(&identities_dir).await?; + // Protocol configuration is already saved to the config file above + // The identity directory structure auto-discovers protocols via directory scanning println!("โž• Added protocol binding to identity '{}'", identity); println!(" Protocol: {} as '{}'", protocol, bind_alias); diff --git a/fastn-p2p/src/cli/init.rs b/fastn-p2p/src/cli/init.rs new file mode 100644 index 0000000..77d0344 --- /dev/null +++ b/fastn-p2p/src/cli/init.rs @@ -0,0 +1,21 @@ +//! FASTN_HOME initialization + +use std::path::PathBuf; + +/// Initialize FASTN_HOME directory structure +pub async fn run(fastn_home: PathBuf) -> Result<(), Box> { + // Create basic directory structure + tokio::fs::create_dir_all(&fastn_home).await?; + tokio::fs::create_dir_all(fastn_home.join("identities")).await?; + + println!("โœ… FASTN_HOME initialized: {}", fastn_home.display()); + println!("๐Ÿ“ Directory structure created"); + println!(""); + println!("Next steps:"); + println!(" 1. Create identity: fastn-p2p create-identity alice"); + println!(" 2. Add protocol: fastn-p2p add-protocol alice --protocol echo.fastn.com --config '{{\"max_length\": 1000}}'"); + println!(" 3. Set online: fastn-p2p identity-online alice"); + println!(" 4. Start protocol server: cargo run --bin "); + + Ok(()) +} \ No newline at end of file diff --git a/fastn-p2p/src/cli/mod.rs b/fastn-p2p/src/cli/mod.rs index ef3109a..5805cf3 100644 --- a/fastn-p2p/src/cli/mod.rs +++ b/fastn-p2p/src/cli/mod.rs @@ -3,9 +3,9 @@ use std::path::PathBuf; pub mod client; -pub mod daemon; pub mod identity; pub mod status; +pub mod init; /// Get the FASTN_HOME directory from clap args, environment variable, or default pub fn get_fastn_home(custom_home: Option) -> Result> { diff --git a/fastn-p2p/src/lib.rs b/fastn-p2p/src/lib.rs index 6d6334d..d737e1f 100644 --- a/fastn-p2p/src/lib.rs +++ b/fastn-p2p/src/lib.rs @@ -153,8 +153,14 @@ mod macros; // Export server module (client is now separate fastn-p2p-client crate) pub mod server; +// Export CLI module for magic CLI capabilities in serve_all +pub mod cli; + // Re-export modern server API for convenience -pub use server::{serve_all, echo_request_handler}; +pub use server::serve_all; + +// Re-export serve_all types for convenience +pub use server::serve_all::{BindingContext, RequestContext}; // Re-export essential types from fastn-net that users need pub use fastn_net::{Graceful, Protocol}; diff --git a/fastn-p2p/src/main.rs b/fastn-p2p/src/main.rs index de825ba..18865f9 100644 --- a/fastn-p2p/src/main.rs +++ b/fastn-p2p/src/main.rs @@ -1,7 +1,7 @@ -//! fastn-p2p: P2P daemon and client +//! fastn-p2p: P2P client and identity management //! -//! This binary provides both daemon and client functionality for P2P communication. -//! It uses Unix domain sockets for communication between client and daemon. +//! This binary provides client functionality and identity management. +//! Protocol servers run as separate daemons using serve_all() API. use clap::{Parser, Subcommand}; use std::path::PathBuf; @@ -10,7 +10,7 @@ mod cli; #[derive(Parser)] #[command(name = "fastn-p2p")] -#[command(about = "P2P daemon and client for fastn")] +#[command(about = "P2P client and identity management for fastn")] struct Cli { #[command(subcommand)] command: Commands, @@ -18,8 +18,8 @@ struct Cli { #[derive(Subcommand)] enum Commands { - /// Start the P2P daemon in foreground mode - Daemon { + /// Initialize FASTN_HOME directory structure + Init { /// Custom FASTN_HOME directory (defaults to FASTN_HOME env var or ~/.fastn) #[arg(long, env = "FASTN_HOME")] home: Option, @@ -118,11 +118,10 @@ async fn main() -> Result<(), Box> { let cli = Cli::parse(); match cli.command { - Commands::Daemon { home } => { + Commands::Init { home } => { let fastn_home = cli::get_fastn_home(home)?; - println!("๐Ÿš€ Starting fastn-p2p daemon"); - println!("๐Ÿ“ FASTN_HOME: {}", fastn_home.display()); - cli::daemon::run(fastn_home).await + println!("๐Ÿ”ง Initializing FASTN_HOME: {}", fastn_home.display()); + cli::init::run(fastn_home).await } Commands::Call { peer, protocol, bind_alias, as_identity, home } => { let fastn_home = cli::get_fastn_home(home)?; diff --git a/fastn-p2p/src/server/daemon.rs b/fastn-p2p/src/server/daemon.rs index dea014c..d5fef53 100644 --- a/fastn-p2p/src/server/daemon.rs +++ b/fastn-p2p/src/server/daemon.rs @@ -58,30 +58,31 @@ impl IdentityConfig { self } - /// Save this identity config to the identities directory + /// Save this identity config to conventional directory structure pub async fn save_to_dir(&self, identities_dir: &PathBuf) -> Result<(), Box> { - // Only save secret key if it doesn't exist yet - let key_path = identities_dir.join(format!("{}.private-key", self.alias)); + let identity_dir = identities_dir.join(&self.alias); + tokio::fs::create_dir_all(&identity_dir).await?; + + // Save secret key inside identity directory if it doesn't exist yet + let key_path = identity_dir.join("identity.private-key"); if !key_path.exists() { - self.secret_key.save_to_dir(identities_dir, &self.alias)?; + self.secret_key.save_to_dir(&identity_dir, "identity")?; } - // Always save the configuration (without secret key) - let config_path = identities_dir.join(format!("{}.config.json", self.alias)); - let serializable = IdentityConfigSerialized { - alias: self.alias.clone(), - protocols: self.protocols.clone(), - online: self.online, - }; - let config_json = serde_json::to_string_pretty(&serializable)?; - tokio::fs::write(&config_path, config_json).await?; + // Save online/offline state as marker file + let online_marker = identity_dir.join("online"); + if self.online { + tokio::fs::write(&online_marker, "").await?; + } else if online_marker.exists() { + tokio::fs::remove_file(&online_marker).await?; + } Ok(()) } /// Load identity config from conventional directory structure pub async fn load_from_conventional_dir(identity_dir: &PathBuf, alias: &str) -> Result> { - // Load the secret key + // Load the secret key from inside identity directory let (_id52, secret_key) = fastn_id52::SecretKey::load_from_dir(identity_dir, "identity")?; // Check if identity is online (online file exists) diff --git a/fastn-p2p/src/server/mod.rs b/fastn-p2p/src/server/mod.rs index cf3b2b6..9a0b0b1 100644 --- a/fastn-p2p/src/server/mod.rs +++ b/fastn-p2p/src/server/mod.rs @@ -29,4 +29,4 @@ pub use daemon::{ }; // Modern multi-identity server with callbacks -pub use serve_all::{serve_all, echo_request_handler}; +pub use serve_all::serve_all; diff --git a/fastn-p2p/src/server/serve_all.rs b/fastn-p2p/src/server/serve_all.rs index 6dd674d..4cc4c39 100644 --- a/fastn-p2p/src/server/serve_all.rs +++ b/fastn-p2p/src/server/serve_all.rs @@ -10,25 +10,17 @@ use std::pin::Pin; /// Async callback type for request/response protocol commands pub type RequestCallback = fn( - &str, // identity - &str, // bind_alias - &str, // protocol (e.g., "mail.fastn.com") - &str, // command (e.g., "settings.add-forwarding") - &PathBuf, // protocol_dir + RequestContext, // context (identity, bind_alias, protocol_dir, command, args) serde_json::Value, // request ) -> Pin>> + Send>>; /// Async callback type for streaming protocol commands pub type StreamCallback = fn( - &str, // identity - &str, // bind_alias - &str, // protocol (e.g., "filetransfer.fastn.com") - &str, // command (e.g., "transfer.large-file") - &PathBuf, // protocol_dir + RequestContext, // context (identity, bind_alias, protocol_dir, command, args) serde_json::Value, // initial_data ) -> Pin>> + Send>>; -/// Protocol binding context passed to all handlers +/// Protocol binding context passed to lifecycle handlers #[derive(Debug, Clone)] pub struct BindingContext { pub identity: fastn_id52::PublicKey, @@ -36,6 +28,16 @@ pub struct BindingContext { pub protocol_dir: PathBuf, } +/// Request/stream context passed to protocol command handlers +#[derive(Debug, Clone)] +pub struct RequestContext { + pub identity: fastn_id52::PublicKey, + pub bind_alias: String, + pub protocol_dir: PathBuf, + pub command: String, + pub args: Vec, +} + /// Lifecycle callback types for protocol management (per binding) - clean async fn signatures pub type CreateCallback = fn(BindingContext) -> Pin>> + Send>>; pub type ActivateCallback = fn(BindingContext) -> Pin>> + Send>>; @@ -49,6 +51,7 @@ pub type GlobalLoadCallback = fn(&str) -> Pin Pin>> + Send>>; /// Protocol command handlers for a specific protocol +#[derive(Clone)] pub struct ProtocolBuilder { protocol_name: String, request_callbacks: HashMap, // Key: command name @@ -218,6 +221,46 @@ impl ServeAllBuilder { /// Start serving all configured identities and protocols pub async fn serve(self) -> Result<(), Box> { + // Magic CLI detection - check if args look like CLI commands + let args: Vec = std::env::args().collect(); + if args.len() > 1 { + match args[1].as_str() { + "init" => { + return self.handle_init_command().await; + }, + "call" => { + return self.handle_call_command(args).await; + }, + "stream" => { + return self.handle_stream_command(args).await; + }, + "create-identity" => { + return self.handle_create_identity_command(args).await; + }, + "add-protocol" => { + return self.handle_add_protocol_command(args).await; + }, + "remove-protocol" => { + return self.handle_remove_protocol_command(args).await; + }, + "status" => { + return self.handle_status_command().await; + }, + "identity-online" => { + return self.handle_identity_online_command(args).await; + }, + "identity-offline" => { + return self.handle_identity_offline_command(args).await; + }, + "run" => { + // Continue to server mode + }, + _ => { + // If not recognized as CLI command, continue to server mode + } + } + } + println!("๐Ÿš€ Starting multi-identity P2P server"); println!("๐Ÿ“ FASTN_HOME: {}", self.fastn_home.display()); @@ -247,37 +290,206 @@ impl ServeAllBuilder { protocol_dir.display()); // Check if we have a handler for this protocol - if let Some(callback) = self.request_callbacks.get(&protocol_binding.protocol) { - println!(" ๐Ÿ”„ Starting request handler for {}", protocol_binding.protocol); + if let Some(protocol_builder) = self.protocols.get(&protocol_binding.protocol) { + if !protocol_builder.request_callbacks.is_empty() { + println!(" ๐Ÿ”„ Starting request handlers for {}", protocol_binding.protocol); + + // TODO: Start actual P2P listener and route requests to callbacks + // For now, just log that we would start it + let identity = identity_config.alias.clone(); + let bind_alias = protocol_binding.bind_alias.clone(); + let protocol = protocol_binding.protocol.clone(); + let protocol_dir_clone = protocol_dir.clone(); + + tokio::spawn(async move { + println!("๐ŸŽง Would start P2P listener for {} {} ({})", protocol, bind_alias, identity); + println!(" Working dir: {}", protocol_dir_clone.display()); + // TODO: Start fastn_p2p::listen() and route to callbacks + }); + } - // TODO: Start actual P2P listener and route requests to callback - // For now, just log that we would start it - let identity = identity_config.alias.clone(); - let bind_alias = protocol_binding.bind_alias.clone(); - let protocol = protocol_binding.protocol.clone(); - let protocol_dir_clone = protocol_dir.clone(); - - tokio::spawn(async move { - println!("๐ŸŽง Would start P2P listener for {} {} ({})", protocol, bind_alias, identity); - println!(" Working dir: {}", protocol_dir_clone.display()); - // TODO: Start fastn_p2p::listen() and route to callback - }); - } - - if let Some(callback) = self.stream_callbacks.get(&protocol_binding.protocol) { - println!(" ๐ŸŒŠ Starting stream handler for {}", protocol_binding.protocol); - // TODO: Similar to request handler but for streaming + if !protocol_builder.stream_callbacks.is_empty() { + println!(" ๐ŸŒŠ Starting stream handlers for {}", protocol_binding.protocol); + // TODO: Similar to request handler but for streaming + } } } } - println!("๐ŸŽฏ Multi-identity server ready (TODO: implement actual P2P listening)"); + println!("๐ŸŽฏ Multi-identity server ready"); + println!("๐Ÿ“ก TODO: Implement P2P listeners with enhanced ProtocolHeader.extra routing"); + println!("๐ŸŽง TODO: Create Unix socket daemon interface"); - // Keep server running + // Keep server running for now loop { tokio::time::sleep(tokio::time::Duration::from_secs(1)).await; } } + + // Magic CLI command handlers + async fn handle_init_command(&self) -> Result<(), Box> { + crate::cli::init::run(self.fastn_home.clone()).await + } + + async fn handle_call_command(&self, args: Vec) -> Result<(), Box> { + if args.len() < 4 { + eprintln!("Usage: {} call [command] [bind_alias] [extra_args...] [--as-identity ]", args[0]); + std::process::exit(1); + } + let peer = args[2].clone(); + let protocol = args[3].clone(); + let command = args.get(4).cloned().unwrap_or_else(|| "basic-echo".to_string()); + let bind_alias = args.get(5).cloned().unwrap_or_else(|| "default".to_string()); + + // Collect extra args (issue #13: stdargs support) + let extra_args: Vec = args.iter().skip(6) + .filter(|arg| !arg.starts_with("--")) + .cloned() + .collect(); + + let as_identity = None; // TODO: parse --as-identity flag + + crate::cli::client::call_with_args(self.fastn_home.clone(), peer, protocol, command, bind_alias, extra_args, as_identity).await + } + + async fn handle_stream_command(&self, args: Vec) -> Result<(), Box> { + if args.len() < 4 { + eprintln!("Usage: {} stream ", args[0]); + std::process::exit(1); + } + let peer = args[2].clone(); + let protocol = args[3].clone(); + + crate::cli::client::stream(self.fastn_home.clone(), peer, protocol).await + } + + async fn handle_create_identity_command(&self, args: Vec) -> Result<(), Box> { + if args.len() < 3 { + eprintln!("Usage: {} create-identity ", args[0]); + std::process::exit(1); + } + let alias = args[2].clone(); + + crate::cli::identity::create_identity(self.fastn_home.clone(), alias).await + } + + async fn handle_add_protocol_command(&self, args: Vec) -> Result<(), Box> { + // Parse: add-protocol alice --protocol Echo --alias default --config '{...}' + let mut identity = None; + let mut protocol = None; + let mut alias = "default".to_string(); + let mut config = "{}".to_string(); + + let mut i = 2; + while i < args.len() { + match args[i].as_str() { + "--protocol" => { + if i + 1 < args.len() { + protocol = Some(args[i + 1].clone()); + i += 2; + } else { + eprintln!("--protocol requires a value"); + std::process::exit(1); + } + }, + "--alias" => { + if i + 1 < args.len() { + alias = args[i + 1].clone(); + i += 2; + } else { + eprintln!("--alias requires a value"); + std::process::exit(1); + } + }, + "--config" => { + if i + 1 < args.len() { + config = args[i + 1].clone(); + i += 2; + } else { + eprintln!("--config requires a value"); + std::process::exit(1); + } + }, + _ => { + if identity.is_none() { + identity = Some(args[i].clone()); + } + i += 1; + } + } + } + + let identity = identity.ok_or("Missing identity argument")?; + let protocol = protocol.ok_or("Missing --protocol argument")?; + + crate::cli::identity::add_protocol(self.fastn_home.clone(), identity, protocol, alias, config).await + } + + async fn handle_remove_protocol_command(&self, args: Vec) -> Result<(), Box> { + // Parse similar to add_protocol + let mut identity = None; + let mut protocol = None; + let mut alias = "default".to_string(); + + let mut i = 2; + while i < args.len() { + match args[i].as_str() { + "--protocol" => { + if i + 1 < args.len() { + protocol = Some(args[i + 1].clone()); + i += 2; + } else { + eprintln!("--protocol requires a value"); + std::process::exit(1); + } + }, + "--alias" => { + if i + 1 < args.len() { + alias = args[i + 1].clone(); + i += 2; + } else { + eprintln!("--alias requires a value"); + std::process::exit(1); + } + }, + _ => { + if identity.is_none() { + identity = Some(args[i].clone()); + } + i += 1; + } + } + } + + let identity = identity.ok_or("Missing identity argument")?; + let protocol = protocol.ok_or("Missing --protocol argument")?; + + crate::cli::identity::remove_protocol(self.fastn_home.clone(), identity, protocol, alias).await + } + + async fn handle_status_command(&self) -> Result<(), Box> { + crate::cli::status::show_status(self.fastn_home.clone()).await + } + + async fn handle_identity_online_command(&self, args: Vec) -> Result<(), Box> { + if args.len() < 3 { + eprintln!("Usage: {} identity-online ", args[0]); + std::process::exit(1); + } + let identity = args[2].clone(); + + crate::cli::identity::set_identity_online(self.fastn_home.clone(), identity).await + } + + async fn handle_identity_offline_command(&self, args: Vec) -> Result<(), Box> { + if args.len() < 3 { + eprintln!("Usage: {} identity-offline ", args[0]); + std::process::exit(1); + } + let identity = args[2].clone(); + + crate::cli::identity::set_identity_offline(self.fastn_home.clone(), identity).await + } } /// Create a new multi-identity server builder @@ -295,46 +507,3 @@ pub fn serve_all() -> ServeAllBuilder { } } -/// Echo request handler callback for basic-echo command -pub fn echo_request_handler( - identity: &str, - bind_alias: &str, - protocol: &str, - command: &str, - protocol_dir: &PathBuf, - request: serde_json::Value, -) -> Pin>> + Send>> { - let identity = identity.to_string(); - let bind_alias = bind_alias.to_string(); - let protocol = protocol.to_string(); - let command = command.to_string(); - let protocol_dir = protocol_dir.clone(); - - Box::pin(async move { - println!("๐Ÿ’ฌ Echo handler called:"); - println!(" Identity: {}", identity); - println!(" Bind alias: {}", bind_alias); - println!(" Protocol: {}", protocol); - println!(" Command: {}", command); - println!(" Protocol dir: {}", protocol_dir.display()); - - // Parse request - let message = request.get("message") - .and_then(|v| v.as_str()) - .unwrap_or("(no message)"); - - if message.is_empty() { - return Err("Message cannot be empty".into()); - } - - println!(" Message: '{}'", message); - - // Create response - let response = serde_json::json!({ - "echoed": format!("Echo from {} ({}): {}", identity, command, message) - }); - - println!("๐Ÿ“ค Echo response: {}", response); - Ok(response) - }) -} \ No newline at end of file diff --git a/test-request-response-magic.sh b/test-request-response-magic.sh new file mode 100755 index 0000000..3f51de6 --- /dev/null +++ b/test-request-response-magic.sh @@ -0,0 +1,79 @@ +#!/bin/bash +set -e + +echo "๐Ÿงช Testing Magic CLI functionality with request_response binary" + +# Kill any existing processes +pkill -f "request_response" || true + +# Clean test directories +rm -rf /tmp/test-alice /tmp/test-bob + +echo "๐Ÿ“ Setting up test environments" + +# Initialize Alice +echo "๐Ÿ”ง Initializing Alice..." +FASTN_HOME="/tmp/test-alice" cargo run --bin request_response -- init + +# Initialize Bob +echo "๐Ÿ”ง Initializing Bob..." +FASTN_HOME="/tmp/test-bob" cargo run --bin request_response -- init + +# Create identities and capture peer IDs +echo "๐Ÿ‘ค Creating Alice identity..." +ALICE_OUTPUT=$(FASTN_HOME="/tmp/test-alice" cargo run --bin request_response -- create-identity alice 2>&1) +echo "$ALICE_OUTPUT" +ALICE_ID=$(echo "$ALICE_OUTPUT" | grep "Peer ID:" | cut -d':' -f2 | xargs) + +echo "๐Ÿ‘ค Creating Bob identity..." +BOB_OUTPUT=$(FASTN_HOME="/tmp/test-bob" cargo run --bin request_response -- create-identity bob 2>&1) +echo "$BOB_OUTPUT" +BOB_ID=$(echo "$BOB_OUTPUT" | grep "Peer ID:" | cut -d':' -f2 | xargs) + +# Add echo protocol to both +echo "๐Ÿ“ก Adding echo protocol to Alice..." +FASTN_HOME="/tmp/test-alice" cargo run --bin request_response -- add-protocol alice --protocol echo.fastn.com --config '{"max_length": 1000}' + +echo "๐Ÿ“ก Adding echo protocol to Bob..." +FASTN_HOME="/tmp/test-bob" cargo run --bin request_response -- add-protocol bob --protocol echo.fastn.com --config '{"max_length": 1000}' + +# Set identities online +echo "๐ŸŸข Setting Alice online..." +FASTN_HOME="/tmp/test-alice" cargo run --bin request_response -- identity-online alice + +echo "๐ŸŸข Setting Bob online..." +FASTN_HOME="/tmp/test-bob" cargo run --bin request_response -- identity-online bob + +# Check status +echo "๐Ÿ“Š Alice status:" +FASTN_HOME="/tmp/test-alice" cargo run --bin request_response -- status + +echo "๐Ÿ“Š Bob status:" +FASTN_HOME="/tmp/test-bob" cargo run --bin request_response -- status + +# Start Alice server in background +echo "๐Ÿš€ Starting Alice server..." +FASTN_HOME="/tmp/test-alice" cargo run --bin request_response -- run & +ALICE_PID=$! + +# Start Bob server in background +echo "๐Ÿš€ Starting Bob server..." +FASTN_HOME="/tmp/test-bob" cargo run --bin request_response -- run & +BOB_PID=$! + +echo "โณ Waiting for servers to start..." +sleep 3 + +# Peer IDs already captured from create-identity output above + +echo "๐Ÿ”‘ Alice peer ID: $ALICE_ID" +echo "๐Ÿ”‘ Bob peer ID: $BOB_ID" + +# Try to make a call from Bob to Alice +echo "๐Ÿ“ž Attempting call from Bob to Alice..." +echo '{"message": "Hello Alice from Bob!"}' | FASTN_HOME="/tmp/test-bob" cargo run --bin request_response -- call "$ALICE_ID" echo.fastn.com + +echo "โœ… Magic CLI test complete!" + +# Cleanup +kill $ALICE_PID $BOB_PID 2>/dev/null || true \ No newline at end of file