diff --git a/lattice-cli/src/commands.rs b/lattice-cli/src/commands.rs index 8a927f7..c50769e 100644 --- a/lattice-cli/src/commands.rs +++ b/lattice-cli/src/commands.rs @@ -1,24 +1,21 @@ //! CLI command handlers use lattice_core::{Node, StoreHandle}; -use lattice_core::{Uuid, PeerStatus}; use lattice_net::LatticeEndpoint; -use chrono::DateTime; -use std::time::Instant; -/// Result of a command that may switch stores +/// Result of a command that may switch stores or exit pub enum CommandResult { /// No store change Ok, /// Switch to this store SwitchTo(StoreHandle), + /// Exit the CLI + Quit, } /// Helper to call async code from sync command handlers -fn block_async(f: F) -> F::Output { - tokio::task::block_in_place(|| { - tokio::runtime::Handle::current().block_on(f) - }) +pub fn block_async(f: F) -> F::Output { + tokio::task::block_in_place(|| tokio::runtime::Handle::current().block_on(f)) } pub type Handler = fn(&Node, Option<&StoreHandle>, Option<&LatticeEndpoint>, &[String]) -> CommandResult; @@ -26,758 +23,57 @@ pub type Handler = fn(&Node, Option<&StoreHandle>, Option<&LatticeEndpoint>, &[S pub struct Command { pub name: &'static str, pub args: &'static str, - pub description: &'static str, + pub desc: &'static str, + pub group: &'static str, pub min_args: usize, pub max_args: usize, pub handler: Handler, } +/// Get all available commands pub fn commands() -> Vec { - vec![ - Command { - name: "init", - args: "", - description: "Initialize node with root store", - min_args: 0, - max_args: 0, - handler: cmd_init, - }, - Command { - name: "create-store", - args: "", - description: "Create a new store", - min_args: 0, - max_args: 0, - handler: cmd_create_store, - }, - Command { - name: "use", - args: "", - description: "Switch to a store", - min_args: 1, - max_args: 1, - handler: cmd_use_store, - }, - Command { - name: "list-stores", - args: "", - description: "List all stores", - min_args: 0, - max_args: 0, - handler: cmd_list_stores, - }, - Command { - name: "put", - args: " ", - description: "Store a key-value pair", - min_args: 2, - max_args: 2, - handler: cmd_put, - }, - Command { - name: "get", - args: " [-v]", - description: "Retrieve a value by key", - min_args: 1, - max_args: 2, - handler: cmd_get, - }, - Command { - name: "delete", - args: "", - description: "Delete a key", - min_args: 1, - max_args: 1, - handler: cmd_delete, - }, - Command { - name: "list", - args: "[-v]", - description: "List all key-value pairs (-v for verbose)", - min_args: 0, - max_args: 1, - handler: cmd_list, - }, - Command { - name: "status", - args: "", - description: "Show node/store info", - min_args: 0, - max_args: 0, - handler: cmd_status, - }, - Command { - name: "author-state", - args: "[author-hex]", - description: "Show author state (default: self)", - min_args: 0, - max_args: 1, - handler: cmd_author_state, - }, - Command { - name: "invite", - args: "", - description: "Invite a peer node (writes to root store)", - min_args: 1, - max_args: 1, - handler: cmd_invite, - }, - Command { - name: "peers", - args: "", - description: "List known peers from root store", - min_args: 0, - max_args: 0, - handler: cmd_peers, - }, - Command { - name: "remove", - args: "", - description: "Remove a peer (set status to removed)", - min_args: 1, - max_args: 1, - handler: cmd_remove, - }, - Command { - name: "join", - args: "", - description: "Join a mesh by connecting to a peer (requires no local store)", - min_args: 1, - max_args: 1, - handler: cmd_join, - }, - Command { - name: "sync", - args: "[nodeid]", - description: "Sync entries with a peer (or all peers if none specified)", - min_args: 0, - max_args: 1, - handler: cmd_sync, - }, - Command { - name: "help", - args: "", - description: "Show this help message", - min_args: 0, - max_args: 0, - handler: cmd_help, - }, - ] -} - -// --- Store management --- - -fn cmd_init(node: &Node, _store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, _args: &[String]) -> CommandResult { - match block_async(node.init()) { - Ok((store_id, handle)) => { - println!("Initialized with root store: {}", store_id); - println!("Node info stored in /nodes/{}/*", hex::encode(node.node_id())); - CommandResult::SwitchTo(handle) - } - Err(e) => { - eprintln!("Error: {}", e); - CommandResult::Ok - } - } -} - -fn cmd_create_store(node: &Node, _store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, _args: &[String]) -> CommandResult { - match node.create_store() { - Ok(store_id) => { - println!("Created store: {}", store_id); - match node.open_store(store_id) { - Ok((handle, _)) => { - println!("Switched to new store"); - CommandResult::SwitchTo(handle) - } - Err(e) => { - eprintln!("Warning: {}", e); - CommandResult::Ok - } - } - } - Err(e) => { - eprintln!("Error: {}", e); - CommandResult::Ok - } - } -} - -fn cmd_use_store(node: &Node, _store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { - let store_id = match Uuid::parse_str(&args[0]) { - Ok(id) => id, - Err(_) => { - eprintln!("Error: invalid UUID '{}'", args[0]); - return CommandResult::Ok; - } - }; + let mut cmds = Vec::new(); - let start = Instant::now(); - match node.open_store(store_id) { - Ok((handle, info)) => { - if info.entries_replayed > 0 { - println!("Replayed {} entries ({:.2?})", info.entries_replayed, start.elapsed()); - } else { - println!("Switched to store {}", store_id); - } - CommandResult::SwitchTo(handle) - } - Err(e) => { - eprintln!("Error: {}", e); - CommandResult::Ok - } - } -} - -fn cmd_list_stores(node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, _args: &[String]) -> CommandResult { - let stores = match node.list_stores() { - Ok(s) => s, - Err(e) => { - eprintln!("Error: {}", e); - return CommandResult::Ok; - } - }; - let current_id = store.map(|s| s.id()); + // General CLI commands + cmds.push(Command { + name: "help", args: "", desc: "Show this help", + group: "general", min_args: 0, max_args: 0, handler: cmd_help as Handler + }); + cmds.push(Command { + name: "quit", args: "", desc: "Exit", + group: "general", min_args: 0, max_args: 0, handler: cmd_quit as Handler + }); - if stores.is_empty() { - println!("No stores. Use 'init' or 'create-store'."); - } else { - for store_id in stores { - let marker = if Some(store_id) == current_id { " *" } else { "" }; - println!("{}{}", store_id, marker); - } - } - CommandResult::Ok + // Node commands (operations on the node) + cmds.extend(crate::node_commands::node_commands()); + + // Store commands (raw KV operations) + cmds.extend(crate::store_commands::store_commands()); + + cmds } -// --- Info --- - fn cmd_help(_node: &Node, _store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, _args: &[String]) -> CommandResult { - println!("\nCommands:"); - for cmd in commands() { - if cmd.args.is_empty() { - println!(" {:<16} {}", cmd.name, cmd.description); - } else { - println!(" {} {:<8} {}", cmd.name, cmd.args, cmd.description); + let cmds = commands(); + let mut last_group = ""; + for cmd in &cmds { + if cmd.group != last_group { + println!(); + println!("[{}]", cmd.group); + last_group = cmd.group; } + let usage = if cmd.args.is_empty() { + cmd.name.to_string() + } else { + format!("{} {}", cmd.name, cmd.args) + }; + println!(" {:18} {}", usage, cmd.desc); } - println!(" quit Exit"); println!(); CommandResult::Ok } -fn cmd_status(node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, _args: &[String]) -> CommandResult { - println!("Node ID: {}", hex::encode(node.node_id())); - println!("Data: {}", node.data_path().display()); - match node.root_store() { - Ok(Some(id)) => println!("Root: {}", id), - Ok(None) => println!("Root: (not set)"), - Err(_) => println!("Root: (error)"), - } - if let Some(h) = store { - println!("Store: {}", h.id()); - println!("Log Seq: {}", block_async(h.log_seq())); - println!("Applied: {}", block_async(h.applied_seq()).unwrap_or(0)); - - // Show sync state summary - if let Ok(sync_state) = block_async(h.sync_state()) { - let authors = sync_state.authors(); - let total_entries: u64 = authors.values().map(|a| a.seq).sum(); - let num_authors = authors.len(); - println!("Authors: {} ({} total entries)", num_authors, total_entries); - - // Show per-author details - for (author, info) in authors { - println!(" {}...: seq={}, heads={}", - hex::encode(&author[..6]), - info.seq, - info.heads.len()); - } - } - - // Show log directory size - let logs_dir = node.data_path().join("stores").join(h.id().to_string()).join("logs"); - if logs_dir.exists() { - let mut total_size = 0u64; - let mut file_count = 0; - if let Ok(entries) = std::fs::read_dir(&logs_dir) { - for entry in entries.flatten() { - if let Ok(meta) = entry.metadata() { - if meta.is_file() { - total_size += meta.len(); - file_count += 1; - } - } - } - } - println!("Logs: {} files, {} bytes", file_count, total_size); - } - } else { - println!("Store: (none)"); - } - CommandResult::Ok -} - -// --- KV --- - -fn cmd_put(_node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { - let Some(h) = store else { - println!("No store selected. Use 'init' or 'use '"); - return CommandResult::Ok; - }; - let start = Instant::now(); - match block_async(h.put(args[0].as_bytes(), args[1].as_bytes())) { - Ok(seq) => println!("OK (seq: {}, {:.2?})", seq, start.elapsed()), - Err(e) => eprintln!("Error: {}", e), - } - CommandResult::Ok -} - -fn cmd_get(_node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { - let Some(h) = store else { - println!("No store selected. Use 'init' or 'use '"); - return CommandResult::Ok; - }; - let verbose = args.get(1).map(|a| a == "-v").unwrap_or(false); - let start = Instant::now(); - let key = args[0].as_bytes(); - - if verbose { - // Show all heads - match block_async(h.get_heads(key)) { - Ok(heads) if heads.is_empty() => println!("(nil)"), - Ok(heads) => { - for (i, head) in heads.iter().enumerate() { - let winner = if i == 0 { "→" } else { " " }; - let tombstone = if head.tombstone { "⊗" } else { "" }; - let author_short = hex::encode(&head.author).chars().take(8).collect::(); - if head.tombstone { - println!("{} {} (deleted) (hlc:{}, author:{})", - winner, tombstone, head.hlc, author_short); - } else { - println!("{} {} (hlc:{}, author:{})", - winner, format_value(&head.value), head.hlc, author_short); - } - } - if heads.len() > 1 { - println!("⚠ {} heads (conflict)", heads.len()); - } - println!("({:.2?})", start.elapsed()); - } - Err(e) => eprintln!("Error: {}", e), - } - } else { - match block_async(h.get(key)) { - Ok(Some(v)) => { - let heads = block_async(h.get_heads(key)).unwrap_or_default(); - if heads.len() > 1 { - println!("{} (⚠ {} heads)", format_value(&v), heads.len()); - } else { - println!("{}", format_value(&v)); - } - println!("({:.2?})", start.elapsed()); - } - Ok(None) => println!("(nil)"), - Err(e) => eprintln!("Error: {}", e), - } - } - CommandResult::Ok -} - -fn cmd_delete(_node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { - let Some(h) = store else { - println!("No store selected. Use 'init' or 'use '"); - return CommandResult::Ok; - }; - let start = Instant::now(); - match block_async(h.delete(args[0].as_bytes())) { - Ok(seq) => println!("OK (seq: {}, {:.2?})", seq, start.elapsed()), - Err(e) => eprintln!("Error: {}", e), - } - CommandResult::Ok -} - -fn cmd_list(_node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { - let Some(h) = store else { - println!("No store selected. Use 'init' or 'use '"); - return CommandResult::Ok; - }; - let verbose = args.first().map(|a| a == "-v").unwrap_or(false); - let start = Instant::now(); - match block_async(h.list()) { - Ok(entries) => { - if entries.is_empty() { - println!("(empty)"); - } else { - for (k, v) in &entries { - let key_str = format_value(k); - if verbose { - // Show all heads for this key - let heads = block_async(h.get_heads(k)).unwrap_or_default(); - println!("{}:", key_str); - for (i, head) in heads.iter().enumerate() { - let winner = if i == 0 { "→" } else { " " }; - let author_short = hex::encode(&head.author).chars().take(8).collect::(); - if head.tombstone { - println!(" {} ⊗ (deleted) (hlc:{}, author:{})", - winner, head.hlc, author_short); - } else { - println!(" {} {} (hlc:{}, author:{})", - winner, format_value(&head.value), head.hlc, author_short); - } - } - } else { - // Check for multiple heads - let heads = block_async(h.get_heads(k)).unwrap_or_default(); - if heads.len() > 1 { - println!("{} = {} (⚠ {} heads)", key_str, format_value(v), heads.len()); - } else { - println!("{} = {}", key_str, format_value(v)); - } - } - } - println!("({} keys, {:.2?})", entries.len(), start.elapsed()); - } - } - Err(e) => eprintln!("Error: {}", e), - } - CommandResult::Ok -} - -fn format_value(v: &[u8]) -> String { - std::str::from_utf8(v).map(String::from).unwrap_or_else(|_| format!("0x{}", hex::encode(v))) -} - -fn cmd_author_state(node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { - let store = match store { - Some(s) => s, - None => { - eprintln!("Error: no store selected"); - return CommandResult::Ok; - } - }; - - // Get author: from arg or default to self - let author_bytes: [u8; 32] = if args.is_empty() { - node.node_id() - } else { - let hex_str = args[0].trim_start_matches("0x"); - match hex::decode(hex_str) { - Ok(bytes) if bytes.len() == 32 => bytes.try_into().unwrap(), - Ok(bytes) => { - eprintln!("Error: author must be 32 bytes, got {}", bytes.len()); - return CommandResult::Ok; - } - Err(e) => { - eprintln!("Error: invalid hex: {}", e); - return CommandResult::Ok; - } - } - }; - - match block_async(store.author_state(&author_bytes)) { - Ok(Some(state)) => { - println!("Author: {}", hex::encode(&author_bytes)); - println!(" seq: {}", state.seq); - println!(" hash: {}", hex::encode(&state.hash)); - println!(" log_offset: {}", state.log_offset); - } - Ok(None) => { - println!("No state for author: {}", hex::encode(&author_bytes)); - } - Err(e) => eprintln!("Error: {}", e), - } - CommandResult::Ok -} - -// --- Peer management --- - -fn cmd_invite(node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { - let store = match store { - Some(s) => s, - None => { - eprintln!("Not in a store. Run 'use' or 'init' first."); - return CommandResult::Ok; - } - }; - - let pubkey_hex = &args[0]; - let _peer_pubkey = match hex::decode(pubkey_hex) { - Ok(bytes) if bytes.len() == 32 => bytes, - _ => { - eprintln!("Invalid pubkey: expected 64 hex chars (32 bytes)"); - return CommandResult::Ok; - } - }; - - // Write peer info as separate keys - let inviter_hex = hex::encode(node.node_id()); - let added_at = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .map(|d| d.as_secs()) - .unwrap_or(0); - - let added_by_key = format!("/nodes/{}/added_by", pubkey_hex); - match block_async(store.put(added_by_key.as_bytes(), inviter_hex.as_bytes())) { - Ok(_) => {} - Err(e) => { - eprintln!("Error writing added_by: {}", e); - return CommandResult::Ok; - } - } - - let added_at_key = format!("/nodes/{}/added_at", pubkey_hex); - match block_async(store.put(added_at_key.as_bytes(), added_at.to_string().as_bytes())) { - Ok(_) => {} - Err(e) => { - eprintln!("Error writing added_at: {}", e); - return CommandResult::Ok; - } - } - - // Write /nodes/{pubkey}/status = invited (becomes active after sync) - let status_key = format!("/nodes/{}/status", pubkey_hex); - match block_async(store.put(status_key.as_bytes(), PeerStatus::Invited.as_str().as_bytes())) { - Ok(_) => {} - Err(e) => { - eprintln!("Error writing status: {}", e); - return CommandResult::Ok; - } - } - - println!("Invited peer: {}", pubkey_hex); - println!(" /nodes/{}/added_by", pubkey_hex); - println!(" /nodes/{}/added_at", pubkey_hex); - println!(" /nodes/{}/status = {} (will become active after sync)", pubkey_hex, PeerStatus::Invited.as_str()); - CommandResult::Ok -} - -fn cmd_peers(_node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, _args: &[String]) -> CommandResult { - let store = match store { - Some(s) => s, - None => { - eprintln!("Not in a store. Run 'use' or 'init' first."); - return CommandResult::Ok; - } - }; - - // List all keys under /nodes/ - let all = match block_async(store.list()) { - Ok(v) => v, - Err(e) => { - eprintln!("Error listing: {}", e); - return CommandResult::Ok; - } - }; - - // Collect unique pubkeys with status - let mut peers: std::collections::HashMap = std::collections::HashMap::new(); - for (key, value) in &all { - let key_str = String::from_utf8_lossy(key); - if key_str.ends_with("/status") { - if let Some(pubkey) = key_str.strip_prefix("/nodes/").and_then(|s| s.strip_suffix("/status")) { - let status_str = String::from_utf8_lossy(value); - if let Some(status) = PeerStatus::from_str(&status_str) { - peers.insert(pubkey.to_string(), status); - } - } - } - } - - if peers.is_empty() { - println!("No peers found."); - } else { - // Group peers by status - let mut by_status: std::collections::HashMap> = - std::collections::HashMap::new(); - - for (pubkey, status) in &peers { - // Try to get name and added_at from separate keys - let name_key = format!("/nodes/{}/name", pubkey); - let added_at_key = format!("/nodes/{}/added_at", pubkey); - - let name = match block_async(store.get(name_key.as_bytes())) { - Ok(Some(bytes)) => String::from_utf8_lossy(&bytes).to_string(), - _ => String::new(), - }; - - let added = match block_async(store.get(added_at_key.as_bytes())) { - Ok(Some(bytes)) => { - if let Ok(ts) = String::from_utf8_lossy(&bytes).parse::() { - DateTime::from_timestamp(ts, 0) - .map(|dt| dt.format("%Y-%m-%d").to_string()) - .unwrap_or_default() - } else { - String::new() - } - } - _ => String::new(), - }; - - by_status.entry(*status) - .or_default() - .push((pubkey.clone(), name, added)); - } - - // Print grouped by status in order: active, invited, removed - let status_order = [PeerStatus::Active, PeerStatus::Invited, PeerStatus::Removed]; - for status in &status_order { - if let Some(peer_list) = by_status.get(status) { - println!("\n[{}] ({}):", status.as_str(), peer_list.len()); - let mut sorted = peer_list.clone(); - sorted.sort_by(|a, b| a.0.cmp(&b.0)); - for (pubkey, name, added) in &sorted { - let info_str = match (name.is_empty(), added.is_empty()) { - (false, false) => format!(" {} ({})", name, added), - (false, true) => format!(" {}", name), - (true, false) => format!(" ({})", added), - _ => String::new(), - }; - println!(" {}{}", pubkey, info_str); - } - } - } - } - CommandResult::Ok -} - -fn cmd_remove(node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { - let store = match store { - Some(s) => s, - None => { - eprintln!("Not in a store. Run 'use' or 'init' first."); - return CommandResult::Ok; - } - }; - - let pubkey_hex = &args[0]; - - // Validate pubkey format (should be 64 hex chars) - if pubkey_hex.len() != 64 || !pubkey_hex.chars().all(|c| c.is_ascii_hexdigit()) { - eprintln!("Invalid pubkey: expected 64 hex characters"); - return CommandResult::Ok; - } - - // Prevent self-removal - let my_pubkey = hex::encode(node.node_id()); - if pubkey_hex == &my_pubkey { - eprintln!("Cannot remove yourself."); - return CommandResult::Ok; - } - - // Check if peer exists - let status_key = format!("/nodes/{}/status", pubkey_hex); - match block_async(store.get(status_key.as_bytes())) { - Ok(Some(status)) => { - if status == PeerStatus::Removed.as_str().as_bytes() { - println!("Peer {} is already removed.", &pubkey_hex[..10]); - return CommandResult::Ok; - } - } - Ok(None) => { - eprintln!("Peer {} not found.", &pubkey_hex[..10]); - return CommandResult::Ok; - } - Err(e) => { - eprintln!("Error checking peer: {}", e); - return CommandResult::Ok; - } - } - - // Set status to removed - match block_async(store.put(status_key.as_bytes(), PeerStatus::Removed.as_str().as_bytes())) { - Ok(_) => println!("Removed peer: {}...", &pubkey_hex[..10]), - Err(e) => eprintln!("Error removing peer: {}", e), - } - - CommandResult::Ok -} - -fn cmd_join(node: &Node, store: Option<&StoreHandle>, endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { - let endpoint = match endpoint { - Some(ep) => ep, - None => { - eprintln!("Iroh endpoint not started."); - return CommandResult::Ok; - } - }; - - if store.is_some() { - eprintln!("Already initialized. Use 'sync' to sync with peers."); - return CommandResult::Ok; - } - - let peer_id = match lattice_net::parse_node_id(&args[0]) { - Ok(id) => id, - Err(e) => { - eprintln!("Invalid node ID: {}", e); - return CommandResult::Ok; - } - }; - - println!("Joining mesh via {}...", peer_id.fmt_short()); - - match block_async(lattice_net::join_mesh(node, endpoint, peer_id)) { - Ok(handle) => { - println!("Joined mesh! Use 'sync' command to sync entries."); - CommandResult::SwitchTo(handle) - } - Err(e) => { - eprintln!("Join failed: {}", e); - CommandResult::Ok - } - } -} - -fn cmd_sync(node: &Node, store: Option<&StoreHandle>, endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { - let endpoint = match endpoint { - Some(ep) => ep, - None => { - eprintln!("Iroh endpoint not started."); - return CommandResult::Ok; - } - }; - - let store = match store { - Some(s) => s, - None => { - eprintln!("No store open. Use 'init' or 'join' first."); - return CommandResult::Ok; - } - }; - - if args.is_empty() { - // Sync with all active peers - match block_async(lattice_net::sync_all(node, endpoint, store)) { - Ok(results) => { - if results.is_empty() { - println!("No peers to sync with."); - } else { - let total: u64 = results.iter().map(|r| r.entries_applied).sum(); - println!("\nSync complete! Applied {} entries from {} peer(s).", total, results.len()); - } - } - Err(e) => eprintln!("Sync failed: {}", e), - } - } else { - // Sync with specific peer - let peer_id = match lattice_net::parse_node_id(&args[0]) { - Ok(id) => id, - Err(e) => { - eprintln!("Invalid node ID: {}", e); - return CommandResult::Ok; - } - }; - - println!("Syncing with {}...", peer_id.fmt_short()); - match block_async(lattice_net::sync_with_peer(node, endpoint, store, peer_id)) { - Ok(result) => { - println!("Sync complete! Applied {} entries (peer sent {})", - result.entries_applied, result.entries_sent_by_peer); - } - Err(e) => eprintln!("Sync failed: {}", e), - } - } - - CommandResult::Ok +fn cmd_quit(_node: &Node, _store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, _args: &[String]) -> CommandResult { + println!("Goodbye!"); + CommandResult::Quit } diff --git a/lattice-cli/src/main.rs b/lattice-cli/src/main.rs index 4ef29b0..6deb276 100644 --- a/lattice-cli/src/main.rs +++ b/lattice-cli/src/main.rs @@ -1,6 +1,8 @@ //! Lattice Interactive CLI mod commands; +mod node_commands; +mod store_commands; use lattice_net::spawn_accept_loop; use commands::CommandResult; @@ -52,15 +54,17 @@ async fn main() { } let mut current_store: Option = match node.open_root_store() { - Ok(Some((h, open_info))) => { + Ok(Some(open_info)) => { if open_info.entries_replayed > 0 { println!("Root: {} (replayed {})", open_info.store_id, open_info.entries_replayed); } else { println!("Root: {}", open_info.store_id); } - // Update shared store for accept loop - *shared_store.write().await = Some(h.clone()); - Some(h) + let h = node.root_store().as_ref().cloned(); + if let Some(ref handle) = h { + *shared_store.write().await = Some(handle.clone()); + } + h } Ok(None) => { println!("Status: Not initialized (use 'init')"); @@ -97,13 +101,8 @@ async fn main() { }; let cmd_name = args.first().map(|s| s.as_str()).unwrap_or(""); - - if cmd_name == "quit" || cmd_name == "exit" { - println!("Goodbye!"); - break; - } - match cmds.iter().find(|c| c.name == cmd_name) { + match cmds.iter().find(|c| c.name == cmd_name || (cmd_name == "exit" && c.name == "quit")) { Some(cmd) => { let cmd_args = &args[1..]; if cmd_args.len() < cmd.min_args || cmd_args.len() > cmd.max_args { @@ -116,6 +115,7 @@ async fn main() { *shared_store.write().await = Some(h.clone()); current_store = Some(h); } + CommandResult::Quit => break, } } } diff --git a/lattice-cli/src/node_commands.rs b/lattice-cli/src/node_commands.rs new file mode 100644 index 0000000..c1470ae --- /dev/null +++ b/lattice-cli/src/node_commands.rs @@ -0,0 +1,313 @@ +//! Node commands - operations on the node (mesh, peers, status) + +use crate::commands::{block_async, Command, CommandResult, Handler}; +use lattice_core::{Node, StoreHandle, PeerStatus, Uuid}; +use lattice_net::LatticeEndpoint; +use chrono::DateTime; +use std::time::Instant; + +pub fn node_commands() -> Vec { + vec![ + // Store management + Command { name: "init", args: "", desc: "Initialize root store", group: "node", min_args: 0, max_args: 0, handler: cmd_init as Handler }, + Command { name: "create-store", args: "", desc: "Create a new store", group: "node", min_args: 0, max_args: 0, handler: cmd_create_store as Handler }, + Command { name: "use", args: "", desc: "Switch to a store", group: "node", min_args: 1, max_args: 1, handler: cmd_use_store as Handler }, + Command { name: "list-stores", args: "", desc: "List all stores", group: "node", min_args: 0, max_args: 0, handler: cmd_list_stores as Handler }, + Command { name: "node-status", args: "", desc: "Show node info", group: "node", min_args: 0, max_args: 0, handler: cmd_node_status as Handler }, + // Peer management + Command { name: "invite", args: "", desc: "Invite a peer", group: "peers", min_args: 1, max_args: 1, handler: cmd_invite as Handler }, + Command { name: "peers", args: "", desc: "List all peers", group: "peers", min_args: 0, max_args: 0, handler: cmd_peers as Handler }, + Command { name: "remove", args: "", desc: "Remove a peer", group: "peers", min_args: 1, max_args: 1, handler: cmd_remove as Handler }, + // Networking + Command { name: "join", args: "", desc: "Join an existing mesh", group: "network", min_args: 1, max_args: 1, handler: cmd_join as Handler }, + Command { name: "sync", args: "[node_id]", desc: "Sync with peers", group: "network", min_args: 0, max_args: 1, handler: cmd_sync as Handler }, + ] +} + +// --- Store management --- + +fn cmd_init(node: &Node, _store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, _args: &[String]) -> CommandResult { + match block_async(node.init()) { + Ok(store_id) => { + println!("Initialized with root store: {}", store_id); + println!("Node info stored in /nodes/{}/*", hex::encode(node.node_id())); + match node.root_store().as_ref() { + Some(h) => CommandResult::SwitchTo(h.clone()), + None => CommandResult::Ok, + } + } + Err(e) => { + eprintln!("Error: {}", e); + CommandResult::Ok + } + } +} + +fn cmd_create_store(node: &Node, _store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, _args: &[String]) -> CommandResult { + match node.create_store() { + Ok(store_id) => { + println!("Created store: {}", store_id); + match node.open_store(store_id) { + Ok((handle, _)) => { + println!("Switched to new store"); + CommandResult::SwitchTo(handle) + } + Err(e) => { + eprintln!("Warning: {}", e); + CommandResult::Ok + } + } + } + Err(e) => { + eprintln!("Error: {}", e); + CommandResult::Ok + } + } +} + +fn cmd_use_store(node: &Node, _store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { + let store_id = match Uuid::parse_str(&args[0]) { + Ok(id) => id, + Err(_) => { + eprintln!("Error: invalid UUID '{}'", args[0]); + return CommandResult::Ok; + } + }; + + let start = Instant::now(); + match node.open_store(store_id) { + Ok((handle, info)) => { + if info.entries_replayed > 0 { + println!("Replayed {} entries ({:.2?})", info.entries_replayed, start.elapsed()); + } else { + println!("Switched to store {}", store_id); + } + CommandResult::SwitchTo(handle) + } + Err(e) => { + eprintln!("Error: {}", e); + CommandResult::Ok + } + } +} + +fn cmd_list_stores(node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, _args: &[String]) -> CommandResult { + let stores = match node.list_stores() { + Ok(s) => s, + Err(e) => { + eprintln!("Error: {}", e); + return CommandResult::Ok; + } + }; + let current_id = store.map(|s| s.id()); + + if stores.is_empty() { + println!("No stores. Use 'init' or 'create-store'."); + } else { + for store_id in stores { + let marker = if Some(store_id) == current_id { " *" } else { "" }; + println!("{}{}", store_id, marker); + } + } + CommandResult::Ok +} + +// --- Info --- + +fn cmd_node_status(node: &Node, _store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, _args: &[String]) -> CommandResult { + println!("Node ID: {}", hex::encode(node.node_id())); + if let Some(name) = node.name() { + println!("Name: {}", name); + } + println!("Data: {}", node.data_path().display()); + match node.root_store_id() { + Ok(Some(id)) => println!("Root: {}", id), + Ok(None) => println!("Root: (not set)"), + Err(_) => println!("Root: (error)"), + } + + // Count peers using node.list_peers() + if let Ok(peers) = block_async(node.list_peers()) { + let active = peers.iter().filter(|p| p.status == PeerStatus::Active).count(); + let invited = peers.iter().filter(|p| p.status == PeerStatus::Invited).count(); + println!("Peers: {} active, {} invited", active, invited); + } + + CommandResult::Ok +} + +// --- Peer management --- + +fn cmd_invite(node: &Node, _store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { + let pubkey_hex = &args[0]; + let pubkey: [u8; 32] = match hex::decode(pubkey_hex) { + Ok(bytes) if bytes.len() == 32 => bytes.try_into().unwrap(), + _ => { + eprintln!("Invalid pubkey: expected 64 hex chars (32 bytes)"); + return CommandResult::Ok; + } + }; + + match block_async(node.invite_peer(&pubkey)) { + Ok(()) => { + println!("Invited peer: {}", pubkey_hex); + println!(" Status: {} (will become active after sync)", PeerStatus::Invited.as_str()); + } + Err(e) => eprintln!("Error: {}", e), + } + CommandResult::Ok +} + +fn cmd_peers(node: &Node, _store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, _args: &[String]) -> CommandResult { + let peers = match block_async(node.list_peers()) { + Ok(p) => p, + Err(e) => { + eprintln!("Error: {}", e); + return CommandResult::Ok; + } + }; + + if peers.is_empty() { + println!("No peers found."); + return CommandResult::Ok; + } + + // Group peers by status + let mut by_status: std::collections::HashMap> = + std::collections::HashMap::new(); + for peer in &peers { + by_status.entry(peer.status).or_default().push(peer); + } + + // Print grouped by status in order: active, invited, removed + let status_order = [PeerStatus::Active, PeerStatus::Invited, PeerStatus::Removed]; + for status in &status_order { + if let Some(peer_list) = by_status.get(status) { + println!("\n[{}] ({}):", status.as_str(), peer_list.len()); + let mut sorted: Vec<_> = peer_list.iter().collect(); + sorted.sort_by(|a, b| a.pubkey.cmp(&b.pubkey)); + for peer in sorted { + let added_str = peer.added_at + .and_then(|ts| DateTime::from_timestamp(ts as i64, 0)) + .map(|dt| dt.format("%Y-%m-%d").to_string()) + .unwrap_or_default(); + let info_str = match (peer.name.as_ref(), added_str.is_empty()) { + (Some(name), false) => format!(" {} ({})", name, added_str), + (Some(name), true) => format!(" {}", name), + (None, false) => format!(" ({})", added_str), + (None, true) => String::new(), + }; + println!(" {}{}", peer.pubkey, info_str); + } + } + } + CommandResult::Ok +} + +fn cmd_remove(node: &Node, _store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { + let pubkey_hex = &args[0]; + let pubkey: [u8; 32] = match hex::decode(pubkey_hex) { + Ok(bytes) if bytes.len() == 32 => bytes.try_into().unwrap(), + _ => { + eprintln!("Invalid pubkey: expected 64 hex characters"); + return CommandResult::Ok; + } + }; + + match block_async(node.remove_peer(&pubkey)) { + Ok(()) => println!("Removed peer: {}...", &pubkey_hex[..10]), + Err(e) => eprintln!("Error: {}", e), + } + CommandResult::Ok +} + +// --- Networking --- + +fn cmd_join(node: &Node, store: Option<&StoreHandle>, endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { + let endpoint = match endpoint { + Some(ep) => ep, + None => { + eprintln!("Iroh endpoint not started."); + return CommandResult::Ok; + } + }; + + if store.is_some() { + eprintln!("Already initialized. Use 'sync' to sync with peers."); + return CommandResult::Ok; + } + + let peer_id = match lattice_net::parse_node_id(&args[0]) { + Ok(id) => id, + Err(e) => { + eprintln!("Invalid node ID: {}", e); + return CommandResult::Ok; + } + }; + + println!("Joining mesh via {}...", peer_id.fmt_short()); + + match block_async(lattice_net::join_mesh(node, endpoint, peer_id)) { + Ok(handle) => { + println!("Joined mesh! Use 'sync' command to sync entries."); + CommandResult::SwitchTo(handle) + } + Err(e) => { + eprintln!("Join failed: {}", e); + CommandResult::Ok + } + } +} + +fn cmd_sync(node: &Node, store: Option<&StoreHandle>, endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { + let endpoint = match endpoint { + Some(ep) => ep, + None => { + eprintln!("Iroh endpoint not started."); + return CommandResult::Ok; + } + }; + + let store = match store { + Some(s) => s, + None => { + eprintln!("No store open. Use 'init' or 'join' first."); + return CommandResult::Ok; + } + }; + + if args.is_empty() { + // Sync with all active peers + match block_async(lattice_net::sync_all(node, endpoint, store)) { + Ok(results) => { + if results.is_empty() { + println!("No peers to sync with."); + } else { + let total: u64 = results.iter().map(|r| r.entries_applied).sum(); + println!("\nSync complete! Applied {} entries from {} peer(s).", total, results.len()); + } + } + Err(e) => eprintln!("Sync failed: {}", e), + } + } else { + // Sync with specific peer + let peer_id = match lattice_net::parse_node_id(&args[0]) { + Ok(id) => id, + Err(e) => { + eprintln!("Invalid node ID: {}", e); + return CommandResult::Ok; + } + }; + + println!("Syncing with {}...", peer_id.fmt_short()); + match block_async(lattice_net::sync_with_peer(endpoint, store, peer_id)) { + Ok(result) => { + println!("Sync complete! Applied {} entries (peer sent {})", + result.entries_applied, result.entries_sent_by_peer); + } + Err(e) => eprintln!("Sync failed: {}", e), + } + } + + CommandResult::Ok +} diff --git a/lattice-cli/src/store_commands.rs b/lattice-cli/src/store_commands.rs new file mode 100644 index 0000000..f46e39a --- /dev/null +++ b/lattice-cli/src/store_commands.rs @@ -0,0 +1,209 @@ +//! Store commands - direct KV operations + +use crate::commands::{block_async, Command, CommandResult, Handler}; +use lattice_core::{Node, StoreHandle}; +use lattice_net::LatticeEndpoint; +use std::time::Instant; + +pub fn store_commands() -> Vec { + vec![ + Command { name: "store-status", args: "", desc: "Show store info", group: "store", min_args: 0, max_args: 0, handler: cmd_store_status as Handler }, + Command { name: "put", args: " ", desc: "Store a key-value pair", group: "store", min_args: 2, max_args: 2, handler: cmd_put as Handler }, + Command { name: "get", args: " [-v]", desc: "Get value for key", group: "store", min_args: 1, max_args: 2, handler: cmd_get as Handler }, + Command { name: "delete", args: "", desc: "Delete a key", group: "store", min_args: 1, max_args: 1, handler: cmd_delete as Handler }, + Command { name: "list", args: "[-v]", desc: "List all keys", group: "store", min_args: 0, max_args: 1, handler: cmd_list as Handler }, + Command { name: "author-state", args: "[pubkey]", desc: "Show author sync state", group: "store", min_args: 0, max_args: 1, handler: cmd_author_state as Handler }, + ] +} + +fn cmd_store_status(_node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, _args: &[String]) -> CommandResult { + let Some(h) = store else { + println!("No store selected. Use 'init' or 'use '"); + return CommandResult::Ok; + }; + + println!("Store ID: {}", h.id()); + println!("Log Seq: {}", block_async(h.log_seq())); + println!("Applied: {}", block_async(h.applied_seq()).unwrap_or(0)); + + let all = block_async(h.list()).unwrap_or_default(); + println!("Keys: {}", all.len()); + + // Show log directory size + let (file_count, total_size) = block_async(h.log_stats()); + if file_count > 0 { + println!("Logs: {} files, {} bytes", file_count, total_size); + } + + CommandResult::Ok +} + +fn cmd_put(_node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { + let Some(h) = store else { + println!("No store selected. Use 'init' or 'use '"); + return CommandResult::Ok; + }; + let start = Instant::now(); + match block_async(h.put(args[0].as_bytes(), args[1].as_bytes())) { + Ok(seq) => println!("OK (seq: {}, {:.2?})", seq, start.elapsed()), + Err(e) => eprintln!("Error: {}", e), + } + CommandResult::Ok +} + +fn cmd_get(_node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { + let Some(h) = store else { + println!("No store selected. Use 'init' or 'use '"); + return CommandResult::Ok; + }; + let verbose = args.get(1).map(|a| a == "-v").unwrap_or(false); + let start = Instant::now(); + let key = args[0].as_bytes(); + + if verbose { + // Show all heads + match block_async(h.get_heads(key)) { + Ok(heads) if heads.is_empty() => println!("(nil)"), + Ok(heads) => { + for (i, head) in heads.iter().enumerate() { + let winner = if i == 0 { "→" } else { " " }; + let tombstone = if head.tombstone { "⊗" } else { "" }; + let author_short = hex::encode(&head.author).chars().take(8).collect::(); + if head.tombstone { + println!("{} {} (deleted) (hlc:{}, author:{})", + winner, tombstone, head.hlc, author_short); + } else { + println!("{} {} (hlc:{}, author:{})", + winner, format_value(&head.value), head.hlc, author_short); + } + } + if heads.len() > 1 { + println!("⚠ {} heads (conflict)", heads.len()); + } + println!("({:.2?})", start.elapsed()); + } + Err(e) => eprintln!("Error: {}", e), + } + } else { + match block_async(h.get(key)) { + Ok(Some(v)) => { + let heads = block_async(h.get_heads(key)).unwrap_or_default(); + if heads.len() > 1 { + println!("{} (⚠ {} heads)", format_value(&v), heads.len()); + } else { + println!("{}", format_value(&v)); + } + println!("({:.2?})", start.elapsed()); + } + Ok(None) => println!("(nil)"), + Err(e) => eprintln!("Error: {}", e), + } + } + CommandResult::Ok +} + +fn cmd_delete(_node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { + let Some(h) = store else { + println!("No store selected. Use 'init' or 'use '"); + return CommandResult::Ok; + }; + let start = Instant::now(); + match block_async(h.delete(args[0].as_bytes())) { + Ok(seq) => println!("OK (seq: {}, {:.2?})", seq, start.elapsed()), + Err(e) => eprintln!("Error: {}", e), + } + CommandResult::Ok +} + +fn cmd_list(_node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { + let Some(h) = store else { + println!("No store selected. Use 'init' or 'use '"); + return CommandResult::Ok; + }; + let verbose = args.first().map(|a| a == "-v").unwrap_or(false); + let start = Instant::now(); + match block_async(h.list()) { + Ok(entries) => { + if entries.is_empty() { + println!("(empty)"); + } else { + for (k, v) in &entries { + let key_str = format_value(k); + if verbose { + // Show all heads for this key + let heads = block_async(h.get_heads(k)).unwrap_or_default(); + println!("{}:", key_str); + for (i, head) in heads.iter().enumerate() { + let winner = if i == 0 { "→" } else { " " }; + let author_short = hex::encode(&head.author).chars().take(8).collect::(); + if head.tombstone { + println!(" {} ⊗ (deleted) (hlc:{}, author:{})", + winner, head.hlc, author_short); + } else { + println!(" {} {} (hlc:{}, author:{})", + winner, format_value(&head.value), head.hlc, author_short); + } + } + } else { + // Check for multiple heads + let heads = block_async(h.get_heads(k)).unwrap_or_default(); + if heads.len() > 1 { + println!("{} = {} (⚠ {} heads)", key_str, format_value(v), heads.len()); + } else { + println!("{} = {}", key_str, format_value(v)); + } + } + } + println!("({} keys, {:.2?})", entries.len(), start.elapsed()); + } + } + Err(e) => eprintln!("Error: {}", e), + } + CommandResult::Ok +} + +fn cmd_author_state(node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult { + let store = match store { + Some(s) => s, + None => { + eprintln!("Error: no store selected"); + return CommandResult::Ok; + } + }; + + // Get author: from arg or default to self + let author_bytes: [u8; 32] = if args.is_empty() { + node.node_id() + } else { + let hex_str = args[0].trim_start_matches("0x"); + match hex::decode(hex_str) { + Ok(bytes) if bytes.len() == 32 => bytes.try_into().unwrap(), + Ok(bytes) => { + eprintln!("Error: author must be 32 bytes, got {}", bytes.len()); + return CommandResult::Ok; + } + Err(e) => { + eprintln!("Error: invalid hex: {}", e); + return CommandResult::Ok; + } + } + }; + + match block_async(store.author_state(&author_bytes)) { + Ok(Some(state)) => { + println!("Author: {}", hex::encode(&author_bytes)); + println!(" seq: {}", state.seq); + println!(" hash: {}", hex::encode(&state.hash)); + println!(" log_offset: {}", state.log_offset); + } + Ok(None) => { + println!("No state for author: {}", hex::encode(&author_bytes)); + } + Err(e) => eprintln!("Error: {}", e), + } + CommandResult::Ok +} + +fn format_value(v: &[u8]) -> String { + std::str::from_utf8(v).map(String::from).unwrap_or_else(|_| format!("0x{}", hex::encode(v))) +} diff --git a/lattice-core/src/lib.rs b/lattice-core/src/lib.rs index b960cce..2bd2fb6 100644 --- a/lattice-core/src/lib.rs +++ b/lattice-core/src/lib.rs @@ -35,7 +35,7 @@ pub mod store_actor; pub const MAX_ENTRY_SIZE: usize = 16 * 1024 * 1024; pub use node_identity::{NodeIdentity, PeerStatus}; -pub use node::{Node, NodeBuilder, NodeInfo, StoreInfo, StoreHandle, NodeError}; +pub use node::{Node, NodeBuilder, NodeInfo, StoreInfo, StoreHandle, NodeError, PeerInfo}; pub use sigchain::{SigChain, SigChainManager}; pub use entry::Entry; pub use sync_state::{SyncState, AuthorInfo, MissingRange}; diff --git a/lattice-core/src/node.rs b/lattice-core/src/node.rs index dab82b4..6bc8327 100644 --- a/lattice-core/src/node.rs +++ b/lattice-core/src/node.rs @@ -54,6 +54,15 @@ pub struct StoreInfo { pub entries_replayed: u64, } +/// Information about a peer in the mesh +pub struct PeerInfo { + pub pubkey: String, + pub name: Option, + pub added_at: Option, + pub added_by: Option, + pub status: PeerStatus, +} + pub struct NodeBuilder { pub data_dir: DataDir, } @@ -90,6 +99,7 @@ impl NodeBuilder { data_dir: self.data_dir, node: Rc::new(node), meta, + root_store: std::cell::RefCell::new(None), }) } } @@ -103,6 +113,7 @@ pub struct Node { data_dir: DataDir, node: Rc, meta: MetaStore, + root_store: std::cell::RefCell>, } impl Node { @@ -133,13 +144,13 @@ impl Node { } /// Set the node's display name. - /// Updates meta.db and if a store handle is provided, also updates /nodes/{pubkey}/name - pub async fn set_name(&self, name: &str, store: Option<&StoreHandle>) -> Result<(), NodeError> { + /// Updates meta.db and if root store is open, also updates /nodes/{pubkey}/name + pub async fn set_name(&self, name: &str) -> Result<(), NodeError> { // Update meta.db self.meta.set_name(name)?; - // If store provided, update there too - if let Some(handle) = store { + // If root store is open, update there too + if let Some(handle) = self.root_store.borrow().as_ref() { let pubkey_hex = hex::encode(self.node.public_key_bytes()); let name_key = format!("/nodes/{}/name", pubkey_hex); handle.put(name_key.as_bytes(), name.as_bytes()).await?; @@ -149,20 +160,31 @@ impl Node { } /// Get the root store ID - pub fn root_store(&self) -> Result, NodeError> { + pub fn root_store_id(&self) -> Result, NodeError> { Ok(self.meta.root_store()?) } - /// Open the root store if set - pub fn open_root_store(&self) -> Result, NodeError> { + + /// Get reference to the cached root store handle (if open) + pub fn root_store(&self) -> std::cell::Ref<'_, Option> { + self.root_store.borrow() + } + + /// Open the root store if set. Node owns the handle internally. + /// Returns StoreInfo on success, or None if no root store is set. + pub fn open_root_store(&self) -> Result, NodeError> { match self.meta.root_store()? { - Some(id) => Ok(Some(self.open_store(id)?)), + Some(id) => { + let (handle, info) = self.open_store(id)?; + *self.root_store.borrow_mut() = Some(handle); + Ok(Some(info)) + } None => Ok(None), } } /// Initialize the node with a root store (fails if already initialized). - /// Writes the node's pubkey to `/nodes/{pubkey}/info` in the root store. - pub async fn init(&self) -> Result<(Uuid, StoreHandle), NodeError> { + /// Node owns the store handle internally. Access via root_store(). + pub async fn init(&self) -> Result { if self.meta.root_store()?.is_some() { return Err(NodeError::AlreadyInitialized); } @@ -190,7 +212,140 @@ impl Node { let status_key = format!("/nodes/{}/status", pubkey_hex); handle.put(status_key.as_bytes(), PeerStatus::Active.as_str().as_bytes()).await?; - Ok((store_id, handle)) + // Store the handle - node owns it + *self.root_store.borrow_mut() = Some(handle); + + Ok(store_id) + } + + // --- Peer Management --- + + /// Invite a peer to the mesh. Writes their info with status = invited. + pub async fn invite_peer(&self, pubkey: &[u8; 32]) -> Result<(), NodeError> { + let store = self.root_store.borrow(); + let store = store.as_ref() + .ok_or_else(|| NodeError::Actor("No root store open".to_string()))?; + + let pubkey_hex = hex::encode(pubkey); + let my_pubkey_hex = hex::encode(self.node.public_key_bytes()); + + let added_at = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_secs()) + .unwrap_or(0); + + // Write added_by + let added_by_key = format!("/nodes/{}/added_by", pubkey_hex); + store.put(added_by_key.as_bytes(), my_pubkey_hex.as_bytes()).await?; + + // Write added_at + let added_at_key = format!("/nodes/{}/added_at", pubkey_hex); + store.put(added_at_key.as_bytes(), added_at.to_string().as_bytes()).await?; + + // Write status = invited + let status_key = format!("/nodes/{}/status", pubkey_hex); + store.put(status_key.as_bytes(), PeerStatus::Invited.as_str().as_bytes()).await?; + + Ok(()) + } + + /// List all peers in the mesh with their info + pub async fn list_peers(&self) -> Result, NodeError> { + let store = self.root_store.borrow(); + let store = store.as_ref() + .ok_or_else(|| NodeError::Actor("No root store open".to_string()))?; + + let all = store.list().await?; + + // Collect unique pubkeys with status + let mut peers_map: std::collections::HashMap = std::collections::HashMap::new(); + for (key, value) in &all { + let key_str = String::from_utf8_lossy(key); + if key_str.ends_with("/status") { + if let Some(pubkey) = key_str.strip_prefix("/nodes/").and_then(|s| s.strip_suffix("/status")) { + let status_str = String::from_utf8_lossy(value); + if let Some(status) = PeerStatus::from_str(&status_str) { + peers_map.insert(pubkey.to_string(), status); + } + } + } + } + + // Build PeerInfo for each peer + let mut peers = Vec::new(); + for (pubkey, status) in peers_map { + let name_key = format!("/nodes/{}/name", pubkey); + let added_at_key = format!("/nodes/{}/added_at", pubkey); + let added_by_key = format!("/nodes/{}/added_by", pubkey); + + let name = store.get(name_key.as_bytes()).await? + .map(|b| String::from_utf8_lossy(&b).to_string()); + + let added_at = store.get(added_at_key.as_bytes()).await? + .and_then(|b| String::from_utf8_lossy(&b).parse().ok()); + + let added_by = store.get(added_by_key.as_bytes()).await? + .map(|b| String::from_utf8_lossy(&b).to_string()); + + peers.push(PeerInfo { + pubkey, + name, + added_at, + added_by, + status, + }); + } + + Ok(peers) + } + + /// Remove a peer from the mesh (sets status to removed) + pub async fn remove_peer(&self, pubkey: &[u8; 32]) -> Result<(), NodeError> { + let store = self.root_store.borrow(); + let store = store.as_ref() + .ok_or_else(|| NodeError::Actor("No root store open".to_string()))?; + + let pubkey_hex = hex::encode(pubkey); + + // Prevent self-removal + if pubkey == &self.node.public_key_bytes() { + return Err(NodeError::Actor("Cannot remove yourself".to_string())); + } + + // Check if peer exists + let status_key = format!("/nodes/{}/status", pubkey_hex); + match store.get(status_key.as_bytes()).await? { + Some(status) if status == PeerStatus::Removed.as_str().as_bytes() => { + return Err(NodeError::Actor("Peer already removed".to_string())); + } + None => { + return Err(NodeError::Actor("Peer not found".to_string())); + } + _ => {} + } + + // Set status to removed + store.put(status_key.as_bytes(), PeerStatus::Removed.as_str().as_bytes()).await?; + + Ok(()) + } + + /// Get a peer's status + pub async fn get_peer_status(&self, pubkey: &[u8; 32]) -> Result, NodeError> { + let store = self.root_store.borrow(); + let store = store.as_ref() + .ok_or_else(|| NodeError::Actor("No root store open".to_string()))?; + + let pubkey_hex = hex::encode(pubkey); + let status_key = format!("/nodes/{}/status", pubkey_hex); + + match store.get(status_key.as_bytes()).await? { + Some(bytes) => { + let status_str = String::from_utf8_lossy(&bytes); + Ok(PeerStatus::from_str(&status_str)) + } + None => Ok(None), + } } pub fn list_stores(&self) -> Result, NodeError> { @@ -336,6 +491,14 @@ impl StoreHandle { .map_err(NodeError::Store) } + /// Get log directory statistics (file count, total bytes) + pub async fn log_stats(&self) -> (usize, u64) { + use StoreCmd; + let (resp_tx, resp_rx) = tokio::sync::oneshot::channel(); + let _ = self.tx.send(StoreCmd::LogStats { resp: resp_tx }).await; + resp_rx.await.unwrap_or((0, 0)) + } + pub async fn sync_state(&self) -> Result { use StoreCmd; let (resp_tx, resp_rx) = tokio::sync::oneshot::channel(); @@ -464,11 +627,11 @@ mod tests { .expect("create node"); // Initially no root store - assert!(node.root_store().unwrap().is_none()); + assert!(node.root_store().is_none()); // Init creates root store - let (root_id, _handle) = node.init().await.expect("init failed"); - assert_eq!(node.root_store().unwrap(), Some(root_id)); + let root_id = node.init().await.expect("init failed"); + assert_eq!(node.root_store_id().unwrap(), Some(root_id)); let _ = std::fs::remove_dir_all(data_dir.base()); } @@ -503,7 +666,7 @@ mod tests { let node = NodeBuilder { data_dir: data_dir.clone() } .build() .expect("create node"); - let (root_id, _) = node.init().await.expect("init"); + let root_id = node.init().await.expect("init"); drop(node); // End first session // Second session: root_store should persist @@ -511,7 +674,7 @@ mod tests { .build() .expect("reload node"); - assert_eq!(node.root_store().unwrap(), Some(root_id)); + assert_eq!(node.root_store_id().unwrap(), Some(root_id)); let _ = std::fs::remove_dir_all(data_dir.base()); } @@ -523,7 +686,9 @@ mod tests { let node = NodeBuilder { data_dir: data_dir.clone() } .build() .expect("create node"); - let (_, store) = node.init().await.expect("init"); + node.init().await.expect("init"); + let store = node.root_store(); + let store = store.as_ref().unwrap(); // Get baseline seq after init let baseline = store.log_seq().await; @@ -562,24 +727,32 @@ mod tests { let initial_name = node.name().unwrap(); // Init creates root store - let (_, store) = node.init().await.expect("init"); + node.init().await.expect("init"); // Verify initial name is in store let pubkey_hex = hex::encode(node.node_id()); let name_key = format!("/nodes/{}/name", pubkey_hex); - let stored_name = store.get(name_key.as_bytes()).await.unwrap(); - assert_eq!(stored_name, Some(initial_name.as_bytes().to_vec())); + { + let store = node.root_store(); + let store = store.as_ref().unwrap(); + let stored_name = store.get(name_key.as_bytes()).await.unwrap(); + assert_eq!(stored_name, Some(initial_name.as_bytes().to_vec())); + } // Change name let new_name = "my-custom-name"; - node.set_name(new_name, Some(&store)).await.expect("set_name"); + node.set_name(new_name).await.expect("set_name"); // Verify meta.db updated assert_eq!(node.name(), Some(new_name.to_string())); // Verify store updated - let stored_name = store.get(name_key.as_bytes()).await.unwrap(); - assert_eq!(stored_name, Some(new_name.as_bytes().to_vec())); + { + let store = node.root_store(); + let store = store.as_ref().unwrap(); + let stored_name = store.get(name_key.as_bytes()).await.unwrap(); + assert_eq!(stored_name, Some(new_name.as_bytes().to_vec())); + } let _ = std::fs::remove_dir_all(data_dir.base()); } diff --git a/lattice-core/src/sigchain.rs b/lattice-core/src/sigchain.rs index a2602d1..a6d2d03 100644 --- a/lattice-core/src/sigchain.rs +++ b/lattice-core/src/sigchain.rs @@ -315,6 +315,26 @@ impl SigChainManager { pub fn logs_dir(&self) -> &Path { &self.logs_dir } + + /// Get log directory statistics (file count, total bytes) + pub fn log_stats(&self) -> (usize, u64) { + if !self.logs_dir.exists() { + return (0, 0); + } + let mut total_size = 0u64; + let mut file_count = 0; + if let Ok(entries) = std::fs::read_dir(&self.logs_dir) { + for entry in entries.flatten() { + if let Ok(meta) = entry.metadata() { + if meta.is_file() { + total_size += meta.len(); + file_count += 1; + } + } + } + } + (file_count, total_size) + } } #[cfg(test)] diff --git a/lattice-core/src/store_actor.rs b/lattice-core/src/store_actor.rs index 8e3c06e..7b0bf7d 100644 --- a/lattice-core/src/store_actor.rs +++ b/lattice-core/src/store_actor.rs @@ -58,6 +58,9 @@ pub enum StoreCmd { entry: SignedEntry, resp: oneshot::Sender>, }, + LogStats { + resp: oneshot::Sender<(usize, u64)>, + }, Shutdown, } @@ -185,6 +188,9 @@ impl StoreActor { let result = self.store.apply_entry(&entry); let _ = resp.send(result); } + StoreCmd::LogStats { resp } => { + let _ = resp.send(self.chain_manager.log_stats()); + } StoreCmd::Shutdown => { break; } diff --git a/lattice-net/src/mesh/sync.rs b/lattice-net/src/mesh/sync.rs index 3bea46d..bf2f7e5 100644 --- a/lattice-net/src/mesh/sync.rs +++ b/lattice-net/src/mesh/sync.rs @@ -60,7 +60,7 @@ pub async fn join_mesh( // Immediately sync with the peer to get initial data println!("[Join] Syncing with peer to get initial data..."); - match sync_with_peer(node, endpoint, &handle, peer_id).await { + match sync_with_peer(endpoint, &handle, peer_id).await { Ok(result) => { println!("[Join] Initial sync complete: applied {} entries", result.entries_applied); } @@ -87,7 +87,6 @@ pub async fn join_mesh( /// Sync with a specific peer (bidirectional). /// Both sides exchange states and send missing entries to each other. pub async fn sync_with_peer( - node: &Node, endpoint: &LatticeEndpoint, store: &StoreHandle, peer_id: iroh::PublicKey, @@ -192,7 +191,7 @@ pub async fn sync_all( // Sync with each peer let mut results = Vec::new(); for peer_id in peer_ids { - match sync_with_peer(node, endpoint, store, peer_id).await { + match sync_with_peer(endpoint, store, peer_id).await { Ok(result) => results.push(result), Err(e) => { // Log error but continue with other peers