AdminRealtimeTest.php 38 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626
  1. <?php
  2. namespace Tests\Integration;
  3. use App\Http\Controllers\admin\PaymentOrder as PaymentOrderController;
  4. use App\Http\Controllers\admin\Withdraw as WithdrawController;
  5. use App\Models\PaymentOrder;
  6. use App\Models\User;
  7. use App\Models\Withdraw;
  8. use App\Services\AdminRealtime\AdminRealtime;
  9. use App\Services\AdminRealtime\AdminRealtimeExchangeList;
  10. use App\Services\AdminRealtime\AdminRealtimeFeed;
  11. use App\Services\AdminRealtime\AdminRealtimeRows;
  12. use App\Services\AdminRealtime\AdminRealtimeAccess;
  13. use App\Services\JwtService;
  14. use App\Http\Controllers\admin\Realtime as RealtimeController;
  15. use Illuminate\Broadcasting\BroadcastManager;
  16. use Illuminate\Contracts\Broadcasting\Factory as BroadcastingFactory;
  17. use Carbon\Carbon;
  18. use Illuminate\Cache\ArrayStore;
  19. use Illuminate\Cache\Repository as CacheRepository;
  20. use Illuminate\Config\Repository;
  21. use Illuminate\Database\Capsule\Manager;
  22. use Illuminate\Database\DatabaseTransactionsManager;
  23. use Illuminate\Database\Eloquent\Model;
  24. use Illuminate\Database\Schema\Blueprint;
  25. use Illuminate\Events\Dispatcher;
  26. use Illuminate\Foundation\Application;
  27. use Illuminate\Http\JsonResponse;
  28. use Illuminate\Http\Request;
  29. use Illuminate\Support\Facades\DB;
  30. use Illuminate\Support\Facades\Facade;
  31. use Illuminate\Support\Facades\Schema;
  32. use Illuminate\Translation\ArrayLoader;
  33. use Illuminate\Translation\Translator;
  34. use Illuminate\Validation\Factory;
  35. use PHPUnit\Framework\TestCase;
  36. use Illuminate\Support\Facades\Broadcast;
  37. use Symfony\Component\Console\Tester\CommandTester;
  38. class AdminRealtimeTest extends TestCase
  39. {
  40. private Application $app;
  41. private object $log;
  42. protected function setUp(): void
  43. {
  44. Carbon::setTestNow(Carbon::parse('2026-10-31 10:00:00', 'Asia/Shanghai'));
  45. Facade::clearResolvedInstances();
  46. Model::clearBootedModels();
  47. $this->app = new Application(dirname(__DIR__, 2));
  48. $this->app->instance('config', new Repository([
  49. 'app' => ['locale' => 'zh', 'timezone' => 'Asia/Shanghai'],
  50. 'admin_realtime' => ['enabled' => true, 'connection' => 'reverb', 'batch_size' => 200, 'gap_seconds' => 60, 'replay_limit' => 1000,
  51. 'permission_cache_seconds' => 60, 'online_scan_seconds' => 15, 'retention_hours' => 72],
  52. ]));
  53. Facade::setFacadeApplication($this->app);
  54. $events = new Dispatcher($this->app); $this->app->instance('events', $events);
  55. $db = new Manager($this->app); $db->addConnection(['driver' => 'sqlite', 'database' => ':memory:', 'prefix' => 'bot_']);
  56. $db->setEventDispatcher($events); $db->setAsGlobal(); $db->bootEloquent();
  57. $db->getConnection()->setTransactionManager(new DatabaseTransactionsManager());
  58. $this->app->instance('db', $db->getDatabaseManager());
  59. $this->app->bind('db.schema', fn () => $db->schema());
  60. $translator = new Translator(new ArrayLoader(), 'zh'); $this->app->instance('translator', $translator);
  61. $this->app->instance('validator', new Factory($translator, $this->app));
  62. $this->log = new class {
  63. public array $records = [];
  64. public function channel($name = null) { return $this; }
  65. public function __call($level, $args) { $this->records[] = [$level, $args[0] ?? '']; }
  66. };
  67. $this->app->instance('log', $this->log);
  68. $this->app->instance('cache', new CacheRepository(new ArrayStore()));
  69. $this->app->instance(\Illuminate\Contracts\Routing\ResponseFactory::class, new \Tests\Support\JsonResponseFactory());
  70. Request::macro('validate', function (array $rules) { return app('validator')->make($this->all(), $rules)->validate(); });
  71. $this->app->instance('request', Request::create('/admin/realtime/events', 'GET'));
  72. (require dirname(__DIR__, 2) . '/database/migrations/2026_10_31_160000_create_admin_realtime_events.php')->up();
  73. (require dirname(__DIR__, 2) . '/database/migrations/2026_10_31_210000_add_seq_to_admin_realtime_events.php')->up();
  74. (require dirname(__DIR__, 2) . '/database/migrations/2026_11_01_000000_add_delivery_state_to_admin_realtime_events.php')->up();
  75. Schema::create('users', function (Blueprint $t) {
  76. $t->id(); $t->string('member_id'); $t->string('user_id')->default(''); $t->string('account')->nullable(); $t->string('first_name')->default('');
  77. $t->string('admin_note')->nullable(); $t->string('language')->nullable(); $t->integer('recharge_channel_group_id')->nullable(); $t->string('real_name', 64)->nullable(); $t->timestamps();
  78. });
  79. Schema::create('withdraws', function (Blueprint $t) {
  80. $t->id(); $t->string('member_id'); $t->decimal('amount', 20, 10)->default(0); $t->decimal('service_charge', 20, 10)->default(0);
  81. $t->decimal('to_account', 20, 10)->default(0); $t->string('address')->default(''); $t->decimal('exchange_rate', 20, 10)->default(0);
  82. $t->integer('status')->default(0); $t->decimal('after_balance', 20, 10)->default(0); $t->string('remark')->nullable();
  83. $t->string('admin_note')->nullable(); $t->integer('is_locked')->default(0); $t->integer('is_send')->default(0); $t->timestamps();
  84. });
  85. DB::table('users')->insert(['member_id' => '9001', 'user_id' => 'u9001', 'first_name' => 'Alice', 'admin_note' => 'vip', 'real_name' => '王小明']);
  86. }
  87. protected function tearDown(): void { Carbon::setTestNow(); DB::disconnect(); Model::unsetEventDispatcher(); Model::clearBootedModels(); Facade::clearResolvedInstances(); parent::tearDown(); }
  88. private function events(): array
  89. {
  90. return DB::table('admin_realtime_events')->orderBy('id')->get()->map(fn ($r) => "{$r->resource}:{$r->resource_id}:{$r->action}")->all();
  91. }
  92. private function consume(int $cursor): array
  93. {
  94. $batch = AdminRealtimeFeed::next($cursor);
  95. AdminRealtimeFeed::acknowledge($batch['ids']);
  96. return $batch;
  97. }
  98. public function test_exchange_list_changes_coalesce_and_can_be_replayed(): void
  99. {
  100. Schema::create('admin_exchange_list_changes', fn (Blueprint $table) => $table->id());
  101. $changes = DB::table('admin_exchange_list_changes');
  102. $changes->insert([['id' => 2], ['id' => 3]]);
  103. $batch = AdminRealtimeFeed::next(0);
  104. $this->assertCount(1, $batch['events']);
  105. $event = $batch['events'][0];
  106. $this->assertSame(AdminRealtimeExchangeList::CHANNEL, $event['channel']);
  107. $this->assertSame('exchange_list.updated', $event['event']);
  108. $this->assertSame(['scope' => 'all'], $event['key']);
  109. $this->assertNull($event['data']);
  110. $broadcast = new \App\Events\AdminRealtimeBroadcast($event);
  111. $this->assertSame('private-admin.operation.exchange_list', $broadcast->broadcastOn()[0]->name);
  112. $this->assertSame($event['event'], $broadcast->broadcastAs());
  113. $this->assertSame(0, $changes->count());
  114. $this->assertSame($batch, AdminRealtimeFeed::next($batch['cursor']));
  115. AdminRealtimeFeed::acknowledge($batch['ids']);
  116. // An earlier buffer ID can become visible after a later ID has been broadcast.
  117. $changes->insert(['id' => 1]);
  118. $second = $this->consume($batch['cursor']);
  119. $this->assertCount(1, $second['events']);
  120. $this->assertSame($second['events'], AdminRealtimeFeed::replay($event['seq'], [$event['channel']])['events']);
  121. $this->assertSame([], AdminRealtimeFeed::replay(0, ['admin.users'])['events']);
  122. $this->assertSame([], $this->consume($second['cursor'])['events']);
  123. }
  124. public function test_record_waits_for_commit_and_drops_on_rollback(): void
  125. {
  126. AdminRealtime::record('withdraws', 1);
  127. $this->assertSame(['withdraws:1:updated'], $this->events());
  128. DB::beginTransaction();
  129. AdminRealtime::record('withdraws', [2, 2, 3], 'created');
  130. $this->assertCount(1, $this->events(), '事务内不写入');
  131. DB::commit();
  132. $this->assertSame(['withdraws:1:updated', 'withdraws:2:created', 'withdraws:3:created'], $this->events());
  133. DB::beginTransaction();
  134. AdminRealtime::record('withdraws', 4);
  135. DB::rollBack();
  136. $this->assertCount(3, $this->events(), '回滚后丢弃');
  137. $this->app['config']->set('admin_realtime.enabled', false);
  138. AdminRealtime::record('withdraws', 5);
  139. $this->assertCount(3, $this->events(), '关闭时不登记');
  140. }
  141. public function test_record_failure_never_breaks_business_transaction(): void
  142. {
  143. Schema::drop('admin_realtime_events');
  144. DB::transaction(function () {
  145. Withdraw::create(['member_id' => '9001', 'amount' => 10]);
  146. });
  147. $this->assertSame(1, Withdraw::count(), '事件表缺失也不影响主业务');
  148. $this->assertSame('error', $this->log->records[0][0] ?? null);
  149. }
  150. public function test_model_hooks_capture_orders_and_new_users_only(): void
  151. {
  152. $withdraw = Withdraw::create(['member_id' => '9001', 'amount' => 10]);
  153. $withdraw->update(['is_send' => 1]);
  154. $withdraw->update(['status' => 1]);
  155. $user = User::query()->create(['member_id' => '9002', 'first_name' => 'Bob']);
  156. $user->first_name = 'Bobby'; $user->save();
  157. $withdraw->delete();
  158. $this->assertSame([
  159. "withdraws:{$withdraw->id}:created", "withdraws:{$withdraw->id}:updated",
  160. "users:{$user->id}:created", "withdraws:{$withdraw->id}:deleted",
  161. ], $this->events());
  162. }
  163. public function test_agent_application_audit_lists_push_rows_and_member_submissions(): void
  164. {
  165. Schema::create('agents', function (Blueprint $t) {
  166. $t->id(); $t->integer('type')->default(1); $t->string('username'); $t->string('real_name')->default('');
  167. $t->decimal('balance', 20, 4)->default(0); $t->decimal('margin_amount', 20, 4)->default(0);
  168. });
  169. // 用真实迁移建表:Eloquent 按模型类缓存可批量赋值的列,手写缺列的表会影响后续测试
  170. foreach (['2026_10_29_140000_create_agent_applications.php', '2026_11_08_150000_add_type_to_agent_applications.php'] as $file) {
  171. (require dirname(__DIR__, 2) . '/database/migrations/' . $file)->up();
  172. }
  173. Schema::create('agent_profit_commission_profiles', function (Blueprint $t) { $t->id(); $t->string('name'); $t->unsignedBigInteger('owner_agent_id')->nullable(); $t->timestamps(); });
  174. Schema::create('agent_plan_applications', function (Blueprint $t) {
  175. $t->id(); $t->unsignedBigInteger('agent_id'); $t->unsignedBigInteger('profile_id'); $t->string('profile_name')->default('');
  176. $t->string('status')->default('pending'); $t->string('remark')->default(''); $t->string('reject_reason')->default('');
  177. $t->unsignedBigInteger('reviewed_by')->nullable(); $t->dateTime('reviewed_at')->nullable(); $t->timestamps();
  178. });
  179. DB::table('agent_applications')->insert(['user_id' => 1, 'username' => 'old']);
  180. $this->consume(0); // 首轮只记下游标,不补推历史申请
  181. $this->assertSame([], $this->events());
  182. // 会员前台(melbet_sport-api)直接插表
  183. DB::table('agent_applications')->insert(['user_id' => 1, 'username' => 'newagent', 'type' => 2, 'created_at' => '2026-10-31 10:00:00']);
  184. DB::table('agents')->insert(['id' => 5, 'username' => 'joint5']);
  185. DB::table('agent_profit_commission_profiles')->insert(['id' => 3, 'name' => 'A方案']);
  186. $plan = \App\Models\Agent\AgentPlanApplication::create(['agent_id' => 5, 'profile_id' => 3, 'profile_name' => 'A方案', 'status' => 'pending']);
  187. $batch = $this->consume(0);
  188. $events = collect($batch['events'])->keyBy('channel');
  189. $app = $events['admin.agents.applications'];
  190. $this->assertSame(['application.created', 'created', ['id' => 2], 'melbet_sport-api'], [$app['event'], $app['action'], $app['key'], $app['source']]);
  191. $this->assertSame('9001', $app['data']['member_id']);
  192. $this->assertSame('入股', $app['data']['type_text']);
  193. $this->assertSame(['application.created', ['id' => $plan->id]], [$events['admin.agents.plan_applications']['event'], $events['admin.agents.plan_applications']['key']]);
  194. $this->assertSame('joint5', $events['admin.agents.plan_applications']['data']['agent']['username']);
  195. $this->assertSame('待审核', $events['admin.agents.plan_applications']['data']['status_text']);
  196. // 后台审核(经模型)推 updated
  197. \App\Models\Agent\AgentApplication::find(2)->update(['status' => 'rejected', 'reject_reason' => 'no']);
  198. $plan->update(['status' => 'approved', 'reviewed_by' => 1]);
  199. $updated = collect($this->consume($batch['cursor'])['events'])->keyBy('channel');
  200. $this->assertSame(['application.updated', 'rejected'], [$updated['admin.agents.applications']['event'], $updated['admin.agents.applications']['data']['status']]);
  201. $this->assertSame('已通过', $updated['admin.agents.plan_applications']['data']['status_text']);
  202. $channels = AdminRealtimeAccess::allowedChannels(1);
  203. $this->assertContains('admin.agents.applications', $channels);
  204. $this->assertContains('admin.agents.plan_applications', $channels);
  205. }
  206. public function test_feed_resolves_rows_with_the_list_endpoint_shape(): void
  207. {
  208. $this->assertSame(0, AdminRealtimeFeed::startCursor());
  209. $this->assertSame(['cursor' => 0, 'ids' => [], 'events' => [], 'has_more' => false], $this->consume(0));
  210. $withdraw = Withdraw::create(['member_id' => '9001', 'amount' => 12.5, 'to_account' => 12.5]);
  211. $withdraw->update(['status' => 1]);
  212. Withdraw::create(['member_id' => '9001', 'amount' => 3]);
  213. $result = $this->consume(0);
  214. $this->assertSame(3, $result['cursor']);
  215. $this->assertSame([1, 2, 3], DB::table('admin_realtime_events')->orderBy('id')->pluck('seq')->map(fn ($v) => (int) $v)->all());
  216. $this->assertCount(2, $result['events'], '同一记录一批内合并');
  217. $first = $result['events'][0];
  218. $this->assertSame(['seq' => 2, 'id' => 2, 'channel' => 'admin.orders.withdraw', 'event' => 'order.created', 'action' => 'created', 'key' => ['id' => 1]],
  219. array_intersect_key($first, array_flip(['seq', 'id', 'channel', 'event', 'action', 'key'])));
  220. $listRequest = Request::create('/admin/withdraw', 'GET');
  221. $this->app->instance('request', $listRequest);
  222. $list = (new WithdrawController())->index()->getData(true)['data']['data'];
  223. $row = collect($list)->firstWhere('id', 1);
  224. $this->assertSame($row, $first['data'], '推送行与列表接口返回的行一致');
  225. $this->assertSame('Alice', $first['data']['first_name']);
  226. $this->assertSame('王小明', $first['data']['real_name']);
  227. $this->assertSame(1, $first['data']['status']);
  228. $withdraw->delete();
  229. $deleted = $this->consume(3)['events'];
  230. $this->assertSame([['seq' => 4, 'channel' => 'admin.orders.withdraw', 'action' => 'deleted', 'key' => ['id' => 1], 'data' => null]],
  231. array_map(fn ($e) => array_intersect_key($e, array_flip(['seq', 'channel', 'action', 'key', 'data'])), $deleted));
  232. $this->assertSame(4, AdminRealtimeFeed::startCursor(), '重启后从上次广播到的 id 继续');
  233. $this->assertNotEmpty(array_filter($this->log->records, fn ($r) => $r[1] === 'admin_realtime_online_scan_failed'),
  234. 'SQLite 不支持在线判定 SQL,扫描失败只记日志不影响广播');
  235. }
  236. public function test_withdrawal_lists_filter_by_member_real_name_and_keep_legacy_rows(): void
  237. {
  238. Schema::create('payment_orders', function (Blueprint $t) {
  239. $t->id(); $t->integer('type'); $t->string('member_id'); $t->decimal('amount', 20, 2);
  240. $t->integer('status')->default(0); $t->string('account'); $t->string('bank_name'); $t->string('card_no'); $t->timestamps();
  241. });
  242. DB::table('users')->insert(['member_id' => '9002', 'account' => 'legacy', 'first_name' => '王小明']);
  243. foreach (['9001', '9002'] as $memberId) {
  244. DB::table('withdraws')->insert(['member_id' => $memberId, 'amount' => 10]);
  245. foreach ([2, 4] as $type) {
  246. DB::table('payment_orders')->insert(['type' => $type, 'member_id' => $memberId, 'amount' => 10,
  247. 'account' => '历史收款人', 'bank_name' => '银行', 'card_no' => '1234']);
  248. }
  249. }
  250. $controller = new WithdrawController();
  251. foreach ([['index', '/admin/withdraw', []], ['rmb', '/admin/withdraw/rmb', ['type' => 2]], ['rmb', '/admin/withdraw/rmb', ['type' => 4]]] as [$method, $path, $filters]) {
  252. $this->app->instance('request', Request::create($path, 'GET', $filters));
  253. $all = $controller->{$method}()->getData(true)['data'];
  254. $this->assertSame(2, $all['total']);
  255. $this->assertNull(collect($all['data'])->firstWhere('member_id', '9002')['real_name']);
  256. $this->app->instance('request', Request::create($path, 'GET', $filters + ['real_name' => ' 小明 ']));
  257. $filtered = $controller->{$method}()->getData(true)['data'];
  258. $this->assertSame(1, $filtered['total']);
  259. $this->assertSame(['9001'], array_column($filtered['data'], 'member_id'));
  260. $this->assertSame('王小明', $filtered['data'][0]['real_name']);
  261. if ($method === 'rmb') {
  262. $this->assertSame('历史收款人', $filtered['data'][0]['recipient_name']);
  263. }
  264. }
  265. }
  266. public function test_manual_recharge_orders_push_to_the_payment_order_channel(): void
  267. {
  268. Schema::create('payment_orders', function (Blueprint $t) {
  269. $t->id(); $t->integer('type'); $t->string('order_no')->default(''); $t->string('member_id'); $t->decimal('amount', 20, 2)->default(0);
  270. $t->decimal('fee', 20, 2)->default(0); $t->string('channel')->default(''); $t->integer('payment_type')->default(0); $t->integer('status')->default(0);
  271. $t->text('pay_data')->nullable(); $t->string('remark')->nullable(); $t->text('recharge_bonus_snapshot')->nullable();
  272. $t->integer('is_send')->default(0); $t->integer('is_locked')->default(0); $t->timestamps();
  273. });
  274. $order = PaymentOrder::create(['type' => 3, 'order_no' => 'rgcz_1', 'member_id' => '9001', 'amount' => 50, 'channel' => 'rgcz',
  275. 'payment_type' => 1, 'status' => 0, 'recharge_bonus_snapshot' => null]);
  276. $order->update(['status' => 5]);
  277. $events = $this->consume(0)['events'];
  278. $this->assertSame([['channel' => 'admin.orders.payment', 'action' => 'created', 'key' => ['id' => 1]]],
  279. array_map(fn ($e) => array_intersect_key($e, array_flip(['channel', 'action', 'key'])), $events), '人工充值只推「代收订单」频道');
  280. $this->app->instance('request', Request::create('/admin/paymentOrder', 'GET', ['channel' => 'rgcz']));
  281. $list = (new PaymentOrderController())->index()->getData(true)['data']['data'];
  282. $this->assertSame(collect($list)->firstWhere('id', 1), $events[0]['data'], '推送行与人工充值审查列表的行一致');
  283. $this->assertSame(5, $events[0]['data']['status']);
  284. }
  285. public function test_an_event_that_cannot_be_resolved_does_not_block_the_manual_recharge_push(): void
  286. {
  287. Schema::create('payment_orders', function (Blueprint $t) {
  288. $t->id(); $t->integer('type'); $t->string('order_no')->default(''); $t->string('member_id'); $t->decimal('amount', 20, 2)->default(0);
  289. $t->decimal('fee', 20, 2)->default(0); $t->string('channel')->default(''); $t->integer('payment_type')->default(0); $t->integer('status')->default(0);
  290. $t->text('pay_data')->nullable(); $t->string('remark')->nullable(); $t->text('recharge_bonus_snapshot')->nullable();
  291. $t->integer('is_send')->default(0); $t->integer('is_locked')->default(0); $t->timestamps();
  292. });
  293. // third_game_orders has no table here, so resolving this event throws.
  294. AdminRealtime::record('third_game_orders', 1, 'created');
  295. PaymentOrder::create(['type' => 3, 'order_no' => 'rgcz_1', 'member_id' => '9001', 'amount' => 50, 'channel' => 'rgcz',
  296. 'payment_type' => 1, 'status' => 0, 'recharge_bonus_snapshot' => null]);
  297. $batch = $this->consume(0);
  298. $this->assertSame([['channel' => 'admin.orders.payment', 'action' => 'created', 'key' => ['id' => 1]]],
  299. array_map(fn ($e) => array_intersect_key($e, array_flip(['channel', 'action', 'key'])), $batch['events']));
  300. $this->assertContains(['error', 'admin_realtime_resolve_failed'], $this->log->records);
  301. $this->assertSame([], $this->consume($batch['cursor'])['events'], '失败事件已确认,不会反复重试');
  302. }
  303. public function test_feed_delivers_late_commits_with_a_later_seq(): void
  304. {
  305. $insert = fn (int $id, string $at) => DB::table('admin_realtime_events')->insert([
  306. 'id' => $id, 'resource' => 'withdraws', 'resource_id' => (string) $id, 'action' => 'deleted', 'source' => 'melbet', 'created_at' => $at,
  307. ]);
  308. $insert(1, '2026-10-31 09:59:59');
  309. $insert(3, '2026-10-31 09:59:59');
  310. $this->assertSame([1, 3], array_column($this->consume(0)['events'], 'id'));
  311. $insert(2, '2026-10-31 09:59:30');
  312. $insert(4, '2026-10-31 09:58:00');
  313. $insert(5, '2026-10-31 09:58:00');
  314. $batch = $this->consume(3);
  315. $this->assertSame([[3, 2], [4, 4], [5, 5]], array_map(fn ($e) => [$e['seq'], $e['id']], $batch['events']),
  316. '游标之前晚提交的 id 仍会广播,seq 按广播先后递增');
  317. $this->assertSame(5, $batch['cursor']);
  318. $insert(6, '2026-10-31 10:00:00');
  319. $insert(7, '2026-10-31 10:00:00');
  320. DB::table('admin_realtime_events')->where('id', 6)->delete();
  321. $this->consume(5);
  322. $insert(6, '2026-10-31 09:58:00');
  323. $this->assertSame([], $this->consume(7)['events'], '超过 gap_seconds 的晚提交不再补发');
  324. }
  325. public function test_replay_by_seq_filters_channels_and_requests_resync(): void
  326. {
  327. $withdraw = Withdraw::create(['member_id' => '9001', 'amount' => 1]);
  328. Withdraw::create(['member_id' => '9001', 'amount' => 2]);
  329. AdminRealtime::record('sport_order', 99, 'deleted');
  330. $withdraw->delete();
  331. $this->consume(0);
  332. $this->assertSame(4, AdminRealtimeFeed::lastSeq());
  333. $replay = AdminRealtimeFeed::replay(1, ['admin.orders.withdraw']);
  334. $this->assertFalse($replay['resync']);
  335. $this->assertSame(4, $replay['last_seq']);
  336. $this->assertSame([[2, 'order.created'], [4, 'order.deleted']], array_map(fn ($e) => [$e['seq'], $e['event']], $replay['events']),
  337. '只返回请求的频道,并按当前数据解析');
  338. $this->assertSame([], AdminRealtimeFeed::replay(4, ['admin.orders.withdraw'])['events']);
  339. $this->assertSame('invalid_cursor', AdminRealtimeFeed::replay(9, ['admin.orders.withdraw'])['reason']);
  340. $this->app['config']->set('admin_realtime.replay_limit', 2);
  341. $this->assertSame('too_many_events', AdminRealtimeFeed::replay(0, ['admin.orders.withdraw'])['reason']);
  342. DB::table('admin_realtime_events')->where('seq', 1)->delete();
  343. $this->assertSame('history_unavailable', AdminRealtimeFeed::replay(1, ['admin.orders.withdraw'])['reason'], '游标已被清理时重新拉列表');
  344. }
  345. public function test_channel_auth_signs_session_user_for_reverb(): void
  346. {
  347. $this->bindBroadcasting();
  348. Schema::create('admin', function (Blueprint $t) { $t->id(); $t->string('username'); $t->timestamps(); });
  349. DB::table('admin')->insert(['id' => 1, 'username' => 'root']);
  350. $exp = time() + 3600;
  351. $token = \Firebase\JWT\JWT::encode(['iat' => time(), 'exp' => $exp, 'sub' => 1, 'user_id' => 1], str_repeat('k', 32), 'HS256');
  352. $call = function (?string $token, string $channel) {
  353. $request = Request::create('/admin/broadcasting/auth', 'POST', ['socket_id' => '123.456', 'channel_name' => $channel]);
  354. if ($token) $request->headers->set('Authorization', 'Bearer ' . $token);
  355. $this->app->instance('request', $request);
  356. return (new RealtimeController())->auth(new \App\Services\AdminJwtSession(new JwtService()));
  357. };
  358. $response = $call($token, 'private-admin.orders.withdraw');
  359. $this->assertSame(200, $response->getStatusCode());
  360. $body = $response->getData(true);
  361. $this->assertSame(['user_id' => "1:{$exp}"], json_decode($body['channel_data'], true));
  362. // 与 Reverb 校验私有频道的签名算法一致(InteractsWithPrivateChannels::verify)
  363. $this->assertSame('app-key:' . hash_hmac('sha256', "123.456:private-admin.orders.withdraw:{$body['channel_data']}", 'app-secret'), $body['auth']);
  364. $this->assertSame(401, $call(null, 'private-admin.orders.withdraw')->getStatusCode());
  365. $this->assertSame(401, $call($token . 'x', 'private-admin.orders.withdraw')->getStatusCode());
  366. $this->assertSame(403, $call($token, 'private-admin.orders.unknown')->getStatusCode());
  367. $this->assertSame(403, $call($token, 'admin.orders.withdraw')->getStatusCode(), '只授权私有频道');
  368. $this->assertSame(array_keys(AdminRealtimeRows::CHANNELS), AdminRealtimeAccess::allowedChannels(1));
  369. DB::table('admin')->insert(['id' => 2, 'username' => 'limited']);
  370. $this->app['cache']->put('admin_realtime:channels:2', ['admin.users'], 60);
  371. $limited = \Firebase\JWT\JWT::encode(['iat' => time(), 'exp' => $exp, 'sub' => 2, 'user_id' => 2], str_repeat('k', 32), 'HS256');
  372. foreach (['admin.stats.platform_count', 'admin.stats.unprocessed', AdminRealtimeExchangeList::CHANNEL] as $statsChannel) {
  373. $this->assertSame(200, $call($token, 'private-' . $statsChannel)->getStatusCode());
  374. $this->assertSame(403, $call($limited, 'private-' . $statsChannel)->getStatusCode(), '统计频道也必须校验管理员权限');
  375. }
  376. }
  377. public function test_failed_termination_retries_without_affecting_a_newer_login(): void
  378. {
  379. $pusher = $this->fakePusher();
  380. $expired = time() - 10;
  381. $future = time() + 600;
  382. AdminRealtimeAccess::authorize(7, $expired, 'private-admin.orders.withdraw', '1.1');
  383. AdminRealtimeAccess::authorize(7, $expired, 'private-admin.users', '1.1');
  384. AdminRealtimeAccess::authorize(7, $future, 'private-admin.users', '2.2');
  385. $pusher->fail = true;
  386. $this->assertSame([], AdminRealtimeAccess::terminateExpiredSessions());
  387. $this->assertArrayHasKey("7:{$expired}", $this->app['cache']->get('admin_realtime:sessions'));
  388. $pusher->fail = false;
  389. // 模拟发出 HTTP 请求期间有另一个管理员完成授权;合并清理不能覆盖它。
  390. $pusher->onTerminate = static fn() => AdminRealtimeAccess::authorize(8, $future, 'private-admin.users', '3.3');
  391. $this->assertSame(["7:{$expired}"], AdminRealtimeAccess::terminateExpiredSessions());
  392. $this->assertSame([], AdminRealtimeAccess::terminateExpiredSessions(), '已断开的会话不重复处理');
  393. $this->assertSame(["7:{$expired}", "7:{$expired}"], $pusher->attempts);
  394. $this->assertSame(["7:{$future}" => $future, "8:{$future}" => $future], $this->app['cache']->get('admin_realtime:sessions'));
  395. $this->assertNotEmpty(array_filter($this->log->records, fn ($r) => $r[1] === 'admin_realtime_terminate_failed'),
  396. 'Reverb 不可达时只记日志');
  397. }
  398. public function test_session_registration_failure_denies_channel_authorization(): void
  399. {
  400. $this->fakePusher();
  401. $this->app->instance('cache', new CacheRepository(new class extends ArrayStore {
  402. public function forever($key, $value) { return false; }
  403. }));
  404. Facade::clearResolvedInstance('cache');
  405. $this->expectException(\RuntimeException::class);
  406. $this->expectExceptionMessage('登记后台推送会话失败');
  407. AdminRealtimeAccess::authorize(7, time() + 600, 'private-admin.users', '1.1');
  408. }
  409. public function test_session_without_expiry_cannot_be_authorized(): void
  410. {
  411. $this->fakePusher();
  412. $this->expectException(\RuntimeException::class);
  413. AdminRealtimeAccess::authorize(7, null, 'private-admin.users', '1.1');
  414. }
  415. public function test_delayed_expired_subscription_is_rejected_using_verified_metadata(): void
  416. {
  417. $connection = $this->reverbConnection();
  418. $application = $connection->app();
  419. $subscription = new \Laravel\Reverb\Protocols\Pusher\Channels\ChannelConnection($connection, ['user_id' => '7:' . (time() - 10)]);
  420. $channel = $this->createMock(\Laravel\Reverb\Protocols\Pusher\Channels\Channel::class);
  421. $channel->method('find')->with($connection)->willReturn($subscription);
  422. $manager = $this->createMock(\Laravel\Reverb\Protocols\Pusher\Contracts\ChannelManager::class);
  423. $manager->method('for')->with($application)->willReturnSelf();
  424. $manager->method('find')->with('private-admin.users')->willReturn($channel);
  425. $this->app->instance(\Laravel\Reverb\Protocols\Pusher\Contracts\ChannelManager::class, $manager);
  426. $listener = new \App\Listeners\RejectExpiredAdminRealtimeSubscription();
  427. // 客户端伪造的未来时间不能替代 Reverb 已验签的过期订阅数据。
  428. $listener->handle(new \Laravel\Reverb\Events\MessageReceived($connection, json_encode([
  429. 'event' => 'pusher:subscribe', 'data' => ['channel' => 'private-admin.users', 'channel_data' => json_encode(['user_id' => '7:' . (time() + 600)])],
  430. ])));
  431. $this->assertSame(1, $connection->disconnects);
  432. }
  433. public function test_valid_subscription_remains_connected(): void
  434. {
  435. $connection = $this->reverbConnection();
  436. $subscription = new \Laravel\Reverb\Protocols\Pusher\Channels\ChannelConnection($connection, ['user_id' => '7:' . (time() + 600)]);
  437. $channel = $this->createStub(\Laravel\Reverb\Protocols\Pusher\Channels\Channel::class);
  438. $channel->method('find')->willReturn($subscription);
  439. $manager = $this->createStub(\Laravel\Reverb\Protocols\Pusher\Contracts\ChannelManager::class);
  440. $manager->method('for')->willReturnSelf();
  441. $manager->method('find')->willReturn($channel);
  442. $this->app->instance(\Laravel\Reverb\Protocols\Pusher\Contracts\ChannelManager::class, $manager);
  443. (new \App\Listeners\RejectExpiredAdminRealtimeSubscription())->handle(new \Laravel\Reverb\Events\MessageReceived($connection,
  444. json_encode(['event' => 'pusher:subscribe', 'data' => json_encode(['channel' => 'private-admin.users'])])));
  445. $this->assertSame(0, $connection->disconnects);
  446. }
  447. public function test_mixed_session_identities_cannot_hide_an_expiring_channel_on_the_same_socket(): void
  448. {
  449. $this->app->bind(\Laravel\Reverb\Protocols\Pusher\Contracts\ChannelConnectionManager::class,
  450. \Laravel\Reverb\Protocols\Pusher\Managers\ArrayChannelConnectionManager::class);
  451. $manager = new \Laravel\Reverb\Protocols\Pusher\Managers\ArrayChannelManager();
  452. $this->app->instance(\Laravel\Reverb\Protocols\Pusher\Contracts\ChannelManager::class, $manager);
  453. $mixed = $this->reverbConnection('1.1');
  454. $newer = $this->reverbConnection('2.2');
  455. $later = time() + 600;
  456. $earlier = time() + 60;
  457. $subscribe = function ($connection, string $name, int $expires) use ($manager): void {
  458. $data = json_encode(['user_id' => "7:{$expires}"]);
  459. $auth = 'key:' . hash_hmac('sha256', "{$connection->id()}:{$name}:{$data}", 'secret');
  460. $manager->for($connection->app())->findOrCreate($name)->subscribe($connection, $auth, $data);
  461. (new \App\Listeners\RejectExpiredAdminRealtimeSubscription())->handle(new \Laravel\Reverb\Events\MessageReceived($connection,
  462. json_encode(['event' => 'pusher:subscribe', 'data' => ['channel' => $name]])));
  463. };
  464. $subscribe($mixed, 'private-admin.users', $later);
  465. $subscribe($mixed, 'private-admin.orders.withdraw', $earlier);
  466. $this->assertSame("7:{$later}", $manager->connections()['1.1']->data('user_id'), '真实 Reverb 管理器只保留首个 user_id');
  467. $this->assertSame(1, $mixed->disconnects, '不同到期时间不能共用同一 socket');
  468. $subscribe($newer, 'private-admin.users', $later);
  469. $subscribe($newer, 'private-admin.orders.withdraw', $later);
  470. $this->assertSame(0, $newer->disconnects, '独立的新登录及同一会话的多个频道保持有效');
  471. }
  472. private function reverbConnection(string $id = '1.1'): \Laravel\Reverb\Contracts\Connection
  473. {
  474. return new class($id) extends \Laravel\Reverb\Contracts\Connection {
  475. public int $disconnects = 0;
  476. public function __construct(private string $socketId) { $this->application = new \Laravel\Reverb\Application('app', 'key', 'secret', 60, 30, ['*'], 10000); }
  477. public function identifier(): string { return $this->socketId; }
  478. public function id(): string { return $this->socketId; }
  479. public function send(string $message): void {}
  480. public function control(string $type = '9'): void {}
  481. public function terminate(): void { $this->disconnects++; }
  482. };
  483. }
  484. public function test_broadcast_failure_is_retried_in_order_after_relay_restart(): void
  485. {
  486. $broadcaster = new class {
  487. public bool $fail = true;
  488. public array $attempts = [];
  489. public function queue($event) {
  490. $this->attempts[] = $event->message['seq'];
  491. if ($this->fail && $event->message['seq'] === 2) throw new \RuntimeException('temporary publish failure');
  492. }
  493. };
  494. Broadcast::swap($broadcaster);
  495. AdminRealtime::record('withdraws', [77, 88, 99], 'deleted');
  496. $run = function (): int {
  497. $command = new \App\Console\Commands\RelayAdminRealtimeEvents();
  498. $command->setLaravel($this->app);
  499. // 隔离测试不启动 ConsoleKernel,只验证命令投递状态;不注册进程信号。
  500. $previous = (new \ReflectionProperty(\Illuminate\Console\Signals::class, 'availabilityResolver'))->getValue();
  501. \Illuminate\Console\Signals::resolveAvailabilityUsing(static fn() => false);
  502. try {
  503. return (new CommandTester($command))->execute(['--once' => true]);
  504. } finally {
  505. \Illuminate\Console\Signals::resolveAvailabilityUsing($previous);
  506. }
  507. };
  508. $this->assertSame(1, $run());
  509. $this->assertSame([1, 2], $broadcaster->attempts, '失败后不能发送更大的 seq');
  510. $this->assertSame(3, DB::table('admin_realtime_events')->whereNull('published_at')->count());
  511. $broadcaster->fail = false;
  512. $this->assertSame(0, $run());
  513. $this->assertSame([1, 2, 1, 2, 3], $broadcaster->attempts, '新进程恢复未确认批次,允许重复但不丢失');
  514. $this->assertSame(0, DB::table('admin_realtime_events')->whereNull('published_at')->count());
  515. $this->assertSame(0, $run());
  516. $this->assertSame([1, 2, 1, 2, 3], $broadcaster->attempts);
  517. }
  518. public function test_pruning_all_history_never_reuses_seq_and_keeps_pending_delivery(): void
  519. {
  520. AdminRealtime::record('withdraws', 77, 'deleted');
  521. $first = $this->consume(0);
  522. DB::table('admin_realtime_events')->update(['created_at' => '2026-10-27 00:00:00']);
  523. $this->assertSame(1, AdminRealtimeFeed::prune());
  524. $this->assertSame(1, AdminRealtimeFeed::lastSeq());
  525. AdminRealtime::record('withdraws', 88, 'deleted');
  526. $second = AdminRealtimeFeed::next(AdminRealtimeFeed::startCursor());
  527. $this->assertSame(2, $second['events'][0]['seq']);
  528. $this->assertSame('history_unavailable', AdminRealtimeFeed::replay($first['events'][0]['seq'], ['admin.orders.withdraw'])['reason']);
  529. DB::table('admin_realtime_events')->update(['created_at' => '2026-10-27 00:00:00']);
  530. $this->assertSame(0, AdminRealtimeFeed::prune(), '待发送批次即使超期也不能清理');
  531. $retry = AdminRealtimeFeed::next(AdminRealtimeFeed::startCursor());
  532. $this->assertSame($second['ids'], $retry['ids']);
  533. $this->assertSame(array_column($second['events'], 'seq'), array_column($retry['events'], 'seq'));
  534. AdminRealtimeFeed::acknowledge($second['ids']);
  535. $this->assertSame(1, AdminRealtimeFeed::prune());
  536. $this->assertSame(2, AdminRealtimeFeed::lastSeq());
  537. }
  538. private function fakePusher(): \Pusher\Pusher
  539. {
  540. $pusher = new class('app-key', 'app-secret', 'app-id', ['host' => '127.0.0.1', 'port' => 1, 'scheme' => 'http', 'useTLS' => false]) extends \Pusher\Pusher {
  541. public bool $fail = false;
  542. public array $attempts = [];
  543. public $onTerminate = null;
  544. public function terminateUserConnections(string $user_id): object {
  545. $this->attempts[] = $user_id;
  546. if ($this->fail) throw new \RuntimeException('temporary termination failure');
  547. if ($this->onTerminate) ($this->onTerminate)();
  548. return (object) [];
  549. }
  550. };
  551. Broadcast::swap(new class($pusher) {
  552. public function __construct(private $pusher) {}
  553. public function connection($name) { return $this; }
  554. public function getPusher() { return $this->pusher; }
  555. });
  556. return $pusher;
  557. }
  558. private function bindBroadcasting(): void
  559. {
  560. $this->app['config']->set('broadcasting.connections.reverb', [
  561. 'driver' => 'reverb', 'key' => 'app-key', 'secret' => 'app-secret', 'app_id' => 'app-id',
  562. 'options' => ['host' => '127.0.0.1', 'port' => 1, 'scheme' => 'http', 'useTLS' => false],
  563. 'client_options' => ['connect_timeout' => 1, 'timeout' => 1],
  564. ]);
  565. $this->app['config']->set('app.jwt_secret', str_repeat('k', 32));
  566. $this->app['config']->set('app.jwt_exp', 3600);
  567. $this->app->instance(BroadcastingFactory::class, new BroadcastManager($this->app));
  568. }
  569. }