Hamza/musadaq-saas
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Update: 2026-05-08 15:02:13

Browse Source
This commit is contained in:
Hamza-Ayed
2026-05-08 15:02:13 +03:00
parent 80949e584c
commit 7ea42f0f3b
4 changed files with 430 additions and 156 deletions
Download Patch File Download Diff File Expand all files Collapse all files

104
app/Services/InvoiceProcessor.php
View File

@@ -7,28 +7,46 @@ use App\Core\Database;
use App\Core\AI;
use App\Core\Encryption;
use App\Middleware\QuotaMiddleware;
use App\Services\NotificationService;
use Exception;
class InvoiceProcessor
{
private static function log(string $msg): void
{
$line = "[" . date('Y-m-d H:i:s') . "] [InvoiceProcessor] " . $msg . "\n";
@file_put_contents(STORAGE_PATH . '/logs/worker.log', $line, FILE_APPEND);
// Also echo for CLI/terminal usage
if (php_sapi_name() === 'cli') {
echo $line;
}
}
/**
* Processes a single invoice queue item by its ID.
*/
public static function processQueueItem(int $queueId): bool
{
$db = Database::getInstance();
self::log("Starting processQueueItem($queueId)");
try {
$db = Database::getInstance();
} catch (\Throwable $e) {
self::log("FATAL: Cannot connect to DB: " . $e->getMessage());
return false;
}
try {
// Fetch the queue item and its batch info
$stmt = $db->prepare("
SELECT q.*, b.tenant_id, b.company_id, b.uploaded_by, b.total_images
FROM invoice_processing_queue q
JOIN invoice_batches b ON q.batch_id = b.id COLLATE utf8mb4_unicode_ci
JOIN invoice_batches b ON q.batch_id = b.id
WHERE q.id = ? AND q.status = 'pending'
");
$stmt->execute([$queueId]);
$item = $stmt->fetch();
if (!$item) {
self::log("Queue ID $queueId: Not found or not pending. Skipping.");
return false;
}
@@ -38,27 +56,38 @@ class InvoiceProcessor
$userId = $item['uploaded_by'];
$imagePath = $item['image_path'];
self::log("Queue ID $queueId: Image=$imagePath, Batch=$batchId");
// Mark as processing
$db->prepare("UPDATE invoice_processing_queue SET status = 'processing' WHERE id = ?")->execute([$queueId]);
// Check file exists
if (!file_exists($imagePath)) {
self::log("Queue ID $queueId: FILE NOT FOUND: $imagePath");
$db->prepare("UPDATE invoice_processing_queue SET status = 'failed', error_message = 'File not found' WHERE id = ?")->execute([$queueId]);
return false;
}
self::log("Queue ID $queueId: File exists (" . filesize($imagePath) . " bytes). Starting AI extraction...");
$mimeType = mime_content_type($imagePath) ?: 'image/jpeg';
$fileContent = file_get_contents($imagePath);
$base64Data = base64_encode($fileContent);
// AI Extraction (this takes ~5-15 seconds)
$extracted = AI::extractInvoiceData($base64Data, $mimeType);
if (!$extracted) {
$db->prepare("UPDATE invoice_processing_queue SET status = 'failed', error_message = 'AI failed to extract' WHERE id = ?")->execute([$queueId]);
self::log("Queue ID $queueId: AI extraction returned NULL (failed).");
$db->prepare("UPDATE invoice_processing_queue SET status = 'failed', error_message = 'AI failed to extract data from image' WHERE id = ?")->execute([$queueId]);
return false;
}
try {
self::log("Queue ID $queueId: AI extraction successful. Saving to DB...");
// Save to database in a transaction
$db->beginTransaction();
try {
$invoiceId = vsprintf('%s%s-%s-%s-%s-%s%s%s', str_split(bin2hex(random_bytes(16)), 4));
$supplierTin = $extracted['supplier']['tin'] ?? '';
@@ -90,7 +119,7 @@ class InvoiceProcessor
$extracted['subtotal'] ?? 0, $extracted['tax_amount'] ?? 0, $extracted['discount_total'] ?? 0, $extracted['grand_total'] ?? 0, $extracted['currency_code'] ?? 'JOD'
]);
// Save Lines
// Save invoice line items
if (!empty($extracted['lines'])) {
$lineStmt = $db->prepare("INSERT INTO invoice_lines (id, invoice_id, line_number, description, quantity, unit_price, tax_rate, line_total) VALUES (?,?,?,?,?,?,?,?)");
foreach ($extracted['lines'] as $idx => $line) {
@@ -99,16 +128,34 @@ class InvoiceProcessor
$invoiceId, $line['line_number'] ?? ($idx + 1), $line['description'] ?? '', $line['quantity'] ?? 1, $line['unit_price'] ?? 0, $line['tax_rate'] ?? 0, $line['line_total'] ?? $line['total_amount'] ?? 0
]);
}
self::log("Queue ID $queueId: Saved " . count($extracted['lines']) . " line items.");
}
// Mark queue item done
$db->prepare("UPDATE invoice_processing_queue SET status = 'done', invoice_id = ?, processed_at = NOW() WHERE id = ?")->execute([$invoiceId, $queueId]);
// Update batch progress
$db->prepare("UPDATE invoice_batches SET processed_images = processed_images + 1 WHERE id = ?")->execute([$batchId]);
// Increment quota
QuotaMiddleware::incrementInvoiceUsage($tenantId);
$db->commit();
self::log("Queue ID $queueId: ✓ Invoice $invoiceId created and committed.");
// Progress Push
} catch (\Throwable $e) {
if ($db->inTransaction()) {
$db->rollBack();
}
self::log("Queue ID $queueId: DB ERROR: " . $e->getMessage());
try {
$db->prepare("UPDATE invoice_processing_queue SET status = 'failed', error_message = ? WHERE id = ?")->execute([$e->getMessage(), $queueId]);
} catch (\Throwable $e2) {}
return false;
}
// Check if entire batch is complete
self::checkBatchCompletion($batchId);
// Send push notification (non-critical, don't fail on error)
try {
$stmt = $db->prepare("SELECT total_images, processed_images, uploaded_by FROM invoice_batches WHERE id = ?");
$stmt->execute([$batchId]);
@@ -122,39 +169,44 @@ class InvoiceProcessor
'total' => $currentBatch['total_images']
]);
}
} catch (\Exception $pushErr) {}
} catch (\Exception $e) {
if ($db->inTransaction()) {
$db->rollBack();
} catch (\Throwable $pushErr) {
self::log("Queue ID $queueId: Push notification failed (non-critical): " . $pushErr->getMessage());
}
try {
$db->prepare("UPDATE invoice_processing_queue SET status = 'failed', error_message = ? WHERE id = ?")->execute([$e->getMessage(), $queueId]);
} catch (\Exception $e2) {}
return false;
}
// Check if batch complete
self::checkBatchCompletion($batchId);
return true;
} catch (\Throwable $e) {
self::log("Queue ID $queueId: UNHANDLED EXCEPTION: " . $e->getMessage() . "\n" . $e->getTraceAsString());
return false;
}
}
public static function checkBatchCompletion(string $batchId): void
{
try {
$db = Database::getInstance();
$stmt = $db->prepare("SELECT total_images, processed_images, uploaded_by FROM invoice_batches WHERE id = ?");
$stmt->execute([$batchId]);
$batch = $stmt->fetch();
if ($batch && $batch['processed_images'] >= $batch['total_images']) {
$db->prepare("UPDATE invoice_batches SET status = 'done', completed_at = NOW() WHERE id = ?")->execute([$batchId]);
self::log("Batch $batchId: COMPLETE ({$batch['processed_images']}/{$batch['total_images']})");
try {
$notifier = new NotificationService();
$title = "اكتملت معالجة الدفعة";
$body = "تمت معالجة جميع الفواتير بنجاح. يمكنك الآن مراجعتها وتدقيقها في لوحة التحكم قبل اعتمادها.";
$notifier->sendNotification($batch['uploaded_by'], $title, $body, ['batch_id' => $batchId]);
} catch (\Exception $e) {}
$notifier->sendNotification(
$batch['uploaded_by'],
"اكتملت معالجة الدفعة",
"تمت معالجة جميع الفواتير بنجاح. يمكنك الآن مراجعتها وتدقيقها في لوحة التحكم قبل اعتمادها.",
['batch_id' => $batchId]
);
} catch (\Throwable $e) {
self::log("Batch $batchId: Completion notification failed: " . $e->getMessage());
}
}
} catch (\Throwable $e) {
self::log("Batch $batchId: checkBatchCompletion error: " . $e->getMessage());
}
}
}

