package com.ym.mec.biz.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.fasterxml.jackson.annotation.JsonFormat;
import com.google.common.collect.Lists;
import com.ym.mec.auth.api.client.SysUserFeignService;
import com.ym.mec.auth.api.entity.SysUser;
import com.ym.mec.biz.dal.dao.ImLiveBroadcastRoomDao;
import com.ym.mec.biz.dal.dao.LiveGoodsMapperDao;
import com.ym.mec.biz.dal.dao.SysConfigDao;
import com.ym.mec.biz.dal.dto.ImLiveBroadcastRoomDto;
import com.ym.mec.biz.dal.dto.RoomReservationUserSearch;
import com.ym.mec.biz.dal.entity.ImLiveBroadcastRoom;
import com.ym.mec.biz.dal.entity.ImLiveBroadcastRoomData;
import com.ym.mec.biz.dal.entity.ImLiveBroadcastRoomMember;
import com.ym.mec.biz.dal.entity.ImLiveRoomBlack;
import com.ym.mec.biz.dal.enums.MessageTypeEnum;
import com.ym.mec.biz.dal.page.LiveRoomGoodsOrderQueryInfo;
import com.ym.mec.biz.dal.vo.*;
import com.ym.mec.biz.service.*;
import com.ym.mec.common.entity.ImRoomMessage;
import com.ym.mec.common.entity.ImUserState;
import com.ym.mec.common.exception.BizException;
import com.ym.mec.common.page.PageInfo;
import com.ym.mec.common.page.PageUtil;
import com.ym.mec.common.page.WrapperUtil;
import com.ym.mec.common.tenant.TenantContextHolder;
import com.ym.mec.im.ImFeignService;
import com.ym.mec.thirdparty.message.MessageSenderPluginContext;
import com.ym.mec.util.collection.MapUtil;
import com.ym.mec.util.date.DateUtil;
import com.ym.mec.util.excel.POIUtil;
import com.ym.mec.util.http.HttpUtil;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.poi.hssf.usermodel.HSSFWorkbook;
import org.redisson.api.RBucket;
import org.redisson.api.RLock;
import org.redisson.api.RMap;
import org.redisson.api.RedissonClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
import java.io.OutputStream;
import java.io.Serializable;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.function.BiFunction;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * 直播房间管理表(ImLiveBroadcastRoom)表服务实现类
 *
 * @author hgw
 * @since 2022-02-17 20:52:05
 */
@Service("imLiveBroadcastRoomService")
public class ImLiveBroadcastRoomServiceImpl extends ServiceImpl<ImLiveBroadcastRoomDao, ImLiveBroadcastRoom> implements ImLiveBroadcastRoomService {

    private final static Logger log = LoggerFactory.getLogger(ImLiveBroadcastRoomServiceImpl.class);

    @Autowired
    private SysUserFeignService sysUserFeignService;
    @Autowired
    private ImFeignService imFeignService;
    @Autowired
    private RedissonClient redissonClient;
    @Autowired
    private SysMessageService sysMessageService;
    @Autowired
    private SysConfigDao sysConfigDao;
    @Autowired
    private ImLiveBroadcastRoomMemberService liveBroadcastRoomMemberService;
    @Autowired
    private ImLiveBroadcastRoomDataService liveBroadcastRoomDataService;
    @Autowired
    private LiveGoodsMapperDao liveGoodsMapperDao;
    @Autowired
    private ImLiveRoomReservationService imLiveRoomReservationService;
    @Autowired
    private ImLiveRoomBlackService imLiveRoomBlackService;

    //待替换的变量
    public static final String USER_ID = "${userId}";
    public static final String ROOM_UID = "${roomUid}";

    //直播间累计用户信息-指只要进入到该房间的用户都要记录
    public static final String LIVE_ROOM_TOTAL_USER_LIST = String.join(":", "IM:LIVE_ROOM_TOTAL_USER_LIST", ROOM_UID);
    //直播间在线用户信息
    public static final String LIVE_ROOM_ONLINE_USER_LIST = String.join(":", "IM:LIVE_ROOM_ONLINE_USER_LIST", ROOM_UID);
    //当前房间主讲人心跳
    public static final String LIVE_ROOM_SPEAKER_HEART_BEAT = String.join(":", "IM:LIVE_ROOM_SPEAKER_HEART_BEAT", ROOM_UID);
    //房间点赞数
    public static final String LIVE_ROOM_LIKE = String.join(":", "IM:LIVE_ROOM_LIKE", ROOM_UID);
    //计算人员观看时长锁
    public static final String LIVE_LOOK_LOCK = String.join(":", "IM:LIVE_LOOK_LOCK", ROOM_UID);
    //用户对应的直播间Uid
    public static final String LIVE_USER_ROOM = String.join(":", "IM:LIVE_ROOM_USER", USER_ID);
    //记录人员最后变更的状态消息时间
    public static final String LIVE_USER_STATE_TIME = String.join(":", "IM:LIVE_USER_STATE_TIME", USER_ID);
    //主讲人信息
    public static final String LIVE_SPEAKER_INFO = String.join(":", "IM:LIVE_SPEAKER_INFO", ROOM_UID, USER_ID);
    //主讲人最近一次加入房间的clientIp
    public static final String LIVE_SPEAKER_LAST_CLIENT_IP = String.join(":", "IM:LIVE_SPEAKER_LAST_CLIENT_IP", ROOM_UID, USER_ID);
    //直播提前开始时间
    public static final int PRE_LIVE_TIME_MINUTE = 30;

    /**
     * 进入直播间检查数据
     *
     * @param roomUid 房间uid
     * @param userId  用户id
     * @param osType  默认学生端  0:pc端 1:学生端 2:老师端
     */
    @Override
    public ImLiveBroadcastRoomVo queryRoomAndCheck(String roomUid, Integer userId, Integer osType) {
        SysUser sysUser = Optional.ofNullable(userId)
                .map(this::getSysUser)
                .orElseGet(this::getSysUser);
        osType = Optional.ofNullable(osType).orElse(1);
        if (osType == 1) {
            //学生端
            return studentQueryRoomAndCheck(roomUid, sysUser);
        } else if (osType == 2) {
            //老师端
            return teacherQueryRoomAndCheck(roomUid, sysUser);
        }
        return queryRoomAndCheck(roomUid, sysUser);
    }

    public ImLiveBroadcastRoomVo studentQueryRoomAndCheck(String roomUid, SysUser sysUser) {
        //如果是学生端，则需要检查是否有权限进入
        Map<String, Object> param = new HashMap<>();
        param.put("roomUid", roomUid);
        ImLiveBroadcastRoomVo vo = Optional.of(param).map(this::getImLiveBroadcastRoomVo)
                .orElseThrow(() -> new BizException("您无法观看该直播"));
        if (vo.getSpeakerId().equals(sysUser.getId())) {
            throw new BizException("您是该直播间的主讲人，请从直播端进入");
        }
        if (vo.getLiveState() == 0) {
            throw new BizException("直播尚未开始");
        }
        return queryRoomAndCheck(vo, sysUser);
    }

    public ImLiveBroadcastRoomVo teacherQueryRoomAndCheck(String roomUid, SysUser sysUser) {
        ImLiveBroadcastRoomVo vo = Optional.ofNullable(roomUid).map(this::queryRoomInfo)
                .orElseThrow(() -> new BizException("直播间不存在"));
        if (vo.getOs().equals("pc")) {
            throw new BizException("该直播间仅可在电脑端直播");
        }
        return queryRoomAndCheck(vo, sysUser);
    }

    public ImLiveBroadcastRoomVo queryRoomAndCheck(String roomUid, SysUser sysUser) {
        ImLiveBroadcastRoomVo vo = Optional.ofNullable(roomUid).map(this::queryRoomInfo)
                .orElseThrow(() -> new BizException("直播间不存在"));
        return queryRoomAndCheck(vo, sysUser);
    }

    public ImLiveBroadcastRoomVo queryRoomAndCheck(ImLiveBroadcastRoomVo room, SysUser sysUser) {
        //直播间信息校验
        if (!room.getTenantId().equals(sysUser.getTenantId())) {
            throw new BizException("您不是该直播机构人员,不可观看!");
        }
        if (room.getRoomState() == 1) {
            throw new BizException("直播间不存在");
        }
        if (room.getLiveState() == 0) {
            Date liveStartTime = DateUtil.addMinutes(room.getLiveStartTime(), -PRE_LIVE_TIME_MINUTE);
            throw new BizException(DateUtil.format(liveStartTime, "yyyy年MM月dd日 HH点mm分") + " 可进入直播间准备");
        }
        if (room.getLiveState() == 2) {
            //如果直播结束了还是推广状态则将推广修改为取消
            if (room.getPopularize() == 1) {
                this.update(Wrappers.<ImLiveBroadcastRoom>lambdaUpdate()
                        .set(ImLiveBroadcastRoom::getPopularize, 0)
                        .eq(ImLiveBroadcastRoom::getId, room.getId()));
            }
            throw new BizException("直播已结束！");
        }
        room.setBlacklistFlag(0);
        //黑名单查询-查询当前用户是否在黑名单中
        int count = imLiveRoomBlackService.count(Wrappers.<ImLiveRoomBlack>lambdaQuery()
                .eq(ImLiveRoomBlack::getRoomUid, room.getRoomUid())
                .eq(ImLiveRoomBlack::getUserId, sysUser.getId()));
        if (count > 0) {
            room.setBlacklistFlag(1);
        }
        return room;
    }

