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

import cn.axzo.msg.center.dal.MNSChannelMessageTemplateDao;
import cn.axzo.msg.center.domain.entity.*;
import cn.axzo.msg.center.notices.common.domain.MessageContext;
import cn.axzo.msg.center.notices.common.enums.*;
import cn.axzo.msg.center.notices.common.exception.BizException;
import cn.axzo.msg.center.notices.common.lang.Page;
import cn.axzo.msg.center.notices.common.utils.DateUtils;
import cn.axzo.msg.center.notices.integration.client.AliYunSmsClient;
import cn.axzo.msg.center.notices.manager.api.*;
import cn.axzo.msg.center.notices.manager.api.dto.request.BatchMessageRequestDto;
import cn.axzo.msg.center.notices.manager.api.dto.request.MessageQueryRequestDto;
import cn.axzo.msg.center.notices.manager.api.dto.request.MessageRedoQueryRequestDto;
import cn.axzo.msg.center.notices.manager.api.dto.request.MessageSendRequestDto;
import cn.axzo.msg.center.notices.manager.api.dto.response.SendSmsCommonResponseDto;
import cn.axzo.msg.center.notices.service.api.TaskService;
import cn.azxo.framework.common.model.JobPageRequest;
import cn.azxo.framework.common.utils.LogUtil;
import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.IdUtil;
import com.alibaba.fastjson.JSON;
import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.joda.time.DateTime;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.Date;
import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;

/**
 * 定时任务服务实现类
 *
 * @author zhaoyong_sh
 * @see TaskServiceImpl
 * @since 2021-05-19 15:48
 */
@Slf4j
@Service("taskService")
public class TaskServiceImpl implements TaskService {

    @Resource(name = "messageManager")
    private MessageManager messageManager;

    @Resource(name = "messageRedoManger")
    private MessageRedoManger messageRedoManger;

    @Resource(name = "aliYunSmsClient")
    private AliYunSmsClient aliYunSmsClient;

    @Resource(name = "smsSendManagerComposite")
    private SmsSendManager smsSendManagerComposite;

    @Resource(name = "retryStrategy")
    private RetryStrategy retryStrategy;

    @Resource(name = "batchMessageRequestManger")
    private BatchMessageRequestManger batchMessageRequestManger;

    @Resource(name = "batchMessageManger")
    private BatchMessageManger batchMessageManger;

    @Resource
    private MNSChannelMessageTemplateDao mnsChannelMessageTemplateDao;

    @Value("${realtime.batch.message.threshold.minutes:10}")
    private Integer thresholdMinutes;

