package com.fs.live.websocket.service;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.fs.common.constant.LiveKeysConstant;
import com.fs.common.exception.base.BaseException;
import com.fs.common.utils.date.DateUtil;
import com.fs.his.domain.FsUser;
import com.fs.his.service.IFsUserService;
import com.fs.hisStore.domain.FsUserScrm;
import com.fs.hisStore.service.IFsUserScrmService;
import com.fs.live.config.ProductionWordFilter;
import com.fs.live.mapper.LiveCouponMapper;
import com.fs.live.vo.LiveWatchUserEntry;
import com.fs.live.domain.LiveWatchLog;
import com.fs.live.domain.LiveVideo;
import com.fs.live.domain.LiveUserBehaviorTrack;
import com.fs.live.service.ILiveUserBehaviorTrackService;
import com.fs.live.util.LiveBehaviorTrackUtil;
import com.fs.live.enums.LiveBehaviorTypeEnum;
import com.fs.live.enums.LiveBehaviorResourceTypeEnum;
import com.fs.live.vo.LiveWatchUserEntry;
import com.fs.live.domain.LiveWatchLog;
import com.fs.live.domain.LiveVideo;
import com.fs.live.websocket.auth.WebSocketConfigurator;
import com.fs.live.websocket.bean.SendMsgVo;
import com.fs.common.core.domain.R;
import com.fs.common.core.redis.RedisCache;
import com.fs.common.utils.StringUtils;
import com.fs.common.utils.redis.LiveDelayedTaskRedisUtil;
import com.fs.common.utils.spring.SpringUtils;
import com.fs.live.domain.*;
import com.fs.live.param.RedPO;
import com.fs.live.service.*;
import com.fs.live.vo.LiveGoodsVo;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.time.DateUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.websocket.*;
import javax.websocket.server.ServerEndpoint;
import java.io.EOFException;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import static com.fs.common.constant.LiveKeysConstant.*;

@ServerEndpoint(value = "/ws/app/webSocket",configurator = WebSocketConfigurator.class)
@Component
@Slf4j
public class WebSocketServer {


    // 上次点赞数缓存
    private final ConcurrentHashMap<Long, Integer> lastLikeCountCache = new ConcurrentHashMap<>();
    // 直播间用户session
    private final static ConcurrentHashMap<Long, ConcurrentHashMap<Long, Session>> rooms = new ConcurrentHashMap<>();
    // 管理端连接
    private final static ConcurrentHashMap<Long, CopyOnWriteArrayList<Session>> adminRooms = new ConcurrentHashMap<>();

    // Session发送锁，避免同一会话并发发送消息
    private final static ConcurrentHashMap<String, Lock> sessionLocks = new ConcurrentHashMap<>();
    // 心跳超时缓存：key=sessionId，value=最后心跳时间戳
    private final static ConcurrentHashMap<String, Long> heartbeatCache = new ConcurrentHashMap<>();
    // 心跳超时时间（毫秒）：3分钟无心跳则认为超时
    private final static long HEARTBEAT_TIMEOUT = 2 * 60 * 1000;
    // admin房间消息发送线程池（单线程，保证串行化）
    private final static ConcurrentHashMap<Long, ExecutorService> adminExecutors = new ConcurrentHashMap<>();

    // 消息队列系统
    // 每个直播间的消息队列，使用优先级队列支持管理员消息插队
    private final static ConcurrentHashMap<Long, PriorityBlockingQueue<QueueMessage>> messageQueues = new ConcurrentHashMap<>();
    // 每个直播间的消费者线程
    private final static ConcurrentHashMap<Long, Thread> consumerThreads = new ConcurrentHashMap<>();
    // 每个直播间的消费者线程控制标志
    private final static ConcurrentHashMap<Long, AtomicBoolean> consumerRunningFlags = new ConcurrentHashMap<>();
    // 每个直播间队列的总大小（字节数）
    private final static ConcurrentHashMap<Long, AtomicLong> queueSizes = new ConcurrentHashMap<>();
    // 消息队列最大容量：10000
    private final static int MAX_QUEUE_SIZE = 10000;
    // 消息队列最大大小：200MB
    private final static long MAX_QUEUE_SIZE_BYTES = 200L * 1024L * 1024L; // 200MB
    // 上下线消息采样率：10%
    private final static double ENTRY_EXIT_SAMPLE_RATE = 0.1;

    private final RedisCache redisCache = SpringUtils.getBean(RedisCache.class);
    private final ILiveMsgService liveMsgService = SpringUtils.getBean(ILiveMsgService.class);
    private final ILiveService liveService = SpringUtils.getBean(ILiveService.class);
    private final ILiveWatchUserService liveWatchUserService = SpringUtils.getBean(ILiveWatchUserService.class);
    private final IFsUserScrmService fsUserService = SpringUtils.getBean(IFsUserScrmService.class);
    private final ILiveDataService liveDataService = SpringUtils.getBean(ILiveDataService.class);
    private final ProductionWordFilter productionWordFilter = SpringUtils.getBean(ProductionWordFilter.class);
    private final ILiveRedConfService liveRedConfService =  SpringUtils.getBean(ILiveRedConfService.class);
    private final ILiveLotteryConfService liveLotteryConfService =  SpringUtils.getBean(ILiveLotteryConfService.class);
    private final ILiveGoodsService liveGoodsService =  SpringUtils.getBean(ILiveGoodsService.class);
    private final ILiveUserFirstEntryService liveUserFirstEntryService =  SpringUtils.getBean(ILiveUserFirstEntryService.class);
    private final ILiveCouponIssueService liveCouponIssueService =  SpringUtils.getBean(ILiveCouponIssueService.class);
    private final LiveCouponMapper liveCouponMapper = SpringUtils.getBean(LiveCouponMapper.class);
    private final ILiveWatchLogService liveWatchLogService = SpringUtils.getBean(ILiveWatchLogService.class);
    private final ILiveVideoService liveVideoService = SpringUtils.getBean(ILiveVideoService.class);
    private final ILiveCompletionPointsRecordService completionPointsRecordService = SpringUtils.getBean(ILiveCompletionPointsRecordService.class);
    private final ILiveUserBehaviorTrackService behaviorTrackService = SpringUtils.getBean(ILiveUserBehaviorTrackService.class);
    private static Random random = new Random();

    // Redis key 前缀：用户进入直播间时间
    public static final String USER_ENTRY_TIME_KEY = "live:user:entry:time:%s:%s"; // liveId:userId

    // 直播间在线用户缓存
//    private static final ConcurrentHashMap<Long, Integer> liveOnlineUsers = new ConcurrentHashMap<>();


    //建立连接成功调用
    @OnOpen
    public void onOpen(Session session) {

        Map<String, Object> userProperties = session.getUserProperties();
        long liveId = (long) userProperties.get("liveId");
        long userId = (long) userProperties.get("userId");
        long userType = (long) userProperties.get("userType");
        long qwUserId = -1;
        long externalContactId = -1;
        String location = (String) userProperties.get("location");  // 获取location参数

        Live live = liveService.selectLiveByLiveId(liveId);
        if (live == null) {
            throw new BaseException("未找到直播间");
        }
        long companyId = -1L;
        long companyUserId = -1L;
        if (!Objects.isNull(userProperties.get("companyId"))) {
            companyId = (long) userProperties.get("companyId");
        }
        if (!Objects.isNull(userProperties.get("companyUserId"))) {
            companyUserId = (long) userProperties.get("companyUserId");
        }
        if (!Objects.isNull(userProperties.get("qwUserId"))) {
            qwUserId = (long) userProperties.get("qwUserId");
        }
        if (!Objects.isNull(userProperties.get("externalContactId"))) {
            externalContactId = (long) userProperties.get("externalContactId");
        }


        ConcurrentHashMap<Long, Session> room = getRoom(liveId);
        List<Session> adminRoom = getAdminRoom(liveId);

        // 记录连接信息 管理员不记录
        if (userType == 0) {
            // 缓存用户信息，过期时间4小时
            String userCacheKey = "fs:user:" + userId;
            FsUserScrm fsUser = redisCache.getCacheObject(userCacheKey);
            if (fsUser == null) {
                fsUser = fsUserService.selectFsUserById(userId);
                if (fsUser != null) {
                    redisCache.setCacheObject(userCacheKey, fsUser, 4, TimeUnit.HOURS);
                }
            }
            if (Objects.isNull(fsUser)) {
                throw new BaseException("用户信息错误");
            }

            LiveWatchUser liveWatchUserVO = liveWatchUserService.join(fsUser,liveId, userId, location);
            room.put(userId, session);

            // 存储用户进入直播间的时间到 Redis（用于计算在线时长）
            // 如果已经存在进入时间，说明是重连，不应该覆盖，保持原来的进入时间
            String entryTimeKey = String.format(USER_ENTRY_TIME_KEY, liveId, userId);
            Long existingEntryTime = redisCache.getCacheObject(entryTimeKey);
            if (existingEntryTime == null) {
                // 首次连接，记录进入时间
                redisCache.setCacheObject(entryTimeKey, System.currentTimeMillis(), 24, TimeUnit.HOURS);
            }
            // 如果是重连，不覆盖进入时间，保持原来的进入时间以便正确计算总时长

            // 直播间浏览量 +1
            redisCache.incr(PAGE_VIEWS_KEY + liveId, 1);

            // 累计观看人次 +1
            redisCache.incr(TOTAL_VIEWS_KEY + liveId, 1);

            // 记录在线人数
            redisCache.incr(ONLINE_USERS_KEY + liveId, 1);
            // 将用户ID添加到在线用户Set中
            String onlineUsersSetKey = ONLINE_USERS_SET_KEY + liveId;
            redisCache.redisTemplate.opsForSet().add(onlineUsersSetKey, String.valueOf(userId));
            // 获取Set的大小作为当前在线人数
            Long currentOnlineCount = redisCache.redisTemplate.opsForSet().size(onlineUsersSetKey);
            //最大同时在线人数 - 使用Set大小来判断
            Integer maxOnline = redisCache.getCacheObject(MAX_ONLINE_USERS_KEY + liveId);
            int currentOnline = currentOnlineCount != null ? currentOnlineCount.intValue() : 0;
            if (maxOnline == null || currentOnline > maxOnline) {
                redisCache.setCacheObject(MAX_ONLINE_USERS_KEY + liveId, currentOnline);
            }

            // 判断是否是该直播间的首次访客（独立访客统计）
            boolean isFirstVisit = redisCache.setIfAbsent(USER_VISIT_KEY + liveId + ":" + userId, 1, 1, TimeUnit.DAYS);
            if (isFirstVisit) {
                redisCache.incr(UNIQUE_VISITORS_KEY + liveId, 1);
            }

            // 判断是否是首次进入直播间的观众
            boolean isFirstViewer = redisCache.setIfAbsent(UNIQUE_VIEWERS_KEY + liveId + ":" + userId, 1, 1, TimeUnit.DAYS);
            if (isFirstViewer) {
                redisCache.incr(UNIQUE_VIEWERS_KEY + liveId, 1);
            }
            liveWatchUserVO.setMsgStatus(liveWatchUserVO.getMsgStatus());
            // 上线消息采样10%进入队列
            if (random.nextDouble() < ENTRY_EXIT_SAMPLE_RATE) {
                SendMsgVo sendMsgVo = new SendMsgVo();
                sendMsgVo.setLiveId(liveId);
                sendMsgVo.setUserId(userId);
                sendMsgVo.setUserType(userType);
                sendMsgVo.setCmd("entry");
                sendMsgVo.setMsg("用户进入");
                sendMsgVo.setData(JSONObject.toJSONString(liveWatchUserVO));
                sendMsgVo.setNickName(fsUser.getNickname());
                sendMsgVo.setAvatar(fsUser.getAvatar());
                // 将上线消息加入队列
                enqueueMessage(liveId, JSONObject.toJSONString(R.ok().put("data", sendMsgVo)), false);
            }

            // 缓存用户首次进入记录，过期时间4小时
            String liveUserFirstEntryCacheKey = "live:userFirstEntry:" + liveId + ":" + userId;
            LiveUserFirstEntry liveUserFirstEntry = redisCache.getCacheObject(liveUserFirstEntryCacheKey);
            if (liveUserFirstEntry == null) {
                liveUserFirstEntry = liveUserFirstEntryService.selectEntityByLiveIdUserId(liveId, userId);
                if (liveUserFirstEntry != null) {
                    redisCache.setCacheObject(liveUserFirstEntryCacheKey, liveUserFirstEntry, 4, TimeUnit.HOURS);
                }
            }
            // 如果用户连上了 socket，并且公司ID和销售ID大于0，更新 LiveWatchLog 的 logType

            if ((qwUserId > 0 && externalContactId > 0) || (liveUserFirstEntry != null && liveUserFirstEntry.getCompanyId() > 0 && liveUserFirstEntry.getCompanyUserId() > 0 )) {
                // 获取当前直播/回放状态
                Map<String, Integer> flagMap = liveWatchUserService.getLiveFlagWithCache(liveId);
                Integer currentLiveFlag = flagMap.get("liveFlag");

                // 如果当前是直播状态（liveFlag = 1），更新 logType
                if (currentLiveFlag != null && currentLiveFlag == 1) {
                    updateLiveWatchLogTypeOnConnect(liveId, userId, qwUserId, externalContactId);
                }
            }


            if (liveUserFirstEntry != null) {
                // 处理第一次自己进入，第二次扫码销售进入
                if (liveUserFirstEntry.getCompanyUserId() == -1L && companyUserId != -1L) {
                    liveUserFirstEntry.setCompanyId(companyId);
                    liveUserFirstEntry.setCompanyUserId(companyUserId);
                    if (qwUserId != -1) {
                        liveUserFirstEntry.setQwUserId(qwUserId);
                    }
                    if (externalContactId!= -1) {
                        liveUserFirstEntry.setExternalContactId(externalContactId);
                    }
                    liveUserFirstEntryService.updateLiveUserFirstEntry(liveUserFirstEntry);
                }
            } else {
                // 这个用户A邀请用户b，b的业绩算a的销售的
                if (companyId == -2L) {
                    // 缓存用户首次进入记录，过期时间4小时
                    String clientBCacheKey = "live:userFirstEntry:" + liveId + ":" + companyUserId;
                    LiveUserFirstEntry clientB = redisCache.getCacheObject(clientBCacheKey);
                    if (clientB == null) {
                        clientB = liveUserFirstEntryService.selectEntityByLiveIdUserId(liveId, companyUserId);
                        if (clientB != null) {
                            redisCache.setCacheObject(clientBCacheKey, clientB, 4, TimeUnit.HOURS);
                        }
                    }
                    if (clientB != null) {
                        companyId = clientB.getCompanyId();
                        companyUserId = clientB.getCompanyUserId();
                    }
                }
                Date date = new Date();
                liveUserFirstEntry = new LiveUserFirstEntry();
                liveUserFirstEntry.setUserId(userId);
                liveUserFirstEntry.setLiveId(liveId);
                liveUserFirstEntry.setCompanyId(companyId);
                liveUserFirstEntry.setCompanyUserId(companyUserId);
                liveUserFirstEntry.setEntryDate(date);
                liveUserFirstEntry.setFirstEntryTime(date);
                liveUserFirstEntry.setUpdateTime( date);
                if (qwUserId != -1) {
                    liveUserFirstEntry.setQwUserId(qwUserId);
                }
                if (externalContactId!= -1) {
                    liveUserFirstEntry.setExternalContactId(externalContactId);
                }
                liveUserFirstEntryService.insertLiveUserFirstEntry(liveUserFirstEntry);
            }
            redisCache.setCacheObject( "live:user:first:entry:" + liveId + ":" + userId, liveUserFirstEntry, 4, TimeUnit.HOURS);

            // 推送完课积分倒计时配置信息给前端
//            sendCompletionPointsConfigToUser(session, liveId, userId, live);


        } else {
            adminRoom.add(session);
            // 为admin房间创建单线程执行器，保证串行化发送
            adminExecutors.computeIfAbsent(liveId, k -> Executors.newSingleThreadExecutor());
        }

        // 初始化Session锁
        sessionLocks.putIfAbsent(session.getId(), new ReentrantLock());
        // 初始化心跳时间
        heartbeatCache.put(session.getId(), System.currentTimeMillis());

        // 如果有session，启动消费者线程
        ConcurrentHashMap<Long, Session> tempRoom = getRoom(liveId);
        List<Session> tempAdminRoom = getAdminRoom(liveId);
        boolean hasSession = (tempRoom != null && !tempRoom.isEmpty()) ||
                            (tempAdminRoom != null && !tempAdminRoom.isEmpty());
        if (hasSession) {
            startConsumerThread(liveId);
        }

        pushLiveCommentBootstrap(session, liveId, userType);

    }

