diff --git a/changelog.md b/changelog.md index 78f9f7e..39cdd28 100644 --- a/changelog.md +++ b/changelog.md @@ -1,5 +1,12 @@ # Changelog +## 2026-05-02 - 2.10.0 - feat(rustdb) +extract service API logic into a dedicated Rust module and expose shared service types + +- adds a new rustdb service_api module to handle health, tenant, and database import/export operations +- moves SmartDB service interfaces into a dedicated TypeScript service-types module and re-exports them through the public API +- updates management request handling to delegate service operations through shared service API helpers + ## 2026-05-02 - 2.9.0 - feat(server) add tenant management, health checks, and database export/import APIs diff --git a/rust/crates/rustdb/src/lib.rs b/rust/crates/rustdb/src/lib.rs index 6f67a1c..f574ca5 100644 --- a/rust/crates/rustdb/src/lib.rs +++ b/rust/crates/rustdb/src/lib.rs @@ -1,4 +1,5 @@ pub mod management; +pub mod service_api; use std::fs::File; use std::io::BufReader; diff --git a/rust/crates/rustdb/src/management.rs b/rust/crates/rustdb/src/management.rs index 8a45ce8..a7cbf5a 100644 --- a/rust/crates/rustdb/src/management.rs +++ b/rust/crates/rustdb/src/management.rs @@ -1,11 +1,11 @@ use anyhow::Result; -use bson::{Bson, Document}; use serde::{Deserialize, Serialize}; use tokio::io::{AsyncBufReadExt, BufReader}; use tracing::{info, error}; +use crate::service_api; use crate::RustDb; -use rustdb_config::{RustDbOptions, StorageType}; +use rustdb_config::RustDbOptions; /// A management request from the TypeScript wrapper. #[derive(Debug, Deserialize)] @@ -140,19 +140,36 @@ async fn handle_request( "start" => handle_start(&id, &request.params, db).await, "stop" => handle_stop(&id, db).await, "getStatus" => handle_get_status(&id, db), - "getHealth" => handle_get_health(&id, db).await, + "getHealth" => ManagementResponse::ok(id, service_api::get_health(db.as_ref()).await), "getMetrics" => handle_get_metrics(&id, db).await, - "createDatabaseTenant" => handle_create_database_tenant(&id, &request.params, db).await, - "deleteDatabaseTenant" => handle_delete_database_tenant(&id, &request.params, db).await, - "rotateDatabaseTenantPassword" => { - handle_rotate_database_tenant_password(&id, &request.params, db).await - } - "listDatabaseTenants" => handle_list_database_tenants(&id, db), - "getDatabaseTenantDescriptor" => { - handle_get_database_tenant_descriptor(&id, &request.params, db) - } - "exportDatabase" => handle_export_database(&id, &request.params, db).await, - "importDatabase" => handle_import_database(&id, &request.params, db).await, + "createDatabaseTenant" => match db.as_ref() { + Some(d) => service_response(&id, service_api::create_database_tenant(d, &request.params).await), + None => server_not_running_response(&id), + }, + "deleteDatabaseTenant" => match db.as_ref() { + Some(d) => service_response(&id, service_api::delete_database_tenant(d, &request.params).await), + None => server_not_running_response(&id), + }, + "rotateDatabaseTenantPassword" => match db.as_ref() { + Some(d) => service_response(&id, service_api::rotate_database_tenant_password(d, &request.params).await), + None => server_not_running_response(&id), + }, + "listDatabaseTenants" => match db.as_ref() { + Some(d) => service_response(&id, service_api::list_database_tenants(d)), + None => server_not_running_response(&id), + }, + "getDatabaseTenantDescriptor" => match db.as_ref() { + Some(d) => service_response(&id, service_api::get_database_tenant_descriptor(d, &request.params)), + None => server_not_running_response(&id), + }, + "exportDatabase" => match db.as_ref() { + Some(d) => service_response(&id, service_api::export_database(d, &request.params).await), + None => server_not_running_response(&id), + }, + "importDatabase" => match db.as_ref() { + Some(d) => service_response(&id, service_api::import_database(d, &request.params).await), + None => server_not_running_response(&id), + }, "getOpLog" => handle_get_oplog(&id, &request.params, db), "getOpLogStats" => handle_get_oplog_stats(&id, db), "revertToSeq" => handle_revert_to_seq(&id, &request.params, db).await, @@ -162,6 +179,17 @@ async fn handle_request( } } +fn service_response(id: &str, result: service_api::ServiceResult) -> ManagementResponse { + match result { + Ok(value) => ManagementResponse::ok(id.to_string(), value), + Err(message) => ManagementResponse::err(id.to_string(), message), + } +} + +fn server_not_running_response(id: &str) -> ManagementResponse { + ManagementResponse::err(id.to_string(), "Server is not running".to_string()) +} + async fn handle_start( id: &str, params: &serde_json::Value, @@ -244,42 +272,6 @@ fn handle_get_status( } } -async fn handle_get_health(id: &str, db: &Option) -> ManagementResponse { - match db.as_ref() { - Some(d) => { - let ctx = d.ctx(); - let (database_count, collection_count) = database_and_collection_counts(ctx).await; - let options = d.options(); - let storage = match &options.storage { - StorageType::Memory => "memory", - StorageType::File => "file", - }; - ManagementResponse::ok( - id.to_string(), - serde_json::json!({ - "running": true, - "storage": storage, - "storagePath": options.storage_path.clone().or_else(|| options.persist_path.clone()), - "authEnabled": ctx.auth.enabled(), - "authUsers": ctx.auth.user_count(), - "usersPathConfigured": options.auth.users_path.is_some(), - "databaseCount": database_count, - "collectionCount": collection_count, - "uptimeSeconds": ctx.start_time.elapsed().as_secs(), - }), - ) - } - None => ManagementResponse::ok( - id.to_string(), - serde_json::json!({ - "running": false, - "databaseCount": 0, - "collectionCount": 0, - }), - ), - } -} - async fn handle_get_metrics( id: &str, db: &Option, @@ -316,501 +308,6 @@ async fn handle_get_metrics( } } -async fn handle_create_database_tenant( - id: &str, - params: &serde_json::Value, - db: &Option, -) -> ManagementResponse { - let d = match db.as_ref() { - Some(d) => d, - None => return ManagementResponse::err(id.to_string(), "Server is not running".to_string()), - }; - let ctx = d.ctx(); - if !ctx.auth.enabled() { - return ManagementResponse::err( - id.to_string(), - "Authentication must be enabled to create database tenants".to_string(), - ); - } - - let database_name = match string_param(params, "databaseName") { - Ok(value) => value, - Err(message) => return ManagementResponse::err(id.to_string(), message), - }; - if let Err(message) = validate_database_name(database_name) { - return ManagementResponse::err(id.to_string(), message); - } - let username = match string_param(params, "username") { - Ok(value) => value, - Err(message) => return ManagementResponse::err(id.to_string(), message), - }; - if let Err(message) = validate_username(username) { - return ManagementResponse::err(id.to_string(), message); - } - let password = match string_param(params, "password") { - Ok(value) => value, - Err(message) => return ManagementResponse::err(id.to_string(), message), - }; - if password.is_empty() { - return ManagementResponse::err(id.to_string(), "password must not be empty".to_string()); - } - let roles = match roles_param(params) { - Ok(roles) => roles, - Err(message) => return ManagementResponse::err(id.to_string(), message), - }; - - if let Err(e) = ctx.storage.create_database(database_name).await { - if !is_already_exists(&e.to_string()) { - return ManagementResponse::err( - id.to_string(), - format!("Failed to create database: {e}"), - ); - } - } - - match ctx - .auth - .create_user(database_name, username, password, roles) - { - Ok(()) => { - let users = ctx.auth.users_info(database_name, Some(username)); - match users.first() { - Some(user) => ManagementResponse::ok(id.to_string(), tenant_descriptor_json(user)), - None => ManagementResponse::err( - id.to_string(), - "Tenant user was created but could not be read back".to_string(), - ), - } - } - Err(e) => { - ManagementResponse::err(id.to_string(), format!("Failed to create tenant user: {e}")) - } - } -} - -async fn handle_delete_database_tenant( - id: &str, - params: &serde_json::Value, - db: &Option, -) -> ManagementResponse { - let d = match db.as_ref() { - Some(d) => d, - None => { - return ManagementResponse::err(id.to_string(), "Server is not running".to_string()) - } - }; - let ctx = d.ctx(); - let database_name = match string_param(params, "databaseName") { - Ok(value) => value, - Err(message) => return ManagementResponse::err(id.to_string(), message), - }; - if let Err(message) = validate_database_name(database_name) { - return ManagementResponse::err(id.to_string(), message); - } - let username = params.get("username").and_then(|v| v.as_str()); - if let Some(username) = username { - if let Err(message) = validate_username(username) { - return ManagementResponse::err(id.to_string(), message); - } - } - - if let Err(e) = ctx.storage.drop_database(database_name).await { - return ManagementResponse::err(id.to_string(), format!("Failed to drop database: {e}")); - } - remove_database_indexes(ctx, database_name); - - let mut deleted_users = 0usize; - if ctx.auth.enabled() { - if let Some(username) = username { - match ctx.auth.drop_user(database_name, username) { - Ok(()) => deleted_users = 1, - Err(rustdb_auth::AuthError::UserNotFound(_)) => deleted_users = 0, - Err(e) => { - return ManagementResponse::err( - id.to_string(), - format!("Failed to drop tenant user: {e}"), - ) - } - } - } else { - match ctx.auth.drop_users_for_database(database_name) { - Ok(count) => deleted_users = count, - Err(e) => { - return ManagementResponse::err( - id.to_string(), - format!("Failed to drop tenant users: {e}"), - ) - } - } - } - } - - ManagementResponse::ok( - id.to_string(), - serde_json::json!({ - "databaseName": database_name, - "deletedUsers": deleted_users, - "databaseDropped": true, - }), - ) -} - -async fn handle_rotate_database_tenant_password( - id: &str, - params: &serde_json::Value, - db: &Option, -) -> ManagementResponse { - let d = match db.as_ref() { - Some(d) => d, - None => { - return ManagementResponse::err(id.to_string(), "Server is not running".to_string()) - } - }; - let ctx = d.ctx(); - if !ctx.auth.enabled() { - return ManagementResponse::err( - id.to_string(), - "Authentication must be enabled to rotate database tenant passwords".to_string(), - ); - } - - let username = match string_param(params, "username") { - Ok(value) => value, - Err(message) => return ManagementResponse::err(id.to_string(), message), - }; - if let Err(message) = validate_username(username) { - return ManagementResponse::err(id.to_string(), message); - } - let password = match string_param(params, "password") { - Ok(value) => value, - Err(message) => return ManagementResponse::err(id.to_string(), message), - }; - if password.is_empty() { - return ManagementResponse::err(id.to_string(), "password must not be empty".to_string()); - } - - let matches: Vec<_> = ctx - .auth - .list_users() - .into_iter() - .filter(|user| user.username == username) - .collect(); - if matches.is_empty() { - return ManagementResponse::err( - id.to_string(), - format!("tenant user not found: {username}"), - ); - } - if matches.len() > 1 { - return ManagementResponse::err( - id.to_string(), - format!("tenant username is ambiguous across databases: {username}"), - ); - } - let user = &matches[0]; - match ctx - .auth - .update_user(&user.database, username, Some(password), None) - { - Ok(()) => { - let users = ctx.auth.users_info(&user.database, Some(username)); - match users.first() { - Some(user) => ManagementResponse::ok(id.to_string(), tenant_descriptor_json(user)), - None => ManagementResponse::err( - id.to_string(), - "Tenant user was updated but could not be read back".to_string(), - ), - } - } - Err(e) => ManagementResponse::err( - id.to_string(), - format!("Failed to rotate tenant password: {e}"), - ), - } -} - -fn handle_list_database_tenants(id: &str, db: &Option) -> ManagementResponse { - let d = match db.as_ref() { - Some(d) => d, - None => { - return ManagementResponse::err(id.to_string(), "Server is not running".to_string()) - } - }; - let tenants: Vec = d - .ctx() - .auth - .list_users() - .into_iter() - .filter(|user| user.database != "admin") - .map(|user| tenant_descriptor_json(&user)) - .collect(); - ManagementResponse::ok(id.to_string(), serde_json::json!({ "tenants": tenants })) -} - -fn handle_get_database_tenant_descriptor( - id: &str, - params: &serde_json::Value, - db: &Option, -) -> ManagementResponse { - let d = match db.as_ref() { - Some(d) => d, - None => { - return ManagementResponse::err(id.to_string(), "Server is not running".to_string()) - } - }; - let database_name = match string_param(params, "databaseName") { - Ok(value) => value, - Err(message) => return ManagementResponse::err(id.to_string(), message), - }; - let username = match string_param(params, "username") { - Ok(value) => value, - Err(message) => return ManagementResponse::err(id.to_string(), message), - }; - let users = d.ctx().auth.users_info(database_name, Some(username)); - match users.first() { - Some(user) => ManagementResponse::ok(id.to_string(), tenant_descriptor_json(user)), - None => ManagementResponse::err( - id.to_string(), - format!("tenant user not found: {database_name}.{username}"), - ), - } -} - -async fn handle_export_database( - id: &str, - params: &serde_json::Value, - db: &Option, -) -> ManagementResponse { - let d = match db.as_ref() { - Some(d) => d, - None => { - return ManagementResponse::err(id.to_string(), "Server is not running".to_string()) - } - }; - let ctx = d.ctx(); - let database_name = match string_param(params, "databaseName") { - Ok(value) => value, - Err(message) => return ManagementResponse::err(id.to_string(), message), - }; - if let Err(message) = validate_database_name(database_name) { - return ManagementResponse::err(id.to_string(), message); - } - match ctx.storage.database_exists(database_name).await { - Ok(true) => {} - Ok(false) => { - return ManagementResponse::err( - id.to_string(), - format!("database not found: {database_name}"), - ) - } - Err(e) => { - return ManagementResponse::err( - id.to_string(), - format!("Failed to check database: {e}"), - ) - } - } - - let collection_names = match ctx.storage.list_collections(database_name).await { - Ok(collections) => collections, - Err(e) => { - return ManagementResponse::err( - id.to_string(), - format!("Failed to list collections: {e}"), - ) - } - }; - let mut collections = Vec::with_capacity(collection_names.len()); - for collection_name in collection_names { - let documents = match ctx.storage.find_all(database_name, &collection_name).await { - Ok(docs) => docs - .into_iter() - .map(|doc| bson_doc_to_json(&doc)) - .collect::>(), - Err(e) => { - return ManagementResponse::err( - id.to_string(), - format!("Failed to export collection '{collection_name}': {e}"), - ) - } - }; - let indexes = match ctx - .storage - .get_indexes(database_name, &collection_name) - .await - { - Ok(specs) => specs - .into_iter() - .map(|doc| bson_doc_to_json(&doc)) - .collect::>(), - Err(_) => Vec::new(), - }; - collections.push(serde_json::json!({ - "name": collection_name, - "documents": documents, - "indexes": indexes, - })); - } - - ManagementResponse::ok( - id.to_string(), - serde_json::json!({ - "format": "smartdb.database.export.v1", - "databaseName": database_name, - "exportedAtMs": now_ms(), - "collections": collections, - }), - ) -} - -async fn handle_import_database( - id: &str, - params: &serde_json::Value, - db: &Option, -) -> ManagementResponse { - let d = match db.as_ref() { - Some(d) => d, - None => { - return ManagementResponse::err(id.to_string(), "Server is not running".to_string()) - } - }; - let ctx = d.ctx(); - let database_name = match string_param(params, "databaseName") { - Ok(value) => value, - Err(message) => return ManagementResponse::err(id.to_string(), message), - }; - if let Err(message) = validate_database_name(database_name) { - return ManagementResponse::err(id.to_string(), message); - } - let source = match params.get("source") { - Some(value) => value, - None => { - return ManagementResponse::err( - id.to_string(), - "Missing 'source' parameter".to_string(), - ) - } - }; - let source_collections = match source.get("collections").and_then(|value| value.as_array()) { - Some(collections) => collections, - None => { - return ManagementResponse::err( - id.to_string(), - "source.collections must be an array".to_string(), - ) - } - }; - - if let Err(e) = ctx.storage.drop_database(database_name).await { - return ManagementResponse::err( - id.to_string(), - format!("Failed to clear database before import: {e}"), - ); - } - remove_database_indexes(ctx, database_name); - if let Err(e) = ctx.storage.create_database(database_name).await { - if !is_already_exists(&e.to_string()) { - return ManagementResponse::err( - id.to_string(), - format!("Failed to create database: {e}"), - ); - } - } - - let mut imported_collections = 0usize; - let mut imported_documents = 0usize; - for collection in source_collections { - let collection_name = match collection.get("name").and_then(|value| value.as_str()) { - Some(value) => value, - None => { - return ManagementResponse::err( - id.to_string(), - "source collection is missing a string 'name'".to_string(), - ) - } - }; - if let Err(message) = validate_collection_name(collection_name) { - return ManagementResponse::err(id.to_string(), message); - } - if let Err(e) = ctx - .storage - .create_collection(database_name, collection_name) - .await - { - if !is_already_exists(&e.to_string()) { - return ManagementResponse::err( - id.to_string(), - format!("Failed to create collection '{collection_name}': {e}"), - ); - } - } - - if let Some(documents) = collection - .get("documents") - .and_then(|value| value.as_array()) - { - for document_value in documents { - let document = match json_to_bson_doc(document_value) { - Ok(document) => document, - Err(message) => { - return ManagementResponse::err( - id.to_string(), - format!("Invalid document in '{collection_name}': {message}"), - ) - } - }; - if let Err(e) = ctx - .storage - .insert_one(database_name, collection_name, document) - .await - { - return ManagementResponse::err( - id.to_string(), - format!("Failed to import document into '{collection_name}': {e}"), - ); - } - imported_documents += 1; - } - } - - if let Some(indexes) = collection.get("indexes").and_then(|value| value.as_array()) { - for index_value in indexes { - let index_doc = match json_to_bson_doc(index_value) { - Ok(document) => document, - Err(message) => { - return ManagementResponse::err( - id.to_string(), - format!("Invalid index in '{collection_name}': {message}"), - ) - } - }; - let name = index_doc.get_str("name").unwrap_or("_id_").to_string(); - if let Err(e) = ctx - .storage - .save_index(database_name, collection_name, &name, index_doc) - .await - { - return ManagementResponse::err( - id.to_string(), - format!("Failed to import index '{name}' into '{collection_name}': {e}"), - ); - } - } - } - - imported_collections += 1; - } - - ManagementResponse::ok( - id.to_string(), - serde_json::json!({ - "databaseName": database_name, - "collections": imported_collections, - "documents": imported_documents, - }), - ) -} - fn handle_get_oplog( id: &str, params: &serde_json::Value, @@ -1105,129 +602,6 @@ async fn handle_get_documents( ) } -async fn database_and_collection_counts(ctx: &rustdb_commands::CommandContext) -> (usize, u64) { - let databases = ctx.storage.list_databases().await.unwrap_or_default(); - let mut collections = 0u64; - for database in &databases { - if let Ok(database_collections) = ctx.storage.list_collections(database).await { - collections += database_collections.len() as u64; - } - } - (databases.len(), collections) -} - -fn remove_database_indexes(ctx: &rustdb_commands::CommandContext, database_name: &str) { - let prefix = format!("{}.", database_name); - let keys_to_remove: Vec = ctx - .indexes - .iter() - .filter(|entry| entry.key().starts_with(&prefix)) - .map(|entry| entry.key().clone()) - .collect(); - for key in keys_to_remove { - ctx.indexes.remove(&key); - } -} - -fn tenant_descriptor_json(user: &rustdb_auth::AuthenticatedUser) -> serde_json::Value { - serde_json::json!({ - "databaseName": user.database.clone(), - "username": user.username.clone(), - "roles": user.roles.clone(), - "authSource": user.database.clone(), - }) -} - -fn string_param<'a>(params: &'a serde_json::Value, key: &str) -> Result<&'a str, String> { - params - .get(key) - .and_then(|value| value.as_str()) - .ok_or_else(|| format!("Missing '{key}' parameter")) -} - -fn roles_param(params: &serde_json::Value) -> Result, String> { - let Some(value) = params.get("roles") else { - return Ok(vec!["readWrite".to_string(), "dbAdmin".to_string()]); - }; - let roles = value - .as_array() - .ok_or_else(|| "roles must be an array of strings".to_string())?; - let mut result = Vec::with_capacity(roles.len()); - for role in roles { - let Some(role_name) = role.as_str() else { - return Err("roles must be an array of strings".to_string()); - }; - if role_name.is_empty() { - return Err("roles must not contain empty role names".to_string()); - } - result.push(role_name.to_string()); - } - Ok(result) -} - -fn validate_database_name(name: &str) -> Result<(), String> { - if name.is_empty() { - return Err("databaseName must not be empty".to_string()); - } - if name == "." - || name == ".." - || name.contains('/') - || name.contains('\\') - || name.contains('\0') - { - return Err(format!( - "databaseName contains invalid path characters: {name}" - )); - } - Ok(()) -} - -fn validate_collection_name(name: &str) -> Result<(), String> { - if name.is_empty() { - return Err("collection name must not be empty".to_string()); - } - if name == "." - || name == ".." - || name.contains('/') - || name.contains('\\') - || name.contains('\0') - { - return Err(format!( - "collection name contains invalid path characters: {name}" - )); - } - Ok(()) -} - -fn validate_username(username: &str) -> Result<(), String> { - if username.is_empty() { - return Err("username must not be empty".to_string()); - } - if username.contains('\0') { - return Err("username must not contain NUL bytes".to_string()); - } - Ok(()) -} - -fn is_already_exists(message: &str) -> bool { - message.contains("AlreadyExists") || message.contains("already exists") -} - -fn json_to_bson_doc(value: &serde_json::Value) -> Result { - let bson_value: Bson = serde_json::from_value(value.clone()).map_err(|e| e.to_string())?; - match bson_value { - Bson::Document(document) => Ok(document), - _ => Err("expected BSON document".to_string()), - } -} - -fn now_ms() -> u64 { - std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap_or_default() - .as_millis() as u64 -} - /// Convert a BSON Document to a serde_json::Value. fn bson_doc_to_json(doc: &bson::Document) -> serde_json::Value { // Use bson's built-in relaxed extended JSON serialization. diff --git a/rust/crates/rustdb/src/service_api.rs b/rust/crates/rustdb/src/service_api.rs new file mode 100644 index 0000000..f5047b7 --- /dev/null +++ b/rust/crates/rustdb/src/service_api.rs @@ -0,0 +1,446 @@ +use bson::{Bson, Document}; +use rustdb_config::StorageType; + +use crate::RustDb; + +pub type ServiceResult = Result; + +pub async fn get_health(db: Option<&RustDb>) -> serde_json::Value { + match db { + Some(d) => { + let ctx = d.ctx(); + let (database_count, collection_count) = database_and_collection_counts(ctx).await; + let options = d.options(); + let storage = match &options.storage { + StorageType::Memory => "memory", + StorageType::File => "file", + }; + serde_json::json!({ + "running": true, + "storage": storage, + "storagePath": options.storage_path.clone().or_else(|| options.persist_path.clone()), + "authEnabled": ctx.auth.enabled(), + "authUsers": ctx.auth.user_count(), + "usersPathConfigured": options.auth.users_path.is_some(), + "databaseCount": database_count, + "collectionCount": collection_count, + "uptimeSeconds": ctx.start_time.elapsed().as_secs(), + }) + } + None => serde_json::json!({ + "running": false, + "databaseCount": 0, + "collectionCount": 0, + }), + } +} + +pub async fn create_database_tenant(db: &RustDb, params: &serde_json::Value) -> ServiceResult { + let ctx = db.ctx(); + if !ctx.auth.enabled() { + return Err("Authentication must be enabled to create database tenants".to_string()); + } + + let database_name = string_param(params, "databaseName")?; + validate_database_name(database_name)?; + let username = string_param(params, "username")?; + validate_username(username)?; + let password = string_param(params, "password")?; + if password.is_empty() { + return Err("password must not be empty".to_string()); + } + let roles = roles_param(params)?; + + if let Err(e) = ctx.storage.create_database(database_name).await { + if !is_already_exists(&e.to_string()) { + return Err(format!("Failed to create database: {e}")); + } + } + + match ctx + .auth + .create_user(database_name, username, password, roles) + { + Ok(()) => { + let users = ctx.auth.users_info(database_name, Some(username)); + users + .first() + .map(tenant_descriptor_json) + .ok_or_else(|| "Tenant user was created but could not be read back".to_string()) + } + Err(e) => Err(format!("Failed to create tenant user: {e}")), + } +} + +pub async fn delete_database_tenant(db: &RustDb, params: &serde_json::Value) -> ServiceResult { + let ctx = db.ctx(); + let database_name = string_param(params, "databaseName")?; + validate_database_name(database_name)?; + let username = params.get("username").and_then(|v| v.as_str()); + if let Some(username) = username { + validate_username(username)?; + } + + if let Err(e) = ctx.storage.drop_database(database_name).await { + return Err(format!("Failed to drop database: {e}")); + } + remove_database_indexes(ctx, database_name); + + let mut deleted_users = 0usize; + if ctx.auth.enabled() { + if let Some(username) = username { + match ctx.auth.drop_user(database_name, username) { + Ok(()) => deleted_users = 1, + Err(rustdb_auth::AuthError::UserNotFound(_)) => deleted_users = 0, + Err(e) => return Err(format!("Failed to drop tenant user: {e}")), + } + } else { + deleted_users = ctx + .auth + .drop_users_for_database(database_name) + .map_err(|e| format!("Failed to drop tenant users: {e}"))?; + } + } + + Ok(serde_json::json!({ + "databaseName": database_name, + "deletedUsers": deleted_users, + "databaseDropped": true, + })) +} + +pub async fn rotate_database_tenant_password( + db: &RustDb, + params: &serde_json::Value, +) -> ServiceResult { + let ctx = db.ctx(); + if !ctx.auth.enabled() { + return Err( + "Authentication must be enabled to rotate database tenant passwords".to_string(), + ); + } + + let username = string_param(params, "username")?; + validate_username(username)?; + let password = string_param(params, "password")?; + if password.is_empty() { + return Err("password must not be empty".to_string()); + } + + let matches: Vec<_> = ctx + .auth + .list_users() + .into_iter() + .filter(|user| user.username == username) + .collect(); + if matches.is_empty() { + return Err(format!("tenant user not found: {username}")); + } + if matches.len() > 1 { + return Err(format!( + "tenant username is ambiguous across databases: {username}" + )); + } + + let user = &matches[0]; + ctx.auth + .update_user(&user.database, username, Some(password), None) + .map_err(|e| format!("Failed to rotate tenant password: {e}"))?; + let users = ctx.auth.users_info(&user.database, Some(username)); + users + .first() + .map(tenant_descriptor_json) + .ok_or_else(|| "Tenant user was updated but could not be read back".to_string()) +} + +pub fn list_database_tenants(db: &RustDb) -> ServiceResult { + let tenants: Vec = db + .ctx() + .auth + .list_users() + .into_iter() + .filter(|user| user.database != "admin") + .map(|user| tenant_descriptor_json(&user)) + .collect(); + Ok(serde_json::json!({ "tenants": tenants })) +} + +pub fn get_database_tenant_descriptor(db: &RustDb, params: &serde_json::Value) -> ServiceResult { + let database_name = string_param(params, "databaseName")?; + let username = string_param(params, "username")?; + let users = db.ctx().auth.users_info(database_name, Some(username)); + users + .first() + .map(tenant_descriptor_json) + .ok_or_else(|| format!("tenant user not found: {database_name}.{username}")) +} + +pub async fn export_database(db: &RustDb, params: &serde_json::Value) -> ServiceResult { + let ctx = db.ctx(); + let database_name = string_param(params, "databaseName")?; + validate_database_name(database_name)?; + match ctx.storage.database_exists(database_name).await { + Ok(true) => {} + Ok(false) => return Err(format!("database not found: {database_name}")), + Err(e) => return Err(format!("Failed to check database: {e}")), + } + + let collection_names = ctx + .storage + .list_collections(database_name) + .await + .map_err(|e| format!("Failed to list collections: {e}"))?; + let mut collections = Vec::with_capacity(collection_names.len()); + for collection_name in collection_names { + let documents = ctx + .storage + .find_all(database_name, &collection_name) + .await + .map_err(|e| format!("Failed to export collection '{collection_name}': {e}"))? + .into_iter() + .map(|doc| bson_doc_to_json(&doc)) + .collect::>(); + let indexes = match ctx + .storage + .get_indexes(database_name, &collection_name) + .await + { + Ok(specs) => specs + .into_iter() + .map(|doc| bson_doc_to_json(&doc)) + .collect::>(), + Err(_) => Vec::new(), + }; + collections.push(serde_json::json!({ + "name": collection_name, + "documents": documents, + "indexes": indexes, + })); + } + + Ok(serde_json::json!({ + "format": "smartdb.database.export.v1", + "databaseName": database_name, + "exportedAtMs": now_ms(), + "collections": collections, + })) +} + +pub async fn import_database(db: &RustDb, params: &serde_json::Value) -> ServiceResult { + let ctx = db.ctx(); + let database_name = string_param(params, "databaseName")?; + validate_database_name(database_name)?; + let source = params + .get("source") + .ok_or_else(|| "Missing 'source' parameter".to_string())?; + let source_collections = source + .get("collections") + .and_then(|value| value.as_array()) + .ok_or_else(|| "source.collections must be an array".to_string())?; + + if let Err(e) = ctx.storage.drop_database(database_name).await { + return Err(format!("Failed to clear database before import: {e}")); + } + remove_database_indexes(ctx, database_name); + if let Err(e) = ctx.storage.create_database(database_name).await { + if !is_already_exists(&e.to_string()) { + return Err(format!("Failed to create database: {e}")); + } + } + + let mut imported_collections = 0usize; + let mut imported_documents = 0usize; + for collection in source_collections { + let collection_name = collection + .get("name") + .and_then(|value| value.as_str()) + .ok_or_else(|| "source collection is missing a string 'name'".to_string())?; + validate_collection_name(collection_name)?; + if let Err(e) = ctx + .storage + .create_collection(database_name, collection_name) + .await + { + if !is_already_exists(&e.to_string()) { + return Err(format!( + "Failed to create collection '{collection_name}': {e}" + )); + } + } + + if let Some(documents) = collection + .get("documents") + .and_then(|value| value.as_array()) + { + for document_value in documents { + let document = json_to_bson_doc(document_value).map_err(|message| { + format!("Invalid document in '{collection_name}': {message}") + })?; + if let Err(e) = ctx + .storage + .insert_one(database_name, collection_name, document) + .await + { + return Err(format!( + "Failed to import document into '{collection_name}': {e}" + )); + } + imported_documents += 1; + } + } + + if let Some(indexes) = collection.get("indexes").and_then(|value| value.as_array()) { + for index_value in indexes { + let index_doc = json_to_bson_doc(index_value).map_err(|message| { + format!("Invalid index in '{collection_name}': {message}") + })?; + let name = index_doc.get_str("name").unwrap_or("_id_").to_string(); + if let Err(e) = ctx + .storage + .save_index(database_name, collection_name, &name, index_doc) + .await + { + return Err(format!( + "Failed to import index '{name}' into '{collection_name}': {e}" + )); + } + } + } + + imported_collections += 1; + } + + Ok(serde_json::json!({ + "databaseName": database_name, + "collections": imported_collections, + "documents": imported_documents, + })) +} + +async fn database_and_collection_counts(ctx: &rustdb_commands::CommandContext) -> (usize, u64) { + let databases = ctx.storage.list_databases().await.unwrap_or_default(); + let mut collections = 0u64; + for database in &databases { + if let Ok(database_collections) = ctx.storage.list_collections(database).await { + collections += database_collections.len() as u64; + } + } + (databases.len(), collections) +} + +fn remove_database_indexes(ctx: &rustdb_commands::CommandContext, database_name: &str) { + let prefix = format!("{}.", database_name); + let keys_to_remove: Vec = ctx + .indexes + .iter() + .filter(|entry| entry.key().starts_with(&prefix)) + .map(|entry| entry.key().clone()) + .collect(); + for key in keys_to_remove { + ctx.indexes.remove(&key); + } +} + +fn tenant_descriptor_json(user: &rustdb_auth::AuthenticatedUser) -> serde_json::Value { + serde_json::json!({ + "databaseName": user.database.clone(), + "username": user.username.clone(), + "roles": user.roles.clone(), + "authSource": user.database.clone(), + }) +} + +fn string_param<'a>(params: &'a serde_json::Value, key: &str) -> Result<&'a str, String> { + params + .get(key) + .and_then(|value| value.as_str()) + .ok_or_else(|| format!("Missing '{key}' parameter")) +} + +fn roles_param(params: &serde_json::Value) -> Result, String> { + let Some(value) = params.get("roles") else { + return Ok(vec!["readWrite".to_string(), "dbAdmin".to_string()]); + }; + let roles = value + .as_array() + .ok_or_else(|| "roles must be an array of strings".to_string())?; + let mut result = Vec::with_capacity(roles.len()); + for role in roles { + let Some(role_name) = role.as_str() else { + return Err("roles must be an array of strings".to_string()); + }; + if role_name.is_empty() { + return Err("roles must not contain empty role names".to_string()); + } + result.push(role_name.to_string()); + } + Ok(result) +} + +fn validate_database_name(name: &str) -> Result<(), String> { + if name.is_empty() { + return Err("databaseName must not be empty".to_string()); + } + if name == "." + || name == ".." + || name.contains('/') + || name.contains('\\') + || name.contains('\0') + { + return Err(format!( + "databaseName contains invalid path characters: {name}" + )); + } + Ok(()) +} + +fn validate_collection_name(name: &str) -> Result<(), String> { + if name.is_empty() { + return Err("collection name must not be empty".to_string()); + } + if name == "." + || name == ".." + || name.contains('/') + || name.contains('\\') + || name.contains('\0') + { + return Err(format!( + "collection name contains invalid path characters: {name}" + )); + } + Ok(()) +} + +fn validate_username(username: &str) -> Result<(), String> { + if username.is_empty() { + return Err("username must not be empty".to_string()); + } + if username.contains('\0') { + return Err("username must not contain NUL bytes".to_string()); + } + Ok(()) +} + +fn is_already_exists(message: &str) -> bool { + message.contains("AlreadyExists") || message.contains("already exists") +} + +fn json_to_bson_doc(value: &serde_json::Value) -> Result { + let bson_value: Bson = serde_json::from_value(value.clone()).map_err(|e| e.to_string())?; + match bson_value { + Bson::Document(document) => Ok(document), + _ => Err("expected BSON document".to_string()), + } +} + +fn now_ms() -> u64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_millis() as u64 +} + +fn bson_doc_to_json(doc: &bson::Document) -> serde_json::Value { + let bson_val = bson::Bson::Document(doc.clone()); + bson_val.into_relaxed_extjson() +} diff --git a/ts/00_commitinfo_data.ts b/ts/00_commitinfo_data.ts index c156b57..94def18 100644 --- a/ts/00_commitinfo_data.ts +++ b/ts/00_commitinfo_data.ts @@ -3,6 +3,6 @@ */ export const commitinfo = { name: '@push.rocks/smartdb', - version: '2.9.0', + version: '2.10.0', description: 'A MongoDB-compatible embedded database server with wire protocol support, backed by a high-performance Rust engine.' } diff --git a/ts/ts_smartdb/index.ts b/ts/ts_smartdb/index.ts index 9978237..4681c27 100644 --- a/ts/ts_smartdb/index.ts +++ b/ts/ts_smartdb/index.ts @@ -21,6 +21,10 @@ export type { ICollectionInfo, IDocumentsResult, ISmartDbMetrics, +} from './rust-db-bridge.js'; + +// Export service API types +export type { ISmartDbHealth, ISmartDbDatabaseTenantInput, ISmartDbDeleteDatabaseTenantInput, @@ -31,4 +35,4 @@ export type { ISmartDbDatabaseExport, ISmartDbImportDatabaseInput, ISmartDbImportDatabaseResult, -} from './rust-db-bridge.js'; +} from './service-types.js'; diff --git a/ts/ts_smartdb/rust-db-bridge.ts b/ts/ts_smartdb/rust-db-bridge.ts index a08f6d2..8e1723b 100644 --- a/ts/ts_smartdb/rust-db-bridge.ts +++ b/ts/ts_smartdb/rust-db-bridge.ts @@ -2,6 +2,30 @@ import * as plugins from './plugins.js'; import * as path from 'path'; import * as url from 'url'; import { EventEmitter } from 'events'; +import type { + ISmartDbHealth, + ISmartDbDatabaseTenantInput, + ISmartDbDeleteDatabaseTenantInput, + ISmartDbRotateDatabaseTenantPasswordInput, + ISmartDbDatabaseTenantDescriptor, + ISmartDbDeleteDatabaseTenantResult, + ISmartDbDatabaseExport, + ISmartDbImportDatabaseInput, + ISmartDbImportDatabaseResult, +} from './service-types.js'; + +export type { + ISmartDbHealth, + ISmartDbDatabaseTenantInput, + ISmartDbDeleteDatabaseTenantInput, + ISmartDbRotateDatabaseTenantPasswordInput, + ISmartDbDatabaseTenantDescriptor, + ISmartDbDeleteDatabaseTenantResult, + ISmartDbDatabaseExportCollection, + ISmartDbDatabaseExport, + ISmartDbImportDatabaseInput, + ISmartDbImportDatabaseResult, +} from './service-types.js'; /** * A single oplog entry returned from the Rust engine. @@ -83,73 +107,6 @@ export interface ISmartDbMetrics { uptimeSeconds: number; } -export interface ISmartDbHealth { - running: boolean; - storage?: 'memory' | 'file'; - storagePath?: string; - authEnabled?: boolean; - authUsers?: number; - usersPathConfigured?: boolean; - databaseCount: number; - collectionCount: number; - uptimeSeconds?: number; -} - -export interface ISmartDbDatabaseTenantInput { - databaseName: string; - username: string; - password: string; - roles?: string[]; -} - -export interface ISmartDbDeleteDatabaseTenantInput { - databaseName: string; - username?: string; -} - -export interface ISmartDbRotateDatabaseTenantPasswordInput { - username: string; - password: string; -} - -export interface ISmartDbDatabaseTenantDescriptor { - databaseName: string; - username: string; - roles: string[]; - authSource: string; - mongodbUri?: string; -} - -export interface ISmartDbDeleteDatabaseTenantResult { - databaseName: string; - deletedUsers: number; - databaseDropped: boolean; -} - -export interface ISmartDbDatabaseExportCollection { - name: string; - documents: Record[]; - indexes: Record[]; -} - -export interface ISmartDbDatabaseExport { - format: 'smartdb.database.export.v1'; - databaseName: string; - exportedAtMs: number; - collections: ISmartDbDatabaseExportCollection[]; -} - -export interface ISmartDbImportDatabaseInput { - databaseName: string; - source: ISmartDbDatabaseExport; -} - -export interface ISmartDbImportDatabaseResult { - databaseName: string; - collections: number; - documents: number; -} - /** * Type-safe command definitions for the RustDb IPC protocol. */ diff --git a/ts/ts_smartdb/server/SmartdbServer.ts b/ts/ts_smartdb/server/SmartdbServer.ts index 2b5ff94..6060122 100644 --- a/ts/ts_smartdb/server/SmartdbServer.ts +++ b/ts/ts_smartdb/server/SmartdbServer.ts @@ -8,6 +8,8 @@ import type { ICollectionInfo, IDocumentsResult, ISmartDbMetrics, +} from '../rust-db-bridge.js'; +import type { ISmartDbHealth, ISmartDbDatabaseTenantInput, ISmartDbDeleteDatabaseTenantInput, @@ -17,7 +19,7 @@ import type { ISmartDbDatabaseExport, ISmartDbImportDatabaseInput, ISmartDbImportDatabaseResult, -} from '../rust-db-bridge.js'; +} from '../service-types.js'; /** * Server configuration options diff --git a/ts/ts_smartdb/service-types.ts b/ts/ts_smartdb/service-types.ts new file mode 100644 index 0000000..c9c54b2 --- /dev/null +++ b/ts/ts_smartdb/service-types.ts @@ -0,0 +1,66 @@ +export interface ISmartDbHealth { + running: boolean; + storage?: 'memory' | 'file'; + storagePath?: string; + authEnabled?: boolean; + authUsers?: number; + usersPathConfigured?: boolean; + databaseCount: number; + collectionCount: number; + uptimeSeconds?: number; +} + +export interface ISmartDbDatabaseTenantInput { + databaseName: string; + username: string; + password: string; + roles?: string[]; +} + +export interface ISmartDbDeleteDatabaseTenantInput { + databaseName: string; + username?: string; +} + +export interface ISmartDbRotateDatabaseTenantPasswordInput { + username: string; + password: string; +} + +export interface ISmartDbDatabaseTenantDescriptor { + databaseName: string; + username: string; + roles: string[]; + authSource: string; + mongodbUri?: string; +} + +export interface ISmartDbDeleteDatabaseTenantResult { + databaseName: string; + deletedUsers: number; + databaseDropped: boolean; +} + +export interface ISmartDbDatabaseExportCollection { + name: string; + documents: Record[]; + indexes: Record[]; +} + +export interface ISmartDbDatabaseExport { + format: 'smartdb.database.export.v1'; + databaseName: string; + exportedAtMs: number; + collections: ISmartDbDatabaseExportCollection[]; +} + +export interface ISmartDbImportDatabaseInput { + databaseName: string; + source: ISmartDbDatabaseExport; +} + +export interface ISmartDbImportDatabaseResult { + databaseName: string; + collections: number; + documents: number; +}