kubus 0.4.2

//! Kubernetes operator framework for Rust
//!
//! Kubus provides a `#[kubus]` macro for building Kubernetes operators with minimal boilerplate.
//! It wraps [kube-rs](https://kube.rs) controllers with an ergonomic function-based API.
//!
//! # Examples
//!
//! ## Basic operator
//!
//! ```rust,no_run
//! use std::{sync::Arc, time::Duration};
//! use k8s_openapi::api::core::v1::Pod;
//! use kube::{Client, ResourceExt};
//! use kubus::{Context, HandlerError, Operator, kubus};
//!
//! #[tokio::main]
//! async fn main() -> Result<(), kubus::Error> {
//!     let client = Client::try_default().await?;
//!
//!     Operator::builder()
//!         .with_context(client)
//!         .handler(on_pod)
//!         .run()
//!         .await
//! }
//!
//! #[kubus(event = Apply)]
//! async fn on_pod(pod: Arc<Pod>, _ctx: Arc<Context>) -> Result<(), HandlerError> {
//!     println!("Pod {} in namespace {}",
//!         pod.name_unchecked(),
//!         pod.namespace().unwrap()
//!     );
//!     Ok(())
//! }
//! ```
//!
//! ## Multiple handlers with label selectors
//!
//! ```rust,no_run
//! use std::{sync::Arc, time::Duration};
//! use k8s_openapi::api::core::v1::Pod;
//! use kube::{Client, ResourceExt};
//! use kubus::{Context, HandlerError, Operator, kubus};
//!
//! #[tokio::main]
//! async fn main() -> Result<(), kubus::Error> {
//!     let client = Client::try_default().await?;
//!
//!     Operator::builder()
//!         .with_context(client)
//!         .handler(on_pod_apply)
//!         .handler(on_pod_delete)
//!         .run()
//!         .await
//! }
//!
//! #[kubus(
//!     event = Apply,
//!     label_selector = "app.kubernetes.io/managed-by=kubus"
//! )]
//! async fn on_pod_apply(pod: Arc<Pod>, _ctx: Arc<Context>) -> Result<(), HandlerError> {
//!     println!("Apply: {}", pod.name_unchecked());
//!     Ok(())
//! }
//!
//! #[kubus(
//!     event = Delete,
//!     label_selector = "app.kubernetes.io/managed-by=kubus"
//! )]
//! async fn on_pod_delete(pod: Arc<Pod>, _ctx: Arc<Context>) -> Result<(), HandlerError> {
//!     println!("Delete: {}", pod.name_unchecked());
//!     Ok(())
//! }
//! ```
//!
//! ## Custom state and finalizers
//!
//! ```rust,no_run
//! use std::{sync::Arc, time::Duration};
//! use k8s_openapi::api::core::v1::ConfigMap;
//! use kube::{Client, ResourceExt};
//! use kubus::{Context, HandlerError, Operator, kubus};
//!
//! #[derive(Debug, Clone)]
//! struct State {
//!     db_pool: String,
//! }
//!
//! #[tokio::main]
//! async fn main() -> Result<(), kubus::Error> {
//!     let client = Client::try_default().await?;
//!     let state = State { db_pool: "connection_string".to_string() };
//!
//!     Operator::builder()
//!         .with_context((client, state))
//!         .handler(on_configmap_apply)
//!         .handler(on_configmap_delete)
//!         .run()
//!         .await
//! }
//!
//! #[kubus(event = Apply, finalizer = "kubus.io/cleanup")]
//! async fn on_configmap_apply(
//!     cm: Arc<ConfigMap>,
//!     ctx: Arc<Context<State>>
//! ) -> Result<(), HandlerError> {
//!     println!("ConfigMap {} - db: {}", cm.name_unchecked(), ctx.data.db_pool);
//!     Ok(())
//! }
//!
//! #[kubus(event = Delete, finalizer = "kubus.io/cleanup")]
//! async fn on_configmap_delete(
//!     cm: Arc<ConfigMap>,
//!     _ctx: Arc<Context<State>>
//! ) -> Result<(), HandlerError> {
//!     println!("Cleanup for {}", cm.name_unchecked());
//!     Ok(())
//! }
//! ```
//!
//! # Macro attributes
//!
//! - `event`: `Apply` or `Delete` - required
//! - `finalizer`: Finalizer name for cleanup on deletion
//! - `label_selector`: Filter resources by labels
//! - `field_selector`: Filter resources by fields

