> * @throws Exception */ public function listOperations(int $gatewayId, int $limit = 20, bool $includeEvents = true): array { $this->requireGateway($gatewayId); $this->failTimedOutOperations($gatewayId); $rows = (new edge_gateway_operations_o())->getFieldsWhere([ 'gateway_id' => $gatewayId, 'deleted_at' => null, ], ['id']); $ids = array_map(static fn(array $row): int => (int)$row['id'], $rows); rsort($ids); $ids = array_slice($ids, 0, max(1, $limit)); $operations = []; foreach ($ids as $id) { $operations[] = $this->serializeOperation((new edge_gateway_operations_o())->select($id), $includeEvents); } return $operations; } /** * @return array> * @throws Exception */ public function listOperationEvents(int $gatewayId, int $operationId, int $limit = 100): array { $this->requireOperation($gatewayId, $operationId); $rows = (new edge_gateway_operation_events_o())->getFieldsWhere([ 'gateway_id' => $gatewayId, 'operation_id' => $operationId, ], ['id']); $ids = array_map(static fn(array $row): int => (int)$row['id'], $rows); sort($ids); $ids = array_slice($ids, max(0, count($ids) - max(1, $limit))); $events = []; foreach ($ids as $id) { $events[] = (new edge_gateway_operation_events_o())->select($id)->asArray(); } return $events; } /** * @throws Exception */ public function getActiveOperation(int $gatewayId, bool $includeEvents = true): ?array { $this->requireGateway($gatewayId); $this->failTimedOutOperations($gatewayId); $statement = db::getPDO()->prepare( "SELECT id FROM edge_gateway_operations WHERE gateway_id = :gateway_id AND deleted_at IS NULL AND status IN ('PENDING', 'IN_PROGRESS') ORDER BY FIELD(status, 'IN_PROGRESS', 'PENDING'), id ASC LIMIT 1" ); $statement->execute([':gateway_id' => $gatewayId]); $row = $statement->fetch(); if (!is_array($row) || empty($row['id'])) { return null; } return $this->serializeOperation((new edge_gateway_operations_o())->select((int)$row['id']), $includeEvents); } /** * @throws Exception */ public function buildRecentOperationsSummary(int $gatewayId): array { $operations = $this->listOperations($gatewayId, 25, false); $summary = [ 'total' => count($operations), 'pending' => 0, 'in_progress' => 0, 'completed' => 0, 'failed' => 0, 'latest_completed_at' => null, 'latest_failed_at' => null, 'latest_type' => $operations[0]['type'] ?? null, 'latest_status' => $operations[0]['status'] ?? null, ]; foreach ($operations as $operation) { $status = (string)($operation['status'] ?? self::STATUS_PENDING); if ($status === self::STATUS_PENDING) { $summary['pending'] += 1; } elseif ($status === self::STATUS_IN_PROGRESS) { $summary['in_progress'] += 1; } elseif ($status === self::STATUS_COMPLETED) { $summary['completed'] += 1; $summary['latest_completed_at'] ??= $operation['completed_at'] ?? null; } elseif ($status === self::STATUS_FAILED) { $summary['failed'] += 1; $summary['latest_failed_at'] ??= $operation['completed_at'] ?? null; } } return $summary; } /** * @throws Exception */ public function queueOperation(int $gatewayId, string $type, array $request = [], ?int $requestedBy = null): array { $gateway = $this->requireGateway($gatewayId); $type = self::normalizeOperationType($type); $request = $this->validateOperationRequest($gateway, $type, $request); if ($this->getActiveOperation($gatewayId) !== null) { throw new edge_gateway_operation_exception( 'Another gateway operation is already active', self::ERROR_CONFLICT, 409 ); } if ($type === self::TYPE_DISCOVERY) { $gateway->discovery_status->set('PENDING'); } elseif ($type === self::TYPE_UPDATE) { $targetVersion = trim((string)($request['target_version'] ?? '')); if ($targetVersion !== '') { $gateway->target_version->set($targetVersion); } } $summary = [ 'label' => match ($type) { self::TYPE_DISCOVERY => 'Discovery queued', self::TYPE_UPDATE => 'Update queued', self::TYPE_UNINSTALL => 'Uninstall queued', }, 'progress' => 0, 'retryable' => true, ]; $operationId = (new edge_gateway_operations_o())->add_object([ 'gateway_id' => $gatewayId, 'type' => $type, 'operation_type' => $type, 'status' => self::STATUS_PENDING, 'request_json' => $request, 'summary_json' => $summary, 'result_json' => [], 'error_code' => null, 'error_message' => null, 'correlation_id' => bin2hex(random_bytes(16)), 'agent_instance_id' => null, 'lease_expires_at' => null, 'last_progress_at' => null, 'attempt_count' => 0, 'requested_by' => $requestedBy, 'requested_at' => $this->now(), 'started_at' => null, 'completed_at' => null, ]); $this->appendEventRecord( $gatewayId, $operationId, self::LEVEL_INFO, 'OPERATION_QUEUED', 'Operation queued for gateway execution', ['type' => $type, 'request' => $request] ); $this->manager()->logGatewayAudit( $gatewayId, (int)$gateway->department_id->value(), 'GATEWAY_OPERATION_QUEUED', $requestedBy, ['operation_id' => $operationId, 'type' => $type] ); return $this->serializeOperation((new edge_gateway_operations_o())->select($operationId), true); } /** * @throws Exception */ public function queueDiscoveryOperation(int $gatewayId, ?int $requestedBy = null): array { return $this->queueOperation($gatewayId, self::TYPE_DISCOVERY, [], $requestedBy); } /** * @throws Exception */ public function rotateCredentials(int $gatewayId, ?int $userId = null): array { return $this->manager()->rotateGatewayCredentials($gatewayId, $userId); } /** * @throws Exception */ public function claimNextOperation( int $gatewayId, string $plainToken, int $waitSeconds = edge_gateway_manager::COMMAND_POLL_TIMEOUT_SECONDS, ?string $agentInstanceId = null ): ?array { try { $this->manager()->authenticateGateway($gatewayId, $plainToken); } catch (Exception) { throw new edge_gateway_operation_exception('Invalid edge gateway token', self::ERROR_INVALID_TOKEN, 401); } $deadline = microtime(true) + max(0, $waitSeconds); do { $this->failTimedOutOperations($gatewayId); $operation = $this->claimPendingOperation($gatewayId, $this->normalizeAgentInstanceId($agentInstanceId, $gatewayId)); if ($operation !== null) { return $operation; } if (microtime(true) >= $deadline) { break; } usleep(self::POLL_INTERVAL_MICROSECONDS); } while (true); return null; } /** * @throws Exception */ public function appendAgentOperationEvent(int $gatewayId, int $operationId, string $plainToken, array $payload): array { try { $this->manager()->authenticateGateway($gatewayId, $plainToken); } catch (Exception) { throw new edge_gateway_operation_exception('Invalid edge gateway token', self::ERROR_INVALID_TOKEN, 401); } $operation = $this->requireOperation($gatewayId, $operationId); if (in_array((string)$operation->status->value(), [self::STATUS_COMPLETED, self::STATUS_FAILED], true)) { return $this->serializeOperation($operation, true); } $level = strtoupper(trim((string)($payload['level'] ?? self::LEVEL_INFO))); if (!in_array($level, [self::LEVEL_INFO, self::LEVEL_WARNING, self::LEVEL_ERROR], true)) { $level = self::LEVEL_INFO; } $message = trim((string)($payload['message'] ?? 'Operation event received')); if ($message === '') { $message = 'Operation event received'; } $code = isset($payload['code']) ? trim((string)$payload['code']) : null; $context = isset($payload['context']) && is_array($payload['context']) ? (array)$payload['context'] : []; $this->appendEventRecord($gatewayId, $operationId, $level, $code, $message, $context); $summary = (array)($operation->summary_json->value() ?? []); $summary['last_event_at'] = $this->now(); $summary['last_event_message'] = $message; if (isset($context['progress'])) { $summary['progress'] = max(0, min(100, (int)$context['progress'])); } if (isset($context['label']) && trim((string)$context['label']) !== '') { $summary['label'] = trim((string)$context['label']); } $operation->summary_json->set($summary); $this->refreshOperationLease($operation); return $this->serializeOperation($operation, true); } /** * @throws Exception */ public function completeAgentOperation(int $gatewayId, int $operationId, string $plainToken, array $payload): array { try { $this->manager()->authenticateGateway($gatewayId, $plainToken); } catch (Exception) { throw new edge_gateway_operation_exception('Invalid edge gateway token', self::ERROR_INVALID_TOKEN, 401); } $operation = $this->requireOperation($gatewayId, $operationId); if (in_array((string)$operation->status->value(), [self::STATUS_COMPLETED, self::STATUS_FAILED], true)) { return $this->serializeOperation($operation, true); } $ok = (bool)($payload['ok'] ?? false); $result = isset($payload['result']) && is_array($payload['result']) ? (array)$payload['result'] : []; $errorMessage = trim((string)($payload['error_message'] ?? $payload['error'] ?? '')); $errorCode = trim((string)($payload['error_code'] ?? '')); if (!$ok && $errorCode === '') { $errorCode = $this->classifyCompletionError($gatewayId, $errorMessage); } if (!$ok && $errorMessage === '') { $errorMessage = 'Gateway operation failed'; } $operation->status->set($ok ? self::STATUS_COMPLETED : self::STATUS_FAILED); $operation->result_json->set($result); $operation->error_code->set($ok ? null : $errorCode); $operation->error_message->set($ok ? null : $errorMessage); $operation->completed_at->set($this->now()); $operation->lease_expires_at->set(null); $operation->last_progress_at->set($this->now()); $summary = (array)($operation->summary_json->value() ?? []); $summary['label'] = $ok ? 'Completed' : 'Failed'; $summary['progress'] = 100; $summary['retryable'] = !$ok && $errorCode !== self::ERROR_UNSUPPORTED_VERSION; $operation->summary_json->set($summary); $this->appendEventRecord( $gatewayId, $operationId, $ok ? self::LEVEL_INFO : self::LEVEL_ERROR, $ok ? 'OPERATION_COMPLETED' : $errorCode, $ok ? 'Operation completed successfully' : $errorMessage, $result ); $this->applyCompletionSideEffects($gatewayId, $operation, $ok, $result, $errorCode, $errorMessage); return $this->serializeOperation($operation, true); } private static function normalizeOperationType(string $type): string { $normalized = strtoupper(trim($type)); if (in_array($normalized, [self::TYPE_DISCOVERY, self::TYPE_UPDATE, self::TYPE_UNINSTALL], true)) { return $normalized; } throw new edge_gateway_operation_exception( 'Unsupported gateway operation type', self::ERROR_VALIDATION, 422 ); } /** * @throws Exception */ private function validateOperationRequest(edge_gateways_o $gateway, string $type, array $request): array { if ($type === self::TYPE_DISCOVERY) { return $request; } if ($type === self::TYPE_UPDATE) { $targetVersion = trim((string)($request['target_version'] ?? '')); if ($targetVersion === '') { throw new edge_gateway_operation_exception( 'Update operations require target_version', self::ERROR_VALIDATION, 422 ); } $releaseChannel = trim((string)($request['release_channel'] ?? $gateway->release_channel->value() ?? edge_gateway_manager::DEFAULT_RELEASE_CHANNEL)); if ($releaseChannel === '') { $releaseChannel = edge_gateway_manager::DEFAULT_RELEASE_CHANNEL; } return array_merge( $this->manager()->buildUpdateOperationRequest($targetVersion, $releaseChannel), $request, [ 'target_version' => $targetVersion, 'release_channel' => $releaseChannel, ] ); } if ($type === self::TYPE_UNINSTALL) { return array_merge([ 'service_name' => edge_gateway_manager::DEFAULT_AGENT_SERVICE_NAME, 'install_dir' => '/opt/truckwash-edge-agent', ], $request); } return $request; } /** * @throws Exception */ private function claimPendingOperation(int $gatewayId, string $agentInstanceId): ?array { $pdo = db::getPDO(); $pdo->beginTransaction(); try { $statement = $pdo->prepare( "SELECT id FROM edge_gateway_operations WHERE gateway_id = :gateway_id AND deleted_at IS NULL AND status = 'PENDING' ORDER BY id ASC LIMIT 1 FOR UPDATE" ); $statement->execute([':gateway_id' => $gatewayId]); $row = $statement->fetch(); if (!is_array($row) || empty($row['id'])) { $pdo->commit(); return null; } $operation = (new edge_gateway_operations_o())->select((int)$row['id']); $operation->status->set(self::STATUS_IN_PROGRESS); $operation->started_at->set($this->now()); $operation->agent_instance_id->set($agentInstanceId); $operation->last_progress_at->set($this->now()); $operation->lease_expires_at->set($this->leaseExpiry()); $operation->attempt_count->set(((int)($operation->attempt_count->value() ?? 0)) + 1); $summary = (array)($operation->summary_json->value() ?? []); $summary['label'] = 'Gateway is processing the operation'; $summary['progress'] = max(5, (int)($summary['progress'] ?? 0)); $summary['claimed_by'] = $agentInstanceId; $operation->summary_json->set($summary); $pdo->commit(); $this->appendEventRecord( $gatewayId, (int)$operation->id, self::LEVEL_INFO, 'OPERATION_STARTED', 'Gateway started processing the operation', [ 'type' => (string)$operation->type->value(), 'agent_instance_id' => $agentInstanceId, 'attempt_count' => (int)($operation->attempt_count->value() ?? 1), ] ); return $this->serializeOperation($operation, true); } catch (\Throwable $throwable) { if ($pdo->inTransaction()) { $pdo->rollBack(); } throw $throwable; } } /** * @throws Exception */ private function failTimedOutOperations(int $gatewayId): void { $rows = (new edge_gateway_operations_o())->getFieldsWhere([ 'gateway_id' => $gatewayId, 'status' => self::STATUS_IN_PROGRESS, 'deleted_at' => null, ], ['id']); $now = time(); foreach ($rows as $row) { $operation = (new edge_gateway_operations_o())->select((int)$row['id']); $startedAt = $operation->started_at->value() === null ? null : strtotime((string)$operation->started_at->value()); $leaseExpiresAt = $operation->lease_expires_at->value() === null ? null : strtotime((string)$operation->lease_expires_at->value()); $timedOut = $startedAt !== false && $startedAt !== null && ($now - $startedAt) >= self::OPERATION_TIMEOUT_SECONDS; $leaseExpired = $leaseExpiresAt !== false && $leaseExpiresAt !== null && $leaseExpiresAt <= $now; if (!$timedOut && !$leaseExpired) { continue; } $errorMessage = $leaseExpired ? 'Gateway stopped reporting operation progress before the lease expired' : 'Gateway operation timed out'; $summaryLabel = $leaseExpired ? 'Lease expired' : 'Timed out'; $operation->status->set(self::STATUS_FAILED); $operation->completed_at->set($this->now()); $operation->error_code->set(self::ERROR_OPERATION_TIMEOUT); $operation->error_message->set($errorMessage); $operation->lease_expires_at->set(null); $summary = (array)($operation->summary_json->value() ?? []); $summary['label'] = $summaryLabel; $summary['retryable'] = true; $operation->summary_json->set($summary); $this->appendEventRecord( $gatewayId, (int)$operation->id, self::LEVEL_ERROR, self::ERROR_OPERATION_TIMEOUT, $errorMessage, [ 'agent_instance_id' => $operation->agent_instance_id->value(), 'last_progress_at' => $operation->last_progress_at->value(), ] ); } } /** * @throws Exception */ private function applyCompletionSideEffects( int $gatewayId, edge_gateway_operations_o $operation, bool $ok, array $result, string $errorCode, string $errorMessage ): void { $gateway = $this->requireGateway($gatewayId); $metadata = (array)($gateway->metadata_json->value() ?? []); $type = (string)$operation->type->value(); $now = $this->now(); $metadata['last_operation'] = [ 'id' => (int)$operation->id, 'type' => $type, 'status' => $ok ? self::STATUS_COMPLETED : self::STATUS_FAILED, 'completed_at' => $now, 'error_code' => $ok ? null : $errorCode, ]; if ($type === self::TYPE_DISCOVERY) { if ($ok) { $inventory = isset($result['inventory']) && is_array($result['inventory']) ? (array)$result['inventory'] : []; $this->manager()->syncGatewayInventory($gatewayId, $inventory); $gateway->discovery_status->set('READY'); $metadata['last_discovery_completed_at'] = $now; $metadata['last_discovery_error'] = null; } else { $gateway->discovery_status->set('FAILED'); $metadata['last_discovery_error'] = [ 'code' => $errorCode, 'message' => $errorMessage, 'at' => $now, ]; } } elseif ($type === self::TYPE_UPDATE) { if ($ok) { $installedVersion = trim((string)($result['installed_version'] ?? $result['target_version'] ?? $operation->request_json->value()['target_version'] ?? '')); if ($installedVersion !== '') { $gateway->installed_version->set($installedVersion); $gateway->target_version->set($installedVersion); } $metadata['last_update_completed_at'] = $now; $metadata['last_update_error'] = null; } else { $metadata['last_update_error'] = [ 'code' => $errorCode, 'message' => $errorMessage, 'at' => $now, ]; } } elseif ($type === self::TYPE_UNINSTALL) { if ($ok) { $gateway->status->set(edge_gateway_manager::STATUS_OFFLINE); $metadata['uninstalled_at'] = $now; $metadata['uninstall_error'] = null; } else { $metadata['uninstall_error'] = [ 'code' => $errorCode, 'message' => $errorMessage, 'at' => $now, ]; } } $gateway->metadata_json->set($metadata); $this->manager()->logGatewayAudit( $gatewayId, (int)$gateway->department_id->value(), $ok ? 'GATEWAY_OPERATION_COMPLETED' : 'GATEWAY_OPERATION_FAILED', null, [ 'operation_id' => (int)$operation->id, 'type' => $type, 'status' => $ok ? self::STATUS_COMPLETED : self::STATUS_FAILED, 'error_code' => $ok ? null : $errorCode, ] ); } /** * @throws Exception */ private function serializeOperation(edge_gateway_operations_o $operation, bool $includeEvents = true): array { $operationArray = $operation->asArray(); if ($includeEvents) { $operationArray['events'] = $this->listOperationEvents( (int)$operation->gateway_id->value(), (int)$operation->id, 20 ); } return $operationArray; } /** * @throws Exception */ private function requireGateway(int $gatewayId): edge_gateways_o { $gateway = (new edge_gateways_o())->select($gatewayId); if (!$gateway->exists() || $gateway->deleted_at->value() !== null) { throw new Exception('Edge gateway not found'); } return $gateway; } /** * @throws Exception */ private function requireOperation(int $gatewayId, int $operationId): edge_gateway_operations_o { $operation = (new edge_gateway_operations_o())->select($operationId); if (!$operation->exists() || $operation->deleted_at->value() !== null) { throw new Exception('Edge gateway operation not found'); } if ((int)$operation->gateway_id->value() !== $gatewayId) { throw new Exception('Edge gateway operation does not belong to this gateway'); } return $operation; } private function appendEventRecord( int $gatewayId, int $operationId, string $level, ?string $code, string $message, array $context ): array { $eventId = (new edge_gateway_operation_events_o())->add_object([ 'operation_id' => $operationId, 'gateway_id' => $gatewayId, 'level' => $level, 'code' => $code, 'message' => $message, 'context_json' => $context, ]); return (new edge_gateway_operation_events_o())->select($eventId)->asArray(); } private function refreshOperationLease(edge_gateway_operations_o $operation): void { $operation->last_progress_at->set($this->now()); $operation->lease_expires_at->set($this->leaseExpiry()); } /** * @throws Exception */ private function classifyCompletionError(int $gatewayId, string $errorMessage): string { $gateway = $this->requireGateway($gatewayId); $gatewayStatus = edge_gateway_manager::resolveGatewayStatus( (string)$gateway->status->value(), $gateway->last_heartbeat_at->value() === null ? null : (string)$gateway->last_heartbeat_at->value() ); if ($gatewayStatus === edge_gateway_manager::STATUS_OFFLINE) { return self::ERROR_OFFLINE; } if ($gatewayStatus === edge_gateway_manager::STATUS_DEGRADED) { return self::ERROR_STALE_HEARTBEAT; } $normalizedMessage = strtolower(trim($errorMessage)); if (str_contains($normalizedMessage, 'version')) { return self::ERROR_UNSUPPORTED_VERSION; } if (str_contains($normalizedMessage, 'validation')) { return self::ERROR_VALIDATION; } return self::ERROR_VALIDATION; } private function manager(): edge_gateway_manager { return $this->manager ?? new edge_gateway_manager(); } private function now(): string { return date('Y-m-d H:i:s'); } private function leaseExpiry(): string { return date('Y-m-d H:i:s', time() + self::OPERATION_LEASE_SECONDS); } private function normalizeAgentInstanceId(?string $agentInstanceId, int $gatewayId): string { $candidate = trim((string)$agentInstanceId); if ($candidate === '') { return 'gateway-' . $gatewayId; } return substr($candidate, 0, 128); } }