loa_core/core/alerts/

mod.rs

//! Alerts Factory - Manages workers for different alert types
//!
//! The factory routes alert messages to specific worker implementations:
//! - heartbeat: API authentication and liveness checks
//! - email: Send email notifications (future)
//! - slack: Send Slack notifications (future)

mod heartbeat;
// Future: mod email;
// Future: mod slack;

use ractor::async_trait;
use ractor::factory::*;
use ractor::{Actor, ActorProcessingErr, ActorRef, SpawnErr};
use tokio::task::JoinHandle;

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

#[derive(Debug, Clone)]
pub enum AlertMessage {
    /// Send a heartbeat to the backend API
    /// Returns ws_url in response for ConfigManager to connect WebSocket
    Heartbeat,

    // Future alert types:
    // Email { to: String, subject: String, body: String },
    // Slack { channel: String, message: String },
    // PagerDuty { severity: String, summary: String },
}

// ============================================================================
// WORKER STATE
// ============================================================================

pub struct AlertsState {
    api_url: String,
    identity: std::sync::Arc<crate::AgentIdentity>,
    workspace_id: Option<String>,
    claim_token: Option<String>,
    storage_path: std::path::PathBuf,
    heartbeat_count: u64,
    http_client: reqwest_middleware::ClientWithMiddleware,
}

impl AlertsState {
    /// Attempt to register the agent with the backend.
    /// Returns true if registration succeeded, false otherwise.
    /// On failure, logs warnings but never fails - allows embedded agents to operate without cloud.
    async fn try_register(&mut self) -> bool {
        match heartbeat::handle_register(
            &self.api_url,
            &self.identity,
            &self.workspace_id,
            &self.claim_token,
            &self.http_client,
        )
        .await
        {
            Ok(response) => {
                tracing::info!(
                    "Agent registered as '{}' (id: {}, is_new: {})",
                    response.name,
                    response.agent_id,
                    response.is_new
                );

                // Update local state with server response
                if let Some(ref new_token) = response.claim_token {
                    self.claim_token = Some(new_token.clone());
                }

                // Persist to agent.toml
                if let Err(e) = crate::identity::update_agent_toml_registration(
                    &self.storage_path,
                    Some(response.name.as_str()),
                    response.claim_token.as_deref(),
                ) {
                    tracing::error!("Failed to update agent.toml: {}", e);
                }

                true
            }
            Err(e) => {
                if self.claim_token.is_some() {
                    tracing::warn!(
                        "Registration failed: {} - continuing with existing claim_token",
                        e
                    );
                } else {
                    // No existing claim_token - agent will operate unclaimed until fixed
                    // This allows embedded agents to function locally without cloud connectivity
                    tracing::warn!(
                        "Registration failed and no existing claim_token: {} - agent will remain unclaimed",
                        e
                    );
                }
                false
            }
        }
    }
}

// ============================================================================
// CONFIGURATION
// ============================================================================
//
// Note: API URL is now defined in crate::constants module for centralized management

// ============================================================================
// WORKER ARGUMENTS
// ============================================================================

pub struct AlertsWorkerArgs {
    pub api_url: String,
    pub identity: std::sync::Arc<crate::AgentIdentity>,
    pub workspace_id: Option<String>,
    pub storage_path: std::path::PathBuf,
}

// ============================================================================
// WORKER IMPLEMENTATION
// ============================================================================

pub struct AlertsWorker;

#[async_trait]
impl Worker for AlertsWorker {
    type Key = ();
    type Message = AlertMessage;
    type State = AlertsState;
    type Arguments = AlertsWorkerArgs;

    async fn pre_start(
        &self,
        wid: WorkerId,
        _factory: &ActorRef<FactoryMessage<(), AlertMessage>>,
        args: AlertsWorkerArgs,
    ) -> Result<AlertsState, ActorProcessingErr> {
        tracing::debug!("AlertsWorker {} starting", wid);
        tracing::debug!("  API URL: {}", args.api_url);
        tracing::debug!("  PeerId: {}", args.identity.peer_id());

        // Load claim token from storage (TOML or legacy file)
        let claim_token = crate::identity::read_claim_token(&args.storage_path);
        if claim_token.is_some() {
            tracing::debug!("  Loaded claim_token from storage");
        }

        Ok(AlertsState {
            api_url: args.api_url,
            identity: args.identity,
            workspace_id: args.workspace_id,
            claim_token,
            storage_path: args.storage_path,
            heartbeat_count: 0,
            http_client: crate::http::build_http_client(),
        })
    }

    async fn post_start(
        &self,
        wid: WorkerId,
        _factory: &ActorRef<FactoryMessage<(), AlertMessage>>,
        state: &mut AlertsState,
    ) -> Result<(), ActorProcessingErr> {
        // Always register on startup to ensure agent is known and claims are applied
        state.try_register().await;
        tracing::debug!("AlertsWorker {} ready", wid);
        Ok(())
    }

