herolib-virt 0.3.13

//! Kubernetes Manager - Core functionality for namespace-scoped Kubernetes operations
//!
//! This module provides the [`KubernetesManager`] struct, which is the primary entry point
//! for all Kubernetes operations. Each manager instance is scoped to a single namespace
//! and provides methods for managing pods, services, deployments, and other resources.
//!
//! # Features
//!
//! - Namespace-scoped resource management
//! - Pod, Service, Deployment, ConfigMap, and Secret CRUD operations
//! - Pattern-based resource deletion with PCRE regex
//! - Service connection information retrieval
//! - Production safety features (timeouts, retries, rate limiting)
//!
//! # Rust Example
//!
//! ```rust,no_run
//! use herolib_system::kubernetes::{KubernetesManager, KubernetesConfig};
//!
//! #[tokio::main]
//! async fn main() -> Result<(), Box<dyn std::error::Error>> {
//!     // Create a manager for the "default" namespace
//!     let km = KubernetesManager::new("default").await?;
//!
//!     // Or with custom configuration
//!     let config = KubernetesConfig::default();
//!     let km = KubernetesManager::with_config("my-namespace", config).await?;
//!
//!     // List all pods
//!     let pods = km.pods_list().await?;
//!     println!("Found {} pods", pods.len());
//!
//!     // Create a namespace
//!     km.namespace_create("new-namespace").await?;
//!
//!     // Create a pod
//!     let labels = std::collections::HashMap::from([
//!         ("app".to_string(), "my-app".to_string()),
//!     ]);
//!     km.pod_create("my-pod", "nginx:alpine", Some(labels), None).await?;
//!
//!     // Delete resources matching a pattern
//!     let deleted = km.delete("test-.*").await?;
//!     println!("Deleted {} resources", deleted);
//!
//!     Ok(())
//! }
//! ```
//!
//! # Rhai Scripting Examples
//!
//! The KubernetesManager is fully accessible from Rhai scripts via `herodo`.
//!
//! ## Creating a Manager
//!
//! ```rhai
//! // Create a manager for a namespace
//! let km = kubernetes_manager_new("my-namespace");
//!
//! // Get the namespace this manager operates on
//! let ns = km.namespace();
//! print("Operating on namespace: " + ns);
//! ```
//!
//! ## Listing Resources
//!
//! ```rhai
//! let km = kubernetes_manager_new("default");
//!
//! // List all pods
//! let pods = km.pods_list();
//! for pod in pods {
//!     print("Pod: " + pod);
//! }
//!
//! // List other resources
//! let services = km.services_list();
//! let deployments = km.deployments_list();
//! let configmaps = km.configmaps_list();
//! let secrets = km.secrets_list();
//! let namespaces = km.namespaces_list();
//! ```
//!
//! ## Creating Resources
//!
//! ```rhai
//! let km = kubernetes_manager_new("my-namespace");
//!
//! // Create a pod with labels
//! let labels = #{
//!     "app": "my-app",
//!     "environment": "development"
//! };
//! km.create_pod("my-pod", "nginx:alpine", labels);
//!
//! // Create a pod with environment variables
//! let env_vars = #{
//!     "MY_VAR": "my-value",
//!     "DEBUG": "true"
//! };
//! km.create_pod_with_env("my-pod-env", "nginx:alpine", labels, env_vars);
//!
//! // Create a service
//! let selector = #{ "app": "my-app" };
//! km.create_service("my-service", selector, 80, 8080);
//!
//! // Create a deployment
//! km.create_deployment("my-deployment", "nginx:alpine", 3, labels, env_vars);
//!
//! // Create a ConfigMap
//! let config_data = #{
//!     "config.json": `{"key": "value"}`,
//!     "settings.yaml": "debug: true"
//! };
//! km.create_configmap("my-config", config_data);
//!
//! // Create a Secret
//! let secret_data = #{
//!     "username": "admin",
//!     "password": "secret123"
//! };
//! km.create_secret("my-secret", secret_data, "Opaque");
//! ```
//!
//! ## Getting Resources
//!
//! ```rhai
//! let km = kubernetes_manager_new("my-namespace");
//!
//! let pod = km.get_pod("my-pod");
//! let service = km.get_service("my-service");
//! let deployment = km.get_deployment("my-deployment");
//! ```
//!
//! ## Deleting Resources
//!
//! ```rhai
//! let km = kubernetes_manager_new("my-namespace");
//!
//! // Delete specific resources
//! km.delete_pod("my-pod");
//! km.delete_service("my-service");
//! km.delete_deployment("my-deployment");
//! km.delete_configmap("my-config");
//! km.delete_secret("my-secret");
//!
//! // Delete resources matching a PCRE pattern
//! let deleted_count = km.delete("test-.*");
//! print("Deleted " + deleted_count + " resources");
//! ```
//!
//! ## Namespace Management
//!
//! ```rhai
//! let km = kubernetes_manager_new("default");
//!
//! // Create a namespace (idempotent)
//! km.create_namespace("my-new-namespace");
//!
//! // Check if namespace exists
//! if km.namespace_exists("my-new-namespace") {
//!     print("Namespace exists!");
//! }
//!
//! // Reset a namespace (delete and recreate)
//! km.reset_namespace("my-namespace");
//!
//! // Delete a namespace
//! km.delete_namespace("my-new-namespace");
//! ```
//!
//! ## Service Connection Info
//!
//! ```rhai
//! let km = kubernetes_manager_new("my-namespace");
//!
//! // Get service connection info
//! let info = km.service_connection_info("my-service");
//!
//! print("Service: " + info.name);
//! print("Namespace: " + info.namespace);
//! print("Hostname: " + info.hostname);
//! print("Cluster IP: " + info.cluster_ip);
//! print("Service Type: " + info.service_type);
//!
//! // Get connection strings
//! print("Connection: " + info.connection_string());
//! print("Host Connection: " + info.host_connection_string());
//! print("Localhost: " + info.localhost_connection_string());
//! ```
//!
//! ## Resource Counts and Application Deployment
//!
//! ```rhai
//! let km = kubernetes_manager_new("my-namespace");
//!
//! // Get resource counts
//! let counts = km.resource_counts();
//! print("Pods: " + counts["pods"]);
//! print("Services: " + counts["services"]);
//! print("Deployments: " + counts["deployments"]);
//!
//! // Deploy a complete application (creates deployment + service)
//! let labels = #{ "app": "my-app", "version": "1.0" };
//! let env = #{ "DATABASE_URL": "postgres://localhost:5432/mydb" };
//! km.deploy_application("my-app", "my-image:latest", 3, 8080, labels, env);
//! ```
//!
//! ## Error Handling
//!
//! ```rhai
//! try {
//!     let km = kubernetes_manager_new("my-namespace");
//!     let pods = km.pods_list();
//! } catch(error) {
//!     print("Kubernetes error: " + error);
//! }
//! ```

use super::config::KubernetesConfig;
use super::error::{KubernetesError, KubernetesResult};
use base64::Engine;
use k8s_openapi::api::apps::v1::Deployment;
use k8s_openapi::api::core::v1::{ConfigMap, Namespace, Node, Pod, Secret, Service};
use k8s_openapi::apimachinery::pkg::apis::meta::v1::ObjectMeta;
use kube::config::{KubeConfigOptions, Kubeconfig};
use kube::{Api, Client, Config};
use regex::Regex;
use std::collections::HashMap;
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::Semaphore;
use tokio::time::timeout;
use tokio_retry::Retry;
use tokio_retry::strategy::ExponentialBackoff;

/// Core Kubernetes namespaces that should never be deleted
const PROTECTED_NAMESPACES: &[&str] = &[
    "default",
    "kube-system",
    "kube-public",
    "kube-node-lease",
    "kube-apiserver",
];

/// KubernetesManager provides namespace-scoped operations for Kubernetes resources
///
/// Each instance operates on a single namespace and provides methods for
/// managing pods, services, deployments, and other Kubernetes resources.
///
/// Includes production safety features:
/// - Configurable timeouts for all operations
/// - Exponential backoff retry logic for transient failures
/// - Rate limiting to prevent API overload
#[derive(Clone)]
pub struct KubernetesManager {
    /// Kubernetes client
    client: Client,
    /// Target namespace for operations
    namespace: String,
    /// Configuration for production safety features
    config: KubernetesConfig,
    /// Semaphore for rate limiting API calls
    rate_limiter: Arc<Semaphore>,
    /// Last request time for rate limiting
    last_request: Arc<tokio::sync::Mutex<Instant>>,
}

impl KubernetesManager {
    /// Create a new KubernetesManager for the specified namespace with default configuration
    ///
    /// # Arguments
    ///
    /// * `namespace` - The Kubernetes namespace to operate on
    ///
    /// # Returns
    ///
    /// * `KubernetesResult<Self>` - The manager instance or an error
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// use herolib_kubernetes::KubernetesManager;
    ///
    /// #[tokio::main]
    /// async fn main() -> Result<(), Box<dyn std::error::Error>> {
    ///     // This requires a running Kubernetes cluster
    ///     let km = KubernetesManager::new("default").await?;
    ///     Ok(())
    /// }
    /// ```
    pub async fn new(namespace: impl Into<String>) -> KubernetesResult<Self> {
        Self::with_config(namespace, KubernetesConfig::default()).await
    }

