Code Coverage
 
Lines
Functions and Methods
Classes and Traits
Total
98.85%
86 / 87
85.71%
6 / 7
 CRAP
0.00%
0 / 1
ProcessDavQueueTask
98.84%
85 / 86
85.71%
6 / 7
 17
0.00%
0 / 1
 __construct
100.00%
1 / 1
100.00%
1 / 1
 1
 run
100.00%
28 / 28
100.00%
1 / 1
 5
 executeJob
90.91%
10 / 11
0.00%
0 / 1
 5.02
 fetchPendingJobs
100.00%
14 / 14
100.00%
1 / 1
 1
 markJobRunning
100.00%
8 / 8
100.00%
1 / 1
 1
 markJobCompleted
100.00%
8 / 8
100.00%
1 / 1
 1
 markJobFailed
100.00%
16 / 16
100.00%
1 / 1
 3
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\Dav\Task;
8
9defined('AMMONLY_APP') || exit('Direct script access is forbidden.');
10
11use App\Modules\Dav\Application\Service\DavAccountResolver;
12use App\Modules\Dav\Application\Service\DavCalendarSyncService;
13use App\Modules\Dav\Application\Service\DavContactSyncService;
14use App\Modules\Dav\Application\Service\DavQueueDispatcher;
15use Exception;
16use PDO;
17
18/**
19 * Background DAV Push Queue Worker CRON Task.
20 *
21 * Implements CronTaskInterface to process asynchronous CalDAV & CardDAV push jobs.
22 *
23 * @package App\Modules\Dav\Task
24 */
25final readonly class ProcessDavQueueTask extends AbstractDavTask
26{
27    /**
28     * ProcessDavQueueTask constructor.
29     *
30     * @param PDO|null $pdo Optional PDO database connection instance.
31     * @param DavCalendarSyncService|null $calendarSync Calendar synchronization service.
32     * @param DavContactSyncService|null $contactSync Contact synchronization service.
33     * @param DavAccountResolver|null $accountResolver Account resolver service.
34     * @param string $tablePrefix Table prefix.
35     * @param int $maxJobsPerRun Batch size per cron execution.
36     */
37    public function __construct(
38        ?PDO $pdo = null,
39        ?DavCalendarSyncService $calendarSync = null,
40        ?DavContactSyncService $contactSync = null,
41        ?DavAccountResolver $accountResolver = null,
42        string $tablePrefix = 'a_',
43        private int $maxJobsPerRun = 25
44    ) {
45        parent::__construct($pdo, $calendarSync, $contactSync, $accountResolver, $tablePrefix);
46    }
47
48    /**
49     * {@inheritdoc}
50     */
51    public function run(): string
52    {
53        $pdo = $this->resolvePdo();
54        $calendarSync = $this->resolveCalendarSync($pdo);
55        $contactSync = $this->resolveContactSync($pdo);
56        $accountResolver = $this->resolveAccountResolver($pdo);
57
58        $jobs = $this->fetchPendingJobs($pdo);
59        if (empty($jobs)) {
60            return 'DAV push queue worker: No pending jobs.';
61        }
62
63        $successCount = 0;
64        $failCount = 0;
65
66        foreach ($jobs as $job) {
67            $jobId = (int) $job['id'];
68            $jobType = (string) $job['job_type'];
69            $payload = json_decode((string) $job['payload'], true) ?: [];
70
71            $this->markJobRunning($pdo, $jobId);
72
73            try {
74                $this->executeJob($jobType, $payload, $accountResolver, $calendarSync, $contactSync);
75                $this->markJobCompleted($pdo, $jobId);
76                $successCount++;
77            } catch (Exception $e) {
78                $attempts = (int) $job['attempts'];
79                $maxAttempts = (int) $job['max_attempts'];
80                $this->markJobFailed($pdo, $jobId, $attempts, $maxAttempts, $e->getMessage());
81                $failCount++;
82            }
83        }
84
85        return sprintf(
86            'DAV push queue worker completed %d jobs (%d succeeded, %d failed).',
87            count($jobs),
88            $successCount,
89            $failCount
90        );
91    }
92
93    /**
94     * Executes a single queue job payload.
95     *
96     * @param string $jobType Type identifier of the job.
97     * @param array<string, mixed> $payload Job payload array.
98     * @param DavAccountResolver $accountResolver User account resolver.
99     * @param DavCalendarSyncService $calendarSync Calendar service.
100     * @param DavContactSyncService $contactSync Contact service.
101     * @return void
102     */
103    private function executeJob(
104        string $jobType,
105        array $payload,
106        DavAccountResolver $accountResolver,
107        DavCalendarSyncService $calendarSync,
108        DavContactSyncService $contactSync
109    ): void {
110        $userId = (int) ($payload['user_id'] ?? 1);
111        $account = $accountResolver->resolveByUserId($userId);
112
113        if ($account === null) {
114            // User does not have a DAV mailbox configured, mark completed as no-op.
115            return;
116        }
117
118        $entityId = (int) ($payload['entity_id'] ?? 0);
119        $action = (string) ($payload['action'] ?? 'update');
120        $snapshot = is_array($payload['snapshot'] ?? null) ? $payload['snapshot'] : [];
121
122        if ($jobType === DavQueueDispatcher::JOB_CALENDAR_PUSH) {
123            $calendarSync->pushEvent($entityId, $action, $account, $snapshot);
124        } elseif ($jobType === DavQueueDispatcher::JOB_CONTACT_PUSH) {
125            $contactSync->pushContact($entityId, $action, $account, $snapshot);
126        }
127    }
128
129    /**
130     * Fetches pending DAV jobs from database.
131     *
132     * @param PDO $pdo PDO connection.
133     * @return array<int, array<string, mixed>> List of pending job records.
134     */
135    private function fetchPendingJobs(PDO $pdo): array
136    {
137        $table = $this->tablePrefix . 'mod_queue_records';
138        $sql = sprintf(
139            'SELECT `id`, `job_type`, `payload`, `attempts`, `max_attempts` FROM `%s` '
140            . 'WHERE `job_type` IN (:cal_push, :con_push) AND `status` = \'pending\' '
141            . 'ORDER BY `id` ASC LIMIT %d',
142            $table,
143            $this->maxJobsPerRun
144        );
145
146        $stmt = $pdo->prepare($sql);
147        $stmt->execute([
148            ':cal_push' => DavQueueDispatcher::JOB_CALENDAR_PUSH,
149            ':con_push' => DavQueueDispatcher::JOB_CONTACT_PUSH,
150        ]);
151
152        return $stmt->fetchAll(PDO::FETCH_ASSOC);
153    }
154
155    /**
156     * Marks job status as running.
157     *
158     * @param PDO $pdo PDO connection.
159     * @param int $jobId Job ID.
160     * @return void
161     */
162    private function markJobRunning(PDO $pdo, int $jobId): void
163    {
164        $table = $this->tablePrefix . 'mod_queue_records';
165        $sql = sprintf(
166            'UPDATE `%s` SET `status` = \'running\', `started_at` = CURRENT_TIMESTAMP, '
167            . '`heartbeat_at` = CURRENT_TIMESTAMP WHERE `id` = :id',
168            $table
169        );
170        $stmt = $pdo->prepare($sql);
171        $stmt->execute([':id' => $jobId]);
172    }
173
174    /**
175     * Marks job status as completed.
176     *
177     * @param PDO $pdo PDO connection.
178     * @param int $jobId Job ID.
179     * @return void
180     */
181    private function markJobCompleted(PDO $pdo, int $jobId): void
182    {
183        $table = $this->tablePrefix . 'mod_queue_records';
184        $sql = sprintf(
185            'UPDATE `%s` SET `status` = \'completed\', `progress_percent` = 100, `processed_items` = 1, '
186            . '`completed_at` = CURRENT_TIMESTAMP WHERE `id` = :id',
187            $table
188        );
189        $stmt = $pdo->prepare($sql);
190        $stmt->execute([':id' => $jobId]);
191    }
192
193    /**
194     * Marks job status as failed or increments attempts.
195     *
196     * @param PDO $pdo PDO connection.
197     * @param int $jobId Job ID.
198     * @param int $currentAttempts Current attempt count.
199     * @param int $maxAttempts Maximum allowed attempts.
200     * @param string $errorMessage Encountered error text.
201     * @return void
202     */
203    private function markJobFailed(
204        PDO $pdo,
205        int $jobId,
206        int $currentAttempts,
207        int $maxAttempts,
208        string $errorMessage
209    ): void {
210        $nextAttempts = $currentAttempts + 1;
211        $status = ($nextAttempts >= $maxAttempts) ? 'failed' : 'pending';
212        $failedAtClause = ($status === 'failed') ? '`failed_at` = CURRENT_TIMESTAMP,' : '';
213
214        $table = $this->tablePrefix . 'mod_queue_records';
215        $sql = sprintf(
216            'UPDATE `%s` SET `status` = :status, `attempts` = :attempts, %s `error_message` = :err WHERE `id` = :id',
217            $table,
218            $failedAtClause
219        );
220        $stmt = $pdo->prepare($sql);
221        $stmt->execute([
222            ':status' => $status,
223            ':attempts' => $nextAttempts,
224            ':err' => substr($errorMessage, 0, 1000),
225            ':id' => $jobId,
226        ]);
227    }
228}