package com.yeejoin.equip.eqmx;

import com.alibaba.fastjson.JSONObject;
import com.yeejoin.equip.kafka.KafkaProducerService;
import com.yeejoin.equip.utils.SpringUtils;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
public class MqttConnector implements Connector, MqttCallback {

    // 定时重连线程池
    private static final ScheduledExecutorService reconnectScheduler = Executors.newScheduledThreadPool(4);
    // 初始重连延迟时间（秒）
    private static final int INITIAL_RECONNECT_DELAY = 5;
    // 最大重连延迟时间（秒）
    private static final int MAX_RECONNECT_DELAY = 60;
    // 最大重连尝试次数
    private static final int MAX_RECONNECT_ATTEMPTS = 10;

    private MqttClient mqttClient;
    private final AtomicBoolean connected = new AtomicBoolean(false);
    private final ConnectionConfigDTO connectionConfig;
    // 记录当前重连尝试次数
    private final AtomicInteger reconnectAttempts = new AtomicInteger(0);
    // 用于管理重连任务
    private ScheduledFuture<?> reconnectFuture;

    public MqttConnector(ConnectionConfigDTO connectionConfig) {
        this.connectionConfig = connectionConfig;
    }

    @Override
    public void connect() throws MqttException {
        if (connectionConfig==null) {
            log.info("mqtt config is null or not enabled");
            return;
        }
        log.info("mqtt config: {}", connectionConfig);
        MqttConnectOptions options = buildConnectOptions();
        mqttClient = new MqttClient(connectionConfig.getUrl(), "equip" + connectionConfig.getClientId(), new MemoryPersistence());
        mqttClient.setCallback(this);
        mqttClient.connect(options);
        mqttClient.setTimeToWait(5000);
        mqttClient.subscribe(connectionConfig.getTopics());
        connected.set(true);
        // 连接成功，重置重连尝试次数
        reconnectAttempts.set(0);
        // 取消重连任务
        cancelReconnectTask();
    }

    @Override
    public void disconnect() throws MqttException {
        if (mqttClient != null && mqttClient.isConnected()) {
            mqttClient.disconnect();
            mqttClient.close();
        }
        connected.set(false);
        // 断开连接时取消重连任务
        cancelReconnectTask();
    }

    @Override
    public void connectionLost(Throwable throwable) {
        connected.set(false);
        log.error("mqtt connection lost", throwable);
        // 启动重连流程
        scheduleReconnect();
    }

    @Override
    public void messageArrived(String topic, MqttMessage message) {
       // log.info("mqtt message: {}", message);
        String address="";
        String gatewayId="";
        String value="";
        if("/farm/xiazao/yitihua".equals(topic))
        {
            com.alibaba.fastjson.JSONObject msg = com.alibaba.fastjson.JSONObject.parseObject(message.toString());
            String node = msg.getString("node");
            Map<String,String> map = null;
            if(node.contains("fan2.5"))
            {
                gatewayId="1668801435891929089";
                map=SystemInit.fanmap.get("fan2.5");
            }else if(node.contains("fan3.45"))
            {
                gatewayId="1668801435891929089";
                map=SystemInit.fanmap.get("fan3.45");
            }else if(node.contains("station"))
            {
                gatewayId="1668801570352926721";
                map=SystemInit.map.get(gatewayId);
            }else
            {
                return;
            }
            String values = msg.getString("values");
            JSONObject vobj = JSONObject.parseObject(values,JSONObject.class);
            for(String key:vobj.keySet())
            {
                JSONObject result = new JSONObject();
                if(map!=null&&map.containsKey(key))
                {
                    String oldAddress = map.get(key);
                    result.put("address",oldAddress);
                }else
                {
                    result.put("address",key);
                }
                result.put("value",vobj.get(key));
                result.put("gatewayId",gatewayId);
                //log.info("===========接收MQTT订阅消息，address:{},gatewayId:{},value:{}", address,gatewayId,value);
                //log.info("===========接收IOT订阅消息，address:{},gatewayId:{},dateType:{},value:{},signalType:{}", address,gatewayId,dataType,value,signalType);
                this.connectionConfig.getKafkaProducerService().sendMessageAsync(this.connectionConfig.getKafkaTopic(), com.alibaba.fastjson.JSONObject.toJSONString(result));
            }
        }
        else if("/solar/taihe/yitihua".equals(topic))
        {
            com.alibaba.fastjson.JSONObject msg = com.alibaba.fastjson.JSONObject.parseObject(message.toString());
            String node = msg.getString("node");
            if(node.contains("solar"))
            {
                gatewayId="1669524885619085313";
            }else if(node.contains("station"))
            {
                gatewayId="1669525017559306241";
            }else
            {
                return;
            }
            String values = msg.getString("values");
            JSONObject vobj = JSONObject.parseObject(values,JSONObject.class);
            Map<String,String> map = SystemInit.map.get(gatewayId);
            for(String key:vobj.keySet())
            {
                JSONObject result = new JSONObject();
                if(map!=null&&map.containsKey(key))
                {
                    String oldAddress = map.get(key);
                    result.put("address",oldAddress);
                }else
                {
                    result.put("address",key);
                }
                result.put("value",vobj.get(key));
                result.put("gatewayId",gatewayId);
                //log.info("===========接收MQTT订阅消息，address:{},gatewayId:{},value:{}", address,gatewayId,value);
                //log.info("===========接收IOT订阅消息，address:{},gatewayId:{},dateType:{},value:{},signalType:{}", address,gatewayId,dataType,value,signalType);
                this.connectionConfig.getKafkaProducerService().sendMessageAsync(this.connectionConfig.getKafkaTopic(), com.alibaba.fastjson.JSONObject.toJSONString(result));
            }
        }

    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        try {
            log.info("delivery to {} success: {}", token.getTopics(), token.getMessage());
        } catch (Exception e) {
            log.error("delivery to {} failed: {}", token.getTopics(), e.getMessage());
        }
    }

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

