services;
+
+}
\ No newline at end of file
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/device/IotDeviceMessageDO.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/device/IotDeviceMessageDO.java
new file mode 100644
index 0000000000..9f1f6a6a0c
--- /dev/null
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/device/IotDeviceMessageDO.java
@@ -0,0 +1,109 @@
+package cn.iocoder.yudao.module.iot.dal.dataobject.device;
+
+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.thingmodel.IotThingModelDO;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/**
+ * IoT 设备消息数据 DO
+ *
+ * 目前使用 TDengine 存储
+ *
+ * @author alwayssuper
+ */
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class IotDeviceMessageDO {
+
+ /**
+ * 消息编号
+ */
+ private String id;
+ /**
+ * 上报时间戳
+ */
+ private Long reportTime;
+ /**
+ * 存储时间戳
+ */
+ private Long ts;
+
+ /**
+ * 设备编号
+ *
+ * 关联 {@link IotDeviceDO#getId()}
+ */
+ private Long deviceId;
+ /**
+ * 租户编号
+ */
+ private Long tenantId;
+
+ /**
+ * 服务编号，该消息由哪个 server 发送
+ */
+ private String serverId;
+
+ /**
+ * 是否上行消息
+ *
+ * 由 {@link cn.iocoder.yudao.module.iot.core.util.IotDeviceMessageUtils#isUpstreamMessage(IotDeviceMessage)} 计算。
+ * 计算并存储的目的：方便计算多少条上行、多少条下行
+ */
+ private Boolean upstream;
+ /**
+ * 是否回复消息
+ *
+ * 由 {@link cn.iocoder.yudao.module.iot.core.util.IotDeviceMessageUtils#isReplyMessage(IotDeviceMessage)} 计算。
+ * 计算并存储的目的：方便计算多少条请求、多少条回复
+ */
+ private Boolean reply;
+ /**
+ * 标识符
+ *
+ * 例如说：{@link IotThingModelDO#getIdentifier()}
+ * 目前，只有事件上报、服务调用才有！！！
+ */
+ private String identifier;
+
+ // ========== codec（编解码）字段 ==========
+
+ /**
+ * 请求编号
+ *
+ * 由设备生成，对应阿里云 IoT 的 Alink 协议中的 id、华为云 IoTDA 协议的 request_id
+ */
+ private String requestId;
+ /**
+ * 请求方法
+ *
+ * 枚举 {@link IotDeviceMessageMethodEnum}
+ * 例如说：thing.property.report 属性上报
+ */
+ private String method;
+ /**
+ * 请求参数
+ *
+ * 例如说：属性上报的 properties、事件上报的 params
+ */
+ private Object params;
+ /**
+ * 响应结果
+ */
+ private Object data;
+ /**
+ * 响应错误码
+ */
+ private Integer code;
+ /**
+ * 响应提示
+ */
+ private String msg;
+
+}
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/ota/IotOtaTaskDO.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/ota/IotOtaTaskDO.java
new file mode 100644
index 0000000000..4c9124b89f
--- /dev/null
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/ota/IotOtaTaskDO.java
@@ -0,0 +1,70 @@
+package cn.iocoder.yudao.module.iot.dal.dataobject.ota;
+
+import cn.iocoder.yudao.framework.mybatis.core.dataobject.BaseDO;
+import cn.iocoder.yudao.module.iot.enums.ota.IotOtaTaskDeviceScopeEnum;
+import cn.iocoder.yudao.module.iot.enums.ota.IotOtaTaskStatusEnum;
+import com.baomidou.mybatisplus.annotation.KeySequence;
+import com.baomidou.mybatisplus.annotation.TableId;
+import com.baomidou.mybatisplus.annotation.TableName;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/**
+ * IoT OTA 升级任务 DO
+ *
+ * @author 芋道源码
+ */
+@TableName(value = "iot_ota_task", autoResultMap = true)
+@KeySequence("iot_ota_task_seq") // 用于 Oracle、PostgreSQL、Kingbase、DB2、H2 数据库的主键自增。如果是 MySQL 等数据库，可不写。
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class IotOtaTaskDO extends BaseDO {
+
+ /**
+ * 任务编号
+ */
+ @TableId
+ private Long id;
+ /**
+ * 任务名称
+ */
+ private String name;
+ /**
+ * 任务描述
+ */
+ private String description;
+
+ /**
+ * 固件编号
+ *
+ * 关联 {@link IotOtaFirmwareDO#getId()}
+ */
+ private Long firmwareId;
+
+ /**
+ * 任务状态
+ *
+ * 关联 {@link IotOtaTaskStatusEnum}
+ */
+ private Integer status;
+
+ /**
+ * 设备升级范围
+ *
+ * 关联 {@link IotOtaTaskDeviceScopeEnum}
+ */
+ private Integer deviceScope;
+ /**
+ * 设备总数数量
+ */
+ private Integer deviceTotalCount;
+ /**
+ * 设备成功数量
+ */
+ private Integer deviceSuccessCount;
+
+}
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/ota/IotOtaTaskRecordDO.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/ota/IotOtaTaskRecordDO.java
new file mode 100644
index 0000000000..d99a1bb60a
--- /dev/null
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/ota/IotOtaTaskRecordDO.java
@@ -0,0 +1,82 @@
+package cn.iocoder.yudao.module.iot.dal.dataobject.ota;
+
+import cn.iocoder.yudao.framework.mybatis.core.dataobject.BaseDO;
+import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceDO;
+import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceMessageDO;
+import cn.iocoder.yudao.module.iot.enums.ota.IotOtaTaskRecordStatusEnum;
+import com.baomidou.mybatisplus.annotation.KeySequence;
+import com.baomidou.mybatisplus.annotation.TableId;
+import com.baomidou.mybatisplus.annotation.TableName;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/**
+ * IoT OTA 升级任务记录 DO
+ *
+ * @author 芋道源码
+ */
+@TableName(value = "iot_ota_task_record", autoResultMap = true)
+@KeySequence("iot_ota_task_record_seq") // 用于 Oracle、PostgreSQL、Kingbase、DB2、H2 数据库的主键自增。如果是 MySQL 等数据库，可不写。
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class IotOtaTaskRecordDO extends BaseDO {
+
+ public static final String DESCRIPTION_CANCEL_BY_TASK = "管理员手动取消升级任务（批量）";
+
+ public static final String DESCRIPTION_CANCEL_BY_RECORD = "管理员手动取消升级记录（单个）";
+
+ /**
+ * 升级记录编号
+ */
+ @TableId
+ private Long id;
+
+ /**
+ * 固件编号
+ *
+ * 关联 {@link IotOtaFirmwareDO#getId()}
+ */
+ private Long firmwareId;
+ /**
+ * 任务编号
+ *
+ * 关联 {@link IotOtaTaskDO#getId()}
+ */
+ private Long taskId;
+
+ /**
+ * 设备编号
+ *
+ * 关联 {@link IotDeviceDO#getId()}
+ */
+ private Long deviceId;
+ /**
+ * 来源的固件编号
+ *
+ * 关联 {@link IotDeviceDO#getFirmwareId()}
+ */
+ private Long fromFirmwareId;
+
+ /**
+ * 升级状态
+ *
+ * 关联 {@link IotOtaTaskRecordStatusEnum}
+ */
+ private Integer status;
+ /**
+ * 升级进度，百分比
+ */
+ private Integer progress;
+ /**
+ * 升级进度描述
+ *
+ * 注意，只记录设备最后一次的升级进度描述
+ * 如果想看历史记录，可以查看 {@link IotDeviceMessageDO} 设备日志
+ */
+ private String description;
+
+}
\ No newline at end of file
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/rule/IotDataRuleDO.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/rule/IotDataRuleDO.java
new file mode 100644
index 0000000000..191df10d06
--- /dev/null
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/rule/IotDataRuleDO.java
@@ -0,0 +1,109 @@
+package cn.iocoder.yudao.module.iot.dal.dataobject.rule;
+
+import cn.iocoder.yudao.framework.common.enums.CommonStatusEnum;
+import cn.iocoder.yudao.framework.mybatis.core.dataobject.BaseDO;
+import cn.iocoder.yudao.framework.mybatis.core.type.LongListTypeHandler;
+import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
+import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceDO;
+import cn.iocoder.yudao.module.iot.dal.dataobject.product.IotProductDO;
+import cn.iocoder.yudao.module.iot.dal.dataobject.thingmodel.IotThingModelDO;
+import com.baomidou.mybatisplus.annotation.KeySequence;
+import com.baomidou.mybatisplus.annotation.TableField;
+import com.baomidou.mybatisplus.annotation.TableName;
+import com.baomidou.mybatisplus.extension.handlers.JacksonTypeHandler;
+import jakarta.validation.constraints.NotEmpty;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.util.List;
+
+/**
+ * IoT 数据流转规则 DO
+ *
+ * 监听 {@link SourceConfig} 数据源，转发到 {@link IotDataSinkDO} 数据目的
+ *
+ * @author 芋道源码
+ */
+@TableName(value = "iot_data_rule", autoResultMap = true)
+@KeySequence("iot_data_rule_seq") // 用于 Oracle、PostgreSQL、Kingbase、DB2、H2 数据库的主键自增。如果是 MySQL 等数据库，可不写。
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class IotDataRuleDO extends BaseDO {
+
+ /**
+ * 数据流转规格编号
+ */
+ private Long id;
+ /**
+ * 数据流转规格名称
+ */
+ private String name;
+ /**
+ * 数据流转规格描述
+ */
+ private String description;
+ /**
+ * 数据流转规格状态
+ *
+ * 枚举 {@link CommonStatusEnum}
+ */
+ private Integer status;
+
+ /**
+ * 数据源配置数组
+ */
+ @TableField(typeHandler = JacksonTypeHandler.class)
+ private List sourceConfigs;
+ /**
+ * 数据目的编号数组
+ *
+ * 关联 {@link IotDataSinkDO#getId()}
+ */
+ @TableField(typeHandler = LongListTypeHandler.class)
+ private List sinkIds;
+
+ // TODO @芋艿：未来考虑使用 groovy；支持数据处理；
+
+ /**
+ * 数据源配置
+ */
+ @Data
+ public static class SourceConfig {
+
+ /**
+ * 消息方法
+ *
+ * 枚举 {@link IotDeviceMessageMethodEnum} 中的 upstream 上行部分
+ */
+ @NotEmpty(message = "消息方法不能为空")
+ private String method;
+
+ /**
+ * 产品编号
+ *
+ * 关联 {@link IotProductDO#getId()}
+ */
+ private Long productId;
+ /**
+ * 设备编号
+ *
+ * 关联 {@link IotDeviceDO#getId()}
+ * 特殊：如果为 {@link IotDeviceDO#DEVICE_ID_ALL} 时，则是全部设备
+ */
+ @NotEmpty(message = "设备编号不能为空")
+ private Long deviceId;
+
+ /**
+ * 标识符
+ *
+ * 1. 物模型时，对应：{@link IotThingModelDO#getIdentifier()}
+ */
+ private String identifier;
+
+ }
+
+}
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/rule/IotSceneRuleDO.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/rule/IotSceneRuleDO.java
new file mode 100644
index 0000000000..94aa1eb5a3
--- /dev/null
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/rule/IotSceneRuleDO.java
@@ -0,0 +1,245 @@
+package cn.iocoder.yudao.module.iot.dal.dataobject.rule;
+
+import cn.iocoder.yudao.framework.common.enums.CommonStatusEnum;
+import cn.iocoder.yudao.framework.tenant.core.db.TenantBaseDO;
+import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
+import cn.iocoder.yudao.module.iot.dal.dataobject.alert.IotAlertConfigDO;
+import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceDO;
+import cn.iocoder.yudao.module.iot.dal.dataobject.product.IotProductDO;
+import cn.iocoder.yudao.module.iot.dal.dataobject.thingmodel.IotThingModelDO;
+import cn.iocoder.yudao.module.iot.enums.rule.IotSceneRuleActionTypeEnum;
+import cn.iocoder.yudao.module.iot.enums.rule.IotSceneRuleConditionOperatorEnum;
+import cn.iocoder.yudao.module.iot.enums.rule.IotSceneRuleConditionTypeEnum;
+import cn.iocoder.yudao.module.iot.enums.rule.IotSceneRuleTriggerTypeEnum;
+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.List;
+
+/**
+ * IoT 场景联动规则 DO
+ *
+ * @author 芋道源码
+ */
+@TableName(value = "iot_scene_rule", autoResultMap = true)
+@KeySequence("iot_scene_rule_seq") // 用于 Oracle、PostgreSQL、Kingbase、DB2、H2 数据库的主键自增。如果是 MySQL 等数据库，可不写。
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class IotSceneRuleDO extends TenantBaseDO {
+
+ /**
+ * 场景联动编号
+ */
+ @TableId
+ private Long id;
+ /**
+ * 场景联动名称
+ */
+ private String name;
+ /**
+ * 场景联动描述
+ */
+ private String description;
+ /**
+ * 场景联动状态
+ *
+ * 枚举 {@link CommonStatusEnum}
+ */
+ private Integer status;
+
+ /**
+ * 场景定义配置
+ */
+ @TableField(typeHandler = JacksonTypeHandler.class)
+ private List triggers;
+
+ /**
+ * 场景动作配置
+ */
+ @TableField(typeHandler = JacksonTypeHandler.class)
+ private List actions;
+
+ /**
+ * 场景定义配置
+ */
+ @Data
+ public static class Trigger {
+
+ // ========== 事件部分 ==========
+
+ /**
+ * 场景事件类型
+ *
+ * 枚举 {@link IotSceneRuleTriggerTypeEnum}
+ * 1. {@link IotSceneRuleTriggerTypeEnum#DEVICE_STATE_UPDATE} 时，operator 非空，并且 value 为在线状态
+ * 2. {@link IotSceneRuleTriggerTypeEnum#DEVICE_PROPERTY_POST}
+ * {@link IotSceneRuleTriggerTypeEnum#DEVICE_EVENT_POST} 时，identifier、operator 非空，并且 value 为属性值
+ * 3. {@link IotSceneRuleTriggerTypeEnum#DEVICE_EVENT_POST}
+ * {@link IotSceneRuleTriggerTypeEnum#DEVICE_SERVICE_INVOKE} 时，identifier 非空，但是 operator、value 为空
+ * 4. {@link IotSceneRuleTriggerTypeEnum#TIMER} 时，conditions 非空，并且设备无关（无需 productId、deviceId 字段）
+ */
+ private Integer type;
+
+ /**
+ * 产品编号
+ *
+ * 关联 {@link IotProductDO#getId()}
+ */
+ private Long productId;
+ /**
+ * 设备编号
+ *
+ * 关联 {@link IotDeviceDO#getId()}
+ * 特殊：如果为 {@link IotDeviceDO#DEVICE_ID_ALL} 时，则是全部设备
+ */
+ private Long deviceId;
+ /**
+ * 物模型标识符
+ *
+ * 对应：{@link IotThingModelDO#getIdentifier()}
+ */
+ private String identifier;
+ /**
+ * 操作符
+ *
+ * 枚举 {@link IotSceneRuleConditionOperatorEnum}
+ */
+ private String operator;
+ /**
+ * 参数（属性值、在线状态）
+ *
+ * 如果有多个值，则使用 "," 分隔，类似 "1,2,3"。
+ * 例如说，{@link IotSceneRuleConditionOperatorEnum#IN}、{@link IotSceneRuleConditionOperatorEnum#BETWEEN}
+ */
+ private String value;
+
+ /**
+ * CRON 表达式
+ */
+ private String cronExpression;
+
+ // ========== 条件部分 ==========
+
+ /**
+ * 触发条件分组（状态条件分组）的数组
+ *
+ * 第一层 List：分组与分组之间，是“或”的关系
+ * 第二层 List：条件与条件之间，是“且”的关系
+ */
+ private List> conditionGroups;
+
+ }
+
+ /**
+ * 触发条件（状态条件）
+ */
+ @Data
+ public static class TriggerCondition {
+
+ /**
+ * 触发条件类型
+ *
+ * 枚举 {@link IotSceneRuleConditionTypeEnum}
+ * 1. {@link IotSceneRuleConditionTypeEnum#DEVICE_STATE} 时，operator 非空，并且 value 为在线状态
+ * 2. {@link IotSceneRuleConditionTypeEnum#DEVICE_PROPERTY} 时，identifier、operator 非空，并且 value 为属性值
+ * 3. {@link IotSceneRuleConditionTypeEnum#CURRENT_TIME} 时，operator 非空（使用 DATE_TIME_ 和 TIME_ 部分），并且 value 非空
+ */
+ private Integer type;
+
+ /**
+ * 产品编号
+ *
+ * 关联 {@link IotProductDO#getId()}
+ */
+ private Long productId;
+ /**
+ * 设备编号
+ *
+ * 关联 {@link IotDeviceDO#getId()}
+ */
+ private Long deviceId;
+ /**
+ * 标识符（属性）
+ *
+ * 关联 {@link IotThingModelDO#getIdentifier()}
+ */
+ private String identifier;
+ /**
+ * 操作符
+ *
+ * 枚举 {@link IotSceneRuleConditionOperatorEnum}
+ */
+ private String operator;
+ /**
+ * 参数
+ *
+ * 如果有多个值，则使用 "," 分隔，类似 "1,2,3"。
+ * 例如说，{@link IotSceneRuleConditionOperatorEnum#IN}、{@link IotSceneRuleConditionOperatorEnum#BETWEEN}
+ */
+ private String param;
+
+ }
+
+ /**
+ * 场景动作配置
+ */
+ @Data
+ public static class Action {
+
+ /**
+ * 执行类型
+ *
+ * 枚举 {@link IotSceneRuleActionTypeEnum}
+ * 1. {@link IotSceneRuleActionTypeEnum#DEVICE_PROPERTY_SET} 时，params 非空
+ * {@link IotSceneRuleActionTypeEnum#DEVICE_SERVICE_INVOKE} 时，params 非空
+ * 2. {@link IotSceneRuleActionTypeEnum#ALERT_TRIGGER} 时，alertConfigId 为空，因为是 {@link IotAlertConfigDO} 里面关联它
+ * 3. {@link IotSceneRuleActionTypeEnum#ALERT_RECOVER} 时，alertConfigId 非空
+ */
+ private Integer type;
+
+ /**
+ * 产品编号
+ *
+ * 关联 {@link IotProductDO#getId()}
+ */
+ private Long productId;
+ /**
+ * 设备编号
+ *
+ * 关联 {@link IotDeviceDO#getId()}
+ */
+ private Long deviceId;
+
+ /**
+ * 标识符（服务）
+ *
+ * 关联 {@link IotThingModelDO#getIdentifier()}
+ */
+ private String identifier;
+
+ /**
+ * 请求参数
+ *
+ * 一般来说，对应 {@link IotDeviceMessage#getParams()} 请求参数
+ */
+ private String params;
+
+ /**
+ * 告警配置编号
+ *
+ * 关联 {@link IotAlertConfigDO#getId()}
+ */
+ private Long alertConfigId;
+
+ }
+
+}
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/rule/config/IotAbstractDataSinkConfig.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/rule/config/IotAbstractDataSinkConfig.java
new file mode 100644
index 0000000000..68a8fd699b
--- /dev/null
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/rule/config/IotAbstractDataSinkConfig.java
@@ -0,0 +1,35 @@
+package cn.iocoder.yudao.module.iot.dal.dataobject.rule.config;
+
+import cn.iocoder.yudao.module.iot.enums.rule.IotDataSinkTypeEnum;
+import com.fasterxml.jackson.annotation.JsonSubTypes;
+import com.fasterxml.jackson.annotation.JsonTypeInfo;
+import lombok.Data;
+
+/**
+ * IoT IotDataBridgeConfig 抽象类
+ *
+ * 用于表示数据目的配置数据的通用类型，根据具体的 "type" 字段动态映射到对应的子类
+ * 提供多态支持，适用于不同类型的数据结构序列化和反序列化场景。
+ *
+ * @author HUIHUI
+ */
+@Data
+@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type", visible = true)
+@JsonSubTypes({
+ @JsonSubTypes.Type(value = IotDataSinkHttpConfig.class, name = "1"),
+ @JsonSubTypes.Type(value = IotDataSinkMqttConfig.class, name = "10"),
+ @JsonSubTypes.Type(value = IotDataSinkRedisConfig.class, name = "21"),
+ @JsonSubTypes.Type(value = IotDataSinkRocketMQConfig.class, name = "30"),
+ @JsonSubTypes.Type(value = IotDataSinkRabbitMQConfig.class, name = "31"),
+ @JsonSubTypes.Type(value = IotDataSinkKafkaConfig.class, name = "32"),
+})
+public abstract class IotAbstractDataSinkConfig {
+
+ /**
+ * 配置类型
+ *
+ * 枚举 {@link IotDataSinkTypeEnum#getType()}
+ */
+ private String type;
+
+}
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/rule/config/IotDataSinkRedisConfig.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/rule/config/IotDataSinkRedisConfig.java
new file mode 100644
index 0000000000..07460ac368
--- /dev/null
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/rule/config/IotDataSinkRedisConfig.java
@@ -0,0 +1,64 @@
+package cn.iocoder.yudao.module.iot.dal.dataobject.rule.config;
+
+import cn.iocoder.yudao.framework.common.validation.InEnum;
+import cn.iocoder.yudao.module.iot.enums.rule.IotRedisDataStructureEnum;
+import lombok.Data;
+
+/**
+ * IoT Redis 配置 {@link IotAbstractDataSinkConfig} 实现类
+ *
+ * @author HUIHUI
+ */
+@Data
+public class IotDataSinkRedisConfig extends IotAbstractDataSinkConfig {
+
+ /**
+ * Redis 服务器地址
+ */
+ private String host;
+ /**
+ * 端口
+ */
+ private Integer port;
+ /**
+ * 密码
+ */
+ private String password;
+ /**
+ * 数据库索引
+ */
+ private Integer database;
+
+ /**
+ * Redis 数据结构类型
+ *
+ * 枚举 {@link IotRedisDataStructureEnum}
+ */
+ @InEnum(IotRedisDataStructureEnum.class)
+ private Integer dataStructure;
+
+ /**
+ * 主题/键名
+ *
+ * 对于不同的数据结构：
+ * - Stream: 流的键名
+ * - Hash: Hash 的键名
+ * - List: 列表的键名
+ * - Set: 集合的键名
+ * - ZSet: 有序集合的键名
+ * - String: 字符串的键名
+ */
+ private String topic;
+
+ /**
+ * Hash 字段名（仅当 dataStructure 为 HASH 时使用）
+ */
+ private String hashField;
+
+ /**
+ * ZSet 分数字段（仅当 dataStructure 为 ZSET 时使用）
+ * 指定消息中哪个字段作为分数，如果不指定则使用当前时间戳
+ */
+ private String scoreField;
+
+}
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/thingmodel/model/dataType/ThingModelBoolOrEnumDataSpecs.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/thingmodel/model/dataType/ThingModelBoolOrEnumDataSpecs.java
new file mode 100644
index 0000000000..8533fcc6f5
--- /dev/null
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/thingmodel/model/dataType/ThingModelBoolOrEnumDataSpecs.java
@@ -0,0 +1,30 @@
+package cn.iocoder.yudao.module.iot.dal.dataobject.thingmodel.model.dataType;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import jakarta.validation.constraints.NotEmpty;
+import jakarta.validation.constraints.NotNull;
+import jakarta.validation.constraints.Pattern;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+/**
+ * IoT 物模型数据类型为布尔型或枚举型的 DataSpec 定义
+ *
+ * 数据类型，取值为 bool 或 enum
+ *
+ * @author HUIHUI
+ */
+@Data
+@EqualsAndHashCode(callSuper = true)
+@JsonIgnoreProperties({"dataType"}) // 忽略子类中的 dataType 字段，从而避免重复
+public class ThingModelBoolOrEnumDataSpecs extends ThingModelDataSpecs {
+
+ @NotEmpty(message = "枚举项的名称不能为空")
+ @Pattern(regexp = "^[\\u4e00-\\u9fa5a-zA-Z0-9][\\u4e00-\\u9fa5a-zA-Z0-9_-]{0,19}$",
+ message = "枚举项的名称只能包含中文、大小写英文字母、数字、下划线和短划线，必须以中文、英文字母或数字开头，长度不超过 20 个字符")
+ private String name;
+
+ @NotNull(message = "枚举值不能为空")
+ private Integer value;
+
+}
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/alert/IotAlertConfigMapper.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/alert/IotAlertConfigMapper.java
new file mode 100644
index 0000000000..c5d7154ff6
--- /dev/null
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/alert/IotAlertConfigMapper.java
@@ -0,0 +1,39 @@
+package cn.iocoder.yudao.module.iot.dal.mysql.alert;
+
+import cn.iocoder.yudao.framework.common.pojo.PageResult;
+import cn.iocoder.yudao.framework.mybatis.core.mapper.BaseMapperX;
+import cn.iocoder.yudao.framework.mybatis.core.query.LambdaQueryWrapperX;
+import cn.iocoder.yudao.framework.mybatis.core.util.MyBatisUtils;
+import cn.iocoder.yudao.module.iot.controller.admin.alert.vo.config.IotAlertConfigPageReqVO;
+import cn.iocoder.yudao.module.iot.dal.dataobject.alert.IotAlertConfigDO;
+import org.apache.ibatis.annotations.Mapper;
+
+import java.util.List;
+
+/**
+ * IoT 告警配置 Mapper
+ *
+ * @author 芋道源码
+ */
+@Mapper
+public interface IotAlertConfigMapper extends BaseMapperX {
+
+ default PageResult selectPage(IotAlertConfigPageReqVO reqVO) {
+ return selectPage(reqVO, new LambdaQueryWrapperX()
+ .likeIfPresent(IotAlertConfigDO::getName, reqVO.getName())
+ .eqIfPresent(IotAlertConfigDO::getStatus, reqVO.getStatus())
+ .betweenIfPresent(IotAlertConfigDO::getCreateTime, reqVO.getCreateTime())
+ .orderByDesc(IotAlertConfigDO::getId));
+ }
+
+ default List selectListByStatus(Integer status) {
+ return selectList(IotAlertConfigDO::getStatus, status);
+ }
+
+ default List selectListBySceneRuleIdAndStatus(Long sceneRuleId, Integer status) {
+ return selectList(new LambdaQueryWrapperX()
+ .eq(IotAlertConfigDO::getStatus, status)
+ .apply(MyBatisUtils.findInSet("scene_rule_id", sceneRuleId)));
+ }
+
+}
\ No newline at end of file
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/alert/IotAlertRecordMapper.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/alert/IotAlertRecordMapper.java
new file mode 100644
index 0000000000..f23fe60f74
--- /dev/null
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/alert/IotAlertRecordMapper.java
@@ -0,0 +1,46 @@
+package cn.iocoder.yudao.module.iot.dal.mysql.alert;
+
+import cn.iocoder.yudao.framework.common.pojo.PageResult;
+import cn.iocoder.yudao.framework.mybatis.core.mapper.BaseMapperX;
+import cn.iocoder.yudao.framework.mybatis.core.query.LambdaQueryWrapperX;
+import cn.iocoder.yudao.module.iot.controller.admin.alert.vo.recrod.IotAlertRecordPageReqVO;
+import cn.iocoder.yudao.module.iot.dal.dataobject.alert.IotAlertRecordDO;
+import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
+import org.apache.ibatis.annotations.Mapper;
+
+import java.util.Collection;
+import java.util.List;
+
+/**
+ * IoT 告警记录 Mapper
+ *
+ * @author 芋道源码
+ */
+@Mapper
+public interface IotAlertRecordMapper extends BaseMapperX {
+
+ default PageResult selectPage(IotAlertRecordPageReqVO reqVO) {
+ return selectPage(reqVO, new LambdaQueryWrapperX()
+ .eqIfPresent(IotAlertRecordDO::getConfigId, reqVO.getConfigId())
+ .eqIfPresent(IotAlertRecordDO::getConfigLevel, reqVO.getLevel())
+ .eqIfPresent(IotAlertRecordDO::getProductId, reqVO.getProductId())
+ .eqIfPresent(IotAlertRecordDO::getDeviceId, reqVO.getDeviceId())
+ .eqIfPresent(IotAlertRecordDO::getProcessStatus, reqVO.getProcessStatus())
+ .betweenIfPresent(IotAlertRecordDO::getCreateTime, reqVO.getCreateTime())
+ .orderByDesc(IotAlertRecordDO::getId));
+ }
+
+ default List selectListBySceneRuleId(Long sceneRuleId, Long deviceId, Boolean processStatus) {
+ return selectList(new LambdaQueryWrapperX()
+ .eq(IotAlertRecordDO::getSceneRuleId, sceneRuleId)
+ .eqIfPresent(IotAlertRecordDO::getDeviceId, deviceId)
+ .eqIfPresent(IotAlertRecordDO::getProcessStatus, processStatus)
+ .orderByDesc(IotAlertRecordDO::getId));
+ }
+
+ default int updateList(Collection ids, IotAlertRecordDO updateObj) {
+ return update(updateObj, new LambdaUpdateWrapper()
+ .in(IotAlertRecordDO::getId, ids));
+ }
+
+}
\ No newline at end of file
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/ota/IotOtaTaskMapper.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/ota/IotOtaTaskMapper.java
new file mode 100644
index 0000000000..cf73231234
--- /dev/null
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/ota/IotOtaTaskMapper.java
@@ -0,0 +1,32 @@
+package cn.iocoder.yudao.module.iot.dal.mysql.ota;
+
+import cn.iocoder.yudao.framework.common.pojo.PageResult;
+import cn.iocoder.yudao.framework.mybatis.core.mapper.BaseMapperX;
+import cn.iocoder.yudao.framework.mybatis.core.query.LambdaQueryWrapperX;
+import cn.iocoder.yudao.module.iot.controller.admin.ota.vo.task.IotOtaTaskPageReqVO;
+import cn.iocoder.yudao.module.iot.dal.dataobject.ota.IotOtaTaskDO;
+import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
+import org.apache.ibatis.annotations.Mapper;
+
+@Mapper
+public interface IotOtaTaskMapper extends BaseMapperX {
+
+ default IotOtaTaskDO selectByFirmwareIdAndName(Long firmwareId, String name) {
+ return selectOne(IotOtaTaskDO::getFirmwareId, firmwareId,
+ IotOtaTaskDO::getName, name);
+ }
+
+ default PageResult selectPage(IotOtaTaskPageReqVO pageReqVO) {
+ return selectPage(pageReqVO, new LambdaQueryWrapperX()
+ .eqIfPresent(IotOtaTaskDO::getFirmwareId, pageReqVO.getFirmwareId())
+ .likeIfPresent(IotOtaTaskDO::getName, pageReqVO.getName())
+ .orderByDesc(IotOtaTaskDO::getId));
+ }
+
+ default int updateByIdAndStatus(Long id, Integer whereStatus, IotOtaTaskDO updateObj) {
+ return update(updateObj, new LambdaUpdateWrapper()
+ .eq(IotOtaTaskDO::getId, id)
+ .eq(IotOtaTaskDO::getStatus, whereStatus));
+ }
+
+}
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/ota/IotOtaTaskRecordMapper.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/ota/IotOtaTaskRecordMapper.java
new file mode 100644
index 0000000000..017adc9192
--- /dev/null
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/ota/IotOtaTaskRecordMapper.java
@@ -0,0 +1,80 @@
+package cn.iocoder.yudao.module.iot.dal.mysql.ota;
+
+import cn.iocoder.yudao.framework.common.pojo.PageResult;
+import cn.iocoder.yudao.framework.mybatis.core.mapper.BaseMapperX;
+import cn.iocoder.yudao.framework.mybatis.core.query.LambdaQueryWrapperX;
+import cn.iocoder.yudao.module.iot.controller.admin.ota.vo.task.record.IotOtaTaskRecordPageReqVO;
+import cn.iocoder.yudao.module.iot.dal.dataobject.ota.IotOtaTaskRecordDO;
+import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
+import org.apache.ibatis.annotations.Mapper;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.Set;
+
+@Mapper
+public interface IotOtaTaskRecordMapper extends BaseMapperX {
+
+ default List selectListByFirmwareIdAndTaskId(Long firmwareId, Long taskId) {
+ return selectList(new LambdaQueryWrapperX()
+ .eqIfPresent(IotOtaTaskRecordDO::getFirmwareId, firmwareId)
+ .eqIfPresent(IotOtaTaskRecordDO::getTaskId, taskId)
+ .select(IotOtaTaskRecordDO::getDeviceId, IotOtaTaskRecordDO::getStatus));
+ }
+
+ default PageResult selectPage(IotOtaTaskRecordPageReqVO pageReqVO) {
+ return selectPage(pageReqVO, new LambdaQueryWrapperX()
+ .eqIfPresent(IotOtaTaskRecordDO::getTaskId, pageReqVO.getTaskId())
+ .eqIfPresent(IotOtaTaskRecordDO::getStatus, pageReqVO.getStatus()));
+ }
+
+ default List selectListByTaskIdAndStatus(Long taskId, Collection statuses) {
+ return selectList(new LambdaQueryWrapperX()
+ .eq(IotOtaTaskRecordDO::getTaskId, taskId)
+ .in(IotOtaTaskRecordDO::getStatus, statuses));
+ }
+
+ default Long selectCountByTaskIdAndStatus(Long taskId, Collection statuses) {
+ return selectCount(new LambdaQueryWrapperX()
+ .eq(IotOtaTaskRecordDO::getTaskId, taskId)
+ .in(IotOtaTaskRecordDO::getStatus, statuses));
+ }
+
+ default int updateByIdAndStatus(Long id, Integer status,
+ IotOtaTaskRecordDO updateObj) {
+ return update(updateObj, new LambdaUpdateWrapper()
+ .eq(IotOtaTaskRecordDO::getId, id)
+ .eq(IotOtaTaskRecordDO::getStatus, status));
+ }
+
+ default int updateByIdAndStatus(Long id, Collection whereStatuses,
+ IotOtaTaskRecordDO updateObj) {
+ return update(updateObj, new LambdaUpdateWrapper()
+ .eq(IotOtaTaskRecordDO::getId, id)
+ .in(IotOtaTaskRecordDO::getStatus, whereStatuses));
+ }
+
+ default void updateListByIdAndStatus(Collection ids, Collection whereStatuses,
+ IotOtaTaskRecordDO updateObj) {
+ update(updateObj, new LambdaUpdateWrapper()
+ .in(IotOtaTaskRecordDO::getId, ids)
+ .in(IotOtaTaskRecordDO::getStatus, whereStatuses));
+ }
+
+ default List selectListByDeviceIdAndStatus(Set deviceIds, Set statuses) {
+ return selectList(new LambdaQueryWrapperX()
+ .inIfPresent(IotOtaTaskRecordDO::getDeviceId, deviceIds)
+ .inIfPresent(IotOtaTaskRecordDO::getStatus, statuses));
+ }
+
+ default List selectListByDeviceIdAndStatus(Long deviceId, Set statuses) {
+ return selectList(new LambdaQueryWrapperX()
+ .eqIfPresent(IotOtaTaskRecordDO::getDeviceId, deviceId)
+ .inIfPresent(IotOtaTaskRecordDO::getStatus, statuses));
+ }
+
+ default List selectListByStatus(Integer status) {
+ return selectList(IotOtaTaskRecordDO::getStatus, status);
+ }
+
+}
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/rule/IotDataRuleMapper.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/rule/IotDataRuleMapper.java
new file mode 100644
index 0000000000..7c0c17d3bc
--- /dev/null
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/rule/IotDataRuleMapper.java
@@ -0,0 +1,38 @@
+package cn.iocoder.yudao.module.iot.dal.mysql.rule;
+
+import cn.iocoder.yudao.framework.common.pojo.PageResult;
+import cn.iocoder.yudao.framework.mybatis.core.mapper.BaseMapperX;
+import cn.iocoder.yudao.framework.mybatis.core.query.LambdaQueryWrapperX;
+import cn.iocoder.yudao.framework.mybatis.core.util.MyBatisUtils;
+import cn.iocoder.yudao.module.iot.controller.admin.rule.vo.data.rule.IotDataRulePageReqVO;
+import cn.iocoder.yudao.module.iot.dal.dataobject.rule.IotDataRuleDO;
+import org.apache.ibatis.annotations.Mapper;
+
+import java.util.List;
+
+/**
+ * IoT 数据流转规则 Mapper
+ *
+ * @author 芋道源码
+ */
+@Mapper
+public interface IotDataRuleMapper extends BaseMapperX {
+
+ default PageResult selectPage(IotDataRulePageReqVO reqVO) {
+ return selectPage(reqVO, new LambdaQueryWrapperX()
+ .likeIfPresent(IotDataRuleDO::getName, reqVO.getName())
+ .eqIfPresent(IotDataRuleDO::getStatus, reqVO.getStatus())
+ .betweenIfPresent(IotDataRuleDO::getCreateTime, reqVO.getCreateTime())
+ .orderByDesc(IotDataRuleDO::getId));
+ }
+
+ default List selectListBySinkId(Long sinkId) {
+ return selectList(new LambdaQueryWrapperX()
+ .apply(MyBatisUtils.findInSet("sink_ids", sinkId)));
+ }
+
+ default List selectListByStatus(Integer status) {
+ return selectList(IotDataRuleDO::getStatus, status);
+ }
+
+}
\ No newline at end of file
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/rule/IotDataSinkMapper.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/rule/IotDataSinkMapper.java
new file mode 100644
index 0000000000..e65001db86
--- /dev/null
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/rule/IotDataSinkMapper.java
@@ -0,0 +1,32 @@
+package cn.iocoder.yudao.module.iot.dal.mysql.rule;
+
+import cn.iocoder.yudao.framework.common.pojo.PageResult;
+import cn.iocoder.yudao.framework.mybatis.core.mapper.BaseMapperX;
+import cn.iocoder.yudao.framework.mybatis.core.query.LambdaQueryWrapperX;
+import cn.iocoder.yudao.module.iot.controller.admin.rule.vo.data.sink.IotDataSinkPageReqVO;
+import cn.iocoder.yudao.module.iot.dal.dataobject.rule.IotDataSinkDO;
+import org.apache.ibatis.annotations.Mapper;
+
+import java.util.List;
+
+/**
+ * IoT 数据流转目的 Mapper
+ *
+ * @author HUIHUI
+ */
+@Mapper
+public interface IotDataSinkMapper extends BaseMapperX {
+
+ default PageResult selectPage(IotDataSinkPageReqVO reqVO) {
+ return selectPage(reqVO, new LambdaQueryWrapperX()
+ .likeIfPresent(IotDataSinkDO::getName, reqVO.getName())
+ .eqIfPresent(IotDataSinkDO::getStatus, reqVO.getStatus())
+ .betweenIfPresent(IotDataSinkDO::getCreateTime, reqVO.getCreateTime())
+ .orderByDesc(IotDataSinkDO::getId));
+ }
+
+ default List selectListByStatus(Integer status) {
+ return selectList(IotDataSinkDO::getStatus, status);
+ }
+
+}
\ No newline at end of file
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/rule/IotSceneRuleMapper.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/rule/IotSceneRuleMapper.java
new file mode 100644
index 0000000000..4fd6490d15
--- /dev/null
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/rule/IotSceneRuleMapper.java
@@ -0,0 +1,33 @@
+package cn.iocoder.yudao.module.iot.dal.mysql.rule;
+
+import cn.iocoder.yudao.framework.common.pojo.PageResult;
+import cn.iocoder.yudao.framework.mybatis.core.mapper.BaseMapperX;
+import cn.iocoder.yudao.framework.mybatis.core.query.LambdaQueryWrapperX;
+import cn.iocoder.yudao.module.iot.controller.admin.rule.vo.scene.IotSceneRulePageReqVO;
+import cn.iocoder.yudao.module.iot.dal.dataobject.rule.IotSceneRuleDO;
+import org.apache.ibatis.annotations.Mapper;
+
+import java.util.List;
+
+/**
+ * IoT 场景联动 Mapper
+ *
+ * @author HUIHUI
+ */
+@Mapper
+public interface IotSceneRuleMapper extends BaseMapperX {
+
+ default PageResult selectPage(IotSceneRulePageReqVO reqVO) {
+ return selectPage(reqVO, new LambdaQueryWrapperX()
+ .likeIfPresent(IotSceneRuleDO::getName, reqVO.getName())
+ .likeIfPresent(IotSceneRuleDO::getDescription, reqVO.getDescription())
+ .eqIfPresent(IotSceneRuleDO::getStatus, reqVO.getStatus())
+ .betweenIfPresent(IotSceneRuleDO::getCreateTime, reqVO.getCreateTime())
+ .orderByDesc(IotSceneRuleDO::getId));
+ }
+
+ default List selectListByStatus(Integer status) {
+ return selectList(IotSceneRuleDO::getStatus, status);
+ }
+
+}
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/redis/device/DeviceServerIdRedisDAO.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/redis/device/DeviceServerIdRedisDAO.java
new file mode 100644
index 0000000000..cef78f3cff
--- /dev/null
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/redis/device/DeviceServerIdRedisDAO.java
@@ -0,0 +1,30 @@
+package cn.iocoder.yudao.module.iot.dal.redis.device;
+
+import cn.iocoder.yudao.module.iot.dal.redis.RedisKeyConstants;
+import jakarta.annotation.Resource;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.stereotype.Repository;
+
+/**
+ * 设备关联的网关 serverId 的 Redis DAO
+ *
+ * @author 芋道源码
+ */
+@Repository
+public class DeviceServerIdRedisDAO {
+
+ @Resource
+ private StringRedisTemplate stringRedisTemplate;
+
+ public void update(Long deviceId, String serverId) {
+ stringRedisTemplate.opsForHash().put(RedisKeyConstants.DEVICE_SERVER_ID,
+ String.valueOf(deviceId), serverId);
+ }
+
+ public String get(Long deviceId) {
+ Object value = stringRedisTemplate.opsForHash().get(RedisKeyConstants.DEVICE_SERVER_ID,
+ String.valueOf(deviceId));
+ return value != null ? (String) value : null;
+ }
+
+}
\ No newline at end of file
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/tdengine/IotDeviceMessageMapper.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/tdengine/IotDeviceMessageMapper.java
new file mode 100644
index 0000000000..b09895fd36
--- /dev/null
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/tdengine/IotDeviceMessageMapper.java
@@ -0,0 +1,79 @@
+package cn.iocoder.yudao.module.iot.dal.tdengine;
+
+import cn.iocoder.yudao.module.iot.controller.admin.device.vo.message.IotDeviceMessagePageReqVO;
+import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceMessageDO;
+import cn.iocoder.yudao.module.iot.framework.tdengine.core.annotation.TDengineDS;
+import com.baomidou.mybatisplus.annotation.InterceptorIgnore;
+import com.baomidou.mybatisplus.core.metadata.IPage;
+import org.apache.ibatis.annotations.Mapper;
+import org.apache.ibatis.annotations.Param;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * 设备消息 {@link IotDeviceMessageDO} Mapper 接口
+ */
+@Mapper
+@TDengineDS
+@InterceptorIgnore(tenantLine = "true") // 避免 SQL 解析，因为 JSqlParser 对 TDengine 的 SQL 解析会报错
+public interface IotDeviceMessageMapper {
+
+ /**
+ * 创建设备消息超级表
+ */
+ void createSTable();
+
+ /**
+ * 查询设备消息表是否存在
+ *
+ * @return 存在则返回表名；不存在则返回 null
+ */
+ String showSTable();
+
+ /**
+ * 插入设备消息数据
+ *
+ * 如果子表不存在，会自动创建子表
+ *
+ * @param message 设备消息数据
+ */
+ void insert(IotDeviceMessageDO message);
+
+ /**
+ * 获得设备消息分页
+ *
+ * @param reqVO 分页查询条件
+ * @return 设备消息列表
+ */
+ IPage selectPage(IPage page,
+ @Param("reqVO") IotDeviceMessagePageReqVO reqVO);
+
+ /**
+ * 统计设备消息数量
+ *
+ * @param createTime 创建时间，如果为空，则统计所有消息数量
+ * @return 消息数量
+ */
+ Long selectCountByCreateTime(@Param("createTime") Long createTime);
+
+ /**
+ * 按照 requestIds 批量查询消息
+ *
+ * @param deviceId 设备编号
+ * @param requestIds 请求编号集合
+ * @param reply 是否回复消息
+ * @return 消息列表
+ */
+ List selectListByRequestIdsAndReply(@Param("deviceId") Long deviceId,
+ @Param("requestIds") Collection requestIds,
+ @Param("reply") Boolean reply);
+
+ /**
+ * 按照时间范围（小时），统计设备的消息数量
+ */
+ List> selectDeviceMessageCountGroupByDate(@Param("startTime") Long startTime,
+ @Param("endTime") Long endTime);
+
+}
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/enums/DictTypeConstants.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/enums/DictTypeConstants.java
new file mode 100644
index 0000000000..4f07ddfc1c
--- /dev/null
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/enums/DictTypeConstants.java
@@ -0,0 +1,25 @@
+package cn.iocoder.yudao.module.iot.enums;
+
+/**
+ * IoT 字典类型的枚举类
+ *
+ * @author 芋道源码
+ */
+public class DictTypeConstants {
+
+ public static final String NET_TYPE = "iot_net_type";
+ public static final String LOCATION_TYPE = "iot_location_type";
+ public static final String CODEC_TYPE = "iot_codec_type";
+
+ public static final String PRODUCT_STATUS = "iot_product_status";
+ public static final String PRODUCT_DEVICE_TYPE = "iot_product_device_type";
+
+ public static final String DEVICE_STATE = "iot_device_state";
+
+ public static final String ALERT_LEVEL = "iot_alert_level";
+
+ public static final String OTA_TASK_DEVICE_SCOPE = "iot_ota_task_device_scope";
+ public static final String OTA_TASK_STATUS = "iot_ota_task_status";
+ public static final String OTA_TASK_RECORD_STATUS = "iot_ota_task_record_status";
+
+}
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/enums/ErrorCodeConstants.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/enums/ErrorCodeConstants.java
new file mode 100644
index 0000000000..d1cf60e206
--- /dev/null
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/enums/ErrorCodeConstants.java
@@ -0,0 +1,82 @@
+package cn.iocoder.yudao.module.iot.enums;
+
+import cn.iocoder.yudao.framework.common.exception.ErrorCode;
+
+/**
+ * iot 错误码枚举类
+ *
+ * iot 系统，使用 1-050-000-000 段
+ */
+public interface ErrorCodeConstants {
+
+ // ========== 产品相关 1-050-001-000 ============
+ ErrorCode PRODUCT_NOT_EXISTS = new ErrorCode(1_050_001_000, "产品不存在");
+ ErrorCode PRODUCT_KEY_EXISTS = new ErrorCode(1_050_001_001, "产品标识已经存在");
+ ErrorCode PRODUCT_STATUS_NOT_DELETE = new ErrorCode(1_050_001_002, "产品状是发布状态，不允许删除");
+ ErrorCode PRODUCT_STATUS_NOT_ALLOW_THING_MODEL = new ErrorCode(1_050_001_003, "产品状是发布状态，不允许操作物模型");
+ ErrorCode PRODUCT_DELETE_FAIL_HAS_DEVICE = new ErrorCode(1_050_001_004, "产品下存在设备，不允许删除");
+
+ // ========== 产品物模型 1-050-002-000 ============
+ ErrorCode THING_MODEL_NOT_EXISTS = new ErrorCode(1_050_002_000, "产品物模型不存在");
+ ErrorCode THING_MODEL_EXISTS_BY_PRODUCT_KEY = new ErrorCode(1_050_002_001, "ProductKey 对应的产品物模型已存在");
+ ErrorCode THING_MODEL_IDENTIFIER_EXISTS = new ErrorCode(1_050_002_002, "存在重复的功能标识符。");
+ ErrorCode THING_MODEL_NAME_EXISTS = new ErrorCode(1_050_002_003, "存在重复的功能名称。");
+ ErrorCode THING_MODEL_IDENTIFIER_INVALID = new ErrorCode(1_050_002_003, "产品物模型标识无效");
+
+ // ========== 设备 1-050-003-000 ============
+ ErrorCode DEVICE_NOT_EXISTS = new ErrorCode(1_050_003_000, "设备不存在");
+ ErrorCode DEVICE_NAME_EXISTS = new ErrorCode(1_050_003_001, "设备名称在同一产品下必须唯一");
+ ErrorCode DEVICE_HAS_CHILDREN = new ErrorCode(1_050_003_002, "有子设备，不允许删除");
+ ErrorCode DEVICE_KEY_EXISTS = new ErrorCode(1_050_003_003, "设备标识已经存在");
+ ErrorCode DEVICE_GATEWAY_NOT_EXISTS = new ErrorCode(1_050_003_004, "网关设备不存在");
+ ErrorCode DEVICE_NOT_GATEWAY = new ErrorCode(1_050_003_005, "设备不是网关设备");
+ ErrorCode DEVICE_IMPORT_LIST_IS_EMPTY = new ErrorCode(1_050_003_006, "导入设备数据不能为空！");
+ ErrorCode DEVICE_DOWNSTREAM_FAILED_SERVER_ID_NULL = new ErrorCode(1_050_003_007, "下行设备消息失败，原因：设备未连接网关");
+ ErrorCode DEVICE_SERIAL_NUMBER_EXISTS = new ErrorCode(1_050_003_008, "设备序列号已存在，序列号必须全局唯一");
+
+ // ========== 产品分类 1-050-004-000 ==========
+ ErrorCode PRODUCT_CATEGORY_NOT_EXISTS = new ErrorCode(1_050_004_000, "产品分类不存在");
+
+ // ========== 设备分组 1-050-005-000 ==========
+ ErrorCode DEVICE_GROUP_NOT_EXISTS = new ErrorCode(1_050_005_000, "设备分组不存在");
+ ErrorCode DEVICE_GROUP_DELETE_FAIL_DEVICE_EXISTS = new ErrorCode(1_050_005_001, "设备分组下存在设备，不允许删除");
+
+ // ========== OTA 固件相关 1-050-008-000 ==========
+
+ ErrorCode OTA_FIRMWARE_NOT_EXISTS = new ErrorCode(1_050_008_000, "固件信息不存在");
+ ErrorCode OTA_FIRMWARE_PRODUCT_VERSION_DUPLICATE = new ErrorCode(1_050_008_001, "产品版本号重复");
+
+ // ========== OTA 升级任务相关 1-050-008-100 ==========
+
+ ErrorCode OTA_TASK_NOT_EXISTS = new ErrorCode(1_050_008_100, "升级任务不存在");
+ ErrorCode OTA_TASK_CREATE_FAIL_NAME_DUPLICATE = new ErrorCode(1_050_008_101, "创建 OTA 任务失败，原因：任务名称重复");
+ ErrorCode OTA_TASK_CREATE_FAIL_DEVICE_FIRMWARE_EXISTS = new ErrorCode(1_050_008_102,
+ "创建 OTA 任务失败，原因：设备({})已经是该固件版本");
+ ErrorCode OTA_TASK_CREATE_FAIL_DEVICE_OTA_IN_PROCESS = new ErrorCode(1_050_008_102,
+ "创建 OTA 任务失败，原因：设备({})已经在升级中...");
+ ErrorCode OTA_TASK_CREATE_FAIL_DEVICE_EMPTY = new ErrorCode(1_050_008_103, "创建 OTA 任务失败，原因：没有可升级的设备");
+ ErrorCode OTA_TASK_CANCEL_FAIL_STATUS_END = new ErrorCode(1_050_008_104, "取消 OTA 任务失败，原因：任务状态不是进行中");
+
+ // ========== OTA 升级任务记录相关 1-050-008-200 ==========
+
+ ErrorCode OTA_TASK_RECORD_NOT_EXISTS = new ErrorCode(1_050_008_200, "升级记录不存在");
+ ErrorCode OTA_TASK_RECORD_CANCEL_FAIL_STATUS_ERROR = new ErrorCode(1_050_008_201, "取消 OTA 升级记录失败，原因：记录状态不是进行中");
+ ErrorCode OTA_TASK_RECORD_UPDATE_PROGRESS_FAIL_NO_EXISTS = new ErrorCode(1_050_008_202, "更新 OTA 升级记录进度失败，原因：该设备没有进行中的升级记录");
+
+ // ========== IoT 数据流转规则 1-050-010-000 ==========
+ ErrorCode DATA_RULE_NOT_EXISTS = new ErrorCode(1_050_010_000, "数据流转规则不存在");
+
+ // ========== IoT 数据流转目的 1-050-011-000 ==========
+ ErrorCode DATA_SINK_NOT_EXISTS = new ErrorCode(1_050_011_000, "数据桥梁不存在");
+ ErrorCode DATA_SINK_DELETE_FAIL_USED_BY_RULE = new ErrorCode(1_050_011_001, "数据流转目的正在被数据流转规则使用，无法删除");
+
+ // ========== IoT 场景联动 1-050-012-000 ==========
+ ErrorCode RULE_SCENE_NOT_EXISTS = new ErrorCode(1_050_012_000, "场景联动不存在");
+
+ // ========== IoT 告警配置 1-050-013-000 ==========
+ ErrorCode ALERT_CONFIG_NOT_EXISTS = new ErrorCode(1_050_013_000, "IoT 告警配置不存在");
+
+ // ========== IoT 告警记录 1-050-014-000 ==========
+ ErrorCode ALERT_RECORD_NOT_EXISTS = new ErrorCode(1_050_014_000, "IoT 告警记录不存在");
+
+}
\ No newline at end of file
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/enums/ota/IotOtaTaskRecordStatusEnum.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/enums/ota/IotOtaTaskRecordStatusEnum.java
new file mode 100644
index 0000000000..0f95eb79cc
--- /dev/null
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/enums/ota/IotOtaTaskRecordStatusEnum.java
@@ -0,0 +1,57 @@
+package cn.iocoder.yudao.module.iot.enums.ota;
+
+
+import cn.hutool.core.util.ArrayUtil;
+import cn.iocoder.yudao.framework.common.core.ArrayValuable;
+import cn.iocoder.yudao.framework.common.util.collection.SetUtils;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * IoT OTA 升级任务记录的状态枚举
+ *
+ * @author 芋道源码
+ */
+@RequiredArgsConstructor
+@Getter
+public enum IotOtaTaskRecordStatusEnum implements ArrayValuable {
+
+ PENDING(0), // 待推送
+ PUSHED(10), // 已推送
+ UPGRADING(20), // 升级中
+ SUCCESS(30), // 升级成功
+ FAILURE(40), // 升级失败
+ CANCELED(50),; // 升级取消
+
+ public static final Integer[] ARRAYS = Arrays.stream(values())
+ .map(IotOtaTaskRecordStatusEnum::getStatus).toArray(Integer[]::new);
+
+ public static final Set IN_PROCESS_STATUSES = SetUtils.asSet(
+ PENDING.getStatus(),
+ PUSHED.getStatus(),
+ UPGRADING.getStatus());
+
+ public static final List PRIORITY_STATUSES = Arrays.asList(
+ SUCCESS.getStatus(),
+ PENDING.getStatus(), PUSHED.getStatus(), UPGRADING.getStatus(),
+ FAILURE.getStatus(), CANCELED.getStatus());
+
+ /**
+ * 状态
+ */
+ private final Integer status;
+
+ @Override
+ public Integer[] array() {
+ return ARRAYS;
+ }
+
+ public static IotOtaTaskRecordStatusEnum of(Integer status) {
+ return ArrayUtil.firstMatch(o -> o.getStatus().equals(status), values());
+ }
+
+}
\ No newline at end of file
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/enums/rule/IotDataSinkTypeEnum.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/enums/rule/IotDataSinkTypeEnum.java
new file mode 100644
index 0000000000..45a557db61
--- /dev/null
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/enums/rule/IotDataSinkTypeEnum.java
@@ -0,0 +1,42 @@
+package cn.iocoder.yudao.module.iot.enums.rule;
+
+import cn.iocoder.yudao.framework.common.core.ArrayValuable;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+
+import java.util.Arrays;
+
+/**
+ * IoT 数据目的的类型枚举
+ *
+ * @author 芋道源码
+ */
+@RequiredArgsConstructor
+@Getter
+public enum IotDataSinkTypeEnum implements ArrayValuable {
+
+ HTTP(1, "HTTP"),
+ TCP(2, "TCP"), // TODO @puhui999：待实现；
+ WEBSOCKET(3, "WebSocket"), // TODO @puhui999：待实现；
+
+ MQTT(10, "MQTT"), // TODO 待实现；
+
+ DATABASE(20, "Database"), // TODO @puhui999：待实现；可以简单点，对应的表名是什么，字段先固定了。
+ REDIS(21, "Redis"),
+
+ ROCKETMQ(30, "RocketMQ"),
+ RABBITMQ(31, "RabbitMQ"),
+ KAFKA(32, "Kafka");
+
+ private final Integer type;
+
+ private final String name;
+
+ public static final Integer[] ARRAYS = Arrays.stream(values()).map(IotDataSinkTypeEnum::getType).toArray(Integer[]::new);
+
+ @Override
+ public Integer[] array() {
+ return ARRAYS;
+ }
+
+}
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/enums/rule/IotSceneRuleActionTypeEnum.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/enums/rule/IotSceneRuleActionTypeEnum.java
new file mode 100644
index 0000000000..7e9e4de631
--- /dev/null
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/enums/rule/IotSceneRuleActionTypeEnum.java
@@ -0,0 +1,52 @@
+package cn.iocoder.yudao.module.iot.enums.rule;
+
+import cn.iocoder.yudao.framework.common.core.ArrayValuable;
+import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+
+import java.util.Arrays;
+
+/**
+ * IoT 规则场景的触发类型枚举
+ *
+ * 设备触发，定时触发
+ */
+@RequiredArgsConstructor
+@Getter
+public enum IotSceneRuleActionTypeEnum implements ArrayValuable {
+
+ /**
+ * 设备属性设置
+ *
+ * 对应 {@link IotDeviceMessageMethodEnum#PROPERTY_SET}
+ */
+ DEVICE_PROPERTY_SET(1),
+ /**
+ * 设备服务调用
+ *
+ * 对应 {@link IotDeviceMessageMethodEnum#SERVICE_INVOKE}
+ */
+ DEVICE_SERVICE_INVOKE(2),
+
+ /**
+ * 告警触发
+ */
+ ALERT_TRIGGER(100),
+ /**
+ * 告警恢复
+ */
+ ALERT_RECOVER(101),
+
+ ;
+
+ private final Integer type;
+
+ public static final Integer[] ARRAYS = Arrays.stream(values()).map(IotSceneRuleActionTypeEnum::getType).toArray(Integer[]::new);
+
+ @Override
+ public Integer[] array() {
+ return ARRAYS;
+ }
+
+}
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/enums/rule/IotSceneRuleConditionTypeEnum.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/enums/rule/IotSceneRuleConditionTypeEnum.java
new file mode 100644
index 0000000000..81d7e6e1f5
--- /dev/null
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/enums/rule/IotSceneRuleConditionTypeEnum.java
@@ -0,0 +1,40 @@
+package cn.iocoder.yudao.module.iot.enums.rule;
+
+import cn.hutool.core.util.ArrayUtil;
+import cn.iocoder.yudao.framework.common.core.ArrayValuable;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+
+import java.util.Arrays;
+
+/**
+ * IoT 条件类型枚举
+ *
+ * @author 芋道源码
+ */
+@RequiredArgsConstructor
+@Getter
+public enum IotSceneRuleConditionTypeEnum implements ArrayValuable {
+
+ DEVICE_STATE(1, "设备状态"),
+ DEVICE_PROPERTY(2, "设备属性"),
+
+ CURRENT_TIME(100, "当前时间"),
+
+ ;
+
+ private final Integer type;
+ private final String name;
+
+ public static final Integer[] ARRAYS = Arrays.stream(values()).map(IotSceneRuleConditionTypeEnum::getType).toArray(Integer[]::new);
+
+ @Override
+ public Integer[] array() {
+ return ARRAYS;
+ }
+
+ public static IotSceneRuleConditionTypeEnum typeOf(Integer type) {
+ return ArrayUtil.firstMatch(item -> item.getType().equals(type), values());
+ }
+
+}
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/enums/rule/IotSceneRuleTriggerTypeEnum.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/enums/rule/IotSceneRuleTriggerTypeEnum.java
new file mode 100644
index 0000000000..bfc84c9f60
--- /dev/null
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/enums/rule/IotSceneRuleTriggerTypeEnum.java
@@ -0,0 +1,68 @@
+package cn.iocoder.yudao.module.iot.enums.rule;
+
+import cn.hutool.core.util.ArrayUtil;
+import cn.iocoder.yudao.framework.common.core.ArrayValuable;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+
+import java.util.Arrays;
+
+/**
+ * IoT 场景流转的触发类型枚举
+ *
+ * 为什么不直接使用 IotDeviceMessageMethodEnum 呢？
+ * 原因是，物模型属性上报，存在批量上报的情况，不只对应一个 method！！！
+ *
+ * @author 芋道源码
+ */
+@RequiredArgsConstructor
+@Getter
+public enum IotSceneRuleTriggerTypeEnum implements ArrayValuable {
+
+ // TODO @芋艿：后续“对应”部分，要 @下，等包结构梳理完；
+ /**
+ * 设备上下线变更
+ *
+ * 对应 IotDeviceMessageMethodEnum.STATE_UPDATE
+ */
+ DEVICE_STATE_UPDATE(1),
+ /**
+ * 物模型属性上报
+ *
+ * 对应 IotDeviceMessageMethodEnum.DEVICE_PROPERTY_POST
+ */
+ DEVICE_PROPERTY_POST(2),
+ /**
+ * 设备事件上报
+ *
+ * 对应 IotDeviceMessageMethodEnum.DEVICE_EVENT_POST
+ */
+ DEVICE_EVENT_POST(3),
+ /**
+ * 设备服务调用
+ *
+ * 对应 IotDeviceMessageMethodEnum.DEVICE_SERVICE_INVOKE
+ */
+ DEVICE_SERVICE_INVOKE(4),
+
+ /**
+ * 定时触发
+ */
+ TIMER(100)
+
+ ;
+
+ private final Integer type;
+
+ public static final Integer[] ARRAYS = Arrays.stream(values()).map(IotSceneRuleTriggerTypeEnum::getType).toArray(Integer[]::new);
+
+ @Override
+ public Integer[] array() {
+ return ARRAYS;
+ }
+
+ public static IotSceneRuleTriggerTypeEnum typeOf(Integer type) {
+ return ArrayUtil.firstMatch(item -> item.getType().equals(type), values());
+ }
+
+}
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/framework/iot/config/YudaoIotProperties.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/framework/iot/config/YudaoIotProperties.java
new file mode 100644
index 0000000000..07473c0293
--- /dev/null
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/framework/iot/config/YudaoIotProperties.java
@@ -0,0 +1,28 @@
+package cn.iocoder.yudao.module.iot.framework.iot.config;
+
+import lombok.Data;
+import org.springframework.stereotype.Component;
+
+import java.time.Duration;
+
+/**
+ * 芋道 IoT 全局配置类
+ *
+ * @author 芋道源码
+ */
+@Component
+@Data
+public class YudaoIotProperties {
+
+ /**
+ * 设备连接超时时间
+ */
+ private Duration keepAliveTime = Duration.ofMinutes(10);
+ /**
+ * 设备连接超时时间的因子
+ *
+ * 因为设备可能会有网络抖动，所以需要乘以一个因子，避免误判
+ */
+ private double keepAliveFactor = 1.5D;
+
+}
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/framework/iot/package-info.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/framework/iot/package-info.java
new file mode 100644
index 0000000000..0930a1409c
--- /dev/null
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/framework/iot/package-info.java
@@ -0,0 +1,4 @@
+/**
+ * iot 模块的【全局】拓展封装
+ */
+package cn.iocoder.yudao.module.iot.framework.iot;
\ No newline at end of file
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/job/ota/IotOtaUpgradeJob.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/job/ota/IotOtaUpgradeJob.java
new file mode 100644
index 0000000000..8a15c5e7bb
--- /dev/null
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/job/ota/IotOtaUpgradeJob.java
@@ -0,0 +1,86 @@
+package cn.iocoder.yudao.module.iot.job.ota;
+
+import cn.hutool.core.collection.CollUtil;
+import cn.hutool.core.util.StrUtil;
+import cn.iocoder.yudao.framework.quartz.core.handler.JobHandler;
+import cn.iocoder.yudao.framework.tenant.core.job.TenantJob;
+import cn.iocoder.yudao.module.iot.core.enums.IotDeviceStateEnum;
+import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceDO;
+import cn.iocoder.yudao.module.iot.dal.dataobject.ota.IotOtaFirmwareDO;
+import cn.iocoder.yudao.module.iot.dal.dataobject.ota.IotOtaTaskRecordDO;
+import cn.iocoder.yudao.module.iot.enums.ota.IotOtaTaskRecordStatusEnum;
+import cn.iocoder.yudao.module.iot.service.device.IotDeviceService;
+import cn.iocoder.yudao.module.iot.service.ota.IotOtaFirmwareService;
+import cn.iocoder.yudao.module.iot.service.ota.IotOtaTaskRecordService;
+import jakarta.annotation.Resource;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * IoT OTA 升级推送 Job：查询待推送的 OTA 升级记录，并推送给设备
+ *
+ * @author 芋道源码
+ */
+@Component
+@Slf4j
+public class IotOtaUpgradeJob implements JobHandler {
+
+ @Resource
+ private IotOtaTaskRecordService otaTaskRecordService;
+ @Resource
+ private IotOtaFirmwareService otaFirmwareService;
+ @Resource
+ private IotDeviceService deviceService;
+
+ @Override
+ @TenantJob
+ public String execute(String param) throws Exception {
+ // 1. 查询待推送的 OTA 升级记录
+ List records = otaTaskRecordService.getOtaRecordListByStatus(
+ IotOtaTaskRecordStatusEnum.PENDING.getStatus());
+ if (CollUtil.isEmpty(records)) {
+ return null;
+ }
+
+ // TODO 芋艿：可以优化成批量获取 原因是：1. N+1 问题；2. offline 的设备无需查询
+ // 2. 遍历推送记录
+ int successCount = 0;
+ int failureCount = 0;
+ Map otaFirmwares = new HashMap<>();
+ for (IotOtaTaskRecordDO record : records) {
+ try {
+ // 2.1 设备如果不在线，直接跳过
+ IotDeviceDO device = deviceService.getDeviceFromCache(record.getDeviceId());
+ // TODO 芋艿：【优化】当前逻辑跳过了离线的设备，但未充分利用 MQTT 的离线消息能力。
+ // 1. MQTT 协议本身支持持久化会话（Clean Session=false）和 QoS > 0 的消息，允许 broker 为离线设备缓存消息。
+ // 2. 对于 OTA 升级这类非实时性强的任务，即使设备当前离线，也应该可以推送升级指令。设备在下次上线时即可收到。
+ // 3. 后续可以考虑：增加一个“允许离线推送”的选项。如果开启，即使设备状态为 OFFLINE，也应尝试推送消息，依赖 MQTT Broker 的能力进行离线缓存。
+ if (device == null || IotDeviceStateEnum.isNotOnline(device.getState())) {
+ continue;
+ }
+ // 2.2 获取 OTA 固件信息
+ IotOtaFirmwareDO fireware = otaFirmwares.get(record.getFirmwareId());
+ if (fireware == null) {
+ fireware = otaFirmwareService.getOtaFirmware(record.getFirmwareId());
+ otaFirmwares.put(record.getFirmwareId(), fireware);
+ }
+ // 2.3 推送 OTA 升级任务
+ boolean result = otaTaskRecordService.pushOtaTaskRecord(record, fireware, device);
+ if (result) {
+ successCount++;
+ } else {
+ failureCount++;
+ }
+ } catch (Exception e) {
+ failureCount++;
+ log.error("[execute][推送 OTA 升级任务({})发生异常]", record.getId(), e);
+ }
+ }
+ return StrUtil.format("升级任务推送成功：{} 条，送失败：{} 条", successCount, failureCount);
+ }
+
+}
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/job/rule/IotSceneRuleJob.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/job/rule/IotSceneRuleJob.java
new file mode 100644
index 0000000000..9967ccc3b1
--- /dev/null
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/job/rule/IotSceneRuleJob.java
@@ -0,0 +1,58 @@
+package cn.iocoder.yudao.module.iot.job.rule;
+
+import cn.hutool.core.map.MapUtil;
+import cn.iocoder.yudao.module.iot.enums.rule.IotSceneRuleTriggerTypeEnum;
+import cn.iocoder.yudao.module.iot.service.rule.scene.IotSceneRuleService;
+import jakarta.annotation.Resource;
+import lombok.extern.slf4j.Slf4j;
+import org.quartz.JobExecutionContext;
+import org.springframework.scheduling.quartz.QuartzJobBean;
+
+import java.util.Map;
+
+/**
+ * IoT 规则场景 Job，用于执行 {@link IotSceneRuleTriggerTypeEnum#TIMER} 类型的规则场景
+ *
+ * @author 芋道源码
+ */
+@Slf4j
+public class IotSceneRuleJob extends QuartzJobBean {
+
+ /**
+ * JobData Key - 规则场景编号
+ */
+ public static final String JOB_DATA_KEY_RULE_SCENE_ID = "sceneRuleId";
+
+ @Resource
+ private IotSceneRuleService sceneRuleService;
+
+ @Override
+ protected void executeInternal(JobExecutionContext context) {
+ // 获得规则场景编号
+ Long sceneRuleId = context.getMergedJobDataMap().getLong(JOB_DATA_KEY_RULE_SCENE_ID);
+
+ // 执行规则场景
+ sceneRuleService.executeSceneRuleByTimer(sceneRuleId);
+ }
+
+ /**
+ * 创建 JobData Map
+ *
+ * @param sceneRuleId 规则场景编号
+ * @return JobData Map
+ */
+ public static Map buildJobDataMap(Long sceneRuleId) {
+ return MapUtil.of(JOB_DATA_KEY_RULE_SCENE_ID, sceneRuleId);
+ }
+
+ /**
+ * 创建 Job 名字
+ *
+ * @param sceneRuleId 规则场景编号
+ * @return Job 名字
+ */
+ public static String buildJobName(Long sceneRuleId) {
+ return String.format("%s_%d", IotSceneRuleJob.class.getSimpleName(), sceneRuleId);
+ }
+
+}
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/mq/consumer/device/IotDeviceMessageSubscriber.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/mq/consumer/device/IotDeviceMessageSubscriber.java
new file mode 100644
index 0000000000..7e039d0327
--- /dev/null
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/mq/consumer/device/IotDeviceMessageSubscriber.java
@@ -0,0 +1,101 @@
+package cn.iocoder.yudao.module.iot.mq.consumer.device;
+
+import cn.hutool.core.util.ObjectUtil;
+import cn.iocoder.yudao.framework.tenant.core.util.TenantUtils;
+import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
+import cn.iocoder.yudao.module.iot.core.enums.IotDeviceStateEnum;
+import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus;
+import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageSubscriber;
+import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
+import cn.iocoder.yudao.module.iot.core.util.IotDeviceMessageUtils;
+import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceDO;
+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.PostConstruct;
+import jakarta.annotation.Resource;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+import java.time.LocalDateTime;
+import java.util.Objects;
+
+/**
+ * 针对 {@link IotDeviceMessage} 的业务处理器：调用 method 对应的逻辑。例如说：
+ * 1. {@link IotDeviceMessageMethodEnum#PROPERTY_POST} 属性上报时，记录设备属性
+ *
+ * @author alwayssuper
+ */
+@Component
+@Slf4j
+public class IotDeviceMessageSubscriber implements IotMessageSubscriber {
+
+ @Resource
+ private IotDeviceService deviceService;
+ @Resource
+ private IotDevicePropertyService devicePropertyService;
+ @Resource
+ private IotDeviceMessageService deviceMessageService;
+
+ @Resource
+ private IotMessageBus messageBus;
+
+ @PostConstruct
+ public void init() {
+ messageBus.register(this);
+ }
+
+ @Override
+ public String getTopic() {
+ return IotDeviceMessage.MESSAGE_BUS_DEVICE_MESSAGE_TOPIC;
+ }
+
+ @Override
+ public String getGroup() {
+ return "iot_device_message_consumer";
+ }
+
+ @Override
+ public void onMessage(IotDeviceMessage message) {
+ if (!IotDeviceMessageUtils.isUpstreamMessage(message)) {
+ log.error("[onMessage][message({}) 非上行消息，不进行处理]", message);
+ return;
+ }
+
+ TenantUtils.execute(message.getTenantId(), () -> {
+ // 1.1 更新设备的最后时间
+ IotDeviceDO device = deviceService.validateDeviceExistsFromCache(message.getDeviceId());
+ devicePropertyService.updateDeviceReportTimeAsync(device.getId(), LocalDateTime.now());
+ // 1.2 更新设备的连接 server
+ // TODO 芋艿：HTTP 网关的上行消息，不应该更新 serverId，会覆盖掉 MQTT 等长连接的 serverId，导致下行消息无法发送。
+ devicePropertyService.updateDeviceServerIdAsync(device.getId(), message.getServerId());
+
+ // 2. 未上线的设备，强制上线
+ forceDeviceOnline(message, device);
+
+ // 3. 核心：处理消息
+ deviceMessageService.handleUpstreamDeviceMessage(message, device);
+ });
+ }
+
+ private void forceDeviceOnline(IotDeviceMessage message, IotDeviceDO device) {
+ // 已经在线，无需处理
+ if (ObjectUtil.equal(device.getState(), IotDeviceStateEnum.ONLINE.getState())) {
+ return;
+ }
+ // 如果是 STATE 相关的消息，无需处理，不然就重复处理状态了
+ if (Objects.equals(message.getMethod(), IotDeviceMessageMethodEnum.STATE_UPDATE.getMethod())) {
+ return;
+ }
+
+ // 特殊：设备非在线时，主动标记设备为在线
+ // 为什么不直接更新状态呢？因为通过 IotDeviceMessage 可以经过一系列的处理，例如说记录日志、规则引擎等等
+ try {
+ deviceMessageService.sendDeviceMessage(IotDeviceMessage.buildStateUpdateOnline().setDeviceId(device.getId()));
+ } catch (Exception e) {
+ // 注意：即使执行失败，也不影响主流程
+ log.error("[forceDeviceOnline][message({}) device({}) 强制设备上线失败]", message, device, e);
+ }
+ }
+
+}
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/mq/consumer/rule/IotDataRuleMessageHandler.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/mq/consumer/rule/IotDataRuleMessageHandler.java
new file mode 100644
index 0000000000..843592a272
--- /dev/null
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/mq/consumer/rule/IotDataRuleMessageHandler.java
@@ -0,0 +1,48 @@
+package cn.iocoder.yudao.module.iot.mq.consumer.rule;
+
+import cn.iocoder.yudao.framework.tenant.core.util.TenantUtils;
+import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus;
+import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageSubscriber;
+import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
+import cn.iocoder.yudao.module.iot.service.rule.data.IotDataRuleService;
+import jakarta.annotation.PostConstruct;
+import jakarta.annotation.Resource;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+/**
+ * 针对 {@link IotDeviceMessage} 的消费者，处理数据流转
+ *
+ * @author 芋道源码
+ */
+@Component
+@Slf4j
+public class IotDataRuleMessageHandler implements IotMessageSubscriber