api/services/nginx/app/tests/Integration/Invoicing/EconomicTransferQueueIntegrationTest.php

<?php

use classes\db;
use classes\economic_transfer_executor;
use classes\economic_transfer_queue;

app_require('classes/economic_transfer_executor.php');

if (!class_exists('EconomicTransferQueueIntegrationStubExecutor')) {
    class EconomicTransferQueueIntegrationStubExecutor extends economic_transfer_executor
    {
        /** @var array<int, bool> */
        private array $fail_once_order_draft = [];

        public function failNextOrderDraft(int $order_id): void
        {
            $this->fail_once_order_draft[$order_id] = true;
        }

        public function exportOrderDraftInvoice(int $order_id, int $user_id = 0): array
        {
            if (($this->fail_once_order_draft[$order_id] ?? false) === true) {
                unset($this->fail_once_order_draft[$order_id]);
                throw new Exception('Simulated draft export failure for order ' . $order_id);
            }

            return [
                'order_id' => $order_id,
                'user_id' => $user_id,
                'mode' => 'draft',
            ];
        }

        public function exportOrderInvoice(int $order_id, int $user_id = 0): array
        {
            return [
                'order_id' => $order_id,
                'user_id' => $user_id,
                'mode' => 'invoice',
            ];
        }

        public function exportCollectedInvoice(int $collected_invoice_id, bool $send_as_is = false, int $user_id = 0): array
        {
            return [
                'collected_invoice_id' => $collected_invoice_id,
                'send_as_is' => $send_as_is,
                'user_id' => $user_id,
                'mode' => 'collected',
            ];
        }
    }
}

function economic_transfer_queue_integration_db(): db
{
    if (!integration_enabled()) {
        test()->markTestSkipped('Set RUN_INTEGRATION_TESTS=1 to run DB integration tests.');
    }

    $host = getenv('CONFIG_DB_HOST') ?: null;
    $user = getenv('CONFIG_DB_USER') ?: null;
    $password = getenv('CONFIG_DB_PASSWORD') ?: '';
    $database = getenv('CONFIG_DB_DATABASE') ?: null;
    $port = (int)(getenv('CONFIG_DB_PORT') ?: 3306);
    if (!$host || !$user || !$database) {
        test()->markTestSkipped('Missing DB env vars: CONFIG_DB_HOST/CONFIG_DB_USER/CONFIG_DB_DATABASE.');
    }

    app_require('classes/db.php');
    app_require('classes/economic_transfer_executor.php');
    app_require('classes/economic_transfer_queue_schema_bootstrap.php');
    app_require('classes/economic_transfer_queue.php');

    $GLOBALS['response'] = new class {
        public function internal_server_error(string $message): void
        {
            throw new RuntimeException($message);
        }
    };

    $db = new db([
        'host' => $host,
        'user' => $user,
        'password' => $password,
        'database' => $database,
        'port' => $port,
    ]);
    try {
        $db->connect();
    } catch (Throwable $throwable) {
        test()->markTestSkipped('Integration DB unavailable: ' . $throwable->getMessage());
    }
    $GLOBALS['db'] = $db;
    return $db;
}

function economic_transfer_queue_cleanup_for_created_by(db $db, int $created_by): void
{
    $db->query(
        "DELETE requester FROM economic_transfer_queue_job_requesters requester
         LEFT JOIN economic_transfer_queue_jobs job ON job.id = requester.queue_job_id
         WHERE requester.user_id = $created_by OR job.created_by = $created_by"
    );
    $db->query("DELETE FROM economic_transfer_queue_jobs WHERE created_by = $created_by");
}

it('processes queued transfer jobs to completion', function (): void {
    $db = economic_transfer_queue_integration_db();
    $created_by = 920000 + random_int(1000, 9999);
    $order_id = 930000 + random_int(1000, 9999);
    $queue = new economic_transfer_queue(new EconomicTransferQueueIntegrationStubExecutor());

    try {
        economic_transfer_queue_cleanup_for_created_by($db, $created_by);

        $job = $queue->enqueue(
            economic_transfer_queue::TYPE_ORDER_DRAFT_EXPORT,
            [
                'order_id' => $order_id,
                'requested_by' => $created_by,
            ],
            $created_by
        );

        expect((int)$job['id'])->toBeGreaterThan(0);
        expect((string)$job['status'])->toBe(economic_transfer_queue::STATUS_QUEUED);

        $summary = $queue->processPending(1);
        expect((int)$summary['processed'])->toBe(1);
        expect((int)$summary['completed'])->toBe(1);
        expect((int)$summary['failed'])->toBe(0);

        $processed_job = $queue->getJobById((int)$job['id']);
        expect($processed_job)->not->toBeNull();
        expect((string)$processed_job['status'])->toBe(economic_transfer_queue::STATUS_COMPLETED);
        expect((int)($processed_job['result']['order_id'] ?? 0))->toBe($order_id);
    } finally {
        economic_transfer_queue_cleanup_for_created_by($db, $created_by);
        $db->close();
    }
});

