package com.iot.scheduler.utils;

import io.micrometer.common.util.StringUtils;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;

/**
 * MQTT客户端工具类
 * 支持连接、发布、订阅、断开连接等操作
 */
@Slf4j
public class MqttUtils {

    private MqttClient mqttClient;
    private MqttConnectOptions connectOptions;
    private ConnectionCallback connectionCallback;
    private MessageCallback messageCallback;
    private boolean isConnected = false;
    private List<String> subscribedTopics = new ArrayList<>();

    /**
     * 连接回调接口
     */
    public interface ConnectionCallback {
        void onConnected();

        void onDisconnected();

        void onConnectionFailed(Throwable cause);

        void onConnectionLost(Throwable cause);
    }

    /**
     * 消息回调接口
     */
    public interface MessageCallback {
        void onMessageReceived(String topic, String message);

        void onMessageDelivered(IMqttDeliveryToken token);
    }

    /**
     * MQTT连接配置构建器
     */
    public static class MqttConfig {
        private String broker;
        private String clientId;
        private String username;
        private String password;
        private int connectionTimeout = 10;
        private int keepAliveInterval = 60;
        private boolean cleanSession = true;
        private boolean automaticReconnect = true;
        private int maxInflight = 10;
        private int qos = 1;

        public MqttConfig(String broker) {
            this.broker = broker;
            this.clientId = "Client_" + UUID.randomUUID().toString().substring(0, 8);
        }

        public MqttConfig setBroker(String broker) {
            this.broker = broker;
            return this;
        }

        public MqttConfig setClientId(String clientId) {
            this.clientId = clientId;
            return this;
        }

        public MqttConfig setUsername(String username) {
            this.username = username;
            return this;
        }

        public MqttConfig setPassword(String password) {
            this.password = password;
            return this;
        }

        public MqttConfig setConnectionTimeout(int connectionTimeout) {
            this.connectionTimeout = connectionTimeout;
            return this;
        }

        public MqttConfig setKeepAliveInterval(int keepAliveInterval) {
            this.keepAliveInterval = keepAliveInterval;
            return this;
        }

        public MqttConfig setCleanSession(boolean cleanSession) {
            this.cleanSession = cleanSession;
            return this;
        }

        public MqttConfig setAutomaticReconnect(boolean automaticReconnect) {
            this.automaticReconnect = automaticReconnect;
            return this;
        }

        public MqttConfig setMaxInflight(int maxInflight) {
            this.maxInflight = maxInflight;
            return this;
        }

        public MqttConfig setQos(int qos) {
            this.qos = qos;
            return this;
        }

        public String getBroker() {
            return broker;
        }

        public String getClientId() {
            return clientId;
        }

        public String getUsername() {
            return username;
        }

        public String getPassword() {
            return password;
        }

        public int getConnectionTimeout() {
            return connectionTimeout;
        }

        public int getKeepAliveInterval() {
            return keepAliveInterval;
        }

        public boolean isCleanSession() {
            return cleanSession;
        }

        public boolean isAutomaticReconnect() {
            return automaticReconnect;
        }

        public int getMaxInflight() {
            return maxInflight;
        }

        public int getQos() {
            return qos;
        }
    }

    /**
     * 创建MQTT工具实例
     */
    public MqttUtils() {
    }

    /**
     * 连接到MQTT服务器
     *
     * @param config 连接配置
     * @throws MqttException 连接异常
     */
    public void connect(MqttConfig config) throws MqttException {
        connect(config, null, null);
    }

    /**
     * 连接到MQTT服务器（带回调）
     *
     * @param config       连接配置
     * @param connCallback 连接回调
     * @param msgCallback  消息回调
     * @throws MqttException 连接异常
     */
    public void connect(MqttConfig config, ConnectionCallback connCallback, MessageCallback msgCallback)
            throws MqttException {

        this.connectionCallback = connCallback;
        this.messageCallback = msgCallback;

        try {
            // 创建MQTT客户端
            String brokerUrl = config.getBroker().startsWith("tcp://") ?
                    config.getBroker() : "tcp://" + config.getBroker();

            mqttClient = new MqttClient(brokerUrl, config.getClientId(), new MemoryPersistence());

            // 设置连接选项
            connectOptions = new MqttConnectOptions();
            connectOptions.setCleanSession(config.isCleanSession());
            connectOptions.setConnectionTimeout(config.getConnectionTimeout());
            connectOptions.setKeepAliveInterval(config.getKeepAliveInterval());
            connectOptions.setAutomaticReconnect(config.isAutomaticReconnect());
            connectOptions.setMaxInflight(config.getMaxInflight());

            // 设置用户名密码（如果有）
            if (config.getUsername() != null && !config.getUsername().isEmpty()) {
                connectOptions.setUserName(config.getUsername());
                if (config.getPassword() != null && !config.getPassword().isEmpty()) {
                    connectOptions.setPassword(config.getPassword().toCharArray());
                }
            }

            // 设置回调
            mqttClient.setCallback(new MqttCallbackExtended() {
                @Override
                public void connectComplete(boolean reconnect, String serverURI) {
                    isConnected = true;
                    log.info("MQTT连接成功: " + serverURI + (reconnect ? " (重连)" : ""));
                    if (connectionCallback != null) {
                        connectionCallback.onConnected();
                    }
                }

                @Override
                public void connectionLost(Throwable cause) {
                    isConnected = false;
                    log.error("MQTT连接断开: " + cause.getMessage());
                    if (connectionCallback != null) {
                        connectionCallback.onConnectionLost(cause);
                    }
                }

                @Override
                public void messageArrived(String topic, MqttMessage message) {
                    String payload = new String(message.getPayload(), StandardCharsets.UTF_8);
                    log.info("收到消息 - Topic: " + topic + ", QoS: " + message.getQos() +
                            ", Message: " + payload);

                    if (messageCallback != null) {
                        messageCallback.onMessageReceived(topic, payload);
                    }
                }

                @Override
                public void deliveryComplete(IMqttDeliveryToken token) {
                    log.info("消息发送完成");
                    if (messageCallback != null) {
                        messageCallback.onMessageDelivered(token);
                    }
                }
            });

            // 连接到服务器
            log.info("正在连接到: " + brokerUrl);
            mqttClient.connect(connectOptions);
            isConnected = mqttClient.isConnected();

        } catch (MqttException e) {
            isConnected = false;
            log.error("连接失败: " + e.getMessage());
            if (connectionCallback != null) {
                connectionCallback.onConnectionFailed(e);
            }
            throw e;
        }
    }

