bot-28/tests/Integration/AdminRealtimeSummaryTest.php

<?php

namespace Tests\Integration;

use App\Http\Controllers\admin\Online;
use App\Http\Controllers\admin\PaymentOrder;
use App\Http\Controllers\admin\User as AdminUser;
use App\Http\Controllers\agent\MemberController as AgentMemberController;
use App\Models\Agent\Agent;
use App\Services\AdminRealtime\AdminRealtimeFeed;
use App\Services\AdminRealtime\AdminRealtimeSummaries;
use App\Services\OnlineUserService;
use App\Services\UserService;
use Carbon\Carbon;
use Illuminate\Cache\ArrayStore;
use Illuminate\Cache\Repository as Cache;
use Illuminate\Config\Repository;
use Illuminate\Database\Capsule\Manager;
use Illuminate\Database\Eloquent\Model;
use Illuminate\Database\Schema\Blueprint;
use Illuminate\Events\Dispatcher;
use Illuminate\Foundation\Application;
use Illuminate\Http\Request;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Facade;
use Illuminate\Support\Facades\Schema;
use Illuminate\Translation\ArrayLoader;
use Illuminate\Translation\Translator;
use Illuminate\Validation\Factory;
use PHPUnit\Framework\TestCase;

class AdminRealtimeSummaryTest extends TestCase
{
    private Application $app;
    private Carbon $clock;

    protected function setUp(): void
    {
        Facade::clearResolvedInstances(); Model::clearBootedModels();
        $this->clock = Carbon::now('Asia/Shanghai'); Carbon::setTestNow($this->clock);
        $this->app = new Application(dirname(__DIR__, 2));
        $this->app->instance('config', new Repository(['app' => ['locale' => 'zh', 'timezone' => 'Asia/Shanghai'],
            'admin_realtime' => ['enabled' => true, 'summary_scan_seconds' => 5, 'online_scan_seconds' => 15]]));
        $this->app->instance('events', new Dispatcher($this->app));
        $this->app->instance('cache', new Cache(new ArrayStore()));
        $this->app->instance('translator', new Translator(new ArrayLoader(), 'zh'));
        $this->app->instance('request', Request::create('/admin/realtime/replay'));
        $this->app->instance(\Illuminate\Contracts\Routing\ResponseFactory::class, new \Tests\Support\JsonResponseFactory());
        $this->app->instance('log', new class {
            public array $records = [];
            public function channel($name) { return $this; }
            public function error($message, $context = []) { $this->records[] = [$message, $context]; }
        });
        Facade::setFacadeApplication($this->app);
        $db = new Manager($this->app);
        $db->addConnection(['driver' => 'sqlite', 'database' => ':memory:', 'prefix' => 'bot_']);
        $db->setEventDispatcher($this->app['events']); $db->setAsGlobal(); $db->bootEloquent();
        $this->app->instance('db', $db->getDatabaseManager()); $this->app->bind('db.schema', fn() => $db->schema());
        // 只在隔离 SQLite 中提供本服务所用排序规则，执行原始统计 SQL 而非替换统计服务。
        DB::connection()->getPdo()->sqliteCreateCollation('utf8mb4_general_ci', 'strcasecmp');
        foreach (['2026_10_31_160000_create_admin_realtime_events.php', '2026_10_31_210000_add_seq_to_admin_realtime_events.php',
            '2026_11_01_000000_add_delivery_state_to_admin_realtime_events.php'] as $file) {
            (require dirname(__DIR__, 2) . '/database/migrations/' . $file)->up();
        }
        Schema::create('users', function (Blueprint $t) {
            $t->id(); $t->string('user_id'); $t->string('member_id'); $t->integer('from')->default(1); $t->integer('last_active_time')->nullable();
            $t->timestamp('created_at')->nullable(); $t->integer('level')->nullable(); $t->string('agent_user_code')->nullable();
        });
        Schema::create('user_session', function (Blueprint $t) {
            $t->id(); $t->string('user_id'); $t->integer('expire_time'); $t->integer('create_time'); $t->integer('update_time')->nullable();
        });
        Schema::create('user_login', function (Blueprint $t) { $t->id(); $t->string('user_id'); $t->string('platform'); $t->integer('status')->default(1); });
        foreach (['recharges', 'withdraws', 'payment_orders'] as $table) {
            Schema::create($table, function (Blueprint $t) use ($table) {
                $t->id(); $t->integer('status'); $t->dateTime('created_at');
                if ($table === 'payment_orders') $t->integer('type');
            });
        }
        Schema::create('share_claims', function (Blueprint $t) { $t->id(); $t->string('status'); });
        DB::table('users')->insert(['user_id' => 'account1', 'member_id' => '1001', 'last_active_time' => $this->clock->timestamp]);
        DB::table('user_login')->insert(['user_id' => 'account1', 'platform' => 'iOS', 'status' => 1]);
        $now = date('Y-m-d H:i:s');
        DB::table('recharges')->insert(['status' => 0, 'created_at' => $now]);
        DB::table('withdraws')->insert(['status' => 0, 'created_at' => $now]);
        foreach ([[3, 0], [3, 1], [3, 5], [1, 0], [1, 1], [4, 0], [2, 0], [1, 6]] as [$type, $status]) {
            DB::table('payment_orders')->insert(compact('type', 'status') + ['created_at' => $now]);
        }
        DB::table('share_claims')->insert([['status' => 'pending'], ['status' => 'pending'], ['status' => 'approved']]);
    }

