crabka-operator 0.3.1

//! `SchemaRegistry` reconciler. Renders a stateless Deployment + headless
//! Service + `ClusterIP` Service for the `crabka-schema-registry` binary,
//! associated with a managed `Kafka` via the `crabka.io/cluster` label.

use std::collections::BTreeMap;
use std::sync::Arc;
use std::time::Duration;

use futures::StreamExt as _;
use k8s_openapi::api::apps::v1::Deployment;
use k8s_openapi::api::core::v1::Service;
use kube::api::Api;
use kube::runtime::controller::{Action, Controller};
use kube::runtime::reflector::ObjectRef;
use kube::runtime::watcher;
use kube::{Resource, ResourceExt as _};
use serde_json::json;

use crate::context::Context;
use crate::controller::common::{ReconcileError, apply_object, condition, owner_ref, patch_status};
use crate::controller::topic::internal_listener_bootstrap;
use crate::crd::{Kafka, SchemaRegistry, SchemaRegistryStatus, TlsClientAuth};

const APP_NAME: &str = "crabka-schema-registry";
const SR_PORT: i32 = 8081;
const DEFAULT_IMAGE: &str = concat!(
    "ghcr.io/robot-head/crabka-schema-registry:",
    env!("CARGO_PKG_VERSION")
);

pub async fn run(ctx: Context) -> anyhow::Result<()> {
    let sr_api: Api<SchemaRegistry> = Api::all(ctx.client.clone());
    let kafka_api: Api<Kafka> = Api::all(ctx.client.clone());
    Controller::new(sr_api, watcher::Config::default())
        .watches(kafka_api, watcher::Config::default(), |_kafka| {
            Vec::<ObjectRef<SchemaRegistry>>::new().into_iter()
        })
        .run(reconcile, error_policy, Arc::new(ctx))
        .for_each(|res| async move {
            match res {
                Ok((obj, _)) => tracing::debug!(?obj, "schemaregistry reconciled"),
                Err(e) => tracing::warn!(error = %e, "schemaregistry reconcile error"),
            }
        })
        .await;
    Ok(())
}

pub fn error_policy(_obj: Arc<SchemaRegistry>, err: &ReconcileError, _ctx: Arc<Context>) -> Action {
    tracing::warn!(error = %err, "schemaregistry reconcile error, requeueing");
    Action::requeue(Duration::from_secs(15))
}

