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

import cn.hutool.core.util.ObjUtil;
import cn.iocoder.yudao.module.iot.core.biz.dto.IotModbusDeviceConfigRespDTO;
import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService;
import com.ghgande.j2mod.modbus.net.TCPMasterConnection;
import io.vertx.core.Context;
import io.vertx.core.Future;
import io.vertx.core.Vertx;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;

import java.net.InetAddress;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * IoT Modbus TCP 连接管理器
 * <p>
 * 统一管理 Modbus TCP 连接：
 * 1. 管理 TCP 连接（相同 ip:port 共用连接）
 * 2. 分布式锁管理（连接级别），避免多节点重复创建连接
 * 3. 连接重试和故障恢复
 *
 * @author 芋道源码
 */
@Slf4j
public class IotModbusTcpClientConnectionManager {

    private static final String LOCK_KEY_PREFIX = "iot:modbus-tcp:connection:";

    private final RedissonClient redissonClient;
    private final Vertx vertx;
    private final IotDeviceMessageService messageService;
    private final IotModbusTcpClientConfigCacheService configCacheService;
    private final String serverId;

    /**
     * 连接池：key = ip:port
     */
    private final Map<String, ModbusConnection> connectionPool = new ConcurrentHashMap<>();

    /**
     * 设备 ID 到连接 key 的映射
     */
    private final Map<Long, String> deviceConnectionMap = new ConcurrentHashMap<>();

    public IotModbusTcpClientConnectionManager(RedissonClient redissonClient, Vertx vertx,
                                                IotDeviceMessageService messageService,
                                                IotModbusTcpClientConfigCacheService configCacheService,
                                                String serverId) {
        this.redissonClient = redissonClient;
        this.vertx = vertx;
        this.messageService = messageService;
        this.configCacheService = configCacheService;
        this.serverId = serverId;
    }

    /**
     * 确保连接存在
     * <p>
     * 首次建连成功时，直接发送设备上线消息
     *
     * @param config 设备配置
     */
    public void ensureConnection(IotModbusDeviceConfigRespDTO config) {
        // 1.1 检查设备是否切换了 IP/端口，若是则先清理旧连接
        String connectionKey = buildConnectionKey(config.getIp(), config.getPort());
        String oldConnectionKey = deviceConnectionMap.get(config.getDeviceId());
        if (oldConnectionKey != null && ObjUtil.notEqual(oldConnectionKey, connectionKey)) {
            log.info("[ensureConnection][设备 {} IP/端口变更: {} -> {}, 清理旧连接]",
                    config.getDeviceId(), oldConnectionKey, connectionKey);
            removeDevice(config.getDeviceId());
        }
        // 1.2 记录设备与连接的映射
        deviceConnectionMap.put(config.getDeviceId(), connectionKey);

        // 2. 情况一：连接已存在，注册设备并发送上线消息
        ModbusConnection connection = connectionPool.get(connectionKey);
        if (connection != null) {
            addDeviceAndOnline(connection, config);
            return;
        }

        // 3. 情况二：连接不存在，加分布式锁创建新连接
        RLock lock = redissonClient.getLock(LOCK_KEY_PREFIX + connectionKey);
        if (!lock.tryLock()) {
            log.debug("[ensureConnection][获取锁失败, 由其他节点负责: {}]", connectionKey);
            return;
        }
        try {
            // 3.1 double-check：拿到锁后再次检查，避免并发创建重复连接
            connection = connectionPool.get(connectionKey);
            if (connection != null) {
                addDeviceAndOnline(connection, config);
                lock.unlock();
                return;
            }
            // 3.2 创建新连接
            connection = createConnection(config);
            connection.setLock(lock);
            connectionPool.put(connectionKey, connection);
            log.info("[ensureConnection][创建 Modbus 连接成功: {}]", connectionKey);
            // 3.3 注册设备并发送上线消息
            addDeviceAndOnline(connection, config);
        } catch (Exception e) {
            log.error("[ensureConnection][创建 Modbus 连接失败: {}]", connectionKey, e);
            // 建连失败，释放锁让其他节点可重试
            lock.unlock();
        }
    }

    /**
     * 创建 Modbus TCP 连接
     */
    private ModbusConnection createConnection(IotModbusDeviceConfigRespDTO config) throws Exception {
        // 1. 创建 TCP 连接
        TCPMasterConnection tcpConnection = new TCPMasterConnection(InetAddress.getByName(config.getIp()));
        tcpConnection.setPort(config.getPort());
        tcpConnection.setTimeout(config.getTimeout());
        tcpConnection.connect();

        // 2. 创建 Modbus 连接对象
        return new ModbusConnection()
                .setConnectionKey(buildConnectionKey(config.getIp(), config.getPort()))
                .setTcpConnection(tcpConnection).setContext(vertx.getOrCreateContext())
                .setTimeout(config.getTimeout()).setRetryInterval(config.getRetryInterval());
    }

    /**
     * 获取连接
     */
    public ModbusConnection getConnection(Long deviceId) {
        String connectionKey = deviceConnectionMap.get(deviceId);
        if (connectionKey == null) {
            return null;
        }
        return connectionPool.get(connectionKey);
    }

