product_os_command_control/

lib.rs

//! # Product OS Command and Control
//!
//! This crate provides a distributed command and control system for coordinating
//! multiple Product OS server instances. It enables secure communication, service
//! discovery, and workload distribution across a cluster of nodes.
//!
//! ## Features
//!
//! - **Secure Communication**: Authentication framework using Diffie-Hellman key exchange
//! - **Service Discovery**: Automatic node registration and discovery
//! - **Load Balancing**: Smart routing to available nodes based on capabilities
//! - **Health Monitoring**: Pulse checks and automatic failure detection
//! - **Feature Management**: Dynamic feature and service registration
//!
//! ## Basic Usage
//!
//! ```no_run
//! use product_os_command_control::ProductOSController;
//! use product_os_command_control::CommandControl;
//! use product_os_security::certificates::Certificates;
//!
//! # async fn example() {
//! let config = CommandControl::default();
//! let certs = Certificates::new_self_signed(
//!     vec![("CN".to_string(), "localhost".to_string())],
//!     None, None, None, None, None,
//! );
//!
//! let controller = ProductOSController::new(
//!     config,
//!     certs,
//!     None, // Optional key-value store
//!     #[cfg(feature = "relational_store")]
//!     None, // Optional relational store
//! );
//!
//! let registry = controller.get_registry();
//! let my_node = registry.get_me();
//! println!("Node ID: {}", my_node.get_identifier());
//! # }
//! ```
//!
//! ## Architecture
//!
//! The command and control system consists of several key components:
//!
//! - **ProductOSController**: Main coordinator that manages nodes and services
//! - **Registry**: Tracks available nodes and their capabilities
//! - **Node**: Represents a single server instance in the cluster
//! - **Commands**: Structured way to send instructions to remote nodes
//!
//! ## Security
//!
//! All node-to-node communication is secured using:
//! - TLS/SSL certificates for transport security
//! - Diffie-Hellman key exchange for establishing shared secrets
//! - Message authentication using HMAC
//!
//! ## Cargo Features
//!
//! | Feature | Default | Description |
//! |---------|---------|-------------|
//! | `std_base` | Yes | Std support with core Product OS dependencies |
//! | `framework_axum` | Yes | Axum HTTP framework |
//! | `framework_flow` | No | Flow HTTP framework |
//! | `monitor` | Yes | Monitoring via `product-os-monitoring` |
//! | `tokio` | No | Tokio async executor |
//! | `distributed` | No | Distributed node discovery (placeholder) |
//! | `relational_store` | No | Base relational store support |
//! | `postgres_store` | No | PostgreSQL relational store |
//! | `sqlite_store` | No | SQLite relational store |
//! | `redis_key_value_store` | No | Redis key-value store |
//! | `memory_key_value_store` | No | In-memory key-value store |
//! | `file_key_value_store` | No | File-based key-value store |
//! | `redis_queue_store` | No | Redis queue store |
//! | `memory_queue_store` | No | In-memory queue store |

#![no_std]
#![warn(rust_2018_idioms)]
#![warn(clippy::all)]
#![allow(clippy::module_name_repetitions)]
#![allow(clippy::must_use_candidate)]
#![allow(clippy::too_many_arguments)]
#![allow(clippy::missing_errors_doc)]
#![allow(clippy::missing_panics_doc)]
#![allow(missing_docs)]
#![allow(ambiguous_glob_imports)]
#![allow(ambiguous_panic_imports)]

extern crate no_std_compat as std;

use std::prelude::v1::*;

mod config;
pub use config::CommandControl;

pub mod authentication;
mod registry;
pub mod commands;

use std::collections::BTreeMap;

use std::sync::Arc;
#[cfg(feature = "distributed")]
use product_os_request::Bytes;
use parking_lot::Mutex;
#[cfg(feature = "distributed")]
use parking_lot::MutexGuard;
#[cfg(feature = "relational_store")]
use product_os_store::{KeyValueKind, KeyValueStore, RelationalKind, RelationalStore};
#[cfg(not(feature = "relational_store"))]
use product_os_store::{KeyValueKind, KeyValueStore};

use std::str::FromStr;
use std::time::Duration;

pub use product_os_request::Method;
use product_os_request::{ProductOSClient, ProductOSResponse};
pub use registry::Node;

/// Constant for the "true" string used in query maps.
const ENABLED_STR: &str = "true";


