package cn.axzo.msg.center.inside.notices.service.impl;

import cn.axzo.basics.common.constant.enums.TableIsDeleteEnum;
import cn.axzo.basics.common.util.AssertUtil;
import cn.axzo.core.domain.PageResult;
import cn.axzo.msg.center.api.enums.MsgRecordTerminalTypeEnum;
import cn.axzo.msg.center.api.enums.MsgStateEnum;
import cn.axzo.msg.center.api.enums.MsgTypeEnum;
import cn.axzo.msg.center.api.enums.ReceiveTypeEnum;
import cn.axzo.msg.center.api.mq.SendMessageRecordMessage;
import cn.axzo.msg.center.api.request.CmsMsgQueryReq;
import cn.axzo.msg.center.api.request.CmsReadMsgReq;
import cn.axzo.msg.center.api.request.GeneralMessageReq;
import cn.axzo.msg.center.api.request.MessageRecordDeleteReq;
import cn.axzo.msg.center.api.request.MessageRecordReq;
import cn.axzo.msg.center.api.request.MessageStatisticsReq;
import cn.axzo.msg.center.api.request.MsgBody4Guest;
import cn.axzo.msg.center.api.request.MsgReturnParamRes;
import cn.axzo.msg.center.api.request.QueryMessageRecordReq;
import cn.axzo.msg.center.api.response.MessageNewRes;
import cn.axzo.msg.center.api.response.MessageRecordRes;
import cn.axzo.msg.center.api.response.MessageTotalRes;
import cn.axzo.msg.center.api.response.MyPage;
import cn.axzo.msg.center.common.enums.IdentityType;
import cn.axzo.msg.center.common.exception.ServiceException;
import cn.axzo.msg.center.common.model.ProfileIdRepair;
import cn.axzo.msg.center.common.utils.BeanConvertUtils;
import cn.axzo.msg.center.common.utils.PlaceholderResolver;
import cn.axzo.msg.center.common.utils.SpringUtils;
import cn.axzo.msg.center.dal.MessageModuleDao;
import cn.axzo.msg.center.dal.MessageRecordDao;
import cn.axzo.msg.center.dal.MessageRelationDao;
import cn.axzo.msg.center.dal.MessageRouterDao;
import cn.axzo.msg.center.dal.MessageTemplateDao;
import cn.axzo.msg.center.domain.dto.CmsMsgQueryReqDTO;
import cn.axzo.msg.center.domain.dto.MessageNewResDTO;
import cn.axzo.msg.center.domain.dto.MsgStatisticsDTO;
import cn.axzo.msg.center.domain.dto.UpdateReadDTO;
import cn.axzo.msg.center.domain.entity.MessageModule;
import cn.axzo.msg.center.domain.entity.MessageRecord;
import cn.axzo.msg.center.domain.entity.MessageRelation;
import cn.axzo.msg.center.domain.entity.MessageRouter;
import cn.axzo.msg.center.domain.entity.MessageTemplate;
import cn.axzo.msg.center.domain.enums.ModuleBizTypeEnum;
import cn.axzo.msg.center.domain.enums.MsgRouteTypeEnum;
import cn.axzo.msg.center.domain.enums.NativeTypeEnum;
import cn.axzo.msg.center.domain.enums.UserTypeEnum;
import cn.axzo.msg.center.domain.enums.YesNoEnum;
import cn.axzo.msg.center.domain.request.InsideCmsReadMsgReq;
import cn.axzo.msg.center.inside.notices.event.SendMessageEvent;
import cn.axzo.msg.center.inside.notices.service.MessageRecordService;
import cn.axzo.msg.center.message.service.GeneralMessageMapperService;
import cn.axzo.msg.center.mq.MqMessageRecord;
import cn.axzo.msg.center.mq.MqProducer;
import cn.axzo.msg.center.service.dto.IdentifyAndReceiveType;
import cn.axzo.msg.center.service.dto.IdentityDTO;
import cn.axzo.msg.center.service.enums.MqMessageType;
import cn.axzo.msg.center.utils.PersonIdentityUtil;
import cn.azxo.framework.common.utils.LogUtil;
import cn.azxo.framework.common.utils.LogUtil.ErrorLevel;
import cn.azxo.framework.common.utils.LogUtil.ErrorType;
import cn.azxo.framework.common.utils.StringUtils;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.map.MapUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.text.StrSubstitutor;
import org.apache.logging.log4j.util.Strings;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.stream.Collectors;


