loa_core/core/

config.rs

//! Config Actor - Manages agent configuration via HTTP polling
//!
//! This actor:
//! - Fetches configuration from backend via HTTP when notified by heartbeat
//! - Applies config changes to agent state (heartbeat interval, metrics collectors)
//! - Maintains local-first fallback config (Tier 0)

use ractor::factory::{FactoryMessage, Job, JobOptions};
use ractor::{async_trait, registry, Actor, ActorProcessingErr, ActorRef, SpawnErr};
use serde::{Deserialize, Serialize};
use std::sync::Arc;
use std::time::Duration;
use tokio::task::JoinHandle;

// ============================================================================
// MESSAGES
// ============================================================================

#[derive(Debug, Clone)]
pub enum ConfigMessage {
    /// Fetch configuration from backend via HTTP
    FetchConfig,
    /// Apply config update (called after HTTP fetch completes)
    ApplyConfig(serde_json::Value),
    /// Spawn metrics collectors based on config (called after config applied)
    SpawnMetricsCollectors,
    /// Check if tier has changed (called from heartbeat response)
    CheckTier(u32),
}

// ============================================================================
// STATE
// ============================================================================

pub struct ConfigState {
    #[allow(dead_code)]
    identity: Arc<crate::AgentIdentity>,
    peer_id: String,
    api_url: String,
    heartbeat_interval_secs: Option<u64>,
    heartbeat_task: Option<tokio::task::JoinHandle<()>>,
    /// Current tier ID (used to detect tier changes from heartbeat)
    tier_id: Option<u32>,
    /// Metrics configuration from remote
    pub metrics_config: Option<MetricsConfig>,
    /// HTTP client with retry middleware for config fetches
    http_client: reqwest_middleware::ClientWithMiddleware,
}

/// Arguments for spawning the Config actor
pub struct ConfigArgs {
    pub identity: Arc<crate::AgentIdentity>,
    pub api_url: Option<String>,
}

// ============================================================================
// CONFIGURATION
// ============================================================================

/// Metrics configuration - native sink settings
#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct NativeSinkConfig {
    pub enabled: bool,
    pub interval_secs: u64,
    pub lttb_threshold: usize,
}

/// Metrics configuration - controls which sinks are enabled
#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct MetricsConfig {
    pub native: Option<NativeSinkConfig>,
    // Future: prometheus, grafana, etc.
}

/// Configuration response from HTTP endpoint
#[derive(Debug, Deserialize, Serialize)]
struct ConfigResponse {
    config: TierConfig,
}

/// Tier configuration (heartbeat, metrics)
#[derive(Debug, Deserialize, Serialize)]
struct TierConfig {
    tier_id: u32,
    heartbeat: serde_json::Value,
    metrics: MetricsConfig,
    // Boost thresholds (not used by agent, but included in API response)
    // Backend sends these in camelCase
    #[serde(rename = "boostMin")]
    #[allow(dead_code)]
    boost_min: u32,
    #[serde(rename = "boostMax")]
    #[allow(dead_code)]
    boost_max: Option<u32>,
}

// ============================================================================
// ACTOR IMPLEMENTATION
// ============================================================================

pub struct Config;

#[async_trait]
impl Actor for Config {
    type Msg = ConfigMessage;
    type State = ConfigState;
    type Arguments = ConfigArgs;

    async fn pre_start(
        &self,
        _myself: ActorRef<Self::Msg>,
        args: Self::Arguments,
    ) -> Result<Self::State, ActorProcessingErr> {
        tracing::debug!("Config actor starting");
        tracing::debug!("  PeerId: {}", args.identity.peer_id());

        let peer_id = args.identity.peer_id().to_string();
        let api_url = args.api_url.unwrap_or_else(crate::constants::api_url);

        tracing::debug!("  API URL: {}", api_url);

        Ok(ConfigState {
            identity: args.identity,
            peer_id,
            api_url,
            heartbeat_interval_secs: None,
            heartbeat_task: None,
            tier_id: None,
            metrics_config: None,
            http_client: crate::http::build_http_client(),
        })
    }

