package com.inspect.metadata.mqtt;

import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;

@Slf4j
@Service
@ConditionalOnProperty(
        name = {"hua-ruan.mqtt.enable"},
        havingValue = "true"
)
public class MQTTUtils {

    private final MqttProperties properties;

    private MqttClient client;
    private Long id = 0L;

    public MQTTUtils(MqttProperties properties) {
        this.properties = properties;
    }

    public Long getId() {
        return this.id;
    }

    @PostConstruct
    public void init() {
        log.info("MQTT Utils initializing...");
        connectWithRetry();
    }

    private void connectWithRetry() {
        MemoryPersistence persistence = new MemoryPersistence();
        MqttConnectOptions connOpts = new MqttConnectOptions();
        connOpts.setUserName(properties.getUser());
        connOpts.setPassword(properties.getPwd().toCharArray());
        connOpts.setCleanSession(true);

        String broker = String.format("tcp://%s:%d", properties.getIp(), properties.getPort());

        while (true) {
            try {
                client = new MqttClient(broker, properties.getClientId(), persistence);
                client.setCallback(new MqttCallback() {
                    @Override
                    public void connectionLost(Throwable cause) {
                        log.error("Connection lost: {}", cause.getMessage());
                        // 自动重连
                        connectWithRetry();
                    }

                    @Override
                    public void messageArrived(String topic, MqttMessage message) {
                        log.info("messageArrived: {}, message: {}", topic, message.toString());
                    }

                    @Override
                    public void deliveryComplete(IMqttDeliveryToken token) {
                        log.info("Delivery complete: {}", token.getMessageId());
                    }
                });

                log.info("Connecting to MQTT broker: {}", broker);
                client.connect(connOpts);
                log.info("Connected to MQTT broker");

                subscribe(properties.getTopicFilters(), properties.getQos());
                break;
            } catch (MqttException e) {
                log.error("Failed to connect to MQTT broker. Reason: {}", e.getMessage());
                try {
                    log.info("Retrying connection in 30 seconds...");
                    Thread.sleep(30000);
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                    log.error("Connection retry interrupted", ie);
                }
            }
        }
    }

    public void subscribe(String[] topicFilters, int[] qos) {
        try {
            log.info("Subscribing to topicFilters: {}, qos: {}", topicFilters, qos);
            client.subscribe(topicFilters, qos);
        } catch (MqttException e) {
            log.error("Failed to subscribe. Reason: {}", e.getMessage(), e);
        }
    }

    public void disconnect() {
        try {
            log.info("Disconnecting from MQTT broker");
            if (client != null && client.isConnected()) {
                client.disconnect();
            }
            if (client != null) {
                client.close();
            }
        } catch (MqttException e) {
            log.error("Failed to disconnect. Reason: {}", e.getMessage(), e);
        }
    }

    public boolean publish(int qos, String pubTopic, String content) {
        try {
            MqttMessage message = new MqttMessage(content.getBytes());
            message.setQos(qos);
            log.info("Publishing message to topic: {}", pubTopic);
            client.publish(pubTopic, message);
            return true;
        } catch (MqttException e) {
            log.error("Failed to publish message. Reason: {}", e.getMessage(), e);
            return false;
        }
    }

    @PreDestroy
    public void cleanup() {
        disconnect();
    }

    public MqttClient getClient() {
        return client;
    }
}