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

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.njcn.msgpush.framework.common.pojo.PageResult;
import com.njcn.msgpush.module.push.constant.MsgStatusConstant;
import com.njcn.msgpush.module.push.controller.admin.message.vo.MessageSendResultVO;
import com.njcn.msgpush.module.push.controller.admin.retry.vo.MessageRetryQueueReqVO;
import com.njcn.msgpush.module.push.dal.dataobject.message.MessageRecordDO;
import com.njcn.msgpush.module.push.dal.dataobject.retry.MessageRetryQueueDO;
import com.njcn.msgpush.module.push.dal.dataobject.retry.RetryStrategyConfigDO;
import com.njcn.msgpush.module.push.dal.mysql.retry.MessageRetryQueueMapper;
import com.njcn.msgpush.module.push.dal.redis.MessageConfirmRedisDAO;
import com.njcn.msgpush.module.push.dal.redis.MessageRetryRedisDAO;
import com.njcn.msgpush.module.push.enums.ChannelTypeEnum;
import com.njcn.msgpush.module.push.service.message.MessageRecordService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.time.LocalDateTime;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;

@Slf4j
@Service
public class MessageRetryQueueServiceImpl extends ServiceImpl<MessageRetryQueueMapper, MessageRetryQueueDO> implements MessageRetryQueueService, CommandLineRunner {
    @Autowired
    private MessageRetryRedisDAO messageRetryRedisDAO;
    @Autowired
    private MessageRecordService messageRecordService;
    @Autowired
    public RetryStrategyConfigService retryStrategyConfigService;
    @Autowired
    private MessageConfirmRedisDAO messageConfirmRedisDAO;

    private static final int DEFAULT_BATCH_SIZE = 100;

    private static final int DEFAULT_MAX_RETRY_COUNT = 5;

    private static final List<String> CHANNELS = Arrays.asList(
            ChannelTypeEnum.SMS.getCode(),
            ChannelTypeEnum.EMAIL.getCode(),
            ChannelTypeEnum.APP_PUSH.getCode());

    private final AtomicBoolean startupRebuildCompleted = new AtomicBoolean(false);

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void saveOrUpdateRetryMessage(MessageRecordDO message) {
        MessageRetryQueueDO existing = super.baseMapper.selectOne(
                new LambdaQueryWrapper<MessageRetryQueueDO>()
                        .eq(MessageRetryQueueDO::getMessageId, message.getId())
        );
        message.setLastRetryTime(message.getSendTime());

        if (ObjectUtil.isNull(existing)) {
            MessageRetryQueueDO retryRecord = new MessageRetryQueueDO();
            retryRecord.setMessageId(message.getId());
            retryRecord.setChannel(message.getChannel());
            retryRecord.setReceiver(message.getReceiver());

            retryRecord.setRetryCount(1);
            RetryStrategyConfigDO strategyConfig = retryStrategyConfigService.getStrategyConfig(message.getChannel());
            retryRecord.setMaxRetry(ObjectUtil.isNull(strategyConfig) ? DEFAULT_MAX_RETRY_COUNT : strategyConfig.getMaxRetryCount());
            retryRecord.setFirstFailTime(message.getSendTime());
            retryRecord.setLastRetryTime(message.getSendTime());
            retryRecord.setLastErrorMsg(message.getErrorMsg());
            LocalDateTime nextRetryTime = this.calculateNextRetryTime(message.getChannel(), retryRecord.getRetryCount());
            retryRecord.setNextRetryTime(nextRetryTime);
            message.setNextRetryTime(nextRetryTime);
            this.save(retryRecord);
            if (!messageRetryRedisDAO.addToRetryQueue(message)) {
                log.warn("消息已写入数据库重试队列，但 Redis 入队失败: messageId={}", message.getId());
            }
            return;
        }

        int newRetryCount = existing.getRetryCount() + 1;
        existing.setLastRetryTime(message.getSendTime());
        existing.setLastErrorMsg(message.getErrorMsg());

        if (newRetryCount >= existing.getMaxRetry()) {
            message.setStatus(MsgStatusConstant.FINALFAILED);
            message.setNextRetryTime(null);
            existing.setNextRetryTime(null);
            this.updateById(existing);
            this.deleteByMessageIds(Collections.singletonList(message.getId()));
            messageRetryRedisDAO.removeFromRetryQueue(message.getChannel(), message.getId());
            return;
        }

        LocalDateTime nextRetryTime = this.calculateNextRetryTime(message.getChannel(), newRetryCount);
        existing.setRetryCount(newRetryCount);
        existing.setNextRetryTime(nextRetryTime);
        message.setNextRetryTime(nextRetryTime);
        this.updateById(existing);
        if (!messageRetryRedisDAO.addToRetryQueue(message)) {
            log.warn("消息已更新数据库重试队列，但 Redis 入队失败: messageId={}", message.getId());
        }
    }

