package com.xzzn.web.controller.ems;

import com.xzzn.common.constant.RedisKeyConstants;
import com.xzzn.common.core.redis.RedisCache;
import com.xzzn.common.enums.TopicHandleType;
import com.xzzn.common.utils.StringUtils;
import com.xzzn.ems.domain.EmsMqttTopicConfig;
import com.xzzn.ems.mapper.EmsMqttTopicConfigMapper;
import com.xzzn.ems.service.IDeviceDataProcessService;
import com.xzzn.ems.service.IMqttSyncLogService;
import com.xzzn.framework.manager.MqttLifecycleManager;
import com.xzzn.framework.web.service.MqttPublisher;
import com.xzzn.framework.web.service.MqttSubscriber;

import java.util.List;

import javax.annotation.PostConstruct;

import org.apache.commons.collections4.CollectionUtils;
import org.eclipse.paho.client.mqttv3.IMqttMessageListener;
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.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Service
public class MqttMessageController implements MqttPublisher, MqttSubscriber {

    private static final Logger log = LoggerFactory.getLogger(MqttMessageController.class);


    private final MqttLifecycleManager mqttLifecycleManager;


    @Autowired
    private IDeviceDataProcessService deviceDataProcessService;

    @Autowired
    private EmsMqttTopicConfigMapper emsMqttTopicConfigMapper;
    @Autowired
    private IMqttSyncLogService iMqttSyncLogService;

    @Autowired
    private RedisCache redisCache;
    @Autowired
    private ThreadPoolTaskExecutor threadPoolTaskExecutor;

    @Autowired
    public MqttMessageController(MqttLifecycleManager mqttLifecycleManager) {
        this.mqttLifecycleManager = mqttLifecycleManager;
    }

    @PostConstruct
    public void init() {
        List<EmsMqttTopicConfig> topicConfigList = emsMqttTopicConfigMapper.selectEmsMqttTopicConfigList(null);

        if (CollectionUtils.isEmpty (topicConfigList)) {
            log.info ("未查询到任何 MQTT 主题配置，跳过订阅");
            return;
        }

        for (EmsMqttTopicConfig topicConfig : topicConfigList) {
            String topic = topicConfig.getMqttTopic();
            if (StringUtils.isEmpty(topic)) {
                log.info ("主题配置 ID:" +topicConfig.getId() +"的 mqttTopic 为空，跳过订阅");
                continue;
            }

            int qos =  topicConfig.getQos() == null ? 1 : topicConfig.getQos();
            if (qos < 0 || qos > 2) {
                log.info ("主题:" + topic +"的 QoS值"+ qos + "不合法，自动调整为1");
                qos = 1;
            }

            IMqttMessageListener listener = getMqttListenerByTopic(topic, topicConfig.getHandleType());
            subscribe(topic, qos, listener);
        }
        // 订阅奉贤系统状态主题
        /*subscribe("021_FXX_01_UP", 1, this::handleDeviceData);
        subscribe("021_FXX_01_RECALL", 1, this::handleDeviceData);
        subscribe("021_FXX_01_DOWN", 1, this::handleDeviceData);
        subscribe("021_FXX_01", 1, this::handleSystemStatus);
        subscribe("021_FXX_01_ALARM_UP", 1, this::handleAlarmData);

        // 订阅电动所系统状态主题
        subscribe("021_DDS_01_UP", 1, this::handleDeviceData);
        subscribe("021_DDS_01_RECALL", 1, this::handleDeviceData);
        subscribe("021_DDS_01_DOWN", 1, this::handleDeviceData);
        subscribe("021_DDS_01", 1, this::handleSystemStatus);*/

    }

    private IMqttMessageListener getMqttListenerByTopic(String topic, String handleType) {
        TopicHandleType topicHandleType = TopicHandleType.getEnumByCode(handleType);
        switch (topicHandleType) {
            case DEVICE:
                return this::handleDeviceData;
            case DEVICE_ALARM:
                return this::handleAlarmData;
            case STRATEGY:
                if (topic.equals("FAULT_PROTECTION_PLAN_UP")) {
                    return this::handleFaultProtPlanData;
                } else if (topic.equals("FAULT_ALARM_RECORD_UP")) {
                    return this::handleFaultAlarmData;
                } else if (topic.equals("FAULT_PLAN_ISSUE_UP")) {
                    return this::handleFaultPlanIssueData;
                } else if (topic.equals("DEVICE_CHANGE_LOG_UP")) {
                    return this::handleDeviceChangeLogData;
                } else {
                    return this::handleStrategyData;
                }
            default:
                log.warn("Unknown handle type: " + handleType + ", using default handler");
                return this::handleSystemStatus;
        }
    }