/**
 * 消息记录(MessageRecord)表服务实现类
 *
 * @author zhangran
 * @since 2022-03-28 14:59:16
 */
@Service
@Slf4j
public class MessageRecordServiceImpl implements MessageRecordService {

  /**
   * 一次更新数据操作所允许的最大行数，特指通过id去更新数据的操作
   */
  private static final int MAX_SIZE_OF_BATCH_UPDATE_ONCE = 500;

  @Value("${msg.center.server.partition-size:5000}")
  private Integer partitionSize;
  @Value("${msg.center.server.push-athena:true}")
  private Boolean pushAthena;

  @Resource
  private MessageRecordDao messageRecordDao;
  @Resource
  private MessageRelationDao messageRelationDao;
  @Resource
  private MessageTemplateDao messageTemplateDao;
  @Resource
  private MessageModuleDao messageModuleDao;
  @Resource
  private MessageRouterDao messageRouterDao;
  @Resource
  private MqProducer mqProducer;
  /*@Resource
  private IdentityProfileService identityProfileService;*/

  @Resource
  private GeneralMessageMapperService generalMessageMapperService;

  /**
   * 新增推送消息接口
   *
   * @param message
   * @return 消息唯一主键
   */
  @Override
  public List<MsgReturnParamRes> pushMsg(GeneralMessageReq message) {
    log.info("pushMsg message:{}", JSON.toJSONString(message));
    MessageRecord basic = BeanConvertUtils.copyBean(message, MessageRecord.class);
    basic.setExtra(org.springframework.util.StringUtils.hasLength(message.getExt())
        ? message.getExt() : "{}");
    basic.setRouterParams(Objects.nonNull(message.getRouterParams())
        ? JSON.toJSONString(message.getRouterParams()) : "{}");

    //如果moduleId和templateCode不为空，通过moduleId和templateCode优先查询模板，否则使用relationId查询
    MessageRelation relation = null;
    if(message.getModuleId() != null && org.apache.commons.lang3.StringUtils.isNotEmpty(message.getTemplateCode())){
      relation = messageRelationDao.getTemplateByCode(message.getModuleId(),message.getTemplateCode());
    }else {
      // 消息模板参数替换
      relation = messageRelationDao.getById(message.getRelationId());
    }
    AssertUtil.notNull(relation, "消息模板不存在");

    MessageTemplate messageTemplate = messageTemplateDao.getById(relation.getTemplateId());
    AssertUtil.notNull(messageTemplate, "消息模板不存在");

    StrSubstitutor strSubstitutor = new StrSubstitutor(message.getMsgParams());
    String title = strSubstitutor.replace(messageTemplate.getTitle());
    String content = strSubstitutor.replace(messageTemplate.getContent());

    // 消息模块
    MessageModule messageModule = messageModuleDao.getById(relation.getModuleId());
    AssertUtil.notNull(messageModule, "消息模块不存在");

    basic.setModuleId(messageModule.getId());
    basic.setTitle(title);
    basic.setContent(content);
    basic.setReceiveType(message.getReceiveType());
    basic.setState(MsgStateEnum.UNSENT);

    if (message.getTerminalType() == null) {
      basic.setTerminalType(MsgRecordTerminalTypeEnum.UNKNOWN);
    }

    List<MessageRecord> pushMessages = new ArrayList<>();
    Lists.partition(Lists.newArrayList(message.getToId()), partitionSize).forEach(toIds -> {
      Map<Long, MessageRecord> toIdRecordMap = Maps.newHashMap();
      List<MessageRecord> messageRecords = saveBatch(basic, toIds, message.getToldIdPersonIdMap(), toIdRecordMap);
      try {
        generalMessageMapperService.asyncBatchSendMessage(message, toIdRecordMap);
      } catch (Exception e) {
        log.warn("asyncBatchSendMessage push error, param= [{}]", JSON.toJSONString(message), e);
      }
      // 不影响原有消息通道
      pushMessages.addAll(messageRecords);
    });
    SendMessageRecordMessage mqMessage = new SendMessageRecordMessage();
    mqMessage.setSendRequest(message);
    mqProducer.send(MqMessageRecord
            .builder(MqMessageType.OLD_MSG_SEND, mqMessage)
            .build());
    if(pushAthena) {
      asyncPushAthena(message, messageTemplate.getAudioFileName(), messageModule.getModuleName(), pushMessages);
    }

    return pushMessages.stream().map(item -> {
      MsgReturnParamRes res = new MsgReturnParamRes();
      res.setToId(0L == item.getToId() ? item.getPersonId(): item.getToId());
      res.setRecordId(item.getId());
      res.setReceiveType(null == item.getReceiveType() ? 0 : item.getReceiveType().getCode());
      res.setPersonId(null == item.getPersonId() ? 0 : item.getPersonId());
      return res;
    }).collect(Collectors.toList());
  }

