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

import cn.axzo.msg.center.api.enums.MsgStateEnum;
import cn.axzo.msg.center.api.enums.MsgTempBizCategoryEnum;
import cn.axzo.msg.center.api.enums.MsgTypeEnum;
import cn.axzo.msg.center.api.enums.ReceiveTypeEnum;
import cn.axzo.msg.center.api.request.PendingMessagePageReq;
import cn.axzo.msg.center.api.request.PendingMessageTemporarilyStatisticReq;
import cn.axzo.msg.center.api.response.MessageRouterInfoRes;
import cn.axzo.msg.center.api.response.PendingMessageBriefRes;
import cn.axzo.msg.center.api.response.PendingMessageStatisticRes;
import cn.axzo.msg.center.api.response.PendingMessageTemporarilyStatisticRes;
import cn.axzo.msg.center.api.response.PendingMessageTemporarilyTypeRes;
import cn.axzo.msg.center.common.utils.BeanConvertUtils;
import cn.axzo.msg.center.common.utils.PlaceholderResolver;
import cn.axzo.msg.center.dal.MessageRecordDao;
import cn.axzo.msg.center.domain.dto.MsgModuleDTO;
import cn.axzo.msg.center.domain.entity.MessageModule;
import cn.axzo.msg.center.domain.entity.MessageRecord;
import cn.axzo.msg.center.domain.entity.MessageRouter;
import cn.axzo.msg.center.domain.enums.MsgRouteTypeEnum;
import cn.axzo.msg.center.domain.enums.NativeTypeEnum;
import cn.axzo.msg.center.domain.enums.YesNoEnum;
import cn.axzo.msg.center.inside.notices.config.MessageSystemConfig;
import cn.axzo.msg.center.inside.notices.config.PendingMessageBizConfig;
import cn.axzo.msg.center.inside.notices.config.PendingMessageBizConfig.MsgTempBizCategoryConfig;
import cn.axzo.msg.center.inside.notices.service.MessageModuleService;
import cn.axzo.msg.center.inside.notices.service.MessageRelationService;
import cn.axzo.msg.center.inside.notices.service.MessageRouterService;
import cn.axzo.msg.center.inside.notices.service.PendingMessageService;
import cn.axzo.msg.center.inside.notices.utils.MessageRecordHelper;
import cn.axzo.msg.center.message.service.PendingMessageNewService;
import cn.axzo.msg.center.notices.common.enums.ReturnCodeEnum;
import cn.axzo.msg.center.notices.common.exception.BizException;
import cn.azxo.framework.common.model.Page;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.google.common.collect.Lists;
import jodd.util.concurrent.ThreadFactoryBuilder;
import lombok.AllArgsConstructor;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Service;

import java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.Collection;
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.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.stream.Collectors;

@Slf4j
@Service
@RequiredArgsConstructor
public class PendingMessageServiceImpl implements PendingMessageService {

    private static final List<MsgStateEnum> SEND_SUCCESSFULLY_STATES = Lists.newArrayList(
            MsgStateEnum.HAS_BEEN_SENT,
            MsgStateEnum.RECEIVED
    );

    private static final List<MsgStateEnum> ALL_WITHOUT_UNSENT = Lists.newArrayList(
            MsgStateEnum.HAS_BEEN_SENT,
            MsgStateEnum.RECEIVED,
            MsgStateEnum.COMPLETE
    );

    private static final List<ReceiveTypeEnum> SPECIFY_IDENTIFIES = Lists.newArrayList(
            ReceiveTypeEnum.CM_WORKER,
            ReceiveTypeEnum.CM_LEADER
    );

    private final ThreadFactory statisticTaskThreadFactory = ThreadFactoryBuilder.create()
            .setDaemon(true)
            .setNameFormat("MODULE_STATISTIC_TASK_%d")
            .get();
    private final ExecutorService stateStatisticTaskThreadPool = new ThreadPoolExecutor(5, 10, 10, TimeUnit.SECONDS,
            new ArrayBlockingQueue<>(1024), statisticTaskThreadFactory);

