Code Coverage
 
Lines
Functions and Methods
Classes and Traits
Total
86.84%
66 / 76
60.00%
3 / 5
 CRAP
0.00%
0 / 1
ImportQueueJobHandler
86.67%
65 / 75
60.00%
3 / 5
 11.29
0.00%
0 / 1
 __construct
100.00%
1 / 1
100.00%
1 / 1
 1
 supports
100.00%
1 / 1
100.00%
1 / 1
 1
 handle
95.00%
38 / 40
0.00%
0 / 1
 5
 updateImportJobProgress
0.00%
0 / 8
0.00%
0 / 1
 2
 updateImportJobStatus
100.00%
25 / 25
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\Core\DataExchange\Application\Handler;
8
9defined('AMMONLY_APP') || exit('Direct script access is forbidden.');
10
11use App\Core\DataExchange\Application\Service\DataImportServiceInterface;
12use App\Core\DataExchange\Domain\Model\ImportMappingConfig;
13use App\Core\DataExchange\Domain\Model\ImportResultDto;
14use App\Core\Engine\Domain\Model\PermissionContext;
15use App\Modules\Automation\Queue\Application\Handler\JobHandlerInterface;
16use App\Modules\Automation\Queue\Domain\Model\QueueJob;
17use App\Modules\Automation\Queue\Domain\Repository\QueueRepositoryInterface;
18use InvalidArgumentException;
19use PDO;
20use Throwable;
21
22/**
23 * Asynchronous job handler executing background batch data imports.
24 *
25 * @package App\Core\DataExchange\Application\Handler
26 */
27final class ImportQueueJobHandler implements JobHandlerInterface
28{
29    public const string JOB_TYPE = 'data_import';
30
31    /**
32     * ImportQueueJobHandler constructor.
33     *
34     * @param DataImportServiceInterface $importService Data import application service.
35     * @param PDO                        $pdo           Database connection.
36     * @param string                     $tablePrefix   Database table prefix.
37     */
38    public function __construct(
39        private readonly DataImportServiceInterface $importService,
40        private readonly PDO $pdo,
41        private readonly string $tablePrefix = 'a_'
42    ) {
43    }
44
45    /** {@inheritdoc} */
46    public function supports(): string
47    {
48        return self::JOB_TYPE;
49    }
50
51    /** {@inheritdoc} */
52    public function handle(QueueJob $job, QueueRepositoryInterface $queueRepository): string
53    {
54        $payload = $job->payload;
55        $importJobId = (int) ($payload['import_job_id'] ?? 0);
56        $moduleName = (string) ($payload['module_name'] ?? '');
57        $filePath = (string) ($payload['file_path'] ?? '');
58
59        if ($importJobId === 0 || $moduleName === '' || !file_exists($filePath)) {
60            throw new InvalidArgumentException('Invalid payload for background data import.');
61        }
62
63        $mappingConfig = ImportMappingConfig::fromArray((array) ($payload['mapping_config'] ?? []));
64        $context = new PermissionContext(
65            actorUserId: (int) ($payload['user_id'] ?? 1),
66            actorIp: '127.0.0.1',
67            isSuperuser: (bool) ($payload['is_super_admin'] ?? false),
68            ownerScopeEnabled: false,
69            actorProfileId: (int) ($payload['profile_id'] ?? 1)
70        );
71
72        $this->updateImportJobStatus($importJobId, 'processing');
73
74        try {
75            $result = $this->importService->executeImport(
76                $moduleName,
77                $filePath,
78                $mappingConfig,
79                $context,
80                function (int $processed, int $total) use ($importJobId, $queueRepository, $job): void {
81                    $this->updateImportJobProgress($importJobId, $processed);
82                    $queueRepository->updateProgress($job->id, $processed, $total);
83                }
84            );
85
86            $this->updateImportJobStatus($importJobId, 'completed', $result);
87
88            return sprintf(
89                'Import completed for [%s]. Total: %d, Imported: %d, Updated: %d, Skipped: %d, Failed: %d.',
90                $moduleName,
91                $result->totalRows,
92                $result->importedRows,
93                $result->updatedRows,
94                $result->skippedRows,
95                $result->failedRows
96            );
97        } catch (Throwable $e) {
98            $this->updateImportJobStatus($importJobId, 'failed', null, [
99                ['row' => 0, 'message' => $e->getMessage()],
100            ]);
101            throw $e;
102        }
103    }
104
105    /**
106     * Updates processed rows counter for live progress bar polling.
107     */
108    private function updateImportJobProgress(int $importJobId, int $processed): void
109    {
110        $sql = "UPDATE {$this->tablePrefix}mod_import_jobs_records " .
111            'SET processed_rows = :processed, updated_at = :updated_at WHERE id = :id';
112        $stmt = $this->pdo->prepare($sql);
113        $stmt->execute([
114            ':processed' => $processed,
115            ':updated_at' => date('Y-m-d H:i:s'),
116            ':id' => $importJobId,
117        ]);
118    }
119
120    /**
121     * Updates the full metrics and completion status of the import job record.
122     *
123     * @param list<array{row: int, message: string}> $fallbackErrors
124     */
125    private function updateImportJobStatus(
126        int $importJobId,
127        string $status,
128        ?ImportResultDto $result = null,
129        array $fallbackErrors = []
130    ): void {
131        $total = $result?->totalRows ?? 0;
132        $imported = $result?->importedRows ?? 0;
133        $updated = $result?->updatedRows ?? 0;
134        $skipped = $result?->skippedRows ?? 0;
135        $failed = $result?->failedRows ?? 0;
136        $errors = $result !== null ? $result->errors : $fallbackErrors;
137        $errorSummary = !empty($errors) ? (string) json_encode(array_slice($errors, 0, 50)) : null;
138
139        $sql = "UPDATE {$this->tablePrefix}mod_import_jobs_records SET " .
140            'status = :status, total_rows = :total, processed_rows = :processed, ' .
141            'imported_rows = :imported, updated_rows = :updated, skipped_rows = :skipped, ' .
142            'failed_rows = :failed, error_summary = :error_summary, updated_at = :updated_at ' .
143            'WHERE id = :id';
144
145        $stmt = $this->pdo->prepare($sql);
146        $stmt->execute([
147            ':status' => $status,
148            ':total' => $total,
149            ':processed' => $total,
150            ':imported' => $imported,
151            ':updated' => $updated,
152            ':skipped' => $skipped,
153            ':failed' => $failed,
154            ':error_summary' => $errorSummary,
155            ':updated_at' => date('Y-m-d H:i:s'),
156            ':id' => $importJobId,
157        ]);
158    }
159}