feat(mqtt): add per-slot packet filters for MQTT slots

Introduce per-slot packet filters to allow users to specify which
packet types are uploaded for each MQTT slot. This feature enhances
the flexibility of the MQTT bridge by enabling users to configure
allowlists for packet types, improving the efficiency of data
transmissions. The implementation includes updates to the WebConfig
interface, internal handling of packet filters, and necessary
modifications to the MQTT preferences structure.

test/README.md

@@ -30,6 +30,7 @@ does not reflect the GoogleTest count — run the built binary directly
 | `test_mqtt_topic_router` | `src/helpers/MQTTTopicRouter.h` | complete preset/custom topic-routing contract; MeshRank packets-only behavior; required identifiers; invalid inputs/slots; exact buffer boundaries |
 | `test_mqtt_connection_policy` | `src/helpers/MQTTConnectionPolicy.h` | reconnect guard/backoff/stagger and breaker transitions; stable reset; JWT lifetime/renewal policy; exact timing boundaries and 32-bit `millis()` rollover |
 | `test_mqtt_packet_queue_policy` | `src/helpers/MQTTPacketQueuePolicy.h` | queue-full eviction; stale-disconnect flush; adaptive drain limits; bounded QoS0 retries; exact timing boundaries and 32-bit `millis()` rollover |
+| `test_mqtt_packet_filter` | `src/helpers/MQTTPacketFilter.h` | per-slot 0-15 allowlist parsing/formatting; exact bounds; membership; eligible-target and retry-completion policy |
 | `test_mqtt_runtime_buffer_lifecycle` | `src/helpers/MQTTRuntimeBufferLifecycle.h` | idempotent allocation/release; partial-allocation degradation; retry of only missing buffers |
 | `test_mqtt_prefs_codec` | `src/helpers/MQTTPrefsStorage.h`, `src/helpers/MQTTPrefsCodec.h` | binary pre-slot/3-slot/6-slot migration fixtures; v1 header integrity; downgrade preservation |
 | `test_mqtt_prefs_atomic_store` | `src/helpers/MQTTPrefsAtomicStore.h` | transactional MQTT writes and legacy `/node_prefs` handoff; exact short-write detection; begin/finish/rename failure cleanup; original-file preservation |

test/test_mqtt_packet_filter/test_mqtt_packet_filter.cpp

