diff --git a/docs/websocket-api-schema-changelog.md b/docs/websocket-api-schema-changelog.md index d67a5450..0603a768 100644 --- a/docs/websocket-api-schema-changelog.md +++ b/docs/websocket-api-schema-changelog.md @@ -10,6 +10,24 @@ The server-side constants live in `packages/ws-controller/src/server/WebSocketCo (`SCHEMA_VERSION`, `MIN_SUPPORTED_SCHEMA_VERSION`). Versions predating this document are not listed retroactively; entries start at the first version maintained here. +## Schema 13 + +Minimum supported: 11 (older clients keep working with the pre-13 command shapes). + +### Network topology + +- **New command `get_network_topology`** → a `NetworkTopology` (`{ collected_at, nodes[], connections[] }`) describing the whole Matter network as a graph: the Thread mesh (nodes + neighbor/route-table links, with external neighbors and mDNS-discovered Border Routers) and the Wi-Fi star (one `wifi_ap` pseudo-node per BSSID — id `ap_` with colons stripped, e.g. `ap_112233445566` — stations linked to it). Ethernet nodes appear unlinked. Optional `refresh` argument re-reads the Thread neighbor/route tables, routing role and network name (and Wi-Fi diagnostics) from every online node before building the snapshot (slower — seconds; best-effort, concurrency-capped, under an overall deadline, and shared between concurrent requesters); omitted/`false` builds from the current attribute cache. + - Node kinds: `matter` (commissioned here; carries `node_id`), `border_router`, `thread_unknown` (a neighbour not commissioned on this fabric), `wifi_ap`. Thread connections keep both observed directions (`source_to_target` / `target_to_source`) so asymmetric links stay legible; the top-level `strength` is the strongest observed direction. `strength` values are `strong` / `medium` / `weak` / `none` / `unknown` — `none` means the link was observed dead (Thread pairs with no live direction are omitted entirely), `unknown` means no measurement was available (e.g. a Wi-Fi station whose RSSI can't be read) and must not be rendered as a dead link. Border Router classification depends on mDNS discovery: when the server runs with Thread diagnostics disabled, no `border_router` nodes appear and every external neighbour is reported as `thread_unknown`. +- **New event `network_topology_updated`** → a `NetworkTopology`, emitted (debounced, latest-wins coalesced) whenever the derived graph changes, plus a slow periodic refresh so sleepy-device drift is eventually reflected. **Delivered only to connections that have issued `get_network_topology`** during their lifetime — mirroring the schema-12 `thread_diagnostics_updated` opt-in so pre-schema-13 clients never receive an event type they didn't subscribe to. **Outgoing events carry no `require_schema`**; clients detect support via `server_info.schema_version >= 13`. + +See `packages/ws-client/src/models/model.ts` for the exact `NetworkTopology` / `NetworkTopologyNode` / +`NetworkTopologyConnection` wire shapes (each field documented inline). + +> This first iteration derives Thread links from the nodes' own `ThreadNetworkDiagnostics` (neighbor + +> route tables) and classifies externals against the passively-discovered Border Router registry. The +> richer MeshCoP diagnostic enrichment (route64 / childTable → router-to-router links and +> diagnostic-only mesh nodes) is a planned follow-up; the wire model already accommodates it. + ## Schema 12 Minimum supported: 11 (older clients keep working with the pre-12 command shapes). diff --git a/packages/matter-server/test/IntegrationTest.ts b/packages/matter-server/test/IntegrationTest.ts index 1efae3b8..304b05ad 100644 --- a/packages/matter-server/test/IntegrationTest.ts +++ b/packages/matter-server/test/IntegrationTest.ts @@ -129,7 +129,7 @@ describe("Integration Test", function () { expect(info).to.have.property("fabric_id"); expect(info).to.have.property("compressed_fabric_id"); - expect(info.schema_version).to.equal(12); + expect(info.schema_version).to.equal(13); expect(info.min_supported_schema_version).to.equal(11); expect(info.sdk_version).to.be.a("string").that.includes("matter-server"); expect(info.sdk_version).to.be.a("string").that.includes("matter.js"); @@ -172,7 +172,7 @@ describe("Integration Test", function () { expect(diag).to.have.property("info"); expect(diag).to.have.property("nodes"); expect(diag).to.have.property("events"); - expect(diag.info.schema_version).to.equal(12); + expect(diag.info.schema_version).to.equal(13); expect(diag.nodes).to.be.an("array"); expect(diag.events).to.be.an("array"); }); diff --git a/packages/ws-client/src/models/model.ts b/packages/ws-client/src/models/model.ts index e0505ebb..f4a10348 100644 --- a/packages/ws-client/src/models/model.ts +++ b/packages/ws-client/src/models/model.ts @@ -211,6 +211,110 @@ export interface ThreadDiagnosticsBatch { partialReason?: ThreadDiagnosticsPartialReason; } +/** + * Kind of node in a {@link NetworkTopology} graph. @since schema 13 + * - `matter`: a commissioned Matter node on this fabric (carries `node_id`). + * - `border_router`: an mDNS-discovered Thread Border Router (not commissioned here). + * - `thread_unknown`: a Thread device seen in a commissioned node's neighbor table but not + * commissioned on this fabric (e.g. a neighbour on another fabric). + * - `wifi_ap`: a synthetic access-point node grouping Wi-Fi stations by BSSID (id `ap_`, + * colons stripped — e.g. BSSID `11:22:33:44:55:66` → `ap_112233445566`). + */ +export type TopologyNodeKind = "matter" | "border_router" | "thread_unknown" | "wifi_ap"; + +/** + * Role of a topology node. Thread roles mirror ThreadNetworkDiagnostics RoutingRole; + * `station`/`ap` describe Wi-Fi. `unassigned`/absent when unknown. @since schema 13 + */ +export type TopologyRole = + | "leader" + | "router" + | "reed" + | "end_device" + | "sleepy_end_device" + | "unassigned" + | "station" + | "ap"; + +/** + * Link-quality bucket. Thread: LQI 0-3 → none/weak/medium/strong. Wi-Fi: RSSI thresholds. + * `none` means the link was observed dead; `unknown` means no measurement was available + * (the link is still there), so it must not be treated as a dead link. @since schema 13 + */ +export type TopologyStrength = "strong" | "medium" | "weak" | "none" | "unknown"; + +/** One direction's observed link quality. @since schema 13 */ +export interface TopologyDirectionInfo { + strength: TopologyStrength; + /** Thread Link Quality Indicator (0-3 on OpenThread). */ + lqi?: number; + /** RSSI in dBm (Wi-Fi, or Thread neighbor RSSI when available). */ + rssi?: number; +} + +/** + * A node in the network topology graph. Every field beyond `id`/`kind`/`network_type` is + * optional so the shape degrades gracefully across kinds and schema growth. @since schema 13 + */ +export interface NetworkTopologyNode { + /** Stable graph id. Commissioned nodes use the decimal node id; externals use `br_` / + * `unknown_`; Wi-Fi APs use `ap_` with the colons stripped (e.g. `ap_112233445566`). */ + id: string; + kind: TopologyNodeKind; + network_type: "thread" | "wifi" | "ethernet" | "unknown"; + /** Present for `matter` nodes only. */ + node_id?: number | bigint; + role?: TopologyRole; + /** Present for `matter` nodes: reachability of the commissioned node. */ + available?: boolean; + is_bridge?: boolean; + /** 16-char uppercase hex Thread extended (MAC) address, when known. */ + ext_address?: string; + /** Thread short address within the mesh. */ + rloc16?: number; + /** 16-char uppercase hex Thread extended PAN id (network key). */ + ext_pan_id?: string; + /** Thread network name, or Wi-Fi BSSID for `wifi_ap` nodes. */ + network_name?: string; + vendor_name?: string; + model_name?: string; + /** Epoch ms the node was last observed (border routers). */ + last_seen?: number; +} + +/** + * An edge between two topology nodes. Thread neighbor links are directional and may be + * asymmetric: `source_to_target` / `target_to_source` carry each observed direction, while + * top-level `strength` is a summary (the strongest observed direction) for simple renderers. + * @since schema 13 + */ +export interface NetworkTopologyConnection { + source: string; + target: string; + network: "thread" | "wifi"; + /** Summary strength = strongest observed direction. Use the per-direction fields for exact rendering. */ + strength: TopologyStrength; + /** source → target observation, when that direction reported the link. */ + source_to_target?: TopologyDirectionInfo; + /** target → source observation, when that direction reported the link. */ + target_to_source?: TopologyDirectionInfo; + /** True when the edge was supplemented from the Thread route table rather than the neighbor table. */ + via_route_table?: boolean; + /** Thread path cost (1 = direct), when known. */ + path_cost?: number; +} + +/** + * Snapshot of the Matter network topology (Thread mesh + Wi-Fi), returned by + * `get_network_topology` and streamed via `network_topology_updated`. @since schema 13 + */ +export interface NetworkTopology { + /** Epoch ms the snapshot was assembled. */ + collected_at: number; + nodes: NetworkTopologyNode[]; + connections: NetworkTopologyConnection[]; +} + export type WebRtcEventType = "offer" | "answer" | "ice_candidates" | "end"; export interface WebRtcIceCandidate { @@ -327,6 +431,16 @@ export interface APICommands { requestArgs: { ext_pan_id?: string; force?: boolean }; response: ThreadDiagnosticsBatch | ThreadDiagnosticsBatch[] | null; }; + /** + * Full Matter network topology (Thread mesh + Wi-Fi). `refresh` re-reads the Thread + * neighbor/route tables (and Wi-Fi diagnostics) from online nodes before building the + * snapshot; omitted/false builds from the current attribute cache. Issuing this command + * also opts the connection in to the `network_topology_updated` event. @since schema 13 + */ + get_network_topology: { + requestArgs: { refresh?: boolean }; + response: NetworkTopology; + }; open_commissioning_window: { requestArgs: { node_id: number | bigint; @@ -610,6 +724,9 @@ export interface APIEvents { thread_diagnostics_updated: { data: ThreadDiagnosticsBatch; }; + network_topology_updated: { + data: NetworkTopology; + }; webrtc_callback: { data: WebRtcCallbackData; }; @@ -658,6 +775,10 @@ interface ServerEventThreadDiagnosticsUpdated { event: "thread_diagnostics_updated"; data: ThreadDiagnosticsBatch; } +interface ServerEventNetworkTopologyUpdated { + event: "network_topology_updated"; + data: NetworkTopology; +} interface ServerEventWebRtcCallback { event: "webrtc_callback"; data: WebRtcCallbackData; @@ -674,6 +795,7 @@ export type EventMessage = | ServerEventEndpointRemoved | ServerEventInfoUpdated | ServerEventThreadDiagnosticsUpdated + | ServerEventNetworkTopologyUpdated | ServerEventWebRtcCallback; export interface ResultMessageBase { diff --git a/packages/ws-client/src/topology/topology-utils.ts b/packages/ws-client/src/topology/topology-utils.ts index fcacbc12..1ebe6aaa 100644 --- a/packages/ws-client/src/topology/topology-utils.ts +++ b/packages/ws-client/src/topology/topology-utils.ts @@ -150,6 +150,15 @@ export function getThreadRole(node: TopologySourceNode): number | undefined { return typeof v === "number" ? v : undefined; } +/** + * Gets the Thread network name a node reports for itself. + * Uses attribute 0/53/2 (NetworkName, nullable per Matter spec). + */ +export function getThreadNetworkName(node: TopologySourceNode): string | undefined { + const v = node.attributes["0/53/2"]; + return typeof v === "string" && v.length > 0 ? v : undefined; +} + /** * Gets the Thread channel for a node. * Uses attribute 0/53/0 (Channel, nullable per Matter spec). diff --git a/packages/ws-controller/src/controller/MatterController.ts b/packages/ws-controller/src/controller/MatterController.ts index 34fb6953..f2ad4597 100644 --- a/packages/ws-controller/src/controller/MatterController.ts +++ b/packages/ws-controller/src/controller/MatterController.ts @@ -39,6 +39,7 @@ import { Readable } from "node:stream"; import { ConfigStorage } from "../server/ConfigStorage.js"; import { ControllerCommandHandler } from "./ControllerCommandHandler.js"; import { LegacyDataInjector, LegacyServerData } from "./LegacyDataInjector.js"; +import { NetworkTopologyService } from "./NetworkTopologyService.js"; import { resolveServerId } from "./ServerIdResolver.js"; import { ThreadDiagnosticsService } from "./ThreadDiagnosticsService.js"; @@ -163,6 +164,20 @@ function formatDatasetForLog(ds: OperationalDataset): string { return fields.join(", "); } +/** + * The attribute reader {@link NetworkTopologyService} refreshes through. + * + * `fabricFiltered` MUST stay `true`: matter.js only integrates a read into the subscribed + * datasource when the read's filter matches the subscription's (which defaults to filtered), + * so a `false` read is discarded — no cache write, no `attributeChanged`, and the rebuild that + * follows sees the same stale values the refresh was issued to replace. + */ +export function topologyAttributeReader( + handler: Pick, +): (nodeId: number | bigint, paths: string[]) => Promise { + return (nodeId, paths) => handler.handleReadAttributes(NodeId(nodeId), paths, true).then(() => undefined); +} + /** * Split an `OtbrRestCapability.baseUrl` (e.g. `http://[fd00::1]:8081`) into the * host + port the {@link OtbrRestClient} constructor expects. Square-bracketed @@ -196,6 +211,7 @@ export class MatterController { #stopped = false; readonly #credentials = new ThreadCredentialsRegistry(); readonly #threadDiagnostics: ThreadDiagnosticsService; + #networkTopology?: NetworkTopologyService; #webRtcRequestor?: Endpoint; #services: SharedEnvironmentServices; @@ -457,6 +473,33 @@ export class MatterController { return this.#threadDiagnostics; } + /** + * Lazily-constructed network topology service. Created on first access (i.e. the first + * client that queries/subscribes topology), wired to the command handler's node cache + + * events and the Border Router registry. Reused across connections. Throws once the + * controller is stopped rather than starting a service nothing will shut down again. + */ + get networkTopology(): NetworkTopologyService { + if (this.#networkTopology === undefined) { + if (this.#stopped) { + throw new Error("Controller is stopped"); + } + const handler = this.commandHandler; + this.#networkTopology = new NetworkTopologyService({ + listNodes: () => handler.getNodeIds().map(nodeId => handler.getNodeDetails(nodeId)), + readAttributes: topologyAttributeReader(handler), + borderRouters: this.#borderRouterRegistry, + controllerEvents: { + attributeChanged: handler.events.attributeChanged, + nodeAvailabilityChanged: handler.events.nodeAvailabilityChanged, + nodeAdded: handler.events.nodeAdded, + nodeDecommissioned: handler.events.nodeDecommissioned, + }, + }); + } + return this.#networkTopology; + } + #registerStoredThreadCredentials(): void { for (const entry of this.#config.listThreadCredentials()) { registerThreadCredentialsFromHex(this.#credentials, entry.dataset, `stored:${entry.id}`); @@ -578,6 +621,7 @@ export class MatterController { async stop() { this.#stopped = true; await this.#settleBackgroundInit(); + this.#networkTopology?.stop(); if (!this.#threadDiagnosticsDisabled) { await this.#threadDiagnostics.stop(); await this.#borderRouterRegistry.stop(); diff --git a/packages/ws-controller/src/controller/NetworkTopologyService.ts b/packages/ws-controller/src/controller/NetworkTopologyService.ts new file mode 100644 index 00000000..adf291fb --- /dev/null +++ b/packages/ws-controller/src/controller/NetworkTopologyService.ts @@ -0,0 +1,535 @@ +/** + * @license + * Copyright 2025-2026 Open Home Foundation + * SPDX-License-Identifier: Apache-2.0 + */ + +import { + buildExtAddrMap, + buildRloc16Map, + buildThreadEdgePairs, + findUnknownDevices, + getNetworkType, + getThreadExtendedAddressHex, + getThreadExtendedPanId, + getThreadNetworkName, + getThreadRloc16, + getThreadRole, + getWiFiDiagnostics, + type NetworkTopology, + type NetworkTopologyConnection, + type NetworkTopologyNode, + type SignalLevel, + type ThreadConnection, + type ThreadEdgePair, + type ThreadExternalDevice, + type TopologyDirectionInfo, + type TopologyRole, + type TopologySourceNode, + type TopologyStrength, +} from "@matter-server/ws-client"; +import { Logger, Observable, ObserverGroup } from "@matter/main"; +import type { BorderRouterEntry, BorderRouterRegistry } from "@matter/thread-br-client"; + +const logger = Logger.get("NetworkTopologyService"); + +/** + * Clusters whose attribute updates can change the derived topology: + * 0x31/49 NetworkCommissioning (FeatureMap → network type) + * 0x33/51 GeneralDiagnostics (NetworkInterfaces → extended address / MAC) + * 0x35/53 ThreadNetworkDiagnostics (role, neighbor/route tables, rloc16, extPanId) + * 0x36/54 WiFiNetworkDiagnostics (BSSID, RSSI, channel) + * An attribute change outside this set can't affect the graph, so it doesn't schedule a rebuild. + */ +const TOPOLOGY_CLUSTERS = new Set([49, 51, 53, 54]); + +/** + * Thread neighbor + route tables, interface list, own RLOC16, routing role, network name and + * extended PAN id — the inputs to Thread edge derivation and to the built node. RLOC16 (0/53/64) + * is included because edge/unknown matching falls back to it when extended-address matching fails; + * RLOC16, routing role (0/53/1), network name (0/53/2) and ext PAN id (0/53/4) all change when a + * device re-attaches to the mesh — the case a user triggers a refresh to reflect. + */ +export const THREAD_REFRESH_PATHS = ["0/53/7", "0/53/8", "0/51/0", "0/53/64", "0/53/1", "0/53/2", "0/53/4"]; +/** WiFi BSSID + channel + RSSI — the inputs to the WiFi star. */ +export const WIFI_REFRESH_PATHS = ["0/54/0", "0/54/3", "0/54/4"]; + +/** An Observable this service only needs to subscribe to; payloads are inspected structurally. */ +type AnyObservable = Observable; + +/** + * A source node plus the availability + bridge flags surfaced on the wire. Neither is part of + * the pure derivation ({@link TopologySourceNode}, which reads only `node_id`/`attributes`); + * both ride along from the controller's node details ({@link MatterNodeData}). + */ +export type TopologyNode = TopologySourceNode & { available?: boolean; is_bridge?: boolean }; + +/** + * An additional provider of graph nodes (e.g. imported test nodes), registered via + * {@link NetworkTopologyService.addNodeSource}. Its nodes are derivation inputs only: + * {@link NetworkTopologyService.refresh} never issues attribute reads to them. + */ +export interface TopologyNodeSource { + listNodes: () => TopologyNode[]; + nodeAdded?: AnyObservable; + nodeRemoved?: AnyObservable; +} + +export interface NetworkTopologyServiceOptions { + /** Snapshot of the currently commissioned nodes (node id, availability, bridge flag, attribute cache). */ + listNodes: () => TopologyNode[]; + /** Read the given attribute paths from a node, refreshing the server's attribute cache. */ + readAttributes: (nodeId: number | bigint, paths: string[]) => Promise; + /** Passively-discovered Thread Border Router registry (used to classify neighbors + label networks). */ + borderRouters: Pick; + /** Controller lifecycle/attribute events that can change the topology. */ + controllerEvents: { + attributeChanged: AnyObservable; + nodeAvailabilityChanged: AnyObservable; + nodeAdded: AnyObservable; + nodeDecommissioned: AnyObservable; + }; + /** Trailing debounce for change-driven rebuilds. Defaults to {@link NetworkTopologyService.DEFAULT_DEBOUNCE_MS}. */ + debounceMs?: number; + /** Periodic rebuild interval (catches sleepy-device drift). Defaults to {@link NetworkTopologyService.DEFAULT_PERIODIC_MS}. */ + periodicMs?: number; + /** Overall deadline for a {@link NetworkTopologyService.refresh}. Defaults to {@link NetworkTopologyService.DEFAULT_REFRESH_TIMEOUT_MS}. */ + refreshTimeoutMs?: number; + /** Max concurrent per-node reads during {@link NetworkTopologyService.refresh}. Defaults to {@link NetworkTopologyService.DEFAULT_REFRESH_CONCURRENCY}. */ + refreshConcurrency?: number; +} + +/** + * Derives the Matter network topology (Thread mesh + Wi-Fi star) from the controller's + * attribute cache + the passively-discovered Border Router registry, and streams a fresh + * snapshot whenever the derived graph changes. + * + * The heavy derivation (neighbor/route-table parsing, edge-pair building, unknown/BR + * classification) is the shared, DOM-free pipeline in `@matter-server/ws-client`, so the + * server and the dashboard compute the same graph. + * + * Update model (hybrid): change-driven rebuilds are debounced (a burst of attribute + * reports collapses to one), emitted only when the graph actually changed (hash compare, + * excluding the timestamp), plus a slow periodic rebuild so sleepy-device table drift that + * arrives via routers' subscriptions is eventually reflected without an explicit request. + * + * NOTE: this v1 derives from the nodes' own ThreadNetworkDiagnostics (neighbor/route tables) + * plus mDNS-discovered Border Routers. The richer MeshCoP diagnostic enrichment (route64 / + * childTable → router-router links and diagnostic-only mesh nodes) is a planned follow-up; + * the wire model already accommodates it. + */ +export class NetworkTopologyService { + static readonly DEFAULT_DEBOUNCE_MS = 2_000; + static readonly DEFAULT_PERIODIC_MS = 60_000; + static readonly DEFAULT_REFRESH_TIMEOUT_MS = 30_000; + static readonly DEFAULT_REFRESH_CONCURRENCY = 4; + + readonly events = { + topologyUpdated: new Observable<[NetworkTopology]>(), + }; + + readonly #opts: NetworkTopologyServiceOptions; + readonly #auxSources: TopologyNodeSource[] = []; + readonly #observers = new ObserverGroup(); + readonly #debounceMs: number; + readonly #refreshTimeoutMs: number; + readonly #refreshConcurrency: number; + #debounceTimer?: NodeJS.Timeout; + #periodicTimer?: NodeJS.Timeout; + #refreshInFlight?: Promise; + #lastHash?: string; + #stopped = false; + + constructor(opts: NetworkTopologyServiceOptions) { + this.#opts = opts; + this.#debounceMs = opts.debounceMs ?? NetworkTopologyService.DEFAULT_DEBOUNCE_MS; + this.#refreshTimeoutMs = opts.refreshTimeoutMs ?? NetworkTopologyService.DEFAULT_REFRESH_TIMEOUT_MS; + this.#refreshConcurrency = opts.refreshConcurrency ?? NetworkTopologyService.DEFAULT_REFRESH_CONCURRENCY; + const periodicMs = opts.periodicMs ?? NetworkTopologyService.DEFAULT_PERIODIC_MS; + + this.#observers.on(opts.controllerEvents.attributeChanged, (...args: any[]) => { + const clusterId = args[1]?.path?.clusterId; + if (typeof clusterId === "number" && TOPOLOGY_CLUSTERS.has(clusterId)) { + this.#scheduleRebuild(); + } + }); + this.#observers.on(opts.controllerEvents.nodeAvailabilityChanged, () => this.#scheduleRebuild()); + this.#observers.on(opts.controllerEvents.nodeAdded, () => this.#scheduleRebuild()); + this.#observers.on(opts.controllerEvents.nodeDecommissioned, () => this.#scheduleRebuild()); + this.#observers.on(opts.borderRouters.events.added, () => this.#scheduleRebuild()); + this.#observers.on(opts.borderRouters.events.updated, () => this.#scheduleRebuild()); + this.#observers.on(opts.borderRouters.events.removed, () => this.#scheduleRebuild()); + + this.#periodicTimer = setInterval(() => this.#rebuildAndEmit(), periodicMs); + // Don't let the periodic rebuild keep the process alive on shutdown. + this.#periodicTimer.unref?.(); + } + + /** Build a fresh snapshot from the current caches. Pure read — does not emit. */ + getTopology(): NetworkTopology { + return this.#build(); + } + + /** + * Merge an additional node source into the graph (e.g. the WS handler's imported test + * nodes, so the derived topology matches what `get_nodes` serves). Idempotent per source. + */ + addNodeSource(source: TopologyNodeSource): void { + if (this.#stopped || this.#auxSources.includes(source)) return; + this.#auxSources.push(source); + if (source.nodeAdded !== undefined) this.#observers.on(source.nodeAdded, () => this.#scheduleRebuild()); + if (source.nodeRemoved !== undefined) this.#observers.on(source.nodeRemoved, () => this.#scheduleRebuild()); + this.#scheduleRebuild(); + } + + /** + * Re-read the Thread neighbor/route tables (and WiFi diagnostics) from every online + * controller node (aux-source nodes are never read), then rebuild. Reads are + * concurrency-capped and best-effort (a failing/slow node is skipped), mirroring the + * dashboard's "update connections" action. Once the overall deadline expires no further + * reads start; reads already in flight run to completion (their results land in the + * cache and surface via the next rebuild). + * + * Concurrent callers share one in-flight refresh: the read fan-out costs real radio + * traffic on every node, so N clients requesting a refresh at once must not multiply it. + */ + async refresh(): Promise { + if (this.#stopped) return this.#build(); + if (this.#refreshInFlight !== undefined) return this.#refreshInFlight; + + const inFlight = this.#runRefresh().finally(() => { + if (this.#refreshInFlight === inFlight) this.#refreshInFlight = undefined; + }); + this.#refreshInFlight = inFlight; + return inFlight; + } + + async #runRefresh(): Promise { + const tasks: Array<() => Promise> = []; + for (const node of this.#opts.listNodes()) { + if (node.available === false) continue; + const networkType = getNetworkType(node); + const paths = + networkType === "thread" + ? THREAD_REFRESH_PATHS + : networkType === "wifi" + ? WIFI_REFRESH_PATHS + : undefined; + if (paths === undefined) continue; + const nodeId = node.node_id; + tasks.push(() => + this.#opts + .readAttributes(nodeId, paths) + .catch(err => logger.debug(`refresh read failed node=${nodeId}`, err)), + ); + } + + await this.#runWithDeadline(tasks); + + // stop() may have landed while awaiting the reads; a stopped service must not emit. + const topology = this.#build(); + if (!this.#stopped) this.#emitIfChanged(topology); + return topology; + } + + /** Cancel timers and unsubscribe. Idempotent. */ + stop(): void { + this.#stopped = true; + this.#observers.close(); + if (this.#debounceTimer !== undefined) { + clearTimeout(this.#debounceTimer); + this.#debounceTimer = undefined; + } + if (this.#periodicTimer !== undefined) { + clearInterval(this.#periodicTimer); + this.#periodicTimer = undefined; + } + } + + #scheduleRebuild(): void { + if (this.#stopped) return; + if (this.#debounceTimer !== undefined) clearTimeout(this.#debounceTimer); + this.#debounceTimer = setTimeout(() => { + this.#debounceTimer = undefined; + this.#rebuildAndEmit(); + }, this.#debounceMs); + this.#debounceTimer.unref?.(); + } + + #rebuildAndEmit(): void { + if (this.#stopped) return; + try { + this.#emitIfChanged(this.#build()); + } catch (err) { + logger.warn("topology rebuild failed", err); + } + } + + #emitIfChanged(topology: NetworkTopology): void { + const hash = hashTopology(topology); + if (hash === this.#lastHash) return; + this.#lastHash = hash; + logger.debug(`topology changed: ${topology.nodes.length} nodes, ${topology.connections.length} connections`); + this.events.topologyUpdated.emit(topology); + } + + async #runWithDeadline(tasks: Array<() => Promise>): Promise { + if (tasks.length === 0) return; + let expired = false; + let timer: NodeJS.Timeout | undefined; + const deadline = new Promise(resolve => { + timer = setTimeout(() => { + expired = true; + resolve(); + }, this.#refreshTimeoutMs); + timer.unref?.(); + }); + try { + await Promise.race([runWithConcurrency(tasks, this.#refreshConcurrency, () => expired), deadline]); + } finally { + if (timer !== undefined) clearTimeout(timer); + } + } + + #build(): NetworkTopology { + const collectedAt = Date.now(); + const allNodes = [...this.#opts.listNodes(), ...this.#auxSources.flatMap(source => source.listNodes())]; + + const brList = this.#opts.borderRouters.list(); + const brByExt = new Map(); + const networkNameByXp = new Map(); + for (const br of brList) { + brByExt.set(br.extAddressHex.toUpperCase(), br); + if (br.extendedPanIdHex !== undefined && br.networkName !== undefined) { + networkNameByXp.set(br.extendedPanIdHex.toUpperCase(), br.networkName); + } + } + + const nodes: NetworkTopologyNode[] = []; + const connections: NetworkTopologyConnection[] = []; + const threadNodes: Record = {}; + + // --- Commissioned Matter nodes --- + for (const node of allNodes) { + const id = String(node.node_id); + const networkType = getNetworkType(node); + const available = node.available !== false; + const isBridge = node.is_bridge === true ? true : undefined; + + if (networkType === "thread") { + threadNodes[id] = node; + const xp = getThreadExtendedPanId(node); + const extPanIdHex = xp !== undefined ? xp.toString(16).padStart(16, "0").toUpperCase() : undefined; + nodes.push({ + id, + kind: "matter", + network_type: "thread", + node_id: node.node_id, + role: mapThreadRole(getThreadRole(node)), + available, + is_bridge: isBridge, + ext_address: getThreadExtendedAddressHex(node), + rloc16: getThreadRloc16(node), + ext_pan_id: extPanIdHex, + network_name: + getThreadNetworkName(node) ?? + (extPanIdHex !== undefined ? networkNameByXp.get(extPanIdHex) : undefined), + }); + } else if (networkType === "wifi") { + nodes.push({ + id, + kind: "matter", + network_type: "wifi", + node_id: node.node_id, + role: "station", + available, + is_bridge: isBridge, + }); + } else { + nodes.push({ + id, + kind: "matter", + network_type: networkType, + node_id: node.node_id, + available, + is_bridge: isBridge, + }); + } + } + + // --- Thread externals (neighbors not commissioned here) + edges --- + const extAddrMap = buildExtAddrMap(threadNodes); + const rloc16Map = buildRloc16Map(threadNodes); + const externals = findUnknownDevices(threadNodes, extAddrMap, rloc16Map, brByExt); + const edgePairs = buildThreadEdgePairs(threadNodes, extAddrMap, rloc16Map, externals); + const linkedIds = new Set(); + for (const pair of edgePairs.values()) { + const connection = mapThreadPair(pair); + if (connection === undefined) continue; + connections.push(connection); + linkedIds.add(connection.source); + linkedIds.add(connection.target); + } + for (const ext of externals) { + // An external is materialized from any neighbor record, including dead ones (LQI 0), + // whose pair mapThreadPair drops — shipping it anyway would be a node with no edges. + if (!linkedIds.has(ext.id)) continue; + nodes.push(mapExternal(ext)); + } + + // --- WiFi star: one AP pseudo-node per BSSID, station → AP edges --- + const seenAps = new Set(); + for (const node of allNodes) { + if (getNetworkType(node) !== "wifi") continue; + const { bssid, rssi } = getWiFiDiagnostics(node); + if (bssid === null) continue; + const apId = `ap_${bssid.replace(/:/g, "")}`; + if (!seenAps.has(apId)) { + seenAps.add(apId); + nodes.push({ id: apId, kind: "wifi_ap", network_type: "wifi", role: "ap", network_name: bssid }); + } + const strength = rssiToStrength(rssi); + connections.push({ + source: String(node.node_id), + target: apId, + network: "wifi", + strength, + // No RSSI measurement → no direction detail (rather than a synthetic level). + source_to_target: rssi === null ? undefined : { strength, rssi }, + }); + } + + return { collected_at: collectedAt, nodes, connections }; + } +} + +/** Map ThreadNetworkDiagnostics RoutingRole (0/53/1) to a wire role. */ +function mapThreadRole(role: number | undefined): TopologyRole | undefined { + switch (role) { + case 6: + return "leader"; + case 5: + return "router"; + case 4: + return "reed"; + case 3: + return "end_device"; + case 2: + return "sleepy_end_device"; + case 1: + return "unassigned"; + default: + return undefined; + } +} + +function mapExternal(ext: ThreadExternalDevice): NetworkTopologyNode { + if (ext.kind === "br") { + return { + id: ext.id, + kind: "border_router", + network_type: "thread", + role: "router", + ext_address: ext.extAddressHex, + ext_pan_id: ext.extendedPanIdHex?.toUpperCase(), + network_name: ext.networkName, + vendor_name: ext.vendorName, + model_name: ext.modelName, + last_seen: ext.lastSeen, + }; + } + return { + id: ext.id, + kind: "thread_unknown", + network_type: "thread", + role: ext.isRouter ? "router" : undefined, + ext_address: ext.extAddressHex, + ext_pan_id: ext.extendedPanIdHex?.toUpperCase(), + network_name: ext.networkName, + }; +} + +/** + * Fold the 0-2 directional edges of a Thread pair into one wire connection. Both directions are + * preserved (asymmetry stays legible); the summary `strength` is the strongest observed direction. + * Pairs where every observed direction is dead (LQI 0 → "none") are dropped. + */ +function mapThreadPair(pair: ThreadEdgePair): NetworkTopologyConnection | undefined { + const { edgeAB, edgeBA } = pair; // edgeAB: nodeA→nodeB (source→target), edgeBA: nodeB→nodeA + const levels = [edgeAB?.signalLevel, edgeBA?.signalLevel].filter((l): l is SignalLevel => l !== undefined); + if (levels.length === 0 || !levels.some(level => level !== "none")) return undefined; + + return { + source: pair.nodeA, + target: pair.nodeB, + network: "thread", + strength: strongestLevel(levels), + source_to_target: edgeAB !== undefined ? directionInfo(edgeAB) : undefined, + target_to_source: edgeBA !== undefined ? directionInfo(edgeBA) : undefined, + via_route_table: edgeAB?.fromRouteTable === true || edgeBA?.fromRouteTable === true ? true : undefined, + path_cost: edgeAB?.pathCost ?? edgeBA?.pathCost, + }; +} + +function directionInfo(conn: ThreadConnection): TopologyDirectionInfo { + return { strength: conn.signalLevel, lqi: conn.lqi, rssi: conn.rssi ?? undefined }; +} + +const LEVEL_RANK: Record = { unknown: -1, none: 0, weak: 1, medium: 2, strong: 3 }; + +function strongestLevel(levels: TopologyStrength[]): TopologyStrength { + return levels.reduce((best, level) => (LEVEL_RANK[level] > LEVEL_RANK[best] ? level : best), "none"); +} + +/** + * WiFi RSSI → strength (ports the dashboard's -70/-85 dBm thresholds). An unmeasured RSSI is + * "unknown": "none" is reserved for an observed dead link, which consumers filter out of the graph. + */ +function rssiToStrength(rssi: number | null): TopologyStrength { + if (rssi === null) return "unknown"; + if (rssi > -70) return "strong"; + if (rssi > -85) return "medium"; + return "weak"; +} + +/** + * Stable hash of the graph, excluding `collected_at`, so a rebuild that produced an identical + * graph does not re-emit. Nodes/connections are sorted by identity first: their build order isn't + * guaranteed stable (listNodes iteration, neighbor-table arrival order), and the graph is a set — + * a reordering is not a change. Node ids (bigint) are stringified for JSON. + */ +function hashTopology(topology: NetworkTopology): string { + const nodes = [...topology.nodes].sort((a, b) => compareStrings(a.id, b.id)); + const connections = [...topology.connections].sort( + (a, b) => + compareStrings(a.source, b.source) || + compareStrings(a.target, b.target) || + compareStrings(a.network, b.network), + ); + return JSON.stringify({ nodes, connections }, (_key, value) => (typeof value === "bigint" ? `${value}n` : value)); +} + +function compareStrings(a: string, b: string): number { + return a < b ? -1 : a > b ? 1 : 0; +} + +/** + * Run thunks with a bounded number in flight. Each thunk owns its own error handling. + * `cancelled` stops workers from STARTING further tasks; in-flight tasks still run to + * completion (attribute reads offer no cancellation). + */ +async function runWithConcurrency( + tasks: Array<() => Promise>, + concurrency: number, + cancelled: () => boolean = () => false, +): Promise { + let next = 0; + const worker = async (): Promise => { + while (next < tasks.length && !cancelled()) { + const task = tasks[next++]; + await task(); + } + }; + const workerCount = Math.max(1, Math.min(concurrency, tasks.length)); + await Promise.all(Array.from({ length: workerCount }, () => worker())); +} diff --git a/packages/ws-controller/src/server/WebSocketControllerHandler.ts b/packages/ws-controller/src/server/WebSocketControllerHandler.ts index c4e25b5a..b058408e 100644 --- a/packages/ws-controller/src/server/WebSocketControllerHandler.ts +++ b/packages/ws-controller/src/server/WebSocketControllerHandler.ts @@ -24,6 +24,7 @@ import { NodeStates } from "@project-chip/matter.js/device"; import { WebSocketServer } from "ws"; import { ControllerCommandHandler } from "../controller/ControllerCommandHandler.js"; import { MatterController, registerThreadCredentialsFromHex } from "../controller/MatterController.js"; +import type { TopologyNodeSource } from "../controller/NetworkTopologyService.js"; import { TestNodeCommandHandler } from "../controller/TestNodeCommandHandler.js"; import { VendorIds } from "../data/VendorIDs.js"; import { ClusterMap, ClusterMapEntry } from "../model/ModelMapper.js"; @@ -87,14 +88,20 @@ function generateConnectionId(): string { return id.toString(16); } -const SCHEMA_VERSION = 12; +const SCHEMA_VERSION = 13; const MIN_SUPPORTED_SCHEMA_VERSION = 11; // Issuing any of these (schema 12) proves the connection is Thread-aware, so it opts the connection // in to `thread_diagnostics_updated` — even if the request itself errors. See the schema changelog. const THREAD_DIAGNOSTICS_OPT_IN_COMMANDS = new Set(["get_thread_diagnostics", "get_thread_border_routers"]); -const skipMessageContentInLogFor = ["start_listening"]; +// Issuing this (schema 13) opts the connection in to `network_topology_updated`, mirroring the +// thread-diagnostics opt-in: pre-schema-13 clients never subscribed, so they must not receive it. +const NETWORK_TOPOLOGY_OPT_IN_COMMANDS = new Set(["get_network_topology"]); + +// Responses whose payload is large enough that logging it in full just bloats the debug log +// (the full node/attribute dump, or the whole topology graph — hundreds of nodes/edges). +const skipMessageContentInLogFor = ["start_listening", "get_network_topology"]; /** Normalize a requested fabric label: matter.js requires a non-empty label of 1-32 chars. */ function normalizeFabricLabel(label: string | null): string { @@ -127,6 +134,8 @@ export class WebSocketControllerHandler implements WebServerHandler { #controller: MatterController; #commandHandler: ControllerCommandHandler; #testNodeHandler: TestNodeCommandHandler; + /** Stable identity: NetworkTopologyService.addNodeSource dedupes registrations by source object. */ + #testNodeSource: TopologyNodeSource; #config: ConfigStorage; #serverVersion: string; #wss?: WebSocketServer; @@ -154,6 +163,11 @@ export class WebSocketControllerHandler implements WebServerHandler { this.#controller = controller; this.#commandHandler = controller.commandHandler; this.#testNodeHandler = new TestNodeCommandHandler(); + this.#testNodeSource = { + listNodes: () => this.#testNodeHandler.getNodes(), + nodeAdded: this.#testNodeHandler.nodeAdded, + nodeRemoved: this.#testNodeHandler.nodeRemoved, + }; this.#config = config; this.#serverVersion = serverVersion; } @@ -255,6 +269,9 @@ export class WebSocketControllerHandler implements WebServerHandler { // Thread request, so schema-11 clients (all currently deployed HA installs) never receive an // event type they'd crash on. See the schema changelog. let wantsThreadDiagnostics = false; + // network_topology_updated (schema 13) is likewise sent only to connections that have + // issued get_network_topology, so pre-schema-13 clients never receive it. + let wantsNetworkTopology = false; // webrtc_callback is likewise sent only to a connection that has issued a WebRTC provider // command, so it reaches the client driving that camera session rather than every client. let wantsWebRtc = false; @@ -511,6 +528,27 @@ export class WebSocketControllerHandler implements WebServerHandler { } }); + // Registered lazily on first opt-in: touching the networkTopology getter here would + // instantiate the service (timers, event subscriptions) for every connection, even + // ones that never request topology. + const ensureTopologyObserver = () => { + if (wantsNetworkTopology || this.#closed || this.#shuttingDown) return; + wantsNetworkTopology = true; + observers.on(this.#controller.networkTopology.events.topologyUpdated, topology => { + if (this.#closed || this.#shuttingDown) return; + // topologyUpdated is a shared Observable; isolate serialize/send per connection so a + // throw here can't starve other connections. The graph is global (not per-network), so + // coalesce latest-wins under a single key — an older snapshot is worthless. + try { + connection.sendCoalescable("network_topology", () => + toBigIntAwareJson({ event: "network_topology_updated", data: topology }), + ); + } catch (err) { + logger.error(`[${connId}] Failed to send network_topology_updated`, err); + } + }); + }; + observers.on(this.#commandHandler.events.webRtcCallback, data => { if (this.#closed || this.#shuttingDown || !wantsWebRtc) return; // WebRTC signaling is control-plane: never coalesced or dropped, so send reliably. @@ -541,7 +579,13 @@ export class WebSocketControllerHandler implements WebServerHandler { ws.on("message", data => { this.#handleWebSocketRequest(connId, connection, data.toString()) .then( - ({ response, enableListeners, wantsThreadDiagnostics: requested, wantsWebRtc: reqWebRtc }) => { + ({ + response, + enableListeners, + wantsThreadDiagnostics: requested, + wantsNetworkTopology: reqTopology, + wantsWebRtc: reqWebRtc, + }) => { if (this.#closed) return; if (enableListeners) { listening = true; @@ -549,6 +593,9 @@ export class WebSocketControllerHandler implements WebServerHandler { if (requested) { wantsThreadDiagnostics = true; } + if (reqTopology) { + ensureTopologyObserver(); + } if (reqWebRtc) { wantsWebRtc = true; } @@ -625,6 +672,7 @@ export class WebSocketControllerHandler implements WebServerHandler { response: ErrorResultMessage | SuccessResultMessage; enableListeners?: boolean; wantsThreadDiagnostics?: boolean; + wantsNetworkTopology?: boolean; wantsWebRtc?: boolean; }> { let messageId: string | undefined; @@ -727,6 +775,9 @@ export class WebSocketControllerHandler implements WebServerHandler { case "get_thread_diagnostics": result = await this.#handleGetThreadDiagnostics(args); break; + case "get_network_topology": + result = await this.#handleGetNetworkTopology(args); + break; case "open_commissioning_window": result = await this.#handleOpenCommissioningWindow(args); break; @@ -784,6 +835,7 @@ export class WebSocketControllerHandler implements WebServerHandler { }, enableListeners, wantsThreadDiagnostics: command !== undefined && THREAD_DIAGNOSTICS_OPT_IN_COMMANDS.has(command), + wantsNetworkTopology: command !== undefined && NETWORK_TOPOLOGY_OPT_IN_COMMANDS.has(command), wantsWebRtc: command === "send_webrtc_provider_command", }; } catch (err) { @@ -796,6 +848,7 @@ export class WebSocketControllerHandler implements WebServerHandler { details: (err as Error).message, }, wantsThreadDiagnostics: command !== undefined && THREAD_DIAGNOSTICS_OPT_IN_COMMANDS.has(command), + wantsNetworkTopology: command !== undefined && NETWORK_TOPOLOGY_OPT_IN_COMMANDS.has(command), wantsWebRtc: command === "send_webrtc_provider_command", }; } @@ -1350,6 +1403,16 @@ export class WebSocketControllerHandler implements WebServerHandler { return batch === undefined ? null : serializeBatch(batch); } + async #handleGetNetworkTopology(args: ArgsOf<"get_network_topology">): Promise> { + // Imported test nodes are served by get_nodes, so the derived graph must include them + // too. Attached here (not at construction) to keep the service lazily instantiated. + this.#controller.networkTopology.addNodeSource(this.#testNodeSource); + if (args?.refresh === true) { + return this.#controller.networkTopology.refresh(); + } + return this.#controller.networkTopology.getTopology(); + } + async #handleRemoveWifiCredentials( args: ArgsOf<"remove_wifi_credentials">, ): Promise> { diff --git a/packages/ws-controller/test/NetworkTopologyServiceTest.ts b/packages/ws-controller/test/NetworkTopologyServiceTest.ts new file mode 100644 index 00000000..77b81907 --- /dev/null +++ b/packages/ws-controller/test/NetworkTopologyServiceTest.ts @@ -0,0 +1,735 @@ +/** + * @license + * Copyright 2025-2026 Open Home Foundation + * SPDX-License-Identifier: Apache-2.0 + */ + +import type { NetworkTopology, TopologySourceNode } from "@matter-server/ws-client"; +import { NodeId, Observable } from "@matter/main"; +import type { BorderRouterEntry, BorderRouterRegistry } from "@matter/thread-br-client"; +import type { ControllerCommandHandler } from "../src/controller/ControllerCommandHandler.js"; +import { topologyAttributeReader } from "../src/controller/MatterController.js"; +import { + NetworkTopologyService, + THREAD_REFRESH_PATHS, + WIFI_REFRESH_PATHS, +} from "../src/controller/NetworkTopologyService.js"; + +/** A commissioned node as the controller surfaces it to the topology service. */ +type Node = TopologySourceNode & { available?: boolean; is_bridge?: boolean }; + +/** base64 of a byte array — mirrors what the wire delivers for EUI-64 / BSSID attributes. */ +function b64(bytes: number[]): string { + return Buffer.from(bytes).toString("base64"); +} + +const EXT_PAN_HEX = "1122334455667788"; +const EXT_PAN_BIGINT = 0x1122334455667788n; +const NODE1_EXT_BYTES = [0xaa, 0xbb, 0xcc, 0xdd, 0xee, 0xff, 0x00, 0x01]; +const NODE1_EXT_HEX = "AABBCCDDEEFF0001"; +const NODE2_EXT_BYTES = [0xaa, 0xbb, 0xcc, 0xdd, 0xee, 0xff, 0x00, 0x02]; +const BR_EXT_BYTES = [0x11, 0x22, 0xaa, 0xbb, 0xcc, 0x33, 0x44, 0x55]; +const BR_EXT_HEX = "1122AABBCC334455"; +const BR_NODE_ID = `br_${BR_EXT_HEX}`; +const BSSID_BYTES = [0x11, 0x22, 0x33, 0x44, 0x55, 0x66]; +const BSSID_STR = "11:22:33:44:55:66"; +const AP_NODE_ID = "ap_112233445566"; + +/** One entry of a Thread NeighborTable (0/53/7), numeric-keyed per the Matter spec. */ +function neighbor( + ext: number | string, + rloc16: number, + lqi: number, + rssi: number | null, + rxOnWhenIdle = false, +): Record { + return { "0": ext, "2": rloc16, "5": lqi, "6": rssi, "10": rxOnWhenIdle }; +} + +interface ThreadNodeOpts { + role?: number; + rloc16?: number; + extBytes?: number[]; + neighbors?: Record[]; + available?: boolean; + isBridge?: boolean; + networkName?: string; +} + +function mkThread(nodeId: number, opts: ThreadNodeOpts = {}): Node { + const attributes: Record = { + "0/49/65532": 1 << 1, // NetworkCommissioning FeatureMap: Thread + "0/53/4": EXT_PAN_BIGINT, + }; + if (opts.role !== undefined) attributes["0/53/1"] = opts.role; + if (opts.rloc16 !== undefined) attributes["0/53/64"] = opts.rloc16; + if (opts.extBytes !== undefined) attributes["0/51/0"] = [{ "4": b64(opts.extBytes), "7": 4 }]; + if (opts.neighbors !== undefined) attributes["0/53/7"] = opts.neighbors; + if (opts.networkName !== undefined) attributes["0/53/2"] = opts.networkName; + return { node_id: nodeId, available: opts.available ?? true, attributes, is_bridge: opts.isBridge }; +} + +function mkWifi(nodeId: number, opts: { rssi?: number | null; available?: boolean } = {}): Node { + return { + node_id: nodeId, + available: opts.available ?? true, + attributes: { + "0/49/65532": 1 << 0, // NetworkCommissioning FeatureMap: WiFi + "0/54/0": b64(BSSID_BYTES), + // preserve an explicit null (unknown RSSI); only default when omitted + "0/54/4": opts.rssi === undefined ? -55 : opts.rssi, + }, + }; +} + +function mkEthernet(nodeId: number): Node { + return { node_id: nodeId, available: true, attributes: { "0/49/65532": 1 << 2 } }; +} + +function makeBr(overrides: Partial = {}): BorderRouterEntry { + return { + extAddressHex: BR_EXT_HEX, + extendedPanIdHex: EXT_PAN_HEX, + networkName: "OpenThread", + vendorName: "Apple", + modelName: "eero", + addresses: ["fd00::1"], + sources: ["meshcop"], + lastSeen: 1000, + ...overrides, + }; +} + +interface BrEvents { + added: Observable<[BorderRouterEntry]>; + updated: Observable<[BorderRouterEntry]>; + removed: Observable<[BorderRouterEntry]>; +} + +function makeBrEvents(): BrEvents { + return { + added: new Observable<[BorderRouterEntry]>(), + updated: new Observable<[BorderRouterEntry]>(), + removed: new Observable<[BorderRouterEntry]>(), + }; +} + +interface ControllerEventsStub { + attributeChanged: Observable; + nodeAvailabilityChanged: Observable; + nodeAdded: Observable; + nodeDecommissioned: Observable; +} + +function makeControllerEvents(): ControllerEventsStub { + return { + attributeChanged: new Observable(), + nodeAvailabilityChanged: new Observable(), + nodeAdded: new Observable(), + nodeDecommissioned: new Observable(), + }; +} + +function brRegistryFrom( + list: () => BorderRouterEntry[], + events: BrEvents, +): Pick { + return { list, events }; +} + +interface HarnessConfig { + nodes: () => Node[]; + brs?: () => BorderRouterEntry[]; + readAttributes?: (nodeId: number | bigint, paths: string[]) => Promise; + debounceMs?: number; + periodicMs?: number; + refreshTimeoutMs?: number; + refreshConcurrency?: number; +} + +interface Harness { + service: NetworkTopologyService; + emitted: NetworkTopology[]; + controller: ControllerEventsStub; + brEvents: BrEvents; + readCalls: Array<{ nodeId: number | bigint; paths: string[] }>; + fireAttr: (clusterId: number) => void; +} + +const activeHarnesses: Harness[] = []; + +function makeHarness(config: HarnessConfig): Harness { + const controller = makeControllerEvents(); + const brEvents = makeBrEvents(); + const readCalls: Array<{ nodeId: number | bigint; paths: string[] }> = []; + const readAttributes = + config.readAttributes ?? + (async (nodeId: number | bigint, paths: string[]) => { + readCalls.push({ nodeId, paths }); + }); + + const service = new NetworkTopologyService({ + listNodes: config.nodes, + readAttributes, + borderRouters: brRegistryFrom(config.brs ?? (() => []), brEvents), + controllerEvents: controller, + debounceMs: config.debounceMs ?? 5, + // Default the periodic rebuild far out so it can't race the change-driven assertions. + periodicMs: config.periodicMs ?? 600_000, + refreshTimeoutMs: config.refreshTimeoutMs, + refreshConcurrency: config.refreshConcurrency, + }); + + const emitted: NetworkTopology[] = []; + service.events.topologyUpdated.on(t => { + emitted.push(t); + }); + + const harness: Harness = { + service, + emitted, + controller, + brEvents, + readCalls, + fireAttr: clusterId => controller.attributeChanged.emit(1, { path: { clusterId } }), + }; + activeHarnesses.push(harness); + return harness; +} + +const delay = (ms: number): Promise => new Promise(resolve => setTimeout(resolve, ms)); + +describe("NetworkTopologyService", () => { + afterEach(() => { + while (activeHarnesses.length > 0) { + activeHarnesses.pop()!.service.stop(); + } + }); + + describe("getTopology", () => { + function fullMeshHarness(): Harness { + const node1 = mkThread(1, { + role: 6, // leader + rloc16: 1024, + extBytes: NODE1_EXT_BYTES, + neighbors: [ + neighbor(0, 1025, 3, -40), // → node 2 (matched by rloc16) + neighbor(b64(BR_EXT_BYTES), 61440, 2, -70, true), // → external Border Router + ], + }); + const node2 = mkThread(2, { + role: 5, // router + rloc16: 1025, + extBytes: NODE2_EXT_BYTES, + neighbors: [neighbor(0, 1024, 2, -60)], // → node 1 + }); + const node5 = mkWifi(5, { rssi: -55 }); + return makeHarness({ nodes: () => [node1, node2, node5], brs: () => [makeBr()] }); + } + + it("maps a synthetic mesh to matter / border_router / wifi_ap wire nodes", () => { + const { service } = fullMeshHarness(); + const topology = service.getTopology(); + const byId = new Map(topology.nodes.map(n => [n.id, n])); + + expect(topology.nodes).to.have.lengthOf(5); + + const n1 = byId.get("1")!; + expect(n1.kind).to.equal("matter"); + expect(n1.network_type).to.equal("thread"); + expect(n1.node_id).to.equal(1); + expect(n1.role).to.equal("leader"); + expect(n1.available).to.equal(true); + expect(n1.rloc16).to.equal(1024); + expect(n1.ext_address).to.equal(NODE1_EXT_HEX); + expect(n1.ext_pan_id).to.equal(EXT_PAN_HEX); + expect(n1.network_name).to.equal("OpenThread"); + + expect(byId.get("2")!.role).to.equal("router"); + + const wifi = byId.get("5")!; + expect(wifi.kind).to.equal("matter"); + expect(wifi.network_type).to.equal("wifi"); + expect(wifi.role).to.equal("station"); + + const br = byId.get(BR_NODE_ID)!; + expect(br.kind).to.equal("border_router"); + expect(br.network_type).to.equal("thread"); + expect(br.role).to.equal("router"); + expect(br.ext_address).to.equal(BR_EXT_HEX); + expect(br.vendor_name).to.equal("Apple"); + expect(br.model_name).to.equal("eero"); + expect(br.network_name).to.equal("OpenThread"); + + const ap = byId.get(AP_NODE_ID)!; + expect(ap.kind).to.equal("wifi_ap"); + expect(ap.network_type).to.equal("wifi"); + expect(ap.role).to.equal("ap"); + expect(ap.network_name).to.equal(BSSID_STR); + }); + + it("folds both directions of a Thread link and summarises the strongest", () => { + const { service } = fullMeshHarness(); + const topology = service.getTopology(); + + const link = topology.connections.find(c => c.source === "1" && c.target === "2")!; + expect(link.network).to.equal("thread"); + expect(link.strength).to.equal("strong"); + expect(link.source_to_target).to.deep.equal({ strength: "strong", lqi: 3, rssi: -40 }); + expect(link.target_to_source).to.deep.equal({ strength: "medium", lqi: 2, rssi: -60 }); + }); + + it("links a Matter node to the external Border Router it neighbours", () => { + const { service } = fullMeshHarness(); + const topology = service.getTopology(); + + const brLink = topology.connections.find(c => c.target === BR_NODE_ID)!; + expect(brLink.source).to.equal("1"); + expect(brLink.network).to.equal("thread"); + expect(brLink.strength).to.equal("medium"); + expect(brLink.source_to_target).to.deep.equal({ strength: "medium", lqi: 2, rssi: -70 }); + expect(brLink.target_to_source).to.equal(undefined); + }); + + it("builds a Wi-Fi star: one AP pseudo-node per BSSID with a station→AP edge", () => { + const { service } = fullMeshHarness(); + const topology = service.getTopology(); + + const wifiLink = topology.connections.find(c => c.network === "wifi")!; + expect(wifiLink.source).to.equal("5"); + expect(wifiLink.target).to.equal(AP_NODE_ID); + expect(wifiLink.strength).to.equal("strong"); + expect(wifiLink.source_to_target).to.deep.equal({ strength: "strong", rssi: -55 }); + }); + + it("reports an unmeasured Wi-Fi RSSI as strength unknown with no direction detail", () => { + const { service } = makeHarness({ nodes: () => [mkWifi(5, { rssi: null })] }); + const topology = service.getTopology(); + + const wifiLink = topology.connections.find(c => c.network === "wifi")!; + expect(wifiLink.strength).to.equal("unknown"); + expect(wifiLink.source_to_target).to.equal(undefined); + }); + + it("classifies an unmatched neighbour as thread_unknown when no BR matches", () => { + const node1 = mkThread(1, { + role: 5, + rloc16: 1024, + neighbors: [neighbor(b64(BR_EXT_BYTES), 61440, 1, -90, true)], + }); + const { service } = makeHarness({ nodes: () => [node1], brs: () => [] }); + const topology = service.getTopology(); + + const unknown = topology.nodes.find(n => n.kind === "thread_unknown"); + expect(unknown).to.not.equal(undefined); + expect(unknown!.network_type).to.equal("thread"); + expect(topology.nodes.some(n => n.kind === "border_router")).to.equal(false); + }); + + it("drops an external whose only neighbour record is a dead link", () => { + const node1 = mkThread(1, { + role: 5, + rloc16: 1024, + neighbors: [neighbor(b64(BR_EXT_BYTES), 61440, 0, null, true)], // LQI 0 → "none" + }); + const { service } = makeHarness({ nodes: () => [node1], brs: () => [makeBr()] }); + const topology = service.getTopology(); + + expect(topology.connections).to.have.lengthOf(0); + expect(topology.nodes.map(n => n.id)).to.deep.equal(["1"]); + }); + + it("prefers the node's own Thread network name over the Border Router registry", () => { + const node1 = mkThread(1, { role: 5, rloc16: 1024, networkName: "OwnNet" }); + const { service } = makeHarness({ nodes: () => [node1], brs: () => [makeBr()] }); + + expect(service.getTopology().nodes[0].network_name).to.equal("OwnNet"); + }); + + it("marks an offline node unavailable and passes the bridge flag through", () => { + const online = mkThread(1, { role: 5, rloc16: 1024, isBridge: true }); + const offline = mkThread(2, { role: 5, rloc16: 1025, available: false }); + const { service } = makeHarness({ nodes: () => [online, offline] }); + const topology = service.getTopology(); + const byId = new Map(topology.nodes.map(n => [n.id, n])); + + expect(byId.get("1")!.is_bridge).to.equal(true); + expect(byId.get("2")!.available).to.equal(false); + }); + + it("emits an unlinked node for an ethernet device", () => { + const eth = mkEthernet(1); + const { service } = makeHarness({ nodes: () => [eth] }); + const topology = service.getTopology(); + + expect(topology.nodes).to.have.lengthOf(1); + expect(topology.nodes[0].network_type).to.equal("ethernet"); + expect(topology.connections).to.have.lengthOf(0); + }); + }); + + describe("change-driven emission", () => { + it("ignores attribute changes on clusters that can't affect the graph", async () => { + const { emitted, fireAttr } = makeHarness({ + nodes: () => [mkThread(1, { role: 5, rloc16: 1024 })], + }); + + fireAttr(6); // OnOff — irrelevant + await delay(25); + expect(emitted).to.have.lengthOf(0); + + fireAttr(53); // ThreadNetworkDiagnostics — relevant + await delay(25); + expect(emitted).to.have.lengthOf(1); + }); + + it("collapses a burst of attribute reports into a single debounced emit", async () => { + const { emitted, fireAttr } = makeHarness({ + nodes: () => [mkThread(1, { role: 5, rloc16: 1024 })], + debounceMs: 10, + }); + + fireAttr(53); + fireAttr(53); + fireAttr(53); + await delay(30); + + expect(emitted).to.have.lengthOf(1); + }); + + it("does not re-emit when a rebuild produces an identical graph", async () => { + const { emitted, fireAttr } = makeHarness({ + nodes: () => [mkThread(1, { role: 5, rloc16: 1024 })], + }); + + fireAttr(53); + await delay(20); + expect(emitted).to.have.lengthOf(1); + + fireAttr(53); // same underlying data → same hash + await delay(20); + expect(emitted).to.have.lengthOf(1); + }); + + it("does not re-emit when only node/connection ordering changes", async () => { + const node1 = mkThread(1, { role: 5, rloc16: 1024, neighbors: [neighbor(0, 1025, 3, -40)] }); + const node2 = mkThread(2, { role: 5, rloc16: 1025, neighbors: [neighbor(0, 1024, 2, -60)] }); + let order: Node[] = [node1, node2]; + const { emitted, controller } = makeHarness({ nodes: () => order }); + + controller.nodeAdded.emit(1); + await delay(20); + expect(emitted).to.have.lengthOf(1); + + // Same graph, reversed iteration order → the set is unchanged → no re-emit. + order = [node2, node1]; + controller.nodeAdded.emit(2); + await delay(20); + expect(emitted).to.have.lengthOf(1); + }); + + it("re-emits when node lifecycle events change the graph", async () => { + let nodeList: Node[] = [mkThread(1, { role: 5, rloc16: 1024 })]; + const { emitted, controller } = makeHarness({ nodes: () => nodeList }); + + controller.nodeAdded.emit(2); + await delay(20); + expect(emitted).to.have.lengthOf(1); + + // No change → no new emit. + controller.nodeAvailabilityChanged.emit(1, true); + await delay(20); + expect(emitted).to.have.lengthOf(1); + + // A node goes offline → the graph changes → a fresh emit. + nodeList = [mkThread(1, { role: 5, rloc16: 1024, available: false })]; + controller.nodeAvailabilityChanged.emit(1, false); + await delay(20); + expect(emitted).to.have.lengthOf(2); + expect(emitted[1].nodes[0].available).to.equal(false); + + // A node is removed → the graph changes again. + nodeList = []; + controller.nodeDecommissioned.emit(1); + await delay(20); + expect(emitted).to.have.lengthOf(3); + expect(emitted[2].nodes).to.have.lengthOf(0); + }); + + it("rebuilds against the current Border Router registry when a BR is discovered", async () => { + let brList: BorderRouterEntry[] = []; + const node1 = mkThread(1, { + role: 5, + rloc16: 1024, + neighbors: [neighbor(b64(BR_EXT_BYTES), 61440, 2, -70, true)], + }); + const { emitted, brEvents } = makeHarness({ nodes: () => [node1], brs: () => brList }); + + brList = [makeBr()]; + brEvents.added.emit(makeBr()); + await delay(20); + + expect(emitted.length).to.be.greaterThanOrEqual(1); + const latest = emitted[emitted.length - 1]; + expect(latest.nodes.find(n => n.id === BR_NODE_ID)?.kind).to.equal("border_router"); + expect(latest.nodes.some(n => n.kind === "thread_unknown")).to.equal(false); + }); + + it("rebuilds when a Border Router entry is updated (e.g. its network name resolves)", async () => { + let brList: BorderRouterEntry[] = [makeBr({ networkName: undefined })]; + const node1 = mkThread(1, { + role: 5, + rloc16: 1024, + neighbors: [neighbor(b64(BR_EXT_BYTES), 61440, 2, -70, true)], + }); + const { emitted, brEvents } = makeHarness({ nodes: () => [node1], brs: () => brList }); + + brList = [makeBr({ networkName: "ResolvedNet" })]; + brEvents.updated.emit(brList[0]); + await delay(20); + + expect(emitted.length).to.be.greaterThanOrEqual(1); + const latest = emitted[emitted.length - 1]; + expect(latest.nodes.find(n => n.id === BR_NODE_ID)?.network_name).to.equal("ResolvedNet"); + }); + + it("emits once from the periodic rebuild and stays quiet while the graph is stable", async () => { + const { emitted } = makeHarness({ + nodes: () => [mkThread(1, { role: 5, rloc16: 1024 })], + periodicMs: 15, + }); + + await delay(60); // several periodic ticks + expect(emitted).to.have.lengthOf(1); + }); + }); + + describe("addNodeSource", () => { + it("merges aux-source nodes into the graph, with edges resolving across sources", async () => { + const primary = mkThread(1, { + role: 5, + rloc16: 1024, + neighbors: [neighbor(0, 1025, 3, -40)], // → aux node 2, matched by rloc16 + }); + const aux = mkThread(2, { role: 3, rloc16: 1025 }); + const { service, emitted } = makeHarness({ nodes: () => [primary] }); + + const nodeAdded = new Observable(); + service.addNodeSource({ listNodes: () => [aux], nodeAdded }); + await delay(20); // registration schedules a rebuild + + expect(emitted).to.have.lengthOf(1); + const topology = emitted[0]; + expect(topology.nodes.map(n => n.id).sort()).to.deep.equal(["1", "2"]); + const link = topology.connections.find(c => c.source === "1" && c.target === "2"); + expect(link?.strength).to.equal("strong"); + }); + + it("rebuilds when the aux source reports a node added", async () => { + let auxNodes = [mkThread(2, { role: 5, rloc16: 1025 })]; + const nodeAdded = new Observable(); + const { service, emitted } = makeHarness({ nodes: () => [] }); + service.addNodeSource({ listNodes: () => auxNodes, nodeAdded }); + await delay(20); + expect(emitted).to.have.lengthOf(1); + + auxNodes = [...auxNodes, mkThread(3, { role: 5, rloc16: 1026 })]; + nodeAdded.emit(3); + await delay(20); + + expect(emitted).to.have.lengthOf(2); + expect(emitted[1].nodes).to.have.lengthOf(2); + }); + + it("never issues refresh() reads to aux-source nodes and is idempotent per source", async () => { + const { service, readCalls } = makeHarness({ + nodes: () => [mkThread(1, { role: 5, rloc16: 1024 })], + }); + const source = { listNodes: () => [mkThread(2, { role: 5, rloc16: 1025 })] }; + service.addNodeSource(source); + service.addNodeSource(source); // duplicate registration must be a no-op + + const topology = await service.refresh(); + + expect(topology.nodes).to.have.lengthOf(2); // not 3: the duplicate source isn't merged twice + expect(readCalls.map(c => c.nodeId)).to.deep.equal([1]); + }); + }); + + describe("refresh", () => { + it("fans out reads to online Thread/Wi-Fi nodes and skips offline/other nodes", async () => { + const nodes = [ + mkThread(1, { role: 5, rloc16: 1024 }), + mkWifi(2, { rssi: -50 }), + mkThread(3, { role: 5, rloc16: 1026, available: false }), + mkEthernet(4), + ]; + const { service, readCalls } = makeHarness({ nodes: () => nodes }); + + const topology = await service.refresh(); + + expect(topology.nodes.length).to.be.greaterThan(0); + const byNode = new Map(readCalls.map(c => [c.nodeId, c.paths])); + expect(byNode.get(1)).to.deep.equal(THREAD_REFRESH_PATHS); + expect(byNode.get(2)).to.deep.equal(WIFI_REFRESH_PATHS); + expect(byNode.has(3)).to.equal(false); // offline + expect(byNode.has(4)).to.equal(false); // ethernet + expect(readCalls).to.have.lengthOf(2); + }); + + it("caps the number of concurrent reads at refreshConcurrency", async () => { + const nodes = Array.from({ length: 5 }, (_unused, i) => mkThread(i + 1, { role: 5, rloc16: 1024 + i })); + let active = 0; + let maxActive = 0; + const { service } = makeHarness({ + nodes: () => nodes, + refreshConcurrency: 2, + readAttributes: async () => { + active += 1; + maxActive = Math.max(maxActive, active); + await delay(10); + active -= 1; + }, + }); + + await service.refresh(); + + expect(maxActive).to.equal(2); + }); + + it("stops starting reads once the refresh deadline expires", async () => { + const nodes = Array.from({ length: 3 }, (_unused, i) => mkThread(i + 1, { role: 5, rloc16: 1024 + i })); + const started: number[] = []; + let releaseFirst!: () => void; + const firstBlocked = new Promise(resolve => { + releaseFirst = resolve; + }); + const { service } = makeHarness({ + nodes: () => nodes, + refreshConcurrency: 1, + refreshTimeoutMs: 20, + readAttributes: async nodeId => { + started.push(Number(nodeId)); + await firstBlocked; + }, + }); + + await service.refresh(); // the deadline beats the blocked first read + expect(started).to.deep.equal([1]); + + // Releasing the in-flight read must not let the worker pick up the queued ones. + releaseFirst(); + await delay(20); + expect(started).to.deep.equal([1]); + }); + + it("emits after a refresh that changes the graph", async () => { + let nodeList: Node[] = [mkThread(1, { role: 5, rloc16: 1024 })]; + const { service, emitted } = makeHarness({ + nodes: () => nodeList, + readAttributes: async () => { + // Model a read that surfaces a second node the cache didn't have yet. + nodeList = [mkThread(1, { role: 5, rloc16: 1024 }), mkThread(2, { role: 5, rloc16: 1025 })]; + }, + }); + + await service.refresh(); + expect(emitted).to.have.lengthOf(1); + expect(emitted[0].nodes).to.have.lengthOf(2); + }); + + it("coalesces concurrent callers onto a single read fan-out", async () => { + let releaseRead!: () => void; + const readBlocked = new Promise(resolve => { + releaseRead = resolve; + }); + const reads: Array = []; + const { service } = makeHarness({ + nodes: () => [mkThread(1, { role: 5, rloc16: 1024 }), mkThread(2, { role: 5, rloc16: 1025 })], + readAttributes: async nodeId => { + reads.push(nodeId); + await readBlocked; + }, + }); + + const first = service.refresh(); + const second = service.refresh(); + releaseRead(); + const [a, b] = await Promise.all([first, second]); + + expect(reads).to.deep.equal([1, 2]); // one read per node, not one per caller per node + expect(a).to.equal(b); // both callers observe the same snapshot + + // Once settled, a later refresh starts a fresh run. + await service.refresh(); + expect(reads).to.deep.equal([1, 2, 1, 2]); + }); + }); + + describe("stop", () => { + it("stops emitting and refresh() returns a snapshot without emitting", async () => { + const { service, emitted, fireAttr } = makeHarness({ + nodes: () => [mkThread(1, { role: 5, rloc16: 1024 })], + }); + + fireAttr(53); + await delay(20); + expect(emitted).to.have.lengthOf(1); + + service.stop(); + + fireAttr(53); + await delay(20); + expect(emitted).to.have.lengthOf(1); // no further emits after stop + + const topology = await service.refresh(); + expect(topology.nodes).to.have.lengthOf(1); + expect(emitted).to.have.lengthOf(1); + }); + + it("does not emit when stop() lands mid-refresh", async () => { + let releaseRead!: () => void; + const readBlocked = new Promise(resolve => { + releaseRead = resolve; + }); + let nodeList: Node[] = [mkThread(1, { role: 5, rloc16: 1024 })]; + const { service, emitted } = makeHarness({ + nodes: () => nodeList, + refreshTimeoutMs: 10_000, // long enough that only releaseRead settles the read + readAttributes: async () => { + // A read whose result would change the graph (so a build would otherwise emit). + nodeList = [mkThread(1, { role: 5, rloc16: 1024 }), mkThread(2, { role: 5, rloc16: 1025 })]; + await readBlocked; + }, + }); + + const refreshDone = service.refresh(); + await delay(10); // enter the blocked read + service.stop(); // stop lands while the refresh awaits + releaseRead(); + const topology = await refreshDone; + + expect(topology.nodes).to.have.lengthOf(2); // caller still gets the fresh snapshot + expect(emitted).to.have.lengthOf(0); // but a stopped service does not broadcast it + }); + }); + + describe("controller wiring", () => { + it("reads fabric-filtered so matter.js integrates the values into the node cache", async () => { + const reads: Array<{ nodeId: NodeId; paths: string[]; fabricFiltered?: boolean }> = []; + const handler: Pick = { + async handleReadAttributes(nodeId, paths, fabricFiltered) { + reads.push({ nodeId, paths, fabricFiltered }); + return {}; + }, + }; + + await topologyAttributeReader(handler)(1, THREAD_REFRESH_PATHS); + + expect(reads).to.have.lengthOf(1); + expect(reads[0].nodeId).to.equal(NodeId(1)); + expect(reads[0].paths).to.deep.equal(THREAD_REFRESH_PATHS); + expect(reads[0].fabricFiltered).to.equal(true); + }); + }); +}); diff --git a/packages/ws-controller/test/WebSocketCredentialsApiTest.ts b/packages/ws-controller/test/WebSocketCredentialsApiTest.ts index fba994fd..ff375804 100644 --- a/packages/ws-controller/test/WebSocketCredentialsApiTest.ts +++ b/packages/ws-controller/test/WebSocketCredentialsApiTest.ts @@ -69,6 +69,17 @@ function makeStubController(credentials: ThreadCredentialsRegistry) { refreshAllKnown() {}, }; + const stubNetworkTopology = { + events: { topologyUpdated: new Observable() }, + addNodeSource() {}, + getTopology() { + return { collected_at: 0, nodes: [], connections: [] }; + }, + async refresh() { + return { collected_at: 0, nodes: [], connections: [] }; + }, + }; + const stubBorderRouters = { list() { return []; @@ -89,6 +100,11 @@ function makeStubController(credentials: ThreadCredentialsRegistry) { typeof import("../src/controller/ThreadDiagnosticsService.js").ThreadDiagnosticsService >; }, + get networkTopology() { + return stubNetworkTopology as unknown as InstanceType< + typeof import("../src/controller/NetworkTopologyService.js").NetworkTopologyService + >; + }, get borderRouters() { return stubBorderRouters as unknown as InstanceType< typeof import("@matter/thread-br-client").BorderRouterRegistry @@ -103,6 +119,7 @@ interface TestHarness { sendOn(ws: WebSocket, command: string, args: unknown): Promise; emitDiagnosticsBatch(batch: unknown): void; emitWebRtcCallback(data: unknown): void; + emitTopologyUpdated(topology: unknown): void; config: ConfigStorage; close(): Promise; } @@ -201,7 +218,20 @@ async function createHarness(): Promise { (controller.commandHandler.events.webRtcCallback as unknown as Observable<[unknown]>).emit(data); } - return { handle, openClient, sendOn, emitDiagnosticsBatch, emitWebRtcCallback, config, close }; + function emitTopologyUpdated(topology: unknown): void { + (controller.networkTopology.events.topologyUpdated as unknown as Observable<[unknown]>).emit(topology); + } + + return { + handle, + openClient, + sendOn, + emitDiagnosticsBatch, + emitWebRtcCallback, + emitTopologyUpdated, + config, + close, + }; } describe("WebSocket Credentials API", () => { @@ -332,6 +362,49 @@ describe("WebSocket Credentials API", () => { ws.close(); }); + it("get_network_topology returns the built snapshot", async () => { + const res = await h.handle<{ nodes: unknown[]; connections: unknown[] }>("get_network_topology", {}); + expect(res.nodes).to.deep.equal([]); + expect(res.connections).to.deep.equal([]); + }); + + it("withholds network_topology_updated until the connection requests topology", async () => { + const ws = await h.openClient(); + const events = new Array(); + ws.on("message", raw => { + const msg = JSON.parse(raw.toString()) as { event?: string }; + if (msg.event !== undefined) events.push(msg.event); + }); + + const topology = { collected_at: 0, nodes: [], connections: [] }; + + // A pre-schema-13 client that never asks for topology must not receive the new event. + h.emitTopologyUpdated(topology); + await new Promise(r => setTimeout(r, 50)); + expect(events).to.not.include("network_topology_updated"); + + // Issuing get_network_topology opts this connection in. + await new Promise((resolve, reject) => { + const id = "req-topology"; + const onMsg = (raw: WebSocket.RawData) => { + const msg = JSON.parse(raw.toString()) as { message_id?: string }; + if (msg.message_id === id) { + ws.off("message", onMsg); + resolve(); + } + }; + ws.on("message", onMsg); + ws.once("error", reject); + ws.send(JSON.stringify({ message_id: id, command: "get_network_topology", args: {} })); + }); + + h.emitTopologyUpdated(topology); + await new Promise(r => setTimeout(r, 50)); + expect(events).to.include("network_topology_updated"); + + ws.close(); + }); + it("get_thread_diagnostics(ext_pan_id) returns null (not an error) when nothing is cached", async () => { const res = await h.handle("get_thread_diagnostics", { ext_pan_id: "1122334455667788" }); expect(res).to.equal(null); @@ -387,12 +460,12 @@ describe("WebSocket Credentials API", () => { expect(def?.extPanId).to.equal(def?.extPanId?.toUpperCase()); }); - it("server_info reports schema 12 / min 11", async () => { + it("server_info reports schema 13 / min 11", async () => { const info = await h.handle<{ schema_version: number; min_supported_schema_version: number }>( "server_info", {}, ); - expect(info.schema_version).to.equal(12); + expect(info.schema_version).to.equal(13); expect(info.min_supported_schema_version).to.equal(11); }); diff --git a/packages/ws-controller/test/WebSocketRequestSerializationTest.ts b/packages/ws-controller/test/WebSocketRequestSerializationTest.ts index 91c6626a..0be48a8f 100644 --- a/packages/ws-controller/test/WebSocketRequestSerializationTest.ts +++ b/packages/ws-controller/test/WebSocketRequestSerializationTest.ts @@ -57,7 +57,8 @@ async function createHarness(pingNode: () => unknown) { const config = await ConfigStorage.create(env); const fakeCommandHandler = createFakeCommandHandler(pingNode); - // register() subscribes to the controller-level thread-diagnostics observable; supply just that. + // register() eagerly subscribes only to thread-diagnostics; the network-topology observer is + // registered lazily on first get_network_topology, which this test never issues. const fakeController = { commandHandler: fakeCommandHandler, threadDiagnostics: { events: { batchUpdated: new Observable() } }, diff --git a/packages/ws-controller/test/WsBackpressureReproTest.ts b/packages/ws-controller/test/WsBackpressureReproTest.ts index 977894f5..2b3b42d6 100644 --- a/packages/ws-controller/test/WsBackpressureReproTest.ts +++ b/packages/ws-controller/test/WsBackpressureReproTest.ts @@ -114,7 +114,8 @@ async function createHarness() { const config = await ConfigStorage.create(env); const fakeCommandHandler = createFakeCommandHandler(); - // register() subscribes to the controller-level thread-diagnostics observable; supply just that. + // register() eagerly subscribes only to thread-diagnostics; the network-topology observer is + // registered lazily on first get_network_topology, which this test never issues. const fakeController = { commandHandler: fakeCommandHandler, threadDiagnostics: { events: { batchUpdated: new Observable() } }, diff --git a/python_client/tests/test_integration.py b/python_client/tests/test_integration.py index afb60c64..d7b6df35 100644 --- a/python_client/tests/test_integration.py +++ b/python_client/tests/test_integration.py @@ -162,7 +162,7 @@ async def test_02_server_info_command(self, env): assert "fabric_id" in info assert "compressed_fabric_id" in info - assert info["schema_version"] == 12 + assert info["schema_version"] == 13 assert info["min_supported_schema_version"] == 11 assert "matter-server" in info["sdk_version"] assert "matter.js" in info["sdk_version"] @@ -207,7 +207,7 @@ async def test_07_diagnostics(self, env): client: MatterTestClient = env["client"] diag = await client.get_diagnostics() assert diag.info is not None - assert diag.info.schema_version == 12 + assert diag.info.schema_version == 13 assert isinstance(diag.nodes, list) assert isinstance(diag.events, list)