139
app/cron/diagnose.php Normal file
View File

@@ -0,0 +1,139 @@
<?php
/**
* Diagnostic Script — Run on server to verify processing works
*
* Usage: php app/cron/diagnose.php
*/
declare(strict_types=1);
require_once __DIR__ . '/../bootstrap/init.php';
use App\Core\Database;
echo "=== Musadaq Processing Diagnostics ===\n";
echo "Time: " . date('Y-m-d H:i:s') . "\n";
echo "PHP: " . PHP_VERSION . "\n";
echo "SAPI: " . php_sapi_name() . "\n\n";
// 1. Check DB connection
echo "--- Database ---\n";
try {
$db = Database::getInstance();
echo " ✓ Database connected\n";
} catch (\Throwable $e) {
echo " ✗ Database FAILED: " . $e->getMessage() . "\n";
exit(1);
}
// 2. Check pending queue items
echo "\n--- Queue Status ---\n";
$stmt = $db->query("SELECT status, COUNT(*) as cnt FROM invoice_processing_queue GROUP BY status");
$rows = $stmt->fetchAll();
if (empty($rows)) {
echo " (empty — no items in queue at all)\n";
} else {
foreach ($rows as $r) {
echo " {$r['status']}: {$r['cnt']}\n";
}
}
// 3. Check batch statuses
echo "\n--- Batch Status ---\n";
$stmt = $db->query("SELECT status, COUNT(*) as cnt FROM invoice_batches GROUP BY status");
$rows = $stmt->fetchAll();
if (empty($rows)) {
echo " (empty — no batches)\n";
} else {
foreach ($rows as $r) {
echo " {$r['status']}: {$r['cnt']}\n";
}
}
// 4. Check for stuck items (processing but no worker)
echo "\n--- Stuck Items (processing for >5 minutes) ---\n";
$stmt = $db->query("
SELECT q.id, q.batch_id, q.status, q.image_path, q.created_at, q.error_message
FROM invoice_processing_queue q
WHERE q.status IN ('pending', 'processing')
ORDER BY q.created_at DESC
LIMIT 10
");
$stuck = $stmt->fetchAll();
if (empty($stuck)) {
echo " (none — all clear)\n";
} else {
foreach ($stuck as $s) {
$exists = file_exists($s['image_path']) ? '✓ file exists' : '✗ FILE MISSING';
echo " ID={$s['id']} | Status={$s['status']} | $exists\n";
echo " Path: {$s['image_path']}\n";
if ($s['error_message']) echo " Error: {$s['error_message']}\n";
}
}
// 5. Check lock file
echo "\n--- Lock File ---\n";
$lockFile = STORAGE_PATH . '/logs/process_batches.lock';
if (file_exists($lockFile)) {
$age = time() - filemtime($lockFile);
$content = trim(file_get_contents($lockFile));
echo " ⚠ Lock file EXISTS (age: {$age}s, content: $content)\n";
if ($age > 300) {
echo " → This lock is STALE. Removing...\n";
@unlink($lockFile);
echo " ✓ Removed.\n";
}
} else {
echo " ✓ No lock file (good)\n";
}
// 6. Check key files
echo "\n--- Key Files ---\n";
$files = [
'InvoiceProcessor' => APP_PATH . '/Services/InvoiceProcessor.php',
'AI' => APP_PATH . '/Core/AI.php',
'process_batches' => APP_PATH . '/cron/process_batches.php',
'worker.log' => STORAGE_PATH . '/logs/worker.log',
];
foreach ($files as $name => $path) {
if (file_exists($path)) {
echo " ✓ $name: $path (" . filesize($path) . " bytes)\n";
} else {
echo " ✗ $name: MISSING — $path\n";
}
}
// 7. Check Gemini API key
echo "\n--- Configuration ---\n";
$apiKey = env('GEMINI_API_KEY');
echo " GEMINI_API_KEY: " . ($apiKey ? "✓ Set (" . strlen($apiKey) . " chars)" : "✗ MISSING!") . "\n";
echo " APP_DEBUG: " . env('APP_DEBUG', 'false') . "\n";
echo " fastcgi_finish_request: " . (function_exists('fastcgi_finish_request') ? '✓ Available' : '✗ Not available (CLI mode)') . "\n";
// 8. Show last lines of worker.log
echo "\n--- Last 20 lines of worker.log ---\n";
$workerLog = STORAGE_PATH . '/logs/worker.log';
if (file_exists($workerLog)) {
$lines = file($workerLog);
$last = array_slice($lines, -20);
foreach ($last as $line) {
echo " " . rtrim($line) . "\n";
}
} else {
echo " (worker.log does not exist yet)\n";
}
// 9. Try to reset any stuck 'processing' items back to 'pending'
echo "\n--- Fix Stuck Items? ---\n";
$stmt = $db->query("SELECT COUNT(*) FROM invoice_processing_queue WHERE status = 'processing'");
$stuckCount = (int)$stmt->fetchColumn();
if ($stuckCount > 0) {
echo " Found $stuckCount items stuck in 'processing' state.\n";
$db->query("UPDATE invoice_processing_queue SET status = 'pending' WHERE status = 'processing'");
echo " ✓ Reset them to 'pending' so they can be reprocessed.\n";
} else {
echo " ✓ No stuck items.\n";
}
echo "\n=== Diagnostics Complete ===\n";
echo "Next step: Run 'php app/cron/process_batches.php' to process pending items.\n";

89
app/cron/process_batches.php
View File

@@ -1,56 +1,91 @@
<?php
/**
* Background Worker for AI Invoice Extraction
* Processes images in the invoice_processing_queue.
* 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;
// Prevent multiple instances (Lock file)
// Simple lock: prevent overlapping runs
$lockFile = STORAGE_PATH . '/logs/process_batches.lock';
$fp = fopen($lockFile, 'c+');
if (!flock($fp, LOCK_EX | LOCK_NB)) {
exit("Worker already running.\n");
// 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);
}
}
echo "Starting Musadaq AI Worker [" . date('Y-m-d H:i:s') . "]\n";
// 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;
while (true) {
// Get ALL pending items (no infinite loop!)
$stmt = $db->prepare("
SELECT id FROM invoice_processing_queue
WHERE status = 'pending'
ORDER BY created_at ASC
LIMIT 1
LIMIT 20
");
$stmt->execute();
$queueId = $stmt->fetchColumn();
$items = $stmt->fetchAll(\PDO::FETCH_COLUMN);
if (!$queueId) {
echo "Queue empty. Waiting...\n";
sleep(5);
continue;
}
echo "Processing Queue ID: $queueId\n";
$success = InvoiceProcessor::processQueueItem((int)$queueId);
if ($success) {
echo "Success for Queue ID $queueId\n";
if (empty($items)) {
workerLog("No pending items. Exiting.");
} else {
echo "Failed for Queue ID $queueId\n";
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());
}
}
} catch (Exception $e) {
echo "Fatal Worker Error: " . $e->getMessage() . "\n";
} finally {
flock($fp, LOCK_UN);
fclose($fp);
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);
}

82
app/modules_app/batches/finalize.php
View File

@@ -4,7 +4,7 @@
* POST /v1/batches/finalize
*
* Marks a batch as ready for processing.
* Triggers background processing (or processes synchronously depending on setup).
* Sends instant response to mobile app, then processes in background via fastcgi_finish_request.
*/
declare(strict_types=1);
@@ -40,8 +40,6 @@ if (!$batch || ($decoded['role'] !== 'super_admin' && $batch['tenant_id'] !== $t
json_error('الدفعة غير موجودة', 404);
}
if ($batch['status'] !== 'uploading') {
json_error('تم إنهاء هذه الدفعة مسبقاً', 400);
}
@@ -58,11 +56,9 @@ $stmt = $db->prepare("
");
$stmt->execute([$batchId]);
// 3. Post-Response Processing (Background)
// Manually send response instead of json_success() because it calls exit;
header('Content-Type: application/json; charset=utf-8');
http_response_code(200);
echo json_encode([
// 3. Send response IMMEDIATELY to mobile app
// We manually build the response instead of using json_success() because it calls exit()
$responsePayload = json_encode([
'success' => true,
'data' => [
'batch_id' => $batchId,
@@ -73,22 +69,74 @@ echo json_encode([
'timestamp' => date('c')
], JSON_UNESCAPED_UNICODE);
// This allows the mobile app to get an instant success message,
// while the server continues processing the AI extraction in the background.
// Set headers
header('Content-Type: application/json; charset=utf-8');
header('Content-Length: ' . strlen($responsePayload));
http_response_code(200);
// Flush ALL output buffers to send response to client NOW
echo $responsePayload;
// Flush PHP output buffers
if (ob_get_level() > 0) {
ob_end_flush();
}
flush();
// Log the API call for app.log (mimicking json_response behavior)
$logEntry = sprintf(
"API %s %s | 200 | OK | %s",
$_SERVER['REQUEST_METHOD'] ?? 'CLI',
$_SERVER['REQUEST_URI'] ?? '',
'تم إنهاء الدفعة بنجاح وبدء المعالجة الفورية'
);
error_log($logEntry);
@file_put_contents(
STORAGE_PATH . '/logs/app.log',
"[" . date('Y-m-d H:i:s') . "] " . $logEntry . "\n",
FILE_APPEND
);
// 4. Tell PHP-FPM: "The client response is done. But keep this PHP process alive."
if (function_exists('fastcgi_finish_request')) {
fastcgi_finish_request();
}
// Ensure the script keeps running
// 5. Now process in the background (client has already received the response)
ignore_user_abort(true);
set_time_limit(300); // Give it 5 minutes
set_time_limit(300); // 5 minutes max
$queueStmt = $db->prepare("SELECT id FROM invoice_processing_queue WHERE batch_id = ? AND status = 'pending'");
$bgLog = function(string $msg) {
@file_put_contents(
STORAGE_PATH . '/logs/worker.log',
"[" . date('Y-m-d H:i:s') . "] [finalize-bg] " . $msg . "\n",
FILE_APPEND
);
};
$bgLog("Background processing started for batch: $batchId");
try {
$queueStmt = $db->prepare("SELECT id FROM invoice_processing_queue WHERE batch_id = ? AND status = 'pending' ORDER BY created_at ASC");
$queueStmt->execute([$batchId]);
$items = $queueStmt->fetchAll();
$items = $queueStmt->fetchAll(\PDO::FETCH_COLUMN);
foreach ($items as $item) {
InvoiceProcessor::processQueueItem((int)$item['id']);
$bgLog("Found " . count($items) . " pending item(s) for batch $batchId");
foreach ($items as $queueId) {
$bgLog("Processing queue item: $queueId");
try {
$success = InvoiceProcessor::processQueueItem((int)$queueId);
$bgLog("Queue item $queueId: " . ($success ? "SUCCESS" : "FAILED"));
} catch (\Throwable $e) {
$bgLog("Queue item $queueId EXCEPTION: " . $e->getMessage());
}
}
exit; // End here
$bgLog("Background processing finished for batch: $batchId");
} catch (\Throwable $e) {
$bgLog("FATAL ERROR in background processing: " . $e->getMessage());
}
exit;