/// Main controller for managing a distributed command and control system.
///
/// `ProductOSController` is the primary interface for coordinating multiple
/// Product OS server instances. It handles node discovery, service management,
/// secure communication, and workload distribution.
///
/// # Examples
///
/// ```no_run
/// use product_os_command_control::ProductOSController;
/// use product_os_command_control::CommandControl;
/// use product_os_security::certificates::Certificates;
///
/// let config = CommandControl::default();
/// let certs = Certificates::new_self_signed(
///     vec![("CN".to_string(), "localhost".to_string())],
///     None, None, None, None, None,
/// );
///
/// let controller = ProductOSController::new(
///     config,
///     certs,
///     None,
///     #[cfg(feature = "relational_store")]
///     None,
/// );
/// ```
pub struct ProductOSController {
    command_control: crate::config::CommandControl,

    certificates: product_os_security::certificates::Certificates,

    max_servers: u8,

    registry: registry::Registry,

    pulse_check: bool,
    pulse_check_cron: String,

    #[cfg(feature = "monitor")]
    monitor: bool,
    #[cfg(feature = "monitor")]
    monitor_cron: String,

    requester: product_os_request::ProductOSRequester,
    client: product_os_request::ProductOSRequestClient,

    #[cfg(feature = "relational_store")]
    relational_store: Arc<product_os_store::ProductOSRelationalStore>,
    key_value_store: Arc<product_os_store::ProductOSKeyValueStore>,

    /// Optional callback invoked when the node loses its presence on the remote registry.
    /// If `None`, `std::process::exit(8)` is called instead (the legacy behavior).
    pub on_presence_lost: Option<Arc<dyn Fn() + Send + Sync>>,
}

impl ProductOSController {
    /// Creates a new `ProductOSController` instance.
    ///
    /// Initializes the command and control system with the provided configuration,
    /// certificates, and optional data stores.
    ///
    /// # Arguments
    ///
    /// * `command_control` - Command and control configuration
    /// * `certificates` - TLS/SSL certificates for secure communication
    /// * `key_value_store` - Optional key-value store for node registry persistence
    /// * `relational_store` - Optional relational database store (requires `relational_store` feature)
    pub fn new(command_control: crate::config::CommandControl, certificates:
               product_os_security::certificates::Certificates,
               key_value_store: Option<Arc<product_os_store::ProductOSKeyValueStore>>,
               #[cfg(feature = "relational_store")] relational_store: Option<Arc<product_os_store::ProductOSRelationalStore>>) -> Self {
        let key_value_store: Arc<product_os_store::ProductOSKeyValueStore> = key_value_store.unwrap_or_else(|| Arc::new(product_os_store::ProductOSKeyValueStore::try_new(&KeyValueStore {
            enabled: false,
            kind: KeyValueKind::Sink,
            host: "".to_string(),
            port: 0,
            secure: false,
            db_number: 0,
            db_name: None,
            username: None,
            password: None,
            pool_size: 0,
            default_limit: 0,
            default_offset: 0,
            prefix: None,
        }).expect("Sink key-value store creation should not fail")));

        #[cfg(feature = "relational_store")]
        let relational_store: Arc<product_os_store::ProductOSRelationalStore> = relational_store.unwrap_or_else(|| Arc::new(product_os_store::ProductOSRelationalStore::try_new(&RelationalStore {
            enabled: false,
            kind: RelationalKind::Sink,
            host: "".to_string(),
            port: 0,
            secure: false,
            db_name: "".to_string(),
            username: None,
            password: None,
            pool_size: 0,
            default_limit: 0,
            default_offset: 0,
            prefix: None,
        }).expect("Sink relational store creation should not fail")));

        let registry = registry::Registry::new(&command_control, key_value_store.clone(), certificates.to_owned());

        let mut requester = product_os_request::ProductOSRequester::new();
        requester.add_header("x-product-os-command", registry.get_me().get_identifier().as_str(), false);
        let mut request_client = product_os_request::ProductOSRequestClient::new();
        request_client.build(&requester);

        Self {
            certificates,

            max_servers: command_control.max_servers,

            pulse_check: command_control.pulse_check,
            pulse_check_cron: command_control.pulse_check_cron.clone(),

            #[cfg(feature = "monitor")]
            monitor: command_control.monitor,
            #[cfg(feature = "monitor")]
            monitor_cron: command_control.monitor_cron.clone(),

            command_control,

            #[cfg(feature = "relational_store")]
            relational_store,
            key_value_store,

            registry,

            requester,
            client: request_client,

            on_presence_lost: None,
        }
    }

