package com.xzzn.framework.manager;

import com.xzzn.ems.service.IEmsAlarmRecordsService;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.context.SmartLifecycle;
import org.springframework.stereotype.Component;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

@Component
public class MqttLifecycleManager implements ApplicationRunner, SmartLifecycle, MqttCallbackExtended {

    private static final Logger log = LoggerFactory.getLogger(MqttLifecycleManager.class);
    private static final long RECONNECT_DELAY_SECONDS = 5;

    private final MqttConnectOptions connectOptions;
    private final IEmsAlarmRecordsService iEmsAlarmRecordsService;
    private final ScheduledExecutorService reconnectExecutor = Executors.newSingleThreadScheduledExecutor(r -> {
        Thread thread = new Thread(r);
        thread.setName("mqtt-reconnect");
        thread.setDaemon(true);
        return thread;
    });
    private final AtomicBoolean reconnectScheduled = new AtomicBoolean(false);
    private volatile ScheduledFuture<?> reconnectFuture;
    private MqttClient mqttClient;
    private volatile boolean running = false;

    // 存储订阅关系: topic -> (listener, qos)
    private final ConcurrentHashMap<String, SubscriptionInfo> subscriptions = new ConcurrentHashMap<>();

    @Autowired
    public MqttLifecycleManager(MqttConnectOptions connectOptions, IEmsAlarmRecordsService iEmsAlarmRecordsService) {
        this.connectOptions = connectOptions;
        this.iEmsAlarmRecordsService = iEmsAlarmRecordsService;
    }

    // Spring Boot 启动完成后执行
    @Override
    public void run(ApplicationArguments args) throws Exception {
        start();
    }

    @Override
    public void start() {
        if (running) return;

        try {
            String prefix = connectOptions.getUserName() == null || connectOptions.getUserName().isEmpty()
                    ? "mqtt-client" : connectOptions.getUserName();
            String clientId = prefix + "-" + System.currentTimeMillis();
            mqttClient = new MqttClient(
                    connectOptions.getServerURIs()[0],
                    clientId,
                    new MemoryPersistence()
            );

            mqttClient.setCallback(this);
            mqttClient.connect(connectOptions);

            resubscribeAll();
            running = true;
            log.info("MQTT client connected to: {}", connectOptions.getServerURIs()[0]);
        } catch (MqttException e) {
            running = false;
            log.error("MQTT connection failed: {}", e.getMessage(), e);
            scheduleReconnect();
        }
    }

    @Override
    public void stop() {
        cancelReconnectTask();
        if (mqttClient != null && mqttClient.isConnected()) {
            try {
                mqttClient.disconnect();
                mqttClient.close();
            } catch (MqttException e) {
                log.warn("Error disconnecting MQTT client: {}", e.getMessage(), e);
            }
        }
        reconnectExecutor.shutdownNow();
        running = false;
    }

    @Override
    public boolean isRunning() {
        return running;
    }

    // MQTT 回调方法
    @Override
    public void connectionLost(Throwable cause) {
        log.warn("MQTT connection lost: {}", cause == null ? "unknown" : cause.getMessage(), cause);
        running = false;
        scheduleReconnect();
    }

    @Override
    public void connectComplete(boolean reconnect, String serverURI) {
        running = true;
        cancelReconnectTask();
        log.info("MQTT connection complete, reconnect: {}, serverURI: {}", reconnect, serverURI);
        resubscribeAll();
    }

    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        SubscriptionInfo info = subscriptions.get(topic);
        if (info != null && info.getListener() != null) {
            info.getListener().messageArrived(topic, message);
        }
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        // 消息发布完成处理
    }

    // 订阅方法
    public void subscribe(String topic, int qos, IMqttMessageListener listener) {
        try {
            if (mqttClient != null && mqttClient.isConnected()) {
                mqttClient.subscribe(topic, qos);
                log.info("MQTT subscribe success, topic: {}, qos: {}", topic, qos);
            } else {
                log.warn("MQTT subscribe deferred, client not connected, topic: {}", topic);
            }
            subscriptions.put(topic, new SubscriptionInfo(listener, qos));
        } catch (MqttException e) {
            log.error("Subscribe failed, topic: {}, err: {}", topic, e.getMessage(), e);
            iEmsAlarmRecordsService.addSubFailedAlarmRecord(topic);
            scheduleReconnect();
        }
        iEmsAlarmRecordsService.checkFailedRecord(topic);
    }

    // 发布方法
    public void publish(String topic, String payload, int qos) throws MqttException {
        if (mqttClient != null && mqttClient.isConnected()) {
            MqttMessage message = new MqttMessage(payload.getBytes());
            message.setQos(qos);
            mqttClient.publish(topic, message);
        } else {
            throw new MqttException(MqttException.REASON_CODE_CLIENT_NOT_CONNECTED);
        }
    }

    // 重新订阅所有主题
    private void resubscribeAll() {
        if (mqttClient == null || !mqttClient.isConnected()) return;

        subscriptions.forEach((topic, info) -> {
            try {
                mqttClient.subscribe(topic, info.getQos());
                log.info("MQTT resubscribe success, topic: {}, qos: {}", topic, info.getQos());
            } catch (MqttException e) {
                log.error("Resubscribe failed for topic: {}, err: {}", topic, e.getMessage(), e);
            }
        });
    }

    private void scheduleReconnect() {
        if (mqttClient == null || reconnectExecutor.isShutdown()) {
            return;
        }
        if (!reconnectScheduled.compareAndSet(false, true)) {
            return;
        }

        reconnectFuture = reconnectExecutor.scheduleWithFixedDelay(() -> {
            if (mqttClient == null) {
                cancelReconnectTask();
                return;
            }
            if (mqttClient.isConnected()) {
                cancelReconnectTask();
                return;
            }
            try {
                log.info("MQTT reconnecting...");
                mqttClient.connect(connectOptions);
                running = true;
                cancelReconnectTask();
                resubscribeAll();
                log.info("MQTT reconnect success.");
            } catch (MqttException e) {
                running = false;
                log.warn("MQTT reconnect failed: {}", e.getMessage());
            }
        }, RECONNECT_DELAY_SECONDS, RECONNECT_DELAY_SECONDS, TimeUnit.SECONDS);
    }

    private void cancelReconnectTask() {
        reconnectScheduled.set(false);
        ScheduledFuture<?> future = reconnectFuture;
        if (future != null && !future.isCancelled()) {
            future.cancel(false);
        }
    }

    // 订阅信息内部类
    private static class SubscriptionInfo {
        private final IMqttMessageListener listener;
        private final int qos;

        public SubscriptionInfo(IMqttMessageListener listener, int qos) {
            this.listener = listener;
            this.qos = qos;
        }

        public IMqttMessageListener getListener() {
            return listener;
        }

        public int getQos() {
            return qos;
        }
    }
}