    /// Create a new KubernetesManager with custom configuration
    ///
    /// # Arguments
    ///
    /// * `namespace` - The Kubernetes namespace to operate on
    /// * `config` - Configuration for production safety features
    ///
    /// # Returns
    ///
    /// * `KubernetesResult<Self>` - The manager instance or an error
    pub async fn with_config(
        namespace: impl Into<String>,
        config: KubernetesConfig,
    ) -> KubernetesResult<Self> {
        let client = Self::create_kube_client().await?;

        // Validate cluster connectivity
        Self::validate_cluster_connectivity(&client).await?;

        // Create rate limiter semaphore with burst capacity
        let rate_limiter = Arc::new(Semaphore::new(config.rate_limit_burst as usize));
        let last_request = Arc::new(tokio::sync::Mutex::new(Instant::now()));

        Ok(Self {
            client,
            namespace: namespace.into(),
            config,
            rate_limiter,
            last_request,
        })
    }

    /// Create a Kubernetes client, checking multiple config locations
    ///
    /// This function tries the following locations in order:
    /// 1. Default locations (in-cluster, ~/.kube/config, KUBECONFIG env)
    /// 2. k3s config at /etc/rancher/k3s/k3s.yaml
    /// 3. microk8s config at /var/snap/microk8s/current/credentials/client.config
    /// 4. Detection via kubectl (if kubectl works, tries to find the config it's using)
    async fn create_kube_client() -> KubernetesResult<Client> {
        // First try the default locations (in-cluster, ~/.kube/config, KUBECONFIG env)
        if let Ok(k8s_config) = Config::infer().await {
            if let Ok(client) = Client::try_from(k8s_config) {
                return Ok(client);
            }
        }

        // Try k3s config location
        let k3s_path = "/etc/rancher/k3s/k3s.yaml";
        if std::path::Path::new(k3s_path).exists() {
            log::debug!("Trying k3s config at {}", k3s_path);
            if let Ok(kubeconfig) = Kubeconfig::read_from(k3s_path) {
                if let Ok(config) =
                    Config::from_custom_kubeconfig(kubeconfig, &KubeConfigOptions::default()).await
                {
                    if let Ok(client) = Client::try_from(config) {
                        log::info!("Using k3s kubeconfig from {}", k3s_path);
                        return Ok(client);
                    }
                }
            }
        }

        // Try microk8s config location
        let microk8s_path = "/var/snap/microk8s/current/credentials/client.config";
        if std::path::Path::new(microk8s_path).exists() {
            log::debug!("Trying microk8s config at {}", microk8s_path);
            if let Ok(kubeconfig) = Kubeconfig::read_from(microk8s_path) {
                if let Ok(config) =
                    Config::from_custom_kubeconfig(kubeconfig, &KubeConfigOptions::default()).await
                {
                    if let Ok(client) = Client::try_from(config) {
                        log::info!("Using microk8s kubeconfig from {}", microk8s_path);
                        return Ok(client);
                    }
                }
            }
        }

        // Try to detect from kubectl - if kubectl works, try to find its config
        if let Some(kubeconfig_path) = Self::detect_kubeconfig_from_kubectl() {
            log::debug!("Detected kubeconfig at {} (kubectl works)", kubeconfig_path);
            if let Ok(kubeconfig) = Kubeconfig::read_from(&kubeconfig_path) {
                if let Ok(config) =
                    Config::from_custom_kubeconfig(kubeconfig, &KubeConfigOptions::default()).await
                {
                    if let Ok(client) = Client::try_from(config) {
                        log::info!(
                            "Using kubeconfig from {} (detected via kubectl)",
                            kubeconfig_path
                        );
                        return Ok(client);
                    }
                }
            }
        }

        // All attempts failed - return a helpful error
        Err(KubernetesError::config_error(
            "❌ No Kubernetes cluster configuration found!\n\n\
             Tried:\n  \
             - In-cluster config (not running inside Kubernetes)\n  \
             - Kubeconfig file at ~/.kube/config (not found)\n  \
             - KUBECONFIG environment variable (not set or invalid)\n  \
             - k3s config at /etc/rancher/k3s/k3s.yaml\n  \
             - microk8s config at /var/snap/microk8s/current/credentials/client.config\n  \
             - Detection via kubectl\n\n\
             To fix this, either:\n  \
             - Set the KUBECONFIG environment variable to point to your kubeconfig file\n  \
             - Copy your kubeconfig to ~/.kube/config\n  \
             - Ensure the kubeconfig file has correct permissions\n  \
             - Run this command inside a Kubernetes cluster",
        ))
    }

    /// Try to detect kubeconfig path from kubectl
    ///
    /// If kubectl works, this function tries to find the kubeconfig file it's using
    /// by checking common locations.
    fn detect_kubeconfig_from_kubectl() -> Option<String> {
        use std::process::Command;

        // Try to verify kubectl works by checking if it can connect to a cluster
        let output = Command::new("kubectl")
            .args([
                "config",
                "view",
                "--minify",
                "-o",
                "jsonpath={.clusters[0].name}",
            ])
            .output()
            .ok()?;

        if !output.status.success() {
            return None;
        }

        // kubectl works, so let's check common locations
        // First check KUBECONFIG env var
        if let Ok(kubeconfig) = std::env::var("KUBECONFIG") {
            if std::path::Path::new(&kubeconfig).exists() {
                return Some(kubeconfig);
            }
        }

        // Check k3s location
        let k3s_path = "/etc/rancher/k3s/k3s.yaml";
        if std::path::Path::new(k3s_path).exists() {
            return Some(k3s_path.to_string());
        }

        // Check microk8s location
        let microk8s_path = "/var/snap/microk8s/current/credentials/client.config";
        if std::path::Path::new(microk8s_path).exists() {
            return Some(microk8s_path.to_string());
        }

        None
    }

    /// Validate that we can connect to the Kubernetes cluster
    async fn validate_cluster_connectivity(client: &Client) -> KubernetesResult<()> {
        log::info!("🔍 Validating Kubernetes cluster connectivity...");

        // Try to get server version as a connectivity test
        match client.apiserver_version().await {
            Ok(version) => {
                log::info!(
                    "✅ Connected to Kubernetes cluster (version: {})",
                    version.git_version
                );
                Ok(())
            }
            Err(e) => {
                let error_msg = e.to_string();
                if error_msg.contains("connection refused") {
                    Err(KubernetesError::config_error(
                        "❌ Kubernetes cluster is not reachable!\n\n\
                         The cluster appears to be down or unreachable.\n\
                         Try: `kubectl get nodes` to verify connectivity.\n\n\
                         If using minikube: `minikube start`\n\
                         If using kind: `kind create cluster`",
                    ))
                } else if error_msg.contains("Unauthorized") || error_msg.contains("Forbidden") {
                    Err(KubernetesError::permission_denied(
                        "❌ Access denied to Kubernetes cluster!\n\n\
                         You don't have permission to access this cluster.\n\
                         Check your kubeconfig and RBAC permissions.",
                    ))
                } else {
                    Err(KubernetesError::config_error(format!(
                        "❌ Failed to connect to Kubernetes cluster!\n\n\
                         Error: {error_msg}\n\n\
                         Please verify:\n\
                         1. Cluster is running: `kubectl get nodes`\n\
                         2. Network connectivity\n\
                         3. Authentication credentials"
                    )))
                }
            }
        }
    }

    /// Get the namespace this manager operates on
    pub fn namespace(&self) -> &str {
        &self.namespace
    }

    /// Get the Kubernetes client
    pub fn client(&self) -> &Client {
        &self.client
    }

    /// Get the configuration
    pub fn config(&self) -> &KubernetesConfig {
        &self.config
    }

    /// Execute an operation with production safety features (timeout, retry, rate limiting)
    async fn execute_with_safety<F, Fut, T>(&self, operation: F) -> KubernetesResult<T>
    where
        F: Fn() -> Fut + Send + Sync,
        Fut: std::future::Future<Output = KubernetesResult<T>> + Send,
        T: Send,
    {
        // Rate limiting
        self.rate_limit().await?;

        // Retry logic with exponential backoff
        let retry_strategy =
            ExponentialBackoff::from_millis(self.config.retry_base_delay.as_millis() as u64)
                .max_delay(self.config.retry_max_delay)
                .take(self.config.max_retries as usize);

        let result = Retry::spawn(retry_strategy, || async {
            // Apply timeout to the operation
            match timeout(self.config.operation_timeout, operation()).await {
                Ok(result) => result.map_err(|e| {
                    // Only retry on certain types of errors
                    match &e {
                        KubernetesError::ApiError(kube_err) => {
                            // Retry on transient errors
                            if is_retryable_error(kube_err) {
                                log::warn!("Retryable error encountered: {e}");
                                e
                            } else {
                                log::error!("Non-retryable error: {e}");
                                // Convert to a non-retryable error type
                                KubernetesError::operation_error(format!("Non-retryable: {e}"))
                            }
                        }
                        _ => {
                            log::warn!("Retrying operation due to error: {e}");
                            e
                        }
                    }
                }),
                Err(_) => {
                    let timeout_err = KubernetesError::timeout(format!(
                        "Operation timed out after {:?}",
                        self.config.operation_timeout
                    ));
                    log::error!("Operation timeout: {:?}", self.config.operation_timeout);
                    Err(timeout_err)
                }
            }
        })
        .await;

        result
    }