    @Override
    public void aliYunProcessSendMessageException(JobPageRequest request) {
        log.info("TaskServiceImpl#aliYunProcessSendMessageException request param is {}", JSON.toJSONString(request));
        MessageRedoQueryRequestDto requestDto = new MessageRedoQueryRequestDto();
        requestDto.setStartCreateTime(request.getStartTime());
        if(StringUtils.isNoneBlank(request.getReferenceNo())) {
            requestDto.setId(Long.valueOf(request.getReferenceNo()));
        } else {
            requestDto.setChannelCode(MessageChannelEnum.ALIYUN.getCode());
            requestDto.setEndCreateTime(request.getEndTime());
            requestDto.setPageSize(request.getPageSize());
            requestDto.setStatuses(Lists.newArrayList(MessageStatusEnum.EXCEPTION.getCode()));
        }
        Page<MNSMessageRedo> messagePage = messageRedoManger.queryPageByMessageRedoQueryRequestDto(requestDto);
        if(messagePage == null) {
            log.info("TaskServiceImpl.aliYunProcessSendMessageException result is null");
            return;
        }
        List<MNSMessageRedo> redoMessages = messagePage.getList();

        if(CollectionUtils.isEmpty(redoMessages)){
            log.info("TaskServiceImpl.aliYunProcessSendMessageException result size is null");
            return;
        }
        log.info("TaskServiceImpl.aliYunProcessSendMessageException result size is {}", redoMessages.size());
        log.info("短信发送异常重试列表-------{}",JSON.toJSONString(redoMessages));
        for (MNSMessageRedo redoMessage : redoMessages) {
            String dateFormat = DateUtil.format(redoMessage.getCreateAt(), DatePattern.PURE_DATE_PATTERN);
            SendDetailsStatusEnum statusEnum = aliYunSmsClient
                .queryExceptionSendDetails(redoMessage.getTargetAddress(), dateFormat, redoMessage.getMessageOrderNo());
            if(statusEnum != null) {
                // 更新状态
                String messageStatus = SendDetailsStatusEnum.getMappingCode(statusEnum);
                messageManager.updateExceptionToQueryStatus(messageStatus, redoMessage.getId());
                messageRedoManger.updateMessageRedoStatusById(redoMessage.getId(),MessageStatusEnum.EXCEPTION.getCode(),messageStatus);
            } else {
                // 锁定消息
                MessageContext context = MessageContext.builder().retryCount(redoMessage.getRetryCount())
                    .notifyRule(redoMessage.getNotifyRule()).build();
                messageRedoManger.lockMessage(context, redoMessage.getId());

                MNSChannelMessageTemplate channelMessageTemplate = mnsChannelMessageTemplateDao
                    .queryByTemplateNo(redoMessage.getTemplateNo(), redoMessage.getChannelCode());
                if (Objects.isNull(channelMessageTemplate)){
                    LogUtil.error("短信重试渠道模板不存在--{}",JSON.toJSONString(redoMessage));
                    continue;
                }
                // 发送消息
                String redoOrderNo = IdUtil.simpleUUID();
                log.info("当前发送短信的重试单号为===={}",redoOrderNo);
                MessageSendRequestDto dto = new MessageSendRequestDto();
                dto.setAppCode(redoMessage.getAppNo());
                dto.setChannelCode(redoMessage.getChannelCode());
                dto.setPhoneNo(redoMessage.getTargetAddress());
                dto.setTemplateMap(JSON.parseObject(redoMessage.getSubject()));
                dto.setAppRequestNo(redoMessage.getRequestNo());
                dto.setRequestChannelNo(redoOrderNo);
                dto.setTemplateNo(channelMessageTemplate.getTemplateNo());
                try {
                    SendSmsCommonResponseDto sendSmsCommonResponseDto = smsSendManagerComposite.sendMessage(dto);
                    log.info("短信重试返回=========={}",JSON.toJSONString(sendSmsCommonResponseDto));
                    if (Objects.nonNull(sendSmsCommonResponseDto)){
                        messageRedoManger.updateMessageRedoChannelById(redoMessage.getId(),sendSmsCommonResponseDto.getBizId(),dto.getRequestChannelNo());
                    }
                } catch (BizException e) {
                    messageRedoManger.updateExceptionToFail(redoMessage.getId(), redoMessage.getRedoOrderNo());
                } catch (Exception e) {
                    // 获取下次重试秒数
                    Integer nextNotifySecond = retryStrategy.getNextNotifySecond(context);
                    // 到达最大重试次数,更新消息为失败
                    if(nextNotifySecond == null) {
                        messageRedoManger.updateExceptionToFail(redoMessage.getId(), redoMessage.getRedoOrderNo());
                    } else {
                        Date retryAt = DateUtils.plusSeconds(nextNotifySecond);
                        messageRedoManger.updateRetry(retryAt, redoMessage.getId());
                    }
                }
            }
        }

    }

