redisctl-mcp 0.4.0

//! Composed Redis diagnostic tools (health_check, key_summary, hotkeys, connection_summary)

use std::collections::HashMap;
use std::sync::Arc;

use schemars::JsonSchema;
use serde::Deserialize;
use tower_mcp::extract::{Json, State};
use tower_mcp::{CallToolResult, McpRouter, ResultExt, Tool, ToolBuilder};

use crate::state::AppState;

/// Build a sub-router containing all diagnostic Redis tools
pub fn router(state: Arc<AppState>) -> McpRouter {
    McpRouter::new()
        .tool(health_check(state.clone()))
        .tool(key_summary(state.clone()))
        .tool(hotkeys(state.clone()))
        .tool(connection_summary(state))
}

// ---------------------------------------------------------------------------
// Helpers
// ---------------------------------------------------------------------------

/// Parse a Redis INFO response into a key-value map.
///
/// The INFO format has `# Section` headers and `key:value` lines.
fn parse_info(info: &str) -> HashMap<String, String> {
    info.lines()
        .filter(|line| !line.starts_with('#') && !line.is_empty())
        .filter_map(|line| {
            let mut parts = line.splitn(2, ':');
            Some((parts.next()?.to_string(), parts.next()?.to_string()))
        })
        .collect()
}

/// Parse a Redis CLIENT LIST response into a vector of field maps.
///
/// Each line contains space-separated `key=value` pairs.
fn parse_client_list(clients: &str) -> Vec<HashMap<String, String>> {
    clients
        .lines()
        .filter(|line| !line.is_empty())
        .map(|line| {
            line.split_whitespace()
                .filter_map(|pair| {
                    let mut parts = pair.splitn(2, '=');
                    Some((parts.next()?.to_string(), parts.next()?.to_string()))
                })
                .collect()
        })
        .collect()
}

/// Format a byte count into a human-readable string.
fn format_bytes(bytes: i64) -> String {
    const KB: f64 = 1024.0;
    const MB: f64 = KB * 1024.0;
    const GB: f64 = MB * 1024.0;

    let b = bytes as f64;
    if b >= GB {
        format!("{:.2} GB", b / GB)
    } else if b >= MB {
        format!("{:.2} MB", b / MB)
    } else if b >= KB {
        format!("{:.2} KB", b / KB)
    } else {
        format!("{} bytes", bytes)
    }
}

// ---------------------------------------------------------------------------
// 1. redis_health_check
// ---------------------------------------------------------------------------

/// Input for redis_health_check
#[derive(Debug, Deserialize, JsonSchema)]
pub struct HealthCheckInput {
    /// Optional Redis URL (overrides profile, uses configured URL if not provided)
    #[serde(default)]
    pub url: Option<String>,
    /// Optional profile name to resolve connection from (uses default profile if not set)
    #[serde(default)]
    pub profile: Option<String>,
}

