package cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpserver;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.lang.Assert;
import cn.hutool.extra.spring.SpringUtil;
import cn.iocoder.yudao.module.iot.core.biz.IotDeviceCommonApi;
import cn.iocoder.yudao.module.iot.core.biz.dto.IotModbusDeviceConfigRespDTO;
import cn.iocoder.yudao.module.iot.core.enums.IotProtocolTypeEnum;
import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus;
import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
import cn.iocoder.yudao.module.iot.core.util.IotDeviceMessageUtils;
import cn.iocoder.yudao.module.iot.gateway.config.IotGatewayProperties.ProtocolProperties;
import cn.iocoder.yudao.module.iot.gateway.protocol.IotProtocol;
import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpserver.codec.IotModbusFrameDecoder;
import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpserver.codec.IotModbusFrameEncoder;
import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpserver.handler.downstream.IotModbusTcpServerDownstreamHandler;
import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpserver.handler.downstream.IotModbusTcpServerDownstreamSubscriber;
import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpserver.handler.upstream.IotModbusTcpServerUpstreamHandler;
import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpserver.manager.IotModbusTcpServerConfigCacheService;
import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpserver.manager.IotModbusTcpServerConnectionManager;
import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpserver.manager.IotModbusTcpServerConnectionManager.ConnectionInfo;
import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpserver.manager.IotModbusTcpServerPendingRequestManager;
import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpserver.manager.IotModbusTcpServerPollScheduler;
import cn.iocoder.yudao.module.iot.gateway.service.device.IotDeviceService;
import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService;
import io.vertx.core.Vertx;
import io.vertx.core.net.NetServer;
import io.vertx.core.net.NetServerOptions;
import io.vertx.core.net.NetSocket;
import io.vertx.core.parsetools.RecordParser;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * IoT 网关 Modbus TCP Server 协议
 * <p>
 * 作为 TCP Server 接收设备主动连接：
 * 1. 设备通过自定义功能码（FC 65）发送认证请求
 * 2. 认证成功后，网关主动发送 Modbus 读请求，设备响应（云端轮询模式）
 *
 * @author 芋道源码
 */
@Slf4j
public class IotModbusTcpServerProtocol implements IotProtocol {

    /**
     * 协议配置
     */
    private final ProtocolProperties properties;
    /**
     * 服务器 ID（用于消息追踪，全局唯一）
     */
    @Getter
    private final String serverId;

    /**
     * 运行状态
     */
    @Getter
    private volatile boolean running = false;

    /**
     * Vert.x 实例
     */
    private final Vertx vertx;
    /**
     * TCP Server
     */
    private NetServer netServer;
    /**
     * 配置刷新定时器 ID
     */
    private Long configRefreshTimerId;
    /**
     * Pending Request 清理定时器 ID
     */
    private Long requestCleanupTimerId;

    /**
     * 连接管理器
     */
    private final IotModbusTcpServerConnectionManager connectionManager;
    /**
     * 下行消息订阅者
     */
    private IotModbusTcpServerDownstreamSubscriber downstreamSubscriber;

    private final IotModbusFrameDecoder frameDecoder;
    @SuppressWarnings("FieldCanBeLocal")
    private final IotModbusFrameEncoder frameEncoder;

    private final IotModbusTcpServerConfigCacheService configCacheService;
    private final IotModbusTcpServerPendingRequestManager pendingRequestManager;
    private final IotModbusTcpServerUpstreamHandler upstreamHandler;
    private final IotModbusTcpServerPollScheduler pollScheduler;
    private final IotDeviceMessageService messageService;

