package com.agri.framework.manager;

import com.agri.common.utils.wechat.WxUtil;
import com.agri.framework.web.dispatcher.MqttMessageDispatcher;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.SmartLifecycle;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * MQTT客户端生命周期管理器
 * 核心功能：
 * 1. 订阅设备状态、前端控制指令主题
 * 2. MQTT客户端启动/停止/重连
 * 3. 管理自动关闭任务线程池
 * 4. 监听MQTT连接状态，设置消息回调
 * 适配JDK 8，无心跳包相关逻辑
 */
@Component
public class MqttClientManager implements SmartLifecycle {

    /**
     * 优化：统一使用SLF4J日志（JDK 8兼容）
     */
    private static final Logger log = LoggerFactory.getLogger(MqttClientManager.class);

    /**
     * 新增：生命周期管理标识，控制MQTT客户端启动/关闭
     */
    private final AtomicBoolean isRunning = new AtomicBoolean(false);

    /**
     * MQTT客户端（由MqttConfig配置类注入）
     */
    @Resource
    private MqttClient mqttClient;

    /**
     * MQTT连接配置项（从MqttConfig注入）
     */
    @Resource
    private MqttConnectOptions mqttConnectOptions;

    /**
     * MQTT消息分发器，转发消息到对应处理器
     */
    @Resource
    private MqttMessageDispatcher mqttMessageDispatcher;

    /**
     * 自动关任务管理器，初始化/关闭线程池
     */
    @Resource
    private MqttAutoOffManager mqttAutoOffManager;

    // 读取配置文件中的默认订阅主题（移除心跳主题）
    @Value("${spring.mqtt.default-topic}")
    private String defaultTopic;

    // 新增：自动关闭任务线程池核心线程数（可配置）
    @Value("${spring.mqtt.auto-off-thread-pool-size:5}")
    private int autoOffThreadPoolSize;

    private final ThreadPoolExecutor mqttBizPool =
            new ThreadPoolExecutor(
                    8,                     // core
                    16,                    // max
                    60, TimeUnit.SECONDS,
                    new ArrayBlockingQueue<>(2000), // 有界、较小
                    r -> {
                        Thread t = new Thread(r);
                        t.setName("mqtt-biz-" + t.getId());
                        t.setDaemon(true);
                        return t;
                    },
                    new ThreadPoolExecutor.DiscardPolicy() // 直接丢
            );
    /**
     * 初始化：订阅主题+设置回调
     * （移除@PostConstruct，改为由SmartLifecycle的start()触发）
     * <p>
     * 【方案A】不做自写重连；Paho会在连接断开后自动重连（前提：connectOptions.setAutomaticReconnect(true)）
     */
    public void subscribeTopics() throws MqttException {
        // 关键补充1：判空
        if (mqttClient == null) {
            log.error("【MQTT初始化】客户端实例为空，无法订阅主题");
            throw new MqttException(MqttException.REASON_CODE_CLIENT_NOT_CONNECTED);
        }

        // 解析配置的主题列表
        final String[] topicsFinal = Arrays.stream(defaultTopic.split(","))
                .map(String::trim).toArray(String[]::new);
        final int[] qosFinal = new int[topicsFinal.length];
        Arrays.fill(qosFinal, 0);


        // 设置MQTT消息回调：处理连接断开、消息接收、消息发布完成
        mqttClient.setCallback(new MqttCallbackExtended() {
            @Override
            public void connectComplete(boolean reconnect, String serverURI) {
                log.info("【MQTT连接完成】reconnect={}, serverURI={}, clientId={}",
                        reconnect, serverURI, safeClientId());
                // cleanSession=true：重连后必须补订阅
                if (reconnect) {
                    try {
                        mqttClient.subscribe(topicsFinal, qosFinal);
                        log.info("【MQTT订阅恢复】topicsFinal={}", String.join(",", topicsFinal));
                    } catch (Exception e) {
                        log.error("【MQTT订阅恢复失败】", e);
                    }
                }
            }

            /**
             * MQTT连接断开回调
             * @param cause 断开原因
             */
            @Override
            public void connectionLost(Throwable cause) {
                log.info("autoReconnect={}, cleanSession={}, keepAlive={}",
                        mqttConnectOptions.isAutomaticReconnect(),
                        mqttConnectOptions.isCleanSession(),
                        mqttConnectOptions.getKeepAliveInterval());

                log.error("【MQTT连接异常】连接断开，clientId：{}，原因：{}",
                        safeClientId(), (cause == null ? "unknown" : cause.getMessage()), cause);
                WxUtil.pushText("【MQTT连接异常】连接断开：\n  clientId："+safeClientId()+"\n  Cause: "+cause);
                // 【方案A】不再触发自写重连；Paho自动重连会接管重连过程
                // 这里只记录日志即可
                if (isRunning.get()) {
                    log.warn("【MQTT自动重连】已开启automaticReconnect，等待Paho自动重连...");
                }
            }

            /**
             * 收到MQTT消息回调：核心处理入口
             * @param topic 消息主题
             * @param message 消息内容
             * @throws Exception 消息处理异常
             */
            @Override
            public void messageArrived(String topic, MqttMessage message) throws Exception {

                final String payload = new String(message.getPayload(), StandardCharsets.UTF_8);
                if (message.isRetained()) {
                    log.info("ignore retained snapshot: {}", topic);
                    return;
                }

                mqttBizPool.execute(() -> {
                    // log.debug("mqttBizPool active={}, queue={}",
                    //         mqttBizPool.getActiveCount(),
                    //         mqttBizPool.getQueue().size());
                    if (mqttBizPool.getActiveCount()>10 || mqttBizPool.getQueue().size()>1000) {
                        WxUtil.pushText("线程池繁忙! \n  正在处理中任务："+mqttBizPool.getActiveCount()+"\n  剩余待进行任务："+mqttBizPool.getQueue().size());
                    }
                    try {
                        // 优化：显式指定UTF-8编码，避免乱码（JDK 8兼容）
                        mqttMessageDispatcher.handleMessage(topic, payload);
                    } catch (Exception e) {
                        log.error("【MQTT消息处理异常】topic={}, payload={}", topic, payload, e);
                    }
                });

            }

            /**
             * 消息发布完成回调（仅日志记录，无业务逻辑）
             * @param token 发布令牌
             */
            @Override
            public void deliveryComplete(IMqttDeliveryToken token) {
                if (token != null && token.getTopics() != null && token.getTopics().length > 0) {
                    // log.info("【MQTT确认】消息发布完成，clientId：{}，主题：{}", safeClientId(), token.getTopics()[0]);
                }
            }
        });
        if (!mqttClient.isConnected()) {
            try {
                // 使用注入的连接配置项连接Broker（带用户名密码、自动重连等配置）
                mqttClient.connect(mqttConnectOptions);
                log.info("【MQTT连接】客户端已成功连接到Broker，clientId：{}", mqttClient.getClientId());
            } catch (MqttException e) {
                log.error("【MQTT连接】连接Broker失败，clientId：{}", mqttClient.getClientId(), e);
                throw e;
            }
        }
        // 订阅主题
        // connect 后首次订阅
        try {
            mqttClient.subscribe(topicsFinal, qosFinal);
            log.info("【MQTT初始化】订阅主题完成，clientId：{}，topicsFinal={}",
                    mqttClient.getClientId(), String.join(",", topicsFinal));
        } catch (MqttException e) {
            log.error("【MQTT初始化】订阅失败，clientId：{}，topicsFinal={}",
                    safeClientId(), String.join(",", topicsFinal), e);
            throw e;
        }
    }

