mesh.hpp

Go to the documentation of this file.

#ifndef _PAINLESS_MESH_MESH_HPP_
#define _PAINLESS_MESH_MESH_HPP_
 
#include <algorithm>
#include <list>
#include <vector>
#include <map>
#include <set>
#include <queue>
#include <type_traits>
 
#include "painlessmesh/configuration.hpp"
 
#include "painlessmesh/ack.hpp"
#include "painlessmesh/connection.hpp"
#include "painlessmesh/gateway.hpp"
#include "painlessmesh/logger.hpp"
#include "painlessmesh/message_queue.hpp"
#include "painlessmesh/ntp.hpp"
#include "painlessmesh/plugin.hpp"
#include "painlessmesh/protocol.hpp"
#include "painlessmesh/rtc.hpp"
#include "painlessmesh/tcp.hpp"
#include "painlessmesh/validation.hpp"
 
#ifdef PAINLESSMESH_ENABLE_OTA
#include "painlessmesh/ota.hpp"
#endif
 
namespace painlessmesh {
typedef std::function<void(uint32_t nodeId)> newConnectionCallback_t;
typedef std::function<void(uint32_t nodeId)> droppedConnectionCallback_t;
typedef std::function<void(uint32_t from, TSTRING &msg)> receivedCallback_t;
typedef std::function<void()> changedConnectionsCallback_t;
typedef std::function<void(int32_t offset)> nodeTimeAdjustedCallback_t;
typedef std::function<void(uint32_t nodeId, int32_t delay)> nodeDelayCallback_t;
typedef std::function<void(uint32_t bridgeNodeId, bool internetAvailable)> bridgeStatusChangedCallback_t;
typedef std::function<void(uint32_t timestamp)> rtcSyncCompleteCallback_t;
typedef std::function<void(bool available)> localInternetChangedCallback_t;
typedef std::function<void(uint32_t oldPrimary, uint32_t newPrimary)> gatewayChangedCallback_t;
 
struct SendOptions {
  uint8_t priority = protocol::PRIORITY_NORMAL;
  ack::deliveryCallback_t ackCallback = nullptr;
  uint32_t ackTimeoutMs = 5000;
};
 
typedef std::function<void(bool success, uint16_t httpStatus, TSTRING error)>
    internetResultCallback_t;
 
#define PAINLESSMESH_HAS_INTERNET_RESULT 1
 
struct InternetResult {
  uint32_t messageId = 0;
  bool success = false;
  uint16_t httpStatus = 0;
  TSTRING error;
  TSTRING response;
  bool retryable = false;
  uint8_t attempts = 0;
};
 
typedef std::function<void(const InternetResult& result)>
    internetResponseCallback_t;
 
struct PendingInternetRequest {
  uint32_t messageId = 0;          
  uint32_t timestamp = 0;          
  uint8_t retryCount = 0;          
  uint8_t maxRetries = 3;          
  uint8_t priority = 2;            
  uint32_t timeoutMs = 30000;      
  uint32_t retryDelayMs = 1000;    
  uint32_t gatewayNodeId = 0;      
  TSTRING destination = "";        
  TSTRING payload = "";            
  internetResultCallback_t callback;  
  internetResponseCallback_t onResult;  
  uint8_t attempts = 0;
  uint32_t requestNonce = 0;
  TSTRING lastError = "";
 
  bool isTimedOut() const {
    return (static_cast<uint32_t>(millis()) - timestamp) > timeoutMs;
  }
};
 
class BridgeInfo {
public:
  uint32_t nodeId = 0;              // Bridge node ID
  bool internetConnected = false;   // Is bridge connected to Internet?
  int8_t routerRSSI = 0;           // Router WiFi signal strength in dBm
  uint8_t routerChannel = 0;       // Router WiFi channel
  uint32_t lastSeen = 0;           // Timestamp when last status received (millis)
  uint32_t uptime = 0;             // Bridge uptime in milliseconds
  TSTRING gatewayIP = "";          // Router gateway IP address
  uint32_t timestamp = 0;          // Timestamp from bridge status message
  
  bool isHealthy(uint32_t timeoutMs = 60000) const {
    // Cast millis() to uint32_t to handle overflow correctly
    // This ensures consistent behavior on both embedded and test environments
    return (static_cast<uint32_t>(millis()) - lastSeen) < timeoutMs;
  }
};
 
struct BridgeHealthMetrics {
  // Connectivity
  uint32_t uptimeSeconds = 0;
  uint32_t internetUptimeSeconds = 0;
  uint32_t totalDisconnects = 0;
  uint32_t currentUptime = 0;
  
  // Signal Quality
  int8_t currentRSSI = 0;
  int8_t avgRSSI = 0;
  int8_t minRSSI = 0;
  int8_t maxRSSI = -127;
  
  // Traffic
  uint64_t bytesRx = 0;
  uint64_t bytesTx = 0;
  uint32_t messagesRx = 0;
  uint32_t messagesTx = 0;
  uint32_t messagesQueued = 0;
  uint32_t messagesDropped = 0;
  
  // Performance
  uint32_t avgLatencyMs = 0;
  uint8_t packetLossPercent = 0;
  uint32_t meshNodeCount = 0;
};
 
struct BridgeStatus {
  bool isBridge = false;              // Is this node acting as a bridge?
  bool internetConnected = false;     // Is Internet connection available?
  TSTRING role = "regular";           // Role: "regular", "bridge", "root"
  uint32_t bridgeNodeId = 0;          // Current bridge node ID (0 if none)
  int8_t bridgeRSSI = 0;             // Signal strength to bridge/router (dBm)
  uint32_t timeSinceBridgeChange = 0; // Time since last bridge change (ms)
};
 
struct ElectionRecord {
  uint32_t timestamp = 0;          // When election occurred (millis)
  uint32_t winnerNodeId = 0;       // Node that won election
  int8_t winnerRSSI = 0;          // Winner's router RSSI
  uint32_t candidateCount = 0;    // Number of candidates
  TSTRING reason = "";             // Why election was triggered
};
 
struct BridgeChangeEvent {
  uint32_t timestamp = 0;            // When change occurred (millis)
  uint32_t oldBridgeId = 0;         // Previous bridge node ID
  uint32_t newBridgeId = 0;         // New bridge node ID
  TSTRING reason = "";               // Reason for change
  bool internetAvailable = false;    // Internet available after change
};
 
struct BridgeTestResult {
  bool success = false;              // Overall test success
  bool bridgeReachable = false;      // Can reach bridge node
  bool internetReachable = false;    // Can reach Internet (if bridge available)
  uint32_t latencyMs = 0;           // Round-trip latency to bridge
  TSTRING message = "";              // Detailed test message
};
 
template <class T>
class Mesh : public ntp::MeshTime, public plugin::PackageHandler<T> {
 public:
  void init(uint32_t id) {
    using namespace logger;
    if (!isExternalScheduler) {
      mScheduler = new Scheduler();
    }
    this->nodeId = id;
 
#ifdef ESP32
    xSemaphore = xSemaphoreCreateMutex();
#endif
 
    // Add package handlers
    painlessmesh::ntp::addPackageCallback(this->callbackList, (*this));
    painlessmesh::router::addPackageCallback(this->callbackList, (*this));
 
    // Every node answers a gateway request it cannot serve. A bridge that
    // reboots, crashes or is reflashed as a regular node announces nothing
    // -- only a bridge stepping down in-process sends `leaving` -- so its
    // peers keep it in knownBridges until its last status ages out
    // (bridgeTimeoutMs, 60 s) and may route every Internet request to it.
    // A regular node had no GATEWAY_DATA handler and dropped them without a
    // word; the origin waited out its request timeout, 30 s, each time.
    // Found on the HIL rig while validating 2.0.3.
    //
    // Registered first, so a gateway's own handler (initGatewayInternetHandler
    // in wifi.hpp, or a test's stand-in) is always the second: this one
    // stays silent whenever any other GATEWAY_DATA handler exists.
    this->callbackList.onPackage(
        protocol::GATEWAY_DATA,
        [this](protocol::Variant& variant, std::shared_ptr<T> ingress,
               uint32_t) {
          if (this->callbackList.count(protocol::GATEWAY_DATA) > 1) {
            return false;  // this node is a gateway; its handler answers
          }
          auto pkg = variant.to<gateway::GatewayDataPackage>();
          char errorBuf[96];
          snprintf(errorBuf, sizeof(errorBuf), "Node %u %s",
                   static_cast<unsigned>(this->nodeId),
                   gateway::GATEWAY_NOT_A_GATEWAY_PHRASE);
          Log(logger::ERROR,
              "GATEWAY_DATA from node %u answered: %s\n",
              static_cast<unsigned>(pkg.originNode), errorBuf);
          this->sendGatewayAck(pkg, false, 0, TSTRING(errorBuf), ingress);
          return true;
        });
 
    // Seed delivery-confirmation ids with a random value so they do not
    // restart at 1 after a reboot — a delayed ACK for a pre-reboot
    // message could otherwise match a fresh message's id and report a
    // false delivered = true.
    ackTracker.seed(validation::SecureRandom::generate());
 
    // Per-message delivery confirmation (issue #379): reply with an ACK
    // when a received application message carries a msgId. Uses the
    // Variant field peeks — the common no-ack case must not pay for a
    // full package materialization on every received message.
    auto autoAck = [this](protocol::Variant& variant, std::shared_ptr<T>,
                          uint32_t) {
      auto msgId = variant.msgId();
      if (msgId == 0) return false;
      auto origin = variant.from();
      if (origin == 0 || origin == this->nodeId) return false;
      if (variant.type() == protocol::BROADCAST) {
        // Stagger broadcast ACKs by nodeId so N nodes do not fire N
        // unicast ACKs at the sender in the same instant
        this->queueBroadcastAck(origin, msgId);
      } else {
        auto ackPkg = ack::MessageAckPackage(this->nodeId, origin, msgId);
        this->sendPackage(&ackPkg);
      }
      return false;
    };
    this->callbackList.onPackage(protocol::SINGLE, autoAck);
    this->callbackList.onPackage(protocol::BROADCAST, autoAck);
 
    // Match incoming ACKs to messages awaiting delivery confirmation
    this->callbackList.onPackage(
        protocol::MESSAGE_ACK,
        [this](protocol::Variant& variant, std::shared_ptr<T>, uint32_t) {
          std::list<ack::DeliveryResult> results;
          this->ackTracker.collectAck(variant.msgId(), variant.from(),
                                      static_cast<uint32_t>(millis()),
                                      results);
          if (!results.empty()) {
            this->addTask([results]() {
              for (const auto& result : results) {
                result.callback(result.nodeId, result.delivered,
                                result.latencyMs);
              }
            });
          }
          return false;
        });
 
    // Add bridge status package handler (Type BRIDGE_STATUS)
    // This will be called when any node receives a bridge status broadcast
    this->callbackList.onPackage(
        protocol::BRIDGE_STATUS,
        [this](protocol::Variant& variant, std::shared_ptr<T>, uint32_t) {
          // We need to manually parse the JSON since BridgeStatusPackage is in alteriom namespace
          // and may not be available in all contexts. We'll parse the critical fields directly.
          JsonDocument doc;
          TSTRING str;
          variant.printTo(str);
          deserializeJson(doc, str);
          JsonObject obj = doc.as<JsonObject>();
          
          if (obj["internetConnected"].is<bool>()) {
            uint32_t bridgeNodeId = obj["from"];
            if (obj["leaving"] | false) {
              // A bridge stepping down says so, and is forgotten at once
              // rather than when its last status ages out — a minute and
              // more during which every candidate held it healthy.
              Log(GENERAL, "Bridge %u is stepping down\n", bridgeNodeId);
              this->forgetBridge(bridgeNodeId);
              return false;
            }
            bool internetConnected = obj["internetConnected"];
            int8_t routerRSSI = obj["routerRSSI"] | 0;
            uint8_t routerChannel = obj["routerChannel"] | 0;
            uint32_t uptime = obj["uptime"] | 0;
            TSTRING gatewayIP = obj["gatewayIP"].as<TSTRING>();
            uint32_t timestamp = obj["timestamp"] | 0;
            
            // Update bridge status
            this->updateBridgeStatus(bridgeNodeId, internetConnected, routerRSSI, 
                                    routerChannel, uptime, gatewayIP, timestamp);
            
            Log(GENERAL, "Bridge status received from %u: Internet %s\n",
                bridgeNodeId, internetConnected ? "Connected" : "Disconnected");
          }
          return false;  // Don't consume the package, allow other handlers
        });
 
    this->changedConnectionCallbacks.push_back([this](uint32_t nodeId) {
      Log(MESH_STATUS, "Changed connections in neighbour %u\n", nodeId);
      if (nodeId != 0) layout::syncLayout<T>((*this), nodeId);
    });
    this->droppedConnectionCallbacks.push_back([this](uint32_t nodeId,
                                                      bool station) {
      Log(MESH_STATUS, "Dropped connection %u, station %d\n", nodeId, station);
      this->metricsDisconnectCount++;
      this->eraseClosedConnections();
    });
    this->newConnectionCallbacks.push_back([](uint32_t nodeId) {
      Log(MESH_STATUS, "New connection %u\n", nodeId);
    });
  }
 