    /// Returns a reference to the internal node registry.
    ///
    /// The registry contains information about all known nodes in the cluster,
    /// including their capabilities, services, and features.
    pub fn get_registry(&self) -> &registry::Registry {
        &self.registry
    }

    /// Discovers remote nodes and adds their certificates to the trusted set.
    pub async fn discover_nodes(&mut self) {
        self.registry.discover_nodes().await;

        for cert in self.registry.get_nodes_raw_certificates(0, true) {
            self.requester.add_trusted_certificate_der(cert);
        }

        self.client.build(&self.requester);
    }

    /// Returns the maximum number of servers this controller manages.
    pub fn get_max_servers(&self) -> u8 {
        self.max_servers
    }

    /// Inserts or updates a node in the local registry.
    pub fn upsert_node_local(&mut self, identifier: String, node: Node) {
        self.registry.upsert_node_local(identifier, node);
    }

    /// Returns the TLS certificate chain as a vector of byte slices.
    pub fn get_certificates(&self) -> Vec<&[u8]> {
        self.certificates.certificates.iter().map(|cert| cert.as_slice()).collect()
    }

    /// Returns the TLS private key as a byte slice.
    pub fn get_private_key(&self) -> &[u8] {
        self.certificates.private.as_slice()
    }

    /// Returns the full certificates and private key structure.
    pub fn get_certificates_and_private_key(&self) -> product_os_security::certificates::Certificates {
        self.certificates.to_owned()
    }

    /// Validates that the given certificate matches one of the controller's certificates.
    pub fn validate_certificate(&self, certificate: &[u8]) -> bool {
        tracing::trace!("Checking certificate stored {:?} vs given {:?}", self.certificates.certificates, certificate);

        for cert in self.get_certificates() {
            if cert.eq(certificate) { return true; }
        }

        false
    }

    /// Validates a verification tag against the stored key for the given identifier.
    pub fn validate_verify_tag<T>(&self, query: Option<&BTreeMap<String, String>>, payload: Option<T>,
                                  headers: Option<&BTreeMap<String, String>>, verify_identifier: &str, verify_tag: &str) -> bool
    where T: product_os_security::AsByteVector
    {
        tracing::trace!("Checking verification provided {} from {}", verify_tag, verify_identifier);
        let key = self.registry.get_key(verify_identifier);
        tracing::trace!("Checking verification with key {:?}", key);

        match key {
            Some(verify_key) => {
                product_os_security::verify_auth_request(query, false, payload, headers,
                                                       &["x-product-os-verify", "x-product-os-command", "x-product-os-control"],
                                                       verify_tag, Some(verify_key.as_slice()))
            },
            None => false
        }
    }

    /// Authenticates a command and control request using the verify identifier and tag.
    ///
    /// Returns `Ok(true)` if authentication succeeds, or an error if the keys are invalid.
    pub fn authenticate_command_control<T>(&self, query: Option<&BTreeMap<String, String>>, payload: Option<T>, headers: Option<&BTreeMap<String, String>>,
                                        verify_identifier: Option<&str>, verify_tag: Option<&str>) -> Result<bool, authentication::CommandControlAuthenticateError>
    where T: product_os_security::AsByteVector
    {
        tracing::trace!("Verify identifier {:?} and tag {:?} verify result", &verify_identifier, &verify_tag);
        if verify_identifier.is_some() && verify_tag.is_some() && self.validate_verify_tag(query, payload, headers, verify_identifier.unwrap(), verify_tag.unwrap()) {
            Ok(true)
        }
        else {
            Err(authentication::CommandControlAuthenticateError {
                error: authentication::CommandControlAuthenticateErrorState::KeyError("One of the keys provided was not valid".to_string())
            })
        }
    }

    /// Searches for a node matching the query and prepares a command to send to it.
    pub fn search_and_prepare_command(&self, query: BTreeMap<&str, &str>, module: String, instruction: String, data: Option<serde_json::Value>)  -> Result<crate::commands::Command, product_os_request::ProductOSRequestError> {
        let matching_node = self.registry.pick_node(query);
        let client = &self.client;

        match matching_node {
            Some(node) => {
                match self.registry.get_key(node.get_identifier().as_str()) {
                    None => Err(product_os_request::ProductOSRequestError::Error("No key found for matching node".to_string())),
                    Some(key) => Ok(commands::Command {
                        requester: client.clone(),
                        #[allow(deprecated)]
                        node_url: node.get_address().to_owned(),
                        verify_key: key,
                        module,
                        instruction,
                        data
                    })
                }
            },
            None => Err(product_os_request::ProductOSRequestError::Error("No matching node found".to_string()))
        }
    }