    public IotModbusTcpServerProtocol(ProtocolProperties properties) {
        IotModbusTcpServerConfig slaveConfig = properties.getModbusTcpServer();
        Assert.notNull(slaveConfig, "Modbus TCP Server 协议配置（modbusTcpServer）不能为空");
        this.properties = properties;
        this.serverId = IotDeviceMessageUtils.generateServerId(properties.getPort());

        // 初始化 Vertx
        this.vertx = Vertx.vertx();

        // 初始化 Manager
        IotDeviceCommonApi deviceApi = SpringUtil.getBean(IotDeviceCommonApi.class);
        this.connectionManager = new IotModbusTcpServerConnectionManager();
        this.configCacheService = new IotModbusTcpServerConfigCacheService(deviceApi);
        this.pendingRequestManager = new IotModbusTcpServerPendingRequestManager();

        // 初始化帧编解码器
        this.frameDecoder = new IotModbusFrameDecoder(slaveConfig.getCustomFunctionCode());
        this.frameEncoder = new IotModbusFrameEncoder(slaveConfig.getCustomFunctionCode());

        // 初始化共享事务 ID 自增器（PollScheduler 和 DownstreamHandler 共用，避免 transactionId 冲突）
        AtomicInteger transactionIdCounter = new AtomicInteger(0);
        // 初始化轮询调度器
        this.pollScheduler = new IotModbusTcpServerPollScheduler(
                vertx, connectionManager, frameEncoder, pendingRequestManager,
                slaveConfig.getRequestTimeout(), transactionIdCounter, configCacheService);

        // 初始化 Handler
        this.messageService = SpringUtil.getBean(IotDeviceMessageService.class);
        IotDeviceService deviceService = SpringUtil.getBean(IotDeviceService.class);
        this.upstreamHandler = new IotModbusTcpServerUpstreamHandler(
                deviceApi, this.messageService, frameEncoder,
                connectionManager, configCacheService, pendingRequestManager,
                pollScheduler, deviceService, serverId);
    }

    @Override
    public String getId() {
        return properties.getId();
    }

    @Override
    public IotProtocolTypeEnum getType() {
        return IotProtocolTypeEnum.MODBUS_TCP_SERVER;
    }

    @Override
    public void start() {
        if (running) {
            log.warn("[start][IoT Modbus TCP Server 协议 {} 已经在运行中]", getId());
            return;
        }

        try {
            // 1. 启动配置刷新定时器
            IotModbusTcpServerConfig slaveConfig = properties.getModbusTcpServer();
            configRefreshTimerId = vertx.setPeriodic(
                    TimeUnit.SECONDS.toMillis(slaveConfig.getConfigRefreshInterval()),
                    id -> refreshConfig());

            // 2.1 启动 TCP Server
            startTcpServer();

            // 2.2 启动 PendingRequest 清理定时器
            requestCleanupTimerId = vertx.setPeriodic(
                    slaveConfig.getRequestCleanupInterval(),
                    id -> pendingRequestManager.cleanupExpired());
            running = true;
            log.info("[start][IoT Modbus TCP Server 协议 {} 启动成功, serverId={}, port={}]",
                    getId(), serverId, properties.getPort());

            // 3. 启动下行消息订阅
            IotMessageBus messageBus = SpringUtil.getBean(IotMessageBus.class);
            IotModbusTcpServerDownstreamHandler downstreamHandler = new IotModbusTcpServerDownstreamHandler(
                    connectionManager, configCacheService, frameEncoder, this.pollScheduler.getTransactionIdCounter());
            this.downstreamSubscriber = new IotModbusTcpServerDownstreamSubscriber(
                    this, downstreamHandler, messageBus);
            downstreamSubscriber.start();
        } catch (Exception e) {
            log.error("[start][IoT Modbus TCP Server 协议 {} 启动失败]", getId(), e);
            stop0();
            throw e;
        }
    }

    @Override
    public void stop() {
        if (!running) {
            return;
        }
        stop0();
    }

    private void stop0() {
        // 1. 停止下行消息订阅
        if (downstreamSubscriber != null) {
            try {
                downstreamSubscriber.stop();
            } catch (Exception e) {
                log.error("[stop][下行消息订阅器停止失败]", e);
            }
            downstreamSubscriber = null;
        }

        // 2.1 取消定时器
        if (configRefreshTimerId != null) {
            vertx.cancelTimer(configRefreshTimerId);
            configRefreshTimerId = null;
        }
        if (requestCleanupTimerId != null) {
            vertx.cancelTimer(requestCleanupTimerId);
            requestCleanupTimerId = null;
        }
        // 2.2 停止轮询
        pollScheduler.stopAll();
        // 2.3 清理 PendingRequest
        pendingRequestManager.clear();
        // 2.4 关闭所有连接
        connectionManager.closeAll();
        // 2.5 关闭 TCP Server
        if (netServer != null) {
            try {
                netServer.close().result();
                log.info("[stop][TCP Server 已关闭]");
            } catch (Exception e) {
                log.error("[stop][TCP Server 关闭失败]", e);
            }
            netServer = null;
        }

        // 3. 关闭 Vertx
        if (vertx != null) {
            try {
                vertx.close().result();
            } catch (Exception e) {
                log.error("[stop][Vertx 关闭失败]", e);
            }
        }
        running = false;
        log.info("[stop][IoT Modbus TCP Server 协议 {} 已停止]", getId());
    }