    //关闭连接时调用
    @OnClose
    public void onClose(Session session) {
        Map<String, Object> userProperties = session.getUserProperties();
        // 获取公司ID和销售ID
        long companyId = -1L;
        long companyUserId = -1L;
        if (!Objects.isNull(userProperties.get("companyId"))) {
            companyId = (long) userProperties.get("companyId");
        }
        if (!Objects.isNull(userProperties.get("companyUserId"))) {
            companyUserId = (long) userProperties.get("companyUserId");
        }

        long liveId = (long) userProperties.get("liveId");
        long userId = (long) userProperties.get("userId");
        long userType = (long) userProperties.get("userType");

        ConcurrentHashMap<Long, Session> room = getRoom(liveId);
        List<Session> adminRoom = getAdminRoom(liveId);
        if (userType == 0) {
            // 缓存用户信息，过期时间4小时
            String userCacheKey = "fs:user:" + userId;
            FsUserScrm fsUser = redisCache.getCacheObject(userCacheKey);
            if (fsUser == null) {
                fsUser = fsUserService.selectFsUserById(userId);
                if (fsUser != null) {
                    redisCache.setCacheObject(userCacheKey, fsUser, 4, TimeUnit.HOURS);
                }
            }
            if (Objects.isNull(fsUser)) {
                throw new BaseException("用户信息错误");
            }
            // 计算并更新用户在线时长
            room.remove(userId);
            if (room.isEmpty()) {
                rooms.remove(liveId);
            }


            // 直播间在线人数 -1
            redisCache.incr(ONLINE_USERS_KEY + liveId, -1);
            // 从在线用户Set中移除用户ID
            String onlineUsersSetKey = ONLINE_USERS_SET_KEY + liveId;
            redisCache.redisTemplate.opsForSet().remove(onlineUsersSetKey, String.valueOf(userId));

            LiveWatchUser liveWatchUserVO = liveWatchUserService.close(fsUser,liveId, userId);


            // 下线消息采样10%进入队列
            if (random.nextDouble() < ENTRY_EXIT_SAMPLE_RATE) {
                SendMsgVo sendMsgVo = new SendMsgVo();
                sendMsgVo.setLiveId(liveId);
                sendMsgVo.setUserId(userId);
                sendMsgVo.setUserType(userType);
                sendMsgVo.setCmd("out");
                sendMsgVo.setMsg("用户离开");
                sendMsgVo.setData(JSONObject.toJSONString(liveWatchUserVO));
                sendMsgVo.setNickName(fsUser.getNickname());
                sendMsgVo.setAvatar(fsUser.getAvatar());
                // 将下线消息加入队列
                enqueueMessage(liveId, JSONObject.toJSONString(R.ok().put("data", sendMsgVo)), false);
            }

        } else {
            adminRoom.remove(session);
            // 如果admin房间为空，关闭并清理执行器
            if (adminRoom.isEmpty()) {
                ExecutorService executor = adminExecutors.remove(liveId);
                if (executor != null) {
                    executor.shutdown();
                }
                adminRooms.remove(liveId);
            }
        }

        // 清理Session相关资源
        heartbeatCache.remove(session.getId());
        sessionLocks.remove(session.getId());

        // 检查并清理空的直播间资源
        cleanupEmptyRoom(liveId);
    }