    protected function tearDown(): void
    {
        DB::disconnect(); Facade::clearResolvedInstances(); Carbon::setTestNow(); parent::tearDown();
    }

    private function tick(int $seconds): array
    {
        Carbon::setTestNow($this->clock->copy()->addSeconds($seconds));
        $batch = AdminRealtimeFeed::next(AdminRealtimeFeed::startCursor());
        AdminRealtimeFeed::acknowledge($batch['ids']);
        return $batch['events'];
    }

    public function test_initial_snapshots_match_http_responses_and_support_replay(): void
    {
        $events = $this->tick(0);
        $this->assertSame(['admin.stats.platform_count', 'admin.stats.unprocessed'], array_column($events, 'channel'));
        $this->assertSame(['stats.updated', 'stats.updated'], array_column($events, 'event'));
        $this->assertSame([['scope' => 'all'], ['scope' => 'all']], array_column($events, 'key'));
        $this->assertSame((new Online())->platformCount()->getData(true)['data'], $events[0]['data']);
        $this->assertSame((new PaymentOrder())->unProcessed()->getData(true)['data'], $events[1]['data']);
        $this->assertSame(['usdtRecharge' => 1, 'rgRecharge' => 3, 'rmbRecharge' => 2, 'usdtWithdraw' => 1,
            'rgWithdraw' => 1, 'rmbWithdraw' => 1, 'shareClaim' => 2], $events[1]['data']);
        $replay = AdminRealtimeFeed::replay(0, ['admin.stats.unprocessed']);
        $this->assertSame(1, count($replay['events']));
        $this->assertSame($events[1]['data'], $replay['events'][0]['data']);
        $this->assertSame([], $this->tick(6), '统计值未变，不重复推送');
    }

    public function test_platform_switch_is_detected_even_when_online_user_ids_do_not_change(): void
    {
        $this->tick(0); $before = OnlineUserService::onlineUserIds();
        DB::table('user_login')->insert(['user_id' => 'account1', 'platform' => 'Android', 'status' => 1]);
        $events = $this->tick(6);
        $this->assertSame($before, OnlineUserService::onlineUserIds());
        $this->assertCount(1, $events);
        $this->assertSame('admin.stats.platform_count', $events[0]['channel']);
        $this->assertSame(0, $events[0]['data']['apple']);
        $this->assertSame(1, $events[0]['data']['android']);
        $this->assertSame([], $this->tick(12));
        DB::table('users')->update(['last_active_time' => $this->clock->timestamp - 300]);
        $events = $this->tick(18);
        $summary = array_values(array_filter($events, fn($e) => $e['channel'] === 'admin.stats.platform_count'));
        $this->assertSame(0, $summary[0]['data']['total'], '活跃窗口过期时归零');
    }

