package com.njcn.msgpush.module.push.service.message;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.njcn.msgpush.framework.common.util.object.PageUtils;
import com.njcn.msgpush.module.push.checker.MsgPushGuardChain;
import com.njcn.msgpush.module.push.client.factory.MessageProviderFactory;
import com.njcn.msgpush.module.push.client.factory.MessageProviderFactoryRegistry;
import com.njcn.msgpush.module.push.client.sender.SendOutcome;
import com.njcn.msgpush.module.push.client.sender.SendResult;
import com.njcn.msgpush.module.push.client.sender.Sender;
import com.njcn.msgpush.module.push.constant.MsgStatusConstant;
import com.njcn.msgpush.module.push.controller.admin.message.vo.MessageRecordReqVO;
import com.njcn.msgpush.module.push.controller.admin.message.vo.MessageSendResultVO;
import com.njcn.msgpush.module.push.dal.dataobject.channel.ChannelProviderConfigDO;
import com.njcn.msgpush.module.push.dal.dataobject.message.MessageRecordDO;
import com.njcn.msgpush.module.push.dal.dataobject.retry.MessageRetryHistoryDO;
import com.njcn.msgpush.module.push.dal.mysql.message.MessageRecordMapper;
import com.njcn.msgpush.module.push.dal.redis.MessageConfirmRedisDAO;
import com.njcn.msgpush.module.push.dal.redis.RateLimitRedisDAO;
import com.njcn.msgpush.module.push.dal.redis.SystemQuotaRedisDAO;
import com.njcn.msgpush.module.push.enums.ChannelTypeEnum;
import com.njcn.msgpush.module.push.enums.RetrySourceEnum;
import com.njcn.msgpush.module.push.service.channel.ChannelProviderConfigService;
import com.njcn.msgpush.module.push.service.retry.MessageRetryHistoryService;
import com.njcn.msgpush.module.push.service.retry.MessageRetryQueueService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;

@Service
public class MessageRecordServiceImpl extends ServiceImpl<MessageRecordMapper, MessageRecordDO> implements MessageRecordService {

    @Autowired
    private Sender sender;

    @Autowired
    private MessageRetryHistoryService messageRetryHistoryService;

    @Autowired
    private MessageRetryQueueService messageRetryQueueService;

    @Autowired
    public ChannelProviderConfigService channelProviderConfigService;

    @Autowired
    private MessageProviderFactoryRegistry messageProviderFactoryRegistry;

    @Autowired
    private MsgPushGuardChain msgPushGuardChain;

    @Autowired
    private SystemQuotaRedisDAO systemQuotaRedisDAO;

    @Autowired
    private RateLimitRedisDAO rateLimitRedisDAO;

    @Autowired
    private MessageConfirmRedisDAO messageConfirmRedisDAO;

    @Override
    public List<MessageSendResultVO> send(List<MessageRecordReqVO> reqVOList, ChannelTypeEnum channelTypeEnum, String systemName) {
        List<MessageRecordDO> messageRecordDOList = this.createMessageRecords(reqVOList, channelTypeEnum, systemName);
        return this.processSendMsg(messageRecordDOList, null);
    }

    @Transactional(rollbackFor = Exception.class)
    public List<MessageRecordDO> createMessageRecords(List<MessageRecordReqVO> reqVOList, ChannelTypeEnum channelTypeEnum, String systemName) {
        List<MessageRecordDO> messageRecordDOList = BeanUtil.copyToList(reqVOList, MessageRecordDO.class);
        messageRecordDOList.forEach(messageRecordDO -> {
            messageRecordDO.setChannel(channelTypeEnum.getCode());
            messageRecordDO.setStatus(MsgStatusConstant.PENDING);
            messageRecordDO.setAppName(systemName);
//            messageRecordDO.setRetryCount(0);
        });
        this.saveBatch(messageRecordDOList);
        return messageRecordDOList;
    }

    /**
     * 发送主流程。
     * 这里统一接收 sender 返回的 SendResult，再决定消息状态、重试队列和服务商健康度如何变化。
     */
    @Override
    public List<MessageSendResultVO> processSendMsg(List<MessageRecordDO> messageRecordDOList, RetrySourceEnum retrySource) {
        msgPushGuardChain.checkAll(messageRecordDOList);
        List<MessageSendResultVO> resultList = new ArrayList<>();
        for (MessageRecordDO messageRecordDO : messageRecordDOList) {
            try {
                MessageSendResultVO resultVO = this.processSingleMessage(messageRecordDO, retrySource);
                resultList.add(resultVO);
            } catch (Exception e) {
                MessageSendResultVO failedVO = new MessageSendResultVO();
                failedVO.setMessageId(messageRecordDO.getId());
                failedVO.setResult(false);
                failedVO.setDetail("消息处理异常：" + e.getMessage());
                resultList.add(failedVO);
            }
        }
        return resultList;
    }

