package com.njcn.msgpush.module.push.dal.redis;

import com.njcn.msgpush.module.push.dal.dataobject.message.MessageRecordDO;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Component;

import java.time.ZoneId;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;

/**
 * {@link com.njcn.msgpush.module.push.dal.dataobject.retry.MessageRetryQueueDO} RedisDAO
 *
 * @author caozehui
 * @data 2026-02-27
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class MessageRetryRedisDAO {
    private final RedisTemplate<String, String> redisTemplate;

    /**
     * Redis中消息重试队列的Key前缀
     */
    private static final String RETRY_QUEUE_KEY_PREFIX = "msPush:retry_queue:";

    /**
     * 获取指定渠道的重试队列Key
     */
    private String getRetryQueueKey(String channel) {
        return RETRY_QUEUE_KEY_PREFIX + channel;
    }

    /**
     * 原子认领并移除到期消息（Lua 脚本保证原子性）
     * 适用于单实例场景，直接删除不再单独维护认领状态
     *
     * @param channel     渠道类型
     * @param currentTime 当前时间戳
     * @param limit       限制数量
     * @return 认领成功的消息 ID 列表
     */
    public List<String> claimAndRemoveDueMessages(String channel, long currentTime, int limit) {
        String key = getRetryQueueKey(channel);

        String luaScript =
                "local key = KEYS[1] " +
                        "local currentTime = tonumber(ARGV[1]) " +
                        "local limit = tonumber(ARGV[2]) " +
                        "local dueMessages = redis.call('ZRANGEBYSCORE', key, 0, currentTime, 'LIMIT', 0, limit) " +
                        "if #dueMessages == 0 then return {} end " +
                        "for _, messageId in ipairs(dueMessages) do " +
                        "    redis.call('ZREM', key, messageId) " +
                        "end " +
                        "return dueMessages";

        DefaultRedisScript<List> redisScript = new DefaultRedisScript<>();
        redisScript.setScriptText(luaScript);
        redisScript.setResultType(List.class);

        try {
            List<String> claimedMessageIds = redisTemplate.execute(
                    redisScript,
                    Arrays.asList(key),
                    String.valueOf(currentTime),
                    String.valueOf(limit)
            );

            log.debug("原子认领并移除到期消息成功：channel={}, 认领数量={}", channel,
                    claimedMessageIds != null ? claimedMessageIds.size() : 0);

            return claimedMessageIds != null ? claimedMessageIds : Collections.emptyList();
        } catch (Exception e) {
            return Collections.emptyList();
        }
    }


    /**
     * 添加消息到重试队列
     *
     * @param message 消息
     */
    public boolean addToRetryQueue(MessageRecordDO message) {
        String key = getRetryQueueKey(message.getChannel());
        double score = message.getNextRetryTime().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();

        return redisTemplate.opsForZSet().add(key, message.getId() + "", score);
    }

    /**
     * 从重试队列移除消息
     *
     * @param channel   渠道类型
     * @param messageId 消息ID
     */
    public boolean removeFromRetryQueue(String channel, Long messageId) {
        String key = getRetryQueueKey(channel);

        redisTemplate.opsForZSet().remove(key, messageId);
        return true;
    }


    /**
     * 清空指定渠道的重试队列
     *
     * @param channel 渠道类型
     */
    public void clearRetryQueue(String channel) {
        String key = getRetryQueueKey(channel);
        redisTemplate.delete(key);
    }

    /**
     * 检查消息是否在重试队列中
     *
     * @param channel   渠道类型
     * @param messageId 消息 ID
     * @return 是否存在
     */
    public boolean existsInRetryQueue(String channel, Long messageId) {
        String key = getRetryQueueKey(channel);
        try {
            Double score = redisTemplate.opsForZSet().score(key, messageId + "");
            return score != null;
        } catch (Exception e) {
            return false;
        }
    }

    /**
     * 批量添加消息到重试队列
     *
     * @param messages 消息列表
     * @return 成功添加的数量
     */
    public int batchAddToRetryQueue(List<MessageRecordDO> messages) {
        if (messages == null || messages.isEmpty()) {
            return 0;
        }

        int successCount = 0;
        for (MessageRecordDO message : messages) {
            if (addToRetryQueue(message)) {
                successCount++;
            }
        }
        return successCount;
    }

    /**
     * 检查 Redis 是否可用
     *
     * @return Redis 是否可用
     */
    public boolean isRedisAvailable() {
        try {
            String pong = redisTemplate.execute((RedisCallback<String>) connection ->
                    new String(connection.ping()));
            return "PONG".equals(pong);
        } catch (Exception e) {
            log.error("Redis 不可用", e);
            return false;
        }
    }
}
