use anyhow::Result;
use async_trait::async_trait;
use chrono::{DateTime, Utc};
use k8s_openapi::api::apps::v1::{Deployment as K8sDeployment, DeploymentSpec, ReplicaSet};
use k8s_openapi::api::core::v1::{
    Capabilities, Container, ContainerPort, Event, HTTPGetAction, HostAlias, LocalObjectReference,
    Namespace, PodSecurityContext, PodSpec, PodTemplateSpec, Probe, ProjectedVolumeSource,
    ResourceRequirements, SeccompProfile, Secret, SecurityContext, Service,
    ServiceAccountTokenProjection, ServicePort, ServiceSpec, Volume, VolumeMount, VolumeProjection,
};
use k8s_openapi::api::networking::v1::{
    HTTPIngressPath, HTTPIngressRuleValue, Ingress, IngressBackend, IngressRule,
    IngressServiceBackend, IngressSpec, NetworkPolicy, NetworkPolicySpec, ServiceBackendPort,
};
use k8s_openapi::apimachinery::pkg::apis::meta::v1::{LabelSelector, ObjectMeta};
use kube::api::{Api, DeleteParams, Patch, PatchParams, PostParams};
use kube::Client;
use serde::{Deserialize, Serialize};
use std::collections::BTreeMap;
use std::sync::Arc;
use std::time::Duration;
use tokio::time::interval;
use tracing::{debug, error, info, warn};

use super::{DeploymentBackend, DeploymentUrls, HealthStatus, ReconcileResult};
use crate::db::deployments as db_deployments;
use crate::db::models::{Deployment, DeploymentStatus, Project};
use crate::db::projects as db_projects;
use crate::server::deployment::models::DEFAULT_DEPLOYMENT_GROUP;
use crate::server::registry::{
    models::{RegistryAuthMethod, RegistryCredentials},
    RegistryProvider,
};
use crate::server::settings::AccessRequirement;
use crate::server::state::ControllerState;

// Kubernetes label and annotation constants
const LABEL_MANAGED_BY: &str = "rise.dev/managed-by";
const LABEL_PROJECT: &str = "rise.dev/project";
const LABEL_DEPLOYMENT_GROUP: &str = "rise.dev/deployment-group";
const LABEL_DEPLOYMENT_ID: &str = "rise.dev/deployment-id";
const LABEL_DEPLOYMENT_UUID: &str = "rise.dev/deployment-uuid";
const ANNOTATION_LAST_REFRESH: &str = "rise.dev/last-refresh";

/// Image pull secret name used for private registry authentication
const IMAGE_PULL_SECRET_NAME: &str = "rise-registry-creds";
const EXTRA_SERVICE_TOKENS_VOLUME_NAME: &str = "rise-extra-service-tokens";
const EXTRA_SERVICE_TOKENS_MOUNT_PATH: &str = "/var/run/secrets/rise/tokens";

/// Finalizer name for Kubernetes namespaces
/// Added when a namespace is created for a project, removed when namespace cleanup is complete.
pub const KUBERNETES_NAMESPACE_FINALIZER: &str = "kubernetes.rise.dev/namespace";

/// Container waiting state reasons that indicate irrecoverable errors
/// These errors require user intervention and won't resolve automatically
const IRRECOVERABLE_CONTAINER_REASONS: &[&str] = &[
    "InvalidImageName",
    "ErrImagePull",
    "ImagePullBackOff",
    "ImageInspectError",
    "CrashLoopBackOff",
    "CreateContainerConfigError",
    "CreateContainerError",
    "RunContainerError",
];

/// Critical event reasons that indicate deployment should fail
/// These are specific event reasons that won't resolve without user intervention
const CRITICAL_EVENT_REASONS: &[&str] = &[
    "FailedCreate",
    "FailedKillPod",
    "FailedPostStartHook",
    "FailedPreStopHook",
];

/// Minimum event count to consider a Warning event as critical
/// This helps filter out transient warnings (e.g., FailedScheduling)
const MIN_EVENT_COUNT_FOR_CRITICAL: i32 = 3;
const MAX_EVENT_QUERY_LIMIT: u32 = 200;
const MAX_EVENTS_PER_POD: usize = 5;
const MAX_PODS_IN_STATUS: usize = 20;
const MAX_CONDITIONS_PER_POD: usize = 10;
const MAX_CONTAINERS_PER_POD: usize = 10;
const MAX_POD_TEXT_CHARS: usize = 500;

fn truncate_text(input: &str, max_chars: usize) -> String {
    if input.chars().count() <= max_chars {
        return input.to_string();
    }
    let mut truncated: String = input.chars().take(max_chars).collect();
    truncated.push_str("...");
    truncated
}

fn truncate_optional_text(input: Option<String>, max_chars: usize) -> Option<String> {
    input.map(|text| truncate_text(&text, max_chars))
}

fn reconcile_in_progress_status(deployment: &Deployment) -> DeploymentStatus {
    if deployment.first_healthy_at.is_some() {
        DeploymentStatus::Unhealthy
    } else {
        DeploymentStatus::Deploying
    }
}

/// Kubernetes-specific metadata stored in deployment.controller_metadata
#[derive(Serialize, Deserialize, Default, Clone)]
struct KubernetesMetadata {
    namespace: Option<String>,
    #[serde(alias = "replicaset_name")]
    deployment_name: Option<String>,
    service_name: Option<String>,
    ingress_name: Option<String>,
    image_tag: Option<String>,
    http_port: u16,
    #[serde(default)]
    reconcile_phase: ReconcilePhase,
    #[serde(alias = "previous_replicaset")]
    previous_deployment: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none", default)]
    pod_status: Option<PodStatus>,
}

/// Detailed Pod status information
#[derive(Serialize, Deserialize, Clone, Debug)]
struct PodStatus {
    /// Desired number of pods (from Deployment spec)
    desired_replicas: i32,
    /// Current number of ready pods
    ready_replicas: i32,
    /// Current number of pods (any state)
    current_replicas: i32,
    /// Per-pod details
    #[serde(skip_serializing_if = "Vec::is_empty", default)]
    pods: Vec<PodInfo>,
    /// Timestamp of last pod status check
    last_checked: DateTime<Utc>,
}

/// Per-pod information
#[derive(Serialize, Deserialize, Clone, Debug)]
struct PodInfo {
    name: String,
    phase: PodPhase,
    /// Pod conditions (Ready, ContainersReady, etc.)
    #[serde(skip_serializing_if = "Vec::is_empty", default)]
    conditions: Vec<PodCondition>,
    /// Container statuses
    #[serde(skip_serializing_if = "Vec::is_empty", default)]
    containers: Vec<ContainerStatusInfo>,
    /// Recent warning/error events
    #[serde(skip_serializing_if = "Vec::is_empty", default)]
    events: Vec<PodEvent>,
}

#[derive(Serialize, Deserialize, Clone, Debug)]
#[serde(rename_all = "PascalCase")]
enum PodPhase {
    Pending,
    Running,
    Succeeded,
    Failed,
    Unknown,
}

/// Pod condition (e.g., Ready, ContainersReady)
#[derive(Serialize, Deserialize, Clone, Debug)]
struct PodCondition {
    #[serde(rename = "type")]
    type_: String,
    status: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    reason: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    message: Option<String>,
}

/// Container status within a pod
#[derive(Serialize, Deserialize, Clone, Debug)]
struct ContainerStatusInfo {
    name: String,
    ready: bool,
    restart_count: i32,
    #[serde(skip_serializing_if = "Option::is_none")]
    state: Option<ContainerState>,
}

/// Container state (waiting, running, or terminated)
#[derive(Serialize, Deserialize, Clone, Debug)]
struct ContainerState {
    /// "waiting", "running", or "terminated"
    state_type: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    reason: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    message: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    exit_code: Option<i32>,
}

/// Pod event (from Kubernetes events API)
#[derive(Serialize, Deserialize, Clone, Debug)]
struct PodEvent {
    #[serde(rename = "type")]
    type_: String,
    reason: String,
    message: String,
    count: i32,
    last_timestamp: DateTime<Utc>,
}

/// Reconciliation phases for Kubernetes deployments
#[derive(Serialize, Deserialize, Default, Clone, Debug, PartialEq)]
#[serde(rename_all = "PascalCase")]
enum ReconcilePhase {
    #[default]
    NotStarted,
    // Old phases - kept temporarily until reconciliation loop is refactored
    CreatingNamespace,
    CreatingImagePullSecret,
    CreatingBackendService,
    CreatingService,
    #[serde(alias = "CreatingReplicaSet")]
    CreatingDeployment,
    #[serde(alias = "WaitingForReplicaSet")]
    WaitingForDeployment,
    UpdatingIngress,
    WaitingForHealth,
    SwitchingTraffic,
    Completed,
}

/// Check if an error indicates the namespace is missing
/// Handles both direct kube::Error and anyhow::Error with kube::Error source
fn is_namespace_not_found_error(
    error: &(impl std::fmt::Display + std::fmt::Debug + ?Sized),
) -> bool {
    let error_string = error.to_string();
    // Check if error message indicates namespace is missing
    // Kubernetes returns messages like: "namespaces \"rise-test\" not found"
    error_string.contains("namespaces") && error_string.contains("not found")
}

/// Parsed ingress URL components
#[derive(Debug, Clone)]
struct IngressUrl {
    /// The hostname (e.g., "rise.dev")
    host: String,
    /// Optional path prefix (e.g., "/myapp")
    /// None for subdomain-based routing
    path_prefix: Option<String>,
}

/// Configuration parameters for the Kubernetes controller
pub struct KubernetesControllerConfig {
    pub production_ingress_url_template: String,
    pub staging_ingress_url_template: Option<String>,
    pub ingress_port: Option<u16>,
    pub ingress_schema: String,
    pub registry_provider: Arc<dyn RegistryProvider>,
    pub auth_backend_url: String,
    pub auth_signin_url: String,
    /// Backend address for routing /.rise path prefix traffic to the Rise backend
    pub backend_address: Option<crate::server::settings::BackendAddress>,
    pub namespace_labels: std::collections::HashMap<String, String>,
    pub namespace_annotations: std::collections::HashMap<String, String>,
    pub ingress_annotations: std::collections::HashMap<String, String>,
    pub ingress_tls_secret_name: Option<String>,
    pub custom_domain_tls_mode: crate::server::settings::CustomDomainTlsMode,
    pub custom_domain_ingress_annotations: std::collections::HashMap<String, String>,
    pub node_selector: std::collections::HashMap<String, String>,
    pub image_pull_secret_name: Option<String>,
    pub access_classes: std::collections::HashMap<String, crate::server::settings::AccessClass>,
    /// Host aliases to inject into pod specs (hostname -> IP address)
    pub host_aliases: std::collections::HashMap<String, String>,
    pub extra_service_token_audiences: std::collections::HashMap<String, String>,
    pub network_policy: crate::server::settings::NetworkPolicyConfig,
    pub pod_security_enabled: bool,
    pub pod_resources: Option<crate::server::settings::PodResourceLimits>,
    pub health_probes: Option<crate::server::settings::HealthProbeConfig>,
}

/// Kubernetes controller implementation
pub struct KubernetesController {
    state: ControllerState,
    kube_client: Client,
    production_ingress_url_template: String,
    staging_ingress_url_template: Option<String>,
    ingress_port: Option<u16>,
    ingress_schema: String,
    registry_provider: Arc<dyn RegistryProvider>,
    auth_backend_url: String,
    auth_signin_url: String,
    /// Backend address for routing /.rise path prefix traffic to the Rise backend
    backend_address: Option<crate::server::settings::BackendAddress>,
    namespace_labels: std::collections::HashMap<String, String>,
    namespace_annotations: std::collections::HashMap<String, String>,
    ingress_annotations: std::collections::HashMap<String, String>,
    ingress_tls_secret_name: Option<String>,
    custom_domain_tls_mode: crate::server::settings::CustomDomainTlsMode,
    custom_domain_ingress_annotations: std::collections::HashMap<String, String>,
    node_selector: std::collections::HashMap<String, String>,
    image_pull_secret_name: Option<String>,
    access_classes: std::collections::HashMap<String, crate::server::settings::AccessClass>,
    /// In-memory cache of resource versions for drift detection
    /// Key: "deployment-{id}-{resource_type}" (e.g., "deployment-123-service")
    /// Value: last observed resourceVersion from Kubernetes API
    /// Lost on controller restart → causes one re-apply, then cached again
    resource_versions: Arc<std::sync::RwLock<std::collections::HashMap<String, String>>>,
    /// Host aliases to inject into pod specs (hostname -> IP address)
    host_aliases: std::collections::HashMap<String, String>,
    extra_service_token_audiences: std::collections::HashMap<String, String>,
    network_policy: crate::server::settings::NetworkPolicyConfig,
    pod_security_enabled: bool,
    pod_resources: Option<crate::server::settings::PodResourceLimits>,
    health_probes: Option<crate::server::settings::HealthProbeConfig>,
}

/// Helper enum for probe type
#[derive(Debug, Clone, Copy)]
enum ProbeType {
    Liveness,
    Readiness,
}

impl KubernetesController {
    /// Create a new Kubernetes controller
    pub fn new(
        state: ControllerState,
        kube_client: Client,
        config: KubernetesControllerConfig,
    ) -> Result<Self> {
        Ok(Self {
            state,
            kube_client,
            production_ingress_url_template: config.production_ingress_url_template,
            staging_ingress_url_template: config.staging_ingress_url_template,
            ingress_port: config.ingress_port,
            ingress_schema: config.ingress_schema,
            registry_provider: config.registry_provider,
            auth_backend_url: config.auth_backend_url,
            auth_signin_url: config.auth_signin_url,
            backend_address: config.backend_address,
            namespace_labels: config.namespace_labels,
            namespace_annotations: config.namespace_annotations,
            ingress_annotations: config.ingress_annotations,
            ingress_tls_secret_name: config.ingress_tls_secret_name,
            custom_domain_tls_mode: config.custom_domain_tls_mode,
            custom_domain_ingress_annotations: config.custom_domain_ingress_annotations,
            node_selector: config.node_selector,
            image_pull_secret_name: config.image_pull_secret_name,
            access_classes: config.access_classes,
            resource_versions: Arc::new(std::sync::RwLock::new(std::collections::HashMap::new())),
            host_aliases: config.host_aliases,
            extra_service_token_audiences: config.extra_service_token_audiences,
            network_policy: config.network_policy,
            pod_security_enabled: config.pod_security_enabled,
            pod_resources: config.pod_resources,
            health_probes: config.health_probes,
        })
    }

    /// Get cache key for a deployment's resource
    fn resource_cache_key(deployment_id: uuid::Uuid, resource_type: &str) -> String {
        format!("deployment-{}-{}", deployment_id, resource_type)
    }

    /// Check if resource needs apply based on version cache
    /// Returns true if: no cached version OR current version != cached version
    fn needs_apply(
        &self,
        deployment_id: uuid::Uuid,
        resource_type: &str,
        current_version: Option<&str>,
    ) -> bool {
        let key = Self::resource_cache_key(deployment_id, resource_type);
        let cache = self.resource_versions.read().unwrap();

        match (cache.get(&key), current_version) {
            (None, _) => true,                                  // Never seen before
            (Some(_), None) => true, // Resource exists but has no version?
            (Some(cached), Some(current)) => cached != current, // Version changed
        }
    }

    /// Update cached resource version
    ///
    /// When `version` is `Some`, inserts/updates the cached version.
    /// When `version` is `None`, removes the cache entry to force re-evaluation on next reconcile.
    fn update_version_cache(
        &self,
        deployment_id: uuid::Uuid,
        resource_type: &str,
        version: Option<String>,
    ) {
        let key = Self::resource_cache_key(deployment_id, resource_type);
        let mut cache = self.resource_versions.write().unwrap();
        if let Some(version) = version {
            cache.insert(key, version);
        } else {
            cache.remove(&key);
        }
    }

    /// Clear all cached versions for a deployment (on termination/deletion)
    fn clear_deployment_cache(&self, deployment_id: uuid::Uuid) {
        let mut cache = self.resource_versions.write().unwrap();
        cache.retain(|k, _| !k.starts_with(&format!("deployment-{}-", deployment_id)));
    }

    /// Test connection to Kubernetes API
    ///
    /// Validates that the controller can successfully connect to and communicate with
    /// the Kubernetes API server. This should be called during initialization to catch
    /// credential or connectivity issues early.
    ///
    /// Returns Ok with server version info on success, or Err on connection failure.
    pub async fn test_connection(&self) -> Result<()> {
        info!("Testing Kubernetes API connection...");

        // Try to get the server version - this is a lightweight API call that validates:
        // 1. Network connectivity to the API server
        // 2. Authentication credentials are valid
        // 3. Basic API permissions are working
        let version_info = self
            .kube_client
            .apiserver_version()
            .await
            .map_err(|e| anyhow::anyhow!("Failed to connect to Kubernetes API: {}", e))?;

        info!(
            "Successfully connected to Kubernetes API server (version: {}.{})",
            version_info.major, version_info.minor
        );

        Ok(())
    }

    /// Start namespace cleanup loop
    ///
    /// This loop runs independently and handles namespace deletion when projects are deleted.
    /// It follows the finalizer pattern to coordinate with project deletion.
    pub fn start(self: Arc<Self>) {
        let cleanup_self = Arc::clone(&self);
        tokio::spawn(async move {
            cleanup_self.namespace_cleanup_loop().await;
        });
    }

    /// Start secret refresh loop (Kubernetes-specific)
    pub fn start_secret_refresh_loop(self: Arc<Self>, interval_duration: Duration) {
        tokio::spawn(async move {
            let mut ticker = interval(interval_duration);
            loop {
                ticker.tick().await;
                if let Err(e) = self.refresh_image_pull_secrets().await {
                    error!("Error refreshing image pull secrets: {}", e);
                }
            }
        });
    }

    /// Namespace cleanup loop - deletes namespaces for deleted projects
    ///
    /// Runs every 5 seconds and:
    /// 1. Finds projects in Deleting status with the Kubernetes namespace finalizer
    /// 2. Deletes the Kubernetes namespace
    /// 3. Removes the finalizer so project can be fully deleted
    async fn namespace_cleanup_loop(&self) {
        info!("Kubernetes namespace cleanup loop started");
        let mut ticker = interval(Duration::from_secs(5));

        loop {
            ticker.tick().await;
            if let Err(e) = self.cleanup_namespaces().await {
                error!("Error in Kubernetes namespace cleanup loop: {}", e);
            }
        }
    }

    /// Process namespace cleanup for all deleting projects with Kubernetes finalizer
    async fn cleanup_namespaces(&self) -> Result<()> {
        // Find projects marked for deletion that still have Kubernetes namespace finalizer
        let projects = db_projects::find_deleting_with_finalizer(
            &self.state.db_pool,
            KUBERNETES_NAMESPACE_FINALIZER,
            10,
        )
        .await?;

        for project in projects {
            debug!(
                "Cleaning up Kubernetes namespace for project: {}",
                project.name
            );

            let namespace_name = Self::namespace_name(&project);
            let ns_api: Api<Namespace> = Api::all(self.kube_client.clone());

            // Try to delete the namespace
            match ns_api
                .delete(&namespace_name, &DeleteParams::default())
                .await
            {
                Ok(_) => {
                    info!(
                        "Deleted Kubernetes namespace '{}' for project: {}",
                        namespace_name, project.name
                    );
                }
                Err(kube::Error::Api(err)) if err.code == 404 => {
                    // Namespace doesn't exist, that's fine
                    info!(
                        "Kubernetes namespace '{}' did not exist for project: {} (already deleted)",
                        namespace_name, project.name
                    );
                }
                Err(e) => {
                    warn!(
                        "Failed to delete Kubernetes namespace '{}' for project {}: {}",
                        namespace_name, project.name, e
                    );
                    // Continue to next project, will retry on next loop
                    continue;
                }
            }

            // Remove finalizer so project can be deleted
            db_projects::remove_finalizer(
                &self.state.db_pool,
                project.id,
                KUBERNETES_NAMESPACE_FINALIZER,
            )
            .await?;
            info!(
                "Removed Kubernetes namespace finalizer from project: {}, cleanup complete",
                project.name
            );
        }

        Ok(())
    }

    /// Refresh image pull secrets for all projects with active deployments
    async fn refresh_image_pull_secrets(&self) -> Result<()> {
        // Skip if the registry provider doesn't require pull secret management
        if !self.registry_provider.requires_pull_secret() {
            debug!("Registry provider does not require pull secret management, skipping refresh");
            return Ok(());
        }

        // Skip if using externally-managed secrets
        if self.image_pull_secret_name.is_some() {
            debug!("Using externally-managed imagePullSecret, skipping refresh");
            return Ok(());
        }

        // Get registry provider
        let provider = &self.registry_provider;

        // Find all projects that have active Kubernetes deployments
        // We'll refresh the secret for each namespace (one per project)
        let healthy_deployments =
            db_deployments::find_by_status(&self.state.db_pool, DeploymentStatus::Healthy).await?;
        let unhealthy_deployments =
            db_deployments::find_by_status(&self.state.db_pool, DeploymentStatus::Unhealthy)
                .await?;
        let deployments = [healthy_deployments, unhealthy_deployments].concat();

        // Group deployments by namespace, collecting the project_id for each
        use std::collections::{HashMap, HashSet};
        let mut namespace_to_project_id: HashMap<String, uuid::Uuid> = HashMap::new();

        for deployment in &deployments {
            let metadata: Option<KubernetesMetadata> =
                serde_json::from_value(deployment.controller_metadata.clone()).ok();

            let Some(metadata) = metadata else {
                continue; // Not a Kubernetes deployment
            };

            let Some(namespace) = metadata.namespace else {
                continue;
            };

            namespace_to_project_id
                .entry(namespace)
                .or_insert(deployment.project_id);
        }

        // Batch-fetch all needed projects in one query
        let project_ids: Vec<uuid::Uuid> = namespace_to_project_id
            .values()
            .copied()
            .collect::<HashSet<_>>()
            .into_iter()
            .collect();
        let projects = db_projects::find_by_ids(&self.state.db_pool, &project_ids).await?;
        let project_name_map: HashMap<uuid::Uuid, String> =
            projects.into_iter().map(|p| (p.id, p.name)).collect();

        // Build final namespace -> project_name map
        let mut namespaces_to_refresh: HashMap<String, String> = HashMap::new();
        for (namespace, project_id) in namespace_to_project_id {
            if let Some(name) = project_name_map.get(&project_id) {
                namespaces_to_refresh.insert(namespace, name.clone());
            }
        }

        // Refresh secret for each namespace
        for (namespace, project_name) in namespaces_to_refresh {
            if let Err(e) = self
                .refresh_namespace_pull_secret(&namespace, &project_name, provider)
                .await
            {
                warn!(
                    "Failed to refresh pull secret for namespace {}: {}",
                    namespace, e
                );
            } else {
                debug!("Refreshed pull secret for namespace {}", namespace);
            }
        }

        Ok(())
    }

    /// Refresh the image pull secret for a specific namespace
    async fn refresh_namespace_pull_secret(
        &self,
        namespace: &str,
        project_name: &str,
        provider: &Arc<dyn RegistryProvider>,
    ) -> Result<()> {
        let secret_api: Api<Secret> = Api::namespaced(self.kube_client.clone(), namespace);

        // Check if secret exists and get its last refresh time
        match secret_api.get(IMAGE_PULL_SECRET_NAME).await {
            Ok(existing_secret) => {
                // Check annotation for last refresh time
                if let Some(annotations) = &existing_secret.metadata.annotations {
                    if let Some(last_refresh_str) = annotations.get(ANNOTATION_LAST_REFRESH) {
                        // Parse the timestamp
                        if let Ok(last_refresh) =
                            chrono::DateTime::parse_from_rfc3339(last_refresh_str)
                        {
                            let age =
                                Utc::now().signed_duration_since(last_refresh.with_timezone(&Utc));

                            // Refresh if older than 6 hours (50% of 12-hour ECR token lifetime)
                            if age.num_seconds() < 6 * 3600 {
                                debug!(
                                    "Secret in namespace {} is fresh (age: {}s), skipping refresh",
                                    namespace,
                                    age.num_seconds()
                                );
                                return Ok(());
                            }
                        }
                    }
                }

                // If we get here, either annotation is missing or secret is old enough
                debug!(
                    "Secret in namespace {} needs refresh (missing annotation or expired)",
                    namespace
                );
            }
            Err(kube::Error::Api(ae)) if ae.code == 404 => {
                // Secret doesn't exist, we'll create it below
                debug!(
                    "Secret in namespace {} does not exist, will create",
                    namespace
                );
            }
            Err(e) => return Err(e.into()),
        }

        // Get fresh credentials and create/update secret
        let credentials = provider.get_k8s_pull_credentials(project_name).await?;

        let secret = self.create_dockerconfigjson_secret(
            IMAGE_PULL_SECRET_NAME,
            provider.registry_host(),
            &credentials,
        )?;

        secret_api
            .patch(
                IMAGE_PULL_SECRET_NAME,
                &PatchParams::apply("rise-controller").force(),
                &Patch::Apply(&secret),
            )
            .await?;

        info!(
            "Refreshed image pull secret for namespace {} (project-wide secret)",
            namespace
        );
        Ok(())
    }

