# 故障录波 MQTT 整体设计

> 状态：设计草案，2026-09-30；2026-10-16 首版已按本文实现，实现与代码位置见 [implementation-status.md](implementation-status.md)；2026-10-14 按 [FaultFilter_MQTT协议文档](FaultFilter_MQTT协议文档.md) 更新业务协议；2026-10-16 补充文件上传链路、双环境、分层架构、异步任务、存储和取消等用户已确认的决策。功能目标见 [requirements.md](requirements.md)，页面设计见 [ui-design.md](ui-design.md)，报文依据见 [protocol-observations.md](protocol-observations.md) 和 [protocol-specification.md](protocol-specification.md)。

## 1. 设计目标与原则

建立直接经 MQTT 与设备交互的故障录波能力，覆盖设备信息（实时数据）、录波记录获取、参数读取和参数设置，同时支持**生产服**和**测试服**。

- **只发起、不应答**：本工具只发送 `377`（以及取消用的 `cmd=4`），其余消息全部监听。`380`、`607`、`610` 由 FWH 平台发出，平台在收到 `379(cmd=2)` 后自动发起文件上传。
- **分层清晰**：每层只负责自己的事，只依赖下层。MQTT 客户端只提供底层收发，业务判断全部在上层。
- **复用现有模块**：asynq 任务系统、MongoDB、MinIO、Redis、`fwh_pro`／`fwh_test` 站点信息接口、uPlot。
- **只显示原始值**：录波数据不做物理量换算（B15）。
- **DataDownload 暂不接入**：本功能完整实现并测试通过后，再考虑开放给 DataDownload；DataDownload 继续使用原 FWH API 方案。

## 2. 分层架构

```text
L5 Route / Controller        HTTP 入参校验、出参转换
L4 Service                   对外用例：创建操作、查询状态与结果、取消、导出、查询设备列表
L3 Task Executor (asynq)     操作执行器：按状态机推进、持久化每一步、重启恢复
L2 Protocol / Parser         纯函数，不访问网络、数据库和文件系统
     ├─ protocol：构造 377 dataArea，解析 378／379／607～610（兼容旧版本 608 把 `id` 误写为第二个 `result`）
     └─ parser：解压、hexData(code=1/2/3)、failuredata、waveformdata、CSV 生成
L1 FWH MQTT Client           按环境实例化；FWH 报文外层（len／crc／信封）+ Request／Publish／Subscribe
L0 utils/mqttClient          paho 封装

存储：MongoDB（操作记录） · MinIO（原始文件） · Redis（网关锁、asynq）
外部：FWH 平台 HTTP（站点信息、hes-upload 下载）
```

| 层 | 建议位置 | 只负责 | 不负责 |
| --- | --- | --- | --- |
| L0 | `utils/mqttClient` | 连接、订阅、发布、重连 | FWH 报文格式 |
| L1 | `repositories/mqtt/fwh_mqtt` | 按环境的连接实例；FWH 信封编解码；三种收发方法；订阅期间保持连接 | 任何 cmdType／cmd 的业务含义 |
| L2 | `function/FaultRecordFile`（解析器）、`function/FaultRecordProtocol`（协议编解码） | 字节和 JSON 的结构化转换 | IO |
| L3 | `task/queue/faultRecordQueue` | 一次操作的完整执行和恢复 | HTTP 入参、页面格式 |
| L4 | `services/faultRecordServices.go` | 用例编排，调用 Repository 和入队 | 协议细节 |
| L5 | `controllers/faultRecordControllers.go`、`routes/faultRecordRoutes.go` | HTTP | 业务 |
| 存储 | `repositories/mongo/faultRecordRepo.go`、`repositories/minio/faultRecordRepo.go` | 数据读写 | 业务 |
| 下载 | `fwh_pro`／`fwh_test` 中各一个无 token 下载函数 | 按环境域名下载 | 解压、解析 |

L2 是纯函数，用仓库中的样本文件（[CloudLog_2026-10-14_104644.csv](CloudLog_2026-10-14_104644.csv)、[2026-10-14_05_46_20.tar.bz2](2026-10-14_05_46_20.tar.bz2)）做单元测试。以后 DataDownload 接入时，复用 L1 和 L2。

## 3. L1 FWH MQTT 客户端

### 3.1 对外方法