    /**
     * 查询直播间信息
     *
     * @param roomUid 直播间uid
     */
    @Override
    public ImLiveBroadcastRoomVo queryRoomInfo(String roomUid) {
        ImLiveBroadcastRoomVo roomVo = getImLiveBroadcastRoomVo(roomUid);
        if (roomVo == null) {
            return null;
        }
        getRoomData(roomVo);
        return roomVo;
    }

    private ImLiveBroadcastRoomVo getImLiveBroadcastRoomVo(String roomUid) {
        List<ImLiveBroadcastRoomVo> list = baseMapper.queryPage(new HashMap<String, Object>() {{
            put("roomUid", roomUid);
            put("allRoom", 1);
        }});
        if (CollectionUtils.isEmpty(list)) {
            return null;
        }
        return list.get(0);
    }

    /**
     * 分页查询直播间列表
     */
    @Override
    public PageInfo<ImLiveBroadcastRoomVo> queryPage(Map<String, Object> param) {
        String os = WrapperUtil.toStr(param, "os");
        if (StringUtils.isNotBlank(os) && os.equals("teacher")) {
            param.put("speakerId", getSysUser().getId());
        }
        Page<ImLiveBroadcastRoomVo> pageInfo = PageUtil.concatTimePage(param, "startTime", "endTime");
        pageInfo.setDesc("a.created_time_");
        param.put("tenantId", TenantContextHolder.getTenantId());
        IPage<ImLiveBroadcastRoomVo> page = baseMapper.queryPage(pageInfo, param);

        // 查询预约人数
        List<ImLiveBroadcastRoomVo> records = page.getRecords();
        if (CollectionUtils.isEmpty(records)) {
            return PageUtil.pageInfo(page);
        }

        List<String> roomIdList = records.stream().map(ImLiveBroadcastRoomVo::getRoomUid).collect(Collectors.toList());

        List<ImLiveBroadcastRoomVo> countRoomReservationList = imLiveRoomReservationService.countRoomReservation(roomIdList);
        Map<String, List<ImLiveBroadcastRoomVo>> collect = countRoomReservationList.stream()
                .collect(Collectors.groupingBy(ImLiveBroadcastRoomVo::getRoomUid));
        for (ImLiveBroadcastRoomVo record : records) {
            List<ImLiveBroadcastRoomVo> imLiveBroadcastRoomVos = collect.get(record.getRoomUid());
            if (CollectionUtils.isEmpty(imLiveBroadcastRoomVos)) {
                continue;
            }
            record.setRoomReservationNum(imLiveBroadcastRoomVos.get(0).getRoomReservationNum());
        }

        return PageUtil.pageInfo(page);
    }

    /**
     * 预创建直播间
     *
     * @param dto 直播间信息
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void add(ImLiveBroadcastRoomDto dto) {
        SysUser sysUser = getSysUser(dto.getSpeakerId());
        Date now = new Date();
        if (now.getTime() > dto.getLiveStartTime().getTime()) {
            throw new BizException("设置的直播开始时间不能小于当前时间");
        }
        ImLiveBroadcastRoom obj = new ImLiveBroadcastRoom();
        BeanUtils.copyProperties(dto, obj);
        String roomUid = "LIVE-" + sysUser.getId() + "-" + now.getTime();
        obj.setTenantId(TenantContextHolder.getTenantId());
        obj.setRoomUid(roomUid);
        obj.setRoomConfig(JSONObject.toJSONString(dto.getRoomConfig()));
        obj.setLiveState(0);
        obj.setRoomState(0);
        obj.setCreatedBy(getSysUser().getId());
        obj.setCreatedTime(now);
        this.save(obj);
        //推送老师端-直播已经创建
        sendRoomLiveState(sysUser, obj, MessageTypeEnum.JIGUANG_LIVE_CREATED);
    }

    /**
     * 修改直播间信息
     *
     * @param dto
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void update(ImLiveBroadcastRoomDto dto) {
        Date now = new Date();
        Integer userId = getSysUser().getId();
        ImLiveBroadcastRoom obj = this.getById(dto.getId());
        obj.setUpdatedBy(userId);
        obj.setUpdatedTime(now);
        //如果直播已经开始则只能修改是否全员禁麦 和聊天功能
        if (obj.getLiveState() == 1) {
            obj.setRoomConfig(JSONObject.toJSONString(dto.getRoomConfig()));
            this.updateById(obj);
            return;
        }
        if (now.getTime() > dto.getLiveStartTime().getTime()) {
            throw new BizException("设置的直播开始时间不能小于当前时间");
        }
        //本次修改后的类型
        BeanUtils.copyProperties(dto, obj);
        obj.setRoomConfig(JSONObject.toJSONString(dto.getRoomConfig()));
        log.info("update room  >>>  :{}", JSONObject.toJSONString(obj));
        this.updateById(obj);
    }

    /**
     * 是否禁言
     *
     * @param id          房间id
     * @param whetherChat 是否允许聊天互动  0允许 1不允许
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void whetherChat(Integer id, Integer whetherChat) {
        ImLiveBroadcastRoom obj = this.getById(id);
        ImLiveBroadcastRoomDto.RoomConfig roomConfig = getRoomConfig(obj.getRoomConfig()).orElse(null);
        if (Objects.isNull(roomConfig)) {
            return;
        }
        roomConfig.setWhether_chat(whetherChat);
        obj.setRoomConfig(JSONObject.toJSONString(roomConfig));
        this.updateById(obj);
    }

    /**
     * 删除直播间
     *
     * @param id 直播间id
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void delete(Integer id) {
        ImLiveBroadcastRoom obj = this.getById(id);
        if (obj.getLiveState() == 1) {
            throw new BizException("直播已经开始，无法删除");
        }
        obj.setId(id);
        obj.setRoomState(1);
        obj.setUpdatedBy(getSysUser().getId());
        obj.setUpdatedTime(new Date());
        obj.setPopularize(0);
        this.updateById(obj);
    }

    /**
     * 推广直播间-每个机构只能有一个直播间在首页推广
     *
     * @param id         直播间id
     * @param popularize 是否在首页推广 0否 1是
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void opsPopularize(Integer id, Integer popularize) {
        if (!WrapperUtil.checkInObj(popularize, 0, 1)) {
            throw new BizException("参数错误");
        }
        Optional<ImLiveBroadcastRoom> roomOptional = Optional.ofNullable(id)
                .map(this::getById);
        roomOptional.orElseThrow(() -> new BizException("直播间不存在"));
        roomOptional.filter(room -> room.getRoomState() == 0)
                .orElseThrow(() -> new BizException("直播间已经删除，无法设置推广"));
        roomOptional.filter(room -> room.getLiveState() != 2)
                .orElseThrow(() -> new BizException("直播已结束，无法设置推广"));
        ImLiveBroadcastRoom obj = roomOptional.get();
        if (Objects.equals(obj.getPopularize(), popularize)) {
            return;
        }
        //推广该直播间，先清除其该机构他直播间的推广状态
        if (popularize == 1) {
            this.update(Wrappers.<ImLiveBroadcastRoom>lambdaUpdate()
                    .set(ImLiveBroadcastRoom::getPopularize, 0)
                    .eq(ImLiveBroadcastRoom::getRoomState, 0)
                    .eq(ImLiveBroadcastRoom::getTenantId, obj.getTenantId()));
        }
        //更新直播间推广状态
        obj.setPopularize(popularize);
        this.updateById(obj);
    }

    /**
     * 查询该机构目前推广的直播间
     */
    @Override
    public ImLiveBroadcastRoomVo queryPopularizeRoom() {
        Map<String, Object> param = new HashMap<>();
        param.put("tenantId", TenantContextHolder.getTenantId());
        param.put("popularize", 1);
        param.put("liveState", 1);
        return getImLiveBroadcastRoomVo(param);
    }

    private ImLiveBroadcastRoomVo getImLiveBroadcastRoomVo(Map<String, Object> param) {
        PageInfo<ImLiveBroadcastRoomVo> pageInfo = imLiveRoomReservationService.queryPageStudent(param);
        List<ImLiveBroadcastRoomVo> list = pageInfo.getRows();
        if (CollectionUtils.isNotEmpty(list)) {
            getRoomData(list.get(0));
            return list.get(0);
        }
        return null;
    }