    @Override
    public boolean isConnected() {
        return connected.get();
    }

    @Override
    public boolean checkAlive() {
        // 主动检查MQTT连接状态
        boolean isAlive = mqttClient != null && mqttClient.isConnected();
        // 更新连接状态缓存
        connected.set(isAlive);
        return isAlive;
    }

    private MqttConnectOptions buildConnectOptions() {
        MqttConnectOptions options = new MqttConnectOptions();
        options.setAutomaticReconnect(false);
        options.setCleanSession(true);
        options.setConnectionTimeout(10);
        options.setUserName(this.connectionConfig.getUsername());
//        String password = "";
//        try {
//            password = SymmetricEncryptionUtil.decrypt(this.connectionConfig.getPassword());
//        } catch (Exception e) {
//            log.error("decrypt mqtt password error,pw={}", this.connectionConfig.getPassword(), e);
//        }
        options.setPassword(this.connectionConfig.getPassword().toCharArray());
        return options;
    }

    private void scheduleReconnect() {
        // 若已有未完成的重连任务，不再重复创建
        if (reconnectFuture != null && !reconnectFuture.isDone()) {
            return;
        }
        // 若达到最大重连尝试次数，停止重连
        if (reconnectAttempts.get() >= MAX_RECONNECT_ATTEMPTS) {
            log.error("Reached maximum reconnect attempts ({}) for MQTT client [{}]. Stopping reconnection.",
                    MAX_RECONNECT_ATTEMPTS, connectionConfig.getId());
            return;
        }
        int currentAttempt = reconnectAttempts.incrementAndGet();
        // 计算重连延迟时间，采用指数退避策略
        int delay = Math.min(INITIAL_RECONNECT_DELAY * (int) Math.pow(2, currentAttempt - 1), MAX_RECONNECT_DELAY);

        log.info("Scheduling MQTT client [{}] reconnect attempt {} in {} seconds", connectionConfig.getId(), currentAttempt, delay);

        reconnectFuture = reconnectScheduler.schedule(() -> {
            try {
                if (!connected.get()) {
                    log.info("Trying to reconnect MQTT client [{}]...", connectionConfig.getId());
                    connect();
                    if (connected.get()) {
                        log.info("MQTT client [{}] reconnect success", connectionConfig.getId());
                    }
                }
            } catch (Exception e) {
                log.error("MQTT client [{}] reconnect attempt {} failed",
                        connectionConfig.getId(), currentAttempt, e);
                // 重连失败，继续尝试下一次重连
                scheduleReconnect();
            }
        }, delay, TimeUnit.SECONDS);
    }

    private void cancelReconnectTask() {
        if (reconnectFuture != null) {
            reconnectFuture.cancel(true);
            reconnectFuture = null;
        }
    }
}