    // 处理系统状态消息
    private void handleSystemStatus(String topic, MqttMessage message) {
        String payload = new String(message.getPayload());
        System.out.println("[SYSTEM] Status update: " + payload);
    }

    // 处理设备数据
    private void handleDeviceData(String topic, MqttMessage message) {
        String payload = new String(message.getPayload());
        threadPoolTaskExecutor.execute(() -> {
            log.debug("[DEVICE] data: {}", payload);
            try {
                deviceDataProcessService.handleDeviceData(payload, getSiteIdByTopic(topic));
            } catch (Exception e) {
                log.error("Failed to process device data message: {}", e.getMessage(), e);
            }
        });

    }

    // 处理告警数据
    private void handleAlarmData(String topic, MqttMessage message) {
        String payload = new String(message.getPayload());
        threadPoolTaskExecutor.execute(() -> {
            try {
                deviceDataProcessService.handleAlarmData(payload, getSiteIdByTopic(topic));
            } catch (Exception e) {
                log.error("Failed to process device alarm data message: {}", e.getMessage(), e);
            }
        });

    }

    private String getSiteIdByTopic(String topic) {
        String siteId = redisCache.getCacheObject(RedisKeyConstants.SITE_ID + topic);
        if (StringUtils.isEmpty(siteId)) {
            EmsMqttTopicConfig topicConfig = emsMqttTopicConfigMapper.selectOneByTopic(topic);
            siteId = topicConfig.getSiteId();
            redisCache.setCacheObject(RedisKeyConstants.SITE_ID + topic, siteId);
        }
        return siteId;
    }

    // 处理运行策略数据：云端-本地
    private void handleStrategyData(String topic, MqttMessage message) {
        String payload = new String(message.getPayload());
        threadPoolTaskExecutor.execute(() -> {
            try {
                iMqttSyncLogService.handleMqttStrategyData(payload);
            } catch (Exception e) {
                log.error("Failed to process strategy data message: {}", e.getMessage(), e);
            }
        });
    }

    // 处理设备保护告警策略数据：云端-本地
    private void handleFaultProtPlanData(String topic, MqttMessage message) {
        String payload = new String(message.getPayload());
        threadPoolTaskExecutor.execute(() -> {
            try {
                iMqttSyncLogService.handleMqttPlanData(payload);
            } catch (Exception e) {
                log.error("Failed to process fault plan data message: {}", e.getMessage(), e);
            }
        });
    }

    // 处理保护策略告警信息：本地-云端
    private void handleFaultAlarmData(String topic, MqttMessage message) {
        String payload = new String(message.getPayload());
        threadPoolTaskExecutor.execute(() -> {
            try {
                iMqttSyncLogService.handleFaultAlarmData(payload);
            } catch (Exception e) {
                log.error("Failed to process fault plan alarm data message: {}", e.getMessage(), e);
            }
        });
    }

    // 处理保护策略下发日志：本地-云端
    private void handleFaultPlanIssueData(String topic, MqttMessage message) {
        String payload = new String(message.getPayload());
        threadPoolTaskExecutor.execute(() -> {
            try {
                iMqttSyncLogService.handleFaultPlanIssueData(payload);
            } catch (Exception e) {
                log.error("Failed to process fault plan issue log message: {}", e.getMessage(), e);
            }
        });
    }

    // 处理设备状态变更日志：本地-云端
    private void handleDeviceChangeLogData(String topic, MqttMessage message) {
        String payload = new String(message.getPayload());
        threadPoolTaskExecutor.execute(() -> {
            try {
                iMqttSyncLogService.handleDeviceChangeLogData(payload);
            } catch (Exception e) {
                log.error("Failed to process device change log message: {}", e.getMessage(), e);
            }
        });
    }

    @Override
    public void publish(String topic, String message) throws MqttException {
        mqttLifecycleManager.publish(topic, message, 0);
    }

    @Override
    public void publish(String topic, String message, int qos) throws MqttException {
        mqttLifecycleManager.publish(topic, message, qos);
    }

    @Override
    public void subscribe(String topic, int qos, IMqttMessageListener listener) {
        mqttLifecycleManager.subscribe(topic, qos, listener);


    }

    // 发送设备控制命令
    public void sendDeviceCommand(String deviceId, String command) {
        try {
            String topic = "devices/" + deviceId + "/commands";
            publish(topic, command, 1);
        } catch (MqttException e) {
            System.err.println("Failed to send command to device " + deviceId);
        }
    }

}