    /**
     * 获取设备的 slave ID
     */
    public Integer getSlaveId(Long deviceId) {
        ModbusConnection connection = getConnection(deviceId);
        if (connection == null) {
            return null;
        }
        return connection.getSlaveId(deviceId);
    }

    /**
     * 移除设备
     * <p>
     * 移除时直接发送设备下线消息
     */
    public void removeDevice(Long deviceId) {
        // 1.1 移除设备时，发送下线消息
        sendOfflineMessage(deviceId);
        // 1.2 移除设备引用
        String connectionKey = deviceConnectionMap.remove(deviceId);
        if (connectionKey == null) {
            return;
        }

        // 2.1 移除连接中的设备引用
        ModbusConnection connection = connectionPool.get(connectionKey);
        if (connection == null) {
            return;
        }
        connection.removeDevice(deviceId);
        // 2.2 如果没有设备引用了，关闭连接
        if (connection.getDeviceCount() == 0) {
            closeConnection(connectionKey);
        }
    }

    // ==================== 设备连接 & 上下线消息 ====================

    /**
     * 注册设备到连接，并发送上线消息
     */
    private void addDeviceAndOnline(ModbusConnection connection,
                                    IotModbusDeviceConfigRespDTO config) {
        Integer previous = connection.addDevice(config.getDeviceId(), config.getSlaveId());
        // 首次注册，发送上线消息
        if (previous == null) {
            sendOnlineMessage(config);
        }
    }

    /**
     * 发送设备上线消息
     */
    private void sendOnlineMessage(IotModbusDeviceConfigRespDTO config) {
        try {
            IotDeviceMessage onlineMessage = IotDeviceMessage.buildStateUpdateOnline();
            messageService.sendDeviceMessage(onlineMessage,
                    config.getProductKey(), config.getDeviceName(), serverId);
        } catch (Exception ex) {
            log.error("[sendOnlineMessage][发送设备上线消息失败, deviceId={}]", config.getDeviceId(), ex);
        }
    }

    /**
     * 发送设备下线消息
     */
    private void sendOfflineMessage(Long deviceId) {
        IotModbusDeviceConfigRespDTO config = configCacheService.getConfig(deviceId);
        if (config == null) {
            return;
        }
        try {
            IotDeviceMessage offlineMessage = IotDeviceMessage.buildStateOffline();
            messageService.sendDeviceMessage(offlineMessage,
                    config.getProductKey(), config.getDeviceName(), serverId);
        } catch (Exception ex) {
            log.error("[sendOfflineMessage][发送设备下线消息失败, deviceId={}]", deviceId, ex);
        }
    }

    /**
     * 关闭指定连接
     */
    private void closeConnection(String connectionKey) {
        ModbusConnection connection = connectionPool.remove(connectionKey);
        if (connection == null) {
            return;
        }

        try {
            if (connection.getTcpConnection() != null) {
                connection.getTcpConnection().close();
            }
            // 释放分布式锁，让其他节点可接管
            RLock lock = connection.getLock();
            if (lock != null && lock.isHeldByCurrentThread()) {
                lock.unlock();
            }
            log.info("[closeConnection][关闭 Modbus 连接: {}]", connectionKey);
        } catch (Exception e) {
            log.error("[closeConnection][关闭连接失败: {}]", connectionKey, e);
        }
    }

    /**
     * 关闭所有连接
     */
    public void closeAll() {
        // 先复制再遍历，避免 closeConnection 中 remove 导致并发修改
        List<String> connectionKeys = new ArrayList<>(connectionPool.keySet());
        for (String connectionKey : connectionKeys) {
            closeConnection(connectionKey);
        }
        deviceConnectionMap.clear();
    }

    private String buildConnectionKey(String ip, Integer port) {
        return ip + ":" + port;
    }

    /**
     * Modbus 连接信息
     */
    @Data
    public static class ModbusConnection {

        private String connectionKey;
        private TCPMasterConnection tcpConnection;
        private Integer timeout;
        private Integer retryInterval;
        /**
         * 设备 ID 到 slave ID 的映射
         */
        private final Map<Long, Integer> deviceSlaveMap = new ConcurrentHashMap<>();

        /**
         * 分布式锁，锁住连接的创建和销毁，避免多节点重复连接同一从站
         */
        private RLock lock;

        /**
         * Vert.x Context，用于 executeBlocking 执行 Modbus 操作，保证同一连接的操作串行执行
         */
        private Context context;

        public Integer addDevice(Long deviceId, Integer slaveId) {
            return deviceSlaveMap.putIfAbsent(deviceId, slaveId);
        }

        public void removeDevice(Long deviceId) {
            deviceSlaveMap.remove(deviceId);
        }

        public int getDeviceCount() {
            return deviceSlaveMap.size();
        }

        public Integer getSlaveId(Long deviceId) {
            return deviceSlaveMap.get(deviceId);
        }

        /**
         * 执行 Modbus 读取操作（阻塞方式，在 Vert.x worker 线程执行）
         */
        public <T> Future<T> executeBlocking(java.util.function.Function<TCPMasterConnection, T> operation) {
            // ordered=true 保证同一 Context 的操作串行执行，不同连接之间可并行
            return context.executeBlocking(() -> operation.apply(tcpConnection), true);
        }
    }

}