    async fn handle(
        &self,
        _wid: WorkerId,
        _factory: &ActorRef<FactoryMessage<(), AlertMessage>>,
        Job { msg, key, .. }: Job<(), AlertMessage>,
        state: &mut AlertsState,
    ) -> Result<(), ActorProcessingErr> {
        // Route message to specific handler
        match msg {
            AlertMessage::Heartbeat => {
                // Heartbeat-first design: try heartbeat, if 404 → register → heartbeat
                match heartbeat::handle_heartbeat(
                    &state.api_url,
                    &state.identity,
                    &state.http_client,
                    &mut state.heartbeat_count,
                )
                .await
                {
                    Ok(()) => {
                        // Success - heartbeat handler already sent CheckTier to Config
                    }
                    Err(e) if e.to_string().contains("404") => {
                        // Not registered - register now
                        tracing::info!("Heartbeat 404 - re-registering agent");

                        if state.try_register().await {
                            // Immediately send heartbeat to trigger config fetch
                            // (heartbeat response has tier_id → CheckTier → FetchConfig)
                            if let Err(e) = heartbeat::handle_heartbeat(
                                &state.api_url,
                                &state.identity,
                                &state.http_client,
                                &mut state.heartbeat_count,
                            )
                            .await
                            {
                                tracing::warn!("Post-registration heartbeat failed: {}", e);
                            }
                        }
                    }
                    Err(e) => {
                        // Other error (network, 5xx) - retry middleware handles retries
                        tracing::warn!("Heartbeat failed: {}", e);
                    }
                }
            }
            // Future alert types will be routed here
        }

        Ok(key)
    }

    async fn post_stop(
        &self,
        wid: WorkerId,
        _factory: &ActorRef<FactoryMessage<(), AlertMessage>>,
        state: &mut AlertsState,
    ) -> Result<(), ActorProcessingErr> {
        tracing::debug!(
            "AlertsWorker {} stopped (sent {} heartbeats)",
            wid,
            state.heartbeat_count
        );
        Ok(())
    }
}

// ============================================================================
// WORKER BUILDER
// ============================================================================

pub struct AlertsWorkerBuilder {
    identity: std::sync::Arc<crate::AgentIdentity>,
    workspace_id: Option<String>,
    api_url: Option<String>,
    storage_path: std::path::PathBuf,
}

impl WorkerBuilder<AlertsWorker, AlertsWorkerArgs> for AlertsWorkerBuilder {
    fn build(&mut self, _wid: WorkerId) -> (AlertsWorker, AlertsWorkerArgs) {
        (
            AlertsWorker,
            AlertsWorkerArgs {
                // Use custom api_url if provided, otherwise fall back to default
                api_url: self.api_url.clone().unwrap_or_else(crate::constants::api_url),
                identity: self.identity.clone(),
                workspace_id: self.workspace_id.clone(),
                storage_path: self.storage_path.clone(),
            },
        )
    }
}

// ============================================================================
// FACTORY LIFECYCLE HOOKS
// ============================================================================

pub struct AlertsFactoryHooks;

#[async_trait]
impl FactoryLifecycleHooks<(), AlertMessage> for AlertsFactoryHooks {
    async fn on_factory_started(
        &self,
        _factory: ActorRef<FactoryMessage<(), AlertMessage>>,
    ) -> Result<(), ActorProcessingErr> {
        tracing::debug!("Alerts started");
        Ok(())
    }

    async fn on_factory_stopped(&self) -> Result<(), ActorProcessingErr> {
        tracing::info!("Alerts stopped");
        Ok(())
    }

    async fn on_factory_draining(
        &self,
        _factory: ActorRef<FactoryMessage<(), AlertMessage>>,
    ) -> Result<(), ActorProcessingErr> {
        tracing::debug!("AlertsFactory draining");
        Ok(())
    }
}

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

/// Spawn the Alerts factory with a single worker
///
/// Returns a factory ActorRef and its join handle.
pub async fn spawn_alerts_factory(
    identity: std::sync::Arc<crate::AgentIdentity>,
    workspace_id: Option<String>,
    api_url: Option<String>,
    storage_path: std::path::PathBuf,
) -> Result<(ActorRef<FactoryMessage<(), AlertMessage>>, JoinHandle<()>), SpawnErr> {
    tracing::debug!("Spawning AlertsFactory...");

    let factory_def = Factory::<
        (),
        AlertMessage,
        AlertsWorkerArgs,
        AlertsWorker,
        routing::QueuerRouting<(), AlertMessage>,
        queues::DefaultQueue<(), AlertMessage>,
    >::default();

    let factory_args = FactoryArguments::builder()
        .worker_builder(Box::new(AlertsWorkerBuilder {
            identity,
            workspace_id,
            api_url,
            storage_path,
        }))
        .queue(Default::default())
        .router(Default::default())
        .num_initial_workers(1)
        .lifecycle_hooks(Box::new(AlertsFactoryHooks))
        .build();

    let (factory, factory_handle) = Actor::spawn(
        Some("AlertsFactory".to_string()),
        factory_def,
        factory_args,
    )
    .await
    .map_err(|e| SpawnErr::StartupFailed(Box::new(e)))?;

    tracing::debug!("\u{2713} AlertsFactory spawned successfully (1 worker)");

    Ok((factory, factory_handle))
}