From e3740278e5cb5070b56091c79d989895d39356d7 Mon Sep 17 00:00:00 2001 From: Strycher Date: Mon, 22 Jun 2026 22:36:08 -0400 Subject: [PATCH] feat(#80): broker editor service write ops (save handshake, clear, re-GET) The write half of the broker editor, against RedCreek's confirmed field-at-a-time contract. The existing 0xC0 key/value command already carries the broker keys -- no new wire codes -- so the codec comment at observer_config_client.dart:166 is stale, not a gap. - saveBroker(): disable a live slot first, write each changed field, write `enabled` LAST. Stops at the first ERR/timeout and reports the failed field -- a partial save leaves the slot disabled, never live-corrupt (activation guard). TDD: ordering + partial-failure tests. - clearBroker(): mqtt.broker..clear. - getBroker(): single-slot re-GET via 14 per-field GETs (recovery / post-save read), not the 84-frame pool dump. Password is write-only -> best-effort. Key spellings (mqtt.broker.N.enabled / .clear, scalar field GET) are inferred from the existing scheme; they're the one thing the joint firmware test confirms. 12/12 service tests green. UI (brokers screen + editor) next. Co-Authored-By: Claude Opus 4.8 --- lib/services/observer_config_service.dart | 76 +++++++++++++ .../observer_config_service_test.dart | 101 ++++++++++++++++++ 2 files changed, 177 insertions(+) diff --git a/lib/services/observer_config_service.dart b/lib/services/observer_config_service.dart index 928c23f..860c031 100644 --- a/lib/services/observer_config_service.dart +++ b/lib/services/observer_config_service.dart @@ -132,6 +132,73 @@ class ObserverConfigService extends ChangeNotifier { Future setBrokerField(int slot, String field, String value) => setFlat('mqtt.broker.$slot.$field', value); + /// Save a broker [slot] field-at-a-time per the firmware contract: a live slot + /// is disabled first, each changed [fields] entry is written, then `enabled` + /// is written LAST (when [enable]). Stops at the first ERR/timeout and reports + /// the failed field — a partial save therefore leaves the slot disabled, never + /// live-corrupt (#80). The caller re-GETs the slot to show true state + offer + /// retry/clear. + Future saveBroker( + int slot, { + required Map fields, + required bool enable, + required bool wasLive, + }) async { + if (wasLive) { + if (!await setBrokerField(slot, 'enabled', '0')) { + return const BrokerSaveResult.failed('enabled'); + } + } + for (final e in fields.entries) { + if (!await setBrokerField(slot, e.key, e.value)) { + return BrokerSaveResult.failed(e.key); + } + } + if (enable) { + if (!await setBrokerField(slot, 'enabled', '1')) { + return const BrokerSaveResult.failed('enabled'); + } + } + return const BrokerSaveResult.ok(); + } + + /// Definitively wipe a broker slot — `mqtt.broker..clear` (#80). + Future clearBroker(int slot) => setBrokerField(slot, 'clear', '1'); + + /// Re-read a single broker [slot] field-by-field — the post-save / recovery + /// re-GET. Cheaper than the whole pool: 14 scalar GETs for one slot, not the + /// 84-frame OCFG_BROKERS dump (#80). Returns null if any field GET fails. + Future getBroker(int slot) async { + const fields = [ + 'enabled', + 'url', + 'port', + 'transport', + 'auth_type', + 'username', + 'password', + 'topic_prefix', + 'iata_override', + 'jwt_audience', + 'jwt_refresh', + 'jwt_owner', + 'jwt_email', + 'ca_cert', + ]; + final kv = {}; + for (final f in fields) { + final v = await getFlat('mqtt.broker.$slot.$f'); + if (v == null) { + // password is write-only — a missing read is its presence, not a slot + // failure; any other missing field means the re-GET is incomplete. + if (f == 'password') continue; + return null; + } + kv[f] = v; + } + return BrokerConfig.fromWireFields(slot, kv); + } + /// Read the broker pool (paginated START → KV → END). Null on timeout. Future?> getBrokers() { return _serialized(() async { @@ -225,3 +292,12 @@ class ObserverConfigService extends ChangeNotifier { ); } } + +/// Outcome of [ObserverConfigService.saveBroker]. On failure [failedField] names +/// the field whose SET failed (or `enabled`); the slot is left disabled, safe. +class BrokerSaveResult { + const BrokerSaveResult.ok() : ok = true, failedField = null; + const BrokerSaveResult.failed(this.failedField) : ok = false; + final bool ok; + final String? failedField; +} diff --git a/test/services/observer_config_service_test.dart b/test/services/observer_config_service_test.dart index f299498..a975da9 100644 --- a/test/services/observer_config_service_test.dart +++ b/test/services/observer_config_service_test.dart @@ -80,6 +80,14 @@ class _AutoConnector extends MeshCoreConnector { ? _resp(ObserverConfigClient.rErr, 'ERROR not available') : _resp(ObserverConfigClient.rValue, '$key = ${_valueFor(key)}'); Future.microtask(() => _frames.add(frame)); + } else if (op == ObserverConfigClient.opSet) { + // SET payload is "key value"; the key is everything before the 1st space. + final payload = utf8.decode(data.sublist(2, data.indexOf(0, 2))); + final key = payload.split(' ').first; + final frame = failKeys.contains(key) + ? _resp(ObserverConfigClient.rErr, 'ERROR not available') + : _resp(ObserverConfigClient.rAck, '$key = ok'); + Future.microtask(() => _frames.add(frame)); } else if (op == ObserverConfigClient.opBrokers) { if (failBrokers) return; // no response -> getBrokers times out Future.microtask(() { @@ -272,4 +280,97 @@ void main() { auto.closeStream(); }, ); + + // ---- #80: broker save handshake (enabled-last, per-field ACK, recovery) ---- + List setKeys(_AutoConnector c) => c.sent + .where((f) => f.length > 2 && f[1] == ObserverConfigClient.opSet) + .map((f) => utf8.decode(f.sublist(2, f.indexOf(0, 2))).split(' ').first) + .toList(); + + test( + 'saveBroker on a live slot disables first, writes fields, enables LAST', + () async { + final auto = _AutoConnector(); + final s = ObserverConfigService(auto); + final r = await s.saveBroker( + 2, + fields: {'url': 'mqtt://h', 'port': '1883'}, + enable: true, + wasLive: true, + ); + expect(r.ok, isTrue); + final keys = setKeys(auto); + expect( + keys.first, + 'mqtt.broker.2.enabled', + reason: 'a live slot is disabled first', + ); + expect( + keys.last, + 'mqtt.broker.2.enabled', + reason: 'enabled is written LAST (activation guard)', + ); + expect(keys.where((k) => k == 'mqtt.broker.2.enabled').length, 2); + expect( + keys.indexOf('mqtt.broker.2.url'), + greaterThan(keys.indexOf('mqtt.broker.2.enabled')), + ); + expect( + keys.indexOf('mqtt.broker.2.port'), + lessThan(keys.lastIndexOf('mqtt.broker.2.enabled')), + ); + auto.closeStream(); + }, + ); + + test('saveBroker stops on a field ERR and never reaches the enabled-last ' + 'activation', () async { + final auto = _AutoConnector(failKeys: {'mqtt.broker.2.port'}); + final s = ObserverConfigService(auto); + final r = await s.saveBroker( + 2, + fields: {'url': 'mqtt://h', 'port': '1883'}, + enable: true, + wasLive: false, + ); + expect(r.ok, isFalse); + expect(r.failedField, 'port'); + expect( + setKeys(auto), + isNot(contains('mqtt.broker.2.enabled')), + reason: 'a failed field must never reach enabled=1 — slot stays safe', + ); + auto.closeStream(); + }); + + test('clearBroker sends the slot clear op', () async { + final auto = _AutoConnector(); + final s = ObserverConfigService(auto); + expect(await s.clearBroker(3), isTrue); + expect(setKeys(auto), contains('mqtt.broker.3.clear')); + auto.closeStream(); + }); + + test( + 'getBroker re-reads one slot field-by-field, never the pool dump', + () async { + final auto = _AutoConnector(); + final s = ObserverConfigService(auto); + final b = await s.getBroker(4); + expect(b, isNotNull); + expect(b!.slot, 4); + final getKeys = auto.sent + .where((f) => f.length > 2 && f[1] == ObserverConfigClient.opGet) + .map((f) => utf8.decode(f.sublist(2, f.indexOf(0, 2)))) + .toList(); + expect(getKeys, contains('mqtt.broker.4.url')); + expect(getKeys, contains('mqtt.broker.4.enabled')); + expect( + auto.sent.any((f) => f[1] == ObserverConfigClient.opBrokers), + isFalse, + reason: 'single-slot re-GET must not trigger the 84-frame pool dump', + ); + auto.closeStream(); + }, + ); }