package cn.iocoder.yudao.module.iot.gateway.protocol.modbus.common.manager;

import cn.hutool.core.collection.CollUtil;
import cn.iocoder.yudao.module.iot.core.biz.dto.IotModbusDeviceConfigRespDTO;
import cn.iocoder.yudao.module.iot.core.biz.dto.IotModbusPointRespDTO;
import io.vertx.core.Vertx;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;

import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.convertList;
import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.convertSet;

/**
 * Modbus 轮询调度器基类
 * <p>
 * 封装通用的定时器管理、per-device 请求队列限速逻辑。
 * 子类只需实现 {@link #pollPoint(Long, Long)} 定义具体的轮询动作。
 * 如需将多个点位合并为一个轮询任务，可覆盖 {@link #buildPollTasks(IotModbusDeviceConfigRespDTO)}
 * 和 {@link #pollTask(Long, String)}。
 * <p>
 *
 * @author 芋道源码
 */
@Slf4j
public abstract class AbstractIotModbusPollScheduler {

    protected final Vertx vertx;

    /**
     * 同设备最小请求间隔（毫秒），防止 Modbus 设备性能不足时请求堆积
     */
    private static final long MIN_REQUEST_INTERVAL = 1000;
    /**
     * 每个设备请求队列的最大长度，超出时丢弃最旧请求
     */
    private static final int MAX_QUEUE_SIZE = 1000;

    /**
     * 设备轮询任务的定时器映射：deviceId -> (taskKey -> PollTimerInfo)
     */
    private final Map<Long, Map<String, PollTimerInfo>> devicePollTimers = new ConcurrentHashMap<>();

    /**
     * per-device 请求队列：deviceId -> 待执行请求队列
     */
    private final Map<Long, Queue<Runnable>> deviceRequestQueues = new ConcurrentHashMap<>();
    /**
     * per-device 上次请求时间戳：deviceId -> lastRequestTimeMs
     */
    private final Map<Long, Long> deviceLastRequestTime = new ConcurrentHashMap<>();
    /**
     * per-device 延迟 timer 标记：deviceId -> 是否有延迟 timer 在等待
     */
    private final Map<Long, Boolean> deviceDelayTimerActive = new ConcurrentHashMap<>();

    protected AbstractIotModbusPollScheduler(Vertx vertx) {
        this.vertx = vertx;
    }

    /**
     * 轮询任务信息
     */
    @Data
    @AllArgsConstructor
    protected static class PollTask {

        /**
         * 任务标识
         */
        private String key;
        /**
         * 轮询间隔（用于判断是否需要更新定时器）
         */
        private Integer pollInterval;

    }

    /**
     * 轮询定时器信息
     */
    @Data
    @AllArgsConstructor
    private static class PollTimerInfo {

        /**
         * Vert.x 定时器 ID
         */
        private Long timerId;
        /**
         * 轮询间隔（用于判断是否需要更新定时器）
         */
        private Integer pollInterval;

    }

    // ========== 轮询管理 ==========

    /**
     * 更新轮询任务（增量更新）
     *
     * 1. 【删除】任务：停止对应的轮询定时器
     * 2. 【新增】任务：创建对应的轮询定时器
     * 3. 【修改】任务：pollInterval 变化，重建对应的轮询定时器
     *    【修改】其他属性变化：不需要重建定时器（pollTask 运行时从 configCache 取最新配置）
     */
    public void updatePolling(IotModbusDeviceConfigRespDTO config) {
        Long deviceId = config.getDeviceId();
        List<PollTask> newTasks = buildPollTasks(config);
        Map<String, PollTimerInfo> currentTimers = devicePollTimers
                .computeIfAbsent(deviceId, k -> new ConcurrentHashMap<>());
        // 1.1 计算新配置中的任务 Key 集合
        Set<String> newTaskKeys = convertSet(newTasks, PollTask::getKey);
        // 1.2 计算删除的任务 Key 集合
        Set<String> removedTaskKeys = new HashSet<>(currentTimers.keySet());
        removedTaskKeys.removeAll(newTaskKeys);

        // 2. 处理删除的任务：停止不再存在的定时器
        for (String taskKey : removedTaskKeys) {
            PollTimerInfo timerInfo = currentTimers.remove(taskKey);
            if (timerInfo != null) {
                vertx.cancelTimer(timerInfo.getTimerId());
                log.debug("[updatePolling][设备 {} 轮询任务 {} 定时器已删除]", deviceId, taskKey);
            }
        }

        // 3. 处理新增和修改的任务
        if (CollUtil.isEmpty(newTasks)) {
            return;
        }
        for (PollTask task : newTasks) {
            String taskKey = task.getKey();
            Integer newPollInterval = task.getPollInterval();
            PollTimerInfo existingTimer = currentTimers.get(taskKey);
            // 3.1 新增任务：创建定时器
            if (existingTimer == null) {
                Long timerId = createPollTimer(deviceId, taskKey, newPollInterval);
                if (timerId != null) {
                    currentTimers.put(taskKey, new PollTimerInfo(timerId, newPollInterval));
                    log.debug("[updatePolling][设备 {} 轮询任务 {} 定时器已创建, interval={}ms]",
                            deviceId, taskKey, newPollInterval);
                }
            } else if (!Objects.equals(existingTimer.getPollInterval(), newPollInterval)) {
                // 3.2 pollInterval 变化：重建定时器
                vertx.cancelTimer(existingTimer.getTimerId());
                Long timerId = createPollTimer(deviceId, taskKey, newPollInterval);
                if (timerId != null) {
                    currentTimers.put(taskKey, new PollTimerInfo(timerId, newPollInterval));
                    log.debug("[updatePolling][设备 {} 轮询任务 {} 定时器已更新, interval={}ms -> {}ms]",
                            deviceId, taskKey, existingTimer.getPollInterval(), newPollInterval);
                } else {
                    currentTimers.remove(taskKey);
                }
            }
            // 3.3 其他属性变化：无需重建定时器，因为 pollTask() 运行时从 configCache 获取最新配置
        }
    }

