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.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.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.spring.SpringUtils;
import com.fs.live.domain.*;
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.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.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;

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<>();
    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 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");  // 获取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) {
            FsUserScrm fsUser = fsUserService.selectFsUserByUserId(userId);
            if (Objects.isNull(fsUser)) {
                throw new BaseException("用户信息错误");
            }

            LiveWatchUser liveWatchUserVO = liveWatchUserService.join(liveId, userId, location);
            room.put(userId, session);
            // 直播间浏览量 +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());
            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);
        }

        log.debug("加入webSocket liveId: {}, userId: {}, 直播间人数: {}, 管理端人数: {}", liveId, userId, room.size(), adminRoom.size());
    }

    //关闭连接时调用
    @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) {
            FsUserScrm fsUser = fsUserService.selectFsUserByUserId(userId);
            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(liveId, userId);
            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);
        }

        log.debug("离开webSocket liveId: {}, userId: {}, 直播间人数: {}, 管理端人数: {}", liveId, userId, room.size(), adminRoom.size());
    }

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

                        // 根据直播状态设置live_flag或replay_flag
                        Live msgLive = liveService.selectLiveByLiveId(msg.getLiveId());
                        if (msgLive != null && msgLive.getFinishTime() != null) {
                            Date finishTime = java.sql.Timestamp.valueOf(msgLive.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);
                    }

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

                    // 广播消息
                    broadcastMessage(liveId, JSONObject.toJSONString(R.ok().put("data", msg)));
                    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);
                    msg.setOn(true);
                    msg.setData(JSONObject.toJSONString(liveMsg));
                    msg.setCmd("sendMsg");
                    // 广播消息
                    broadcastMessage(liveId, JSONObject.toJSONString(R.ok().put("data", msg)));
                    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));
                    // 广播消息
                    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 "globalVisible":
                    msg.setOn(true);
                    liveWatchUserService.updateGlobalVisible(liveId, msg.getStatus());
                    liveService.updateGlobalVisible(liveId, msg.getStatus());
                    // 广播消息
                    broadcastMessage(liveId, JSONObject.toJSONString(R.ok().put("data", msg)));
                    break;
                case "singleVisible":
                    liveWatchUserService.updateSingleVisible(liveId, msg.getStatus(),msg.getUserId());
                    // 广播消息
                    broadcastMessage(liveId, JSONObject.toJSONString(R.ok().put("data", msg)));
                    break;
                case "sendGift":
                    break;
                case "blockUser":
                    sendBlockMessage(liveId, msg.getUserId());
                    break;
                case "goods":
                    sendGoodsMessage(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 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) {
        if (throwable instanceof EOFException) {
            log.info("WebSocket连接被客户端正常关闭（EOF），sessionId: {}", session.getId());
        } else {
            log.error("WebSocket连接错误", throwable);
        }
    }

    /**
     * 获取房间
     * @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 {
        session.getAsyncRemote().sendText(message);
    }

    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,7);
            }
        });
        adminRoom.forEach(v -> {
            if (v.isOpen()) {
                sendWithRetry(v,message,7);
            }
        });
    }

    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) {
                log.error("点赞数格式错误，liveId: {}, value: {}", liveId, cacheObject, 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));
    }

    /**
     * 广播点赞消息
     * @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,7);
            }
        });
    }

    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;
                }
                LiveCouponIssue liveCouponIssue = liveCouponIssueService.selectLiveCouponIssueByCouponId(liveCoupon.getCouponId());
                LiveCouponIssueRelation relation = liveCouponMapper.selectCouponRelation(task.getLiveId(), 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 ;
                // 更新直播间配置缓存
//                liveService.asyncToCacheLiveConfig(task.getLiveId());
                // 查询商品信息并广播
//                LiveGoodsVo liveGoodsVo = liveGoodsService.selectLiveGoodsVoByGoodsId(goodsId);
//                if (liveGoodsVo != null) {
//                    msg.setData(JSON.toJSONString(liveGoodsVo));
//                    msg.setStatus(status);
//                }
            }
            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);
    }
}
