package cn.iocoder.yudao.module.mes.framework.collect.mqtt;

import cn.hutool.core.util.StrUtil;
import cn.iocoder.yudao.module.mes.framework.collect.config.MesCollectProperties;
import cn.iocoder.yudao.module.mes.service.collect.MesCollectMessageDispatcher;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.context.SmartLifecycle;
import org.springframework.stereotype.Component;

import java.net.InetAddress;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * MES 数采 MQTT 客户端
 *
 * @author 超级管理员
 */
@Component
@Slf4j
public class MesCollectMqttClient implements SmartLifecycle {

    private static final long RECONNECT_DELAY_SECONDS = 30L;

    @Resource
    private MesCollectProperties properties;
    @Resource
    private MesCollectMessageDispatcher dispatcher;

    private volatile boolean running = false;
    private MqttClient client;
    private ScheduledExecutorService scheduler;

    @Override
    public void start() {
        MesCollectProperties.Mqtt mqtt = properties.getMqtt();
        if (!Boolean.TRUE.equals(mqtt.getEnabled())) {
            log.info("[start][数采 MQTT 未启用]");
            return;
        }
        if (StrUtil.isBlank(mqtt.getUrl())) {
            log.warn("[start][数采 MQTT 已启用但 url 为空，跳过连接]");
            return;
        }
        running = true;
        scheduler = Executors.newSingleThreadScheduledExecutor(runnable -> {
            Thread thread = new Thread(runnable, "mes-collect-mqtt-reconnect");
            thread.setDaemon(true);
            return thread;
        });
        scheduler.scheduleWithFixedDelay(this::ensureConnected, 0, RECONNECT_DELAY_SECONDS, TimeUnit.SECONDS);
    }

    private void ensureConnected() {
        if (!running) {
            return;
        }
        MesCollectProperties.Mqtt mqtt = properties.getMqtt();
        try {
            if (client == null) {
                client = new MqttClient(mqtt.getUrl(), buildClientId(), new MemoryPersistence());
                client.setCallback(new MqttCallbackExtended() {
                    @Override
                    public void connectComplete(boolean reconnect, String serverURI) {
                        subscribe();
                    }

                    @Override
                    public void connectionLost(Throwable cause) {
                        log.warn("[connectionLost][数采 MQTT 连接断开：{}]", cause == null ? "未知原因" : cause.getMessage());
                    }

                    @Override
                    public void messageArrived(String topic, MqttMessage message) {
                        String payload = new String(message.getPayload(), StandardCharsets.UTF_8);
                        try {
                            dispatcher.dispatch(topic, payload);
                        } catch (Exception e) {
                            log.error("[messageArrived][主题({}) 入队失败]", topic, e);
                        }
                    }

                    @Override
                    public void deliveryComplete(IMqttDeliveryToken token) {
                        // 当前只订阅，不发布
                    }
                });
            }
            if (!client.isConnected()) {
                MqttConnectOptions options = new MqttConnectOptions();
                options.setAutomaticReconnect(true);
                options.setCleanSession(true);
                options.setKeepAliveInterval(mqtt.getKeepAlive());
                options.setConnectionTimeout(mqtt.getConnectionTimeout());
                if (StrUtil.isNotBlank(mqtt.getUsername())) {
                    options.setUserName(mqtt.getUsername());
                }
                if (mqtt.getPassword() != null) {
                    options.setPassword(mqtt.getPassword().toCharArray());
                }
                client.connect(options);
                log.info("[ensureConnected][数采 MQTT 已连接：{}]", mqtt.getUrl());
            }
        } catch (Exception e) {
            log.warn("[ensureConnected][数采 MQTT 连接/订阅失败，{} 秒后重试：{}]",
                    RECONNECT_DELAY_SECONDS, e.getMessage());
        }
    }

    private void subscribe() {
        MesCollectProperties.Mqtt mqtt = properties.getMqtt();
        for (String topic : resolveSubscribeTopics(mqtt)) {
            try {
                client.subscribe(topic, mqtt.getQos());
                log.info("[subscribe][数采 MQTT 订阅成功：{}]", topic);
            } catch (Exception e) {
                log.error("[subscribe][数采 MQTT 订阅失败：{}]", topic, e);
            }
        }
    }

    /**
     * 解析需要订阅的主题。
     *
     * <p>显式配置 topics 优先；网关格式订阅整个前缀（gateway/#），
     * 因为网关除 attributes 外还会发 connect / disconnect，后续也可能新增主题；
     * 其余格式沿用 /YLCY/IMM/+/+ 两级通配。</p>
     */
    private List<String> resolveSubscribeTopics(MesCollectProperties.Mqtt mqtt) {
        if (StrUtil.isNotBlank(mqtt.getTopics())) {
            return Arrays.stream(mqtt.getTopics().split(","))
                    .map(String::trim)
                    .filter(StrUtil::isNotBlank)
                    .toList();
        }
        String prefix = normalizeTopicPrefix(mqtt.getTopicPrefix());
        return Collections.singletonList(mqtt.isGatewayFormat() ? prefix + "#" : prefix + "+/+");
    }

    private String normalizeTopicPrefix(String prefix) {
        if (StrUtil.isBlank(prefix)) {
            return "";
        }
        return prefix.endsWith("/") ? prefix : prefix + "/";
    }

    private String buildClientId() {
        String host;
        try {
            host = InetAddress.getLocalHost().getHostName();
        } catch (Exception e) {
            host = "host";
        }
        host = host.replaceAll("[^a-zA-Z0-9_-]", "_");
        String clientId = properties.getMqtt().getClientIdPrefix() + "-" + host + "-"
                + Long.toHexString(System.nanoTime());
        return StrUtil.maxLength(clientId, 64);
    }

    @Override
    public void stop() {
        running = false;
        if (scheduler != null) {
            scheduler.shutdownNow();
            scheduler = null;
        }
        if (client != null) {
            try {
                if (client.isConnected()) {
                    client.disconnect();
                }
            } catch (Exception e) {
                log.warn("[stop][数采 MQTT 断开失败：{}]", e.getMessage());
            }
            try {
                client.close(true);
            } catch (Exception e) {
                log.warn("[stop][数采 MQTT 关闭失败：{}]", e.getMessage());
            }
            client = null;
        }
        log.info("[stop][数采 MQTT 已停止]");
    }

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

}