host_application.hpp

#pragma once
 
#include "detail/compat.hpp"
#include "detail/secure_buffer.hpp"
#include "logging.hpp"
#include "mqtt_handle.hpp"
#include "payload_builder.hpp"
#include "sparkplug_b.pb.h"
#include "topic.hpp"
 
#include <atomic>
#include <functional>
#include <memory>
#include <mutex>
#include <optional>
#include <span>
#include <string>
#include <unordered_map>
#include <utility>
#include <vector>
 
#include <MQTTAsync.h>
 
namespace sparkplug {
 
using MessageCallback =
    std::function<void(const Topic&, const org::eclipse::tahu::protobuf::Payload&)>;
 
class HostApplication {
public:
  struct TlsOptions {
    std::string trust_store; 
    std::string key_store;   
    std::string private_key; 
    std::string private_key_password; 
    std::string
        enabled_cipher_suites; 
    bool enable_server_cert_auth = true; 
    bool verify_hostname = true; 
  };
 
  struct DeviceState {
    bool is_online{false};         
    uint64_t last_seq{255};        
    bool birth_received{false};    
    uint64_t offline_timestamp{0}; 
    bool metrics_stale{false};     
    std::unordered_map<uint64_t, std::string>
        alias_map; 
  };
 
  struct TransparentStringHash {
    using is_transparent = void;
    using hash_type = std::hash<std::string_view>;
    [[nodiscard]] size_t operator()(std::string_view str) const noexcept {
      return hash_type{}(str);
    }
    [[nodiscard]] size_t operator()(const std::string& str) const noexcept {
      return hash_type{}(str);
    }
  };
 
  struct NodeState {
    bool is_online{false};       
    uint64_t last_seq{255};      
    uint64_t bd_seq{0};          
    uint64_t birth_timestamp{0}; 
    bool birth_received{false};  
    std::unordered_map<std::string, DeviceState, TransparentStringHash, std::equal_to<>>
        devices; 
    std::unordered_map<uint64_t, std::string>
        alias_map; 
  };
 
  struct Config {
    std::string broker_url; 
    std::string client_id;  
    std::string host_id;    
    int qos = 1;            
    bool clean_session =
        true; 
    int keep_alive_interval = 60; 
    int max_inflight = 100; 
    bool validate_sequence =
        true; 
    std::optional<TlsOptions>
        tls{}; 
    std::optional<std::string>
        username{}; 
    std::optional<std::string>
        password{};                     
    MessageCallback message_callback{}; 
    LogCallback log_callback{};         
    size_t max_payload_bytes = 1 << 20; 
    size_t max_tracked_nodes = 4096; 
    size_t max_metrics_per_birth = 10000; 
  };
 
  HostApplication(Config config);
 
  ~HostApplication();
 
  HostApplication(const HostApplication&) = delete;
  HostApplication& operator=(const HostApplication&) = delete;
  HostApplication(HostApplication&&) noexcept;
  HostApplication& operator=(HostApplication&&) noexcept;
 
  void set_credentials(std::optional<std::string> username,
                       std::optional<std::string> password);
 
  void set_tls(std::optional<TlsOptions> tls);
 
  void set_message_callback(MessageCallback callback);
 
  void set_log_callback(LogCallback callback);
 
  [[nodiscard]] stdx::expected<void, std::string> connect();
 
  [[nodiscard]] stdx::expected<void, std::string> disconnect();
 
  [[nodiscard]] stdx::expected<void, std::string> subscribe_all_groups();
 
  [[nodiscard]] stdx::expected<void, std::string>
  subscribe_group(std::string_view group_id);
 
  [[nodiscard]] stdx::expected<void, std::string>
  subscribe_node(std::string_view group_id, std::string_view edge_node_id);
 
  [[nodiscard]] stdx::expected<void, std::string>
  subscribe_state(std::string_view host_id);
 
  struct NodeStateSnapshot {
    bool is_online{false};
    uint64_t last_seq{255};
    uint64_t bd_seq{0};
    uint64_t birth_timestamp{0};
    bool birth_received{false};
  };
 
  [[nodiscard]] std::optional<NodeStateSnapshot>
  get_node_state(std::string_view group_id, std::string_view edge_node_id) const;
 
  [[nodiscard]] std::optional<std::string> get_metric_name(std::string_view group_id,
                                                           std::string_view edge_node_id,
                                                           std::string_view device_id,
                                                           uint64_t alias) const;
 
