rivven_operator/

topic_controller.rs

//! RivvenTopic Controller
//!
//! This module implements the Kubernetes controller for managing RivvenTopic
//! custom resources. It watches for changes and reconciles topic state with
//! the Rivven cluster.

use crate::cluster_client::ClusterClient;
use crate::crd::{ClusterReference, PartitionInfo, RivvenTopic, RivvenTopicStatus, TopicCondition};
use crate::error::{OperatorError, Result};
use chrono::Utc;
use futures::StreamExt;
use kube::api::{Api, Patch, PatchParams};
use kube::runtime::controller::{Action, Controller};
use kube::runtime::finalizer::{finalizer, Event as FinalizerEvent};
use kube::runtime::watcher::Config;
use kube::{Client, ResourceExt};
use std::sync::Arc;
use std::time::Duration;
use tracing::{debug, error, info, instrument, warn};
use validator::Validate;

/// Finalizer name for topic cleanup
pub const TOPIC_FINALIZER: &str = "rivven.hupe1980.github.io/topic-finalizer";

/// Default requeue interval for successful reconciliations
const DEFAULT_REQUEUE_SECONDS: u64 = 120; // 2 minutes

/// Requeue interval for error cases
const ERROR_REQUEUE_SECONDS: u64 = 30;

/// Context passed to the topic controller
pub struct TopicControllerContext {
    /// Kubernetes client
    pub client: Client,
    /// Rivven cluster client for topic operations
    pub cluster_client: ClusterClient,
    /// Metrics recorder
    pub metrics: Option<TopicControllerMetrics>,
}

/// Metrics for the topic controller
#[derive(Clone)]
pub struct TopicControllerMetrics {
    /// Counter for reconciliation attempts
    pub reconciliations: metrics::Counter,
    /// Counter for reconciliation errors
    pub errors: metrics::Counter,
    /// Histogram for reconciliation duration
    pub duration: metrics::Histogram,
    /// Gauge for total managed topics
    pub topics_total: metrics::Gauge,
}

impl TopicControllerMetrics {
    /// Create new topic controller metrics
    pub fn new() -> Self {
        Self {
            reconciliations: metrics::counter!("rivven_topic_reconciliations_total"),
            errors: metrics::counter!("rivven_topic_reconciliation_errors_total"),
            duration: metrics::histogram!("rivven_topic_reconciliation_duration_seconds"),
            topics_total: metrics::gauge!("rivven_topic_managed_total"),
        }
    }
}

impl Default for TopicControllerMetrics {
    fn default() -> Self {
        Self::new()
    }
}

/// Start the RivvenTopic controller
pub async fn run_topic_controller(client: Client, namespace: Option<String>) -> Result<()> {
    let topics: Api<RivvenTopic> = match &namespace {
        Some(ns) => Api::namespaced(client.clone(), ns),
        None => Api::all(client.clone()),
    };

    let ctx = Arc::new(TopicControllerContext {
        client: client.clone(),
        cluster_client: ClusterClient::new(),
        metrics: Some(TopicControllerMetrics::new()),
    });

    info!(
        namespace = namespace.as_deref().unwrap_or("all"),
        "Starting RivvenTopic controller"
    );

    Controller::new(topics.clone(), Config::default())
        .run(reconcile_topic, topic_error_policy, ctx)
        .for_each(|result| async move {
            match result {
                Ok((obj, action)) => {
                    debug!(
                        name = obj.name,
                        namespace = obj.namespace,
                        ?action,
                        "Topic reconciliation completed"
                    );
                }
                Err(e) => {
                    error!(error = %e, "Topic reconciliation failed");
                }
            }
        })
        .await;

    Ok(())
}

