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

Update: 2026-07-30 02:27:45

Browse Source
This commit is contained in:
Hamza-Ayed
2026-07-30 02:27:45 +03:00
parent 5f62455113
commit ca4a7c2e70
56 changed files with 3391 additions and 709 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/Services/InvoiceProcessor.php
+342 -129
View File
@@ -20,6 +20,90 @@ class InvoiceProcessor
}
}
/**
* Atomically claim a pending queue item.
*
* Two consumers race for the queue (the background task spawned by
* batches/finalize.php and the every-minute cron). A SELECT-then-UPDATE
* would let both see the same row as 'pending' and process it twice,
* producing duplicate invoices and double-charging the tenant's quota.
* A single conditional UPDATE makes the claim exclusive: only the process
* whose UPDATE affected a row owns the item.
*
* @return array|null The claimed item joined with its batch, or null if
* another process got there first / it is exhausted.
*/
private static function claimQueueItem(\PDO $db, int $queueId): ?array
{
$claim = $db->prepare("
UPDATE invoice_processing_queue
SET status = 'processing',
attempts = attempts + 1,
claimed_at = NOW(),
error_message = NULL
WHERE id = ?
AND status = 'pending'
AND attempts < COALESCE(max_attempts, 3)
");
$claim->execute([$queueId]);
if ($claim->rowCount() !== 1) {
return null;
}
$stmt = $db->prepare("
SELECT q.*, b.tenant_id AS batch_tenant_id, b.company_id AS batch_company_id,
b.uploaded_by, b.total_images, b.status AS batch_status
FROM invoice_processing_queue q
JOIN invoice_batches b ON q.batch_id = b.id
WHERE q.id = ?
");
$stmt->execute([$queueId]);
$item = $stmt->fetch();
return $item ?: null;
}
/**
* Release a claimed item back to the queue for a later retry, or bury it as
* permanently failed once max_attempts is exhausted.
*
* Only a permanent failure counts towards the batch's failed_images, so a
* transient AI hiccup does not prematurely "complete" a batch.
*/
private static function failQueueItem(\PDO $db, array $item, string $reason): void
{
$queueId = (int)$item['id'];
$attempts = (int)($item['attempts'] ?? 1);
$maxAttempts = (int)($item['max_attempts'] ?? 3);
$batchId = (string)$item['batch_id'];
if ($attempts < $maxAttempts) {
$db->prepare("
UPDATE invoice_processing_queue
SET status = 'pending', error_message = ?
WHERE id = ?
")->execute([$reason, $queueId]);
self::log("Queue ID $queueId: will retry ($attempts/$maxAttempts) - $reason");
return;
}
$db->prepare("
UPDATE invoice_processing_queue
SET status = 'failed', error_message = ?, processed_at = NOW()
WHERE id = ?
")->execute([$reason, $queueId]);
$db->prepare("
UPDATE invoice_batches SET failed_images = failed_images + 1 WHERE id = ?
")->execute([$batchId]);
self::log("Queue ID $queueId: PERMANENTLY FAILED after $attempts attempt(s) - $reason");
self::checkBatchCompletion($batchId);
}
/**
* Processes a single invoice queue item by its ID.
*/
@@ -34,37 +118,41 @@ class InvoiceProcessor
return false;
}
$item = null;
// Once the invoice is committed the item must never be released back to
// 'pending', or a later worker would re-insert the same invoice.
$committed = 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
WHERE q.id = ? AND q.status = 'pending'
");
$stmt->execute([$queueId]);
$item = $stmt->fetch();
$item = self::claimQueueItem($db, $queueId);
if (!$item) {
self::log("Queue ID $queueId: Not found or not pending. Skipping.");
self::log("Queue ID $queueId: not claimable (already taken, not pending, or attempts exhausted). Skipping.");
return false;
}
$batchId = $item['batch_id'];
$tenantId = $item['tenant_id'];
$companyId = $item['company_id'];
$batchId = (string)$item['batch_id'];
$tenantId = (string)$item['batch_tenant_id'];
$companyId = (string)$item['batch_company_id'];
$userId = $item['uploaded_by'];
$imagePath = $item['image_path'];
$imagePath = (string)$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
// Check file exists. A missing file will never appear, so bury it
// immediately rather than burning the remaining retries.
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]);
$item['attempts'] = $item['max_attempts'] ?? 3;
self::failQueueItem($db, $item, 'File not found');
return false;
}
// Refuse to spend AI credit the tenant no longer has.
if (!QuotaMiddleware::hasInvoiceQuota($tenantId)) {
self::log("Queue ID $queueId: tenant $tenantId is out of invoice quota.");
$item['attempts'] = $item['max_attempts'] ?? 3;
self::failQueueItem($db, $item, 'تم استنفاد رصيد الفواتير لهذا الشهر');
return false;
}
@@ -75,124 +163,92 @@ class InvoiceProcessor
$base64Data = base64_encode($fileContent);
// AI Extraction (this takes ~5-15 seconds)
$extracted = AI::extractInvoiceData($base64Data, $mimeType);
AI::setTenantContext($tenantId);
$extractedInvoices = AI::extractInvoices($base64Data, $mimeType);
AI::setTenantContext(null);
if (!$extracted) {
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]);
if (empty($extractedInvoices)) {
self::log("Queue ID $queueId: AI extraction returned nothing.");
self::failQueueItem($db, $item, 'AI failed to extract data from image');
return false;
}
self::log("Queue ID $queueId: AI extraction successful. Saving to DB...");
self::log("Queue ID $queueId: AI extracted " . count($extractedInvoices) . " invoice(s). Saving to DB...");
$createdInvoiceIds = [];
// 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'] ?? '';
$invoiceNum = $extracted['invoice_number'] ?? '';
$invoiceDate = $extracted['invoice_date'] ?? '';
$validDate = (!empty($invoiceDate) && strtotime($invoiceDate)) ? $invoiceDate : null;
$stmt = $db->prepare("
INSERT INTO invoices (
id, tenant_id, company_id, uploaded_by, original_file_path, status,
invoice_number, invoice_date, invoice_type, invoice_category,
supplier_tin, supplier_name, supplier_address,
buyer_tin, buyer_name, buyer_national_id,
subtotal, tax_amount, discount_total, grand_total, currency_code,
created_at
) VALUES (
?, ?, ?, ?, ?, 'extracted',
?, ?, ?, ?, ?, ?, ?, ?, ?, ?,
?, ?, ?, ?, ?,
NOW()
)
");
$stmt->execute([
$invoiceId, $tenantId, $companyId, $userId, $imagePath,
$invoiceNum, $validDate, $extracted['invoice_type'] ?? 'cash', $extracted['invoice_category'] ?? 'simplified',
Encryption::encrypt($supplierTin), Encryption::encrypt($extracted['supplier']['name'] ?? ''), Encryption::encrypt($extracted['supplier']['address'] ?? ''),
Encryption::encrypt($extracted['buyer']['tin'] ?? ''), Encryption::encrypt($extracted['buyer']['name'] ?? ''), Encryption::encrypt($extracted['buyer']['national_id'] ?? ''),
$extracted['subtotal'] ?? 0, $extracted['tax_amount'] ?? 0, $extracted['discount_total'] ?? 0, $extracted['grand_total'] ?? 0, $extracted['currency_code'] ?? 'JOD'
]);
// 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, tax_amount,
discount_amount, net_total, line_total, tax_category
) VALUES (?,?,?,?,?,?,?,?,?,?,?,?)
");
foreach ($extracted['lines'] as $idx => $line) {
$quantity = (float)($line['quantity'] ?? 1);
$unitPrice = (float)($line['unit_price'] ?? 0);
$taxRate = (float)($line['tax_rate'] ?? 0);
$discount = (float)($line['discount'] ?? $line['discount_amount'] ?? 0);
$subtotal = $quantity * $unitPrice;
$taxAmount = (float)($line['tax_amount'] ?? ($subtotal * $taxRate));
$netTotal = (float)($line['net_total'] ?? ($line['line_total'] ?? ($subtotal + $taxAmount - $discount)));
$lineStmt->execute([
vsprintf('%s%s-%s-%s-%s-%s%s%s', str_split(bin2hex(random_bytes(16)), 4)),
$invoiceId,
$line['line_number'] ?? ($idx + 1),
$line['description'] ?? '',
$quantity,
$unitPrice,
$taxRate,
$taxAmount,
$discount,
$netTotal,
$netTotal, // line_total
$line['tax_category'] ?? 'standard'
]);
}
self::log("Queue ID $queueId: Saved " . count($extracted['lines']) . " line items.");
foreach ($extractedInvoices as $extracted) {
$createdInvoiceIds[] = self::insertInvoice(
$db,
$extracted,
$batchId,
$tenantId,
$companyId,
$userId,
$imagePath
);
}
$primaryInvoiceId = $createdInvoiceIds[0];
// Mark queue item done
$db->prepare("UPDATE invoice_processing_queue SET status = 'done', invoice_id = ?, processed_at = NOW() WHERE id = ?")->execute([$invoiceId, $queueId]);
$db->prepare("UPDATE invoice_processing_queue SET status = 'done', invoice_id = ?, error_message = NULL, processed_at = NOW() WHERE id = ?")
->execute([$primaryInvoiceId, $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->prepare("UPDATE invoice_batches SET processed_images = processed_images + 1 WHERE id = ?")
->execute([$batchId]);
$db->commit();
self::log("Queue ID $queueId: ✓ Invoice $invoiceId created and committed.");
$committed = true;
self::log("Queue ID $queueId: OK - invoice(s) " . implode(', ', $createdInvoiceIds) . " committed.");
} 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) {}
self::failQueueItem($db, $item, $e->getMessage());
return false;
}
// Charge quota only for invoices that are actually committed. This runs
// outside the transaction so a quota-write failure cannot roll back
// (or be rolled back by) the invoice itself.
//
// Wrapped separately: letting this bubble to the outer catch would
// release an ALREADY-COMMITTED item back to 'pending' and the next
// worker would insert the same invoice a second time.
try {
foreach ($createdInvoiceIds as $_) {
QuotaMiddleware::incrementInvoiceUsage($tenantId);
}
} catch (\Throwable $quotaErr) {
self::log("Queue ID $queueId: quota increment failed (invoice already saved): " . $quotaErr->getMessage());
}
// Check if entire batch is complete
self::checkBatchCompletion($batchId);
// Progress/Completion Push
// Progress push (silent data message + Live Activity update)
try {
$stmt = $db->prepare("SELECT total_images, processed_images, uploaded_by FROM invoice_batches WHERE id = ?");
$stmt = $db->prepare("SELECT total_images, processed_images, failed_images, status, uploaded_by FROM invoice_batches WHERE id = ?");
$stmt->execute([$batchId]);
$currentBatch = $stmt->fetch();
if ($currentBatch) {
$notifier = new NotificationService();
// Send data notification with invoice_id for auto-navigation
// 'batch_progress' is the type NotificationService recognises for
// iOS Live Activity / Android Live Update payloads.
$notifier->sendDataNotification($currentBatch['uploaded_by'], [
'type' => 'invoice_processed',
'batch_id' => $batchId,
'invoice_id' => $invoiceId,
'processed' => $currentBatch['processed_images'],
'total' => $currentBatch['total_images']
'type' => 'batch_progress',
'batch_id' => $batchId,
'invoice_id' => $createdInvoiceIds[0],
'processed' => $currentBatch['processed_images'],
'failed' => $currentBatch['failed_images'],
'total' => $currentBatch['total_images'],
'is_done' => in_array($currentBatch['status'], ['done', 'partial_fail', 'failed'], true) ? 1 : 0,
]);
}
} catch (\Throwable $pushErr) {
@@ -203,42 +259,199 @@ class InvoiceProcessor
} catch (\Throwable $e) {
self::log("Queue ID $queueId: UNHANDLED EXCEPTION: " . $e->getMessage() . "\n" . $e->getTraceAsString());
// Never leave a claimed row stuck in 'processing' - unless the invoice
// was already committed, in which case retrying would duplicate it.
if ($item !== null && !$committed) {
try {
self::failQueueItem($db, $item, $e->getMessage());
} catch (\Throwable $e2) {
self::log("Queue ID $queueId: could not release item: " . $e2->getMessage());
}
}
return false;
}
}
/**
* Insert one extracted invoice (plus its line items) and return its id.
*/
private static function insertInvoice(
\PDO $db,
array $extracted,
string $batchId,
string $tenantId,
string $companyId,
?string $userId,
string $imagePath
): string {
$invoiceId = vsprintf('%s%s-%s-%s-%s-%s%s%s', str_split(bin2hex(random_bytes(16)), 4));
$supplierTin = $extracted['supplier']['tin'] ?? '';
$invoiceNum = $extracted['invoice_number'] ?? '';
$invoiceDate = $extracted['invoice_date'] ?? '';
$validDate = (!empty($invoiceDate) && strtotime($invoiceDate)) ? $invoiceDate : null;
$stmt = $db->prepare("
INSERT INTO invoices (
id, tenant_id, company_id, batch_id, uploaded_by, original_file_path, status,
invoice_number, invoice_date, invoice_type, invoice_category,
supplier_tin, supplier_name, supplier_address,
buyer_tin, buyer_name, buyer_national_id,
subtotal, tax_amount, discount_total, grand_total, currency_code,
created_at
) VALUES (
?, ?, ?, ?, ?, ?, 'extracted',
?, ?, ?, ?, ?, ?, ?, ?, ?, ?,
?, ?, ?, ?, ?,
NOW()
)
");
$stmt->execute([
$invoiceId, $tenantId, $companyId, $batchId, $userId, $imagePath,
$invoiceNum, $validDate, $extracted['invoice_type'] ?? 'cash', $extracted['invoice_category'] ?? 'simplified',
Encryption::encrypt($supplierTin), Encryption::encrypt($extracted['supplier']['name'] ?? ''), Encryption::encrypt($extracted['supplier']['address'] ?? ''),
Encryption::encrypt($extracted['buyer']['tin'] ?? ''), Encryption::encrypt($extracted['buyer']['name'] ?? ''), Encryption::encrypt($extracted['buyer']['national_id'] ?? ''),
$extracted['subtotal'] ?? 0, $extracted['tax_amount'] ?? 0, $extracted['discount_total'] ?? 0, $extracted['grand_total'] ?? 0, $extracted['currency_code'] ?? 'JOD'
]);
// 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, tax_amount,
discount_amount, net_total, line_total, tax_category
) VALUES (?,?,?,?,?,?,?,?,?,?,?,?)
");
foreach ($extracted['lines'] as $idx => $line) {
$quantity = (float)($line['quantity'] ?? 1);
$unitPrice = (float)($line['unit_price'] ?? 0);
$taxRate = (float)($line['tax_rate'] ?? 0);
$discount = (float)($line['discount'] ?? $line['discount_amount'] ?? 0);
$subtotal = $quantity * $unitPrice;
$taxAmount = (float)($line['tax_amount'] ?? ($subtotal * $taxRate));
$netTotal = (float)($line['net_total'] ?? ($line['line_total'] ?? ($subtotal + $taxAmount - $discount)));
$lineStmt->execute([
vsprintf('%s%s-%s-%s-%s-%s%s%s', str_split(bin2hex(random_bytes(16)), 4)),
$invoiceId,
$line['line_number'] ?? ($idx + 1),
$line['description'] ?? '',
$quantity,
$unitPrice,
$taxRate,
$taxAmount,
$discount,
$netTotal,
$netTotal, // line_total
$line['tax_category'] ?? 'standard'
]);
}
self::log("Invoice $invoiceId: saved " . count($extracted['lines']) . " line items.");
}
return $invoiceId;
}
/**
* Close a batch once every image has reached a terminal state.
*
* Two rules matter here:
* 1. A batch still in 'uploading' is NEVER completed. total_images grows as
* each image arrives, so a worker that processed the first image would
* otherwise see 1 >= 1 and lock the batch while the user is still
* uploading - after which upload-image and finalize both reject with a
* misleading "already processed" error.
* 2. Permanent failures count towards completion, otherwise one bad photo
* leaves the batch (and the app's progress bar) hanging forever.
*/
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 = $db->prepare("
SELECT total_images, processed_images, failed_images, status, 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 {
// Try to get the last invoice_id for this batch for completion navigation
$invStmt = $db->prepare("SELECT id FROM invoices WHERE original_file_path IN (SELECT image_path FROM invoice_processing_queue WHERE batch_id = ?) ORDER BY created_at DESC LIMIT 1");
$invStmt->execute([$batchId]);
$lastInvoiceId = $invStmt->fetchColumn();
if (!$batch) {
return;
}
$notifier = new NotificationService();
$notifier->sendNotification(
$batch['uploaded_by'],
"اكتملت معالجة الدفعة",
"تمت معالجة جميع الفواتير بنجاح. يمكنك الآن مراجعتها وتدقيقها.",
[
'type' => 'batch_complete',
'batch_id' => $batchId,
'invoice_id' => $lastInvoiceId ?: ''
]
);
} catch (\Throwable $e) {
self::log("Batch $batchId: Completion notification failed: " . $e->getMessage());
}
// Rule 1: the user is still uploading - nothing to close yet.
if ($batch['status'] === 'uploading') {
self::log("Batch $batchId: still uploading, completion check deferred.");
return;
}
// Already terminal.
if (in_array($batch['status'], ['done', 'partial_fail', 'failed'], true)) {
return;
}
$total = (int)$batch['total_images'];
$processed = (int)$batch['processed_images'];
$failed = (int)$batch['failed_images'];
// Rule 2: terminal means done OR permanently failed.
if ($total <= 0 || ($processed + $failed) < $total) {
return;
}
if ($failed === 0) {
$finalStatus = 'done';
} elseif ($processed === 0) {
$finalStatus = 'failed';
} else {
$finalStatus = 'partial_fail';
}
$db->prepare("UPDATE invoice_batches SET status = ?, completed_at = NOW() WHERE id = ?")
->execute([$finalStatus, $batchId]);
self::log("Batch $batchId: $finalStatus ($processed ok / $failed failed / $total total)");
try {
// invoices.batch_id is populated now, so this is a direct lookup.
$invStmt = $db->prepare("SELECT id FROM invoices WHERE batch_id = ? ORDER BY created_at DESC LIMIT 1");
$invStmt->execute([$batchId]);
$lastInvoiceId = $invStmt->fetchColumn();
[$title, $body] = match ($finalStatus) {
'done' => [
'اكتملت معالجة الدفعة',
'تمت معالجة جميع الفواتير بنجاح. يمكنك الآن مراجعتها وتدقيقها.',
],
'partial_fail' => [
'اكتملت المعالجة مع أخطاء',
"تمت معالجة {$processed} فاتورة بنجاح، وفشلت {$failed}. يمكنك إعادة تصوير الفواتير الفاشلة.",
],
default => [
'فشلت معالجة الدفعة',
"لم نتمكن من استخراج البيانات من {$failed} فاتورة. يرجى إعادة التصوير بإضاءة أفضل.",
],
};
$notifier = new NotificationService();
$notifier->sendNotification(
$batch['uploaded_by'],
$title,
$body,
[
'type' => 'batch_complete',
'batch_id' => $batchId,
'status' => $finalStatus,
'processed' => $processed,
'failed' => $failed,
'total' => $total,
'is_done' => 1,
'invoice_id' => $lastInvoiceId ?: '',
]
);
} catch (\Throwable $e) {
self::log("Batch $batchId: Completion notification failed: " . $e->getMessage());
}
} catch (\Throwable $e) {
self::log("Batch $batchId: checkBatchCompletion error: " . $e->getMessage());