    /// Finds a node with a matching manager key/kind and prepares a command.
    pub fn find_and_prepare_command(&self, key: &str, manager: &str, instruction: String, data: Option<serde_json::Value>) -> Result<commands::Command, product_os_request::ProductOSRequestError> {
        let query = BTreeMap::from([
            ("manager.key", key),
            ("manager.kind", manager),
            ("manager.enabled", ENABLED_STR)
        ]);

        self.search_and_prepare_command(query, manager.to_string(), instruction, data)
    }

    /// Finds a node with a matching capability and prepares a command.
    pub fn find_any_and_prepare_command(&self, capability: &str, manager: &str, instruction: &str, data: Option<serde_json::Value>) -> Result<commands::Command, product_os_request::ProductOSRequestError> {
        let query = BTreeMap::from([
            ("capability", capability),
            ("manager.enabled", ENABLED_STR)
        ]);

        self.search_and_prepare_command(query, manager.to_string(), instruction.to_string(), data)
    }

    /// Finds a node with a matching kind and prepares a command.
    ///
    /// Note: despite the `async` signature, this method performs no asynchronous work.
    /// A non-async version is available as `find_kind_and_prepare_command_sync`.
    #[deprecated(since = "0.0.28", note = "Use find_kind_and_prepare_command_sync instead; this method is not actually async")]
    pub async fn find_kind_and_prepare_command(&self, kind: &str, manager: &str, instruction: &str, data: Option<serde_json::Value>) -> Result<commands::Command, product_os_request::ProductOSRequestError> {
        self.find_kind_and_prepare_command_sync(kind, manager, instruction, data)
    }

    /// Finds a node with a matching kind and prepares a command (non-async).
    pub fn find_kind_and_prepare_command_sync(&self, kind: &str, manager: &str, instruction: &str, data: Option<serde_json::Value>) -> Result<commands::Command, product_os_request::ProductOSRequestError> {
        let query = BTreeMap::from([
            ("kind", kind),
            ("manager.enabled", ENABLED_STR)
        ]);

        self.search_and_prepare_command(query, manager.to_string(), instruction.to_string(), data)
    }

    /// Searches for a node matching the query and prepares an HTTP request to send to it.
    pub fn search_and_prepare_ask(&self, query: BTreeMap<&str, &str>, path: &str, data: Option<serde_json::Value>, headers: BTreeMap<String, String>,
                                params: BTreeMap<String, String>, method: Method) -> Result<commands::Ask, product_os_request::ProductOSRequestError> {
        let matching_node = self.registry.pick_node(query);
        let client = &self.client;
        match matching_node {
            Some(node) => {
                match self.registry.get_key(node.get_identifier().as_str()) {
                    None => Err(product_os_request::ProductOSRequestError::Error("No key found for matching node".to_string())),
                    Some(key) => Ok(commands::Ask {
                        requester: client.clone(),
                        #[allow(deprecated)]
                        node_url: node.get_address().to_owned(),
                        verify_key: key,
                        path: path.to_string(),
                        data,
                        headers,
                        params,
                        method
                    })
                }

            },
            None => Err(product_os_request::ProductOSRequestError::Error("No matching node found".to_string()))
        }
    }

    /// Finds a node with the specified feature and sends an HTTP request to it.
    pub async fn find_feature_and_ask(&self, feature: &str, path: &str, data: &Option<serde_json::Value>, headers: &BTreeMap<String, String>,
                                params: &BTreeMap<String, String>, method: &Method) -> Result<ProductOSResponse<product_os_request::BodyBytes>, product_os_request::ProductOSRequestError> {
        let matching_node = self.registry.pick_node_for_feature(feature);
        let client = &self.client;
        match matching_node {
            Some(node) => {
                match self.registry.get_key(node.get_identifier().as_str()) {
                    None => Err(product_os_request::ProductOSRequestError::Error("No key found for matching node".to_string())),
                    Some(key) => commands::ask_node(client, node, key.as_slice(), path, data, headers, params, method).await
                }

            },
            None => Err(product_os_request::ProductOSRequestError::Error("No matching node found".to_string()))
        }
    }

