package com.fastbee.iot.data.emq;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.fastbee.common.enums.DeviceStatus;
import com.fastbee.common.enums.ServerType;
import com.fastbee.common.enums.TopicType;
import com.fastbee.common.extend.core.domin.mq.DeviceReportBo;
import com.fastbee.common.extend.core.domin.mq.DeviceStatusBo;
import com.fastbee.common.utils.DateUtils;
import com.fastbee.common.utils.StringUtils;
import com.fastbee.common.utils.gateway.mq.TopicsUtils;
import com.fastbee.iot.domain.Device;
import com.fastbee.iot.service.IDeviceService;
import com.fastbee.mq.producer.MessageProducer;
import com.fastbee.ruleEngine.context.MsgContext;
import com.fastbee.iot.ruleEngine.RuleProcess;
import com.fastbee.mqttclient.IEmqxMessageProducer;
import io.netty.buffer.ByteBufUtil;
import io.netty.buffer.Unpooled;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.IMqttMessageListener;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.OutputStream;
import java.net.HttpURLConnection;
import java.net.URL;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.Objects;

@Slf4j
@Component
public class subscribeCallback implements IMqttMessageListener {
    @Resource
    private RuleProcess ruleProcess;
    @Resource
    private TopicsUtils topicsUtils;
    @Resource
    private IEmqxMessageProducer emqxMessageSerice;
    @Resource
    private IDeviceService deviceService;

    @Override
    public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
        String message = new String(mqttMessage.getPayload());
        log.info(" 接收消息主题 : {}", topic);
        log.info(" 接收消息Qos : {}", mqttMessage.getQos());
        log.info(" 接收消息内容 : {}", message);

        if (topic.startsWith("/special/device"))  {
            log.info("[  特殊设备] 监听到特殊主题消息，内容: {}", message);

            try {
                JSONObject msgJson = JSON.parseObject(message);
                String serialNumber = msgJson.getString("serialNumber");
                String productId = msgJson.getString("productId");
                Integer status = msgJson.getInteger("status");

                if (StringUtils.isNotEmpty(serialNumber)  && status != null) {
                    // 构造模拟Webhook的JSON
                    JSONObject simulatedWebhook = new JSONObject();
                    simulatedWebhook.put("clientid",  "S&" + serialNumber + "&" + productId + "&1");
                    simulatedWebhook.put("event",  status == 1 ? "client.connected"  : "client.disconnected");
                    simulatedWebhook.put("peername",  msgJson.getString("ip") + ":127.0.0.1");

                    // 通过HTTP POST调用本地接口
                    postToWebhookV5(simulatedWebhook);
                }
            } catch (Exception e) {
                log.error("[  特殊设备] 消息处理异常: {}", e.getMessage());
            }
            return;
        }

        // 解决16进制数据转换错误
        byte[] bytes = message.getBytes(StandardCharsets.UTF_8);
        if (bytes.length  != mqttMessage.getPayload().length)  {
            bytes = mqttMessage.getPayload();
            message = ByteBufUtil.hexDump(Unpooled.wrappedBuffer(mqttMessage.getPayload()));
        }

        //这里默认设备编号长度超过9位
        String[] split = topic.split("/");
        String sn = Arrays.stream(split).filter(imei  -> imei.length()  > 9).findFirst().get();
        // 规则引擎脚本处理,完成后返回结果
        MsgContext context = ruleProcess.processRuleScript(sn,  1, topic, message);
        if (!Objects.isNull(context)  && StringUtils.isNotEmpty(context.getPayload())
                && StringUtils.isNotEmpty(context.getTopic()))  {
            topic = context.getTopic();
            message = context.getPayload();
            bytes = message.getBytes(StandardCharsets.UTF_8);
        }
        String serialNumber = topicsUtils.parseSerialNumber(topic);
        Long productId;
        if (topic.contains(TopicType.FETCH_UPGRADE_REPLY.getTopicSuffix())  || topic.contains(TopicType.HTTP_UPGRADE_REPLY.getTopicSuffix()))  {
            productId = 0L;
        } else {
            productId = topicsUtils.parseProductId(topic);
        }

        String name = topicsUtils.parseTopicName(topic);
        DeviceReportBo reportBo = DeviceReportBo.builder()
                .serialNumber(serialNumber)
                .productId(productId)
                .data(bytes)
                .platformDate(DateUtils.getNowDate())
                .topicName(topic)
                .serverType(ServerType.MQTT)
                .build();
        /*将mqtt的消息发送至MQ队列处理消息 ,减轻mqtt客户端消息压力*/
        emqxMessageSerice.sendEmqxMessage(name,  reportBo);
    }

    /**
     * 向指定HTTP接口发送POST请求
     * @param data 要发送的JSON数据
     */
    private void postToWebhookV5(JSONObject data) {
//        String apiUrl = "http://121.36.111.27:8080/bridge/get";
//        String apiUrl = "http://localhost:8080/iot/tool/mqtt/webhookv5";
        String apiUrl = "http://121.36.111.27:8080/iot/tool/mqtt/webhookv5";
        HttpURLConnection connection = null;

        try {
            URL url = new URL(apiUrl);
            connection = (HttpURLConnection) url.openConnection();
            connection.setRequestMethod("POST");
            connection.setRequestProperty("Content-Type",  "application/json; charset=UTF-8");
            connection.setDoOutput(true);

            // 发送请求体
            try (OutputStream os = connection.getOutputStream())  {
                byte[] input = data.toJSONString().getBytes(StandardCharsets.UTF_8);
                os.write(input,  0, input.length);
            }

            // 获取响应状态码
            int responseCode = connection.getResponseCode();
            if (responseCode == HttpURLConnection.HTTP_OK) {
                log.info("HTTP 请求成功，状态码: {}", responseCode);
            } else {
                log.error("HTTP 请求失败，状态码: {}", responseCode);
            }
        } catch (Exception e) {
            log.error("HTTP 请求异常: {}", e.getMessage());
        } finally {
            if (connection != null) {
                connection.disconnect();
            }
        }
    }
}