管理员可订阅的频道，口径与 CheckButtonPermission 对列表接口的按钮权限一致 */ public static function allowedChannels(int $adminId): array { $ttl = max(0, (int) config('admin_realtime.permission_cache_seconds', 60)); if ($ttl === 0) { return self::resolveChannels($adminId); } return Cache::remember('admin_realtime:channels:' . $adminId, $ttl, static fn(): array => self::resolveChannels($adminId)); } private static function resolveChannels(int $adminId): array { $channels = array_keys(AdminRealtimeRows::CHANNELS); if ($adminId === 1) { return $channels; } $roleIds = collect(RoleUserService::findAll(['user_id' => $adminId]))->pluck('role_id')->toArray() ?: [-1]; $granted = []; foreach (RoleService::model()::with(['menus' => function ($query) { $query->where('type', MenuService::model()::TYPE_BUTTON); }])->whereIn('id', $roleIds)->get() as $role) { $granted = array_merge($granted, $role['menus_uris'] ?: []); } return array_values(array_filter($channels, static function (string $channel) use ($granted): bool { $uri = AdminRealtimeRows::CHANNELS[$channel][1]; $menu = MenuService::findOne(['uri' => $uri, 'type' => MenuService::model()::TYPE_BUTTON, 'status' => MenuService::model()::STATUS_SHOW]); return empty($menu) || in_array($uri, $granted, true); })); } /** * Pusher 协议私有频道授权，返回 {auth, channel_data} * * @param string $channelName 带 private- 前缀的频道名 * @throws \Pusher\PusherException socket_id 或频道名格式不合法 */ public static function authorize(int $adminId, ?int $sessionExpiresAt, string $channelName, string $socketId): array { if (!$sessionExpiresAt) { throw new RuntimeException('后台推送要求 JWT 带有过期时间'); } $userId = self::sessionUserId($adminId, $sessionExpiresAt); $auth = self::pusher()->authorizeChannel($channelName, $socketId, json_encode(['user_id' => $userId])); if ($sessionExpiresAt) { self::rememberSession($userId, $sessionExpiresAt); } return json_decode($auth, true); } public static function sessionUserId(int $adminId, ?int $sessionExpiresAt): string { return $adminId . ':' . (int) $sessionExpiresAt; } /** 断开 JWT 已过期的会话，由 relay 定期调用；返回断开的会话 user_id */ public static function terminateExpiredSessions(): array { $sessions = Cache::get(self::SESSIONS, []); $terminated = []; foreach ($sessions as $userId => $exp) { if ($exp > time()) continue; try { self::pusher()->terminateUserConnections((string) $userId); $terminated[(string) $userId] = $exp; } catch (\Throwable $e) { Log::channel('realtime')->error('admin_realtime_terminate_failed', ['user_id' => $userId, 'error' => $e->getMessage()]); } } if ($terminated) { Cache::lock(self::SESSIONS . ':lock', 5)->block(3, function () use ($terminated) { $current = Cache::get(self::SESSIONS, []); foreach ($terminated as $userId => $exp) { if (($current[$userId] ?? null) === $exp) unset($current[$userId]); } if (!Cache::forever(self::SESSIONS, $current)) { throw new RuntimeException('保存后台推送会话状态失败'); } }); } return array_keys($terminated); } /** 同一会话订阅多个频道会重复授权，已登记的不再加锁写缓存 */ private static function rememberSession(string $userId, int $exp): void { if (isset(Cache::get(self::SESSIONS, [])[$userId])) { return; } Cache::lock(self::SESSIONS . ':lock', 5)->block(3, function () use ($userId, $exp) { $sessions = Cache::get(self::SESSIONS, []); $sessions[$userId] = $exp; if (!Cache::forever(self::SESSIONS, $sessions)) { throw new RuntimeException('登记后台推送会话失败'); } }); } private static function pusher(): Pusher { return Broadcast::connection(config('admin_realtime.connection'))->getPusher(); } }