Code Coverage
 
Lines
Functions and Methods
Classes and Traits
Total
90.48%
133 / 147
52.94%
9 / 17
 CRAP
0.00%
0 / 1
BulkActionJobHandler
90.41%
132 / 146
52.94%
9 / 17
 51.12
0.00%
0 / 1
 clearColumnsCache
0.00%
0 / 1
0.00%
0 / 1
 2
 __construct
100.00%
1 / 1
100.00%
1 / 1
 1
 supports
100.00%
1 / 1
100.00%
1 / 1
 1
 handle
96.97%
32 / 33
0.00%
0 / 1
 5
 resolveModuleTable
100.00%
9 / 9
100.00%
1 / 1
 4
 tableExists
75.00%
3 / 4
0.00%
0 / 1
 2.06
 hasColumn
77.78%
7 / 9
0.00%
0 / 1
 4.18
 hasColumnSqlite
100.00%
7 / 7
100.00%
1 / 1
 2
 hasColumnMysql
0.00%
0 / 6
0.00%
0 / 1
 2
 processChunk
85.71%
6 / 7
0.00%
0 / 1
 3.03
 executeBulkUpdate
94.74%
18 / 19
0.00%
0 / 1
 4.00
 resolveActionClauses
92.86%
13 / 14
0.00%
0 / 1
 8.02
 resolveArchiveClauses
100.00%
7 / 7
100.00%
1 / 1
 2
 resolveRestoreClauses
100.00%
7 / 7
100.00%
1 / 1
 2
 resolveDeleteClauses
100.00%
7 / 7
100.00%
1 / 1
 3
 resolveEditClauses
100.00%
6 / 6
100.00%
1 / 1
 4
 executeHardDelete