    private LocalDateTime calculateNextRetryTime(String channel, int retryCount) {
        RetryStrategyConfigDO strategyConfig = retryStrategyConfigService.getStrategyConfig(channel);
        long plusSeconds = 0;
        if (ObjectUtil.isNull(strategyConfig)) {
            switch (ChannelTypeEnum.getByCode(channel)) {
                case SMS -> {
                    if (retryCount == 1) {
                        plusSeconds = 60 * 5;
                    } else if (retryCount == 2) {
                        plusSeconds = 60 * 10;
                    } else {
                        plusSeconds = 60 * 30;
                    }
                }
                case EMAIL -> {
                    if (retryCount == 1) {
                        plusSeconds = 60 * 2;
                    } else if (retryCount == 2) {
                        plusSeconds = 60 * 30;
                    } else if (retryCount == 3) {
                        plusSeconds = 60 * 60;
                    } else if (retryCount == 4) {
                        plusSeconds = 60 * 60 * 2;
                    } else {
                        plusSeconds = 60 * 60 * 4;
                    }
                }
                case APP_PUSH -> {
                    if (retryCount == 1) {
                        plusSeconds = 60;
                    } else {
                        plusSeconds = 60 * 60;
                    }
                }
            }
        } else {
            String retryIntervals = strategyConfig.getRetryIntervals();
            String[] split = retryIntervals.split(String.valueOf(StrUtil.C_COMMA));
            if (retryCount >= split.length) {
                plusSeconds = Long.parseLong(split[split.length - 1]);
            } else {
                plusSeconds = Long.parseLong(split[retryCount - 1]);
            }
        }
        return LocalDateTime.now().plusSeconds(plusSeconds);
    }

    @Override
    public void processRetryBatch(String channel) {
        long epochMilli = System.currentTimeMillis();
        // 使用原子认领机制，一次性完成读取 + 删除
        List<String> claimedMessageIds = messageRetryRedisDAO.claimAndRemoveDueMessages(channel, epochMilli, DEFAULT_BATCH_SIZE);

        if (CollUtil.isEmpty(claimedMessageIds)) {
            log.debug("Redis 中没有待重试消息：channel={}", channel);
            fallbackToDatabase(channel);

            epochMilli = System.currentTimeMillis();
            claimedMessageIds = messageRetryRedisDAO.claimAndRemoveDueMessages(channel, epochMilli, DEFAULT_BATCH_SIZE);

            if (CollUtil.isEmpty(claimedMessageIds)) {
                return;
            }
        }

        List<MessageRecordDO> messageRecordDOList = messageRecordService.listByIds(claimedMessageIds);
        for (MessageRecordDO messageRecordDO : messageRecordDOList) {
            processSingleRetry(messageRecordDO);
        }
    }

    /**
     * 处理单条消息重试，根据执行结果决定是否重新入队
     * 成功：不再回 Redis，并逻辑删除数据库队列记录
     * 失败且继续重试：重新计算 nextRetryTime，重新写入 Redis 和数据库队列表
     * 失败且终止：不再回 Redis，并逻辑删除数据库队列记录
     *
     * @param messageRecordDO 消息记录
     */
    private void processSingleRetry(MessageRecordDO messageRecordDO) {
        messageRecordDO.setStatus(MsgStatusConstant.PENDING);
        messageRecordService.processSendMsg(Collections.singletonList(messageRecordDO)).get(0);
        messageRecordService.updateRetryCount(messageRecordDO.getId());
    }

