From 7eff0fd4ab992bf79f2f31f8e6559e9870489c26 Mon Sep 17 00:00:00 2001 From: Henry Schokkenbroek Date: Sat, 10 Oct 2026 13:34:55 +0200 Subject: [PATCH] Add Ethernet MQTT observer for RAK4631 + RAK13800 (W5100S) New build target RAK_4631_repeater_observer_mqtt_eth: MQTT bridge over wired Ethernet for the nRF52-based RAK4631 with a RAK13800 in the IO slot (same pin mapping as Meshtastic's rak4631_eth_gw). - MQTTBridgeEthernet: same MQTTBridge class/API as the ESP32 WiFi bridge, so MyMesh, CommonCLI and AlertReporter work unchanged. Network I/O runs in its own FreeRTOS task; the W5100S is never touched from the mesh loop. - BearTLSClient: BearSSL TLS 1.2 client on any Arduino Client, full 16 KB input record buffer, trust anchors decoded at runtime from the preset CA PEMs. - WSClient: RFC 6455 client (subprotocol "mqtt") exposed as an Arduino Client. - Stack per slot: PubSubClient -> WSClient -> BearTLSClient -> EthernetClient, supporting mqtt://, mqtts://, ws:// and wss:// incl. Ed25519 JWT presets (dutchmeshcore-1/2 by default). Max 2 active slots. - CLI: 'get eth.status' (wifi.status maps to it); alerts say "Ethernet". - Uses upstream's ETHERNET_ENABLED flag (not a fork-specific WITH_ETHERNET), so RAK4631Board.cpp gets upstream's early WB_IO2 power-on constructor (PoE boot). The constructor is byte-identical to upstream so a future merge stays clean. Also fixes the plain nRF52 builds, which no longer compiled on this branch: - JWTHelper / MQTTMessageBuilder only compile with WITH_MQTT_BRIDGE - JWTHelper base64url no longer depends on mbedTLS - 'memory' CLI command is platform specific - nrf52_base ignores the ESP-IDF-only PsychicMqttClient library Co-Authored-By: Claude Opus 5.5 (1M context) --- MQTT_IMPLEMENTATION.md | 48 + examples/simple_repeater/MyMesh.cpp | 4 + examples/simple_repeater/UITask.cpp | 16 +- platformio.ini | 2 + src/helpers/AlertReporter.cpp | 13 +- src/helpers/CommonCLI.cpp | 12 + src/helpers/JWTHelper.cpp | 53 +- src/helpers/MQTTMessageBuilder.cpp | 6 +- src/helpers/bridges/MQTTBridge.h | 7 + src/helpers/bridges/MQTTBridgeEthernet.cpp | 1488 ++++++++++++++++++++ src/helpers/bridges/MQTTBridgeEthernet.h | 256 ++++ src/helpers/net/BearTLSClient.cpp | 439 ++++++ src/helpers/net/BearTLSClient.h | 91 ++ src/helpers/net/NetRandom.cpp | 47 + src/helpers/net/NetRandom.h | 10 + src/helpers/net/WSClient.cpp | 301 ++++ src/helpers/net/WSClient.h | 67 + variants/rak4631/RAK4631Board.cpp | 13 + variants/rak4631/platformio.ini | 40 + 19 files changed, 2885 insertions(+), 28 deletions(-) create mode 100644 src/helpers/bridges/MQTTBridgeEthernet.cpp create mode 100644 src/helpers/bridges/MQTTBridgeEthernet.h create mode 100644 src/helpers/net/BearTLSClient.cpp create mode 100644 src/helpers/net/BearTLSClient.h create mode 100644 src/helpers/net/NetRandom.cpp create mode 100644 src/helpers/net/NetRandom.h create mode 100644 src/helpers/net/WSClient.cpp create mode 100644 src/helpers/net/WSClient.h diff --git a/MQTT_IMPLEMENTATION.md b/MQTT_IMPLEMENTATION.md index 21804a8e56..520e735070 100644 --- a/MQTT_IMPLEMENTATION.md +++ b/MQTT_IMPLEMENTATION.md @@ -73,6 +73,54 @@ get mqtt.status --- +## Ethernet Observer (RAK4631 + RAK13800) + +Besides the ESP32/WiFi observers there is a wired Ethernet variant for the nRF52-based +**RAK4631** with a **RAK13800** (WIZnet W5100S) module in the WisBlock IO slot. + +**Build target:** `RAK_4631_repeater_observer_mqtt_eth` + +```bash +pio run -e RAK_4631_repeater_observer_mqtt_eth -t upload +``` + +**What is the same as the WiFi observers** +- Same slot/preset system and CLI (`set mqttN.preset`, `set mqtt.iata`, `get mqtt.status`, `get mqttN.diag`, ...) +- Same presets, including `wss://` + Ed25519 JWT (default `dutchmeshcore-1` / `dutchmeshcore-2`) +- Same topics and JSON payloads (status / packets / raw) + +**What is different** +- No WiFi settings needed: the board uses DHCP on the Ethernet port. `wifi.ssid` / `wifi.pwd` are ignored. +- `get eth.status` (also `get wifi.status`) shows link, IP, MAC and uptime. +- Max **2 active MQTT slots** (W5100S has 4 sockets; each TLS slot uses ~19 KB RAM for BearSSL buffers). +- MQTT publishes are QoS 0 (status messages too); retained flag follows the preset. +- TLS: BearSSL, TLS 1.2 (ECDHE + AES-GCM / ChaCha20). Trust anchors are the CA certificates of the built-in + presets (GTS Root R4, ISRG Root X1); custom `mqtts://` / `wss://` brokers must chain to one of those. +- SNMP is not available in this build. + +**Hardware / pins** (same as Meshtastic `rak4631_eth_gw`, overridable via build flags): + +| Signal | RAK4631 pin | Build flag | +|--------|-------------|------------| +| SPI CS | 26 (WB_SPI_CS) | `ETH_PIN_CS` | +| SPI SCK / MISO / MOSI | 3 / 29 / 30 (IO slot SPI, own SPIM2 instance) | `ETH_PIN_SCK` / `ETH_PIN_MISO` / `ETH_PIN_MOSI` | +| Reset | 21 (WB_IO3) | `ETH_PIN_RESET` | +| Power (3V3_S) | 34 (WB_IO2) | `ETH_PIN_POWER` | + +**Implementation** +- `src/helpers/bridges/MQTTBridgeEthernet.{h,cpp}` — same `MQTTBridge` class/API, selected by `MQTTBridge.h` when `ETHERNET_ENABLED` is defined +- `src/helpers/net/BearTLSClient.{h,cpp}` — BearSSL TLS client on top of any Arduino `Client` (full 16 KB input record buffer) +- `src/helpers/net/WSClient.{h,cpp}` — WebSocket client (`Sec-WebSocket-Protocol: mqtt`) as an Arduino `Client` +- Stack per slot: `PubSubClient -> WSClient -> BearTLSClient -> EthernetClient` +- All network I/O runs in its own FreeRTOS task; the W5100S is never accessed from the mesh loop. NTP time is + written to the RTC from the mesh loop task (I2C safety). +- Uses upstream's `ETHERNET_ENABLED` flag, so `RAK4631Board.cpp` powers the IO slot (WB_IO2) from an early + constructor, before the framework starts (required for PoE boot). +- Upstream's Ethernet CLI (`helpers/nrf52/EthernetCLI.h`, telnet on port 23) also keys off `ETHERNET_ENABLED`. + It must not run in the MQTT build: it would initialise the W5100S a second time from another task, and the + 4 hardware sockets are already used by 2 MQTT slots, NTP and DHCP. When merging upstream, guard its blocks in + `examples/simple_repeater/main.cpp` with `#if defined(ETHERNET_ENABLED) && !defined(WITH_MQTT_BRIDGE)`. + ## Overview The MQTT bridge implementation provides: diff --git a/examples/simple_repeater/MyMesh.cpp b/examples/simple_repeater/MyMesh.cpp index af02d433a6..0edfe59511 100644 --- a/examples/simple_repeater/MyMesh.cpp +++ b/examples/simple_repeater/MyMesh.cpp @@ -1437,6 +1437,10 @@ void MyMesh::loop() { #ifdef WITH_BRIDGE // bridge.loop() is now handled by FreeRTOS task on Core 0 - no need to call it here #endif +#if defined(WITH_MQTT_BRIDGE) && defined(ETHERNET_ENABLED) + // Ethernet bridge: network I/O runs in its own task; loop() only applies NTP time to the RTC + if (bridge) bridge->loop(); +#endif if (next_flood_advert && millisHasNowPassed(next_flood_advert)) { mesh::Packet *pkt = createSelfAdvert(); diff --git a/examples/simple_repeater/UITask.cpp b/examples/simple_repeater/UITask.cpp index e648ef5fb8..0b32cf80ac 100644 --- a/examples/simple_repeater/UITask.cpp +++ b/examples/simple_repeater/UITask.cpp @@ -6,7 +6,9 @@ #define USER_BTN_PRESSED LOW #endif -#ifdef WITH_MQTT_BRIDGE +#if defined(WITH_MQTT_BRIDGE) && defined(ETHERNET_ENABLED) +#include +#elif defined(WITH_MQTT_BRIDGE) #include #endif @@ -95,7 +97,17 @@ void UITask::renderCurrScreen() { sprintf(tmp, "BW: %03.2f CR: %d", _node_prefs->bw, _node_prefs->cr); _display->print(tmp); -#ifdef WITH_MQTT_BRIDGE +#if defined(WITH_MQTT_BRIDGE) && defined(ETHERNET_ENABLED) + // Display Ethernet IP address (cached by the bridge task; never touches the W5100S here) + uint8_t eth_ip[4]; + MQTTBridge::getLocalIP(eth_ip); + if (eth_ip[0] != 0) { + _display->setCursor(0, 40); + _display->setColor(DisplayDriver::LIGHT); + snprintf(tmp, sizeof(tmp), "IP: %d.%d.%d.%d", eth_ip[0], eth_ip[1], eth_ip[2], eth_ip[3]); + _display->print(tmp); + } +#elif defined(WITH_MQTT_BRIDGE) // Display IP address for MQTT bridge devices if (WiFi.status() == WL_CONNECTED) { IPAddress ip = WiFi.localIP(); diff --git a/platformio.ini b/platformio.ini index e16f7b8304..6150161be4 100644 --- a/platformio.ini +++ b/platformio.ini @@ -93,6 +93,8 @@ build_flags = ${arduino_base.build_flags} lib_deps = ${arduino_base.lib_deps} https://github.com/oltaco/CustomLFS#0.2.2 +; PsychicMqttClient (lib/) is ESP-IDF only; the LDF would otherwise pick it up via MQTTBridge.h +lib_ignore = PsychicMqttClient ; ----------------- RP2040 --------------------- [rp2040_base] diff --git a/src/helpers/AlertReporter.cpp b/src/helpers/AlertReporter.cpp index df38d6fc5e..8d25376d5e 100644 --- a/src/helpers/AlertReporter.cpp +++ b/src/helpers/AlertReporter.cpp @@ -26,6 +26,13 @@ #define ALERT_DEBUG_PRINTLN(...) do {} while (0) #endif +// Uplink name used in the network fault alert texts +#ifdef ETHERNET_ENABLED +#define NET_ALERT_LABEL "Ethernet" +#else +#define NET_ALERT_LABEL "WiFi" +#endif + AlertReporter::AlertReporter() : _prefs(nullptr), _mesh(nullptr), _callbacks(nullptr), #ifdef WITH_MQTT_BRIDGE @@ -236,9 +243,9 @@ void AlertReporter::onLoop(unsigned long now_ms) { uint8_t reason = MQTTBridge::getLastWifiDisconnectReason(); char text[80]; if (reason != 0) { - snprintf(text, sizeof(text), "WiFi down %s (reason %u)", age, (unsigned)reason); + snprintf(text, sizeof(text), NET_ALERT_LABEL " down %s (reason %u)", age, (unsigned)reason); } else { - snprintf(text, sizeof(text), "WiFi down %s", age); + snprintf(text, sizeof(text), NET_ALERT_LABEL " down %s", age); } if (sendChannel(text)) { _wifi.state = FIRING; @@ -253,7 +260,7 @@ void AlertReporter::onLoop(unsigned long now_ms) { char age[16]; formatAge(total, age, sizeof(age)); char text[80]; - snprintf(text, sizeof(text), "WiFi recovered after %s", age); + snprintf(text, sizeof(text), NET_ALERT_LABEL " recovered after %s", age); sendChannel(text); _wifi.state = OK; } diff --git a/src/helpers/CommonCLI.cpp b/src/helpers/CommonCLI.cpp index 67f6a59e48..07709e7a44 100644 --- a/src/helpers/CommonCLI.cpp +++ b/src/helpers/CommonCLI.cpp @@ -778,6 +778,7 @@ void CommonCLI::handleCommand(uint32_t sender_timestamp, char* command, char* re strcpy(reply, "ERR: clock cannot go backwards"); } } else if (memcmp(command, "memory", 6) == 0) { +#ifdef ESP_PLATFORM sprintf(reply, "Free: %d, Min: %d, Max: %d, Queue: %d, IntFree: %d, IntMax: %d, PSRAM: %d/%d", ESP.getFreeHeap(), ESP.getMinFreeHeap(), ESP.getMaxAllocHeap(), _callbacks->getQueueSize(), @@ -785,6 +786,12 @@ void CommonCLI::handleCommand(uint32_t sender_timestamp, char* command, char* re (int)heap_caps_get_largest_free_block(MALLOC_CAP_INTERNAL), (int)heap_caps_get_free_size(MALLOC_CAP_SPIRAM), (int)heap_caps_get_total_size(MALLOC_CAP_SPIRAM)); +#elif defined(NRF52_PLATFORM) + sprintf(reply, "Heap: %d/%d used, Queue: %d", + dbgHeapUsed(), dbgHeapTotal(), _callbacks->getQueueSize()); +#else + sprintf(reply, "Queue: %d", _callbacks->getQueueSize()); +#endif } else if (memcmp(command, "tls.bundletest ", 15) == 0) { #ifdef ESP_PLATFORM if (WiFi.status() != WL_CONNECTED) { @@ -2070,6 +2077,10 @@ void CommonCLI::handleGetCmd(uint32_t sender_timestamp, char* command, char* rep sprintf(reply, "> %s", _prefs->wifi_ssid); } else if (memcmp(config, "wifi.pwd", 8) == 0) { sprintf(reply, "> %s", _prefs->wifi_password); +#ifdef ETHERNET_ENABLED + } else if (memcmp(config, "eth.status", 10) == 0 || memcmp(config, "wifi.status", 11) == 0) { + MQTTBridge::formatNetStatusReply(reply, 160); +#else } else if (memcmp(config, "wifi.status", 11) == 0) { wl_status_t status = WiFi.status(); const char* status_str; @@ -2123,6 +2134,7 @@ void CommonCLI::handleGetCmd(uint32_t sender_timestamp, char* command, char* rep sprintf(reply, "> %s (code: %d)", status_str, status); #endif } +#endif // ETHERNET_ENABLED } else if (memcmp(config, "wifi.powersave", 14) == 0) { uint8_t ps = _prefs->wifi_power_save; const char* ps_name = (ps == 1) ? "none" : (ps == 2) ? "max" : "min"; diff --git a/src/helpers/JWTHelper.cpp b/src/helpers/JWTHelper.cpp index 0b73d87ec7..d2dc29836e 100644 --- a/src/helpers/JWTHelper.cpp +++ b/src/helpers/JWTHelper.cpp @@ -1,9 +1,11 @@ +#ifdef WITH_MQTT_BRIDGE + #include "JWTHelper.h" #include #include #include +#include #include "ed_25519.h" -#include "mbedtls/base64.h" // Base64 URL encoding table (without padding) static const char base64url_chars[] = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_"; @@ -111,29 +113,35 @@ size_t JWTHelper::base64UrlEncode(const uint8_t* input, size_t inputLen, char* o if (!input || !output || outputSize == 0) { return 0; } - - size_t outlen = 0; - int ret = mbedtls_base64_encode((unsigned char*)output, outputSize - 1, &outlen, input, inputLen); - - if (ret != 0) { + + // Unpadded base64url output length (portable; no mbedTLS dependency so nRF52 builds can use it) + size_t outlen = (inputLen / 3) * 4 + ((inputLen % 3) ? (inputLen % 3) + 1 : 0); + if (outlen >= outputSize) { return 0; } - - // Convert to base64 URL format in-place (replace + with -, / with _, remove padding =) - for (size_t i = 0; i < outlen; i++) { - if (output[i] == '+') { - output[i] = '-'; - } else if (output[i] == '/') { - output[i] = '_'; - } - } - - // Remove padding '=' characters - while (outlen > 0 && output[outlen-1] == '=') { - outlen--; - } - output[outlen] = '\0'; - return outlen; + + size_t o = 0; + size_t i = 0; + while (i + 2 < inputLen) { + uint32_t v = ((uint32_t)input[i] << 16) | ((uint32_t)input[i + 1] << 8) | input[i + 2]; + output[o++] = base64url_chars[(v >> 18) & 0x3F]; + output[o++] = base64url_chars[(v >> 12) & 0x3F]; + output[o++] = base64url_chars[(v >> 6) & 0x3F]; + output[o++] = base64url_chars[v & 0x3F]; + i += 3; + } + if (inputLen - i == 1) { + uint32_t v = (uint32_t)input[i] << 16; + output[o++] = base64url_chars[(v >> 18) & 0x3F]; + output[o++] = base64url_chars[(v >> 12) & 0x3F]; + } else if (inputLen - i == 2) { + uint32_t v = ((uint32_t)input[i] << 16) | ((uint32_t)input[i + 1] << 8); + output[o++] = base64url_chars[(v >> 18) & 0x3F]; + output[o++] = base64url_chars[(v >> 12) & 0x3F]; + output[o++] = base64url_chars[(v >> 6) & 0x3F]; + } + output[o] = '\0'; + return o; } size_t JWTHelper::createHeader(char* output, size_t outputSize) { @@ -196,3 +204,4 @@ size_t JWTHelper::createPayload( return base64UrlEncode((uint8_t*)jsonBuffer, len, output, outputSize); } +#endif // WITH_MQTT_BRIDGE diff --git a/src/helpers/MQTTMessageBuilder.cpp b/src/helpers/MQTTMessageBuilder.cpp index e20ec631c9..f966640569 100644 --- a/src/helpers/MQTTMessageBuilder.cpp +++ b/src/helpers/MQTTMessageBuilder.cpp @@ -1,3 +1,5 @@ +#ifdef WITH_MQTT_BRIDGE + #include "MQTTMessageBuilder.h" #include #include @@ -450,4 +452,6 @@ void MQTTMessageBuilder::packetToHex(mesh::Packet* packet, char* hex, size_t hex // Convert serialized packet to hex bytesToHex(raw_buf, raw_len, hex, hex_size); -} \ No newline at end of file +} + +#endif // WITH_MQTT_BRIDGE diff --git a/src/helpers/bridges/MQTTBridge.h b/src/helpers/bridges/MQTTBridge.h index 049df3b35e..67993e6f71 100644 --- a/src/helpers/bridges/MQTTBridge.h +++ b/src/helpers/bridges/MQTTBridge.h @@ -1,5 +1,10 @@ #pragma once +#if defined(WITH_MQTT_BRIDGE) && defined(ETHERNET_ENABLED) +// Wired Ethernet (W5100S) variant with the same class name and API (nRF52 / RAK4631 + RAK13800) +#include "MQTTBridgeEthernet.h" +#else + #include "MeshCore.h" #include "helpers/bridges/BridgeBase.h" #include @@ -424,3 +429,5 @@ class MQTTBridge : public BridgeBase { }; #endif + +#endif // WITH_MQTT_BRIDGE && ETHERNET_ENABLED diff --git a/src/helpers/bridges/MQTTBridgeEthernet.cpp b/src/helpers/bridges/MQTTBridgeEthernet.cpp new file mode 100644 index 0000000000..1c46114b46 --- /dev/null +++ b/src/helpers/bridges/MQTTBridgeEthernet.cpp @@ -0,0 +1,1488 @@ +#if defined(WITH_MQTT_BRIDGE) && defined(ETHERNET_ENABLED) + +#include "MQTTBridge.h" +#include "../MQTTMessageBuilder.h" +#include "../TxtDataHelpers.h" +#include "../net/BearTLSClient.h" +#include "../net/WSClient.h" + +#include +#include +#include +#include +#include +#include +#include +#include + +// --------------------------------------------------------------------------- +// Hardware configuration (defaults: RAK13800 in the WisBlock IO slot of a RAK4631, +// same wiring as Meshtastic's rak4631_eth_gw). Override via build_flags. +// --------------------------------------------------------------------------- +#ifndef ETH_PIN_CS +#define ETH_PIN_CS 26 // WB_SPI_CS +#endif +#ifndef ETH_PIN_RESET +#define ETH_PIN_RESET 21 // WB_IO3 +#endif +#ifndef ETH_PIN_POWER +#define ETH_PIN_POWER 34 // WB_IO2 (3V3_S rail enable) +#endif +#ifndef ETH_PIN_SCK +#define ETH_PIN_SCK 3 // WB_SPI_CLK +#endif +#ifndef ETH_PIN_MISO +#define ETH_PIN_MISO 29 // WB_SPI_MISO +#endif +#ifndef ETH_PIN_MOSI +#define ETH_PIN_MOSI 30 // WB_SPI_MOSI +#endif +#ifndef ETH_DHCP_TIMEOUT_MS +#define ETH_DHCP_TIMEOUT_MS 10000 +#endif +#ifndef MQTT_ETH_MAX_ACTIVE_SLOTS +#define MQTT_ETH_MAX_ACTIVE_SLOTS 2 // W5100S has 4 sockets; each TLS slot needs ~19 KB RAM +#endif +#ifndef MQTT_ETH_BUFFER_SIZE +#define MQTT_ETH_BUFFER_SIZE 2304 // CONNECT with JWT, or PUBLISH with 2 KB JSON payload +#endif +#ifndef MQTT_TASK_STACK_WORDS +#define MQTT_TASK_STACK_WORDS 3072 // 12 KB: BearSSL handshake + Ed25519 JWT signing +#endif + +#ifdef NRF52_PLATFORM +// The global SPI (SPIM3) is re-pinned to the LoRa radio; Ethernet gets its own SPIM2. +static SPIClass s_eth_spi(NRF_SPIM2, ETH_PIN_MISO, ETH_PIN_SCK, ETH_PIN_MOSI); +#else +static SPIClass& s_eth_spi = SPI; +#endif + +// --------------------------------------------------------------------------- +// Shared state (written by the MQTT task, read by CLI/UI/AlertReporter) +// --------------------------------------------------------------------------- +enum NetState : uint8_t { NET_OFF, NET_STARTING, NET_UP, NET_NO_HW, NET_NO_LINK, NET_DHCP_FAILED }; +enum NetReason : uint8_t { NET_REASON_NONE = 0, NET_REASON_NO_HW = 1, NET_REASON_NO_LINK = 2, NET_REASON_DHCP = 3 }; + +static MQTTBridge* s_instance = nullptr; +static volatile uint8_t s_net_state = NET_OFF; +static volatile uint8_t s_net_reason = NET_REASON_NONE; +static volatile unsigned long s_link_up_at = 0; // millis() when IP acquired, 0 while down +static volatile unsigned long s_link_down_at = 0; // millis() of last loss/failure +static uint8_t s_ip[4] = { 0, 0, 0, 0 }; +static uint8_t s_mac[6]; + +static BearTrustAnchors s_anchors; + +static EthernetUDP s_ntp_udp; +static NTPClient s_ntp_client(s_ntp_udp, "pool.ntp.org", 0, 60000); + +// --------------------------------------------------------------------------- +// Wall clock: epoch base captured from NTP (or the RTC at boot), advanced by millis(). +// Avoids touching an I2C RTC from the MQTT task. +// --------------------------------------------------------------------------- +static volatile uint32_t s_epoch_base = 0; +static volatile unsigned long s_epoch_millis = 0; +static const uint32_t MIN_VALID_EPOCH = 1735689600UL; // 2025-01-01 + +static uint32_t netNow() { + if (s_epoch_base == 0) return 0; + return s_epoch_base + (uint32_t)((millis() - s_epoch_millis) / 1000UL); +} + +static void setEpoch(uint32_t epoch) { + s_epoch_millis = millis(); + s_epoch_base = epoch; +} + +#ifdef NRF52_PLATFORM +// newlib's time() ends up here; MQTTMessageBuilder and JWTHelper use time(nullptr). +extern "C" int _gettimeofday(struct timeval* tv, void* tz) { + (void)tz; + if (tv) { + tv->tv_sec = netNow(); + tv->tv_usec = 0; + } + return 0; +} +#endif + +// --------------------------------------------------------------------------- +// Transport stack +// --------------------------------------------------------------------------- + +// PubSubClient busy-waits on available(). The nRF52 FreeRTOS port has time slicing +// disabled, so yield whenever no data is ready to keep the mesh loop task running. +class YieldingClient : public Client { +public: + void setInner(Client* inner) { _inner = inner; } + int connect(IPAddress ip, uint16_t port) override { return _inner ? _inner->connect(ip, port) : 0; } + int connect(const char* host, uint16_t port) override { return _inner ? _inner->connect(host, port) : 0; } + size_t write(uint8_t b) override { return _inner ? _inner->write(b) : 0; } + size_t write(const uint8_t* buf, size_t size) override { return _inner ? _inner->write(buf, size) : 0; } + int available() override { + int n = _inner ? _inner->available() : 0; + if (n <= 0) yield(); + return n; + } + int read() override { return _inner ? _inner->read() : -1; } + int read(uint8_t* buf, size_t size) override { return _inner ? _inner->read(buf, size) : -1; } + int peek() override { return _inner ? _inner->peek() : -1; } + void flush() override { if (_inner) _inner->flush(); } + void stop() override { if (_inner) _inner->stop(); } + uint8_t connected() override { return _inner ? _inner->connected() : 0; } + operator bool() override { return connected(); } + +private: + Client* _inner = nullptr; +}; + +struct MQTTSlotTransport { + EthernetClient tcp; + BearTLSClient tls; + WSClient ws; + YieldingClient top; + PubSubClient mqtt; + + MQTTSlotTransport() : tls(tcp) {} +}; + +// --------------------------------------------------------------------------- +// Static helpers / CLI accessors +// --------------------------------------------------------------------------- + +static void applyEffectiveOrigin(const NodePrefs* prefs, char* dest, size_t dest_size) { + if (!prefs || !dest || dest_size == 0) return; + if (prefs->mqtt_origin[0] == '\0') { + strncpy(dest, prefs->node_name, dest_size - 1); + } else { + strncpy(dest, prefs->mqtt_origin, dest_size - 1); + } + dest[dest_size - 1] = '\0'; + StrHelper::stripSurroundingQuotes(dest, dest_size); +} + +void MQTTBridge::getEffectiveMqttOrigin(const NodePrefs* prefs, char* buf, size_t buf_size) { + if (!buf || buf_size == 0) return; + if (!prefs) { + buf[0] = '\0'; + return; + } + applyEffectiveOrigin(prefs, buf, buf_size); +} + +void MQTTBridge::refreshOriginFromPrefs() { + applyEffectiveOrigin(_prefs, _origin, sizeof(_origin)); +} + +bool MQTTBridge::isConfigValid(const NodePrefs* prefs) { + if (!prefs) return false; + for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) { + const char* preset_name = prefs->mqtt_slot_preset[i]; + if (preset_name[0] == '\0' || strcmp(preset_name, MQTT_PRESET_NONE) == 0) continue; + if (strcmp(preset_name, MQTT_PRESET_CUSTOM) == 0) { + if (prefs->mqtt_slot_host[i][0] != '\0' && prefs->mqtt_slot_port[i] != 0) return true; + } else if (findMQTTPreset(preset_name) != nullptr) { + return true; + } + } + return false; +} + +unsigned long MQTTBridge::getWifiConnectedAtMillis() { return s_link_up_at; } +uint8_t MQTTBridge::getLastWifiDisconnectReason() { return s_net_reason; } +unsigned long MQTTBridge::getLastWifiDisconnectTime() { return s_link_down_at; } + +const char* MQTTBridge::wifiReasonStr(uint8_t reason) { + switch (reason) { + case NET_REASON_NO_HW: return "no Ethernet module"; + case NET_REASON_NO_LINK: return "cable unplugged"; + case NET_REASON_DHCP: return "DHCP failed"; + default: return nullptr; + } +} + +const char* MQTTBridge::tlsErrorStr(int32_t err) { + return BearTLSClient::errorStr((int)err); +} + +void MQTTBridge::getLocalIP(uint8_t ip[4]) { + if (s_net_state == NET_UP) { + memcpy(ip, s_ip, 4); + } else { + memset(ip, 0, 4); + } +} + +static void formatUptime(char* buf, size_t size, unsigned long ms) { + unsigned long s = ms / 1000; + unsigned long d = s / 86400, h = (s % 86400) / 3600, m = (s % 3600) / 60; + if (d > 0) snprintf(buf, size, "%lud %luh %lum", d, h, m); + else if (h > 0) snprintf(buf, size, "%luh %lum", h, m); + else snprintf(buf, size, "%lum %lus", m, s % 60); +} + +void MQTTBridge::formatNetStatusReply(char* buf, size_t bufsize) { + if (!buf || bufsize == 0) return; + switch (s_net_state) { + case NET_UP: { + char up[24]; + formatUptime(up, sizeof(up), millis() - s_link_up_at); + snprintf(buf, bufsize, "> ethernet up, IP: %u.%u.%u.%u, MAC: %02X:%02X:%02X:%02X:%02X:%02X, uptime: %s", + s_ip[0], s_ip[1], s_ip[2], s_ip[3], + s_mac[0], s_mac[1], s_mac[2], s_mac[3], s_mac[4], s_mac[5], up); + break; + } + case NET_STARTING: + snprintf(buf, bufsize, "> ethernet starting (DHCP)"); + break; + case NET_OFF: + snprintf(buf, bufsize, "> ethernet off (bridge not running)"); + break; + default: { + const char* why = wifiReasonStr(s_net_reason); + snprintf(buf, bufsize, "> ethernet down: %s", why ? why : "unknown"); + break; + } + } +} + +void MQTTBridge::formatMqttStatusReply(char* buf, size_t bufsize, const NodePrefs* prefs) { + if (buf == nullptr || bufsize == 0) return; + const char* msgs = (prefs->mqtt_status_enabled) ? "on" : "off"; + if (s_instance == nullptr || !s_instance->_initialized) { + snprintf(buf, bufsize, "> msgs: %s (bridge not running)", msgs); + return; + } + MQTTBridge* b = s_instance; + + int pos = snprintf(buf, bufsize, "> msgs: %s", msgs); + if (s_net_state != NET_UP) { + pos += snprintf(buf + pos, bufsize - pos, ", eth: down"); + } + for (int i = 0; i < RUNTIME_MQTT_SLOTS && pos < (int)bufsize - 1; i++) { + const MQTTSlot& slot = b->_slots[i]; + const char* name = slot.preset ? slot.preset->name : "custom"; + const char* state; + if (!slot.enabled && slot.preset) { + state = "inactive"; + } else if (!slot.enabled) { + continue; + } else if (!b->isSlotReady(i)) { + state = "wait"; + } else if (slot.connected) { + state = "ok"; + } else if (slot.circuit_breaker_tripped) { + state = "fail"; + } else { + state = "disc"; + } + pos += snprintf(buf + pos, bufsize - pos, ", %d: %s (%s)", i + 1, name, state); + } + snprintf(buf + pos, bufsize - pos, ", q:%d", b->_queue_count); +} + +void MQTTBridge::formatSlotDiagReply(char* buf, size_t bufsize, int slot_index) { + if (!buf || bufsize == 0) return; + if (!s_instance || !s_instance->_initialized) { + snprintf(buf, bufsize, "> mqtt%d: bridge not running", slot_index + 1); + return; + } + if (slot_index < 0 || slot_index >= RUNTIME_MQTT_SLOTS) { + snprintf(buf, bufsize, "> invalid slot"); + return; + } + const MQTTSlot& slot = s_instance->_slots[slot_index]; + if (!slot.enabled && !slot.preset && slot.host[0] == '\0') { + snprintf(buf, bufsize, "> mqtt%d: not configured", slot_index + 1); + return; + } + const char* state = !slot.enabled ? "inactive" + : slot.connected ? "ok" + : slot.circuit_breaker_tripped ? "fail" : "disc"; + int pos = snprintf(buf, bufsize, "> mqtt%d: %s", slot_index + 1, state); + if (slot.disconnect_count > 0) { + pos += snprintf(buf + pos, bufsize - pos, ", dc:%lu", (unsigned long)slot.disconnect_count); + } + if (slot.last_error_time == 0) { + snprintf(buf + pos, bufsize - pos, slot.connected ? ", no errors" : ", no error info"); + return; + } + if (slot.last_tls_err != 0) { + const char* desc = BearTLSClient::errorStr(slot.last_tls_err); + if (desc) pos += snprintf(buf + pos, bufsize - pos, ", tls: %s", desc); + else pos += snprintf(buf + pos, bufsize - pos, ", tls:%d", slot.last_tls_err); + } + if (slot.last_http_status != 0 && slot.last_http_status != 101) { + pos += snprintf(buf + pos, bufsize - pos, ", ws http:%d", slot.last_http_status); + } + if (slot.last_mqtt_state != 0) { + pos += snprintf(buf + pos, bufsize - pos, ", mqtt:%d", slot.last_mqtt_state); + } + unsigned long ago = (millis() - slot.last_error_time) / 1000; + if (ago < 60) snprintf(buf + pos, bufsize - pos, ", %lus ago", ago); + else if (ago < 3600) snprintf(buf + pos, bufsize - pos, ", %lum ago", ago / 60); + else snprintf(buf + pos, bufsize - pos, ", %luh ago", ago / 3600); +} + +unsigned long MQTTBridge::getSlotCurrentOutageStartMs(int slot_index) const { + if (slot_index < 0 || slot_index >= RUNTIME_MQTT_SLOTS) return 0; + return _slots[slot_index].current_outage_started_ms; +} + +bool MQTTBridge::isSlotEnabledAndAttempted(int slot_index) const { + if (slot_index < 0 || slot_index >= RUNTIME_MQTT_SLOTS) return false; + return _slots[slot_index].enabled && _slots[slot_index].initial_connect_done; +} + +const char* MQTTBridge::getSlotPresetName(int slot_index) const { + if (slot_index < 0 || slot_index >= RUNTIME_MQTT_SLOTS) return "?"; + const MQTTSlot& s = _slots[slot_index]; + if (s.preset && s.preset->name) return s.preset->name; + if (!s.enabled) return MQTT_PRESET_NONE; + return MQTT_PRESET_CUSTOM; +} + +// --------------------------------------------------------------------------- +// Construction / lifecycle +// --------------------------------------------------------------------------- + +MQTTBridge::MQTTBridge(NodePrefs *prefs, mesh::PacketManager *mgr, mesh::RTCClock *rtc, mesh::LocalIdentity *identity) + : BridgeBase(prefs, mgr, rtc), + _status_enabled(true), _packets_enabled(true), _raw_enabled(false), + _last_status_publish(0), _last_status_retry(0), _status_interval(300000), + _packet_queue(nullptr), _task_handle(nullptr), _reload_pending(false), _paused(false), _queue_count(0), + _queue_disconnected_since(0), + _staged_raw_len(0), _staged_snr(0), _staged_rssi(0), _staged_raw_valid(false), + _ntp_synced(false), _last_ntp_sync(0), _last_ntp_attempt(0), _pending_rtc_epoch(0), + _slots_setup_done(false), _max_active_slots(MQTT_ETH_MAX_ACTIVE_SLOTS), + _timezone_utc(TimeChangeRule{"UTC", Last, Sun, Mar, 0, 0}, TimeChangeRule{"UTC", Last, Sun, Mar, 0, 0}), + _identity(identity), _last_slot_connect_ms(0), _last_no_broker_log(0), _last_config_warning(0), + _dispatcher(nullptr), _radio(nullptr), _board(nullptr), _ms(nullptr) { + strcpy(_origin, "MeshCore-Repeater"); + strcpy(_iata, "XXX"); + strcpy(_device_id, "DEVICE_ID_PLACEHOLDER"); + strcpy(_firmware_version, "unknown"); + strcpy(_board_model, "unknown"); + strcpy(_build_date, "unknown"); + _jwt_username[0] = '\0'; + for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) { + memset(&_slots[i], 0, sizeof(MQTTSlot)); + _slots[i].port = 1883; + _slot_reconfigure_pending[i] = false; + } +} + +void MQTTBridge::loadSlotsFromPrefs() { + for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) { + MQTTSlot& slot = _slots[i]; + slot.enabled = false; + slot.preset = nullptr; + const char* preset_name = _prefs->mqtt_slot_preset[i]; + if (preset_name[0] == '\0' || strcmp(preset_name, MQTT_PRESET_NONE) == 0) continue; + + if (strcmp(preset_name, MQTT_PRESET_CUSTOM) == 0) { + StrHelper::strncpy(slot.host, _prefs->mqtt_slot_host[i], sizeof(slot.host)); + if (slot.host[0] == '\0') { + MQTT_DEBUG_PRINTLN("MQTT%d: custom preset has no server configured, disabling", i + 1); + continue; + } + slot.port = _prefs->mqtt_slot_port[i]; + StrHelper::strncpy(slot.username, _prefs->mqtt_slot_username[i], sizeof(slot.username)); + StrHelper::strncpy(slot.password, _prefs->mqtt_slot_password[i], sizeof(slot.password)); + StrHelper::strncpy(slot.audience, _prefs->mqtt_slot_audience[i], sizeof(slot.audience)); + slot.enabled = true; + } else { + const MQTTPresetDef* preset = findMQTTPreset(preset_name); + if (!preset) { + MQTT_DEBUG_PRINTLN("MQTT%d: unknown preset '%s', disabling", i + 1, preset_name); + continue; + } + slot.preset = preset; + slot.enabled = true; + if (mqttPresetNeedsSlotCredentials(preset)) { + StrHelper::strncpy(slot.username, _prefs->mqtt_slot_username[i], sizeof(slot.username)); + StrHelper::strncpy(slot.password, _prefs->mqtt_slot_password[i], sizeof(slot.password)); + } + } + } +} + +void MQTTBridge::begin() { + if (_initialized) return; + MQTT_DEBUG_PRINTLN("Initializing Ethernet MQTT bridge..."); + + refreshOriginFromPrefs(); + StrHelper::strncpy(_iata, _prefs->mqtt_iata, sizeof(_iata)); + StrHelper::stripSurroundingQuotes(_iata, sizeof(_iata)); + for (int i = 0; _iata[i]; i++) _iata[i] = toupper(_iata[i]); + + _status_enabled = _prefs->mqtt_status_enabled; + _packets_enabled = _prefs->mqtt_packets_enabled; + _raw_enabled = _prefs->mqtt_raw_enabled; + if (_prefs->mqtt_status_interval >= 1000 && _prefs->mqtt_status_interval <= 3600000) { + _status_interval = _prefs->mqtt_status_interval; + } else { + _prefs->mqtt_status_interval = 300000; + _status_interval = 300000; + } + + // Seed the wall clock from the RTC (read here, on the mesh task) until NTP syncs. + if (s_epoch_base == 0 && _rtc) { + uint32_t rtc_now = _rtc->getCurrentTime(); + if (rtc_now >= MIN_VALID_EPOCH) setEpoch(rtc_now); + } + + if (_identity && _jwt_username[0] == '\0') { + char public_key_hex[65]; + mesh::Utils::toHex(public_key_hex, _identity->pub_key, PUB_KEY_SIZE); + snprintf(_jwt_username, sizeof(_jwt_username), "v1_%s", public_key_hex); + } + + if (_task_handle != nullptr) { + // Restart (e.g. after a CLI setting change): the running task owns the slots and + // the W5100S, so it reloads the configuration itself. + _reload_pending = true; + _initialized = true; + s_instance = this; + MQTT_DEBUG_PRINTLN("Ethernet MQTT bridge restarted"); + return; + } + + loadSlotsFromPrefs(); + + _packet_queue = xQueueCreate(MAX_QUEUE_SIZE, sizeof(QueuedPacket)); + if (!_packet_queue) { + MQTT_DEBUG_PRINTLN("Failed to create packet queue"); + return; + } + + for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) { + if (!_slots[i].net) { + _slots[i].net = new MQTTSlotTransport(); + _slots[i].net->mqtt.setBufferSize(MQTT_ETH_BUFFER_SIZE); + } + } + + _slots_setup_done = false; + _initialized = true; + s_instance = this; + + TaskHandle_t handle = nullptr; + if (xTaskCreate(taskEntry, "mqtt_eth", MQTT_TASK_STACK_WORDS, this, TASK_PRIO_LOW, &handle) != pdPASS) { + MQTT_DEBUG_PRINTLN("Failed to create MQTT task"); + _initialized = false; + s_instance = nullptr; + vQueueDelete((QueueHandle_t)_packet_queue); + _packet_queue = nullptr; + return; + } + _task_handle = handle; + MQTT_DEBUG_PRINTLN("Ethernet MQTT bridge started"); +} + +void MQTTBridge::end() { + if (!_initialized) return; + // Non-blocking: the task notices, closes its MQTT sessions and idles. The task is + // never deleted, since it may be in the middle of a W5100S/SPI transaction. + _initialized = false; + MQTT_DEBUG_PRINTLN("Ethernet MQTT bridge stopping"); +} + +void MQTTBridge::loop() { + // Runs on the mesh loop task: the only place where the (possibly I2C) RTC is written. + uint32_t epoch = _pending_rtc_epoch; + if (epoch != 0 && _rtc) { + _pending_rtc_epoch = 0; + _rtc->setCurrentTime(netNow()); + } +} + +// --------------------------------------------------------------------------- +// MQTT task +// --------------------------------------------------------------------------- + +void MQTTBridge::taskEntry(void* param) { + static_cast(param)->taskLoop(); +} + +void MQTTBridge::taskLoop() { + ethernetInit(); + + // Trust anchors: every distinct CA used by the built-in presets (also used for custom TLS slots). + if (s_anchors.count() == 0) { + for (int i = 0; i < MQTT_PRESET_COUNT; i++) { + if (MQTT_PRESETS[i].ca_cert && !s_anchors.addPEM(MQTT_PRESETS[i].ca_cert)) { + MQTT_DEBUG_PRINTLN("Failed to load CA for preset %s", MQTT_PRESETS[i].name); + } + } + MQTT_DEBUG_PRINTLN("Loaded %u trust anchors", (unsigned)s_anchors.count()); + } + + while (true) { + unsigned long now = millis(); + ethernetMaintain(now); + + if (_reload_pending) { + _reload_pending = false; + for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) { + teardownSlot(i); + _slot_reconfigure_pending[i] = false; + } + loadSlotsFromPrefs(); + _slots_setup_done = false; + _last_status_publish = 0; + _paused = false; + } + + if (!_initialized) { + if (!_paused) { + for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) teardownSlot(i); + xQueueReset((QueueHandle_t)_packet_queue); + _queue_count = 0; + _paused = true; + } + vTaskDelay(pdMS_TO_TICKS(100)); + continue; + } + + if (isNetworkUp()) { + bool want_ntp = !_ntp_synced ? (_last_ntp_attempt == 0 || now - _last_ntp_attempt >= 30000UL) + : (now - _last_ntp_sync >= 3600000UL); + if (want_ntp) syncTimeWithNTP(); + + for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) { + if (_slot_reconfigure_pending[i]) { + _slot_reconfigure_pending[i] = false; + applySlotPreset(i, _prefs->mqtt_slot_preset[i]); + } + } + + // Slot activation waits for a valid clock (JWT iat/exp and certificate validity). + if (!_slots_setup_done && netNow() >= MIN_VALID_EPOCH) { + _slots_setup_done = true; + int active = 0; + for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) { + if (!_slots[i].enabled) continue; + if (active >= _max_active_slots) { + MQTT_DEBUG_PRINTLN("MQTT%d skipped: max active slots (%d) reached", i + 1, _max_active_slots); + _slots[i].enabled = false; + continue; + } + active++; + } + } + + if (_slots_setup_done) maintainSlots(now); + } + + processPacketQueue(); + + if (_status_enabled && anySlotConnected()) { + bool due; + if (_last_status_retry != 0) { + due = (now - _last_status_retry) >= STATUS_RETRY_INTERVAL; + } else { + due = (_last_status_publish == 0) || (now - _last_status_publish) >= _status_interval; + } + if (due) { + _last_status_retry = now; + if (publishStatus()) { + _last_status_publish = now; + _last_status_retry = 0; + } + } + } else { + _last_status_retry = 0; + } + + checkConfigurationMismatch(); + vTaskDelay(pdMS_TO_TICKS(_queue_count > 0 ? 5 : 20)); + } +} + +// --------------------------------------------------------------------------- +// Ethernet (W5100S) +// --------------------------------------------------------------------------- + +void MQTTBridge::ethernetInit() { + s_net_state = NET_STARTING; + + // Same bring-up sequence as the RAK13800 examples / Meshtastic rak4631_eth_gw. + pinMode(ETH_PIN_POWER, OUTPUT); + digitalWrite(ETH_PIN_POWER, HIGH); + delay(100); + pinMode(ETH_PIN_RESET, OUTPUT); + digitalWrite(ETH_PIN_RESET, LOW); + delay(100); + digitalWrite(ETH_PIN_RESET, HIGH); + delay(100); + + s_eth_spi.begin(); + Ethernet.init(s_eth_spi, ETH_PIN_CS); + +#ifdef NRF52_PLATFORM + // Stable, locally administered unicast MAC derived from the chip's device address. + const uint8_t* src = (const uint8_t*)NRF_FICR->DEVICEADDR; + s_mac[0] = (uint8_t)((src[5] | 0xC2) & 0xFE); + s_mac[1] = src[4]; + s_mac[2] = src[3]; + s_mac[3] = src[2]; + s_mac[4] = src[1]; + s_mac[5] = src[0]; +#else + const uint8_t def_mac[6] = { 0xC2, 0x4D, 0x43, 0x00, 0x00, 0x01 }; + memcpy(s_mac, def_mac, 6); +#endif +} + +bool MQTTBridge::ethernetBegin() { + s_net_state = NET_STARTING; + MQTT_DEBUG_PRINTLN("Ethernet: starting DHCP..."); + if (Ethernet.begin(s_mac, ETH_DHCP_TIMEOUT_MS, 4000) == 0) { + if (Ethernet.hardwareStatus() == EthernetNoHardware) { + s_net_state = NET_NO_HW; + s_net_reason = NET_REASON_NO_HW; + MQTT_DEBUG_PRINTLN("Ethernet: W5100S not found (check RAK13800 / IO slot)"); + } else if (Ethernet.linkStatus() == LinkOFF) { + s_net_state = NET_NO_LINK; + s_net_reason = NET_REASON_NO_LINK; + MQTT_DEBUG_PRINTLN("Ethernet: cable not connected"); + } else { + s_net_state = NET_DHCP_FAILED; + s_net_reason = NET_REASON_DHCP; + MQTT_DEBUG_PRINTLN("Ethernet: DHCP failed"); + } + s_link_up_at = 0; + if (s_link_down_at == 0) s_link_down_at = millis(); + return false; + } + + IPAddress ip = Ethernet.localIP(); + for (int i = 0; i < 4; i++) s_ip[i] = ip[i]; + s_net_reason = NET_REASON_NONE; + s_link_up_at = millis(); + if (s_link_up_at == 0) s_link_up_at = 1; + s_net_state = NET_UP; + MQTT_DEBUG_PRINTLN("Ethernet up: IP %u.%u.%u.%u", s_ip[0], s_ip[1], s_ip[2], s_ip[3]); + + // Network (re)gained: let disconnected slots retry right away. + for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) { + _slots[i].last_reconnect_attempt = 0; + _slots[i].reconnect_backoff = 0; + } + return true; +} + +void MQTTBridge::ethernetMaintain(unsigned long now) { + static unsigned long next_attempt = 0; + static unsigned long last_link_check = 0; + + if (s_net_state != NET_UP) { + if ((long)(now - next_attempt) < 0) return; + if (s_net_state == NET_NO_LINK && Ethernet.linkStatus() != LinkON) { + next_attempt = now + 2000; // wait for the cable instead of running a doomed DHCP cycle + return; + } + if (!ethernetBegin()) { + next_attempt = millis() + (s_net_state == NET_NO_HW ? 60000UL : 10000UL); + } + return; + } + + if (now - last_link_check < 2000) return; + last_link_check = now; + + if (Ethernet.linkStatus() == LinkOFF) { + MQTT_DEBUG_PRINTLN("Ethernet: link lost"); + s_net_state = NET_NO_LINK; + s_net_reason = NET_REASON_NO_LINK; + s_link_up_at = 0; + s_link_down_at = now; + for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) { + if (_slots[i].connected) disconnectSlot(i, true); + } + next_attempt = now + 3000; + return; + } + + Ethernet.maintain(); // DHCP lease renewal + IPAddress ip = Ethernet.localIP(); + if (ip[0] == 0 && ip[1] == 0 && ip[2] == 0 && ip[3] == 0) { + MQTT_DEBUG_PRINTLN("Ethernet: DHCP lease lost"); + s_net_state = NET_DHCP_FAILED; + s_net_reason = NET_REASON_DHCP; + s_link_up_at = 0; + s_link_down_at = now; + for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) { + if (_slots[i].connected) disconnectSlot(i, true); + } + next_attempt = now; + return; + } + for (int i = 0; i < 4; i++) s_ip[i] = ip[i]; +} + +bool MQTTBridge::isNetworkUp() const { + return s_net_state == NET_UP; +} + +// --------------------------------------------------------------------------- +// NTP +// --------------------------------------------------------------------------- + +void MQTTBridge::syncTimeWithNTP() { + _last_ntp_attempt = millis(); + MQTT_DEBUG_PRINTLN("Syncing time with NTP..."); + + bool ok = false; + unsigned long epoch = 0; + s_ntp_client.begin(); + for (int attempt = 0; attempt < 3 && !ok; attempt++) { + if (attempt > 0) delay(1000); + if (s_ntp_client.forceUpdate()) { + epoch = s_ntp_client.getEpochTime(); + ok = epoch >= MIN_VALID_EPOCH; + } + } + s_ntp_client.end(); + + if (!ok) { + MQTT_DEBUG_PRINTLN("NTP sync failed"); + return; + } + setEpoch(epoch); + _ntp_synced = true; + _last_ntp_sync = millis(); + _pending_rtc_epoch = epoch; // RTC itself is updated from the mesh task (see loop()) + MQTT_DEBUG_PRINTLN("Time synced: %lu", epoch); +} + +// --------------------------------------------------------------------------- +// Slot management +// --------------------------------------------------------------------------- + +bool MQTTBridge::slotUsesJwt(int index) const { + const MQTTSlot& slot = _slots[index]; + return (slot.preset && slot.preset->auth_type == MQTT_AUTH_JWT) || + (!slot.preset && slot.audience[0] != '\0'); +} + +bool MQTTBridge::resolveEndpoint(int index) { + MQTTSlot& slot = _slots[index]; + char url[160]; + + if (slot.preset) { + StrHelper::strncpy(url, slot.preset->server_url, sizeof(url)); + } else if (strstr(slot.host, "://") != nullptr) { + StrHelper::strncpy(url, slot.host, sizeof(url)); + } else { + // Bare host: infer protocol from the port, like the ESP32 bridge. + const char* proto = (slot.port == 8883) ? "mqtts" : (slot.port == 443) ? "wss" : "mqtt"; + snprintf(url, sizeof(url), "%s://%s:%u", proto, slot.host, (unsigned)slot.port); + } + + const char* p = url; + if (strncmp(p, "mqtts://", 8) == 0) { slot.scheme = SCHEME_MQTTS; p += 8; } + else if (strncmp(p, "mqtt://", 7) == 0) { slot.scheme = SCHEME_MQTT; p += 7; } + else if (strncmp(p, "wss://", 6) == 0) { slot.scheme = SCHEME_WSS; p += 6; } + else if (strncmp(p, "ws://", 5) == 0) { slot.scheme = SCHEME_WS; p += 5; } + else return false; + + const char* path = strchr(p, '/'); + const char* host_end = path ? path : p + strlen(p); + const char* colon = (const char*)memchr(p, ':', host_end - p); + const char* name_end = colon ? colon : host_end; + size_t host_len = name_end - p; + if (host_len == 0 || host_len >= sizeof(slot.ep_host)) return false; + memcpy(slot.ep_host, p, host_len); + slot.ep_host[host_len] = '\0'; + + if (colon) { + slot.ep_port = (uint16_t)atoi(colon + 1); + } else if (!slot.preset && slot.port != 0) { + slot.ep_port = slot.port; + } else { + static const uint16_t DEFAULT_PORTS[] = { 1883, 8883, 80, 443 }; + slot.ep_port = DEFAULT_PORTS[slot.scheme]; + } + if (slot.ep_port == 0) return false; + + StrHelper::strncpy(slot.ep_path, path ? path : "/", sizeof(slot.ep_path)); + return true; +} + +bool MQTTBridge::createSlotAuthToken(int index) { + MQTTSlot& slot = _slots[index]; + if (!_identity) return false; + + const char* audience = nullptr; + unsigned long base_lifetime = 86400; + if (slot.preset && slot.preset->auth_type == MQTT_AUTH_JWT) { + audience = slot.preset->jwt_audience; + if (slot.preset->token_lifetime > 0) base_lifetime = slot.preset->token_lifetime; + } else if (slot.audience[0] != '\0') { + audience = slot.audience; + } + if (!audience || audience[0] == '\0') return false; + + const char* owner_key = nullptr; + char owner_key_upper[65]; + if (_prefs->mqtt_owner_public_key[0] != '\0') { + StrHelper::strncpy(owner_key_upper, _prefs->mqtt_owner_public_key, sizeof(owner_key_upper)); + for (int i = 0; owner_key_upper[i]; i++) owner_key_upper[i] = toupper(owner_key_upper[i]); + owner_key = owner_key_upper; + } + char client_version[64]; + getClientVersion(client_version, sizeof(client_version)); + const char* email = (_prefs->mqtt_email[0] != '\0') ? _prefs->mqtt_email : nullptr; + + uint32_t now = netNow(); + if (now < MIN_VALID_EPOCH) return false; + unsigned long stagger = index * min((unsigned long)300, base_lifetime / 20); + unsigned long expires_in = base_lifetime - stagger; + + if (JWTHelper::createAuthToken(*_identity, audience, now, expires_in, + slot.auth_token, AUTH_TOKEN_SIZE, owner_key, client_version, email)) { + slot.token_expires_at = now + expires_in; + return true; + } + slot.token_expires_at = 0; + return false; +} + +bool MQTTBridge::connectSlot(int index) { + MQTTSlot& slot = _slots[index]; + MQTTSlotTransport* n = slot.net; + if (!n) return false; + + slot.last_tls_err = 0; + slot.last_http_status = 0; + slot.last_mqtt_state = 0; + + if (!resolveEndpoint(index)) { + MQTT_DEBUG_PRINTLN("MQTT%d: invalid broker URL", index + 1); + slot.last_error_time = millis(); + slot.last_mqtt_state = MQTT_CONNECT_FAILED; + return false; + } + + bool use_tls = (slot.scheme == SCHEME_MQTTS || slot.scheme == SCHEME_WSS); + bool use_ws = (slot.scheme == SCHEME_WS || slot.scheme == SCHEME_WSS); + + n->tcp.setConnectionTimeout(10000); + Client* base = &n->tcp; + if (use_tls) { + n->tls.setTrustAnchors(&s_anchors); + n->tls.setUnixTime(netNow()); + base = &n->tls; + } + Client* top = base; + if (use_ws) { + n->ws.setTransport(base); + n->ws.setPath(slot.ep_path); + top = &n->ws; + } + n->top.setInner(top); + + PubSubClient& m = n->mqtt; + m.setClient(n->top); + m.setServer(slot.ep_host, slot.ep_port); + m.setKeepAlive((slot.preset && slot.preset->keepalive > 0) ? slot.preset->keepalive : 60); + m.setSocketTimeout(15); + + const char* user = nullptr; + const char* pass = nullptr; + if (slotUsesJwt(index)) { + if (!createSlotAuthToken(index)) { + MQTT_DEBUG_PRINTLN("MQTT%d: JWT creation failed", index + 1); + slot.last_error_time = millis(); + return false; + } + user = _jwt_username; + pass = slot.auth_token; + } else if (slot.preset && slot.preset->auth_type == MQTT_AUTH_USERPASS) { + if (slot.preset->userpass_username && slot.preset->userpass_password) { + user = slot.preset->userpass_username; + pass = slot.preset->userpass_password; + } else if (slot.username[0] != '\0') { + user = slot.username; + pass = slot.password; + } + } else if (!slot.preset && slot.username[0] != '\0') { + user = slot.username; + pass = slot.password; + } + + char client_id[40]; + snprintf(client_id, sizeof(client_id), "meshcore_%.16s_%d", _device_id, index + 1); + + MQTT_DEBUG_PRINTLN("MQTT%d connecting to %s://%s:%u%s", index + 1, + slot.scheme == SCHEME_MQTT ? "mqtt" : slot.scheme == SCHEME_MQTTS ? "mqtts" + : slot.scheme == SCHEME_WS ? "ws" : "wss", + slot.ep_host, (unsigned)slot.ep_port, use_ws ? slot.ep_path : ""); + + bool ok = m.connect(client_id, user, pass); + if (!ok) { + slot.last_mqtt_state = m.state(); + if (use_tls) slot.last_tls_err = n->tls.lastError(); + if (use_ws) slot.last_http_status = n->ws.lastHttpStatus(); + slot.last_error_time = millis(); + n->top.stop(); + MQTT_DEBUG_PRINTLN("MQTT%d connect failed: mqtt=%d tls=%d http=%d", index + 1, + slot.last_mqtt_state, slot.last_tls_err, slot.last_http_status); + return false; + } + MQTT_DEBUG_PRINTLN("MQTT%d connected", index + 1); + return true; +} + +void MQTTBridge::disconnectSlot(int index, bool count_as_disconnect) { + MQTTSlot& slot = _slots[index]; + if (slot.net) { + if (slot.net->mqtt.connected()) slot.net->mqtt.disconnect(); + slot.net->top.stop(); + } + if (slot.connected && count_as_disconnect) { + unsigned long now = millis(); + slot.disconnect_count++; + if (slot.first_disconnect_time == 0) slot.first_disconnect_time = now; + if (slot.current_outage_started_ms == 0) slot.current_outage_started_ms = now; + } + slot.connected = false; +} + +void MQTTBridge::teardownSlot(int index) { + disconnectSlot(index, false); + MQTTSlot& slot = _slots[index]; + slot.auth_token[0] = '\0'; + slot.token_expires_at = 0; + slot.initial_connect_done = false; + slot.reconnect_backoff = 0; + slot.max_backoff_failures = 0; + slot.circuit_breaker_tripped = false; + slot.last_reconnect_attempt = 0; + slot.current_outage_started_ms = 0; +} + +void MQTTBridge::maintainSlots(unsigned long now) { + static const unsigned long SLOT_BACKOFF_MS[] = { 10000, 30000, 60000, 120000, 300000 }; + static const uint8_t MAX_FAILURES_AT_MAX_BACKOFF = 3; + static const unsigned long CIRCUIT_BREAKER_PROBE_MS = 1800000UL; + static const unsigned long CONNECT_GUARD_MS = 5000UL; // one handshake at a time + + for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) { + MQTTSlot& slot = _slots[i]; + if (!slot.enabled || !slot.net) continue; + + if (slot.connected) { + if (!slot.net->mqtt.loop()) { + slot.last_mqtt_state = slot.net->mqtt.state(); + slot.last_error_time = now; + MQTT_DEBUG_PRINTLN("MQTT%d connection lost (state %d)", i + 1, slot.last_mqtt_state); + disconnectSlot(i, true); + slot.last_reconnect_attempt = now; + } else if (slotUsesJwt(i) && slot.token_expires_at != 0 && netNow() + 60 >= slot.token_expires_at) { + // Reconnect with a fresh token shortly before the current one expires. + MQTT_DEBUG_PRINTLN("MQTT%d token expiring, reconnecting", i + 1); + disconnectSlot(i, false); + slot.last_reconnect_attempt = 0; + slot.reconnect_backoff = 0; + } + continue; + } + + if (!isSlotReady(i)) continue; + if (_last_slot_connect_ms != 0 && now - _last_slot_connect_ms < CONNECT_GUARD_MS) continue; + + if (slot.initial_connect_done && slot.last_reconnect_attempt != 0) { + unsigned long elapsed = now - slot.last_reconnect_attempt; + if (slot.circuit_breaker_tripped) { + if (elapsed < CIRCUIT_BREAKER_PROBE_MS) continue; + } else { + unsigned int idx = slot.reconnect_backoff < 5 ? slot.reconnect_backoff : 4; + if (elapsed < SLOT_BACKOFF_MS[idx] + i * 3000UL) continue; + } + } + + slot.initial_connect_done = true; + slot.last_reconnect_attempt = now; + _last_slot_connect_ms = now; + + if (connectSlot(i)) { + slot.connected = true; + slot.reconnect_backoff = 0; + slot.max_backoff_failures = 0; + slot.circuit_breaker_tripped = false; + slot.last_error_time = 0; + slot.current_outage_started_ms = 0; + publishStatusToSlot(i); + } else { + if (slot.current_outage_started_ms == 0) slot.current_outage_started_ms = millis(); + if (slot.reconnect_backoff < 5) { + slot.reconnect_backoff++; + } else if (!slot.circuit_breaker_tripped && ++slot.max_backoff_failures >= MAX_FAILURES_AT_MAX_BACKOFF) { + slot.circuit_breaker_tripped = true; + MQTT_DEBUG_PRINTLN("MQTT%d circuit breaker tripped; probing every 30 min", i + 1); + } + } + return; // at most one (blocking) connect attempt per cycle + } +} + +void MQTTBridge::setSlotPreset(int slot_index, const char* preset_name) { + if (slot_index < 0 || slot_index >= RUNTIME_MQTT_SLOTS) return; + (void)preset_name; // re-read from prefs by the MQTT task + // Not running: the next begin() loads all slots from prefs anyway. + if (_initialized && _task_handle != nullptr) { + _slot_reconfigure_pending[slot_index] = true; + } +} + +void MQTTBridge::applySlotPreset(int slot_index, const char* preset_name) { + MQTTSlot& slot = _slots[slot_index]; + teardownSlot(slot_index); + slot.enabled = false; + slot.preset = nullptr; + + if (preset_name[0] == '\0' || strcmp(preset_name, MQTT_PRESET_NONE) == 0) return; + + if (strcmp(preset_name, MQTT_PRESET_CUSTOM) == 0) { + StrHelper::strncpy(slot.host, _prefs->mqtt_slot_host[slot_index], sizeof(slot.host)); + slot.port = _prefs->mqtt_slot_port[slot_index]; + StrHelper::strncpy(slot.username, _prefs->mqtt_slot_username[slot_index], sizeof(slot.username)); + StrHelper::strncpy(slot.password, _prefs->mqtt_slot_password[slot_index], sizeof(slot.password)); + StrHelper::strncpy(slot.audience, _prefs->mqtt_slot_audience[slot_index], sizeof(slot.audience)); + slot.enabled = slot.host[0] != '\0'; + } else { + const MQTTPresetDef* preset = findMQTTPreset(preset_name); + if (!preset) return; + slot.preset = preset; + slot.enabled = true; + if (mqttPresetNeedsSlotCredentials(preset)) { + StrHelper::strncpy(slot.username, _prefs->mqtt_slot_username[slot_index], sizeof(slot.username)); + StrHelper::strncpy(slot.password, _prefs->mqtt_slot_password[slot_index], sizeof(slot.password)); + } + } + + // Respect the active-slot limit (sockets/RAM). + if (slot.enabled) { + int active = 0; + for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) { + if (i != slot_index && _slots[i].enabled) active++; + } + if (active >= _max_active_slots) { + MQTT_DEBUG_PRINTLN("MQTT%d not activated: max active slots (%d) reached", slot_index + 1, _max_active_slots); + slot.enabled = false; + } + } +} + +void MQTTBridge::setSlotCustomBroker(int slot_index, const char* host, uint16_t port, + const char* username, const char* password) { + if (slot_index < 0 || slot_index >= RUNTIME_MQTT_SLOTS) return; + MQTTSlot& slot = _slots[slot_index]; + StrHelper::strncpy(slot.host, host ? host : "", sizeof(slot.host)); + slot.port = port; + StrHelper::strncpy(slot.username, username ? username : "", sizeof(slot.username)); + StrHelper::strncpy(slot.password, password ? password : "", sizeof(slot.password)); +} + +bool MQTTBridge::isIATAValid() const { + return _iata[0] != '\0' && strcmp(_iata, "XXX") != 0; +} + +bool MQTTBridge::isSlotReady(int index, char* reason_buf, size_t reason_size) const { + if (index < 0 || index >= RUNTIME_MQTT_SLOTS) return false; + const MQTTSlot& slot = _slots[index]; + if (!slot.enabled) return true; + + if (slot.preset) { + if (slot.preset->topic_style == MQTT_TOPIC_MESHRANK) { + if (_prefs->mqtt_slot_token[index][0] == '\0') { + if (reason_buf) snprintf(reason_buf, reason_size, "set mqtt%d.token ", index + 1); + return false; + } + } else if (!isIATAValid()) { + if (reason_buf) snprintf(reason_buf, reason_size, "set mqtt.iata "); + return false; + } + if (mqttPresetNeedsSlotCredentials(slot.preset)) { + if (_prefs->mqtt_slot_username[index][0] == '\0' || _prefs->mqtt_slot_password[index][0] == '\0') { + if (reason_buf) snprintf(reason_buf, reason_size, "set mqtt%d.username/password", index + 1); + return false; + } + } + } else if (_prefs->mqtt_slot_topic[index][0] == '\0' && !isIATAValid()) { + if (reason_buf) snprintf(reason_buf, reason_size, "set mqtt.iata or set mqtt%d.topic