132 lines
4.5 KiB
PHP
132 lines
4.5 KiB
PHP
<?php
|
|
/**
|
|
* Cron Worker for AI Invoice Extraction
|
|
*
|
|
* Designed to run via cron every minute: * * * * *
|
|
* Processes ALL pending items in the queue, then EXITS.
|
|
* NO infinite loop. NO lock file issues.
|
|
*/
|
|
|
|
declare(strict_types=1);
|
|
|
|
require_once __DIR__ . '/../bootstrap/init.php';
|
|
|
|
use App\Core\Database;
|
|
use App\Services\InvoiceProcessor;
|
|
|
|
// Simple lock: prevent overlapping runs
|
|
$lockFile = STORAGE_PATH . '/logs/process_batches.lock';
|
|
|
|
// Check if lock file exists and is stale (older than 5 minutes = dead process)
|
|
if (file_exists($lockFile)) {
|
|
$lockAge = time() - filemtime($lockFile);
|
|
if ($lockAge > 300) {
|
|
// Stale lock from a crashed process - remove it
|
|
@unlink($lockFile);
|
|
workerLog("Removed stale lock file (age: {$lockAge}s)");
|
|
} else {
|
|
workerLog("Worker already running (lock age: {$lockAge}s). Exiting.");
|
|
exit(0);
|
|
}
|
|
}
|
|
|
|
// Create lock
|
|
file_put_contents($lockFile, getmypid() . "\n" . date('c'));
|
|
|
|
function workerLog(string $msg): void {
|
|
$line = "[" . date('Y-m-d H:i:s') . "] " . $msg . "\n";
|
|
echo $line;
|
|
// Also write to dedicated log file
|
|
@file_put_contents(STORAGE_PATH . '/logs/worker.log', $line, FILE_APPEND);
|
|
}
|
|
|
|
workerLog("=== Musadaq AI Worker Started ===");
|
|
|
|
try {
|
|
$db = Database::getInstance();
|
|
$processed = 0;
|
|
$failed = 0;
|
|
|
|
// Get pending items whose batch is ready for processing.
|
|
//
|
|
// The b.status <> 'uploading' filter is essential: queue rows are inserted as
|
|
// 'pending' the moment each image lands, while total_images is still growing.
|
|
// Picking them up mid-upload made the batch look complete after the first
|
|
// image, which locked the batch and made every later upload fail with
|
|
// "لا يمكن إضافة صور لدفعة تمت معالجتها".
|
|
$stmt = $db->prepare("
|
|
SELECT q.id
|
|
FROM invoice_processing_queue q
|
|
JOIN invoice_batches b ON q.batch_id = b.id
|
|
WHERE q.status = 'pending'
|
|
AND q.attempts < COALESCE(q.max_attempts, 3)
|
|
AND b.status <> 'uploading'
|
|
ORDER BY q.created_at ASC
|
|
LIMIT 20
|
|
");
|
|
$stmt->execute();
|
|
$items = $stmt->fetchAll(\PDO::FETCH_COLUMN);
|
|
|
|
// Rescue items abandoned in 'processing' by a crashed worker. claimed_at is
|
|
// set at claim time, so "claimed over 10 minutes ago and still processing"
|
|
// reliably means the owning process died mid-flight.
|
|
$rescued = $db->prepare("
|
|
UPDATE invoice_processing_queue
|
|
SET status = 'pending',
|
|
error_message = 'Reclaimed from a stalled worker'
|
|
WHERE status = 'processing'
|
|
AND claimed_at IS NOT NULL
|
|
AND claimed_at < (NOW() - INTERVAL 10 MINUTE)
|
|
");
|
|
$rescued->execute();
|
|
if ($rescued->rowCount() > 0) {
|
|
workerLog("Reclaimed {$rescued->rowCount()} stalled item(s) back to pending.");
|
|
}
|
|
|
|
// Also close out batches whose images all reached a terminal state but whose
|
|
// status was never flipped (e.g. the worker died right after the last item).
|
|
$stuck = $db->query("
|
|
SELECT id FROM invoice_batches
|
|
WHERE status = 'processing'
|
|
AND total_images > 0
|
|
AND (processed_images + failed_images) >= total_images
|
|
LIMIT 50
|
|
")->fetchAll(\PDO::FETCH_COLUMN);
|
|
foreach ($stuck as $stuckBatchId) {
|
|
workerLog("Closing stuck batch: $stuckBatchId");
|
|
InvoiceProcessor::checkBatchCompletion((string)$stuckBatchId);
|
|
}
|
|
|
|
if (empty($items)) {
|
|
workerLog("No pending items. Exiting.");
|
|
} else {
|
|
workerLog("Found " . count($items) . " pending item(s).");
|
|
|
|
foreach ($items as $queueId) {
|
|
workerLog("Processing Queue ID: $queueId ...");
|
|
|
|
try {
|
|
$success = InvoiceProcessor::processQueueItem((int)$queueId);
|
|
if ($success) {
|
|
$processed++;
|
|
workerLog(" ✓ Queue ID $queueId processed successfully.");
|
|
} else {
|
|
$failed++;
|
|
workerLog(" ✗ Queue ID $queueId failed (returned false).");
|
|
}
|
|
} catch (\Throwable $e) {
|
|
$failed++;
|
|
workerLog(" ✗ Queue ID $queueId EXCEPTION: " . $e->getMessage());
|
|
}
|
|
}
|
|
|
|
workerLog("=== Worker Done: $processed success, $failed failed ===");
|
|
}
|
|
|
|
} catch (\Throwable $e) {
|
|
workerLog("FATAL ERROR: " . $e->getMessage() . "\n" . $e->getTraceAsString());
|
|
} finally {
|
|
// ALWAYS remove lock file
|
|
@unlink($lockFile);
|
|
}
|