461 lines
19 KiB
PHP
461 lines
19 KiB
PHP
<?php
|
|
declare(strict_types=1);
|
|
|
|
namespace App\Services;
|
|
|
|
use App\Core\Database;
|
|
use App\Core\AI;
|
|
use App\Core\Encryption;
|
|
use App\Middleware\QuotaMiddleware;
|
|
|
|
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;
|
|
}
|
|
}
|
|
|
|
/**
|
|
* 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.
|
|
*/
|
|
public static function processQueueItem(int $queueId): bool
|
|
{
|
|
self::log("Starting processQueueItem($queueId)");
|
|
|
|
try {
|
|
$db = Database::getInstance();
|
|
} catch (\Throwable $e) {
|
|
self::log("FATAL: Cannot connect to DB: " . $e->getMessage());
|
|
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 {
|
|
$item = self::claimQueueItem($db, $queueId);
|
|
|
|
if (!$item) {
|
|
self::log("Queue ID $queueId: not claimable (already taken, not pending, or attempts exhausted). Skipping.");
|
|
return false;
|
|
}
|
|
|
|
$batchId = (string)$item['batch_id'];
|
|
$tenantId = (string)$item['batch_tenant_id'];
|
|
$companyId = (string)$item['batch_company_id'];
|
|
$userId = $item['uploaded_by'];
|
|
$imagePath = (string)$item['image_path'];
|
|
|
|
self::log("Queue ID $queueId: Image=$imagePath, Batch=$batchId");
|
|
|
|
// 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");
|
|
$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;
|
|
}
|
|
|
|
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)
|
|
AI::setTenantContext($tenantId);
|
|
$extractedInvoices = AI::extractInvoices($base64Data, $mimeType);
|
|
AI::setTenantContext(null);
|
|
|
|
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 extracted " . count($extractedInvoices) . " invoice(s). Saving to DB...");
|
|
|
|
$createdInvoiceIds = [];
|
|
|
|
// Save to database in a transaction
|
|
$db->beginTransaction();
|
|
try {
|
|
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 = ?, 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]);
|
|
|
|
$db->commit();
|
|
$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());
|
|
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 push (silent data message + Live Activity update)
|
|
try {
|
|
$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();
|
|
// 'batch_progress' is the type NotificationService recognises for
|
|
// iOS Live Activity / Android Live Update payloads.
|
|
$notifier->sendDataNotification($currentBatch['uploaded_by'], [
|
|
'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) {
|
|
self::log("Queue ID $queueId: Push notification failed (non-critical): " . $pushErr->getMessage());
|
|
}
|
|
|
|
return true;
|
|
|
|
} 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, failed_images, status, uploaded_by
|
|
FROM invoice_batches WHERE id = ?
|
|
");
|
|
$stmt->execute([$batchId]);
|
|
$batch = $stmt->fetch();
|
|
|
|
if (!$batch) {
|
|
return;
|
|
}
|
|
|
|
// 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());
|
|
}
|
|
}
|
|
}
|