<?php

namespace classes;

use Exception;
use RuntimeException;
use objects\edge_gateway_operation_events_o;
use objects\edge_gateway_operations_o;
use objects\edge_gateways_o;

class edge_gateway_operation_service
{
    public const TYPE_DISCOVERY = 'DISCOVERY';
    public const TYPE_UPDATE = 'UPDATE';
    public const TYPE_UNINSTALL = 'UNINSTALL';

    public const STATUS_PENDING = 'PENDING';
    public const STATUS_IN_PROGRESS = 'IN_PROGRESS';
    public const STATUS_CANCEL_REQUESTED = 'CANCEL_REQUESTED';
    public const STATUS_CANCELLED = 'CANCELLED';
    public const STATUS_COMPLETED = 'COMPLETED';
    public const STATUS_FAILED = 'FAILED';

    public const LEVEL_INFO = 'INFO';
    public const LEVEL_WARNING = 'WARNING';
    public const LEVEL_ERROR = 'ERROR';

    public const ERROR_OFFLINE = 'EDGE_GATEWAY_OFFLINE';
    public const ERROR_STALE_HEARTBEAT = 'EDGE_GATEWAY_STALE_HEARTBEAT';
    public const ERROR_INVALID_TOKEN = 'EDGE_GATEWAY_INVALID_TOKEN';
    public const ERROR_OPERATION_TIMEOUT = 'EDGE_GATEWAY_OPERATION_TIMEOUT';
    public const ERROR_UNSUPPORTED_VERSION = 'EDGE_GATEWAY_UNSUPPORTED_VERSION';
    public const ERROR_CONFLICT = 'EDGE_GATEWAY_CONFLICT';
    public const ERROR_VALIDATION = 'EDGE_GATEWAY_VALIDATION_FAILED';
    public const ERROR_CANCELLED = 'EDGE_GATEWAY_CANCELLED';

    public const POLL_INTERVAL_MICROSECONDS = 250000;
    public const OPERATION_TIMEOUT_SECONDS = 900;
    public const OPERATION_LEASE_SECONDS = 45;

    public function __construct(private readonly ?edge_gateway_manager $manager = null)
    {
        edge_gateway_schema_bootstrap::ensureTables();
    }