    /**
     * 定时任务，每分钟执行
     * 销毁主讲人退出超过30分钟的直播间
     */
    public void destroyExpiredLiveRoom() {
        //查询状态是 未开始和已开始 的直播间
        List<ImLiveBroadcastRoom> list = this.list(Wrappers.<ImLiveBroadcastRoom>lambdaQuery()
                .in(ImLiveBroadcastRoom::getLiveState, 0, 1)
                .eq(ImLiveBroadcastRoom::getRoomState, 0));
        if (CollectionUtils.isEmpty(list)) {
            return;
        }
        //查询房间过期时间
        String expiredMinuteStr = sysConfigDao.findConfigValue("destroy_expired_live_room_minute");
        if (StringUtils.isEmpty(expiredMinuteStr)) {
            log.info("roomDestroy>>>> 未查询到配置：destroy_expired_live_room_minute");
            return;
        }
        int expiredMinute = Integer.parseInt(expiredMinuteStr);
        Date now = new Date();
        list.forEach(room -> {
            try {
                destroyExpiredLiveRoom(now, room, expiredMinute);
            } catch (Exception e) {
                log.error("roomDestroy>>>> failed roomId：{} msg：{}", room.getId(), e.getMessage());
            }
        });
    }

    private void destroyExpiredLiveRoom(Date now, ImLiveBroadcastRoom room, int expiredMinute) {
        log.error("roomDestroy destroyExpiredLiveRoom >>>> now {} roomInfo : {} expiredMinute:{}", now, JSONObject.toJSONString(room), expiredMinute);
        //过期时间 = LiveStartTime + expiredMinute
        Date expiredTime = DateUtil.addMinutes(room.getLiveStartTime(), expiredMinute);
        //当前时间 小于 过期时间 则不销毁
        if (now.getTime() <= expiredTime.getTime()) {
            return;
        }
        //获取直播间主讲人信息
        RBucket<RoomSpeakerInfo> speakerCache = getRoomSpeakerInfoCache(room.getRoomUid(), room.getSpeakerId().toString());
        if (speakerCache.isExists()) {
            RoomSpeakerInfo speakerInfo = speakerCache.get();
            //查询用户是否在线
            if (imFeignService.checkOnline(speakerInfo.getSpeakerId().toString())) {
                log.info("roomDestroy destroyExpiredLiveRoom  is online >>>> roomId:{} speakerId:{}", room.getId(), speakerInfo.getSpeakerId());
                return;
            }
            //校验房间心跳是否过期没续租
            RBucket<Date> lastRoomHeartbeatCache = redissonClient.getBucket(LIVE_ROOM_SPEAKER_HEART_BEAT.replace(ROOM_UID, room.getRoomUid()));
            if (lastRoomHeartbeatCache.isExists()) {
                //获取最后一次房间心跳时间
                Date lastDate = lastRoomHeartbeatCache.get();
                //房间心跳过期时间 = 将房间心跳时间+过期分钟
                Date lastRoomDateExpired = DateUtil.addMinutes(lastDate, expiredMinute);
                //当前时间 小于 房间心跳过期时间 则不销毁
                if (now.getTime() <= lastRoomDateExpired.getTime()) {
                    log.info("roomDestroy destroyExpiredLiveRoom  last room heartbeat  >>>> roomId:{} speakerId:{}", room.getId(), speakerInfo.getSpeakerId());
                    return;
                }
            }
            //1.主播没有进入房间，则直接销毁房间
            if (Objects.isNull(speakerInfo.getJoinRoomTime())) {
                log.info("roomDestroy not joinRoom >>>> cache : {}", JSONObject.toJSONString(test(room.getRoomUid())));
                roomDestroy(room);
                return;
            }

            //2.已知主播已进入了房间 就判断是否退出过房间
            if (Objects.nonNull(speakerInfo.getExitRoomTime())) {
                //如果退出时间 大于 进入时间，则证明退出后再也没有进入房间
                if (speakerInfo.getExitRoomTime().getTime() > speakerInfo.getJoinRoomTime().getTime()) {
                    //最终过期时间 =  退出时间+expiredMinute过期时间
                    Date exitExpiredTime = DateUtil.addMinutes(speakerInfo.getExitRoomTime(), expiredMinute);
                    //现在时间 大于等于 最终过期时间 则证明退出后至少30分钟没进入过房间
                    if (now.getTime() >= exitExpiredTime.getTime()) {
                        log.info("roomDestroy exitExpiredTime >>>> cache : {}", JSONObject.toJSONString(test(room.getRoomUid())));
                        roomDestroy(room);
                    }
                }
            }
        }
    }

    /**
     * 关闭房间-融云
     * 获取所有直播间缓存数据并写入数据库后并清理缓存
     *
     * @param roomUid 直播间Uid
     */
    @Override
    public void roomDestroy(String roomUid) {
        ImLiveBroadcastRoom room = this.getOne(Wrappers.<ImLiveBroadcastRoom>lambdaQuery()
                .eq(ImLiveBroadcastRoom::getRoomUid, roomUid));
        opsRoomDestroy(room);
    }

    /**
     * 关闭房间-融云
     * 获取所有直播间缓存数据并写入数据库后并清理缓存
     *
     * @param id 直播房间表id
     */
    @Override
    public void roomDestroy(Integer id) {
        ImLiveBroadcastRoom room = this.getById(id);
        opsRoomDestroy(room);
    }

    private void opsRoomDestroy(ImLiveBroadcastRoom room) {
        if (Objects.isNull(room)) {
            return;
        }
        if (room.getLiveState() == 0) {
            throw new BizException("直播未开始");
        }
        if (room.getLiveState() == 2) {
            throw new BizException("直播已经结束,请刷新数据！");
        }
        //销毁房间
        roomDestroy(room);
    }

    public void roomDestroy(ImLiveBroadcastRoom room) {
        //10秒内同一个房间不能重复销毁-防重复销毁
        RBucket<Object> bucket = redissonClient.getBucket("IM:ROOMDESTROY:" + room.getRoomUid());
        if (!bucket.trySet(1, 10, TimeUnit.SECONDS)) {
            return;
        }
        log.error("roomDestroy>>>> room : {}", JSONObject.toJSONString(room));
        String roomUid = room.getRoomUid();
        Integer speakerId = room.getSpeakerId();

        //获取所有直播间缓存数据并写入数据库后并清理缓存
        CompletableFuture.runAsync(() -> insertAndCleanLiveData(roomUid, speakerId));
        log.info("roomDestroy>>>> insertAndCleanLiveData {}", JSONObject.toJSONString(room));

        //将房间状态改为已销毁
        Date date = new Date();
        room.setRoomState(2);
        room.setLiveState(2);
        Integer userId = -1;
        try {
            userId = getSysUser().getId();
        } catch (Exception ignored) {
        }
        room.setUpdatedBy(userId);
        room.setUpdatedTime(date);
        room.setLiveEndTime(date);
        room.setPopularize(0);//销毁直播间后要关闭推广
        this.updateById(room);

        //下架商品
        liveGoodsMapperDao.updateStatus(room.getRoomUid(), null, false);

        //向聊天室发自定义消息踢出所有人
        try {
            ImRoomMessage message = new ImRoomMessage();
            message.setIsIncludeSender(1);
            message.setFromUserId(speakerId.toString());
            message.setToChatroomId(roomUid);
            message.setObjectName(ImRoomMessage.FORCED_OFFLINE);
            imFeignService.publishRoomMsg(message);
            log.info("roomDestroy>>>> FORCED_OFFLINE {}", JSONObject.toJSONString(message));
            //销毁直播间
//            imFeignService.destroyLiveRoom(roomUid);
            log.info("roomDestroy>>>> destroyLiveRoom {}", JSONObject.toJSONString(message));
        } catch (Exception e) {
            log.error("roomDestroy>>>> errorMsg{}", e.getMessage(), e.getCause());
        }
    }

