package com.tongran.rocketmq.consumer;

import com.alibaba.fastjson.JSON;
import com.tongran.common.core.enums.MsgEnum;
import com.tongran.rocketmq.domain.DeviceMessage;
import com.tongran.rocketmq.domain.InitialBandwidthTraffic;
import com.tongran.rocketmq.enums.MessageCodeEnum;
import com.tongran.rocketmq.handler.MessageHistoryDataHandler;
import com.tongran.rocketmq.utils.JsonDataParser;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.text.SimpleDateFormat;
import java.time.Instant;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 时间范围重消费消费者（两阶段优化版）
 * 第一阶段：并行消费并缓存所有消息
 * 第二阶段：按设备顺序处理缓存的消息
 */
@Slf4j
@Component
public class TimeRangeReConsumer {


    @Autowired(required = false)
    private MessageHistoryDataHandler messageHandler;

    // 时间范围配置
    private static final String START_TIME = "2026-02-02 06:00:00";
    private static final String END_TIME = "2026-02-02 12:00:00";

    // 统计信息
    private final AtomicLong receivedCount = new AtomicLong(0);  // 已接收消息数
    private final AtomicLong skippedCount = new AtomicLong(0);
    private final AtomicLong processedCount = new AtomicLong(0); // 已处理消息数
    private final Map<Integer, Long> queueCurrentOffsets = new ConcurrentHashMap<>();
    private final Set<Integer> completedQueues = ConcurrentHashMap.newKeySet();

    // 控制是否继续消费
    private AtomicBoolean shouldContinue = new AtomicBoolean(true);
    private AtomicBoolean isProcessingPhase = new AtomicBoolean(false); // 第二阶段处理标记

    // 消息缓存：按clientId分组，按时间戳排序
    private final ConcurrentHashMap<String, PriorityBlockingQueue<MessageWrapper>> messageCache =
            new ConcurrentHashMap<>();

    // 第二阶段顺序处理器线程池
    private final ExecutorService sequentialProcessor = Executors.newFixedThreadPool(
            Math.min(Runtime.getRuntime().availableProcessors() * 4, 32)
    );

    // 第二阶段处理跟踪
    private final Map<String, CompletableFuture<Void>> processingFutures = new ConcurrentHashMap<>();
    private final AtomicInteger activeSequentialTasks = new AtomicInteger(0);

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class MessageWrapper implements Comparable<MessageWrapper> {
        private String msgId;
        private DeviceMessage message;
        private MessageExt messageExt;
        private long timestamp;  // 从data中解析的时间戳
        private long bornTimestamp;  // 消息产生时间

        @Override
        public int compareTo(MessageWrapper other) {
            return Long.compare(this.timestamp, other.timestamp);
        }
    }


    // 添加一个标识字段
    private final AtomicBoolean isRunning = new AtomicBoolean(false);

    // 暴露启动方法
    public String startReconsumeTask(String startTime, String endTime, String namesrvAddr) {
        if (isRunning.compareAndSet(false, true)) {
            new Thread(() -> {
                try {
                    doReconsume(startTime, endTime, namesrvAddr);
                } catch (Exception e) {
                    log.error("重消费失败", e);
                } finally {
                    isRunning.set(false);
                }
            }).start();
            return "重消费任务已启动";
        }
        return "已有任务正在运行";
    }
    public void startTimeRangeConsumer(String startTime, String endTime, String namesrvAddr) {
        new Thread(() -> {
            try {
                doReconsume(startTime, endTime, namesrvAddr);
            } catch (Exception e) {
                log.error("重消费失败", e);
            }
        }).start();
    }

    private void doReconsume(String startTime, String endTime, String namesrvAddr) throws Exception {
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        long startTimestamp = sdf.parse(startTime).getTime();
        long endTimestamp = sdf.parse(endTime).getTime();

        String consumerGroup = "RECONSUME_GROUP_TEST" + Instant.now().getEpochSecond();
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setNamesrvAddr(namesrvAddr);

        // ================== 第一阶段：开始消息缓存 ==================
        log.info("===== 第一阶段：开始消息缓存 =====");

        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_TIMESTAMP);
        consumer.setConsumeTimestamp(sdf.format(new Date(startTimestamp)));