  void init(Scheduler *scheduler, uint32_t id) {
    this->setScheduler(scheduler);
    this->init(id);
  }
 
#ifdef PAINLESSMESH_ENABLE_OTA
  std::shared_ptr<Task> offerOTA(painlessmesh::plugin::ota::Announce announce) {
    auto announceTask = this->addTask(TASK_SECOND * 60, 60, [this, announce]() {
      this->sendPackage(&announce);
    });
    return announceTask;
  }
 
  std::shared_ptr<Task> offerOTA(TSTRING role, TSTRING hardware, TSTRING md5,
                                 size_t noPart, bool forced = false,
                                 bool broadcasted = false, bool compressed = false) {
    painlessmesh::plugin::ota::Announce announce;
    announce.md5 = md5;
    announce.role = role;
    announce.hardware = hardware;
    announce.from = this->nodeId;
    announce.noPart = noPart;
    announce.forced = forced;
    announce.broadcasted = broadcasted;
    announce.compressed = compressed;
    return offerOTA(announce);
  }
 
  void initOTASend(
      painlessmesh::plugin::ota::otaDataPacketCallbackType_t callback,
      size_t otaPartSize) {
    painlessmesh::plugin::ota::addSendPackageCallback(
        *this->mScheduler, (*this), callback, otaPartSize);
  }
  void initOTAReceive(TSTRING role = "",
                      std::function<void(int, int)> progress_cb = NULL) {
    painlessmesh::plugin::ota::addReceivePackageCallback(
        *this->mScheduler, (*this), role, progress_cb);
  }
#endif
 
  void setRoot(bool on = true) { this->root = on; };
 
  void setContainsRoot(bool on = true) { shouldContainRoot = on; };
 
  bool isRoot() { return this->root; };
 
  void setDebugMsgTypes(uint16_t types) { Log.setLogLevel(types); }
 
  void stop() {
    using namespace logger;
    // Close all connections
    while (this->subs.size() > 0) {
      auto conn = this->subs.begin();
      (*conn)->close();
      this->eraseClosedConnections();
    }
    // Disable the ack task BEFORE nulling the member: if stop() was
    // reached from inside the task's own callback, the lambda's
    // this->ackCheckTask guard is useless once the member is null and
    // the task would keep firing forever on an external scheduler.
    if (ackCheckTask) ackCheckTask->disable();
    if (broadcastAckTask) broadcastAckTask->disable();
    ackTracker.clear();
    pendingBroadcastAcks.clear();
    ackCheckTask = nullptr;
    broadcastAckTask = nullptr;
 
    plugin::PackageHandler<T>::stop(mScheduler);
 
    newConnectionCallbacks.clear();
    droppedConnectionCallbacks.clear();
    changedConnectionCallbacks.clear();
 
    if (!isExternalScheduler) {
      if (mScheduler && mScheduler->getCurrentTask() != nullptr) {
        // stop() was reached from inside a scheduler callback (e.g. a
        // delivery/timeout callback). Deleting the scheduler here would
        // free the object whose execute() is still on the call stack —
        // the same bug class as issue #373. Retire it until execute()
        // unwinds, then reclaim it during the next update.
        retiredSchedulers.push_back(mScheduler);
        mScheduler = nullptr;
      } else {
        delete mScheduler;
        mScheduler = nullptr;
      }
    }
  }
 
  void update(void) {
    if (semaphoreTake()) {
      plugin::PackageHandler<T>::reclaimQuarantinedTasks();
      reclaimRetiredSchedulers();
      // Check if something is executed (returns false)
      if (!mScheduler->execute())
        // DEBUG, not GENERAL: this fires every time any task runs, which on a
        // busy node is thousands of lines a second. At GENERAL it drowned the
        // level that carries the bridge and gateway diagnostics —
        // sendBridgeStatus(), "Bridge status received from %u" — so nobody
        // can turn those on to investigate a failover without being flooded.
        Log(logger::DEBUG, "update(): Scheduler executed a task\n");
      semaphoreGive();
    }
    return;
  }
 
  bool sendSingle(uint32_t destId, TSTRING msg) {
    return sendSingle(destId, msg, SendOptions());
  }
 
  bool sendSingle(uint32_t destId, TSTRING msg, uint8_t priorityLevel) {
    SendOptions options;
    options.priority = priorityLevel;
    return sendSingle(destId, msg, options);
  }
 
  bool sendSingle(uint32_t destId, TSTRING msg, const SendOptions& options) {
    Log(logger::COMMUNICATION, "sendSingle(): dest=%u msg=%s priority=%u%s\n",
        destId, msg.c_str(), options.priority,
        options.ackCallback ? " with ack" : "");
    auto single = painlessmesh::protocol::Single(this->nodeId, destId, msg);
    single.priority = options.priority;
    if (!options.ackCallback)
      return painlessmesh::router::send<T>(single, (*this));
    if (ackTracker.full()) {
      Log(logger::ERROR, "sendSingle(): pending-ack limit reached\n");
      return false;
    }
    single.msgId = ackTracker.nextMessageId();
    if (!painlessmesh::router::send<T>(single, (*this))) return false;
    ackTracker.track(single.msgId, {destId}, options.ackCallback,
                     options.ackTimeoutMs, static_cast<uint32_t>(millis()));
    this->ensureAckScheduling();
    return true;
  }
 
  bool sendSingle(uint32_t destId, TSTRING msg,
                  ack::deliveryCallback_t ackCallback,
                  uint32_t ackTimeoutMs = 5000) {
    SendOptions options;
    options.ackCallback = ackCallback;
    options.ackTimeoutMs = ackTimeoutMs;
    return sendSingle(destId, msg, options);
  }
 
  bool sendBroadcast(TSTRING msg, bool includeSelf = false) {
    return sendBroadcast(msg, SendOptions(), includeSelf);
  }
 
  bool sendBroadcast(TSTRING msg, const SendOptions& options,
                     bool includeSelf = false) {
    using namespace logger;
    Log(COMMUNICATION, "sendBroadcast(): msg=%s priority=%u%s\n", msg.c_str(),
        options.priority, options.ackCallback ? " with ack" : "");
    painlessmesh::protocol::Broadcast pkg(this->nodeId, 0, msg);
    pkg.priority = options.priority;
 
    if (!options.ackCallback) {
      auto success = router::broadcast<protocol::Broadcast, T>(pkg, (*this), 0);
      if (includeSelf) {
        protocol::Variant var(pkg);
        this->callbackList.execute(var.type(), var, NULL, 0);
      }
      return success > 0;
    }
 
    if (ackTracker.full()) {
      Log(ERROR, "sendBroadcast(): pending-ack limit reached\n");
      return false;
    }
    // Broadcast recipients intentionally jitter ACK replies across this
    // window. A shorter deadline would deterministically time out nodes
    // whose slot falls after it, even when delivery is immediate.
    auto ackTimeoutMs = options.ackTimeoutMs;
    if (ackTimeoutMs < ACK_BROADCAST_MIN_TIMEOUT_MS)
      ackTimeoutMs = ACK_BROADCAST_MIN_TIMEOUT_MS;
    auto expected = this->getNodeList(false);
    pkg.msgId = ackTracker.nextMessageId();
    auto success = router::broadcast<protocol::Broadcast, T>(pkg, (*this), 0);
    const bool trackingPeers = success > 0 && !expected.empty();
    if (trackingPeers) {
      ackTracker.track(pkg.msgId, expected, options.ackCallback, ackTimeoutMs,
                       static_cast<uint32_t>(millis()));
      this->ensureAckScheduling();
    }
    if (includeSelf) {
      protocol::Variant var(pkg);
      this->callbackList.execute(var.type(), var, NULL, 0);
    }
    // Do not access mesh state after self-delivery: an onReceive callback
    // may have called stop(). Local delivery still occurs without peers,
    // while the return value continues to describe peer ACK tracking.
    return trackingPeers;
  }
  
  bool sendBroadcast(TSTRING msg, bool includeSelf,
                     ack::deliveryCallback_t ackCallback,
                     uint32_t ackTimeoutMs = 5000) {
    SendOptions options;
    options.ackCallback = ackCallback;
    options.ackTimeoutMs = ackTimeoutMs;
    return sendBroadcast(msg, options, includeSelf);
  }
 
  size_t checkAcks() {
    std::list<ack::DeliveryResult> results;
    auto pending = ackTracker.collectExpired(static_cast<uint32_t>(millis()),
                                             results);
    if (!results.empty()) {
      this->addTask([results]() {
        for (const auto& result : results) {
          result.callback(result.nodeId, result.delivered, result.latencyMs);
        }
      });
    }
    return pending;
  }
 
  size_t pendingAcks() const { return ackTracker.pending(); }
 
  template <
      typename Priority,
      typename IncludeSelf = bool,
      typename std::enable_if<
          (std::is_integral<typename std::decay<Priority>::type>::value ||
           std::is_enum<typename std::decay<Priority>::type>::value) &&
              !std::is_same<typename std::decay<Priority>::type, bool>::value &&
              std::is_same<typename std::decay<IncludeSelf>::type, bool>::value,
          int>::type = 0>
  bool sendBroadcast(TSTRING msg, Priority priorityLevel,
                     IncludeSelf includeSelf = false) {
    SendOptions options;
    options.priority = static_cast<uint8_t>(priorityLevel);
    return sendBroadcast(msg, options, includeSelf);
  }
 
  bool startDelayMeas(uint32_t id) {
    using namespace logger;
    Log(S_TIME, "startDelayMeas(): NodeId %u\n", id);
    auto conn = painlessmesh::router::findRoute<T>((*this), id);
    if (!conn) return false;
    return router::send<protocol::TimeDelay, T>(
        protocol::TimeDelay(this->nodeId, id, this->getNodeTime()), conn);
  }
 
  void onReceive(receivedCallback_t onReceive) {
    using namespace painlessmesh;
    this->callbackList.onPackage(
        protocol::SINGLE,
        [onReceive](protocol::Variant &variant, std::shared_ptr<T>, uint32_t) {
          auto pkg = variant.to<protocol::Single>();
          onReceive(pkg.from, pkg.msg);
          return false;
        });
    this->callbackList.onPackage(
        protocol::BROADCAST,
        [onReceive](protocol::Variant &variant, std::shared_ptr<T>, uint32_t) {
          auto pkg = variant.to<protocol::Broadcast>();
          onReceive(pkg.from, pkg.msg);
          return false;
        });
  }
 
  void onNewConnection(newConnectionCallback_t onNewConnection) {
    Log(logger::GENERAL, "onNewConnection():\n");
    newConnectionCallbacks.push_back([onNewConnection](uint32_t nodeId) {
      if (nodeId != 0) onNewConnection(nodeId);
    });
  }
 
  void onDroppedConnection(droppedConnectionCallback_t onDroppedConnection) {
    droppedConnectionCallbacks.push_back(
        [onDroppedConnection](uint32_t nodeId, bool station) {
          if (nodeId != 0) onDroppedConnection(nodeId);
        });
  }
 
  void onChangedConnections(changedConnectionsCallback_t onChangedConnections) {
    Log(logger::GENERAL, "onChangedConnections():\n");
    changedConnectionCallbacks.push_back(
        [onChangedConnections](uint32_t nodeId) {
          if (nodeId != 0) onChangedConnections();
        });
  }
 
  void onNodeTimeAdjusted(nodeTimeAdjustedCallback_t onTimeAdjusted) {
    Log(logger::GENERAL, "onNodeTimeAdjusted():\n");
    nodeTimeAdjustedCallback = onTimeAdjusted;
  }
 
  void onNodeDelayReceived(nodeDelayCallback_t onDelayReceived) {
    Log(logger::GENERAL, "onNodeDelayReceived():\n");
    nodeDelayReceivedCallback = onDelayReceived;
  }
 
  void onBridgeStatusChanged(bridgeStatusChangedCallback_t onBridgeStatusChanged) {
    Log(logger::GENERAL, "onBridgeStatusChanged():\n");
    bridgeStatusChangedCallback = onBridgeStatusChanged;
  }
 
  bool hasInternetConnection() {
    // Always require healthy bridge status to prevent false positives
    // when mesh is disconnected. Stale bridge data is unreliable.
    for (const auto& bridge : knownBridges) {
      if (bridge.isHealthy(bridgeTimeoutMs) && bridge.internetConnected) {
        return true;
      }
    }
    return false;
  }
 
  std::vector<uint32_t> getNodesWithInternet() {
    std::vector<uint32_t> nodesWithInternet;
    for (const auto& bridge : knownBridges) {
      if (bridge.isHealthy(bridgeTimeoutMs) && bridge.internetConnected) {
        nodesWithInternet.push_back(bridge.nodeId);
      }
    }
    return nodesWithInternet;
  }
 
  std::vector<BridgeInfo> getBridges() {
    return knownBridges;
  }
 
  BridgeInfo* getPrimaryBridge() {
    BridgeInfo* primary = nullptr;
    int8_t bestRSSI = -127;  // Worst possible RSSI
    
    // Always require healthy bridge status to prevent routing to stale/unreachable bridges
    for (auto& bridge : knownBridges) {
      if (bridge.isHealthy(bridgeTimeoutMs) && bridge.internetConnected) {
        if (bridge.routerRSSI > bestRSSI) {
          bestRSSI = bridge.routerRSSI;
          primary = &bridge;
        }
      }
    }
    
    return primary;
  }
 
