staircase 0.0.7

use std::collections::BTreeMap;

use k8s_openapi::api::core::v1::Service;
use kube::api::{ListParams, ObjectList};
use thiserror::Error;

use crate::resources::{ApiResource, ComponentizedResource, LabeledResource, NamedResource};

/// gets the internal host of a service within the cluster.
pub fn service_host<T: ComponentizedResource<Resource = Service>, O>(owner: &O) -> String
where
    O: NamedResource<T>,
{
    format!(
        "{}.{}.svc.cluster.local",
        NamedResource::<T>::name(owner),
        NamedResource::<T>::ns(owner).unwrap(), /* SAFETY: services are always [`kube::NamespaceResourceScope`],
                                                 * which always have a namespace */
    )
}

/// gets the internal full url of a service within the cluster.
/// parameters:
///  - *scheme*: like `http` or `https`
///  - *port_path_query*: postfix of the url, e.g.
///    `:8080/metrics?since=2025-01-01`
pub fn service_url<T: ComponentizedResource<Resource = Service>, O>(
    owner: &O,
    scheme: &str,
    port_path_query: &str,
) -> String
where
    O: NamedResource<T>,
{
    format!(
        "{}://{}.{}.svc.cluster.local{}",
        scheme,
        NamedResource::<T>::name(owner),
        NamedResource::<T>::ns(owner).unwrap(), /* SAFETY: services are always [`kube::NamespaceResourceScope`],
                                                 * which always have a namespace */
        port_path_query,
    )
}

/// performs a [`kube::Api::list`] via the [`ApiResource`] and
/// [`LabeledResource`] trait
pub async fn list_all_by_label<I, T: ComponentizedResource>(
    ident: &I,
    client: &kube::Client,
) -> Result<ObjectList<T::Resource>, kube::Error>
where
    I: ApiResource<T>,
    I: LabeledResource<T>,
{
    let api = ident.api(client.clone());
    let lp = ListParams {
        label_selector: Some(ident.selector_all()),
        ..Default::default()
    };

    let list = api.list(&lp).await?;
    Ok(list)
}

/// Error returned when a label selector is expected to resolve to exactly one
/// resource.
#[derive(Error, Debug)]
pub enum ListOneError<R: Clone> {
    /// No resource matched the selector.
    #[error("expected exactly one resource, found none for selector `{0}`")]
    NoResourceFound(String),
    /// Kubernetes API request failed while listing resources.
    #[error("Kubernetes API request failed: `{0}`")]
    Kube(#[from] kube::Error),
    /// More than one resource matched the selector.
    #[error("expected exactly one resource, found more than one for selector `{1}`")]
    TooManyResourcesFound(Box<ObjectList<R>>, String),
}

/// like [`list_one_by_label`] but just returns a single resource.
pub async fn list_one_by_label<I, T: ComponentizedResource>(
    ident: &I,
    client: &kube::Client,
) -> Result<T::Resource, ListOneError<T::Resource>>
where
    I: ApiResource<T>,
    I: LabeledResource<T>,
{
    let list = list_all_by_label(ident, client).await?;
    if list.items.len() > 1 {
        return Err(ListOneError::TooManyResourcesFound(
            Box::new(list),
            ident.selector_all(),
        ));
    }
    if let Some(resource) = list.items.into_iter().next() {
        Ok(resource)
    } else {
        Err(ListOneError::NoResourceFound(ident.selector_all()))
    }
}

/// performs a [`kube::Api::get_opt`] via the [`ApiResource`] and
/// [`NamedResource`] trait.
pub async fn get_opt_by_name<I, T: ComponentizedResource>(
    ident: &I,
    client: &kube::Client,
) -> Result<Option<T::Resource>, kube::Error>
where
    I: ApiResource<T>,
    I: NamedResource<T>,
{
    let api = ident.api(client.clone());
    api.get_opt(&ident.name()).await
}

pub fn identifier_component_label<I, T: ComponentizedResource, F: FnOnce(T::Component) -> String>(
    ident: &I,
    comp_fn: F,
    identifier_key: &str,
    component_key: &str,
) -> BTreeMap<String, String>
where
    I: NamedResource<T>,
{
    BTreeMap::from([
        (identifier_key.to_string(), NamedResource::name(ident)),
        (component_key.to_string(), comp_fn(T::COMPONENT)),
    ])
}

#[cfg(test)]
mod tests {
    use std::{
        collections::VecDeque,
        convert::Infallible,
        sync::{Arc, Mutex},
    };

    use http::{Method, StatusCode, Uri};
    use k8s_openapi::{
        api::core::v1::{Pod, Service},
        apimachinery::pkg::apis::meta::v1::ObjectMeta,
    };
    use kube::{CustomResource, client::Body};
    use schemars::JsonSchema;
    use serde::{Deserialize, Serialize};
    use serde_json::{Value, json};

    use super::*;

    #[derive(Debug, PartialEq, Eq)]
    enum Component {
        Service,
    }

    struct ServiceComponent;