    //收到客户端信息
    @OnMessage
    public void onMessage(Session session,String message) throws IOException {
        Map<String, Object> userProperties = session.getUserProperties();

        long liveId = (long) userProperties.get("liveId");
        long userType = (long) userProperties.get("userType");
        long companyUserId = -1L;
        if (!Objects.isNull(userProperties.get("companyUserId"))) {
            companyUserId = (long) userProperties.get("companyUserId");
        }
        boolean isAdmin = false;

        SendMsgVo msg = JSONObject.parseObject(message, SendMsgVo.class);
        if(msg.isOn()) return;
        LiveMsg liveMsg;
        try {
            switch (msg.getCmd()) {
                case "heartbeat":
                    // 更新心跳时间
                    heartbeatCache.put(session.getId(), System.currentTimeMillis());

                    // 心跳时同步更新观看时长到Redis Hash
                    long watchUserId = (long) userProperties.get("userId");



//                    if (msg.getData() != null && !msg.getData().isEmpty()) {
//                        try {
//                            Long currentDuration = Long.parseLong(msg.getData());
//
//                            Live currentLive = liveService.selectLiveByLiveId(liveId);
//                            if (currentLive == null) {
//                                break;
//                            }
//
//
//                            // 判断直播是否已开始：status=2(直播中) 或 当前时间 >= 开播时间
//                            boolean isLiveStarted = false;
//                            if (currentLive.getStatus() != null && currentLive.getStatus() == 2) {
//                                // status=2 表示直播中
//                                isLiveStarted = true;
//                            } else if (currentLive.getStartTime() != null) {
//                                // 判断当前时间是否已超过开播时间
//                                LocalDateTime now = LocalDateTime.now();
//                                isLiveStarted = now.isAfter(currentLive.getStartTime()) || now.isEqual(currentLive.getStartTime());
//                            }
//
//                            // 使用Hash结构存储：一个直播间一个Hash，包含所有用户的时长
//                            String hashKey = "live:watch:duration:hash:" + liveId;
//                            String userIdField = String.valueOf(watchUserId);
//
//                            if (!isLiveStarted) {
//                                redisCache.hashDelete(hashKey, userIdField);
//                                log.debug("[心跳-观看时长] 直播未开始，清除预播时长, liveId={}, userId={}", liveId, watchUserId);
//                                break;
//                            }
//
//                            // 直播已开始，记录观看时长
//                            // 获取现有时长
//                            Object existingDuration = redisCache.hashGet(hashKey, userIdField);
//                            // 只有当新的时长更大时才更新
//                            if (existingDuration == null || currentDuration > Long.parseLong(existingDuration.toString())) {
//                                // 更新Hash中的用户时长
//                                redisCache.hashPut(hashKey, userIdField, currentDuration.toString());
//                                // 设置过期时间（2小时）
//                                redisCache.expire(hashKey, 2, TimeUnit.HOURS);
//
//                                checkAndSendCompletionPointsInRealTime(liveId, watchUserId, currentDuration);
//
//                            }
//                        } catch (Exception e) {
//                            log.error("[心跳-观看时长] 更新失败, liveId={}, userId={}, data={}",
//                                    liveId, watchUserId, msg.getData(), e);
//                        }
//                    }

                    sendMessage(session, JSONObject.toJSONString(R.ok().put("data", msg)));
                    break;
                case "sendMsg":
                    msg.setMsg(productionWordFilter.filter(msg.getMsg()).getFilteredText());
                    if(StringUtils.isEmpty(msg.getMsg())) return;
                    liveMsg = new LiveMsg();
                    liveMsg.setLiveId(msg.getLiveId());
                    liveMsg.setUserId(msg.getUserId());
                    liveMsg.setNickName(msg.getNickName());
                    liveMsg.setAvatar(msg.getAvatar());
                    liveMsg.setMsg(msg.getMsg());
                    liveMsg.setCreateTime(new Date());

                    if (userType == 0) {
                        List<LiveWatchUser> liveWatchUser = liveWatchUserService.getByLiveIdAndUserId(msg.getLiveId(), msg.getUserId());
                        if(!liveWatchUser.isEmpty() && liveWatchUser.get(0).getMsgStatus() == 1){
                            sendMessage(session, JSONObject.toJSONString(R.error("你已被禁言")));
                            return;
                        }

                        Map<String, Integer> flagMap = liveWatchUserService.getLiveFlagWithCache(liveId);
                        Integer liveFlag = flagMap.get("liveFlag");
                        Integer replayFlag = flagMap.get("replayFlag");
                        liveMsg.setLiveFlag(liveFlag);
                        liveMsg.setReplayFlag(replayFlag);

                        liveMsgService.insertLiveMsg(liveMsg);

                        // 埋点：记录发送消息行为
                        try {
                            LiveUserBehaviorTrack track = LiveBehaviorTrackUtil.trackSendMessage(
                                msg.getUserId(), msg.getLiveId(), liveMsg.getMsgId());
                            behaviorTrackService.asyncInsertLiveUserBehaviorTrack(track);
                        } catch (Exception e) {
                            log.error("[埋点] 记录发送消息行为失败, liveId={}, userId={}", msg.getLiveId(), msg.getUserId(), e);
                        }

                        // 检查是否有进行中的抽奖，如果评论匹配关键词，写入Redis
                        try {
                            handleCommentLottery(msg.getLiveId(), msg.getUserId(), msg.getMsg());
                        } catch (Exception e) {
                            log.error("[评论抽奖] 处理失败, liveId={}, userId={}, msg={}", msg.getLiveId(), msg.getUserId(), msg.getMsg(), e);
                        }

                        // 检查是否有进行中的红包，如果评论匹配关键词，写入Redis
                        try {
                            handleCommentRed(msg.getLiveId(), msg.getUserId(), msg.getMsg());
                        } catch (Exception e) {
                            log.error("[评论抽红包] 处理失败, liveId={}, userId={}, msg={}", msg.getLiveId(), msg.getUserId(), msg.getMsg(), e);
                        }
                    }

                    msg.setOn(true);
                    msg.setData(JSONObject.toJSONString(liveMsg));

                    // 将消息加入队列（普通用户消息）
                    isAdmin = (userType == 1);
                    enqueueMessage(liveId, JSONObject.toJSONString(R.ok().put("data", msg)), isAdmin);
                    break;
                case "sendNormalMsg":
                    msg.setMsg(productionWordFilter.filter(msg.getMsg()).getFilteredText());
                    if(StringUtils.isEmpty(msg.getMsg())) return;
                    liveMsg = new LiveMsg();
                    liveMsg.setLiveId(msg.getLiveId());
                    liveMsg.setUserId(msg.getUserId());
                    liveMsg.setNickName(msg.getNickName());
                    liveMsg.setAvatar(msg.getAvatar());
                    liveMsg.setMsg(msg.getMsg());
                    liveMsg.setCreateTime(new Date());

                    // 根据直播状态设置live_flag或replay_flag
                    Live normalMsgLive = liveService.selectLiveByLiveId(msg.getLiveId());
                    if (normalMsgLive != null && normalMsgLive.getFinishTime() != null) {
                        Date finishTime = java.sql.Timestamp.valueOf(normalMsgLive.getFinishTime());
                        if (new Date().after(finishTime)) {
                            liveMsg.setReplayFlag(1);
                            liveMsg.setLiveFlag(0);
                        } else {
                            liveMsg.setLiveFlag(1);
                            liveMsg.setReplayFlag(0);
                        }
                    } else {
                        liveMsg.setLiveFlag(1);
                        liveMsg.setReplayFlag(0);
                    }

                    liveMsgService.insertLiveMsg(liveMsg);

                    // 埋点：记录普通消息行为
                    try {
                        LiveUserBehaviorTrack track = LiveBehaviorTrackUtil.trackSendMessage(
                            msg.getUserId(), msg.getLiveId(), liveMsg.getMsgId());
                        behaviorTrackService.asyncInsertLiveUserBehaviorTrack(track);
                    } catch (Exception e) {
                        log.error("[埋点] 记录普通消息行为失败, liveId={}, userId={}", msg.getLiveId(), msg.getUserId(), e);
                    }

                    msg.setOn(true);
                    msg.setData(JSONObject.toJSONString(liveMsg));
                    msg.setCmd("sendMsg");
                    enqueueMessage(liveId, JSONObject.toJSONString(R.ok().put("data", msg)), true);
                    break;
                case "sendPopMsg":
                    msg.setMsg(productionWordFilter.filter(msg.getMsg()).getFilteredText());
                    if(StringUtils.isEmpty(msg.getMsg())) return;
                    liveMsg = new LiveMsg();
                    liveMsg.setLiveId(msg.getLiveId());
                    liveMsg.setUserId(msg.getUserId());
                    liveMsg.setNickName(msg.getNickName());
                    liveMsg.setAvatar(msg.getAvatar());
                    liveMsg.setMsg(msg.getMsg());
                    msg.setOn(true);
                    msg.setData(JSONObject.toJSONString(liveMsg));
                    enqueueMessage(liveId, JSONObject.toJSONString(R.ok().put("data", msg)), true);
                    break;
                case "sendTopMsg":
                    msg.setMsg(productionWordFilter.filter(msg.getMsg()).getFilteredText());
                    if(StringUtils.isEmpty(msg.getMsg())) return;
                    liveMsg = new LiveMsg();
                    liveMsg.setLiveId(msg.getLiveId());
                    liveMsg.setUserId(msg.getUserId());
                    liveMsg.setNickName(msg.getNickName());
                    liveMsg.setAvatar(msg.getAvatar());
                    liveMsg.setMsg(msg.getMsg());
                    liveMsg.setEndTime(DateUtils.addMinutes(new Date(),msg.getDuration()).toString());
                    msg.setOn(true);
                    msg.setData(JSONObject.toJSONString(liveMsg));
                    enqueueMessage(liveId, JSONObject.toJSONString(R.ok().put("data", msg)), true);
                    // 放在当前活动里面
                    redisCache.deleteObject(String.format(LiveKeysConstant.LIVE_HOME_PAGE_CONFIG, liveId, TOP_MSG));
                    redisCache.setCacheObject(String.format(LiveKeysConstant.LIVE_HOME_PAGE_CONFIG, liveId, TOP_MSG), JSONObject.toJSONString(liveMsg));
                    redisCache.expire(String.format(LiveKeysConstant.LIVE_HOME_PAGE_CONFIG, liveId, TOP_MSG), msg.getDuration(), TimeUnit.MINUTES);
                    break;
                case "globalVisible":
                    msg.setOn(true);
                    liveWatchUserService.updateGlobalVisible(liveId, msg.getStatus());
                    liveService.updateGlobalVisible(liveId, msg.getStatus());
                    // 管理员消息插队
                    enqueueMessage(liveId, JSONObject.toJSONString(R.ok().put("data", msg)), true);
                    break;
                case "singleVisible":
                    liveWatchUserService.updateSingleVisible(liveId, msg.getStatus(),msg.getUserId());
                    // 管理员消息插队
                    enqueueMessage(liveId, JSONObject.toJSONString(R.ok().put("data", msg)), true);
                    break;
                case "sendGift":
                    break;
                case "blockUser":
                    sendBlockMessage(liveId, msg.getUserId());
                    break;
                case "goods":
                    sendGoodsMessage(msg);
                    break;
                case "deleteMsg":
                    deleteMsg(liveId,msg);
                    break;
                case "red":
                    processRed(liveId, msg);
                    break;
                case "lottery":
                    processLottery(liveId, msg);
                    break;
                case "coupon":
                    processCoupon(liveId, msg);
                    break;
                case "floatScreenMsg":
                    msg.setMsg(productionWordFilter.filter(msg.getMsg()).getFilteredText());
                    if (StringUtils.isEmpty(msg.getMsg())) {
                        return;
                    }
                    ILiveCommentFloatScreenService floatScreenService = SpringUtils.getBean(ILiveCommentFloatScreenService.class);
                    Long checkCompanyUserId = (companyUserId > 0)
                            ? companyUserId
                            : (msg.getCompanyUserId() != null && msg.getCompanyUserId() > 0 ? msg.getCompanyUserId() : msg.getUserId());
                    R floatR = floatScreenService.handleFloatScreen(liveId, msg.getUserId(), msg.getNickName(), msg.getAvatar(),
                            msg.getMsg(), msg.getLiveRoleCode(), checkCompanyUserId);
                    if (!Integer.valueOf(200).equals(floatR.get("code"))) {
                        sendMessage(session, JSONObject.toJSONString(floatR));
                        break;
                    }
                    LiveMsg flm = (LiveMsg) floatR.get("liveMsg");
                    msg.setCmd("floatScreenDisplay");
                    msg.setOn(true);
                    msg.setData(JSONObject.toJSONString(flm));
                    enqueueMessage(liveId, JSONObject.toJSONString(R.ok().put("data", msg)), true);
                    break;
                case "pinComment":
                    if (msg.getTargetMsgId() == null) {
                        sendMessage(session, JSONObject.toJSONString(R.error("缺少 targetMsgId")));
                        break;
                    }
                    int pinDur = msg.getDuration() != null ? msg.getDuration() : -1;
                    ILiveCommentPinService pinService = SpringUtils.getBean(ILiveCommentPinService.class);
                    R pinR = pinService.pinComment(liveId, msg.getTargetMsgId(), msg.getUserId(), msg.getNickName(),
                            msg.getLiveRoleCode(), (int) userType, pinDur,
                            (companyUserId > 0 ? companyUserId :
                                    (msg.getCompanyUserId() != null && msg.getCompanyUserId() > 0 ? msg.getCompanyUserId() : msg.getUserId())));
                    sendMessage(session, JSONObject.toJSONString(pinR));
                    break;
                case "unpinComment":
                    if (msg.getTargetMsgId() == null) {
                        sendMessage(session, JSONObject.toJSONString(R.error("缺少 targetMsgId")));
                        break;
                    }
                    ILiveCommentPinService pinService2 = SpringUtils.getBean(ILiveCommentPinService.class);
                    R unpinR = pinService2.unpinComment(liveId, msg.getTargetMsgId(), msg.getUserId(), msg.getLiveRoleCode(), (int) userType,
                            (companyUserId > 0 ? companyUserId :
                                    (msg.getCompanyUserId() != null && msg.getCompanyUserId() > 0 ? msg.getCompanyUserId() : msg.getUserId())));
                    sendMessage(session, JSONObject.toJSONString(unpinR));
                    break;
                case "delAutoTask":
                    if (userType == 1) {
                        delAutoTask(liveId, DateUtils.parseDate(msg.getData(),"yyyy-MM-dd'T'HH:mm:ss.SSSZ").getTime());
                    }
                    break;
            }
        } catch (Exception e) {
            log.error("webSocket 消息处理失败 msg: {}", e.getMessage(), e);
        }
    }

    private void deleteMsg(long liveId,SendMsgVo msg) {
        SendMsgVo sendMsgVo = new SendMsgVo();
        sendMsgVo.setLiveId(liveId);
        sendMsgVo.setUserType(0L);
        sendMsgVo.setCmd("deleteMsg");
        sendMsgVo.setMsg(msg.getMsg());
        // 管理员消息插队
        enqueueMessage(liveId, JSONObject.toJSONString(R.ok().put("data", sendMsgVo)), true);
    }

    private void processCoupon(long liveId, SendMsgVo msg) {
        JSONObject jsonObject = JSON.parseObject(msg.getData());
        Integer status = jsonObject.getInteger("status");
        msg.setStatus(status);
        Long couponIssueId = jsonObject.getLong("couponIssueId");
        // ①  检查  缓存是否存在  ② 如果是发布 放入缓存 ③ 删除缓存
        String couponNumKey = LiveKeysConstant.liveCouponNumKey(couponIssueId);
        LiveDelayedTaskRedisUtil delayedTaskRedisUtil = SpringUtils.getBean(LiveDelayedTaskRedisUtil.class);
        if (status == 1) {
            Object cacheObject = redisCache.getCacheObject(couponNumKey);
            if (cacheObject == null) {
                LiveCouponIssue liveCoupon = liveCouponIssueService.selectLiveCouponIssueById(couponIssueId);
                if (liveCoupon != null) {
                    redisCache.setCacheObject(couponNumKey, liveCoupon.getRemainCount().intValue(), 30, TimeUnit.MINUTES);
                    delayedTaskRedisUtil.trackCouponNum(couponIssueId);
                }
            }
        } else {
            redisCache.deleteObject(couponNumKey);
            delayedTaskRedisUtil.untrackCouponNum(couponIssueId);
        }
        // 管理员消息插队
        enqueueMessage(liveId, JSONObject.toJSONString(R.ok().put("data", msg)), true);
    }


    private void sendGoodsMessage(SendMsgVo msg) {
        JSONObject jsonObject = JSON.parseObject(msg.getData());
        Long goodsId = jsonObject.getLong("goodsId");
        Long liveId = jsonObject.getLong("liveId");
        Integer status = jsonObject.getInteger("status");
        msg.setStatus(status);
        LiveGoodsVo liveGoods = liveGoodsService.selectLiveGoodsVoByGoodsId(goodsId);
        if(liveGoods == null) return;
        liveService.asyncToCacheLiveConfig(liveId);
        msg.setLiveId(liveId);
        msg.setData(JSONObject.toJSONString(liveGoods));
        // 管理员消息插队
        enqueueMessage(liveId, JSONObject.toJSONString(R.ok().put("data", msg)), true);
    }

    /**
     * 处理红包变动消息
     */
    private void processRed(Long liveId, SendMsgVo msg) {
        JSONObject jsonObject = JSON.parseObject(msg.getData());
        Integer status = jsonObject.getInteger("status");
        msg.setStatus( status);
        LiveRedConf liveRedConf = liveRedConfService.selectLiveRedConfByRedId(jsonObject.getLong("redId"));
        if (Objects.nonNull(liveRedConf)) {
            liveService.asyncToCacheLiveConfig(liveId);
            msg.setData(JSONObject.toJSONString(liveRedConf));
            // 管理员消息插队
            enqueueMessage(liveId, JSONObject.toJSONString(R.ok().put("data", msg)), true);
        }
    }

