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

<?php

namespace MintyPHP\Service\Scheduler;

use MintyPHP\App\Module\ModuleEventDispatcher;
use MintyPHP\App\Module\ModuleEvents;
use MintyPHP\Domain\Taxonomy\ScheduledJobRunStatus;
use MintyPHP\Domain\Taxonomy\ScheduledJobTriggerType;
use MintyPHP\Domain\Taxonomy\SchedulerRuntimeResult;
use MintyPHP\Repository\Scheduler\ScheduledJobRepositoryInterface;
use MintyPHP\Repository\Scheduler\ScheduledJobRunRepositoryInterface;
use MintyPHP\Repository\Scheduler\SchedulerRuntimeRepositoryInterface;
use MintyPHP\Repository\Support\DatabaseSessionRepository;
use MintyPHP\Repository\Support\RepoQuery;
use MintyPHP\Service\Audit\AuditRecorderInterface;

/**
 * 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 function __construct(
        private readonly ScheduledJobService $scheduledJobService,
        private readonly ScheduledJobRepositoryInterface $scheduledJobRepository,
        private readonly ScheduledJobRunRepositoryInterface $scheduledJobRunRepository,
        private readonly SchedulerRuntimeRepositoryInterface $schedulerRuntimeRepository,
        private readonly ScheduledJobRegistry $scheduledJobRegistry,
        private readonly ScheduleCalculator $scheduleCalculator,
        private readonly DatabaseSessionRepository $databaseSessionRepository,
        private readonly AuditRecorderInterface $systemAuditService,
        private readonly ?ModuleEventDispatcher $eventDispatcher = null
    ) {
    }

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

        if (!$this->acquireRunnerLock()) {
            $result['ok'] = false;
            $result['error'] = SchedulerRuntimeResult::LockNotAcquired->value;
            $result['duration_ms'] = $this->durationMs($startedAt);
            $this->updateRuntimeHeartbeat(
                SchedulerRuntimeResult::LockNotAcquired->value,
                SchedulerRuntimeResult::LockNotAcquired->value
            );
            $this->recordSchedulerRunEvent(ScheduledJobTriggerType::Scheduler->value, $result);
            return $result;
        }

        try {
            $nowUtc = gmdate('Y-m-d H:i:s');
            $dueJobs = $this->scheduledJobRepository->listDueJobs($nowUtc, $limit);
            foreach ($dueJobs as $job) {
                $run = $this->runOne($job, ScheduledJobTriggerType::Scheduler->value, null, false);
                $result['processed']++;
                $status = ScheduledJobRunStatus::normalizeOr(
                    (string) ($run['status'] ?? ''),
                    ScheduledJobRunStatus::Failed
                )->value;
                if ($status === ScheduledJobRunStatus::Success->value) {
                    $result['success']++;
                } elseif ($status === ScheduledJobRunStatus::Skipped->value) {
                    $result['skipped']++;
                } else {
                    $result['failed']++;
                }
            }
        } catch (\Throwable $exception) {
            $result['ok'] = false;
            $result['error'] = SchedulerRuntimeResult::UnexpectedError->value;
        } finally {
            $this->releaseRunnerLock();
            $result['duration_ms'] = $this->durationMs($startedAt);
            if ($result['ok']) {
                $this->updateRuntimeHeartbeat(SchedulerRuntimeResult::Ok->value, null);
            } else {
                $errorCode = $this->nullableString($result['error']) ?? SchedulerRuntimeResult::UnexpectedError->value;
                $this->updateRuntimeHeartbeat(SchedulerRuntimeResult::UnexpectedError->value, $errorCode);
            }
        }

        $this->recordSchedulerRunEvent(ScheduledJobTriggerType::Scheduler->value, $result);

        return $result;
    }

    public function runJobNow(int $jobId, int $actorUserId): array
    {
        $this->scheduledJobService->ensureSystemJobs();
        if ($jobId <= 0 || $actorUserId <= 0) {
            $result = ['ok' => false, 'error' => 'invalid_request'];
            $this->recordSchedulerRunEvent(ScheduledJobTriggerType::Manual->value, $result, $actorUserId);
            return $result;
        }
        if (!$this->acquireRunnerLock()) {
            $result = ['ok' => false, 'error' => SchedulerRuntimeResult::LockNotAcquired->value];
            $this->recordSchedulerRunEvent(ScheduledJobTriggerType::Manual->value, $result, $actorUserId);
            return $result;
        }

        try {
            $job = $this->scheduledJobRepository->find($jobId);
            if (!$job) {
                $result = ['ok' => false, 'error' => 'job_not_found'];
                $this->recordSchedulerRunEvent(ScheduledJobTriggerType::Manual->value, $result, $actorUserId);
                return $result;
            }
            $run = $this->runOne($job, ScheduledJobTriggerType::Manual->value, $actorUserId, true);
            $runStatus = ScheduledJobRunStatus::normalizeOr(
                (string) ($run['status'] ?? ''),
                ScheduledJobRunStatus::Failed
            )->value;
            $result = [
                'ok' => in_array($runStatus, [ScheduledJobRunStatus::Success->value, ScheduledJobRunStatus::Skipped->value], true),
                'status' => $runStatus,
                'error' => (string) ($run['error_code'] ?? ''),
            ];
            $this->recordSchedulerRunEvent(ScheduledJobTriggerType::Manual->value, $result, $actorUserId, [
                'processed' => 1,
                'success' => (int) ($result['status'] === ScheduledJobRunStatus::Success->value),
                'failed' => (int) ($result['status'] === ScheduledJobRunStatus::Failed->value),
                'skipped' => (int) ($result['status'] === ScheduledJobRunStatus::Skipped->value),
            ]);
            return $result;
        } finally {
            $this->releaseRunnerLock();
        }
    }

    private 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 = ScheduledJobTriggerType::normalizeOr($triggerType, ScheduledJobTriggerType::Scheduler)->value;
        if ($jobId <= 0) {
            return ['status' => ScheduledJobRunStatus::Failed->value, 'error_code' => 'job_not_found', 'error_message' => null];
        }

        if (!$force && ((int) ($job['enabled'] ?? 0) !== 1 || empty($job['next_run_at']))) {
            $run = ['status' => ScheduledJobRunStatus::Skipped->value, 'error_code' => 'job_disabled', 'error_message' => null];
            $this->recordSchedulerJobEvent($jobId, (string) ($job['job_key'] ?? ''), $triggerType, $actorUserId, $run, 0);
            return $run;
        }

        // markRunning uses a conditional update — failure means another process already claimed this job.
        $markedRunning = $this->scheduledJobRepository->markRunning($jobId, $startedAtUtc);
        if (!$markedRunning) {
            $this->insertRunLog([
                'job_id' => $jobId,
                'job_key' => (string) ($job['job_key'] ?? ''),
                'trigger_type' => $triggerType,
                'status' => ScheduledJobRunStatus::Skipped->value,
                'actor_user_id' => $actorUserId,
                'started_at' => $startedAtUtc,
                'finished_at' => $startedAtUtc,
                'duration_ms' => 0,
                'error_code' => 'job_already_running',
                'error_message' => null,
                'result_json' => null,
            ]);
            $run = ['status' => ScheduledJobRunStatus::Skipped->value, 'error_code' => 'job_already_running', 'error_message' => null];
            $this->recordSchedulerJobEvent($jobId, (string) ($job['job_key'] ?? ''), $triggerType, $actorUserId, $run, 0);
            return $run;
        }

        $startedMs = microtime(true);
        $status = ScheduledJobRunStatus::Failed->value;
        $errorCode = null;
        $errorMessage = null;
        $resultPayload = [];

        try {
            $definition = $this->scheduledJobRegistry->get((string) ($job['job_key'] ?? ''));
            if (!$definition) {
                $status = ScheduledJobRunStatus::Failed->value;
                $errorCode = 'job_not_registered';
            } else {
                $execution = $this->scheduledJobRegistry->execute((string) $job['job_key'], $actorUserId);
                $status = ScheduledJobRunStatus::tryNormalize((string) $execution['status'])->value
                    ?? ScheduledJobRunStatus::Failed->value;
                $errorCode = $this->nullableString($execution['error_code']);
                $errorMessage = $this->nullableString($execution['error_message']);
                $resultPayload = $execution['result'];
            }
        } catch (\Throwable $exception) {
            $status = ScheduledJobRunStatus::Failed->value;
            $errorCode = SchedulerRuntimeResult::UnexpectedError->value;
            $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 === ScheduledJobTriggerType::Scheduler->value
                && (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 = $this->parseUtc((string) $job['next_run_at']) ?? $finishedAt;
                $next = $this->scheduleCalculator->calculateNextRunUtc($job, $reference);
                $guard = 0;
                while ($next && $next <= $finishedAt && $guard < 500) {
                    $next = $this->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 = $this->scheduleCalculator->calculateNextRunUtc($job, $finishedAt);
                $nextRunUtc = $next ? $next->format('Y-m-d H:i:s') : null;
            }
        }

        $this->scheduledJobRepository->finishRun(
            $jobId,
            $status,
            $startedAtUtc,
            $finishedAtUtc,
            $nextRunUtc,
            $errorCode,
            $errorMessage
        );

        $this->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' => $this->encodeResult($resultPayload),
        ]);

        $run = ['status' => $status, 'error_code' => $errorCode, 'error_message' => $errorMessage];
        $this->recordSchedulerJobEvent($jobId, (string) ($job['job_key'] ?? ''), $triggerType, $actorUserId, $run, $durationMs);

        if ($status === ScheduledJobRunStatus::Failed->value) {
            $this->dispatchJobFailedEvent($job, $errorCode, $errorMessage, $triggerType);
        }

        return $run;
    }

    // Fail-open: a logging failure must not kill the job run that already completed.
    private function insertRunLog(array $data): void
    {
        try {
            $this->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'] ?? ScheduledJobTriggerType::Scheduler->value),
                'status' => (string) ($data['status'] ?? ScheduledJobRunStatus::Failed->value),
                '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 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 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 function nullableString(mixed $value): ?string
    {
        $value = trim((string) $value);
        return $value !== '' ? $value : null;
    }

    private function acquireRunnerLock(): bool
    {
        return $this->databaseSessionRepository->acquireAdvisoryLock(self::RUNNER_LOCK_NAME, 0);
    }

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

    private function releaseRunnerLock(): void
    {
        try {
            $this->databaseSessionRepository->releaseAdvisoryLock(self::RUNNER_LOCK_NAME);
        } catch (\Throwable $exception) {
            // no-op
        }
    }

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

    private function recordSchedulerRunEvent(string $triggerType, array $result, ?int $actorUserId = null, array $counts = []): void
    {
        $status = ScheduledJobRunStatus::tryNormalize((string) ($result['status'] ?? ''))?->value;
        if ($status === null) {
            if (!($result['ok'] ?? false)) {
                $status = ScheduledJobRunStatus::Failed->value;
            } elseif ((int) ($counts['failed'] ?? $result['failed'] ?? 0) > 0) {
                $status = ScheduledJobRunStatus::Failed->value;
            } elseif ((int) ($counts['success'] ?? $result['success'] ?? 0) > 0) {
                $status = ScheduledJobRunStatus::Success->value;
            } else {
                $status = ScheduledJobRunStatus::Skipped->value;
            }
        }

        $this->recordAudit(
            'scheduler.run',
            $this->auditOutcomeForRunStatus($status),
            [
                'actor_user_id' => ($actorUserId ?? 0) > 0 ? (int) $actorUserId : null,
                'error_code' => $this->nullableString($result['error'] ?? null),
                'metadata' => [
                    'trigger_type' => ScheduledJobTriggerType::normalizeOr($triggerType, ScheduledJobTriggerType::Scheduler)->value,
                    'run_status' => $status,
                    'processed' => (int) ($counts['processed'] ?? $result['processed'] ?? 0),
                    'success_count' => (int) ($counts['success'] ?? $result['success'] ?? 0),
                    'failed_count' => (int) ($counts['failed'] ?? $result['failed'] ?? 0),
                    'skipped_count' => (int) ($counts['skipped'] ?? $result['skipped'] ?? 0),
                    'duration_ms' => (int) ($result['duration_ms'] ?? 0),
                ],
            ]
        );
    }

    private function recordSchedulerJobEvent(
        int $jobId,
        string $jobKey,
        string $triggerType,
        ?int $actorUserId,
        array $runResult,
        int $durationMs
    ): void {
        $status = ScheduledJobRunStatus::normalizeOr((string) ($runResult['status'] ?? ''), ScheduledJobRunStatus::Failed)->value;

        $this->recordAudit(
            'scheduler.job.run',
            $this->auditOutcomeForRunStatus($status),
            [
                'actor_user_id' => ($actorUserId ?? 0) > 0 ? (int) $actorUserId : null,
                'target_type' => 'scheduled_job',
                'target_id' => $jobId > 0 ? $jobId : null,
                'error_code' => $this->nullableString($runResult['error_code'] ?? null),
                'metadata' => [
                    'job_id' => $jobId,
                    'job_key' => $jobKey,
                    'trigger_type' => ScheduledJobTriggerType::normalizeOr($triggerType, ScheduledJobTriggerType::Scheduler)->value,
                    'run_status' => $status,
                    'duration_ms' => max(0, $durationMs),
                ],
            ]
        );
    }

    private function dispatchJobFailedEvent(array $job, ?string $errorCode, ?string $errorMessage, string $triggerType): void
    {
        if (!$this->eventDispatcher) {
            return;
        }

        try {
            $this->eventDispatcher->dispatch(ModuleEvents::SCHEDULER_JOB_FAILED, [
                'job_id' => (int) ($job['id'] ?? 0),
                'job_key' => (string) ($job['job_key'] ?? ''),
                'label' => (string) ($job['label'] ?? ''),
                'error_code' => $errorCode,
                'error_message' => $errorMessage,
                'trigger_type' => $triggerType,
            ]);
        } catch (\Throwable) {
            // fail-open: notification failure must not affect scheduler
        }
    }

    private function recordAudit(string $eventType, string $outcome, array $context): void
    {
        try {
            $this->systemAuditService->record($eventType, $outcome, $context);
        } catch (\Throwable) {
            // fail-open
        }
    }

    private function auditOutcomeForRunStatus(string $status): string
    {
        if ($status === ScheduledJobRunStatus::Success->value) {
            return 'success';
        }
        if ($status === ScheduledJobRunStatus::Failed->value) {
            return 'failed';
        }
        // "skipped" is an operational non-error state (e.g. already running),
        // not an authorization denial.
        return 'success';
    }
}