breadcrumb-the-shire/lib/Service/Scheduler/SchedulerRunService.php

<?php

namespace MintyPHP\Service\Scheduler;

use MintyPHP\DB;
use MintyPHP\Repository\Scheduler\ScheduledJobRepository;
use MintyPHP\Repository\Scheduler\ScheduledJobRunRepository;
use MintyPHP\Repository\Scheduler\SchedulerRuntimeRepository;
use MintyPHP\Repository\Support\RepoQuery;

/**
 * Orchestrates the execution of scheduled jobs.
 *
 * Acquires a MySQL advisory lock before any work to ensure only one runner
 * process is active at a time (safe for minute-by-minute cron invocations).
 * Individual jobs are dispatched via ScheduledJobRegistry, and every execution
 * attempt – including skips and failures – is written to the run log.
 */
class SchedulerRunService
{
    private const RUNNER_LOCK_NAME = 'scheduled_jobs_runner';

    public static function runDueJobs(int $limit = 20): array
    {
        ScheduledJobService::ensureSystemJobs();
        $startedAt = microtime(true);
        $result = [
            'ok' => true,
            'error' => null,
            'duration_ms' => 0,
            'processed' => 0,
            'success' => 0,
            'failed' => 0,
            'skipped' => 0,
        ];

        if (!self::acquireRunnerLock()) {
            $result['ok'] = false;
            $result['error'] = 'lock_not_acquired';
            $result['duration_ms'] = self::durationMs($startedAt);
            self::updateRuntimeHeartbeat('lock_not_acquired', 'lock_not_acquired');
            return $result;
        }

        try {
            $nowUtc = gmdate('Y-m-d H:i:s');
            $dueJobs = ScheduledJobRepository::listDueJobs($nowUtc, $limit);
            foreach ($dueJobs as $job) {
                $run = self::runOne($job, 'scheduler', null, false);
                $result['processed']++;
                $status = (string) ($run['status'] ?? 'failed');
                if ($status === 'success') {
                    $result['success']++;
                } elseif ($status === 'skipped') {
                    $result['skipped']++;
                } else {
                    $result['failed']++;
                }
            }
        } catch (\Throwable $exception) {
            $result['ok'] = false;
            $result['error'] = 'unexpected_error';
        } finally {
            self::releaseRunnerLock();
            $result['duration_ms'] = self::durationMs($startedAt);
            if ($result['ok']) {
                self::updateRuntimeHeartbeat('ok', null);
            } else {
                $errorCode = self::nullableString($result['error'] ?? null) ?? 'unexpected_error';
                self::updateRuntimeHeartbeat('unexpected_error', $errorCode);
            }
        }

        return $result;
    }

    public static function runJobNow(int $jobId, int $actorUserId): array
    {
        ScheduledJobService::ensureSystemJobs();
        if ($jobId <= 0 || $actorUserId <= 0) {
            return ['ok' => false, 'error' => 'invalid_request'];
        }
        if (!self::acquireRunnerLock()) {
            return ['ok' => false, 'error' => 'lock_not_acquired'];
        }

        try {
            $job = ScheduledJobRepository::find($jobId);
            if (!$job) {
                return ['ok' => false, 'error' => 'job_not_found'];
            }
            $run = self::runOne($job, 'manual', $actorUserId, true);
            return [
                'ok' => in_array((string) ($run['status'] ?? ''), ['success', 'skipped'], true),
                'status' => (string) ($run['status'] ?? 'failed'),
                'error' => (string) ($run['error_code'] ?? ''),
            ];
        } finally {
            self::releaseRunnerLock();
        }
    }