    /// Finds a node with the specified feature and prepares an HTTP request to send to it.
    pub fn find_feature_and_prepare_ask(&self, feature: &str, path: &str, data: Option<serde_json::Value>, headers: BTreeMap<String, String>,
                                      params: BTreeMap<String, String>, method: Method) -> Result<commands::Ask, product_os_request::ProductOSRequestError> {
        let matching_node = self.registry.pick_node_for_feature(feature);
        let client = &self.client;
        match matching_node {
            Some(node) => {
                match self.registry.get_key(node.get_identifier().as_str()) {
                    None => Err(product_os_request::ProductOSRequestError::Error("No key found for matching node".to_string())),
                    Some(key) => Ok(commands::Ask {
                        requester: client.clone(),
                        #[allow(deprecated)]
                        node_url: node.get_address().to_owned(),
                        verify_key: key,
                        path: path.to_string(),
                        data,
                        headers,
                        params,
                        method
                    })
                }

            },
            None => Err(product_os_request::ProductOSRequestError::Error("No matching node found".to_string()))
        }
    }

    /// Returns a clone of the command and control configuration.
    pub fn get_configuration(&self) -> crate::config::CommandControl {
        self.command_control.clone()
    }

    /// Returns the cryptographic key associated with the given node identifier.
    pub fn get_key(&self, identifier: &str) -> Option<Vec<u8>> {
        self.registry.get_key(identifier)
    }

    /// Creates a new Diffie-Hellman key session and returns the session ID and public key.
    pub fn create_key_session(&mut self) -> (String, [u8; 32]) {
        self.registry.create_key_session()
    }

    /// Generates a shared key from a Diffie-Hellman exchange session.
    pub fn generate_key(&mut self, session_identifier: &str, remote_public_key: &[u8], association: String, remote_session_identifier: Option<String>) {
        self.registry.generate_key(session_identifier, remote_public_key, association, remote_session_identifier);
    }

    /// Returns the relational store (requires `relational_store` feature).
    #[cfg(feature = "relational_store")]
    pub fn get_relational_store(&mut self) -> Arc<product_os_store::ProductOSRelationalStore> {
        self.relational_store.clone()
    }

    /// Returns the key-value store.
    pub fn get_key_value_store(&mut self) -> Arc<product_os_store::ProductOSKeyValueStore> {
        self.key_value_store.clone()
    }

    /// Registers a feature on the local node and updates the router.
    pub async fn add_feature(&mut self, feature: Arc<dyn product_os_capabilities::Feature>, base_path: String, router: &mut product_os_router::ProductOSRouter) {
        self.registry.add_feature(feature, base_path, router).await;
    }

    /// Registers a mutable feature on the local node and updates the router.
    pub async fn add_feature_mut(&mut self, feature: Arc<Mutex<dyn product_os_capabilities::Feature>>, base_path: String, router: &mut product_os_router::ProductOSRouter) {
        self.registry.add_feature_mut(feature, base_path, router).await;
    }

    /// Removes a feature from the local node.
    pub async fn remove_feature(&mut self, identifier: &str) {
        self.registry.remove_feature(identifier).await;
    }

    /// Registers a service on the local node.
    pub async fn add_service(&mut self, service: Arc<dyn product_os_capabilities::Service>) {
        self.registry.add_service(service).await;
    }

    /// Registers a mutable service on the local node.
    pub async fn add_service_mut(&mut self, service: Arc<Mutex<dyn product_os_capabilities::Service>>) {
        self.registry.add_service_mut(service).await;
    }

    /// Sets the active status of a service on the local node.
    pub async fn set_service_active(&mut self, identifier: String, status: bool) {
        self.registry.set_service_active(identifier, status).await;
    }

    /// Removes a service from the local node.
    pub async fn remove_service(&mut self, identifier: &str) {
        self.registry.remove_service(identifier).await;
    }

    /// Removes all inactive services matching the query.
    pub async fn remove_inactive_services(&mut self, query: BTreeMap<&str, &str>) {
        self.registry.remove_inactive_services(query).await;
    }

    /// Starts all registered services on the local node.
    pub async fn start_services(&mut self) -> Result<(), ()> {
        self.registry.start_services().await
    }

    /// Starts the service with the given identifier.
    pub async fn start_service(&mut self, identifier: &str) -> Result<(), ()> {
        self.registry.start_service(identifier).await
    }

    /// Stops the service with the given identifier.
    pub async fn stop_service(&mut self, identifier: &str) -> Result<(), ()> {
        self.registry.stop_service(identifier).await
    }
}




