Scope economic transfer queue jobs by creator

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(
+        $sql = "UPDATE economic_transfer_queue_jobs
-            "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;

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);
             }

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')) {
+        $created_by = max(0, $created_by);
-            return max(0, (int)$queue->countJobs(
+        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 !== []) {

services/nginx/app/tests/Unit/Invoicing/EconomicTransferQueueAccessScopeTest.php

@@ -0,0 +1,24 @@
+<?php
+
+it('scopes economic transfer queue route reads and retries to the current user', function (): void {
+    $economic_invoice_route = file_get_contents(app_path('routes/economicInvoiceRoute.php'));
+    $collected_invoice_route = file_get_contents(app_path('routes/orderInvoicesRoute.php'));
+    $queue = file_get_contents(app_path('classes/economic_transfer_queue.php'));
+
+    expect($economic_invoice_route)->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 = ?');
+});

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');