Code Coverage
 
Lines
Functions and Methods
Classes and Traits
Total
99.10%
329 / 332
94.87%
37 / 39
 CRAP
0.00%
0 / 1
Database
99.10%
329 / 332
94.87%
37 / 39
 73
0.00%
0 / 1
 __construct
100.00%
8 / 8
100.00%
1 / 1
 2
 create
100.00%
1 / 1
100.00%
1 / 1
 1
 getDb
100.00%
1 / 1
100.00%
1 / 1
 1
 db
100.00%
1 / 1
100.00%
1 / 1
 1
 getTable
100.00%
1 / 1
100.00%
1 / 1
 1
 ensureReservedUntilColumn
100.00%
10 / 10
100.00%
1 / 1
 2
 ensureReservedByColumn
100.00%
6 / 6
100.00%
1 / 1
 2
 claimedBy
100.00%
5 / 5
100.00%
1 / 1
 1
 buildTaskClaimEligibleWhere
100.00%
10 / 10
100.00%
1 / 1
 1
 claimedByTaskId
100.00%
7 / 7
100.00%
1 / 1
 1
 claimTaskRun
100.00%
15 / 15
100.00%
1 / 1
 1
 getEndIndex
100.00%
7 / 7
100.00%
1 / 1
 2
 push
100.00%
18 / 18
100.00%
1 / 1
 1
 buildEligibleWhere
100.00%
9 / 9
100.00%
1 / 1
 1
 reserve
100.00%
43 / 43
100.00%
1 / 1
 11
 release
100.00%
18 / 18
100.00%
1 / 1
 1
 delete
100.00%
6 / 6
100.00%
1 / 1
 1
 bury
100.00%
15 / 15
100.00%
1 / 1
 1
 hasJobs
100.00%
1 / 1
100.00%
1 / 1
 1
 count
100.00%
5 / 5
100.00%
1 / 1
 2
 clear
100.00%
4 / 4
100.00%
1 / 1
 1
 hasDeadJobs
100.00%
1 / 1
100.00%
1 / 1
 1
 countDead
100.00%
5 / 5
100.00%
1 / 1
 2
 getDeadJobs
83.33%
10 / 12
0.00%
0 / 1
 4.07
 getDeadJob
91.67%
11 / 12
0.00%
0 / 1
 4.01
 retryDeadJob
100.00%
5 / 5
100.00%
1 / 1
 2
 deleteDeadJob
100.00%
6 / 6
100.00%
1 / 1
 1
 clearDead
100.00%
4 / 4
100.00%
1 / 1
 1
 schedule
100.00%
16 / 16
100.00%
1 / 1
 2
 getTasks
100.00%
8 / 8
100.00%
1 / 1
 2
 getTask
100.00%
11 / 11
100.00%
1 / 1
 4
 getAllTasks
100.00%
10 / 10
100.00%
1 / 1
 4
 updateTask
100.00%
14 / 14
100.00%
1 / 1
 2
 removeTask
100.00%
6 / 6
100.00%
1 / 1
 1
 getTaskCount
100.00%
5 / 5
100.00%
1 / 1
 2
 hasTasks
100.00%
1 / 1
100.00%
1 / 1
 1
 clearTasks
100.00%
4 / 4
100.00%
1 / 1
 1
 createTable
100.00%
14 / 14
100.00%
1 / 1
 1
 createIndexes
100.00%
7 / 7
100.00%
1 / 1
 2
