package com.shzg.project.worn.sensor.config;

import com.shzg.project.worn.domain.MqttTopicConfig;
import com.shzg.project.worn.sensor.mqtt.dispatcher.MqttMessageDispatcher;
import com.shzg.project.worn.service.IMqttTopicConfigService;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Set;
import java.util.UUID;

/**
 * MQTT客户端配置类
 *
 * 支持：
 * 1. 动态订阅（新增topic自动生效）
 * 2. 动态取消订阅（删除topic自动取消）
 * 3. 重连自动恢复订阅
 * 4. 无需重启服务
 */
@Slf4j
@Configuration
public class MqttClientConfig {

    /** MQTT客户端 */
    private MqttClient mqttClient;

    /** 当前已订阅topic缓存 */
    private final Set<String> subscribedTopics = new LinkedHashSet<>();

    /** MQTT配置（缓存一份，供refresh使用） */
    private MqttProperties mqttProperties;

    @Resource
    private MqttMessageDispatcher mqttMessageDispatcher;

    @Resource
    private IMqttTopicConfigService mqttTopicConfigService;

    @Bean
    public MqttClient mqttClient(MqttProperties props) throws MqttException {

        // 开关控制
        if (!props.isEnabled()) {
            log.warn("[MQTT] mqtt已关闭（mqtt.enabled=false），跳过初始化");
            return null;
        }

        // 缓存配置
        this.mqttProperties = props;

        // clientId
        String clientId = "worn-backend-" + UUID.randomUUID();
        mqttClient = new MqttClient(props.getBroker(), clientId, new MemoryPersistence());

        // 连接参数
        MqttConnectOptions options = new MqttConnectOptions();
        options.setCleanSession(props.isCleanSession());
        options.setKeepAliveInterval(props.getKeepAlive());
        options.setConnectionTimeout(props.getTimeout());
        options.setAutomaticReconnect(true);

        if (props.getUsername() != null && !props.getUsername().trim().isEmpty()) {
            options.setUserName(props.getUsername().trim());
        }
        if (props.getPassword() != null && !props.getPassword().trim().isEmpty()) {
            options.setPassword(props.getPassword().toCharArray());
        }

        // 回调
        mqttClient.setCallback(new MqttCallbackExtended() {

            @Override
            public void connectComplete(boolean reconnect, String serverURI) {
                log.info("[MQTT] connected reconnect={}, serverURI={}", reconnect, serverURI);
                refreshSubscribe(); // 重连自动恢复订阅
            }

            @Override
            public void connectionLost(Throwable cause) {
                log.warn("[MQTT] 连接已断开", cause);
            }

            @Override
            public void messageArrived(String topic, MqttMessage message) {
                String payload = new String(message.getPayload(), StandardCharsets.UTF_8);
                log.info("[MQTT] message arrived topic={} payload={}", topic, payload);
                mqttMessageDispatcher.dispatch(topic, payload);
            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken token) {
                // 可扩展
            }
        });

        // 连接
        log.info("[MQTT] connecting broker={}", props.getBroker());
        mqttClient.connect(options);
        log.info("[MQTT] connected");

        // 初始订阅
        refreshSubscribe();

        return mqttClient;
    }

    /**
     * 动态刷新订阅（线程安全）
     */
    public synchronized void refreshSubscribe() {

        if (mqttClient == null || !mqttClient.isConnected()) {
            log.warn("[MQTT] 刷新订阅跳过（客户端未连接）");
            return;
        }

        try {
            // 查询数据库
            List<MqttTopicConfig> configs =
                    mqttTopicConfigService.selectEnabledMqttTopicConfigList();

            Set<String> newTopics = new LinkedHashSet<>();

            for (MqttTopicConfig c : configs) {
                if (c.getTopicUp() != null && !c.getTopicUp().trim().isEmpty()) {
                    newTopics.add(c.getTopicUp().trim());
                }
            }

            if (newTopics.isEmpty()) {
                log.warn("[MQTT] 未查询到任何可用的topic_up配置");
            }

            // ================== 新增订阅 ==================
            for (String topic : newTopics) {
                if (!subscribedTopics.contains(topic)) {
                    mqttClient.subscribe(topic, mqttProperties.getQos());
                    log.info("[MQTT] subscribe new topic={}", topic);
                }
            }

            // ================== 取消订阅 ==================
            for (String topic : new LinkedHashSet<>(subscribedTopics)) {
                if (!newTopics.contains(topic)) {
                    mqttClient.unsubscribe(topic);
                    log.info("[MQTT] unsubscribe topic={}", topic);
                }
            }

            // 更新缓存
            subscribedTopics.clear();
            subscribedTopics.addAll(newTopics);

        } catch (Exception e) {
            log.error("[MQTT] refreshSubscribe failed", e);
        }
    }

    @PreDestroy
    public void destroy() {
        try {
            if (mqttClient != null) {
                if (mqttClient.isConnected()) {
                    mqttClient.disconnect();
                }
                mqttClient.close();
            }
        } catch (Exception ignored) {
        }
    }
}