# 平台对接需求与技术设计 ## 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. 前端功能设计 一级菜单：**平台对接**，只有 3 个子菜单。 ```text 平台对接 平台管理 ← 平台模板 + 接口 + 流程 + 租户接入，全部内嵌 设备接入 ← 设备绑定 + 字段映射 + 交换任务，向导式操作 交换记录 ← 数据交换记录 + 尝试日志 + 重试 ``` 资源映射不单独作为菜单，在设备接入页面的绑定详情中内嵌查看。 ### 6.1 平台管理 平台管理页面采用左右布局：左侧是平台列表，右侧是选中平台的详情，通过 Tab 切换。 **左侧：平台列表** - 展示所有平台模板（DCP、通用 HTTP 等）。 - 每行显示平台名称、类型、适配器、启用状态。 - 支持新增平台模板。 **右侧 Tab 1：基本信息** 平台模板主要字段： | 字段 | 说明 | | ------ | ----------------------------------------- | | 平台名称 | 用户可读名称，例如 DCP。 | | 平台编码 | 全局唯一，例如 `DCP`。 | | 平台类型 | DCP、GENERIC_HTTP、GENERIC_MQTT、GENERIC_DB。 | | 适配器编码 | dcp、genericHttp 等。 | | 基础地址模板 | 可作为租户实例默认 baseUrl。 | | 认证方式 | OAuth2、自定义 Token、固定 Token 等。 | | 默认请求配置 | 超时、重试、Content-Type 等。 | | 启用状态 | 启用、停用。 | **右侧 Tab 2：接口列表** 接口属于平台模板，在此 Tab 内直接管理，不单独作为菜单。 接口按类型分组展示，点击接口行展开查看/编辑接口字段： | 分组 | 接口类型 | 说明 | | ------ | ------------------------------------- | ----------------------- | | 认证接口 | TOKEN | 获取或刷新 token。 | | 公共接口 | QUERY、RESOURCE_CREATE、RESOURCE_BIND | 查询用户/项目、创建/绑定资源等。 | | 设备接口 | PUSH、CALLBACK | 按设备类型分组的数据推送接口。 | | 接收入口 | RECEIVE | FastBee 被动接收第三方数据的入口定义。 | 每个接口展开后显示：请求方法、URL 模板、Header/Query/Body 模板、成功条件、响应变量提取规则、重试策略，以及该接口的 REQUEST/RESPONSE 字段列表（可增删改）。 接口字段用于设备接入时的字段映射，字段路径使用 JSONPath 或表单字段名。 **右侧 Tab 3：流程配置** 流程属于平台模板，在此 Tab 内管理。 流程列表按类型展示（OUTBOUND_PUSH、INBOUND_PULL、INBOUND_RECEIVE、RESOURCE_SYNC），点击流程行展开步骤列表，支持拖拽排序步骤。 步骤类型：API（引用接口）、ADAPTER（适配器方法）、CONDITION（条件判断）、TRANSFORM（数据转换）、SINK（写入业务落点）。 **右侧 Tab 4：租户接入** 租户接入表示当前租户启用该平台模板，在此 Tab 内配置，不单独作为菜单。 主要字段： | 字段 | 说明 | | ------- | -------------------------------- | | 实例名称 | 租户内展示名。 | | baseUrl | 可覆盖模板默认地址。 | | 授权配置 | 租户自己的账号、clientId、secret、token 等。 | | 默认第三方用户 | DCP userId 或其他平台用户。 | | 默认项目 | DCP projectId 或其他平台项目。 | | 接收安全配置 | webhook token、签名密钥、IP 白名单等。 | | 启用状态 | 启用、停用。 | 租户接入配置完成后，才能在设备接入页面选择该平台实例。 ### 6.2 设备接入 设备接入页面是租户操作的核心入口，采用列表 + 抽屉向导的方式。 **列表视图** 展示当前租户所有设备绑定，列显示：平台实例、设备类型、FastBee 设备名/编号、交换模式、状态、最近成功时间。 支持按平台实例、设备类型、状态筛选。 操作列：编辑映射、启用/停用、初始化资源、查看资源映射、查看交换记录、解绑。 **新增绑定：向导式抽屉（3步）** **第 1 步：选择平台与设备类型** - 下拉选择平台实例（从当前租户已启用的租户接入中加载，不是写死的枚举）。 - 选择平台实例后，动态加载该平台模板下的设备接口分组，作为设备类型选项（例如 DCP 加载后显示：雨量计、倾角仪、静力水准、轴力计）。 - 设备类型对应平台模板中的 PUSH 接口分组，由平台模板定义，不在前端硬编码。 **第 2 步：选择 FastBee 设备** - 选择产品（下拉）。 - 选择设备（从该产品下的设备列表选，支持搜索设备名/编号）。 - 展示设备基本信息和物模型字段列表（供下一步参考）。 - 可选：填写绑定级第三方用户 id、项目 id（留空则使用租户接入的默认值）。 **第 3 步：字段映射** 左右分栏： - 左侧：FastBee 物模型字段列表（identifier、字段名、数据类型），以及系统字段（serialNumber、deviceName、reportTime）。 - 右侧：第 1 步选定的设备接口的 REQUEST 字段列表（字段路径、字段名、是否必填）。 - 中间：映射关系，每行对应一个目标字段，来源可选：物模型字段、系统字段、常量、资源映射（pointId 等）、JSONPath 表达式。 - 支持从平台模板复制默认映射（如果平台模板预置了默认映射规则）。 字段映射同时支持 OUTBOUND（FastBee → 第三方）和 INBOUND（第三方 → FastBee），通过方向 Tab 切换。 **绑定详情：资源映射内嵌** 点击列表行的"查看资源映射"，在侧边抽屉中展示该绑定下的资源映射记录（USER、PROJECT、POINT、DEVICE、SENSOR 等），不单独作为菜单。 **交换任务** 在设备接入列表页顶部提供"交换任务"入口（或作为 Tab），管理当前租户的定时推送、主动拉取、资源同步任务。 任务配置：选择平台实例、选择流程、设置 Cron 表达式、时间窗口类型、批次大小、最大重试次数。 任务与绑定的关联在任务详情中配置（选择哪些设备绑定参与该任务）。 ### 6.3 交换记录 交换记录是运维监控视图，展示所有数据交换的执行状态。 筛选条件：平台实例、方向（OUTBOUND_PUSH/INBOUND_PULL/INBOUND_RECEIVE）、设备编号、状态、时间范围。 列表列：方向、平台实例、设备名/编号、数据时间、状态、重试次数、最近错误、创建时间。 操作： - 点击行展开查看：原始输入、标准消息、最终 payload、幂等 key、请求响应详情、错误信息。 - 失败记录：单条重试、批量重试。 - 成功记录：强制重放（新建 REPLAY 记录，幂等 key 追加批次号）。 - 点击"尝试日志"查看每次请求的 URL、请求头（脱敏）、请求体、响应状态、响应内容、耗时。 ## 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 账号、用户、项目、设备绑定、字段映射和数据交换策略。