copenhagentruckwash/api
1
0
Fork 0
Code Issues Pull Requests 1 Actions 2 Packages Projects Releases Wiki Activity

Add unit and integration tests for collected invoice queue handling, route hardening, lifecycle validation, and manual batch processing logic.

Browse Source
This commit is contained in:
Jeppe Bundgaard
2026-04-08 15:53:22 +02:00
parent c5cd42be7f
commit 653680376a
26 changed files with 2974 additions and 428 deletions
Download Patch File Download Diff File Expand all files Collapse all files
services/nginx/app/classes/economic_transfer_queue.php
+106 -32
View File
@@ -114,33 +114,14 @@ class economic_transfer_queue
return $this->normalizeJobRow($row);
}
public function listJobs(array $statuses = [], int $limit = 50, int $offset = 0): array
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 = '';
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) . ')';
}
}
$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) {
@@ -154,6 +135,25 @@ class economic_transfer_queue
return $jobs;
}
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']);
}
/**
* @throws Exception
*/
@@ -201,6 +201,19 @@ class economic_transfer_queue
}
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();
@@ -212,7 +225,7 @@ class economic_transfer_queue
$empty_claims = 0;
for ($i = 0; $i < $limit; $i++) {
$job = $this->claimNextJob();
$job = $this->claimNextJob($transfer_type);
if ($job === null) {
$empty_claims++;
if ($empty_claims >= 3) {
@@ -247,22 +260,40 @@ class economic_transfer_queue
];
}
private function claimNextJob(): ?array
private function claimNextJob(?string $transfer_type = null): ?array
{
global $db;
$sql = "SELECT id FROM economic_transfer_queue_jobs
WHERE status = '" . self::STATUS_QUEUED . "'
$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())
ORDER BY id ASC
LIMIT 1";
$result = $db->query($sql);
if (!$result instanceof mysqli_result) {
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;
}
$row = $result->fetch_assoc();
$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;
}
@@ -278,7 +309,6 @@ class economic_transfer_queue
}
$processing = self::STATUS_PROCESSING;
$queued = self::STATUS_QUEUED;
$stmt->bind_param('sis', $processing, $job_id, $queued);
$stmt->execute();
$affected = $stmt->affected_rows;
@@ -650,4 +680,48 @@ class economic_transfer_queue
}
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);
}
}