fix(O): 涂鸦消息订阅连接状态可观测, DP上报落device_state, 修复上下线误判

- MqConsumer.start() 移入独立守护线程, 主循环监控存活并自动重连(原阻塞主循环)
- connected 语义改为'收到过消息', 新增 connecting/messageCount/lastConnectAt/lastError
- 新增 syncDeviceStateFromDps: 涂鸦属性上报写入 device_state(此前该表仅中控屏一条写入路径)
- TuyaPulsarEventAdapter 修复: protocol==4 无条件判 ONLINE 导致离线推送被写成上线
- 新增 TuyaPulsarEventAdapterTest(5用例)

backend/iot-device/src/main/java/com/iot/device/collector/TuyaPulsarEventAdapter.java

@@ -92,18 +92,74 @@ public class TuyaPulsarEventAdapter implements IPlatformEventAdapter {
         return events;
     }
 
-    /** 归一化事件类型（BizCode 与数字协议号双模式，与旧实现语义一致） */
+    /**
+     * 归一化事件类型（BizCode 与数字协议号双模式）
+     *
+     * <p><b>修正</b>：原实现把 {@code protocol == 4}（设备上下线）一律判为 ONLINE，
+     * 也把"含 status/online 字段"一律判为 ONLINE，导致<b>离线推送被写成上线</b>，
+     * 设备在线状态失真。现在一律按字段取值判定。</p>
+     */
     private String resolveType(Integer protocol, String bizCode, JsonNode d) {
         if ("deviceOnline".equals(bizCode)) return "ONLINE";
         if ("deviceOffline".equals(bizCode)) return "OFFLINE";
         if ("devicePropertyMessage".equals(bizCode)
                 || "deviceReportMessage".equals(bizCode)) return "DP";
         if (protocol != null) {
-            if (protocol == 4) return "ONLINE";
+            if (protocol == 4) {
+                // 上下线：按 payload 里的状态值判定，不能一律当上线
+                return resolveOnlineByPayload(d) ? "ONLINE" : "OFFLINE";
+            }
             if (protocol == 8 || protocol == 9) return "DP";
         }
         // 数字 protocol 老结构里 4 之外，data 里带 status/online 的也算上下线
-        if (d.has("status") || d.has("online")) return "ONLINE";
+        if (d.has("status") || d.has("online")) {
+            return resolveOnlineByPayload(d) ? "ONLINE" : "OFFLINE";
+        }
         return "OTHER";
     }
+
+    /**
+     * 从推送负载里解析设备是否在线
+     *
+     * <p>兼容三种取值：布尔 true/false、数字 1/0、字符串 "online"/"offline"（或 "1"/"0"）。</p>
+     */
+    private boolean resolveOnlineByPayload(JsonNode d) {
+        if (d == null) {
+            return false;
+        }
+        JsonNode online = d.get("online");
+        if (online != null && !online.isNull()) {
+            if (online.isBoolean()) {
+                return online.asBoolean();
+            }
+            if (online.isNumber()) {
+                return online.asInt() != 0;
+            }
+            String text = online.asText();
+            if ("online".equalsIgnoreCase(text) || "1".equals(text)) {
+                return true;
+            }
+            if ("offline".equalsIgnoreCase(text) || "0".equals(text)) {
+                return false;
+            }
+        }
+        JsonNode status = d.get("status");
+        if (status != null && !status.isNull()) {
+            if (status.isBoolean()) {
+                return status.asBoolean();
+            }
+            if (status.isNumber()) {
+                return status.asInt() != 0;
+            }
+            String text = status.asText();
+            if ("online".equalsIgnoreCase(text) || "1".equals(text)) {
+                return true;
+            }
+            if ("offline".equalsIgnoreCase(text) || "0".equals(text)) {
+                return false;
+            }
+        }
+        // 取不到明确状态时，按"无状态证据"处理为离线，避免把离线误判成在线
+        return false;
+    }
 }

