package com.agri.framework.interceptor;

import com.agri.common.utils.wechat.WxUtil;
import com.agri.framework.config.MqttConfig;
import com.agri.framework.manager.MqttAutoOffManager;
import com.agri.framework.manager.MqttSubscriptionManager;
import com.agri.system.domain.SysAgriInfo;
import com.agri.system.domain.SysMessage;
import com.agri.system.domain.SysUserAgri;
import com.agri.system.service.AgriService;
import com.agri.system.service.ISysAgriInfoService;
import com.agri.system.util.UrlEncodeUtil;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import org.apache.commons.collections4.CollectionUtils;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.math.BigDecimal;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.TimeUnit;

/**
 * 设备状态消息处理器
 * 核心功能：
 * 1. 处理设备状态上报、设备回执消息
 * 2. 触发自动关闭任务、取消自动关闭任务
 * 3. 转发设备状态到订阅的前端
 * 4. 维护设备最新状态缓存
 * 适配JDK 8，无心跳包相关逻辑
 */
@Component
public class DeviceStatusHandler {
    /**
     * 优化：统一使用SLF4J日志（JDK 8兼容）
     */
    private static final Logger log = LoggerFactory.getLogger(DeviceStatusHandler.class);

    /**
     * Redis模板，用于存储订阅关系、设备在线状态、分布式锁
     */
    @Resource
    private StringRedisTemplate stringRedisTemplate;

    /**
     * MQTT消息发送工具类（由MqttConfig配置类注入）
     */
    @Resource
    private MqttConfig.MqttMessageSender mqttMessageSender;

    /**
     * MQTT订阅关系管理器，处理批量Redis操作
     */
    @Resource
    private MqttSubscriptionManager mqttSubscriptionManager;

    @Autowired
    private DeviceAckHandler deviceAckHandler;

    @Autowired
    private ISysAgriInfoService agriInfoService;
    // 新增：最新状态缓存TTL（设备每10秒上报一次，缓存一小段时间即可）
    @Value("${spring.mqtt.latest-ttl-seconds:120}")
    private int latestTtlSeconds;

    @Autowired
    private AgriService agriService;

    private static final String AUTO_MODE = "auto_mode";
    private static final String AUTO_OFF = "autooff";

    @Resource
    private FrontendControlHandler frontendControlHandler;
    /**
     * 自动关任务管理器，调度/取消自动关任务
     */
    @Resource
    private MqttAutoOffManager mqttAutoOffManager;

    private static final Set<String> VALID_FUNC_CODES = new HashSet<>();
    static {
        VALID_FUNC_CODES.add("jm1g");
        VALID_FUNC_CODES.add("jm2g");
        VALID_FUNC_CODES.add("jbg");
        VALID_FUNC_CODES.add("jm3g");
        VALID_FUNC_CODES.add("jm2k");
        VALID_FUNC_CODES.add("jm3k");
        VALID_FUNC_CODES.add("jbk");
        VALID_FUNC_CODES.add("jm1k");
    }