  BridgeInfo* getLastKnownBridge() {
    BridgeInfo* best = nullptr;
    int8_t bestRSSI = -127;  // Worst possible RSSI
    
    for (auto& bridge : knownBridges) {
      // Include any bridge that reported Internet at some point
      if (bridge.internetConnected) {
        if (bridge.routerRSSI > bestRSSI) {
          bestRSSI = bridge.routerRSSI;
          best = &bridge;
        }
      }
    }
    
    return best;
  }
 
  bool hasActiveMeshConnections() {
    // Check if we have any active connections in our subs list
    for (auto& conn : this->subs) {
      if (conn && conn->connected()) {
        return true;
      }
    }
    return false;
  }
 
  bool isBridge() {
    return this->root;
  }
 
  // ==================== Capacity (ESP8266) ====================
 
  size_t apChildren() {
    size_t n = 0;
    for (auto&& sub : this->subs) {
      if (sub && sub->connected() && !sub->station) ++n;
    }
    return n;
  }
 
  bool overCapacity() { return _overCapacity; }
 
  bool capacityCheck(uint32_t freeHeap, uint32_t floor) {
    using namespace logger;
    size_t children = apChildren();
    bool over = freeHeap < floor && children > 1;
    uint32_t now = millis();
    if (over && (!_overCapacity ||
                 now - _capacityWarnedAt > CAPACITY_WARN_INTERVAL_MS)) {
      Log(ERROR,
          "capacity: %u B free with %u AP children on ESP8266. This part is "
          "specified as a leaf in meshes of this size; init(..., maxconn=0) "
          "or reduce the mesh.\n",
          freeHeap, children);
      _capacityWarnedAt = now;
    }
    _overCapacity = over;
    return over;
  }
 
  static constexpr uint32_t ESP8266_CAPACITY_FLOOR = 12 * 1024;
  static constexpr uint32_t CAPACITY_WARN_INTERVAL_MS = 5 * 60 * 1000;
 
  // ==================== Gateway Status API ====================
 
  bool isPrimaryGateway() {
    BridgeInfo* primary = getPrimaryBridge();
    if (primary == nullptr) {
      return false;
    }
    return primary->nodeId == this->nodeId;
  }
 
  uint32_t getPrimaryGateway() {
    BridgeInfo* primary = getPrimaryBridge();
    if (primary == nullptr) {
      return 0;
    }
    return primary->nodeId;
  }
 
  std::list<uint32_t> getGateways() {
    // Convert vector to list for API consistency with getNodeList()
    // The copy overhead is acceptable since gateway lists are typically small
    auto nodes = getNodesWithInternet();
    return std::list<uint32_t>(nodes.begin(), nodes.end());
  }
 
  size_t getGatewayCount() {
    size_t count = 0;
    for (const auto& bridge : knownBridges) {
      if (bridge.isHealthy(bridgeTimeoutMs) && bridge.internetConnected) count++;
    }
    return count;
  }
 
  void onGatewayChanged(gatewayChangedCallback_t callback) {
    Log(logger::GENERAL, "onGatewayChanged():\n");
    gatewayChangedCallback = callback;
  }
 
  void setBridgeStatusInterval(uint32_t intervalMs) {
    bridgeStatusIntervalMs = intervalMs;
  }
 
  void setBridgeTimeout(uint32_t timeoutMs) {
    bridgeTimeoutMs = timeoutMs;
  }
 
  void enableBridgeStatusBroadcast(bool enabled) {
    bridgeStatusBroadcastEnabled = enabled;
  }
 
  bool forgetBridge(uint32_t bridgeNodeId) {
    size_t sizeBefore = knownBridges.size();
    knownBridges.erase(
        std::remove_if(knownBridges.begin(), knownBridges.end(),
                       [bridgeNodeId](const BridgeInfo& bridge) {
                         return bridge.nodeId == bridgeNodeId;
                       }),
        knownBridges.end());
    return knownBridges.size() != sizeBefore;
  }
 
  void cleanupExpiredBridges() {
    using namespace logger;
    size_t sizeBefore = knownBridges.size();
    
    // Remove bridges that haven't reported within the timeout period
    knownBridges.erase(
        std::remove_if(knownBridges.begin(), knownBridges.end(),
                       [this](const BridgeInfo& bridge) {
                         return !bridge.isHealthy(bridgeTimeoutMs);
                       }),
        knownBridges.end());
    
    size_t removed = sizeBefore - knownBridges.size();
    if (removed > 0) {
      Log(GENERAL, "cleanupExpiredBridges(): Removed %lu expired bridges\n",
          static_cast<unsigned long>(removed));
    }
  }
 
  void enableBridgeCleanup() {
    using namespace logger;
    if (bridgeCleanupTask != nullptr) {
      Log(GENERAL, "enableBridgeCleanup(): Already enabled\n");
      return;
    }
    
    Log(GENERAL, "enableBridgeCleanup(): Starting cleanup task (interval: %u ms)\n",
        bridgeTimeoutMs);
    
    bridgeCleanupTask = this->addTask(
        bridgeTimeoutMs,
        TASK_FOREVER,
        [this]() {
          cleanupExpiredBridges();
        });
  }
 
  void disableBridgeCleanup() {
    using namespace logger;
    if (bridgeCleanupTask != nullptr) {
      bridgeCleanupTask->disable();
      // Task remains in scheduler (disabled) — negligible overhead.
      // The scheduler will skip it on each cycle.
      bridgeCleanupTask = nullptr;
      Log(GENERAL, "disableBridgeCleanup(): Cleanup task disabled\n");
    }
  }
 
  bool isBridgeCleanupEnabled() const {
    return bridgeCleanupTask != nullptr && bridgeCleanupTask->isEnabled();
  }
 
  // ==================== Local Internet Health Check API ====================
 
  bool hasLocalInternet() {
    return internetHealthChecker.hasLocalInternet();
  }
 
  bool uplinkServesLocally() {
    if (hasLocalInternet()) return true;
    if (!isInternetHealthCheckEnabled()) return false;
    if (internetHealthChecker.checkOnDemand()) {
      Log(logger::COMMUNICATION,
          "uplinkServesLocally(): on-demand probe found the uplink up\n");
      return true;
    }
    return false;
  }
 
  void onLocalInternetChanged(localInternetChangedCallback_t callback) {
    Log(logger::GENERAL, "onLocalInternetChanged():\n");
    localInternetChangedCallback = callback;
    internetHealthChecker.onConnectivityChanged(callback);
  }
 
  gateway::InternetStatus getInternetStatus() {
    return internetHealthChecker.getStatus();
  }
 
  void configureInternetHealthCheck(const gateway::SharedGatewayConfig& config) {
    internetHealthChecker.setConfig(config);
  }
 
  void setInternetCheckTarget(const TSTRING& host, uint16_t port = 53) {
    internetHealthChecker.setCheckTarget(host, port);
  }
 
  void setInternetCheckInterval(uint32_t intervalMs) {
    internetHealthChecker.setCheckInterval(intervalMs);
  }
 
  void setInternetCheckTimeout(uint32_t timeoutMs) {
    internetHealthChecker.setCheckTimeout(timeoutMs);
  }
 
  bool checkInternetNow() {
    return internetHealthChecker.checkNow();
  }
 
  void enableInternetHealthCheck() {
    using namespace logger;
    if (internetHealthCheckTask != nullptr) {
      if (internetHealthCheckTask->isEnabled()) {
        Log(GENERAL, "enableInternetHealthCheck(): Already enabled\n");
        return;
      }
      // A stop()/re-init cycle -- what bridge promotion does before calling
      // initAsBridge() again -- disables every task and drops it from the
      // reusable pool, but this member still points at it. Refusing here
      // would leave a re-initialised node with no health check at all, and
      // so with hasLocalInternet() stuck false (issue #445). Re-arm instead,
      // releasing the spent task so the pool can reclaim it.
      internetHealthCheckTask = nullptr;
    }
    
    Log(GENERAL, "enableInternetHealthCheck(): Starting health check task (interval: %u ms)\n",
        internetHealthChecker.getCheckInterval());
    
    internetHealthCheckTask = this->addTask(
        internetHealthChecker.getCheckInterval(),
        TASK_FOREVER,
        [this]() {
          this->internetHealthChecker.checkNow();
        });
  }
 
  void disableInternetHealthCheck() {
    using namespace logger;
    if (internetHealthCheckTask != nullptr) {
      internetHealthCheckTask->disable();
      internetHealthCheckTask = nullptr;
      Log(GENERAL, "disableInternetHealthCheck(): Health check task disabled\n");
    }
  }
 
  bool isInternetHealthCheckEnabled() const {
    return internetHealthCheckTask != nullptr && internetHealthCheckTask->isEnabled();
  }
 
  void resetInternetHealthStats() {
    internetHealthChecker.resetStats();
  }
 
#ifdef PAINLESSMESH_BOOST
  void setMockInternetConnected(bool connected) {
    internetHealthChecker.setMockConnected(connected);
  }
#endif
 
  // ==================== Send to Internet API ====================
 
  uint32_t sendToInternet(
      TSTRING destination,
      TSTRING payload,
      internetResultCallback_t callback,
      uint8_t priority = static_cast<uint8_t>(gateway::GatewayPriority::PRIORITY_NORMAL)) {
    return startInternetRequest(destination, payload, callback, nullptr, priority);
  }
 
  uint32_t sendToInternet(
      TSTRING destination,
      TSTRING payload,
      internetResponseCallback_t onResult,
      uint8_t priority = static_cast<uint8_t>(gateway::GatewayPriority::PRIORITY_NORMAL)) {
    return startInternetRequest(destination, payload, nullptr, onResult, priority);
  }
 
 private:
  static void deliverInternetResult(const PendingInternetRequest& request,
                                    InternetResult result) {
    result.messageId = request.messageId;
    result.attempts = request.attempts;
    // A request that never left this node -- refused for want of a mesh or a
    // gateway, or whose every routing attempt failed to send -- cannot have
    // reached a server, whatever ended it: resending it cannot duplicate.
    if (!result.success && request.attempts == 0) {
      result.retryable = true;
    }
    if (request.onResult) {
      request.onResult(result);
    } else if (request.callback) {
      request.callback(result.success, result.httpStatus, result.error);
    }
  }
 
  static InternetResult failedInternetResult(const TSTRING& error,
                                             uint16_t httpStatus = 0) {
    InternetResult result;
    result.httpStatus = httpStatus;
    result.error = error;
    return result;
  }
 
  uint32_t startInternetRequest(
      TSTRING destination,
      TSTRING payload,
      internetResultCallback_t callback,
      internetResponseCallback_t onResult,
      uint8_t priority) {
    using namespace logger;
 
    // Generate unique message ID
    uint32_t messageId = gateway::GatewayDataPackage::generateMessageId(this->nodeId);
 
    Log(COMMUNICATION, "sendToInternet(): msgId=%u dest=%s priority=%u\n",
        messageId, destination.c_str(), priority);
 
    // A shared gateway can execute its own gateway package locally. Keep the
    // same package handler, pending-request tracking and callback semantics,
    // but do not require a mesh peer merely to reach this node's own uplink.
    if (uplinkServesLocally()) {
      PendingInternetRequest request;
      request.messageId = messageId;
      request.timestamp = millis();
      request.retryCount = 0;
      request.maxRetries = internetRetryCount;
      request.priority = priority;
      request.timeoutMs = internetRequestTimeout;
      request.retryDelayMs = internetRetryDelay;
      request.gatewayNodeId = this->nodeId;
      request.destination = destination;
      request.payload = payload;
      request.callback = callback;
      request.onResult = onResult;
      request.requestNonce = gateway::newRequestNonce();
      // Handed to this node's own handler below, unconditionally.
      request.attempts = 1;
      const uint32_t nonce = request.requestNonce;
      pendingInternetRequests[messageId] = request;
 
      dispatchInternetRequestLocally(messageId, priority, 0, destination,
                                     payload, nonce);
      return messageId;
    }
 
    // A request refused before it is tracked still reports through the
    // callback it was given, on the scheduler rather than inside this call.
    PendingInternetRequest refused;
    refused.callback = callback;
    refused.onResult = onResult;
    refused.attempts = 0;
 
    // Validate mesh connectivity before attempting to send
    if (!hasActiveMeshConnections()) {
      Log(ERROR, "sendToInternet(): No active mesh connections\n");
      if (callback || onResult) {
        this->addTask([refused]() {
          deliverInternetResult(refused, failedInternetResult(
              "No mesh connections - cannot route to gateway"));
        });
      }
      return 0;
    }
 
    // Find the best gateway to route through
    BridgeInfo* gateway = getPrimaryBridge();
    if (gateway == nullptr) {
      Log(ERROR, "sendToInternet(): No gateway available\n");
      if (callback || onResult) {
        this->addTask([refused]() {
          deliverInternetResult(refused, failedInternetResult("No gateway available"));
        });
      }
      return 0;
    }
 
    // Create and store the pending request
    PendingInternetRequest request;
    request.messageId = messageId;
    request.timestamp = millis();
    request.retryCount = 0;
    request.maxRetries = internetRetryCount;
    request.priority = priority;
    request.timeoutMs = internetRequestTimeout;
    request.retryDelayMs = internetRetryDelay;
    request.gatewayNodeId = gateway->nodeId;
    request.destination = destination;
    request.payload = payload;
    request.callback = callback;
    request.onResult = onResult;
    request.requestNonce = gateway::newRequestNonce();
 
    // Store pending request
    pendingInternetRequests[messageId] = request;
 
    // Create gateway data package
    gateway::GatewayDataPackage pkg;
    pkg.from = this->nodeId;
    pkg.dest = gateway->nodeId;
    pkg.messageId = messageId;
    pkg.originNode = this->nodeId;
    pkg.timestamp = this->getNodeTime();
    pkg.priority = priority;
    pkg.destination = destination;
    pkg.payload = payload;
    pkg.contentType = "application/json";
    pkg.retryCount = 0;
    pkg.requiresAck = true;
    pkg.requestNonce = request.requestNonce;
 
    // Send the package with priority
    bool sent = false;
    auto conn = painlessmesh::router::findRoute<T>((*this), gateway->nodeId);
    if (conn) {
      // HTTP runs on the gateway's cooperative scheduler. Preserve this
      // requester's route until the corresponding acknowledgment can return.
      painlessmesh::gateway::reserveGatewayBlockingBudget(*conn);
      sent = painlessmesh::router::sendWithPriority(pkg, conn, priority);
      if (!sent) {
        Log(ERROR, "sendToInternet(): sendWithPriority failed to gateway %u (send buffer full?)\n", gateway->nodeId);
      }
    } else {
      Log(ERROR, "sendToInternet(): No route to gateway %u\n", gateway->nodeId);
    }
 
    if (!sent) {
      Log(ERROR, "sendToInternet(): Failed to send to gateway %u, scheduling retry\n", gateway->nodeId);
      // Schedule retry or failure callback
      scheduleInternetRetry(messageId);
    } else {
      Log(COMMUNICATION, "sendToInternet(): Sent to gateway %u\n", gateway->nodeId);
      // Only a send that left this node is an attempt.
      pendingInternetRequests[messageId].attempts = 1;
      // Schedule timeout check
      scheduleInternetTimeout(messageId);
    }
 
    return messageId;
  }
 
