智慧水产系统设计指南

版本：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、水产行业需求

1.1 行业痛点

（需求总纲）

来自看塘人官网（kantangren.com）与总规划书，按优先级排序：

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

1.2 用户角色与场景

多租户模型（对应涂鸦"厂家-家庭-用户"，米家"家庭-用户"，并对齐通用提纲"厂家 + 一级租户 + 用户"）：

平台运营方（我们，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 非功能需求

2、水产行业标准

2.1 溶氧安全阈值

（核心业务标准）

行业通用参考区间（以品种、放养密度、水温为准，平台内可按塘口覆盖）：

落地：做成 threshold_template；建塘选品种自动填充，用户可覆盖。

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

（四档映射）

配套保护规则（固件本地执行，平台只做展示与参数下发）：

• 传感器离线 > 30 分钟 → 自动切时间表模式（跑 30min / 停 15min）
• 多路错峰启动：正常间隔 4s / 紧急 2s / 断电恢复 5s
• 最短运行 10 分钟，防频繁启停
• RCD 跳闸：30 天内首次 30s 后自动重试，2 次以上停止自动重试并报警

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

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

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

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

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

• 每个水产品类有标准物模型模板；
• 每个养殖品种有标准阈值模板；
• 每类风险有标准告警规则模板。

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

3、用数据赋能水产行业

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

3.1 数据集设计

（平台沉淀的九类核心数据）

3.2 功能点 → 数据映射

（保证"每个功能都有数据支撑"）

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 对动态物模型最友好；吸收 FastBee 标签维度（is_chart/is_monitor/is_history/is_readonly/model_order），量大再迁表驱动（§4.5）。

4.1 平台功能定义总表

（必须在前，可选在后）

必须功能（P0，缺一不可，对应 §4.3 各节）：

可选功能（P1/P2，按需启用，对应 §4.4 各节）：

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

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

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

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 技术选型

4.2.2.1 后端语言：二选一，推荐 Node.js + Express

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

推荐理由（按权重）：

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 时）：

• API 进程只做轻量编排；MQTT 消费、历史批量写、聚合、报表、通知重试全部独立 worker（BullMQ/独立进程）。
• 单进程 UV_THREADPOOL_SIZE、连接池、Redis pipeline 按压测调；PM2/cluster 或 K8s 多实例 + SLB。
• 规则引擎内存匹配保持 O(候选规则数)，禁止在请求链路里做全表扫描/大 JSON 深拷贝。
• Express 5 已支持 async 错误冒泡；统一错误中间件 + requestId，不要每个路由 try/catch。
• 压测基线仍是 §8.12.9：1 万连接、5k msg/s、API P95 <300ms；语言选型不改变验收标准。

两套实现的模块映射（第 8 章契约不变）：

同一项目只选一套；第 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。

【数据结构】

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);  

【关键设计】

• 三套凭据严格分离，不混用：用户走 JWT；开放 API 走 AccessKey（见 M7）；设备走一机一密（见 M2）；
• 厂家（vendor）维度：product.vendor_id 指向 tenant 树中 level=1 的厂家节点，决定固件与"厂家自定义 HTML"（O1）的归属；厂家可视为"更细的租户维度"（与 FastBee 通用 tenant_id 隔离机制一致，可叠加）；
• 权限检查：Express 用 requireRole('tenant_admin') / requirePerm('device:control') 中间件（推荐），FastAPI 用 Depends(require_role("tenant_admin"))；角色多了再引 Casbin；
• 审计三问必答：谁（user_id）在什么时候（created_at）对什么（target_type/target_id）做了什么（action/detail）。

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）。

平台概念 ↔ 协议身份对照（核心映射，平台内部模型不变）：

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

• Redis 在线状态键：device:{id}:online = 1（TTL 180s = 3×Keepalive）；device:{id}:last_seen = 时间戳；
• 设备凭据字段：见 M4 device 表的 device_secret_hash（只存哈希）。

【关键设计】

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"} ]  
}  

【关键设计】

• 物模型（平台 TSL）↔ api-mqtt 品类字典对照（新增品类时按此入库，两套概念一一对应）：

• 字段命名：identifier 小写 snake_case，新增 ≤10 字符；同概念跨品类同名（oxygen/ec）；单位唯一、量程固定；事件/服务可更长；存量字段不改（air_pressure 豁免）；
• 运行时解释流程（无代码生成）：设备上报 JSON → 查该产品已发布物模型 → 逐字段校验类型/范围 → 通过写 Redis（device:{id}:state，Hash）+ property_history（仅 is_history 属性）→ 交规则引擎；不通过则丢弃 + 记设备日志，不影响设备连接；
• 服务下发：入参按 input_params 校验 → 打包发下行 Topic（cmd/config）→ 等 cmd_result（报文协议见 M2，超时 timeout_ms 默认 5s，超时幂等重试或标记"设备未确认"）；
• 事件处理：写设备日志 → 生成 alarm_record（若事件带 level）→ 交规则引擎匹配动作；
• 版本规则：编辑产生新草稿 → 发布生成新 version → 已绑定设备按其固件版本解析，避免上报格式与定义不一致；品类不兼容升级递增版本号（如 vfd → vfd2）。

【水产旗舰示例：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 改成协议字段）：

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);  

【关键设计】

• 默认组件映射表（"属性→组件"，任何合法物模型开箱即显示；与 O2 看板组件共享同一套映射）：

• 设备生命周期：创建（发凭据）→ 烧录 → 首连激活 → 在线 → （离线/禁用）→ 解绑/删除；
• 产品创建→设备上线全流程（涂鸦式）：①创建产品 → ②定义物模型 → ③发布生成 product_key → ④创建设备获 device_secret → ⑤烧录品类凭据按 api-mqtt 协议连接（M2）→ ⑥虚拟设备/真实设备验证 → ⑦用户扫码绑定 → 绑定即订阅告警。

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}    期望状态(设备离线时的下发暂存)  

【关键设计】

• 配置与数据分离（全平台铁律）：物模型/规则/权限在关系库，上报值在缓存/时序，靠 product_key/device_id 关联；
• 容量估算：1000 台 × 每分钟 1 条 = 1 年 5.25 亿条；TDengine 压缩后约 500MB，硬盘成本可忽略；PG 分区起步够用，>500 台迁 TDengine（超级表 + sn TAG，每设备一子表）；
• 聚合策略是配置不是硬编码（TDengine 流计算或 PG 定时任务均可）。

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  
}  