    /// Create a dockerconfigjson Secret for image pulling
    fn create_dockerconfigjson_secret(
        &self,
        name: &str,
        registry_host: &str,
        credentials: &RegistryCredentials,
    ) -> Result<Secret> {
        use base64::Engine;

        let auths_entry = match credentials.auth_method {
            RegistryAuthMethod::LoginCredentials => {
                let auth = base64::engine::general_purpose::STANDARD
                    .encode(format!("{}:{}", credentials.username, credentials.password));
                serde_json::json!({
                    "username": credentials.username,
                    "password": credentials.password,
                    "auth": auth,
                })
            }
            RegistryAuthMethod::RegistryToken => {
                serde_json::json!({ "registrytoken": credentials.password })
            }
        };

        let docker_config = serde_json::json!({ "auths": { registry_host: auths_entry } });

        let docker_config_bytes = docker_config.to_string().into_bytes();

        let mut data = BTreeMap::new();
        data.insert(
            ".dockerconfigjson".to_string(),
            k8s_openapi::ByteString(docker_config_bytes),
        );

        // Add annotation with current timestamp for tracking refresh time
        let mut annotations = BTreeMap::new();
        annotations.insert(ANNOTATION_LAST_REFRESH.to_string(), Utc::now().to_rfc3339());

        Ok(Secret {
            metadata: ObjectMeta {
                name: Some(name.to_string()),
                annotations: Some(annotations),
                ..Default::default()
            },
            type_: Some("kubernetes.io/dockerconfigjson".to_string()),
            data: Some(data),
            ..Default::default()
        })
    }

    /// Get namespace name for a project
    fn namespace_name(project: &Project) -> String {
        format!("rise-{}", project.name)
    }

    /// Sanitize a string to be a valid Kubernetes label value.
    /// Delegates to [`normalize_deployment_group`] which replaces sequences of invalid
    /// characters with `--` (e.g., "mr/26" → "mr--26") and trims non-alphanumeric
    /// leading/trailing characters.
    ///
    /// The result is guaranteed to be <= 63 chars because deployment group validation
    /// (`is_valid_group_name`) rejects names whose normalized form exceeds this limit.
    fn sanitize_label_value(value: &str) -> String {
        crate::server::deployment::models::normalize_deployment_group(value)
    }

    /// Get the escaped deployment group name for use in resource names
    fn escaped_group_name(deployment_group: &str) -> String {
        if deployment_group == crate::server::deployment::models::DEFAULT_DEPLOYMENT_GROUP {
            "default".to_string()
        } else {
            Self::sanitize_label_value(deployment_group)
        }
    }

    /// Get Service name for a deployment group
    fn service_name(_project: &Project, deployment: &Deployment) -> String {
        Self::escaped_group_name(&deployment.deployment_group)
    }

    /// Get Ingress name for a deployment group
    fn ingress_name(_project: &Project, deployment: &Deployment) -> String {
        Self::escaped_group_name(&deployment.deployment_group)
    }

    /// Get Ingress name for custom domains
    fn custom_domain_ingress_name(_project: &Project, deployment: &Deployment) -> String {
        format!(
            "{}-custom-domains",
            Self::escaped_group_name(&deployment.deployment_group)
        )
    }

    /// Get NetworkPolicy name for a deployment group
    fn network_policy_name(_project: &Project, deployment: &Deployment) -> String {
        Self::escaped_group_name(&deployment.deployment_group)
    }

    /// Get hostname (without path) for deployment
    #[allow(dead_code)]
    fn hostname(&self, project: &Project, deployment: &Deployment) -> String {
        let url = self.resolved_ingress_url(project, deployment);
        let parsed = Self::parse_ingress_url(&url);
        parsed.host
    }

    /// Parse a fully-resolved URL into (host, path_prefix)
    ///
    /// Examples:
    ///   "myapp.apps.rise.dev" → IngressUrl { host: "myapp.apps.rise.dev", path_prefix: None }
    ///   "rise.dev/myapp" → IngressUrl { host: "rise.dev", path_prefix: Some("/myapp") }
    fn parse_ingress_url(url: &str) -> IngressUrl {
        match url.find('/') {
            Some(slash_pos) => {
                let host = url[..slash_pos].to_string();
                let path = url[slash_pos..].to_string();
                IngressUrl {
                    host,
                    path_prefix: Some(path),
                }
            }
            None => IngressUrl {
                host: url.to_string(),
                path_prefix: None,
            },
        }
    }

    /// Get fully resolved ingress URL for a deployment group (no Deployment object required)
    fn resolved_ingress_url_for_group(&self, project: &Project, deployment_group: &str) -> String {
        if deployment_group == crate::server::deployment::models::DEFAULT_DEPLOYMENT_GROUP {
            self.production_ingress_url_template
                .replace("{project_name}", &project.name)
        } else if let Some(ref staging_template) = self.staging_ingress_url_template {
            staging_template
                .replace("{project_name}", &project.name)
                .replace(
                    "{deployment_group}",
                    &Self::escaped_group_name(deployment_group),
                )
        } else {
            let base_url = self
                .production_ingress_url_template
                .replace("{project_name}", &project.name);
            if let Some(dot_pos) = base_url.find('.') {
                format!(
                    "{}-{}{}",
                    &base_url[..dot_pos],
                    Self::escaped_group_name(deployment_group),
                    &base_url[dot_pos..]
                )
            } else {
                format!(
                    "{}-{}",
                    base_url,
                    Self::escaped_group_name(deployment_group)
                )
            }
        }
    }

    /// Apply port to a host URL string
    fn full_ingress_url_from_host(&self, url: &str) -> String {
        if let Some(port) = self.ingress_port {
            let parsed = Self::parse_ingress_url(url);
            let host_with_port = format!("{}:{}", parsed.host, port);
            match parsed.path_prefix {
                Some(path) => format!("{}{}", host_with_port, path),
                None => host_with_port,
            }
        } else {
            url.to_string()
        }
    }

    /// Get fully resolved ingress URL with placeholders replaced
    fn resolved_ingress_url(&self, project: &Project, deployment: &Deployment) -> String {
        if deployment.deployment_group
            == crate::server::deployment::models::DEFAULT_DEPLOYMENT_GROUP
        {
            self.production_ingress_url_template
                .replace("{project_name}", &project.name)
        } else if let Some(ref staging_template) = self.staging_ingress_url_template {
            staging_template
                .replace("{project_name}", &project.name)
                .replace(
                    "{deployment_group}",
                    &Self::escaped_group_name(&deployment.deployment_group),
                )
        } else {
            // Fallback: insert "-{group}" before first dot
            let base_url = self
                .production_ingress_url_template
                .replace("{project_name}", &project.name);
            if let Some(dot_pos) = base_url.find('.') {
                format!(
                    "{}-{}{}",
                    &base_url[..dot_pos],
                    Self::escaped_group_name(&deployment.deployment_group),
                    &base_url[dot_pos..]
                )
            } else {
                format!(
                    "{}-{}",
                    base_url,
                    Self::escaped_group_name(&deployment.deployment_group)
                )
            }
        }
    }

    /// Get parsed URL components for Ingress creation
    fn ingress_url_components(&self, project: &Project, deployment: &Deployment) -> IngressUrl {
        let url = self.resolved_ingress_url(project, deployment);
        Self::parse_ingress_url(&url)
    }

    /// Get full ingress URL for deployment_url field
    fn full_ingress_url(&self, project: &Project, deployment: &Deployment) -> String {
        let url = self.resolved_ingress_url(project, deployment);

        if let Some(port) = self.ingress_port {
            // Parse URL to separate host and path
            let parsed = Self::parse_ingress_url(&url);

            // Append port to host
            let host_with_port = format!("{}:{}", parsed.host, port);

            // Reconstruct URL
            match parsed.path_prefix {
                Some(path) => format!("{}{}", host_with_port, path),
                None => host_with_port,
            }
        } else {
            url
        }
    }

    /// Filter custom domains to remove any that would conflict with project default domain patterns.
    /// This is a defensive check in case invalid domains were added before validation was implemented
    /// or if the default domain configuration changed.
    fn filter_valid_custom_domains(
        &self,
        custom_domains: Vec<crate::db::models::CustomDomain>,
    ) -> Vec<crate::db::models::CustomDomain> {
        use crate::server::custom_domains::validation;

        custom_domains
            .into_iter()
            .filter(|domain| {
                match validation::validate_custom_domain(
                    &domain.domain,
                    &self.production_ingress_url_template,
                    self.staging_ingress_url_template.as_deref(),
                    None, // Kubernetes controller doesn't have access to public_url
                ) {
                    Ok(()) => true,
                    Err(reason) => {
                        warn!(
                            domain_id = %domain.id,
                            project_id = %domain.project_id,
                            domain = %domain.domain,
                            "Ignoring custom domain that conflicts with project default domain pattern: {}",
                            reason
                        );
                        false
                    }
                }
            })
            .collect()
    }

    /// Clean up deployment group resources (Service and Ingress) if no other deployments exist in the group
    async fn cleanup_group_resources_if_empty(
        &self,
        deployment: &Deployment,
        metadata: &KubernetesMetadata,
    ) -> Result<()> {
        let namespace = metadata
            .namespace
            .as_ref()
            .ok_or_else(|| anyhow::anyhow!("No namespace in metadata"))?;

        // Check if there are other active deployments in this group
        use crate::db::deployments as db_deployments;
        let other_in_group = db_deployments::list_for_project_and_group(
            &self.state.db_pool,
            deployment.project_id,
            Some(&deployment.deployment_group),
            Some(100), // Get up to 100 deployments to check
            None,
        )
        .await?;

        // Count deployments that aren't this one and aren't in terminal/cleanup states
        let active_count = other_in_group
            .iter()
            .filter(|d| {
                d.id != deployment.id
                    && !matches!(
                        d.status,
                        DeploymentStatus::Terminating
                            | DeploymentStatus::Cancelled
                            | DeploymentStatus::Stopped
                            | DeploymentStatus::Failed
                            | DeploymentStatus::Superseded
                            | DeploymentStatus::Expired
                    )
            })
            .count();

        if active_count == 0 {
            info!(
                "Last deployment in group '{}', cleaning up Service and Ingress",
                deployment.deployment_group
            );

            // Get project info to construct resource names
            use crate::db::projects as db_projects;
            let project = db_projects::find_by_id(&self.state.db_pool, deployment.project_id)
                .await?
                .ok_or_else(|| anyhow::anyhow!("Project not found"))?;

            // Delete Service
            let service_name = Self::service_name(&project, deployment);
            let svc_api: Api<Service> = Api::namespaced(self.kube_client.clone(), namespace);
            if let Err(e) = svc_api
                .delete(&service_name, &DeleteParams::default())
                .await
            {
                if !e.to_string().contains("404") {
                    warn!("Error deleting Service {}: {}", service_name, e);
                }
            } else {
                info!("Deleted Service {} for empty group", service_name);
            }

            // Delete primary Ingress
            let ingress_name = Self::ingress_name(&project, deployment);
            let ingress_api: Api<Ingress> = Api::namespaced(self.kube_client.clone(), namespace);
            if let Err(e) = ingress_api
                .delete(&ingress_name, &DeleteParams::default())
                .await
            {
                if !e.to_string().contains("404") {
                    warn!("Error deleting primary Ingress {}: {}", ingress_name, e);
                }
            } else {
                info!("Deleted primary Ingress {} for empty group", ingress_name);
            }

            // Delete custom domain Ingress
            let custom_domain_ingress_name = Self::custom_domain_ingress_name(&project, deployment);
            if let Err(e) = ingress_api
                .delete(&custom_domain_ingress_name, &DeleteParams::default())
                .await
            {
                if !e.to_string().contains("404") {
                    warn!(
                        "Error deleting custom domain Ingress {}: {}",
                        custom_domain_ingress_name, e
                    );
                }
            } else {
                info!(
                    "Deleted custom domain Ingress {} for empty group",
                    custom_domain_ingress_name
                );
            }

            // Delete NetworkPolicy
            let np_name = Self::network_policy_name(&project, deployment);
            let np_api: Api<NetworkPolicy> = Api::namespaced(self.kube_client.clone(), namespace);
            if let Err(e) = np_api.delete(&np_name, &DeleteParams::default()).await {
                if !e.to_string().contains("404") {
                    warn!("Error deleting NetworkPolicy {}: {}", np_name, e);
                }
            } else {
                info!("Deleted NetworkPolicy {} for empty group", np_name);
            }
        } else {
            debug!(
                "Group '{}' still has {} active deployment(s), keeping resources",
                deployment.deployment_group, active_count
            );
        }

        Ok(())
    }

    /// Check pods for irrecoverable errors (e.g., InvalidImageName, ImagePullBackOff)
    /// Returns (is_failed, error_message) tuple
    async fn check_pod_errors(
        &self,
        namespace: &str,
        _deploy_name: &str,
        deployment_id: &str,
    ) -> Result<(bool, Option<String>)> {
        use k8s_openapi::api::core::v1::Pod;

        let pod_api: Api<Pod> = Api::namespaced(self.kube_client.clone(), namespace);

        // List pods owned by this Deployment
        let pods = pod_api
            .list(
                &kube::api::ListParams::default()
                    .labels(&format!("{}={}", LABEL_DEPLOYMENT_ID, deployment_id)),
            )
            .await?;

        for pod in pods.items {
            if let Some(status) = pod.status {
                // Check container statuses for irrecoverable errors
                if let Some(container_statuses) = status.container_statuses {
                    for container_status in container_statuses {
                        if let Some(waiting) = container_status
                            .state
                            .as_ref()
                            .and_then(|s| s.waiting.as_ref())
                        {
                            let reason = waiting.reason.as_deref().unwrap_or("");

                            // Check for irrecoverable errors using shared constant
                            let is_irrecoverable =
                                IRRECOVERABLE_CONTAINER_REASONS.contains(&reason);

                            if is_irrecoverable {
                                let message = waiting.message.as_deref().unwrap_or(reason);
                                warn!(
                                    "Pod {} has irrecoverable error: {} - {}",
                                    pod.metadata.name.as_deref().unwrap_or("unknown"),
                                    reason,
                                    message
                                );
                                return Ok((true, Some(format!("{}: {}", reason, message))));
                            }
                        }

                        // Check for terminated containers with non-zero exit codes
                        if let Some(terminated) = container_status
                            .state
                            .as_ref()
                            .and_then(|s| s.terminated.as_ref())
                        {
                            if terminated.exit_code != 0 {
                                let reason =
                                    terminated.reason.as_deref().unwrap_or("ContainerFailed");
                                let default_message =
                                    format!("Exit code: {}", terminated.exit_code);
                                let message =
                                    terminated.message.as_deref().unwrap_or(&default_message);

                                // Only fail if container has restarted multiple times
                                if container_status.restart_count >= 3 {
                                    warn!(
                                        "Pod {} container has failed {} times: {} - {}",
                                        pod.metadata.name.as_deref().unwrap_or("unknown"),
                                        container_status.restart_count,
                                        reason,
                                        message
                                    );
                                    return Ok((
                                        true,
                                        Some(format!(
                                            "{}: {} (restarts: {})",
                                            reason, message, container_status.restart_count
                                        )),
                                    ));
                                }
                            }
                        }
                    }
                }
            }
        }

        Ok((false, None))
    }

    /// Extract error message from pod status if there's an irrecoverable error
    fn extract_pod_error(pod_status: &PodStatus) -> Option<String> {
        for pod in &pod_status.pods {
            // Check for container errors (waiting state with bad reasons)
            for container in &pod.containers {
                if let Some(state) = &container.state {
                    if state.state_type == "waiting" {
                        if let Some(reason) = &state.reason {
                            // Check for irrecoverable errors using shared constant
                            let is_irrecoverable =
                                IRRECOVERABLE_CONTAINER_REASONS.contains(&reason.as_str());

                            if is_irrecoverable {
                                let message = state
                                    .message
                                    .as_ref()
                                    .and_then(|m| m.lines().next())
                                    .unwrap_or(reason);
                                return Some(format!("{}: {}", reason, message));
                            }
                        }
                    } else if state.state_type == "terminated" {
                        // Check for repeated failures (3+ restarts with non-zero exit)
                        if let Some(exit_code) = state.exit_code {
                            if exit_code != 0 && container.restart_count >= 3 {
                                let reason = state.reason.as_deref().unwrap_or("ContainerFailed");
                                let default_msg = format!("Exit code: {}", exit_code);
                                let message = state.message.as_deref().unwrap_or(&default_msg);
                                return Some(format!(
                                    "{}: {} (restarts: {})",
                                    reason, message, container.restart_count
                                ));
                            }
                        }
                    }
                }
            }

            // Check for critical events
            for event in &pod.events {
                // Error events are always critical
                if event.type_ == "Error" {
                    return Some(format!("{}: {}", event.reason, event.message));
                }

                // Warning events are only critical if they're specific known issues
                // or repeated multiple times (to filter out transient warnings)
                if event.type_ == "Warning" {
                    let is_critical_reason =
                        CRITICAL_EVENT_REASONS.contains(&event.reason.as_str());
                    let is_repeated = event.count >= MIN_EVENT_COUNT_FOR_CRITICAL;

                    if is_critical_reason || is_repeated {
                        return Some(format!("{}: {}", event.reason, event.message));
                    }
                }
            }
        }

        None
    }

    /// Collect current Pod status for a deployment
    async fn collect_pod_status(
        &self,
        namespace: &str,
        deploy_name: &str,
        deployment_id: &str,
    ) -> Result<PodStatus> {
        use k8s_openapi::api::{apps::v1::Deployment as K8sDeployment, core::v1::Pod};
        use std::collections::{HashMap, HashSet};

        // Get Deployment to know desired replica count
        let deploy_api: Api<K8sDeployment> = Api::namespaced(self.kube_client.clone(), namespace);
        let deployment = deploy_api.get(deploy_name).await?;
        let desired_replicas = deployment.spec.and_then(|s| s.replicas).unwrap_or(1);

        // List pods owned by this Deployment
        let pod_api: Api<Pod> = Api::namespaced(self.kube_client.clone(), namespace);
        let pods = pod_api
            .list(
                &kube::api::ListParams::default()
                    .labels(&format!("{}={}", LABEL_DEPLOYMENT_ID, deployment_id)),
            )
            .await?;
        let current_replicas = pods.items.len() as i32;
        let pod_names: HashSet<String> = pods
            .items
            .iter()
            .filter_map(|pod| pod.metadata.name.clone())
            .collect();

        let ready_count = pods
            .items
            .iter()
            .filter(|pod| {
                let conditions = pod
                    .status
                    .as_ref()
                    .and_then(|s| s.conditions.as_ref())
                    .map(|conds| {
                        conds
                            .iter()
                            .any(|c| c.type_ == "Ready" && c.status == "True")
                    })
                    .unwrap_or(false);
                let all_containers_ready = pod
                    .status
                    .as_ref()
                    .and_then(|s| s.container_statuses.as_ref())
                    .map(|statuses| !statuses.is_empty() && statuses.iter().all(|cs| cs.ready))
                    .unwrap_or(false);
                conditions || all_containers_ready
            })
            .count() as i32;

        // Fetch Pod events (server-side filtered by kind to avoid listing all namespace events)
        let event_api: Api<Event> = Api::namespaced(self.kube_client.clone(), namespace);
        let all_events = event_api
            .list(
                &kube::api::ListParams::default()
                    .fields("involvedObject.kind=Pod")
                    .limit(MAX_EVENT_QUERY_LIMIT),
            )
            .await?;

        // Group events by pod name (only Warning/Error events for matching pods)
        let mut events_by_pod: HashMap<String, Vec<PodEvent>> = HashMap::new();
        for event in all_events.items {
            if let Some(event_type) = &event.type_ {
                let involved_obj = &event.involved_object;
                if involved_obj.kind.as_deref() == Some("Pod") {
                    if let Some(pod_name) = &involved_obj.name {
                        if !pod_names.contains(pod_name) {
                            continue;
                        }
                        // Only include Warning and Error events
                        if event_type == "Warning" || event_type == "Error" {
                            let pod_event = PodEvent {
                                type_: event_type.clone(),
                                reason: truncate_text(
                                    &event.reason.clone().unwrap_or_default(),
                                    MAX_POD_TEXT_CHARS,
                                ),
                                message: truncate_text(
                                    &event.message.clone().unwrap_or_default(),
                                    MAX_POD_TEXT_CHARS,
                                ),
                                count: event.count.unwrap_or(1),
                                last_timestamp: event
                                    .last_timestamp
                                    .as_ref()
                                    .and_then(|ts| {
                                        chrono::DateTime::from_timestamp(
                                            ts.0.as_second(),
                                            ts.0.subsec_nanosecond() as u32,
                                        )
                                    })
                                    .unwrap_or_else(Utc::now),
                            };
                            events_by_pod
                                .entry(pod_name.clone())
                                .or_default()
                                .push(pod_event);
                        }
                    }
                }
            }
        }

        // Sort events by timestamp (most recent first) and cap per pod
        for events in events_by_pod.values_mut() {
            events.sort_by_key(|e| std::cmp::Reverse(e.last_timestamp));
            events.truncate(MAX_EVENTS_PER_POD);
        }

        let mut pod_infos = Vec::new();

        for pod in pods.items.iter().take(MAX_PODS_IN_STATUS) {
            let pod_name = pod.metadata.name.clone().unwrap_or_default();

            let phase = pod
                .status
                .as_ref()
                .and_then(|s| s.phase.as_ref())
                .map(|p| match p.as_str() {
                    "Running" => PodPhase::Running,
                    "Pending" => PodPhase::Pending,
                    "Failed" => PodPhase::Failed,
                    "Succeeded" => PodPhase::Succeeded,
                    _ => PodPhase::Unknown,
                })
                .unwrap_or(PodPhase::Unknown);

            // Collect pod conditions
            let conditions: Vec<PodCondition> = pod
                .status
                .as_ref()
                .and_then(|s| s.conditions.as_ref())
                .map(|conds| {
                    conds
                        .iter()
                        .take(MAX_CONDITIONS_PER_POD)
                        .map(|c| PodCondition {
                            type_: c.type_.clone(),
                            status: c.status.clone(),
                            reason: c.reason.clone(),
                            message: truncate_optional_text(c.message.clone(), MAX_POD_TEXT_CHARS),
                        })
                        .collect()
                })
                .unwrap_or_default();

            // Collect container statuses
            let containers: Vec<ContainerStatusInfo> = pod
                .status
                .as_ref()
                .and_then(|s| s.container_statuses.as_ref())
                .map(|css| {
                    css.iter()
                        .take(MAX_CONTAINERS_PER_POD)
                        .map(|cs| {
                            let state = if let Some(waiting) =
                                cs.state.as_ref().and_then(|s| s.waiting.as_ref())
                            {
                                Some(ContainerState {
                                    state_type: "waiting".to_string(),
                                    reason: waiting.reason.clone(),
                                    message: truncate_optional_text(
                                        waiting.message.clone(),
                                        MAX_POD_TEXT_CHARS,
                                    ),
                                    exit_code: None,
                                })
                            } else if let Some(_running) =
                                cs.state.as_ref().and_then(|s| s.running.as_ref())
                            {
                                Some(ContainerState {
                                    state_type: "running".to_string(),
                                    reason: None,
                                    message: None,
                                    exit_code: None,
                                })
                            } else {
                                cs.state.as_ref().and_then(|s| s.terminated.as_ref()).map(
                                    |terminated| ContainerState {
                                        state_type: "terminated".to_string(),
                                        reason: terminated.reason.clone(),
                                        message: truncate_optional_text(
                                            terminated.message.clone(),
                                            MAX_POD_TEXT_CHARS,
                                        ),
                                        exit_code: Some(terminated.exit_code),
                                    },
                                )
                            };

                            ContainerStatusInfo {
                                name: cs.name.clone(),
                                ready: cs.ready,
                                restart_count: cs.restart_count,
                                state,
                            }
                        })
                        .collect()
                })
                .unwrap_or_default();

            // Get events for this pod
            let events = events_by_pod.remove(&pod_name).unwrap_or_default();

            pod_infos.push(PodInfo {
                name: pod_name,
                phase,
                conditions,
                containers,
                events,
            });
        }

        Ok(PodStatus {
            desired_replicas,
            ready_replicas: ready_count,
            current_replicas,
            pods: pod_infos,
            last_checked: Utc::now(),
        })
    }

