Skip to content

feat: Command Response API WebSocket plumbing #409

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
rusty-art wants to merge 1 commit into from
Closed

feat: Command Response API WebSocket plumbing #409

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
160 changes: 151 additions & 9 deletions mocks/ws.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,14 +17,37 @@ import { NETWORK_MAP_RESPONSE } from "./networkMapResponse.js";
import { PERMIT_JOIN_RESPONSE } from "./permitJoinResponse.js";
import { TOUCHLINK_RESPONSE } from "./touchlinkResponse.js";

// Mutable device state cache — updated on successful SETs, used for GET and reconnect
const deviceStateCache = new Map<string, Record<string, unknown>>();
for (const ds of DEVICE_STATES) {
deviceStateCache.set(ds.topic, merge({}, ds.payload));
}

const cloneDeviceState = (ieee: string) => {
const device = BRIDGE_DEVICES.payload.find((d) => d.ieee_address === ieee);

if (device) {
const deviceState = DEVICE_STATES.find((state) => state.topic === device.friendly_name || state.topic === device.ieee_address);
const topic = deviceStateCache.has(device.friendly_name) ? device.friendly_name : device.ieee_address;
const cached = deviceStateCache.get(topic);

if (cached) {
return { topic, payload: merge({}, cached) };
}
}
};

return merge({}, deviceState);
/**
* Resolve a topic (IEEE address or friendly name) to the friendly name.
* In real Z2M, state updates are always published to the friendly_name topic.
*/
const resolveToFriendlyName = (topic: string): string => {
if (topic.startsWith("0x")) {
const device = BRIDGE_DEVICES.payload.find((d) => d.ieee_address === topic);
if (device) {
return device.friendly_name;
}
}
return topic;
};

const randomString = (len: number): string =>
Expand All @@ -35,8 +58,9 @@ const randomString = (len: number): string =>
// const randomIntInclusive = (min: number, max: number) => Math.floor(Math.random() * (max - min + 1)) + min;

