bot-28/app/Services/AdminRealtime/AdminRealtimeAccess.php

<?php

namespace App\Services\AdminRealtime;

use App\Services\Agent\AgentAdminAccess;
use App\Services\MenuService;
use App\Services\RoleService;
use App\Services\RoleUserService;
use Illuminate\Support\Facades\Broadcast;
use Illuminate\Support\Facades\Cache;
use Illuminate\Support\Facades\Log;
use Pusher\Pusher;
use RuntimeException;

/**
 * 后台实时推送的访问控制：频道授权与登录过期断开。
 *
 * 频道授权响应在 channel_data 中携带「会话 user_id」（管理员 id + JWT 过期时间），Reverb 把它记在连接上；
 * relay 定期把已过期的会话登记出来，调用 Reverb 的 terminate 接口断开这些连接。
 * 断开后前端自动重连会重新授权，过期的 JWT 授权失败（401），前端据此跳转登录。
 */
final class AdminRealtimeAccess
{
    private const SESSIONS = 'admin_realtime:sessions';

    /** @return list<string> 管理员可订阅的频道，口径与 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;
        }
        if (!AgentAdminAccess::allowed($adminId)) {
            $channels = array_values(array_diff($channels, AdminRealtimeRows::AGENT_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();
    }
}