package cn.iocoder.yudao.module.iot.gateway.protocol.mqtt;

import cn.hutool.core.lang.Assert;
import cn.hutool.core.util.StrUtil;
import cn.hutool.extra.spring.SpringUtil;
import cn.iocoder.yudao.module.iot.core.biz.IotDeviceCommonApi;
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;
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.mqtt.handler.downstream.IotMqttDownstreamHandler;
import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.handler.downstream.IotMqttDownstreamSubscriber;
import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.handler.upstream.IotMqttAuthHandler;
import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.handler.upstream.IotMqttRegisterHandler;
import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.handler.upstream.IotMqttUpstreamHandler;
import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.manager.IotMqttConnectionManager;
import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService;
import cn.iocoder.yudao.module.iot.gateway.util.IotMqttTopicUtils;
import io.netty.handler.codec.mqtt.MqttConnectReturnCode;
import io.netty.handler.codec.mqtt.MqttQoS;
import io.vertx.core.Vertx;
import io.vertx.core.net.PemKeyCertOptions;
import io.vertx.mqtt.MqttEndpoint;
import io.vertx.mqtt.MqttServer;
import io.vertx.mqtt.MqttServerOptions;
import io.vertx.mqtt.MqttTopicSubscription;
import io.vertx.mqtt.messages.MqttPublishMessage;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;

/**
 * IoT 网关 MQTT 协议：接收设备上行消息
 *
 * @author 芋道源码
 */
@Slf4j
public class IotMqttProtocol implements IotProtocol {

    /**
     * 注册连接的 clientId 标识
     *
     * @see #handleEndpoint(MqttEndpoint)
     */
    private static final String AUTH_TYPE_REGISTER = "|authType=register|";

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

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

    /**
     * Vert.x 实例
     */
    private Vertx vertx;
    /**
     * MQTT 服务器
     */
    private MqttServer mqttServer;

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

    private final IotDeviceMessageService deviceMessageService;

    private final IotMqttAuthHandler authHandler;
    private final IotMqttRegisterHandler registerHandler;
    private final IotMqttUpstreamHandler upstreamHandler;

    public IotMqttProtocol(ProtocolProperties properties) {
        IotMqttConfig mqttConfig = properties.getMqtt();
        Assert.notNull(mqttConfig, "MQTT 协议配置（mqtt）不能为空");
        this.properties = properties;
        this.serverId = IotDeviceMessageUtils.generateServerId(properties.getPort());

        // 初始化连接管理器
        this.connectionManager = new IotMqttConnectionManager();

        // 初始化 Handler
        this.deviceMessageService = SpringUtil.getBean(IotDeviceMessageService.class);
        IotDeviceCommonApi deviceApi = SpringUtil.getBean(IotDeviceCommonApi.class);
        this.authHandler = new IotMqttAuthHandler(connectionManager, deviceMessageService, deviceApi, serverId);
        this.registerHandler = new IotMqttRegisterHandler(connectionManager, deviceMessageService);
        this.upstreamHandler = new IotMqttUpstreamHandler(connectionManager, deviceMessageService, serverId);
    }

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

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

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

        // 1.1 创建 Vertx 实例
        this.vertx = Vertx.vertx();

        // 1.2 创建服务器选项
        IotMqttConfig mqttConfig = properties.getMqtt();
        MqttServerOptions options = new MqttServerOptions()
                .setPort(properties.getPort())
                .setMaxMessageSize(mqttConfig.getMaxMessageSize())
                .setTimeoutOnConnect(mqttConfig.getConnectTimeoutSeconds());
        IotGatewayProperties.SslConfig sslConfig = properties.getSsl();
        if (sslConfig != null && Boolean.TRUE.equals(sslConfig.getSsl())) {
            PemKeyCertOptions pemKeyCertOptions = new PemKeyCertOptions()
                    .setKeyPath(sslConfig.getSslKeyPath())
                    .setCertPath(sslConfig.getSslCertPath());
            options.setSsl(true).setKeyCertOptions(pemKeyCertOptions);
        }

