api/services/nginx/app/classes/cron_worker.php

<?php

namespace classes;

use Throwable;
use traits\boolean_normalization_t;

require_once __DIR__ . '/../traits/boolean_normalization_t.php';

class cron_worker
{
    use boolean_normalization_t;

    private cron_scheduler $scheduler;
    private string $worker_id;
    private string $name;
    private string $source;
    private int $poll_seconds;
    private int $heartbeat_seconds;
    private int $max_runtime_seconds;
    private bool $should_stop = false;
    private int $last_heartbeat = 0;

    public function __construct(?cron_scheduler $scheduler = null, array $options = [])
    {
        $this->scheduler = $scheduler ?? new cron_scheduler();
        $this->name = $this->stringOption($options, 'name', 'CRON_WORKER_NAME', 'cron-worker');
        $this->worker_id = $this->stringOption($options, 'worker_id', 'CRON_WORKER_ID', $this->name);
        $this->source = $this->stringOption($options, 'source', 'CRON_WORKER_SOURCE', 'coolify_worker');
        $this->poll_seconds = $this->intOption($options, 'poll_seconds', 'CRON_WORKER_POLL_SECONDS', 15, 1, 300);
        $this->heartbeat_seconds = $this->intOption($options, 'heartbeat_seconds', 'CRON_WORKER_HEARTBEAT_SECONDS', 30, 5, 300);
        $this->max_runtime_seconds = $this->intOption($options, 'max_runtime_seconds', 'CRON_WORKER_MAX_RUNTIME_SECONDS', 0, 0, 86400);
    }

    public function run(): int
    {
        if (!$this->boolOption('CRON_WORKER_ENABLED', true)) {
            $this->heartbeat('disabled', 0, 0, null, true);
            return 0;
        }

        $this->registerSignalHandlers();
        $started = time();
        $this->heartbeat('starting', 0, 0, null, true);

        while (!$this->should_stop) {
            $pollStarted = microtime(true);
            $result = $this->tick();
            $this->writeStatusLine($result);

            if ($this->max_runtime_seconds > 0 && time() - $started >= $this->max_runtime_seconds) {
                $this->should_stop = true;
                break;
            }

            $this->sleepUntilNextPoll($pollStarted + $this->poll_seconds);
        }

        $this->heartbeat('stopped', 0, 0, null, true, true);
        return 0;
    }

    public function tick(): array
    {
        $this->heartbeat('running');
        $loopStartedAt = date('Y-m-d H:i:s');
        $staleRuns = 0;
        $ran = ['count' => 0, 'ran' => []];
        $error = null;
        $status = 'running';

        try {
            $staleRuns = $this->scheduler->markExpiredRunningRuns();
            $ran = $this->scheduler->runDue($this->source);
        } catch (Throwable $throwable) {
            $status = 'failed';
            $error = $throwable->getMessage();
        }

        $this->heartbeat($status, (int)($ran['count'] ?? 0), $staleRuns, $error, true, false, $loopStartedAt);

        return [
            'worker_id' => $this->worker_id,
            'status' => $status,
            'ran' => (int)($ran['count'] ?? 0),
            'stale_runs' => $staleRuns,
            'error' => $error,
        ];
    }

    public function listWorkers(): array
    {
        cron_schema_bootstrap::ensureTables();
        $rows = $this->fetchAll('SELECT * FROM cron_worker_state ORDER BY last_heartbeat_at DESC, worker_id');
        $workers = [];
        foreach ($rows as $row) {
            $workers[] = $this->publicWorker($row);
        }

        return [
            'workers' => $workers,
            'summary' => [
                'total' => count($workers),
                'running' => count(array_filter($workers, static fn(array $worker): bool => ($worker['status'] ?? '') === 'running')),
                'stale' => count(array_filter($workers, static fn(array $worker): bool => (bool)($worker['stale'] ?? false))),
            ],
        ];
    }

    private function registerSignalHandlers(): void
    {
        if (!function_exists('pcntl_signal')) {
            return;
        }

        if (function_exists('pcntl_async_signals')) {
            pcntl_async_signals(true);
        }

        pcntl_signal(SIGTERM, function (): void {
            $this->should_stop = true;
        });
        pcntl_signal(SIGINT, function (): void {
            $this->should_stop = true;
        });
    }

    private function sleepUntilNextPoll(float $nextPollAt): void
    {
        while (!$this->should_stop) {
            $remaining = $nextPollAt - microtime(true);
            if ($remaining <= 0) {
                return;
            }
            usleep((int)(min(1.0, $remaining) * 1000000));
            if (time() - $this->last_heartbeat >= $this->heartbeat_seconds) {
                $this->heartbeat('running');
            }
        }
    }