    //获取该直播间所有数据写入数据库-并清理缓存
    private void insertAndCleanLiveData(String roomUid, Integer speakerId) {
        log.info("insertAndCleanLiveData >>>> roomUid : {}", roomUid);
        Date now = new Date();
        //总观看人数
        List<ImLiveBroadcastRoomMember> memberList = new ArrayList<>();
        //获取直播间所有人数据写入 im_live_broadcast_room_member
        RMap<Integer, String> roomTotalUserCache = getTotalUserCache(roomUid);
        if (roomTotalUserCache.isExists()) {
            List<RoomUserInfoVo> roomTotalUser = roomTotalUserCache.values().parallelStream()
                    .map(a -> JSONObject.toJavaObject(JSONObject.parseObject(a), RoomUserInfoVo.class))
                    .collect(Collectors.toList());
            for (RoomUserInfoVo v : roomTotalUser) {
                ImLiveBroadcastRoomMember member = new ImLiveBroadcastRoomMember();
                member.setTenantId(v.getTenantId());
                member.setRoomUid(roomUid);
                member.setUserId(v.getUserId());
                member.setJoinTime(v.getFirstJoinTime());
                member.setTotalTime(getLookMinutes(v.getDynamicLookTime(), now, v.getTotalViewTime()));
                memberList.add(member);
            }
        }

        //获取直播点赞
        int like = 0;
        RBucket<Object> likeCache = redissonClient.getBucket(LIVE_ROOM_LIKE.replace(ROOM_UID, roomUid));
        if (likeCache.isExists()) {
            like = (int) likeCache.get();
            //删除房间点赞数据
            likeCache.delete();
        }
        int speakerLiveTime = 0;
        //获取直播间主讲人信息 写入im_live_broadcast_room_data
        RBucket<RoomSpeakerInfo> speakerCache = getRoomSpeakerInfoCache(roomUid, speakerId.toString());
        if (speakerCache.isExists()) {
            ImLiveBroadcastRoomData liveData = new ImLiveBroadcastRoomData();
            RoomSpeakerInfo speakerInfo = speakerCache.get();
            log.info("insertAndCleanLiveData >>>> speakerInfo : {}", JSONObject.toJSONString(speakerInfo));
            liveData.setTenantId(speakerInfo.getTenantId());
            liveData.setRoomUid(roomUid);
            liveData.setLikeNum(like);
            liveData.setTotalUserNum(CollectionUtils.isNotEmpty(memberList) ? memberList.size() : 0);
            liveData.setUpdatedTime(now);
            liveData.setLiveTime(getLookMinutes(speakerInfo.getStartLiveTime(), now, speakerInfo.getTotalLiveTime()));
            liveBroadcastRoomDataService.save(liveData);
            //删除房间主讲人数据
            speakerCache.delete();
            //获取主讲人直播时长
            speakerLiveTime = liveData.getLiveTime();
        }

        //写入im_live_broadcast_room_member表，校验用户观看时长，不能大于主讲人直播时长
        if (CollectionUtils.isNotEmpty(memberList)) {
            for (ImLiveBroadcastRoomMember member : memberList) {
                if (member.getTotalTime() > speakerLiveTime) {
                    member.setTotalTime(speakerLiveTime);
                }
            }
            //添加人员数据
            Lists.partition(memberList, 500)
                    .forEach(list -> liveBroadcastRoomMemberService.getDao().insertBatch(list));
        }
        //获取在线人员信息
        RMap<Integer, String> onlineUserCache = getOnlineUserCache(roomUid);
        //删除人员对应直播间编号信息
        onlineUserCache.forEach((id, s) -> redissonClient.getBucket(LIVE_USER_ROOM.replace(USER_ID, id.toString())).delete());
        //删除直播间所有用户数据
        roomTotalUserCache.delete();
        //删除在线用户数据
        onlineUserCache.delete();
    }

    /**
     * 同步点赞数量
     *
     * @param roomUid 房间uid
     * @param likeNum 点赞数
     */
    @Override
    public void syncLike(String roomUid, Integer likeNum) {
        redissonClient.getBucket(LIVE_ROOM_LIKE.replace(ROOM_UID, roomUid)).set(likeNum);
        //增加房间心跳
        redissonClient.getBucket(LIVE_ROOM_SPEAKER_HEART_BEAT.replace(ROOM_UID, roomUid)).set(new Date());
    }

    /**
     * <p>主讲人处理进入和退出房间数据
     * <p>观看者只处理退出房间数据
     *
     * @param userState 用户状态数据
     */
    @Override
    public void opsRoom(List<ImUserState> userState) {
        if (CollectionUtils.isEmpty(userState)) {
            return;
        }
        userState.forEach(user -> {
            log.info("opsRoom>>>> {}", JSONObject.toJSONString(user));
            if (StringUtils.isBlank(user.getStatus())) {
                return;
            }
            Date now = new Date();
            //获取当前用户状态变更的时间
            long userStateTime = Optional.ofNullable(user.getTime()).orElse(now.getTime());
            String userid = user.getUserid();
            RBucket<Long> userStateTimeCache = redissonClient.getBucket(LIVE_USER_STATE_TIME.replace(USER_ID, userid));
            if (userStateTimeCache.isExists()) {
                //缓存的时间比当前传入时间大则放弃这条数据
                long cacheTime = userStateTimeCache.get();
                if (cacheTime > userStateTime) {
                    return;
                }
            }
            //将最新的时间写入缓存
            userStateTimeCache.set(userStateTime, 5L, TimeUnit.MINUTES);
            //获取当前用户所在房间的uid
            RBucket<String> userRoom = redissonClient.getBucket(LIVE_USER_ROOM.replace(USER_ID, userid));
            if (!userRoom.isExists()) {
                return;
            }
            String roomUid = userRoom.get();
            //查询userId是不是主讲人 ,如果是主讲人则返回
            if (isSpeaker(user, now, userid, roomUid)) {
                return;
            }
            //这里开始只处理观看者的数据，观看者只接受退出消息 status=0 是进入房间
            if (user.getStatus().equals("0")) {
                return;
            }
            Integer userId = Integer.valueOf(userid);

            //从房间累计用户信息中查询该用户的信息
            RMap<Integer, String> roomTotalUser = this.getTotalUserCache(roomUid);
            //该房间未查询到用户数据则不处理
            if (!roomTotalUser.isExists()) {
                return;
            }
            if (!roomTotalUser.containsKey(userId)) {
                return;
            }
            String userJsonStr = roomTotalUser.get(userId);
            log.info("opsRoom>>>> userJsonStr  {}", userJsonStr);
            //查询用户数据
            RoomUserInfoVo userInfo = JSONObject.toJavaObject(JSONObject.parseObject(userJsonStr), RoomUserInfoVo.class);
            //如果有动态观看时间则证明主播开播过，需要计算当前用户观看时长
            if (Objects.nonNull(userInfo.getDynamicLookTime())) {
                userInfo.setTotalViewTime(this.getLookMinutes(userInfo.getDynamicLookTime(), userInfo.getTotalViewTime()));
                userInfo.setDynamicLookTime(null);
            }
            roomTotalUser.fastPut(userId, JSONObject.toJSONString(userInfo));
            //查询在线人员列表
            RMap<Integer, String> onlineUserInfo = this.getOnlineUserCache(roomUid);
            if (!onlineUserInfo.isExists()) {
                return;
            }
            //从在线人员列表删除该人员
            onlineUserInfo.fastRemove(userId);
            //向直播间发送当前在线人数消息
            this.sendOnlineUserCount(roomUid, userId, onlineUserInfo.size());
            log.info("opsRoom>>>> looker userInfo: {}", JSONObject.toJSONString(userInfo));
            //用户离开直播间发送退出房间消息给主讲人
            ImRoomMessage message = new ImRoomMessage();
            message.setIsIncludeSender(1);
            message.setFromUserId(userId.toString());
            message.setToChatroomId(roomUid);
            message.setObjectName(ImRoomMessage.LOOKER_LOGIN_OUT);
            message.setContent(userId);
            try {
                imFeignService.publishRoomMsg(message);
                log.info("opsRoom>>>> looker LOOKER_LOGIN_OUT : {}", JSONObject.toJSONString(userInfo));
            } catch (Exception e) {
                log.error("opsRoom>>>>  looker error LOOKER_LOGIN_OUT {}", e.getMessage());
                log.error("opsRoom>>>>  looker error sendMessage {} : LOOKER_LOGIN_OUT : {}", message, JSONObject.toJSONString(userInfo));
            }
        });
    }

    /**
     * 向直播间发送当前在线人数消息
     *
     * @param roomUid    房间uid
     * @param fromUserId 发送人id
     * @param count      人数
     */
    private void sendOnlineUserCount(String roomUid, Integer fromUserId, Integer count) {
        //校验传入参数，房间uid和发送人id不能为空
        if (!WrapperUtil.checkObj(roomUid, fromUserId, count)) {
            log.info(" sendOnlineUserCount>>>> param is null   roomUid: {}  fromUserId:{}  count:{}", roomUid, fromUserId, count);
            return;
        }
        ImRoomMessage message = new ImRoomMessage();
        message.setIsIncludeSender(1);
        message.setObjectName(ImRoomMessage.MEMBER_COUNT);
        message.setToChatroomId(roomUid);
        HashMap<String, Integer> sendMap = new HashMap<>();
        sendMap.put("count", count);
        message.setFromUserId(fromUserId.toString());
        message.setContent(sendMap);
        //发送消息
        try {
            imFeignService.publishRoomMsg(message);
            log.info("sendOnlineUserCount>>>> message: {}", JSONObject.toJSONString(message));
        } catch (Exception e) {
            log.error("sendOnlineUserCount>>>> error {}", e.getMessage());
            log.error("sendOnlineUserCount>>>> sendMessage {} :", JSONObject.toJSONString(message));
        }
    }