    private final MessageRecordDao messageRecordDao;
    private final MessageSystemConfig messageSystemConfig;
    private final MessageRouterService messageRouterService;
    private final MessageModuleService messageModuleService;
    private final MessageRelationService messageRelationService;
    private final PendingMessageBizConfig pendingMessageBizConfig;
    private final PendingMessageNewService pendingMessageNewService;

    @Override
    public List<PendingMessageTemporarilyStatisticRes> temporarilyStatistic(PendingMessageTemporarilyStatisticReq request) {
        // 参数检测
        checkParam(request);
        final List<MsgTempBizCategoryConfig> categoryConfigs = pendingMessageBizConfig.tempCategoriesConfig();
        if (CollectionUtils.isEmpty(categoryConfigs)) {
            log.warn("the nacos config is lost......");
            return Collections.emptyList();
        }
        try {
            return categoryConfigs.stream()
                    // 自然排序,前端仅做渲染,展示顺序的工作由后端配置
                    .sorted()
                    .map(e -> statistic(request, e))
                    .collect(Collectors.toList());
        } catch (Exception e) {
            log.error("broke out some exception while statistic pending message.", e);
            throw new BizException(ReturnCodeEnum.SYSTEM_ERROR, "system exception.");
        }
    }

    @Override
    public PendingMessageStatisticRes temporarilyPage(PendingMessagePageReq request) {
        IPage<MessageRecord> page = request.toPage();
        // 获取数据冷热隔离的分界
        final LocalDateTime queryFrom = getQueryFromTime();
        // 消息临时分类的配置
        final List<MsgTempBizCategoryConfig> categoryConfigs = pendingMessageBizConfig.tempCategoriesConfig();
        // FIXME: 这里由于是临时接口不具有普适性，这里是链路上传参的问题的临时解决方案 {@see ReceiveTypeEnum#CM_WORKER}
        request.setIdentifyTypes(SPECIFY_IDENTIFIES);
        // FIXME: 这里前端可以不传临时的业务分类标识，若不传的话查询配置指定的几个分类的合集
        Set<Long> relationIds = fetchRelationIdFromRequestOrConfig(request, categoryConfigs);
        if (CollectionUtils.isEmpty(relationIds)) {
            log.warn("the tempBizCategory is invalid. {}", request.getTempBizCategory());
            return PendingMessageStatisticRes.builder()
                    .totalElements(0L)
                    .pageNum(page.getCurrent())
                    .pageSize(page.getSize())
                    .unCompleteCnt(0)
                    .completeCnt(0)
                    .briefResList(Collections.emptyList())
                    .build();
        }
        // 这里对前端未传入参的情况做一个兜底
        Collection<MsgStateEnum> msgStates = CollectionUtils.isEmpty(request.getStates()) ? ALL_WITHOUT_UNSENT :
                request.getStates();
        String unCompleteStateCntKey = "unCompleteCnt";
        String completeStateCntKey = "completeCnt";
        List<StateStatisticModule> modules = Lists.newArrayList(
                StateStatisticModule.build(unCompleteStateCntKey, SEND_SUCCESSFULLY_STATES),
                StateStatisticModule.build(completeStateCntKey, Collections.singletonList(MsgStateEnum.COMPLETE))
        );
        Map<String, Integer> stateStatisticMap = statistic(modules, relationIds, request, queryFrom);
        // 分页查询message_record的数据
        page = page(request.getPersonId(), request.getIdentifyId(), request.getIdentifyTypes(), relationIds,
                msgStates, queryFrom, page);
        if (CollectionUtils.isEmpty(page.getRecords())) {
            // 列表为空的场景
            return PendingMessageStatisticRes.builder()
                    .totalElements(page.getTotal())
                    .pageNum(page.getCurrent())
                    .pageSize(page.getSize())
                    .unCompleteCnt(stateStatisticMap.get(unCompleteStateCntKey))
                    .completeCnt(stateStatisticMap.get(completeStateCntKey))
                    .briefResList(Collections.emptyList())
                    .build();
        }
        // 查询消息对应的路由集
        Map<Long, MessageRouterInfoRes> routerMap = listRouters(page.getRecords(), request.getSystemType());
        // 原始的数据库模型数据转化为页面展示的数据模型
        List<PendingMessageBriefRes> msgRecordBriefs = page.getRecords().stream()
                .map(e -> convert(e, categoryConfigs, routerMap))
                .collect(Collectors.toList());
        // 列表为空的场景
        return PendingMessageStatisticRes.builder()
                .totalElements(page.getTotal())
                .pageNum(page.getCurrent())
                .pageSize(page.getSize())
                .unCompleteCnt(stateStatisticMap.get(unCompleteStateCntKey))
                .completeCnt(stateStatisticMap.get(completeStateCntKey))
                .briefResList(msgRecordBriefs)
                .build();
    }