    /// Create common labels for all resources
    #[allow(dead_code)] // Will be used in reconciliation implementation
    fn common_labels(project: &Project) -> BTreeMap<String, String> {
        let mut labels = BTreeMap::new();
        labels.insert(LABEL_MANAGED_BY.to_string(), "rise".to_string());
        labels.insert(LABEL_PROJECT.to_string(), project.name.clone());
        labels
    }

    /// Create deployment group labels (for resources shared across deployments in a group)
    /// Used by NetworkPolicy podSelector to target all pods in a deployment group.
    fn group_labels(project: &Project, deployment: &Deployment) -> BTreeMap<String, String> {
        let mut labels = Self::common_labels(project);
        labels.insert(
            LABEL_DEPLOYMENT_GROUP.to_string(),
            Self::sanitize_label_value(&deployment.deployment_group),
        );
        labels
    }

    /// Create deployment-specific labels
    #[allow(dead_code)] // Will be used in reconciliation implementation
    fn deployment_labels(project: &Project, deployment: &Deployment) -> BTreeMap<String, String> {
        let mut labels = Self::common_labels(project);
        labels.insert(
            LABEL_DEPLOYMENT_GROUP.to_string(),
            Self::sanitize_label_value(&deployment.deployment_group),
        );
        labels.insert(
            LABEL_DEPLOYMENT_ID.to_string(),
            deployment.deployment_id.clone(),
        );
        labels.insert(LABEL_DEPLOYMENT_UUID.to_string(), deployment.id.to_string());
        labels
    }

    /// Apply all supporting resources for a deployment (idempotent)
    /// - Namespace
    /// - ImagePullSecret (if needed)
    /// - BackendService (if configured)
    /// - Service
    /// - Ingress
    ///
    /// Uses in-memory version cache to skip unnecessary applies
    async fn apply_supporting_resources(
        &self,
        deployment: &Deployment,
        project: &Project,
        metadata: &mut KubernetesMetadata,
    ) -> Result<()> {
        let namespace = Self::namespace_name(project);

        // 1. Apply Namespace
        self.apply_namespace(deployment.id, project, &namespace)
            .await?;

        // 2. Apply ImagePullSecret (if needed)
        if self.image_pull_secret_name.is_none() {
            self.apply_image_pull_secret(
                deployment.id,
                project,
                &namespace,
                &self.registry_provider,
            )
            .await?;
        }

        // 3. Apply BackendService (if configured)
        if let Some(ref backend_address) = self.backend_address {
            self.apply_backend_service(deployment.id, project, &namespace, backend_address)
                .await?;
        }

        // 4. Apply Service
        self.apply_service(deployment.id, project, deployment, &namespace, metadata)
            .await?;

        // 5. Apply Ingress
        self.apply_ingress(deployment.id, project, deployment, &namespace, metadata)
            .await?;

        // 6. Apply NetworkPolicy
        self.apply_network_policy(deployment.id, project, deployment, &namespace)
            .await?;

        Ok(())
    }

    /// Apply namespace with drift detection
    async fn apply_namespace(
        &self,
        deployment_id: uuid::Uuid,
        project: &Project,
        namespace_name: &str,
    ) -> Result<()> {
        let ns_api: Api<Namespace> = Api::all(self.kube_client.clone());

        match ns_api.get(namespace_name).await {
            Ok(existing_ns) => {
                let current_version = existing_ns.metadata.resource_version.as_deref();

                if self.needs_apply(deployment_id, "namespace", current_version) {
                    let ns = self.create_namespace(project);
                    let result = ns_api
                        .patch(
                            namespace_name,
                            &PatchParams::apply("rise-controller"),
                            &Patch::Apply(&ns),
                        )
                        .await?;
                    self.update_version_cache(
                        deployment_id,
                        "namespace",
                        result.metadata.resource_version,
                    );
                    info!("Applied namespace '{}' (drift detected)", namespace_name);
                } else {
                    debug!("Namespace '{}' is up-to-date", namespace_name);
                }
            }
            Err(kube::Error::Api(err)) if err.code == 404 => {
                let ns = self.create_namespace(project);
                let result = ns_api.create(&PostParams::default(), &ns).await?;
                self.update_version_cache(
                    deployment_id,
                    "namespace",
                    result.metadata.resource_version,
                );

                // Add finalizer to project
                use crate::db::projects as db_projects;
                db_projects::add_finalizer(
                    &self.state.db_pool,
                    project.id,
                    KUBERNETES_NAMESPACE_FINALIZER,
                )
                .await?;

                info!("Created namespace '{}'", namespace_name);
            }
            Err(e) => return Err(e.into()),
        }

        Ok(())
    }

    /// Apply image pull secret with drift detection
    async fn apply_image_pull_secret(
        &self,
        deployment_id: uuid::Uuid,
        project: &Project,
        namespace: &str,
        registry_provider: &Arc<dyn RegistryProvider>,
    ) -> Result<()> {
        let secret_name = IMAGE_PULL_SECRET_NAME;
        let secret_api: Api<Secret> = Api::namespaced(self.kube_client.clone(), namespace);

        match secret_api.get(secret_name).await {
            Ok(existing_secret) => {
                let current_version = existing_secret.metadata.resource_version.as_deref();

                if self.needs_apply(deployment_id, "image-pull-secret", current_version) {
                    let credentials = registry_provider
                        .get_k8s_pull_credentials(&project.name)
                        .await?;
                    let secret = self.create_dockerconfigjson_secret(
                        secret_name,
                        registry_provider.registry_host(),
                        &credentials,
                    )?;
                    let result = secret_api
                        .patch(
                            secret_name,
                            &PatchParams::apply("rise-controller").force(),
                            &Patch::Apply(&secret),
                        )
                        .await?;
                    self.update_version_cache(
                        deployment_id,
                        "image-pull-secret",
                        result.metadata.resource_version,
                    );
                    info!(
                        project = project.name,
                        namespace = namespace,
                        "Applied image pull secret (drift detected)"
                    );
                } else {
                    debug!("Image pull secret is up-to-date");
                }
            }
            Err(kube::Error::Api(err)) if err.code == 404 => {
                let credentials = registry_provider
                    .get_k8s_pull_credentials(&project.name)
                    .await?;
                let secret = self.create_dockerconfigjson_secret(
                    secret_name,
                    registry_provider.registry_host(),
                    &credentials,
                )?;
                let result = secret_api.create(&PostParams::default(), &secret).await?;
                self.update_version_cache(
                    deployment_id,
                    "image-pull-secret",
                    result.metadata.resource_version,
                );
                info!(
                    project = project.name,
                    namespace = namespace,
                    "Created image pull secret"
                );
            }
            Err(e) => return Err(e.into()),
        }

        Ok(())
    }

    /// Apply backend service with drift detection
    async fn apply_backend_service(
        &self,
        deployment_id: uuid::Uuid,
        project: &Project,
        namespace: &str,
        backend_address: &crate::server::settings::BackendAddress,
    ) -> Result<()> {
        let service_name = "rise-backend";
        let service_api: Api<Service> = Api::namespaced(self.kube_client.clone(), namespace);

        match service_api.get(service_name).await {
            Ok(existing_svc) => {
                let current_version = existing_svc.metadata.resource_version.as_deref();

                if self.needs_apply(deployment_id, "backend-service", current_version) {
                    if backend_address.is_ip_address() {
                        // ClusterIP with Endpoints
                        let svc = self.create_backend_service_clusterip(
                            project,
                            namespace,
                            backend_address.port,
                        );
                        let result = service_api
                            .patch(
                                service_name,
                                &PatchParams::apply("rise").force(),
                                &Patch::Apply(&svc),
                            )
                            .await?;
                        self.update_version_cache(
                            deployment_id,
                            "backend-service",
                            result.metadata.resource_version,
                        );

                        // Apply Endpoints
                        let endpoints = self.create_backend_endpoints(
                            project,
                            namespace,
                            &backend_address.host,
                            backend_address.port,
                        );
                        let endpoints_api: Api<k8s_openapi::api::core::v1::Endpoints> =
                            Api::namespaced(self.kube_client.clone(), namespace);
                        endpoints_api
                            .patch(
                                service_name,
                                &PatchParams::apply("rise").force(),
                                &Patch::Apply(&endpoints),
                            )
                            .await?;

                        info!(
                            project = project.name,
                            backend_ip = backend_address.host,
                            backend_port = backend_address.port,
                            "Applied backend service (ClusterIP+Endpoints, drift detected)"
                        );
                    } else {
                        // ExternalName - clean up Endpoints if they exist from previous IP-based config
                        let endpoints_api: Api<k8s_openapi::api::core::v1::Endpoints> =
                            Api::namespaced(self.kube_client.clone(), namespace);
                        if let Err(e) = endpoints_api
                            .delete(service_name, &DeleteParams::default())
                            .await
                        {
                            // Ignore 404 errors (Endpoints don't exist, which is fine)
                            if !e.to_string().contains("404") {
                                warn!(
                                    "Failed to delete orphaned Endpoints during ExternalName transition: {}",
                                    e
                                );
                            }
                        } else {
                            info!(
                                project = project.name,
                                "Deleted orphaned Endpoints (switched from IP to DNS backend)"
                            );
                        }

                        let svc = self.create_backend_service_externalname(
                            project,
                            namespace,
                            &backend_address.host,
                        );
                        let result = service_api
                            .patch(
                                service_name,
                                &PatchParams::apply("rise").force(),
                                &Patch::Apply(&svc),
                            )
                            .await?;
                        self.update_version_cache(
                            deployment_id,
                            "backend-service",
                            result.metadata.resource_version,
                        );
                        info!(
                            project = project.name,
                            backend_dns = backend_address.host,
                            backend_port = backend_address.port,
                            "Applied backend service (ExternalName, drift detected)"
                        );
                    }
                } else {
                    debug!("Backend service is up-to-date");
                }
            }
            Err(kube::Error::Api(err)) if err.code == 404 => {
                // Create new service
                if backend_address.is_ip_address() {
                    let svc = self.create_backend_service_clusterip(
                        project,
                        namespace,
                        backend_address.port,
                    );
                    let result = service_api.create(&PostParams::default(), &svc).await?;
                    self.update_version_cache(
                        deployment_id,
                        "backend-service",
                        result.metadata.resource_version,
                    );

                    let endpoints = self.create_backend_endpoints(
                        project,
                        namespace,
                        &backend_address.host,
                        backend_address.port,
                    );
                    let endpoints_api: Api<k8s_openapi::api::core::v1::Endpoints> =
                        Api::namespaced(self.kube_client.clone(), namespace);
                    endpoints_api
                        .create(&PostParams::default(), &endpoints)
                        .await?;

                    info!(
                        project = project.name,
                        backend_ip = backend_address.host,
                        backend_port = backend_address.port,
                        "Created backend service (ClusterIP+Endpoints)"
                    );
                } else {
                    // ExternalName - clean up any orphaned Endpoints
                    let endpoints_api: Api<k8s_openapi::api::core::v1::Endpoints> =
                        Api::namespaced(self.kube_client.clone(), namespace);
                    if let Err(e) = endpoints_api
                        .delete(service_name, &DeleteParams::default())
                        .await
                    {
                        // Ignore 404 errors (expected when creating for the first time)
                        if !e.to_string().contains("404") {
                            warn!(
                                "Failed to delete orphaned Endpoints before ExternalName creation: {}",
                                e
                            );
                        }
                    }

                    let svc = self.create_backend_service_externalname(
                        project,
                        namespace,
                        &backend_address.host,
                    );
                    let result = service_api.create(&PostParams::default(), &svc).await?;
                    self.update_version_cache(
                        deployment_id,
                        "backend-service",
                        result.metadata.resource_version,
                    );
                    info!(
                        project = project.name,
                        backend_dns = backend_address.host,
                        backend_port = backend_address.port,
                        "Created backend service (ExternalName)"
                    );
                }
            }
            Err(e) => return Err(e.into()),
        }

        Ok(())
    }

    /// Apply service with drift detection
    async fn apply_service(
        &self,
        deployment_id: uuid::Uuid,
        project: &Project,
        deployment: &Deployment,
        namespace: &str,
        metadata: &mut KubernetesMetadata,
    ) -> Result<()> {
        let service_name = Self::service_name(project, deployment);
        let service_api: Api<Service> = Api::namespaced(self.kube_client.clone(), namespace);

        match service_api.get(&service_name).await {
            Ok(existing_svc) => {
                let current_version = existing_svc.metadata.resource_version.as_deref();

                if self.needs_apply(deployment_id, "service", current_version) {
                    let svc = self.create_service(project, deployment, metadata);
                    let result = service_api
                        .patch(
                            &service_name,
                            &PatchParams::apply("rise").force(),
                            &Patch::Apply(&svc),
                        )
                        .await?;
                    self.update_version_cache(
                        deployment_id,
                        "service",
                        result.metadata.resource_version,
                    );
                    info!("Applied Service '{}' (drift detected)", service_name);
                } else {
                    debug!("Service '{}' is up-to-date", service_name);
                }
            }
            Err(kube::Error::Api(err)) if err.code == 404 => {
                let svc = self.create_service(project, deployment, metadata);
                let result = service_api.create(&PostParams::default(), &svc).await?;
                self.update_version_cache(
                    deployment_id,
                    "service",
                    result.metadata.resource_version,
                );
                info!("Created Service '{}'", service_name);
            }
            Err(e) => return Err(e.into()),
        }

        metadata.service_name = Some(service_name);
        Ok(())
    }

    /// Apply ingress with drift detection
    /// This now creates two separate Ingresses when custom domains exist:
    /// 1. Primary ingress for the default project URL (may include path-based routing)
    /// 2. Custom domain ingress for custom domains (always uses root path)
    async fn apply_ingress(
        &self,
        deployment_id: uuid::Uuid,
        project: &Project,
        deployment: &Deployment,
        namespace: &str,
        metadata: &mut KubernetesMetadata,
    ) -> Result<()> {
        let ingress_name = Self::ingress_name(project, deployment);
        let custom_domain_ingress_name = Self::custom_domain_ingress_name(project, deployment);

        // Fetch custom domains
        use crate::db::custom_domains as db_custom_domains;
        let custom_domains =
            db_custom_domains::list_project_custom_domains(&self.state.db_pool, project.id).await?;

        // Filter out custom domains that conflict with project default patterns
        let custom_domains = self.filter_valid_custom_domains(custom_domains);

        let ingress_api: Api<Ingress> = Api::namespaced(self.kube_client.clone(), namespace);

        // 1. Apply primary ingress (default project URL)
        match ingress_api.get(&ingress_name).await {
            Ok(existing_ingress) => {
                let current_version = existing_ingress.metadata.resource_version.as_deref();

                // When needs_reconcile is set (e.g., access_class changed), use replace
                // to fully replace the resource including removing old annotations.
                // SSA patch may not properly remove annotations when switching access classes.
                if deployment.needs_reconcile {
                    let mut ingress = self.create_primary_ingress(project, deployment, metadata);
                    // Set resourceVersion for optimistic concurrency
                    ingress.metadata.resource_version = current_version.map(String::from);
                    let result = ingress_api
                        .replace(&ingress_name, &PostParams::default(), &ingress)
                        .await?;
                    self.update_version_cache(
                        deployment_id,
                        "ingress",
                        result.metadata.resource_version,
                    );
                    info!(
                        "Replaced primary Ingress '{}' with updated annotations (needs_reconcile)",
                        ingress_name
                    );
                } else if self.needs_apply(deployment_id, "ingress", current_version) {
                    let ingress = self.create_primary_ingress(project, deployment, metadata);
                    let result = ingress_api
                        .patch(
                            &ingress_name,
                            &PatchParams::apply("rise").force(),
                            &Patch::Apply(&ingress),
                        )
                        .await?;
                    self.update_version_cache(
                        deployment_id,
                        "ingress",
                        result.metadata.resource_version,
                    );
                    info!(
                        "Applied primary Ingress '{}' (drift detected)",
                        ingress_name
                    );
                } else {
                    debug!("Primary Ingress '{}' is up-to-date", ingress_name);
                }
            }
            Err(kube::Error::Api(err)) if err.code == 404 => {
                let ingress = self.create_primary_ingress(project, deployment, metadata);
                let result = ingress_api.create(&PostParams::default(), &ingress).await?;
                self.update_version_cache(
                    deployment_id,
                    "ingress",
                    result.metadata.resource_version,
                );
                info!("Created primary Ingress '{}'", ingress_name);
            }
            Err(e) => return Err(e.into()),
        }

        // 2. Handle custom domain ingress
        if !custom_domains.is_empty() {
            // Custom domains exist - create/update custom domain ingress
            match ingress_api.get(&custom_domain_ingress_name).await {
                Ok(existing_ingress) => {
                    let current_version = existing_ingress.metadata.resource_version.as_deref();

                    // When needs_reconcile is set, use replace to fully replace the resource
                    if deployment.needs_reconcile {
                        let mut ingress = self.create_custom_domain_ingress(
                            project,
                            deployment,
                            metadata,
                            &custom_domains,
                        );
                        ingress.metadata.resource_version = current_version.map(String::from);
                        let result = ingress_api
                            .replace(
                                &custom_domain_ingress_name,
                                &PostParams::default(),
                                &ingress,
                            )
                            .await?;
                        self.update_version_cache(
                            deployment_id,
                            "ingress-custom-domains",
                            result.metadata.resource_version,
                        );
                        info!(
                            "Replaced custom domain Ingress '{}' with updated annotations (needs_reconcile)",
                            custom_domain_ingress_name
                        );
                    } else if self.needs_apply(
                        deployment_id,
                        "ingress-custom-domains",
                        current_version,
                    ) {
                        let ingress = self.create_custom_domain_ingress(
                            project,
                            deployment,
                            metadata,
                            &custom_domains,
                        );
                        let result = ingress_api
                            .patch(
                                &custom_domain_ingress_name,
                                &PatchParams::apply("rise").force(),
                                &Patch::Apply(&ingress),
                            )
                            .await?;
                        self.update_version_cache(
                            deployment_id,
                            "ingress-custom-domains",
                            result.metadata.resource_version,
                        );
                        info!(
                            "Applied custom domain Ingress '{}' (drift detected)",
                            custom_domain_ingress_name
                        );
                    } else {
                        debug!(
                            "Custom domain Ingress '{}' is up-to-date",
                            custom_domain_ingress_name
                        );
                    }
                }
                Err(kube::Error::Api(err)) if err.code == 404 => {
                    let ingress = self.create_custom_domain_ingress(
                        project,
                        deployment,
                        metadata,
                        &custom_domains,
                    );
                    let result = ingress_api.create(&PostParams::default(), &ingress).await?;
                    self.update_version_cache(
                        deployment_id,
                        "ingress-custom-domains",
                        result.metadata.resource_version,
                    );
                    info!(
                        "Created custom domain Ingress '{}'",
                        custom_domain_ingress_name
                    );
                }
                Err(e) => return Err(e.into()),
            }
        } else {
            // No custom domains - delete custom domain ingress if it exists
            match ingress_api.get(&custom_domain_ingress_name).await {
                Ok(_) => {
                    // Custom domain ingress exists but shouldn't - delete it
                    match ingress_api
                        .delete(&custom_domain_ingress_name, &DeleteParams::default())
                        .await
                    {
                        Ok(_) => {
                            info!(
                                "Deleted custom domain Ingress '{}' (no custom domains configured)",
                                custom_domain_ingress_name
                            );
                            // Clear cache entry for deleted ingress
                            self.update_version_cache(
                                deployment_id,
                                "ingress-custom-domains",
                                None,
                            );
                        }
                        Err(e) if !e.to_string().contains("404") => {
                            warn!(
                                "Error deleting custom domain Ingress '{}': {}",
                                custom_domain_ingress_name, e
                            );
                        }
                        _ => {}
                    }
                }
                Err(kube::Error::Api(err)) if err.code == 404 => {
                    // Doesn't exist, that's fine
                    debug!(
                        "Custom domain Ingress '{}' doesn't exist (expected)",
                        custom_domain_ingress_name
                    );
                }
                Err(e) => {
                    // Log but don't fail - this is just cleanup
                    warn!(
                        "Error checking for custom domain Ingress '{}': {}",
                        custom_domain_ingress_name, e
                    );
                }
            }
        }

        metadata.ingress_name = Some(ingress_name);
        Ok(())
    }

    /// Look up the ClusterIP of the `kubernetes` service in the `default` namespace.
    ///
    /// The result is cached for 5 minutes. On lookup failure, a previously cached value is
    /// retained so that a transient error doesn't remove a working egress rule.
    ///
    /// We use the service ClusterIP rather than pod selector labels because the labels on
    /// kube-apiserver pods are not standardized across Kubernetes distributions (e.g. kubeadm
    /// uses `component: kube-apiserver`, others use `component: apiserver`, and managed clusters
    /// like EKS/GKE may not expose API server pods at all). The `kubernetes` service in the
    /// `default` namespace is part of the Kubernetes spec and always exists with a stable,
    /// immutable ClusterIP.
    /// Apply NetworkPolicy with drift detection
    async fn apply_network_policy(
        &self,
        deployment_id: uuid::Uuid,
        project: &Project,
        deployment: &Deployment,
        namespace: &str,
    ) -> Result<()> {
        let np_name = Self::network_policy_name(project, deployment);
        let np_api: Api<NetworkPolicy> = Api::namespaced(self.kube_client.clone(), namespace);

        match np_api.get(&np_name).await {
            Ok(existing_np) => {
                let current_version = existing_np.metadata.resource_version.as_deref();

                if self.needs_apply(deployment_id, "network-policy", current_version) {
                    let np = self.create_network_policy(project, deployment, namespace);
                    let result = np_api
                        .patch(
                            &np_name,
                            &PatchParams::apply("rise").force(),
                            &Patch::Apply(&np),
                        )
                        .await?;
                    self.update_version_cache(
                        deployment_id,
                        "network-policy",
                        result.metadata.resource_version,
                    );
                    info!("Applied NetworkPolicy '{}' (drift detected)", np_name);
                } else {
                    debug!("NetworkPolicy '{}' is up-to-date", np_name);
                }
            }
            Err(kube::Error::Api(err)) if err.code == 404 => {
                let np = self.create_network_policy(project, deployment, namespace);
                let result = np_api.create(&PostParams::default(), &np).await?;
                self.update_version_cache(
                    deployment_id,
                    "network-policy",
                    result.metadata.resource_version,
                );
                info!("Created NetworkPolicy '{}'", np_name);
            }
            Err(e) => return Err(e.into()),
        }

        Ok(())
    }