it('deduplicates active jobs per transfer target', function (): void {
    $db = economic_transfer_queue_integration_db();
    $created_by = 921000 + random_int(1000, 9999);
    $collected_invoice_id = 931000 + random_int(1000, 9999);
    $queue = new economic_transfer_queue(new EconomicTransferQueueIntegrationStubExecutor());

    try {
        economic_transfer_queue_cleanup_for_created_by($db, $created_by);

        $first = $queue->enqueue(
            economic_transfer_queue::TYPE_COLLECTED_INVOICE_EXPORT,
            [
                'collected_invoice_id' => $collected_invoice_id,
                'send_as_is' => false,
                'requested_by' => $created_by,
            ],
            $created_by
        );

        $second = $queue->enqueue(
            economic_transfer_queue::TYPE_COLLECTED_INVOICE_EXPORT,
            [
                'collected_invoice_id' => $collected_invoice_id,
                'send_as_is' => true,
                'requested_by' => $created_by,
            ],
            $created_by
        );

        expect((int)$first['id'])->toBeGreaterThan(0);
        expect((int)$second['id'])->toBe((int)$first['id']);

        $row = $db->fetch_assoc($db->query(
            "SELECT COUNT(*) AS cnt
             FROM economic_transfer_queue_jobs
             WHERE created_by = $created_by
               AND transfer_type = '" . economic_transfer_queue::TYPE_COLLECTED_INVOICE_EXPORT . "'"
        ));
        expect((int)($row['cnt'] ?? 0))->toBe(1);
    } finally {
        economic_transfer_queue_cleanup_for_created_by($db, $created_by);
        $db->close();
    }
});

it('registers later deduplicated requesters for narrowly scoped queue monitoring', function (): void {
    $db = economic_transfer_queue_integration_db();
    $creator = 921200 + random_int(1000, 9999);
    $requester = 941200 + random_int(1000, 9999);
    $unrelated = 951200 + random_int(1000, 9999);
    $collected_invoice_id = 961200 + random_int(1000, 9999);
    $queue = new economic_transfer_queue(new EconomicTransferQueueIntegrationStubExecutor());

    try {
        economic_transfer_queue_cleanup_for_created_by($db, $creator);
        economic_transfer_queue_cleanup_for_created_by($db, $requester);

        $first = $queue->enqueue(
            economic_transfer_queue::TYPE_COLLECTED_INVOICE_EXPORT,
            ['collected_invoice_id' => $collected_invoice_id, 'requested_by' => $creator],
            $creator
        );
        $deduplicated = $queue->enqueue(
            economic_transfer_queue::TYPE_COLLECTED_INVOICE_EXPORT,
            ['collected_invoice_id' => $collected_invoice_id, 'requested_by' => $requester],
            $requester
        );

        expect((int)$deduplicated['id'])->toBe((int)$first['id'])
            ->and($queue->getJobByIdForUser((int)$first['id'], $requester))->not->toBeNull()
            ->and(array_column($queue->listJobsForCreatedBy([], 50, 0, null, $requester), 'id'))
            ->toContain((int)$first['id'])
            ->and(array_column($queue->listMonitorJobsForUser($requester), 'id'))
            ->toContain((int)$first['id'])
            ->and($queue->getJobByIdForUser((int)$first['id'], $unrelated))->toBeNull();
    } finally {
        economic_transfer_queue_cleanup_for_created_by($db, $requester);
        economic_transfer_queue_cleanup_for_created_by($db, $creator);
        $db->close();
    }
});