@@ -0,0 +1,143 @@
+#include <gtest/gtest.h>
+
+#include "helpers/MQTTPacketFilter.h"
+
+namespace Filter = MQTTPacketFilter;
+
+TEST(MQTTPacketFilter, EmptyAndAllMeanEveryType) {
+  for (const char* value : {"", " ", "\tall\r\n"}) {
+    uint16_t mask = 0;
+    ASSERT_TRUE(Filter::parse(value, &mask)) << value;
+    EXPECT_EQ(Filter::kAllPacketTypes, mask);
+  }
+}
+
+TEST(MQTTPacketFilter, NoneMeansNoTypes) {
+  uint16_t mask = Filter::kAllPacketTypes;
+  ASSERT_TRUE(Filter::parse(" none ", &mask));
+  EXPECT_EQ(0u, mask);
+}
+
+TEST(MQTTPacketFilter, ParsesDecimalCsvWithWhitespaceAndDuplicates) {
+  uint16_t mask = 0;
+  ASSERT_TRUE(Filter::parse(" 2, 4,\t15,2 ", &mask));
+  EXPECT_EQ(static_cast<uint16_t>((1u << 2) | (1u << 4) | (1u << 15)), mask);
+}
+
+TEST(MQTTPacketFilter, FullNumericListCanonicalizesToAll) {
+  uint16_t mask = 0;
+  ASSERT_TRUE(Filter::parse("0,1,2,3,4,5,6,7,8,9,10,11,12,13,14,15", &mask));
+  EXPECT_EQ(Filter::kAllPacketTypes, mask);
+  char output[Filter::kFilterTextSize];
+  ASSERT_TRUE(Filter::format(mask, output, sizeof(output)));
+  EXPECT_STREQ("all", output);
+}
+
+TEST(MQTTPacketFilter, RejectsMalformedOrOutOfRangeValuesWithoutChangingOutput) {
+  const char* invalid[] = {
+    ",", "1,", ",1", "1,,2", "1, ,2", "-1", "+1", "0x2", "16",
+    "999999999999999999999", "ALL", "All", "all,2", "none,2", "2-4", "2x",
+  };
+  for (const char* value : invalid) {
+    uint16_t mask = 0x1234;
+    EXPECT_FALSE(Filter::parse(value, &mask)) << value;
+    EXPECT_EQ(0x1234, mask) << value;
+  }
+  EXPECT_FALSE(Filter::parse(nullptr, nullptr));
+}
+
+TEST(MQTTPacketFilter, FormatsSubsetsInAscendingOrder) {
+  const uint16_t mask = static_cast<uint16_t>(
+      (1u << 15) | (1u << 2) | (1u << 10) | (1u << 0));
+  char output[Filter::kFilterTextSize];
+  ASSERT_TRUE(Filter::format(mask, output, sizeof(output)));
+  EXPECT_STREQ("0,2,10,15", output);
+}
+
+TEST(MQTTPacketFilter, FormatsNoneAndHonorsExactBufferBoundaries) {
+  char none[5];
+  ASSERT_TRUE(Filter::format(0, none, sizeof(none)));
+  EXPECT_STREQ("none", none);
+
+  char too_small[4] = {'x', 'x', 'x', '\0'};
+  EXPECT_FALSE(Filter::format(0, too_small, sizeof(too_small)));
+  EXPECT_STREQ("", too_small);
+
+  char all_types[Filter::kFilterTextSize];
+  const uint16_t subset = static_cast<uint16_t>(Filter::kAllPacketTypes & ~(1u << 14));
+  ASSERT_TRUE(Filter::format(subset, all_types, sizeof(all_types)));
+  EXPECT_STREQ("0,1,2,3,4,5,6,7,8,9,10,11,12,13,15", all_types);
+  char one_short[34];
+  EXPECT_FALSE(Filter::format(subset, one_short, sizeof(one_short)));
+  EXPECT_STREQ("", one_short);
+}
+
+TEST(MQTTPacketFilter, MembershipIsBoundedToFourBitTypes) {
+  EXPECT_TRUE(Filter::allows(static_cast<uint16_t>(1u << 0), 0));
+  EXPECT_TRUE(Filter::allows(static_cast<uint16_t>(1u << 15), 15));
+  EXPECT_FALSE(Filter::allows(Filter::kAllPacketTypes, 16));
+  EXPECT_FALSE(Filter::allows(0, 0));
+}
+
+TEST(MQTTPacketFilter, EligibilityIgnoresConnectionButRequiresTopicAndFilter) {
+  const uint16_t only_advert = static_cast<uint16_t>(1u << 4);
+  EXPECT_TRUE(Filter::slotEligible(true, true, only_advert, 4));
+  EXPECT_FALSE(Filter::slotEligible(false, true, only_advert, 4));
+  EXPECT_FALSE(Filter::slotEligible(true, false, only_advert, 4));
+  EXPECT_FALSE(Filter::slotEligible(true, true, only_advert, 2));
+}
+
+TEST(MQTTPacketFilter, AllFilteredOrTopicIncompatibleTargetsCompleteIntentionally) {
+  const uint16_t only_text = static_cast<uint16_t>(1u << 2);
+  const bool filtered = Filter::slotEligible(true, true, only_text, 4);
+  const bool topic_incompatible =
+      Filter::slotEligible(true, false, Filter::kAllPacketTypes, 4);
+  EXPECT_FALSE(filtered || topic_incompatible);
+  EXPECT_TRUE(Filter::publishComplete(filtered || topic_incompatible, false));
+}
+
+TEST(MQTTPacketFilter, EligibleDisconnectedTargetStillRequiresRetry) {
+  const uint16_t only_advert = static_cast<uint16_t>(1u << 4);
+  // Connection state is intentionally absent from slotEligible(): the first
+  // slot remains a target while disconnected. A second connected slot whose
+  // filter rejects the advert cannot turn that into intentional completion.
+  const bool disconnected_eligible =
+      Filter::slotEligible(true, true, only_advert, 4);
+  const bool connected_filtered =
+      Filter::slotEligible(true, true, static_cast<uint16_t>(1u << 2), 4);
+  ASSERT_TRUE(disconnected_eligible);
+  ASSERT_FALSE(connected_filtered);
+  EXPECT_FALSE(Filter::publishComplete(
+      disconnected_eligible || connected_filtered, false));
+}
+
+TEST(MQTTPacketFilter, PublishFailureRetriesAndAnySuccessCompletes) {
+  EXPECT_FALSE(Filter::publishComplete(true, false));
+  EXPECT_TRUE(Filter::publishComplete(true, true));
+  EXPECT_TRUE(Filter::publishComplete(false, true));  // defensive: success wins
+}
+
+TEST(MQTTPacketFilter, UnsupportedRawPathCannotHideEligiblePacketFailure) {
+  const bool packet_eligible = true;
+  const bool packet_published = false;
+  const bool raw_eligible = false;  // e.g. MeshRank has no raw topic
+  const bool raw_published = false;
+
+  EXPECT_FALSE(Filter::publishComplete(
+      packet_eligible || raw_eligible,
+      packet_published || raw_published));
+  EXPECT_TRUE(Filter::publishComplete(false, false));
+}
+
+TEST(MQTTPacketFilter, PacketAndRawCanShareTheSameMaskDecision) {
+  const uint16_t text_and_advert = static_cast<uint16_t>((1u << 2) | (1u << 4));
+  for (uint8_t type = 0; type <= 15; ++type) {
+    const bool expected = type == 2 || type == 4;
+    EXPECT_EQ(expected, Filter::allows(text_and_advert, type)) << unsigned(type);
+  }
+}
+
+int main(int argc, char** argv) {
+  ::testing::InitGoogleTest(&argc, argv);
+  return RUN_ALL_TESTS();
+}