    /// Create Namespace resource
    fn create_namespace(&self, project: &Project) -> Namespace {
        // Start with common labels and merge in configured namespace labels
        let mut labels = Self::common_labels(project);
        for (k, v) in &self.namespace_labels {
            labels.insert(k.clone(), v.clone());
        }

        // Convert HashMap to BTreeMap for annotations
        let annotations = if !self.namespace_annotations.is_empty() {
            Some(
                self.namespace_annotations
                    .iter()
                    .map(|(k, v)| (k.clone(), v.clone()))
                    .collect(),
            )
        } else {
            None
        };

        Namespace {
            metadata: ObjectMeta {
                name: Some(Self::namespace_name(project)),
                labels: Some(labels),
                annotations,
                ..Default::default()
            },
            ..Default::default()
        }
    }

    /// Create Service resource
    fn create_service(
        &self,
        project: &Project,
        deployment: &Deployment,
        metadata: &KubernetesMetadata,
    ) -> Service {
        Service {
            metadata: ObjectMeta {
                name: Some(Self::service_name(project, deployment)),
                namespace: metadata.namespace.clone(),
                labels: Some(Self::common_labels(project)),
                ..Default::default()
            },
            spec: Some(ServiceSpec {
                type_: Some("ClusterIP".to_string()),
                selector: Some(Self::deployment_labels(project, deployment)),
                ports: Some(vec![ServicePort {
                    name: Some("http".to_string()),
                    port: 80,
                    target_port: Some(
                        k8s_openapi::apimachinery::pkg::util::intstr::IntOrString::Int(
                            metadata.http_port as i32,
                        ),
                    ),
                    protocol: Some("TCP".to_string()),
                    ..Default::default()
                }]),
                ..Default::default()
            }),
            ..Default::default()
        }
    }

    /// Create ExternalName service for routing to Rise backend (DNS names)
    fn create_backend_service_externalname(
        &self,
        project: &Project,
        namespace: &str,
        external_name: &str,
    ) -> Service {
        Service {
            metadata: ObjectMeta {
                name: Some("rise-backend".to_string()),
                namespace: Some(namespace.to_string()),
                labels: Some(Self::common_labels(project)),
                ..Default::default()
            },
            spec: Some(ServiceSpec {
                type_: Some("ExternalName".to_string()),
                external_name: Some(external_name.to_string()),
                // Note: ports are not used by ExternalName services
                // (kube-proxy returns CNAME record, actual port is in Ingress config)
                ..Default::default()
            }),
            ..Default::default()
        }
    }

    /// Create ClusterIP service for routing to Rise backend (IP addresses)
    /// Requires manual Endpoints resource to specify the target IP
    fn create_backend_service_clusterip(
        &self,
        project: &Project,
        namespace: &str,
        port: u16,
    ) -> Service {
        Service {
            metadata: ObjectMeta {
                name: Some("rise-backend".to_string()),
                namespace: Some(namespace.to_string()),
                labels: Some(Self::common_labels(project)),
                ..Default::default()
            },
            spec: Some(ServiceSpec {
                type_: Some("ClusterIP".to_string()),
                // No selector - we'll create manual Endpoints
                ports: Some(vec![ServicePort {
                    name: Some("http".to_string()),
                    port: port as i32,
                    target_port: Some(
                        k8s_openapi::apimachinery::pkg::util::intstr::IntOrString::Int(port as i32),
                    ),
                    protocol: Some("TCP".to_string()),
                    ..Default::default()
                }]),
                ..Default::default()
            }),
            ..Default::default()
        }
    }

    /// Create Endpoints for ClusterIP backend service pointing to an IP address
    fn create_backend_endpoints(
        &self,
        project: &Project,
        namespace: &str,
        ip: &str,
        port: u16,
    ) -> k8s_openapi::api::core::v1::Endpoints {
        use k8s_openapi::api::core::v1::{EndpointAddress, EndpointPort, EndpointSubset};

        k8s_openapi::api::core::v1::Endpoints {
            metadata: ObjectMeta {
                name: Some("rise-backend".to_string()),
                namespace: Some(namespace.to_string()),
                labels: Some(Self::common_labels(project)),
                ..Default::default()
            },
            subsets: Some(vec![EndpointSubset {
                addresses: Some(vec![EndpointAddress {
                    ip: ip.to_string(),
                    ..Default::default()
                }]),
                ports: Some(vec![EndpointPort {
                    name: Some("http".to_string()),
                    port: port as i32,
                    protocol: Some("TCP".to_string()),
                    ..Default::default()
                }]),
                ..Default::default()
            }]),
        }
    }

    /// Create NetworkPolicy for a deployment group from explicit config
    fn create_network_policy(
        &self,
        project: &Project,
        deployment: &Deployment,
        namespace: &str,
    ) -> NetworkPolicy {
        let (egress_rules, policy_types) = match &self.network_policy.egress {
            None => (None, vec!["Ingress".to_string()]),
            Some(rules) => (
                Some(rules.clone()),
                vec!["Ingress".to_string(), "Egress".to_string()],
            ),
        };

        NetworkPolicy {
            metadata: ObjectMeta {
                name: Some(Self::network_policy_name(project, deployment)),
                namespace: Some(namespace.to_string()),
                labels: Some(Self::common_labels(project)),
                ..Default::default()
            },
            spec: Some(NetworkPolicySpec {
                pod_selector: Some(LabelSelector {
                    match_labels: Some(Self::group_labels(project, deployment)),
                    ..Default::default()
                }),
                policy_types: Some(policy_types),
                egress: egress_rules,
                ingress: Some(self.network_policy.ingress.clone()),
            }),
        }
    }

    /// Load and decrypt environment variables for a deployment
    async fn load_env_vars(
        &self,
        _project: &crate::db::Project,
        deployment: &crate::db::Deployment,
    ) -> Result<Vec<k8s_openapi::api::core::v1::EnvVar>> {
        use k8s_openapi::api::core::v1::EnvVar;

        // Load and decrypt environment variables using shared helper
        let env_vars = crate::db::env_vars::load_deployment_env_vars_decrypted(
            &self.state.db_pool,
            deployment.id,
            self.state.encryption_provider.as_deref(),
        )
        .await?;

        // Format as Kubernetes EnvVar objects
        // Note: PORT, RISE_ISSUER, RISE_APP_URL, and RISE_APP_URLS
        // are persisted in deployment_env_vars and loaded from the database
        let k8s_env_vars: Vec<EnvVar> = env_vars
            .into_iter()
            .map(|(key, value)| EnvVar {
                name: key,
                value: Some(value),
                ..Default::default()
            })
            .collect();

        Ok(k8s_env_vars)
    }

    /// Create Pod-level security context with restricted profile
    /// Returns None if pod_security_enabled is false
    fn create_pod_security_context(&self) -> Option<PodSecurityContext> {
        if !self.pod_security_enabled {
            return None;
        }

        Some(PodSecurityContext {
            run_as_non_root: Some(true), // Enforce non-root, but let image choose UID
            seccomp_profile: Some(SeccompProfile {
                type_: "RuntimeDefault".to_string(),
                ..Default::default()
            }),
            ..Default::default()
        })
    }

    /// Create Container-level security context with minimal privileges
    /// Returns None if pod_security_enabled is false
    fn create_container_security_context(&self) -> Option<SecurityContext> {
        if !self.pod_security_enabled {
            return None;
        }

        Some(SecurityContext {
            allow_privilege_escalation: Some(false),
            run_as_non_root: Some(true),
            capabilities: Some(Capabilities {
                drop: Some(vec!["ALL".to_string()]),
                ..Default::default()
            }),
            read_only_root_filesystem: Some(false), // Writable filesystem for compatibility
            ..Default::default()
        })
    }

    /// Create resource requirements from configuration
    /// Returns resource requirements with configured or default values
    fn create_resource_requirements(&self) -> Option<ResourceRequirements> {
        use k8s_openapi::apimachinery::pkg::api::resource::Quantity;

        // Use configured values or defaults from PodResourceLimits::default()
        let config = self.pod_resources.clone().unwrap_or_default();
        let cpu_request = config.cpu_request;
        let memory_request = config.memory_request;
        let cpu_limit = config.cpu_limit;
        let memory_limit = config.memory_limit;

        Some(ResourceRequirements {
            requests: Some({
                let mut map = BTreeMap::new();
                map.insert("cpu".to_string(), Quantity(cpu_request));
                map.insert("memory".to_string(), Quantity(memory_request));
                map
            }),
            limits: Some({
                let mut map = BTreeMap::new();
                map.insert("cpu".to_string(), Quantity(cpu_limit));
                map.insert("memory".to_string(), Quantity(memory_limit));
                map
            }),
            ..Default::default()
        })
    }

    /// Create HTTP health probe for application port
    /// Returns None if probes are disabled
    fn create_http_probe(&self, port: i32, probe_type: ProbeType) -> Option<Probe> {
        // Get config or use defaults
        let config = self.health_probes.as_ref().cloned().unwrap_or_else(|| {
            crate::server::settings::HealthProbeConfig {
                liveness_enabled: true,
                readiness_enabled: true,
                path: "/".to_string(),
                initial_delay_seconds: 10,
                period_seconds: 10,
                timeout_seconds: 5,
                failure_threshold: 3,
            }
        });

        // Check if this probe type is enabled
        let enabled = match probe_type {
            ProbeType::Liveness => config.liveness_enabled,
            ProbeType::Readiness => config.readiness_enabled,
        };

        if !enabled {
            return None;
        }

        // Validate and normalize probe path
        let path = if config.path.is_empty() || !config.path.starts_with('/') {
            warn!(
                "Invalid health probe path '{}', using default '/'",
                config.path
            );
            "/".to_string()
        } else {
            config.path.clone()
        };

        Some(Probe {
            http_get: Some(HTTPGetAction {
                path: Some(path),
                port: k8s_openapi::apimachinery::pkg::util::intstr::IntOrString::Int(port),
                scheme: Some("HTTP".to_string()),
                ..Default::default()
            }),
            initial_delay_seconds: Some(config.initial_delay_seconds),
            period_seconds: Some(config.period_seconds),
            timeout_seconds: Some(config.timeout_seconds),
            failure_threshold: Some(config.failure_threshold),
            success_threshold: Some(1),
            ..Default::default()
        })
    }

    fn create_extra_service_token_volume(&self) -> Option<Volume> {
        if self.extra_service_token_audiences.is_empty() {
            return None;
        }

        let mut token_names: Vec<_> = self.extra_service_token_audiences.keys().cloned().collect();
        token_names.sort();

        let sources = token_names
            .into_iter()
            .map(|name| VolumeProjection {
                service_account_token: Some(ServiceAccountTokenProjection {
                    audience: Some(
                        self.extra_service_token_audiences
                            .get(&name)
                            .expect("token name collected from map keys")
                            .clone(),
                    ),
                    path: name,
                    ..Default::default()
                }),
                ..Default::default()
            })
            .collect();

        Some(Volume {
            name: EXTRA_SERVICE_TOKENS_VOLUME_NAME.to_string(),
            projected: Some(ProjectedVolumeSource {
                sources: Some(sources),
                ..Default::default()
            }),
            ..Default::default()
        })
    }

    fn create_extra_service_token_volume_mount(&self) -> Option<VolumeMount> {
        if self.extra_service_token_audiences.is_empty() {
            return None;
        }

        Some(VolumeMount {
            name: EXTRA_SERVICE_TOKENS_VOLUME_NAME.to_string(),
            mount_path: EXTRA_SERVICE_TOKENS_MOUNT_PATH.to_string(),
            read_only: Some(true),
            ..Default::default()
        })
    }

    /// Create Deployment resource
    async fn create_deployment(
        &self,
        project: &Project,
        deployment: &Deployment,
        metadata: &KubernetesMetadata,
        env_vars: Vec<k8s_openapi::api::core::v1::EnvVar>,
    ) -> K8sDeployment {
        // Build image reference - same logic as API layer's get_deployment_image_tag()
        // This ensures rollback deployments use the correct source image tag
        let image = if let Some(ref image_digest) = deployment.image_digest {
            // Pre-built images use the pinned digest
            image_digest.clone()
        } else {
            // For rollback deployments, use the source deployment's deployment_id for the image tag
            let deployment_id_for_tag = if let Some(source_deployment_id) =
                deployment.rolled_back_from_deployment_id
            {
                // Fetch the source deployment to get its deployment_id
                match db_deployments::find_by_id(&self.state.db_pool, source_deployment_id).await {
                    Ok(Some(source_deployment)) => source_deployment.deployment_id,
                    Ok(None) => {
                        tracing::warn!(
                            "Rollback deployment {} references non-existent source deployment {}",
                            deployment.deployment_id,
                            source_deployment_id
                        );
                        deployment.deployment_id.clone()
                    }
                    Err(e) => {
                        tracing::error!(
                            "Failed to fetch source deployment {} for rollback {}: {}",
                            source_deployment_id,
                            deployment.deployment_id,
                            e
                        );
                        deployment.deployment_id.clone()
                    }
                }
            } else {
                deployment.deployment_id.clone()
            };

            // Build-from-source: construct from registry config using the appropriate deployment_id
            self.registry_provider.get_image_tag(
                &project.name,
                &deployment_id_for_tag,
                crate::server::registry::ImageTagType::Internal,
            )
        };

        let volumes = self
            .create_extra_service_token_volume()
            .map(|volume| vec![volume]);
        let volume_mounts = self
            .create_extra_service_token_volume_mount()
            .map(|mount| vec![mount]);

        K8sDeployment {
            metadata: ObjectMeta {
                name: Some(format!("{}-{}", project.name, deployment.deployment_id)),
                namespace: metadata.namespace.clone(),
                labels: Some(Self::deployment_labels(project, deployment)),
                ..Default::default()
            },
            spec: Some(DeploymentSpec {
                replicas: Some(1),
                min_ready_seconds: None,
                selector: LabelSelector {
                    match_labels: Some(Self::deployment_labels(project, deployment)),
                    ..Default::default()
                },
                strategy: Some(k8s_openapi::api::apps::v1::DeploymentStrategy {
                    type_: Some("RollingUpdate".to_string()),
                    rolling_update: Some(k8s_openapi::api::apps::v1::RollingUpdateDeployment {
                        max_surge: Some(
                            k8s_openapi::apimachinery::pkg::util::intstr::IntOrString::Int(1),
                        ),
                        max_unavailable: Some(
                            k8s_openapi::apimachinery::pkg::util::intstr::IntOrString::Int(0),
                        ),
                    }),
                }),
                template: PodTemplateSpec {
                    metadata: Some(ObjectMeta {
                        labels: Some(Self::deployment_labels(project, deployment)),
                        ..Default::default()
                    }),
                    spec: Some(PodSpec {
                        security_context: self.create_pod_security_context(),
                        image_pull_secrets: {
                            if self.registry_provider.requires_pull_secret() {
                                let secret_name = self
                                    .image_pull_secret_name
                                    .as_deref()
                                    .or(Some(IMAGE_PULL_SECRET_NAME));
                                secret_name.map(|name| {
                                    vec![LocalObjectReference {
                                        name: name.to_string(),
                                    }]
                                })
                            } else {
                                None
                            }
                        },
                        containers: vec![Container {
                            name: "app".to_string(),
                            image: Some(image),
                            ports: Some(vec![ContainerPort {
                                container_port: metadata.http_port as i32,
                                ..Default::default()
                            }]),
                            image_pull_policy: Some("Always".to_string()),
                            env: Some(env_vars), // Always set env vars (at minimum, PORT is injected)
                            security_context: self.create_container_security_context(),
                            resources: self.create_resource_requirements(),
                            liveness_probe: self
                                .create_http_probe(metadata.http_port as i32, ProbeType::Liveness),
                            readiness_probe: self
                                .create_http_probe(metadata.http_port as i32, ProbeType::Readiness),
                            volume_mounts,
                            ..Default::default()
                        }],
                        volumes,
                        node_selector: if self.node_selector.is_empty() {
                            None
                        } else {
                            Some(self.node_selector.clone().into_iter().collect())
                        },
                        host_aliases: if self.host_aliases.is_empty() {
                            None
                        } else {
                            Some(
                                self.host_aliases
                                    .iter()
                                    .map(|(hostname, ip)| HostAlias {
                                        hostnames: Some(vec![hostname.clone()]),
                                        ip: ip.clone(),
                                    })
                                    .collect(),
                            )
                        },
                        ..Default::default()
                    }),
                },
                ..Default::default()
            }),
            ..Default::default()
        }
    }

    /// Build common ingress annotations (auth annotations based on access class)
    /// Returns a BTreeMap starting with user-provided annotations, with auth annotations added if needed
    fn build_ingress_annotations(&self, project: &Project) -> BTreeMap<String, String> {
        // Start with user-provided annotations from config (convert HashMap to BTreeMap)
        let mut annotations: BTreeMap<String, String> = self
            .ingress_annotations
            .iter()
            .map(|(k, v)| (k.clone(), v.clone()))
            .collect();

        // Get access class configuration
        let access_class = self
            .access_classes
            .get(&project.access_class)
            .unwrap_or_else(|| {
                panic!(
                    "Access class '{}' not configured in deployment controller. Available: {}. \
                     This should have been validated on project creation.",
                    project.access_class,
                    self.access_classes
                        .keys()
                        .cloned()
                        .collect::<Vec<_>>()
                        .join(", ")
                )
            });

        // Apply authentication based on access requirement
        match access_class.access_requirement {
            AccessRequirement::None => {
                // No authentication required - fully public
                // No nginx auth annotations needed
            }
            AccessRequirement::Authenticated | AccessRequirement::Member => {
                // Both require authentication, but differ in authorization logic
                // The /auth/ingress endpoint will handle the membership check
                // Note: All API routes are nested under /api/v1 prefix
                let auth_url = format!(
                    "{}/api/v1/auth/ingress?project={}",
                    self.auth_backend_url, project.name
                );

                // Use dynamic host routing if backend service exists
                let signin_url = if self.backend_address.is_some() {
                    // Use $http_host to preserve port, but use literal scheme from config to avoid parsing error
                    // (nginx parser fails on $scheme:// at start of URL, but accepts http:// or https://)
                    format!(
                        "{}://$http_host/.rise/auth/signin?project={}&redirect=$scheme://$http_host$escaped_request_uri",
                        self.ingress_schema,  // "http" or "https" from config
                        urlencoding::encode(&project.name)
                    )
                } else {
                    // Fall back to configured auth_signin_url for deployments without custom domains
                    format!(
                        "{}/api/v1/auth/signin?project={}&redirect=$scheme://$http_host$escaped_request_uri",
                        self.auth_signin_url,
                        urlencoding::encode(&project.name)
                    )
                };

                // Nginx auth subrequest - returns 2xx for allowed, 401/403 for denied
                // If auth-url is unreachable or returns 5xx, nginx defaults to DENY (fail-closed)
                // IMPORTANT: Ensure auth-url points to the correct endpoint (/api/v1/auth/ingress)
                // If misconfigured to hit the frontend catch-all route, it may return 200 OK and grant access
                annotations.insert("nginx.ingress.kubernetes.io/auth-url".to_string(), auth_url);
                annotations.insert(
                    "nginx.ingress.kubernetes.io/auth-signin".to_string(),
                    signin_url,
                );
                annotations.insert(
                    "nginx.ingress.kubernetes.io/auth-response-headers".to_string(),
                    "X-Auth-Request-Email,X-Auth-Request-User".to_string(),
                );
            }
        }

        // Apply custom annotations from access class
        for (key, value) in &access_class.custom_annotations {
            annotations.insert(key.clone(), value.clone());
        }

        annotations
    }

    /// Get the ingress class for a project based on its access class
    fn get_ingress_class_for_project(&self, project: &Project) -> &str {
        let access_class = self
            .access_classes
            .get(&project.access_class)
            .unwrap_or_else(|| {
                panic!(
                    "Access class '{}' not configured in deployment controller. Available: {}. \
                     This should have been validated on project creation.",
                    project.access_class,
                    self.access_classes
                        .keys()
                        .cloned()
                        .collect::<Vec<_>>()
                        .join(", ")
                )
            });
        &access_class.ingress_class
    }

    /// Build HTTP paths for an ingress rule, including optional backend service routing
    fn build_ingress_paths(
        &self,
        service_name: &str,
        app_path: &str,
        app_path_type: &str,
    ) -> Vec<HTTPIngressPath> {
        let mut paths = vec![HTTPIngressPath {
            path: Some(app_path.to_string()),
            path_type: app_path_type.to_string(),
            backend: IngressBackend {
                service: Some(IngressServiceBackend {
                    name: service_name.to_string(),
                    port: Some(ServiceBackendPort {
                        name: Some("http".to_string()),
                        ..Default::default()
                    }),
                }),
                ..Default::default()
            },
        }];

        // Add /.rise path routing if backend service exists
        if let Some(ref backend_addr) = self.backend_address {
            paths.push(HTTPIngressPath {
                path: Some("/.rise".to_string()),
                path_type: "Prefix".to_string(),
                backend: IngressBackend {
                    service: Some(IngressServiceBackend {
                        name: "rise-backend".to_string(),
                        port: Some(ServiceBackendPort {
                            number: Some(backend_addr.port as i32),
                            ..Default::default()
                        }),
                    }),
                    ..Default::default()
                },
            });
        }

        paths
    }

    /// Create primary Ingress resource for default project URL
    fn create_primary_ingress(
        &self,
        project: &Project,
        deployment: &Deployment,
        metadata: &KubernetesMetadata,
    ) -> Ingress {
        let url_components = self.ingress_url_components(project, deployment);

        // Build common annotations
        let mut annotations = self.build_ingress_annotations(project);

        // Add Nginx rewrite annotations for sub-path routing
        // This is safe here because we only handle the primary URL (not custom domains)
        if let Some(ref path) = url_components.path_prefix {
            // Strip path prefix: /myapp/foo → /foo
            annotations.insert(
                "nginx.ingress.kubernetes.io/rewrite-target".to_string(),
                "/$2".to_string(),
            );

            // Pass original prefix as header using the built-in annotation
            // This annotation is now only applied to the primary ingress with path-based routing
            annotations.insert(
                "nginx.ingress.kubernetes.io/x-forwarded-prefix".to_string(),
                path.trim_end_matches('/').to_string(),
            );
        }

        // Determine path and path type based on routing mode
        let (ingress_path, path_type) = if let Some(ref path) = url_components.path_prefix {
            // Regex: /myapp(/|$)(.*)
            // Matches: /myapp, /myapp/, /myapp/anything
            // $2 captures everything after /myapp/
            let pattern = format!("{}(/|$)(.*)", path.trim_end_matches('/'));
            (pattern, "ImplementationSpecific")
        } else {
            ("/".to_string(), "Prefix")
        };

        // Build paths for primary host
        let service_name = Self::service_name(project, deployment);
        let primary_paths = self.build_ingress_paths(&service_name, &ingress_path, path_type);

        let rules = vec![IngressRule {
            host: Some(url_components.host.clone()),
            http: Some(HTTPIngressRuleValue {
                paths: primary_paths,
            }),
        }];

        // Build TLS configuration for primary host only
        let tls = self.build_primary_tls_config(&url_components.host);

        Ingress {
            metadata: ObjectMeta {
                name: Some(Self::ingress_name(project, deployment)),
                namespace: metadata.namespace.clone(),
                labels: Some(Self::common_labels(project)),
                // Always include annotations (even if empty) so SSA removes old keys
                // when access_class changes from authenticated to public
                annotations: Some(annotations),
                ..Default::default()
            },
            spec: Some(IngressSpec {
                ingress_class_name: Some(self.get_ingress_class_for_project(project).to_string()),
                tls,
                rules: Some(rules),
                ..Default::default()
            }),
            ..Default::default()
        }
    }