    private MessageSendResultVO processSingleMessage(MessageRecordDO messageRecordDO, RetrySourceEnum retrySource) {
        MessageSendResultVO resultVO = new MessageSendResultVO();
        resultVO.setMessageId(messageRecordDO.getId());

        if (!MsgStatusConstant.PENDING.equals(messageRecordDO.getStatus())) {
            resultVO.setResult(false);
            resultVO.setDetail(this.buildBlockedDetail(messageRecordDO));
            this.updateById(messageRecordDO);
            return resultVO;
        }
        List<ChannelProviderConfigDO> enabledProviders = channelProviderConfigService.getEnabledProviders(messageRecordDO.getChannel());
        SendResult sendResult = null;
        if (CollectionUtil.isNotEmpty(enabledProviders)) {
            ChannelProviderConfigDO channelProviderConfigDO = enabledProviders.get(0);
            MessageProviderFactory messageProviderFactory = messageProviderFactoryRegistry.getFactory(channelProviderConfigDO.getProviderType());
            sendResult = this.validateProviderAndChannel(messageRecordDO, channelProviderConfigDO, messageProviderFactory);

            messageRecordDO.setProviderType(channelProviderConfigDO.getProviderType());
            if (sendResult == null) {
                sendResult = this.sendMessage(messageRecordDO, channelProviderConfigDO, messageProviderFactory);
            }
            this.applySendResult(messageRecordDO, sendResult);
            this.recordRetryHistory(messageRecordDO, retrySource);
        } else {
            sendResult = this.validateProviderAndChannel(messageRecordDO, null, null);
            this.applySendResult(messageRecordDO, sendResult);
        }


        // 在这里修改接口返回的是投递结果，还是调用接口结果
        resultVO.setResult(this.isSuccessOutcome(sendResult.getOutcome()));
        resultVO.setDetail(this.buildResultDetail(sendResult));
        return resultVO;
    }

    private SendResult validateProviderAndChannel(MessageRecordDO messageRecordDO, ChannelProviderConfigDO channelProviderConfigDO, MessageProviderFactory messageProviderFactory) {
        ChannelTypeEnum channelType = ChannelTypeEnum.getByCode(messageRecordDO.getChannel());
        if (channelType == null) {
            return SendResult.unsupported("当前发送渠道类型暂不支持");
        }
        if (ObjectUtil.isNull(messageProviderFactory)) {
            return SendResult.configInvalid("当前服务商未启用或未注册");
        }
        if (ObjectUtil.isNull(channelProviderConfigDO)) {
            return SendResult.configInvalid("当前服务商渠道配置不存在");
        }
        return null;
    }

    private SendResult sendMessage(MessageRecordDO messageRecordDO,
                                   ChannelProviderConfigDO channelProviderConfigDO,
                                   MessageProviderFactory messageProviderFactory) {
        ChannelTypeEnum channelType = ChannelTypeEnum.getByCode(messageRecordDO.getChannel());
        return switch (channelType) {
            case SMS -> messageProviderFactory.createSmsSender(channelProviderConfigDO, sender).sendSms(messageRecordDO);
            case EMAIL -> messageProviderFactory.createEmailSender(channelProviderConfigDO, sender)
                    .sendEmail(messageRecordDO, new HashMap<>());
            case APP -> messageProviderFactory.createAppPushSender(channelProviderConfigDO, sender).appPush(messageRecordDO);
        };
    }