| 方法 | 用途 | 行为 |
| --- | --- | --- |
| `Request(ctx, req) (*Msg, error)` | 一问一答（保留原方式） | 订阅应答 topic → 发送 → 等到第一条满足 `Match` 的消息或超时 → 取消订阅 |
| `Publish(ctx, topic, msg) error` | 只请求 | 发送后立即返回 |
| `Subscribe(topic, handler) (cancel func(), error)` | 持续监听 | 同一 topic 可有多个订阅者，各自收到每条消息；`cancel` 只移除自己 |

- `handler` 收到 `{Topic, Envelope, Raw []byte, ReceivedAt}`。`Envelope` 为解析后的外层，`Raw` 为原始字节，供上层按需重新解析（如旧版本 608 的重复 `result` 键）。
- `Request` 的匹配条件由调用方传入（如 `cmdType=378 && dataArea.cmd=1`），客户端不内置任何业务过滤。
- 信封构造（`len`、`crc`、`timeStamp`、`snno`）属于 L1 的通用能力，以 `NewEnvelope(cmdType, equipNo, dataArea)` 提供。
- 现有 `EnableXlog`、`FaultRecordParamSetting`、`ChargeChannel` 中的业务内容移到上层，L1 只保留通用部分。

### 3.2 连接与订阅管理

- 内部维护 `topic → 订阅者集合`。第一个订阅者加入时向 broker 订阅；最后一个退出时向 broker 取消订阅。
- **有活动订阅时不做空闲断开**；没有订阅且空闲 5 分钟后才断开（保持现有省连接的行为）。
- 断线自动重连后恢复全部 broker 订阅（L0 已有 `ResumeSubscribe`）。L1 需要记录断线区间，供上层判断“这段时间可能漏了消息”。
- 发布仍经过一个 worker 串行发送；订阅回调不阻塞 paho 线程，按订阅者各自的缓冲 channel 分发，满了就丢弃并记日志。

## 4. 操作模型

### 4.1 操作类型

| 操作 | `cmd` | 输入 | 结果 |
| --- | --- | --- | --- |
| `device_info` | 0 | `codeList ⊆ {1,2,3}` | `hexList` 原文 + 解析结果 |
| `record_fetch` | 2 | **一个**设备、一条或多条记录序号（最多 32） | 原始文件（MinIO）+ 解析结果 |
| `param_read` | 3 | `idList`（功能操作页只用于 4、7） | `idInfo` 原文 + 解析结果 |
| `param_write` | 1 | `paraList` | `setResultList` |

每次操作只针对一个网关下的一个设备（`aGate` 或某个 `pe`）。

`cmd=3`（查询参数）和 `cmd=4`（退出）只在 FaultFilter 文档中定义，旧描述、现有代码和实报只有 `0/1/2`。用户判断外层可能支持，先按支持实现，实测后确认（B01）。如不支持：功能操作页的 4、7 只能设置不能读取，取消只停止本工具的等待。参数设置页（5、6）读取和复读走 `cmd=0`，不受影响。`record_fetch` 可一次获取多条记录（2026-10-19 用户确认设备支持）：文件按压缩包中的 `index_<n>` 目录区分，不会混淆；379 中的多组 `4/5` 结果按“每条记录一组、顺序与请求一致”对应（待实测）。解析后核对压缩包是否包含全部请求的序号，缺少时在结果中提示。等待时间按记录数延长：379 每多一条加 1 分钟，609 每多一条加 30 秒。

下载时机：收到 609 成功后推进到 `uploaded`，收到 610 后 15 秒开始下载，失败后每 15 秒重试，一直到总超时（从 379 起算 5 分钟，每多一条记录加 30 秒）；仍失败记为“下载失败”并注明尝试次数（用户要求，2026-10-19）。连续相同的失败只保留最后一条下载记录，并记录累计失败次数和首次失败时间。610 没有文件名，按“收到 609 之后的下一条 610”对应，其 `result` 只记录不判断。未收到 609 或 610 时只在超时时按规则直接下载一次；服务重启恢复时先尝试下载一次。不做收到 379 后定期试探下载的兜底（用户决定，2026-10-19）。

**手动重试下载**：上传已完成（收到 609）但下载失败或超时的操作，页面提供“重试下载”按钮：操作重新置为“上传完成”，立即下载，失败后每 15 秒重试，**最多尝试 3 次**（用户要求，2026-10-19）；每次触发都在原始报文中追加一条 `action` 记录（操作人、第几次重试、重试前状态与错误）；每次重试使用新的任务 ID，服务重启后按当前重试序号恢复。

