package com.ruoyi.im.service.impl;

import cn.hutool.core.collection.CollStreamUtil;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.ruoyi.basic.service.YjAppUserService;
import com.ruoyi.common.core.domain.entity.AppUser;
import com.ruoyi.common.exception.GlobalException;
import com.ruoyi.common.im.constant.IMConstant;
import com.ruoyi.common.im.constant.RedisKey;
import com.ruoyi.common.im.enums.IMTerminalType;
import com.ruoyi.common.im.enums.MessageStatus;
import com.ruoyi.common.im.enums.MessageType;
import com.ruoyi.common.im.model.IMGroupMessage;
import com.ruoyi.common.im.model.IMUserInfo;
import com.ruoyi.common.im.util.BeanUtils;
import com.ruoyi.common.im.util.CommaTextUtils;
import com.ruoyi.common.im.util.ThreadPoolExecutorFactory;
import com.ruoyi.common.utils.JpushClientUtil;
import com.ruoyi.common.utils.SecurityUtils;
import com.ruoyi.im.domain.Group;
import com.ruoyi.im.domain.GroupMember;
import com.ruoyi.im.domain.GroupMessage;
import com.ruoyi.im.domain.PrivateMessage;
import com.ruoyi.im.domain.dto.GroupMessageDTO;
import com.ruoyi.im.domain.vo.GroupMessageVO;
import com.ruoyi.im.mapper.GroupMessageMapper;
import com.ruoyi.im.service.GroupMemberService;
import com.ruoyi.im.service.GroupMessageService;
import com.ruoyi.im.service.GroupService;
import com.ruoyi.im.util.SensitiveFilterUtil;
import com.ruoyi.imclient.IMClient;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.time.DateUtils;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.*;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

