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

import cn.axzo.core.domain.PageResult;
import cn.axzo.msg.center.api.enums.MsgStateEnum;
import cn.axzo.msg.center.api.enums.MsgTypeEnum;
import cn.axzo.msg.center.api.mq.SendMessageRecordMessage;
import cn.axzo.msg.center.api.request.CmsMsgQueryReq;
import cn.axzo.msg.center.api.request.MessageModuleStatisticReq;
import cn.axzo.msg.center.api.request.MessagePageQueryReq;
import cn.axzo.msg.center.api.request.MessageReadAllReq;
import cn.axzo.msg.center.api.request.MessageStatisticReq;
import cn.axzo.msg.center.api.response.MessageDetailRes;
import cn.axzo.msg.center.api.response.MessageModuleBriefRes;
import cn.axzo.msg.center.api.response.MessageModuleStatisticRes;
import cn.axzo.msg.center.api.response.MessageNewRes;
import cn.axzo.msg.center.api.response.MessageStatisticRes;
import cn.axzo.msg.center.api.response.MessageTotalRes;
import cn.axzo.msg.center.common.enums.SystemTypeEnum;
import cn.axzo.msg.center.common.utils.BeanConvertUtils;
import cn.axzo.msg.center.common.utils.PlaceholderResolver;
import cn.axzo.msg.center.domain.dto.MessagePageQueryDTO;
import cn.axzo.msg.center.domain.dto.MsgModuleDTO;
import cn.axzo.msg.center.domain.entity.MessageModule;
import cn.axzo.msg.center.domain.entity.MessageRelation;
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.UserTypeEnum;
import cn.axzo.msg.center.inside.notices.service.IMessageService;
import cn.axzo.msg.center.inside.notices.service.MessageCoreService;
import cn.axzo.msg.center.inside.notices.service.MessageModuleService;
import cn.axzo.msg.center.inside.notices.service.MessageRecordService;
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.RawMessageRecordService;
import cn.axzo.msg.center.mq.MqMessageRecord;
import cn.axzo.msg.center.mq.MqProducer;
import cn.axzo.msg.center.service.enums.MqMessageType;
import cn.azxo.framework.common.model.Page;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import com.google.common.collect.Lists;
import jodd.util.concurrent.ThreadFactoryBuilder;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.HashMap;
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;

import static cn.axzo.msg.center.domain.enums.MsgRouteTypeEnum.WECHAT_MINI_PROGRAM;

/**
 * @author : liuchuntao
 * @date : 2022/3/28 11:52
 * @description : 消息中心实现
 */
@Service
@Slf4j
public class MessageCoreServiceImpl implements MessageCoreService {

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

    @Resource
    private MessageRecordService messageRecordService;

    @Resource
    private MessageModuleService messageModuleService;

    @Resource
    private MessageRouterService messageRouterService;

    @Resource
    private MessageRelationService relationService;
    @Resource
    private IMessageService messageService;

    @Resource
    private RawMessageRecordService rawMessageRecordService;
    @Resource
    private MqProducer mqProducer;

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

    /**
     * 消息统计
     *
     * @return
     */
    @Override
    public MessageTotalRes statisticsMsg(Long personId, Long identityId, UserTypeEnum userTypeEnum) {
        //根据用户和当前身份获取未读和待办消息
        return messageRecordService.statisticsMsg4Total(personId, identityId, userTypeEnum);
    }


    /**
     * 消息列表 (原施工逻辑中兼容交易业务的无身份类型)
     *
     * @param req
     * @return
     */
    @Override
    public PageResult<MessageNewRes> listMsgInfo(CmsMsgQueryReq req) {
        PageResult<MessageNewRes> resPageResult;
        /*Long identityId = ContextInfoHolder.get().getUserInfo().getIdentityId();
        Long personId = null;

        // 如果是工人 APP,则需要走特殊的查询逻辑, 需要将施工业务的对应身份数据查询出来,交易业务全量数据查询出来,合并为一个列表返回
        if (Objects.equals(NT_CM_APP_GENERAL, ContextInfoHolder.get().getSystemAndDeviceInfo().getLoginDevice())) {
            personId = ContextInfoHolder.get().getUserInfo().getPersonId();
        }*/
        resPageResult = messageRecordService.pageMsgInfo(req, req.getPersonId(), req.determineIdentities());

        List<MessageNewRes> resList = refineMessageRes(resPageResult, req);

        complementMsg(resList);
        return new PageResult<>(resList, resPageResult.getTotalCount());
    }

