bot-28/tests/Integration/SharingMysqlConcurrencyTest.php

<?php

namespace Tests\Integration;

use App\Services\Sharing\ShareSettlementService;
use Illuminate\Support\Facades\DB;
use PHPUnit\Framework\TestCase;
use Tests\Fixtures\SharingMysqlSupport;

require_once dirname(__DIR__) . '/fixtures/sharing_mysql_support.php';

/**
 * Opt-in, real InnoDB row-lock tests. Never loads either project's .env or connects over TCP.
 * SHARING_MYSQL_SOCKET must belong to a new --skip-networking server under the guarded tmp prefix.
 * The runner starts and stops its own server; this suite owns only randomly named test databases.
 */
class SharingMysqlConcurrencyTest extends TestCase
{
    private ?\PDO $server = null;
    private string $socket;
    private string $database;
    private string $runDir;
    private array $processes = [];

    protected function setUp(): void
    {
        $socket = getenv('SHARING_MYSQL_SOCKET');
        if (!$socket) $this->markTestSkipped('Set SHARING_MYSQL_SOCKET to a new isolated, socket-only MySQL fixture; no existing database is allowed');
        if (!in_array('mysql', \PDO::getAvailableDrivers(), true)) $this->markTestSkipped('The isolated PHP runtime has no pdo_mysql driver');
        $this->socket = $socket;
        $this->server = SharingMysqlSupport::guardedPdo($socket);
        $this->database = 'share_concurrency_' . bin2hex(random_bytes(6));
        $this->server->exec('CREATE DATABASE `' . $this->database . '` CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci');
        $this->runDir = dirname($socket) . '/workers-' . bin2hex(random_bytes(6));
        mkdir($this->runDir, 0700);
        SharingMysqlSupport::laravel($socket, $this->database);
        SharingMysqlSupport::schema();
        SharingMysqlSupport::seed();
    }

    protected function tearDown(): void
    {
        foreach ($this->processes as $worker) {
            if (is_resource($worker['process'])) {
                if (proc_get_status($worker['process'])['running']) proc_terminate($worker['process']);
                foreach ($worker['pipes'] as $pipe) if (is_resource($pipe)) fclose($pipe);
                proc_close($worker['process']);
            }
        }
        $this->processes = [];
        if ($this->server !== null) {
            DB::disconnect();
            $this->server->exec('DROP DATABASE IF EXISTS `' . $this->database . '`');
            foreach (glob($this->runDir . '/*') ?: [] as $file) unlink($file);
            rmdir($this->runDir);
            $this->server = null;
            \Carbon\Carbon::setTestNow();
            \Illuminate\Support\Facades\Facade::clearResolvedInstances();
        }
        parent::tearDown();
    }

    public function test_same_tp_request_in_two_processes_reserves_once_and_returns_the_same_claim(): void
    {
        [$a, $b] = $this->race(['framework' => 'tp', 'request_id' => 'same-request'], ['framework' => 'tp', 'request_id' => 'same-request']);
        $this->assertTrue($a['ok']); $this->assertTrue($b['ok']);
        $this->assertSame($a['result']['id'], $b['result']['id']);
        $this->assertSame('10.01', $a['result']['amount']);
        $this->assertSame(1, DB::table('share_claims')->count());
        $this->assertSame('10.0100000000', $this->amount('reserved_amount'));
        $this->assertSame('20.0000000000', $this->wallet());
    }

    public function test_different_tp_requests_in_two_processes_leave_only_one_pending_claim(): void
    {
        [$a, $b] = $this->race(['framework' => 'tp', 'request_id' => 'request-a'], ['framework' => 'tp', 'request_id' => 'request-b']);
        $this->assertTrue($a['ok']); $this->assertFalse($b['ok']);
        $this->assertStringContainsString('已有待审核', $b['message']);
        $this->assertSame(1, DB::table('share_claims')->where('active_user_id', 1)->count());
        $this->assertSame('10.0100000000', $this->amount('reserved_amount'));
        $this->assertSame('0.0000000000', $this->amount('paid_amount'));
    }