    public function test_online_filter_and_offline_push_use_the_same_five_minute_window(): void
    {
        Schema::create('wallets', function (Blueprint $t) { $t->id(); $t->string('member_id'); });
        Schema::create('balance_logs', function (Blueprint $t) {
            $t->id(); $t->string('member_id'); $t->integer('type'); $t->string('change_type'); $t->decimal('amount');
        });
        $now = $this->clock->timestamp;
        DB::table('users')->where('id', 1)->update(['last_active_time' => $now - 320]);
        DB::table('users')->insert([
            ['user_id' => 'account2', 'member_id' => '1002', 'last_active_time' => $now - 300, 'from' => 1],
            ['user_id' => 'account3', 'member_id' => '1003', 'last_active_time' => null, 'from' => 1],
            ['user_id' => 'guest', 'member_id' => '1004', 'last_active_time' => $now, 'from' => 2],
            ['user_id' => 'account5', 'member_id' => '1005', 'last_active_time' => $now - 299, 'from' => 1],
        ]);
        DB::table('user_session')->insert([
            ['user_id' => 'account1', 'expire_time' => $now + 86400, 'create_time' => $now - 3600, 'update_time' => $now - 299],
            ['user_id' => '1002', 'expire_time' => $now + 86400, 'create_time' => $now - 300, 'update_time' => null],
            ['user_id' => 'account3', 'expire_time' => $now, 'create_time' => $now - 10, 'update_time' => $now],
        ]);

        $all = UserService::paginate();
        $this->assertSame(5, $all['total']);
        $this->assertSame(['1005' => true, '1004' => false, '1003' => false, '1002' => false, '1001' => true], $all['data']->pluck('is_online', 'member_id')->all());
        $this->app->instance('validator', new Factory($this->app['translator'], $this->app));
        Request::macro('validate', function (array $rules) { return app('validator')->make($this->all(), $rules)->validate(); });
        $this->app->instance('request', Request::create('/admin/user?is_online=1&limit=1'));
        $online = (new AdminUser())->index()->getData(true)['data'];
        $this->assertSame(2, $online['total']);
        $this->assertCount(1, $online['data']);
        $this->assertTrue($online['data'][0]['is_online']);
        $this->assertSame(3, UserService::paginate(['is_online' => '0'])['total']);
        $this->assertSame(0, UserService::paginate(['is_online' => '1', 'member_id' => '1002'])['total']);
        $this->assertEqualsCanonicalizing([1, 5], OnlineUserService::onlineUserIds());
        $this->assertSame(2, OnlineUserService::platformCounts()['total']);

        $this->tick(0);
        Carbon::setTestNow($this->clock->copy()->addSecond());
        $this->assertSame(0, UserService::paginate(['is_online' => '1'])['total'], '满五分钟即离线');
        $events = array_values(array_filter($this->tick(16), fn ($event) => $event['event'] === 'user.offline'));
        $this->assertEqualsCanonicalizing(['1001', '1005'], array_column(array_column($events, 'key'), 'member_id'));
    }

