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

<?php

namespace App\Services\AdminRealtime;

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;

/**
 * 后台实时推送的访问控制：频道授权与登录过期断开。
 *
 * 频道授权响应在 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;
        }
        $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();
    }
}