diff --git a/docs/knowledge-base/06-platform-integration-design.md b/docs/knowledge-base/06-platform-integration-design.md new file mode 100644 index 00000000..56e2ab13 --- /dev/null +++ b/docs/knowledge-base/06-platform-integration-design.md @@ -0,0 +1,1238 @@ +# 平台对接需求与技术设计 + +## 1. 背景 + +需求目录 `docs/platform/` 中目前包含 DCP 雨量计、倾角、静力水准、轴力计接口文档。DCP 是当前优先落地的平台,但平台对接能力不能只服务一个平台、一个租户。平台一旦完成接入,后续其他租户应复用同一套平台定义、接口定义、流程编排、适配器和数据交换执行能力;租户只维护自己的第三方账号/用户、项目、设备、字段映射和交换任务。 + +平台与平台之间的数据交换本质上分三类: + +| 方向 | FastBee 角色 | 触发方 | 说明 | +| --------------- | ---------- | ------- | ---------------------------------------- | +| OUTBOUND_PUSH | Producer | FastBee | FastBee 主动把本系统数据推送给第三方。 | +| INBOUND_PULL | Consumer | FastBee | FastBee 主动从第三方拉取数据,但消费方仍是 FastBee。 | +| INBOUND_RECEIVE | Consumer | 第三方 | 第三方主动调用 FastBee 接口或消息入口,FastBee 接收并消费数据。 | + +因此本方案不再以“推送”为中心,而是以更普适的 **数据交换 Exchange** 为中心: + +```text +平台对接 = 平台模板 + 租户接入 + 资源映射 + 流程编排 + 数据交换任务 + 数据交换记录 +``` + +当前 FastBee 已具备以下基础能力: + +- 设备、产品、物模型:设备通过 `iot_device` 管理,物模型通过产品和设备编号查询。 +- 设备上报处理:设备数据解析后进入原有入库和规则处理链路。 +- 历史数据查询:数据中心接口支持按设备、时间窗口和物模型标识符查询历史数据。 +- 数据桥接:现有 `bridge`、`iot_script`、`iot_script_bridge` 支持 HTTP、MQTT、数据库桥接,可作为轻量转发能力保留。 +- 定时任务:系统已有 Quartz 任务体系,可复用为 OUTBOUND_PUSH、INBOUND_PULL、RESOURCE_SYNC 等任务触发器。 +- 多租户:IoT 主要业务表已有 `tenant_id`、`tenant_name` 约定,平台对接也必须按租户隔离业务映射和交换数据。 + +## 2. 设计目标 + +平台对接需要同时满足“平台共用”“租户隔离”“方向清晰”“数据边界清晰”: + +- 平台接入人员只维护一次第三方平台模板:认证方式、接口、字段、流程、适配器。 +- 多个租户可启用同一个平台模板,不重复配置接口和流程。 +- 每个租户维护自己的第三方账号/用户授权、项目映射、设备绑定、字段映射和交换任务。 +- 支持三类数据方向:主动推送、主动拉取、被动接收。 +- 主动拉取和被动接收都属于 INBOUND,FastBee 是 consumer,最终要落到 FastBee 的设备、物模型、事件、告警、资源映射或业务表。 +- 主动推送属于 OUTBOUND,FastBee 是 producer,重点是可靠投递、重试、重推和审计。 +- 支持幂等、失败重试、手动重试、补偿、日志追踪。 +- DCP 等复杂平台通过专用适配器处理 token、项目、测点、设备、传感器、雨量递增等平台规则。 +- DCP 必须遵守 **一个测点对应一个 FastBee 设备**,不能将测点设计为平台级或产品级共享资源。 + +## 3. 总体架构 + +平台对接按四层拆分。 + +```text +平台模板层(全局/系统级,可复用) + -> 平台定义、接口定义、字段定义、流程编排、适配器 + +租户接入层(租户级,隔离) + -> 租户平台实例、第三方账号/用户、项目映射、资源映射、设备绑定、字段映射 + +数据交换层(租户级,隔离) + -> 数据交换任务、交换记录、交换尝试、幂等、重试、补偿 + +业务落点层(FastBee 内部边界) + -> 设备数据、设备日志、物模型值、事件、告警、资源映射、业务表 +``` + +### 3.1 平台模板层 + +平台模板层由平台管理员维护,配置一次、多租户复用。 + +包含: + +- 第三方平台类型,例如 DCP、通用 HTTP、通用 MQTT、通用数据库。 +- 认证方式,例如无认证、固定 Token、Basic、Bearer、OAuth2、自定义 Token 接口。 +- 接口定义,例如获取用户、获取项目、创建测点、创建设备、绑定传感器、推送数据、查询数据、接收数据。 +- 接口字段定义,用于页面展示和字段映射。 +- 流程编排,例如“获取用户 -> 获取项目 -> 确保测点 -> 确保设备 -> 推送数据”。 +- 适配器编码,例如 `genericHttp`、`dcp`。 + +平台模板不保存租户账号、项目、设备和业务数据。 + +### 3.2 租户接入层 + +租户接入层由租户管理员或实施人员维护。 + +包含: + +- 租户启用哪个平台模板。 +- 租户自己的第三方账号、token、clientId/clientSecret 等授权配置。 +- 租户第三方用户和项目映射。 +- FastBee 设备与第三方资源的绑定。 +- FastBee 物模型字段与第三方接口字段的映射。 +- 接收入口密钥、签名配置、白名单等租户级安全配置。 + +同一个平台模板下,不同租户的数据必须隔离:授权配置、项目、用户、资源映射、设备绑定、交换任务、交换记录都必须带 `tenant_id`。 + +### 3.3 数据交换层 + +数据交换层是运行时核心。它不只处理推送,也处理拉取和接收。 + +```text +OUTBOUND_PUSH + FastBee 内部数据 -> 第三方平台 + +INBOUND_PULL + 第三方平台 -> FastBee 主动拉取 -> FastBee 内部业务落点 + +INBOUND_RECEIVE + 第三方主动提交 -> FastBee 接收入口 -> FastBee 内部业务落点 +``` + +三种方向共用平台模板、租户接入、认证、字段映射、流程编排、资源映射和交换记录,但使用不同触发器和执行器。 + +### 3.4 业务落点层 + +INBOUND 数据必须明确落点,避免平台对接层无限扩张。 + +推荐落点类型: + +| sink_type | 说明 | +| ---------------- | ------------------- | +| DEVICE_PROPERTY | 写入设备属性/物模型数据。 | +| DEVICE_EVENT | 写入设备事件。 | +| ALERT_EVENT | 写入告警或告警日志。 | +| RESOURCE_MAPPING | 写入用户、项目、测点、设备等资源映射。 | +| BUSINESS_TABLE | 写入明确的业务表,需要白名单。 | +| RAW_ONLY | 只记录原始数据,不进入业务表。 | + +平台对接层负责接入、鉴权、映射、转换、幂等和审计;具体业务落库应通过明确的 sink adapter 完成,不允许任意 SQL 或任意代码写入。 + +## 4. 三类方向的边界 + +### 4.1 OUTBOUND_PUSH:推送数据 + +FastBee 是 producer,第三方平台是 consumer。 + +```text +设备上报 / 历史补推 / 业务事件 + -> 构造标准出站数据 PlatformExchangeMessage + -> 匹配租户平台实例和设备绑定 + -> 字段映射和资源补齐 + -> 执行 PUSH 流程 + -> 调用第三方接口/MQTT/DB + -> 写 exchange_record / exchange_attempt +``` + +边界: + +- 数据来源是 FastBee 内部可信业务数据。 +- 失败可重试,可手动重推。 +- 需要 outbox,避免阻塞设备上报主链路。 +- 幂等 key 应基于本地来源、绑定、时间和 payload hash。 + +### 4.2 INBOUND_PULL:拉取数据 + +FastBee 主动发起请求,但 FastBee 仍是 consumer。 + +```text +Quartz / 手动触发 + -> 加载租户平台实例和授权配置 + -> 执行 PULL / RESOURCE_SYNC 流程 + -> 调第三方查询接口 + -> 处理分页、游标、时间窗口 + -> 响应字段映射 + -> 识别本地租户、项目、设备、资源 + -> 写入业务落点 + -> 写 exchange_record / exchange_attempt +``` + +边界: + +- 数据来源是第三方平台,必须按外部输入处理。 +- 必须有游标、时间窗口或分页边界,禁止无界拉取。 +- 必须定义 sink_type,明确拉取结果落到哪里。 +- 必须支持幂等,避免同一第三方数据重复入库。 +- 拉取状态建议保存在任务游标中,例如 `cursor_json`、`last_success_time`、`last_success_external_id`。 + +### 4.3 INBOUND_RECEIVE:接收数据 + +第三方主动调用 FastBee,FastBee 是 consumer。 + +```text +第三方 HTTP/MQTT 调用 FastBee + -> 接收入口鉴权/签名/限流 + -> 匹配租户平台实例和 RECEIVE 流程 + -> 提取幂等 key + -> 请求字段映射 + -> 识别本地租户、项目、设备、资源 + -> 写入业务落点 + -> 写 exchange_record / exchange_attempt + -> 返回处理结果 +``` + +边界: + +- 数据来源是第三方平台,必须按外部输入处理。 +- 接收入口必须具备租户识别、鉴权、签名校验、幂等、限流和日志。 +- 同步响应只代表 FastBee 是否接收/处理成功,不代表后续所有异步流程都完成。 +- 如果需要异步处理,接收入口应先写 exchange_record,再由 worker 处理。 +- 不建议复用旧 `/bridge/**` 作为主入口,可保留旧桥接用于轻量场景。 + +## 5. 与现有系统功能的关系 + +| 现有能力 | 复用方式 | +| ----------------- | ----------------------------------------------------------------- | +| 设备管理 `iot_device` | 设备绑定引用 `device_id`、`serial_number`、`product_id`、`tenant_id`。 | +| 产品物模型 | 绑定和字段映射页面通过现有物模型接口展示字段,使用 `identifier`。 | +| 设备实时上报链路 | OUTBOUND_PUSH 可在原有数据处理完成后生成交换记录,不改变协议解析逻辑。 | +| 数据中心历史查询 | OUTBOUND_PUSH 补推可按设备、时间窗口、identifier 取历史数据。 | +| 规则引擎和数据桥接 | 保留给轻量 HTTP/MQTT/DB 转发;平台对接面向多步骤、多租户、可审计场景。 | +| Quartz 定时任务 | INBOUND_PULL、OUTBOUND_PUSH 定时补偿、RESOURCE_SYNC 可复用 sys_job/Quartz。 | +| Redis/Redisson | 缓存 token,资源创建时做并发锁,接收入口可做幂等和限流。 | +| 若依权限体系 | 继续使用菜单、角色、`@PreAuthorize` 权限控制。 | + +现有数据桥接仍保留。它适合“固定 URL + 固定模板 + 脚本触发”的轻量转发;平台对接用于第三方平台长期集成,需要多租户账号、项目、设备映射、流程编排、资源幂等、数据落点、失败重试和审计。 + +## 6. 前端功能设计 + +一级菜单:**平台对接**。 + +建议菜单: + +```text +平台对接 + 平台模板 + 接口配置 + 流程编排 + 租户接入 + 资源映射 + 设备绑定 + 数据交换任务 + 数据交换记录 +``` + +### 6.1 平台模板 + +平台模板是系统级可复用配置。 + +主要字段: + +| 字段 | 说明 | +| ------ | ----------------------------------------- | +| 平台名称 | 用户可读名称,例如 DCP。 | +| 平台编码 | 全局唯一,例如 `DCP`。 | +| 平台类型 | DCP、GENERIC_HTTP、GENERIC_MQTT、GENERIC_DB。 | +| 适配器编码 | dcp、genericHttp 等。 | +| 基础地址模板 | 可作为租户实例默认 baseUrl。 | +| 认证方式 | OAuth2、自定义 Token、固定 Token 等。 | +| 默认请求配置 | 超时、重试、Content-Type 等。 | +| 启用状态 | 启用、停用。 | + +### 6.2 接口配置 + +接口配置属于平台模板,可被所有租户复用。 + +接口类型: + +| 类型 | 说明 | +| --------------- | ----------------------- | +| TOKEN | 获取或刷新 token。 | +| QUERY | 查询第三方基础数据,如用户、项目、设备、数据。 | +| RECEIVE | FastBee 接收入口定义。 | +| RESOURCE_CREATE | 创建外部资源,如测点、设备、传感器。 | +| RESOURCE_BIND | 绑定外部资源。 | +| PUSH | 推送设备数据。 | +| CALLBACK | 回调第三方处理结果,可选。 | + +配置项: + +- 请求方法:GET、POST、PUT、DELETE。 +- URL 模板:支持 `${变量}`。 +- Header、Query、Body 模板。 +- Content-Type。 +- 成功条件:HTTP 状态码、响应 JSONPath。 +- 响应变量提取:例如 `$.id -> pointId`。 +- 超时和重试策略。 +- RECEIVE 接口还需要配置入口路径、鉴权方式、签名规则、幂等 key 提取规则。 + +### 6.3 接口字段 + +接口字段用于页面展示和字段映射。 + +| 字段 | 说明 | +| ---- | ----------------------------------- | +| 接口 | 所属接口。 | +| 方向 | REQUEST、RESPONSE。 | +| 字段路径 | JSONPath 或表单字段名。 | +| 字段名称 | 展示名称。 | +| 数据类型 | string、number、boolean、array、object。 | +| 是否必填 | 是否必填。 | +| 默认值 | 默认值。 | +| 示例值 | 示例值。 | + +### 6.4 流程编排 + +流程属于平台模板,用于将多个接口和适配器方法组成可复用业务流程。 + +流程类型: + +| flow_type | 说明 | +| --------------- | --------------------- | +| OUTBOUND_PUSH | FastBee 推送数据到第三方。 | +| INBOUND_PULL | FastBee 从第三方拉取并消费数据。 | +| INBOUND_RECEIVE | FastBee 接收第三方提交并消费数据。 | +| RESOURCE_SYNC | 同步第三方用户、项目、设备、测点等资源。 | +| CALLBACK | 回调第三方处理结果,可选。 | + +流程步骤类型: + +| step_type | 说明 | +| --------- | ---------------- | +| API | 调用平台接口配置。 | +| ADAPTER | 调用专用适配器方法。 | +| CONDITION | 条件判断。 | +| TRANSFORM | 数据转换。 | +| SINK | 写入 FastBee 业务落点。 | + +流程模板可以被多个租户复用;步骤执行时使用当前租户的授权配置、项目映射、设备绑定、资源映射和字段映射。 + +### 6.5 租户接入 + +租户接入表示某个租户启用某个平台模板。 + +主要字段: + +| 字段 | 说明 | +| ------- | -------------------------------- | +| 租户 | 当前租户。 | +| 平台模板 | 选择 DCP 等平台模板。 | +| 实例名称 | 租户内展示名。 | +| baseUrl | 可覆盖模板默认地址。 | +| 授权配置 | 租户自己的账号、clientId、secret、token 等。 | +| 默认第三方用户 | DCP userId 或其他平台用户。 | +| 默认项目 | DCP projectId 或其他平台项目。 | +| 接收安全配置 | webhook token、签名密钥、IP 白名单等。 | +| 启用状态 | 启用、停用。 | + +租户接入配置完成后,该租户才能建立资源映射、设备绑定、交换任务和接收入口。 + +### 6.6 资源映射 + +资源映射是平台对接的核心边界之一,表示 FastBee 本地对象和第三方对象的对应关系。 + +常见资源类型: + +| resource_type | 说明 | +| ------------- | -------- | +| USER | 第三方用户。 | +| PROJECT | 第三方项目。 | +| DEVICE | 第三方设备。 | +| POINT | 第三方测点。 | +| SENSOR | 第三方传感器。 | +| MODULE | 第三方模块。 | +| VIRTUAL | 第三方虚拟资源。 | +| CUSTOM | 其他平台资源。 | + +资源映射必须带 `tenant_id` 和 `tenant_channel_id`。除平台模板定义外,第三方资源 id 不允许全局共享。 + +### 6.7 设备绑定 + +设备绑定是租户级配置。 + +绑定内容: + +```text +租户平台实例 + FastBee 设备 + 平台流程 + 第三方用户/项目/资源 + 字段映射 + 交换模式 +``` + +绑定页面左侧使用现有 FastBee 产品、设备、物模型;右侧使用平台模板的接口字段和租户的第三方用户/项目/资源。 + +绑定支持:新增、修改、启用、停用、解绑、初始化资源、查看资源映射、查看交换记录。 + +DCP 绑定规则: + +```text +FastBee 一个设备 = DCP 一个测点 = DCP 一个设备资源 +``` + +DCP 的 `pointId`、`dcpDeviceId`、sensor/module/virtual 等必须按租户设备绑定保存,不能平台级共享。 + +### 6.8 字段映射 + +字段映射按设备绑定维护,也可以支持从平台模板复制默认映射。 + +映射来源: + +| source_type | 说明 | +| ----------- | ------------------------------------------- | +| THING_MODEL | 物模型字段,例如 rainfall。 | +| SYSTEM | 系统字段,例如 serialNumber、deviceName、reportTime。 | +| CONSTANT | 常量。 | +| RESOURCE | 外部资源映射,例如 pointId、dcpDeviceId。 | +| JSONPATH | 从 rawPayload 或第三方响应中取值。 | +| EXPRESSION | 受限表达式,后续增强。 | + +字段映射同时服务 OUTBOUND 和 INBOUND: + +- OUTBOUND:FastBee 内部字段映射到第三方字段。 +- INBOUND:第三方字段映射到 FastBee 标准消息或业务落点字段。 + +表达式必须受限,不能执行任意代码。 + +### 6.9 数据交换任务与记录 + +数据交换任务是租户级配置。 + +任务类型: + +| task_type | 说明 | +| ------------- | ----------------------- | +| OUTBOUND_PUSH | 定时推送或补推 FastBee 数据到第三方。 | +| INBOUND_PULL | 定时拉取第三方数据并消费。 | +| RESOURCE_SYNC | 定时同步第三方用户、项目、设备、测点等资源。 | + +INBOUND_RECEIVE 一般不依赖定时任务,而是通过接收入口触发;如果接收后异步处理,也会生成交换记录。 + +数据交换记录展示业务层面的交换状态: + +- 按租户、平台、方向、流程、设备、状态、时间筛选。 +- 查看原始输入、标准消息、最终 payload、幂等 key、请求响应、错误信息。 +- 失败重试、批量重试、成功记录强制重放。 + +## 7. 标准数据模型 + +平台对接内部统一使用标准数据交换消息,避免直接依赖原始 MQTT/HTTP payload 或第三方响应结构。 + +```json +{ + "direction": "OUTBOUND_PUSH", + "tenantId": 1, + "tenantChannelId": 10, + "flowId": 100, + "bindingId": 1001, + "sourceType": "DEVICE_PROPERTY", + "sourceRefId": "device_log_id_or_external_message_id", + "eventTime": "2026-05-27 10:00:00", + "device": { + "productId": 10, + "deviceId": 1001, + "serialNumber": "ABC001", + "deviceName": "雨量计-001" + }, + "values": { + "rainfall": { + "value": 12.3, + "unit": "mm", + "ts": "2026-05-27 10:00:00" + } + }, + "resources": { + "projectId": "P001", + "pointId": "PT001", + "externalDeviceId": "D001" + }, + "rawPayload": {} +} +``` + +说明: + +- OUTBOUND 的 `rawPayload` 通常是 FastBee 原始上报或业务事件。 +- INBOUND 的 `rawPayload` 是第三方原始请求或响应。 +- `values` 的 key 优先使用 FastBee 物模型 `identifier`。 +- `resources` 来自租户级资源映射。 +- INBOUND 处理完成后必须进入明确的 `sink_type`,不能停留在平台对接层。 + +## 8. DCP 适配设计 + +DCP 通过平台模板 + 流程编排 + 专用适配器落地。DCP 当前主要覆盖 OUTBOUND_PUSH 和 RESOURCE_SYNC,也可以扩展 INBOUND_PULL 同步项目、测点或历史数据。 + +### 8.1 DCP 平台模板 + +DCP 平台模板包含: + +- token 接口:`POST /connect/token`。 +- 用户接口:获取 userId。 +- 项目接口:按 userId 获取项目。 +- 资源接口:创建测点、创建设备、创建/绑定传感器等。 +- 数据接口:雨量、倾角、静力水准、轴力计数据推送;如 DCP 提供查询接口,也可配置为 INBOUND_PULL。 +- DCP 专用适配器:处理 token 缓存、资源幂等、payload 构造、特殊规则。 + +这些配置由系统维护一次,所有租户复用。 + +### 8.2 DCP 租户映射 + +每个租户需要维护自己的 DCP 映射: + +```text +租户 + -> DCP 租户平台实例 + -> DCP 账号/用户 userId + -> DCP 项目 projectId + -> FastBee 设备绑定 DCP 测点 pointId + -> FastBee 设备绑定 DCP 设备 dcpDeviceId + -> 设备附属 sensor/module/virtual +``` + +同一个 DCP 平台模板下,不同租户可以使用不同 DCP 账号、用户和项目。资源映射必须按 `tenant_id` 隔离。 + +### 8.3 DCP 一测点一设备约束 + +DCP 资源映射必须按“一测点一设备”建模: + +```text +FastBee Device(serialNumber) + -> DCP Project(projectId) + -> DCP Point(pointId) + -> DCP Device(dcpDeviceId) + -> DCP Sensor/Module/Virtual(按设备类型可选) +``` + +规则: + +- `PROJECT` 可以是租户平台实例级默认资源,也可以绑定级覆盖。 +- `USER`、`PROJECT` 是租户级或绑定级映射,不是全局共享。 +- `POINT` 必须是设备级资源,一个 FastBee 设备对应一个 DCP 测点。 +- `DEVICE` 必须是设备级资源,一个 FastBee 设备对应一个 DCP 设备资源。 +- `SENSOR`、`MODULE`、`VIRTUAL` 是设备附属资源,按绑定保存。 +- 静力水准 payload 中可包含多条 `sensorData`,但仍归属同一个 FastBee 设备对应的 DCP 测点和 DCP 设备下。 + +### 8.4 DCP OUTBOUND_PUSH 流程 + +```text +设备上报或补推任务 + -> 匹配租户 DCP 设备绑定 + -> 加载租户 DCP 授权配置 + -> 获取/刷新 token + -> 加载 userId/projectId 映射 + -> 加载 pointId/dcpDeviceId/sensor 映射 + -> 资源缺失时按流程懒加载或手动初始化 + -> 构造对应设备类型 payload + -> 调用 DCP 数据接口 + -> 写 exchange_record 和 exchange_attempt +``` + +资源创建必须幂等。并发上报同一设备时,使用 Redis/Redisson 锁: + +```text +platform:dcp:resource:{tenantId}:{tenantChannelId}:{serialNumber} +``` + +### 8.5 DCP 特殊规则 + +- 雨量 `value` 为累计降雨量,DCP 要求新值大于历史值。系统使用绑定的 `last_success_value` 或资源扩展字段记录最近成功值;不满足时记录为 `SKIPPED`,不反复重试。 +- 倾角需要同一时间的 x、y、z 和温度,必须按一次设备快照组包,不能拆成多个单字段推送。 +- 静力水准的 `sensorData` 可包含多条传感器数据,但这些数据属于同一个 FastBee 设备对应的 DCP 设备资源。 +- 轴力计按独立设备类型处理,字段映射和 payload 由 `DcpPlatformAdapter` 中对应 builder 负责。 +- DCP token 缓存在 Redis,key 建议:`platform:dcp:token:{tenantId}:{tenantChannelId}`。 +- 401 或 token 过期时刷新 token 后重试一次,再失败则记录错误。 + +## 9. 表设计 + +表名前缀使用 `iot_platform_`,避免和现有 `iot_social_platform` 混淆。 + +### 9.1 表清单 + +| 表名 | 层级 | 说明 | +| ------------------------------- | ----- | ------------------------- | +| `iot_platform_template` | 平台模板层 | 平台模板,全局可复用。 | +| `iot_platform_api` | 平台模板层 | 第三方接口定义,含调用接口和接收入口。 | +| `iot_platform_api_field` | 平台模板层 | 接口字段定义。 | +| `iot_platform_flow` | 平台模板层 | 流程模板。 | +| `iot_platform_flow_step` | 平台模板层 | 流程步骤。 | +| `iot_platform_tenant_channel` | 租户接入层 | 租户启用的平台实例、授权和接收安全配置。 | +| `iot_platform_resource_mapping` | 租户接入层 | 用户、项目、测点、设备、传感器等资源映射。 | +| `iot_platform_binding` | 租户接入层 | FastBee 设备与平台流程/项目/资源绑定。 | +| `iot_platform_field_mapping` | 租户接入层 | 绑定级字段映射。 | +| `iot_platform_exchange_task` | 数据交换层 | 定时推送、主动拉取、资源同步任务。 | +| `iot_platform_task_binding` | 数据交换层 | 交换任务与绑定关系。 | +| `iot_platform_exchange_record` | 数据交换层 | 数据交换记录,兼 outbox/inbox 审计。 | +| `iot_platform_exchange_attempt` | 数据交换层 | 每次交换尝试日志。 | + +### 9.2 iot_platform_template + +平台模板,全局可复用。 + +| 字段 | 类型 | 说明 | +| ----------------- | ------------ | ---------------------------------------- | +| id | bigint | 主键 | +| platform_code | varchar(64) | 平台编码,全局唯一 | +| platform_name | varchar(128) | 平台名称 | +| platform_type | varchar(32) | DCP、GENERIC_HTTP、GENERIC_MQTT、GENERIC_DB | +| adapter_code | varchar(64) | dcp、genericHttp 等 | +| base_url_template | varchar(512) | 默认基础地址模板 | +| auth_type | varchar(32) | 认证类型 | +| auth_schema | json | 授权配置字段定义,用于租户填写 | +| request_config | json | 默认超时、重试、TLS 等 | +| status | tinyint | 0停用,1启用 | +| create_time | datetime | 创建时间 | +| update_time | datetime | 更新时间 | +| remark | varchar(500) | 备注 | + +建议索引: + +```text +uk_platform_code(platform_code) +idx_platform_type(platform_type) +``` + +### 9.3 iot_platform_api + +第三方接口定义,属于平台模板。对 OUTBOUND/PULL 是要调用的第三方接口;对 RECEIVE 是 FastBee 暴露的接收入口定义。 + +| 字段 | 类型 | 说明 | +| ------------------ | ------------ | --------------------------------------------------------------- | +| id | bigint | 主键 | +| template_id | bigint | 平台模板 id | +| api_code | varchar(64) | 接口编码 | +| api_name | varchar(128) | 接口名称 | +| api_type | varchar(32) | TOKEN、QUERY、RECEIVE、PUSH、RESOURCE_CREATE、RESOURCE_BIND、CALLBACK | +| method | varchar(16) | GET、POST、PUT、DELETE | +| path_template | varchar(512) | URL 路径模板或接收入口路径模板 | +| content_type | varchar(64) | application/json、form-urlencoded | +| header_template | json | Header 模板 | +| query_template | json | Query 模板 | +| body_template | json | Body 模板 | +| auth_config | json | 接口级认证或接收校验配置 | +| idempotency_config | json | 幂等 key 提取规则 | +| success_condition | json | 成功条件 | +| response_mapping | json | 响应变量提取规则 | +| retry_policy | json | 接口级重试策略 | +| timeout_ms | int | 超时时间 | +| status | tinyint | 0停用,1启用 | +| create_time | datetime | 创建时间 | +| update_time | datetime | 更新时间 | + +建议索引: + +```text +uk_template_api_code(template_id, api_code) +idx_template_api_type(template_id, api_type) +``` + +### 9.4 iot_platform_api_field + +接口字段定义,用于页面展示和字段映射。 + +| 字段 | 类型 | 说明 | +| ------------- | ------------ | ---------------------------------- | +| id | bigint | 主键 | +| api_id | bigint | 接口 id | +| direction | varchar(16) | REQUEST、RESPONSE | +| field_path | varchar(256) | JSONPath 或表单字段名 | +| field_name | varchar(128) | 字段名称 | +| data_type | varchar(32) | string、number、boolean、array、object | +| required | tinyint | 是否必填 | +| default_value | varchar(512) | 默认值 | +| example_value | varchar(512) | 示例值 | +| description | varchar(500) | 说明 | +| sort_order | int | 排序 | + +建议索引: + +```text +idx_api_direction(api_id, direction) +``` + +### 9.5 iot_platform_flow + +流程模板,属于平台模板。 + +| 字段 | 类型 | 说明 | +| ------------ | ------------ | ----------------------------------------------------------------- | +| id | bigint | 主键 | +| template_id | bigint | 平台模板 id | +| flow_code | varchar(64) | 流程编码 | +| flow_name | varchar(128) | 流程名称 | +| flow_type | varchar(32) | OUTBOUND_PUSH、INBOUND_PULL、INBOUND_RECEIVE、RESOURCE_SYNC、CALLBACK | +| adapter_code | varchar(64) | 可覆盖平台适配器 | +| sink_type | varchar(32) | INBOUND 数据落点类型,OUTBOUND 可为空 | +| status | tinyint | 0停用,1启用 | +| create_time | datetime | 创建时间 | +| update_time | datetime | 更新时间 | +| remark | varchar(500) | 备注 | + +建议索引: + +```text +uk_template_flow_code(template_id, flow_code) +idx_template_flow_type(template_id, flow_type) +``` + +### 9.6 iot_platform_flow_step + +流程步骤。 + +| 字段 | 类型 | 说明 | +| -------------- | ------------ | ------------------------------------ | +| id | bigint | 主键 | +| flow_id | bigint | 流程 id | +| step_code | varchar(64) | 步骤编码 | +| step_name | varchar(128) | 步骤名称 | +| step_type | varchar(32) | API、ADAPTER、CONDITION、TRANSFORM、SINK | +| api_id | bigint | API 步骤引用接口 id | +| adapter_method | varchar(128) | 适配器方法,例如 ensurePoint | +| input_mapping | json | 步骤入参映射 | +| output_mapping | json | 步骤输出变量映射 | +| condition_expr | varchar(512) | 执行条件 | +| on_fail | varchar(32) | STOP、SKIP、RETRY | +| sort_order | int | 执行顺序 | +| status | tinyint | 0停用,1启用 | + +建议索引: + +```text +idx_flow_sort(flow_id, sort_order) +``` + +### 9.7 iot_platform_tenant_channel + +租户启用平台模板后的平台实例。授权配置、baseUrl、默认用户/项目、接收安全配置属于租户级。 + +| 字段 | 类型 | 说明 | +| ------------------ | ------------ | ------------------------- | +| id | bigint | 主键 | +| tenant_id | bigint | 租户 id | +| tenant_name | varchar(64) | 租户名称 | +| template_id | bigint | 平台模板 id | +| channel_code | varchar(64) | 租户内平台实例编码 | +| channel_name | varchar(128) | 租户内平台实例名称 | +| base_url | varchar(512) | 实际基础地址 | +| auth_config | json | 租户授权配置,敏感值加密 | +| receive_config | json | 接收安全配置,token、签名密钥、IP 白名单等 | +| default_user_id | varchar(128) | 默认第三方用户 id | +| default_project_id | varchar(128) | 默认第三方项目 id | +| config_json | json | 租户级扩展配置 | +| status | tinyint | 0停用,1启用 | +| del_flag | char(1) | 0正常,2删除 | +| create_by | varchar(64) | 创建者 | +| create_time | datetime | 创建时间 | +| update_by | varchar(64) | 更新者 | +| update_time | datetime | 更新时间 | +| remark | varchar(500) | 备注 | + +建议索引: + +```text +uk_tenant_channel_code(tenant_id, channel_code, del_flag) +idx_tenant_template(tenant_id, template_id) +``` + +### 9.8 iot_platform_resource_mapping + +外部资源映射,租户级隔离。 + +| 字段 | 类型 | 说明 | +| ----------------- | ------------ | ------------------------------------------------------ | +| id | bigint | 主键 | +| tenant_id | bigint | 租户 id | +| tenant_channel_id | bigint | 租户平台实例 id | +| binding_id | bigint | 绑定 id,用户/项目可为空 | +| resource_scope | varchar(32) | TENANT、PROJECT、DEVICE、BINDING | +| resource_type | varchar(64) | USER、PROJECT、POINT、DEVICE、SENSOR、MODULE、VIRTUAL、CUSTOM | +| local_key | varchar(128) | 本地唯一键,设备级使用 serialNumber | +| external_id | varchar(128) | 外部资源 id | +| external_code | varchar(128) | 外部资源编码 | +| external_name | varchar(128) | 外部资源名称 | +| resource_payload | json | 外部资源详情 | +| status | tinyint | 0失效,1有效 | +| create_time | datetime | 创建时间 | +| update_time | datetime | 更新时间 | + +建议索引: + +```text +uk_resource(tenant_channel_id, binding_id, resource_type, local_key) +idx_resource_tenant_type(tenant_id, tenant_channel_id, resource_type) +idx_external_id(tenant_channel_id, resource_type, external_id) +``` + +### 9.9 iot_platform_binding + +设备绑定,租户级配置。 + +| 字段 | 类型 | 说明 | +| -------------------- | ------------ | ----------------------------------------------- | +| id | bigint | 主键 | +| tenant_id | bigint | 租户 id | +| tenant_channel_id | bigint | 租户平台实例 id | +| flow_id | bigint | 平台流程 id | +| product_id | bigint | 产品 id | +| product_name | varchar(128) | 产品名称冗余 | +| device_id | bigint | FastBee 设备 id | +| device_name | varchar(128) | FastBee 设备名称 | +| serial_number | varchar(64) | FastBee 设备编号 | +| external_user_id | varchar(128) | 绑定级第三方用户 id,可为空 | +| external_project_id | varchar(128) | 绑定级第三方项目 id,可为空 | +| external_device_type | varchar(32) | 第三方设备类型,例如 DCP RAIN、INCLINO | +| binding_name | varchar(128) | 绑定名称 | +| exchange_mode | varchar(32) | OUTBOUND_PUSH、INBOUND_PULL、INBOUND_RECEIVE、BOTH | +| config_json | json | 绑定级配置 | +| last_success_time | datetime | 最近成功交换时间 | +| last_success_value | varchar(128) | 最近成功值,雨量递增等场景使用 | +| status | tinyint | 0停用,1启用 | +| del_flag | char(1) | 0正常,2删除 | +| create_time | datetime | 创建时间 | +| update_time | datetime | 更新时间 | + +建议索引: + +```text +idx_binding_device(tenant_id, product_id, serial_number) +idx_binding_channel_flow(tenant_channel_id, flow_id) +uk_binding(tenant_channel_id, flow_id, serial_number, del_flag) +``` + +### 9.10 iot_platform_field_mapping + +字段映射,按绑定维护。 + +| 字段 | 类型 | 说明 | +| ---------------- | ------------ | --------------------------------------------------------------------- | +| id | bigint | 主键 | +| tenant_id | bigint | 租户 id | +| binding_id | bigint | 绑定 id | +| direction | varchar(32) | OUTBOUND、INBOUND | +| target_path | varchar(256) | 目标字段路径 | +| target_name | varchar(128) | 目标字段名称 | +| source_type | varchar(32) | THING_MODEL、SYSTEM、CONSTANT、JSONPATH、EXPRESSION、RESOURCE | +| source_key | varchar(256) | 物模型 identifier、系统字段、表达式等 | +| source_name | varchar(128) | 来源字段名称 | +| transform_type | varchar(32) | DIRECT、DATE_FORMAT、UNIT_CONVERT、ENUM、NUMBER_SCALE、TEMPLATE、EXPRESSION | +| transform_config | json | 转换配置 | +| default_value | varchar(512) | 默认值 | +| required | tinyint | 是否必填 | +| sort_order | int | 排序 | +| status | tinyint | 0停用,1启用 | + +建议索引: + +```text +idx_mapping_binding(binding_id, direction) +``` + +### 9.11 iot_platform_exchange_task + +数据交换任务,租户级。用于定时推送、主动拉取、资源同步。 + +| 字段 | 类型 | 说明 | +| ----------------- | ------------ | ----------------------------------------- | +| id | bigint | 主键 | +| tenant_id | bigint | 租户 id | +| tenant_channel_id | bigint | 租户平台实例 id | +| flow_id | bigint | 流程 id | +| task_name | varchar(128) | 任务名称 | +| task_type | varchar(32) | OUTBOUND_PUSH、INBOUND_PULL、RESOURCE_SYNC | +| cron_expression | varchar(128) | Cron 表达式 | +| window_type | varchar(32) | LAST_SUCCESS、RECENT_MINUTES、CUSTOM、CURSOR | +| window_config | json | 时间窗口配置 | +| cursor_json | json | 拉取游标、分页位置、最近外部 id 等 | +| sink_type | varchar(32) | INBOUND 数据落点类型 | +| batch_size | int | 批次大小 | +| max_retry | int | 最大重试次数 | +| status | tinyint | 0停用,1启用 | +| last_run_time | datetime | 最近执行时间 | +| next_run_time | datetime | 下次执行时间 | +| create_time | datetime | 创建时间 | +| update_time | datetime | 更新时间 | + +### 9.12 iot_platform_task_binding + +交换任务与设备绑定关系。 + +| 字段 | 类型 | 说明 | +| ---------- | ------- | ------- | +| id | bigint | 主键 | +| task_id | bigint | 任务 id | +| binding_id | bigint | 绑定 id | +| status | tinyint | 0停用,1启用 | + +建议索引: + +```text +uk_task_binding(task_id, binding_id) +``` + +### 9.13 iot_platform_exchange_record + +数据交换记录。OUTBOUND 时是 outbox;INBOUND 时是 inbox/消费审计。 + +| 字段 | 类型 | 说明 | +| ------------------- | ------------- | --------------------------------------------------------------------------------- | +| id | bigint | 主键 | +| tenant_id | bigint | 租户 id | +| tenant_channel_id | bigint | 租户平台实例 id | +| flow_id | bigint | 流程 id | +| binding_id | bigint | 绑定 id,可为空 | +| task_id | bigint | 任务 id,接收入口可为空 | +| direction | varchar(32) | OUTBOUND_PUSH、INBOUND_PULL、INBOUND_RECEIVE | +| source_type | varchar(32) | DEVICE_PROPERTY、DEVICE_EVENT、ALERT_EVENT、RESOURCE、RAW 等 | +| sink_type | varchar(32) | DEVICE_PROPERTY、DEVICE_EVENT、ALERT_EVENT、RESOURCE_MAPPING、BUSINESS_TABLE、RAW_ONLY | +| source_device_id | bigint | FastBee 设备 id,可为空 | +| serial_number | varchar(64) | FastBee 设备编号,可为空 | +| external_message_id | varchar(128) | 第三方消息 id,可为空 | +| source_ref_id | varchar(128) | 本地来源或外部来源 id | +| source_time | datetime | 数据时间 | +| time_window_start | datetime | 批量窗口开始 | +| time_window_end | datetime | 批量窗口结束 | +| raw_payload | json | 原始输入或原始响应 | +| standard_payload | json | 标准数据交换消息 | +| request_payload | json | 最终请求 payload,OUTBOUND 使用 | +| response_payload | json | 响应或处理结果 | +| payload_hash | varchar(64) | payload 哈希 | +| idempotency_key | varchar(128) | 幂等 key | +| status | varchar(32) | PENDING、PROCESSING、SUCCESS、FAILED、RETRY_WAIT、SKIPPED、CANCELLED | +| retry_count | int | 已重试次数 | +| max_retry | int | 最大重试次数 | +| next_retry_time | datetime | 下次重试时间 | +| last_error | varchar(1000) | 最近错误 | +| locked_by | varchar(64) | worker 锁 | +| locked_time | datetime | 锁定时间 | +| success_time | datetime | 成功时间 | +| create_time | datetime | 创建时间 | +| update_time | datetime | 更新时间 | + +建议索引: + +```text +uk_idempotency(idempotency_key) +idx_record_poll(status, next_retry_time) +idx_record_direction(tenant_channel_id, direction, source_time) +idx_record_device(tenant_channel_id, serial_number, source_time) +idx_record_binding(binding_id, source_time) +``` + +### 9.14 iot_platform_exchange_attempt + +数据交换尝试日志。 + +| 字段 | 类型 | 说明 | +| --------------- | ------------- | ------------------------------------------ | +| id | bigint | 主键 | +| tenant_id | bigint | 租户 id | +| record_id | bigint | 交换记录 id | +| attempt_no | int | 第几次尝试 | +| direction | varchar(32) | OUTBOUND_PUSH、INBOUND_PULL、INBOUND_RECEIVE | +| request_url | varchar(1024) | 请求地址或接收入口 | +| request_headers | json | 请求头,敏感信息脱敏 | +| request_payload | json | 请求体 | +| response_status | int | HTTP 状态码或处理状态码 | +| response_body | mediumtext | 响应内容 | +| success | tinyint | 是否成功 | +| error_msg | varchar(1000) | 错误信息 | +| duration_ms | int | 耗时 | +| create_time | datetime | 创建时间 | + +建议索引: + +```text +idx_attempt_record(record_id, attempt_no) +idx_attempt_tenant(tenant_id, create_time) +``` + +## 10. 技术实现建议 + +### 10.1 模块与包 + +后端先放在现有 IoT 服务和 open-api 中: + +```text +fastbee-open-api/src/main/java/com/fastbee/controller/platform +fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform +fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/platform +``` + +前端目录: + +```text +yack-web/src/views/iot/platformConnect +yack-web/src/api/iot/platformConnect.js +``` + +权限前缀建议使用 `iot:platformConnect`,避免和现有第三方登录 `iot:platform` 混淆。 + +### 10.2 核心接口 + +```java +public interface PlatformAdapter { + String adapterCode(); + void beforeExecute(PlatformContext context); + void ensureResources(PlatformContext context); + Object transform(PlatformContext context); + ExchangeResult execute(PlatformContext context, Object payload); +} +``` + +```java +public interface PlatformFlowExecutor { + ExchangeResult execute(Long tenantChannelId, Long flowId, PlatformExchangeMessage message, Long bindingId); +} +``` + +```java +public interface PlatformExchangeRecordService { + void createOutboundRecords(PlatformExchangeMessage message); + void createPullRecords(Long taskId, Date beginTime, Date endTime); + void createReceiveRecord(Long tenantChannelId, Long flowId, Object rawPayload); + void retry(Long recordId, boolean force); + void retryBatch(List recordIds, boolean force); +} +``` + +### 10.3 OUTBOUND_PUSH 接入点 + +设备上报原流程保持不变,在设备数据入库或规则处理附近增加平台交换事件: + +```text +设备数据解析 + -> 原有设备日志/历史数据处理 + -> 构造 PlatformExchangeMessage(direction=OUTBOUND_PUSH) + -> PlatformExchangeRecordService.createOutboundRecords(message) + -> 匹配 tenant_id + device_id 的启用绑定 + -> 生成 PENDING 交换记录 + -> worker 异步执行流程 +``` + +### 10.4 INBOUND_PULL 调度 + +优先复用 Quartz/sys_job: + +```text +Quartz 触发 + -> 读取 iot_platform_exchange_task + -> direction=INBOUND_PULL 或 RESOURCE_SYNC + -> 根据 window_type/cursor_json 计算拉取范围 + -> 调第三方查询接口 + -> 分页/游标处理 + -> 生成或直接处理 exchange_record + -> 写入 sink_type 对应业务落点 + -> 更新 cursor_json +``` + +### 10.5 INBOUND_RECEIVE 接收入口 + +建议新增统一接收入口,不把旧 `/bridge/**` 作为平台对接主入口。 + +```text +POST /platform/receive/{tenantChannelCode}/{flowCode} +``` + +处理流程: + +```text +解析 tenantChannelCode + flowCode + -> 加载租户平台实例和 RECEIVE 流程 + -> 校验 token/签名/IP 白名单 + -> 提取幂等 key + -> 创建 INBOUND_RECEIVE exchange_record + -> 同步处理或交给 worker 异步处理 + -> 返回接收结果 +``` + +### 10.6 worker 处理 + +worker 轮询: + +```text +status in (PENDING, RETRY_WAIT) +and next_retry_time <= now +``` + +抢占记录时使用数据库乐观更新,避免多实例重复消费: + +```sql +UPDATE iot_platform_exchange_record +SET status = 'PROCESSING', locked_by = ?, locked_time = NOW() +WHERE id = ? + AND status IN ('PENDING', 'RETRY_WAIT') +``` + +处理结果:成功写 `SUCCESS`,可重试失败写 `RETRY_WAIT`,超过次数写 `FAILED`,业务跳过写 `SKIPPED`。 + +## 11. 幂等、游标与重试 + +### 11.1 OUTBOUND_PUSH 幂等 + +实时数据优先使用原始日志 id: + +```text +tenantChannelId + flowId + bindingId + OUTBOUND_PUSH + sourceRefId + payloadHash +``` + +没有 `sourceRefId` 时使用: + +```text +tenantChannelId + flowId + bindingId + OUTBOUND_PUSH + sourceTime + payloadHash +``` + +### 11.2 INBOUND_PULL 幂等 + +优先使用第三方数据 id: + +```text +tenantChannelId + flowId + INBOUND_PULL + externalMessageId +``` + +没有第三方 id 时使用: + +```text +tenantChannelId + flowId + INBOUND_PULL + sourceTime + payloadHash +``` + +拉取任务必须维护 `cursor_json` 或时间窗口,避免重复和无界拉取。 + +### 11.3 INBOUND_RECEIVE 幂等 + +优先从第三方请求中提取 messageId、eventId、deviceCode + time 等业务幂等键: + +```text +tenantChannelId + flowId + INBOUND_RECEIVE + externalMessageId +``` + +如果第三方没有幂等字段,可使用请求体 hash,但需要注意同一业务重复发送和相同 payload 的边界。 + +### 11.4 外部资源幂等 + +资源创建前先查 `iot_platform_resource_mapping`。 + +DCP 设备级资源必须按 `tenantChannelId + bindingId + serialNumber + resourceType` 唯一。 + +```text +查 mapping + -> 存在则复用 + -> 不存在则加 Redis 锁 + -> 再查一次 + -> 调第三方创建接口或适配器方法 + -> 保存 mapping +``` + +DCP 锁 key: + +```text +platform:dcp:resource:{tenantId}:{tenantChannelId}:{serialNumber} +``` + +### 11.5 重试和重放 + +- OUTBOUND_PUSH:失败可自动重试和手动重推。 +- INBOUND_PULL:失败可重试当前窗口;成功后再推进游标。 +- INBOUND_RECEIVE:同步处理失败由第三方重试;异步处理失败可由 FastBee 内部重试。 +- 成功记录强制重放:新建 MANUAL/REPLAY 记录,幂等 key 追加手动批次号。 + +## 12. 权限建议 + +权限前缀:`iot:platformConnect`。 + +```text +iot:platformConnect:template:list/query/add/edit/remove +iot:platformConnect:api:list/add/edit/remove +iot:platformConnect:flow:list/add/edit/remove +iot:platformConnect:tenantChannel:list/query/add/edit/remove +iot:platformConnect:resource:list/query/sync +iot:platformConnect:binding:list/query/add/edit/unbind/initResource +iot:platformConnect:task:list/add/edit/remove/run +iot:platformConnect:record:list/query/retry/replay +``` + +平台模板、接口、流程建议仅平台管理员可维护;租户管理员主要维护租户接入、资源映射、设备绑定、交换任务和交换记录。 + +## 13. 实施计划 + +### 阶段一:平台模板与租户接入骨架 + +- 新增平台模板、接口、字段、流程、流程步骤表。 +- 新增租户平台实例表。 +- 建立 DCP 平台模板和 DCP OUTBOUND_PUSH/RESOURCE_SYNC 基础流程。 +- 实现租户级授权配置、接收安全配置、默认用户/项目映射。 + +### 阶段二:资源映射、设备绑定与字段映射 + +- 新增资源映射、设备绑定、字段映射页面和接口。 +- 实现 DCP 一设备一测点、一设备一 DCP 设备资源建模。 +- 实现 DCP token、userId、projectId、pointId、dcpDeviceId、sensor/module/virtual 资源初始化。 + +### 阶段三:OUTBOUND_PUSH 数据交换闭环 + +- 新增 exchange_task、exchange_record、exchange_attempt。 +- 实现设备上报生成 OUTBOUND_PUSH 交换记录。 +- 实现 worker、重试、幂等和手动重放。 +- 支持雨量计实时推送闭环。 + +### 阶段四:INBOUND_PULL 和 RESOURCE_SYNC + +- 实现平台交换任务的定时拉取。 +- 支持分页、游标、时间窗口。 +- 支持用户、项目、设备、测点等资源同步。 +- 支持拉取数据写入明确 sink_type。 + +### 阶段五:INBOUND_RECEIVE 接收能力 + +- 新增统一接收入口。 +- 支持 token、签名、IP 白名单、幂等 key 提取。 +- 支持同步处理和异步 exchange_record 处理。 +- 支持接收数据写入设备属性、事件、告警、资源映射或 RAW_ONLY。 + +### 阶段六:通用平台增强 + +- 通用 HTTP 平台模板。 +- 接口调试。 +- 模板导入导出。 +- MQTT/DB 适配器按需扩展。 +- 日志归档和敏感信息脱敏策略。 + +## 14. 风险与约束 + +- 平台模板是共享基础能力,租户授权、项目、资源、设备绑定、交换记录必须按 `tenant_id` 隔离。 +- INBOUND_PULL 和 INBOUND_RECEIVE 都是外部输入,必须做鉴权、校验、幂等和明确落点。 +- DCP 一测点一设备是核心建模约束,不能把测点设计成产品级或平台级共享资源。 +- OUTBOUND_PUSH 必须异步,不影响设备上报吞吐。 +- token、密码、请求头、签名密钥等敏感信息需要加密保存和脱敏展示。 +- 交换日志增长较快,需要后续增加归档或清理策略。 +- 字段映射表达式必须受限,不能执行任意代码。 +- INBOUND 写业务表必须通过白名单 sink adapter,不能开放任意 SQL。 +- 定时拉取和补推要限制窗口、批次大小和并发数,避免压垮第三方平台或本系统。 +- 雨量累计值不递增时应跳过并记录原因,不应无限重试。 + +## 15. 结论 + +平台对接应定位为可复用的数据交换/iPaaS Connector 能力,而不是单个平台或单一“推送”能力。 + +推荐模型: + +```text +平台模板(共享) + -> 接口配置(共享) + -> 流程编排(共享) + -> 适配器(共享) + -> 租户平台实例(租户级) + -> 用户/项目/资源映射(租户级) + -> 设备绑定和字段映射(租户级) + -> 数据交换任务(租户级) + -> 数据交换记录和尝试日志(租户级) +``` + +其中数据方向明确分为: + +```text +OUTBOUND_PUSH:FastBee 作为 producer 推送数据 +INBOUND_PULL:FastBee 主动拉取,但作为 consumer 消费第三方数据 +INBOUND_RECEIVE:FastBee 被动接收,并作为 consumer 消费第三方数据 +``` + +DCP 是第一批落地模板和适配器,必须保留“一测点一设备”的资源约束;其他租户复用 DCP 平台模板后,只需配置自己的 DCP 账号、用户、项目、设备绑定、字段映射和数据交换策略。 diff --git a/docs/knowledge-base/06_1-platform-intergration-design.md b/docs/knowledge-base/06_1-platform-intergration-design.md new file mode 100644 index 00000000..7dbe09c2 --- /dev/null +++ b/docs/knowledge-base/06_1-platform-intergration-design.md @@ -0,0 +1,1013 @@ +# 平台对接需求与技术设计 + +## 1. 背景 + +需求目录 `docs/platform/` 中目前包含 DCP 雨量计、倾角、静力水准、轴力计接口文档。DCP 是当前优先落地的平台,但平台对接能力不能只服务一个平台、一个租户。平台一旦完成接入,后续其他租户应复用同一套平台定义、接口定义、流程编排、适配器和推送执行能力;租户只需要维护自己的第三方账号/用户、项目、设备、字段映射和推送任务。 + +当前 FastBee 已具备以下基础能力: + +- 设备、产品、物模型:设备通过 `iot_device` 管理,物模型通过产品和设备编号查询。 +- 设备上报处理:设备数据解析后进入原有入库和规则处理链路。 +- 历史数据查询:数据中心接口支持按设备、时间窗口和物模型标识符查询历史数据。 +- 数据桥接:现有 `bridge`、`iot_script`、`iot_script_bridge` 支持 HTTP、MQTT、数据库桥接。 +- 定时任务:系统已有 Quartz 任务体系,可复用为定时推送和补偿任务触发器。 +- 多租户:IoT 主要业务表已有 `tenant_id`、`tenant_name` 约定,平台对接也应按租户隔离业务映射和推送数据。 + +因此推荐建设系统级 **平台对接 / iPaaS Connector** 能力:平台基础接入可复用,租户业务映射可隔离,DCP 等复杂平台通过专用适配器补充规则。 + +## 2. 核心目标 + +平台对接需要同时满足“平台共用”和“租户隔离”: + +- 平台接入人员只需维护一次第三方平台模板:认证方式、接口、字段、流程、适配器。 +- 多个租户可启用同一个平台模板,不重复配置接口和流程。 +- 每个租户维护自己的第三方账号/用户授权、项目映射、设备绑定、字段映射和推送任务。 +- 设备上报后按租户和设备绑定匹配推送目标,生成推送记录并异步执行。 +- 支持定时推送和按时间范围补推,复用现有历史数据查询能力。 +- 支持幂等、失败重试、手动重推、推送日志和审计。 +- DCP 这类平台保留专用适配器,处理 token、项目、测点、设备、传感器、雨量递增等平台规则。 +- DCP 必须遵守 **一个测点对应一个 FastBee 设备**,不能将测点设计为平台级或产品级共享资源。 + +## 3. 分层模型 + +平台对接按三层拆分: + +```text +平台模板层(全局/系统级,可复用) + -> 平台定义、接口定义、字段定义、流程编排、适配器 + +租户接入层(租户级,隔离) + -> 租户平台实例、第三方账号/用户、项目映射、资源映射、设备绑定、字段映射、推送任务 + +执行审计层(租户级,隔离) + -> 推送记录 outbox、推送尝试日志、重试、重推、补偿 +``` + +### 3.1 平台模板层 + +平台模板层由平台管理员维护,配置一次、多租户复用。 + +包含: + +- 第三方平台类型,例如 DCP、通用 HTTP、通用 MQTT、通用数据库。 +- 认证方式,例如无认证、固定 Token、Basic、Bearer、OAuth2、自定义 Token 接口。 +- 接口定义,例如获取用户、获取项目、创建测点、创建设备、绑定传感器、推送数据。 +- 接口字段定义,用于页面展示和字段映射。 +- 流程编排,例如“获取用户 -> 获取项目 -> 确保测点 -> 确保设备 -> 推送数据”。 +- 适配器编码,例如 `genericHttp`、`dcp`。 + +### 3.2 租户接入层 + +租户接入层由租户管理员或实施人员维护。 + +包含: + +- 租户启用哪个平台模板。 +- 租户自己的第三方账号、token、clientId/clientSecret 等授权配置。 +- 租户第三方用户和项目映射。 +- FastBee 设备与第三方资源的绑定。 +- FastBee 物模型字段与第三方接口字段的映射。 +- 实时/定时推送任务和补推策略。 + +同一个平台模板下,不同租户的数据必须隔离:授权配置、项目、用户、资源映射、设备绑定、推送记录都必须带 `tenant_id`。 + +### 3.3 执行审计层 + +执行审计层面向运行时。 + +```text +设备上报 / Quartz 定时任务 / 手动补推 + -> 标准上报快照 PlatformReport + -> 按 tenant_id + device_id 匹配启用绑定 + -> 加载租户平台实例和流程 + -> 字段映射与数据转换 + -> 资源检查/懒加载 + -> 生成推送记录 iot_platform_push_record + -> worker 异步执行 + -> 写推送尝试日志 iot_platform_push_attempt +``` + +## 4. 与现有系统功能的关系 + +| 现有能力 | 复用方式 | +| --- | --- | +| 设备管理 `iot_device` | 设备绑定引用 `device_id`、`serial_number`、`product_id`、`tenant_id`。 | +| 产品物模型 | 绑定页面通过现有物模型接口展示字段,字段映射使用物模型 `identifier`。 | +| 设备实时上报链路 | 在原有数据处理完成后发布平台推送事件,不改变协议解析逻辑。 | +| 数据中心历史查询 | 定时推送和手动补推按设备、时间窗口、identifier 取历史数据。 | +| 规则引擎和数据桥接 | 保留给轻量 HTTP/MQTT/DB 转发;平台对接面向多步骤、多租户、可重推场景。 | +| Quartz 定时任务 | 平台推送任务可复用 sys_job/Quartz 触发。 | +| Redis/Redisson | 缓存 token,资源创建时做并发锁。 | +| 若依权限体系 | 继续使用菜单、角色、`@PreAuthorize` 权限控制。 | + +现有数据桥接仍保留。它适合“固定 URL + 固定模板 + 脚本触发”的轻量转发;平台对接用于第三方平台长期集成,需要多租户账号、项目、设备映射、流程编排、资源幂等、失败重试和审计。 + +## 5. 前端功能设计 + +一级菜单:**平台对接**。 + +建议菜单: + +```text +平台对接 + 平台模板 + 接口配置 + 流程编排 + 租户接入 + 设备绑定 + 推送任务 + 推送记录 +``` + +### 5.1 平台模板 + +平台模板是系统级可复用配置。 + +主要字段: + +| 字段 | 说明 | +| --- | --- | +| 平台名称 | 用户可读名称,例如 DCP。 | +| 平台编码 | 全局唯一,例如 `DCP`。 | +| 平台类型 | DCP、GENERIC_HTTP、GENERIC_MQTT、GENERIC_DB。 | +| 适配器编码 | dcp、genericHttp 等。 | +| 基础地址模板 | 可作为租户实例默认 baseUrl。 | +| 认证方式 | OAuth2、自定义 Token、固定 Token 等。 | +| 默认请求配置 | 超时、重试、Content-Type 等。 | +| 启用状态 | 启用、停用。 | + +### 5.2 接口配置 + +接口配置属于平台模板,可被所有租户复用。 + +接口类型: + +| 类型 | 说明 | +| --- | --- | +| TOKEN | 获取或刷新 token。 | +| QUERY | 查询第三方基础数据,如用户、项目。 | +| RESOURCE_CREATE | 创建外部资源,如测点、设备、传感器。 | +| RESOURCE_BIND | 绑定外部资源。 | +| PUSH | 推送设备数据。 | + +配置项: + +- 请求方法:GET、POST、PUT、DELETE。 +- URL 模板:支持 `${变量}`。 +- Header、Query、Body 模板。 +- Content-Type。 +- 成功条件:HTTP 状态码、响应 JSONPath。 +- 响应变量提取:例如 `$.id -> pointId`。 +- 超时和重试策略。 + +### 5.3 接口字段 + +接口字段用于设备绑定页面右侧展示,帮助租户配置字段映射。 + +| 字段 | 说明 | +| --- | --- | +| 接口 | 所属接口。 | +| 方向 | REQUEST、RESPONSE。 | +| 字段路径 | JSONPath 或表单字段名。 | +| 字段名称 | 展示名称。 | +| 数据类型 | string、number、boolean、array、object。 | +| 是否必填 | 是否必填。 | +| 默认值 | 默认值。 | +| 示例值 | 示例值。 | + +### 5.4 流程编排 + +流程属于平台模板,用于将多个接口和适配器方法组成可复用业务流程。 + +示例:DCP 数据推送流程。 + +```text +获取/刷新 token + -> 获取用户 userId + -> 获取/选择项目 projectId + -> 确保测点 pointId + -> 确保设备 dcpDeviceId + -> 确保传感器/附属资源 + -> 构造 payload + -> 推送数据 +``` + +流程步骤类型: + +| 步骤类型 | 说明 | +| --- | --- | +| API | 调用平台接口配置。 | +| ADAPTER | 调用专用适配器方法。 | +| CONDITION | 条件判断。 | +| TRANSFORM | 数据转换。 | + +流程模板可以被多个租户复用;步骤执行时使用当前租户的授权配置、项目映射、设备绑定和资源映射。 + +### 5.5 租户接入 + +租户接入表示某个租户启用某个平台模板。 + +主要字段: + +| 字段 | 说明 | +| --- | --- | +| 租户 | 当前租户。 | +| 平台模板 | 选择 DCP 等平台模板。 | +| 实例名称 | 租户内展示名。 | +| baseUrl | 可覆盖模板默认地址。 | +| 授权配置 | 租户自己的账号、clientId、secret、token 等。 | +| 默认第三方用户 | DCP userId 或其他平台用户。 | +| 默认项目 | DCP projectId 或其他平台项目。 | +| 启用状态 | 启用、停用。 | + +租户接入配置完成后,该租户才能建立设备绑定、推送任务和资源映射。 + +### 5.6 设备绑定 + +设备绑定是租户级配置。 + +绑定内容: + +```text +租户平台实例 + FastBee 设备 + 平台流程 + 第三方用户/项目 + 字段映射 + 推送模式 +``` + +绑定页面左侧使用现有 FastBee 产品、设备、物模型;右侧使用平台模板的接口字段和租户的第三方用户/项目/资源。 + +绑定支持:新增、修改、启用、停用、解绑、初始化资源、查看资源映射、查看推送记录。 + +DCP 绑定规则: + +```text +FastBee 一个设备 = DCP 一个测点 = DCP 一个设备资源 +``` + +DCP 的 `pointId`、`dcpDeviceId`、sensor/module/virtual 等必须按租户设备绑定保存,不能平台级共享。 + +### 5.7 字段映射 + +字段映射按设备绑定维护,也可以支持从平台模板复制默认映射。 + +映射来源: + +| 来源类型 | 说明 | +| --- | --- | +| THING_MODEL | 物模型字段,例如 rainfall。 | +| SYSTEM | 系统字段,例如 serialNumber、deviceName、reportTime。 | +| CONSTANT | 常量。 | +| RESOURCE | 外部资源映射,例如 pointId、dcpDeviceId。 | +| JSONPATH | 从 rawPayload 中取值。 | +| EXPRESSION | 受限表达式,后续增强。 | + +第一阶段转换能力建议支持:直接映射、时间格式化、单位换算、数值倍率、枚举转换、默认值。表达式必须受限,不能执行任意代码。 + +### 5.8 推送任务与推送记录 + +推送任务是租户级配置,支持: + +- 实时推送:设备上报后生成推送记录。 +- 定时推送:按 Quartz/sys_job 触发,从历史数据按窗口生成推送记录。 +- 手动补推:按设备和时间范围生成推送记录。 + +推送记录展示业务层面的推送状态: + +- 按租户、平台、流程、设备、状态、时间筛选。 +- 查看最终 payload、幂等 key、请求响应、错误信息。 +- 失败重推、批量重推、成功记录强制重推。 + +## 6. 标准上报快照 + +字段映射不要直接依赖原始 MQTT/HTTP payload。系统应先把设备数据整理成统一快照。 + +```json +{ + "tenantId": 1, + "productId": 10, + "deviceId": 1001, + "serialNumber": "ABC001", + "deviceName": "雨量计-001", + "reportTime": "2026-05-27 10:00:00", + "sourceRefId": "device_log_id_or_history_id", + "values": { + "rainfall": { + "value": 12.3, + "unit": "mm", + "ts": "2026-05-27 10:00:00" + }, + "temperature": { + "value": 22.1, + "unit": "℃", + "ts": "2026-05-27 10:00:00" + } + }, + "rawPayload": {} +} +``` + +`values` 的 key 使用物模型 `identifier`,便于复用现有物模型和历史数据查询能力。 + +## 7. DCP 适配设计 + +DCP 通过平台模板 + 流程编排 + 专用适配器落地。 + +### 7.1 DCP 平台模板 + +DCP 平台模板包含: + +- token 接口:`POST /connect/token`。 +- 用户接口:获取 userId。 +- 项目接口:按 userId 获取项目。 +- 资源接口:创建测点、创建设备、创建/绑定传感器等。 +- 推送接口:雨量、倾角、静力水准、轴力计数据推送。 +- DCP 专用适配器:处理 token 缓存、资源幂等、payload 构造、特殊规则。 + +这些配置由系统维护一次,所有租户复用。 + +### 7.2 DCP 租户映射 + +每个租户需要维护自己的 DCP 映射: + +```text +租户 + -> DCP 租户平台实例 + -> DCP 账号/用户 userId + -> DCP 项目 projectId + -> FastBee 设备绑定 DCP 测点 pointId + -> FastBee 设备绑定 DCP 设备 dcpDeviceId + -> 设备附属 sensor/module/virtual +``` + +同一个 DCP 平台模板下,不同租户可以使用不同 DCP 账号、用户和项目。资源映射必须按 `tenant_id` 隔离。 + +### 7.3 DCP 一测点一设备约束 + +DCP 资源映射必须按“一测点一设备”建模: + +```text +FastBee Device(serialNumber) + -> DCP Project(projectId) + -> DCP Point(pointId) + -> DCP Device(dcpDeviceId) + -> DCP Sensor/Module/Virtual(按设备类型可选) +``` + +规则: + +- `PROJECT` 可以是租户平台实例级默认资源,也可以绑定级覆盖。 +- `USER`、`PROJECT` 是租户级或绑定级映射,不是全局共享。 +- `POINT` 必须是设备级资源,一个 FastBee 设备对应一个 DCP 测点。 +- `DEVICE` 必须是设备级资源,一个 FastBee 设备对应一个 DCP 设备资源。 +- `SENSOR`、`MODULE`、`VIRTUAL` 是设备附属资源,按绑定保存。 +- 静力水准 payload 中可包含多条 `sensorData`,但仍归属同一个 FastBee 设备对应的 DCP 测点和 DCP 设备下。 + +### 7.4 DCP 运行流程 + +```text +设备上报或定时任务 + -> 匹配租户 DCP 设备绑定 + -> 加载租户 DCP 授权配置 + -> 获取/刷新 token + -> 加载 userId/projectId 映射 + -> 加载 pointId/dcpDeviceId/sensor 映射 + -> 资源缺失时按流程懒加载或手动初始化 + -> 构造对应设备类型 payload + -> 调用 DCP 数据接口 + -> 写推送记录和尝试日志 +``` + +资源创建必须幂等。并发上报同一设备时,使用 Redis/Redisson 锁: + +```text +platform:dcp:resource:{tenantId}:{tenantChannelId}:{serialNumber} +``` + +### 7.5 DCP 特殊规则 + +- 雨量 `value` 为累计降雨量,DCP 要求新值大于历史值。系统使用绑定的 `last_success_value` 或资源扩展字段记录最近成功值;不满足时记录为 `SKIPPED`,不反复重试。 +- 倾角需要同一时间的 x、y、z 和温度,必须按一次设备快照组包,不能拆成多个单字段推送。 +- 静力水准的 `sensorData` 可包含多条传感器数据,但这些数据属于同一个 FastBee 设备对应的 DCP 设备资源。 +- 轴力计按独立设备类型处理,字段映射和 payload 由 `DcpPlatformAdapter` 中对应 builder 负责。 +- DCP token 缓存在 Redis,key 建议:`platform:dcp:token:{tenantId}:{tenantChannelId}`。 +- 401 或 token 过期时刷新 token 后重试一次,再失败则记录错误。 + +## 8. 表设计 + +表名前缀使用 `iot_platform_`,避免和现有 `iot_social_platform` 混淆。 + +### 8.1 表清单 + +| 表名 | 层级 | 说明 | +| --- | --- | --- | +| `iot_platform_template` | 平台模板层 | 平台模板,全局可复用。 | +| `iot_platform_api` | 平台模板层 | 第三方接口定义。 | +| `iot_platform_api_field` | 平台模板层 | 接口字段定义。 | +| `iot_platform_flow` | 平台模板层 | 流程模板。 | +| `iot_platform_flow_step` | 平台模板层 | 流程步骤。 | +| `iot_platform_tenant_channel` | 租户接入层 | 租户启用的平台实例和授权配置。 | +| `iot_platform_resource_mapping` | 租户接入层 | 用户、项目、测点、设备、传感器等资源映射。 | +| `iot_platform_binding` | 租户接入层 | FastBee 设备与平台流程/项目/资源绑定。 | +| `iot_platform_field_mapping` | 租户接入层 | 绑定级字段映射。 | +| `iot_platform_push_task` | 租户接入层 | 定时推送任务。 | +| `iot_platform_task_binding` | 租户接入层 | 推送任务与绑定关系。 | +| `iot_platform_push_record` | 执行审计层 | 推送记录,兼 outbox。 | +| `iot_platform_push_attempt` | 执行审计层 | 每次推送尝试日志。 | + +### 8.2 iot_platform_template + +平台模板,全局可复用。 + +| 字段 | 类型 | 说明 | +| --- | --- | --- | +| id | bigint | 主键 | +| platform_code | varchar(64) | 平台编码,全局唯一 | +| platform_name | varchar(128) | 平台名称 | +| platform_type | varchar(32) | DCP、GENERIC_HTTP、GENERIC_MQTT、GENERIC_DB | +| adapter_code | varchar(64) | dcp、genericHttp 等 | +| base_url_template | varchar(512) | 默认基础地址模板 | +| auth_type | varchar(32) | 认证类型 | +| auth_schema | json | 授权配置字段定义,用于租户填写 | +| request_config | json | 默认超时、重试、TLS 等 | +| status | tinyint | 0停用,1启用 | +| create_time | datetime | 创建时间 | +| update_time | datetime | 更新时间 | +| remark | varchar(500) | 备注 | + +建议索引: + +```text +uk_platform_code(platform_code) +idx_platform_type(platform_type) +``` + +### 8.3 iot_platform_api + +第三方接口定义,属于平台模板。 + +| 字段 | 类型 | 说明 | +| --- | --- | --- | +| id | bigint | 主键 | +| template_id | bigint | 平台模板 id | +| api_code | varchar(64) | 接口编码 | +| api_name | varchar(128) | 接口名称 | +| api_type | varchar(32) | TOKEN、QUERY、PUSH、RESOURCE_CREATE、RESOURCE_BIND | +| method | varchar(16) | GET、POST、PUT、DELETE | +| path_template | varchar(512) | URL 路径模板 | +| content_type | varchar(64) | application/json、form-urlencoded | +| header_template | json | Header 模板 | +| query_template | json | Query 模板 | +| body_template | json | Body 模板 | +| success_condition | json | 成功条件 | +| response_mapping | json | 响应变量提取规则 | +| retry_policy | json | 接口级重试策略 | +| timeout_ms | int | 超时时间 | +| status | tinyint | 0停用,1启用 | +| create_time | datetime | 创建时间 | +| update_time | datetime | 更新时间 | + +建议索引: + +```text +uk_template_api_code(template_id, api_code) +idx_template_api_type(template_id, api_type) +``` + +### 8.4 iot_platform_api_field + +接口字段定义,用于页面展示和字段映射。 + +| 字段 | 类型 | 说明 | +| --- | --- | --- | +| id | bigint | 主键 | +| api_id | bigint | 接口 id | +| direction | varchar(16) | REQUEST、RESPONSE | +| field_path | varchar(256) | JSONPath 或表单字段名 | +| field_name | varchar(128) | 字段名称 | +| data_type | varchar(32) | string、number、boolean、array、object | +| required | tinyint | 是否必填 | +| default_value | varchar(512) | 默认值 | +| example_value | varchar(512) | 示例值 | +| description | varchar(500) | 说明 | +| sort_order | int | 排序 | + +建议索引: + +```text +idx_api_direction(api_id, direction) +``` + +### 8.5 iot_platform_flow + +流程模板,属于平台模板。 + +| 字段 | 类型 | 说明 | +| --- | --- | --- | +| id | bigint | 主键 | +| template_id | bigint | 平台模板 id | +| flow_code | varchar(64) | 流程编码 | +| flow_name | varchar(128) | 流程名称 | +| flow_type | varchar(32) | PUSH、QUERY、RESOURCE_SYNC | +| adapter_code | varchar(64) | 可覆盖平台适配器 | +| status | tinyint | 0停用,1启用 | +| create_time | datetime | 创建时间 | +| update_time | datetime | 更新时间 | +| remark | varchar(500) | 备注 | + +建议索引: + +```text +uk_template_flow_code(template_id, flow_code) +``` + +### 8.6 iot_platform_flow_step + +流程步骤。 + +| 字段 | 类型 | 说明 | +| --- | --- | --- | +| id | bigint | 主键 | +| flow_id | bigint | 流程 id | +| step_code | varchar(64) | 步骤编码 | +| step_name | varchar(128) | 步骤名称 | +| step_type | varchar(32) | API、ADAPTER、CONDITION、TRANSFORM | +| api_id | bigint | API 步骤引用接口 id | +| adapter_method | varchar(128) | 适配器方法,例如 ensurePoint | +| input_mapping | json | 步骤入参映射 | +| output_mapping | json | 步骤输出变量映射 | +| condition_expr | varchar(512) | 执行条件 | +| on_fail | varchar(32) | STOP、SKIP、RETRY | +| sort_order | int | 执行顺序 | +| status | tinyint | 0停用,1启用 | + +建议索引: + +```text +idx_flow_sort(flow_id, sort_order) +``` + +### 8.7 iot_platform_tenant_channel + +租户启用平台模板后的平台实例。授权配置、baseUrl、默认用户/项目属于租户级。 + +| 字段 | 类型 | 说明 | +| --- | --- | --- | +| id | bigint | 主键 | +| tenant_id | bigint | 租户 id | +| tenant_name | varchar(64) | 租户名称 | +| template_id | bigint | 平台模板 id | +| channel_code | varchar(64) | 租户内平台实例编码 | +| channel_name | varchar(128) | 租户内平台实例名称 | +| base_url | varchar(512) | 实际基础地址 | +| auth_config | json | 租户授权配置,敏感值加密 | +| default_user_id | varchar(128) | 默认第三方用户 id | +| default_project_id | varchar(128) | 默认第三方项目 id | +| config_json | json | 租户级扩展配置 | +| status | tinyint | 0停用,1启用 | +| del_flag | char(1) | 0正常,2删除 | +| create_by | varchar(64) | 创建者 | +| create_time | datetime | 创建时间 | +| update_by | varchar(64) | 更新者 | +| update_time | datetime | 更新时间 | +| remark | varchar(500) | 备注 | + +建议索引: + +```text +uk_tenant_channel_code(tenant_id, channel_code, del_flag) +idx_tenant_template(tenant_id, template_id) +``` + +### 8.8 iot_platform_resource_mapping + +外部资源映射,租户级隔离。DCP 的 user、project、point、device、sensor 等都保存在这里。 + +| 字段 | 类型 | 说明 | +| --- | --- | --- | +| id | bigint | 主键 | +| tenant_id | bigint | 租户 id | +| tenant_channel_id | bigint | 租户平台实例 id | +| binding_id | bigint | 绑定 id,用户/项目可为空 | +| resource_scope | varchar(32) | TENANT、PROJECT、DEVICE、BINDING | +| resource_type | varchar(64) | USER、PROJECT、POINT、DEVICE、SENSOR、MODULE、VIRTUAL | +| local_key | varchar(128) | 本地唯一键,设备级使用 serialNumber | +| external_id | varchar(128) | 外部资源 id | +| external_code | varchar(128) | 外部资源编码 | +| external_name | varchar(128) | 外部资源名称 | +| resource_payload | json | 外部资源详情 | +| status | tinyint | 0失效,1有效 | +| create_time | datetime | 创建时间 | +| update_time | datetime | 更新时间 | + +建议索引: + +```text +uk_resource(tenant_channel_id, binding_id, resource_type, local_key) +idx_resource_tenant_type(tenant_id, tenant_channel_id, resource_type) +idx_external_id(tenant_channel_id, resource_type, external_id) +``` + +### 8.9 iot_platform_binding + +设备绑定,租户级配置。 + +| 字段 | 类型 | 说明 | +| --- | --- | --- | +| id | bigint | 主键 | +| tenant_id | bigint | 租户 id | +| tenant_channel_id | bigint | 租户平台实例 id | +| flow_id | bigint | 平台流程 id | +| product_id | bigint | 产品 id | +| product_name | varchar(128) | 产品名称冗余 | +| device_id | bigint | FastBee 设备 id | +| device_name | varchar(128) | FastBee 设备名称 | +| serial_number | varchar(64) | FastBee 设备编号 | +| external_user_id | varchar(128) | 绑定级第三方用户 id,可为空 | +| external_project_id | varchar(128) | 绑定级第三方项目 id,可为空 | +| external_device_type | varchar(32) | 第三方设备类型,例如 DCP RAIN、INCLINO | +| binding_name | varchar(128) | 绑定名称 | +| push_mode | varchar(16) | REALTIME、SCHEDULED、BOTH | +| config_json | json | 绑定级配置 | +| last_success_time | datetime | 最近成功推送时间 | +| last_success_value | varchar(128) | 最近成功值,雨量递增等场景使用 | +| status | tinyint | 0停用,1启用 | +| del_flag | char(1) | 0正常,2删除 | +| create_time | datetime | 创建时间 | +| update_time | datetime | 更新时间 | + +建议索引: + +```text +idx_binding_device(tenant_id, product_id, serial_number) +idx_binding_channel_flow(tenant_channel_id, flow_id) +uk_binding(tenant_channel_id, flow_id, serial_number, del_flag) +``` + +### 8.10 iot_platform_field_mapping + +字段映射,按绑定维护。 + +| 字段 | 类型 | 说明 | +| --- | --- | --- | +| id | bigint | 主键 | +| tenant_id | bigint | 租户 id | +| binding_id | bigint | 绑定 id | +| target_path | varchar(256) | 外部目标字段路径 | +| target_name | varchar(128) | 外部字段名称 | +| source_type | varchar(32) | THING_MODEL、SYSTEM、CONSTANT、JSONPATH、EXPRESSION、RESOURCE | +| source_key | varchar(256) | 物模型 identifier、系统字段、表达式等 | +| source_name | varchar(128) | 内部字段名称 | +| transform_type | varchar(32) | DIRECT、DATE_FORMAT、UNIT_CONVERT、ENUM、NUMBER_SCALE、TEMPLATE、EXPRESSION | +| transform_config | json | 转换配置 | +| default_value | varchar(512) | 默认值 | +| required | tinyint | 是否必填 | +| sort_order | int | 排序 | +| status | tinyint | 0停用,1启用 | + +建议索引: + +```text +idx_mapping_binding(binding_id) +``` + +### 8.11 iot_platform_push_task + +定时推送任务,租户级。 + +| 字段 | 类型 | 说明 | +| --- | --- | --- | +| id | bigint | 主键 | +| tenant_id | bigint | 租户 id | +| tenant_channel_id | bigint | 租户平台实例 id | +| flow_id | bigint | 流程 id | +| task_name | varchar(128) | 任务名称 | +| cron_expression | varchar(128) | Cron 表达式 | +| window_type | varchar(32) | LAST_SUCCESS、RECENT_MINUTES、CUSTOM | +| window_config | json | 时间窗口配置 | +| batch_size | int | 批次大小 | +| max_retry | int | 最大重试次数 | +| status | tinyint | 0停用,1启用 | +| last_run_time | datetime | 最近执行时间 | +| next_run_time | datetime | 下次执行时间 | +| create_time | datetime | 创建时间 | +| update_time | datetime | 更新时间 | + +### 8.12 iot_platform_task_binding + +定时任务与设备绑定关系。 + +| 字段 | 类型 | 说明 | +| --- | --- | --- | +| id | bigint | 主键 | +| task_id | bigint | 任务 id | +| binding_id | bigint | 绑定 id | +| status | tinyint | 0停用,1启用 | + +建议索引: + +```text +uk_task_binding(task_id, binding_id) +``` + +### 8.13 iot_platform_push_record + +推送记录,租户级 outbox。 + +| 字段 | 类型 | 说明 | +| --- | --- | --- | +| id | bigint | 主键 | +| tenant_id | bigint | 租户 id | +| tenant_channel_id | bigint | 租户平台实例 id | +| flow_id | bigint | 流程 id | +| binding_id | bigint | 绑定 id | +| task_id | bigint | 定时任务 id,实时推送为空 | +| push_mode | varchar(16) | REALTIME、SCHEDULED、MANUAL | +| source_device_id | bigint | FastBee 设备 id | +| serial_number | varchar(64) | FastBee 设备编号 | +| source_ref_id | varchar(128) | 来源日志或历史数据 id | +| source_time | datetime | 数据时间 | +| time_window_start | datetime | 批量窗口开始 | +| time_window_end | datetime | 批量窗口结束 | +| request_payload | json | 待推送 payload | +| payload_hash | varchar(64) | payload 哈希 | +| idempotency_key | varchar(128) | 幂等 key | +| status | varchar(32) | PENDING、PROCESSING、SUCCESS、FAILED、RETRY_WAIT、SKIPPED、CANCELLED | +| retry_count | int | 已重试次数 | +| max_retry | int | 最大重试次数 | +| next_retry_time | datetime | 下次重试时间 | +| last_error | varchar(1000) | 最近错误 | +| locked_by | varchar(64) | worker 锁 | +| locked_time | datetime | 锁定时间 | +| success_time | datetime | 成功时间 | +| create_time | datetime | 创建时间 | +| update_time | datetime | 更新时间 | + +建议索引: + +```text +uk_idempotency(idempotency_key) +idx_record_poll(status, next_retry_time) +idx_record_device(tenant_channel_id, serial_number, source_time) +idx_record_binding(binding_id, source_time) +``` + +### 8.14 iot_platform_push_attempt + +推送尝试日志。 + +| 字段 | 类型 | 说明 | +| --- | --- | --- | +| id | bigint | 主键 | +| tenant_id | bigint | 租户 id | +| record_id | bigint | 推送记录 id | +| attempt_no | int | 第几次尝试 | +| request_url | varchar(1024) | 请求地址 | +| request_headers | json | 请求头,敏感信息脱敏 | +| request_payload | json | 请求体 | +| response_status | int | HTTP 状态码 | +| response_body | mediumtext | 响应内容 | +| success | tinyint | 是否成功 | +| error_msg | varchar(1000) | 错误信息 | +| duration_ms | int | 耗时 | +| create_time | datetime | 创建时间 | + +建议索引: + +```text +idx_attempt_record(record_id, attempt_no) +idx_attempt_tenant(tenant_id, create_time) +``` + +## 9. 技术实现建议 + +### 9.1 模块与包 + +后端先放在现有 IoT 服务和 open-api 中: + +```text +fastbee-open-api/src/main/java/com/fastbee/controller/platform +fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform +fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/platform +``` + +前端目录: + +```text +yack-web/src/views/iot/platformConnect +yack-web/src/api/iot/platformConnect.js +``` + +权限前缀建议使用 `iot:platformConnect`,避免和现有第三方登录 `iot:platform` 混淆。 + +### 9.2 核心接口 + +```java +public interface PlatformAdapter { + String adapterCode(); + void beforeExecute(PlatformContext context); + void ensureResources(PlatformContext context); + Object buildPayload(PlatformContext context); + PushResult push(PlatformContext context, Object payload); +} +``` + +```java +public interface PlatformFlowExecutor { + PushResult execute(Long tenantChannelId, Long flowId, PlatformReport report, Long bindingId); +} +``` + +```java +public interface PlatformPushRecordService { + void createRealtimeRecords(PlatformReport report); + void createScheduledRecords(Long taskId, Date beginTime, Date endTime); + void retry(Long recordId, boolean force); + void retryBatch(List recordIds, boolean force); +} +``` + +### 9.3 实时推送接入点 + +设备上报原流程保持不变,在设备数据入库或规则处理附近增加平台推送事件: + +```text +设备数据解析 + -> 原有设备日志/历史数据处理 + -> 构造 PlatformReport + -> PlatformPushRecordService.createRealtimeRecords(report) + -> 匹配 tenant_id + device_id 的启用绑定 + -> 生成 PENDING 推送记录 + -> worker 异步执行流程 +``` + +如果短期不改主链路,也可先通过现有规则引擎脚本触发平台推送服务,作为过渡方案。 + +### 9.4 定时推送 + +优先复用 Quartz/sys_job: + +```text +Quartz 触发 + -> 读取 iot_platform_push_task + -> 根据 window_type 计算时间窗口 + -> 查询任务关联绑定 + -> 调用现有历史数据服务按设备取数 + -> 构造 PlatformReport + -> 批量生成推送记录 + -> worker 执行流程 +``` + +不建议第一阶段新增独立调度框架。 + +### 9.5 worker 处理 + +worker 轮询: + +```text +status in (PENDING, RETRY_WAIT) +and next_retry_time <= now +``` + +抢占记录时使用数据库乐观更新,避免多实例重复消费: + +```sql +UPDATE iot_platform_push_record +SET status = 'PROCESSING', locked_by = ?, locked_time = NOW() +WHERE id = ? + AND status IN ('PENDING', 'RETRY_WAIT') +``` + +处理结果:成功写 `SUCCESS`,可重试失败写 `RETRY_WAIT`,超过次数写 `FAILED`,业务跳过写 `SKIPPED`。 + +## 10. 幂等与重推 + +### 10.1 推送记录幂等 + +实时数据优先使用原始日志 id: + +```text +tenantChannelId + flowId + bindingId + pushMode + sourceRefId + payloadHash +``` + +没有 `sourceRefId` 时使用: + +```text +tenantChannelId + flowId + bindingId + pushMode + sourceTime + payloadHash +``` + +定时窗口使用: + +```text +tenantChannelId + flowId + bindingId + windowStart + windowEnd + sourceTime + payloadHash +``` + +`idempotency_key` 必须唯一。 + +### 10.2 外部资源幂等 + +资源创建前先查 `iot_platform_resource_mapping`。 + +DCP 设备级资源必须按 `tenantChannelId + bindingId + serialNumber + resourceType` 唯一。 + +```text +查 mapping + -> 存在则复用 + -> 不存在则加 Redis 锁 + -> 再查一次 + -> 调第三方创建接口或适配器方法 + -> 保存 mapping +``` + +DCP 锁 key: + +```text +platform:dcp:resource:{tenantId}:{tenantChannelId}:{serialNumber} +``` + +### 10.3 手动重推 + +- 失败记录重推:原记录状态重置为 `PENDING`,保留历史 attempt。 +- 成功记录强制重推:新建 `MANUAL` 记录,幂等 key 追加手动批次号。 +- 按设备时间范围补推:查询历史数据后生成新的 `MANUAL` 推送记录。 + +## 11. 权限建议 + +权限前缀:`iot:platformConnect`。 + +```text +iot:platformConnect:template:list/query/add/edit/remove +iot:platformConnect:api:list/add/edit/remove +iot:platformConnect:flow:list/add/edit/remove +iot:platformConnect:tenantChannel:list/query/add/edit/remove +iot:platformConnect:binding:list/query/add/edit/unbind/initResource +iot:platformConnect:task:list/add/edit/remove +iot:platformConnect:record:list/query/retry +``` + +平台模板、接口、流程建议仅平台管理员可维护;租户管理员主要维护租户接入、设备绑定、推送任务和推送记录。 + +## 12. 实施计划 + +### 阶段一:平台模板与租户接入骨架 + +- 新增平台模板、接口、字段、流程、流程步骤表。 +- 新增租户平台实例表。 +- 建立 DCP 平台模板和 DCP 基础流程。 +- 实现租户级授权配置和默认用户/项目映射。 + +### 阶段二:设备绑定与 DCP 资源映射 + +- 新增设备绑定、字段映射、资源映射页面和接口。 +- 实现 DCP 一设备一测点、一设备一 DCP 设备资源建模。 +- 实现 DCP token、userId、projectId、pointId、dcpDeviceId、sensor/module/virtual 资源初始化。 + +### 阶段三:推送执行与审计 + +- 新增推送记录和推送尝试日志。 +- 实现实时推送记录生成。 +- 实现 worker、重试、幂等和手动重推。 +- 支持雨量计实时推送闭环。 + +### 阶段四:补齐 DCP 设备类型与定时补偿 + +- 倾角 payload 组包。 +- 静力水准 sensorData 配置和组包。 +- 轴力计 payload 组包。 +- 雨量递增校验和 `SKIPPED` 状态。 +- 接入 Quartz/sys_job 定时推送和按时间范围补推。 + +### 阶段五:通用平台增强 + +- 通用 HTTP 平台模板。 +- 接口调试。 +- 模板导入导出。 +- MQTT/DB 适配器按需扩展。 +- 日志归档和敏感信息脱敏策略。 + +## 13. 风险与约束 + +- 平台模板是共享基础能力,租户授权、项目、设备绑定、推送记录必须按 `tenant_id` 隔离。 +- DCP 一测点一设备是核心建模约束,不能把测点设计成产品级或平台级共享资源。 +- 实时推送必须异步,不影响设备上报吞吐。 +- token、密码、请求头等敏感信息需要加密保存和脱敏展示。 +- 推送日志增长较快,需要后续增加归档或清理策略。 +- 字段映射表达式必须受限,不能执行任意代码。 +- 定时补推要限制窗口、批次大小和并发数,避免压垮第三方平台。 +- 雨量累计值不递增时应跳过并记录原因,不应无限重试。 + +## 14. 结论 + +平台对接应定位为可复用的 iPaaS/Connector 能力,而不是单个平台的硬编码适配。 + +推荐模型: + +```text +平台模板(共享) + -> 接口配置(共享) + -> 流程编排(共享) + -> 适配器(共享) + -> 租户平台实例(租户级) + -> 用户/项目/资源映射(租户级) + -> 设备绑定和字段映射(租户级) + -> 推送任务、推送记录、推送日志(租户级) +``` + +DCP 是第一批落地模板和适配器,必须保留“一测点一设备”的资源约束;其他租户复用 DCP 平台模板后,只需配置自己的 DCP 账号、用户、项目、设备绑定、字段映射和推送策略。 diff --git a/docs/knowledge-base/README.md b/docs/knowledge-base/README.md index 5d145ec5..4708428a 100644 --- a/docs/knowledge-base/README.md +++ b/docs/knowledge-base/README.md @@ -9,6 +9,7 @@ 3. [数据库与表关系](./03-data-model.md):表分组、核心实体、业务关联、真实外键和约定关联。 4. [数据库关系图](./04-er-diagrams.md):按业务域拆分的 Mermaid ER 图。 5. [数据桥接对接文档](./05-data-bridge-integration.md):设备/物模型查询、设备上报数据推送和第三方主动查询。 +6. [平台对接需求与技术设计](./06-platform-integration-design.md):可复用 iPaaS/数据交换能力,覆盖 OUTBOUND_PUSH、INBOUND_PULL、INBOUND_RECEIVE,支持租户级资源映射、设备绑定、字段映射、交换任务与记录,明确 DCP 一测点一设备约束。 ## 当前结论 diff --git a/docs/platform/DCP接口文档(轴力计).pdf b/docs/platform/DCP接口文档(轴力计).pdf index abc9a2bc..4cb5014a 100644 Binary files a/docs/platform/DCP接口文档(轴力计).pdf and b/docs/platform/DCP接口文档(轴力计).pdf differ