/// Build the health_check tool
pub fn health_check(state: Arc<AppState>) -> Tool {
    ToolBuilder::new("redis_health_check")
        .description(
            "Comprehensive Redis health check combining PING, INFO (server, memory, stats), \
             and DBSIZE into a single structured summary. Returns connectivity, version, \
             uptime, memory usage, operations rate, and key count.",
        )
        .read_only_safe()
        .extractor_handler_typed::<_, _, _, HealthCheckInput>(
            state,
            |State(state): State<Arc<AppState>>, Json(input): Json<HealthCheckInput>| async move {
                let url = super::resolve_redis_url(input.url, input.profile.as_deref(), &state)?;

                let client = redis::Client::open(url.as_str()).tool_context("Invalid URL")?;

                let mut conn = client
                    .get_multiplexed_async_connection()
                    .await
                    .tool_context("Connection failed")?;

                // PING
                let ping_response: String = redis::cmd("PING")
                    .query_async(&mut conn)
                    .await
                    .tool_context("PING failed")?;

                // INFO (server + memory + stats combined)
                let info_text: String = redis::cmd("INFO")
                    .query_async(&mut conn)
                    .await
                    .tool_context("INFO failed")?;

                let info = parse_info(&info_text);

                // DBSIZE
                let db_size: i64 = redis::cmd("DBSIZE")
                    .query_async(&mut conn)
                    .await
                    .tool_context("DBSIZE failed")?;

                // Extract fields with fallbacks
                let version = info
                    .get("redis_version")
                    .cloned()
                    .unwrap_or_else(|| "unknown".to_string());
                let uptime_seconds = info
                    .get("uptime_in_seconds")
                    .cloned()
                    .unwrap_or_else(|| "unknown".to_string());
                let uptime_days = info
                    .get("uptime_in_days")
                    .cloned()
                    .unwrap_or_else(|| "unknown".to_string());
                let used_memory_human = info
                    .get("used_memory_human")
                    .cloned()
                    .unwrap_or_else(|| "unknown".to_string());
                let maxmemory = info
                    .get("maxmemory")
                    .cloned()
                    .unwrap_or_else(|| "0".to_string());
                let maxmemory_human = info
                    .get("maxmemory_human")
                    .cloned()
                    .unwrap_or_else(|| "unlimited".to_string());
                let frag_ratio = info
                    .get("mem_fragmentation_ratio")
                    .cloned()
                    .unwrap_or_else(|| "unknown".to_string());
                let ops_per_sec = info
                    .get("instantaneous_ops_per_sec")
                    .cloned()
                    .unwrap_or_else(|| "unknown".to_string());
                let total_commands = info
                    .get("total_commands_processed")
                    .cloned()
                    .unwrap_or_else(|| "unknown".to_string());
                let connected_clients = info
                    .get("connected_clients")
                    .cloned()
                    .unwrap_or_else(|| "unknown".to_string());

                let maxmemory_display = if maxmemory == "0" {
                    "unlimited".to_string()
                } else {
                    maxmemory_human
                };

                let output = format!(
                    "Redis Health Check\n\
                     ==================\n\
                     \n\
                     Connectivity: {}\n\
                     Version: {}\n\
                     Uptime: {} seconds ({} days)\n\
                     \n\
                     Memory:\n\
                     - Used: {}\n\
                     - Max: {}\n\
                     - Fragmentation ratio: {}\n\
                     \n\
                     Stats:\n\
                     - Ops/sec: {}\n\
                     - Total commands processed: {}\n\
                     - Connected clients: {}\n\
                     \n\
                     Keys: {}",
                    ping_response,
                    version,
                    uptime_seconds,
                    uptime_days,
                    used_memory_human,
                    maxmemory_display,
                    frag_ratio,
                    ops_per_sec,
                    total_commands,
                    connected_clients,
                    db_size,
                );

                Ok(CallToolResult::text(output))
            },
        )
        .build()
}

// ---------------------------------------------------------------------------
// 2. redis_key_summary
// ---------------------------------------------------------------------------

/// Input for redis_key_summary
#[derive(Debug, Deserialize, JsonSchema)]
pub struct KeySummaryInput {
    /// Optional Redis URL (overrides profile, uses configured URL if not provided)
    #[serde(default)]
    pub url: Option<String>,
    /// Optional profile name to resolve connection from (uses default profile if not set)
    #[serde(default)]
    pub profile: Option<String>,
    /// Key to inspect
    pub key: String,
}