    /**
     * 处理抽奖变动消息
     */
    private void processLottery(Long liveId, SendMsgVo msg) {
        JSONObject jsonObject = JSON.parseObject(msg.getData());
        Integer status = jsonObject.getInteger("status");
        msg.setStatus( status);
        LiveLotteryConf liveLotteryConf = liveLotteryConfService.selectLiveLotteryConfByLotteryId(jsonObject.getLong("lotteryId"));
        if (Objects.nonNull(liveLotteryConf)) {
            liveService.asyncToCacheLiveConfig(liveId);
            msg.setData(JSONObject.toJSONString(liveLotteryConf));
            // 管理员消息插队
            enqueueMessage(liveId, JSONObject.toJSONString(R.ok().put("data", msg)), true);
        }
    }

    //错误时调用
    @OnError
    public void onError(Session session, Throwable throwable) {

        try {
            this.onClose(session);
        } catch (Exception e) {
            log.error("webSocket 错误处理失败", e);
        }
    }

    /**
     * 获取房间
     * @param liveId 直播间ID
     * @return 容器
     */
    private ConcurrentHashMap<Long, Session> getRoom(Long liveId) {
        return rooms.computeIfAbsent(liveId, k -> new ConcurrentHashMap<>());
    }

    /**
     * 获取当前所有"有在线用户"的直播间ID集合及其在线人数快照。
     * 供订单动效、在线统计等外部模块按房间状态调度任务使用。
     */
    public static java.util.Map<Long, Integer> getActiveRoomsSnapshot() {
        java.util.Map<Long, Integer> snapshot = new java.util.HashMap<>(rooms.size());
        for (Map.Entry<Long, ConcurrentHashMap<Long, Session>> entry : rooms.entrySet()) {
            ConcurrentHashMap<Long, Session> room = entry.getValue();
            if (room != null && !room.isEmpty()) {
                snapshot.put(entry.getKey(), room.size());
            }
        }
        return snapshot;
    }

    /**
     * 获取管理端房间
     * @param liveId  直播间ID
     * @return  容器
     */
    private List<Session> getAdminRoom(Long liveId) {
        return adminRooms.computeIfAbsent(liveId, k -> new CopyOnWriteArrayList<>());
    }

    //发送消息（带锁机制，避免并发发送）
    public void sendMessage(Session session, String message) throws IOException {
        if (session == null || !session.isOpen()) {
            return;
        }

        // 获取Session锁
        Lock lock = sessionLocks.get(session.getId());
        if (lock == null) {
            // 如果锁不存在，创建一个新锁
            lock = sessionLocks.computeIfAbsent(session.getId(), k -> new ReentrantLock());
        }

        // 使用锁保证同一Session的消息串行发送
        lock.lock();
        try {
            if (session.isOpen()) {
                session.getAsyncRemote().sendText(message);
            }
        } finally {
            lock.unlock();
        }
    }

    public void sendIntegralMessage(Long liveId, Long userId,Long scoreAmount) {
        ConcurrentHashMap<Long, Session> room = getRoom(liveId);
        Session session = room.get(userId);
        if (session == null || !session.isOpen()) {
            return;
        }
        SendMsgVo sendMsgVo = new SendMsgVo();
        sendMsgVo.setLiveId(liveId);
        sendMsgVo.setUserId(userId);
        sendMsgVo.setUserType(0L);
        sendMsgVo.setCmd("Integral");
        sendMsgVo.setMsg("恭喜你成功获得观看奖励:" + scoreAmount + "芳华币");
        sendMsgVo.setData(String.valueOf(scoreAmount));

        if(Objects.isNull( session)) return;
        // 使用带锁的sendMessage方法，保证线程安全
        try {
            sendMessage(session, JSONObject.toJSONString(R.ok().put("data", sendMsgVo)));
        } catch (IOException e) {
            log.error("发送积分消息失败: liveId={}, userId={}, error={}", liveId, userId, e.getMessage(), e);
        }
    }

    /**
     * 发送完课积分弹窗通知给特定用户
     */
    public void sendCompletionPointsMessage(Long liveId, Long userId, SendMsgVo sendMsgVo) {
        ConcurrentHashMap<Long, Session> room = getRoom(liveId);
        Session session = room.get(userId);
        if (session == null || !session.isOpen()) {
            return;
        }
        try {
            sendMessage(session, JSONObject.toJSONString(R.ok().put("data", sendMsgVo)));
        } catch (IOException e) {
            log.error("发送完课积分消息失败: liveId={}, userId={}, error={}", liveId, userId, e.getMessage(), e);
        }
    }

    private void sendBlockMessage(Long liveId, Long userId) {

        ConcurrentHashMap<Long, Session> room = getRoom(liveId);
        Session session = room.get(userId);
        if (session == null || !session.isOpen()) {
            return;
        }

        SendMsgVo sendMsgVo = new SendMsgVo();
        sendMsgVo.setLiveId(liveId);
        sendMsgVo.setUserId(userId);
        sendMsgVo.setUserType(0L);
        sendMsgVo.setCmd("blockUser");
        sendMsgVo.setMsg("账号已被停用");
        sendMsgVo.setData(null);

        if(Objects.isNull( session)) return;
        // 使用带锁的sendMessage方法，保证线程安全
        try {
            sendMessage(session, JSONObject.toJSONString(R.ok().put("data", sendMsgVo)));
        } catch (IOException e) {
            log.error("发送封禁消息失败: liveId={}, userId={}, error={}", liveId, userId, e.getMessage(), e);
        }
    }

    /**
     * 向指定直播间推送订单动效提示（走现有优先级消息队列，作为普通优先级消息）。
     *
     * <p>走队列的好处：</p>
     * <ul>
     *   <li>与管理员消息共享优先级机制 —— 管理员消息仍可插队</li>
     *   <li>触发队列容量与大小保护 —— 高峰期可自动丢弃</li>
     *   <li>由消费者线程异步广播，外部调用方不阻塞</li>
     * </ul>
     *
     * @param liveId  直播间ID
     * @param message 完整消息 JSON（R.ok().put("data", SendMsgVo) 的序列化结果）
     */
    public boolean enqueueOrderTipMessage(Long liveId, String message) {
        if (liveId == null || message == null) {
            return false;
        }
        // 仅在直播间有在线用户时才入队，避免无效堆积
        ConcurrentHashMap<Long, Session> room = rooms.get(liveId);
        if (room == null || room.isEmpty()) {
            return false;
        }
        return enqueueMessage(liveId, message, false);
    }

    /**
     * 广播消息
     * @param liveId   直播间ID
     * @param message  消息内容
     * 优化：使用快照遍历，避免在遍历过程中修改集合
     */
    public void broadcastWebMessage(Long liveId, String message) {
        ConcurrentHashMap<Long, Session> room = getRoom(liveId);

        if (room.isEmpty()) {
            return;
        }

        // 普通用户房间：并行发送（使用快照遍历，避免并发修改）
        // ConcurrentHashMap 的 entrySet() 是弱一致性的，但为了更安全，我们显式创建快照
        for (Map.Entry<Long, Session> entry : room.entrySet()) {
            Session session = entry.getValue();
            if (session != null && session.isOpen()) {
                sendWithRetry(session, message, 3);
            }
        }
    }

    /**
     * 广播消息
     * @param liveId   直播间ID
     * @param message  消息内容
     * 优化：使用快照遍历，避免在遍历过程中修改集合
     */
    public void broadcastMessage(Long liveId, String message) {
        ConcurrentHashMap<Long, Session> room = getRoom(liveId);
        List<Session> adminRoom = getAdminRoom(liveId);

        // 普通用户房间：并行发送（使用快照遍历，避免并发修改）
        if (!room.isEmpty()) {
            for (Map.Entry<Long, Session> entry : room.entrySet()) {
                Session session = entry.getValue();
                if (session != null && session.isOpen()) {
                    sendWithRetry(session, message, 1);
                }
            }
        }

        // admin房间：串行发送，使用单线程执行器
        if (!adminRoom.isEmpty()) {
            ExecutorService executor = adminExecutors.get(liveId);
            if (executor != null && !executor.isShutdown()) {
                executor.submit(() -> {
                    for (Session session : adminRoom) {
                        if (session.isOpen()) {
                            sendWithRetry(session, message, 1);
                        }
                    }
                });
            } else {
                // 如果执行器不存在或已关闭，直接发送
                adminRoom.forEach(v -> {
                    if (v.isOpen()) {
                        sendWithRetry(v, message, 1);
                    }
                });
            }
        }
    }

    public void removeLikeCountCache(Long liveId) {
        lastLikeCountCache.remove(liveId);
    }
    @Scheduled(fixedRate = 300)// 每分钟执行一次
    public void broadcastLikeMessage() {
        Set<Long> activeLiveIds = new HashSet<>();
        for (Map.Entry<Long, ConcurrentHashMap<Long, Session>> entry : rooms.entrySet()) {
            Long liveId = entry.getKey();
            activeLiveIds.add(liveId);
            String likeKey = "live:like:" + liveId;
            Object cacheObject = redisCache.getCacheObject(likeKey);
            if(cacheObject == null) continue;
            Integer current = null;
            try {
                String valueStr = cacheObject.toString().trim();
                current = Integer.parseInt(valueStr);
            } catch (NumberFormatException e) {
                continue;
            }
            Integer last = lastLikeCountCache.getOrDefault(liveId, 0);
            if (!current.equals(last)) {
                SendMsgVo sendMsgVo = new SendMsgVo();
                sendMsgVo.setLiveId(liveId);
                sendMsgVo.setCmd("likeDetail");
                sendMsgVo.setData(JSON.toJSONString(current));
                broadcastLikeMessage(liveId, JSONObject.toJSONString(R.ok().put("data",sendMsgVo)));
                lastLikeCountCache.put(liveId, current); // 更新上次值
            }
        }
        lastLikeCountCache.keySet().removeIf(liveId -> !activeLiveIds.contains(liveId));
    }


    @Scheduled(fixedRate = 2000)// 每2秒执行一次
    public void broadcastUserNumMessage() {
        // 遍历每个直播间
        for (Map.Entry<Long, ConcurrentHashMap<Long, Session>> entry : rooms.entrySet()) {
            Long liveId = entry.getKey();
            ConcurrentHashMap<Long, Session> room = entry.getValue();

            // 统计当前直播间的在线人数
            int onlineCount = room.size();

            // 构造消息
            SendMsgVo sendMsgVo = new SendMsgVo();
            sendMsgVo.setLiveId(liveId);
            sendMsgVo.setCmd("userCount");
            sendMsgVo.setData(String.valueOf(onlineCount));

            // 广播当前直播间的在线人数
            broadcastMessage(liveId, JSONObject.toJSONString(R.ok().put("data", sendMsgVo)));
        }
    }