test/test_mqtt_prefs_codec/test_mqtt_prefs_codec.cpp

@@ -5,6 +5,7 @@
 
 #define WITH_MQTT_BRIDGE 1
 #include "helpers/MQTTPrefsCodec.h"
+#include "helpers/MQTTPacketFilter.h"
 
 namespace Codec = MQTTPrefsCodec;
 
@@ -20,6 +21,7 @@ MQTTPrefs defaults() {
   prefs.wifi_power_save = 1;
   for (int i = 0; i < MQTT_PREFS_SLOT_COUNT; ++i) {
     strncpy(prefs.mqtt_slot_preset[i], "none", sizeof(prefs.mqtt_slot_preset[i]) - 1);
+    prefs.mqtt_slot_packet_filter[i] = MQTTPacketFilter::kAllPacketTypes;
   }
   strncpy(prefs.snmp_community, "public", sizeof(prefs.snmp_community) - 1);
   prefs.radio_watchdog_minutes = 5;
@@ -218,6 +220,10 @@ TEST(MQTTPrefsCodec, MigratesAllLegacySixSlotPrefixesWithoutClobberingDefaults)
     } else {
       EXPECT_STREQ("time.example", prefs.mqtt_ntp_server);
     }
+    for (int slot = 0; slot < MQTT_PREFS_SLOT_COUNT; ++slot) {
+      EXPECT_EQ(MQTTPacketFilter::kAllPacketTypes,
+                prefs.mqtt_slot_packet_filter[slot]) << size << ":" << slot;
+    }
   }
 }
 
