Jeppe B
068f9e254f
Add the authoritative revision-bound invoice collection tree and guarded cleanup, merge, price-reset, and transfer operations.
1077 lines
37 KiB
PHP
1077 lines
37 KiB
PHP
<?php
|
|
|
|
namespace classes;
|
|
|
|
use Exception;
|
|
use mysqli_result;
|
|
use objects\logs_o;
|
|
|
|
/**
|
|
* Queue service for asynchronous e-conomic transfer jobs.
|
|
*/
|
|
class economic_transfer_queue
|
|
{
|
|
public const STATUS_QUEUED = 'QUEUED';
|
|
public const STATUS_PROCESSING = 'PROCESSING';
|
|
public const STATUS_COMPLETED = 'COMPLETED';
|
|
public const STATUS_FAILED = 'FAILED';
|
|
|
|
public const TYPE_ORDER_DRAFT_EXPORT = 'ORDER_DRAFT_EXPORT';
|
|
public const TYPE_ORDER_INVOICE_EXPORT = 'ORDER_INVOICE_EXPORT';
|
|
public const TYPE_COLLECTED_INVOICE_EXPORT = 'COLLECTED_INVOICE_EXPORT';
|
|
private const STALE_PROCESSING_LOCK_SECONDS = 900;
|
|
|
|
private economic_transfer_executor $executor;
|
|
|
|
public function __construct(?economic_transfer_executor $executor = null)
|
|
{
|
|
$this->executor = $executor ?? new economic_transfer_executor();
|
|
economic_transfer_queue_schema_bootstrap::ensureTables();
|
|
}
|
|
|
|
/**
|
|
* @throws Exception
|
|
*/
|
|
public function enqueue(string $transfer_type, array $payload, int $created_by = 0, int $max_attempts = 3): array
|
|
{
|
|
global $db;
|
|
|
|
$created_by = max(0, $created_by);
|
|
$max_attempts = max(1, min(10, $max_attempts));
|
|
$transfer_type = $this->validateTransferType($transfer_type);
|
|
$payload = $this->normalizePayloadForTransferType($transfer_type, $payload, $created_by);
|
|
$collected_invoice_lock = $this->acquireCollectedInvoiceExportLock($transfer_type, $payload);
|
|
if ($collected_invoice_lock !== null) {
|
|
$this->assertCollectedInvoiceExportIsStillEligible($payload);
|
|
}
|
|
|
|
$active_job = $this->findActiveJobByTarget($transfer_type, $payload, $created_by);
|
|
if ($active_job !== null) {
|
|
$this->registerJobRequester((int)($active_job['id'] ?? 0), $created_by);
|
|
$target_label = $this->buildTargetLabel($transfer_type, $payload);
|
|
$this->logQueueEvent(
|
|
1,
|
|
$created_by,
|
|
'ECONOMIC_TRANSFER_JOB_DEDUPED',
|
|
'Transfer job deduped for active target (' . $target_label . '), returning existing job #' . (int)($active_job['id'] ?? 0)
|
|
);
|
|
return $active_job;
|
|
}
|
|
|
|
$payload_json = json_encode($payload, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES);
|
|
if ($payload_json === false) {
|
|
throw new Exception('Failed to serialize queue payload');
|
|
}
|
|
|
|
$stmt = $db->prepare(
|
|
"INSERT INTO economic_transfer_queue_jobs
|
|
(transfer_type, payload_json, status, progress_percent, progress_message, attempts, max_attempts, created_by)
|
|
VALUES (?, ?, ?, 0, 'Queued', 0, ?, ?)"
|
|
);
|
|
if (!$stmt) {
|
|
throw new Exception('Failed to prepare queue insert statement');
|
|
}
|
|
|
|
$status = self::STATUS_QUEUED;
|
|
$stmt->bind_param('sssii', $transfer_type, $payload_json, $status, $max_attempts, $created_by);
|
|
if (!$stmt->execute()) {
|
|
throw new Exception('Failed to enqueue transfer job');
|
|
}
|
|
$job_id = (int)$db->insert_id();
|
|
$stmt->close();
|
|
|
|
$this->registerJobRequester($job_id, $created_by);
|
|
|
|
$this->logQueueEvent(
|
|
1,
|
|
$created_by,
|
|
'ECONOMIC_TRANSFER_JOB_ENQUEUED',
|
|
'Transfer job #' . $job_id . ' queued: ' . $transfer_type
|
|
);
|
|
|
|
$job = $this->getJobById($job_id);
|
|
if ($job === null) {
|
|
throw new Exception('Failed to load queued transfer job');
|
|
}
|
|
return $job;
|
|
}
|
|
|
|
public function getJobById(int $job_id): ?array
|
|
{
|
|
global $db;
|
|
|
|
$stmt = $db->prepare("SELECT * FROM economic_transfer_queue_jobs WHERE id = ? LIMIT 1");
|
|
if (!$stmt) {
|
|
return null;
|
|
}
|
|
|
|
$stmt->bind_param('i', $job_id);
|
|
if (!$stmt->execute()) {
|
|
$stmt->close();
|
|
return null;
|
|
}
|
|
|
|
$result = $stmt->get_result();
|
|
$row = $result instanceof mysqli_result ? $result->fetch_assoc() : null;
|
|
$stmt->close();
|
|
|
|
if (!$row) {
|
|
return null;
|
|
}
|
|
return $this->normalizeJobRow($row);
|
|
}
|
|
|
|
public function listJobs(array $statuses = [], int $limit = 50, int $offset = 0, ?string $transfer_type = null): array
|
|
{
|
|
global $db;
|
|
|
|
$limit = max(1, min(500, $limit));
|
|
$offset = max(0, $offset);
|
|
|
|
$where = $this->buildListJobsWhereClause($statuses, $transfer_type);
|
|
$sql = "SELECT * FROM economic_transfer_queue_jobs $where ORDER BY id DESC LIMIT $limit OFFSET $offset";
|
|
$result = $db->query($sql);
|
|
if (!$result instanceof mysqli_result) {
|
|
return [];
|
|
}
|
|
|
|
$jobs = [];
|
|
while ($row = $result->fetch_assoc()) {
|
|
$jobs[] = $this->normalizeJobRow($row);
|
|
}
|
|
return $jobs;
|
|
}
|
|
|
|
public function getJobByIdForUser(int $job_id, int $created_by): ?array
|
|
{
|
|
global $db;
|
|
|
|
$job_id = max(0, $job_id);
|
|
$created_by = max(0, $created_by);
|
|
if ($job_id < 1 || $created_by < 1) {
|
|
return null;
|
|
}
|
|
|
|
$stmt = $db->prepare(
|
|
"SELECT q.*
|
|
FROM economic_transfer_queue_jobs q
|
|
LEFT JOIN economic_transfer_queue_job_requesters r
|
|
ON r.queue_job_id = q.id AND r.user_id = ?
|
|
WHERE q.id = ? AND (q.created_by = ? OR r.user_id = ?)
|
|
LIMIT 1"
|
|
);
|
|
if (!$stmt) {
|
|
return null;
|
|
}
|
|
|
|
$stmt->bind_param('iiii', $created_by, $job_id, $created_by, $created_by);
|
|
if (!$stmt->execute()) {
|
|
$stmt->close();
|
|
return null;
|
|
}
|
|
|
|
$result = $stmt->get_result();
|
|
$row = $result instanceof mysqli_result ? $result->fetch_assoc() : null;
|
|
$stmt->close();
|
|
|
|
if (!$row) {
|
|
return null;
|
|
}
|
|
return $this->normalizeJobRow($row);
|
|
}
|
|
|
|
public function listJobsForCreatedBy(array $statuses = [], int $limit = 50, int $offset = 0, ?string $transfer_type = null, int $created_by = 0): array
|
|
{
|
|
global $db;
|
|
|
|
$created_by = max(0, $created_by);
|
|
if ($created_by < 1) {
|
|
return [];
|
|
}
|
|
|
|
$limit = max(1, min(500, $limit));
|
|
$offset = max(0, $offset);
|
|
|
|
$where = $this->buildListJobsWhereClause($statuses, $transfer_type);
|
|
$visibility = $this->jobVisibilitySql('economic_transfer_queue_jobs', $created_by);
|
|
$where .= $where === '' ? 'WHERE ' . $visibility : ' AND ' . $visibility;
|
|
$sql = "SELECT * FROM economic_transfer_queue_jobs $where ORDER BY id DESC LIMIT $limit OFFSET $offset";
|
|
$result = $db->query($sql);
|
|
if (!$result instanceof mysqli_result) {
|
|
return [];
|
|
}
|
|
|
|
$jobs = [];
|
|
while ($row = $result->fetch_assoc()) {
|
|
$jobs[] = $this->normalizeJobRow($row);
|
|
}
|
|
return $jobs;
|
|
}
|
|
|
|
public function countJobsForCreatedBy(array $statuses = [], ?string $transfer_type = null, int $created_by = 0): int
|
|
{
|
|
global $db;
|
|
|
|
$created_by = max(0, $created_by);
|
|
if ($created_by < 1) {
|
|
return 0;
|
|
}
|
|
|
|
$where = $this->buildListJobsWhereClause($statuses, $transfer_type);
|
|
$visibility = $this->jobVisibilitySql('economic_transfer_queue_jobs', $created_by);
|
|
$where .= $where === '' ? 'WHERE ' . $visibility : ' AND ' . $visibility;
|
|
$sql = "SELECT COUNT(*) AS total FROM economic_transfer_queue_jobs $where";
|
|
$result = $db->query($sql);
|
|
if (!$result instanceof mysqli_result) {
|
|
return 0;
|
|
}
|
|
|
|
$row = $result->fetch_assoc();
|
|
if (!is_array($row) || !isset($row['total'])) {
|
|
return 0;
|
|
}
|
|
|
|
return max(0, (int)$row['total']);
|
|
}
|
|
|
|
public function countJobs(array $statuses = [], ?string $transfer_type = null): int
|
|
{
|
|
global $db;
|
|
|
|
$where = $this->buildListJobsWhereClause($statuses, $transfer_type);
|
|
$sql = "SELECT COUNT(*) AS total FROM economic_transfer_queue_jobs $where";
|
|
$result = $db->query($sql);
|
|
if (!$result instanceof mysqli_result) {
|
|
return 0;
|
|
}
|
|
|
|
$row = $result->fetch_assoc();
|
|
if (!is_array($row) || !isset($row['total'])) {
|
|
return 0;
|
|
}
|
|
|
|
return max(0, (int)$row['total']);
|
|
}
|
|
|
|
public function listMonitorJobsForUser(int $user_id, int $limit = 50, ?string $transfer_type = null): array
|
|
{
|
|
global $db;
|
|
|
|
$user_id = max(0, $user_id);
|
|
if ($user_id < 1) {
|
|
return [];
|
|
}
|
|
$limit = max(1, min(100, $limit));
|
|
try {
|
|
$normalized_transfer_type = $transfer_type !== null && trim($transfer_type) !== ''
|
|
? $this->validateTransferType($transfer_type)
|
|
: null;
|
|
} catch (Exception) {
|
|
return [];
|
|
}
|
|
|
|
$transfer_condition = '';
|
|
if ($normalized_transfer_type !== null) {
|
|
$transfer_condition = "AND q.transfer_type = '" . $db->escape_string($normalized_transfer_type) . "'";
|
|
}
|
|
|
|
$sql = "SELECT q.*
|
|
FROM economic_transfer_queue_jobs q
|
|
LEFT JOIN economic_transfer_queue_job_dismissals d
|
|
ON d.queue_job_id = q.id
|
|
AND d.user_id = $user_id
|
|
AND d.dismissed_status = q.status
|
|
WHERE " . $this->jobVisibilitySql('q', $user_id) . "
|
|
$transfer_condition
|
|
AND (
|
|
q.status IN ('" . self::STATUS_QUEUED . "', '" . self::STATUS_PROCESSING . "')
|
|
OR d.queue_job_id IS NULL
|
|
)
|
|
ORDER BY
|
|
CASE WHEN q.status IN ('" . self::STATUS_QUEUED . "', '" . self::STATUS_PROCESSING . "') THEN 0 ELSE 1 END,
|
|
q.id DESC
|
|
LIMIT $limit";
|
|
$result = $db->query($sql);
|
|
if (!$result instanceof mysqli_result) {
|
|
return [];
|
|
}
|
|
|
|
$jobs = [];
|
|
while ($row = $result->fetch_assoc()) {
|
|
$jobs[] = $this->normalizeJobRow($row);
|
|
}
|
|
return $jobs;
|
|
}
|
|
|
|
/**
|
|
* @throws Exception
|
|
*/
|
|
public function dismissTerminalJobForUser(int $job_id, int $user_id): array
|
|
{
|
|
global $db;
|
|
|
|
$job_id = max(0, $job_id);
|
|
$user_id = max(0, $user_id);
|
|
if ($job_id < 1 || $user_id < 1) {
|
|
throw new Exception('Queue job and user are required');
|
|
}
|
|
|
|
$job = $this->getJobByIdForUser($job_id, $user_id);
|
|
if ($job === null) {
|
|
throw new Exception('Queue job not found');
|
|
}
|
|
|
|
$status = strtoupper((string)($job['status'] ?? ''));
|
|
if (!in_array($status, [self::STATUS_COMPLETED, self::STATUS_FAILED], true)) {
|
|
throw new Exception('Only completed or failed queue jobs can be dismissed');
|
|
}
|
|
|
|
$stmt = $db->prepare(
|
|
"INSERT INTO economic_transfer_queue_job_dismissals (queue_job_id, user_id, dismissed_status, dismissed_at)
|
|
VALUES (?, ?, ?, NOW())
|
|
ON DUPLICATE KEY UPDATE dismissed_status = VALUES(dismissed_status), dismissed_at = NOW()"
|
|
);
|
|
if (!$stmt) {
|
|
throw new Exception('Failed to prepare queue dismissal statement');
|
|
}
|
|
|
|
$stmt->bind_param('iis', $job_id, $user_id, $status);
|
|
if (!$stmt->execute()) {
|
|
$stmt->close();
|
|
throw new Exception('Failed to dismiss queue job');
|
|
}
|
|
$stmt->close();
|
|
|
|
return $job;
|
|
}
|
|
|
|
public function dismissTerminalJobsForUser(int $user_id, ?string $transfer_type = null): int
|
|
{
|
|
global $db;
|
|
|
|
$user_id = max(0, $user_id);
|
|
if ($user_id < 1) {
|
|
return 0;
|
|
}
|
|
|
|
try {
|
|
$normalized_transfer_type = $transfer_type !== null && trim($transfer_type) !== ''
|
|
? $this->validateTransferType($transfer_type)
|
|
: null;
|
|
} catch (Exception) {
|
|
return 0;
|
|
}
|
|
|
|
$transfer_condition = '';
|
|
if ($normalized_transfer_type !== null) {
|
|
$transfer_condition = "AND q.transfer_type = '" . $db->escape_string($normalized_transfer_type) . "'";
|
|
}
|
|
|
|
$sql = "INSERT INTO economic_transfer_queue_job_dismissals (queue_job_id, user_id, dismissed_status, dismissed_at)
|
|
SELECT q.id, $user_id, q.status, NOW()
|
|
FROM economic_transfer_queue_jobs q
|
|
LEFT JOIN economic_transfer_queue_job_dismissals d
|
|
ON d.queue_job_id = q.id
|
|
AND d.user_id = $user_id
|
|
AND d.dismissed_status = q.status
|
|
WHERE " . $this->jobVisibilitySql('q', $user_id) . "
|
|
AND q.status IN ('" . self::STATUS_COMPLETED . "', '" . self::STATUS_FAILED . "')
|
|
$transfer_condition
|
|
AND d.queue_job_id IS NULL
|
|
ON DUPLICATE KEY UPDATE dismissed_status = VALUES(dismissed_status), dismissed_at = NOW()";
|
|
$db->query($sql);
|
|
return max(0, (int)($db->affected_rows ?? 0));
|
|
}
|
|
|
|
/**
|
|
* @throws Exception
|
|
*/
|
|
public function retryJob(int $job_id): array
|
|
{
|
|
return $this->retryJobInternal($job_id);
|
|
}
|
|
|
|
public function retryJobForUser(int $job_id, int $created_by): array
|
|
{
|
|
return $this->retryJobInternal($job_id, $created_by);
|
|
}
|
|
|
|
private function retryJobInternal(int $job_id, ?int $created_by = null): array
|
|
{
|
|
global $db;
|
|
|
|
$job_id = max(0, $job_id);
|
|
$created_by = $created_by === null ? null : max(0, $created_by);
|
|
$existing_job = $created_by === null
|
|
? $this->getJobById($job_id)
|
|
: $this->getJobByIdForUser($job_id, $created_by);
|
|
if ($existing_job === null) {
|
|
throw new Exception('Queue job not found');
|
|
}
|
|
if ($created_by !== null && (int)($existing_job['created_by'] ?? 0) !== $created_by) {
|
|
throw new Exception('Only the queue job creator can retry this job');
|
|
}
|
|
if ((string)($existing_job['status'] ?? '') !== self::STATUS_FAILED) {
|
|
throw new Exception('Only failed jobs can be retried');
|
|
}
|
|
if ((int)($existing_job['attempts'] ?? 0) >= (int)($existing_job['max_attempts'] ?? 1)) {
|
|
throw new Exception('Queue job reached max retry attempts');
|
|
}
|
|
|
|
$sql = "UPDATE economic_transfer_queue_jobs
|
|
SET status = ?, progress_percent = 0, progress_message = 'Queued for retry',
|
|
error_message = NULL, result_json = NULL, started_at = NULL, completed_at = NULL, locked_at = NULL
|
|
WHERE id = ? AND status = ?";
|
|
if ($created_by !== null) {
|
|
$sql .= " AND created_by = ?";
|
|
}
|
|
|
|
$stmt = $db->prepare($sql);
|
|
if (!$stmt) {
|
|
throw new Exception('Failed to prepare retry statement');
|
|
}
|
|
|
|
$queued = self::STATUS_QUEUED;
|
|
$failed = self::STATUS_FAILED;
|
|
if ($created_by !== null) {
|
|
$stmt->bind_param('sisi', $queued, $job_id, $failed, $created_by);
|
|
} else {
|
|
$stmt->bind_param('sis', $queued, $job_id, $failed);
|
|
}
|
|
$stmt->execute();
|
|
$affected = $stmt->affected_rows;
|
|
$stmt->close();
|
|
|
|
if ($affected < 1) {
|
|
throw new Exception('Failed to retry queue job');
|
|
}
|
|
|
|
$this->clearDismissalsForJob($job_id);
|
|
|
|
$job = $created_by === null
|
|
? $this->getJobById($job_id)
|
|
: $this->getJobByIdForUser($job_id, $created_by);
|
|
if ($job === null) {
|
|
throw new Exception('Retry updated job could not be loaded');
|
|
}
|
|
return $job;
|
|
}
|
|
|
|
public function processPending(int $limit = 5): array
|
|
{
|
|
return $this->processPendingInternal($limit);
|
|
}
|
|
|
|
/**
|
|
* @throws Exception
|
|
*/
|
|
public function processPendingByTransferType(string $transfer_type, int $limit = 10): array
|
|
{
|
|
return $this->processPendingInternal($limit, $this->validateTransferType($transfer_type));
|
|
}
|
|
|
|
private function processPendingInternal(int $limit = 5, ?string $transfer_type = null): array
|
|
{
|
|
$limit = max(1, min(100, $limit));
|
|
$this->releaseStaleProcessingLocks();
|
|
|
|
$processed = 0;
|
|
$completed = 0;
|
|
$failed = 0;
|
|
$jobs = [];
|
|
$empty_claims = 0;
|
|
|
|
for ($i = 0; $i < $limit; $i++) {
|
|
$job = $this->claimNextJob($transfer_type);
|
|
if ($job === null) {
|
|
$empty_claims++;
|
|
if ($empty_claims >= 3) {
|
|
break;
|
|
}
|
|
continue;
|
|
}
|
|
$empty_claims = 0;
|
|
|
|
$processed++;
|
|
$jobs[] = $job['id'];
|
|
$this->updateProgress((int)$job['id'], 15, 'Running transfer');
|
|
|
|
try {
|
|
$result = $this->executeJob($job);
|
|
$this->markCompleted((int)$job['id'], $result);
|
|
$completed++;
|
|
} catch (Exception $e) {
|
|
$this->markFailed((int)$job['id'], $e->getMessage());
|
|
$failed++;
|
|
} catch (\Throwable $e) {
|
|
$this->markFailed((int)$job['id'], $e->getMessage());
|
|
$failed++;
|
|
}
|
|
}
|
|
|
|
return [
|
|
'processed' => $processed,
|
|
'completed' => $completed,
|
|
'failed' => $failed,
|
|
'jobs' => $jobs,
|
|
];
|
|
}
|
|
|
|
private function claimNextJob(?string $transfer_type = null): ?array
|
|
{
|
|
global $db;
|
|
|
|
$sql = "SELECT id
|
|
FROM economic_transfer_queue_jobs
|
|
WHERE status = ?
|
|
AND attempts < max_attempts
|
|
AND (next_retry_at IS NULL OR next_retry_at <= NOW())";
|
|
if ($transfer_type !== null) {
|
|
$sql .= " AND transfer_type = ?";
|
|
}
|
|
$sql .= " ORDER BY id ASC LIMIT 1";
|
|
|
|
$stmt = $db->prepare($sql);
|
|
if (!$stmt) {
|
|
return null;
|
|
}
|
|
|
|
$queued = self::STATUS_QUEUED;
|
|
if ($transfer_type !== null) {
|
|
$stmt->bind_param('ss', $queued, $transfer_type);
|
|
} else {
|
|
$stmt->bind_param('s', $queued);
|
|
}
|
|
|
|
if (!$stmt->execute()) {
|
|
$stmt->close();
|
|
return null;
|
|
}
|
|
|
|
$result = $stmt->get_result();
|
|
$row = $result instanceof mysqli_result ? $result->fetch_assoc() : null;
|
|
$stmt->close();
|
|
if (!$row || !isset($row['id'])) {
|
|
return null;
|
|
}
|
|
|
|
$job_id = (int)$row['id'];
|
|
$stmt = $db->prepare(
|
|
"UPDATE economic_transfer_queue_jobs
|
|
SET status = ?, progress_percent = 5, progress_message = 'Processing', started_at = NOW(), locked_at = NOW()
|
|
WHERE id = ? AND status = ?"
|
|
);
|
|
if (!$stmt) {
|
|
return null;
|
|
}
|
|
|
|
$processing = self::STATUS_PROCESSING;
|
|
$stmt->bind_param('sis', $processing, $job_id, $queued);
|
|
$stmt->execute();
|
|
$affected = $stmt->affected_rows;
|
|
$stmt->close();
|
|
|
|
if ($affected < 1) {
|
|
return null;
|
|
}
|
|
|
|
return $this->getJobById($job_id);
|
|
}
|
|
|
|
/**
|
|
* @throws Exception
|
|
*/
|
|
private function executeJob(array $job): array
|
|
{
|
|
$payload = (array)($job['payload'] ?? []);
|
|
$transfer_type = (string)($job['transfer_type'] ?? '');
|
|
$requested_by = (int)($payload['requested_by'] ?? ($job['created_by'] ?? 0));
|
|
|
|
$this->updateProgress((int)$job['id'], 40, 'Validating job payload');
|
|
|
|
return match ($transfer_type) {
|
|
self::TYPE_ORDER_DRAFT_EXPORT => $this->executeOrderDraftExport($job, $payload, $requested_by),
|
|
self::TYPE_ORDER_INVOICE_EXPORT => $this->executeOrderInvoiceExport($job, $payload, $requested_by),
|
|
self::TYPE_COLLECTED_INVOICE_EXPORT => $this->executeCollectedInvoiceExport($job, $payload, $requested_by),
|
|
default => throw new Exception('Unsupported transfer type: ' . $transfer_type),
|
|
};
|
|
}
|
|
|
|
/**
|
|
* @throws Exception
|
|
*/
|
|
private function executeOrderDraftExport(array $job, array $payload, int $requested_by): array
|
|
{
|
|
$order_id = (int)($payload['order_id'] ?? 0);
|
|
if ($order_id < 1) {
|
|
throw new Exception('order_id is required');
|
|
}
|
|
$this->updateProgress((int)$job['id'], 65, 'Exporting order draft invoice');
|
|
return $this->executor->exportOrderDraftInvoice($order_id, $requested_by);
|
|
}
|
|
|
|
/**
|
|
* @throws Exception
|
|
*/
|
|
private function executeOrderInvoiceExport(array $job, array $payload, int $requested_by): array
|
|
{
|
|
$order_id = (int)($payload['order_id'] ?? 0);
|
|
if ($order_id < 1) {
|
|
throw new Exception('order_id is required');
|
|
}
|
|
$this->updateProgress((int)$job['id'], 65, 'Exporting booked invoice');
|
|
return $this->executor->exportOrderInvoice($order_id, $requested_by);
|
|
}
|
|
|
|
/**
|
|
* @throws Exception
|
|
*/
|
|
private function executeCollectedInvoiceExport(array $job, array $payload, int $requested_by): array
|
|
{
|
|
$collected_invoice_id = (int)($payload['collected_invoice_id'] ?? 0);
|
|
if ($collected_invoice_id < 1) {
|
|
throw new Exception('collected_invoice_id is required');
|
|
}
|
|
$collection_lock = $this->acquireCollectedInvoiceExportLock(
|
|
self::TYPE_COLLECTED_INVOICE_EXPORT,
|
|
$payload
|
|
);
|
|
$this->assertCollectedInvoiceExportIsStillEligible($payload);
|
|
$send_as_is = (bool)($payload['send_as_is'] ?? false);
|
|
$this->updateProgress((int)$job['id'], 65, 'Exporting collected invoice');
|
|
return $this->executor->exportCollectedInvoice($collected_invoice_id, $send_as_is, $requested_by);
|
|
}
|
|
|
|
/**
|
|
* Enqueue, worker execution, payments, and invoice-tree mutations share the
|
|
* same collection lock. The returned object intentionally stays in scope
|
|
* for the complete enqueue/export operation and releases in its destructor.
|
|
*
|
|
* @throws Exception
|
|
*/
|
|
private function acquireCollectedInvoiceExportLock(string $transfer_type, array $payload): ?order_payment_lock
|
|
{
|
|
if ($transfer_type !== self::TYPE_COLLECTED_INVOICE_EXPORT) {
|
|
return null;
|
|
}
|
|
$collected_invoice_id = (int)($payload['collected_invoice_id'] ?? 0);
|
|
if ($collected_invoice_id < 1) {
|
|
throw new Exception('collected_invoice_id is required');
|
|
}
|
|
$lock = order_payment_lock::tryAcquireInvoiceCollection($collected_invoice_id);
|
|
if ($lock === null) {
|
|
throw new Exception('Invoice collection is currently being changed or paid. Try again.');
|
|
}
|
|
return $lock;
|
|
}
|
|
|
|
/**
|
|
* Re-read eligibility after acquiring the collection lock so a queued job
|
|
* cannot export a collection that was booked or superseded while waiting.
|
|
*
|
|
* @throws Exception
|
|
*/
|
|
private function assertCollectedInvoiceExportIsStillEligible(array $payload): void
|
|
{
|
|
$collection = (new \objects\collected_order_invoices_o())->select(
|
|
(int)($payload['collected_invoice_id'] ?? 0)
|
|
);
|
|
invoice_collection_bulk_action_service::assertCollectionCanQueueEconomic($collection);
|
|
}
|
|
|
|
private function updateProgress(int $job_id, int $percent, string $message): void
|
|
{
|
|
global $db;
|
|
|
|
$percent = max(0, min(100, $percent));
|
|
$escaped_message = $db->escape_string($message);
|
|
$sql = "UPDATE economic_transfer_queue_jobs
|
|
SET progress_percent = $percent, progress_message = '$escaped_message'
|
|
WHERE id = $job_id";
|
|
$db->query($sql);
|
|
}
|
|
|
|
private function markCompleted(int $job_id, array $result): void
|
|
{
|
|
global $db;
|
|
|
|
$result_json = json_encode($result, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES);
|
|
if ($result_json === false) {
|
|
$result_json = json_encode(['result' => 'serialization_error']);
|
|
}
|
|
$escaped_result = $db->escape_string((string)$result_json);
|
|
|
|
$sql = "UPDATE economic_transfer_queue_jobs
|
|
SET status = '" . self::STATUS_COMPLETED . "',
|
|
progress_percent = 100,
|
|
progress_message = 'Completed',
|
|
result_json = '$escaped_result',
|
|
error_message = NULL,
|
|
completed_at = NOW(),
|
|
locked_at = NULL
|
|
WHERE id = $job_id";
|
|
$db->query($sql);
|
|
|
|
$this->logQueueEvent(
|
|
1,
|
|
0,
|
|
'ECONOMIC_TRANSFER_JOB_COMPLETED',
|
|
'Transfer job #' . $job_id . ' completed'
|
|
);
|
|
}
|
|
|
|
private function markFailed(int $job_id, string $error_message): void
|
|
{
|
|
global $db;
|
|
|
|
$error_message = trim($error_message);
|
|
if ($error_message === '') {
|
|
$error_message = 'Unknown transfer queue error';
|
|
}
|
|
$escaped_error = $db->escape_string($error_message);
|
|
|
|
$sql = "UPDATE economic_transfer_queue_jobs
|
|
SET status = '" . self::STATUS_FAILED . "',
|
|
progress_message = 'Failed',
|
|
error_message = '$escaped_error',
|
|
attempts = attempts + 1,
|
|
completed_at = NOW(),
|
|
locked_at = NULL
|
|
WHERE id = $job_id";
|
|
$db->query($sql);
|
|
|
|
$this->logQueueEvent(
|
|
3,
|
|
0,
|
|
'ECONOMIC_TRANSFER_JOB_FAILED',
|
|
'Transfer job #' . $job_id . ' failed: ' . $error_message
|
|
);
|
|
}
|
|
|
|
private function decodeJsonValue(mixed $json): mixed
|
|
{
|
|
if (!is_string($json) || trim($json) === '') {
|
|
return null;
|
|
}
|
|
$decoded = json_decode($json, true);
|
|
return json_last_error() === JSON_ERROR_NONE ? $decoded : null;
|
|
}
|
|
|
|
private function normalizeJobRow(array $row): array
|
|
{
|
|
return [
|
|
'id' => (int)$row['id'],
|
|
'transfer_type' => (string)($row['transfer_type'] ?? ''),
|
|
'status' => (string)($row['status'] ?? self::STATUS_QUEUED),
|
|
'progress_percent' => (int)($row['progress_percent'] ?? 0),
|
|
'progress_message' => $row['progress_message'] ?? null,
|
|
'attempts' => (int)($row['attempts'] ?? 0),
|
|
'max_attempts' => (int)($row['max_attempts'] ?? 0),
|
|
'error_message' => $row['error_message'] ?? null,
|
|
'payload' => $this->decodeJsonValue($row['payload_json'] ?? null),
|
|
'result' => $this->decodeJsonValue($row['result_json'] ?? null),
|
|
'created_by' => isset($row['created_by']) ? (int)$row['created_by'] : null,
|
|
'created_at' => $row['created_at'] ?? null,
|
|
'updated_at' => $row['updated_at'] ?? null,
|
|
'started_at' => $row['started_at'] ?? null,
|
|
'completed_at' => $row['completed_at'] ?? null,
|
|
'next_retry_at' => $row['next_retry_at'] ?? null,
|
|
];
|
|
}
|
|
|
|
private function clearDismissalsForJob(int $job_id): void
|
|
{
|
|
global $db;
|
|
|
|
$job_id = max(0, $job_id);
|
|
if ($job_id < 1) {
|
|
return;
|
|
}
|
|
|
|
$db->query("DELETE FROM economic_transfer_queue_job_dismissals WHERE queue_job_id = $job_id");
|
|
}
|
|
|
|
private function registerJobRequester(int $job_id, int $user_id): void
|
|
{
|
|
global $db;
|
|
|
|
if ($job_id < 1 || $user_id < 1) {
|
|
return;
|
|
}
|
|
$stmt = $db->prepare(
|
|
"INSERT INTO economic_transfer_queue_job_requesters (queue_job_id, user_id, requested_at)
|
|
VALUES (?, ?, NOW())
|
|
ON DUPLICATE KEY UPDATE requested_at = VALUES(requested_at)"
|
|
);
|
|
if (!$stmt) {
|
|
throw new Exception('Failed to prepare queue requester registration');
|
|
}
|
|
$stmt->bind_param('ii', $job_id, $user_id);
|
|
if (!$stmt->execute()) {
|
|
$stmt->close();
|
|
throw new Exception('Failed to register queue requester');
|
|
}
|
|
$stmt->close();
|
|
}
|
|
|
|
private function jobVisibilitySql(string $alias, int $user_id): string
|
|
{
|
|
$user_id = max(0, $user_id);
|
|
return "($alias.created_by = $user_id OR EXISTS (
|
|
SELECT 1 FROM economic_transfer_queue_job_requesters requester
|
|
WHERE requester.queue_job_id = $alias.id AND requester.user_id = $user_id
|
|
))";
|
|
}
|
|
|
|
/**
|
|
* Release jobs stuck in PROCESSING due to crashes or killed workers.
|
|
*/
|
|
private function releaseStaleProcessingLocks(): void
|
|
{
|
|
global $db;
|
|
|
|
$timeout = (int)self::STALE_PROCESSING_LOCK_SECONDS;
|
|
$sql = "UPDATE economic_transfer_queue_jobs
|
|
SET status = '" . self::STATUS_QUEUED . "',
|
|
progress_message = 'Re-queued after stale processing lock',
|
|
locked_at = NULL,
|
|
started_at = NULL
|
|
WHERE status = '" . self::STATUS_PROCESSING . "'
|
|
AND locked_at IS NOT NULL
|
|
AND locked_at < (NOW() - INTERVAL $timeout SECOND)";
|
|
$db->query($sql);
|
|
}
|
|
|
|
/**
|
|
* @throws Exception
|
|
*/
|
|
private function normalizePayloadForTransferType(string $transfer_type, array $payload, int $created_by): array
|
|
{
|
|
$normalized_payload = $payload;
|
|
if (isset($normalized_payload['requested_by'])) {
|
|
$normalized_payload['requested_by'] = max(0, (int)$normalized_payload['requested_by']);
|
|
}
|
|
|
|
if ($transfer_type === self::TYPE_COLLECTED_INVOICE_EXPORT) {
|
|
return $this->normalizeCollectedInvoicePayload($normalized_payload, $created_by);
|
|
}
|
|
if (!in_array($transfer_type, [self::TYPE_ORDER_DRAFT_EXPORT, self::TYPE_ORDER_INVOICE_EXPORT], true)) {
|
|
$this->rejectPayload($created_by, 'Unsupported transfer type payload: ' . $transfer_type);
|
|
}
|
|
|
|
return $this->normalizeOrderPayload($normalized_payload, $created_by);
|
|
}
|
|
|
|
/**
|
|
* @throws Exception
|
|
*/
|
|
private function normalizeOrderPayload(array $payload, int $created_by): array
|
|
{
|
|
$order_id = $payload['order_id'] ?? null;
|
|
if ($order_id === null || !is_numeric($order_id) || (int)$order_id < 1) {
|
|
$this->rejectPayload($created_by, 'order_id is required and must be a positive number');
|
|
}
|
|
$payload['order_id'] = (int)$order_id;
|
|
return $payload;
|
|
}
|
|
|
|
/**
|
|
* @throws Exception
|
|
*/
|
|
private function normalizeCollectedInvoicePayload(array $payload, int $created_by): array
|
|
{
|
|
$collected_invoice_id = $payload['collected_invoice_id'] ?? null;
|
|
if ($collected_invoice_id === null || !is_numeric($collected_invoice_id) || (int)$collected_invoice_id < 1) {
|
|
$this->rejectPayload($created_by, 'collected_invoice_id is required and must be a positive number');
|
|
}
|
|
|
|
$payload['collected_invoice_id'] = (int)$collected_invoice_id;
|
|
$payload['send_as_is'] = $this->normalizeBooleanPayloadValue($payload['send_as_is'] ?? false, 'send_as_is', $created_by);
|
|
return $payload;
|
|
}
|
|
|
|
/**
|
|
* @throws Exception
|
|
*/
|
|
private function normalizeBooleanPayloadValue(mixed $value, string $field_name, int $created_by): bool
|
|
{
|
|
if (is_bool($value)) {
|
|
return $value;
|
|
}
|
|
if (is_int($value) || is_float($value) || (is_string($value) && is_numeric($value))) {
|
|
$numeric = (int)$value;
|
|
if ($numeric === 0 || $numeric === 1) {
|
|
return $numeric === 1;
|
|
}
|
|
$this->rejectPayload($created_by, $field_name . ' must be a boolean');
|
|
}
|
|
if (is_string($value)) {
|
|
$normalized = strtolower(trim($value));
|
|
if (in_array($normalized, ['true', 'false', '1', '0'], true)) {
|
|
return in_array($normalized, ['true', '1'], true);
|
|
}
|
|
$this->rejectPayload($created_by, $field_name . ' must be a boolean');
|
|
}
|
|
|
|
$this->rejectPayload($created_by, $field_name . ' must be a boolean');
|
|
}
|
|
|
|
private function findActiveJobByTarget(string $transfer_type, array $payload, int $created_by): ?array
|
|
{
|
|
return match ($transfer_type) {
|
|
self::TYPE_ORDER_DRAFT_EXPORT, self::TYPE_ORDER_INVOICE_EXPORT => $this->findActiveJobByJsonNumericTarget(
|
|
$transfer_type,
|
|
'$.order_id',
|
|
(int)($payload['order_id'] ?? 0),
|
|
$created_by
|
|
),
|
|
self::TYPE_COLLECTED_INVOICE_EXPORT => $this->findActiveJobByJsonNumericTarget(
|
|
$transfer_type,
|
|
'$.collected_invoice_id',
|
|
(int)($payload['collected_invoice_id'] ?? 0),
|
|
$created_by
|
|
),
|
|
default => null,
|
|
};
|
|
}
|
|
|
|
private function findActiveJobByJsonNumericTarget(string $transfer_type, string $json_path, int $target_value, int $created_by): ?array
|
|
{
|
|
global $db;
|
|
|
|
// Active work is unique by transfer type and business target across all requesting users.
|
|
if ($target_value < 1) {
|
|
return null;
|
|
}
|
|
|
|
$stmt = $db->prepare(
|
|
"SELECT id
|
|
FROM economic_transfer_queue_jobs
|
|
WHERE transfer_type = ?
|
|
AND status IN (?, ?)
|
|
AND CAST(JSON_UNQUOTE(JSON_EXTRACT(payload_json, '$json_path')) AS UNSIGNED) = ?
|
|
ORDER BY id DESC
|
|
LIMIT 1"
|
|
);
|
|
if (!$stmt) {
|
|
return null;
|
|
}
|
|
|
|
$queued = self::STATUS_QUEUED;
|
|
$processing = self::STATUS_PROCESSING;
|
|
$stmt->bind_param('sssi', $transfer_type, $queued, $processing, $target_value);
|
|
if (!$stmt->execute()) {
|
|
$stmt->close();
|
|
return null;
|
|
}
|
|
|
|
$result = $stmt->get_result();
|
|
$row = $result instanceof mysqli_result ? $result->fetch_assoc() : null;
|
|
$stmt->close();
|
|
if (!$row || !isset($row['id'])) {
|
|
return null;
|
|
}
|
|
|
|
return $this->getJobById((int)$row['id']);
|
|
}
|
|
|
|
private function buildTargetLabel(string $transfer_type, array $payload): string
|
|
{
|
|
return match ($transfer_type) {
|
|
self::TYPE_ORDER_DRAFT_EXPORT, self::TYPE_ORDER_INVOICE_EXPORT => 'order_id=' . (int)($payload['order_id'] ?? 0),
|
|
self::TYPE_COLLECTED_INVOICE_EXPORT => 'collected_invoice_id=' . (int)($payload['collected_invoice_id'] ?? 0),
|
|
default => 'unknown',
|
|
};
|
|
}
|
|
|
|
/**
|
|
* @throws Exception
|
|
*/
|
|
private function rejectPayload(int $created_by, string $message): never
|
|
{
|
|
$this->logQueueEvent(
|
|
3,
|
|
max(0, $created_by),
|
|
'ECONOMIC_TRANSFER_JOB_VALIDATION_REJECTED',
|
|
'Transfer job enqueue rejected: ' . $message
|
|
);
|
|
throw new Exception($message);
|
|
}
|
|
|
|
private function logQueueEvent(int $status_code, int $user_id, string $event, string $message): void
|
|
{
|
|
try {
|
|
(new logs_o())->add(
|
|
'economic_transfer_queue',
|
|
'global',
|
|
$status_code,
|
|
$user_id,
|
|
$event,
|
|
$message
|
|
);
|
|
} catch (\Throwable) {
|
|
// Logging is best effort for queue operations.
|
|
}
|
|
}
|
|
|
|
/**
|
|
* @throws Exception
|
|
*/
|
|
private function validateTransferType(string $transfer_type): string
|
|
{
|
|
$transfer_type = strtoupper(trim($transfer_type));
|
|
if (!in_array($transfer_type, [
|
|
self::TYPE_ORDER_DRAFT_EXPORT,
|
|
self::TYPE_ORDER_INVOICE_EXPORT,
|
|
self::TYPE_COLLECTED_INVOICE_EXPORT,
|
|
], true)) {
|
|
throw new Exception('Unsupported transfer type: ' . $transfer_type);
|
|
}
|
|
return $transfer_type;
|
|
}
|
|
|
|
private function sanitizeStatuses(array $statuses): array
|
|
{
|
|
return array_values(array_unique(array_filter(array_map(static function ($status): string {
|
|
return strtoupper(trim((string)$status));
|
|
}, $statuses), static function ($status): bool {
|
|
return in_array($status, [
|
|
self::STATUS_QUEUED,
|
|
self::STATUS_PROCESSING,
|
|
self::STATUS_COMPLETED,
|
|
self::STATUS_FAILED,
|
|
], true);
|
|
})));
|
|
}
|
|
|
|
private function buildListJobsWhereClause(array $statuses = [], ?string $transfer_type = null): string
|
|
{
|
|
global $db;
|
|
|
|
$conditions = [];
|
|
|
|
$clean_statuses = $this->sanitizeStatuses($statuses);
|
|
if (!empty($clean_statuses)) {
|
|
$escaped_statuses = array_map(static function ($status) use ($db): string {
|
|
return "'" . $db->escape_string($status) . "'";
|
|
}, $clean_statuses);
|
|
$conditions[] = 'status IN (' . implode(',', $escaped_statuses) . ')';
|
|
}
|
|
|
|
if ($transfer_type !== null && trim($transfer_type) !== '') {
|
|
try {
|
|
$normalized_transfer_type = $this->validateTransferType($transfer_type);
|
|
} catch (Exception) {
|
|
return 'WHERE 1 = 0';
|
|
}
|
|
$conditions[] = "transfer_type = '" . $db->escape_string($normalized_transfer_type) . "'";
|
|
}
|
|
|
|
if (empty($conditions)) {
|
|
return '';
|
|
}
|
|
|
|
return 'WHERE ' . implode(' AND ', $conditions);
|
|
}
|
|
}
|