    public function test_concurrent_laravel_approvals_credit_one_wallet_log_and_audit(): void
    {
        $claim = $this->single(['framework' => 'tp', 'request_id' => 'approve-once']);
        $this->assertTrue($claim['ok']);
        $args = ['framework' => 'laravel', 'action' => 'audit', 'claim_id' => $claim['result']['id']];
        [$a, $b] = $this->race($args, $args);
        $this->assertTrue($a['ok']); $this->assertTrue($b['ok']);
        $this->assertSame('approved', $a['result']['status']);
        $this->assertSame($a['result']['credited_log_id'], $b['result']['credited_log_id']);
        $this->assertSame('30.0100000000', $this->wallet());
        $this->assertSame('10.0100000000', $this->amount('paid_amount'));
        $this->assertSame('0.0000000000', $this->amount('reserved_amount'));
        $this->assertSame(1, DB::table('balance_logs')->where('change_type', '分享佣金领取')->count());
        $this->assertSame(1, DB::table('operation_audits')->where('resource_type', 'share_claim')->count());
    }

    public function test_correction_holding_account_blocks_approval_and_preserves_the_original_pending_request(): void
    {
        $id = $this->settledClaim();
        DB::table('share_bet_facts')->update(['effective_amount' => '50', 'net_profit' => '-50']);
        [$corrected, $approval] = $this->race(['framework' => 'laravel', 'action' => 'settle'], ['framework' => 'laravel', 'action' => 'audit', 'claim_id' => $id]);
        $this->assertTrue($corrected['ok']); $this->assertFalse($approval['ok']);
        $this->assertStringContainsString('重新申请', $approval['message']);
        $this->assertSame('5.0000000000', $this->amount('earned_amount'));
        $this->assertSame('0.0000000000', $this->amount('paid_amount'));
        $this->assertSame('10.0000000000', $this->amount('reserved_amount'));
        $this->assertSame('pending', DB::table('share_claims')->where('id', $id)->value('status'));
        $this->assertSame('20.0000000000', $this->wallet());
        $this->assertSame(0, DB::table('balance_logs')->where('change_type', '分享佣金领取')->count());
        $before = DB::table('share_ledger')->count();
        $this->assertSame(0, (new ShareSettlementService())->settle('2026-10-24')['changed']);
        $this->assertSame($before, DB::table('share_ledger')->count());
    }

    public function test_approval_holding_account_credits_once_then_correction_becomes_debt_without_wallet_debit(): void
    {
        $id = $this->settledClaim();
        DB::table('share_bet_facts')->update(['effective_amount' => '50', 'net_profit' => '-50']);
        [$approval, $correction] = $this->race(['framework' => 'laravel', 'action' => 'audit', 'claim_id' => $id], ['framework' => 'laravel', 'action' => 'settle']);
        $this->assertTrue($approval['ok']); $this->assertTrue($correction['ok']);
        $this->assertSame('30.0000000000', $this->wallet());
        $this->assertSame('10.0000000000', $this->amount('paid_amount'));
        $this->assertSame('5.0000000000', $this->amount('earned_amount'));
        $this->assertSame('0.0000000000', $this->amount('reserved_amount'));
        $this->assertSame('approved', DB::table('share_claims')->where('id', $id)->value('status'));
        $this->assertSame(1, DB::table('balance_logs')->where('change_type', '分享佣金领取')->count());
        $this->assertSame(2, DB::table('share_ledger')->count());
        $this->assertSame(0, (new ShareSettlementService())->settle('2026-10-24')['changed']);
        $this->assertSame('30.0000000000', $this->wallet());
    }

    public function test_two_settlement_processes_serialize_one_revision_and_one_ledger_delta(): void
    {
        DB::table('share_accounts')->where('user_id', 1)->update(['earned_amount' => 0]);
        $args = ['framework' => 'laravel', 'action' => 'settle'];
        [$a, $b] = $this->race($args, $args);
        $this->assertTrue($a['ok']); $this->assertTrue($b['ok']);
        $this->assertSame(1, $a['result']['changed']); $this->assertSame(0, $b['result']['changed']);
        $this->assertSame('10.0000000000', $this->amount('earned_amount'));
        $this->assertSame(1, DB::table('share_ledger')->count());
        $this->assertSame(1, DB::table('share_settlements')->count());
        $this->assertSame(1, (int)DB::table('share_settlements')->value('revision'));
    }

