edge_node.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 <mutex>
#include <optional>
#include <span>
#include <string>
#include <unordered_map>
#include <vector>
 
#include <MQTTAsync.h>
 
namespace sparkplug {
 
using CommandCallback =
    std::function<void(const Topic&, const org::eclipse::tahu::protobuf::Payload&)>;
 
class EdgeNode {
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 Config {
    std::string broker_url;   
    std::string client_id;    
    std::string group_id;     
    std::string edge_node_id; 
    int data_qos = 0;         
    int death_qos = 1;        
    bool clean_session = true; 
    int keep_alive_interval =
        60; 
    std::optional<TlsOptions>
        tls{}; 
    std::optional<std::string>
        username{}; 
    std::optional<std::string>
        password{}; 
    std::optional<CommandCallback> command_callback{};
    std::optional<std::string> primary_host_id{};
    std::optional<LogCallback> log_callback{};
    size_t max_payload_bytes = 1 << 20; 
  };
 
  EdgeNode(Config config);
 
  ~EdgeNode();
 
  EdgeNode(const EdgeNode&) = delete;
  EdgeNode& operator=(const EdgeNode&) = delete;
  EdgeNode(EdgeNode&&) noexcept;
  EdgeNode& operator=(EdgeNode&&) noexcept;
 
  void set_credentials(std::optional<std::string> username,
                       std::optional<std::string> password);
 
  void set_tls(std::optional<TlsOptions> tls);
 
  void set_log_callback(std::optional<LogCallback> callback);
 
  [[nodiscard]] stdx::expected<void, std::string> connect();
 
  [[nodiscard]] stdx::expected<void, std::string> disconnect();
 
  [[nodiscard]] stdx::expected<void, std::string> publish_birth(PayloadBuilder& payload);
 
  [[nodiscard]] stdx::expected<void, std::string> publish_data(PayloadBuilder& payload);
 
  [[nodiscard]] stdx::expected<void, std::string> publish_death();
 
  [[nodiscard]] stdx::expected<void, std::string> rebirth();
 
  [[nodiscard]] uint64_t get_seq() const {
    std::scoped_lock lock(mutex_);
    return seq_num_;
  }
 
  [[nodiscard]] uint64_t get_bd_seq() const {
    std::scoped_lock lock(mutex_);
    return bd_seq_num_;
  }
 
  [[nodiscard]] bool is_primary_host_online() const {
    std::scoped_lock lock(mutex_);
    return primary_host_online_;
  }
 
  [[nodiscard]] stdx::expected<void, std::string>
  publish_device_birth(std::string_view device_id, PayloadBuilder& payload);
 
  [[nodiscard]] stdx::expected<void, std::string>
  publish_device_data(std::string_view device_id, PayloadBuilder& payload);
 
  [[nodiscard]] stdx::expected<void, std::string>
  publish_device_death(std::string_view device_id);
 
  [[nodiscard]] stdx::expected<void, std::string>
  publish_node_command(std::string_view target_edge_node_id, PayloadBuilder& payload);
 
  [[nodiscard]] stdx::expected<void, std::string>
  publish_device_command(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:
  struct DeviceState {
    std::vector<uint8_t> last_birth_payload; // Last DBIRTH for rebirth
    bool is_online{false};                   // True if DBIRTH sent and device online
  };
 
  Config config_;
  MQTTAsyncHandle client_;
  uint64_t seq_num_{0};    // Node message sequence (0-255)
  uint64_t bd_seq_num_{0}; // Birth/Death sequence
 
  // Store the NDEATH payload for the MQTT Will
  std::vector<uint8_t> death_payload_data_;
  std::string death_topic_str_; // Topic string for MQTT Will (must outlive async connect)
  MQTTAsync_willOptions will_opts_; // Will options struct (must outlive async connect)
  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_;
 
  // Store last NBIRTH for rebirth command
  std::vector<uint8_t> last_birth_payload_;
 
  // Hash and equality functors that support heterogeneous lookup (string_view)
  struct StringHash {
    using is_transparent = void;
    [[nodiscard]] size_t operator()(std::string_view sv) const noexcept {
      return std::hash<std::string_view>{}(sv);
    }
  };
 
  struct StringEqual {
    using is_transparent = void;
    [[nodiscard]] bool operator()(std::string_view lhs,
                                  std::string_view rhs) const noexcept {
      return lhs == rhs;
    }
  };
 
  // Track state of attached devices (device_id -> state, with heterogeneous lookup)
  std::unordered_map<std::string, DeviceState, StringHash, StringEqual> device_states_;
 
  std::atomic<bool> is_connected_{false};
  std::atomic<bool> primary_host_online_{
      false}; // True if primary host is online (or no primary host configured)
 
  // Mutex for thread-safe access to all 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]] static stdx::expected<void, std::string>
  publish_message(MQTTAsync client,
                  const std::string& topic_str,
                  std::span<const uint8_t> payload_data,
                  int qos,
                  bool retain);
 
  // Static MQTT callback for message arrived (NCMD)
  static int on_message_arrived(void* context,
                                char* topicName,
                                int topicLen,
                                MQTTAsync_message* message);
 
  // Static MQTT callback for connection lost
  static void on_connection_lost(void* context, char* cause);
};
 
} // namespace sparkplug