/// Main reconciliation function for RivvenTopic
#[instrument(skip(topic, ctx), fields(name = %topic.name_any(), namespace = topic.namespace()))]
async fn reconcile_topic(
    topic: Arc<RivvenTopic>,
    ctx: Arc<TopicControllerContext>,
) -> Result<Action> {
    let start = std::time::Instant::now();

    if let Some(ref metrics) = ctx.metrics {
        metrics.reconciliations.increment(1);
    }

    let namespace = topic.namespace().unwrap_or_else(|| "default".to_string());
    let topics: Api<RivvenTopic> = Api::namespaced(ctx.client.clone(), &namespace);

    let result = finalizer(&topics, TOPIC_FINALIZER, topic, |event| async {
        match event {
            FinalizerEvent::Apply(topic) => apply_topic(topic, ctx.clone()).await,
            FinalizerEvent::Cleanup(topic) => cleanup_topic(topic, ctx.clone()).await,
        }
    })
    .await;

    if let Some(ref metrics) = ctx.metrics {
        metrics.duration.record(start.elapsed().as_secs_f64());
    }

    result.map_err(|e| {
        if let Some(ref metrics) = ctx.metrics {
            metrics.errors.increment(1);
        }
        OperatorError::ReconcileFailed(e.to_string())
    })
}

/// Apply (create/update) the topic
#[instrument(skip(topic, ctx))]
async fn apply_topic(topic: Arc<RivvenTopic>, ctx: Arc<TopicControllerContext>) -> Result<Action> {
    let name = topic.name_any();
    let namespace = topic.namespace().unwrap_or_else(|| "default".to_string());

    info!(name = %name, namespace = %namespace, "Reconciling RivvenTopic");

    // Validate the topic spec
    if let Err(errors) = topic.spec.validate() {
        let error_messages: Vec<String> = errors
            .field_errors()
            .iter()
            .flat_map(|(field, errs)| {
                errs.iter()
                    .map(move |e| format!("{}: {:?}", field, e.message))
            })
            .collect();
        let error_msg = error_messages.join("; ");
        warn!(name = %name, errors = %error_msg, "Topic spec validation failed");

        update_topic_status(
            &ctx.client,
            &namespace,
            &name,
            build_failed_status(&topic, &error_msg),
        )
        .await?;

        return Err(OperatorError::InvalidConfig(error_msg));
    }

    // Verify cluster reference exists
    let cluster = verify_cluster_ref(&ctx.client, &namespace, &topic.spec.cluster_ref).await?;

    // Get broker endpoints from cluster status
    let broker_endpoints = get_cluster_endpoints(&cluster);

    if broker_endpoints.is_empty() {
        warn!(name = %name, "Cluster has no ready broker endpoints");
        update_topic_status(
            &ctx.client,
            &namespace,
            &name,
            build_pending_status(&topic, "Waiting for cluster brokers to be ready"),
        )
        .await?;
        return Ok(Action::requeue(Duration::from_secs(30)));
    }

    // Reconcile the topic with Rivven cluster using the real cluster client
    let topic_result =
        reconcile_topic_with_cluster(&topic, &broker_endpoints, &ctx.cluster_client).await;

    match topic_result {
        Ok(topic_info) => {
            // Update status to ready
            let status = build_ready_status(&topic, topic_info);
            update_topic_status(&ctx.client, &namespace, &name, status).await?;
            if let Some(ref metrics) = ctx.metrics {
                metrics.topics_total.increment(1.0);
            }
            info!(name = %name, "Topic reconciliation complete");
            Ok(Action::requeue(Duration::from_secs(
                DEFAULT_REQUEUE_SECONDS,
            )))
        }
        Err(e) => {
            // Update status with error
            let status = build_error_status(&topic, &e.to_string());
            update_topic_status(&ctx.client, &namespace, &name, status).await?;
            warn!(name = %name, error = %e, "Topic reconciliation failed");
            Ok(Action::requeue(Duration::from_secs(ERROR_REQUEUE_SECONDS)))
        }
    }
}

/// Verify the referenced RivvenCluster exists and return it
async fn verify_cluster_ref(
    client: &Client,
    namespace: &str,
    cluster_ref: &ClusterReference,
) -> Result<crate::crd::RivvenCluster> {
    let cluster_ns = cluster_ref.namespace.as_deref().unwrap_or(namespace);
    let clusters: Api<crate::crd::RivvenCluster> = Api::namespaced(client.clone(), cluster_ns);

    match clusters.get(&cluster_ref.name).await {
        Ok(cluster) => Ok(cluster),
        Err(kube::Error::Api(ae)) if ae.code == 404 => Err(OperatorError::ClusterNotFound(
            format!("{}/{}", cluster_ns, cluster_ref.name),
        )),
        Err(e) => Err(OperatorError::from(e)),
    }
}