    private function heartbeat(
        string $status,
        int $runCount = 0,
        int $staleRunCount = 0,
        ?string $error = null,
        bool $force = false,
        bool $stopped = false,
        ?string $loopStartedAt = null
    ): void {
        if (!$force && time() - $this->last_heartbeat < $this->heartbeat_seconds) {
            return;
        }

        cron_schema_bootstrap::ensureTables();
        $this->last_heartbeat = time();
        $now = date('Y-m-d H:i:s');
        $workerId = $this->sql($this->worker_id);
        $name = $this->sql($this->name);
        $hostname = $this->nullableSql(gethostname() ?: null);
        $pid = getmypid() ?: 0;
        $source = $this->sql($this->source);
        $statusSql = $this->sql($status);
        $releaseChannelId = $this->nullableInt($this->env('CRON_WORKER_RELEASE_CHANNEL_ID'));
        $releaseTargetId = $this->nullableInt($this->env('CRON_WORKER_RELEASE_TARGET_ID'));
        $resourceUuid = $this->nullableSql($this->env('COOLIFY_RESOURCE_UUID') ?: $this->env('CRON_WORKER_COOLIFY_RESOURCE_UUID'));
        $resourceType = $this->nullableSql($this->env('COOLIFY_RESOURCE_TYPE') ?: $this->env('CRON_WORKER_COOLIFY_RESOURCE_TYPE') ?: 'application');
        $commitSha = $this->nullableSql($this->commitSha());
        $errorSql = $this->nullableSql($error);
        $loopStarted = $this->nullableSql($loopStartedAt);
        $stoppedAt = $stopped ? $this->sql($now) : 'NULL';
        $nowSql = $this->sql($now);

        $this->query(
            "INSERT INTO cron_worker_state (
                worker_id, name, hostname, pid, source, status, release_channel_id, release_target_id,
                coolify_resource_uuid, coolify_resource_type, commit_sha, poll_seconds, last_run_count,
                last_stale_run_count, last_error, started_at, last_heartbeat_at, last_loop_started_at,
                last_loop_finished_at, last_loop_gap_seconds, consecutive_minute_loops, stopped_at
            ) VALUES (
                $workerId, $name, $hostname, $pid, $source, $statusSql, $releaseChannelId, $releaseTargetId,
                $resourceUuid, $resourceType, $commitSha, $this->poll_seconds, $runCount,
                $staleRunCount, $errorSql, $nowSql, $nowSql, $loopStarted,
                $nowSql, NULL, " . ($loopStartedAt !== null ? '1' : '0') . ", $stoppedAt
            )
            ON DUPLICATE KEY UPDATE
                name = VALUES(name),
                hostname = VALUES(hostname),
                pid = VALUES(pid),
                source = VALUES(source),
                status = VALUES(status),
                release_channel_id = VALUES(release_channel_id),
                release_target_id = VALUES(release_target_id),
                coolify_resource_uuid = VALUES(coolify_resource_uuid),
                coolify_resource_type = VALUES(coolify_resource_type),
                commit_sha = VALUES(commit_sha),
                poll_seconds = VALUES(poll_seconds),
                last_run_count = VALUES(last_run_count),
                last_stale_run_count = VALUES(last_stale_run_count),
                last_error = VALUES(last_error),
                last_heartbeat_at = VALUES(last_heartbeat_at),
                last_loop_gap_seconds = CASE
                    WHEN VALUES(last_loop_started_at) IS NULL OR last_loop_started_at IS NULL THEN last_loop_gap_seconds
                    ELSE GREATEST(0, TIMESTAMPDIFF(SECOND, last_loop_started_at, VALUES(last_loop_started_at)))
                END,
                consecutive_minute_loops = CASE
                    WHEN VALUES(last_loop_started_at) IS NULL THEN consecutive_minute_loops
                    WHEN last_loop_started_at IS NULL THEN 1
                    WHEN TIMESTAMPDIFF(SECOND, last_loop_started_at, VALUES(last_loop_started_at)) BETWEEN 0 AND 60
                        THEN consecutive_minute_loops + 1
                    ELSE 1
                END,
                last_loop_started_at = COALESCE(VALUES(last_loop_started_at), last_loop_started_at),
                last_loop_finished_at = VALUES(last_loop_finished_at),
                stopped_at = VALUES(stopped_at)"
        );
    }

    private function publicWorker(array $row): array
    {
        $heartbeatAt = (string)($row['last_heartbeat_at'] ?? '');
        $heartbeatTs = strtotime($heartbeatAt);
        $threshold = max(60, ((int)($row['poll_seconds'] ?? 15) * 4) + 30);
        $age = $heartbeatTs !== false ? max(0, time() - $heartbeatTs) : null;
        $loopStartedAt = (string)($row['last_loop_started_at'] ?? '');
        $loopStartedTs = strtotime($loopStartedAt);
        $loopAge = $loopStartedTs !== false ? max(0, time() - $loopStartedTs) : null;
        $loopGap = isset($row['last_loop_gap_seconds']) ? (int)$row['last_loop_gap_seconds'] : null;
        $consecutiveMinuteLoops = (int)($row['consecutive_minute_loops'] ?? 0);
        $minuteCadenceVerified = ($row['status'] ?? '') === 'running'
            && $loopAge !== null
            && $loopAge <= 60
            && $loopGap !== null
            && $loopGap <= 60
            && $consecutiveMinuteLoops >= 2;

        return [
            'worker_id' => (string)($row['worker_id'] ?? ''),
            'name' => (string)($row['name'] ?? ''),
            'hostname' => $row['hostname'] ?? null,
            'pid' => isset($row['pid']) ? (int)$row['pid'] : null,
            'source' => (string)($row['source'] ?? ''),
            'status' => (string)($row['status'] ?? 'unknown'),
            'release_channel_id' => isset($row['release_channel_id']) ? (int)$row['release_channel_id'] : null,
            'release_target_id' => isset($row['release_target_id']) ? (int)$row['release_target_id'] : null,
            'coolify_resource_uuid' => $row['coolify_resource_uuid'] ?? null,
            'coolify_resource_type' => $row['coolify_resource_type'] ?? null,
            'commit_sha' => $row['commit_sha'] ?? null,
            'poll_seconds' => (int)($row['poll_seconds'] ?? 0),
            'last_run_count' => (int)($row['last_run_count'] ?? 0),
            'last_stale_run_count' => (int)($row['last_stale_run_count'] ?? 0),
            'last_error' => $row['last_error'] ?? null,
            'started_at' => $row['started_at'] ?? null,
            'last_heartbeat_at' => $heartbeatAt !== '' ? $heartbeatAt : null,
            'last_heartbeat_age_seconds' => $age,
            'last_loop_started_at' => $row['last_loop_started_at'] ?? null,
            'last_loop_finished_at' => $row['last_loop_finished_at'] ?? null,
            'last_loop_age_seconds' => $loopAge,
            'last_loop_gap_seconds' => $loopGap,
            'consecutive_minute_loops' => $consecutiveMinuteLoops,
            'minute_cadence' => [
                'verified' => $minuteCadenceVerified,
                'maximum_gap_seconds' => 60,
                'last_gap_seconds' => $loopGap,
                'consecutive_loops' => $consecutiveMinuteLoops,
            ],
            'stopped_at' => $row['stopped_at'] ?? null,
            'stale' => $age === null || $age > $threshold,
            'stale_after_seconds' => $threshold,
        ];
    }

    private function writeStatusLine(array $result): void
    {
        echo '[' . date('Y-m-d H:i:s') . '][CRON_WORKER] '
            . json_encode($result, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES)
            . PHP_EOL;
    }

    private function stringOption(array $options, string $key, string $env, string $default): string
    {
        $value = trim((string)($options[$key] ?? $this->env($env) ?? ''));
        return $value !== '' ? $value : $default;
    }

    private function intOption(array $options, string $key, string $env, int $default, int $min, int $max): int
    {
        $value = (int)($options[$key] ?? $this->env($env) ?? $default);
        return max($min, min($max, $value));
    }

    private function boolOption(string $env, bool $default): bool
    {
        $value = $this->env($env);
        if ($value === null || trim($value) === '') {
            return $default;
        }

        return self::normalizeBoolean($value);
    }

    private function commitSha(): string
    {
        foreach (['CRON_WORKER_COMMIT_SHA', 'API_COMMIT_SHA', 'RELEASE_COMMIT_SHA', 'COMMIT_SHA', 'GITHUB_SHA'] as $key) {
            $value = trim((string)($this->env($key) ?? ''));
            if ($value !== '') {
                return $value;
            }
        }

        return '';
    }

    private function env(string $key): ?string
    {
        $value = getenv($key);
        if ($value !== false) {
            return (string)$value;
        }

        return isset($_SERVER[$key]) ? (string)$_SERVER[$key] : null;
    }

    private function nullableInt(?string $value): string
    {
        $value = trim((string)$value);
        if ($value === '' || filter_var($value, FILTER_VALIDATE_INT) === false) {
            return 'NULL';
        }

        return (string)max(0, (int)$value);
    }

    private function nullableSql(?string $value): string
    {
        $value = $value !== null ? trim($value) : '';
        return $value === '' ? 'NULL' : $this->sql($value);
    }

    private function fetchAll(string $sql): array
    {
        $result = $this->query($sql);
        if ($result === false || $result === true) {
            return [];
        }

        return $result->fetch_all(MYSQLI_ASSOC);
    }

    private function query(string $sql): \mysqli_result|bool
    {
        global $db;
        return $db->query($sql);
    }

    private function sql(string $value): string
    {
        global $db;
        return "'" . $db->escape_string($value) . "'";
    }
}