    public function test_lottery_fact_swap_is_one_publication_and_settlement_waits_for_the_complete_parent_order(): void
    {
        DB::table('share_bet_facts')->delete();
        DB::table('share_accounts')->where('user_id', 1)->update(['earned_amount' => 0]);
        $schema = DB::connection()->getSchemaBuilder();
        $schema->create('lhc_order', function ($t) {
            $t->id(); $t->string('member_id'); $t->string('ordernum'); $t->integer('type'); $t->integer('lottery_status');
            $t->decimal('amount', 24, 10); $t->decimal('total_amount', 24, 10); $t->bigInteger('created_at'); $t->bigInteger('updated_at');
        });
        $schema->create('lhc_order_item', function ($t) {
            $t->id(); $t->string('member_id'); $t->string('ordernum'); $t->integer('lottery_status');
            $t->decimal('win_amount', 24, 10); $t->decimal('profit_and_loss', 24, 10); $t->bigInteger('updated_at');
        });
        $time = \Carbon\Carbon::parse('2026-10-24 12:00:00', 'Asia/Shanghai')->timestamp;
        DB::table('lhc_order')->insert(['id' => 1, 'member_id' => 'M2', 'ordernum' => 'atomic-swap', 'type' => 5,
            'lottery_status' => 2, 'amount' => 100, 'total_amount' => 200, 'created_at' => $time, 'updated_at' => $time]);
        DB::table('lhc_order_item')->insert([
            ['id' => 101, 'member_id' => 'M2', 'ordernum' => 'atomic-swap', 'lottery_status' => 2,
                'win_amount' => 150, 'profit_and_loss' => 50, 'updated_at' => $time],
            ['id' => 102, 'member_id' => 'M2', 'ordernum' => 'atomic-swap', 'lottery_status' => 1,
                'win_amount' => 0, 'profit_and_loss' => -100, 'updated_at' => $time],
        ]);
        $baseline = (new \App\Services\Sharing\BetFactSynchronizer())->sync('2026-10-24', '2026-10-24', true);
        $this->assertSame([], $baseline['errors']); $this->assertSame(2, $baseline['created']);
        $this->assertSame('150.0000000000', bcadd((string)DB::table('share_bet_facts')->sum('effective_amount'), '0', 10));
        $this->assertSame(1, (new ShareSettlementService())->settle('2026-10-24')['changed']);
        $this->assertSame('15.0000000000', $this->amount('earned_amount'));
        DB::table('lhc_order_item')->where('id', 101)->update(['lottery_status' => 1, 'win_amount' => 0, 'profit_and_loss' => -100, 'updated_at' => $time + 1]);
        DB::table('lhc_order_item')->where('id', 102)->update(['lottery_status' => 2, 'win_amount' => 150, 'profit_and_loss' => 50, 'updated_at' => $time + 1]);
        [$sync, $settled] = $this->race(['framework' => 'laravel', 'action' => 'sync'], ['framework' => 'laravel', 'action' => 'settle'], function ($held) {
            // Inside the source-order transaction the first changed child temporarily makes 200.
            // Another connection must still see the old, committed 150 and settlement must wait.
            $this->assertSame('200.0000000000', file_get_contents($held['partial']));
            $this->assertSame('150.0000000000', bcadd((string)DB::table('share_bet_facts')->sum('effective_amount'), '0', 10));
        });
        $this->assertTrue($sync['ok']); $this->assertSame([], $sync['result']['errors']); $this->assertSame(2, $sync['result']['updated']);
        $this->assertTrue($settled['ok']); $this->assertSame(0, $settled['result']['changed']);
        $this->assertSame('150.0000000000', bcadd((string)DB::table('share_settlements')->value('direct_performance'), '0', 10));
        $this->assertSame('150.0000000000', bcadd((string)DB::table('share_bet_facts')->sum('effective_amount'), '0', 10));
        $this->assertSame('15.0000000000', $this->amount('earned_amount'));
        $this->assertSame(1, DB::table('share_ledger')->count());
        $this->assertSame(1, (int)DB::table('share_settlements')->value('revision'));
    }