use std::error::Error as StdError;
use std::fmt::Debug;
use std::hash::Hash;
use std::sync::Arc;
use std::time::Duration;

use async_trait::async_trait;
use futures::StreamExt;
use k8s_openapi::{ClusterResourceScope, NamespaceResourceScope};
use kube::api::{Api, DeleteParams, Patch, PatchParams};
use kube::runtime::controller::{Action, Controller};
use kube::runtime::watcher::Config;
use kube::{Client, Resource, ResourceExt};
use serde::de::DeserializeOwned;
use serde::ser::Serialize;
use serde_json::json;
use thiserror::Error;

pub use kubus_derive::*;
use tokio::task::JoinSet;
use tracing::{Instrument, info_span};

/// Errors that can occur during operator execution
#[derive(Error, Debug)]
pub enum Error {
    /// Error during JSON serialization/deserialization
    #[error("SerializationError: {0}")]
    SerializationError(#[source] serde_json::Error),
    /// Error from the Kubernetes client
    #[error("Kube Error: {0}")]
    KubeError(#[from] kube::Error),
    /// Error returned from the event handler
    #[error("Handler Error: {0}")]
    Handler(#[source] Box<dyn StdError + Send + Sync>),
}

/// Errors that can occur during event handler execution
#[derive(Error, Debug)]
pub enum HandlerError {
    /// Error from the Kubernetes client
    #[error("Kube Error: {0}")]
    KubeError(#[from] kube::Error),
    /// Error from the Kubus
    #[error("Kubus Error: {0}")]
    KubusError(#[from] Error),
}

/// Result type for Kubus operations
pub type Result<T, E = Error> = std::result::Result<T, E>;

/// Shared handler context
#[derive(Clone)]
pub struct Context<T = ()>
where
    T: Clone,
{
    /// Kube client
    pub client: Client,
    /// User defined data type
    pub data: T,
}

impl<T> From<(Client, T)> for Context<T>
where
    T: Clone,
{
    fn from((client, data): (Client, T)) -> Self {
        Self { client, data }
    }
}

impl From<Client> for Context {
    fn from(client: Client) -> Self {
        Self { client, data: () }
    }
}

/// Kubernetes resource event types
#[derive(Debug, PartialEq, Eq)]
pub enum EventType {
    /// Resource created or updated
    Apply,
    /// Resource deleted
    Delete,
}

/// Handler trait for Kubernetes resource events
///
/// Implement this trait (typically via the `#[kubus]` derive macro) to define
/// custom logic for responding to resource changes.
#[async_trait]
pub trait EventHandler<K, S, E = HandlerError>
where
    K: Resource + Clone + Debug + DeserializeOwned + Send + Sync + 'static,
    K::DynamicType: Clone + Debug + Default + Hash + Unpin + Eq,
    S: Clone + Send + Sync + 'static,
    E: StdError + Send + Sync + 'static,
{
    const NAME: &'static str;

    const LABEL_SELECTOR: Option<&'static str> = None;

    const FIELD_SELECTOR: Option<&'static str> = None;

    /// Handles a resource event
    ///
    /// Called when a resource is created, updated, or needs reconciliation.
    /// Returns an `Action` indicating when to reconcile again.
    async fn handler(resource: Arc<K>, context: Arc<Context<S>>) -> Result<Action, E>;

    /// Defines error handling policy for the handler
    ///
    /// Called when `handler` returns an error. Default implementation logs a warning
    /// and requeues after 5 seconds.
    fn error_policy(_resource: Arc<K>, err: &E, _ctx: Arc<Context<S>>) -> Action {
        tracing::error!({ err = err as &dyn StdError }, "Handler error");

        Action::requeue(Duration::from_secs(5))
    }

    /// Starts the controller watching for resource events
    ///
    /// Runs until the process receives a shutdown signal.
    async fn watch(client: Client, context: Arc<Context<S>>) -> Result<(), E>
    where
        Self: Sized + 'static,
    {
        tracing::info!("starting controller");

        let api = Api::<K>::all(client);
        let mut config = Config::default();
        config.label_selector = Self::LABEL_SELECTOR.map(String::from);
        config.field_selector = Self::FIELD_SELECTOR.map(String::from);

        Controller::new(api, config)
            .shutdown_on_signal()
            .run(Self::handler, Self::error_policy, context)
            .filter_map(|x| async move { std::result::Result::ok(x) })
            .for_each(|_| futures::future::ready(()))
            .await;

        Ok(())
    }
}

struct EventHandlerWrapper<H, K, S, E>
where
    H: EventHandler<K, S, E>,
    K: Resource + Clone + DeserializeOwned + Debug + Send + Sync + 'static,
    K::DynamicType: Clone + Debug + Default + Hash + Unpin + Eq,
    S: Clone + Send + Sync + 'static,
    E: StdError + Sync + Send + 'static,
{
    context: Arc<Context<S>>,
    _phantom: std::marker::PhantomData<(H, K, E)>,
}

impl<H, K, S, E> EventHandlerWrapper<H, K, S, E>
where
    H: EventHandler<K, S, E>,
    K: Resource + Clone + DeserializeOwned + Debug + Send + Sync + 'static,
    K::DynamicType: Clone + Debug + Default + Hash + Unpin + Eq,
    S: Clone + Send + Sync + 'static,
    E: StdError + Sync + Send + 'static,
{
    const fn new(context: Arc<Context<S>>) -> Self {
        Self {
            context,
            _phantom: std::marker::PhantomData,
        }
    }
}

#[async_trait]
trait DynEventHandler<E>: Send + Sync
where
    E: StdError + Send + Sync + 'static,
{
    fn name(&self) -> &'static str;

    async fn run(&self, client: Client) -> Result<(), E>;
}

#[async_trait]
impl<H, K, S, E> DynEventHandler<E> for EventHandlerWrapper<H, K, S, E>
where
    H: EventHandler<K, S, E> + Send + Sync + 'static,
    K: Resource + Clone + DeserializeOwned + Debug + Send + Sync + 'static,
    K::DynamicType: Clone + Debug + Default + Hash + Unpin + Eq,
    S: Clone + Send + Sync + 'static,
    E: StdError + Sync + Send + 'static,
{
    fn name(&self) -> &'static str {
        H::NAME
    }

    async fn run(&self, client: Client) -> Result<(), E> {
        let context = self.context.clone();
        H::watch(client, context).await
    }
}

/// Extensions for Kubernetes resource scopes
///
/// Provides methods to create appropriate `Api` instances for namespaced
/// or cluster-scoped resources.
pub trait ScopeExt<K>
where
    K: Resource<Scope = Self>,
{
    /// Creates an API client for the resource
    ///
    /// Returns a namespaced API if namespace is provided and the resource is namespaced,
    /// otherwise returns a cluster-wide API.
    fn api(client: Client, namespace: Option<impl AsRef<str>>) -> Api<K>;
}

impl<K> ScopeExt<K> for NamespaceResourceScope
where
    K: Resource<Scope = Self>,
    K::DynamicType: Default,
{
    fn api(client: Client, namespace: Option<impl AsRef<str>>) -> Api<K> {
        if let Some(namespace) = namespace {
            Api::namespaced(client, namespace.as_ref())
        } else {
            Api::all(client)
        }
    }
}

impl<K> ScopeExt<K> for ClusterResourceScope
where
    K: Resource<Scope = Self>,
    K::DynamicType: Default,
{
    fn api(client: Client, _: Option<impl AsRef<str>>) -> Api<K> {
        Api::all(client)
    }
}

#[async_trait]
pub trait ApiExt<K>
where
    K: Resource,
{
    async fn apply(self, client: &Client) -> kube::Result<K>;
    async fn apply_with_api(self, api: Api<K>) -> kube::Result<K>;
    async fn apply_if_not_exists(self, client: &Client) -> kube::Result<()>;
    async fn delete(self, client: &Client) -> kube::Result<()>;
    async fn exists(self, client: &Client) -> kube::Result<bool>;
}

#[async_trait]
impl<K, S> ApiExt<K> for K
where
    S: ScopeExt<K>,
    K: Resource<Scope = S> + Clone + Serialize + DeserializeOwned + Debug + Send + Sync + 'static,
    K::DynamicType: Clone + Debug + Default + Hash + Unpin + Eq,
{
    async fn apply(self, client: &Client) -> kube::Result<K> {
        let api = K::Scope::api(client.clone(), self.namespace());
        self.apply_with_api(api).await
    }

    async fn apply_with_api(self, api: Api<K>) -> kube::Result<K> {
        api.patch(
            &self.name_unchecked(),
            &PatchParams::apply(env!("CARGO_PKG_NAME")),
            &Patch::Apply(self),
        )
        .await
    }

    async fn apply_if_not_exists(self, client: &Client) -> kube::Result<()> {
        let api = K::Scope::api(client.clone(), self.namespace());
        let name = self.name_unchecked();

        if let Err(kube::Error::Api(kube::core::ErrorResponse { code: 404, .. })) =
            api.get(&name).await
        {
            self.apply_with_api(api).await?;
        }

        Ok(())
    }

    async fn delete(self, client: &Client) -> kube::Result<()> {
        let api = K::Scope::api(client.clone(), self.namespace());
        let name = self.name_unchecked();
        let params = DeleteParams::default();
        api.delete(&name, &params).await?;
        Ok(())
    }

    async fn exists(self, client: &Client) -> kube::Result<bool> {
        let name = self.name_unchecked();
        let api = K::Scope::api(client.clone(), self.namespace());

        match api.get(&name).await {
            Ok(_) => Ok(true),
            Err(kube::Error::Api(kube::core::ErrorResponse { code: 404, .. })) => Ok(false),
            err => {
                err?;
                Ok(false)
            }
        }
    }
}

/// Kubernetes operator managing multiple resource handlers
///
/// Use with the `#[kubus]` derive macro to register handlers for different resource types.
pub struct Operator<S, E>
where
    S: Clone,
{
    context: Arc<Context<S>>,
    handlers: Vec<Box<dyn DynEventHandler<E>>>,
}

impl<S, E> Operator<S, E>
where
    S: Clone + Send + Sync + 'static,
    E: StdError + Send + Sync + 'static,
{
    /// Creates a new operator
    pub fn new(context: Arc<Context<S>>) -> Self {
        Self {
            context,
            handlers: Default::default(),
        }
    }

    /// Registers an event handler for a resource type
    ///
    /// Chain multiple calls to register handlers for different resources.
    #[must_use]
    pub fn handler<H, K>(mut self, _: H) -> Self
    where
        H: EventHandler<K, S, E> + Send + Sync + 'static,
        K: Resource + Clone + DeserializeOwned + Debug + Send + Sync + 'static,
        K::DynamicType: Clone + Debug + Default + Hash + Unpin + Eq,
    {
        let wrapper = EventHandlerWrapper::<H, K, S, E>::new(self.context.clone());
        self.handlers.push(Box::new(wrapper));
        self
    }

    /// Runs the operator, starting all registered handlers
    ///
    /// Blocks until all handlers complete (typically on shutdown).
    pub async fn run(self) -> Result<()> {
        tracing::info!(
            "starting kubus operator with {} handlers",
            self.handlers.len()
        );

        let mut set = JoinSet::new();

        for handler in self.handlers {
            let client = self.context.client.clone();
            let name = handler.name();
            set.spawn(
                async move {
                    let client = client.clone();
                    if let Err(err) = handler.run(client).await {
                        tracing::error!({ err = &err as &dyn StdError }, "handler error");
                    }
                }
                .instrument(info_span!("handler", name = name)),
            );
        }

        set.join_all().await;

        Ok(())
    }
}

#[doc(hidden)]
pub struct OperatorBuilder;

impl OperatorBuilder {
    /// Attach context to operator builder
    pub fn with_context<S>(self, context: impl Into<Context<S>>) -> OperatorBuilderWithContext<S>
    where
        S: Clone + Send + Sync + 'static,
    {
        let context = Arc::new(context.into());
        OperatorBuilderWithContext { context }
    }
}

impl Operator<(), ()> {
    /// Start building an Operator
    #[must_use]
    pub const fn builder() -> OperatorBuilder {
        OperatorBuilder
    }
}

#[doc(hidden)]
pub struct OperatorBuilderWithContext<S>
where
    S: Clone + Send + Sync + 'static,
{
    context: Arc<Context<S>>,
}

impl<S> OperatorBuilderWithContext<S>
where
    S: Clone + Send + Sync + 'static,
{
    /// Registers an event handler for a resource type
    ///
    /// Chain multiple calls to register handlers for different resources.
    #[must_use]
    pub fn handler<H, K, E>(self, _: H) -> Operator<S, E>
    where
        H: EventHandler<K, S, E> + Send + Sync + 'static,
        E: StdError + Send + Sync + 'static,
        K: Resource + Clone + DeserializeOwned + Debug + Send + Sync + 'static,
        K::DynamicType: Clone + Debug + Default + Hash + Unpin + Eq,
    {
        let wrapper = EventHandlerWrapper::<H, K, S, E>::new(self.context.clone());
        let mut operator = Operator::<S, E>::new(self.context);
        operator.handlers.push(Box::new(wrapper));
        operator
    }
}

async fn patch_object<K>(api: &Api<K>, obj: Arc<K>, patch: serde_json::Value) -> kube::Result<()>
where
    K: Resource + Clone + Debug + Serialize + DeserializeOwned,
{
    api.patch::<K>(
        &obj.meta().name.clone().unwrap(),
        &PatchParams::default(),
        &Patch::Json(serde_json::from_value(patch).unwrap()),
    )
    .await?;

    Ok(())
}

pub async fn apply_finalizer<K>(api: &Api<K>, name: &str, obj: Arc<K>) -> kube::Result<()>
where
    K: Resource + Clone + Debug + Serialize + DeserializeOwned,
{
    let patch = if obj.finalizers().is_empty() {
        json!([
            { "op": "test", "path": "/metadata/finalizers", "value": null },
            { "op": "add", "path": "/metadata/finalizers", "value": [name] }
        ])
    } else {
        json!([
            { "op": "test", "path": "/metadata/finalizers", "value": obj.finalizers() },
            { "op": "add", "path": "/metadata/finalizers", "value": [name] }
        ])
    };

    patch_object(api, obj, patch).await
}

pub async fn remove_finalizer<K>(api: &Api<K>, name: &str, obj: Arc<K>) -> kube::Result<()>
where
    K: Resource + Clone + Debug + Serialize + DeserializeOwned,
{
    let Some(idx) = obj
        .finalizers()
        .iter()
        .enumerate()
        .find(|(_, n)| n == &name)
        .map(|(idx, _)| idx)
    else {
        return Ok(());
    };

    let finalizer_path = format!("/metadata/finalizers/{idx}");

    let patch = json!([
      { "op": "test", "path": finalizer_path, "value": name },
      { "op": "remove", "path": finalizer_path }
    ]);

    patch_object(api, obj, patch).await
}

/// Print list of CRD's to stdout as serialized yaml
///
/// ```rust,no_run
/// print_crds![Database, Backup];
/// ```
#[macro_export]
macro_rules! print_crds {
    [$($resource:ident),+] => {{
            let list = ::k8s_openapi::List {
                items: vec![$($resource::crd(),)+],
                ..::core::default::Default::default()
            };
            let yaml = ::serde_yaml::to_string(&list).unwrap();
            let mut stdout = ::std::io::stdout();
            stdout.write_all(yaml.as_bytes())?;
    }};
}