  @Async
  public void asyncPushAthena(GeneralMessageReq message, String audioFileName, String moduleName,
                              List<MessageRecord> pushMessages) {
    sendMessage(message, audioFileName, moduleName, pushMessages);
  }


  @Transactional(rollbackFor = Exception.class, propagation = Propagation.REQUIRES_NEW)
  public List<MessageRecord> saveBatch(MessageRecord basic, List<Long> toIds, Map<Long, Long> toldIdPersonIdMap,
                                       Map<Long, MessageRecord> toIdRecordMap) {
    if (CollectionUtils.isEmpty(toIds)) {
      return Collections.emptyList();
    }
    List<MessageRecord> pushMessages = new ArrayList<>();
    if(basic.getReceiveType()==null){
      basic.setReceiveType(ReceiveTypeEnum.NORMAL);
    }
    if (ReceiveTypeEnum.NOT_IDENTITY.getCode().equals(basic.getReceiveType().getCode())) {
      toIds.forEach(i -> {
        MessageRecord messageRecord = BeanConvertUtils.copyBean(basic, MessageRecord.class);
        messageRecord.setToId(0L);
        messageRecord.setPersonId(i);
        pushMessages.add(messageRecord);
        toIdRecordMap.put(i, messageRecord);
      });
    } else {
      toIds.forEach(i -> {
        MessageRecord messageRecord = BeanConvertUtils.copyBean(basic, MessageRecord.class);
        messageRecord.setToId(i);
        if (MapUtil.isEmpty(toldIdPersonIdMap) || toldIdPersonIdMap.get(i) == null) {
          messageRecord.setPersonId(0L);
        } else {
          messageRecord.setPersonId(toldIdPersonIdMap.get(i));
        }
        pushMessages.add(messageRecord);
        toIdRecordMap.put(i, messageRecord);
      });
    }
    messageRecordDao.saveBatch(pushMessages);
    return pushMessages;
  }

  public static IdentityType ConvertIdentity(ReceiveTypeEnum receiveTypeEnum) {
    IdentityType identityType;
    switch (receiveTypeEnum) {
      case CM_LEADER:
        identityType = IdentityType.WORKER_LEADER;
        break;
      case  CM_WORKER:
        identityType = IdentityType.WORKER;
        break;
      case CMP_USER:
        identityType = IdentityType.PRACTITIONER;
        break;
      default:
        identityType = null;
        break;
    }
    return identityType;
  }

