package com.wecloud.im.service.impl;

import com.wecloud.im.entity.ImClientBlacklist;
import com.wecloud.im.param.ImConversationQueryVo;
import com.wecloud.im.post.MessageBuilder;
import io.geekidea.springbootplus.framework.common.api.ApiCode;
import io.geekidea.springbootplus.framework.common.api.ApiResult;
import io.geekidea.springbootplus.framework.common.exception.BusinessException;
import io.geekidea.springbootplus.framework.common.service.impl.BaseServiceImpl;
import io.geekidea.springbootplus.framework.core.pagination.PageInfo;
import io.geekidea.springbootplus.framework.core.pagination.Paging;
import io.geekidea.springbootplus.framework.shiro.util.SecurityUtils;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.function.Function;
import java.util.stream.Collectors;

import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.core.metadata.OrderItem;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.wecloud.dispatch.common.BaseRequest;
import com.wecloud.im.entity.ImApiMessageOnlineSend;
import com.wecloud.im.entity.ImApplication;
import com.wecloud.im.entity.ImClient;
import com.wecloud.im.entity.ImConversation;
import com.wecloud.im.entity.ImConversationMembers;
import com.wecloud.im.entity.ImInbox;
import com.wecloud.im.entity.ImMessage;
import com.wecloud.im.entity.ImMessageOnlineSend;
import com.wecloud.im.event.ClientSendMessageEvent;
import com.wecloud.im.mapper.ImMessageMapper;
import com.wecloud.im.mq.MqSender;
import com.wecloud.im.param.ChatContentVo;
import com.wecloud.im.param.ConversationExtParam;
import com.wecloud.im.param.GetReadersParam;
import com.wecloud.im.param.GroupChatMessageParam;
import com.wecloud.im.param.GroupChatStatusMessageParam;
import com.wecloud.im.param.ImClientSimpleDto;
import com.wecloud.im.param.ImHistoryMessagePageParam;
import com.wecloud.im.param.MsgDeleteParam;
import com.wecloud.im.param.PrivateChatMessageParam;
import com.wecloud.im.param.PrivateChatStatusMessageParam;
import com.wecloud.im.param.PushExtParam;
import com.wecloud.im.param.SingleUserNotificationParam;
import com.wecloud.im.param.add.ImMsgRecall;
import com.wecloud.im.param.add.ImMsgSendToOnlineClient;
import com.wecloud.im.param.add.ImMsgUpdate;
import com.wecloud.im.param.add.ServerImConversationCreate;
import com.wecloud.im.sdk.enums.ChatTypeEnum;
import com.wecloud.im.service.ContextService;
import com.wecloud.im.service.ImApplicationService;
import com.wecloud.im.service.ImClientService;
import com.wecloud.im.service.ImConversationMembersService;
import com.wecloud.im.service.ImConversationService;
import com.wecloud.im.service.ImInboxService;
import com.wecloud.im.service.ImMessageService;
import com.wecloud.im.thousandchat.service.ThousandChatService;
import com.wecloud.im.vo.ImMessageOfflineListVo;
import com.wecloud.im.vo.OfflineMsgDto;
import com.wecloud.im.vo.ReaderList;
import com.wecloud.im.vo.ReaderVo;
import com.wecloud.im.ws.enums.MsgTypeEnum;
import com.wecloud.im.ws.enums.WsResponseCmdEnum;
import com.wecloud.im.ws.model.WsResponse;
import com.wecloud.im.ws.model.request.PushVO;
import com.wecloud.im.ws.model.request.ReceiveDataVO;
import com.wecloud.im.ws.model.request.ReceiveVO;
import com.wecloud.im.ws.sender.ChannelSender;
import com.wecloud.pushserver.client.model.constant.MqConstant;
import com.wecloud.pushserver.client.model.dto.PushDTO;
import com.wecloud.utils.JsonUtils;
import com.wecloud.utils.SnowflakeUtil;


/**
 * 消息存储表 服务实现类
 *
 * @author wei
 * @since 2021-04-29
 */
@Slf4j
@Service
public class ImMessageServiceImpl extends BaseServiceImpl<ImMessageMapper, ImMessage> implements ImMessageService {

    @Autowired
    private ImMessageMapper imMessageMapper;

    @Autowired
    private ImClientService imClientService;

    @Autowired
    private ImConversationService imConversationService;

    @Autowired
    private ImApplicationService imApplicationService;

    @Autowired
    private ImConversationMembersService imConversationMembersService;

    @Autowired
    private ContextService contextService;

    @Autowired
    private ChannelSender channelSender;

    @Autowired
    private ThousandChatService thousandChatService;

    @Autowired
    private MqSender mqSender;

    @Autowired
    private ImInboxService imInboxService;

    @Autowired
    private ApplicationEventPublisher eventPublisher;