 public:
  void setInternetRequestTimeout(uint32_t timeoutMs) {
    internetRequestTimeout = timeoutMs;
  }
 
  void setInternetRetryCount(uint8_t retryCount) {
    internetRetryCount = retryCount;
  }
 
  void setInternetRetryDelay(uint32_t delayMs) {
    internetRetryDelay = delayMs;
  }
 
  void setTcpRetryConfig(const painlessmesh::tcp::TcpRetryConfig &config) {
    tcpRetryConfig = painlessmesh::tcp::clampTcpRetryConfig(config);
  }
 
  painlessmesh::tcp::TcpRetryConfig getTcpRetryConfig() const {
    return tcpRetryConfig;
  }
 
  size_t getPendingInternetRequestCount() const {
    return pendingInternetRequests.size();
  }
 
  bool cancelInternetRequest(uint32_t messageId) {
    auto it = pendingInternetRequests.find(messageId);
    if (it != pendingInternetRequests.end()) {
      const PendingInternetRequest cancelled = it->second;
      pendingInternetRequests.erase(it);
      if (cancelled.callback || cancelled.onResult) {
        this->addTask([cancelled]() {
          deliverInternetResult(cancelled, failedInternetResult("Request cancelled"));
        });
      }
      Log(logger::GENERAL, "cancelInternetRequest(): Cancelled msgId=%u\n", messageId);
      return true;
    }
    return false;
  }
 
  void enableSendToInternet() {
    using namespace logger;
    Log(GENERAL, "enableSendToInternet(): Enabling Internet routing API\n");
 
    // Register handler for gateway acknowledgment packages
    this->callbackList.onPackage(
        protocol::GATEWAY_ACK,
        [this](protocol::Variant& variant, std::shared_ptr<T>, uint32_t) {
          auto ack = variant.to<gateway::GatewayAckPackage>();
          this->handleGatewayAck(ack);
          return false;  // Don't consume, allow other handlers
        });
 
    // Start cleanup task for timed-out requests
    if (internetCleanupTask == nullptr) {
      internetCleanupTask = this->addTask(
          5000,  // Check every 5 seconds
          TASK_FOREVER,
          [this]() {
            this->cleanupTimedOutRequests();
          });
    }
 
    sendToInternetEnabled = true;
  }
 
  void disableSendToInternet() {
    using namespace logger;
    Log(GENERAL, "disableSendToInternet(): Disabling Internet routing API\n");
 
    if (internetCleanupTask != nullptr) {
      internetCleanupTask->disable();
      internetCleanupTask = nullptr;
    }
 
    // Cancel all pending requests
    for (auto& pair : pendingInternetRequests) {
      deliverInternetResult(pair.second, failedInternetResult("API disabled"));
    }
    pendingInternetRequests.clear();
 
    sendToInternetEnabled = false;
  }
 
  bool isSendToInternetEnabled() const {
    return sendToInternetEnabled;
  }
 
 private:
 public:
  // Public: the gateway handler in wifi.hpp, a derived class, answers with it.
  void sendGatewayAck(const gateway::GatewayDataPackage& request, bool success,
                      uint16_t httpStatus, const TSTRING& error,
                      std::shared_ptr<T> ingressConnection = nullptr,
                      const TSTRING& response = TSTRING(), int8_t retryable = -1,
                      uint32_t retryAfterMs = 0) {
    using namespace logger;
 
    gateway::GatewayAckPackage ack;
    ack.from = this->nodeId;
    ack.dest = request.originNode;
    ack.messageId = request.messageId;
    ack.originNode = request.originNode;
    ack.success = success;
    ack.httpStatus = httpStatus;
    ack.error = error;
    ack.timestamp = this->getNodeTime();
    ack.response = response;
    ack.retryable = retryable;
    ack.retryAfterMs = retryAfterMs;
 
    if (request.originNode == this->nodeId) {
      // Completed from the scheduler, not from inside the gateway handler
      // that called this. There the handler's HTTPClient, its WiFiClient
      // and the response buffers are all still allocated, and the
      // application's callback runs on what is left: on the HIL rig an
      // ESP8266 shared gateway with ~11 KB free delivered its own result
      // into a callback that could no longer allocate the event it built.
      //
      // The ack is moved into one heap block and handed to the handler as
      // it is. It used to be copied into the task's closure, copied again
      // inside it, serialised into a JSON document by the Variant and
      // parsed back into a third package by the handler -- the reply
      // excerpt and the error text five times over on the heap the
      // handler had already spent -- and on the rig the closure's own
      // 76-byte allocation was the one that failed (#469). Nothing but
      // handleGatewayAck() listens for GATEWAY_ACK, and a request that
      // originated here exists only if enableSendToInternet() registered
      // it, so the round trip through the callback list bought nothing.
      // A shared_ptr rather than a C++14 init-capture: the ESP32 Arduino
      // 2.x core still builds as gnu++11.
      auto held = std::make_shared<gateway::GatewayAckPackage>(std::move(ack));
      this->addTask([this, held]() { this->handleGatewayAck(*held); });
      Log(COMMUNICATION,
          "Completing local GATEWAY_ACK from the scheduler (success=%d, http=%d)\n",
          success, httpStatus);
      return;
    }
 
    auto conn = router::findRoute<T>((*this), request.originNode);
    if (!conn && ingressConnection) {
      conn = ingressConnection;
      Log(COMMUNICATION,
          "Routing GATEWAY_ACK to node %u through request ingress while "
          "topology converges\n",
          request.originNode);
    }
    if (conn) {
      protocol::Variant variant(&ack);
      router::send(std::move(variant), conn);
      Log(COMMUNICATION, "Sent GATEWAY_ACK to node %u (success=%d, http=%d)\n",
          request.originNode, success, httpStatus);
    } else {
      Log(ERROR, "Failed to send GATEWAY_ACK: no route to node %u\n",
          request.originNode);
    }
  }
 
 private:
  void handleGatewayAck(const gateway::GatewayAckPackage& ack) {
    using namespace logger;
    Log(COMMUNICATION, "handleGatewayAck(): msgId=%u success=%d http=%u\n",
        ack.messageId, ack.success, ack.httpStatus);
 
    auto it = pendingInternetRequests.find(ack.messageId);
    if (it == pendingInternetRequests.end()) {
      Log(GENERAL, "handleGatewayAck(): Unknown message ID %u\n", ack.messageId);
      return;
    }
 
    PendingInternetRequest& request = it->second;
 
    InternetResult result;
    result.success = ack.success;
    result.httpStatus = ack.httpStatus;
    result.error = ack.error;
    result.response = ack.response;
 
    // Check if this is a success response
    if (ack.success) {
      // Taken out of the map, not copied out of it, and the result handed
      // on rather than copied: on an ESP8266 shared gateway every extra
      // copy of the destination, the payload and the reply excerpt lands
      // on a heap the request itself has nearly spent (#469).
      PendingInternetRequest done = std::move(request);
      pendingInternetRequests.erase(it);
      deliverInternetResult(done, std::move(result));
      return;
    }
 
    // The node this request went to is not a gateway (issue found on the
    // HIL rig while validating 2.0.3): it rebooted or was reflashed as a
    // regular node without announcing it, and its last bridge status is
    // still inside bridgeTimeoutMs. Forget it and send again through the
    // next gateway at once. The retry budget is not spent: nothing was
    // attempted against the Internet, and every such answer removes one
    // entry from knownBridges, so this cannot loop.
    if (gateway::responseContains(ack.error,
                                  gateway::GATEWAY_NOT_A_GATEWAY_PHRASE)) {
      const uint32_t staleGateway = request.gatewayNodeId;
      bool forgot = this->forgetBridge(staleGateway);
      if (ack.from != staleGateway) forgot = this->forgetBridge(ack.from) || forgot;
      Log(ERROR, "handleGatewayAck(): %s; forgot it as a gateway for msgId=%u\n",
          ack.error.c_str(), ack.messageId);
 
      if (!forgot) {
        // Nothing to forget, so a redispatch would reach the same node.
        // Fall through to the ordinary failure path below.
      } else if (this->getPrimaryBridge() == nullptr &&
                 !this->uplinkServesLocally()) {
        // No gateway left right now. On the rig the live bridge's first
        // status arrived half a second after the ex-bridge answered, so an
        // accepted request rides the ordinary retry backoff rather than
        // failing on the spot; retryInternetRequest() reschedules while no
        // gateway is known, and the final callback carries this reason.
        char errorBuf[160];
        snprintf(errorBuf, sizeof(errorBuf),
                 "No Internet gateway available: %s", ack.error.c_str());
        request.lastError = TSTRING(errorBuf);
        scheduleInternetRetry(ack.messageId);
        return;
      } else {
        const uint32_t messageId = ack.messageId;
        this->addTask([this, messageId]() {
          this->retryInternetRequest(messageId);
        });
        return;
      }
    }
 
    // Failure: may the identical request be sent again?
    //
    // A gateway that says so decides (2.1.0 and later): it saw the transport
    // error or the status, and it knows whether the request can have reached
    // the server. Resending one that may have been processed delivers it
    // twice -- on the rig one timed-out send was issued four times -- so only
    // a request that cannot have arrived, or one the server told us to bring
    // back (429, 503), is retried.
    bool isRetryable = false;
    uint32_t minDelayMs = 0;
    if (ack.retryable >= 0) {
      isRetryable = ack.retryable == 1;
      if (isRetryable && ack.retryAfterMs > gateway::GATEWAY_RETRY_AFTER_MAX_MS) {
        // The server asked to wait longer than a retry will: say when, and
        // leave the resend to the application.
        isRetryable = false;
        char errorBuf[224];
        snprintf(errorBuf, sizeof(errorBuf), "%s (the server asked to retry after %lu s)",
                 ack.error.c_str(),
                 static_cast<unsigned long>(ack.retryAfterMs / 1000));
        result.error = TSTRING(errorBuf);
      } else if (isRetryable) {
        minDelayMs = ack.retryAfterMs;
      }
    } else {
      // A gateway that predates the field. Keep its behaviour -- it has
      // retried network errors and 5xx since 1.x -- except for a 2xx: the
      // server answered, so a resend is a second copy.
      isRetryable = legacyAckIsRetryable(ack);
    }
    result.retryable = isRetryable;
 
    // If retryable and have retries left, schedule retry
    if (isRetryable && request.retryCount < request.maxRetries) {
      Log(COMMUNICATION, "handleGatewayAck(): Scheduling retry for msgId=%u (attempt %u/%u)\n",
          ack.messageId, request.retryCount + 1, request.maxRetries);
      request.lastError = result.error;
      scheduleInternetRetry(ack.messageId, minDelayMs);
    } else {
      if (isRetryable) {
        Log(ERROR, "handleGatewayAck(): Max retries reached for msgId=%u\n", ack.messageId);
      } else {
        Log(COMMUNICATION, "handleGatewayAck(): Final failure for msgId=%u (HTTP %u)\n",
            ack.messageId, ack.httpStatus);
      }
      PendingInternetRequest done = std::move(request);
      pendingInternetRequests.erase(it);
      deliverInternetResult(done, std::move(result));
    }
  }
 
  static bool legacyAckIsRetryable(const gateway::GatewayAckPackage& ack) {
    if (ack.httpStatus >= 500 && ack.httpStatus < 600) return true;
    if (ack.httpStatus == 429) return true;
    if (ack.httpStatus != 0) return false;
    // Gateway-level connectivity failures are infrastructure, not transient.
    static const char* const FINAL[] = {"Router has no internet",
                                        "Gateway WiFi not connected",
                                        "Captive portal detected"};
    for (const char* phrase : FINAL) {
      if (gateway::responseContains(ack.error, phrase)) return false;
    }
    return true;
  }
 
