From ce8e6d0dab798c976978cc005baa61163e034246 Mon Sep 17 00:00:00 2001 From: Jeppe B <2jepp9350@gmail.com> Date: Tue, 2 Jun 2026 00:20:30 +0200 Subject: [PATCH] Scope economic transfer queue jobs by creator --- .../app/classes/economic_transfer_queue.php | 145 ++++++++++++++++-- .../nginx/app/routes/economicInvoiceRoute.php | 12 +- .../nginx/app/routes/orderInvoicesRoute.php | 34 ++-- .../EconomicTransferQueueAccessScopeTest.php | 24 +++ .../EconomicTransferQueueHardeningTest.php | 8 +- 5 files changed, 185 insertions(+), 38 deletions(-) create mode 100644 services/nginx/app/tests/Unit/Invoicing/EconomicTransferQueueAccessScopeTest.php diff --git a/services/nginx/app/classes/economic_transfer_queue.php b/services/nginx/app/classes/economic_transfer_queue.php index 48392921..c0ccc953 100644 --- a/services/nginx/app/classes/economic_transfer_queue.php +++ b/services/nginx/app/classes/economic_transfer_queue.php @@ -41,7 +41,7 @@ class economic_transfer_queue $transfer_type = $this->validateTransferType($transfer_type); $payload = $this->normalizePayloadForTransferType($transfer_type, $payload, $created_by); - $active_job = $this->findActiveJobByTarget($transfer_type, $payload); + $active_job = $this->findActiveJobByTarget($transfer_type, $payload, $created_by); if ($active_job !== null) { $target_label = $this->buildTargetLabel($transfer_type, $payload); $this->logQueueEvent( @@ -135,6 +135,89 @@ class economic_transfer_queue 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 * FROM economic_transfer_queue_jobs WHERE id = ? AND created_by = ? LIMIT 1"); + if (!$stmt) { + return null; + } + + $stmt->bind_param('ii', $job_id, $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); + $where .= $where === '' ? 'WHERE created_by = ' . $created_by : ' AND created_by = ' . $created_by; + $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); + $where .= $where === '' ? 'WHERE created_by = ' . $created_by : ' AND created_by = ' . $created_by; + $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; @@ -179,7 +262,7 @@ class economic_transfer_queue ON d.queue_job_id = q.id AND d.user_id = $user_id AND d.dismissed_status = q.status - WHERE 1 = 1 + WHERE q.created_by = $user_id $transfer_condition AND ( q.status IN ('" . self::STATUS_QUEUED . "', '" . self::STATUS_PROCESSING . "') @@ -214,7 +297,7 @@ class economic_transfer_queue throw new Exception('Queue job and user are required'); } - $job = $this->getJobById($job_id); + $job = $this->getJobByIdForUser($job_id, $user_id); if ($job === null) { throw new Exception('Queue job not found'); } @@ -272,7 +355,8 @@ class economic_transfer_queue ON d.queue_job_id = q.id AND d.user_id = $user_id AND d.dismissed_status = q.status - WHERE q.status IN ('" . self::STATUS_COMPLETED . "', '" . self::STATUS_FAILED . "') + WHERE q.created_by = $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()"; @@ -284,10 +368,24 @@ class economic_transfer_queue * @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; - $existing_job = $this->getJobById($job_id); + $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'); } @@ -298,19 +396,26 @@ class economic_transfer_queue throw new Exception('Queue job reached max retry attempts'); } - $stmt = $db->prepare( - "UPDATE economic_transfer_queue_jobs + $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 = ?" - ); + 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; - $stmt->bind_param('sis', $queued, $job_id, $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(); @@ -321,7 +426,9 @@ class economic_transfer_queue $this->clearDismissalsForJob($job_id); - $job = $this->getJobById($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'); } @@ -710,28 +817,31 @@ class economic_transfer_queue return $this->rejectPayload($created_by, $field_name . ' must be a boolean'); } - private function findActiveJobByTarget(string $transfer_type, array $payload): ?array + 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) + (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) + (int)($payload['collected_invoice_id'] ?? 0), + $created_by ), default => null, }; } - private function findActiveJobByJsonNumericTarget(string $transfer_type, string $json_path, int $target_value): ?array + private function findActiveJobByJsonNumericTarget(string $transfer_type, string $json_path, int $target_value, int $created_by): ?array { global $db; - if ($target_value < 1) { + $created_by = max(0, $created_by); + if ($target_value < 1 || $created_by < 1) { return null; } @@ -741,6 +851,7 @@ class economic_transfer_queue WHERE transfer_type = ? AND status IN (?, ?) AND CAST(JSON_UNQUOTE(JSON_EXTRACT(payload_json, '$json_path')) AS UNSIGNED) = ? + AND created_by = ? ORDER BY id DESC LIMIT 1" ); @@ -750,7 +861,7 @@ class economic_transfer_queue $queued = self::STATUS_QUEUED; $processing = self::STATUS_PROCESSING; - $stmt->bind_param('sssi', $transfer_type, $queued, $processing, $target_value); + $stmt->bind_param('sssii', $transfer_type, $queued, $processing, $target_value, $created_by); if (!$stmt->execute()) { $stmt->close(); return null; diff --git a/services/nginx/app/routes/economicInvoiceRoute.php b/services/nginx/app/routes/economicInvoiceRoute.php index 4997e818..76bda89f 100644 --- a/services/nginx/app/routes/economicInvoiceRoute.php +++ b/services/nginx/app/routes/economicInvoiceRoute.php @@ -239,7 +239,7 @@ class economicInvoiceRoute } $queue = new economic_transfer_queue(); - $job = $queue->getJobById((int)$job_id); + $job = $queue->getJobByIdForUser((int)$job_id, (int)$user->id); if ($job === null || ($job['transfer_type'] ?? null) !== economic_transfer_queue::TYPE_ORDER_DRAFT_EXPORT) { $response->error('Draft export queue job not found', 404); } @@ -264,13 +264,13 @@ class economicInvoiceRoute } $queue = new economic_transfer_queue(); - $existing_job = $queue->getJobById((int)$job_id); + $existing_job = $queue->getJobByIdForUser((int)$job_id, (int)$user->id); if ($existing_job === null || ($existing_job['transfer_type'] ?? null) !== economic_transfer_queue::TYPE_ORDER_DRAFT_EXPORT) { $response->error('Draft export queue job not found', 404); } try { - $job = $queue->retryJob((int)$job_id); + $job = $queue->retryJobForUser((int)$job_id, (int)$user->id); } catch (\Throwable $e) { $response->error($e->getMessage(), 400); } @@ -298,7 +298,7 @@ class economicInvoiceRoute } $queue = new economic_transfer_queue(); - $job = $queue->getJobById((int)$job_id); + $job = $queue->getJobByIdForUser((int)$job_id, (int)$user->id); if ($job === null || ($job['transfer_type'] ?? null) !== economic_transfer_queue::TYPE_ORDER_INVOICE_EXPORT) { $response->error('Invoice export queue job not found', 404); } @@ -323,13 +323,13 @@ class economicInvoiceRoute } $queue = new economic_transfer_queue(); - $existing_job = $queue->getJobById((int)$job_id); + $existing_job = $queue->getJobByIdForUser((int)$job_id, (int)$user->id); if ($existing_job === null || ($existing_job['transfer_type'] ?? null) !== economic_transfer_queue::TYPE_ORDER_INVOICE_EXPORT) { $response->error('Invoice export queue job not found', 404); } try { - $job = $queue->retryJob((int)$job_id); + $job = $queue->retryJobForUser((int)$job_id, (int)$user->id); } catch (\Throwable $e) { $response->error($e->getMessage(), 400); } diff --git a/services/nginx/app/routes/orderInvoicesRoute.php b/services/nginx/app/routes/orderInvoicesRoute.php index 01da8864..5cb99763 100644 --- a/services/nginx/app/routes/orderInvoicesRoute.php +++ b/services/nginx/app/routes/orderInvoicesRoute.php @@ -752,13 +752,14 @@ class orderInvoicesRoute ['limit' => $limit, 'offset' => $offset] = $this->parseCollectedInvoiceQueuePagination(); $queue = new economic_transfer_queue(); - $jobs = $queue->listJobs( + $jobs = $queue->listJobsForCreatedBy( $statuses, $limit, $offset, - economic_transfer_queue::TYPE_COLLECTED_INVOICE_EXPORT + economic_transfer_queue::TYPE_COLLECTED_INVOICE_EXPORT, + (int)$user->id ); - $total_jobs = $this->countCollectedInvoiceQueueJobs($queue, $statuses); + $total_jobs = $this->countCollectedInvoiceQueueJobs($queue, $statuses, (int)$user->id); $has_more = ($offset + count($jobs)) < $total_jobs; $response->success([ @@ -785,7 +786,7 @@ class orderInvoicesRoute $this->ensureEconomicTransferQueueIsAvailable(); $job_id = $this->requireCollectedInvoiceQueueJobId(); - $job = $this->requireCollectedInvoiceQueueJobById($job_id); + $job = $this->requireCollectedInvoiceQueueJobById($job_id, (int)$user->id); $response->success($this->withCollectedInvoiceQueueDetailsSummary($job)); }, @@ -827,14 +828,14 @@ class orderInvoicesRoute $this->ensureEconomicTransferQueueIsAvailable(); $job_id = $this->requireCollectedInvoiceQueueJobId(); - $job = $this->requireCollectedInvoiceQueueJobById($job_id, true); + $job = $this->requireCollectedInvoiceQueueJobById($job_id, (int)$user->id, true); if ((int)($job['attempts'] ?? 0) >= (int)($job['max_attempts'] ?? 1)) { $response->error('Collected invoice queue job reached max retry attempts', 409); } try { $queue = new economic_transfer_queue(); - $retried = $queue->retryJob($job_id); + $retried = $queue->retryJobForUser($job_id, (int)$user->id); } catch (\Throwable $e) { $message = trim((string)$e->getMessage()); $status_code = $this->resolveCollectedInvoiceQueueRetryErrorStatus($message); @@ -861,7 +862,7 @@ class orderInvoicesRoute $this->ensureEconomicTransferQueueIsAvailable(); $job_id = $this->requireCollectedInvoiceQueueJobId(); - $job = $this->requireCollectedInvoiceQueueJobById($job_id); + $job = $this->requireCollectedInvoiceQueueJobById($job_id, (int)$user->id); $status = strtoupper((string)($job['status'] ?? '')); if (!in_array($status, [ economic_transfer_queue::STATUS_COMPLETED, @@ -2153,12 +2154,12 @@ class orderInvoicesRoute return (int)$job_id; } - private function requireCollectedInvoiceQueueJobById(int $job_id, bool $mustBeFailed = false): array + private function requireCollectedInvoiceQueueJobById(int $job_id, int $created_by, bool $mustBeFailed = false): array { global $response; $queue = new economic_transfer_queue(); - $job = $queue->getJobById($job_id); + $job = $queue->getJobByIdForUser($job_id, $created_by); if ($job === null || ($job['transfer_type'] ?? null) !== economic_transfer_queue::TYPE_COLLECTED_INVOICE_EXPORT) { $response->error('Collected invoice queue job not found', 404); } @@ -2263,19 +2264,26 @@ class orderInvoicesRoute ]; } - private function countCollectedInvoiceQueueJobs(economic_transfer_queue $queue, array $statuses): int + private function countCollectedInvoiceQueueJobs(economic_transfer_queue $queue, array $statuses, int $created_by): int { global $db; - if (method_exists($queue, 'countJobs')) { - return max(0, (int)$queue->countJobs( + $created_by = max(0, $created_by); + if ($created_by < 1) { + return 0; + } + + if (method_exists($queue, 'countJobsForCreatedBy')) { + return max(0, (int)$queue->countJobsForCreatedBy( $statuses, - economic_transfer_queue::TYPE_COLLECTED_INVOICE_EXPORT + economic_transfer_queue::TYPE_COLLECTED_INVOICE_EXPORT, + $created_by )); } $conditions = [ "transfer_type = '" . $db->escape_string(economic_transfer_queue::TYPE_COLLECTED_INVOICE_EXPORT) . "'", + 'created_by = ' . $created_by, ]; if ($statuses !== []) { diff --git a/services/nginx/app/tests/Unit/Invoicing/EconomicTransferQueueAccessScopeTest.php b/services/nginx/app/tests/Unit/Invoicing/EconomicTransferQueueAccessScopeTest.php new file mode 100644 index 00000000..07d76eb9 --- /dev/null +++ b/services/nginx/app/tests/Unit/Invoicing/EconomicTransferQueueAccessScopeTest.php @@ -0,0 +1,24 @@ +not->toBeFalse(); + expect($collected_invoice_route)->not->toBeFalse(); + expect($queue)->not->toBeFalse(); + + expect($economic_invoice_route)->toContain('$queue->getJobByIdForUser((int)$job_id, (int)$user->id)'); + expect($economic_invoice_route)->toContain('$queue->retryJobForUser((int)$job_id, (int)$user->id)'); + + expect($collected_invoice_route)->toContain('$queue->listJobsForCreatedBy('); + expect($collected_invoice_route)->toContain('$this->countCollectedInvoiceQueueJobs($queue, $statuses, (int)$user->id)'); + expect($collected_invoice_route)->toContain('private function requireCollectedInvoiceQueueJobById(int $job_id, int $created_by, bool $mustBeFailed = false): array'); + expect($collected_invoice_route)->toContain('$queue->getJobByIdForUser($job_id, $created_by)'); + expect($collected_invoice_route)->toContain('$queue->retryJobForUser($job_id, (int)$user->id)'); + + expect($queue)->toContain('SELECT * FROM economic_transfer_queue_jobs WHERE id = ? AND created_by = ? LIMIT 1'); + expect($queue)->toContain('WHERE q.created_by = $user_id'); + expect($queue)->toContain('AND created_by = ?'); +}); diff --git a/services/nginx/app/tests/Unit/Invoicing/EconomicTransferQueueHardeningTest.php b/services/nginx/app/tests/Unit/Invoicing/EconomicTransferQueueHardeningTest.php index f94bb759..715a90f5 100644 --- a/services/nginx/app/tests/Unit/Invoicing/EconomicTransferQueueHardeningTest.php +++ b/services/nginx/app/tests/Unit/Invoicing/EconomicTransferQueueHardeningTest.php @@ -7,16 +7,20 @@ it('hardens transfer queue with type validation retry caps and stale lock recove expect($content)->toContain('private const STALE_PROCESSING_LOCK_SECONDS = 900'); expect($content)->toContain('$this->validateTransferType($transfer_type)'); expect($content)->toContain('$this->normalizePayloadForTransferType($transfer_type, $payload, $created_by)'); - expect($content)->toContain('$this->findActiveJobByTarget($transfer_type, $payload)'); + expect($content)->toContain('$this->findActiveJobByTarget($transfer_type, $payload, $created_by)'); expect($content)->toContain('$max_attempts = max(1, min(10, $max_attempts));'); expect($content)->toContain('private function normalizePayloadForTransferType(string $transfer_type, array $payload, int $created_by): array'); expect($content)->toContain('private function normalizeBooleanPayloadValue(mixed $value, string $field_name, int $created_by): bool'); - expect($content)->toContain('private function findActiveJobByJsonNumericTarget(string $transfer_type, string $json_path, int $target_value): ?array'); + expect($content)->toContain('private function findActiveJobByJsonNumericTarget(string $transfer_type, string $json_path, int $target_value, int $created_by): ?array'); expect($content)->toContain('ECONOMIC_TRANSFER_JOB_DEDUPED'); expect($content)->toContain('ECONOMIC_TRANSFER_JOB_VALIDATION_REJECTED'); expect($content)->toContain('Queue job reached max retry attempts'); expect($content)->toContain('AND attempts < max_attempts'); expect($content)->toContain('public function listJobs(array $statuses = [], int $limit = 50, int $offset = 0, ?string $transfer_type = null): array'); + expect($content)->toContain('public function getJobByIdForUser(int $job_id, int $created_by): ?array'); + expect($content)->toContain('public function listJobsForCreatedBy(array $statuses = [], int $limit = 50, int $offset = 0, ?string $transfer_type = null, int $created_by = 0): array'); + expect($content)->toContain('public function countJobsForCreatedBy(array $statuses = [], ?string $transfer_type = null, int $created_by = 0): int'); + expect($content)->toContain('public function retryJobForUser(int $job_id, int $created_by): array'); expect($content)->toContain('public function countJobs(array $statuses = [], ?string $transfer_type = null): int'); expect($content)->toContain('public function listMonitorJobsForUser(int $user_id, int $limit = 50, ?string $transfer_type = null): array'); expect($content)->toContain('public function dismissTerminalJobForUser(int $job_id, int $user_id): array');