        consumer.subscribe(MessageCodeEnum.TR_AGENT_UP.getCode(), "*");

        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(
                    List<MessageExt> list,
                    ConsumeConcurrentlyContext context) {

                if (!shouldContinue.get()) {
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }

                try {
                    for (MessageExt messageExt : list) {
                        long bornTimestamp = messageExt.getBornTimestamp();
                        int queueId = messageExt.getQueueId();

                        // 记录当前队列的offset
                        queueCurrentOffsets.put(queueId, messageExt.getQueueOffset());

                        // 1. 时间太早，跳过
                        if (bornTimestamp < startTimestamp) {
                            skippedCount.incrementAndGet();
                            if (skippedCount.get() % 1000 != 0) {
                                continue;
                            }
                        }

                        // 2. 超过结束时间，标记队列完成
                        if (bornTimestamp > endTimestamp) {
                            log.info("队列 {} 已超过结束时间，标记为完成", queueId);
                            completedQueues.add(queueId);

                            // 检查是否所有队列都完成
                            if (completedQueues.size() >= 4) { // 假设有4个队列
                                log.info("所有队列均已超过结束时间，第一阶段完成");
                                shouldContinue.set(false);
                            }
                            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                        }

                        String body = new String(messageExt.getBody(), StandardCharsets.UTF_8);
                        DeviceMessage message = JSON.parseObject(body, DeviceMessage.class);
                        String dataType = message.getDataType();
                        String clientId = message.getClientId();
                        // 只处理"网络上报重试"类型的消息
                        if(MsgEnum.网络上报重试.getValue().equals(dataType)){

                            receivedCount.incrementAndGet();

                            // 只缓存消息，不处理
                            cacheMessageOnly(message, messageExt, bornTimestamp, sdf);

                            // 定期打印进度
                            if (receivedCount.get() % 100 == 0) {
                                printPhase1Progress();
                            }
                        }
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                } catch (Exception e) {
                    log.error("重消费消息异常", e);
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
            }
        });

        // 启动消费者
        consumer.start();
        log.info("时间范围重消费者启动成功，消费者组: {}", consumerGroup);

        // 预热：查找大致的起始位置
        findApproximateStartOffset(consumer, startTimestamp);

        // 监控第一阶段完成
        waitForPhase1Completion(consumer);

        // 第一阶段完成，关闭消费者
        consumer.shutdown();
        log.info("===== 第一阶段完成，共缓存{}条消息，涉及{}个设备 =====",
                receivedCount.get(), messageCache.size());

        // ================== 第二阶段：开始顺序处理 ==================
        log.info("===== 第二阶段：开始顺序处理 =====");
        startPhase2Processing();

        // 等待第二阶段完成
        waitForPhase2Completion();

