Code Coverage
 
Lines
Functions and Methods
Classes and Traits
Total
100.00%
175 / 175
100.00%
9 / 9
 CRAP
100.00%
1 / 1
CronRunner
100.00%
174 / 174
100.00%
9 / 9
 39
100.00%
1 / 1
 __construct
100.00%
1 / 1
100.00%
1 / 1
 1
 run
100.00%
23 / 23
100.00%
1 / 1
 5
 fetchActiveJobs
100.00%
17 / 17
100.00%
1 / 1
 3
 handleStaleRunningJob
100.00%
7 / 7
100.00%
1 / 1
 5
 dispatchStaleJobUnlock
100.00%
13 / 13
100.00%
1 / 1
 3
 executeJob
100.00%
32 / 32
100.00%
1 / 1
 4
 acquireLock
100.00%
16 / 16
100.00%
1 / 1
 3
 releaseLock
100.00%
25 / 25
100.00%
1 / 1
 6
 writeLog
100.00%
40 / 40
100.00%
1 / 1
 9
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\Core\Cron;
8
9defined('AMMONLY_APP') || exit('Direct script access is forbidden.');
10
11use App\Core\Api\ApiClientInterface;
12use App\Core\Cron\Presentation\Api\CronApiController;
13use DateTimeImmutable;
14use PDO;
15use Throwable;
16
17/**
18 * Enterprise Background CRON Automation Engine.
19 *
20 * Dispatches due background tasks via REST API /api/v1/cron with timeout handling.
21 *
22 * @package App\Core\Cron
23 */
24final class CronRunner
25{
26    private const string DATETIME_FORMAT = 'Y-m-d H:i:s';
27
28    /**
29     * CronRunner constructor.
30     *
31     * @param CronApiController|ApiClientInterface|PDO $source API controller, client, or database connection.
32     * @param string $tablePrefix Database table prefix.
33     * @param CronExpressionEvaluator|null $evaluator Evaluator for standard 5-field cron syntax.
34     */
35    public function __construct(
36        private readonly CronApiController|ApiClientInterface|PDO $source,
37        private readonly string $tablePrefix = 'a_',
38        private ?CronExpressionEvaluator $evaluator = null
39    ) {
40        $this->evaluator = $evaluator ?? new CronExpressionEvaluator();
41    }
42
43    /**
44     * Executes all active and due scheduled tasks.
45     *
46     * @return array<int, array<string, mixed>> Execution summary.
47     */
48    public function run(): array
49    {
50        $jobs = $this->fetchActiveJobs();
51        $results = [];
52        $now = new DateTimeImmutable();
53        $nowTs = $now->getTimestamp();
54
55        foreach ($jobs as $job) {
56            $jobId = (int)$job['id'];
57            $name = (string)$job['name'];
58            $isRunning = (bool)$job['is_running'];
59            $lastRunAt = isset($job['last_run_at']) ? (string)$job['last_run_at'] : null;
60            $timeoutSeconds = (int)($job['timeout_seconds'] ?? 300);
61
62            if ($isRunning) {
63                $this->handleStaleRunningJob($jobId, $lastRunAt, $timeoutSeconds);
64                $results[] = [
65                    'job_id'      => $jobId,
66                    'name'        => $name,
67                    'status'      => false,
68                    'duration_ms' => 0,
69                ];
70                continue;
71            }
72
73            $expression = (string)$job['expression'];
74            if (!$this->evaluator->isDue($expression, $nowTs)) {
75                continue;
76            }
77
78            $results[] = $this->executeJob($job);
79        }
80
81        return $results;
82    }
83
84    /**
85     * Fetches active scheduled jobs from the configured source.
86     *
87     * @return array<int, array<string, mixed>> Active cron task records.
88     */
89    private function fetchActiveJobs(): array
90    {
91        if ($this->source instanceof CronApiController) {
92            $response = $this->source->tasks();
93            $body = (string)$response->getBody();
94            /** @var array{data?: array<int, array<string, mixed>>} $payload */
95            $payload = (array)json_decode($body, true);
96            return (array)($payload['data'] ?? []);
97        }
98
99        if ($this->source instanceof ApiClientInterface) {
100            $payload = $this->source->get('/api/v1/cron/tasks');
101            return (array)($payload['data'] ?? []);
102        }
103
104        $cronTable = $this->tablePrefix . 'mod_cron_records';
105        $sql = sprintf(
106            'SELECT `id`, `name`, `label`, `command_class`, `expression`, `timeout_seconds`, ' .
107            '`is_running`, `last_run_at` FROM `%s` WHERE `is_active` = 1',
108            $cronTable
109        );
110
111        $stmt = $this->source->prepare($sql);
112        $stmt->execute();
113
114        return $stmt->fetchAll(PDO::FETCH_ASSOC);
115    }
116
117    /**
118     * Handles timed out running job release and logging.
119     */
120    private function handleStaleRunningJob(int $jobId, ?string $lastRunAt, int $timeoutSeconds): void
121    {
122        if ($lastRunAt === null) {
123            return;
124        }
125
126        $lastRunTs = strtotime($lastRunAt);
127        $elapsed = $lastRunTs !== false ? (time() - $lastRunTs) : 0;
128        if ($elapsed <= $timeoutSeconds || $lastRunTs === false) {
129            return;
130        }
131
132        $this->dispatchStaleJobUnlock($jobId, $elapsed, $timeoutSeconds);
133    }
134
135    /**
136     * Dispatches unlock and logging for timed out stale job.
137     */
138    private function dispatchStaleJobUnlock(int $jobId, int $elapsed, int $timeoutSeconds): void
139    {
140        if ($this->source instanceof CronApiController) {
141            $factory = new \Nyholm\Psr7\Factory\Psr17Factory();
142            $req = $factory->createServerRequest('POST', '/api/v1/cron/unlock-timeout/' . $jobId)
143                ->withParsedBody(['elapsed' => $elapsed, 'timeout_seconds' => $timeoutSeconds]);
144            $this->source->unlockTimeout($req, $jobId);
145            return;
146        }
147
148        if ($this->source instanceof ApiClientInterface) {
149            $this->source->post('/api/v1/cron/unlock-timeout/' . $jobId, [
150                'elapsed'         => $elapsed,
151                'timeout_seconds' => $timeoutSeconds,
152            ]);
153            return;
154        }
155
156        CronTimeoutHandler::unlockJob($this->source, $this->tablePrefix, $jobId, $elapsed, $timeoutSeconds);
157    }
158
159    /**
160     * Executes a single cron task instance with timing, memory tracking, and error logging.
161     *
162     * @param array<string, mixed> $job Job configuration record.
163     * @return array<string, mixed> Execution result.
164     */
165    private function executeJob(array $job): array
166    {
167        $jobId = (int)$job['id'];
168        $name = (string)$job['name'];
169        $commandClass = (string)$job['command_class'];
170
171        $this->acquireLock($jobId);
172
173        $startMemory = memory_get_peak_usage(true);
174        $startTime = microtime(true);
175        $output = '';
176        $errorMessage = null;
177        $isSuccess = true;
178
179        try {
180            if (!class_exists($commandClass)) {
181                throw new \DomainException(sprintf('Command class "%s" not found.', $commandClass));
182            }
183
184            /** @var mixed $task */
185            $task = new $commandClass();
186            if (!$task instanceof CronTaskInterface) {
187                throw new \DomainException(sprintf(
188                    'Command class "%s" does not implement CronTaskInterface.',
189                    $commandClass
190                ));
191            }
192
193            $output = $task->run();
194        } catch (Throwable $e) {
195            $isSuccess = false;
196            $errorMessage = $e->getMessage() . "\n" . $e->getTraceAsString();
197        }
198
199        $durationMs = (int)round((microtime(true) - $startTime) * 1000);
200        $memoryPeak = memory_get_peak_usage(true) - $startMemory;
201
202        $this->releaseLock($jobId, $isSuccess, $durationMs);
203        $this->writeLog($jobId, $isSuccess, $durationMs, $memoryPeak, $output, $errorMessage);
204
205        return [
206            'job_id'      => $jobId,
207            'name'        => $name,
208            'status'      => $isSuccess,
209            'duration_ms' => $durationMs,
210            'output'      => $output,
211        ];
212    }
213
214    /**
215     * Acquires lock on cron job record.
216     */
217    private function acquireLock(int $jobId): void
218    {
219        if ($this->source instanceof CronApiController) {
220            $this->source->lock($jobId);
221            return;
222        }
223
224        if ($this->source instanceof ApiClientInterface) {
225            $this->source->post('/api/v1/cron/lock/' . $jobId);
226            return;
227        }
228
229        $cronTable = $this->tablePrefix . 'mod_cron_records';
230        $sql = sprintf(
231            'UPDATE `%s` SET `is_running` = 1, `last_run_at` = :now WHERE `id` = :id',
232            $cronTable
233        );
234        $stmt = $this->source->prepare($sql);
235        $stmt->execute([
236            ':now' => date(self::DATETIME_FORMAT),
237            ':id'  => $jobId,
238        ]);
239    }
240
241    /**
242     * Releases lock on cron job record.
243     */
244    private function releaseLock(int $jobId, bool $isSuccess, int $durationMs): void
245    {
246        if ($this->source instanceof CronApiController) {
247            $factory = new \Nyholm\Psr7\Factory\Psr17Factory();
248            $req = $factory->createServerRequest('POST', '/api/v1/cron/unlock/' . $jobId)
249                ->withParsedBody(['status' => $isSuccess ? 1 : 0, 'duration_ms' => $durationMs]);
250            $this->source->unlock($req, $jobId);
251            return;
252        }
253
254        if ($this->source instanceof ApiClientInterface) {
255            $this->source->post('/api/v1/cron/unlock/' . $jobId, [
256                'status'      => $isSuccess ? 1 : 0,
257                'duration_ms' => $durationMs,
258            ]);
259            return;
260        }
261
262        $cronTable = $this->tablePrefix . 'mod_cron_records';
263        $sql = sprintf(
264            'UPDATE `%s` SET `is_running` = 0, `last_status` = :status, `last_duration_ms` = :duration, ' .
265            '`updated_at` = :now WHERE `id` = :id',
266            $cronTable
267        );
268        $stmt = $this->source->prepare($sql);
269        $stmt->execute([
270            ':status'   => $isSuccess ? 1 : 0,
271            ':duration' => $durationMs,
272            ':now'      => date(self::DATETIME_FORMAT),
273            ':id'       => $jobId,
274        ]);
275    }
276
277    /**
278     * Writes execution log record.
279     */
280    private function writeLog(
281        int $jobId,
282        bool $isSuccess,
283        int $durationMs,
284        int $memoryPeak,
285        string $output,
286        ?string $errorMessage
287    ): void {
288        if ($this->source instanceof CronApiController) {
289            $factory = new \Nyholm\Psr7\Factory\Psr17Factory();
290            $req = $factory->createServerRequest('POST', '/api/v1/cron/log')
291                ->withParsedBody([
292                    'job_id'            => $jobId,
293                    'status'            => $isSuccess ? 1 : 0,
294                    'duration_ms'       => $durationMs,
295                    'memory_peak_bytes' => max(0, $memoryPeak),
296                    'output_log'        => $output !== '' ? $output : null,
297                    'error_message'     => $errorMessage,
298                ]);
299            $this->source->log($req);
300            return;
301        }
302
303        if ($this->source instanceof ApiClientInterface) {
304            $this->source->post('/api/v1/cron/log', [
305                'job_id'            => $jobId,
306                'status'            => $isSuccess ? 1 : 0,
307                'duration_ms'       => $durationMs,
308                'memory_peak_bytes' => max(0, $memoryPeak),
309                'output_log'        => $output !== '' ? $output : null,
310                'error_message'     => $errorMessage,
311            ]);
312            return;
313        }
314
315        $logTable = $this->tablePrefix . 'logs_cron_records';
316        $sql = sprintf(
317            'INSERT INTO `%s` (`job_id`, `status`, `duration_ms`, `memory_peak_bytes`, `output_log`, ' .
318            '`error_message`, `created_at`) ' .
319            'VALUES (:job_id, :status, :duration_ms, :mem_peak, :output, :error, :now)',
320            $logTable
321        );
322
323        $stmt = $this->source->prepare($sql);
324        $stmt->execute([
325            ':job_id'      => $jobId,
326            ':status'      => $isSuccess ? 1 : 0,
327            ':duration_ms' => $durationMs,
328            ':mem_peak'    => max(0, $memoryPeak),
329            ':output'      => $output !== '' ? $output : null,
330            ':error'       => $errorMessage,
331            ':now'         => date(self::DATETIME_FORMAT),
332        ]);
333    }
334}