    private function settledClaim(): int
    {
        DB::table('share_accounts')->where('user_id', 1)->update(['earned_amount' => 0]);
        $this->assertSame(1, (new ShareSettlementService())->settle('2026-10-24')['changed']);
        $claim = $this->single(['framework' => 'tp', 'request_id' => 'corrected-claim']);
        $this->assertTrue($claim['ok']); $this->assertSame('10.00', $claim['result']['amount']);
        return (int)$claim['result']['id'];
    }

    private function race(array $first, array $second, ?callable $whileHeld = null): array
    {
        $a = $this->spawn($first + ['hold' => true]);
        try {
            $this->waitFile($a['locked']);
            if ($whileHeld) $whileHeld($a);
            $b = $this->spawn($second);
            $this->waitFile($b['ready']);
            $deadline = microtime(true) + 5;
            do {
                $waiting = (int)$this->server->query('SELECT COUNT(*) FROM performance_schema.data_lock_waits')->fetchColumn();
                if ($waiting > 0) break;
                usleep(10000);
            } while (microtime(true) < $deadline);
            $this->assertGreaterThan(0, $waiting, 'The second real database transaction must contend on an InnoDB row lock');
        } finally {
            touch($a['release']);
        }
        return [$this->collect($a), $this->collect($b)];
    }

    private function single(array $args): array { return $this->collect($this->spawn($args)); }

    private function spawn(array $args): array
    {
        $token = bin2hex(random_bytes(6));
        $params = $args + ['socket' => $this->socket, 'database' => $this->database,
            'tp_root' => getenv('SHARING_TP_ROOT') ?: dirname(__DIR__, 3) . '/melbet_sport-api',
            'laravel_autoload' => getenv('AGENT_TEST_AUTOLOAD') ?: dirname(__DIR__, 2) . '/vendor/autoload.php',
            'ready' => $this->runDir . '/' . $token . '.ready', 'locked' => $this->runDir . '/' . $token . '.locked',
            'partial' => $this->runDir . '/' . $token . '.partial',
            'release' => $this->runDir . '/' . $token . '.release'];
        $process = proc_open([PHP_BINARY, '-d', 'xdebug.mode=off', '-d', 'error_reporting=8191',
            dirname(__DIR__) . '/fixtures/sharing_mysql_worker.php', json_encode($params, JSON_THROW_ON_ERROR)],
            [0 => ['pipe', 'r'], 1 => ['pipe', 'w'], 2 => ['pipe', 'w']], $pipes);
        if (!is_resource($process)) throw new \RuntimeException('Cannot start an isolated worker process');
        fclose($pipes[0]); unset($pipes[0]);
        foreach ($pipes as $pipe) stream_set_blocking($pipe, false);
        $worker = $params + ['process' => $process, 'pipes' => $pipes, 'token' => $token];
        $this->processes[$token] = $worker;
        return $worker;
    }

    private function collect(array $worker): array
    {
        $deadline = microtime(true) + 20;
        while (proc_get_status($worker['process'])['running']) {
            if (microtime(true) > $deadline) throw new \RuntimeException('A fixture worker did not finish');
            usleep(10000);
        }
        $stdout = stream_get_contents($worker['pipes'][1]); $stderr = stream_get_contents($worker['pipes'][2]);
        foreach ($worker['pipes'] as $pipe) fclose($pipe);
        proc_close($worker['process']); unset($this->processes[$worker['token']]);
        $decoded = json_decode($stdout, true);
        if (!is_array($decoded)) throw new \RuntimeException('Worker produced no JSON result: ' . $stdout . $stderr);
        return $decoded;
    }

    private function waitFile(string $path): void
    {
        $deadline = microtime(true) + 15;
        while (!is_file($path)) {
            if (microtime(true) > $deadline) throw new \RuntimeException('Worker did not reach expected synchronization point: ' . basename($path));
            usleep(10000);
        }
    }

    private function amount(string $field): string { return bcadd((string)DB::table('share_accounts')->where('user_id', 1)->value($field), '0', 10); }
    private function wallet(): string { return bcadd((string)DB::table('wallets')->where('member_id', 'M1')->value('available_balance'), '0', 10); }
}