Restore working code

Cargo.toml

@@ -1,8 +1,8 @@
 [package]
 name = "herodb"
 version = "0.0.1"
-authors = ["Pin Fang <fpfangpin@hotmail.com>"]
+authors = ["ThreeFold Tech NV"]
-edition = "2021"
+edition = "2024"
 
 [dependencies]
 anyhow = "1.0.59"
@@ -24,6 +24,7 @@ age = "0.10"
 secrecy = "0.8"
 ed25519-dalek = "2"
 base64 = "0.22"
+jsonrpsee = { version = "0.26.0", features = ["http-client", "ws-client", "server", "macros"] }
 
 [dev-dependencies]
 redis = { version = "0.24", features = ["aio", "tokio-comp"] }

README.md

@@ -47,13 +47,13 @@ You can start HeroDB with different backends and encryption options:
 #### `redb` with Encryption
 
 ```bash
-./target/release/herodb --dir /tmp/herodb_encrypted --port 6379 --encrypt --key mysecretkey
+./target/release/herodb --dir /tmp/herodb_encrypted --port 6379 --encrypt --encryption_key mysecretkey
 ```
 
 #### `sled` with Encryption
 
 ```bash
-./target/release/herodb --dir /tmp/herodb_sled_encrypted --port 6379 --sled --encrypt --key mysecretkey
+./target/release/herodb --dir /tmp/herodb_sled_encrypted --port 6379 --sled --encrypt --encryption_key mysecretkey
 ```
 
 ## Usage with Redis Clients

src/lib.rs

@@ -4,6 +4,8 @@ pub mod crypto;
 pub mod error;
 pub mod options;
 pub mod protocol;
+pub mod rpc;
+pub mod rpc_server;
 pub mod server;
 pub mod storage;
 pub mod storage_trait;  // Add this

src/main.rs

@@ -3,6 +3,7 @@
 use tokio::net::TcpListener;
 
 use herodb::server;
+use herodb::rpc_server;
 
 use clap::Parser;
 
