Add economic_transfer_executor and economic_transfer_queue classes for handling e-conomic invoice transfer logic, queue management, and processing. Include unit tests for Redis cache validation.
This commit is contained in:
Jeppe Bundgaard
@@ -0,0 +1,480 @@
|
||||
<?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;
|
||||
|
||||
$transfer_type = $this->validateTransferType($transfer_type);
|
||||
$max_attempts = max(1, min(10, $max_attempts));
|
||||
$created_by = max(0, $created_by);
|
||||
|
||||
$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();
|
||||
|
||||
(new logs_o())->add(
|
||||
'economic_transfer_queue',
|
||||
'global',
|
||||
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): array
|
||||
{
|
||||
global $db;
|
||||
|
||||
$limit = max(1, min(500, $limit));
|
||||
$offset = max(0, $offset);
|
||||
|
||||
$where = '';
|
||||
if (!empty($statuses)) {
|
||||
$clean_statuses = array_values(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);
|
||||
}));
|
||||
if (!empty($clean_statuses)) {
|
||||
$escaped = array_map(static function ($status) use ($db): string {
|
||||
return "'" . $db->escape_string($status) . "'";
|
||||
}, array_values(array_unique($clean_statuses)));
|
||||
$where = 'WHERE status IN (' . implode(',', $escaped) . ')';
|
||||
}
|
||||
}
|
||||
|
||||
$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;
|
||||
}
|
||||
|
||||
/**
|
||||
* @throws Exception
|
||||
*/
|
||||
public function retryJob(int $job_id): array
|
||||
{
|
||||
global $db;
|
||||
|
||||
$existing_job = $this->getJobById($job_id);
|
||||
if ($existing_job === null) {
|
||||
throw new Exception('Queue job not found');
|
||||
}
|
||||
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');
|
||||
}
|
||||
|
||||
$stmt = $db->prepare(
|
||||
"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 (!$stmt) {
|
||||
throw new Exception('Failed to prepare retry statement');
|
||||
}
|
||||
|
||||
$queued = self::STATUS_QUEUED;
|
||||
$failed = self::STATUS_FAILED;
|
||||
$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');
|
||||
}
|
||||
|
||||
$job = $this->getJobById($job_id);
|
||||
if ($job === null) {
|
||||
throw new Exception('Retry updated job could not be loaded');
|
||||
}
|
||||
return $job;
|
||||
}
|
||||
|
||||
public function processPending(int $limit = 5): 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();
|
||||
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(): ?array
|
||||
{
|
||||
global $db;
|
||||
|
||||
$sql = "SELECT id FROM economic_transfer_queue_jobs
|
||||
WHERE status = '" . self::STATUS_QUEUED . "'
|
||||
AND attempts < max_attempts
|
||||
AND (next_retry_at IS NULL OR next_retry_at <= NOW())
|
||||
ORDER BY id ASC
|
||||
LIMIT 1";
|
||||
$result = $db->query($sql);
|
||||
if (!$result instanceof mysqli_result) {
|
||||
return null;
|
||||
}
|
||||
|
||||
$row = $result->fetch_assoc();
|
||||
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;
|
||||
$queued = self::STATUS_QUEUED;
|
||||
$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');
|
||||
}
|
||||
$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);
|
||||
}
|
||||
|
||||
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);
|
||||
|
||||
(new logs_o())->add(
|
||||
'economic_transfer_queue',
|
||||
'global',
|
||||
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);
|
||||
|
||||
(new logs_o())->add(
|
||||
'economic_transfer_queue',
|
||||
'global',
|
||||
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,
|
||||
];
|
||||
}
|
||||
|
||||
/**
|
||||
* 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 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;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user