MeshCore-mqtt-observer/MQTT_INTERNALS.md

# MQTT Bridge Internals

Developer-facing notes on how the MQTT observer feature is structured in the codebase: source files, the seams that keep it isolated from upstream MeshCore code, and how on-device settings are migrated across firmware versions. For user-facing setup and CLI reference, see [MQTT_IMPLEMENTATION.md](MQTT_IMPLEMENTATION.md).

## Files

### Core Implementation
- `src/helpers/bridges/MQTTBridge.h` - MQTT bridge class definition
- `src/helpers/bridges/MQTTBridge.cpp` - MQTT bridge implementation
- `src/helpers/MQTTPresets.h` - Preset definitions, CA certificates, and lookup functions
- `src/helpers/MQTTDefaults.h` - Compile-time defaults for fresh `/mqtt.json`
- `src/helpers/MQTTPrefsSerializer.h` - Versioned, semantic observer JSON schema
- `src/helpers/MQTTMessageBuilder.h` - JSON message formatting utilities
- `src/helpers/MQTTMessageBuilder.cpp` - JSON message formatting implementation
- `src/helpers/JWTHelper.h` - JWT token generation for Ed25519-based authentication
- `src/helpers/CommonCLI_Observer.cpp` - All observer CLI command handling (MQTT, WiFi,
  timezone, NTP, OTA, SNMP, alerts)

### Integration seams with upstream code

The observer feature is kept out of upstream-tracked files through three mechanisms:

- **CLI hook methods** — upstream `CommonCLI.cpp` delegates to three `CommonCLI`
  methods defined in the fork-owned `CommonCLI_Observer.cpp`: `handleObserverCommand()`,
  `handleObserverSetCmd()`, and `handleObserverGetCmd()`. Each returns `true` if it
  consumed the command, otherwise the upstream parser runs. Only these three call
  sites touch upstream CLI code.
- **Callback virtuals** — observer behaviour needed from the application is exposed
  as default-no-op virtuals on `CommonCLICallbacks` (e.g. `restartBridgeSlot`,
  `isMqttBridgeRunning`, `syncMqttNtp`, `onAlertConfigChanged`, `sendAlertText`,
  `resolveAlertScope`, `beginDeferredOtaUpdate`). The example apps override them
  behind `#ifdef WITH_MQTT_BRIDGE`.
- **Separate settings file** — observer settings (MQTT slots, WiFi, timezone, SNMP,
  radio watchdog, fault alerts) live in the runtime `MQTTPrefs` object and are
  field-serialized to `/mqtt.json`, keeping `NodePrefs` / `/prefs.json` aligned with
  upstream. The two files are independent transactions, not one atomic snapshot.

Remaining integration points in upstream files:
- `examples/simple_repeater/MyMesh.{h,cpp}`, `examples/simple_room_server/MyMesh.{h,cpp}` -
  bridge/alerter/SNMP wiring and packet-feed hooks, guarded by `#ifdef WITH_MQTT_BRIDGE`;
  plus the `createObserverPacketManager()` call in each constructor (see below)
- `src/helpers/CommonCLI.{h,cpp}` - the three CLI hooks, `MQTTPrefs` load/save/migration
- `src/Dispatcher.{h,cpp}` - radio watchdog block, guarded by `#ifdef WITH_MQTT_BRIDGE`

### Capture vs. duty-cycle throttling

RX processing needs a free packet from the static pool before `logRx()` (and thus the
MQTT uplink) can run — `Dispatcher::checkRecv()` silently discards received data when
the pool is empty. Because the outbound queue holds pool packets with no expiry,
duty-cycle throttling can park the entire pool waiting on TX budget, capping capture at
the TX rate — and the parked repeats absorb every budget refill, starving the node's
own CLI responses and making it un-administrable over the mesh. Observer builds
therefore use `RxReservePacketManager` (fork-owned,
`src/helpers/RxReservePacketManager.h`): below the RX reserve (a quarter of the pool)
it sheds only low-priority outbound (multi-hop flood repeats, adverts, trace), keeping
the node's own responses/ACKs queueable; below a smaller emergency floor it sheds
everything to keep capture alive. Queued packets still untransmitted 30 s past their
scheduled time are expired at dequeue, so under throttle the queue holds only fresh
traffic and admin responses reach the trickle of TX budget. Non-observer builds keep
the upstream pool behavior.