    /// Create Ingress resource for custom domains
    /// Custom domains always use root path (/) without path prefix or x-forwarded-prefix annotation
    fn create_custom_domain_ingress(
        &self,
        project: &Project,
        deployment: &Deployment,
        metadata: &KubernetesMetadata,
        custom_domains: &[crate::db::models::CustomDomain],
    ) -> Ingress {
        // Build common annotations (auth for private projects)
        // NOTE: We do NOT add x-forwarded-prefix annotation here
        // Custom domains always serve from root path (/) without any path prefix
        let mut annotations = self.build_ingress_annotations(project);

        // Add custom domain specific annotations (e.g., cert-manager annotations)
        for (k, v) in &self.custom_domain_ingress_annotations {
            annotations.insert(k.clone(), v.clone());
        }

        let service_name = Self::service_name(project, deployment);

        // Build rules for custom domains (always from root, no path prefix)
        let mut rules = Vec::new();
        for domain in custom_domains {
            let paths = self.build_ingress_paths(&service_name, "/", "Prefix");
            rules.push(IngressRule {
                host: Some(domain.domain.clone()),
                http: Some(HTTPIngressRuleValue { paths }),
            });
        }

        // Build TLS configuration for custom domains
        let tls = self.build_custom_domain_tls_config(custom_domains);

        Ingress {
            metadata: ObjectMeta {
                name: Some(Self::custom_domain_ingress_name(project, deployment)),
                namespace: metadata.namespace.clone(),
                labels: Some(Self::common_labels(project)),
                // Always include annotations (even if empty) so SSA removes old keys
                // when access_class changes from authenticated to public
                annotations: Some(annotations),
                ..Default::default()
            },
            spec: Some(IngressSpec {
                ingress_class_name: Some(self.get_ingress_class_for_project(project).to_string()),
                tls,
                rules: Some(rules),
                ..Default::default()
            }),
            ..Default::default()
        }
    }

    /// Build TLS configuration for primary host only
    fn build_primary_tls_config(
        &self,
        primary_host: &str,
    ) -> Option<Vec<k8s_openapi::api::networking::v1::IngressTLS>> {
        let shared_secret = self.ingress_tls_secret_name.as_ref()?;

        Some(vec![k8s_openapi::api::networking::v1::IngressTLS {
            hosts: Some(vec![primary_host.to_string()]),
            secret_name: Some(shared_secret.clone()),
        }])
    }

    /// Build TLS configuration for custom domains
    fn build_custom_domain_tls_config(
        &self,
        custom_domains: &[crate::db::models::CustomDomain],
    ) -> Option<Vec<k8s_openapi::api::networking::v1::IngressTLS>> {
        // Early return if no custom domains
        if custom_domains.is_empty() {
            return None;
        }

        let mut tls_configs = Vec::new();

        match self.custom_domain_tls_mode {
            crate::server::settings::CustomDomainTlsMode::Shared => {
                // All custom domains share the same TLS secret
                // Requires ingress_tls_secret_name to be configured
                let shared_secret = self.ingress_tls_secret_name.as_ref()?;

                let all_hosts: Vec<String> =
                    custom_domains.iter().map(|d| d.domain.clone()).collect();

                tls_configs.push(k8s_openapi::api::networking::v1::IngressTLS {
                    hosts: Some(all_hosts),
                    secret_name: Some(shared_secret.clone()),
                });
            }
            crate::server::settings::CustomDomainTlsMode::PerDomain => {
                // Each custom domain gets its own tls-{domain} secret
                // This works with cert-manager to automatically provision certificates
                // Does not require ingress_tls_secret_name to be configured
                for domain in custom_domains {
                    tls_configs.push(k8s_openapi::api::networking::v1::IngressTLS {
                        hosts: Some(vec![domain.domain.clone()]),
                        secret_name: Some(format!("tls-{}", domain.domain)),
                    });
                }
            }
        }

        Some(tls_configs)
    }
}

#[async_trait]
impl DeploymentBackend for KubernetesController {
    async fn reconcile(
        &self,
        deployment: &Deployment,
        project: &Project,
    ) -> Result<ReconcileResult> {
        // Wait for image to be pushed before starting Kubernetes deployment
        // The image must be available in the registry before we can create pods
        if !matches!(
            deployment.status,
            DeploymentStatus::Pushed
                | DeploymentStatus::Deploying
                | DeploymentStatus::Unhealthy
                | DeploymentStatus::Healthy
        ) {
            debug!(
                "Deployment {} not yet pushed (status={:?}), skipping Kubernetes reconciliation",
                deployment.deployment_id, deployment.status
            );
            return Ok(ReconcileResult {
                status: deployment.status.clone(),
                controller_metadata: deployment.controller_metadata.clone(),
                error_message: None,
            });
        }

        // Parse existing metadata (or create default)
        let mut metadata: KubernetesMetadata =
            serde_json::from_value(deployment.controller_metadata.clone()).unwrap_or_default();

        // Default HTTP port
        if metadata.http_port == 0 {
            metadata.http_port = deployment.http_port as u16;
        }

        // Clear Ingress cache entries if needs_reconcile is set
        // This ensures Ingress resources are re-applied when project settings change
        // (e.g., access_class changes requiring updated auth annotations)
        if deployment.needs_reconcile {
            debug!(
                "Clearing Ingress cache for deployment {} due to needs_reconcile flag",
                deployment.deployment_id
            );
            // Clear both primary and custom domain Ingress cache entries
            self.update_version_cache(deployment.id, "ingress", None);
            self.update_version_cache(deployment.id, "ingress-custom-domains", None);
        }

        debug!(
            "Reconciling deployment {} (status={:?}) in phase {:?}",
            deployment.deployment_id, deployment.status, metadata.reconcile_phase
        );

        // Once a deployment has been healthy, recovery/recreate progress must stay post-deploy.
        let status = reconcile_in_progress_status(deployment);

        // Recovery logic: If deployment is Unhealthy and Deployment is missing, reset to recreate it
        if deployment.status == DeploymentStatus::Unhealthy
            && matches!(
                metadata.reconcile_phase,
                ReconcilePhase::WaitingForDeployment
                    | ReconcilePhase::UpdatingIngress
                    | ReconcilePhase::WaitingForHealth
                    | ReconcilePhase::SwitchingTraffic
                    | ReconcilePhase::Completed
            )
        {
            // Check if Deployment still exists
            if let (Some(ref deploy_name), Some(ref namespace)) =
                (&metadata.deployment_name, &metadata.namespace)
            {
                let deploy_api: Api<K8sDeployment> =
                    Api::namespaced(self.kube_client.clone(), namespace);
                match deploy_api.get(deploy_name).await {
                    Ok(_) => {
                        // Deployment exists, continue normal reconciliation
                        debug!(
                            "Deployment {} exists, continuing normal reconciliation",
                            deploy_name
                        );
                    }
                    Err(kube::Error::Api(ae)) if ae.code == 404 => {
                        // Deployment is missing - reset to recreate it
                        warn!(
                            "Unhealthy deployment {} has missing Deployment {} in phase {:?}, resetting to recreate",
                            deployment.deployment_id, deploy_name, metadata.reconcile_phase
                        );

                        // Reset to CreatingDeployment to recreate the Deployment
                        metadata.reconcile_phase = ReconcilePhase::CreatingDeployment;
                        info!(
                            "Reset reconciliation phase to CreatingDeployment for deployment {}",
                            deployment.deployment_id
                        );
                    }
                    Err(e) => {
                        // Other errors, continue normal reconciliation (will likely fail and retry)
                        warn!(
                            "Error checking Deployment {} for unhealthy deployment {}: {}",
                            deploy_name, deployment.deployment_id, e
                        );
                    }
                }
            } else {
                // No Deployment name in metadata - this shouldn't happen in these phases
                warn!(
                    "Unhealthy deployment {} in phase {:?} has no Deployment name in metadata",
                    deployment.deployment_id, metadata.reconcile_phase
                );
            }
        }

        // Loop through phases until we hit one that requires waiting
        loop {
            match metadata.reconcile_phase {
                ReconcilePhase::NotStarted => {
                    // Initialize metadata and continue immediately
                    metadata.namespace = Some(Self::namespace_name(project));
                    metadata.service_name = Some(Self::service_name(project, deployment));
                    metadata.ingress_name = Some(Self::ingress_name(project, deployment));
                    metadata.reconcile_phase = ReconcilePhase::CreatingNamespace;
                    // Continue to next phase
                    continue;
                }

                ReconcilePhase::CreatingNamespace => {
                    let namespace = Self::namespace_name(project);
                    let ns_api: Api<Namespace> = Api::all(self.kube_client.clone());

                    // Check if namespace exists
                    match ns_api.get(&namespace).await {
                        Ok(existing_ns) => {
                            debug!("Namespace {} already exists, checking for drift", namespace);

                            // Check if annotations have drifted
                            let desired_annotations: BTreeMap<String, String> = self
                                .namespace_annotations
                                .iter()
                                .map(|(k, v)| (k.clone(), v.clone()))
                                .collect();

                            let current_annotations = existing_ns
                                .metadata
                                .annotations
                                .as_ref()
                                .cloned()
                                .unwrap_or_default();

                            // Check if our managed annotations match
                            let mut needs_update = false;
                            for (key, desired_value) in &desired_annotations {
                                if current_annotations.get(key) != Some(desired_value) {
                                    needs_update = true;
                                    break;
                                }
                            }

                            if needs_update {
                                // Merge current annotations with desired ones
                                let mut updated_annotations = current_annotations;
                                for (key, value) in desired_annotations {
                                    updated_annotations.insert(key, value);
                                }

                                // Create updated namespace with merged annotations
                                let mut updated_ns = existing_ns.clone();
                                updated_ns.metadata.annotations = Some(updated_annotations);

                                // Apply the update
                                let patch_params = PatchParams::apply("rise-controller");
                                let patch = Patch::Apply(&updated_ns);
                                ns_api.patch(&namespace, &patch_params, &patch).await?;
                                info!(
                                    "Updated namespace {} annotations to match configuration",
                                    namespace
                                );
                            }
                        }
                        Err(kube::Error::Api(ae)) if ae.code == 404 => {
                            // Create namespace
                            let ns = self.create_namespace(project);
                            ns_api.create(&PostParams::default(), &ns).await?;
                            info!("Created namespace {}", namespace);
                        }
                        Err(e) => return Err(e.into()),
                    }

                    // Add Kubernetes namespace finalizer if not already present
                    // This ensures namespace cleanup when project is deleted
                    if !project
                        .finalizers
                        .contains(&KUBERNETES_NAMESPACE_FINALIZER.to_string())
                    {
                        db_projects::add_finalizer(
                            &self.state.db_pool,
                            project.id,
                            KUBERNETES_NAMESPACE_FINALIZER,
                        )
                        .await?;
                        debug!(
                            "Added Kubernetes namespace finalizer to project: {}",
                            project.name
                        );
                    }

                    metadata.reconcile_phase = ReconcilePhase::CreatingImagePullSecret;
                    // Continue to next phase
                    continue;
                }

                ReconcilePhase::CreatingImagePullSecret => {
                    // Skip entirely when the registry provider manages pull credentials externally
                    if !self.registry_provider.requires_pull_secret() {
                        debug!(
                            project = project.name,
                            "Registry provider does not require pull secret management, skipping"
                        );
                        metadata.reconcile_phase = ReconcilePhase::CreatingBackendService;
                        continue;
                    }

                    let namespace = metadata
                        .namespace
                        .as_ref()
                        .ok_or_else(|| anyhow::anyhow!("No namespace in metadata"))?;

                    // Check if using an externally-managed image pull secret
                    if let Some(ref external_secret_name) = self.image_pull_secret_name {
                        // Using external secret - just verify it exists, don't manage it
                        let secret_api: Api<Secret> =
                            Api::namespaced(self.kube_client.clone(), namespace);

                        match secret_api.get(external_secret_name).await {
                            Ok(_) => {
                                debug!(
                                    "External imagePullSecret '{}' exists in namespace {}",
                                    external_secret_name, namespace
                                );
                            }
                            Err(kube::Error::Api(ae)) if ae.code == 404 => {
                                warn!(
                                    "External imagePullSecret '{}' not found in namespace {}. \
                                     Secret must be created manually or deployment will fail to pull images.",
                                    external_secret_name, namespace
                                );
                                // Continue anyway - deployment will fail with clear error if secret is truly missing
                            }
                            Err(e) if is_namespace_not_found_error(&e) => {
                                warn!(
                                    "Namespace missing during imagePullSecret check, resetting to CreatingNamespace"
                                );
                                metadata.reconcile_phase = ReconcilePhase::CreatingNamespace;
                                metadata.namespace = None;
                                continue;
                            }
                            Err(e) => return Err(e.into()),
                        }

                        // Skip to next phase - we don't manage external secrets
                        metadata.reconcile_phase = ReconcilePhase::CreatingBackendService;
                        continue;
                    }

                    // Using registry provider - create/update secret dynamically
                    let provider = &self.registry_provider;
                    let credentials = provider.get_k8s_pull_credentials(&project.name).await?;
                    let secret_api: Api<Secret> =
                        Api::namespaced(self.kube_client.clone(), namespace);

                    let secret = self.create_dockerconfigjson_secret(
                        IMAGE_PULL_SECRET_NAME,
                        provider.registry_host(),
                        &credentials,
                    )?;

                    // Use server-side apply to upsert the secret (creates or updates as needed)
                    match secret_api
                        .patch(
                            IMAGE_PULL_SECRET_NAME,
                            &PatchParams::apply("rise-controller").force(),
                            &Patch::Apply(&secret),
                        )
                        .await
                    {
                        Ok(_) => {
                            info!(
                                project = project.name,
                                namespace = namespace,
                                "ImagePullSecret applied (created or updated)"
                            );
                        }
                        Err(e) if is_namespace_not_found_error(&e) => {
                            warn!(
                                    "Namespace missing during ImagePullSecret apply, resetting to CreatingNamespace"
                                );
                            metadata.reconcile_phase = ReconcilePhase::CreatingNamespace;
                            metadata.namespace = None;
                            continue;
                        }
                        Err(e) => return Err(e.into()),
                    }

                    metadata.reconcile_phase = ReconcilePhase::CreatingBackendService;
                    // Continue to next phase
                    continue;
                }

                ReconcilePhase::CreatingBackendService => {
                    // Skip if backend_address not configured
                    if self.backend_address.is_none() {
                        metadata.reconcile_phase = ReconcilePhase::CreatingService;
                        continue;
                    }

                    let backend_address = self.backend_address.as_ref().unwrap();
                    let namespace = metadata
                        .namespace
                        .as_ref()
                        .ok_or_else(|| anyhow::anyhow!("No namespace in metadata"))?;

                    let service_api: Api<Service> =
                        Api::namespaced(self.kube_client.clone(), namespace);

                    if backend_address.is_ip_address() {
                        // IP address: Create ClusterIP service + manual Endpoints
                        let service = self.create_backend_service_clusterip(
                            project,
                            namespace,
                            backend_address.port,
                        );
                        let endpoints = self.create_backend_endpoints(
                            project,
                            namespace,
                            &backend_address.host,
                            backend_address.port,
                        );

                        let endpoints_api: Api<k8s_openapi::api::core::v1::Endpoints> =
                            Api::namespaced(self.kube_client.clone(), namespace);

                        // Create/update service
                        match service_api
                            .patch(
                                service.metadata.name.as_ref().unwrap(),
                                &PatchParams::apply("rise-controller"),
                                &Patch::Apply(&service),
                            )
                            .await
                        {
                            Ok(_) => {
                                info!(
                                    project = project.name,
                                    deployment_id = %deployment.id,
                                    backend_ip = backend_address.host,
                                    backend_port = backend_address.port,
                                    "Backend ClusterIP service created for /.rise/ routing"
                                );
                            }
                            Err(e) if is_namespace_not_found_error(&e) => {
                                warn!(
                                    "Namespace missing during backend service creation, resetting to CreatingNamespace"
                                );
                                metadata.reconcile_phase = ReconcilePhase::CreatingNamespace;
                                metadata.namespace = None;
                                continue;
                            }
                            Err(e) => {
                                return Err(anyhow::anyhow!(
                                    "Failed to create backend ClusterIP service: {}",
                                    e
                                ))
                            }
                        }

                        // Create/update endpoints
                        match endpoints_api
                            .patch(
                                endpoints.metadata.name.as_ref().unwrap(),
                                &PatchParams::apply("rise-controller"),
                                &Patch::Apply(&endpoints),
                            )
                            .await
                        {
                            Ok(_) => {
                                info!(
                                    project = project.name,
                                    deployment_id = %deployment.id,
                                    backend_ip = backend_address.host,
                                    backend_port = backend_address.port,
                                    "Backend Endpoints created for /.rise/ routing"
                                );
                            }
                            Err(e) if is_namespace_not_found_error(&e) => {
                                warn!(
                                    "Namespace missing during backend endpoints creation, resetting to CreatingNamespace"
                                );
                                metadata.reconcile_phase = ReconcilePhase::CreatingNamespace;
                                metadata.namespace = None;
                                continue;
                            }
                            Err(e) => {
                                return Err(anyhow::anyhow!(
                                    "Failed to create backend Endpoints: {}",
                                    e
                                ))
                            }
                        }
                    } else {
                        // DNS name: Create ExternalName service
                        let service = self.create_backend_service_externalname(
                            project,
                            namespace,
                            &backend_address.host,
                        );

                        match service_api
                            .patch(
                                service.metadata.name.as_ref().unwrap(),
                                &PatchParams::apply("rise-controller"),
                                &Patch::Apply(&service),
                            )
                            .await
                        {
                            Ok(_) => {
                                info!(
                                    project = project.name,
                                    deployment_id = %deployment.id,
                                    backend_dns = backend_address.host,
                                    backend_port = backend_address.port,
                                    "Backend ExternalName service created for /.rise/ routing"
                                );
                            }
                            Err(e) if is_namespace_not_found_error(&e) => {
                                warn!(
                                    "Namespace missing during backend service creation, resetting to CreatingNamespace"
                                );
                                metadata.reconcile_phase = ReconcilePhase::CreatingNamespace;
                                metadata.namespace = None;
                                continue;
                            }
                            Err(e) => {
                                return Err(anyhow::anyhow!(
                                    "Failed to create backend ExternalName service: {}",
                                    e
                                ))
                            }
                        }
                    }

                    metadata.reconcile_phase = ReconcilePhase::CreatingService;
                    continue;
                }

                ReconcilePhase::CreatingService => {
                    let namespace = metadata
                        .namespace
                        .as_ref()
                        .ok_or_else(|| anyhow::anyhow!("No namespace in metadata"))?;

                    let service_name = Self::service_name(project, deployment);
                    let svc_api: Api<Service> =
                        Api::namespaced(self.kube_client.clone(), namespace);

                    // Create service with server-side apply (allows future updates)
                    let svc = self.create_service(project, deployment, &metadata);
                    let patch_params = PatchParams::apply("rise").force();
                    let patch = Patch::Apply(&svc);
                    let result = match svc_api.patch(&service_name, &patch_params, &patch).await {
                        Ok(r) => r,
                        Err(e) if is_namespace_not_found_error(&e) => {
                            warn!(
                                "Namespace missing during Service creation, resetting to CreatingNamespace"
                            );
                            metadata.reconcile_phase = ReconcilePhase::CreatingNamespace;
                            metadata.namespace = None;
                            continue;
                        }
                        Err(e) => return Err(e.into()),
                    };

                    info!(
                        project = project.name,
                        deployment_id = %deployment.id,
                        resource_version = ?result.metadata.resource_version,
                        "Service applied (any drift corrected)"
                    );

                    metadata.reconcile_phase = ReconcilePhase::CreatingDeployment;
                    // Continue to next phase
                    continue;
                }

                ReconcilePhase::CreatingDeployment => {
                    let namespace = metadata
                        .namespace
                        .as_ref()
                        .ok_or_else(|| anyhow::anyhow!("No namespace in metadata"))?;

                    // MIGRATION: Check if we need to clean up an old ReplicaSet from pre-Deployment era
                    // This handles the migration from ReplicaSets to Deployments (commit 716283c)
                    let deploy_name = format!("{}-{}", project.name, deployment.deployment_id);

                    // Check if there's an old ReplicaSet with this name that needs cleanup
                    let rs_api: Api<ReplicaSet> =
                        Api::namespaced(self.kube_client.clone(), namespace);
                    match rs_api.get(&deploy_name).await {
                        Ok(_) => {
                            // Old ReplicaSet exists - delete it before creating Deployment
                            info!(
                                project = project.name,
                                deployment_id = %deployment.id,
                                replicaset = deploy_name,
                                "Found old ReplicaSet during migration, deleting before creating Deployment"
                            );

                            match rs_api.delete(&deploy_name, &DeleteParams::default()).await {
                                Ok(_) => {
                                    info!(
                                        project = project.name,
                                        deployment_id = %deployment.id,
                                        "Old ReplicaSet deleted, will create Deployment"
                                    );
                                }
                                Err(e) if is_namespace_not_found_error(&e) => {
                                    warn!(
                                        "Namespace missing during ReplicaSet migration cleanup, resetting to CreatingNamespace"
                                    );
                                    metadata.reconcile_phase = ReconcilePhase::CreatingNamespace;
                                    metadata.namespace = None;
                                    continue;
                                }
                                Err(e) => {
                                    warn!(
                                        "Failed to delete old ReplicaSet during migration: {}. Continuing anyway.",
                                        e
                                    );
                                    // Continue anyway - the Deployment creation might still work
                                }
                            }
                        }
                        Err(kube::Error::Api(ae)) if ae.code == 404 => {
                            // No ReplicaSet found - this is normal for new deployments or already-migrated ones
                            debug!(
                                project = project.name,
                                deployment_id = %deployment.id,
                                "No old ReplicaSet found (normal for new or already-migrated deployments)"
                            );
                        }
                        Err(e) if is_namespace_not_found_error(&e) => {
                            warn!(
                                "Namespace missing during ReplicaSet migration check, resetting to CreatingNamespace"
                            );
                            metadata.reconcile_phase = ReconcilePhase::CreatingNamespace;
                            metadata.namespace = None;
                            continue;
                        }
                        Err(e) => {
                            warn!(
                                "Error checking for old ReplicaSet during migration: {}. Continuing anyway.",
                                e
                            );
                            // Continue anyway - might just be a transient error
                        }
                    }

                    // Load and decrypt environment variables
                    let env_vars = self.load_env_vars(project, deployment).await?;

                    // deploy_name already calculated above for migration check
                    let deploy_api: Api<K8sDeployment> =
                        Api::namespaced(self.kube_client.clone(), namespace);

                    // Check if Deployment exists
                    match deploy_api.get(&deploy_name).await {
                        Ok(existing_deploy) => {
                            // Deployment exists - check for drift
                            let desired_deploy = self
                                .create_deployment(project, deployment, &metadata, env_vars.clone())
                                .await;

                            if self.deployment_has_drifted(&existing_deploy, &desired_deploy) {
                                info!(
                                    project = project.name,
                                    deployment_id = %deployment.id,
                                    "Deployment has drifted, recreating"
                                );

                                // Delete and wait for deletion to complete
                                if let Err(e) = self
                                    .delete_and_wait_deployment(&deploy_api, &deploy_name)
                                    .await
                                {
                                    if is_namespace_not_found_error(&e) {
                                        warn!(
                                            "Namespace missing during Deployment deletion, resetting to CreatingNamespace"
                                        );
                                        metadata.reconcile_phase =
                                            ReconcilePhase::CreatingNamespace;
                                        metadata.namespace = None;
                                        continue;
                                    } else {
                                        return Err(e);
                                    }
                                }

                                // Create new Deployment
                                match deploy_api
                                    .create(&PostParams::default(), &desired_deploy)
                                    .await
                                {
                                    Ok(_) => {
                                        info!(
                                            project = project.name,
                                            deployment_id = %deployment.id,
                                            "Deployment recreated after drift detected"
                                        );
                                    }
                                    Err(e) if is_namespace_not_found_error(&e) => {
                                        warn!(
                                            "Namespace missing during Deployment creation, resetting to CreatingNamespace"
                                        );
                                        metadata.reconcile_phase =
                                            ReconcilePhase::CreatingNamespace;
                                        metadata.namespace = None;
                                        continue;
                                    }
                                    Err(e) => return Err(e.into()),
                                }
                            } else {
                                debug!(
                                    project = project.name,
                                    deployment_id = %deployment.id,
                                    "Deployment exists and matches desired state"
                                );
                            }
                        }
                        Err(kube::Error::Api(ae)) if ae.code == 404 => {
                            // Deployment doesn't exist - create it
                            let rs = self
                                .create_deployment(project, deployment, &metadata, env_vars)
                                .await;
                            match deploy_api.create(&PostParams::default(), &rs).await {
                                Ok(_) => {
                                    info!(
                                        project = project.name,
                                        deployment_id = %deployment.id,
                                        "Deployment created"
                                    );
                                }
                                Err(e) if is_namespace_not_found_error(&e) => {
                                    warn!(
                                        "Namespace missing during Deployment creation, resetting to CreatingNamespace"
                                    );
                                    metadata.reconcile_phase = ReconcilePhase::CreatingNamespace;
                                    metadata.namespace = None;
                                    continue;
                                }
                                Err(e) => return Err(e.into()),
                            }
                        }
                        Err(e) if is_namespace_not_found_error(&e) => {
                            warn!(
                                "Namespace missing during Deployment get, resetting to CreatingNamespace"
                            );
                            metadata.reconcile_phase = ReconcilePhase::CreatingNamespace;
                            metadata.namespace = None;
                            continue;
                        }
                        Err(e) => return Err(e.into()),
                    }

                    metadata.deployment_name = Some(deploy_name);
                    metadata.reconcile_phase = ReconcilePhase::WaitingForDeployment;

                    // Return here - need to wait for pods to become ready
                    return Ok(ReconcileResult {
                        status,
                        controller_metadata: serde_json::to_value(&metadata)?,
                        error_message: None,
                    });
                }

                ReconcilePhase::WaitingForDeployment => {
                    let namespace = metadata
                        .namespace
                        .as_ref()
                        .ok_or_else(|| anyhow::anyhow!("No namespace in metadata"))?;
                    let deploy_name = metadata
                        .deployment_name
                        .as_ref()
                        .ok_or_else(|| anyhow::anyhow!("No Deployment name in metadata"))?;

                    // Collect and store pod status
                    match self
                        .collect_pod_status(namespace, deploy_name, &deployment.deployment_id)
                        .await
                    {
                        Ok(pod_status) => {
                            metadata.pod_status = Some(pod_status.clone());

                            // Check for irrecoverable errors in pods
                            if let Some(error_msg) = Self::extract_pod_error(&pod_status) {
                                error!(
                                    "Deployment {} has pod errors: {}",
                                    deployment.deployment_id, error_msg
                                );

                                // Mark deployment as Failed
                                return Ok(ReconcileResult {
                                    status: DeploymentStatus::Failed,
                                    controller_metadata: serde_json::to_value(&metadata)?,
                                    error_message: Some(error_msg),
                                });
                            }
                        }
                        Err(e) if is_namespace_not_found_error(&e) => {
                            warn!("Namespace missing during pod status collection, resetting to CreatingNamespace");
                            metadata.reconcile_phase = ReconcilePhase::CreatingNamespace;
                            metadata.namespace = None;
                            continue;
                        }
                        Err(e) => {
                            warn!("Failed to collect pod status: {:?}", e);
                            // Continue reconciliation - don't fail if we can't get pod status
                        }
                    }

                    // Check for irrecoverable pod errors first (fallback check)
                    let (has_errors, error_msg) = match self
                        .check_pod_errors(namespace, deploy_name, &deployment.deployment_id)
                        .await
                    {
                        Ok((errors, msg)) => (errors, msg),
                        Err(e) if is_namespace_not_found_error(&e) => {
                            warn!("Namespace missing during pod error check, resetting to CreatingNamespace");
                            metadata.reconcile_phase = ReconcilePhase::CreatingNamespace;
                            metadata.namespace = None;
                            continue;
                        }
                        Err(e) => return Err(e),
                    };
                    if has_errors {
                        error!(
                            "Deployment {} has irrecoverable pod errors: {}",
                            deployment.deployment_id,
                            error_msg.as_ref().unwrap_or(&"Unknown error".to_string())
                        );

                        // Mark deployment as Failed
                        return Ok(ReconcileResult {
                            status: DeploymentStatus::Failed,
                            controller_metadata: serde_json::to_value(&metadata)?,
                            error_message: error_msg,
                        });
                    }

                    let deploy_api: Api<K8sDeployment> =
                        Api::namespaced(self.kube_client.clone(), namespace);
                    let rs = match deploy_api.get(deploy_name).await {
                        Ok(r) => r,
                        Err(e) if is_namespace_not_found_error(&e) => {
                            warn!("Namespace missing during Deployment get in WaitingForDeployment, resetting to CreatingNamespace");
                            metadata.reconcile_phase = ReconcilePhase::CreatingNamespace;
                            metadata.namespace = None;
                            continue;
                        }
                        Err(e) => return Err(e.into()),
                    };

                    let spec_replicas = rs.spec.and_then(|s| s.replicas).unwrap_or(1);
                    let ready_replicas = rs.status.and_then(|s| s.ready_replicas).unwrap_or(0);

                    if ready_replicas >= spec_replicas {
                        info!(
                            "Deployment {} is ready ({}/{})",
                            deploy_name, ready_replicas, spec_replicas
                        );
                        metadata.reconcile_phase = ReconcilePhase::UpdatingIngress;
                        // Continue to updating ingress
                        continue;
                    } else {
                        debug!(
                            "Waiting for Deployment {} ({}/{})",
                            deploy_name, ready_replicas, spec_replicas
                        );

                        // Return here - still waiting for pods
                        return Ok(ReconcileResult {
                            status,
                            controller_metadata: serde_json::to_value(&metadata)?,
                            error_message: None,
                        });
                    }
                }

                ReconcilePhase::UpdatingIngress => {
                    let namespace = metadata
                        .namespace
                        .as_ref()
                        .ok_or_else(|| anyhow::anyhow!("No namespace in metadata"))?;

                    // Fetch custom domains (only for DEFAULT group)
                    let custom_domains = if deployment.deployment_group == DEFAULT_DEPLOYMENT_GROUP
                    {
                        let domains = crate::db::custom_domains::list_project_custom_domains(
                            &self.state.db_pool,
                            project.id,
                        )
                        .await
                        .unwrap_or_default();

                        // Filter out custom domains that conflict with project default patterns
                        self.filter_valid_custom_domains(domains)
                    } else {
                        Vec::new()
                    };

                    // Apply both primary and custom domain ingresses
                    let ingress_name = Self::ingress_name(project, deployment);
                    let custom_domain_ingress_name =
                        Self::custom_domain_ingress_name(project, deployment);
                    let ingress_api: Api<Ingress> =
                        Api::namespaced(self.kube_client.clone(), namespace);

                    // 1. Apply primary ingress
                    let primary_ingress =
                        self.create_primary_ingress(project, deployment, &metadata);
                    let patch_params = PatchParams::apply("rise").force();
                    let patch = Patch::Apply(&primary_ingress);
                    let result = match ingress_api
                        .patch(&ingress_name, &patch_params, &patch)
                        .await
                    {
                        Ok(r) => r,
                        Err(e) if is_namespace_not_found_error(&e) => {
                            warn!(
                                "Namespace missing during primary Ingress creation, resetting to CreatingNamespace"
                            );
                            metadata.reconcile_phase = ReconcilePhase::CreatingNamespace;
                            metadata.namespace = None;
                            continue;
                        }
                        Err(e) => return Err(e.into()),
                    };

                    info!(
                        project = project.name,
                        deployment_id = %deployment.id,
                        resource_version = ?result.metadata.resource_version,
                        "Primary Ingress applied (any drift corrected)"
                    );

                    // 2. Handle custom domain ingress
                    if !custom_domains.is_empty() {
                        let custom_ingress = self.create_custom_domain_ingress(
                            project,
                            deployment,
                            &metadata,
                            &custom_domains,
                        );
                        let patch = Patch::Apply(&custom_ingress);
                        match ingress_api
                            .patch(&custom_domain_ingress_name, &patch_params, &patch)
                            .await
                        {
                            Ok(_) => {
                                info!(
                                    project = project.name,
                                    deployment_id = %deployment.id,
                                    "Custom domain Ingress applied"
                                );
                            }
                            Err(e) if is_namespace_not_found_error(&e) => {
                                warn!(
                                    "Namespace missing during custom domain Ingress creation, resetting to CreatingNamespace"
                                );
                                metadata.reconcile_phase = ReconcilePhase::CreatingNamespace;
                                metadata.namespace = None;
                                continue;
                            }
                            Err(e) => return Err(e.into()),
                        }
                    } else {
                        // No custom domains - delete custom domain ingress if it exists
                        match ingress_api
                            .delete(&custom_domain_ingress_name, &DeleteParams::default())
                            .await
                        {
                            Ok(_) => {
                                info!(
                                    project = project.name,
                                    deployment_id = %deployment.id,
                                    "Deleted custom domain Ingress (no custom domains configured)"
                                );
                            }
                            Err(e) if !e.to_string().contains("404") => {
                                warn!(
                                    "Error deleting custom domain Ingress '{}': {}",
                                    custom_domain_ingress_name, e
                                );
                            }
                            _ => {}
                        }
                    }

                    metadata.reconcile_phase = ReconcilePhase::WaitingForHealth;
                    // Continue to health check
                    continue;
                }

                ReconcilePhase::WaitingForHealth => {
                    // Use health_check to verify deployment is healthy before switching traffic
                    let health = self.health_check(deployment).await?;

                    if health.healthy {
                        info!(
                            "Deployment {} is healthy, ready for traffic switch",
                            deployment.deployment_id
                        );
                        metadata.reconcile_phase = ReconcilePhase::SwitchingTraffic;

                        // Continue to traffic switching
                        continue;
                    } else {
                        debug!(
                            "Waiting for deployment {} to become healthy",
                            deployment.deployment_id
                        );

                        return Ok(ReconcileResult {
                            status,
                            controller_metadata: serde_json::to_value(&metadata)?,
                            error_message: health.message,
                        });
                    }
                }

                ReconcilePhase::SwitchingTraffic => {
                    let namespace = metadata
                        .namespace
                        .as_ref()
                        .ok_or_else(|| anyhow::anyhow!("No namespace in metadata"))?;

                    // BLUE/GREEN TRAFFIC SWITCH: Update Service selector to point to new deployment
                    let service_name = Self::service_name(project, deployment);
                    let svc_api: Api<Service> =
                        Api::namespaced(self.kube_client.clone(), namespace);

                    // Create updated service with selector pointing to this deployment
                    let svc = self.create_service(project, deployment, &metadata);

                    // Use server-side apply with force to update the service selector
                    let patch_params = PatchParams::apply("rise").force();
                    let patch = Patch::Apply(&svc);
                    match svc_api.patch(&service_name, &patch_params, &patch).await {
                        Ok(_) => {
                            info!(
                                "Switched traffic: Service {} selector now points to deployment {}",
                                service_name, deployment.deployment_id
                            );
                        }
                        Err(e) if is_namespace_not_found_error(&e) => {
                            warn!(
                                "Namespace missing during traffic switch, resetting to CreatingNamespace"
                            );
                            metadata.reconcile_phase = ReconcilePhase::CreatingNamespace;
                            metadata.namespace = None;
                            continue;
                        }
                        Err(e) => return Err(e.into()),
                    }

                    // Mark this deployment as active in the database
                    // This automatically unmarks other deployments in same (project, group)
                    crate::db::deployments::mark_as_active(
                        &self.state.db_pool,
                        deployment.id,
                        deployment.project_id,
                        &deployment.deployment_group,
                    )
                    .await
                    .map_err(|e| {
                        tracing::error!(
                            deployment_id = %deployment.deployment_id,
                            error = %e,
                            "Failed to mark deployment as active in database"
                        );
                        e
                    })?;

                    info!(
                        deployment_id = %deployment.deployment_id,
                        "Marked deployment as active in database"
                    );

                    metadata.reconcile_phase = ReconcilePhase::Completed;

                    return Ok(ReconcileResult {
                        status: DeploymentStatus::Healthy,
                        controller_metadata: serde_json::to_value(&metadata)?,
                        error_message: None,
                    });
                }

                ReconcilePhase::Completed => {
                    // Load and decrypt environment variables for drift detection
                    let env_vars = self.load_env_vars(project, deployment).await?;

                    // Apply all supporting resources with drift detection
                    // This reconciles: Namespace, ImagePullSecret, BackendService, Service, Ingress
                    // Uses in-memory version cache to skip unchanged resources
                    match self
                        .apply_supporting_resources(deployment, project, &mut metadata)
                        .await
                    {
                        Ok(_) => {}
                        Err(e) if is_namespace_not_found_error(&e) => {
                            warn!("Namespace missing during Completed phase, resetting to CreatingNamespace");
                            metadata.reconcile_phase = ReconcilePhase::CreatingNamespace;
                            metadata.namespace = None;
                            continue;
                        }
                        Err(e) => return Err(e),
                    }

                    // Get namespace after apply_supporting_resources (which may update metadata)
                    let namespace = metadata
                        .namespace
                        .as_ref()
                        .ok_or_else(|| anyhow::anyhow!("No namespace in metadata"))?;

                    // Check Deployment for drift
                    let deploy_name = metadata
                        .deployment_name
                        .as_ref()
                        .ok_or_else(|| anyhow::anyhow!("No Deployment name in metadata"))?;
                    let deploy_api: Api<K8sDeployment> =
                        Api::namespaced(self.kube_client.clone(), namespace);

                    match deploy_api.get(deploy_name).await {
                        Ok(existing_deploy) => {
                            let desired_deploy = self
                                .create_deployment(project, deployment, &metadata, env_vars.clone())
                                .await;

                            if self.deployment_has_drifted(&existing_deploy, &desired_deploy) {
                                info!(
                                    project = project.name,
                                    deployment_id = %deployment.id,
                                    "Deployment has drifted in Completed phase, recreating"
                                );

                                // Delete and wait for deletion to complete
                                if let Err(e) = self
                                    .delete_and_wait_deployment(&deploy_api, deploy_name)
                                    .await
                                {
                                    if is_namespace_not_found_error(&e) {
                                        warn!("Namespace missing during Deployment deletion in Completed phase, resetting to CreatingNamespace");
                                        metadata.reconcile_phase =
                                            ReconcilePhase::CreatingNamespace;
                                        metadata.namespace = None;
                                        continue;
                                    } else {
                                        return Err(e);
                                    }
                                }

                                // Create new Deployment
                                match deploy_api
                                    .create(&PostParams::default(), &desired_deploy)
                                    .await
                                {
                                    Ok(_) => {
                                        info!(
                                            project = project.name,
                                            deployment_id = %deployment.id,
                                            "Deployment recreated after drift detected in Completed phase"
                                        );

                                        // Move back to WaitingForDeployment to ensure it becomes ready
                                        metadata.reconcile_phase =
                                            ReconcilePhase::WaitingForDeployment;
                                        return Ok(ReconcileResult {
                                            status,
                                            controller_metadata: serde_json::to_value(&metadata)?,
                                            error_message: None,
                                        });
                                    }
                                    Err(e) if is_namespace_not_found_error(&e) => {
                                        warn!("Namespace missing during Deployment creation in Completed phase, resetting to CreatingNamespace");
                                        metadata.reconcile_phase =
                                            ReconcilePhase::CreatingNamespace;
                                        metadata.namespace = None;
                                        continue;
                                    }
                                    Err(e) => return Err(e.into()),
                                }
                            }
                        }
                        Err(kube::Error::Api(ae)) if ae.code == 404 => {
                            // Deployment is missing - recreate it
                            warn!(
                                project = project.name,
                                deployment_id = %deployment.id,
                                "Deployment missing in Completed phase, recreating"
                            );

                            metadata.reconcile_phase = ReconcilePhase::CreatingDeployment;
                            let metadata_json = serde_json::to_value(&metadata)?;
                            db_deployments::update_controller_metadata(
                                &self.state.db_pool,
                                deployment.id,
                                &metadata_json,
                            )
                            .await?;

                            return Ok(ReconcileResult {
                                status,
                                controller_metadata: serde_json::to_value(&metadata)?,
                                error_message: None,
                            });
                        }
                        Err(e) if is_namespace_not_found_error(&e) => {
                            warn!(
                                "Namespace missing during Deployment get in Completed phase, resetting to CreatingNamespace"
                            );
                            metadata.reconcile_phase = ReconcilePhase::CreatingNamespace;
                            metadata.namespace = None;
                            continue;
                        }
                        Err(e) => return Err(e.into()),
                    }

                    return Ok(ReconcileResult {
                        status: DeploymentStatus::Healthy,
                        controller_metadata: serde_json::to_value(&metadata)?,
                        error_message: None,
                    });
                }
            }
        }
    }

