Code Coverage
 
Lines
Functions and Methods
Classes and Traits
Total
100.00%
28 / 28
100.00%
6 / 6
 CRAP
100.00%
1 / 1
Queue
100.00%
28 / 28
100.00%
6 / 6
 12
100.00%
1 / 1
 __construct
100.00%
1 / 1
100.00%
1 / 1
 1
 push
100.00%
19 / 19
100.00%
1 / 1
 6
 pending
100.00%
5 / 5
100.00%
1 / 1
 1
 failed
100.00%
1 / 1
100.00%
1 / 1
 1
 retry
100.00%
1 / 1
100.00%
1 / 1
 2
 prune
100.00%
1 / 1
100.00%
1 / 1
 1
1<?php
2
3declare(strict_types=1);
4
5namespace Cosray;
6
7use Celema\Quma\Database;
8use Cosray\Contract\Job;
9use Cosray\Exception\RuntimeException;
10use JsonException;
11
12/**
13 * The job queue in the site's database: pushes jobs and manages the ones
14 * that failed. Runners take the jobs from there; see `docs/queue.md`.
15 *
16 * A push inside a transaction belongs to it. The job is queued only if the
17 * transaction commits, so a write and the work it causes, like a stored
18 * form submission and the mail about it, happen together or not at all.
19 *
20 *     $queue->push(SendReceipt::class, ['order' => $order], delay: 60);
21 *
22 * @api
23 */
24final class Queue
25{
26    private const int MAX_NAME_LENGTH = 64;
27
28    public function __construct(
29        private readonly Database $db,
30    ) {}
31
32    /**
33     * @param class-string<Job> $job
34     * @param array<array-key, mixed> $payload handed to the job after a JSON round trip
35     * @param int $delay seconds until the job is due
36     * @param string $queue a name runners can pick jobs by
37     * @return int the job's id
38     */
39    public function push(string $job, array $payload = [], int $delay = 0, string $queue = 'default'): int
40    {
41        if (!is_a($job, Job::class, true)) {
42            throw new RuntimeException("A queued job implements Cosray\\Contract\\Job; {$job} does not.");
43        }
44
45        if ($delay < 0) {
46            throw new RuntimeException('A job cannot be due in the past.');
47        }
48
49        if ($queue === '' || mb_strlen($queue) > self::MAX_NAME_LENGTH) {
50            throw new RuntimeException('A queue name has 1 to ' . self::MAX_NAME_LENGTH . ' characters.');
51        }
52
53        try {
54            $json = json_encode(
55                $payload,
56                JSON_THROW_ON_ERROR | JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE | JSON_PRESERVE_ZERO_FRACTION,
57            );
58        } catch (JsonException $e) {
59            throw new RuntimeException("The payload for {$job} is not JSON: {$e->getMessage()}", previous: $e);
60        }
61
62        /** @var array{id: int} $row */
63        $row = $this->db->queue->push([
64            'queue' => $queue,
65            'job' => $job,
66            'payload' => $json,
67            'delay' => $delay,
68        ])->one();
69
70        return (int) $row['id'];
71    }
72
73    /**
74     * The payloads of a job class's jobs that have yet to succeed, in the
75     * order they were pushed; failed jobs are left out. Tests read what the
76     * code under test queued with it.
77     *
78     * @param class-string<Job> $job
79     * @return list<array<array-key, mixed>>
80     */
81    public function pending(string $job): array
82    {
83        /** @var list<array{payload: string}> $rows */
84        $rows = $this->db->queue->pending(['job' => $job])->all();
85
86        return array_map(
87            static fn(array $row): array => (array) json_decode($row['payload'], true, flags: JSON_THROW_ON_ERROR),
88            $rows,
89        );
90    }
91
92    /** @return list<array{id: int, queue: string, job: string, attempts: int, error: ?string, created: string, failed: string}> */
93    public function failed(): array
94    {
95        /** @var list<array{id: int, queue: string, job: string, attempts: int, error: ?string, created: string, failed: string}> */
96        return $this->db->queue->failed()->all();
97    }
98
99    /** Queues failed jobs again with fresh attempts, one or all; returns how many. */
100    public function retry(?int $id = null): int
101    {
102        return $this->db->queue->requeue($id === null ? [] : ['id' => $id])->len();
103    }
104
105    /** Deletes the jobs that failed more than the given number of days ago; returns how many. */
106    public function prune(int $days): int
107    {
108        return $this->db->queue->prune(['days' => max(0, $days)])->len();
109    }
110}