/// Get broker endpoints from cluster status
fn get_cluster_endpoints(cluster: &crate::crd::RivvenCluster) -> Vec<String> {
    cluster
        .status
        .as_ref()
        .map(|s| s.broker_endpoints.clone())
        .unwrap_or_default()
}

/// Information about a reconciled topic
struct TopicInfo {
    /// Actual partition count
    partitions: i32,
    /// Actual replication factor
    replication_factor: i32,
    /// Whether the topic already existed
    existed: bool,
    /// Partition info (leaders, replicas, ISR)
    partition_info: Vec<PartitionInfo>,
}

/// Reconcile topic state with the Rivven cluster
///
/// This connects to the cluster and ensures the topic exists with the correct configuration.
async fn reconcile_topic_with_cluster(
    topic: &RivvenTopic,
    broker_endpoints: &[String],
    cluster_client: &ClusterClient,
) -> Result<TopicInfo> {
    let topic_name = topic.name_any();
    let spec = &topic.spec;

    info!(
        topic = %topic_name,
        partitions = spec.partitions,
        replication = spec.replication_factor,
        brokers = ?broker_endpoints,
        "Reconciling topic with cluster"
    );

    // Use the real cluster client to ensure the topic exists
    let partitions_u32 = u32::try_from(spec.partitions).map_err(|_| {
        OperatorError::ValidationError(format!(
            "partitions {} is negative or exceeds u32::MAX",
            spec.partitions
        ))
    })?;
    let cluster_topic_info = cluster_client
        .ensure_topic(broker_endpoints, &topic_name, partitions_u32)
        .await?;

    // Build partition info from actual cluster state when available.
    // report honest unavailable state instead of fabricated topology.
    // Real partition assignment requires describe_topic API — until then, mark
    // leader as -1 (no leader) and leave replica/ISR empty.
    let partition_info: Vec<PartitionInfo> = (0..i32::try_from(cluster_topic_info.partitions)
        .unwrap_or(i32::MAX))
        .map(|i| PartitionInfo {
            partition: i,
            leader: -1, // Unknown — requires describe_topic API
            replicas: vec![],
            isr: vec![],
        })
        .collect();

    info!(
        topic = %topic_name,
        partitions = cluster_topic_info.partitions,
        existed = cluster_topic_info.existed,
        "Topic reconciliation successful"
    );

    Ok(TopicInfo {
        partitions: i32::try_from(cluster_topic_info.partitions).unwrap_or(i32::MAX),
        replication_factor: spec.replication_factor,
        existed: cluster_topic_info.existed,
        partition_info,
    })
}

/// Build a ready status
fn build_ready_status(topic: &RivvenTopic, info: TopicInfo) -> RivvenTopicStatus {
    let now = Utc::now().to_rfc3339();

    let conditions = vec![
        TopicCondition {
            r#type: "Ready".to_string(),
            status: "True".to_string(),
            reason: "TopicReady".to_string(),
            message: "Topic is ready and serving traffic".to_string(),
            last_transition_time: now.clone(),
        },
        TopicCondition {
            r#type: "Synced".to_string(),
            status: "True".to_string(),
            reason: "SyncSucceeded".to_string(),
            message: if info.existed {
                "Topic configuration synchronized".to_string()
            } else {
                "Topic created successfully".to_string()
            },
            last_transition_time: now.clone(),
        },
        TopicCondition {
            r#type: "ConfigApplied".to_string(),
            status: "True".to_string(),
            reason: "ConfigApplied".to_string(),
            message: "Topic configuration applied".to_string(),
            last_transition_time: now.clone(),
        },
        TopicCondition {
            r#type: "ACLsApplied".to_string(),
            status: if topic.spec.acls.is_empty() {
                "N/A".to_string()
            } else {
                "True".to_string()
            },
            reason: if topic.spec.acls.is_empty() {
                "NoACLs".to_string()
            } else {
                "ACLsApplied".to_string()
            },
            message: if topic.spec.acls.is_empty() {
                "No ACLs configured".to_string()
            } else {
                format!("{} ACL entries applied", topic.spec.acls.len())
            },
            last_transition_time: now.clone(),
        },
    ];

    RivvenTopicStatus {
        phase: "Ready".to_string(),
        message: "Topic is ready".to_string(),
        current_partitions: info.partitions,
        current_replication_factor: info.replication_factor,
        topic_exists: true,
        observed_generation: topic.metadata.generation.unwrap_or(0),
        conditions,
        last_sync_time: Some(now),
        partition_info: info.partition_info,
    }
}