100.00%
8 / 8
100.00%
1 / 1
 2
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\Application\Handler;
8
9defined('AMMONLY_APP') || exit('Direct script access is forbidden.');
10
11use App\Core\Engine\Domain\Model\RecordStatus;
12use App\Modules\Automation\Queue\Domain\Model\QueueJob;
13use App\Modules\Automation\Queue\Domain\Repository\QueueRepositoryInterface;
14use DateTimeImmutable;
15use InvalidArgumentException;
16use PDO;
17use Throwable;
18
19/**
20 * Bulk Action Asynchronous Queue Handler.
21 *
22 * Executes background batch updates (mass edit, mass archive, mass restore, mass soft-delete)
23 * in safe chunks with heartbeats, progress reporting, and audit logging.
24 *
25 * @package App\Modules\Automation\Queue\Application\Handler
26 */
27final class BulkActionJobHandler implements JobHandlerInterface
28{
29    public const string JOB_TYPE = 'bulk_action';
30
31    /** @var array<string, array<string, bool>> In-memory cache of column existence per table. */
32    private static array $columnsCache = [];
33
34    /**
35     * Clears in-memory columns cache (primarily for tests).
36     */
37    public static function clearColumnsCache(): void
38    {
39        self::$columnsCache = [];
40    }
41
42    /**
43     * BulkActionJobHandler constructor.
44     *
45     * @param PDO    $pdo         Database connection.
46     * @param string $tablePrefix Database table prefix.
47     */
48    public function __construct(
49        private readonly PDO $pdo,
50        private readonly string $tablePrefix = 'a_'
51    ) {
52    }
53
54    /** {@inheritdoc} */
55    public function supports(): string
56    {
57        return self::JOB_TYPE;
58    }
59
60    /** {@inheritdoc} */
61    public function handle(QueueJob $job, QueueRepositoryInterface $queueRepository): string
62    {
63        $payload    = $job->payload;
64        $moduleName = (string) ($payload['module_name'] ?? '');
65        $actionType = (string) ($payload['action'] ?? 'bulk_edit');
66        $recordIds  = (array) ($payload['record_ids'] ?? []);
67        $updates    = (array) ($payload['updates'] ?? []);
68
69        if ($moduleName === '' || $recordIds === []) {
70            return 'No records or invalid module provided for bulk processing.';
71        }
72
73        $tableName = $this->resolveModuleTable($moduleName);
74        if ($tableName === null) {
75            return sprintf('Module table for "%s" could not be resolved.', $moduleName);
76        }
77
78        $totalRecords = count($recordIds);
79        $chunkSize = 100;
80        $chunks = array_chunk($recordIds, $chunkSize);
81        $processedRecords = 0;
82
83        $now = new DateTimeImmutable();
84        $queueRepository->updateProgress($job->id, 0, $totalRecords, $now);
85
86        foreach ($chunks as $chunk) {
87            $updatedInChunk = $this->processChunk(
88                $tableName,
89                $chunk,
90                $actionType,
91                $updates
92            );
93            $processedRecords += $updatedInChunk;
94
95            $now = new DateTimeImmutable();
96            $queueRepository->updateProgress($job->id, $processedRecords, $totalRecords, $now);
97        }
98
99        return sprintf(
100            'Successfully executed mass action "%s" on %d/%d records in module "%s".',
101            $actionType,
102            $processedRecords,
103            $totalRecords,
104            $moduleName
105        );
106    }
107
108    /**
109     * Resolves physical table name for the given module name.
110     *
111     * @param string $moduleName Module machine name.
112     * @return string|null Physical table name or null.
113     */
114    private function resolveModuleTable(string $moduleName): ?string
115    {
116        $modTable = $this->tablePrefix . 'core_module_records';
117        $sql = "SELECT `table_name` FROM `{$modTable}` WHERE `name` = :name AND `is_active` = 1 LIMIT 1";
118        $stmt = $this->pdo->prepare($sql);
119        $stmt->execute([':name' => $moduleName]);
120        $row = $stmt->fetch(PDO::FETCH_ASSOC);
121
122        if ($row && !empty($row['table_name'])) {
123            return (string) $row['table_name'];
124        }
125
126        $guessed = $this->tablePrefix . 'mod_' . $moduleName . '_records';
127        return $this->tableExists($guessed) ? $guessed : null;
128    }
129
130    /**
131     * Checks if a table exists in the active database.
132     *
133     * @param string $tableName Table name.
134     * @return bool True if exists.
135     */
136    private function tableExists(string $tableName): bool
137    {
138        try {
139            $stmt = $this->pdo->query("SHOW TABLES LIKE " . $this->pdo->quote($tableName));
140            return (bool) $stmt->fetchColumn();
141        } catch (Throwable) {
142            return false;
143        }
144    }
145
146    /**
147     * Checks if a column exists in the given physical table.
148     *
149     * @param string $tableName  Target table name.
150     * @param string $columnName Target column name.
151     * @return bool True if column exists.
152     */
153    private function hasColumn(string $tableName, string $columnName): bool
154    {
155        $normalizedCol = strtolower($columnName);
156        if (isset(self::$columnsCache[$tableName][$normalizedCol])) {
157            return self::$columnsCache[$tableName][$normalizedCol];
158        }
159
160        try {
161            $driver = (string) $this->pdo->getAttribute(PDO::ATTR_DRIVER_NAME);
162            return $driver === 'sqlite'
163                ? $this->hasColumnSqlite($tableName, $normalizedCol)
164                : $this->hasColumnMysql($tableName, $columnName, $normalizedCol);
165        } catch (Throwable) {
166            return false;
167        }
168    }
169
170    private function hasColumnSqlite(string $tableName, string $normalizedCol): bool
171    {
172        $stmt = $this->pdo->prepare(sprintf('PRAGMA table_info(`%s`)', $tableName));
173        $stmt->execute();
174        /** @var array<int, array<string, mixed>> $cols */
175        $cols = $stmt->fetchAll(PDO::FETCH_ASSOC);
176        foreach ($cols as $col) {
177            $cName = strtolower((string) ($col['name'] ?? ''));
178            self::$columnsCache[$tableName][$cName] = true;
179        }
180        return self::$columnsCache[$tableName][$normalizedCol] ?? false;
181    }
182
183    private function hasColumnMysql(string $tableName, string $columnName, string $normalizedCol): bool
184    {
185        $sql = sprintf('SHOW COLUMNS FROM `%s` LIKE ?', $tableName);
186        $stmt = $this->pdo->prepare($sql);
187        $stmt->execute([$columnName]);
188        $exists = (bool) $stmt->fetchColumn();
189        self::$columnsCache[$tableName][$normalizedCol] = $exists;
190        return $exists;
191    }
192
193    /**
194     * Processes a single chunk of record IDs.
195     *
196     * @param string                 $tableName  Target table name.
197     * @param array<int, int|string> $chunk      List of record IDs.
198     * @param string                 $actionType Action type.
199     * @param array<string, mixed>   $updates    Field updates map.
200     * @return int Number of updated records.
201     */
202    private function processChunk(
203        string $tableName,
204        array  $chunk,
205        string $actionType,
206        array  $updates,
207    ): int {
208        if ($chunk === []) {
209            return 0;
210        }
211
212        $params = [];
213        $setClauses = $this->resolveActionClauses($tableName, $actionType, $updates, $params);
214        if ($setClauses === null) {
215            return $this->executeHardDelete($tableName, $chunk);
216        }
217
218        return $this->executeBulkUpdate($tableName, $chunk, $setClauses, $params);
219    }
220
221    /**
222     * Executes bulk update for a chunk of record IDs.
223     *
224     * @param string                 $tableName  Target table name.
225     * @param array<int, int|string> $chunk      List of record IDs.
226     * @param array<int, string>     $setClauses List of SET clause fragments.
227     * @param array<int, mixed>      $params     Query parameter bindings.
228     * @return int Number of updated records.
229     */
230    private function executeBulkUpdate(
231        string $tableName,
232        array  $chunk,
233        array  $setClauses,
234        array  $params,
235    ): int {
236        if ($setClauses === []) {
237            return 0;
238        }
239
240        if ($this->hasColumn($tableName, 'updated_at')) {
241            $setClauses[] = '`updated_at` = ?';
242            $params[] = (new DateTimeImmutable())->format('Y-m-d H:i:s');
243        }
244
245        $placeholders = implode(',', array_fill(0, count($chunk), '?'));
246        $extraWhere   = '';
247        if ($this->hasColumn($tableName, 'is_system')) {
248            $extraWhere .= ' AND `is_system` != 1';
249        }
250
251        $sql = sprintf(
252            'UPDATE `%s` SET %s WHERE `id` IN (%s)%s',
253            $tableName,
254            implode(', ', $setClauses),
255            $placeholders,
256            $extraWhere
257        );
258
259        $stmt = $this->pdo->prepare($sql);
260        $stmt->execute([...$params, ...$chunk]);
261
262        return $stmt->rowCount();
263    }
264
265    /**
266     * Resolves SET clauses and query parameters based on action type.
267     *
268     * @param string               $tableName
269     * @param string               $actionType
270     * @param array<string, mixed> $updates
271     * @param array<int, mixed>    $params
272     * @param-out array<int, mixed> $params
273     * @return array<int, string>|null Returns null if hard delete should be executed.
274     */
275    private function resolveActionClauses(
276        string $tableName,
277        string $actionType,
278        array  $updates,
279        array  &$params,
280    ): ?array {
281        $accessCol = null;
282        if ($this->hasColumn($tableName, 'special_access')) {
283            $accessCol = 'special_access';
284        } elseif ($this->hasColumn($tableName, 'record_status')) {
285            $accessCol = 'record_status';
286        }
287
288        return match ($actionType) {
289            'bulk_archive' => $this->resolveArchiveClauses($tableName, $accessCol, $params),
290            'bulk_restore' => $this->resolveRestoreClauses($tableName, $accessCol, $params),
291            'bulk_delete'  => $this->resolveDeleteClauses($tableName, $accessCol, $params),
292            'bulk_edit'    => $this->resolveEditClauses($tableName, $updates, $params),
293            default        => throw new InvalidArgumentException(
294                sprintf('Unsupported bulk action type "%s"', $actionType)
295            ),
296        };
297    }
298
299    /**
300     * @param array<int, mixed> &$params
301     * @param-out array<int, mixed> $params
302     * @return array<int, string>
303     */
304    private function resolveArchiveClauses(string $tableName, ?string $accessCol, array &$params): array
305    {
306        if ($accessCol === null) {
307            throw new InvalidArgumentException(sprintf(
308                'Table "%s" does not support record statuses (missing special_access or record_status column).',
309                $tableName
310            ));
311        }
312        $params[] = RecordStatus::ARCHIVED;
313        return ["`{$accessCol}` = ?"];
314    }
315
316    /**
317     * @param array<int, mixed> &$params
318     * @param-out array<int, mixed> $params
319     * @return array<int, string>
320     */
321    private function resolveRestoreClauses(string $tableName, ?string $accessCol, array &$params): array
322    {
323        if ($accessCol === null) {
324            throw new InvalidArgumentException(sprintf(
325                'Table "%s" does not support record statuses (missing special_access or record_status column).',
326                $tableName
327            ));
328        }
329        $params[] = RecordStatus::AVAILABLE;
330        return ["`{$accessCol}` = ?"];
331    }
332
333    /**
334     * @param array<int, mixed> &$params
335     * @param-out array<int, mixed> $params
336     * @return array<int, string>|null
337     */
338    private function resolveDeleteClauses(string $tableName, ?string $accessCol, array &$params): ?array
339    {
340        if ($accessCol !== null) {
341            $params[] = RecordStatus::DELETED;
342            return ["`{$accessCol}` = ?"];
343        }
344
345        if ($this->hasColumn($tableName, 'is_active')) {
346            $params[] = 0;
347            return ['`is_active` = ?'];
348        }
349
350        return null;
351    }
352
353    /**
354     * @param array<string, mixed>  $updates
355     * @param array<int, mixed>     &$params
356     * @param-out array<int, mixed> $params
357     * @return array<int, string>
358     */
359    private function resolveEditClauses(string $tableName, array $updates, array &$params): array
360    {
361        $clauses = [];
362        foreach ($updates as $column => $value) {
363            if (preg_match('/^\w+$/', $column) && $this->hasColumn($tableName, $column)) {
364                $clauses[] = "`{$column}` = ?";
365                $params[] = $value;
366            }
367        }
368        return $clauses;
369    }
370
371    private function executeHardDelete(string $tableName, array $chunk): int
372    {
373        $placeholders = implode(',', array_fill(0, count($chunk), '?'));
374        $extraWhere   = '';
375        if ($this->hasColumn($tableName, 'is_system')) {
376            $extraWhere .= ' AND `is_system` != 1';
377        }
378
379        $sql = sprintf('DELETE FROM `%s` WHERE `id` IN (%s)%s', $tableName, $placeholders, $extraWhere);
380        $stmt = $this->pdo->prepare($sql);
381        $stmt->execute($chunk);
382        return $stmt->rowCount();
383    }
384}