export function startServer() {
const port = Number.parseInt(process.env.MOCK_WS_PORT || "8579", 10);
const wss = new WebSocketServer({
port: 8579,
port,
});

wss.on("connection", (ws) => {
Expand All @@ -59,8 +83,8 @@ export function startServer() {
ws.send(JSON.stringify(message));
}

for (const message of DEVICE_STATES) {
ws.send(JSON.stringify(message));
for (const [topic, payload] of deviceStateCache) {
ws.send(JSON.stringify({ topic, payload }));
}

for (const ds of DEVICE_STATES) {
Expand Down Expand Up @@ -331,15 +355,133 @@ export function startServer() {
break;
}
default: {
if (msg.topic.endsWith("/set")) {
if ("command" in msg.payload) {
const isDeviceSet = msg.topic.endsWith("/request/set") || msg.topic.endsWith("/set");
const isDeviceGet = msg.topic.endsWith("/request/get") || msg.topic.endsWith("/get");

if (isDeviceSet || isDeviceGet) {
const deviceTopic = msg.topic.replace(/\/(?:request\/)?(set|get)$/, "");
const commandType = isDeviceGet ? "get" : "set";

// SET-only special cases: command execution and attribute reading
if (isDeviceSet && "command" in msg.payload) {
setTimeout(() => {
ws.send(JSON.stringify(BRIDGE_LOGGING_EXECUTE_COMMAND));
}, 500);
} else if ("read" in msg.payload) {
} else if (isDeviceSet && "read" in msg.payload) {
setTimeout(() => {
ws.send(JSON.stringify(BRIDGE_LOGGING_READ_ATTR));
}, 500);
} else {
// Transaction Response API: Send response on {device}/response/{set|get}
const requestId = msg.payload?.z2m_transaction ?? msg.payload?.z2m?.request_id;

if (requestId) {
const friendlyName = resolveToFriendlyName(deviceTopic);
const sleepyDelays: Record<string, [number, number]> = {
"test/sleepy-device-fast": [5000, 0],
"test/sleepy-device-slow": [30000, 5000],
};
const sleepyDelay = sleepyDelays[friendlyName];

// Strip z2m_transaction from payload (real backend strips before converter processing)
const { z2m_transaction: _tx, z2m: _z2m, ...dataPayload } = msg.payload;

// Ping: no attribute keys beyond z2m_transaction
const isPing = Object.keys(dataPayload).length === 0;

/** Send state topic update after successful command */
const sendStateUpdate = () => {
const stateTopic = resolveToFriendlyName(deviceTopic);

if (!isDeviceGet) {
// SET: merge set values into persistent cache
const cached = deviceStateCache.get(stateTopic) || deviceStateCache.get(deviceTopic);
if (cached) {
merge(cached, dataPayload);
}
}

// Send full cached state (GET or SET)
const cached = deviceStateCache.get(stateTopic) || deviceStateCache.get(deviceTopic);
if (cached) {
cached.last_seen = new Date().toISOString();
ws.send(JSON.stringify({ topic: stateTopic, payload: { ...cached } }));
}
};

if (isPing) {
// Ping response — immediate
setTimeout(() => {
ws.send(
JSON.stringify({
topic: `${deviceTopic}/response/${commandType}`,
payload: {
data: {},
status: "ok",
z2m_transaction: requestId,
},
}),
);
}, 25);
} else if (sleepyDelay) {
// Sleepy device: converter blocks until device wakes up.
// Fast variant (~5s) responds before frontend's 10s timeout.
// Slow variant (~30-35s) triggers "queued" UX after timeout.
const wakeupDelay = sleepyDelay[0] + Math.random() * sleepyDelay[1];

setTimeout(() => {
ws.send(
JSON.stringify({
topic: `${deviceTopic}/response/${commandType}`,
payload: {
status: "ok",
z2m_transaction: requestId,
data: isDeviceGet ? {} : dataPayload,
},
}),
);

sendStateUpdate();
}, wakeupDelay);
} else {
// Regular device: 50-200ms delay, 90% success
const delay = 50 + Math.random() * 150;

setTimeout(() => {
const isSuccess = Math.random() > 0.1;

if (isSuccess) {
ws.send(
JSON.stringify({
topic: `${deviceTopic}/response/${commandType}`,
payload: {
status: "ok",
z2m_transaction: requestId,
data: isDeviceGet ? {} : dataPayload,
},
}),
);

sendStateUpdate();
} else {
// Error response with actual failed key names
const failedKeys = Object.keys(dataPayload).join(",");

ws.send(
JSON.stringify({
topic: `${deviceTopic}/response/${commandType}`,
payload: {
data: {},
status: "error",
z2m_transaction: requestId,
error: `failed:${failedKeys || "unknown"}`,
},
}),
);
}
}, delay);
}
}
}
} else if (msg.topic.startsWith("bridge/request/")) {
sendResponseOK();
Expand All @@ -351,5 +493,5 @@ export function startServer() {
});
});

console.log("Started WebSocket server");
console.log(`Started WebSocket server on port ${port}`);
}
16 changes: 2 additions & 14 deletions src/components/dashboard-page/DashboardItem.tsx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,10 +2,9 @@ import NiceModal from "@ebay/nice-modal-react";
import { faTrash } from "@fortawesome/free-solid-svg-icons";
import { FontAwesomeIcon } from "@fortawesome/react-fontawesome";
import type { Row } from "@tanstack/react-table";
import { useCallback } from "react";
import { useTranslation } from "react-i18next";
import { useDeviceCommands } from "../../hooks/useDeviceCommands.js";
import type { DashboardTableData } from "../../pages/Dashboard.js";
import { sendMessage } from "../../websocket/WebSocketManager.js";
import Button from "../Button.js";
import DeviceCard from "../device/DeviceCard.js";
import { RemoveDeviceModal } from "../modal/components/RemoveDeviceModal.js";
Expand All @@ -15,18 +14,7 @@ const DashboardItem = ({
original: { sourceIdx, device, deviceState, deviceAvailability, features, lastSeenConfig, removeDevice },
}: Row<DashboardTableData>) => {
const { t } = useTranslation("zigbee");

const onCardChange = useCallback(
async (value: unknown) => {
await sendMessage<"{friendlyNameOrId}/set">(
sourceIdx,
// @ts-expect-error templated API endpoint
`${device.ieee_address}/set`,
value,
);
},
[sourceIdx, device.ieee_address],
);
const { onChange: onCardChange } = useDeviceCommands(sourceIdx, device);

return (
<div
Expand Down
28 changes: 2 additions & 26 deletions src/components/device-page/tabs/Exposes.tsx
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,9 +1,8 @@
import { useCallback } from "react";
import { useTranslation } from "react-i18next";
import { useShallow } from "zustand/react/shallow";
import { useDeviceCommands } from "../../../hooks/useDeviceCommands.js";
import { useAppStore } from "../../../store.js";
import type { Device } from "../../../types.js";
import { sendMessage } from "../../../websocket/WebSocketManager.js";
import Feature from "../../features/Feature.js";
import FeatureWrapper from "../../features/FeatureWrapper.js";
import { getFeatureKey } from "../../features/index.js";
Expand All @@ -16,30 +15,7 @@ type ExposesProps = {
export default function Exposes({ sourceIdx, device }: ExposesProps) {
const { t } = useTranslation("common");
const deviceState = useAppStore(useShallow((state) => state.deviceStates[sourceIdx][device.friendly_name] ?? {}));

const onChange = useCallback(
async (value: Record<string, unknown>) => {
await sendMessage<"{friendlyNameOrId}/set">(
sourceIdx,
// @ts-expect-error templated API endpoint
`${device.ieee_address}/set`,
value,
);
},
[sourceIdx, device.ieee_address],
);

const onRead = useCallback(
async (value: Record<string, unknown>) => {
await sendMessage<"{friendlyNameOrId}/get">(
sourceIdx,
// @ts-expect-error templated API endpoint
`${device.ieee_address}/get`,
value,
);
},
[sourceIdx, device.ieee_address],
);
const { onChange, onRead } = useDeviceCommands(sourceIdx, device);

return device.definition?.exposes?.length ? (
<div className="list bg-base-100">
Expand Down
2 changes: 1 addition & 1 deletion src/components/group-page/GroupMembers.tsx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,7 @@ const GroupMembers = memo(({ sourceIdx, devices, group }: GroupMembersProps) =>
await sendMessage<"{friendlyNameOrId}/set">(
sourceIdx,
// @ts-expect-error templated API endpoint
`${ieee}/set`,
`${ieee}/request/set`,
value,
);
},
Expand Down
35 changes: 35 additions & 0 deletions src/hooks/useDeviceCommands.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,35 @@
import { useCallback } from "react";
import type { Device } from "../types.js";
import { sendMessage } from "../websocket/WebSocketManager.js";

/**
* Hook that routes device set/get commands through the Transaction Response API topics.
* Sends to {ieee}/request/set and {ieee}/request/get instead of the legacy {ieee}/set and {ieee}/get.
*/
export function useDeviceCommands(sourceIdx: number, device: Device) {
const onChange = useCallback(
async (value: unknown) => {
await sendMessage<"{friendlyNameOrId}/set">(
sourceIdx,
// @ts-expect-error templated API endpoint
`${device.ieee_address}/request/set`,
value,
);
},
[sourceIdx, device.ieee_address],
);

const onRead = useCallback(
async (value: Record<string, unknown>) => {
await sendMessage<"{friendlyNameOrId}/get">(
sourceIdx,
// @ts-expect-error templated API endpoint
`${device.ieee_address}/request/get`,
value,
);
},
[sourceIdx, device.ieee_address],
);

return { onChange, onRead };
}
20 changes: 13 additions & 7 deletions src/store.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -466,15 +466,21 @@ export const useAppStore = create<AppState & AppActions>((set, _get, store) => (
const match = newEntry.message.match(PUBLISH_GET_SET_REGEX);

if (match) {
addedToasts = true;
const [, type, key, name, error] = match;

newToasts.push({
sourceIdx,
topic: `${name}/${type}(${key})`,
status: "error",
error,
});
// Suppress toast for superseded commands. When rapidly sending commands
// to sleepy devices, herdsman cancels the older queued command and rejects
// it with "Request superseded". This is expected behavior, not an error.
// The log entry still appears in notifications for debugging.
if (!/request superseded/i.test(newEntry.message)) {
addedToasts = true;
newToasts.push({
sourceIdx,
topic: `${name}/${type}(${key})`,
status: "error",
error,
});
}
}
}
}
Expand Down
15 changes: 15 additions & 0 deletions src/types.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -165,6 +165,21 @@ export type AnySubFeature = BasicFeature | WithAnySubFeatures<FeatureWithSubFeat

export type Toast = { sourceIdx: number; topic: string; status: "ok" | "error"; error: string | undefined; transaction?: string };

/**
* Transaction Response from Z2M backend.
* Received on {device}/response/set or {device}/response/get topics.
*/
export type CommandResponse = {
/** SET: echoed request values; GET: always {} */
data: Record<string, unknown>;
/** Result status */
status: "ok" | "error";
/** Error message — "group:key1,key2|group:key3" format (present when status is "error") */
error?: string;
/** Echoed correlation ID from request */
z2m_transaction?: string;
};

export type RGBColor = {
r: number;
g: number;
Expand Down
Loading