  void dispatchInternetRequestLocally(uint32_t messageId, uint8_t priority,
                                      uint8_t retryCount,
                                      const TSTRING& destination,
                                      const TSTRING& payload,
                                      uint32_t requestNonce) {
    gateway::GatewayDataPackage pkg;
    pkg.from = this->nodeId;
    pkg.dest = this->nodeId;
    pkg.messageId = messageId;
    pkg.originNode = this->nodeId;
    pkg.timestamp = this->getNodeTime();
    pkg.priority = priority;
    pkg.destination = destination;
    pkg.payload = payload;
    pkg.contentType = "application/json";
    pkg.retryCount = retryCount;
    pkg.requiresAck = true;
    pkg.requestNonce = requestNonce;
    protocol::Variant variant(&pkg);
    this->callbackList.execute(protocol::GATEWAY_DATA, variant, nullptr, 0);
  }
 
  void scheduleInternetRetry(uint32_t messageId, uint32_t minDelayMs = 0) {
    auto it = pendingInternetRequests.find(messageId);
    if (it == pendingInternetRequests.end()) {
      return;
    }
 
    PendingInternetRequest& request = it->second;
 
    if (request.retryCount >= request.maxRetries) {
      // Max retries reached - fail the request, with the reason the retries
      // were spent on when there is one
      Log(logger::ERROR, "scheduleInternetRetry(): Max retries reached for msgId=%u\n",
          messageId);
      const PendingInternetRequest done = request;
      pendingInternetRequests.erase(it);
      deliverInternetResult(done, failedInternetResult(
          done.lastError.length() > 0 ? done.lastError : TSTRING("Max retries exceeded")));
      return;
    }
 
    // Exponential backoff, never shorter than what the server asked for
    uint32_t delay = request.retryDelayMs * (1 << request.retryCount);
    if (delay < minDelayMs) delay = minDelayMs;
    request.retryCount++;
 
    // A wait the server asked for moves the request's deadline with it:
    // otherwise a Retry-After longer than what is left of the request
    // timeout (30 s by default) times the request out before the retry the
    // server invited can run. The retry gets a full timeout of its own.
    if (minDelayMs > 0) {
      const uint32_t elapsed = static_cast<uint32_t>(millis()) - request.timestamp;
      const uint32_t needed = elapsed + delay + internetRequestTimeout;
      if (needed > request.timeoutMs) request.timeoutMs = needed;
    }
 
    Log(logger::COMMUNICATION, "scheduleInternetRetry(): Retry %u for msgId=%u in %u ms\n",
        request.retryCount, messageId, delay);
 
    // Schedule retry
    this->addTask(delay, TASK_ONCE, [this, messageId]() {
      this->retryInternetRequest(messageId);
    });
  }
 
  void retryInternetRequest(uint32_t messageId) {
    auto it = pendingInternetRequests.find(messageId);
    if (it == pendingInternetRequests.end()) {
      return;
    }
 
    PendingInternetRequest& request = it->second;
 
    // A request this node is serving from its own uplink never had a mesh
    // hop to repeat. The routing path below needs an active peer and a
    // discovered bridge, so a lone gateway would burn every retry without
    // issuing a single HTTP request and then report "Max retries exceeded"
    // in place of the transport error that actually occurred -- the same
    // dishonest failure issue #446 was about.
    //
    // Copy what the redispatch needs first: the local handler runs
    // synchronously and can erase this entry, leaving `request` dangling.
    if (request.gatewayNodeId == this->nodeId && uplinkServesLocally()) {
      const uint8_t priority = request.priority;
      const uint8_t retryCount = request.retryCount;
      const TSTRING destination = request.destination;
      const TSTRING payload = request.payload;
      const uint32_t nonce = request.requestNonce;
      request.attempts++;
      Log(logger::COMMUNICATION,
          "retryInternetRequest(): Retrying msgId=%u on the local uplink\n",
          messageId);
      dispatchInternetRequestLocally(messageId, priority, retryCount,
                                     destination, payload, nonce);
      return;
    }
 
    // Check mesh connectivity before attempting retry
    // During bridge failover, connection may be temporarily lost
    if (!hasActiveMeshConnections()) {
      Log(logger::ERROR, "retryInternetRequest(): No active mesh connections for retry msgId=%u, rescheduling\n",
          messageId);
      scheduleInternetRetry(messageId);
      return;
    }
 
    // Find gateway (may have changed)
    BridgeInfo* gateway = getPrimaryBridge();
    if (gateway == nullptr) {
      Log(logger::ERROR, "retryInternetRequest(): No gateway for retry msgId=%u\n",
          messageId);
      scheduleInternetRetry(messageId);
      return;
    }
 
    // Update gateway node ID (may have changed)
    request.gatewayNodeId = gateway->nodeId;
 
    // Create gateway data package
    gateway::GatewayDataPackage pkg;
    pkg.from = this->nodeId;
    pkg.dest = gateway->nodeId;
    pkg.messageId = messageId;
    pkg.originNode = this->nodeId;
    pkg.timestamp = this->getNodeTime();
    pkg.priority = request.priority;
    pkg.destination = request.destination;
    pkg.payload = request.payload;
    pkg.contentType = "application/json";
    pkg.retryCount = request.retryCount;
    pkg.requiresAck = true;
    pkg.requestNonce = request.requestNonce;
 
    // Send the package
    auto conn = painlessmesh::router::findRoute<T>((*this), gateway->nodeId);
    bool sent = false;
    if (conn) {
      painlessmesh::gateway::reserveGatewayBlockingBudget(*conn);
      sent = painlessmesh::router::sendWithPriority(pkg, conn, request.priority);
    }
 
    if (!sent) {
      Log(logger::ERROR, "retryInternetRequest(): Retry send failed msgId=%u\n", messageId);
      scheduleInternetRetry(messageId);
    } else {
      request.attempts++;
      Log(logger::COMMUNICATION, "retryInternetRequest(): Retry sent msgId=%u to gateway %u\n",
          messageId, gateway->nodeId);
    }
  }
 
  void scheduleInternetTimeout(uint32_t messageId) {
    auto it = pendingInternetRequests.find(messageId);
    if (it == pendingInternetRequests.end()) {
      return;
    }
 
    uint32_t timeout = it->second.timeoutMs;
 
    this->addTask(timeout, TASK_ONCE, [this, messageId]() {
      this->checkInternetRequestTimeout(messageId);
    });
  }
 
  void checkInternetRequestTimeout(uint32_t messageId) {
    auto it = pendingInternetRequests.find(messageId);
    if (it == pendingInternetRequests.end()) {
      return;  // Already handled
    }
 
    PendingInternetRequest& request = it->second;
 
    if (request.isTimedOut()) {
      Log(logger::ERROR, "checkInternetRequestTimeout(): Request timed out msgId=%u\n",
          messageId);
      const PendingInternetRequest done = request;
      pendingInternetRequests.erase(it);
      deliverInternetResult(done, failedInternetResult("Request timed out"));
    }
  }
 
  void cleanupTimedOutRequests() {
    auto it = pendingInternetRequests.begin();
    while (it != pendingInternetRequests.end()) {
      if (it->second.isTimedOut()) {
        Log(logger::GENERAL, "cleanupTimedOutRequests(): Cleaning up msgId=%u\n",
            it->first);
        deliverInternetResult(it->second, failedInternetResult("Request timed out"));
        it = pendingInternetRequests.erase(it);
      } else {
        ++it;
      }
    }
  }
 
 public:
  void updateBridgeStatus(uint32_t bridgeNodeId, bool internetConnected, 
                         int8_t routerRSSI, uint8_t routerChannel,
                         uint32_t uptime, TSTRING gatewayIP, uint32_t timestamp) {
    // Find existing bridge or add new one
    BridgeInfo* bridge = nullptr;
    uint32_t oldPrimaryBridgeId = 0;
    auto oldPrimary = this->getPrimaryBridge();
    if (oldPrimary != nullptr) {
      oldPrimaryBridgeId = oldPrimary->nodeId;
    }
    
    for (auto& b : knownBridges) {
      if (b.nodeId == bridgeNodeId) {
        bridge = &b;
        break;
      }
    }
    
    bool wasConnected = false;
    bool isNewBridge = false;
    if (bridge == nullptr) {
      // New bridge - enforce MAX_KNOWN_BRIDGES limit before adding
      if (knownBridges.size() >= MAX_KNOWN_BRIDGES) {
        // First try to remove expired bridges
        cleanupExpiredBridges();
        
        // If still at capacity, remove the bridge with worst RSSI (lowest signal)
        if (knownBridges.size() >= MAX_KNOWN_BRIDGES && !knownBridges.empty()) {
          auto worstBridge = knownBridges.begin();
          int8_t worstRSSI = worstBridge->routerRSSI;
          
          for (auto it = knownBridges.begin(); it != knownBridges.end(); ++it) {
            if (it->routerRSSI < worstRSSI) {
              worstRSSI = it->routerRSSI;
              worstBridge = it;
            }
          }
          
          Log(logger::GENERAL, "updateBridgeStatus(): Removing bridge %u (RSSI: %d) to make room\n",
              worstBridge->nodeId, worstBridge->routerRSSI);
          knownBridges.erase(worstBridge);
        }
      }
      
      // Add new bridge
      BridgeInfo newBridge;
      newBridge.nodeId = bridgeNodeId;
      knownBridges.push_back(newBridge);
      bridge = &knownBridges.back();
      isNewBridge = true;
    } else {
      wasConnected = bridge->internetConnected;
    }
    
    // Update bridge info
    bridge->internetConnected = internetConnected;
    bridge->routerRSSI = routerRSSI;
    bridge->routerChannel = routerChannel;
    bridge->lastSeen = millis();
    bridge->uptime = uptime;
    bridge->gatewayIP = gatewayIP;
    bridge->timestamp = timestamp;
    
    // Check if primary bridge changed
    auto newPrimary = this->getPrimaryBridge();
    uint32_t newPrimaryBridgeId = (newPrimary != nullptr) ? newPrimary->nodeId : 0;
    
    // Trigger gateway changed callback if primary gateway changed
    if (oldPrimaryBridgeId != newPrimaryBridgeId && gatewayChangedCallback) {
      gatewayChangedCallback(oldPrimaryBridgeId, newPrimaryBridgeId);
    }
    
    if (diagnosticsEnabled && oldPrimaryBridgeId != newPrimaryBridgeId) {
      // Record bridge change
      lastBridgeChange.timestamp = millis();
      lastBridgeChange.oldBridgeId = oldPrimaryBridgeId;
      lastBridgeChange.newBridgeId = newPrimaryBridgeId;
      lastBridgeChange.internetAvailable = internetConnected;
      if (isNewBridge) {
        lastBridgeChange.reason = "New bridge discovered";
      } else if (internetConnected && !wasConnected) {
        lastBridgeChange.reason = "Bridge Internet restored";
      } else if (!internetConnected && wasConnected) {
        lastBridgeChange.reason = "Bridge Internet lost";
      } else {
        lastBridgeChange.reason = "Primary bridge changed";
      }
      
      lastBridgeChangeTime = millis();
      Log(logger::GENERAL, "updateBridgeStatus(): Bridge change recorded\n");
    }
    
    // Trigger callback if status changed
    if (wasConnected != internetConnected && bridgeStatusChangedCallback) {
      bridgeStatusChangedCallback(bridgeNodeId, internetConnected);
    }
  }
 
  bool enableRTC(rtc::RTCInterface* rtcInterface) {
    using namespace logger;
    Log(GENERAL, "enableRTC(): Initializing RTC\n");
    bool success = rtcManager.enable(rtcInterface);
    if (success) {
      // RTC enabled successfully - mark node as having time authority
      setTimeAuthority(true);
    }
    return success;
  }
 
  void disableRTC() {
    using namespace logger;
    Log(GENERAL, "disableRTC(): Disabling RTC\n");
    rtcManager.disable();
    // RTC disabled - remove time authority if no other source
    // Note: Bridge nodes may still have time authority from Internet
    setTimeAuthority(false);
  }
 
  bool syncRTCFromNTP(uint32_t ntpTimestamp) {
    using namespace logger;
    Log(GENERAL, "syncRTCFromNTP(): Syncing RTC to timestamp %u\n", ntpTimestamp);
    
    if (!rtcManager.isEnabled()) {
      Log(ERROR, "syncRTCFromNTP(): RTC not enabled\n");
      return false;
    }
    
    bool success = rtcManager.syncFromNTP(ntpTimestamp);
    
    if (success && rtcSyncCompleteCallback) {
      rtcSyncCompleteCallback(ntpTimestamp);
    }
    
    return success;
  }
 
  uint32_t getAccurateTime() {
    if (rtcManager.isEnabled()) {
      uint32_t rtcTime = rtcManager.getTime();
      if (rtcTime > 0) {
        return rtcTime;
      }
    }
    // Fallback to mesh time (microseconds), converted to seconds for consistency
    return getNodeTime() / 1000000;
  }
 
  bool hasRTC() const {
    return rtcManager.isEnabled();
  }
 
  rtc::RTCType getRTCType() const {
    return rtcManager.getType();
  }
 
  uint32_t getTimeSinceRTCSync() const {
    return rtcManager.getTimeSinceLastSync();
  }
 
  void onRTCSyncComplete(rtcSyncCompleteCallback_t onRTCSyncComplete) {
    Log(logger::GENERAL, "onRTCSyncComplete():\n");
    rtcSyncCompleteCallback = onRTCSyncComplete;
  }
 
