coolify/app/Actions/CoolifyTask/RunRemoteProcess.php

<?php

namespace App\Actions\CoolifyTask;

use App\Enums\ActivityTypes;
use App\Enums\ProcessStatus;
use App\Helpers\SshMultiplexingHelper;
use App\Jobs\ApplicationDeploymentJob;
use App\Models\Server;
use App\Support\DatabaseImport\DatabaseImportCleanup;
use App\Support\RemoteProcessCommand;
use App\Support\ResourceStartActivity;
use App\Traits\BroadcastsToTeam;
use Illuminate\Contracts\Process\ProcessResult;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Log;
use Illuminate\Support\Facades\Process;
use Spatie\Activitylog\Models\Activity;

class RunRemoteProcess
{
    public Activity $activity;

    public bool $hide_from_output;

    public bool $ignore_errors;

    public $call_event_on_finish = null;

    public $call_event_data = null;

    protected $time_start;

    protected $current_time;

    protected $last_write_at = 0;

    protected $throttle_interval_ms = 200;

    protected int $counter = 1;

    /**
     * Create a new job instance.
     */
    public function __construct(Activity $activity, bool $hide_from_output = false, bool $ignore_errors = false, $call_event_on_finish = null, $call_event_data = null)
    {
        if ($activity->getExtraProperty('type') !== ActivityTypes::INLINE->value && $activity->getExtraProperty('type') !== ActivityTypes::COMMAND->value) {
            throw new \RuntimeException('Incompatible Activity to run a remote command.');
        }

        $this->activity = $activity;
        $this->hide_from_output = $hide_from_output;
        $this->ignore_errors = $ignore_errors;
        $this->call_event_on_finish = $call_event_on_finish;
        $this->call_event_data = $call_event_data;
    }

    public static function decodeOutput(?Activity $activity = null): string
    {
        if (is_null($activity)) {
            return '';
        }

        try {
            $decoded = json_decode(
                data_get($activity, 'description'),
                associative: true,
                flags: JSON_THROW_ON_ERROR | JSON_UNESCAPED_UNICODE
            );
        } catch (\JsonException $exception) {
            return '';
        }

        return collect($decoded)
            ->sortBy(fn ($i) => $i['order'])
            ->map(fn ($i) => $i['output'])
            ->implode('');
    }

    public function __invoke(): ProcessResult
    {
        $this->time_start = hrtime(true);

        $status = ProcessStatus::IN_PROGRESS;
        $timeout = config('constants.ssh.command_timeout');
        $process = Process::timeout($timeout)->start($this->getCommand(), $this->handleOutput(...));
        // Start from the stored properties and save now, so later output writes only change the log and
        // never write back an old copy over a status that Coolify set meanwhile (for example a stopped import).
        $stored = Activity::query()->whereKey($this->activity->getKey())->first()?->properties ?? $this->activity->properties;
        $this->activity->properties = $stored->merge([
            'process_id' => $process->id(),
        ]);
        $this->activity->save();

        $processResult = $process->wait();

        // A database import whose SSH command ended without the restore result keeps blocking the
        // database until its restore is stopped. The stop event also cleans up, so the normal
        // finish event must not run first.
        if (array_key_exists($processResult->exitCode(), DatabaseImportCleanup::TRANSPORT_FAILURE_REASONS)
            && $this->activity->properties->get('operation') === ResourceStartActivity::DATABASE_IMPORT_OPERATION
            && blank(Activity::query()->whereKey($this->activity->getKey())->first()?->properties?->get(DatabaseImportCleanup::STOP_REQUESTED_PROPERTY))
            && DatabaseImportCleanup::stopAfterTransportFailure($this->activity, $processResult->exitCode())) {
            $this->activity->properties = $this->activity->properties->merge([
                'exitCode' => $processResult->exitCode(),
                'stdout' => $processResult->output(),
                'stderr' => $processResult->errorOutput(),
            ]);
            $this->activity->save();

            if (! $this->ignore_errors) {
                throw new \RuntimeException($processResult->errorOutput(), $processResult->exitCode());
            }

            return $processResult;
        }

        if ($this->activity->properties->get('status') === ProcessStatus::ERROR->value) {
            $status = ProcessStatus::ERROR;
        } else {
            if ($processResult->exitCode() == 0) {
                $status = ProcessStatus::FINISHED;
            } else {
                $status = ProcessStatus::ERROR;
            }
        }

        $properties = [
            'exitCode' => $processResult->exitCode(),
            'stdout' => $processResult->output(),
            'stderr' => $processResult->errorOutput(),
            'status' => $status->value,
        ];

        // Coolify can stop a database import while this job runs (restart or stale import). Keep the
        // stop status and message, because this job only has its own older copy of the properties.
        $stored = Activity::query()->whereKey($this->activity->getKey())->first()?->properties;
        if ($stored !== null && filled($stored->get(DatabaseImportCleanup::STOP_REQUESTED_PROPERTY))) {
            $properties = [
                ...$properties,
                ...$stored->only([DatabaseImportCleanup::STOP_REQUESTED_PROPERTY, 'error'])->all(),
                'status' => ProcessStatus::ERROR->value,
            ];
        }

        $this->activity->properties = $this->activity->properties->merge($properties);
        $this->activity->save();
        if ($this->call_event_on_finish) {
            try {
                self::dispatchFinishEvent($this->activity, $this->call_event_on_finish, $this->call_event_data);
            } catch (\Throwable $e) {
                Log::error('Error calling event: '.$e->getMessage());
            }
        }
        if ($processResult->exitCode() != 0 && ! $this->ignore_errors) {
            throw new \RuntimeException($processResult->errorOutput(), $processResult->exitCode());
        }

        return $processResult;
    }