    /**
     * 查询userId是不是主讲人
     *
     * @return true 是主讲人 false 不是主讲人
     */
    private boolean isSpeaker(ImUserState user, Date now, String userid, String roomUid) {
        RBucket<RoomSpeakerInfo> speakerCache = getRoomSpeakerInfoCache(roomUid, userid);
        if (!speakerCache.isExists()) {
            return false;
        }
        RoomSpeakerInfo speakerInfo = speakerCache.get();
        //最后一次进入房间的clientIp
        RBucket<String> lastClientIp = redissonClient.getBucket(LIVE_SPEAKER_LAST_CLIENT_IP.replace(ROOM_UID, roomUid).replace(USER_ID, userid));
        //主讲人进入房间
        if (user.getStatus().equals("0")) {
            speakerInfo.setJoinRoomTime(now);
            log.info("opsRoom>>>> join speakerCache {}", JSONObject.toJSONString(speakerInfo));
            speakerCache.set(speakerInfo);
            //将本次进入房间的clientIp添加到主讲人最后一次clientIp缓存中
            if (StringUtils.isNotBlank(user.getClientIp())) {
                lastClientIp.set(user.getClientIp());
            }
            return true;
        }
        //校验本次退出直播间的clientIp 是不是上次进入房间的clientIp
        if (StringUtils.isNotBlank(user.getClientIp())) {
            if (lastClientIp.isExists()) {
                //如果是上次进入房间的clientIp和本次退出房间的clientIp不相同，则直接忽略
                if (!user.getClientIp().equals(lastClientIp.get())) {
                    return true;
                }
            }
        }
        //如果退出时间大于进入时间就无需再次退出-直接返回
        if (compareDate.apply(speakerInfo.getExitRoomTime(), speakerInfo.getJoinRoomTime())) {
            return true;
        }
        //主讲人退出房间关闭录像
        closeLive(speakerInfo);
        speakerInfo.setExitRoomTime(now);
        log.info("opsRoom>>>> exit speakerCache {}", JSONObject.toJSONString(speakerInfo));
        speakerCache.set(speakerInfo);
        return true;
    }

    /**
     * 主讲人登录专用
     *
     * @param roomUid
     */
    public ImLiveBroadcastRoomVo speakerJoinRoom(String roomUid, String os) {
        SysUser sysUser = getSysUser();
        ImLiveBroadcastRoomVo roomVo;
        if (os.equals("mobile")) {
            roomVo = teacherQueryRoomAndCheck(roomUid, sysUser);
        } else {
            roomVo = queryRoomAndCheck(roomUid, sysUser);
        }
        Integer userId = sysUser.getId();
        if (!userId.equals(roomVo.getSpeakerId())) {
            throw new BizException("您不是该直播间的主讲人");
        }
        Date now = new Date();
        //获取直播间信息
        RBucket<RoomSpeakerInfo> speakerCache = getRoomSpeakerInfoCache(roomUid, userId.toString());
        if (!speakerCache.isExists()) {
            //没有主讲人信息则生成一个
            createSpeakerInfo(this.getById(roomVo.getId()), sysUser);
            speakerCache = getRoomSpeakerInfoCache(roomUid, userId.toString());
        }
        RoomSpeakerInfo speakerInfo = speakerCache.get();
        speakerInfo.setJoinRoomTime(now);
        speakerCache.set(speakerInfo);
        //记录用户当前房间uid
        redissonClient.getBucket(LIVE_USER_ROOM.replace(USER_ID, userId.toString())).set(roomUid, 12L, TimeUnit.HOURS);
        return roomVo;
    }

    /**
     * 进入房间-修改用户对应的房间号信息
     *
     * @param roomUid 房间uid
     * @param userId  用户id
     */
    public void joinRoom(String roomUid, Integer userId) {
        //查询房间信息
        ImLiveBroadcastRoomVo imLiveBroadcastRoomVo = getImLiveBroadcastRoomVo(roomUid);
        if (Objects.isNull(imLiveBroadcastRoomVo)) {
            log.info("opsRoom>>>> joinRoom error roomUid: {}", roomUid);
            return;
        }
        //记录用户当前房间uid
        redissonClient.getBucket(LIVE_USER_ROOM.replace(USER_ID, userId.toString())).set(roomUid, 12L, TimeUnit.HOURS);
        //房间累计用户信息-指只要进入到该房间的用户都要记录
        RMap<Integer, String> roomTotalUser = getTotalUserCache(roomUid);
        //判断是否第一次进房间
        RoomUserInfoVo userInfo;
        Date now = new Date();
        if (roomTotalUser.containsKey(userId)) {
            //多次进入更新动态进入时间
            userInfo = JSONObject.toJavaObject(JSONObject.parseObject(roomTotalUser.get(userId)), RoomUserInfoVo.class);
        } else {
            //第一次进该房间 写入用户首次进入时间
            userInfo = getUserInfo(userId);
            userInfo.setFirstJoinTime(now);
            userInfo.setTotalViewTime(0);
        }
        //查询主讲人信息
        RBucket<RoomSpeakerInfo> speakerCache = getRoomSpeakerInfoCache(roomUid, imLiveBroadcastRoomVo.getSpeakerId().toString());
        if (speakerCache.isExists()) {
            //如果用户进来时主讲人已经开启直播则修改学生观看时间
            Integer state = speakerCache.get().getState();
            if (Objects.nonNull(state) && state == 0) {
                userInfo.setDynamicLookTime(now);
            }
        }
        roomTotalUser.fastPut(userId, JSONObject.toJSONString(userInfo));
        //在线人员列表
        RMap<Integer, String> onlineUserInfo = getOnlineUserCache(roomUid);
        //进入房间写如在线人员列表
        onlineUserInfo.fastPut(userId, JSONObject.toJSONString(userInfo));
        //向直播间发送当前在线人数消息
        this.sendOnlineUserCount(roomUid, userId, onlineUserInfo.size());
        log.info("join sendOnlineUserCount>>>>   roomUid: {}  fromUserId:{}  count:{}", roomUid, userId, onlineUserInfo.size());
        log.info("joinRoom>>>> userInfo: {}", JSONObject.toJSONString(userInfo));
    }

    /**
     * 开始直播-录像
     *
     * @param roomUid 房间uid
     */
    public void startLive(String roomUid, Integer userId, String videoResolution) {
        //查询房间信息
        RBucket<RoomSpeakerInfo> speakerCache = getRoomSpeakerInfoCache(roomUid, userId.toString());
        if (!speakerCache.isExists()) {
            return;
        }
        RoomSpeakerInfo roomSpeakerInfo = speakerCache.get();
        //已是直播状态则直接返回
        if (intEquals(roomSpeakerInfo.getState(), 0)) {
            return;
        }
        //是否允许录像
        if (intEquals(roomSpeakerInfo.getWhetherVideo(), 0)) {
            //开始录制视频
            try {
                imFeignService.startRecord(roomUid, videoResolution);
            } catch (Exception e) {
                log.error("startRecord error: {}", e.getMessage());
            }
        }
        Date now = new Date();
        //开始直播
        roomSpeakerInfo.setState(0);
        roomSpeakerInfo.setStartLiveTime(now);
        speakerCache.set(roomSpeakerInfo);
        log.info("startLive>>>> roomSpeakerInfo: {}", JSONObject.toJSONString(roomSpeakerInfo));
        //主播开启直播，查询所有在直播间的用户并写入观看时间
        CompletableFuture.runAsync(() -> this.asyncOpsLiveLookTime(roomUid, 1, now));
    }

    /**
     * 关闭直播-录像
     *
     * @param roomUid 房间uid
     * @param userId  老师id
     */
    public void closeLive(String roomUid, Integer userId) {
        //查询房间主播信息
        RBucket<RoomSpeakerInfo> speakerCache = getRoomSpeakerInfoCache(roomUid, userId.toString());
        if (!speakerCache.isExists()) {
            return;
        }
        RoomSpeakerInfo roomSpeakerInfo = speakerCache.get();
        //关闭直播
        closeLive(roomSpeakerInfo);
        speakerCache.set(roomSpeakerInfo);
    }

