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.live.config.ProductionWordFilter;
import com.fs.live.mapper.LiveCouponMapper;
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.exception.BaseException;
import com.fs.common.utils.StringUtils;
import com.fs.common.utils.spring.SpringUtils;
import com.fs.live.domain.*;
import com.fs.live.service.*;
import com.fs.live.vo.LiveGoodsVo;
import com.fs.store.domain.FsUser;
import com.fs.store.service.IFsUserService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.time.DateUtils;
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.util.*;
import java.util.concurrent.*;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

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

@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 = 3 * 60 * 1000;
    // admin房间消息发送线程池（单线程，保证串行化）
    private final static ConcurrentHashMap<Long, ExecutorService> adminExecutors = new ConcurrentHashMap<>();

    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 IFsUserService fsUserService = SpringUtils.getBean(IFsUserService.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 static Random random = new Random();

    // 直播间在线用户缓存
//    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");
        String location = (String) userProperties.get("location");
        Live live = liveService.selectLiveByLiveId(liveId);
        if (live == null) {
            throw new BaseException("未找到直播间");
        }
        long companyId = live.getCompanyId() == null ? -1L : live.getCompanyId();
        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");
        }


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

        // 记录连接信息 管理员不记录
        if (userType == 0) {
            FsUser fsUser = fsUserService.selectFsUserByUserId(userId);
            if (Objects.isNull(fsUser)) {
                throw new BaseException("用户信息错误");
            }

            LiveWatchUser liveWatchUserVO = liveWatchUserService.join(fsUser,liveId, userId, location);
            room.put(userId, session);
            // 直播间浏览量 +1
            redisCache.increment(PAGE_VIEWS_KEY + liveId, 1);

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

            // 记录在线人数
            redisCache.increment(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.increment(UNIQUE_VISITORS_KEY + liveId, 1);
            }

            // 判断是否是首次进入直播间的观众
            boolean isFirstViewer = redisCache.setIfAbsent(UNIQUE_VIEWERS_KEY + liveId + ":" + userId, 1, 1, TimeUnit.DAYS);
            if (isFirstViewer) {
                redisCache.increment(UNIQUE_VIEWERS_KEY + liveId, 1);
            }

            liveWatchUserVO.setMsgStatus(liveWatchUserVO.getMsgStatus());
            if (1 == random.nextInt(10)) {
                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());
                // 广播连接消息
                broadcastMessage(liveId, JSONObject.toJSONString(R.ok().put("data", sendMsgVo)));
            }

            LiveUserFirstEntry liveUserFirstEntry = liveUserFirstEntryService.selectEntityByLiveIdUserId(liveId, userId);
            if (liveUserFirstEntry != null) {
                // 处理第一次自己进入，第二次扫码销售进入
                if (liveUserFirstEntry.getCompanyUserId() == -1L && companyUserId != -1L) {
                    liveUserFirstEntry.setCompanyId(companyId);
                    liveUserFirstEntry.setCompanyUserId(companyUserId);
                    liveUserFirstEntryService.updateLiveUserFirstEntry(liveUserFirstEntry);
                }
            } else {
                // 这个用户A邀请用户b，b的业绩算a的销售的
                if (companyUserId == -2L) {
                    LiveUserFirstEntry clientB = liveUserFirstEntryService.selectEntityByLiveIdUserId(liveId, companyUserId);
                    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);
                liveUserFirstEntryService.insertLiveUserFirstEntry(liveUserFirstEntry);
            }


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

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

    //关闭连接时调用
    @OnClose
    public void onClose(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");

        ConcurrentHashMap<Long, Session> room = getRoom(liveId);
        List<Session> adminRoom = getAdminRoom(liveId);
        if (userType == 0) {
            FsUser fsUser = fsUserService.selectFsUserByUserId(userId);
            if (Objects.isNull(fsUser)) {
                throw new BaseException("用户信息错误");
            }

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

            if (room.isEmpty()) {
                rooms.remove(liveId);
            }


            // 直播间在线人数 -1
            redisCache.increment(ONLINE_USERS_KEY + liveId, -1);
            // 从在线用户Set中移除用户ID
            String onlineUsersSetKey = ONLINE_USERS_SET_KEY + liveId;
            redisCache.redisTemplate.opsForSet().remove(onlineUsersSetKey, String.valueOf(userId));
            // 广播离开消息 添加一个概率问题 摇塞子，1-4 当为1的时候广播消息
            if (1 == new Random().nextInt(10)) {
                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());
                broadcastMessage(liveId, JSONObject.toJSONString(R.ok().put("data", sendMsgVo)));
            }
        } 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());

    }

    //收到客户端信息
    @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");

        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());
                    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);
                    }

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

                    // 广播消息
                    broadcastMessage(liveId, JSONObject.toJSONString(R.ok().put("data", msg)));
                    break;
                case "sendGift":
                    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));
                    broadcastMessage(liveId, JSONObject.toJSONString(R.ok().put("data", msg)));
                    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));
                    // 广播消息
                    broadcastMessage(liveId, JSONObject.toJSONString(R.ok().put("data", msg)));
                    // 放在当前活动里面
                    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 "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 "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());
        broadcastMessage(liveId, JSONObject.toJSONString(R.ok().put("data", sendMsgVo)));
    }

    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");
        // ①  检查  缓存是否存在  ② 如果是发布 放入缓存 ③ 删除缓存
        if (status == 1) {
            Object cacheObject = redisCache.getCacheObject(String.format(LiveKeysConstant.LIVE_COUPON_NUM , couponIssueId));
            if (cacheObject == null) {
                LiveCouponIssue liveCoupon = liveCouponIssueService.selectLiveCouponIssueById(couponIssueId);
                if (liveCoupon != null) {
                    redisCache.setCacheObject(String.format(LiveKeysConstant.LIVE_COUPON_NUM , couponIssueId), liveCoupon.getRemainCount().intValue(), 30, TimeUnit.MINUTES);
                }
            }
        } else {
            redisCache.deleteObject(String.format(LiveKeysConstant.LIVE_COUPON_NUM , couponIssueId));
        }
        broadcastMessage(liveId, JSONObject.toJSONString(R.ok().put("data", msg)));
    }


    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));
        broadcastMessage(liveId, JSONObject.toJSONString(R.ok().put("data", msg)));
    }

    /**
     * 处理红包变动消息
     */
    private void processRed(Long liveId, SendMsgVo msg) {
        log.debug("redData: {}", 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));
            broadcastMessage(liveId, JSONObject.toJSONString(R.ok().put("data", msg)));
        }
    }

    /**
     * 处理抽奖变动消息
     */
    private void processLottery(Long liveId, SendMsgVo msg) {
        log.debug("lotteryData: {}", 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));
            broadcastMessage(liveId, JSONObject.toJSONString(R.ok().put("data", msg)));
        }
    }

    //错误时调用
    @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<>());
    }

    /**
     * 获取管理端房间
     * @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) {
        SendMsgVo sendMsgVo = new SendMsgVo();
        sendMsgVo.setLiveId(liveId);
        sendMsgVo.setUserId(userId);
        sendMsgVo.setUserType(0L);
        sendMsgVo.setCmd("Integral");
        sendMsgVo.setMsg("恭喜你成功获得观看奖励:" + scoreAmount + "芳华币");
        sendMsgVo.setData(String.valueOf(scoreAmount));
        ConcurrentHashMap<Long, Session> room = getRoom(liveId);
        Session session = room.get(userId);
        if(Objects.isNull( session)) return;
        session.getAsyncRemote().sendText(JSONObject.toJSONString(R.ok().put("data", sendMsgVo)));
    }

    private void sendBlockMessage(Long liveId, Long userId) {
        SendMsgVo sendMsgVo = new SendMsgVo();
        sendMsgVo.setLiveId(liveId);
        sendMsgVo.setUserId(userId);
        sendMsgVo.setUserType(0L);
        sendMsgVo.setCmd("blockUser");
        sendMsgVo.setMsg("账号已被停用");
        sendMsgVo.setData(null);
        ConcurrentHashMap<Long, Session> room = getRoom(liveId);
        Session session = room.get(userId);
        if(Objects.isNull( session)) return;
        session.getAsyncRemote().sendText(JSONObject.toJSONString(R.ok().put("data", sendMsgVo)));
    }

    /**
     * 广播消息
     * @param liveId   直播间ID
     * @param message  消息内容
     */
    public void broadcastMessage(Long liveId, String message) {
        ConcurrentHashMap<Long, Session> room = getRoom(liveId);
        List<Session> adminRoom = getAdminRoom(liveId);

        // 普通用户房间：并行发送
        room.forEach((k, v) -> {
            if (v.isOpen()) {
                sendWithRetry(v,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);
                    }
                });
            }
        }
    }

    /**
     * 广播点赞消息
     * @param liveId   直播间ID
     * @param message  消息内容
     */
    public void broadcastLikeMessage(Long liveId, String message) {
        ConcurrentHashMap<Long, Session> room = getRoom(liveId);
        room.forEach((k, v) -> {
            if (v.isOpen()) {
                sendWithRetry(v,message,1);
            }
        });
    }

    private void sendWithRetry(Session session, String message, int maxRetries) {
        int attempts = 0;
        while (attempts < maxRetries) {
            try {
                if(session.isOpen()) {
                    session.getAsyncRemote().sendText(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 {
                    throw e;
                }
            }
        }
        log.info("超过重试次数, 消息 {}",message);
    }

    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;
                }
                if (liveCoupon.getIssueId() == null) {
                    log.error("优惠券未发布");
                    return;
                }
                LiveCouponIssue liveCouponIssue = liveCouponIssueService.selectLiveCouponIssueById(liveCoupon.getIssueId());
                LiveCouponIssueRelation relation = liveCouponMapper.selectCouponRelation(task.getLiveId(), liveCouponIssue.getId());
                if (liveCoupon != null) {
                    redisCache.setCacheObject(String.format(LiveKeysConstant.LIVE_COUPON_NUM , liveCouponIssue.getId()), liveCouponIssue.getRemainCount().intValue(), 30, TimeUnit.MINUTES);
                }
                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 ;

            }
            msg.setStatus(1);
            broadcastMessage(task.getLiveId(), JSONObject.toJSONString(R.ok().put("data", msg)));
        } catch (Exception e) {
            log.error("定时任务执行异常：{}", e.getMessage());
        }

    }
    private void delAutoTask(long liveId, Long data) {
        String key = "live:auto_task:";
        redisCache.redisTemplate.opsForZSet().removeRangeByScore(key + liveId, data, data);
    }

    public void removeLikeCountCache(Long liveId) {
        lastLikeCountCache.remove(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 = 300)// 每0.3分钟执行一次
    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 = 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();

            // 检查普通用户会话
            List<Long> toRemove = new ArrayList<>();
            room.forEach((userId, session) -> {
                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, "心跳超时"));
                        }
                    } catch (Exception e) {
                        log.error("关闭超时会话失败: sessionId={}, liveId={}, userId={}",
                                session.getId(), liveId, userId, e);
                    }
                }
            });

            // 移除超时的会话
            toRemove.forEach(room::remove);
            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);
            }
        }
    }
}
