copenhagentruckwash/api
1
0
Fork 0
Code Issues Pull Requests 1 Actions 2 Packages Projects Releases Wiki Activity

Update relay-handling logic and test assertions for device binding and local IP resolution

Browse Source 
- Correct test cases to ensure proper relay IDs are switched.
- Add robust local IP resolution for relay-device bindings, including caching and inventory backfill.
- Introduce fast-path options for relay status and switch dispatch.
- Validate PHP extensions (`curl`, `sqlite3`) in edge agent images.
This commit is contained in:
Jeppe Bundgaard
2026-04-27 16:02:36 +02:00
parent 7bd940de37
commit f2e2b9a8f4
25 changed files with 938 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files
services/edge-agent/Dockerfile.test-gateway
+1 -3
View File
@@ -3,9 +3,7 @@ FROM php:8.2-cli
WORKDIR /opt/truckwash-edge-agent
RUN set -eux; \
apt-get update; \
apt-get install -y --no-install-recommends ca-certificates; \
rm -rf /var/lib/apt/lists/*
php -r 'foreach (["curl", "sqlite3"] as $extension) { if (!extension_loaded($extension)) { fwrite(STDERR, "Missing PHP extension: {$extension}\n"); exit(1); } }'
COPY services/nginx/app/resources/edge-gateway-agent/ ./
services/edge-agent/dist/agent.mjs
Vendored
+67 -9
View File
@@ -17,6 +17,7 @@ const DEFAULT_UPDATE_VERIFICATION_TIMEOUT_SECONDS = 45;
const DEFAULT_UPDATE_VERIFY_INTERVAL_MS = 500;
const DEFAULT_UPDATE_RESTART_GRACE_MS = 150;
const DEFAULT_BROKER_RECONNECT_DELAY_MS = 1500;
const DEFAULT_SHELLY_LOCAL_HTTP_TIMEOUT_MS = 1200;
const UPDATE_VERIFY_COMMAND = "post-update-verify";
const execFile = promisify(execFileCallback);
@@ -426,8 +427,55 @@ export async function claimIfNeeded(config, configPath, fetchImpl = fetch) {
return nextConfig;
}
async function fetchJson(url, fetchImpl = fetch) {
const response = await fetchImpl(url);
function makeHttpTimeoutError(timeoutMs) {
const error = new Error(`HTTP request timed out after ${timeoutMs}ms`);
error.code = "EDGE_AGENT_HTTP_TIMEOUT";
return error;
}
function isHttpTimeoutError(error) {
return error?.code === "EDGE_AGENT_HTTP_TIMEOUT";
}
function resolveShellyLocalHttpTimeoutMs(options = {}) {
const configured = Number(options.timeoutMs ?? process.env.EDGE_SHELLY_LOCAL_HTTP_TIMEOUT_MS);
if (Number.isFinite(configured) && configured > 0) {
return Math.max(50, Math.floor(configured));
}
return DEFAULT_SHELLY_LOCAL_HTTP_TIMEOUT_MS;
}
async function fetchJson(url, fetchImpl = fetch, options = {}) {
const timeoutMs = Number(options.timeoutMs || 0);
let timeout = null;
let controller = null;
const requestOptions = {};
if (timeoutMs > 0 && typeof AbortController !== "undefined") {
controller = new AbortController();
requestOptions.signal = controller.signal;
}
let response;
try {
const fetchPromise = Promise.resolve().then(() => fetchImpl(url, requestOptions));
response = timeoutMs > 0
? await Promise.race([
fetchPromise,
new Promise((_, reject) => {
timeout = setTimeout(() => {
controller?.abort();
reject(makeHttpTimeoutError(timeoutMs));
}, timeoutMs);
}),
])
: await fetchPromise;
} finally {
if (timeout !== null) {
clearTimeout(timeout);
}
}
if (!response.ok) {
throw new Error(`HTTP ${response.status}`);
}
@@ -458,7 +506,9 @@ export async function discoverShellyDevices(options = {}, fetchImpl = fetch) {
await Promise.all(candidateIps.map(async (ip) => {
try {
const identity = await fetchJson(`http://${ip}/shelly`, fetchImpl);
const identity = await fetchJson(`http://${ip}/shelly`, fetchImpl, {
timeoutMs: resolveShellyLocalHttpTimeoutMs(options),
});
discovered.push({
id: identity.mac || identity.id || ip,
device_id: identity.mac || identity.id || ip,
@@ -482,19 +532,23 @@ export async function discoverShellyDevices(options = {}, fetchImpl = fetch) {
export async function getRelayStatus(payload, fetchImpl = fetch) {
const ip = payload.localIp || payload.local_ip || payload.ip;
const channel = Number.isInteger(payload.channel) ? payload.channel : 0;
const timeoutMs = resolveShellyLocalHttpTimeoutMs(payload);
if (!ip) {
throw new Error("Missing relay local IP");
}
try {
const rpc = await fetchJson(`http://${ip}/rpc/Switch.GetStatus?id=${channel}`, fetchImpl);
const rpc = await fetchJson(`http://${ip}/rpc/Switch.GetStatus?id=${channel}`, fetchImpl, { timeoutMs });
return {
online: true,
on: Boolean(rpc.output),
raw: rpc,
};
} catch {
const legacy = await fetchJson(`http://${ip}/relay/${channel}`, fetchImpl);
} catch (error) {
if (isHttpTimeoutError(error)) {
throw error;
}
const legacy = await fetchJson(`http://${ip}/relay/${channel}`, fetchImpl, { timeoutMs });
return {
online: true,
on: Boolean(legacy.ison ?? legacy.output),
@@ -507,19 +561,23 @@ export async function setRelayState(payload, fetchImpl = fetch) {
const ip = payload.localIp || payload.local_ip || payload.ip;
const channel = Number.isInteger(payload.channel) ? payload.channel : 0;
const on = Boolean(payload.on);
const timeoutMs = resolveShellyLocalHttpTimeoutMs(payload);
if (!ip) {
throw new Error("Missing relay local IP");
}
try {
const rpc = await fetchJson(`http://${ip}/rpc/Switch.Set?id=${channel}&on=${on ? "true" : "false"}`, fetchImpl);
const rpc = await fetchJson(`http://${ip}/rpc/Switch.Set?id=${channel}&on=${on ? "true" : "false"}`, fetchImpl, { timeoutMs });
return {
online: true,
on: Boolean(rpc.output ?? on),
raw: rpc,
};
} catch {
const legacy = await fetchJson(`http://${ip}/relay/${channel}?turn=${on ? "on" : "off"}`, fetchImpl);
} catch (error) {
if (isHttpTimeoutError(error)) {
throw error;
}
const legacy = await fetchJson(`http://${ip}/relay/${channel}?turn=${on ? "on" : "off"}`, fetchImpl, { timeoutMs });
return {
online: true,
on: Boolean(legacy.ison ?? legacy.output ?? on),
services/edge-agent/test/agent.test.mjs
+17
View File
@@ -120,6 +120,23 @@ test("relay status and switch commands support both Shelly RPC and legacy endpoi
assert.equal(switched.on, false);
});
test("relay switch commands fail quickly when the local Shelly request stalls", async () => {
let calls = 0;
const hangingFetch = async () => {
calls += 1;
return new Promise(() => {});
};
const startedAt = Date.now();
await assert.rejects(
() => setRelayState({ localIp: "10.1.0.31", channel: 0, on: true, timeoutMs: 50 }, hangingFetch),
/HTTP request timed out after 50ms/
);
assert.equal(calls, 1);
assert.ok(Date.now() - startedAt < 500);
});
test("runUpdate stages a pending verification restart after installing new artifacts", async () => {
const tempDir = await mkdtemp(path.join(os.tmpdir(), "edge-agent-update-"));
const configPath = path.join(tempDir, "config.json");
services/edge-broker/server.mjs
+77 -4
View File
@@ -106,6 +106,19 @@ function sendJson(ws, payload) {
return true;
}
function currentTimestamp() {
return new Date().toISOString();
}
function normalizeErrorMessage(error, fallback = "Connection step failed") {
if (error instanceof Error && error.message) {
return error.message;
}
const message = String(error || "").trim();
return message || fallback;
}
export function createBrokerServer(options = {}) {
const sharedSecret = options.sharedSecret ?? process.env.EDGE_BROKER_SHARED_SECRET ?? "";
const managerUrl = resolveManagerUrl(options);
@@ -331,6 +344,44 @@ export function createBrokerServer(options = {}) {
return syncPromise;
};
const sendAgentWelcome = (ws) =>
sendJson(ws, {
type: "WELCOME",
target: "edge-agent",
gatewayId: String(ws.gatewayId || ""),
connectionId: ws.connectionId || null,
agentInstanceId: ws.agentInstanceId || null,
serverTime: currentTimestamp(),
broker: {
authMode,
},
});
const sendAgentConnectionProgress = (ws, stage, status, message, details = {}) =>
sendJson(ws, {
type: "CONNECTION_PROGRESS",
gatewayId: String(ws.gatewayId || ""),
connectionId: ws.connectionId || null,
stage,
status,
message,
...details,
serverTime: currentTimestamp(),
});
const sendAgentConnectionError = (ws, stage, error, details = {}) =>
sendJson(ws, {
type: "CONNECTION_ERROR",
gatewayId: String(ws.gatewayId || ""),
connectionId: ws.connectionId || null,
stage,
status: "failed",
message: normalizeErrorMessage(error),
error: normalizeErrorMessage(error),
...details,
serverTime: currentTimestamp(),
});
const server = http.createServer(async (req, res) => {
try {
const url = new URL(req.url, "http://localhost");
@@ -421,6 +472,9 @@ export function createBrokerServer(options = {}) {
ws.agentInstanceId = agentInstanceId || null;
ws.connectionId = randomUUID();
agents.set(gatewayId, ws);
wss.emit("connection", ws, req);
sendAgentWelcome(ws);
sendAgentConnectionProgress(ws, "presence", "started", "Reporting broker presence to the edge manager.");
reportGatewayPresence(gatewayId, {
status: "connected",
connectionId: ws.connectionId,
@@ -428,15 +482,30 @@ export function createBrokerServer(options = {}) {
remote_address: req.socket.remoteAddress || null,
agent_instance_id: ws.agentInstanceId,
},
}).catch(() => {});
})
.then(() => {
sendAgentConnectionProgress(ws, "presence", "succeeded", "Broker presence was reported.");
})
.catch((error) => {
sendAgentConnectionError(ws, "presence", error);
});
broadcastGatewayEvent(gatewayId, {
type: "presence.changed",
gatewayId,
status: "connected",
connectionId: ws.connectionId,
});
syncGatewayBacklog(gatewayId, ws).catch(() => {});
wss.emit("connection", ws, req);
sendAgentConnectionProgress(ws, "backlog", "started", "Synchronizing queued gateway work.");
syncGatewayBacklog(gatewayId, ws)
.then((result) => {
sendAgentConnectionProgress(ws, "backlog", "succeeded", "Queued gateway work was synchronized.", {
queued: Boolean(result?.queued),
dispatchCount: Array.isArray(result?.dispatch) ? result.dispatch.length : 0,
});
})
.catch((error) => {
sendAgentConnectionError(ws, "backlog", error);
});
});
return;
}
@@ -549,7 +618,11 @@ export function createBrokerServer(options = {}) {
}
if (message.type === "TELEMETRY") {
const payload = message.payload || {};
const payload = {
...(message.payload || {}),
broker_connection_id: ws.connectionId || null,
broker_agent_instance_id: ws.agentInstanceId || null,
};
let ingested = null;
let ingestError = null;
try {
services/edge-broker/test/broker.test.mjs
+127 -9
View File
@@ -4,12 +4,6 @@ import WebSocket from "ws";
import { createBrokerServer } from "../server.mjs";
function waitForMessage(socket) {
return new Promise((resolve) => {
socket.once("message", (raw) => resolve(JSON.parse(raw.toString())));
});
}
function collectMessages(socket) {
const messages = [];
socket.on("message", (raw) => {
@@ -132,7 +126,10 @@ test("broker forwards browser shell input, resize, and close events to the agent
const agentMessages = collectMessages(agent);
const browser = new WebSocket(`ws://127.0.0.1:${port}/ws/browser-shell?token=session-token`);
await new Promise((resolve) => browser.once("open", resolve));
await waitForMessage(agent);
await waitFor(
() => agentMessages.some((message) => message.type === "OPEN_ROOT_SHELL" && message.payload.sessionId === "shell-2"),
{ description: "agent shell open request" }
);
browser.send(JSON.stringify({ type: "input", data: "ls\r" }));
browser.send(JSON.stringify({ type: "resize", cols: 140, rows: 44 }));
@@ -214,7 +211,10 @@ test("broker closes browser shell sessions when the agent disconnects before she
const browser = new WebSocket(`ws://127.0.0.1:${port}/ws/browser-shell?token=session-token`);
const browserMessages = collectMessages(browser);
await new Promise((resolve) => browser.once("open", resolve));
await waitForMessage(agent);
await waitFor(
() => agentMessages.some((message) => message.type === "OPEN_ROOT_SHELL" && message.payload.sessionId === "shell-4"),
{ description: "agent shell open request before disconnect" }
);
agent.terminate();
await waitForClose(browser);
@@ -227,6 +227,119 @@ test("broker closes browser shell sessions when the agent disconnects before she
await broker.close();
});
test("broker sends an agent welcome before connection progress and backlog dispatch", async () => {
const broker = createBrokerServer({
authMode: "stub",
reportGatewayPresence: async () => ({}),
requestGatewayBacklog: async () => ({
dispatch: [
{
type: "TASK_DISPATCH",
taskType: "OPERATION",
operation: {
id: 91,
type: "DISCOVERY",
request: {},
},
},
],
}),
});
const address = await broker.listen(0);
const port = address.port;
const agent = new WebSocket(`ws://127.0.0.1:${port}/ws/agent?gatewayId=701&token=agent-token`);
const messages = collectMessages(agent);
await new Promise((resolve) => agent.once("open", resolve));
await waitFor(
() => messages.some((message) => message.type === "TASK_DISPATCH" && message.operation?.id === 91),
{ description: "backlog dispatch after connection welcome" }
);
assert.equal(messages[0].type, "WELCOME");
assert.equal(messages[0].gatewayId, "701");
assert.equal(typeof messages[0].connectionId, "string");
assert.ok(messages[0].connectionId.length > 0);
const firstProgressIndex = messages.findIndex((message) => message.type === "CONNECTION_PROGRESS");
const dispatchIndex = messages.findIndex((message) => message.type === "TASK_DISPATCH");
assert.ok(firstProgressIndex > 0);
assert.ok(dispatchIndex > firstProgressIndex);
assert.ok(
messages.some(
(message) =>
message.type === "CONNECTION_PROGRESS" &&
message.stage === "presence" &&
message.status === "started"
)
);
assert.ok(
messages.some(
(message) =>
message.type === "CONNECTION_PROGRESS" &&
message.stage === "backlog" &&
message.status === "started"
)
);
assert.ok(
messages.some(
(message) =>
message.type === "CONNECTION_PROGRESS" &&
message.stage === "backlog" &&
message.status === "succeeded" &&
message.dispatchCount === 1
)
);
agent.terminate();
await broker.close();
});
test("broker sends connection errors to agents after the welcome message", async () => {
const broker = createBrokerServer({
authMode: "stub",
reportGatewayPresence: async () => {
throw new Error("presence callback unavailable");
},
requestGatewayBacklog: async () => {
throw new Error("backlog sync unavailable");
},
});
const address = await broker.listen(0);
const port = address.port;
const agent = new WebSocket(`ws://127.0.0.1:${port}/ws/agent?gatewayId=701&token=agent-token`);
const messages = collectMessages(agent);
await new Promise((resolve) => agent.once("open", resolve));
await waitFor(
() => messages.filter((message) => message.type === "CONNECTION_ERROR").length >= 2,
{ description: "agent connection error notifications" }
);
assert.equal(messages[0].type, "WELCOME");
assert.ok(
messages.some(
(message) =>
message.type === "CONNECTION_ERROR" &&
message.stage === "presence" &&
message.error === "presence callback unavailable"
)
);
assert.ok(
messages.some(
(message) =>
message.type === "CONNECTION_ERROR" &&
message.stage === "backlog" &&
message.error === "backlog sync unavailable"
)
);
agent.terminate();
await broker.close();
});
test("broker syncs queued gateway backlog on agent connect and manual sync", async () => {
const backlogRequests = [];
const broker = createBrokerServer({
@@ -285,6 +398,7 @@ test("broker syncs queued gateway backlog on agent connect and manual sync", asy
});
test("broker fans out telemetry, task, log, and presence updates to browser gateway streams", async () => {
const telemetryPayloads = [];
const broker = createBrokerServer({
authMode: "stub",
validateAgent: async () => ({ id: "701", gateway_id: "701", label: "CPH Edge 01" }),
@@ -293,7 +407,10 @@ test("broker fans out telemetry, task, log, and presence updates to browser gate
gateway_id: "701",
scopes: ["overview", "tasks", "logs", "statistics"],
}),
ingestTelemetry: async (_gatewayId, payload) => ({ gateway: { id: 701, metadata: payload.metadata || {} } }),
ingestTelemetry: async (_gatewayId, payload) => {
telemetryPayloads.push(payload);
return { gateway: { id: 701, metadata: payload.metadata || {} } };
},
ingestTaskEvent: async (_gatewayId, operationId, payload) => ({
id: operationId,
status: "IN_PROGRESS",
@@ -356,6 +473,7 @@ test("broker fans out telemetry, task, log, and presence updates to browser gate
assert.ok(browserMessages.some((message) => message.type === "stats.updated"));
assert.ok(browserMessages.some((message) => message.type === "task.updated" && message.operationId === 41));
assert.ok(browserMessages.some((message) => message.type === "log.append" && /heartbeat/.test(message.entry?.message)));
assert.equal(telemetryPayloads[0]?.broker_connection_id?.length > 0, true);
browser.terminate();
agent.terminate();
services/nginx/app/.phpunit.cache/test-results
+1 -1
View File
File diff suppressed because one or more lines are too long
services/nginx/app/classes/gateway_shelly_transport.php
+18 -7
View File
@@ -9,7 +9,10 @@ use interfaces\shelly_transport_i;
class gateway_shelly_transport implements shelly_transport_i
{
public function __construct(private readonly ?edge_gateway_manager $manager = null)
public function __construct(
private readonly ?edge_gateway_manager $manager = null,
private readonly bool $localOnly = false
)
{
}
@@ -52,7 +55,9 @@ class gateway_shelly_transport implements shelly_transport_i
$result = [];
foreach ($ids as $logicalRelayId) {
$status = $this->manager()->dispatchRelayStatus($departmentId, $logicalRelayId);
$status = $this->localOnly
? $this->manager()->dispatchRelayStatusLocalOnly($departmentId, $logicalRelayId)
: $this->manager()->dispatchRelayStatus($departmentId, $logicalRelayId);
$result[] = $this->normalizeRelayPayload($logicalRelayId, $status);
}
@@ -69,11 +74,17 @@ class gateway_shelly_transport implements shelly_transport_i
throw new Exception('Shelly gateway switch requests require an id');
}
$status = $this->manager()->dispatchRelaySwitch(
$departmentId,
$logicalRelayId,
(bool)($data['on'] ?? false)
);
$status = $this->localOnly
? $this->manager()->dispatchRelaySwitchLocalOnly(
$departmentId,
$logicalRelayId,
(bool)($data['on'] ?? false)
)
: $this->manager()->dispatchRelaySwitch(
$departmentId,
$logicalRelayId,
(bool)($data['on'] ?? false)
);
return [$this->normalizeRelayPayload($logicalRelayId, $status)];
}
services/nginx/app/classes/shelly_relay_inventory.php
+40
View File
@@ -198,6 +198,7 @@ class shelly_relay_inventory
$online = $this->normalizeBoolean($catalog_entry['cloud_online'] ?? null);
}
$status_color = $this->extractStatusColor($online);
$local_ip = $this->extractLocalIp($device, $catalog_entry);
return [
'id' => $device_id,
@@ -215,6 +216,7 @@ class shelly_relay_inventory
'code' => $device_code !== '' ? $device_code : null,
'control_type' => $control_type,
'control_name' => $control_name !== '' ? $control_name : null,
'local_ip' => $local_ip,
'status_color' => $status_color,
'online' => $online,
];
@@ -352,6 +354,44 @@ class shelly_relay_inventory
]);
}
/**
* @param array<string,mixed> $device
* @param array<string,mixed>|null $catalog_entry
*/
private function extractLocalIp(array $device, ?array $catalog_entry = null): ?string
{
$candidates = [
$device['local_ip'] ?? null,
$device['localIp'] ?? null,
$device['ip'] ?? null,
$device['_dev_info']['local_ip'] ?? null,
$device['_dev_info']['localIp'] ?? null,
$device['_dev_info']['ip'] ?? null,
$device['wifi_sta']['ip'] ?? null,
$device['wifi']['ip'] ?? null,
$device['eth']['ip'] ?? null,
$device['status']['wifi_sta']['ip'] ?? null,
$device['status']['wifi']['ip'] ?? null,
$device['status']['eth']['ip'] ?? null,
$device['status']['sta_ip'] ?? null,
$device['settings']['wifi_sta']['ip'] ?? null,
$device['settings']['wifi']['ip'] ?? null,
$device['settings']['eth']['ip'] ?? null,
$catalog_entry['local_ip'] ?? null,
$catalog_entry['localIp'] ?? null,
$catalog_entry['ip'] ?? null,
];
foreach ($candidates as $candidate) {
$normalized = trim((string)($candidate ?? ''));
if ($normalized !== '' && filter_var($normalized, FILTER_VALIDATE_IP) !== false) {
return $normalized;
}
}
return null;
}
private function buildRelayLabel(
?string $device_type,
string $cloud_name,
services/nginx/app/classes/shelly_transport_resolver.php
+15 -1
View File
@@ -17,8 +17,22 @@ class shelly_transport_resolver
) {
}
public function resolveForDepartment(int $departmentId): shelly_transport_i
public function resolveForDepartment(int $departmentId, ?string $transportOverride = null): shelly_transport_i
{
$override = strtolower(trim((string)$transportOverride));
$explicitLocalOverride = $override === 'local';
if ($override === 'local') {
$override = edge_gateway_manager::TRANSPORT_MODE_GATEWAY;
}
if ($override === edge_gateway_manager::TRANSPORT_MODE_GATEWAY) {
return $this->gatewayTransport ?? new gateway_shelly_transport($this->manager(), $explicitLocalOverride);
}
if ($override === edge_gateway_manager::TRANSPORT_MODE_CLOUD) {
return $this->cloudTransport ?? new cloud_shelly_transport();
}
$mode = $this->manager()->getDepartmentTransportMode($departmentId);
if ($mode === edge_gateway_manager::TRANSPORT_MODE_GATEWAY) {
return $this->gatewayTransport ?? new gateway_shelly_transport($this->manager());
services/nginx/app/modules/edgegateway/classes/edge_gateway_manager.php
+296 -26
View File
@@ -74,6 +74,8 @@ class edge_gateway_manager
public const SHELL_SESSION_TTL_SECONDS = 900;
public const DEVICE_FRESHNESS_DEGRADED_AFTER_SECONDS = 180;
public const CREDENTIAL_FRESH_AFTER_SECONDS = 2592000;
/** @var array<int,array<string,mixed>>|null */
private ?array $shellyRelayOptionsCache = null;
public function __construct()
{
@@ -621,8 +623,15 @@ class edge_gateway_manager
if ($existing !== []) {
$bindingObject = (new edge_gateway_relay_bindings_o())->select((int)$existing[0]['id']);
$existingDeviceId = trim((string)$bindingObject->device_id->value());
$localIp = $this->resolveRelayBindingLocalIp(
$gatewayId,
$binding,
$deviceId,
$existingDeviceId === $deviceId ? $bindingObject->local_ip->value() : null
);
$bindingObject->device_id->set($deviceId);
$bindingObject->local_ip->set($binding['local_ip'] ?? null);
$bindingObject->local_ip->set($localIp);
$bindingObject->channel->set((int)($binding['channel'] ?? 0));
$bindingObject->binding_source->set((string)($binding['binding_source'] ?? 'MANUAL'));
$bindingObject->approved_by->set($userId);
@@ -631,12 +640,13 @@ class edge_gateway_manager
continue;
}
$localIp = $this->resolveRelayBindingLocalIp($gatewayId, $binding, $deviceId);
(new edge_gateway_relay_bindings_o())->add_object([
'gateway_id' => $gatewayId,
'department_id' => $departmentId,
'relay_id' => $relayId,
'device_id' => $deviceId,
'local_ip' => $binding['local_ip'] ?? null,
'local_ip' => $localIp,
'channel' => (int)($binding['channel'] ?? 0),
'binding_source' => (string)($binding['binding_source'] ?? 'MANUAL'),
'approved_by' => $userId,
@@ -837,10 +847,33 @@ class edge_gateway_manager
* @throws Exception
*/
public function dispatchRelayStatus(int $departmentId, string $logicalRelayId): array
{
return $this->dispatchRelayStatusWithOptions($departmentId, $logicalRelayId);
}
/**
* @throws Exception
*/
public function dispatchRelayStatusLocalOnly(int $departmentId, string $logicalRelayId): array
{
return $this->dispatchRelayStatusWithOptions($departmentId, $logicalRelayId, true);
}
/**
* @throws Exception
*/
private function dispatchRelayStatusWithOptions(
int $departmentId,
string $logicalRelayId,
bool $requireFastLocalPath = false
): array
{
$binding = $this->resolveRelayBinding($departmentId, $logicalRelayId);
$gateway = $this->requireGateway((int)$binding['gateway_id']);
$resolution = $this->resolveRelayExecutionPlan($gateway, $binding, $logicalRelayId);
if ($requireFastLocalPath) {
$resolution = $this->forceLocalRelayExecutionPlan($resolution);
}
if (($resolution['execution_path'] ?? 'local') === 'cloud') {
return $this->dispatchRelayThroughCloud($departmentId, $logicalRelayId, null, $binding, $resolution);
@@ -854,6 +887,7 @@ class edge_gateway_manager
], null, [
'preferred_channel' => (string)($resolution['preferred_channel'] ?? self::DELIVERY_CHANNEL_API),
'fallback_reason' => $resolution['reason'] ?? null,
'require_fast_path' => $requireFastLocalPath,
]);
try {
@@ -875,10 +909,34 @@ class edge_gateway_manager
* @throws Exception
*/
public function dispatchRelaySwitch(int $departmentId, string $logicalRelayId, bool $on): array
{
return $this->dispatchRelaySwitchWithOptions($departmentId, $logicalRelayId, $on);
}
/**
* @throws Exception
*/
public function dispatchRelaySwitchLocalOnly(int $departmentId, string $logicalRelayId, bool $on): array
{
return $this->dispatchRelaySwitchWithOptions($departmentId, $logicalRelayId, $on, true);
}
/**
* @throws Exception
*/
private function dispatchRelaySwitchWithOptions(
int $departmentId,
string $logicalRelayId,
bool $on,
bool $requireFastLocalPath = false
): array
{
$binding = $this->resolveRelayBinding($departmentId, $logicalRelayId);
$gateway = $this->requireGateway((int)$binding['gateway_id']);
$resolution = $this->resolveRelayExecutionPlan($gateway, $binding, $logicalRelayId);
if ($requireFastLocalPath) {
$resolution = $this->forceLocalRelayExecutionPlan($resolution);
}
if (($resolution['execution_path'] ?? 'local') === 'cloud') {
return $this->dispatchRelayThroughCloud($departmentId, $logicalRelayId, $on, $binding, $resolution);
@@ -893,6 +951,7 @@ class edge_gateway_manager
], null, [
'preferred_channel' => (string)($resolution['preferred_channel'] ?? self::DELIVERY_CHANNEL_API),
'fallback_reason' => $resolution['reason'] ?? null,
'require_fast_path' => $requireFastLocalPath,
]);
try {
@@ -1870,16 +1929,43 @@ BASH;
public function recordTelemetryFromBroker(int $gatewayId, array $payload): array
{
$gateway = $this->requireGateway($gatewayId);
$now = $this->now();
$metadata = array_merge(
(array)($gateway->metadata_json->value() ?? []),
isset($payload['metadata']) && is_array($payload['metadata']) ? (array)$payload['metadata'] : []
);
$existingPresence = isset($metadata['broker_presence']) && is_array($metadata['broker_presence'])
? (array)$metadata['broker_presence']
: [];
$presenceMetadata = isset($existingPresence['metadata']) && is_array($existingPresence['metadata'])
? (array)$existingPresence['metadata']
: [];
$payloadMetadata = isset($payload['metadata']) && is_array($payload['metadata']) ? (array)$payload['metadata'] : [];
$connectionId = trim((string)($payload['broker_connection_id'] ?? $existingPresence['connection_id'] ?? ''));
$presence = [
'gateway_id' => $gatewayId,
'connected' => true,
'connection_id' => $connectionId !== '' ? $connectionId : null,
'last_seen_at' => $now,
'disconnect_reason' => null,
'last_error' => null,
'metadata' => array_merge($presenceMetadata, array_filter([
'agent_instance_id' => $payload['broker_agent_instance_id']
?? $payloadMetadata['agent_instance_id']
?? null,
], static fn(mixed $value): bool => $value !== null && $value !== '')),
];
$metadata['broker_presence'] = $presence;
$metadata['broker_connected'] = true;
$metadata['broker_connected_at'] = $now;
$metadata['broker_last_error'] = null;
$this->writeBrokerPresence($gatewayId, $presence);
$gateway->status->set((string)($payload['status'] ?? self::STATUS_ONLINE));
$gateway->hostname->set($payload['hostname'] ?? $gateway->hostname->value());
$gateway->installed_version->set($payload['installed_version'] ?? $gateway->installed_version->value());
$gateway->target_version->set($payload['target_version'] ?? $gateway->target_version->value());
$gateway->last_heartbeat_at->set($this->now());
$gateway->last_heartbeat_at->set($now);
$gateway->metadata_json->set($metadata);
if (isset($payload['inventory']) && is_array($payload['inventory'])) {
@@ -2834,6 +2920,145 @@ BASH;
}
}
private function resolveRelayBindingLocalIp(
int $gatewayId,
array $binding,
string $deviceId,
mixed $existingLocalIp = null
): ?string {
$incomingLocalIp = $this->normalizeLocalIp(
$binding['local_ip']
?? $binding['localIp']
?? $binding['ip']
?? $binding['metadata']['local_ip']
?? null
);
if ($incomingLocalIp !== null) {
return $incomingLocalIp;
}
$preservedLocalIp = $this->normalizeLocalIp($existingLocalIp);
if ($preservedLocalIp !== null) {
return $preservedLocalIp;
}
$inventoryLocalIp = $this->findInventoryLocalIp($gatewayId, $deviceId);
if ($inventoryLocalIp !== null) {
return $inventoryLocalIp;
}
return $this->findShellyCloudRelayLocalIp(
$deviceId,
trim((string)($binding['relay_id'] ?? ''))
);
}
private function findInventoryLocalIp(int $gatewayId, string $deviceId): ?string
{
$rows = (new edge_gateway_device_inventory_o())->getFieldsWhere([
'gateway_id' => $gatewayId,
'device_id' => $deviceId,
'deleted_at' => null,
], ['id']);
if ($rows === []) {
return null;
}
$inventoryObject = (new edge_gateway_device_inventory_o())->select((int)$rows[0]['id']);
return $this->normalizeLocalIp($inventoryObject->local_ip->value());
}
private function findShellyCloudRelayLocalIp(string $deviceId, string $relayId): ?string
{
foreach ($this->listShellyRelayOptionsForLocalIpLookup() as $option) {
$optionDeviceId = trim((string)($option['device_id'] ?? ''));
$optionRelayId = trim((string)($option['id'] ?? ''));
if ($optionDeviceId !== $deviceId && $optionRelayId !== $relayId && $optionRelayId !== $deviceId) {
continue;
}
$localIp = $this->normalizeLocalIp(
$option['local_ip']
?? $option['localIp']
?? $option['ip']
?? null
);
if ($localIp !== null) {
return $localIp;
}
}
return null;
}
/**
* @return array<int,array<string,mixed>>
*/
private function listShellyRelayOptionsForLocalIpLookup(): array
{
if (is_array($this->shellyRelayOptionsCache)) {
return $this->shellyRelayOptionsCache;
}
try {
$this->shellyRelayOptionsCache = (new shelly_relay_inventory())->listRelayOptions();
} catch (\Throwable) {
$this->shellyRelayOptionsCache = [];
}
return $this->shellyRelayOptionsCache;
}
private function normalizeLocalIp(mixed $value): ?string
{
$localIp = trim((string)($value ?? ''));
if ($localIp === '') {
return null;
}
return filter_var($localIp, FILTER_VALIDATE_IP) !== false ? $localIp : null;
}
/**
* @param array<string,mixed> $device
*/
private function extractDeviceLocalIp(array $device): ?string
{
return $this->normalizeLocalIp(
$device['local_ip']
?? $device['localIp']
?? $device['ip']
?? $device['metadata']['local_ip']
?? $device['metadata']['localIp']
?? $device['metadata']['ip']
?? null
);
}
private function backfillRelayBindingLocalIpFromInventory(int $gatewayId, string $deviceId, ?string $localIp): void
{
$localIp = $this->normalizeLocalIp($localIp);
if ($localIp === null) {
return;
}
$rows = (new edge_gateway_relay_bindings_o())->getFieldsWhere([
'gateway_id' => $gatewayId,
'device_id' => $deviceId,
'deleted_at' => null,
], ['id']);
foreach ($rows as $row) {
$bindingObject = (new edge_gateway_relay_bindings_o())->select((int)$row['id']);
if ($this->normalizeLocalIp($bindingObject->local_ip->value()) !== null) {
continue;
}
$bindingObject->local_ip->set($localIp);
}
}
/**
* @param array<int,array<string,mixed>> $inventory
*/
@@ -2845,6 +3070,7 @@ BASH;
continue;
}
$localIp = $this->extractDeviceLocalIp($device);
$rows = (new edge_gateway_device_inventory_o())->getFieldsWhere([
'gateway_id' => $gatewayId,
'device_id' => $deviceId,
@@ -2855,7 +3081,7 @@ BASH;
(new edge_gateway_device_inventory_o())->add_object([
'gateway_id' => $gatewayId,
'device_id' => $deviceId,
'local_ip' => $device['local_ip'] ?? $device['ip'] ?? null,
'local_ip' => $localIp,
'model' => $device['model'] ?? null,
'channel_count' => (int)($device['channel_count'] ?? $device['channels'] ?? 1),
'capabilities_json' => (array)($device['capabilities'] ?? []),
@@ -2863,17 +3089,21 @@ BASH;
'last_seen_at' => $this->now(),
'metadata_json' => (array)($device['metadata'] ?? []),
]);
$this->backfillRelayBindingLocalIpFromInventory($gatewayId, $deviceId, $localIp);
continue;
}
$inventoryObject = (new edge_gateway_device_inventory_o())->select((int)$rows[0]['id']);
$inventoryObject->local_ip->set($device['local_ip'] ?? $device['ip'] ?? null);
if ($localIp !== null) {
$inventoryObject->local_ip->set($localIp);
}
$inventoryObject->model->set($device['model'] ?? null);
$inventoryObject->channel_count->set((int)($device['channel_count'] ?? $device['channels'] ?? 1));
$inventoryObject->capabilities_json->set((array)($device['capabilities'] ?? []));
$inventoryObject->online->set((bool)($device['online'] ?? true));
$inventoryObject->last_seen_at->set($this->now());
$inventoryObject->metadata_json->set((array)($device['metadata'] ?? []));
$this->backfillRelayBindingLocalIpFromInventory($gatewayId, $deviceId, $localIp);
}
}
@@ -2931,7 +3161,7 @@ BASH;
$gateway = $this->requireGateway($gatewayId);
$normalizedStatus = trim(strtolower($status));
$connected = $normalizedStatus === 'connected';
$presence = array_filter([
$presence = [
'gateway_id' => $gatewayId,
'connected' => $connected,
'connection_id' => $connectionId,
@@ -2939,7 +3169,7 @@ BASH;
'disconnect_reason' => $connected ? null : $reason,
'last_error' => !$connected && $reason !== null && trim($reason) !== '' ? trim($reason) : null,
'metadata' => $metadata,
], static fn(mixed $value): bool => $value !== null);
];
$this->writeBrokerPresence($gatewayId, $presence);
@@ -3142,7 +3372,7 @@ BASH;
}
$presence = $this->readBrokerPresence($gatewayId);
return !empty($presence['connected']) ? self::DELIVERY_CHANNEL_BROKER : self::DELIVERY_CHANNEL_API;
return self::isBrokerPresenceConnected($presence) ? self::DELIVERY_CHANNEL_BROKER : self::DELIVERY_CHANNEL_API;
}
public function validateBrokerSharedSecret(?string $secret): bool
@@ -3301,33 +3531,66 @@ BASH;
]);
}
private function forceLocalRelayExecutionPlan(array $resolution): array
{
$wasCloud = (string)($resolution['execution_path'] ?? 'local') === 'cloud';
$recoveryActions = array_values(array_filter(array_unique(array_merge(
(array)($resolution['recovery_actions'] ?? []),
['retry_local_command']
))));
return array_merge($resolution, [
'execution_path' => 'local',
'preferred_channel' => self::DELIVERY_CHANNEL_BROKER,
'fallback_mode' => self::RELAY_FALLBACK_LOCAL_ONLY,
'reason' => $wasCloud ? 'local_transport_override' : ($resolution['reason'] ?? null),
'recommended_action' => $resolution['recommended_action'] ?? 'retry_local_command',
'recovery_actions' => $recoveryActions,
]);
}
/**
* @throws Exception
*/
private function dispatchGatewayCommand(edge_gateways_o $gateway, edge_gateway_command_jobs_o $job): array
{
$delivery = (array)($job->delivery_json->value() ?? []);
$preferredChannel = (string)($delivery['preferred_channel'] ?? self::DELIVERY_CHANNEL_API);
$requireFastPath = !empty($delivery['require_fast_path']);
$effectiveStatus = self::resolveGatewayStatus(
$gateway->status->value() === null ? null : (string)$gateway->status->value(),
$gateway->last_heartbeat_at->value() === null ? null : (string)$gateway->last_heartbeat_at->value()
);
if (!in_array($effectiveStatus, [self::STATUS_ONLINE, self::STATUS_DEGRADED], true)) {
if (!$requireFastPath && !in_array($effectiveStatus, [self::STATUS_ONLINE, self::STATUS_DEGRADED], true)) {
throw new Exception('Gateway agent is offline');
}
$delivery = (array)($job->delivery_json->value() ?? []);
$preferredChannel = (string)($delivery['preferred_channel'] ?? self::DELIVERY_CHANNEL_API);
if ($preferredChannel === self::DELIVERY_CHANNEL_BROKER && $this->buildBrokerInternalUrl() !== null) {
try {
$this->markCommandJobDispatching($job, self::DELIVERY_CHANNEL_BROKER);
$payload = $this->dispatchBrokerCommand($gateway, $job);
$this->finalizeCommandJob($job, true, $payload, null, $gateway);
return $payload;
} catch (Exception $exception) {
$this->markCommandDeliveryFailure($job, self::DELIVERY_CHANNEL_BROKER, $exception->getMessage(), 'broker_dispatch_failed');
if ($preferredChannel === self::DELIVERY_CHANNEL_BROKER) {
if ($this->buildBrokerInternalUrl() === null) {
$this->markCommandDeliveryFailure($job, self::DELIVERY_CHANNEL_BROKER, 'Edge broker is not configured', 'broker_not_configured');
if ($requireFastPath) {
throw new Exception('Edge broker is not configured');
}
} else {
try {
$this->markCommandJobDispatching($job, self::DELIVERY_CHANNEL_BROKER);
$payload = $this->dispatchBrokerCommand($gateway, $job);
$this->finalizeCommandJob($job, true, $payload, null, $gateway);
return $payload;
} catch (Exception $exception) {
$this->markCommandDeliveryFailure($job, self::DELIVERY_CHANNEL_BROKER, $exception->getMessage(), 'broker_dispatch_failed');
if ($requireFastPath) {
throw $exception;
}
}
}
}
if ($requireFastPath) {
throw new Exception('Edge broker fast path is unavailable');
}
return $this->waitForCommandResult((int)$job->id);
}
@@ -3386,11 +3649,6 @@ BASH;
throw new Exception('Edge broker is not configured');
}
$presence = $this->readBrokerPresence((int)$gateway->id);
if (empty($presence['connected'])) {
throw new Exception('Edge broker fast path is unavailable');
}
$result = $this->httpJsonRequest(
$brokerUrl . '/api/gateways/' . (int)$gateway->id . '/commands',
[
@@ -4101,9 +4359,9 @@ BASH;
$brokerPresence = isset($metadata['broker_presence']) && is_array($metadata['broker_presence'])
? (array)$metadata['broker_presence']
: [];
$brokerConnected = !empty($brokerPresence['connected']);
$brokerLastSeenAt = isset($brokerPresence['last_seen_at']) ? (string)$brokerPresence['last_seen_at'] : null;
$brokerAgeSeconds = self::heartbeatAgeSeconds($brokerLastSeenAt, $now);
$brokerConnected = self::isBrokerPresenceConnected($brokerPresence, $now);
$brokerHealthy = $brokerConnected
&& $brokerAgeSeconds !== null
&& $brokerAgeSeconds < self::BROKER_CONNECTIVITY_DEGRADED_AFTER_SECONDS;
@@ -4132,6 +4390,18 @@ BASH;
];
}
private static function isBrokerPresenceConnected(array $presence, ?int $now = null): bool
{
if (empty($presence['connected'])) {
return false;
}
$lastSeenAt = isset($presence['last_seen_at']) ? (string)$presence['last_seen_at'] : null;
$ageSeconds = self::heartbeatAgeSeconds($lastSeenAt, $now);
return $ageSeconds !== null && $ageSeconds < self::BROKER_PRESENCE_TTL_SECONDS;
}
private static function buildRelayHealth(array $gateway, string $effectiveStatus, ?int $now = null): array
{
$bindings = is_array($gateway['bindings'] ?? null) ? (array)$gateway['bindings'] : [];
services/nginx/app/modules/selfserve/traits/selfserve_lane_port_controller_t.php
+4 -4
View File
@@ -32,8 +32,8 @@ trait selfserve_lane_port_controller_t
if ($this->getLaneStatus()->equals(selfserve_lane_status::FAULT)) throw new \Exception("Cannot open port on FAULT lane");
// Get the relay ID based on the port
$relay_id = match ($port) {
selfserve_lane_port::ENTRANCE => $this->department_lane->relay_out_id->value(),
selfserve_lane_port::EXIT => $this->department_lane->relay_in_id->value(),
selfserve_lane_port::ENTRANCE => $this->department_lane->relay_in_id->value(),
selfserve_lane_port::EXIT => $this->department_lane->relay_out_id->value(),
default => throw new \Exception("Invalid port specified, must be ENTRANCE or EXIT"),
};
// Make sure relay ID is valid
@@ -63,8 +63,8 @@ trait selfserve_lane_port_controller_t
{
// Queue the relay switch command to be executed asynchronously
$relay_id = match ($port) {
selfserve_lane_port::ENTRANCE => $this->department_lane->relay_out_id->value(),
selfserve_lane_port::EXIT => $this->department_lane->relay_in_id->value(),
selfserve_lane_port::ENTRANCE => $this->department_lane->relay_in_id->value(),
selfserve_lane_port::EXIT => $this->department_lane->relay_out_id->value(),
default => throw new \Exception("Invalid port specified, must be ENTRANCE or EXIT"),
};
// If the relay ID indicates a demo port, skip the Shelly switch call but keep the logging and state changes
services/nginx/app/modules/selfserve/traits/selfserve_lane_relay_controller_t.php
+29 -2
View File
@@ -21,6 +21,27 @@ trait selfserve_lane_relay_controller_t
private const SHELLY_DEFAULT_CHANNEL = 0;
private const DEMO_RELAY_ID_PREFIX = 'demo-';
protected ?string $shelly_transport_override = null;
/**
* @throws \Exception
*/
public function setShellyTransportOverride(?string $transport): self
{
$normalized = strtolower(trim((string)$transport));
if ($normalized === '') {
$this->shelly_transport_override = null;
return $this;
}
if (!in_array($normalized, ['cloud', 'gateway', 'local'], true)) {
throw new \Exception('Invalid Shelly transport override. Expected local or cloud.');
}
$this->shelly_transport_override = $normalized;
return $this;
}
/**
* Get current MACHINE relay status from Shelly.
* @return array{relay_id: string, online: bool, on: bool}
@@ -622,7 +643,10 @@ trait selfserve_lane_relay_controller_t
private function getLaneShellyStatusSnapshotCacheKey(): string
{
return self::SHELLY_STATUS_SNAPSHOT_KEY_PREFIX . (int)$this->id;
$transport = strtolower(trim((string)$this->shelly_transport_override));
$safeTransport = $transport !== '' ? (preg_replace('/[^a-z0-9_-]+/', '_', $transport) ?: $transport) : '';
$transportSuffix = $safeTransport !== '' ? '_' . $safeTransport : '';
return self::SHELLY_STATUS_SNAPSHOT_KEY_PREFIX . (int)$this->id . $transportSuffix;
}
/**
@@ -695,7 +719,10 @@ trait selfserve_lane_relay_controller_t
protected function createShellyTransport(): shelly_transport_i
{
return (new shelly_transport_resolver())->resolveForDepartment($this->resolveShellyTransportDepartmentId());
return (new shelly_transport_resolver())->resolveForDepartment(
$this->resolveShellyTransportDepartmentId(),
$this->shelly_transport_override
);
}
/**
services/nginx/app/resources/edge-gateway-agent/Dockerfile.auto-updater
+2 -1
View File
@@ -4,7 +4,8 @@ FROM ${BASE_IMAGE}
RUN set -eux; \
apt-get update; \
apt-get install -y --no-install-recommends bash ca-certificates curl docker.io docker-compose; \
rm -rf /var/lib/apt/lists/*
rm -rf /var/lib/apt/lists/*; \
php -r 'foreach (["curl", "sqlite3"] as $extension) { if (!extension_loaded($extension)) { fwrite(STDERR, "Missing PHP extension: {$extension}\n"); exit(1); } }'
COPY auto-updater.php /usr/local/bin/auto-updater.php
services/nginx/app/resources/edge-gateway-agent/Dockerfile.edge-agent
+3
View File
@@ -1,6 +1,9 @@
ARG BASE_IMAGE=php:8.2-cli-bookworm
FROM ${BASE_IMAGE}
RUN set -eux; \
php -r 'foreach (["curl", "sqlite3"] as $extension) { if (!extension_loaded($extension)) { fwrite(STDERR, "Missing PHP extension: {$extension}\n"); exit(1); } }'
WORKDIR /opt/truckwash-edge-agent
COPY agent.php /opt/truckwash-edge-agent/agent.php
services/nginx/app/resources/edge-gateway-agent/Dockerfile.lan-worker
+3
View File
@@ -1,6 +1,9 @@
ARG BASE_IMAGE=php:8.2-cli-bookworm
FROM ${BASE_IMAGE}
RUN set -eux; \
php -r 'foreach (["curl", "sqlite3"] as $extension) { if (!extension_loaded($extension)) { fwrite(STDERR, "Missing PHP extension: {$extension}\n"); exit(1); } }'
WORKDIR /opt/truckwash-edge-agent
COPY lan-worker.php /opt/truckwash-edge-agent/lan-worker.php
services/nginx/app/resources/edge-gateway-agent/truckwash-edge-agent.service
+9 -7
View File
@@ -1,14 +1,16 @@
[Unit]
Description=TruckWash Edge Agent
After=network.target
Description=TruckWash Edge Agent Compatibility Unit
After=network-online.target docker.service
Wants=network-online.target docker.service
[Service]
Type=simple
Type=oneshot
RemainAfterExit=yes
WorkingDirectory=/opt/truckwash-edge-agent
ExecStart=/usr/bin/php /opt/truckwash-edge-agent/agent.php --config /opt/truckwash-edge-agent/config.json
Restart=always
RestartSec=5
User=root
ExecStart=/opt/truckwash-edge-agent/gateway-launcher.sh up
ExecReload=/opt/truckwash-edge-agent/gateway-launcher.sh reconcile
ExecStop=/opt/truckwash-edge-agent/gateway-launcher.sh down
TimeoutStartSec=900
[Install]
WantedBy=multi-user.target
services/nginx/app/routes/moduleSelfServeRoute.php
+57
View File
@@ -441,12 +441,14 @@ class moduleSelfServeRoute
$lane = $selfserve->lane($lane_id);
try {
$this->applyShellyTransportOverride($lane);
$lane->open($gate);
$response->success([
'lane_id' => $lane_id,
'gate' => $gate->name,
'opened' => true,
'state' => $lane->getLaneState()->name,
'transport' => $this->requestedShellyTransportOverride(),
]);
} catch (\Exception $e) {
error_log('Failed to open self-serve lane gate ' . $gate->name . ' for lane ' . $lane_id . ': ' . $e->getMessage());
@@ -468,6 +470,7 @@ class moduleSelfServeRoute
$lane = $selfserve->lane($lane_id);
try {
$this->applyShellyTransportOverride($lane);
$status = $lane->getMachineProgramPickerRelayStatus();
$response->success([
'lane_id' => $lane_id,
@@ -475,6 +478,7 @@ class moduleSelfServeRoute
'relay_id' => (string)$status['relay_id'],
'online' => (bool)$status['online'],
'on' => (bool)$status['on'],
'transport' => $this->requestedShellyTransportOverride(),
]);
} catch (\Exception $e) {
$response->error('Failed to get MACHINE_PROGRAM_PICKER relay status: ' . $e->getMessage(), 400);
@@ -501,6 +505,7 @@ class moduleSelfServeRoute
$lane = $selfserve->lane($lane_id);
try {
$this->applyShellyTransportOverride($lane);
$lane->setMachineProgramPickerRelayStatus((bool)$on);
$status = $lane->getMachineProgramPickerRelayStatus();
$response->success([
@@ -510,6 +515,7 @@ class moduleSelfServeRoute
'relay_id' => (string)$status['relay_id'],
'online' => (bool)$status['online'],
'on' => (bool)$status['on'],
'transport' => $this->requestedShellyTransportOverride(),
]);
} catch (\Exception $e) {
$response->error('Failed to set MACHINE_PROGRAM_PICKER relay status: ' . $e->getMessage(), 400);
@@ -530,6 +536,7 @@ class moduleSelfServeRoute
$lane = $selfserve->lane($lane_id);
try {
$this->applyShellyTransportOverride($lane);
$status = $lane->getMachineCleanerRelayStatus();
$response->success([
'lane_id' => $lane_id,
@@ -537,6 +544,7 @@ class moduleSelfServeRoute
'relay_id' => (string)$status['relay_id'],
'online' => (bool)$status['online'],
'on' => (bool)$status['on'],
'transport' => $this->requestedShellyTransportOverride(),
]);
} catch (\Exception $e) {
$response->error('Failed to get MACHINE_CLEANER relay status: ' . $e->getMessage(), 400);
@@ -563,6 +571,7 @@ class moduleSelfServeRoute
$lane = $selfserve->lane($lane_id);
try {
$this->applyShellyTransportOverride($lane);
$lane->setMachineCleanerRelayStatus((bool)$on);
$status = $lane->getMachineCleanerRelayStatus();
$response->success([
@@ -572,6 +581,7 @@ class moduleSelfServeRoute
'relay_id' => (string)$status['relay_id'],
'online' => (bool)$status['online'],
'on' => (bool)$status['on'],
'transport' => $this->requestedShellyTransportOverride(),
]);
} catch (\Exception $e) {
$response->error('Failed to set MACHINE_CLEANER relay status: ' . $e->getMessage(), 400);
@@ -592,6 +602,7 @@ class moduleSelfServeRoute
$lane = $selfserve->lane($lane_id);
try {
$this->applyShellyTransportOverride($lane);
$status = $lane->getMachineRelayStatus();
$response->success([
'lane_id' => $lane_id,
@@ -599,6 +610,7 @@ class moduleSelfServeRoute
'relay_id' => (string)$status['relay_id'],
'online' => (bool)$status['online'],
'on' => (bool)$status['on'],
'transport' => $this->requestedShellyTransportOverride(),
]);
} catch (\Exception $e) {
$response->error('Failed to get MACHINE relay status: ' . $e->getMessage(), 400);
@@ -625,6 +637,7 @@ class moduleSelfServeRoute
$lane = $selfserve->lane($lane_id);
try {
$this->applyShellyTransportOverride($lane);
$lane->setMachineRelayStatus((bool)$on);
// Keep lane cache state aligned with the latest explicit relay action
try {
@@ -651,6 +664,7 @@ class moduleSelfServeRoute
'relay_id' => (string)$status['relay_id'],
'online' => (bool)$status['online'],
'on' => (bool)$status['on'],
'transport' => $this->requestedShellyTransportOverride(),
]);
} catch (\Exception $e) {
$response->error('Failed to set MACHINE relay status: ' . $e->getMessage(), 400);
@@ -984,4 +998,47 @@ class moduleSelfServeRoute
'modules_selfserve_lane_force_machine_disable' => 'Force disable MACHINE relay but keep lane as in-wash (superusers only)'
]);
}
/**
* @throws \Exception
*/
private function applyShellyTransportOverride(object $lane): void
{
$transport = $this->requestedShellyTransportOverride();
if ($transport === null) {
return;
}
if (!method_exists($lane, 'setShellyTransportOverride')) {
throw new \Exception('This lane does not support Shelly transport overrides.');
}
$lane->setShellyTransportOverride($transport);
}
private function requestedShellyTransportOverride(): ?string
{
$transport = null;
foreach (['transport', 'test_transport', 'transport_mode'] as $parameter) {
if (self::isParametersSet([$parameter])) {
$transport = self::getParameter($parameter);
break;
}
}
$normalized = strtolower(trim((string)$transport));
if ($normalized === '') {
return null;
}
if ($normalized === 'gateway') {
return 'local';
}
if (!in_array($normalized, ['local', 'cloud'], true)) {
throw new \InvalidArgumentException('Invalid transport. Expected local or cloud.');
}
return $normalized;
}
}
services/nginx/app/tests/Unit/Selfserve/EdgeGatewayManagerCommandQueueTest.php
+16
View File
@@ -11,8 +11,18 @@ it('keeps relay dispatch and discovery queueing on the edge gateway manager', fu
expect($managerSource)->not->toBeFalse();
expect($managerSource)->toContain('public function queueDiscovery');
expect($managerSource)->toContain('public function dispatchRelayStatus');
expect($managerSource)->toContain('public function dispatchRelayStatusLocalOnly');
expect($managerSource)->toContain('public function dispatchRelaySwitch');
expect($managerSource)->toContain('public function dispatchRelaySwitchLocalOnly');
expect($managerSource)->toContain('private function createCommandJob');
expect($managerSource)->toContain("'require_fast_path' => \$requireFastLocalPath");
expect($managerSource)->toContain('local_transport_override');
expect($managerSource)->toContain('private function resolveRelayBindingLocalIp');
expect($managerSource)->toContain('private function findShellyCloudRelayLocalIp');
expect($managerSource)->toContain('private function backfillRelayBindingLocalIpFromInventory');
expect($managerSource)->toContain('$bindingObject->local_ip->set($localIp);');
expect($managerSource)->toContain("'local_ip' => \$localIp");
expect($managerSource)->toContain('(new shelly_relay_inventory())->listRelayOptions()');
expect($managerSource)->toContain('public function buildUpdateOperationRequest');
expect($managerSource)->toContain('public function rotateGatewayCredentials');
expect($operationServiceSource)->toContain('public function queueOperation');
@@ -38,11 +48,17 @@ it('loads relay command helpers on the manager and gateway operations on the ded
expect($reflection->hasMethod('pollCommand'))->toBeTrue();
expect($reflection->hasMethod('submitCommandResult'))->toBeTrue();
expect($reflection->hasMethod('dispatchRelayStatus'))->toBeTrue();
expect($reflection->hasMethod('dispatchRelayStatusLocalOnly'))->toBeTrue();
expect($reflection->hasMethod('dispatchRelaySwitch'))->toBeTrue();
expect($reflection->hasMethod('dispatchRelaySwitchLocalOnly'))->toBeTrue();
expect($reflection->hasMethod('claimNextCommandJob'))->toBeTrue();
expect($reflection->getMethod('claimNextCommandJob')->isPrivate())->toBeTrue();
expect($reflection->hasMethod('syncDeviceInventory'))->toBeTrue();
expect($reflection->getMethod('syncDeviceInventory')->isPrivate())->toBeTrue();
expect($reflection->hasMethod('resolveRelayBindingLocalIp'))->toBeTrue();
expect($reflection->getMethod('resolveRelayBindingLocalIp')->isPrivate())->toBeTrue();
expect($reflection->hasMethod('backfillRelayBindingLocalIpFromInventory'))->toBeTrue();
expect($reflection->getMethod('backfillRelayBindingLocalIpFromInventory')->isPrivate())->toBeTrue();
expect($reflection->hasMethod('syncGatewayInventory'))->toBeTrue();
expect($operationServiceReflection->hasMethod('queueOperation'))->toBeTrue();
expect($operationServiceReflection->hasMethod('cancelOperation'))->toBeTrue();
services/nginx/app/tests/Unit/Selfserve/EdgeGatewayManagerHeartbeatStatusTest.php
+41
View File
@@ -72,6 +72,38 @@ it('marks ready discovery as stale when the gateway heartbeat has expired', func
expect($gateway['discovery_status'])->toBe('STALE');
});
it('only trusts broker presence while the broker heartbeat is fresh', function (): void {
$recent = edge_gateway_manager::deriveGatewayRuntimeState([
'status' => edge_gateway_manager::STATUS_ONLINE,
'last_heartbeat_at' => '2026-04-08 10:04:50',
'metadata' => [
'broker_presence' => [
'connected' => true,
'last_seen_at' => '2026-04-08 10:04:30',
],
],
], edge_gateway_heartbeat_test_now('2026-04-08 10:05:00'));
expect($recent['channel_status']['broker']['connected'])->toBeTrue();
expect($recent['channel_status']['broker']['state'])->toBe(edge_gateway_manager::STATUS_ONLINE);
expect($recent['channel_status']['command']['preferred'])->toBe(edge_gateway_manager::DELIVERY_CHANNEL_BROKER);
$stale = edge_gateway_manager::deriveGatewayRuntimeState([
'status' => edge_gateway_manager::STATUS_ONLINE,
'last_heartbeat_at' => '2026-04-08 10:04:50',
'metadata' => [
'broker_presence' => [
'connected' => true,
'last_seen_at' => '2026-04-08 10:03:00',
],
],
], edge_gateway_heartbeat_test_now('2026-04-08 10:05:00'));
expect($stale['channel_status']['broker']['connected'])->toBeFalse();
expect($stale['channel_status']['broker']['state'])->toBe(edge_gateway_manager::STATUS_OFFLINE);
expect($stale['channel_status']['command']['preferred'])->toBe(edge_gateway_manager::DELIVERY_CHANNEL_API);
});
it('merges incoming heartbeat metadata with existing gateway metadata', function (): void {
$source = file_get_contents(app_path('classes/edge_gateway_manager.php'));
@@ -79,6 +111,15 @@ it('merges incoming heartbeat metadata with existing gateway metadata', function
expect($source)->toContain("\$gateway->metadata_json->set(array_merge(\$existingMetadata, (array)(\$payload['metadata'] ?? [])));");
});
it('refreshes broker presence from broker telemetry heartbeats', function (): void {
$source = file_get_contents(app_path('classes/edge_gateway_manager.php'));
expect($source)->toContain('public function recordTelemetryFromBroker');
expect($source)->toContain("'broker_connection_id'");
expect($source)->toContain('$this->writeBrokerPresence($gatewayId, $presence);');
expect($source)->toContain('private static function isBrokerPresenceConnected');
});
it('derives relay fallback and transport health details without shell or update runtime state', function (): void {
$gateway = edge_gateway_manager::deriveGatewayRuntimeState([
'status' => edge_gateway_manager::STATUS_ONLINE,
services/nginx/app/tests/Unit/Selfserve/EdgeGatewayUpdateLifecycleTest.php
+10 -3
View File
@@ -50,7 +50,11 @@ it('builds the installer around the compose stack artifacts and management polli
expect($managerSource)->toContain('journalctl -u truckwash-edge-gateway-stack.service -n 60 --no-pager || true');
expect($managerSource)->not->toContain('agent.mjs');
expect($managerSource)->toContain("'brokerUrl' => \$this->buildBrokerPublicUrl()");
expect($serviceSource)->toContain('ExecStart=/usr/bin/php /opt/truckwash-edge-agent/agent.php --config /opt/truckwash-edge-agent/config.json');
expect($serviceSource)->toContain('Description=TruckWash Edge Agent Compatibility Unit');
expect($serviceSource)->toContain('ExecStart=/opt/truckwash-edge-agent/gateway-launcher.sh up');
expect($serviceSource)->toContain('ExecReload=/opt/truckwash-edge-agent/gateway-launcher.sh reconcile');
expect($serviceSource)->toContain('ExecStop=/opt/truckwash-edge-agent/gateway-launcher.sh down');
expect($serviceSource)->not->toContain('/usr/bin/php /opt/truckwash-edge-agent/agent.php');
expect($stackServiceSource)->toContain('ExecStart=/opt/truckwash-edge-agent/gateway-launcher.sh up');
expect($stackServiceSource)->toContain('TimeoutStartSec=900');
expect($launcherSource)->toContain('STACK_HEALTHCHECK_TIMEOUT_SECONDS="${STACK_HEALTHCHECK_TIMEOUT_SECONDS:-300}"');
@@ -92,14 +96,17 @@ it('builds the installer around the compose stack artifacts and management polli
expect($agentSource)->toContain("'last_transport_error'");
expect($agentSource)->toContain('private function recordTransportFailure(string $context, Throwable $throwable): void');
expect($edgeDockerfileSource)->toContain('FROM ${BASE_IMAGE}');
expect($edgeDockerfileSource)->toContain('extension_loaded($extension)');
expect($edgeDockerfileSource)->toContain('Missing PHP extension: {$extension}');
expect($edgeDockerfileSource)->toContain('COPY agent.php /opt/truckwash-edge-agent/agent.php');
expect($edgeDockerfileSource)->not->toContain('docker-php-ext-install');
expect($workerDockerfileSource)->toContain('FROM ${BASE_IMAGE}');
expect($workerDockerfileSource)->toContain('extension_loaded($extension)');
expect($workerDockerfileSource)->toContain('Missing PHP extension: {$extension}');
expect($workerDockerfileSource)->toContain('COPY lan-worker.php /opt/truckwash-edge-agent/lan-worker.php');
expect($workerDockerfileSource)->not->toContain('docker-php-ext-install');
expect($autoUpdaterSource)->toContain("'/bin/bash ' . escapeshellarg(\$launcherPath) . ' reconcile 2>&1'");
expect($autoUpdaterDockerfileSource)->toContain('COPY auto-updater.php /usr/local/bin/auto-updater.php');
expect($autoUpdaterDockerfileSource)->toContain('apt-get install -y --no-install-recommends bash ca-certificates curl docker.io docker-compose;');
expect($autoUpdaterDockerfileSource)->toContain('extension_loaded($extension)');
expect($serviceSource)->not->toContain('node /opt/truckwash-edge-agent/agent.mjs');
});
services/nginx/app/tests/Unit/Selfserve/GatewayShellyTransportTest.php
+57
View File
@@ -12,6 +12,10 @@ class GatewayShellyTransportManagerFake extends edge_gateway_manager
public array $statusCalls = [];
/** @var array<int,array<string,mixed>> */
public array $switchCalls = [];
/** @var array<int,array<string,mixed>> */
public array $localOnlyStatusCalls = [];
/** @var array<int,array<string,mixed>> */
public array $localOnlySwitchCalls = [];
public ?Exception $statusException = null;
public ?Exception $switchException = null;
@@ -55,6 +59,35 @@ class GatewayShellyTransportManagerFake extends edge_gateway_manager
'raw' => ['source' => 'switch'],
];
}
public function dispatchRelayStatusLocalOnly(int $departmentId, string $logicalRelayId): array
{
$this->localOnlyStatusCalls[] = [
'department_id' => $departmentId,
'relay_id' => $logicalRelayId,
];
return [
'online' => true,
'on' => true,
'raw' => ['source' => 'local-status'],
];
}
public function dispatchRelaySwitchLocalOnly(int $departmentId, string $logicalRelayId, bool $on): array
{
$this->localOnlySwitchCalls[] = [
'department_id' => $departmentId,
'relay_id' => $logicalRelayId,
'on' => $on,
];
return [
'online' => true,
'on' => $on,
'raw' => ['source' => 'local-switch'],
];
}
}
it('maps gateway relay status responses into the Shelly cloud payload shape', function (): void {
@@ -92,6 +125,30 @@ it('maps gateway relay switch responses into the Shelly cloud payload shape', fu
expect($result[0]['status']['switch:0']['output'])->toBeFalse();
});
it('uses local-only gateway dispatch for explicit local transport overrides', function (): void {
$manager = new GatewayShellyTransportManagerFake();
$transport = new gateway_shelly_transport($manager, true);
$status = $transport->sendPostRequest('/v2/devices/api/get', [
'ids' => ['relay-entry'],
], 17);
$switch = $transport->sendPostRequest('/v2/devices/api/set/switch', [
'id' => 'relay-entry',
'on' => true,
], 17);
expect($manager->localOnlyStatusCalls)->toBe([
['department_id' => 17, 'relay_id' => 'relay-entry'],
]);
expect($manager->localOnlySwitchCalls)->toBe([
['department_id' => 17, 'relay_id' => 'relay-entry', 'on' => true],
]);
expect($manager->statusCalls)->toBe([]);
expect($manager->switchCalls)->toBe([]);
expect($status[0]['raw']['source'])->toBe('local-status');
expect($switch[0]['raw']['source'])->toBe('local-switch');
});
it('requires a valid department id for gateway transport requests', function (): void {
$transport = new gateway_shelly_transport(new GatewayShellyTransportManagerFake());
services/nginx/app/tests/Unit/Selfserve/SelfserveLanePortControllerTest.php
+4 -4
View File
@@ -97,7 +97,7 @@ class SelfserveLanePortControllerHarness
public function logLaneAction(...$args): void {}
}
it('opens exit port by switching relay_in_id on', function (): void {
it('opens exit port by switching relay_out_id on', function (): void {
$lane = new SelfserveLanePortControllerHarness(
relayInId: 'relay-in-123',
relayOutId: 'relay-out-123'
@@ -108,13 +108,13 @@ it('opens exit port by switching relay_in_id on', function (): void {
expect($result)->toBeTrue();
expect($lane->switchFake->switchCalls)->toBe([
[
'id' => 'relay-in-123',
'id' => 'relay-out-123',
'on' => true,
],
]);
});
it('opens entrance port by switching relay_out_id on', function (): void {
it('opens entrance port by switching relay_in_id on', function (): void {
$lane = new SelfserveLanePortControllerHarness(
relayInId: 'relay-in-123',
relayOutId: 'relay-out-123'
@@ -125,7 +125,7 @@ it('opens entrance port by switching relay_out_id on', function (): void {
expect($result)->toBeTrue();
expect($lane->switchFake->switchCalls)->toBe([
[
'id' => 'relay-out-123',
'id' => 'relay-in-123',
'on' => true,
],
]);
services/nginx/app/tests/Unit/Selfserve/SelfserveRouteWiringTest.php
+2
View File
@@ -84,6 +84,8 @@ it('wires machine relay status get and set endpoints', function (): void {
expect($moduleSelfServeRoute)->toContain('getMachineCleanerRelayStatus');
expect($moduleSelfServeRoute)->toContain('setMachineCleanerRelayStatus');
expect($moduleSelfServeRoute)->toContain('setMachineCleanerRelayStatusHard(true)');
expect($moduleSelfServeRoute)->toContain('applyShellyTransportOverride($lane)');
expect($moduleSelfServeRoute)->toContain("'transport' => \$this->requestedShellyTransportOverride()");
});
it('wires allowed services route through machine relay visibility sync', function (): void {
services/nginx/app/tests/Unit/Selfserve/ShellyRelayInventoryTest.php
+13
View File
@@ -19,6 +19,9 @@ it('normalizes owned Shelly devices into relay select options', function (): voi
'online' => 1,
],
'name' => 'Entry Gate',
'wifi_sta' => [
'ip' => '10.32.0.11',
],
'status' => [
'switch:0' => [
'output' => false,
@@ -33,6 +36,11 @@ it('normalizes owned Shelly devices into relay select options', function (): voi
'model' => 'Shelly Pro 2PM',
],
'name' => 'Machine Cabinet',
'status' => [
'eth' => [
'ip' => '10.32.0.12',
],
],
'relays' => [
['ison' => false, 'name' => 'Program Picker'],
],
@@ -96,6 +104,7 @@ it('normalizes owned Shelly devices into relay select options', function (): voi
'code' => 'SPSW-001PE16EU',
'control_type' => 'Switch',
'control_name' => 'Entrance Relay',
'local_ip' => '10.32.0.11',
'status_color' => 'Green',
'online' => true,
],
@@ -109,6 +118,7 @@ it('normalizes owned Shelly devices into relay select options', function (): voi
'code' => 'SPSW-201XE16EU',
'control_type' => 'Relay',
'control_name' => 'Program Picker',
'local_ip' => '10.32.0.12',
'status_color' => 'Red',
'online' => false,
],
@@ -122,6 +132,7 @@ it('normalizes owned Shelly devices into relay select options', function (): voi
'code' => 'SHSW-1',
'control_type' => 'Relay',
'control_name' => null,
'local_ip' => null,
'status_color' => 'Red',
'online' => false,
],
@@ -142,6 +153,7 @@ it('falls back to local device and control names when Shelly cloud list metadata
'model' => 'Shelly Plus 1PM',
'online' => 1,
],
'ip' => '10.32.0.21',
'name' => 'Entry Gate',
'status' => [
'switch:0' => [
@@ -174,6 +186,7 @@ it('falls back to local device and control names when Shelly cloud list metadata
'code' => 'SPSW-001PE16EU',
'control_type' => 'Switch',
'control_name' => 'Entrance Relay',
'local_ip' => '10.32.0.21',
'status_color' => 'Green',
'online' => true,
],
services/nginx/app/tests/Unit/Selfserve/ShellyTransportResolverTest.php
+29
View File
@@ -4,6 +4,7 @@ app_require('classes/edge_gateway_manager.php');
app_require('classes/shelly_transport_resolver.php');
use classes\edge_gateway_manager;
use classes\gateway_shelly_transport;
use classes\shelly_transport_resolver;
use interfaces\shelly_transport_i;
@@ -75,3 +76,31 @@ it('resolves the injected cloud transport when a department is in cloud mode', f
expect($resolver->resolveForDepartment(22))->toBe($cloudTransport);
expect($manager->calls)->toBe([22]);
});
it('lets relay tests override the department transport without changing department mode', function (): void {
$manager = new ShellyTransportResolverManagerFake();
$manager->modesByDepartment[17] = edge_gateway_manager::TRANSPORT_MODE_CLOUD;
$cloudTransport = new ShellyTransportResolverTransportFake('cloud');
$gatewayTransport = new ShellyTransportResolverTransportFake('gateway');
$resolver = new shelly_transport_resolver($manager, $cloudTransport, $gatewayTransport);
expect($resolver->resolveForDepartment(17, 'local'))->toBe($gatewayTransport);
expect($resolver->resolveForDepartment(17, 'gateway'))->toBe($gatewayTransport);
expect($resolver->resolveForDepartment(17, 'cloud'))->toBe($cloudTransport);
expect($manager->calls)->toBe([]);
});
it('marks non-injected local overrides as local-only gateway transport', function (): void {
$manager = new ShellyTransportResolverManagerFake();
$resolver = new shelly_transport_resolver($manager);
$transport = $resolver->resolveForDepartment(17, 'local');
$localOnly = new ReflectionProperty($transport, 'localOnly');
$localOnly->setAccessible(true);
expect($transport)->toBeInstanceOf(gateway_shelly_transport::class);
expect($localOnly->getValue($transport))->toBeTrue();
expect($manager->calls)->toBe([]);
});