    @Override
    public void aliYunMessageProcessingHandle(JobPageRequest request) {
        log.info("TaskServiceImpl#aliYunMessageProcessingHandle request param is {}", JSON.toJSONString(request));
        MessageQueryRequestDto dto = new MessageQueryRequestDto();
        dto.setStartCreateTime(null);
        if(StringUtils.isNoneBlank(request.getReferenceNo())) {
            dto.setId(Long.valueOf(request.getReferenceNo()));
        } else {
            dto.setChannelCode(MessageChannelEnum.ALIYUN.getCode());
            dto.setStartCreateTime(request.getStartTime());
            dto.setEndCreateTime(request.getEndTime());
            dto.setPageSize(request.getPageSize());
            dto.setStatuses(Lists.newArrayList(MessageStatusEnum.PROCESSING.getCode()));
        }
        Page<MNSMessage> messagePage = messageManager.queryPageByMessageQueryRequestDto(dto);
        if(messagePage == null) {
            log.info("TaskServiceImpl.aliYunMessageProcessingHandle result is null");
            return;
        }
        List<MNSMessage> messages = messagePage.getList();

        if(CollectionUtils.isEmpty(messages)){
            log.info("TaskServiceImpl.aliYunMessageProcessingHandle result size is null");
            return;
        }
        log.info("TaskServiceImpl.aliYunMessageProcessingHandle result size is {}", messages.size());
        for (MNSMessage message : messages) {
            String dateFormat = DateUtil.format(message.getCreateAt(), DatePattern.PURE_DATE_PATTERN);
            //查询短信状态
            SendDetailsStatusEnum statusEnum = aliYunSmsClient
                .queryProcessingSendDetail(message.getTargetAddress(), dateFormat, message.getChannelBizId());
            String messageStatus;
            if( Objects.isNull(statusEnum)) {
                // 查询不到的更新状态为异常
                messageStatus = MessageStatusEnum.FAIL.getCode();
            }else{
                messageStatus = SendDetailsStatusEnum.getMappingCode(statusEnum);
                //发送中的话，不做处理
                if (MessageStatusEnum.PROCESSING.getCode().equals(messageStatus)){
                    continue;
                }
            }
            // 更新状态
            messageManager.updateProcessingMessageStatusById(message.getId(),messageStatus);
        }

    }

    @Override
    public void batchMessageRequestSplitHandle(JobPageRequest request) {
        log.info("TaskServiceImpl#batchMessageRequestSplitHandle request param is {}",
            JSON.toJSONString(request));
        BatchMessageRequestDto dto = buildBatchMessageRequestDto(request,
            BatchMessageRequestStatusEnum.INITIALIZE.getCode(), BatchSendTypeEnum.NOT_REALTIME.getCode());
        log.info("TaskServiceImpl#batchMessageRequestSplitHandle request dto is {}",
            JSON.toJSONString(dto));
        Page<MNSBatchMessageRequest> requestPage = batchMessageRequestManger
            .queryPageByBatchMessageQueryRequestDto(dto);
        if (requestPage == null || CollectionUtils
            .isEmpty(requestPage.getList())) {
            log.info("TaskServiceImpl.batchMessageRequestSplitHandle result is null");
            return;
        }
        List<MNSBatchMessageRequest> list = requestPage.getList();
        log.info("TaskServiceImpl.batchMessageRequestSplitHandle result size is {}", list.size());
        list.forEach(batchMessageRequest -> {
            //拆分 batch_message_request 到 batch_message
            log.info("TaskServiceImpl.batchMessageRequestSplitHandle 开始拆分 batchMessageRequest = {}",
                JSON.toJSONString(batchMessageRequest));
            batchMessageRequestManger.batchMessageRequestSplitAndSave(batchMessageRequest);

        });
    }


    @Override
    public void batchMessageSendSplitHandle(JobPageRequest request) {
        log.info("TaskServiceImpl#batchMessageSendSplitHandle request param is {}",
            JSON.toJSONString(request));
        BatchMessageRequestDto dto = buildBatchMessageRequestDto(request,
            BatchMessageStatusEnum.INITIALIZE.getCode(), BatchSendTypeEnum.NOT_REALTIME.getCode());
        log.info("TaskServiceImpl#batchMessageSendSplitHandle request dto is {}",
            JSON.toJSONString(dto));
        Page<MNSBatchMessage> page = batchMessageManger.queryByPage(dto);
        if (page == null || CollectionUtils
            .isEmpty(page.getList())) {
            log.info("TaskServiceImpl.batchMessageSendSplitHandle result is null");
            return;
        }
        List<MNSBatchMessage> list = page.getList();
        log.info("TaskServiceImpl.batchMessageSendSplitHandle result size is {}", list.size());
        list.forEach(batchMessage -> {
            //拆分批次保存到message和更新状态为拆分完成
            log.info("TaskServiceImpl.batchMessageSendSplitHandle 开始拆分 batchMessage = {}",
                JSON.toJSONString(batchMessage));
            batchMessageManger.batchMessageSplitToMessage(batchMessage);
        });

    }

