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

Add batch processing for lane hardware commands and status retrieval

Browse Source
This commit is contained in:
Jeppe Bundgaard
2026-06-30 09:30:43 +02:00
parent 5f13242cfa
commit 0da02dfeb5
10 changed files with 1914 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
services/nginx/app/modules/edgegateway/classes/edge_gateway_manager.php
+519 -1
View File
@@ -23,6 +23,8 @@ class edge_gateway_manager
public const DELIVERY_CHANNEL_BROKER = 'BROKER_FAST_PATH';
public const DELIVERY_CHANNEL_API = 'API_POLLING';
public const DELIVERY_CHANNEL_CLOUD = 'CLOUD';
public const COMMAND_BATCH_RELAY_STATUS = 'BATCH_RELAY_STATUS';
public const COMMAND_BATCH_SET_RELAY_STATE = 'BATCH_SET_RELAY_STATE';
public const RELAY_FALLBACK_PREFER_LOCAL = 'PREFER_LOCAL';
public const RELAY_FALLBACK_LOCAL_ONLY = 'LOCAL_ONLY';
public const RELAY_FALLBACK_CLOUD_ONLY = 'CLOUD_ONLY';
@@ -68,6 +70,11 @@ class edge_gateway_manager
public const COMMAND_POLL_INTERVAL_MICROSECONDS = 250000;
public const COMMAND_DISPATCH_STALE_AFTER_SECONDS = 30;
public const COMMAND_EXPIRES_AFTER_SECONDS = 90;
public const RELAY_BATCH_DEDUPE_SECONDS = 30;
public const RELAY_STATUS_RATE_LIMIT_PER_MINUTE = 60;
public const RELAY_SET_RATE_LIMIT_PER_MINUTE = 30;
public const RELAY_GATE_RATE_LIMIT_PER_MINUTE = 4;
public const RELAY_OPERATIONAL_WINDOW_SECONDS = 900;
public const BROKER_PRESENCE_TTL_SECONDS = 90;
public const BROKER_CONNECTIVITY_DEGRADED_AFTER_SECONDS = 45;
public const BROKER_HTTP_TIMEOUT_SECONDS = 5;
@@ -1127,6 +1134,179 @@ class edge_gateway_manager
);
}
/**
* @param array<int,array<string,mixed>> $requests
* @throws Exception
*/
public function queueRelayStatusBatch(
int $departmentId,
array $requests,
?int $userId = null,
bool $requireFastLocalPath = false
): array {
return $this->queueRelayBatch(
$departmentId,
self::COMMAND_BATCH_RELAY_STATUS,
$requests,
$userId,
$requireFastLocalPath
);
}
/**
* @param array<int,array<string,mixed>> $requests
* @throws Exception
*/
public function queueRelaySwitchBatch(
int $departmentId,
array $requests,
?int $userId = null,
bool $requireFastLocalPath = false
): array {
return $this->queueRelayBatch(
$departmentId,
self::COMMAND_BATCH_SET_RELAY_STATE,
$requests,
$userId,
$requireFastLocalPath
);
}
/**
* @return array<string,mixed>
* @throws Exception
*/
public function relayBatchStatus(string $batchId): array
{
$batchId = trim($batchId);
if ($batchId === '' || !preg_match('/^[a-f0-9]{32}$/', $batchId)) {
throw new Exception('Invalid relay batch id');
}
$statement = db::getPDO()->prepare(
"SELECT id
FROM edge_gateway_command_jobs
WHERE deleted_at IS NULL
AND JSON_UNQUOTE(JSON_EXTRACT(delivery_json, '$.batch_id')) = :batch_id
ORDER BY id ASC
LIMIT 10"
);
$statement->execute([':batch_id' => $batchId]);
$jobs = [];
foreach ($statement->fetchAll() ?: [] as $row) {
$jobId = (int)($row['id'] ?? 0);
if ($jobId > 0) {
$jobs[] = (new edge_gateway_command_jobs_o())->select($jobId);
}
}
if ($jobs === []) {
throw new Exception('Relay batch not found');
}
$items = [];
$pending = 0;
$failed = 0;
$completed = 0;
$unknownOutcomes = 0;
foreach ($jobs as $job) {
if (!$job->exists()) {
continue;
}
$delivery = (array)($job->delivery_json->value() ?? []);
if (
(string)$job->status->value() === 'DISPATCHING'
&& !empty($delivery['no_auto_retry'])
&& self::parseApplicationDateTime((string)($job->updated_at->value() ?? $job->requested_at->value())) <= time() - self::COMMAND_DISPATCH_STALE_AFTER_SECONDS
) {
$this->finalizeCommandJob(
$job,
false,
[],
'Edge gateway command outcome is unknown after dispatch interruption; not retrying non-idempotent gate pulse',
null,
'TIMED_OUT'
);
}
$status = strtoupper((string)$job->status->value());
$outcomeUnknown = $status === 'TIMED_OUT' && !empty($delivery['no_auto_retry']);
if ($outcomeUnknown) {
$unknownOutcomes++;
}
if (in_array($status, ['PENDING', 'DISPATCHING'], true)) {
$pending++;
} elseif ($status === 'COMPLETED') {
$completed++;
} else {
$failed++;
}
$request = (array)($job->request_json->value() ?? []);
$response = (array)($job->response_json->value() ?? []);
$payload = (array)($response['payload'] ?? []);
$resultItems = isset($payload['results']) && is_array($payload['results'])
? (array)$payload['results']
: [];
if ($resultItems === [] && $status === 'COMPLETED') {
$resultItems = [[
'target' => (string)($request['target'] ?? ''),
'relayId' => (string)($request['relayId'] ?? $request['relay_id'] ?? ''),
'relay_id' => (string)($request['relay_id'] ?? $request['relayId'] ?? ''),
'ok' => true,
'payload' => $payload,
]];
}
if ($resultItems === [] && $status !== 'COMPLETED') {
$items[] = [
'target' => (string)($request['target'] ?? ''),
'relay_id' => (string)($request['relayId'] ?? $request['relay_id'] ?? ''),
'status' => $status,
'ok' => false,
'outcome_unknown' => $outcomeUnknown,
'error' => $job->error_message->value() === null ? null : (string)$job->error_message->value(),
];
continue;
}
foreach ($resultItems as $item) {
if (!is_array($item)) {
continue;
}
$items[] = $item + [
'status' => !empty($item['ok']) ? 'COMPLETED' : 'FAILED',
'outcome_unknown' => $outcomeUnknown,
];
}
}
$itemFailures = count(array_filter($items, static fn(array $item): bool => empty($item['ok']) && (($item['status'] ?? '') !== 'PENDING')));
$status = 'PENDING';
if ($pending > 0 && ($completed > 0 || $failed > 0)) {
$status = 'RUNNING';
} elseif ($pending === 0 && $unknownOutcomes > 0 && $completed === 0) {
$status = 'UNKNOWN_OUTCOME';
} elseif ($pending === 0 && ($failed > 0 || $itemFailures > 0) && ($completed > 0 || count($items) > $itemFailures)) {
$status = 'PARTIAL';
} elseif ($pending === 0 && ($failed > 0 || $itemFailures > 0)) {
$status = 'FAILED';
} elseif ($pending === 0) {
$status = 'COMPLETED';
}
return [
'batch_id' => $batchId,
'status' => $status,
'unknown_outcome_count' => $unknownOutcomes,
'jobs' => array_map(static fn(edge_gateway_command_jobs_o $job): array => $job->asArray(), $jobs),
'items' => $items,
];
}
/**
* @throws Exception
*/
@@ -1201,6 +1381,275 @@ class edge_gateway_manager
}
}
/**
* @param array<int,array<string,mixed>> $requests
* @return array<string,mixed>
* @throws Exception
*/
private function queueRelayBatch(
int $departmentId,
string $commandType,
array $requests,
?int $userId,
bool $requireFastLocalPath
): array {
if ($departmentId <= 0) {
throw new Exception('A department_id is required for relay batches');
}
if ($requests === []) {
throw new Exception('At least one relay batch command is required');
}
if (count($requests) > 5) {
throw new Exception('Relay batch commands are limited to five targets');
}
$batchId = bin2hex(random_bytes(16));
$commands = [];
$jobs = [];
$gateway = null;
$seenTargets = [];
foreach ($requests as $request) {
if (!is_array($request)) {
throw new Exception('Invalid relay batch command');
}
$target = strtoupper(trim((string)($request['target'] ?? $request['relay'] ?? '')));
$logicalRelayId = trim((string)($request['relay_id'] ?? $request['relayId'] ?? ''));
if ($target === '' || $logicalRelayId === '') {
throw new Exception('Relay batch commands require target and relay_id');
}
if (isset($seenTargets[$target])) {
throw new Exception('Duplicate relay batch target: ' . $target);
}
$seenTargets[$target] = true;
$binding = $this->resolveRelayBinding($departmentId, $logicalRelayId);
$commandGateway = $this->requireGateway((int)$binding['gateway_id']);
if ($gateway === null) {
$gateway = $commandGateway;
} elseif ((int)$gateway->id !== (int)$commandGateway->id) {
throw new Exception('Relay batch commands must target the same edge gateway');
}
$resolution = $this->resolveRelayExecutionPlan($commandGateway, $binding, $logicalRelayId);
if ($requireFastLocalPath) {
$resolution = $this->forceLocalRelayExecutionPlan($resolution);
}
if (($resolution['execution_path'] ?? 'local') === 'cloud') {
throw new Exception('Cloud Shelly transport does not support asynchronous relay batches');
}
$command = [
'target' => $target,
'relayId' => $logicalRelayId,
'relay_id' => $logicalRelayId,
'deviceId' => $binding['device_id'],
'device_id' => $binding['device_id'],
'localIp' => $binding['local_ip'],
'local_ip' => $binding['local_ip'],
'channel' => (int)$binding['channel'],
];
$deviceGeneration = $this->resolveRelayBindingDeviceGeneration($binding);
if ($deviceGeneration !== null) {
$command['deviceGeneration'] = $deviceGeneration;
$command['device_generation'] = $deviceGeneration;
}
if ($commandType === self::COMMAND_BATCH_SET_RELAY_STATE) {
$command['on'] = (bool)($request['on'] ?? true);
$rawToggleAfter = $request['toggle_after'] ?? $request['toggleAfter'] ?? $request['timer'] ?? null;
$toggleAfter = $rawToggleAfter === null || $rawToggleAfter === ''
? null
: $this->normalizeRelayToggleAfter((int)$rawToggleAfter);
if ($toggleAfter !== null) {
$command['toggleAfter'] = $toggleAfter;
$command['toggle_after'] = $toggleAfter;
}
}
$commands[] = $command;
}
if ($gateway === null) {
throw new Exception('Unable to resolve edge gateway for relay batch');
}
$batchDedupeKey = $this->buildRelayBatchDedupeKey($departmentId, $commandType, $commands, $userId);
$existingBatch = $this->findRecentRelayBatchByDedupeKey((int)$gateway->id, $batchDedupeKey, self::RELAY_BATCH_DEDUPE_SECONDS);
if ($existingBatch !== null) {
$existingBatch['deduplicated'] = true;
$existingBatch['dedupe_window_seconds'] = self::RELAY_BATCH_DEDUPE_SECONDS;
return $existingBatch;
}
foreach ($commands as $command) {
$this->enforceRelayCommandRateLimit(
(int)$gateway->id,
$departmentId,
$commandType === self::COMMAND_BATCH_RELAY_STATUS ? 'GET_RELAY_STATUS' : 'SET_RELAY_STATE',
(string)$command['target'],
$userId
);
}
foreach ($commands as $command) {
$isGatePulse = $commandType === self::COMMAND_BATCH_SET_RELAY_STATE
&& in_array((string)$command['target'], ['ENTRANCE', 'EXIT'], true);
$jobs[] = $this->createCommandJob(
(int)$gateway->id,
$commandType === self::COMMAND_BATCH_RELAY_STATUS ? 'GET_RELAY_STATUS' : 'SET_RELAY_STATE',
$command,
$userId,
[
'department_id' => $departmentId,
'batch_id' => $batchId,
'batch_dedupe_key' => $batchDedupeKey,
'idempotency_key' => $batchId . ':' . (string)$command['target'],
'target' => (string)$command['target'],
'physical_command' => true,
'preferred_channel' => self::DELIVERY_CHANNEL_API,
'require_fast_path' => false,
'no_auto_retry' => $isGatePulse,
]
);
}
edge_gateway_view_cache::syncGateway($this->getGateway((int)$gateway->id));
return [
'batch_id' => $batchId,
'status' => 'PENDING',
'gateway_id' => (int)$gateway->id,
'deduplicated' => false,
'dedupe_window_seconds' => self::RELAY_BATCH_DEDUPE_SECONDS,
'jobs' => array_map(static fn(edge_gateway_command_jobs_o $job): array => $job->asArray(), $jobs),
'items' => array_map(static fn(array $command, edge_gateway_command_jobs_o $job): array => [
'target' => (string)$command['target'],
'relay_id' => (string)$command['relayId'],
'job_id' => (int)$job->id,
'status' => 'PENDING',
], $commands, $jobs),
];
}
/**
* @param array<int,array<string,mixed>> $commands
*/
private function buildRelayBatchDedupeKey(int $departmentId, string $commandType, array $commands, ?int $userId): string
{
$normalizedCommands = array_map(static function (array $command): array {
return [
'target' => (string)($command['target'] ?? ''),
'relay_id' => (string)($command['relay_id'] ?? $command['relayId'] ?? ''),
'device_id' => (string)($command['device_id'] ?? $command['deviceId'] ?? ''),
'channel' => (int)($command['channel'] ?? 0),
'on' => array_key_exists('on', $command) ? (bool)$command['on'] : null,
'toggle_after' => isset($command['toggle_after']) ? (int)$command['toggle_after'] : null,
];
}, $commands);
usort($normalizedCommands, static fn(array $left, array $right): int => ($left['target'] <=> $right['target']) ?: ($left['relay_id'] <=> $right['relay_id']));
return hash('sha256', json_encode([
'department_id' => $departmentId,
'command_type' => $commandType,
'requested_by' => $userId ?? 0,
'commands' => $normalizedCommands,
], JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES) ?: '');
}
/**
* @return array<string,mixed>|null
* @throws Exception
*/
private function findRecentRelayBatchByDedupeKey(int $gatewayId, string $dedupeKey, int $windowSeconds): ?array
{
if ($gatewayId <= 0 || $dedupeKey === '') {
return null;
}
$statement = db::getPDO()->prepare(
"SELECT JSON_UNQUOTE(JSON_EXTRACT(delivery_json, '$.batch_id')) AS batch_id
FROM edge_gateway_command_jobs
WHERE gateway_id = :gateway_id
AND deleted_at IS NULL
AND requested_at >= :cutoff
AND JSON_UNQUOTE(JSON_EXTRACT(delivery_json, '$.batch_dedupe_key')) = :dedupe_key
ORDER BY requested_at DESC, id DESC
LIMIT 1"
);
$statement->execute([
':gateway_id' => $gatewayId,
':cutoff' => $this->formatDateTime(time() - max(1, $windowSeconds)),
':dedupe_key' => $dedupeKey,
]);
$row = $statement->fetch();
$batchId = is_array($row) ? trim((string)($row['batch_id'] ?? '')) : '';
return $batchId === '' ? null : $this->relayBatchStatus($batchId);
}
/**
* @throws Exception
*/
private function enforceRelayCommandRateLimit(
int $gatewayId,
int $departmentId,
string $commandType,
string $target,
?int $userId
): void {
$target = strtoupper(trim($target));
$isGate = in_array($target, ['ENTRANCE', 'EXIT'], true);
$limit = $commandType === 'GET_RELAY_STATUS'
? self::RELAY_STATUS_RATE_LIMIT_PER_MINUTE
: ($isGate ? self::RELAY_GATE_RATE_LIMIT_PER_MINUTE : self::RELAY_SET_RATE_LIMIT_PER_MINUTE);
$clauses = [
'gateway_id = :gateway_id',
'deleted_at IS NULL',
'requested_at >= :cutoff',
'command_type = :command_type',
"JSON_UNQUOTE(JSON_EXTRACT(delivery_json, '$.department_id')) = :department_id",
];
$params = [
':gateway_id' => $gatewayId,
':cutoff' => $this->formatDateTime(time() - 60),
':command_type' => $commandType,
':department_id' => (string)$departmentId,
];
if ($userId === null) {
$clauses[] = 'requested_by IS NULL';
} else {
$clauses[] = 'requested_by = :requested_by';
$params[':requested_by'] = $userId;
}
if ($isGate) {
$clauses[] = "JSON_UNQUOTE(JSON_EXTRACT(delivery_json, '$.target')) = :target";
$params[':target'] = $target;
}
$statement = db::getPDO()->prepare(
'SELECT COUNT(*) AS c
FROM edge_gateway_command_jobs
WHERE ' . implode(' AND ', $clauses)
);
$statement->execute($params);
$row = $statement->fetch();
$count = isset($row['c']) ? (int)$row['c'] : 0;
if ($count >= $limit) {
throw new Exception(sprintf(
'Relay command rate limit exceeded for %s. Try again shortly.',
$isGate ? strtolower($target) . ' gate' : strtolower(str_replace('_', ' ', $commandType))
));
}
}
/**
* @return array<int,array<string,mixed>>
*/
@@ -3199,6 +3648,7 @@ BASH;
OR (
status = :dispatching_status_match
AND COALESCE(updated_at, created_at, requested_at) <= :stale_before
AND COALESCE(JSON_UNQUOTE(JSON_EXTRACT(delivery_json, \'$.no_auto_retry\')), \'false\') <> \'true\'
)
)
ORDER BY CASE WHEN status = :pending_status_order THEN 0 ELSE 1 END, requested_at ASC, id ASC
@@ -3841,6 +4291,13 @@ BASH;
$counts = [
'command_backlog' => 0,
'operation_backlog' => 0,
'stale_dispatching_commands' => 0,
'recent_command_failures' => 0,
'recent_command_timeouts' => 0,
'recent_unknown_gate_outcomes' => 0,
'recent_relay_commands' => 0,
'recent_relay_failure_rate' => 0.0,
'recent_command_avg_latency_seconds' => null,
'last_successful_command_at' => null,
'last_successful_discovery_at' => null,
'last_successful_operation_at' => null,
@@ -3858,15 +4315,76 @@ BASH;
WHERE gateway_id = :gateway_id
AND deleted_at IS NULL
AND status IN ('PENDING', 'IN_PROGRESS')",
'stale_dispatching_commands' => "SELECT COUNT(*) AS c
FROM edge_gateway_command_jobs
WHERE gateway_id = :gateway_id
AND deleted_at IS NULL
AND status = 'DISPATCHING'
AND COALESCE(updated_at, created_at, requested_at) <= :stale_before",
'recent_command_failures' => "SELECT COUNT(*) AS c
FROM edge_gateway_command_jobs
WHERE gateway_id = :gateway_id
AND deleted_at IS NULL
AND status IN ('FAILED', 'TIMED_OUT')
AND requested_at >= :recent_cutoff",
'recent_command_timeouts' => "SELECT COUNT(*) AS c
FROM edge_gateway_command_jobs
WHERE gateway_id = :gateway_id
AND deleted_at IS NULL
AND status = 'TIMED_OUT'
AND requested_at >= :recent_cutoff",
'recent_unknown_gate_outcomes' => "SELECT COUNT(*) AS c
FROM edge_gateway_command_jobs
WHERE gateway_id = :gateway_id
AND deleted_at IS NULL
AND status = 'TIMED_OUT'
AND requested_at >= :recent_cutoff
AND COALESCE(JSON_UNQUOTE(JSON_EXTRACT(delivery_json, '$.no_auto_retry')), 'false') = 'true'",
'recent_relay_commands' => "SELECT COUNT(*) AS c
FROM edge_gateway_command_jobs
WHERE gateway_id = :gateway_id
AND deleted_at IS NULL
AND command_type IN ('GET_RELAY_STATUS', 'SET_RELAY_STATE')
AND requested_at >= :recent_cutoff",
];
foreach ($countQueries as $key => $sql) {
$statement = $pdo->prepare($sql);
$statement->execute([':gateway_id' => $gatewayId]);
$params = [
':gateway_id' => $gatewayId,
];
if (str_contains($sql, ':stale_before')) {
$params[':stale_before'] = $this->formatDateTime(time() - self::COMMAND_DISPATCH_STALE_AFTER_SECONDS);
}
if (str_contains($sql, ':recent_cutoff')) {
$params[':recent_cutoff'] = $this->formatDateTime(time() - self::RELAY_OPERATIONAL_WINDOW_SECONDS);
}
$statement->execute($params);
$row = $statement->fetch();
$counts[$key] = isset($row['c']) ? (int)$row['c'] : 0;
}
$counts['recent_relay_failure_rate'] = $counts['recent_relay_commands'] > 0
? round($counts['recent_command_failures'] / $counts['recent_relay_commands'], 4)
: 0.0;
$latencyStatement = $pdo->prepare(
"SELECT AVG(TIMESTAMPDIFF(SECOND, requested_at, completed_at)) AS avg_latency
FROM edge_gateway_command_jobs
WHERE gateway_id = :gateway_id
AND deleted_at IS NULL
AND completed_at IS NOT NULL
AND requested_at >= :recent_cutoff"
);
$latencyStatement->execute([
':gateway_id' => $gatewayId,
':recent_cutoff' => $this->formatDateTime(time() - self::RELAY_OPERATIONAL_WINDOW_SECONDS),
]);
$latencyRow = $latencyStatement->fetch();
if (is_array($latencyRow) && $latencyRow['avg_latency'] !== null) {
$counts['recent_command_avg_latency_seconds'] = round((float)$latencyRow['avg_latency'], 2);
}
$timestampQueries = [
'last_successful_command_at' => "SELECT completed_at AS ts
FROM edge_gateway_command_jobs