    /**
     * 定时清理无效会话（每分钟执行一次）
     * 检查心跳超时的会话并关闭
     * 优化：使用快照遍历，避免在遍历过程中修改集合
     */
    @Scheduled(fixedRate = 60000) // 每分钟执行一次
    public void cleanInactiveSessions() {
        long currentTime = System.currentTimeMillis();
        int cleanedCount = 0;

        // 遍历所有直播间（使用快照，避免在遍历过程中被修改影响）
        for (Map.Entry<Long, ConcurrentHashMap<Long, Session>> roomEntry : rooms.entrySet()) {
            Long liveId = roomEntry.getKey();
            ConcurrentHashMap<Long, Session> room = roomEntry.getValue();

            // 如果房间为空，跳过
            if (room.isEmpty()) {
                continue;
            }

            // 检查普通用户会话（使用快照遍历，避免并发修改异常）
            List<Long> toRemove = new ArrayList<>();
            // 创建快照，避免在遍历过程中修改原集合
            for (Map.Entry<Long, Session> userEntry : room.entrySet()) {
                Long userId = userEntry.getKey();
                Session session = userEntry.getValue();

                if (session == null) {
                    toRemove.add(userId);
                    continue;
                }

                Long lastHeartbeat = heartbeatCache.get(session.getId());
                if (lastHeartbeat != null && (currentTime - lastHeartbeat) > HEARTBEAT_TIMEOUT) {
                    toRemove.add(userId);
                    try {
                        if (session.isOpen()) {
                            session.close(new CloseReason(CloseReason.CloseCodes.NORMAL_CLOSURE, "心跳超时"));
                        }
                        liveWatchUserService.close(null, liveId, userId);
                    } catch (Exception e) {
                        log.error("关闭超时会话失败: sessionId={}, liveId={}, userId={}",
                                session.getId(), liveId, userId, e);
                    }
                }
            }

            // 移除超时的会话
            if (!toRemove.isEmpty()) {
                String hashKey = String.format(LiveKeysConstant.LIVE_WATCH_USERS, liveId);
                for (Long userId : toRemove) {
                    room.remove(userId);
                    // 从 Redis hash 中删除无效用户
                    redisCache.hashDelete(hashKey, String.valueOf(userId));
                }
                cleanedCount += toRemove.size();
            }
        }

        // 检查admin房间
        for (Map.Entry<Long, CopyOnWriteArrayList<Session>> adminEntry : adminRooms.entrySet()) {
            Long liveId = adminEntry.getKey();
            CopyOnWriteArrayList<Session> adminRoom = adminEntry.getValue();

            List<Session> toRemoveAdmin = new ArrayList<>();
            for (Session session : adminRoom) {
                Long lastHeartbeat = heartbeatCache.get(session.getId());
                if (lastHeartbeat != null && (currentTime - lastHeartbeat) > HEARTBEAT_TIMEOUT) {
                    toRemoveAdmin.add(session);
                    try {
                        if (session.isOpen()) {
                            session.close(new CloseReason(CloseReason.CloseCodes.NORMAL_CLOSURE, "心跳超时"));
                        }
                    } catch (Exception e) {
                        log.error("关闭admin超时会话失败: sessionId={}, liveId={}",
                                session.getId(), liveId, e);
                    }
                }
            }

            // 移除超时的admin会话
            toRemoveAdmin.forEach(adminRoom::remove);
            cleanedCount += toRemoveAdmin.size();
        }

        if (cleanedCount > 0) {
            if (random.nextInt(10) == 1) {
                log.info("已清理 {} 个无效会话", cleanedCount);
            }
        }
    }


    /**
     * 广播点赞消息
     * @param liveId   直播间ID
     * @param message  消息内容
     * 优化：使用快照遍历，避免在遍历过程中修改集合
     */
    public void broadcastLikeMessage(Long liveId, String message) {
        ConcurrentHashMap<Long, Session> room = getRoom(liveId);

        if (room.isEmpty()) {
            return;
        }

        // 使用快照遍历，避免并发修改
        for (Map.Entry<Long, Session> entry : room.entrySet()) {
            Session session = entry.getValue();
            if (session != null && session.isOpen()) {
                sendWithRetry(session, message, 3);
            }
        }
    }

    private void sendWithRetry(Session session, String message, int maxRetries) {
        if (session == null || !session.isOpen()) {
            return;
        }

        int attempts = 0;
        while (attempts < maxRetries) {
            try {
                // 使用带锁的sendMessage方法，避免并发发送
                sendMessage(session, message);
                return;  // 发送成功，退出
            } catch (Exception e) {
                if (e.getMessage() != null && e.getMessage().contains("TEXT_FULL_WRITING")) {
                    attempts++;
                    try {
                        TimeUnit.MILLISECONDS.sleep(ThreadLocalRandom.current().nextInt(5, 100));
                    } catch (InterruptedException ie) {
                        Thread.currentThread().interrupt();
                        break;
                    }
                } else {
                    log.error("发送消息失败: sessionId={}, error={}", session.getId(), e.getMessage(), e);
                    break;
                }
            }
        }

        if (attempts >= maxRetries) {
            log.warn("超过重试次数({})，放弃发送消息: sessionId={}", maxRetries, session.getId());
        }
    }



    public void handleAutoTask(LiveAutoTask task) {
        SendMsgVo msg = new SendMsgVo();
        msg.setLiveId(task.getLiveId());
        msg.setData(task.getContent());
        Date now = new Date();
        try {
            if (task.getTaskType() == 1L) {
                msg.setCmd("goods");
                LiveGoodsVo liveGoodsVo = JSON.parseObject(task.getContent(), LiveGoodsVo.class);
                liveGoodsService.updateLiveIsShow(liveGoodsVo.getGoodsId(), task.getLiveId());
                liveService.asyncToCacheLiveConfig(task.getLiveId());
            }else if (task.getTaskType() == 2L) {
                msg.setCmd("red");
                LiveRedConf liveRedConf = JSON.parseObject(task.getContent(), LiveRedConf.class);
                if(liveRedConf.getRedStatus() != 0){
                    log.error("红包状态应该为：未发放");
                    return;
                }
                liveRedConf.setRedStatus(1L);
                liveRedConf.setUpdateTime( now);
                msg.setData(JSON.toJSONString(liveRedConf));
                liveRedConfService.updateLiveRedConf(liveRedConf);
                liveService.asyncToCacheLiveConfig(task.getLiveId());
            }else if (task.getTaskType() == 4L) {
                msg.setCmd("lottery");
                LiveLotteryConf liveLotteryConf = JSON.parseObject(task.getContent(), LiveLotteryConf.class);
                if(!"0".equals(liveLotteryConf.getLotteryStatus()) ){
                    log.error("抽奖状态应该为：未发放");
                    return;
                }
                liveLotteryConf.setLotteryStatus("1");
                liveLotteryConf.setUpdateTime( now);
                msg.setData(JSON.toJSONString(liveLotteryConf));
                liveLotteryConfService.updateLiveLotteryConf(liveLotteryConf);
                liveService.asyncToCacheLiveConfig(task.getLiveId());
            }else if (task.getTaskType() == 3L) {
                msg.setCmd("sendMsg");
                msg.setMsg(task.getContent());
                msg.setUserType(1L);
                msg.setNickName(task.getTaskName());
                LiveMsg liveMsg = new LiveMsg();
                liveMsg.setLiveId(task.getLiveId());
                liveMsg.setUserId(0L);
                liveMsg.setNickName(task.getTaskName());
                liveMsg.setMsg(task.getContent());
                liveMsg.setCreateTime(now);
                liveMsg.setUpdateTime(now);
                liveMsgService.insertLiveMsg(liveMsg);
            } else if (task.getTaskType() == 5L) {
                msg.setCmd("coupon");
                LiveCoupon liveCoupon = JSON.parseObject(task.getContent(), LiveCoupon.class);
                if(liveCoupon.getStatus() != 1 ){
                    log.error("优惠券状态已经关闭");
                    return;
                }
                LiveCouponIssue liveCouponIssue = liveCouponIssueService.selectLiveCouponIssueByCouponId(liveCoupon.getCouponId());
                LiveCouponIssueRelation relation = liveCouponMapper.selectCouponRelation(task.getLiveId(), liveCouponIssue.getId());
                if (liveCoupon != null) {
                    redisCache.setCacheObject(LiveKeysConstant.liveCouponNumKey(liveCouponIssue.getId()), liveCouponIssue.getRemainCount().intValue(), 30, TimeUnit.MINUTES);
                    SpringUtils.getBean(LiveDelayedTaskRedisUtil.class).trackCouponNum(liveCouponIssue.getId());
                }
                HashMap<String, Object> data = new HashMap<>();
                data.put("liveId", task.getLiveId());
                data.put("couponIssueId", liveCouponIssue.getId());
                data.put("status", 1);
                data.put("goodsId", relation.getGoodsId());
                data.put("couponName", liveCoupon.getTitle());
                data.put("couponPrice", liveCoupon.getCouponPrice());
                data.put("useMinPrice", liveCoupon.getUseMinPrice());
                data.put("couponTime", liveCoupon.getCouponTime());
                msg.setData(JSON.toJSONString(data));
                liveCouponMapper.updateChangeShow(task.getLiveId(), liveCouponIssue.getId());
            } else if (task.getTaskType() == 6L) {
                // 上架/下架商品
                msg.setCmd("goods");
                JSONObject jsonObject = JSON.parseObject(task.getContent());
                Long goodsId = jsonObject.getLong("goodsId");
                Integer status = jsonObject.getInteger("status");
                if (goodsId == null || status == null) {
                    log.error("商品ID或状态为空");
                    return;
                }
                liveGoodsService.updateLiveGoodsStatus(goodsId, status);
                return;
            } else if (task.getTaskType() == 7L) {
                // 自动化抽奖：开始
                handleAutoLotteryStart(task);
                return;
            } else if (task.getTaskType() == 8L) {
                // 自动化抽奖：结算
                handleAutoLotterySettle(task);
                return;
            } else if (task.getTaskType() == 9L) {
                // 自动化红包：开始
                handleAutoRedStart(task);
                return;
            } else if (task.getTaskType() == 10L) {
                // 自动化红包：结算
                handleAutoRedSettle(task);
                return;
            }
            msg.setStatus(1);
            // 定时任务消息作为管理员消息插队
            enqueueMessage(task.getLiveId(), JSONObject.toJSONString(R.ok().put("data", msg)), true);
        } catch (Exception e) {
            log.error("定时任务执行异常：{}", e.getMessage());
        }

    }
    private void delAutoTask(long liveId, Long data) {
        String cacheKey = LiveKeysConstant.liveAutoTaskKey(liveId);
        redisCache.redisTemplate.opsForZSet().removeRangeByScore(cacheKey, data, data);
        SpringUtils.getBean(LiveDelayedTaskRedisUtil.class).refreshAutoTaskIndex(liveId);
    }

    /**
     * 计算并更新用户在线时长
     * @param liveId 直播间ID
     * @param userId 用户ID
     * @param companyId 公司ID
     * @param companyUserId 销售ID
     */
    private void updateUserOnlineDuration(Long liveId, Long userId, Long companyId, Long companyUserId) {
        try {
            // 从 Redis 获取用户进入时间
            String entryTimeKey = String.format(USER_ENTRY_TIME_KEY, liveId, userId);
            Long entryTime = redisCache.getCacheObject(entryTimeKey);

            if (entryTime == null) {
                // 如果没有进入时间记录，可能是旧数据，跳过
                return;
            }

            long currentTimeMillis = System.currentTimeMillis();
            Date now = new Date();

            // 计算在线时长（秒）
            long durationSeconds = (currentTimeMillis - entryTime) / 1000;

            if (durationSeconds <= 0) {
                return;
            }

            // 获取当前直播/回放状态
            Map<String, Integer> flagMap = liveWatchUserService.getLiveFlagWithCache(liveId);
            Integer currentLiveFlag = flagMap.get("liveFlag");
            Integer currentReplayFlag = flagMap.get("replayFlag");

            // 查询用户记录
            LiveWatchUserEntry liveWatchUser = liveWatchUserService.selectLiveWatchAndCompanyUserByFlag(
                    liveId, userId, currentLiveFlag, currentReplayFlag);

            if (liveWatchUser != null) {
                // 累加在线时长
                Long onlineSeconds = liveWatchUser.getOnlineSeconds();
                if (onlineSeconds == null) {
                    onlineSeconds = 0L;
                }
                liveWatchUser.setOnlineSeconds(onlineSeconds + durationSeconds);
                liveWatchUser.setUpdateTime(now);

                // 更新数据库
                liveWatchUserService.updateLiveWatchUserEntry(liveWatchUser);
                // 如果 LiveWatchUserEntry 存在，并且当前是直播状态（liveFlag = 1），更新 LiveWatchLog
//                if (currentLiveFlag != null && currentLiveFlag == 1
//                        && liveWatchUser.getCompanyId() != null && liveWatchUser.getCompanyId() > 0
//                        && liveWatchUser.getCompanyUserId() != null && liveWatchUser.getCompanyUserId() > 0) {
//                    updateLiveWatchLogTypeByDuration(liveId, userId,
//                            liveWatchUser.getCompanyId(), liveWatchUser.getCompanyUserId(),
//                            liveWatchUser.getOnlineSeconds());
//                }
            }

            // 删除 Redis 中的进入时间记录
            redisCache.deleteObject(entryTimeKey);
        } catch (Exception e) {
            log.error("更新用户在线时长异常：liveId={}, userId={}, error={}",
                    liveId, userId, e.getMessage(), e);
        }
    }

    /**
     * 在连接时更新 LiveWatchLog 的 logType
     * 如果 logType 类型不是 2，修改 logType 类型为 1（看课中）
     */
    private void updateLiveWatchLogTypeOnConnect(Long liveId, Long userId, Long qwUserId, Long externalContactId) {
        try {
            LiveWatchLog queryLog = new LiveWatchLog();
            queryLog.setLiveId(liveId);
            queryLog.setQwUserId(String.valueOf(qwUserId));
            queryLog.setExternalContactId(externalContactId);
            DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
            List<LiveWatchLog> logs = liveWatchLogService.selectLiveWatchLogList(queryLog);
            if (logs != null && !logs.isEmpty()) {
                for (LiveWatchLog log : logs) {
                    // 如果 logType 不是 2（完课），则更新为 1（看课中）
                    if (log.getLogType() == null || log.getLogType() != 2) {
                        log.setLogType(1);
                        liveWatchLogService.updateLiveWatchLog(log);
                        String liveUserWatchLogKey = LiveKeysConstant.liveUserWatchLogKey(liveId, userId, externalContactId, qwUserId);
                        LocalDateTime now = LocalDateTime.now();
                        redisCache.setCacheObject(liveUserWatchLogKey, formatter.format(now), 5, TimeUnit.MINUTES);
                        SpringUtils.getBean(LiveDelayedTaskRedisUtil.class).trackUserWatchLog(liveUserWatchLogKey);
                    }
                }
            }
        } catch (Exception e) {
            log.error("更新 LiveWatchLog logType 异常（连接时）：liveId={}, userId={}, error={}",
                    liveId, userId, e.getMessage(), e);
        }
    }