    async fn post_start(
        &self,
        myself: ActorRef<Self::Msg>,
        state: &mut Self::State,
    ) -> Result<(), ActorProcessingErr> {
        tracing::info!("Config actor ready - starting heartbeat loop");

        // Start heartbeat loop FIRST with default Tier 0 interval
        // This ensures heartbeats run even if config fetch fails
        // Heartbeat-first design: if 404, alerts worker will register then heartbeat
        let default_interval = crate::constants::DEFAULT_HEARTBEAT_INTERVAL_SECS;
        state.heartbeat_interval_secs = Some(default_interval);
        let task = tokio::spawn(schedule_heartbeat_loop(default_interval));
        state.heartbeat_task = Some(task);

        tracing::info!("Started heartbeat loop (default: {}s)", default_interval);

        // Trigger initial config fetch (non-blocking)
        // If this fails, heartbeats still run and will trigger config fetch on success
        if let Err(e) = myself.cast(ConfigMessage::FetchConfig) {
            tracing::warn!("Config: Failed to trigger initial config fetch: {}", e);
        }

        Ok(())
    }

    async fn handle(
        &self,
        myself: ActorRef<Self::Msg>,
        message: Self::Msg,
        state: &mut Self::State,
    ) -> Result<(), ActorProcessingErr> {
        match message {
            ConfigMessage::FetchConfig => {
                // Fetch config via HTTP GET using shared client with retry middleware
                let api_url = state.api_url.clone();
                let peer_id = state.peer_id.clone();
                let config_url = format!("{}/v1/agents/{}/config", api_url, peer_id);
                let http_client = state.http_client.clone();

                tracing::debug!("Config: Fetching config from {}", config_url);

                let myself_clone = myself.clone();
                tokio::spawn(async move {
                    match http_client.get(&config_url).send().await {
                        Ok(response) => {
                            if response.status().is_success() {
                                match response.json::<serde_json::Value>().await {
                                    Ok(config) => {
                                        tracing::debug!("Config: Successfully fetched config");
                                        let _ =
                                            myself_clone.cast(ConfigMessage::ApplyConfig(config));
                                    }
                                    Err(e) => {
                                        tracing::warn!(
                                            "Config: Failed to parse config response: {}",
                                            e
                                        );
                                    }
                                }
                            } else {
                                tracing::warn!(
                                    "Config: HTTP {} from config endpoint",
                                    response.status()
                                );
                            }
                        }
                        Err(e) => {
                            tracing::warn!("Config: Failed to fetch config: {}", e);
                        }
                    }
                });
            }

            ConfigMessage::ApplyConfig(config) => {
                tracing::debug!("Config: Applying config update");

                // Parse configuration response
                match serde_json::from_value::<ConfigResponse>(config) {
                    Ok(config_response) => {
                        tracing::debug!("Config: Parsed config successfully");
                        tracing::debug!("  Tier ID: {}", config_response.config.tier_id);

                        // Store tier_id in state (used to detect tier changes)
                        state.tier_id = Some(config_response.config.tier_id);

                        // Store metrics config in state
                        state.metrics_config = Some(config_response.config.metrics.clone());

                        // Determine metrics status for summary
                        let metrics_status =
                            if let Some(native) = &config_response.config.metrics.native {
                                if native.enabled {
                                    format!("enabled ({}s interval)", native.interval_secs)
                                } else {
                                    "disabled".to_string()
                                }
                            } else {
                                "disabled".to_string()
                            };

                        // Extract heartbeat interval, fallback to default if missing
                        let interval_secs: u64 = config_response
                            .config
                            .heartbeat
                            .get("interval_secs")
                            .and_then(|v| v.as_u64())
                            .unwrap_or_else(|| {
                                tracing::warn!(
                                    "Config: No interval_secs found in heartbeat config, using default {}s",
                                    crate::constants::DEFAULT_HEARTBEAT_INTERVAL_SECS
                                );
                                crate::constants::DEFAULT_HEARTBEAT_INTERVAL_SECS
                            });

                        // Cancel existing heartbeat task if any
                        if let Some(task) = state.heartbeat_task.take() {
                            task.abort();
                            tracing::debug!("Config: Cancelled previous heartbeat task");
                        }

                        // Schedule new heartbeat task
                        state.heartbeat_interval_secs = Some(interval_secs);
                        let task = tokio::spawn(async move {
                            schedule_heartbeat_loop(interval_secs).await;
                        });
                        state.heartbeat_task = Some(task);

                        // Single summary log
                        tracing::info!(
                            "Loaded tier {} config (heartbeat: {}s, metrics: {})",
                            config_response.config.tier_id,
                            interval_secs,
                            metrics_status
                        );

                        // Trigger metrics collectors spawn based on config
                        if let Err(e) = myself.cast(ConfigMessage::SpawnMetricsCollectors) {
                            tracing::warn!("Config: Failed to trigger metrics spawn: {}", e);
                        }
                    }
                    Err(e) => {
                        tracing::warn!("Config: Failed to parse config response: {}", e);
                        tracing::info!(
                            "Config: Falling back to default heartbeat interval ({}s)",
                            crate::constants::DEFAULT_HEARTBEAT_INTERVAL_SECS
                        );

                        // Fallback to default Tier 0 interval
                        let interval_secs = crate::constants::DEFAULT_HEARTBEAT_INTERVAL_SECS;

                        // Cancel existing heartbeat task if any
                        if let Some(task) = state.heartbeat_task.take() {
                            task.abort();
                            tracing::debug!("Config: Cancelled previous heartbeat task");
                        }

                        // Schedule heartbeat with default interval
                        state.heartbeat_interval_secs = Some(interval_secs);
                        let task = tokio::spawn(async move {
                            schedule_heartbeat_loop(interval_secs).await;
                        });
                        state.heartbeat_task = Some(task);
                    }
                }
            }

            ConfigMessage::CheckTier(backend_tier_id) => {
                tracing::debug!(
                    "Config: Checking tier - backend: {}, local: {:?}",
                    backend_tier_id,
                    state.tier_id
                );

                // If we don't have a local tier yet, or tier changed, fetch config
                let should_fetch = match state.tier_id {
                    None => {
                        tracing::info!("Config: No local tier - fetching config");
                        true
                    }
                    Some(local_tier_id) => {
                        if backend_tier_id != local_tier_id {
                            tracing::info!(
                                "Config: Tier changed (local: {}, backend: {}) - fetching updated config",
                                local_tier_id,
                                backend_tier_id
                            );
                            true
                        } else {
                            tracing::debug!("Config: Tier {} up-to-date", local_tier_id);
                            false
                        }
                    }
                };

                if should_fetch {
                    if let Err(e) = myself.cast(ConfigMessage::FetchConfig) {
                        tracing::warn!("Config: Failed to trigger config fetch: {}", e);
                    }
                }
            }

            ConfigMessage::SpawnMetricsCollectors => {
                tracing::info!("Config: Spawning metrics collectors based on config");

                // Check if we have metrics config with native sink enabled
                if let Some(metrics_config) = &state.metrics_config {
                    if let Some(native) = &metrics_config.native {
                        if native.enabled {
                            tracing::info!(
                                "Config: Native sink enabled - interval: {}s, lttb_threshold: {}",
                                native.interval_secs,
                                native.lttb_threshold
                            );

                            // Look up MetricsSupervisor and tell it to spawn NativeSink
                            if let Some(metrics_cell) =
                                registry::where_is("MetricsSupervisor".to_string())
                            {
                                let metrics_ref: ActorRef<crate::core::metrics::MetricsMessage> =
                                    metrics_cell.into();

                                // Send message to spawn NativeSink with config
                                if let Err(e) = metrics_ref.cast(
                                    crate::core::metrics::MetricsMessage::SpawnNativeSink {
                                        interval_secs: native.interval_secs,
                                        lttb_threshold: native.lttb_threshold,
                                    },
                                ) {
                                    tracing::warn!(
                                        "Config: Failed to send SpawnNativeSink message: {}",
                                        e
                                    );
                                }
                            } else {
                                tracing::warn!("Config: MetricsSupervisor not found in registry");
                            }
                        } else {
                            tracing::info!("Config: Native sink disabled (enabled = false)");
                        }
                    } else {
                        tracing::info!("Config: Native sink disabled (no native config)");
                    }
                } else {
                    tracing::warn!("Config: No metrics config available");
                }
            }
        }

        Ok(())
    }

