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