breadcrumb-the-shire/core/Repository/Scheduler/ScheduledJobRepository.php

<?php

namespace MintyPHP\Repository\Scheduler;

use MintyPHP\DB;
use MintyPHP\Domain\Taxonomy\ScheduledJobStatus;
use MintyPHP\Repository\Support\RepoQuery;

/** Persists scheduled jobs with cron expressions, next-run calculation, and metadata updates. */
class ScheduledJobRepository implements ScheduledJobRepositoryInterface
{
    public function create(array $data): int|false
    {
        $id = DB::insert(
            'insert into scheduled_jobs (
                job_key, label, description, enabled, timezone,
                schedule_type, schedule_interval, schedule_time, schedule_weekdays_csv,
                catchup_once, next_run_at
            ) values (?,?,?,?,?,?,?,?,?,?,?)',
            (string) ($data['job_key'] ?? ''),
            (string) ($data['label'] ?? ''),
            $data['description'] ?? null,
            (string) ((int) ($data['enabled'] ?? 0)),
            (string) ($data['timezone'] ?? ''),
            (string) ($data['schedule_type'] ?? 'daily'),
            (string) ((int) ($data['schedule_interval'] ?? 1)),
            $data['schedule_time'] ?? null,
            $data['schedule_weekdays_csv'] ?? null,
            (string) ((int) ($data['catchup_once'] ?? 1)),
            $data['next_run_at'] ?? null
        );
        return $id ? (int) $id : false;
    }

    public function find(int $id): ?array
    {
        if ($id <= 0) {
            return null;
        }
        $row = DB::selectOne('select * from scheduled_jobs where id = ? limit 1', (string) $id);
        return $this->normalizeRow($row);
    }

    public function findByKey(string $jobKey): ?array
    {
        $jobKey = trim($jobKey);
        if ($jobKey === '') {
            return null;
        }
        $row = DB::selectOne('select * from scheduled_jobs where job_key = ? limit 1', $jobKey);
        return $this->normalizeRow($row);
    }

    public function listPaged(array $filters): array
    {
        $search = trim((string) ($filters['search'] ?? ''));
        $enabled = trim((string) ($filters['enabled'] ?? ''));
        $status = strtolower(trim((string) ($filters['status'] ?? '')));

        [$limit, $offset] = RepoQuery::sanitizeLimitOffset($filters, 20, 1, 200, 0);
        [$order, $dir] = RepoQuery::sanitizeOrder(
            $filters,
            ['id', 'job_key', 'label', 'enabled', 'next_run_at', 'last_run_finished_at', 'last_run_status', 'updated_at'],
            'job_key',
            'asc'
        );

        $where = [];
        $params = [];
        RepoQuery::addLikeFilter(
            $where,
            $params,
            ['job_key', 'label', 'description', 'last_error_code', 'last_error_message'],
            $search
        );
        if (in_array($enabled, ['0', '1'], true)) {
            $where[] = 'enabled = ?';
            $params[] = $enabled;
        }
        if (ScheduledJobStatus::tryNormalize($status) !== null) {
            $where[] = 'last_run_status = ?';
            $params[] = $status;
        }

        $whereSql = $where ? (' where ' . implode(' and ', $where)) : '';
        $total = (int) (DB::selectValue('select count(*) from scheduled_jobs' . $whereSql, ...$params) ?? 0);
        $rows = DB::select(
            'select * from scheduled_jobs' .
            $whereSql .
            sprintf(' order by `%s` %s limit ? offset ?', $order, $dir),
            ...array_merge($params, [(string) $limit, (string) $offset])
        );

        $normalized = [];
        if (is_array($rows)) {
            foreach ($rows as $row) {
                $item = $this->normalizeRow($row);
                if ($item !== null) {
                    $normalized[] = $item;
                }
            }
        }

        return [
            'total' => $total,
            'rows' => $normalized,
        ];
    }

    public function listDueJobs(string $nowUtc, int $limit = 20): array
    {
        $limit = max(1, min(200, $limit));
        $rows = DB::select(
            'select * from scheduled_jobs
             where enabled = 1
               and next_run_at is not null
               and next_run_at <= ?
             order by next_run_at asc
             limit ?',
            $nowUtc,
            (string) $limit
        );

        $normalized = [];
        if (is_array($rows)) {
            foreach ($rows as $row) {
                $item = $this->normalizeRow($row);
                if ($item !== null) {
                    $normalized[] = $item;
                }
            }
        }
        return $normalized;
    }

    public function updateJobMeta(int $id, array $data): bool
    {
        if ($id <= 0) {
            return false;
        }

        $updated = DB::update(
            'update scheduled_jobs
             set label = ?,
                 description = ?,
                 enabled = ?,
                 timezone = ?,
                 schedule_type = ?,
                 schedule_interval = ?,
                 schedule_time = ?,
                 schedule_weekdays_csv = ?,
                 catchup_once = ?,
                 next_run_at = ?
             where id = ?',
            (string) ($data['label'] ?? ''),
            $data['description'] ?? null,
            (string) ((int) ($data['enabled'] ?? 0)),
            (string) ($data['timezone'] ?? ''),
            (string) ($data['schedule_type'] ?? 'daily'),
            (string) ((int) ($data['schedule_interval'] ?? 1)),
            $data['schedule_time'] ?? null,
            $data['schedule_weekdays_csv'] ?? null,
            (string) ((int) ($data['catchup_once'] ?? 1)),
            $data['next_run_at'] ?? null,
            (string) $id
        );
        return $updated !== false;
    }

    /**
     * Atomically marks a job as running if it is not already in a running state.
     *
     * Includes stale-running protection: if a job has been stuck in 'running' status
     * for more than 2 hours, it is considered abandoned (e.g. the runner process died)
     * and a new run is allowed to proceed. This prevents a crashed runner from blocking
     * the job permanently.
     *
     * Returns true only when the UPDATE affected a row, i.e. the job was successfully
     * claimed by this runner. Returns false if the job is already running (not stale).
     */
    public function markRunning(int $id, string $startedAtUtc): bool
    {
        if ($id <= 0) {
            return false;
        }
        // A job that has been 'running' for longer than this threshold is treated as stale.
        $staleBefore = (new \DateTimeImmutable($startedAtUtc, new \DateTimeZone('UTC')))
            ->modify('-2 hours')
            ->format('Y-m-d H:i:s');

        $updated = DB::update(
            'update scheduled_jobs
             set last_run_status = ?,
                 last_run_started_at = ?,
                 last_run_finished_at = null,
                 last_error_code = null,
                 last_error_message = null
             where id = ?
               and (last_run_status is null
                    or last_run_status <> ?
                    or (last_run_started_at is not null and last_run_started_at < ?))',
            ScheduledJobStatus::Running->value,
            $startedAtUtc,
            (string) $id,
            ScheduledJobStatus::Running->value,
            $staleBefore
        );
        return (int) $updated > 0;
    }

    public function finishRun(
        int $id,
        string $status,
        string $startedAtUtc,
        string $finishedAtUtc,
        ?string $nextRunAtUtc,
        ?string $errorCode,
        ?string $errorMessage
    ): bool {
        if ($id <= 0) {
            return false;
        }
        $updated = DB::update(
            'update scheduled_jobs
             set last_run_status = ?,
                 last_run_started_at = ?,
                 last_run_finished_at = ?,
                 next_run_at = ?,
                 last_error_code = ?,
                 last_error_message = ?
             where id = ?',
            $status,
            $startedAtUtc,
            $finishedAtUtc,
            $nextRunAtUtc,
            $errorCode,
            $errorMessage,
            (string) $id
        );
        return $updated !== false;
    }

    public function deleteOrphanedJobs(array $validKeys): int
    {
        $validKeys = array_filter(array_map('trim', $validKeys), static fn (string $k): bool => $k !== '');
        if ($validKeys === []) {
            return 0;
        }

        $placeholders = implode(',', array_fill(0, count($validKeys), '?'));
        $affected = DB::update(
            "DELETE FROM scheduled_jobs WHERE job_key NOT IN ({$placeholders})",
            ...array_map('strval', $validKeys)
        );

        return (int) $affected;
    }

    private function normalizeRow(mixed $row): ?array
    {
        if (!is_array($row)) {
            return null;
        }
        $item = $row['scheduled_jobs'] ?? $row;
        if (!is_array($item) || !isset($item['id'])) {
            return null;
        }
        return $item;
    }
}