  void setTimeAuthority(bool hasAuthority) {
    using namespace logger;
    if (this->hasTimeAuthority != hasAuthority) {
      this->hasTimeAuthority = hasAuthority;
      Log(GENERAL, "setTimeAuthority(): Time authority %s\n", 
          hasAuthority ? "enabled" : "disabled");
      
      // Trigger time sync with all connections to propagate authority status
      for (auto&& connection : this->subs) {
        if (connection->nodeId != 0) {
          connection->timeSyncTask.forceNextIteration();
        }
      }
    }
  }
 
  bool getTimeAuthority() const {
    return this->hasTimeAuthority;
  }
 
  //
  // Message Queue API
  //
  
  void enableMessageQueue(bool enabled, uint32_t maxSize = 1000) {
    if (enabled && !messageQueue) {
      messageQueue = new MessageQueue(maxSize);
      Log(logger::GENERAL, "enableMessageQueue(): Queue enabled with capacity %u\n", maxSize);
    } else if (!enabled && messageQueue) {
      delete messageQueue;
      messageQueue = nullptr;
      Log(logger::GENERAL, "enableMessageQueue(): Queue disabled\n");
    }
  }
  
  uint32_t queueMessage(const TSTRING& payload, 
                       const TSTRING& destination = "",
                       MessagePriority priority = PRIORITY_NORMAL) {
    if (!messageQueue) {
      Log(logger::ERROR, "queueMessage(): Message queue not enabled\n");
      return 0;
    }
    
    return messageQueue->enqueue(priority, payload, destination);
  }
  
  std::vector<QueuedMessage> flushMessageQueue() {
    if (!messageQueue) {
      return std::vector<QueuedMessage>();
    }
    
    return messageQueue->getMessages();
  }
  
  bool removeQueuedMessage(uint32_t messageId) {
    if (!messageQueue) {
      return false;
    }
    
    return messageQueue->remove(messageId);
  }
  
  uint32_t incrementQueuedMessageAttempts(uint32_t messageId) {
    if (!messageQueue) {
      return 0;
    }
    
    return messageQueue->incrementAttempts(messageId);
  }
  
  uint32_t getQueuedMessageCount(MessagePriority priority) {
    if (!messageQueue) {
      return 0;
    }
    
    return messageQueue->size(priority);
  }
  
  uint32_t getQueuedMessageCount() {
    if (!messageQueue) {
      return 0;
    }
    
    return messageQueue->size();
  }
  
  QueueStats getQueueStats() {
    if (!messageQueue) {
      return QueueStats();
    }
    
    return messageQueue->getStats();
  }
  
  void onQueueStateChanged(queueStateChangedCallback_t callback) {
    if (messageQueue) {
      messageQueue->onStateChanged(callback);
    }
  }
  
  uint32_t pruneQueue(uint32_t maxAgeMs) {
    if (!messageQueue) {
      return 0;
    }
    
    return messageQueue->pruneOldMessages(maxAgeMs);
  }
  
  void clearQueue() {
    if (messageQueue) {
      messageQueue->clear();
    }
  }
 
  bool isConnected(uint32_t nodeId) {
    return painlessmesh::router::findRoute<T>((*this), nodeId) != NULL;
  }
 
  std::list<uint32_t> getNodeList(bool includeSelf = false) {
    return painlessmesh::layout::asList(this->asNodeTree(), includeSelf);
  }
 
  inline TSTRING subConnectionJson(bool pretty = false) {
    return this->asNodeTree().toString(pretty);
  }
 
  struct ConnectionInfo {
    uint32_t nodeId = 0;           // Connected node ID
    uint32_t lastSeen = 0;         // Timestamp of last message (ms)
    int rssi = 0;                  // Signal strength (dBm)
    int avgLatency = 0;            // Average round-trip time (ms)
    int hopCount = 0;              // Hops from current node
    int quality = 0;               // Connection quality (0-100)
    uint32_t messagesRx = 0;       // Messages received
    uint32_t messagesTx = 0;       // Messages sent
    uint32_t messagesDropped = 0;  // Failed transmissions
  };
 
  std::vector<ConnectionInfo> getConnectionDetails() {
    std::vector<ConnectionInfo> connections;
    
    for (auto conn : this->subs) {
      if (conn->connected()) {
        ConnectionInfo info;
        info.nodeId = conn->nodeId;
        info.lastSeen = conn->timeLastReceived;
        info.rssi = conn->getRSSI();
        info.avgLatency = conn->getLatency();
        info.hopCount = 1;  // Direct connection
        info.quality = conn->getQuality();
        info.messagesRx = conn->messagesRx;
        info.messagesTx = conn->messagesTx;
        info.messagesDropped = conn->messagesDropped;
        
        connections.push_back(info);
      }
    }
    
    return connections;
  }
 
  int getHopCount(uint32_t nodeId) {
    // Check if requesting hop count to self
    if (nodeId == this->getNodeId()) {
      return 0;
    }
    
    // Check if it's a direct connection (1 hop)
    for (auto conn : this->subs) {
      if (conn->nodeId == nodeId && conn->connected()) {
        return 1;
      }
    }
    
    // For indirect connections, use BFS to find shortest path
    auto tree = this->asNodeTree();
    
    // BFS to find hop count
    std::queue<std::pair<uint32_t, uint16_t>> queue;  // (nodeId, hops)
    std::set<uint32_t> visited;
    
    // Start from this node
    queue.push({this->getNodeId(), 0});
    visited.insert(this->getNodeId());
    
    while (!queue.empty()) {
      auto current = queue.front();
      queue.pop();
      
      uint32_t currentNode = current.first;
      uint16_t hops = current.second;
      
      // Found the target
      if (currentNode == nodeId) {
        return hops;
      }
      
      // Get neighbors of current node
      std::list<uint32_t> neighbors;
      if (getNodeNeighbors(tree, currentNode, neighbors)) {
        // Add unvisited neighbors to queue
        for (auto neighbor : neighbors) {
          if (visited.find(neighbor) == visited.end()) {
            visited.insert(neighbor);
            queue.push({neighbor, static_cast<uint16_t>(hops + 1)});
          }
        }
      }
    }
    
    return -1;  // Node not found or unreachable
  }
 
  std::map<uint32_t, uint32_t> getRoutingTable() {
    std::map<uint32_t, uint32_t> table;
    auto tree = this->asNodeTree();
    
    // BFS from this node to build routing table
    std::queue<uint32_t> queue;
    std::set<uint32_t> visited;
    std::map<uint32_t, uint32_t> nextHop;  // dest -> next hop from source
    
    // Start from this node
    queue.push(this->getNodeId());
    visited.insert(this->getNodeId());
    
    while (!queue.empty()) {
      uint32_t current = queue.front();
      queue.pop();
      
      // Get neighbors of current node
      std::list<uint32_t> neighbors;
      if (getNodeNeighbors(tree, current, neighbors)) {
        for (auto neighbor : neighbors) {
          if (visited.find(neighbor) == visited.end()) {
            visited.insert(neighbor);
            queue.push(neighbor);
            
            // Determine next hop for this neighbor
            if (current == this->getNodeId()) {
              // Direct connection - next hop is the neighbor itself
              nextHop[neighbor] = neighbor;
            } else {
              // Multi-hop - inherit parent's next hop
              nextHop[neighbor] = nextHop[current];
            }
            
            // Add to routing table
            table[neighbor] = nextHop[neighbor];
          }
        }
      }
    }
    
    return table;
  }
 
  std::vector<uint32_t> getPathToNode(uint32_t nodeId) {
    std::vector<uint32_t> path;
    
    // Check if requesting path to self
    if (nodeId == this->getNodeId()) {
      path.push_back(this->getNodeId());
      return path;
    }
    
    auto tree = this->asNodeTree();
    
    // BFS to find path
    std::queue<uint32_t> queue;
    std::set<uint32_t> visited;
    std::map<uint32_t, uint32_t> parent;  // child -> parent mapping
    
    // Start from this node
    queue.push(this->getNodeId());
    visited.insert(this->getNodeId());
    parent[this->getNodeId()] = 0;  // Root has no parent
    
    bool found = false;
    while (!queue.empty() && !found) {
      uint32_t current = queue.front();
      queue.pop();
      
      // Check if we reached the target
      if (current == nodeId) {
        found = true;
        break;
      }
      
      // Get neighbors of current node
      std::list<uint32_t> neighbors;
      if (getNodeNeighbors(tree, current, neighbors)) {
        for (auto neighbor : neighbors) {
          if (visited.find(neighbor) == visited.end()) {
            visited.insert(neighbor);
            parent[neighbor] = current;
            queue.push(neighbor);
          }
        }
      }
    }
    
    // Reconstruct path from parent map
    if (found) {
      uint32_t current = nodeId;
      size_t maxIter = parent.size() + 1;
      while (current != 0 && maxIter-- > 0) {
        path.insert(path.begin(), current);
        current = parent[current];
      }
    }
    
    return path;
  }
 
  BridgeHealthMetrics getBridgeHealthMetrics() {
    BridgeHealthMetrics metrics;
    
    // Connectivity metrics
    metrics.uptimeSeconds = millis() / 1000;
    metrics.currentUptime = millis();
    
    // Check for primary bridge with Internet connection
    auto primaryBridge = getPrimaryBridge();
    if (primaryBridge != nullptr) {
      metrics.internetUptimeSeconds = (millis() - primaryBridge->lastSeen) / 1000;
      metrics.currentRSSI = primaryBridge->routerRSSI;
    }
    
    // Aggregate traffic and performance from all connections
    uint64_t totalBytesRx = 0;
    uint64_t totalBytesTx = 0;
    uint32_t totalMessagesRx = 0;
    uint32_t totalMessagesTx = 0;
    uint32_t totalMessagesDropped = 0;
    uint32_t totalLatency = 0;
    uint32_t latencySampleCount = 0;
    int32_t sumRSSI = 0;
    int8_t minRSSI = 0;
    int8_t maxRSSI = -127;
    uint32_t rssiCount = 0;
    
    for (auto conn : this->subs) {
      if (conn->connected()) {
        totalBytesRx += conn->bytesRx;
        totalBytesTx += conn->bytesTx;
        totalMessagesRx += conn->messagesRx;
        totalMessagesTx += conn->messagesTx;
        totalMessagesDropped += conn->messagesDropped;
        
        // Latency
        int latency = conn->getLatency();
        if (latency >= 0) {
          totalLatency += latency;
          latencySampleCount++;
        }
        
        // RSSI
        int rssi = conn->getRSSI();
        if (rssi != 0) {
          sumRSSI += rssi;
          rssiCount++;
          if (rssi < minRSSI || minRSSI == 0) minRSSI = rssi;
          if (rssi > maxRSSI) maxRSSI = rssi;
        }
      }
    }
    
    // Set traffic metrics
    metrics.bytesRx = totalBytesRx;
    metrics.bytesTx = totalBytesTx;
    metrics.messagesRx = totalMessagesRx;
    metrics.messagesTx = totalMessagesTx;
    metrics.messagesDropped = totalMessagesDropped;
    
    // Calculate averages
    if (latencySampleCount > 0) {
      metrics.avgLatencyMs = totalLatency / latencySampleCount;
    }
    
    if (rssiCount > 0) {
      metrics.avgRSSI = sumRSSI / rssiCount;
      metrics.minRSSI = minRSSI;
      metrics.maxRSSI = maxRSSI;
    }
    
    // Calculate packet loss percentage
    uint32_t totalAttempted = totalMessagesTx + totalMessagesDropped;
    if (totalAttempted > 0) {
      metrics.packetLossPercent = (totalMessagesDropped * 100) / totalAttempted;
    }
    
    // Mesh node count
    metrics.meshNodeCount = this->getNodeList(true).size();
    
    // Track disconnects (stored in protected member)
    metrics.totalDisconnects = metricsDisconnectCount;
    
    return metrics;
  }
 
  void resetHealthMetrics() {
    using namespace logger;
    Log(GENERAL, "resetHealthMetrics(): Resetting all metric counters\n");
    
    // Reset connection metrics
    for (auto conn : this->subs) {
      if (conn->connected()) {
        conn->messagesRx = 0;
        conn->messagesTx = 0;
        conn->messagesDropped = 0;
        conn->bytesRx = 0;
        conn->bytesTx = 0;
        conn->latencySamples.clear();
      }
    }
    
    // Reset disconnect counter
    metricsDisconnectCount = 0;
  }
 