/// Build the key_summary tool
pub fn key_summary(state: Arc<AppState>) -> Tool {
    ToolBuilder::new("redis_key_summary")
        .description(
            "Get a complete metadata summary for a single key combining TYPE, TTL, \
             MEMORY USAGE, and OBJECT ENCODING into one result. Gracefully handles \
             cases where MEMORY USAGE or OBJECT ENCODING are unavailable.",
        )
        .read_only_safe()
        .extractor_handler_typed::<_, _, _, KeySummaryInput>(
            state,
            |State(state): State<Arc<AppState>>, Json(input): Json<KeySummaryInput>| async move {
                let url = super::resolve_redis_url(input.url, input.profile.as_deref(), &state)?;

                let client = redis::Client::open(url.as_str()).tool_context("Invalid URL")?;

                let mut conn = client
                    .get_multiplexed_async_connection()
                    .await
                    .tool_context("Connection failed")?;

                // TYPE
                let key_type: String = redis::cmd("TYPE")
                    .arg(&input.key)
                    .query_async(&mut conn)
                    .await
                    .tool_context("TYPE failed")?;

                if key_type == "none" {
                    return Ok(CallToolResult::text(format!(
                        "Key '{}' does not exist",
                        input.key
                    )));
                }

                // TTL
                let ttl: i64 = redis::cmd("TTL")
                    .arg(&input.key)
                    .query_async(&mut conn)
                    .await
                    .tool_context("TTL failed")?;

                let ttl_display = match ttl {
                    -2 => "key does not exist".to_string(),
                    -1 => "no expiry".to_string(),
                    _ => format!("{} seconds", ttl),
                };

                // MEMORY USAGE (may fail for some key types or Redis versions)
                let memory_display = match redis::cmd("MEMORY")
                    .arg("USAGE")
                    .arg(&input.key)
                    .query_async::<Option<i64>>(&mut conn)
                    .await
                {
                    Ok(Some(bytes)) => format_bytes(bytes),
                    Ok(None) => "unknown".to_string(),
                    Err(_) => "unavailable".to_string(),
                };

                // OBJECT ENCODING (may fail for some key types)
                let encoding_display = match redis::cmd("OBJECT")
                    .arg("ENCODING")
                    .arg(&input.key)
                    .query_async::<Option<String>>(&mut conn)
                    .await
                {
                    Ok(Some(enc)) => enc,
                    Ok(None) => "unknown".to_string(),
                    Err(_) => "unavailable".to_string(),
                };

                let output = format!(
                    "Key Summary: {}\n\
                     =============={}\n\
                     \n\
                     Type: {}\n\
                     TTL: {}\n\
                     Memory: {}\n\
                     Encoding: {}",
                    input.key,
                    "=".repeat(input.key.len()),
                    key_type,
                    ttl_display,
                    memory_display,
                    encoding_display,
                );

                Ok(CallToolResult::text(output))
            },
        )
        .build()
}

// ---------------------------------------------------------------------------
// 3. redis_hotkeys
// ---------------------------------------------------------------------------

/// Maximum allowed sample size to prevent runaway scans.
const MAX_SAMPLE_SIZE: usize = 10_000;

/// Number of top keys to return by memory usage.
const TOP_N: usize = 20;

/// Input for redis_hotkeys
#[derive(Debug, Deserialize, JsonSchema)]
pub struct HotkeysInput {
    /// Optional Redis URL (overrides profile, uses configured URL if not provided)
    #[serde(default)]
    pub url: Option<String>,
    /// Optional profile name to resolve connection from (uses default profile if not set)
    #[serde(default)]
    pub profile: Option<String>,
    /// Key pattern to match (default: "*")
    #[serde(default)]
    pub pattern: Option<String>,
    /// Maximum number of keys to sample (default: 1000, max: 10000)
    #[serde(default)]
    pub sample_size: Option<usize>,
}