    @Override
    public List<PendingMessageTemporarilyTypeRes> listTemporarilyTypes() {
        return pendingMessageBizConfig.tempCategoriesConfig().stream()
                .sorted()
                .map(MsgTempBizCategoryConfig::toTemporarilyType)
                .collect(Collectors.toList());
    }

    @Override
    public Page<PendingMessageBriefRes> pageQuery(PendingMessagePageReq request) {
        IPage<MessageRecord> page = request.toPage();
        Set<Long> relationIds = null;
        if (CollectionUtils.isNotEmpty(request.getModuleIds())) {
            List<Long> relationIdList = listRelationIds(request.getModuleIds());
            if (CollectionUtils.isEmpty(relationIdList)) {
                // 对应模块及其关联的模板删除的场景
                return Page.toPage(page.getCurrent(), page.getSize(), page.getTotal(), Collections.emptyList());
            }
            relationIds = new HashSet<>(relationIdList);
        }
        // 获取数据冷热隔离的分界
        final LocalDateTime queryFrom = getQueryFromTime();
        // 这里对前端未传入参的情况做一个兜底
        Collection<MsgStateEnum> msgStates = CollectionUtils.isEmpty(request.getStates()) ? ALL_WITHOUT_UNSENT :
                request.getStates();
        // 分页查询message_record的数据
        page = page(request.getPersonId(), request.getIdentifyId(), request.getIdentifyTypes(), relationIds,
                msgStates, queryFrom, page);
        if (CollectionUtils.isEmpty(page.getRecords())) {
            // 列表为空的场景
            return Page.toPage(page.getCurrent(), page.getSize(), page.getTotal(), Collections.emptyList());
        }
        // 查询消息对应的路由集
        Map<Long, MessageRouterInfoRes> routerMap = listRouters(page.getRecords(), request.getSystemType());
        // 查询消息对应的模块集
        Map<Long, MsgModuleDTO> moduleMap = listMessageModules(page.getRecords());
        // 原始的数据库模型数据转化为页面展示的数据模型
        List<PendingMessageBriefRes> msgRecordBriefs = page.getRecords().stream()
                .filter(e -> moduleMap.containsKey(e.getId()))
                .map(e -> convert(e, moduleMap.get(e.getId()), routerMap))
                .collect(Collectors.toList());
        return Page.toPage(page.getCurrent(), page.getSize(), page.getTotal(), msgRecordBriefs);
    }

    private List<Long> listRelationIds(List<Long> moduleIds) {
        List<MessageModule> modules = messageModuleService.getModuleByIds(moduleIds);
        if (CollectionUtils.isEmpty(modules)) {
            return Collections.emptyList();
        }
        moduleIds = modules.stream().map(MessageModule::getId).collect(Collectors.toList());
        return messageRelationService.listRelationIds(moduleIds);
    }

