package cn.axzo.msg.center.message.service.impl;

import cn.axzo.basics.common.util.AssertUtil;
import cn.axzo.msg.center.api.enums.MsgTypeEnum;
import cn.axzo.msg.center.api.request.GeneralMessageReq;
import cn.axzo.msg.center.domain.entity.BizEventMapping;
import cn.axzo.msg.center.domain.entity.MessageRecord;
import cn.axzo.msg.center.domain.enums.BizActionCategory;
import cn.axzo.msg.center.message.domain.dto.SendImMessageDTO;
import cn.axzo.msg.center.message.send.processor.ISendProcessor;
import cn.axzo.msg.center.message.send.processor.SendContext;
import cn.axzo.msg.center.message.service.BizEventMappingService;
import cn.axzo.msg.center.message.service.GeneralMessageMapperService;
import cn.axzo.msg.center.message.service.GeneralMessageService;
import cn.axzo.msg.center.message.service.MessageSendTwiceRecordService;
import cn.axzo.msg.center.message.service.RelationTemplateMapService;
import cn.axzo.msg.center.service.dto.PersonDTO;
import cn.axzo.msg.center.service.enums.IdentityTypeEnum;
import cn.axzo.msg.center.service.enums.OrganizationTypeEnum;
import cn.axzo.msg.center.service.general.request.GeneralMessageSendRequest;
import cn.axzo.msg.center.utils.PersonIdentityUtil;
import cn.hutool.core.map.MapUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import java.util.Collection;
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.ExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import jodd.util.concurrent.ThreadFactoryBuilder;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.context.ApplicationContext;
import org.springframework.stereotype.Service;