    /**
     * 关闭直播-录像
     *
     * @param roomSpeakerInfo 房间主播信息
     */
    private void closeLive(RoomSpeakerInfo roomSpeakerInfo) {
        //直播状态 true 直播中 false关闭直播
        boolean stateFlag = intEquals(roomSpeakerInfo.getState(), 0);
        //是否录像 true允许 false不允许
        boolean whetherVideoFlag = intEquals(roomSpeakerInfo.getWhetherVideo(), 0);
        //允许录像并在直播中
        if (whetherVideoFlag && stateFlag) {
            try {
                //停止录制视频
                imFeignService.stopRecord(roomSpeakerInfo.getRoomUid());
            } catch (Exception e) {
                log.error("stopRecord error: {}", e.getMessage());
            }
        }
        //直播状态 true 直播中
        if (stateFlag) {
            Date now = new Date();
            roomSpeakerInfo.setEndLiveTime(now);
            roomSpeakerInfo.setState(1);
            //如果开播时间和本次操作结束播放时间小于1分钟则不计算观看时间
            int liveMinutes = getLookMinutes(roomSpeakerInfo.getStartLiveTime(), null);
            if (liveMinutes > 1) {
                //写入本次直播时长
                roomSpeakerInfo.setTotalLiveTime(getLookMinutes(roomSpeakerInfo.getStartLiveTime(), roomSpeakerInfo.getTotalLiveTime()));
                //关闭直播后异步执行计算房间人员观看时长
                CompletableFuture.runAsync(() -> this.asyncOpsLiveLookTime(roomSpeakerInfo.getRoomUid(), 2, now));
            }
            //计算完后将开始直播时间设置为空，待下次开启后再计算
            roomSpeakerInfo.setStartLiveTime(null);
        }
        log.info("closeLive>>>> roomSpeakerInfo: {}", JSONObject.toJSONString(roomSpeakerInfo));
    }

    /**
     * 打开/关闭直播后-异步计算房间人员观看时长
     *
     * @param roomUid 房间uid
     * @param type    type 1:开始直播-开始录像     2:关闭直播关闭录像
     */
    private void asyncOpsLiveLookTime(String roomUid, Integer type, Date now) {
        //加锁-避免快速点击开启直播和关闭直后异步执行后直播数据错误
        boolean b = this.runIfLockCanGet(LIVE_LOOK_LOCK.replace(ROOM_UID, roomUid), () -> {
            //查询所有在直播间的用户并计算观看时长
            RMap<Integer, String> roomTotalUser = getTotalUserCache(roomUid);
            if (!roomTotalUser.isExists()) {
                return;
            }
            //查询在线人员列表
            RMap<Integer, String> onlineUserInfo = getOnlineUserCache(roomUid);
            boolean onlineUserInfoExists = onlineUserInfo.isExists();
            roomTotalUser.forEach((id, userInfoStr) -> {
                RoomUserInfoVo userInfo = JSONObject.toJavaObject(JSONObject.parseObject(userInfoStr), RoomUserInfoVo.class);
                //获取当前用户是否在房间状态 true在 false不在
                if (onlineUserInfoExists && onlineUserInfo.containsKey(id)) {
                    if (type.equals(1)) {
                        //开启直播后对当前在房间的用户写入观看时间
                        userInfo.setDynamicLookTime(now);
                    } else if (type.equals(2)) {
                        userInfo.setTotalViewTime(getLookMinutes(userInfo.getDynamicLookTime(), now, userInfo.getTotalViewTime()));
                        userInfo.setDynamicLookTime(null);
                    } else {
                        return;
                    }
                    roomTotalUser.fastPut(id, JSONObject.toJSONString(userInfo));
                }
            });
        }, 2, 1, TimeUnit.MINUTES);
        if (!b) {
            try {
                Thread.sleep(200L);
            } catch (InterruptedException e) {
                log.error("asyncOpsLiveLookTime error: {}", e.getMessage());
            }
            this.asyncOpsLiveLookTime(roomUid, type, now);
        }
    }

    /**
     * 定时任务，每分钟执行
     * 提前30分钟主动去融云注册并创建房间
     */
    public void createLiveRoom() {
        RBucket<Object> createLock = redissonClient.getBucket("IM:LIVE_ROOM_CREATE_LOCK");
        if (!createLock.trySet(1, 1, TimeUnit.MINUTES)) {
            return;
        }
        Date now = new Date();
        Date endTime = DateUtil.addMinutes(now, PRE_LIVE_TIME_MINUTE);
        List<ImLiveBroadcastRoom> list = this.list(new WrapperUtil<ImLiveBroadcastRoom>()
                .hasEq("live_state_", 0)
                .hasEq("room_state_", 0)
                .hasGe("live_start_time_", now)
                .hasLe("live_start_time_", endTime)
                .queryWrapper());
        if (CollectionUtils.isEmpty(list)) {
            return;
        }
        list.forEach(this::createLiveRoom);
        createLock.delete();
    }

    /**
     * 去融云创建房间
     */
    private void createLiveRoom(ImLiveBroadcastRoom room) {
        log.info("createLiveRoom>>>>>>roomUid:{}", room.getRoomUid());
        try {
            //生成主讲人信息到缓存
            SysUser sysUser = getSysUser(room.getSpeakerId());
            createSpeakerInfo(room, sysUser);
            //去融云创建房间
            imFeignService.createLiveRoom(room.getRoomUid(), room.getRoomTitle());
            //推送预约直播间消息
            imLiveRoomReservationService.push(room);
            //推送直播开始消息
            sendRoomLiveState(sysUser, room, MessageTypeEnum.JIGUANG_LIVE_STARTED);
        } catch (Exception e) {
            log.error(">>>>>>>>>> createLiveRoom error roomUid:{} msg:{}", room.getRoomUid(), e.getMessage());
        }
    }

    //生成主讲人信息
    private void createSpeakerInfo(ImLiveBroadcastRoom room, SysUser sysUser) {
        Date now = new Date();
        RoomSpeakerInfo speakerInfo = new RoomSpeakerInfo();
        speakerInfo.setSpeakerId(sysUser.getId());
        speakerInfo.setSpeakerName(sysUser.getRealName());
        speakerInfo.setCreateRoomTime(now);
        speakerInfo.setRoomUid(room.getRoomUid());
        speakerInfo.setTenantId(sysUser.getTenantId());
        speakerInfo.setTotalLiveTime(0);

        //获取是否允许录像
        boolean video = getRoomConfig(room.getRoomConfig())
                .filter(c -> Objects.nonNull(c.getWhether_video()))
                .map(c -> c.getWhether_video() == 0)
                .orElse(false);
        if (video) {
            //可以录制视频
            speakerInfo.setWhetherVideo(0);
        } else {
            //不可以录制视频
            speakerInfo.setWhetherVideo(1);
        }
        speakerInfo.setOs(room.getOs());
        //写入主讲人信息
        getRoomSpeakerInfoCache(room.getRoomUid(), room.getSpeakerId().toString()).set(speakerInfo);

        //生成0点赞
        redissonClient.getBucket(LIVE_ROOM_LIKE.replace(ROOM_UID, room.getRoomUid())).set(0);

        //修改房间状态为开始
        room.setLiveState(1);
        room.setUpdatedTime(now);
        room.setUpdatedBy(-2);
        baseMapper.updateById(room);
    }

    private Optional<ImLiveBroadcastRoomDto.RoomConfig> getRoomConfig(String roomConfig) {
        return Optional.ofNullable(roomConfig)
                .map(c -> JSON.parseObject(c, ImLiveBroadcastRoomDto.RoomConfig.class));
    }

    /**
     * 推送老师端-直播已经创建/开始的消息
     *
     * @param user 用户信息
     * @param room 房间信息
     * @param en   直播列表地址
     *             <P> /#/liveRoomList?t=0 未开始
     *             <P>/#/liveRoomList 直播中
     */
    private void sendRoomLiveState(SysUser user, ImLiveBroadcastRoom room, MessageTypeEnum en) {
        String teacherBaseUrl = sysConfigDao.findConfigValue(SysConfigService.TEACHER_BASE_URL);
        if (StringUtils.isBlank(teacherBaseUrl)) {
            log.error("sendRoomState error: teacherBaseUrl is null");
            return;
        }
        teacherBaseUrl = "13?" + teacherBaseUrl;
        Map<Integer, String> pushMap = new HashMap<>();
        pushMap.put(user.getId(), user.getId().toString());
        String memo;
        Object[] objs;
        try {
            String liveDateStr = DateUtil.format(room.getLiveStartTime(), "yyyy年MM月dd日 HH点mm分");
            //创建直播间
            if (en.equals(MessageTypeEnum.JIGUANG_LIVE_CREATED)) {
                //创建直播间-发送消息-跳到未开始页面
                memo = teacherBaseUrl + "/#/liveRoomList?t=0";
                SysUser createUser = getSysUser(room.getCreatedBy());
                objs = new Object[]{createUser.getUsername(), liveDateStr, room.getLiveRemark()};
            } else if (en.equals(MessageTypeEnum.JIGUANG_LIVE_STARTED)) {
                //直播开始-发送消息-跳到直播中页面
                memo = teacherBaseUrl + "/#/liveRoomList";
                objs = new Object[]{liveDateStr, room.getRoomTitle()};
            } else {
                log.error("sendRoomState error: MessageTypeEnum error " + en);
                return;
            }
            //发送消息
            sysMessageService.batchSendMessage(MessageSenderPluginContext.MessageSender.JIGUANG, en, pushMap,
                    null, 0, memo, "TEACHER", objs);
        } catch (Exception e) {
            log.error("sendRoomState error ", e.getCause());
        }
    }