pub async fn reconcile(
    obj: Arc<SchemaRegistry>,
    ctx: Arc<Context>,
) -> Result<Action, ReconcileError> {
    let ns = obj.namespace().unwrap_or_else(|| "default".into());
    let name = obj.name_any();
    let sr_api: Api<SchemaRegistry> = Api::namespaced(ctx.client.clone(), &ns);

    // 1. Cluster label (unless an explicit bootstrap override is set)
    let cluster = obj
        .meta()
        .labels
        .as_ref()
        .and_then(|l| l.get("crabka.io/cluster").cloned());

    // 2. Resolve bootstrap: spec override wins, else derive from the Kafka.
    let bootstrap = if let Some(b) = obj.spec.bootstrap_servers.clone() {
        Some(b)
    } else {
        let Some(cluster) = cluster.clone() else {
            set_status(
                &sr_api,
                &name,
                &obj,
                "MissingClusterLabel",
                "set metadata.labels[\"crabka.io/cluster\"] or spec.bootstrapServers",
                None,
                None,
            )
            .await?;
            return Ok(Action::requeue(Duration::from_mins(1)));
        };
        let kafka_api: Api<Kafka> = Api::namespaced(ctx.client.clone(), &ns);
        let kafka = kafka_api.get_opt(&cluster).await?;
        kafka.as_ref().and_then(internal_listener_bootstrap)
    };
    let Some(bootstrap) = bootstrap else {
        set_status(
            &sr_api,
            &name,
            &obj,
            "KafkaNotReady",
            "referenced Kafka is not Ready or has no internal listener",
            None,
            None,
        )
        .await?;
        return Ok(Action::requeue(Duration::from_secs(30)));
    };

    // 3. Render + apply children (Deployment + 2 Services).
    let svc_api: Api<Service> = Api::namespaced(ctx.client.clone(), &ns);
    let dep_api: Api<Deployment> = Api::namespaced(ctx.client.clone(), &ns);

    let headless = render_headless_service(&obj)?;
    apply_object(&svc_api, &headless_name(&name), &headless).await?;
    let clusterip = render_clusterip_service(&obj)?;
    apply_object(&svc_api, &service_name(&name), &clusterip).await?;
    let image = obj
        .spec
        .image
        .clone()
        .or_else(|| ctx.config.default_schema_registry_image.clone())
        .unwrap_or_else(|| DEFAULT_IMAGE.to_string());
    let deployment = render_deployment(&obj, &bootstrap, &image)?;
    apply_object(&dep_api, &deployment_name(&name), &deployment).await?;

    // 4. Status from the live Deployment.
    let live = dep_api.get_opt(&deployment_name(&name)).await?;
    let (replicas, ready) = live
        .as_ref()
        .and_then(|d| d.status.as_ref())
        .map_or((None, None), |s| (s.replicas, s.ready_replicas));
    let desired = obj.spec.replicas;
    let url = format!(
        "{}://{}.{}.svc.cluster.local:{SR_PORT}",
        scheme(&obj),
        service_name(&name),
        ns
    );
    if ready.unwrap_or(0) >= desired {
        set_status(
            &sr_api,
            &name,
            &obj,
            "Available",
            &format!("{desired} replica(s) ready"),
            Some((replicas, ready)),
            Some(url),
        )
        .await?;
    } else {
        set_status(
            &sr_api,
            &name,
            &obj,
            "Progressing",
            &format!("{}/{desired} replica(s) ready", ready.unwrap_or(0)),
            Some((replicas, ready)),
            Some(url),
        )
        .await?;
    }
    Ok(Action::requeue(Duration::from_mins(1)))
}

fn deployment_name(n: &str) -> String {
    format!("{n}-sr")
}
fn service_name(n: &str) -> String {
    format!("{n}-sr")
}
fn headless_name(n: &str) -> String {
    format!("{n}-sr-headless")
}
fn scheme(obj: &SchemaRegistry) -> &'static str {
    if obj.spec.tls.is_some() {
        "https"
    } else {
        "http"
    }
}

/// Stable label set used for Deployment `selector.matchLabels`, the pod
/// template labels, and BOTH Services' `spec.selector`. Deployment
/// selectors are immutable, so this map must NOT carry the version label
/// (a churning value there would make the Deployment un-updatable, and a
/// selector/template mismatch would mean pods never become Ready).
fn selector_labels(obj: &SchemaRegistry) -> BTreeMap<String, String> {
    let instance = obj.name_any();
    let mut m = BTreeMap::new();
    m.insert("app.kubernetes.io/name".into(), APP_NAME.into());
    m.insert("app.kubernetes.io/instance".into(), instance);
    m.insert(
        "app.kubernetes.io/component".into(),
        "schema-registry".into(),
    );
    m
}

/// Metadata labels for the rendered objects: the stable selector labels
/// plus the version / managed-by metadata labels. Used only for
/// `metadata.labels`, never for selectors.
fn meta_labels(obj: &SchemaRegistry) -> BTreeMap<String, String> {
    let mut m = selector_labels(obj);
    m.insert("app.kubernetes.io/version".into(), "0.1.1".into());
    m.insert(
        "app.kubernetes.io/managed-by".into(),
        "crabka-operator".into(),
    );
    m
}

fn render_headless_service(obj: &SchemaRegistry) -> Result<Service, ReconcileError> {
    let name = obj.name_any();
    let svc = serde_json::from_value(json!({
        "metadata": {
            "name": headless_name(&name),
            "namespace": obj.meta().namespace.clone(),
            "labels": meta_labels(obj),
            "ownerReferences": [owner_ref::<SchemaRegistry>(obj)?],
        },
        "spec": {
            "clusterIP": "None",
            "selector": selector_labels(obj),
            "ports": [{ "name": "rest", "port": SR_PORT, "protocol": "TCP", "targetPort": SR_PORT }],
        }
    }))?;
    Ok(svc)
}