    @Override
    public void batchSendMessage(JobPageRequest request) {
        log.info("TaskServiceImpl#batchSendMessage request param is {}",
            JSON.toJSONString(request));
        BatchMessageRequestDto dto = buildBatchMessageRequestDto(request,
            BatchMessageStatusEnum.SPLIT_COMPLETED.getCode(), BatchSendTypeEnum.NOT_REALTIME.getCode());
        log.info("TaskServiceImpl#batchSendMessage request dto is {}",
            JSON.toJSONString(dto));
        Page<MNSBatchMessage> page = batchMessageManger.queryByPage(dto);
        if (page == null || CollectionUtils
            .isEmpty(page.getList())) {
            log.info("TaskServiceImpl.batchSendMessage result is null");
            return;
        }
        List<MNSBatchMessage> list = page.getList();
        log.info("TaskServiceImpl.batchSendMessage result size is {}", list.size());
        list.forEach(batchMessage -> {
            //发送短信、更新状态和回执单号
            log.info("TaskServiceImpl.batchSendMessage 开始才分批次 batchMessage = {}",batchMessage);
            batchMessageManger.sendBatchMessageAndHandleReceipt(batchMessage);
        });

    }

    @Override
    public void realBatchMessageRequestSplitCompensation(JobPageRequest request) {
        log.info("TaskServiceImpl#realBatchMessageRequestSplitCompensation request param is {}",
            JSON.toJSONString(request));
        BatchMessageRequestDto dto = buildBatchMessageRequestDto(request,
            BatchMessageRequestStatusEnum.INITIALIZE.getCode(), BatchSendTypeEnum.REALTIME.getCode());
        log.info("TaskServiceImpl#realBatchMessageRequestSplitCompensation request dto is {}",
            JSON.toJSONString(dto));
        Page<MNSBatchMessageRequest> batchMessageRequestPage = batchMessageRequestManger
            .queryPageByBatchMessageQueryRequestDto(dto);
        if (batchMessageRequestPage == null || CollectionUtils
            .isEmpty(batchMessageRequestPage.getList())) {
            log.info("TaskServiceImpl.realBatchMessageRequestSplitCompensation result is null");
            return;
        }

        batchMessageRequestPage.getList().stream()
            .filter(v -> canCompensation(v.getCreateAt(), thresholdMinutes))
            .forEach(batchMessageRequest -> {
            //拆分 batch_message_request 到 batch_message
            log.info("TaskServiceImpl.realBatchMessageRequestSplitCompensation 开始拆分 batchMessageRequest = {}",
                JSON.toJSONString(batchMessageRequest));

            // 拆分短信
            List<String> batchNos = batchMessageRequestManger.batchMessageRequestSplitAndSave(batchMessageRequest);

            // 消息幂等
            if(CollectionUtils.isEmpty(batchNos)) {
                log.info("TaskServiceImpl.realBatchMessageRequestSplitCompensation is idempotent {}",
                    JSON.toJSONString(batchMessageRequest));
                return;
            }
            log.info("TaskServiceImpl.realBatchMessageRequestSplitCompensation size is {}", batchNos.size());

            // remove rocketmq
        });
    }