    async fn post_stop(
        &self,
        _myself: ActorRef<Self::Msg>,
        state: &mut Self::State,
    ) -> Result<(), ActorProcessingErr> {
        // Cancel heartbeat task on shutdown
        if let Some(task) = state.heartbeat_task.take() {
            task.abort();
            tracing::debug!("Config: Cancelled heartbeat task on shutdown");
        }

        tracing::info!("Config actor stopped");
        Ok(())
    }
}

// ============================================================================
// HEARTBEAT SCHEDULING
// ============================================================================

/// Periodic heartbeat loop that dispatches heartbeats to AlertsFactory
async fn schedule_heartbeat_loop(interval_secs: u64) {
    let mut interval = tokio::time::interval(Duration::from_secs(interval_secs));
    // First tick fires immediately, triggering registration if needed

    loop {
        interval.tick().await;

        // Look up AlertsFactory from registry
        match registry::where_is("AlertsFactory".to_string()) {
            Some(factory_cell) => {
                let factory_ref: ActorRef<FactoryMessage<(), crate::core::alerts::AlertMessage>> =
                    factory_cell.into();

                // Dispatch heartbeat job to factory
                if let Err(e) = factory_ref.cast(FactoryMessage::Dispatch(Job {
                    key: (),
                    msg: crate::core::alerts::AlertMessage::Heartbeat,
                    options: JobOptions::default(),
                    accepted: None,
                })) {
                    tracing::error!(
                        "Config: Failed to dispatch heartbeat to AlertsFactory: {}",
                        e
                    );
                } else {
                    tracing::debug!("Config: Dispatched heartbeat to AlertsFactory");
                }
            }
            None => {
                tracing::warn!("Config: AlertsFactory not found in registry");
            }
        }
    }
}

