uuid(); $now = date('Y-m-d H:i:s'); $data['created_at'] = $now; $data['expires_at'] ??= gmdate('Y-m-d H:i:s', time() + (int) config('Site')->retentionHours * 3600); $this->insert($data); return $data['id']; } /** * Atomically claim the next queued job. Uses SKIP LOCKED so N workers * never grab the same row and the API stays responsive. */ public function claimNext(): ?array { $db = $this->db; $sql = "SELECT id FROM {$this->table} WHERE status = 'queued' AND created_at <= ? ORDER BY priority ASC, created_at ASC LIMIT 1"; // One transaction per claim; InnoDB row locks do the rest. $db->transBegin(); try { $query = $db->query($sql . ' FOR UPDATE SKIP LOCKED', [date('Y-m-d H:i:s')]); $row = $query->getFirstRow(); if ($row === null) { $db->transRollback(); return null; } $db->table($this->table) ->where('id', $row->id) ->update([ 'status' => 'processing', 'stage' => 'preparing', 'progress' => 1, 'started_at' => date('Y-m-d H:i:s'), ]); $this->db->query("UPDATE {$this->table} SET attempts = attempts + 1 WHERE id = ?", [$row->id]); $db->transCommit(); return $this->find((string) $row->id); } catch (\Throwable $e) { $db->transRollback(); throw $e; } } /** @return list */ public function forSession(string $sessionHash, int $limit = 20): array { return $this->where('session_id', $sessionHash) ->orderBy('created_at', 'DESC') ->limit($limit) ->findAll(); } public function countQueued(): int { return $this->where('status', 'queued')->countAllResults(); } public function countProcessing(): int { return $this->where('status', 'processing')->countAllResults(); } /** Jobs stuck in processing longer than the max runtime — failed by reaper. */ public function findStale(int $maxSeconds): array { return $this->where('status', 'processing') ->where('started_at <', gmdate('Y-m-d H:i:s', time() - $maxSeconds - 60)) ->findAll(); } /** Rows past retention (any terminal/abandoned state) for the reaper. */ public function findExpired(int $batchSize = 200): array { $now = gmdate('Y-m-d H:i:s'); return $this->where('expires_at <', $now) ->whereIn('status', ['queued', 'processing', 'completed', 'cancelled', 'expired']) ->limit($batchSize) ->findAll(); } /** Failed jobs older than an hour, cleaned on the same cadence. */ public function findOldFailed(int $olderThanSeconds = 3600, int $batchSize = 100): array { return $this->where('status', 'failed') ->where('completed_at <', gmdate('Y-m-d H:i:s', time() - $olderThanSeconds)) ->limit($batchSize) ->findAll(); } private function uuid(): string { $bytes = random_bytes(16); $bytes[6] = chr((ord($bytes[6]) & 0x0f) | 0x40); $bytes[8] = chr((ord($bytes[8]) & 0x3f) | 0x80); $hex = bin2hex($bytes); return sprintf('%s-%s-%s-%s-%s', substr($hex, 0, 8), substr($hex, 8, 4), substr($hex, 12, 4), substr($hex, 16, 4), substr($hex, 20, 12)); } }