  /**
   * 推送消息
   *
   * @param pushMessages  推送消息
   * @param audioFileName 语音文件
   */
  private void sendMessage(GeneralMessageReq message, String audioFileName, String moduleName,
      List<MessageRecord> pushMessages) {
    Long relationId = message.getRelationId();
    Map<String, String> routerParams = message.getRouterParams();
    Map<Long,Integer> personAppVersionCode = Optional.ofNullable(message.getPersonAppVersionCode()).orElse(new HashMap<>());

    ArrayList<MsgRouteTypeEnum> msgRouteTypeEnums = Lists.newArrayList(MsgRouteTypeEnum.ANDROID,
        MsgRouteTypeEnum.IOS, MsgRouteTypeEnum.MINI_PROGRAM, MsgRouteTypeEnum.WEBVIEW);
    List<MessageRouter> messageRouters = messageRouterDao.getByRelationId(relationId,
        msgRouteTypeEnums);

    MsgBody4Guest imMessage = new MsgBody4Guest();
    imMessage.setTy(0);
    imMessage.setF("0");
    imMessage.setNickName(moduleName);
    imMessage.setAppClient(message.getAppClient());
    pushMessages.forEach(item -> {
      String toId = buildToId(item.getToId(), item.getPersonId(), item.getReceiveType(), item.getBizId());
      Integer versionCode = personAppVersionCode.getOrDefault(item.getPersonId(), 0);
      log.info("发给用户：{}, versionCode:{}, personId: {},ReceiveType:{}",item.getToId(), versionCode,item.getPersonId(), item.getReceiveType());

      if (Objects.nonNull(versionCode) && versionCode >= 330 && ReceiveTypeEnum.CM_LEADER.equals(item.getReceiveType())) {
        log.info("班组长转到cmp,发给用户：{}, versionCode:{}, personId: {}",item.getToId(), versionCode,item.getPersonId());
        imMessage.setAppClient("cmp");
      }
      imMessage.setT(toId);
      imMessage.setM(item.getContent());

      JSONObject jsonObject = constructImMessage(item, audioFileName, routerParams, messageRouters);
      log.info("sendMessage ImMessage param= [{}]", jsonObject.toJSONString());
      imMessage.setM2(jsonObject.toJSONString());
      imMessage.setM3(String.valueOf(item.getPersonId()));

      // 发送推送消息事件
      SpringUtils.publishEvent(new SendMessageEvent(item.getId(), imMessage));
    });
  }

  private String buildToId(Long itemToId, Long personId, ReceiveTypeEnum receiveType, Long bizId) {
    String toId = "";
    switch (receiveType) {
      case CM_LEADER:
        toId = ReceiveTypeEnum.CM_LEADER.getMessage() + ProfileIdRepair.workerLeaderNewId2Old(
            itemToId);
        break;
      case CM_WORKER:
        toId = ReceiveTypeEnum.CM_WORKER.getMessage() + ProfileIdRepair.workerNewId2OldId(itemToId);
        break;
      case CMP_USER:
        toId = ReceiveTypeEnum.CMP_USER.getMessage() + ProfileIdRepair.entUserNew2OldId(itemToId);
        break;
      case NOT_IDENTITY:
        toId = ReceiveTypeEnum.NOT_IDENTITY.getMessage() + personId;
        break;
      case NORMAL:
        toId = ReceiveTypeEnum.NORMAL.getMessage() + personId;
        break;
      default:
        LogUtil.error(ErrorLevel.P0, ErrorType.ERROR_BUSINESS, "消息推送用户类型错误 bizId", bizId);
        throw new ServiceException("消息推送失败");
    }
    return toId;
  }

