caixa-feira 0.1.30

//! Cluster-side runtime for `feira ephemeral up|down|wait|status|list`.
//!
//! Every operation goes through `kube-rs` against the Process CRD in
//! `tatara.pleme.io/v1alpha1`. NO SHELL — no `kubectl` wrapping.
//!
//! Operator UX contract: a single `feira ephemeral up form.lisp --wait`
//! applies the Process CR and blocks until it reaches `Attested`, so
//! CI/test pipelines can branch on the return code without polling.

use std::time::Duration;

use anyhow::{Context, Result, anyhow};
use kube::Client;
use kube::api::{Api, DeleteParams, ListParams, Patch, PatchParams, PostParams};
use tatara_process::phase::ProcessPhase;
use tatara_process::prelude::Process;

/// Build an in-cluster or kubeconfig-based client (whichever applies).
pub async fn client() -> Result<Client> {
    Client::try_default()
        .await
        .context("failed to build kube client (need KUBECONFIG or in-cluster auth)")
}

/// Apply a `Process` via server-side apply (create-or-update) — the
/// caller has already filled in name + namespace.
pub async fn apply_process(client: Client, process: &Process) -> Result<Process> {
    let ns = process
        .metadata
        .namespace
        .clone()
        .ok_or_else(|| anyhow!("Process has no metadata.namespace"))?;
    let name = process
        .metadata
        .name
        .clone()
        .ok_or_else(|| anyhow!("Process has no metadata.name"))?;
    let api: Api<Process> = Api::namespaced(client, &ns);
    // Server-side apply with field manager = "feira" — idempotent
    // across re-runs, conflict-detectable if a competing controller
    // touches the same fields.
    let params = PatchParams::apply("feira").force();
    api.patch(&name, &params, &Patch::Apply(process))
        .await
        .with_context(|| format!("server-side apply Process {ns}/{name}"))
}

/// Plain create — fails with 409 if the Process exists.
pub async fn create_process(client: Client, process: &Process) -> Result<Process> {
    let ns = process
        .metadata
        .namespace
        .clone()
        .ok_or_else(|| anyhow!("Process has no metadata.namespace"))?;
    let api: Api<Process> = Api::namespaced(client, &ns);
    api.create(&PostParams::default(), process)
        .await
        .context("create Process")
}

/// Delete a Process by name. Returns Ok(true) if delete was issued,
/// Ok(false) if the Process didn't exist.
pub async fn delete_process(client: Client, ns: &str, name: &str) -> Result<bool> {
    let api: Api<Process> = Api::namespaced(client, ns);
    match api.delete(name, &DeleteParams::default()).await {
        Ok(_) => Ok(true),
        Err(kube::Error::Api(e)) if e.code == 404 => Ok(false),
        Err(e) => Err(anyhow!("delete Process {ns}/{name}: {e}")),
    }
}

/// Fetch a Process by name. None if it doesn't exist.
pub async fn get_process(client: Client, ns: &str, name: &str) -> Result<Option<Process>> {
    let api: Api<Process> = Api::namespaced(client, ns);
    api.get_opt(name)
        .await
        .with_context(|| format!("get Process {ns}/{name}"))
}

/// List Processes in a namespace. Optionally filter to ephemeral-lifetime.
pub async fn list_processes(
    client: Client,
    ns: &str,
    ephemeral_only: bool,
) -> Result<Vec<Process>> {
    let api: Api<Process> = Api::namespaced(client, ns);
    let list = api
        .list(&ListParams::default())
        .await
        .with_context(|| format!("list Processes in {ns}"))?;
    if ephemeral_only {
        Ok(list
            .items
            .into_iter()
            .filter(|p| p.spec.lifetime.is_ephemeral())
            .collect())
    } else {
        Ok(list.items)
    }
}

/// Poll a Process until it reaches `target` phase or `timeout` elapses.
/// Returns the final observed Process. Errors on timeout, on terminal
/// non-target phases (Reaped before Attested), and on kube errors.
pub async fn wait_for_phase(
    client: Client,
    ns: &str,
    name: &str,
    target: ProcessPhase,
    timeout: Duration,
    poll_interval: Duration,
) -> Result<Process> {
    let deadline = tokio::time::Instant::now() + timeout;
    loop {
        let current = get_process(client.clone(), ns, name)
            .await?
            .ok_or_else(|| anyhow!("Process {ns}/{name} disappeared while waiting"))?;
        let phase = current
            .status
            .as_ref()
            .map(|s| s.phase)
            .unwrap_or(ProcessPhase::Pending);

        if phase == target {
            return Ok(current);
        }

        // Terminal sinks short-circuit so we don't spin forever waiting
        // for Attested from a Failed/Zombie/Reaped Process.
        if phase == ProcessPhase::Reaped {
            return Err(anyhow!(
                "Process {ns}/{name} reached Reaped while waiting for {target:?}"
            ));
        }
        if phase == ProcessPhase::Failed && target != ProcessPhase::Failed {
            let msg = current
                .status
                .as_ref()
                .and_then(|s| s.message.clone())
                .unwrap_or_else(|| "no message".into());
            return Err(anyhow!(
                "Process {ns}/{name} reached Failed while waiting for {target:?}: {msg}"
            ));
        }

        if tokio::time::Instant::now() >= deadline {
            return Err(anyhow!(
                "timed out waiting for Process {ns}/{name} to reach {target:?} (current: {phase:?})"
            ));
        }
        tokio::time::sleep(poll_interval).await;
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    // The kube-touching paths can't be unit-tested without an apiserver.
    // We test the pure shape decisions: deadline math + wait-for-phase
    // terminal-short-circuit semantics via a small mock.

    #[test]
    fn target_phase_can_be_any_processphase() {
        // Sanity: ProcessPhase is Copy + comparable.
        let a = ProcessPhase::Attested;
        let b = ProcessPhase::Attested;
        assert_eq!(a, b);
    }

    #[test]
    fn duration_arithmetic_for_poll_loop() {
        // The poll loop uses `tokio::time::Instant::now() + timeout`
        // and `>= deadline`. Sanity-check the math.
        let now = std::time::Instant::now();
        let later = now + Duration::from_secs(60);
        assert!(later > now);
        assert!(later.duration_since(now).as_secs() == 60);
    }
}