    async fn health_check(&self, deployment: &Deployment) -> Result<HealthStatus> {
        let metadata: KubernetesMetadata =
            serde_json::from_value(deployment.controller_metadata.clone())?;

        let deploy_name = metadata
            .deployment_name
            .ok_or_else(|| anyhow::anyhow!("No Deployment name"))?;
        let namespace = metadata
            .namespace
            .ok_or_else(|| anyhow::anyhow!("No namespace"))?;

        // Collect current pod status
        let pod_status = match self
            .collect_pod_status(&namespace, &deploy_name, &deployment.deployment_id)
            .await
        {
            Ok(ps) => Some(ps),
            Err(e) if is_namespace_not_found_error(&e) => {
                warn!("Namespace missing during pod status collection in health check");
                None
            }
            Err(e) => {
                warn!("Failed to collect pod status during health check: {:?}", e);
                None
            }
        };

        // 1. Check for pod-level errors FIRST
        // This prevents race conditions where Deployment reports ready_replicas
        // but pods are actually in CrashLoopBackOff or other error states
        let (has_errors, error_msg) = match self
            .check_pod_errors(&namespace, &deploy_name, &deployment.deployment_id)
            .await
        {
            Ok((errors, msg)) => (errors, msg),
            Err(e) if is_namespace_not_found_error(&e) => {
                // Namespace missing - return unhealthy status
                warn!("Namespace missing during health check, marking deployment as unhealthy");
                return Ok(HealthStatus {
                    healthy: false,
                    message: Some("Namespace missing - recovery in progress".to_string()),
                    last_check: Utc::now(),
                    pod_status: pod_status.and_then(|ps| serde_json::to_value(ps).ok()),
                });
            }
            Err(e) => return Err(e),
        };
        if has_errors {
            return Ok(HealthStatus {
                healthy: false,
                message: error_msg,
                last_check: Utc::now(),
                pod_status: pod_status.and_then(|ps| serde_json::to_value(ps).ok()),
            });
        }

        // 2. Then check Deployment readiness
        let deploy_api: Api<K8sDeployment> = Api::namespaced(self.kube_client.clone(), &namespace);

        // Get Deployment, handling 404 errors gracefully
        match deploy_api.get(&deploy_name).await {
            Ok(rs) => {
                // Check Deployment status
                let spec_replicas = rs.spec.and_then(|s| s.replicas).unwrap_or(1);
                let ready_replicas = rs.status.and_then(|s| s.ready_replicas).unwrap_or(0);

                let healthy = ready_replicas >= spec_replicas;

                Ok(HealthStatus {
                    healthy,
                    message: if !healthy {
                        Some(format!(
                            "Deployment ready: {}/{}",
                            ready_replicas, spec_replicas
                        ))
                    } else {
                        None
                    },
                    last_check: Utc::now(),
                    pod_status: pod_status.and_then(|ps| serde_json::to_value(ps).ok()),
                })
            }
            Err(kube::Error::Api(ae)) if ae.code == 404 => {
                // Deployment doesn't exist - mark as unhealthy to trigger recreation
                warn!(
                    "Deployment {} not found in namespace {}, marking deployment as unhealthy",
                    deploy_name, namespace
                );
                Ok(HealthStatus {
                    healthy: false,
                    message: Some(format!("Deployment {} not found", deploy_name)),
                    last_check: Utc::now(),
                    pod_status: pod_status.and_then(|ps| serde_json::to_value(ps).ok()),
                })
            }
            Err(e) if is_namespace_not_found_error(&e) => {
                // Namespace missing - mark as unhealthy
                warn!("Namespace missing during Deployment health check");
                Ok(HealthStatus {
                    healthy: false,
                    message: Some("Namespace missing - recovery in progress".to_string()),
                    last_check: Utc::now(),
                    pod_status: pod_status.and_then(|ps| serde_json::to_value(ps).ok()),
                })
            }
            Err(e) => Err(e.into()),
        }
    }

    async fn cancel(&self, deployment: &Deployment) -> Result<()> {
        info!(
            "Cancelling deployment {} (pre-infrastructure)",
            deployment.deployment_id
        );

        // Parse metadata, but don't fail if it's missing or incomplete (deployment may not have started)
        let metadata: Option<KubernetesMetadata> =
            serde_json::from_value(deployment.controller_metadata.clone()).ok();

        if let Some(metadata) = metadata {
            // Clean up any partially created Deployment
            if let (Some(deploy_name), Some(namespace)) =
                (metadata.deployment_name, metadata.namespace)
            {
                let deploy_api: Api<K8sDeployment> =
                    Api::namespaced(self.kube_client.clone(), &namespace);
                if let Err(e) = deploy_api
                    .delete(&deploy_name, &DeleteParams::default())
                    .await
                {
                    // Ignore 404 errors (already deleted)
                    if !e.to_string().contains("404") {
                        warn!("Error deleting Deployment during cancellation: {}", e);
                    }
                }
            }
        } else {
            debug!(
                "No metadata found for deployment {}, nothing to clean up",
                deployment.deployment_id
            );
        }

        // Service and namespace are shared, don't delete
        Ok(())
    }

    async fn terminate(&self, deployment: &Deployment) -> Result<()> {
        info!(
            "Terminating deployment {} - deleting Deployment",
            deployment.deployment_id
        );

        // Clear cached resource versions for this deployment
        self.clear_deployment_cache(deployment.id);

        // Parse metadata, but don't fail if it's missing or incomplete
        let metadata: Option<KubernetesMetadata> =
            serde_json::from_value(deployment.controller_metadata.clone()).ok();

        if let Some(metadata) = metadata {
            // Delete ONLY the Deployment (cascading deletes pods)
            if let (Some(deploy_name), Some(namespace)) =
                (metadata.deployment_name.clone(), metadata.namespace.clone())
            {
                let deploy_api: Api<K8sDeployment> =
                    Api::namespaced(self.kube_client.clone(), &namespace);
                if let Err(e) = deploy_api
                    .delete(&deploy_name, &DeleteParams::default())
                    .await
                {
                    // Ignore 404 errors (already deleted)
                    if !e.to_string().contains("404") {
                        warn!("Error deleting Deployment during termination: {}", e);
                    }
                }
            }

            // For non-default deployment groups, check if we should clean up group-specific resources
            if deployment.deployment_group
                != crate::server::deployment::models::DEFAULT_DEPLOYMENT_GROUP
            {
                // Check if there are any other active deployments in this group
                if let Err(e) = self
                    .cleanup_group_resources_if_empty(deployment, &metadata)
                    .await
                {
                    warn!(
                        "Error cleaning up group resources for deployment {}: {}",
                        deployment.deployment_id, e
                    );
                }
            }
        } else {
            debug!(
                "No metadata found for deployment {}, nothing to terminate",
                deployment.deployment_id
            );
        }

        // DO NOT delete Secret or Namespace (shared across all groups in the project)
        Ok(())
    }

    async fn get_deployment_urls(
        &self,
        deployment: &Deployment,
        project: &Project,
    ) -> Result<DeploymentUrls> {
        // Calculate default URL from templates
        let default_url_host = self.full_ingress_url(project, deployment);
        let default_url = format!("{}://{}", self.ingress_schema, default_url_host);

        // Custom domains only apply to deployments in the default group
        let (custom_domain_urls, primary_url) =
            if deployment.deployment_group == DEFAULT_DEPLOYMENT_GROUP {
                // Fetch custom domains from database
                let custom_domains = crate::db::custom_domains::list_project_custom_domains(
                    &self.state.db_pool,
                    project.id,
                )
                .await?;

                // Filter out custom domains that conflict with project default patterns
                let custom_domains = self.filter_valid_custom_domains(custom_domains);

                // Find the starred (primary) custom domain
                let starred = custom_domains.iter().find(|d| d.is_primary);
                let primary = if let Some(starred) = starred {
                    let host = if let Some(port) = self.ingress_port {
                        format!("{}:{}", starred.domain, port)
                    } else {
                        starred.domain.clone()
                    };
                    format!("{}://{}", self.ingress_schema, host)
                } else {
                    default_url.clone()
                };

                // Build custom domain URLs with schema
                let mut urls = Vec::new();
                for domain in custom_domains {
                    let url_host = if let Some(port) = self.ingress_port {
                        format!("{}:{}", domain.domain, port)
                    } else {
                        domain.domain
                    };
                    let url = format!("{}://{}", self.ingress_schema, url_host);
                    urls.push(url);
                }
                (urls, primary)
            } else {
                // Non-default groups don't get custom domain URLs
                (Vec::new(), default_url.clone())
            };

        Ok(DeploymentUrls {
            default_url,
            primary_url,
            custom_domain_urls,
        })
    }