/// Runs a single pulse check cycle, verifying node health and connectivity.
///
/// Acquires a lock on the controller, checks the local node's
/// presence on the remote registry (when `distributed` feature is enabled),
/// pings remote nodes, and handles failures.
///
/// If the node loses presence and no `on_presence_lost` callback is set,
/// `std::process::exit(8)` is called.
#[allow(clippy::await_holding_lock)]
pub async fn pulse_run(controller_unlocked: Arc<Mutex<ProductOSController>>) {
    tracing::info!("Starting pulse run...");
    let controller_locked = controller_unlocked.try_lock_for(core::time::Duration::new(10, 0));
    match controller_locked {
        #[allow(unused_mut)]
        Some(mut controller) => {
            #[cfg(feature = "distributed")]
            controller.discover_nodes().await;

            #[cfg(feature = "distributed")]
            let self_identifier = controller.registry.get_me().get_identifier();
            #[cfg(feature = "distributed")]
            #[allow(deprecated)]
            let control_url = controller.registry.get_me().get_address();

            #[cfg(feature = "distributed")]
            let nodes = controller.registry.get_nodes_endpoints(0, true);

            #[cfg(feature = "distributed")] {
                match controller.registry.check_me_remote().await {
                    Some(_) => { controller.registry.update_me_status(true); },
                    None => {
                        let alive = controller.registry.update_me_status(false);

                        if !alive {
                            tracing::error!("Terminating server due to lost presence on remote registry");
                            match &controller.on_presence_lost {
                                Some(callback) => callback(),
                                None => std::process::exit(8),
                            }
                        };
                    }
                }
            }

            // Check status of services
            let services = controller.get_registry().get_me().get_services().list();
            for (_, service) in services {
                let _ = service.status().await;
            }

            #[cfg(feature = "distributed")]
            let client = controller.client.clone();

            // Explicitly drop controller
            std::mem::drop(controller);

            #[cfg(feature = "distributed")]
            for (identifier, (url, key)) in nodes {
                if url != control_url {
                    match key {
                        Some(verify_key) => {
                            match commands::command(&client, url.clone(), verify_key, "status", "ping", None).await {
                                Ok(response) => {
                                    let Some(status) = response.try_status() else {
                                        tracing::error!("Failed to get status from response for {}", url);
                                        continue;
                                    };

                                    match status {
                                        product_os_request::StatusCode::UNAUTHORIZED => {
                                            let text = {
                                                let controller_locked = controller_unlocked.try_lock_for(core::time::Duration::new(10, 0));
                                                match controller_locked {
                                                    Some(controller) => {
                                                        controller.client.text(response).await.unwrap_or_else(|e| {
                                                            tracing::error!("Failed to parse response text: {:?}", e);
                                                            String::new()
                                                        })
                                                    }
                                                    None => String::new()
                                                }
                                            };

                                            let auth: authentication::CommandControlAuthenticateError = match serde_json::from_str(text.as_str()) {
                                                Ok(auth_error) => auth_error,
                                                Err(_) => authentication::CommandControlAuthenticateError { error: authentication::CommandControlAuthenticateErrorState::None }
                                            };

                                            tracing::error!("Error object auth {:?}", auth);

                                            match auth.error {
                                                authentication::CommandControlAuthenticateErrorState::KeyError(_) => {
                                                    tracing::info!("Attempting to purge remote node - keys don't match {}", identifier);

                                                    let controller_locked = controller_unlocked.try_lock_for(core::time::Duration::new(10, 0));
                                                    match controller_locked {
                                                        Some(mut controller) => {
                                                            controller.registry.remove_node(identifier.as_str()).await;
                                                        }
                                                        None => tracing::error!("Failed to lock controller")
                                                    }
                                                },
                                                authentication::CommandControlAuthenticateErrorState::None => ()
                                            };
                                        },
                                        product_os_request::StatusCode::OK => {
                                            let body = {
                                                let controller_locked = controller_unlocked.try_lock_for(core::time::Duration::new(10, 0));
                                                match controller_locked {
                                                    Some(controller) => {
                                                        controller.client.bytes(response).await.unwrap_or_else(|e| {
                                                            tracing::error!("Failed to parse response bytes: {:?}", e);
                                                            Bytes::new()
                                                        })
                                                    }
                                                    None => Bytes::new()
                                                }
                                            };

                                            tracing::info!("Response received from {}: {} {:?}", url, status, body);
                                        }
                                        _ => {
                                            let body = {
                                                let controller_locked = controller_unlocked.try_lock_for(core::time::Duration::new(10, 0));
                                                match controller_locked {
                                                    Some(controller) => {
                                                        controller.client.bytes(response).await.unwrap_or_else(|e| {
                                                            tracing::error!("Failed to parse error response bytes: {:?}", e);
                                                            Bytes::new()
                                                        })
                                                    }
                                                    None => Bytes::new()
                                                }
                                            };
                                            tracing::error!("Error response received from {}: {} {:?}", url, status, body);
                                            tracing::info!("Attempting to purge remote node - keys don't match {}", identifier);

                                            let controller_locked = controller_unlocked.try_lock_for(core::time::Duration::new(10, 0));
                                            match controller_locked {
                                                Some(mut controller) => {
                                                    controller.registry.update_pulse_status(identifier.as_str(), false).await;
                                                }
                                                None => tracing::error!("Failed to lock controller")
                                            }
                                        }
                                    }
                                },
                                Err(e) => {
                                    tracing::error!("Error encountered {:?} from {}", e, url);

                                    let controller_locked = controller_unlocked.try_lock_for(core::time::Duration::new(10, 0));
                                    match controller_locked {
                                        Some(mut controller) => {
                                            controller.registry.update_pulse_status(identifier.as_str(), false).await;
                                        }
                                        None => tracing::error!("Failed to lock controller")
                                    }
                                }
                            }
                        },
                        None => ()
                    }
                } else if self_identifier != identifier {
                    tracing::info!("Attempting to purge self duplicate node - identifier mismatch {}", identifier);
                    let controller_locked = controller_unlocked.try_lock_for(core::time::Duration::new(10, 0));
                    match controller_locked {
                        Some(mut controller) => {
                            controller.registry.remove_node(identifier.as_str()).await;
                        }
                        None => tracing::error!("Failed to lock controller")
                    }
                }
            }


            tracing::info!("Finished pulse run...");
        }
        None => tracing::error!("Failed to lock controller")
    }
}