**文件名**：不同固件的 609.fileName 有的带 `.tar.bz2`、有的不带。**每次下载（含自动和手动重试）都由批次名重新拼接 `<批次>.tar.bz2` 生成下载链接**，不依赖已保存的文件名（10-19 测试服实测缺后缀导致下载为空）；MinIO 对象键和原始包导出文件名同样使用补齐后缀的名称。

**前端轮询**：开始后 10 秒内每 2 秒查询一次，之后每 5 秒一次；同一操作只保留一条轮询。

文件下载是 HTTP 请求，不是 MQTT 报文；每次下载作为一条 `http` 记录写入操作的原始报文（路径、结果、大小或错误、耗时）。操作结束后再记录约 2 秒内到达的报文（如 609 之后的 610）。

### 4.2 异步执行

```text
前端 ──POST 创建操作──▶ Service ──写 Mongo(pending) + 入队 asynq──▶ 返回 operationId
前端 ──GET 轮询 operationId──▶ Service ──读 Mongo──▶ 状态、阶段时间、结果
asynq worker ──▶ Task Executor ──▶ 执行并逐步写回 Mongo
```

- 队列：新建 `faultRecordQueue`，在 `task/queue/taskQueue.go` 注册，沿用 `dataDownloadQueue` 的写法（`asynq.TaskID` 保证幂等，`Retention` 保留记录）。
- **不依赖 asynq 重试做业务重试**：重复执行由执行器根据 Mongo 状态自行判断，避免重复发送 `377(cmd=2)` 清空设备文件。
- 任务超时 15 分钟；各阶段等待：378 为 30 秒，379 为 2 分钟（`record_fetch` 4 分钟），379 到 609 为 5 分钟（暂定，实测后调整，见 `faultRecordQueue/deps.go`）。
- `MaxRetry` 设为 3 只用于进程崩溃后重新投递；业务失败由执行器写入状态并正常返回，不触发 asynq 重试。（asynq 对崩溃任务按重试计数处理，`MaxRetry(0)` 会直接归档、无法恢复。）
- 收到 609 并下载完成后操作即结束，之后到达的 610 不再记录。

### 4.3 执行器流程（以 `record_fetch` 为例）

1. 取 Redis 网关锁 `fault-record:lock:<env>:<gatewaySN>`（带 TTL 并续租）。取不到就结束操作，状态为 `gateway_locked`（本工具内已有操作在进行）。
2. 订阅 `fwh/c2s/mng/<sn>` 和 `fwh/s2c/mng/<sn>`（378/379/380）；`record_fetch` 另订阅 `fwh/s2c/debug/<sn>` 和 `fwh/c2s/debug/<sn>`（607/610 与 608/609，用户确认，2026-10-19）。debug topic 与 Xlog 共用，客户端支持同一 topic 多个订阅者，互不影响。
3. 用 `Publish` 发 `377`，写 `sent`。
4. 收到报文就交给 L2 解析，推进状态，并把原始报文追加到记录：

```text
pending → sent → accepted(378) → ready(379) → uploading(607) → uploaded(609) → downloaded → done
```

5. 收到 `609.result=0` 后推进到 `uploaded`；收到 610 后 15 秒开始下载（失败每 15 秒重试至总超时），存入 MinIO，解析，写 `done`。
6. 释放锁，取消订阅。

终止状态分开记录：`busy`（设备回复 busy，明确提示，不重试）、`rejected`（378 失败）、`item_failed`（379 中 `resultOne` 不为 ok）、`timeout_<阶段>`、`upload_failed`（609 失败）、`download_failed`、`parse_failed`、`cancelled`、`interrupted`。

其他操作按同一框架执行，只到 `ready(379)` 为止，再解析结果写 `done`。

### 4.4 关联规则

- 377 → 378／379：环境 + `equipNo` + `cmdType` + `dataArea.cmd`。网关锁保证本工具内同一网关同一时间只有一个操作。
- 379 → 607／608／609：`equipNo` + 文件名（`609.fileName` = `379.filename + ".tar.bz2"`）。
- 608 在设备上行方向：订阅不到 607 时，收到文件名匹配的 608 也推进到“平台上传中”，并从 `608.id` 取得平台记录 ID。
- 发送的 `377` 也会出现在 s2c topic 上，按 `cmdType` 过滤掉。

### 4.5 取消（`cmd=4`）