    /**
     * 构建轮询任务列表
     *
     * 默认每个点位一个轮询任务。TCP Client 等协议可覆盖该方法，将多个点位合并为一个批量读取任务。
     */
    protected List<PollTask> buildPollTasks(IotModbusDeviceConfigRespDTO config) {
        return convertList(config.getPoints(), point -> new PollTask(String.valueOf(point.getId()), point.getPollInterval()));
    }

    /**
     * 创建轮询定时器
     */
    private Long createPollTimer(Long deviceId, String taskKey, Integer pollInterval) {
        if (pollInterval == null || pollInterval <= 0) {
            return null;
        }
        return vertx.setPeriodic(pollInterval, timerId -> {
            try {
                submitPollRequest(deviceId, taskKey);
            } catch (Exception e) {
                log.error("[createPollTimer][轮询任务失败, deviceId={}, taskKey={}]", deviceId, taskKey, e);
            }
        });
    }

    // ========== 请求队列（per-device 限速） ==========

    /**
     * 提交轮询请求到设备请求队列（保证同设备请求间隔）
     */
    private void submitPollRequest(Long deviceId, String taskKey) {
        // 1. 【重要】将请求添加到设备的请求队列
        Queue<Runnable> queue = deviceRequestQueues.computeIfAbsent(deviceId, k -> new ConcurrentLinkedQueue<>());
        while (queue.size() >= MAX_QUEUE_SIZE) {
            // 超出上限时，丢弃最旧的请求
            queue.poll();
            log.warn("[submitPollRequest][设备 {} 请求队列已满({}), 丢弃最旧请求]", deviceId, MAX_QUEUE_SIZE);
        }
        queue.offer(() -> pollTask(deviceId, taskKey));

        // 2. 处理设备请求队列（如果没有延迟 timer 在等待）
        processDeviceQueue(deviceId);
    }

    /**
     * 处理设备请求队列
     */
    private void processDeviceQueue(Long deviceId) {
        Queue<Runnable> queue = deviceRequestQueues.get(deviceId);
        if (CollUtil.isEmpty(queue)) {
            return;
        }
        // 检查是否已有延迟 timer 在等待
        if (Boolean.TRUE.equals(deviceDelayTimerActive.get(deviceId))) {
            return;
        }

        // 不满足间隔要求，延迟执行
        long now = System.currentTimeMillis();
        long lastTime = deviceLastRequestTime.getOrDefault(deviceId, 0L);
        long elapsed = now - lastTime;
        if (elapsed < MIN_REQUEST_INTERVAL) {
            scheduleNextRequest(deviceId, MIN_REQUEST_INTERVAL - elapsed);
            return;
        }

        // 满足间隔要求，立即执行
        Runnable task = queue.poll();
        if (task == null) {
            return;
        }
        deviceLastRequestTime.put(deviceId, now);
        task.run();
        // 继续处理队列中的下一个（如果有的话，需要延迟）
        if (CollUtil.isNotEmpty(queue)) {
            scheduleNextRequest(deviceId);
        }
    }

    private void scheduleNextRequest(Long deviceId) {
        scheduleNextRequest(deviceId, MIN_REQUEST_INTERVAL);
    }

    private void scheduleNextRequest(Long deviceId, long delayMs) {
        deviceDelayTimerActive.put(deviceId, true);
        vertx.setTimer(delayMs, id -> {
            deviceDelayTimerActive.put(deviceId, false);
            Queue<Runnable> queue = deviceRequestQueues.get(deviceId);
            if (CollUtil.isEmpty(queue)) {
                return;
            }

            // 满足间隔要求，立即执行
            Runnable task = queue.poll();
            if (task == null) {
                return;
            }
            deviceLastRequestTime.put(deviceId, System.currentTimeMillis());
            task.run();
            // 继续处理队列中的下一个（如果有的话，需要延迟）
            if (CollUtil.isNotEmpty(queue)) {
                scheduleNextRequest(deviceId);
            }
        });
    }

    // ========== 轮询执行 ==========

    /**
     * 轮询任务
     *
     * 默认将任务标识作为点位 ID，执行单点轮询。
     */
    protected void pollTask(Long deviceId, String taskKey) {
        pollPoint(deviceId, Long.valueOf(taskKey));
    }

    /**
     * 轮询单个点位（子类实现具体的读取逻辑）
     *
     * @param deviceId 设备 ID
     * @param pointId  点位 ID
     */
    protected abstract void pollPoint(Long deviceId, Long pointId);

    // ========== 停止 ==========

    /**
     * 停止设备的轮询
     */
    public void stopPolling(Long deviceId) {
        Map<String, PollTimerInfo> timers = devicePollTimers.remove(deviceId);
        if (CollUtil.isNotEmpty(timers)) {
            for (PollTimerInfo timerInfo : timers.values()) {
                vertx.cancelTimer(timerInfo.getTimerId());
            }
        }
        // 清理请求队列
        deviceRequestQueues.remove(deviceId);
        deviceLastRequestTime.remove(deviceId);
        deviceDelayTimerActive.remove(deviceId);
        log.debug("[stopPolling][设备 {} 停止了 {} 个轮询定时器]", deviceId,
                CollUtil.isEmpty(timers) ? 0 : timers.size());
    }

    /**
     * 停止所有轮询
     */
    public void stopAll() {
        for (Long deviceId : new ArrayList<>(devicePollTimers.keySet())) {
            stopPolling(deviceId);
        }
    }

}