    /// Rate limiting implementation
    async fn rate_limit(&self) -> KubernetesResult<()> {
        // Acquire semaphore permit
        let _permit = self
            .rate_limiter
            .acquire()
            .await
            .map_err(|_| KubernetesError::operation_error("Rate limiter semaphore closed"))?;

        // Enforce minimum time between requests
        let mut last_request = self.last_request.lock().await;
        let now = Instant::now();
        let min_interval = Duration::from_millis(1000 / self.config.rate_limit_rps as u64);

        if let Some(sleep_duration) = min_interval.checked_sub(now.duration_since(*last_request)) {
            tokio::time::sleep(sleep_duration).await;
        }

        *last_request = Instant::now();
        Ok(())
    }

    /// List all pods in the namespace
    ///
    /// # Returns
    ///
    /// * `KubernetesResult<Vec<Pod>>` - List of pods or an error
    pub async fn pods_list(&self) -> KubernetesResult<Vec<Pod>> {
        self.execute_with_safety(|| async {
            let pods: Api<Pod> = Api::namespaced(self.client.clone(), &self.namespace);
            let pod_list = pods.list(&Default::default()).await?;
            Ok(pod_list.items)
        })
        .await
    }

    /// List all services in the namespace
    ///
    /// # Returns
    ///
    /// * `KubernetesResult<Vec<Service>>` - List of services or an error
    pub async fn services_list(&self) -> KubernetesResult<Vec<Service>> {
        self.execute_with_safety(|| async {
            let services: Api<Service> = Api::namespaced(self.client.clone(), &self.namespace);
            let service_list = services.list(&Default::default()).await?;
            Ok(service_list.items)
        })
        .await
    }

    /// List all deployments in the namespace
    ///
    /// # Returns
    ///
    /// * `KubernetesResult<Vec<Deployment>>` - List of deployments or an error
    pub async fn deployments_list(&self) -> KubernetesResult<Vec<Deployment>> {
        let deployments: Api<Deployment> = Api::namespaced(self.client.clone(), &self.namespace);
        let deployment_list = deployments.list(&Default::default()).await?;
        Ok(deployment_list.items)
    }

    /// List all configmaps in the namespace
    ///
    /// # Returns
    ///
    /// * `KubernetesResult<Vec<ConfigMap>>` - List of configmaps or an error
    pub async fn configmaps_list(&self) -> KubernetesResult<Vec<ConfigMap>> {
        let configmaps: Api<ConfigMap> = Api::namespaced(self.client.clone(), &self.namespace);
        let configmap_list = configmaps.list(&Default::default()).await?;
        Ok(configmap_list.items)
    }

    /// List all secrets in the namespace
    ///
    /// # Returns
    ///
    /// * `KubernetesResult<Vec<Secret>>` - List of secrets or an error
    pub async fn secrets_list(&self) -> KubernetesResult<Vec<Secret>> {
        let secrets: Api<Secret> = Api::namespaced(self.client.clone(), &self.namespace);
        let secret_list = secrets.list(&Default::default()).await?;
        Ok(secret_list.items)
    }

    /// Create a ConfigMap
    ///
    /// # Arguments
    ///
    /// * `name` - The name of the ConfigMap
    /// * `data` - Key-value pairs for the ConfigMap data
    ///
    /// # Returns
    ///
    /// * `KubernetesResult<ConfigMap>` - The created ConfigMap or an error
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// use herolib_kubernetes::KubernetesManager;
    /// use std::collections::HashMap;
    ///
    /// #[tokio::main]
    /// async fn main() -> Result<(), Box<dyn std::error::Error>> {
    ///     let km = KubernetesManager::new("default").await?;
    ///
    ///     let mut data = HashMap::new();
    ///     data.insert("config.yaml".to_string(), "key: value".to_string());
    ///     data.insert("app.properties".to_string(), "debug=true".to_string());
    ///
    ///     let configmap = km.configmap_create("my-config", data).await?;
    ///     println!("Created ConfigMap: {}", configmap.metadata.name.unwrap_or_default());
    ///     Ok(())
    /// }
    /// ```
    pub async fn configmap_create(
        &self,
        name: &str,
        data: HashMap<String, String>,
    ) -> KubernetesResult<ConfigMap> {
        let configmaps: Api<ConfigMap> = Api::namespaced(self.client.clone(), &self.namespace);

        let configmap = ConfigMap {
            metadata: ObjectMeta {
                name: Some(name.to_string()),
                namespace: Some(self.namespace.clone()),
                ..Default::default()
            },
            data: Some(data.into_iter().collect()),
            ..Default::default()
        };

        let created_configmap = configmaps.create(&Default::default(), &configmap).await?;
        log::info!("Created ConfigMap '{name}'");
        Ok(created_configmap)
    }

    /// Get a specific ConfigMap by name
    ///
    /// # Arguments
    ///
    /// * `name` - The name of the ConfigMap to retrieve
    ///
    /// # Returns
    ///
    /// * `KubernetesResult<ConfigMap>` - The ConfigMap or an error
    pub async fn configmap_get(&self, name: &str) -> KubernetesResult<ConfigMap> {
        let configmaps: Api<ConfigMap> = Api::namespaced(self.client.clone(), &self.namespace);
        let configmap = configmaps.get(name).await?;
        Ok(configmap)
    }

    /// Create a Secret
    ///
    /// # Arguments
    ///
    /// * `name` - The name of the Secret
    /// * `data` - Key-value pairs for the Secret data (will be base64 encoded)
    /// * `secret_type` - The type of secret (defaults to "Opaque")
    ///
    /// # Returns
    ///
    /// * `KubernetesResult<Secret>` - The created Secret or an error
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// use herolib_kubernetes::KubernetesManager;
    /// use std::collections::HashMap;
    ///
    /// #[tokio::main]
    /// async fn main() -> Result<(), Box<dyn std::error::Error>> {
    ///     let km = KubernetesManager::new("default").await?;
    ///
    ///     let mut data = HashMap::new();
    ///     data.insert("username".to_string(), "admin".to_string());
    ///     data.insert("password".to_string(), "secret123".to_string());
    ///
    ///     let secret = km.secret_create("my-secret", data, None).await?;
    ///     println!("Created Secret: {}", secret.metadata.name.unwrap_or_default());
    ///     Ok(())
    /// }
    /// ```
    pub async fn secret_create(
        &self,
        name: &str,
        data: HashMap<String, String>,
        secret_type: Option<&str>,
    ) -> KubernetesResult<Secret> {
        use k8s_openapi::ByteString;

        let secrets: Api<Secret> = Api::namespaced(self.client.clone(), &self.namespace);

        // Convert string data to base64 encoded bytes
        let encoded_data: std::collections::BTreeMap<String, ByteString> = data
            .into_iter()
            .map(|(k, v)| {
                let encoded = base64::engine::general_purpose::STANDARD.encode(v.as_bytes());
                (k, ByteString(encoded.into_bytes()))
            })
            .collect();

        let secret = Secret {
            metadata: ObjectMeta {
                name: Some(name.to_string()),
                namespace: Some(self.namespace.clone()),
                ..Default::default()
            },
            data: Some(encoded_data),
            type_: Some(secret_type.unwrap_or("Opaque").to_string()),
            ..Default::default()
        };

        let created_secret = secrets.create(&Default::default(), &secret).await?;
        log::info!("Created Secret '{name}'");
        Ok(created_secret)
    }

    /// Get a specific Secret by name
    ///
    /// # Arguments
    ///
    /// * `name` - The name of the Secret to retrieve
    ///
    /// # Returns
    ///
    /// * `KubernetesResult<Secret>` - The Secret or an error
    pub async fn secret_get(&self, name: &str) -> KubernetesResult<Secret> {
        let secrets: Api<Secret> = Api::namespaced(self.client.clone(), &self.namespace);
        let secret = secrets.get(name).await?;
        Ok(secret)
    }

    /// Create a namespace (idempotent operation)
    ///
    /// # Arguments
    ///
    /// * `name` - The name of the namespace to create
    ///
    /// # Returns
    ///
    /// * `KubernetesResult<()>` - Success or an error
    pub async fn namespace_create(&self, name: &str) -> KubernetesResult<()> {
        let name = name.to_string(); // Clone for move into closure
        self.execute_with_safety(move || {
            let name = name.clone();
            let client = self.client.clone();
            async move {
                let namespaces: Api<Namespace> = Api::all(client);

                // Check if namespace already exists
                match namespaces.get(&name).await {
                    Ok(_) => {
                        log::info!("Namespace '{name}' already exists");
                        return Ok(());
                    }
                    Err(kube::Error::Api(api_err)) if api_err.code == 404 => {
                        // Namespace doesn't exist, we'll create it
                    }
                    Err(e) => return Err(KubernetesError::ApiError(e)),
                }

                // Create the namespace
                let namespace = Namespace {
                    metadata: k8s_openapi::apimachinery::pkg::apis::meta::v1::ObjectMeta {
                        name: Some(name.clone()),
                        ..Default::default()
                    },
                    ..Default::default()
                };

                namespaces.create(&Default::default(), &namespace).await?;
                log::info!("Created namespace '{name}'");

                // Verify namespace was actually created
                match namespaces.get(&name).await {
                    Ok(_) => {
                        log::info!("Verified namespace '{name}' exists");
                        Ok(())
                    }
                    Err(e) => {
                        log::error!("Failed to verify namespace '{name}' was created: {e}");
                        Err(KubernetesError::operation_error(format!(
                            "Namespace '{name}' was not created successfully. Verification failed: {e}"
                        )))
                    }
                }
            }
        })
        .await
    }