it('processes only collected-invoice jobs and respects the manual batch limit', function (): void {
    $db = economic_transfer_queue_integration_db();
    $created_by = 921500 + random_int(1000, 9999);
    $order_id = 932000 + random_int(1000, 9999);
    $first_collected_invoice_id = 932500 + random_int(1000, 9999);
    $second_collected_invoice_id = 933000 + random_int(1000, 9999);
    $queue = new economic_transfer_queue(new EconomicTransferQueueIntegrationStubExecutor());

    try {
        economic_transfer_queue_cleanup_for_created_by($db, $created_by);

        $order_job = $queue->enqueue(
            economic_transfer_queue::TYPE_ORDER_DRAFT_EXPORT,
            [
                'order_id' => $order_id,
                'requested_by' => $created_by,
            ],
            $created_by
        );

        $first_collected_job = $queue->enqueue(
            economic_transfer_queue::TYPE_COLLECTED_INVOICE_EXPORT,
            [
                'collected_invoice_id' => $first_collected_invoice_id,
                'send_as_is' => false,
                'requested_by' => $created_by,
            ],
            $created_by
        );

        $second_collected_job = $queue->enqueue(
            economic_transfer_queue::TYPE_COLLECTED_INVOICE_EXPORT,
            [
                'collected_invoice_id' => $second_collected_invoice_id,
                'send_as_is' => true,
                'requested_by' => $created_by,
            ],
            $created_by
        );

        $summary = $queue->processPendingByTransferType(
            economic_transfer_queue::TYPE_COLLECTED_INVOICE_EXPORT,
            1
        );

        expect((int)$summary['processed'])->toBe(1);
        expect((int)$summary['completed'])->toBe(1);
        expect((int)$summary['failed'])->toBe(0);
        expect($summary['jobs'])->toHaveCount(1);
        expect($summary['jobs'][0])->toBe((int)$first_collected_job['id']);

        $processed_collected_job = $queue->getJobById((int)$first_collected_job['id']);
        $queued_collected_job = $queue->getJobById((int)$second_collected_job['id']);
        $queued_order_job = $queue->getJobById((int)$order_job['id']);

        expect($processed_collected_job)->not->toBeNull();
        expect((string)$processed_collected_job['status'])->toBe(economic_transfer_queue::STATUS_COMPLETED);
        expect((int)($processed_collected_job['result']['collected_invoice_id'] ?? 0))->toBe($first_collected_invoice_id);

        expect($queued_collected_job)->not->toBeNull();
        expect((string)$queued_collected_job['status'])->toBe(economic_transfer_queue::STATUS_QUEUED);

        expect($queued_order_job)->not->toBeNull();
        expect((string)$queued_order_job['status'])->toBe(economic_transfer_queue::STATUS_QUEUED);
    } finally {
        economic_transfer_queue_cleanup_for_created_by($db, $created_by);
        $db->close();
    }
});

it('rejects invalid payloads without inserting queue rows', function (): void {
    $db = economic_transfer_queue_integration_db();
    $created_by = 922000 + random_int(1000, 9999);
    $queue = new economic_transfer_queue(new EconomicTransferQueueIntegrationStubExecutor());

    try {
        economic_transfer_queue_cleanup_for_created_by($db, $created_by);

        $before = $db->fetch_assoc($db->query(
            "SELECT COUNT(*) AS cnt
             FROM economic_transfer_queue_jobs
             WHERE created_by = $created_by"
        ));
        $before_count = (int)($before['cnt'] ?? 0);

        expect(fn () => $queue->enqueue(
            economic_transfer_queue::TYPE_ORDER_DRAFT_EXPORT,
            [
                'order_id' => 0,
                'requested_by' => $created_by,
            ],
            $created_by
        ))->toThrow(Exception::class, 'order_id is required and must be a positive number');

        $after = $db->fetch_assoc($db->query(
            "SELECT COUNT(*) AS cnt
             FROM economic_transfer_queue_jobs
             WHERE created_by = $created_by"
        ));
        expect((int)($after['cnt'] ?? 0))->toBe($before_count);
    } finally {
        economic_transfer_queue_cleanup_for_created_by($db, $created_by);
        $db->close();
    }
});

it('supports fail retry and reprocess lifecycle transitions', function (): void {
    $db = economic_transfer_queue_integration_db();
    $created_by = 923000 + random_int(1000, 9999);
    $order_id = 933000 + random_int(1000, 9999);
    $executor = new EconomicTransferQueueIntegrationStubExecutor();
    $executor->failNextOrderDraft($order_id);
    $queue = new economic_transfer_queue($executor);

    try {
        economic_transfer_queue_cleanup_for_created_by($db, $created_by);

        $job = $queue->enqueue(
            economic_transfer_queue::TYPE_ORDER_DRAFT_EXPORT,
            [
                'order_id' => $order_id,
                'requested_by' => $created_by,
            ],
            $created_by
        );
        $job_id = (int)$job['id'];
        expect($job_id)->toBeGreaterThan(0);

        $first = $queue->processPending(1);
        expect((int)$first['processed'])->toBe(1);
        expect((int)$first['failed'])->toBe(1);

        $failed_job = $queue->getJobById($job_id);
        expect($failed_job)->not->toBeNull();
        expect((string)$failed_job['status'])->toBe(economic_transfer_queue::STATUS_FAILED);
        expect((int)$failed_job['attempts'])->toBe(1);

        $retried = $queue->retryJob($job_id);
        expect((string)$retried['status'])->toBe(economic_transfer_queue::STATUS_QUEUED);

        $second = $queue->processPending(1);
        expect((int)$second['processed'])->toBe(1);
        expect((int)$second['completed'])->toBe(1);

        $completed_job = $queue->getJobById($job_id);
        expect($completed_job)->not->toBeNull();
        expect((string)$completed_job['status'])->toBe(economic_transfer_queue::STATUS_COMPLETED);
        expect((int)($completed_job['result']['order_id'] ?? 0))->toBe($order_id);
    } finally {
        economic_transfer_queue_cleanup_for_created_by($db, $created_by);
        $db->close();
    }
});