bot-28/tests/Integration/AdminRealtimeTest.php

<?php

namespace Tests\Integration;

use App\Http\Controllers\admin\Withdraw as WithdrawController;
use App\Models\User;
use App\Models\Withdraw;
use App\Services\AdminRealtime\AdminRealtime;
use App\Services\AdminRealtime\AdminRealtimeFeed;
use App\Services\AdminRealtime\AdminRealtimeRows;
use App\Services\AdminRealtime\AdminRealtimeAccess;
use App\Services\JwtService;
use App\Http\Controllers\admin\Realtime as RealtimeController;
use Illuminate\Broadcasting\BroadcastManager;
use Illuminate\Contracts\Broadcasting\Factory as BroadcastingFactory;
use Carbon\Carbon;
use Illuminate\Cache\ArrayStore;
use Illuminate\Cache\Repository as CacheRepository;
use Illuminate\Config\Repository;
use Illuminate\Database\Capsule\Manager;
use Illuminate\Database\DatabaseTransactionsManager;
use Illuminate\Database\Eloquent\Model;
use Illuminate\Database\Schema\Blueprint;
use Illuminate\Events\Dispatcher;
use Illuminate\Foundation\Application;
use Illuminate\Http\JsonResponse;
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 AdminRealtimeTest extends TestCase
{
    private Application $app;
    private object $log;

    protected function setUp(): void
    {
        Carbon::setTestNow(Carbon::parse('2026-10-31 10:00:00', 'Asia/Shanghai'));
        Facade::clearResolvedInstances();
        Model::clearBootedModels();
        $this->app = new Application(dirname(__DIR__, 2));
        $this->app->instance('config', new Repository([
            'app' => ['locale' => 'zh', 'timezone' => 'Asia/Shanghai'],
            'admin_realtime' => ['enabled' => true, 'connection' => 'reverb', 'batch_size' => 200, 'gap_seconds' => 60, 'replay_limit' => 1000,
                'permission_cache_seconds' => 60, 'online_scan_seconds' => 15, 'retention_hours' => 72],
        ]));
        Facade::setFacadeApplication($this->app);
        $events = new Dispatcher($this->app); $this->app->instance('events', $events);
        $db = new Manager($this->app); $db->addConnection(['driver' => 'sqlite', 'database' => ':memory:', 'prefix' => 'bot_']);
        $db->setEventDispatcher($events); $db->setAsGlobal(); $db->bootEloquent();
        $db->getConnection()->setTransactionManager(new DatabaseTransactionsManager());
        $this->app->instance('db', $db->getDatabaseManager());
        $this->app->bind('db.schema', fn () => $db->schema());
        $translator = new Translator(new ArrayLoader(), 'zh'); $this->app->instance('translator', $translator);
        $this->app->instance('validator', new Factory($translator, $this->app));
        $this->log = new class {
            public array $records = [];
            public function channel($name = null) { return $this; }
            public function __call($level, $args) { $this->records[] = [$level, $args[0] ?? '']; }
        };
        $this->app->instance('log', $this->log);
        $this->app->instance('cache', new CacheRepository(new ArrayStore()));
        $this->app->instance(\Illuminate\Contracts\Routing\ResponseFactory::class, new \Tests\Support\JsonResponseFactory());
        Request::macro('validate', function (array $rules) { return app('validator')->make($this->all(), $rules)->validate(); });
        $this->app->instance('request', Request::create('/admin/realtime/events', 'GET'));

        (require dirname(__DIR__, 2) . '/database/migrations/2026_10_31_160000_create_admin_realtime_events.php')->up();
        (require dirname(__DIR__, 2) . '/database/migrations/2026_10_31_210000_add_seq_to_admin_realtime_events.php')->up();
        Schema::create('users', function (Blueprint $t) {
            $t->id(); $t->string('member_id'); $t->string('user_id')->default(''); $t->string('first_name')->default('');
            $t->string('admin_note')->nullable(); $t->string('language')->nullable(); $t->integer('recharge_channel_group_id')->nullable(); $t->timestamps();
        });
        Schema::create('withdraws', function (Blueprint $t) {
            $t->id(); $t->string('member_id'); $t->decimal('amount', 20, 10)->default(0); $t->decimal('service_charge', 20, 10)->default(0);
            $t->decimal('to_account', 20, 10)->default(0); $t->string('address')->default(''); $t->decimal('exchange_rate', 20, 10)->default(0);
            $t->integer('status')->default(0); $t->decimal('after_balance', 20, 10)->default(0); $t->string('remark')->nullable();
            $t->string('admin_note')->nullable(); $t->integer('is_locked')->default(0); $t->integer('is_send')->default(0); $t->timestamps();
        });
        DB::table('users')->insert(['member_id' => '9001', 'user_id' => 'u9001', 'first_name' => 'Alice', 'admin_note' => 'vip']);
    }

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

    private function events(): array
    {
        return DB::table('admin_realtime_events')->orderBy('id')->get()->map(fn ($r) => "{$r->resource}:{$r->resource_id}:{$r->action}")->all();
    }

    public function test_record_waits_for_commit_and_drops_on_rollback(): void
    {
        AdminRealtime::record('withdraws', 1);
        $this->assertSame(['withdraws:1:updated'], $this->events());

        DB::beginTransaction();
        AdminRealtime::record('withdraws', [2, 2, 3], 'created');
        $this->assertCount(1, $this->events(), '事务内不写入');
        DB::commit();
        $this->assertSame(['withdraws:1:updated', 'withdraws:2:created', 'withdraws:3:created'], $this->events());

        DB::beginTransaction();
        AdminRealtime::record('withdraws', 4);
        DB::rollBack();
        $this->assertCount(3, $this->events(), '回滚后丢弃');

        $this->app['config']->set('admin_realtime.enabled', false);
        AdminRealtime::record('withdraws', 5);
        $this->assertCount(3, $this->events(), '关闭时不登记');
    }

    public function test_record_failure_never_breaks_business_transaction(): void
    {
        Schema::drop('admin_realtime_events');
        DB::transaction(function () {
            Withdraw::create(['member_id' => '9001', 'amount' => 10]);
        });
        $this->assertSame(1, Withdraw::count(), '事件表缺失也不影响主业务');
        $this->assertSame('error', $this->log->records[0][0] ?? null);
    }

    public function test_model_hooks_capture_orders_and_new_users_only(): void
    {
        $withdraw = Withdraw::create(['member_id' => '9001', 'amount' => 10]);
        $withdraw->update(['is_send' => 1]);
        $withdraw->update(['status' => 1]);
        $user = User::query()->create(['member_id' => '9002', 'first_name' => 'Bob']);
        $user->first_name = 'Bobby'; $user->save();
        $withdraw->delete();

        $this->assertSame([
            "withdraws:{$withdraw->id}:created", "withdraws:{$withdraw->id}:updated",
            "users:{$user->id}:created", "withdraws:{$withdraw->id}:deleted",
        ], $this->events());
    }

    public function test_feed_resolves_rows_with_the_list_endpoint_shape(): void
    {
        $this->assertSame(0, AdminRealtimeFeed::startCursor());
        $this->assertSame(['cursor' => 0, 'events' => [], 'has_more' => false], AdminRealtimeFeed::next(0));

        $withdraw = Withdraw::create(['member_id' => '9001', 'amount' => 12.5, 'to_account' => 12.5]);
        $withdraw->update(['status' => 1]);
        Withdraw::create(['member_id' => '9001', 'amount' => 3]);
        $result = AdminRealtimeFeed::next(0);

        $this->assertSame(3, $result['cursor']);
        $this->assertSame([1, 2, 3], DB::table('admin_realtime_events')->orderBy('id')->pluck('seq')->map(fn ($v) => (int) $v)->all());
        $this->assertCount(2, $result['events'], '同一记录一批内合并');
        $first = $result['events'][0];
        $this->assertSame(['seq' => 2, 'id' => 2, 'channel' => 'admin.orders.withdraw', 'event' => 'order.created', 'action' => 'created', 'key' => ['id' => 1]],
            array_intersect_key($first, array_flip(['seq', 'id', 'channel', 'event', 'action', 'key'])));

        $listRequest = Request::create('/admin/withdraw', 'GET');
        $this->app->instance('request', $listRequest);
        $list = (new WithdrawController())->index()->getData(true)['data']['data'];
        $row = collect($list)->firstWhere('id', 1);
        $this->assertSame($row, $first['data'], '推送行与列表接口返回的行一致');
        $this->assertSame('Alice', $first['data']['first_name']);
        $this->assertSame(1, $first['data']['status']);

        $withdraw->delete();
        $deleted = AdminRealtimeFeed::next(3)['events'];
        $this->assertSame([['seq' => 4, 'channel' => 'admin.orders.withdraw', 'action' => 'deleted', 'key' => ['id' => 1], 'data' => null]],
            array_map(fn ($e) => array_intersect_key($e, array_flip(['seq', 'channel', 'action', 'key', 'data'])), $deleted));
        $this->assertSame(4, AdminRealtimeFeed::startCursor(), '重启后从上次广播到的 id 继续');
        $this->assertNotEmpty(array_filter($this->log->records, fn ($r) => $r[1] === 'admin_realtime_online_scan_failed'),
            'SQLite 不支持在线判定 SQL，扫描失败只记日志不影响广播');
    }

    public function test_feed_delivers_late_commits_with_a_later_seq(): void
    {
        $insert = fn (int $id, string $at) => DB::table('admin_realtime_events')->insert([
            'id' => $id, 'resource' => 'withdraws', 'resource_id' => (string) $id, 'action' => 'deleted', 'source' => 'melbet', 'created_at' => $at,
        ]);
        $insert(1, '2026-10-31 09:59:59');
        $insert(3, '2026-10-31 09:59:59');
        $this->assertSame([1, 3], array_column(AdminRealtimeFeed::next(0)['events'], 'id'));

        $insert(2, '2026-10-31 09:59:30');
        $insert(4, '2026-10-31 09:58:00');
        $insert(5, '2026-10-31 09:58:00');
        $batch = AdminRealtimeFeed::next(3);
        $this->assertSame([[3, 2], [4, 4], [5, 5]], array_map(fn ($e) => [$e['seq'], $e['id']], $batch['events']),
            '游标之前晚提交的 id 仍会广播，seq 按广播先后递增');
        $this->assertSame(5, $batch['cursor']);

        $insert(6, '2026-10-31 10:00:00');
        $insert(7, '2026-10-31 10:00:00');
        DB::table('admin_realtime_events')->where('id', 6)->delete();
        AdminRealtimeFeed::next(5);
        $insert(6, '2026-10-31 09:58:00');
        $this->assertSame([], AdminRealtimeFeed::next(7)['events'], '超过 gap_seconds 的晚提交不再补发');
    }

    public function test_replay_by_seq_filters_channels_and_requests_resync(): void
    {
        $withdraw = Withdraw::create(['member_id' => '9001', 'amount' => 1]);
        Withdraw::create(['member_id' => '9001', 'amount' => 2]);
        User::query()->create(['member_id' => '9003', 'first_name' => 'Carol']);
        $withdraw->delete();
        AdminRealtimeFeed::next(0);
        $this->assertSame(4, AdminRealtimeFeed::lastSeq());

        $replay = AdminRealtimeFeed::replay(1, ['admin.orders.withdraw']);
        $this->assertFalse($replay['resync']);
        $this->assertSame(4, $replay['last_seq']);
        $this->assertSame([[2, 'order.created'], [4, 'order.deleted']], array_map(fn ($e) => [$e['seq'], $e['event']], $replay['events']),
            '只返回请求的频道，并按当前数据解析');
        $this->assertSame([], AdminRealtimeFeed::replay(4, ['admin.orders.withdraw'])['events']);
        $this->assertSame('invalid_cursor', AdminRealtimeFeed::replay(9, ['admin.orders.withdraw'])['reason']);

        $this->app['config']->set('admin_realtime.replay_limit', 2);
        $this->assertSame('too_many_events', AdminRealtimeFeed::replay(0, ['admin.orders.withdraw'])['reason']);

        DB::table('admin_realtime_events')->where('seq', 1)->delete();
        $this->assertSame('history_unavailable', AdminRealtimeFeed::replay(1, ['admin.orders.withdraw'])['reason'], '游标已被清理时重新拉列表');
    }

    public function test_channel_auth_signs_session_user_for_reverb(): void
    {
        $this->bindBroadcasting();
        Schema::create('admin', function (Blueprint $t) { $t->id(); $t->string('username'); $t->timestamps(); });
        DB::table('admin')->insert(['id' => 1, 'username' => 'root']);
        $exp = time() + 3600;
        $token = \Firebase\JWT\JWT::encode(['iat' => time(), 'exp' => $exp, 'sub' => 1, 'user_id' => 1], str_repeat('k', 32), 'HS256');

        $call = function (?string $token, string $channel) {
            $request = Request::create('/admin/broadcasting/auth', 'POST', ['socket_id' => '123.456', 'channel_name' => $channel]);
            if ($token) $request->headers->set('Authorization', 'Bearer ' . $token);
            $this->app->instance('request', $request);
            return (new RealtimeController())->auth(new JwtService());
        };

        $response = $call($token, 'private-admin.orders.withdraw');
        $this->assertSame(200, $response->getStatusCode());
        $body = $response->getData(true);
        $this->assertSame(['user_id' => "1:{$exp}"], json_decode($body['channel_data'], true));
        // 与 Reverb 校验私有频道的签名算法一致（InteractsWithPrivateChannels::verify）
        $this->assertSame('app-key:' . hash_hmac('sha256', "123.456:private-admin.orders.withdraw:{$body['channel_data']}", 'app-secret'), $body['auth']);

        $this->assertSame(401, $call(null, 'private-admin.orders.withdraw')->getStatusCode());
        $this->assertSame(401, $call($token . 'x', 'private-admin.orders.withdraw')->getStatusCode());
        $this->assertSame(403, $call($token, 'private-admin.orders.unknown')->getStatusCode());
        $this->assertSame(403, $call($token, 'admin.orders.withdraw')->getStatusCode(), '只授权私有频道');
        $this->assertSame(array_keys(AdminRealtimeRows::CHANNELS), AdminRealtimeAccess::allowedChannels(1));
    }

    public function test_expired_sessions_are_terminated_once(): void
    {
        $this->bindBroadcasting();
        AdminRealtimeAccess::authorize(7, time() - 1, 'private-admin.orders.withdraw', '1.1');
        AdminRealtimeAccess::authorize(7, time() - 1, 'private-admin.users', '1.1');
        AdminRealtimeAccess::authorize(8, time() + 600, 'private-admin.users', '2.2');

        $this->assertSame(['7:' . (time() - 1)], AdminRealtimeAccess::terminateExpiredSessions());
        $this->assertSame([], AdminRealtimeAccess::terminateExpiredSessions(), '已断开的会话不重复处理');
        $this->assertNotEmpty(array_filter($this->log->records, fn ($r) => $r[1] === 'admin_realtime_terminate_failed'),
            'Reverb 不可达时只记日志');
    }

    private function bindBroadcasting(): void
    {
        $this->app['config']->set('broadcasting.connections.reverb', [
            'driver' => 'reverb', 'key' => 'app-key', 'secret' => 'app-secret', 'app_id' => 'app-id',
            'options' => ['host' => '127.0.0.1', 'port' => 1, 'scheme' => 'http', 'useTLS' => false],
            'client_options' => ['connect_timeout' => 1, 'timeout' => 1],
        ]);
        $this->app['config']->set('app.jwt_secret', str_repeat('k', 32));
        $this->app['config']->set('app.jwt_exp', 3600);
        $this->app->instance(BroadcastingFactory::class, new BroadcastManager($this->app));
    }
}