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(); (require dirname(__DIR__, 2) . '/database/migrations/2026_11_01_000000_add_delivery_state_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(); } private function consume(int $cursor): array { $batch = AdminRealtimeFeed::next($cursor); AdminRealtimeFeed::acknowledge($batch['ids']); return $batch; } 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, 'ids' => [], 'events' => [], 'has_more' => false], $this->consume(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 = $this->consume(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 = $this->consume(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($this->consume(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 = $this->consume(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(); $this->consume(5); $insert(6, '2026-10-31 09:58:00'); $this->assertSame([], $this->consume(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]); AdminRealtime::record('sport_order', 99, 'deleted'); $withdraw->delete(); $this->consume(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)); DB::table('admin')->insert(['id' => 2, 'username' => 'limited']); $this->app['cache']->put('admin_realtime:channels:2', ['admin.users'], 60); $limited = \Firebase\JWT\JWT::encode(['iat' => time(), 'exp' => $exp, 'sub' => 2, 'user_id' => 2], str_repeat('k', 32), 'HS256'); foreach (['admin.stats.platform_count', 'admin.stats.unprocessed'] as $statsChannel) { $this->assertSame(200, $call($token, 'private-' . $statsChannel)->getStatusCode()); $this->assertSame(403, $call($limited, 'private-' . $statsChannel)->getStatusCode(), '统计频道也必须校验管理员权限'); } } public function test_failed_termination_retries_without_affecting_a_newer_login(): void { $pusher = $this->fakePusher(); $expired = time() - 10; $future = time() + 600; AdminRealtimeAccess::authorize(7, $expired, 'private-admin.orders.withdraw', '1.1'); AdminRealtimeAccess::authorize(7, $expired, 'private-admin.users', '1.1'); AdminRealtimeAccess::authorize(7, $future, 'private-admin.users', '2.2'); $pusher->fail = true; $this->assertSame([], AdminRealtimeAccess::terminateExpiredSessions()); $this->assertArrayHasKey("7:{$expired}", $this->app['cache']->get('admin_realtime:sessions')); $pusher->fail = false; // 模拟发出 HTTP 请求期间有另一个管理员完成授权;合并清理不能覆盖它。 $pusher->onTerminate = static fn() => AdminRealtimeAccess::authorize(8, $future, 'private-admin.users', '3.3'); $this->assertSame(["7:{$expired}"], AdminRealtimeAccess::terminateExpiredSessions()); $this->assertSame([], AdminRealtimeAccess::terminateExpiredSessions(), '已断开的会话不重复处理'); $this->assertSame(["7:{$expired}", "7:{$expired}"], $pusher->attempts); $this->assertSame(["7:{$future}" => $future, "8:{$future}" => $future], $this->app['cache']->get('admin_realtime:sessions')); $this->assertNotEmpty(array_filter($this->log->records, fn ($r) => $r[1] === 'admin_realtime_terminate_failed'), 'Reverb 不可达时只记日志'); } public function test_session_registration_failure_denies_channel_authorization(): void { $this->fakePusher(); $this->app->instance('cache', new CacheRepository(new class extends ArrayStore { public function forever($key, $value) { return false; } })); Facade::clearResolvedInstance('cache'); $this->expectException(\RuntimeException::class); $this->expectExceptionMessage('登记后台推送会话失败'); AdminRealtimeAccess::authorize(7, time() + 600, 'private-admin.users', '1.1'); } public function test_session_without_expiry_cannot_be_authorized(): void { $this->fakePusher(); $this->expectException(\RuntimeException::class); AdminRealtimeAccess::authorize(7, null, 'private-admin.users', '1.1'); } public function test_delayed_expired_subscription_is_rejected_using_verified_metadata(): void { $connection = $this->reverbConnection(); $application = $connection->app(); $subscription = new \Laravel\Reverb\Protocols\Pusher\Channels\ChannelConnection($connection, ['user_id' => '7:' . (time() - 10)]); $channel = $this->createMock(\Laravel\Reverb\Protocols\Pusher\Channels\Channel::class); $channel->method('find')->with($connection)->willReturn($subscription); $manager = $this->createMock(\Laravel\Reverb\Protocols\Pusher\Contracts\ChannelManager::class); $manager->method('for')->with($application)->willReturnSelf(); $manager->method('find')->with('private-admin.users')->willReturn($channel); $this->app->instance(\Laravel\Reverb\Protocols\Pusher\Contracts\ChannelManager::class, $manager); $listener = new \App\Listeners\RejectExpiredAdminRealtimeSubscription(); // 客户端伪造的未来时间不能替代 Reverb 已验签的过期订阅数据。 $listener->handle(new \Laravel\Reverb\Events\MessageReceived($connection, json_encode([ 'event' => 'pusher:subscribe', 'data' => ['channel' => 'private-admin.users', 'channel_data' => json_encode(['user_id' => '7:' . (time() + 600)])], ]))); $this->assertSame(1, $connection->disconnects); } public function test_valid_subscription_remains_connected(): void { $connection = $this->reverbConnection(); $subscription = new \Laravel\Reverb\Protocols\Pusher\Channels\ChannelConnection($connection, ['user_id' => '7:' . (time() + 600)]); $channel = $this->createStub(\Laravel\Reverb\Protocols\Pusher\Channels\Channel::class); $channel->method('find')->willReturn($subscription); $manager = $this->createStub(\Laravel\Reverb\Protocols\Pusher\Contracts\ChannelManager::class); $manager->method('for')->willReturnSelf(); $manager->method('find')->willReturn($channel); $this->app->instance(\Laravel\Reverb\Protocols\Pusher\Contracts\ChannelManager::class, $manager); (new \App\Listeners\RejectExpiredAdminRealtimeSubscription())->handle(new \Laravel\Reverb\Events\MessageReceived($connection, json_encode(['event' => 'pusher:subscribe', 'data' => json_encode(['channel' => 'private-admin.users'])]))); $this->assertSame(0, $connection->disconnects); } public function test_mixed_session_identities_cannot_hide_an_expiring_channel_on_the_same_socket(): void { $this->app->bind(\Laravel\Reverb\Protocols\Pusher\Contracts\ChannelConnectionManager::class, \Laravel\Reverb\Protocols\Pusher\Managers\ArrayChannelConnectionManager::class); $manager = new \Laravel\Reverb\Protocols\Pusher\Managers\ArrayChannelManager(); $this->app->instance(\Laravel\Reverb\Protocols\Pusher\Contracts\ChannelManager::class, $manager); $mixed = $this->reverbConnection('1.1'); $newer = $this->reverbConnection('2.2'); $later = time() + 600; $earlier = time() + 60; $subscribe = function ($connection, string $name, int $expires) use ($manager): void { $data = json_encode(['user_id' => "7:{$expires}"]); $auth = 'key:' . hash_hmac('sha256', "{$connection->id()}:{$name}:{$data}", 'secret'); $manager->for($connection->app())->findOrCreate($name)->subscribe($connection, $auth, $data); (new \App\Listeners\RejectExpiredAdminRealtimeSubscription())->handle(new \Laravel\Reverb\Events\MessageReceived($connection, json_encode(['event' => 'pusher:subscribe', 'data' => ['channel' => $name]]))); }; $subscribe($mixed, 'private-admin.users', $later); $subscribe($mixed, 'private-admin.orders.withdraw', $earlier); $this->assertSame("7:{$later}", $manager->connections()['1.1']->data('user_id'), '真实 Reverb 管理器只保留首个 user_id'); $this->assertSame(1, $mixed->disconnects, '不同到期时间不能共用同一 socket'); $subscribe($newer, 'private-admin.users', $later); $subscribe($newer, 'private-admin.orders.withdraw', $later); $this->assertSame(0, $newer->disconnects, '独立的新登录及同一会话的多个频道保持有效'); } private function reverbConnection(string $id = '1.1'): \Laravel\Reverb\Contracts\Connection { return new class($id) extends \Laravel\Reverb\Contracts\Connection { public int $disconnects = 0; public function __construct(private string $socketId) { $this->application = new \Laravel\Reverb\Application('app', 'key', 'secret', 60, 30, ['*'], 10000); } public function identifier(): string { return $this->socketId; } public function id(): string { return $this->socketId; } public function send(string $message): void {} public function control(string $type = '9'): void {} public function terminate(): void { $this->disconnects++; } }; } public function test_broadcast_failure_is_retried_in_order_after_relay_restart(): void { $broadcaster = new class { public bool $fail = true; public array $attempts = []; public function queue($event) { $this->attempts[] = $event->message['seq']; if ($this->fail && $event->message['seq'] === 2) throw new \RuntimeException('temporary publish failure'); } }; Broadcast::swap($broadcaster); AdminRealtime::record('withdraws', [77, 88, 99], 'deleted'); $run = function (): int { $command = new \App\Console\Commands\RelayAdminRealtimeEvents(); $command->setLaravel($this->app); // 隔离测试不启动 ConsoleKernel,只验证命令投递状态;不注册进程信号。 $previous = (new \ReflectionProperty(\Illuminate\Console\Signals::class, 'availabilityResolver'))->getValue(); \Illuminate\Console\Signals::resolveAvailabilityUsing(static fn() => false); try { return (new CommandTester($command))->execute(['--once' => true]); } finally { \Illuminate\Console\Signals::resolveAvailabilityUsing($previous); } }; $this->assertSame(1, $run()); $this->assertSame([1, 2], $broadcaster->attempts, '失败后不能发送更大的 seq'); $this->assertSame(3, DB::table('admin_realtime_events')->whereNull('published_at')->count()); $broadcaster->fail = false; $this->assertSame(0, $run()); $this->assertSame([1, 2, 1, 2, 3], $broadcaster->attempts, '新进程恢复未确认批次,允许重复但不丢失'); $this->assertSame(0, DB::table('admin_realtime_events')->whereNull('published_at')->count()); $this->assertSame(0, $run()); $this->assertSame([1, 2, 1, 2, 3], $broadcaster->attempts); } public function test_pruning_all_history_never_reuses_seq_and_keeps_pending_delivery(): void { AdminRealtime::record('withdraws', 77, 'deleted'); $first = $this->consume(0); DB::table('admin_realtime_events')->update(['created_at' => '2026-10-27 00:00:00']); $this->assertSame(1, AdminRealtimeFeed::prune()); $this->assertSame(1, AdminRealtimeFeed::lastSeq()); AdminRealtime::record('withdraws', 88, 'deleted'); $second = AdminRealtimeFeed::next(AdminRealtimeFeed::startCursor()); $this->assertSame(2, $second['events'][0]['seq']); $this->assertSame('history_unavailable', AdminRealtimeFeed::replay($first['events'][0]['seq'], ['admin.orders.withdraw'])['reason']); DB::table('admin_realtime_events')->update(['created_at' => '2026-10-27 00:00:00']); $this->assertSame(0, AdminRealtimeFeed::prune(), '待发送批次即使超期也不能清理'); $retry = AdminRealtimeFeed::next(AdminRealtimeFeed::startCursor()); $this->assertSame($second['ids'], $retry['ids']); $this->assertSame(array_column($second['events'], 'seq'), array_column($retry['events'], 'seq')); AdminRealtimeFeed::acknowledge($second['ids']); $this->assertSame(1, AdminRealtimeFeed::prune()); $this->assertSame(2, AdminRealtimeFeed::lastSeq()); } private function fakePusher(): \Pusher\Pusher { $pusher = new class('app-key', 'app-secret', 'app-id', ['host' => '127.0.0.1', 'port' => 1, 'scheme' => 'http', 'useTLS' => false]) extends \Pusher\Pusher { public bool $fail = false; public array $attempts = []; public $onTerminate = null; public function terminateUserConnections(string $user_id): object { $this->attempts[] = $user_id; if ($this->fail) throw new \RuntimeException('temporary termination failure'); if ($this->onTerminate) ($this->onTerminate)(); return (object) []; } }; Broadcast::swap(new class($pusher) { public function __construct(private $pusher) {} public function connection($name) { return $this; } public function getPusher() { return $this->pusher; } }); return $pusher; } 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)); } }