### Neighbors publication path (PSRAM only)

Periodic neighbors publishing is gated on `WITH_MQTT_NEIGHBORS`
(`defined(BOARD_HAS_PSRAM) && defined(MAX_NEIGHBOURS) && MAX_NEIGHBOURS > 0`,
defined in `MQTTBridge.h`). It spans two subsystems and two cores:

- **Mesh side (Core 1), `MyMesh`**: the `loop()` runs a two-stage refresh driven by
  `mqtt_neighbors_interval`. Stage 1 sends a zero-hop `sendNodeDiscoverReq()` and waits
  out its 60 s collection window to refresh `neighbours[]`. Stage 2
  (`startNeighborDiscover`) fires one anon-regions scope query per heard neighbor,
  overlaying them onto the peer-index space at `NEIGHBOR_DISCOVER_PEER_BASE` so their
  `PAYLOAD_TYPE_RESPONSE` packets decrypt via `searchPeersByHash`/`getPeerSharedSecret`/
  `onPeerDataRecv` even when the neighbor is not an ACL client. A reply is zero-hop by
  request, so `handleNeighborDiscoverResponse` also re-stamps `heard_timestamp` in both
  the snapshot and `neighbours[]` — proof of reception, and the only thing that heals a
  stamp taken before the clock was set. After all responses land or a 30 s window
  expires, `finishNeighborDiscover()` builds the JSON with
  `MQTTMessageBuilder::buildNeighborsMessage` into a transient buffer (PSRAM where
  available, internal DRAM otherwise) and hands it to the bridge. The entry table and its
  hex strings share one heap block sized to the pass — at `MAX_NEIGHBOURS` they reach
  ~4.5 KB, which does not fit the mesh loop task's 8 KB stack. Ages that still span a clock epoch publish as `null` rather than a
  fabricated delta (`neighborHeardAgeUsable`, see `UPSTREAM_BUGS.md` #1).
- **Buffer sizing**: `NEIGHBORS_JSON_BUFFER_SIZE` is 10 KB with PSRAM and 4 KB without,
  since a non-PSRAM board pays for the persistent buffer, the transient build buffer and
  the ArduinoJson pool out of the same internal DRAM each TLS slot needs ~40 KB of
  (~13 KB peak instead of ~35 KB). The pool has its own budget
  (`NEIGHBORS_DOC_POOL_BUDGET`) because ArduinoJson v7 hands out pool blocks in fixed
  4096-byte chunks, so a table that just fits the text buffer can still need well over
  it in pool — and a starved pool sets `doc.overflowed()`, which drops the entire publish
  instead of truncating it. `NEIGHBORS_MAX_PUBLISH_ENTRIES` (20 without PSRAM) keeps the
  pool inside a single block.
- **Bridge side, handoff**: `requestPublishNeighbors(json, len)` (Core 1) memcpys into the
  persistent buffer (`NEIGHBORS_JSON_BUFFER_SIZE`, PSRAM where available) and sets
  `_neighbors_publish_pending` with a release store; the MQTT task (`mqttTaskLoop`, Core 0)
  consumes it with an acquire load, calls `publishNeighbors()`, and clears the flag. A
  second snapshot is dropped while one is in flight. `publishNeighbors()` sends QoS 1,
  retain = `preset->allow_retain` (custom slots non-retained). MeshRank slots are included,
  publishing to `meshrank/uplink/{token}/{device}/neighbors` (non-retained, since the
  preset sets `allow_retain = false`).
- **Status reporting**: `MyMesh` reports the schedule each loop via
  `setNeighborsSchedule(phase, secs)`; `formatMqttStatusReply` renders it as the trailing
  `nbr: <when>/<last>` field in `get mqtt.status` while the feature is enabled.

The JSON builder lives in the pure, host-tested `MQTTPayloadBuilder`
(`test/test_mqtt_payload_builder`); the topic type in `MQTTTopicRouter`
(`test/test_mqtt_topic_router`). The mesh↔bridge orchestration above is on-target only.

### Runtime construction and slot memory

- **Deferred construction** — `MQTTBridge` is heap-allocated in each app's `begin()`
  (`bridge = new MQTTBridge(...)` in `MyMesh.cpp`) rather than held as a static member,
  because constructing it at static-init time crashes on ESP32 classic.
- **Runtime slot array** — `RUNTIME_MQTT_SLOTS` (`MQTTPresets.h`) is 6 with PSRAM and 3
  without, saving ~1.2 KB of heap on non-PSRAM boards. `MAX_MQTT_SLOTS` stays 6 on every
  build because it fixes the persisted `MQTTPrefs` layout, so slot config survives moving
  firmware between board classes. Three runtime slots suffice without PSRAM:
  `_max_active_slots` caps those boards at 2 live connections, leaving one spare for
  reconfiguration. Configured slots past the cap report `(inactive)`.
- **Buffers** — the 768-byte JWT `auth_token` is inline in every `MQTTSlot`, not allocated
  per JWT-auth slot. What varies is the MQTT client's TX/RX buffer: 896 bytes (the minimum
  that fits a CONNECT plus a 768-byte JWT) uniformly on PSRAM boards to limit
  fragmentation from mixed allocations, and 896 or 512 per slot on non-PSRAM boards so
  non-JWT slots leave smaller holes across teardown/recreate cycles. The large
  JSON/raw-packet buffers go through `psram_malloc()`, which prefers PSRAM and falls back
  to internal heap.