【关键设计】

• 水产告警矩阵（默认规则模板，与 §2.2 保护规则配套）：

• 执行保障：数据流入按 product_key + identifier 内存索引匹配 → 命中查冷却 → 异步执行动作 → 写执行日志；规则执行失败不阻塞数据落库；
• 事件恢复闭环：设备端恢复事件（同名 event + clear:true，见 api-mqtt §3.8）自动把对应 alarm_record 置"已恢复"；恢复事件同样走规则引擎（可配"恢复通知"）；
• 边缘 + 云端双闭环（铁律）：

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

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}/...             # 历史版本可回滚  

【关键设计】

• 上传白名单：仅 .html/.css/.js/图片，大小上限 5MB，禁止 .env/.git 等敏感文件；上传目录独立于应用代码，生产由 Nginx 直接服务，FastAPI 只管上传和元数据；
• postMessage 桥接协议：

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

• 桥接能力清单：getLatestProps / subscribeProps / invokeService，凭短期设备 token（15min）鉴权——绝不暴露平台内部 API 与用户 JWT；

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: {} }  

演进路线

备份与降级

• PG：RDS 自动备份每日 + 变更前手动快照；TDengine：文件 rsync 每日；
• 云端全挂不影响设备本地控制（M6 双闭环铁律）——这是向客户承诺可靠性的底气。

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 天）

6、后期可扩展行业应用

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

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

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

一句话总结：设备按 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 身份与命名口径（最容易写错的地方）

必须成立的三条等式（建议写成数据库约束 + 单元测试）：

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

8.1.2 时间、单位、ID、编码

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 边缘/云端职责矩阵

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

8.2.1 租户树与生命周期

platform(level=0, id=1)
 └── vendor(level=1)                 厂家：产品/固件/自定义UI归属
      └── tenant(level=2)            代理商/基地/集团子公司
           └── user_group(level=3)   家庭/班组/用户组（C端自助注册默认建这一层）

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

内置角色权限：

实现方式：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
}

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;

刷新与登出：

• refresh token 存 Redis auth:refresh:{jti} = user_id，TTL 30d；刷新时轮换（旧 jti 删除，新 jti 写入），检测到旧 token 复用则整用户下线。
• 登出/改密/停用：user:{id}:ver +1，所有 access token 因 ver 不匹配失效；refresh 全删。
• 微信 unionid 用于同主体多小程序合并；openid 唯一键；手机号绑定后可用于短信告警。
• 首次微信登录必须让用户勾选《隐私政策》并记录 consent_at、consent_version（合规要求）。

短信验证码：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已撤销

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。

• 记录范围：所有写操作 + 登录/登出 + 设备控制 + 权限变更 + OTA + 计费 + 导出。
• 敏感字段（密码、secret、AccessKey）写入前脱敏为 ***。
• 查询：租户管理员看本租户；平台管理员看全部；保留 180 天，之后归档到 OSS（Parquet/JSONL）。
• 审计写入异步，但控制类操作必须与业务同事务或先写审计再执行（至少保证失败也有记录）。

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)

• 所有业务表必须有 tenant_id（device_state/property_history 通过 device 关联校验，查询必须先取设备再校验租户）。
• 所有列表接口默认 tenant_id 过滤；跨租户统计走独立 platform 路由 + 显式权限。
• 缓存键带租户：t:{tenant_id}:device:{id}:state，防止键碰撞越权。
• 静态文件目录：static/ui/{tenant_id}/{product_key}/...，Nginx internal + 签名 URL，禁止目录遍历。
• 测试：每个资源至少一条"租户 A 访问租户 B 返回 404/403"的集成测试。

8.2.7 账号安全基线

8.3 设备接入实现规范（M2 补差）【P0】

8.3.1 连接与主题最终口径

上行（设备→平台）  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

• 平台只订阅 up，不订阅 cmd（避免自回环）；下行由 API/规则引擎发布。
• 每个网关一个 MQTT 连接；子设备复用连接，用主题中间段区分。
• 设备侧 LWT：aqua/<gw4g_sn>.gw4g/up，payload {"type":"status","sn":"<gw4g_sn>.gw4g","online":false}；子设备离线由网关 status/gw4g.devices 与超时判定，不单独设 LWT。

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}

校验逻辑：