    /**
     * Dispatches the event that a remote process asked for when it finishes.
     *
     * Without explicit event data, a team event goes to the team of the server that ran the
     * process. The causer is a user, so its id is only used for events on a user channel.
     */
    public static function dispatchFinishEvent(Activity $activity, string $eventName, mixed $eventData = null): void
    {
        $eventClass = "App\\Events\\{$eventName}";

        if (is_null($eventData)) {
            $eventData = in_array(BroadcastsToTeam::class, class_uses_recursive($eventClass), true)
                ? $activity->getExtraProperty('team_id')
                : $activity->causer_id;
        }

        event(new $eventClass($eventData));
    }

    protected function getCommand(): string
    {
        $server_uuid = $this->activity->getExtraProperty('server_uuid');
        $command = RemoteProcessCommand::read($this->activity) ?? throw new \RuntimeException('The command of this task was already removed.');
        $server = Server::whereUuid($server_uuid)->firstOrFail();

        return SshMultiplexingHelper::generateSshCommand($server, $command);
    }

    protected function handleOutput(string $type, string $output)
    {
        if ($this->hide_from_output) {
            return;
        }
        $this->current_time = $this->elapsedTime();
        $this->activity->description = $this->encodeOutput($type, $output);
        if ($this->isAfterLastThrottle()) {
            // Let's write to database.
            DB::transaction(function () {
                $this->activity->save();
                $this->last_write_at = $this->current_time;
            });
        }
    }

    protected function elapsedTime(): int
    {
        $timeMs = (hrtime(true) - $this->time_start) / 1_000_000;

        return intval($timeMs);
    }

    public function encodeOutput($type, $output)
    {
        $outputStack = json_decode($this->activity->description, associative: true, flags: JSON_THROW_ON_ERROR | JSON_UNESCAPED_UNICODE);
        $outputStack[] = [
            'type' => $type,
            'output' => $output,
            'timestamp' => hrtime(true),
            'batch' => ApplicationDeploymentJob::$batch_counter,
            'order' => $this->getLatestCounter(),
        ];

        return json_encode($outputStack, flags: JSON_THROW_ON_ERROR | JSON_UNESCAPED_UNICODE);
    }

    protected function getLatestCounter(): int
    {
        $description = json_decode($this->activity->description, associative: true, flags: JSON_THROW_ON_ERROR | JSON_UNESCAPED_UNICODE);
        if ($description === null || count($description) === 0) {
            return 1;
        }

        return end($description)['order'] + 1;
    }

    /**
     * Determines if it's time to write again to database.
     *
     * @return bool
     */
    protected function isAfterLastThrottle()
    {
        // If DB was never written, then we immediately decide we have to write.
        if ($this->last_write_at === 0) {
            return true;
        }

        return ($this->current_time - $this->throttle_interval_ms) > $this->last_write_at;
    }
}