#![cfg(feature = "discovery-kube")]

use std::time::{Duration, SystemTime, UNIX_EPOCH};

use k8s_openapi::api::discovery::v1::{Endpoint, EndpointConditions, EndpointPort, EndpointSlice};
use k8s_openapi::apimachinery::pkg::apis::meta::v1::ObjectMeta;
use kube::api::{DeleteParams, PostParams};
use kube::{Api, Client};
use rs_zero::discovery::Discovery;
use rs_zero::discovery_kube::{KubeDiscovery, KubeDiscoveryConfig};

#[tokio::test]
#[ignore = "requires local Kubernetes cluster from Docker Desktop or compatible kubeconfig"]
async fn kube_discovery_external_endpoint_slice_lifecycle() {
    if std::env::var("RS_ZERO_KUBE_EXTERNAL").ok().as_deref() != Some("1") {
        eprintln!("set RS_ZERO_KUBE_EXTERNAL=1 to run this external Kubernetes test");
        return;
    }

    let context = std::env::var("RS_ZERO_KUBE_CONTEXT")
        .ok()
        .or_else(|| Some("docker-desktop".to_string()));
    let client = kubernetes_client(context.as_deref())
        .await
        .expect("connect Kubernetes");
    let namespace =
        std::env::var("RS_ZERO_KUBE_NAMESPACE").unwrap_or_else(|_| "default".to_string());
    let service = format!("rs-zero-kube-{}", unique_suffix());
    let slice_name = format!("{service}-slice");
    let slices: Api<EndpointSlice> = Api::namespaced(client.clone(), &namespace);

    slices
        .create(
            &PostParams::default(),
            &endpoint_slice(&slice_name, &service, true),
        )
        .await
        .expect("create endpoint slice");

    let discovery = KubeDiscovery::from_client(
        KubeDiscoveryConfig {
            context,
            namespace: namespace.clone(),
            service_name: Some(service.clone()),
            port_name: Some("http".to_string()),
            sync_timeout: Duration::from_secs(10),
            ..KubeDiscoveryConfig::default()
        },
        client,
    )
    .await
    .expect("start watcher");

    let discovered = discover_until_ready(&discovery, &service).await;
    assert_eq!(discovered.len(), 1);
    assert_eq!(discovered[0].endpoint.authority(), "10.0.0.10:8080");

    slices
        .delete(&slice_name, &DeleteParams::default())
        .await
        .expect("delete endpoint slice");
    assert_discover_empty(&discovery, &service).await;
}

async fn kubernetes_client(context: Option<&str>) -> Result<Client, kube::Error> {
    if let Some(context) = context {
        let mut config = kube::Config::from_kubeconfig(&kube::config::KubeConfigOptions {
            context: Some(context.to_string()),
            cluster: None,
            user: None,
        })
        .await
        .map_err(kube::Error::InferKubeconfig)?;
        config.apply_debug_overrides();
        Client::try_from(config)
    } else {
        Client::try_default().await
    }
}

fn endpoint_slice(name: &str, service: &str, ready: bool) -> EndpointSlice {
    EndpointSlice {
        address_type: "IPv4".to_string(),
        metadata: ObjectMeta {
            name: Some(name.to_string()),
            labels: Some(std::collections::BTreeMap::from([(
                "kubernetes.io/service-name".to_string(),
                service.to_string(),
            )])),
            ..ObjectMeta::default()
        },
        ports: Some(vec![EndpointPort {
            name: Some("http".to_string()),
            port: Some(8080),
            protocol: Some("TCP".to_string()),
            ..EndpointPort::default()
        }]),
        endpoints: vec![Endpoint {
            addresses: vec!["10.0.0.10".to_string()],
            conditions: Some(EndpointConditions {
                ready: Some(ready),
                terminating: Some(false),
                serving: Some(ready),
            }),
            ..Endpoint::default()
        }],
    }
}

fn unique_suffix() -> String {
    let millis = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_millis();
    format!("{millis}-{}", std::process::id())
}

async fn discover_until_ready(
    discovery: &KubeDiscovery,
    service: &str,
) -> Vec<rs_zero::discovery::ServiceInstance> {
    for _ in 0..40 {
        if let Ok(instances) = discovery.discover(service).await {
            return instances;
        }
        tokio::time::sleep(Duration::from_millis(100)).await;
    }
    discovery
        .discover(service)
        .await
        .expect("discover eventually")
}

async fn assert_discover_empty(discovery: &KubeDiscovery, service: &str) {
    for _ in 0..40 {
        if discovery.discover(service).await.is_err() {
            return;
        }
        tokio::time::sleep(Duration::from_millis(100)).await;
    }
    panic!("service {service} still discoverable");
}