/// Runs the controller lifecycle: self-trust, discovery, announcements, and periodic tasks.
///
/// Sets up pulse check and monitoring timers using the provided async executor.
#[allow(clippy::await_holding_lock)]
pub async fn run_controller<X, E>(controller_mutex: Arc<Mutex<ProductOSController>>, executor: Arc<E>)
where
    E: product_os_async_executor::Executor<X> + product_os_async_executor::ExecutorPerform<X> + product_os_async_executor::Timer
{
    let controller_unlocked = controller_mutex.clone();
    let controller_locked = controller_unlocked.try_lock_for(core::time::Duration::new(10, 0));
    match controller_locked {
        Some(mut controller) => {

            #[cfg(feature = "distributed")] {
                authentication::perform_self_trust(&mut controller);

                controller.registry.update_me().await;
                tracing::info!("Added self...");

                tracing::info!("Command and Control registered for {}", controller.registry.get_me().get_identifier());

                controller.discover_nodes().await;
                perform_announce(&mut controller).await;
            }

            // Start up services
            if controller.command_control.auto_start_services {
                match controller.start_services().await {
                    Ok(_) => {
                        tracing::info!("Command Control services started");
                    },
                    Err(_) => {
                        tracing::error!("Command Control services failed to start");
                    }
                }
            }

            let pulse_check = controller.pulse_check;
            let pulse_check_cron = cron::Schedule::from_str(controller.pulse_check_cron.as_str()).unwrap();
            let pulse_check_next = pulse_check_cron.upcoming(chrono::Utc).next().unwrap();
            let pulse_check_following = pulse_check_cron.upcoming(chrono::Utc).nth(1).unwrap();

            #[cfg(feature = "monitor")]
            let monitor = controller.monitor;
            #[cfg(feature = "monitor")]
            let monitor_cron = cron::Schedule::from_str(controller.monitor_cron.as_str()).unwrap();
            #[cfg(feature = "monitor")]
            let _monitor_next = monitor_cron.upcoming(chrono::Utc).next().unwrap();
            #[cfg(feature = "monitor")]
            let _monitor_following = monitor_cron.upcoming(chrono::Utc).nth(1).unwrap();

            // Explicitly drop controller
            std::mem::drop(controller);

            if pulse_check {
                drop(E::spawn_from_executor(executor.as_ref(), async move {
                    tracing::info!("Pulse check service starting...");

                    let start_duration = match u32::try_from(match pulse_check_next.signed_duration_since(chrono::Utc::now()).to_std() {
                        Ok(d) => d,
                        Err(_) => Duration::new(1, 0)
                    }.as_millis()) {
                        Ok(u) => u,
                        Err(_) => panic!("Period defined is too large for timer")
                    };
                    let interval_duration = match u32::try_from(match pulse_check_following.signed_duration_since(pulse_check_next).to_std() {
                        Ok(d) => d,
                        Err(_) => Duration::new(1, 0)
                    }.as_millis()) {
                        Ok(u) => u,
                        Err(_) => panic!("Period defined is too large for timer")
                    };

                    let mut delay = E::once(start_duration).await;
                    let mut interval = E::interval(interval_duration).await;

                    delay.tick().await;
                    loop {
                        interval.tick().await;
                        tracing::debug!("Pulse check service running");

                        pulse_run(controller_mutex.clone()).await;
                    }
                }));
            }

            #[cfg(feature = "monitor")]
            if monitor {
                drop(E::spawn_from_executor(executor.as_ref(), async move {
                    tracing::info!("Monitor service starting...");

                    let start_duration = match u32::try_from(match pulse_check_next.signed_duration_since(chrono::Utc::now()).to_std() {
                        Ok(d) => d,
                        Err(_) => Duration::new(1, 0)
                    }.as_millis()) {
                        Ok(u) => u,
                        Err(_) => panic!("Period defined is too large for timer")
                    };
                    let interval_duration = match u32::try_from(match pulse_check_following.signed_duration_since(pulse_check_next).to_std() {
                        Ok(d) => d,
                        Err(_) => Duration::new(1, 0)
                    }.as_millis()) {
                        Ok(u) => u,
                        Err(_) => panic!("Period defined is too large for timer")
                    };

                    let mut delay = E::once(start_duration).await;
                    let mut interval = E::interval(interval_duration).await;

                    delay.tick().await;
                    loop {
                        interval.tick().await;
                        tracing::debug!("Monitor service running");

                        let _ = product_os_monitoring::process_statistics_info(None);
                    }
                }));
            }
        }
        None => tracing::error!("Failed to lock controller")
    }
}