    private Map<Long, MsgModuleDTO> listMessageModules(List<MessageRecord> msgRecords) {
        //获取relationId集合
        Set<Long> relationIds = msgRecords.stream().map(MessageRecord::getRelationId).collect(Collectors.toSet());
        if (CollectionUtils.isNotEmpty(relationIds)) {
            relationIds.remove(null);
            relationIds.remove(0L);
        }
        //获取oldTypeId集合
        Set<Integer> oldTypeIds = msgRecords.stream().map(MessageRecord::getOldTypeId).collect(Collectors.toSet());
        if (CollectionUtils.isNotEmpty(oldTypeIds)) {
            oldTypeIds.remove(null);
            oldTypeIds.remove(0);
        }
        if (CollectionUtils.isEmpty(relationIds) && CollectionUtils.isEmpty(oldTypeIds)) {
            return Collections.emptyMap();
        }
        //获取模板列表
        List<MsgModuleDTO> modules = messageModuleService.listModuleByRelationIdsOrTypeIds(oldTypeIds, relationIds);
        //新关联键分组
        Map<Long, List<MsgModuleDTO>> relationModuleMap = modules.stream()
                .filter(e -> Objects.nonNull(e.getRelationId()) && e.getRelationId() != 0)
                .collect(Collectors.groupingBy(MsgModuleDTO::getRelationId));
        //老关联键分组
        Map<Integer, List<MsgModuleDTO>> typeIdModuleMap = modules.stream()
                .filter(e -> Objects.nonNull(e.getTypeId()) && e.getTypeId() != 0)
                .collect(Collectors.groupingBy(MsgModuleDTO::getTypeId));
        return msgRecords.stream()
                .map(e -> selectModule(e, relationModuleMap, typeIdModuleMap))
                .filter(Objects::nonNull)
                .collect(Collectors.toMap(MsgModuleDTO::getMsgId, Function.identity(), (pre, cur) -> cur));
    }

    private MsgModuleDTO selectModule(MessageRecord record, Map<Long, List<MsgModuleDTO>> relationModuleMap,
                                      Map<Integer, List<MsgModuleDTO>> typeIdModuleMap) {
        MsgModuleDTO module = null;
        // 优先使用relation关联的模板
        if (MessageRecordHelper.isRelationIdValid(record.getRelationId())) {
            module = relationModuleMap.get(record.getRelationId()).stream()
                    .findFirst().orElse(null);
        }
        if (Objects.isNull(module) && MessageRecordHelper.isOldTypeIdValid(record.getOldTypeId())) {
            module = typeIdModuleMap.get(record.getOldTypeId()).stream()
                    .findFirst().orElse(null);
        }
        Optional.ofNullable(module)
                .ifPresent(v -> v.setMsgId(record.getId()));
        return module;
    }

    private PendingMessageTemporarilyStatisticRes statistic(PendingMessageTemporarilyStatisticReq request,
                                                            MsgTempBizCategoryConfig categoryConfig) {
        // 统计数量
        Integer count = pendingMessageNewService.countUncompleted(request.getPersonId(), categoryConfig.getConfigRelationIds());
        PendingMessageTemporarilyStatisticRes res = new PendingMessageTemporarilyStatisticRes();
        res.setUnCompletedCount(count);
        res.setCategoryCode(categoryConfig.getCategory().name());
        res.setCategoryDesc(categoryConfig.getCategory().getDesc());
        res.setIconUrl(categoryConfig.getIconUrl());
        if (StringUtils.isNotBlank(categoryConfig.getJumpUrl())) {
            // 分类指定了具体的跳转地址,则优先使用其指定的地址，这里指定为原生APP
            res.setRouter(new MessageRouterInfoRes(NativeTypeEnum.H5.getMessage(), categoryConfig.getJumpUrl(), ""));
        }
        return res;
    }

