3159 lines
116 KiB
PHP
3159 lines
116 KiB
PHP
#!/usr/bin/env php
|
|
<?php
|
|
|
|
if (PHP_SAPI !== 'cli') {
|
|
fwrite(STDERR, "This agent must run from the CLI.\n");
|
|
exit(1);
|
|
}
|
|
|
|
final class AgentConfig
|
|
{
|
|
public function __construct(public array $data, public string $path)
|
|
{
|
|
}
|
|
|
|
public static function load(string $path): self
|
|
{
|
|
if (!is_file($path)) {
|
|
throw new RuntimeException('Missing config file: ' . $path);
|
|
}
|
|
|
|
$decoded = json_decode((string)file_get_contents($path), true);
|
|
if (!is_array($decoded)) {
|
|
throw new RuntimeException('Invalid config file: ' . $path);
|
|
}
|
|
|
|
return new self($decoded, $path);
|
|
}
|
|
|
|
public function get(string $key, mixed $default = null): mixed
|
|
{
|
|
return array_key_exists($key, $this->data) ? $this->data[$key] : $default;
|
|
}
|
|
|
|
public function set(string $key, mixed $value): void
|
|
{
|
|
$this->data[$key] = $value;
|
|
}
|
|
|
|
public function save(): void
|
|
{
|
|
file_put_contents($this->path, json_encode($this->data, JSON_PRETTY_PRINT | JSON_UNESCAPED_SLASHES) . PHP_EOL);
|
|
}
|
|
}
|
|
|
|
final class HttpRequestTimeoutException extends RuntimeException
|
|
{
|
|
}
|
|
|
|
final class OperationAbortException extends RuntimeException
|
|
{
|
|
public function __construct(string $message, private readonly bool $cancelled = false)
|
|
{
|
|
parent::__construct($message);
|
|
}
|
|
|
|
public function isCancellation(): bool
|
|
{
|
|
return $this->cancelled;
|
|
}
|
|
}
|
|
|
|
final class HttpJsonClient
|
|
{
|
|
public function __construct(private readonly string $baseUrl, private readonly array $defaultHeaders = [])
|
|
{
|
|
}
|
|
|
|
public function post(string $path, array $payload, int $timeoutSeconds = 30): ?array
|
|
{
|
|
$response = $this->requestJson(
|
|
'POST',
|
|
rtrim($this->baseUrl, '/') . '/' . ltrim($path, '/'),
|
|
$payload,
|
|
$timeoutSeconds
|
|
);
|
|
|
|
return $response['decoded'];
|
|
}
|
|
|
|
public function getJson(string $url, int $timeoutSeconds = 10): array
|
|
{
|
|
$response = $this->requestJson('GET', $url, null, $timeoutSeconds);
|
|
return $response['decoded'] ?? [];
|
|
}
|
|
|
|
public function download(string $url, int $timeoutSeconds = 60): string
|
|
{
|
|
$response = $this->requestRaw('GET', $url, null, $timeoutSeconds, ['Accept: */*']);
|
|
return $response['body'];
|
|
}
|
|
|
|
private function requestJson(string $method, string $url, ?array $payload, int $timeoutSeconds): array
|
|
{
|
|
$headers = array_values(array_merge(['Accept: application/json'], $this->defaultHeaders));
|
|
if ($payload !== null) {
|
|
$headers[] = 'Content-Type: application/json';
|
|
}
|
|
|
|
$response = $this->requestRaw($method, $url, $payload, $timeoutSeconds, $headers);
|
|
$decoded = json_decode($response['body'], true);
|
|
if (!is_array($decoded)) {
|
|
throw new RuntimeException('Invalid JSON response from ' . $method . ' ' . $url);
|
|
}
|
|
|
|
$response['decoded'] = $decoded;
|
|
return $response;
|
|
}
|
|
|
|
private function requestRaw(
|
|
string $method,
|
|
string $url,
|
|
?array $payload,
|
|
int $timeoutSeconds,
|
|
array $headers
|
|
): array {
|
|
$responseHeaders = [];
|
|
$ch = curl_init($url);
|
|
curl_setopt_array($ch, [
|
|
CURLOPT_CUSTOMREQUEST => $method,
|
|
CURLOPT_RETURNTRANSFER => true,
|
|
CURLOPT_TIMEOUT => $timeoutSeconds,
|
|
CURLOPT_CONNECTTIMEOUT => min(10, $timeoutSeconds),
|
|
CURLOPT_FOLLOWLOCATION => true,
|
|
CURLOPT_HTTPHEADER => $headers,
|
|
CURLOPT_HEADERFUNCTION => static function ($curl, string $headerLine) use (&$responseHeaders): int {
|
|
$trimmed = trim($headerLine);
|
|
if ($trimmed !== '') {
|
|
$responseHeaders[] = $trimmed;
|
|
}
|
|
return strlen($headerLine);
|
|
},
|
|
]);
|
|
|
|
if ($payload !== null) {
|
|
curl_setopt($ch, CURLOPT_POSTFIELDS, json_encode($payload, JSON_UNESCAPED_SLASHES));
|
|
}
|
|
|
|
$raw = curl_exec($ch);
|
|
$status = (int)curl_getinfo($ch, CURLINFO_RESPONSE_CODE);
|
|
$errno = curl_errno($ch);
|
|
$error = curl_error($ch);
|
|
curl_close($ch);
|
|
|
|
if ($raw === false) {
|
|
$message = sprintf('%s %s failed: %s', $method, $url, $error !== '' ? $error : 'unknown curl error');
|
|
if ($errno === CURLE_OPERATION_TIMEDOUT) {
|
|
throw new HttpRequestTimeoutException($message);
|
|
}
|
|
|
|
throw new RuntimeException($message);
|
|
}
|
|
|
|
if ($status >= 400) {
|
|
$preview = substr(preg_replace('/\s+/', ' ', (string)$raw) ?? '', 0, 240);
|
|
$headerPreview = implode('; ', array_slice($responseHeaders, 0, 8));
|
|
throw new RuntimeException(sprintf(
|
|
'%s %s returned HTTP %d. Headers: %s. Body preview: %s',
|
|
$method,
|
|
$url,
|
|
$status,
|
|
$headerPreview !== '' ? $headerPreview : 'none',
|
|
$preview !== '' ? $preview : 'empty'
|
|
));
|
|
}
|
|
|
|
return [
|
|
'status' => $status,
|
|
'body' => (string)$raw,
|
|
'headers' => $responseHeaders,
|
|
];
|
|
}
|
|
}
|
|
|
|
final class Logger
|
|
{
|
|
/** @var callable|null */
|
|
private $sink = null;
|
|
|
|
public function __construct(private readonly string $logFile)
|
|
{
|
|
}
|
|
|
|
public function setSink(?callable $sink): void
|
|
{
|
|
$this->sink = $sink;
|
|
}
|
|
|
|
public function info(string $message): void
|
|
{
|
|
$this->write('INFO', $message);
|
|
}
|
|
|
|
public function warning(string $message): void
|
|
{
|
|
$this->write('WARNING', $message);
|
|
}
|
|
|
|
public function error(string $message): void
|
|
{
|
|
$this->write('ERROR', $message);
|
|
}
|
|
|
|
private function write(string $level, string $message): void
|
|
{
|
|
$line = '[' . date('c') . "] {$level} {$message}\n";
|
|
file_put_contents($this->logFile, $line, FILE_APPEND);
|
|
fwrite($level === 'ERROR' ? STDERR : STDOUT, $line);
|
|
if ($this->sink !== null) {
|
|
try {
|
|
call_user_func($this->sink, $level, $message);
|
|
} catch (Throwable) {
|
|
// Logging must never break the agent loop.
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
final class LocalStateStore
|
|
{
|
|
private SQLite3 $db;
|
|
|
|
public function __construct(string $path)
|
|
{
|
|
if (!class_exists('SQLite3')) {
|
|
throw new RuntimeException('The edge-agent container requires sqlite3 support.');
|
|
}
|
|
|
|
$directory = dirname($path);
|
|
if (!is_dir($directory)) {
|
|
@mkdir($directory, 0777, true);
|
|
}
|
|
|
|
$this->db = new SQLite3($path);
|
|
$this->db->busyTimeout(5000);
|
|
$this->db->exec('PRAGMA journal_mode = WAL;');
|
|
$this->db->exec('PRAGMA synchronous = NORMAL;');
|
|
$this->db->exec(
|
|
'CREATE TABLE IF NOT EXISTS kv (
|
|
key TEXT PRIMARY KEY,
|
|
value_json TEXT NOT NULL,
|
|
updated_at TEXT NOT NULL
|
|
)'
|
|
);
|
|
$this->db->exec(
|
|
'CREATE TABLE IF NOT EXISTS outbox (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
item_type TEXT NOT NULL,
|
|
endpoint TEXT NOT NULL,
|
|
payload_json TEXT NOT NULL,
|
|
created_at TEXT NOT NULL
|
|
)'
|
|
);
|
|
}
|
|
|
|
public function getJson(string $key, mixed $default = null): mixed
|
|
{
|
|
$statement = $this->db->prepare('SELECT value_json FROM kv WHERE key = :key LIMIT 1');
|
|
$statement->bindValue(':key', $key, SQLITE3_TEXT);
|
|
$result = $statement->execute();
|
|
$row = $result instanceof SQLite3Result ? $result->fetchArray(SQLITE3_ASSOC) : false;
|
|
if (!is_array($row) || !array_key_exists('value_json', $row)) {
|
|
return $default;
|
|
}
|
|
|
|
$decoded = json_decode((string)$row['value_json'], true);
|
|
return json_last_error() === JSON_ERROR_NONE ? $decoded : $default;
|
|
}
|
|
|
|
public function setJson(string $key, mixed $value): void
|
|
{
|
|
$statement = $this->db->prepare(
|
|
'INSERT INTO kv (key, value_json, updated_at)
|
|
VALUES (:key, :value_json, :updated_at)
|
|
ON CONFLICT(key) DO UPDATE SET value_json = excluded.value_json, updated_at = excluded.updated_at'
|
|
);
|
|
$statement->bindValue(':key', $key, SQLITE3_TEXT);
|
|
$statement->bindValue(':value_json', json_encode($value, JSON_UNESCAPED_SLASHES), SQLITE3_TEXT);
|
|
$statement->bindValue(':updated_at', date('c'), SQLITE3_TEXT);
|
|
$statement->execute();
|
|
}
|
|
|
|
public function enqueue(string $type, string $endpoint, array $payload): void
|
|
{
|
|
$statement = $this->db->prepare(
|
|
'INSERT INTO outbox (item_type, endpoint, payload_json, created_at)
|
|
VALUES (:item_type, :endpoint, :payload_json, :created_at)'
|
|
);
|
|
$statement->bindValue(':item_type', $type, SQLITE3_TEXT);
|
|
$statement->bindValue(':endpoint', $endpoint, SQLITE3_TEXT);
|
|
$statement->bindValue(':payload_json', json_encode($payload, JSON_UNESCAPED_SLASHES), SQLITE3_TEXT);
|
|
$statement->bindValue(':created_at', date('c'), SQLITE3_TEXT);
|
|
$statement->execute();
|
|
}
|
|
|
|
/**
|
|
* @return array<int,array<string,mixed>>
|
|
*/
|
|
public function queuedItems(int $limit = 25): array
|
|
{
|
|
$statement = $this->db->prepare(
|
|
'SELECT id, item_type, endpoint, payload_json, created_at
|
|
FROM outbox
|
|
ORDER BY id ASC
|
|
LIMIT :limit'
|
|
);
|
|
$statement->bindValue(':limit', max(1, $limit), SQLITE3_INTEGER);
|
|
$result = $statement->execute();
|
|
$items = [];
|
|
if (!$result instanceof SQLite3Result) {
|
|
return $items;
|
|
}
|
|
|
|
while (($row = $result->fetchArray(SQLITE3_ASSOC)) !== false) {
|
|
$items[] = [
|
|
'id' => (int)$row['id'],
|
|
'type' => (string)$row['item_type'],
|
|
'endpoint' => (string)$row['endpoint'],
|
|
'payload' => json_decode((string)$row['payload_json'], true) ?: [],
|
|
'created_at' => (string)$row['created_at'],
|
|
];
|
|
}
|
|
|
|
return $items;
|
|
}
|
|
|
|
public function removeOutboxItem(int $id): void
|
|
{
|
|
$statement = $this->db->prepare('DELETE FROM outbox WHERE id = :id');
|
|
$statement->bindValue(':id', $id, SQLITE3_INTEGER);
|
|
$statement->execute();
|
|
}
|
|
|
|
public function outboxSummary(): array
|
|
{
|
|
$countResult = $this->db->querySingle('SELECT COUNT(*) FROM outbox');
|
|
$oldestResult = $this->db->querySingle('SELECT created_at FROM outbox ORDER BY id ASC LIMIT 1');
|
|
$lastSyncAt = $this->getJson('last_sync_at');
|
|
|
|
return [
|
|
'queued' => (int)$countResult,
|
|
'oldest_queued_at' => is_string($oldestResult) && $oldestResult !== '' ? $oldestResult : null,
|
|
'last_replayed_at' => is_string($lastSyncAt) && $lastSyncAt !== '' ? $lastSyncAt : null,
|
|
];
|
|
}
|
|
}
|
|
|
|
final class BrokerWebSocketClient
|
|
{
|
|
private const CONNECT_TIMEOUT_SECONDS = 15;
|
|
private const RECONNECT_DELAY_SECONDS = 2;
|
|
|
|
/** @var resource|null */
|
|
private $socket = null;
|
|
private ?string $brokerUrl = null;
|
|
private ?int $gatewayId = null;
|
|
private ?string $agentToken = null;
|
|
private ?string $agentInstanceId = null;
|
|
private string $readBuffer = '';
|
|
private int $reconnectAfterEpoch = 0;
|
|
private array $state = [
|
|
'url' => null,
|
|
'connected' => false,
|
|
'lastError' => null,
|
|
'disconnectReason' => null,
|
|
'lastConnectedAt' => null,
|
|
'lastDisconnectedAt' => null,
|
|
];
|
|
|
|
public function __construct(private readonly Logger $logger)
|
|
{
|
|
}
|
|
|
|
public function configure(?string $brokerUrl, ?int $gatewayId, ?string $agentToken, ?string $agentInstanceId): void
|
|
{
|
|
$normalizedUrl = $this->normalizeBrokerBaseUrl($brokerUrl);
|
|
$gatewayId = $gatewayId !== null && $gatewayId > 0 ? $gatewayId : null;
|
|
$agentToken = trim((string)($agentToken ?? '')) !== '' ? trim((string)$agentToken) : null;
|
|
$agentInstanceId = trim((string)($agentInstanceId ?? '')) !== '' ? trim((string)$agentInstanceId) : null;
|
|
|
|
$didChange = $this->brokerUrl !== $normalizedUrl
|
|
|| $this->gatewayId !== $gatewayId
|
|
|| $this->agentToken !== $agentToken
|
|
|| $this->agentInstanceId !== $agentInstanceId;
|
|
|
|
$this->brokerUrl = $normalizedUrl;
|
|
$this->gatewayId = $gatewayId;
|
|
$this->agentToken = $agentToken;
|
|
$this->agentInstanceId = $agentInstanceId;
|
|
$this->state['url'] = $normalizedUrl;
|
|
|
|
if ($didChange && $this->socket !== null) {
|
|
$this->disconnect('config_changed');
|
|
}
|
|
}
|
|
|
|
public function isConnected(): bool
|
|
{
|
|
return is_resource($this->socket) && ($this->state['connected'] ?? false) === true;
|
|
}
|
|
|
|
public function state(): array
|
|
{
|
|
return $this->state;
|
|
}
|
|
|
|
public function ensureConnected(): bool
|
|
{
|
|
if ($this->isConnected()) {
|
|
return true;
|
|
}
|
|
|
|
if ($this->brokerUrl === null || $this->gatewayId === null || $this->agentToken === null) {
|
|
return false;
|
|
}
|
|
|
|
if (time() < $this->reconnectAfterEpoch) {
|
|
return false;
|
|
}
|
|
|
|
try {
|
|
$this->connect();
|
|
return true;
|
|
} catch (Throwable $throwable) {
|
|
$this->state['connected'] = false;
|
|
$this->state['lastError'] = $throwable->getMessage();
|
|
$this->state['disconnectReason'] = 'connect_failed';
|
|
$this->state['lastDisconnectedAt'] = date('c');
|
|
$this->reconnectAfterEpoch = time() + self::RECONNECT_DELAY_SECONDS;
|
|
return false;
|
|
}
|
|
}
|
|
|
|
public function disconnect(string $reason = 'closed'): void
|
|
{
|
|
if (is_resource($this->socket)) {
|
|
@fclose($this->socket);
|
|
}
|
|
|
|
$this->socket = null;
|
|
$this->readBuffer = '';
|
|
$this->state['connected'] = false;
|
|
$this->state['disconnectReason'] = $reason;
|
|
$this->state['lastDisconnectedAt'] = date('c');
|
|
$this->reconnectAfterEpoch = time() + self::RECONNECT_DELAY_SECONDS;
|
|
}
|
|
|
|
public function send(array $message): bool
|
|
{
|
|
if (!$this->ensureConnected()) {
|
|
return false;
|
|
}
|
|
|
|
try {
|
|
$this->writeFrame(json_encode($message, JSON_UNESCAPED_SLASHES));
|
|
return true;
|
|
} catch (Throwable $throwable) {
|
|
$this->state['lastError'] = $throwable->getMessage();
|
|
$this->disconnect('send_failed');
|
|
return false;
|
|
}
|
|
}
|
|
|
|
/**
|
|
* @return array<int,array<string,mixed>>
|
|
*/
|
|
public function readMessages(int $maxMessages = 16): array
|
|
{
|
|
if (!$this->ensureConnected()) {
|
|
return [];
|
|
}
|
|
|
|
try {
|
|
$this->readAvailableBytes();
|
|
} catch (Throwable $throwable) {
|
|
$this->state['lastError'] = $throwable->getMessage();
|
|
$this->disconnect('read_failed');
|
|
return [];
|
|
}
|
|
|
|
$messages = [];
|
|
for ($index = 0; $index < max(1, $maxMessages); $index++) {
|
|
$frame = $this->extractFrame();
|
|
if ($frame === null) {
|
|
break;
|
|
}
|
|
|
|
$opcode = (int)($frame['opcode'] ?? 0x1);
|
|
$payload = (string)($frame['payload'] ?? '');
|
|
|
|
if ($opcode === 0x8) {
|
|
$this->disconnect('server_closed');
|
|
break;
|
|
}
|
|
|
|
if ($opcode === 0x9) {
|
|
try {
|
|
$this->writeFrame($payload, 0xA);
|
|
} catch (Throwable) {
|
|
$this->disconnect('pong_failed');
|
|
}
|
|
continue;
|
|
}
|
|
|
|
if ($opcode !== 0x1 || trim($payload) === '') {
|
|
continue;
|
|
}
|
|
|
|
$decoded = json_decode($payload, true);
|
|
if (is_array($decoded)) {
|
|
$messages[] = $decoded;
|
|
}
|
|
}
|
|
|
|
return $messages;
|
|
}
|
|
|
|
private function connect(): void
|
|
{
|
|
$parts = $this->buildSocketParts();
|
|
$transport = ($parts['scheme'] ?? 'ws') === 'wss' ? 'ssl' : 'tcp';
|
|
$host = (string)$parts['host'];
|
|
$port = (int)$parts['port'];
|
|
$path = (string)$parts['path'];
|
|
|
|
$socket = @stream_socket_client(
|
|
sprintf('%s://%s:%d', $transport, $host, $port),
|
|
$errno,
|
|
$error,
|
|
self::CONNECT_TIMEOUT_SECONDS,
|
|
STREAM_CLIENT_CONNECT
|
|
);
|
|
if (!is_resource($socket)) {
|
|
throw new RuntimeException(sprintf(
|
|
'Unable to connect to broker %s:%d: %s',
|
|
$host,
|
|
$port,
|
|
trim((string)$error) !== '' ? trim((string)$error) : 'connection refused'
|
|
));
|
|
}
|
|
|
|
stream_set_timeout($socket, self::CONNECT_TIMEOUT_SECONDS);
|
|
stream_set_blocking($socket, true);
|
|
|
|
$key = base64_encode(random_bytes(16));
|
|
$request = implode("\r\n", [
|
|
'GET ' . $path . ' HTTP/1.1',
|
|
'Host: ' . $host . ($this->isDefaultPort($parts) ? '' : ':' . $port),
|
|
'Upgrade: websocket',
|
|
'Connection: Upgrade',
|
|
'Sec-WebSocket-Version: 13',
|
|
'Sec-WebSocket-Key: ' . $key,
|
|
"\r\n",
|
|
]);
|
|
|
|
$written = fwrite($socket, $request);
|
|
if ($written === false || $written < strlen($request)) {
|
|
fclose($socket);
|
|
throw new RuntimeException('Unable to write broker handshake request');
|
|
}
|
|
|
|
$response = '';
|
|
$startedAt = microtime(true);
|
|
while (!str_contains($response, "\r\n\r\n")) {
|
|
$chunk = fread($socket, 2048);
|
|
if ($chunk === false || $chunk === '') {
|
|
if ((microtime(true) - $startedAt) >= self::CONNECT_TIMEOUT_SECONDS) {
|
|
fclose($socket);
|
|
throw new RuntimeException('Timed out while reading broker handshake response');
|
|
}
|
|
usleep(25000);
|
|
continue;
|
|
}
|
|
|
|
$response .= $chunk;
|
|
}
|
|
|
|
if (!preg_match('/\AHTTP\/1\.[01]\s+101\b/i', $response)) {
|
|
fclose($socket);
|
|
throw new RuntimeException('Broker rejected websocket upgrade: ' . trim(strtok($response, "\r\n")));
|
|
}
|
|
|
|
$expectedAccept = base64_encode(
|
|
sha1($key . '258EAFA5-E914-47DA-95CA-C5AB0DC85B11', true)
|
|
);
|
|
if (!preg_match('/Sec-WebSocket-Accept:\s*(.+)\r\n/i', $response, $matches)) {
|
|
fclose($socket);
|
|
throw new RuntimeException('Broker handshake missing Sec-WebSocket-Accept header');
|
|
}
|
|
|
|
if (!hash_equals($expectedAccept, trim((string)$matches[1]))) {
|
|
fclose($socket);
|
|
throw new RuntimeException('Broker handshake validation failed');
|
|
}
|
|
|
|
stream_set_blocking($socket, false);
|
|
stream_set_timeout($socket, 0);
|
|
$this->socket = $socket;
|
|
$this->readBuffer = '';
|
|
$this->reconnectAfterEpoch = 0;
|
|
$this->state['connected'] = true;
|
|
$this->state['lastError'] = null;
|
|
$this->state['disconnectReason'] = null;
|
|
$this->state['lastConnectedAt'] = date('c');
|
|
}
|
|
|
|
private function readAvailableBytes(): void
|
|
{
|
|
if (!is_resource($this->socket)) {
|
|
return;
|
|
}
|
|
|
|
while (true) {
|
|
$chunk = @fread($this->socket, 8192);
|
|
if ($chunk === false) {
|
|
throw new RuntimeException('Unable to read from broker socket');
|
|
}
|
|
|
|
if ($chunk === '') {
|
|
if (feof($this->socket)) {
|
|
$this->disconnect('server_eof');
|
|
}
|
|
break;
|
|
}
|
|
|
|
$this->readBuffer .= $chunk;
|
|
if (strlen($chunk) < 8192) {
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
|
|
private function extractFrame(): ?array
|
|
{
|
|
$length = strlen($this->readBuffer);
|
|
if ($length < 2) {
|
|
return null;
|
|
}
|
|
|
|
$first = ord($this->readBuffer[0]);
|
|
$second = ord($this->readBuffer[1]);
|
|
$opcode = $first & 0x0F;
|
|
$masked = ($second & 0x80) === 0x80;
|
|
$payloadLength = $second & 0x7F;
|
|
$offset = 2;
|
|
|
|
if ($payloadLength === 126) {
|
|
if ($length < 4) {
|
|
return null;
|
|
}
|
|
$payloadLength = unpack('n', substr($this->readBuffer, $offset, 2))[1];
|
|
$offset += 2;
|
|
} elseif ($payloadLength === 127) {
|
|
if ($length < 10) {
|
|
return null;
|
|
}
|
|
$parts = unpack('Nhigh/Nlow', substr($this->readBuffer, $offset, 8));
|
|
$payloadLength = ((int)$parts['high'] << 32) | (int)$parts['low'];
|
|
$offset += 8;
|
|
}
|
|
|
|
$maskingKey = '';
|
|
if ($masked) {
|
|
if ($length < ($offset + 4)) {
|
|
return null;
|
|
}
|
|
$maskingKey = substr($this->readBuffer, $offset, 4);
|
|
$offset += 4;
|
|
}
|
|
|
|
if ($length < ($offset + $payloadLength)) {
|
|
return null;
|
|
}
|
|
|
|
$payload = substr($this->readBuffer, $offset, $payloadLength);
|
|
$this->readBuffer = (string)substr($this->readBuffer, $offset + $payloadLength);
|
|
|
|
if ($masked) {
|
|
$payload = $this->unmaskPayload($payload, $maskingKey);
|
|
}
|
|
|
|
return [
|
|
'opcode' => $opcode,
|
|
'payload' => $payload,
|
|
];
|
|
}
|
|
|
|
private function writeFrame(string $payload, int $opcode = 0x1): void
|
|
{
|
|
if (!is_resource($this->socket)) {
|
|
throw new RuntimeException('Broker socket is not connected');
|
|
}
|
|
|
|
$frame = chr(0x80 | ($opcode & 0x0F));
|
|
$length = strlen($payload);
|
|
if ($length < 126) {
|
|
$frame .= chr(0x80 | $length);
|
|
} elseif ($length <= 0xFFFF) {
|
|
$frame .= chr(0x80 | 126) . pack('n', $length);
|
|
} else {
|
|
$frame .= chr(0x80 | 127) . pack('NN', 0, $length);
|
|
}
|
|
|
|
$mask = random_bytes(4);
|
|
$frame .= $mask . $this->maskPayload($payload, $mask);
|
|
|
|
$written = @fwrite($this->socket, $frame);
|
|
if ($written === false || $written < strlen($frame)) {
|
|
throw new RuntimeException('Unable to write websocket frame to broker');
|
|
}
|
|
}
|
|
|
|
private function buildSocketParts(): array
|
|
{
|
|
$url = $this->brokerUrl;
|
|
if ($url === null || $this->gatewayId === null || $this->agentToken === null) {
|
|
throw new RuntimeException('Broker websocket is not configured');
|
|
}
|
|
|
|
$query = http_build_query([
|
|
'gatewayId' => $this->gatewayId,
|
|
'token' => $this->agentToken,
|
|
'agentInstanceId' => $this->agentInstanceId,
|
|
]);
|
|
$socketUrl = rtrim($url, '/') . '/ws/agent?' . $query;
|
|
$parts = parse_url($socketUrl);
|
|
if (!is_array($parts) || empty($parts['host'])) {
|
|
throw new RuntimeException('Invalid broker websocket URL: ' . $socketUrl);
|
|
}
|
|
|
|
$host = (string)$parts['host'];
|
|
$path = (string)($parts['path'] ?? '/');
|
|
if ($host === 'edge-broker' && ($path === '/edge-broker' || str_starts_with($path, '/edge-broker/'))) {
|
|
$path = substr($path, strlen('/edge-broker'));
|
|
if ($path === '') {
|
|
$path = '/';
|
|
}
|
|
}
|
|
|
|
return [
|
|
'scheme' => strtolower((string)($parts['scheme'] ?? 'ws')),
|
|
'host' => $host,
|
|
'port' => (int)($parts['port'] ?? (((string)($parts['scheme'] ?? 'ws')) === 'wss' ? 443 : 80)),
|
|
'path' => $path
|
|
. (isset($parts['query']) && trim((string)$parts['query']) !== '' ? '?' . $parts['query'] : ''),
|
|
];
|
|
}
|
|
|
|
private function isDefaultPort(array $parts): bool
|
|
{
|
|
$scheme = (string)($parts['scheme'] ?? 'ws');
|
|
$port = (int)($parts['port'] ?? 0);
|
|
return ($scheme === 'wss' && $port === 443) || ($scheme !== 'wss' && $port === 80);
|
|
}
|
|
|
|
private function normalizeBrokerBaseUrl(?string $value): ?string
|
|
{
|
|
$trimmed = rtrim(trim((string)($value ?? '')), '/');
|
|
if ($trimmed === '') {
|
|
return null;
|
|
}
|
|
|
|
if (str_starts_with($trimmed, 'ws://') || str_starts_with($trimmed, 'wss://')) {
|
|
return $trimmed;
|
|
}
|
|
|
|
if (str_starts_with($trimmed, 'https://')) {
|
|
return 'wss://' . substr($trimmed, 8);
|
|
}
|
|
|
|
if (str_starts_with($trimmed, 'http://')) {
|
|
return 'ws://' . substr($trimmed, 7);
|
|
}
|
|
|
|
return 'ws://' . $trimmed;
|
|
}
|
|
|
|
private function maskPayload(string $payload, string $mask): string
|
|
{
|
|
$length = strlen($payload);
|
|
$result = '';
|
|
for ($index = 0; $index < $length; $index++) {
|
|
$result .= chr(ord($payload[$index]) ^ ord($mask[$index % 4]));
|
|
}
|
|
return $result;
|
|
}
|
|
|
|
private function unmaskPayload(string $payload, string $mask): string
|
|
{
|
|
return $this->maskPayload($payload, $mask);
|
|
}
|
|
}
|
|
|
|
final class AgentShellBridge
|
|
{
|
|
/** @var array<string,array<string,mixed>> */
|
|
private array $sessions = [];
|
|
|
|
public function __construct(private readonly string $defaultCwd)
|
|
{
|
|
}
|
|
|
|
public function open(array $payload, callable $send): void
|
|
{
|
|
$sessionId = trim((string)($payload['sessionId'] ?? ''));
|
|
if ($sessionId === '') {
|
|
return;
|
|
}
|
|
|
|
$this->close(['sessionId' => $sessionId], $send);
|
|
|
|
$cwd = trim((string)($payload['cwd'] ?? $this->defaultCwd));
|
|
if ($cwd === '' || !is_dir($cwd)) {
|
|
$cwd = $this->defaultCwd;
|
|
}
|
|
|
|
[$command, $args] = $this->defaultShellCommand(
|
|
trim((string)($payload['shellCommand'] ?? '')),
|
|
isset($payload['shellArgs']) && is_array($payload['shellArgs']) ? (array)$payload['shellArgs'] : []
|
|
);
|
|
|
|
$descriptorSpec = [
|
|
0 => ['pipe', 'r'],
|
|
1 => ['pipe', 'w'],
|
|
2 => ['pipe', 'w'],
|
|
];
|
|
|
|
$process = @proc_open(
|
|
array_merge([$command], $args),
|
|
$descriptorSpec,
|
|
$pipes,
|
|
$cwd,
|
|
array_merge($_ENV, ['TERM' => 'xterm-256color'])
|
|
);
|
|
|
|
if (!is_resource($process)) {
|
|
$send([
|
|
'type' => 'SHELL_OUTPUT',
|
|
'sessionId' => $sessionId,
|
|
'data' => "Failed to start root shell.\r\n",
|
|
]);
|
|
$send([
|
|
'type' => 'SHELL_EXIT',
|
|
'sessionId' => $sessionId,
|
|
'code' => 1,
|
|
]);
|
|
return;
|
|
}
|
|
|
|
foreach ($pipes as $pipe) {
|
|
stream_set_blocking($pipe, false);
|
|
}
|
|
|
|
$this->sessions[$sessionId] = [
|
|
'process' => $process,
|
|
'pipes' => $pipes,
|
|
];
|
|
|
|
$send([
|
|
'type' => 'SHELL_OPENED',
|
|
'sessionId' => $sessionId,
|
|
]);
|
|
$send([
|
|
'type' => 'SHELL_OUTPUT',
|
|
'sessionId' => $sessionId,
|
|
'data' => sprintf(
|
|
"Connected to gateway host shell.\r\nWorking directory: %s\r\nShortcuts: docker ps | systemctl status truckwash-edge-gateway-stack.service | ./gateway-launcher.sh reconcile\r\n\r\n",
|
|
$cwd
|
|
),
|
|
]);
|
|
}
|
|
|
|
public function input(array $payload): void
|
|
{
|
|
$sessionId = trim((string)($payload['sessionId'] ?? ''));
|
|
if ($sessionId === '' || !isset($this->sessions[$sessionId])) {
|
|
return;
|
|
}
|
|
|
|
$stdin = $this->sessions[$sessionId]['pipes'][0] ?? null;
|
|
if (is_resource($stdin)) {
|
|
@fwrite($stdin, (string)($payload['data'] ?? ''));
|
|
}
|
|
}
|
|
|
|
public function resize(array $payload): void
|
|
{
|
|
$sessionId = trim((string)($payload['sessionId'] ?? ''));
|
|
if ($sessionId === '' || !isset($this->sessions[$sessionId])) {
|
|
return;
|
|
}
|
|
|
|
$this->sessions[$sessionId]['cols'] = (int)($payload['cols'] ?? 0);
|
|
$this->sessions[$sessionId]['rows'] = (int)($payload['rows'] ?? 0);
|
|
}
|
|
|
|
public function close(array $payload, callable $send): void
|
|
{
|
|
$sessionId = trim((string)($payload['sessionId'] ?? ''));
|
|
if ($sessionId === '' || !isset($this->sessions[$sessionId])) {
|
|
return;
|
|
}
|
|
|
|
$session = $this->sessions[$sessionId];
|
|
$process = $session['process'] ?? null;
|
|
if (is_resource($process)) {
|
|
@proc_terminate($process);
|
|
}
|
|
|
|
$this->finalizeSession($sessionId, $send);
|
|
}
|
|
|
|
public function pump(callable $send): void
|
|
{
|
|
foreach (array_keys($this->sessions) as $sessionId) {
|
|
$session = $this->sessions[$sessionId];
|
|
$stdout = $session['pipes'][1] ?? null;
|
|
$stderr = $session['pipes'][2] ?? null;
|
|
|
|
foreach ([$stdout, $stderr] as $pipe) {
|
|
if (!is_resource($pipe)) {
|
|
continue;
|
|
}
|
|
|
|
$chunk = stream_get_contents($pipe);
|
|
if ($chunk !== false && $chunk !== '') {
|
|
$send([
|
|
'type' => 'SHELL_OUTPUT',
|
|
'sessionId' => $sessionId,
|
|
'data' => $chunk,
|
|
]);
|
|
}
|
|
}
|
|
|
|
$process = $session['process'] ?? null;
|
|
if (!is_resource($process)) {
|
|
$this->finalizeSession($sessionId, $send);
|
|
continue;
|
|
}
|
|
|
|
$status = proc_get_status($process);
|
|
if (!is_array($status) || ($status['running'] ?? false) === true) {
|
|
continue;
|
|
}
|
|
|
|
$this->finalizeSession($sessionId, $send, (int)($status['exitcode'] ?? 0));
|
|
}
|
|
}
|
|
|
|
public function dispose(callable $send): void
|
|
{
|
|
foreach (array_keys($this->sessions) as $sessionId) {
|
|
$this->close(['sessionId' => $sessionId], $send);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* @return array{0:string,1:array<int,string>}
|
|
*/
|
|
private function defaultShellCommand(string $command, array $args): array
|
|
{
|
|
if ($command !== '') {
|
|
return [$command, array_values(array_map('strval', $args))];
|
|
}
|
|
|
|
if (DIRECTORY_SEPARATOR === '\\') {
|
|
return ['cmd.exe', ['/Q']];
|
|
}
|
|
|
|
return ['/bin/bash', ['-l']];
|
|
}
|
|
|
|
private function finalizeSession(string $sessionId, callable $send, int $exitCode = 0): void
|
|
{
|
|
if (!isset($this->sessions[$sessionId])) {
|
|
return;
|
|
}
|
|
|
|
$session = $this->sessions[$sessionId];
|
|
foreach ((array)($session['pipes'] ?? []) as $pipe) {
|
|
if (is_resource($pipe)) {
|
|
@fclose($pipe);
|
|
}
|
|
}
|
|
|
|
$process = $session['process'] ?? null;
|
|
if (is_resource($process)) {
|
|
$status = proc_get_status($process);
|
|
$exitCode = is_array($status) && isset($status['exitcode']) ? (int)$status['exitcode'] : $exitCode;
|
|
@proc_close($process);
|
|
}
|
|
|
|
unset($this->sessions[$sessionId]);
|
|
$send([
|
|
'type' => 'SHELL_EXIT',
|
|
'sessionId' => $sessionId,
|
|
'code' => $exitCode >= 0 ? $exitCode : 0,
|
|
]);
|
|
}
|
|
}
|
|
|
|
final class TruckwashEdgeAgent
|
|
{
|
|
private const DEFAULT_INSTALL_DIR = '/opt/truckwash-edge-agent';
|
|
private const DEFAULT_RUNTIME_DIR = '/opt/truckwash-edge-agent/runtime';
|
|
private const DEFAULT_STATE_DATABASE = '/opt/truckwash-edge-agent/runtime/gateway-state.sqlite';
|
|
private const DEFAULT_WORKER_BASE_URL = 'http://lan-worker:8090';
|
|
private const DEFAULT_UPDATE_WINDOW = '02:00-04:00';
|
|
private const OPERATION_COMPLETE_TIMEOUT_SECONDS = 120;
|
|
private const OUTBOX_REPLAY_BATCH_LIMIT = 3;
|
|
private const OUTBOX_REPLAY_TIMEOUT_SECONDS = 3;
|
|
private const OUTBOX_OPERATION_COMPLETE_REPLAY_TIMEOUT_SECONDS = 10;
|
|
private const OUTBOX_REPLAY_FAILURE_COOLDOWN_SECONDS = 15;
|
|
private const MACHINE_SIGNAL_TIMEOUT_SECONDS = 3;
|
|
private const BROKER_MESSAGE_PUMP_LIMIT = 12;
|
|
private const BROKER_CONNECTED_COMMAND_POLL_INTERVAL_SECONDS = 1;
|
|
private const LOOP_STALE_AFTER_SECONDS = 30;
|
|
private const CONTROL_PLANE_SYNC_STALE_AFTER_SECONDS = 90;
|
|
private const MAX_RELAY_TOGGLE_AFTER_SECONDS = 5;
|
|
|
|
private AgentConfig $config;
|
|
private HttpJsonClient $http;
|
|
private HttpJsonClient $workerHttp;
|
|
private Logger $logger;
|
|
private LocalStateStore $stateStore;
|
|
private BrokerWebSocketClient $brokerClient;
|
|
private AgentShellBridge $shellBridge;
|
|
private string $installDir;
|
|
private string $runtimeDir;
|
|
private string $statePath;
|
|
private string $lastOperationSnapshotPath;
|
|
private string $lastHeartbeatMarkerPath;
|
|
private string $controlPlaneStatusPath;
|
|
private string $stagedUpdatePath;
|
|
private int $lastHeartbeatAt = 0;
|
|
private int $lastMachineSignalPollAt = 0;
|
|
private int $lastBrokerConnectedCommandPollAt = 0;
|
|
private int $lastMachineSignalMonitorRefreshAt = 0;
|
|
private int $lastOutboxFailureAt = 0;
|
|
private ?array $lastControlPlaneResponse = null;
|
|
private string $agentInstanceId;
|
|
|
|
public function __construct(string $configPath)
|
|
{
|
|
$this->config = AgentConfig::load($configPath);
|
|
$this->installDir = rtrim((string)$this->config->get('installDir', self::DEFAULT_INSTALL_DIR), DIRECTORY_SEPARATOR);
|
|
$this->runtimeDir = rtrim((string)$this->config->get('runtimeDir', self::DEFAULT_RUNTIME_DIR), DIRECTORY_SEPARATOR);
|
|
if (!is_dir($this->runtimeDir)) {
|
|
@mkdir($this->runtimeDir, 0777, true);
|
|
}
|
|
|
|
$this->http = new HttpJsonClient((string)$this->config->get('apiUrl'));
|
|
$this->workerHttp = new HttpJsonClient(
|
|
(string)$this->config->get('workerBaseUrl', self::DEFAULT_WORKER_BASE_URL),
|
|
$this->workerAuthorizationHeaders()
|
|
);
|
|
$this->logger = new Logger($this->runtimeDir . DIRECTORY_SEPARATOR . 'agent.log');
|
|
$this->stateStore = new LocalStateStore((string)$this->config->get('stateDatabasePath', self::DEFAULT_STATE_DATABASE));
|
|
$this->statePath = $this->runtimeDir . DIRECTORY_SEPARATOR . 'current-operation.json';
|
|
$this->lastOperationSnapshotPath = $this->runtimeDir . DIRECTORY_SEPARATOR . 'last-operation.json';
|
|
$this->lastHeartbeatMarkerPath = $this->runtimeDir . DIRECTORY_SEPARATOR . 'last-heartbeat-ok.txt';
|
|
$this->controlPlaneStatusPath = $this->runtimeDir . DIRECTORY_SEPARATOR . 'control-plane-status.json';
|
|
$this->stagedUpdatePath = $this->runtimeDir . DIRECTORY_SEPARATOR . 'staged-update.json';
|
|
$this->agentInstanceId = $this->ensureAgentInstanceId();
|
|
$this->brokerClient = new BrokerWebSocketClient($this->logger);
|
|
$this->shellBridge = new AgentShellBridge($this->installDir);
|
|
$this->logger->setSink(fn(string $level, string $message): bool => $this->emitLogFrame($level, $message));
|
|
$this->configureBrokerClient();
|
|
$this->initializeControlPlaneStatus();
|
|
}
|
|
|
|
public function run(): void
|
|
{
|
|
$this->logger->info('Truckwash compose edge-agent starting.');
|
|
$this->recoverPreviousOperationState();
|
|
|
|
while (true) {
|
|
try {
|
|
$this->touchLoopHeartbeat();
|
|
$this->reloadConfigFromDisk();
|
|
$this->ensureClaimed();
|
|
$this->configureBrokerClient();
|
|
$this->pumpBrokerTransport();
|
|
$this->heartbeat();
|
|
$this->pollMachineStartSignals();
|
|
if ($this->resumePendingOperationCompletion()) {
|
|
$this->pumpBrokerTransport();
|
|
continue;
|
|
}
|
|
$brokerConnected = $this->isBrokerConnected();
|
|
$processedManagementOperation = false;
|
|
if (!$brokerConnected) {
|
|
$processedManagementOperation = $this->processManagementOperation();
|
|
}
|
|
if (!$processedManagementOperation && $this->shouldPollApiCommandQueue($brokerConnected)) {
|
|
$this->processCommandQueue($brokerConnected ? 0 : null);
|
|
}
|
|
$this->pumpBrokerTransport();
|
|
$this->flushOutbox();
|
|
} catch (Throwable $throwable) {
|
|
$this->logger->error($throwable->getMessage());
|
|
sleep(2);
|
|
}
|
|
|
|
usleep(250000);
|
|
}
|
|
}
|
|
|
|
private function ensureClaimed(): void
|
|
{
|
|
if ($this->config->get('gatewayId') && $this->config->get('agentToken')) {
|
|
return;
|
|
}
|
|
|
|
$installedVersion = (string)$this->config->get('installedVersion', 'compose-php-agent-v3');
|
|
try {
|
|
$response = $this->http->post('/edge-agent/claim', [
|
|
'token' => (string)$this->config->get('installToken'),
|
|
'hostname' => gethostname() ?: 'truckwash-edge',
|
|
'installed_version' => $installedVersion,
|
|
'metadata' => [
|
|
'runtime' => 'compose-php',
|
|
'runtime_mode' => 'compose',
|
|
'php_version' => PHP_VERSION,
|
|
'agent_instance_id' => $this->agentInstanceId,
|
|
'update_window' => (string)$this->config->get('updateWindow', self::DEFAULT_UPDATE_WINDOW),
|
|
'container_health' => $this->buildContainerHealth(),
|
|
'outbox_status' => $this->buildOutboxStatus(),
|
|
'last_sync_at' => $this->currentLastSuccessfulSyncAt(),
|
|
'control_plane_status' => $this->buildControlPlaneStatusPayload(),
|
|
'rollback_status' => $this->readRollbackStatus(),
|
|
'staged_version' => $this->currentStagedUpdate(),
|
|
],
|
|
]);
|
|
} catch (Throwable $throwable) {
|
|
$this->recordTransportFailure('Gateway claim failed', $throwable);
|
|
throw $throwable;
|
|
}
|
|
|
|
$payload = $response['data'] ?? [];
|
|
$gateway = is_array($payload['gateway'] ?? null) ? (array)$payload['gateway'] : [];
|
|
$this->config->set('gatewayId', $gateway['id'] ?? null);
|
|
$this->config->set('agentToken', $payload['agent_token'] ?? null);
|
|
if (!empty($payload['broker_url'])) {
|
|
$this->config->set('brokerUrl', (string)$payload['broker_url']);
|
|
}
|
|
$this->config->set('installedVersion', $installedVersion);
|
|
$this->config->set('targetVersion', (string)($gateway['target_version'] ?? $installedVersion));
|
|
$this->config->set('agentInstanceId', $this->agentInstanceId);
|
|
$this->config->save();
|
|
$this->recordSuccessfulSync();
|
|
$this->logger->info('Claimed gateway ' . (string)($gateway['id'] ?? 'unknown') . '.');
|
|
}
|
|
|
|
private function configureBrokerClient(): void
|
|
{
|
|
$this->brokerClient->configure(
|
|
$this->config->get('brokerUrl'),
|
|
(int)$this->config->get('gatewayId', 0),
|
|
$this->config->get('agentToken'),
|
|
$this->agentInstanceId
|
|
);
|
|
}
|
|
|
|
private function isBrokerConnected(): bool
|
|
{
|
|
return $this->brokerClient->isConnected();
|
|
}
|
|
|
|
private function pumpBrokerTransport(): void
|
|
{
|
|
$this->configureBrokerClient();
|
|
$this->brokerClient->ensureConnected();
|
|
$messages = $this->brokerClient->readMessages(self::BROKER_MESSAGE_PUMP_LIMIT);
|
|
foreach ($messages as $message) {
|
|
$this->handleBrokerMessage($message);
|
|
}
|
|
|
|
$this->shellBridge->pump(fn(array $message): bool => $this->sendBrokerMessage($message));
|
|
}
|
|
|
|
private function sendBrokerMessage(array $message): bool
|
|
{
|
|
$sent = $this->brokerClient->send($message);
|
|
if ($sent) {
|
|
$this->recordSuccessfulSync();
|
|
}
|
|
return $sent;
|
|
}
|
|
|
|
private function handleBrokerMessage(array $message): void
|
|
{
|
|
$type = strtoupper(trim((string)($message['type'] ?? '')));
|
|
if ($type === '') {
|
|
return;
|
|
}
|
|
|
|
switch ($type) {
|
|
case 'COMMAND':
|
|
$this->processBrokerCommand($message);
|
|
return;
|
|
|
|
case 'TASK_DISPATCH':
|
|
$taskType = strtoupper(trim((string)($message['taskType'] ?? '')));
|
|
if ($taskType === 'OPERATION' && isset($message['operation']) && is_array($message['operation'])) {
|
|
$this->executeManagementOperation((array)$message['operation']);
|
|
}
|
|
return;
|
|
|
|
case 'TASK_CANCEL':
|
|
$taskType = strtoupper(trim((string)($message['taskType'] ?? '')));
|
|
if ($taskType === 'OPERATION') {
|
|
$operation = isset($message['operation']) && is_array($message['operation']) ? (array)$message['operation'] : [];
|
|
$operationId = (int)($operation['id'] ?? 0);
|
|
if ($operationId > 0) {
|
|
$this->markOperationCancellationRequested($operationId);
|
|
$this->logger->warning('Cancellation requested for operation ' . $operationId . ' via broker.');
|
|
}
|
|
}
|
|
return;
|
|
|
|
case 'OPEN_ROOT_SHELL':
|
|
$this->shellBridge->open(
|
|
isset($message['payload']) && is_array($message['payload']) ? (array)$message['payload'] : [],
|
|
fn(array $payload): bool => $this->sendBrokerMessage($payload)
|
|
);
|
|
return;
|
|
|
|
case 'SHELL_INPUT':
|
|
$this->shellBridge->input(
|
|
isset($message['payload']) && is_array($message['payload']) ? (array)$message['payload'] : []
|
|
);
|
|
return;
|
|
|
|
case 'RESIZE_ROOT_SHELL':
|
|
$this->shellBridge->resize(
|
|
isset($message['payload']) && is_array($message['payload']) ? (array)$message['payload'] : []
|
|
);
|
|
return;
|
|
|
|
case 'CLOSE_ROOT_SHELL':
|
|
$this->shellBridge->close(
|
|
isset($message['payload']) && is_array($message['payload']) ? (array)$message['payload'] : [],
|
|
fn(array $payload): bool => $this->sendBrokerMessage($payload)
|
|
);
|
|
return;
|
|
|
|
default:
|
|
return;
|
|
}
|
|
}
|
|
|
|
private function heartbeat(bool $force = false, array $metadata = []): void
|
|
{
|
|
$interval = (int)$this->config->get('heartbeatIntervalSeconds', 15);
|
|
if (!$force && (time() - $this->lastHeartbeatAt) < max(5, $interval)) {
|
|
return;
|
|
}
|
|
|
|
$gatewayId = (int)$this->config->get('gatewayId');
|
|
if ($gatewayId <= 0) {
|
|
return;
|
|
}
|
|
|
|
$this->recordHeartbeatAttempt();
|
|
$operationState = $this->readOperationState();
|
|
$payload = [
|
|
'agent_token' => (string)$this->config->get('agentToken'),
|
|
'status' => 'ONLINE',
|
|
'hostname' => gethostname() ?: 'truckwash-edge',
|
|
'installed_version' => (string)$this->config->get('installedVersion', 'compose-php-agent-v3'),
|
|
'target_version' => (string)$this->config->get(
|
|
'targetVersion',
|
|
$this->config->get('installedVersion', 'compose-php-agent-v3')
|
|
),
|
|
'metadata' => array_merge([
|
|
'agent_instance_id' => $this->agentInstanceId,
|
|
'runtime' => 'compose-php',
|
|
'runtime_mode' => 'compose',
|
|
'current_operation' => $operationState,
|
|
'system_metrics' => $this->buildSystemMetrics(),
|
|
'container_health' => $this->buildContainerHealth(),
|
|
'outbox_status' => $this->buildOutboxStatus(),
|
|
'last_sync_at' => $this->currentLastSuccessfulSyncAt(),
|
|
'control_plane_status' => $this->buildControlPlaneStatusPayload(),
|
|
'update_window' => (string)$this->config->get('updateWindow', self::DEFAULT_UPDATE_WINDOW),
|
|
'staged_version' => $this->currentStagedUpdate(),
|
|
'rollback_status' => $this->readRollbackStatus(),
|
|
'command_transport' => $this->isBrokerConnected() ? 'BROKER_WS' : 'API_POLLING',
|
|
'broker_connected' => $this->isBrokerConnected(),
|
|
'broker_url' => $this->brokerClient->state()['url'] ?? $this->config->get('brokerUrl'),
|
|
'broker_last_error' => $this->brokerClient->state()['lastError'] ?? null,
|
|
'broker_last_connected_at' => $this->brokerClient->state()['lastConnectedAt'] ?? null,
|
|
'broker_last_disconnected_at' => $this->brokerClient->state()['lastDisconnectedAt'] ?? null,
|
|
'broker_disconnect_reason' => $this->brokerClient->state()['disconnectReason'] ?? null,
|
|
], $metadata),
|
|
];
|
|
|
|
$posted = $this->sendControlPlaneEvent(
|
|
'/edge-agent/gateways/' . $gatewayId . '/heartbeat',
|
|
$payload,
|
|
'heartbeat'
|
|
);
|
|
if (!$posted) {
|
|
return;
|
|
}
|
|
|
|
$heartbeatSucceededAt = date('c');
|
|
$this->applyBrokerUrlFromControlPlaneResponse($this->lastControlPlaneResponse);
|
|
$this->lastHeartbeatAt = time();
|
|
$this->recordSuccessfulSync($heartbeatSucceededAt, [
|
|
'last_heartbeat_success_at' => $heartbeatSucceededAt,
|
|
]);
|
|
file_put_contents($this->lastHeartbeatMarkerPath, json_encode([
|
|
'gateway_id' => $gatewayId,
|
|
'at' => $heartbeatSucceededAt,
|
|
'agent_instance_id' => $this->agentInstanceId,
|
|
], JSON_UNESCAPED_SLASHES) . PHP_EOL);
|
|
}
|
|
|
|
private function pollMachineStartSignals(): void
|
|
{
|
|
$interval = max(1, (int)$this->config->get('machineSignalPollIntervalSeconds', 2));
|
|
if ((time() - $this->lastMachineSignalPollAt) < $interval) {
|
|
return;
|
|
}
|
|
$this->lastMachineSignalPollAt = time();
|
|
|
|
$gatewayId = (int)$this->config->get('gatewayId');
|
|
$agentToken = (string)$this->config->get('agentToken');
|
|
if ($gatewayId <= 0 || trim($agentToken) === '') {
|
|
return;
|
|
}
|
|
|
|
foreach ($this->machineSignalMonitors($gatewayId, $agentToken) as $monitor) {
|
|
$localIp = trim((string)($monitor['local_ip'] ?? ''));
|
|
if ($localIp === '') {
|
|
continue;
|
|
}
|
|
|
|
try {
|
|
$component = strtolower(trim((string)($monitor['component'] ?? 'input'))) === 'switch' ? 'switch' : 'input';
|
|
$channel = (int)($monitor['channel'] ?? 0);
|
|
$status = $this->machineSignalMonitorStatus($localIp, $channel, $component);
|
|
$on = $this->machineSignalOnState($status, $component);
|
|
if ($on === null) {
|
|
continue;
|
|
}
|
|
|
|
$stateKey = sprintf(
|
|
'selfserve_machine_signal:%s:%s:%d',
|
|
preg_replace('/[^A-Za-z0-9_\-:.]+/', '_', (string)($monitor['relay_id'] ?? 'relay')),
|
|
$component,
|
|
$channel
|
|
);
|
|
$previous = $this->stateStore->getJson($stateKey, null);
|
|
$previousOn = is_array($previous) && array_key_exists('on', $previous) ? (bool)$previous['on'] : false;
|
|
|
|
if ($on && !$previousOn) {
|
|
$event = $component === 'switch' ? 'switch.on' : 'input.toggle_on';
|
|
$payload = [
|
|
'agent_token' => $agentToken,
|
|
'agent_instance_id' => $this->agentInstanceId,
|
|
'lane_id' => (int)($monitor['lane_id'] ?? 0),
|
|
'relay_id' => (string)($monitor['relay_id'] ?? ''),
|
|
'device_id' => (string)($monitor['device_id'] ?? ''),
|
|
'component' => $component,
|
|
'channel' => $channel,
|
|
'event' => $event,
|
|
'source' => 'edge_gateway_poll',
|
|
'status' => $status,
|
|
];
|
|
$payload[$component === 'switch' ? 'output' : 'state'] = true;
|
|
|
|
$this->sendControlPlaneEvent(
|
|
'/edge-agent/gateways/' . $gatewayId . '/selfserve/machine-signal',
|
|
$payload,
|
|
'machine_signal',
|
|
self::MACHINE_SIGNAL_TIMEOUT_SECONDS
|
|
);
|
|
}
|
|
|
|
$this->stateStore->setJson($stateKey, [
|
|
'on' => $on,
|
|
'component' => $component,
|
|
'channel' => $channel,
|
|
'relay_id' => (string)($monitor['relay_id'] ?? ''),
|
|
'updated_at' => date('c'),
|
|
]);
|
|
} catch (Throwable $throwable) {
|
|
$this->logger->warning('Machine start signal poll failed: ' . $throwable->getMessage());
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* @return array<int,array<string,mixed>>
|
|
*/
|
|
private function machineSignalMonitors(int $gatewayId, string $agentToken): array
|
|
{
|
|
$cache = $this->stateStore->getJson('selfserve_machine_signal_monitors', []);
|
|
if (
|
|
is_array($cache)
|
|
&& isset($cache['monitors'], $cache['refreshed_at'])
|
|
&& is_array($cache['monitors'])
|
|
&& (time() - (int)$cache['refreshed_at']) < 60
|
|
) {
|
|
return array_values(array_filter($cache['monitors'], 'is_array'));
|
|
}
|
|
|
|
try {
|
|
$response = $this->http->post('/edge-agent/gateways/' . $gatewayId . '/selfserve/machine-signal-bindings', [
|
|
'agent_token' => $agentToken,
|
|
'agent_instance_id' => $this->agentInstanceId,
|
|
], 10);
|
|
$payload = is_array($response['data'] ?? null) ? (array)$response['data'] : (is_array($response) ? $response : []);
|
|
$monitors = is_array($payload['monitors'] ?? null) ? array_values(array_filter((array)$payload['monitors'], 'is_array')) : [];
|
|
$this->lastMachineSignalMonitorRefreshAt = time();
|
|
$this->stateStore->setJson('selfserve_machine_signal_monitors', [
|
|
'refreshed_at' => $this->lastMachineSignalMonitorRefreshAt,
|
|
'monitors' => $monitors,
|
|
]);
|
|
|
|
return $monitors;
|
|
} catch (Throwable $throwable) {
|
|
if (is_array($cache) && isset($cache['monitors']) && is_array($cache['monitors'])) {
|
|
return array_values(array_filter($cache['monitors'], 'is_array'));
|
|
}
|
|
$this->logger->warning('Machine start signal monitor refresh failed: ' . $throwable->getMessage());
|
|
return [];
|
|
}
|
|
}
|
|
|
|
private function machineSignalMonitorStatus(string $localIp, int $channel, string $component): array
|
|
{
|
|
if ($component === 'input') {
|
|
try {
|
|
$input = $this->workerHttp->post('/relay/input-status', [
|
|
'local_ip' => $localIp,
|
|
'channel' => $channel,
|
|
], 5) ?? [];
|
|
return [
|
|
'input_state' => $input['state'] ?? null,
|
|
'input' => $input,
|
|
];
|
|
} catch (Throwable) {
|
|
// Fall back to the combined status endpoint below.
|
|
}
|
|
}
|
|
|
|
return $this->workerHttp->post('/relay/status', [
|
|
'local_ip' => $localIp,
|
|
'channel' => $channel,
|
|
'include_input' => $component === 'input',
|
|
], 8) ?? [];
|
|
}
|
|
|
|
private function machineSignalOnState(array $status, string $component): ?bool
|
|
{
|
|
if ($component === 'switch') {
|
|
if (array_key_exists('output', $status)) {
|
|
return (bool)$status['output'];
|
|
}
|
|
if (array_key_exists('on', $status)) {
|
|
return (bool)$status['on'];
|
|
}
|
|
if (isset($status['raw']) && is_array($status['raw']) && array_key_exists('output', $status['raw'])) {
|
|
return (bool)$status['raw']['output'];
|
|
}
|
|
|
|
return null;
|
|
}
|
|
|
|
if (array_key_exists('input_state', $status)) {
|
|
return $status['input_state'] === null ? null : (bool)$status['input_state'];
|
|
}
|
|
if (isset($status['input']) && is_array($status['input']) && array_key_exists('state', $status['input'])) {
|
|
return (bool)$status['input']['state'];
|
|
}
|
|
if (array_key_exists('state', $status)) {
|
|
return (bool)$status['state'];
|
|
}
|
|
|
|
return null;
|
|
}
|
|
|
|
private function shouldPollApiCommandQueue(bool $brokerConnected): bool
|
|
{
|
|
if (!$brokerConnected) {
|
|
return true;
|
|
}
|
|
|
|
$now = time();
|
|
if (($now - $this->lastBrokerConnectedCommandPollAt) < self::BROKER_CONNECTED_COMMAND_POLL_INTERVAL_SECONDS) {
|
|
return false;
|
|
}
|
|
|
|
$this->lastBrokerConnectedCommandPollAt = $now;
|
|
return true;
|
|
}
|
|
|
|
private function processCommandQueue(?int $waitSecondsOverride = null): void
|
|
{
|
|
$gatewayId = (int)$this->config->get('gatewayId');
|
|
if ($gatewayId <= 0) {
|
|
return;
|
|
}
|
|
|
|
$waitSeconds = $waitSecondsOverride ?? (int)$this->config->get('operationPollTimeoutSeconds', 20);
|
|
try {
|
|
$response = $this->http->post('/edge-agent/gateways/' . $gatewayId . '/commands/poll', [
|
|
'agent_token' => (string)$this->config->get('agentToken'),
|
|
'wait_seconds' => $waitSeconds,
|
|
], $this->pollRequestTimeoutSeconds($waitSeconds));
|
|
} catch (HttpRequestTimeoutException) {
|
|
return;
|
|
} catch (Throwable $throwable) {
|
|
$this->logger->warning('Command polling failed: ' . $throwable->getMessage());
|
|
return;
|
|
}
|
|
|
|
$command = $response['data'] ?? null;
|
|
if (!is_array($command) || empty($command['id'])) {
|
|
return;
|
|
}
|
|
|
|
$jobId = (int)$command['id'];
|
|
$type = (string)($command['command_type'] ?? $command['commandType'] ?? '');
|
|
$request = is_array($command['payload'] ?? null) ? (array)$command['payload'] : [];
|
|
|
|
try {
|
|
$result = $this->executeEdgeCommand($type, $request);
|
|
|
|
$this->sendControlPlaneEvent(
|
|
'/edge-agent/gateways/' . $gatewayId . '/commands/' . $jobId . '/result',
|
|
[
|
|
'agent_token' => (string)$this->config->get('agentToken'),
|
|
'ok' => true,
|
|
'result' => $result,
|
|
],
|
|
'command_result'
|
|
);
|
|
} catch (Throwable $throwable) {
|
|
$this->sendControlPlaneEvent(
|
|
'/edge-agent/gateways/' . $gatewayId . '/commands/' . $jobId . '/result',
|
|
[
|
|
'agent_token' => (string)$this->config->get('agentToken'),
|
|
'ok' => false,
|
|
'error' => $throwable->getMessage(),
|
|
],
|
|
'command_result'
|
|
);
|
|
}
|
|
}
|
|
|
|
private function processBrokerCommand(array $message): void
|
|
{
|
|
$commandId = trim((string)($message['commandId'] ?? ''));
|
|
$type = (string)($message['commandType'] ?? '');
|
|
$request = isset($message['payload']) && is_array($message['payload']) ? (array)$message['payload'] : [];
|
|
if ($commandId === '') {
|
|
return;
|
|
}
|
|
|
|
try {
|
|
$result = $this->executeEdgeCommand($type, $request);
|
|
$this->sendBrokerMessage([
|
|
'type' => 'COMMAND_RESULT',
|
|
'commandId' => $commandId,
|
|
'ok' => true,
|
|
'payload' => $result,
|
|
]);
|
|
} catch (Throwable $throwable) {
|
|
$this->sendBrokerMessage([
|
|
'type' => 'COMMAND_RESULT',
|
|
'commandId' => $commandId,
|
|
'ok' => false,
|
|
'error' => $throwable->getMessage(),
|
|
]);
|
|
}
|
|
}
|
|
|
|
private function executeEdgeCommand(string $type, array $request): array
|
|
{
|
|
return match (strtoupper(trim($type))) {
|
|
'DISCOVER_SHELLY' => $this->runDiscovery(),
|
|
'GET_RELAY_STATUS' => $this->readRelayStatus($request),
|
|
'SET_RELAY_STATE' => $this->switchRelay($request),
|
|
default => throw new RuntimeException('Unsupported command type: ' . $type),
|
|
};
|
|
}
|
|
|
|
private function processManagementOperation(): bool
|
|
{
|
|
$gatewayId = (int)$this->config->get('gatewayId');
|
|
if ($gatewayId <= 0) {
|
|
return false;
|
|
}
|
|
|
|
$waitSeconds = (int)$this->config->get('operationPollTimeoutSeconds', 20);
|
|
try {
|
|
$response = $this->http->post('/edge-agent/gateways/' . $gatewayId . '/operations/next', [
|
|
'agent_token' => (string)$this->config->get('agentToken'),
|
|
'wait_seconds' => $waitSeconds,
|
|
'agent_instance_id' => $this->agentInstanceId,
|
|
], $this->pollRequestTimeoutSeconds($waitSeconds));
|
|
} catch (HttpRequestTimeoutException) {
|
|
return false;
|
|
} catch (Throwable $throwable) {
|
|
$this->logger->warning('Operation polling failed: ' . $throwable->getMessage());
|
|
return false;
|
|
}
|
|
|
|
$operation = $response['data'] ?? null;
|
|
if (!is_array($operation) || empty($operation['id'])) {
|
|
return false;
|
|
}
|
|
|
|
return $this->executeManagementOperation((array)$operation);
|
|
}
|
|
|
|
private function executeManagementOperation(array $operation): bool
|
|
{
|
|
$gatewayId = (int)$this->config->get('gatewayId');
|
|
if ($gatewayId <= 0) {
|
|
return false;
|
|
}
|
|
|
|
$operationId = (int)($operation['id'] ?? 0);
|
|
$type = (string)($operation['type'] ?? '');
|
|
$request = is_array($operation['request'] ?? null) ? (array)$operation['request'] : [];
|
|
if ($operationId <= 0 || trim($type) === '') {
|
|
return false;
|
|
}
|
|
|
|
$this->persistOperationState([
|
|
'gateway_id' => $gatewayId,
|
|
'operation_id' => $operationId,
|
|
'type' => $type,
|
|
'status' => 'IN_PROGRESS',
|
|
'stage' => 'claimed',
|
|
'progress' => 5,
|
|
'agent_instance_id' => $this->agentInstanceId,
|
|
'started_at' => date('c'),
|
|
'request' => $request,
|
|
]);
|
|
|
|
$this->emitOperationProgress(
|
|
$gatewayId,
|
|
$operationId,
|
|
'OPERATION_AGENT_STARTED',
|
|
'Compose edge-agent started processing the operation',
|
|
10,
|
|
['type' => $type, 'agent_instance_id' => $this->agentInstanceId],
|
|
'starting'
|
|
);
|
|
|
|
try {
|
|
$result = match ($type) {
|
|
'DISCOVERY' => $this->runManagementDiscovery($gatewayId, $operationId, $request),
|
|
'UPDATE' => $this->runUpdate($gatewayId, $operationId, $request),
|
|
'UNINSTALL' => $this->runUninstall($gatewayId, $operationId, $request),
|
|
default => throw new RuntimeException('Unsupported operation type: ' . $type),
|
|
};
|
|
$this->abortIfOperationCancelled($operationId);
|
|
|
|
$this->finalizeOperationCompletion(
|
|
$gatewayId,
|
|
$operationId,
|
|
[
|
|
'agent_token' => (string)$this->config->get('agentToken'),
|
|
'ok' => true,
|
|
'result' => $result,
|
|
],
|
|
'COMPLETED',
|
|
['result' => $result]
|
|
);
|
|
} catch (Throwable $throwable) {
|
|
$errorCode = $throwable instanceof OperationAbortException && $throwable->isCancellation()
|
|
? 'EDGE_GATEWAY_CANCELLED'
|
|
: $this->classifyManagementError($throwable, $type);
|
|
if (!($throwable instanceof OperationAbortException)) {
|
|
$this->postOperationEvent($gatewayId, $operationId, [
|
|
'level' => 'ERROR',
|
|
'code' => $errorCode,
|
|
'message' => $throwable->getMessage(),
|
|
'context' => ['type' => $type, 'progress' => 100, 'agent_instance_id' => $this->agentInstanceId],
|
|
]);
|
|
}
|
|
|
|
$this->finalizeOperationCompletion(
|
|
$gatewayId,
|
|
$operationId,
|
|
[
|
|
'agent_token' => (string)$this->config->get('agentToken'),
|
|
'ok' => false,
|
|
'error_code' => $errorCode,
|
|
'error_message' => $throwable->getMessage(),
|
|
],
|
|
'FAILED',
|
|
[
|
|
'error_code' => $errorCode,
|
|
'error_message' => $throwable->getMessage(),
|
|
]
|
|
);
|
|
}
|
|
|
|
$this->clearOperationCancellationRequested($operationId);
|
|
return true;
|
|
}
|
|
|
|
private function postOperationEvent(int $gatewayId, int $operationId, array $payload): ?array
|
|
{
|
|
return $this->requestControlPlaneEvent(
|
|
'/edge-agent/gateways/' . $gatewayId . '/operations/' . $operationId . '/events',
|
|
[
|
|
'agent_token' => (string)$this->config->get('agentToken'),
|
|
'level' => (string)($payload['level'] ?? 'INFO'),
|
|
'code' => $payload['code'] ?? null,
|
|
'message' => (string)($payload['message'] ?? 'Operation event'),
|
|
'context' => is_array($payload['context'] ?? null) ? (array)$payload['context'] : [],
|
|
],
|
|
'operation_event'
|
|
);
|
|
}
|
|
|
|
private function emitOperationProgress(
|
|
int $gatewayId,
|
|
int $operationId,
|
|
string $code,
|
|
string $message,
|
|
int $progress,
|
|
array $context = [],
|
|
?string $stage = null
|
|
): void {
|
|
$this->abortIfOperationCancelled($operationId);
|
|
|
|
$operationState = $this->readOperationState();
|
|
if ($operationState !== null) {
|
|
$operationState['progress'] = max(0, min(100, $progress));
|
|
$operationState['stage'] = $stage ?? ($operationState['stage'] ?? 'running');
|
|
$operationState['updated_at'] = date('c');
|
|
$this->persistOperationState($operationState);
|
|
}
|
|
|
|
$this->heartbeat(true, [
|
|
'current_operation' => $this->readOperationState(),
|
|
]);
|
|
|
|
$response = $this->postOperationEvent($gatewayId, $operationId, [
|
|
'level' => 'INFO',
|
|
'code' => $code,
|
|
'message' => $message,
|
|
'context' => array_merge($context, [
|
|
'progress' => max(0, min(100, $progress)),
|
|
'agent_instance_id' => $this->agentInstanceId,
|
|
]),
|
|
]);
|
|
$this->guardOperationEventResponse($response);
|
|
}
|
|
|
|
private function runDiscovery(): array
|
|
{
|
|
try {
|
|
$response = $this->workerHttp->post('/discover', [
|
|
'hostname' => gethostname() ?: 'truckwash-edge',
|
|
'gateway_id' => (int)$this->config->get('gatewayId', 0),
|
|
], 20);
|
|
$inventory = isset($response['inventory']) && is_array($response['inventory']) ? (array)$response['inventory'] : [];
|
|
return ['inventory' => $inventory];
|
|
} catch (Throwable) {
|
|
$hostname = gethostname() ?: 'truckwash-edge';
|
|
return ['inventory' => [[
|
|
'device_id' => 'gateway-runtime-' . substr(sha1($hostname), 0, 10),
|
|
'local_ip' => gethostbyname($hostname),
|
|
'model' => 'TruckWash Edge Gateway',
|
|
'channel_count' => 1,
|
|
'online' => true,
|
|
'capabilities' => [
|
|
'local_discovery' => true,
|
|
'relay_commands' => true,
|
|
'gateway_management_v2' => true,
|
|
],
|
|
'metadata' => [
|
|
'hostname' => $hostname,
|
|
'runtime' => 'compose-php',
|
|
'php_version' => PHP_VERSION,
|
|
],
|
|
]]];
|
|
}
|
|
}
|
|
|
|
private function runManagementDiscovery(int $gatewayId, int $operationId, array $request): array
|
|
{
|
|
$this->emitOperationProgress(
|
|
$gatewayId,
|
|
$operationId,
|
|
'DISCOVERY_COLLECTING',
|
|
'Collecting local gateway inventory from lan-worker',
|
|
35,
|
|
[],
|
|
'discovering'
|
|
);
|
|
$inventory = isset($request['inventory']) && is_array($request['inventory']) && $request['inventory'] !== []
|
|
? (array)$request['inventory']
|
|
: (array)($this->runDiscovery()['inventory'] ?? []);
|
|
|
|
$this->emitOperationProgress($gatewayId, $operationId, 'DISCOVERY_COMPLETED', 'Inventory collected', 90, [
|
|
'device_count' => count($inventory),
|
|
], 'finishing');
|
|
|
|
return ['inventory' => $inventory];
|
|
}
|
|
|
|
private function runUpdate(int $gatewayId, int $operationId, array $request): array
|
|
{
|
|
$targetVersion = trim((string)($request['target_version'] ?? $request['targetVersion'] ?? ''));
|
|
if ($targetVersion === '') {
|
|
throw new RuntimeException('Update request is missing target version');
|
|
}
|
|
|
|
$updateWindow = trim((string)($request['updateWindow'] ?? $this->config->get('updateWindow', self::DEFAULT_UPDATE_WINDOW)));
|
|
if ($updateWindow === '') {
|
|
$updateWindow = self::DEFAULT_UPDATE_WINDOW;
|
|
}
|
|
|
|
$requiredArtifacts = [
|
|
[
|
|
'url' => (string)($request['artifactUrl'] ?? ''),
|
|
'sha256' => (string)($request['artifactSha256'] ?? ''),
|
|
'path' => $this->installDir . DIRECTORY_SEPARATOR . 'agent.php',
|
|
'label' => 'edge-agent runtime',
|
|
'progress' => 35,
|
|
'code' => 'UPDATE_DOWNLOAD_AGENT',
|
|
],
|
|
[
|
|
'url' => (string)($request['lanWorkerArtifactUrl'] ?? ''),
|
|
'sha256' => (string)($request['lanWorkerArtifactSha256'] ?? ''),
|
|
'path' => $this->installDir . DIRECTORY_SEPARATOR . 'lan-worker.php',
|
|
'label' => 'lan-worker runtime',
|
|
'progress' => 45,
|
|
'code' => 'UPDATE_DOWNLOAD_WORKER',
|
|
],
|
|
[
|
|
'url' => (string)($request['autoUpdaterArtifactUrl'] ?? ''),
|
|
'sha256' => (string)($request['autoUpdaterArtifactSha256'] ?? ''),
|
|
'path' => $this->installDir . DIRECTORY_SEPARATOR . 'auto-updater.php',
|
|
'label' => 'auto-updater runtime',
|
|
'progress' => 50,
|
|
'code' => 'UPDATE_DOWNLOAD_AUTO_UPDATER',
|
|
],
|
|
[
|
|
'url' => (string)($request['composeFileUrl'] ?? ''),
|
|
'sha256' => (string)($request['composeFileSha256'] ?? ''),
|
|
'path' => $this->installDir . DIRECTORY_SEPARATOR . 'docker-compose.gateway.yml',
|
|
'label' => 'compose stack',
|
|
'progress' => 55,
|
|
'code' => 'UPDATE_DOWNLOAD_COMPOSE',
|
|
],
|
|
[
|
|
'url' => (string)($request['edgeAgentDockerfileUrl'] ?? ''),
|
|
'sha256' => (string)($request['edgeAgentDockerfileSha256'] ?? ''),
|
|
'path' => $this->installDir . DIRECTORY_SEPARATOR . 'Dockerfile.edge-agent',
|
|
'label' => 'edge-agent Dockerfile',
|
|
'progress' => 60,
|
|
'code' => 'UPDATE_DOWNLOAD_EDGE_DOCKERFILE',
|
|
],
|
|
[
|
|
'url' => (string)($request['lanWorkerDockerfileUrl'] ?? ''),
|
|
'sha256' => (string)($request['lanWorkerDockerfileSha256'] ?? ''),
|
|
'path' => $this->installDir . DIRECTORY_SEPARATOR . 'Dockerfile.lan-worker',
|
|
'label' => 'lan-worker Dockerfile',
|
|
'progress' => 65,
|
|
'code' => 'UPDATE_DOWNLOAD_WORKER_DOCKERFILE',
|
|
],
|
|
[
|
|
'url' => (string)($request['autoUpdaterDockerfileUrl'] ?? ''),
|
|
'sha256' => (string)($request['autoUpdaterDockerfileSha256'] ?? ''),
|
|
'path' => $this->installDir . DIRECTORY_SEPARATOR . 'Dockerfile.auto-updater',
|
|
'label' => 'auto-updater Dockerfile',
|
|
'progress' => 68,
|
|
'code' => 'UPDATE_DOWNLOAD_AUTO_UPDATER_DOCKERFILE',
|
|
],
|
|
[
|
|
'url' => (string)($request['launcherScriptUrl'] ?? ''),
|
|
'sha256' => (string)($request['launcherScriptSha256'] ?? ''),
|
|
'path' => $this->installDir . DIRECTORY_SEPARATOR . 'gateway-launcher.sh',
|
|
'label' => 'gateway launcher',
|
|
'progress' => 72,
|
|
'code' => 'UPDATE_DOWNLOAD_LAUNCHER',
|
|
],
|
|
[
|
|
'url' => (string)($request['stackServiceUnitUrl'] ?? ''),
|
|
'sha256' => (string)($request['stackServiceUnitSha256'] ?? ''),
|
|
'path' => $this->installDir . DIRECTORY_SEPARATOR . 'truckwash-edge-gateway-stack.service',
|
|
'label' => 'compose systemd unit',
|
|
'progress' => 80,
|
|
'code' => 'UPDATE_DOWNLOAD_STACK_SERVICE',
|
|
],
|
|
[
|
|
'url' => (string)($request['serviceUnitUrl'] ?? ''),
|
|
'sha256' => (string)($request['serviceUnitSha256'] ?? ''),
|
|
'path' => $this->installDir . DIRECTORY_SEPARATOR . 'truckwash-edge-agent.service',
|
|
'label' => 'compatibility systemd unit',
|
|
'progress' => 84,
|
|
'code' => 'UPDATE_DOWNLOAD_LEGACY_SERVICE',
|
|
],
|
|
];
|
|
|
|
foreach ([
|
|
'composeFileUrl',
|
|
'launcherScriptUrl',
|
|
'stackServiceUnitUrl',
|
|
'autoUpdaterArtifactUrl',
|
|
'autoUpdaterDockerfileUrl',
|
|
] as $requiredKey) {
|
|
if (trim((string)($request[$requiredKey] ?? '')) === '') {
|
|
throw new RuntimeException('Update request is missing compose artifact ' . $requiredKey);
|
|
}
|
|
}
|
|
|
|
$this->emitOperationProgress($gatewayId, $operationId, 'UPDATE_VALIDATED', 'Validated compose update request', 20, [
|
|
'target_version' => $targetVersion,
|
|
'update_window' => $updateWindow,
|
|
], 'validating');
|
|
|
|
$changedFiles = [];
|
|
foreach ($requiredArtifacts as $artifact) {
|
|
if ($artifact['url'] === '') {
|
|
continue;
|
|
}
|
|
|
|
$this->emitOperationProgress(
|
|
$gatewayId,
|
|
$operationId,
|
|
(string)$artifact['code'],
|
|
'Downloading ' . (string)$artifact['label'],
|
|
(int)$artifact['progress'],
|
|
['url' => (string)$artifact['url']],
|
|
'downloading'
|
|
);
|
|
$changedFiles[] = $this->downloadToFileAtomic(
|
|
(string)$artifact['url'],
|
|
(string)$artifact['path'],
|
|
(string)$artifact['sha256']
|
|
);
|
|
}
|
|
|
|
$this->ensureExecutable($this->installDir . DIRECTORY_SEPARATOR . 'agent.php');
|
|
$this->ensureExecutable($this->installDir . DIRECTORY_SEPARATOR . 'lan-worker.php');
|
|
$this->ensureExecutable($this->installDir . DIRECTORY_SEPARATOR . 'auto-updater.php');
|
|
$this->ensureExecutable($this->installDir . DIRECTORY_SEPARATOR . 'gateway-launcher.sh');
|
|
|
|
$stagedAt = date('c');
|
|
$stagedUpdate = [
|
|
'target_version' => $targetVersion,
|
|
'staged_at' => $stagedAt,
|
|
'apply_after' => $this->nextUpdateWindowStartIso($updateWindow),
|
|
'status' => 'STAGED',
|
|
'update_window' => $updateWindow,
|
|
'compose_project_name' => (string)($request['composeProjectName'] ?? $this->config->get('composeProjectName', 'truckwash-edge-gateway')),
|
|
'stack_service_name' => (string)($request['stackServiceName'] ?? $this->config->get('stackServiceName', 'truckwash-edge-gateway-stack.service')),
|
|
'changed_files' => $changedFiles,
|
|
];
|
|
|
|
$this->emitOperationProgress($gatewayId, $operationId, 'UPDATE_STAGE_METADATA', 'Writing staged update metadata', 90, [], 'writing-config');
|
|
$this->config->set('targetVersion', $targetVersion);
|
|
$this->config->set('lastStagedUpdate', $stagedUpdate);
|
|
$this->config->save();
|
|
$this->stateStore->setJson('staged_update', $stagedUpdate);
|
|
$this->stateStore->setJson('rollback_status', [
|
|
'state' => 'IDLE',
|
|
'reason' => null,
|
|
'rolled_back_to' => null,
|
|
'at' => null,
|
|
]);
|
|
file_put_contents($this->stagedUpdatePath, json_encode($stagedUpdate, JSON_PRETTY_PRINT | JSON_UNESCAPED_SLASHES) . PHP_EOL);
|
|
|
|
$this->emitOperationProgress($gatewayId, $operationId, 'UPDATE_STAGED', 'Compose rollout staged for the maintenance window', 96, [
|
|
'target_version' => $targetVersion,
|
|
'apply_after' => $stagedUpdate['apply_after'],
|
|
], 'staged');
|
|
|
|
return [
|
|
'applied' => false,
|
|
'installed_version' => (string)$this->config->get('installedVersion', 'compose-php-agent-v3'),
|
|
'staged_version' => $targetVersion,
|
|
'target_version' => $targetVersion,
|
|
'staged_at' => $stagedAt,
|
|
'apply_after' => $stagedUpdate['apply_after'],
|
|
'update_window' => $updateWindow,
|
|
'restart_required' => true,
|
|
'service_name' => (string)($request['stackServiceName'] ?? $this->config->get('stackServiceName', 'truckwash-edge-gateway-stack.service')),
|
|
'changed_files' => $changedFiles,
|
|
'agent_instance_id' => $this->agentInstanceId,
|
|
'rollback_status' => $this->readRollbackStatus(),
|
|
];
|
|
}
|
|
|
|
private function runUninstall(int $gatewayId, int $operationId, array $request): array
|
|
{
|
|
$this->emitOperationProgress($gatewayId, $operationId, 'UNINSTALL_PREPARE', 'Preparing uninstall manifest', 40, [], 'preparing-uninstall');
|
|
$manifestPath = $this->runtimeDir . DIRECTORY_SEPARATOR . 'uninstall-plan.json';
|
|
$manifest = [
|
|
'service_name' => (string)($request['service_name'] ?? $this->config->get('stackServiceName', 'truckwash-edge-gateway-stack.service')),
|
|
'install_dir' => (string)($request['install_dir'] ?? $this->installDir),
|
|
'generated_at' => date('c'),
|
|
'agent_instance_id' => $this->agentInstanceId,
|
|
];
|
|
file_put_contents($manifestPath, json_encode($manifest, JSON_PRETTY_PRINT | JSON_UNESCAPED_SLASHES) . PHP_EOL);
|
|
file_put_contents($this->runtimeDir . DIRECTORY_SEPARATOR . 'uninstall.flag', date('c') . PHP_EOL);
|
|
$this->config->set('pendingUninstall', $manifest);
|
|
$this->config->save();
|
|
|
|
$this->emitOperationProgress($gatewayId, $operationId, 'UNINSTALL_READY', 'Gateway marked for controlled uninstall', 90, [
|
|
'manifest_path' => $manifestPath,
|
|
], 'uninstall-ready');
|
|
|
|
return [
|
|
'uninstalled' => true,
|
|
'service_name' => $manifest['service_name'],
|
|
'install_dir' => $manifest['install_dir'],
|
|
'manual_cleanup_required' => true,
|
|
'cleanup_manifest_path' => $manifestPath,
|
|
];
|
|
}
|
|
|
|
private function readRelayStatus(array $request): array
|
|
{
|
|
$localIp = (string)($request['localIp'] ?? $request['local_ip'] ?? '');
|
|
$channel = (int)($request['channel'] ?? 0);
|
|
|
|
try {
|
|
return $this->workerHttp->post('/relay/status', [
|
|
'local_ip' => $localIp,
|
|
'channel' => $channel,
|
|
], 8) ?? [];
|
|
} catch (Throwable) {
|
|
return $this->fetchShellyState($localIp, $channel);
|
|
}
|
|
}
|
|
|
|
private function switchRelay(array $request): array
|
|
{
|
|
$localIp = (string)($request['localIp'] ?? $request['local_ip'] ?? '');
|
|
$channel = (int)($request['channel'] ?? 0);
|
|
$on = (bool)($request['on'] ?? false);
|
|
$toggleAfter = $this->resolveRelayToggleAfterSeconds($request);
|
|
$deviceGeneration = $this->resolveShellyCommandGeneration($request);
|
|
$workerPayload = [
|
|
'local_ip' => $localIp,
|
|
'channel' => $channel,
|
|
'on' => $on,
|
|
];
|
|
if ($toggleAfter !== null) {
|
|
$workerPayload['toggle_after'] = $toggleAfter;
|
|
$workerPayload['toggleAfter'] = $toggleAfter;
|
|
}
|
|
if ($deviceGeneration !== null) {
|
|
$workerPayload['device_generation'] = $deviceGeneration;
|
|
$workerPayload['deviceGeneration'] = $deviceGeneration;
|
|
}
|
|
|
|
try {
|
|
return $this->workerHttp->post('/relay/switch', $workerPayload, 8) ?? [];
|
|
} catch (Throwable) {
|
|
$timerQuery = $toggleAfter === null ? '' : '&toggle_after=' . rawurlencode((string)$toggleAfter);
|
|
$legacyTimerQuery = $toggleAfter === null ? '' : '&timer=' . rawurlencode((string)$toggleAfter);
|
|
$runRpcSwitch = fn(): array => $this->http->getJson(
|
|
sprintf('http://%s/rpc/Switch.Set?id=%d&on=%s%s', $localIp, $channel, $on ? 'true' : 'false', $timerQuery),
|
|
8
|
|
);
|
|
$runLegacySwitch = fn(): array => $this->http->getJson(
|
|
sprintf('http://%s/relay/%d?turn=%s%s', $localIp, $channel, $on ? 'on' : 'off', $legacyTimerQuery),
|
|
8
|
|
);
|
|
|
|
if ($toggleAfter !== null) {
|
|
$attempts = $deviceGeneration === 1
|
|
? [$runLegacySwitch, $runRpcSwitch]
|
|
: [$runRpcSwitch, $runLegacySwitch];
|
|
$lastError = null;
|
|
foreach ($attempts as $attempt) {
|
|
try {
|
|
$attempt();
|
|
return $this->fetchShellyState($localIp, $channel);
|
|
} catch (Throwable $throwable) {
|
|
$lastError = $throwable;
|
|
}
|
|
}
|
|
|
|
throw $lastError ?? new RuntimeException('Unable to switch relay');
|
|
}
|
|
|
|
try {
|
|
$runRpcSwitch();
|
|
} catch (Throwable) {
|
|
$runLegacySwitch();
|
|
}
|
|
|
|
return $this->fetchShellyState($localIp, $channel);
|
|
}
|
|
}
|
|
|
|
private function resolveRelayToggleAfterSeconds(array $request): ?int
|
|
{
|
|
$configured = $request['toggleAfter'] ?? $request['toggle_after'] ?? $request['timer'] ?? null;
|
|
if (!is_int($configured) && !is_float($configured) && !(is_string($configured) && is_numeric(trim($configured)))) {
|
|
return null;
|
|
}
|
|
|
|
$seconds = (int)floor((float)$configured);
|
|
if ($seconds <= 0) {
|
|
return null;
|
|
}
|
|
|
|
return min($seconds, self::MAX_RELAY_TOGGLE_AFTER_SECONDS);
|
|
}
|
|
|
|
private function normalizeShellyDeviceGeneration(mixed $value): ?int
|
|
{
|
|
if (is_int($value) || is_float($value) || (is_string($value) && is_numeric(trim($value)))) {
|
|
$generation = (int)$value;
|
|
return $generation > 0 ? $generation : null;
|
|
}
|
|
|
|
return null;
|
|
}
|
|
|
|
private function inferShellyDeviceGenerationFromString(mixed $value): ?int
|
|
{
|
|
$normalized = trim((string)$value);
|
|
if ($normalized === '') {
|
|
return null;
|
|
}
|
|
|
|
if (preg_match('/\bgen(?:eration)?\s*([1-9]\d*)\b/i', $normalized, $explicit) === 1) {
|
|
return (int)$explicit[1];
|
|
}
|
|
|
|
$upper = strtoupper($normalized);
|
|
if (preg_match('/\bS([3-9])(?:[A-Z0-9]+)?-[A-Z0-9-]+\b/', $upper, $series) === 1) {
|
|
return (int)$series[1];
|
|
}
|
|
|
|
if (preg_match('/\b(?:SHELLY\s+)?(?:PLUS|PRO)\b/i', $normalized) === 1 || preg_match('/\bSP[A-Z0-9]+-[A-Z0-9-]+\b/', $upper) === 1) {
|
|
return 2;
|
|
}
|
|
|
|
if (preg_match('/\bSH[A-Z0-9]+-?[A-Z0-9-]*\b/', $upper) === 1) {
|
|
return 1;
|
|
}
|
|
|
|
return null;
|
|
}
|
|
|
|
private function resolveShellyCommandGeneration(array $request): ?int
|
|
{
|
|
foreach ([
|
|
$request['gen'] ?? null,
|
|
$request['generation'] ?? null,
|
|
$request['deviceGeneration'] ?? null,
|
|
$request['device_generation'] ?? null,
|
|
is_array($request['capabilities'] ?? null) ? ($request['capabilities']['generation'] ?? null) : null,
|
|
] as $candidate) {
|
|
$generation = $this->normalizeShellyDeviceGeneration($candidate);
|
|
if ($generation !== null) {
|
|
return $generation;
|
|
}
|
|
}
|
|
|
|
foreach ([
|
|
$request['model'] ?? null,
|
|
$request['deviceModel'] ?? null,
|
|
$request['device_model'] ?? null,
|
|
$request['type'] ?? null,
|
|
$request['deviceType'] ?? null,
|
|
$request['device_type'] ?? null,
|
|
$request['app'] ?? null,
|
|
$request['name'] ?? null,
|
|
$request['deviceName'] ?? null,
|
|
$request['device_name'] ?? null,
|
|
$request['deviceId'] ?? null,
|
|
$request['device_id'] ?? null,
|
|
$request['relayId'] ?? null,
|
|
$request['relay_id'] ?? null,
|
|
$request['mac'] ?? null,
|
|
] as $candidate) {
|
|
$generation = $this->inferShellyDeviceGenerationFromString($candidate);
|
|
if ($generation !== null) {
|
|
return $generation;
|
|
}
|
|
}
|
|
|
|
return null;
|
|
}
|
|
|
|
private function downloadToFileAtomic(string $url, string $path, string $expectedSha256 = ''): array
|
|
{
|
|
$directory = dirname($path);
|
|
if (!is_dir($directory)) {
|
|
@mkdir($directory, 0777, true);
|
|
}
|
|
|
|
$temporaryPath = $path . '.download';
|
|
$backupPath = $path . '.bak';
|
|
$raw = $this->http->download($url, 60);
|
|
file_put_contents($temporaryPath, $raw);
|
|
|
|
if ($expectedSha256 !== '') {
|
|
$actualSha = hash_file('sha256', $temporaryPath);
|
|
if (!is_string($actualSha) || !hash_equals(strtolower($expectedSha256), strtolower($actualSha))) {
|
|
@unlink($temporaryPath);
|
|
throw new RuntimeException('Artifact checksum mismatch for ' . basename($path));
|
|
}
|
|
}
|
|
|
|
if (is_file($backupPath)) {
|
|
@unlink($backupPath);
|
|
}
|
|
|
|
if (is_file($path) && !@rename($path, $backupPath)) {
|
|
@unlink($temporaryPath);
|
|
throw new RuntimeException('Could not create backup for ' . $path);
|
|
}
|
|
|
|
if (!@rename($temporaryPath, $path)) {
|
|
if (is_file($backupPath)) {
|
|
@rename($backupPath, $path);
|
|
}
|
|
@unlink($temporaryPath);
|
|
throw new RuntimeException('Could not promote downloaded artifact for ' . $path);
|
|
}
|
|
|
|
return [
|
|
'path' => $path,
|
|
'url' => $url,
|
|
'backup_path' => is_file($backupPath) ? $backupPath : null,
|
|
'sha256' => hash_file('sha256', $path),
|
|
];
|
|
}
|
|
|
|
private function classifyManagementError(Throwable $throwable, string $type): string
|
|
{
|
|
$message = strtolower(trim($throwable->getMessage()));
|
|
if (str_contains($message, 'token') || str_contains($message, 'credential')) {
|
|
return 'EDGE_GATEWAY_INVALID_TOKEN';
|
|
}
|
|
if (str_contains($message, 'timeout') || str_contains($message, 'http 5')) {
|
|
return 'EDGE_GATEWAY_OPERATION_TIMEOUT';
|
|
}
|
|
if ($type === 'UPDATE' && (str_contains($message, 'version') || str_contains($message, 'checksum'))) {
|
|
return 'EDGE_GATEWAY_UNSUPPORTED_VERSION';
|
|
}
|
|
|
|
return 'EDGE_GATEWAY_VALIDATION_FAILED';
|
|
}
|
|
|
|
private function fetchShellyState(string $localIp, int $channel): array
|
|
{
|
|
if ($localIp === '') {
|
|
throw new RuntimeException('Missing Shelly IP address');
|
|
}
|
|
|
|
try {
|
|
$payload = $this->http->getJson(sprintf('http://%s/rpc/Switch.GetStatus?id=%d', $localIp, $channel), 8);
|
|
return [
|
|
'online' => true,
|
|
'on' => (bool)($payload['output'] ?? false),
|
|
'output' => (bool)($payload['output'] ?? false),
|
|
'raw' => $payload,
|
|
];
|
|
} catch (Throwable) {
|
|
$payload = $this->http->getJson(sprintf('http://%s/relay/%d', $localIp, $channel), 8);
|
|
return [
|
|
'online' => true,
|
|
'on' => (bool)($payload['ison'] ?? false),
|
|
'output' => (bool)($payload['ison'] ?? false),
|
|
'raw' => $payload,
|
|
];
|
|
}
|
|
}
|
|
|
|
private function buildSystemMetrics(): array
|
|
{
|
|
$diskTotal = @disk_total_space($this->installDir);
|
|
$diskFree = @disk_free_space($this->installDir);
|
|
$diskUsed = (is_numeric($diskTotal) && is_numeric($diskFree)) ? ((float)$diskTotal - (float)$diskFree) : null;
|
|
$diskUsagePct = ($diskTotal && $diskUsed !== null && $diskTotal > 0)
|
|
? (int)round(($diskUsed / (float)$diskTotal) * 100)
|
|
: null;
|
|
$memoryLimitBytes = $this->iniBytes((string)ini_get('memory_limit'));
|
|
$memoryUsage = memory_get_usage(true);
|
|
$memoryUsagePct = ($memoryLimitBytes !== null && $memoryLimitBytes > 0)
|
|
? (int)round(($memoryUsage / $memoryLimitBytes) * 100)
|
|
: null;
|
|
$loadAverage = function_exists('sys_getloadavg') ? sys_getloadavg() : [];
|
|
|
|
return [
|
|
'memory_usage_bytes' => $memoryUsage,
|
|
'memory_peak_bytes' => memory_get_peak_usage(true),
|
|
'memory_usage_pct' => $memoryUsagePct,
|
|
'cpu_usage_pct' => is_array($loadAverage) && isset($loadAverage[0]) ? max(0, (int)round((float)$loadAverage[0] * 100)) : null,
|
|
'load_average' => $loadAverage,
|
|
'disk_usage_pct' => $diskUsagePct,
|
|
'disk_used_bytes' => $diskUsed,
|
|
'disk_total_bytes' => is_numeric($diskTotal) ? (int)$diskTotal : null,
|
|
'disk_mount' => $this->installDir,
|
|
'latency_ms' => null,
|
|
];
|
|
}
|
|
|
|
private function buildContainerHealth(): array
|
|
{
|
|
$services = [[
|
|
'name' => 'edge-agent',
|
|
'status' => 'healthy',
|
|
'updated_at' => date('c'),
|
|
]];
|
|
|
|
$services[] = $this->probeWorkerHealth();
|
|
$services[] = $this->probeTcpService('redis', 'redis', 6379);
|
|
$services[] = $this->probeTcpService('mariadb', 'mariadb', 3306);
|
|
$services[] = $this->probeHttpService('minio', 'http://minio:9000/minio/health/live');
|
|
$services[] = $this->probeAutoUpdaterHealth();
|
|
|
|
$healthyCount = count(array_filter($services, static fn(array $service): bool => (string)($service['status'] ?? '') === 'healthy'));
|
|
$state = $healthyCount === count($services) ? 'ONLINE' : 'DEGRADED';
|
|
|
|
return [
|
|
'state' => $state,
|
|
'summary' => sprintf('%d/%d containers healthy', $healthyCount, count($services)),
|
|
'services' => $services,
|
|
];
|
|
}
|
|
|
|
private function buildOutboxStatus(): array
|
|
{
|
|
$summary = $this->stateStore->outboxSummary();
|
|
$queued = (int)($summary['queued'] ?? 0);
|
|
return [
|
|
'state' => $queued > 0 ? 'QUEUED' : 'IN_SYNC',
|
|
'queued' => $queued,
|
|
'oldest_queued_at' => $summary['oldest_queued_at'] ?? null,
|
|
'last_replayed_at' => $summary['last_replayed_at'] ?? null,
|
|
'summary' => $queued > 0 ? sprintf('%d outbound items queued', $queued) : 'Outbox is empty',
|
|
];
|
|
}
|
|
|
|
private function initializeControlPlaneStatus(): void
|
|
{
|
|
$this->writeControlPlaneStatus();
|
|
}
|
|
|
|
private function touchLoopHeartbeat(): void
|
|
{
|
|
$this->writeControlPlaneStatus([
|
|
'last_loop_at' => date('c'),
|
|
]);
|
|
}
|
|
|
|
private function currentLastSuccessfulSyncAt(): ?string
|
|
{
|
|
$current = $this->readControlPlaneStatus()['last_successful_sync_at'] ?? $this->stateStore->getJson('last_sync_at');
|
|
return $this->normalizeControlPlaneStatusTimestamp($current);
|
|
}
|
|
|
|
private function recordHeartbeatAttempt(): string
|
|
{
|
|
$attemptedAt = date('c');
|
|
$this->writeControlPlaneStatus([
|
|
'last_heartbeat_attempt_at' => $attemptedAt,
|
|
]);
|
|
return $attemptedAt;
|
|
}
|
|
|
|
private function recordSuccessfulSync(?string $at = null, array $statusOverrides = []): void
|
|
{
|
|
$syncedAt = $this->normalizeControlPlaneStatusTimestamp($at ?? date('c')) ?? date('c');
|
|
$this->stateStore->setJson('last_sync_at', $syncedAt);
|
|
$statusOverrides['last_successful_sync_at'] = $syncedAt;
|
|
$this->writeControlPlaneStatus($statusOverrides);
|
|
}
|
|
|
|
private function recordTransportFailure(string $context, Throwable $throwable): void
|
|
{
|
|
$message = $this->normalizeControlPlaneStatusString($throwable->getMessage()) ?? $throwable::class;
|
|
$this->writeControlPlaneStatus([
|
|
'last_transport_failure_at' => date('c'),
|
|
'last_transport_error' => $message,
|
|
]);
|
|
$this->logger->warning($context . ': ' . $message);
|
|
}
|
|
|
|
/**
|
|
* @return array<string,mixed>
|
|
*/
|
|
private function buildControlPlaneStatusPayload(): array
|
|
{
|
|
return $this->writeControlPlaneStatus();
|
|
}
|
|
|
|
/**
|
|
* @return array<string,mixed>
|
|
*/
|
|
private function readControlPlaneStatus(): array
|
|
{
|
|
if (!is_file($this->controlPlaneStatusPath)) {
|
|
return [];
|
|
}
|
|
|
|
$decoded = json_decode((string)file_get_contents($this->controlPlaneStatusPath), true);
|
|
return is_array($decoded) ? $decoded : [];
|
|
}
|
|
|
|
/**
|
|
* @param array<string,mixed> $overrides
|
|
* @return array<string,mixed>
|
|
*/
|
|
private function writeControlPlaneStatus(array $overrides = []): array
|
|
{
|
|
$current = $this->readControlPlaneStatus();
|
|
$outboxSummary = $this->stateStore->outboxSummary();
|
|
$lastSuccessfulSyncAt = array_key_exists('last_successful_sync_at', $overrides)
|
|
? $overrides['last_successful_sync_at']
|
|
: ($current['last_successful_sync_at'] ?? $this->stateStore->getJson('last_sync_at'));
|
|
$lastTransportError = array_key_exists('last_transport_error', $overrides)
|
|
? $overrides['last_transport_error']
|
|
: ($current['last_transport_error'] ?? null);
|
|
$lastTransportFailureAt = array_key_exists('last_transport_failure_at', $overrides)
|
|
? $overrides['last_transport_failure_at']
|
|
: ($current['last_transport_failure_at'] ?? null);
|
|
$lastHeartbeatAttemptAt = array_key_exists('last_heartbeat_attempt_at', $overrides)
|
|
? $overrides['last_heartbeat_attempt_at']
|
|
: ($current['last_heartbeat_attempt_at'] ?? null);
|
|
$lastHeartbeatSuccessAt = array_key_exists('last_heartbeat_success_at', $overrides)
|
|
? $overrides['last_heartbeat_success_at']
|
|
: ($current['last_heartbeat_success_at'] ?? null);
|
|
$lastLoopAt = array_key_exists('last_loop_at', $overrides)
|
|
? $overrides['last_loop_at']
|
|
: ($current['last_loop_at'] ?? null);
|
|
|
|
$status = [
|
|
'started_at' => $this->normalizeControlPlaneStatusTimestamp($current['started_at'] ?? null) ?? date('c'),
|
|
'agent_instance_id' => $this->agentInstanceId,
|
|
'last_loop_at' => $this->normalizeControlPlaneStatusTimestamp($lastLoopAt),
|
|
'last_heartbeat_attempt_at' => $this->normalizeControlPlaneStatusTimestamp($lastHeartbeatAttemptAt),
|
|
'last_heartbeat_success_at' => $this->normalizeControlPlaneStatusTimestamp($lastHeartbeatSuccessAt),
|
|
'last_successful_sync_at' => $this->normalizeControlPlaneStatusTimestamp($lastSuccessfulSyncAt),
|
|
'outbox_queued' => array_key_exists('outbox_queued', $overrides)
|
|
? max(0, (int)$overrides['outbox_queued'])
|
|
: max(0, (int)($outboxSummary['queued'] ?? 0)),
|
|
'broker_connected' => array_key_exists('broker_connected', $overrides)
|
|
? (bool)$overrides['broker_connected']
|
|
: $this->isBrokerConnected(),
|
|
'last_transport_error' => $this->normalizeControlPlaneStatusString($lastTransportError),
|
|
'last_transport_failure_at' => $this->normalizeControlPlaneStatusTimestamp($lastTransportFailureAt),
|
|
];
|
|
|
|
file_put_contents(
|
|
$this->controlPlaneStatusPath,
|
|
json_encode($status, JSON_PRETTY_PRINT | JSON_UNESCAPED_SLASHES) . PHP_EOL
|
|
);
|
|
|
|
return $status;
|
|
}
|
|
|
|
private function normalizeControlPlaneStatusTimestamp(mixed $value): ?string
|
|
{
|
|
return is_string($value) && trim($value) !== '' ? trim($value) : null;
|
|
}
|
|
|
|
private function normalizeControlPlaneStatusString(mixed $value): ?string
|
|
{
|
|
if (is_string($value)) {
|
|
$normalized = trim($value);
|
|
return $normalized !== '' ? $normalized : null;
|
|
}
|
|
if (is_scalar($value)) {
|
|
$normalized = trim((string)$value);
|
|
return $normalized !== '' ? $normalized : null;
|
|
}
|
|
|
|
return null;
|
|
}
|
|
|
|
private function flushOutbox(): void
|
|
{
|
|
if ($this->shouldSkipOutboxReplay()) {
|
|
return;
|
|
}
|
|
|
|
$items = $this->stateStore->queuedItems(self::OUTBOX_REPLAY_BATCH_LIMIT);
|
|
foreach ($items as $item) {
|
|
try {
|
|
$timeoutSeconds = $this->outboxReplayTimeoutSeconds((string)($item['type'] ?? ''));
|
|
$endpoint = (string)$item['endpoint'];
|
|
$payload = is_array($item['payload'] ?? null) ? (array)$item['payload'] : [];
|
|
$brokerDispatch = $this->dispatchBrokerControlPlaneEvent($endpoint, $payload);
|
|
if ($brokerDispatch !== true || $this->shouldPersistControlPlaneEventOverHttp($endpoint)) {
|
|
$this->http->post($endpoint, $payload, $timeoutSeconds);
|
|
}
|
|
$this->stateStore->removeOutboxItem((int)$item['id']);
|
|
$this->lastOutboxFailureAt = 0;
|
|
$this->recordSuccessfulSync();
|
|
} catch (Throwable $throwable) {
|
|
$this->lastOutboxFailureAt = time();
|
|
$this->recordTransportFailure(
|
|
'Outbox replay blocked on ' . (string)$item['type'] . ' for ' . (string)$item['endpoint'],
|
|
$throwable
|
|
);
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
|
|
private function shouldSkipOutboxReplay(): bool
|
|
{
|
|
return $this->lastOutboxFailureAt > 0
|
|
&& (time() - $this->lastOutboxFailureAt) < self::OUTBOX_REPLAY_FAILURE_COOLDOWN_SECONDS;
|
|
}
|
|
|
|
private function outboxReplayTimeoutSeconds(string $type): int
|
|
{
|
|
return $type === 'operation_complete'
|
|
? self::OUTBOX_OPERATION_COMPLETE_REPLAY_TIMEOUT_SECONDS
|
|
: self::OUTBOX_REPLAY_TIMEOUT_SECONDS;
|
|
}
|
|
|
|
private function probeWorkerHealth(): array
|
|
{
|
|
try {
|
|
$workerHealth = $this->workerHttp->getJson(
|
|
rtrim((string)$this->config->get('workerBaseUrl', self::DEFAULT_WORKER_BASE_URL), '/') . '/health',
|
|
5
|
|
);
|
|
return [
|
|
'name' => 'lan-worker',
|
|
'status' => (string)($workerHealth['status'] ?? 'healthy'),
|
|
'updated_at' => (string)($workerHealth['timestamp'] ?? date('c')),
|
|
];
|
|
} catch (Throwable $throwable) {
|
|
return [
|
|
'name' => 'lan-worker',
|
|
'status' => 'degraded',
|
|
'updated_at' => date('c'),
|
|
'error' => $throwable->getMessage(),
|
|
];
|
|
}
|
|
}
|
|
|
|
private function probeHttpService(string $name, string $url): array
|
|
{
|
|
try {
|
|
$this->http->download($url, 5);
|
|
return [
|
|
'name' => $name,
|
|
'status' => 'healthy',
|
|
'updated_at' => date('c'),
|
|
];
|
|
} catch (Throwable $throwable) {
|
|
return [
|
|
'name' => $name,
|
|
'status' => 'degraded',
|
|
'updated_at' => date('c'),
|
|
'error' => $throwable->getMessage(),
|
|
];
|
|
}
|
|
}
|
|
|
|
private function probeTcpService(string $name, string $host, int $port): array
|
|
{
|
|
$socket = @fsockopen($host, $port, $errno, $error, 3);
|
|
if (is_resource($socket)) {
|
|
fclose($socket);
|
|
return [
|
|
'name' => $name,
|
|
'status' => 'healthy',
|
|
'updated_at' => date('c'),
|
|
];
|
|
}
|
|
|
|
return [
|
|
'name' => $name,
|
|
'status' => 'degraded',
|
|
'updated_at' => date('c'),
|
|
'error' => trim((string)$error) !== '' ? trim((string)$error) : 'TCP connection failed',
|
|
'code' => $errno > 0 ? $errno : null,
|
|
];
|
|
}
|
|
|
|
private function probeAutoUpdaterHealth(): array
|
|
{
|
|
$path = $this->runtimeDir . DIRECTORY_SEPARATOR . 'auto-updater-heartbeat.json';
|
|
if (!is_file($path)) {
|
|
return [
|
|
'name' => 'auto-updater',
|
|
'status' => 'degraded',
|
|
'updated_at' => date('c'),
|
|
'error' => 'No auto-updater heartbeat recorded',
|
|
];
|
|
}
|
|
|
|
$decoded = json_decode((string)file_get_contents($path), true);
|
|
$status = trim((string)($decoded['status'] ?? 'idle'));
|
|
$ageSeconds = max(0, time() - filemtime($path));
|
|
$healthy = $ageSeconds <= 90 && $status !== 'error';
|
|
|
|
return [
|
|
'name' => 'auto-updater',
|
|
'status' => $healthy ? 'healthy' : 'degraded',
|
|
'updated_at' => (string)($decoded['updated_at'] ?? date('c')),
|
|
'mode' => $status,
|
|
'error' => $healthy ? null : (string)($decoded['last_output'] ?? 'Auto-updater heartbeat is stale'),
|
|
];
|
|
}
|
|
|
|
private function sendControlPlaneEvent(string $endpoint, array $payload, string $type, int $timeoutSeconds = 20): bool
|
|
{
|
|
$this->lastControlPlaneResponse = null;
|
|
$brokerDispatch = $this->dispatchBrokerControlPlaneEvent($endpoint, $payload);
|
|
if ($brokerDispatch === true && !$this->shouldPersistControlPlaneEventOverHttp($endpoint)) {
|
|
return true;
|
|
}
|
|
|
|
try {
|
|
$response = $this->http->post($endpoint, $payload, $timeoutSeconds);
|
|
$this->lastControlPlaneResponse = is_array($response) ? $response : null;
|
|
$this->recordSuccessfulSync();
|
|
return true;
|
|
} catch (Throwable $throwable) {
|
|
if ($brokerDispatch === true && !$this->shouldPersistControlPlaneEventOverHttp($endpoint)) {
|
|
$this->recordTransportFailure(
|
|
'Broker dispatched ' . $type . ' but HTTP persistence failed on ' . $endpoint,
|
|
$throwable
|
|
);
|
|
return true;
|
|
}
|
|
|
|
$this->stateStore->enqueue($type, $endpoint, $payload);
|
|
$this->recordTransportFailure(
|
|
'Queued ' . $type . ' to local outbox after transport failure on ' . $endpoint,
|
|
$throwable
|
|
);
|
|
return false;
|
|
}
|
|
}
|
|
|
|
private function applyBrokerUrlFromControlPlaneResponse(?array $response): void
|
|
{
|
|
if (!is_array($response)) {
|
|
return;
|
|
}
|
|
|
|
$payload = is_array($response['data'] ?? null) ? (array)$response['data'] : $response;
|
|
$brokerUrl = trim((string)($payload['broker_url'] ?? $payload['gateway']['broker_url'] ?? ''));
|
|
if ($brokerUrl === '') {
|
|
return;
|
|
}
|
|
|
|
$current = trim((string)$this->config->get('brokerUrl'));
|
|
$currentNormalized = $this->normalizeBrokerUrlForComparison($current);
|
|
$nextNormalized = $this->normalizeBrokerUrlForComparison($brokerUrl);
|
|
if ($currentNormalized !== null && $currentNormalized === $nextNormalized) {
|
|
return;
|
|
}
|
|
|
|
$this->config->set('brokerUrl', rtrim($brokerUrl, '/'));
|
|
$this->config->save();
|
|
$this->configureBrokerClient();
|
|
$this->logger->info('Updated broker URL from control plane heartbeat response.');
|
|
}
|
|
|
|
private function normalizeBrokerUrlForComparison(?string $value): ?string
|
|
{
|
|
$trimmed = rtrim(trim((string)($value ?? '')), '/');
|
|
if ($trimmed === '') {
|
|
return null;
|
|
}
|
|
|
|
if (str_starts_with($trimmed, 'https://')) {
|
|
$trimmed = 'wss://' . substr($trimmed, 8);
|
|
} elseif (str_starts_with($trimmed, 'http://')) {
|
|
$trimmed = 'ws://' . substr($trimmed, 7);
|
|
} elseif (!str_starts_with($trimmed, 'ws://') && !str_starts_with($trimmed, 'wss://')) {
|
|
$trimmed = 'ws://' . $trimmed;
|
|
}
|
|
|
|
$parts = parse_url($trimmed);
|
|
if (!is_array($parts) || empty($parts['host'])) {
|
|
return strtolower($trimmed);
|
|
}
|
|
|
|
$scheme = strtolower((string)($parts['scheme'] ?? 'ws'));
|
|
$host = strtolower((string)$parts['host']);
|
|
$port = isset($parts['port']) ? (int)$parts['port'] : ($scheme === 'wss' ? 443 : 80);
|
|
$path = rtrim((string)($parts['path'] ?? ''), '/');
|
|
if ($host === 'edge-broker' && ($path === '/edge-broker' || str_starts_with($path, '/edge-broker/'))) {
|
|
$path = rtrim(substr($path, strlen('/edge-broker')), '/');
|
|
}
|
|
|
|
$isDefaultPort = ($scheme === 'wss' && $port === 443) || ($scheme !== 'wss' && $port === 80);
|
|
return $scheme . '://' . $host . ($isDefaultPort ? '' : ':' . $port) . ($path === '' ? '' : $path);
|
|
}
|
|
|
|
private function shouldPersistControlPlaneEventOverHttp(string $endpoint): bool
|
|
{
|
|
return preg_match('#/edge-agent/gateways/\d+/heartbeat$#', $endpoint) === 1;
|
|
}
|
|
|
|
private function requestControlPlaneEvent(string $endpoint, array $payload, string $type, int $timeoutSeconds = 20): ?array
|
|
{
|
|
$brokerDispatch = $this->dispatchBrokerControlPlaneEvent($endpoint, $payload);
|
|
if ($brokerDispatch === true) {
|
|
return ['data' => ['status' => 'ACKNOWLEDGED']];
|
|
}
|
|
|
|
try {
|
|
$response = $this->http->post($endpoint, $payload, $timeoutSeconds);
|
|
$this->recordSuccessfulSync();
|
|
return is_array($response) ? $response : null;
|
|
} catch (Throwable $throwable) {
|
|
$this->stateStore->enqueue($type, $endpoint, $payload);
|
|
$this->recordTransportFailure(
|
|
'Queued ' . $type . ' to local outbox after transport failure on ' . $endpoint,
|
|
$throwable
|
|
);
|
|
return null;
|
|
}
|
|
}
|
|
|
|
private function guardOperationEventResponse(?array $response): void
|
|
{
|
|
$operation = is_array($response['data'] ?? null) ? (array)$response['data'] : null;
|
|
if ($operation === null) {
|
|
return;
|
|
}
|
|
|
|
$status = strtoupper(trim((string)($operation['status'] ?? '')));
|
|
if ($status === 'CANCEL_REQUESTED' || $status === 'CANCELLED') {
|
|
throw new OperationAbortException('Operation cancelled by operator', true);
|
|
}
|
|
|
|
if ($status === 'FAILED') {
|
|
throw new OperationAbortException(
|
|
trim((string)($operation['error_message'] ?? '')) !== ''
|
|
? (string)$operation['error_message']
|
|
: 'Gateway operation failed'
|
|
);
|
|
}
|
|
}
|
|
|
|
private function markOperationCancellationRequested(int $operationId): void
|
|
{
|
|
$ids = $this->operationCancellationRequests();
|
|
if (!in_array($operationId, $ids, true)) {
|
|
$ids[] = $operationId;
|
|
$this->stateStore->setJson('cancel_requested_operations', array_values($ids));
|
|
}
|
|
}
|
|
|
|
private function clearOperationCancellationRequested(int $operationId): void
|
|
{
|
|
$ids = array_values(array_filter(
|
|
$this->operationCancellationRequests(),
|
|
static fn(int $currentId): bool => $currentId !== $operationId
|
|
));
|
|
$this->stateStore->setJson('cancel_requested_operations', $ids);
|
|
}
|
|
|
|
/**
|
|
* @return array<int,int>
|
|
*/
|
|
private function operationCancellationRequests(): array
|
|
{
|
|
$stored = $this->stateStore->getJson('cancel_requested_operations', []);
|
|
if (!is_array($stored)) {
|
|
return [];
|
|
}
|
|
|
|
return array_values(array_filter(
|
|
array_map(static fn(mixed $value): int => (int)$value, $stored),
|
|
static fn(int $value): bool => $value > 0
|
|
));
|
|
}
|
|
|
|
private function abortIfOperationCancelled(int $operationId): void
|
|
{
|
|
if (in_array($operationId, $this->operationCancellationRequests(), true)) {
|
|
throw new OperationAbortException('Operation cancelled by operator', true);
|
|
}
|
|
}
|
|
|
|
private function readRollbackStatus(): array
|
|
{
|
|
$rollbackPath = $this->runtimeDir . DIRECTORY_SEPARATOR . 'rollback-status.json';
|
|
if (is_file($rollbackPath)) {
|
|
$decoded = json_decode((string)file_get_contents($rollbackPath), true);
|
|
if (is_array($decoded)) {
|
|
return $decoded;
|
|
}
|
|
}
|
|
|
|
$rollback = $this->stateStore->getJson('rollback_status', null);
|
|
if (is_array($rollback)) {
|
|
return $rollback;
|
|
}
|
|
|
|
return [
|
|
'state' => 'IDLE',
|
|
'reason' => null,
|
|
'rolled_back_to' => null,
|
|
'at' => null,
|
|
];
|
|
}
|
|
|
|
private function currentStagedUpdate(): ?array
|
|
{
|
|
if (is_file($this->stagedUpdatePath)) {
|
|
$decoded = json_decode((string)file_get_contents($this->stagedUpdatePath), true);
|
|
if (is_array($decoded)) {
|
|
return $decoded;
|
|
}
|
|
}
|
|
|
|
$staged = $this->stateStore->getJson('staged_update', null);
|
|
return is_array($staged) ? $staged : null;
|
|
}
|
|
|
|
private function ensureExecutable(string $path): void
|
|
{
|
|
if (is_file($path)) {
|
|
@chmod($path, 0755);
|
|
}
|
|
}
|
|
|
|
private function workerAuthorizationHeaders(): array
|
|
{
|
|
$agentToken = trim((string)$this->config->get('agentToken', ''));
|
|
return $agentToken !== '' ? ['X-Truckwash-Worker-Token: ' . $agentToken] : [];
|
|
}
|
|
|
|
private function ensureAgentInstanceId(): string
|
|
{
|
|
$configured = trim((string)$this->config->get('agentInstanceId', ''));
|
|
if ($configured !== '') {
|
|
return substr($configured, 0, 128);
|
|
}
|
|
|
|
$generated = sprintf(
|
|
'%s-%s',
|
|
preg_replace('/[^A-Za-z0-9\-]+/', '-', gethostname() ?: 'truckwash-edge') ?: 'truckwash-edge',
|
|
substr(bin2hex(random_bytes(6)), 0, 12)
|
|
);
|
|
$this->config->set('agentInstanceId', $generated);
|
|
$this->config->save();
|
|
return $generated;
|
|
}
|
|
|
|
private function recoverPreviousOperationState(): void
|
|
{
|
|
$state = $this->readOperationState();
|
|
if ($state === null) {
|
|
return;
|
|
}
|
|
|
|
$completion = is_array($state['completion'] ?? null) ? (array)$state['completion'] : null;
|
|
$message = $completion !== null
|
|
? 'Recovered previous operation awaiting backend completion acknowledgement: operation %s (%s) last stage %s.'
|
|
: 'Recovered previous unfinished operation snapshot: operation %s (%s) last stage %s.';
|
|
|
|
$this->logger->warning(sprintf(
|
|
$message,
|
|
(string)($state['operation_id'] ?? 'unknown'),
|
|
(string)($state['type'] ?? 'unknown'),
|
|
(string)($state['stage'] ?? 'unknown')
|
|
));
|
|
}
|
|
|
|
private function finalizeOperationCompletion(
|
|
int $gatewayId,
|
|
int $operationId,
|
|
array $payload,
|
|
string $snapshotStatus,
|
|
array $snapshotExtra = []
|
|
): void {
|
|
$state = $this->readOperationState() ?? [];
|
|
$completion = [
|
|
'endpoint' => '/edge-agent/gateways/' . $gatewayId . '/operations/' . $operationId . '/complete',
|
|
'payload' => $payload,
|
|
'snapshot_status' => $snapshotStatus,
|
|
'snapshot_extra' => $snapshotExtra,
|
|
];
|
|
|
|
$state['status'] = 'COMPLETION_PENDING';
|
|
$state['stage'] = 'awaiting_completion_ack';
|
|
$state['updated_at'] = date('c');
|
|
$state['completion'] = $completion;
|
|
$this->persistOperationState($state);
|
|
|
|
if (!$this->dispatchOperationCompletion($completion, true)) {
|
|
return;
|
|
}
|
|
|
|
$this->snapshotOperationState($snapshotStatus, $snapshotExtra);
|
|
$this->clearOperationState();
|
|
}
|
|
|
|
private function resumePendingOperationCompletion(): bool
|
|
{
|
|
$state = $this->readOperationState();
|
|
if ($state === null) {
|
|
return false;
|
|
}
|
|
|
|
$completion = is_array($state['completion'] ?? null) ? (array)$state['completion'] : null;
|
|
if ($completion === null) {
|
|
return false;
|
|
}
|
|
|
|
if (!$this->dispatchOperationCompletion($completion, false)) {
|
|
return true;
|
|
}
|
|
|
|
$snapshotStatus = trim((string)($completion['snapshot_status'] ?? $state['status'] ?? 'FAILED'));
|
|
if ($snapshotStatus === '') {
|
|
$snapshotStatus = 'FAILED';
|
|
}
|
|
|
|
$snapshotExtra = is_array($completion['snapshot_extra'] ?? null)
|
|
? (array)$completion['snapshot_extra']
|
|
: [];
|
|
$this->snapshotOperationState($snapshotStatus, $snapshotExtra);
|
|
$this->clearOperationState();
|
|
$this->logger->info(sprintf(
|
|
'Acknowledged completion for recovered operation %s.',
|
|
(string)($state['operation_id'] ?? 'unknown')
|
|
));
|
|
|
|
return true;
|
|
}
|
|
|
|
private function dispatchOperationCompletion(array $completion, bool $queueOnFailure): bool
|
|
{
|
|
$endpoint = trim((string)($completion['endpoint'] ?? ''));
|
|
$payload = is_array($completion['payload'] ?? null) ? (array)$completion['payload'] : [];
|
|
if ($endpoint === '') {
|
|
$this->logger->error('Unable to dispatch operation completion: missing endpoint.');
|
|
return false;
|
|
}
|
|
|
|
$brokerDispatch = $this->dispatchBrokerControlPlaneEvent($endpoint, $payload);
|
|
if ($brokerDispatch === true) {
|
|
return true;
|
|
}
|
|
|
|
try {
|
|
$this->http->post($endpoint, $payload, self::OPERATION_COMPLETE_TIMEOUT_SECONDS);
|
|
$this->recordSuccessfulSync();
|
|
return true;
|
|
} catch (Throwable $throwable) {
|
|
if ($queueOnFailure) {
|
|
$this->stateStore->enqueue('operation_complete', $endpoint, $payload);
|
|
$this->recordTransportFailure(
|
|
'Queued operation_complete to local outbox after completion acknowledgement failure on ' . $endpoint,
|
|
$throwable
|
|
);
|
|
return false;
|
|
}
|
|
|
|
$this->recordTransportFailure(
|
|
'Retrying backend completion acknowledgement later for ' . $endpoint,
|
|
$throwable
|
|
);
|
|
return false;
|
|
}
|
|
}
|
|
|
|
private function dispatchBrokerControlPlaneEvent(string $endpoint, array $payload): ?bool
|
|
{
|
|
if (!$this->isBrokerConnected()) {
|
|
return null;
|
|
}
|
|
|
|
if (preg_match('#/edge-agent/gateways/(\d+)/heartbeat$#', $endpoint)) {
|
|
return $this->sendBrokerMessage([
|
|
'type' => 'TELEMETRY',
|
|
'payload' => $this->stripAgentAuthentication($payload),
|
|
]);
|
|
}
|
|
|
|
if (preg_match('#/edge-agent/gateways/\d+/operations/(\d+)/events$#', $endpoint, $matches)) {
|
|
return $this->sendBrokerMessage([
|
|
'type' => 'TASK_EVENT',
|
|
'operationId' => (int)$matches[1],
|
|
'payload' => $this->stripAgentAuthentication($payload),
|
|
]);
|
|
}
|
|
|
|
if (preg_match('#/edge-agent/gateways/\d+/operations/(\d+)/complete$#', $endpoint, $matches)) {
|
|
return $this->sendBrokerMessage([
|
|
'type' => 'TASK_RESULT',
|
|
'operationId' => (int)$matches[1],
|
|
'payload' => $this->stripAgentAuthentication($payload),
|
|
]);
|
|
}
|
|
|
|
if (preg_match('#/edge-agent/gateways/\d+/logs$#', $endpoint)) {
|
|
return $this->sendBrokerMessage([
|
|
'type' => 'LOG_FRAME',
|
|
'payload' => $this->stripAgentAuthentication($payload),
|
|
]);
|
|
}
|
|
|
|
if (preg_match('#/edge-agent/gateways/\d+/selfserve/machine-signal$#', $endpoint)) {
|
|
return $this->sendBrokerMessage([
|
|
'type' => 'MACHINE_SIGNAL',
|
|
'payload' => $this->stripAgentAuthentication($payload),
|
|
]);
|
|
}
|
|
|
|
return null;
|
|
}
|
|
|
|
private function stripAgentAuthentication(array $payload): array
|
|
{
|
|
unset($payload['agent_token']);
|
|
return $payload;
|
|
}
|
|
|
|
private function emitLogFrame(string $level, string $message): bool
|
|
{
|
|
$gatewayId = (int)$this->config->get('gatewayId', 0);
|
|
if ($gatewayId <= 0 || trim($message) === '') {
|
|
return false;
|
|
}
|
|
|
|
$payload = [
|
|
'level' => strtoupper(trim($level)) ?: 'INFO',
|
|
'message' => $message,
|
|
'stream' => 'agent',
|
|
'source' => 'EDGE_AGENT',
|
|
'context' => [
|
|
'agent_instance_id' => $this->agentInstanceId,
|
|
],
|
|
];
|
|
|
|
$endpoint = '/edge-agent/gateways/' . $gatewayId . '/logs';
|
|
$brokerDispatch = $this->dispatchBrokerControlPlaneEvent($endpoint, $payload);
|
|
if ($brokerDispatch === true) {
|
|
return true;
|
|
}
|
|
|
|
try {
|
|
$this->http->post($endpoint, $payload, 10);
|
|
return true;
|
|
} catch (Throwable) {
|
|
return false;
|
|
}
|
|
}
|
|
|
|
private function persistOperationState(array $state): void
|
|
{
|
|
file_put_contents($this->statePath, json_encode($state, JSON_PRETTY_PRINT | JSON_UNESCAPED_SLASHES) . PHP_EOL);
|
|
}
|
|
|
|
private function snapshotOperationState(string $status, array $extra = []): void
|
|
{
|
|
$state = $this->readOperationState() ?? [];
|
|
unset($state['completion']);
|
|
$state['status'] = $status;
|
|
$state['finished_at'] = date('c');
|
|
$state = array_merge($state, $extra);
|
|
file_put_contents($this->lastOperationSnapshotPath, json_encode($state, JSON_PRETTY_PRINT | JSON_UNESCAPED_SLASHES) . PHP_EOL);
|
|
}
|
|
|
|
private function clearOperationState(): void
|
|
{
|
|
if (is_file($this->statePath)) {
|
|
@unlink($this->statePath);
|
|
}
|
|
}
|
|
|
|
private function readOperationState(): ?array
|
|
{
|
|
if (!is_file($this->statePath)) {
|
|
return null;
|
|
}
|
|
|
|
$decoded = json_decode((string)file_get_contents($this->statePath), true);
|
|
return is_array($decoded) ? $decoded : null;
|
|
}
|
|
|
|
private function nextUpdateWindowStartIso(string $window): string
|
|
{
|
|
$parts = explode('-', $window, 2);
|
|
$start = trim((string)($parts[0] ?? '02:00'));
|
|
if (!preg_match('/^\d{2}:\d{2}$/', $start)) {
|
|
$start = '02:00';
|
|
}
|
|
|
|
$now = new DateTimeImmutable('now');
|
|
[$hour, $minute] = array_map('intval', explode(':', $start));
|
|
$candidate = $now->setTime($hour, $minute, 0);
|
|
if ($candidate <= $now) {
|
|
$candidate = $candidate->modify('+1 day');
|
|
}
|
|
|
|
return $candidate->format(DateTimeInterface::ATOM);
|
|
}
|
|
|
|
private function reloadConfigFromDisk(): void
|
|
{
|
|
$reloaded = AgentConfig::load($this->config->path);
|
|
$this->config = $reloaded;
|
|
$this->workerHttp = new HttpJsonClient(
|
|
(string)$this->config->get('workerBaseUrl', self::DEFAULT_WORKER_BASE_URL),
|
|
$this->workerAuthorizationHeaders()
|
|
);
|
|
$this->configureBrokerClient();
|
|
}
|
|
|
|
private function pollRequestTimeoutSeconds(int $waitSeconds): int
|
|
{
|
|
return max(5, $waitSeconds + 2);
|
|
}
|
|
|
|
private function iniBytes(string $value): ?int
|
|
{
|
|
$normalized = trim(strtolower($value));
|
|
if ($normalized === '' || $normalized === '-1') {
|
|
return null;
|
|
}
|
|
|
|
$unit = substr($normalized, -1);
|
|
$number = (float)$normalized;
|
|
return match ($unit) {
|
|
'g' => (int)round($number * 1024 * 1024 * 1024),
|
|
'm' => (int)round($number * 1024 * 1024),
|
|
'k' => (int)round($number * 1024),
|
|
default => is_numeric($normalized) ? (int)$normalized : null,
|
|
};
|
|
}
|
|
}
|
|
|
|
$configPath = null;
|
|
foreach ($argv as $index => $argument) {
|
|
if ($argument === '--config' && isset($argv[$index + 1])) {
|
|
$configPath = $argv[$index + 1];
|
|
}
|
|
}
|
|
|
|
if ($configPath === null) {
|
|
fwrite(STDERR, "Usage: php agent.php --config /path/to/config.json\n");
|
|
exit(1);
|
|
}
|
|
|
|
$agent = new TruckwashEdgeAgent($configPath);
|
|
$agent->run();
|