    @Override
    @Transactional(rollbackFor = Exception.class)
    public ApiResult<Boolean> restApiImMessageSend(ImMsgSendToOnlineClient imMsgSendToOnlineClient, ImApplication imApplication) {


        // 查询该会话所有成员
        List<ImConversationMembers> membersList = imConversationMembersService.list(
                new QueryWrapper<ImConversationMembers>().lambda()
                        .eq(ImConversationMembers::getFkConversationId, imMsgSendToOnlineClient.getConversationId())
        );
        if (membersList.isEmpty()) {
            log.info("membersList为空,toConversationId:" + imMsgSendToOnlineClient.getConversationId());
            return ApiResult.fail();

        }

        ImApiMessageOnlineSend imApiMessageOnlineSend = new ImApiMessageOnlineSend();
        imApiMessageOnlineSend.setCreateTime(new Date());

        String attributes = JsonUtils.encodeJson(imMsgSendToOnlineClient.getContent());
        imApiMessageOnlineSend.setContent(attributes);

        imApiMessageOnlineSend.setConversationId(imMsgSendToOnlineClient.getConversationId());


        // 遍历发送
        for (ImConversationMembers conversationMembers : membersList) {

            // 查询接收方
            ImClient imClientReceiver = imClientService.getOne(new QueryWrapper<ImClient>().lambda()
                    .eq(ImClient::getFkAppid, imApplication.getId())
                    .eq(ImClient::getId, conversationMembers.getFkClientId()));
            if (imClientReceiver == null) {
                continue;
            }

            WsResponse<ImApiMessageOnlineSend> responseModel = new WsResponse<>();
            responseModel.setCmd(WsResponseCmdEnum.REST_API_MSG.getCmdCode());
            ApiResult<Boolean> result = ApiResult.result(ApiCode.SUCCESS);
            responseModel.setCode(result.getCode());
            responseModel.setMsg(result.getMessage());
            responseModel.setData(imApiMessageOnlineSend);
            responseModel.setReqId(null);

            //  向接收方推送
            channelSender.sendMsg(responseModel, imClientReceiver.getId());

        }


        return null;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public ImMessage saveImMessage(ImApplication imApplication, ImClient imClientSender, long messageId, ReceiveVO receiveVO, ReceiveDataVO sysParam) {

        ImMessage imMessage = new ImMessage();

        //  数据库字段类型为JSON格式
        //  因mysql关系型数据库非MongoDB文档类型数据库,第三方应用拓展的自定义参数名和值需使用json格式落库
        String contentJsonString = JsonUtils.encodeJson(receiveVO.getData());
        imMessage.setContent(contentJsonString);

        imMessage.setId(messageId);
        imMessage.setCreateTime(new Date());
        imMessage.setFkAppid(imApplication.getId());
        imMessage.setSender(imClientSender.getId());
        imMessage.setWithdraw(false);
        imMessage.setEvent(false);
        imMessage.setSystemFlag(false);
        imMessage.setSendStatus(2);
        imMessage.setMsgType(sysParam.getType());
        imMessage.setFkConversationId(sysParam.getToConversation());
        this.save(imMessage);

//        eventPublisher.publishEvent(new ClientSendMessageEvent(imApplication.getId(), imMessage));

        return imMessage;
    }

    @Override
    public ImMessage saveImMessage(ImClientSimpleDto client, ChatContentVo data) {

        ImMessage imMessage = new ImMessage();

        //  数据库字段类型为JSON格式
        //  因mysql关系型数据库非MongoDB文档类型数据库,第三方应用拓展的自定义参数名和值需使用json格式落库
        String contentJsonString = JsonUtils.encodeJson(data);
        imMessage.setContent(contentJsonString);

        imMessage.setId(SnowflakeUtil.getId());
        imMessage.setCreateTime(new Date());
        imMessage.setFkAppid(client.getFkAppid());
        imMessage.setSender(client.getId());
        imMessage.setWithdraw(false);
        imMessage.setEvent(false);
        imMessage.setSystemFlag(false);
        imMessage.setSendStatus(2);
        imMessage.setMsgType(data.getType());
        imMessage.setAt(data.getAt());
        imMessage.setFkConversationId(data.getToConversation());
        this.save(imMessage);

//        eventPublisher.publishEvent(new ClientSendMessageEvent(client.getFkAppid(), imMessage));

        return imMessage;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public Boolean withdrawMsg(ImMsgRecall imMsgRecall) {
        ImClient imClientSender = contextService.getImClientIfNotNullOrThrow();
        ImApplication imApplication = contextService.getImApplicationIfNotNullOrThrow(imClientSender.getFkAppid());

        ImMessage messageById = this.getById(imMsgRecall.getMsgId());
        // 判断该消息是否是该客户端发送 todo 单向撤回、双向撤回开关可配置
        boolean withdrawOther = true;
        if (!withdrawOther && !messageById.getSender().equals(imClientSender.getId())) {
            throw new BusinessException("不可撤回别人发送的消息");
        }
        if (messageById.getWithdraw()) {
            return Boolean.TRUE;
        }
        ImClient msgOwner = imClientService.getCacheImClient(messageById.getSender());
        if (msgOwner == null) {
            throw new BusinessException("未查找到消息发送者");
        }

        // 修改消息体
        messageById.setWithdraw(Boolean.TRUE);
        messageById.setWithdrawTime(new Date());
        // 清空消息
        messageById.setContent("{}");
        boolean saveOk = this.updateById(messageById);

        eventPublisher.publishEvent(new ClientSendMessageEvent(imApplication.getId(), messageById));

        // 组装消息
        Map<String, Object> content = new HashMap<>();
        content.put("operator", imClientSender.getClientId());
        content.put("beOperatedMsgId", imMsgRecall.getMsgId());
        content.put("msgOwner", msgOwner.getClientId());
        ImMessage imMessage = MessageBuilder.buildEventMessage(MsgTypeEnum.MSG_WITHDRAW, imApplication,
                imClientSender, new ImConversation().setId(messageById.getFkConversationId()), JsonUtils.encodeJson(content));
        this.save(imMessage);

        if (saveOk) {
            // 查询该会话所有成员
            List<ImConversationMembers> membersList = imConversationMembersService.list(
                    new QueryWrapper<ImConversationMembers>().lambda()
                            .eq(ImConversationMembers::getFkConversationId, messageById.getFkConversationId())
                            .notIn(ImConversationMembers::getFkClientId, imClientSender.getId())
            );
            if (membersList.isEmpty()) {
                log.info("membersList为空,toConversationId:" + messageById.getFkConversationId());
                throw new BusinessException("该会话成员列表为空");
            }
            // 删除inbox数据
            imInboxService.remove(new QueryWrapper<ImInbox>().lambda()
                    .eq(ImInbox::getFkMsgId, imMsgRecall.getMsgId()));
            // 遍历发送
            for (ImConversationMembers member : membersList) {
                // 封装响应的实体
                ImMessageOnlineSend imMessageOnlineSend = new ImMessageOnlineSend();
                BeanUtils.copyProperties(messageById, imMessageOnlineSend);
                imMessageOnlineSend.setType(MsgTypeEnum.MSG_WITHDRAW.getUriCode());
                imMessageOnlineSend.setMsgId(messageById.getId());
                imMessageOnlineSend.setCreateTime(new Date());
                imMessageOnlineSend.setSender(imClientSender.getClientId());
                Map<String, Object> contentMap = Maps.newHashMap();
                contentMap.put("msgOwner", msgOwner.getClientId());
                imMessageOnlineSend.setContent(contentMap);
                imMessageOnlineSend.setConversationId(member.getFkConversationId());
                imMessageOnlineSend.setWithdraw(Boolean.TRUE);
                imMessageOnlineSend.setEvent(Boolean.TRUE);

                //  向接收方推送
                WsResponse<ImMessageOnlineSend> responseModel = new WsResponse<>();
                responseModel.setCmd(WsResponseCmdEnum.ONLINE_EVENT_MSG.getCmdCode());
                ApiResult<Boolean> result = ApiResult.result(ApiCode.SUCCESS);
                responseModel.setCode(result.getCode());
                responseModel.setMsg(result.getMessage());
                responseModel.setData(imMessageOnlineSend);
                responseModel.setReqId(null);
                channelSender.sendMsg(responseModel, member.getFkClientId());

                // 异步推送系统通知消息
                PushDTO pushDTO = mqSender.buildPushDto(imMsgRecall.getPush(), member.getFkClientId(), member.getClientId(), imApplication);
                if (pushDTO != null) {
                    mqSender.orderSend(MqConstant.Topic.IM_ORDER_MSG_TOPIC, MqConstant.Tag.IM_ORDER_MSG_TAG, pushDTO);
                }
            }

            return Boolean.TRUE;

        } else {
            return Boolean.TRUE;
        }
    }

    @Override
    public Boolean deleteMsg(MsgDeleteParam param) {
        ImClient imClientSender = contextService.getImClientIfNotNullOrThrow();
        ImApplication imApplication = contextService.getImApplicationIfNotNullOrThrow(imClientSender.getFkAppid());
        // todo 单向撤回、双向撤回开关可配置
        Boolean deleteOther = true;
        List<ImMessage> imMessageList = this.listByIds(param.getMsgIds());
        if (CollectionUtils.isEmpty(imMessageList)) {
            throw new BusinessException("查无消息");
        }
        for (ImMessage message : imMessageList) {
            // 判断该消息是否是该客户端发送
            if (!deleteOther && !message.getSender().equals(imClientSender.getId())) {
                throw new BusinessException("不可删除别人发送的消息");
            }
            if (message.getIsDelete() == 2) {
                continue;
            }
            // 修改消息体
            message.setIsDelete(2);
            message.setUpdateDate(new Date());
            boolean saveOk = this.updateById(message);
            ImClient msgOwner = imClientService.getCacheImClient(message.getSender());
            if (msgOwner == null) {
                throw new BusinessException("未查找到消息发送者");
            }

            // 组装消息
            Map<String, Object> content = new HashMap<>();
            content.put("operator", imClientSender.getClientId());
            content.put("beOperatedMsgId", message.getId());
            content.put("msgOwner", msgOwner.getClientId());
            ImMessage imMessage = MessageBuilder.buildEventMessage(MsgTypeEnum.MSG_DELETE, imApplication,
                    imClientSender, new ImConversation().setId(message.getFkConversationId()), JsonUtils.encodeJson(content));
            this.save(imMessage);

            if (saveOk) {
                // 查询该会话所有成员
                List<ImConversationMembers> membersList = imConversationMembersService.list(
                        new QueryWrapper<ImConversationMembers>().lambda()
                                .eq(ImConversationMembers::getFkConversationId, message.getFkConversationId())
                                .notIn(ImConversationMembers::getFkClientId, imClientSender.getId())
                );
                if (membersList.isEmpty()) {
                    log.info("membersList为空,toConversationId:" + message.getFkConversationId());
                    throw new BusinessException("该会话成员列表为空");
                }
                // 删除inbox数据
                imInboxService.remove(new QueryWrapper<ImInbox>().lambda()
                        .eq(ImInbox::getFkMsgId, message.getId()));
                // 遍历发送
                for (ImConversationMembers member : membersList) {
                    // 封装响应的实体
                    ImMessageOnlineSend imMessageOnlineSend = new ImMessageOnlineSend();
                    BeanUtils.copyProperties(message, imMessageOnlineSend);
                    imMessageOnlineSend.setType(MsgTypeEnum.MSG_DELETE.getUriCode());
                    imMessageOnlineSend.setMsgId(message.getId());
                    imMessageOnlineSend.setCreateTime(new Date());
                    imMessageOnlineSend.setSender(imClientSender.getClientId());
                    Map<String, Object> contentMap = Maps.newHashMap();
                    contentMap.put("msgOwner", msgOwner.getClientId());
                    imMessageOnlineSend.setContent(contentMap);
                    imMessageOnlineSend.setConversationId(member.getFkConversationId());
                    imMessageOnlineSend.setEvent(Boolean.TRUE);

                    //  向接收方推送
                    WsResponse<ImMessageOnlineSend> responseModel = new WsResponse<>();
                    responseModel.setCmd(WsResponseCmdEnum.ONLINE_EVENT_MSG.getCmdCode());
                    ApiResult<Boolean> result = ApiResult.result(ApiCode.SUCCESS);
                    responseModel.setCode(result.getCode());
                    responseModel.setMsg(result.getMessage());
                    responseModel.setData(imMessageOnlineSend);
                    responseModel.setReqId(null);
                    channelSender.sendMsg(responseModel, member.getFkClientId());
                }
                return Boolean.TRUE;
            } else {
                throw new BusinessException("删除消息错误，稍后重试");
            }
        }

        return Boolean.TRUE;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public Boolean updateMsgById(ImMsgUpdate imMsgUpdate) {

//        ImClient client = imClientService.getCurentClient();

/*        // 判断该消息是否是该客户端发送
        ImMessage messageById = this.getById(imMsgUpdate.getId());
        if (!messageById.getSender().equals(client.getId())) {
            log.error("判断该消息是否是该客户端发送");
            return ApiResult.fail();
        }*/

        ImMessage imMessage = new ImMessage();
        imMessage.setId(imMsgUpdate.getMsgId());

        String content = JsonUtils.encodeJson(imMsgUpdate.getContent());
        imMessage.setContent(content);

        if (this.updateById(imMessage)) {
            return Boolean.TRUE;
        } else {
            throw new BusinessException("修改消息错误");
        }
    }

    @Transactional(rollbackFor = Exception.class)
    @Override
    public boolean saveImMessage(ImMessage imMessage) throws Exception {
        return super.save(imMessage);
    }

    /**
     * 获取读取人员
     *
     * @param param
     * @return
     */
    @Override
    public ReaderList getReaders(GetReadersParam param) {
        ImClient imClientSender = contextService.getImClientIfNotNullOrThrow();
        List<ReaderVo> readerVos = imMessageMapper.getReaders(imClientSender.getId(), param);
        if (CollectionUtils.isEmpty(readerVos)) {
            return null;
        }
        List<ReaderVo> readList = Lists.newArrayList();
        List<ReaderVo> unReadList = Lists.newArrayList();
        for (ReaderVo readerVo : readerVos) {
            if (readerVo.getReadMsgStatus() == 0) {
                unReadList.add(readerVo);
            } else {
                readList.add(readerVo);
            }
        }
        ReaderList readerList = new ReaderList();
        readerList.setReadList(readList);
        readerList.setUnReadList(unReadList);
        return readerList;
    }

    @Override
    public Paging<OfflineMsgDto> getHistoryMsgConversationId(ImHistoryMessagePageParam param) {
        Page<ImMessage> page = new PageInfo<>(param, OrderItem.desc(getLambdaColumn(ImMessage::getCreateTime)));
        ImClient currentClient = imClientService.getCurrentClient();
        param.setCurrentFkClientId(currentClient.getId());
        ImConversation imConversation = imConversationService.getById(param.getConversationId());
        if (imConversation == null) {
            return new Paging<>(null);
        }
        if (ChatTypeEnum.NORMAL_GROUP.getCode().equals(imConversation.getChatType())) {
            // 普通群聊判断是否已被踢出群
            List<ImConversationMembers> members = imConversationMembersService.list(
                    new QueryWrapper<ImConversationMembers>().lambda()
                            .eq(ImConversationMembers::getFkConversationId, param.getConversationId())
                            .eq(ImConversationMembers::getClientId, currentClient.getClientId()));
            if (CollectionUtils.isEmpty(members)) {
                return new Paging<>(null);
            }
        }
        IPage<OfflineMsgDto> iPage = imMessageMapper.getHistoryMsgConversationId(page, param);

        return new Paging<>(iPage);
    }
    @Override
    public Paging<OfflineMsgDto> getHistoryMsgConversationIdNew(ImHistoryMessagePageParam param) {
        Page<ImMessage> page = new PageInfo<>(param, OrderItem.desc(getLambdaColumn(ImMessage::getCreateTime)));
        ImClient currentClient = imClientService.getCurrentClient();
        param.setCurrentFkClientId(currentClient.getId());
        IPage<OfflineMsgDto> iPage = imMessageMapper.getHistoryMsgConversationIdNew(page, param);

        return new Paging<>(iPage);
    }


    /**
     * 查询用户所有离线消息
     *
     * @return
     * @throws Exception
     */
    @Override
    public List<ImMessageOfflineListVo> getOfflineList() throws Exception {

        // 返回数据
        List<ImMessageOfflineListVo> imMessageOfflineListVoList = new ArrayList<>();

        ImClient client = imClientService.getCurrentClient();

        // 获取加入的所有会话
        List<ImConversation> myImConversationList = imConversationService.getMyImConversationList();

        List<ImConversation> thousandConversations = new ArrayList<>();
        // 遍历会话列表, 查询每个会话列表的离线消息
        for (ImConversation imConversation : myImConversationList) {

            //万人群 暂时跳过，后面统一处理
            if (ChatTypeEnum.THOUSAND_GROUP.getCode().equals(imConversation.getChatType())) {
                thousandConversations.add(imConversation);
                continue;
            }

            //根据客户端id与会话id 查询离线消息
            List<OfflineMsgDto> offlineListByClientAndConversation = this.getOfflineListByClientAndConversation(client.getId(), imConversation.getId());

            // 房间消息为空则不添加
            if (offlineListByClientAndConversation.isEmpty()) {
                continue;
            }
            ImMessageOfflineListVo imMessageOfflineListVo = new ImMessageOfflineListVo();
            imMessageOfflineListVo.setMsgList(offlineListByClientAndConversation);
            imMessageOfflineListVo.setConversationId(imConversation.getId());
            imMessageOfflineListVoList.add(imMessageOfflineListVo);
        }
        // 万人群的部分
        if (CollectionUtils.isNotEmpty(thousandConversations)) {
            List<ImMessageOfflineListVo> offlineMsgs = thousandChatService.findOfflineMsgs(thousandConversations);
            imMessageOfflineListVoList.addAll(offlineMsgs);
        }

        return imMessageOfflineListVoList;
    }

    /**
     * 根据客户端id与会话id 查询离线消息
     *
     * @param clientId
     * @param conversationId
     * @return
     */
    @Override
    public List<OfflineMsgDto> getOfflineListByClientAndConversation(Long clientId, Long conversationId) {
        return imMessageMapper.getOfflineListByClientAndConversation(clientId, conversationId);
    }


    @Override
    public OfflineMsgDto getReceivedLastMsgByConversationId(Long clientId, Long conversationId) {
        return imMessageMapper.getReceivedLastMsgByConversationId(clientId, conversationId);
    }

    @Override
    public OfflineMsgDto getLastMsgByConversationId(Long conversationId, Long currentFkClientId) {
        return imMessageMapper.getLastMsgByConversationId(conversationId, currentFkClientId);
    }

    @Override
    public Boolean groupMessagePublish(GroupChatMessageParam param) {
        Long appId = SecurityUtils.getCurrentAppId();
        ImApplication application = imApplicationService.getCacheById(appId);
        // 获取发件人信息
        String senderClientId = param.getFromUserId();
        ImClient sender = imClientService.getCacheImClient(appId, senderClientId);
        if (sender == null) {
            throw new BusinessException("id为 " + senderClientId + " 的发件人不存在");
        }
        // 获取群聊信息
        String conversationIdsStr = param.getToGroupIds();
        List<String> conversationIdList = Arrays.asList(conversationIdsStr.split(","));
        List<ImConversation> conversationList = imConversationService.listByIds(conversationIdList);
        // 校验群聊信息
        if (conversationList.size() <= 0) {
            throw new BusinessException("群聊ids为 " + conversationIdsStr + " 的群聊列表不存在");
        }
        // 是否指定用户
        List<String> toUserIds = Collections.emptyList();
        boolean isToUser = false;
        if (StringUtils.isNotBlank(param.getToUserIds())) {
            isToUser = true;
            toUserIds = Arrays.asList(param.getToUserIds().split(","));
        }
        for (ImConversation conversation : conversationList) {
            // 获取群成员
            List<ImConversationMembers> membersList = Collections.emptyList();
            if (isToUser) {
                // 指定群成员
                membersList = imConversationMembersService.list(
                        new QueryWrapper<ImConversationMembers>().lambda()
                                .eq(ImConversationMembers::getFkConversationId, conversation.getId()).in(ImConversationMembers::getClientId, toUserIds));
            } else {
                membersList = imConversationMembersService.list(
                        new QueryWrapper<ImConversationMembers>().lambda()
                                .eq(ImConversationMembers::getFkConversationId, conversation.getId()));
            }

            // 组装消息
            ImMessage message = assembleImMessage(appId, sender, conversation.getId(), param.getMessageType(), false,
                    JSONObject.toJSONString(param.getContent()));
            // 持久化
            this.save(message);
            // 拼装发送消息体
            ImMessageOnlineSend imMessageOnlineSend = assembleImMessageOnlineSend(message, sender, appId);
            // 入库 保存收件箱
            batchSaveInbox(application, senderClientId, conversation, imMessageOnlineSend.getMsgId(), membersList);
            // 发送消息
            for (ImConversationMembers members : membersList) {
                if (members.getClientId().equals(senderClientId) && !param.getIncludeSender()) {
                    // 是否是发送者 且includeSender不为true
                    continue;
                }
                // 在线用户直接发消息
                sendMsgForOnline(members.getFkClientId(), imMessageOnlineSend);

                // 离线消息推送
                pushMsgToOfflineMembers(application, members, param.getPushContent(), param.getPushExt());
            }
        }

        return true;
    }


    private void batchSaveInbox(ImApplication imApplication, String senderClientId, ImConversation conversation, long messageId,
                                List<ImConversationMembers> membersList) {
        List<ImInbox> inboxes = Lists.newArrayList();
        for (ImConversationMembers member : membersList) {
            if (senderClientId.equals(member.getClientId())) {
                continue;
            }
            ImInbox imInbox = new ImInbox();
            imInbox.setId(SnowflakeUtil.getId());
            imInbox.setCreateTime(new Date());
            imInbox.setFkAppid(imApplication.getId());
            imInbox.setReceiver(member.getFkClientId());
            imInbox.setFkMsgId(messageId);
            imInbox.setReadMsgStatus(0);
            imInbox.setReceiverMsgStatus(0);
            imInbox.setFkConversationId(conversation.getId());
            inboxes.add(imInbox);
        }
        if (CollectionUtils.isNotEmpty(inboxes)) {
            imInboxService.saveBatch(inboxes);
        }
    }

    @Override
    public Boolean groupStatusMessagePublish(GroupChatStatusMessageParam param) {
        Long appId = SecurityUtils.getCurrentAppId();
        // 获取发件人信息
        String senderClientId = param.getFromUserId();
        ImClient sender = imClientService.getCacheImClient(appId, senderClientId);
        if (sender == null) {
            throw new BusinessException("id为 " + senderClientId + " 的发件人不存在");
        }
        // 获取群聊信息
        String conversationIdsStr = param.getToGroupIds();
        List<String> conversationIdList = Arrays.asList(conversationIdsStr.split(","));
        List<ImConversation> conversationList = imConversationService.listByIds(conversationIdList);
        // 校验群聊信息
        if (conversationList.size() <= 0) {
            throw new BusinessException("群聊ids为 " + conversationIdsStr + " 的群聊列表不存在");
        }
        // 是否指定用户
        List<String> toUserIds = Collections.emptyList();
        boolean isToUser = false;
        if (StringUtils.isNotBlank(param.getToUserIds())) {
            isToUser = true;
            toUserIds = Arrays.asList(param.getToUserIds().split(","));
        }
        // 开始发送
        for (ImConversation conversation : conversationList) {
            // 获取群成员
            List<ImConversationMembers> membersList = Collections.emptyList();
            if (isToUser) {
                // 指定群成员
                membersList = imConversationMembersService.list(
                        new QueryWrapper<ImConversationMembers>().lambda()
                                .eq(ImConversationMembers::getFkConversationId, conversation.getId()).in(ImConversationMembers::getClientId, toUserIds));
            } else {
                membersList = imConversationMembersService.list(
                        new QueryWrapper<ImConversationMembers>().lambda()
                                .eq(ImConversationMembers::getFkConversationId, conversation.getId()));
            }

            // 组装消息
            ImMessage message = assembleImMessage(appId, sender, conversation.getId(), param.getMessageType(), true,
                    JSONObject.toJSONString(param.getContent()));
            if (isToUser && CollectionUtils.isNotEmpty(membersList)) {
                // 指定群内成员发送 落库接收人  仅限少量接收人场景时落库
                List<Long> memberFkClientIds = membersList.stream().map(m -> m.getFkClientId()).collect(Collectors.toList());
                String receivers = StringUtils.join(memberFkClientIds, ",");
                if (receivers.length() < 200) {
                    message.setReceivers(receivers);
                }
            }
            // 持久化
            this.save(message);
            // 拼装发送消息体
            ImMessageOnlineSend imMessageOnlineSend = assembleImMessageOnlineSend(message, sender, appId);

            // 发送消息
            for (ImConversationMembers members : membersList) {
                if (members.getClientId().equals(senderClientId) && !param.getIncludeSender()) {
                    // 是否是发送者 且includeSender不为true
                    continue;
                }
                // 在线用户直接发消息
                sendMsgForOnline(members.getFkClientId(), imMessageOnlineSend);
            }
        }
        return true;
    }

    @Override
    public Boolean privateMessagePublish(PrivateChatMessageParam param) {
        Long appId = SecurityUtils.getCurrentAppId();
        ImApplication application = imApplicationService.getCacheById(appId);
        String senderClientId = param.getFromUserId();
        String toUserIdsStr = param.getToUserIds();
        List<String> toUserIdList = Arrays.asList(toUserIdsStr.split(","));
        // 获取发件人信息
        ImClient sender =
                imClientService.getOne(Wrappers.<ImClient>lambdaQuery().eq(ImClient::getFkAppid, appId).eq(ImClient::getClientId, senderClientId));
        if (sender == null) {
            throw new BusinessException("id为 " + senderClientId + " 的发件人不存在");
        }

        Map<Long, ImConversation> conversationMapGroupById = new HashMap<>();
        Map<Long, List<ImConversationMembers>> conMembersMapGroupByConId = new HashMap<>();
        beforePublishPrivateMessage(application, sender, toUserIdList, conversationMapGroupById, conMembersMapGroupByConId);
        // 开始发送消息
        conversationMapGroupById.forEach((conversationId, conversation) -> {
            List<ImConversationMembers> membersList = conMembersMapGroupByConId.getOrDefault(conversationId,
                    Collections.emptyList());
            // 组装消息
            ImMessage message = assembleImMessage(appId, sender, conversation.getId(), param.getMessageType(), false, JSONObject.toJSONString(param.getContent()));
            // 持久化
            this.save(message);
            // 拼装发送消息体
            ImMessageOnlineSend imMessageOnlineSend = assembleImMessageOnlineSend(message, sender, appId);

            // 发送消息
            for (ImConversationMembers members : membersList) {
                if (members.getClientId().equals(senderClientId) && !param.getIncludeSender()) {
                    // 是否是发送者 且includeSender不为true
                    continue;
                }

                // 入库 保存收件箱
                saveImInbox(application, conversation.getId(), imMessageOnlineSend.getMsgId(),
                        members, SnowflakeUtil.getId());

                // 在线用户直接发消息
                sendMsgForOnline(members.getFkClientId(), imMessageOnlineSend);

                // 离线消息推送
                pushMsgToOfflineMembers(application, members, param.getPushContent(), param.getPushExt());
            }

        });
        return true;
    }

    @Override
    public Boolean privateStatusMessagePublish(PrivateChatStatusMessageParam param) {
        Long appId = SecurityUtils.getCurrentAppId();
        ImApplication application = imApplicationService.getCacheById(appId);
        String senderClientId = param.getFromUserId();
        String toUserIdsStr = param.getToUserIds();
        List<String> toUserIdList = Arrays.asList(toUserIdsStr.split(","));
        // 获取发件人信息
        ImClient sender = imClientService.getCacheImClient(appId, senderClientId);
        if (sender == null) {
            throw new BusinessException("id为 " + senderClientId + " 的发件人不存在");
        }

        Map<Long, ImConversation> conversationMapGroupById = new HashMap<>();
        Map<Long, List<ImConversationMembers>> conMembersMapGroupByConId = new HashMap<>();
        beforePublishPrivateMessage(application, sender, toUserIdList, conversationMapGroupById, conMembersMapGroupByConId);
        // 开始发送消息
        conversationMapGroupById.forEach((conversationId, conversation) -> {
            List<ImConversationMembers> membersList = conMembersMapGroupByConId.getOrDefault(conversationId,
                    Collections.emptyList());
            // 组装消息
            ImMessage message = assembleImMessage(appId, sender, conversation.getId(), param.getMessageType(), true,
                    JSONObject.toJSONString(param.getContent()));
            // 持久化
            this.save(message);
            // 拼装发送消息体
            ImMessageOnlineSend imMessageOnlineSend = assembleImMessageOnlineSend(message, sender, appId);

            // 发送消息
            for (ImConversationMembers members : membersList) {
                if (members.getClientId().equals(senderClientId) && !param.getIncludeSender()) {
                    // 是否是发送者 且includeSender不为true
                    continue;
                }
                log.info("下发消息 clientId: {} 消息内容: {}", members.getFkClientId(), JSON.toJSONString(imMessageOnlineSend));
                // 在线用户直接发消息
                sendMsgForOnline(members.getFkClientId(), imMessageOnlineSend);
            }

        });
        return true;
    }

    @Override
    public Boolean singleUserNotification(SingleUserNotificationParam param) {
        ImApplication application = imApplicationService.getCacheById(SecurityUtils.getCurrentAppId());
        String senderClientId = param.getFromUserId();
        String toUserIdsStr = param.getToUserIds();
        List<String> toUserIdList = Arrays.asList(toUserIdsStr.split(","));
        // 获取发件人信息
        ImClient sender = imClientService.getCacheImClient(application.getId(), senderClientId);
        if (sender == null) {
            throw new BusinessException("id为 " + senderClientId + " 的发件人不存在");
        }
        Map<Long, ImConversation> conversationMapGroupById = new HashMap<>();
        Map<Long, List<ImConversationMembers>> conMembersMapGroupByConId = new HashMap<>();

        if (param.getConversationExt() == null) {
            ConversationExtParam conversationExt = new ConversationExtParam();
            conversationExt.setChatType(ChatTypeEnum.SYSTEM);
            param.setConversationExt(conversationExt);
        } else {
            param.getConversationExt().setChatType(ChatTypeEnum.SYSTEM);
        }
        beforePublishPrivateMessage(application, sender, toUserIdList, param.getConversationExt(),
                conversationMapGroupById, conMembersMapGroupByConId);

        // 开始发送消息
        conversationMapGroupById.forEach((conversationId, conversation) -> {
            List<ImConversationMembers> membersList = conMembersMapGroupByConId.getOrDefault(conversationId,
                    Collections.emptyList());
            // 组装消息
            ImMessage message = assembleImMessage(application.getId(), sender, conversation.getId(), param.getMessageType(),
                    false, param.getContent());
            // 持久化
            this.save(message);
            // 拼装发送消息体
            ImMessageOnlineSend imMessageOnlineSend = assembleImMessageOnlineSend(message, sender, application.getId());

            // 发送消息
            for (ImConversationMembers members : membersList) {
                if (members.getClientId().equals(senderClientId)) {
                    // 系统消息不给自己发
                    continue;
                }

                // 入库 保存收件箱
                saveImInbox(application, conversation.getId(), imMessageOnlineSend.getMsgId(),
                        members, SnowflakeUtil.getId());

                // 在线用户直接发消息
                sendMsgForOnline(members.getFkClientId(), imMessageOnlineSend);

                // 离线消息推送
                pushMsgToOfflineMembers(application, members, param.getPushContent(), param.getPushExt());
            }

        });
        return true;
    }

    /**
     *
     * @Author luozh
     * @Date 2022年05月25日 10:48:58
     * @param application
     * @param sender
     * @param toUserIdList
     * @param conversationMapGroupById
     * @param conMembersMapGroupByConId
     * @Return
     */
    private void beforePublishPrivateMessage(ImApplication application,
                                             ImClient sender,
                                             List<String> toUserIdList,
                                             Map<Long, ImConversation> conversationMapGroupById,
                                             Map<Long, List<ImConversationMembers>> conMembersMapGroupByConId) {
        this.beforePublishPrivateMessage(application, sender, toUserIdList, null, conversationMapGroupById, conMembersMapGroupByConId);
    }

    /**
     * 私聊消息发送前
     * @Author luozh
     * @Date 2022年05月07日 11:28:59
     * @param application 应用
     * @param sender 发送者
     * @param toUserIdList 接收人用户id
     * @param conversationExt 会话拓展信息，该参数非必填，如果传入，则会话名称和
     * @param conversationMapGroupById
     * @param conMembersMapGroupByConId
     * @Return
     */
    private void beforePublishPrivateMessage(ImApplication application,
                                             ImClient sender,
                                             List<String> toUserIdList,
                                             ConversationExtParam conversationExt,
                                             Map<Long, ImConversation> conversationMapGroupById,
                                             Map<Long, List<ImConversationMembers>> conMembersMapGroupByConId) {

        // 排除发件人自己
        toUserIdList = toUserIdList.stream().filter(userId -> !sender.getClientId().equals(userId)).collect(Collectors.toList());

        Long appId = application.getId();
        // 获取收件人信息
        List<ImClient> receiverList = imClientService.list(Wrappers.<ImClient>lambdaQuery().eq(ImClient::getFkAppid,
                appId).in(ImClient::getClientId, toUserIdList));
        Map<Long, ImClient> clientMapById = receiverList.stream().collect(Collectors.toMap(ImClient::getId, Function.identity()));
        if (receiverList.isEmpty()) {
            throw new BusinessException("收件人列表为空");
        }


        // 查找conversation列表
        Long senderImClientId = sender.getId();
        Set<Long> receiverImClientIds = receiverList.stream().map(ImClient::getId).collect(Collectors.toSet());
        List<ImConversation> conversationList =
                imConversationService.getConversationBySenderAndReceivers(senderImClientId, receiverImClientIds);

        // 根据收件人imClientId对会话/会话成员进行分组
        Set<Long> notExistConversationReceiverClientIds = new HashSet<>(receiverImClientIds);


        if (!conversationList.isEmpty()) {
            conversationMapGroupById.putAll(conversationList.stream().collect(Collectors.toMap(ImConversation::getId,
                    Function.identity(), (v1, v2) -> v1)));

            // 查找会话成员
            List<Long> conversationIdList = conversationList.stream().map(ImConversation::getId).collect(Collectors.toList());
            List<ImConversationMembers> conversationMembersList =
                    imConversationMembersService.list(Wrappers.<ImConversationMembers>lambdaQuery().in(ImConversationMembers::getFkConversationId, conversationIdList));
            conMembersMapGroupByConId.putAll(conversationMembersList.stream().collect(Collectors.groupingBy(ImConversationMembers::getFkConversationId)));
            Set<Long> hasExistConversationReceiverImClientIds =
                    conversationMembersList.stream()
                            .map(ImConversationMembers::getFkClientId)
                            .filter(fkClientId -> !fkClientId.equals(senderImClientId)).collect(Collectors.toSet());


            notExistConversationReceiverClientIds.removeAll(hasExistConversationReceiverImClientIds);
        }

        // 不存在会话的先创建会话
        for (Long receiverImClientId : notExistConversationReceiverClientIds) {
            ImClient receiver = clientMapById.get(receiverImClientId);
            List<ImClient> members = new ArrayList<>();
            members.add(receiver);
            // 创建会话
            ServerImConversationCreate conversationCreate = new ServerImConversationCreate();
            conversationCreate.setName(sender.getNickname());
            conversationCreate.setAttributes(null);
            conversationCreate.setCreator(sender);
            conversationCreate.setMembers(members);
            conversationCreate.setChatType(ChatTypeEnum.SINGLE);
            conversationCreate.setApplication(application);

            // 会话拓展信息不为空
            if (conversationExt != null) {
                if (StringUtils.isNotBlank(conversationExt.getName())) {
                    conversationCreate.setName(conversationExt.getName());
                }
                if (conversationExt.getChatType() != null) {
                    conversationCreate.setChatType(conversationExt.getChatType());
                }
                if (conversationExt.getAttributes() != null) {
                    conversationCreate.setAttributes(conversationExt.getAttributes());
                }
            }

            ImConversation conversation = imConversationService.serverCreateImConversation(conversationCreate);

            if (conversation == null) {
                continue;
            }

            // 构建ImConversationMembers(只保存必要信息)
            ImConversationMembers receiverMember = new ImConversationMembers();
            receiverMember.setClientId(receiver.getClientId());
            receiverMember.setFkClientId(receiver.getId());
            receiverMember.setFkConversationId(conversation.getId());

            ImConversationMembers creatorMember = new ImConversationMembers();
            creatorMember.setClientId(sender.getClientId());
            creatorMember.setFkClientId(sender.getId());
            creatorMember.setFkConversationId(conversation.getId());

            List<ImConversationMembers> conversationMembers = new ArrayList<>();
            conversationMembers.add(receiverMember);
            conversationMembers.add(creatorMember);

            conversationMapGroupById.put(conversation.getId(), conversation);
            conMembersMapGroupByConId.put(conversation.getId(), conversationMembers);
        }

    }

    /**
     * 拼装发送消息体
     * @param message
     * @param imClientSender
     * @param appId
     * @return
     */
    private ImMessageOnlineSend assembleImMessageOnlineSend(ImMessage message, ImClient imClientSender, Long appId) {
        // 封装响应的实体
        ImMessageOnlineSend imMessageOnlineSend = new ImMessageOnlineSend();
        imMessageOnlineSend.setMsgId(message.getId());
        imMessageOnlineSend.setSender(imClientSender.getClientId());
        Map<String, Object> content = JSONObject.parseObject(message.getContent(), Map.class);
        //action的属性无需要返回
        content.remove(BaseRequest.ACTION);
        imMessageOnlineSend.setContent(content);
        imMessageOnlineSend.setConversationId(message.getFkConversationId());
        imMessageOnlineSend.setCreateTime(message.getCreateTime());
        imMessageOnlineSend.setWithdrawTime(message.getWithdrawTime());
        imMessageOnlineSend.setWithdraw(message.getWithdraw());
        imMessageOnlineSend.setEvent(message.getEvent());
        imMessageOnlineSend.setSystemFlag(message.getSystemFlag());
        imMessageOnlineSend.setType(message.getMsgType());
        imMessageOnlineSend.setAt(message.getAt());
        return imMessageOnlineSend;
    }

    /**
     * 组装imMessage
     * @Author luozh
     * @Date 2022年05月05日 06:17:00
     * @param appId 发送群聊消息入参
     * @param sender 发送群聊消息入参
     * @param toConversationId 发送群聊消息入参
     * @param messageType 发送群聊消息入参
     * @param content 发送群聊消息入参
     * @Return ImMessage
     */
    private ImMessage assembleImMessage(Long appId, ImClient sender, Long toConversationId, String messageType,
                                        Boolean event, String content) {
        long messageId = SnowflakeUtil.getId();

        ImMessage imMessage = new ImMessage();
        imMessage.setContent(content);

        imMessage.setId(messageId);
        imMessage.setCreateTime(new Date());
        imMessage.setFkAppid(appId);
        imMessage.setSender(sender.getId());
        imMessage.setWithdraw(false);
        imMessage.setEvent(event);
        imMessage.setSystemFlag(false);
        imMessage.setSendStatus(2);
        imMessage.setMsgType(Integer.valueOf(messageType));
        imMessage.setAt("");
        imMessage.setFkConversationId(toConversationId);
        return imMessage;
    }

    /**
     * 入库 保存收件箱
     *
     * @param imApplication
     * @param toConversationId
     * @param messageId
     * @param conversationMembers
     * @param imInboxId
     */
    private void saveImInbox(ImApplication imApplication, Long toConversationId, long messageId, ImConversationMembers conversationMembers, long imInboxId) {
        ImInbox imInbox = new ImInbox();
        imInbox.setId(imInboxId);
        imInbox.setCreateTime(new Date());
        imInbox.setFkAppid(imApplication.getId());
        imInbox.setReceiver(conversationMembers.getFkClientId());
        imInbox.setFkMsgId(messageId);
        imInbox.setReadMsgStatus(0);
        imInbox.setReceiverMsgStatus(0);
        imInbox.setFkConversationId(toConversationId);
        imInboxService.save(imInbox);
    }

    /**
     * 发送消息给在线客户
     * @param receiverClientId
     * @param imMessageOnlineSend
     */
    private void sendMsgForOnline(Long receiverClientId, ImMessageOnlineSend imMessageOnlineSend) {
        //  封装要推给接收方的消息
        WsResponse<ImMessageOnlineSend> responseModel = new WsResponse<>();
        responseModel.setCmd(WsResponseCmdEnum.ONLINE_MSG.getCmdCode());
        ApiResult<Boolean> result = ApiResult.result(ApiCode.SUCCESS);
        responseModel.setCode(result.getCode());
        responseModel.setMsg(result.getMessage());
        responseModel.setData(imMessageOnlineSend);
        responseModel.setReqId(null);
        channelSender.sendMsg(responseModel, receiverClientId);
    }

    /**
     * 推送消息至离线成员
     * @Author luozh
     * @Date 2022年05月06日 05:23:58
     * @param application
     * @param members
     * @param pushContent
     * @param pushExt
     * @Return
     */
    private void pushMsgToOfflineMembers(ImApplication application, ImConversationMembers members,
                                         String pushContent, PushExtParam pushExt) {
        // 不是静默推送
        PushVO pushVO = new PushVO();
        pushVO.setData(JSONObject.parseObject(pushContent, HashMap.class));
        if (pushExt != null) {
            pushVO.setTitle(pushExt.getTitle());
        }
        ImClient receiver = new ImClient();
        receiver.setId(members.getFkClientId());
        receiver.setClientId(members.getClientId());
        PushDTO pushDTO = mqSender.buildPushDto(pushVO, receiver.getId(), receiver.getClientId(), application);
        if (pushDTO != null) {
            mqSender.orderSend(MqConstant.Topic.IM_ORDER_MSG_TOPIC, MqConstant.Tag.IM_ORDER_MSG_TAG, pushDTO);
        }
    }


}