    public function test_member_list_shows_the_independent_agent_as_superior(): void
    {
        Schema::table('users', function (Blueprint $t) {
            $t->unsignedBigInteger('agent_id')->nullable(); $t->string('user_code')->nullable(); $t->string('account')->nullable(); $t->string('first_name')->nullable();
        });
        Schema::create('agents', function (Blueprint $t) {
            $t->id(); $t->string('username'); $t->string('real_name')->default(''); $t->integer('type')->default(1); $t->integer('is_test')->default(0);
        });
        Schema::create('wallets', function (Blueprint $t) {
            $t->id(); $t->string('member_id'); $t->decimal('available_balance')->default(0); $t->decimal('wager_required')->default(0); $t->decimal('wager_completed')->default(0);
        });
        Schema::create('balance_logs', function (Blueprint $t) {
            $t->id(); $t->string('member_id'); $t->integer('type'); $t->string('change_type'); $t->decimal('amount');
        });
        Schema::create('share_members', function (Blueprint $t) {
            $t->unsignedBigInteger('user_id')->primary(); $t->unsignedBigInteger('parent_user_id')->nullable();
        });
        DB::table('agents')->insert([
            ['id' => 6, 'username' => 'agent06', 'type' => 1, 'is_test' => 0],
            ['id' => 7, 'username' => 'holder07', 'type' => 2, 'is_test' => 1],
            ['id' => 8, 'username' => 'agent08', 'type' => 1, 'is_test' => 0],
        ]);
        DB::table('users')->where('member_id', '1001')->update(['agent_id' => 8, 'user_code' => 'INV1001', 'account' => 'inviter1001']);
        foreach ([['1002', 6, null], ['1003', 7, null], ['1004', null, "\tinv1001\t"], ['1005', 7, 'INV1001'],
            ['1006', 6, null], ['1007', null, 'GONE'], ['1008', null, '0'], ['1009', null, 'DUPE'], ['1010', null, 'SELF'],
            ['1011', null, 'TRIAL'], ['1012', null, null]] as [$member, $agent, $code]) {
            DB::table('users')->insert(['user_id' => 'u' . $member, 'member_id' => $member, 'agent_id' => $agent, 'agent_user_code' => $code, 'account' => 'm' . $member]);
        }
        DB::table('users')->where('member_id', '1003')->update(['user_code' => '0']);
        DB::table('users')->whereIn('member_id', ['1002', '1005'])->update(['user_code' => 'DUPE']);
        DB::table('users')->where('member_id', '1010')->update(['user_code' => 'SELF']);
        DB::table('users')->where('member_id', '1012')->update(['user_code' => 'TRIAL', 'from' => 2]);
        DB::table('share_members')->insert([
            ['user_id' => DB::table('users')->where('member_id', '1006')->value('id'), 'parent_user_id' => 1],
            ['user_id' => DB::table('users')->where('member_id', '1007')->value('id'), 'parent_user_id' => 99],
        ]);
        $rows = collect(UserService::paginate(['limit' => 15])['data'])->keyBy('member_id');
        $this->assertSame(['agent', '合营：agent06', 'agent06', '合营'], [$rows['1002']['superior']['type'], $rows['1002']['superior']['label'],
            $rows['1002']['superior']['account'], $rows['1002']['superior']['agent_type_text']]);
        $this->assertNull($rows['1003']['superior'], 'equity agents are not shown as superior');
        $this->assertSame('邀请人：inviter1001（邀请码：INV1001）', $rows['1005']['superior']['label'], 'equity agent member falls back to the inviter');
        $this->assertSame(['member', '邀请人：inviter1001（邀请码：INV1001）', 'INV1001'],
            [$rows['1004']['superior']['type'], $rows['1004']['superior']['label'], $rows['1004']['superior']['invite_code']]);
        $this->assertSame($rows['1004']['superior'], $rows['1006']['superior'], 'fixed parent takes precedence over the agent');
        $this->assertNull($rows['1007']['superior'], 'parent record is missing');
        $this->assertNull($rows['1008']['superior'], 'zero is a placeholder, not an invitation code');
        foreach (['1009', '1010', '1011'] as $memberId) {
            $this->assertNull($rows[$memberId]['superior'], 'invalid legacy inviter');
        }
        $canBind = $rows->pluck('can_bind_superior', 'member_id')->sortKeys();
        $this->assertSame([1001 => false, 1002 => false, 1003 => true, 1004 => false, 1005 => false,
            1006 => false, 1007 => false, 1008 => true, 1009 => false, 1010 => false, 1011 => false, 1012 => false], $canBind->all());
        $this->assertSame('agent06', $rows['1002']->toArray()['system_agent']['username']);

        $this->app->instance('validator', new Factory($this->app['translator'], $this->app));
        Request::macro('validate', function (array $rules) { return app('validator')->make($this->all(), $rules)->validate(); });
        $request = Request::create('/agent/members');
        $request->attributes->set('agent', Agent::query()->findOrFail(6));
        $this->app->instance('request', $request);
        $response = (new AgentMemberController())->index($request)->getData(true);
        $this->assertSame(0, $response['code'], $response['msg']);
        $agentRows = collect($response['data']['data'])->keyBy('member_id');
        $this->assertEqualsCanonicalizing([1002, 1006], $agentRows->keys()->all());
        $this->assertArrayNotHasKey('superior', $agentRows['1006']);
        $this->assertArrayNotHasKey('share_parent', $agentRows['1006']);
        $this->assertStringNotContainsString('inviter1001', json_encode($agentRows['1006'], JSON_THROW_ON_ERROR));
    }