    /// Delete resources matching a PCRE pattern
    ///
    /// ⚠️  **WARNING**: This operation is destructive and irreversible!
    /// This method walks over all resources in the namespace and deletes
    /// those whose names match the provided regular expression pattern.
    ///
    /// # Safety
    /// - Always test patterns in a safe environment first
    /// - Use specific patterns to avoid accidental deletion of critical resources
    /// - Consider the impact on dependent resources before deletion
    ///
    /// # Arguments
    ///
    /// * `pattern` - PCRE pattern to match resource names against
    ///
    /// # Returns
    ///
    /// * `KubernetesResult<usize>` - Number of resources deleted or an error
    pub async fn delete(&self, pattern: &str) -> KubernetesResult<usize> {
        let regex = Regex::new(pattern)?;

        // Log warning about destructive operation
        log::warn!(
            "🚨 DESTRUCTIVE OPERATION: Starting bulk deletion with pattern '{}' in namespace '{}'",
            pattern,
            self.namespace
        );

        let mut deleted_count = 0;
        let mut failed_deletions = Vec::new();

        // Delete matching pods
        match self.delete_pods_matching(&regex).await {
            Ok(count) => deleted_count += count,
            Err(e) => {
                log::error!("Failed to delete pods matching pattern '{pattern}': {e}");
                failed_deletions.push(format!("pods: {e}"));
            }
        }

        // Delete matching services
        match self.delete_services_matching(&regex).await {
            Ok(count) => deleted_count += count,
            Err(e) => {
                log::error!("Failed to delete services matching pattern '{pattern}': {e}");
                failed_deletions.push(format!("services: {e}"));
            }
        }

        // Delete matching deployments
        match self.delete_deployments_matching(&regex).await {
            Ok(count) => deleted_count += count,
            Err(e) => {
                log::error!("Failed to delete deployments matching pattern '{pattern}': {e}");
                failed_deletions.push(format!("deployments: {e}"));
            }
        }

        // Delete matching configmaps
        match self.delete_configmaps_matching(&regex).await {
            Ok(count) => deleted_count += count,
            Err(e) => {
                log::error!("Failed to delete configmaps matching pattern '{pattern}': {e}");
                failed_deletions.push(format!("configmaps: {e}"));
            }
        }

        // Delete matching secrets
        match self.delete_secrets_matching(&regex).await {
            Ok(count) => deleted_count += count,
            Err(e) => {
                log::error!("Failed to delete secrets matching pattern '{pattern}': {e}");
                failed_deletions.push(format!("secrets: {e}"));
            }
        }

        if !failed_deletions.is_empty() {
            log::error!(
                "Bulk deletion completed with {} successes and {} failures. Failed: [{}]",
                deleted_count,
                failed_deletions.len(),
                failed_deletions.join(", ")
            );
            return Err(KubernetesError::operation_error(format!(
                "Partial deletion failure: {} resources deleted, {} resource types failed: {}",
                deleted_count,
                failed_deletions.len(),
                failed_deletions.join(", ")
            )));
        }

        log::info!(
            "✅ Successfully deleted {} resources matching pattern '{}' in namespace '{}'",
            deleted_count,
            pattern,
            self.namespace
        );
        Ok(deleted_count)
    }

    /// Delete pods matching the regex pattern
    async fn delete_pods_matching(&self, regex: &Regex) -> KubernetesResult<usize> {
        let pods: Api<Pod> = Api::namespaced(self.client.clone(), &self.namespace);
        let pod_list = pods.list(&Default::default()).await?;
        let mut deleted = 0;

        for pod in pod_list.items {
            if let Some(name) = &pod.metadata.name {
                if regex.is_match(name) {
                    match pods.delete(name, &Default::default()).await {
                        Ok(_) => {
                            log::info!("Deleted pod '{name}'");
                            deleted += 1;
                        }
                        Err(e) => {
                            log::error!("Failed to delete pod '{name}': {e}");
                        }
                    }
                }
            }
        }

        Ok(deleted)
    }

    /// Delete services matching the regex pattern
    async fn delete_services_matching(&self, regex: &Regex) -> KubernetesResult<usize> {
        let services: Api<Service> = Api::namespaced(self.client.clone(), &self.namespace);
        let service_list = services.list(&Default::default()).await?;
        let mut deleted = 0;

        for service in service_list.items {
            if let Some(name) = &service.metadata.name {
                if regex.is_match(name) {
                    match services.delete(name, &Default::default()).await {
                        Ok(_) => {
                            log::info!("Deleted service '{name}'");
                            deleted += 1;
                        }
                        Err(e) => {
                            log::error!("Failed to delete service '{name}': {e}");
                        }
                    }
                }
            }
        }

        Ok(deleted)
    }

    /// Delete deployments matching the regex pattern
    async fn delete_deployments_matching(&self, regex: &Regex) -> KubernetesResult<usize> {
        let deployments: Api<Deployment> = Api::namespaced(self.client.clone(), &self.namespace);
        let deployment_list = deployments.list(&Default::default()).await?;
        let mut deleted = 0;

        for deployment in deployment_list.items {
            if let Some(name) = &deployment.metadata.name {
                if regex.is_match(name) {
                    match deployments.delete(name, &Default::default()).await {
                        Ok(_) => {
                            log::info!("Deleted deployment '{name}'");
                            deleted += 1;
                        }
                        Err(e) => {
                            log::error!("Failed to delete deployment '{name}': {e}");
                        }
                    }
                }
            }
        }

        Ok(deleted)
    }

    /// Delete configmaps matching the regex pattern
    async fn delete_configmaps_matching(&self, regex: &Regex) -> KubernetesResult<usize> {
        let configmaps: Api<ConfigMap> = Api::namespaced(self.client.clone(), &self.namespace);
        let configmap_list = configmaps.list(&Default::default()).await?;
        let mut deleted = 0;

        for configmap in configmap_list.items {
            if let Some(name) = &configmap.metadata.name {
                if regex.is_match(name) {
                    match configmaps.delete(name, &Default::default()).await {
                        Ok(_) => {
                            log::info!("Deleted configmap '{name}'");
                            deleted += 1;
                        }
                        Err(e) => {
                            log::error!("Failed to delete configmap '{name}': {e}");
                        }
                    }
                }
            }
        }

        Ok(deleted)
    }

    /// Delete secrets matching the regex pattern
    async fn delete_secrets_matching(&self, regex: &Regex) -> KubernetesResult<usize> {
        let secrets: Api<Secret> = Api::namespaced(self.client.clone(), &self.namespace);
        let secret_list = secrets.list(&Default::default()).await?;
        let mut deleted = 0;

        for secret in secret_list.items {
            if let Some(name) = &secret.metadata.name {
                if regex.is_match(name) {
                    match secrets.delete(name, &Default::default()).await {
                        Ok(_) => {
                            log::info!("Deleted secret '{name}'");
                            deleted += 1;
                        }
                        Err(e) => {
                            log::error!("Failed to delete secret '{name}': {e}");
                        }
                    }
                }
            }
        }

        Ok(deleted)
    }

    /// Create a simple pod with a single container
    ///
    /// # Arguments
    ///
    /// * `name` - The name of the pod
    /// * `image` - The container image to use
    /// * `labels` - Optional labels for the pod
    /// * `env_vars` - Optional environment variables for the container
    ///
    /// # Returns
    ///
    /// * `KubernetesResult<Pod>` - The created pod or an error
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// use herolib_kubernetes::KubernetesManager;
    /// use std::collections::HashMap;
    ///
    /// #[tokio::main]
    /// async fn main() -> Result<(), Box<dyn std::error::Error>> {
    ///     let km = KubernetesManager::new("default").await?;
    ///
    ///     let mut labels = HashMap::new();
    ///     labels.insert("app".to_string(), "my-app".to_string());
    ///
    ///     let pod = km.pod_create("my-pod", "nginx:latest", Some(labels), None).await?;
    ///     println!("Created pod: {}", pod.metadata.name.unwrap_or_default());
    ///     Ok(())
    /// }
    /// ```
    pub async fn pod_create(
        &self,
        name: &str,
        image: &str,
        labels: Option<HashMap<String, String>>,
        env_vars: Option<HashMap<String, String>>,
    ) -> KubernetesResult<Pod> {
        use k8s_openapi::api::core::v1::{Container, PodSpec};

        let pods: Api<Pod> = Api::namespaced(self.client.clone(), &self.namespace);

        let pod = Pod {
            metadata: ObjectMeta {
                name: Some(name.to_string()),
                namespace: Some(self.namespace.clone()),
                labels: labels.map(|l| l.into_iter().collect()),
                ..Default::default()
            },
            spec: Some(PodSpec {
                containers: vec![{
                    let mut container = Container {
                        name: name.to_string(),
                        image: Some(image.to_string()),
                        ..Default::default()
                    };

                    // Add environment variables if provided
                    if let Some(env_vars) = env_vars {
                        use k8s_openapi::api::core::v1::EnvVar;
                        container.env = Some(
                            env_vars
                                .into_iter()
                                .map(|(key, value)| EnvVar {
                                    name: key,
                                    value: Some(value),
                                    ..Default::default()
                                })
                                .collect(),
                        );
                    }

                    container
                }],
                ..Default::default()
            }),
            ..Default::default()
        };

        let created_pod = pods.create(&Default::default(), &pod).await?;
        log::info!("Created pod '{name}' with image '{image}'");
        Ok(created_pod)
    }

