keystone/app/Jobs/Services/RunStep.php

<?php

namespace App\Jobs\Services;

use App\Enums\BuildArtifactStatus;
use App\Enums\OperationStatus;
use App\Enums\RegistryType;
use App\Enums\ServiceStatus;
use App\Models\BuildArtifact;
use App\Models\Environment;
use App\Models\Operation;
use App\Models\OperationStep;
use App\Models\Registry;
use App\Models\Server;
use App\Models\Service;
use App\Models\ServiceReplica;
use App\Models\ServiceSlice;
use App\Services\Operations\RemoteCommandRunner;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Queue\Queueable;

class RunStep implements ShouldQueue
{
    use Queueable;

    public function __construct(
        protected OperationStep $step,
    ) {
        //
    }

    public function handle(): void
    {
        $this->step->load('operation.target');
        $this->step->operation->update([
            'status' => OperationStatus::IN_PROGRESS,
        ]);
        $this->step->update([
            'status' => OperationStatus::IN_PROGRESS,
            'started_at' => now(),
        ]);

        $server = $this->targetServer();

        try {
            $output = app(RemoteCommandRunner::class)->run($server, $this->step->scriptForExecution());
        } catch (\Throwable $exception) {
            $this->failStep($exception->getMessage());

            return;
        }

        $this->step->update([
            'status' => OperationStatus::COMPLETED,
            'finished_at' => now(),
            'logs' => trim($this->step->logs."\n".$output),
            'secrets' => null,
        ]);
        $this->captureRuntimeState();

        // Dispatch the next step if available
        if ($nextStep = $this->step->operation->steps()->where('order', '>', $this->step->order)->orderBy('order', 'asc')->first()) {
            $nextStep->dispatchJob();
        } elseif ($this->dispatchNextChildOperation($this->step->operation)) {
            return;
        } else {
            $this->completeOperation($this->step->operation);
            $this->dispatchNextOperationAfter($this->step->operation);
        }
    }

    private function captureRuntimeState(): void
    {
        $target = $this->step->operation->target;

        if (! $target instanceof ServiceReplica) {
            return;
        }

        $state = $this->step->refresh()->capturedRuntimeState();

        if ($state !== []) {
            $target->update($state);
        }
    }

    private function markTargetCompleted(): void
    {
        $target = $this->step->operation->target;

        if ($target instanceof Service) {
            $target->update([
                'status' => ServiceStatus::RUNNING,
            ]);

            return;
        }

        if ($target instanceof ServiceReplica) {
            $target->update([
                'status' => 'running',
                'health_status' => $target->health_status === 'unknown' ? 'healthy' : $target->health_status,
            ]);
        }
    }

    private function completeOperation(Operation $operation): void
    {
        $operation->update([
            'status' => OperationStatus::COMPLETED,
            'finished_at' => now(),
        ]);

        if ($operation->is($this->step->operation)) {
            $this->markTargetCompleted();
        }

        $this->markRegistryHealthOperationCompleted($operation);
        $this->markRegistryMaintenanceOperationCompleted($operation);
    }

    private function dispatchNextChildOperation(Operation $operation): bool
    {
        $child = $operation->children()
            ->where('status', OperationStatus::PENDING)
            ->orderBy('id')
            ->get()
            ->first(fn (Operation $child): bool => $child->steps()->exists());

        if (! $child) {
            return false;
        }

        $child->steps()->orderBy('order')->first()?->dispatchJob();

        return true;
    }

    private function dispatchNextOperationAfter(Operation $operation): void
    {
        $operation->loadMissing('parent');

        if (! $operation->parent_id) {
            return;
        }

        $nextSibling = $operation->parent
            ?->children()
            ->where('id', '>', $operation->id)
            ->where('status', OperationStatus::PENDING)
            ->orderBy('id')
            ->first();

        if ($nextSibling) {
            $nextSibling->steps()->orderBy('order')->first()?->dispatchJob();

            return;
        }

        $parent = $operation->parent;

        if ($parent && $parent->status === OperationStatus::IN_PROGRESS) {
            $this->completeOperation($parent);
            $this->dispatchNextOperationAfter($parent);
        }
    }

    private function targetServer(): Server
    {
        $target = $this->step->operation->target;

        $server = match (true) {
            $target instanceof ServiceReplica => $target->server,
            $target instanceof Service => $target->replicas()->with('server')->first()?->server ?: $target->server,
            $target instanceof ServiceSlice => $target->service->replicas()->with('server')->first()?->server ?: $target->service->server,
            $target instanceof Server => $target,
            $target instanceof Environment => $target->services()->with(['server', 'replicas.server'])->get()
                ->flatMap(fn (Service $service) => $service->replicas->pluck('server')->filter())
                ->first() ?: $target->services()->with('server')->get()->pluck('server')->filter()->first(),
            default => null,
        };

        if (! $server instanceof Server) {
            throw new \RuntimeException('Operation target does not have a server for SSH execution.');
        }

        return $server;
    }

    public function failed(\Throwable $exception): void
    {
        $this->failStep($exception->getMessage());
    }