    /**
     * 实时更新用户看课状态（在心跳时调用）
     * 在直播期间实时更新用户的看课状态，而不是等到关闭 WebSocket 或清理无效会话时才更新
     * @param liveId 直播间ID
     * @param userId 用户ID
     * @param watchDuration 观看时长（秒）
     */
    public void updateWatchLogTypeInRealTime(Long liveId, Long userId, Long watchDuration) {
        try {
            // 获取当前直播/回放状态
            Map<String, Integer> flagMap = liveWatchUserService.getLiveFlagWithCache(liveId);
            Integer currentLiveFlag = flagMap.get("liveFlag");

            // 只在直播状态（liveFlag = 1）时更新
            if (currentLiveFlag == null || currentLiveFlag != 1) {
                return;
            }

            // 获取用户的 companyId 和 companyUserId（使用带缓存的查询方法）
            LiveUserFirstEntry liveUserFirstEntry = liveUserFirstEntryService.selectEntityByLiveIdUserIdWithCache(liveId, userId);
            if (liveUserFirstEntry == null) {
                return;
            }

            Long companyId = liveUserFirstEntry.getCompanyId();
            Long companyUserId = liveUserFirstEntry.getCompanyUserId();

            // 如果 companyId 和 companyUserId 有效，则更新看课状态
            if (companyId != null && companyId > 0 && companyUserId != null && companyUserId > 0) {
                // 检查是否达到关键观看时长节点，在这些节点实时更新
                // 关键节点：3分钟（180秒）、20分钟（1200秒）、30分钟（1800秒）
                boolean isKeyDuration = (watchDuration == 180 || watchDuration == 1200 || watchDuration == 1800) ||
                                       (watchDuration > 180 && watchDuration % 60 == 0); // 每分钟更新一次

                // 使用 Redis 缓存控制更新频率，避免频繁更新数据库
                // 策略：在关键节点立即更新，其他时候每60秒更新一次
                String updateLockKey = "live:watch:log:update:lock:" + liveId + ":" + userId;
                String lastUpdateKey = "live:watch:log:last:duration:" + liveId + ":" + userId;

                // 获取上次更新的时长
                Long lastUpdateDuration = redisCache.getCacheObject(lastUpdateKey);

                // 如果达到关键节点，或者距离上次更新已超过60秒，则更新
                boolean shouldUpdate = false;
                if (isKeyDuration) {
                    // 关键节点立即更新
                    shouldUpdate = true;
                } else if (lastUpdateDuration == null || (watchDuration - lastUpdateDuration) >= 60) {
                    // 每60秒更新一次
                    shouldUpdate = true;
                }

                if (shouldUpdate) {
                    // 使用分布式锁，避免并发更新（锁超时时间10秒）
                    Boolean canUpdate = redisCache.setIfAbsent(updateLockKey, "1", 10, TimeUnit.SECONDS);

                    if (Boolean.TRUE.equals(canUpdate)) {
                        // 异步更新，避免阻塞心跳处理
                        CompletableFuture.runAsync(() -> {
                            try {
                                updateLiveWatchLogTypeByDuration(liveId, userId, companyId, companyUserId, watchDuration);
                                // 更新上次更新的时长
                                redisCache.setCacheObject(lastUpdateKey, watchDuration, 2, TimeUnit.HOURS);
                            } catch (Exception e) {
                                log.error("实时更新看课状态异常：liveId={}, userId={}, error={}",
                                        liveId, userId, e.getMessage(), e);
                            } finally {
                                // 释放锁
                                redisCache.deleteObject(updateLockKey);
                            }
                        });
                    }
                }
            }
        } catch (Exception e) {
            log.error("实时更新看课状态异常：liveId={}, userId={}, error={}",
                    liveId, userId, e.getMessage(), e);
        }
    }

    /**
     * 根据在线时长更新 LiveWatchLog 的 logType
     * @param liveId 直播间ID
     * @param userId 用户ID
     * @param companyId 公司ID
     * @param companyUserId 销售ID
     * @param onlineSeconds 在线时长（秒）
     */
    private void updateLiveWatchLogTypeByDuration(Long liveId, Long userId, Long companyId,
                                                   Long companyUserId, Long onlineSeconds) {
        try {
            // 获取直播视频总时长（videoType = 1 的视频，使用带缓存的查询方法）
            List<LiveVideo> videos = liveVideoService.listByLiveIdWithCache(liveId, 1);
            long totalVideoDuration = 0L;
            if (videos != null && !videos.isEmpty()) {
                totalVideoDuration = videos.stream()
                        .filter(v -> v.getDuration() != null)
                        .mapToLong(LiveVideo::getDuration)
                        .sum();
            }

            // 查询 LiveWatchLog
            LiveWatchLog queryLog = new LiveWatchLog();
            queryLog.setLiveId(liveId);
            queryLog.setCompanyId(companyId);
            queryLog.setCompanyUserId(companyUserId);

            List<LiveWatchLog> logs = liveWatchLogService.selectLiveWatchLogList(queryLog);
            if (logs == null || logs.isEmpty()) {
                return;
            }
            Date now = DateUtil.getDate();
            for (LiveWatchLog log : logs) {
                boolean needUpdate = false;
                Integer newLogType = log.getLogType();

                // ① 如果在线时长 <= 3分钟，修改 logType 为 4（看课中断）
                if (onlineSeconds <= 180) { // 3分钟 = 180秒
                    newLogType = 4;
                    needUpdate = true;
                }
                // ③ 如果直播视频 >= 40分钟，在线时长 >= 30分钟，logType 设置为 2（完课）
                else if (totalVideoDuration >= 2400 && onlineSeconds >= 1800) { // 40分钟 = 2400秒，30分钟 = 1800秒
                    newLogType = 2;
                    log.setFinishTime(now);
                    needUpdate = true;
                }
                // 如果直播视频 >= 20分钟且 < 40分钟，在线时长 >= 20分钟，logType 设置为 2（完课）
                else if (totalVideoDuration >= 1200 && totalVideoDuration < 2400 && onlineSeconds >= 1200) { // 20分钟 = 1200秒
                    newLogType = 2;
                    log.setFinishTime(now);
                    needUpdate = true;
                }

                // 如果 logType 已经是 2（完课），不再更新
                if (needUpdate && (log.getLogType() == null || log.getLogType() != 2)) {
                    log.setLogType(newLogType);
                    liveWatchLogService.updateLiveWatchLog(log);
                }
            }
        } catch (Exception e) {
            log.error("根据在线时长更新 LiveWatchLog logType 异常：liveId={}, userId={}, error={}",
                    liveId, userId, e.getMessage(), e);
        }
    }

    /**
     * 实时检查并推送完课积分
     * 在用户观看时长更新时立即检查是否达到完课条件，达到则立即推送
     * @param liveId 直播间ID
     * @param userId 用户ID
     * @param duration 当前观看时长（秒）
     */
    private void checkAndSendCompletionPointsInRealTime(long liveId, long userId, Long duration) {
        try {
            log.debug("[实时完课检查] liveId={}, userId={}, duration={}秒", liveId, userId, duration);

            // 1. 调用完课记录服务检查并创建完课记录
            completionPointsRecordService.checkAndCreateCompletionRecord(liveId, userId, duration);

            // 2. 查询是否有新的未领取完课记录
            List<LiveCompletionPointsRecord> unreceivedRecords =
                completionPointsRecordService.getUserUnreceivedRecords(liveId, userId);

            if (unreceivedRecords == null || unreceivedRecords.isEmpty()) {
                // 没有待领取的完课记录
                return;
            }

            // 3. 构建推送消息
            SendMsgVo sendMsgVo = new SendMsgVo();
            sendMsgVo.setLiveId(liveId);
            sendMsgVo.setUserId(userId);
            sendMsgVo.setCmd("completionPoints");
            sendMsgVo.setMsg("完成任务!");
            sendMsgVo.setData(JSONObject.toJSONString(unreceivedRecords.get(0)));

            // 4. 实时推送完课积分弹窗
            sendCompletionPointsMessage(liveId, userId, sendMsgVo);

            log.info("[实时完课推送] 发送完课积分弹窗通知, liveId={}, userId={}, points={}, duration={}秒",
                    liveId, userId, unreceivedRecords.get(0).getPointsAwarded(), duration);

        } catch (Exception e) {
            log.error("[实时完课推送] 实时检查完课积分失败, liveId={}, userId={}, duration={}",
                    liveId, userId, duration, e);
        }
    }

    /**
     * 向用户推送完课积分倒计时配置信息
     * 在用户连接WebSocket时调用，让前端能够显示倒计时
     * @param session WebSocket会话
     * @param liveId 直播间ID
     * @param userId 用户ID
     * @param live 直播信息
     */
    private void sendCompletionPointsConfigToUser(Session session, Long liveId, Long userId, Live live) {
        try {

            boolean isLiveStarted = false;
            if (live.getStatus() != null && live.getStatus() == 2) {
                isLiveStarted = true;
            } else if (live.getStartTime() != null) {
                LocalDateTime now = LocalDateTime.now();
                isLiveStarted = now.isAfter(live.getStartTime()) || now.isEqual(live.getStartTime());
            }

            if (!isLiveStarted) {
                // 直播未开始，不推送完课配置
                log.debug("[完课配置推送] 直播未开始，跳过推送, liveId={}, userId={}", liveId, userId);
                return;
            }

            String configJson = live.getConfigJson();
            if (configJson == null || configJson.isEmpty()) {
                return;
            }

            JSONObject jsonConfig = JSON.parseObject(configJson);
            boolean enabled = jsonConfig.getBooleanValue("enabled");
            if (!enabled) {
                return;
            }

            Integer completionRate = jsonConfig.getInteger("completionRate");
            if (completionRate == null || completionRate <= 0 || completionRate > 100) {
                return;
            }

            // 3. 计算完课所需观看时长
            Long videoDuration = live.getDuration();
            if (videoDuration == null || videoDuration <= 0) {
                return;
            }

            // 完课所需时长（秒） = 视频总时长 × 完课比例 / 100
            long requiredDuration = (long) Math.ceil(videoDuration * completionRate / 100.0);

            // 4. 获取用户当前观看时长
            String hashKey = "live:watch:duration:hash:" + liveId;
            String userIdField = String.valueOf(userId);
            Object existingDuration = redisCache.hashGet(hashKey, userIdField);
            long currentDuration = existingDuration != null ? Long.parseLong(existingDuration.toString()) : 0L;

            // 5. 检查今天是否已有完课记录
            LocalDate today = LocalDate.now();
            Date currentDate = Date.from(today.atStartOfDay(ZoneId.systemDefault()).toInstant());
            LiveCompletionPointsRecord todayRecord = completionPointsRecordService.selectByUserAndDate(liveId, userId, currentDate);

            boolean hasCompletedToday = (todayRecord != null);

            // 6. 构建配置信息
            JSONObject configData = new JSONObject();
            configData.put("videoDuration", videoDuration);  // 视频总时长（秒）
            configData.put("completionRate", completionRate);  // 完课比例（%）
            configData.put("requiredDuration", requiredDuration);  // 完课所需时长（秒）
            configData.put("currentDuration", currentDuration);  // 当前观看时长（秒）
            configData.put("remainingDuration", Math.max(0, requiredDuration - currentDuration));  // 剩余时长（秒）
            configData.put("hasCompletedToday", hasCompletedToday);  // 今天是否已完课

            // 7. 推送配置消息
            SendMsgVo sendMsgVo = new SendMsgVo();
            sendMsgVo.setLiveId(liveId);
            sendMsgVo.setUserId(userId);
            sendMsgVo.setCmd("completionPointsConfig");
            sendMsgVo.setMsg("完课积分配置");
            sendMsgVo.setData(configData.toJSONString());

            sendMessage(session, JSONObject.toJSONString(R.ok().put("data", sendMsgVo)));

            log.debug("[完课配置推送] 推送成功, liveId={}, userId={}, 所需时长={}秒, 当前时长={}秒, 剩余={}秒",
                    liveId, userId, requiredDuration, currentDuration, Math.max(0, requiredDuration - currentDuration));

        } catch (Exception e) {
            log.error("[完课配置推送] 推送失败, liveId={}, userId={}", liveId, userId, e);
        }
    }

