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; } 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; } }