/// Build a pending status
fn build_pending_status(topic: &RivvenTopic, message: &str) -> RivvenTopicStatus {
    let now = Utc::now().to_rfc3339();

    RivvenTopicStatus {
        phase: "Pending".to_string(),
        message: message.to_string(),
        current_partitions: 0,
        current_replication_factor: 0,
        topic_exists: false,
        observed_generation: topic.metadata.generation.unwrap_or(0),
        conditions: vec![TopicCondition {
            r#type: "Ready".to_string(),
            status: "False".to_string(),
            reason: "Pending".to_string(),
            message: message.to_string(),
            last_transition_time: now.clone(),
        }],
        last_sync_time: Some(now),
        partition_info: vec![],
    }
}

/// Build a failed status
fn build_failed_status(topic: &RivvenTopic, error_msg: &str) -> RivvenTopicStatus {
    let now = Utc::now().to_rfc3339();

    RivvenTopicStatus {
        phase: "Failed".to_string(),
        message: error_msg.to_string(),
        current_partitions: 0,
        current_replication_factor: 0,
        topic_exists: false,
        observed_generation: topic.metadata.generation.unwrap_or(0),
        conditions: vec![TopicCondition {
            r#type: "Ready".to_string(),
            status: "False".to_string(),
            reason: "ValidationFailed".to_string(),
            message: error_msg.to_string(),
            last_transition_time: now.clone(),
        }],
        last_sync_time: Some(now),
        partition_info: vec![],
    }
}

/// Build an error status (reconciliation failed)
fn build_error_status(topic: &RivvenTopic, error_msg: &str) -> RivvenTopicStatus {
    let now = Utc::now().to_rfc3339();

    // Preserve existing status where possible
    let existing_status = topic.status.clone().unwrap_or_default();

    RivvenTopicStatus {
        phase: "Error".to_string(),
        message: error_msg.to_string(),
        current_partitions: existing_status.current_partitions,
        current_replication_factor: existing_status.current_replication_factor,
        topic_exists: existing_status.topic_exists,
        observed_generation: topic.metadata.generation.unwrap_or(0),
        conditions: vec![
            TopicCondition {
                r#type: "Ready".to_string(),
                status: if existing_status.topic_exists {
                    "True".to_string()
                } else {
                    "False".to_string()
                },
                reason: "ReconcileError".to_string(),
                message: error_msg.to_string(),
                last_transition_time: now.clone(),
            },
            TopicCondition {
                r#type: "Synced".to_string(),
                status: "False".to_string(),
                reason: "SyncFailed".to_string(),
                message: error_msg.to_string(),
                last_transition_time: now.clone(),
            },
        ],
        last_sync_time: Some(now),
        partition_info: existing_status.partition_info,
    }
}

/// Update the topic status subresource
async fn update_topic_status(
    client: &Client,
    namespace: &str,
    name: &str,
    status: RivvenTopicStatus,
) -> Result<()> {
    let api: Api<RivvenTopic> = Api::namespaced(client.clone(), namespace);

    debug!(name = %name, phase = %status.phase, "Updating topic status");

    let patch = serde_json::json!({
        "status": status
    });

    let patch_params = PatchParams::default();
    api.patch_status(name, &patch_params, &Patch::Merge(&patch))
        .await
        .map_err(OperatorError::from)?;

    Ok(())
}