    /**
     * 发布消息
     *
     * @param topic    主题
     * @param message  消息内容
     * @param qos      QoS级别 (0, 1, 2)
     * @param retained 是否保留消息
     * @throws MqttException 发布异常
     */
    public void publish(String topic, String message, int qos, boolean retained) throws MqttException {
        if (mqttClient == null) {
            throw new MqttException(MqttException.REASON_CODE_CLIENT_NOT_CONNECTED);
        }

        int attempts = 0;
        while (true) {
            if (!mqttClient.isConnected()) {
                try {
                    if (connectOptions != null) {
                        mqttClient.connect(connectOptions);
                        isConnected = mqttClient.isConnected();
                    }
                } catch (MqttException ce) {
                    attempts++;
                    if (attempts >= 2) {
                        log.error("重连失败: " + ce.getMessage());
                        throw ce;
                    }
                    try {
                        Thread.sleep(100);
                    } catch (InterruptedException ie) {
                        Thread.currentThread().interrupt();
                    }
                    continue;
                }
            }

            try {
                MqttMessage mqttMessage = new MqttMessage(message.getBytes(StandardCharsets.UTF_8));
                mqttMessage.setQos(qos);
                mqttMessage.setRetained(retained);

                mqttClient.publish(topic, mqttMessage);
                log.info("发布成功 - Topic: " + topic + ", QoS: " + qos +
                        ", Retained: " + retained + ", Message: " + message);
                return;
            } catch (MqttException e) {
                attempts++;
                if (attempts >= 2) {
                    log.error("发布失败: " + e.getMessage());
                    throw e;
                }
                try {
                    Thread.sleep(100);
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                }
            }
        }
    }

    /**
     * 发布消息（简化版）
     *
     * @param topic   主题
     * @param message 消息内容
     * @throws MqttException 发布异常
     */
    public void publish(String topic, String message) throws MqttException {
        publish(topic, message, 1, false);
    }

    /**
     * 订阅主题
     *
     * @param topic 主题
     * @param qos   QoS级别
     * @throws MqttException 订阅异常
     */
    public void subscribe(String topic, int qos) throws MqttException {
        if (!isConnected) {
            throw new MqttException(MqttException.REASON_CODE_CLIENT_NOT_CONNECTED);
        }

        try {
            mqttClient.subscribe(topic, qos);
            subscribedTopics.add(topic);
            log.info("订阅成功 - Topic: " + topic + ", QoS: " + qos);
        } catch (MqttException e) {
            log.error("订阅失败: " + e.getMessage());
            throw e;
        }
    }

    /**
     * 订阅多个主题
     *
     * @param topics 主题数组
     * @param qos    QoS级别数组
     * @throws MqttException 订阅异常
     */
    public void subscribe(String[] topics, int[] qos) throws MqttException {
        if (!isConnected) {
            throw new MqttException(MqttException.REASON_CODE_CLIENT_NOT_CONNECTED);
        }

        try {
            mqttClient.subscribe(topics, qos);
            for (String topic : topics) {
                subscribedTopics.add(topic);
            }
            log.info("批量订阅成功 - 主题数量: " + topics.length);
        } catch (MqttException e) {
            log.error("批量订阅失败: " + e.getMessage());
            throw e;
        }
    }

