故障录波 MQTT 整体设计

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

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、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 异步执行

前端 ──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` 为例）

取 Redis 网关锁 fault-record:lock:<env>:<gatewaySN>（带 TTL 并续租）。取不到就结束操作，状态为 gateway_locked（本工具内已有操作在进行）。
订阅 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 多个订阅者，互不影响。
用 Publish 发 377，写 sent。

收到报文就交给 L2 解析，推进状态，并把原始报文追加到记录：

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

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

终止状态分开记录：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（测试服网关 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）：

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. 待实测前提

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

协议遗留问题见 protocol-specification.md B12、B14。

technical-design.md 19 KB Історія Запис