  private JSONObject constructImMessage(MessageRecord message, String audioFileName,
      Map routerParams, List<MessageRouter> messageRouters) {

    JSONObject extra = new JSONObject();

    if (messageRouters.isEmpty()) {
      extra.put("ar", "");
      extra.put("ir", "");
    } else {

      // 业务上不会跳转微信小程序
      messageRouters.forEach(item -> {
        PlaceholderResolver pr = PlaceholderResolver.getDefaultResolver();
        String url = pr.resolveByMap(item.getRouterUrl(), routerParams);
        extra.put("rt", NativeTypeEnum.getByCode(item.getRouterType().getCode()).getMessage());
        switch (item.getRouterType()) {
          case IOS:
            extra.put("ir", url);
            break;
          case ANDROID:
            extra.put("ar", url);
            break;
          case WEBVIEW:
          case MINI_PROGRAM:
            // 小程序
            extra.put("ar", url);
            extra.put("ir", url);
            break;
          default:
        }
      });
    }
    extra.put("terminalId", message.getTerminalId());
    extra.put("msgId", message.getId());
    extra.put("p", message.getRouterParams());
    extra.put("t", message.getTitle());
    extra.put("extra", message.getExtra());
    // 临时处理发送类型，如果消息type为1时，需要前端清空teamId,并且重新拉取数据
    Integer type = 0;
    if (StringUtils.isNotBlank(message.getExtra())) {
      JSONObject jsonObject = JSON.parseObject(message.getExtra());
      Integer exType = jsonObject.getInteger("type");
      if (type != null) {
        type = exType;
      }
    }
    extra.put("type", type);

    if (StringUtils.isNotBlank(audioFileName)) {
      extra.put("audio", audioFileName);
    }
    return extra;
  }


  /**
   * 变更执行消息的状态
   *
   * @param msgId
   * @param state
   * @return
   */
  @Override
  public Boolean changeMessageState(List<Long> msgId, MsgStateEnum state) {
    List<MessageRecord> messageRecords = messageRecordDao.listByIds(msgId);
    messageRecords = messageRecords.stream().map(item -> {
      item.setState(state);
      return item;
    }).collect(Collectors.toList());
    return messageRecordDao.updateBatchById(messageRecords);
  }

  @Override
  public Boolean changeMessageState(Long toId, Long relationId, Long bizId, MsgStateEnum state) {
    List<MessageRecord> records = messageRecordDao.lambdaQuery().eq(MessageRecord::getToId, toId)
        .eq(MessageRecord::getRelationId, relationId)
        .eq(MessageRecord::getBizId, bizId)
        .in(MessageRecord::getState, Arrays.asList(2, 3))
        .list();
    if (CollectionUtils.isEmpty(records)) {
      log.info("未找到 toId={}, relationId={}, bizId={} 的数据, 无法修改状态!", toId, relationId, bizId);
      return false;
    }
    if (records.size() > 1) {
      throw new IllegalStateException("存在多条相同 toId,relationId,bizId 的消息记录");
    }
    MessageRecord record = records.get(0);
    log.info("消息主键为: {} 的状态被更新为: {}({})", record.getId(),
        record.getState().getMessage(),
        record.getState().getCode());
    record.setState(state);
    return messageRecordDao.updateById(record);
  }

  /**
   * 消息Id+当前登录类型进行消息读取
   *
   * @param req
   */
  @Override
  public void updateReadMsgStatus(CmsReadMsgReq req) {
    InsideCmsReadMsgReq request= BeanConvertUtils.copyBean(req,InsideCmsReadMsgReq.class);
    if (request.getMsgId() == -1L) {
      // 已读全部
      messageRecordDao.readAllMsg(request, req.getPersonId(), req.getIdentityId());
    } else {
      messageRecordDao.readMsg(request, req.getPersonId(), req.getIdentityId());
    }
    SendMessageRecordMessage mqMessage = new SendMessageRecordMessage();
    mqMessage.setCmsReadRequest(req);
    mqProducer.send(MqMessageRecord
            .builder(MqMessageType.OLD_MSG_SEND, mqMessage)
            .build());
  }