    /**
     * @return array<int,array<string,mixed>>
     * @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<int,array<string,mixed>>
     * @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', 'CANCEL_REQUESTED')
             ORDER BY FIELD(status, 'IN_PROGRESS', 'CANCEL_REQUESTED', '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,
            'cancel_requested' => 0,
            'cancelled' => 0,
            'completed' => 0,
            'failed' => 0,
            'latest_cancelled_at' => null,
            '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_CANCEL_REQUESTED) {
                $summary['cancel_requested'] += 1;
            } elseif ($status === self::STATUS_CANCELLED) {
                $summary['cancelled'] += 1;
                $summary['latest_cancelled_at'] ??= $operation['completed_at'] ?? null;
            } 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]
        );

        $this->refreshGatewayViewCache($gatewayId);
        $this->manager()->notifyBrokerGatewaySync($gatewayId);

        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 cancelOperation(int $gatewayId, int $operationId, ?int $requestedBy = null): array
    {
        $gateway = $this->requireGateway($gatewayId);
        $operation = $this->requireOperation($gatewayId, $operationId);
        $status = (string)$operation->status->value();

        if (in_array($status, [self::STATUS_COMPLETED, self::STATUS_FAILED, self::STATUS_CANCELLED], true)) {
            return $this->serializeOperation($operation, true);
        }

        $message = 'Operation cancelled by operator';
        if ($status === self::STATUS_PENDING) {
            $this->markOperationCancelled(
                $gatewayId,
                $operation,
                $message,
                'OPERATION_CANCELLED',
                ['requested_by' => $requestedBy]
            );
            $this->manager()->logGatewayAudit(
                $gatewayId,
                (int)$gateway->department_id->value(),
                'GATEWAY_OPERATION_CANCELLED',
                $requestedBy,
                ['operation_id' => $operationId, 'type' => (string)$operation->type->value()]
            );
        } elseif ($status === self::STATUS_IN_PROGRESS) {
            $operation->status->set(self::STATUS_CANCEL_REQUESTED);
            $operation->error_code->set(self::ERROR_CANCELLED);
            $operation->error_message->set('Operation cancellation requested by operator');
            $operation->last_progress_at->set($this->now());
            $summary = (array)($operation->summary_json->value() ?? []);
            $summary['label'] = 'Cancellation requested';
            $summary['retryable'] = true;
            $operation->summary_json->set($summary);
            $this->appendEventRecord(
                $gatewayId,
                $operationId,
                self::LEVEL_WARNING,
                'OPERATION_CANCEL_REQUESTED',
                'Operation cancellation requested by operator',
                ['requested_by' => $requestedBy]
            );
            $this->manager()->logGatewayAudit(
                $gatewayId,
                (int)$gateway->department_id->value(),
                'GATEWAY_OPERATION_CANCEL_REQUESTED',
                $requestedBy,
                ['operation_id' => $operationId, 'type' => (string)$operation->type->value()]
            );
        }

        $this->refreshGatewayViewCache($gatewayId);
        $this->manager()->notifyBrokerGatewaySync($gatewayId);

        return $this->serializeOperation($operation, true);
    }

    /**
     * @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 claimBrokerOperation(int $gatewayId, ?string $agentInstanceId = null): ?array
    {
        $this->requireGateway($gatewayId);
        $this->failTimedOutOperations($gatewayId);
        return $this->claimPendingOperation($gatewayId, $this->normalizeAgentInstanceId($agentInstanceId, $gatewayId));
    }

    /**
     * @return array<int,array<string,mixed>>
     * @throws Exception
     */
    public function listBrokerCancellationRequests(int $gatewayId, ?string $agentInstanceId = null): array
    {
        $this->requireGateway($gatewayId);
        $rows = (new edge_gateway_operations_o())->getFieldsWhere([
            'gateway_id' => $gatewayId,
            'status' => self::STATUS_CANCEL_REQUESTED,
            'deleted_at' => null,
        ], ['id']);

        $requestedInstance = $this->normalizeAgentInstanceId($agentInstanceId, $gatewayId);
        $operations = [];
        foreach ($rows as $row) {
            $operation = (new edge_gateway_operations_o())->select((int)$row['id']);
            if (!$operation->exists()) {
                continue;
            }

            $claimedBy = trim((string)($operation->agent_instance_id->value() ?? ''));
            if ($claimedBy !== '' && $requestedInstance !== '' && $claimedBy !== $requestedInstance) {
                continue;
            }

            $operations[] = $this->serializeOperation($operation, true);
        }

        return $operations;
    }

    /**
     * @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,
            self::STATUS_CANCEL_REQUESTED,
            self::STATUS_CANCELLED,
        ], true)) {
            return $this->serializePersistedOperation($operationId, 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);

        if ($level === self::LEVEL_ERROR) {
            $this->markOperationFailedFromEvent($gatewayId, $operation, $code, $message, $context);
        }

        $this->clearObjectPropertyCache('edge_gateway_operations', $operationId);
        $this->clearGatewayViewCache($gatewayId);

        return $this->serializePersistedOperation($operationId, true);
    }

    /**
     * @throws Exception
     */
    public function appendBrokerOperationEvent(int $gatewayId, int $operationId, array $payload): array
    {
        $this->requireGateway($gatewayId);
        return $this->appendOperationEventWithoutAuthentication($gatewayId, $operationId, $payload);
    }