    @Override
    public void realBatchMessageSplitCompensation(JobPageRequest request) {
        log.info("TaskServiceImpl#realBatchMessageSplitCompensation request param is {}",
            JSON.toJSONString(request));
        BatchMessageRequestDto dto = buildBatchMessageRequestDto(request,
            BatchMessageStatusEnum.INITIALIZE.getCode(), BatchSendTypeEnum.REALTIME.getCode());
        log.info("TaskServiceImpl#realBatchMessageSplitCompensation request dto is {}",
            JSON.toJSONString(dto));
        Page<MNSBatchMessage> page = batchMessageManger.queryByPage(dto);
        if (page == null || CollectionUtils
            .isEmpty(page.getList())) {
            log.info("TaskServiceImpl.realBatchMessageSplitCompensation result is null");
            return;
        }
        List<MNSBatchMessage> list = page.getList().stream()
            .filter(v -> canCompensation(v.getCreateAt(), thresholdMinutes))
            .collect(Collectors.toList());
        log.info("TaskServiceImpl.realBatchMessageSplitCompensation size is {}", list.size());

        list.forEach(message -> {
            //拆分批次保存到message和更新状态为拆分完成
            log.info("TaskServiceImpl.realBatchMessageSplitCompensation 开始拆分 message = {}",
                JSON.toJSONString(message));

            // 批量消息拆分为 Message
            boolean result = batchMessageManger.batchMessageSplitToMessage(message);

            if(!result) {
                log.info("TaskServiceImpl.realBatchMessageSplitCompensation is idempotent {}",
                    JSON.toJSONString(message));
                return;
            }

            // 发送渠道消息
            try {
                batchMessageManger.sendBatchMessageAndHandleReceipt(message);
            } catch (Exception e) {
                LogUtil.error("TaskServiceImpl.realBatchMessageSplitCompensation send channel fail", e);
            }
        });
    }

    @Override
    public void batchMessageRequestCountCompensation(JobPageRequest request) {
        log.info("TaskServiceImpl#batchMessageRequestCountCompensation request param is {}",
            JSON.toJSONString(request));
        BatchMessageRequestDto dto = buildBatchMessageRequestDto(request,
            BatchMessageRequestStatusEnum.SPLIT_COMPLETED.getCode(), null);

        Page<MNSBatchMessageRequest> messagePage = batchMessageRequestManger
            .queryPageByBatchMessageQueryRequestDto(dto);
        if (messagePage == null || CollectionUtils
            .isEmpty(messagePage.getList())) {
            log.info("TaskServiceImpl.batchMessageRequestCountCompensation result is null");
            return;
        }

        List<MNSBatchMessageRequest> list = messagePage.getList();
        log.info("TaskServiceImpl.batchMessageRequestCountCompensation result size is {}", list.size());

        list.stream().filter(v -> canCompensation(v.getCreateAt(), thresholdMinutes * 3))
            .forEach(v -> {
                log.info("TaskServiceImpl.batchMessageRequestCountCompensation request no is {}", v.getRequestNo());

                batchMessageRequestManger.handleBatchMessageRequestStatistics(v);
            });
    }

    /**
     * 构建请求对象
     * @param request
     * @param status
     * @param sendType
     * @return
     */
    private BatchMessageRequestDto buildBatchMessageRequestDto(JobPageRequest request,
        Integer status, Integer sendType) {
        BatchMessageRequestDto dto = new BatchMessageRequestDto();
        if (StringUtils.isNoneBlank(request.getReferenceNo())) {
            dto.setId(Long.valueOf(request.getReferenceNo()));
        }else {
            dto.setStartCreateTime(request.getStartTime());
            dto.setEndCreateTime(request.getEndTime());
            dto.setPageSize(request.getPageSize());
            dto.setStatuses(Lists.newArrayList(status));
            dto.setSendType(sendType);
        }
        return dto;
    }

    /**
     * 超过10分钟(默认)没有执行能够补偿
     * @param createAt
     * @return
     */
    private boolean canCompensation(Date createAt, int plusMinutes){
        DateTime createDateTime = new DateTime(createAt);
        Date createAtPlusThresholdMinutes = createDateTime.plusMinutes(plusMinutes).toDate();
        Date currentTime = new Date();
        return createAtPlusThresholdMinutes.before(currentTime);
    }


}
