🗃️ Add priority to queue

app/Queue/DatabaseConnector.php

@@ -0,0 +1,19 @@
+<?php
+
+namespace App\Queue;
+
+use Illuminate\Queue\Connectors\DatabaseConnector as BaseConnector;
+
+class DatabaseConnector extends BaseConnector
+{
+    public function connect(array $config)
+    {
+        return new DatabaseQueue(
+            $this->connections->connection($config['connection'] ?? null),
+            $config['table'],
+            $config['queue'],
+            $config['retry_after'] ?? 60,
+            $config['after_commit'] ?? null
+        );
+    }
+}

app/Queue/DatabaseQueue.php

@@ -0,0 +1,132 @@
+<?php
+
+namespace App\Queue;
+
+use Illuminate\Queue\DatabaseQueue as BaseQ;
+use Illuminate\Queue\Jobs\DatabaseJobRecord;
+use App\Jobs\Concerns\HasPriority;
+
+class DatabaseQueue extends BaseQ
+{
+    protected function getPriority($job): ?int
+    {
+        if ($job instanceof DatabaseRecord) {
+            return $job->priority ?? null;
+        }
+
+        if (!($job instanceof HasPriority)) {
+            return null;
+        }
+
+        return $job->getPriority();
+    }
+
+    public function push($job, $data = '', $queue = null)
+    {
+        $this->later(null, $job, $data, $queue);
+    }
+
+    public function pushRaw($payload, $queue = null, array $options = [])
+    {
+        $prio = $options['priority'] ?? null;
+
+        return $this->pushToDatabase($queue, $payload, priority: $prio);
+    }
+
+    public function release($queue, $job, $delay)
+    {
+        $prio = $this->getPriority($job);
+
+        return $this->pushToDatabase(
+            $queue,
+            $job->payload,
+            $delay,
+            $job->attempts,
+            $prio
+        );
+    }
+
+    public function later($delay, $job, $data = '', $queue = null)
+    {
+        $prio = $this->getPriority($job);
+
+        return $this->enqueueUsing(
+            $job,
+            $this->createPayload($job, $this->getQueue($queue), $data),
+            $queue,
+            $delay,
+            function ($payload, $queue, $delay) use ($prio) {
+                return $this->pushToDatabase($queue, $payload, $delay, priority: $prio);
+            }
+        );
+    }
+
+    public function bulk($jobs, $data = '', $queue = null)
+    {
+        $queue = $this->getQueue($queue);
+
+        $availableAt = $this->availableAt();
+
+        return $this->database->table($this->table)->insert(collect((array) $jobs)->map(
+            function ($job) use ($queue, $data, $availableAt) {
+                $prio = $this->getPriority($job);
+
+                return $this->buildDatabaseRecord(
+                    $queue,
+                    $this->createPayload($job, $this->getQueue($queue), $data),
+                    $availableAt,
+                    priority: $prio
+                );
+            }
+        )->all());
+    }
+
+    protected function getNextAvailableJob($queue)
+    {
+        $job = $this
+             ->database
+             ->table($this->table)
+             ->lock($this->getLockForPopping())
+             ->where('queue', $this->getQueue($queue))
+             ->where(function ($query) {
+                 $this->isAvailable($query);
+                 $this->isReservedButExpired($query);
+             })
+             ->orderBy('priority', 'desc')
+             ->orderBy('id', 'asc')
+             ->first();
+
+        return $job ? new DatabaseJobRecord((object) $job) : null;
+    }
+
+    protected function pushToDatabase(
+        $queue,
+        $payload,
+        $delay = 0,
+        $attempts = 0,
+        int $priority = null
+    ) {
+        return $this->database->table($this->table)->insertGetId($this->buildDatabaseRecord(
+            $this->getQueue($queue),
+            $payload,
+            $this->availableAt($delay ?? 0),
+            $attempts,
+            $priority
+        ));
+    }
+
+    protected function buildDatabaseRecord(
+        $queue,
+        $payload,
+        $availableAt,
+        $attempts = 0,
+        int $priority = null
+    ) {
+        $job = parent::buildDatabaseRecord($queue, $payload, $availableAt, $attempts);
+        if (!is_null($priority)) {
+            $job['priority'] = $priority;
+        }
+
+        return $job;
+    }
+}