    async fn get_project_urls(
        &self,
        project: &Project,
        deployment_group: &str,
    ) -> Result<DeploymentUrls> {
        // Build a URL using ingress templates and the deployment group
        let url_host = self.resolved_ingress_url_for_group(project, deployment_group);
        let full_host = self.full_ingress_url_from_host(&url_host);
        let default_url = format!("{}://{}", self.ingress_schema, full_host);

        // Custom domains only apply to the default group
        let (custom_domain_urls, primary_url) = if deployment_group == DEFAULT_DEPLOYMENT_GROUP {
            let custom_domains = crate::db::custom_domains::list_project_custom_domains(
                &self.state.db_pool,
                project.id,
            )
            .await?;

            let custom_domains = self.filter_valid_custom_domains(custom_domains);

            let starred = custom_domains.iter().find(|d| d.is_primary);
            let primary = if let Some(starred) = starred {
                let host = if let Some(port) = self.ingress_port {
                    format!("{}:{}", starred.domain, port)
                } else {
                    starred.domain.clone()
                };
                format!("{}://{}", self.ingress_schema, host)
            } else {
                default_url.clone()
            };

            let mut urls = Vec::new();
            for domain in custom_domains {
                let url_host = if let Some(port) = self.ingress_port {
                    format!("{}:{}", domain.domain, port)
                } else {
                    domain.domain
                };
                urls.push(format!("{}://{}", self.ingress_schema, url_host));
            }
            (urls, primary)
        } else {
            (Vec::new(), default_url.clone())
        };

        Ok(DeploymentUrls {
            default_url,
            primary_url,
            custom_domain_urls,
        })
    }

    async fn stream_logs(
        &self,
        deployment: &Deployment,
        follow: bool,
        tail_lines: Option<i64>,
        timestamps: bool,
        since_seconds: Option<i64>,
    ) -> Result<futures::stream::BoxStream<'static, Result<bytes::Bytes, anyhow::Error>>> {
        use futures::StreamExt;
        use k8s_openapi::api::core::v1::Pod;
        use kube::api::{Api, ListParams, LogParams};

        // Extract namespace from controller_metadata
        let namespace = deployment
            .controller_metadata
            .get("namespace")
            .and_then(|v| v.as_str())
            .ok_or_else(|| anyhow::anyhow!("Namespace not found in controller metadata"))?
            .to_string();

        // Find pod using label selector
        let pod_api: Api<Pod> = Api::namespaced(self.kube_client.clone(), &namespace);
        let pods = pod_api
            .list(&ListParams::default().labels(&format!(
                "rise.dev/deployment-id={}",
                deployment.deployment_id
            )))
            .await?;

        let pod = pods
            .items
            .first()
            .ok_or_else(|| anyhow::anyhow!("Pod not found - deployment may not be ready yet"))?;

        let pod_name = pod
            .metadata
            .name
            .as_ref()
            .ok_or_else(|| anyhow::anyhow!("Pod name not found"))?
            .clone();

        // Build LogParams
        let mut log_params = LogParams {
            follow,
            timestamps,
            ..Default::default()
        };
        if let Some(tail) = tail_lines {
            log_params.tail_lines = Some(tail);
        }
        if let Some(since) = since_seconds {
            log_params.since_seconds = Some(since);
        }

        // Stream logs from pod (will stream from first container)
        let mut log_stream = pod_api.log_stream(&pod_name, &log_params).await?;

        // Convert AsyncBufRead to Stream of Bytes by reading chunks
        use futures::AsyncReadExt;
        let stream = async_stream::stream! {
            let mut buffer = vec![0u8; 8192];
            loop {
                match log_stream.read(&mut buffer).await {
                    Ok(0) => break, // EOF
                    Ok(n) => {
                        yield Ok(bytes::Bytes::copy_from_slice(&buffer[..n]));
                    }
                    Err(e) => {
                        yield Err(anyhow::anyhow!("Log stream error: {}", e));
                        break;
                    }
                }
            }
        };

        Ok(stream.boxed())
    }
}

impl KubernetesController {
    /// Compare actual vs desired Deployment state to detect drift
    fn deployment_has_drifted(&self, actual: &K8sDeployment, desired: &K8sDeployment) -> bool {
        // Compare critical fields that should never drift

        // 1. Replica count
        let actual_replicas = actual.spec.as_ref().and_then(|s| s.replicas).unwrap_or(0);
        let desired_replicas = desired.spec.as_ref().and_then(|s| s.replicas).unwrap_or(0);
        if actual_replicas != desired_replicas {
            warn!(
                "Deployment drift: replicas {} != {}",
                actual_replicas, desired_replicas
            );
            return true;
        }

        // 2. Container image
        let actual_image = actual
            .spec
            .as_ref()
            .and_then(|s| s.template.spec.as_ref())
            .and_then(|ps| ps.containers.first())
            .and_then(|c| c.image.as_ref());

        let desired_image = desired
            .spec
            .as_ref()
            .and_then(|s| s.template.spec.as_ref())
            .and_then(|ps| ps.containers.first())
            .and_then(|c| c.image.as_ref());

        if actual_image != desired_image {
            warn!(
                "Deployment drift: image {:?} != {:?}",
                actual_image, desired_image
            );
            return true;
        }

        // 3. Labels (deployment-id must match)
        let actual_labels = actual
            .spec
            .as_ref()
            .and_then(|s| s.selector.match_labels.as_ref());
        let desired_labels = desired
            .spec
            .as_ref()
            .and_then(|s| s.selector.match_labels.as_ref());

        if actual_labels != desired_labels {
            warn!("Deployment drift: labels don't match");
            return true;
        }

        // Note: We do NOT check environment variables for drift.
        // Environment variables are snapshots at deployment time and should not
        // trigger reconciliation when project-level env vars change.

        false
    }