@Slf4j
@Service
@RequiredArgsConstructor
public class GroupMessageServiceImpl extends ServiceImpl<GroupMessageMapper, GroupMessage>
    implements GroupMessageService {
    private final GroupService groupService;
    private final YjAppUserService appUserService;
    private final GroupMemberService groupMemberService;
    private final RedisTemplate<String, Object> redisTemplate;
    private final IMClient imClient;
    private final SensitiveFilterUtil sensitiveFilterUtil;
    private static final ScheduledThreadPoolExecutor EXECUTOR = ThreadPoolExecutorFactory.getThreadPoolExecutor();

    @Override
    public GroupMessageVO sendMessage(GroupMessageDTO dto) {
        AppUser session = SecurityUtils.getAppLoginUser().getAppUser();
        Group group = groupService.getAndCheckById(dto.getGroupId());
        // 是否在群聊里面
        GroupMember member = groupMemberService.findByGroupAndUserId(dto.getGroupId(), session.getId());
        if (Objects.isNull(member) || member.getQuit()) {
            throw new GlobalException("您已不在群聊里面，无法发送消息");
        }
        // 群聊成员列表
        List<Long> userIds = groupMemberService.findUserIdsByGroupId(group.getId());
        if (dto.getReceipt() && userIds.size() > 10L) {
            // 大群的回执消息过于消耗资源，不允许发送
            throw new GlobalException(
                String.format("当前群聊大于%s人,不支持发送回执消息", 10L));
        }
        // 不用发给自己
        userIds = userIds.stream().filter(id -> !session.getId().equals(id)).collect(Collectors.toList());
        // 保存消息
        GroupMessage msg = BeanUtils.copyProperties(dto, GroupMessage.class);
        msg.setSendId(session.getId());
        msg.setSendTime(new Date());
        msg.setSendNickName(member.getShowNickName());
        msg.setAtUserIds(CommaTextUtils.asText(dto.getAtUserIds()));
        // 过滤内容中的敏感词
        if (MessageType.TEXT.code().equals(dto.getType())) {
            msg.setContent(sensitiveFilterUtil.filter(dto.getContent()));
        }
        this.save(msg);
        // 群发
        GroupMessageVO msgInfo = BeanUtils.copyProperties(msg, GroupMessageVO.class);
        msgInfo.setAtUserIds(dto.getAtUserIds());
        IMGroupMessage<GroupMessageVO> sendMessage = new IMGroupMessage<>();
        sendMessage.setSender(new IMUserInfo(session.getId(), 1));
        sendMessage.setRecvIds(userIds);
        sendMessage.setSendResult(false);
        sendMessage.setData(msgInfo);
        imClient.sendGroupMessage(sendMessage);
        jpush(group.getId(),group.getName(),msg.getContent(),userIds);
        log.info("发送群聊消息，发送id:{},群聊id:{},内容:{}", session.getId(), dto.getGroupId(), dto.getContent());
        return msgInfo;
    }
    public void jpush(Long groupId,String groupName,String contentd,List<Long> recvIds){
        List<AppUser> recvUsers= appUserService.listByIds(recvIds);
        List<String>  registerIds=recvUsers.stream().filter(user->StrUtil.isNotBlank(user.getRegisterId())).map(it -> it.getRegisterId()).collect(Collectors.toList());
        if (registerIds.size()>0){
            String msg = String.format(
                    "%s：%s",
                    groupName, contentd
            );
            Map<String, String> extras=new HashMap<>();
            extras.put("type","groupMessage");
            extras.put("groupId",groupId.toString());
            extras.put("pagePath","/pages/chatbox/chat-box");
            JpushClientUtil.pushByRegistrationId(registerIds,"瑜伽汇",msg,extras);
        }
    }
    @Transactional
    @Override
    public GroupMessageVO recallMessage(Long id) {
        AppUser session = SecurityUtils.getAppLoginUser().getAppUser();
        GroupMessage msg = this.getById(id);
        if (Objects.isNull(msg)) {
            throw new GlobalException("消息不存在");
        }
        if (!msg.getSendId().equals(session.getId())) {
            throw new GlobalException("这条消息不是由您发送,无法撤回");
        }
        if (System.currentTimeMillis() - msg.getSendTime().getTime() > IMConstant.ALLOW_RECALL_SECOND * 1000) {
            throw new GlobalException("消息已发送超过5分钟，无法撤回");
        }
        // 判断是否在群里
        GroupMember member = groupMemberService.findByGroupAndUserId(msg.getGroupId(), session.getId());
        if (Objects.isNull(member) || Boolean.TRUE.equals(member.getQuit())) {
            throw new GlobalException("您已不在群聊里面，无法撤回消息");
        }
        // 修改数据库
        msg.setStatus(MessageStatus.RECALL.code());
        this.updateById(msg);
        // 生成一条撤回消息
        GroupMessage recallMsg = new GroupMessage();
        recallMsg.setStatus(MessageStatus.UNSEND.code());
        recallMsg.setType(MessageType.RECALL.code());
        recallMsg.setGroupId(msg.getGroupId());
        recallMsg.setSendId(session.getId());
        recallMsg.setSendNickName(member.getShowNickName());
        recallMsg.setContent(id.toString());
        recallMsg.setSendTime(new Date());
        this.save(recallMsg);
        // 群发
        List<Long> userIds = groupMemberService.findUserIdsByGroupId(msg.getGroupId());
        GroupMessageVO msgInfo = BeanUtils.copyProperties(recallMsg, GroupMessageVO.class);
        IMGroupMessage<GroupMessageVO> sendMessage = new IMGroupMessage<>();
        sendMessage.setSender(new IMUserInfo(session.getId(), 1));
        sendMessage.setRecvIds(userIds);
        sendMessage.setData(msgInfo);
        imClient.sendGroupMessage(sendMessage);
        log.info("撤回群聊消息，发送id:{},群聊id:{},内容:{}", session.getId(), msg.getGroupId(), msg.getContent());
        return msgInfo;
    }

    @Override
    public void pullOfflineMessage(Long minId) {
        AppUser session = SecurityUtils.getAppLoginUser().getAppUser();
        if (!imClient.isOnline(session.getId())) {
            throw new GlobalException("网络连接失败，无法拉取离线消息");
        }
        // 查询用户加入的群组
        List<GroupMember> members = groupMemberService.findByUserId(session.getId());
        Map<Long, GroupMember> groupMemberMap = CollStreamUtil.toIdentityMap(members, GroupMember::getGroupId);
        Set<Long> groupIds = groupMemberMap.keySet();
        if (CollectionUtil.isEmpty(groupIds)) {
            // 关闭加载中标志
            this.sendLoadingMessage(false,session);
            return;
        }


        // 只能拉取最近3个月的,移动端只拉最近一个月
        int months = new Integer(1).equals(IMTerminalType.APP.code()) ? 1 : 3;
        Date minDate = DateUtils.addMonths(new Date(), -months);
        LambdaQueryWrapper<GroupMessage> wrapper = Wrappers.lambdaQuery();
        wrapper.gt(GroupMessage::getId, minId)
                .gt(GroupMessage::getSendTime, minDate)
                .in(GroupMessage::getGroupId, groupIds)
                .orderByAsc(GroupMessage::getId);
        List<GroupMessage> messages = this.list(wrapper);
        // 通过群聊对消息进行分组
        Map<Long, List<GroupMessage>> messageGroupMap =
            messages.stream().collect(Collectors.groupingBy(GroupMessage::getGroupId));
        // 退群前的消息
        List<GroupMember> quitMembers = groupMemberService.findQuitInMonth(session.getId());
        for (GroupMember quitMember : quitMembers) {
            wrapper = Wrappers.lambdaQuery();
            wrapper.gt(GroupMessage::getId, minId).between(GroupMessage::getSendTime, minDate, quitMember.getQuitTime())
                .eq(GroupMessage::getGroupId, quitMember.getGroupId())
                .ne(GroupMessage::getStatus, MessageStatus.RECALL.code()).orderByAsc(GroupMessage::getId);
            List<GroupMessage> groupMessages = this.list(wrapper);
            messageGroupMap.put(quitMember.getGroupId(), groupMessages);
            groupMemberMap.put(quitMember.getGroupId(), quitMember);
        }
        EXECUTOR.execute(() -> {
            // 开启加载中标志
            this.sendLoadingMessage(true,session);
            // 推送消息
            AtomicInteger sendCount = new AtomicInteger();
            messageGroupMap.forEach((groupId, groupMessages) -> {

                // 第一次拉取时,一个群最多推送1w条消息，防止前端接收能力溢出导致卡顿
                List<GroupMessage> sendMessages = groupMessages;
                if (minId <= 0 && groupMessages.size() > 10000) {
                    sendMessages = groupMessages.subList(groupMessages.size() - 10000, groupMessages.size());
                }
                // 填充消息状态
                String key = StrUtil.join(":", RedisKey.IM_GROUP_READED_POSITION, groupId);
                Object o = redisTemplate.opsForHash().get(key,session.getId().toString());
                long readedMaxId = Objects.isNull(o) ? -1 : Long.parseLong(o.toString());
                Map<Object, Object> maxIdMap = null;

                for (GroupMessage m : sendMessages) {

                    // 排除加群之前的消息
                    GroupMember member = groupMemberMap.get(m.getGroupId());
                    if (DateUtil.compare(member.getCreatedTime(), m.getSendTime()) > 0) {
                        continue;
                    }

                    // 排除不需要接收的消息
                    List<String> recvIds = CommaTextUtils.asList(m.getRecvIds());
                    if (!recvIds.isEmpty() && !recvIds.contains(session.getId().toString())) {
                        continue;
                    }

                    // 组装vo
                    GroupMessageVO vo = BeanUtils.copyProperties(m, GroupMessageVO.class);

                    // 被@用户列表
                    List<String> atIds = CommaTextUtils.asList(m.getAtUserIds());

                    vo.setAtUserIds(atIds.stream().map(Long::parseLong).collect(Collectors.toList()));

                    // 填充状态
                    vo.setStatus(readedMaxId >= m.getId() ? MessageStatus.READED.code() : MessageStatus.UNSEND.code());

                    // 针对回执消息填充已读人数
                    if (m.getReceipt()) {
                        if (Objects.isNull(maxIdMap)) {
                            maxIdMap = redisTemplate.opsForHash().entries(key);
                        }
                        int count = getReadedUserIds(maxIdMap, m.getId(), m.getSendId()).size();
                        vo.setReadedCount(count);
                    }

                    // 推送
                    IMGroupMessage<GroupMessageVO> sendMessage = new IMGroupMessage<>();
                    sendMessage.setSender(new IMUserInfo(m.getSendId(), IMTerminalType.WEB.code()));
                    sendMessage.setRecvIds(Arrays.asList(session.getId()));
                    sendMessage.setRecvTerminals(Arrays.asList(1));
                    sendMessage.setSendResult(false);
                    sendMessage.setSendToSelf(false);
                    sendMessage.setData(vo);
                    imClient.sendGroupMessage(sendMessage);
                    sendCount.getAndIncrement();
                }
                System.out.println("推送结束123");
            });
            // 关闭加载中标志
            this.sendLoadingMessage(false,session);
            log.info("拉取离线群聊消息,用户id:{},数量:{}",session.getId(), sendCount.get());
        });
    }

    @Override
    public void readedMessage(Long groupId) {
        AppUser session = SecurityUtils.getAppLoginUser().getAppUser();
        // 取出最后的消息id
        LambdaQueryWrapper<GroupMessage> wrapper = Wrappers.lambdaQuery();
        wrapper.eq(GroupMessage::getGroupId, groupId).orderByDesc(GroupMessage::getId).last("limit 1")
            .select(GroupMessage::getId);
        GroupMessage message = this.getOne(wrapper);
        if (Objects.isNull(message)) {
            return;
        }
        // 推送消息给自己的其他终端,同步清空会话列表中的未读数量
        GroupMessageVO msgInfo = new GroupMessageVO();
        msgInfo.setType(MessageType.READED.code());
        msgInfo.setSendTime(new Date());
        msgInfo.setSendId(session.getId());
        msgInfo.setGroupId(groupId);
        IMGroupMessage<GroupMessageVO> sendMessage = new IMGroupMessage<>();
        sendMessage.setSender(new IMUserInfo(session.getId(), 1));
        sendMessage.setSendToSelf(true);
        sendMessage.setData(msgInfo);
        sendMessage.setSendResult(true);
        imClient.sendGroupMessage(sendMessage);
        // 已读消息key
        String key = StrUtil.join(":", RedisKey.IM_GROUP_READED_POSITION, groupId);
        // 原来的已读消息位置
        Object maxReadedId = redisTemplate.opsForHash().get(key,session.getId().toString());
        // 记录已读消息位置
        redisTemplate.opsForHash().put(key, session.getId().toString(), message.getId());
        // 推送消息回执，刷新已读人数显示
        wrapper = Wrappers.lambdaQuery();
        wrapper.eq(GroupMessage::getGroupId, groupId);
        wrapper.gt(!Objects.isNull(maxReadedId), GroupMessage::getId, maxReadedId);
        wrapper.le(!Objects.isNull(maxReadedId), GroupMessage::getId, message.getId());
        wrapper.ne(GroupMessage::getStatus, MessageStatus.RECALL.code());
        wrapper.eq(GroupMessage::getReceipt, true);
        List<GroupMessage> receiptMessages = this.list(wrapper);
        if (CollectionUtil.isNotEmpty(receiptMessages)) {
            List<Long> userIds = groupMemberService.findUserIdsByGroupId(groupId);
            Map<Object, Object> maxIdMap = redisTemplate.opsForHash().entries(key);
            for (GroupMessage receiptMessage : receiptMessages) {
                Integer readedCount =
                    getReadedUserIds(maxIdMap, receiptMessage.getId(), receiptMessage.getSendId()).size();
                // 如果所有人都已读，记录回执消息完成标记
                if (readedCount >= userIds.size() - 1) {
                    receiptMessage.setReceiptOk(true);
                    this.updateById(receiptMessage);
                }
                msgInfo = new GroupMessageVO();
                msgInfo.setId(receiptMessage.getId());
                msgInfo.setGroupId(groupId);
                msgInfo.setReadedCount(readedCount);
                msgInfo.setReadedId(session.getId());
                msgInfo.setReceiptOk(receiptMessage.getReceiptOk());
                msgInfo.setType(MessageType.RECEIPT.code());
                sendMessage = new IMGroupMessage<>();
                sendMessage.setSender(new IMUserInfo(session.getId(), 1));
                sendMessage.setRecvIds(userIds);
                sendMessage.setData(msgInfo);
                sendMessage.setSendToSelf(false);
                sendMessage.setSendResult(false);
                imClient.sendGroupMessage(sendMessage);
            }
        }
    }

    @Override
    public List<Long> findReadedUsers(Long groupId, Long messageId) {
        AppUser session = SecurityUtils.getAppLoginUser().getAppUser();
        GroupMessage message = this.getById(messageId);
        if (Objects.isNull(message)) {
            throw new GlobalException("消息不存在");
        }
        // 是否在群聊里面
        GroupMember member = groupMemberService.findByGroupAndUserId(groupId, session.getId());
        if (Objects.isNull(member) || member.getQuit()) {
            throw new GlobalException("您已不在群聊里面");
        }
        // 已读位置key
        String key = StrUtil.join(":", RedisKey.IM_GROUP_READED_POSITION, groupId);
        // 一次获取所有用户的已读位置
        Map<Object, Object> maxIdMap = redisTemplate.opsForHash().entries(key);
        // 返回已读用户的id集合
        return getReadedUserIds(maxIdMap, message.getId(), message.getSendId());
    }

    @Override
    public List<GroupMessageVO> findHistoryMessage(Long groupId, Long page, Long size) {
        AppUser session = SecurityUtils.getAppLoginUser().getAppUser();
        page = page > 0 ? page : 1;
        size = size > 0 ? size : 10;
        Long userId = session.getId();
        long stIdx = (page - 1) * size;
        // 群聊成员信息
        GroupMember member = groupMemberService.findByGroupAndUserId(groupId, userId);
        if (Objects.isNull(member) || member.getQuit()) {
            throw new GlobalException("您已不在群聊中");
        }
        // 查询聊天记录，只查询加入群聊时间之后的消息
        QueryWrapper<GroupMessage> wrapper = new QueryWrapper<>();
        wrapper.lambda().eq(GroupMessage::getGroupId, groupId).gt(GroupMessage::getSendTime, member.getCreatedTime())
            .ne(GroupMessage::getStatus, MessageStatus.RECALL.code()).orderByDesc(GroupMessage::getId)
            .last("limit " + stIdx + "," + size);
        List<GroupMessage> messages = this.list(wrapper);
        List<GroupMessageVO> messageInfos =
            messages.stream().map(m -> BeanUtils.copyProperties(m, GroupMessageVO.class)).collect(Collectors.toList());
        log.info("拉取群聊记录，用户id:{},群聊id:{}，数量:{}", userId, groupId, messageInfos.size());
        return messageInfos;
    }

    private List<Long> getReadedUserIds(Map<Object, Object> maxIdMap, Long messageId, Long sendId) {
        List<Long> userIds = new LinkedList<>();
        maxIdMap.forEach((k, v) -> {
            Long userId = Long.valueOf(k.toString());
            long maxId = Long.parseLong(v.toString());
            // 发送者不计入已读人数
            if (!sendId.equals(userId) && maxId >= messageId) {
                userIds.add(userId);
            }
        });
        return userIds;
    }

    private void sendLoadingMessage(Boolean isLoadding,  AppUser session) {
        GroupMessageVO msgInfo = new GroupMessageVO();
        msgInfo.setType(MessageType.LOADING.code());
        msgInfo.setContent(isLoadding.toString());
        IMGroupMessage sendMessage = new IMGroupMessage<>();

        sendMessage.setSender(new IMUserInfo(session.getId(), 1));
        sendMessage.setRecvIds(Arrays.asList(session.getId()));
        sendMessage.setRecvTerminals(Arrays.asList(1));
        sendMessage.setData(msgInfo);
        sendMessage.setSendToSelf(false);
        sendMessage.setSendResult(false);
        System.out.println("发送LOADING"+isLoadding);
        imClient.sendGroupMessage(sendMessage);
    }

    @Override
    public String getMaxReadedId(Long groupId) {
        AppUser session = SecurityUtils.getAppLoginUser().getAppUser();
        // 已读消息key
        String key = StrUtil.join(":", RedisKey.IM_GROUP_READED_POSITION, groupId);
        // 原来的已读消息位置
        Object maxReadedId = redisTemplate.opsForHash().get(key,session.getId().toString());

        if (ObjectUtil.isEmpty(maxReadedId)) {
            return "-1";
        }
//        LambdaQueryWrapper<GroupMessage> wrapper = Wrappers.lambdaQuery();
//        wrapper
//                .eq(GroupMessage::getGroupId, groupId)
//                .eq(GroupMessage::getStatus, MessageStatus.READED.code()).orderByDesc(GroupMessage::getId)
//                .select(GroupMessage::getId).last("limit 1");
//        GroupMessage message = this.getOne(wrapper);
//        if (Objects.isNull(message)) {
//            return "-1";
//        }
        return maxReadedId.toString();
    }
}