/// Announces the local node to all other nodes in the cluster.
///
/// Performs key exchange and sends the local node's information to each
/// remote node in the registry.
#[cfg(feature = "distributed")]
pub async fn perform_announce(controller: &mut MutexGuard<'_, ProductOSController>) {
    tracing::info!("Announce searching for nodes...");

    authentication::perform_key_exchange(controller).await;

    let self_identifier = controller.registry.get_me().get_identifier();
    #[allow(deprecated)]
    let control_url = controller.registry.get_me().get_address();

    let nodes = controller.registry.get_nodes_endpoints(0, true);

    tracing::info!("Starting announce...");
    tracing::info!("Nodes data {:?}", nodes);

    for (identifier, (url, key)) in nodes {
        if url != control_url {
            match key {
                Some(verify_key) => {
                    tracing::info!("Announcing {}: {}", identifier, url);

                    match commands::command(&controller.client, url.clone(), verify_key, "status", "announce", Some(serde_json::value::to_value(controller.get_registry().get_me()).unwrap())).await {
                        Ok(response) => {
                            let Some(status) = response.try_status() else {
                                tracing::error!("Failed to get status from announce response for {}", url);
                                continue;
                            };

                            match status {
                                product_os_request::StatusCode::UNAUTHORIZED => {
                                    let text = controller.client.text(response).await.unwrap();
                                    let auth: authentication::CommandControlAuthenticateError = match serde_json::from_str(text.as_str()) {
                                        Ok(auth_error) => auth_error,
                                        Err(_) => authentication::CommandControlAuthenticateError { error: authentication::CommandControlAuthenticateErrorState::None }
                                    };

                                    tracing::error!("Error object auth {:?}", auth);

                                    match auth.error {
                                        authentication::CommandControlAuthenticateErrorState::KeyError(_) => {
                                            tracing::info!("Error from remote node - keys don't match {}", identifier);
                                        },
                                        authentication::CommandControlAuthenticateErrorState::None => ()
                                    };
                                },
                                product_os_request::StatusCode::OK => {
                                    let body = controller.client.bytes(response).await.unwrap();
                                    tracing::info!("Response received from {}: {} {:?}", url, status, body);
                                }
                                _ => {
                                    let body = controller.client.bytes(response).await.unwrap();
                                    tracing::error!("Error response received from {}: {} {:?}", url, status, body);
                                }
                            }
                        },
                        Err(e) => {
                            tracing::error!("Error encountered {:?} from {}", e, url);
                        }
                    };
                },
                None => ()
            }
        }
        else {
            if identifier != self_identifier {
                tracing::info!("Attempting to purge self duplicate node - identifier mismatch {}", identifier);
                controller.registry.remove_node(identifier.as_str()).await;
            }
        }
    }

    tracing::info!("Finished announcing...");
}