    /// Get a specific pod by name
    ///
    /// # Arguments
    ///
    /// * `name` - The name of the pod to retrieve
    ///
    /// # Returns
    ///
    /// * `KubernetesResult<Pod>` - The pod or an error
    pub async fn pod_get(&self, name: &str) -> KubernetesResult<Pod> {
        let pods: Api<Pod> = Api::namespaced(self.client.clone(), &self.namespace);
        let pod = pods.get(name).await?;
        Ok(pod)
    }

    /// Create a simple service
    ///
    /// # Arguments
    ///
    /// * `name` - The name of the service
    /// * `selector` - Labels to select pods
    /// * `port` - The port to expose
    /// * `target_port` - The target port on pods (defaults to port if None)
    ///
    /// # Returns
    ///
    /// * `KubernetesResult<Service>` - The created service or an error
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// use herolib_kubernetes::KubernetesManager;
    /// use std::collections::HashMap;
    ///
    /// #[tokio::main]
    /// async fn main() -> Result<(), Box<dyn std::error::Error>> {
    ///     let km = KubernetesManager::new("default").await?;
    ///
    ///     let mut selector = HashMap::new();
    ///     selector.insert("app".to_string(), "my-app".to_string());
    ///
    ///     let service = km.service_create("my-service", selector, 80, Some(8080)).await?;
    ///     println!("Created service: {}", service.metadata.name.unwrap_or_default());
    ///     Ok(())
    /// }
    /// ```
    pub async fn service_create(
        &self,
        name: &str,
        selector: HashMap<String, String>,
        port: i32,
        target_port: Option<i32>,
    ) -> KubernetesResult<Service> {
        use k8s_openapi::api::core::v1::{ServicePort, ServiceSpec};
        use k8s_openapi::apimachinery::pkg::util::intstr::IntOrString;

        let services: Api<Service> = Api::namespaced(self.client.clone(), &self.namespace);

        let service = Service {
            metadata: ObjectMeta {
                name: Some(name.to_string()),
                namespace: Some(self.namespace.clone()),
                ..Default::default()
            },
            spec: Some(ServiceSpec {
                selector: Some(selector.into_iter().collect()),
                ports: Some(vec![ServicePort {
                    port,
                    target_port: Some(IntOrString::Int(target_port.unwrap_or(port))),
                    ..Default::default()
                }]),
                ..Default::default()
            }),
            ..Default::default()
        };

        let created_service = services.create(&Default::default(), &service).await?;
        log::info!("Created service '{name}' on port {port}");
        Ok(created_service)
    }

    /// Create a LoadBalancer service for external access
    ///
    /// This is similar to `service_create` but creates a LoadBalancer type service,
    /// which is useful for environments like OrbStack, Docker Desktop, or cloud providers
    /// that support LoadBalancer services for external access.
    ///
    /// # Arguments
    ///
    /// * `name` - The name of the service
    /// * `selector` - Labels to select pods
    /// * `port` - The port to expose
    /// * `target_port` - The target port on pods (defaults to port if None)
    ///
    /// # Returns
    ///
    /// * `KubernetesResult<Service>` - The created service or an error
    pub async fn service_create_lb(
        &self,
        name: &str,
        selector: HashMap<String, String>,
        port: i32,
        target_port: Option<i32>,
    ) -> KubernetesResult<Service> {
        use k8s_openapi::api::core::v1::{ServicePort, ServiceSpec};
        use k8s_openapi::apimachinery::pkg::util::intstr::IntOrString;

        let services: Api<Service> = Api::namespaced(self.client.clone(), &self.namespace);

        let service = Service {
            metadata: ObjectMeta {
                name: Some(name.to_string()),
                namespace: Some(self.namespace.clone()),
                ..Default::default()
            },
            spec: Some(ServiceSpec {
                type_: Some("LoadBalancer".to_string()),
                selector: Some(selector.into_iter().collect()),
                ports: Some(vec![ServicePort {
                    port,
                    target_port: Some(IntOrString::Int(target_port.unwrap_or(port))),
                    ..Default::default()
                }]),
                ..Default::default()
            }),
            ..Default::default()
        };

        let created_service = services.create(&Default::default(), &service).await?;
        log::info!("Created LoadBalancer service '{name}' on port {port}");
        Ok(created_service)
    }

    /// Get a specific service by name
    ///
    /// # Arguments
    ///
    /// * `name` - The name of the service to retrieve
    ///
    /// # Returns
    ///
    /// * `KubernetesResult<Service>` - The service or an error
    pub async fn service_get(&self, name: &str) -> KubernetesResult<Service> {
        let services: Api<Service> = Api::namespaced(self.client.clone(), &self.namespace);
        let service = services.get(name).await?;
        Ok(service)
    }

    /// Create a simple deployment
    ///
    /// # Arguments
    ///
    /// * `name` - The name of the deployment
    /// * `image` - The container image to use
    /// * `replicas` - Number of replicas to create
    /// * `labels` - Optional labels for the deployment and pods
    ///
    /// # Returns
    ///
    /// * `KubernetesResult<Deployment>` - The created deployment or an error
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// use herolib_kubernetes::KubernetesManager;
    /// use std::collections::HashMap;
    ///
    /// #[tokio::main]
    /// async fn main() -> Result<(), Box<dyn std::error::Error>> {
    ///     let km = KubernetesManager::new("default").await?;
    ///
    ///     let mut labels = HashMap::new();
    ///     labels.insert("app".to_string(), "my-app".to_string());
    ///
    ///     let deployment = km.deployment_create("my-deployment", "nginx:latest", 3, Some(labels), None).await?;
    ///     println!("Created deployment: {}", deployment.metadata.name.unwrap_or_default());
    ///     Ok(())
    /// }
    /// ```
    pub async fn deployment_create(
        &self,
        name: &str,
        image: &str,
        replicas: i32,
        labels: Option<HashMap<String, String>>,
        env_vars: Option<HashMap<String, String>>,
    ) -> KubernetesResult<Deployment> {
        use k8s_openapi::api::apps::v1::DeploymentSpec;
        use k8s_openapi::api::core::v1::{Container, PodSpec, PodTemplateSpec};
        use k8s_openapi::apimachinery::pkg::apis::meta::v1::LabelSelector;

        let deployments: Api<Deployment> = Api::namespaced(self.client.clone(), &self.namespace);

        let labels_btree = labels
            .as_ref()
            .map(|l| l.iter().map(|(k, v)| (k.clone(), v.clone())).collect());
        let selector_labels = labels.clone().unwrap_or_else(|| {
            let mut default_labels = HashMap::new();
            default_labels.insert("app".to_string(), name.to_string());
            default_labels
        });

        let deployment = Deployment {
            metadata: ObjectMeta {
                name: Some(name.to_string()),
                namespace: Some(self.namespace.clone()),
                labels: labels_btree.clone(),
                ..Default::default()
            },
            spec: Some(DeploymentSpec {
                replicas: Some(replicas),
                selector: LabelSelector {
                    match_labels: Some(selector_labels.clone().into_iter().collect()),
                    ..Default::default()
                },
                template: PodTemplateSpec {
                    metadata: Some(ObjectMeta {
                        labels: Some(selector_labels.into_iter().collect()),
                        ..Default::default()
                    }),
                    spec: Some(PodSpec {
                        containers: vec![{
                            let mut container = Container {
                                name: name.to_string(),
                                image: Some(image.to_string()),
                                ..Default::default()
                            };

                            // Add environment variables if provided
                            if let Some(env_vars) = env_vars {
                                use k8s_openapi::api::core::v1::EnvVar;
                                container.env = Some(
                                    env_vars
                                        .into_iter()
                                        .map(|(key, value)| EnvVar {
                                            name: key,
                                            value: Some(value),
                                            ..Default::default()
                                        })
                                        .collect(),
                                );
                            }

                            container
                        }],
                        ..Default::default()
                    }),
                },
                ..Default::default()
            }),
            ..Default::default()
        };

        let created_deployment = deployments.create(&Default::default(), &deployment).await?;
        log::info!("Created deployment '{name}' with {replicas} replicas using image '{image}'");
        Ok(created_deployment)
    }

    /// Get a specific deployment by name
    ///
    /// # Arguments
    ///
    /// * `name` - The name of the deployment to retrieve
    ///
    /// # Returns
    ///
    /// * `KubernetesResult<Deployment>` - The deployment or an error
    pub async fn deployment_get(&self, name: &str) -> KubernetesResult<Deployment> {
        let deployments: Api<Deployment> = Api::namespaced(self.client.clone(), &self.namespace);
        let deployment = deployments.get(name).await?;
        Ok(deployment)
    }