    /**
     * 消息队列包装类，支持优先级（管理员消息优先级更高）
     */
    private static class QueueMessage implements Comparable<QueueMessage> {
        private final String message;
        private final long timestamp;
        private final int priority; // 0=普通消息, 1=管理员消息（优先级更高）
        private final long sequence; // 序列号，用于相同优先级消息的FIFO排序
        private final long sizeBytes; // 消息大小（字节数）

        private static final AtomicLong sequenceGenerator = new AtomicLong(0);

        public QueueMessage(String message, boolean isAdmin) {
            this.message = message;
            this.timestamp = System.currentTimeMillis();
            this.priority = isAdmin ? 1 : 0;
            this.sequence = sequenceGenerator.getAndIncrement();
            // 计算消息大小（UTF-8编码）
            this.sizeBytes = message != null ? message.getBytes(StandardCharsets.UTF_8).length : 0;
        }

        public String getMessage() {
            return message;
        }

        public long getSizeBytes() {
            return sizeBytes;
        }

        @Override
        public int compareTo(QueueMessage other) {
            // 优先级高的先处理（管理员消息）
            int priorityCompare = Integer.compare(other.priority, this.priority);
            if (priorityCompare != 0) {
                return priorityCompare;
            }
            // 相同优先级按序列号排序（FIFO）
            return Long.compare(this.sequence, other.sequence);
        }
    }

    /**
     * 获取或创建消息队列
     */
    private PriorityBlockingQueue<QueueMessage> getMessageQueue(Long liveId) {
        return messageQueues.computeIfAbsent(liveId, k -> new PriorityBlockingQueue<>());
    }

    /**
     * 启动消费者线程（如果还没有启动）
     */
    private void startConsumerThread(Long liveId) {
        consumerRunningFlags.computeIfAbsent(liveId, k -> new AtomicBoolean(false));
        AtomicBoolean runningFlag = consumerRunningFlags.get(liveId);

        // 如果线程已经在运行，直接返回
        if (runningFlag.get()) {
            return;
        }

        // 尝试启动消费者线程
        synchronized (consumerRunningFlags) {
            if (runningFlag.compareAndSet(false, true)) {
                Thread consumerThread = new Thread(() -> {
                    PriorityBlockingQueue<QueueMessage> queue = getMessageQueue(liveId);
                    log.info("[消息队列] 启动消费者线程, liveId={}", liveId);

                    while (runningFlag.get()) {
                        try {
                            // 检查是否还有session，如果没有则退出
                            ConcurrentHashMap<Long, Session> room = rooms.get(liveId);
                            List<Session> adminRoom = adminRooms.get(liveId);

                            boolean hasSession = (room != null && !room.isEmpty()) ||
                                                (adminRoom != null && !adminRoom.isEmpty());

                            if (!hasSession) {
                                log.info("[消息队列] 直播间无session，停止消费者线程, liveId={}", liveId);
                                break;
                            }

                            // 从队列中取消息，最多等待1秒
                            QueueMessage queueMessage = queue.poll(1, TimeUnit.SECONDS);
                            if (queueMessage != null) {
                                // 更新队列大小（减少）
                                AtomicLong currentSize = queueSizes.get(liveId);
                                if (currentSize != null) {
                                    currentSize.addAndGet(-queueMessage.getSizeBytes());
                                }
                                // 广播消息
                                broadcastMessageFromQueue(liveId, queueMessage.getMessage());
                            }
                        } catch (InterruptedException e) {
                            Thread.currentThread().interrupt();
                            log.info("[消息队列] 消费者线程被中断, liveId={}", liveId);
                            break;
                        } catch (Exception e) {
                            log.error("[消息队列] 消费消息异常, liveId={}", liveId, e);
                        }
                    }

                    // 清理资源
                    runningFlag.set(false);
                    consumerThreads.remove(liveId);
                    log.info("[消息队列] 消费者线程已停止, liveId={}", liveId);
                }, "MessageConsumer-" + liveId);

                consumerThread.setDaemon(true);
                consumerThread.start();
                consumerThreads.put(liveId, consumerThread);
            }
        }
    }

    /**
     * 停止消费者线程
     */
    private void stopConsumerThread(Long liveId) {
        AtomicBoolean runningFlag = consumerRunningFlags.get(liveId);
        if (runningFlag != null) {
            runningFlag.set(false);
        }
        Thread consumerThread = consumerThreads.remove(liveId);
        if (consumerThread != null) {
            consumerThread.interrupt();
        }
    }

    /**
     * 将消息加入队列
     * @param liveId 直播间ID
     * @param message 消息内容
     * @param isAdmin 是否是管理员消息（管理员消息会插队）
     * @return 是否成功加入队列
     */
    private boolean enqueueMessage(Long liveId, String message, boolean isAdmin) {
        PriorityBlockingQueue<QueueMessage> queue = getMessageQueue(liveId);
        AtomicLong currentSize = queueSizes.computeIfAbsent(liveId, k -> new AtomicLong(0));

        // 计算新消息的大小
        long messageSize = message != null ? message.getBytes(StandardCharsets.UTF_8).length : 0;

        // 检查队列条数限制
        if (!isAdmin && queue.size() >= MAX_QUEUE_SIZE) {
            log.warn("[消息队列] 队列条数已满，丢弃消息, liveId={}, queueSize={}", liveId, queue.size());
            return false;
        }

        // 检查队列大小限制（200MB）
        long newTotalSize = currentSize.get() + messageSize;
        if (newTotalSize > MAX_QUEUE_SIZE_BYTES) {
            if (!isAdmin) {
                // 普通消息超过大小限制，直接丢弃
                log.warn("[消息队列] 队列大小超过限制，丢弃普通消息, liveId={}, currentSize={}MB, messageSize={}KB",
                        liveId, currentSize.get() / (1024.0 * 1024.0), messageSize / 1024.0);
                return false;
            } else {
                // 管理员消息：需要移除一些普通消息以腾出空间
                long needToFree = newTotalSize - MAX_QUEUE_SIZE_BYTES;
                long freedSize = removeMessagesToFreeSpace(queue, currentSize, needToFree, true);
                if (freedSize < needToFree) {
                    log.warn("[消息队列] 无法释放足够空间，管理员消息可能无法入队, liveId={}, needToFree={}KB, freed={}KB",
                            liveId, needToFree / 1024.0, freedSize / 1024.0);
                    // 即使空间不足，也尝试入队（可能会超过限制，但管理员消息优先级高）
                }
            }
        }

        // 如果是管理员消息且队列条数已满，移除一个普通消息
        if (isAdmin && queue.size() >= MAX_QUEUE_SIZE) {
            // 由于是优先级队列，普通消息（priority=0）会在队列末尾
            // 尝试移除一个普通消息，为管理员消息腾出空间
            QueueMessage removed = null;
            Iterator<QueueMessage> iterator = queue.iterator();
            while (iterator.hasNext()) {
                QueueMessage msg = iterator.next();
                if (msg.priority == 0) {
                    removed = msg;
                    break;
                }
            }
            if (removed != null) {
                queue.remove(removed);
                currentSize.addAndGet(-removed.getSizeBytes());
                log.debug("[消息队列] 管理员消息插队，移除普通消息, liveId={}", liveId);
            } else {
                // 如果没有普通消息，移除队列末尾的消息（可能是最早的管理员消息）
                // 这种情况很少发生，因为管理员消息通常较少
                log.warn("[消息队列] 队列条数已满且无普通消息可移除, liveId={}", liveId);
            }
        }

        QueueMessage queueMessage = new QueueMessage(message, isAdmin);
        queue.offer(queueMessage);
        currentSize.addAndGet(messageSize);

        // 如果有session，确保消费者线程在运行
        ConcurrentHashMap<Long, Session> room = rooms.get(liveId);
        List<Session> adminRoom = adminRooms.get(liveId);
        boolean hasSession = (room != null && !room.isEmpty()) ||
                            (adminRoom != null && !adminRoom.isEmpty());

        if (hasSession) {
            startConsumerThread(liveId);
        }

        return true;
    }

    /**
     * 移除消息以释放空间
     * @param queue 消息队列
     * @param currentSize 当前队列大小（原子变量）
     * @param needToFree 需要释放的空间（字节数）
     * @param onlyRemoveNormal 是否只移除普通消息（true=只移除普通消息，false=可以移除任何消息）
     * @return 实际释放的空间（字节数）
     */
    private long removeMessagesToFreeSpace(PriorityBlockingQueue<QueueMessage> queue,
                                          AtomicLong currentSize,
                                          long needToFree,
                                          boolean onlyRemoveNormal) {
        long freedSize = 0;
        List<QueueMessage> toRemove = new ArrayList<>();

        // 收集需要移除的消息（优先移除普通消息）
        Iterator<QueueMessage> iterator = queue.iterator();
        while (iterator.hasNext() && freedSize < needToFree) {
            QueueMessage msg = iterator.next();
            if (!onlyRemoveNormal || msg.priority == 0) {
                toRemove.add(msg);
                freedSize += msg.getSizeBytes();
            }
        }

        // 如果只移除普通消息但空间还不够，可以移除管理员消息
        if (onlyRemoveNormal && freedSize < needToFree) {
            iterator = queue.iterator();
            while (iterator.hasNext() && freedSize < needToFree) {
                QueueMessage msg = iterator.next();
                if (msg.priority == 1 && !toRemove.contains(msg)) {
                    toRemove.add(msg);
                    freedSize += msg.getSizeBytes();
                }
            }
        }

        // 移除消息并更新大小
        for (QueueMessage msg : toRemove) {
            if (queue.remove(msg)) {
                currentSize.addAndGet(-msg.getSizeBytes());
            }
        }

        if (freedSize > 0) {
            log.info("[消息队列] 释放队列空间, removedCount={}, freedSize={}KB",
                    toRemove.size(), freedSize / 1024.0);
        }

        return freedSize;
    }

    /**
     * 从队列中消费消息并广播
     */
    private void broadcastMessageFromQueue(Long liveId, String message) {
        broadcastMessage(liveId, message);
    }

    /**
     * 检查并清理空的直播间资源
     */
    private void cleanupEmptyRoom(Long liveId) {
        ConcurrentHashMap<Long, Session> room = rooms.get(liveId);
        List<Session> adminRoom = adminRooms.get(liveId);

        boolean hasSession = (room != null && !room.isEmpty()) ||
                            (adminRoom != null && !adminRoom.isEmpty());

        if (!hasSession) {
            // 停止消费者线程
            stopConsumerThread(liveId);
            // 清理消息队列
            messageQueues.remove(liveId);
            consumerRunningFlags.remove(liveId);
            queueSizes.remove(liveId);
            log.info("[消息队列] 清理空直播间资源, liveId={}", liveId);
        }
    }

    /**
     * 处理自动化抽奖开始
     */
    private void handleAutoLotteryStart(LiveAutoTask task) {
        try {
            LiveLotteryConf conf = JSON.parseObject(task.getContent(), LiveLotteryConf.class);
            if (conf == null || conf.getLotteryId() == null) {
                log.error("[自动抽奖-开始] 配置解析失败");
                return;
            }

            // 调用Service层处理
            liveLotteryConfService.autoStartLottery(conf.getLotteryId());

            // 查询最新状态
            LiveLotteryConf latest = liveLotteryConfService.selectLiveLotteryConfByLotteryId(conf.getLotteryId());
            if (latest == null) {
                log.error("[自动抽奖-开始] 抽奖不存在，lotteryId={}", conf.getLotteryId());
                return;
            }

            // 更新缓存
            liveService.asyncToCacheLiveConfig(task.getLiveId());

            // 广播消息
            SendMsgVo msg = new SendMsgVo();
            msg.setLiveId(task.getLiveId());
            msg.setCmd("lottery");
            msg.setStatus(1);
            msg.setData(JSON.toJSONString(latest));

            enqueueMessage(task.getLiveId(), JSONObject.toJSONString(R.ok().put("data", msg)), true);

            log.info("[自动抽奖-开始] 推送消息成功，liveId={}, lotteryId={}", task.getLiveId(), conf.getLotteryId());

        } catch (Exception e) {
            log.error("[自动抽奖-开始] 处理失败，taskId={}", task.getId(), e);
        }
    }