    private String safeClientId() {
        try {
            return (mqttClient == null ? "null" : mqttClient.getClientId());
        } catch (Exception e) {
            return "unknown";
        }
    }

    // ========== 手动重连接口（供Controller调用） ==========

    /**
     * 手动触发MQTT重连（最小改动：不替换client实例，只用同一个client重连）
     */
    public synchronized String manualReconnect() {
        isRunning.set(true);
        try {
            // 强制断开旧连接（如果存在）
            if (mqttClient != null && mqttClient.isConnected()) {
                mqttClient.disconnect();
            }
            // 重新初始化订阅（内部会connect + subscribe）
            subscribeTopics();
            log.info("【手动重连】MQTT客户端重连成功");
            return "MQTT手动重连成功";
        } catch (MqttException e) {
            log.error("【手动重连】MQTT客户端重连失败", e);
            return "MQTT手动重连失败：" + e.getMessage();
        }
    }

    /**
     * 获取当前MQTT连接状态
     */
    public Map<String,Object> getMqttStatus() {
        boolean connected = (mqttClient != null && mqttClient.isConnected());
        String status = connected ? "已连接" : "已断开";

        Map<String, Object> data = new HashMap<>();
        data.put("enabled", mqttAutoOffManager.isEnabled());
        data.put("mqtt", status);
        data.put("clientId", safeClientId());
        data.put("executorInited", mqttAutoOffManager.isExecutorInited());
        data.put("totalTaskCount", mqttAutoOffManager.getTotalTaskCount());
        data.put("deviceTaskCount", mqttAutoOffManager.getDeviceTaskCount());
        return data;
    }

    // ======================== SmartLifecycle 生命周期管理（核心修复） ========================

    /**
     * 启动MQTT客户端（Spring上下文初始化/重启时触发）
     * 核心：替代@PostConstruct，保证上下文重启时重新初始化MQTT连接
     */
    @Override
    public void start() {
        log.info("开始监听");
        if (isRunning.compareAndSet(false, true)) {
            try {
                // 初始化自动关任务线程池
                mqttAutoOffManager.initExecutor(autoOffThreadPoolSize);
                // 核心修改：无论是否已连接，都执行订阅
                subscribeTopics();
                log.info("【MQTT生命周期】客户端启动成功（已设置回调+订阅主题），自动关闭任务线程池大小：{}", autoOffThreadPoolSize);
            } catch (MqttException e) {
                log.error("【MQTT生命周期】客户端启动失败", e);
                isRunning.set(false);
            }
        }
    }

    /**
     * 停止MQTT客户端
     * 改造点：优化多线程池的优雅关闭
     */
    @Override
    public void stop() {
        if (isRunning.compareAndSet(true, false)) {
            try {
                // 关闭自动关任务线程池
                mqttAutoOffManager.shutdownExecutor();
                // 关闭MQTT客户端
                if (mqttClient != null) {
                    if (mqttClient.isConnected()) {
                        mqttClient.disconnect();
                    }
                    mqttClient.close();
                    log.info("【MQTT生命周期】客户端已优雅关闭");
                }
            } catch (Exception e) {
                log.error("【MQTT生命周期】客户端关闭失败", e);
            }
        }
    }

    /**
     * 异步停止
     */
    @Override
    public void stop(Runnable callback) {
        stop();
        callback.run();
    }

    /**
     * 判断MQTT客户端是否运行中
     */
    @Override
    public boolean isRunning() {
        return isRunning.get();
    }

    /**
     * 启动优先级（保证MQTT在Redis之后启动）
     */
    @Override
    public int getPhase() {
        return 10;
    }

    /**
     * 是否自动启动（默认true，Spring上下文初始化时自动调用start()）
     */
    @Override
    public boolean isAutoStartup() {
        return true;
    }
}