/// Build the hotkeys tool
pub fn hotkeys(state: Arc<AppState>) -> Tool {
    ToolBuilder::new("redis_hotkeys")
        .description(
            "Sample keys to identify the largest by memory usage and show type distribution. \
             Uses SCAN to iterate keys, then TYPE and MEMORY USAGE on each sampled key. \
             Returns top 20 keys by memory, type counts, and total memory sampled. \
             Capped at sample_size (default 1000, max 10000) to limit impact.",
        )
        .read_only_safe()
        .extractor_handler_typed::<_, _, _, HotkeysInput>(
            state,
            |State(state): State<Arc<AppState>>, Json(input): Json<HotkeysInput>| async move {
                let url = super::resolve_redis_url(input.url, input.profile.as_deref(), &state)?;

                let client = redis::Client::open(url.as_str()).tool_context("Invalid URL")?;

                let mut conn = client
                    .get_multiplexed_async_connection()
                    .await
                    .tool_context("Connection failed")?;

                let pattern = input.pattern.as_deref().unwrap_or("*");
                let sample_size = input.sample_size.unwrap_or(1000).min(MAX_SAMPLE_SIZE);

                // SCAN to collect keys
                let mut cursor: u64 = 0;
                let mut scanned_keys: Vec<String> = Vec::new();

                loop {
                    let (new_cursor, keys): (u64, Vec<String>) = redis::cmd("SCAN")
                        .arg(cursor)
                        .arg("MATCH")
                        .arg(pattern)
                        .arg("COUNT")
                        .arg(100)
                        .query_async(&mut conn)
                        .await
                        .tool_context("SCAN failed")?;

                    scanned_keys.extend(keys);
                    cursor = new_cursor;

                    if cursor == 0 || scanned_keys.len() >= sample_size {
                        break;
                    }
                }

                scanned_keys.truncate(sample_size);

                if scanned_keys.is_empty() {
                    return Ok(CallToolResult::text(format!(
                        "No keys found matching pattern '{}'",
                        pattern
                    )));
                }

                // Collect TYPE and MEMORY USAGE for each key
                let mut type_counts: HashMap<String, usize> = HashMap::new();
                let mut key_sizes: Vec<(String, i64, String)> = Vec::new();
                let mut total_memory: i64 = 0;

                for key in &scanned_keys {
                    // TYPE
                    let key_type: String =
                        match redis::cmd("TYPE").arg(key).query_async(&mut conn).await {
                            Ok(t) => t,
                            Err(_) => continue,
                        };

                    *type_counts.entry(key_type.clone()).or_insert(0) += 1;

                    // MEMORY USAGE -- may return None or fail
                    let mem_bytes: Option<i64> = redis::cmd("MEMORY")
                        .arg("USAGE")
                        .arg(key)
                        .query_async(&mut conn)
                        .await
                        .unwrap_or_default();

                    if let Some(bytes) = mem_bytes {
                        total_memory += bytes;
                        key_sizes.push((key.clone(), bytes, key_type));
                    }
                }

                // Sort by memory descending and take top N
                key_sizes.sort_by(|a, b| b.1.cmp(&a.1));
                key_sizes.truncate(TOP_N);

                // Build output
                let mut output = format!(
                    "Redis Hotkeys Analysis\n\
                     ======================\n\
                     \n\
                     Keys scanned: {}\n\
                     Total memory sampled: {}\n\
                     \n\
                     Type Distribution:\n",
                    scanned_keys.len(),
                    format_bytes(total_memory),
                );

                let mut type_list: Vec<_> = type_counts.iter().collect();
                type_list.sort_by(|a, b| b.1.cmp(a.1));
                for (t, count) in &type_list {
                    output.push_str(&format!("  {}: {}\n", t, count));
                }

                output.push_str(&format!(
                    "\nTop {} Keys by Memory:\n",
                    key_sizes.len().min(TOP_N)
                ));

                for (i, (key, bytes, key_type)) in key_sizes.iter().enumerate() {
                    output.push_str(&format!(
                        "  {}. {} ({}) - {}\n",
                        i + 1,
                        key,
                        key_type,
                        format_bytes(*bytes),
                    ));
                }

                Ok(CallToolResult::text(output))
            },
        )
        .build()
}

// ---------------------------------------------------------------------------
// 4. redis_connection_summary
// ---------------------------------------------------------------------------

/// Input for redis_connection_summary
#[derive(Debug, Deserialize, JsonSchema)]
pub struct ConnectionSummaryInput {
    /// Optional Redis URL (overrides profile, uses configured URL if not provided)
    #[serde(default)]
    pub url: Option<String>,
    /// Optional profile name to resolve connection from (uses default profile if not set)
    #[serde(default)]
    pub profile: Option<String>,
}

