Code Coverage
 
Lines
Functions and Methods
Classes and Traits
Total
95.68%
155 / 162
90.00%
9 / 10
 CRAP
0.00%
0 / 1
SqlQueueRepository
95.65%
154 / 161
90.00%
9 / 10
 32
0.00%
0 / 1
 __construct
100.00%
1 / 1
100.00%
1 / 1
 1
 enqueue
100.00%
14 / 14
100.00%
1 / 1
 1
 fetchNextPending
84.44%
38 / 45
0.00%
0 / 1
 14.74
 recoverStaleJobs
100.00%
34 / 34
100.00%
1 / 1
 3
 updateProgress
100.00%
12 / 12
100.00%
1 / 1
 2
 markCompleted
100.00%
10 / 10
100.00%
1 / 1
 1
 markFailed
100.00%
10 / 10
100.00%
1 / 1
 1
 findById
100.00%
9 / 9
100.00%
1 / 1
 2
 hydrateJob
100.00%
23 / 23
100.00%
1 / 1
 4
 parseDateTime
100.00%
3 / 3
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\Automation\Queue\Infrastructure\Repository;
8
9defined('AMMONLY_APP') || exit('Direct script access is forbidden.');
10
11use App\Modules\Automation\Queue\Domain\Model\QueueJob;
12use App\Modules\Automation\Queue\Domain\Repository\QueueRepositoryInterface;
13use DateTimeImmutable;
14use PDO;
15
16/**
17 * SQL Persistence Implementation for Background Queue Tasks.
18 *
19 * Implements transaction-safe queuing, atomic worker locking, progress tracking,
20 * and automated zombie task recovery with heartbeat timeouts.
21 *
22 * @package App\Modules\Automation\Queue\Infrastructure\Repository
23 */
24final readonly class SqlQueueRepository implements QueueRepositoryInterface
25{
26    private const string DATE_FORMAT = 'Y-m-d H:i:s.u';
27    private const string PARAM_PENDING = ':pending';
28
29    /**
30     * SqlQueueRepository constructor.
31     *
32     * @param PDO $pdo Database connection handle.
33     * @param string $tablePrefix Database table prefix.
34     */
35    public function __construct(
36        private PDO $pdo,
37        private string $tablePrefix = 'a_'
38    ) {
39    }
40
41    /**
42     * {@inheritdoc}
43     */
44    public function enqueue(
45        string $jobType,
46        string $label,
47        array $payload,
48        int $createdBy = 1,
49        int $owner = 1
50    ): int {
51        $tableName = $this->tablePrefix . 'mod_queue_records';
52        $sql = "INSERT INTO {$tableName}
53                (job_type, label, status, payload, progress_percent, total_items,
54                 processed_items, attempts, max_attempts, created_by, owner)
55                VALUES
56                (:job_type, :label, :status, :payload, 0, 0, 0, 0, 3, :created_by, :owner)";
57
58        $stmt = $this->pdo->prepare($sql);
59        $payloadJson = (string)json_encode($payload, JSON_UNESCAPED_UNICODE | JSON_THROW_ON_ERROR);
60        $status = QueueJob::STATUS_PENDING;
61
62        $stmt->bindValue(':job_type', $jobType);
63        $stmt->bindValue(':label', $label);
64        $stmt->bindValue(':status', $status);
65        $stmt->bindValue(':payload', $payloadJson);
66        $stmt->bindValue(':created_by', $createdBy, PDO::PARAM_INT);
67        $stmt->bindValue(':owner', $owner, PDO::PARAM_INT);
68
69        $stmt->execute();
70
71        return (int)$this->pdo->lastInsertId();
72    }
73
74    /**
75     * {@inheritdoc}
76     */
77    public function fetchNextPending(DateTimeImmutable $now): ?QueueJob
78    {
79        $tableName = $this->tablePrefix . 'mod_queue_records';
80
81        $driver = '';
82        try {
83            $driver = (string) $this->pdo->getAttribute(PDO::ATTR_DRIVER_NAME);
84        } catch (\Throwable) {
85            // Unmocked driver attribute in unit tests
86        }
87        $isMySql = str_contains(strtolower($driver), 'mysql');
88        $skipLocked = $isMySql ? ' FOR UPDATE SKIP LOCKED' : '';
89
90        $ownsTransaction = false;
91        if ($isMySql && !$this->pdo->inTransaction()) {
92            $this->pdo->beginTransaction();
93            $ownsTransaction = true;
94        }
95
96        try {
97            // Find oldest pending job with optional concurrency skip-locked
98            $selectSql = "SELECT id, job_type, label, status, payload, progress_percent,
99                                 total_items, processed_items, attempts, max_attempts,
100                                 heartbeat_at, started_at, completed_at, failed_at,
101                                 error_message, output_log, created_at, updated_at,
102                                 created_by, owner
103                          FROM {$tableName}
104                          WHERE status = :pending
105                          ORDER BY id ASC
106                          LIMIT 1{$skipLocked}";
107
108            $stmt = $this->pdo->prepare($selectSql);
109            $stmt->bindValue(self::PARAM_PENDING, QueueJob::STATUS_PENDING);
110            $stmt->execute();
111
112            /** @var array<string, mixed>|false $row */
113            $row = $stmt->fetch(PDO::FETCH_ASSOC);
114            if ($row === false) {
115                if ($ownsTransaction && $this->pdo->inTransaction()) {
116                    $this->pdo->commit();
117                }
118                return null;
119            }
120
121            $jobId = (int)$row['id'];
122            $nowStr = $now->format(self::DATE_FORMAT);
123
124            // Atomic lock transition from pending to running
125            $updateSql = "UPDATE {$tableName}
126                          SET status = :running,
127                              started_at = :started_at,
128                              heartbeat_at = :heartbeat_at,
129                              attempts = attempts + 1
130                          WHERE id = :id AND status = :pending";
131
132            $updateStmt = $this->pdo->prepare($updateSql);
133            $updateStmt->bindValue(':running', QueueJob::STATUS_RUNNING);
134            $updateStmt->bindValue(':started_at', $nowStr);
135            $updateStmt->bindValue(':heartbeat_at', $nowStr);
136            $updateStmt->bindValue(':id', $jobId, PDO::PARAM_INT);
137            $updateStmt->bindValue(self::PARAM_PENDING, QueueJob::STATUS_PENDING);
138            $updateStmt->execute();
139
140            if ($ownsTransaction && $this->pdo->inTransaction()) {
141                $this->pdo->commit();
142            }
143
144            if ($updateStmt->rowCount() === 0) {
145                return null; // Another worker locked this job concurrently
146            }
147
148            $row['status'] = QueueJob::STATUS_RUNNING;
149            $row['started_at'] = $nowStr;
150            $row['heartbeat_at'] = $nowStr;
151            $row['attempts'] = (int)$row['attempts'] + 1;
152
153            return $this->hydrateJob($row);
154        } catch (\Throwable $e) {
155            if ($ownsTransaction && $this->pdo->inTransaction()) {
156                $this->pdo->rollBack();
157            }
158            throw $e;
159        }
160    }
161
162    /**
163     * {@inheritdoc}
164     */
165    public function recoverStaleJobs(int $staleSeconds, DateTimeImmutable $now): int
166    {
167        $tableName = $this->tablePrefix . 'mod_queue_records';
168        $cutoff = $now->modify("-{$staleSeconds} seconds");
169        $cutoffStr = $cutoff->format(self::DATE_FORMAT);
170        $nowStr = $now->format(self::DATE_FORMAT);
171
172        // Find running jobs with stale heartbeat
173        $sql = "SELECT id, attempts, max_attempts FROM {$tableName}
174                WHERE status = :running
175                  AND (heartbeat_at IS NULL OR heartbeat_at < :cutoff)";
176
177        $stmt = $this->pdo->prepare($sql);
178        $stmt->bindValue(':running', QueueJob::STATUS_RUNNING);
179        $stmt->bindValue(':cutoff', $cutoffStr);
180        $stmt->execute();
181
182        /** @var array<int, array{id: int|string, attempts: int|string, max_attempts: int|string}> $staleRows */
183        $staleRows = $stmt->fetchAll(PDO::FETCH_ASSOC);
184        $recoveredCount = 0;
185
186        foreach ($staleRows as $staleRow) {
187            $id = (int)$staleRow['id'];
188            $attempts = (int)$staleRow['attempts'];
189            $maxAttempts = (int)$staleRow['max_attempts'];
190
191            if ($attempts >= $maxAttempts) {
192                // Exhausted retries -> mark as failed
193                $failSql = "UPDATE {$tableName}
194                            SET status = :failed,
195                                failed_at = :failed_at,
196                                error_message = :err
197                            WHERE id = :id";
198                $failStmt = $this->pdo->prepare($failSql);
199                $failStmt->bindValue(':failed', QueueJob::STATUS_FAILED);
200                $failStmt->bindValue(':failed_at', $nowStr);
201                $failStmt->bindValue(':err', 'Task heartbeat timed out and exceeded max retry attempts.');
202                $failStmt->bindValue(':id', $id, PDO::PARAM_INT);
203                $failStmt->execute();
204            } else {
205                // Retryable -> reset back to pending
206                $retrySql = "UPDATE {$tableName}
207                             SET status = :pending,
208                                 started_at = NULL,
209                                 heartbeat_at = NULL,
210                                 error_message = :err
211                             WHERE id = :id";
212                $retryStmt = $this->pdo->prepare($retrySql);
213                $retryStmt->bindValue(':pending', QueueJob::STATUS_PENDING);
214                $retryStmt->bindValue(':err', 'Recovered from stale running state (heartbeat timeout).');
215                $retryStmt->bindValue(':id', $id, PDO::PARAM_INT);
216                $retryStmt->execute();
217            }
218            $recoveredCount++;
219        }
220
221        return $recoveredCount;
222    }
223
224    /**
225     * {@inheritdoc}
226     */
227    public function updateProgress(int $jobId, int $processedItems, int $totalItems, DateTimeImmutable $now): void
228    {
229        $tableName = $this->tablePrefix . 'mod_queue_records';
230        $progress = $totalItems > 0 ? (int)min(100, round(($processedItems / $totalItems) * 100)) : 100;
231        $nowStr = $now->format(self::DATE_FORMAT);
232
233        $sql = "UPDATE {$tableName}
234                SET processed_items = :processed,
235                    total_items = :total,
236                    progress_percent = :progress,
237                    heartbeat_at = :heartbeat_at
238                WHERE id = :id";
239
240        $stmt = $this->pdo->prepare($sql);
241        $stmt->bindValue(':processed', $processedItems, PDO::PARAM_INT);
242        $stmt->bindValue(':total', $totalItems, PDO::PARAM_INT);
243        $stmt->bindValue(':progress', $progress, PDO::PARAM_INT);
244        $stmt->bindValue(':heartbeat_at', $nowStr);
245        $stmt->bindValue(':id', $jobId, PDO::PARAM_INT);
246        $stmt->execute();
247    }
248
249    /**
250     * {@inheritdoc}
251     */
252    public function markCompleted(int $jobId, DateTimeImmutable $now, ?string $outputLog = null): void
253    {
254        $tableName = $this->tablePrefix . 'mod_queue_records';
255        $nowStr = $now->format(self::DATE_FORMAT);
256
257        $sql = "UPDATE {$tableName}
258                SET status = :completed,
259                    completed_at = :completed_at,
260                    progress_percent = 100,
261                    output_log = :output_log
262                WHERE id = :id";
263
264        $stmt = $this->pdo->prepare($sql);
265        $stmt->bindValue(':completed', QueueJob::STATUS_COMPLETED);
266        $stmt->bindValue(':completed_at', $nowStr);
267        $stmt->bindValue(':output_log', $outputLog);
268        $stmt->bindValue(':id', $jobId, PDO::PARAM_INT);
269        $stmt->execute();
270    }
271
272    /**
273     * {@inheritdoc}
274     */
275    public function markFailed(int $jobId, string $errorMessage, DateTimeImmutable $now): void
276    {
277        $tableName = $this->tablePrefix . 'mod_queue_records';
278        $nowStr = $now->format(self::DATE_FORMAT);
279
280        $sql = "UPDATE {$tableName}
281                SET status = :failed,
282                    failed_at = :failed_at,
283                    error_message = :err
284                WHERE id = :id";
285
286        $stmt = $this->pdo->prepare($sql);
287        $stmt->bindValue(':failed', QueueJob::STATUS_FAILED);
288        $stmt->bindValue(':failed_at', $nowStr);
289        $stmt->bindValue(':err', $errorMessage);
290        $stmt->bindValue(':id', $jobId, PDO::PARAM_INT);
291        $stmt->execute();
292    }
293
294    /**
295     * {@inheritdoc}
296     */
297    public function findById(int $jobId): ?QueueJob
298    {
299        $tableName = $this->tablePrefix . 'mod_queue_records';
300        $sql = "SELECT id, job_type, label, status, payload, progress_percent,
301                       total_items, processed_items, attempts, max_attempts,
302                       heartbeat_at, started_at, completed_at, failed_at,
303                       error_message, output_log, created_at, updated_at,
304                       created_by, owner
305                FROM {$tableName}
306                WHERE id = :id";
307
308        $stmt = $this->pdo->prepare($sql);
309        $stmt->bindValue(':id', $jobId, PDO::PARAM_INT);
310        $stmt->execute();
311
312        /** @var array<string, mixed>|false $row */
313        $row = $stmt->fetch(PDO::FETCH_ASSOC);
314
315        return $row !== false ? $this->hydrateJob($row) : null;
316    }
317
318    /**
319     * Hydrates a QueueJob domain entity from a database associative array.
320     *
321     * @param array<string, mixed> $row Database row.
322     * @return QueueJob Hydrated entity.
323     */
324    private function hydrateJob(array $row): QueueJob
325    {
326        /** @var array<string, mixed> $payload */
327        $payload = is_string($row['payload']) ? (array)json_decode($row['payload'], true) : [];
328
329        return new QueueJob(
330            (int)$row['id'],
331            (string)$row['job_type'],
332            (string)$row['label'],
333            (string)$row['status'],
334            $payload,
335            (int)($row['progress_percent'] ?? 0),
336            (int)($row['total_items'] ?? 0),
337            (int)($row['processed_items'] ?? 0),
338            (int)($row['attempts'] ?? 0),
339            (int)($row['max_attempts'] ?? 3),
340            $this->parseDateTime($row['heartbeat_at'] ?? null),
341            $this->parseDateTime($row['started_at'] ?? null),
342            $this->parseDateTime($row['completed_at'] ?? null),
343            $this->parseDateTime($row['failed_at'] ?? null),
344            isset($row['error_message']) ? (string)$row['error_message'] : null,
345            isset($row['output_log']) ? (string)$row['output_log'] : null,
346            $this->parseDateTime($row['created_at'] ?? null),
347            $this->parseDateTime($row['updated_at'] ?? null),
348            (int)($row['created_by'] ?? 1),
349            (int)($row['owner'] ?? 1)
350        );
351    }
352
353    /**
354     * Parses nullable timestamp string into DateTimeImmutable.
355     *
356     * @param mixed $val Raw datetime string.
357     * @return DateTimeImmutable|null Parsed object or null.
358     */
359    private function parseDateTime(mixed $val): ?DateTimeImmutable
360    {
361        if (!is_string($val) || $val === '') {
362            return null;
363        }
364
365        return new DateTimeImmutable($val);
366    }
367}