### Reconnection, backoff, and circuit breaker

The client's own auto-reconnect is disabled (`setAutoReconnect(false)`); the bridge drives
reconnection per slot.

- Backoff ladder: 10 s → 30 s → 60 s → 120 s → 300 s, staggered by 3 s × slot index so
  slots don't all handshake at once.
- The ladder resets only after a connection has held for 2 minutes
  (`BACKOFF_STABLE_RESET_MS`), which is longer than the 75 s keepalive — a link that can't
  survive one keepalive round-trip keeps its earned rung instead of hammering TLS
  handshakes at the 10 s rung. CONNACK alone does not reset it.
- After 3 more failures at the top rung (~15 min) the slot's circuit breaker trips and
  routine reconnects stop. A tripped slot is probed once every 30 minutes (with a fresh
  JWT where applicable); a successful connect clears the breaker, as does reconfiguring
  the slot.
- Message retransmit timeout is 15 s — one retry inside esp-mqtt's 30 s outbox expiry,
  preserving at-least-once delivery while capping duplicates at one.

### Message building

- The `hash` field in `packets` messages is MeshCore's own packet hash,
  `Packet::calculatePacketHash()` — SHA256 over the payload type and payload (plus
  `path_len` for TRACE), truncated to `MAX_HASH_SIZE`. It is the same value the dispatcher
  uses, so uplinked hashes match the mesh.
- `score` is recomputed at publish time from the packet's SNR and length via the radio's
  `packetScore()`, so it matches the value the firmware used on receive.
- Timezone: the JChristensen/Timezone object (`_timezone_storage`, inline since the
  memory-defrag work) is kept current from `timezone_string` via `setRules()`, but
  `formatIsoTimestampForMqtt()` explicitly ignores it — every published timestamp, time,
  and date field is UTC off `gmtime()`, matching Python's
  `datetime.now(timezone.utc).isoformat()`. The timezone prefs therefore do not affect
  MQTT message content.

### Command namespacing

CLI commands sit at two levels. `bridge.*` is low-level and shared by all bridge types
(MQTT, RS232, ESP-NOW): `bridge.enabled` is the master switch, and `bridge.source` selects
which packet events non-MQTT bridges capture. The MQTT bridge ignores `bridge.source` in
favour of independent `mqtt.rx` / `mqtt.tx` controls. Everything MQTT-specific lives under
`mqtt.*` (shared settings), `mqttN.*` (per-slot broker config), `wifi.*`, and `timezone.*`.

### `/mqtt.json` file format

Observer preferences use the same `ConfigSerializer` object notation as upstream
`/prefs.json`: semantic unquoted keys, quoted strings, decimal numbers, and nested
objects. Schema version 1 requires the root `version:1` field, which must be the
first property of the root object. The main shape is:

```text
{version:1,
 wifi:{ssid:"...",password:"...",power_save:1},
 time:{timezone:"...",utc_offset:0,ntp_server:"..."},
 mqtt:{origin:"...",iata:"SEA",packets_enabled:1,raw_enabled:0,
       tx_enabled:2,rx_enabled:1,
       status:{enabled:1,interval_ms:300000},
       neighbors:{enabled:0,interval_ms:86400000},
       owner:{public_key:"...",email:"..."},
       slot1:{preset:"analyzer-us",host:"",port:0,username:"",password:"",
              token:"",topic:"",audience:"",packet_filter:65535},
       ... slot2 through slot6 ...},
 snmp:{enabled:0,community:"public"},
 radio:{watchdog_min:5},
 alert:{enabled:0,psk_hex:"",wifi_minutes:30,mqtt_minutes:240,
        rate_limit_min:60,hashtag:"",region:""}}
```

Keys may contain digits after their first character, which permits the readable
`slot1` ... `slot6` names. Every schema key fits `ConfigSerializer`'s 15-character
visible-key limit. Known strings and numbers use strict parsing: duplicates, overlong
strings, malformed decimals, and overflow reject the complete file. A supported file
with a semantically out-of-range value is repaired to that field's safe default and
rewritten atomically. Unknown fields are ignored only within the serializer's general
limits (15-character keys, 127-byte decoded values, and six nested object levels below the root).

Loading always starts with defaults and parses into a separate heap scratch object.
The live preferences change only after the complete file parses, has an explicit
supported version, and passes validation. A missing/future version, syntax error,
overlength value, or allocation/read failure leaves `/mqtt.json` untouched, runs this
boot on defaults, and holds observer saves so a later CLI/WebConfig write cannot erase
the opaque source. If an observer setter cannot commit its JSON transaction, it restores
the pre-command in-memory preferences and does not restart or apply a bridge change as
though the setting were durable.

Saves stream to `/mqtt.json.tmp` through a sticky short-write detector while computing
size and checksum. The firmware closes and rereads the temp, verifies size/checksum,
parses it into another scratch object, then publishes with
`/mqtt.json` -> `/mqtt.json.bak` and temp -> primary. Failure safety is paid for in
transient heap: a CLI setter holds one `MQTTPrefs` rollback snapshot (2876 bytes) for the
whole command, and the save allocates one more for normalization defaults (released
before file I/O) and then one for verification — about 5.6 KiB peak above baseline.
Each allocation is checked, so exhaustion refuses the setting rather than crashing.

If publishing fails partway, the transaction is rolled back rather than left for boot
recovery: the backup is renamed back over the empty primary and the verified temp is
discarded. This is what makes the setter's "change rolled back" reply true — without it
the next boot would promote that temp and activate the value the CLI just refused. When
the rollback itself cannot complete, the save reports an indeterminate outcome instead,
the artifacts stay for recovery, and the CLI says the flash state is unresolved. Further
saves are refused until then, because a transaction cannot start while a temp or backup
exists. Boot recovery selects the usable
primary/temp/backup without overwriting an opaque future or corrupt primary. A valid
future-version temp that reached the rename phase wins over the stale backup and is held
for newer firmware. If a temp claims a future version but uses grammar this firmware
cannot parse, or cannot be classified because scratch allocation fails, recovery renames
nothing at all: it reads the last usable backup straight out of `/mqtt.json.bak`, leaves
the primary name empty, and holds all observer writes. The empty primary name is the only
record that the candidate had already passed the backup rename, so publishing the backup
would make the candidate indistinguishable from a stale artifact — and the next boot, the
one with enough heap or new enough firmware to finally read it, would delete it. Leaving
the names alone means that boot promotes the candidate through the ordinary temp rule, or
falls back to the backup if it turns out to be definitively corrupt. A
definitively corrupt/incomplete current-version temp may be discarded for that backup.
If power fails during the very first migration, there is no JSON primary or backup yet;
recovery discards a definitively invalid temp so the intact `/mqtt_prefs` source can be
loaded and the migration retried. An uncertain/future temp is still preserved and held.

Unknown slot preset names are repaired to `none`, never to a build-specific default
broker. Duplicate known presets from historical firmware are preserved on load; current
CLI and WebConfig setters prevent creating new duplicates without silently changing an
existing deployment during migration.

Schema changes that reinterpret existing names or values must increment `version`.
Additive fields may remain in version 1 only when old readers can safely ignore them
within the limits above. A future version is treated as opaque rather than partially
loading its known-looking fields. Literal schema keys are compile-time checked against
the 15-character visible-key limit so a new version-1 field cannot accidentally violate
that downgrade contract.