fn render_clusterip_service(obj: &SchemaRegistry) -> Result<Service, ReconcileError> {
    let name = obj.name_any();
    let svc = serde_json::from_value(json!({
        "metadata": {
            "name": service_name(&name),
            "namespace": obj.meta().namespace.clone(),
            "labels": meta_labels(obj),
            "ownerReferences": [owner_ref::<SchemaRegistry>(obj)?],
        },
        "spec": {
            "type": "ClusterIP",
            "selector": selector_labels(obj),
            "ports": [{ "name": "rest", "port": SR_PORT, "protocol": "TCP", "targetPort": SR_PORT }],
        }
    }))?;
    Ok(svc)
}

fn render_deployment(
    obj: &SchemaRegistry,
    bootstrap: &str,
    image: &str,
) -> Result<Deployment, ReconcileError> {
    let name = obj.name_any();
    let ns = obj
        .meta()
        .namespace
        .clone()
        .unwrap_or_else(|| "default".into());
    let selector = selector_labels(obj);
    let (args, volumes, mounts) = build_args_and_mounts(obj, bootstrap);
    let advertised = format!(
        "{}://$(POD_NAME).{}.{}.svc.cluster.local:{SR_PORT}",
        scheme(obj),
        headless_name(&name),
        ns
    );
    let dep = serde_json::from_value(json!({
        "metadata": {
            "name": deployment_name(&name),
            "namespace": obj.meta().namespace.clone(),
            "labels": meta_labels(obj),
            "ownerReferences": [owner_ref::<SchemaRegistry>(obj)?],
        },
        "spec": {
            "replicas": obj.spec.replicas,
            "selector": { "matchLabels": selector },
            "template": {
                "metadata": { "labels": selector },
                "spec": {
                    "securityContext": { "runAsNonRoot": true, "runAsUser": 65532, "fsGroup": 65532 },
                    "volumes": volumes,
                    "containers": [{
                        "name": "schema-registry",
                        "image": image,
                        "args": args,
                        "env": [
                            { "name": "POD_NAME", "valueFrom": { "fieldRef": { "fieldPath": "metadata.name" } } },
                            { "name": "SCHEMA_REGISTRY_ADVERTISED_URL", "value": advertised },
                        ],
                        "ports": [{ "name": "rest", "containerPort": SR_PORT, "protocol": "TCP" }],
                        "volumeMounts": mounts,
                        "readinessProbe": { "tcpSocket": { "port": SR_PORT }, "initialDelaySeconds": 2, "periodSeconds": 5 },
                        "livenessProbe": { "tcpSocket": { "port": SR_PORT }, "initialDelaySeconds": 5, "periodSeconds": 10 },
                        "resources": obj.spec.resources.clone().unwrap_or_default(),
                    }],
                }
            }
        }
    }))?;
    Ok(dep)
}