    /// Delete a specific pod by name (idempotent - succeeds if pod doesn't exist)
    ///
    /// # Arguments
    ///
    /// * `name` - The name of the pod to delete
    ///
    /// # Returns
    ///
    /// * `KubernetesResult<()>` - Success or an error
    pub async fn pod_delete(&self, name: &str) -> KubernetesResult<()> {
        let pods: Api<Pod> = Api::namespaced(self.client.clone(), &self.namespace);
        match pods.delete(name, &Default::default()).await {
            Ok(_) => {
                log::info!("Deleted pod '{name}'");
                Ok(())
            }
            Err(kube::Error::Api(api_err)) if api_err.code == 404 => {
                log::info!("Pod '{name}' does not exist, nothing to delete");
                Ok(())
            }
            Err(e) => Err(KubernetesError::ApiError(e)),
        }
    }

    /// Delete a specific service by name (idempotent - succeeds if service doesn't exist)
    ///
    /// # Arguments
    ///
    /// * `name` - The name of the service to delete
    ///
    /// # Returns
    ///
    /// * `KubernetesResult<()>` - Success or an error
    pub async fn service_delete(&self, name: &str) -> KubernetesResult<()> {
        let services: Api<Service> = Api::namespaced(self.client.clone(), &self.namespace);
        match services.delete(name, &Default::default()).await {
            Ok(_) => {
                log::info!("Deleted service '{name}'");
                Ok(())
            }
            Err(kube::Error::Api(api_err)) if api_err.code == 404 => {
                log::info!("Service '{name}' does not exist, nothing to delete");
                Ok(())
            }
            Err(e) => Err(KubernetesError::ApiError(e)),
        }
    }

    /// Delete a specific deployment by name (idempotent - succeeds if deployment doesn't exist)
    ///
    /// # Arguments
    ///
    /// * `name` - The name of the deployment to delete
    ///
    /// # Returns
    ///
    /// * `KubernetesResult<()>` - Success or an error
    pub async fn deployment_delete(&self, name: &str) -> KubernetesResult<()> {
        let deployments: Api<Deployment> = Api::namespaced(self.client.clone(), &self.namespace);
        match deployments.delete(name, &Default::default()).await {
            Ok(_) => {
                log::info!("Deleted deployment '{name}'");
                Ok(())
            }
            Err(kube::Error::Api(api_err)) if api_err.code == 404 => {
                log::info!("Deployment '{name}' does not exist, nothing to delete");
                Ok(())
            }
            Err(e) => Err(KubernetesError::ApiError(e)),
        }
    }

    /// Delete a specific ConfigMap by name (idempotent - succeeds if configmap doesn't exist)
    ///
    /// # Arguments
    ///
    /// * `name` - The name of the ConfigMap to delete
    ///
    /// # Returns
    ///
    /// * `KubernetesResult<()>` - Success or an error
    pub async fn configmap_delete(&self, name: &str) -> KubernetesResult<()> {
        let configmaps: Api<ConfigMap> = Api::namespaced(self.client.clone(), &self.namespace);
        match configmaps.delete(name, &Default::default()).await {
            Ok(_) => {
                log::info!("Deleted ConfigMap '{name}'");
                Ok(())
            }
            Err(kube::Error::Api(api_err)) if api_err.code == 404 => {
                log::info!("ConfigMap '{name}' does not exist, nothing to delete");
                Ok(())
            }
            Err(e) => Err(KubernetesError::ApiError(e)),
        }
    }

    /// Delete a specific Secret by name (idempotent - succeeds if secret doesn't exist)
    ///
    /// # Arguments
    ///
    /// * `name` - The name of the Secret to delete
    ///
    /// # Returns
    ///
    /// * `KubernetesResult<()>` - Success or an error
    pub async fn secret_delete(&self, name: &str) -> KubernetesResult<()> {
        let secrets: Api<Secret> = Api::namespaced(self.client.clone(), &self.namespace);
        match secrets.delete(name, &Default::default()).await {
            Ok(_) => {
                log::info!("Deleted Secret '{name}'");
                Ok(())
            }
            Err(kube::Error::Api(api_err)) if api_err.code == 404 => {
                log::info!("Secret '{name}' does not exist, nothing to delete");
                Ok(())
            }
            Err(e) => Err(KubernetesError::ApiError(e)),
        }
    }

    /// Get resource counts for the namespace
    ///
    /// # Returns
    ///
    /// * `KubernetesResult<HashMap<String, usize>>` - Resource counts by type
    pub async fn resource_counts(&self) -> KubernetesResult<HashMap<String, usize>> {
        let mut counts = HashMap::new();

        // Count pods
        let pods = self.pods_list().await?;
        counts.insert("pods".to_string(), pods.len());

        // Count services
        let services = self.services_list().await?;
        counts.insert("services".to_string(), services.len());

        // Count deployments
        let deployments = self.deployments_list().await?;
        counts.insert("deployments".to_string(), deployments.len());

        // Count configmaps
        let configmaps = self.configmaps_list().await?;
        counts.insert("configmaps".to_string(), configmaps.len());

        // Count secrets
        let secrets = self.secrets_list().await?;
        counts.insert("secrets".to_string(), secrets.len());

        Ok(counts)
    }

    /// Check if a namespace exists
    ///
    /// # Arguments
    ///
    /// * `name` - The name of the namespace to check
    ///
    /// # Returns
    ///
    /// * `KubernetesResult<bool>` - True if namespace exists, false otherwise
    pub async fn namespace_exists(&self, name: &str) -> KubernetesResult<bool> {
        let namespaces: Api<Namespace> = Api::all(self.client.clone());
        match namespaces.get(name).await {
            Ok(_) => Ok(true),
            Err(kube::Error::Api(api_err)) if api_err.code == 404 => Ok(false),
            Err(e) => Err(KubernetesError::ApiError(e)),
        }
    }

    /// List all namespaces (cluster-wide operation)
    ///
    /// # Returns
    ///
    /// * `KubernetesResult<Vec<Namespace>>` - List of all namespaces
    pub async fn namespaces_list(&self) -> KubernetesResult<Vec<Namespace>> {
        let namespaces: Api<Namespace> = Api::all(self.client.clone());
        let namespace_list = namespaces.list(&Default::default()).await?;
        Ok(namespace_list.items)
    }

    /// Delete a namespace (cluster-wide operation, idempotent - succeeds if namespace doesn't exist)
    ///
    /// ⚠️  **WARNING**: This operation is destructive and will delete all resources in the namespace!
    ///
    /// # Arguments
    ///
    /// * `name` - The name of the namespace to delete
    ///
    /// # Returns
    ///
    /// * `KubernetesResult<()>` - Success or an error
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// use herolib_kubernetes::KubernetesManager;
    ///
    /// #[tokio::main]
    /// async fn main() -> Result<(), Box<dyn std::error::Error>> {
    ///     let km = KubernetesManager::new("default").await?;
    ///
    ///     // ⚠️ This will delete the entire namespace and all its resources!
    ///     km.namespace_delete("test-namespace").await?;
    ///     Ok(())
    /// }
    /// ```
    pub async fn namespace_delete(&self, name: &str) -> KubernetesResult<()> {
        // Protect core Kubernetes namespaces from deletion
        if PROTECTED_NAMESPACES.contains(&name) {
            return Err(KubernetesError::operation_error(format!(
                "Cannot delete protected namespace '{}'. Protected namespaces: {}",
                name,
                PROTECTED_NAMESPACES.join(", ")
            )));
        }

        let namespaces: Api<Namespace> = Api::all(self.client.clone());

        // Check if namespace exists first
        match namespaces.get(name).await {
            Err(kube::Error::Api(api_err)) if api_err.code == 404 => {
                log::info!("Namespace '{name}' does not exist, nothing to delete");
                return Ok(());
            }
            Err(e) => return Err(KubernetesError::ApiError(e)),
            Ok(_) => {}
        }

        // Log warning about destructive operation
        log::warn!("🚨 DESTRUCTIVE OPERATION: Deleting namespace '{name}' and ALL its resources!");

        // Delete resources in the correct order for faster cleanup:
        // 1. Controllers first (they recreate pods)
        // 2. Then pods and other resources
        // 3. Finally the namespace itself
        self.delete_all_resources_in_namespace(name).await?;

        // Now delete the namespace itself
        match namespaces.delete(name, &Default::default()).await {
            Ok(_) => {
                log::info!("Deleted namespace '{name}'");
                Ok(())
            }
            Err(kube::Error::Api(api_err)) if api_err.code == 404 => {
                log::info!("Namespace '{name}' already deleted");
                Ok(())
            }
            Err(e) => Err(KubernetesError::ApiError(e)),
        }
    }