    /// Delete Deployment and wait for deletion to complete
    async fn delete_and_wait_deployment(
        &self,
        deploy_api: &Api<K8sDeployment>,
        name: &str,
    ) -> Result<()> {
        // Delete the Deployment
        let delete_params = DeleteParams::default();
        match deploy_api.delete(name, &delete_params).await {
            Ok(_) => {}
            Err(kube::Error::Api(ae)) if ae.code == 404 => {
                // Already deleted
                return Ok(());
            }
            Err(e) => return Err(e.into()),
        }

        // Wait for deletion to complete (max 30 seconds)
        for i in 0..30 {
            tokio::time::sleep(Duration::from_secs(1)).await;

            match deploy_api.get(name).await {
                Err(kube::Error::Api(ae)) if ae.code == 404 => {
                    debug!("Deployment {} deleted successfully", name);
                    return Ok(());
                }
                Err(e) => return Err(e.into()),
                Ok(_) => {
                    debug!(
                        "Waiting for Deployment {} deletion (attempt {})",
                        name,
                        i + 1
                    );
                }
            }
        }

        Err(anyhow::anyhow!(
            "Timeout waiting for Deployment {} deletion",
            name
        ))
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use k8s_openapi::api::core::v1::{Container, PodSpec, PodTemplateSpec};
    use std::collections::BTreeMap;

    // Test database URL for mock controllers
    const TEST_DATABASE_URL: &str = "postgres://localhost/test";

    /// Mock registry provider for unit tests
    struct MockRegistryProvider;

    #[async_trait::async_trait]
    impl crate::server::registry::RegistryProvider for MockRegistryProvider {
        async fn get_credentials(
            &self,
            _repository: &str,
        ) -> anyhow::Result<crate::server::registry::models::RegistryCredentials> {
            Ok(crate::server::registry::models::RegistryCredentials {
                username: "test".to_string(),
                password: "test".to_string(),
                registry_url: "localhost:5000".to_string(),
                expires_in: None,
                auth_method: Default::default(),
            })
        }

        async fn get_pull_credentials(&self) -> anyhow::Result<(String, String)> {
            Ok(("test".to_string(), "test".to_string()))
        }

        fn registry_host(&self) -> &str {
            "localhost:5000"
        }

        fn registry_url(&self) -> &str {
            "localhost:5000"
        }

        fn get_image_tag(
            &self,
            repository: &str,
            tag: &str,
            _tag_type: crate::server::registry::ImageTagType,
        ) -> String {
            format!("localhost:5000/{repository}:{tag}")
        }
    }

    fn create_test_deployment(
        name: &str,
        replicas: i32,
        image: &str,
        labels: BTreeMap<String, String>,
    ) -> K8sDeployment {
        K8sDeployment {
            metadata: ObjectMeta {
                name: Some(name.to_string()),
                ..Default::default()
            },
            spec: Some(DeploymentSpec {
                replicas: Some(replicas),
                selector: LabelSelector {
                    match_labels: Some(labels.clone()),
                    ..Default::default()
                },
                template: PodTemplateSpec {
                    metadata: Some(ObjectMeta {
                        labels: Some(labels),
                        ..Default::default()
                    }),
                    spec: Some(PodSpec {
                        containers: vec![Container {
                            name: "app".to_string(),
                            image: Some(image.to_string()),
                            ..Default::default()
                        }],
                        ..Default::default()
                    }),
                },
                ..Default::default()
            }),
            ..Default::default()
        }
    }

    #[tokio::test]
    async fn test_deployment_has_drifted_no_drift() {
        // Create a mock controller (we only need the method, not actual K8s connection)
        let controller = create_mock_controller();

        let mut labels = BTreeMap::new();
        labels.insert("app".to_string(), "test".to_string());

        let d1 = create_test_deployment("test-deploy", 1, "nginx:1.0", labels.clone());
        let d2 = create_test_deployment("test-deploy", 1, "nginx:1.0", labels);

        assert!(!controller.deployment_has_drifted(&d1, &d2));
    }

    #[tokio::test]
    async fn test_deployment_has_drifted_replica_count() {
        let controller = create_mock_controller();

        let mut labels = BTreeMap::new();
        labels.insert("app".to_string(), "test".to_string());

        let d1 = create_test_deployment("test-deploy", 1, "nginx:1.0", labels.clone());
        let d2 = create_test_deployment("test-deploy", 3, "nginx:1.0", labels);

        assert!(controller.deployment_has_drifted(&d1, &d2));
    }

    #[tokio::test]
    async fn test_deployment_has_drifted_image() {
        let controller = create_mock_controller();

        let mut labels = BTreeMap::new();
        labels.insert("app".to_string(), "test".to_string());

        let d1 = create_test_deployment("test-deploy", 1, "nginx:1.0", labels.clone());
        let d2 = create_test_deployment("test-deploy", 1, "nginx:2.0", labels);

        assert!(controller.deployment_has_drifted(&d1, &d2));
    }

    #[tokio::test]
    async fn test_deployment_has_drifted_labels() {
        let controller = create_mock_controller();

        let mut labels1 = BTreeMap::new();
        labels1.insert("app".to_string(), "test".to_string());

        let mut labels2 = BTreeMap::new();
        labels2.insert("app".to_string(), "other".to_string());

        let d1 = create_test_deployment("test-deploy", 1, "nginx:1.0", labels1);
        let d2 = create_test_deployment("test-deploy", 1, "nginx:1.0", labels2);

        assert!(controller.deployment_has_drifted(&d1, &d2));
    }

    fn pod_status_with_pod(pod: PodInfo) -> PodStatus {
        PodStatus {
            desired_replicas: 1,
            ready_replicas: 0,
            current_replicas: 1,
            pods: vec![pod],
            last_checked: Utc::now(),
        }
    }

    fn empty_pod() -> PodInfo {
        PodInfo {
            name: "pod-1".to_string(),
            phase: PodPhase::Pending,
            conditions: vec![],
            containers: vec![],
            events: vec![],
        }
    }

    #[test]
    fn test_extract_pod_error_waiting_irrecoverable_reason() {
        let mut pod = empty_pod();
        pod.containers.push(ContainerStatusInfo {
            name: "app".to_string(),
            ready: false,
            restart_count: 0,
            state: Some(ContainerState {
                state_type: "waiting".to_string(),
                reason: Some("ImagePullBackOff".to_string()),
                message: Some("Back-off pulling image \"bad/image:latest\"".to_string()),
                exit_code: None,
            }),
        });
        let status = pod_status_with_pod(pod);

        let error = KubernetesController::extract_pod_error(&status);
        assert_eq!(
            error,
            Some("ImagePullBackOff: Back-off pulling image \"bad/image:latest\"".to_string())
        );
    }

    #[test]
    fn test_extract_pod_error_terminated_with_restart_threshold() {
        let mut pod = empty_pod();
        pod.containers.push(ContainerStatusInfo {
            name: "app".to_string(),
            ready: false,
            restart_count: 3,
            state: Some(ContainerState {
                state_type: "terminated".to_string(),
                reason: Some("Error".to_string()),
                message: Some("process exited".to_string()),
                exit_code: Some(137),
            }),
        });
        let status = pod_status_with_pod(pod);

        let error = KubernetesController::extract_pod_error(&status);
        assert_eq!(
            error,
            Some("Error: process exited (restarts: 3)".to_string())
        );
    }

    #[test]
    fn test_extract_pod_error_repeated_warning_event() {
        let mut pod = empty_pod();
        pod.events.push(PodEvent {
            type_: "Warning".to_string(),
            reason: "FailedScheduling".to_string(),
            message: "0/3 nodes are available".to_string(),
            count: MIN_EVENT_COUNT_FOR_CRITICAL,
            last_timestamp: Utc::now(),
        });
        let status = pod_status_with_pod(pod);

        let error = KubernetesController::extract_pod_error(&status);
        assert_eq!(
            error,
            Some("FailedScheduling: 0/3 nodes are available".to_string())
        );
    }

    #[test]
    fn test_extract_pod_error_ignores_transient_warning_event() {
        let mut pod = empty_pod();
        pod.events.push(PodEvent {
            type_: "Warning".to_string(),
            reason: "FailedScheduling".to_string(),
            message: "0/3 nodes are available".to_string(),
            count: 1,
            last_timestamp: Utc::now(),
        });
        let status = pod_status_with_pod(pod);

        let error = KubernetesController::extract_pod_error(&status);
        assert_eq!(error, None);
    }

    /// Construct a test NetworkPolicyConfig with standard ingress rules and unrestricted egress.
    fn test_network_policy() -> crate::server::settings::NetworkPolicyConfig {
        use k8s_openapi::api::networking::v1::{NetworkPolicyIngressRule, NetworkPolicyPeer};
        use k8s_openapi::apimachinery::pkg::apis::meta::v1::LabelSelector;

        crate::server::settings::NetworkPolicyConfig {
            ingress: vec![
                NetworkPolicyIngressRule {
                    from: Some(vec![NetworkPolicyPeer {
                        namespace_selector: Some(LabelSelector {
                            match_labels: Some(
                                [(
                                    "kubernetes.io/metadata.name".to_string(),
                                    "ingress-nginx".to_string(),
                                )]
                                .into_iter()
                                .collect(),
                            ),
                            ..Default::default()
                        }),
                        pod_selector: Some(LabelSelector {
                            match_labels: Some(
                                [(
                                    "app.kubernetes.io/name".to_string(),
                                    "ingress-nginx".to_string(),
                                )]
                                .into_iter()
                                .collect(),
                            ),
                            ..Default::default()
                        }),
                        ..Default::default()
                    }]),
                    ports: None,
                },
                NetworkPolicyIngressRule {
                    from: Some(vec![NetworkPolicyPeer {
                        pod_selector: Some(LabelSelector {
                            match_labels: Some(Default::default()),
                            ..Default::default()
                        }),
                        ..Default::default()
                    }]),
                    ports: None,
                },
            ],
            egress: None,
        }
    }

    // Helper function to create a mock controller for testing
    fn create_mock_controller() -> KubernetesController {
        use crate::server::state::ControllerState;
        use axum::http::Uri;
        use sqlx::postgres::PgPoolOptions;

        // Install default CryptoProvider for rustls (required for kube-rs HTTPS connections)
        rustls::crypto::ring::default_provider()
            .install_default()
            .ok();

        // Create a minimal controller with mock values
        // Note: We won't actually connect to K8s or DB for these unit tests
        let pool = PgPoolOptions::new()
            .max_connections(1)
            .connect_lazy(TEST_DATABASE_URL)
            .expect("Failed to create pool");

        let state = ControllerState {
            db_pool: pool,
            encryption_provider: None,
        };

        // Create a fake kube client (won't be used in these tests)
        let cluster_url = "http://localhost:8080"
            .parse::<Uri>()
            .expect("Failed to parse URI");
        let kube_config = kube::Config::new(cluster_url);
        let kube_client = kube::Client::try_from(kube_config).expect("Failed to create client");

        // Create default access classes for testing
        let mut access_classes = std::collections::HashMap::new();
        access_classes.insert(
            "public".to_string(),
            crate::server::settings::AccessClass {
                display_name: "Public".to_string(),
                description: "Publicly accessible".to_string(),
                ingress_class: "nginx".to_string(),
                access_requirement: crate::server::settings::AccessRequirement::None,
                custom_annotations: std::collections::HashMap::new(),
            },
        );

        KubernetesController {
            state,
            kube_client,
            production_ingress_url_template: "{project_name}.test.local".to_string(),
            staging_ingress_url_template: None,
            ingress_port: None,
            ingress_schema: "https".to_string(),
            registry_provider: Arc::new(MockRegistryProvider),
            auth_backend_url: "http://localhost:3000".to_string(),
            auth_signin_url: "http://localhost:3000".to_string(),
            backend_address: None,
            namespace_labels: std::collections::HashMap::new(),
            namespace_annotations: std::collections::HashMap::new(),
            ingress_annotations: std::collections::HashMap::new(),
            ingress_tls_secret_name: None,
            custom_domain_tls_mode: crate::server::settings::CustomDomainTlsMode::PerDomain,
            custom_domain_ingress_annotations: std::collections::HashMap::new(),
            node_selector: std::collections::HashMap::new(),
            image_pull_secret_name: None,
            access_classes,
            resource_versions: Arc::new(std::sync::RwLock::new(std::collections::HashMap::new())),
            host_aliases: std::collections::HashMap::new(),
            extra_service_token_audiences: std::collections::HashMap::new(),
            network_policy: test_network_policy(),
            pod_security_enabled: true,
            pod_resources: None,
            health_probes: None,
        }
    }

    fn sample_project() -> crate::db::models::Project {
        use crate::db::models::ProjectStatus;
        use uuid::Uuid;

        crate::db::models::Project {
            id: Uuid::new_v4(),
            name: "test-project".to_string(),
            status: ProjectStatus::Running,
            access_class: "public".to_string(),
            owner_user_id: None,
            owner_team_id: None,
            finalizers: vec![],
            created_at: chrono::Utc::now(),
            updated_at: chrono::Utc::now(),
        }
    }

    fn sample_deployment(project_id: uuid::Uuid) -> crate::db::models::Deployment {
        crate::db::models::Deployment {
            id: uuid::Uuid::new_v4(),
            project_id,
            deployment_id: "20240108-103000".to_string(),
            created_by_id: uuid::Uuid::new_v4(),
            deployment_group: "default".to_string(),
            status: crate::db::models::DeploymentStatus::Deploying,
            expires_at: None,
            termination_reason: None,
            completed_at: None,
            error_message: None,
            build_logs: None,
            controller_metadata: serde_json::json!({}),
            image: None,
            image_digest: None,
            rolled_back_from_deployment_id: None,
            http_port: 8080,
            needs_reconcile: false,
            is_active: false,
            deploying_started_at: None,
            first_healthy_at: None,
            created_at: chrono::Utc::now(),
            updated_at: chrono::Utc::now(),
        }
    }

    fn sample_metadata(namespace: &str, deployment_name: &str) -> KubernetesMetadata {
        KubernetesMetadata {
            namespace: Some(namespace.to_string()),
            deployment_name: Some(deployment_name.to_string()),
            service_name: Some("default".to_string()),
            ingress_name: Some("default".to_string()),
            image_tag: None,
            http_port: 8080,
            reconcile_phase: ReconcilePhase::Completed,
            previous_deployment: None,
            pod_status: None,
        }
    }

    #[tokio::test]
    async fn test_create_deployment_omits_extra_service_token_projection_when_not_configured() {
        let controller = create_mock_controller();
        let project = sample_project();
        let deployment = sample_deployment(project.id);
        let metadata = sample_metadata("rise-test-project", "test-project-20240108-103000");

        let generated = controller
            .create_deployment(&project, &deployment, &metadata, vec![])
            .await;

        let pod_spec = generated
            .spec
            .and_then(|spec| spec.template.spec)
            .expect("pod spec should be present");

        assert!(pod_spec.volumes.is_none());
        assert!(pod_spec.containers[0].volume_mounts.is_none());
    }

    #[tokio::test]
    async fn test_create_deployment_adds_extra_service_token_projection() {
        let mut controller = create_mock_controller();
        controller.extra_service_token_audiences = std::collections::HashMap::from([
            ("vault".to_string(), "https://vault.example.com".to_string()),
            ("metrics".to_string(), "metrics-service".to_string()),
        ]);

        let project = sample_project();
        let deployment = sample_deployment(project.id);
        let metadata = sample_metadata("rise-test-project", "test-project-20240108-103000");

        let generated = controller
            .create_deployment(&project, &deployment, &metadata, vec![])
            .await;

        let pod_spec = generated
            .spec
            .and_then(|spec| spec.template.spec)
            .expect("pod spec should be present");

        let volumes = pod_spec.volumes.expect("volumes should be present");
        assert_eq!(volumes.len(), 1);
        assert_eq!(volumes[0].name, EXTRA_SERVICE_TOKENS_VOLUME_NAME);

        let projected = volumes[0]
            .projected
            .as_ref()
            .expect("projected volume should be present");
        let sources = projected
            .sources
            .as_ref()
            .expect("projected volume sources should be present");
        assert_eq!(sources.len(), 2);

        let mut actual_sources: Vec<(String, String)> = sources
            .iter()
            .map(|projection| {
                let token = projection
                    .service_account_token
                    .as_ref()
                    .expect("serviceAccountToken projection should be present");
                (
                    token.path.clone(),
                    token
                        .audience
                        .clone()
                        .expect("audience should be populated"),
                )
            })
            .collect();
        actual_sources.sort();

        assert_eq!(
            actual_sources,
            vec![
                ("metrics".to_string(), "metrics-service".to_string()),
                ("vault".to_string(), "https://vault.example.com".to_string()),
            ]
        );

        let volume_mounts = pod_spec.containers[0]
            .volume_mounts
            .as_ref()
            .expect("volume mounts should be present");
        assert_eq!(volume_mounts.len(), 1);
        assert_eq!(volume_mounts[0].name, EXTRA_SERVICE_TOKENS_VOLUME_NAME);
        assert_eq!(volume_mounts[0].mount_path, EXTRA_SERVICE_TOKENS_MOUNT_PATH);
        assert_eq!(volume_mounts[0].read_only, Some(true));
    }

    #[tokio::test]
    async fn test_resource_requirements_defaults() {
        let controller = create_mock_controller();
        let resources = controller
            .create_resource_requirements()
            .expect("should return resource requirements");

        let requests = resources.requests.expect("should have requests");
        assert_eq!(requests["cpu"].0, "500m");
        assert_eq!(requests["memory"].0, "256Mi");

        let limits = resources.limits.expect("should have limits");
        assert_eq!(limits["cpu"].0, "2");
        assert_eq!(limits["memory"].0, "2Gi");
    }

    #[tokio::test]
    async fn test_resource_requirements_custom_values() {
        let mut controller = create_mock_controller();
        controller.pod_resources = Some(crate::server::settings::PodResourceLimits {
            cpu_request: "100m".to_string(),
            memory_request: "128Mi".to_string(),
            cpu_limit: "4".to_string(),
            memory_limit: "1Gi".to_string(),
        });

        let resources = controller
            .create_resource_requirements()
            .expect("should return resource requirements");

        let requests = resources.requests.expect("should have requests");
        assert_eq!(requests["cpu"].0, "100m");
        assert_eq!(requests["memory"].0, "128Mi");

        let limits = resources.limits.expect("should have limits");
        assert_eq!(limits["cpu"].0, "4");
        assert_eq!(limits["memory"].0, "1Gi");
    }

    #[test]
    fn test_parse_ingress_url_subdomain() {
        let url = "myapp.apps.rise.dev";
        let parsed = KubernetesController::parse_ingress_url(url);
        assert_eq!(parsed.host, "myapp.apps.rise.dev");
        assert!(parsed.path_prefix.is_none());
    }

    #[test]
    fn test_parse_ingress_url_single_path() {
        let url = "rise.dev/myapp";
        let parsed = KubernetesController::parse_ingress_url(url);
        assert_eq!(parsed.host, "rise.dev");
        assert_eq!(parsed.path_prefix, Some("/myapp".to_string()));
    }

    #[test]
    fn test_parse_ingress_url_multi_level_path() {
        let url = "rise.dev/myapp/staging";
        let parsed = KubernetesController::parse_ingress_url(url);
        assert_eq!(parsed.host, "rise.dev");
        assert_eq!(parsed.path_prefix, Some("/myapp/staging".to_string()));
    }

    #[test]
    fn test_deployment_labels_includes_deployment_id() {
        use crate::db::models::{Deployment, Project, ProjectStatus};
        use uuid::Uuid;

        let deployment_uuid = Uuid::new_v4();
        let project = Project {
            id: Uuid::new_v4(),
            name: "test-project".to_string(),
            status: ProjectStatus::Running,
            access_class: "public".to_string(),
            owner_user_id: None,
            owner_team_id: None,
            finalizers: vec![],
            created_at: chrono::Utc::now(),
            updated_at: chrono::Utc::now(),
        };

        let deployment = Deployment {
            id: deployment_uuid,
            project_id: project.id,
            deployment_id: "20240108-103000".to_string(),
            created_by_id: Uuid::new_v4(),
            deployment_group: "default".to_string(),
            status: crate::db::models::DeploymentStatus::Deploying,
            expires_at: None,
            termination_reason: None,
            completed_at: None,
            error_message: None,
            build_logs: None,
            controller_metadata: serde_json::json!({}),
            image: None,
            image_digest: None,
            rolled_back_from_deployment_id: None,
            http_port: 8080,
            needs_reconcile: false,
            is_active: false,
            deploying_started_at: None,
            first_healthy_at: None,
            created_at: chrono::Utc::now(),
            updated_at: chrono::Utc::now(),
        };

        let labels = KubernetesController::deployment_labels(&project, &deployment);

        // Verify the rise.dev/deployment-id label is present
        assert!(
            labels.contains_key(LABEL_DEPLOYMENT_ID),
            "Labels should contain rise.dev/deployment-id"
        );
        assert_eq!(
            labels.get(LABEL_DEPLOYMENT_ID).unwrap(),
            &deployment.deployment_id,
            "deployment-id label should match deployment.deployment_id"
        );

        // Verify the rise.dev/deployment-uuid label is present
        assert!(
            labels.contains_key(LABEL_DEPLOYMENT_UUID),
            "Labels should contain rise.dev/deployment-uuid"
        );
        assert_eq!(
            labels.get(LABEL_DEPLOYMENT_UUID).unwrap(),
            &deployment.id.to_string(),
            "deployment-uuid label should match deployment.id"
        );

        // Also verify other expected labels
        assert!(labels.contains_key(LABEL_MANAGED_BY));
        assert!(labels.contains_key(LABEL_PROJECT));
        assert!(labels.contains_key(LABEL_DEPLOYMENT_GROUP));
    }

    #[test]
    fn test_ingress_names() {
        use crate::db::models::{Deployment, Project, ProjectStatus};
        use uuid::Uuid;

        let project = Project {
            id: Uuid::new_v4(),
            name: "test-project".to_string(),
            status: ProjectStatus::Running,
            access_class: "public".to_string(),
            owner_user_id: None,
            owner_team_id: None,
            finalizers: vec![],
            created_at: chrono::Utc::now(),
            updated_at: chrono::Utc::now(),
        };

        let deployment = Deployment {
            id: Uuid::new_v4(),
            project_id: project.id,
            deployment_id: "20240108-103000".to_string(),
            created_by_id: Uuid::new_v4(),
            deployment_group: "default".to_string(),
            status: crate::db::models::DeploymentStatus::Deploying,
            expires_at: None,
            termination_reason: None,
            completed_at: None,
            error_message: None,
            build_logs: None,
            controller_metadata: serde_json::json!({}),
            image: None,
            image_digest: None,
            rolled_back_from_deployment_id: None,
            http_port: 8080,
            needs_reconcile: false,
            is_active: false,
            deploying_started_at: None,
            first_healthy_at: None,
            created_at: chrono::Utc::now(),
            updated_at: chrono::Utc::now(),
        };

        // Test primary ingress name
        let ingress_name = KubernetesController::ingress_name(&project, &deployment);
        assert_eq!(ingress_name, "default");

        // Test custom domain ingress name
        let custom_domain_ingress_name =
            KubernetesController::custom_domain_ingress_name(&project, &deployment);
        assert_eq!(custom_domain_ingress_name, "default-custom-domains");
    }

    #[tokio::test]
    async fn test_primary_ingress_with_path_prefix_has_annotation() {
        let controller = create_mock_controller_with_path_based_routing();

        use crate::db::models::{Deployment, Project, ProjectStatus};
        use uuid::Uuid;

        let project = Project {
            id: Uuid::new_v4(),
            name: "compass".to_string(),
            status: ProjectStatus::Running,
            access_class: "public".to_string(),
            owner_user_id: None,
            owner_team_id: None,
            finalizers: vec![],
            created_at: chrono::Utc::now(),
            updated_at: chrono::Utc::now(),
        };

        let deployment = Deployment {
            id: Uuid::new_v4(),
            project_id: project.id,
            deployment_id: "20240108-103000".to_string(),
            created_by_id: Uuid::new_v4(),
            deployment_group: "default".to_string(),
            status: crate::db::models::DeploymentStatus::Deploying,
            expires_at: None,
            termination_reason: None,
            completed_at: None,
            error_message: None,
            build_logs: None,
            controller_metadata: serde_json::json!({}),
            image: None,
            image_digest: None,
            rolled_back_from_deployment_id: None,
            http_port: 8080,
            needs_reconcile: false,
            is_active: false,
            deploying_started_at: None,
            first_healthy_at: None,
            created_at: chrono::Utc::now(),
            updated_at: chrono::Utc::now(),
        };

        let metadata = KubernetesMetadata {
            namespace: Some("rise-compass".to_string()),
            deployment_name: Some("compass-20240108-103000".to_string()),
            service_name: Some("default".to_string()),
            ingress_name: Some("default".to_string()),
            image_tag: None,
            http_port: 8080,
            reconcile_phase: ReconcilePhase::Completed,
            previous_deployment: None,
            pod_status: None,
        };

        let ingress = controller.create_primary_ingress(&project, &deployment, &metadata);

        // Verify the x-forwarded-prefix annotation is present
        let annotations = ingress.metadata.annotations.unwrap();
        assert!(
            annotations.contains_key("nginx.ingress.kubernetes.io/x-forwarded-prefix"),
            "Primary ingress should have x-forwarded-prefix annotation for path-based routing"
        );
        assert_eq!(
            annotations.get("nginx.ingress.kubernetes.io/x-forwarded-prefix"),
            Some(&"/compass".to_string()),
            "x-forwarded-prefix should be /compass"
        );

        // Verify the rewrite-target annotation is also present
        assert!(
            annotations.contains_key("nginx.ingress.kubernetes.io/rewrite-target"),
            "Primary ingress should have rewrite-target annotation for path-based routing"
        );

        // Verify there's only one rule (primary host)
        let rules = ingress.spec.unwrap().rules.unwrap();
        assert_eq!(rules.len(), 1, "Primary ingress should have exactly 1 rule");
        assert_eq!(
            rules[0].host,
            Some("rise.dev".to_string()),
            "Primary ingress should have the configured host"
        );
    }

    #[tokio::test]
    async fn test_custom_domain_ingress_without_annotation() {
        let controller = create_mock_controller_with_path_based_routing();

        use crate::db::models::{CustomDomain, Deployment, Project, ProjectStatus};
        use uuid::Uuid;

        let project = Project {
            id: Uuid::new_v4(),
            name: "compass".to_string(),
            status: ProjectStatus::Running,
            access_class: "public".to_string(),
            owner_user_id: None,
            owner_team_id: None,
            finalizers: vec![],
            created_at: chrono::Utc::now(),
            updated_at: chrono::Utc::now(),
        };

        let deployment = Deployment {
            id: Uuid::new_v4(),
            project_id: project.id,
            deployment_id: "20240108-103000".to_string(),
            created_by_id: Uuid::new_v4(),
            deployment_group: "default".to_string(),
            status: crate::db::models::DeploymentStatus::Deploying,
            expires_at: None,
            termination_reason: None,
            completed_at: None,
            error_message: None,
            build_logs: None,
            controller_metadata: serde_json::json!({}),
            image: None,
            image_digest: None,
            rolled_back_from_deployment_id: None,
            http_port: 8080,
            needs_reconcile: false,
            is_active: false,
            deploying_started_at: None,
            first_healthy_at: None,
            created_at: chrono::Utc::now(),
            updated_at: chrono::Utc::now(),
        };

        let metadata = KubernetesMetadata {
            namespace: Some("rise-compass".to_string()),
            deployment_name: Some("compass-20240108-103000".to_string()),
            service_name: Some("default".to_string()),
            ingress_name: Some("default".to_string()),
            image_tag: None,
            http_port: 8080,
            reconcile_phase: ReconcilePhase::Completed,
            previous_deployment: None,
            pod_status: None,
        };

        let custom_domains = vec![CustomDomain {
            id: Uuid::new_v4(),
            domain: "compass-dev.example.com".to_string(),
            project_id: project.id,
            is_primary: false,
            created_at: chrono::Utc::now(),
            updated_at: chrono::Utc::now(),
        }];

        let ingress = controller.create_custom_domain_ingress(
            &project,
            &deployment,
            &metadata,
            &custom_domains,
        );

        // Verify the x-forwarded-prefix annotation is NOT present
        if let Some(annotations) = &ingress.metadata.annotations {
            assert!(
                !annotations.contains_key("nginx.ingress.kubernetes.io/x-forwarded-prefix"),
                "Custom domain ingress should NOT have x-forwarded-prefix annotation"
            );
            // Also verify rewrite-target is not present
            assert!(
                !annotations.contains_key("nginx.ingress.kubernetes.io/rewrite-target"),
                "Custom domain ingress should NOT have rewrite-target annotation"
            );
        }

        // Verify there's one rule for the custom domain
        let rules = ingress.spec.unwrap().rules.unwrap();
        assert_eq!(
            rules.len(),
            1,
            "Custom domain ingress should have exactly 1 rule"
        );
        assert_eq!(
            rules[0].host,
            Some("compass-dev.example.com".to_string()),
            "Custom domain ingress should have the custom domain as host"
        );

        // Verify the path is "/" (not "/compass")
        let paths = rules[0].http.as_ref().unwrap().paths.clone();
        assert!(
            paths.iter().any(|p| p.path == Some("/".to_string())),
            "Custom domain ingress should use root path '/'"
        );
    }

    #[tokio::test]
    async fn test_subdomain_based_ingress_without_annotation() {
        let controller = create_mock_controller(); // Uses subdomain-based routing

        use crate::db::models::{Deployment, Project, ProjectStatus};
        use uuid::Uuid;

        let project = Project {
            id: Uuid::new_v4(),
            name: "myapp".to_string(),
            status: ProjectStatus::Running,
            access_class: "public".to_string(),
            owner_user_id: None,
            owner_team_id: None,
            finalizers: vec![],
            created_at: chrono::Utc::now(),
            updated_at: chrono::Utc::now(),
        };

        let deployment = Deployment {
            id: Uuid::new_v4(),
            project_id: project.id,
            deployment_id: "20240108-103000".to_string(),
            created_by_id: Uuid::new_v4(),
            deployment_group: "default".to_string(),
            status: crate::db::models::DeploymentStatus::Deploying,
            expires_at: None,
            termination_reason: None,
            completed_at: None,
            error_message: None,
            build_logs: None,
            controller_metadata: serde_json::json!({}),
            image: None,
            image_digest: None,
            rolled_back_from_deployment_id: None,
            http_port: 8080,
            needs_reconcile: false,
            is_active: false,
            deploying_started_at: None,
            first_healthy_at: None,
            created_at: chrono::Utc::now(),
            updated_at: chrono::Utc::now(),
        };

        let metadata = KubernetesMetadata {
            namespace: Some("rise-myapp".to_string()),
            deployment_name: Some("myapp-20240108-103000".to_string()),
            service_name: Some("default".to_string()),
            ingress_name: Some("default".to_string()),
            image_tag: None,
            http_port: 8080,
            reconcile_phase: ReconcilePhase::Completed,
            previous_deployment: None,
            pod_status: None,
        };

        let ingress = controller.create_primary_ingress(&project, &deployment, &metadata);

        // Verify the x-forwarded-prefix annotation is NOT present (subdomain-based routing)
        if let Some(annotations) = &ingress.metadata.annotations {
            assert!(
                !annotations.contains_key("nginx.ingress.kubernetes.io/x-forwarded-prefix"),
                "Subdomain-based ingress should NOT have x-forwarded-prefix annotation"
            );
        }

        // Verify the host is a subdomain
        let rules = ingress.spec.unwrap().rules.unwrap();
        assert_eq!(rules.len(), 1);
        assert_eq!(
            rules[0].host,
            Some("myapp.test.local".to_string()),
            "Subdomain-based ingress should use subdomain"
        );
    }

    // Helper function to create a mock controller with path-based routing
    fn create_mock_controller_with_path_based_routing() -> KubernetesController {
        use crate::server::state::ControllerState;
        use axum::http::Uri;
        use sqlx::postgres::PgPoolOptions;

        rustls::crypto::ring::default_provider()
            .install_default()
            .ok();

        let pool = PgPoolOptions::new()
            .max_connections(1)
            .connect_lazy(TEST_DATABASE_URL)
            .expect("Failed to create pool");

        let state = ControllerState {
            db_pool: pool,
            encryption_provider: None,
        };

        let cluster_url = "http://localhost:8080"
            .parse::<Uri>()
            .expect("Failed to parse URI");
        let kube_config = kube::Config::new(cluster_url);
        let kube_client = kube::Client::try_from(kube_config).expect("Failed to create client");

        // Create default access classes for testing
        let mut access_classes = std::collections::HashMap::new();
        access_classes.insert(
            "public".to_string(),
            crate::server::settings::AccessClass {
                display_name: "Public".to_string(),
                description: "Publicly accessible".to_string(),
                ingress_class: "nginx".to_string(),
                access_requirement: crate::server::settings::AccessRequirement::None,
                custom_annotations: std::collections::HashMap::new(),
            },
        );

        KubernetesController {
            state,
            kube_client,
            // Path-based routing template: rise.dev/{project_name}
            production_ingress_url_template: "rise.dev/{project_name}".to_string(),
            staging_ingress_url_template: None,
            ingress_port: None,
            ingress_schema: "https".to_string(),
            registry_provider: Arc::new(MockRegistryProvider),
            auth_backend_url: "http://localhost:3000".to_string(),
            auth_signin_url: "http://localhost:3000".to_string(),
            backend_address: None,
            namespace_labels: std::collections::HashMap::new(),
            namespace_annotations: std::collections::HashMap::new(),
            ingress_annotations: std::collections::HashMap::new(),
            ingress_tls_secret_name: None,
            custom_domain_tls_mode: crate::server::settings::CustomDomainTlsMode::PerDomain,
            custom_domain_ingress_annotations: std::collections::HashMap::new(),
            node_selector: std::collections::HashMap::new(),
            image_pull_secret_name: None,
            access_classes,
            resource_versions: Arc::new(std::sync::RwLock::new(std::collections::HashMap::new())),
            host_aliases: std::collections::HashMap::new(),
            extra_service_token_audiences: std::collections::HashMap::new(),
            network_policy: test_network_policy(),
            pod_security_enabled: true,
            pod_resources: None,
            health_probes: None,
        }
    }

    fn create_test_project_and_deployment() -> (Project, Deployment) {
        use crate::db::models::{Deployment, Project, ProjectStatus};
        use uuid::Uuid;

        let project = Project {
            id: Uuid::new_v4(),
            name: "test-project".to_string(),
            status: ProjectStatus::Running,
            access_class: "public".to_string(),
            owner_user_id: None,
            owner_team_id: None,
            finalizers: vec![],
            created_at: chrono::Utc::now(),
            updated_at: chrono::Utc::now(),
        };
        let deployment = Deployment {
            id: Uuid::new_v4(),
            project_id: project.id,
            deployment_id: "20240108-103000".to_string(),
            created_by_id: Uuid::new_v4(),
            deployment_group: "default".to_string(),
            status: crate::db::models::DeploymentStatus::Deploying,
            expires_at: None,
            termination_reason: None,
            completed_at: None,
            error_message: None,
            build_logs: None,
            controller_metadata: serde_json::json!({}),
            image: None,
            image_digest: None,
            rolled_back_from_deployment_id: None,
            http_port: 8080,
            needs_reconcile: false,
            is_active: false,
            deploying_started_at: None,
            first_healthy_at: None,
            created_at: chrono::Utc::now(),
            updated_at: chrono::Utc::now(),
        };
        (project, deployment)
    }

    #[test]
    fn reconcile_in_progress_status_stays_post_deploy_after_first_healthy() {
        let (_, mut deployment) = create_test_project_and_deployment();
        deployment.status = DeploymentStatus::Healthy;
        deployment.first_healthy_at = Some(Utc::now());

        assert_eq!(
            reconcile_in_progress_status(&deployment),
            DeploymentStatus::Unhealthy
        );
    }

    #[test]
    fn reconcile_in_progress_status_uses_deploying_before_first_healthy() {
        let (_, deployment) = create_test_project_and_deployment();

        assert_eq!(
            reconcile_in_progress_status(&deployment),
            DeploymentStatus::Deploying
        );
    }

    #[tokio::test]
    async fn test_network_policy_standard_config() {
        let controller = create_mock_controller();
        let (project, deployment) = create_test_project_and_deployment();
        let np = controller.create_network_policy(&project, &deployment, "ns");

        let spec = np.spec.as_ref().unwrap();

        // Standard config: policyTypes is ["Ingress"] only (no egress restriction)
        assert_eq!(
            spec.policy_types.as_ref().unwrap(),
            &vec!["Ingress".to_string()]
        );

        // No egress rules
        assert!(spec.egress.is_none());

        // 2 ingress rules: ingress-nginx + intra-namespace
        let ingress = spec.ingress.as_ref().unwrap();
        assert_eq!(ingress.len(), 2);

        // First rule: ingress-nginx namespace+pod selector
        let first_from = ingress[0].from.as_ref().unwrap();
        assert_eq!(first_from.len(), 1);
        let ns_selector = first_from[0].namespace_selector.as_ref().unwrap();
        assert_eq!(
            ns_selector
                .match_labels
                .as_ref()
                .unwrap()
                .get("kubernetes.io/metadata.name"),
            Some(&"ingress-nginx".to_string())
        );
        let pod_selector = first_from[0].pod_selector.as_ref().unwrap();
        assert_eq!(
            pod_selector
                .match_labels
                .as_ref()
                .unwrap()
                .get("app.kubernetes.io/name"),
            Some(&"ingress-nginx".to_string())
        );

        // Second rule: intra-namespace (empty pod selector, no namespace selector)
        let second_from = ingress[1].from.as_ref().unwrap();
        assert_eq!(second_from.len(), 1);
        assert!(second_from[0].namespace_selector.is_none());
        let pod_selector = second_from[0].pod_selector.as_ref().unwrap();
        assert!(pod_selector.match_labels.as_ref().unwrap().is_empty());
    }

    #[tokio::test]
    async fn test_network_policy_explicit_egress() {
        use k8s_openapi::api::networking::v1::{
            IPBlock, NetworkPolicyEgressRule, NetworkPolicyPeer, NetworkPolicyPort,
        };
        use k8s_openapi::apimachinery::pkg::util::intstr::IntOrString;

        let mut controller = create_mock_controller();
        controller.network_policy.egress = Some(vec![NetworkPolicyEgressRule {
            to: Some(vec![NetworkPolicyPeer {
                ip_block: Some(IPBlock {
                    cidr: "10.0.0.0/8".to_string(),
                    except: None,
                }),
                ..Default::default()
            }]),
            ports: Some(vec![NetworkPolicyPort {
                protocol: Some("TCP".to_string()),
                port: Some(IntOrString::Int(443)),
                ..Default::default()
            }]),
        }]);

        let (project, deployment) = create_test_project_and_deployment();
        let np = controller.create_network_policy(&project, &deployment, "ns");
        let spec = np.spec.as_ref().unwrap();

        // policyTypes includes both Ingress and Egress
        assert_eq!(
            spec.policy_types.as_ref().unwrap(),
            &vec!["Ingress".to_string(), "Egress".to_string()]
        );

        // Egress rules present
        let egress = spec.egress.as_ref().unwrap();
        assert_eq!(egress.len(), 1);
        let peer = &egress[0].to.as_ref().unwrap()[0];
        assert_eq!(peer.ip_block.as_ref().unwrap().cidr, "10.0.0.0/8");
    }

    #[tokio::test]
    async fn test_network_policy_empty_egress_denies_all() {
        let mut controller = create_mock_controller();
        controller.network_policy.egress = Some(vec![]);

        let (project, deployment) = create_test_project_and_deployment();
        let np = controller.create_network_policy(&project, &deployment, "ns");
        let spec = np.spec.as_ref().unwrap();

        // policyTypes includes Egress (deny-all since rules are empty)
        assert_eq!(
            spec.policy_types.as_ref().unwrap(),
            &vec!["Ingress".to_string(), "Egress".to_string()]
        );
        assert!(spec.egress.as_ref().unwrap().is_empty());
    }

    #[tokio::test]
    async fn test_network_policy_custom_ingress() {
        use k8s_openapi::api::networking::v1::{
            IPBlock, NetworkPolicyIngressRule, NetworkPolicyPeer, NetworkPolicyPort,
        };
        use k8s_openapi::apimachinery::pkg::util::intstr::IntOrString;

        let mut controller = create_mock_controller();
        controller.network_policy.ingress = vec![NetworkPolicyIngressRule {
            from: Some(vec![NetworkPolicyPeer {
                ip_block: Some(IPBlock {
                    cidr: "203.0.113.0/24".to_string(),
                    except: None,
                }),
                ..Default::default()
            }]),
            ports: Some(vec![NetworkPolicyPort {
                protocol: Some("TCP".to_string()),
                port: Some(IntOrString::Int(80)),
                ..Default::default()
            }]),
        }];

        let (project, deployment) = create_test_project_and_deployment();
        let np = controller.create_network_policy(&project, &deployment, "ns");
        let spec = np.spec.as_ref().unwrap();

        let ingress = spec.ingress.as_ref().unwrap();
        assert_eq!(ingress.len(), 1);
        let peer = &ingress[0].from.as_ref().unwrap()[0];
        assert_eq!(peer.ip_block.as_ref().unwrap().cidr, "203.0.113.0/24");
        let ports = ingress[0].ports.as_ref().unwrap();
        assert_eq!(ports.len(), 1);
        assert_eq!(
            ports[0].port,
            Some(k8s_openapi::apimachinery::pkg::util::intstr::IntOrString::Int(80))
        );
    }
}
