feat(iot): 合入设备影子能力并完善状态同步

本地合入 Gitee !1561，沿用数据库 version CAS 与三次重试。修正影子缓存、删除事务、状态开关及清空校验，补充必要单测。新增影子权限种子，业务表已通过仓库外升级 SQL 在本地 MySQL 补建。

feat(iot): 合入设备影子能力并完善状态同步
本地合入 Gitee !1561，沿用数据库 version CAS 与三次重试。修正影子缓存、删除事务、状态开关及清空校验，补充必要单测。新增影子权限种子，业务表已通过仓库外升级 SQL 在本地 MySQL 补建。
YunaiV
2 parents 39df7c49 8865481b
Showing 25 changed files with 1986 additions and 5 deletions
sql/mysql/ruoyi-vue-pro.sql
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/device/IotDeviceShadowController.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/device/vo/shadow/IotDeviceShadowPropertyRespVO.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/device/vo/shadow/IotDeviceShadowRespVO.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/device/vo/shadow/IotDeviceShadowUpdateDesiredReqVO.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/device/vo/shadow/IotDeviceShadowUpdateStatusReqVO.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/device/IotDeviceShadowDO.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/device/IotDeviceShadowMapper.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/redis/RedisKeyConstants.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/redis/device/DeviceShadowRedisDAO.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/enums/ErrorCodeConstants.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/IotDeviceServiceImpl.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/message/IotDeviceMessageService.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/message/IotDeviceMessageServiceImpl.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/shadow/IotDeviceShadowService.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/shadow/IotDeviceShadowServiceImpl.java
yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/device/IotDeviceServiceImplTest.java
yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/device/shadow/IotDeviceShadowServiceImplTest.java
yudao-module-iot/yudao-module-iot-biz/src/test/resources/sql/clean.sql
yudao-module-iot/yudao-module-iot-biz/src/test/resources/sql/create_tables.sql
@@ -2799,7 +2799,7 @@ CREATE TABLE `system_menu`  (
   `update_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
   `deleted` bit(1) NOT NULL DEFAULT b'0' COMMENT '是否删除',
   PRIMARY KEY (`id`) USING BTREE
-) ENGINE = InnoDB AUTO_INCREMENT = 13478 CHARACTER SET = utf8mb4 COLLATE = utf8mb4_unicode_ci COMMENT = '菜单权限表';
+) ENGINE = InnoDB AUTO_INCREMENT = 13480 CHARACTER SET = utf8mb4 COLLATE = utf8mb4_unicode_ci COMMENT = '菜单权限表';
 -- ----------------------------
 -- Records of system_menu
@@ -5349,6 +5349,8 @@ INSERT INTO `system_menu` (`id`, `name`, `permission`, `type`, `sort`, `parent_i
 INSERT INTO `system_menu` (`id`, `name`, `permission`, `type`, `sort`, `parent_id`, `path`, `icon`, `component`, `component_name`, `status`, `visible`, `keep_alive`, `always_show`, `creator`, `create_time`, `updater`, `update_time`, `deleted`) VALUES (13475, '维修备件领用', 'eam:repair-order-act:maintenance', 3, 5, 13249, '', '', '', NULL, 0, b'1', b'1', b'1', '', '2026-09-26 20:50:45', '', '2026-09-26 20:50:45', b'0');
 INSERT INTO `system_menu` (`id`, `name`, `permission`, `type`, `sort`, `parent_id`, `path`, `icon`, `component`, `component_name`, `status`, `visible`, `keep_alive`, `always_show`, `creator`, `create_time`, `updater`, `update_time`, `deleted`) VALUES (13476, '维修执行单取消', 'eam:repair-order-act:cancel', 3, 6, 13249, '', '', '', NULL, 0, b'1', b'1', b'1', '', '2026-09-26 20:50:45', '', '2026-09-26 20:50:45', b'0');
 INSERT INTO `system_menu` (`id`, `name`, `permission`, `type`, `sort`, `parent_id`, `path`, `icon`, `component`, `component_name`, `status`, `visible`, `keep_alive`, `always_show`, `creator`, `create_time`, `updater`, `update_time`, `deleted`) VALUES (13477, '维修执行单修改', 'eam:repair-order-act:update', 3, 7, 13249, '', '', '', NULL, 0, b'1', b'1', b'1', '', '2026-09-26 20:50:45', '', '2026-09-26 20:50:45', b'0');
+INSERT INTO `system_menu` (`id`, `name`, `permission`, `type`, `sort`, `parent_id`, `path`, `icon`, `component`, `component_name`, `status`, `visible`, `keep_alive`, `always_show`, `creator`, `create_time`, `updater`, `update_time`, `deleted`) VALUES (13478, '设备影子查询', 'iot:device:shadow-query', 3, 13, 868, '', '', '', '', 0, b'1', b'1', b'1', '1', '2026-10-26 00:00:00', '1', '2026-10-26 00:00:00', b'0');
+INSERT INTO `system_menu` (`id`, `name`, `permission`, `type`, `sort`, `parent_id`, `path`, `icon`, `component`, `component_name`, `status`, `visible`, `keep_alive`, `always_show`, `creator`, `create_time`, `updater`, `update_time`, `deleted`) VALUES (13479, '设备影子更新', 'iot:device:shadow-update', 3, 14, 868, '', '', '', '', 0, b'1', b'1', b'1', '1', '2026-10-26 00:00:00', '1', '2026-10-26 00:00:00', b'0');
 COMMIT;
 -- ----------------------------
+package cn.iocoder.yudao.module.iot.controller.admin.device;
+
+import cn.hutool.core.lang.Assert;
+import cn.hutool.core.map.MapUtil;
+import cn.iocoder.yudao.framework.common.pojo.CommonResult;
+import cn.iocoder.yudao.framework.common.util.object.BeanUtils;
+import cn.iocoder.yudao.module.iot.controller.admin.device.vo.shadow.IotDeviceShadowPropertyRespVO;
+import cn.iocoder.yudao.module.iot.controller.admin.device.vo.shadow.IotDeviceShadowRespVO;
+import cn.iocoder.yudao.module.iot.controller.admin.device.vo.shadow.IotDeviceShadowUpdateDesiredReqVO;
+import cn.iocoder.yudao.module.iot.controller.admin.device.vo.shadow.IotDeviceShadowUpdateStatusReqVO;
+import cn.iocoder.yudao.module.iot.core.util.IotDeviceShadowProtocolUtils;
+import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceDO;
+import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceShadowDO;
+import cn.iocoder.yudao.module.iot.dal.dataobject.thingmodel.IotThingModelDO;
+import cn.iocoder.yudao.module.iot.dal.dataobject.thingmodel.model.ThingModelProperty;
+import cn.iocoder.yudao.module.iot.enums.thingmodel.IotThingModelTypeEnum;
+import cn.iocoder.yudao.module.iot.service.device.IotDeviceService;
+import cn.iocoder.yudao.module.iot.service.device.shadow.IotDeviceShadowService;
+import cn.iocoder.yudao.module.iot.service.thingmodel.IotThingModelService;
+import io.swagger.v3.oas.annotations.Operation;
+import io.swagger.v3.oas.annotations.Parameter;
+import io.swagger.v3.oas.annotations.tags.Tag;
+import jakarta.annotation.Resource;
+import jakarta.validation.Valid;
+import org.springframework.security.access.prepost.PreAuthorize;
+import org.springframework.validation.annotation.Validated;
+import org.springframework.web.bind.annotation.*;
+
+import java.util.List;
+import java.util.Map;
+
+import static cn.iocoder.yudao.framework.common.pojo.CommonResult.success;
+import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.convertList;
+
+/**
+ * 管理后台 - IoT 设备影子 Controller
+ *
+ * @author 芋道源码
+ */
+@Tag(name = "管理后台 - IoT 设备影子")
+@RestController
+@RequestMapping("/iot/device/shadow")
+@Validated
+public class IotDeviceShadowController {
+
+    @Resource
+    private IotDeviceShadowService deviceShadowService;
+    @Resource
+    private IotDeviceService deviceService;
+    @Resource
+    private IotThingModelService thingModelService;
+
+    @GetMapping("/get")
+    @Operation(summary = "获取设备影子")
+    @Parameter(name = "deviceId", description = "设备编号", required = true)
+    @PreAuthorize("@ss.hasPermission('iot:device:shadow-query')")
+    public CommonResult<IotDeviceShadowRespVO> getDeviceShadow(@RequestParam("deviceId") Long deviceId) {
+        IotDeviceDO device = deviceService.validateDeviceExists(deviceId);
+        IotDeviceShadowDO shadow = deviceShadowService.getOrCreateShadow(device);
+
+        List<IotThingModelDO> thingModels = thingModelService.getThingModelListByProductIdAndType(
+                device.getProductId(), IotThingModelTypeEnum.PROPERTY.getType());
+        Map<String, Object> desired = shadow.getDesired() != null ? shadow.getDesired() : MapUtil.empty();
+        Map<String, Object> reported = shadow.getReported() != null ? shadow.getReported() : MapUtil.empty();
+        Map<String, Object> metadata = shadow.getMetadata() != null ? shadow.getMetadata() : MapUtil.empty();
+        // 拼接 VO
+        List<IotDeviceShadowPropertyRespVO> properties = convertList(thingModels, thingModel -> {
+            ThingModelProperty prop = thingModel.getProperty();
+            Assert.notNull(prop, "属性不能为空");
+            return new IotDeviceShadowPropertyRespVO()
+                    .setIdentifier(thingModel.getIdentifier()).setName(thingModel.getName())
+                    .setDataType(prop.getDataType()).setDataSpecs(prop.getDataSpecs())
+                    .setDataSpecsList(prop.getDataSpecsList())
+                    .setDesiredValue(desired.get(thingModel.getIdentifier()))
+                    .setReportedValue(reported.get(thingModel.getIdentifier()))
+                    .setDesiredTime(IotDeviceShadowProtocolUtils.getDesiredTimestamp(metadata, thingModel.getIdentifier()))
+                    .setReportedTime(IotDeviceShadowProtocolUtils.getReportedTimestamp(metadata, thingModel.getIdentifier()));
+        });
+
+        IotDeviceShadowRespVO respVO = BeanUtils.toBean(shadow, IotDeviceShadowRespVO.class)
+                .setProperties(properties)
+                .setDesired(desired)
+                .setReported(reported)
+                .setMetadata(metadata);
+        return success(respVO);
+    }
+
+    @PutMapping("/update-desired")
+    @Operation(summary = "更新设备影子期望状态")
+    @PreAuthorize("@ss.hasPermission('iot:device:shadow-update')")
+    public CommonResult<Boolean> updateDesiredState(@Valid @RequestBody IotDeviceShadowUpdateDesiredReqVO reqVO) {
+        deviceShadowService.updateDesiredState(reqVO.getDeviceId(), reqVO.getDesired());
+        return success(true);
+    }
+
+    @PutMapping("/update-status")
+    @Operation(summary = "更新设备影子开关状态")
+    @PreAuthorize("@ss.hasPermission('iot:device:shadow-update')")
+    public CommonResult<Boolean> updateShadowStatus(@Valid @RequestBody IotDeviceShadowUpdateStatusReqVO reqVO) {
+        deviceShadowService.updateStatus(reqVO.getDeviceId(), reqVO.getStatus());
+        return success(true);
+    }
+
+    @DeleteMapping("/reset")
+    @Operation(summary = "重置设备影子")
+    @Parameter(name = "deviceId", description = "设备编号", required = true)
+    @PreAuthorize("@ss.hasPermission('iot:device:shadow-update')")
+    public CommonResult<Boolean> resetShadow(@RequestParam("deviceId") Long deviceId) {
+        deviceShadowService.resetShadow(deviceId);
+        return success(true);
+    }
+
+}
+package cn.iocoder.yudao.module.iot.controller.admin.device.vo.shadow;
+
+import cn.iocoder.yudao.module.iot.dal.dataobject.thingmodel.model.dataType.ThingModelDataSpecs;
+import io.swagger.v3.oas.annotations.media.Schema;
+import lombok.Data;
+
+import java.util.List;
+
+/**
+ * 管理后台 - IoT 设备影子属性对比 Response VO
+ *
+ * @author 芋道源码
+ */
+@Schema(description = "管理后台 - IoT 设备影子属性对比 Response VO")
+@Data
+public class IotDeviceShadowPropertyRespVO {
+
+    @Schema(description = "标识符", example = "Temperature")
+    private String identifier;
+
+    @Schema(description = "属性名称", example = "温度")
+    private String name;
+
+    @Schema(description = "数据类型", example = "int")
+    private String dataType;
+
+    @Schema(description = "数据定义")
+    private ThingModelDataSpecs dataSpecs;
+
+    @Schema(description = "数据定义列表")
+    private List<ThingModelDataSpecs> dataSpecsList;
+
+    @Schema(description = "期望值")
+    private Object desiredValue;
+
+    @Schema(description = "上报值")
+    private Object reportedValue;
+
+    @Schema(description = "期望更新时间（epoch 秒）")
+    private Long desiredTime;
+
+    @Schema(description = "上报更新时间（epoch 秒）")
+    private Long reportedTime;
+
+}
+package cn.iocoder.yudao.module.iot.controller.admin.device.vo.shadow;
+
+import io.swagger.v3.oas.annotations.media.Schema;
+import lombok.Data;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * 管理后台 - IoT 设备影子 Response VO
+ *
+ * @author 芋道源码
+ */
+@Schema(description = "管理后台 - IoT 设备影子 Response VO")
+@Data
+public class IotDeviceShadowRespVO {
+
+    @Schema(description = "版本号", example = "3")
+    private Long version;
+
+    @Schema(description = "最后更新时间（epoch 秒）", example = "1469564576")
+    private Long shadowTime;
+
+    @Schema(description = "是否启用", example = "0")
+    private Integer status;
+
+    @Schema(description = "属性对比列表")
+    private List<IotDeviceShadowPropertyRespVO> properties;
+
+    @Schema(description = "期望状态")
+    private Map<String, Object> desired;
+
+    @Schema(description = "上报状态")
+    private Map<String, Object> reported;
+
+    @Schema(description = "元数据")
+    private Map<String, Object> metadata;
+
+}
+package cn.iocoder.yudao.module.iot.controller.admin.device.vo.shadow;
+
+import io.swagger.v3.oas.annotations.media.Schema;
+import jakarta.validation.constraints.NotNull;
+import lombok.Data;
+
+import java.util.Map;
+
+/**
+ * 管理后台 - IoT 设备影子更新期望状态 Request VO
+ *
+ * @author 芋道源码
+ */
+@Schema(description = "管理后台 - IoT 设备影子更新期望状态 Request VO")
+@Data
+public class IotDeviceShadowUpdateDesiredReqVO {
+
+    @Schema(description = "设备编号", requiredMode = Schema.RequiredMode.REQUIRED, example = "1")
+    @NotNull(message = "设备编号不能为空")
+    private Long deviceId;
+
+    @Schema(description = "期望状态", requiredMode = Schema.RequiredMode.REQUIRED)
+    @NotNull(message = "期望状态不能为空")
+    private Map<String, Object> desired;
+
+}
+package cn.iocoder.yudao.module.iot.controller.admin.device.vo.shadow;
+
+import cn.iocoder.yudao.framework.common.enums.CommonStatusEnum;
+import cn.iocoder.yudao.framework.common.validation.InEnum;
+import io.swagger.v3.oas.annotations.media.Schema;
+import jakarta.validation.constraints.NotNull;
+import lombok.Data;
+
+/**
+ * 管理后台 - IoT 设备影子更新状态 Request VO
+ *
+ * @author 芋道源码
+ */
+@Schema(description = "管理后台 - IoT 设备影子更新状态 Request VO")
+@Data
+public class IotDeviceShadowUpdateStatusReqVO {
+
+    @Schema(description = "设备编号", requiredMode = Schema.RequiredMode.REQUIRED, example = "1")
+    @NotNull(message = "设备编号不能为空")
+    private Long deviceId;
+
+    @Schema(description = "状态", requiredMode = Schema.RequiredMode.REQUIRED, example = "0")
+    @NotNull(message = "状态不能为空")
+    @InEnum(CommonStatusEnum.class)
+    private Integer status;
+
+}
+package cn.iocoder.yudao.module.iot.dal.dataobject.device;
+
+import cn.iocoder.yudao.framework.tenant.core.db.TenantBaseDO;
+import cn.iocoder.yudao.module.iot.dal.dataobject.product.IotProductDO;
+import com.baomidou.mybatisplus.annotation.KeySequence;
+import com.baomidou.mybatisplus.annotation.TableField;
+import com.baomidou.mybatisplus.annotation.TableId;
+import com.baomidou.mybatisplus.annotation.TableName;
+import com.baomidou.mybatisplus.extension.handlers.JacksonTypeHandler;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.util.Map;
+
+/**
+ * IoT 设备影子 DO
+ *
+ * @author 芋道源码
+ */
+@TableName(value = "iot_device_shadow", autoResultMap = true)
+@KeySequence("iot_device_shadow_seq")
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class IotDeviceShadowDO extends TenantBaseDO {
+
+    /**
+     * 主键
+     */
+    @TableId
+    private Long id;
+    /**
+     * 设备编号
+     */
+    private Long deviceId;
+    /**
+     * 产品编号
+     * <p>
+     * 关联 {@link IotProductDO#getId()}
+     */
+    private Long productId;
+    /**
+     * 产品标识
+     */
+    private String productKey;
+    /**
+     * 设备名称
+     */
+    private String deviceName;
+    /**
+     * 上报状态
+     */
+    @TableField(typeHandler = JacksonTypeHandler.class)
+    private Map<String, Object> reported;
+    /**
+     * 期望状态
+     */
+    @TableField(typeHandler = JacksonTypeHandler.class)
+    private Map<String, Object> desired;
+    /**
+     * 元数据：{ desired: { key: { timestamp } }, reported: { ... } }
+     */
+    @TableField(typeHandler = JacksonTypeHandler.class)
+    private Map<String, Object> metadata;
+    /**
+     * 版本号（乐观锁）
+     */
+    private Long version;
+    /**
+     * 文档最后更新时间（epoch 秒）
+     */
+    private Long shadowTime;
+    /**
+     * 状态
+     * <p>
+     * 枚举 {@link cn.iocoder.yudao.framework.common.enums.CommonStatusEnum}
+     */
+    private Integer status;
+
+}
+package cn.iocoder.yudao.module.iot.dal.mysql.device;
+
+import cn.iocoder.yudao.framework.mybatis.core.mapper.BaseMapperX;
+import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceShadowDO;
+import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
+import org.apache.ibatis.annotations.Mapper;
+
+/**
+ * IoT 设备影子 Mapper
+ *
+ * @author 芋道源码
+ */
+@Mapper
+public interface IotDeviceShadowMapper extends BaseMapperX<IotDeviceShadowDO> {
+
+    default IotDeviceShadowDO selectByDeviceId(Long deviceId) {
+        return selectOne(IotDeviceShadowDO::getDeviceId, deviceId);
+    }
+
+    default int updateByDeviceId(IotDeviceShadowDO update, Long deviceId) {
+        return update(update, new LambdaUpdateWrapper<IotDeviceShadowDO>()
+                .eq(IotDeviceShadowDO::getDeviceId, deviceId));
+    }
+
+    default int updateByDeviceIdAndVersion(IotDeviceShadowDO update, Long deviceId, Long expectVersion) {
+        return update(update, new LambdaUpdateWrapper<IotDeviceShadowDO>()
+                .eq(IotDeviceShadowDO::getDeviceId, deviceId)
+                .eq(IotDeviceShadowDO::getVersion, expectVersion));
+    }
+
+    default int deleteByDeviceId(Long deviceId) {
+        return delete(IotDeviceShadowDO::getDeviceId, deviceId);
+    }
+
+}
@@ -19,6 +19,15 @@ public interface RedisKeyConstants {
     String DEVICE_PROPERTY = "iot:device_property:%d";
     /**
+     * 设备影子数据缓存，采用 STRING 结构
+     * <p>
+     * 影子缓存：DB 为准，Redis 仅作读缓存（cache-aside），写后失效，过期时间 60 秒。
+     * KEY 格式：iot:device_shadow:{deviceId}
+     * VALUE 数据类型：String(JSON) {@link cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceShadowDO}
+     */
+    String DEVICE_SHADOW = "iot:device_shadow:%d";
+
+    /**
      * 设备的最后上报时间，采用 ZSET 结构
      *
      * KEY 格式：{deviceId}
+package cn.iocoder.yudao.module.iot.dal.redis.device;
+
+import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
+import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceShadowDO;
+import jakarta.annotation.Resource;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.stereotype.Repository;
+
+import java.time.Duration;
+
+import static cn.iocoder.yudao.module.iot.dal.redis.RedisKeyConstants.DEVICE_SHADOW;
+
+/**
+ * {@link IotDeviceShadowDO} 的 Redis DAO
+ * <p>
+ * 影子缓存：DB 为准，Redis 仅作读缓存（cache-aside），写后失效，并设置较短的过期时间兜底。
+ *
+ * @author 芋道源码
+ */
+@Repository
+public class DeviceShadowRedisDAO {
+
+    /**
+     * 缓存过期时间：并发回填旧数据时，最多在该时间内读到旧影子
+     */
+    private static final Duration TIMEOUT = Duration.ofSeconds(60);
+
+    @Resource
+    private StringRedisTemplate stringRedisTemplate;
+
+    public IotDeviceShadowDO get(Long deviceId) {
+        String redisKey = formatKey(deviceId);
+        String json = stringRedisTemplate.opsForValue().get(redisKey);
+        return JsonUtils.parseObject(json, IotDeviceShadowDO.class);
+    }
+
+    public void set(IotDeviceShadowDO shadow) {
+        String redisKey = formatKey(shadow.getDeviceId());
+        stringRedisTemplate.opsForValue().set(redisKey, JsonUtils.toJsonString(shadow), TIMEOUT);
+    }
+
+    public void delete(Long deviceId) {
+        stringRedisTemplate.delete(formatKey(deviceId));
+    }
+
+    private static String formatKey(Long deviceId) {
+        return String.format(DEVICE_SHADOW, deviceId);
+    }
+
+}
@@ -67,6 +67,11 @@ public interface ErrorCodeConstants {
     ErrorCode DEVICE_MODBUS_POINT_NOT_EXISTS = new ErrorCode(1_050_007_000, "设备 Modbus 点位配置不存在");
     ErrorCode DEVICE_MODBUS_POINT_EXISTS = new ErrorCode(1_050_007_001, "设备 Modbus 点位配置已存在");
+    // ========== 设备影子 1-050-007-100 ==========
+    ErrorCode DEVICE_SHADOW_NOT_ENABLED = new ErrorCode(1_050_007_100, "设备影子未启用");
+    ErrorCode DEVICE_SHADOW_VERSION_CONFLICT = new ErrorCode(1_050_007_101, "设备影子版本冲突");
+    ErrorCode DEVICE_SHADOW_PROPERTY_COUNT_EXCEEDED = new ErrorCode(1_050_007_102, "设备影子属性数超过限制");
+
     // ========== OTA 固件相关 1-050-008-000 ==========
     ErrorCode OTA_FIRMWARE_NOT_EXISTS = new ErrorCode(1_050_008_000, "固件信息不存在");
@@ -37,6 +37,7 @@ import cn.iocoder.yudao.module.iot.dal.mysql.device.IotDeviceMapper;
 import cn.iocoder.yudao.module.iot.dal.redis.RedisKeyConstants;
 import cn.iocoder.yudao.module.iot.enums.product.IotProductDeviceTypeEnum;
 import cn.iocoder.yudao.module.iot.service.device.message.IotDeviceMessageService;
+import cn.iocoder.yudao.module.iot.service.device.shadow.IotDeviceShadowService;
 import cn.iocoder.yudao.module.iot.service.product.IotProductService;
 import jakarta.annotation.Resource;
 import jakarta.validation.ConstraintViolationException;
@@ -81,6 +82,9 @@ public class IotDeviceServiceImpl implements IotDeviceService {
     @Resource
     @Lazy // 延迟加载，解决循环依赖
     private IotDeviceMessageService deviceMessageService;
+    @Resource
+    @Lazy // 延迟加载，解决循环依赖
+    private IotDeviceShadowService deviceShadowService;
     private IotDeviceServiceImpl getSelf() {
         return SpringUtil.getBean(getClass());
@@ -197,6 +201,7 @@ public class IotDeviceServiceImpl implements IotDeviceService {
     }
     @Override
+    @Transactional(rollbackFor = Exception.class)
     public void deleteDevice(Long id) {
         // 1.1 校验存在
         IotDeviceDO device = validateDeviceExists(id);
@@ -209,7 +214,10 @@ public class IotDeviceServiceImpl implements IotDeviceService {
         // 2. 删除设备
         deviceMapper.deleteById(id);
-        // 3. 清空对应缓存
+        // 3. 级联删除影子
+        deviceShadowService.deleteByDeviceId(id);
+
+        // 4. 清空对应缓存（事务提交后生效）
         deleteDeviceCache(device);
     }
@@ -235,7 +243,10 @@ public class IotDeviceServiceImpl implements IotDeviceService {
         // 2. 删除设备
         deviceMapper.deleteByIds(ids);
-        // 3. 清空对应缓存
+        // 3. 级联删除影子
+        ids.forEach(deviceShadowService::deleteByDeviceId);
+
+        // 4. 清空对应缓存（事务提交后生效）
         deleteDeviceCache(devices);
     }
@@ -30,6 +30,16 @@ public interface IotDeviceMessageService {
     IotDeviceMessage sendDeviceMessage(IotDeviceMessage message, IotDeviceDO device);
     /**
+     * 发送设备消息（指定 serverId，用于上行回复等场景）
+     *
+     * @param message  消息
+     * @param device   设备
+     * @param serverId 网关 serverId
+     * @return 设备消息
+     */
+    IotDeviceMessage sendDeviceMessage(IotDeviceMessage message, IotDeviceDO device, String serverId);
+
+    /**
      * 发送设备消息
      *
      * @param message 消息（“codec（编解码）字段” 部分字段）
@@ -28,6 +28,7 @@ import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceMessageDO;
 import cn.iocoder.yudao.module.iot.dal.storage.IotDeviceMessageDAO;
 import cn.iocoder.yudao.module.iot.service.device.IotDeviceService;
 import cn.iocoder.yudao.module.iot.service.device.property.IotDevicePropertyService;
+import cn.iocoder.yudao.module.iot.service.device.shadow.IotDeviceShadowService;
 import cn.iocoder.yudao.module.iot.service.ota.IotOtaTaskRecordService;
 import com.google.common.base.Objects;
 import jakarta.annotation.Resource;
@@ -63,6 +64,9 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService {
     @Resource
     @Lazy // 延迟加载，避免循环依赖
     private IotOtaTaskRecordService otaTaskRecordService;
+    @Resource
+    @Lazy // 延迟加载，避免循环依赖
+    private IotDeviceShadowService deviceShadowService;
     @Resource
     private IotDeviceMessageDAO deviceMessageDAO;
@@ -110,7 +114,8 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService {
         return sendDeviceMessage(message, device, null);
     }
-    private IotDeviceMessage sendDeviceMessage(IotDeviceMessage message, IotDeviceDO device, String serverId) {
+    @Override
+    public IotDeviceMessage sendDeviceMessage(IotDeviceMessage message, IotDeviceDO device, String serverId) {
         // 1. 补充信息
         appendDeviceMessage(message, device);
@@ -219,6 +224,12 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService {
             return null;
         }
+        // 设备影子更新
+        if (Objects.equal(message.getMethod(), IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod())) {
+            deviceShadowService.handleDeviceShadowUpdate(device, message);
+            return null;
+        }
+
         // 添加拓扑关系
         if (Objects.equal(message.getMethod(), IotDeviceMessageMethodEnum.TOPO_ADD.getMethod())) {
             return deviceService.handleTopoAddMessage(message, device);
+package cn.iocoder.yudao.module.iot.service.device.shadow;
+
+import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
+import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceDO;
+import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceShadowDO;
+
+import java.util.Map;
+
+/**
+ * IoT 设备影子 Service 接口
+ *
+ * @author 芋道源码
+ */
+public interface IotDeviceShadowService {
+
+    /**
+     * 获取或创建设备影子
+     * <p>
+     * 优先读缓存，仅用于展示；写入、版本校验等，需以 DB 为准
+     *
+     * @param device 设备
+     * @return 设备影子
+     */
+    IotDeviceShadowDO getOrCreateShadow(IotDeviceDO device);
+
+    /**
+     * 处理设备影子上行更新
+     *
+     * @param device  设备
+     * @param message 上行消息
+     */
+    void handleDeviceShadowUpdate(IotDeviceDO device, IotDeviceMessage message);
+
+    /**
+     * 更新期望状态
+     *
+     * @param deviceId 设备编号
+     * @param desired  期望状态
+     */
+    void updateDesiredState(Long deviceId, Map<String, Object> desired);
+
+    /**
+     * 重置影子
+     *
+     * @param deviceId 设备编号
+     */
+    void resetShadow(Long deviceId);
+
+    /**
+     * 更新影子开关状态
+     *
+     * @param deviceId 设备编号
+     * @param status   状态
+     */
+    void updateStatus(Long deviceId, Integer status);
+
+    /**
+     * 删除设备影子（设备删除时级联）
+     *
+     * @param deviceId 设备编号
+     */
+    void deleteByDeviceId(Long deviceId);
+
+}
+package cn.iocoder.yudao.module.iot.service.device.shadow;
+
+import cn.hutool.core.collection.CollUtil;
+import cn.hutool.core.date.DateUtil;
+import cn.hutool.core.lang.Assert;
+import cn.hutool.core.util.ObjUtil;
+import cn.hutool.core.util.StrUtil;
+import cn.iocoder.yudao.framework.common.enums.CommonStatusEnum;
+import cn.iocoder.yudao.framework.common.exception.ServiceException;
+import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
+import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
+import cn.iocoder.yudao.module.iot.core.enums.shadow.IotDeviceShadowErrorCodeEnum;
+import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
+import cn.iocoder.yudao.module.iot.core.topic.shadow.IotDeviceShadowUpdateReqDTO;
+import cn.iocoder.yudao.module.iot.core.util.IotDeviceShadowProtocolUtils;
+import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceDO;
+import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceShadowDO;
+import cn.iocoder.yudao.module.iot.dal.mysql.device.IotDeviceShadowMapper;
+import cn.iocoder.yudao.module.iot.dal.redis.device.DeviceShadowRedisDAO;
+import cn.iocoder.yudao.module.iot.service.device.IotDeviceService;
+import cn.iocoder.yudao.module.iot.service.device.message.IotDeviceMessageService;
+import cn.iocoder.yudao.module.iot.service.device.property.IotDevicePropertyService;
+import jakarta.annotation.Resource;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.context.annotation.Lazy;
+import org.springframework.dao.DuplicateKeyException;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.support.TransactionSynchronization;
+import org.springframework.transaction.support.TransactionSynchronizationManager;
+import org.springframework.validation.annotation.Validated;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+
+import static cn.iocoder.yudao.framework.common.exception.enums.GlobalErrorCodeConstants.SUCCESS;
+import static cn.iocoder.yudao.framework.common.exception.util.ServiceExceptionUtil.exception;
+import static cn.iocoder.yudao.module.iot.core.util.IotDeviceShadowProtocolUtils.INNER_METHOD_GET;
+import static cn.iocoder.yudao.module.iot.core.util.IotDeviceShadowProtocolUtils.INNER_METHOD_UPDATE;
+import static cn.iocoder.yudao.module.iot.enums.ErrorCodeConstants.*;
+
+/**
+ * IoT 设备影子 Service 实现类
+ * <p>
+ * DB 为准：写入、版本校验、开关判断都读 DB；Redis 仅用于管理后台查询的读加速
+ *
+ * @author 芋道源码
+ */
+@Service
+@Validated
+@Slf4j
+public class IotDeviceShadowServiceImpl implements IotDeviceShadowService {
+
+    /**
+     * desired、reported 各自的最大属性数
+     */
+    private static final int MAX_PROPERTY_COUNT = 128;
+    /**
+     * 乐观锁冲突时的最大尝试次数
+     */
+    private static final int OPTIMISTIC_LOCK_RETRY = 3;
+
+    @Resource
+    private IotDeviceShadowMapper deviceShadowMapper;
+    @Resource
+    private DeviceShadowRedisDAO deviceShadowRedisDAO;
+
+    @Resource
+    private IotDeviceService deviceService;
+    @Resource
+    private IotDevicePropertyService devicePropertyService;
+    @Resource
+    @Lazy // 延迟加载，解决循环依赖
+    private IotDeviceMessageService deviceMessageService;
+
+    @Override
+    public IotDeviceShadowDO getOrCreateShadow(IotDeviceDO device) {
+        // 1. 优先读缓存：仅用于展示，允许在缓存过期前读到旧影子
+        IotDeviceShadowDO shadow = deviceShadowRedisDAO.get(device.getId());
+        if (shadow != null) {
+            return shadow;
+        }
+        // 2. 读 DB（不存在则创建），并回填缓存
+        shadow = getOrCreateDeviceShadowFromDB(device, CommonStatusEnum.ENABLE.getStatus());
+        deviceShadowRedisDAO.set(shadow);
+        return shadow;
+    }
+
+    @Override
+    public void handleDeviceShadowUpdate(IotDeviceDO device, IotDeviceMessage message) {
+        try {
+            handleDeviceShadowUpdate0(device, message);
+        } catch (Exception ex) {
+            log.error("[handleDeviceShadowUpdate][device({}) message({}) 处理异常]", device.getId(), message, ex);
+            sendShadowErrorReply(device, message, IotDeviceShadowErrorCodeEnum.SERVER_ERROR, 0L);
+        }
+    }
+
+    private void handleDeviceShadowUpdate0(IotDeviceDO device, IotDeviceMessage message) {
+        // 1.1 解析请求：params 结构为 { method, state: { reported }, version }
+        IotDeviceShadowUpdateReqDTO reqDTO = JsonUtils.convertObject(message.getParams(),
+                IotDeviceShadowUpdateReqDTO.class);
+        if (reqDTO == null) {
+            sendShadowErrorReply(device, message, IotDeviceShadowErrorCodeEnum.JSON_FORMAT_ERROR, 0L);
+            return;
+        }
+        // 1.2 校验影子已启用：读 DB，避免缓存中的旧开关
+        IotDeviceShadowDO shadow = getOrCreateDeviceShadowFromDB(device, CommonStatusEnum.ENABLE.getStatus());
+        if (CommonStatusEnum.isDisable(shadow.getStatus())) {
+            log.info("[handleDeviceShadowUpdate0][device({}) 影子未启用，忽略上行]", device.getId());
+            return;
+        }
+
+        // 2. 内层 method 分发
+        if (INNER_METHOD_GET.equals(reqDTO.getMethod())) {
+            sendShadowGetReply(device, message.getServerId(), message.getRequestId(), shadow);
+            return;
+        }
+        if (INNER_METHOD_UPDATE.equals(reqDTO.getMethod())) {
+            handleReportedUpdate(device, message, reqDTO, shadow);
+            return;
+        }
+        sendShadowErrorReply(device, message, StrUtil.isEmpty(reqDTO.getMethod())
+                ? IotDeviceShadowErrorCodeEnum.METHOD_MISSING : IotDeviceShadowErrorCodeEnum.INVALID_METHOD,
+                shadow.getVersion());
+    }
+
+    private void handleReportedUpdate(IotDeviceDO device, IotDeviceMessage message,
+                                      IotDeviceShadowUpdateReqDTO reqDTO, IotDeviceShadowDO shadow) {
+        // 1. version 为 -1 时，清空影子，无需 state
+        if (Objects.equals(reqDTO.getVersion(), -1L)) {
+            if (!resetDeviceShadow0(device.getId())) {
+                sendShadowVersionConflictReply(device, message);
+                return;
+            }
+            sendShadowReply(device, message.getServerId(), message.getRequestId(), SUCCESS.getCode(), SUCCESS.getMsg(),
+                    IotDeviceShadowProtocolUtils.buildUpdateData(0L, DateUtil.currentSeconds()));
+            return;
+        }
+
+        // 2. 校验 reported
+        IotDeviceShadowUpdateReqDTO.ShadowState state = reqDTO.getState();
+        if (state == null) {
+            sendShadowErrorReply(device, message, IotDeviceShadowErrorCodeEnum.STATE_MISSING, shadow.getVersion());
+            return;
+        }
+        if (state.getReported() == null) {
+            sendShadowErrorReply(device, message, IotDeviceShadowErrorCodeEnum.REPORTED_MISSING, shadow.getVersion());
+            return;
+        }
+        if (state.getReported().isEmpty()) {
+            sendShadowErrorReply(device, message, IotDeviceShadowErrorCodeEnum.REPORTED_EMPTY, shadow.getVersion());
+            return;
+        }
+
+        // 3. 合并更新 reported
+        IotDeviceShadowDO newShadow;
+        try {
+            newShadow = updateDeviceShadowState(device.getId(), false, state.getReported(), reqDTO.getVersion());
+        } catch (ServiceException ex) { // 合并后的属性数超过限制
+            sendShadowErrorReply(device, message, IotDeviceShadowErrorCodeEnum.PROPERTY_COUNT_EXCEEDED,
+                    shadow.getVersion());
+            return;
+        }
+        if (newShadow == null) {
+            sendShadowVersionConflictReply(device, message);
+            return;
+        }
+        sendShadowReply(device, message.getServerId(), message.getRequestId(), SUCCESS.getCode(), SUCCESS.getMsg(),
+                IotDeviceShadowProtocolUtils.buildUpdateData(newShadow.getVersion(), newShadow.getShadowTime()));
+    }
+
+    @Override
+    public void updateDesiredState(Long deviceId, Map<String, Object> desired) {
+        // 1. 校验影子已启用：读 DB，避免缓存中的旧开关
+        IotDeviceDO device = deviceService.validateDeviceExists(deviceId);
+        IotDeviceShadowDO shadow = getOrCreateDeviceShadowFromDB(device, CommonStatusEnum.ENABLE.getStatus());
+        if (CommonStatusEnum.isDisable(shadow.getStatus())) {
+            throw exception(DEVICE_SHADOW_NOT_ENABLED);
+        }
+        if (CollUtil.isEmpty(desired)) {
+            return;
+        }
+
+        // 2. 合并更新 desired
+        IotDeviceShadowDO newShadow = updateDeviceShadowState(deviceId, true, desired, null);
+        if (newShadow == null) {
+            throw exception(DEVICE_SHADOW_VERSION_CONFLICT);
+        }
+
+        // 3. 设备在线时，主动推送全量影子
+        String serverId = devicePropertyService.getDeviceServerId(deviceId);
+        if (StrUtil.isNotEmpty(serverId)) {
+            sendShadowGetReply(device, serverId, null, newShadow);
+        }
+    }
+
+    @Override
+    public void resetShadow(Long deviceId) {
+        deviceService.validateDeviceExists(deviceId);
+        if (!resetDeviceShadow0(deviceId)) {
+            throw exception(DEVICE_SHADOW_VERSION_CONFLICT);
+        }
+    }
+
+    @Override
+    public void updateStatus(Long deviceId, Integer status) {
+        // 1. 获得影子：不存在时，直接按目标状态创建
+        IotDeviceDO device = deviceService.validateDeviceExists(deviceId);
+        IotDeviceShadowDO shadow = getOrCreateDeviceShadowFromDB(device, status);
+
+        // 2. 更新状态
+        if (ObjUtil.notEqual(shadow.getStatus(), status)) {
+            deviceShadowMapper.updateByDeviceId(new IotDeviceShadowDO().setStatus(status), deviceId);
+        }
+        deleteDeviceShadowCache(deviceId);
+    }
+
+    @Override
+    public void deleteByDeviceId(Long deviceId) {
+        deviceShadowMapper.deleteByDeviceId(deviceId);
+        deleteDeviceShadowCache(deviceId);
+    }
+
+    /**
+     * 从 DB 获得影子，不存在时创建
+     *
+     * @param device 设备
+     * @param status 创建时的状态
+     * @return 影子
+     */
+    private IotDeviceShadowDO getOrCreateDeviceShadowFromDB(IotDeviceDO device, Integer status) {
+        IotDeviceShadowDO shadow = deviceShadowMapper.selectByDeviceId(device.getId());
+        if (shadow != null) {
+            return shadow;
+        }
+        // lazy 创建：首次访问时，才插入空影子；并发插入时，以先插入的为准
+        shadow = new IotDeviceShadowDO().setDeviceId(device.getId()).setProductId(device.getProductId())
+                .setProductKey(device.getProductKey()).setDeviceName(device.getDeviceName())
+                .setDesired(new HashMap<>()).setReported(new HashMap<>()).setMetadata(new HashMap<>())
+                .setVersion(0L).setShadowTime(DateUtil.currentSeconds()).setStatus(status);
+        try {
+            deviceShadowMapper.insert(shadow);
+            return shadow;
+        } catch (DuplicateKeyException ex) {
+            return deviceShadowMapper.selectByDeviceId(device.getId());
+        }
+    }
+
+    /**
+     * 乐观锁合并更新 desired 或 reported
+     * <p>
+     * 每次尝试都读 DB 最新影子重新合并，避免丢失并发更新
+     *
+     * @param deviceId      设备编号
+     * @param desired       true 更新 desired；false 更新 reported
+     * @param values        待合并的属性
+     * @param expectVersion 期望的当前版本；null 表示不校验
+     * @return 更新后的影子；版本冲突时，返回 null
+     */
+    private IotDeviceShadowDO updateDeviceShadowState(Long deviceId, boolean desired, Map<String, Object> values,
+                                                      Long expectVersion) {
+        for (int i = 0; i < OPTIMISTIC_LOCK_RETRY; i++) {
+            // 1. 校验版本
+            IotDeviceShadowDO latest = deviceShadowMapper.selectByDeviceId(deviceId);
+            Assert.notNull(latest, "设备影子({})不存在", deviceId);
+            if (expectVersion != null && ObjUtil.notEqual(expectVersion, latest.getVersion())) {
+                return null;
+            }
+
+            // 2.1 合并属性，并校验属性数
+            Map<String, Object> state = mergeMap(desired ? latest.getDesired() : latest.getReported(), values);
+            if (state.size() > MAX_PROPERTY_COUNT) {
+                throw exception(DEVICE_SHADOW_PROPERTY_COUNT_EXCEEDED);
+            }
+            // 2.2 记录属性的更新时间
+            long now = DateUtil.currentSeconds();
+            Map<String, Object> metadata = mergeMap(latest.getMetadata(), null);
+            if (desired) {
+                IotDeviceShadowProtocolUtils.putDesiredTimestamps(metadata, values.keySet(), now);
+            } else {
+                IotDeviceShadowProtocolUtils.putReportedTimestamps(metadata, values.keySet(), now);
+            }
+
+            // 3. 按 version 更新：成功则返回；失败则重试
+            IotDeviceShadowDO update = new IotDeviceShadowDO()
+                    .setDesired(desired ? state : latest.getDesired())
+                    .setReported(desired ? latest.getReported() : state)
+                    .setMetadata(metadata).setVersion(latest.getVersion() + 1).setShadowTime(now);
+            if (deviceShadowMapper.updateByDeviceIdAndVersion(update, deviceId, latest.getVersion()) > 0) {
+                deleteDeviceShadowCache(deviceId);
+                return update;
+            }
+        }
+        return null;
+    }
+
+    /**
+     * 乐观锁清空影子：desired、reported、metadata 置空，version 归零
+     *
+     * @param deviceId 设备编号
+     * @return 是否成功；影子不存在时，无需清空，也返回 true；版本冲突重试耗尽时，返回 false
+     */
+    private boolean resetDeviceShadow0(Long deviceId) {
+        for (int i = 0; i < OPTIMISTIC_LOCK_RETRY; i++) {
+            IotDeviceShadowDO latest = deviceShadowMapper.selectByDeviceId(deviceId);
+            if (latest == null) {
+                return true;
+            }
+            IotDeviceShadowDO update = new IotDeviceShadowDO()
+                    .setDesired(new HashMap<>()).setReported(new HashMap<>()).setMetadata(new HashMap<>())
+                    .setVersion(0L).setShadowTime(DateUtil.currentSeconds());
+            if (deviceShadowMapper.updateByDeviceIdAndVersion(update, deviceId, latest.getVersion()) > 0) {
+                deleteDeviceShadowCache(deviceId);
+                return true;
+            }
+        }
+        return false;
+    }
+
+    /**
+     * 删除影子缓存：存在事务时，在提交后删除，避免并发查询回填提交前的旧影子
+     *
+     * @param deviceId 设备编号
+     */
+    private void deleteDeviceShadowCache(Long deviceId) {
+        if (!TransactionSynchronizationManager.isSynchronizationActive()) {
+            deviceShadowRedisDAO.delete(deviceId);
+            return;
+        }
+        TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
+
+            @Override
+            public void afterCommit() {
+                deviceShadowRedisDAO.delete(deviceId);
+            }
+
+        });
+    }
+
+    private void sendShadowGetReply(IotDeviceDO device, String serverId, String requestId, IotDeviceShadowDO shadow) {
+        sendShadowReply(device, serverId, requestId, SUCCESS.getCode(), SUCCESS.getMsg(),
+                IotDeviceShadowProtocolUtils.buildGetData(shadow.getDesired(), shadow.getReported(),
+                        shadow.getVersion(), shadow.getShadowTime()));
+    }
+
+    /**
+     * 回复版本冲突：带上 DB 当前版本，便于设备重新同步
+     */
+    private void sendShadowVersionConflictReply(IotDeviceDO device, IotDeviceMessage message) {
+        IotDeviceShadowDO latest = deviceShadowMapper.selectByDeviceId(device.getId());
+        sendShadowErrorReply(device, message, IotDeviceShadowErrorCodeEnum.VERSION_CONFLICT,
+                latest != null ? latest.getVersion() : 0L);
+    }
+
+    /**
+     * 回复错误：data 仍带 version，便于设备重新同步
+     */
+    private void sendShadowErrorReply(IotDeviceDO device, IotDeviceMessage message,
+                                      IotDeviceShadowErrorCodeEnum errorCode, Long version) {
+        sendShadowReply(device, message.getServerId(), message.getRequestId(),
+                errorCode.getCode(), errorCode.getMessage(),
+                IotDeviceShadowProtocolUtils.buildUpdateData(version, DateUtil.currentSeconds()));
+    }
+
+    /**
+     * 下发影子回复（标准 IotDeviceMessage 封套）
+     * <p>
+     * method 使用 {@link IotDeviceMessageMethodEnum#SHADOW_UPDATE}，带 code 即为回复，
+     * 网关下发到 thing/shadow/update_reply Topic；发送失败只打印日志，不影响上行处理
+     */
+    private void sendShadowReply(IotDeviceDO device, String serverId, String requestId,
+                                 Integer code, String msg, Map<String, Object> data) {
+        IotDeviceMessage reply = IotDeviceMessage.replyOf(requestId,
+                IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), data, code, msg);
+        reply.setDeviceId(device.getId()).setTenantId(device.getTenantId()).setServerId(serverId);
+        try {
+            deviceMessageService.sendDeviceMessage(reply, device, serverId);
+        } catch (Exception ex) {
+            log.error("[sendShadowReply][device({}) 下发影子回复失败]", device.getId(), ex);
+        }
+    }
+
+    private static Map<String, Object> mergeMap(Map<String, Object> oldMap, Map<String, Object> newMap) {
+        Map<String, Object> result = oldMap != null ? new HashMap<>(oldMap) : new HashMap<>();
+        if (CollUtil.isNotEmpty(newMap)) {
+            result.putAll(newMap);
+        }
+        return result;
+    }
+
+}
+package cn.iocoder.yudao.module.iot.service.device;
+
+import cn.iocoder.yudao.framework.test.core.ut.BaseDbUnitTest;
+import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceDO;
+import cn.iocoder.yudao.module.iot.dal.mysql.device.IotDeviceMapper;
+import cn.iocoder.yudao.module.iot.enums.product.IotProductDeviceTypeEnum;
+import cn.iocoder.yudao.module.iot.service.device.message.IotDeviceMessageService;
+import cn.iocoder.yudao.module.iot.service.device.shadow.IotDeviceShadowService;
+import cn.iocoder.yudao.module.iot.service.product.IotProductService;
+import jakarta.annotation.Resource;
+import org.junit.jupiter.api.Test;
+import org.springframework.context.annotation.Import;
+import org.springframework.test.context.bean.override.mockito.MockitoBean;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.verify;
+
+/**
+ * {@link IotDeviceServiceImpl} 的单元测试
+ *
+ * @author 芋道源码
+ */
+@Import(IotDeviceServiceImpl.class)
+public class IotDeviceServiceImplTest extends BaseDbUnitTest {
+
+    @Resource
+    private IotDeviceServiceImpl deviceService;
+
+    @Resource
+    private IotDeviceMapper deviceMapper;
+
+    @MockitoBean
+    private IotProductService productService;
+    @MockitoBean
+    private IotDeviceGroupService deviceGroupService;
+    @MockitoBean
+    private IotDeviceMessageService deviceMessageService;
+    @MockitoBean
+    private IotDeviceShadowService deviceShadowService;
+
+    @Test
+    public void testDeleteDevice_success() {
+        // mock 数据
+        IotDeviceDO dbDevice = buildDevice();
+        deviceMapper.insert(dbDevice);
+        // 准备参数
+        Long id = dbDevice.getId();
+
+        // 调用
+        deviceService.deleteDevice(id);
+        // 断言
+        assertNull(deviceMapper.selectById(id));
+        verify(deviceShadowService).deleteByDeviceId(id);
+    }
+
+    @Test
+    public void testDeleteDevice_shadowDeleteFailed() {
+        // mock 数据
+        IotDeviceDO dbDevice = buildDevice();
+        deviceMapper.insert(dbDevice);
+        // 准备参数
+        Long id = dbDevice.getId();
+        // mock 影子删除失败
+        RuntimeException failure = new RuntimeException("shadow delete failed");
+        doThrow(failure).when(deviceShadowService).deleteByDeviceId(id);
+
+        // 调用，并断言异常
+        assertSame(failure, assertThrows(RuntimeException.class, () -> deviceService.deleteDevice(id)));
+        // 断言：设备删除已回滚
+        assertNotNull(deviceMapper.selectById(id));
+    }
+
+    private static IotDeviceDO buildDevice() {
+        return IotDeviceDO.builder().deviceName("testDevice").productId(10L).productKey("testProduct")
+                .deviceType(IotProductDeviceTypeEnum.DIRECT.getType()).build();
+    }
+
+}
+package cn.iocoder.yudao.module.iot.service.device.shadow;
+
+import cn.iocoder.yudao.framework.common.enums.CommonStatusEnum;
+import cn.iocoder.yudao.framework.test.core.ut.BaseMockitoUnitTest;
+import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
+import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
+import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceDO;
+import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceShadowDO;
+import cn.iocoder.yudao.module.iot.dal.mysql.device.IotDeviceShadowMapper;
+import cn.iocoder.yudao.module.iot.dal.redis.device.DeviceShadowRedisDAO;
+import cn.iocoder.yudao.module.iot.service.device.IotDeviceService;
+import cn.iocoder.yudao.module.iot.service.device.message.IotDeviceMessageService;
+import cn.iocoder.yudao.module.iot.service.device.property.IotDevicePropertyService;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.springframework.dao.DuplicateKeyException;
+import org.springframework.transaction.support.TransactionSynchronization;
+import org.springframework.transaction.support.TransactionSynchronizationManager;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static cn.iocoder.yudao.framework.test.core.util.AssertUtils.assertServiceException;
+import static cn.iocoder.yudao.module.iot.enums.ErrorCodeConstants.*;
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.ArgumentMatchers.*;
+import static org.mockito.Mockito.*;
+
+/**
+ * {@link IotDeviceShadowServiceImpl} 的单元测试
+ *
+ * @author 芋道源码
+ */
+public class IotDeviceShadowServiceImplTest extends BaseMockitoUnitTest {
+
+    @InjectMocks
+    private IotDeviceShadowServiceImpl deviceShadowService;
+
+    @Mock
+    private IotDeviceShadowMapper deviceShadowMapper;
+    @Mock
+    private DeviceShadowRedisDAO deviceShadowRedisDAO;
+    @Mock
+    private IotDeviceService deviceService;
+    @Mock
+    private IotDevicePropertyService devicePropertyService;
+    @Mock
+    private IotDeviceMessageService deviceMessageService;
+
+    // ========== getOrCreateShadow ==========
+
+    @Test
+    public void testGetOrCreateShadow_cacheHit() {
+        // mock 数据
+        IotDeviceDO device = buildDevice();
+        IotDeviceShadowDO cacheShadow = buildShadow(device, 2L);
+        // mock 方法
+        when(deviceShadowRedisDAO.get(device.getId())).thenReturn(cacheShadow);
+
+        // 调用
+        IotDeviceShadowDO shadow = deviceShadowService.getOrCreateShadow(device);
+        // 断言
+        assertSame(cacheShadow, shadow);
+        verifyNoInteractions(deviceShadowMapper);
+    }
+
+    @Test
+    public void testGetOrCreateShadow_notExists() {
+        // 准备参数
+        IotDeviceDO device = buildDevice();
+        // mock 方法
+        when(deviceShadowRedisDAO.get(device.getId())).thenReturn(null);
+        when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(null);
+
+        // 调用
+        IotDeviceShadowDO shadow = deviceShadowService.getOrCreateShadow(device);
+        // 断言：插入启用的空影子，并回填缓存
+        verify(deviceShadowMapper).insert(shadow);
+        assertEquals(device.getId(), shadow.getDeviceId());
+        assertEquals(CommonStatusEnum.ENABLE.getStatus(), shadow.getStatus());
+        assertEquals(0L, shadow.getVersion());
+        assertTrue(shadow.getReported().isEmpty());
+        verify(deviceShadowRedisDAO).set(shadow);
+    }
+
+    // ========== handleDeviceShadowUpdate：get ==========
+
+    @Test
+    public void testHandleDeviceShadowUpdate_get() {
+        // mock 数据
+        IotDeviceDO device = buildDevice();
+        IotDeviceShadowDO dbShadow = buildShadow(device, 2L);
+        dbShadow.getDesired().put("PowerSwitch", 1);
+        // mock 方法
+        when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(dbShadow);
+        // 准备参数
+        IotDeviceMessage message = buildMessage(device, Map.of("method", "get"));
+
+        // 调用
+        deviceShadowService.handleDeviceShadowUpdate(device, message);
+        // 断言：读 DB 回复全量影子
+        IotDeviceMessage reply = captureReply(device);
+        assertEquals(IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), reply.getMethod());
+        assertEquals("req-1", reply.getRequestId());
+        assertEquals(0, reply.getCode());
+        assertEquals(2L, getReplyVersion(reply));
+        assertEquals(Map.of("PowerSwitch", 1), getReplyState(reply).get("desired"));
+        verify(deviceShadowRedisDAO, never()).get(any());
+    }
+
+    @Test
+    public void testHandleDeviceShadowUpdate_disabledInDb() {
+        // mock 数据：缓存为启用的旧影子，DB 已关闭
+        IotDeviceDO device = buildDevice();
+        lenient().when(deviceShadowRedisDAO.get(device.getId())).thenReturn(buildShadow(device, 0L));
+        when(deviceShadowMapper.selectByDeviceId(device.getId()))
+                .thenReturn(buildShadow(device, 0L, CommonStatusEnum.DISABLE.getStatus()));
+        // 准备参数
+        IotDeviceMessage message = buildMessage(device, buildUpdateParams(Map.of("Temperature", 25), null));
+
+        // 调用
+        deviceShadowService.handleDeviceShadowUpdate(device, message);
+        // 断言：以 DB 为准，忽略上行
+        verify(deviceShadowMapper, never()).updateByDeviceIdAndVersion(any(), any(), any());
+        verify(deviceMessageService, never()).sendDeviceMessage(any(), any(), any());
+    }
+
+    // ========== handleDeviceShadowUpdate：update ==========
+
+    @Test
+    public void testHandleDeviceShadowUpdate_update_success() {
+        // mock 数据
+        IotDeviceDO device = buildDevice();
+        IotDeviceShadowDO dbShadow = buildShadow(device, 1L);
+        dbShadow.getReported().put("Temperature", 20);
+        dbShadow.getDesired().put("PowerSwitch", 1);
+        // mock 方法
+        when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(dbShadow);
+        when(deviceShadowMapper.updateByDeviceIdAndVersion(any(), eq(device.getId()), eq(1L))).thenReturn(1);
+        // 准备参数
+        IotDeviceMessage message = buildMessage(device,
+                buildUpdateParams(Map.of("Temperature", 25, "Humidity", 60), 1L));
+
+        // 调用
+        deviceShadowService.handleDeviceShadowUpdate(device, message);
+        // 断言：合并 reported，保留 desired，version + 1
+        IotDeviceShadowDO update = captureUpdate(device, 1L);
+        assertEquals(Map.of("Temperature", 25, "Humidity", 60), update.getReported());
+        assertEquals(Map.of("PowerSwitch", 1), update.getDesired());
+        assertEquals(2L, update.getVersion());
+        verify(deviceShadowRedisDAO).delete(device.getId());
+        // 断言：回复新版本
+        IotDeviceMessage reply = captureReply(device);
+        assertEquals(0, reply.getCode());
+        assertEquals(2L, getReplyVersion(reply));
+    }
+
+    @Test
+    public void testHandleDeviceShadowUpdate_update_staleCacheUseDbVersion() {
+        // mock 数据：缓存为 v1 的旧影子，DB 已是 v2
+        IotDeviceDO device = buildDevice();
+        lenient().when(deviceShadowRedisDAO.get(device.getId())).thenReturn(buildShadow(device, 1L));
+        when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(buildShadow(device, 2L));
+        when(deviceShadowMapper.updateByDeviceIdAndVersion(any(), eq(device.getId()), eq(2L))).thenReturn(1);
+        // 准备参数：设备携带 DB 当前版本 v2
+        IotDeviceMessage message = buildMessage(device, buildUpdateParams(Map.of("Temperature", 25), 2L));
+
+        // 调用
+        deviceShadowService.handleDeviceShadowUpdate(device, message);
+        // 断言：按 DB 版本写入成功
+        assertEquals(3L, captureUpdate(device, 2L).getVersion());
+        assertEquals(0, captureReply(device).getCode());
+        verify(deviceShadowRedisDAO, never()).get(any());
+    }
+
+    @Test
+    public void testHandleDeviceShadowUpdate_update_versionConflict() {
+        // mock 数据
+        IotDeviceDO device = buildDevice();
+        // mock 方法
+        when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(buildShadow(device, 5L));
+        // 准备参数：设备携带旧版本 v3
+        IotDeviceMessage message = buildMessage(device, buildUpdateParams(Map.of("Temperature", 25), 3L));
+
+        // 调用
+        deviceShadowService.handleDeviceShadowUpdate(device, message);
+        // 断言：不写入，回复 409 及 DB 当前版本
+        verify(deviceShadowMapper, never()).updateByDeviceIdAndVersion(any(), any(), any());
+        IotDeviceMessage reply = captureReply(device);
+        assertEquals(409, reply.getCode());
+        assertEquals(5L, getReplyVersion(reply));
+    }
+
+    @Test
+    public void testHandleDeviceShadowUpdate_update_versionNull() {
+        // mock 数据
+        IotDeviceDO device = buildDevice();
+        // mock 方法
+        when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(buildShadow(device, 5L));
+        when(deviceShadowMapper.updateByDeviceIdAndVersion(any(), eq(device.getId()), eq(5L))).thenReturn(1);
+        // 准备参数：不传 version，强制更新
+        IotDeviceMessage message = buildMessage(device, buildUpdateParams(Map.of("Temperature", 25), null));
+
+        // 调用
+        deviceShadowService.handleDeviceShadowUpdate(device, message);
+        // 断言
+        assertEquals(6L, captureUpdate(device, 5L).getVersion());
+    }
+
+    @Test
+    public void testHandleDeviceShadowUpdate_update_retryFromLatest() {
+        // mock 数据：前两次读到 v1，之后读到 v2
+        IotDeviceDO device = buildDevice();
+        IotDeviceShadowDO staleShadow = buildShadow(device, 1L);
+        staleShadow.getReported().put("Temperature", 20);
+        IotDeviceShadowDO latestShadow = buildShadow(device, 2L);
+        latestShadow.getReported().put("Temperature", 30);
+        latestShadow.getReported().put("Humidity", 50);
+        // mock 方法：v1 写入冲突，v2 写入成功
+        when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(staleShadow, staleShadow, latestShadow);
+        when(deviceShadowMapper.updateByDeviceIdAndVersion(any(), eq(device.getId()), eq(1L))).thenReturn(0);
+        when(deviceShadowMapper.updateByDeviceIdAndVersion(any(), eq(device.getId()), eq(2L))).thenReturn(1);
+        // 准备参数
+        IotDeviceMessage message = buildMessage(device, buildUpdateParams(Map.of("Temperature", 25), null));
+
+        // 调用
+        deviceShadowService.handleDeviceShadowUpdate(device, message);
+        // 断言：重试时基于 v2 合并，保留 Humidity
+        IotDeviceShadowDO update = captureUpdate(device, 2L);
+        assertEquals(Map.of("Temperature", 25, "Humidity", 50), update.getReported());
+        assertEquals(3L, update.getVersion());
+        assertEquals(3L, getReplyVersion(captureReply(device)));
+    }
+
+    @Test
+    public void testHandleDeviceShadowUpdate_update_retryExhausted() {
+        // mock 数据
+        IotDeviceDO device = buildDevice();
+        // mock 方法：每次写入都冲突
+        when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(buildShadow(device, 4L));
+        when(deviceShadowMapper.updateByDeviceIdAndVersion(any(), eq(device.getId()), eq(4L))).thenReturn(0);
+        // 准备参数
+        IotDeviceMessage message = buildMessage(device, buildUpdateParams(Map.of("Temperature", 25), null));
+
+        // 调用
+        deviceShadowService.handleDeviceShadowUpdate(device, message);
+        // 断言：重试 3 次后，回复 409
+        verify(deviceShadowMapper, times(3)).updateByDeviceIdAndVersion(any(), eq(device.getId()), eq(4L));
+        verify(deviceShadowRedisDAO, never()).delete(any());
+        IotDeviceMessage reply = captureReply(device);
+        assertEquals(409, reply.getCode());
+        assertEquals(4L, getReplyVersion(reply));
+    }
+
+    @Test
+    public void testHandleDeviceShadowUpdate_update_propertyCountExceeded() {
+        // mock 数据：已有 128 个属性
+        IotDeviceDO device = buildDevice();
+        IotDeviceShadowDO dbShadow = buildShadow(device, 1L);
+        for (int i = 0; i < 128; i++) {
+            dbShadow.getReported().put("p" + i, i);
+        }
+        // mock 方法
+        when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(dbShadow);
+        // 准备参数
+        IotDeviceMessage message = buildMessage(device, buildUpdateParams(Map.of("Temperature", 25), 1L));
+
+        // 调用
+        deviceShadowService.handleDeviceShadowUpdate(device, message);
+        // 断言：不写入，回复 408
+        verify(deviceShadowMapper, never()).updateByDeviceIdAndVersion(any(), any(), any());
+        IotDeviceMessage reply = captureReply(device);
+        assertEquals(408, reply.getCode());
+        assertEquals(1L, getReplyVersion(reply));
+    }
+
+    @Test
+    public void testHandleDeviceShadowUpdate_resetWithoutState() {
+        // mock 数据
+        IotDeviceDO device = buildDevice();
+        // mock 方法
+        when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(buildShadow(device, 3L));
+        when(deviceShadowMapper.updateByDeviceIdAndVersion(any(), eq(device.getId()), eq(3L))).thenReturn(1);
+        // 准备参数：version 为 -1，且不传 state
+        IotDeviceMessage message = buildMessage(device, buildResetParams());
+
+        // 调用
+        deviceShadowService.handleDeviceShadowUpdate(device, message);
+        // 断言：按 DB 版本清空影子，version 归零
+        IotDeviceShadowDO update = captureUpdate(device, 3L);
+        assertTrue(update.getDesired().isEmpty());
+        assertTrue(update.getReported().isEmpty());
+        assertTrue(update.getMetadata().isEmpty());
+        assertEquals(0L, update.getVersion());
+        verify(deviceShadowRedisDAO).delete(device.getId());
+        // 断言：回复成功
+        IotDeviceMessage reply = captureReply(device);
+        assertEquals(0, reply.getCode());
+        assertEquals(0L, getReplyVersion(reply));
+    }
+
+    @Test
+    public void testHandleDeviceShadowUpdate_resetRetryExhausted() {
+        // mock 数据
+        IotDeviceDO device = buildDevice();
+        // mock 方法：每次清空都冲突
+        when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(buildShadow(device, 3L));
+        when(deviceShadowMapper.updateByDeviceIdAndVersion(any(), eq(device.getId()), eq(3L))).thenReturn(0);
+        // 准备参数
+        IotDeviceMessage message = buildMessage(device, buildResetParams());
+
+        // 调用
+        deviceShadowService.handleDeviceShadowUpdate(device, message);
+        // 断言：重试 3 次后，回复 409 及 DB 当前版本
+        verify(deviceShadowMapper, times(3)).updateByDeviceIdAndVersion(any(), eq(device.getId()), eq(3L));
+        verify(deviceShadowRedisDAO, never()).delete(any());
+        IotDeviceMessage reply = captureReply(device);
+        assertEquals(409, reply.getCode());
+        assertEquals(3L, getReplyVersion(reply));
+    }
+
+    @Test
+    public void testHandleDeviceShadowUpdate_stateMissing() {
+        // mock 数据
+        IotDeviceDO device = buildDevice();
+        // mock 方法
+        when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(buildShadow(device, 3L));
+        // 准备参数
+        IotDeviceMessage message = buildMessage(device, Map.of("method", "update", "version", 3L));
+
+        // 调用
+        deviceShadowService.handleDeviceShadowUpdate(device, message);
+        // 断言
+        IotDeviceMessage reply = captureReply(device);
+        assertEquals(402, reply.getCode());
+        assertEquals(3L, getReplyVersion(reply));
+    }
+
+    @Test
+    public void testHandleDeviceShadowUpdate_invalidParams() {
+        // 准备参数
+        IotDeviceDO device = buildDevice();
+        IotDeviceMessage message = buildMessage(device, null);
+
+        // 调用
+        deviceShadowService.handleDeviceShadowUpdate(device, message);
+        // 断言
+        assertEquals(400, captureReply(device).getCode());
+    }
+
+    @Test
+    public void testHandleDeviceShadowUpdate_methodMissing() {
+        // mock 数据
+        IotDeviceDO device = buildDevice();
+        // mock 方法
+        when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(buildShadow(device, 3L));
+        // 准备参数
+        IotDeviceMessage message = buildMessage(device, Map.of("state", Map.of("reported", Map.of("Temperature", 25))));
+
+        // 调用
+        deviceShadowService.handleDeviceShadowUpdate(device, message);
+        // 断言
+        assertEquals(401, captureReply(device).getCode());
+    }
+
+    @Test
+    public void testHandleDeviceShadowUpdate_dbFailure() {
+        // mock 数据
+        IotDeviceDO device = buildDevice();
+        // mock 方法
+        when(deviceShadowMapper.selectByDeviceId(device.getId())).thenThrow(new RuntimeException("DB down"));
+        // 准备参数
+        IotDeviceMessage message = buildMessage(device, Map.of("method", "get"));
+
+        // 调用
+        assertDoesNotThrow(() -> deviceShadowService.handleDeviceShadowUpdate(device, message));
+        // 断言：仍回复 500
+        IotDeviceMessage reply = captureReply(device);
+        assertEquals(500, reply.getCode());
+        assertEquals(0L, getReplyVersion(reply));
+    }
+
+    @Test
+    public void testHandleDeviceShadowUpdate_shadowDeletedDuringUpdate() {
+        // mock 数据
+        IotDeviceDO device = buildDevice();
+        // mock 方法：校验时存在，更新时已被删除
+        when(deviceShadowMapper.selectByDeviceId(device.getId()))
+                .thenReturn(buildShadow(device, 2L), (IotDeviceShadowDO) null);
+        // 准备参数
+        IotDeviceMessage message = buildMessage(device, buildUpdateParams(Map.of("Temperature", 25), 2L));
+
+        // 调用
+        deviceShadowService.handleDeviceShadowUpdate(device, message);
+        // 断言
+        verify(deviceShadowMapper, never()).updateByDeviceIdAndVersion(any(), any(), any());
+        assertEquals(500, captureReply(device).getCode());
+    }
+
+    // ========== updateDesiredState ==========
+
+    @Test
+    public void testUpdateDesiredState_online() {
+        // mock 数据：缓存为关闭的旧影子，DB 已启用
+        IotDeviceDO device = buildDevice();
+        IotDeviceShadowDO dbShadow = buildShadow(device, 1L);
+        dbShadow.getReported().put("Temperature", 20);
+        lenient().when(deviceShadowRedisDAO.get(device.getId()))
+                .thenReturn(buildShadow(device, 1L, CommonStatusEnum.DISABLE.getStatus()));
+        when(deviceService.validateDeviceExists(device.getId())).thenReturn(device);
+        when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(dbShadow);
+        when(deviceShadowMapper.updateByDeviceIdAndVersion(any(), eq(device.getId()), eq(1L))).thenReturn(1);
+        when(devicePropertyService.getDeviceServerId(device.getId())).thenReturn("server-1");
+        // 准备参数：数值、结构体、数组保持原类型
+        Map<String, Object> desired = Map.of("Temperature", 25, "Location", Map.of("lat", 31.2),
+                "Modes", List.of(1, 2));
+
+        // 调用
+        deviceShadowService.updateDesiredState(device.getId(), desired);
+        // 断言：合并 desired，保留 reported
+        IotDeviceShadowDO update = captureUpdate(device, 1L);
+        assertEquals(desired, update.getDesired());
+        assertEquals(Map.of("Temperature", 20), update.getReported());
+        assertEquals(2L, update.getVersion());
+        verify(deviceShadowRedisDAO).delete(device.getId());
+        // 断言：推送全量影子，desired 类型不变
+        IotDeviceMessage push = captureReply(device);
+        assertEquals(IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), push.getMethod());
+        assertEquals(0, push.getCode());
+        assertEquals(2L, getReplyVersion(push));
+        assertEquals(desired, getReplyState(push).get("desired"));
+        assertEquals(Map.of("Temperature", 20), getReplyState(push).get("reported"));
+    }
+
+    @Test
+    public void testUpdateDesiredState_offline() {
+        // mock 数据
+        IotDeviceDO device = buildDevice();
+        // mock 方法
+        when(deviceService.validateDeviceExists(device.getId())).thenReturn(device);
+        when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(buildShadow(device, 0L));
+        when(deviceShadowMapper.updateByDeviceIdAndVersion(any(), eq(device.getId()), eq(0L))).thenReturn(1);
+        when(devicePropertyService.getDeviceServerId(device.getId())).thenReturn(null);
+
+        // 调用
+        deviceShadowService.updateDesiredState(device.getId(), Map.of("PowerSwitch", 1));
+        // 断言
+        verify(deviceMessageService, never()).sendDeviceMessage(any(), any(), any());
+    }
+
+    @Test
+    public void testUpdateDesiredState_disabledInDb() {
+        // mock 数据：缓存为启用的旧影子，DB 已关闭
+        IotDeviceDO device = buildDevice();
+        lenient().when(deviceShadowRedisDAO.get(device.getId())).thenReturn(buildShadow(device, 0L));
+        when(deviceService.validateDeviceExists(device.getId())).thenReturn(device);
+        when(deviceShadowMapper.selectByDeviceId(device.getId()))
+                .thenReturn(buildShadow(device, 0L, CommonStatusEnum.DISABLE.getStatus()));
+
+        // 调用，并断言异常
+        assertServiceException(() -> deviceShadowService.updateDesiredState(device.getId(), Map.of("PowerSwitch", 1)),
+                DEVICE_SHADOW_NOT_ENABLED);
+        verify(deviceShadowMapper, never()).updateByDeviceIdAndVersion(any(), any(), any());
+    }
+
+    @Test
+    public void testUpdateDesiredState_propertyCountExceeded() {
+        // mock 数据：已有 128 个属性
+        IotDeviceDO device = buildDevice();
+        IotDeviceShadowDO dbShadow = buildShadow(device, 1L);
+        for (int i = 0; i < 128; i++) {
+            dbShadow.getDesired().put("p" + i, i);
+        }
+        // mock 方法
+        when(deviceService.validateDeviceExists(device.getId())).thenReturn(device);
+        when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(dbShadow);
+
+        // 调用，并断言异常
+        assertServiceException(() -> deviceShadowService.updateDesiredState(device.getId(), Map.of("PowerSwitch", 1)),
+                DEVICE_SHADOW_PROPERTY_COUNT_EXCEEDED);
+        verify(deviceShadowMapper, never()).updateByDeviceIdAndVersion(any(), any(), any());
+    }
+
+    @Test
+    public void testUpdateDesiredState_retryExhausted() {
+        // mock 数据
+        IotDeviceDO device = buildDevice();
+        // mock 方法：每次写入都冲突
+        when(deviceService.validateDeviceExists(device.getId())).thenReturn(device);
+        when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(buildShadow(device, 1L));
+        when(deviceShadowMapper.updateByDeviceIdAndVersion(any(), eq(device.getId()), eq(1L))).thenReturn(0);
+
+        // 调用，并断言异常
+        assertServiceException(() -> deviceShadowService.updateDesiredState(device.getId(), Map.of("PowerSwitch", 1)),
+                DEVICE_SHADOW_VERSION_CONFLICT);
+        verify(deviceShadowMapper, times(3)).updateByDeviceIdAndVersion(any(), eq(device.getId()), eq(1L));
+        verify(deviceMessageService, never()).sendDeviceMessage(any(), any(), any());
+    }
+
+    // ========== resetShadow ==========
+
+    @Test
+    public void testResetShadow() {
+        // mock 数据
+        IotDeviceDO device = buildDevice();
+        // mock 方法
+        when(deviceService.validateDeviceExists(device.getId())).thenReturn(device);
+        when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(buildShadow(device, 2L));
+        when(deviceShadowMapper.updateByDeviceIdAndVersion(any(), eq(device.getId()), eq(2L))).thenReturn(1);
+
+        // 调用
+        deviceShadowService.resetShadow(device.getId());
+        // 断言
+        IotDeviceShadowDO update = captureUpdate(device, 2L);
+        assertTrue(update.getDesired().isEmpty());
+        assertTrue(update.getReported().isEmpty());
+        assertEquals(0L, update.getVersion());
+        verify(deviceShadowRedisDAO).delete(device.getId());
+    }
+
+    @Test
+    public void testResetShadow_notExists() {
+        // mock 数据
+        IotDeviceDO device = buildDevice();
+        // mock 方法
+        when(deviceService.validateDeviceExists(device.getId())).thenReturn(device);
+        when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(null);
+
+        // 调用
+        deviceShadowService.resetShadow(device.getId());
+        // 断言：影子不存在，无需清空
+        verify(deviceShadowMapper, never()).updateByDeviceIdAndVersion(any(), any(), any());
+        verify(deviceShadowRedisDAO, never()).delete(any());
+    }
+
+    @Test
+    public void testResetShadow_retryExhausted() {
+        // mock 数据
+        IotDeviceDO device = buildDevice();
+        // mock 方法：每次清空都冲突
+        when(deviceService.validateDeviceExists(device.getId())).thenReturn(device);
+        when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(buildShadow(device, 2L));
+        when(deviceShadowMapper.updateByDeviceIdAndVersion(any(), eq(device.getId()), eq(2L))).thenReturn(0);
+
+        // 调用，并断言异常
+        assertServiceException(() -> deviceShadowService.resetShadow(device.getId()), DEVICE_SHADOW_VERSION_CONFLICT);
+        verify(deviceShadowMapper, times(3)).updateByDeviceIdAndVersion(any(), eq(device.getId()), eq(2L));
+        verify(deviceShadowRedisDAO, never()).delete(any());
+    }
+
+    // ========== updateStatus ==========
+
+    @Test
+    public void testUpdateStatus_shadowMissing() {
+        // mock 数据
+        IotDeviceDO device = buildDevice();
+        // mock 方法
+        when(deviceService.validateDeviceExists(device.getId())).thenReturn(device);
+        when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(null);
+
+        // 调用
+        deviceShadowService.updateStatus(device.getId(), CommonStatusEnum.DISABLE.getStatus());
+        // 断言：直接按目标状态创建
+        ArgumentCaptor<IotDeviceShadowDO> insertCaptor = ArgumentCaptor.forClass(IotDeviceShadowDO.class);
+        verify(deviceShadowMapper).insert(insertCaptor.capture());
+        assertEquals(CommonStatusEnum.DISABLE.getStatus(), insertCaptor.getValue().getStatus());
+        verify(deviceShadowMapper, never()).updateByDeviceId(any(), any());
+    }
+
+    @Test
+    public void testUpdateStatus_duplicateKey() {
+        // mock 数据
+        IotDeviceDO device = buildDevice();
+        // mock 方法：并发创建，插入冲突后读到已启用的影子
+        when(deviceService.validateDeviceExists(device.getId())).thenReturn(device);
+        when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(null, buildShadow(device, 0L));
+        when(deviceShadowMapper.insert(any(IotDeviceShadowDO.class))).thenThrow(new DuplicateKeyException("duplicate"));
+
+        // 调用
+        deviceShadowService.updateStatus(device.getId(), CommonStatusEnum.DISABLE.getStatus());
+        // 断言：改为更新状态，并删除缓存
+        ArgumentCaptor<IotDeviceShadowDO> updateCaptor = ArgumentCaptor.forClass(IotDeviceShadowDO.class);
+        verify(deviceShadowMapper).updateByDeviceId(updateCaptor.capture(), eq(device.getId()));
+        assertEquals(CommonStatusEnum.DISABLE.getStatus(), updateCaptor.getValue().getStatus());
+        assertNull(updateCaptor.getValue().getVersion());
+        verify(deviceShadowRedisDAO).delete(device.getId());
+    }
+
+    @Test
+    public void testUpdateStatus_inTransaction() {
+        // mock 数据
+        IotDeviceDO device = buildDevice();
+        // mock 方法
+        when(deviceService.validateDeviceExists(device.getId())).thenReturn(device);
+        when(deviceShadowMapper.selectByDeviceId(device.getId())).thenReturn(buildShadow(device, 0L));
+        // mock 事务
+        TransactionSynchronizationManager.initSynchronization();
+        try {
+            // 调用
+            deviceShadowService.updateStatus(device.getId(), CommonStatusEnum.DISABLE.getStatus());
+            // 断言：提交前不删除缓存
+            verify(deviceShadowMapper).updateByDeviceId(any(), eq(device.getId()));
+            verify(deviceShadowRedisDAO, never()).delete(any());
+            // 断言：提交后删除缓存
+            TransactionSynchronizationManager.getSynchronizations().forEach(TransactionSynchronization::afterCommit);
+            verify(deviceShadowRedisDAO).delete(device.getId());
+        } finally {
+            TransactionSynchronizationManager.clearSynchronization();
+        }
+    }
+
+    // ========== deleteByDeviceId ==========
+
+    @Test
+    public void testDeleteByDeviceId_inTransaction() {
+        // 准备参数
+        Long deviceId = 2L;
+        // mock 事务
+        TransactionSynchronizationManager.initSynchronization();
+        try {
+            // 调用
+            deviceShadowService.deleteByDeviceId(deviceId);
+            // 断言：提交前不删除缓存
+            verify(deviceShadowMapper).deleteByDeviceId(deviceId);
+            verify(deviceShadowRedisDAO, never()).delete(any());
+            // 断言：回滚时不删除缓存
+            TransactionSynchronizationManager.getSynchronizations()
+                    .forEach(sync -> sync.afterCompletion(TransactionSynchronization.STATUS_ROLLED_BACK));
+            verify(deviceShadowRedisDAO, never()).delete(any());
+            // 断言：提交后删除缓存
+            TransactionSynchronizationManager.getSynchronizations().forEach(TransactionSynchronization::afterCommit);
+            verify(deviceShadowRedisDAO).delete(deviceId);
+        } finally {
+            TransactionSynchronizationManager.clearSynchronization();
+        }
+    }
+
+    // ========== 私有方法 ==========
+
+    private IotDeviceDO buildDevice() {
+        return IotDeviceDO.builder().id(2L).productId(10L).productKey("testProduct").deviceName("testDevice").build();
+    }
+
+    private IotDeviceShadowDO buildShadow(IotDeviceDO device, Long version) {
+        return buildShadow(device, version, CommonStatusEnum.ENABLE.getStatus());
+    }
+
+    private IotDeviceShadowDO buildShadow(IotDeviceDO device, Long version, Integer status) {
+        return new IotDeviceShadowDO().setId(100L).setDeviceId(device.getId()).setProductId(device.getProductId())
+                .setProductKey(device.getProductKey()).setDeviceName(device.getDeviceName())
+                .setDesired(new HashMap<>()).setReported(new HashMap<>()).setMetadata(new HashMap<>())
+                .setVersion(version).setShadowTime(1000L).setStatus(status);
+    }
+
+    private static Map<String, Object> buildUpdateParams(Map<String, Object> reported, Long version) {
+        Map<String, Object> params = new HashMap<>();
+        params.put("method", "update");
+        params.put("state", Map.of("reported", reported));
+        params.put("version", version);
+        return params;
+    }
+
+    private static Map<String, Object> buildResetParams() {
+        Map<String, Object> params = new HashMap<>();
+        params.put("method", "update");
+        params.put("version", -1L);
+        return params;
+    }
+
+    private static IotDeviceMessage buildMessage(IotDeviceDO device, Object params) {
+        return IotDeviceMessage.requestOf(device.getId(), 1L, "server-1",
+                IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod(), params).setRequestId("req-1");
+    }
+
+    private IotDeviceShadowDO captureUpdate(IotDeviceDO device, Long expectVersion) {
+        ArgumentCaptor<IotDeviceShadowDO> updateCaptor = ArgumentCaptor.forClass(IotDeviceShadowDO.class);
+        verify(deviceShadowMapper).updateByDeviceIdAndVersion(updateCaptor.capture(), eq(device.getId()),
+                eq(expectVersion));
+        return updateCaptor.getValue();
+    }
+
+    private IotDeviceMessage captureReply(IotDeviceDO device) {
+        ArgumentCaptor<IotDeviceMessage> replyCaptor = ArgumentCaptor.forClass(IotDeviceMessage.class);
+        verify(deviceMessageService).sendDeviceMessage(replyCaptor.capture(), eq(device), eq("server-1"));
+        return replyCaptor.getValue();
+    }
+
+    @SuppressWarnings("unchecked")
+    private static Long getReplyVersion(IotDeviceMessage reply) {
+        Object version = ((Map<String, Object>) reply.getData()).get("version");
+        return version instanceof Number ? ((Number) version).longValue() : null;
+    }
+
+    @SuppressWarnings("unchecked")
+    private static Map<String, Object> getReplyState(IotDeviceMessage reply) {
+        return (Map<String, Object>) ((Map<String, Object>) reply.getData()).get("state");
+    }
+
+}
 DELETE FROM "iot_scene_rule";
 DELETE FROM "iot_product";
 DELETE FROM "iot_device";
+DELETE FROM "iot_device_shadow";
 DELETE FROM "iot_thing_model";
 DELETE FROM "iot_device_data";
 DELETE FROM "iot_alert_config";
@@ -189,6 +189,29 @@ CREATE TABLE IF NOT EXISTS &quot;iot_ota_record&quot; (
     PRIMARY KEY ("id")
 ) COMMENT 'IoT OTA 升级记录表';
+CREATE TABLE IF NOT EXISTS "iot_device_shadow"
+(
+    "id"          bigint       NOT NULL GENERATED BY DEFAULT AS IDENTITY,
+    "device_id"   bigint       NOT NULL,
+    "product_id"  bigint       NOT NULL,
+    "product_key" varchar(64)  NOT NULL DEFAULT '',
+    "device_name" varchar(255) NOT NULL DEFAULT '',
+    "reported"    varchar(10000)        DEFAULT NULL,
+    "desired"     varchar(10000)        DEFAULT NULL,
+    "metadata"    varchar(10000)        DEFAULT NULL,
+    "version"     bigint       NOT NULL DEFAULT 0,
+    "shadow_time" bigint       NOT NULL DEFAULT 0,
+    "status"      tinyint      NOT NULL DEFAULT 0,
+    "creator"     varchar(64)           DEFAULT '',
+    "create_time" timestamp    NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "updater"     varchar(64)           DEFAULT '',
+    "update_time" timestamp    NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "deleted"     bit          NOT NULL DEFAULT FALSE,
+    "tenant_id"   bigint       NOT NULL DEFAULT '0',
+    PRIMARY KEY ("id"),
+    UNIQUE ("device_id")
+) COMMENT 'IoT 设备影子表';
+
 CREATE TABLE IF NOT EXISTS "iot_data_rule" (
     "id" bigint NOT NULL GENERATED BY DEFAULT AS IDENTITY,
     "name" varchar(128) NOT NULL,
@@ -67,6 +67,10 @@ public enum IotDeviceMessageMethodEnum implements ArrayValuable&lt;String&gt; {
     OTA_UPGRADE("thing.ota.upgrade", "OTA 固件信息推送", false),
     OTA_PROGRESS("thing.ota.progress", "OTA 升级进度上报", true),
+    // ========== 设备影子 ==========
+
+    SHADOW_UPDATE("thing.shadow.update", "设备影子更新", true),
+
     ;
     public static final String[] ARRAYS = Arrays.stream(values()).map(IotDeviceMessageMethodEnum::getMethod)
@@ -77,7 +81,8 @@ public enum IotDeviceMessageMethodEnum implements ArrayValuable&lt;String&gt; {
      */
     public static final Set<String> REPLY_DISABLED = SetUtils.asSet(
             STATE_UPDATE.getMethod(),
-            OTA_PROGRESS.getMethod() // 参考阿里云，OTA 升级进度上报，不进行回复
+            OTA_PROGRESS.getMethod(), // 参考阿里云，OTA 升级进度上报，不进行回复
+            SHADOW_UPDATE.getMethod() // 设备影子由影子 Service 主动下发回复（thing/shadow/update_reply）
     );
     private final String method;
+package cn.iocoder.yudao.module.iot.core.enums.shadow;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+
+/**
+ * IoT 设备影子回复的错误码枚举
+ * <p>
+ * 用于设备影子回复（Topic：thing/shadow/update_reply）的 code、msg；
+ * 管理后台接口的业务异常，使用 biz 模块的 ErrorCodeConstants
+ *
+ * @author 芋道源码
+ */
+@Getter
+@AllArgsConstructor
+public enum IotDeviceShadowErrorCodeEnum {
+
+    JSON_FORMAT_ERROR(400, "JSON 格式错误"),
+    METHOD_MISSING(401, "缺少 method"),
+    STATE_MISSING(402, "缺少 state"),
+    REPORTED_MISSING(404, "缺少 reported"),
+    REPORTED_EMPTY(405, "reported 为空"),
+    INVALID_METHOD(406, "无效内层 method"),
+    PROPERTY_COUNT_EXCEEDED(408, "影子属性数超过限制"),
+    VERSION_CONFLICT(409, "影子版本冲突"),
+    SERVER_ERROR(500, "服务端异常");
+
+    private final Integer code;
+
+    private final String message;
+
+}
+package cn.iocoder.yudao.module.iot.core.topic.shadow;
+
+import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.util.Map;
+
+/**
+ * IoT 设备影子上行 Request DTO
+ * <p>
+ * 用于 {@link IotDeviceMessageMethodEnum#SHADOW_UPDATE} 标准封套的 params 参数
+ *
+ * @author 芋道源码
+ */
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+public class IotDeviceShadowUpdateReqDTO {
+
+    /**
+     * 内层方法：get / update
+     */
+    private String method;
+
+    /**
+     * 影子状态
+     */
+    private ShadowState state;
+
+    /**
+     * 设备已知版本号，可缺省；-1 表示清空影子
+     */
+    private Long version;
+
+    @Data
+    @NoArgsConstructor
+    @AllArgsConstructor
+    public static class ShadowState {
+
+        /**
+         * 上报状态
+         */
+        private Map<String, Object> reported;
+
+    }
+
+}
+package cn.iocoder.yudao.module.iot.core.util;
+
+import cn.hutool.core.map.MapUtil;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * IoT 设备影子协议工具
+ * <p>
+ * 内层 method、state 键名、reply data 字段等与设备侧标准封套 params/data 对齐；并提供 metadata 读写。
+ *
+ * @author 芋道源码
+ */
+public final class IotDeviceShadowProtocolUtils {
+
+    private IotDeviceShadowProtocolUtils() {
+    }
+
+    // ========== 内层 method（上行 params.method） ==========
+
+    public static final String INNER_METHOD_GET = "get";
+    public static final String INNER_METHOD_UPDATE = "update";
+
+    // ========== state / metadata 键 ==========
+
+    public static final String STATE_DESIRED = "desired";
+    public static final String STATE_REPORTED = "reported";
+    public static final String METADATA_TIMESTAMP = "timestamp";
+
+    // ========== reply data 键 ==========
+
+    public static final String PAYLOAD_STATE = "state";
+    public static final String PAYLOAD_VERSION = "version";
+
+    /**
+     * 构建 get 响应 data：{ state:{desired,reported}, version, timestamp }
+     */
+    public static Map<String, Object> buildGetData(Map<String, Object> desired, Map<String, Object> reported,
+                                                   Long version, Long timestamp) {
+        Map<String, Object> state = new HashMap<>();
+        state.put(STATE_DESIRED, desired != null ? desired : new HashMap<>());
+        state.put(STATE_REPORTED, reported != null ? reported : new HashMap<>());
+        Map<String, Object> data = new HashMap<>();
+        data.put(PAYLOAD_STATE, state);
+        data.put(PAYLOAD_VERSION, version);
+        data.put(METADATA_TIMESTAMP, timestamp);
+        return data;
+    }
+
+    /**
+     * 构建 update 响应 / 错误 data：{ version, timestamp }
+     */
+    public static Map<String, Object> buildUpdateData(Long version, Long timestamp) {
+        Map<String, Object> data = new HashMap<>();
+        data.put(PAYLOAD_VERSION, version);
+        data.put(METADATA_TIMESTAMP, timestamp);
+        return data;
+    }
+
+    /**
+     * 获取 desired 下某属性的更新时间戳
+     *
+     * @param metadata   元数据
+     * @param identifier 属性标识符
+     * @return 时间戳（epoch 秒），不存在返回 null
+     */
+    public static Long getDesiredTimestamp(Map<String, Object> metadata, String identifier) {
+        return getTimestamp(metadata, STATE_DESIRED, identifier);
+    }
+
+    /**
+     * 获取 reported 下某属性的更新时间戳
+     *
+     * @param metadata   元数据
+     * @param identifier 属性标识符
+     * @return 时间戳（epoch 秒），不存在返回 null
+     */
+    public static Long getReportedTimestamp(Map<String, Object> metadata, String identifier) {
+        return getTimestamp(metadata, STATE_REPORTED, identifier);
+    }
+
+    /**
+     * 写入 desired 下多个属性的更新时间戳（原地修改 metadata）
+     *
+     * @param metadata  元数据
+     * @param keys      属性标识符集合
+     * @param timestamp 时间戳（epoch 秒）
+     */
+    public static void putDesiredTimestamps(Map<String, Object> metadata, Set<String> keys, long timestamp) {
+        putTimestamps(metadata, STATE_DESIRED, keys, timestamp);
+    }
+
+    /**
+     * 写入 reported 下多个属性的更新时间戳（原地修改 metadata）
+     *
+     * @param metadata  元数据
+     * @param keys      属性标识符集合
+     * @param timestamp 时间戳（epoch 秒）
+     */
+    public static void putReportedTimestamps(Map<String, Object> metadata, Set<String> keys, long timestamp) {
+        putTimestamps(metadata, STATE_REPORTED, keys, timestamp);
+    }
+
+    @SuppressWarnings("unchecked")
+    private static Long getTimestamp(Map<String, Object> metadata, String type, String identifier) {
+        if (metadata == null) {
+            return null;
+        }
+        Object typeMeta = metadata.get(type);
+        if (!(typeMeta instanceof Map)) {
+            return null;
+        }
+        Object keyMeta = ((Map<String, Object>) typeMeta).get(identifier);
+        if (!(keyMeta instanceof Map)) {
+            return null;
+        }
+        Object timestamp = ((Map<String, Object>) keyMeta).get(METADATA_TIMESTAMP);
+        return timestamp instanceof Number ? ((Number) timestamp).longValue() : null;
+    }
+
+    @SuppressWarnings("unchecked")
+    private static void putTimestamps(Map<String, Object> metadata, String type, Set<String> keys, long timestamp) {
+        Map<String, Object> typeMeta = (Map<String, Object>) metadata.computeIfAbsent(type, k -> new HashMap<>());
+        for (String key : keys) {
+            typeMeta.put(key, MapUtil.of(METADATA_TIMESTAMP, timestamp));
+        }
+    }
+
+}
+package cn.iocoder.yudao.module.iot.gateway.util;
+
+import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * {@link IotMqttTopicUtils} 的单元测试
+ *
+ * @author 芋道源码
+ */
+public class IotMqttTopicUtilsTest {
+
+    @Test
+    public void testBuildTopicByMethod_shadowUpdateReply() {
+        // 准备参数
+        String method = IotDeviceMessageMethodEnum.SHADOW_UPDATE.getMethod();
+
+        // 调用
+        String topic = IotMqttTopicUtils.buildTopicByMethod(method, "pk", "dn", true);
+        // 断言
+        assertEquals("/sys/pk/dn/thing/shadow/update_reply", topic);
+    }
+
+    @Test
+    public void testBuildTopicByMethod_serviceInvokeReply() {
+        // 准备参数
+        String method = IotDeviceMessageMethodEnum.SERVICE_INVOKE.getMethod();
+
+        // 调用
+        String topic = IotMqttTopicUtils.buildTopicByMethod(method, "pk", "dn", true);
+        // 断言
+        assertEquals("/sys/pk/dn/thing/service/invoke_reply", topic);
+    }
+
+}