use anyhow::{bail, Result};
use redis::{Client, Commands};
use std::process::Command;
use std::time::Duration;

/// Cache key prefix for pod SSH enablement status.
const POD_SSH_CACHE_KEY_PREFIX: &str = "pod_ssh_enabled:";

/// Configuration for connecting to a Kubernetes pod via SSH.
#[derive(Debug, Clone)]
pub struct PodConnectionConfig {
    /// Kubernetes namespace
    pub namespace: String,
    /// Pod name
    pub pod_name: String,
    /// SSH user in the pod (default: root)
    pub user: String,
    /// SSH port in the pod (default: 22)
    pub port: u16,
    /// Redis connection string for caching (optional)
    pub redis_url: Option<String>,
    /// Cache TTL in seconds (default: 3600)
    pub cache_ttl: usize,
}

impl PodConnectionConfig {
    /// Creates a new pod connection configuration.
    pub fn new<S: Into<String>>(namespace: S, pod_name: S) -> Self {
        Self {
            namespace: namespace.into(),
            pod_name: pod_name.into(),
            user: "root".to_string(),
            port: 22,
            redis_url: None,
            cache_ttl: 3600,
        }
    }

    /// Sets the SSH user in the pod.
    pub fn user<S: Into<String>>(mut self, user: S) -> Self {
        self.user = user.into();
        self
    }

    /// Sets the SSH port in the pod.
    pub fn port(mut self, port: u16) -> Self {
        self.port = port;
        self
    }

    /// Sets the Redis connection URL for caching pod SSH status.
    pub fn redis_url<S: Into<String>>(mut self, url: S) -> Self {
        self.redis_url = Some(url.into());
        self
    }

    /// Sets the cache TTL in seconds.
    pub fn cache_ttl(mut self, ttl: usize) -> Self {
        self.cache_ttl = ttl;
        self
    }

    /// Gets the pod identifier for caching purposes.
    pub fn pod_id(&self) -> String {
        format!("{}:{}", self.namespace, self.pod_name)
    }

    /// Gets the cache key for this pod's SSH enablement status.
    pub fn cache_key(&self) -> String {
        format!("{}{}", POD_SSH_CACHE_KEY_PREFIX, self.pod_id())
    }

    /// Checks if SSH is already enabled in the pod (uses cache if available).
    ///
    /// Returns:
    /// - `Ok(true)`: SSH is enabled in the pod
    /// - `Ok(false)`: SSH is not enabled
    /// - `Err`: If there's an error checking the pod
    pub fn is_ssh_enabled(&self) -> Result<bool> {
        // Try to get from Redis cache first
        if let Some(ref redis_url) = self.redis_url {
            if let Ok(is_enabled) = self.check_redis_cache(redis_url) {
                return Ok(is_enabled);
            }
        }

        // If not in cache or no Redis, check directly
        self.check_pod_ssh_direct()
    }

    /// Checks the Redis cache for pod SSH enablement status.
    fn check_redis_cache(&self, redis_url: &str) -> Result<bool> {
        let client = Client::open(redis_url.to_string())
            .map_err(|e| anyhow::anyhow!("Failed to create Redis client: {}", e))?;
        let mut conn = client
            .get_connection()
            .map_err(|e| anyhow::anyhow!("Failed to connect to Redis: {}", e))?;

        let cache_key = self.cache_key();

        // Try to get the cached value
        let cached_value: Option<String> = conn
            .get(&cache_key)
            .map_err(|e| anyhow::anyhow!("Failed to get cache value: {}", e))?;

        match cached_value {
            Some(value) => {
                // Return true if cached value is "1"
                Ok(value == "1")
            }
            None => {
                // Cache miss - return error to trigger direct check
                bail!("Cache miss")
            }
        }
    }

    /// Checks if SSH is enabled in the pod directly via kubectl.
    fn check_pod_ssh_direct(&self) -> Result<bool> {
        // Try to execute a simple SSH command in the pod
        let check_cmd = format!(
            "kubectl exec -n {} {} -- which sshd",
            self.namespace, self.pod_name
        );

        let output = Command::new("sh")
            .arg("-c")
            .arg(&check_cmd)
            .output()?;

        Ok(output.status.success())
    }

    /// Marks the pod as SSH-enabled in the cache.
    pub fn mark_ssh_enabled(&self) -> Result<()> {
        if let Some(ref redis_url) = self.redis_url {
            self.set_redis_cache(redis_url, true)?;
        }
        Ok(())
    }

    /// Marks the pod as SSH-disabled in the cache.
    pub fn mark_ssh_disabled(&self) -> Result<()> {
        if let Some(ref redis_url) = self.redis_url {
            self.set_redis_cache(redis_url, false)?;
        }
        Ok(())
    }

    /// Clears the SSH enablement status from cache.
    pub fn clear_cache(&self) -> Result<()> {
        if let Some(ref redis_url) = self.redis_url {
            self.delete_redis_cache(redis_url)?;
        }
        Ok(())
    }

