智慧水产系统设计指南

版本:v5.4 | 日期:2026-10-24
定位:一份可落地实施的"通用物联网平台 + 智慧水产应用"设计指南;读第 1~9 章即可施工。
结构:必须功能 M1~M7(§4.3)→ 可选 O1~O7(§4.4)→ 实现级规范(第 8 章)→ 验收清单(第 9 章)。
原则:以本文为准;参考文档只作对照(通用提纲 / api-mqtt / 涂鸦米家 / FastBee)。
v5.2:后端二选一,推荐 Node.js + Express + TypeScript,备选 Python + FastAPI(§4.2.2.1);实现级契约见第 8 章,验收见第 9 章。
v5.3:补齐 P1 必须项——全量 API 路由表(§8.11.4)、用户通知偏好、Webhook payload、数据导出/删除、OTA 密钥/回滚、全局限流与租户配额。
v5.4:新增最小经营记录表 pond_operation_log(投喂/用药/巡塘/死亡/出塘,V3),补齐经营数据落点与 API。
配套:【参考】重要在线文档.txt、【参考】api-mqtt【可扩展更多】.md、【参考】通用物联网平台提纲.md、【参考】涂鸦和米家物联网大纲.md、【参考】物平台-FastBee框架讲解.md;上级 01_总规划/总规划书.md、04_软件方案/云平台规划.md。


目录

章节 内容 读者
1 水产行业需求 产品/全员
2 水产行业标准 产品/后端
3 用数据赋能水产行业 产品/后端
4 (最核心)如何设计一个云平台 后端/架构
5 最小闭环与开发路线 全员
6 后期可扩展行业应用 架构/产品
7 与参考平台的对照与借鉴 架构
8 实现级完整规范(审查补差:契约、DDL、接口、前端、工程、测试) 后端/前端/运维
9 验收与实施检查表(逐项可勾选) 全员
附录 报文示例、工程目录、术语表 后端

1、水产行业需求

1.1 行业痛点

(需求总纲)

来自看塘人官网(kantangren.com)与总规划书,按优先级排序:

# 痛点 说明 平台必须提供的能力
P1 停电/缺氧翻塘 凌晨 2–5 点溶氧全天最低,增氧机一停 3~5 点溶氧断崖下跌 DO 实时监测 + 分级报警(微信/短信/电话)+ 增氧机自动控制
P2 凌晨人工巡塘 半夜看浮头,请人贵、自己累、肉眼易误判 手机端随时看数据,替代巡塘
P3 电费居高不下 定时开关不反映塘口实时状态,增氧机"一开到底" DO 四档联动变频调速(RS485 Modbus)
P4 传感器漂移、维护频繁 生物膜附着导致漂移 传感器状态监测、清洗联动、易损件周期提醒
P5 多塘难管 5 个塘以上人工跑不动 多塘集中管理、按塘分组告警、Web 后台
P6 4G 断网是常态 偏远塘口信号不稳,云断=失控 边缘闭环:断网时设备本地阈值控制不停;数据补传

关键:P1/P6 要求控制下沉到设备本地(边缘 DO 闭环);云端只负责"看得见、管得住、算得清"。这是与纯云端方案的本质差异。

1.2 用户角色与场景

角色 场景 核心诉求
养殖户(散户 1~5 塘) 手机看 DO、收报警 简单、不用学、报警必达
养殖户家人 共享设备 看数据、收报警
中小基地(5~20 塘) 多塘集中管理 一屏多塘、按塘分组、历史追溯
大型企业/集团(20+ 塘) 集中监控、数据报表、API 对接 多租户分级、导出报表、开放接口
代理商/集成商 卖设备、装设备、做售后 自有客户隔离、工单、远程诊断
运维/安装师傅 调试、维修 设备调试工具、远程参数、OTA、工单流转
平台运营(我们自己) 设备全生命周期、订阅计费 后台管理、固件灰度、运营数据

多租户模型(对应涂鸦"厂家-家庭-用户",米家"家庭-用户",并对齐通用提纲"厂家 + 一级租户 + 用户"):

平台运营方(我们,platform_admin 跨租户管理)  
 └── 厂家 vendor(产品/固件/自定义UI 的归属方,智慧水产中=泰基或品类 ODM 方)  
      └── 一级租户(代理商 / 大型基地 / 集团子公司)  
           └── 普通用户(养殖户 / 家人共享 / 运维员)  

简化:通用提纲的 tenant + vendor 合并为一棵 tenant 树;vendor 是 level=1 节点,product.vendor_id 指向它。

1.3 功能需求清单

(平台侧)

按必须/应该/可以分级,对应 §4.1:

MVP(必须有)
1. 设备接入:MQTT over TLS、一机一密、上下线感知(M2)
2. 数据看板:DO/水温/设备状态实时展示 + 24h/7d/30d 历史曲线(M4/M5)
3. 告警:DO 低/危险、设备离线、传感器离线、RCD 跳闸、箱内高温,微信推送(M6)
4. 远程控制:增氧机开关、变频调速(手动模式,仅运维角色)(M4)
5. 设备绑定:扫码添加、设备共享给家人(M1/M4)
6. 参数配置:DO 阈值下发(设备本地生效)(M3/M4)

V2(应该有)
7. 多塘管理与设备分组(M4)
8. 订阅计费:试用/基础/标准/旗舰,到期降级(本地控制永不停)(O4)
9. OTA:固件上传、灰度推送、进度追踪、失败回滚(O3)
10. Web 管理后台:设备台账、客户管理、告警统计、操作日志(M1/M4)
11. 远程诊断:信号强度、重启、日志回捞、工单 SLA(O5)
12. 开放 API + Webhook + 数据转发(HTTP/MQTT 必须,Kafka/DB 可选)(M7/M6)

V3(可以有)
13. 拖拽式可视化看板(类简化 Grafana)(O2)
14. 水质报告/投喂建议等数据增值服务(O6)
15. 规则引擎可视化编排、厂家自定义产品页面(O1)

1.4 非功能需求

维度 要求 设计对策
可靠性 云端全挂,塘口控制不中断 边缘闭环(固件本地阈值控制)+ 设备离线兜底告警
规模 起步 100 台,撑到 10,000 台再重构 单机 docker-compose 起步;EMQX 单机 5 万连接富余
数据量 1000 台 × 每分钟 1 条 = 1 年 5.25 亿条 时序库(TDengine)或 PG 按月分区 + 降采样
安全 一机一密、TLS、租户隔离、固件签名 EMQX HTTP 认证回调 + JWT + tenant_id 强制过滤
成本 起步月费 ≤ ¥800 自建 EMQX/TDengine(¥0)+ 小规格 RDS
响应 报警从触发到手机 ≤ 60s MQTT 共享订阅消费 + 规则引擎内存匹配 + 微信模板消息

2、水产行业标准

2.1 溶氧安全阈值

(核心业务标准)

行业通用参考区间(以品种、放养密度、水温为准,平台内可按塘口覆盖):

养殖品种 适宜溶氧 警戒线 危险线(泛塘风险)
四大家鱼(青/草/鲢/鳙) ≥ 5 mg/L ≤ 3 mg/L < 2 mg/L
罗非鱼 ≥ 5 mg/L ≤ 3 mg/L < 1.5 mg/L
鳜鱼 / 鲈鱼(肉食性) ≥ 5~6 mg/L ≤ 3 mg/L < 2 mg/L
南美白对虾 ≥ 5 mg/L ≤ 3 mg/L < 2 mg/L(最敏感)
河蟹 / 小龙虾 ≥ 5 mg/L ≤ 3 mg/L < 2 mg/L
黄颡鱼 / 鲫鱼 ≥ 5 mg/L ≤ 3 mg/L < 2 mg/L

落地:做成 threshold_template;建塘选品种自动填充,用户可覆盖。

2.2 溶氧-增氧控制策略标准

(四档映射)

档位 条件 动作
停机档 DO ≥ 上限(默认 6.0) 停机或最低频 15Hz
维持档 下限 < DO < 上限(4.0~6.0) 维持频率 25~35Hz
加速档 危险 < DO ≤ 下限(2.0~4.0) 全速 50Hz
报警档 DO ≤ 危险(≤ 2.0) 全速 + 告警推送

配套保护规则(固件本地执行,平台只做展示与参数下发):

固化原因:这些数字是行业 know-how,沉淀为物模型默认参数 + 告警规则模板,新装设备开箱即用。

2.3 相关国家标准与行业规范

标准 内容 对平台的约束
GB 11607-89《渔业水质标准》 溶氧:连续 24h 中 16h 以上 >5mg/L,其余任何时候 ≥3mg/L(鲑科冰封期 ≥4mg/L) 阈值模板的法规底线
SC/T 9101《淡水池塘养殖水水质要求》 池塘水质要求 数据集字段设计参照(pH/氨氮/亚硝酸盐等)
NY 5051《无公害食品 淡水养殖用水水质》(及其修订版) 养殖用水水质 数据增值报告的依据
GB/T 19582《基于 Modbus 协议的工业自动化网络规范》 Modbus RTU 传感器/变频器接入协议
OASIS MQTT 3.1.1/5.0 设备通信协议 EMQX 接入层
GB/T 3805 / IEC 61140 SELV 低压安全 DC48V 路线(硬件侧,平台仅展示状态)

注:标准会修订,以最新版为准;数据模型保留 standard_ref,报告必须引用。

2.4 平台数据标准 = 物模型标准

核心主张:行业标准落地为物模型模板 + 阈值模板 + 告警规则模板三件套,全部以数据存平台(不是代码):

厂家/代理商复用模板开箱即用;头部客户可在模板上自定义。


3、用数据赋能水产行业

目标:一个平台管全流程——数据集、功能点、行业标准、业务过程全数据化,降本增效。

3.1 数据集设计

(平台沉淀的九类核心数据)

# 数据集 内容 来源 用途
D1 塘口档案 位置/面积/水深/品种/密度/投放日期 用户录入 报警责任归属、数据增值
D2 水质时序 DO/水温/pH/氨氮/盐度… 设备上报(物模型属性) 曲线、报告、阈值优化
D3 设备运行 通道状态/运行时长/电流/频率/启停次数 物模型属性 能耗分析、预测性运维
D4 告警记录 类型/等级/触发值/处理结果/耗时 规则引擎 售后 SLA、事故追溯
D5 阈值/规则配置 品种阈值、四档参数、告警规则 模板+用户覆盖 行业标准数字化
D6 养殖批次 放苗→投喂→用药→出塘全程记录 用户+设备 产量追溯、成本核算
D7 能耗数据 电量/变频节能对比 电表/估算 节能收益证明(回本测算)
D8 运维工单 报修/派单/处理/回访 售后系统 代理商考核、故障率统计
D9 订阅交易 套餐/订单/支付/到期 计费模块 经营数据

3.2 功能点 → 数据映射

(保证"每个功能都有数据支撑")

功能点 输入数据集 输出数据集
DO 看板 D2 D2(聚合)
自动增氧联动 D2 + D5 D3(动作日志)
分级报警 D2 + D5 D4
历史曲线 D2 —
节能报告 D3 + D7 + D5 D7(报告)
水质报告 D1 + D2 + D6(引用标准号) 报告文档
传感器维护提醒 D3(运行时长)+ D5(易损件周期) D4(提醒)
事故追溯 D2 + D3 + D4 + D5 时间线报告
经营看板 D4 + D8 + D9 运营报表

3.3 业务过程全数据化原则

  1. 能自动记录的不靠人填:水质/设备/告警由物模型上报;
  2. 必须人填的做成模板:投喂/用药/巡塘用移动端快捷表单(V3,落 pond_operation_log);
  3. 数据带 tenant_id + pond_id + batch_id 三级归属,可聚合、可追溯;
  4. 先存原始值,聚合后置:原始 90 天 → 小时 365 天 → 天长期,策略配置化。

3.4 数据如何"赋能"


4、(最核心)如何设计一个云平台

基座是通用物联网平台。验收标准:动态物模型定义产品、属性可跨设备/产品联动、统一规则引擎、登录/接入/存储/开放 API。
涂鸦/米家逻辑:产品定义驱动接入,规则引擎打通数据与动作,多租户隔离使用者。

4.0 实现路线抉择

对照点 路线 A:JSONB 文档式(本文主推) 路线 B:表驱动 EAV(FastBee 式)
物模型存储 thing_model 表整体存 JSONB,一行一产品 iot_things_model 一行一属性/功能/事件
动态性 前端定义完直接落库,改字段零迁移 加属性即加行,查询维度多
开发量 小(解释器一处) 中(行级 CRUD + 拼装)
适合 起步期、快速迭代、物模型结构相对稳定 管理维度要求极多(排序/分享权限/图表标记等)

决定:路线 A 起步。创业公司要快,JSONB 对动态物模型最友好;吸收 FastBee 标签维度(is_chart/is_monitor/is_history/is_readonly/model_order),量大再迁表驱动(§4.5)。

4.1 平台功能定义总表

(必须在前,可选在后)

必须功能(P0,缺一不可,对应 §4.3 各节):

编号 模块 功能定义(一句话) 核心数据结构
M1 登录与账号 多租户三级体系 + JWT/RBAC + 设备绑定共享 + 操作审计 tenant, user, audit_log
M2 通信接入层 MQTT 接入(api-mqtt 报文协议)+ 一机一密认证 + LWT 上下线感知 + 消息消费 + 命令/配置闭环 + 断网补传 Topic 规范(up/cmd+type), device 凭据字段, Redis 在线键
M3 产品与物模型 动态创建产品 + 属性/服务/事件三要素 + 扁平产品分组(一级,无树) + 模板 + 版本管理 + 设备授权 product, product_group, thing_model, model_template, threshold_template
M4 设备管理 设备全生命周期 + 批量导入 + 扫码绑定/共享 + 分组 + 网关子设备拓扑 + 影子 + 日志 + 在线调试 + 默认组件展示 device, device_state, device_log, device_group, pond/farm/batch
M5 数据存储与性能优化 最新状态缓存 + 历史落库 + 降采样聚合 + 保留策略 property_history, Redis 键设计
M6 规则引擎与告警 条件→动作 + 跨属性联动 + 数据转发 + 冷却 + 分级告警闭环 + 场景联动 rule, alarm_record, data_forward, 规则 JSON Schema
M7 开放第三方接口 AccessKey 鉴权 + 只读优先 API + Webhook 推送 access_key, API 清单

可选功能(P1/P2,按需启用,对应 §4.4 各节):

编号 模块 功能定义(一句话) 核心数据结构
O1 静态文件显示系统 厂家自定义单页 HTML 替换默认展示 + 沙箱桥接 product_ui, UI 目录树
O2 可视化看板 拖拽式看板(简化版 Grafana) dashboard
O3 OTA 固件升级 固件上传 + 灰度推送 + 回滚 + BLE 备通道 firmware, ota_task
O4 订阅计费 套餐 + 微信支付 + 到期降级 subscription, order
O5 工单与远程运维 报修派单 + SLA + 远程诊断 work_order
O6 数据增值服务 水质/节能/投喂报告 + 投喂/用药/巡塘/死亡/出塘记录(V3) 报告表 + pond_operation_log
O7 能力扩展 视频监控、能耗采集、AI 识别等 预留扩展位

判定原则:缺了它设备就"看不见、管不住、连不上、控不了"的,是必须;提升体验/商业化/规模化的,是可选。O1 的"默认公共组件"属于 M4(必须),O1 本身只管"自定义 HTML 替换"这一增值能力。

4.1.1 基本功能自查清单(对照《通用物联网平台提纲》速览 + api-mqtt)

实施完成后逐行验收,保证基本功能一个不少;"可选"项按客户/项目需要启用。这是本文作为"完整物联网平台"的验收底线。

基础能力域 功能点 必/可选 本文落点
设备接入 MQTT 一机一密、TLS、QoS1 去重、LWT/重连退避 必须 M2
设备接入 命令闭环 cmd_result、config 配置下发(cfg_ver)、离线补传 必须 M2 / 附录A/B
设备接入 网关与子设备拓扑(网关代理上下线、子设备清单) 必须 M2 / M4
设备接入 一型一密/动态注册、HTTP 直连上报、协议包热加载 可选 M2 / O7
设备管理 注册/激活/批量导入、生命周期状态、禁用删除 必须 M4
设备管理 设备影子(desired/reported)、实时控制、数据监测 必须 M4 / M5
设备管理 设备分组、设备分享(只读/可控)、设备日志、在线调试 必须 M4
设备管理 设备定位(经纬度/地图分布) 可选 M4
物模型 三要素、动态校验、版本管理、免代码生成 必须 M3
产品管理 产品创建/发布、产品分组(一级,无树)、设备授权(secret 重置/作废) 必须 M3
产品管理 固件管理与 OTA(灰度/回滚/BLE 备通道) 可选 O3
规则引擎 属性/事件/定时触发、跨属性联动、场景联动(串/并行) 必须 M6
规则引擎 下发/通知/虚拟属性动作、告警中心(级别/确认/恢复闭环) 必须 M6
规则引擎 数据转发(HTTP/MQTT 推送必须;Kafka/数据库直写可选) 必须 + 可选 M6 / M7
数据存储 元数据/最新值/历史三层分离、降采样、多库分工演进 必须 M5
多租户安全 厂家+租户+用户隔离、RBAC、tenant_id 强制过滤、凭据只存哈希 必须 M1 / M2
可视化 默认公共组件(开箱即显示) 必须 M4
可视化 自定义 HTML、轻量看板、大屏/组态/报表 可选 O1 / O2 / O6
日志审计 操作日志 + 设备日志双轨 必须 M1 / M4
开放能力 开放 API(AccessKey)、Webhook 推送、数据转发 必须 M7 / M6
通知 站内信/Webhook/短信/邮件(微信模板消息/电话为水产扩展) 必须 M6
平台监控 EMQX 连接数/消息吞吐/worker 负载/规则执行成功率 可选 O5

4.2 总体架构与技术选型

4.2.1 总体架构:五层 + 一个静态文件子系统

设备 / 网关(4G 控制器 ESP32-S3 + ML307A/EC600N,RS485 传感器与变频器)  
    │  MQTT over TLS(一机一密)  
    ▼  
接入层    EMQX Broker(开源版)  
    │  共享订阅消费上行 / HTTP 认证回调  
    ▼  
核心层    应用服务(Node.js+Express 或 Python FastAPI,二选一;推荐 Node.js+Express)
          (物模型解析 · 设备生命周期 · 规则引擎 · 告警)  
    │  
    ├── 数据层    PostgreSQL(元数据,JSONB)+ Redis(最新状态/在线标记)  
    │             + TDengine 或 PG 按月分区(历史时序,可后置)  
    │  
    ├── 静态文件子系统   默认组件库 · 厂家自定义 HTML · 租户隔离存储  
    │  
    ▼  
应用层    多租户(平台+一级租户+用户) / 权限 / 产品管理 / 设备管理  
          / 可视化 / 日志审计 / OTA / 订阅计费 / 开放 API  

关键设计原则(配置与数据分离):产品定义、物模型、规则、权限存关系库;设备上报的属性值走时序/缓存。二者通过 product_key / device_id 关联,绝不混在一张表里。

4.2.2 技术选型

层 选型 理由 备选
核心服务 Node.js + Express + TypeScript(推荐) 或 Python FastAPI(备选) 二选一,不并存;Node 与前端同栈、单人上手快、I/O/WebSocket 够用、与现有云平台规划一致;FastAPI 的 Pydantic 物模型 JSON 校验更自然 另一方案;同一项目只能选一套
MQTT Broker EMQX 开源版 单机 5 万连接、WebSocket 直连小程序、HTTP 认证/共享订阅齐全 自研 Netty Broker(FastBee 路线,不建议自研)
元数据库 PostgreSQL(JSONB) 动态物模型最友好;阿里云 RDS 免运维 MySQL 8.0(JSON 列,功能弱一点)
缓存/状态 Redis 设备最新状态、在线标记(TTL 判离线)、JWT 黑名单 —
时序存储 起步 PG 按月分区;>500 台上 TDengine 10 倍压缩、标准 SQL、免运维 InfluxDB、IoTDB
消息/解耦 起步进程内队列;>5000 台上 EMQX Rule Engine → Kafka 避免早期过度设计 RocketMQ
前端 Web Vue3 + Element Plus 生态成熟,后台模板多 React + AntD
移动端 微信小程序(养殖户)+ Web 自适应 见 04_软件方案/微信小程序规划.md Uniapp 跨端
对象存储 阿里云 OSS 固件、自定义 UI 包、报告文件 MinIO 自建

4.2.2.1 后端语言:二选一,推荐 Node.js + Express

结论:推荐 Node.js + Express + TypeScript;Python + FastAPI 备选。两者都能实现第 8 章契约,只影响工程与库。
不引入 NestJS/Fastify:单人维护 + 写硬件 C,抽象越少越快;Express 5 当前规模足够。

维度 Node.js + Express + TypeScript(推荐) Python + FastAPI(备选)
与现有基础 已在用 Node.js/Express,04_软件方案/云平台规划.md 也选 Node.js 需团队具备 Python 异步/类型经验
前后端同栈 Vue3/小程序 + TS,共享物模型/规则类型(Zod) 前后端两套语言,类型靠 OpenAPI 同步
并发模型 Express 5 事件循环 + PM2/cluster 多实例;I/O 密集够用 asyncio + uvicorn 多 worker;I/O 密集同样够用
CPU 密集 规则聚合/报表放 BullMQ worker/worker_threads,不阻塞主循环 放独立进程/Celery,规避 GIL
物模型校验 Zod(固定 DTO)+ Ajv(动态 JSON Schema) Pydantic 更自然,JSON Schema 生态成熟
WebSocket ws / socket.io 生态成熟,小程序/网页都方便 FastAPI WebSocket 可用,生态略弱
任务调度 BullMQ(Redis)/ node-cron APScheduler
ORM/迁移 Prisma 或 Drizzle + Prisma Migrate/Drizzle Kit SQLAlchemy + Alembic
工程风格 轻量、自由、路由+中间件,单人上手最快 约定更强,DI/Pydantic 自带结构
测试 Vitest/Jest + Supertest pytest + httpx
AI/数据分析 通过独立 Python 服务/API 调用 同进程更方便
主要风险 自由度高,需自己立目录/错误码/校验规范 异步生态碎片化;需注意同步库阻塞事件循环
适合 单人/小团队、快速交付、TS 全栈、已有 Express 经验 Python 团队、AI/数据分析、Schema 复杂

推荐理由(按权重):

  1. 单人 + 硬件 C:Express 只有路由 + 中间件,文件少、抽象少、排错直;NestJS 的模块/DI 反而拖慢。
  2. 现有基础 + 同栈:云平台规划已选 Node,你也用过 Express;Vue3/小程序/后端全 TS,Zod 一套 Schema 前后端共用。
  3. 性能够:EMQX 承担连接,后端只做共享订阅消费;瓶颈在 DB/Redis/规则。Express + PM2/cluster + BullMQ 足够。
  4. FastAPI:Python 团队、AI/数据分析、Schema 极复杂时选;同一项目只选一套。

并发落地铁律(选 Node/Express 时):

两套实现的模块映射(第 8 章契约不变):

能力 Node.js + Express(推荐) Python + FastAPI(备选)
HTTP/中间件 Express 5 + 自写 auth/tenant/permission 中间件 FastAPI + Depends
校验 Zod(DTO)+ Ajv(物模型 JSON Schema) Pydantic
ORM/迁移 Prisma/Drizzle + Prisma Migrate/Drizzle Kit SQLAlchemy + Alembic
MQTT mqtt.js aiomqtt/paho-mqtt
Redis ioredis redis-py
队列/调度 BullMQ + node-cron Celery/APScheduler
WebSocket ws / socket.io FastAPI WebSocket
JWT jsonwebtoken / passport-jwt python-jose
日志/追踪 Pino + OpenTelemetry structlog + OpenTelemetry
API 文档 swagger-jsdoc / zod-to-openapi 内置 OpenAPI
测试 Vitest/Jest + Supertest pytest + httpx
文件上传 multer python-multipart
微信支付 wechatpay-node-v3 wechatpy

同一项目只选一套;第 8 章所有契约(表结构、Topic、命令闭环、错误码、验收)与语言无关。

成本基线(与云平台规划一致):起步 ¥650~800/月(ECS 2C4G + RDS PG 1C2G + Redis 1G + OSS + 短信按量)。不选阿里云 IoT 托管:按消息计费 1000 台月费 ¥2000+,且绑死云厂商。