    /**
     * 取消订阅
     *
     * @param topic 主题
     * @throws MqttException 取消订阅异常
     */
    public void unsubscribe(String topic) throws MqttException {
        if (!isConnected) {
            throw new MqttException(MqttException.REASON_CODE_CLIENT_NOT_CONNECTED);
        }

        try {
            mqttClient.unsubscribe(topic);
            subscribedTopics.remove(topic);
            log.info("取消订阅成功 - Topic: " + topic);
        } catch (MqttException e) {
            log.error("取消订阅失败: " + e.getMessage());
            throw e;
        }
    }

    /**
     * 断开连接
     */
    public void disconnect() {
        if (mqttClient != null && isConnected) {
            try {
                // 取消所有订阅
                if (!subscribedTopics.isEmpty()) {
                    String[] topics = subscribedTopics.toArray(new String[0]);
                    mqttClient.unsubscribe(topics);
                    subscribedTopics.clear();
                }

                // 断开连接
                mqttClient.disconnect();
                isConnected = false;
                log.info("MQTT连接已断开");

                if (connectionCallback != null) {
                    connectionCallback.onDisconnected();
                }

            } catch (MqttException e) {
                log.error("断开连接异常: " + e.getMessage());
            }
        }
    }

    /**
     * 判断是否已连接
     *
     * @return 连接状态
     */
    public boolean isConnected() {
        return isConnected && mqttClient != null && mqttClient.isConnected();
    }

    /**
     * 获取客户端ID
     *
     * @return 客户端ID
     */
    public String getClientId() {
        return mqttClient != null ? mqttClient.getClientId() : null;
    }

    /**
     * 获取服务器URI
     *
     * @return 服务器URI
     */
    public String getServerURI() {
        return mqttClient != null ? mqttClient.getServerURI() : null;
    }

    /**
     * 获取已订阅的主题列表
     *
     * @return 主题列表
     */
    public List<String> getSubscribedTopics() {
        return new ArrayList<>(subscribedTopics);
    }

    /**
     * 关闭客户端（释放资源）
     */
    public void close() {
        disconnect();
        try {
            if (mqttClient != null) {
                mqttClient.close();
                log.info("MQTT客户端已关闭");
            }
        } catch (MqttException e) {
            log.error("关闭客户端异常: " + e.getMessage());
        }
    }

    /**
     * 快速连接并发送消息
     */
    public static void quickPublish(String broker, String topic, String username,
                                    String password, String clientId, String message) throws Exception {
        MqttUtils mqttUtils = new MqttUtils();

        MqttConfig config = new MqttConfig(broker);

        if (StringUtils.isNotBlank(username)) {
            config.setUsername(username);
        }
        if (StringUtils.isNotBlank(password)) {
            config.setPassword(password);
        }
        if (StringUtils.isNotBlank(clientId)) {
            config.setClientId(clientId);
        }

        config.setConnectionTimeout(15)
                .setKeepAliveInterval(30)
                .setQos(1)
                .setMaxInflight(50);

        try {
            mqttUtils.connect(config);
            mqttUtils.publish(topic, message);
            log.info("消息发送成功!");
        } finally {
            mqttUtils.disconnect();
        }
    }

    public static void publish(String broker, String topic, String username,
                               String password, String clientId, String message) throws Exception {
        MqttUtils mqttUtils = new MqttUtils();
        MqttUtils.MqttConfig config = new MqttUtils.MqttConfig(broker)
                .setClientId(clientId)
                .setUsername(username)
                .setPassword(password)
                .setConnectionTimeout(15)
                .setKeepAliveInterval(30)
                .setQos(1)
                .setMaxInflight(50);

        try {
            mqttUtils.connect(config);

            mqttUtils.publish(topic, message);

        } catch (MqttException e) {
            log.error("首次发布失败: " + e.getMessage());
            try {
                mqttUtils.disconnect();
            } catch (Exception ignore) {
            }
            try {
                mqttUtils.connect(config);
                mqttUtils.publish(topic, message);
            } catch (MqttException ee) {
                log.error("重试发布失败: " + ee.getMessage());
                throw ee;
            }
        } finally {
            mqttUtils.disconnect();
        }
    }


    public static void main(String[] args) {

        String broker = "10.9.0.205:1883";
        String clientId = "QesP7NB4PpAQhupFxlY";
        String user = "admin";
        String pwd = "123456";
        String topic = "v1/devices/me/telemetry";
        String message = "{\"temp\": 12.8,\"status\": \"ON\"}";

        try {
            quickPublish(broker, topic, user, pwd, clientId, message);
        } catch (Exception e) {
            e.printStackTrace();
        }

//        MqttUtils mqttUtils = new MqttUtils();
//        MqttUtils.MqttConfig config = new MqttUtils.MqttConfig(broker)
//                .setClientId(clientId)
//                .setUsername(user)
//                .setPassword(pwd)
//                .setConnectionTimeout(15)
//                .setKeepAliveInterval(30)
//                .setQos(1);
//
//        try {
//            // 连接
//            mqttUtils.connect(config);
//
//            // 发布消息
//            mqttUtils.publish(topic, message);
//
//        } catch (MqttException e) {
//            e.printStackTrace();
//        } finally {
//            // 断开连接
//            mqttUtils.disconnect();
//        }
    }
}