    private void checkParam(PendingMessageTemporarilyStatisticReq request) {
        if (Objects.isNull(request)
                || Objects.isNull(request.getIdentifyId())) {
            log.error("the param is invalid. {}", request);
            throw new BizException(ReturnCodeEnum.INVALID_PARAMETER);
        }
    }

    private LocalDateTime getQueryFromTime() {
        return Optional.ofNullable(messageSystemConfig.getDataDivideDays())
                .filter(e -> e > 0)
                .map(e -> LocalDate.now().minusDays(e).atStartOfDay())
                .orElse(null);
    }

    private Set<Long> fetchRelationIdFromRequestOrConfig(PendingMessagePageReq request,
                                                         List<MsgTempBizCategoryConfig> categoryConfigs) {
        MsgTempBizCategoryEnum category = request.getTempBizCategory();
        if (Objects.isNull(category)) {
            // 若前端未指定具体的分类,则取配置的所有的分类
            return categoryConfigs.stream()
                    .flatMap(e -> e.getConfigRelationIds().stream())
                    .collect(Collectors.toSet());
        }
        return categoryConfigs.stream()
                .filter(e -> e.getCategory().equals(category))
                .flatMap(e -> e.getConfigRelationIds().stream())
                .collect(Collectors.toSet());
    }

    private PendingMessageBriefRes convert(MessageRecord record, List<MsgTempBizCategoryConfig> categoryConfigs,
                                           Map<Long, MessageRouterInfoRes> routerMap) {
        PendingMessageBriefRes msgBrief = BeanConvertUtils.copyBean(record, PendingMessageBriefRes.class);
        msgBrief.setMsgId(record.getId());
        if (CollectionUtils.isNotEmpty(categoryConfigs)) {
            msgBrief.setTempBizCategory(getTempCategory(record.getRelationId(), categoryConfigs));
        }
        msgBrief.setRouter(routerMap.get(record.getId()));
        msgBrief.setCreateTimestamp(Optional.ofNullable(record.getCreateAt()).map(Date::getTime).orElse(0L));
        return msgBrief;
    }

    private PendingMessageBriefRes convert(MessageRecord record, MsgModuleDTO module,
                                           Map<Long, MessageRouterInfoRes> routerMap) {
        PendingMessageBriefRes msgBrief = BeanConvertUtils.copyBean(record, PendingMessageBriefRes.class);
        msgBrief.setMsgId(record.getId());
        msgBrief.setModuleName(module.getMsgModuleName());
        msgBrief.setModuleIcon(module.getMsgModuleIcon());
        msgBrief.setRouter(routerMap.get(record.getId()));
        msgBrief.setCreateTimestamp(Optional.ofNullable(record.getCreateAt()).map(Date::getTime).orElse(0L));
        return msgBrief;
    }

    private MsgTempBizCategoryEnum getTempCategory(Long relationId, List<MsgTempBizCategoryConfig> categoryConfigs) {
        return categoryConfigs.stream()
                .filter(e -> e.match(relationId))
                .map(MsgTempBizCategoryConfig::getCategory)
                .findFirst()
                .orElse(MsgTempBizCategoryEnum.PENDING_TASK);
    }

    private Map<Long, MessageRouterInfoRes> listRouters(Collection<MessageRecord> msgRecords, String systemType) {
        final Set<Long> relationIds = msgRecords.stream().map(MessageRecord::getRelationId).collect(Collectors.toSet());
        Map<Long, List<MessageRouter>> rawRouterMap = messageRouterService.listRoutersByRelationIds(relationIds).stream()
                .collect(Collectors.groupingBy(MessageRouter::getRelationId));
        final Map<Long, MessageRouterInfoRes> resRouterMap = new HashMap<>();
        msgRecords.stream()
                .filter(e -> rawRouterMap.containsKey(e.getRelationId()))
                .forEach(e -> {
                    MessageRouterInfoRes router = pickAndParseMessageRouter(rawRouterMap.get(e.getRelationId()),
                            systemType, e.getRouterParams());
                    resRouterMap.put(e.getId(), router);
                });
        return resRouterMap;
    }