  /**
   * 分页查询-不含路由信息
   */
  @Override
  public PageResult<MessageNewRes> pageMsgInfo(CmsMsgQueryReq req, Long personId, List<IdentityDTO> identities) {
    IPage page = req.toPage();
    List<Integer> states = buildStates(req);
    CmsMsgQueryReqDTO param=BeanConvertUtils.copyBean(req, CmsMsgQueryReqDTO.class);
    List<IdentifyAndReceiveType> convertedIdentifies = identities.stream()
            .map(i -> new IdentifyAndReceiveType(i, PersonIdentityUtil.toReceiveType(i.getType())))
            .collect(Collectors.toList());
    IPage<MessageNewResDTO> iPage = messageRecordDao.pageMsgInfo(param, page, personId, states, convertedIdentifies);
    return convertPageMessageNewRes(iPage);
  }

  /**
   *  转换返回值
   * @param iPage
   * @return
   */
  private PageResult<MessageNewRes> convertPageMessageNewRes(IPage<MessageNewResDTO> iPage){
    if(iPage == null || CollectionUtils.isEmpty(iPage.getRecords())){
      return new PageResult<>(new ArrayList<>(),0L);
    }
    List<MessageNewRes> records=new ArrayList<>(iPage.getRecords().size());
    for (MessageNewResDTO record : iPage.getRecords()) {
      MessageNewRes messageNewRes = BeanConvertUtils.copyBean(record, MessageNewRes.class);
      records.add(messageNewRes);
    }
    return new PageResult<>(records, iPage.getTotal());
  }

  @Override
  public PageResult<MessageNewRes> pageMsgInfo4WechatAndTrade(CmsMsgQueryReq req, Long personId) {
    List<Long> moduleIds = messageModuleDao.lambdaQuery()
        .eq(MessageModule::getIsDelete, TableIsDeleteEnum.NORMAL.value)
        .eq(MessageModule::getBizType, ModuleBizTypeEnum.TRADE.getCode())
        .select(MessageModule::getId)
        .list().stream().map(MessageModule::getId).collect(Collectors.toList());
    ArrayList<Integer> states = buildStates(req);
    CmsMsgQueryReqDTO param= BeanConvertUtils.copyBean(req, CmsMsgQueryReqDTO.class);
    IPage<MessageNewResDTO> iPage = messageRecordDao.pageMsgInfo4WechatAndTrade(param, moduleIds, personId,
        states);
    return convertPageMessageNewRes(iPage);
  }

  private static ArrayList<Integer> buildStates(CmsMsgQueryReq req) {
    // 普通消息、待办消息未读及未处理消息默认初始状态 2，3
    ArrayList<Integer> states = Lists.newArrayList(2, 3);
    if (req.getMsgType().equals(MsgTypeEnum.GENERAL_MESSAGE.getCode())) {
      if (req.getMsgStatus() == 0) {
        // 全部
        states.add(4);
      }
    } else if (req.getMsgType().equals(MsgTypeEnum.PENDING_MESSAGE.getCode())) {
      if (req.getMsgStatus() == 1) {
        // 未处理
        states.add(4);
      } else if (req.getMsgStatus() == 2) {
        // 已完成
        states = Lists.newArrayList(5);
      }else if (req.getMsgStatus() == 3) {
        //查询全部
        states.addAll( Lists.newArrayList(4, 5));
      }
    }
    return states;
  }


  /**
   * 用户Id + 用户类型进行消息统计
   *
   * @param personId
   * @param identityId
   * @param userTypeEnum
   * @return
   */
  @Override
  public MessageTotalRes statisticsMsg4Total(Long personId, Long identityId, UserTypeEnum userTypeEnum) {
    ReceiveTypeEnum typeEnum = ReceiveTypeEnum.getByMessage(userTypeEnum.getName());
    if (typeEnum == null) {
      throw new ServiceException("未获取到用户类型信息");
    }
    MessageTotalRes res = new MessageTotalRes();

    List<MsgStatisticsDTO> constructionList = messageRecordDao.statisticsMsg4Construction(
        identityId,
        typeEnum.getCode()
        );
    calcCount(constructionList, res);

//    List<MsgStatisticsDTO> tradeList = messageRecordDao.statisticsMsg4Trade(personId);
//    calcCount(tradeList, res);
    return res;
  }