1<?php
2declare(strict_types=1);
3/**
4 * Pop PHP Framework (https://www.popphp.org/)
5 *
6 * @link       https://github.com/popphp/popphp-framework
7 * @author     Nick Sagona, III <nick@popphp.org>
8 * @copyright  Copyright (c) 2009-2026 Nick Sagona, III
9 * @license    https://www.popphp.org/license     New BSD License
10 */
11
12/**
13 * @namespace
14 */
15namespace Pop\Queue\Adapter;
16
17use Pop\Db\Adapter\AbstractAdapter as DbAdapter;
18use Pop\Db\Gateway\Table as DbTable;
19use Pop\Db\Sql\AbstractSql as DbSql;
20use Pop\Db\Sql\Where;
21use Pop\Queue\Process\AbstractJob;
22use Pop\Queue\Process\PayloadSigner;
23use Pop\Queue\Process\Task;
24
25/**
26 * Database adapter class
27 *
28 * @category   Pop
29 * @package    Pop\Queue
30 * @author     Nick Sagona, III <nick@popphp.org>
31 * @copyright  Copyright (c) 2009-2026 Nick Sagona, III
32 * @license    https://www.popphp.org/license     New BSD License
33 * @version    3.0.0
34 */
35class Database extends AbstractTaskAdapter
36{
37    /**
38     * How many candidate rows reserve() pulls back per scan.
39     *
40     * reserve() returns a single job, but it has to look past any candidate it
41     * can't claim (delayed, backed off, corrupt, or won by another worker), so
42     * it needs more than one row in hand - and, when a whole batch turns out to
43     * be unclaimable, the freedom to page further back. Without a bound it read
44     * the entire eligible set into PHP to hand back one job, which made the cost
45     * of reserving grow with the depth of the queue: every worker paid for every
46     * queued job, on every job it ran.
47     *
48     * Sized to clear the ordinary case in one round trip - a run of delayed jobs
49     * at the head of the queue is the usual reason for a skip, and it is rarely
50     * dozens long - while staying small enough that the payloads fetched
51     * alongside them stay cheap.
52     */
53    protected const RESERVE_BATCH_SIZE = 50;
54
55    /**
56     * Database adapter
57     * @var ?DbAdapter
58     */
59    protected ?DbAdapter $db = null;
60
61    /**
62     * Database table
63     * @var ?string
64     */
65    protected ?string $table = null;
66
67    /**
68     * Reservation lease length, in seconds
69     * @var int
70     */
71    protected int $leaseSeconds = 60;
72
73    /**
74     * Constructor
75     *
76     * Instantiate the database adapter object
77     *
78     * @param DbAdapter $db
79     * @param string    $table
80     * @param ?string   $priority
81     * @param int       $leaseSeconds
82     */
83    public function __construct(DbAdapter $db, string $table = 'pop_queue', ?string $priority = null, int $leaseSeconds = 60)
84    {
85        $this->db           = $db;
86        $this->table        = $table;
87        $this->leaseSeconds = $leaseSeconds;
88
89        if (!$this->db->hasTable($table)) {
90            $this->createTable($table);
91        } else {
92            $this->ensureReservedUntilColumn($table);
93            $this->ensureReservedByColumn($table);
94        }
95
96        parent::__construct($priority);
97    }
98
99    /**
100     * Create database adapter
101     *
102     * @param  DbAdapter $db
103     * @param  string    $table
104     * @param  ?string   $priority
105     * @param  int       $leaseSeconds
106     * @return Database
107     */
108    public static function create(DbAdapter $db, string $table = 'pop_queue', ?string $priority = null, int $leaseSeconds = 60): Database
109    {
110        return new self($db, $table, $priority, $leaseSeconds);
111    }
112
113    /**
114     * Get database adapter
115     *
116     * @return ?DbAdapter
117     */
118    public function getDb(): ?DbAdapter
119    {
120        return $this->db;
121    }
122
123    /**
124     * Get database adapter (alias)
125     *
126     * @return ?DbAdapter
127     */
128    public function db(): ?DbAdapter
129    {
130        return $this->db;
131    }
132
133    /**
134     * Get database table
135     *
136     * @return ?string
137     */
138    public function getTable(): ?string
139    {
140        return $this->table;
141    }
142
143    /**
144     * Add the reserved_until column to a table created by an earlier version
145     * of this adapter, if it isn't there already. Gated on a real
146     * column-existence check (Pop\Db\Gateway\Table::getTableInfo(), backed
147     * by PRAGMA table_info on SQLite / information_schema.columns on
148     * Postgres+SQL Server / SHOW COLUMNS elsewhere - genuinely portable
149     * across every backend this adapter supports), NOT on catching the
150     * exception a duplicate-column ALTER is expected to throw: this repo's
151     * SQLite adapter's query() only calls throwError() when the driver
152     * reports a non-zero error code, and a duplicate-column ALTER TABLE ...
153     * ADD COLUMN against SQLite returns false with error code 0 - it fails
154     * silently (surfacing only as a PHP warning), so a try/catch around it
155     * never fires. That previously let the backfill below re-run on every
156     * single construction against an already-migrated table, zeroing out
157     * reserved_until (and the lease it represents) for every currently
158     * in-flight job, table-wide, on ordinary worker startup - exactly the
159     * double-execution failure this task exists to prevent. Gating on the
160     * real column check means the ALTER and backfill run exactly once, ever,
161     * per table, and a genuine migration failure (permissions, locked table)
162     * now throws normally instead of failing silently.
163     *
164     * The backfill sets reserved_until = 0 for any row already sitting at
165     * status = 0 - a job reserved under the pre-lease Phase 1 contract.
166     * Without it, such a row would have reserved_until = NULL forever, and
167     * "NULL <= now" evaluates to NULL in SQL - matching neither the
168     * "status = 1" nor the "status = 0 AND reserved_until <= now" branch of
169     * reserve()'s eligibility check, making the row invisible to reserve()
170     * permanently. Backfilling it to 0 makes it immediately eligible for
171     * reclaim on the very next reserve() call, which is the correct
172     * behavior for a job whose reservation state predates leasing entirely.
173     *
174     * @param  string $table
175     * @return void
176     */
177    protected function ensureReservedUntilColumn(string $table): void
178    {
179        $info = (new DbTable($table))->getTableInfo($this->db);
180        if (isset($info['columns']['reserved_until'])) {
181            return;
182        }
183
184        $schema = $this->db->createSchema();
185        $schema->alter($table)->addColumn('reserved_until', 'int', 16)->nullable();
186        $this->db->query($schema);
187
188        $backfill = $this->db->createSql();
189        $backfill->update($table)->values(['reserved_until' => 0])
190            ->where("type = 'job'")->andWhere('status = 0');
191        $this->db->query($backfill);
192    }
193
194    /**
195     * Add the reserved_by column to a table created by an earlier version of
196     * this adapter, if it isn't there already. Same real column-existence
197     * check as ensureReservedUntilColumn() (see that method's docblock for
198     * why exception-based detection is unsafe here). reserved_by holds the
199     * random claim token reserve() writes and re-reads to prove its own
200     * claiming UPDATE actually won a given row - see reserve()'s docblock
201     * for why that replaced an affected-row count.
202     *
203     * @param  string $table
204     * @return void
205     */
206    protected function ensureReservedByColumn(string $table): void
207    {
208        $info = (new DbTable($table))->getTableInfo($this->db);
209        if (isset($info['columns']['reserved_by'])) {
210            return;
211        }
212
213        $schema = $this->db->createSchema();
214        $schema->alter($table)->addColumn('reserved_by', 'varchar', 64)->nullable();
215        $this->db->query($schema);
216    }
217
218    /**
219     * Read back the reserved_by token currently stored for a row. Used by
220     * reserve() to prove, via the real resulting row state rather than
221     * driver-reported execute metadata, whether its own claiming UPDATE was
222     * the one that actually won the row.
223     *
224     * @param  int $id
225     * @return ?string
226     */
227    protected function claimedBy(int $id): ?string
228    {
229        $sql = $this->db->createSql();
230        $sql->select('reserved_by')->from($this->table)->where('id = ' . $id);
231        $this->db->query($sql);
232        $rows = $this->db->fetchAll();
233
234        return $rows[0]['reserved_by'] ?? null;
235    }
236
237    /**
238     * Build the eligibility predicate for claiming a task's current
239     * due-window: eligible if no claim exists yet, the existing claim is
240     * for a *different* window (a new tick is always claimable
241     * immediately, regardless of the old claim's expiry), or the existing
242     * claim is for the *same* window but has expired. The window is
243     * encoded as a prefix of reserved_by itself ("<window>:<token>"), so
244     * this needs no new column - reserved_until still means "this claim
245     * expires at", reserved_by's prefix now also answers "for which
246     * window". "reserved_by NOT LIKE '<window>:%'" is plain, portable SQL
247     * LIKE - no dialect-specific string functions - and the window value
248     * is purely numeric so it can never contain a LIKE wildcard.
249     *
250     * @param  DbSql  $sql
251     * @param  string $taskId
252     * @param  string $window
253     * @param  int    $now
254     * @return Where
255     */
256    protected function buildTaskClaimEligibleWhere(DbSql $sql, string $taskId, string $window, int $now): Where
257    {
258        $where = new Where($sql);
259        $where->equalTo('type', 'task');
260        $where->equalTo('job_id', $taskId);
261
262        $group = $where->andNest();
263        $group->isNull('reserved_by');
264        $group->or();
265        $group->notLike('reserved_by', $window . ':%');
266        $group->or();
267        $group->lessThanOrEqualTo('reserved_until', $now);
268
269        return $where;
270    }
271
272    /**
273     * Read back the reserved_by token currently stored for a task row.
274     * Used by claimTaskRun() to prove, via the real resulting row state,
275     * whether its own claiming UPDATE actually won - the same approach
276     * claimedBy() uses for job claiming (see reserve()'s docblock for why
277     * an affected-row count isn't portable enough for this).
278     *
279     * @param  string $taskId
280     * @return ?string
281     */
282    protected function claimedByTaskId(string $taskId): ?string
283    {
284        $sql = $this->db->createSql();
285        $sql->select('reserved_by')->from($this->table)->where("type = 'task'")->andWhere('job_id = :job_id');
286        $this->db->prepare($sql);
287        $this->db->bindParams(['job_id' => $taskId]);
288        $this->db->execute();
289        $rows = $this->db->fetchAll();
290
291        return $rows[0]['reserved_by'] ?? null;
292    }
293
294    /**
295     * Atomically claim a task's current due-window. See
296     * buildTaskClaimEligibleWhere() for the eligibility rule.
297     *
298     * @param  string $taskId
299     * @param  string $window
300     * @return bool
301     */
302    public function claimTaskRun(string $taskId, string $window): bool
303    {
304        $now   = time();
305        $token = $window . ':' . bin2hex(random_bytes(8));
306
307        $sql    = $this->db->createSql();
308        $update = $sql->update($this->table)->values([
309            'reserved_until' => ':reserved_until',
310            'reserved_by'    => ':reserved_by'
311        ]);
312        $update->where($this->buildTaskClaimEligibleWhere($update, $taskId, $window, $now));
313
314        $this->db->prepare($sql);
315        $this->db->bindParams([
316            'reserved_until' => ($now + self::TASK_CLAIM_TTL),
317            'reserved_by'    => $token
318        ]);
319        $this->db->execute();
320
321        return ($this->claimedByTaskId($taskId) === $token);
322    }
323
324    /**
325     * Get queue end index
326     *
327     * @return int
328     */
329    protected function getEndIndex(): int
330    {
331        $sql    = $this->db->createSql();
332        $select = $sql->select('index')->from($this->table);
333
334        // Kept off the where() chain on purpose - where() is declared to return
335        // AbstractPredicateClause, which has no orderBy()/limit(), so chaining
336        // past it loses the Select type even though the object really is one.
337        $select->where('index IS NOT NULL');
338        $select->orderBy('index', 'DESC')->limit(1);
339
340        $this->db->query($sql);
341
342        $rows = $this->db->fetchAll();
343        return (isset($rows[0]['index'])) ? (int)$rows[0]['index'] : 0;
344    }
345
346    /**
347     * Push job on to queue
348     *
349     * @param  AbstractJob $job
350     * @return Database
351     */
352    public function push(AbstractJob $job): Database
353    {
354        $sql = $this->db->createSql();
355        $sql->insert($this->table)->values([
356            'index'   => ':index',
357            'type'    => ':type',
358            'job_id'  => ':job_id',
359            'payload' => ':payload',
360            'status'  => ':status'
361        ]);
362
363        $this->db->prepare($sql);
364        $this->db->bindParams([
365            'index'   => ($this->getEndIndex() + 1),
366            'type'    => 'job',
367            'job_id'  => $job->getJobId(),
368            'payload' => base64_encode(PayloadSigner::sign(serialize(clone $job))),
369            'status'  => 1
370        ]);
371        $this->db->execute();
372
373        return $this;
374    }
375
376    /**
377     * Build the "pending, or claimed with an expired lease" predicate shared
378     * by reserve()'s scan and its per-row claiming UPDATE: type = 'job' AND
379     * (status = 1 OR (status = 0 AND reserved_until <= $now)). Built via
380     * pop-db's nested PredicateSet API (Where::andNest()/orNest()) rather
381     * than a raw SQL string, because pop-db's where()/andWhere() string
382     * parser only understands simple "column operator value" expressions -
383     * handing it a compound, parenthesized boolean expression silently
384     * misparses it instead of raising an error.
385     *
386     * @param  DbSql $sql
387     * @param  int   $now
388     * @return Where
389     */
390    protected function buildEligibleWhere(DbSql $sql, int $now): Where
391    {
392        $where = new Where($sql);
393        $where->equalTo('type', 'job');
394
395        $leaseGroup = $where->andNest();
396        $leaseGroup->equalTo('status', 1);
397
398        $expiredGroup = $leaseGroup->orNest();
399        $expiredGroup->equalTo('status', 0);
400        $expiredGroup->and();
401        $expiredGroup->lessThanOrEqualTo('reserved_until', $now);
402
403        return $where;
404    }
405
406    /**
407     * Atomically claim the next eligible job. Scans pending/expired-lease
408     * rows in queue order, skipping any job that isn't yet available (delayed
409     * or backed off), and atomically claims the first eligible one via a
410     * conditional UPDATE that writes a random claim token into reserved_by
411     * alongside status/reserved_until - all in the same statement.
412     *
413     * The scan runs in batches of RESERVE_BATCH_SIZE rather than over the whole
414     * eligible set at once, paging forward only when an entire batch turns out
415     * to be unclaimable. Paging is inherently approximate - rows claimed by
416     * other workers between batches shift the offset, so a candidate can slip
417     * past - but the scan was already racy by construction (that is what the
418     * read-back below exists to handle), and anything missed is simply picked up
419     * by the next reserve() call rather than lost.
420     *
421     * Success is proven by re-reading reserved_by immediately after the
422     * UPDATE and comparing it to the token this call generated, not by
423     * inspecting the UPDATE's driver-reported affected-row count.
424     * getNumberOfAffectedRows() is not portable enough for this: pop-db's
425     * Pgsql adapter reads it off the *prepare* result rather than the
426     * *execute* result, so pg_affected_rows() on a PREPARE always reports 0
427     * - meaning a real, successful claim on Postgres would still read back
428     * as "0 affected rows" and reserve() would treat every genuine win as a
429     * lost race, lease every row it touches, and never return a job. Reading
430     * back the actual resulting row state instead works identically across
431     * every backend, because it isn't asking the driver to describe what it
432     * did - it's asking the database what's really there now.
433     *
434     * If reserved_by doesn't come back as this call's token (someone else's
435     * token is there, or the row's state changed/vanished), this worker lost
436     * the race between the scan and its UPDATE, so this moves on to the next
437     * candidate instead of assuming success.
438     *
439     * @return ?AbstractJob
440     */
441    public function reserve(): ?AbstractJob
442    {
443        $now    = time();
444        $offset = 0;
445
446        while (true) {
447            $sql    = $this->db->createSql();
448            $select = $sql->select(['id', 'index', 'payload'])->from($this->table);
449            $select->where($this->buildEligibleWhere($select, $now));
450            $select->orderBy('index', ($this->isFifo()) ? 'ASC' : 'DESC');
451            $select->limit(self::RESERVE_BATCH_SIZE);
452            if ($offset > 0) {
453                $select->offset($offset);
454            }
455
456            $this->db->query($sql);
457            $rows = $this->db->fetchAll();
458
459            if (empty($rows)) {
460                return null;
461            }
462
463            foreach ($rows as $row) {
464                // Suppressed: a corrupt/truncated payload makes unserialize() emit a
465                // warning and return false, and a payload whose class no longer
466                // exists (renamed/removed in a deploy) yields a
467                // __PHP_Incomplete_Class - the instanceof check below handles both.
468                // A payload that fails PayloadSigner::verify() (tampered, or written
469                // by something other than this application when a signing key is
470                // configured) is treated identically - $raw is false, so
471                // unserialize() is never called on it at all.
472                $raw = PayloadSigner::verify(base64_decode($row['payload']));
473                $job = ($raw !== false) ? @unserialize($raw) : false;
474
475                if (!($job instanceof AbstractJob)) {
476                    // Corrupt/unloadable payload - skip rather than claim it and
477                    // then blow up returning a non-AbstractJob. Claiming it would
478                    // be worse than a one-shot crash now that leases exist: the
479                    // claim would expire, get reclaimed back to eligible, and
480                    // poison the next worker too, forever.
481                    continue;
482                }
483
484                if (!$job->isAvailable()) {
485                    continue;
486                }
487
488                $id    = (int)$row['id'];
489                $token = bin2hex(random_bytes(16));
490
491                $sql    = $this->db->createSql();
492                $update = $sql->update($this->table)->values([
493                    'status'         => ':status',
494                    'reserved_until' => ':reserved_until',
495                    'reserved_by'    => ':reserved_by'
496                ]);
497                $update->where($this->buildEligibleWhere($update, $now));
498                $update->andWhere('id = ' . $id);
499
500                $this->db->prepare($sql);
501                $this->db->bindParams([
502                    'status'         => 0,
503                    'reserved_until' => ($now + $this->leaseSeconds),
504                    'reserved_by'    => $token
505                ]);
506                $this->db->execute();
507
508                if ($this->claimedBy($id) === $token) {
509                    return $job;
510                }
511                // Lost the race to another worker between the scan and this UPDATE - try the next candidate.
512            }
513
514            // Nothing in this batch was claimable - every row was corrupt, not
515            // yet available, or lost to another worker. A short batch means
516            // there is nothing further back to look at; a full one means there
517            // might be, so page forward rather than give up. A queue whose head
518            // is a run of delayed jobs still has runnable ones behind them.
519            if (count($rows) < self::RESERVE_BATCH_SIZE) {
520                return null;
521            }
522
523            $offset += self::RESERVE_BATCH_SIZE;
524        }
525    }
526
527    /**
528     * Put a job back to pending, honoring its backoff schedule unless an
529     * explicit delay is given
530     *
531     * @param  AbstractJob $job
532     * @param  ?int        $delay
533     * @return Database
534     */
535    public function release(AbstractJob $job, ?int $delay = null): Database
536    {
537        $job->delay($delay ?? $job->getBackoffDelay());
538
539        $sql = $this->db->createSql();
540        $sql->update($this->table)->values([
541            'payload'        => ':payload',
542            'status'         => ':status',
543            'reserved_until' => ':reserved_until',
544            'reserved_by'    => ':reserved_by'
545        ])->where("type = 'job'")->andWhere('job_id = :job_id');
546
547        $this->db->prepare($sql);
548        $this->db->bindParams([
549            'payload'        => base64_encode(PayloadSigner::sign(serialize(clone $job))),
550            'status'         => 1,
551            'reserved_until' => null,
552            'reserved_by'    => null,
553            'job_id'         => $job->getJobId()
554        ]);
555        $this->db->execute();
556
557        return $this;
558    }
559
560    /**
561     * Permanently remove a job
562     *
563     * @param  AbstractJob $job
564     * @return Database
565     */
566    public function delete(AbstractJob $job): Database
567    {
568        $sql = $this->db->createSql();
569        $sql->delete()->from($this->table)->where("type = 'job'")->andWhere('job_id = :job_id');
570        $this->db->prepare($sql);
571        $this->db->bindParams(['job_id' => $job->getJobId()]);
572        $this->db->execute();
573
574        return $this;
575    }
576
577    /**
578     * Move a job to the dead-letter store
579     *
580     * @param  AbstractJob $job
581     * @param  ?string     $reason
582     * @return Database
583     */
584    public function bury(AbstractJob $job, ?string $reason = null): Database
585    {
586        $this->delete($job);
587
588        $sql = $this->db->createSql();
589        $sql->insert($this->table)->values([
590            'type'    => ':type',
591            'job_id'  => ':job_id',
592            'payload' => ':payload'
593        ]);
594
595        $this->db->prepare($sql);
596        $this->db->bindParams([
597            'type'    => 'dead',
598            'job_id'  => $job->getJobId(),
599            'payload' => base64_encode(PayloadSigner::sign(serialize(clone $job)))
600        ]);
601        $this->db->execute();
602
603        return $this;
604    }
605
606    /**
607     * Check if adapter has jobs
608     *
609     * @return bool
610     */
611    public function hasJobs(): bool
612    {
613        return ($this->count() > 0);
614    }
615
616    /**
617     * Count of pending + reserved jobs
618     *
619     * @return int
620     */
621    public function count(): int
622    {
623        $sql = $this->db->createSql();
624        $sql->select(['total' => 'COUNT(1)'])->from($this->table)->where("type = 'job'");
625        $this->db->query($sql);
626        $rows = $this->db->fetchAll();
627
628        return (isset($rows[0]['total'])) ? (int)$rows[0]['total'] : 0;
629    }
630
631    /**
632     * Clear pending and reserved jobs (not tasks or dead-letter jobs)
633     *
634     * @return Database
635     */
636    public function clear(): Database
637    {
638        $sql = $this->db->createSql();
639        $sql->delete()->from($this->table)->where("type = 'job'");
640        $this->db->query($sql);
641
642        return $this;
643    }
644
645    /**
646     * Check if adapter has dead-letter jobs
647     *
648     * @return bool
649     */
650    public function hasDeadJobs(): bool
651    {
652        return ($this->countDead() > 0);
653    }
654
655    /**
656     * Count of dead-letter jobs
657     *
658     * @return int
659     */
660    public function countDead(): int
661    {
662        $sql = $this->db->createSql();
663        $sql->select(['total' => 'COUNT(1)'])->from($this->table)->where("type = 'dead'");
664        $this->db->query($sql);
665        $rows = $this->db->fetchAll();
666
667        return (isset($rows[0]['total'])) ? (int)$rows[0]['total'] : 0;
668    }
669
670    /**
671     * Get dead-letter jobs
672     *
673     * @param  bool $unserialize
674     * @return array
675     */
676    public function getDeadJobs(bool $unserialize = true): array
677    {
678        $sql = $this->db->createSql();
679        $sql->select()->from($this->table)->where("type = 'dead'");
680        $this->db->query($sql);
681        $rows = $this->db->fetchAll();
682        $jobs = [];
683
684        foreach ($rows as $row) {
685            if (!$unserialize) {
686                $jobs[$row['job_id']] = $row;
687                continue;
688            }
689            $raw = PayloadSigner::verify(base64_decode($row['payload']));
690            $jobs[$row['job_id']] = ($raw !== false) ? unserialize($raw) : false;
691        }
692
693        return $jobs;
694    }
695
696    /**
697     * Get a dead-letter job
698     *
699     * @param  string $jobId
700     * @param  bool   $unserialize
701     * @return mixed
702     */
703    public function getDeadJob(string $jobId, bool $unserialize = true): mixed
704    {
705        $sql = $this->db->createSql();
706        $sql->select()->from($this->table)->where("type = 'dead'")->andWhere('job_id = :job_id');
707        $this->db->prepare($sql);
708        $this->db->bindParams(['job_id' => $jobId]);
709        $this->db->execute();
710        $rows = $this->db->fetchAll();
711
712        if (!isset($rows[0]['payload'])) {
713            return null;
714        }
715
716        if (!$unserialize) {
717            return $rows[0];
718        }
719
720        $raw = PayloadSigner::verify(base64_decode($rows[0]['payload']));
721        return ($raw !== false) ? unserialize($raw) : false;
722    }
723
724    /**
725     * Move a dead-letter job back to pending
726     *
727     * @param  string $jobId
728     * @return Database
729     */
730    public function retryDeadJob(string $jobId): Database
731    {
732        $job = $this->getDeadJob($jobId);
733        if ($job instanceof AbstractJob) {
734            $this->deleteDeadJob($jobId);
735            $this->push($job);
736        }
737
738        return $this;
739    }
740
741    /**
742     * Permanently remove a dead-letter job
743     *
744     * @param  string $jobId
745     * @return Database
746     */
747    public function deleteDeadJob(string $jobId): Database
748    {
749        $sql = $this->db->createSql();
750        $sql->delete()->from($this->table)->where("type = 'dead'")->andWhere('job_id = :job_id');
751        $this->db->prepare($sql);
752        $this->db->bindParams(['job_id' => $jobId]);
753        $this->db->execute();
754
755        return $this;
756    }
757
758    /**
759     * Clear all dead-letter jobs
760     *
761     * @return Database
762     */
763    public function clearDead(): Database
764    {
765        $sql = $this->db->createSql();
766        $sql->delete()->from($this->table)->where("type = 'dead'");
767        $this->db->query($sql);
768
769        return $this;
770    }
771
772    /**
773     * Schedule job with queue
774     *
775     * @param  Task $task
776     * @return Database
777     */
778    public function schedule(Task $task): Database
779    {
780        if ($task->isValid()) {
781            $sql = $this->db->createSql();
782            $sql->insert($this->table)->values([
783                'type'    => ':type',
784                'job_id'  => ':job_id',
785                'payload' => ':payload'
786            ]);
787
788            $jobData = [
789                'type'    => 'task',
790                'job_id'  => $task->getJobId(),
791                'payload' => base64_encode(PayloadSigner::sign(serialize(clone $task)))
792            ];
793
794            $this->db->prepare($sql);
795            $this->db->bindParams($jobData);
796            $this->db->execute();
797        }
798
799        return $this;
800    }
801
802    /**
803     * Get scheduled tasks
804     *
805     * @return array
806     */
807    public function getTasks(): array
808    {
809        $sql = $this->db->createSql();
810        $sql->select('job_id')->from($this->table)->where("type = 'task'");
811        $this->db->query($sql);
812        $rows = $this->db->fetchAll();
813
814        $tasks = [];
815
816        foreach ($rows as $row) {
817            $tasks[] = $row['job_id'];
818        }
819
820        return $tasks;
821    }
822
823    /**
824     * Get scheduled task
825     *
826     * @param  string $taskId
827     * @return ?Task
828     */
829    public function getTask(string $taskId): ?Task
830    {
831        $sql = $this->db->createSql();
832        $sql->select('payload')->from($this->table)->where("type = 'task'")->andWhere('job_id = :job_id');
833        $this->db->prepare($sql);
834        $this->db->bindParams(['job_id' => $taskId]);
835        $this->db->execute();
836        $rows = $this->db->fetchAll();
837
838        if (!isset($rows[0]['payload'])) {
839            return null;
840        }
841
842        // Guarded with instanceof rather than returning unserialize()'s result
843        // directly, the same way reserve() does: a corrupt or tampered payload
844        // makes unserialize() return false, and false out of a ": ?Task" method
845        // is a TypeError, not a null.
846        $raw  = PayloadSigner::verify(base64_decode($rows[0]['payload']));
847        $task = ($raw !== false) ? @unserialize($raw) : false;
848
849        return ($task instanceof Task) ? $task : null;
850    }
851
852    /**
853     * Get every scheduled task, keyed by task ID.
854     *
855     * One query for the whole set, instead of the inherited "list the IDs, then
856     * SELECT each payload by ID" - which is a query per scheduled task, run on
857     * every Queue::run() and so on every tick of a worker's schedule loop. The
858     * payloads are decoded exactly as getTask() decodes its one, corrupt entries
859     * omitted rather than returned.
860     *
861     * @return array  taskId => Task
862     */
863    public function getAllTasks(): array
864    {
865        $sql = $this->db->createSql();
866        $sql->select(['job_id', 'payload'])->from($this->table)->where("type = 'task'");
867        $this->db->query($sql);
868
869        $tasks = [];
870
871        foreach ($this->db->fetchAll() as $row) {
872            $raw  = PayloadSigner::verify(base64_decode($row['payload']));
873            $task = ($raw !== false) ? @unserialize($raw) : false;
874
875            if ($task instanceof Task) {
876                $tasks[$row['job_id']] = $task;
877            }
878        }
879
880        return $tasks;
881    }
882
883    /**
884     * Update scheduled task
885     *
886     * @param  Task $task
887     * @return Database
888     */
889    public function updateTask(Task $task): Database
890    {
891        if ($task->isValid()) {
892            $sql = $this->db->createSql();
893            $sql->update($this->table)->values([
894                'payload' => ':payload'
895            ])->where("type = 'task'")->andWhere('job_id = :job_id');
896
897            $jobData = [
898                'payload' => base64_encode(PayloadSigner::sign(serialize(clone $task))),
899                'job_id'  => $task->getJobId()
900            ];
901
902            $this->db->prepare($sql);
903            $this->db->bindParams($jobData);
904            $this->db->execute();
905        } else {
906            $this->removeTask($task->getJobId());
907        }
908
909        return $this;
910    }
911
912    /**
913     * Remove scheduled task
914     *
915     * @param  string $taskId
916     * @return Database
917     */
918    public function removeTask(string $taskId): Database
919    {
920        $sql = $this->db->createSql();
921        $sql->delete()->from($this->table)->where("type = 'task'")->andWhere('job_id = :job_id');
922        $this->db->prepare($sql);
923        $this->db->bindParams(['job_id' => $taskId]);
924        $this->db->execute();
925
926        return $this;
927    }
928
929    /**
930     * Get scheduled tasks count
931     *
932     * @return int
933     */
934    public function getTaskCount(): int
935    {
936        $sql = $this->db->createSql();
937        $sql->select(['total' => 'COUNT(1)'])->from($this->table)->where("type = 'task'");
938        $this->db->query($sql);
939        $rows = $this->db->fetchAll();
940
941        return (isset($rows[0]['total'])) ? (int)$rows[0]['total'] : 0;
942    }
943
944    /**
945     * Has scheduled tasks
946     *
947     * @return bool
948     */
949    public function hasTasks(): bool
950    {
951        return ($this->getTaskCount() > 0);
952    }
953
954    /**
955     * Clear all scheduled task
956     *
957     * @return Database
958     */
959    public function clearTasks(): Database
960    {
961        $sql = $this->db->createSql();
962        $sql->delete()->from($this->table)->where("type = 'task'");
963        $this->db->query($sql);
964
965        return $this;
966    }
967
968    /**
969     * Create the database table
970     *
971     * @param  string $table
972     * @return Database
973     */
974    public function createTable(string $table): Database
975    {
976        $schema = $this->db->createSchema();
977
978        $schema->create($table)
979            ->int('id', 16)->increment()
980            ->int('index', 16)->nullable()
981            ->varchar('type', 255)
982            ->varchar('job_id', 255)
983            ->text('payload')
984            ->int('status', 1)->defaultIs(1)
985            ->int('reserved_until', 16)->nullable()
986            ->varchar('reserved_by', 64)->nullable()
987            ->primary('id');
988
989        $this->db->query($schema);
990
991        $this->createIndexes($table);
992
993        return $this;
994    }
995
996    /**
997     * Add the indexes the hot queries need to a table this adapter just created.
998     *
999     * Every query on the hot paths filters on type and then either orders by
1000     * index (reserve(), getEndIndex()) or looks a row up by job_id (getTask(),
1001     * claimedByTaskId(), the dead-letter accessors). Unindexed, all of those are
1002     * full table scans, and reserve()'s is a scan plus a sort - paid by every
1003     * worker on every job it runs, against a table whose whole purpose is to
1004     * accumulate rows.
1005     *
1006     * Issued as separate statements, and separate schema objects, for two
1007     * reasons that are easy to get wrong:
1008     *
1009     *  - Chaining index() onto the create() above renders valid DDL but does not
1010     *    execute it. The rendered schema becomes several statements separated by
1011     *    semicolons, and the adapters hand the whole string to one driver call -
1012     *    SQLite3::query() runs the first statement and silently discards the
1013     *    rest, so the table would appear and the indexes just wouldn't.
1014     *  - Casting a schema object to string consumes it. Rendering one to inspect
1015     *    it (or reusing one across two query() calls) leaves an empty builder
1016     *    behind, which also fails silently.
1017     *
1018     * The index names are given explicitly because deriving them would involve
1019     * the "index" column, a reserved word on every backend here.
1020     *
1021     * Only ever called for a table this adapter is creating from scratch. An
1022     * existing table is left alone deliberately: building an index on a live
1023     * queue table can hold a lock for as long as the table is large, which is a
1024     * decision for whoever operates the database, not one a library should make
1025     * on their behalf on first connect.
1026     *
1027     * @param  string $table
1028     * @return void
1029     */
1030    protected function createIndexes(string $table): void
1031    {
1032        foreach ([
1033            $table . '_type_index_idx'  => ['type', 'index'],
1034            $table . '_type_job_id_idx' => ['type', 'job_id'],
1035        ] as $name => $columns) {
1036            $schema = $this->db->createSchema();
1037            $schema->alter($table)->index($columns, $name);
1038            $this->db->query($schema);
1039        }
1040    }
1041
1042}