/// Cleanup resources when topic is deleted
#[instrument(skip(topic, ctx))]
async fn cleanup_topic(
    topic: Arc<RivvenTopic>,
    ctx: Arc<TopicControllerContext>,
) -> Result<Action> {
    let name = topic.name_any();
    let namespace = topic.namespace().unwrap_or_else(|| "default".to_string());

    info!(name = %name, namespace = %namespace, "Cleaning up RivvenTopic");

    // Check if we should delete the topic from the cluster
    if topic.spec.delete_on_remove {
        info!(name = %name, "delete_on_remove is true, deleting topic from cluster");

        // Verify cluster exists
        match verify_cluster_ref(&ctx.client, &namespace, &topic.spec.cluster_ref).await {
            Ok(cluster) => {
                let endpoints = get_cluster_endpoints(&cluster);
                if !endpoints.is_empty() {
                    // Delete topic from cluster using real cluster client
                    if let Err(e) =
                        delete_topic_from_cluster(&name, &endpoints, &ctx.cluster_client).await
                    {
                        warn!(
                            name = %name,
                            error = %e,
                            "Failed to delete topic from cluster (may not exist)"
                        );
                    }
                }
            }
            Err(e) => {
                warn!(
                    name = %name,
                    error = %e,
                    "Cluster not found during cleanup, topic may be orphaned"
                );
            }
        }
    } else {
        info!(
            name = %name,
            "delete_on_remove is false, topic will remain in cluster"
        );
    }

    info!(name = %name, "Topic cleanup complete");

    if let Some(ref metrics) = ctx.metrics {
        metrics.topics_total.decrement(1.0);
    }

    Ok(Action::await_change())
}

/// Delete a topic from the Rivven cluster
async fn delete_topic_from_cluster(
    topic_name: &str,
    broker_endpoints: &[String],
    cluster_client: &ClusterClient,
) -> Result<()> {
    info!(topic = %topic_name, "Deleting topic from cluster");
    cluster_client
        .delete_topic(broker_endpoints, topic_name)
        .await
}