  @Override
  public MessageTotalRes statisticsMsg4Construction(Long identityId, UserTypeEnum userTypeEnum) {
    List<MsgStatisticsDTO> res = messageRecordDao.statisticsMsg4Construction(identityId,
        userTypeEnum.getValue());
    return calcCount(res, null);
  }

  private MessageTotalRes calcCount(List<MsgStatisticsDTO> list, MessageTotalRes res) {
    if(Objects.isNull(res)) {
      res = new MessageTotalRes();
    }
    for (MsgStatisticsDTO item : list) {
      if (item.getType().equals(MsgTypeEnum.GENERAL_MESSAGE.getCode())) {
        res.setMsgTotal(res.getMsgTotal() + item.getNum());
      }
      if (item.getType().equals(MsgTypeEnum.PENDING_MESSAGE.getCode())) {
        res.setUpcomingTotal(res.getUpcomingTotal() + item.getNum());
      }
    }
    res.setAllTotal(res.getMsgTotal() + res.getUpcomingTotal());
    return res;
  }

  @Override
  public MessageTotalRes statisticsMsg4Trade(Long personId) {
    List<MsgStatisticsDTO> res = messageRecordDao.statisticsMsg4Trade(personId);
    return calcCount(res, null);
  }

  @Override
  public Boolean changeHasRead(UpdateReadDTO request) {
    return messageRecordDao.updateHasRead(request) > 0;
  }

  private List<MessageRouter> saveRouterList(Long relationId, String typeId,
      HashMap<Long, List<String>> hashMap) {

    List<String> stringList = hashMap.getOrDefault(Long.parseLong(typeId), Lists.newArrayList());
    if (stringList.isEmpty()) {
      return Lists.newArrayList();
    }

    return stringList.stream().map(item -> {
      MessageRouter messageRouter = new MessageRouter();
      messageRouter.setRelationId(relationId);
      if (item.startsWith("AX")) {
        messageRouter.setRouterType(MsgRouteTypeEnum.IOS);
        messageRouter.setRouterUrl(item);
      } else if (item.startsWith("cn")) {
        messageRouter.setRouterType(MsgRouteTypeEnum.ANDROID);
        messageRouter.setRouterUrl(item);
      } else {
        messageRouter.setRouterType(MsgRouteTypeEnum.MINI_PROGRAM);
        messageRouter.setRouterUrl(item);
      }
      return messageRouter;
    }).collect(Collectors.toList());


  }

  @Override
  public IPage<MessageRecord> querySendFailMsg(IPage<MessageRecord> page) {
    return messageRecordDao.lambdaQuery()
        .eq(MessageRecord::getState, MsgStateEnum.UNSENT.getCode())
//        .ne(MessageRecord::getEventSource, "")
        .page(page);
  }

  private void addList(String param, List<String> list) {
    if (!StringUtils.isBlank(param)) {
      list.add(param);
    }
  }


  @Override
  public IPage<MessageRecordRes> queryUnReadList(QueryMessageRecordReq request) {
    IPage<MessageRecordRes> resultList=new MyPage<>();

    IPage<MessageNewResDTO> newResDTOIPage = messageRecordDao.queryUnReadList(request);
    if(newResDTOIPage == null || newResDTOIPage.getTotal()==0){
      return resultList;
    }
    List<MessageRecordRes> messageList=new ArrayList<>();
    for (MessageNewResDTO record : newResDTOIPage.getRecords()) {
      MessageRecordRes item = BeanConvertUtils.copyBean(record, MessageRecordRes.class);
      messageList.add(item);
    }
    resultList.setRecords(messageList);
    return resultList;
  }

  @Override
  public IPage<MessageRecordRes> querySent(QueryMessageRecordReq request) {
    return messageRecordDao.querySent(request);
  }