• 密码只存哈希，平台无法反查；重置 = 生成新 secret 覆盖哈希，旧连接被踢下线（EMQX REST API DELETE /clients/{clientid}）。
• 内部回调必须校验 X-Internal-Token + 来源 IP 白名单；/internal/* 不暴露公网。
• 认证失败按 IP+client_id 限流（Redis 10 次/分钟），防爆破。

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/#

实现二选一：

• EMQX 5 支持 HTTP 授权源时：/internal/mqtt/acl 接收 clientid/username/topic/action，查网关子设备表 / 用户绑定表返回 allow/deny。
• 不支持时：平台在网关注册/子设备变更时，通过 EMQX REST API 把 ACL 规则写入内置数据库（按 client_id 精确规则），变更后同步。
• 开发期可先 allow all，但上线前必须有 ACL 集成测试（伪造网关发布他人主题必须被拒绝）。
• 匹配顺序：精确 allow 必须先于通配 deny；若 EMQX 授权源不支持顺序语义，则不要把 deny aqua/+/up 写成通配，改用"默认拒绝 + 逐条精确 allow"的白名单模式。
• 子设备级隔离：网关内部必须按 sn.product_id 二次校验，防止一个网关代理越权访问未登记子设备；平台侧 gateway_subdevice 是唯一权威清单。

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 推送（设备详情实时刷新）

并发与可靠性：

• 每个 worker 内 asyncio.Queue(maxsize=10000)；队列满时不丢：写入 Redis Stream mqtt:overflow，由补偿 worker 消费。
• 历史写入按 500 条/批或 200ms 刷盘；失败重试 3 次后写 write_dlq 表并告警。
• 去重键必须覆盖 msg_id（若设备提供）与 (sn, seq)；补传再按 (sn, collect_ts) 二次去重。
• 消费位点由 EMQX 共享订阅保证；平台不保存 offset；重复消息靠去重表兜底。
• 单设备消息乱序：data 按 timestamp/collect_ts 排序写入，最新状态只在 timestamp >= 当前 时覆盖；cmd_result 按 request_id 关联，不参与排序。

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，写告警"设备未确认"。

离线控制（影子）：

• 设备离线时，set 类命令写 Redis device:{id}:desired（Hash: identifier→value, __ver 版本, __ts），TTL 24h；返回"已暂存，设备上线后下发"。
• 设备上线 status(online=true) 后，平台比较 desired 与 reported，按属性逐个下发 cmd；收到 cmd_result 后删除对应 desired。
• 安全类命令不暂存：restart、clean_now、relay_ctrl 中涉及增氧机启停的，离线时直接拒绝（防止上线瞬间误动作）；只允许"参数/阈值/模式"暂存。
• desired 与 reported 冲突：以"用户最后一次手动操作"为准，但设备本地安全规则可拒绝（code=6），平台把 desired 标记失败并通知。

命令优先级（设备端仲裁，平台记录 source）：

• 同优先级按到达时间；低优先级不得打断正在执行的高优先级动作（最短运行时间内）。
• 设备拒绝时回 code=6 + message，平台必须展示"被本地安全规则拒绝"，不得显示"成功"。

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) → 订阅告警

• device_secret 只在创建/重置响应中返回一次，之后只存哈希；支持批量导出"sn+secret"加密压缩包（一次性下载，下载后删除）。
• 二维码内容：aqua://bind?sn={sn}&code={activation_code}；activation_code 存哈希，绑定成功后作废。
• 一型一密（P1）：产品存 product_secret_hash；首连用 product_key + product_secret + sn 调 /internal/device/register 换一机一密，校验并防重；量产可关。
• BLE 配网（P1）：设备未联网时通过 BLE 写入 WiFi/APN、MQTT 地址、sn、secret；协议与页面见 04_软件方案/蓝牙调试小程序规划.md；平台侧只记录配网结果（device.provisioned_at、provision_source=ble），不参与 BLE 链路。

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

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

• product_key 在创建产品时确定（发布不会重新生成）；创建后不可修改；必须改品类时新建产品 + 设备迁移（换品类 = 换 sn 主题，属于不兼容变更）。
• vendor_id 必填（level=1 租户），决定固件/UI 归属；平台级模板产品 tenant_id=0，租户复制后生成自己的产品。

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":    {}
      }
    }
  }
}

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

平台校验流程（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 版本管理与设备绑定

• thing_model 每次发布插入新行 (product_id, version)，status=1；同一产品最多一个 status=1 的"当前版本"。
• device.model_version 在设备创建时写入当前版本；设备上报时按该版本解析。升级物模型后：
• 兼容变更：平台可批量把设备 model_version 升到新版本（需设备固件支持新字段，否则旧字段继续按旧版本解析）。
• 不兼容变更：新建产品（新 product_key），设备走"换品类迁移"（重新建档/换 sn 主题），不原地升级。
• 解析缓存：Redis product:{product_key}:model:{version} TTL 10min；发布/下线时主动删除。
• 版本查询 API：GET /products/{id}/models、GET /products/{id}/models/{version}、POST /products/{id}/models/{version}/publish。

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

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

品类种子清单（与 api-mqtt §7 一一对应，平台启动时幂等 upsert 到 model_template）：

字段映射规则：

• 新品类：identifier == source_field == api-mqtt 字段名，不再造第二套命名。
• 历史产品（如平台早期用 do_value）：保留 identifier=do_value，设 source_field=oxygen；规则/UI 用 do_value，解析器按 oxygen 取值。迁移完成后统一到 oxygen。
• 公共字段 fw/signal/voltage/temp/uptime_s/fault_code 放 model_template.common，各品类模板可引用，避免重复定义。
• 事件码表：do_low→2，do_danger/rcd_trip/power_loss→3，sensor_offline/device_offline→2，ota_progress→1；恢复事件同名 + clear=true。

8.4.5 在线模拟器（无真机验证物模型）

• 入口：产品详情 → "在线模拟"；创建虚拟设备（device.simulated=true，sn 以 SIM- 前缀，禁止绑定真实用户）。
• 实现：后端内置一个 MQTT 客户端或直接调用解析管道（推荐后者，避免污染真实 Topic）；两种模式：
• 上报模拟：按物模型生成随机/手填 JSON，走 thing_model 校验 → 写 Redis/历史 → 规则引擎；UI 实时显示。
• 下发模拟：平台下发 cmd，模拟器按物模型自动回 cmd_result（可配置延迟/失败/拒绝码 6）。
• 模拟数据隔离：property_history 加 simulated BOOLEAN，查询默认排除；告警默认不通知真实用户；30 天自动清理。
• 验收：模拟器能覆盖属性/服务/事件三类 + 非法字段（类型错/超范围/只读写入）四类用例。

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 / 厂内默认）

• 创建塘口选品种 → 自动复制 threshold_template.items 到 pond_threshold；用户修改只影响本塘。
• 修改生效：服务端计算"每设备有效配置"→ 写 device_config 快照 → 生成 config 下行（带新 cfg_ver）→ 设备回 cmd_result 持久化 → 平台比对 cfg_ver，不一致补下发。
• 边缘优先：配置下发失败不影响设备当前本地参数；平台显示"待同步"，不得显示"已生效"。
• 配置项白名单：do_map/min_run_s/hysteresis/report_interval_s/sensor_offline_min/fast_report/log_level/tank_id；未知键拒绝。
• 审计：谁在什么时候改了哪个塘/设备的哪个参数，旧值/新值都记录。
• API：GET/PUT /ponds/{id}/threshold、GET/PUT /devices/{id}/config、POST /devices/{id}/config/sync。

8.5 设备管理实现规范（M4 补差）【P0】

8.5.1 设备状态机

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

• 状态只由三类事件驱动：EMQX lifecycle、设备 status 报文、管理员操作；禁止 API 直接改 status=1。
• 离线判定：now - last_seen > 3×keepalive（默认 180s）；LWT/断连事件立即置离线，但离线告警延迟 5min（防抖动）。
• 在线状态双写：device.status/last_seen + Redis device:{id}:online（TTL 180s）。

8.5.2 批量导入（CSV/Excel）

• 流程：上传 → 预校验（逐行返回错误行号/字段/原因）→ 用户确认 → 事务分批（500/批）创建 → 生成 secret 加密包（一次性下载）。
• 幂等：同一 sn 已存在则跳过并计入 skipped，不覆盖密钥。
• 失败处理：部分成功返回 {created, skipped, failed:[{row, reason}]}；不整体回滚（避免 1000 行因 1 行失败全废）。
• 导出：支持按筛选导出设备台账（不含 secret）。

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

• device_group 是通用分组；水产映射 farm → pond → device。两者可并存：设备可同时属于业务分组和塘口，但 device.pond_id 只能一个。
• 分组树最大 5 层；组内设备数上限 5000；组级批量控制 = 对组内在线设备逐个下发 cmd，聚合结果返回 {total, success, failed, offline}。
• 批量控制必须二次确认 + 操作审计；增氧机启停类批量操作要求 operator 及以上，且单次最多 200 台。
• 按塘口聚合：GET /ponds/{id}/overview 返回最新 DO/水温/设备在线数/未处理告警数。

8.5.4 设备日志与在线调试

• device_log.log_type：1 属性、2 服务、3 事件、4 生命周期、5 非法报文、6 补传摘要、7 OTA、8 配置。
• 日志保留：热库 30 天；之后按月分区归档到 OSS（Parquet/JSONL）；查询默认 7 天，最长 90 天。
• 日志查询：按设备/类型/时间/identifier 分页；支持导出 CSV（脱敏）。
• 在线调试工具：
• 实时报文流（WebSocket 推送该设备最近 100 条上下行，脱敏 secret）。
• 手动构造 cmd/config 下发（按物模型表单生成，禁止手写任意 JSON 绕过校验）。
• 虚拟设备模拟（§8.4.5）。
• 日志级别下发：config {log_level: off|error|warn|info|debug}。

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"
}

• 组件顺序：model_order 降序；分组按 group 字段（物模型可加 ui_group）。
• 无物模型/字段缺失：显示"暂无数据"，不得白屏。
• 自定义 UI（O1）失败时回退本契约。

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}}
}

• state 只返回 is_share_perm=true 的属性（共享只读用户）。
• 控制类接口返回 device_command.request_id，前端轮询/WebSocket 等 cmd_result；不得同步阻塞等待设备。

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，兜底持久）
  → 交规则引擎

• 最新状态：Redis 是读主，device_state 是持久兜底；Redis 丢失时从 device_state 回填。
• 历史：只写 is_history=true 的属性；report_only 不写最新状态但可写历史（按需）。
• 失败：Redis 失败 → 直接写 DB 并告警；DB 失败 → 写 write_dlq + 重试；规则引擎失败不影响落库。
• 一致性：最新状态允许最终一致（秒级）；历史以 (device_id, identifier, ts) 唯一，重复写入用 ON CONFLICT DO NOTHING。
• 顺序：同一设备同一属性按 ts 覆盖最新值；乱序旧数据只写历史，不回写最新状态。

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

响应：

{
  "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]]}]
}

• 最新值：GET /data/{device_id}/latest?identifiers=...，先读 Redis，miss 读 device_state。
• 聚合值：GET /data/{device_id}/stats?...&interval=1d，返回 avg/min/max/sum/count。
• 导出：POST /data/export 异步生成 CSV/Excel，任务表 export_task，完成后站内信 + 签名 URL（24h）。
• 删除/注销：POST /data/deletion（scope=account/device/pond/tenant/history）→ 审核 → 异步执行；可删除的删除，订单/审计/发票匿名化保留；GET /data/deletion/{id} 查进度，完成写审计。

8.6.4 Redis 键注册表（必须按此实现，禁止随手起键名）

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

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');

• 迁移策略：双写 1 周 → 校验一致 → 读切 TDengine → 停写 PG 分区；保留 PG 历史只读 90 天。
• 查询适配层 HistoryRepository 对上层屏蔽 PG/TDengine 差异；聚合优先用 TDengine 流计算，PG 模式用定时任务。
• 时序库只存值，不存物模型/设备档案；备份用 taosdump 每日全量 + binlog。

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}
]

• 动作串行执行；delay 后动作入调度队列；单个动作失败不阻断后续（除非 stop_on_match=true）。
• service 动作跨设备下发时，平台为每台设备生成独立 device_command；批量上限 200 台。
• notify 的 targets 解析：device_binders=绑定该设备的所有用户；operators=租户运维；custom=指定用户/角色；排除已静音/无联系方式用户。
• virtual_property 写 device_state + Redis，并标记 source=virtual，不触发设备上报；可作为其他规则条件。

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);

• 保留 30 天；只记录"命中"与"执行失败"的规则（全量记录量大）；规则详情页可查最近 100 条。
• 指标：规则执行成功率、平均耗时、动作失败率，接入监控（§8.12）。

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

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)

• 同一 alarm_key 在未关闭前只保留一条，重复触发 repeat_count++、last_at=now、trigger_value 更新；通知按冷却策略决定是否再发。
• 级别：1 提示 / 2 警告 / 3 严重；严重告警未确认 15 分钟升级（通知上级/电话）。
• 恢复：设备 clear=true 或规则恢复条件；status=2 后仍可人工关闭；不可恢复事件（recoverable=false）只能人工关闭。
• 告警查询/导出、按塘口/设备/级别/状态/时间筛选；未处理数进首页红点。
• 告警与工单联动：level=3 且 30 分钟未确认 → 自动建 P0 工单（O5）。

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。

渠道实现要点：

• 通知模板变量白名单，禁止渲染用户输入 HTML（防 XSS）。
• notify_log 记录脱敏后的目标；原文不落库。
• 静音时段：quiet_hours 内非 level=3 只发站内信，延迟到时段结束再发（可配）。

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()
);

• 场景 = 规则的预设模板 + 有序动作；用户一键执行（POST /scenes/{id}/run）或由规则触发。
• 支持"串行/并行"：动作项 {"parallel":[...]}；并行失败聚合返回。
• 水产预设：凌晨保命（DO 低 → 全速 + 电话）、省电模式（DO 高 → 降频）、雷雨预案（停电 → 报警 + 柴发）、收塘（批量停机）。

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()
);

• HTTP：POST JSON，Header X-Aqua-Signature: sha256=HMAC(secret, timestamp+"."+body)；超时 5s；重试 3 次（1m/5m/30m）；最终失败进 data_forward_log(status=2) 并告警。
• Kafka：至少一次，幂等生产者；MQTT 转发：QoS1，独立 client_id；DB 直写：只允许平台预置的只读账号/白名单表。
• 批量：默认 100 条/批或 1s 窗口；payload 含 tenant_id/device_id/product_key/ts/values。
• 敏感字段加密存储；转发目的启停/测试按钮；每租户最多 20 个目的。

8.7.10 调度任务清单（BullMQ/node-cron 或 APScheduler）

• 所有任务必须幂等、可重入、分布式锁；执行结果写 job_log；失败告警到运维群。
• 调度器独立进程（Node：npm run scheduler / BullMQ Worker；FastAPI：python -m app.scheduler），与 API 分开部署，避免多实例重复执行。

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

• 校验顺序：AK 存在/启用/未过期 → IP 白名单 → 时间窗 → nonce 未用过（Redis SETNX open:nonce:{ak}:{nonce} 1 EX 600）→ 签名 → scope → 限流。
• 限流：按 AccessKey rate_limit（默认 100 次/分钟，令牌桶）；超限 429 + Retry-After。
• 错误码：401 INVALID_SIGNATURE、403 SCOPE_DENIED、429 RATE_LIMITED、400 INVALID_ARGUMENT；不泄露内部细节。
• Secret 只展示一次；支持轮换（新旧双活 24h）与吊销（立即生效，Redis 黑名单）。

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}             删除

• 写操作默认关闭，需租户管理员在后台逐项授权 + IP 白名单；控制类接口必须记录 source=openapi、操作人（AK）、审计。
• 所有响应带 X-Request-Id；分页 page/page_size/total，page_size<=200。
• OpenAPI 文档：/open/docs（Swagger UI），按 scope 过滤展示；提供 Postman 集合与签名示例（Python/Java/Go/Node/curl）。

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"}}

• payload ≤64KB；字段只增不改；敏感字段（手机号、secret、坐标）不推给第三方。
• 2xx 成功；非 2xx/超时 5s 失败；重试 1m/5m/30m/2h/6h（5 次）；仍失败 status=3 并告警。
• 至少一次投递，接收方按 X-Aqua-Delivery 幂等。
• 事件 payload 稳定 schema，新增字段向后兼容；禁止把内部错误堆栈推给第三方。
• 每租户最多 20 个 Webhook；支持"测试发送"。

8.8.4 数据转发授权

• 高频数据不走轮询 API；用 M6 data_forward（HTTP/Kafka/MQTT）。
• 转发目的必须在 M7 登记并授权 scope data:forward；转发日志可查；停用立即生效。
• 第三方只能收到其租户/授权设备的数据；payload 不含 secret/手机号等敏感字段（按需脱敏）。

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：独立域名 ui.example.com，sandbox="allow-scripts"；同源时禁止 allow-scripts + allow-same-origin；不加 top-navigation/popups/forms。
• CSP：default-src 'self'; script-src 'self' 'unsafe-inline'; img-src 'self' data: https:; connect-src 'none'（自定义页不得直接发网络请求，只能走桥接）。
• 服务端：禁止执行上传目录内脚本；Nginx 只读；X-Content-Type-Options: nosniff；目录不可列。
• postMessage 必须校验 event.origin 与 event.source；消息 schema 白名单；nonce 防重放；短期 token 15min、绑定 device_id + user_id + scope。

桥接协议（完整）：

// 平台 → 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":"..."}

• 桥接只暴露三个能力：读最新属性、订阅属性、调用物模型服务；不暴露用户 JWT、租户 API、文件系统。
• 自定义页崩溃/超时 5s 未 ready → 回退默认组件。
• 版本回滚：POST /products/{id}/ui/{version}/rollback；保留最近 10 个版本。

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;

• 组件目录：line/bar/gauge/stat/table/map/switch/alarm-list/image/text；每个组件 source={product_key, identifiers[], device_ids[], agg}。
• 布局：12 列栅格，{x,y,w,h}；前端用 grid 渲染；后端只校验 schema。
• 数据：看板按组件批量取数（一个请求聚合多个 series），默认 30s 刷新；支持暂停。
• 共享：scope=private/tenant/platform；share_token 只读、可设过期、可撤销；公开链接不返回敏感属性。
• 预设模板：单塘总览、多塘地图、运营大屏；平台模板 tenant_id=0，租户可复制。
• 导出：PNG/PDF（前端截图或后端 headless 渲染），异步任务。

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()
);

套餐能力矩阵（默认，可后台改）：

• entitlement 检查：设备创建/绑定、历史查询、API、OTA、报表入口统一调用 entitlement.check(tenant, feature)；超限返回 402 PLAN_LIMIT 并引导升级。
• 状态机：正常 → 到期前 7 天提醒 → 到期进入宽限 7 天（全功能）→ 受限（只读+本地控制）→ 过期（仅登录/续费）；本地控制永不停。
• 微信支付 v3：统一下单 → 小程序 wx.requestPayment → 回调验签/解密 → 幂等更新订单/订阅 → 开票（可选）。回调必须验签 + 校验金额 + 订单状态机。
• 定时任务：每日扫描到期/宽限/试用；30/7/3/1 天提醒；重复支付退款；对账。
• 退款/发票/手动赠送：后台操作，写审计；金额单位统一"分"。

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()
);

• 自动派单：level=3 告警 30 分钟未确认 / 设备离线 + DO 极低 / RCD 跳闸 → 建 P0 工单，按"塘口归属运维/代理商"路由。
• SLA：P0 15min 响应/4h 解决；P1 30min/24h；P2 2h/48h；P3 4h/7d；超时升级（通知上级 + 站内信）。
• 远程诊断：信号/电压/电流/总线状态、最近日志、重启、参数回读、日志级别调整、链路质量；全部走 device_command 并留痕。
• 回访：关闭后 24h 推送评分（1~5）；低分自动建复盘工单。
• 备件/耗材：spare_part 表（可选），与设备型号/易损件周期关联。

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;

• 生成管道：定时/手动 → 取 D1/D2/D3/D7 数据 → 按模板计算（均值/超标次数/达标率/能耗对比）→ 存 content JSONB → 渲染 PDF/Excel 到 OSS → 站内信/邮件推送。
• 水质报告必须引用 standard_ref（GB 11607-89 / SC/T 9101 等）与数据时间范围；缺测标注"数据不足"，不得编造。
• 节能报告：变频前后电量对比、按电价估算节省金额、回本周期；口径写清楚（估算 vs 实测）。
• 投喂建议（V3）：基于摄食行为/水温/DO 的规则建议，标注"建议非处方"，用户可采纳/忽略。
• 维护提醒：易损件周期（探头清洗 30 天、校准 90 天、继电器寿命）→ 生成 report(maintenance) + 站内信/工单。

养殖经营记录（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);

• 一张表覆盖 5 类经营记录，不拆 5 张表；移动端快捷表单录入，可选自动带入 DO/水温。
• payload 约定：feed{feed_type,amount_kg,duration_min}；med{medicine,dose,unit,withdraw_days,reason}；patrol{result,abnormal,action}；mortality{count,cause,disposal}；harvest{weight_kg,count,size,buyer,price}。
• 用药记录保留 withdraw_days（休药期），供追溯/合规；附件存 OSS，DB 只存 key。
• 权限：写 pond:manage，读 data:read；租户隔离 + 审计；保留 3 年，可导出。
• API：GET/POST /ponds/{id}/operations、GET/PUT/DELETE /operations/{id}、GET /ponds/{id}/operations/stats。

8.9.7 O7 能力扩展位

• 协议插件接口（参考 FastBee protocol）：

// 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 定义同名三方法即可

• 插件以独立 npm 包/独立进程接入，注册到 protocol_registry；不改主消费链路；插件崩溃隔离，超时丢弃并告警。
• 视频：video_channel 表（device_id、国标 ID、流地址、状态），SIP/ZLMediaKit 独立服务；平台只做通道管理与鉴权，不转发视频流。
• AI 事件：ai_event 表（device_id、type、confidence、image_key、ts），经规则引擎联动增氧机；图片存 OSS，保留 30 天。

-- 可选扩展表（按项目启用；不启用不影响基础平台）
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
);

• 扩展原则：不新增表也能扩展（新品类走 product/thing_model；新协议走 adapter；新数据复用 property_history/alarm_record）；只有独立元数据才建表。

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

8.10.1 Web 管理后台页面地图

前端实现约定：

• Vue3 + Element Plus + Pinia + Vue Router；按权限动态路由（后端返回菜单/权限码）。
• 统一请求封装：注入 JWT、X-Request-Id、错误提示、401 跳登录、403 提示无权限、429 重试提示。
• 列表页统一：筛选/分页/排序/导出/空态/加载态/错误重试；批量操作二次确认。
• 图表：ECharts；历史查询防抖；大数据量降采样由后端完成。
• 实时：设备详情/告警中心用 WebSocket/SSE（/ws?token=），断线指数退避重连；页面隐藏时暂停。
• 危险操作（控制/OTA/删除/转移）必须输入设备名或二次确认；操作结果用 request_id 跟踪。

8.10.2 微信小程序页面地图（养殖户）

• 实时数据优先 WebSocket；不支持时 10s 轮询；页面隐藏停止。
• 弱网：本地缓存最近数据 + "数据可能延迟"提示；控制命令离线时明确提示"设备离线，已暂存/不可用"。
• 订阅消息：每次告警前引导授权；未授权则站内信 + 短信（套餐允许）。
• 分享：设备/报表分享给家人用小程序卡片；公开报表用 H5 签名链接（只读、过期）。
• 蓝牙配网：wx.openBluetoothAdapter → 扫描 → 连接 → 写 WiFi/APN/sn/secret → 读回执 → 绑定时校验；失败提示重试；不把 secret 显示在页面。

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}

• 单连接最多订阅 200 个设备；每 30s ping/pong；服务端限流 10 条/秒/连接。
• 水平扩展：多 API 实例用 Redis Pub/Sub 广播；客户端只连一个实例。
• 断线重连后拉一次全量最新状态，避免丢消息。

8.10.4 交互与可用性基线

• 所有列表/详情必须有：加载骨架、空态、错误重试、无权限提示。
• 数值展示：按物模型 precision/unit；缺测显示 --，不得显示 0。
• 告警颜色：1 蓝/2 黄/3 红；色盲友好图标 + 文本。
• 时间统一展示设备时区，hover 显示完整时间。
• 移动端适配 320px~430px；按钮触达 ≥44px；表单键盘类型正确。
• 文案避免术语：DO→"溶氧"，RCD→"漏电保护"，VFD→"变频器"。

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":"不能为空"}]}

• HTTP 状态码表达大类（200/201/400/401/403/404/409/422/429/500/503），业务码在 code。
• 分页：{"items":[],"page":1,"page_size":20,"total":100}；游标分页用于大表（历史/日志）。
• 时间：ISO8601 带时区；金额：整数分；ID：字符串。
• 所有写接口支持 Idempotency-Key；重复请求返回首次结果。
• API 版本：URL /api/v1；破坏性变更升 /api/v2；旧版本至少保留 6 个月并在响应头 Deprecation 提示。
• OpenAPI：Node/Express 用 swagger-jsdoc 或 zod-to-openapi，FastAPI 内置生成；统一输出 /api/docs、/api/openapi.json；每个路由必须有 summary/description/tags 与示例。

8.11.2 错误码目录（节选，完整表进代码常量）

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

8.11.3 接口幂等与并发

• 创建类：Idempotency-Key + 唯一约束（如 sn）双保险。
• 控制类：request_id 由平台生成；同一设备同一属性 1s 内重复点击合并。
• 更新类：乐观锁 version 字段（产品/物模型/规则/看板）；冲突返回 409。
• 删除类：软删除 + 依赖检查（有设备的产品不可删）。
• 批量类：返回逐项结果，允许部分成功；前端展示失败原因。

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()
);

• 三层限流：Nginx limit_req 做粗粒度 IP 限流；应用层 Redis 滑动窗口按 tenant/user/ip/access_key 限流；MQTT 按网关限流。
• 默认值：登录 10 次/分/IP；认证 5 次/分/用户；写 60 次/分/用户；读 600 次/分/用户；开放 API 按 access_key.rate_limit；MQTT 每网关 120 条/分（可配）。
• 超限返回 429 + X-RateLimit-Limit/Remaining/Reset + 业务码 42901；连续超限 5 分钟临时封禁。
• 配额检查：设备创建/绑定、用户创建、API Key 创建、历史导出、存储写入统一调 quota.check(tenant, key)；tenant_quota 覆盖 plan.features，平台管理员可调。

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

共同约束：

• 分层依赖方向：controller/router → service → repository → model；禁止 controller 直接写 SQL。
• 所有 DB 操作通过 Repository；跨表事务在 service 层；外部调用（MQTT/HTTP/SMS）不放在事务内。
• 异步：Node 用 async/await + Prisma/Drizzle；FastAPI 用 async + SQLAlchemy 2.x async。
• CPU 密集/阻塞调用必须离开主请求链路：Node 放 BullMQ worker / worker_threads；FastAPI 放独立进程/Celery。
• 选 Node 时，src/modules/<domain> 与 M1~M7 一一对应；选 FastAPI 时 app/routers/<domain> 对应，二者目录可互相映射。

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 环境变量与密钥（禁止硬编码）

• 生产密钥放 KMS/Secrets Manager 或 Docker secrets；.env 只用于本地且不进 Git。
• 启动时校验必填变量，缺失直接失败（fail fast）。
• 密钥轮换：JWT kid 双密钥；AccessKey 新旧双活；设备 secret 支持批量轮换（OTA 配置）。

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";
  }
}

• /internal/* 只允许 Docker 内网（allow 172.16.0.0/12; deny all;）。
• 静态 UI 独立子域可进一步隔离；禁止目录列表；大文件上传限 20MB。
• 所有 HTTP 跳 HTTPS；HSTS；gzip/br；API 超时 30s（导出/OTA 用异步任务）。

8.12.5 数据库迁移与初始化

• 迁移：Node 用 Prisma Migrate（prisma migrate deploy）或 Drizzle Kit；FastAPI 用 Alembic（alembic upgrade head）。禁止手改生产表。
• 初始化顺序：扩展 → 基础表 → 索引/分区 → 种子（权限点/内置角色/套餐/品类模板/告警模板/通知模板）→ 平台管理员。
• 种子脚本幂等（ON CONFLICT DO NOTHING）；Node npm run seed，FastAPI python scripts/seed.py，均可重复执行。
• 回滚：每个 migration 必须有 down/rollback；不可逆变更（删列）分两次发布（先停用再删）。

CI/CD 流水线（最小）：

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

• 生产发布必须可回滚（保留上一镜像 + 数据库向下迁移）；发布窗口避开凌晨告警高峰。
• 配置变更（规则/模板/套餐）走后台 + 审计，不发版；代码变更才走流水线。

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

8.12.7 备份、容灾与降级

降级：

8.12.8 安全基线（上线前必须）

• OWASP Top 10 自查：注入（ORM 参数化）、越权（租户校验）、认证、敏感数据、XXE、SSRF（Webhook 目标校验内网 IP 黑名单）、XSS（自定义 UI/通知模板）、反序列化、依赖漏洞（pip-audit/npm audit）。
• 上传：白名单 + 大小 + 路径 + MIME + 沙箱；禁止执行。
• MQTT：一机一密/ACL/TLS/限流/防重放；匿名仅开发。
• 接口：限流、幂等、CORS 白名单、CSRF（Cookie 模式时）、安全响应头。
• 数据：手机号/地址/ICCID 脱敏；导出审计；备份加密；删除请求可执行。
• 供应链：镜像扫描（Trivy）、依赖锁版本、CI 签名。
• 渗透测试：上线前一次；每年一次；修复 P0/P1。

8.12.9 容量与性能目标

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

8.13.1 测试金字塔

8.13.2 设备模拟器（必须交付）

• 独立 CLI/脚本 tools/device_sim.ts（Node，推荐）或 tools/device_sim.py（FastAPI 备选）：支持指定 product_key、sn、上报周期、字段、事件、故障注入（离线/乱序/重复/非法/补传）。
• 支持网关模式：一个连接代理多子设备，上报 gw4g.devices。
• 支持命令响应：成功/失败/超时/拒绝（code=6）。
• 用途：开发联调、CI E2E、压测、演示；不依赖真实硬件。

8.13.3 必测场景（验收用例）

8.13.4 上线检查（Go/No-Go）

• [ ] 数据库迁移 + 种子 + 回滚脚本演练通过
• [ ] EMQX TLS/认证/ACL 开启，匿名关闭
• [ ] 所有 /internal/* 仅内网；密钥不在代码/镜像
• [ ] 备份/恢复演练通过；监控/告警接通
• [ ] 压测达标；E2E 全绿；安全扫描无 P0/P1
• [ ] 隐私政策/用户授权/数据删除流程就绪
• [ ] 客服/运维 runbook、回滚方案、值班表就绪

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

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

8.15 实施顺序与完成定义（DoD）

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

排期映射：§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）

• [ ] MQTT 3.1.1 + TLS 8883 + WSS 8084 可用；匿名关闭
• [ ] 一机一密/网关级认证；错误凭据拒绝；密钥只存哈希
• [ ] 主题 aqua/<sn>.<product_key>/up|cmd 与 api-mqtt 完全一致
• [ ] 共享订阅 $share/aqua-core/aqua/+/up 多 worker 负载均衡
• [ ] status/data/event/cmd_result/config/log/ota/err 全类型处理
• [ ] 未知 type 不丢弃；非法报文记日志
• [ ] LWT + 3×keepalive 离线判定；重连退避
• [ ] 命令闭环：seq/request_id/超时/重试/code=6 透传
• [ ] 离线影子 desired/reported；上线同步；安全命令不暂存
• [ ] 断网补传 cache/collect_ts 去重；不触发实时告警
• [ ] 网关/子设备拓扑同步；换网关迁移；子设备丢失告警
• [ ] EMQX ACL 防伪造；App 只读自己设备
• [ ] 时间同步/时钟偏差记录
• [ ] 设备模拟器可离线/乱序/重复/故障注入

9.2 产品与物模型（M3）

• [ ] 产品创建/发布/下线/删除约束；product_key=product_id
• [ ] 物模型三要素完整 Schema；7 种数据类型
• [ ] 字段命名 ≤10、单位/量程/精度；枚举/数组/对象
• [ ] 动态校验：类型/范围/只读/未知字段
• [ ] 版本管理：草稿/发布/兼容性/设备绑定版本
• [ ] 模板实例化；15 个品类种子；api-mqtt 字段映射
• [ ] 事件级别 info/warning/critical 映射 1/2/3
• [ ] 在线模拟器覆盖属性/服务/事件/非法用例
• [ ] 产品扁平分组（一级，无树）；vendor 归属；设备授权/密钥重置

9.3 设备管理（M4）

• [ ] 设备注册/激活/批量导入/导出；secret 一次性展示
• [ ] 二维码绑定/激活码/申请转移；一设备一 owner
• [ ] 共享只读/可控/运维授权/过期；解绑/转移/换网关
• [ ] 生命周期状态机；禁用/启用/软删除
• [ ] 设备影子/实时控制/数据监测
• [ ] 分组/塘口/基地；组级批量控制二次确认
• [ ] 设备日志 8 类；保留/归档/导出
• [ ] 在线调试：报文流/表单下发/虚拟设备
• [ ] 默认组件覆盖全部数据类型；自定义 UI 回退
• [ ] 设备定位/地图（可选）

9.4 数据存储（M5）

• [ ] Redis 最新状态 + 在线键 + 期望状态 + seq + 去重
• [ ] property_history 按月分区；唯一约束；quality
• [ ] 小时/天聚合；保留策略可配置
• [ ] 分区预建/归档任务；分布式锁；幂等
• [ ] 历史查询 raw/1h/1d/auto；缺测/补传/权限
• [ ] 最新值/统计/导出 API
• [ ] 时序库演进方案（>500 台）
• [ ] 容量估算与压测达标
• [ ] 备份/恢复演练

9.5 规则与告警（M6）

• [ ] 属性/事件/定时触发；跨设备/跨产品
• [ ] 嵌套条件/持续时长/回差/时间窗
• [ ] 动作：服务/通知/虚拟属性/转发/工单/延迟/场景
• [ ] 冷却/防抖/恢复；规则优先级/冲突
• [ ] 规则执行日志；失败不阻塞落库
• [ ] 告警去重/级别/确认/恢复/关闭/升级/导出
• [ ] 通知渠道（站内信/微信/短信/语音/邮件/Webhook）
• [ ] 通知模板/静音/重试/失败降级
• [ ] 场景联动串行/并行；水产预设场景
• [ ] 数据转发 HTTP/Kafka/MQTT/DB；签名/重试/日志
• [ ] 调度任务幂等/锁/监控

9.6 开放平台（M7）

• [ ] AccessKey 创建/轮换/吊销/过期/IP 白名单
• [ ] HMAC 签名/时间窗/nonce/body hash
• [ ] 限流/scope/审计；错误码不泄露内部
• [ ] 只读 API 全量；写操作白名单
• [ ] 全量 API 路由表覆盖各模块；权限/分页/幂等
• [ ] Webhook 注册/签名/重试/幂等/停用
• [ ] Webhook payload schema/示例/字段兼容
• [ ] OpenAPI 文档 + 多语言示例
• [ ] 数据转发授权与脱敏

9.7 多租户与安全（M1）

• [ ] 四级租户树；创建/停用/过户/删除约束
• [ ] 用户/角色/权限/成员邀请；四个内置角色
• [ ] 微信登录/账密/短信/JWT/刷新/登出/强制下线
• [ ] 设备绑定/共享/转移；跨租户 404/403
• [ ] Repository 强制 tenant_id；缓存键带租户
• [ ] 审计日志：谁/何时/对什么/做了什么/结果
• [ ] 用户通知偏好（渠道/设备静音/免打扰）
• [ ] 密码/密钥哈希；TLS；敏感数据脱敏
• [ ] 隐私政策/授权记录；数据导出/删除 API（法定保留匿名化）
• [ ] 全局 API 限流 + 租户配额
• [ ] OWASP 自查/依赖扫描/渗透测试

9.8 可选模块（O1~O7）

• [ ] O1 上传白名单/沙箱/桥接/版本回滚/回退
• [ ] O2 看板拖拽/组件/共享/预设模板
• [ ] O3 OTA 灰度/进度/回滚/兼容/BLE；签名公钥/轮换/吊销、回滚任务 API
• [ ] O4 套餐/支付/回调/降级/本地控制不停
• [ ] O5 工单/SLA/派单/诊断/回访
• [ ] O6 报告模板/生成/标准引用/导出；投喂/用药/巡塘/死亡/出塘记录（V3）
• [ ] O7 协议插件/视频/AI 扩展位

9.9 前端与体验

• [ ] Web 后台页面/权限/空态/错误/批量/确认
• [ ] 小程序页面/实时/弱网/订阅消息/分享/蓝牙配网
• [ ] WebSocket 鉴权/订阅/重连/限流
• [ ] 数值/单位/精度/缺测/时区/颜色/文案
• [ ] 移动端适配/触达/键盘类型
• [ ] 自定义 UI 与默认组件数据通路一致

9.10 工程与运维

• [ ] 目录分层/依赖方向/异步/事务边界
• [ ] 环境变量/密钥管理/fail fast
• [ ] docker-compose 一键起；健康检查
• [ ] Nginx TLS/反代/内网限制/静态隔离
• [ ] 迁移/种子/回滚（Prisma Migrate 或 Alembic）
• [ ] 结构化日志/指标/追踪/告警
• [ ] 备份/RPO/RTO/恢复演练/降级
• [ ] CI/CD/镜像扫描/依赖审计
• [ ] 压测/ E2E / 混沌 / 安全测试
• [ ] 上线 Go/No-Go 全绿

9.11 智慧水产业务闭环

• [ ] DO 实时监测 + 24h/7d/30d 曲线
• [ ] DO 低/危险、离线、传感器离线、RCD、箱温告警
• [ ] 微信/短信/电话分级通知；静音/升级
• [ ] 增氧机开关/变频调速；仅授权角色
• [ ] DO 阈值下发设备本地生效；断网仍闭环
• [ ] 多塘/基地/分组/一屏总览
• [ ] 传感器维护/校准提醒；易损件周期
• [ ] 水质报告引用 GB 11607-89 等标准
• [ ] 节能报告/回本测算；投喂建议（V3）
• [ ] 投喂/用药/巡塘/死亡/出塘记录（V3，pond_operation_log）
• [ ] 订阅计费/到期降级；本地控制永不停
• [ ] 工单/SLA/远程诊断/回访
• [ ] 事故追溯时间线（数据+动作+告警+工单）

最终判定：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. 术语表