    /**
     * 删除重试记录（数据库逻辑删除 + Redis 已提前删除）
     */
    private void deleteRetryRecord(MessageRecordDO message) {
        this.deleteByMessageIds(Collections.singletonList(message.getId()));
        messageRecordService.updateById(message);
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void manualRetry(List<String> messageIds) {
        if (CollUtil.isNotEmpty(messageIds)) {
            List<MessageRecordDO> messageRecordDOList = messageRecordService.listByIds(messageIds);
            for (MessageRecordDO messageRecordDO : messageRecordDOList) {
                processSingleRetry(messageRecordDO);
            }
        }
    }

    @Override
    public PageResult<MessageRetryQueueDO> getPage(MessageRetryQueueReqVO reqVO) {
        LambdaQueryWrapper<MessageRetryQueueDO> queryWrapper = new LambdaQueryWrapper<MessageRetryQueueDO>()
                .eq(ObjectUtil.isNotNull(reqVO.getMessageId()), MessageRetryQueueDO::getMessageId, reqVO.getMessageId())
                .eq(StrUtil.isNotBlank(reqVO.getChannel()), MessageRetryQueueDO::getChannel, reqVO.getChannel())
                .eq(StrUtil.isNotBlank(reqVO.getReceiver()), MessageRetryQueueDO::getReceiver, reqVO.getReceiver())
                .ge(ObjectUtil.isNotNull(reqVO.getMinRetryCount()), MessageRetryQueueDO::getRetryCount, reqVO.getMinRetryCount())
                .le(ObjectUtil.isNotNull(reqVO.getMaxRetryCount()), MessageRetryQueueDO::getRetryCount, reqVO.getMaxRetryCount())
                .orderByAsc(MessageRetryQueueDO::getNextRetryTime)
                .orderByDesc(MessageRetryQueueDO::getId);
        return this.baseMapper.selectPage(reqVO, queryWrapper);
    }

    @Override
    public boolean deleteByMessageIds(List<Long> messageIds) {
        if (CollUtil.isEmpty(messageIds)) {
            return false;
        }

        return this.remove(new LambdaQueryWrapper<MessageRetryQueueDO>()
                .in(MessageRetryQueueDO::getMessageId, messageIds));
    }


    @Override
    public void run(String... args) throws Exception {
        rebuildRetryQueueOnStartup();
        startupRebuildCompleted.set(true);
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void rebuildRetryQueueOnStartup() {
        log.info("========== 开始重建重试队列 ==========");

        for (String channel : CHANNELS) {
            try {
                log.info("正在清空渠道 [{}] 的 Redis 重试队列", channel);
                messageRetryRedisDAO.clearRetryQueue(channel);

                log.info("正在从数据库加载渠道 [{}] 的待重试消息", channel);
                List<MessageRetryQueueDO> dbRecords = this.lambdaQuery()
                        .eq(MessageRetryQueueDO::getChannel, channel)
                        .isNotNull(MessageRetryQueueDO::getNextRetryTime)
                        .list();

                if (CollUtil.isEmpty(dbRecords)) {
                    log.info("渠道 [{}] 没有待重试的消息", channel);
                    continue;
                }

                List<Long> messageIds = dbRecords.stream().map(MessageRetryQueueDO::getMessageId).collect(Collectors.toList());
                List<MessageRecordDO> messages = messageRecordService.listByIds(messageIds);

                if (CollUtil.isEmpty(messages)) {
                    log.warn("渠道 [{}] 的消息记录不存在", channel);
                    continue;
                }

                int successCount = messageRetryRedisDAO.batchAddToRetryQueue(messages);
                log.info("渠道 [{}] 重建完成：数据库记录数={}, 成功入队 Redis 数={}",
                        channel, dbRecords.size(), successCount);

            } catch (Exception e) {
                log.error("重建渠道 [{}] 的重试队列失败", channel, e);
            }
        }

        log.info("========== 重试队列重建完成 ==========");
    }

    @Scheduled(fixedRate = 600000)
    @Override
    public void syncRetryQueueConsistency() {
        if (!startupRebuildCompleted.get()) {
            log.debug("启动重建未完成，跳过本次同步");
            return;
        }
        log.info("开始同步redis与数据库重试队列一致性");

        for (String channel : CHANNELS) {
            List<MessageRetryQueueDO> dbRecords = this.lambdaQuery()
                    .eq(MessageRetryQueueDO::getChannel, channel)
                    .isNotNull(MessageRetryQueueDO::getNextRetryTime)
                    .list();

            if (CollUtil.isEmpty(dbRecords)) {
                continue;
            }

            for (MessageRetryQueueDO dbRecord : dbRecords) {
                boolean existsInRedis = messageRetryRedisDAO.existsInRetryQueue(channel, dbRecord.getMessageId());

                if (!existsInRedis) {
                    Long messageId = dbRecord.getMessageId();
                    MessageRecordDO message = messageRecordService.getById(messageId);

                    if (message != null && message.getNextRetryTime() != null) {
                        boolean added = messageRetryRedisDAO.addToRetryQueue(message);
                        if (added) {
                            log.debug("补回消息到 Redis: channel={}, messageId={}", channel, messageId);
                        } else {
                            log.warn("补回消息到 Redis 失败：channel={}, messageId={}", channel, messageId);
                        }
                    } else {
                        log.warn("消息记录不存在或 next_retry_time 为空：channel={}, messageId={}", channel, messageId);
                    }
                }
            }
        }
    }

    @Override
    public void fallbackToDatabase(String channel) {
        boolean redisAvailable = messageRetryRedisDAO.isRedisAvailable();
        if (!redisAvailable) {
            log.warn("Redis 不可用，channel={}", channel);
            return;
        }

        fallbackToDatabaseDirectly(channel);
    }


    /**
     * 从数据库补拉
     *
     * @param channel
     */
    private void fallbackToDatabaseDirectly(String channel) {
        List<MessageRetryQueueDO> dbRecords = baseMapper.selectNeedRetryMessages(LocalDateTime.now(), DEFAULT_BATCH_SIZE);

        if (CollUtil.isEmpty(dbRecords)) {
            return;
        }

        List<Long> messageIds = dbRecords.stream().map(MessageRetryQueueDO::getMessageId).collect(Collectors.toList());
        List<MessageRecordDO> messages = messageRecordService.listByIds(messageIds);

        if (CollUtil.isEmpty(messages)) {
            return;
        }

        if (messageRetryRedisDAO.isRedisAvailable()) {
            messageRetryRedisDAO.batchAddToRetryQueue(messages);
        }
    }
}