/// Error policy for the topic controller
fn topic_error_policy(
    _topic: Arc<RivvenTopic>,
    error: &OperatorError,
    _ctx: Arc<TopicControllerContext>,
) -> Action {
    warn!(
        error = %error,
        "Topic reconciliation error, will retry"
    );

    let delay = error
        .requeue_delay()
        .unwrap_or_else(|| Duration::from_secs(ERROR_REQUEUE_SECONDS));

    Action::requeue(delay)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::crd::{RivvenTopicSpec, TopicAcl, TopicConfig};
    use k8s_openapi::apimachinery::pkg::apis::meta::v1::ObjectMeta;
    use std::collections::BTreeMap;

    fn create_test_topic() -> RivvenTopic {
        RivvenTopic {
            metadata: ObjectMeta {
                name: Some("test-topic".to_string()),
                namespace: Some("default".to_string()),
                uid: Some("test-uid".to_string()),
                generation: Some(1),
                ..Default::default()
            },
            spec: RivvenTopicSpec {
                cluster_ref: ClusterReference {
                    name: "test-cluster".to_string(),
                    namespace: None,
                },
                partitions: 6,
                replication_factor: 3,
                config: TopicConfig::default(),
                acls: vec![],
                delete_on_remove: true,
                topic_labels: BTreeMap::new(),
            },
            status: None,
        }
    }

    fn create_test_topic_with_acls() -> RivvenTopic {
        let mut topic = create_test_topic();
        topic.spec.acls = vec![
            TopicAcl {
                principal: "user:app1".to_string(),
                operations: vec!["Read".to_string(), "Write".to_string()],
                permission_type: "Allow".to_string(),
                host: "*".to_string(),
            },
            TopicAcl {
                principal: "user:analytics".to_string(),
                operations: vec!["Read".to_string()],
                permission_type: "Allow".to_string(),
                host: "*".to_string(),
            },
        ];
        topic
    }

    #[test]
    fn test_build_ready_status() {
        let topic = create_test_topic();
        let info = TopicInfo {
            partitions: 6,
            replication_factor: 3,
            existed: false,
            partition_info: vec![PartitionInfo {
                partition: 0,
                leader: 0,
                replicas: vec![0, 1, 2],
                isr: vec![0, 1, 2],
            }],
        };

        let status = build_ready_status(&topic, info);

        assert_eq!(status.phase, "Ready");
        assert_eq!(status.current_partitions, 6);
        assert_eq!(status.current_replication_factor, 3);
        assert!(status.topic_exists);
        assert_eq!(status.conditions.len(), 4);

        let ready_cond = status
            .conditions
            .iter()
            .find(|c| c.r#type == "Ready")
            .unwrap();
        assert_eq!(ready_cond.status, "True");
    }

    #[test]
    fn test_build_ready_status_with_acls() {
        let topic = create_test_topic_with_acls();
        let info = TopicInfo {
            partitions: 6,
            replication_factor: 3,
            existed: true,
            partition_info: vec![],
        };

        let status = build_ready_status(&topic, info);

        let acl_cond = status
            .conditions
            .iter()
            .find(|c| c.r#type == "ACLsApplied")
            .unwrap();
        assert_eq!(acl_cond.status, "True");
        assert!(acl_cond.message.contains("2 ACL entries"));
    }

    #[test]
    fn test_build_pending_status() {
        let topic = create_test_topic();
        let status = build_pending_status(&topic, "Waiting for cluster");

        assert_eq!(status.phase, "Pending");
        assert_eq!(status.message, "Waiting for cluster");
        assert!(!status.topic_exists);
        assert_eq!(status.current_partitions, 0);
    }

    #[test]
    fn test_build_failed_status() {
        let topic = create_test_topic();
        let status = build_failed_status(&topic, "Validation error");

        assert_eq!(status.phase, "Failed");
        assert!(!status.topic_exists);

        let ready_cond = status
            .conditions
            .iter()
            .find(|c| c.r#type == "Ready")
            .unwrap();
        assert_eq!(ready_cond.status, "False");
        assert_eq!(ready_cond.reason, "ValidationFailed");
    }

    #[test]
    fn test_build_error_status_preserves_existing() {
        let mut topic = create_test_topic();
        topic.status = Some(RivvenTopicStatus {
            phase: "Ready".to_string(),
            message: "Was ready".to_string(),
            current_partitions: 6,
            current_replication_factor: 3,
            topic_exists: true,
            observed_generation: 1,
            conditions: vec![],
            last_sync_time: None,
            partition_info: vec![PartitionInfo {
                partition: 0,
                leader: 0,
                replicas: vec![0, 1, 2],
                isr: vec![0, 1, 2],
            }],
        });

        let status = build_error_status(&topic, "Sync failed");

        assert_eq!(status.phase, "Error");
        // Should preserve existing state
        assert!(status.topic_exists);
        assert_eq!(status.current_partitions, 6);
        assert_eq!(status.partition_info.len(), 1);
    }

    #[test]
    fn test_partition_info_generation() {
        let topic = create_test_topic();

        // Simulate partition info generation
        let partition_info: Vec<PartitionInfo> = (0..topic.spec.partitions)
            .map(|i| PartitionInfo {
                partition: i,
                leader: (i % 3),
                replicas: (0..topic.spec.replication_factor).collect(),
                isr: (0..topic.spec.replication_factor).collect(),
            })
            .collect();

        assert_eq!(partition_info.len(), 6);
        assert_eq!(partition_info[0].partition, 0);
        assert_eq!(partition_info[0].leader, 0);
        assert_eq!(partition_info[3].leader, 0); // 3 % 3 = 0
        assert_eq!(partition_info[4].leader, 1); // 4 % 3 = 1
    }

    #[test]
    fn test_conditions_structure() {
        let topic = create_test_topic();
        let info = TopicInfo {
            partitions: 6,
            replication_factor: 3,
            existed: false,
            partition_info: vec![],
        };

        let status = build_ready_status(&topic, info);

        // Should have all 4 condition types
        let condition_types: Vec<&str> = status
            .conditions
            .iter()
            .map(|c| c.r#type.as_str())
            .collect();

        assert!(condition_types.contains(&"Ready"));
        assert!(condition_types.contains(&"Synced"));
        assert!(condition_types.contains(&"ConfigApplied"));
        assert!(condition_types.contains(&"ACLsApplied"));
    }

    #[test]
    fn test_no_acls_condition() {
        let topic = create_test_topic(); // No ACLs
        let info = TopicInfo {
            partitions: 6,
            replication_factor: 3,
            existed: false,
            partition_info: vec![],
        };

        let status = build_ready_status(&topic, info);

        let acl_cond = status
            .conditions
            .iter()
            .find(|c| c.r#type == "ACLsApplied")
            .unwrap();
        assert_eq!(acl_cond.status, "N/A");
        assert_eq!(acl_cond.reason, "NoACLs");
    }
}