Code Coverage
 
Lines
Functions and Methods
Classes and Traits
Total
100.00%
28 / 28
100.00%
6 / 6
 CRAP
100.00%
1 / 1
QueueManager
100.00%
27 / 27
100.00%
6 / 6
 10
100.00%
1 / 1
 __construct
100.00%
2 / 2
100.00%
1 / 1
 2
 registerHandler
100.00%
1 / 1
100.00%
1 / 1
 1
 enqueue
100.00%
1 / 1
100.00%
1 / 1
 1
 processNext
100.00%
5 / 5
100.00%
1 / 1
 2
 executeJobWithSafety
100.00%
16 / 16
100.00%
1 / 1
 3
 recoverZombies
100.00%
2 / 2
100.00%
1 / 1
 1
1<?php
2
3declare(strict_types=1);
4
5/** @license For full copyright and license information, please see the LICENSE.md file. */
6
7namespace App\Modules\Automation\Queue\Application\Service;
8
9defined('AMMONLY_APP') || exit('Direct script access is forbidden.');
10
11use App\Modules\Automation\Queue\Application\Handler\JobHandlerInterface;
12use App\Modules\Automation\Queue\Domain\Model\QueueJob;
13use App\Modules\Automation\Queue\Domain\Repository\QueueRepositoryInterface;
14use DateTimeImmutable;
15use RuntimeException;
16use Throwable;
17
18/**
19 * Queue Application Manager Service.
20 *
21 * Orchestrates background job registration, handler discovery, safe execution,
22 * and automatic zombie recovery with heartbeat timeout detection.
23 *
24 * @package App\Modules\Automation\Queue\Application\Service
25 */
26final class QueueManager
27{
28    /**
29     * @var array<string, JobHandlerInterface> Registered job handlers keyed by job type.
30     */
31    private array $handlers = [];
32
33    /**
34     * QueueManager constructor.
35     *
36     * @param QueueRepositoryInterface $queueRepository Queue persistence repository.
37     * @param array<int, JobHandlerInterface> $handlers Initial handlers list.
38     */
39    public function __construct(
40        private readonly QueueRepositoryInterface $queueRepository,
41        array $handlers = []
42    ) {
43        foreach ($handlers as $handler) {
44            $this->registerHandler($handler);
45        }
46    }
47
48    /**
49     * Registers a specialized job handler.
50     *
51     * @param JobHandlerInterface $handler Handler instance.
52     * @return void
53     */
54    public function registerHandler(JobHandlerInterface $handler): void
55    {
56        $this->handlers[$handler->supports()] = $handler;
57    }
58
59    /**
60     * Enqueues a new background task.
61     *
62     * @param string $jobType Machine name of the job type.
63     * @param string $label Human-readable title.
64     * @param array<string, mixed> $payload Serialized execution parameters.
65     * @param int $createdBy User ID.
66     * @param int $owner Owner User ID.
67     * @return int Enqueued job ID.
68     */
69    public function enqueue(
70        string $jobType,
71        string $label,
72        array $payload,
73        int $createdBy = 1,
74        int $owner = 1
75    ): int {
76        return $this->queueRepository->enqueue($jobType, $label, $payload, $createdBy, $owner);
77    }
78
79    /**
80     * Fetches and processes the next pending queue job.
81     *
82     * @param DateTimeImmutable $now Current execution timestamp.
83     * @return bool True if a job was found and processed, false if queue is idle.
84     */
85    public function processNext(DateTimeImmutable $now): bool
86    {
87        $job = $this->queueRepository->fetchNextPending($now);
88        if ($job === null) {
89            return false;
90        }
91
92        $this->executeJobWithSafety($job, $now);
93
94        return true;
95    }
96
97    /**
98     * Executes single job with strict exception safety and state transitions.
99     *
100     * @param QueueJob $job Job to execute.
101     * @param DateTimeImmutable $now Execution start timestamp.
102     * @return void
103     */
104    private function executeJobWithSafety(QueueJob $job, DateTimeImmutable $now): void
105    {
106        $jobId = $job->getId();
107        $jobType = $job->getJobType();
108
109        if (!isset($this->handlers[$jobType])) {
110            $this->queueRepository->markFailed(
111                $jobId,
112                "No registered handler found for job type '{$jobType}'.",
113                $now
114            );
115            return;
116        }
117
118        $handler = $this->handlers[$jobType];
119
120        try {
121            $outputLog = $handler->handle($job, $this->queueRepository);
122            $completionNow = new DateTimeImmutable();
123            $this->queueRepository->markCompleted($jobId, $completionNow, $outputLog);
124        } catch (Throwable $e) {
125            $failureNow = new DateTimeImmutable();
126            $this->queueRepository->markFailed($jobId, $e->getMessage(), $failureNow);
127        }
128    }
129
130    /**
131     * Recovers dead/stale zombie jobs with expired heartbeat.
132     *
133     * @param int $staleSeconds Stale threshold in seconds (default 300 = 5 min).
134     * @param DateTimeImmutable|null $now Optional timestamp.
135     * @return int Count of recovered/failed jobs.
136     */
137    public function recoverZombies(int $staleSeconds = 300, ?DateTimeImmutable $now = null): int
138    {
139        $currentNow = $now ?? new DateTimeImmutable();
140        return $this->queueRepository->recoverStaleJobs($staleSeconds, $currentNow);
141    }
142}