    /**
     * 处理设备状态：转发给订阅的前端、处理回执、触发自动关
     */
    public void handle(String topic, String payload) {

        // log.info("【设备处理】JSON解析：{}",payloadObj);
        // 解析设备ID：主题格式为dtu/{deviceId}/up，分割后第2个元素是设备ID
        String deviceId = extractDeviceId(topic);
        if (deviceId == null) return;
        String[] segments = topic.split("/");
        String action = segments[2];
        if ("down".equals(action) || !isJsonObjectLike(payload)) return;

        // 第一步：解析JSON，非有效JSON直接return
        JSONObject payloadObj;
        try {
            payloadObj = JSON.parseObject(payload);
        } catch (Exception e) {
            log.error("【设备处理】JSON解析失败，payload={}", payload, e);
            return;
        }
        if (payloadObj == null || payloadObj.isEmpty()) {
            log.warn("【设备处理】JSON解析后为空，payload={}", payload);
            return;
        }

        boolean isAck = payloadObj.containsKey("suc") && payloadObj.containsKey("prop");
        JSONObject sendObj = payloadObj;  // 默认直接用原对象
        // 如果是回执，先拿 funcType

        if (isAck) {
            JSONObject propObj = payloadObj.getJSONObject("prop");
            if (propObj != null && !propObj.isEmpty()) {
                String funcType = propObj.entrySet().iterator().next().getKey();
                String lockKey = "lock:" + deviceId + ":" + funcType;

                // 读取锁的 value（比如 autooff / user:1001）
                String lockHolder = stringRedisTemplate.opsForValue().get(lockKey);
                if (lockHolder != null) {
                    sendObj = new JSONObject(payloadObj);  // 只在需要时复制
                    sendObj.put("clientId", lockHolder);
                    // 如果相等则为自动模式直接退出方法 不转发ack消息 自动关也不应该转发
                    if (AUTO_MODE.equals(lockHolder) || AUTO_OFF.equals(lockHolder)) {
                        return;
                    }
                }
            }
        }

        //  转发消息
        forwardPayload(deviceId, payload,payloadObj,action, sendObj, isAck);
        // 获取第二个动态段，如"up"或"ack"
        if ("ack".equals(action)) {
            deviceAckHandler.isStartAutoOffTask(payloadObj,deviceId,payload);
            return;
        }
        // payload全部为数字
        if ("up".equals(action) && isAllKeysDigit(payloadObj)) {
            SysAgriInfo agriInfo = agriInfoService.lambdaQuery()
                    .eq(SysAgriInfo::getImei, deviceId)
                    .one();
            if (agriInfo == null) {
                return;
            }
            // 温度上下限
            BigDecimal tempUp = agriInfo.getTempUp();
            BigDecimal tempLow = agriInfo.getTempLow();
            // 湿度上下限
            BigDecimal humiUp = agriInfo.getHumiUp();
            BigDecimal humiLow = agriInfo.getHumiLow();
            List<SysAgriInfo> msgList = new ArrayList<>();
            // 仅当up且所有key为数字时，才更新最新状态缓存
            for (String key : payloadObj.keySet()) {
                BigDecimal value = payloadObj.getBigDecimal(key);
                String valueIndex = key.substring(2);
                SysAgriInfo sysAgriInfo = new SysAgriInfo();
                BeanUtils.copyProperties(agriInfo, sysAgriInfo);
                // 温度
                if (key.startsWith("20")) {
                    // 温度大于温度上限
                    if (value.compareTo(tempUp) > 0 ) {
                        agriInfo.setTitle("温度异常");
                        agriInfo.setMsg("温度"+valueIndex+ "异常！高于上限"+tempUp+"℃！");
                    }
                    // 温度小于温度下限
                    if (value.compareTo(tempLow) < 0) {
                        agriInfo.setTitle("温度异常");
                        agriInfo.setMsg("温度"+valueIndex+ "异常！低于下限"+tempLow+"℃！");
                    }
                    msgList.add(agriInfo);
                }
                // 湿度
                if (key.startsWith("10")) {
                    // 湿度大于湿度上限
                    if (value.compareTo(humiUp) > 0 ) {
                        agriInfo.setTitle("湿度异常");
                        agriInfo.setMsg("湿度"+valueIndex+ "异常！高于上限"+humiUp+"RH%！");
                    }
                    // 湿度小于湿度下限
                    if (value.compareTo(humiLow) < 0) {
                        agriInfo.setTitle("湿度异常");
                        agriInfo.setMsg("湿度"+valueIndex+ "异常！低于下限"+humiLow+"RH%！");
                    }
                    msgList.add(agriInfo);
                }
            }
            if (!msgList.isEmpty()) {
                List<SysMessage> messages = agriService.saveMessage(msgList);
                try {
                    frontendControlHandler.sendAlarmMessage(messages);
                } catch (Exception e) {
                    throw new RuntimeException(e);
                }
            }
        }
    }

