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

Add unit tests for invoicing, orders normalization, gateway commands, and department complaints. Update schema bootstraps and improve agent command execution logic.

Browse Source
This commit is contained in:
Jeppe Bundgaard
2026-04-09 12:24:22 +02:00
parent 9e9372db05
commit 22f856c62a
35 changed files with 2346 additions and 167 deletions
Download Patch File Download Diff File Expand all files Collapse all files
services/nginx/app/classes/edge_gateway_manager.php
+426 -63
View File
@@ -28,6 +28,10 @@ class edge_gateway_manager
public const SHELL_SESSION_TTL_SECONDS = 900;
public const HEARTBEAT_DEGRADED_AFTER_SECONDS = 60;
public const HEARTBEAT_OFFLINE_AFTER_SECONDS = 300;
public const COMMAND_WAIT_TIMEOUT_SECONDS = 10;
public const COMMAND_POLL_TIMEOUT_SECONDS = 20;
public const COMMAND_POLL_INTERVAL_MICROSECONDS = 250000;
public const COMMAND_DISPATCH_STALE_AFTER_SECONDS = 30;
public function __construct(private readonly ?edge_broker_client $brokerClient = null)
{
@@ -168,7 +172,6 @@ class edge_gateway_manager
$gateway->target_version->set($payload['target_version'] ?? $gateway->target_version->value());
$gateway->last_heartbeat_at->set($this->now());
$gateway->last_seen_ip->set($this->remoteIp());
$gateway->discovery_status->set((string)($payload['discovery_status'] ?? $gateway->discovery_status->value()));
$gateway->metadata_json->set((array)($payload['metadata'] ?? $gateway->metadata_json->value() ?? []));
if (isset($payload['inventory']) && is_array($payload['inventory'])) {
@@ -324,15 +327,9 @@ class edge_gateway_manager
*/
public function queueDiscovery(int $gatewayId, ?int $userId = null): array
{
$gateway = $this->requireGateway($gatewayId);
$job = $this->createCommandJob($gatewayId, 'DISCOVER_SHELLY', [], $userId);
$response = $this->dispatchCommandJob($job, $gateway, [
'requestedBy' => $userId,
]);
if (isset($response['inventory']) && is_array($response['inventory'])) {
$this->syncDeviceInventory($gatewayId, $response['inventory']);
}
$gateway = $this->requireDispatchableGateway($gatewayId);
$gateway->discovery_status->set('PENDING');
$this->createCommandJob($gatewayId, 'DISCOVER_SHELLY', [], $userId);
return $this->getGateway($gatewayId);
}
@@ -342,10 +339,13 @@ class edge_gateway_manager
*/
public function queueUpdate(int $gatewayId, string $targetVersion, string $releaseChannel, ?int $userId = null): array
{
$gateway = $this->requireGateway($gatewayId);
$gateway = $this->requireDispatchableGateway($gatewayId);
$gateway->target_version->set($targetVersion);
$jobObject = new edge_gateway_update_jobs_o();
$jobId = $jobObject->add_object([
'gateway_id' => $gatewayId,
'command_job_id' => null,
'target_version' => $targetVersion,
'release_channel' => $releaseChannel,
'status' => 'PENDING',
@@ -359,22 +359,7 @@ class edge_gateway_manager
'targetVersion' => $targetVersion,
'releaseChannel' => $releaseChannel,
], $userId);
try {
$response = $this->dispatchCommandJob($command, $gateway, [
'targetVersion' => $targetVersion,
'releaseChannel' => $releaseChannel,
]);
$jobObject->status->set('COMPLETED');
$jobObject->started_at->set($this->now());
$jobObject->completed_at->set($this->now());
$jobObject->result_json->set($response);
} catch (\Throwable $throwable) {
$jobObject->status->set('FAILED');
$jobObject->completed_at->set($this->now());
$jobObject->result_json->set(['error' => $throwable->getMessage()]);
throw $throwable;
}
$jobObject->command_job_id->set((int)$command->id);
$this->writeAudit(
$gatewayId,
@@ -387,6 +372,71 @@ class edge_gateway_manager
return $jobObject->asArray();
}
/**
* @throws Exception
*/
public function pollCommand(int $gatewayId, string $plainToken, int $waitSeconds = self::COMMAND_POLL_TIMEOUT_SECONDS): ?array
{
$gateway = $this->authenticateGateway($gatewayId, $plainToken);
$deadline = microtime(true) + max(0, $waitSeconds);
do {
$job = $this->claimNextCommandJob($gateway);
if ($job !== null) {
return $this->formatAgentCommandJob($job, $gateway);
}
if (microtime(true) >= $deadline) {
break;
}
usleep(self::COMMAND_POLL_INTERVAL_MICROSECONDS);
} while (true);
return null;
}
/**
* @throws Exception
*/
public function submitCommandResult(
int $gatewayId,
int $jobId,
string $plainToken,
bool $ok,
array $payload = [],
?string $error = null
): array {
$gateway = $this->authenticateGateway($gatewayId, $plainToken);
$job = (new edge_gateway_command_jobs_o())->select($jobId);
if (!$job->exists()) {
throw new Exception('Edge gateway command job not found');
}
if ((int)$job->gateway_id->value() !== (int)$gateway->id) {
throw new Exception('Edge gateway command job does not belong to this gateway');
}
$status = (string)$job->status->value();
if (in_array($status, ['COMPLETED', 'FAILED'], true)) {
return [
'acknowledged' => true,
'job' => $job->asArray(),
];
}
$errorMessage = $ok ? null : trim((string)$error);
if (!$ok && $errorMessage === '') {
$errorMessage = 'Edge gateway command failed';
}
$this->finalizeCommandJob($job, $ok, $payload, $errorMessage, $gateway);
return [
'acknowledged' => true,
'job' => $job->asArray(),
];
}
/**
* @throws Exception
*/
@@ -502,7 +552,7 @@ class edge_gateway_manager
public function dispatchRelayStatus(int $departmentId, string $logicalRelayId): array
{
$binding = $this->resolveRelayBinding($departmentId, $logicalRelayId);
$gateway = $this->requireGateway((int)$binding['gateway_id']);
$gateway = $this->requireDispatchableGateway((int)$binding['gateway_id']);
$job = $this->createCommandJob((int)$gateway->id, 'GET_RELAY_STATUS', [
'relayId' => $logicalRelayId,
'deviceId' => $binding['device_id'],
@@ -510,12 +560,9 @@ class edge_gateway_manager
'channel' => (int)$binding['channel'],
], null);
return $this->dispatchCommandJob($job, $gateway, [
'relayId' => $logicalRelayId,
'deviceId' => $binding['device_id'],
'localIp' => $binding['local_ip'],
'channel' => (int)$binding['channel'],
]);
$this->tryImmediateBrokerDispatch($job, $gateway);
return $this->waitForCommandResult((int)$job->id);
}
/**
@@ -524,7 +571,7 @@ class edge_gateway_manager
public function dispatchRelaySwitch(int $departmentId, string $logicalRelayId, bool $on): array
{
$binding = $this->resolveRelayBinding($departmentId, $logicalRelayId);
$gateway = $this->requireGateway((int)$binding['gateway_id']);
$gateway = $this->requireDispatchableGateway((int)$binding['gateway_id']);
$job = $this->createCommandJob((int)$gateway->id, 'SET_RELAY_STATE', [
'relayId' => $logicalRelayId,
'deviceId' => $binding['device_id'],
@@ -533,13 +580,9 @@ class edge_gateway_manager
'on' => $on,
], null);
return $this->dispatchCommandJob($job, $gateway, [
'relayId' => $logicalRelayId,
'deviceId' => $binding['device_id'],
'localIp' => $binding['local_ip'],
'channel' => (int)$binding['channel'],
'on' => $on,
]);
$this->tryImmediateBrokerDispatch($job, $gateway);
return $this->waitForCommandResult((int)$job->id);
}
/**
@@ -592,7 +635,7 @@ INSTALL_DIR=/opt/truckwash-edge-agent
mkdir -p "$INSTALL_DIR"
export DEBIAN_FRONTEND=noninteractive
apt-get update
apt-get install -y curl ca-certificates nodejs npm
apt-get install -y curl ca-certificates nodejs npm python3 make g++
curl -fsSL "__PACKAGE_URL__" -o "$INSTALL_DIR/package.json"
curl -fsSL "__AGENT_URL__" -o "$INSTALL_DIR/agent.mjs"
cat > "$INSTALL_DIR/config.json" <<'EOF_JSON'
@@ -654,7 +697,7 @@ BASH;
{
$configured = trim((string)(getenv('EDGE_BROKER_PUBLIC_URL') ?: ''));
if ($configured !== '') {
return $configured;
return $this->normalizeBrokerPublicUrl($configured);
}
$apiBaseUrl = $this->getApiBaseUrl();
@@ -665,7 +708,7 @@ BASH;
$scheme = ($parsed['scheme'] ?? 'https') === 'https' ? 'https' : 'http';
$port = getenv('EDGE_BROKER_PUBLIC_PORT') ?: '4300';
return $scheme . '://' . $parsed['host'] . ':' . $port;
return $this->normalizeBrokerPublicUrl($scheme . '://' . $parsed['host'] . ':' . $port);
}
public function getApiBaseUrl(): string
@@ -740,6 +783,76 @@ BASH;
return null;
}
private function normalizeBrokerPublicUrl(string $url): string
{
$parsed = parse_url($url);
if (!is_array($parsed) || !isset($parsed['host'])) {
return trim($url);
}
$host = (string)$parsed['host'];
$scheme = strtolower((string)($parsed['scheme'] ?? ''));
if ($scheme === '' || ($scheme === 'http' && $this->shouldUseSecureBrokerScheme($host))) {
$scheme = $this->shouldUseSecureBrokerScheme($host) ? 'https' : 'http';
}
$normalized = $scheme . '://' . $host;
if (isset($parsed['port'])) {
$normalized .= ':' . $parsed['port'];
}
if (isset($parsed['path'])) {
$normalized .= $parsed['path'];
}
if (isset($parsed['query'])) {
$normalized .= '?' . $parsed['query'];
}
if (isset($parsed['fragment'])) {
$normalized .= '#' . $parsed['fragment'];
}
return $normalized;
}
private function shouldUseSecureBrokerScheme(string $host): bool
{
$normalized = strtolower(trim($host, '[]'));
if ($normalized === '' || $normalized === 'localhost' || $normalized === 'edge-broker') {
return false;
}
if (str_ends_with($normalized, '.localhost')
|| str_ends_with($normalized, '.local')
|| str_ends_with($normalized, '.lan')
|| str_ends_with($normalized, '.internal')
|| str_ends_with($normalized, '.home.arpa')) {
return false;
}
if (filter_var($normalized, FILTER_VALIDATE_IP) !== false) {
return filter_var($normalized, FILTER_VALIDATE_IP, FILTER_FLAG_NO_PRIV_RANGE | FILTER_FLAG_NO_RES_RANGE) !== false;
}
return str_contains($normalized, '.');
}
/**
* @throws Exception
*/
private function requireDispatchableGateway(int $gatewayId): edge_gateways_o
{
$gateway = $this->requireGateway($gatewayId);
$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)) {
throw new Exception('Gateway agent is offline');
}
return $gateway;
}
/**
* @throws Exception
*/
@@ -856,32 +969,282 @@ BASH;
/**
* @throws Exception
*/
private function dispatchCommandJob(edge_gateway_command_jobs_o $job, edge_gateways_o $gateway, array $payload): array
private function tryImmediateBrokerDispatch(edge_gateway_command_jobs_o $job, edge_gateways_o $gateway): void
{
$job->status->set('DISPATCHING');
$this->markCommandJobDispatching($job);
$response = $this->broker()->dispatchCommand(
(int)$gateway->id,
(string)$job->command_type->value(),
[
'jobId' => (int)$job->id,
'gatewayId' => (int)$gateway->id,
'departmentId' => (int)$gateway->department_id->value(),
'payload' => $payload,
]
);
try {
$response = $this->broker()->dispatchCommand(
(int)$gateway->id,
(string)$job->command_type->value(),
$this->buildCommandExecutionPayload($job, $gateway)
);
$ok = (bool)($response['ok'] ?? false);
$payload = (array)($response['payload'] ?? []);
$errorMessage = $ok ? null : trim((string)($response['error'] ?? 'Edge broker command failed'));
$this->finalizeCommandJob($job, $ok, $payload, $errorMessage, $gateway);
if (!$ok) {
throw new Exception($errorMessage ?: 'Edge broker command failed');
}
} catch (\Throwable $throwable) {
if ($this->shouldFallbackToQueuedDelivery($throwable)) {
$this->releaseCommandJobToQueue($job);
return;
}
$this->finalizeCommandJob($job, false, [], $throwable->getMessage(), $gateway);
throw $throwable;
}
}
/**
* @throws Exception
*/
private function waitForCommandResult(int $jobId, int $timeoutSeconds = self::COMMAND_WAIT_TIMEOUT_SECONDS): array
{
$deadline = microtime(true) + max(0, $timeoutSeconds);
do {
$job = (new edge_gateway_command_jobs_o())->select($jobId);
if (!$job->exists()) {
throw new Exception('Edge gateway command job not found');
}
$status = (string)$job->status->value();
if ($status === 'COMPLETED') {
$response = (array)($job->response_json->value() ?? []);
return (array)($response['payload'] ?? []);
}
if ($status === 'FAILED') {
$errorMessage = trim((string)($job->error_message->value() ?? ''));
throw new Exception($errorMessage !== '' ? $errorMessage : 'Edge gateway command failed');
}
if (microtime(true) >= $deadline) {
break;
}
usleep(self::COMMAND_POLL_INTERVAL_MICROSECONDS);
} while (true);
throw new Exception('Edge gateway command timed out');
}
private function claimNextCommandJob(edge_gateways_o $gateway): ?edge_gateway_command_jobs_o
{
$pdo = db::getPDO();
$pdo->beginTransaction();
try {
$statement = $pdo->prepare(
'SELECT id
FROM edge_gateway_command_jobs
WHERE gateway_id = :gateway_id
AND deleted_at IS NULL
AND (
status = :pending_status_match
OR (
status = :dispatching_status_match
AND COALESCE(updated_at, created_at, requested_at) <= :stale_before
)
)
ORDER BY CASE WHEN status = :pending_status_order THEN 0 ELSE 1 END, requested_at ASC, id ASC
LIMIT 1
FOR UPDATE'
);
$statement->execute([
':gateway_id' => (int)$gateway->id,
':pending_status_match' => 'PENDING',
':dispatching_status_match' => 'DISPATCHING',
':pending_status_order' => 'PENDING',
':stale_before' => $this->formatDateTime(time() - self::COMMAND_DISPATCH_STALE_AFTER_SECONDS),
]);
$row = $statement->fetch();
if (!is_array($row) || !isset($row['id'])) {
$pdo->commit();
return null;
}
$update = $pdo->prepare(
'UPDATE edge_gateway_command_jobs
SET status = :status,
response_json = :response_json,
error_message = NULL,
completed_at = NULL
WHERE id = :id'
);
$update->execute([
':status' => 'DISPATCHING',
':response_json' => json_encode([], JSON_UNESCAPED_UNICODE),
':id' => (int)$row['id'],
]);
$pdo->commit();
} catch (\Throwable $throwable) {
if ($pdo->inTransaction()) {
$pdo->rollBack();
}
throw $throwable;
}
$job = (new edge_gateway_command_jobs_o())->select((int)$row['id']);
$this->markLinkedUpdateJobStarted((int)$job->id);
return $job;
}
private function formatAgentCommandJob(edge_gateway_command_jobs_o $job, edge_gateways_o $gateway): array
{
return [
'id' => (int)$job->id,
'gateway_id' => (int)$gateway->id,
'department_id' => (int)$gateway->department_id->value(),
'command_type' => (string)$job->command_type->value(),
'commandType' => (string)$job->command_type->value(),
'payload' => $this->buildCommandExecutionPayload($job, $gateway),
'requested_at' => (string)$job->requested_at->value(),
];
}
private function buildCommandExecutionPayload(edge_gateway_command_jobs_o $job, edge_gateways_o $gateway): array
{
return array_merge([
'jobId' => (int)$job->id,
'gatewayId' => (int)$gateway->id,
'departmentId' => (int)$gateway->department_id->value(),
'correlationId' => (string)$job->correlation_id->value(),
], (array)($job->request_json->value() ?? []));
}
private function finalizeCommandJob(
edge_gateway_command_jobs_o $job,
bool $ok,
array $payload = [],
?string $errorMessage = null,
?edge_gateways_o $gateway = null
): void {
$gatewayObject = $gateway ?? $this->requireGateway((int)$job->gateway_id->value());
$response = [
'ok' => $ok,
'payload' => $payload,
];
if (!$ok && $errorMessage !== null && trim($errorMessage) !== '') {
$response['error'] = $errorMessage;
}
$job->response_json->set($response);
$job->completed_at->set($this->now());
$ok = (bool)($response['ok'] ?? false);
$job->error_message->set($ok ? null : $errorMessage);
$job->status->set($ok ? 'COMPLETED' : 'FAILED');
if (!$ok) {
$job->error_message->set((string)($response['error'] ?? 'Edge broker command failed'));
throw new Exception((string)($response['error'] ?? 'Edge broker command failed'));
$this->applyCommandResult($gatewayObject, $job, $ok, $payload, $errorMessage);
}
private function applyCommandResult(
edge_gateways_o $gateway,
edge_gateway_command_jobs_o $job,
bool $ok,
array $payload,
?string $errorMessage
): void {
$commandType = (string)$job->command_type->value();
if ($commandType === 'DISCOVER_SHELLY') {
if ($ok) {
$inventory = isset($payload['inventory']) && is_array($payload['inventory']) ? $payload['inventory'] : [];
$this->syncDeviceInventory((int)$gateway->id, $inventory);
$gateway->discovery_status->set('READY');
} else {
$gateway->discovery_status->set('FAILED');
}
return;
}
return (array)($response['payload'] ?? $response);
if ($commandType === 'RUN_UPDATE') {
$this->finalizeLinkedUpdateJob((int)$job->id, $ok, $payload, $errorMessage);
}
}
private function releaseCommandJobToQueue(edge_gateway_command_jobs_o $job): void
{
$job->response_json->set([]);
$job->error_message->set(null);
$job->completed_at->set(null);
$job->status->set('PENDING');
}
private function markCommandJobDispatching(edge_gateway_command_jobs_o $job): void
{
$job->response_json->set([]);
$job->error_message->set(null);
$job->completed_at->set(null);
$job->status->set('DISPATCHING');
}
private function shouldFallbackToQueuedDelivery(\Throwable $throwable): bool
{
if ($throwable instanceof edge_broker_transport_exception) {
return true;
}
if ($throwable instanceof edge_broker_http_exception) {
return $throwable->statusCode() === 503;
}
$message = $throwable->getMessage();
return str_contains($message, 'Gateway agent is offline')
|| str_contains($message, 'Could not resolve host:')
|| str_contains($message, 'Failed to connect')
|| str_contains($message, 'Connection refused');
}
private function markLinkedUpdateJobStarted(int $commandJobId): void
{
$updateJob = $this->findLinkedUpdateJobByCommandId($commandJobId);
if ($updateJob === null) {
return;
}
if ($updateJob->started_at->value() === null) {
$updateJob->started_at->set($this->now());
}
}
private function finalizeLinkedUpdateJob(int $commandJobId, bool $ok, array $payload, ?string $errorMessage): void
{
$updateJob = $this->findLinkedUpdateJobByCommandId($commandJobId);
if ($updateJob === null) {
return;
}
if ($updateJob->started_at->value() === null) {
$updateJob->started_at->set($this->now());
}
$updateJob->status->set($ok ? 'COMPLETED' : 'FAILED');
$updateJob->completed_at->set($this->now());
$updateJob->result_json->set($ok
? $payload
: ['error' => $errorMessage ?: 'Edge gateway update failed']);
}
private function findLinkedUpdateJobByCommandId(int $commandJobId): ?edge_gateway_update_jobs_o
{
$rows = (new edge_gateway_update_jobs_o())->getFieldsWhere([
'command_job_id' => $commandJobId,
'deleted_at' => null,
], ['id']);
if ($rows === []) {
return null;
}
return (new edge_gateway_update_jobs_o())->select((int)$rows[0]['id']);
}
/**