package com.zy.integration.iot.client.impl;

import com.core.common.Cools;
import com.zy.integration.iot.client.IotMqttClient;
import com.zy.integration.iot.handler.IotInboundMessageHandler;
import com.zy.integration.iot.util.IotMqttSslUtils;
import com.zy.iot.config.IotProperties;
import com.zy.iot.entity.IotTopicConfig;
import lombok.extern.slf4j.Slf4j;
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.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.io.File;
import java.nio.charset.StandardCharsets;

/**
 * IoT MQTT 客户端。
 * 兼容 AWS IoT 的 mTLS 模式，也兼容局域网模拟 broker 的明文 TCP 模式。
 */
@Slf4j
@Service
public class PahoIotMqttClient implements IotMqttClient {

    @Autowired
    private IotProperties iotProperties;
    @Autowired
    private IotInboundMessageHandler inboundMessageHandler;

    private volatile MqttClient mqttClient;

    @PostConstruct
    public void init() {
        if (!iotProperties.isEnabled()) {
            return;
        }
        try {
            ensureConnected();
        } catch (Exception e) {
            log.error("initial connect to iot mqtt failed", e);
        }
    }

    @PreDestroy
    public void destroy() {
        if (mqttClient == null) {
            return;
        }
        try {
            if (mqttClient.isConnected()) {
                mqttClient.disconnect();
            }
            mqttClient.close();
        } catch (Exception e) {
            log.warn("close iot mqtt client failed", e);
        }
    }

    /**
     * 懒加载建连，首次发送或启动时自动连接并补订阅主题。
     */
    @Override
    public synchronized void ensureConnected() throws Exception {
        if (!iotProperties.isEnabled()) {
            return;
        }
        if (!hasRequiredConfig()) {
            log.warn("iot config incomplete, mqtt client will not connect");
            return;
        }
        if (mqttClient == null) {
            mqttClient = createClient();
        }
        if (mqttClient.isConnected()) {
            return;
        }
        mqttClient.connect(buildOptions());
        subscribeTopics();
    }

    /**
     * 统一按 QoS1 发布，兼顾可靠性和实现复杂度。
     */
    @Override
    public void publish(String topic, String payload) throws Exception {
        if (!iotProperties.isEnabled() || Cools.isEmpty(topic) || payload == null) {
            return;
        }
        ensureConnected();
        if (mqttClient == null || !mqttClient.isConnected()) {
            throw new MqttException(MqttException.REASON_CODE_CLIENT_NOT_CONNECTED);
        }
        MqttMessage mqttMessage = new MqttMessage(payload.getBytes(StandardCharsets.UTF_8));
        mqttMessage.setQos(1);
        mqttMessage.setRetained(false);
        mqttClient.publish(topic, mqttMessage);
    }

    @Override
    public boolean isConnected() {
        return mqttClient != null && mqttClient.isConnected();
    }

    private boolean hasRequiredConfig() {
        if (Cools.isEmpty(iotProperties.getServerUri()) || Cools.isEmpty(iotProperties.getResolvedClientId())) {
            return false;
        }
        if (!iotProperties.isTlsEnabled()) {
            return true;
        }
        return !Cools.isEmpty(iotProperties.getCaCertPath())
                && !Cools.isEmpty(iotProperties.getClientCertPath())
                && !Cools.isEmpty(iotProperties.getPrivateKeyPath());
    }

    private MqttClient createClient() throws Exception {
        MqttClient client = new MqttClient(
                iotProperties.getServerUri(),
                iotProperties.getResolvedClientId(),
                new MqttDefaultFilePersistence(resolvePersistenceDir())
        );
        client.setCallback(new MqttCallbackExtended() {
            @Override
            public void connectionLost(Throwable cause) {
                log.warn("iot mqtt connection lost", cause);
            }

            @Override
            public void messageArrived(String topic, MqttMessage message) {
                inboundMessageHandler.handleMessage(topic, new String(message.getPayload(), StandardCharsets.UTF_8));
            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken token) {
            }

            @Override
            public void connectComplete(boolean reconnect, String serverURI) {
                try {
                    subscribeTopics();
                    log.info("iot mqtt connected, reconnect={}, serverURI={}", reconnect, serverURI);
                } catch (Exception e) {
                    log.error("subscribe iot topics failed", e);
                }
            }
        });
        return client;
    }

    private MqttConnectOptions buildOptions() throws Exception {
        MqttConnectOptions options = new MqttConnectOptions();
        options.setAutomaticReconnect(iotProperties.isAutomaticReconnect());
        options.setCleanSession(iotProperties.isCleanSession());
        options.setKeepAliveInterval(iotProperties.getKeepAliveSeconds());
        options.setConnectionTimeout(iotProperties.getConnectionTimeoutSeconds());
        if (iotProperties.isTlsEnabled()) {
            options.setSocketFactory(IotMqttSslUtils.buildSocketFactory(
                    iotProperties.getCaCertPath(),
                    iotProperties.getClientCertPath(),
                    iotProperties.getPrivateKeyPath()
            ));
        }
        return options;
    }

    /**
     * 只订阅 XBPS 下发的 stow/pick 指令和外部 feedback 回执。
     */
    private synchronized void subscribeTopics() throws Exception {
        if (mqttClient == null || !mqttClient.isConnected()) {
            return;
        }
        IotTopicConfig topics = iotProperties.getTopics();
        subscribe(topics.getEgressStow());
        subscribe(topics.getEgressPick());
        subscribe(topics.getEgressFeedback());
    }

    private void subscribe(String topic) throws Exception {
        if (Cools.isEmpty(topic)) {
            return;
        }
        mqttClient.subscribe(topic, 1);
    }

    private String resolvePersistenceDir() {
        String configured = iotProperties.getPersistenceDir();
        String dir = Cools.isEmpty(configured)
                ? System.getProperty("java.io.tmpdir") + File.separator + "asrs-iot-mqtt"
                : configured;
        File file = new File(dir);
        if (!file.exists()) {
            file.mkdirs();
        }
        return file.getAbsolutePath();
    }
}