4.3 平台必须功能设计

(M1~M7)

M1 登录与账号(多租户 RBAC)

【功能定义】

  1. 租户管理:四级租户树(平台→厂家 vendor→一级租户→用户组)的创建/停用/过户;
  2. 用户认证:微信小程序 wx.login 换 openid 登录;手机号+密码登录(后台);JWT 签发与刷新;
  3. 角色权限(租户内 RBAC,四级角色,见下表);
  4. 设备绑定与共享:用户与设备的归属/共享关系;
  5. 操作审计:所有写操作记录到 audit_log。
角色 权限
platform_admin(平台管理员) 跨租户:租户管理、产品模板、固件、全局监控
tenant_admin(一级租户管理员) 本租户:产品创建、物模型编辑、UI 模板上传、成员与设备分配、规则配置
member(普通用户/养殖户) 绑定设备的查看/告警/共享,无控制(控制留给自动逻辑与运维)
operator(运维员) 分配到的设备:远程控制、参数修改、OTA、日志

【数据结构】

CREATE TABLE tenant (  
  id         BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,  
  parent_id  BIGINT DEFAULT 0,  
  name       VARCHAR(64) NOT NULL,  
  level      SMALLINT NOT NULL DEFAULT 1,  -- 0平台 1厂家(vendor) 2一级租户 3用户组  
  status     SMALLINT NOT NULL DEFAULT 1,  
  created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP  
);  

CREATE TABLE "user" (  
  id            BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,  
  tenant_id     BIGINT NOT NULL REFERENCES tenant(id),  
  openid        VARCHAR(64) UNIQUE,        -- 微信登录  
  phone         VARCHAR(20) UNIQUE,  
  password_hash VARCHAR(128),  
  role          VARCHAR(20) NOT NULL DEFAULT 'member',  
  status        SMALLINT NOT NULL DEFAULT 1,  
  created_at    TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP  
);  
CREATE INDEX idx_user_tenant ON "user" (tenant_id);  

-- 设备绑定/共享关系(谁拥有/谁可见)  
CREATE TABLE device_bind (  
  id         BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,  
  device_id  BIGINT NOT NULL,  
  user_id    BIGINT NOT NULL,  
  bind_type  SMALLINT NOT NULL,              -- 1拥有 2共享只读 3共享可控 4运维授权(存量 3=运维的迁移为 4,见 §8.2.4)  
  created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,  
  UNIQUE (device_id, user_id)  
);  

CREATE TABLE audit_log (  
  id          BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,  
  tenant_id   BIGINT NOT NULL,  
  user_id     BIGINT,  
  action      VARCHAR(64) NOT NULL,          -- 如 thing_model.publish  
  target_type VARCHAR(32),                   -- product/device/rule/ui/...  
  target_id   BIGINT,  
  detail      JSONB,  
  created_at  TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP  
);  
CREATE INDEX idx_audit_tenant_time ON audit_log (tenant_id, created_at DESC);  

【关键设计】

M2 通信接入层(MQTT + api-mqtt 报文协议 + 一机一密)

【功能定义】

  1. MQTT Broker 接入:MQTT 3.1.1、QoS1、Keepalive 60s;生产环境 TLS(8883),开发可 1883;WebSocket(8083)供小程序直连;
  2. 设备认证:一机一密,EMQX HTTP 认证回调平台校验;鉴权主体是网关(client_id = 网关 dev_id,其代理的全部子设备共用该连接);
  3. 报文协议:直接采用 api-mqtt v1.2;主题固定 up/cmd,type 区分 status/data/event/cmd_result/config/log/ota/err;未知 type 记录不丢弃;
  4. 命令与配置闭环:下发 cmd/config → 设备回 cmd_result(request_id 原样回传 + success/message/code 码表),超时按同 seq 幂等重试;配置持久化并上报 cfg_ver,平台比对不一致即补下发;
  5. 上下线感知:keepalive 超时(3 周期判离线)+ 遗嘱 LWT(异常掉线自动报 offline)+ CONNECT/DISCONNECT Hook → 设备状态变更;
  6. 上行消费:共享订阅 aqua/+/up,多 worker 负载均衡,按 type 分派处理、按 product_id 路由品类解析器;
  7. 断网补传接收:设备离线缓存(Flash 环形缓冲 ≥10000 条)恢复后带 cache:true + collect_ts 补传,平台按 (sn, collect_ts) 归档去重;
  8. 接入扩展(可选):一型一密动态注册(同型号预烧 product_secret,首连调平台动态注册接口换一机一密凭据,便于量产)、HTTP 直连上报、TCP/CoAP/Modbus 协议包热加载(参考 FastBee protocol 机制,不改主流程)。

【数据结构】

Topic 规范(api-mqtt 三段式 <域>/<sn.product_id>/<动作>,动作固定 up/cmd,业务类型用 type 扩展):

上行  aqua/<sn.product_id>/up     设备→平台:status / data / event / cmd_result / log ...(type 区分)  
下行  aqua/<sn.product_id>/cmd    平台→设备:cmd / config / ota ...(type 区分)  
通配  aqua/+/up、aqua/+/cmd       平台侧共享订阅收全量  

口径注:api-mqtt 原文把品类字段写作 product_id;本指南统一称 product_key,二者是同一个字符串。Topic 中间段实际值必须等于 device.sn + "." + product.product_key,不得同时存在两个不同值(详见 §8.1.1)。

平台概念 ↔ 协议身份对照(核心映射,平台内部模型不变):

平台概念(本指南) 协议概念(api-mqtt) 说明
product_key(产品) product_id(品类) 同一字符串,必须完全相等(如 rly4p3、wq8in1、gw4g);决定字段字典/物模型,禁止另造 pk_* 内部键作为主题段
device_name(设备) sn(设备序列号,全局唯一) 品类下的具体设备实例
—(连接层) client_id = 网关 dev_id(MAC Hex + .gw4g) 网关级身份,一网关一连接,子设备共用
物模型属性 data 报文业务字段(品类字典) 对照表见 M3
物模型服务 cmd/config 的 params 与字典可写(W)字段 只读字段拒绝下发
物模型事件 event 报文 + 事件码表 事件码见 api-mqtt §11.4
子设备 gw4g.devices 声明的从设备 拓扑管理见 M4

公共字段:sn(完整标识)、timestamp(ISO8601)、msg_id/seq(去重/命令,0~65535 环形)、source(app/cloud/auto)、collect_ts/cache(补传)。