    private static function runOne(array $job, string $triggerType, ?int $actorUserId, bool $force): array
    {
        $jobId = (int) ($job['id'] ?? 0);
        $nowUtc = new \DateTimeImmutable('now', new \DateTimeZone('UTC'));
        $startedAtUtc = $nowUtc->format('Y-m-d H:i:s');

        $triggerType = in_array($triggerType, ['scheduler', 'manual'], true) ? $triggerType : 'scheduler';
        if ($jobId <= 0) {
            return ['status' => 'failed', 'error_code' => 'job_not_found', 'error_message' => null];
        }

        if (!$force && ((int) ($job['enabled'] ?? 0) !== 1 || empty($job['next_run_at']))) {
            return ['status' => 'skipped', 'error_code' => 'job_disabled', 'error_message' => null];
        }

        $markedRunning = ScheduledJobRepository::markRunning($jobId, $startedAtUtc);
        if (!$markedRunning) {
            self::insertRunLog([
                'job_id' => $jobId,
                'job_key' => (string) ($job['job_key'] ?? ''),
                'trigger_type' => $triggerType,
                'status' => 'skipped',
                'actor_user_id' => $actorUserId,
                'started_at' => $startedAtUtc,
                'finished_at' => $startedAtUtc,
                'duration_ms' => 0,
                'error_code' => 'job_already_running',
                'error_message' => null,
                'result_json' => null,
            ]);
            return ['status' => 'skipped', 'error_code' => 'job_already_running', 'error_message' => null];
        }

        $startedMs = microtime(true);
        $status = 'failed';
        $errorCode = null;
        $errorMessage = null;
        $resultPayload = [];

        try {
            $definition = ScheduledJobRegistry::get((string) ($job['job_key'] ?? ''));
            if (!$definition) {
                $status = 'failed';
                $errorCode = 'job_not_registered';
            } else {
                $execution = ScheduledJobRegistry::execute((string) $job['job_key'], $actorUserId);
                $status = in_array((string) $execution['status'], ['success', 'failed', 'skipped'], true)
                    ? (string) $execution['status']
                    : 'failed';
                $errorCode = self::nullableString($execution['error_code']);
                $errorMessage = self::nullableString($execution['error_message']);
                $resultPayload = $execution['result'];
            }
        } catch (\Throwable $exception) {
            $status = 'failed';
            $errorCode = 'unexpected_error';
            $errorMessage = substr($exception->getMessage(), 0, 255) ?: null;
        }

        $finishedAt = new \DateTimeImmutable('now', new \DateTimeZone('UTC'));
        $finishedAtUtc = $finishedAt->format('Y-m-d H:i:s');
        $durationMs = max(0, (int) round((microtime(true) - $startedMs) * 1000));

        $nextRunUtc = null;
        if ((int) ($job['enabled'] ?? 0) === 1) {
            if ($triggerType === 'scheduler' && (int) ($job['catchup_once'] ?? 1) === 0 && !empty($job['next_run_at'])) {
                // catchup_once = 0: advance next_run_at strictly from the originally scheduled
                // time, skipping any windows that already passed. This prevents a backlog storm
                // after downtime – missed slots are dropped, only the next future slot is kept.
                // The guard cap (500) prevents an infinite loop if calculateNextRunUtc has a bug
                // and keeps returning a time that is still in the past.
                $reference = self::parseUtc((string) $job['next_run_at']) ?? $finishedAt;
                $next = ScheduleCalculator::calculateNextRunUtc($job, $reference);
                $guard = 0;
                while ($next && $next <= $finishedAt && $guard < 500) {
                    $next = ScheduleCalculator::calculateNextRunUtc($job, $next);
                    $guard++;
                }
                $nextRunUtc = $next ? $next->format('Y-m-d H:i:s') : null;
            } else {
                // catchup_once = 1 (default): calculate next_run_at from the actual finish time.
                // If a run was delayed or a slot was missed, exactly one catch-up run is scheduled
                // starting from now, then normal scheduling resumes.
                $next = ScheduleCalculator::calculateNextRunUtc($job, $finishedAt);
                $nextRunUtc = $next ? $next->format('Y-m-d H:i:s') : null;
            }
        }

        ScheduledJobRepository::finishRun(
            $jobId,
            $status,
            $startedAtUtc,
            $finishedAtUtc,
            $nextRunUtc,
            $errorCode,
            $errorMessage
        );

        self::insertRunLog([
            'job_id' => $jobId,
            'job_key' => (string) ($job['job_key'] ?? ''),
            'trigger_type' => $triggerType,
            'status' => $status,
            'actor_user_id' => $actorUserId,
            'started_at' => $startedAtUtc,
            'finished_at' => $finishedAtUtc,
            'duration_ms' => $durationMs,
            'error_code' => $errorCode,
            'error_message' => $errorMessage,
            'result_json' => self::encodeResult($resultPayload),
        ]);

        return ['status' => $status, 'error_code' => $errorCode, 'error_message' => $errorMessage];
    }

    private static function insertRunLog(array $data): void
    {
        try {
            ScheduledJobRunRepository::create([
                'run_uuid' => RepoQuery::uuidV4(),
                'job_id' => (int) ($data['job_id'] ?? 0),
                'job_key' => (string) ($data['job_key'] ?? ''),
                'trigger_type' => (string) ($data['trigger_type'] ?? 'scheduler'),
                'status' => (string) ($data['status'] ?? 'failed'),
                'actor_user_id' => ($data['actor_user_id'] ?? null) !== null ? (int) $data['actor_user_id'] : null,
                'started_at' => (string) ($data['started_at'] ?? ''),
                'finished_at' => $data['finished_at'] ?? null,
                'duration_ms' => ($data['duration_ms'] ?? null) !== null ? (int) $data['duration_ms'] : null,
                'error_code' => $data['error_code'] ?? null,
                'error_message' => $data['error_message'] ?? null,
                'result_json' => $data['result_json'] ?? null,
            ]);
        } catch (\Throwable $exception) {
            // fail-open
        }
    }

    private static function encodeResult(array $result): ?string
    {
        if (!$result) {
            return null;
        }
        $encoded = json_encode($result, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES);
        return is_string($encoded) ? $encoded : null;
    }

    private static function parseUtc(string $value): ?\DateTimeImmutable
    {
        $value = trim($value);
        if ($value === '') {
            return null;
        }
        try {
            return new \DateTimeImmutable($value, new \DateTimeZone('UTC'));
        } catch (\Throwable $exception) {
            return null;
        }
    }

    private static function nullableString(mixed $value): ?string
    {
        $value = trim((string) $value);
        return $value !== '' ? $value : null;
    }

    private static function acquireRunnerLock(): bool
    {
        $got = DB::selectValue('select GET_LOCK(?, 0) as got_lock', self::RUNNER_LOCK_NAME);
        return (int) $got === 1;
    }

    private static function updateRuntimeHeartbeat(string $status, ?string $errorCode): void
    {
        try {
            SchedulerRuntimeRepository::touchHeartbeat($status, $errorCode);
        } catch (\Throwable $exception) {
            // fail-open
        }
    }

    private static function releaseRunnerLock(): void
    {
        try {
            DB::selectValue('select RELEASE_LOCK(?) as released_lock', self::RUNNER_LOCK_NAME);
        } catch (\Throwable $exception) {
            // no-op
        }
    }

    private static function durationMs(float $startedAt): int
    {
        return (int) max(0, round((microtime(true) - $startedAt) * 1000));
    }
}