nodered: its MQTT broker comes from the mesh

Node-RED's one broker node pointed at zurag.be:1884, where nothing listens. nodered now requires mqtt-topic (asking for every topic: flows follow the devices' own) and a run-once `mqtt` step — declared last, restarted when the binding, credential or settings change — points the mesh's broker nodes at the bound broker through Node-RED's admin API with the module's api-token: the node the step makes itself when none is named, or the ones an assignment names in `mqtt.brokers`. Only host, port, TLS and the login change; the broker is asked first whether it takes the login; the deploy is against the revision read ("nodes", so only that node restarts) and a digest makes a rerun a no-op. A broker node nobody named is never touched. settings.js keeps `mqtt` and `topics` out of Node-RED.
2026-09-30 13:01:14 +02:00
parent c1a65e2354
commit 3c7aafdc21
9 changed files with 668 additions and 5 deletions
@@ -0,0 +1,124 @@
+// What holds nodered's MQTT step (mqtt/connection.ts): the broker nodes the mesh owns — the one it
+// makes, or the ones settings name — are made to use the broker and login the mesh bound, only after
+// the broker takes that login; every other field of a node is kept; nothing is deployed when nothing
+// differs; a node that is not named is never touched; a loopback broker address is refused.
+//
+// Node-RED and the broker are fakes answering as the real ones do (admin API v2 of nodered/node-red
+// 5.0.7, the build ace runs).
+
+import { test } from "node:test";
+import assert from "node:assert/strict";
+
+import { MESH_BROKER_ID, namedBrokers, reconcileBrokers, type Binding, type FlowNode, type Marks, type NodeRed } from "../mqtt/connection.ts";
+import type { Probe } from "../mqtt/probe.ts";
+
+const MINTED = "mesh-minted-password";
+const binding = (at = "ace.internal"): Binding => ({ provision: "mqtt-topic", from: "ace", at, as: "mesh_ace_nodered", serves: { scheme: "mqtt", port: 1883 } });
+
+/** ace's flows, reduced: its one broker node (dead, zurag.be:1884) and a node that uses it. */
+function aceFlows(): FlowNode[] {
+  return [
+    { id: "2b0aece9c5f3b307", type: "mqtt-broker", name: "MQTT Broker", broker: "zurag.be", port: "1884", clientid: "", usetls: false, protocolVersion: "4", keepalive: "60" },
+    { id: "fe0cae96f1e3ae4d", type: "mqtt in", topic: "stat/sonoff_office_light_switch/RESULT", broker: "2b0aece9c5f3b307", z: "t" },
+    { id: "other-broker", type: "mqtt-broker", name: "someone else's", broker: "test.mosquitto.org", port: "1883" },
+  ];
+}
+
+function fakeNodeRed(flows: FlowNode[], creds: Record<string, { user?: string; password?: string }> = {}) {
+  let rev = "r1";
+  const deploys: FlowNode[][] = [];
+  const nodered: NodeRed = {
+    async flows() {
+      return { rev, flows: structuredClone(flows) };
+    },
+    async credentials(_type, id) {
+      const c = creds[id] ?? {};
+      return { user: c.user, has_password: Boolean(c.password) };
+    },
+    async deploy(at, next) {
+      if (at !== rev) throw new Error("Node-RED /flows: 409 version_mismatch");
+      for (const n of next) {
+        if (n.credentials) creds[n.id] = { ...(creds[n.id] ?? {}), ...(n.credentials as object) };
+      }
+      flows.splice(0, flows.length, ...next.map(({ credentials: _c, ...n }) => n as FlowNode));
+      deploys.push(next);
+      rev = `r${deploys.length + 1}`;
+    },
+  };
+  return { nodered, deploys, flows, creds };
+}
+
+const marks = (): Marks & { store: Map<string, string> } => {
+  const store = new Map<string, string>();
+  return { store, get: async (k) => store.get(k), set: async (k, v) => void store.set(k, v) };
+};
+const takes: Probe = async (_h, _p, user, pass) => ({ connack: user === "mesh_ace_nodered" && pass === MINTED ? 0 : 5, suback: 0 });
+
+test("ace: the named broker node is moved to the bound broker and login; the other broker is not touched", async () => {
+  const f = fakeNodeRed(aceFlows(), { "2b0aece9c5f3b307": { user: "luffy", password: "old" }, "other-broker": { user: "x", password: "y" } });
+  const m = marks();
+  const out = await reconcileBrokers({ nodered: f.nodered, probe: takes, marks: m }, binding(), `${MINTED}\n`, ["2b0aece9c5f3b307"]);
+  assert.deepEqual(out, [{ what: "broker 2b0aece9c5f3b307", result: "written", fields: ["broker", "port", "user", "password"] }]);
+  const node = f.flows.find((n) => n.id === "2b0aece9c5f3b307");
+  assert.deepEqual(node, { ...aceFlows()[0], broker: "ace.internal", port: "1883", usetls: false });
+  assert.deepEqual(f.creds["2b0aece9c5f3b307"], { user: "mesh_ace_nodered", password: MINTED });
+  assert.deepEqual(f.flows.find((n) => n.id === "other-broker"), aceFlows()[2]);
+  assert.deepEqual(f.creds["other-broker"], { user: "x", password: "y" });
+  // Only the changed node carried credentials in the deploy.
+  assert.deepEqual(f.deploys[0].filter((n) => n.credentials).map((n) => n.id), ["2b0aece9c5f3b307"]);
+
+  // Again: nothing differs, nothing is deployed.
+  const again = await reconcileBrokers({ nodered: f.nodered, probe: takes, marks: m }, binding(), MINTED, ["2b0aece9c5f3b307"]);
+  assert.deepEqual(again, [{ what: "broker 2b0aece9c5f3b307", result: "unchanged" }]);
+  assert.equal(f.deploys.length, 1);
+});
+
+test("fresh: with nothing named, the step makes its own broker node", async () => {
+  const f = fakeNodeRed([]);
+  const out = await reconcileBrokers({ nodered: f.nodered, probe: takes, marks: marks() }, binding(), MINTED, []);
+  assert.deepEqual(out, [{ what: `broker ${MESH_BROKER_ID}`, result: "written", fields: ["node"] }]);
+  assert.equal(f.flows[0].type, "mqtt-broker");
+  assert.equal(f.flows[0].broker, "ace.internal");
+  assert.deepEqual(f.creds[MESH_BROKER_ID], { user: "mesh_ace_nodered", password: MINTED });
+});
+
+test("a login the broker does not take is never written", async () => {
+  const f = fakeNodeRed(aceFlows());
+  const out = await reconcileBrokers({ nodered: f.nodered, probe: async () => ({ connack: 5 }), marks: marks() }, binding(), MINTED, ["2b0aece9c5f3b307"]);
+  assert.equal(out[0].result, "refused");
+  assert.equal(f.deploys.length, 0);
+});
+
+test("a named node that is not there, or a loopback broker, is refused", async () => {
+  const f = fakeNodeRed(aceFlows());
+  const out = await reconcileBrokers({ nodered: f.nodered, probe: takes, marks: marks() }, binding(), MINTED, ["gone"]);
+  assert.match((out[0] as { problem: string }).problem, /no such node/);
+  const lo = await reconcileBrokers({ nodered: f.nodered, probe: takes, marks: marks() }, binding("127.0.0.1"), MINTED, []);
+  assert.match((lo[0] as { problem: string }).problem, /Node-RED itself/);
+  assert.equal(f.deploys.length, 0);
+});
+
+test("a deploy that raced another is read again once", async () => {
+  const f = fakeNodeRed(aceFlows());
+  let first = true;
+  const racing: NodeRed = {
+    ...f.nodered,
+    async flows() {
+      const got = await f.nodered.flows();
+      if (first) {
+        first = false;
+        return { ...got, rev: "stale" };
+      }
+      return got;
+    },
+  };
+  const out = await reconcileBrokers({ nodered: racing, probe: takes, marks: marks() }, binding(), MINTED, ["2b0aece9c5f3b307"]);
+  assert.equal(out[0].result, "written");
+  assert.equal(f.deploys.length, 1);
+});
+
+test("settings name broker nodes under mqtt.brokers", () => {
+  assert.deepEqual(namedBrokers({ mqtt: { brokers: ["a", "", 3, "b"] } }), ["a", "b"]);
+  assert.deepEqual(namedBrokers({ endpoints: {} }), []);
+  assert.deepEqual(namedBrokers(undefined), []);
+});