product-os-command-control 0.0.27

//! # 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_configuration::Configuration;
//! use product_os_security::certificates::Certificates;
//!
//! # async fn example() {
//! // Create configuration and certificates
//! let config = Configuration::new();
//! let certs = Certificates::new_self_signed(
//!     vec![("CN".to_string(), "localhost".to_string())],
//!     None, None, None, None, None,
//! );
//!
//! // Initialize the controller
//! let controller = ProductOSController::new(
//!     config,
//!     certs,
//!     None, // Optional key-value store
//! );
//!
//! // Access the registry
//! 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

#![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)]

extern crate no_std_compat as std;

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

pub mod authentication;
mod registry;
pub mod commands;

use std::collections::BTreeMap;

use std::sync::Arc;
use parking_lot::Mutex;
#[cfg(feature = "relational_store")]
use product_os_configuration::{KeyValueKind, KeyValueStore, RelationalKind, RelationalStore};
#[cfg(not(feature = "relational_store"))]
use product_os_configuration::{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;



/// 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_configuration::Configuration;
/// use product_os_security::certificates::Certificates;
///
/// let config = Configuration::new();
/// 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);
/// ```
pub struct ProductOSController {
    configuration: product_os_configuration::Configuration,

    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>,
}

impl ProductOSController {
    /// Creates a new `ProductOSController` instance.
    ///
    /// Initializes the command and control system with the provided configuration,
    /// certificates, and optional data stores.
    ///
    /// # Arguments
    ///
    /// * `configuration` - System configuration including network and service settings
    /// * `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)
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use product_os_command_control::ProductOSController;
    /// use product_os_configuration::Configuration;
    /// use product_os_security::certificates::Certificates;
    ///
    /// let config = Configuration::new();
    /// 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);
    /// ```
    pub fn new(configuration: product_os_configuration::Configuration, 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::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,
        })));

        #[cfg(feature = "relational_store")]
        let relational_store: Arc<product_os_store::ProductOSRelationalStore> = relational_store.unwrap_or_else(|| Arc::new(product_os_store::ProductOSRelationalStore::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,
        })));

        let registry = registry::Registry::new(&configuration, 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: configuration.get_cc_max_servers(),

            pulse_check: configuration.is_pulse_check_enabled(),
            pulse_check_cron: configuration.get_hearbeat_cron().to_string(),

            #[cfg(feature = "monitor")]
            monitor: configuration.is_monitor_enabled(),
            #[cfg(feature = "monitor")]
            monitor_cron: configuration.get_monitor_cron().to_string(),

            configuration,

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

            registry,

            requester,
            client: request_client
        }
    }

    /// 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.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # use product_os_command_control::ProductOSController;
    /// # use product_os_configuration::Configuration;
    /// # use product_os_security::certificates::Certificates;
    /// # let controller = ProductOSController::new(
    /// #     Configuration::new(),
    /// #     Certificates::new_self_signed(
    /// #         vec![("CN".to_string(), "localhost".to_string())],
    /// #         None, None, None, None, None,
    /// #     ),
    /// #     None,
    /// # );
    /// let registry = controller.get_registry();
    /// let my_node = registry.get_me();
    /// println!("My node ID: {}", my_node.get_identifier());
    /// ```
    pub fn get_registry(&self) -> &registry::Registry {
        &self.registry
    }

    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_pem(cert);
        }

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

    pub fn get_max_servers(&self) -> u8 {
        self.max_servers.to_owned()
    }

    pub fn upsert_node_local(&mut self, identifier: String, node: Node) {
        self.registry.upsert_node_local(identifier, node);
    }

    pub fn get_certificates(&self) -> Vec<&[u8]> {
        let mut certificates = vec!();
        for cert in &self.certificates.certificates {
            certificates.push(cert.as_slice())
        }

        certificates
    }

    pub fn get_private_key(&self) -> &[u8] {
        self.certificates.private.as_slice()
    }

    pub fn get_certificates_and_private_key(&self) -> product_os_security::certificates::Certificates {
        self.certificates.to_owned()
    }

    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
    }

    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
        }
    }

    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(String::from("One of the keys provided was not valid"))
            })
        }
    }

    /*
    async fn search_and_command(&self, query: BTreeMap<String, String>, module: String, instruction: String, data: Option<serde_json::Value>)  -> Result<ProductOSResponse, 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()) {
                    None => Err(product_os_request::ProductOSRequestError::Error(format!("No key found for matching node"))),
                    Some(key) => commands::command_node(client, node, key, module, instruction, data).await
                }
            },
            None => Err(product_os_request::ProductOSRequestError::Error(format!("No matching node found")))
        }
    }
    */

    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(format!("No key found for matching node"))),
                    Some(key) => Ok(commands::Command {
                        requester: client.clone(),
                        node_url: node.get_address().to_owned(),
                        verify_key: key,
                        module,
                        instruction,
                        data
                    })
                }
            },
            None => Err(product_os_request::ProductOSRequestError::Error(format!("No matching node found")))
        }
    }

    /*
    async fn find_and_command(&self, key: String, manager: String, instruction: String, data: Option<serde_json::Value>) -> Result<ProductOSResponse, product_os_request::ProductOSRequestError> {
        let query = BTreeMap::from([
            ("manager.key".to_string(), key),
            ("manager.kind".to_string(), manager.to_string()),
            ("manager.enabled".to_string(), true.to_string())
        ]);

        self.search_and_command(query, manager.to_string(), instruction, data).await
    }
    */

    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 enabled = true.to_string();

        let query = BTreeMap::from([
            ("manager.key", key),
            ("manager.kind", manager),
            ("manager.enabled", enabled.as_str())
        ]);

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

    /*
    async fn find_any_and_command(&self, capability: String, manager: String, instruction: String, data: Option<serde_json::Value>) -> Result<ProductOSResponse, product_os_request::ProductOSRequestError> {
        let query = BTreeMap::from([
            ("capability".to_string(), capability),
            ("manager.enabled".to_string(), true.to_string())
        ]);

        self.search_and_command(query, manager.to_string(), instruction, data).await
    }
    */

    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 enabled = true.to_string();

        let query = BTreeMap::from([
            ("capability", capability),
            ("manager.enabled", enabled.as_str())
        ]);

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

    /*
    async fn find_kind_and_command(&self, kind: String, manager: String, instruction: String, data: Option<serde_json::Value>) -> Result<ProductOSResponse, product_os_request::ProductOSRequestError> {
        let query = BTreeMap::from([
            ("kind".to_string(), kind),
            ("manager.enabled".to_string(), true.to_string())
        ]);

        self.search_and_command(query, manager.to_string(), instruction, data).await
    }
    */

    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> {
        let enabled = true.to_string();
        let query = BTreeMap::from([
            ("kind", kind),
            ("manager.enabled", enabled.as_str())
        ]);

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

    /*
    async fn search_and_ask(&self, query: BTreeMap<String, String>, path: String, data: Option<serde_json::Value>, headers: BTreeMap<String, String>,
                          params: BTreeMap<String, String>, method: product_os_request::Method) -> Result<ProductOSResponse, 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()) {
                    None => Err(product_os_request::ProductOSRequestError::Error(format!("No key found for matching node"))),
                    Some(key) => commands::ask_node(client, node, key, path, data, headers, params, method).await
                }
            },
            None => Err(product_os_request::ProductOSRequestError::Error(format!("No matching node found")))
        }
    }
    */

    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(format!("No key found for matching node"))),
                    Some(key) => Ok(commands::Ask {
                        requester: client.clone(),
                        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(format!("No matching node found")))
        }
    }

    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(format!("No key found for matching node"))),
                    Some(key) => commands::ask_node(client, node, key.as_slice(), path, data, headers, params, method).await
                }

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

    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(format!("No key found for matching node"))),
                    Some(key) => Ok(commands::Ask {
                        requester: client.clone(),
                        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(format!("No matching node found")))
        }
    }

    /*
    pub fn find_feature_and_ask_sync(&mut self, feature: String, path: String, data: Option<serde_json::Value>, headers: BTreeMap<String, String>,
                                      params: BTreeMap<String, String>, method: product_os_request::Method) -> Result<product_os_request::ProductOSResponse, product_os_request::ProductOSRequestSyncError> {
        let matching_node = self.registry.pick_node_for_feature(feature);
        let client = &mut self.client;
        match matching_node {
            Some(node) => {
                match self.registry.get_key(node.get_identifier()) {
                    None => Err(product_os_request::ProductOSRequestSyncError {
                        error: product_os_request::ProductOSRequestError::Error("No key found for matching node".to_string()),
                        generated_error: None
                    }),
                    Some(key) => commands::ask_node_sync(client, node, key, path, data, headers, params, method)
                }

            },
            None => Err(product_os_request::ProductOSRequestSyncError {
                error: product_os_request::ProductOSRequestError::Error("No matching node found".to_string()),
                generated_error: None
            })
        }
    }
    */

    pub fn get_configuration(&self) -> product_os_configuration::Configuration {
        self.configuration.clone()
    }

    pub fn get_key(&self, identifier: &str) -> Option<Vec<u8>> {
        self.registry.get_key(identifier)
    }

    pub fn create_key_session(&mut self) -> (String, [u8; 32]) {
        self.registry.create_key_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);
    }

    #[cfg(feature = "relational_store")]
    pub fn get_relational_store(&mut self) -> Arc<product_os_store::ProductOSRelationalStore> {
        self.relational_store.clone()
    }

    pub fn get_key_value_store(&mut self) -> Arc<product_os_store::ProductOSKeyValueStore> {
        self.key_value_store.clone()
    }

    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;
    }

    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;
    }

    pub async fn remove_feature(&mut self, identifier: &str) {
        self.registry.remove_feature(identifier).await;
    }

    pub async fn add_service(&mut self, service: Arc<dyn product_os_capabilities::Service>) {
        self.registry.add_service(service).await;
    }

    pub async fn add_service_mut(&mut self, service: Arc<Mutex<dyn product_os_capabilities::Service>>) {
        self.registry.add_service_mut(service).await;
    }

    pub async fn set_service_active(&mut self, identifier: String, status: bool) {
        self.registry.set_service_active(identifier, status).await;
    }

    pub async fn remove_service(&mut self, identifier: &str) {
        self.registry.remove_service(identifier).await;
    }

    pub async fn remove_inactive_services(&mut self, query: BTreeMap<&str, &str>) {
        self.registry.remove_inactive_services(query).await;
    }

    pub async fn start_services(&mut self) -> Result<(), ()> {
        self.registry.start_services().await
    }

    pub async fn start_service(&mut self, identifier: &str) -> Result<(), ()> {
        self.registry.start_service(identifier).await
    }

    pub async fn stop_service(&mut self, identifier: &str) -> Result<(), ()> {
        self.registry.stop_service(identifier).await
    }


}





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 {
        Some(controller) => {
            #[cfg(feature = "distributed")]
            controller.discover_nodes().await;

            #[cfg(feature = "distributed")]
            let self_identifier = controller.registry.get_me().get_identifier();
            #[cfg(feature = "distributed")]
            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");
                            std::process::exit(8);
                        };
                    }
                }
            }

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

            #[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 status = response.status();

                                    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")
    }
}


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
            match &controller.configuration.command_control {
                Some(command_control_config) => {
                    if command_control_config.auto_start_services {
                        match controller.start_services().await {
                            Ok(_) => {
                                tracing::info!("Command Control services started");
                            },
                            Err(_) => {
                                tracing::error!("Command Control services failed to start");
                            }
                        }
                    }
                }
                None => {
                    tracing::error!("Command Control configuration not enabled to start services");
                }
            }

            let pulse_check = controller.pulse_check.clone();
            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).nth(0).unwrap();
            let pulse_check_following = pulse_check_cron.upcoming(chrono::Utc).nth(1).unwrap();

            #[cfg(feature = "monitor")]
            let monitor = controller.monitor.clone();
            #[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).nth(0).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 {
                let _ = 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 {
                let _ = 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");

                        product_os_monitoring::process_statistics(None);
                    }
                }); // await on the last loop
            }
        }
        None => tracing::error!("Failed to lock controller")
    }
}


#[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();
    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 status = response.status();

                            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...");
}