staircase 0.0.7

use std::{error::Error, future::Future};

use k8s_openapi::{
    NamespaceResourceScope,
    serde::{Serialize, de::DeserializeOwned},
};
use kube::{
    Api, Resource,
    api::{Patch, PatchParams},
    core::object::HasStatus,
};
use serde_json::json;
use thiserror::Error;

use super::RunContext;

/// Writes status updates produced by modified steps.
///
/// Status updates are best-effort in the step-based model. If a status update
/// fails after a step modified external state, staircase logs and ignores that
/// status error so the next run can continue from observed state.
///
/// Use [`KubeStatusUpdater`] for the default Kubernetes API implementation.
pub trait StatusUpdater<R>
where
    R: HasStatus,
{
    type Error: Error + std::fmt::Debug;

    fn update(
        &self,
        context: &RunContext<R>,
        status: R::Status,
    ) -> impl Future<Output = Result<(), Self::Error>> + Send;
}

#[derive(Error, Debug)]
pub enum KubeStatusUpdaterError {
    /// Kubernetes API request failed while patching the status subresource.
    #[error("error updating status via the Kubernetes API server: `{0}`")]
    Kube(#[from] kube::Error),
    /// Status value could not be serialized into the merge patch body.
    #[error("error serializing status, this is a programming error: `{0}`")]
    Serde(#[from] serde_json::Error),
}

/// Default implementation of [`StatusUpdater`].
pub struct KubeStatusUpdater {}

impl<R> StatusUpdater<R> for KubeStatusUpdater
where
    R: Resource<Scope = NamespaceResourceScope, DynamicType: Default> + HasStatus + Send,
    R::Status: Serialize + Send,
    R: DeserializeOwned,
{
    type Error = KubeStatusUpdaterError;

    fn update(
        &self,
        context: &RunContext<R>,
        status: R::Status,
    ) -> impl Future<Output = Result<(), Self::Error>> + Send {
        let meta = context.resource.meta();
        use kube::ResourceExt;
        let name = context.resource.name_any();
        let namespace = meta.namespace.clone();
        let client = context.client.clone();

        async move {
            if let Some(namespace) = &namespace {
                let api: Api<R> = Api::namespaced(client, namespace);
                let patch = json!({ "status": status });
                api.patch_status(&name, &PatchParams::default(), &Patch::Merge(&patch))
                    .await?;
            } else {
                #[cfg(feature = "controller_trace")]
                tracing::error!("It looks like the resource does not contain have namespace information");
            }
            Ok(())
        }
    }
}

#[allow(unused_variables)]
pub(crate) fn status_log_and_ignore<E: Error + std::fmt::Debug>(error: E) {
    #[cfg(feature = "controller_trace")]
    tracing::error!(
        ?error,
        "status update did not succeed, this is considered a non-issue if the status will be eventually updated."
    );
}

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

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

    use super::*;

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

    #[derive(Deserialize, Serialize, Clone, Debug, Default, PartialEq, Eq, JsonSchema)]
    struct StatusUpdateTestStatus {
        observed: i32,
        phase:    String,
    }

    #[derive(Clone, Debug)]
    struct CapturedRequest {
        method:  Method,
        uri:     Uri,
        headers: HeaderMap,
        body:    Vec<u8>,
    }

    fn client_for_status_response(
        response: StatusUpdateTest,
        captured: Arc<Mutex<Vec<CapturedRequest>>>,
    ) -> kube::Client {
        let service = tower::service_fn(move |request: http::Request<Body>| {
            let captured = captured.clone();
            let response = response.clone();
            async move {
                let (parts, body) = request.into_parts();
                let body = body.collect_bytes().await.unwrap().to_vec();
                captured.lock().unwrap().push(CapturedRequest {
                    method: parts.method,
                    uri: parts.uri,
                    headers: parts.headers,
                    body,
                });

                Ok::<_, Infallible>(
                    http::Response::builder()
                        .header("content-type", "application/json")
                        .body(Body::from(serde_json::to_vec(&response).unwrap()))
                        .unwrap(),
                )
            }
        });

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

    fn client_for_status_error(captured: Arc<Mutex<Vec<CapturedRequest>>>) -> kube::Client {
        let service = tower::service_fn(move |request: http::Request<Body>| {
            let captured = captured.clone();
            async move {
                let (parts, body) = request.into_parts();
                let body = body.collect_bytes().await.unwrap().to_vec();
                captured.lock().unwrap().push(CapturedRequest {
                    method: parts.method,
                    uri: parts.uri,
                    headers: parts.headers,
                    body,
                });

                Ok::<_, Infallible>(
                    http::Response::builder()
                        .status(StatusCode::INTERNAL_SERVER_ERROR)
                        .header("content-type", "application/json")
                        .body(Body::from(
                            serde_json::to_vec(&json!({
                                "kind": "Status",
                                "apiVersion": "v1",
                                "metadata": {},
                                "status": "Failure",
                                "message": "mock status patch failure",
                                "reason": "InternalError",
                                "code": 500
                            }))
                            .unwrap(),
                        ))
                        .unwrap(),
                )
            }
        });

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

    fn resource(namespace: Option<&str>) -> StatusUpdateTest {
        StatusUpdateTest {
            metadata: ObjectMeta {
                name: namespace.map(|_| "sample".to_string()),
                namespace: namespace.map(str::to_string),
                ..Default::default()
            },
            spec:     StatusUpdateTestSpec {},
            status:   None,
        }
    }

    fn context(resource: StatusUpdateTest, client: kube::Client) -> RunContext<StatusUpdateTest> {
        RunContext {
            resource: Arc::new(resource),
            client,
            run_start: Timestamp::now(),
        }
    }

    fn request_json(request: &CapturedRequest) -> Value { serde_json::from_slice(&request.body).unwrap() }

    #[test]
    fn kube_status_updater_sends_merge_patch_to_status_subresource() {
        let captured = Arc::new(Mutex::new(Vec::new()));
        let status = StatusUpdateTestStatus {
            observed: 42,
            phase:    "Ready".to_string(),
        };
        let rt = tokio::runtime::Builder::new_current_thread().build().unwrap();

        rt.block_on(async {
            let mut response = resource(Some("apps"));
            response.status = Some(status.clone());
            let client = client_for_status_response(response, captured.clone());
            let context = context(resource(Some("apps")), client);

            KubeStatusUpdater {}.update(&context, status.clone()).await.unwrap();
        });

        let captured = captured.lock().unwrap();
        assert_eq!(captured.len(), 1);
        let request = &captured[0];
        assert_eq!(request.method, Method::PATCH);
        assert_eq!(
            request.uri.path(),
            "/apis/tests.staircase.dev/v1/namespaces/apps/statusupdatetests/sample/status"
        );
        assert_eq!(
            request.headers.get("content-type").unwrap(),
            "application/merge-patch+json"
        );
        assert_eq!(
            request_json(request),
            json!({
                "status": {
                    "observed": 42,
                    "phase": "Ready"
                }
            })
        );
    }

    #[test]
    fn kube_status_updater_without_namespace_is_noop() {
        let captured = Arc::new(Mutex::new(Vec::new()));
        let status = StatusUpdateTestStatus {
            observed: 42,
            phase:    "Ready".to_string(),
        };
        let rt = tokio::runtime::Builder::new_current_thread().build().unwrap();

        rt.block_on(async {
            let client = client_for_status_response(resource(None), captured.clone());
            let context = context(resource(None), client);

            KubeStatusUpdater {}.update(&context, status).await.unwrap();
        });

        assert!(captured.lock().unwrap().is_empty());
    }

    #[test]
    fn kube_status_updater_returns_kube_error_when_patch_fails() {
        let captured = Arc::new(Mutex::new(Vec::new()));
        let status = StatusUpdateTestStatus {
            observed: 42,
            phase:    "Ready".to_string(),
        };
        let rt = tokio::runtime::Builder::new_current_thread().build().unwrap();

        let error = rt.block_on(async {
            let client = client_for_status_error(captured.clone());
            let context = context(resource(Some("apps")), client);

            KubeStatusUpdater {}.update(&context, status).await.unwrap_err()
        });

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