    impl ComponentizedResource for ServiceComponent {
        type Component = Component;
        type Resource = Service;

        const COMPONENT: Self::Component = Component::Service;
    }

    struct PodComponent;

    impl ComponentizedResource for PodComponent {
        type Component = Component;
        type Resource = Pod;

        const COMPONENT: Self::Component = Component::Service;
    }

    struct Owner {
        name:      String,
        namespace: String,
    }

    impl NamedResource<ServiceComponent> for Owner {
        fn name(&self) -> String { self.name.clone() }

        fn ns(&self) -> Option<String> { Some(self.namespace.clone()) }
    }

    fn owner() -> Owner {
        Owner {
            name:      "sample-web".to_string(),
            namespace: "apps".to_string(),
        }
    }

    #[derive(CustomResource, Deserialize, Serialize, Clone, Debug, JsonSchema)]
    #[kube(group = "tests.staircase.dev", version = "v1", kind = "LookupOwner", namespaced)]
    struct LookupOwnerSpec {}

    impl LabeledResource<PodComponent> for LookupOwner {
        fn selector_all(&self) -> String { format!("app={},component=pod", self.metadata.name.as_ref().unwrap()) }
    }

    impl NamedResource<PodComponent> for LookupOwner {
        fn name(&self) -> String { format!("{}-pod", self.metadata.name.as_ref().unwrap()) }

        fn ns(&self) -> Option<String> { self.metadata.namespace.clone() }
    }

    #[derive(Clone, Debug)]
    struct CapturedRequest {
        method: Method,
        uri:    Uri,
    }

    #[derive(Clone, Debug)]
    struct QueuedResponse {
        status: StatusCode,
        body:   Value,
    }

    fn lookup_owner() -> LookupOwner {
        LookupOwner {
            metadata: ObjectMeta {
                name: Some("sample".to_string()),
                namespace: Some("apps".to_string()),
                ..Default::default()
            },
            spec:     LookupOwnerSpec {},
        }
    }

    fn pod(name: &str) -> Pod {
        Pod {
            metadata: ObjectMeta {
                name: Some(name.to_string()),
                ..Default::default()
            },
            ..Default::default()
        }
    }

    fn pod_list(items: Vec<Pod>) -> Value {
        serde_json::to_value(ObjectList {
            types: Default::default(),
            metadata: Default::default(),
            items,
        })
        .unwrap()
    }

    fn json_response(body: impl Serialize) -> QueuedResponse {
        QueuedResponse {
            status: StatusCode::OK,
            body:   serde_json::to_value(body).unwrap(),
        }
    }

    fn not_found_response() -> QueuedResponse {
        QueuedResponse {
            status: StatusCode::NOT_FOUND,
            body:   json!({
                "kind": "Status",
                "apiVersion": "v1",
                "metadata": {},
                "status": "Failure",
                "message": "pods \"sample-pod\" not found",
                "reason": "NotFound",
                "code": 404
            }),
        }
    }

    fn server_error_response() -> QueuedResponse {
        QueuedResponse {
            status: StatusCode::INTERNAL_SERVER_ERROR,
            body:   json!({
                "kind": "Status",
                "apiVersion": "v1",
                "metadata": {},
                "status": "Failure",
                "message": "mock resource lookup failure",
                "reason": "InternalError",
                "code": 500
            }),
        }
    }

    fn client_with_responses(responses: Vec<QueuedResponse>) -> (kube::Client, Arc<Mutex<Vec<CapturedRequest>>>) {
        let captured = Arc::new(Mutex::new(Vec::new()));
        let responses = Arc::new(Mutex::new(VecDeque::from(responses)));
        let service = tower::service_fn({
            let captured = captured.clone();
            let responses = responses.clone();
            move |request: http::Request<Body>| {
                let captured = captured.clone();
                let responses = responses.clone();
                async move {
                    let (parts, _) = request.into_parts();
                    captured.lock().unwrap().push(CapturedRequest {
                        method: parts.method,
                        uri:    parts.uri,
                    });

                    let response = responses.lock().unwrap().pop_front().expect("unexpected request");
                    Ok::<_, Infallible>(
                        http::Response::builder()
                            .status(response.status)
                            .header("content-type", "application/json")
                            .body(Body::from(serde_json::to_vec(&response.body).unwrap()))
                            .unwrap(),
                    )
                }
            }
        });

        (kube::Client::new(service, "default"), captured)
    }

    #[test]
    fn service_host_uses_named_resource_name_and_namespace() {
        assert_eq!(
            service_host::<ServiceComponent, _>(&owner()),
            "sample-web.apps.svc.cluster.local"
        );
    }

    #[test]
    fn service_url_uses_scheme_name_namespace_and_suffix() {
        assert_eq!(
            service_url::<ServiceComponent, _>(&owner(), "https", ":8443/healthz"),
            "https://sample-web.apps.svc.cluster.local:8443/healthz"
        );
    }