【关键设计】

  1. 平台侧不实现 MQTT 协议栈,只做"认证回调 + 数据消费"(与 FastBee 内置 Netty Broker 对照:外包给 EMQX,少一个自研组件,多一份社区保障);
  2. 一机一密流程:创建设备时生成 device_secret(仅创建时返回一次)→ 出厂烧录品类凭据(或蓝牙配网写入)→ 设备(网关)CONNECT 携带 clientId / username / password → EMQX 回调 /internal/mqtt/auth → 平台校验哈希 → allow/deny;未登记设备拒绝接入;
  3. 安全演进五步(api-mqtt §9,按上线节奏分步实施):匿名(现状)→ 一机一密 + 命令白名单/频率限制 → TLS 8883 → 主题 ACL(每网关仅读写自己的 aqua/<...>/#,子设备级隔离由网关内部按 sn 二次管控)→ 防重放(msg_id 去重 + seq 校验)→ 数据脱敏(位置/ICCID/SIM);
  4. 生命周期:网关上电报 gw4g status → 子设备逐个 status → 周期 data 心跳 → 正常下线 status(false)+DISCONNECT;异常掉线 LWT 标记;重连 5s→60s+抖动,恢复后补状态→补传→校时;
  5. 命令闭环:平台发 seq/request_id/timeout_ms → 设备校验/仲裁(code=6=本地安全拒绝)→ 执行 → cmd_result → 超时幂等重试或标失败;
  6. 与云平台规划原方案的差异:原方案用 X.509 证书,起步一机一密更简单(免 PKI 运维),后期按安全演进表升级双向 TLS;
  7. 租户隔离仍在服务层强制:Topic 不带 tenant 前缀(设备只知道品类凭据,且租户归属会随过户/转代理变化),各模块 Repository 强制注入 tenant_id;若需租户级 Topic 隔离,可平滑加 /t/{tenant_id} 前缀 + EMQX ACL。

M3 产品与物模型(动态定义的核心抽象)

【功能定义】

  1. 产品管理:创建/编辑/发布产品,生成全局唯一 product_key(简化可用 product_a / product_b);产品是设备的配置模板,承载物模型;
  2. 产品分组:租户下一级扁平分组(product_group,只有组号/名称,无父子、无树),仅用于检索与归类;不建分类树;
  3. 物模型三要素可视化定义:属性(Property)/ 服务(Service)/ 事件(Event);
  4. 物模型模板:按品类预置标准模板(§2.4 三件套之一);
  5. 品种阈值模板:按养殖品种预置默认阈值(§2.1 落地);
  6. 版本管理:草稿 → 发布生成新 version,已绑定设备按其固件声明版本解析;
  7. 设备授权:device_secret 的生成/重置/作废(凭据只存哈希,支持授权码下发);
  8. 在线模拟:发布前用虚拟设备模拟上报,验证定义正确性。

术语:本文"品类"= api-mqtt product_id(产品类型/字段字典),不是分类树;产品归类只用 product_group(一级扁平)。device_group 可保留 parent_id 做设备空间层级,二者不要混。

【数据结构】

CREATE TABLE product (  
  id          BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,  
  tenant_id   BIGINT NOT NULL,  
  vendor_id   BIGINT,                        -- 厂家(tenant.level=1),固件/自定义UI 归属  
  group_id    BIGINT,                        -- 产品分组(product_group.id;一级扁平,可空)  
  product_key VARCHAR(32) NOT NULL UNIQUE,   -- **对外协议品类标识,必须等于 api-mqtt product_id**(如 rly4p3 / product_a / product_b);全局唯一  
  model_no    VARCHAR(64),                    -- 内部 SKU / 型号,便于销售与固件管理;不参与 MQTT 主题  
  name        VARCHAR(64) NOT NULL,  
  product_type VARCHAR(32) NOT NULL,         -- do_controller / water_sensor / vfd / gateway;仅类型标签,不是分类树  
  device_type SMALLINT NOT NULL DEFAULT 1,   -- 1直连 2网关子设备  
  status      SMALLINT NOT NULL DEFAULT 0,   -- 0开发 1发布  
  created_at  TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP  
);  

-- 产品分组:一级扁平,只有组号/名称,无 parent_id、无树  
CREATE TABLE product_group (  
  id         BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,  
  tenant_id  BIGINT NOT NULL,  
  code       VARCHAR(32) NOT NULL,           -- 组号,如 A / B / water / control  
  name       VARCHAR(64) NOT NULL,  
  sort_order INT NOT NULL DEFAULT 0,  
  status     SMALLINT NOT NULL DEFAULT 1,  
  UNIQUE (tenant_id, code)  
);  

CREATE TABLE thing_model (  
  id         BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,  
  product_id BIGINT NOT NULL REFERENCES product(id),  
  version    VARCHAR(16) NOT NULL DEFAULT '1.0.0',  
  properties JSONB NOT NULL DEFAULT '[]',  
  services   JSONB NOT NULL DEFAULT '[]',  
  events     JSONB NOT NULL DEFAULT '[]',  
  status     SMALLINT NOT NULL DEFAULT 0,    -- 0草稿 1已发布  
  created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,  
  UNIQUE (product_id, version)  
);  

-- 物模型模板(品类级,平台预置,租户可复制;group_code 为扁平分组标签)  
CREATE TABLE model_template (  
  id         BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,  
  group_code VARCHAR(32) NOT NULL,           -- 扁平分组,如 control / water / energy  
  name       VARCHAR(64) NOT NULL,  
  model      JSONB NOT NULL,                 -- properties/services/events 整体  
  version    VARCHAR(16) NOT NULL DEFAULT '1.0.0'  
);  

-- 品种阈值模板(§2.1 落地:品种 → 默认阈值 + 四档参数 + 易损件周期)  
CREATE TABLE threshold_template (  
  id           BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,  
  species      VARCHAR(32) NOT NULL UNIQUE,  -- south_american_shrimp / four_major_carp ...  
  items        JSONB NOT NULL,               -- {"do_suit":5,"do_warn":3,"do_danger":2,"standard_ref":"GB 11607-89",...}  
  standard_ref VARCHAR(64)                   -- 数据依据标准号(§2.3)  
);  

三要素 JSON 结构(物模型的"数据结构"本体):

// properties[] 元素  
{  
  "identifier": "oxygen",              // 产品内唯一,程序引用;默认 = api-mqtt 报文字段名  
  "source_field": "oxygen",            // 协议报文字段映射;默认与 identifier 相同,历史改名/别名时才不同  
  "name": "溶解氧",  
  "data_type": "float",                // int/float/bool/enum/string/array/object  
  "access": "r",                       // r只读 / rw读写 / report_only只上报  
  "unit": "mg/L",  
  "min": 0, "max": 20, "step": 0.1, "precision": 1,  
  "is_chart": true,                    // 是否默认画曲线(吸收FastBee标签维度)  
  "is_history": true,                  // 是否存历史  
  "is_alarm": false                    // 是否参与告警监测  
}  
// services[] 元素  
{  
  "identifier": "set_do_threshold",  
  "name": "设置DO阈值",  
  "input_params":  [ {"identifier":"low","data_type":"float","min":0,"max":20},  
                     {"identifier":"danger","data_type":"float","min":0,"max":20} ],  
  "output_params": []  
}  
// events[] 元素  
{  
  "identifier": "do_danger",  
  "name": "溶氧危险告警",  
  "level": 3,  
  "output_params": [ {"identifier":"value","data_type":"float"} ]  
}  

【关键设计】

物模型三要素 品类字典/报文来源 示例(rly4p3 / vfd)
属性(只读) data 报文字段(字典 R 项) oxygen / water_temp / switch1~4 / voltage
属性(可写) 字典标 W 项,经 config 下发 do_map / min_run_s(vfd)
服务 cmd 报文的 cmd + params cmd:"set" params:{"switch1":true}
事件 event 报文 + 事件码表(api-mqtt §11.4) do_low / rcd_trip / device_offline
公共属性 品类字典 §7.0 公共字段 fw / signal / voltage / fault_code / uptime_s

【水产旗舰示例:4G 增氧控制器物模型】

{  
  "properties": [  
    {"identifier":"do_value","name":"溶解氧","data_type":"float","access":"r","unit":"mg/L","min":0,"max":20,"is_chart":true,"is_history":true,"is_alarm":true},  
    {"identifier":"water_temp","name":"水温","data_type":"float","access":"r","unit":"℃","min":-5,"max":60,"is_chart":true,"is_history":true},  
    {"identifier":"box_temp","name":"箱内温度","data_type":"float","access":"r","unit":"℃","min":-20,"max":90,"is_history":true},  
    {"identifier":"ac_voltage","name":"市电电压","data_type":"int","access":"r","unit":"V","min":0,"max":300},  
    {"identifier":"ch1_state","name":"通道1开关","data_type":"bool","access":"r"},  
    {"identifier":"ch1_mode","name":"通道1模式","data_type":"enum","access":"rw",  
     "enum":["auto","manual_on","manual_off"],"is_history":false},  
    {"identifier":"ch1_vfd_hz","name":"通道1频率","data_type":"float","access":"rw","unit":"Hz","min":0,"max":50},  
    {"identifier":"ch1_current","name":"通道1电流","data_type":"float","access":"r","unit":"A"},  
    {"identifier":"do_low","name":"DO下限阈值","data_type":"float","access":"rw","min":0,"max":20},  
    {"identifier":"do_danger","name":"DO危险阈值","data_type":"float","access":"rw","min":0,"max":20},  
    {"identifier":"sensor_ok","name":"传感器在线","data_type":"bool","access":"r"},  
    {"identifier":"rcd_1","name":"1路漏保状态","data_type":"bool","access":"r"},  
    {"identifier":"csq","name":"4G信号","data_type":"int","access":"r","min":0,"max":31},  
    {"identifier":"fw_ver","name":"固件版本","data_type":"string","access":"r"}  
  ],  
  "services": [  
    {"identifier":"set_do_threshold","name":"设置DO阈值","input_params":[  
        {"identifier":"low","data_type":"float"},{"identifier":"danger","data_type":"float"}]},  
    {"identifier":"set_vfd","name":"设置变频频率","input_params":[  
        {"identifier":"ch","data_type":"int","min":1,"max":4},{"identifier":"hz","data_type":"float","min":0,"max":50}]},  
    {"identifier":"relay_ctrl","name":"通道开关","input_params":[  
        {"identifier":"ch","data_type":"int"},{"identifier":"on","data_type":"bool"}]},  
    {"identifier":"restart","name":"设备重启","input_params":[]},  
    {"identifier":"clean_now","name":"立即清洗探头","input_params":[]}  
  ],  
  "events": [  
    {"identifier":"do_danger","name":"溶氧危险","level":3,"output_params":[{"identifier":"value","data_type":"float"}]},  
    {"identifier":"do_low","name":"溶氧偏低","level":2,"output_params":[{"identifier":"value","data_type":"float"}]},  
    {"identifier":"rcd_trip","name":"漏保跳闸","level":3,"output_params":[{"identifier":"ch","data_type":"int"}]},  
    {"identifier":"sensor_offline","name":"传感器离线","level":2,"output_params":[]}  
  ]  
}  

字段口径:上例为平台物模型;设备按 api-mqtt 上报时用 source_field(缺省=identifier)取值,新品类直接用协议字段名。事件级别 info/warning/critical → 1/2/3;完整 Schema 见 §8.4。

旗舰示例与 api-mqtt rly4p3 的字段映射(实现时按此填 source_field,或直接把 identifier 改成协议字段):

平台 identifier api-mqtt source_field 说明
do_value oxygen 溶氧
water_temp water_temp 水温
box_temp box_temp 箱温(协议扩展字段,非 rly4p3 字典必填)
ac_voltage voltage 市电电压
ch1_state switch1 1 路开关
ch1_mode sw1_mode 1 路模式(api-mqtt 改进建议字段,未上线前用 config 扩展)
ch1_vfd_hz sw1_freq 1 路频率(扩展字段;变频器本体用 vfd.freq)
ch1_current current 电流
do_low / do_danger do_low / do_danger 阈值,经 config 下发
sensor_ok sensor_ok 传感器在线(扩展)
rcd_1 rcd_trip(pmet)/ 扩展 rcd_1 漏保状态
csq signal 信号
fw_ver fw 固件版本

M4 设备管理

【功能定义】

  1. 设备注册与激活:产品下创建设备(device_name + sn),生成一次性 device_secret;出厂烧录 → 首次连接即激活;支持 sn 编号段批量导入建档;
  2. 扫码绑定/共享:用户扫 sn 码绑定设备到名下,可共享给家人/他人(device_bind 细分只读/可控/运维授权,语义同通用提纲 device_share);
  3. 在线状态管理:在线/离线/未激活/禁用,最近在线时间;离线判定 = 超过 3×Keepalive 无任何消息(含 LWT 兜底);
  4. 设备影子:平台侧保存设备最新上报状态(Redis reported + desired 期望值),设备离线时指令先落影子(device:{id}:desired),上线后同步下发,支撑离线控制;
  5. 网关与子设备拓扑:子设备经网关代理接入——网关在 gw4g.devices 中声明子设备清单(sn/product_id/在线状态/modbus_addr),平台据此建立子设备影子;一台子设备只归属一个网关,换网关/迁移走平台侧重新注册流程;
  6. 设备分组:通用分组 device_group(空间/业务维度),支持组级批量控制与按组下发;水产行业映射为塘口(pond),设备挂到塘口、塘口挂到养殖场(farm);
  7. 设备定位:记录设备经纬度(可选),支撑地图分布查看;
  8. 设备日志:上下线、属性上报、服务下发、事件上报全记录;
  9. 远程控制与参数配置:按 RBAC 授权(operator 及以上)下发 service;
  10. 在线调试工具:虚拟设备模拟上报、实时报文日志、服务直接调用;
  11. 默认组件展示(必须的数据可视化基础):设备详情页按物模型动态生成组件。

【数据结构】

CREATE TABLE device (  
  id                 BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,  
  tenant_id          BIGINT NOT NULL,  
  product_id         BIGINT NOT NULL REFERENCES product(id),  
  device_name        VARCHAR(64) NOT NULL,  
  device_secret_hash VARCHAR(128),             -- 一机一密,只存哈希;**子设备走网关级鉴权时可为空**  
  sn                 VARCHAR(64) UNIQUE,        -- 出厂序列号(二维码);协议完整标识 = sn.product_key  
  gateway_id         BIGINT,                    -- 所属网关(gateway.id;子设备经网关代理接入,拓扑 = gw4g.devices)  
  modbus_addr        SMALLINT,                  -- RS485 总线地址(子设备)  
  model_version      VARCHAR(16) NOT NULL DEFAULT '1.0.0', -- 绑定的物模型版本(M3 发布版本)  
  hardware_version   VARCHAR(32),               -- 硬件版本,OTA 兼容性校验  
  activation_code_hash VARCHAR(128),            -- 激活码/授权码哈希(扫码绑定前的一次性凭据)  
  activated_at       TIMESTAMP,                 -- 首次激活时间  
  pond_id            BIGINT,                    -- 所属塘口(device_group 的水产行业映射)  
  longitude          DOUBLE PRECISION,          -- 经纬度(可选,地图分布)  
  latitude           DOUBLE PRECISION,  
  timezone           VARCHAR(32) NOT NULL DEFAULT 'Asia/Shanghai',  
  status             SMALLINT NOT NULL DEFAULT 0, -- 0未激活 1在线 2离线 3禁用  
  fw_ver             VARCHAR(16),  
  last_seen          TIMESTAMP,  
  deleted_at         TIMESTAMP,                 -- 软删除;有历史/告警/订单的设备禁止物理删除  
  created_at         TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,  
  UNIQUE (product_id, device_name)  
);  
CREATE INDEX idx_device_tenant ON device (tenant_id);  

-- 通用设备分组(空间/业务维度,组级批量控制;非水产行业直接用本表)  
CREATE TABLE device_group (  
  id        BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,  
  tenant_id BIGINT NOT NULL,  
  name      VARCHAR(64) NOT NULL,  
  parent_id BIGINT DEFAULT 0  
);  
CREATE TABLE device_group_member (  
  group_id  BIGINT NOT NULL,  
  device_id BIGINT NOT NULL,  
  PRIMARY KEY (group_id, device_id)  
);  

-- 水产业务归属(数据集 D1/D6 落地)  
CREATE TABLE farm (  
  id        BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,  
  tenant_id BIGINT NOT NULL,  
  name      VARCHAR(64) NOT NULL,  
  address   VARCHAR(255)  
);  

CREATE TABLE pond (  
  id           BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,  
  tenant_id    BIGINT NOT NULL,  
  farm_id      BIGINT REFERENCES farm(id),  
  name         VARCHAR(64) NOT NULL,  
  area_mu      DOUBLE PRECISION,             -- 面积(亩)  
  species      VARCHAR(32),                  -- 主养品种 → threshold_template.species  
  stock_date   DATE,                         -- 投放日期  
  stock_density VARCHAR(64)                  -- 放养密度  
);  

CREATE TABLE batch (                           -- 养殖批次  
  id         BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,  
  pond_id    BIGINT NOT NULL REFERENCES pond(id),  
  species    VARCHAR(32) NOT NULL,  
  start_date DATE NOT NULL,  
  end_date   DATE,  
  status     SMALLINT NOT NULL DEFAULT 1      -- 1在养 2已出塘  
);  

-- 最新状态表(与 Redis 双写:Redis 供实时读,本表兜底持久)  
CREATE TABLE device_state (  
  device_id           BIGINT NOT NULL,  
  property_identifier VARCHAR(32) NOT NULL,  
  value               JSONB,  
  updated_at          TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,  
  PRIMARY KEY (device_id, property_identifier)  
);  

CREATE TABLE device_log (  
  id         BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,  
  device_id  BIGINT NOT NULL,  
  log_type   SMALLINT NOT NULL,               -- 1属性 2服务 3事件 4生命周期 5非法报文  
  identifier VARCHAR(64),  
  payload    JSONB,                            -- 上/下行报文或校验结果  
  direction  SMALLINT NOT NULL DEFAULT 1,     -- 1上行 2下行  
  created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP  
);  
CREATE INDEX idx_devlog_device_time ON device_log (device_id, created_at DESC);  

【关键设计】

物模型定义 默认组件
bool(access 含 w) 开关卡片
bool(只读) 状态指示灯
int/float + unit 数值大字卡 + 折线图(is_chart=true)
enum 下拉/分段选择器(可写)或标签(只读)
string 文本卡
event(level≥2) 告警卡片(红/黄分级)
service 操作按钮(点了弹参数表单)

M5 数据存储与性能优化

【功能定义】

  1. 最新状态读写:Redis 实时读(看板毫秒级)+ device_state 表兜底;
  2. 历史数据落库:按 is_history 属性写 property_history 分区表;
  3. 降采样聚合:原始 90 天 → 小时聚合 365 天 → 天聚合长期(§3.3 原则 4);
  4. 保留策略与分区管理:定时任务预建下月分区、过期分区归档;
  5. 历史查询 API:支持时间范围 + 聚合粒度(raw/1h/1d)。

【数据结构】

-- 起步用 PG 按月分区(父表 + 示例分区)  
CREATE TABLE property_history (  
  device_id  BIGINT NOT NULL,  
  identifier VARCHAR(32) NOT NULL,  
  value      DOUBLE PRECISION,  
  value_str  VARCHAR(255),                    -- 枚举/字符串属性值  
  ts         TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP  
) PARTITION BY RANGE (ts);  
CREATE TABLE property_history_2026_11 PARTITION OF property_history  
  FOR VALUES FROM ('2026-11-01') TO ('2026-12-01');  
CREATE INDEX idx_ph_dev_ts ON property_history (device_id, identifier, ts DESC);  

-- Redis 键设计(最新状态与在线标记)  
-- device:{id}:state        Hash{identifier: value}    最新属性(实时读)  
-- device:{id}:online       "1" / TTL 120s             在线标记(隐式心跳续期)  
-- device:{id}:last_seen    时间戳  
-- device:{id}:desired      Hash{identifier: value}    期望状态(设备离线时的下发暂存)  

【关键设计】

M6 规则引擎与告警

【功能定义】

  1. 三类触发:属性条件(支持持续时长防抖)、事件触发(事件码表,api-mqtt §11.4)、定时(cron);
  2. 跨属性联动:条件左侧可以是本租户内任意设备、任意产品的任意属性("所有物模型的属性之间均能相互联动"就落在这里);
  3. 四类动作:下发服务(跨设备)、通知(站内信/微信/短信/电话/Webhook/邮件,渠道可配置)、置位虚拟属性、数据转发(把设备数据推给第三方系统:HTTP 推送 / Kafka / MQTT 转发 / 数据库直写,作为独立"数据目的"配置);
  4. 冷却与防抖:冷却时间内重复触发忽略;条件支持 duration(对齐设备端事件 duration_s 去抖语义);
  5. 分级告警闭环(告警中心):触发 → 生成告警记录(级别:提示/警告/严重 + 触发值)→ 通知 → 用户确认(ack)→ 恢复(设备端恢复事件:同名 event + clear:true,或条件回差)→ 关闭 → 全程记 alarm_record,可查可导出;
  6. 场景联动:多条"触发条件 → 执行动作"可编排为场景(支持串行/并行动作),界面配置即时生效、无需发版(LiteFlow 编排语义的内存匹配实现);
  7. 边缘 + 云端双闭环(关键架构约束,见下)。

【数据结构】

CREATE TABLE rule (  
  id             BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,  
  tenant_id      BIGINT NOT NULL,  
  name           VARCHAR(64) NOT NULL,  
  trigger_config JSONB NOT NULL,  
  action_config  JSONB NOT NULL,  
  cooldown       INT NOT NULL DEFAULT 0,       -- 冷却秒数  
  enabled        BOOLEAN NOT NULL DEFAULT TRUE,  
  created_at     TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP  
);  

CREATE TABLE alarm_record (  
  id            BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,  
  tenant_id     BIGINT NOT NULL,  
  device_id     BIGINT NOT NULL,  
  pond_id       BIGINT,  
  rule_id       BIGINT,  
  level         SMALLINT NOT NULL,             -- 1提示 2警告 3严重  
  type          VARCHAR(32) NOT NULL,          -- do_low/do_danger/offline/rcd_trip/...  
  trigger_value JSONB,  
  status        SMALLINT NOT NULL DEFAULT 0,   -- 0未处理 1已确认 2已恢复 3已关闭  
  created_at    TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,  
  acked_at      TIMESTAMP,  
  handled_at    TIMESTAMP  
);  
CREATE INDEX idx_alarm_tenant_time ON alarm_record (tenant_id, created_at DESC);  

-- 数据转发目的(数据开放:HTTP 推送 / Kafka / MQTT 转发 / 数据库直写)  
CREATE TABLE data_forward (  
  id         BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,  
  tenant_id  BIGINT NOT NULL,  
  name       VARCHAR(64) NOT NULL,  
  type       VARCHAR(16) NOT NULL,           -- http / kafka / mqtt / db  
  config     JSONB NOT NULL,                 -- url/topic/连接串等(敏感字段加密存储)  
  enabled    BOOLEAN NOT NULL DEFAULT TRUE  
);  

规则 JSON Schema(trigger + action 的本体定义):

{  
  "name": "DO危险告警",  
  "trigger": {  
    "type": "property",                       // property | event | cron  
    "scope": {"tenant_id": 1001, "pond_id": 88},  // 作用域:整租户/某塘/某设备  
    "conditions": [                            // 多条件 AND/OR  
      {"device_id": null, "product_key":"rly4p3",  
       "identifier":"oxygen", "op":"<", "value":2.0, "duration":300}  
    ],  
    "logic": "AND"  
  },  
  "actions": [  
    {"type":"notify","channel":["wechat","sms"],"level":3,  
     "template":"do_danger","targets":"device_binders"},  
    {"type":"service","device_id":null,"scope":"same_pond",  
     "service":"relay_ctrl","params":{"on":true}},  
    {"type":"virtual_property","identifier":"manual_confirmed","value":false},  
    {"type":"forward","target":"forward:kafka_aqua","payload":"latest"}  
  ],  
  "cooldown": 900,  
  "enabled": true  
}  

【关键设计】

规则 触发条件 通道 冷却
DO 低告警 do_value < do_low 持续 5min 微信 30min
DO 危险告警 do_value < do_danger 持续 2min 微信+短信(旗舰版+电话) 15min
设备离线 lifecycle 离线事件且 5min 未恢复 微信 1h
传感器离线 sensor_ok = false 微信 1h
RCD 跳闸 rcd_N = false(事件) 微信+短信 无
箱内高温 box_temp > 65 微信 1h
订阅到期 定时任务 微信+短信 每天
闭环 位置 延迟 断网可用 职责
边缘闭环 4G 控制器固件 秒级 ✅ DO 四档调频/开关、错峰启动、传感器离线切时间表
云端闭环 平台规则引擎 秒~分钟 ❌ 分级报警、跨设备联动、定时任务、通知矩阵

保命逻辑必须在边缘。云端规则只做"锦上添花"(通知、跨设备调度、数据增值),云端规则失败不得影响设备本地安全;参数下发(阈值)只改边缘闭环的配置。

M7 开放第三方接口(OpenAPI)

【功能定义】

  1. 鉴权:AccessKey/AccessSecret,HMAC-SHA256 签名,按 key 限流、可吊销;
  2. API 范围(只读优先):设备列表/状态、最新属性、历史查询、告警查询;写操作白名单(控制、参数)单独授权;
  3. Webhook 推送:告警事件、设备上下线,第三方登记 URL,失败重试 3 次指数退避;
  4. 数据流开放:高频数据对接不走轮询 API,用 M6 数据转发(HTTP/Kafka/MQTT 转发)持续推送,M7 负责转发目的的授权与审计;
  5. 用途:集成商系统对接、政府监管平台、客户自有 ERP。

【数据结构】

CREATE TABLE access_key (  
  id            BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,  
  tenant_id     BIGINT NOT NULL,  
  access_key    VARCHAR(32) NOT NULL UNIQUE,  
  secret_hash   VARCHAR(128) NOT NULL,  
  scopes        JSONB,                         -- 允许的接口与设备范围  
  rate_limit    INT NOT NULL DEFAULT 100,      -- 次/分钟  
  status        SMALLINT NOT NULL DEFAULT 1,  
  created_at    TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP  
);  

CREATE TABLE webhook_target (  
  id          BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,  
  tenant_id   BIGINT NOT NULL,  
  url         VARCHAR(255) NOT NULL,  
  events      JSONB,                           -- 订阅的事件类型  
  status      SMALLINT NOT NULL DEFAULT 1  
);  

API 路由骨架(/api/v1):

/auth        POST /login(微信/账密)  POST /refresh  
/products    GET/POST /  PUT/POST /{id}/publish  /{id}/model(物模型CRUD)  
/devices     GET /list  GET /{id}  POST /bind  POST /share  
             POST /{id}/service/{identifier}(下发)  GET /{id}/logs  GET /{id}/state  
/data        GET /{device_id}/latest  GET /{device_id}/history?range&interval  
/rules       GET/POST /  PUT /{id}/enable  
/alarms      GET /list  PUT /{id}/ack  PUT /{id}/close  
/internal    POST /mqtt/auth  POST /mqtt/lifecycle      (仅EMQX内网回调)  
/open/v1     第三方只读接口(AccessKey签名)  POST /webhook/register  

完整管理面 API(租户/用户/角色/产品分组/设备分组/塘口/通知/OTA/计费/工单/报表)见 §8.11.4。


4.4 平台可选功能设计

(O1~O7,按需启用)

O1 静态文件显示系统(厂家自定义 HTML)

【功能定义】

  1. 两级模板策略:default(平台内置动态组件,M4 已含)+ custom(厂家上传单页 HTML 替换展示);
  2. 上传管理:白名单校验、版本化存储、回滚;模板归属厂家(product.vendor_id),按 tenant_id/product_key/version 三级隔离;
  3. 沙箱运行:iframe sandbox + postMessage 桥接,不暴露平台内部 API;
  4. 回退逻辑:custom 不存在或加载失败 → 自动回退 default 组件。

【数据结构】

CREATE TABLE product_ui (  
  id            BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,  
  product_id    BIGINT NOT NULL REFERENCES product(id),  
  template_type VARCHAR(16) NOT NULL DEFAULT 'default',  -- default / custom  
  html_path     VARCHAR(255),                            -- current/index.html 相对路径  
  asset_dir     VARCHAR(255),  
  version       VARCHAR(32) NOT NULL DEFAULT '1.0.0',  
  status        SMALLINT NOT NULL DEFAULT 1,  
  created_at    TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP  
);  

存储结构(租户隔离目录树):

static/ui/{tenant_id}/{product_key}/  
  ├── current/{index.html, assets/...}     # 当前生效  
  └── versions/{timestamp}/...             # 历史版本可回滚  

【关键设计】

// 页面 → 平台  
{"type":"invokeService","service":"relay_ctrl","params":{"ch":1,"on":true},"nonce":"x1"}  
// 平台 → 页面  
{"type":"propsUpdate","data":{"do_value":5.2,"ch1_state":true},"nonce":"x1"}  

O2 可视化看板

【功能定义】:拖拽式看板(简化版 Grafana)——选产品 → 选属性 → 生成折线图/仪表盘/开关卡;水产预设模板:单塘总览、多塘地图、运营大屏。可视化三级:设备详情页(M4/O1)→ 轻量看板(本节)→ 数据大屏/组态(可选扩展:拖拽组件生成行业监控大屏,配置存 JSON 渲染)。

【数据结构】

CREATE TABLE dashboard (  
  id         BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,  
  tenant_id  BIGINT NOT NULL,  
  user_id    BIGINT NOT NULL,  
  name       VARCHAR(64) NOT NULL,  
  scope      VARCHAR(16) NOT NULL DEFAULT 'private',  -- private个人 / tenant租户共享 / platform平台模板  
  layout     JSONB NOT NULL,                 -- 组件数组: [{"type":"line","source":{"product_key":"...","identifier":"do_value"},"pos":{"x":0,"y":0,"w":6,"h":4}},...]  
  created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,  
  updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP  
);  
CREATE INDEX idx_dash_tenant_user ON dashboard (tenant_id, user_id);  

layout 存组件类型/数据源(product_key+identifier)/位置尺寸;看板组件与 M4 默认组件共享同一套"属性→组件"映射,保证数据通路一致。

O3 OTA 固件升级

【功能定义】:固件上传(OSS 私有 + 签名 URL)、灰度策略(按 sn 段/百分比)、进度追踪、失败自动回滚、BLE 蓝牙直传备通道。

【数据结构】:

CREATE TABLE firmware (  
  id         BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,  
  product_id BIGINT NOT NULL,  
  version    VARCHAR(16) NOT NULL,  
  file_key   VARCHAR(255) NOT NULL,          -- OSS 对象键  
  file_md5   VARCHAR(64) NOT NULL,  
  signature  VARCHAR(512) NOT NULL,          -- ECDSA 签名,设备端 Secure Boot V2 验签  
  status     SMALLINT NOT NULL DEFAULT 0,    -- 0草稿 1已发布 2已撤回  
  created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,  
  UNIQUE (product_id, version)  
);  

CREATE TABLE ota_task (  
  id          BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,  
  firmware_id BIGINT NOT NULL REFERENCES firmware(id),  
  strategy    JSONB NOT NULL,                -- {"mode":"percent","value":10,"sn_range":[...]}  
  total       INT, succeeded INT DEFAULT 0, failed INT DEFAULT 0,  
  status      SMALLINT NOT NULL DEFAULT 0,   -- 0进行中 1完成 2中止  
  created_at  TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP  
);  

要点:OTA 报文 {"type":"ota","cmd":"start","url","version","md5"} 发网关 cmd;进度 event=ota_progress、结果 cmd_result;30s 内看门狗复位自动回滚;5 分钟 3 次非预期复位进 Safe Mode。

O4 订阅计费

【功能定义】:套餐(基础¥200/标准¥500/旗舰¥1,200/试用30天)、微信支付下单与回调、到期前 30/7/3/1 天提醒、到期降级。

【数据结构】:

CREATE TABLE subscription (  
  id         BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,  
  tenant_id  BIGINT NOT NULL,  
  plan       VARCHAR(16) NOT NULL,           -- trial/basic/standard/flagship  
  scope      JSONB NOT NULL,                 -- {"device_ids":[...]} 或整租户  
  starts_at  TIMESTAMP, expires_at TIMESTAMP,  
  status     SMALLINT NOT NULL DEFAULT 1     -- 1正常 2宽限期 3功能受限 4过期  
);  

CREATE TABLE "order" (  
  id           BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,  
  tenant_id    BIGINT NOT NULL,  
  plan         VARCHAR(16) NOT NULL,  
  amount       INT NOT NULL,                 -- 分  
  pay_status   SMALLINT NOT NULL DEFAULT 0,  -- 0待支付 1已支付 2失败 3已退款  
  wx_transaction_id VARCHAR(64),  
  created_at   TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,  
  paid_at      TIMESTAMP  
);  

铁律:过期降级只限云端功能(历史数据/增值),设备本地控制永不停(保命功能)。状态机与定时任务详见 04_软件方案/云平台规划.md §3.6。

O5 工单与远程运维

【功能定义】:多入口报修、自动派单(离线+DO 极低→P0)、流转、回访;远程诊断(重启/参数/日志回捞/信号电压链路);平台监控(EMQX 连接与吞吐、worker、规则成功率)。

【数据结构】

CREATE TABLE work_order (  
  id           BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,  
  tenant_id    BIGINT NOT NULL,  
  device_id    BIGINT,                       -- 关联设备(可空=通用咨询)  
  pond_id      BIGINT,  
  level        SMALLINT NOT NULL,            -- 0=P0紧急 / 1=P1严重 / 2=P2一般 / 3=P3轻微  
  type         VARCHAR(32) NOT NULL,         -- offline / do_abnormal / rcd_trip / sensor / vfd / other  
  title        VARCHAR(128) NOT NULL,  
  detail       TEXT,  
  source       VARCHAR(16) NOT NULL DEFAULT 'manual', -- manual报修 / auto云端自动派单 / hotline热线 / wechat微信  
  status       SMALLINT NOT NULL DEFAULT 0,  -- 0新建 1已受理 2诊断中 3处理中 4待客户确认 5已关闭 6已回访  
  assigned_to  BIGINT,                       -- 处理人 user_id  
  sla_deadline TIMESTAMP,                    -- 按级别自动计算:P0=4h / P1=24h / P2=48h / P3=7d  
  created_at   TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,  
  responded_at TIMESTAMP,  
  resolved_at  TIMESTAMP,  
  closed_at    TIMESTAMP,  
  feedback     SMALLINT                      -- 回访评分 1~5  
);  
CREATE INDEX idx_wo_tenant_time ON work_order (tenant_id, created_at DESC);  
CREATE INDEX idx_wo_device ON work_order (device_id);  

SLA 承诺(与官网一致):P0 15min 响应/4h 解决;P1 30min/24h;P2 2h/48h;P3 4h/7d。云端自动派单:监测到"设备离线 + DO 极低"直接生成 P0 单。

O6 数据增值服务

【功能定义】:水质报告(引用 standard_ref 标准号)、节能对比报告、投喂建议(V3)、传感器维护提醒;养殖经营记录(V3):投喂/用药/巡塘/死亡/出塘。数据来源见 §3.1 数据集 D2/D3/D6/D7。

【数据结构】

CREATE TABLE report (  
  id           BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,  
  tenant_id    BIGINT NOT NULL,  
  pond_id      BIGINT,  
  batch_id     BIGINT,                       -- 关联养殖批次(追溯类报告用)  
  type         VARCHAR(32) NOT NULL,         -- water_quality水质 / energy节能 / feeding投喂 / maintenance维护提醒  
  period       VARCHAR(16) NOT NULL,         -- day / week / month / batch  
  period_start DATE,  
  period_end   DATE,  
  content      JSONB NOT NULL,               -- 结构化统计结果,含引用的 standard_ref 标准号(§2.3)  
  file_key     VARCHAR(255),                 -- 导出文件 OSS 键(PDF/Excel)  
  created_by   BIGINT,  
  created_at   TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP  
);  
CREATE INDEX idx_report_tenant_time ON report (tenant_id, created_at DESC);  

内容存 JSONB + 可导出 PDF/Excel;数据来源见 §3.1 数据集 D2/D3/D7。

O7 能力扩展位

视频(GB/T 28181+SIP)、能耗(pmet)、水下 AI(cam)、HTTP/TCP/CoAP/Modbus 协议包热加载;只预留接入位,不在 MVP。

【数据结构】:不新增表——扩展品类走 M3 的 product/thing_model(品类字典),扩展协议走 M2 接入层,扩展数据复用 property_history/alarm_record 存储与规则引擎;仅当某扩展需要独立元数据时(如视频通道、AI 事件库)再按 M1~M7 同风格建表。


4.5 部署与演进

起步(<500 台):单机 docker-compose

services:  
  emqx:      { image: emqx/emqx:5.x, ports: ["1883:1883","8883:8883","8083:8083","18083:18083"] }  
  postgres:  { image: postgres:16, volumes: [pgdata:/var/lib/postgresql/data] }  
  redis:     { image: redis:7 }  
  api:       { build: ./backend, depends_on: [emqx, postgres, redis] }   # Node.js(Express) 多实例;FastAPI 见 §8.12.3  
  scheduler: { build: ./backend, command: npm run scheduler }            # Node BullMQ/node-cron;FastAPI 用 python -m app.scheduler  
  nginx:     { image: nginx:alpine, ports: ["80:80","443:443"] }         # 静态UI + 反代 + TLS  
volumes: { pgdata: {} }  

演进路线

规模 动作
500~5,000 台 ECS 升 4C8G;时序迁 TDengine;EMQX 2 节点集群;API 容器化 + SLB
5,000~10,000 台 按域拆服务:接入消费 / 规则告警 / API / 调度各自独立伸缩;EMQX Rule Engine → Kafka 解耦
>10,000 台 此时再评估微服务化与多库分工(FastBee 架构就是目标形态之一)

备份与降级


5、最小闭环与开发路线

5.1 最小闭环

(2 周内必须跑通)

建"增氧控制器"产品 → 定义 oxygen / switch1(M3)→ 模拟设备一机一密上报 {"oxygen":1.8,"switch1":true}(M2)→ 校验落库、默认组件显示(M4/M5)→ 规则 oxygen<2 持续 2 分钟 → 微信通知(M6)→ 通知/日志可查(M1/M4)→ 上传自定义 HTML 并切换(O1)。

5.2 七个开发块

(可独立迭代、可交给 AI 逐块实现)

  1. 数据模型块:M1/M3/M4/M5/M6/M7 各模块 DDL + 物模型 JSON → ORM 模型 + Schema + CRUD(Node:Prisma/Drizzle + Zod;FastAPI:SQLAlchemy + Pydantic);
  2. MQTT 接入块:M2——EMQX HTTP 认证回调 /internal/mqtt/auth + lifecycle Webhook + 共享订阅消费者;
  3. 物模型解析块:M3——三要素校验器 + 落库管道(Redis 最新状态 + 分区历史);
  4. 规则引擎块:M6——trigger/action JSON schema 校验 + 内存匹配器 + 动作执行器 + 冷却;
  5. 设备管理与展示块:M4——绑定/共享/影子/日志 + 默认组件映射与设备详情页;
  6. UI 与静态文件块:O1——上传 API(白名单)+ iframe 沙箱 postMessage 桥接 + 回退逻辑;
  7. 工程与部署块:附录 C 目录结构 + docker-compose + Nginx + 备份脚本。

每块的验收 = §5.1 闭环中对应环节可演示。

5.3 排期建议

(对齐云平台规划 37 天)

阶段 内容 工期
1 块1+块2:建库、EMQX 部署、模拟设备上线、数据落库 6 天
2 块3:物模型解析 + 最新状态/历史 API 4 天
3 块4:规则引擎 + 告警记录 + 微信推送 4 天
4 块5:默认组件 + 设备详情/看板 + 绑定共享 6 天
5 用户系统(微信登录 + RBAC)并入后台 4 天
6 真实设备联调(4G 控制器 + Y504-A)+ OTA(O3) 6 天
7 Web 管理后台 + 订阅计费(O4)+ 开放 API(M7)+ 压测调优 7 天
合计 37 天(最小可用版 15 天:阶段 1-3+5)

6、后期可扩展行业应用

平台基座是通用物联网平台(第 4 章),扩展新行业不改基座代码:只需新增品类物模型字典(M3)、行业规则/阈值模板(§2.4 三件套)与可选的自定义 UI(O1);设备接入、存储、规则引擎、多租户、开放接口全部复用。api-mqtt 协议天然多品类(一个 product_id 一套字典、主题结构不变),已为多行业预留。

行业应用 平台已有能力如何支撑 行业特有增量(仅新增品类/模板)
智慧能源管理 数据转发(HTTP/Kafka/MQTT)持续推送功率/电量数据流给电价与调度系统(M6/M7);规则引擎做峰谷策略、需量告警、柴发自动补能联动(M6);时序存储与降采样支撑能耗曲线与节能报告(M5/O6);订阅计费支撑能源托管分成(O4) 品类字典已预置:ems(储能/光储柴:四态状态机、SOC、峰谷时段 tou_*)、pmet(配电监测:三相/漏保/SPD)、pv(光伏)、gen(柴发)、wxst(气象辐照)——见 api-mqtt §7.5/7.11~7.14,无需新协议
客房控制 RCU 设备分组按"楼层/房间"组织(device_group,M4);设备分享按角色授权——客人只读可控、前台/保洁分组授权(device_bind,M1/M4);场景联动编排"入住/睡眠/退房"一键场景(串行动作序列,M6);自定义 HTML 做酒店品牌化控制面板(iframe 沙箱 + postMessage,O1);开放 API 对接酒店 PMS/客控系统(M7) 新增 rcu 品类(灯光/空调/窗帘/服务开关的聚合面板字典),命令下发复用 cmd/config 闭环,无需动接入层
智能照明 cron 定时触发(M6)+ 气象站光照 solar_rad + 设备经纬度(M4 定位)做"日落开灯/日出关灯"联动;组级批量控制整层灯具(device_group);光照/人流传感器事件触发告警与通知 新增 light 品类(api-mqtt §7.16 已规划:照度、时段),复用网关代理子设备接入
智能家居 米家式家庭模型:tenant 用户组 = 家庭,设备分享给家人(M1/M4);小程序/Web 默认组件开箱即用(M4),自定义面板可选(O1);规则引擎做家庭自动化(温湿度联动空调、离家一键断电);断网本地闭环同样成立(边缘优先) 新增家居品类字典(温湿度、门锁、窗帘、红外家电等),复用"一个网关挂多品类"架构(api-mqtt §3.4)

共性结论:四个行业与智慧水产共用同一套"设备接入 + 物模型 + 规则引擎 + 多租户 + 可视化 + 开放接口",差异只落在品类字典与行业模板上——这正是 §2.4"平台数据标准 = 物模型模板"在多行业的复制,也验证了"本系统不局限于智慧渔业,基座本质是通用物联网平台"的设计目标。


7、与参考平台的对照与借鉴

平台 核心逻辑 本文吸收 本文取舍
涂鸦 产品定义驱动接入;一机一密;设备绑定/分享 M3/M4 全流程、M2 一机一密、O1 产品 UI 双模板 不做涂鸦级多品类生态
米家 家庭(租户)-设备-用户;简单闭环体验 M1 租户模型、默认组件开箱即用 不做米家级生态互联
FastBee 物模型表驱动(type/datatype/specs/is_chart/is_monitor/model_order);多库分工;LiteFlow 规则;若依 RBAC;设备影子 "标签维度"并入物模型字段(M3)、多库分工的演进目标(§4.5)、告警/设备日志表设计(M4/M6)、device 影子=Redis 最新状态 物模型存储选 JSONB 而非表驱动(§4.0);不自研 Broker;规则引擎自研轻量而非 LiteFlow
ThingsBoard 规则链 思路参考(条件→动作链) 不直接套用(偏重)
通用物联网平台提纲 通用基座:厂家+一级租户+用户、产品分组、设备分组/分享/定位、一型一密动态注册、网关子设备、数据转发、告警中心、场景联动、三级可视化、基本功能速览清单 作为本文"基本功能"对照基线:vendor/product_group/device_group/定位等基础能力已全部落入 M1~M4,数据转发/告警恢复/场景联动落入 M6,自查清单见 §4.1.1 租户模型用 tenant 树合并 vendor(§1.2 简化决策);产品分组只做一级扁平,不做分类树;Topic 采用 api-mqtt 实测协议而非提纲示例风格(M2 已说明)
api-mqtt(设备协议 v1.2) up/cmd 两主题 + type 扩展、cmd_result 命令闭环、config/cfg_ver、事件码表、离线补传、LWT/重连、网关级 client_id、品类字段字典 直接作为 M2 设备侧行为协议与 M3 品类字典来源(已实测在用):报文示例见附录 A/B,事件码表驱动 M6 告警 平台概念(product_key/device_name/TSL)与协议概念(product_id/sn/报文字段)按 M2/M3 对照表映射,平台内部模型不被协议绑架
联犀/铱塔 Go 物模型、场景联动 场景联动语义对照 —

一句话总结:设备按 api-mqtt 协议(up/cmd + type)走 MQTT 一机一密进来,物模型 JSONB 定义"设备是什么",规则引擎在数据流入时判断"该做什么",静态文件系统决定"用户看到什么"——默认组件保证完整可用,自定义 HTML 让厂家专业呈现,多租户从数据库到静态目录全程隔离;而保命的增氧控制永远留在塘口边缘,云端全挂也不翻塘。


8、实现级完整规范(审查补差)

本章回答一个验收问题:"只读第 1~7 章,能不能把平台真正写出来?" 第 1~7 章已经把"做什么、为什么、核心表"讲清楚;本章补齐"怎么接、怎么校验、怎么隔离、怎么失败、怎么上线、怎么验收"的实现级契约。
阅读顺序:8.1 先统一口径(不先读会写出互相冲突的代码)→ 8.2~8.8 按 M1~M7 补实现 → 8.9 可选模块 → 8.10 前端/移动端 → 8.11~8.13 工程/测试。
约定:本章新增表统一使用 PostgreSQL 语法;标注【P0】的是 MVP 必须实现,【P1】是 V2 必须,【P2】是可选。

8.1 全局口径与铁律【P0】

8.1.1 身份与命名口径(最容易写错的地方)

概念 唯一来源 规则 反例(禁止)
品类/产品对外标识 product.product_key = api-mqtt product_id 小写 ^[a-z][a-z0-9_]{1,31}$,全局唯一;产品少时直接 product_a / product_b;创建后不可改 pk_aqua_ctrl_v1 作为 MQTT 主题段
设备序列号 device.sn 全局唯一、出厂写死、二维码内容;建议 ^[A-Za-z0-9_-]{4,32}$ 用数据库自增 id 当 sn
设备展示名 device.device_name 租户内可重复,可改;仅用于 UI 把 device_name 放进 MQTT 主题
协议完整标识 sn + "." + product_key 必须与 Topic 中间段、JSON sn 字段完全一致 sz0001(缺品类)
MQTT 连接层身份 gateway.client_id 网关 MAC 12 位 Hex 按 2-2-2 分组 + .gw4g,如 0025-CA3B-6F01.gw4g 用子设备 sn 当 client_id
物模型属性键 property.identifier 新字段小写 snake_case ≤10 字符;默认等于 api-mqtt 报文字段名 dissolvedOxygen、doValue
协议字段映射 property.source_field 缺省=identifier;仅历史改名/别名时不同 每个字段都强行造别名
设备类型 product.device_type 1 直连 / 2 网关子设备;网关本体也是 product(gw4g) 网关不建产品、直接写死代码

必须成立的三条等式(建议写成数据库约束 + 单元测试):

topic 中间段 == device.sn + "." + product.product_key
JSON sn      == topic 中间段
product.product_key == api-mqtt product_id == thing_model 所属产品键

8.1.2 时间、单位、ID、编码

项 规范
时间存储 数据库统一 TIMESTAMPTZ(PG TIMESTAMP WITH TIME ZONE),服务端按 UTC 存储,API 输出 ISO8601 带 +08:00
设备时间 报文 timestamp / collect_ts 必须 ISO8601 秒级、固定 +08:00;平台记录 received_at 用于时钟偏差判断
时钟偏差 abs(received_at - timestamp) > 300s 记一条 clock_skew 设备日志;不拒绝数据,但补传按 collect_ts 归档
单位 每字段一个单位、固定量程/精度;跨品类同名同单位(oxygen 永远 mg/L)
主键 业务表 BIGINT IDENTITY;对外 API 用字符串返回,避免 JS 精度丢失
业务编号 request_id = REQ-{epoch_ms}-{6位随机};order_no = YYYYMMDD + 10位序列;work_order_no = WO + 日期 + 序列
枚举 数据库存 SMALLINT,API 同时给 code 与 label;禁止在业务代码里散落魔法数字
软删除 设备/产品/规则/用户用 deleted_at;历史、告警、订单、审计禁止物理删除
幂等键 所有写接口支持 Idempotency-Key 请求头;服务端 Redis 保存 24h 结果

8.1.3 五条铁律

  1. 配置与数据分离:物模型/规则/权限在关系库;属性值在 Redis + 时序/分区表;只靠 product_key / device_id 关联。
  2. 租户隔离在服务层强制:Repository 基类自动注入 tenant_id;平台管理员跨租户必须显式 cross_tenant=True 并写审计;禁止跨租户 JOIN。
  3. 三套凭据不混用:用户 JWT / 开放 API AccessKey / 设备一机一密(或网关级凭据);任何一套泄露不影响另外两套。
  4. 保命逻辑在边缘:DO 四档、错峰、RCD 重试、传感器离线切时间表全部在固件;云端只做通知、跨设备联动、参数下发。云端规则失败不得改变设备本地安全行为。
  5. 未知报文不丢弃:未知 type 按公共骨架解析 sn/timestamp 后写 device_log,返回成功消费;未知字段记录但不落属性。

8.1.4 边缘/云端职责矩阵

能力 边缘(固件) 云端 冲突裁决
DO 四档调频/启停 ✅ 主执行 只下发参数 边缘优先,云端不得直接改频率
错峰启动/最短运行 ✅ 只展示 边缘优先
RCD 跳闸重试 ✅ 只告警/工单 边缘优先
传感器离线切时间表 ✅ 只告警 边缘优先
分级报警 事件上报 ✅ 通知矩阵 云端可静音,不可阻断本地保护
跨设备联动 ❌ ✅ 规则引擎 云端失败不影响边缘
远程手动控制 本地仲裁 ✅ 下发 见 §8.3.6 优先级
OTA 验签/双分区 ✅ 灰度 边缘验签失败必须回滚

8.2 身份、租户与权限【P0】

8.2.1 租户树与生命周期

platform(level=0, id=1)
 └── vendor(level=1)                 厂家:产品/固件/自定义UI归属
      └── tenant(level=2)            代理商/基地/集团子公司
           └── user_group(level=3)   家庭/班组/用户组(C端自助注册默认建这一层)
操作 规则
创建租户 校验 parent_id 存在且 parent.level = level-1;平台管理员或上级 tenant_admin 可建
停用租户 该租户下所有用户 JWT 立即失效(Redis 版本号 +1);设备保留但禁止控制;数据不删除
过户/转移 变更 device.tenant_id 与 product.vendor_id 需平台管理员双人确认;写审计;MQTT 连接不断,数据归属即时切换
删除租户 仅允许无设备、无订单、无历史;否则只能停用
vendor 维度 product.vendor_id 指向 level=1 节点;固件、自定义 UI、品牌素材按 vendor 隔离

8.2.2 RBAC 数据模型(补 M1 缺口)

原 M1 只在 user.role 存一个字符串,无法支撑"成员管理/自定义角色/权限点"。补齐以下表;user.role 保留为"主角色"冗余字段,鉴权以 user_role 为准。

CREATE TABLE role (
  id          BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
  tenant_id   BIGINT NOT NULL,                 -- 0=平台内置角色
  code        VARCHAR(32) NOT NULL,            -- platform_admin / tenant_admin / member / operator / custom_xxx
  name        VARCHAR(64) NOT NULL,
  is_builtin  BOOLEAN NOT NULL DEFAULT FALSE,
  UNIQUE (tenant_id, code)
);

CREATE TABLE permission (
  id          BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
  code        VARCHAR(64) NOT NULL UNIQUE,     -- product:create / device:control / rule:update / ota:push ...
  name        VARCHAR(64) NOT NULL,
  module      VARCHAR(32) NOT NULL
);

CREATE TABLE role_permission (
  role_id       BIGINT NOT NULL REFERENCES role(id),
  permission_id BIGINT NOT NULL REFERENCES permission(id),
  PRIMARY KEY (role_id, permission_id)
);

CREATE TABLE user_role (
  user_id BIGINT NOT NULL REFERENCES "user"(id),
  role_id BIGINT NOT NULL REFERENCES role(id),
  PRIMARY KEY (user_id, role_id)
);

CREATE TABLE tenant_invite (
  id         BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
  tenant_id  BIGINT NOT NULL,
  code       VARCHAR(32) NOT NULL UNIQUE,       -- 邀请码
  role_code  VARCHAR(32) NOT NULL DEFAULT 'member',
  expires_at TIMESTAMP NOT NULL,
  used_by    BIGINT,
  created_by BIGINT NOT NULL
);

权限点最小集(种子数据,按 模块:动作):

tenant:read tenant:manage user:read user:manage role:manage
product:read product:create product:update product:publish product:delete
model:read model:update model:publish
device:read device:create device:import device:update device:delete
device:bind device:share device:control device:config device:debug device:transfer
group:manage pond:manage farm:manage
data:read data:export
rule:read rule:create rule:update rule:enable
alarm:read alarm:ack alarm:close
ota:read ota:upload ota:push
order:read order:manage subscription:manage
workorder:read workorder:assign workorder:handle
openapi:manage audit:read

内置角色权限:

角色 权限
platform_admin 全部 + 跨租户
tenant_admin 本租户除平台级外的全部
operator device:read/control/config/debug、ota:push、alarm:ack、workorder:handle、data:read
member device:read、data:read、alarm:read/ack(仅绑定设备)
installer(可扩展) device:bind/debug、device:read,仅安装期临时授权(expires_at)

实现方式:Express 中间件 requirePerm('device:control')(推荐) 或 FastAPI Depends(require_perm("device:control"));权限码从 JWT 缓存 + Redis 缓存读取;角色变更后 Redis 版本号 +1 强制刷新。

8.2.3 认证流程(微信 / 账密 / 刷新 / 登出)

JWT 约定:

{
  "iss": "aqua-platform",
  "sub": "1024",                 // user_id
  "tenant_id": 2001,
  "role": "member",
  "scope": ["device:read", "data:read"],
  "ver": 3,                      // 用户权限版本号,与 Redis user:{id}:ver 比对
  "jti": "uuid",
  "iat": 1790000000,
  "exp": 1790007200              // access token 2h
}
端 登录方式 流程 Token 有效期
微信小程序 wx.login code → jscode2session 换 openid/unionid → 查/建 user → 首次建个人 user_group 租户 → 签发 JWT access 2h / refresh 30d
Web 后台 手机号+密码 密码 Argon2id/bcrypt → 短信二次验证(可选)→ JWT access 2h / refresh 7d
开放 API AccessKey HMAC 签名(§8.8),不使用 JWT 每请求
设备 MQTT 一机一密/网关级 EMQX HTTP 认证回调(§8.3.2) 连接期
ALTER TABLE "user" ADD COLUMN unionid VARCHAR(64);
ALTER TABLE "user" ADD COLUMN nickname VARCHAR(64);
ALTER TABLE "user" ADD COLUMN avatar_url VARCHAR(255);
ALTER TABLE "user" ADD COLUMN consent_at TIMESTAMPTZ;
ALTER TABLE "user" ADD COLUMN consent_version VARCHAR(16);
ALTER TABLE "user" ADD COLUMN last_login_at TIMESTAMPTZ;
CREATE UNIQUE INDEX uq_user_unionid ON "user" (unionid) WHERE unionid IS NOT NULL;

刷新与登出:

短信验证码:Redis sms:{scene}:{phone} 6 位、5 分钟有效、60s 重发、每小时 5 次、每天 20 次;验证码只存哈希。

8.2.4 设备绑定、共享、转移(补 M1/M4 缺口)

device_bind 增加字段并明确语义:

-- bind_type 已在 §4.3 M1 中定义;v5.0 统一取值:1拥有 2共享只读 3共享可控 4运维授权
-- 若存量 3 表示“运维授权”,迁移脚本需把 3 → 4,再启用“3=共享可控”
ALTER TABLE device_bind ADD COLUMN granted_by BIGINT;
ALTER TABLE device_bind ADD COLUMN expires_at TIMESTAMP;
ALTER TABLE device_bind ADD COLUMN status SMALLINT NOT NULL DEFAULT 1; -- 1有效 2已撤销
场景 规则
扫码绑定 二维码内容 = aqua://bind?sn=xxx&code=yyy;code 为一次性激活码(device.activation_code_hash);绑定成功即置 activated_at、status=1、owner bind
绑定冲突 一设备只能有一个 owner;已被他人绑定时返回 409 DEVICE_ALREADY_BOUND,走"申请转移"流程
共享给家人 owner 可授 share_read / share_control;share_control 仍需设备 mode 允许远程;可设 expires_at
运维授权 operator 通过工单获得 bind_type=4,默认 72h 过期;到期自动撤销
解绑 owner 解绑后设备进入 未激活?不——保留设备档案,清空 owner bind,历史仍可查;设备继续上报但无 C 端可见
转移 原 owner 发起 → 新 owner 确认 → 平台/代理商审核(可配置)→ 事务内改 device.tenant_id + bind + 审计;转移不换 sn、不断连接
换网关 子设备 gateway_id 变更:旧网关清单移除、新网关清单加入;平台校验子设备当前离线;写迁移审计;历史数据不迁移

8.2.5 操作审计(补 M1)

ALTER TABLE audit_log ADD COLUMN request_id VARCHAR(64);
ALTER TABLE audit_log ADD COLUMN ip VARCHAR(64);
ALTER TABLE audit_log ADD COLUMN user_agent VARCHAR(255);
ALTER TABLE audit_log ADD COLUMN before JSONB;
ALTER TABLE audit_log ADD COLUMN after JSONB;
ALTER TABLE audit_log ADD COLUMN result SMALLINT NOT NULL DEFAULT 0; -- 0成功 1失败
ALTER TABLE audit_log ADD COLUMN error_code VARCHAR(32);

audit_log 补字段:request_id、ip、user_agent、before、after、result、error_code。

8.2.6 多租户隔离实现(代码级)

// Node.js(推荐):Repository 基类统一注入 tenant_id
export abstract class TenantRepository<T> {
  constructor(
    protected readonly db: PrismaClient | DrizzleDb,
    protected readonly tenantId: bigint,
    protected readonly crossTenant = false,
  ) {}

  protected scoped() {
    return this.crossTenant ? {} : { tenantId: this.tenantId };
  }
}

// FastAPI(备选):Depends 注入 tenant_id,Repository 统一 where
// def scoped(stmt): return stmt if cross_tenant else stmt.where(Model.tenant_id == tenant_id)

8.2.7 账号安全基线

项 要求
密码 最少 8 位,含字母+数字;Argon2id 或 bcrypt cost≥12;禁止明文/可逆加密
登录限流 同 IP 10 次/分钟;同账号 5 次失败锁 15 分钟;验证码防轰炸
JWT 密钥 环境变量/密钥管理,≥32 字节;支持 kid 轮换;不得写进代码/镜像
传输 全站 HTTPS/WSS;HSTS;TLS1.2+;禁用弱套件
敏感数据 手机号/地址/ICCID 展示脱敏;导出需二次授权;日志不打印 token/secret
会话 支持"退出所有设备";管理员可强制下线
合规 隐私政策、用户授权记录、数据导出/删除接口、等保二级基线(日志留存≥6个月)

8.3 设备接入实现规范(M2 补差)【P0】

8.3.1 连接与主题最终口径

项 生产 开发
MQTT 3.1.1 / 5.0 均可,固定 3.1.1 1883
TCP TLS 8883,TLS1.2+,服务端证书 关闭
WebSocket 8084(WSS),路径 /mqtt,供小程序可选直连 8083
Keepalive 60s 60s
QoS 上行/下行均 1 1
Retain false false
Clean Start true true
会话过期 0(设备侧自管离线缓存) 0
最大报文 Broker 上限 256KB;平台业务上限 64KB,超限拒绝并记日志 同
上行(设备→平台)  aqua/<sn>.<product_key>/up     type: status/data/event/cmd_result/log/err/ota
下行(平台→设备)  aqua/<sn>.<product_key>/cmd    type: cmd/config/ota
平台共享订阅       $share/aqua-core/aqua/+/up

8.3.2 EMQX 认证回调契约(POST /internal/mqtt/auth)

平台不实现 Broker,只实现认证/授权回调。EMQX 5 的 HTTP 认证请求体可配置,建议固定为:

// EMQX → 平台(Header: X-Internal-Token: <shared secret>)
{
  "clientid": "0025-CA3B-6F01.gw4g",
  "username": "gw:0025CA3B6F01",       // 网关:gw:<mac>;直连设备:dev:<sn>.<product_key>;App:app:<user_id>
  "password": "<device_secret 或短期 token>",
  "peerhost": "10.0.0.8",
  "protocol": "mqtt",
  "clean_start": true
}
// 平台 → EMQX(HTTP 200)
{"result": "allow", "is_superuser": false, "client_attrs": {"tenant_id": "2001", "kind": "gateway"}}
// 拒绝
{"result": "deny", "is_superuser": false}
// 不归本回调处理(交给下一个认证链)
{"result": "ignore", "is_superuser": false}

校验逻辑:

client_id 前缀 身份 校验
XXXX-XXXX-XXXX.gw4g 网关 解析 MAC → 查 gateway;bcrypt/argon2.verify(password, gateway.secret_hash);status=1;记录 last_connected_at
dev:<sn>.<product_key> 直连设备(可选) 查 device + product;校验 device.device_secret_hash;设备必须 status!=3
app:<user_id>:<uuid> 小程序/App 可选 WSS 校验短期 token(Redis app:mqtt:{token},15min,绑定 user_id);只读 ACL
其他 拒绝 记安全日志,返回 deny

8.3.3 上下线 Webhook 契约(POST /internal/mqtt/lifecycle)

// EMQX client.connected / client.disconnected 归一化后
{
  "event": "connected",              // connected | disconnected
  "clientid": "0025-CA3B-6F01.gw4g",
  "username": "gw:0025CA3B6F01",
  "ts": "2026-10-24T10:00:00+08:00",
  "reason": "normal",                // normal | keepalive_timeout | tcp_closed | ...
  "peerhost": "10.0.0.8",
  "clean_start": true
}

处理规则:

  1. connected:网关 status=1、last_seen=now、写 Redis 在线键;把该网关下所有子设备标记 在线待确认,等待子设备 status 或首帧 data 确认。
  2. disconnected:网关 status=2、所有子设备 status=2、写离线事件;不要立即告警,等 §8.7 的离线规则(5 分钟未恢复)再触发。
  3. 重复/乱序事件按 ts 与当前状态幂等处理。
  4. Webhook 失败重试由 EMQX 负责;平台必须快速返回 200(异步落库),避免阻塞 Broker。

8.3.4 主题 ACL(授权回调或内置 ACL)

目标:网关只能读写自己代理的子设备主题;App 只能读自己绑定的设备。

网关 G(client_id=0025-CA3B-6F01.gw4g):
  allow publish   aqua/{sub_sn}.{sub_product_key}/up     for each sub in gateway.devices
  allow subscribe aqua/{sub_sn}.{sub_product_key}/cmd    for each sub in gateway.devices
  deny  publish/subscribe aqua/+/cmd、aqua/+/up、$SYS/#
App(只读模式):
  allow subscribe aqua/{sn}.{product_key}/up            仅 device_bind 有效设备
  deny  publish aqua/#

实现二选一:

8.3.5 上行消费管道(从 MQTT 到落库)

MQTT $share/aqua-core/aqua/+/up
  → ① 解包/大小限制/JSON 解析失败 → 记 err 日志,丢弃
  → ② 身份解析:topic 中间段 与 JSON sn 必须一致;不一致拒绝并记非法报文
  → ③ 去重:Redis SETNX mqtt:dedup:{sn}:{msg_id 或 sn:seq} TTL 24h
  → ④ 按 type 分派:status / data / event / cmd_result / log / ota / err / unknown
  → ⑤ 查产品已发布物模型(Redis 缓存 product:{key}:model:{version})
  → ⑥ 字段校验(类型/范围/枚举/只读);未知字段记录不落库
  → ⑦ 写 Redis 最新状态 + 批量写 property_history(is_history)+ device_log
  → ⑧ 更新在线/心跳、last_seen
  → ⑨ 交规则引擎(异步,不阻塞落库)
  → ⑩ 交 WebSocket/SSE 推送(设备详情实时刷新)

并发与可靠性:

8.3.6 命令闭环、仲裁与离线控制【P0】

命令状态机:

CREATED → SENT → ACKED(success=true) 
              ├→ FAILED(success=false / code!=0)
              ├→ TIMEOUT → RETRY(同 seq/request_id) → ... → EXPIRED
              └→ REJECTED(设备本地安全规则 code=6)

补差表 device_command(命令审计与超时重试的唯一依据):

CREATE TABLE device_command (
  id           BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
  tenant_id    BIGINT NOT NULL,
  device_id    BIGINT NOT NULL,
  request_id   VARCHAR(64) NOT NULL UNIQUE,
  seq          INT NOT NULL,
  cmd_type     VARCHAR(16) NOT NULL,          -- cmd / config / ota
  cmd          VARCHAR(32),
  params       JSONB,
  source       VARCHAR(16) NOT NULL,          -- app / cloud / auto / schedule
  status       SMALLINT NOT NULL DEFAULT 0,   -- 0创建 1已发 2成功 3失败 4超时 5过期 6拒绝
  retry_count  SMALLINT NOT NULL DEFAULT 0,
  timeout_ms   INT NOT NULL DEFAULT 5000,
  result       JSONB,
  error_code   INT,
  created_by   BIGINT,
  created_at   TIMESTAMPTZ NOT NULL DEFAULT now(),
  sent_at      TIMESTAMPTZ,
  acked_at     TIMESTAMPTZ,
  expires_at   TIMESTAMPTZ
);
CREATE INDEX idx_cmd_device_time ON device_command (device_id, created_at DESC);
CREATE INDEX idx_cmd_status ON device_command (status) WHERE status IN (1,4);

seq 分配:Redis INCR device:{id}:seq,对 65536 取模;设备重启不清零;平台重启从 device_command 最大 seq 恢复(或直接用 Redis 持久化)。

重试:仅对幂等命令(set/config/查询)重试;默认 1 次,间隔 2s;ota 不自动重试。超时后 status=4,写告警"设备未确认"。

离线控制(影子):

命令优先级(设备端仲裁,平台记录 source):

优先级 来源 说明
1(最高) 本地安全联锁 过流/缺相/漏电/超温/传感器离线保护,任何远程命令不可覆盖
2 本地物理开关/面板 现场人工
3 source=app 手动 运维/管理员远程手动
4 source=cloud 自动 云端规则引擎
5 source=schedule 定时 时间表

8.3.7 网关与子设备拓扑(补 M2/M4 缺口)

原 device.gateway_id 没有网关实体。补:

CREATE TABLE gateway (
  id              BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
  tenant_id       BIGINT NOT NULL,
  product_id      BIGINT NOT NULL REFERENCES product(id),  -- 固定为 gw4g 产品
  gw_sn           VARCHAR(32) NOT NULL UNIQUE,             -- MAC 去横线,如 0025CA3B6F01
  client_id       VARCHAR(64) NOT NULL UNIQUE,             -- 0025-CA3B-6F01.gw4g
  secret_hash     VARCHAR(128) NOT NULL,
  fw_ver          VARCHAR(16),
  signal          INT,
  sim_iccid       VARCHAR(32),
  sim_state       VARCHAR(16),
  cache_cnt       INT DEFAULT 0,
  ntp_ok          BOOLEAN DEFAULT FALSE,
  cfg_ver         INT DEFAULT 0,
  status          SMALLINT NOT NULL DEFAULT 0,             -- 0未激活 1在线 2离线 3禁用
  last_seen       TIMESTAMPTZ,
  created_at      TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

CREATE TABLE gateway_subdevice (
  gateway_id  BIGINT NOT NULL REFERENCES gateway(id),
  device_id   BIGINT NOT NULL REFERENCES device(id),
  modbus_addr SMALLINT,
  online      BOOLEAN NOT NULL DEFAULT FALSE,
  last_seen   TIMESTAMPTZ,
  PRIMARY KEY (gateway_id, device_id),
  UNIQUE (device_id)                                       -- 一子设备只属一个网关
);

gw4g.devices 同步规则:

  1. 网关 status 带 devices[]:按 (gateway_id, sn) upsert gateway_subdevice;新子设备自动建档(产品必须已存在且 device_type=2)。
  2. 清单中消失的子设备:不删除,标记 online=false,超过 24h 仍消失则告警"子设备丢失"。
  3. 子设备首次出现但未在平台注册:若产品允许自动注册(product.auto_register=true),创建 status=0 档案;否则记 unregistered_subdevice 日志并告警。
  4. 网关离线:其全部子设备 online=false;网关恢复后以子设备 status/data 逐台确认。
  5. 换网关:旧 gateway_subdevice 删除 → 新网关插入;要求子设备当前离线;写审计;历史数据保留原 device_id。

8.3.8 设备出厂、配网与首次激活【P0】

出厂流程:

平台创建设备 → 生成 sn + device_secret(仅展示一次)+ activation_code(二维码)
  → 工厂烧录 secret / 打印二维码
  → 设备上电连 MQTT → EMQX 认证通过 → 上报 status(online=true, fw, signal, devices)
  → 平台置 activated_at、status=1(首次激活)
  → 用户扫码绑定 → device_bind(owner) → 订阅告警

8.3.9 断网补传、去重与时间同步

项 规则
缓存 设备 Flash 环形缓冲 ≥10000 条,满丢最旧;gw4g.cache_cnt 上报未补传数
补传标记 cache=true + collect_ts(采集时间)+ timestamp(发送时间)
归档 按 (sn, collect_ts) 去重;写 property_history.ts = collect_ts
告警 补传数据不触发实时告警;批内越限生成"补传异常摘要"(event=backfill_summary)供人工查看
顺序 按 collect_ts 升序补发;平台按 collect_ts 排序写入,允许乱序到达
时钟 网关 NTP 优先(ntp.aliyun.com);未同步 ntp_ok=false;平台可 config 下发 server_time 校时
偏差 abs(received_at - timestamp) > 300s 记 clock_skew,不拒收;>24h 的数据标记 stale=true,只归档不展示实时
摘要 补传完成后平台生成一条 device_log(log_type=6 补传摘要),含条数/时间范围/越限数

8.3.10 EMQX 关键配置样例(可直接照抄)

# emqx.conf(节选)
listeners.tcp.default { bind = "0.0.0.0:1883" }
listeners.ssl.default {
  bind = "0.0.0.0:8883"
  ssl_options { certfile = "/etc/emqx/certs/server.crt", keyfile = "/etc/emqx/certs/server.key", verify = verify_none }
}
listeners.ws.default { bind = "0.0.0.0:8083", websocket.mqtt_path = "/mqtt" }
listeners.wss.default { bind = "0.0.0.0:8084", websocket.mqtt_path = "/mqtt" }

authentication = [
  {
    mechanism = password_based
    backend = http
    method = post
    url = "http://api:8000/internal/mqtt/auth"
    headers { "X-Internal-Token" = "${INTERNAL_TOKEN}" }
    body { clientid = "${clientid}", username = "${username}", password = "${password}", peerhost = "${peerhost}" }
    request_timeout = 5s
  }
]

# Webhook:client.connected / client.disconnected → /internal/mqtt/lifecycle
# 共享订阅无需配置,客户端使用 $share/aqua-core/aqua/+/up 即可

EMQX 版本差异:5.x 的认证/授权配置项名称可能随小版本变化,部署时以 emqx.conf 官方 schema 为准;平台侧契约(§8.3.2/8.3.3)不变。

8.4 产品与物模型实现规范(M3 补差)【P0】

8.4.1 产品生命周期与补差字段

draft(0) → published(1) → deprecated(2) → archived(3)
   ↑            │
   └── 编辑草稿 ─┘(已发布产品的物模型改动生成新版本,不原地改)
-- model_no 已在 §4.3 M3 的 product 建表中定义,无需重复添加
ALTER TABLE product ADD COLUMN auto_register BOOLEAN NOT NULL DEFAULT FALSE;
ALTER TABLE product ADD COLUMN product_secret_hash VARCHAR(128);  -- 一型一密,可选
ALTER TABLE product ADD COLUMN published_at TIMESTAMPTZ;
ALTER TABLE product ADD COLUMN deprecated_at TIMESTAMPTZ;
ALTER TABLE product ADD COLUMN deleted_at TIMESTAMPTZ;
ALTER TABLE product ADD COLUMN protocol VARCHAR(16) NOT NULL DEFAULT 'mqtt'; -- mqtt/http/tcp/modbus
操作 前置条件 副作用
发布产品 至少 1 个属性;物模型 JSON 校验通过;product_key 唯一且符合协议 生成新 thing_model.version、product.status=1、刷新缓存、写审计
编辑已发布产品 创建新草稿版本 已绑定设备仍按 device.model_version 解析旧版本,不立即切换
下线产品 无在线设备或已迁移 禁止新建设备;旧设备可上报但 UI 标记"已下线"
删除产品 无设备、无历史、无固件 软删除;product_key 永不复用
重置设备密钥 有 device:update 权限 生成新 secret(只展示一次)、旧连接踢下线、审计

8.4.2 物模型完整 JSON Schema

{
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "type": "object",
  "required": ["properties", "services", "events"],
  "properties": {
    "properties": {
      "type": "array",
      "items": {
        "type": "object",
        "required": ["identifier", "name", "data_type", "access"],
        "properties": {
          "identifier":   {"type": "string", "pattern": "^[a-z][a-z0-9_]{0,9}$"},
          "source_field": {"type": "string", "pattern": "^[a-zA-Z][a-zA-Z0-9_]{0,31}$"},
          "name":         {"type": "string", "maxLength": 64},
          "data_type":    {"enum": ["int", "float", "bool", "enum", "string", "array", "object"]},
          "access":       {"enum": ["r", "rw", "report_only"]},
          "unit":         {"type": "string", "maxLength": 16},
          "min":          {"type": "number"},
          "max":          {"type": "number"},
          "step":         {"type": "number"},
          "precision":    {"type": "integer", "minimum": 0, "maximum": 6},
          "enum":         {"type": "array", "items": {"type": "string"}},
          "max_length":   {"type": "integer", "minimum": 1, "maximum": 1024},
          "items":        {"type": "object"},   // data_type=array 时的元素定义(同 param 结构)
          "schema":       {"type": "object"},   // data_type=object 时的 JSON Schema
          "default":      {},
          "required":     {"type": "boolean", "default": false},
          "is_chart":     {"type": "boolean", "default": false},
          "is_history":   {"type": "boolean", "default": false},
          "is_alarm":     {"type": "boolean", "default": false},
          "is_monitor":   {"type": "boolean", "default": false},
          "is_readonly":  {"type": "boolean", "default": false},
          "is_share_perm":{"type": "boolean", "default": true},
          "model_order":  {"type": "integer", "default": 0},
          "description":  {"type": "string", "maxLength": 255}
        },
        "allOf": [
          {"if": {"properties": {"data_type": {"enum": ["enum"]}}},
           "then": {"required": ["enum"]}},
          {"if": {"properties": {"data_type": {"enum": ["int", "float"]}}},
           "then": {"required": ["min", "max"]}}
        ]
      }
    },
    "services": {
      "type": "array",
      "items": {
        "type": "object",
        "required": ["identifier", "name", "input_params"],
        "properties": {
          "identifier":   {"type": "string", "pattern": "^[a-z][a-z0-9_]{0,31}$"},
          "name":         {"type": "string", "maxLength": 64},
          "input_params": {"type": "array", "items": {"$ref": "#/$defs/param"}},
          "output_params":{"type": "array", "items": {"$ref": "#/$defs/param"}},
          "timeout_ms":   {"type": "integer", "default": 5000, "minimum": 1000, "maximum": 60000},
          "async":        {"type": "boolean", "default": false}
        }
      }
    },
    "events": {
      "type": "array",
      "items": {
        "type": "object",
        "required": ["identifier", "name", "level", "output_params"],
        "properties": {
          "identifier":   {"type": "string", "pattern": "^[a-z][a-z0-9_]{0,31}$"},
          "name":         {"type": "string", "maxLength": 64},
          "level":        {"type": "integer", "enum": [1, 2, 3]},   // info/warning/critical
          "output_params":{"type": "array", "items": {"$ref": "#/$defs/param"}},
          "recoverable":  {"type": "boolean", "default": true},
          "description":  {"type": "string", "maxLength": 255}
        }
      }
    }
  },
  "$defs": {
    "param": {
      "type": "object",
      "required": ["identifier", "data_type"],
      "properties": {
        "identifier": {"type": "string", "pattern": "^[a-z][a-z0-9_]{0,31}$"},
        "name":       {"type": "string"},
        "data_type":  {"enum": ["int", "float", "bool", "enum", "string"]},
        "unit":       {"type": "string"},
        "min":        {"type": "number"},
        "max":        {"type": "number"},
        "step":       {"type": "number"},
        "enum":       {"type": "array", "items": {"type": "string"}},
        "required":   {"type": "boolean", "default": true},
        "default":    {}
      }
    }
  }
}

属性/服务/事件语义补充:

项 规则
access=r 只能设备上报,平台可读;下发 cmd 含该字段直接拒绝
access=rw 可上报也可下发;下发前按 min/max/enum 校验
access=report_only 只上报、不存最新状态(用于高频原始帧/调试字段);仍可写设备日志
is_chart 默认在设备详情画折线;数值型才允许 true
is_history 是否写 property_history;高频字段可关闭,靠聚合
is_alarm 是否参与规则引擎的默认告警候选;不等于强制告警
is_monitor 是否在设备列表/看板实时监测卡展示
is_share_perm 共享用户是否可见该属性(隐私字段如位置可 false)
recoverable 事件是否支持 clear:true 恢复;不可恢复事件只能人工关闭
async=true 服务 立即返回 accepted,结果通过 event=service_result 或 cmd_result 异步回传(长任务如清洗、OTA)

平台校验流程(thing_model/validator.py):

1. JSON Schema 校验(结构/必填/正则)
2. 业务校验:
   - identifier 产品内唯一(属性/服务/事件各自命名空间内唯一,跨类型允许同名但需告警)
   - 属性 source_field 不得重复
   - enum 值唯一且非空;bool 不得有 min/max;数值型 min<max、step>0
   - 服务 input_params 不得含只读属性;不得引用不存在的 identifier
   - 事件 level ∈ {1,2,3}
   - 字段名长度:新增字段 ≤10;豁免清单(switch1~4、air_pressure)硬编码
3. 兼容性校验(发布新版本时):
   - 删除属性/改类型/改单位/缩小量程 = 不兼容,必须升大版本(1.x → 2.0)并新建 product_key 或走设备迁移
   - 新增可选属性/扩大范围/新增枚举值 = 兼容,升小版本(1.0 → 1.1)
4. 失败返回 422 + 逐条错误:{"path":"properties[2].min","code":"MODEL_RANGE_INVALID","message":"..."}

8.4.3 版本管理与设备绑定

8.4.4 模板、品类种子与 api-mqtt 字典映射

模板实例化:model_template 是平台预置的完整物模型;创建产品时选模板 → 深拷贝到 thing_model 草稿 → 租户可增删字段 → 发布。模板升级不自动影响已实例化产品(避免"改模板炸全平台")。

品类种子清单(与 api-mqtt §7 一一对应,平台启动时幂等 upsert 到 model_template):

product_key 模板名 关键字段(identifier 直接采用协议字段) api-mqtt
rly4p3 4路三相增氧控制器 oxygen water_temp air_temp air_pressure voltage current switch1~4 signal §7.1
gw4g 4G 网关 run uptime_s fw mac signal sim_iccid sim_state mqtt_state rs485_bus cache_cnt dev_cnt devices ntp_ok cfg_ver tank_id lat lng §7.2
wq4in1 四合一水质仪 water_temp ph tds ec signal §7.3
wq8in1 八合一水质传感器 water_temp ph oxygen ec turbidity ammonia nitrite orp tds salinity sg oxygen_sat probe_st clean_cnt last_cal §7.4
ems 储能能量管理 run mode soc soh bat_volt bat_curr bat_temp chg_power dis_power pv_power grid_power load_power grid_freq chg_energy dis_energy ems_state source grid_state gen_state chg_limit dis_limit valley_chg gen_auto gen_start gen_stop tou_peak tou_valley fault_code §7.5
vfd 增氧机变频器 run mode freq set_freq mot_volt mot_curr mot_power rpm energy_sum do_map min_run_s hysteresis fault_code §7.6
feed 投饵机 run mode feeding feed_rate feed_total hopper_lvl spread_m schedule fault_code §7.7
pump 水泵 run mode flow_rate pressure mot_curr mot_power run_sum_s fault_code §7.8
gate 电动闸门/阀 pos state target_pos lim_open lim_close torque fault_code §7.9
wlevel 水位计 water_lvl distance water_temp voltage signal §7.10
wxst 气象站 air_temp air_pres humidity wind_speed wind_dir rainfall rain_rate solar_rad uv_index signal §7.11
pv 光伏 MPPT pv_volt pv_curr pv_power chg_power chg_energy bat_volt mppt_state temp fault_code §7.12
gen 柴油发电机 run state fuel_lvl fuel_rate oil_pres cool_temp rpm voltage current power run_sum_s auto_en fault_code §7.13
pmet 配电/电表监测 voltage_a/b/c current_a/b/c p_power q_power s_power pf freq energy_sum day_energy grid_state rcd_trip spd_ok signal §7.14
cam AI 视觉盒 alive cpu_load disk_used detect_cnt events fw §7.15
ras / light / net / muldo 规划品类 先建模板占位,字段按项目立项补充 §7.16

字段映射规则:

8.4.5 在线模拟器(无真机验证物模型)

8.4.6 阈值与运行参数的覆盖链(水产必需)

threshold_template 只是"品种默认值",不能直接改模板(会影响所有客户)。必须有塘口/设备级覆盖:

CREATE TABLE pond_threshold (
  id          BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
  tenant_id   BIGINT NOT NULL,
  pond_id     BIGINT NOT NULL REFERENCES pond(id),
  species     VARCHAR(32),
  items       JSONB NOT NULL,               -- {"do_suit":5,"do_warn":3,"do_danger":2,
                                            --  "do_map":[{"do":3,"freq":50},...],
                                            --  "min_run_s":600,"hysteresis":0.2,
                                            --  "report_interval_s":60,"sensor_offline_min":30}
  source      VARCHAR(16) NOT NULL DEFAULT 'user', -- template/user
  version     INT NOT NULL DEFAULT 1,
  updated_by  BIGINT,
  updated_at  TIMESTAMPTZ NOT NULL DEFAULT NOW(),
  UNIQUE (pond_id)
);

CREATE TABLE device_config (
  device_id   BIGINT NOT NULL,
  config_key  VARCHAR(32) NOT NULL,
  value       JSONB NOT NULL,
  version     INT NOT NULL DEFAULT 1,
  updated_by  BIGINT,
  updated_at  TIMESTAMPTZ NOT NULL DEFAULT NOW(),
  PRIMARY KEY (device_id, config_key)
);

生效优先级(从高到低):

设备级 device_config
  > 塘口级 pond_threshold
    > 品种模板 threshold_template
      > 产品默认(thing_model 的 default / 厂内默认)

8.5 设备管理实现规范(M4 补差)【P0】

8.5.1 设备状态机

未激活(0) --首连--> 在线(1) --keepalive超时/LWT--> 离线(2) --重连--> 在线(1)
   │                    │                              │
   │                    └--禁用(3)<--管理员------------┘
   └--扫码绑定/激活--> 在线(1)(未激活也可先绑定,首连后置1)
禁用(3) --启用--> 离线(2)(等待重连)
任意状态 --软删除--> deleted_at(保留历史,不再接受连接)

8.5.2 批量导入(CSV/Excel)

列 必填 说明
product_key 是 必须已发布
device_name 是 租户内唯一
sn 是 全局唯一;不填则按规则生成(可选)
gateway_sn 子设备必填 网关存在
modbus_addr 子设备必填 1~247,网关内唯一
pond_name 否 不存在则创建
remark 否

8.5.3 设备分组、塘口与批量控制

8.5.4 设备日志与在线调试

8.5.5 默认组件契约(M4 必须)

前端按物模型动态渲染,组件协议固定:

// GET /devices/{id}/ui 返回
{
  "device": {"id":"1024","name":"1号塘增氧机","sn":"sz0001","product_key":"rly4p3","online":true},
  "groups": [
    {"title":"水质","components":[
      {"type":"metric","identifier":"oxygen","label":"溶解氧","value":5.2,"unit":"mg/L","precision":1,"status":"normal","chart":true},
      {"type":"metric","identifier":"water_temp","label":"水温","value":25.6,"unit":"℃","precision":1,"chart":true}
    ]},
    {"title":"控制","components":[
      {"type":"switch","identifier":"switch1","label":"1号增氧机","value":true,"writable":true,"confirm":true},
      {"type":"select","identifier":"sw1_mode","label":"1号模式","value":"auto","options":[{"value":"auto","label":"自动"},{"value":"manual_on","label":"手动开"},{"value":"manual_off","label":"手动关"}],"writable":true}
    ]},
    {"title":"告警","components":[
      {"type":"alarm","identifier":"do_low","level":2,"label":"溶氧偏低","active":false}
    ]}
  ],
  "updated_at":"2026-10-24T10:00:00+08:00"
}
物模型 组件 交互
bool + rw switch 点击弹确认 → POST /devices/{id}/service/relay_ctrl
bool + r indicator 只读状态灯
int/float + unit metric 数值 + 单位 + 图表(is_chart)
enum select / segmented rw 可改;r 只显示标签
string text 文本卡
event level≥2 alarm 红/黄卡片,点击跳告警详情
service action 按钮 + 参数表单(按 input_params 生成)

8.5.6 设备详情 API 响应(前后端联调基线)

// GET /api/v1/devices/{id}
{
  "id": "1024", "tenant_id": "2001", "product_key": "rly4p3", "product_name": "4路三相增氧控制器",
  "device_name": "1号塘增氧机", "sn": "sz0001", "status": 1, "online": true,
  "model_version": "1.2.0", "fw_ver": "1.3.0", "signal": -72,
  "gateway": {"id":"88","gw_sn":"0025CA3B6F01","online":true},
  "pond": {"id":"12","name":"1号塘"}, "farm": {"id":"3","name":"东区基地"},
  "location": {"lng":120.1,"lat":30.2}, "last_seen": "2026-10-24T10:00:00+08:00",
  "bind": {"type":1,"role":"owner"}, "tags": ["增氧","变频"],
  "state": {"oxygen":5.2,"water_temp":25.6,"switch1":true},
  "alarms": {"unhandled":1,"latest":{"id":"9001","type":"do_low","level":2}}
}

8.6 数据存储与查询实现规范(M5 补差)【P0】

8.6.1 写入路径与一致性

物模型校验通过
  → Redis pipeline:HSET device:{id}:state(最新值)+ EXPIRE 7d
  → Redis SET device:{id}:last_seen now EX 180
  → 内存批次(500 条/200ms)→ property_history(按 ts 分区)
  → 更新 device_state(upsert,兜底持久)
  → 交规则引擎

8.6.2 历史表补差 DDL(分区 + 聚合 + 保留)

-- 父表补充列(v5.0)
ALTER TABLE property_history ADD COLUMN tenant_id BIGINT;
ALTER TABLE property_history ADD COLUMN product_key VARCHAR(32);
ALTER TABLE property_history ADD COLUMN quality SMALLINT NOT NULL DEFAULT 1; -- 1正常 2补传 3异常 4模拟
ALTER TABLE property_history ADD COLUMN simulated BOOLEAN NOT NULL DEFAULT FALSE;
ALTER TABLE property_history ADD COLUMN value_json JSONB;  -- array/object 属性;数值/字符串仍走 value/value_str
CREATE UNIQUE INDEX uq_ph_dev_ident_ts ON property_history (device_id, identifier, ts);

-- 小时聚合(原始 90 天 → 小时 365 天)
CREATE TABLE property_hourly (
  device_id  BIGINT NOT NULL,
  identifier VARCHAR(32) NOT NULL,
  bucket     TIMESTAMPTZ NOT NULL,          -- date_trunc('hour', ts)
  avg_v      DOUBLE PRECISION, min_v DOUBLE PRECISION, max_v DOUBLE PRECISION,
  sum_v      DOUBLE PRECISION, cnt BIGINT, first_v DOUBLE PRECISION, last_v DOUBLE PRECISION,
  first_str  VARCHAR(255), last_str VARCHAR(255), first_json JSONB, last_json JSONB,
  tenant_id  BIGINT, product_key VARCHAR(32),
  PRIMARY KEY (device_id, identifier, bucket)
);
CREATE INDEX idx_ph_hourly_tenant ON property_hourly (tenant_id, bucket DESC);

-- 天聚合(长期)
CREATE TABLE property_daily (
  device_id  BIGINT NOT NULL,
  identifier VARCHAR(32) NOT NULL,
  bucket     DATE NOT NULL,
  avg_v      DOUBLE PRECISION, min_v DOUBLE PRECISION, max_v DOUBLE PRECISION,
  sum_v      DOUBLE PRECISION, cnt BIGINT, first_v DOUBLE PRECISION, last_v DOUBLE PRECISION,
  first_str  VARCHAR(255), last_str VARCHAR(255), first_json JSONB, last_json JSONB,
  tenant_id  BIGINT, product_key VARCHAR(32),
  PRIMARY KEY (device_id, identifier, bucket)
);

-- 分区与保留策略(配置化,不硬编码)
CREATE TABLE retention_policy (
  id            BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
  tenant_id     BIGINT NOT NULL DEFAULT 0,   -- 0=平台默认
  data_kind     VARCHAR(16) NOT NULL,        -- raw/hourly/daily/device_log/alarm/audit
  hot_days      INT NOT NULL,                -- 热库保留
  archive_days  INT NOT NULL DEFAULT 0,      -- 归档保留(0=永久)
  enabled       BOOLEAN NOT NULL DEFAULT TRUE
);

CREATE TABLE partition_task (
  id             BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
  table_name     VARCHAR(64) NOT NULL,
  partition_name VARCHAR(64) NOT NULL,
  start_at       DATE NOT NULL, end_at DATE NOT NULL,
  status         SMALLINT NOT NULL DEFAULT 0,   -- 0待建 1已建 2归档 3已删
  created_at     TIMESTAMPTZ NOT NULL DEFAULT now(),
  UNIQUE (table_name, partition_name)
);

-- 异步导出任务
CREATE TABLE export_task (
  id         BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
  tenant_id  BIGINT NOT NULL,
  user_id    BIGINT NOT NULL,
  kind       VARCHAR(32) NOT NULL,            -- history/alarm/device/workorder
  params     JSONB NOT NULL,
  status     SMALLINT NOT NULL DEFAULT 0,     -- 0排队 1生成中 2完成 3失败
  file_key   VARCHAR(255), error TEXT,
  created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
  finished_at TIMESTAMPTZ
);

-- 数据删除/注销申请(合规:可删除的删除,法定保留的匿名化)
CREATE TABLE data_deletion_request (
  id           BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
  tenant_id    BIGINT NOT NULL,
  user_id      BIGINT NOT NULL,
  scope        VARCHAR(16) NOT NULL,            -- account/device/pond/tenant/history
  target_id    BIGINT,
  reason       TEXT,
  status       SMALLINT NOT NULL DEFAULT 0,     -- 0待审 1执行中 2完成 3拒绝 4部分保留
  approved_by  BIGINT,
  result       JSONB,                           -- 删除/匿名化条数与范围
  requested_at TIMESTAMPTZ NOT NULL DEFAULT now(),
  executed_at  TIMESTAMPTZ
);

数据删除规则:账号/设备/塘口按 scope 执行;订单、审计、发票等法定保留数据只做匿名化(用户 ID 替换、手机号/地址脱敏),不做物理删除;删除任务异步执行、可审计、完成后通知;删除期间保留只读凭证供监管核查。

-- 写失败死信(历史/日志)
CREATE TABLE write_dlq (
  id         BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
  kind       VARCHAR(32) NOT NULL,            -- property_history/device_log/...
  payload    JSONB NOT NULL,
  error      TEXT, retry_count SMALLINT NOT NULL DEFAULT 0,
  status     SMALLINT NOT NULL DEFAULT 0,     -- 0待重试 1成功 2放弃
  created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

-- 定时任务日志
CREATE TABLE job_log (
  id         BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
  job_name   VARCHAR(64) NOT NULL,
  status     SMALLINT NOT NULL,               -- 0成功 1失败
  message    TEXT, duration_ms INT,
  created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

属性值落列规则:int/float → value;bool → value(1/0,读出还原);string/enum → value_str;array/object → value_json。聚合只对数值型做 avg/min/max/sum;字符串/数组只保留 first/last。

默认保留策略:原始 90 天 → 小时 365 天 → 天 3650 天;设备日志 30 天热 + 180 天归档;告警/审计 180 天热 + 3 年归档;OTA/订单永久。

分区任务:每月 25 日预建下月分区;每日 02:00 聚合前一天数据(幂等 upsert);每月 1 日归档过期分区(导出 Parquet 到 OSS 后 detach/drop)。任务必须分布式锁(Redis lock:job:{name} TTL 1h)。

8.6.3 查询 API 语义

GET /api/v1/data/{device_id}/history
  ?identifiers=oxygen,water_temp
  &from=2026-10-01T00:00:00%2B08:00&to=2026-10-02T00:00:00%2B08:00
  &interval=raw|1h|1d|auto
  &agg=avg|min|max|first|last|sum
  &fill=none|previous|zero
  &limit=5000
参数 规则
interval=auto 跨度 ≤6h → raw;≤30d → 1h;>30d → 1d
最大点数 5000;超过则自动升采样或返回 422 RANGE_TOO_LARGE
时区 默认设备时区(Asia/Shanghai),tz 可覆盖
缺测 fill=previous 前值填充;none 返回 null;不得用 0 冒充缺测
补传/异常 默认只返回 quality=1;include_quality=true 时返回 quality
权限 用户必须对该设备有读权限;共享只读按 is_share_perm 过滤字段

响应:

{
  "device_id":"1024","from":"...","to":"...","interval":"1h",
  "series":[{"identifier":"oxygen","unit":"mg/L","points":[["2026-10-01T00:00:00+08:00",5.2],["...",5.0]]}]
}

8.6.4 Redis 键注册表(必须按此实现,禁止随手起键名)

M5 中的 device:{id}:state 是"逻辑键";实现统一加租户前缀为 t:{tenant}:device:{id}:state,由 DeviceStateRepository 封装,业务代码不得直接拼接键名。迁移期可双读旧键,写只写新键。

Key 类型 TTL 用途
t:{tenant}:device:{id}:state Hash 7d 最新属性
device:{id}:online String 180s 在线标记(心跳续期)
device:{id}:last_seen String 7d 最近消息时间戳
device:{id}:desired Hash 24h 离线期望状态(含 __ver/__ts)
device:{id}:seq String 永久 命令 seq 计数器
mqtt:dedup:{sn}:{key} String 24h 报文去重
rule:{rule_id}:{device_id}:{ident}:since String 1h 持续时长起点
rule:cooldown:{rule_id}:{scope} String 规则冷却 冷却
auth:refresh:{jti} String 30d refresh token
user:{id}:ver String 永久 权限版本
product:{key}:model:{ver} String 10min 物模型缓存
app:mqtt:{token} String 15min 小程序 MQTT 短期 token
lock:job:{name} String 1h 定时任务锁
ratelimit:{scope}:{id} String 窗口 限流

8.6.5 时序库演进(>500 台)

-- TDengine 超级表(每设备一子表)
CREATE STABLE property_history (
  ts TIMESTAMP, value DOUBLE, quality TINYINT
) TAGS (device_id BIGINT, tenant_id BIGINT, product_key NCHAR(32), identifier NCHAR(32));
CREATE TABLE d_1024_oxygen USING property_history TAGS (1024, 2001, 'rly4p3', 'oxygen');

8.7 规则引擎、告警与通知实现规范(M6 补差)【P0】

8.7.1 规则数据模型补差

ALTER TABLE rule ADD COLUMN version INT NOT NULL DEFAULT 1;
ALTER TABLE rule ADD COLUMN priority INT NOT NULL DEFAULT 100;       -- 越大越先
ALTER TABLE rule ADD COLUMN status SMALLINT NOT NULL DEFAULT 0;      -- 0草稿 1启用 2停用 3测试
ALTER TABLE rule ADD COLUMN stop_on_match BOOLEAN NOT NULL DEFAULT FALSE;
ALTER TABLE rule ADD COLUMN description VARCHAR(255);
ALTER TABLE rule ADD COLUMN created_by BIGINT;
ALTER TABLE rule ADD COLUMN last_run_at TIMESTAMPTZ;
ALTER TABLE rule ADD COLUMN deleted_at TIMESTAMPTZ;

8.7.2 触发器完整 Schema

{
  "type": "property",                        // property | event | cron
  "scope": {"tenant_id": 2001, "pond_id": 12, "device_ids": [1024, 1025]},
  "logic": "AND",                            // AND | OR
  "conditions": [
    {
      "device_id": null,                     // null=按 scope/product_key 批量匹配
      "product_key": "rly4p3",
      "identifier": "oxygen",
      "op": "<",                             // < <= > >= == != in between changed
      "value": 2.0,
      "value_type": "number",                // number | string | bool | property
      "source": null,                        // value_type=property 时:{"device_id":1025,"identifier":"oxygen"}
      "duration": 300,                       // 持续秒数;0=瞬时
      "hysteresis": 0.2,                     // 回差,恢复需 > value+hysteresis
      "time_window": {"start":"22:00","end":"06:00","tz":"Asia/Shanghai"}
    }
  ],
  "group": [                                 // 可选:嵌套条件组,实现 (A AND B) OR C
    {"logic":"OR","conditions":[{"identifier":"sensor_ok","op":"==","value":false}]}
  ],
  "recovery": {                              // 属性规则恢复条件;缺省用回差
    "mode": "hysteresis",                    // hysteresis | condition | event
    "condition": {"identifier":"oxygen","op":">=","value":4.0}
  },
  "event": {"identifier":"do_low","level_min":2},       // type=event 时
  "cron": {"expr":"0 6 * * *","tz":"Asia/Shanghai"}      // type=cron 时
}

跨设备联动:条件左侧支持 value_type=property + source={"device_id":..., "identifier":...},表示"用另一台设备的属性作阈值";例如"1号塘 DO 低于 2 时,开 2 号塘增氧机"。引擎在每次数据流入时解析引用值,跨产品/跨设备均在租户内允许。

8.7.3 动作完整 Schema

[
  {"type":"service","target":{"scope":"same_pond","device_ids":[]},
   "service":"relay_ctrl","params":{"ch":1,"on":true},"source":"auto","timeout_ms":5000},
  {"type":"notify","channel":["wechat","sms","voice","inapp","webhook","email"],
   "level":3,"template":"do_danger","targets":["device_binders","operators","custom"],
   "user_ids":[],"quiet_hours":{"start":"23:00","end":"06:00","except_level":3}},
  {"type":"virtual_property","device_id":1024,"identifier":"manual_confirmed","value":false},
  {"type":"forward","target_id":7,"payload":"latest"},
  {"type":"work_order","level":0,"type":"do_abnormal","title":"DO 极低自动派单","assignee_role":"operator"},
  {"type":"delay","seconds":300,"then":[{"type":"service","target":{"device_ids":[1024]},"service":"relay_ctrl","params":{"ch":1,"on":false}}]},
  {"type":"scene","scene_id":3}
]

8.7.4 规则执行日志(补"写执行日志"缺口)

CREATE TABLE rule_execution_log (
  id          BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
  tenant_id   BIGINT NOT NULL,
  rule_id     BIGINT NOT NULL,
  device_id   BIGINT,
  trigger_type VARCHAR(16) NOT NULL,          -- property/event/cron
  trigger_value JSONB,
  matched     BOOLEAN NOT NULL,
  actions     JSONB,                          -- 每个动作的结果
  success     BOOLEAN NOT NULL,
  error       TEXT,
  duration_ms INT,
  created_at  TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX idx_rel_rule_time ON rule_execution_log (rule_id, created_at DESC);

8.7.5 持续时长、冷却与恢复的实现

机制 实现
duration 首次满足时 Redis SET rule:{r}:{d}:{i}:since now NX EX 3600;后续满足时比较 now-since>=duration 才触发;不满足则删除键。重启后键仍在(Redis 持久化)
冷却 触发后 SET rule:cooldown:{r}:{scope} 1 EX cooldown;冷却内命中只记 matched=false,不执行
恢复 属性规则:回差或 recovery.condition;事件规则:同名事件 clear=true;恢复时更新 alarm_record.status=2,可配恢复通知
防抖 设备端已去抖(duration_s);云端 duration 是第二道;两者取更严格
重启 冷却/持续状态在 Redis,平台重启不丢;Redis 丢失时允许重复触发一次(可接受)
规则变更 规则更新后清空该规则的 since/cooldown 键,避免旧状态误触发

8.7.6 告警中心补差

ALTER TABLE alarm_record ADD COLUMN alarm_key VARCHAR(128);       -- 去重键:rule:device:type
ALTER TABLE alarm_record ADD COLUMN first_at TIMESTAMPTZ;
ALTER TABLE alarm_record ADD COLUMN last_at TIMESTAMPTZ;
ALTER TABLE alarm_record ADD COLUMN repeat_count INT NOT NULL DEFAULT 1;
ALTER TABLE alarm_record ADD COLUMN cleared_at TIMESTAMPTZ;
ALTER TABLE alarm_record ADD COLUMN closed_at TIMESTAMPTZ;
ALTER TABLE alarm_record ADD COLUMN close_reason VARCHAR(255);
ALTER TABLE alarm_record ADD COLUMN assignee BIGINT;
ALTER TABLE alarm_record ADD COLUMN notify_status SMALLINT NOT NULL DEFAULT 0; -- 0未发 1已发 2失败
ALTER TABLE alarm_record ADD COLUMN recovered_value JSONB;
CREATE UNIQUE INDEX uq_alarm_open ON alarm_record (alarm_key) WHERE status IN (0,1);

状态机:

未处理(0) --ack--> 已确认(1) --恢复--> 已恢复(2) --关闭--> 已关闭(3)
    │                  │
    └──恢复────────────┴──> 已恢复(2)
已恢复(2) --人工关闭/超时--> 已关闭(3)

8.7.7 通知渠道与站内信(补"通知"缺口)

CREATE TABLE notify_channel (
  id         BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
  tenant_id  BIGINT NOT NULL,
  type       VARCHAR(16) NOT NULL,          -- wechat/sms/voice/email/webhook/inapp
  name       VARCHAR(64) NOT NULL,
  config     JSONB NOT NULL,                -- 加密存储:appid/secret/template_id/签名/供应商
  enabled    BOOLEAN NOT NULL DEFAULT TRUE
);

CREATE TABLE notify_template (
  id         BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
  tenant_id  BIGINT NOT NULL DEFAULT 0,
  code       VARCHAR(64) NOT NULL,          -- do_danger / offline / subscription_expire
  channel    VARCHAR(16) NOT NULL,
  title      VARCHAR(128),
  content    TEXT NOT NULL,                 -- 变量占位 {{device_name}} {{value}}
  provider_template_id VARCHAR(128),
  UNIQUE (tenant_id, code, channel)
);

CREATE TABLE notification (                 -- 站内信
  id         BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
  tenant_id  BIGINT NOT NULL,
  user_id    BIGINT NOT NULL,
  type       VARCHAR(32) NOT NULL,
  title      VARCHAR(128) NOT NULL,
  content    TEXT,
  ref_type   VARCHAR(32), ref_id BIGINT,
  read_at    TIMESTAMPTZ,
  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX idx_notify_user ON notification (user_id, read_at, created_at DESC);

CREATE TABLE notify_log (
  id           BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
  tenant_id    BIGINT NOT NULL,
  channel      VARCHAR(16) NOT NULL,
  template     VARCHAR(64),
  target       VARCHAR(255),                -- openid/phone/email/url(脱敏)
  ref_type     VARCHAR(32), ref_id BIGINT,
  status       SMALLINT NOT NULL DEFAULT 0, -- 0待发 1成功 2失败 3跳过
  retry_count  SMALLINT NOT NULL DEFAULT 0,
  request      JSONB, response JSONB, error TEXT,
  created_at   TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

-- 用户级通知偏好:渠道、设备静音、级别、免打扰
CREATE TABLE user_notify_pref (
  id            BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
  tenant_id     BIGINT NOT NULL,
  user_id       BIGINT NOT NULL,
  channels      JSONB NOT NULL DEFAULT '["inapp"]',   -- inapp/wechat/sms/voice/email
  device_mute   JSONB NOT NULL DEFAULT '[]',          -- 静音设备 id 列表
  alarm_types   JSONB NOT NULL DEFAULT '[]',          -- 关注的告警类型,空=全部
  level_min     SMALLINT NOT NULL DEFAULT 1,          -- 最低级别:1/2/3
  quiet_hours   JSONB,                                -- {"start":"23:00","end":"07:00","except_level":3,"tz":"Asia/Shanghai"}
  updated_at    TIMESTAMPTZ NOT NULL DEFAULT NOW(),
  UNIQUE (user_id)
);

通知偏好规则:用户级 user_notify_pref 覆盖规则动作的默认渠道;level_min 以下只发站内信;device_mute 命中的设备不发外部渠道;免打扰时段内非 except_level 只发站内信。通知服务按"用户偏好 → 规则渠道 → 租户默认"三级解析,最终写 notify_log。

渠道实现要点:

渠道 接入 约束
微信订阅消息 小程序 wx.requestSubscribeMessage 用户逐次授权;服务号模板消息 必须存用户授权记录;模板 ID 按环境配置;失败不重试超过 3 次
短信 阿里云/腾讯云短信 签名/模板审核;同手机号同模板 1 分钟 1 条、1 天 10 条
语音电话 阿里云语音通知 仅 level=3 且用户已授权;夜间静音可配
邮件 SMTP 用于报表/对账单
Webhook 第三方 URL HMAC 签名 + 重试(§8.8)
站内信 DB + WebSocket 推送 所有通知默认写站内信,其他渠道可失败

8.7.8 场景联动(补"场景"数据模型)

CREATE TABLE scene (
  id         BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
  tenant_id  BIGINT NOT NULL,
  name       VARCHAR(64) NOT NULL,
  trigger_config JSONB NOT NULL,             -- 同 rule.trigger
  actions    JSONB NOT NULL,                 -- 有序动作数组,支持 parallel
  enabled    BOOLEAN NOT NULL DEFAULT TRUE,
  created_by BIGINT,
  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

8.7.9 数据转发(补投递可靠性)

ALTER TABLE data_forward ADD COLUMN last_success_at TIMESTAMPTZ;
ALTER TABLE data_forward ADD COLUMN fail_count INT NOT NULL DEFAULT 0;
CREATE TABLE data_forward_log (
  id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
  forward_id BIGINT NOT NULL,
  tenant_id BIGINT NOT NULL,
  payload JSONB, status SMALLINT NOT NULL,   -- 0待发 1成功 2失败 3丢弃
  retry_count SMALLINT NOT NULL DEFAULT 0,
  error TEXT, created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

8.7.10 调度任务清单(BullMQ/node-cron 或 APScheduler)

任务 频率 说明
partition_maintain 每日 02:00 预建/归档分区
history_aggregate 每小时 05 分 上一小时聚合
daily_aggregate 每日 00:10 前一天聚合
offline_check 每分钟 扫描 last_seen 超时置离线(兜底 LWT)
alarm_escalate 每分钟 严重告警升级
command_timeout 每 10s 命令超时/重试
subscription_scan 每日 00:20 到期提醒/降级
report_generate 每日 01:00 日/周/月报
data_forward_retry 每 5 分钟 失败重投
audit_archive 每月 1 日 审计归档
cleanup 每日 03:00 过期模拟数据/去重键/站内信

8.8 开放平台实现规范(M7 补差)【P1】

8.8.1 AccessKey 与签名

ALTER TABLE access_key ADD COLUMN name VARCHAR(64);
ALTER TABLE access_key ADD COLUMN expires_at TIMESTAMPTZ;
ALTER TABLE access_key ADD COLUMN last_used_at TIMESTAMPTZ;
ALTER TABLE access_key ADD COLUMN ip_whitelist JSONB;
ALTER TABLE access_key ADD COLUMN created_by BIGINT;

请求头:

X-AK: ak_2f8c...
X-Timestamp: 1790000000            # 秒级,±300s
X-Nonce: 8f3a...                   # 16~64 字符,10 分钟内不可重复
X-Signature: base64(HMAC-SHA256(secret, canonical))
Content-Type: application/json

规范串:

canonical = HTTP_METHOD + "\n"
          + PATH + "\n"
          + sorted_query_string + "\n"
          + lower_hex(SHA256(raw_body)) + "\n"
          + timestamp + "\n"
          + nonce

8.8.2 开放 API 清单(只读优先)

GET  /open/v1/devices                     设备列表(分页/筛选)
GET  /open/v1/devices/{id}                设备详情
GET  /open/v1/devices/{id}/state          最新属性
GET  /open/v1/devices/{id}/history        历史(同 §8.6.3)
GET  /open/v1/devices/{id}/alarms         告警
GET  /open/v1/alarms                      告警列表
GET  /open/v1/ponds                       塘口列表
GET  /open/v1/ponds/{id}/overview         塘口聚合
POST /open/v1/devices/{id}/service/{sid}  控制(需 device:control scope,白名单)
POST /open/v1/webhooks                    注册 Webhook
GET  /open/v1/webhooks                    查询
DELETE /open/v1/webhooks/{id}             删除

8.8.3 Webhook

ALTER TABLE webhook_target ADD COLUMN secret_hash VARCHAR(128);
ALTER TABLE webhook_target ADD COLUMN description VARCHAR(255);
ALTER TABLE webhook_target ADD COLUMN last_success_at TIMESTAMPTZ;
ALTER TABLE webhook_target ADD COLUMN fail_count INT NOT NULL DEFAULT 0;
CREATE TABLE webhook_delivery (
  id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
  target_id BIGINT NOT NULL,
  event_id VARCHAR(64) NOT NULL,
  event_type VARCHAR(32) NOT NULL,
  payload JSONB NOT NULL,
  status SMALLINT NOT NULL DEFAULT 0,        -- 0待发 1成功 2失败 3丢弃
  retry_count SMALLINT NOT NULL DEFAULT 0,
  next_retry_at TIMESTAMPTZ,
  response_code INT, response_body TEXT,
  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
  UNIQUE (target_id, event_id)
);

投递协议:

POST <url>
X-Aqua-Event: alarm.created | alarm.recovered | device.online | device.offline | ota.progress
X-Aqua-Delivery: <event_id>
X-Aqua-Timestamp: 1790000000
X-Aqua-Signature: sha256=HMAC(secret, timestamp + "." + body)

Payload 统一包络:

{
  "event": "alarm.created",
  "event_id": "evt_01H...",
  "occurred_at": "2026-10-24T10:00:00+08:00",
  "tenant_id": "2001",
  "data": { }
}
// alarm.created
{"event":"alarm.created","event_id":"evt_1","occurred_at":"2026-10-24T10:00:00+08:00","tenant_id":"2001",
 "data":{"alarm_id":"9001","device_id":"1024","pond_id":"12","type":"do_low","level":2,
         "trigger_value":{"oxygen":2.9},"message":"溶氧偏低","rule_id":"7"}}

// alarm.recovered
{"event":"alarm.recovered","event_id":"evt_2","occurred_at":"2026-10-24T10:20:00+08:00","tenant_id":"2001",
 "data":{"alarm_id":"9001","device_id":"1024","recovered_value":{"oxygen":4.6}}}

// device.online / device.offline
{"event":"device.offline","event_id":"evt_3","occurred_at":"2026-10-24T10:05:00+08:00","tenant_id":"2001",
 "data":{"device_id":"1024","sn":"sz0001","product_key":"rly4p3","reason":"keepalive_timeout"}}

// ota.progress
{"event":"ota.progress","event_id":"evt_4","occurred_at":"2026-10-24T10:10:00+08:00","tenant_id":"2001",
 "data":{"task_id":"31","device_id":"1024","progress":60,"status":"downloading",
         "from_version":"1.2.0","to_version":"1.3.0"}}

8.8.4 数据转发授权

8.9 可选模块实现规范(O1~O7 补差)【P1/P2】

8.9.1 O1 静态文件显示系统(自定义 HTML)

补差字段:

ALTER TABLE product_ui ADD COLUMN vendor_id BIGINT;
ALTER TABLE product_ui ADD COLUMN file_size BIGINT;
ALTER TABLE product_ui ADD COLUMN file_md5 VARCHAR(64);
ALTER TABLE product_ui ADD COLUMN published_at TIMESTAMPTZ;
ALTER TABLE product_ui ADD COLUMN created_by BIGINT;

上传流程:

POST /products/{id}/ui/upload(multipart,zip 或单文件)
  → 校验扩展名白名单(.html/.css/.js/.png/.jpg/.jpeg/.svg/.woff2)
  → 解压到临时目录:拒绝绝对路径/../符号链接/zip bomb(解压后 ≤20MB,文件数 ≤200)
  → MIME 嗅探(不得仅看扩展名);HTML 必须含 index.html
  → 可选病毒扫描(ClamAV)
  → 写入 static/ui/{tenant_id}/{product_key}/versions/{timestamp}/
  → 元数据入库 status=0(待发布)
  → POST /products/{id}/ui/{version}/publish 原子切换 current 软链
  → 失败自动回退上一版本

安全:

桥接协议(完整):

// 平台 → iframe(加载完成)
{"type":"init","deviceId":"1024","model":{...},"props":{"oxygen":5.2},"token":"<15min>","nonce":"n1"}
// iframe → 平台
{"type":"ready","nonce":"n1"}
{"type":"getLatestProps","nonce":"n2"}
{"type":"subscribeProps","identifiers":["oxygen","switch1"],"nonce":"n3"}
{"type":"invokeService","service":"relay_ctrl","params":{"ch":1,"on":true},"nonce":"n4"}
// 平台 → iframe
{"type":"propsUpdate","data":{"oxygen":5.2},"nonce":"n3"}
{"type":"serviceResult","requestId":"REQ-...","success":true,"message":"...","nonce":"n4"}
{"type":"error","code":"TOKEN_EXPIRED","message":"..."}

8.9.2 O2 可视化看板

ALTER TABLE dashboard ADD COLUMN description VARCHAR(255);
ALTER TABLE dashboard ADD COLUMN is_default BOOLEAN NOT NULL DEFAULT FALSE;
ALTER TABLE dashboard ADD COLUMN share_token VARCHAR(64);
ALTER TABLE dashboard ADD COLUMN share_expires_at TIMESTAMPTZ;
ALTER TABLE dashboard ADD COLUMN updated_by BIGINT;

8.9.3 O3 OTA 固件升级

ALTER TABLE firmware ADD COLUMN hardware_version VARCHAR(32);
ALTER TABLE firmware ADD COLUMN file_size BIGINT;
ALTER TABLE firmware ADD COLUMN changelog TEXT;
ALTER TABLE firmware ADD COLUMN min_battery INT;
CREATE TABLE ota_device (
  id           BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
  ota_task_id  BIGINT NOT NULL REFERENCES ota_task(id),
  device_id    BIGINT NOT NULL,
  status       SMALLINT NOT NULL DEFAULT 0,   -- 0待推送 1已下发 2下载中 3校验中 4升级中 5成功 6失败 7回滚
  progress     SMALLINT NOT NULL DEFAULT 0,
  from_version VARCHAR(16), to_version VARCHAR(16),
  error        TEXT, retry_count SMALLINT NOT NULL DEFAULT 0,
  started_at   TIMESTAMPTZ, finished_at TIMESTAMPTZ,
  UNIQUE (ota_task_id, device_id)
);

-- 固件签名公钥(私钥离线/HSM,平台只存公钥)
CREATE TABLE firmware_signing_key (
  id          BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
  vendor_id   BIGINT NOT NULL,
  key_id      VARCHAR(64) NOT NULL,
  algorithm   VARCHAR(16) NOT NULL DEFAULT 'ECDSA_P256',
  public_key  TEXT NOT NULL,
  status      SMALLINT NOT NULL DEFAULT 1,     -- 1启用 2停用 3吊销
  created_at  TIMESTAMPTZ NOT NULL DEFAULT NOW(),
  retired_at  TIMESTAMPTZ,
  revoked_at  TIMESTAMPTZ,
  UNIQUE (vendor_id, key_id)
);
ALTER TABLE firmware ADD COLUMN signing_key_id VARCHAR(64);
ALTER TABLE ota_task ADD COLUMN task_type VARCHAR(16) NOT NULL DEFAULT 'upgrade'; -- upgrade/rollback
ALTER TABLE ota_task ADD COLUMN parent_task_id BIGINT;
ALTER TABLE ota_task ADD COLUMN paused_at TIMESTAMPTZ;

签名与回滚规则:私钥不落平台/不入库,签名在离线机或 HSM 完成;平台登记 key_id + public_key,设备端按 key_id 选公钥验签。密钥轮换时旧 key 先停用、观察一个灰度周期再吊销;吊销后禁止新任务使用。回滚任务 = 新建 ota_task(task_type=rollback, parent_task_id=原任务),复用 ota_device 逐台跟踪,可暂停/取消/重试。

流程:

  1. 上传固件到 OSS 私有桶 → 计算 MD5/SHA256 → ECDSA 签名(私钥离线)→ firmware.status=0。
  2. 发布固件 → 选择产品/硬件版本兼容范围 → 创建 ota_task(按 sn 段/百分比/分组/指定设备)。
  3. 生成每台设备 ota_device;按批次下发 {"type":"ota","cmd":"start","url":"<签名URL>","version":"1.3.0","md5":"...","size":123,"seq":...}。
  4. 设备下载 → 校验 MD5 + 签名 → 写备用分区 → 重启 → 上报 event=ota_progress(0~100)与 cmd_result。
  5. 平台更新 ota_device;失败自动重试 1 次;仍失败标记 6 并告警。
  6. 回滚:设备启动 30s 内看门狗复位 → 自动回滚;5 分钟内 3 次非预期复位 → Safe Mode(只上报不动作);平台可发起"回滚任务"。
  7. BLE 备通道:设备离线且现场有人员时,通过蓝牙小程序直传固件(分片、断点续传、校验),结果回传平台 ota_device.status=5,channel=ble。

约束:灰度按 5%→20%→50%→100%;每批观察 2h;同设备 24h 内最多 1 次 OTA;低电量/信号差(signal<-100)延后;OTA 不影响本地安全控制。

8.9.4 O4 订阅计费

CREATE TABLE plan (
  id          BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
  code        VARCHAR(16) NOT NULL UNIQUE,    -- trial/basic/standard/flagship
  name        VARCHAR(32) NOT NULL,
  price_cents INT NOT NULL,
  period      VARCHAR(8) NOT NULL DEFAULT 'year',
  trial_days  INT NOT NULL DEFAULT 0,
  features    JSONB NOT NULL,                 -- {"max_devices":5,"history_days":30,"sms":true,"voice":false,"ota":true}
  enabled     BOOLEAN NOT NULL DEFAULT TRUE
);

-- subscription 以 §4.4 O4 建表为基础,补充:
ALTER TABLE subscription ADD COLUMN auto_renew BOOLEAN NOT NULL DEFAULT FALSE;
ALTER TABLE subscription ADD COLUMN created_at TIMESTAMPTZ NOT NULL DEFAULT NOW();
-- status 取值统一为:1正常 2宽限 3受限 4过期 5取消

CREATE TABLE payment_log (
  id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
  order_id BIGINT NOT NULL,
  channel VARCHAR(16) NOT NULL DEFAULT 'wechat',
  transaction_id VARCHAR(64),
  amount_cents INT NOT NULL,
  status SMALLINT NOT NULL,                   -- 0待支付 1成功 2失败 3退款
  raw JSONB, created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

套餐能力矩阵(默认,可后台改):

能力 试用 基础 标准 旗舰
设备数 2 5 20 100
历史天数 7 30 180 365
微信告警 ✅ ✅ ✅ ✅
短信 ❌ ✅ ✅ ✅
电话 ❌ ❌ ❌ ✅
OTA ❌ ✅ ✅ ✅
开放 API ❌ ❌ ✅ ✅
自定义 UI ❌ ❌ ✅ ✅
报表 ❌ 月报 周/月 日/周/月

8.9.5 O5 工单与远程运维

CREATE TABLE work_order_log (
  id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
  work_order_id BIGINT NOT NULL,
  action VARCHAR(32) NOT NULL,               -- create/assign/accept/diagnose/resolve/close/reopen/feedback
  from_status SMALLINT, to_status SMALLINT,
  operator_id BIGINT, note TEXT, attachments JSONB,
  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE TABLE work_order_comment (
  id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
  work_order_id BIGINT NOT NULL,
  user_id BIGINT NOT NULL,
  content TEXT NOT NULL,
  internal BOOLEAN NOT NULL DEFAULT FALSE,   -- 内部备注客户不可见
  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

8.9.6 O6 数据增值服务

CREATE TABLE report_template (
  id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
  tenant_id BIGINT NOT NULL DEFAULT 0,
  code VARCHAR(32) NOT NULL,                 -- water_quality/energy/feeding/maintenance
  name VARCHAR(64) NOT NULL,
  content JSONB NOT NULL,                    -- 章节/指标/图表定义
  standard_ref VARCHAR(64),                  -- GB 11607-89 等
  enabled BOOLEAN NOT NULL DEFAULT TRUE,
  UNIQUE (tenant_id, code)
);
ALTER TABLE report ADD COLUMN template_id BIGINT;
ALTER TABLE report ADD COLUMN status SMALLINT NOT NULL DEFAULT 0; -- 0生成中 1完成 2失败
ALTER TABLE report ADD COLUMN error TEXT;

养殖经营记录(V3,最小表)

CREATE TABLE pond_operation_log (
  id          BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
  tenant_id   BIGINT NOT NULL,
  pond_id     BIGINT NOT NULL REFERENCES pond(id),
  batch_id    BIGINT REFERENCES batch(id),
  type        VARCHAR(16) NOT NULL,            -- feed/med/patrol/mortality/harvest
  payload     JSONB NOT NULL,                  -- 按 type 约定,见下
  operator    BIGINT,
  occurred_at TIMESTAMPTZ NOT NULL,
  attachments JSONB,                           -- 照片/单据 OSS key
  created_at  TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE INDEX idx_pol_pond_time ON pond_operation_log (pond_id, occurred_at DESC);
CREATE INDEX idx_pol_tenant_type ON pond_operation_log (tenant_id, type, occurred_at DESC);

8.9.7 O7 能力扩展位

// Node.js(推荐)
export interface ProtocolAdapter {
  name: string;
  match(topic: string, payload: Buffer): boolean;
  decode(raw: Buffer): Record<string, unknown>;   // 转成统一上行骨架
  encode(cmd: Record<string, unknown>): Buffer;   // 统一下行转协议帧
}
// FastAPI(备选)用 typing.Protocol 定义同名三方法即可
-- 可选扩展表(按项目启用;不启用不影响基础平台)
CREATE TABLE video_channel (
  id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
  tenant_id BIGINT NOT NULL, device_id BIGINT,
  gb_id VARCHAR(64), name VARCHAR(64), stream_url VARCHAR(255),
  status SMALLINT NOT NULL DEFAULT 0, created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE TABLE ai_event (
  id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
  tenant_id BIGINT NOT NULL, device_id BIGINT NOT NULL,
  type VARCHAR(32) NOT NULL, confidence DOUBLE PRECISION,
  image_key VARCHAR(255), payload JSONB,
  created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE TABLE spare_part (
  id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
  tenant_id BIGINT NOT NULL, product_key VARCHAR(32),
  name VARCHAR(64) NOT NULL, life_days INT, stock INT DEFAULT 0
);

8.10 前端与移动端实现规范【P0】

8.10.1 Web 管理后台页面地图

模块 路由 页面/能力 权限
登录 /login 账密/短信/扫码;忘记密码 公开
首页 /dashboard 设备总数/在线/离线、今日告警、订阅到期、趋势图 登录
租户 /tenants 树、创建/停用/过户、成员/邀请 platform_admin
用户/角色 /users、/roles 成员、角色、权限点、强制下线 tenant_admin
产品 /products 列表/扁平分组(一级)/创建/发布/物模型编辑器/模板/模拟器 product:*
设备 /devices 列表/详情/实时/历史/日志/分组/绑定/共享/批量导入/地图 device:*
塘口 /farms、/ponds 基地/塘口档案、设备挂载、总览、经营记录(投喂/用药/巡塘/死亡/出塘) pond:manage
规则 /rules 列表/编辑器/测试/启停/执行日志 rule:*
告警 /alarms 列表/详情/确认/关闭/统计/导出 alarm:*
通知 /notify 渠道/模板/站内信/用户偏好/重试 tenant_admin
看板 /dashboards 列表/设计器/共享 data:read
OTA /ota 固件/任务/灰度/进度/回滚/签名密钥 ota:*
工单 /workorders 列表/派单/处理/SLA/回访 workorder:*
计费 /billing 套餐/订单/续费/发票/对账 order:*
开放平台 /openapi AccessKey/Webhook/转发/文档 openapi:manage
配额 /system/quota 租户设备/用户/API/存储配额 platform_admin
系统 /system 字典/参数/日志/监控/备份 platform_admin

前端实现约定:

8.10.2 微信小程序页面地图(养殖户)

页面 能力
登录/授权 wx.login、手机号绑定、隐私政策勾选
首页 我的塘口卡片:DO/水温/在线/告警;一键全部增氧(仅运维)
塘口详情 多设备聚合、24h 曲线、设备列表
设备详情 默认组件/自定义 UI、实时数据、控制(按权限)、历史、告警
告警中心 列表/详情/确认、订阅消息授权、免打扰设置
报表 水质/节能报告列表与分享
我的 家庭/成员、设备共享、订阅续费、消息、设置、联系客服
绑定设备 扫码、输入 sn+激活码、申请转移
运维(独立入口) 工单、远程诊断、批量控制、OTA、调试

8.10.3 实时推送通道

Web/小程序 → wss://api.example.com/ws?token=<JWT>
  → 鉴权(JWT + 设备/租户范围)
  → 订阅:{"action":"subscribe","topics":["device:1024","alarm:tenant"]}
  → 服务端推送:{"type":"state","device_id":1024,"data":{...},"ts":"..."}
                 {"type":"alarm","data":{...}}
                 {"type":"cmd_result","request_id":"...","success":true}

8.10.4 交互与可用性基线

8.11 API 约定、错误码与兼容性【P0】

8.11.1 统一响应

// 成功
{"code":0,"message":"ok","data":{...},"request_id":"req_01H..."}
// 失败
{"code":40001,"message":"参数错误","data":null,"request_id":"req_01H...","details":[{"field":"sn","reason":"不能为空"}]}

8.11.2 错误码目录(节选,完整表进代码常量)

code HTTP 含义 处理建议
0 200 成功 —
40001 400 参数错误 看 details
40101 401 未登录/token 过期 刷新或重新登录
40102 401 签名错误 检查 AccessKey/规范串
40301 403 无权限 申请权限
40302 403 跨租户访问 禁止
40401 404 资源不存在 —
40901 409 设备已被绑定 走转移
40902 409 重复请求 幂等返回
42201 422 物模型校验失败 看 details
42202 422 规则 JSON 非法 看 details
42901 429 限流 退避重试
40201 402 套餐限制 升级套餐
50001 500 内部错误 带 request_id 报障
50301 503 依赖不可用(DB/MQTT) 稍后重试;设备本地不受影响

设备侧错误码独立:api-mqtt §11.3(0 成功/1 未知命令/2 参数错/3 忙/4 未找到/5 无权限/6 本地安全拒绝/7 硬件失败/100+ 自定义)。平台必须原样透传到 UI。

8.11.3 接口幂等与并发

8.11.4 全量 API 路由表(管理面)

统一前缀 /api/v1;除 public/internal 外均需 JWT;权限码见 §8.2.2。列表统一支持 page/page_size/sort/filter;写接口支持 Idempotency-Key。
权限码中的 :* 表示该模块全部权限;实现时按 §8.2.2 的权限点拆分。

限流与租户配额:

CREATE TABLE tenant_quota (
  tenant_id          BIGINT PRIMARY KEY,
  max_devices        INT NOT NULL DEFAULT -1,   -- -1=不限
  max_users          INT NOT NULL DEFAULT -1,
  api_qps            INT NOT NULL DEFAULT -1,
  mqtt_msgs_per_min  INT NOT NULL DEFAULT -1,
  storage_gb         INT NOT NULL DEFAULT -1,
  max_api_keys       INT NOT NULL DEFAULT -1,
  updated_by         BIGINT,
  updated_at         TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
模块 方法与路径 权限
认证/我的 POST /auth/wechat/login(public)、POST /auth/login(public)、POST /auth/refresh(public)、POST /auth/logout、GET/PUT /auth/me、POST /auth/password、POST /auth/sms/send(public) 登录用户
租户 GET/POST /tenants、GET/PUT/DELETE /tenants/{id}、POST /tenants/{id}/disable、POST /tenants/{id}/enable、POST /tenants/{id}/transfer、GET/POST /tenants/{id}/members、DELETE /tenants/{id}/members/{uid}、POST /tenants/{id}/invites、POST /invites/{code}/accept、GET/PUT /tenants/{id}/quota platform_admin / tenant_admin
用户/角色 GET/POST /users、GET/PUT/DELETE /users/{id}、POST /users/{id}/reset-password、POST /users/{id}/force-logout、GET/POST /roles、GET/PUT/DELETE /roles/{id}、GET/PUT /roles/{id}/permissions、GET /permissions user:manage / role:manage
产品分组 GET/POST /product-groups、GET/PUT/DELETE /product-groups/{id} product:read / product:create
产品/模板 GET/POST /products、GET/PUT/DELETE /products/{id}、POST /products/{id}/publish、POST /products/{id}/deprecate、GET/POST /products/{id}/models、GET/PUT /products/{id}/models/{version}、POST /products/{id}/models/{version}/publish、POST /products/{id}/secret/reset、GET /model-templates、POST /products/from-template、POST /products/{id}/simulate product:* / model:*
设备分组 GET/POST /device-groups、GET/PUT/DELETE /device-groups/{id}、POST /device-groups/{id}/devices、DELETE /device-groups/{id}/devices/{deviceId}、POST /device-groups/{id}/commands group:manage / device:control
农场/塘口 GET/POST /farms、GET/PUT/DELETE /farms/{id}、GET/POST /ponds、GET/PUT/DELETE /ponds/{id}、GET /ponds/{id}/overview、GET/PUT /ponds/{id}/threshold、POST /ponds/{id}/devices、DELETE /ponds/{id}/devices/{deviceId}、GET/POST /ponds/{id}/operations、GET/PUT/DELETE /operations/{id}、GET /ponds/{id}/operations/stats pond:manage / data:read
设备 GET/POST /devices、GET/PUT/DELETE /devices/{id}、POST /devices/import、GET /devices/import/{taskId}、POST /devices/{id}/activate、POST /devices/{id}/bind、POST /devices/{id}/unbind、POST /devices/{id}/share、DELETE /devices/{id}/share/{userId}、POST /devices/{id}/transfer、POST /devices/{id}/service/{sid}、GET /devices/{id}/state、GET /devices/{id}/logs、GET/PUT /devices/{id}/config、POST /devices/{id}/config/sync、POST /devices/{id}/enable、POST /devices/{id}/disable、GET /devices/{id}/ui、GET /devices/{id}/commands、POST /devices/{id}/debug device:*
数据 GET /data/{deviceId}/latest、GET /data/{deviceId}/history、GET /data/{deviceId}/stats、POST /data/export、GET /data/export/{id}、DELETE /data/export/{id}、POST /data/deletion、GET /data/deletion/{id} data:read / data:export
规则/场景 GET/POST /rules、GET/PUT/DELETE /rules/{id}、POST /rules/{id}/enable、POST /rules/{id}/disable、POST /rules/{id}/test、GET /rules/{id}/logs、GET/POST /scenes、GET/PUT/DELETE /scenes/{id}、POST /scenes/{id}/run rule:*
告警 GET /alarms、GET /alarms/{id}、GET /alarms/stats、POST /alarms/{id}/ack、POST /alarms/{id}/close、GET /alarms/export alarm:*
通知 GET/PUT /notification-prefs、GET /notifications、POST /notifications/{id}/read、POST /notifications/read-all、GET/POST /notify-channels、PUT/DELETE /notify-channels/{id}、GET/POST /notify-templates、PUT/DELETE /notify-templates/{id} 本人 / tenant_admin
OTA GET/POST /ota/firmware、GET/PUT/DELETE /ota/firmware/{id}、POST /ota/firmware/{id}/publish、GET/POST /ota/keys、PUT /ota/keys/{id}/revoke、GET/POST /ota/tasks、GET /ota/tasks/{id}、GET /ota/tasks/{id}/devices、POST /ota/tasks/{id}/pause、POST /ota/tasks/{id}/cancel、POST /ota/tasks/{id}/rollback、GET /ota/devices/{id}、POST /ota/devices/{id}/retry ota:*
计费 GET /plans、GET /subscription、POST /orders、GET /orders、POST /payments/wechat/notify(public)、POST /subscriptions/{id}/cancel、GET /invoices order:* / subscription:manage
工单 GET/POST /workorders、GET/PUT /workorders/{id}、POST /workorders/{id}/assign、POST /workorders/{id}/accept、POST /workorders/{id}/resolve、POST /workorders/{id}/close、POST /workorders/{id}/reopen、GET/POST /workorders/{id}/comments、POST /workorders/{id}/feedback、GET /workorders/export workorder:*
报表 GET/POST /reports、GET /reports/{id}、POST /reports/{id}/generate、GET /reports/{id}/download、GET/POST /report-templates、PUT/DELETE /report-templates/{id} report:read / report:manage
看板/UI GET/POST /dashboards、GET/PUT/DELETE /dashboards/{id}、POST /dashboards/{id}/share、GET/POST /products/{id}/ui、POST /products/{id}/ui/upload、POST /products/{id}/ui/{version}/publish、POST /products/{id}/ui/{version}/rollback data:read / ui:manage
开放/Webhook/转发 GET/POST /open/v1/*、GET/POST /webhooks、DELETE /webhooks/{id}、POST /webhooks/{id}/test、GET /webhooks/{id}/deliveries、GET/POST /data-forwards、PUT/DELETE /data-forwards/{id}、POST /data-forwards/{id}/test openapi:manage
审计/系统 GET /audit-logs、GET /audit-logs/export、GET/PUT /system/config、GET /system/health(public)、GET /system/metrics(internal)、GET/POST /system/jobs、POST /system/jobs/{name}/run audit:read / platform_admin
内部回调 POST /internal/mqtt/auth、POST /internal/mqtt/acl、POST /internal/mqtt/lifecycle、POST /internal/device/register X-Internal-Token(仅内网)

8.12 工程、部署与运维规范【P0】

8.12.1 后端工程结构(二选一;推荐 Node.js + Express)

A. Node.js + Express(推荐):Express 5 + TypeScript + Prisma/Drizzle

backend/
├── src/
│   ├── server.ts            # http listen / graceful shutdown
│   ├── app.ts               # express app / 中间件顺序 / 路由挂载
│   ├── config/              env.ts logger.ts db.ts redis.ts
│   ├── common/              errors.ts response.ts auth.ts tenant.ts permission.ts request-id.ts
│   ├── modules/
│   │   ├── auth/ tenant/ user/ product/ thing/ device/ data/
│   │   ├── rule/ alarm/ notify/ ui/ dashboard/ ota/ billing/
│   │   ├── workorder/ report/ openapi/ internal/ ws/
│   │   └── <module>/        route.ts service.ts repo.ts schema.ts
│   ├── mqtt/                consumer.ts publisher.ts auth.handler.ts acl.handler.ts
│   ├── thing-model/         validator.ts registry.ts seed.ts
│   ├── rule-engine/         matcher.ts executor.ts state.ts scheduler.ts
│   ├── notify/              wechat.ts sms.ts voice.ts email.ts webhook.ts
│   ├── static-ui/           upload.ts sandbox.ts token.ts
│   └── workers/             history-writer.ts forwarder.ts aggregator.ts
├── prisma/                  schema.prisma migrations/
├── test/                    unit/ integration/ e2e/ fixtures/
├── scripts/                 seed.ts backup.sh restore.sh
├── docker-compose.yml
├── Dockerfile
└── package.json

单人维护建议:每个模块只保留 route.ts(Express Router,薄)→ service.ts(业务)→ repo.ts(SQL/Prisma)→ schema.ts(Zod),不额外造 controller/use-case 层。

B. Python FastAPI(备选)

backend/
├── app/
│   ├── main.py
│   ├── core/              config.py security.py deps.py errors.py logging.py
│   ├── db/                session.py base.py mixins.py
│   ├── models/            tenant.py user.py product.py device.py rule.py alarm.py ...
│   ├── schemas/           thing_model.py rule.py device.py common.py
│   ├── routers/           auth.py tenant.py product.py thing.py device.py data.py
│   │                      rule.py alarm.py ui.py dashboard.py ota.py billing.py
│   │                      workorder.py report.py openapi.py internal.py ws.py
│   ├── services/          product_service.py device_service.py command_service.py ...
│   ├── repositories/      base.py device_repo.py history_repo.py ...
│   ├── mqtt/              consumer.py publisher.py auth_handler.py acl_handler.py
│   ├── thing_model/       validator.py registry.py seed.py
│   ├── rule_engine/       matcher.py executor.py state.py scheduler.py
│   ├── notify/            wechat.py sms.py voice.py email.py webhook.py
│   ├── static_ui/         upload.py sandbox.py token.py
│   └── workers/           history_writer.py forwarder.py aggregator.py
├── alembic/               versions/
├── tests/                 unit/ integration/ e2e/ fixtures/
├── scripts/               seed.py backup.sh restore.sh
├── docker-compose.yml
├── Dockerfile
└── pyproject.toml

共同约束:

8.12.1.1 Node.js + Express 关键代码骨架(推荐栈)

只给"能启动、能接设备、能校验、能跑任务"的最小骨架;业务规则仍按第 8 章契约实现。

依赖清单:

express@5 typescript tsx @types/express
prisma @prisma/client zod ajv ajv-formats
mqtt ioredis bullmq ws pino pino-http
jsonwebtoken passport passport-jwt multer
swagger-jsdoc zod-to-openapi
vitest supertest wechatpay-node-v3

启动(Express 5 + TypeScript):

// src/app.ts
import express from 'express';
import { requestId } from './common/request-id';
import { auth } from './common/auth';
import { tenant } from './common/tenant';
import { errorHandler } from './common/errors';
import { deviceRouter } from './modules/device/route';

export function createApp() {
  const app = express();
  app.use(requestId);                         // requestId → 日志
  app.use(express.json({ limit: '64kb' }));   // 报文上限
  app.use('/api/v1', auth, tenant);           // 鉴权 → 租户
  app.use('/api/v1/devices', deviceRouter);   // 权限在路由内按需挂
  app.use(errorHandler);                      // 统一错误码
  return app;
}
// src/server.ts
import { createApp } from './app';
const server = createApp().listen(8000, '0.0.0.0');
process.on('SIGTERM', () => server.close(() => process.exit(0)));

Express 5 会把 async 路由的 rejected promise 自动交给错误中间件,不需要 express-async-errors;仍建议只抛 AppError,不要在路由里到处 try/catch。

Prisma 模型(核心表最小片段):

model Tenant {
  id       BigInt   @id @default(autoincrement())
  parentId BigInt   @default(0) @map("parent_id")
  name     String   @db.VarChar(64)
  level    Int      @default(1)
  status   Int      @default(1)
  @@map("tenant")
}

model Device {
  id        BigInt   @id @default(autoincrement())
  tenantId  BigInt   @map("tenant_id")
  productId BigInt   @map("product_id")
  deviceName String  @map("device_name") @db.VarChar(64)
  sn        String   @unique @db.VarChar(64)
  gatewayId BigInt?  @map("gateway_id")
  modelVersion String @default("1.0.0") @map("model_version")
  status    Int      @default(0)
  lastSeen  DateTime? @map("last_seen") @db.Timestamptz()
  @@index([tenantId])
  @@map("device")
}

MQTT 共享订阅消费(mqtt.js):

import mqtt from 'mqtt';
import { Redis } from 'ioredis';

const redis = new Redis(process.env.REDIS_URL!);
const client = mqtt.connect(process.env.MQTT_URL!, {
  clientId: `aqua-core-${process.pid}`,
  username: process.env.MQTT_USER,
  password: process.env.MQTT_PASS,
  clean: true, keepalive: 60, reconnectPeriod: 5000,
});

client.on('connect', () => {
  client.subscribe('$share/aqua-core/aqua/+/up', { qos: 1 });
});

client.on('message', async (topic, buf) => {
  const msg = JSON.parse(buf.toString('utf8'));
  const dedupKey = `mqtt:dedup:${msg.sn}:${msg.msg_id ?? `${msg.sn}:${msg.seq}`}`;
  if (!(await redis.set(dedupKey, '1', 'EX', 86400, 'NX'))) return; // 重复消息
  // 身份解析 → type 分派 → 物模型校验 → Redis/Hash + property_history → 规则引擎
  await dispatch(topic, msg);
});

Express 路由 + 权限中间件:

// src/common/permission.ts
import type { RequestHandler } from 'express';

export function requirePerm(perm: string): RequestHandler {
  return (req, _res, next) => {
    if (!req.user?.perms?.includes(perm)) return next(new AppError(40301, '无权限'));
    next();
  };
}
// src/modules/device/route.ts
import { Router } from 'express';
import { requirePerm } from '../../common/permission';
import * as deviceService from './service';

export const deviceRouter = Router();

deviceRouter.get('/', requirePerm('device:read'), async (req, res) => {
  const data = await deviceService.list(req.tenantId!, req.query);
  res.json(ok(data));
});

deviceRouter.post('/:id/service/:sid',
  requirePerm('device:control'),
  async (req, res) => {
    const result = await deviceService.invoke(req.tenantId!, req.params.id, req.params.sid, req.body);
    res.json(ok(result));
  });

BullMQ 命令超时/通知重试:

import { Queue, Worker } from 'bullmq';
const connection = { url: process.env.REDIS_URL! };
export const commandQueue = new Queue('command-timeout', { connection });

await commandQueue.add('timeout', { commandId, tenantId },
  { delay: timeoutMs, jobId: requestId, removeOnComplete: 1000 });

new Worker('command-timeout', async (job) => {
  // 查 device_command:仍为 SENT → 标记 TIMEOUT;按策略重试或告警
}, { connection });

物模型 Zod 校验(片段):

import { z } from 'zod';
export const PropertySchema = z.object({
  identifier: z.string().regex(/^[a-z][a-z0-9_]{0,9}$/),
  source_field: z.string().optional(),
  name: z.string().max(64),
  data_type: z.enum(['int','float','bool','enum','string','array','object']),
  access: z.enum(['r','rw','report_only']),
  unit: z.string().max(16).optional(),
  min: z.number().optional(), max: z.number().optional(),
  enum: z.array(z.string()).optional(),
  is_chart: z.boolean().default(false),
  is_history: z.boolean().default(false),
}).refine(v => v.data_type !== 'enum' || (v.enum?.length ?? 0) > 0, {
  message: 'enum 必须提供取值',
});

动态物模型不用 Zod 递归生成,直接用 Ajv 编译 JSON Schema(见 §8.4.2);Zod 负责固定 DTO 与配置校验。

8.12.2 环境变量与密钥(禁止硬编码)

变量 示例 说明
ENV prod 环境
DATABASE_URL Node:postgresql://...;FastAPI:postgresql+asyncpg://... 主库
REDIS_URL redis://... 缓存
JWT_SECRET 32+ bytes JWT 签名
JWT_ACCESS_TTL 7200 秒
INTERNAL_TOKEN 随机 EMQX 回调鉴权
EMQX_API_URL http://emqx:18083/api/v5 踢下线/ACL 同步
EMQX_API_KEY/SECRET — EMQX REST
MQTT_HOST/PORT/USER/PASS — 平台发布端
OSS_ENDPOINT/BUCKET/AK/SK — 固件/UI/报告
WECHAT_APPID/SECRET — 小程序登录/订阅消息
WECHAT_MCH_ID/API_V3_KEY/CERT — 支付
SMS_PROVIDER/KEY/SECRET/SIGN — 短信
VOICE_PROVIDER/... — 语音
WEBHOOK_SECRET — 出站签名
SENTRY_DSN — 错误追踪
LOG_LEVEL INFO 日志

8.12.3 docker-compose(起步完整版)

services:
  emqx:
    image: emqx/emqx:5.8
    ports: ["1883:1883","8883:8883","8083:8083","8084:8084","18083:18083"]
    volumes: ["./deploy/emqx/emqx.conf:/opt/emqx/etc/emqx.conf", "emqxdata:/opt/emqx/data"]
  postgres:
    image: postgres:16
    environment: { POSTGRES_DB: aqua, POSTGRES_USER: aqua, POSTGRES_PASSWORD: "${PG_PASS}" }
    volumes: ["pgdata:/var/lib/postgresql/data"]
    healthcheck: { test: ["CMD-SHELL","pg_isready -U aqua"], interval: 10s, retries: 5 }
  redis:
    image: redis:7-alpine
    command: ["redis-server","--appendonly","yes","--maxmemory","512mb","--maxmemory-policy","allkeys-lru"]
    volumes: ["redisdata:/data"]
  api:
    build: ./backend
    env_file: .env
    depends_on: { postgres: {condition: service_healthy}, redis: {condition: service_started}, emqx: {condition: service_started} }
    # Node.js + Express(推荐):构建后 npm run start:prod;FastAPI 备选改为下一行
    command: ["node","dist/server.js"]
    # command: ["uvicorn","app.main:app","--host","0.0.0.0","--port","8000","--workers","4"]
    healthcheck: { test: ["CMD","curl","-f","http://localhost:8000/health"], interval: 15s }
  scheduler:
    build: ./backend
    env_file: .env
    command: ["npm","run","scheduler"]        # FastAPI 备选:["python","-m","app.scheduler"]
  nginx:
    image: nginx:alpine
    ports: ["80:80","443:443"]
    volumes: ["./deploy/nginx/nginx.conf:/etc/nginx/nginx.conf","./static:/srv/static"]
volumes: { pgdata: {}, redisdata: {}, emqxdata: {} }

8.12.4 Nginx 关键配置

server {
  listen 443 ssl http2;
  server_name api.example.com;
  ssl_certificate /etc/ssl/fullchain.pem;
  ssl_certificate_key /etc/ssl/privkey.pem;
  ssl_protocols TLSv1.2 TLSv1.3;

  location /api/ { proxy_pass http://api:8000; proxy_set_header X-Real-IP $remote_addr; }
  location /ws { proxy_pass http://api:8000; proxy_http_version 1.1;
                 proxy_set_header Upgrade $http_upgrade; proxy_set_header Connection "upgrade"; }
  location /mqtt { proxy_pass http://emqx:8083; proxy_http_version 1.1;
                   proxy_set_header Upgrade $http_upgrade; proxy_set_header Connection "upgrade"; }
  location /ui/ {
    alias /srv/static/ui/;
    add_header X-Content-Type-Options nosniff;
    add_header Content-Security-Policy "frame-ancestors 'self' https://app.example.com";
  }
}

8.12.5 数据库迁移与初始化

CI/CD 流水线(最小):

push/PR → lint(ruff/eslint) → typecheck(mypy/tsc) → unit → integration(compose)
       → build image → 镜像扫描(Trivy) → 推送仓库 → 部署 staging
       → E2E(设备模拟器+Playwright) → 人工审批 → 生产滚动发布 → 冒烟/健康检查

8.12.6 日志、指标、追踪、健康检查

项 要求
日志 JSON 结构化;字段 ts/level/logger/request_id/tenant_id/user_id/device_id/msg;不打印 secret/token/手机号全量
访问日志 Nginx + API 双记录;保留 30 天
错误追踪 Sentry/OpenTelemetry;按 request_id 关联
指标 Prometheus:QPS、P95、错误率、MQTT 消费速率/积压、DB 连接池、Redis 命中、规则执行成功率、通知失败率、设备在线数
追踪 OpenTelemetry(API→DB→MQTT→规则);采样 10%
健康 /health(进程)、/health/ready(DB/Redis/MQTT)、/metrics(内网)
告警 5xx>1%、消费积压>1万、DB 连接>80%、磁盘>80%、证书<30天、EMQX 连接异常

8.12.7 备份、容灾与降级

项 策略 RPO/RTO
PostgreSQL 每日全量 + WAL 归档(RDS 自动);每周异地 RPO≤5min / RTO≤1h
Redis AOF everysec + 每日 RDB;丢失可从 DB 回填最新状态 RPO≤1s / RTO≤30min
OSS 版本化 + 跨区域复制(固件/UI/报告) RPO≈0
EMQX 配置 Git + 每日导出 RTO≤30min
时序库 每日 dump + binlog RPO≤1h
恢复演练 每季度一次;记录时长 —

降级:

故障 行为
MQTT 挂 设备本地控制继续;API 显示"数据延迟";恢复后补传
API 挂 小程序显示缓存;控制不可用;设备本地不受影响
DB 挂 写 Redis Stream 缓冲(有限);只读降级;恢复回放
Redis 挂 直读 DB;在线状态由 lifecycle 兜底;性能下降但不丢数据
通知商挂 站内信必达;其他渠道重试/降级;告警不丢
云全挂 设备本地闭环保命;这是对客户的核心承诺

8.12.8 安全基线(上线前必须)

8.12.9 容量与性能目标

指标 目标(起步) 说明
设备连接 10,000(EMQX 单机 5 万富余) 每网关 1 连接
消息吞吐 5,000 msg/s 1000 台×1 条/分钟≈17 msg/s,余量大
API P95 <300ms(读)/ <800ms(写) 不含导出/OTA
历史查询 1 天 raw <1s;30 天 1h <2s 5000 点上限
告警延迟 触发→站内信 <5s;微信 <60s 目标 ≤60s
规则执行 P95 <50ms/条 内存匹配
可用性 99.9%(API)/ 设备本地 100% 云端故障不影响保命

8.13 测试、验收与上线【P0】

8.13.1 测试金字塔

层 范围 工具/要求
单元 物模型校验、规则匹配、签名、权限、状态机 Node:Vitest/Jest;FastAPI:pytest;覆盖率≥70%(核心模块≥85%)
集成 API+DB+Redis、EMQX 回调、命令闭环、通知 mock testcontainers/docker compose;每次 CI
E2E 模拟设备上线→上报→告警→控制→OTA Node/Python MQTT 模拟器 + Playwright(Web)
契约 开放 API/Webhook/自定义 UI 桥接 JSON Schema + 消费者契约测试
性能 MQTT 1 万连接、5k msg/s、历史查询 Locust/k6 + emqtt_bench
安全 越权、注入、上传、签名重放、限流 自动化 + 渗透
混沌 杀 EMQX/Redis/DB、网络分区 验证降级与恢复

8.13.2 设备模拟器(必须交付)

8.13.3 必测场景(验收用例)

# 场景 通过标准
1 设备一机一密接入 错误 secret 拒绝;正确 secret 上线;LWT 离线
2 物模型校验 类型/范围/枚举/只读写入均被正确处理;非法不落库
3 命令闭环 cmd_result 关联;超时重试;code=6 透传
4 断网补传 cache=true 按 collect_ts 归档;不触发实时告警;去重
5 网关子设备 拓扑同步;网关离线子设备离线;换网关迁移
6 规则联动 属性/事件/定时;跨设备;持续时长;冷却;恢复
7 告警闭环 触发→通知→确认→恢复→关闭;升级;去重
8 多租户隔离 A 访问 B 全部返回 404/403;缓存/静态文件不越权
9 权限 四个内置角色 + 自定义角色;控制需 operator
10 历史查询 raw/1h/1d;补测/缺测;权限过滤
11 自定义 UI 上传/沙箱/桥接/回退/回滚;XSS 不生效
12 OTA 灰度/进度/失败回滚/兼容性/离线 BLE
13 计费 下单/回调/幂等/到期降级/本地控制不停
14 开放 API 签名/时间窗/nonce/限流/scope/审计
15 Webhook 签名/重试/幂等/停用
16 通知 微信/短信/语音/站内信/静音/失败降级
17 降级 杀 MQTT/DB/Redis 后设备本地控制不受影响
18 压测 1 万连接、5k msg/s、P95 达标、无消息丢失

8.13.4 上线检查(Go/No-Go)

8.14 补差数据表索引与迁移顺序

表 用途 章节
role / permission / role_permission / user_role RBAC 8.2.2
tenant_invite 成员邀请 8.2.2
gateway / gateway_subdevice 网关拓扑 8.3.7
pond_threshold / device_config 塘口/设备级参数覆盖 8.4.6
device_command 命令闭环 8.3.6
property_hourly / property_daily 聚合 8.6.2
retention_policy / partition_task 保留/分区 8.6.2
rule_execution_log 规则执行日志 8.7.4
notify_channel / notify_template / notification / notify_log / user_notify_pref 通知与用户偏好 8.7.7
scene 场景联动 8.7.8
data_forward_log 转发日志 8.7.9
webhook_delivery Webhook 投递 8.8.3
ota_device / firmware_signing_key OTA 逐设备/签名公钥 8.9.3
plan / subscription / payment_log 计费 8.9.4
work_order_log / work_order_comment 工单 8.9.5
report_template / pond_operation_log 报告模板/养殖经营记录 8.9.6
video_channel / ai_event 扩展(可选) 8.9.7
export_task / data_deletion_request 异步导出/数据删除 8.6.3
tenant_quota 租户配额 8.11.4
write_dlq / job_log 可靠性/任务 8.6.1 / 8.7.10

迁移顺序:扩展/枚举 → RBAC → 网关 → 命令 → 存储/聚合 → 规则/告警/通知 → 开放平台/配额 → OTA(含签名密钥)/计费/工单/报告/删除 → 索引/分区 → 种子数据。

8.15 实施顺序与完成定义(DoD)

在 §5.2 七个开发块基础上,v5.0 要求每块的 DoD:

块 交付 DoD
1 数据模型 全部 DDL + 迁移 + 种子 + Repository 迁移成功(Prisma migrate deploy / Alembic upgrade head);种子幂等;租户隔离测试通过
2 MQTT 接入 认证/生命周期/ACL/消费/发布/模拟器 模拟网关上线→上报→落库;错误 secret 拒绝;LWT 离线;补传去重
3 物模型 校验器/版本/模板/解析管道 四类非法用例被拒;属性/服务/事件全通;版本切换不串数据
4 规则告警 匹配/动作/冷却/恢复/通知/告警中心 六条水产默认规则通过;告警闭环;通知失败可降级
5 设备管理 生命周期/绑定/共享/分组/影子/日志/组件 扫码绑定→共享→控制→解绑全流程;离线暂存上线下发
6 UI/静态 默认组件/自定义 HTML/沙箱/回退 任意合法物模型开箱显示;自定义页加载/回退/回滚;XSS 不生效
7 工程部署 compose/Nginx/备份/监控/CI 一键起;健康检查;备份恢复;压测达标;上线检查全绿

排期映射:§5.3 的 37 天保持不变;新增工作量主要是把"隐性契约"显性化,不改变功能范围。建议在阶段 1 先完成 8.1/8.2/8.3 的口径与接入契约,阶段 3 前完成 8.6/8.7,阶段 5 前完成 8.10,阶段 7 完成 8.12/8.13。


9、验收与实施检查表

用法:逐项勾选;每项都要有可演示的证据(接口返回、日志、截图、测试报告)。"必须"项未过 = 不算完整物联网平台。

9.1 设备接入(M2)

9.2 产品与物模型(M3)

9.3 设备管理(M4)

9.4 数据存储(M5)

9.5 规则与告警(M6)

9.6 开放平台(M7)

9.7 多租户与安全(M1)

9.8 可选模块(O1~O7)

9.9 前端与体验

9.10 工程与运维

9.11 智慧水产业务闭环

最终判定:9.1~9.8 的"必须"项全部通过,且 9.11 的水产闭环可演示,才算"跟着本指南能实现一个完整物联网平台"。


附录

A. 设备接入报文示例

(api-mqtt 协议,对应 M2/M3)

主题:上行 aqua/<sn.product_key>/up,下行 aqua/<sn.product_key>/cmd(动作固定 up/cmd 两类,业务类型靠 type 区分;完整规范见 【参考】api-mqtt【可扩展更多】.md)。

// 上线状态(开机必报;异常掉线由遗嘱 LWT 兜底)
{"type":"status","sn":"sz0001.rly4p3","device_sn":"sz0001","product_id":"rly4p3","online":true,"fw":"1.2.0","signal":-72,"timestamp":"2026-09-20T17:36:50+08:00"}

// 属性数据快照(周期上报,字段随品类字典;对应物模型只读属性)
{"type":"data","sn":"sz0001.rly4p3","oxygen":5.2,"water_temp":25.6,"voltage":12.2,"switch1":true,"switch2":false,"switch3":false,"switch4":false,"signal":20,"timestamp":"2026-09-20T17:36:50+08:00"}

// 命令下发(平台→设备;seq 每设备单调递增,source 标来源,timeout_ms 超时判定)
{"type":"cmd","cmd":"set","sn":"sz0001.rly4p3","request_id":"REQ-1790046772049-E7LGFH","params":{"switch1":true},"seq":1001,"source":"cloud","timeout_ms":5000,"timestamp":"2026-09-22T11:15:00+08:00"}

// 命令回执(设备→平台,发往 up 主题;request_id 原样回传,code 码表 0=成功)
{"type":"cmd_result","sn":"sz0001.rly4p3","request_id":"REQ-1790046772049-E7LGFH","cmd":"set","seq":1001,"success":true,"message":"四路继电器已打开","switch1":true,"timestamp":"2026-09-22T11:15:00+08:00"}

// 配置下发(阈值/上报周期/DO-频率映射;设备持久化并回 cmd_result,cfg_ver 比对补下发)
{"type":"config","sn":"sz0001.vfd","config":{"report_interval_s":30,"do_map":[{"do":3,"freq":50},{"do":4,"freq":35},{"do":5,"freq":20},{"do":6,"freq":0}],"min_run_s":600},"seq":2001}

// 事件(去抖后上报;恢复事件同名 + clear:true;事件码表见 api-mqtt §11.4,驱动 M6 告警)
{"type":"event","sn":"sz0012.wq8in1","event":"do_low","level":"warning","value":3.2,"threshold":4.0,"duration_s":60,"timestamp":"2026-09-20T17:36:50+08:00"}

B. 断网补传协议约定

设备断网期间数据缓存 Flash 环形缓冲(≥10000 条,满溢丢弃最旧),恢复联网后逐条补发,报文带 "cache":true,并用 collect_ts(采集时间)与 timestamp(发送时间)分离标识补传语义:

{"type":"data","sn":"sz0012.wq8in1","cache":true,"collect_ts":"2026-09-20T17:36:50+08:00","oxygen":3.9,"ph":7.8,"water_temp":26.1,"timestamp":"2026-09-21T08:00:01+08:00"}

平台按 (sn, collect_ts) 归档去重(重复 msg_id 丢弃),逐条按 collect_ts 写入 property_history;补传数据不触发实时告警(避免过期数据误报),但批内越阈值数据生成"补传异常摘要"事件供人工查看;未补传条数经 gw4g.cache_cnt 监控。

C. 工程目录结构(二选一;推荐 Node.js + Express)

C.1 Node.js + Express(推荐):Express 5 + TypeScript + Prisma

backend/
├── src/
│   ├── server.ts / app.ts
│   ├── config/        env.ts logger.ts db.ts redis.ts
│   ├── common/        errors.ts response.ts auth.ts tenant.ts permission.ts request-id.ts
│   ├── modules/       auth/ tenant/ user/ product/ thing/ device/ data/
│   │                  rule/ alarm/ notify/ ui/ dashboard/ ota/ billing/
│   │                  workorder/ report/ openapi/ internal/ ws/
│   │                  <module>/{route,service,repo,schema}.ts
│   ├── mqtt/          consumer.ts publisher.ts auth.handler.ts acl.handler.ts
│   ├── thing-model/   validator.ts registry.ts seed.ts
│   ├── rule-engine/   matcher.ts executor.ts state.ts scheduler.ts
│   ├── notify/        wechat.ts sms.ts voice.ts email.ts webhook.ts
│   ├── static-ui/     upload.ts sandbox.ts token.ts
│   └── workers/       history-writer.ts forwarder.ts aggregator.ts
├── prisma/            schema.prisma migrations/
├── test/              unit/ integration/ e2e/ fixtures/
├── scripts/           seed.ts backup.sh restore.sh
├── docker-compose.yml / Dockerfile / package.json

C.2 Python FastAPI(备选)

backend/
├── app/
│   ├── main.py
│   ├── core/          config.py security.py deps.py errors.py logging.py
│   ├── db/            session.py base.py mixins.py
│   ├── models/        SQLAlchemy 模型
│   ├── schemas/       Pydantic schema(物模型/规则 JSON 校验)
│   ├── routers/       auth/product/thing/device/data/rule/alarm/ui/dashboard/ota/
│   │                  billing/workorder/report/openapi/internal/ws
│   ├── services/      repositories/ mqtt/ thing_model/ rule_engine/ notify/ static_ui/ workers/
├── static/ui/         # 自定义 UI 存储(Nginx 直服)
├── alembic/           # 迁移
├── docker-compose.yml / Dockerfile / pyproject.toml
└── tests/             # unit / integration / e2e / fixtures

D. 术语表

术语 含义
物模型 (TSL) Thing Specification Language:属性/服务/事件三要素描述设备能力
一机一密 每台设备独立密钥,出厂烧录,泄露不影响其他设备
一型一密 同型号预烧产品密钥,首连动态注册换一机一密凭据
共享订阅 EMQX $share/{group}/{topic}:同组订阅者负载均衡消费
设备影子 平台侧保存的设备最新期望/报告状态(本文用 Redis 实现)
边缘闭环 控制逻辑在设备本地执行,不依赖云端
多租户 平台→厂家(vendor)→一级租户→用户的四级数据隔离体系
产品键 product_key,必须等于 api-mqtt 的 product_id
命令闭环 cmd → 设备执行 → cmd_result(request_id/seq 关联)
补传 断网缓存数据带 cache=true、按 collect_ts 归档
SLA 服务级别协议:响应/解决时限
entitlement 套餐权益校验(设备数/历史天数/渠道/API 等)