More client decomposition (#3414)

* Decouple DatasourceCache from ClientEndpointStore for reuse by remote nodes

DatasourceCache now takes an options object with writer and optional persist/erase
callbacks instead of a direct ClientEndpointStore reference. RemoteWriteParticipant
takes a generic RemoteWriter function and uses it as both the writer and the
transaction dedup key. ClientEndpointStore.participantFor() is removed as its logic
moves into DatasourceCache.set().

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* Normalize ClientBehaviorBacking change keys from attribute IDs to property names

ClientBehaviorBacking uses primaryKey "id", causing its Datasource to
report changed properties as attribute ID strings (e.g. "0" for onOff).
Downstream consumers (ProtocolService, StateStream) expect property
names.  Add a reverse id-to-name map on RootSupervisor and use it in
ClientBehaviorBacking.broadcastChanges to normalize before broadcast.

Also fix StateStream dirty tracking to deepCopy individual changed
properties rather than the full state object.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
Co-authored-by: mergify[bot] <37929162+mergify[bot]@users.noreply.github.com>

Author: 3 people

packages/node/src/behavior/internal/ClientBehaviorBacking.ts

@@ -56,6 +56,18 @@ export class ClientBehaviorBacking extends BehaviorBacking {
         return options;
     }
 
+    /**
+     * Map attribute ID keys back to property names before broadcasting.
+     *
+     * The Datasource reports changed keys using the backing's primaryKey format.  Since this backing uses
+     * `primaryKey: "id"`, props are attribute ID strings (e.g. "0" for onOff).  Downstream consumers
+     * ({@link ChangeNotificationService}, {@link ProtocolService}) expect property names.
+     */
+    protected override broadcastChanges(props: string[]) {
+        const idToName = this.type.supervisor.propertyIdsAndNames;
+        super.broadcastChanges(props.map(id => idToName.get(id) ?? id));
+    }
+
     override close(): MaybePromise {
         // Prepare the store for reuse in the case of reset
         (this.store as DatasourceCache).reclaimValues?.();

packages/node/src/behavior/supervision/RootSupervisor.ts

@@ -61,6 +61,7 @@ export class RootSupervisor implements ValueSupervisor {
     #root: ValueSupervisor;
     #memberNames?: Set<string>;
     #propertyNamesAndIds?: Map<string, AttributeId | undefined>;
+    #propertyIdsAndNames?: Map<string, string>;
 
     /**
      * Create a new supervisor.
@@ -190,6 +191,24 @@ export class RootSupervisor implements ValueSupervisor {
         return names;
     }
 
+    /**
+     * Reverse of {@link propertyNamesAndIds}: maps attribute ID strings to property names.
+     */
+    get propertyIdsAndNames() {
+        let ids = this.#propertyIdsAndNames;
+        if (!ids) {
+            ids = new Map();
+            for (const member of this.#members) {
+                const id = member.id;
+                if (id !== undefined) {
+                    ids.set(id.toString(), camelize(member.name));
+                }
+            }
+            this.#propertyIdsAndNames = ids;
+        }
+        return ids;
+    }
+
     /**
      * Retrieve members for schema in {@link scope}.
      *

packages/node/src/node/integration/StateStream.ts

@@ -10,7 +10,6 @@ import { Node } from "#node/Node.js";
 import { ServerNode } from "#node/ServerNode.js";
 import { Abort, deepCopy, Duration, Gate, Millis, Timer } from "@matter/general";
 import { DatatypeModel, FieldElement } from "@matter/model";
-import { Val } from "@matter/protocol";
 import { EndpointNumber } from "@matter/types";
 import { ChangeNotificationService } from "./ChangeNotificationService.js";
 
@@ -103,14 +102,17 @@ export function StateStream(
                     // Property update
                     const state = stateOfBehavior(node.id, endpoint.number, behavior.id);
                     state.queueEntry = undefined;
-                    let changes = endpoint.stateOf(behavior);
+
+                    let changes: Record<string, unknown>;
                     if (state.dirty) {
-                        changes = Object.fromEntries(
-                            [...state.dirty].map(name => [name, (changes as Val.Struct)[name]]),
-                        );
+                        const allState = endpoint.stateOf(behavior) as Record<string, unknown>;
+                        changes = {};
+                        for (const name of state.dirty) {
+                            changes[name] = deepCopy(allState[name]);
+                        }
                         state.dirty = undefined;
                     } else {
-                        changes = deepCopy(changes);
+                        changes = deepCopy(endpoint.stateOf(behavior));
                     }
                     yield {
                         kind: "update",

packages/node/src/storage/client/ClientEndpointStore.ts

@@ -6,12 +6,11 @@
 
 import { EndpointStore } from "#storage/EndpointStore.js";
 import { DatasourceStore } from "#storage/server/DatasourceStore.js";
-import { StorageContext, Transaction } from "@matter/general";
+import { StorageContext } from "@matter/general";
 import { PeerAddress } from "@matter/protocol";
 import type { EndpointNumber } from "@matter/types";
 import type { ClientNodeStore } from "./ClientNodeStore.js";
 import { DatasourceCache } from "./DatasourceCache.js";
-import { RemoteWriteParticipant } from "./RemoteWriteParticipant.js";
 
 export class ClientEndpointStore extends EndpointStore {
     #owner: ClientNodeStore;
@@ -42,21 +41,18 @@ export class ClientEndpointStore extends EndpointStore {
         return this.initialValues.get("commissioning")?.["peerAddress"];
     }
 
-    participantFor(transaction: Transaction) {
-        let participant = transaction.getParticipant(this.#owner);
-        if (participant === undefined) {
-            participant = new RemoteWriteParticipant(this.#owner);
-            transaction.addParticipants(participant);
-        }
-        return participant;
-    }
-
     /**
      * Create a {@link Datasource.ExternallyMutableStore} for a behavior.
      */
     createStoreForBehavior(behaviorId: string) {
         const initialValues = this.consumeInitialValues(behaviorId);
-        return DatasourceCache(this, behaviorId, initialValues);
+        return DatasourceCache({
+            writer: this.#owner.write,
+            endpointNumber: this.#number,
+            behaviorId,
+            initialValues,
+            localWriter: this.#owner.localWriter,
+        });
     }
 
     /**

packages/node/src/storage/client/ClientNodeStore.ts

@@ -10,6 +10,7 @@ import { InternalError, StorageContext, StorageContextFactory } from "@matter/ge
 import { EndpointNumber } from "@matter/types";
 import { NodeStore } from "../NodeStore.js";
 import { ClientEndpointStore } from "./ClientEndpointStore.js";
+import { LocalWriter } from "./LocalWriter.js";
 import type { RemoteWriter } from "./RemoteWriter.js";
 
 /**
@@ -20,6 +21,7 @@ export class ClientNodeStore extends NodeStore {
     #storage?: StorageContext;
     #stores = new Map<EndpointNumber, ClientEndpointStore>();
     #write?: RemoteWriter;
+    #localWriter?: LocalWriter;
     #isPreexisting: boolean;
     #onErase?: () => void;
 
@@ -54,10 +56,25 @@ export class ClientNodeStore extends NodeStore {
         this.#write = write;
     }
 
+    get localWriter() {
+        if (this.#localWriter === undefined) {
+            this.#localWriter = new LocalWriter(this);
+        }
+        return this.#localWriter;
+    }
+
     get endpointStores() {
         return this.#stores.values();
     }
 
+    storeForEndpointNumber(endpointNumber: EndpointNumber) {
+        const store = this.#stores.get(endpointNumber);
+        if (store === undefined) {
+            throw new InternalError(`No endpoint store for endpoint ${endpointNumber}`);
+        }
+        return store;
+    }
+
     override async erase() {
         this.#stores = new Map();
         this.#onErase?.();

packages/node/src/storage/client/DatasourceCache.ts

@@ -7,8 +7,10 @@
 import { Datasource } from "#behavior/state/managed/Datasource.js";
 import { InternalError, MaybePromise, Transaction } from "@matter/general";
 import { Val } from "@matter/protocol";
-import type { ClientEndpointStore } from "./ClientEndpointStore.js";
-import type { RemoteWriteParticipant } from "./RemoteWriteParticipant.js";
+import { EndpointNumber } from "@matter/types";
+import type { LocalWriter } from "./LocalWriter.js";
+import { RemoteWriteParticipant } from "./RemoteWriteParticipant.js";
+import type { RemoteWriter } from "./RemoteWriter.js";
 
 /**
  * The default implementation of {@link Datasource.ExternallyMutableStore}.
@@ -27,11 +29,9 @@ export interface DatasourceCache extends Datasource.ExternallyMutableStore {
     erase(): MaybePromise<void>;
 }
 
-export function DatasourceCache(
-    store: ClientEndpointStore,
-    behaviorId: string,
-    initialValues: Val.Struct | undefined,
-): DatasourceCache {
+export function DatasourceCache(options: DatasourceCache.Options): DatasourceCache {
+    const { writer, endpointNumber, behaviorId, initialValues } = options;
+
     let version = initialValues?.[DatasourceCache.VERSION_KEY] as number;
     if (typeof version !== "number") {
         version = Datasource.UNKNOWN_VERSION;
@@ -41,8 +41,12 @@ export function DatasourceCache(
         initialValues,
 
         async set(transaction: Transaction, values: Val.Struct) {
-            const participant = store.participantFor(transaction) as RemoteWriteParticipant;
-            participant.set(store.number, behaviorId, values);
+            let participant = transaction.getParticipant(writer);
+            if (participant === undefined) {
+                participant = new RemoteWriteParticipant(writer);
+                transaction.addParticipants(participant);
+            }
+            (participant as RemoteWriteParticipant).set(endpointNumber, behaviorId, values);
         },
 
         async externalSet(values: Val.StructMap) {
@@ -52,7 +56,7 @@ export function DatasourceCache(
             }
 
             const valuesStruct = Object.fromEntries(values) as Val.Struct;
-            await store.set({ [behaviorId]: valuesStruct });
+            await options.localWriter?.persist(endpointNumber, behaviorId, valuesStruct);
 
             if (this.externalChangeListener) {
                 await this.externalChangeListener(values);
@@ -83,7 +87,7 @@ export function DatasourceCache(
         },
 
         async erase() {
-            await store.eraseStoreForBehavior(behaviorId);
+            await options.localWriter?.erase(endpointNumber, behaviorId);
         },
     };
 }
@@ -95,4 +99,12 @@ export namespace DatasourceCache {
      * This conveys the version to the {@link Datasource}.
      */
     export const VERSION_KEY = "__version__";
+
+    export interface Options {
+        writer: RemoteWriter;
+        endpointNumber: EndpointNumber;
+        behaviorId: string;
+        initialValues?: Val.Struct;
+        localWriter?: LocalWriter;
+    }
 }

packages/node/src/storage/client/LocalWriter.ts

@@ -0,0 +1,29 @@
+/**
+ * @license
+ * Copyright 2022-2026 Matter.js Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+import { MaybePromise } from "@matter/general";
+import { Val } from "@matter/protocol";
+import { EndpointNumber } from "@matter/types";
+import type { ClientNodeStore } from "./ClientNodeStore.js";
+
+/**
+ * Handles local storage persistence for client datasource caches.
+ */
+export class LocalWriter {
+    #nodeStore: ClientNodeStore;
+
+    constructor(nodeStore: ClientNodeStore) {
+        this.#nodeStore = nodeStore;
+    }
+
+    async persist(endpointNumber: EndpointNumber, behaviorId: string, values: Val.Struct) {
+        await this.#nodeStore.storeForEndpointNumber(endpointNumber).set({ [behaviorId]: values });
+    }
+
+    erase(endpointNumber: EndpointNumber, behaviorId: string): MaybePromise<void> {
+        return this.#nodeStore.storeForEndpointNumber(endpointNumber).eraseStoreForBehavior(behaviorId);
+    }
+}

packages/node/src/storage/client/RemoteWriteParticipant.ts

@@ -7,7 +7,6 @@
 import { Transaction } from "@matter/general";
 import { Val } from "@matter/protocol";
 import { EndpointNumber } from "@matter/types";
-import type { ClientNodeStore } from "./ClientNodeStore.js";
 import type { RemoteWriter } from "./RemoteWriter.js";
 
 /**
@@ -17,13 +16,13 @@ import type { RemoteWriter } from "./RemoteWriter.js";
  */
 export class RemoteWriteParticipant implements Transaction.Participant {
     #request: RemoteWriter.Request = [];
-    #store: ClientNodeStore;
+    #writer: RemoteWriter;
 
     /**
-     * There is one participant for each transaction/client node pair.  We therefore use the store as the role.
+     * There is one participant for each transaction/writer pair.  We use the writer function itself as the dedup key.
      */
     get role() {
-        return this.#store;
+        return this.#writer;
     }
 
     /**
@@ -45,18 +44,18 @@ export class RemoteWriteParticipant implements Transaction.Participant {
         const request = this.#request;
         this.#request = [];
 
-        await this.#store.write(request);
+        await this.#writer(request);
     }
 
     rollback() {
         this.#request = [];
     }
 
     toString() {
-        return `writer#${this.#store.id}`;
+        return `remote-writer`;
     }
 
-    constructor(store: ClientNodeStore) {
-        this.#store = store;
+    constructor(writer: RemoteWriter) {
+        this.#writer = writer;
     }
 }

packages/node/src/storage/index.ts

@@ -7,6 +7,7 @@
 export * from "./client/ClientNodeStore.js";
 export * from "./client/ClientNodeStores.js";
 export * from "./client/DatasourceCache.js";
+export * from "./client/LocalWriter.js";
 export * from "./NodeStore.js";
 export * from "./server/DatasourceStore.js";
 export * from "./server/ServerEndpointStore.js";