    private MessageRouterInfoRes pickAndParseMessageRouter(List<MessageRouter> rawRouters, String systemType, String routerParam) {
        MsgRouteTypeEnum routeType = getSystemType(systemType);
        return pickAndParseMessageRouter(rawRouters, routeType, routerParam);
    }

    private MessageRouterInfoRes pickAndParseMessageRouter(List<MessageRouter> messageRouters, MsgRouteTypeEnum msgRouteTypeEnum, String routerParam) {
        String rawRouterUrl = "";
        Integer routerType = null;
        if (messageRouters.size() == 1) {
            NativeTypeEnum nativeTypeEnum = NativeTypeEnum.getByCode(messageRouters.get(0).getRouterType().getCode());
            if (Objects.nonNull(nativeTypeEnum)) {
                rawRouterUrl = messageRouters.get(0).getRouterUrl();
                routerType = nativeTypeEnum.getMessage();
            }
        } else {
            //不止一条 获取当前系统对应路由
            List<MessageRouter> routers = messageRouters.stream()
                    .filter(k -> k.getRouterType().equals(msgRouteTypeEnum))
                    .collect(Collectors.toList());
            if (CollectionUtils.isNotEmpty(routers)) {
                NativeTypeEnum nativeTypeEnum = NativeTypeEnum.getByCode(routers.get(0).getRouterType().getCode());
                if (nativeTypeEnum != null) {
                    rawRouterUrl = routers.get(0).getRouterUrl();
                    routerType = nativeTypeEnum.getMessage();
                }
            } else {
                //优先取H5
                List<MessageRouter> routerH5 = messageRouters.stream()
                        .filter(k -> k.getRouterType().equals(MsgRouteTypeEnum.WEBVIEW))
                        .collect(Collectors.toList());
                if (CollectionUtils.isNotEmpty(routerH5)) {
                    NativeTypeEnum nativeTypeEnum = NativeTypeEnum.getByCode(routerH5.get(0).getRouterType().getCode());
                    if (nativeTypeEnum != null) {
                        rawRouterUrl = routerH5.get(0).getRouterUrl();
                        routerType = nativeTypeEnum.getMessage();
                    }
                }

                //再取小程序
                List<MessageRouter> routerUni = messageRouters.stream()
                        .filter(k -> k.getRouterType().equals(MsgRouteTypeEnum.MINI_PROGRAM)).collect(
                                Collectors.toList());
                if (CollectionUtils.isNotEmpty(routerUni)) {
                    NativeTypeEnum nativeTypeEnum = NativeTypeEnum.getByCode(routerUni.get(0).getRouterType().getCode());
                    if (nativeTypeEnum != null) {
                        rawRouterUrl = routerUni.get(0).getRouterUrl();
                        routerType = nativeTypeEnum.getMessage();
                    }
                }
            }
        }
        PlaceholderResolver resolver = PlaceholderResolver.getDefaultResolver();
        if (Objects.nonNull(routerType)) {
            String routerUrl = resolver.resolveByMap(rawRouterUrl,
                    isRouterParamValid(routerParam) ? JSON.parseObject(routerParam) : Collections.emptyMap());
            return new MessageRouterInfoRes(routerType, routerUrl, routerParam);
        }
        return null;
    }

    private boolean isRouterParamValid(String routerParam) {
        return StringUtils.isNotEmpty(routerParam)
                && !"null".equals(routerParam);
    }

    private MsgRouteTypeEnum getSystemType(String systemType) {
        /*String systemType = ContextInfoHolder.get().getSystemAndDeviceInfo().getSystemType();*/
        //判断是否为手机端调用
        if (MsgRouteTypeEnum.IOS.getMessage().equals(systemType)) {
            return MsgRouteTypeEnum.IOS;
        }
        if (MsgRouteTypeEnum.ANDROID.getMessage().equals(systemType)) {
            return MsgRouteTypeEnum.ANDROID;
        }
        if (MsgRouteTypeEnum.WECHAT_MINI_PROGRAM.getMessage().equals(systemType)) {
            return MsgRouteTypeEnum.WECHAT_MINI_PROGRAM;
        }
        //都不是表明确定web端调用
        return MsgRouteTypeEnum.WEB;
    }