    /**
     * 启动 TCP Server
     */
    private void startTcpServer() {
        // 1. 创建 TCP Server
        NetServerOptions options = new NetServerOptions()
                .setPort(properties.getPort());
        netServer = vertx.createNetServer(options);

        // 2. 设置连接处理器
        netServer.connectHandler(this::handleConnection);
        try {
            netServer.listen().toCompletionStage().toCompletableFuture().get();
            log.info("[startTcpServer][TCP Server 启动成功, port={}]", properties.getPort());
        } catch (Exception e) {
            throw new RuntimeException("[startTcpServer][TCP Server 启动失败]", e);
        }
    }

    /**
     * 处理新连接
     */
    private void handleConnection(NetSocket socket) {
        log.info("[handleConnection][新连接, remoteAddress={}]", socket.remoteAddress());

        // 1. 创建 RecordParser 并设置为数据处理器
        RecordParser recordParser =  frameDecoder.createRecordParser((frame, frameFormat) -> {
            // 【重要】帧处理分发，即消息处理
            upstreamHandler.handleFrame(socket, frame, frameFormat);
        });
        socket.handler(recordParser);

        // 2.1 连接关闭处理
        socket.closeHandler(v -> {
            ConnectionInfo info = connectionManager.removeConnection(socket);
            if (info == null || info.getDeviceId() == null) {
                log.info("[handleConnection][未认证连接关闭, remoteAddress={}]", socket.remoteAddress());
                return;
            }
            pollScheduler.stopPolling(info.getDeviceId());
            pendingRequestManager.removeDevice(info.getDeviceId());
            configCacheService.removeConfig(info.getDeviceId());
            // 发送设备下线消息
            try {
                IotDeviceMessage offlineMessage = IotDeviceMessage.buildStateOffline();
                messageService.sendDeviceMessage(offlineMessage, info.getProductKey(), info.getDeviceName(), serverId);
            } catch (Exception ex) {
                log.error("[handleConnection][发送设备下线消息失败, deviceId={}]", info.getDeviceId(), ex);
            }
            log.info("[handleConnection][连接关闭, deviceId={}, remoteAddress={}]",
                    info.getDeviceId(), socket.remoteAddress());
        });
        // 2.2 异常处理
        socket.exceptionHandler(e -> {
            log.error("[handleConnection][连接异常, remoteAddress={}]", socket.remoteAddress(), e);
            socket.close();
        });
    }

    /**
     * 刷新已连接设备的配置（定时调用）
     */
    private synchronized void refreshConfig() {
        try {
            // 1. 只刷新已连接设备的配置
            Set<Long> connectedDeviceIds = connectionManager.getConnectedDeviceIds();
            if (CollUtil.isEmpty(connectedDeviceIds)) {
                return;
            }
            List<IotModbusDeviceConfigRespDTO> configs =
                    configCacheService.refreshConnectedDeviceConfigList(connectedDeviceIds);
            if (configs == null) {
                log.warn("[refreshConfig][刷新配置失败，跳过本次刷新]");
                return;
            }
            log.debug("[refreshConfig][刷新了 {} 个已连接设备的配置]", configs.size());

            // 2. 更新已连接设备的轮询任务
            for (IotModbusDeviceConfigRespDTO config : configs) {
                try {
                    pollScheduler.updatePolling(config);
                } catch (Exception e) {
                    log.error("[refreshConfig][处理设备配置失败, deviceId={}]", config.getDeviceId(), e);
                }
            }

            // 3. 清理本轮不再返回配置的已连接设备，避免继续轮询已删除设备的旧点位
            Set<Long> missingDeviceIds = configCacheService.cleanupMissingConfigs(connectedDeviceIds, configs);
            for (Long deviceId : missingDeviceIds) {
                cleanupMissingDevice(deviceId);
            }
        } catch (Exception e) {
            log.error("[refreshConfig][刷新配置失败]", e);
        }
    }

    private void cleanupMissingDevice(Long deviceId) {
        try {
            pollScheduler.stopPolling(deviceId);
            pendingRequestManager.removeDevice(deviceId);
            configCacheService.removeConfig(deviceId);
            connectionManager.closeConnection(deviceId);
            log.info("[cleanupMissingDevice][设备 {} 配置已失效，已停止轮询并清理连接]", deviceId);
        } catch (Exception e) {
            log.error("[cleanupMissingDevice][清理设备失败, deviceId={}]", deviceId, e);
        }
    }

}