copenhagentruckwash/pleno-vue
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add outbox replay handling with configurable limits and timeouts

Browse Source
This commit is contained in:
Jeppe Bundgaard
2026-06-30 16:15:57 +02:00
parent 48f428a4f0
commit 730ddb1c96
5 changed files with 325 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/features/edgeGateways/edgeGatewayLiveSessions.js
+2
View File
@@ -3,6 +3,7 @@ import {
createEdgeGatewayStreamSession,
unwrapEdgeGatewayResponse,
} from "@/services/edgeGateways.js";
import { allowedPublicBrokerOrigins } from "@/features/edgeGateways/edgeGatewayBrokerConfigSecurity.js";
import { normalizeGatewayWebSocketClose, redactEdgeGatewayUrl } from "@/features/edgeGateways/edgeGatewayErrors.js";
const MOCK_SOCKET_PREFIX = "mock-ws://";
@@ -51,6 +52,7 @@ const getTrustedSocketOrigins = () => {
origins.add(toSocketOrigin(location.origin));
}
allowedPublicBrokerOrigins().forEach((origin) => origins.add(toSocketOrigin(origin)));
getConfiguredTrustedSocketOrigins().forEach((origin) => origins.add(toSocketOrigin(origin)));
return origins;
};
src/views/dashboards/superUserDashboard/selfserve/components/SelfServeConnectivityRelay.vue
+33 -24
View File
@@ -37,55 +37,64 @@ export type ForceStopLaneOptions = {
bill: boolean;
reason?: string | null;
};
type RelayStatusGetter = (_laneId: number) => Promise<any>;
type RelaySetter = (_laneId: number, _on: boolean) => Promise<any>;
const relayStatusGetters: Record<RelayKind, (laneId: number) => Promise<any>> = {
const LOCAL_SELFSERVE_TRANSPORT = "local";
const withLocalTransport = <T extends object>(payload: T): T & { transport: typeof LOCAL_SELFSERVE_TRANSPORT } => ({
...payload,
transport: LOCAL_SELFSERVE_TRANSPORT,
});
const relayStatusGetters: Record<RelayKind, RelayStatusGetter> = {
MACHINE: (laneId: number) => SessionUser.request(
"/modules/self-serve/lane/relay/machine/status",
"GET",
{
withLocalTransport({
lane_id: laneId,
}
})
),
PROGRAM_PICKER: (laneId: number) => SessionUser.request(
"/modules/self-serve/lane/relay/machine_program_picker/status",
"GET",
{
withLocalTransport({
lane_id: laneId,
}
})
),
CLEANER: (laneId: number) => SessionUser.request(
"/modules/self-serve/lane/relay/machine_cleaner/status",
"GET",
{
withLocalTransport({
lane_id: laneId,
}
})
),
};
const relaySetters: Record<RelayKind, (laneId: number, on: boolean) => Promise<any>> = {
const relaySetters: Record<RelayKind, RelaySetter> = {
MACHINE: (laneId: number, on: boolean) => SessionUser.request(
"/modules/self-serve/lane/relay/machine/set",
"POST",
{
withLocalTransport({
lane_id: laneId,
on,
}
})
),
PROGRAM_PICKER: (laneId: number, on: boolean) => SessionUser.request(
"/modules/self-serve/lane/relay/machine_program_picker/set",
"POST",
{
withLocalTransport({
lane_id: laneId,
on,
}
})
),
CLEANER: (laneId: number, on: boolean) => SessionUser.request(
"/modules/self-serve/lane/relay/machine_cleaner/set",
"POST",
{
withLocalTransport({
lane_id: laneId,
on,
}
})
),
};
@@ -127,11 +136,11 @@ const executeLaneCommand = (
) => SessionUser.request(
"/modules/self-serve/lane/command",
"POST",
{
withLocalTransport({
lane_id: normalizeLaneId(laneId),
command,
...payload,
}
})
);
const normalizeSessionListParams = (params: SelfServeSessionListParams = {}): SelfServeSessionListParams => {
@@ -202,10 +211,10 @@ export const relays = {
status: (laneId: number, targets: HardwareTarget[] = ["MACHINE", "PROGRAM_PICKER", "CLEANER"]) => SessionUser.request(
"/modules/self-serve/lane/hardware/batch/status",
"POST",
{
withLocalTransport({
lane_id: normalizeLaneId(laneId),
targets,
}
})
),
set: (
laneId: number,
@@ -213,10 +222,10 @@ export const relays = {
) => SessionUser.request(
"/modules/self-serve/lane/hardware/batch/set",
"POST",
{
withLocalTransport({
lane_id: normalizeLaneId(laneId),
commands,
}
})
),
get: (batchId: string) => SessionUser.request(
`/modules/self-serve/lane/hardware/batch/${encodeURIComponent(String(batchId))}`,
@@ -256,10 +265,10 @@ export const lane = {
open: (laneId: number, gate: LaneGate) => SessionUser.request(
"/modules/self-serve/lane/gate/open",
"POST",
{
withLocalTransport({
lane_id: normalizeLaneId(laneId),
gate: normalizeGate(gate),
}
})
),
},
force: {
@@ -316,7 +325,7 @@ export const lane = {
return SessionUser.request(
"/modules/self-serve/lane/force/machine/enable",
"POST",
payload
withLocalTransport(payload)
);
},
disable: (laneId: number, licensePlate: string | null = null) => {
@@ -335,7 +344,7 @@ export const lane = {
return SessionUser.request(
"/modules/self-serve/lane/force/machine/disable",
"POST",
payload
withLocalTransport(payload)
);
},
},
src/views/dashboards/superUserDashboard/selfserve/components/SelfServeMachineConnectivity.vue
+140 -14
View File
@@ -361,34 +361,141 @@ const resumeActiveHardwareBatches = (laneId: number): Promise<void> => {
return resumePromise;
};
const asRecord = (value: unknown): Record<string, unknown> | null => {
if (!value || typeof value !== "object") {
return null;
}
return value as Record<string, unknown>;
};
const parseBooleanValue = (value: unknown): boolean | null => {
if (typeof value === "boolean") {
return value;
}
if (typeof value === "number" && Number.isFinite(value)) {
return value !== 0;
}
if (typeof value === "string") {
const normalized = value.trim().toLowerCase();
if (["1", "true", "on", "yes", "online"].includes(normalized)) {
return true;
}
if (["0", "false", "off", "no", "offline"].includes(normalized)) {
return false;
}
}
return null;
};
const firstBooleanValue = (...values: unknown[]): boolean | null => {
for (const value of values) {
const parsed = parseBooleanValue(value);
if (parsed !== null) {
return parsed;
}
}
return null;
};
const firstStringValue = (...values: unknown[]): string | null => {
for (const value of values) {
if (typeof value === "string" && value.trim().length > 0) {
return value;
}
}
return null;
};
const firstFiniteNumberValue = (...values: unknown[]): number | null => {
for (const value of values) {
const parsed = Number(value);
if (Number.isFinite(parsed)) {
return parsed;
}
}
return null;
};
const switchZeroStatus = (status: Record<string, unknown> | null): Record<string, unknown> | null => (
asRecord(status?.["switch:0"])
|| asRecord(status?.switch0)
|| asRecord(status?.switch_0)
);
const normalizeRelayStatus = (
laneId: number,
relay: RelayKind,
payload: unknown
): RelayStatus | null => {
if (!payload || typeof payload !== "object") {
const source = asRecord(payload);
if (!source) {
return null;
}
const source = payload as Partial<RelayStatus> & Record<string, unknown>;
const hasRelayState =
Object.prototype.hasOwnProperty.call(source, "online")
|| Object.prototype.hasOwnProperty.call(source, "on");
const status = asRecord(source.status);
const switchStatus = switchZeroStatus(status);
const raw = asRecord(source.raw);
const nestedPayload = asRecord(source.payload);
const nestedStatus = asRecord(nestedPayload?.status);
const nestedSwitchStatus = switchZeroStatus(nestedStatus);
const nestedRaw = asRecord(nestedPayload?.raw);
if (!hasRelayState) {
const online = firstBooleanValue(
source.online,
source.connected,
source.reachable,
status?.online,
switchStatus?.online,
raw?.online,
raw?.connected,
nestedPayload?.online,
nestedPayload?.connected,
nestedStatus?.online,
nestedSwitchStatus?.online,
nestedRaw?.online,
nestedRaw?.connected
);
const on = firstBooleanValue(
source.on,
source.output,
status?.output,
switchStatus?.output,
raw?.on,
raw?.output,
raw?.ison,
nestedPayload?.on,
nestedPayload?.output,
nestedStatus?.output,
nestedSwitchStatus?.output,
nestedRaw?.on,
nestedRaw?.output,
nestedRaw?.ison
);
if (online === null && on === null) {
return null;
}
const relayId = typeof source.relay_id === "string" ? source.relay_id : "";
const relayType = typeof source.relay === "string" ? source.relay : relay;
const lane = Number.isFinite(Number(source.lane_id)) ? Number(source.lane_id) : laneId;
const relayId = firstStringValue(
source.relay_id,
source.relayId,
nestedPayload?.relay_id,
nestedPayload?.relayId
) || "";
const relayType = firstStringValue(source.relay, nestedPayload?.relay) || relay;
const lane = firstFiniteNumberValue(
source.lane_id,
source.laneId,
nestedPayload?.lane_id,
nestedPayload?.laneId
) || laneId;
return {
lane_id: lane,
relay: relayType,
relay_id: relayId,
online: Boolean(source.online),
on: Boolean(source.on),
online: online ?? true,
on: on ?? false,
};
};
@@ -454,13 +561,22 @@ const fetchAllRelayStatusesIndividually = async (
};
};
const refreshRelayStatusAfterCommand = async (
laneId: number,
relay: RelayKind
): Promise<RelayStatus | null> => {
const response = await relayStatusGetters[relay](laneId);
return updateLaneRelayStatus(laneId, relay, extractPayload(response));
};
const setRelayStateIndividually = async (
laneId: number,
relay: RelayKind,
on: boolean
): Promise<RelayStatus | null> => {
const response = await relaySetters[relay](laneId, on);
return updateLaneRelayStatus(laneId, relay, extractPayload(response));
const status = updateLaneRelayStatus(laneId, relay, extractPayload(response));
return status ?? refreshRelayStatusAfterCommand(laneId, relay);
};
const normalizeInProgressDetails = (
@@ -615,10 +731,20 @@ export const setRelayState = async (
{ lane_id: normalizedLaneId, kind: "SET" }
);
const item = batchItems(batch).find((entry) => relayFromBatchTarget(entry?.target) === relay);
if (!item || !item.ok) {
if (!item) {
const batchStatus = String(batch?.status || "").toUpperCase();
if (["FAILED", "PARTIAL", "UNKNOWN_OUTCOME"].includes(batchStatus)) {
throw new Error(`Failed to set ${relay} relay`);
}
await refreshRelayStatusAfterCommand(normalizedLaneId, relay);
} else if (!item.ok) {
throw new Error(item?.error || `Failed to set ${relay} relay`);
} else {
applyHardwareBatchResult(normalizedLaneId, batch);
if (!laneRelayStatuses[normalizedLaneId][relay]) {
await refreshRelayStatusAfterCommand(normalizedLaneId, relay);
}
}
applyHardwareBatchResult(normalizedLaneId, batch);
} catch (error) {
if (!isCloudShellyBatchUnsupportedError(error)) {
throw error;
tests/unit/edge-gateway-live-sessions-security.spec.js
+33
View File
@@ -134,4 +134,37 @@ describe("edge gateway live session websocket security", () => {
await expect(createGatewayShellClient("gateway-7")).rejects.toThrow("Gateway websocket URL must use wss");
expect(MockWebSocket.instances).toHaveLength(0);
});
it("allows approved public broker origins that differ from the app origin", async () => {
shellSessionMock.mockResolvedValue({
data: {
data: {
ws_url: "wss://api.truckwash.io:4433/live-shell",
token: "SHELL-TOKEN-secret-123",
},
},
});
const { createGatewayShellClient } = await import("@/features/edgeGateways/edgeGatewayLiveSessions.js");
await createGatewayShellClient("gateway-7");
expect(MockWebSocket.instances).toHaveLength(1);
expect(MockWebSocket.instances[0].url).toBe("wss://api.truckwash.io:4433/live-shell?gatewayId=gateway-7");
});
it("rejects TLS websocket origins that are not approved for browser sessions", async () => {
shellSessionMock.mockResolvedValue({
data: {
data: {
ws_url: "wss://evil.example/live-shell",
token: "SHELL-TOKEN-secret-123",
},
},
});
const { createGatewayShellClient } = await import("@/features/edgeGateways/edgeGatewayLiveSessions.js");
await expect(createGatewayShellClient("gateway-7")).rejects.toThrow("Gateway websocket URL origin is not trusted");
expect(MockWebSocket.instances).toHaveLength(0);
});
});
tests/unit/self-serve-machine-connectivity.spec.js
+117 -11
View File
@@ -30,7 +30,8 @@ const cleanerStatusEndpoint = "/modules/self-serve/lane/relay/machine_cleaner/st
const machineSetEndpoint = "/modules/self-serve/lane/relay/machine/set";
const gateOpenEndpoint = "/modules/self-serve/lane/gate/open";
const activeBatchStoragePrefix = "truckwash.selfserve.hardwareBatch.";
const cloudShellyBatchUnsupportedError = "Cloud Shelly transport does not support asynchronous relay batches";
const asyncBatchUnsupportedError = "Cloud Shelly transport does not support asynchronous relay batches";
const withLocalTransport = (payload) => ({ ...payload, transport: "local" });
const buildRelayStatus = (
laneId,
@@ -147,6 +148,67 @@ describe("self-serve machine connectivity store", () => {
expect(connectivity.error.value).toBeNull();
});
it("normalizes Shelly-shaped relay status payloads from batch results", async () => {
const laneId = 8;
mocks.request.mockImplementation(async (endpoint) => {
if (endpoint === batchStatusEndpoint) {
return {
data: {
data: buildBatch([
{
target: relayKinds.MACHINE,
relay_id: "machine-relay",
ok: true,
payload: {
status: {
"switch:0": {
output: true,
},
},
},
},
{
target: relayKinds.PROGRAM_PICKER,
relay_id: "program-picker-relay",
ok: true,
payload: {
raw: {
output: 0,
},
},
},
{
target: relayKinds.CLEANER,
relay_id: "cleaner-relay",
ok: true,
payload: {
online: false,
status: {
"switch:0": {
output: true,
},
},
},
},
]),
},
};
}
throw new Error(`Unexpected endpoint: ${endpoint}`);
});
const connectivity = useMachineConnectivity(laneId);
await connectivity.fetchAllRelayStatuses();
expect(connectivity.statuses.value.MACHINE).toMatchObject({ lane_id: laneId, on: true, online: true });
expect(connectivity.statuses.value.PROGRAM_PICKER).toMatchObject({ lane_id: laneId, on: false, online: true });
expect(connectivity.statuses.value.CLEANER).toMatchObject({ lane_id: laneId, on: true, online: false });
expect(connectivity.getDisplayStatus(relayKinds.MACHINE)).toBe("ON");
expect(connectivity.getDisplayStatus(relayKinds.PROGRAM_PICKER)).toBe("OFF");
expect(connectivity.getDisplayStatus(relayKinds.CLEANER)).toBe("OFFLINE");
});
it("toggles a relay and refreshes relay status when set response is envelope-only", async () => {
const laneId = 11;
let machineOn = false;
@@ -156,7 +218,9 @@ describe("self-serve machine connectivity store", () => {
machineOn = Boolean(payload?.commands?.[0]?.on);
return {
data: {
data: buildBatch([buildBatchItem(laneId, relayKinds.MACHINE, { on: Boolean(payload?.commands?.[0]?.on) })]),
data: {
acknowledged: true,
},
},
};
}
@@ -171,6 +235,13 @@ describe("self-serve machine connectivity store", () => {
},
};
}
if (endpoint === machineStatusEndpoint && method === "GET") {
return {
data: {
data: buildRelayStatus(laneId, relayKinds.MACHINE, { on: machineOn }),
},
};
}
throw new Error(`Unexpected endpoint: ${endpoint}`);
});
@@ -181,6 +252,7 @@ describe("self-serve machine connectivity store", () => {
await connectivity.toggleRelay(relayKinds.MACHINE, true);
expect(connectivity.statuses.value.MACHINE?.on).toBe(true);
expect(mocks.request).toHaveBeenCalledWith(machineStatusEndpoint, "GET", withLocalTransport({ lane_id: laneId }));
});
it("waits long enough for slow edge batches to complete", async () => {
@@ -234,15 +306,20 @@ describe("self-serve machine connectivity store", () => {
await connectivity.forceDisableMachine("cd67890");
await connectivity.forceStopLane({ sessionId: 91, bill: true, reason: "clear app runtime" });
await connectivity.stopWash();
await connectivity.executeLaneCommand("START", { customer_number: 42, license_plate: "AB12345" });
await connectivity.openOutsideEntranceGate();
await connectivity.openOutsideExitGate();
expect(mocks.request).toHaveBeenNthCalledWith(1, "/modules/self-serve/lane/force/machine/enable", "POST", {
lane_id: laneId,
duration: 120,
license_plate: "AB12345",
transport: "local",
});
expect(mocks.request).toHaveBeenNthCalledWith(2, "/modules/self-serve/lane/force/machine/disable", "POST", {
lane_id: laneId,
license_plate: "CD67890",
transport: "local",
});
expect(mocks.request).toHaveBeenNthCalledWith(3, "/modules/self-serve/lane/force/stop", "POST", {
lane_id: laneId,
@@ -253,6 +330,24 @@ describe("self-serve machine connectivity store", () => {
expect(mocks.request).toHaveBeenNthCalledWith(4, "/modules/self-serve/lane/command", "POST", {
lane_id: laneId,
command: "STOP",
transport: "local",
});
expect(mocks.request).toHaveBeenNthCalledWith(5, "/modules/self-serve/lane/command", "POST", {
lane_id: laneId,
command: "START",
customer_number: 42,
license_plate: "AB12345",
transport: "local",
});
expect(mocks.request).toHaveBeenNthCalledWith(6, "/modules/self-serve/lane/command", "POST", {
lane_id: laneId,
command: "OPEN_PROPERTY_ACCESS_GATE",
transport: "local",
});
expect(mocks.request).toHaveBeenNthCalledWith(7, "/modules/self-serve/lane/command", "POST", {
lane_id: laneId,
command: "OPEN_PROPERTY_EXIT_GATE",
transport: "local",
});
});
@@ -268,10 +363,12 @@ describe("self-serve machine connectivity store", () => {
expect(mocks.request).toHaveBeenNthCalledWith(1, batchSetEndpoint, "POST", {
lane_id: laneId,
commands: [{ target: "ENTRANCE", action: "OPEN" }],
transport: "local",
});
expect(mocks.request).toHaveBeenNthCalledWith(2, batchSetEndpoint, "POST", {
lane_id: laneId,
commands: [{ target: "EXIT", action: "OPEN" }],
transport: "local",
});
});
@@ -412,12 +509,12 @@ describe("self-serve machine connectivity store", () => {
expect(connectivity.error.value).toContain("program picker unavailable");
});
it("falls back to individual Cloud Shelly relay status requests when async batches are unavailable", async () => {
it("falls back to individual relay status requests when async batches are unavailable", async () => {
const laneId = 23;
mocks.request.mockImplementation(async (endpoint, method) => {
if (endpoint === batchStatusEndpoint && method === "POST") {
throw new Error(cloudShellyBatchUnsupportedError);
throw new Error(asyncBatchUnsupportedError);
}
if (endpoint === machineStatusEndpoint && method === "GET") {
return { data: { data: buildRelayStatus(laneId, relayKinds.MACHINE, { on: true }) } };
@@ -441,18 +538,23 @@ describe("self-serve machine connectivity store", () => {
expect(mocks.request).toHaveBeenCalledWith(batchStatusEndpoint, "POST", {
lane_id: laneId,
targets: ["MACHINE", "PROGRAM_PICKER", "CLEANER"],
transport: "local",
});
expect(mocks.request).toHaveBeenCalledWith(machineStatusEndpoint, "GET", { lane_id: laneId });
expect(mocks.request).toHaveBeenCalledWith(programPickerStatusEndpoint, "GET", { lane_id: laneId });
expect(mocks.request).toHaveBeenCalledWith(cleanerStatusEndpoint, "GET", { lane_id: laneId });
expect(mocks.request).toHaveBeenCalledWith(machineStatusEndpoint, "GET", withLocalTransport({ lane_id: laneId }));
expect(mocks.request).toHaveBeenCalledWith(
programPickerStatusEndpoint,
"GET",
withLocalTransport({ lane_id: laneId })
);
expect(mocks.request).toHaveBeenCalledWith(cleanerStatusEndpoint, "GET", withLocalTransport({ lane_id: laneId }));
});
it("falls back to individual Cloud Shelly relay set requests when async batches are unavailable", async () => {
it("falls back to individual relay set requests when async batches are unavailable", async () => {
const laneId = 25;
mocks.request.mockImplementation(async (endpoint, method, payload) => {
if (endpoint === batchSetEndpoint && method === "POST") {
throw new Error(cloudShellyBatchUnsupportedError);
throw new Error(asyncBatchUnsupportedError);
}
if (endpoint === machineSetEndpoint && method === "POST") {
return {
@@ -472,19 +574,21 @@ describe("self-serve machine connectivity store", () => {
expect(mocks.request).toHaveBeenCalledWith(batchSetEndpoint, "POST", {
lane_id: laneId,
commands: [{ target: "MACHINE", on: true }],
transport: "local",
});
expect(mocks.request).toHaveBeenCalledWith(machineSetEndpoint, "POST", {
lane_id: laneId,
on: true,
transport: "local",
});
});
it("falls back to the synchronous Cloud Shelly gate endpoint when async batches are unavailable", async () => {
it("falls back to the synchronous gate endpoint when async batches are unavailable", async () => {
const laneId = 27;
mocks.request.mockImplementation(async (endpoint, method, payload) => {
if (endpoint === batchSetEndpoint && method === "POST") {
throw new Error(cloudShellyBatchUnsupportedError);
throw new Error(asyncBatchUnsupportedError);
}
if (endpoint === gateOpenEndpoint && method === "POST") {
return { data: { data: { lane_id: payload?.lane_id, gate: payload?.gate, queued: false } } };
@@ -499,10 +603,12 @@ describe("self-serve machine connectivity store", () => {
expect(mocks.request).toHaveBeenCalledWith(batchSetEndpoint, "POST", {
lane_id: laneId,
commands: [{ target: "ENTRANCE", action: "OPEN" }],
transport: "local",
});
expect(mocks.request).toHaveBeenCalledWith(gateOpenEndpoint, "POST", {
lane_id: laneId,
gate: "ENTRANCE",
transport: "local",
});
});