    /// Delete all resources in a namespace in the correct order for fast cleanup
    ///
    /// Order of deletion:
    /// 1. Deployments, StatefulSets, DaemonSets, ReplicaSets (controllers that recreate pods)
    /// 2. Jobs, CronJobs
    /// 3. Pods (any remaining)
    /// 4. Services, Ingresses
    /// 5. ConfigMaps, Secrets
    /// 6. PersistentVolumeClaims
    ///
    /// This ensures controllers are deleted before their pods, preventing
    /// pods from being recreated during deletion.
    async fn delete_all_resources_in_namespace(&self, namespace: &str) -> KubernetesResult<()> {
        use k8s_openapi::api::apps::v1::{DaemonSet, ReplicaSet, StatefulSet};
        use k8s_openapi::api::batch::v1::{CronJob, Job};
        use k8s_openapi::api::core::v1::PersistentVolumeClaim;
        use kube::api::DeleteParams;

        let delete_params = DeleteParams::default();

        // 1. Delete Deployments
        let deployments: Api<Deployment> = Api::namespaced(self.client.clone(), namespace);
        if let Ok(list) = deployments.list(&Default::default()).await {
            for item in list.items {
                if let Some(name) = item.metadata.name {
                    log::info!("Deleting deployment '{name}' in namespace '{namespace}'");
                    let _ = deployments.delete(&name, &delete_params).await;
                }
            }
        }

        // 2. Delete StatefulSets
        let statefulsets: Api<StatefulSet> = Api::namespaced(self.client.clone(), namespace);
        if let Ok(list) = statefulsets.list(&Default::default()).await {
            for item in list.items {
                if let Some(name) = item.metadata.name {
                    log::info!("Deleting statefulset '{name}' in namespace '{namespace}'");
                    let _ = statefulsets.delete(&name, &delete_params).await;
                }
            }
        }

        // 3. Delete DaemonSets
        let daemonsets: Api<DaemonSet> = Api::namespaced(self.client.clone(), namespace);
        if let Ok(list) = daemonsets.list(&Default::default()).await {
            for item in list.items {
                if let Some(name) = item.metadata.name {
                    log::info!("Deleting daemonset '{name}' in namespace '{namespace}'");
                    let _ = daemonsets.delete(&name, &delete_params).await;
                }
            }
        }

        // 4. Delete ReplicaSets (usually managed by Deployments, but clean up orphans)
        let replicasets: Api<ReplicaSet> = Api::namespaced(self.client.clone(), namespace);
        if let Ok(list) = replicasets.list(&Default::default()).await {
            for item in list.items {
                if let Some(name) = item.metadata.name {
                    log::info!("Deleting replicaset '{name}' in namespace '{namespace}'");
                    let _ = replicasets.delete(&name, &delete_params).await;
                }
            }
        }

        // 5. Delete Jobs
        let jobs: Api<Job> = Api::namespaced(self.client.clone(), namespace);
        if let Ok(list) = jobs.list(&Default::default()).await {
            for item in list.items {
                if let Some(name) = item.metadata.name {
                    log::info!("Deleting job '{name}' in namespace '{namespace}'");
                    let _ = jobs.delete(&name, &delete_params).await;
                }
            }
        }

        // 6. Delete CronJobs
        let cronjobs: Api<CronJob> = Api::namespaced(self.client.clone(), namespace);
        if let Ok(list) = cronjobs.list(&Default::default()).await {
            for item in list.items {
                if let Some(name) = item.metadata.name {
                    log::info!("Deleting cronjob '{name}' in namespace '{namespace}'");
                    let _ = cronjobs.delete(&name, &delete_params).await;
                }
            }
        }

        // 7. Delete Pods (any remaining after controllers are gone)
        let pods: Api<Pod> = Api::namespaced(self.client.clone(), namespace);
        if let Ok(list) = pods.list(&Default::default()).await {
            for item in list.items {
                if let Some(name) = item.metadata.name {
                    log::info!("Deleting pod '{name}' in namespace '{namespace}'");
                    let _ = pods.delete(&name, &delete_params).await;
                }
            }
        }

        // 8. Delete Services
        let services: Api<Service> = Api::namespaced(self.client.clone(), namespace);
        if let Ok(list) = services.list(&Default::default()).await {
            for item in list.items {
                if let Some(name) = item.metadata.name {
                    log::info!("Deleting service '{name}' in namespace '{namespace}'");
                    let _ = services.delete(&name, &delete_params).await;
                }
            }
        }

        // 9. Delete ConfigMaps
        let configmaps: Api<ConfigMap> = Api::namespaced(self.client.clone(), namespace);
        if let Ok(list) = configmaps.list(&Default::default()).await {
            for item in list.items {
                if let Some(name) = item.metadata.name {
                    log::info!("Deleting configmap '{name}' in namespace '{namespace}'");
                    let _ = configmaps.delete(&name, &delete_params).await;
                }
            }
        }

        // 10. Delete Secrets
        let secrets: Api<Secret> = Api::namespaced(self.client.clone(), namespace);
        if let Ok(list) = secrets.list(&Default::default()).await {
            for item in list.items {
                if let Some(name) = item.metadata.name {
                    // Skip default service account token secrets
                    if name.starts_with("default-token-") {
                        continue;
                    }
                    log::info!("Deleting secret '{name}' in namespace '{namespace}'");
                    let _ = secrets.delete(&name, &delete_params).await;
                }
            }
        }

        // 11. Delete PersistentVolumeClaims
        let pvcs: Api<PersistentVolumeClaim> = Api::namespaced(self.client.clone(), namespace);
        if let Ok(list) = pvcs.list(&Default::default()).await {
            for item in list.items {
                if let Some(name) = item.metadata.name {
                    log::info!("Deleting PVC '{name}' in namespace '{namespace}'");
                    let _ = pvcs.delete(&name, &delete_params).await;
                }
            }
        }

        log::info!("Finished deleting all resources in namespace '{namespace}'");
        Ok(())
    }

    /// Reset a namespace by deleting it (and all resources) and recreating it
    ///
    /// This is useful for cleaning up a namespace before redeploying resources.
    /// The function waits for the namespace to be fully deleted before recreating it.
    ///
    /// # Arguments
    ///
    /// * `name` - The name of the namespace to reset
    ///
    /// # Returns
    ///
    /// * `KubernetesResult<()>` - Success or an error
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// use herolib_kubernetes::KubernetesManager;
    ///
    /// #[tokio::main]
    /// async fn main() -> Result<(), Box<dyn std::error::Error>> {
    ///     let km = KubernetesManager::new("default").await?;
    ///
    ///     // This will delete the namespace and all its resources, then recreate it
    ///     km.namespace_reset("my-namespace").await?;
    ///     Ok(())
    /// }
    /// ```
    pub async fn namespace_reset(&self, name: &str) -> KubernetesResult<()> {
        log::info!("Resetting namespace '{name}'...");

        // Delete the namespace if it exists
        self.namespace_delete(name).await?;

        // Wait for the namespace to be fully deleted
        let namespaces: Api<Namespace> = Api::all(self.client.clone());
        let max_wait = std::time::Duration::from_secs(60);
        let poll_interval = std::time::Duration::from_millis(500);
        let start = std::time::Instant::now();

        loop {
            match namespaces.get(name).await {
                Ok(ns) => {
                    // Check if namespace is in Terminating state
                    let phase = ns.status.as_ref().and_then(|s| s.phase.as_deref());
                    if phase == Some("Terminating") {
                        log::info!("Namespace '{name}' is terminating, waiting...");
                    } else {
                        log::info!("Namespace '{name}' still exists, waiting for deletion...");
                    }

                    if start.elapsed() > max_wait {
                        return Err(KubernetesError::timeout(format!(
                            "Timeout waiting for namespace '{name}' to be deleted after {:?}",
                            max_wait
                        )));
                    }
                    tokio::time::sleep(poll_interval).await;
                }
                Err(kube::Error::Api(api_err)) if api_err.code == 404 => {
                    log::info!("Namespace '{name}' has been deleted");
                    break;
                }
                Err(e) => return Err(KubernetesError::ApiError(e)),
            }
        }

        // Create the namespace fresh
        self.namespace_create(name).await?;
        log::info!("Namespace '{name}' has been reset successfully");

        Ok(())
    }

    /// Deploy a complete application with deployment and service
    ///
    /// This convenience method creates both a deployment and a service for an application,
    /// making it easy to deploy complete containerized applications with a single call.
    ///
    /// # Arguments
    ///
    /// * `name` - The name for both deployment and service
    /// * `image` - The container image to deploy
    /// * `replicas` - Number of replicas to create
    /// * `port` - The port the application listens on
    /// * `labels` - Optional labels for the resources
    ///
    /// # Returns
    ///
    /// * `KubernetesResult<()>` - Success or an error
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// use herolib_kubernetes::KubernetesManager;
    /// use std::collections::HashMap;
    ///
    /// #[tokio::main]
    /// async fn main() -> Result<(), Box<dyn std::error::Error>> {
    ///     let km = KubernetesManager::new("default").await?;
    ///
    ///     let mut labels = HashMap::new();
    ///     labels.insert("app".to_string(), "my-app".to_string());
    ///
    ///     km.deploy_application("my-app", "node:18", 3, 3000, Some(labels), None).await?;
    ///     Ok(())
    /// }
    /// ```
    pub async fn deploy_application(
        &self,
        name: &str,
        image: &str,
        replicas: i32,
        port: i32,
        labels: Option<HashMap<String, String>>,
        env_vars: Option<HashMap<String, String>>,
    ) -> KubernetesResult<()> {
        log::info!("Deploying application '{name}' with image '{image}'");

        // Create deployment with environment variables
        self.deployment_create(name, image, replicas, labels.clone(), env_vars)
            .await?;

        // Create service selector - use app=name if no labels provided
        let selector = if let Some(ref labels) = labels {
            labels.clone()
        } else {
            let mut default_selector = HashMap::new();
            default_selector.insert("app".to_string(), name.to_string());
            default_selector
        };

        // Create LoadBalancer service for external access (works with OrbStack, Docker Desktop, etc.)
        self.service_create_lb(name, selector, port, Some(port))
            .await?;

        log::info!("Successfully deployed application '{name}'");
        Ok(())
    }
}