    private void getRoomData(ImLiveBroadcastRoomVo roomVo) {
        //点赞数
        Object like = redissonClient.getBucket(LIVE_ROOM_LIKE.replace(ROOM_UID, roomVo.getRoomUid())).get();
        if (Objects.isNull(like)) {
            like = 0;
        }
        roomVo.setLikeNum((int) like);
        //累计总用户数量
        roomVo.setTotalLookNum(getNum.apply(this::getTotalUserCache, roomVo.getRoomUid()));
        //在房间观看用户数量
        roomVo.setLookNum(getNum.apply(this::getOnlineUserCache, roomVo.getRoomUid()));
    }

    /**
     * 获取房间缓存中的用户数量/观看人数
     *
     * <p> func：查询用户数量/观看人数的方法
     * <p> roomUid ：房间uid
     * <p> return ：用户数量/观看人数
     */
    private final BiFunction<Function<String, RMap<Integer, ?>>, String, Integer> getNum = (func, roomUid) -> Optional.of(roomUid)
            .map(func)
            .filter(RMap::isExists)
            .map(RMap::size)
            .orElse(0);

    private SysUser getSysUser(Integer userId) {
        return Optional.ofNullable(userId)
                .map(sysUserFeignService::queryUserById)
                .orElseThrow(() -> new BizException("用户不存在."));
    }

    private SysUser getSysUser() {
        //修改机构基础信息
        return Optional.ofNullable(sysUserFeignService.queryUserInfo())
                .orElseThrow(() -> new BizException("用户不存在."));
    }

    /**
     * 测试
     */
    public Map<String, Object> test(String roomUid) {
        return test(roomUid, null);
    }

    public Map<String, Object> test(String roomUid, String userFlag) {
        //test
        Map<String, Object> result = new HashMap<>();
        //校验房间心跳是否过期没续租
        RBucket<Date> lastRoomHeartbeatCache = redissonClient.getBucket(LIVE_ROOM_SPEAKER_HEART_BEAT.replace(ROOM_UID, roomUid));
        if (lastRoomHeartbeatCache.isExists()) {
            result.put("房间心跳", DateUtil.dateToString(lastRoomHeartbeatCache.get(), DateUtil.EXPANDED_DATE_TIME_FORMAT));
        } else {
            result.put("房间心跳", "房间心跳不存在");
        }
        String userId = "";
        try {
            String[] split = roomUid.split("-");
            userId = split[1];
        } catch (Exception ignored) {
        }
        //获取主讲人信息
        RBucket<RoomSpeakerInfo> speakerCache = getRoomSpeakerInfoCache(roomUid, userId);
        if (speakerCache.isExists()) {
            result.put("主讲人信息", speakerCache.get());
        } else {
            result.put("主讲人信息", "主讲人信息不存在");
        }
        //点赞数
        Object like = redissonClient.getBucket(LIVE_ROOM_LIKE.replace(ROOM_UID, roomUid)).get();
        if (Objects.isNull(like)) {
            like = 0;
        }
        result.put("点赞数", like);

        int totalLook = 0;
        int look = 0;
        //正在房间观看的用户数据
        List<BaseRoomUserVo> inRoomUserInfo = queryRoomOnlineUserInfo(roomUid);
        //累计总观看的用户数量
        List<RoomUserInfoVo> totalUserInfo = queryTotalRoomUserInfo(roomUid);
        if (CollectionUtils.isNotEmpty(totalUserInfo)) {
            if (CollectionUtils.isNotEmpty(inRoomUserInfo)) {
                look = inRoomUserInfo.size();
                Optional.ofNullable(userFlag)
                        .ifPresent(a -> result.put("正在观看的人员信息", inRoomUserInfo));
            } else {
                result.put("正在观看的人员信息", "没有正在观看的人员数据");
            }
            totalLook = totalUserInfo.size();
            Optional.ofNullable(userFlag)
                    .ifPresent(a -> result.put("总人员数据", totalUserInfo));
        } else {
            result.put("总人员数据", "没有人员数据");
        }
        result.put("总观看人数", totalLook);
        result.put("实时观看数", look);
        return result;
    }

    @Override
    public void shareGroup(String roomUid, String groupIds) {
        ImLiveBroadcastRoomVo imLiveBroadcastRoomVo = queryRoomInfo(roomUid);
        if (imLiveBroadcastRoomVo == null) {
            throw new BizException("直播间信息异常！");
        }
        SysUser sysUser = sysUserFeignService.queryUserInfo();
        String baseApiUrl = sysConfigDao.findConfigValue(SysConfigService.BASE_API_URL);
        sysMessageService.batchSendImGroupMessage(MessageTypeEnum.IM_SHARE_LIVE_URL, sysUser.getId().toString(), "liveRoom?" + roomUid, groupIds.split(","), null,
                imLiveBroadcastRoomVo.getTenantName(), imLiveBroadcastRoomVo.getRoomTitle(), imLiveBroadcastRoomVo.getSpeakerName(),
                DateUtil.format(imLiveBroadcastRoomVo.getLiveStartTime(), DateUtil.CHINESE_DATA_FORMAT_1),
                imLiveBroadcastRoomVo.getLiveRemark(), HttpUtil.getSortUrl(baseApiUrl + "/#/liveClassTransfer?roomUid=" + roomUid));
    }

    /**
     * 查询直播间在线的用户
     *
     * @param roomUid 直播间uid
     */
    @Override
    public List<BaseRoomUserVo> queryRoomLimitOnlineUserInfo(String roomUid) {
        RMap<Integer, String> onlineUserCache = getOnlineUserCache(roomUid);
        return onlineUserCache.values().parallelStream()
                .map(a -> JSONObject.toJavaObject(JSONObject.parseObject(a), BaseRoomUserVo.class))
                .limit(200)
                .collect(Collectors.toList());
    }

    /**
     * 查询直播间在线的用户
     *
     * @param roomUid 直播间uid
     */
    @Override
    public List<BaseRoomUserVo> queryRoomOnlineUserInfo(String roomUid) {
        RMap<Integer, String> onlineUserInfo = getOnlineUserCache(roomUid);
        return onlineUserInfo.values().parallelStream()
                .map(a -> JSONObject.toJavaObject(JSONObject.parseObject(a), BaseRoomUserVo.class))
                .collect(Collectors.toList());
    }

    /**
     * 根据传入的用户id查询用户信息
     *
     * @param userIdList 用户id集合
     */
    public List<BaseRoomUserVo> queryBaseUserInfo(List<Integer> userIdList) {
        return Optional.ofNullable(userIdList)
                .filter(CollectionUtils::isNotEmpty)
                .map(baseMapper::queryBaseUserInfo)
                .orElseThrow(() -> new BizException("用户id集合不能为空"));
    }

    @Override
    public PageInfo<RoomReservationUserVo> queryRoomUser(RoomReservationUserSearch query) {
        IPage<RoomReservationUserVo> page = PageUtil.getPage(query.getPage(), query.getRows());
        page = baseMapper.queryRoomUser(page, query);
        return PageUtil.pageInfo(page);
    }