@@ -31,6 +32,14 @@ struct Args {
     #[arg(long)]
     encrypt: bool,
 
+    /// Enable RPC management server
+    #[arg(long)]
+    enable_rpc: bool,
+
+    /// RPC server port (default: 8080)
+    #[arg(long, default_value = "8080")]
+    rpc_port: u16,
+
     /// Use the sled backend
     #[arg(long)]
     sled: bool,
@@ -50,7 +59,7 @@ async fn main() {
 
     // new DB option
     let option = herodb::options::DBOption {
-        dir: args.dir,
+        dir: args.dir.clone(),
         port,
         debug: args.debug,
         encryption_key: args.encryption_key,
@@ -62,12 +71,36 @@ async fn main() {
         },
     };
 
+    let backend = option.backend.clone();
+
     // new server
-    let server = server::Server::new(option).await;
+    let mut server = server::Server::new(option).await;
+
+    // Initialize the default database storage
+    let _ = server.current_storage();
 
     // Add a small delay to ensure the port is ready
     tokio::time::sleep(std::time::Duration::from_millis(100)).await;
 
+    // Start RPC server if enabled
+    let rpc_handle = if args.enable_rpc {
+        let rpc_addr = format!("127.0.0.1:{}", args.rpc_port).parse().unwrap();
+        let base_dir = args.dir.clone();
+
+        match rpc_server::start_rpc_server(rpc_addr, base_dir, backend).await {
+            Ok(handle) => {
+                println!("RPC management server started on port {}", args.rpc_port);
+                Some(handle)
+            }
+            Err(e) => {
+                eprintln!("Failed to start RPC server: {}", e);
+                None
+            }
+        }
+    } else {
+        None
+    };
+
     // accept new connections
     loop {
         let stream = listener.accept().await;

src/rpc.rs

@@ -0,0 +1,342 @@
+use std::collections::HashMap;
+use std::sync::Arc;
+use tokio::sync::RwLock;
+use jsonrpsee::{core::RpcResult, proc_macros::rpc};
+use serde::{Deserialize, Serialize};
+
+use crate::server::Server;
+use crate::options::DBOption;
+
+/// Database backend types
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub enum BackendType {
+    Redb,
+    Sled,
+    // Future: InMemory, Custom(String)
+}
+
+/// Database configuration
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct DatabaseConfig {
+    pub name: Option<String>,
+    pub storage_path: Option<String>,
+    pub max_size: Option<u64>,
+    pub redis_version: Option<String>,
+}
+
+/// Database information returned by metadata queries
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct DatabaseInfo {
+    pub id: u64,
+    pub name: Option<String>,
+    pub backend: BackendType,
+    pub encrypted: bool,
+    pub redis_version: Option<String>,
+    pub storage_path: Option<String>,
+    pub size_on_disk: Option<u64>,
+    pub key_count: Option<u64>,
+    pub created_at: u64,
+    pub last_access: Option<u64>,
+}
+
+/// RPC trait for HeroDB management
+#[rpc(server, client, namespace = "herodb")]
+pub trait Rpc {
+    /// Create a new database with specified configuration
+    #[method(name = "createDatabase")]
+    async fn create_database(
+        &self,
+        backend: BackendType,
+        config: DatabaseConfig,
+        encryption_key: Option<String>,
+    ) -> RpcResult<u64>;
+
+    /// Set encryption for an existing database (write-only key)
+    #[method(name = "setEncryption")]
+    async fn set_encryption(&self, db_id: u64, encryption_key: String) -> RpcResult<bool>;
+
+    /// List all managed databases
+    #[method(name = "listDatabases")]
+    async fn list_databases(&self) -> RpcResult<Vec<DatabaseInfo>>;
+
+    /// Get detailed information about a specific database
+    #[method(name = "getDatabaseInfo")]
+    async fn get_database_info(&self, db_id: u64) -> RpcResult<DatabaseInfo>;
+
+    /// Delete a database
+    #[method(name = "deleteDatabase")]
+    async fn delete_database(&self, db_id: u64) -> RpcResult<bool>;
+
+    /// Get server statistics
+    #[method(name = "getServerStats")]
+    async fn get_server_stats(&self) -> RpcResult<HashMap<String, serde_json::Value>>;
+}
+
+/// RPC Server implementation
+pub struct RpcServerImpl {
+    /// Base directory for database files
+    base_dir: String,
+    /// Managed database servers
+    servers: Arc<RwLock<HashMap<u64, Arc<Server>>>>,
+    /// Next unencrypted database ID to assign
+    next_unencrypted_id: Arc<RwLock<u64>>,
+    /// Next encrypted database ID to assign
+    next_encrypted_id: Arc<RwLock<u64>>,
+    /// Default backend type
+    backend: crate::options::BackendType,
+}
+
+impl RpcServerImpl {
+    /// Create a new RPC server instance
+    pub fn new(base_dir: String, backend: crate::options::BackendType) -> Self {
+        Self {
+            base_dir,
+            servers: Arc::new(RwLock::new(HashMap::new())),
+            next_unencrypted_id: Arc::new(RwLock::new(0)),
+            next_encrypted_id: Arc::new(RwLock::new(10)),
+            backend,
+        }
+    }
+
+    /// Get or create a server instance for the given database ID
+    async fn get_or_create_server(&self, db_id: u64) -> Result<Arc<Server>, jsonrpsee::types::ErrorObjectOwned> {
+        // Check if server already exists
+        {
+            let servers = self.servers.read().await;
+            if let Some(server) = servers.get(&db_id) {
+                return Ok(server.clone());
+            }
+        }
+
+        // Check if database file exists
+        let db_path = std::path::PathBuf::from(&self.base_dir).join(format!("{}.db", db_id));
+        if !db_path.exists() {
+            return Err(jsonrpsee::types::ErrorObjectOwned::owned(
+                -32000,
+                format!("Database {} not found", db_id),
+                None::<()>
+            ));
+        }
+
+        // Create server instance with default options
+        let db_option = DBOption {
+            dir: self.base_dir.clone(),
+            port: 0, // Not used for RPC-managed databases
+            debug: false,
+            encryption_key: None,
+            encrypt: false,
+            backend: self.backend.clone(),
+        };
+
+        let mut server = Server::new(db_option).await;
+
+        // Set the selected database to the db_id for proper file naming
+        server.selected_db = db_id;
+
+        // Store the server
+        let mut servers = self.servers.write().await;
+        servers.insert(db_id, Arc::new(server.clone()));
+
+        Ok(Arc::new(server))
+    }
+
+    /// Discover existing database files in the base directory
+    async fn discover_databases(&self) -> Vec<u64> {
+        let mut db_ids = Vec::new();
+
+        if let Ok(entries) = std::fs::read_dir(&self.base_dir) {
+            for entry in entries.flatten() {
+                if let Ok(file_name) = entry.file_name().into_string() {
+                    // Check if it's a database file (ends with .db)
+                    if file_name.ends_with(".db") {
+                        // Extract database ID from filename (e.g., "11.db" -> 11)
+                        if let Some(id_str) = file_name.strip_suffix(".db") {
+                            if let Ok(db_id) = id_str.parse::<u64>() {
+                                db_ids.push(db_id);
+                            }
+                        }
+                    }
+                }
+            }
+        }
+
+        db_ids
+    }
+
+    /// Get the next available database ID
+    async fn get_next_db_id(&self, is_encrypted: bool) -> u64 {
+        if is_encrypted {
+            let mut id = self.next_encrypted_id.write().await;
+            let current_id = *id;
+            *id += 1;
+            current_id
+        } else {
+            let mut id = self.next_unencrypted_id.write().await;
+            let current_id = *id;
+            *id += 1;
+            current_id
+        }
+    }
+}
+
+#[jsonrpsee::core::async_trait]
+impl RpcServer for RpcServerImpl {
+    async fn create_database(
+        &self,
+        backend: BackendType,
+        config: DatabaseConfig,
+        encryption_key: Option<String>,
+    ) -> RpcResult<u64> {
+        let db_id = self.get_next_db_id(encryption_key.is_some()).await;
+
+        // Handle both Redb and Sled backends
+        match backend {
+            BackendType::Redb | BackendType::Sled => {
+                // Create database directory
+                let db_dir = if let Some(path) = &config.storage_path {
+                    std::path::PathBuf::from(path)
+                } else {
+                    std::path::PathBuf::from(&self.base_dir).join(format!("rpc_db_{}", db_id))
+                };
+
+                // Ensure directory exists
+                std::fs::create_dir_all(&db_dir)
+                    .map_err(|e| jsonrpsee::types::ErrorObjectOwned::owned(
+                        -32000,
+                        format!("Failed to create directory: {}", e),
+                        None::<()>
+                    ))?;
+
+                // Create DB options
+                let encrypt = encryption_key.is_some();
+                let option = DBOption {
+                    dir: db_dir.to_string_lossy().to_string(),
+                    port: 0, // Not used for RPC-managed databases
+                    debug: false,
+                    encryption_key,
+                    encrypt,
+                    backend: match backend {
+                        BackendType::Redb => crate::options::BackendType::Redb,
+                        BackendType::Sled => crate::options::BackendType::Sled,
+                    },
+                };
+
+                // Create server instance
+                let mut server = Server::new(option).await;
+
+                // Set the selected database to the db_id for proper file naming
+                server.selected_db = db_id;
+
+                // Initialize the storage to create the database file
+                let _ = server.current_storage();
+
+                // Store the server
+                let mut servers = self.servers.write().await;
+                servers.insert(db_id, Arc::new(server));
+
+                Ok(db_id)
+            }
+        }
+    }
+
+    async fn set_encryption(&self, db_id: u64, _encryption_key: String) -> RpcResult<bool> {
+        // Note: In a real implementation, we'd need to modify the existing database
+        // For now, return false as encryption can only be set during creation
+        let _servers = self.servers.read().await;
+        // TODO: Implement encryption setting for existing databases
+        Ok(false)
+    }
+
+    async fn list_databases(&self) -> RpcResult<Vec<DatabaseInfo>> {
+        let db_ids = self.discover_databases().await;
+        let mut result = Vec::new();
+
+        for db_id in db_ids {
+            // Try to get or create server for this database
+            if let Ok(server) = self.get_or_create_server(db_id).await {
+                let backend = match server.option.backend {
+                    crate::options::BackendType::Redb => BackendType::Redb,
+                    crate::options::BackendType::Sled => BackendType::Sled,
+                };
+
+                let info = DatabaseInfo {
+                    id: db_id,
+                    name: None, // TODO: Store name in server metadata
+                    backend,
+                    encrypted: server.option.encrypt,
+                    redis_version: Some("7.0".to_string()), // Default Redis compatibility
+                    storage_path: Some(server.option.dir.clone()),
+                    size_on_disk: None, // TODO: Calculate actual size
+                    key_count: None, // TODO: Get key count from storage
+                    created_at: std::time::SystemTime::now()
+                        .duration_since(std::time::UNIX_EPOCH)
+                        .unwrap()
+                        .as_secs(),
+                    last_access: None,
+                };
+                result.push(info);
+            }
+        }
+
+        Ok(result)
+    }
+
+    async fn get_database_info(&self, db_id: u64) -> RpcResult<DatabaseInfo> {
+        let server = self.get_or_create_server(db_id).await?;
+
+        let backend = match server.option.backend {
+            crate::options::BackendType::Redb => BackendType::Redb,
+            crate::options::BackendType::Sled => BackendType::Sled,
+        };
+
+        Ok(DatabaseInfo {
+            id: db_id,
+            name: None,
+            backend,
+            encrypted: server.option.encrypt,
+            redis_version: Some("7.0".to_string()),
+            storage_path: Some(server.option.dir.clone()),
+            size_on_disk: None,
+            key_count: None,
+            created_at: std::time::SystemTime::now()
+                .duration_since(std::time::UNIX_EPOCH)
+                .unwrap()
+                .as_secs(),
+            last_access: None,
+        })
+    }
+
+    async fn delete_database(&self, db_id: u64) -> RpcResult<bool> {
+        let mut servers = self.servers.write().await;
+
+        if let Some(_server) = servers.remove(&db_id) {
+            // Clean up database files
+            let db_path = std::path::PathBuf::from(&self.base_dir).join(format!("{}.db", db_id));
+            if db_path.exists() {
+                if db_path.is_dir() {
+                    std::fs::remove_dir_all(&db_path).ok();
+                } else {
+                    std::fs::remove_file(&db_path).ok();
+                }
+            }
+            Ok(true)
+        } else {
+            Ok(false)
+        }
+    }
+
+    async fn get_server_stats(&self) -> RpcResult<HashMap<String, serde_json::Value>> {
+        let db_ids = self.discover_databases().await;
+        let mut stats = HashMap::new();
+
+        stats.insert("total_databases".to_string(), serde_json::json!(db_ids.len()));
+        stats.insert("uptime".to_string(), serde_json::json!(
+            std::time::SystemTime::now()
+                .duration_since(std::time::UNIX_EPOCH)
+                .unwrap()
+                .as_secs()
+        ));
+
+        Ok(stats)
+    }
+}

src/rpc_server.rs

@@ -0,0 +1,49 @@
+use std::net::SocketAddr;
+use jsonrpsee::server::{ServerBuilder, ServerHandle};
+use jsonrpsee::RpcModule;
+
+use crate::rpc::{RpcServer, RpcServerImpl};
+
+/// Start the RPC server on the specified address
+pub async fn start_rpc_server(addr: SocketAddr, base_dir: String, backend: crate::options::BackendType) -> Result<ServerHandle, Box<dyn std::error::Error + Send + Sync>> {
+    // Create the RPC server implementation
+    let rpc_impl = RpcServerImpl::new(base_dir, backend);
+
+    // Create the RPC module
+    let mut module = RpcModule::new(());
+    module.merge(RpcServer::into_rpc(rpc_impl))?;
+
+    // Build the server with both HTTP and WebSocket support
+    let server = ServerBuilder::default()
+        .build(addr)
+        .await?;
+
+    // Start the server
+    let handle = server.start(module);
+
+    println!("RPC server started on {}", addr);
+
+    Ok(handle)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::time::Duration;
+
+    #[tokio::test]
+    async fn test_rpc_server_startup() {
+        let addr = "127.0.0.1:0".parse().unwrap(); // Use port 0 for auto-assignment
+        let base_dir = "/tmp/test_rpc".to_string();
+        let backend = crate::options::BackendType::Redb; // Default for test
+
+        let handle = start_rpc_server(addr, base_dir, backend).await.unwrap();
+
+        // Give the server a moment to start
+        tokio::time::sleep(Duration::from_millis(100)).await;
+
+        // Stop the server
+        handle.stop().unwrap();
+        handle.stopped().await;
+    }
+}

tests/rpc_tests.rs

@@ -0,0 +1,62 @@
+use std::net::SocketAddr;
+use jsonrpsee::http_client::HttpClientBuilder;
+use jsonrpsee::core::client::ClientT;
+use serde_json::json;
+
+use herodb::rpc::{RpcClient, BackendType, DatabaseConfig};
+
+#[tokio::test]
+async fn test_rpc_server_basic() {
+    // This test would require starting the RPC server in a separate thread
+    // For now, we'll just test that the types compile correctly
+
+    // Test serialization of types
+    let backend = BackendType::Redb;
+    let config = DatabaseConfig {
+        name: Some("test_db".to_string()),
+        storage_path: Some("/tmp/test".to_string()),
+        max_size: Some(1024 * 1024),
+        redis_version: Some("7.0".to_string()),
+    };
+
+    let backend_json = serde_json::to_string(&backend).unwrap();
+    let config_json = serde_json::to_string(&config).unwrap();
+
+    assert_eq!(backend_json, "\"Redb\"");
+    assert!(config_json.contains("test_db"));
+}
+
+#[tokio::test]
+async fn test_database_config_serialization() {
+    let config = DatabaseConfig {
+        name: Some("my_db".to_string()),
+        storage_path: None,
+        max_size: Some(1000000),
+        redis_version: Some("7.0".to_string()),
+    };
+
+    let json = serde_json::to_value(&config).unwrap();
+    assert_eq!(json["name"], "my_db");
+    assert_eq!(json["max_size"], 1000000);
+    assert_eq!(json["redis_version"], "7.0");
+}
+
+#[tokio::test]
+async fn test_backend_type_serialization() {
+    // Test that both Redb and Sled backends serialize correctly
+    let redb_backend = BackendType::Redb;
+    let sled_backend = BackendType::Sled;
+
+    let redb_json = serde_json::to_string(&redb_backend).unwrap();
+    let sled_json = serde_json::to_string(&sled_backend).unwrap();
+
+    assert_eq!(redb_json, "\"Redb\"");
+    assert_eq!(sled_json, "\"Sled\"");
+
+    // Test deserialization
+    let redb_deserialized: BackendType = serde_json::from_str(&redb_json).unwrap();
+    let sled_deserialized: BackendType = serde_json::from_str(&sled_json).unwrap();
+
+    assert!(matches!(redb_deserialized, BackendType::Redb));
+    assert!(matches!(sled_deserialized, BackendType::Sled));
+}

tests/usage_suite.rs

@@ -501,11 +501,11 @@ async fn test_07_age_stateless_suite() {
     let mut s = connect(port).await;
 
     // GENENC -> [recipient, identity]
-    let gen = send_cmd(&mut s, &["AGE", "GENENC"]).await;
+    let genenc = send_cmd(&mut s, &["AGE", "GENENC"]).await;
     assert!(
-        gen.starts_with("*2\r\n$"),
+        genenc.starts_with("*2\r\n$"),
         "AGE GENENC should return array [recipient, identity], got:\n{}",
-        gen
+        genenc
     );
 
     // Parse simple RESP array of two bulk strings to extract keys
@@ -520,7 +520,7 @@ async fn test_07_age_stateless_suite() {
         let ident = lines.next().unwrap_or("").to_string();
         (recip, ident)
     }
-    let (recipient, identity) = parse_two_bulk_array(&gen);
+    let (recipient, identity) = parse_two_bulk_array(&genenc);
     assert!(
         recipient.starts_with("age1") && identity.starts_with("AGE-SECRET-KEY-1"),
         "Unexpected AGE key formats.\nrecipient: {}\nidentity: {}",