    /**
     * sender 只负责调用第三方并返回统一结果；
     * 主流程在这里统一做状态落库、重试编排和健康度更新。
     */
    @Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = Exception.class)
    public void applySendResult(MessageRecordDO messageRecordDO, SendResult sendResult) {
        messageRecordDO.setStatus(this.mapOutcomeToStatus(sendResult.getOutcome()));
        messageRecordDO.setErrorCode(sendResult.getErrorCode());
        messageRecordDO.setErrorMsg(sendResult.getMessage());
        if (sendResult.getSendTime() != null) {
            messageRecordDO.setSendTime(sendResult.getSendTime());
        }
        if (sendResult.getCostTime() != null) {
            messageRecordDO.setCostTime(sendResult.getCostTime());
        }
        if (StrUtil.isNotBlank(sendResult.getThirdPartyId())) {
            messageRecordDO.setThirdPartyId(sendResult.getThirdPartyId());
        }

        // 成功或已受理都视为本次调用已正常投递，更新成功健康度并消耗配额。
        if (SendOutcome.ACCEPTED.equals(sendResult.getOutcome())) {
            messageConfirmRedisDAO.addToConfirmQueue(messageRecordDO);

            systemQuotaRedisDAO.set(messageRecordDO.getChannel(), messageRecordDO.getAppName());
            rateLimitRedisDAO.set(messageRecordDO.getChannel(), messageRecordDO.getAppName(), messageRecordDO.getReceiver());
        } else if (SendOutcome.RETRYABLE_FAILED.equals(sendResult.getOutcome())) {
            messageRetryQueueService.saveOrUpdateRetryMessage(messageRecordDO);
            channelProviderConfigService.failureUpdate(messageRecordDO.getProviderType(), messageRecordDO.getChannel());
        } else if (SendOutcome.FAILED.equals(sendResult.getOutcome())) {
            messageRecordDO.setNextRetryTime(null);
            channelProviderConfigService.failureUpdate(messageRecordDO.getProviderType(), messageRecordDO.getChannel());
        } else {
            // 不支持能力和配置异常都不自动重试。
            messageRecordDO.setNextRetryTime(null);
        }

        this.updateById(messageRecordDO);
    }

    @Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = Exception.class)
    public void recordRetryHistory(MessageRecordDO messageRecordDO, RetrySourceEnum retrySource) {
        MessageRetryHistoryDO messageRetryHistoryDO = BeanUtil.copyProperties(messageRecordDO, MessageRetryHistoryDO.class, "id");
        messageRetryHistoryDO.setMessageId(messageRecordDO.getId());
        messageRetryHistoryDO.setStatus(messageRecordDO.getStatus());
        messageRetryHistoryDO.setRetrySequence(messageRetryHistoryService.getMaxRetrySequence(messageRecordDO.getId()));
        messageRetryHistoryDO.setRetrySource(retrySource != null ? retrySource.getCode() : null);
        messageRetryHistoryService.add(messageRetryHistoryDO);
    }

    private boolean isSuccessOutcome(SendOutcome outcome) {
        return SendOutcome.SUCCESS.equals(outcome) || SendOutcome.ACCEPTED.equals(outcome);
    }

    private String buildBlockedDetail(MessageRecordDO messageRecordDO) {
        if (MsgStatusConstant.BLACKLISTED.equals(messageRecordDO.getStatus())) {
            return "黑名单拦截";
        }
        if (MsgStatusConstant.QUOTAEXCEEDED.equals(messageRecordDO.getStatus())) {
            return "配额超限";
        }
        if (MsgStatusConstant.RATE_LIMITED.equals(messageRecordDO.getStatus())) {
            return "限流拦截";
        }
        return StrUtil.blankToDefault(messageRecordDO.getErrorMsg(), "消息状态不允许发送");
    }

    private String buildResultDetail(SendResult sendResult) {
        if (StrUtil.isNotBlank(sendResult.getMessage())) {
            return sendResult.getMessage();
        }
        return sendResult.getOutcome() == null ? "发送失败" : sendResult.getOutcome().getDesc();
    }

    private String mapOutcomeToStatus(SendOutcome outcome) {
        if (outcome == null) {
            return MsgStatusConstant.FAILED;
        }
        return switch (outcome) {
            case SUCCESS -> MsgStatusConstant.SUCCESS;
            case ACCEPTED -> MsgStatusConstant.SENDING;
            case FAILED -> MsgStatusConstant.FAILED;
            case RETRYABLE_FAILED -> MsgStatusConstant.RETRYABLE_FAILED;
            case UNSUPPORTED_CHANNEL -> MsgStatusConstant.UNSUPPORTED_CHANNEL;
            case CONFIG_INVALID -> MsgStatusConstant.CONFIG_INVALID;
        };
    }


    @Override
    public boolean updateRetryCount(Long id) {
        MessageRecordDO one = this.lambdaQuery().eq(MessageRecordDO::getId, id).one();
        one.setRetryCount(one.getRetryCount() + 1);
        return this.updateById(one);
    }

    @Override
    public boolean add(MessageRecordReqVO messageRecordSendReqVO) {
        MessageRecordDO messageRecordDO = BeanUtil.copyProperties(messageRecordSendReqVO, MessageRecordDO.class);
        return this.save(messageRecordDO);
    }

    @Override
    public Page<MessageRecordDO> getPage(MessageRecordReqVO reqVO) {
        QueryWrapper<MessageRecordDO> wrapper = new QueryWrapper<>();
        wrapper.lambda()
                .eq(StrUtil.isNotBlank(reqVO.getMessageType()), MessageRecordDO::getMessageType, reqVO.getMessageType());
        return this.page(new Page<>(PageUtils.getPageNum(reqVO), PageUtils.getPageSize(reqVO)), wrapper);
    }
}