/// Build the container args + the Secret volumes/mounts from the spec.
/// Non-secret config → args; credentials → mounted Secret files referenced
/// by path args. Returns (args, volumes, volumeMounts) as JSON values.
fn build_args_and_mounts(
    obj: &SchemaRegistry,
    bootstrap: &str,
) -> (Vec<String>, Vec<serde_json::Value>, Vec<serde_json::Value>) {
    let s = &obj.spec;
    // The SR binary has no subcommand — args are flags only. (apko's
    // `cmd: run` default is replaced by the container `args` set here.)
    let mut a: Vec<String> = Vec::new();
    a.push(format!("--bootstrap-servers={bootstrap}"));
    a.push(format!("--listen-addr=0.0.0.0:{SR_PORT}"));
    if let Some(t) = &s.schemas_topic {
        a.push(format!("--schemas-topic={t}"));
    }
    if let Some(rf) = s.schemas_topic_replication_factor {
        a.push(format!("--schemas-topic-rf={rf}"));
    }
    if let Some(g) = &s.group_id {
        a.push(format!("--group-id={g}"));
    }

    let mut volumes = Vec::new();
    let mut mounts = Vec::new();

    // Server TLS
    if let Some(tls) = &s.tls {
        a.push("--tls-cert=/etc/sr/tls/tls.crt".into());
        a.push("--tls-key=/etc/sr/tls/tls.key".into());
        volumes.push(json!({ "name": "tls", "secret": { "secretName": tls.secret_name } }));
        mounts.push(json!({ "name": "tls", "mountPath": "/etc/sr/tls", "readOnly": true }));
        let mode = match tls.client_auth.unwrap_or(TlsClientAuth::Disabled) {
            TlsClientAuth::Disabled => "disabled",
            TlsClientAuth::Optional => "optional",
            TlsClientAuth::Required => "required",
        };
        a.push(format!("--tls-client-auth={mode}"));
        if let Some(ca) = &tls.client_ca_secret_name {
            a.push("--tls-client-ca=/etc/sr/client-ca/ca.crt".into());
            volumes.push(json!({ "name": "client-ca", "secret": { "secretName": ca } }));
            mounts.push(
                json!({ "name": "client-ca", "mountPath": "/etc/sr/client-ca", "readOnly": true }),
            );
        }
    }

    // Authentication
    if let Some(authn) = &s.authentication {
        if authn.require_auth {
            a.push("--require-auth".into());
        }
        if let Some(r) = &authn.realm {
            a.push(format!("--realm={r}"));
        }
        if let Some(b) = &authn.basic {
            let key = b.users_secret_key.clone().unwrap_or_else(|| "users".into());
            a.push("--basic-auth-file=/etc/sr/basic/users".into());
            volumes.push(json!({ "name": "basic", "secret": {
                "secretName": b.users_secret_name,
                "items": [{ "key": key, "path": "users" }]
            }}));
            mounts.push(json!({ "name": "basic", "mountPath": "/etc/sr/basic", "readOnly": true }));
        }
        if authn.bearer.is_some() {
            a.push("--bearer=unsecured".into());
            if let Some(pc) = authn
                .bearer
                .as_ref()
                .and_then(|b| b.principal_claim.clone())
            {
                a.push(format!("--bearer-principal-claim={pc}"));
            }
        }
    }

    // Authorization
    if let Some(az) = &s.authorization {
        if az.enabled {
            a.push("--authz".into());
        }
        for u in &az.super_users {
            a.push(format!("--super-user={u}"));
        }
        if let Some(r) = az.acl_refresh_seconds {
            a.push(format!("--acl-refresh-secs={r}"));
        }
    }

    (a, volumes, mounts)
}

/// Patch status with a single rolled-up `Ready` condition + a `KafkaReady`
/// condition. `reason == "Available"` ⇒ Ready=True.
async fn set_status(
    api: &Api<SchemaRegistry>,
    name: &str,
    obj: &SchemaRegistry,
    reason: &str,
    message: &str,
    counts: Option<(Option<i32>, Option<i32>)>,
    url: Option<String>,
) -> Result<(), ReconcileError> {
    let kafka_ok = !matches!(reason, "MissingClusterLabel" | "KafkaNotReady");
    let ready = if reason == "Available" {
        "True"
    } else {
        "False"
    };
    let (replicas, ready_replicas) = counts.unwrap_or((None, None));
    let observed_generation = if ready == "True" {
        obj.meta().generation
    } else {
        obj.status.as_ref().and_then(|s| s.observed_generation)
    };
    let status = SchemaRegistryStatus {
        conditions: vec![
            condition(
                "KafkaReady",
                if kafka_ok { "True" } else { "False" },
                reason,
                message,
            ),
            condition("Ready", ready, reason, message),
        ],
        observed_generation,
        replicas,
        ready_replicas,
        url,
    };
    patch_status(api, name, status).await?;
    Ok(())
}