  [[nodiscard]] stdx::expected<void, std::string> publish_state_birth(uint64_t timestamp);
 
  [[nodiscard]] stdx::expected<void, std::string> publish_state_death(uint64_t timestamp);
 
  [[nodiscard]] stdx::expected<void, std::string>
  publish_node_command(std::string_view group_id,
                       std::string_view target_edge_node_id,
                       PayloadBuilder& payload);
 
  [[nodiscard]] stdx::expected<void, std::string>
  publish_device_command(std::string_view group_id,
                         std::string_view target_edge_node_id,
                         std::string_view target_device_id,
                         PayloadBuilder& payload);
 
  void log(LogLevel level, std::string_view message) const noexcept;
 
private:
  Config config_;
  MQTTAsyncHandle client_;
  std::atomic<bool> is_connected_{false};
 
  // MQTT connection options that must outlive async operations
  MQTTAsync_SSLOptions ssl_opts_{};
 
  // Secrets adopted from Config into scrub-on-free buffers so plaintext does
  // not linger in std::string heap (CWE-316).
  detail::SecureBuffer password_;
  detail::SecureBuffer tls_key_password_;
 
  // Node state tracking
  struct NodeKey {
    std::string group_id;
    std::string edge_node_id;
 
    [[nodiscard]] bool operator==(const NodeKey& other) const noexcept {
      return group_id == other.group_id && edge_node_id == other.edge_node_id;
    }
  };
 
  struct NodeKeyHash {
    using is_transparent = void;
    [[nodiscard]] size_t operator()(const NodeKey& key) const noexcept {
      size_t h1 = std::hash<std::string>{}(key.group_id);
      size_t h2 = std::hash<std::string>{}(key.edge_node_id);
      return h1 ^ (h2 << 1);
    }
    [[nodiscard]] size_t
    operator()(std::pair<std::string_view, std::string_view> key) const noexcept {
      size_t h1 = std::hash<std::string_view>{}(key.first);
      size_t h2 = std::hash<std::string_view>{}(key.second);
      return h1 ^ (h2 << 1);
    }
  };
 
  struct NodeKeyEqual {
    using is_transparent = void;
    [[nodiscard]] bool operator()(const NodeKey& lhs, const NodeKey& rhs) const noexcept {
      return lhs == rhs;
    }
    [[nodiscard]] bool
    operator()(const NodeKey& lhs,
               std::pair<std::string_view, std::string_view> rhs) const noexcept {
      return lhs.group_id == rhs.first && lhs.edge_node_id == rhs.second;
    }
    [[nodiscard]] bool operator()(std::pair<std::string_view, std::string_view> lhs,
                                  const NodeKey& rhs) const noexcept {
      return lhs.first == rhs.group_id && lhs.second == rhs.edge_node_id;
    }
  };
 
  std::unordered_map<NodeKey, NodeState, NodeKeyHash, NodeKeyEqual> node_states_;
  mutable std::mutex node_states_mutex_; // Protects node_states_ only
 
  // State-tracking caps copied from Config at construction; written only under
  // both mutexes (move), read under node_states_mutex_ in validate_message.
  size_t max_tracked_nodes_{4096};
  size_t max_metrics_per_birth_{10000};
 
  // Mutex for thread-safe access to config and other mutable state
  mutable std::mutex mutex_;
 
  // Caller must hold mutex_ (or be the constructor). Zeroes the Config
  // string copies after adopting them into password_/tls_key_password_.
  void adopt_secrets();
 
  [[nodiscard]] stdx::expected<void, std::string>
  publish_raw_message(std::string_view topic,
                      std::span<const uint8_t> payload_data,
                      int qos,
                      bool retain);
 
  [[nodiscard]] stdx::expected<void, std::string>
  publish_command_message(std::string_view topic, std::span<const uint8_t> payload_data);
 
  // Deferred logs: validate_message runs under node_states_mutex_ and must not
  // invoke the log callback there (user loggers may call public getters that
  // take the same mutex — deadlock, CWE-667). Caller flushes after unlock.
  bool validate_message(const Topic& topic,
                        const org::eclipse::tahu::protobuf::Payload& payload,
                        std::vector<std::pair<LogLevel, std::string>>& deferred_logs);
 
  // Static MQTT callback for message arrived
  static int on_message_arrived(void* context,
                                char* topicName,
                                int topicLen,
                                MQTTAsync_message* message);
 
  static void on_connection_lost(void* context, char* cause);
};
 
} // namespace sparkplug