/// Connection information for a Kubernetes service
#[derive(Debug, Clone)]
pub struct ServiceConnectionInfo {
    /// Service name
    pub name: String,
    /// Namespace
    pub namespace: String,
    /// Full DNS hostname for cluster-internal access (e.g., service.namespace.svc.cluster.local)
    pub hostname: String,
    /// DNS hostname for host access (e.g., for OrbStack: service.namespace.k8s.orb.local)
    pub host_hostname: Option<String>,
    /// Port(s) exposed by the service
    pub ports: Vec<i32>,
    /// NodePort(s) for external access (when service type is LoadBalancer or NodePort)
    pub node_ports: Vec<i32>,
    /// ClusterIP (if assigned)
    pub cluster_ip: Option<String>,
    /// External IP (for LoadBalancer services)
    pub external_ip: Option<String>,
    /// Service type (ClusterIP, NodePort, LoadBalancer)
    pub service_type: String,
}

impl ServiceConnectionInfo {
    /// Get the full connection string for the first port (e.g., "hostname:port")
    pub fn connection_string(&self) -> String {
        if let Some(port) = self.ports.first() {
            format!("{}:{}", self.hostname, port)
        } else {
            self.hostname.clone()
        }
    }

    /// Get a formatted display of connection info
    pub fn display(&self) -> String {
        let mut output = String::new();
        output.push_str(&format!("Service: {}\n", self.name));
        output.push_str(&format!("Namespace: {}\n", self.namespace));
        output.push_str(&format!("Type: {}\n", self.service_type));
        output.push_str(&format!("Hostname (in-cluster): {}\n", self.hostname));
        if let Some(ref host) = self.host_hostname {
            output.push_str(&format!("Hostname (from host): {}\n", host));
        }
        if !self.ports.is_empty() {
            output.push_str(&format!(
                "Ports: {}\n",
                self.ports
                    .iter()
                    .map(|p| p.to_string())
                    .collect::<Vec<_>>()
                    .join(", ")
            ));
        }
        if !self.node_ports.is_empty() {
            output.push_str(&format!(
                "NodePorts: {}\n",
                self.node_ports
                    .iter()
                    .map(|p| p.to_string())
                    .collect::<Vec<_>>()
                    .join(", ")
            ));
        }
        if let Some(ref ip) = self.cluster_ip {
            output.push_str(&format!("ClusterIP: {}\n", ip));
        }
        if let Some(ref ip) = self.external_ip {
            output.push_str(&format!("ExternalIP: {}\n", ip));
        }
        output
    }

    /// Get the best hostname for host access (prefers host_hostname if available)
    pub fn host_accessible_hostname(&self) -> &str {
        self.host_hostname.as_deref().unwrap_or(&self.hostname)
    }

    /// Get the localhost connection string for host access via NodePort
    /// Returns "localhost:PORT" if a NodePort is available
    pub fn localhost_connection_string(&self) -> Option<String> {
        self.node_ports
            .first()
            .map(|port| format!("localhost:{}", port))
    }
}

impl KubernetesManager {
    /// Get connection information for a service
    ///
    /// Returns the DNS hostname, ports, and other connection details for a service.
    /// This is useful for getting the information needed to connect to a deployed application.
    ///
    /// # Arguments
    ///
    /// * `service_name` - The name of the service to get connection info for
    ///
    /// # Returns
    ///
    /// * `KubernetesResult<ServiceConnectionInfo>` - Connection information or an error
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// use herolib_kubernetes::KubernetesManager;
    ///
    /// #[tokio::main]
    /// async fn main() -> Result<(), Box<dyn std::error::Error>> {
    ///     let km = KubernetesManager::new("myapp").await?;
    ///     let info = km.service_connection_info("myapp").await?;
    ///     println!("Connect to: {}", info.connection_string());
    ///     Ok(())
    /// }
    /// ```
    pub async fn service_connection_info(
        &self,
        service_name: &str,
    ) -> KubernetesResult<ServiceConnectionInfo> {
        let service = self.service_get(service_name).await?;

        let namespace = service
            .metadata
            .namespace
            .clone()
            .unwrap_or_else(|| self.namespace.clone());

        let name = service
            .metadata
            .name
            .clone()
            .unwrap_or_else(|| service_name.to_string());

        // Construct the full DNS hostname for in-cluster access
        let hostname = format!("{}.{}.svc.cluster.local", name, namespace);

        // Detect host-accessible hostname (e.g., for OrbStack)
        let host_hostname = self.detect_host_hostname(&name, &namespace).await;

        // Extract ports and node_ports from the service spec
        let (ports, node_ports) = service
            .spec
            .as_ref()
            .and_then(|spec| spec.ports.as_ref())
            .map(|ports| {
                let p: Vec<i32> = ports.iter().map(|p| p.port).collect();
                let np: Vec<i32> = ports.iter().filter_map(|p| p.node_port).collect();
                (p, np)
            })
            .unwrap_or_default();

        // Get ClusterIP
        let cluster_ip = service
            .spec
            .as_ref()
            .and_then(|spec| spec.cluster_ip.clone())
            .filter(|ip| ip != "None");

        // Get External IP (for LoadBalancer services)
        let external_ip = service
            .status
            .as_ref()
            .and_then(|status| status.load_balancer.as_ref())
            .and_then(|lb| lb.ingress.as_ref())
            .and_then(|ingress| ingress.first())
            .and_then(|ing| ing.ip.clone());

        // Get service type
        let service_type = service
            .spec
            .as_ref()
            .and_then(|spec| spec.type_.clone())
            .unwrap_or_else(|| "ClusterIP".to_string());

        Ok(ServiceConnectionInfo {
            name,
            namespace,
            hostname,
            host_hostname,
            ports,
            node_ports,
            cluster_ip,
            external_ip,
            service_type,
        })
    }

    /// Detect the host-accessible hostname for environments like OrbStack
    ///
    /// This checks for common Kubernetes-in-Docker/VM solutions that provide
    /// host-accessible DNS names.
    async fn detect_host_hostname(&self, service_name: &str, namespace: &str) -> Option<String> {
        // Check for OrbStack by looking at node labels or trying known DNS patterns
        // OrbStack uses: <service>.<namespace>.k8s.orb.local

        // Try to detect OrbStack by checking if we're running in an OrbStack environment
        // OrbStack nodes typically have specific labels or the cluster has certain characteristics

        // Check for OrbStack-specific indicators
        if self.is_orbstack_environment().await {
            return Some(format!("{}.{}.k8s.orb.local", service_name, namespace));
        }

        // Could add detection for other environments here (Docker Desktop, Rancher Desktop, etc.)

        None
    }

    /// Check if running in an OrbStack Kubernetes environment
    async fn is_orbstack_environment(&self) -> bool {
        // OrbStack clusters typically have nodes with specific characteristics
        // Check for orbstack in node names or labels
        let nodes: Api<Node> = Api::all(self.client.clone());

        if let Ok(node_list) = nodes.list(&Default::default()).await {
            for node in node_list.items {
                if let Some(name) = &node.metadata.name {
                    // OrbStack nodes are typically named "orbstack"
                    if name.contains("orbstack") {
                        return true;
                    }
                }
                // Check labels for OrbStack indicators
                if let Some(labels) = &node.metadata.labels {
                    for (key, value) in labels {
                        if key.contains("orbstack") || value.contains("orbstack") {
                            return true;
                        }
                    }
                }
                // Check for OrbStack's provisioner or other annotations
                if let Some(annotations) = &node.metadata.annotations {
                    for (key, value) in annotations {
                        if key.contains("orbstack") || value.contains("orbstack") {
                            return true;
                        }
                    }
                }
            }
        }

        false
    }
}

/// Determine if a Kubernetes API error is retryable
pub fn is_retryable_error(error: &kube::Error) -> bool {
    match error {
        // Network-related errors are typically retryable
        kube::Error::HttpError(_) => true,

        // API errors - check status codes
        kube::Error::Api(api_error) => {
            match api_error.code {
                // Temporary server errors
                500..=599 => true,
                // Rate limiting
                429 => true,
                // Conflict (might resolve on retry)
                409 => true,
                // Client errors are generally not retryable
                400..=499 => false,
                // Other codes - be conservative and retry
                _ => true,
            }
        }

        // Auth errors are not retryable
        kube::Error::Auth(_) => false,

        // Discovery errors might be temporary
        kube::Error::Discovery(_) => true,

        // Other errors - be conservative and retry
        _ => true,
    }
}