feat(mqtt): per-slot IATA override (mqttN.iata)

MQTT_IMPLEMENTATION.md

@@ -64,7 +64,8 @@ set mqtt3.preset none         # disable the slot
 
 Broker address, transport, and credentials ship in the firmware. The **Extra setup**
 column lists the presets that need more. Every preset except `meshrank` uses the
-`meshcore/{iata}/…` topic layout, so `mqtt.iata` must be set.
+`meshcore/{iata}/…` topic layout, so `mqtt.iata` must be set (or `mqttN.iata` for
+that slot, see [Per-slot region codes](#per-slot-region-codes)).
 
 `get mqtt.presets` lists what the running firmware ships. The table matches the current build.
 
@@ -147,6 +148,7 @@ Commands work over serial and over a remote admin login unless marked *serial on
 | `get/set mqttN.token <token>` | Per-slot token, e.g. MeshRank; `get` shows the value on serial only |
 | `get/set mqttN.topic <template>` | Topic template, `custom` slots only |
 | `get/set mqttN.audience <aud>` | Switches a `custom` slot to JWT auth; bare `set` clears it |
+| `get/set mqttN.iata <code>` | Region code for this slot's topics; bare `set` clears it back to `mqtt.iata` |
 | `get/set mqttN.filter <all\|none\|list>` | Packet types sent to this slot |
 | `get mqttN.diag` | Last connection error and timing for the slot |
 
@@ -228,11 +230,29 @@ or the scheme's default.
 
 | Placeholder | Value |
 |-------------|-------|
-| `{iata}` | `mqtt.iata` |
+| `{iata}` | `mqttN.iata`, else `mqtt.iata` |
 | `{device}` | public key, 64 hex chars |
 | `{token}` | `mqttN.token` |
 | `{type}` | `status`, `packets`, `raw`, or `neighbors` |
 
+### Per-slot region codes
+
+Brokers don't all draw regions the same way. A code one network expects can be
+unknown to another, and some brokers reject codes they don't recognize.
+`mqttN.iata` lets one observer publish to each slot under the code that network
+uses:
+
+```bash
+set mqtt.iata SAT          # node-wide default, used by every slot without an override
+set mqtt2.iata CRP         # slot 2 publishes meshcore/CRP/... instead
+get mqtt2.iata             # > CRP
+set mqtt2.iata             # clear: slot 2 follows mqtt.iata again
+```
+
+The override applies to every `meshcore/{iata}/…` topic the slot publishes and to
+`{iata}` in a `custom` topic template. MeshRank topics don't carry a region code, so
+it has no effect there. An empty or invalid override falls back to `mqtt.iata`.
+
 ### MQTT Shared Settings
 
 | Command | Purpose | Default |

scripts/webconfig_mock_server.py

@@ -144,7 +144,7 @@ def default_config(setup_mode):
 def _slot():
     return {"preset": "none", "server": "", "port": 8883, "username": "",
             "password": "", "token": "", "topic": "", "audience": "",
-            "filter": "all"}
+            "filter": "all", "iata": ""}
 
 
 class State:
@@ -478,6 +478,14 @@ def apply_slot_set(cfg, idx, field, val):
             return False, "Error: port must be between 1 and 65535"
         slot["port"] = p
         return True, "OK"
+    if field == "iata":
+        if val == "":
+            slot["iata"] = ""
+            return True, "OK - slot %d IATA cleared (using mqtt.iata)" % (idx + 1)
+        if len(val) != 3 or not val.isalnum() or not val.isascii():
+            return False, "Error: IATA code must be exactly 3 letters/digits (e.g. DEN)"
+        slot["iata"] = val.upper()
+        return True, "OK - slot %d IATA: %s" % (idx + 1, slot["iata"])
     if field == "filter":
         canonical = canonical_packet_filter(val)
         if canonical is None:

src/helpers/CommonCLI_Observer.cpp

@@ -31,6 +31,7 @@
 #include "MQTTConnectionPolicy.h"  // classifySlotActivation() — "will this slot connect here?"
 #include "MQTTDefaults.h"
 #include "MQTTPacketFilter.h"
+#include "MQTTTopicRouter.h"   // mqttSlotIata() — a slot's effective region code
 #endif
 
 // Local copy of the busted-libc-safe atoi (the original in CommonCLI.cpp is static).
@@ -510,10 +511,11 @@ bool CommonCLI::handleObserverSetCmd(uint32_t sender_timestamp, const char* conf
           _callbacks->restartBridgeSlot(slot);
           // Check if the slot has everything it needs to connect
           const MQTTPresetDef* p = findMQTTPreset(preset_name);
+          const char* slot_iata = mqttSlotIata(_mqtt_prefs.mqtt_iata, _mqtt_prefs.mqtt_slot_iata[slot]);
           if (p && p->topic_style == MQTT_TOPIC_MESHRANK && _mqtt_prefs.mqtt_slot_token[slot][0] == '\0') {
             sprintf(reply, "OK - slot %d preset: %s (run 'set mqtt%d.token <your_token>' to connect)", slot + 1, preset_name, slot + 1);
           } else if (p && p->topic_style == MQTT_TOPIC_MESHCORE &&
-                     (strlen(_mqtt_prefs.mqtt_iata) == 0 || strcmp(_mqtt_prefs.mqtt_iata, "XXX") == 0)) {
+                     (strlen(slot_iata) == 0 || strcmp(slot_iata, "XXX") == 0)) {
             sprintf(reply, "OK - slot %d preset: %s (run 'set mqtt.iata <airport_code>' to publish)", slot + 1, preset_name);
           } else if (p && mqttPresetNeedsSlotPassword(p) &&
                      _mqtt_prefs.mqtt_slot_password[slot][0] == '\0' &&
@@ -610,6 +612,26 @@ bool CommonCLI::handleObserverSetCmd(uint32_t sender_timestamp, const char* conf
         _callbacks->restartBridgeSlot(slot);
         sprintf(reply, "OK - slot %d topic: %s", slot + 1, _mqtt_prefs.mqtt_slot_topic[slot]);
       }
+    } else if (strcmp(subcmd, "iata") == 0 || memcmp(subcmd, "iata ", 5) == 0) {
+      // Per-slot region code for MeshCore-style topics. Bare or empty clears
+      // the override so the slot follows mqtt.iata again.
+      const char* iata = subcmd[4] == '\0' ? "" : &subcmd[5];
+      if (iata[0] == '\0') {
+        _mqtt_prefs.mqtt_slot_iata[slot][0] = '\0';
+        if (!persistObserverPrefs(reply)) return true;
+        _callbacks->restartBridgeSlot(slot);
+        sprintf(reply, "OK - slot %d IATA cleared (using mqtt.iata)", slot + 1);
+      } else if (!mqttIataValid(iata)) {
+        strcpy(reply, "Error: IATA code must be exactly 3 letters/digits (e.g. DEN)");
+      } else {
+        StrHelper::strncpy(_mqtt_prefs.mqtt_slot_iata[slot], iata, sizeof(_mqtt_prefs.mqtt_slot_iata[slot]));
+        for (int i = 0; _mqtt_prefs.mqtt_slot_iata[slot][i]; i++) {
+          _mqtt_prefs.mqtt_slot_iata[slot][i] = toupper(_mqtt_prefs.mqtt_slot_iata[slot][i]);
+        }
+        if (!persistObserverPrefs(reply)) return true;
+        _callbacks->restartBridgeSlot(slot);
+        sprintf(reply, "OK - slot %d IATA: %s", slot + 1, _mqtt_prefs.mqtt_slot_iata[slot]);
+      }
     } else if (memcmp(subcmd, "audience ", 9) == 0) {
       if (valueTooLong(&subcmd[9], sizeof(_mqtt_prefs.mqtt_slot_audience[slot]), reply, "audience")) return true;
       StrHelper::strncpy(_mqtt_prefs.mqtt_slot_audience[slot], &subcmd[9], sizeof(_mqtt_prefs.mqtt_slot_audience[slot]));
@@ -1005,6 +1027,13 @@ bool CommonCLI::handleObserverGetCmd(uint32_t sender_timestamp, const char* conf
       } else {
         strcpy(reply, "> (default: meshcore/{iata}/{device}/{type})");
       }
+    } else if (strcmp(subcmd, "iata") == 0) {
+      if (_mqtt_prefs.mqtt_slot_iata[slot][0] != '\0') {
+        sprintf(reply, "> %s", _mqtt_prefs.mqtt_slot_iata[slot]);
+      } else {
+        sprintf(reply, "> (not set - using mqtt.iata: %s)",
+                _mqtt_prefs.mqtt_iata[0] ? _mqtt_prefs.mqtt_iata : "not set");
+      }
     } else if (memcmp(subcmd, "audience", 8) == 0) {
       if (_mqtt_prefs.mqtt_slot_audience[slot][0] != '\0') {
         sprintf(reply, "> %s", _mqtt_prefs.mqtt_slot_audience[slot]);

src/helpers/MQTTPrefsSerializer.h

@@ -184,6 +184,7 @@ class MQTTPrefsSerializer : public ConfigSerializer {
     bool _seen_preset = false, _seen_host = false, _seen_port = false;
     bool _seen_username = false, _seen_password = false, _seen_token = false;
     bool _seen_topic = false, _seen_audience = false, _seen_filter = false;
+    bool _seen_iata = false;
   protected:
     void structure() override {
       defStrict("preset", _prefs->mqtt_slot_preset[_index],
@@ -202,6 +203,8 @@ class MQTTPrefsSerializer : public ConfigSerializer {
       defStrict("audience", _prefs->mqtt_slot_audience[_index],
                 sizeof(_prefs->mqtt_slot_audience[_index]), _seen_audience);
       defStrict("packet_filter", _filter, _seen_filter);
+      defStrict("iata", _prefs->mqtt_slot_iata[_index],
+                sizeof(_prefs->mqtt_slot_iata[_index]), _seen_iata);
     }
   public:
     SlotPrefs(MQTTPrefs* prefs, int index)
@@ -218,6 +221,22 @@ class MQTTPrefsSerializer : public ConfigSerializer {
       }
       _prefs->mqtt_slot_port[_index] = static_cast<uint16_t>(_port);
       _prefs->mqtt_slot_packet_filter[_index] = static_cast<uint16_t>(_filter);
+      // An unusable override is dropped rather than repaired to a default: the
+      // slot then falls back to the global IATA, the same as never setting one.
+      char* iata = _prefs->mqtt_slot_iata[_index];
+      if (iata[0] != '\0') {
+        if (!mqttIataValid(iata)) {
+          iata[0] = '\0';
+          *repaired = true;
+        } else {
+          for (char* p = iata; *p; ++p) {
+            if (*p >= 'a' && *p <= 'z') {
+              *p = static_cast<char>(*p - ('a' - 'A'));
+              *repaired = true;
+            }
+          }
+        }
+      }
     }
   };
 

src/helpers/MQTTPrefsStorage.h

@@ -135,6 +135,11 @@ struct MQTTPrefs {
   // Drop received adverts whose Ed25519 signature fails before publishing.
   // Defaults off so RF corruption on the mesh stays visible to collectors.
   uint8_t mqtt_advert_verify;
+
+  // Per-slot IATA override for MeshCore-style topics. Empty uses mqtt_iata, so
+  // one observer can publish under different region codes to brokers whose
+  // region maps differ. Runtime only, like display_timeout_secs.
+  char mqtt_slot_iata[MQTT_PREFS_SLOT_COUNT][8];
 };
 
 static inline bool mqttNetworkSetupComplete(const MQTTPrefs* prefs) {

src/helpers/MQTTTopicRouter.h

@@ -27,6 +27,17 @@ static inline bool mqttTopicSlotIndexValid(int index, size_t slot_count) {
   return index >= 0 && (size_t)index < slot_count;
 }
 
+// The region code a slot publishes under: its own override when one is set,
+// otherwise the node-wide mqtt.iata. The override wins only when it is a usable
+// code, so a stale or invalid value can't stop a slot that would otherwise
+// publish under the global one.
+static inline const char* mqttSlotIata(const char* global_iata, const char* slot_iata) {
+  if (slot_iata && mqttIataValid(slot_iata) && strcmp(slot_iata, "XXX") != 0) {
+    return slot_iata;
+  }
+  return global_iata ? global_iata : "";
+}
+
 static inline const char* mqttPublicationTypeName(int type) {
   switch (type) {
     case MQTT_PUBLICATION_STATUS: return "status";

src/helpers/WebConfigKeys.h

@@ -31,7 +31,7 @@ static const char* const WC_ALLOWED_SET_KEYS[] = {
 };
 static const char* const WC_ALLOWED_SLOT_KEYS[] = {
   "preset", "server", "port", "username", "password", "token", "topic", "audience",
-  "filter",
+  "filter", "iata",
 };
 
 // True when `key` is a well-formed per-slot key ("mqttN.<field>" with N in

src/helpers/bridges/MQTTBridge.cpp

@@ -3059,7 +3059,7 @@ bool MQTTBridge::publishToAllSlots(const char* topic, const char* payload, size_
 // ---------------------------------------------------------------------------
 bool MQTTBridge::substituteTopicTemplate(const char* tmpl, MQTTMessageType type, int slot_index, char* buf, size_t buf_size) {
   return mqttBuildPublicationTopic(MQTT_ROUTE_CUSTOM, (int)type, tmpl,
-                                   _iata, _device_id, _obs->mqtt_slot_token[slot_index],
+                                   slotIata(slot_index), _device_id, _obs->mqtt_slot_token[slot_index],
                                    buf, buf_size);
 }
 
@@ -3079,7 +3079,7 @@ bool MQTTBridge::buildTopicForSlot(int index, MQTTMessageType type, char* topic_
     MQTTTopicRouteStyle style = (slot.preset->topic_style == MQTT_TOPIC_MESHRANK)
         ? MQTT_ROUTE_MESHRANK : MQTT_ROUTE_MESHCORE;
     return mqttBuildPublicationTopic(style, (int)type, nullptr,
-                                     _iata, _device_id, _obs->mqtt_slot_token[index],
+                                     slotIata(index), _device_id, _obs->mqtt_slot_token[index],
                                      topic_buf, buf_size);
   }
 
@@ -3089,7 +3089,7 @@ bool MQTTBridge::buildTopicForSlot(int index, MQTTMessageType type, char* topic_
   }
   // Default: meshcore format
   return mqttBuildPublicationTopic(MQTT_ROUTE_MESHCORE, (int)type, nullptr,
-                                   _iata, _device_id, _obs->mqtt_slot_token[index],
+                                   slotIata(index), _device_id, _obs->mqtt_slot_token[index],
                                    topic_buf, buf_size);
 }
 
@@ -3448,11 +3448,14 @@ bool MQTTBridge::isReady() const {
   return _initialized && isNetworkConfigValid(_obs);
 }
 
-bool MQTTBridge::isIATAValid() const {
-  if (strlen(_iata) == 0 || strcmp(_iata, "XXX") == 0) {
-    return false;
-  }
-  return true;
+const char* MQTTBridge::slotIata(int index) const {
+  if (!mqttTopicSlotIndexValid(index, RUNTIME_MQTT_SLOTS)) return _iata;
+  return mqttSlotIata(_iata, _obs->mqtt_slot_iata[index]);
+}
+
+bool MQTTBridge::isSlotIATAValid(int index) const {
+  const char* iata = slotIata(index);
+  return iata[0] != '\0' && strcmp(iata, "XXX") != 0;
 }
 
 bool MQTTBridge::isSlotReady(int index, char* reason_buf, size_t reason_size) const {
@@ -3468,7 +3471,7 @@ bool MQTTBridge::isSlotReady(int index, char* reason_buf, size_t reason_size) co
         return false;
       }
     } else if (slot.preset->topic_style == MQTT_TOPIC_MESHCORE) {
-      if (!isIATAValid()) {
+      if (!isSlotIATAValid(index)) {
         if (reason_buf) snprintf(reason_buf, reason_size, "set mqtt.iata <airport_code>");
         return false;
       }
@@ -3485,7 +3488,7 @@ bool MQTTBridge::isSlotReady(int index, char* reason_buf, size_t reason_size) co
     }
   } else {
     // Custom slot without a topic template uses meshcore format, needs IATA
-    if (_obs->mqtt_slot_topic[index][0] == '\0' && !isIATAValid()) {
+    if (_obs->mqtt_slot_topic[index][0] == '\0' && !isSlotIATAValid(index)) {
       if (reason_buf) snprintf(reason_buf, reason_size, "set mqtt.iata <airport_code> or set mqtt%d.topic <template>", index + 1);
       return false;
     }

src/helpers/bridges/MQTTBridge.h

@@ -603,7 +603,9 @@ class MQTTBridge : public BridgeBase {
   // (callers should fall back to UTC). Zero-allocation.
   static bool timezoneRulesFromString(const char* tz_string, TimeChangeRule& dst_out, TimeChangeRule& std_out);
   void checkConfigurationMismatch();
-  bool isIATAValid() const;
+  // Region code for one slot's MeshCore topics: mqttN.iata, else mqtt.iata.
+  const char* slotIata(int index) const;
+  bool isSlotIATAValid(int index) const;
   bool isSlotReady(int index, char* reason_buf = nullptr, size_t reason_size = 0) const;
 
   void optimizeMqttClientConfig(PsychicMqttClient* client, bool needs_large_buffer = false);

src/helpers/esp32/WebConfigServer.cpp

@@ -987,6 +987,7 @@ void WebConfigServer::handleConfigGet(AsyncWebServerRequest* req) {
       s["token"] = _obs->mqtt_slot_token[i][0] ? SECRET_SENTINEL : "";
       s["topic"] = (const char*)_obs->mqtt_slot_topic[i];
       s["audience"] = (const char*)_obs->mqtt_slot_audience[i];
+      s["iata"] = (const char*)_obs->mqtt_slot_iata[i];
       char filter_text[MQTTPacketFilter::kFilterTextSize];
       if (MQTTPacketFilter::format(_obs->mqtt_slot_packet_filter[i],
                                    filter_text, sizeof(filter_text))) {

test/test_mqtt_prefs_serializer/test_mqtt_prefs_serializer.cpp

@@ -536,6 +536,62 @@ TEST(MQTTPrefsSerializer, UnknownKeysInsideAKnownGroupAreDowngradeSafe) {
   EXPECT_EQ(2, prefs.wifi_power_save);
 }
 
+TEST(MQTTPrefsSerializer, SlotIataRoundTripsAndDefaultsEmpty) {
+  MQTTPrefs source = defaults();
+  strcpy(source.mqtt_iata, "SAT");
+  strcpy(source.mqtt_slot_iata[1], "CRP");
+
+  OutputStream output;
+  MQTTPrefsSerializer writer(&source);
+  ASSERT_TRUE(writer.saveSerial(output));
+  EXPECT_NE(std::string::npos, output.text().find("iata:\"CRP\"")) << output.text();
+
+  MQTTPrefs loaded = defaults();
+  InputStream input(output.text());
+  MQTTPrefsSerializer reader(&loaded);
+  ASSERT_TRUE(reader.loadSerial(input)) << output.text();
+  bool repaired = true;
+  ASSERT_TRUE(reader.apply(&repaired));
+  EXPECT_FALSE(repaired);
+  EXPECT_STREQ("SAT", loaded.mqtt_iata);
+  EXPECT_STREQ("CRP", loaded.mqtt_slot_iata[1]);
+  for (int i = 0; i < MQTT_PREFS_SLOT_COUNT; ++i) {
+    if (i != 1) EXPECT_STREQ("", loaded.mqtt_slot_iata[i]) << i;
+  }
+}
+
+TEST(MQTTPrefsSerializer, PrefsWrittenBeforeSlotIataStillLoad) {
+  // Upgrade path: a slot group without the key keeps an empty override, so the
+  // slot follows mqtt.iata exactly as it did before.
+  MQTTPrefs prefs = defaults();
+  InputStream input(
+      "{version:1,mqtt:{iata:\"SEA\",slot1:{preset:\"analyzer-us\"}}}");
+  MQTTPrefsSerializer serializer(&prefs);
+  ASSERT_TRUE(serializer.loadSerial(input));
+  bool repaired = false;
+  ASSERT_TRUE(serializer.apply(&repaired));
+  EXPECT_FALSE(repaired);
+  EXPECT_STREQ("", prefs.mqtt_slot_iata[0]);
+}
+
+TEST(MQTTPrefsSerializer, RepairsSlotIataCaseAndDropsInvalidOverrides) {
+  MQTTPrefs prefs = defaults();
+  InputStream input(
+      "{version:1,mqtt:{iata:\"SEA\",slot1:{iata:\"den\"},"
+      "slot2:{iata:\"toolong\"},slot3:{iata:\"a/b\"}}}");
+  MQTTPrefsSerializer serializer(&prefs);
+  ASSERT_TRUE(serializer.loadSerial(input));
+  bool repaired = false;
+  ASSERT_TRUE(serializer.apply(&repaired));
+  EXPECT_TRUE(repaired);
+  EXPECT_STREQ("DEN", prefs.mqtt_slot_iata[0]);
+  // Invalid overrides are cleared, not repaired to a default, so those slots
+  // fall back to the node-wide code.
+  EXPECT_STREQ("", prefs.mqtt_slot_iata[1]);
+  EXPECT_STREQ("", prefs.mqtt_slot_iata[2]);
+  EXPECT_STREQ("SEA", prefs.mqtt_iata);
+}
+
 int main(int argc, char** argv) {
   ::testing::InitGoogleTest(&argc, argv);
   return RUN_ALL_TESTS();

test/test_mqtt_topic_router/test_mqtt_topic_router.cpp

@@ -192,6 +192,30 @@ TEST(MQTTTopicRouter, PublicationTypeEnumValuesAreFrozen) {
   EXPECT_STREQ("neighbors", mqttPublicationTypeName(MQTT_PUBLICATION_NEIGHBORS));
 }
 
+TEST(MQTTTopicRouter, SlotIataOverridesTheGlobalCodeOnlyWhenUsable) {
+  EXPECT_STREQ("CRP", mqttSlotIata("SAT", "CRP"));
+  EXPECT_STREQ("SAT", mqttSlotIata("SAT", ""));
+  EXPECT_STREQ("SAT", mqttSlotIata("SAT", nullptr));
+  EXPECT_STREQ("SAT", mqttSlotIata("SAT", "XXX"));
+  EXPECT_STREQ("SAT", mqttSlotIata("SAT", "TOOLONG"));
+  EXPECT_STREQ("SAT", mqttSlotIata("SAT", "a/b"));
+  // The override still works with no node-wide code configured.
+  EXPECT_STREQ("CRP", mqttSlotIata("", "CRP"));
+  EXPECT_STREQ("", mqttSlotIata(nullptr, nullptr));
+}
+
+TEST(MQTTTopicRouter, SlotIataFeedsMeshCoreAndCustomTopics) {
+  char topic[128];
+  const char* iata = mqttSlotIata("SAT", "CRP");
+  ASSERT_TRUE(mqttBuildPublicationTopic(MQTT_ROUTE_MESHCORE, MQTT_PUBLICATION_PACKETS,
+                                        nullptr, iata, DEVICE, TOKEN, topic, sizeof(topic)));
+  EXPECT_STREQ("meshcore/CRP/0123456789ABCDEF/packets", topic);
+  ASSERT_TRUE(mqttBuildPublicationTopic(MQTT_ROUTE_CUSTOM, MQTT_PUBLICATION_STATUS,
+                                        "mesh/{iata}/{type}", iata, DEVICE, TOKEN,
+                                        topic, sizeof(topic)));
+  EXPECT_STREQ("mesh/CRP/status", topic);
+}
+
 }  // namespace
 
 int main(int argc, char** argv) {

test/test_webconfig_keys/test_webconfig_keys.cpp

@@ -26,6 +26,7 @@ TEST(WebConfigKeys, AllowsPerSlotKeys) {
   EXPECT_TRUE(wcIsAllowedSetKey("mqtt1.token"));
   EXPECT_TRUE(wcIsAllowedSetKey("mqtt2.filter"));
   EXPECT_TRUE(wcIsAllowedSetKey("mqtt6.audience"));  // MAX_MQTT_SLOTS == 6
+  EXPECT_TRUE(wcIsAllowedSetKey("mqtt2.iata"));
 }
 
 TEST(WebConfigKeys, RejectsDangerousOrUnknownKeys) {