    /**
     * 导出直播间预约用户信息
     *
     * @param query
     * @param response
     */
    public void exportReservationRoomUser(RoomReservationUserSearch query, HttpServletResponse response) throws IOException {
        //训练营详情-查询条件时间设置
        List<RoomReservationUserVo> list = baseMapper.queryRoomUser(query);
        OutputStream outputStream = response.getOutputStream();
        HSSFWorkbook workbook = null;
        try {
            String[] header = {"编号", "姓名", "手机号", "分部", "声部"};
            String[] body = {"userId", "username", "phone", "organName", "subjectName"};
            workbook = POIUtil.exportExcel(header, body, list);
            response.setContentType("application/octet-stream");
            response.setHeader("Content-Disposition", "attachment;filename=replacement-" + DateUtil.getDate(new Date()) + ".xls");
            response.flushBuffer();
            outputStream = response.getOutputStream();
            workbook.write(outputStream);
            outputStream.flush();
            workbook.close();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (outputStream != null) {
                try {
                    workbook.close();
                    outputStream.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    @Override
    public PageInfo<LiveRoomGoodsOrderVo> queryLiveRoomGoodsOrderList(LiveRoomGoodsOrderQueryInfo queryInfo) {
        PageInfo<LiveRoomGoodsOrderVo> pageInfo = new PageInfo<>(queryInfo.getPage(), queryInfo.getRows());
        Map<String, Object> params = new HashMap<String, Object>();
        MapUtil.populateMap(params, queryInfo);

        List<LiveRoomGoodsOrderVo> dataList = null;
        int count = baseMapper.countLiveRoomGoodsOrderList(params);
        if (count > 0) {
            pageInfo.setTotal(count);
            params.put("offset", pageInfo.getOffset());
            dataList = baseMapper.queryLiveRoomGoodsOrderList(params);
            pageInfo.setStatInfo(baseMapper.sumLiveRoomGoodsOrderList(params));
        }
        if (count == 0) {
            dataList = new ArrayList<>();
        }
        pageInfo.setRows(dataList);
        return pageInfo;
    }

    /**
     * 查询直播间所有用户信息
     *
     * @param roomUid 直播间uid
     */
    private List<RoomUserInfoVo> queryTotalRoomUserInfo(String roomUid) {
        RMap<Integer, String> roomTotalUser = getTotalUserCache(roomUid);
        return roomTotalUser.values().parallelStream()
                .map(a -> JSONObject.toJavaObject(JSONObject.parseObject(a), RoomUserInfoVo.class))
                .collect(Collectors.toList());
    }

    /**
     * 获取直播间在线用户信息缓存
     *
     * @param roomUid 直播间uid
     * @return RMap key:userId value:BaseRoomUserVo -> jsonToString
     */
    private RMap<Integer, String> getOnlineUserCache(String roomUid) {
        return redissonClient.getMap(LIVE_ROOM_ONLINE_USER_LIST.replace(ROOM_UID, roomUid));
    }

    /**
     * 获取直播间所有用户信息缓存
     *
     * @param roomUid 直播间uid
     * @return RMap key:userId value:RoomUserInfoVo -> jsonToString
     */
    private RMap<Integer, String> getTotalUserCache(String roomUid) {
        return redissonClient.getMap(LIVE_ROOM_TOTAL_USER_LIST.replace(ROOM_UID, roomUid));
    }

    /**
     * 获取房间主讲人消息
     *
     * @param roomUid 房间uid
     * @param userId  用户id
     */
    private RBucket<RoomSpeakerInfo> getRoomSpeakerInfoCache(String roomUid, String userId) {
        return redissonClient.getBucket(LIVE_SPEAKER_INFO.replace(USER_ID, userId).replace(ROOM_UID, roomUid));
    }

    private RoomUserInfoVo getUserInfo(Integer userId) {
        RoomUserInfoVo userInfo = new RoomUserInfoVo();
        userInfo.setUserId(userId);
        SysUser sysUser = sysUserFeignService.queryUserById(userId);
        String name = userId.toString();
        if (Objects.nonNull(sysUser)) {
            name = sysUser.getPhone();
            if (sysUser.getUserType().contains("STUDENT")) {
                name = sysUser.getUsername();
            } else {
                if (StringUtils.isNotBlank(sysUser.getRealName())) {
                    name = sysUser.getRealName();
                }
            }
        }
        userInfo.setUserName(name);
        userInfo.setTenantId(sysUser.getTenantId());
        return userInfo;
    }

    /**
     * 计算观看时长差-分钟数不满一分钟为0
     *
     * @param startDT    开始时间
     * @param nowMinutes 现在观看时长
     */
    private int getLookMinutes(Date startDT, Integer nowMinutes) {
        return getLookMinutes(startDT, new Date(), nowMinutes);
    }

    /**
     * 计算观看时长差-分钟数不满一分钟为0
     *
     * @param startDT    开始时间
     * @param endDT      结束时间
     * @param nowMinutes 现在观看时长
     */
    private int getLookMinutes(Date startDT, Date endDT, Integer nowMinutes) {
        if (Objects.isNull(nowMinutes)) {
            nowMinutes = 0;
        }
        if (Objects.isNull(startDT)) {
            return nowMinutes;
        }
        if (startDT.getTime() > endDT.getTime()) {
            return nowMinutes;
        }
        //课程结束时间-课程开始时间
        long durationTime = endDT.getTime() - startDT.getTime();
        //相差多少分钟
        int minutesBetween = new Long(durationTime / 1000 / 60).intValue();
        minutesBetween += nowMinutes;
        return Math.max(minutesBetween, 0);
    }

    /**
     * 判断Integer是否相等-null值不相等
     *
     * @param key1 第一个Integer
     * @param key2 第二个Integer
     * @return 相等 true 不相等 false
     */
    private boolean intEquals(Integer key1, Integer key2) {
        return Objects.nonNull(key1) && Objects.nonNull(key2) && Objects.equals(key1, key2);
    }

    /**
     * 分布式锁
     *
     * @param lockName lockKey
     * @param runnable 任务
     * @param waitTime 等待抢锁的时间，若该key已被占用则等待抢锁。
     * @param timeout  超时时间
     * @param unit     时间单位
     * @return true 锁成功  false 锁失败
     */
    private boolean runIfLockCanGet(final String lockName, Runnable runnable, final long waitTime, final long timeout, TimeUnit unit) {
        RLock lock = redissonClient.getLock(lockName);
        if (Objects.isNull(lock)) {
            log.info("runIfLockCanGet lock is null lockName : {}", lockName);
            return false;
        }
        try {
            if (lock.tryLock(waitTime, timeout, unit)) {
                if (Objects.nonNull(runnable)) {
                    runnable.run();
                }
                return true;
            } else {
                return false;
            }
        } catch (Exception e) {
            log.error("runIfLockCanGet error lockName : {}", lockName, e);
            throw new BizException("runIfLockCanGet error lockName :" + lockName, e);
        } finally {
            this.unlock(lock);
        }
    }

    /**
     * 解锁
     */
    private void unlock(RLock lock) {
        if (lock.getHoldCount() != 0) {
            lock.unlock();
        }
    }

    /**
     * 对比时间,2个时间一个为空返回false,否则对比时间大小
     * <p>- date1 时间1
     * <p>- date2 时间2
     */
    private final BiFunction<Date, Date, Boolean> compareDate = (date1, date2) -> {
        if (Objects.nonNull(date1) && Objects.nonNull(date2)) {
            return date1.getTime() > date2.getTime();
        } else {
            return false;
        }
    };

    /**
     * 主讲人信息
     */
    public static class RoomSpeakerInfo implements Serializable {
        //主讲人id
        private Integer speakerId;
        //主讲人名称
        private String speakerName;
        //直播状态 0 直播中 1关闭直播
        private Integer state;
        //房间uid
        private String roomUid;
        //房间创建时间
        private Date createRoomTime;
        //进入房间时间
        @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", timezone = "GMT+8")
        private Date joinRoomTime;
        //开始直播时间
        @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", timezone = "GMT+8")
        private Date startLiveTime;
        //开始直播时间
        @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", timezone = "GMT+8")
        private Date endLiveTime;
        //退出房间时间
        @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", timezone = "GMT+8")
        private Date exitRoomTime;
        //总直播时间
        private Integer totalLiveTime = 0;
        //是否录像 0允许 1不允许
        private Integer whetherVideo;
        //机构
        private Integer tenantId;
        //播出端-  pc网页端 移动端mobile
        private String os = "pc";

        public Integer getSpeakerId() {
            return speakerId;
        }

        public void setSpeakerId(Integer speakerId) {
            this.speakerId = speakerId;
        }

        public String getSpeakerName() {
            return speakerName;
        }

        public void setSpeakerName(String speakerName) {
            this.speakerName = speakerName;
        }

        /**
         * 直播状态 0 直播中 1关闭直播
         */
        public Integer getState() {
            return state;
        }

        /**
         * 直播状态 0 直播中 1关闭直播
         */
        public void setState(Integer state) {
            this.state = state;
        }

        public String getRoomUid() {
            return roomUid;
        }

        public void setRoomUid(String roomUid) {
            this.roomUid = roomUid;
        }

        public Date getJoinRoomTime() {
            return joinRoomTime;
        }

        public void setJoinRoomTime(Date joinRoomTime) {
            this.joinRoomTime = joinRoomTime;
        }

        public Date getStartLiveTime() {
            return startLiveTime;
        }

        public void setStartLiveTime(Date startLiveTime) {
            this.startLiveTime = startLiveTime;
        }

        public Integer getTotalLiveTime() {
            return totalLiveTime;
        }

        public void setTotalLiveTime(Integer totalLiveTime) {
            this.totalLiveTime = totalLiveTime;
        }

        public Date getCreateRoomTime() {
            return createRoomTime;
        }

        public void setCreateRoomTime(Date createRoomTime) {
            this.createRoomTime = createRoomTime;
        }

        public Date getExitRoomTime() {
            return exitRoomTime;
        }

        public void setExitRoomTime(Date exitRoomTime) {
            this.exitRoomTime = exitRoomTime;
        }

        public Integer getWhetherVideo() {
            return whetherVideo;
        }

        public void setWhetherVideo(Integer whetherVideo) {
            this.whetherVideo = whetherVideo;
        }

        public Integer getTenantId() {
            return tenantId;
        }

        public void setTenantId(Integer tenantId) {
            this.tenantId = tenantId;
        }

        public Date getEndLiveTime() {
            return endLiveTime;
        }

        public void setEndLiveTime(Date endLiveTime) {
            this.endLiveTime = endLiveTime;
        }

        public String getOs() {
            return os;
        }

        public void setOs(String os) {
            this.os = os;
        }
    }

}