    /**
     * @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,
            self::STATUS_CANCELLED,
        ], true)) {
            return $this->serializePersistedOperation($operationId, 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'] ?? ''));
        $status = (string)$operation->status->value();
        if ($status === self::STATUS_CANCEL_REQUESTED && !$ok && $errorCode === '') {
            $errorCode = self::ERROR_CANCELLED;
        }
        if (!$ok && $errorCode === '') {
            $errorCode = $this->classifyCompletionError($gatewayId, $errorMessage);
        }
        if (!$ok && $errorMessage === '') {
            $errorMessage = $errorCode === self::ERROR_CANCELLED
                ? 'Gateway operation cancelled'
                : 'Gateway operation failed';
        }

        $finalStatus = $ok
            ? self::STATUS_COMPLETED
            : (($status === self::STATUS_CANCEL_REQUESTED && $errorCode === self::ERROR_CANCELLED)
                ? self::STATUS_CANCELLED
                : self::STATUS_FAILED);
        $operation->status->set($finalStatus);
        $operation->result_json->set($result);
        $operation->error_code->set($ok ? null : ($finalStatus === self::STATUS_CANCELLED ? self::ERROR_CANCELLED : $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'] = match ($finalStatus) {
            self::STATUS_COMPLETED => 'Completed',
            self::STATUS_CANCELLED => 'Cancelled',
            default => 'Failed',
        };
        $summary['progress'] = $finalStatus === self::STATUS_CANCELLED
            ? max(0, min(100, (int)($summary['progress'] ?? 0)))
            : 100;
        $summary['retryable'] = $finalStatus === self::STATUS_CANCELLED
            ? true
            : (!$ok && $errorCode !== self::ERROR_UNSUPPORTED_VERSION);
        $operation->summary_json->set($summary);

        $this->appendEventRecord(
            $gatewayId,
            $operationId,
            $finalStatus === self::STATUS_COMPLETED
                ? self::LEVEL_INFO
                : ($finalStatus === self::STATUS_CANCELLED ? self::LEVEL_WARNING : self::LEVEL_ERROR),
            $finalStatus === self::STATUS_COMPLETED
                ? 'OPERATION_COMPLETED'
                : ($finalStatus === self::STATUS_CANCELLED ? 'OPERATION_CANCELLED' : $errorCode),
            $finalStatus === self::STATUS_COMPLETED
                ? 'Operation completed successfully'
                : $errorMessage,
            $result
        );

        if ($finalStatus !== self::STATUS_CANCELLED) {
            $this->applyCompletionSideEffects(
                $gatewayId,
                $operation,
                $ok,
                $result,
                $errorCode,
                $errorMessage
            );
        }
        $this->clearObjectPropertyCache('edge_gateway_operations', $operationId);
        $this->clearGatewayViewCache($gatewayId);
        $this->manager()->notifyBrokerGatewaySync($gatewayId);

        return $this->serializePersistedOperation($operationId, true);
    }

    /**
     * @throws Exception
     */
    public function completeBrokerOperation(int $gatewayId, int $operationId, array $payload): array
    {
        $this->requireGateway($gatewayId);
        return $this->completeOperationWithoutAuthentication($gatewayId, $operationId, $payload);
    }

    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_STACK_SERVICE_NAME,
                'install_dir' => edge_gateway_manager::DEFAULT_INSTALL_DIR,
            ], $request);
        }

        return $request;
    }

    /**
     * @throws Exception
     */
    private function claimPendingOperation(int $gatewayId, string $agentInstanceId): ?array
    {
        $pdo = db::getPDO();
        $pdo->beginTransaction();

        try {
            $statement = $pdo->prepare(
                "SELECT id, type, attempt_count, summary_json
                 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;
            }

            $operationId = (int)$row['id'];
            $startedAt = $this->now();
            $leaseExpiresAt = $this->leaseExpiry();
            $attemptCount = ((int)($row['attempt_count'] ?? 0)) + 1;
            $summary = isset($row['summary_json']) && is_string($row['summary_json'])
                ? json_decode($row['summary_json'], true)
                : [];
            if (!is_array($summary)) {
                $summary = [];
            }
            $summary['label'] = 'Gateway is processing the operation';
            $summary['progress'] = max(5, (int)($summary['progress'] ?? 0));
            $summary['claimed_by'] = $agentInstanceId;
            $encodedSummary = json_encode($summary, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES);
            if (!is_string($encodedSummary)) {
                $encodedSummary = '[]';
            }

            $update = $pdo->prepare(
                "UPDATE edge_gateway_operations
                 SET status = :status,
                     started_at = :started_at,
                     agent_instance_id = :agent_instance_id,
                     last_progress_at = :last_progress_at,
                     lease_expires_at = :lease_expires_at,
                     attempt_count = :attempt_count,
                     summary_json = :summary_json
                 WHERE id = :id"
            );
            $update->execute([
                ':status' => self::STATUS_IN_PROGRESS,
                ':started_at' => $startedAt,
                ':agent_instance_id' => $agentInstanceId,
                ':last_progress_at' => $startedAt,
                ':lease_expires_at' => $leaseExpiresAt,
                ':attempt_count' => $attemptCount,
                ':summary_json' => $encodedSummary,
                ':id' => $operationId,
            ]);
            $pdo->commit();

            $this->clearObjectPropertyCache('edge_gateway_operations', $operationId);
            $operation = $this->fetchOperationRecord($operationId, $pdo);
            if ($operation === null) {
                throw new RuntimeException('Claimed edge gateway operation could not be reloaded');
            }
            $this->appendEventRecord(
                $gatewayId,
                $operationId,
                self::LEVEL_INFO,
                'OPERATION_STARTED',
                'Gateway started processing the operation',
                [
                    'type' => (string)($row['type'] ?? $operation['type'] ?? ''),
                    'agent_instance_id' => $agentInstanceId,
                    'attempt_count' => $attemptCount,
                    'stage' => self::STATUS_IN_PROGRESS,
                ]
            );

            $this->clearGatewayViewCache($gatewayId, $pdo);

            return $this->serializeOperationRecord($operation, true);
        } catch (\Throwable $throwable) {
            if ($pdo->inTransaction()) {
                $pdo->rollBack();
            }
            throw $throwable;
        }
    }

    /**
     * @throws Exception
     */
    private function failTimedOutOperations(int $gatewayId): void
    {
        $statement = db::getPDO()->prepare(
            "SELECT id
             FROM edge_gateway_operations
             WHERE gateway_id = :gateway_id
               AND deleted_at IS NULL
               AND status IN ('IN_PROGRESS', 'CANCEL_REQUESTED')"
        );
        $statement->execute([':gateway_id' => $gatewayId]);
        $rows = $statement->fetchAll();

        $now = time();
        foreach ($rows as $row) {
            $operation = (new edge_gateway_operations_o())->select((int)$row['id']);
            $status = (string)$operation->status->value();
            $startedAt = edge_gateway_manager::parseApplicationDateTime(
                $operation->started_at->value() === null ? null : (string)$operation->started_at->value()
            );
            $leaseExpiresAt = edge_gateway_manager::parseApplicationDateTime(
                $operation->lease_expires_at->value() === null ? null : (string)$operation->lease_expires_at->value()
            );
            $timedOut = $startedAt !== null
                && ($now - $startedAt) >= self::OPERATION_TIMEOUT_SECONDS;
            $leaseExpired = $leaseExpiresAt !== null
                && $leaseExpiresAt <= $now;

            if (!$timedOut && !$leaseExpired) {
                continue;
            }

            if ($status === self::STATUS_CANCEL_REQUESTED) {
                $this->markOperationCancelled(
                    $gatewayId,
                    $operation,
                    'Gateway did not acknowledge cancellation before the operation lease expired',
                    'OPERATION_CANCELLED',
                    [
                        'agent_instance_id' => $operation->agent_instance_id->value(),
                        'last_progress_at' => $operation->last_progress_at->value(),
                    ]
                );
                $this->refreshGatewayViewCache($gatewayId);
                $this->manager()->notifyBrokerGatewaySync($gatewayId);
                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(),
                ]
            );

            $this->refreshGatewayViewCache($gatewayId);
            $this->manager()->notifyBrokerGatewaySync($gatewayId);
        }
    }

    /**
     * @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) {
            $requestedUpdate = is_array($operation->request_json->value()) ? (array)$operation->request_json->value() : [];
            $stagedVersion = trim((string)($result['staged_version'] ?? $result['target_version'] ?? $requestedUpdate['target_version'] ?? ''));
            $applied = array_key_exists('applied', $result)
                ? (bool)$result['applied']
                : trim((string)($result['installed_version'] ?? '')) !== '';
            if ($ok) {
                $installedVersion = trim((string)($result['installed_version'] ?? ($applied ? $stagedVersion : '')));
                if ($applied && $installedVersion !== '') {
                    $gateway->installed_version->set($installedVersion);
                    $gateway->target_version->set($installedVersion);
                }
                $metadata['last_update_completed_at'] = $now;
                $metadata['last_update_error'] = null;
                if ($stagedVersion !== '') {
                    $metadata['staged_version'] = [
                        'target_version' => $stagedVersion,
                        'staged_at' => $result['staged_at'] ?? $now,
                        'apply_after' => $result['apply_after'] ?? null,
                        'status' => $applied ? 'APPLIED' : 'STAGED',
                    ];
                }
                if (isset($result['update_window'])) {
                    $metadata['update_window'] = (string)$result['update_window'];
                }
                if (isset($result['rollback_status']) && is_array($result['rollback_status'])) {
                    $metadata['rollback_status'] = (array)$result['rollback_status'];
                }
            } 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 {
        $stage = trim((string)($context['stage'] ?? ''));
        if ($stage === '') {
            $operationRecord = $this->fetchOperationRecord($operationId);
            $stage = trim((string)($operationRecord['status'] ?? ''));
        }
        if ($stage === '') {
            $stage = 'RECORDED';
        }

        $eventId = (new edge_gateway_operation_events_o())->add_object([
            'operation_id' => $operationId,
            'gateway_id' => $gatewayId,
            'stage' => $stage,
            '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());
    }

    private function refreshGatewayViewCache(int $gatewayId): void
    {
        edge_gateway_view_cache::syncGateway($this->manager()->getGateway($gatewayId));
    }

    private function clearGatewayViewCache(int $gatewayId, ?\PDO $pdo = null): void
    {
        $statement = ($pdo ?? db::getPDO())->prepare(
            "SELECT department_id
             FROM edge_gateways
             WHERE id = :id
               AND deleted_at IS NULL
             LIMIT 1"
        );
        $statement->execute([':id' => $gatewayId]);
        $row = $statement->fetch();
        $departmentId = is_array($row) && isset($row['department_id']) ? (int)$row['department_id'] : null;
        edge_gateway_view_cache::clearGateway($gatewayId, $departmentId);
    }

    private function clearObjectPropertyCache(string $table, int $id): void
    {
        if ($id <= 0 || !defined('redis')) {
            return;
        }

        $normalizedTable = trim($table, " `\t\n\r\0\x0B");
        redis->clear_keys('obj_prop:' . $normalizedTable . ':' . $id . ':*');
    }

    private function fetchOperationRecord(int $operationId, ?\PDO $pdo = null): ?array
    {
        $statement = ($pdo ?? db::getPDO())->prepare(
            "SELECT id,
                    gateway_id,
                    type,
                    operation_type,
                    status,
                    request_json,
                    summary_json,
                    result_json,
                    error_code,
                    error_message,
                    correlation_id,
                    agent_instance_id,
                    lease_expires_at,
                    last_progress_at,
                    attempt_count,
                    requested_by,
                    requested_at,
                    started_at,
                    completed_at,
                    created_at,
                    updated_at
             FROM edge_gateway_operations
             WHERE id = :id
               AND deleted_at IS NULL
             LIMIT 1"
        );
        $statement->execute([':id' => $operationId]);
        $row = $statement->fetch();

        return is_array($row) ? $row : null;
    }

    /**
     * @return array<string,mixed>
     */
    private function serializePersistedOperation(int $operationId, bool $includeEvents = true): array
    {
        $operation = $this->fetchOperationRecord($operationId);
        if ($operation === null) {
            throw new RuntimeException('Edge gateway operation could not be reloaded');
        }

        return $this->serializeOperationRecord($operation, $includeEvents);
    }

    /**
     * @param array<string,mixed> $operation
     * @return array<string,mixed>
     */
    private function serializeOperationRecord(array $operation, bool $includeEvents = true): array
    {
        $operationId = (int)($operation['id'] ?? 0);
        $gatewayId = (int)($operation['gateway_id'] ?? 0);

        $payload = [
            'id' => $operationId,
            'gateway_id' => $gatewayId,
            'type' => (string)($operation['type'] ?? $operation['operation_type'] ?? ''),
            'status' => (string)($operation['status'] ?? self::STATUS_PENDING),
            'request' => $this->decodeJsonRecord($operation['request_json'] ?? []),
            'summary' => $this->decodeJsonRecord($operation['summary_json'] ?? []),
            'result' => $this->decodeJsonRecord($operation['result_json'] ?? []),
            'error_code' => isset($operation['error_code']) ? (string)$operation['error_code'] : null,
            'error_message' => isset($operation['error_message']) ? (string)$operation['error_message'] : null,
            'correlation_id' => (string)($operation['correlation_id'] ?? ''),
            'agent_instance_id' => isset($operation['agent_instance_id']) ? (string)$operation['agent_instance_id'] : null,
            'lease_expires_at' => isset($operation['lease_expires_at']) ? (string)$operation['lease_expires_at'] : null,
            'last_progress_at' => isset($operation['last_progress_at']) ? (string)$operation['last_progress_at'] : null,
            'attempt_count' => (int)($operation['attempt_count'] ?? 0),
            'requested_by' => isset($operation['requested_by']) ? (int)$operation['requested_by'] : null,
            'requested_at' => isset($operation['requested_at']) ? (string)$operation['requested_at'] : '',
            'started_at' => isset($operation['started_at']) ? (string)$operation['started_at'] : null,
            'completed_at' => isset($operation['completed_at']) ? (string)$operation['completed_at'] : null,
            'created_at' => isset($operation['created_at']) ? (string)$operation['created_at'] : '',
            'updated_at' => isset($operation['updated_at']) ? (string)$operation['updated_at'] : null,
        ];

        if ($includeEvents && $gatewayId > 0 && $operationId > 0) {
            $payload['events'] = $this->listOperationEvents($gatewayId, $operationId, 20);
        }

        return $payload;
    }

    /**
     * @return array<mixed>
     */
    private function decodeJsonRecord(mixed $value): array
    {
        if (is_array($value)) {
            return $value;
        }

        if (!is_string($value) || trim($value) === '') {
            return [];
        }

        $decoded = json_decode($value, true);

        return is_array($decoded) ? $decoded : [];
    }

    /**
     * @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, 'cancel')) {
            return self::ERROR_CANCELLED;
        }
        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 edge_gateway_manager::formatApplicationDateTime(time());
    }

    private function leaseExpiry(): string
    {
        return edge_gateway_manager::formatApplicationDateTime(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);
    }

    private function markOperationFailedFromEvent(
        int $gatewayId,
        edge_gateway_operations_o $operation,
        ?string $code,
        string $message,
        array $context
    ): void {
        $operation->status->set(self::STATUS_FAILED);
        $operation->error_code->set($code !== null && trim($code) !== '' ? trim($code) : self::ERROR_VALIDATION);
        $operation->error_message->set($message);
        $operation->completed_at->set($this->now());
        $operation->lease_expires_at->set(null);
        $summary = (array)($operation->summary_json->value() ?? []);
        $summary['label'] = 'Failed';
        $summary['retryable'] = ((string)$operation->error_code->value()) !== self::ERROR_UNSUPPORTED_VERSION;
        if (isset($context['progress'])) {
            $summary['progress'] = max(0, min(100, (int)$context['progress']));
        }
        $operation->summary_json->set($summary);
    }

    private function markOperationCancelled(
        int $gatewayId,
        edge_gateway_operations_o $operation,
        string $message,
        string $eventCode,
        array $context = []
    ): void {
        $operation->status->set(self::STATUS_CANCELLED);
        $operation->error_code->set(self::ERROR_CANCELLED);
        $operation->error_message->set($message);
        $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'] = 'Cancelled';
        $summary['retryable'] = true;
        $summary['progress'] = max(0, min(100, (int)($summary['progress'] ?? 0)));
        $operation->summary_json->set($summary);
        $this->appendEventRecord(
            $gatewayId,
            (int)$operation->id,
            self::LEVEL_WARNING,
            $eventCode,
            $message,
            $context
        );
    }

    /**
     * @throws Exception
     */
    private function appendOperationEventWithoutAuthentication(int $gatewayId, int $operationId, array $payload): array
    {
        $operation = $this->requireOperation($gatewayId, $operationId);
        if (in_array((string)$operation->status->value(), [
            self::STATUS_COMPLETED,
            self::STATUS_FAILED,
            self::STATUS_CANCEL_REQUESTED,
            self::STATUS_CANCELLED,
        ], true)) {
            return $this->serializePersistedOperation($operationId, 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);

        if ($level === self::LEVEL_ERROR) {
            $this->markOperationFailedFromEvent($gatewayId, $operation, $code, $message, $context);
        }

        $this->clearObjectPropertyCache('edge_gateway_operations', $operationId);
        $this->clearGatewayViewCache($gatewayId);

        return $this->serializePersistedOperation($operationId, true);
    }

    /**
     * @throws Exception
     */
    private function completeOperationWithoutAuthentication(int $gatewayId, int $operationId, array $payload): array
    {
        $operation = $this->requireOperation($gatewayId, $operationId);
        if (in_array((string)$operation->status->value(), [
            self::STATUS_COMPLETED,
            self::STATUS_FAILED,
            self::STATUS_CANCELLED,
        ], true)) {
            return $this->serializePersistedOperation($operationId, 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'] ?? ''));
        $status = (string)$operation->status->value();
        if ($status === self::STATUS_CANCEL_REQUESTED && !$ok && $errorCode === '') {
            $errorCode = self::ERROR_CANCELLED;
        }
        if (!$ok && $errorCode === '') {
            $errorCode = $this->classifyCompletionError($gatewayId, $errorMessage);
        }
        if (!$ok && $errorMessage === '') {
            $errorMessage = $errorCode === self::ERROR_CANCELLED
                ? 'Gateway operation cancelled'
                : 'Gateway operation failed';
        }

        $finalStatus = $ok
            ? self::STATUS_COMPLETED
            : (($status === self::STATUS_CANCEL_REQUESTED && $errorCode === self::ERROR_CANCELLED)
                ? self::STATUS_CANCELLED
                : self::STATUS_FAILED);
        $operation->status->set($finalStatus);
        $operation->result_json->set($result);
        $operation->error_code->set($ok ? null : ($finalStatus === self::STATUS_CANCELLED ? self::ERROR_CANCELLED : $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'] = match ($finalStatus) {
            self::STATUS_COMPLETED => 'Completed',
            self::STATUS_CANCELLED => 'Cancelled',
            default => 'Failed',
        };
        $summary['progress'] = $finalStatus === self::STATUS_CANCELLED
            ? max(0, min(100, (int)($summary['progress'] ?? 0)))
            : 100;
        $summary['retryable'] = $finalStatus === self::STATUS_CANCELLED
            ? true
            : (!$ok && $errorCode !== self::ERROR_UNSUPPORTED_VERSION);
        $operation->summary_json->set($summary);

        $this->appendEventRecord(
            $gatewayId,
            $operationId,
            $finalStatus === self::STATUS_COMPLETED
                ? self::LEVEL_INFO
                : ($finalStatus === self::STATUS_CANCELLED ? self::LEVEL_WARNING : self::LEVEL_ERROR),
            $finalStatus === self::STATUS_COMPLETED
                ? 'OPERATION_COMPLETED'
                : ($finalStatus === self::STATUS_CANCELLED ? 'OPERATION_CANCELLED' : $errorCode),
            $finalStatus === self::STATUS_COMPLETED
                ? 'Operation completed successfully'
                : $errorMessage,
            $result
        );

        if ($finalStatus !== self::STATUS_CANCELLED) {
            $this->applyCompletionSideEffects(
                $gatewayId,
                $operation,
                $ok,
                $result,
                $errorCode,
                $errorMessage
            );
        }
        $this->clearObjectPropertyCache('edge_gateway_operations', $operationId);
        $this->clearGatewayViewCache($gatewayId);
        $this->manager()->notifyBrokerGatewaySync($gatewayId);

        return $this->serializePersistedOperation($operationId, true);
    }
}