    public function test_member_amount_columns_sort_before_pagination(): void
    {
        DB::connection()->getPdo()->sqliteCreateFunction('GREATEST', max(...), 2);
        Schema::create('wallets', function (Blueprint $t) {
            $t->id(); $t->string('member_id'); $t->decimal('available_balance');
            $t->decimal('wager_required'); $t->decimal('wager_completed');
        });
        Schema::create('balance_logs', function (Blueprint $t) {
            $t->id(); $t->string('member_id'); $t->integer('type'); $t->string('change_type'); $t->decimal('amount');
        });
        DB::table('users')->insert([
            ['user_id' => 'account2', 'member_id' => '1002'],
            ['user_id' => 'account3', 'member_id' => '1003'],
        ]);
        DB::table('wallets')->insert([
            ['member_id' => '1001', 'available_balance' => 100, 'wager_required' => 50, 'wager_completed' => 40],
            ['member_id' => '1002', 'available_balance' => 2, 'wager_required' => 100, 'wager_completed' => 0],
        ]);
        $this->app->instance('validator', new Factory($this->app['translator'], $this->app));
        Request::macro('validate', function (array $rules) { return app('validator')->make($this->all(), $rules)->validate(); });

        foreach (['available_balance' => ['1002', '1001'], 'wager_remaining' => ['1001', '1002']] as $field => [$lower, $higher]) {
            foreach ([['asc', 2, $lower], ['desc', 1, $higher]] as [$order, $page, $memberId]) {
                $filters = ['sort_field' => $field, 'sort_order' => $order, 'page' => $page, 'limit' => 1];
                $this->app->instance('request', Request::create('/admin/user', 'GET', $filters));
                $body = (new AdminUser())->index()->getData(true);
                $this->assertSame(0, $body['code'], $body['msg']);
                $this->assertSame(3, $body['data']['total']);
                $this->assertSame([$memberId], array_column($body['data']['data'], 'member_id'));
            }
        }
    }

    public function test_pending_badges_track_raw_writes_claim_audits_and_aged_orders(): void
    {
        $this->tick(0);
        // 不经过 Eloquent 或事件登记，模拟共享数据库另一应用的写入。
        DB::table('payment_orders')->where('type', 3)->update(['status' => 2]);
        DB::table('share_claims')->where('id', 1)->update(['status' => 'rejected']);
        DB::table('recharges')->update(['created_at' => date('Y-m-d H:i:s', time() - 90000)]);
        $events = $this->tick(6);
        $this->assertCount(1, $events);
        $this->assertSame('admin.stats.unprocessed', $events[0]['channel']);
        $this->assertSame(0, $events[0]['data']['rgRecharge']);
        $this->assertSame(0, $events[0]['data']['usdtRecharge']);
        $this->assertSame(1, $events[0]['data']['shareClaim']);
        $this->assertSame((new PaymentOrder())->unProcessed()->getData(true)['data'], $events[0]['data']);
        $this->assertSame(8, DB::table('payment_orders')->count(), '扫描只读业务表');
    }

    public function test_failed_outbox_registration_does_not_swallow_the_snapshot_change(): void
    {
        Schema::rename('admin_realtime_events', 'unavailable_realtime_events');
        AdminRealtimeSummaries::scan();
        $this->assertNotEmpty($this->app['log']->records);
        Schema::rename('unavailable_realtime_events', 'admin_realtime_events');
        $events = $this->tick(6);
        $this->assertCount(2, $events, '事件表恢复后仍会登记两个初始快照');
        $this->assertSame([], $this->tick(12));
    }
}