`cmd=4` 是 FaultFilter 协议的“退出”命令：通知设备结束当前处理，`data` 为 `{}`，没有业务回复。它是设备忙时**唯一**会被接受的命令。

| 取消时所处阶段 | 处理 |
| --- | --- |
| `pending`（还没发 377） | 直接取消，不发任何消息 |
| `sent`／`accepted`（设备正在处理我们的命令） | 发 `377(cmd=4)`，标记 `cancelled`，释放锁 |
| `ready` 及之后（设备已处理完，平台在上传） | 不发 `cmd=4`（设备已无处理中的命令，且无法终止平台上传），只停止等待，标记 `cancelled` |
| 状态为 `busy` 等终止态 | 不可取消 |

限制：

- `cmd=4` 作用于设备，不区分发起方。只有在本操作的命令确实被设备接受（没有收到 `busy`）时才发送，避免中断别人发起的处理。
- 设备是否真的停止、外层是否有回复，待实测；页面显示“已发送取消”，不承诺设备一定停止。

### 4.6 重启恢复

asynq 会把服务重启时未完成的任务重新投递。执行器先读 Mongo 状态，从断点继续：

| 重启时状态 | 恢复方式 |
| --- | --- |
| `pending` | 正常执行 |
| `sent`／`accepted` | 不能确定设备是否已回复，且重发会清空设备文件，标记 `interrupted`，由用户手动重试 |
| `ready`／`uploading`（已知文件名） | 直接按下载地址尝试下载；成功则继续解析，失败则重新订阅等待 `609`，直到超时 |
| `uploaded`／`downloaded` | 继续下载／解析 |

服务启动时扫描 Mongo 中未终止、但 asynq 中已不存在的操作（例如 Redis 数据丢失），按上表处理。

## 5. 设备与站点信息

设备列表按操作所属环境（`fwh_pro` 或 `fwh_test`）查询，不复用 `services.GetSiteInfoData`（会写入站点信息表）。设备种类与平台设备类型对应（用户确认，2026-10-20）：

| 平台设备类型 | 名称 | 种类（代码） | SN 来源 | 377 `type` |
| --- | --- | --- | --- | --- |
| 1 | aGate/MCA | `aGate` | 网关 SN | `aGate` |
| 0 | aPower-PE | `aPowerPE` | `/hes-gateway/manage/getDeviceInfo` 的 `fhpSn`（逗号分隔） | `pe` |
| 2 | aPower-MPPT | `aPowerMPPT` | 与 aPower-PE 同一 SN；仅 PE 硬件版本为 21、24 的 aPower 支持 | `mppt` |
| 3 | aHub（即 aPbox2.0） | `aHub` | `/hes-gateway/manage/selectGatewayPower` 的 `apbox20List[].apbox20Sn` | `apbox` |
| — | DIF（Digital Interface） | — | 暂未知 | `dif`（后端协议层支持，前端暂未开放） |

- `getDeviceInfo` 中 `peHwVerList`、`apowerList` 与 `fhpSn` 的 aPower 顺序一一对应；PE 硬件版本 21（aPower S）、24（aPower 2.1S）带 MPPT，后续可能扩展（集中在 `FaultRecordProtocol` 的 `SupportsMPPT`）。
- 站点名称和在线状态仍取自站点信息接口（`QuerySiteInfoList`）。
- `getDeviceInfo` 失败时退回站点信息的 `fhpSn`，无法判断 aPower-MPPT；`selectGatewayPower` 失败时不列出 aHub。两种情况都在页面提示。
- 377 中的设备类型共 5 种：`pe`、`aGate`、`mppt`、`apbox`、`dif`（用户确认，2026-10-20）。
- 样本：[设备信息.test.franklinwh.com.har](设备信息.test.franklinwh.com.har)（测试服网关 `10070052S00X25150083`：2 台 aPower，第 2 台为 aPower S，1 台 aHub）。

PE 硬件版本对照（格式：版本值 - 版本名称 - 对外名称；代码中使用版本名称）：

| 值 | 版本名称 | 对外名称 |
| --- | --- | --- |
| 0 | aPower 1.0 | aPower X |
| 2 | aPower 1.2（澳洲）/aPower 1.3（北美） | aPower X |
| 4 | aPower 1.4 | aPower X |
| 20 | aPower 2/aPower 2(LR) | aPower 2/aPower 2(LR) |
| 21 | aPower S | aPower S |
| 22 | aPower 2(SH) | aPower 2(SH) |
| 23 | aPower 2.1 | aPower 2.1 |
| 24 | aPower 2.1S | aPower 2.1S |