    /**
     * 处理自动化抽奖结算
     */
    private void handleAutoLotterySettle(LiveAutoTask task) {
        try {
            LiveLotteryConf conf = JSON.parseObject(task.getContent(), LiveLotteryConf.class);
            if (conf == null || conf.getLotteryId() == null) {
                log.error("[自加抽奖-结算] 配置解析失败");
                return;
            }

            // 调用Service层处理并获取中奖用户
            List<Map<String, Object>> winners = liveLotteryConfService.autoSettleLottery(conf.getLotteryId());

            // 查询最新状态
            LiveLotteryConf latest = liveLotteryConfService.selectLiveLotteryConfByLotteryId(conf.getLotteryId());
            if (latest == null) {
                log.error("[自加抽奖-结算] 抽奖不存在，lotteryId={}", conf.getLotteryId());
                return;
            }

            // 更新缓存
            liveService.asyncToCacheLiveConfig(task.getLiveId());

            // 广播结算消息
            SendMsgVo msg = new SendMsgVo();
            msg.setLiveId(task.getLiveId());
            msg.setCmd("lottery");
            msg.setStatus(2);
            msg.setData(JSON.toJSONString(latest));

            enqueueMessage(task.getLiveId(), JSONObject.toJSONString(R.ok().put("data", msg)), true);

            // 广播中奖名单
            if (winners != null && !winners.isEmpty()) {
                SendMsgVo winnerMsg = new SendMsgVo();
                winnerMsg.setLiveId(task.getLiveId());
                winnerMsg.setCmd("LotteryDetail");
                winnerMsg.setData(JSON.toJSONString(winners));
                enqueueMessage(task.getLiveId(), JSONObject.toJSONString(R.ok().put("data", winnerMsg)), true);
            }

            log.info("[自加抽奖-结算] 推送消息成功，liveId={}, lotteryId={}, 中奖人数={}",
                    task.getLiveId(), conf.getLotteryId(), winners != null ? winners.size() : 0);

        } catch (Exception e) {
            log.error("[自加抽奖-结算] 处理失败，taskId={}", task.getId(), e);
        }
    }

    /**
     * 处理评论抽奖：检查用户评论是否匹配进行中的抽奖关键词，如果匹配则写入Redis
     */
    private void handleCommentLottery(Long liveId, Long userId, String comment) {
        try {
            // 查询进行中的评论抽奖（require=4）
            LiveLotteryConf queryConf = new LiveLotteryConf();
            queryConf.setLiveId(liveId);
            queryConf.setLotteryStatus("1");
            queryConf.setRequire(4L);
            List<LiveLotteryConf> lotteries = liveLotteryConfService.selectLiveLotteryConfList(queryConf);

            if (lotteries == null || lotteries.isEmpty()) {
                return;
            }

            Date now = new Date();
            for (LiveLotteryConf lottery : lotteries) {
                log.info("[评论抽奖] 检查抽奖，lotteryId={}, 配置关键词={}, 用户评论={}",
                        lottery.getLotteryId(), lottery.getRequireConf(), comment);

                if (lottery.getRequireConf() != null && lottery.getRequireConf().equals(comment)) {
                    String redisKey = String.format(LiveKeysConstant.LIVE_HOME_PAGE_CONFIG_DRAW, liveId, lottery.getLotteryId());

                    Object existing = redisCache.redisTemplate.opsForHash().get(redisKey, String.valueOf(userId));
                    if (existing != null) {
                        log.info("[评论抽奖] 用户已参与，liveId={}, lotteryId={}, userId={}",
                                liveId, lottery.getLotteryId(), userId);
                        continue;
                    }

                    LiveLotteryRegistration registration = new LiveLotteryRegistration();
                    registration.setLotteryId(lottery.getLotteryId());
                    registration.setLiveId(liveId);
                    registration.setUserId(userId);
                    registration.setIsWin(0L);
                    registration.setRizeLevel(-1L);
                    registration.setCreateTime(now);

                    redisCache.redisTemplate.opsForHash().put(
                        redisKey,
                        String.valueOf(userId),
                        JSON.toJSONString(registration)
                    );

                    redisCache.expire(redisKey, 30, TimeUnit.MINUTES);

                    log.info("[评论抽奖] 用户参与成功，liveId={}, lotteryId={}, userId={}, keyword={}",
                            liveId, lottery.getLotteryId(), userId, comment);
                } else {
                    log.info("[评论抽奖] 关键词不匹配，lotteryId={}, 配置={}, 用户评论={}",
                            lottery.getLotteryId(), lottery.getRequireConf(), comment);
                }
            }
        } catch (Exception e) {
            log.error("[评论抽奖] 处理失败，liveId={}, userId={}", liveId, userId, e);
        }
    }

    /**
     * 处理评论领积分：仅处理 red_type=4 的进行中活动，用户评论匹配 requireConf 配置的指定评论则可领取
     * requireConf 为指定评论内容，多个用逗号分隔，用户评论包含任一词即算匹配；为空则不限制评论内容
     */
    private void handleCommentRed(Long liveId, Long userId, String comment) {

        try {
            List<LiveRedConf> reds = liveRedConfService.selectActivedCommentRed(liveId);
            if (reds == null || reds.isEmpty()) {
                return;
            }
            log.info("[评论领积分] 收到评论，liveId={}, userId={}, comment={}", liveId, userId, comment);

            String commentTrim = comment != null ? comment.trim() : "";
            for (LiveRedConf red : reds) {
                if (!matchCommentRequire(red.getRequireConf(), commentTrim)) {
                    log.debug("[评论领积分] 评论未匹配关键词，redId={}, requireConf={}", red.getRedId(), red.getRequireConf());
                    continue;
                }

                String redisKey = String.format(LiveKeysConstant.LIVE_HOME_PAGE_CONFIG_RED, liveId, red.getRedId());
                String userIdStr = String.valueOf(userId);

                Object existing = redisCache.redisTemplate.opsForHash().get(redisKey, userIdStr);
                if (existing != null) {
                    log.info("[评论领积分] 用户已领取，liveId={}, redId={}, userId={}", liveId, red.getRedId(), userId);
                    continue;
                }

                RedPO redPO = new RedPO();
                redPO.setLiveId(liveId);
                redPO.setRedId(red.getRedId());
                redPO.setUserId(userId);

                R result = liveRedConfService.claimRedPacket(redPO);
                if (result.get("code").equals(200)) {
                    log.info("[评论领积分] 用户领取成功，liveId={}, redId={}, userId={}, msg={}",
                            liveId, red.getRedId(), userId, result.get("msg"));
                } else {
                    log.warn("[评论领积分] 用户领取失败，liveId={}, redId={}, userId={}, msg={}",
                            liveId, red.getRedId(), userId, result.get("msg"));
                }
            }
        } catch (Exception e) {
            log.error("[评论领积分] 处理失败，liveId={}, userId={}", liveId, userId, e);
        }
    }

    /**
     * 判断用户评论是否匹配 requireConf：为空或空串视为不限制（匹配）；
     * 否则按逗号分隔为多个关键词，评论包含任一词即匹配
     */
    private boolean matchCommentRequire(String requireConf, String comment) {
        if (requireConf == null || requireConf.trim().isEmpty()) {
            return true;
        }
        if (comment.isEmpty()) {
            return false;
        }
        String[] keywords = requireConf.split("[,，]");
        for (String kw : keywords) {
            if (kw != null && !kw.trim().isEmpty() && comment.contains(kw.trim())) {
                return true;
            }
        }
        return false;
    }

    /**
     * 处理自动化积分开始
     */
    private void handleAutoRedStart(LiveAutoTask task) {
        try {
            LiveRedConf conf = JSON.parseObject(task.getContent(), LiveRedConf.class);
            if (conf == null || conf.getRedId() == null) {
                log.error("[自动积分-开始] 配置解析失败");
                return;
            }

            // 调用Service层处理
            liveRedConfService.autoStartRed(conf.getRedId());

            // 查询最新状态
            LiveRedConf latest = liveRedConfService.selectLiveRedConfByRedId(conf.getRedId());
            if (latest == null) {
                log.error("[自动积分呢-开始] 积分不存在，redId={}", conf.getRedId());
                return;
            }

            // 更新缓存
            liveService.asyncToCacheLiveConfig(task.getLiveId());

            // 广播开始消息
            SendMsgVo msg = new SendMsgVo();
            msg.setLiveId(task.getLiveId());
            msg.setCmd("red");
            msg.setStatus(1);
            msg.setData(JSON.toJSONString(latest));

            enqueueMessage(task.getLiveId(), JSONObject.toJSONString(R.ok().put("data", msg)), true);

            log.info("[自动积分-开始] 推送消息成功，liveId={}, redId={}",
                    task.getLiveId(), conf.getRedId());

        } catch (Exception e) {
            log.error("[自动积分-开始] 处理失败，taskId={}", task.getId(), e);
        }
    }

    /**
     * 处理自动化红包结算
     */
    private void handleAutoRedSettle(LiveAutoTask task) {
        try {
            LiveRedConf conf = JSON.parseObject(task.getContent(), LiveRedConf.class);
            if (conf == null || conf.getRedId() == null) {
                log.error("[自动红包-结算] 配置解析失败");
                return;
            }

            // 调用Service层处理并获取领取用户
            List<Map<String, Object>> result = liveRedConfService.autoSettleRed(conf.getRedId());

            // 查询最新状态
            LiveRedConf latest = liveRedConfService.selectLiveRedConfByRedId(conf.getRedId());
            if (latest == null) {
                log.error("[自动红包-结算] 红包不存在，redId={}", conf.getRedId());
                return;
            }

            // 更新缓存
            liveService.asyncToCacheLiveConfig(task.getLiveId());

            // 广播结算消息
            SendMsgVo msg = new SendMsgVo();
            msg.setLiveId(task.getLiveId());
            msg.setCmd("red");
            msg.setStatus(2);
            msg.setData(JSON.toJSONString(latest));

            enqueueMessage(task.getLiveId(), JSONObject.toJSONString(R.ok().put("data", msg)), true);

            log.info("[自动红包-结算] 推送消息成功，liveId={}, redId={}, 领取人数={}",
                    task.getLiveId(), conf.getRedId(), result != null ? result.size() : 0);

        } catch (Exception e) {
            log.error("[自动红包-结算] 处理失败，taskId={}", task.getId(), e);
        }
    }

    private void pushLiveCommentBootstrap(Session session, long liveId, long userType) {
        try {
            ILiveCommentFeatureConfigService cfgSvc = SpringUtils.getBean(ILiveCommentFeatureConfigService.class);
            String cfgJson = cfgSvc.buildConfigPushJson();
            SendMsgVo cfgVo = SendMsgVo.builder().liveId(liveId).cmd("liveCommentConfig").data(cfgJson).on(true).build();
            sendMessage(session, JSONObject.toJSONString(R.ok().put("data", cfgVo)));
            if (userType == 0) {
                ILiveCommentPinService pinSvc = SpringUtils.getBean(ILiveCommentPinService.class);
                List<LiveCommentPinActive> pins = pinSvc.listActiveByLiveId(liveId);
                SendMsgVo pvo = SendMsgVo.builder().liveId(liveId).cmd("commentPinList").data(JSON.toJSONString(pins)).on(true).build();
                sendMessage(session, JSONObject.toJSONString(R.ok().put("data", pvo)));
            }
        } catch (IOException e) {
            log.warn("推送评论扩展配置 IO 异常 liveId={}", liveId, e);
        } catch (Exception ex) {
            log.warn("推送评论扩展配置失败 liveId={}", liveId, ex);
        }
    }

    /**
     * 总后台修改全局规则或样式后由 HTTP 触发：向所有在线连接广播（cmd=liveCommentConfig）
     */
    public void broadcastLiveCommentConfigToAll() {
        try {
            ILiveCommentFeatureConfigService cfgSvc = SpringUtils.getBean(ILiveCommentFeatureConfigService.class);
            String cfgJson = cfgSvc.buildConfigPushJson();
            SendMsgVo vo = SendMsgVo.builder().liveId(0L).cmd("liveCommentConfig").data(cfgJson).on(true).build();
            String message = JSONObject.toJSONString(R.ok().put("data", vo));
            broadcastToAllLiveConnections(message);
        } catch (Exception e) {
            log.error("全量广播评论配置失败", e);
        }
    }

    public void broadcastToAllLiveConnections(String message) {
        for (ConcurrentHashMap<Long, Session> room : rooms.values()) {
            for (Session s : room.values()) {
                if (s != null && s.isOpen()) {
                    sendWithRetry(s, message, 1);
                }
            }
        }
        for (CopyOnWriteArrayList<Session> adminRoom : adminRooms.values()) {
            for (Session s : adminRoom) {
                if (s != null && s.isOpen()) {
                    sendWithRetry(s, message, 1);
                }
            }
        }
    }

}