        log.info("✓ 时间范围重消费完成，总计接收消息: {}, 处理消息: {}",
                receivedCount.get(), processedCount.get());
    }

    /**
     * 第一阶段：只缓存消息，不处理
     */
    private void cacheMessageOnly(DeviceMessage message, MessageExt messageExt,
                                  long bornTimestamp, SimpleDateFormat sdf) {
        try {
            // 解析data中的时间戳
            long timestamp = extractTimestampFromData(message.getData());
            String clientId = message.getClientId();

            // 创建消息包装
            MessageWrapper wrapper = new MessageWrapper(
                    messageExt.getMsgId(),
                    message,
                    messageExt,
                    timestamp,
                    bornTimestamp
            );

            // 获取或创建该客户端的优先级队列
            PriorityBlockingQueue<MessageWrapper> queue = messageCache.computeIfAbsent(
                    clientId,
                    id -> new PriorityBlockingQueue<>()
            );

            // 添加到队列（会自动按时间戳排序）
            queue.offer(wrapper);

            if (log.isDebugEnabled() && receivedCount.get() % 20 == 0) {
                log.debug("消息已缓存: clientId={}, timestamp={}, bornTime={}, 队列大小={}",
                        clientId,
                        timestamp,
                        sdf.format(new Date(bornTimestamp)),
                        queue.size());
            }

        } catch (Exception e) {
            log.error("缓存消息失败: msgId={}", messageExt.getMsgId(), e);
        }
    }

    /**
     * 从data字段解析时间戳
     */
    private long extractTimestampFromData(String data) {
        try {
            List<InitialBandwidthTraffic> interfaces = JsonDataParser.parseJsonData(data, InitialBandwidthTraffic.class);
            if(!interfaces.isEmpty()) {
                // 时间戳转换
                long timestamp = interfaces.get(0).getTimestamp();
                return timestamp;
            }
        } catch (Exception e) {
            log.warn("解析data时间戳失败，使用默认值", e);
        }
        return System.currentTimeMillis(); // 解析失败使用当前时间
    }

    /**
     * 等待第一阶段（消息缓存）完成
     */
    private void waitForPhase1Completion(DefaultMQPushConsumer consumer) {
        log.info("等待第一阶段完成（消息缓存完成）...");

        int checkCount = 0;
        long lastMessageCount = 0;
        int noProgressCount = 0;

        while (shouldContinue.get()) {
            try {
                Thread.sleep(5000);
                checkCount++;

                long currentMessageCount = receivedCount.get();
                int totalCached = 0;
                for (PriorityBlockingQueue<MessageWrapper> queue : messageCache.values()) {
                    totalCached += queue.size();
                }

                // 每30秒打印一次进度
                if (checkCount % 6 == 0) {
                    log.info("第一阶段进度: 已接收={}, 已缓存={}, 设备数={}, 完成队列={}/4",
                            currentMessageCount, totalCached, messageCache.size(), completedQueues.size());
                }

                // 检查是否有进度
                if (currentMessageCount == lastMessageCount) {
                    noProgressCount++;
                } else {
                    noProgressCount = 0;
                    lastMessageCount = currentMessageCount;
                }

                // 如果30秒内没有新消息，认为第一阶段完成
                if (noProgressCount >= 6 && completedQueues.size() >= 4) {
                    log.info("30秒内无新消息且所有队列完成，第一阶段完成");
                    shouldContinue.set(false);
                    break;
                }

                // 超时保护：最多等待1小时
                if (checkCount >= 720) {
                    log.warn("第一阶段超时（1小时），强制结束");
                    shouldContinue.set(false);
                    break;
                }

            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                break;
            }
        }
    }

    /**
     * 第二阶段：开始顺序处理所有缓存的消息
     */
    private void startPhase2Processing() {
        isProcessingPhase.set(true);

        int totalMessages = messageCache.values().stream()
                .mapToInt(Queue::size)
                .sum();

        log.info("开始顺序处理{}个设备的{}条消息...",
                messageCache.size(), totalMessages);

        // 为每个设备启动顺序处理任务
        for (Map.Entry<String, PriorityBlockingQueue<MessageWrapper>> entry : messageCache.entrySet()) {
            String clientId = entry.getKey();
            PriorityBlockingQueue<MessageWrapper> queue = entry.getValue();

            if (!queue.isEmpty()) {
                CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
                    activeSequentialTasks.incrementAndGet();
                    try {
                        processDeviceMessagesSequentially(clientId, queue);
                    } finally {
                        activeSequentialTasks.decrementAndGet();
                    }
                }, sequentialProcessor);

                processingFutures.put(clientId, future);
            }
        }

        log.info("已启动{}个设备的顺序处理任务", processingFutures.size());
    }

    /**
     * 按顺序处理单个设备的所有消息
     */
    private void processDeviceMessagesSequentially(String clientId,
                                                   PriorityBlockingQueue<MessageWrapper> queue) {
        String threadName = Thread.currentThread().getName();
        Thread.currentThread().setName("SeqProcessor-" + clientId.substring(0, Math.min(8, clientId.length())));

        try {
            int queueSize = queue.size();
            log.info("开始顺序处理设备{}的{}条消息", clientId, queueSize);

            int deviceProcessed = 0;
            MessageWrapper lastMessage = null;
            long startTime = System.currentTimeMillis();

            while (!queue.isEmpty()) {
                MessageWrapper wrapper = queue.poll();
                if (wrapper != null) {
                    try {
                        // 验证时间顺序（可选）
                        if (lastMessage != null && wrapper.getTimestamp() < lastMessage.getTimestamp()) {
                            log.warn("设备{}消息时间顺序异常: 前一条timestamp={}, 当前timestamp={}",
                                    clientId, lastMessage.getTimestamp(), wrapper.getTimestamp());
                        }

                        // 处理消息
                        if (messageHandler != null) {
                            messageHandler.handleMessage(wrapper.getMessage());
                        }

                        deviceProcessed++;
                        processedCount.incrementAndGet();
                        lastMessage = wrapper;

                        // 每处理20条日志一次
                        if (deviceProcessed % 20 == 0) {
                            log.debug("设备{}已处理{}条消息，剩余{}条",
                                    clientId, deviceProcessed, queue.size());
                        }

                    } catch (Exception e) {
                        log.error("处理设备{}消息失败: msgId={}", clientId, wrapper.getMsgId(), e);
                        // 根据业务决定是否重试或跳过
                        // 这里选择跳过，记录错误
                    }
                }
            }

            long timeCost = System.currentTimeMillis() - startTime;
            log.info("设备{}处理完成: 共处理{}条消息，耗时{}ms",
                    clientId, deviceProcessed, timeCost);

        } finally {
            Thread.currentThread().setName(threadName);
        }
    }

    /**
     * 等待第二阶段处理完成
     */
    private void waitForPhase2Completion() {
        log.info("等待所有设备顺序处理完成...");

        CompletableFuture<Void> allFutures = CompletableFuture.allOf(
                processingFutures.values().toArray(new CompletableFuture[0])
        );

        try {
            // 等待所有任务完成
            allFutures.get(2, TimeUnit.HOURS); // 最多等待2小时

            log.info("✓ 所有设备顺序处理完成");

        } catch (TimeoutException e) {
            log.warn("第二阶段处理超时（2小时），强制结束");
            // 可以记录未完成的任务
            List<String> unfinishedDevices = new ArrayList<>();
            for (Map.Entry<String, CompletableFuture<Void>> entry : processingFutures.entrySet()) {
                if (!entry.getValue().isDone()) {
                    unfinishedDevices.add(entry.getKey());
                }
            }
            log.warn("以下设备未完成处理: {}", unfinishedDevices);

        } catch (Exception e) {
            log.error("等待第二阶段处理异常", e);
        } finally {
            shutdownResources();
        }

        // 最后统计
        int remainingMessages = messageCache.values().stream()
                .mapToInt(Queue::size)
                .sum();

        if (remainingMessages > 0) {
            log.warn("仍有{}条消息未处理", remainingMessages);
        } else {
            log.info("所有消息处理完成，清理缓存");
            messageCache.clear();
        }
    }
    /**
     * 处理完成后关闭资源 - 添加状态重置
     */
    private void shutdownResources() {
        log.info("重消费完成，正在关闭资源...");

        shouldContinue.set(false);
        isProcessingPhase.set(false);
        isRunning.set(false);

        // 清理缓存
        messageCache.clear();
        processingFutures.clear();

        // 关闭线程池
        sequentialProcessor.shutdown();
        try {
            if (!sequentialProcessor.awaitTermination(10, TimeUnit.SECONDS)) {
                sequentialProcessor.shutdownNow();
            }
        } catch (InterruptedException e) {
            sequentialProcessor.shutdownNow();
            Thread.currentThread().interrupt();
        }

        // 必须重置计数器和集合
        receivedCount.set(0);
        processedCount.set(0);
        skippedCount.set(0);
        queueCurrentOffsets.clear();
        completedQueues.clear();

        log.info("✓ 重消费者资源已完全释放");
    }
    /**
     * 查找大致的起始offset
     */
    private void findApproximateStartOffset(DefaultMQPushConsumer consumer, long startTimestamp) throws Exception {
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        log.info("查找起始offset，目标时间: {}", sdf.format(new Date(startTimestamp)));

        try {
            // 获取主题队列
            Set<MessageQueue> queues = consumer.fetchSubscribeMessageQueues(MessageCodeEnum.TR_AGENT_UP.getCode());

            if (queues == null || queues.isEmpty()) {
                log.info("无队列信息，跳过offset查找");
                return;
            }

            log.info("找到{}个队列，开始设置起始offset...", queues.size());

            for (MessageQueue queue : queues) {
                try {
                    // 搜索指定时间戳的offset
                    long offset = consumer.searchOffset(queue, startTimestamp);

                    if (offset >= 0) {
                        // 获取队列的offset范围
                        long minOffset = consumer.minOffset(queue);
                        long maxOffset = consumer.maxOffset(queue);

                        // 确保offset在有效范围内
                        if (offset < minOffset) offset = minOffset;
                        if (offset > maxOffset) offset = maxOffset;

                        queueCurrentOffsets.put(queue.getQueueId(), offset);
                        log.info("队列{}: offset={} (范围: {}-{})",
                                queue.getQueueId(), offset, minOffset, maxOffset);
                    } else {
                        // 使用最小offset
                        long minOffset = consumer.minOffset(queue);
                        queueCurrentOffsets.put(queue.getQueueId(), minOffset);
                        log.info("队列{}: 使用最小offset={}", queue.getQueueId(), minOffset);
                    }
                } catch (Exception e) {
                    log.warn("队列{}设置offset失败: {}", queue.getQueueId(), e.getMessage());
                }
            }

            log.info("起始offset设置完成");

        } catch (Exception e) {
            log.warn("查找起始offset失败，将使用默认时间戳方式: {}", e.getMessage());
        }
    }

    /**
     * 打印第一阶段进度
     */
    private void printPhase1Progress() {
        int totalCached = 0;
        for (PriorityBlockingQueue<MessageWrapper> queue : messageCache.values()) {
            totalCached += queue.size();
        }

        log.info("第一阶段进度: 已接收={}, 已缓存={}, 设备数={}, 完成队列={}/4",
                receivedCount.get(), totalCached, messageCache.size(), completedQueues.size());
    }

    /**
     * 打印第二阶段进度
     */
    private void printPhase2Progress() {
        int totalRemaining = 0;
        int activeTasks = 0;

        for (Map.Entry<String, PriorityBlockingQueue<MessageWrapper>> entry : messageCache.entrySet()) {
            totalRemaining += entry.getValue().size();
            CompletableFuture<Void> future = processingFutures.get(entry.getKey());
            if (future != null && !future.isDone()) {
                activeTasks++;
            }
        }

        log.info("第二阶段进度: 已处理={}, 剩余={}, 活跃任务={}",
                processedCount.get(), totalRemaining, activeTasks);
    }
}