/// Build the connection_summary tool
pub fn connection_summary(state: Arc<AppState>) -> Tool {
    ToolBuilder::new("redis_connection_summary")
        .description(
            "Analyze client connections by combining CLIENT LIST and INFO clients. \
             Returns total connections, connections by source IP (top 10), idle \
             connections (>60s), blocked client count, and oldest connection age.",
        )
        .read_only_safe()
        .extractor_handler_typed::<_, _, _, ConnectionSummaryInput>(
            state,
            |State(state): State<Arc<AppState>>,
             Json(input): Json<ConnectionSummaryInput>| async move {
                let url = super::resolve_redis_url(input.url, input.profile.as_deref(), &state)?;

                let client = redis::Client::open(url.as_str())
                    .tool_context("Invalid URL")?;

                let mut conn = client
                    .get_multiplexed_async_connection()
                    .await
                    .tool_context("Connection failed")?;

                // CLIENT LIST
                let client_list_raw: String = redis::cmd("CLIENT")
                    .arg("LIST")
                    .query_async(&mut conn)
                    .await
                    .tool_context("CLIENT LIST failed")?;

                // INFO clients section
                let info_text: String = redis::cmd("INFO")
                    .arg("clients")
                    .query_async(&mut conn)
                    .await
                    .tool_context("INFO clients failed")?;

                let info = parse_info(&info_text);
                let clients = parse_client_list(&client_list_raw);

                let total = clients.len();

                // Connections by source IP
                let mut ip_counts: HashMap<String, usize> = HashMap::new();
                for c in &clients {
                    if let Some(addr) = c.get("addr") {
                        // addr is "ip:port" -- extract just IP
                        let ip = addr
                            .rsplit_once(':')
                            .map(|(ip, _)| ip.to_string())
                            .unwrap_or_else(|| addr.clone());
                        *ip_counts.entry(ip).or_insert(0) += 1;
                    }
                }
                let mut ip_list: Vec<_> = ip_counts.into_iter().collect();
                ip_list.sort_by(|a, b| b.1.cmp(&a.1));
                ip_list.truncate(10);

                // Idle connections (idle > 60s)
                let idle_count = clients
                    .iter()
                    .filter(|c| {
                        c.get("idle")
                            .and_then(|v| v.parse::<u64>().ok())
                            .is_some_and(|idle| idle > 60)
                    })
                    .count();

                // Blocked clients from INFO
                let blocked_clients = info
                    .get("blocked_clients")
                    .cloned()
                    .unwrap_or_else(|| "unknown".to_string());

                // Oldest connection age
                let oldest_age = clients
                    .iter()
                    .filter_map(|c| c.get("age").and_then(|v| v.parse::<u64>().ok()))
                    .max();

                let oldest_display = match oldest_age {
                    Some(age) => {
                        let days = age / 86400;
                        let hours = (age % 86400) / 3600;
                        let minutes = (age % 3600) / 60;
                        let secs = age % 60;
                        if days > 0 {
                            format!("{}d {}h {}m {}s ({} seconds)", days, hours, minutes, secs, age)
                        } else if hours > 0 {
                            format!("{}h {}m {}s ({} seconds)", hours, minutes, secs, age)
                        } else if minutes > 0 {
                            format!("{}m {}s ({} seconds)", minutes, secs, age)
                        } else {
                            format!("{} seconds", age)
                        }
                    }
                    None => "unknown".to_string(),
                };

                // Build output
                let mut output = format!(
                    "Redis Connection Summary\n\
                     ========================\n\
                     \n\
                     Total connections: {}\n\
                     Blocked clients: {}\n\
                     Idle connections (>60s): {}\n\
                     Oldest connection: {}\n\
                     \n\
                     Connections by IP (top 10):\n",
                    total, blocked_clients, idle_count, oldest_display,
                );

                for (ip, count) in &ip_list {
                    output.push_str(&format!("  {}: {}\n", ip, count));
                }

                Ok(CallToolResult::text(output))
            },
        )
        .build()
}