    private void forwardPayload(String deviceId,String payload,
                                JSONObject payloadObj,String action, JSONObject sendObj, boolean isAck)  {
        try {

            // 非回执消息：正常转发给订阅前端
            // 查询Redis中订阅该设备的前端列表：sub:{deviceId}
            Set<String> subscribedClients = stringRedisTemplate.opsForSet().members("sub:" + deviceId);

            if (subscribedClients != null && !subscribedClients.isEmpty()) {
                // 推送给每个订阅的前端
                // 方案B：不再依赖online:；改为校验subc:{clientId}是否仍包含deviceId（取消订阅失败/异常退出兜底）
                List<String> clients = new ArrayList<>(subscribedClients);
                // 判断subc是否还存在  一次性查全部  获取失效的clientId
                List<Boolean> stillSubs = mqttSubscriptionManager.pipeIsMemberSubc(clients, deviceId);

                // 关系不存在：清理sub:{deviceId}残留，避免一直给前端发
                List<String> stale = null;

                for (int i = 0; i < clients.size(); i++) {
                    String clientId = clients.get(i);
                    boolean stillSub = i < stillSubs.size() && Boolean.TRUE.equals(stillSubs.get(i));
                    if (!stillSub) {
                        if (stale == null) {
                            stale = new ArrayList<>();
                        }
                        // false不存在添加队列
                        stale.add(clientId);
                        continue;
                    }
                    // 前端专属主题：frontend/{clientId}/dtu/{deviceId}/listener
                    String frontendTopic = "frontend/" + clientId + "/dtu/" + deviceId + "/listener";
                    if ("ack".equals(action)) {
                        frontendTopic = "frontend/" + clientId + "/dtu/" + deviceId + "/ack";
                    }
                    // 发布消息
                    mqttMessageSender.publish(frontendTopic, sendObj.toJSONString());
                    // log.info("【设备状态转发】设备{} → 前端{}，主题：{}", deviceId, clientId, frontendTopic);
                }
                // 删掉设备对应的客户端
                if (stale != null && !stale.isEmpty()) {
                    mqttSubscriptionManager.pipeSRemSub(deviceId, stale);
                }
            } else {
                // 优化：替换System.out为log.info
                // log.info("【设备状态转发】设备{}无订阅前端，跳过转发", deviceId);
            }

            // 第三步：仅处理非回执的设备状态包，且仅当是8个功能码结构就写入Redis
            // 有没有人订阅都得写，只要发送设备开的指令成功了就得写
            if (!isAck) {
                // 1) 先校验状态包是否包含8个固定功能码（核心：只有这种结构才写入）
                boolean isValidStatus = true;
                for (String validCode : VALID_FUNC_CODES) {
                    if (!payloadObj.containsKey(validCode)) {
                        isValidStatus = false;
                        // log.debug("【设备状态包】结构不合法（非8个功能码），跳过Redis写入，deviceId={}；payload={}", deviceId, payload);
                        break;
                    }
                }
                if (mqttAutoOffManager.hasAutoOffTask(deviceId) && isValidStatus) {
                    // ✅ 8个功能码状态包：无条件写device:latest:{deviceId}，避免自动关读不到最新状态
                    stringRedisTemplate.opsForValue().set(
                            "device:latest:" + deviceId,
                            payload,  // 完整的8功能码JSON
                            latestTtlSeconds,
                            TimeUnit.SECONDS
                    );
                    // log.debug("【设备状态包】写入Redis成功，deviceId={}", deviceId);
                }
            }
        } catch (MqttException e) {
            WxUtil.pushText(
                    "【消息转发失败】\n  deviceId: "+deviceId+"\n  payload: "+payload+"\n  cause: "+e);
            log.error("【消息转发失败】deviceId={}, msg={}", deviceId, e.getMessage(), e);
        }
    }

    private String extractDeviceId(String topic) {
        int first = topic.indexOf('/');
        if (first < 0) return null;
        int second = topic.indexOf('/', first + 1);
        if (second < 0) return null;
        return topic.substring(first + 1, second);
    }

    private boolean isJsonObjectLike(String s) {
        if (s == null) return false;
        int n = s.length();
        for (int i = 0; i < n; i++) {
            char c = s.charAt(i);
            if (!Character.isWhitespace(c)) return c == '{';
        }
        return false;
    }


    public static boolean isAllKeysDigit(JSONObject obj) {
        return obj != null && !obj.isEmpty()
                && obj.keySet().stream().allMatch(k -> k.matches("\\d+"));
    }

}