    /// Sets the Redis cache value.
    fn set_redis_cache(&self, redis_url: &str, value: bool) -> Result<()> {
        let client = Client::open(redis_url.to_string())
            .map_err(|e| anyhow::anyhow!("Failed to create Redis client: {}", e))?;
        let mut conn = client
            .get_connection()
            .map_err(|e| anyhow::anyhow!("Failed to connect to Redis: {}", e))?;

        let cache_key = self.cache_key();
        let cache_value = if value { "1" } else { "0" };
        let ttl: u64 = self.cache_ttl as u64;

        // Set the cache key with TTL
        let _: () = conn.set_ex(&cache_key, cache_value, ttl)
            .map_err(|e| anyhow::anyhow!("Failed to set cache value: {}", e))?;

        Ok(())
    }

    /// Deletes the Redis cache value.
    fn delete_redis_cache(&self, redis_url: &str) -> Result<()> {
        let client = Client::open(redis_url.to_string())
            .map_err(|e| anyhow::anyhow!("Failed to create Redis client: {}", e))?;
        let mut conn = client
            .get_connection()
            .map_err(|e| anyhow::anyhow!("Failed to connect to Redis: {}", e))?;

        let cache_key = self.cache_key();

        let _: () = conn.del(&cache_key)
            .map_err(|e| anyhow::anyhow!("Failed to delete cache value: {}", e))?;

        Ok(())
    }

    /// Gets the cache value from Redis (useful for debugging).
    pub fn get_cache_value(&self) -> Result<Option<bool>> {
        if let Some(ref redis_url) = self.redis_url {
            let client = Client::open(redis_url.clone())
                .map_err(|e| anyhow::anyhow!("Failed to create Redis client: {}", e))?;
            let mut conn = client
                .get_connection()
                .map_err(|e| anyhow::anyhow!("Failed to connect to Redis: {}", e))?;

            let cache_key = self.cache_key();
            let cached_value: Option<String> = conn
                .get(&cache_key)
                .map_err(|e| anyhow::anyhow!("Failed to get cache value: {}", e))?;

            match cached_value {
                Some(value) => Ok(Some(value == "1")),
                None => Ok(None),
            }
        } else {
            Ok(None)
        }
    }
}

/// Gets the pod's SSH host address for kubectl port-forward.
///
/// This is typically in the form of "localhost:{forwarded_port}" when using
/// kubectl port-forward to tunnel SSH connections.
pub struct PodSshAddress {
    /// The host for kubectl exec (pod name)
    pub pod_host: String,
    /// The namespace
    pub namespace: String,
    /// Local port if using port-forward
    pub local_port: Option<u16>,
}

impl PodSshAddress {
    /// Creates a new pod SSH address configuration.
    pub fn new<S: Into<String>>(namespace: S, pod_name: S) -> Self {
        Self {
            pod_host: pod_name.into(),
            namespace: namespace.into(),
            local_port: None,
        }
    }

    /// Sets up a kubectl port-forward tunnel to the pod.
    ///
    /// # Arguments
    ///
    /// * `local_port` - Local port to forward to (if None, finds a free port)
    ///
    /// # Returns
    ///
    /// Returns the local port that was forwarded.
    pub fn setup_port_forward(&mut self, local_port: Option<u16>) -> Result<u16> {
        let target_port = local_port.unwrap_or(2222);

        let forward_cmd = format!(
            "kubectl port-forward -n {} {} {}:22 &",
            self.namespace, self.pod_host, target_port
        );

        let output = Command::new("sh")
            .arg("-c")
            .arg(&forward_cmd)
            .output()?;

        if !output.status.success() {
            let error = String::from_utf8_lossy(&output.stderr);
            bail!("Failed to set up port-forward: {}", error);
        }

        // Give the port-forward a moment to establish
        std::thread::sleep(Duration::from_millis(500));

        self.local_port = Some(target_port);
        Ok(target_port)
    }

    /// Gets the SSH host address.
    pub fn ssh_host(&self) -> &str {
        "localhost"
    }

    /// Gets the SSH port (local port if port-forward is set up).
    pub fn ssh_port(&self) -> u16 {
        self.local_port.unwrap_or(22)
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_pod_config_creation() {
        let config = PodConnectionConfig::new("default", "test-pod");
        assert_eq!(config.namespace, "default");
        assert_eq!(config.pod_name, "test-pod");
        assert_eq!(config.user, "root");
        assert_eq!(config.port, 22);
    }

    #[test]
    fn test_pod_config_cache_key() {
        let config = PodConnectionConfig::new("my-ns", "my-pod");
        let expected = "pod_ssh_enabled:my-ns:my-pod";
        assert_eq!(config.cache_key(), expected);
    }

    #[test]
    fn test_pod_config_builder() {
        let config = PodConnectionConfig::new("prod", "app-pod")
            .user("ubuntu")
            .port(2222)
            .cache_ttl(7200);

        assert_eq!(config.namespace, "prod");
        assert_eq!(config.pod_name, "app-pod");
        assert_eq!(config.user, "ubuntu");
        assert_eq!(config.port, 2222);
        assert_eq!(config.cache_ttl, 7200);
    }

    #[test]
    fn test_pod_ssh_address() {
        let address = PodSshAddress::new("default", "test-pod");
        assert_eq!(address.ssh_host(), "localhost");
        assert_eq!(address.ssh_port(), 22);
    }
}