backend/iot-device/src/main/java/com/iot/device/service/TuyaMessageSubscribeService.java

@@ -7,9 +7,11 @@ import com.iot.device.collector.PlatformEventInput;
 import com.iot.device.collector.TuyaPulsarEventAdapter;
 import com.iot.device.dto.DeviceEvent;
 import com.iot.device.entity.Device;
+import com.iot.device.entity.DeviceState;
 import com.iot.device.entity.PlatformConfig;
 import com.iot.device.event.DeviceEventPublisher;
 import com.iot.device.mapper.DeviceMapper;
+import com.iot.device.mapper.DeviceStateMapper;
 import com.tuya.open.sdk.mq.AESBase64Utils;
 import com.tuya.open.sdk.mq.MqConfigs;
 import com.tuya.open.sdk.mq.MqConsumer;
@@ -63,14 +65,32 @@ public class TuyaMessageSubscribeService implements SmartLifecycle {
 
     private final TuyaCloudService tuyaCloudService;
     private final DeviceMapper deviceMapper;
+    private final DeviceStateMapper deviceStateMapper;
     private final DeviceEventPublisher deviceEventPublisher;
     private final ObjectMapper objectMapper;
     private final DeviceActivityService deviceActivityService;
     private final TuyaPulsarEventAdapter tuyaPulsarEventAdapter;
 
     private volatile boolean running = false;
+    /**
+     * 是否已收到过消息（真正"连上了"的唯一可靠证据）
+     *
+     * <p>涂鸦 SDK 的 MqConsumer 没有暴露连接状态接口，且 {@code start()} 是阻塞式的
+     * （内部 subscribe + receive），因此"调用了 start()"既不能证明连上、也不能阻塞主循环。
+     * 这里用"是否收到过消息"作为连接成功的判据。</p>
+     */
     private volatile boolean connected = false;
+    /** 是否已发起连接（start() 已调用） */
+    private volatile boolean connecting = false;
+    /** 累计收到的消息条数 */
+    private volatile long messageCount = 0L;
+    /** 最近一次发起连接的时间 */
+    private volatile long lastConnectAt = 0L;
+    /** 最近一次连接错误 */
+    private volatile String lastError = "";
     private volatile String currentAccessKey;
+    /** 当前实际使用的订阅服务器地址（按区域解析） */
+    private volatile String currentServerUrl = "";
     private volatile long lastMessageAt = 0L;
     private volatile String lastMessageSummary = "尚未收到消息";
     private Thread subscribeThread;
@@ -119,6 +139,7 @@ public class TuyaMessageSubscribeService implements SmartLifecycle {
                 String region = StringUtils.hasText(config.getApiRegion())
                         ? config.getApiRegion().toLowerCase() : "cn";
                 String serverUrl = resolveServerUrl(region);
+                currentServerUrl = serverUrl;
                 currentAccessKey = config.getAccessSecret();
 
                 MqConsumer consumer = MqConsumer.build()
@@ -129,23 +150,67 @@ public class TuyaMessageSubscribeService implements SmartLifecycle {
 
                 log.info("涂鸦消息订阅启动中: serverUrl={}, accessId={}, region={}",
                         serverUrl, config.getAccessId(), region);
-                consumer.start();
-                // start() 正常返回：连接已建立，SDK 在内部线程接收消息，这里保活等待
-                connected = true;
-                log.info("涂鸦消息订阅已连接，开始接收设备状态推送（上下线/DP上报）");
+
+                // SDK 的 start() 是阻塞式的（内部完成 subscribe 后一直 receive），
+                // 必须在独立线程里调用：
+                //   1) 主循环才能感知"消费者线程已退出"并触发重连；
+                //   2) 否则连接一旦在 start() 内抛异常就没有任何重试机会。
+                final MqConsumer finalConsumer = consumer;
+                Thread consumerThread = new Thread(() -> {
+                    try {
+                        finalConsumer.start();
+                    } catch (Exception e) {
+                        lastError = e.getClass().getSimpleName() + ": " + e.getMessage();
+                        log.warn("涂鸦消息订阅消费者异常退出: {}", e.getMessage());
+                    } finally {
+                        connecting = false;
+                    }
+                }, "tuya-mq-consumer");
+                consumerThread.setDaemon(true);
+                consumerThread.start();
+                connecting = true;
+                connected = false;
+                lastConnectAt = System.currentTimeMillis();
+                lastError = "";
+
+                // 给 SDK 一点时间完成鉴权与订阅：若立刻失败（认证/订阅不存在），
+                // 消费者线程会很快退出，这里据此提前失败并进入退避重试。
+                Thread.sleep(3000L);
+                if (!consumerThread.isAlive()) {
+                    connecting = false;
+                    throw new IllegalStateException("涂鸦消息订阅消费者启动后立即退出，请检查云项目"
+                            + "「消息订阅」是否已开通（且已开通本项目产品的消息类型）");
+                }
+                log.info("涂鸦消息订阅已发起连接（SDK 阻塞式接收，等待首条消息确认连通性）");
+
+                // 保活：监控消费者线程，退出即重连
                 while (running) {
                     Thread.sleep(5000L);
+                    if (!consumerThread.isAlive()) {
+                        connecting = false;
+                        connected = false;
+                        log.warn("涂鸦消息订阅消费者线程已退出，触发重连");
+                        break;
+                    }
                 }
-                connected = false;
-                return;
+                if (!running) {
+                    return;
+                }
+                sleepQuietly(backoffMs);
+                backoffMs = Math.min(backoffMs * 2, 600_000L);
+                continue;
             } catch (InterruptedException ie) {
                 Thread.currentThread().interrupt();
                 connected = false;
+                connecting = false;
                 return;
             } catch (Exception e) {
                 connected = false;
+                connecting = false;
+                lastError = e.getClass().getSimpleName() + ": " + e.getMessage();
                 log.warn("涂鸦消息订阅连接失败/中断: {}，{}秒后重试"
-                                + "（若持续失败请到涂鸦开发者平台云项目的「消息订阅」页面确认已开启）",
+                                + "（若持续失败请到涂鸦开发者平台云项目的「消息订阅」页面确认已开启，"
+                                + "并确认已订阅本项目产品的消息类型）",
                         e.getMessage(), backoffMs / 1000);
                 sleepQuietly(backoffMs);
                 backoffMs = Math.min(backoffMs * 2, 600_000L);
@@ -158,6 +223,8 @@ public class TuyaMessageSubscribeService implements SmartLifecycle {
     private void onMessageArrived(org.apache.pulsar.client.api.Message message) {
         try {
             lastMessageAt = System.currentTimeMillis();
+            messageCount++;
+            connected = true;   // 收到消息 = 连接与订阅均已生效（唯一可靠证据）
             String payload = new String(message.getData(), StandardCharsets.UTF_8);
             JsonNode root = objectMapper.readTree(payload);
             // 涂鸦推送存在两种标识：老协议用数字 protocol（4=上下线, 8/9=DP 上报），
@@ -206,6 +273,11 @@ public class TuyaMessageSubscribeService implements SmartLifecycle {
                     if ("online".equals(ev.getType()) || "offline".equals(ev.getType())) {
                         syncDeviceOnlineStatus(ev.getDeviceId(), "online".equals(ev.getType()));
                     }
+                    // 属性上报落 device_state：子设备（网关下挂）没有独立在线通道，
+                    // 但只要有属性上报就说明它通电且链路正常 —— 这是判断子设备在线的权威证据。
+                    if ("state".equals(ev.getType()) && ev.getDps() != null && !ev.getDps().isEmpty()) {
+                        syncDeviceStateFromDps(ev.getDeviceId(), ev.getDps());
+                    }
                 }
                 lastMessageSummary = effectiveBizCode + " → " + events.size() + " 事件";
             }
@@ -311,14 +383,89 @@ public class TuyaMessageSubscribeService implements SmartLifecycle {
         }
     }
 
+    /**
+     * 把涂鸦属性上报写入 device_state（属性级当前状态）。
+     *
+     * <p>device_state 此前只有中控屏上报一条写入路径，导致"子设备最近是否有活动"这类
+     * 在线证据无处可查（表恒为空，看起来像订阅没通）。这里把云端属性上报也落到同一张表：
+     * 网关下挂的子设备虽无独立在线通道，但只要有属性上报，就说明它通电且链路正常 ——
+     * 这正是判定子设备在线的权威证据（强于"父网关在线"的推断）。</p>
+     */
+    private void syncDeviceStateFromDps(String devId, Map<String, Object> dps) {
+        if (!StringUtils.hasText(devId) || dps == null || dps.isEmpty()) {
+            return;
+        }
+        Device device = deviceMapper.selectByDeviceId(devId);
+        if (device == null) {
+            log.debug("涂鸦推送的设备未入库，忽略属性状态落库: {}", devId);
+            return;
+        }
+        java.time.LocalDateTime now = java.time.LocalDateTime.now();
+        for (Map.Entry<String, Object> entry : dps.entrySet()) {
+            String code = entry.getKey();
+            if (!StringUtils.hasText(code)) {
+                continue;
+            }
+            DeviceState existing = deviceStateMapper.selectOne(
+                    new com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper<DeviceState>()
+                            .eq(DeviceState::getDeviceId, devId)
+                            .eq(DeviceState::getPropertyCode, code)
+                            .last("LIMIT 1"));
+            DeviceState state = existing != null ? existing : new DeviceState();
+            state.setDeviceId(devId);
+            state.setPropertyCode(code);
+            state.setPropertyValue(entry.getValue() == null ? null : String.valueOf(entry.getValue()));
+            state.setQuality("GOOD");
+            state.setReportedAt(now);
+            state.setReceivedAt(now);
+            if (state.getId() == null) {
+                deviceStateMapper.insert(state);
+            } else {
+                deviceStateMapper.updateById(state);
+            }
+        }
+        // 顺带把最新 DP 写入 device.capabilities，供在线判定与详情展示使用
+        try {
+            com.fasterxml.jackson.databind.JsonNode capsNode = device.getCapabilities();
+            java.util.Map<String, Object> capsMap = new java.util.LinkedHashMap<>();
+            if (capsNode != null && capsNode.isObject()) {
+                capsMap.putAll(objectMapper.convertValue(capsNode, java.util.Map.class));
+            }
+            capsMap.put("dps", dps);
+            Device update = new Device();
+            update.setId(device.getId());
+            update.setCapabilities(objectMapper.valueToTree(capsMap));
+            deviceMapper.updateById(update);
+        } catch (Exception e) {
+            log.debug("写入 device.capabilities 失败: {}", e.getMessage());
+        }
+    }
+
     // ==================== 状态查询（供管理端/排查） ====================
 
-    /** 订阅连接状态 */
+    /**
+     * 订阅连接状态
+     *
+     * <p>字段说明：</p>
+     * <ul>
+     *   <li>{@code running}：订阅线程是否在运行；</li>
+     *   <li>{@code connecting}：是否已发起连接（SDK 已调用 start()）；</li>
+     *   <li>{@code connected}：<b>是否收到过消息</b> —— SDK 没有暴露连接状态接口，
+     *       这是唯一可靠的"真的连上了"证据；</li>
+     *   <li>{@code messageCount}：累计收到的消息条数；</li>
+     *   <li>{@code lastError}：最近一次连接错误（便于定位认证/订阅问题）。</li>
+     * </ul>
+     */
     public Map<String, Object> getStatus() {
         Map<String, Object> m = new LinkedHashMap<>();
         m.put("running", running);
+        m.put("connecting", connecting);
         m.put("connected", connected);
-        m.put("serverUrl", MqConfigs.CN_SERVER_URL);
+        m.put("messageCount", messageCount);
+        m.put("lastConnectAt", lastConnectAt);
+        m.put("lastError", lastError);
+        m.put("serverUrl", StringUtils.hasText(currentServerUrl)
+                ? currentServerUrl : MqConfigs.CN_SERVER_URL);
         m.put("lastMessageAt", lastMessageAt);
         m.put("lastMessageSummary", lastMessageSummary);
         return m;

backend/iot-device/src/test/java/com/iot/device/collector/TuyaPulsarEventAdapterTest.java

@@ -0,0 +1,77 @@
+package com.iot.device.collector;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.iot.device.dto.DeviceEvent;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * 涂鸦推送事件归一化单测
+ *
+ * <p>重点：protocol=4（设备上下线）必须按 payload 的 status 取值判定，
+ * 不能一律当成上线 —— 否则离线推送会把设备状态写成在线。</p>
+ */
+@DisplayName("涂鸦推送事件归一化")
+class TuyaPulsarEventAdapterTest {
+
+    private final ObjectMapper mapper = new ObjectMapper();
+    private final TuyaPulsarEventAdapter adapter = new TuyaPulsarEventAdapter(mapper);
+
+    private List<DeviceEvent> normalize(String payloadJson, Integer protocol, String bizCode) throws Exception {
+        PlatformEventInput input = new PlatformEventInput();
+        input.setPlatform(DevicePlatform.TUYA.code());
+        input.setPayload(mapper.readTree(payloadJson));
+        input.setBizCode(bizCode);
+        input.setProtocol(protocol);
+        input.setRawJson(payloadJson);
+        return adapter.normalize(input);
+    }
+
+    @Test
+    @DisplayName("protocol=4 且 status=0 判为离线")
+    void shouldResolveOfflineWhenStatusZero() throws Exception {
+        List<DeviceEvent> events = normalize("{\"devId\":\"dev_1\",\"status\":0}", 4, null);
+        assertEquals(1, events.size());
+        assertEquals("offline", events.get(0).getType());
+    }
+
+    @Test
+    @DisplayName("protocol=4 且 status=1 判为在线")
+    void shouldResolveOnlineWhenStatusOne() throws Exception {
+        List<DeviceEvent> events = normalize("{\"devId\":\"dev_1\",\"status\":1}", 4, null);
+        assertEquals(1, events.size());
+        assertEquals("online", events.get(0).getType());
+    }
+
+    @Test
+    @DisplayName("protocol=4 且 online=false 判为离线")
+    void shouldResolveOfflineWhenOnlineFalse() throws Exception {
+        List<DeviceEvent> events = normalize("{\"devId\":\"dev_1\",\"online\":false}", 4, null);
+        assertEquals(1, events.size());
+        assertEquals("offline", events.get(0).getType());
+    }
+
+    @Test
+    @DisplayName("新版 bizCode=deviceOffline 判为离线")
+    void shouldResolveOfflineByBizCode() throws Exception {
+        List<DeviceEvent> events = normalize("{\"devId\":\"dev_1\",\"status\":0}", 1001, "deviceOffline");
+        assertEquals(1, events.size());
+        assertEquals("offline", events.get(0).getType());
+    }
+
+    @Test
+    @DisplayName("属性上报解析出 dps")
+    void shouldParseDpPayload() throws Exception {
+        List<DeviceEvent> events = normalize(
+                "{\"devId\":\"dev_1\",\"dps\":{\"switch_1\":true}}", 8, null);
+        assertEquals(1, events.size());
+        DeviceEvent event = events.get(0);
+        assertEquals("state", event.getType());
+        assertTrue(event.getDps() != null && event.getDps().containsKey("switch_1"));
+    }
+}