    private IPage<MessageRecord> page(Long personId, Long identifyId, Collection<ReceiveTypeEnum> identifyTypes,
                                      Collection<Long> relationIds, Collection<MsgStateEnum> states,
                                      LocalDateTime queryFrom, IPage<MessageRecord> page) {
        return messageRecordDao.lambdaQuery()
                .eq(MessageRecord::getType, MsgTypeEnum.PENDING_MESSAGE)
                .eq(Objects.nonNull(personId), MessageRecord::getPersonId, personId)
                .eq(Objects.nonNull(identifyId), MessageRecord::getToId, identifyId)
                .in(CollectionUtils.isNotEmpty(identifyTypes), MessageRecord::getReceiveType, identifyTypes)
                .in(CollectionUtils.isNotEmpty(states), MessageRecord::getState, states)
                .in(CollectionUtils.isNotEmpty(relationIds), MessageRecord::getRelationId, relationIds)
                .ge(Objects.nonNull(queryFrom), MessageRecord::getCreateAt, queryFrom)
                .eq(MessageRecord::getIsDelete, YesNoEnum.NO.getCode())
                .orderByAsc(MessageRecord::getState)
                .orderByDesc(MessageRecord::getCreateAt)
                .page(page);
    }

    private Map<String, Integer> statistic(Collection<StateStatisticModule> stateStatisticModules,
                                           Collection<Long> relationIds,
                                           PendingMessagePageReq request, LocalDateTime queryFrom) {
        final Map<String, Integer> resultMap = new ConcurrentHashMap<>();
        CompletableFuture[] asyncStatisticArray = stateStatisticModules.stream()
                .map(e -> CompletableFuture.runAsync(() -> {
                            Integer cnt = doStatistic(request.getPersonId(), request.getIdentifyId(),
                                    relationIds, request.getIdentifyTypes(), e.states, queryFrom);
                            resultMap.put(e.stateKey, cnt);
                        },
                        stateStatisticTaskThreadPool))
                .toArray(CompletableFuture[]::new);
        CompletableFuture<Void> completableFuture = CompletableFuture.allOf(asyncStatisticArray);
        completableFuture.join();
        return resultMap;
    }

    private Integer doStatistic(Long personId, Long identifyId, Collection<Long> relationIds,
                                Collection<ReceiveTypeEnum> identifyTypes,
                                Collection<MsgStateEnum> msgStates, LocalDateTime begin) {
        return messageRecordDao.lambdaQuery()
                .eq(MessageRecord::getType, MsgTypeEnum.PENDING_MESSAGE)
                .eq(Objects.nonNull(personId), MessageRecord::getPersonId, personId)
                .eq(Objects.nonNull(identifyId), MessageRecord::getToId, identifyId)
                .in(CollectionUtils.isNotEmpty(identifyTypes), MessageRecord::getReceiveType, identifyTypes)
                .in(MessageRecord::getState, msgStates)
                .in(CollectionUtils.isNotEmpty(relationIds), MessageRecord::getRelationId, relationIds)
                .ge(Objects.nonNull(begin), MessageRecord::getCreateAt, begin)
                .eq(MessageRecord::getIsDelete, YesNoEnum.NO.getCode())
                .count();
    }

    @AllArgsConstructor
    private static class StateStatisticModule {
        private String stateKey;
        private Collection<MsgStateEnum> states;

        static StateStatisticModule build(String stateKey, Collection<MsgStateEnum> states) {
            return new StateStatisticModule(stateKey, states);
        }
    }
}
