管理员可订阅的频道，口径与 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 { $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 { $expired = []; Cache::lock(self::SESSIONS . ':lock', 5)->block(3, function () use (&$expired) { $sessions = Cache::get(self::SESSIONS, []); $now = time(); foreach ($sessions as $userId => $exp) { if ($exp <= $now) { $expired[] = (string) $userId; unset($sessions[$userId]); } } if ($expired) { Cache::forever(self::SESSIONS, $sessions); } }); foreach ($expired as $userId) { try { self::pusher()->terminateUserConnections($userId); } catch (\Throwable $e) { Log::channel('realtime')->error('admin_realtime_terminate_failed', ['user_id' => $userId, 'error' => $e->getMessage()]); } } return $expired; } /** 同一会话订阅多个频道会重复授权，已登记的不再加锁写缓存 */ private static function rememberSession(string $userId, int $exp): void { if (isset(Cache::get(self::SESSIONS, [])[$userId])) { return; } try { Cache::lock(self::SESSIONS . ':lock', 5)->block(3, function () use ($userId, $exp) { $sessions = Cache::get(self::SESSIONS, []); $sessions[$userId] = $exp; Cache::forever(self::SESSIONS, $sessions); }); } catch (\Throwable $e) { // 登记失败只影响到期主动断开，过期后重连仍会因授权失败断开 Log::channel('realtime')->error('admin_realtime_session_register_failed', ['user_id' => $userId, 'error' => $e->getMessage()]); } } private static function pusher(): Pusher { return Broadcast::connection(config('admin_realtime.connection'))->getPusher(); } }