        // 1.3 创建服务器并设置连接处理器
        mqttServer = MqttServer.create(vertx, options);
        mqttServer.endpointHandler(this::handleEndpoint);

        // 1.4 启动 MQTT 服务器
        try {
            mqttServer.listen().result();
            running = true;
            log.info("[start][IoT MQTT 协议 {} 启动成功，端口：{}，serverId：{}]",
                    getId(), properties.getPort(), serverId);

            // 2. 启动下行消息订阅者
            IotMessageBus messageBus = SpringUtil.getBean(IotMessageBus.class);
            IotMqttDownstreamHandler downstreamHandler = new IotMqttDownstreamHandler(deviceMessageService, connectionManager);
            this.downstreamSubscriber = new IotMqttDownstreamSubscriber(this, downstreamHandler, messageBus);
            this.downstreamSubscriber.start();
        } catch (Exception e) {
            log.error("[start][IoT MQTT 协议 {} 启动失败]", getId(), e);
            stop0();
            throw e;
        }
    }

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

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

        // 2.1 关闭所有连接
        connectionManager.closeAll();
        // 2.2 关闭 MQTT 服务器
        if (mqttServer != null) {
            try {
                mqttServer.close().result();
                log.info("[stop][IoT MQTT 协议 {} 服务器已停止]", getId());
            } catch (Exception e) {
                log.error("[stop][IoT MQTT 协议 {} 服务器停止失败]", getId(), e);
            }
            mqttServer = null;
        }
        // 2.3 关闭 Vertx 实例
        if (vertx != null) {
            try {
                vertx.close().result();
                log.info("[stop][IoT MQTT 协议 {} Vertx 已关闭]", getId());
            } catch (Exception e) {
                log.error("[stop][IoT MQTT 协议 {} Vertx 关闭失败]", getId(), e);
            }
            vertx = null;
        }
        running = false;
        log.info("[stop][IoT MQTT 协议 {} 已停止]", getId());
    }

    // ======================================= MQTT 连接处理 ======================================

    /**
     * 处理 MQTT 连接端点
     *
     * @param endpoint MQTT 连接端点
     */
    private void handleEndpoint(MqttEndpoint endpoint) {
        // 1. 如果是注册请求，注册待认证连接；否则走正常认证流程
        String clientId = endpoint.clientIdentifier();
        if (StrUtil.endWith(clientId, AUTH_TYPE_REGISTER)) {
            // 情况一：设备注册请求
            registerHandler.handleRegister(endpoint);
            return;
        } else {
            // 情况二：普通认证请求
            if (!authHandler.handleAuthenticationRequest(endpoint)) {
                endpoint.reject(MqttConnectReturnCode.CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD);
                return;
            }
        }

        // 2.1 设置异常和关闭处理器
        endpoint.exceptionHandler(ex -> {
            log.warn("[handleEndpoint][连接异常，客户端 ID: {}，地址: {}，异常: {}]",
                    clientId, connectionManager.getEndpointAddress(endpoint), ex.getMessage());
            endpoint.close();
        });
        endpoint.closeHandler(v -> cleanupConnection(endpoint)); // 处理底层连接关闭（网络中断、异常等）
        endpoint.disconnectHandler(v -> { // 处理 MQTT DISCONNECT 报文
            log.debug("[handleEndpoint][设备断开连接，客户端 ID: {}]", clientId);
            cleanupConnection(endpoint);
        });
        // 2.2 设置心跳处理器
        endpoint.pingHandler(v -> log.debug("[handleEndpoint][收到客户端心跳，客户端 ID: {}]", clientId));

        // 3.1 设置消息处理器
        endpoint.publishHandler(message -> processMessage(endpoint, message));
        // 3.2 设置 QoS 2 消息的 PUBREL 处理器
        endpoint.publishReleaseHandler(endpoint::publishComplete);

        // 4.1 设置订阅处理器（带 ACL 校验）
        endpoint.subscribeHandler(subscribe -> {
            IotMqttConnectionManager.ConnectionInfo connectionInfo = connectionManager.getConnectionInfo(endpoint);
            List<MqttQoS> grantedQoSLevels = new ArrayList<>();
            for (MqttTopicSubscription sub : subscribe.topicSubscriptions()) {
                String topicName = sub.topicName();
                // 校验主题是否属于当前设备
                if (connectionInfo != null && IotMqttTopicUtils.isTopicSubscribeAllowed(
                        topicName, connectionInfo.getProductKey(), connectionInfo.getDeviceName())) {
                    grantedQoSLevels.add(sub.qualityOfService());
                    log.debug("[handleEndpoint][订阅成功，客户端 ID: {}，主题: {}]", clientId, topicName);
                } else {
                    log.warn("[handleEndpoint][订阅被拒绝，客户端 ID: {}，主题: {}]", clientId, topicName);
                    grantedQoSLevels.add(MqttQoS.FAILURE);
                }
            }
            endpoint.subscribeAcknowledge(subscribe.messageId(), grantedQoSLevels);
        });
        // 4.2 设置取消订阅处理器
        endpoint.unsubscribeHandler(unsubscribe -> {
            log.debug("[handleEndpoint][设备取消订阅，客户端 ID: {}，主题: {}]", clientId, unsubscribe.topics());
            endpoint.unsubscribeAcknowledge(unsubscribe.messageId());
        });

        // 5. 接受连接
        endpoint.accept(false);
    }

    /**
     * 处理消息（发布）
     *
     * @param endpoint MQTT 连接端点
     * @param message  发布消息
     */
    private void processMessage(MqttEndpoint endpoint, MqttPublishMessage message) {
        String clientId = endpoint.clientIdentifier();
        try {
            // 1. 处理业务消息
            String topic = message.topicName();
            byte[] payload = message.payload().getBytes();
            upstreamHandler.handleBusinessRequest(endpoint, topic, payload);

            // 2. 根据 QoS 级别发送相应的确认消息
            handleQoSAck(endpoint, message);
        } catch (Exception e) {
            log.error("[processMessage][消息处理失败，断开连接，客户端 ID: {}，地址: {}，错误: {}]",
                    clientId, connectionManager.getEndpointAddress(endpoint), e.getMessage());
            endpoint.close();
        }
    }

    /**
     * 处理 QoS 确认
     *
     * @param endpoint MQTT 连接端点
     * @param message  发布消息
     */
    private void handleQoSAck(MqttEndpoint endpoint, MqttPublishMessage message) {
        if (message.qosLevel() == MqttQoS.AT_LEAST_ONCE) {
            // QoS 1: 发送 PUBACK 确认
            endpoint.publishAcknowledge(message.messageId());
        } else if (message.qosLevel() == MqttQoS.EXACTLY_ONCE) {
            // QoS 2: 发送 PUBREC 确认
            endpoint.publishReceived(message.messageId());
        }
        // QoS 0 无需确认
    }

    /**
     * 清理连接
     *
     * @param endpoint MQTT 连接端点
     */
    private void cleanupConnection(MqttEndpoint endpoint) {
        try {
            // 1. 发送设备离线消息
            IotMqttConnectionManager.ConnectionInfo connectionInfo = connectionManager.getConnectionInfo(endpoint);
            if (connectionInfo != null) {
                IotDeviceMessage offlineMessage = IotDeviceMessage.buildStateOffline();
                deviceMessageService.sendDeviceMessage(offlineMessage, connectionInfo.getProductKey(),
                        connectionInfo.getDeviceName(), serverId);
            }

            // 2. 注销连接
            connectionManager.unregisterConnection(endpoint);
        } catch (Exception e) {
            log.error("[cleanupConnection][清理连接失败，客户端 ID: {}，错误: {}]",
                    endpoint.clientIdentifier(), e.getMessage());
        }
    }

}