  TSTRING getHealthMetricsJSON() {
    auto metrics = getBridgeHealthMetrics();
    
    TSTRING json = "{";
    json += "\"connectivity\":{";
    json += "\"uptimeSeconds\":";
    json += std::to_string(metrics.uptimeSeconds);
    json += ",\"internetUptimeSeconds\":";
    json += std::to_string(metrics.internetUptimeSeconds);
    json += ",\"totalDisconnects\":";
    json += std::to_string(metrics.totalDisconnects);
    json += ",\"currentUptime\":";
    json += std::to_string(metrics.currentUptime);
    json += "},";
    
    json += "\"signalQuality\":{";
    json += "\"currentRSSI\":";
    json += std::to_string(metrics.currentRSSI);
    json += ",\"avgRSSI\":";
    json += std::to_string(metrics.avgRSSI);
    json += ",\"minRSSI\":";
    json += std::to_string(metrics.minRSSI);
    json += ",\"maxRSSI\":";
    json += std::to_string(metrics.maxRSSI);
    json += "},";
    
    json += "\"traffic\":{";
    json += "\"bytesRx\":";
    json += std::to_string(metrics.bytesRx);
    json += ",\"bytesTx\":";
    json += std::to_string(metrics.bytesTx);
    json += ",\"messagesRx\":";
    json += std::to_string(metrics.messagesRx);
    json += ",\"messagesTx\":";
    json += std::to_string(metrics.messagesTx);
    json += ",\"messagesQueued\":";
    json += std::to_string(metrics.messagesQueued);
    json += ",\"messagesDropped\":";
    json += std::to_string(metrics.messagesDropped);
    json += "},";
    
    json += "\"performance\":{";
    json += "\"avgLatencyMs\":";
    json += std::to_string(metrics.avgLatencyMs);
    json += ",\"packetLossPercent\":";
    json += std::to_string(metrics.packetLossPercent);
    json += ",\"meshNodeCount\":";
    json += std::to_string(metrics.meshNodeCount);
    json += "}";
    
    json += "}";
    return json;
  }
 
  void onHealthMetricsUpdate(std::function<void(BridgeHealthMetrics)> callback, 
                             uint32_t intervalMs = 60000) {
    using namespace logger;
    Log(GENERAL, "onHealthMetricsUpdate(): Setting up periodic callback every %u ms\n", intervalMs);
    
    // Create a task that periodically collects and reports metrics
    auto metricsTask = this->addTask(intervalMs, TASK_FOREVER, [this, callback]() {
      auto metrics = this->getBridgeHealthMetrics();
      callback(metrics);
    });
    
    metricsTask->enable();
  }
 
  // ==================== Enhanced Diagnostics API ====================
 
  void enableDiagnostics(bool enabled = true) {
    diagnosticsEnabled = enabled;
    if (enabled) {
      Log(logger::GENERAL, "enableDiagnostics(): Diagnostics enabled\n");
    } else {
      Log(logger::GENERAL, "enableDiagnostics(): Diagnostics disabled\n");
    }
  }
 
  BridgeStatus getBridgeStatus() {
    BridgeStatus status;
    
    status.isBridge = this->isBridge();
    status.internetConnected = this->hasInternetConnection();
    
    if (status.isBridge) {
      status.role = "bridge";
    } else if (this->root) {
      status.role = "root";
    } else {
      status.role = "regular";
    }
    
    // Get primary bridge info
    auto primaryBridge = this->getPrimaryBridge();
    if (primaryBridge != nullptr) {
      status.bridgeNodeId = primaryBridge->nodeId;
      status.bridgeRSSI = primaryBridge->routerRSSI;
    }
    
    // Calculate time since last bridge change
    if (lastBridgeChangeTime > 0) {
      status.timeSinceBridgeChange = millis() - lastBridgeChangeTime;
    }
    
    return status;
  }
 
  std::vector<ElectionRecord> getElectionHistory() {
    if (!diagnosticsEnabled) {
      Log(logger::GENERAL, "getElectionHistory(): Diagnostics not enabled\n");
      return std::vector<ElectionRecord>();
    }
    return electionHistory;
  }
 
  BridgeChangeEvent getLastBridgeChange() {
    return lastBridgeChange;
  }
 
  TSTRING exportTopologyDOT() {
    TSTRING dot = "digraph mesh {\n";
    dot += "  rankdir=TB;\n";
    dot += "  node [shape=box];\n\n";
    
    // Add this node
    dot += "  \"" + TSTRING(std::to_string(this->nodeId).c_str()) + "\" ";
    if (this->isBridge()) {
      dot += "[style=filled,fillcolor=lightblue,label=\"" + TSTRING(std::to_string(this->nodeId).c_str()) + "\\nBridge\"];\n";
    } else {
      dot += "[label=\"" + TSTRING(std::to_string(this->nodeId).c_str()) + "\"];\n";
    }
    
    // Add Internet node if bridge exists
    auto primaryBridge = this->getPrimaryBridge();
    if (primaryBridge != nullptr && primaryBridge->internetConnected) {
      dot += "  \"Internet\" [shape=cloud,style=filled,fillcolor=lightgreen];\n";
      dot += "  \"" + TSTRING(std::to_string(primaryBridge->nodeId).c_str()) + "\" -> \"Internet\" [style=dashed,color=green];\n";
    }
    
    // Add all known nodes and connections
    auto nodeList = this->getNodeList(false);
    for (auto node : nodeList) {
      dot += "  \"" + TSTRING(std::to_string(node).c_str()) + "\";\n";
    }
    
    // Add edges for direct connections
    for (auto conn : this->subs) {
      if (conn->connected()) {
        dot += "  \"" + TSTRING(std::to_string(this->nodeId).c_str()) + "\" -> \"" + TSTRING(std::to_string(conn->nodeId).c_str()) + "\"";
        
        // Add edge labels with latency
        int latency = conn->getLatency();
        if (latency >= 0) {
          dot += " [label=\"" + TSTRING(std::to_string(latency).c_str()) + "ms\"]";
        }
        dot += ";\n";
      }
    }
    
    dot += "}\n";
    return dot;
  }
 
  BridgeTestResult testBridgeConnectivity() {
    BridgeTestResult result;
    
    // Check if we have a bridge
    auto primaryBridge = this->getPrimaryBridge();
    if (primaryBridge == nullptr) {
      result.success = false;
      result.message = "No bridge available";
      return result;
    }
    
    // Check if bridge is reachable
    result.bridgeReachable = this->isConnected(primaryBridge->nodeId);
    if (!result.bridgeReachable) {
      result.success = false;
      result.message = "Bridge not reachable";
      return result;
    }
    
    // Estimate latency from connection info
    for (auto conn : this->subs) {
      if (conn->nodeId == primaryBridge->nodeId && conn->connected()) {
        int latency = conn->getLatency();
        if (latency >= 0) {
          result.latencyMs = latency;
        }
        break;
      }
    }
    
    // Check Internet connectivity through bridge
    result.internetReachable = primaryBridge->internetConnected;
    
    result.success = result.bridgeReachable;
    if (result.internetReachable) {
      result.message = "Bridge reachable with Internet";
    } else {
      result.message = "Bridge reachable, no Internet";
    }
    
    return result;
  }
 
  bool isBridgeReachable(uint32_t bridgeNodeId) {
    return this->isConnected(bridgeNodeId);
  }
 
  TSTRING getDiagnosticReport() {
    TSTRING report = "=== painlessMesh Diagnostics ===\n";
    
    // Node info
    auto status = this->getBridgeStatus();
    report += "Node ID: " + TSTRING(std::to_string(this->nodeId).c_str()) + "\n";
    report += "Mode: " + status.role + "\n";
    
    // Mesh info
    auto nodeList = this->getNodeList(true);
    report += "Mesh Nodes: " + TSTRING(std::to_string(nodeList.size()).c_str()) + "\n";
    
    // Bridge info
    if (status.isBridge) {
      report += "Bridge: " + TSTRING(std::to_string(this->nodeId).c_str()) + " (this node)\n";
    } else {
      auto primaryBridge = this->getPrimaryBridge();
      if (primaryBridge != nullptr) {
        report += "Bridge: " + TSTRING(std::to_string(primaryBridge->nodeId).c_str());
        report += " (RSSI: " + TSTRING(std::to_string(primaryBridge->routerRSSI).c_str()) + " dBm";
        report += ", Internet: " + TSTRING(primaryBridge->internetConnected ? "✓" : "✗") + ")\n";
      } else {
        report += "Bridge: None available\n";
      }
    }
    
    // Connection info
    report += "Direct Connections: " + TSTRING(std::to_string(this->subs.size()).c_str()) + "\n";
    
    // Health metrics
    auto metrics = this->getBridgeHealthMetrics();
    report += "Messages RX: " + TSTRING(std::to_string(metrics.messagesRx).c_str()) + "\n";
    report += "Messages TX: " + TSTRING(std::to_string(metrics.messagesTx).c_str()) + "\n";
    report += "Messages Dropped: " + TSTRING(std::to_string(metrics.messagesDropped).c_str()) + "\n";
    
    if (metrics.avgLatencyMs > 0) {
      report += "Avg Latency: " + TSTRING(std::to_string(metrics.avgLatencyMs).c_str()) + " ms\n";
    }
    
    // Uptime
    uint32_t uptimeSeconds = millis() / 1000;
    uint32_t hours = uptimeSeconds / 3600;
    uint32_t minutes = (uptimeSeconds % 3600) / 60;
    uint32_t seconds = uptimeSeconds % 60;
    
    char uptimeStr[32];
    snprintf(uptimeStr, sizeof(uptimeStr), "%02u:%02u:%02u", hours, minutes, seconds);
    report += "Uptime: " + TSTRING(uptimeStr) + "\n";
    
    // Election info
    if (diagnosticsEnabled && !electionHistory.empty()) {
      auto& lastElection = electionHistory.back();
      uint32_t timeSinceElection = millis() - lastElection.timestamp;
      uint32_t electionMinutes = (timeSinceElection / 1000) / 60;
      uint32_t electionSeconds = (timeSinceElection / 1000) % 60;
      
      char electionTimeStr[32];
      snprintf(electionTimeStr, sizeof(electionTimeStr), "%02u:%02u", electionMinutes, electionSeconds);
      
      report += "Last Election: " + TSTRING(electionTimeStr) + " ago";
      report += " (Winner: " + TSTRING(std::to_string(lastElection.winnerNodeId).c_str());
      report += ", " + TSTRING(std::to_string(lastElection.candidateCount).c_str()) + " candidates)\n";
    }
    
    report += "================================\n";
    return report;
  }
 
  inline std::shared_ptr<Task> addTask(unsigned long aInterval,
                                       long aIterations,
                                       std::function<void()> aCallback) {
    return plugin::PackageHandler<T>::addTask((*this->mScheduler), aInterval,
                                              aIterations, aCallback);
  }
 
  inline std::shared_ptr<Task> addTask(std::function<void()> aCallback) {
    return plugin::PackageHandler<T>::addTask((*this->mScheduler), aCallback);
  }
 
  inline std::shared_ptr<Task> addTask(std::function<void()> aCallback,
                                       unsigned long delayMs) {
    return plugin::PackageHandler<T>::addTask((*this->mScheduler), delayMs,
                                              TASK_ONCE, aCallback);
  }
 
  ~Mesh() {
    this->stop();
    plugin::PackageHandler<T>::reclaimQuarantinedTasks();
    reclaimRetiredSchedulers();
    if (!isExternalScheduler) delete mScheduler;
    if (messageQueue) delete messageQueue;
  }
 
 protected:
  void reclaimRetiredSchedulers() {
    for (auto it = retiredSchedulers.begin();
         it != retiredSchedulers.end();) {
      if ((*it)->getCurrentTask() != nullptr) {
        ++it;
        continue;
      }
      delete *it;
      it = retiredSchedulers.erase(it);
    }
  }
 
  std::list<Scheduler*> retiredSchedulers = {};
 
  void queueBroadcastAck(uint32_t origin, uint32_t msgId) {
    const auto key = std::make_pair(origin, msgId);
    if (std::find(pendingBroadcastAcks.begin(), pendingBroadcastAcks.end(),
                  key) != pendingBroadcastAcks.end())
      return;
    if (pendingBroadcastAcks.size() >= MAX_QUEUED_BROADCAST_ACKS) {
      Log(logger::ERROR, "Mesh: broadcast-ack queue limit (%u) reached\n",
          (unsigned)MAX_QUEUED_BROADCAST_ACKS);
      return;
    }
    pendingBroadcastAcks.push_back(key);
    if (!broadcastAckTask) {
      broadcastAckTask = this->addTask(
          [this]() {
            while (!this->pendingBroadcastAcks.empty()) {
              auto queued = std::move(this->pendingBroadcastAcks);
              this->pendingBroadcastAcks.clear();
              for (const auto& item : queued) {
                auto pkg = ack::MessageAckPackage(this->nodeId, item.first,
                                                  item.second);
                this->sendPackage(&pkg);
              }
            }
          },
          this->nodeId % ACK_BROADCAST_JITTER_MS);
    } else if (!broadcastAckTask->isEnabled() ||
               broadcastAckTask->isLastIteration()) {
      broadcastAckTask->restartDelayed();
    }
  }
 
  void ensureAckScheduling() {
    if (!ackCheckTask) {
      ackCheckTask = this->addTask(ACK_CHECK_INTERVAL_MS, TASK_FOREVER,
                                   [this]() {
                                     if (this->checkAcks() == 0 &&
                                         this->ackCheckTask)
                                       this->ackCheckTask->disable();
                                   });
    } else if (!ackCheckTask->isEnabled()) {
      ackCheckTask->enable();
    }
  }
 