/**
 * @author cold_blade
 * @date 2023/10/23
 * @version 1.0
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class GeneralMessageMapperServiceImpl implements GeneralMessageMapperService {

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

    private final GeneralMessageService generalMessageService;
    private final RelationTemplateMapService relationTemplateMapService;
    private final MessageSendTwiceRecordService messageSendTwiceRecordService;
    private final BizEventMappingService bizEventMappingService;
    private final ApplicationContext applicationContext;

    @Override
    public void asyncBatchSendMessage(GeneralMessageReq request, Map<Long, MessageRecord> toIdMessageRecordMap) {
        log.info("do some check before send im message. relationId:[{}]", request.getRelationId());
        if (MsgTypeEnum.PENDING_MESSAGE.equals(request.getType())) {
            // 待办消息暂时不支持
            log.info("pending message is not supported.");
            return;
        }
        if (Objects.isNull(request.getRelationId())) {
            log.info("relation id is null.");
            return;
        }
        log.info("request relationId= [{}], toldIdPersonIdMap= [{}]", request.getRelationId(),
                JSON.toJSONString(request.getToldIdPersonIdMap()));
        // IM 需要personId，过滤掉 personId = 0 的数据
        Map<Long,Long> imToldIdPersonIdMap = request.getToldIdPersonIdMap().entrySet()
                .stream()
                .filter(entry -> entry.getValue() > 0)
                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
        log.info("request relationId= [{}], final toldIdPersonIdMap= [{}]", request.getRelationId(),
                JSON.toJSONString(imToldIdPersonIdMap));
        request.setToldIdPersonIdMap(imToldIdPersonIdMap);
        if (MapUtil.isEmpty(request.getToldIdPersonIdMap())) {
            // 由于IM那边是根据personId来创建账户的,所以强依赖personId
            log.info("toIdPersonIdMap is empty.");
            return;
        }

        String bizCode = String.valueOf(request.getRelationId());
        Optional<BizEventMapping> bizEventOpt = bizEventMappingService.getByBizCode(bizCode);
        AssertUtil.isTrue(bizEventOpt.isPresent(), String.format("业务事件映射%s不存在", bizCode));
        BizEventMapping bizEvent = bizEventOpt.get();
        AssertUtil.notEmpty(bizEvent.getReachConfig(), String.format("业务事件映射%s无业务动作配置", bizCode));

        log.info("start to send reach message. bizCode:[{}]", bizCode);
        bizEvent.getReachConfig().forEach(r -> {
            BizActionCategory category = BizActionCategory.getCategory(r.getCategory());
            ISendProcessor processor = applicationContext.getBean(category.getProcessor(), ISendProcessor.class);

            SendContext sendContext = SendContext.builder()
                   .bizEvent(bizEvent).reachConfig(r)
                   .messageParam(request)
                   .toIdMessageRecordMap(toIdMessageRecordMap)
                   .build();
            CompletableFuture.runAsync(() -> processor.send(sendContext), asyncSendMsgExecutorService);
        });
    }

    private void doBatchSendMessage(GeneralMessageReq request, Map<Long, MessageRecord> toIdMessageRecordMap) {
        String templateCode = relationTemplateMapService.queryByRelationId(request.getRelationId()).orElse(null);
        if (StringUtils.isBlank(templateCode)) {
            log.info("the relationId([{}]) is not map any new message template. ", request.getRelationId());
            return;
        }
        log.info("start to send im message. relationId:[{}], templateCode:[{}]", request.getRelationId(), templateCode);
        try {
            // 发送IM消息
            GeneralMessageSendRequest sendImMsgRequest = buildSendRequest(request, templateCode, toIdMessageRecordMap.keySet());
            // IM消息的发送是基于personId+应用终端的(eg: 工人端/企业端)
            List<SendImMessageDTO> result = generalMessageService.batchSendMessage(sendImMsgRequest);
            if (CollectionUtils.isEmpty(result)) {
                log.info("there is not any person successfully send im message. relationId:[{}]", request.getRelationId());
                return;
            }
            // 记录发送了IM消息的旧消息
            recordSendImMessage(toIdMessageRecordMap, result, request.getToldIdPersonIdMap());
        } catch (Exception e) {
            log.warn("broke out some exception while sending im message. relationId:[{}]", request.getRelationId(), e);
        }
    }

    private void recordSendImMessage(Map<Long, MessageRecord> toIdMessageRecordMap, List<SendImMessageDTO> sendImResult,
                                     Map<Long, Long> toIdPersonIdMap) {
        // 成功发送了IM消息的personId集合
        Set<Long> sucSendImMsgPersonIds = sendImResult.stream()
                .map(SendImMessageDTO::getPersonId).collect(Collectors.toSet());
        // 从入参中的toIdPersonIdMap中筛选发送成功的entry
        Map<Long, Long> subToIdPersonIdMap = toIdPersonIdMap.entrySet().stream()
                // 过滤掉personId维度IM消息发送失败的entry
                .filter(e -> sucSendImMsgPersonIds.contains(e.getValue()))
                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
        Map<Long, Long> msgRecordIdPersonIdMap = toIdMessageRecordMap.entrySet().stream()
                .filter(e -> subToIdPersonIdMap.containsKey(e.getKey()))
                .collect(Collectors.toMap(e -> e.getValue().getId(), e -> subToIdPersonIdMap.get(e.getKey())));
        log.info("record message that has been send im message. msgIds:{}", msgRecordIdPersonIdMap.keySet());
        // 双发记录
        messageSendTwiceRecordService.batchSave(msgRecordIdPersonIdMap);
    }

    private GeneralMessageSendRequest buildSendRequest(GeneralMessageReq request, String templateCode,
                                                       Collection<Long> subReceiverIds) {
        IdentityTypeEnum identityType = PersonIdentityUtil.toIdentityType(request.getReceiveType());
        List<PersonDTO> receivers = subReceiverIds.stream()
                .filter(request.getToldIdPersonIdMap()::containsKey)
                .map(e -> PersonDTO.from(request.getToldIdPersonIdMap().get(e), e, identityType))
                .collect(Collectors.toList());
        OrganizationTypeEnum orgType = Objects.isNull(request.getTerminalType()) ? OrganizationTypeEnum.UNKNOWN :
                OrganizationTypeEnum.valueOf(request.getTerminalType().name());
        return GeneralMessageSendRequest.builder()
                .templateCode(templateCode)
                .receiver(receivers)
                .orgType(orgType)
                .orgName(request.getTerminalName())
                .orgId(request.getTerminalId())
                .bizCode(Optional.ofNullable(request.getBizId()).map(String::valueOf).orElse(""))
                .routerParams(Optional.ofNullable(request.getRouterParams())
                        .map(v -> JSONObject.parseObject(JSON.toJSONString(v)))
                        .orElseGet(JSONObject::new))
                .bizExtParams(Optional.ofNullable(request.getMsgParams())
                        .map(v -> JSONObject.parseObject(JSON.toJSONString(v)))
                        .orElseGet(JSONObject::new))
                .build();
    }
}
