package xiaozhi.modules.parent.mqtt;

import java.util.Map;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import org.apache.commons.lang3.StringUtils;
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.persist.MemoryPersistence;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.stereotype.Component;

import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;

import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import xiaozhi.common.exception.RenException;
import xiaozhi.common.redis.RedisKeys;
import xiaozhi.common.redis.RedisUtils;
import xiaozhi.modules.sys.service.SysParamsService;

/**
 * 根据设备 MQTT 状态主题判断是否在线。未在线则不允许保存/下发家长任务。
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class ParentDevicePresence {
    public static final String OFFLINE_MESSAGE = "设备不在线，无法同步到硬件。请打开设备并连上网络后再试，本次不会保存。";
    private static final long ONLINE_TTL_SEC = 180L;
    private static final long PROBE_WAIT_MS = 2000L;
    private static final TypeReference<Map<String, Object>> MAP = new TypeReference<>() {
    };

    private final SysParamsService sysParamsService;
    private final RedisUtils redisUtils;
    private final ObjectMapper objectMapper;
    private final ApplicationEventPublisher eventPublisher;

    private volatile boolean stopped;
    private volatile MqttClient listener;

    @PostConstruct
    public void start() {
        Thread thread = new Thread(this::listenLoop, "parent-mqtt-presence");
        thread.setDaemon(true);
        thread.start();
    }

    @PreDestroy
    public void stop() {
        stopped = true;
        closeQuietly(listener);
    }

    public void requireOnline(String macAddress) {
        if (!isOnline(macAddress)) {
            throw new RenException(OFFLINE_MESSAGE);
        }
    }

    public boolean isOnline(String macAddress) {
        if (isOnlineCached(macAddress)) {
            return true;
        }
        String mac = ParentMqttPublisher.normalizeMac(macAddress);
        if (StringUtils.isBlank(mac)) {
            return false;
        }
        return probe(mac);
    }

    public boolean isOnlineCached(String macAddress) {
        String mac = ParentMqttPublisher.normalizeMac(macAddress);
        if (StringUtils.isBlank(mac)) {
            return false;
        }
        return redisUtils.get(RedisKeys.getParentDeviceOnlineKey(mac)) != null;
    }

    private void listenLoop() {
        while (!stopped) {
            try {
                listenOnce();
            } catch (Exception ex) {
                log.warn("parent mqtt presence: {}", ex.getMessage());
            }
            sleepQuietly(15_000);
        }
    }

    private void listenOnce() throws MqttException {
        Broker broker = broker();
        if (broker == null) {
            return;
        }
        MqttClient client = new MqttClient(broker.uri(), "parent-api-presence-" + UUID.randomUUID(),
                new MemoryPersistence());
        listener = client;
        client.setCallback(new org.eclipse.paho.client.mqttv3.MqttCallback() {
            @Override
            public void connectionLost(Throwable cause) {
                log.warn("parent mqtt presence disconnected: {}",
                        cause == null ? "" : cause.getMessage());
            }

            @Override
            public void messageArrived(String topic, org.eclipse.paho.client.mqttv3.MqttMessage message) {
                applyState(topic, new String(message.getPayload(), java.nio.charset.StandardCharsets.UTF_8));
            }

            @Override
            public void deliveryComplete(org.eclipse.paho.client.mqttv3.IMqttDeliveryToken token) {
            }
        });
        client.connect(broker.options());
        client.subscribe("snlp/v1/+/rpt/state", 0);
        log.info("parent mqtt presence listening on snlp/v1/+/rpt/state");
        while (!stopped && client.isConnected()) {
            sleepQuietly(1000);
        }
        closeQuietly(client);
        if (listener == client) {
            listener = null;
        }
    }

    private boolean probe(String mac) {
        Broker broker = broker();
        if (broker == null) {
            return false;
        }
        String topic = "snlp/v1/" + mac + "/rpt/state";
        CountDownLatch latch = new CountDownLatch(1);
        MqttClient client = null;
        try {
            client = new MqttClient(broker.uri(), "parent-api-probe-" + UUID.randomUUID(), new MemoryPersistence());
            client.connect(broker.options());
            client.subscribe(topic, 0, (t, message) -> {
                applyState(t, new String(message.getPayload(), java.nio.charset.StandardCharsets.UTF_8));
                latch.countDown();
            });
            latch.await(PROBE_WAIT_MS, TimeUnit.MILLISECONDS);
        } catch (Exception ex) {
            log.warn("parent mqtt presence probe failed: {}", ex.getMessage());
        } finally {
            closeQuietly(client);
        }
        return redisUtils.get(RedisKeys.getParentDeviceOnlineKey(mac)) != null;
    }

    @SuppressWarnings("unchecked")
    private void applyState(String topic, String json) {
        try {
            Map<String, Object> root = objectMapper.readValue(json, MAP);
            Object payloadRaw = root.get("payload");
            Map<String, Object> payload = payloadRaw instanceof Map<?, ?> map ? (Map<String, Object>) map : Map.of();
            String mac = ParentMqttPublisher.normalizeMac(text(payload.get("mac")));
            if (StringUtils.isBlank(mac)) {
                mac = macFromTopic(topic);
            }
            if (StringUtils.isBlank(mac)) {
                return;
            }
            boolean online = payloadMeansOnline(payload);
            String key = RedisKeys.getParentDeviceOnlineKey(mac);
            if (online) {
                redisUtils.set(key, "1", ONLINE_TTL_SEC);
            } else {
                boolean wasOnline = redisUtils.get(key) != null;
                redisUtils.delete(key);
                if (wasOnline) {
                    eventPublisher.publishEvent(new ParentDeviceOfflineEvent(mac));
                }
            }
        } catch (Exception ex) {
            log.debug("ignore parent mqtt state: {}", ex.getMessage());
        }
    }

    private static String macFromTopic(String topic) {
        String[] parts = StringUtils.trimToEmpty(topic).split("/");
        if (parts.length >= 5 && "snlp".equals(parts[0]) && "v1".equals(parts[1])
                && "rpt".equals(parts[3]) && "state".equals(parts[4]) && parts[2].contains(":")) {
            return ParentMqttPublisher.normalizeMac(parts[2]);
        }
        return "";
    }

    private Broker broker() {
        String host = StringUtils.trimToEmpty(sysParamsService.getValue("server.parent_mqtt_host", true));
        String portText = StringUtils.trimToEmpty(sysParamsService.getValue("server.parent_mqtt_port", true));
        String username = StringUtils.trimToEmpty(sysParamsService.getValue("server.parent_mqtt_username", true));
        String password = StringUtils.trimToEmpty(sysParamsService.getValue("server.parent_mqtt_password", true));
        if (StringUtils.isBlank(host) || "null".equalsIgnoreCase(host)) {
            return null;
        }
        int port = 2100;
        try {
            if (StringUtils.isNotBlank(portText) && !"null".equalsIgnoreCase(portText)) {
                port = Integer.parseInt(portText);
            }
        } catch (NumberFormatException ex) {
            return null;
        }
        MqttConnectOptions options = new MqttConnectOptions();
        options.setAutomaticReconnect(false);
        options.setCleanSession(true);
        options.setConnectionTimeout(8);
        options.setKeepAliveInterval(30);
        if (StringUtils.isNotBlank(username) && !"null".equalsIgnoreCase(username)) {
            options.setUserName(username);
        }
        if (StringUtils.isNotBlank(password) && !"null".equalsIgnoreCase(password)) {
            options.setPassword(password.toCharArray());
        }
        return new Broker("tcp://" + host + ":" + port, options);
    }

    static boolean payloadMeansOnline(Map<String, Object> payload) {
        if (payload == null || payload.isEmpty()) {
            return false;
        }
        Boolean online = asBoolean(payload.get("online"));
        if (Boolean.FALSE.equals(online)) {
            return false;
        }
        if (Boolean.TRUE.equals(online)) {
            return true;
        }
        String net = text(payload.get("net"));
        return StringUtils.isNotBlank(net) && !"offline".equalsIgnoreCase(net);
    }

    private static Boolean asBoolean(Object value) {
        if (value instanceof Boolean bool) {
            return bool;
        }
        if (value == null) {
            return null;
        }
        String text = String.valueOf(value).trim();
        if ("true".equalsIgnoreCase(text) || "1".equals(text)) {
            return Boolean.TRUE;
        }
        if ("false".equalsIgnoreCase(text) || "0".equals(text)) {
            return Boolean.FALSE;
        }
        return null;
    }

    private static String text(Object value) {
        return value == null ? "" : String.valueOf(value).trim();
    }

    private static void closeQuietly(MqttClient client) {
        if (client == null) {
            return;
        }
        try {
            if (client.isConnected()) {
                client.disconnect();
            }
        } catch (Exception ignored) {
        }
        try {
            client.close();
        } catch (Exception ignored) {
        }
    }

    private static void sleepQuietly(long ms) {
        try {
            Thread.sleep(ms);
        } catch (InterruptedException ex) {
            Thread.currentThread().interrupt();
        }
    }

    private record Broker(String uri, MqttConnectOptions options) {
    }
}