    public List<MessageNewRes> refineMessageRes(PageResult<MessageNewRes> resPageResult,
                                                CmsMsgQueryReq req) {
        //路由信息获取
        List<MessageNewRes> resList = resPageResult.getData();
        if (CollectionUtils.isEmpty(resList)) {
            return resList;
        }
        //获取relationId集合
        Set<Long> relationIds = resList.stream().map(MessageNewRes::getRelationId).collect(
                Collectors.toSet());
        if (CollectionUtils.isNotEmpty(relationIds)) {
            relationIds.remove(null);
            relationIds.remove(0L);
        }

        //获取oldTypeId集合
        Set<Integer> oldTypeIds = resList.stream().map(MessageNewRes::getOldTypeId).collect(
                Collectors.toSet());
        if (CollectionUtils.isNotEmpty(oldTypeIds)) {
            oldTypeIds.remove(null);
            oldTypeIds.remove(0);
        }
        if (CollectionUtils.isEmpty(relationIds) && CollectionUtils.isEmpty(oldTypeIds)) {
            return resList;
        }
        //获取模板所对应的所有模块及icon
        List<MsgModuleDTO> moduleDtos = messageModuleService.listModuleByRelationIdsOrTypeIds(oldTypeIds, relationIds);
        //新关联键分组
        Map<Long, List<MsgModuleDTO>> relationModule = moduleDtos.stream().filter(k -> k.getRelationId() != null && k.getRelationId() != 0)
                .collect(Collectors.groupingBy(MsgModuleDTO::getRelationId));
        //老关联键分组
        Map<Integer, List<MsgModuleDTO>> typeIdModules = moduleDtos.stream().filter(k -> k.getTypeId() != null && k.getTypeId() != 0)
                .collect(Collectors.groupingBy(MsgModuleDTO::getTypeId));

        //批量获取路由信息
        MsgRouteTypeEnum msgRouteTypeEnum = getSystemType(req.getSystemType());
        List<MessageRouter> routers = new ArrayList<>();
        Map<Long, List<MessageRouter>> mapRout = new HashMap<>();
        if (CollectionUtils.isNotEmpty(relationIds)) {
            routers = messageRouterService.listRoutersByRelationIds(relationIds);
            mapRout = routers.stream().collect(Collectors.groupingBy(MessageRouter::getRelationId));
        }
        Map<Long, List<MessageRouter>> finalMapRout = mapRout;
        resList.forEach(item -> {
            //赋值状态
            if (req.getMsgType() == 1) {
                item.setState(item.getState() == 4 ? 2 : 1);
            } else {
                item.setState(item.getState() == 5 ? 2 : 1);
            }
            //赋值模板
            if (item.getRelationId() != null && item.getRelationId() != 0) {
                List<MsgModuleDTO> modules = relationModule.get(item.getRelationId());
                if (CollectionUtils.isNotEmpty(modules)) {
                    MsgModuleDTO first = modules.stream()
                            .filter(i -> i.getOwnedType().equals(item.getReceiveType()))
                            .findFirst().orElse(modules.get(0));
                    item.setMsgIcon(first.getMsgModuleIcon());
                    item.setMsgName(first.getMsgModuleName());
                }
            }
            if (StrUtil.isEmpty(item.getMsgIcon())) {
                List<MsgModuleDTO> modules = typeIdModules.get(item.getOldTypeId());
                if (CollectionUtils.isNotEmpty(modules)) {
                    MsgModuleDTO first = modules.stream()
                            .filter(i -> i.getOwnedType().equals(item.getReceiveType()))
                            .findFirst().orElse(modules.get(0));
                    item.setMsgIcon(first.getMsgModuleIcon());
                    item.setMsgName(first.getMsgModuleName());
                }
            }

            //如果relationId为空 进行老方法获取路由
            if (item.getRelationId() == null || item.getRelationId() == 0) {
                messageService.setNewRouter(item, msgRouteTypeEnum.getCode());
            } else {
                List<MessageRouter> messageRouters = finalMapRout.get(item.getRelationId());
                if (CollectionUtils.isNotEmpty(messageRouters)) {
                    if (messageRouters.size() == 1) {
                        NativeTypeEnum nativeTypeEnum = NativeTypeEnum.getByCode(messageRouters.get(0).getRouterType().getCode());
                        if (nativeTypeEnum != null) {
                            item.setRouter(messageRouters.get(0).getRouterUrl());
                            item.setRouterType(nativeTypeEnum.getMessage());
                        }
                    } else {
                        //不止一条 获取当前系统对应路由
                        List<MessageRouter> router = messageRouters.stream().filter(k -> k.getRouterType().equals(msgRouteTypeEnum)).collect(
                                Collectors.toList());
                        if (CollectionUtils.isNotEmpty(router)) {
                            NativeTypeEnum nativeTypeEnum = NativeTypeEnum.getByCode(router.get(0).getRouterType().getCode());
                            if (nativeTypeEnum != null) {
                                item.setRouter(router.get(0).getRouterUrl());
                                item.setRouterType(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) {
                                    item.setRouter(routerH5.get(0).getRouterUrl());
                                    item.setRouterType(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) {
                                    item.setRouter(routerUni.get(0).getRouterUrl());
                                    item.setRouterType(nativeTypeEnum.getMessage());
                                }
                            }
                        }
                    }
                }
                PlaceholderResolver resolver = PlaceholderResolver.getDefaultResolver();
                if (StringUtils.hasLength(item.getRouter())) {
                    item.setRouter(resolver.resolveByMap(item.getRouter(),
                            JSON.parseObject(Objects.equals(item.getRouterParam(), "null") ? "{}" : item.getRouterParam(), Map.class)));
                }
            }
        });
        return resList;
    }

    private void complementMsg(List<MessageNewRes> resList) {
        Set<Long> nullModuleRelationIds = resList.stream().filter(
                        i -> !StringUtils.hasLength(i.getMsgIcon()) && !StringUtils.hasLength(i.getMsgName()))
                .map(MessageNewRes::getRelationId).collect(
                        Collectors.toSet());
        if (CollectionUtils.isEmpty(nullModuleRelationIds)) {
            return;
        }
        List<MessageRelation> msgRelation = relationService.getRelationByIds(nullModuleRelationIds);
        List<Long> moduleIds = msgRelation.stream()
                .map(MessageRelation::getModuleId).collect(Collectors.toList());

        Map<Long, MessageModule> moduleMap = messageModuleService.getModuleByIds(moduleIds).stream()
                .collect(Collectors.toMap(MessageModule::getId, Function.identity(), (s, t) -> s));

        Map<Long, MessageModule> relationModuleMapping = new HashMap<>();
        msgRelation.forEach(i -> {
            relationModuleMapping.put(i.getId(), moduleMap.get(i.getModuleId()));
        });

        resList.stream().filter(i -> !StringUtils.hasLength(i.getMsgIcon()) && !StringUtils.hasLength(i.getMsgName())).forEach(i -> {
            MessageModule messageModule = relationModuleMapping.getOrDefault(i.getRelationId(),
                    new MessageModule());
            i.setMsgIcon(messageModule.getIcon());
            i.setMsgName(messageModule.getModuleName());
        });
    }

    @Override
    public PageResult<MessageNewRes> listMsgInfo4Trade(CmsMsgQueryReq req) {
        PageResult<MessageNewRes> pageResult = messageRecordService.pageMsgInfo4WechatAndTrade(
                req, req.getPersonId());

        List<MessageNewRes> resList = refineMessageRes(pageResult, req);
        return new PageResult<>(resList, pageResult.getTotalCount());
    }

    @Override
    public MessageStatisticRes countPersonMessage(MessageStatisticReq request) {
        List<MsgStateEnum> states = Lists.newArrayList(MsgStateEnum.HAS_BEEN_SENT, MsgStateEnum.RECEIVED);
        int generalCnt = rawMessageRecordService.countPersonMessage(request.getPersonId(), MsgTypeEnum.GENERAL_MESSAGE,
                null, null, states);
        // 统计待办的数量 (切换为新接口)
        return new MessageStatisticRes(generalCnt, -1);
    }

    @Override
    public List<MessageModuleStatisticRes> statisticPersonModuleMessage(MessageModuleStatisticReq request) {
        Map<String, List<MessageModule>> moduleNameMap = messageModuleService.getAllModules().stream()
                .collect(Collectors.groupingBy(MessageModule::getModuleName));
        final Map<String, MessageModuleStatisticRes> resultMap = new ConcurrentHashMap<>();

        CompletableFuture[] asyncStatisticArray = moduleNameMap.entrySet().stream()
                .map(e -> CompletableFuture.runAsync(() -> doStatistic(request, e.getKey(), e.getValue(), resultMap),
                        moduleStatisticTaskThreadPool))
                .toArray(CompletableFuture[]::new);
        CompletableFuture<Void> completableFuture = CompletableFuture.allOf(asyncStatisticArray);
        completableFuture.join();
        return Lists.newArrayList(resultMap.values());
    }

    @Override
    public void readAllMessages(MessageReadAllReq request) {
        List<MsgStateEnum> srcStates = Lists.newArrayList(MsgStateEnum.HAS_BEEN_SENT, MsgStateEnum.RECEIVED);
        if (Objects.nonNull(request.getMsgId())
                && request.getMsgId() > 0) {
            // 阅读单条数据
            rawMessageRecordService.updatePersonMessageState(request.getPersonId(), srcStates, MsgStateEnum.HAVE_READ,
                    request.getMsgId());
        } else {
            List<Long> moduleIds = messageModuleService.listByModuleName(request.getModuleName()).stream()
                    .map(MessageModule::getId)
                    .collect(Collectors.toList());
            rawMessageRecordService.updatePersonMessageState(request.getPersonId(), srcStates, MsgStateEnum.HAVE_READ,
                    MsgTypeEnum.GENERAL_MESSAGE, null, moduleIds);
        }
        SendMessageRecordMessage mqMessage = new SendMessageRecordMessage();
        mqMessage.setReadRequest(request);
        mqProducer.send(MqMessageRecord
                .builder(MqMessageType.OLD_MSG_SEND, mqMessage)
                .build());
    }

    @Override
    public Page<MessageDetailRes> pageQueryMessageRecords(MessagePageQueryReq request) {
        MessagePageQueryDTO pageQuery = BeanConvertUtils.copyBean(request, MessagePageQueryDTO.class);
        List<Long> moduleIds = messageModuleService.listByModuleName(request.getModuleName()).stream()
                .map(MessageModule::getId)
                .collect(Collectors.toList());
        pageQuery.setModuleIds(moduleIds);
        return rawMessageRecordService.pagePersonMessages(pageQuery);
    }

    private void doStatistic(MessageModuleStatisticReq request, String moduleName, List<MessageModule> modules,
                             Map<String, MessageModuleStatisticRes> resultMap) {
        List<Long> moduleIds = modules.stream().map(MessageModule::getId).collect(Collectors.toList());
        List<MsgStateEnum> states = Lists.newArrayList(MsgStateEnum.HAS_BEEN_SENT, MsgStateEnum.RECEIVED);
        int cnt = rawMessageRecordService.countPersonMessage(request.getPersonId(), request.getMsgType(), null,
                moduleIds, states);
        Optional<MessageDetailRes> optional = rawMessageRecordService.selectLatestOnePersonMessage(request.getPersonId(),
                request.getMsgType(), request.getRouterType(), null, moduleIds);
        optional.ifPresent(v -> {
            MessageModuleBriefRes moduleBriefRes = new MessageModuleBriefRes();
            moduleBriefRes.setName(moduleName);
            moduleBriefRes.setIcon(modules.stream()
                    .filter(e -> org.apache.commons.lang3.StringUtils.isNotBlank(e.getIcon()))
                    .findFirst()
                    .map(MessageModule::getIcon)
                    .orElse(null));
            MessageModuleStatisticRes res = new MessageModuleStatisticRes();
            res.setContent(v.getContent());
            res.setCount(cnt);
            res.setTitle(v.getMsgTitle());
            res.setModule(moduleBriefRes);
            res.setTerminalName(v.getTerminalName());
            res.setCreateTimestamp(v.getMsgTimestamp());
            resultMap.put(moduleName, res);
        });
    }
}