    private function failStep(string $message): void
    {
        $this->step->update([
            'status' => OperationStatus::FAILED,
            'finished_at' => now(),
            'error_logs' => $this->step->error_logs."\n".trim($message),
            'secrets' => null,
        ]);

        $this->step->operation->steps()->where('order', '>', $this->step->order)->update([
            'status' => OperationStatus::CANCELLED,
            'secrets' => null,
        ]);

        $this->step->operation->update([
            'status' => OperationStatus::FAILED,
            'finished_at' => now(),
        ]);

        $this->cancelDescendants($this->step->operation);
        $this->cancelPendingSiblingsAndAncestors($this->step->operation);
        $this->markRegistryHealthOperationFailed($this->step->operation, trim($message));
    }

    private function markRegistryHealthOperationCompleted(Operation $operation): void
    {
        if ($operation->kind !== \App\Enums\OperationKind::REGISTRY_HEALTH_CHECK) {
            return;
        }

        if ($operation->parent_id !== null) {
            return;
        }

        $registry = $this->managedRegistryForOperation($operation);

        if (! $registry instanceof Registry) {
            return;
        }

        $checks = collect($registry->readiness_checks ?? [])
            ->map(fn (): string => 'passed')
            ->all();

        $registry->forceFill([
            'readiness_checks' => $checks,
        ])->save();
        $registry->markHealthy('Managed registry smoke checks passed.');
    }

    private function markRegistryHealthOperationFailed(Operation $operation, string $message): void
    {
        if ($operation->kind !== \App\Enums\OperationKind::REGISTRY_HEALTH_CHECK) {
            return;
        }

        $registry = $this->managedRegistryForOperation($operation);

        $registry?->markUnhealthy('Managed registry smoke check failed: '.$message);
    }

    private function markRegistryMaintenanceOperationCompleted(Operation $operation): void
    {
        if ($operation->kind !== \App\Enums\OperationKind::REGISTRY_MAINTENANCE) {
            return;
        }

        $registry = $this->managedRegistryForOperation($operation);

        if (! $registry instanceof Registry) {
            return;
        }

        $artifactIds = collect($operation->metadata['artifact_ids'] ?? [])
            ->filter(fn ($id): bool => is_numeric($id))
            ->map(fn ($id): int => (int) $id)
            ->values();

        BuildArtifact::query()
            ->whereIn('id', $artifactIds)
            ->where('status', BuildArtifactStatus::PRUNABLE)
            ->where('registry_ref', 'like', rtrim((string) $registry->url, '/').'/%')
            ->each(function ($artifact): void {
                $artifact->update([
                    'status' => BuildArtifactStatus::PRUNED,
                    'metadata' => [
                        ...($artifact->metadata ?? []),
                        'pruned_at' => now()->toIso8601String(),
                    ],
                ]);
            });
    }

    private function managedRegistryForOperation(Operation $operation): ?Registry
    {
        $registryId = $operation->metadata['registry_id'] ?? $operation->parent?->metadata['registry_id'] ?? null;

        if ($registryId) {
            $registry = Registry::query()
                ->where('type', RegistryType::MANAGED->value)
                ->find($registryId);

            if ($registry instanceof Registry) {
                return $registry;
            }
        }

        $server = $operation->target;

        if (! $server instanceof Server) {
            $server = $operation->parent?->target;
        }

        if (! $server instanceof Server) {
            return null;
        }

        return Registry::query()
            ->where('type', RegistryType::MANAGED->value)
            ->where('control_server_id', $server->id)
            ->first();
    }

    private function cancelDescendants(Operation $operation): void
    {
        $operation->children()->with('children')->get()->each(function (Operation $child): void {
            $child->steps()->whereIn('status', [OperationStatus::PENDING, OperationStatus::IN_PROGRESS])->update([
                'status' => OperationStatus::CANCELLED,
                'secrets' => null,
            ]);
            $child->update([
                'status' => OperationStatus::CANCELLED,
                'finished_at' => now(),
            ]);
            $this->cancelDescendants($child);
        });
    }

    private function cancelPendingSiblingsAndAncestors(Operation $operation): void
    {
        $operation->loadMissing('parent');

        if (! $operation->parent) {
            return;
        }

        $operation->parent->children()
            ->where('id', '!=', $operation->id)
            ->whereIn('status', [OperationStatus::PENDING, OperationStatus::IN_PROGRESS])
            ->get()
            ->each(function (Operation $sibling): void {
                $sibling->steps()->whereIn('status', [OperationStatus::PENDING, OperationStatus::IN_PROGRESS])->update([
                    'status' => OperationStatus::CANCELLED,
                    'secrets' => null,
                ]);
                $sibling->update([
                    'status' => OperationStatus::CANCELLED,
                    'finished_at' => now(),
                ]);
                $this->cancelDescendants($sibling);
            });

        $operation->parent->update([
            'status' => OperationStatus::FAILED,
            'finished_at' => now(),
        ]);

        $this->cancelPendingSiblingsAndAncestors($operation->parent);
    }
}