  @Override
  public boolean deleteMessageRecords(MessageRecordDeleteReq request) {
    if (Objects.isNull(request)) {
      log.warn("==={}=== request param is null.", "deleteMessageRecords");
      return false;
    }
    if (CollectionUtils.isEmpty(request.getMsgIds())
            && (Objects.isNull(request.getRelationId()) || request.getRelationId() < 0)) {
      log.warn("==={}=== request param is invalid. {}", "deleteMessageRecords", request);
      return false;
    }
    if (!CollectionUtils.isEmpty(request.getMsgIds())
            && request.getMsgIds().size() > MAX_SIZE_OF_BATCH_UPDATE_ONCE) {
      log.warn("==={}=== the size of request.msgIds is exceed {}.", "deleteMessageRecords", MAX_SIZE_OF_BATCH_UPDATE_ONCE);
      return false;
    }
    return messageRecordDao.lambdaUpdate().set(MessageRecord::getIsDelete, YesNoEnum.YES.getCode())
            .in(!CollectionUtils.isEmpty(request.getMsgIds()), MessageRecord::getId, request.getMsgIds())
            .eq(Objects.nonNull(request.getRelationId()) && request.getRelationId() >= 0, MessageRecord::getRelationId, request.getRelationId())
            .eq(MessageRecord::getIsDelete, YesNoEnum.NO.getCode())
            .update();
  }

  @Override
  public Integer unReadStatistics(MessageStatisticsReq request) {
    log.info("statistics reqeust:{}，condition:{}", JSON.toJSONString(request),
            Objects.nonNull(request.getMsgType()));
    return  messageRecordDao.lambdaQuery()
            .eq(Objects.nonNull(request.getFromId()), MessageRecord::getFromId, request.getFromId())
            .eq(Objects.nonNull(request.getModuleId()), MessageRecord::getModuleId, request.getFromId())
            .eq(Objects.nonNull(request.getMsgType()), MessageRecord::getType, MsgTypeEnum.valueOf(request.getMsgType()))
            .eq(Objects.nonNull(request.getToId()), MessageRecord::getToId, request.getToId())
            .eq(Objects.nonNull(request.getTenantId()), MessageRecord::getTenantId, request.getTenantId())
            .in(MessageRecord::getState, MsgStateEnum.HAS_BEEN_SENT, MsgStateEnum.RECEIVED)
            .count();
  }

  @Override
  public IPage<MessageRecordRes> queryMessageRecordList(MessageRecordReq request) {

    IPage<MessageRecord> result = messageRecordDao.lambdaQuery()
            .eq(Objects.nonNull(request.getMsgType()), MessageRecord::getType, MsgTypeEnum.valueOf(request.getMsgType()))
            .eq(Objects.nonNull(request.getFromId()), MessageRecord::getFromId, request.getFromId())
            .eq(Objects.nonNull(request.getTenantId()), MessageRecord::getTenantId, request.getTenantId())
            .like(Strings.isNotBlank(request.getSearchContent()), MessageRecord::getContent, request.getSearchContent())
            .eq(Objects.nonNull(request.getToId()), MessageRecord::getToId, request.getToId())
            .in(Objects.nonNull(request.getMsgStatus()), MessageRecord::getState, request.getMsgStatus())
            .eq(Objects.nonNull(request.getModuleId()), MessageRecord::getModuleId, request.getModuleId())
            .orderByDesc(MessageRecord::getCreateAt)
            .page(request.toPage());

    return result.convert(r -> {
      MessageRecordRes messageRecordRes = BeanConvertUtils.copyBean(r, MessageRecordRes.class);
      messageRecordRes.setMsgId(r.getId());
      return messageRecordRes;
    });
  }

  @Override
  public void updateStateReceive(List<Long> msgIdList) {
    if (CollectionUtil.isEmpty(msgIdList)) {
      return;
    }
    messageRecordDao.lambdaUpdate()
            .set(MessageRecord::getState, MsgStateEnum.RECEIVED)
            .in(MessageRecord::getId, msgIdList).update();
  }
}