## 6. 测试服与生产服

每次操作绑定一个环境，MQTT、站点信息和文件下载都取自同一环境，不能混用。

| 项目 | 生产服 | 测试服 |
| --- | --- | --- |
| MQTT broker／证书 | `FWH_MQTT_BROKER`、`FWH_MQTT_CA/CERT/KEY` | `FWH_MQTT_TEST_BROKER`、`FWH_MQTT_TEST_CA/CERT/KEY` |
| FWH 平台域名 | `FWH_ENERGY`（默认 `energy.franklinwh.com`） | `FWH_TEST_ENERGY`（如 `test.franklinwh.com`） |
| 站点信息 | `fwh_pro` | `fwh_test` |
| 下载凭据 | 不需要 | 不需要 |

文件下载地址（用户确认，2026-10-16）：

```text
https://<环境域名>/hes-upload/v1/file/download/logger/<equipNo>/<609.fileName>
```

- `<equipNo>` 取 `609` 外层的 `equipNo`；`pe` 场景下是否仍为网关 SN 待实测。
- 域名取自环境配置，不从 `607.url` 推导（`607.url` 是上传地址，样本为 `http://`）。
- 文件扩展名是 `.tar.bz2`，实际是 gzip tar；解压按文件头判断，同时兼容 gzip 和 bzip2。

## 7. 存储

### 7.1 MongoDB：操作记录

选择 MongoDB：每次操作的报文条数和结果结构不固定，适合文档存储；与 DataDownload 的任务记录方式一致，已有连接与 Repository 写法。不使用 PostgreSQL：波形约 49KB，查看和导出时现场解析即可，无需结构化建表。

集合 `fault_record_operations`（字段草案）：

| 字段 | 说明 |
| --- | --- |
| `_id` | operationId |
| `env` | `prod`／`test` |
| `gatewaySn`、`deviceType`、`deviceSn` | 目标 |
| `kind`、`cmd`、`request` | 操作类型与请求参数 |
| `status`、`stage`、`stageTimes` | 当前状态和各阶段时间 |
| `operator` | 操作人（取自登录用户），历史共享可见 |
| `messages[]` | 原始报文：方向、topic、cmdType、接收时间、原文 |
| `result` | 解析后的结果（目录、设备信息、参数等） |
| `fileName`、`faultRecordId`、`minioObject` | `record_fetch` 专用 |
| `error` | 失败原因 |
| `createdAt`、`updatedAt` | |

索引：`env + gatewaySn + createdAt`、`status`、`createdAt`。暂不实现数据清理。

### 7.2 MinIO：原始文件

- 对象键：`fault-record/<env>/<gatewaySn>/<fileName>`，保留下载到的原始字节和原始文件名。
- 对象键不经 HTTP 暴露，由 Service 按 operationId 读取。

### 7.3 导出

| 格式 | 内容 |
| --- | --- |
| 原始压缩包 | 直接从 MinIO 读出返回 |
| CSV（zip） | `waveform.csv`：每行一个采样点，列为点序号、相对故障时刻毫秒、各通道原始值，表头为通道序号和含义代码；`summary.csv`：基本信息、版本、触发源、快照通道、采样通道、可配通道 |

CSV 现场生成，不缓存。

## 8. 时间口径

目录、`failuredata`、对时功能中的时间均按**设备时间**原样显示，不做时区换算（暂定，实测有问题再改）。操作记录中的时间（创建、阶段时间）用服务器时间存储，前端按本地时间显示。

## 9. 待实测前提

1. 本工具直接发出 `377(cmd=2)` 后，平台是否同样自动发 `607` 并上传。这是“只监听”方案成立的前提。
2. ~~订阅 s2c 方向~~：测试服已实测可订阅（收到 380）；607～610 的 topic 已确认（debug），待实测确认能收到。
3. `cmd=0/1/3` 的 `380` 是否也由平台回复。
4. `pe` 场景：外层 `equipNo` 与 `info[].sn` 的关系、下载路径中的 `<equipNo>`。
5. `cmd=1/3` 经外层的实际报文和失败样本；`busy` 经外层如何体现。
6. `cmd=4` 的外层表现和实际效果。

协议遗留问题见 [protocol-specification.md](protocol-specification.md#10-待补充清单) B12、B14。