From 6b3c543cbae04d37a14a3a8a273186ff471fb10f Mon Sep 17 00:00:00 2001 From: agessaman Date: Sun, 7 Jun 2026 11:24:39 -0700 Subject: [PATCH] Add MQTT neighbors functionality for PSRAM builds - Introduced commands for periodic neighbors publishing: `get mqtt.neighbors`, `set mqtt.neighbors on|off`, `get mqtt.neighbors.interval`, and `set mqtt.neighbors.interval `. - Updated MQTT topics to include a new neighbors topic for cached zero-hop repeater neighbors. - Added JSON message format for neighbors, including fields for SNR, last-heard age, and status. - Enhanced CLI documentation to reflect new commands and settings. - Implemented neighbor discovery logic in MyMesh class, including handling responses and publishing neighbor data. - Adjusted MQTTBridge to manage neighbors JSON buffer and publish logic. This update enhances the MQTT implementation by allowing devices to share neighbor information, improving network awareness and connectivity. --- MQTT_IMPLEMENTATION.md | 34 +++- docs/cli_commands.md | 26 +++ examples/simple_repeater/MyMesh.cpp | 250 ++++++++++++++++++++++++++++ examples/simple_repeater/MyMesh.h | 30 ++++ src/helpers/CommonCLI.h | 6 + src/helpers/CommonCLI_Observer.cpp | 31 ++++ src/helpers/MQTTDefaults.h | 2 + src/helpers/MQTTMessageBuilder.cpp | 35 ++++ src/helpers/MQTTMessageBuilder.h | 23 +++ src/helpers/bridges/MQTTBridge.cpp | 74 +++++++- src/helpers/bridges/MQTTBridge.h | 19 ++- 11 files changed, 525 insertions(+), 5 deletions(-) diff --git a/MQTT_IMPLEMENTATION.md b/MQTT_IMPLEMENTATION.md index 301b281a9..cb1d0153f 100644 --- a/MQTT_IMPLEMENTATION.md +++ b/MQTT_IMPLEMENTATION.md @@ -396,6 +396,8 @@ These settings apply across all MQTT slots: - `get mqtt.rx` - Get RX packet uplinking setting (on/off) - `get mqtt.tx` - Get TX packet uplinking setting (on/off/advert) - `get mqtt.interval` - Get status publish interval +- `get mqtt.neighbors` - Get periodic neighbors publish (PSRAM builds: on/off) +- `get mqtt.neighbors.interval` - Get neighbors publish interval in hours - `get mqtt.ntp` - Get effective NTP server hostname - `get mqtt.ntp.diag` - Probe every configured NTP server for connectivity (does not change the clock; serial console shows each server's reported time, LoRa shows a compact ` ok|fail` list) - `get mqtt.owner` - Get owner public key (serial console only) @@ -413,6 +415,8 @@ These settings apply across all MQTT slots: - `advert` - Uplink only this node's own advert packets (self-originated) - `off` - Disable TX packet uplinking - `set mqtt.interval ` - Set status publish interval (1-60 minutes) +- `set mqtt.neighbors on|off` - Enable/disable periodic neighbors/scopes publish (PSRAM builds only) +- `set mqtt.neighbors.interval ` - Set neighbors publish interval (12-8760 hours, default 24) - `set mqtt.ntp ` - Set custom NTP server (validated with immediate sync); `none` reverts to default - `set mqtt.owner <64-hex-char-public-key>` - Set owner public key - `set mqtt.email ` - Set owner email address @@ -507,7 +511,7 @@ The CLI commands are organized into two levels: ## MQTT Topics -The bridge publishes to three main topics with the following structure: +The bridge publishes to four main topics with the following structure: ### Status Topic: `meshcore/{IATA}/{DEVICE_PUBLIC_KEY}/status` Device connection status and metadata (retained messages). @@ -518,6 +522,9 @@ Full packet data with RF characteristics and metadata. ### Raw Topic: `meshcore/{IATA}/{DEVICE_PUBLIC_KEY}/raw` Minimal raw packet data for map integration. +### Neighbors Topic: `meshcore/{IATA}/{DEVICE_PUBLIC_KEY}/neighbors` +Cached zero-hop repeater neighbors with SNR, last-heard age, and flood-allowed scopes (retained, QoS 1). Published on `discover.scopes` or periodically when `mqtt.neighbors` is enabled (PSRAM observer builds only). + **Note**: `{DEVICE_PUBLIC_KEY}` is the device's public key in hexadecimal format (64 characters). ## JSON Message Formats @@ -596,6 +603,31 @@ Minimal raw packet data for map integration. } ``` +### Neighbors Message (PSRAM observer builds) +```json +{ + "timestamp": "2026-06-07T12:00:00.000000+00:00", + "origin": "MQTT Observer", + "origin_id": "DEVICE_PUBLIC_KEY", + "self": { + "scopes": "Europe,UK,France" + }, + "neighbors": [ + { + "pubkey": "NEIGHBOR_PUBLIC_KEY", + "snr": 8.5, + "heard_secs_ago": 120, + "scopes": "*,Europe", + "status": "responded" + } + ] +} +``` + +**Notes:** +- `status` per neighbor: `responded`, `timeout`, or `send_failed` (scope query only; `self` has no status field). +- Trigger with `discover.scopes` or enable periodic publish via `set mqtt.neighbors on`. + ## Key Features ### Slot-Based Preset System diff --git a/docs/cli_commands.md b/docs/cli_commands.md index b0badbcd3..def7da172 100644 --- a/docs/cli_commands.md +++ b/docs/cli_commands.md @@ -113,6 +113,17 @@ This document provides an overview of CLI commands that can be sent to MeshCore --- +### Discover neighbor scopes and publish to MQTT (PSRAM observer builds) + +**Usage:** +- `discover.scopes` + +Queries cached zero-hop repeater neighbors for flood-allowed region scopes (`ANON_REQ_TYPE_REGIONS`), then publishes a retained JSON snapshot to `meshcore/{IATA}/{device}/neighbors`. Includes this node's own scopes in the `self` object. + +**Note:** Requires `WITH_MQTT_BRIDGE`, `MAX_NEIGHBOURS`, and PSRAM (`BOARD_HAS_PSRAM`). Returns `Err - not supported (requires PSRAM)` on other builds. + +--- + ## Statistics ### Clear Stats @@ -1103,6 +1114,21 @@ Requires WiFi connected and the MQTT bridge running. --- +#### Periodic neighbors/scopes MQTT publish (PSRAM observer builds) + +**Usage:** +- `get mqtt.neighbors` +- `set mqtt.neighbors on|off` +- `get mqtt.neighbors.interval` +- `set mqtt.neighbors.interval ` + +**Parameters:** +- `hours`: Interval between automatic `discover.scopes` runs (minimum **12**, default **24**) + +**Note:** Requires PSRAM. Publishes to `meshcore/{IATA}/{device}/neighbors` (retained, QoS 1). Use `discover.scopes` for a one-shot publish. + +--- + #### View or change the speed of the bridge (RS-232 only) **Usage:** - `get bridge.baud` diff --git a/examples/simple_repeater/MyMesh.cpp b/examples/simple_repeater/MyMesh.cpp index 88a44fb58..72810b796 100644 --- a/examples/simple_repeater/MyMesh.cpp +++ b/examples/simple_repeater/MyMesh.cpp @@ -666,6 +666,16 @@ void MyMesh::onAnonDataRecv(mesh::Packet *packet, const uint8_t *secret, const m int MyMesh::searchPeersByHash(const uint8_t *hash) { int n = 0; +#if defined(WITH_MQTT_NEIGHBORS) + if (neighbor_discover_active) { + for (int i = 0; i < neighbor_discover_count && n < MAX_CLIENTS; i++) { + auto& nb = neighbours[neighbor_discover[i].neighbour_idx]; + if (nb.heard_timestamp > 0 && nb.id.isHashMatch(hash)) { + matching_peer_indexes[n++] = NEIGHBOR_DISCOVER_PEER_BASE + i; + } + } + } +#endif for (int i = 0; i < acl.getNumClients(); i++) { if (acl.getClientByIdx(i)->id.isHashMatch(hash)) { matching_peer_indexes[n++] = i; // store the INDEXES of matching contacts (for subsequent 'peer' methods) @@ -676,6 +686,15 @@ int MyMesh::searchPeersByHash(const uint8_t *hash) { void MyMesh::getPeerSharedSecret(uint8_t *dest_secret, int peer_idx) { int i = matching_peer_indexes[peer_idx]; +#if defined(WITH_MQTT_NEIGHBORS) + if (neighbor_discover_active && i >= NEIGHBOR_DISCOVER_PEER_BASE) { + int oi = i - NEIGHBOR_DISCOVER_PEER_BASE; + if (oi >= 0 && oi < neighbor_discover_count) { + self_id.calcSharedSecret(dest_secret, neighbours[neighbor_discover[oi].neighbour_idx].id); + return; + } + } +#endif if (i >= 0 && i < acl.getNumClients()) { // lookup pre-calculated shared_secret memcpy(dest_secret, acl.getClientByIdx(i)->shared_secret, PUB_KEY_SIZE); @@ -707,6 +726,15 @@ void MyMesh::onAdvertRecv(mesh::Packet *packet, const mesh::Identity &id, uint32 void MyMesh::onPeerDataRecv(mesh::Packet *packet, uint8_t type, int sender_idx, const uint8_t *secret, uint8_t *data, size_t len) { int i = matching_peer_indexes[sender_idx]; +#if defined(WITH_MQTT_NEIGHBORS) + if (neighbor_discover_active && i >= NEIGHBOR_DISCOVER_PEER_BASE) { + int oi = i - NEIGHBOR_DISCOVER_PEER_BASE; + if (type == PAYLOAD_TYPE_RESPONSE && oi >= 0 && oi < neighbor_discover_count) { + handleNeighborDiscoverResponse(oi, data, len); + } + return; + } +#endif if (i < 0 || i >= acl.getNumClients()) { // get from our known_clients table (sender SHOULD already be known in this context) MESH_DEBUG_PRINTLN("onPeerDataRecv: invalid peer idx: %d", i); return; @@ -986,6 +1014,14 @@ MyMesh::MyMesh(mesh::MainBoard &board, mesh::Radio &radio, mesh::MillisecondCloc pending_discover_tag = 0; pending_discover_until = 0; +#if defined(WITH_MQTT_NEIGHBORS) + neighbor_discover_count = 0; + neighbor_discover_active = false; + neighbor_discover_until = 0; + next_neighbors_publish = 0; + self_scopes_buf[0] = 0; +#endif + memset(default_scope.key, 0, sizeof(default_scope.key)); } @@ -1390,6 +1426,19 @@ void MyMesh::handleCommand(uint32_t sender_timestamp, char *command, char *reply sendNodeDiscoverReq(); strcpy(reply, "OK - Discover sent"); } +#if defined(WITH_MQTT_NEIGHBORS) + } else if (memcmp(command, "discover.scopes", 15) == 0) { + const char* sub = command + 15; + while (*sub == ' ') sub++; + if (*sub != 0) { + strcpy(reply, "Err - discover.scopes has no options"); + } else if (!startNeighborDiscover(reply)) { + // reply already set + } +#elif defined(WITH_MQTT_BRIDGE) + } else if (memcmp(command, "discover.scopes", 15) == 0) { + strcpy(reply, "Err - not supported (requires PSRAM)"); +#endif } else{ _cli.handleCommand(sender_timestamp, command, reply); // common CLI commands } @@ -1462,6 +1511,20 @@ void MyMesh::loop() { _alerter.onLoop(now); #endif +#if defined(WITH_MQTT_NEIGHBORS) + if (neighbor_discover_active) { + loopNeighborDiscover(); + } else if (_cli.getObserverPrefs()->mqtt_neighbors_enabled && bridge && bridge->isRunning()) { + if (next_neighbors_publish == 0 || + (next_neighbors_publish != 0 && millisHasNowPassed(next_neighbors_publish))) { + char tmp_reply[80]; + if (startNeighborDiscover(tmp_reply)) { + MESH_DEBUG_PRINTLN("%s", tmp_reply); + } + } + } +#endif + #ifdef WITH_SNMP // Push radio stats to SNMP agent every 2 seconds if (_snmp_agent.isRunning()) { @@ -1482,6 +1545,193 @@ void MyMesh::loop() { #endif } +#if defined(WITH_MQTT_NEIGHBORS) + +#include "helpers/MQTTMessageBuilder.h" +#if defined(ESP_PLATFORM) +#include +#endif + +void MyMesh::getLocalScopes(char* buf, size_t len) { + if (!buf || len == 0) return; + buf[0] = 0; + region_map.exportNamesTo(buf, (int)len, REGION_DENY_FLOOD); +} + +bool MyMesh::sendAnonRegionsReq(const mesh::Identity& target, uint32_t& tag) { + uint8_t secret[PUB_KEY_SIZE]; + self_id.calcSharedSecret(secret, target); + + tag = getRTCClock()->getCurrentTimeUnique(); + uint8_t inner[6]; + memcpy(inner, &tag, 4); + inner[4] = 0x01; // ANON_REQ_TYPE_REGIONS + inner[5] = 0x00; // zero-hop reply path + + mesh::Packet* pkt = createAnonDatagram(PAYLOAD_TYPE_ANON_REQ, self_id, target, secret, inner, sizeof(inner)); + if (!pkt) return false; + sendDirect(pkt, NULL, 0, 0); + return true; +} + +void MyMesh::handleNeighborDiscoverResponse(int overlay_idx, const uint8_t* data, size_t len) { + if (overlay_idx < 0 || overlay_idx >= neighbor_discover_count) return; + NeighborDiscoverEntry& entry = neighbor_discover[overlay_idx]; + if (entry.status != ND_PENDING || len < 8) return; + + uint32_t tag; + memcpy(&tag, data, 4); + if (tag != entry.tag) return; + + size_t scope_len = len - 8; + if (scope_len >= sizeof(entry.scopes)) { + scope_len = sizeof(entry.scopes) - 1; + } + memcpy(entry.scopes, &data[8], scope_len); + entry.scopes[scope_len] = 0; + entry.status = ND_RESPONDED; +} + +void MyMesh::finishNeighborDiscover() { + getLocalScopes(self_scopes_buf, sizeof(self_scopes_buf)); + + char self_pubkey_hex[65]; + mesh::Utils::toHex(self_pubkey_hex, self_id.pub_key, PUB_KEY_SIZE); + + char origin[32]; + MQTTBridge::getEffectiveMqttOrigin(&_prefs, _cli.getObserverPrefs(), origin, sizeof(origin)); + + char timestamp[40]; + MQTTMessageBuilder::formatIsoTimestampForMqtt(getRTCClock()->getCurrentTime(), 0, nullptr, timestamp, sizeof(timestamp)); + + char pubkey_hex[MAX_NEIGHBOURS][65]; + MQTTMessageBuilder::NeighborsMessageEntry entries[MAX_NEIGHBOURS]; + uint32_t now_secs = getRTCClock()->getCurrentTime(); + + for (int i = 0; i < neighbor_discover_count; i++) { + auto& nb = neighbours[neighbor_discover[i].neighbour_idx]; + mesh::Utils::toHex(pubkey_hex[i], nb.id.pub_key, PUB_KEY_SIZE); + entries[i].pubkey_hex = pubkey_hex[i]; + entries[i].snr = nb.snr / 4.0f; + entries[i].heard_secs_ago = (nb.heard_timestamp > 0 && now_secs >= nb.heard_timestamp) + ? (now_secs - nb.heard_timestamp) : 0; + entries[i].scopes = neighbor_discover[i].scopes; + switch (neighbor_discover[i].status) { + case ND_RESPONDED: entries[i].status = "responded"; break; + case ND_SEND_FAILED: entries[i].status = "send_failed"; break; + default: entries[i].status = "timeout"; break; + } + } + +#if defined(ESP_PLATFORM) + char* json_buf = (char*)heap_caps_malloc(MQTTBridge::NEIGHBORS_JSON_BUFFER_SIZE, MALLOC_CAP_SPIRAM); +#else + char* json_buf = (char*)malloc(MQTTBridge::NEIGHBORS_JSON_BUFFER_SIZE); +#endif + if (!json_buf) { + neighbor_discover_active = false; + neighbor_discover_count = 0; + if (_cli.getObserverPrefs()->mqtt_neighbors_enabled) { + next_neighbors_publish = futureMillis(_cli.getObserverPrefs()->mqtt_neighbors_interval); + } + return; + } + + DynamicJsonDocument doc(MQTTBridge::NEIGHBORS_JSON_BUFFER_SIZE); + int json_len = MQTTMessageBuilder::buildNeighborsMessage( + doc, origin, self_pubkey_hex, timestamp, self_scopes_buf, + entries, neighbor_discover_count, + json_buf, MQTTBridge::NEIGHBORS_JSON_BUFFER_SIZE); + + if (json_len > 0 && bridge) { + bridge->requestPublishNeighbors(json_buf, (size_t)json_len); + } + +#if defined(ESP_PLATFORM) + heap_caps_free(json_buf); +#else + free(json_buf); +#endif + + neighbor_discover_active = false; + neighbor_discover_count = 0; + if (_cli.getObserverPrefs()->mqtt_neighbors_enabled) { + next_neighbors_publish = futureMillis(_cli.getObserverPrefs()->mqtt_neighbors_interval); + } +} + +void MyMesh::loopNeighborDiscover() { + if (!neighbor_discover_active) return; + + bool all_done = true; + for (int i = 0; i < neighbor_discover_count; i++) { + if (neighbor_discover[i].status == ND_PENDING) { + all_done = false; + break; + } + } + + if (!all_done && !millisHasNowPassed(neighbor_discover_until)) { + return; + } + + if (!all_done) { + for (int i = 0; i < neighbor_discover_count; i++) { + if (neighbor_discover[i].status == ND_PENDING) { + neighbor_discover[i].status = ND_TIMEOUT; + } + } + } + finishNeighborDiscover(); +} + +bool MyMesh::startNeighborDiscover(char* reply) { + if (neighbor_discover_active) { + strcpy(reply, "Err - neighbor discover already active"); + return false; + } +#if defined(ESP_PLATFORM) + if (!psramFound()) { + strcpy(reply, "Err - PSRAM not available"); + return false; + } +#endif + if (!bridge || !bridge->isRunning()) { + strcpy(reply, "Err - MQTT bridge not running"); + return false; + } + + getLocalScopes(self_scopes_buf, sizeof(self_scopes_buf)); + neighbor_discover_count = 0; + for (int i = 0; i < MAX_NEIGHBOURS; i++) { + if (neighbours[i].heard_timestamp > 0) { + neighbor_discover[neighbor_discover_count].neighbour_idx = (uint8_t)i; + neighbor_discover[neighbor_discover_count].scopes[0] = 0; + neighbor_discover[neighbor_discover_count].status = ND_PENDING; + uint32_t tag; + if (sendAnonRegionsReq(neighbours[i].id, tag)) { + neighbor_discover[neighbor_discover_count].tag = tag; + } else { + neighbor_discover[neighbor_discover_count].status = ND_SEND_FAILED; + } + neighbor_discover_count++; + } + } + + neighbor_discover_active = true; + neighbor_discover_until = futureMillis(NEIGHBOR_DISCOVER_TIMEOUT_MS); + + if (neighbor_discover_count == 0) { + finishNeighborDiscover(); + strcpy(reply, "OK - neighbor discover started (0 neighbors, self only)"); + } else { + sprintf(reply, "OK - neighbor discover started (%u neighbors)", (unsigned)neighbor_discover_count); + } + return true; +} + +#endif // WITH_MQTT_NEIGHBORS + // To check if there is pending work bool MyMesh::hasPendingWork() const { #if defined(WITH_BRIDGE) diff --git a/examples/simple_repeater/MyMesh.h b/examples/simple_repeater/MyMesh.h index d2a45b815..c30c31208 100644 --- a/examples/simple_repeater/MyMesh.h +++ b/examples/simple_repeater/MyMesh.h @@ -136,6 +136,36 @@ class MyMesh : public mesh::Mesh, public CommonCLICallbacks { AlertReporter _alerter; #endif +#if defined(WITH_MQTT_NEIGHBORS) + enum NeighborDiscoverStatus : uint8_t { + ND_PENDING = 1, + ND_RESPONDED = 2, + ND_TIMEOUT = 3, + ND_SEND_FAILED = 4, + }; + struct NeighborDiscoverEntry { + uint8_t neighbour_idx; + uint32_t tag; + char scopes[96]; + uint8_t status; + }; + NeighborDiscoverEntry neighbor_discover[MAX_NEIGHBOURS]; + uint8_t neighbor_discover_count; + bool neighbor_discover_active; + unsigned long neighbor_discover_until; + unsigned long next_neighbors_publish; + char self_scopes_buf[96]; + + bool sendAnonRegionsReq(const mesh::Identity& target, uint32_t& tag); + bool startNeighborDiscover(char* reply); + void loopNeighborDiscover(); + void finishNeighborDiscover(); + void handleNeighborDiscoverResponse(int overlay_idx, const uint8_t* data, size_t len); + void getLocalScopes(char* buf, size_t len); + static const int NEIGHBOR_DISCOVER_PEER_BASE = 1000; + static const unsigned long NEIGHBOR_DISCOVER_TIMEOUT_MS = 30000; +#endif + void putNeighbour(const mesh::Identity& id, uint32_t timestamp, float snr); uint8_t handleLoginReq(const mesh::Identity& sender, const uint8_t* secret, uint32_t sender_timestamp, const uint8_t* data, bool is_flood); uint8_t handleAnonRegionsReq(const mesh::Identity& sender, uint32_t sender_timestamp, const uint8_t* data); diff --git a/src/helpers/CommonCLI.h b/src/helpers/CommonCLI.h index ef4a797c3..d470e7b16 100644 --- a/src/helpers/CommonCLI.h +++ b/src/helpers/CommonCLI.h @@ -156,6 +156,12 @@ struct MQTTPrefs { uint16_t alert_min_interval_min; // min minutes between same-fault alerts, default 60 char alert_hashtag[24]; // readback for `get alert.hashtag` char alert_region[31]; // optional region override; empty = default_scope + + // Neighbors publishing (PSRAM boards only). Appended at the end of the observer + // tail so a shorter /mqtt_prefs payload from earlier firmware still loads with + // these defaulting off (see applyMQTTDefaults); keeps the format at VERSION 1. + uint8_t mqtt_neighbors_enabled; // Periodic neighbors/scopes MQTT publish (PSRAM boards only) + uint32_t mqtt_neighbors_interval; // Neighbors publish interval (ms), default 24h, min 12h }; // /mqtt_prefs is written with an 8-byte header so the format is self-describing. diff --git a/src/helpers/CommonCLI_Observer.cpp b/src/helpers/CommonCLI_Observer.cpp index 22de73cfe..1166553ed 100644 --- a/src/helpers/CommonCLI_Observer.cpp +++ b/src/helpers/CommonCLI_Observer.cpp @@ -246,6 +246,26 @@ bool CommonCLI::handleObserverSetCmd(uint32_t sender_timestamp, const char* conf } else { strcpy(reply, "Error: interval must be between 1-60 minutes"); } +#if defined(BOARD_HAS_PSRAM) && defined(MAX_NEIGHBOURS) && MAX_NEIGHBOURS > 0 + } else if (memcmp(config, "mqtt.neighbors ", 14) == 0) { + _mqtt_prefs.mqtt_neighbors_enabled = memcmp(&config[14], "on", 2) == 0; + savePrefs(); + strcpy(reply, "OK"); + } else if (memcmp(config, "mqtt.neighbors.interval ", 24) == 0) { + uint32_t hours = _atoi(&config[24]); + if (hours >= 12 && hours <= 8760) { + _mqtt_prefs.mqtt_neighbors_interval = hours * 3600000UL; + savePrefs(); + sprintf(reply, "OK - neighbors interval set to %u hours (%lu ms)", hours, + (unsigned long)_mqtt_prefs.mqtt_neighbors_interval); + } else { + strcpy(reply, "Error: neighbors interval must be between 12-8760 hours"); + } +#elif defined(WITH_MQTT_BRIDGE) + } else if (memcmp(config, "mqtt.neighbors ", 14) == 0 || + memcmp(config, "mqtt.neighbors.interval ", 24) == 0) { + strcpy(reply, "Err - not supported (requires PSRAM)"); +#endif } else if (memcmp(config, "mqtt.ntp ", 9) == 0) { const char* host = &config[9]; while (*host == ' ') host++; @@ -709,6 +729,17 @@ bool CommonCLI::handleObserverGetCmd(uint32_t sender_timestamp, const char* conf } else if (memcmp(config, "mqtt.interval", 13) == 0) { uint32_t minutes = (_mqtt_prefs.mqtt_status_interval + 29999) / 60000; sprintf(reply, "> %u minutes (%lu ms)", minutes, (unsigned long)_mqtt_prefs.mqtt_status_interval); +#if defined(BOARD_HAS_PSRAM) && defined(MAX_NEIGHBOURS) && MAX_NEIGHBOURS > 0 + } else if (memcmp(config, "mqtt.neighbors.interval", 23) == 0) { + uint32_t hours = (_mqtt_prefs.mqtt_neighbors_interval + 3599999) / 3600000; + sprintf(reply, "> %u hours (%lu ms)", hours, (unsigned long)_mqtt_prefs.mqtt_neighbors_interval); + } else if (memcmp(config, "mqtt.neighbors", 14) == 0) { + sprintf(reply, "> %s", _mqtt_prefs.mqtt_neighbors_enabled ? "on" : "off"); +#elif defined(WITH_MQTT_BRIDGE) + } else if (memcmp(config, "mqtt.neighbors", 14) == 0 || + memcmp(config, "mqtt.neighbors.interval", 23) == 0) { + strcpy(reply, "Err - not supported (requires PSRAM)"); +#endif } else if (memcmp(config, "mqtt.ntp.diag", 13) == 0 && (config[13] == '\0' || config[13] == ' ')) { #ifdef ESP_PLATFORM // Connectivity probe across all configured NTP servers; never updates the clock. diff --git a/src/helpers/MQTTDefaults.h b/src/helpers/MQTTDefaults.h index f3c117eb6..2617ef7ce 100644 --- a/src/helpers/MQTTDefaults.h +++ b/src/helpers/MQTTDefaults.h @@ -68,6 +68,8 @@ static inline void applyMQTTDefaults(MQTTPrefs* prefs) { prefs->mqtt_tx_enabled = 2; prefs->mqtt_rx_enabled = 1; prefs->mqtt_status_interval = 300000; + prefs->mqtt_neighbors_enabled = 0; + prefs->mqtt_neighbors_interval = 24UL * 3600UL * 1000UL; prefs->wifi_power_save = 1; mqttDefaultSlotPreset(prefs->mqtt_slot_preset[0], sizeof(prefs->mqtt_slot_preset[0]), diff --git a/src/helpers/MQTTMessageBuilder.cpp b/src/helpers/MQTTMessageBuilder.cpp index a29b3d96c..75be1613b 100644 --- a/src/helpers/MQTTMessageBuilder.cpp +++ b/src/helpers/MQTTMessageBuilder.cpp @@ -227,6 +227,41 @@ int MQTTMessageBuilder::buildRawMessage( return (len > 0 && len < buffer_size) ? len : 0; } +int MQTTMessageBuilder::buildNeighborsMessage( + JsonDocument& doc, + const char* origin, + const char* origin_id, + const char* timestamp, + const char* self_scopes, + const NeighborsMessageEntry* neighbors, + int neighbor_count, + char* buffer, + size_t buffer_size +) { + doc.clear(); + JsonObject root = doc.to(); + + root["timestamp"] = timestamp; + root["origin"] = origin; + root["origin_id"] = origin_id; + + JsonObject self = root.createNestedObject("self"); + self["scopes"] = self_scopes ? self_scopes : ""; + + JsonArray arr = root.createNestedArray("neighbors"); + for (int i = 0; i < neighbor_count; i++) { + JsonObject nb = arr.createNestedObject(); + nb["pubkey"] = neighbors[i].pubkey_hex; + nb["snr"] = neighbors[i].snr; + nb["heard_secs_ago"] = neighbors[i].heard_secs_ago; + nb["scopes"] = neighbors[i].scopes ? neighbors[i].scopes : ""; + nb["status"] = neighbors[i].status; + } + + size_t len = serializeJson(root, buffer, buffer_size); + return (len > 0 && len < buffer_size) ? len : 0; +} + int MQTTMessageBuilder::buildPacketJSON( JsonDocument& doc, mesh::Packet* packet, diff --git a/src/helpers/MQTTMessageBuilder.h b/src/helpers/MQTTMessageBuilder.h index 8e3a6d45f..c230741ff 100644 --- a/src/helpers/MQTTMessageBuilder.h +++ b/src/helpers/MQTTMessageBuilder.h @@ -153,6 +153,29 @@ public: size_t buffer_size ); + struct NeighborsMessageEntry { + const char* pubkey_hex; + float snr; + uint32_t heard_secs_ago; + const char* scopes; + const char* status; + }; + + /** + * Build neighbors table JSON for meshcore/{iata}/{device}/neighbors topic. + */ + static int buildNeighborsMessage( + JsonDocument& doc, + const char* origin, + const char* origin_id, + const char* timestamp, + const char* self_scopes, + const NeighborsMessageEntry* neighbors, + int neighbor_count, + char* buffer, + size_t buffer_size + ); + /** * Convert packet to JSON message * diff --git a/src/helpers/bridges/MQTTBridge.cpp b/src/helpers/bridges/MQTTBridge.cpp index eeae09d23..fe09024b3 100644 --- a/src/helpers/bridges/MQTTBridge.cpp +++ b/src/helpers/bridges/MQTTBridge.cpp @@ -506,6 +506,11 @@ MQTTBridge::MQTTBridge(NodePrefs *prefs, MQTTPrefs *obs, mesh::PacketManager *mg _last_raw_data = (uint8_t*)psram_malloc(LAST_RAW_DATA_SIZE); _publish_json_buffer = (char*)psram_malloc(PUBLISH_JSON_BUFFER_SIZE); _status_json_buffer = (char*)psram_malloc(STATUS_JSON_BUFFER_SIZE); + #if defined(WITH_MQTT_NEIGHBORS) + _neighbors_json_buffer = (char*)psram_malloc(NEIGHBORS_JSON_BUFFER_SIZE); + _neighbors_publish_pending = false; + _neighbors_publish_len = 0; + #endif #else memset(_last_raw_data, 0, sizeof(_last_raw_data)); #endif @@ -804,6 +809,11 @@ void MQTTBridge::end() { psram_free(_last_raw_data); _last_raw_data = nullptr; psram_free(_publish_json_buffer); _publish_json_buffer = nullptr; psram_free(_status_json_buffer); _status_json_buffer = nullptr; + #if defined(WITH_MQTT_NEIGHBORS) + psram_free(_neighbors_json_buffer); _neighbors_json_buffer = nullptr; + _neighbors_publish_pending = false; + _neighbors_publish_len = 0; + #endif #endif // JSON documents are now StaticJsonDocument inline members — no heap allocation to free. @@ -1023,6 +1033,17 @@ void MQTTBridge::mqttTaskLoop() { // Process packet queue processPacketQueue(); +#if defined(WITH_MQTT_NEIGHBORS) + if (_neighbors_publish_pending) { + _neighbors_publish_pending = false; + if (publishNeighbors()) { + MQTT_DEBUG_PRINTLN("Neighbors published"); + } else { + MQTT_DEBUG_PRINTLN("Neighbors publish failed"); + } + } +#endif + #ifdef WITH_SNMP // SNMP agent loop — process incoming UDP requests if (_snmp_agent) { @@ -1820,7 +1841,10 @@ bool MQTTBridge::publishToAllSlots(const char* topic, const char* payload, bool // Presets use hardcoded topic logic; custom slots support user-defined templates. // --------------------------------------------------------------------------- bool MQTTBridge::substituteTopicTemplate(const char* tmpl, MQTTMessageType type, int slot_index, char* buf, size_t buf_size) { - const char* type_str = (type == MSG_STATUS) ? "status" : (type == MSG_PACKETS) ? "packets" : "raw"; + const char* type_str = (type == MSG_STATUS) ? "status" + : (type == MSG_PACKETS) ? "packets" + : (type == MSG_RAW) ? "raw" + : "neighbors"; const char* token = _obs->mqtt_slot_token[slot_index]; size_t out = 0; @@ -1878,7 +1902,10 @@ bool MQTTBridge::buildTopicForSlot(int index, MQTTMessageType type, char* topic_ } // MQTT_TOPIC_MESHCORE (default for all other presets) if (!isIATAValid()) return false; - const char* type_str = (type == MSG_STATUS) ? "status" : (type == MSG_PACKETS) ? "packets" : "raw"; + const char* type_str = (type == MSG_STATUS) ? "status" + : (type == MSG_PACKETS) ? "packets" + : (type == MSG_RAW) ? "raw" + : "neighbors"; snprintf(topic_buf, buf_size, "meshcore/%s/%s/%s", _iata, _device_id, type_str); return true; } @@ -1889,7 +1916,10 @@ bool MQTTBridge::buildTopicForSlot(int index, MQTTMessageType type, char* topic_ } // Default: meshcore format if (!isIATAValid()) return false; - const char* type_str = (type == MSG_STATUS) ? "status" : (type == MSG_PACKETS) ? "packets" : "raw"; + const char* type_str = (type == MSG_STATUS) ? "status" + : (type == MSG_PACKETS) ? "packets" + : (type == MSG_RAW) ? "raw" + : "neighbors"; snprintf(topic_buf, buf_size, "meshcore/%s/%s/%s", _iata, _device_id, type_str); return true; } @@ -2716,6 +2746,44 @@ bool MQTTBridge::publishStatus() { return false; } +#if defined(WITH_MQTT_NEIGHBORS) +void MQTTBridge::requestPublishNeighbors(const char* json, size_t len) { + if (!_neighbors_json_buffer || !json || len == 0) return; + if (len >= NEIGHBORS_JSON_BUFFER_SIZE) { + len = NEIGHBORS_JSON_BUFFER_SIZE - 1; + } + memcpy(_neighbors_json_buffer, json, len); + _neighbors_json_buffer[len] = '\0'; + _neighbors_publish_len = len; + _neighbors_publish_pending = true; +} + +bool MQTTBridge::publishNeighbors() { + if (!_neighbors_json_buffer || _neighbors_publish_len == 0) { + return false; + } + if (!_cached_has_connected_slots) { + return false; + } + + refreshOriginFromPrefs(); + + bool published = false; + char topic[128]; + for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) { + if (_slots[i].enabled && _slots[i].client && _slots[i].connected) { + if (buildTopicForSlot(i, MSG_NEIGHBORS, topic, sizeof(topic))) { + bool use_retain = _slots[i].preset ? _slots[i].preset->allow_retain : false; + if (publishToSlot(i, topic, _neighbors_json_buffer, use_retain, 1)) { + published = true; + } + } + } + } + return published; +} +#endif + bool MQTTBridge::publishPacket(mesh::Packet* packet, bool is_tx, const uint8_t* raw_data, int raw_len, float snr, float rssi) { diff --git a/src/helpers/bridges/MQTTBridge.h b/src/helpers/bridges/MQTTBridge.h index ee5b8b988..70de88ad2 100644 --- a/src/helpers/bridges/MQTTBridge.h +++ b/src/helpers/bridges/MQTTBridge.h @@ -35,6 +35,10 @@ class MeshSNMPAgent; // Forward declaration #ifdef WITH_MQTT_BRIDGE +#if defined(BOARD_HAS_PSRAM) && defined(MAX_NEIGHBOURS) && MAX_NEIGHBOURS > 0 +#define WITH_MQTT_NEIGHBORS 1 +#endif + /** * @brief Bridge implementation using MQTT protocol for packet transport * @@ -257,6 +261,11 @@ private: // class object so these allocations don't interleave with large TLS buffers at startup. static const size_t PUBLISH_JSON_BUFFER_SIZE = 2048; static const size_t STATUS_JSON_BUFFER_SIZE = 768; + #if defined(WITH_MQTT_NEIGHBORS) + char* _neighbors_json_buffer; + size_t _neighbors_publish_len; + bool _neighbors_publish_pending; + #endif #if defined(BOARD_HAS_PSRAM) char* _publish_json_buffer; char* _status_json_buffer; @@ -318,7 +327,7 @@ private: mesh::MillisecondClock* _ms; // For uptime // Topic building - enum MQTTMessageType { MSG_STATUS, MSG_PACKETS, MSG_RAW }; + enum MQTTMessageType { MSG_STATUS, MSG_PACKETS, MSG_RAW, MSG_NEIGHBORS }; bool buildTopicForSlot(int index, MQTTMessageType type, char* topic_buf, size_t buf_size); bool substituteTopicTemplate(const char* tmpl, MQTTMessageType type, int slot_index, char* buf, size_t buf_size); @@ -349,6 +358,9 @@ private: void processPacketQueue(); bool publishStatus(); // Returns true if status was successfully published + #if defined(WITH_MQTT_NEIGHBORS) + bool publishNeighbors(); + #endif bool handleWiFiConnection(unsigned long now); // FreeRTOS task function (runs on Core 0) @@ -477,6 +489,11 @@ public: static void formatSlotDiagReply(char* buf, size_t bufsize, int slot_index); static uint8_t getLastWifiDisconnectReason(); static unsigned long getLastWifiDisconnectTime(); + + #if defined(WITH_MQTT_NEIGHBORS) + void requestPublishNeighbors(const char* json, size_t len); + static const size_t NEIGHBORS_JSON_BUFFER_SIZE = 10240; + #endif static const char* wifiReasonStr(uint8_t reason); static const char* tlsErrorStr(int32_t err);