Code Coverage
 
Lines
Functions and Methods
Classes and Traits
Total
89.06%
57 / 64
50.00%
2 / 4
 CRAP
0.00%
0 / 1
ProcessQueueTask
88.89%
56 / 63
50.00%
2 / 4
 11.17
0.00%
0 / 1
 __construct
100.00%
1 / 1
100.00%
1 / 1
 1
 run
100.00%
14 / 14
100.00%
1 / 1
 3
 resolveQueueManager
86.84%
33 / 38
0.00%
0 / 1
 3.02
 resolvePdo
80.00%
8 / 10
0.00%
0 / 1
 4.13
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\Task;
8
9defined('AMMONLY_APP') || exit('Direct script access is forbidden.');
10
11use App\Core\Cron\CronTaskInterface;
12use App\Core\Database\MultiDbRouter;
13use App\Modules\Automation\Queue\Application\Handler\PicklistValueReassignHandler;
14use App\Modules\Automation\Queue\Application\Service\QueueManager;
15use App\Modules\Automation\Queue\Infrastructure\Repository\SqlQueueRepository;
16use DateTimeImmutable;
17use PDO;
18
19/**
20 * Background Task Queue Worker CRON Task.
21 *
22 * Implements CronTaskInterface to process asynchronous queue tasks,
23 * track heartbeats, and automatically unblock zombie processes.
24 *
25 * @package App\Modules\Automation\Queue\Task
26 */
27final readonly class ProcessQueueTask implements CronTaskInterface
28{
29    /**
30     * ProcessQueueTask constructor.
31     *
32     * @param QueueManager|PDO|null $queueSource Queue manager, PDO connection, or null.
33     * @param string $tablePrefix Table prefix.
34     * @param int $maxJobsPerRun Maximum jobs processed in a single run batch.
35     * @param int $zombieTimeoutSeconds Timeout in seconds before a stalled job is recovered.
36     * @param \App\Core\DataExchange\Application\Service\DataImportService|null $dataImportService Data import service.
37     */
38    public function __construct(
39        private QueueManager|PDO|null $queueSource = null,
40        private string $tablePrefix = 'a_',
41        private int $maxJobsPerRun = 10,
42        private int $zombieTimeoutSeconds = 300,
43        private ?\App\Core\DataExchange\Application\Service\DataImportService $dataImportService = null
44    ) {
45    }
46
47    /**
48     * {@inheritdoc}
49     */
50    public function run(): string
51    {
52        $queueManager = $this->resolveQueueManager();
53        $now = new DateTimeImmutable();
54
55        // 1. Recover any zombie/crashed jobs with stale heartbeat
56        $recoveredZombies = $queueManager->recoverZombies($this->zombieTimeoutSeconds, $now);
57
58        // 2. Process up to maxJobsPerRun pending queue jobs
59        $processedCount = 0;
60        for ($i = 0; $i < $this->maxJobsPerRun; $i++) {
61            $jobNow = new DateTimeImmutable();
62            $hasJob = $queueManager->processNext($jobNow);
63            if (!$hasJob) {
64                break;
65            }
66            $processedCount++;
67        }
68
69        return sprintf(
70            'Queue worker processed %d jobs. Recovered %d stale zombie tasks.',
71            $processedCount,
72            $recoveredZombies
73        );
74    }
75
76    /**
77     * Resolves the QueueManager instance with all standard handlers registered.
78     *
79     * @return QueueManager Configured queue manager.
80     */
81    private function resolveQueueManager(): QueueManager
82    {
83        if ($this->queueSource instanceof QueueManager) {
84            return $this->queueSource;
85        }
86
87        $pdo = $this->resolvePdo();
88        $repo = new SqlQueueRepository($pdo, $this->tablePrefix);
89        $reassignHandler = new PicklistValueReassignHandler($pdo, $this->tablePrefix);
90        $bulkActionHandler = new \App\Modules\Automation\Queue\Application\Handler\BulkActionJobHandler(
91            $pdo,
92            $this->tablePrefix
93        );
94
95        $davClient = new \App\Modules\Dav\Infrastructure\Client\CurlDavClient();
96        $davCalSync = new \App\Modules\Dav\Application\Service\DavCalendarSyncService(
97            $pdo,
98            $davClient,
99            new \App\Modules\Dav\Infrastructure\Converter\ICalendarConverter(),
100            $this->tablePrefix
101        );
102        $davConSync = new \App\Modules\Dav\Application\Service\DavContactSyncService(
103            $pdo,
104            $davClient,
105            new \App\Modules\Dav\Infrastructure\Converter\VCardConverter(),
106            $this->tablePrefix
107        );
108        $davResolver = new \App\Modules\Dav\Application\Service\DavAccountResolver($pdo, $this->tablePrefix);
109
110        $davCalHandler = new \App\Modules\Dav\Application\Handler\DavCalendarPushJobHandler($davCalSync, $davResolver);
111        $davConHandler = new \App\Modules\Dav\Application\Handler\DavContactPushJobHandler($davConSync, $davResolver);
112
113        $handlers = [
114            $reassignHandler,
115            $bulkActionHandler,
116            $davCalHandler,
117            $davConHandler,
118        ];
119
120        if ($this->dataImportService !== null) {
121            $handlers[] = new \App\Core\DataExchange\Application\Handler\ImportQueueJobHandler(
122                $this->dataImportService,
123                $pdo,
124                $this->tablePrefix
125            );
126        }
127
128        return new QueueManager($repo, $handlers);
129    }
130
131    /**
132     * Resolves active PDO connection.
133     *
134     * @return PDO Database connection handle.
135     */
136    private function resolvePdo(): PDO
137    {
138        if ($this->queueSource instanceof PDO) {
139            return $this->queueSource;
140        }
141
142        try {
143            $config = \App\Shared\Infrastructure\Config\InstallerConfigLoader::load();
144            $dbParams = is_array($config) ? ($config['db'] ?? []) : [];
145
146            $router = new MultiDbRouter([
147                'default' => $dbParams,
148            ]);
149
150            return $router->getConnection('default');
151        } catch (\Throwable) {
152            return new PDO('sqlite::memory:');
153        }
154    }
155}