Lines 96.51% 83 / 86
Methods 66.66% 4 / 6
Classes 0.00% 0 / 1
Covered by tests of size
Name Lines Methods CRAP
 __construct 100.00% 1 / 1 100.00% 1 / 1 1
 enqueue 100.00% 19 / 19 100.00% 1 / 1 2
 reserve 94.28% 33 / 35 0.00% 0 / 1 6.01
 acknowledge 100.00% 4 / 4 100.00% 1 / 1 1
 release 100.00% 18 / 18 100.00% 1 / 1 2
 decodeHeaders 88.88% 8 / 9 0.00% 0 / 1 4.02
27readonly class DatabaseTransport
28{
29    public function __construct(
30        private Configuration $configuration,
31    ) {
32    }
33
34    /**
35     * @param array<string, mixed> $headers
36     */
37    public function enqueue(
38        string $body,
39        array $headers = [],
40        string $queue = 'default',
41        ?DateTimeImmutable $availableAt = null,
42    ): int {
43        $db = $this->configuration->getDb();
44        $table = Database::getTablePrefix() . 'faqjobs';
45
46        $availableAt ??= new DateTimeImmutable();
47        $availableAtValue = $db->escape($availableAt->format('Y-m-d H:i:s'));
48        $queueValue = $db->escape($queue);
49        $bodyValue = $db->escape($body);
50        $headersValue = $db->escape(json_encode($headers, JSON_THROW_ON_ERROR));
51
52        $query = sprintf(
53            "INSERT INTO %s (queue, body, headers, available_at, delivered_at, created) VALUES ('%s', '%s', '%s', '%s', NULL, %s)",
54            $table,
55            $queueValue,
56            $bodyValue,
57            $headersValue,
58            $availableAtValue,
59            $db->now(),
60        );
61
62        if ($db->query($query) === false) {
63            throw new RuntimeException('Unable to enqueue job: ' . $db->error());
64        }
65
66        return (int) $db->lastInsertId();
67    }
68
69    /**
70     * @return array{id: int, queue: string, body: string, headers: array<string, mixed>}|null
71     */
72    public function reserve(string $queue = 'default'): ?array
73    {
74        $db = $this->configuration->getDb();
75        $table = Database::getTablePrefix() . 'faqjobs';
76        $escapedQueue = $db->escape($queue);
77
78        $nowValue = $db->escape(new DateTimeImmutable()->format('Y-m-d H:i:s'));
79
80        $query = sprintf(
81            "SELECT id, queue, body, headers FROM %s WHERE queue = '%s' AND delivered_at IS NULL AND available_at <= '%s' ORDER BY available_at ASC, id ASC",
82            $table,
83            $escapedQueue,
84            $nowValue,
85        );
86
87        $result = $db->query($query, 0, 10);
88        if ($result === false) {
89            throw new RuntimeException('Unable to fetch queued jobs: ' . $db->error());
90        }
91
92        while (true) {
93            $row = $db->fetchArray($result);
94            if (!is_array($row)) {
95                break;
96            }
97
98            $jobId = (int) $row['id'];
99            $markDelivered = sprintf(
100                'UPDATE %s SET delivered_at = %s WHERE id = %d AND delivered_at IS NULL',
101                $table,
102                $db->now(),
103                $jobId,
104            );
105
106            if ($db->query($markDelivered) === false) {
107                continue;
108            }
109
110            if ($db->affectedRows() !== 1) {
111                continue;
112            }
113
114            return [
115                'id' => $jobId,
116                'queue' => (string) $row['queue'],
117                'body' => (string) $row['body'],
118                'headers' => $this->decodeHeaders((string) ($row['headers'] ?? '')),
119            ];
120        }
121
122        return null;
123    }
124
125    public function acknowledge(int $jobId): bool
126    {
127        $db = $this->configuration->getDb();
128        $table = Database::getTablePrefix() . 'faqjobs';
129        $query = sprintf('DELETE FROM %s WHERE id = %d', $table, $jobId);
130
131        return $db->query($query) !== false;
132    }
133
134    /**
135     * @param array<string, mixed>|null $headers Updated headers to persist, or null to leave unchanged.
136     */
137    public function release(int $jobId, ?DateTimeImmutable $availableAt = null, ?array $headers = null): bool
138    {
139        $db = $this->configuration->getDb();
140        $table = Database::getTablePrefix() . 'faqjobs';
141        $availableAt ??= new DateTimeImmutable('+60 seconds');
142
143        $query = sprintf(
144            "UPDATE %s SET delivered_at = NULL, available_at = '%s' WHERE id = %d",
145            $table,
146            $db->escape($availableAt->format('Y-m-d H:i:s')),
147            $jobId,
148        );
149
150        if ($headers !== null) {
151            $query = sprintf(
152                "UPDATE %s SET delivered_at = NULL, available_at = '%s', headers = '%s' WHERE id = %d",
153                $table,
154                $db->escape($availableAt->format('Y-m-d H:i:s')),
155                $db->escape(json_encode($headers, JSON_THROW_ON_ERROR)),
156                $jobId,
157            );
158        }
159
160        return $db->query($query) !== false;
161    }
162
163    /**
164     * @return array<string, mixed>
165     */
166    private function decodeHeaders(string $rawHeaders): array
167    {
168        if ($rawHeaders === '') {
169            return [];
170        }
171
172        $decoded = json_decode($rawHeaders, associative: true);
173        if (!is_array($decoded)) {
174            return [];
175        }
176
177        $headers = [];
178        foreach ($decoded as $headerName => $headerValue) {
179            $headers[(string) $headerName] = $headerValue;
180        }
181
182        return $headers;
183    }
184}