# 故障录波 MQTT 整体设计

> 状态：设计草案，2026-09-30；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 重复键）
     └─ 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/SnapFile` 的模式放解析器）；协议编解码放同级 `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` | 数据读写 | 业务 |
| 下载 | `repositories/api/fwh_upload`（或在 `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 | **一个**设备、**一条**记录序号 | 原始文件（MinIO）+ 解析结果 |
| `param_read` | 3 | `idList` | `idInfo` 原文 + 解析结果 |
| `param_write` | 1 | `paraList` | `setResultList` |

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

`cmd=3`（查询参数）和 `cmd=4`（退出）只在 FaultFilter 文档中定义，旧描述、现有代码和实报只有 `0/1/2`。用户判断外层可能支持，先按支持实现，实测后确认（B01）。如不支持：参数读取页、参数设置的“读取当前值”和“自动复读”下线，取消只停止本工具的等待。`record_fetch` 每次只取一条记录，避免多组 `4/5` 无法区分。

### 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 自动重试**（`MaxRetry(0)`），重复执行由执行器根据 Mongo 状态自行判断，避免重复发送 `377(cmd=2)` 清空设备文件。
- 任务超时：`record_fetch` 10 分钟，其他 3 分钟（暂定，实测后调整）。

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

1. 取 Redis 网关锁 `fault-record:lock:<env>:<gatewaySN>`（带 TTL 并续租）。取不到就结束操作，状态为 `gateway_locked`（本工具内已有操作在进行）。
2. 订阅 `fwh/c2s/mng/<sn>` 和 `fwh/s2c/mng/<sn>`。
3. 用 `Publish` 发 `377`，写 `sent`。
4. 收到报文就交给 L2 解析，推进状态，并把原始报文追加到记录：

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

5. 收到 `609.result=0` 后，按环境域名下载，存入 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"`）。
- 发送的 `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. 设备与站点信息

- 用户输入网关 SN，Service 按环境调用 `fwh_pro.QuerySiteInfoList` 或 `fwh_test.QuerySiteInfoList` 获取站点信息。
- 可选设备为：`aGate`（网关 SN）+ `fhpSn` 中逗号分隔的每个 aPower SN。**aPower SN 就是 `pe` SN**（用户确认，2026-10-16）。
- 生产服和测试服是两个不同的接口（`fwh_pro` 与 `fwh_test`），必须按操作所属环境调用，不能用生产服站点信息去操作测试服设备，反之亦然。
- 只调用 Repository 层接口，不复用 `services.GetSiteInfoData`：后者会把结果写入站点信息表，测试服数据不能写入。

## 6. 测试服与生产服

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

| 项目 | 生产服 | 测试服 |
| --- | --- | --- |
| MQTT broker／证书 | `FWH_MQTT_BROKER`、`FWH_MQTT_CA/CERT/KEY` | `FWH_TEST_MQTT_BROKER`、`FWH_TEST_MQTT_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 方向（`607/380/610`）；不能时只能依赖 `609` 推进状态。
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。