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.convertSet;

/**
 * Modbus 轮询调度器基类
 * <p>
 * 封装通用的定时器管理、per-device 请求队列限速逻辑。
 * 子类只需实现 {@link #pollPoint(Long, Long)} 定义具体的轮询动作。
 * <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 -> (pointId -> PointTimerInfo)
     */
    private final Map<Long, Map<Long, PointTimerInfo>> devicePointTimers = 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
    private static class PointTimerInfo {

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

    }

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

    /**
     * 更新轮询任务（增量更新）
     *
     * 1. 【删除】点位：停止对应的轮询定时器
     * 2. 【新增】点位：创建对应的轮询定时器
     * 3. 【修改】点位：pollInterval 变化，重建对应的轮询定时器
     *    【修改】其他属性变化：不需要重建定时器（pollPoint 运行时从 configCache 取最新 point）
     */
    public void updatePolling(IotModbusDeviceConfigRespDTO config) {
        Long deviceId = config.getDeviceId();
        List<IotModbusPointRespDTO> newPoints = config.getPoints();
        Map<Long, PointTimerInfo> currentTimers = devicePointTimers
                .computeIfAbsent(deviceId, k -> new ConcurrentHashMap<>());
        // 1.1 计算新配置中的点位 ID 集合
        Set<Long> newPointIds = convertSet(newPoints, IotModbusPointRespDTO::getId);
        // 1.2 计算删除的点位 ID 集合
        Set<Long> removedPointIds = new HashSet<>(currentTimers.keySet());
        removedPointIds.removeAll(newPointIds);

        // 2. 处理删除的点位：停止不再存在的定时器
        for (Long pointId : removedPointIds) {
            PointTimerInfo timerInfo = currentTimers.remove(pointId);
            if (timerInfo != null) {
                vertx.cancelTimer(timerInfo.getTimerId());
                log.debug("[updatePolling][设备 {} 点位 {} 定时器已删除]", deviceId, pointId);
            }
        }

        // 3. 处理新增和修改的点位
        if (CollUtil.isEmpty(newPoints)) {
            return;
        }
        for (IotModbusPointRespDTO point : newPoints) {
            Long pointId = point.getId();
            Integer newPollInterval = point.getPollInterval();
            PointTimerInfo existingTimer = currentTimers.get(pointId);
            // 3.1 新增点位：创建定时器
            if (existingTimer == null) {
                Long timerId = createPollTimer(deviceId, pointId, newPollInterval);
                if (timerId != null) {
                    currentTimers.put(pointId, new PointTimerInfo(timerId, newPollInterval));
                    log.debug("[updatePolling][设备 {} 点位 {} 定时器已创建, interval={}ms]",
                            deviceId, pointId, newPollInterval);
                }
            } else if (!Objects.equals(existingTimer.getPollInterval(), newPollInterval)) {
                // 3.2 pollInterval 变化：重建定时器
                vertx.cancelTimer(existingTimer.getTimerId());
                Long timerId = createPollTimer(deviceId, pointId, newPollInterval);
                if (timerId != null) {
                    currentTimers.put(pointId, new PointTimerInfo(timerId, newPollInterval));
                    log.debug("[updatePolling][设备 {} 点位 {} 定时器已更新, interval={}ms -> {}ms]",
                            deviceId, pointId, existingTimer.getPollInterval(), newPollInterval);
                } else {
                    currentTimers.remove(pointId);
                }
            }
            // 3.3 其他属性变化：无需重建定时器，因为 pollPoint() 运行时从 configCache 获取最新 point，自动使用新配置
        }
    }

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

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

    /**
     * 提交轮询请求到设备请求队列（保证同设备请求间隔）
     */
    private void submitPollRequest(Long deviceId, Long pointId) {
        // 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(() -> pollPoint(deviceId, pointId));

        // 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);
            }
        });
    }

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

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

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

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

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

}