    #[test]
    fn identifier_component_label_uses_custom_keys_and_component_formatter() {
        let labels = identifier_component_label::<_, ServiceComponent, _>(
            &owner(),
            |component| format!("{component:?}").to_lowercase(),
            "app.kubernetes.io/instance",
            "app.kubernetes.io/component",
        );

        assert_eq!(labels["app.kubernetes.io/instance"], "sample-web");
        assert_eq!(labels["app.kubernetes.io/component"], "service");
    }

    #[test]
    fn list_all_by_label_uses_owner_namespace_and_selector() {
        let rt = tokio::runtime::Builder::new_current_thread().build().unwrap();

        rt.block_on(async {
            let (client, captured) = client_with_responses(vec![QueuedResponse {
                status: StatusCode::OK,
                body:   pod_list(vec![pod("sample-pod")]),
            }]);

            let list = list_all_by_label::<_, PodComponent>(&lookup_owner(), &client)
                .await
                .unwrap();

            assert_eq!(list.items.len(), 1);
            let captured = captured.lock().unwrap();
            assert_eq!(captured.len(), 1);
            assert_eq!(captured[0].method, Method::GET);
            assert_eq!(captured[0].uri.path(), "/api/v1/namespaces/apps/pods");
            assert_eq!(
                captured[0].uri.query(),
                Some("&labelSelector=app%3Dsample%2Ccomponent%3Dpod")
            );
        });
    }

    #[test]
    fn list_one_by_label_maps_empty_and_multiple_results_to_selector_errors() {
        let rt = tokio::runtime::Builder::new_current_thread().build().unwrap();

        rt.block_on(async {
            let (client, _) = client_with_responses(vec![QueuedResponse {
                status: StatusCode::OK,
                body:   pod_list(Vec::new()),
            }]);
            let error = list_one_by_label::<_, PodComponent>(&lookup_owner(), &client)
                .await
                .unwrap_err();
            assert!(matches!(
                error,
                ListOneError::NoResourceFound(selector) if selector == "app=sample,component=pod"
            ));

            let (client, _) = client_with_responses(vec![QueuedResponse {
                status: StatusCode::OK,
                body:   pod_list(vec![pod("first"), pod("second")]),
            }]);
            let error = list_one_by_label::<_, PodComponent>(&lookup_owner(), &client)
                .await
                .unwrap_err();
            assert!(matches!(
                error,
                ListOneError::TooManyResourcesFound(list, selector)
                    if list.items.len() == 2 && selector == "app=sample,component=pod"
            ));
        });
    }

    #[test]
    fn list_helpers_propagate_kubernetes_errors() {
        let rt = tokio::runtime::Builder::new_current_thread().build().unwrap();

        rt.block_on(async {
            let (client, captured) = client_with_responses(vec![server_error_response()]);
            let error = list_all_by_label::<_, PodComponent>(&lookup_owner(), &client)
                .await
                .unwrap_err();

            assert!(matches!(error, kube::Error::Api(_)));
            assert_eq!(captured.lock().unwrap().len(), 1);
        });

        rt.block_on(async {
            let (client, captured) = client_with_responses(vec![server_error_response()]);
            let error = list_one_by_label::<_, PodComponent>(&lookup_owner(), &client)
                .await
                .unwrap_err();

            assert!(matches!(error, ListOneError::Kube(kube::Error::Api(_))));
            assert_eq!(captured.lock().unwrap().len(), 1);
        });
    }

    #[test]
    fn get_opt_by_name_uses_named_resource_path_and_maps_not_found_to_none() {
        let rt = tokio::runtime::Builder::new_current_thread().build().unwrap();

        rt.block_on(async {
            let (client, captured) = client_with_responses(vec![json_response(pod("sample-pod"))]);
            let result = get_opt_by_name::<_, PodComponent>(&lookup_owner(), &client)
                .await
                .unwrap();

            assert_eq!(result.unwrap().metadata.name.as_deref(), Some("sample-pod"));
            let captured = captured.lock().unwrap();
            assert_eq!(captured.len(), 1);
            assert_eq!(captured[0].method, Method::GET);
            assert_eq!(captured[0].uri.path(), "/api/v1/namespaces/apps/pods/sample-pod");
        });

        rt.block_on(async {
            let (client, captured) = client_with_responses(vec![not_found_response()]);
            let result = get_opt_by_name::<_, PodComponent>(&lookup_owner(), &client)
                .await
                .unwrap();

            assert!(result.is_none());
            assert_eq!(captured.lock().unwrap().len(), 1);
        });
    }

    #[test]
    fn get_opt_by_name_propagates_non_not_found_kubernetes_errors() {
        let rt = tokio::runtime::Builder::new_current_thread().build().unwrap();

        rt.block_on(async {
            let (client, captured) = client_with_responses(vec![server_error_response()]);
            let error = get_opt_by_name::<_, PodComponent>(&lookup_owner(), &client)
                .await
                .unwrap_err();

            assert!(matches!(error, kube::Error::Api(_)));
            assert_eq!(captured.lock().unwrap().len(), 1);
        });
    }
}