use crate::connection::{AppState, PoolKind}; use crate::db::redis_driver::{ self, RedisCommandResult, RedisConnection, RedisDatabaseInfo, RedisKeyInfo, RedisScanResult, RedisValue, }; async fn ensure_redis_pool(state: &AppState, connection_id: &str) -> Result<(), String> { state.get_or_create_pool(connection_id, None).await.map(|_| ()) } pub async fn redis_list_databases_core( state: &AppState, connection_id: &str, ) -> Result, String> { ensure_redis_pool(state, connection_id).await?; let connections = state.connections.read().await; let pool = connections.get(connection_id).ok_or("Connection not found")?; match pool { PoolKind::Redis(redis) => match redis { RedisConnection::Direct(con) => { let mut con = con.lock().await; redis_driver::list_databases(&mut *con).await } RedisConnection::Cluster(cluster) => redis_driver::list_cluster_databases(cluster).await, }, _ => Err("Not a Redis connection".to_string()), } } pub async fn redis_scan_keys_core( state: &AppState, connection_id: &str, db: u32, cursor: u64, pattern: &str, count: usize, ) -> Result { redis_scan_keys_batch_core(state, connection_id, db, cursor, pattern, count, 1, true).await } /// Batch-scan keys with server-side multi-SCAN support. /// /// Performs up to `max_iterations` SCAN cycles server-side in a single API /// call, dramatically reducing frontend↔backend roundtrips when fetching many /// keys (e.g. "fetch all" in the key browser). TYPE metadata is optional. pub async fn redis_scan_keys_batch_core( state: &AppState, connection_id: &str, db: u32, cursor: u64, pattern: &str, count: usize, max_iterations: usize, include_types: bool, ) -> Result { ensure_redis_pool(state, connection_id).await?; let connections = state.connections.read().await; let pool = connections.get(connection_id).ok_or("Connection not found")?; match pool { PoolKind::Redis(redis) => match redis { RedisConnection::Direct(con) => { let mut con = con.lock().await; redis_driver::select_db(&mut *con, db).await?; redis_driver::scan_keys_batch(&mut *con, cursor, pattern, count, max_iterations, include_types).await } RedisConnection::Cluster(cluster) => { redis_driver::ensure_cluster_db(db)?; // Cluster scan already iterates across nodes; for batch mode we // loop the cluster-level scan to accumulate keys server-side. if max_iterations <= 1 { return redis_driver::scan_cluster_keys_page_with_options( cluster, cursor, pattern, count, include_types, ) .await; } let mut all_keys: Vec = Vec::new(); let mut current_cursor = cursor; let mut total_keys: u64 = 0; for i in 0..max_iterations { let page = redis_driver::scan_cluster_keys_page_with_options( cluster, current_cursor, pattern, count, include_types, ) .await?; if i == 0 { total_keys = page.total_keys; } all_keys.extend(page.keys); if page.cursor == 0 { return Ok(RedisScanResult { cursor: 0, keys: all_keys, total_keys }); } current_cursor = page.cursor; } Ok(RedisScanResult { cursor: current_cursor, keys: all_keys, total_keys }) } }, _ => Err("Not a Redis connection".to_string()), } } #[allow(clippy::too_many_arguments)] pub async fn redis_scan_values_core( state: &AppState, connection_id: &str, db: u32, cursor: u64, pattern: &str, query: &str, include_key_matches: bool, count: usize, ) -> Result { ensure_redis_pool(state, connection_id).await?; let connections = state.connections.read().await; let pool = connections.get(connection_id).ok_or("Connection not found")?; match pool { PoolKind::Redis(redis) => match redis { RedisConnection::Direct(con) => { let mut con = con.lock().await; redis_driver::select_db(&mut *con, db).await?; redis_driver::scan_values_page(&mut *con, cursor, pattern, query, include_key_matches, count).await } RedisConnection::Cluster(cluster) => { redis_driver::ensure_cluster_db(db)?; redis_driver::scan_cluster_values_page(cluster, cursor, pattern, query, include_key_matches, count) .await } }, _ => Err("Not a Redis connection".to_string()), } } pub async fn redis_get_value_core(state: &AppState, connection_id: &str, key: &str) -> Result { redis_get_value_in_db_core(state, connection_id, 0, key).await } pub async fn redis_get_value_in_db_core( state: &AppState, connection_id: &str, db: u32, key_raw: &str, ) -> Result { ensure_redis_pool(state, connection_id).await?; let connections = state.connections.read().await; let pool = connections.get(connection_id).ok_or("Connection not found")?; match pool { PoolKind::Redis(redis) => { let key = redis_driver::redis_key_raw_to_bytes(key_raw)?; match redis { RedisConnection::Direct(con) => { let mut con = con.lock().await; redis_driver::select_db(&mut *con, db).await?; redis_driver::get_value(&mut *con, &key).await } RedisConnection::Cluster(cluster) => { redis_driver::ensure_cluster_db(db)?; let mut con = redis_driver::cluster_key_connection(cluster, &key).await?; redis_driver::get_value(&mut con, &key).await } } } _ => Err("Not a Redis connection".to_string()), } } pub async fn redis_set_string_core( state: &AppState, connection_id: &str, key: &str, value: &str, ttl: Option, ) -> Result<(), String> { redis_set_string_in_db_core(state, connection_id, 0, key, value, ttl).await } pub async fn redis_set_string_in_db_core( state: &AppState, connection_id: &str, db: u32, key_raw: &str, value: &str, ttl: Option, ) -> Result<(), String> { ensure_redis_pool(state, connection_id).await?; let connections = state.connections.read().await; let pool = connections.get(connection_id).ok_or("Connection not found")?; match pool { PoolKind::Redis(redis) => { let key = redis_driver::redis_key_raw_to_bytes(key_raw)?; match redis { RedisConnection::Direct(con) => { let mut con = con.lock().await; redis_driver::select_db(&mut *con, db).await?; redis_driver::set_string(&mut *con, &key, value, ttl).await } RedisConnection::Cluster(cluster) => { redis_driver::ensure_cluster_db(db)?; let mut con = redis_driver::cluster_key_connection(cluster, &key).await?; redis_driver::set_string(&mut con, &key, value, ttl).await } } } _ => Err("Not a Redis connection".to_string()), } } pub async fn redis_delete_key_core(state: &AppState, connection_id: &str, key: &str) -> Result<(), String> { redis_delete_key_in_db_core(state, connection_id, 0, key).await } pub async fn redis_delete_key_in_db_core( state: &AppState, connection_id: &str, db: u32, key_raw: &str, ) -> Result<(), String> { ensure_redis_pool(state, connection_id).await?; let connections = state.connections.read().await; let pool = connections.get(connection_id).ok_or("Connection not found")?; match pool { PoolKind::Redis(redis) => { let key = redis_driver::redis_key_raw_to_bytes(key_raw)?; match redis { RedisConnection::Direct(con) => { let mut con = con.lock().await; redis_driver::select_db(&mut *con, db).await?; redis_driver::delete_key(&mut *con, &key).await } RedisConnection::Cluster(cluster) => { redis_driver::ensure_cluster_db(db)?; let mut con = redis_driver::cluster_key_connection(cluster, &key).await?; redis_driver::delete_key(&mut con, &key).await } } } _ => Err("Not a Redis connection".to_string()), } } pub async fn redis_hash_set_core( state: &AppState, connection_id: &str, key: &str, field: &str, value: &str, ttl: Option, ) -> Result<(), String> { redis_hash_set_in_db_core(state, connection_id, 0, key, field, value, ttl).await } pub async fn redis_hash_set_in_db_core( state: &AppState, connection_id: &str, db: u32, key_raw: &str, field: &str, value: &str, ttl: Option, ) -> Result<(), String> { ensure_redis_pool(state, connection_id).await?; let connections = state.connections.read().await; match connections.get(connection_id).ok_or("Not found")? { PoolKind::Redis(redis) => { let key = redis_driver::redis_key_raw_to_bytes(key_raw)?; match redis { RedisConnection::Direct(con) => { let mut con = con.lock().await; redis_driver::select_db(&mut *con, db).await?; redis_driver::hash_set(&mut *con, &key, field, value, ttl).await } RedisConnection::Cluster(cluster) => { redis_driver::ensure_cluster_db(db)?; let mut con = redis_driver::cluster_key_connection(cluster, &key).await?; redis_driver::hash_set(&mut con, &key, field, value, ttl).await } } } _ => Err("Not a Redis connection".to_string()), } } pub async fn redis_hash_del_core(state: &AppState, connection_id: &str, key: &str, field: &str) -> Result<(), String> { redis_hash_del_in_db_core(state, connection_id, 0, key, field).await } pub async fn redis_hash_del_in_db_core( state: &AppState, connection_id: &str, db: u32, key_raw: &str, field: &str, ) -> Result<(), String> { ensure_redis_pool(state, connection_id).await?; let connections = state.connections.read().await; match connections.get(connection_id).ok_or("Not found")? { PoolKind::Redis(redis) => { let key = redis_driver::redis_key_raw_to_bytes(key_raw)?; match redis { RedisConnection::Direct(con) => { let mut con = con.lock().await; redis_driver::select_db(&mut *con, db).await?; redis_driver::hash_del(&mut *con, &key, field).await } RedisConnection::Cluster(cluster) => { redis_driver::ensure_cluster_db(db)?; let mut con = redis_driver::cluster_key_connection(cluster, &key).await?; redis_driver::hash_del(&mut con, &key, field).await } } } _ => Err("Not a Redis connection".to_string()), } } pub async fn redis_list_push_core( state: &AppState, connection_id: &str, key: &str, value: &str, ttl: Option, ) -> Result<(), String> { redis_list_push_in_db_core(state, connection_id, 0, key, value, ttl).await } pub async fn redis_list_push_in_db_core( state: &AppState, connection_id: &str, db: u32, key_raw: &str, value: &str, ttl: Option, ) -> Result<(), String> { ensure_redis_pool(state, connection_id).await?; let connections = state.connections.read().await; match connections.get(connection_id).ok_or("Not found")? { PoolKind::Redis(redis) => { let key = redis_driver::redis_key_raw_to_bytes(key_raw)?; match redis { RedisConnection::Direct(con) => { let mut con = con.lock().await; redis_driver::select_db(&mut *con, db).await?; redis_driver::list_push(&mut *con, &key, value, ttl).await } RedisConnection::Cluster(cluster) => { redis_driver::ensure_cluster_db(db)?; let mut con = redis_driver::cluster_key_connection(cluster, &key).await?; redis_driver::list_push(&mut con, &key, value, ttl).await } } } _ => Err("Not a Redis connection".to_string()), } } pub async fn redis_list_set_in_db_core( state: &AppState, connection_id: &str, db: u32, key_raw: &str, index: i64, value: &str, ) -> Result<(), String> { ensure_redis_pool(state, connection_id).await?; let connections = state.connections.read().await; match connections.get(connection_id).ok_or("Not found")? { PoolKind::Redis(redis) => { let key = redis_driver::redis_key_raw_to_bytes(key_raw)?; match redis { RedisConnection::Direct(con) => { let mut con = con.lock().await; redis_driver::select_db(&mut *con, db).await?; redis_driver::list_set(&mut *con, &key, index, value).await } RedisConnection::Cluster(cluster) => { redis_driver::ensure_cluster_db(db)?; let mut con = redis_driver::cluster_key_connection(cluster, &key).await?; redis_driver::list_set(&mut con, &key, index, value).await } } } _ => Err("Not a Redis connection".to_string()), } } pub async fn redis_list_remove_core( state: &AppState, connection_id: &str, key: &str, index: i64, ) -> Result<(), String> { redis_list_remove_in_db_core(state, connection_id, 0, key, index).await } pub async fn redis_list_remove_in_db_core( state: &AppState, connection_id: &str, db: u32, key_raw: &str, index: i64, ) -> Result<(), String> { ensure_redis_pool(state, connection_id).await?; let connections = state.connections.read().await; match connections.get(connection_id).ok_or("Not found")? { PoolKind::Redis(redis) => { let key = redis_driver::redis_key_raw_to_bytes(key_raw)?; match redis { RedisConnection::Direct(con) => { let mut con = con.lock().await; redis_driver::select_db(&mut *con, db).await?; redis_driver::list_remove(&mut *con, &key, index).await } RedisConnection::Cluster(cluster) => { redis_driver::ensure_cluster_db(db)?; let mut con = redis_driver::cluster_key_connection(cluster, &key).await?; redis_driver::list_remove(&mut con, &key, index).await } } } _ => Err("Not a Redis connection".to_string()), } } pub async fn redis_set_add_core( state: &AppState, connection_id: &str, key: &str, member: &str, ttl: Option, ) -> Result<(), String> { redis_set_add_in_db_core(state, connection_id, 0, key, member, ttl).await } pub async fn redis_set_add_in_db_core( state: &AppState, connection_id: &str, db: u32, key_raw: &str, member: &str, ttl: Option, ) -> Result<(), String> { ensure_redis_pool(state, connection_id).await?; let connections = state.connections.read().await; match connections.get(connection_id).ok_or("Not found")? { PoolKind::Redis(redis) => { let key = redis_driver::redis_key_raw_to_bytes(key_raw)?; match redis { RedisConnection::Direct(con) => { let mut con = con.lock().await; redis_driver::select_db(&mut *con, db).await?; redis_driver::set_add(&mut *con, &key, member, ttl).await } RedisConnection::Cluster(cluster) => { redis_driver::ensure_cluster_db(db)?; let mut con = redis_driver::cluster_key_connection(cluster, &key).await?; redis_driver::set_add(&mut con, &key, member, ttl).await } } } _ => Err("Not a Redis connection".to_string()), } } pub async fn redis_set_remove_core( state: &AppState, connection_id: &str, key: &str, member: &str, ) -> Result<(), String> { redis_set_remove_in_db_core(state, connection_id, 0, key, member).await } pub async fn redis_set_remove_in_db_core( state: &AppState, connection_id: &str, db: u32, key_raw: &str, member: &str, ) -> Result<(), String> { ensure_redis_pool(state, connection_id).await?; let connections = state.connections.read().await; match connections.get(connection_id).ok_or("Not found")? { PoolKind::Redis(redis) => { let key = redis_driver::redis_key_raw_to_bytes(key_raw)?; match redis { RedisConnection::Direct(con) => { let mut con = con.lock().await; redis_driver::select_db(&mut *con, db).await?; redis_driver::set_remove(&mut *con, &key, member).await } RedisConnection::Cluster(cluster) => { redis_driver::ensure_cluster_db(db)?; let mut con = redis_driver::cluster_key_connection(cluster, &key).await?; redis_driver::set_remove(&mut con, &key, member).await } } } _ => Err("Not a Redis connection".to_string()), } } pub async fn redis_zadd_in_db_core( state: &AppState, connection_id: &str, db: u32, key_raw: &str, member: &str, score: f64, ttl: Option, ) -> Result<(), String> { ensure_redis_pool(state, connection_id).await?; let connections = state.connections.read().await; match connections.get(connection_id).ok_or("Not found")? { PoolKind::Redis(redis) => { let key = redis_driver::redis_key_raw_to_bytes(key_raw)?; match redis { RedisConnection::Direct(con) => { let mut con = con.lock().await; redis_driver::select_db(&mut *con, db).await?; redis_driver::zadd(&mut *con, &key, member, score, ttl).await } RedisConnection::Cluster(cluster) => { redis_driver::ensure_cluster_db(db)?; let mut con = redis_driver::cluster_key_connection(cluster, &key).await?; redis_driver::zadd(&mut con, &key, member, score, ttl).await } } } _ => Err("Not a Redis connection".to_string()), } } pub async fn redis_zrem_in_db_core( state: &AppState, connection_id: &str, db: u32, key_raw: &str, member: &str, ) -> Result<(), String> { ensure_redis_pool(state, connection_id).await?; let connections = state.connections.read().await; match connections.get(connection_id).ok_or("Not found")? { PoolKind::Redis(redis) => { let key = redis_driver::redis_key_raw_to_bytes(key_raw)?; match redis { RedisConnection::Direct(con) => { let mut con = con.lock().await; redis_driver::select_db(&mut *con, db).await?; redis_driver::zrem(&mut *con, &key, member).await } RedisConnection::Cluster(cluster) => { redis_driver::ensure_cluster_db(db)?; let mut con = redis_driver::cluster_key_connection(cluster, &key).await?; redis_driver::zrem(&mut con, &key, member).await } } } _ => Err("Not a Redis connection".to_string()), } } pub async fn redis_stream_add_in_db_core( state: &AppState, connection_id: &str, db: u32, key_raw: &str, entry_id: &str, fields: Vec<(String, String)>, ttl: Option, ) -> Result<(), String> { ensure_redis_pool(state, connection_id).await?; let connections = state.connections.read().await; match connections.get(connection_id).ok_or("Not found")? { PoolKind::Redis(redis) => { let key = redis_driver::redis_key_raw_to_bytes(key_raw)?; match redis { RedisConnection::Direct(con) => { let mut con = con.lock().await; redis_driver::select_db(&mut *con, db).await?; redis_driver::stream_add(&mut *con, &key, entry_id, &fields, ttl).await } RedisConnection::Cluster(cluster) => { redis_driver::ensure_cluster_db(db)?; let mut con = redis_driver::cluster_key_connection(cluster, &key).await?; redis_driver::stream_add(&mut con, &key, entry_id, &fields, ttl).await } } } _ => Err("Not a Redis connection".to_string()), } } pub async fn redis_json_set_in_db_core( state: &AppState, connection_id: &str, db: u32, key_raw: &str, value: &str, ttl: Option, ) -> Result<(), String> { ensure_redis_pool(state, connection_id).await?; let connections = state.connections.read().await; match connections.get(connection_id).ok_or("Not found")? { PoolKind::Redis(redis) => { let key = redis_driver::redis_key_raw_to_bytes(key_raw)?; match redis { RedisConnection::Direct(con) => { let mut con = con.lock().await; redis_driver::select_db(&mut *con, db).await?; redis_driver::json_set(&mut *con, &key, value, ttl).await } RedisConnection::Cluster(cluster) => { redis_driver::ensure_cluster_db(db)?; let mut con = redis_driver::cluster_key_connection(cluster, &key).await?; redis_driver::json_set(&mut con, &key, value, ttl).await } } } _ => Err("Not a Redis connection".to_string()), } } pub async fn redis_check_json_module_in_db_core( state: &AppState, connection_id: &str, db: u32, ) -> Result { ensure_redis_pool(state, connection_id).await?; let connections = state.connections.read().await; match connections.get(connection_id).ok_or("Not found")? { PoolKind::Redis(redis) => match redis { RedisConnection::Direct(con) => { let mut con = con.lock().await; redis_driver::select_db(&mut *con, db).await?; redis_driver::check_json_module(&mut *con).await } RedisConnection::Cluster(cluster) => { redis_driver::ensure_cluster_db(db)?; let mut con = redis_driver::cluster_any_connection(cluster).await?; redis_driver::check_json_module(&mut con).await } }, _ => Err("Not a Redis connection".to_string()), } } pub async fn redis_set_ttl_in_db_core( state: &AppState, connection_id: &str, db: u32, key_raw: &str, ttl: i64, ) -> Result<(), String> { ensure_redis_pool(state, connection_id).await?; let connections = state.connections.read().await; match connections.get(connection_id).ok_or("Not found")? { PoolKind::Redis(redis) => { let key = redis_driver::redis_key_raw_to_bytes(key_raw)?; match redis { RedisConnection::Direct(con) => { let mut con = con.lock().await; redis_driver::select_db(&mut *con, db).await?; redis_driver::set_ttl(&mut *con, &key, ttl).await } RedisConnection::Cluster(cluster) => { redis_driver::ensure_cluster_db(db)?; let mut con = redis_driver::cluster_key_connection(cluster, &key).await?; redis_driver::set_ttl(&mut con, &key, ttl).await } } } _ => Err("Not a Redis connection".to_string()), } } pub async fn redis_delete_keys_in_db_core( state: &AppState, connection_id: &str, db: u32, key_raws: &[String], ) -> Result { ensure_redis_pool(state, connection_id).await?; let connections = state.connections.read().await; match connections.get(connection_id).ok_or("Not found")? { PoolKind::Redis(redis) => { let keys: Result>, String> = key_raws.iter().map(|k| redis_driver::redis_key_raw_to_bytes(k)).collect(); let keys = keys?; match redis { RedisConnection::Direct(con) => { let mut con = con.lock().await; redis_driver::select_db(&mut *con, db).await?; redis_driver::delete_keys(&mut *con, &keys).await } RedisConnection::Cluster(cluster) => { redis_driver::ensure_cluster_db(db)?; let mut deleted = 0; for key in &keys { let mut con = redis_driver::cluster_key_connection(cluster, key).await?; deleted += redis_driver::delete_keys(&mut con, std::slice::from_ref(key)).await?; } Ok(deleted) } } } _ => Err("Not a Redis connection".to_string()), } } pub async fn redis_flush_db_core(state: &AppState, connection_id: &str, db: u32) -> Result<(), String> { ensure_redis_pool(state, connection_id).await?; let connections = state.connections.read().await; match connections.get(connection_id).ok_or("Not found")? { PoolKind::Redis(redis) => match redis { RedisConnection::Direct(con) => { let mut con = con.lock().await; redis_driver::select_db(&mut *con, db).await?; redis_driver::flush_db(&mut *con).await } RedisConnection::Cluster(cluster) => { redis_driver::ensure_cluster_db(db)?; redis_driver::flush_cluster(cluster).await } }, _ => Err("Not a Redis connection".to_string()), } } pub async fn redis_execute_command_core( state: &AppState, connection_id: &str, db: u32, command: &str, skip_safety_check: bool, ) -> Result { ensure_redis_pool(state, connection_id).await?; let connections = state.connections.read().await; match connections.get(connection_id).ok_or("Not found")? { PoolKind::Redis(redis) => match redis { RedisConnection::Direct(con) => { let mut con = con.lock().await; redis_driver::select_db(&mut *con, db).await?; redis_driver::execute_command(&mut *con, command, skip_safety_check).await } RedisConnection::Cluster(cluster) => { redis_driver::ensure_cluster_db(db)?; let mut con = if let Ok(argv) = redis_driver::parse_command_argv(command) { if let Some(command_name) = argv.first() { if command_name.eq_ignore_ascii_case("SELECT") { return Err("Redis Cluster only supports db0; SELECT is not available".to_string()); } } if command_may_target_first_key(&argv) { redis_driver::cluster_key_connection(cluster, argv[1].as_bytes()).await? } else { redis_driver::cluster_any_connection(cluster).await? } } else { redis_driver::cluster_any_connection(cluster).await? }; redis_driver::execute_command(&mut con, command, skip_safety_check).await } }, _ => Err("Not a Redis connection".to_string()), } } fn command_may_target_first_key(argv: &[String]) -> bool { if argv.len() < 2 { return false; } match argv[0].to_ascii_uppercase().as_str() { "PING" | "INFO" | "DBSIZE" | "TIME" | "ROLE" | "CLUSTER" | "CLIENT" | "COMMAND" | "HELLO" | "AUTH" | "QUIT" => { false } _ => true, } } pub async fn redis_load_more_in_db_core( state: &AppState, connection_id: &str, db: u32, key_raw: &str, key_type: &str, cursor: u64, count: usize, ) -> Result { ensure_redis_pool(state, connection_id).await?; let connections = state.connections.read().await; match connections.get(connection_id).ok_or("Not found")? { PoolKind::Redis(redis) => { let key = redis_driver::redis_key_raw_to_bytes(key_raw)?; match redis { RedisConnection::Direct(con) => { let mut con = con.lock().await; redis_driver::select_db(&mut *con, db).await?; redis_driver::load_more_collection(&mut *con, &key, key_type, cursor, count).await } RedisConnection::Cluster(cluster) => { redis_driver::ensure_cluster_db(db)?; let mut con = redis_driver::cluster_key_connection(cluster, &key).await?; redis_driver::load_more_collection(&mut con, &key, key_type, cursor, count).await } } } _ => Err("Not a Redis connection".to_string()), } } pub async fn redis_publish_core( state: &AppState, connection_id: &str, db: u32, channel: &str, message: &str, ) -> Result { ensure_redis_pool(state, connection_id).await?; let connections = state.connections.read().await; match connections.get(connection_id).ok_or("Not found")? { PoolKind::Redis(redis) => match redis { RedisConnection::Direct(con) => { let mut con = con.lock().await; redis_driver::select_db(&mut *con, db).await?; redis_driver::publish_message(&mut *con, channel, message).await } RedisConnection::Cluster(cluster) => { redis_driver::ensure_cluster_db(db)?; let mut con = redis_driver::cluster_any_connection(cluster).await?; redis_driver::publish_message(&mut con, channel, message).await } }, _ => Err("Not a Redis connection".to_string()), } } pub async fn redis_create_pubsub_core(state: &AppState, connection_id: &str) -> Result { let configs = state.configs.read().await; let config = configs.get(connection_id).ok_or("Connection config not found")?.clone(); drop(configs); if config.db_type != crate::models::connection::DatabaseType::Redis { return Err("Not a Redis connection".to_string()); } let (host, port) = state.connection_host_port(connection_id, &config).await?; let timeout = std::time::Duration::from_secs(config.effective_connect_timeout_secs()); redis_driver::connect_pubsub(&config, &host, port, timeout).await } pub async fn redis_slowlog_get_core( state: &AppState, connection_id: &str, count: usize, node_host: Option, node_port: Option, ) -> Result, String> { ensure_redis_pool(state, connection_id).await?; let connections = state.connections.read().await; match connections.get(connection_id).ok_or("Not found")? { PoolKind::Redis(redis) => match redis { RedisConnection::Direct(con) => { let mut con = con.lock().await; // SLOWLOG is a server-level command, no select_db needed redis_driver::get_slowlog(&mut *con, count).await } RedisConnection::Cluster(cluster) => { if let (Some(host), Some(port)) = (node_host.as_ref(), node_port) { let endpoint = redis_driver::RedisNodeEndpoint { host: host.clone(), port }; let mut con = redis_driver::connect_cluster_node(cluster, &endpoint).await?; redis_driver::get_slowlog(&mut con, count).await } else { // No node specified — return empty (frontend enforces selection) Ok(Vec::new()) } } }, _ => Err("Not a Redis connection".to_string()), } } pub async fn redis_cluster_master_nodes_core( state: &AppState, connection_id: &str, ) -> Result, String> { ensure_redis_pool(state, connection_id).await?; let connections = state.connections.read().await; match connections.get(connection_id).ok_or("Not found")? { PoolKind::Redis(redis) => match redis { RedisConnection::Cluster(cluster) => redis_driver::cluster_master_nodes(cluster).await, _ => Ok(Vec::new()), }, _ => Err("Not a Redis connection".to_string()), } }