@@ -228,6 +234,9 @@ TEST(MQTTPrefsCodec, CurrentVersionedPayloadRoundTripsExactly) {
   strncpy(source.alert_region, "PNW", sizeof(source.alert_region) - 1);
   source.mqtt_neighbors_enabled = 1;
   source.mqtt_neighbors_interval = MQTT_NEIGHBORS_MAX_INTERVAL_MS;
+  for (int i = 0; i < MQTT_PREFS_SLOT_COUNT; ++i) {
+    source.mqtt_slot_packet_filter[i] = static_cast<uint16_t>(1u << i);
+  }
   std::vector<uint8_t> bytes(Codec::kEncodedSize);
   ASSERT_EQ(Codec::kEncodedSize, Codec::encode(source, bytes.data(), bytes.size()));
 
@@ -240,6 +249,37 @@ TEST(MQTTPrefsCodec, CurrentVersionedPayloadRoundTripsExactly) {
   EXPECT_EQ(0, memcmp(&source, &loaded, sizeof(source)));
 }
 
+TEST(MQTTPrefsCodec, PreFilterV1PayloadDefaultsEverySlotToAllTypes) {
+  MQTTPrefs source = defaults();
+  strncpy(source.mqtt_origin, "pre-filter-node", sizeof(source.mqtt_origin) - 1);
+  source.mqtt_neighbors_enabled = 1;
+  source.mqtt_neighbors_interval = MQTT_NEIGHBORS_MAX_INTERVAL_MS;
+  for (int i = 0; i < MQTT_PREFS_SLOT_COUNT; ++i) {
+    source.mqtt_slot_packet_filter[i] = 0;
+  }
+
+  std::vector<uint8_t> bytes(sizeof(MQTTPrefsHeader) + Codec::kV1PreFilterPayloadSize, 0);
+  writeHeader(&bytes, MQTT_PREFS_VERSION,
+              static_cast<uint16_t>(Codec::kV1PreFilterPayloadSize));
+  memcpy(bytes.data() + sizeof(MQTTPrefsHeader), &source, Codec::kV1PreFilterPayloadSize);
+
+  const Codec::DecodePlan plan = classify(bytes);
+  ASSERT_EQ(Codec::Source::Current, plan.source);
+  ASSERT_EQ(Codec::kV1PreFilterPayloadSize, plan.payload_len);
+  ASSERT_TRUE(plan.observer_fields_present);
+  ASSERT_FALSE(plan.preserve_file);
+
+  MQTTPrefs loaded = defaults();
+  memcpy(&loaded, bytes.data() + sizeof(MQTTPrefsHeader), plan.payload_len);
+  EXPECT_STREQ("pre-filter-node", loaded.mqtt_origin);
+  EXPECT_EQ(1u, loaded.mqtt_neighbors_enabled);
+  EXPECT_EQ(MQTT_NEIGHBORS_MAX_INTERVAL_MS, loaded.mqtt_neighbors_interval);
+  for (int i = 0; i < MQTT_PREFS_SLOT_COUNT; ++i) {
+    EXPECT_EQ(MQTTPacketFilter::kAllPacketTypes,
+              loaded.mqtt_slot_packet_filter[i]) << i;
+  }
+}
+
 TEST(MQTTPrefsCodec, CompatibleShortV1PayloadPreservesDefaultsBeyondObserverBoundary) {
   MQTTPrefs source = defaults();
   strncpy(source.mqtt_origin, "short-v1-node", sizeof(source.mqtt_origin) - 1);
@@ -304,6 +344,10 @@ TEST(MQTTPrefsCodec, PreNeighborsV1PayloadLoadsObserverFieldsAndDefaultsNeighbor
   // Interval begins at 2860 (beyond the read) -> keeps the caller's default.
   EXPECT_EQ(0u, loaded.mqtt_neighbors_enabled);
   EXPECT_EQ(MQTT_NEIGHBORS_DEFAULT_INTERVAL_MS, loaded.mqtt_neighbors_interval);
+  for (int i = 0; i < MQTT_PREFS_SLOT_COUNT; ++i) {
+    EXPECT_EQ(MQTTPacketFilter::kAllPacketTypes,
+              loaded.mqtt_slot_packet_filter[i]) << i;
+  }
 }
 
 TEST(MQTTPrefsCodec, CorruptOrShortVersionedInputsArePreserved) {

test/test_webconfig_keys/test_webconfig_keys.cpp

@@ -22,6 +22,7 @@ TEST(WebConfigKeys, AllowsPerSlotKeys) {
   EXPECT_TRUE(wcIsAllowedSetKey("mqtt1.preset"));
   EXPECT_TRUE(wcIsAllowedSetKey("mqtt1.server"));
   EXPECT_TRUE(wcIsAllowedSetKey("mqtt1.token"));
+  EXPECT_TRUE(wcIsAllowedSetKey("mqtt2.filter"));
   EXPECT_TRUE(wcIsAllowedSetKey("mqtt6.audience"));  // MAX_MQTT_SLOTS == 6
 }
 
@@ -102,6 +103,7 @@ TEST(WebConfigKeys, NonSecretKeysNotFlagged) {
   EXPECT_FALSE(wcIsSecretKey("wifi.ssid"));
   EXPECT_FALSE(wcIsSecretKey("mqtt1.username"));  // username is not masked
   EXPECT_FALSE(wcIsSecretKey("mqtt1.server"));
+  EXPECT_FALSE(wcIsSecretKey("mqtt1.filter"));
   EXPECT_FALSE(wcIsSecretKey("mqtt.origin"));
   EXPECT_FALSE(wcIsSecretKey("name"));
   EXPECT_FALSE(wcIsSecretKey(""));