Code Coverage
 
Lines
Functions and Methods
Classes and Traits
Total
90.20%
46 / 51
80.00%
4 / 5
 CRAP
0.00%
0 / 1
Worker
90.20%
46 / 51
80.00%
4 / 5
 19.34
0.00%
0 / 1
 __construct
100.00%
1 / 1
100.00%
1 / 1
 1
 registerHandler
100.00%
1 / 1
100.00%
1 / 1
 1
 runOnce
100.00%
20 / 20
100.00%
1 / 1
 5
 run
100.00%
14 / 14
100.00%
1 / 1
 5
 decodeMessage
66.67%
10 / 15
0.00%
0 / 1
 8.81
1<?php
2
3/**
4 * Queue worker.
5 *
6 * This Source Code Form is subject to the terms of the Mozilla Public License,
7 * v. 2.0. If a copy of the MPL was not distributed with this file, You can
8 * obtain one at https://mozilla.org/MPL/2.0/.
9 *
10 * @package   phpMyFAQ
11 * @author    Thorsten Rinne <thorsten@phpmyfaq.de>
12 * @copyright 2026 phpMyFAQ Team
13 * @license   https://www.mozilla.org/MPL/2.0/ Mozilla Public License Version 2.0
14 * @link      https://www.phpmyfaq.de
15 * @since     2026-02-11
16 */
17
18declare(strict_types=1);
19
20namespace phpMyFAQ\Queue;
21
22use DateTimeImmutable;
23use phpMyFAQ\Queue\Message\QueueMessageInterface;
24use phpMyFAQ\Queue\Transport\DatabaseTransport;
25use RuntimeException;
26use Throwable;
27
28class Worker
29{
30    private const int MAX_RETRIES = 3;
31
32    /** @var array<string, callable> */
33    private array $handlers = [];
34
35    public function __construct(
36        private readonly DatabaseTransport $databaseTransport,
37    ) {
38    }
39
40    public function registerHandler(string $messageClass, callable $handler): void
41    {
42        $this->handlers[$messageClass] = $handler;
43    }
44
45    /**
46     * Runs one available job.
47     */
48    public function runOnce(string $queue = 'default'): bool
49    {
50        $job = $this->databaseTransport->reserve($queue);
51        if ($job === null) {
52            return false;
53        }
54
55        $jobId = (int) $job['id'];
56        $headers = $job['headers'];
57
58        try {
59            $message = $this->decodeMessage($job['body']);
60            $handler = $this->handlers[$message::class] ?? null;
61
62            if (!is_callable($handler)) {
63                throw new RuntimeException('No queue handler registered for message class: ' . $message::class);
64            }
65
66            $handler($message);
67            $this->databaseTransport->acknowledge($jobId);
68
69            return true;
70        } catch (Throwable) {
71            $attempts = (int) ($headers['attempts'] ?? 0) + 1;
72            $headers['attempts'] = $attempts;
73
74            if ($attempts >= self::MAX_RETRIES) {
75                $this->databaseTransport->acknowledge($jobId);
76                return true;
77            }
78
79            $this->databaseTransport->release($jobId, new DateTimeImmutable('+60 seconds'), $headers);
80            return true;
81        }
82    }
83
84    /**
85     * Runs until the queue is empty or the maximum number of jobs has been processed.
86     */
87    public function run(int $maxJobs = 0, string $queue = 'default'): int
88    {
89        $processed = 0;
90
91        while ($maxJobs === 0 || $processed < $maxJobs) {
92            try {
93                if (!$this->runOnce($queue)) {
94                    break;
95                }
96
97                ++$processed;
98            } catch (Throwable $exception) {
99                error_log(sprintf(
100                    'Queue worker error in run() while processing queue "%s": %s in %s:%d',
101                    $queue,
102                    $exception->getMessage(),
103                    $exception->getFile(),
104                    $exception->getLine(),
105                ));
106            }
107        }
108
109        return $processed;
110    }
111
112    private function decodeMessage(string $body): QueueMessageInterface
113    {
114        $decoded = json_decode($body, associative: true);
115        if (!is_array($decoded) || !array_key_exists('class', $decoded)) {
116            throw new RuntimeException('Queue job body has an invalid format.');
117        }
118
119        $messageClass = (string) $decoded['class'];
120        $rawPayload = is_array($decoded['payload'] ?? null) ? $decoded['payload'] : [];
121        $payload = [];
122        foreach ($rawPayload as $payloadKey => $payloadValue) {
123            $payload[(string) $payloadKey] = $payloadValue;
124        }
125
126        if (!class_exists($messageClass)) {
127            throw new RuntimeException('Queue job references unknown message class: ' . $messageClass);
128        }
129
130        if (!is_subclass_of($messageClass, QueueMessageInterface::class)) {
131            throw new RuntimeException(
132                'Queue message class ' . $messageClass . ' does not implement ' . QueueMessageInterface::class,
133            );
134        }
135
136        /** @var class-string<QueueMessageInterface> $messageClass */
137        return $messageClass::fromArray($payload);
138    }
139}