FranklinWH-Toolbox/docs/fault-record/technical-design.md

故障录波 MQTT 整体设计

状态：设计草案，2026-09-30；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 重复键）
     └─ 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、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 异步执行

前端 ──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 解析，推进状态，并把原始报文追加到记录：

   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）：

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 B12、B14。