Every future version must also keep `version` as the first root property. Older firmware
detects a future file by probing that field with its own grammar, and the probe stops at
the first construct it cannot tokenize (an array, an overlong key, a value type it does
not expect). Syntax a newer schema introduces *before* the version field therefore makes
its files look corrupt rather than future, which costs them the preservation guarantee
above: as the temp of an interrupted commit such a file is discardable instead of
retained. `test_mqtt_prefs_serializer` pins both the ordering and that consequence.

#### Downgrade and rollback

The old `/mqtt_prefs` binary is read only for one-time migration and is deliberately
not updated or deleted. When it exists and is usable, it is the exact pre-migration
rollback snapshot. Firmware older than this JSON change will therefore see stale
observer settings after a downgrade. A fresh JSON-only install has no binary observer
configuration to recover on downgrade.

The two files are intentionally not reconciled. If an operator changes observer settings
while running old firmware, those changes update only `/mqtt_prefs`. Rolling forward to
this firmware makes the existing `/mqtt.json` authoritative again, so rollback-era edits
are ignored unless the operator exports and reapplies them. Conversely, JSON-only settings
cannot be recovered by old firmware. This is a rollback snapshot, not bidirectional sync.

`LegacyV1MQTTPrefs` and the older pre-slot/3-slot/6-slot structs are frozen migration
ABIs with size/offset assertions. The runtime `MQTTPrefs` layout is not an on-flash ABI.

### Settings upgrade / migration

`loadPrefs()` handles every historical on-device format one-time at boot:
- **`/mqtt_prefs` -> `/mqtt.json`** — if the legacy file has the version header its
  frozen v1 layout is field-copied. Otherwise
  it is a legacy headerless file and its layout is detected by size: pre-slot
  (`OldMQTTPrefs`), 3-slot (`ThreeSlotMQTTPrefs`), or the 6-slot layout shipped on
  `observer-firmware` back when it was named `mqtt-bridge-implementation-flex`
  (`Legacy6SlotMQTTPrefs`). Each is field-copied into
  the runtime `MQTTPrefs` and saved as schema-versioned JSON — which also
  drops the vestigial `_legacy_*` fields the flex layout carried mid-struct. This is a
  one-time rewrite; every deployed device performs it on its first boot of versioned
  firmware, after which `/mqtt.json` is authoritative. The binary source remains as
  a rollback snapshot and is never dual-written.
  The pre-slot (`OldMQTTPrefs`) copy maps the old single-broker keys onto slots:
  `mqtt.analyzer.us = on` → slot 1 `analyzer-us`, `mqtt.analyzer.eu = on` → slot 2
  `analyzer-eu`, and a configured `mqtt.server` / `mqtt.port` / `mqtt.username` /
  `mqtt.password` → slot 3 `custom` with those values preserved. Origin, IATA, message
  types, WiFi, and timezone carry over as-is.
- **`/com_prefs`** — a file written by fork firmware that predates the `MQTTPrefs` split
  (a zero-filled MQTT gap plus a trailing observer block) is detected by size; the
  trailing SNMP / radio-watchdog / fault-alert settings and the `rx_boosted_gain` /
  `flood_max_*` fields are recovered, carried into `/mqtt.json`, and the active files are
  rewritten in the current formats.
- Settings the pre-split firmware stored *inside* the `/com_prefs` MQTT gap (the MQTT
  slot/WiFi config itself) are **not** recovered — users upgrading from firmware that
  old must re-enter their MQTT and WiFi configuration.

### Local testing without hardware

- **Portal UI:** `python3 scripts/webconfig_mock_server.py` (add `--setup` for the
  first-boot wizard), then open `http://localhost:8080/`. It serves `webui/index.html` and
  mimics the firmware's `/api/*` behavior. Standard library only.
- **Boot, WiFi, MQTT, CLI, OLED:** the Wokwi ESP32-S3 sim. Build
  `pio run -e Heltec_v3_repeater_observer_mqtt_sim -t mergebin` (radio stubbed by
  `SimRadio`, WiFi set to `Wokwi-GUEST`), then run it from `wokwi.toml`/`diagram.json`.
  Outbound MQTT works on the free gateway; reaching the on-device portal needs Wokwi's
  paid Private Gateway.
- **Host unit tests:** `pio test -e native`; see [test/README.md](test/README.md).