  // Poll interval for ack timeouts while acks are pending. Note for
  // battery/light-sleep nodes: while any ack is outstanding the
  // scheduler wakes at this cadence for up to ackTimeoutMs; override at
  // build time (coarser is cheaper, timeout resolution degrades
  // accordingly).
#ifndef PAINLESSMESH_ACK_CHECK_INTERVAL_MS
#define PAINLESSMESH_ACK_CHECK_INTERVAL_MS 100
#endif
  static constexpr unsigned long ACK_CHECK_INTERVAL_MS =
      PAINLESSMESH_ACK_CHECK_INTERVAL_MS;
  // Broadcast ACK replies are staggered by nodeId within this window so
  // N nodes do not converge N simultaneous unicasts on the sender
  static constexpr uint32_t ACK_BROADCAST_JITTER_MS = 50;
  // Allow the jittered ACK to traverse a multi-hop route after its slot.
  static constexpr uint32_t ACK_BROADCAST_MIN_TIMEOUT_MS = 1000;
  static constexpr size_t MAX_QUEUED_BROADCAST_ACKS =
      PAINLESSMESH_MAX_QUEUED_BROADCAST_ACKS;
  ack::AckTracker ackTracker;
  std::shared_ptr<Task> ackCheckTask;
  std::vector<std::pair<uint32_t, uint32_t> > pendingBroadcastAcks;
  std::shared_ptr<Task> broadcastAckTask;
 
  void setScheduler(Scheduler *baseScheduler) {
    this->mScheduler = baseScheduler;
    isExternalScheduler = true;
  }
 
 public:  // Windows MSVC: router functions need access
  void startTimeSync(std::shared_ptr<T> conn) {
    using namespace logger;
    Log(S_TIME, "startTimeSync(): from %u with %u\n", this->nodeId,
        conn->nodeId);
    painlessmesh::protocol::TimeSync timeSync;
    if (ntp::adopt(this->asNodeTree(), (*conn))) {
      timeSync = painlessmesh::protocol::TimeSync(this->nodeId, conn->nodeId,
                                                  this->getNodeTime());
      Log(S_TIME, "startTimeSync(): Requesting time from %u\n", conn->nodeId);
    } else {
      timeSync = painlessmesh::protocol::TimeSync(this->nodeId, conn->nodeId);
      Log(S_TIME, "startTimeSync(): Requesting %u to adopt our time\n",
          conn->nodeId);
    }
    router::send<protocol::TimeSync, T>(timeSync, conn, true);
  }
 
  bool getNodeNeighbors(const protocol::NodeTree& tree, uint32_t nodeId,
                       std::list<uint32_t>& neighbors, uint32_t parent = 0) const {
    if (tree.nodeId == nodeId) {
      // Found the target node - collect its neighbors
      // Add parent if it exists
      if (parent != 0) {
        neighbors.push_back(parent);
      }
      // Add all direct children
      for (auto& sub : tree.subs) {
        if (sub.nodeId != 0) {
          neighbors.push_back(sub.nodeId);
        }
      }
      return true;
    }
    
    // Recursively search in children
    for (auto& sub : tree.subs) {
      if (getNodeNeighbors(sub, nodeId, neighbors, tree.nodeId)) {
        return true;
      }
    }
    
    return false;
  }
 
  bool closeConnectionSTA() {
    auto connection = this->subs.begin();
    while (connection != this->subs.end()) {
      if ((*connection)->station) {
        // We found the STA connection, close it
        (*connection)->close();
        return true;
      }
      ++connection;
    }
    return false;
  }
 
  void eraseClosedConnections() {
    using namespace logger;
    Log(CONNECTION, "eraseClosedConnections():\n");
    this->subs.remove_if(
        [](const std::shared_ptr<T> &conn) { 
          // Null check for safety - should never happen but prevents crashes
          if (!conn) return true;
          return !conn->connected(); 
        });
  }
 
 public:  // Windows MSVC: TCP lambdas need access to droppedConnectionCallbacks
  // Callback functions
  callback::List<uint32_t> newConnectionCallbacks;
  callback::List<uint32_t, bool> droppedConnectionCallbacks;
  callback::List<uint32_t> changedConnectionCallbacks;
  nodeTimeAdjustedCallback_t nodeTimeAdjustedCallback;
  nodeDelayCallback_t nodeDelayReceivedCallback;
  bridgeStatusChangedCallback_t bridgeStatusChangedCallback;
  rtcSyncCompleteCallback_t rtcSyncCompleteCallback;
  gatewayChangedCallback_t gatewayChangedCallback;
  
  // Message queue for offline mode
  MessageQueue* messageQueue = nullptr;
  
#ifdef ESP32
  SemaphoreHandle_t xSemaphore = NULL;
#endif
 
  bool isExternalScheduler = false;
 
  bool shouldContainRoot = false;
 
  // init() either creates one or takes the caller's; null until then.
  Scheduler *mScheduler = nullptr;
 
 public:  // Windows MSVC: lambdas in friend functions need public access
 
  bool semaphoreTake() {
#ifdef ESP32
    return xSemaphoreTake(xSemaphore, (TickType_t)1000) == pdTRUE;
#else
    return true;
#endif
  }
 
  void semaphoreGive() {
#ifdef ESP32
    xSemaphoreGive(xSemaphore);
#endif
  }
 
  // Bridge status tracking
  std::vector<BridgeInfo> knownBridges;
  uint32_t bridgeStatusIntervalMs = 30000;  // Default 30 seconds
  uint32_t bridgeTimeoutMs = 60000;         // Default 60 seconds
  bool bridgeStatusBroadcastEnabled = true;
  // See capacityCheck(). Only ever set on ESP8266.
  bool _overCapacity = false;
  uint32_t _capacityWarnedAt = 0;
  
  // Bridge cleanup configuration
  static const size_t MAX_KNOWN_BRIDGES = 20;  // Memory efficient limit for ESP8266
  std::shared_ptr<Task> bridgeCleanupTask = nullptr;
 
  // Health metrics tracking
  uint32_t metricsDisconnectCount = 0;
 
  // RTC management
  rtc::RTCManager rtcManager;
 
  // Diagnostics tracking
  bool diagnosticsEnabled = false;
  std::vector<ElectionRecord> electionHistory;
  static const size_t MAX_ELECTION_HISTORY = 10;
  BridgeChangeEvent lastBridgeChange;
  uint32_t lastBridgeChangeTime = 0;
 
  // Local Internet health check
  gateway::InternetHealthChecker internetHealthChecker;
  std::shared_ptr<Task> internetHealthCheckTask = nullptr;
  localInternetChangedCallback_t localInternetChangedCallback;
 
  // sendToInternet API
  std::map<uint32_t, PendingInternetRequest> pendingInternetRequests;
  std::shared_ptr<Task> internetCleanupTask = nullptr;
  uint32_t internetRequestTimeout = 30000;   // Default 30 seconds
  uint8_t internetRetryCount = 3;            // Default 3 retries
  uint32_t internetRetryDelay = 1000;        // Default 1 second base delay
  bool sendToInternetEnabled = false;
 
  // TCP connect retry parameters, see setTcpRetryConfig().
  // Public because tcp::connect<Connection, wifi::Mesh> reads it but is NOT a
  // friend: the friend declaration below names connect<T, Mesh<T>>, whereas
  // wifi::Mesh::tcpConnect() instantiates connect<Connection, wifi::Mesh> - a
  // different specialisation. Same reason as the public: sections above.
  painlessmesh::tcp::TcpRetryConfig tcpRetryConfig;
 
  friend T;
  friend void onDataCb(void *, AsyncClient *, void *, size_t);
  friend void tcpSentCb(void *, AsyncClient *, size_t, uint32_t);
  friend void meshRecvCb(void *, AsyncClient *, void *, size_t);
  friend void painlessmesh::ntp::handleTimeSync<Mesh, T>(
      Mesh &, painlessmesh::protocol::TimeSync, std::shared_ptr<T>, uint32_t);
  friend void painlessmesh::ntp::handleTimeDelay<Mesh, T>(
      Mesh &, painlessmesh::protocol::TimeDelay, std::shared_ptr<T>, uint32_t);
  friend void painlessmesh::router::handleNodeSync<Mesh, T>(
      Mesh &, protocol::NodeTree, std::shared_ptr<T> conn);
  friend void painlessmesh::tcp::initServer<T, Mesh>(AsyncServer &, Mesh &);
  friend void painlessmesh::tcp::connect<T, Mesh>(AsyncClient &, IPAddress,
                                                  uint16_t, Mesh &, uint8_t);
};
 
class Connection : public painlessmesh::layout::Neighbour,
                   public painlessmesh::tcp::BufferedConnection {
 public:
  Mesh<Connection> *mesh;
  bool station = true;
  bool newConnection = true;
 
  Task timeSyncTask;
  Task nodeSyncTask;
  Task timeOutTask;
 
  // Connection metrics tracking
  uint32_t messagesRx = 0;
  uint32_t messagesTx = 0;
  uint32_t messagesDropped = 0;
  uint32_t timeLastReceived = 0;
  uint64_t bytesRx = 0;
  uint64_t bytesTx = 0;
  
  // Latency tracking (rolling window)
  std::vector<uint32_t> latencySamples;
  static const size_t MAX_LATENCY_SAMPLES = 10;
 
  Connection(AsyncClient *client, Mesh<painlessmesh::Connection> *mesh,
             bool station)
      : painlessmesh::tcp::BufferedConnection(client),
        mesh(mesh),
        station(station) {}
 
  void initTasks() {
    auto self = this->shared_from_this();
    auto mesh = this->mesh;
    this->onReceive([self](const TSTRING &str) {
      auto variant = painlessmesh::protocol::Variant(str);
 
      router::routePackage<painlessmesh::Connection>(
          (*self->mesh), self->shared_from_this(), str,
          self->mesh->callbackList, self->mesh->getNodeTime());
    });
 
    this->onDisconnect([mesh, self]() {
      Log.remote("id:%u AsyncClient disconnect\n", self->nodeId);
      self->timeSyncTask.setCallback(NULL);
      self->timeSyncTask.disable();
      self->nodeSyncTask.setCallback(NULL);
      self->nodeSyncTask.disable();
      self->timeOutTask.setCallback(NULL);
      self->timeOutTask.disable();
      auto nodeId = self->nodeId;
      auto station = self->station;
      mesh->addTask([mesh, nodeId, station]() {
        mesh->changedConnectionCallbacks.execute(nodeId);
        mesh->droppedConnectionCallbacks.execute(nodeId, station);
      });
      self->clear();
    });
 
    using namespace logger;
 
    timeOutTask.set(NODE_TIMEOUT, TASK_ONCE, [self]() {
      Log(CONNECTION, "Time out reached\n");
      Log.remote("id:%u TimeOut\n", self->nodeId);
      self->close();
    });
    mesh->mScheduler->addTask(timeOutTask);
 
    this->nodeSyncTask.set(TASK_MINUTE, TASK_FOREVER, [self]() {
      Log(SYNC, "nodeSyncTask(): request with %u\n", self->nodeId);
      router::send<protocol::NodeSyncRequest, Connection>(
          self->request(self->mesh->asNodeTree()), self);
      self->timeOutTask.disable();
      self->timeOutTask.restartDelayed();
    });
 
    mesh->mScheduler->addTask(this->nodeSyncTask);
    if (station)
      this->nodeSyncTask.enable();
    else
      this->nodeSyncTask.enableDelayed(10 * TASK_SECOND);
 
    Log(CONNECTION, "painlessmesh::Connection: New connection established.\n");
    this->initialize(mesh->mScheduler);
  }
 
  bool addMessage(const TSTRING &msg, bool priority = false) {
    return this->write(msg, priority);
  }
  
  bool addMessageWithPriority(const TSTRING &msg, uint8_t priorityLevel) {
    return this->writeWithPriority(msg, priorityLevel);
  }
 
  void onMessageReceived(size_t bytes = 0) {
    messagesRx++;
    bytesRx += bytes;
    timeLastReceived = millis();
  }
 
  void onMessageSent(bool success, size_t bytes = 0) {
    if (success) {
      messagesTx++;
      bytesTx += bytes;
    } else {
      messagesDropped++;
    }
  }
 
  void recordLatency(uint32_t latencyMs) {
    latencySamples.push_back(latencyMs);
    if (latencySamples.size() > MAX_LATENCY_SAMPLES) {
      latencySamples.erase(latencySamples.begin());
    }
  }
 
  int getLatency() {
    if (latencySamples.empty()) return -1;
 
    uint32_t sum = 0;
    for (auto sample : latencySamples) {
      sum += sample;
    }
    return sum / latencySamples.size();
  }
 
  int getQuality() {
    // Start with perfect quality
    int quality = 100;
 
    // Penalize high latency (>100ms)
    int latency = getLatency();
    if (latency > 100) {
      quality -= (latency - 100) / 5;
    }
 
    // Penalize packet loss
    if (messagesTx > 0) {
      int lossRate = (messagesDropped * 100) / messagesTx;
      quality -= lossRate;
    }
 
    // Penalize weak RSSI (if available)
    int rssi = getRSSI();
    if (rssi < -80 && rssi != 0) {
      quality -= (80 + rssi);  // e.g., -90 dBm = penalty of 10
    }
 
    return (std::max)(0, (std::min)(100, quality));
  }
 
  int getRSSI() {
#if defined(ESP32)
    // On ESP32, get WiFi RSSI
    if (WiFi.status() == WL_CONNECTED) {
      return WiFi.RSSI();
    }
    return 0;
#elif defined(ESP8266)
    // On ESP8266, get WiFi RSSI
    if (WiFi.status() == WL_CONNECTED) {
      return WiFi.RSSI();
    }
    return 0;
#else
    // Not available on non-WiFi platforms
    return 0;
#endif
  }
 
 protected:
  std::shared_ptr<Connection> shared_from_this() { return shared_from(this); }
};
};  // namespace painlessmesh
#endif