// ============================================================================
// SPAWN HELPER
// ============================================================================

/// Spawn the Config actor
///
/// Returns an ActorRef and its join handle.
pub async fn spawn_config(
    identity: Arc<crate::AgentIdentity>,
    api_url: Option<String>,
) -> Result<(ActorRef<ConfigMessage>, JoinHandle<()>), SpawnErr> {
    tracing::debug!("Spawning Config actor...");

    let args = ConfigArgs { identity, api_url };
    let (actor_ref, actor_handle) =
        Actor::spawn(Some("Config".to_string()), Config, args).await?;

    tracing::debug!("✓ Config actor spawned successfully");

    Ok((actor_ref, actor_handle))
}

// ============================================================================
// UNIT TESTS
// ============================================================================

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

    #[test]
    fn test_config_response_deserialization() {
        // Test that we can correctly parse HTTP config response
        let config_json = json!({
            "config": {
                "tier_id": 2,
                "boostMin": 0,
                "boostMax": 3,
                "heartbeat": {
                    "interval_secs": 30
                },
                "metrics": {
                    "native": {
                        "enabled": true,
                        "interval_secs": 300,
                        "lttb_threshold": 200
                    }
                }
            }
        });

        let config: Result<ConfigResponse, _> = serde_json::from_value(config_json);
        assert!(config.is_ok());

        let config = config.unwrap();
        assert_eq!(config.config.tier_id, 2);
        assert_eq!(
            config
                .config
                .heartbeat
                .get("interval_secs")
                .and_then(|v| v.as_u64()),
            Some(30)
        );
    }

    #[test]
    fn test_tier_config_heartbeat_extraction() {
        // Test extracting interval_secs from heartbeat config
        let tier_config = TierConfig {
            tier_id: 2,
            heartbeat: json!({"interval_secs": 60}),
            metrics: MetricsConfig { native: None },
            boost_min: 0,
            boost_max: None,
        };

        let interval = tier_config
            .heartbeat
            .get("interval_secs")
            .and_then(|v| v.as_u64());
        assert_eq!(interval, Some(60));
    }

    #[test]
    fn test_config_message_types() {
        // Verify all ConfigMessage variants can be created
        let _fetch = ConfigMessage::FetchConfig;
        let _apply = ConfigMessage::ApplyConfig(json!({"config": {}}));
        let _spawn = ConfigMessage::SpawnMetricsCollectors;
        let _check = ConfigMessage::CheckTier(2);

        // All variants should compile and be Debug
        assert!(true);
    }
}