datatrace/source/mq.cpp

#define NOMINMAX
#include "pq_app.h"

#include "json.hpp"

#include <windows.h>
#include <winhttp.h>
#include <objbase.h>

#include <algorithm>
#include <chrono>
#include <cctype>
#include <cwctype>
#include <filesystem>
#include <fstream>
#include <iomanip>
#include <map>
#include <mutex>
#include <set>
#include <sstream>
#include <stdexcept>
#include <thread>
#include <unordered_map>
#include <unordered_set>

/*
 * mq.cpp
 *
 * 本文件负责“外部数据输入/输出”：
 *   - HTTP 请求台账、ICD 列表和 XML 文件下载；
 *   - 动态加载 RocketMQ DLL，初始化生产者/消费者；
 *   - 发送数据追踪请求；
 *   - 消费 DATATopic/TraceTopic，把 jsondata/origin 落盘到 trace_data；
 *   - RocketMQ 不可用时把待发消息写到 mq_outbox_*.jsonl，便于离线排查。
 *
 * 数据追踪落盘结构：
 *   runtime\trace_data\process_N\device_x\monitor_y\
 *     trace_context.json        本次追踪上下文，GUI 自动解析依赖它；
 *     sent_trace_request.json   已发送/待发送追踪请求；
 *     jsondata_01.txt           DATA_TYPE=01 的 DATATopic 数据；
 *     jsondata_02.txt/03.txt    闪变等其它 DATA_TYPE；
 *     origin_*.txt              TraceTopic 原始 MMS/报告数据；
 *     final_sorted.xls          自动解析生成的工作簿。
 *
 * 阅读本文件时可留意的 C++/Windows/RocketMQ 语法：
 *   - WinHTTP 的 Open -> Connect -> OpenRequest -> Send/Receive 是成对句柄流程，错误码需逐层检查；
 *   - LoadLibrary/GetProcAddress 在运行时解析 RocketMQ C DLL，成员中的 (*Fn)(...) 是函数指针；
 *   - RocketMqRuntime 是进程级单例；std::mutex + lock_guard 保护 SDK 回调与 GUI 操作共享的状态；
 *   - TraceSession 保存一次点击“数据追踪”的开始时间、GUID 和去重集合，阻断历史/重复消息；
 *   - DetachedHandles 配合析构函数实现 RAII，先从共享状态摘除句柄，再在锁外关闭 SDK；
 *   - nlohmann::json 同时用于构造双层消息体和递归查找字段；filesystem 统一创建/清理落盘目录；
 *   - std::thread([h]{...}).detach() 是异步清理；lambda 按值捕获 h，避免引用已失效的局部变量。
 */

using json = nlohmann::json;
namespace fs = std::filesystem;

namespace {
/**
 * Function: executableDir
 *
 * @brief 实现外部通信层的 “executableDir” 功能，服务于 HTTP、RocketMQ 或运行时文件管理。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @return 构造、解析或查找到的结果；所有权与生命周期遵循返回类型。
 */
fs::path executableDir() {
    wchar_t path[MAX_PATH]{};
    DWORD n = GetModuleFileNameW(nullptr, path, (DWORD)std::size(path));
    if (n == 0 || n >= std::size(path)) return fs::current_path();
    return fs::path(path).parent_path();
}
/**
 * Function: appRootDir
 *
 * @brief 实现外部通信层的 “appRootDir” 功能，服务于 HTTP、RocketMQ 或运行时文件管理。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @return 构造、解析或查找到的结果；所有权与生命周期遵循返回类型。
 */
fs::path appRootDir() {
    fs::path dir = executableDir();
    std::wstring name = dir.filename().wstring();
    std::transform(name.begin(), name.end(), name.begin(), [](wchar_t ch) { return (wchar_t)towlower(ch); });
    // exe 通常在 bin 下运行，runtime/source 与 bin 同级，所以从 bin 回到项目根。
    if (name == L"bin" && dir.has_parent_path()) return dir.parent_path();
    return dir;
}
/**
 * Function: runtimeDir
 *
 * @brief 处理 “runtimeDir” 对应的业务入口或事件，并调度下游函数完成实际工作。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @return 构造、解析或查找到的结果；所有权与生命周期遵循返回类型。
 */
fs::path runtimeDir() {
    return appRootDir() / "runtime";
}
/**
 * Function: runtimeJsonDir
 *
 * @brief 处理 “runtimeJsonDir” 对应的业务入口或事件，并调度下游函数完成实际工作。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @return 构造、解析或查找到的结果；所有权与生命周期遵循返回类型。
 */
fs::path runtimeJsonDir() {
    return runtimeDir() / "json";
}
/**
 * Function: runtimeLogsDir
 *
 * @brief 处理 “runtimeLogsDir” 对应的业务入口或事件，并调度下游函数完成实际工作。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @return 构造、解析或查找到的结果；所有权与生命周期遵循返回类型。
 */
fs::path runtimeLogsDir() {
    return runtimeDir() / "logs";
}
/**
 * Function: runtimeJsonPath
 *
 * @brief 处理 “runtimeJsonPath” 对应的业务入口或事件，并调度下游函数完成实际工作。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @param name 待解析、比较、显示或发送的 UTF-8 文本。
 * @return 构造、解析或查找到的结果；所有权与生命周期遵循返回类型。
 */
fs::path runtimeJsonPath(const std::string& name) {
    return runtimeJsonDir() / fs::u8path(name);
}
/**
 * Function: runtimeLogPath
 *
 * @brief 处理 “runtimeLogPath” 对应的业务入口或事件，并调度下游函数完成实际工作。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @param name 待解析、比较、显示或发送的 UTF-8 文本。
 * @return 构造、解析或查找到的结果；所有权与生命周期遵循返回类型。
 */
fs::path runtimeLogPath(const std::string& name) {
    return runtimeLogsDir() / fs::u8path(name);
}
/**
 * Function: utf8ToWide
 *
 * @brief 实现外部通信层的 “utf8ToWide” 功能，服务于 HTTP、RocketMQ 或运行时文件管理。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @param s 调用方传入的业务参数；其用途与约束由类型、名称及函数体共同定义。
 * @return 构造、解析或查找到的结果；所有权与生命周期遵循返回类型。
 */
std::wstring utf8ToWide(const std::string& s) {
    if (s.empty()) return L"";
    int n = MultiByteToWideChar(CP_UTF8, 0, s.data(), (int)s.size(), nullptr, 0);
    std::wstring out(n, L'\0');
    MultiByteToWideChar(CP_UTF8, 0, s.data(), (int)s.size(), out.data(), n);
    return out;
}
/**
 * Function: wideToUtf8
 *
 * @brief 实现外部通信层的 “wideToUtf8” 功能，服务于 HTTP、RocketMQ 或运行时文件管理。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @param s 调用方传入的业务参数；其用途与约束由类型、名称及函数体共同定义。
 * @return 构造、解析或查找到的结果；所有权与生命周期遵循返回类型。
 */
std::string wideToUtf8(const std::wstring& s) {
    if (s.empty()) return "";
    int n = WideCharToMultiByte(CP_UTF8, 0, s.data(), (int)s.size(), nullptr, 0, nullptr, nullptr);
    std::string out(n, '\0');
    WideCharToMultiByte(CP_UTF8, 0, s.data(), (int)s.size(), out.data(), n, nullptr, nullptr);
    return out;
}
/**
 * Function: jsonString
 *
 * @brief 实现外部通信层的 “jsonString” 功能，服务于 HTTP、RocketMQ 或运行时文件管理。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @param obj 调用方传入的业务参数；其用途与约束由类型、名称及函数体共同定义。
 * @param key 待解析、比较、显示或发送的 UTF-8 文本。
 * @return 构造、解析或查找到的结果；所有权与生命周期遵循返回类型。
 */
std::string jsonString(const json& obj, const char* key) {
    auto it = obj.find(key);
    if (it == obj.end() || it->is_null()) return "";
    if (it->is_string()) return it->get<std::string>();
    if (it->is_number_integer()) return std::to_string(it->get<long long>());
    if (it->is_number_unsigned()) return std::to_string(it->get<unsigned long long>());
    if (it->is_number_float()) return std::to_string(it->get<double>());
    if (it->is_boolean()) return it->get<bool>() ? "true" : "false";
    return it->dump();
}
/**
 * Function: setContains
 *
 * @brief 把 “setContains” 对应的变更应用到模型、控件或输出文档。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @param allowed 调用方传入的业务参数；其用途与约束由类型、名称及函数体共同定义。
 * @param value 待解析、比较、显示或发送的 UTF-8 文本。
 * @return 条件成立或操作成功时为 true，否则为 false。
 */
bool setContains(const std::unordered_set<std::string>& allowed, const std::string& value) {
    return allowed.empty() || allowed.count(value) > 0;
}
/**
 * Function: parseStatusSet
 *
 * @brief 解析 “parseStatusSet” 对应的文本或结构，提取经过校验的业务字段。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @param text 待解析、比较、显示或发送的 UTF-8 文本。
 * @return 构造、解析或查找到的结果；所有权与生命周期遵循返回类型。
 */
std::unordered_set<std::string> parseStatusSet(const std::string& text) {
    std::unordered_set<std::string> out;
    try {
        // 支持 GUI 默认的 JSON 数组写法，例如 "[0]"、"[1,2]"。
        json j = json::parse(text);
        if (j.is_array()) {
            for (const auto& v : j) out.insert(jsonString(json{{"v", v}}, "v"));
        }
    } catch (...) {
        // 也兼容用户手写 "0,1,2" 或带方括号/引号的宽松格式。
        std::stringstream ss(text);
        std::string part;
        while (std::getline(ss, part, ',')) {
            part.erase(std::remove_if(part.begin(), part.end(), [](unsigned char ch) {
                return std::isspace(ch) || ch == '[' || ch == ']' || ch == '"';
            }), part.end());
            if (!part.empty()) out.insert(part);
        }
    }
    return out;
}
/**
 * Function: sanitizeFilePart
 *
 * @brief 规范化 “sanitizeFilePart” 对应的字符串或标识，减少格式差异对匹配造成的影响。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @param s 调用方传入的业务参数；其用途与约束由类型、名称及函数体共同定义。
 * @return 构造、解析或查找到的结果；所有权与生命周期遵循返回类型。
 */
std::string sanitizeFilePart(std::string s) {
    for (char& ch : s) {
        if (!(std::isalnum((unsigned char)ch) || ch == '-' || ch == '_' || ch == '.')) ch = '_';
    }
    if (s.empty()) s = "unnamed";
    return s;
}
/**
 * Function: urlEncode
 *
 * @brief 实现外部通信层的 “urlEncode” 功能，服务于 HTTP、RocketMQ 或运行时文件管理。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @param value 待解析、比较、显示或发送的 UTF-8 文本。
 * @return 构造、解析或查找到的结果；所有权与生命周期遵循返回类型。
 */
std::string urlEncode(const std::string& value) {
    static const char* hex = "0123456789ABCDEF";
    std::string out;
    for (unsigned char ch : value) {
        if (std::isalnum(ch) || ch == '-' || ch == '_' || ch == '.' || ch == '~' || ch == '/' || ch == ':') {
            out.push_back((char)ch);
        } else {
            out.push_back('%');
            out.push_back(hex[ch >> 4]);
            out.push_back(hex[ch & 15]);
        }
    }
    return out;
}
/**
 * Function: httpRequest
 *
 * @brief 实现外部通信层的 “httpRequest” 功能，服务于 HTTP、RocketMQ 或运行时文件管理。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @param url 调用方传入的业务参数；其用途与约束由类型、名称及函数体共同定义。
 * @param query 调用方传入的业务参数；其用途与约束由类型、名称及函数体共同定义。
 * @param body 待解析、比较、显示或发送的 UTF-8 文本。
 * @param log 回调函数；用于向界面或日志系统报告进度和结果。
 * @return 构造、解析或查找到的结果；所有权与生命周期遵循返回类型。
 */
std::string httpRequest(const std::string& url,
                        const std::string& query,
                        const std::string& body,
                        PqLogFn log) {
    std::wstring full = utf8ToWide(url + (query.empty() ? "" : ("?" + query)));
    URL_COMPONENTS uc{};
    uc.dwStructSize = sizeof(uc);
    wchar_t host[512]{};
    wchar_t path[4096]{};
    uc.lpszHostName = host;
    uc.dwHostNameLength = (DWORD)std::size(host);
    uc.lpszUrlPath = path;
    uc.dwUrlPathLength = (DWORD)std::size(path);
    uc.dwSchemeLength = (DWORD)-1;
    uc.dwExtraInfoLength = (DWORD)-1;

    if (!WinHttpCrackUrl(full.c_str(), 0, 0, &uc)) {
        throw std::runtime_error("URL 解析失败: " + url);
    }

    std::wstring object(path, uc.dwUrlPathLength);
    if (uc.lpszExtraInfo && uc.dwExtraInfoLength > 0) {
        object.append(uc.lpszExtraInfo, uc.dwExtraInfoLength);
    }

    HINTERNET session = WinHttpOpen(L"pq_tool/1.0",
                                    WINHTTP_ACCESS_TYPE_DEFAULT_PROXY,
                                    WINHTTP_NO_PROXY_NAME,
                                    WINHTTP_NO_PROXY_BYPASS,
                                    0);
    if (!session) throw std::runtime_error("WinHttpOpen 失败");

    HINTERNET connect = nullptr;
    HINTERNET request = nullptr;
    std::string response;
    try {
        connect = WinHttpConnect(session, std::wstring(host, uc.dwHostNameLength).c_str(), uc.nPort, 0);
        if (!connect) throw std::runtime_error("WinHttpConnect 失败");

        DWORD flags = (uc.nScheme == INTERNET_SCHEME_HTTPS) ? WINHTTP_FLAG_SECURE : 0;
        // 这个项目约定：body 为空走 GET，不为空走 POST JSON。
        const wchar_t* method = body.empty() ? L"GET" : L"POST";
        request = WinHttpOpenRequest(connect, method, object.c_str(),
                                     nullptr, WINHTTP_NO_REFERER,
                                     WINHTTP_DEFAULT_ACCEPT_TYPES, flags);
        if (!request) throw std::runtime_error("WinHttpOpenRequest 失败");

        WinHttpSetTimeouts(request, 10000, 10000, 10000, 10000);
        std::wstring headers = L"Content-Type: application/json\r\n";
        BOOL ok = WinHttpSendRequest(
            request,
            headers.c_str(),
            (DWORD)-1,
            body.empty() ? WINHTTP_NO_REQUEST_DATA : (LPVOID)body.data(),
            (DWORD)body.size(),
            (DWORD)body.size(),
            0);
        if (!ok || !WinHttpReceiveResponse(request, nullptr)) {
            throw std::runtime_error("HTTP 请求失败，错误码: " + std::to_string(GetLastError()));
        }

        DWORD status = 0;
        DWORD len = sizeof(status);
        WinHttpQueryHeaders(request,
                            WINHTTP_QUERY_STATUS_CODE | WINHTTP_QUERY_FLAG_NUMBER,
                            WINHTTP_HEADER_NAME_BY_INDEX,
                            &status,
                            &len,
                            WINHTTP_NO_HEADER_INDEX);
        if (status < 200 || status >= 300) {
            throw std::runtime_error("HTTP 状态码异常: " + std::to_string(status));
        }

        DWORD available = 0;
        do {
            if (!WinHttpQueryDataAvailable(request, &available)) {
                throw std::runtime_error("读取 HTTP 响应失败");
            }
            if (!available) break;
            std::string chunk(available, '\0');
            DWORD read = 0;
            if (!WinHttpReadData(request, chunk.data(), available, &read)) {
                throw std::runtime_error("读取 HTTP 响应内容失败");
            }
            chunk.resize(read);
            response += chunk;
        } while (available > 0);
    } catch (...) {
        if (request) WinHttpCloseHandle(request);
        if (connect) WinHttpCloseHandle(connect);
        WinHttpCloseHandle(session);
        throw;
    }

    if (request) WinHttpCloseHandle(request);
    if (connect) WinHttpCloseHandle(connect);
    WinHttpCloseHandle(session);
    if (log) log("HTTP " + std::string(body.empty() ? "GET " : "POST ") + url + " 返回 " + std::to_string(response.size()) + " 字节");
    return response;
}
/**
 * Function: writeTextFile
 *
 * @brief 把 “writeTextFile” 对应的内存数据序列化并写入目标位置。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @param path 文件或目录路径；函数按 UTF-8/Windows 路径规则读取或写入。
 * @param text 待解析、比较、显示或发送的 UTF-8 文本。
 */
void writeTextFile(const fs::path& path, const std::string& text) {
    fs::create_directories(path.parent_path().empty() ? fs::path(".") : path.parent_path());
    std::ofstream ofs(path, std::ios::binary);
    if (!ofs) throw std::runtime_error("无法写入文件: " + path.string());
    ofs << text;
}
/**
 * Function: parseResponseDataArray
 *
 * @brief 解析 “parseResponseDataArray” 对应的文本或结构，提取经过校验的业务字段。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @param raw 待解析、比较、显示或发送的 UTF-8 文本。
 * @param name 待解析、比较、显示或发送的 UTF-8 文本。
 * @return 构造、解析或查找到的结果；所有权与生命周期遵循返回类型。
 */
json parseResponseDataArray(const std::string& raw, const std::string& name) {
    json root = json::parse(raw);
    if (!root.contains("data") || !root["data"].is_array()) {
        throw std::runtime_error(name + " 响应缺少 data 数组");
    }
    return root["data"];
}

std::vector<std::pair<std::string, std::string>> objectFields(
    const json& obj,
    const std::unordered_set<std::string>& skip = {}) {
    std::vector<std::pair<std::string, std::string>> out;
    if (!obj.is_object()) return out;
    for (auto it = obj.begin(); it != obj.end(); ++it) {
        if (skip.count(it.key())) continue;
        json wrapper;
        wrapper["v"] = it.value();
        out.push_back({it.key(), jsonString(wrapper, "v")});
    }
    return out;
}
/**
 * Function: looksLikeXml
 *
 * @brief 判断 “looksLikeXml” 所表达的条件；该函数只读取输入，不负责修改业务状态。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @param text 待解析、比较、显示或发送的 UTF-8 文本。
 * @return 条件成立或操作成功时为 true，否则为 false。
 */
bool looksLikeXml(const std::string& text) {
    std::string s = text;
    s.erase(s.begin(), std::find_if(s.begin(), s.end(), [](unsigned char ch) { return !std::isspace(ch); }));
    return s.rfind("<?xml", 0) == 0 || s.rfind("<", 0) == 0;
}
/**
 * Function: makeGuid
 *
 * @brief 构造 “makeGuid” 对应的对象、消息、窗口控件或输出结构。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @return 构造、解析或查找到的结果；所有权与生命周期遵循返回类型。
 */
std::string makeGuid() {
    GUID guid;
    if (CoCreateGuid(&guid) != S_OK) {
        auto now = std::chrono::high_resolution_clock::now().time_since_epoch().count();
        return "guid_" + std::to_string(now);
    }
    wchar_t buf[64]{};
    StringFromGUID2(guid, buf, 64);
    std::string s = wideToUtf8(buf);
    s.erase(std::remove(s.begin(), s.end(), '{'), s.end());
    s.erase(std::remove(s.begin(), s.end(), '}'), s.end());
    std::transform(s.begin(), s.end(), s.begin(), [](unsigned char ch) { return (char)std::tolower(ch); });
    return s;
}
/**
 * Function: nowIsoMillis
 *
 * @brief 实现外部通信层的 “nowIsoMillis” 功能，服务于 HTTP、RocketMQ 或运行时文件管理。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @return 构造、解析或查找到的结果；所有权与生命周期遵循返回类型。
 */
std::string nowIsoMillis() {
    using namespace std::chrono;
    auto now = system_clock::now();
    auto ms = duration_cast<milliseconds>(now.time_since_epoch()) % 1000;
    std::time_t tt = system_clock::to_time_t(now);
    std::tm tm{};
    localtime_s(&tm, &tt);
    std::ostringstream oss;
    oss << std::put_time(&tm, "%Y-%m-%dT%H:%M:%S")
        << "." << std::setw(3) << std::setfill('0') << ms.count();
    return oss.str();
}
/**
 * Function: nowUnixMillis
 *
 * @brief 实现外部通信层的 “nowUnixMillis” 功能，服务于 HTTP、RocketMQ 或运行时文件管理。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @return 构造、解析或查找到的结果；所有权与生命周期遵循返回类型。
 */
long long nowUnixMillis() {
    return std::chrono::duration_cast<std::chrono::milliseconds>(
               std::chrono::system_clock::now().time_since_epoch())
        .count();
}
/**
 * Function: appendTextFile
 *
 * @brief 向目标结构追加或保证存在 “appendTextFile” 所描述的数据。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @param path 文件或目录路径；函数按 UTF-8/Windows 路径规则读取或写入。
 * @param text 待解析、比较、显示或发送的 UTF-8 文本。
 */
void appendTextFile(const fs::path& path, const std::string& text) {
    fs::create_directories(path.parent_path().empty() ? fs::path(".") : path.parent_path());
    std::ofstream ofs(path, std::ios::binary | std::ios::app);
    if (!ofs) throw std::runtime_error("无法追加写入文件: " + path.string());
    ofs << text;
}
/**
 * Function: nonEmptyCStr
 *
 * @brief 实现外部通信层的 “nonEmptyCStr” 功能，服务于 HTTP、RocketMQ 或运行时文件管理。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @param value 待解析、比较、显示或发送的 UTF-8 文本。
 * @param fallback 调用方传入的业务参数；其用途与约束由类型、名称及函数体共同定义。
 * @return 构造、解析或查找到的结果；所有权与生命周期遵循返回类型。
 */
const char* nonEmptyCStr(const std::string& value, const char* fallback) {
    return value.empty() ? fallback : value.c_str();
}
/**
 * Function: trimCopy
 *
 * @brief 规范化 “trimCopy” 对应的字符串或标识，减少格式差异对匹配造成的影响。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @param value 待解析、比较、显示或发送的 UTF-8 文本。
 * @return 构造、解析或查找到的结果；所有权与生命周期遵循返回类型。
 */
std::string trimCopy(std::string value) {
    value.erase(value.begin(), std::find_if(value.begin(), value.end(), [](unsigned char ch) {
        return !std::isspace(ch);
    }));
    value.erase(std::find_if(value.rbegin(), value.rend(), [](unsigned char ch) {
        return !std::isspace(ch);
    }).base(), value.end());
    return value;
}
/**
 * Function: isWildcardTag
 *
 * @brief 判断 “isWildcardTag” 所表达的条件；该函数只读取输入，不负责修改业务状态。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @param tag 待解析、比较、显示或发送的 UTF-8 文本。
 * @return 条件成立或操作成功时为 true，否则为 false。
 */
bool isWildcardTag(const std::string& tag) {
    std::string t = trimCopy(tag);
    return t.empty() || t == "*";
}
/**
 * Function: subscribeTagExpression
 *
 * @brief 实现外部通信层的 “subscribeTagExpression” 功能，服务于 HTTP、RocketMQ 或运行时文件管理。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @param tag 待解析、比较、显示或发送的 UTF-8 文本。
 * @return 构造、解析或查找到的结果；所有权与生命周期遵循返回类型。
 */
std::string subscribeTagExpression(const std::string& tag) {
    return isWildcardTag(tag) ? "*" : trimCopy(tag);
}
/**
 * Function: mergeSubscribeTags
 *
 * @brief 实现外部通信层的 “mergeSubscribeTags” 功能，服务于 HTTP、RocketMQ 或运行时文件管理。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @param left 调用方传入的业务参数；其用途与约束由类型、名称及函数体共同定义。
 * @param right 调用方传入的业务参数；其用途与约束由类型、名称及函数体共同定义。
 * @return 构造、解析或查找到的结果；所有权与生命周期遵循返回类型。
 */
std::string mergeSubscribeTags(const std::string& left, const std::string& right) {
    std::string a = subscribeTagExpression(left);
    std::string b = subscribeTagExpression(right);
    if (a == "*" || b == "*") return "*";
    if (a == b) return a;
    return a + " || " + b;
}
/**
 * Function: tagMatches
 *
 * @brief 实现外部通信层的 “tagMatches” 功能，服务于 HTTP、RocketMQ 或运行时文件管理。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @param expectedExpression 调用方传入的业务参数；其用途与约束由类型、名称及函数体共同定义。
 * @param actualTag 调用方传入的业务参数；其用途与约束由类型、名称及函数体共同定义。
 * @return 条件成立或操作成功时为 true，否则为 false。
 */
bool tagMatches(const std::string& expectedExpression, const std::string& actualTag) {
    if (isWildcardTag(expectedExpression)) return true;
    std::string actual = trimCopy(actualTag);
    std::string expr = expectedExpression;
    size_t start = 0;
    while (start <= expr.size()) {
        size_t pos = expr.find("||", start);
        std::string part = trimCopy(expr.substr(start, pos == std::string::npos ? std::string::npos : pos - start));
        // RocketMQ C 接口订阅表达式可能是 "tagA || tagB"，消费时也按同样规则过滤。
        if (part == "*" || part == actual) return true;
        if (pos == std::string::npos) break;
        start = pos + 2;
    }
    return false;
}
/**
 * Function: parseLongLong
 *
 * @brief 解析 “parseLongLong” 对应的文本或结构，提取经过校验的业务字段。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @param value 待解析、比较、显示或发送的 UTF-8 文本。
 * @return 构造、解析或查找到的结果；所有权与生命周期遵循返回类型。
 */
long long parseLongLong(const json& value) {
    try {
        if (value.is_number_integer()) return value.get<long long>();
        if (value.is_number_unsigned()) return static_cast<long long>(value.get<unsigned long long>());
        if (value.is_number_float()) return static_cast<long long>(value.get<double>());
        if (value.is_string()) return std::stoll(value.get<std::string>());
    } catch (...) {
    }
    return 0;
}
/**
 * Function: findTimestampValue
 *
 * @brief 在现有数据结构中查找 “findTimestampValue” 对应的对象或位置。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @param value 待解析、比较、显示或发送的 UTF-8 文本。
 * @return 构造、解析或查找到的结果；所有权与生命周期遵循返回类型。
 */
long long findTimestampValue(const json& value) {
    static const std::unordered_set<std::string> keys = {
        "Value_TIME", "valueTime", "timestamp", "timeStamp", "storeTimestamp", "time", "Time"
    };
    if (value.is_object()) {
        // 不同消息体时间字段命名不统一，因此递归查找一组常见 key。
        for (const auto& key : keys) {
            auto it = value.find(key);
            if (it != value.end()) {
                long long v = parseLongLong(*it);
                if (v > 0) return v;
            }
        }
        for (auto it = value.begin(); it != value.end(); ++it) {
            long long v = findTimestampValue(it.value());
            if (v > 0) return v;
        }
    } else if (value.is_array()) {
        for (const auto& item : value) {
            long long v = findTimestampValue(item);
            if (v > 0) return v;
        }
    }
    return 0;
}
/**
 * Function: timestampFromJsonText
 *
 * @brief 实现外部通信层的 “timestampFromJsonText” 功能，服务于 HTTP、RocketMQ 或运行时文件管理。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @param text 待解析、比较、显示或发送的 UTF-8 文本。
 * @return 构造、解析或查找到的结果；所有权与生命周期遵循返回类型。
 */
long long timestampFromJsonText(const std::string& text) {
    try {
        return findTimestampValue(json::parse(text));
    } catch (...) {
        return 0;
    }
}
/**
 * Function: jsonScalarToString
 *
 * @brief 实现外部通信层的 “jsonScalarToString” 功能，服务于 HTTP、RocketMQ 或运行时文件管理。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @param value 待解析、比较、显示或发送的 UTF-8 文本。
 * @return 构造、解析或查找到的结果；所有权与生命周期遵循返回类型。
 */
std::string jsonScalarToString(const json& value) {
    if (value.is_string()) return value.get<std::string>();
    if (value.is_number_integer()) return std::to_string(value.get<long long>());
    if (value.is_number_unsigned()) return std::to_string(value.get<unsigned long long>());
    if (value.is_number_float()) {
        std::ostringstream oss;
        oss << value.get<double>();
        return oss.str();
    }
    if (value.is_boolean()) return value.get<bool>() ? "true" : "false";
    return value.is_null() ? "" : value.dump();
}
/**
 * Function: normalizeDataType
 *
 * @brief 规范化 “normalizeDataType” 对应的字符串或标识，减少格式差异对匹配造成的影响。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @param dataType 调用方传入的业务参数；其用途与约束由类型、名称及函数体共同定义。
 * @return 构造、解析或查找到的结果；所有权与生命周期遵循返回类型。
 */
std::string normalizeDataType(std::string dataType) {
    dataType.erase(std::remove_if(dataType.begin(), dataType.end(),
                                  [](unsigned char ch) { return std::isspace(ch) != 0; }),
                   dataType.end());
    if (dataType.size() == 1 && std::isdigit(static_cast<unsigned char>(dataType[0]))) {
        dataType = "0" + dataType;
    }
    return dataType.empty() ? "unknown" : dataType;
}
/**
 * Function: findDataTypeValue
 *
 * @brief 在现有数据结构中查找 “findDataTypeValue” 对应的对象或位置。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @param value 待解析、比较、显示或发送的 UTF-8 文本。
 * @param out 输出参数；函数在成功或失败时写入结果或诊断信息。
 * @return 条件成立或操作成功时为 true，否则为 false。
 */
bool findDataTypeValue(const json& value, std::string& out) {
    if (value.is_object()) {
        auto it = value.find("DATA_TYPE");
        if (it != value.end()) {
            out = normalizeDataType(jsonScalarToString(*it));
            return true;
        }
        for (auto iter = value.begin(); iter != value.end(); ++iter) {
            if (findDataTypeValue(iter.value(), out)) return true;
        }
    } else if (value.is_array()) {
        for (const auto& item : value) {
            if (findDataTypeValue(item, out)) return true;
        }
    }
    return false;
}
/**
 * Function: dataTypeFromJsonText
 *
 * @brief 实现外部通信层的 “dataTypeFromJsonText” 功能，服务于 HTTP、RocketMQ 或运行时文件管理。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @param text 待解析、比较、显示或发送的 UTF-8 文本。
 * @return 构造、解析或查找到的结果；所有权与生命周期遵循返回类型。
 */
std::string dataTypeFromJsonText(const std::string& text) {
    try {
        std::string dataType;
        if (findDataTypeValue(json::parse(text), dataType)) return dataType;
    } catch (...) {
    }
    return "unknown";
}
/**
 * Function: findReportNameValue
 *
 * @brief 在现有数据结构中查找 “findReportNameValue” 对应的对象或位置。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @param value 待解析、比较、显示或发送的 UTF-8 文本。
 * @param out 输出参数；函数在成功或失败时写入结果或诊断信息。
 * @return 条件成立或操作成功时为 true，否则为 false。
 */
bool findReportNameValue(const json& value, std::string& out) {
    static const char* keys[] = {"rpt_id", "rpt_name", "report_name", "rpt_no"};
    if (value.is_object()) {
        for (const char* key : keys) {
            auto it = value.find(key);
            if (it != value.end()) {
                out = trimCopy(jsonScalarToString(*it));
                if (!out.empty()) return true;
            }
        }
        for (auto iter = value.begin(); iter != value.end(); ++iter) {
            if (findReportNameValue(iter.value(), out)) return true;
        }
    } else if (value.is_array()) {
        for (const auto& item : value) {
            if (findReportNameValue(item, out)) return true;
        }
    }
    return false;
}
/**
 * Function: reportNameFromJsonText
 *
 * @brief 实现外部通信层的 “reportNameFromJsonText” 功能，服务于 HTTP、RocketMQ 或运行时文件管理。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @param text 待解析、比较、显示或发送的 UTF-8 文本。
 * @return 构造、解析或查找到的结果；所有权与生命周期遵循返回类型。
 */
std::string reportNameFromJsonText(const std::string& text) {
    try {
        std::string reportName;
        if (findReportNameValue(json::parse(text), reportName)) return reportName;
    } catch (...) {
    }
    return "";
}

std::string lowerAscii(std::string value);
/**
 * Function: isMonitorIdKey
 *
 * @brief 判断 “isMonitorIdKey” 所表达的条件；该函数只读取输入，不负责修改业务状态。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @param key 待解析、比较、显示或发送的 UTF-8 文本。
 * @return 条件成立或操作成功时为 true，否则为 false。
 */
bool isMonitorIdKey(const std::string& key) {
    std::string k = lowerAscii(trimCopy(key));
    return k == "monitor" ||
           k == "monitorid" ||
           k == "monitor_id" ||
           k == "mp_id" ||
           k == "mpid" ||
           k == "measurepointid" ||
           k == "measure_point_id" ||
           k == "measureid" ||
           k == "measure_id";
}
/**
 * Function: looksLikeJsonText
 *
 * @brief 判断 “looksLikeJsonText” 所表达的条件；该函数只读取输入，不负责修改业务状态。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @param text 待解析、比较、显示或发送的 UTF-8 文本。
 * @return 条件成立或操作成功时为 true，否则为 false。
 */
bool looksLikeJsonText(const std::string& text) {
    std::string s = trimCopy(text);
    return !s.empty() && (s.front() == '{' || s.front() == '[');
}
/**
 * Function: findMonitorIdValue
 *
 * @brief 在现有数据结构中查找 “findMonitorIdValue” 对应的对象或位置。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @param value 待解析、比较、显示或发送的 UTF-8 文本。
 * @param out 输出参数；函数在成功或失败时写入结果或诊断信息。
 * @return 条件成立或操作成功时为 true，否则为 false。
 */
bool findMonitorIdValue(const json& value, std::string& out) {
    if (value.is_object()) {
        for (auto it = value.begin(); it != value.end(); ++it) {
            if (isMonitorIdKey(it.key())) {
                out = trimCopy(jsonScalarToString(it.value()));
                if (!out.empty()) return true;
            }
        }
        for (auto it = value.begin(); it != value.end(); ++it) {
            if ((it.key() == "messageBody" || it.key() == "body") && it.value().is_string()) {
                // RocketMQ 外层 JSON 的 messageBody 常是字符串化 JSON，需要再 parse 一次。
                std::string embedded = it.value().get<std::string>();
                if (looksLikeJsonText(embedded)) {
                    try {
                        if (findMonitorIdValue(json::parse(embedded), out)) return true;
                    } catch (...) {
                    }
                }
            }
            if (findMonitorIdValue(it.value(), out)) return true;
        }
    } else if (value.is_array()) {
        for (const auto& item : value) {
            if (findMonitorIdValue(item, out)) return true;
        }
    }
    return false;
}
/**
 * Function: monitorIdFromJsonText
 *
 * @brief 实现外部通信层的 “monitorIdFromJsonText” 功能，服务于 HTTP、RocketMQ 或运行时文件管理。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @param text 待解析、比较、显示或发送的 UTF-8 文本。
 * @return 构造、解析或查找到的结果；所有权与生命周期遵循返回类型。
 */
std::string monitorIdFromJsonText(const std::string& text) {
    try {
        std::string monitorId;
        if (findMonitorIdValue(json::parse(text), monitorId)) return monitorId;
    } catch (...) {
    }
    return "";
}
/**
 * Function: findGuidValue
 *
 * @brief 在现有数据结构中查找 “findGuidValue” 对应的对象或位置。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @param value 待解析、比较、显示或发送的 UTF-8 文本。
 * @param out 输出参数；函数在成功或失败时写入结果或诊断信息。
 * @return 条件成立或操作成功时为 true，否则为 false。
 */
bool findGuidValue(const json& value, std::string& out) {
    if (value.is_object()) {
        for (auto it = value.begin(); it != value.end(); ++it) {
            std::string key = lowerAscii(trimCopy(it.key()));
            if (key == "guid" || key == "traceguid" || key == "trace_guid") {
                out = trimCopy(jsonScalarToString(it.value()));
                if (!out.empty()) return true;
            }
        }
        for (auto it = value.begin(); it != value.end(); ++it) {
            if ((it.key() == "messageBody" || it.key() == "body") && it.value().is_string()) {
                std::string embedded = it.value().get<std::string>();
                if (looksLikeJsonText(embedded)) {
                    try {
                        if (findGuidValue(json::parse(embedded), out)) return true;
                    } catch (...) {
                    }
                }
            }
            if (findGuidValue(it.value(), out)) return true;
        }
    } else if (value.is_array()) {
        for (const auto& item : value) {
            if (findGuidValue(item, out)) return true;
        }
    }
    return false;
}
/**
 * Function: guidFromJsonText
 *
 * @brief 实现外部通信层的 “guidFromJsonText” 功能，服务于 HTTP、RocketMQ 或运行时文件管理。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @param text 待解析、比较、显示或发送的 UTF-8 文本。
 * @return 构造、解析或查找到的结果；所有权与生命周期遵循返回类型。
 */
std::string guidFromJsonText(const std::string& text) {
    try {
        std::string guid;
        if (findGuidValue(json::parse(text), guid)) return guid;
        json root = json::parse(text);
        if (root.is_object()) {
            auto it = root.find("key");
            if (it != root.end()) return trimCopy(jsonScalarToString(*it));
        }
    } catch (...) {
    }
    return "";
}
/**
 * Function: normalizeMonitorId
 *
 * @brief 规范化 “normalizeMonitorId” 对应的字符串或标识，减少格式差异对匹配造成的影响。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @param value 待解析、比较、显示或发送的 UTF-8 文本。
 * @return 构造、解析或查找到的结果；所有权与生命周期遵循返回类型。
 */
std::string normalizeMonitorId(std::string value) {
    value = trimCopy(value);
    value.erase(std::remove_if(value.begin(), value.end(), [](unsigned char ch) {
        return std::isspace(ch) != 0 || ch == '"' || ch == '\'' || ch == '{' || ch == '}';
    }), value.end());
    return lowerAscii(value);
}
/**
 * Function: safePathPart
 *
 * @brief 规范化 “safePathPart” 对应的字符串或标识，减少格式差异对匹配造成的影响。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @param value 待解析、比较、显示或发送的 UTF-8 文本。
 * @param fallback 调用方传入的业务参数；其用途与约束由类型、名称及函数体共同定义。
 * @return 构造、解析或查找到的结果；所有权与生命周期遵循返回类型。
 */
std::string safePathPart(std::string value, const std::string& fallback) {
    if (value.empty()) value = fallback;
    for (char& ch : value) {
        unsigned char c = static_cast<unsigned char>(ch);
        if (c < 32 || ch == '<' || ch == '>' || ch == ':' || ch == '"' ||
            ch == '/' || ch == '\\' || ch == '|' || ch == '?' || ch == '*') {
            ch = '_';
        }
    }
    while (!value.empty() && (value.back() == '.' || value.back() == ' ')) value.pop_back();
    if (value.empty()) value = fallback.empty() ? "unnamed" : fallback;
    if (value.size() > 96) value.resize(96);
    return value;
}
/**
 * Function: firstNonEmpty
 *
 * @brief 实现外部通信层的 “firstNonEmpty” 功能，服务于 HTTP、RocketMQ 或运行时文件管理。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @param a 调用方传入的业务参数；其用途与约束由类型、名称及函数体共同定义。
 * @param b 调用方传入的业务参数；其用途与约束由类型、名称及函数体共同定义。
 * @return 构造、解析或查找到的结果；所有权与生命周期遵循返回类型。
 */
std::string firstNonEmpty(const std::string& a, const std::string& b) {
    return !a.empty() ? a : b;
}
/**
 * Function: prettyJsonOrRaw
 *
 * @brief 实现外部通信层的 “prettyJsonOrRaw” 功能，服务于 HTTP、RocketMQ 或运行时文件管理。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @param text 待解析、比较、显示或发送的 UTF-8 文本。
 * @return 构造、解析或查找到的结果；所有权与生命周期遵循返回类型。
 */
std::string prettyJsonOrRaw(const std::string& text) {
    try {
        return json::parse(text).dump(2);
    } catch (...) {
        return text;
    }
}
/**
 * Function: eventTitle
 *
 * @brief 实现外部通信层的 “eventTitle” 功能，服务于 HTTP、RocketMQ 或运行时文件管理。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @param prefix 调用方传入的业务参数；其用途与约束由类型、名称及函数体共同定义。
 * @param path 文件或目录路径；函数按 UTF-8/Windows 路径规则读取或写入。
 * @return 构造、解析或查找到的结果；所有权与生命周期遵循返回类型。
 */
std::string eventTitle(const std::string& prefix, const std::string& path) {
    fs::path p = fs::u8path(path);
    std::string name = p.filename().u8string();
    return prefix + " " + nowIsoMillis() + (name.empty() ? "" : (" " + name));
}
/**
 * Function: existingOriginCount
 *
 * @brief 实现外部通信层的 “existingOriginCount” 功能，服务于 HTTP、RocketMQ 或运行时文件管理。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @param dir 文件或目录路径；函数按 UTF-8/Windows 路径规则读取或写入。
 * @return 整数状态、索引或计数；入口函数以 0 表示成功。
 */
int existingOriginCount(const fs::path& dir) {
    int count = 0;
    if (!fs::exists(dir)) return 0;
    for (const auto& entry : fs::directory_iterator(dir)) {
        if (!entry.is_regular_file()) continue;
        std::string name = entry.path().filename().u8string();
        if (name.rfind("origin", 0) == 0 && entry.path().extension() == ".txt") ++count;
    }
    return count;
}
/**
 * Function: lowerAscii
 *
 * @brief 规范化 “lowerAscii” 对应的字符串或标识，减少格式差异对匹配造成的影响。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @param value 待解析、比较、显示或发送的 UTF-8 文本。
 * @return 构造、解析或查找到的结果；所有权与生命周期遵循返回类型。
 */
std::string lowerAscii(std::string value) {
    std::transform(value.begin(), value.end(), value.begin(),
                   [](unsigned char ch) { return static_cast<char>(std::tolower(ch)); });
    return value;
}
/**
 * Function: isTraceTextRecord
 *
 * @brief 判断 “isTraceTextRecord” 所表达的条件；该函数只读取输入，不负责修改业务状态。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @param path 文件或目录路径；函数按 UTF-8/Windows 路径规则读取或写入。
 * @return 条件成立或操作成功时为 true，否则为 false。
 */
bool isTraceTextRecord(const fs::path& path) {
    std::string ext = lowerAscii(path.extension().u8string());
    if (ext == ".txt" || ext == ".json" || ext == ".jsonl" || ext == ".log") return true;
    return false;
}
/**
 * Function: clearTraceTextRecords
 *
 * @brief 实现外部通信层的 “clearTraceTextRecords” 功能，服务于 HTTP、RocketMQ 或运行时文件管理。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @param dir 文件或目录路径；函数按 UTF-8/Windows 路径规则读取或写入。
 */
void clearTraceTextRecords(const fs::path& dir) {
    if (!fs::exists(dir)) return;
    for (const auto& entry : fs::directory_iterator(dir)) {
        if (!entry.is_regular_file()) continue;
        if (isTraceTextRecord(entry.path())) {
            std::error_code ec;
            fs::remove(entry.path(), ec);
        }
    }
}

struct CProducer;
struct CPushConsumer;
struct CMessage;
struct CMessageExt;

struct CSendResult {
    int sendStatus;
    char msgId[256];
    long long offset;
};

/*
 * RocketMqRuntime 是进程内唯一的 RocketMQ 适配器。
 *
 * 设计原因：
 *   1. RocketMQ C DLL 通过 LoadLibrary 动态加载，运行时可能不存在；
 *   2. GUI 会多次点击“初始化 MQ”，需要避免重复启动/热重启导致 DLL 崩溃；
 *   3. MQ 回调在 SDK 线程触发，必须用互斥锁保护会话状态，再通过 event_ 通知 GUI。
 */
class RocketMqRuntime {
    struct DetachedHandles {
        CProducer* producer = nullptr;
        CPushConsumer* consumer = nullptr;
        HMODULE module = nullptr;
        int (*ShutdownProducer)(CProducer*) = nullptr;
        int (*DestroyProducer)(CProducer*) = nullptr;
        int (*ShutdownPushConsumer)(CPushConsumer*) = nullptr;
        int (*DestroyPushConsumer)(CPushConsumer*) = nullptr;
    };

    struct TraceSession {
        std::string monitorId;
        std::string monitorName;
        std::string traceDir;
        std::string xmlPath;
        std::string frontType;
        std::string guid;
        long long startedAtMs = 0; // 点击“数据追踪”的时间；更早的 MQ 存量消息不得进入本次会话。
        std::unordered_set<std::string> receivedDataTypes; // 同一次追踪中每个 DATA_TYPE 只保留一次。
        std::unordered_set<std::string> receivedTraceReports; // 同一次追踪中每个 rpt_id 只保留第一条。
        long long dataTimestamp = 0;
        int originCounter = 0;
    };
public:
    /**
     * @brief 释放 RocketMQ 生产者、消费者和动态库句柄。
     * @details 先在互斥锁保护下用 detachLocked() 把共享句柄移出运行时对象，再在锁外
     *          调用 shutdownDetached()；这样既避免回调线程看到半释放状态，也避免 SDK
     *          的阻塞式 Shutdown/Destroy 调用长时间占用 mu_。该析构函数由单例在进程
     *          退出时自动执行。
     */
    ~RocketMqRuntime() {
        DetachedHandles old;
        {
            std::lock_guard<std::mutex> lock(mu_);
            old = detachLocked();
        }
        shutdownDetached(old);
    }
    /**
     * Function: initialize
     *
     * @brief 实现外部通信层的 “initialize” 功能，服务于 HTTP、RocketMQ 或运行时文件管理。
     * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
     * @param cfg 运行配置；提供接口地址、MQ 参数、路径和当前数据模式。
     * @param log 回调函数；用于向界面或日志系统报告进度和结果。
     * @param event 回调函数；用于向界面或日志系统报告进度和结果。
     * @return 条件成立或操作成功时为 true，否则为 false。
     */
    bool initialize(const PqToolConfig& cfg, PqLogFn log, PqMqEventFn event) {
        DetachedHandles old;
        {
            std::lock_guard<std::mutex> lock(mu_);
            // 如果生产/消费连接参数未变，只刷新配置和回调；避免重复 StartProducer/StartPushConsumer。
            bool sameRuntimeConfig = cfg.producer == cfg_.producer &&
                                     cfg.ipport == cfg_.ipport &&
                                     cfg.producerAccessKey == cfg_.producerAccessKey &&
                                     cfg.producerSecretKey == cfg_.producerSecretKey &&
                                     cfg.consumer == cfg_.consumer &&
                                     cfg.consumerIpport == cfg_.consumerIpport &&
                                     cfg.consumerAccessKey == cfg_.consumerAccessKey &&
                                     cfg.consumerSecretKey == cfg_.consumerSecretKey &&
                                     cfg.dataStatTopic == cfg_.dataStatTopic &&
                                     cfg.dataRealtimeTopic == cfg_.dataRealtimeTopic &&
                                     cfg.dataStatTag == cfg_.dataStatTag &&
                                     cfg.dataRealtimeTag == cfg_.dataRealtimeTag &&
                                     cfg.traceTopic == cfg_.traceTopic &&
                                     cfg.traceTag == cfg_.traceTag;
            if (active_ && producer_ && consumer_ && sameRuntimeConfig) {
                cfg_ = cfg;
                event_ = std::move(event);
                if (log) log("RocketMQ 已初始化，已刷新界面回调和当前配置。");
                return true;
            }
            if (active_ && producer_ && consumer_) {
                // 连接参数变了也不在进程内重启底层 SDK，提示用户重启工具后生效。
                cfg_ = cfg;
                event_ = std::move(event);
                if (log) {
                    log("MQ 配置已保存；当前 RocketMQ 连接已运行。");
                    log("为避免 RocketMQ SDK 热关闭/重建导致程序崩溃，本次不在进程内重启底层连接。");
                    log("如果修改了 nameserver、生产者/消费者组、Topic 或订阅 Tag，请关闭工具后重新打开并初始化 MQ 后生效。");
                }
                return true;
            }
            old = detachLocked();
        }

        if (old.producer || old.consumer || old.module) {
            if (log) log("MQ 配置变更，正在重新初始化 RocketMQ...");
            shutdownDetachedAsync(old);
        }

        std::lock_guard<std::mutex> lock(mu_);
        cfg_ = cfg;
        event_ = std::move(event);
        dataTimestamp_ = 0;
        originCounter_ = 0;
        currentTraceDir_.clear();
        currentMonitorId_.clear();
        currentMonitorName_.clear();
        sessions_.clear();
        sessionsByGuid_.clear();
        lastError_.clear();

        if (!loadLibraryLocked()) {
            if (log) log("RocketMQ DLL 未加载: " + lastError_);
            return false;
        }

        producer_ = CreateProducer(nonEmptyCStr(cfg_.producer, "TRACE_PRODUCER"));
        if (!producer_) return failLocked("CreateProducer 返回空指针", log);

        if (!okLocked(SetProducerNameServerAddress(producer_, nonEmptyCStr(cfg_.ipport, "127.0.0.1:9876")),
                      "SetProducerNameServerAddress", log)) {
            return false;
        }
        SetProducerSessionCredentials(producer_,
                                      nonEmptyCStr(cfg_.producerAccessKey, ""),
                                      nonEmptyCStr(cfg_.producerSecretKey, ""),
                                      "");
        fs::create_directories(runtimeLogsDir());
        std::string mqLogDir = runtimeLogsDir().u8string();
        SetProducerLogPath(producer_, mqLogDir.c_str());
        if (!okLocked(StartProducer(producer_), "StartProducer", log)) return false;

        consumer_ = CreatePushConsumer(nonEmptyCStr(cfg_.consumer, "TRACE_CONSUMER"));
        if (!consumer_) return failLocked("CreatePushConsumer 返回空指针", log);

        if (!okLocked(SetPushConsumerNameServerAddress(consumer_, nonEmptyCStr(cfg_.consumerIpport, "127.0.0.1:9876")),
                      "SetPushConsumerNameServerAddress", log)) {
            return false;
        }
        SetPushConsumerSessionCredentials(consumer_,
                                          nonEmptyCStr(cfg_.consumerAccessKey, ""),
                                          nonEmptyCStr(cfg_.consumerSecretKey, ""),
                                          "");
        SetPushConsumerLogPath(consumer_, mqLogDir.c_str());
        SetPushConsumerThreadCount(consumer_, 2);

        // 同一个 Topic 可能同时配置统计/实时 Tag，这里合并成 RocketMQ 支持的 "A || B" 表达式。
        std::map<std::string, std::string> subscriptions;
        auto addSubscription = [&](const std::string& topic, const std::string& tag) {
            if (topic.empty()) return;
            auto it = subscriptions.find(topic);
            if (it == subscriptions.end()) {
                subscriptions[topic] = subscribeTagExpression(tag);
            } else {
                it->second = mergeSubscribeTags(it->second, tag);
            }
        };
        addSubscription(cfg_.dataStatTopic, cfg_.dataStatTag);
        addSubscription(cfg_.dataRealtimeTopic, cfg_.dataRealtimeTag);
        addSubscription(cfg_.traceTopic, cfg_.traceTag);

        bool subscribed = false;
        for (const auto& sub : subscriptions) {
            if (!okLocked(Subscribe(consumer_, sub.first.c_str(), sub.second.c_str()),
                          "Subscribe Topic", log)) {
                return false;
            }
            subscribed = true;
            if (log) log("Subscribed Topic by Tag only: " + sub.first + " [" + sub.second + "]");
        }
        if (!subscribed) return failLocked("DATATopic/TraceTopic 为空，未订阅任何 Topic", log);

        if (!okLocked(RegisterMessageCallback(consumer_, &RocketMqRuntime::messageCallback),
                      "RegisterMessageCallback", log)) {
            return false;
        }
        if (!okLocked(StartPushConsumer(consumer_), "StartPushConsumer", log)) return false;

        active_ = true;
        if (log) {
            log("RocketMQ 已连接，生产者和消费者初始化完成。");
            log("DATATopic -> " + (cfg_.jsonDataPath.empty() ? std::string("jsondata.txt") : cfg_.jsonDataPath)
                + "，TraceTopic -> " + originOutputPath().u8string());
        }
        return true;
    }
    /**
     * Function: setTraceTarget
     *
     * @brief 把 “setTraceTarget” 对应的变更应用到模型、控件或输出文档。
     * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
     * @param cfg 运行配置；提供接口地址、MQ 参数、路径和当前数据模式。
     * @param device 调用方传入的业务参数；其用途与约束由类型、名称及函数体共同定义。
     * @param monitor 调用方传入的业务参数；其用途与约束由类型、名称及函数体共同定义。
     * @param frontType 调用方传入的业务参数；其用途与约束由类型、名称及函数体共同定义。
     * @param guid 索引、数量、范围或偏移参数；具体边界由函数体校验。
     */
    void setTraceTarget(const PqToolConfig& cfg,
                        const PqLedgerDevice& device,
                        const PqLedgerMonitor& monitor,
                        const std::string& frontType,
                        const std::string& guid) {
        const long long startedAtMs = nowUnixMillis();
        std::lock_guard<std::mutex> lock(mu_);
        cfg_ = cfg;
        // 记录当前追踪目标。后续 MQ 回调会用 monitorId 或 guid 把消息路由到对应 TraceSession。
        currentTraceDir_ = pqTraceDirectoryFor(cfg, device, monitor);
        currentXmlPath_ = pqFindDeviceXmlPath(cfg, device);
        currentMonitorId_ = monitor.id;
        currentMonitorName_ = monitor.name;
        fs::path dir = fs::u8path(currentTraceDir_);
        fs::create_directories(dir);
        clearTraceTextRecords(dir);
        dataTimestamp_ = 0;
        originCounter_ = 0;
        std::string monitorKey = normalizeMonitorId(monitor.id);
        // 一次按钮点击只允许一个活动会话。否则旧测点后续到达的消息仍会写入旧会话目录。
        sessions_.clear();
        sessionsByGuid_.clear();
        TraceSession& session = sessions_[monitorKey];
        session.monitorId = monitor.id;
        session.monitorName = monitor.name;
        session.traceDir = currentTraceDir_;
        session.xmlPath = currentXmlPath_;
        session.frontType = frontType;
        session.guid = guid;
        session.startedAtMs = startedAtMs;
        session.receivedDataTypes.clear();
        session.receivedTraceReports.clear();
        session.dataTimestamp = 0;
        session.originCounter = 0;
        if (!guid.empty()) sessionsByGuid_[normalizeMonitorId(guid)] = monitorKey;

        // GUI 自动解析优先读取这个上下文，避免用户手动维护 json/origin/output 路径。
        json context = {
            {"deviceId", device.id},
            {"deviceName", device.name},
            {"devType", device.devType},
            {"processNo", device.processNo},
            {"monitorId", monitor.id},
            {"monitorName", monitor.name},
            {"guid", guid},
            {"frontType", frontType},
            {"startedAt", nowIsoMillis()},
            {"startedAtMs", startedAtMs},
            {"xmlPath", currentXmlPath_},
            {"jsonDataPath", (dir / "jsondata_01.txt").u8string()},
            {"jsonDataPaths", json::array({
                (dir / "jsondata_01.txt").u8string(),
                (dir / "jsondata_02.txt").u8string(),
                (dir / "jsondata_03.txt").u8string()
            })},
            {"outputPath", (dir / "final_sorted.xls").u8string()}
        };
        writeTextFile(dir / "trace_context.json", context.dump(2));
    }
    /**
     * Function: sendPayload
     *
     * @brief 执行 “sendPayload” 对应的外部通信，并把结果交给日志、事件或落盘流程。
     * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
     * @param topic 待解析、比较、显示或发送的 UTF-8 文本。
     * @param tag 待解析、比较、显示或发送的 UTF-8 文本。
     * @param messageKey 调用方传入的业务参数；其用途与约束由类型、名称及函数体共同定义。
     * @param payload 调用方传入的业务参数；其用途与约束由类型、名称及函数体共同定义。
     * @param label 待解析、比较、显示或发送的 UTF-8 文本。
     * @param err 输出参数；函数在成功或失败时写入结果或诊断信息。
     * @param log 回调函数；用于向界面或日志系统报告进度和结果。
     * @return 条件成立或操作成功时为 true，否则为 false。
     */
    bool sendPayload(const std::string& topic,
                     const std::string& tag,
                     const std::string& messageKey,
                     const std::string& payload,
                     const std::string& label,
                     std::string& err,
                     PqLogFn log) {
        std::lock_guard<std::mutex> lock(mu_);
        err.clear();
        if (!active_ || !producer_) {
            err = lastError_.empty() ? "RocketMQ 未初始化" : lastError_;
            return false;
        }

        CMessage* msg = CreateMessage(nonEmptyCStr(topic, ""));
        if (!msg) {
            err = "CreateMessage 返回空指针";
            return false;
        }

        SetMessageTags(msg, nonEmptyCStr(tag, ""));
        SetMessageKeys(msg, nonEmptyCStr(messageKey, ""));
        SetMessageBody(msg, payload.c_str());

        CSendResult result{};
        int rc = SendMessageSync(producer_, msg, &result);
        DestroyMessage(msg);
        if (rc != 0) {
            err = "SendMessageSync 返回错误码 " + std::to_string(rc);
            return false;
        }

        if (log) {
            log("RocketMQ " + label + "消息已发送，Topic=" + topic
                + "，MsgId=" + std::string(result.msgId));
        }
        return true;
    }

private:
    using MessageCallback = int (*)(CPushConsumer*, CMessageExt*);
    /**
     * Function: messageCallback
     *
     * @brief 实现外部通信层的 “messageCallback” 功能，服务于 HTTP、RocketMQ 或运行时文件管理。
     * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
     * @param CPushConsumer 调用方传入的业务参数；其用途与约束由类型、名称及函数体共同定义。
     * @param msg 调用方传入的业务参数；其用途与约束由类型、名称及函数体共同定义。
     * @return 整数状态、索引或计数；入口函数以 0 表示成功。
     */
    static int messageCallback(CPushConsumer*, CMessageExt* msg) {
        return instance().onMessage(msg);
    }
    /**
     * Function: findSessionForMessageLocked
     *
     * @brief 在现有数据结构中查找 “findSessionForMessageLocked” 对应的对象或位置。
     * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
     * @param body 待解析、比较、显示或发送的 UTF-8 文本。
     * @param key 待解析、比较、显示或发送的 UTF-8 文本。
     * @param actualMonitor 调用方传入的业务参数；其用途与约束由类型、名称及函数体共同定义。
     * @param reason 调用方传入的业务参数；其用途与约束由类型、名称及函数体共同定义。
     * @return 构造、解析或查找到的结果；所有权与生命周期遵循返回类型。
     */
    TraceSession* findSessionForMessageLocked(const std::string& body,
                                              const std::string& key,
                                              std::string& actualMonitor,
                                              std::string& reason) {
        actualMonitor.clear();
        reason.clear();

        // 优先用消息体中的测点 ID 路由；没有测点 ID 时再用 guid/key 路由。
        actualMonitor = monitorIdFromJsonText(body);
        std::string normalized = normalizeMonitorId(actualMonitor);
        if (!normalized.empty()) {
            auto it = sessions_.find(normalized);
            if (it != sessions_.end()) return &it->second;
            reason = "monitor not being traced, actual=" + actualMonitor;
        }

        std::string guid = guidFromJsonText(body);
        if (guid.empty()) guid = trimCopy(key);
        std::string normalizedGuid = normalizeMonitorId(guid);
        if (!normalizedGuid.empty()) {
            auto git = sessionsByGuid_.find(normalizedGuid);
            if (git != sessionsByGuid_.end()) {
                auto sit = sessions_.find(git->second);
                if (sit != sessions_.end()) return &sit->second;
            }
            if (reason.empty()) reason = "guid not being traced, actual=" + guid;
        }

        if (reason.empty()) reason = "missing monitor id/guid";
        return nullptr;
    }
    /**
     * Function: onMessage
     *
     * @brief 处理 “onMessage” 对应的业务入口或事件，并调度下游函数完成实际工作。
     * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
     * @param msg 调用方传入的业务参数；其用途与约束由类型、名称及函数体共同定义。
     * @return 整数状态、索引或计数；入口函数以 0 表示成功。
     */
    int onMessage(CMessageExt* msg) {
        std::string topic = GetMessageTopic && GetMessageTopic(msg) ? GetMessageTopic(msg) : "";
        std::string tag = GetMessageTags && GetMessageTags(msg) ? GetMessageTags(msg) : "";
        std::string key = GetMessageKeys && GetMessageKeys(msg) ? GetMessageKeys(msg) : "";
        std::string body = GetMessageBody && GetMessageBody(msg) ? GetMessageBody(msg) : "";
        long long storeTs = GetMessageStoreTimestamp ? GetMessageStoreTimestamp(msg) : 0;

        std::lock_guard<std::mutex> lock(mu_);
        try {
            appendTextFile(runtimeLogPath("mq_messages.log"), nowIsoMillis()
                           + " received topic=" + topic
                           + " tag=" + tag
                           + " key=" + key
                           + " bodyLen=" + std::to_string(body.size()) + "\n");
            // 先按 Topic/Tag 判断消息类型，再按 monitorId/guid 过滤到当前追踪会话。
            std::string dataFrontType = dataFrontTypeForMessageLocked(topic, tag);
            bool isData = !dataFrontType.empty();
            bool isTrace = topic == cfg_.traceTopic && tagMatches(cfg_.traceTag, tag);
            TraceSession* session = nullptr;
            if (isData || isTrace) {
                std::string actualMonitor;
                std::string reason;
                session = findSessionForMessageLocked(body, key, actualMonitor, reason);
                if (!session) {
                    appendTextFile(runtimeLogPath("mq_messages.log"), nowIsoMillis()
                                   + " skipped " + (isData ? "DATATopic" : "TraceTopic")
                                   + " by monitor router: " + reason
                                   + " topic=" + topic
                                   + " tag=" + tag + "\n");
                    return 0;
                }
                if (isData && !session->frontType.empty() && session->frontType != dataFrontType) {
                    appendTextFile(runtimeLogPath("mq_messages.log"), nowIsoMillis()
                                   + " skipped DATATopic by frontType router: session="
                                   + session->frontType + ", message=" + dataFrontType
                                   + " topic=" + topic + " tag=" + tag + "\n");
                    return 0;
                }
                // PushConsumer 可能在点击后继续投递积压消息。必须在任何去重、落盘和 GUI 事件之前
                // 使用 Broker 存储时间做会话边界判断，保证 jsondata 与 origin 都来自本次追踪。
                if (storeTs <= 0 || storeTs < session->startedAtMs) {
                    appendTextFile(runtimeLogPath("mq_messages.log"), nowIsoMillis()
                                   + " skipped " + (isData ? "DATATopic" : "TraceTopic")
                                   + " before trace session: storeTs=" + std::to_string(storeTs)
                                   + ", startedAtMs=" + std::to_string(session->startedAtMs)
                                   + ", monitor=" + session->monitorId
                                   + " topic=" + topic + " tag=" + tag + "\n");
                    return 0;
                }
            }
            if (isData) {
                std::string dataType = dataTypeFromJsonText(body);
                std::string normalizedType = normalizeDataType(dataType);
                // 同一 DATA_TYPE 重复到达时跳过，避免后来的旧数据覆盖已解析文件。
                if (!session->receivedDataTypes.insert(normalizedType).second) {
                    appendTextFile(runtimeLogPath("mq_messages.log"), nowIsoMillis()
                                   + " skipped DATATopic duplicate DATA_TYPE=" + normalizedType
                                   + " monitor=" + session->monitorId
                                   + " topic=" + topic + " tag=" + tag + "\n");
                    return 0;
                }
                fs::path out = jsonDataOutputPathLocked(*session, dataType);
                writeTextFile(out, body);
                long long bodyTs = timestampFromJsonText(body);
                session->dataTimestamp = bodyTs > 0 ? bodyTs : storeTs;
                if (normalizeMonitorId(session->monitorId) == normalizeMonitorId(currentMonitorId_)) {
                    dataTimestamp_ = session->dataTimestamp;
                }
                appendTextFile(runtimeLogPath("mq_messages.log"), nowIsoMillis() + " DATATopic DATA_TYPE=" + dataType + " tag=" + tag + " -> " + out.u8string() + "\n");
                if (event_) event_({"data", eventTitle("DATATopic", out.u8string()), body, out.u8string(), session->monitorId, session->traceDir});
            } else if (isTrace) {
                long long bodyTs = timestampFromJsonText(body);
                // 如果 TraceTopic 的时间早于已经收到的 DATATopic，认为是旧追踪结果，避免串入本次追踪。
                if (session->dataTimestamp > 0 && bodyTs > 0 && bodyTs < session->dataTimestamp) {
                    appendTextFile(runtimeLogPath("mq_messages.log"), nowIsoMillis() + " TraceTopic tag=" + tag + " skipped by timestamp\n");
                } else {
                    std::string reportName = reportNameFromJsonText(body);
                    std::string reportKey = lowerAscii(trimCopy(reportName));
                    if (reportKey.empty()) reportKey = "__unknown_report__";
                    // TopicAsk 的 limit=20 可能返回同一 rpt_id 的多份历史采样；本工具每次点击只取第一份。
                    if (!session->receivedTraceReports.insert(reportKey).second) {
                        appendTextFile(runtimeLogPath("mq_messages.log"), nowIsoMillis()
                                       + " skipped TraceTopic duplicate report=" + reportKey
                                       + " monitor=" + session->monitorId
                                       + " topic=" + topic + " tag=" + tag + "\n");
                        return 0;
                    }
                    fs::path out = nextOriginOutputPathLocked(*session, reportName);
                    writeTextFile(out, body);
                    appendTextFile(runtimeLogPath("mq_messages.log"), nowIsoMillis() + " TraceTopic tag=" + tag + " -> " + out.u8string() + "\n");
                    if (event_) event_({"trace", eventTitle("TraceTopic", out.u8string()), body, out.u8string(), session->monitorId, session->traceDir});
                }
            } else {
                appendTextFile(runtimeLogPath("mq_messages.log"), nowIsoMillis() + " skipped topic=" + topic + " tag=" + tag + "\n");
            }
        } catch (...) {
            return 1;
        }
        return 0;
    }
    /**
     * Function: instance
     *
     * @brief 实现外部通信层的 “instance” 功能，服务于 HTTP、RocketMQ 或运行时文件管理。
     * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
     * @return 构造、解析或查找到的结果；所有权与生命周期遵循返回类型。
     */
    static RocketMqRuntime& instance() {
        static RocketMqRuntime runtime;
        return runtime;
    }

    friend RocketMqRuntime& rocketRuntime();
    /**
     * Function: originOutputPath
     *
     * @brief 实现外部通信层的 “originOutputPath” 功能，服务于 HTTP、RocketMQ 或运行时文件管理。
     * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
     * @return 构造、解析或查找到的结果；所有权与生命周期遵循返回类型。
     */
    fs::path originOutputPath() const {
        if (!cfg_.originPaths.empty() && !cfg_.originPaths.front().empty()) {
            return fs::u8path(cfg_.originPaths.front());
        }
        return fs::u8path("origin.txt");
    }
    /**
     * Function: dataFrontTypeForMessageLocked
     *
     * @brief 实现外部通信层的 “dataFrontTypeForMessageLocked” 功能，服务于 HTTP、RocketMQ 或运行时文件管理。
     * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
     * @param topic 待解析、比较、显示或发送的 UTF-8 文本。
     * @param tag 待解析、比较、显示或发送的 UTF-8 文本。
     * @return 构造、解析或查找到的结果；所有权与生命周期遵循返回类型。
     */
    std::string dataFrontTypeForMessageLocked(const std::string& topic, const std::string& tag) const {
        if (topic == cfg_.dataStatTopic && tagMatches(cfg_.dataStatTag, tag)) return "cfg_stat_data";
        if (topic == cfg_.dataRealtimeTopic && tagMatches(cfg_.dataRealtimeTag, tag)) return "cfg_3s_data";
        return "";
    }
    /**
     * Function: jsonDataOutputPathLocked
     *
     * @brief 实现外部通信层的 “jsonDataOutputPathLocked” 功能，服务于 HTTP、RocketMQ 或运行时文件管理。
     * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
     * @param session 调用方传入的业务参数；其用途与约束由类型、名称及函数体共同定义。
     * @param dataType 调用方传入的业务参数；其用途与约束由类型、名称及函数体共同定义。
     * @return 构造、解析或查找到的结果；所有权与生命周期遵循返回类型。
     */
    fs::path jsonDataOutputPathLocked(const TraceSession& session, const std::string& dataType) const {
        std::string suffix = normalizeDataType(dataType);
        std::string filename = "jsondata_" + safePathPart(suffix, "unknown") + ".txt";
        if (!session.traceDir.empty()) return fs::u8path(session.traceDir) / fs::u8path(filename);
        if (!currentTraceDir_.empty()) return fs::u8path(currentTraceDir_) / fs::u8path(filename);
        if (!cfg_.jsonDataPath.empty()) {
            std::vector<std::string> paths;
            std::stringstream ss(cfg_.jsonDataPath);
            std::string part;
            while (std::getline(ss, part)) {
                part = trimCopy(part);
                if (!part.empty()) paths.push_back(part);
            }
            fs::path base = paths.empty() ? fs::u8path(cfg_.jsonDataPath) : fs::u8path(paths.front());
            fs::path dir = base.has_parent_path() ? base.parent_path() : fs::path(".");
            return dir / fs::u8path(filename);
        }
        return runtimeJsonPath(filename);
    }
    /**
     * Function: nextOriginOutputPathLocked
     *
     * @brief 实现外部通信层的 “nextOriginOutputPathLocked” 功能，服务于 HTTP、RocketMQ 或运行时文件管理。
     * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
     * @param session 调用方传入的业务参数；其用途与约束由类型、名称及函数体共同定义。
     * @param reportName 调用方传入的业务参数；其用途与约束由类型、名称及函数体共同定义。
     * @return 构造、解析或查找到的结果；所有权与生命周期遵循返回类型。
     */
    fs::path nextOriginOutputPathLocked(TraceSession& session, const std::string& reportName) {
        if (session.traceDir.empty() && currentTraceDir_.empty()) return originOutputPath();
        std::string safeReport = reportName.empty() ? "" : safePathPart(reportName, "report");
        std::string baseName;
        if (safeReport.empty()) {
            ++session.originCounter;
            baseName = "origin" + std::to_string(session.originCounter);
        } else {
            baseName = "origin_" + safeReport;
        }

        fs::path dir = session.traceDir.empty() ? fs::u8path(currentTraceDir_) : fs::u8path(session.traceDir);
        fs::path out = dir / fs::u8path(baseName + ".txt");
        int suffix = 2;
        while (fs::exists(out)) {
            out = dir / fs::u8path(baseName + "_" + std::to_string(suffix++) + ".txt");
        }
        return out;
    }
    /**
     * Function: failLocked
     *
     * @brief 实现外部通信层的 “failLocked” 功能，服务于 HTTP、RocketMQ 或运行时文件管理。
     * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
     * @param message 输出参数；函数在成功或失败时写入结果或诊断信息。
     * @param log 回调函数；用于向界面或日志系统报告进度和结果。
     * @return 条件成立或操作成功时为 true，否则为 false。
     */
    bool failLocked(const std::string& message, PqLogFn log) {
        lastError_ = message;
        shutdownLocked();
        if (log) log("RocketMQ 初始化失败: " + message);
        return false;
    }
    /**
     * Function: okLocked
     *
     * @brief 实现外部通信层的 “okLocked” 功能，服务于 HTTP、RocketMQ 或运行时文件管理。
     * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
     * @param rc 调用方传入的业务参数；其用途与约束由类型、名称及函数体共同定义。
     * @param op 调用方传入的业务参数；其用途与约束由类型、名称及函数体共同定义。
     * @param log 回调函数；用于向界面或日志系统报告进度和结果。
     * @return 条件成立或操作成功时为 true，否则为 false。
     */
    bool okLocked(int rc, const char* op, PqLogFn log) {
        if (rc == 0) return true;
        return failLocked(std::string(op) + " 返回错误码 " + std::to_string(rc), log);
    }
    /**
     * Function: loadLibraryLocked
     *
     * @brief 读取或加载 “loadLibraryLocked” 对应的数据，并转换为本模块后续逻辑使用的内存结构。
     * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
     * @return 条件成立或操作成功时为 true，否则为 false。
     */
    bool loadLibraryLocked() {
        if (module_) return true;

        // 兼容不同部署目录和 DLL 命名；LoadLibraryEx 用于带路径 DLL 时顺带搜索其依赖。
        std::vector<std::string> candidates = {
            "rocketmq-client-cpp.dll",
            "rocketmq.dll",
            "librocketmq.dll",
            "bin\\rocketmq-client-cpp.dll",
            "bin\\rocketmq.dll",
            "bin\\librocketmq.dll",
            "rocketmq-client-cpp-2.2.0-source-release\\bin\\rocketmq-client-cpp.dll",
            "rocketmq-client-cpp-2.2.0-source-release\\bin\\rocketmq.dll",
            "rocketmq-client-cpp-2.2.0-source-release\\bin\\librocketmq.dll",
            "Win32\\Debug\\rocketmq-client-cpp.dll",
            "Win32\\Release\\rocketmq-client-cpp.dll",
            "Win32\\x86\\Debug\\rocketmq-client-cpp.dll",
            "Win32\\x86\\Release\\rocketmq-client-cpp.dll",
            "rocketmq-client-cpp-2.2.0-source-release\\Win32\\Debug\\rocketmq-client-cpp.dll",
            "rocketmq-client-cpp-2.2.0-source-release\\Win32\\Release\\rocketmq-client-cpp.dll",
            "rocketmq-client-cpp-2.2.0-source-release\\Win32\\x86\\Debug\\rocketmq-client-cpp.dll",
            "rocketmq-client-cpp-2.2.0-source-release\\Win32\\x86\\Release\\rocketmq-client-cpp.dll",
            "Win32\\x64\\Debug\\rocketmq-client-cpp.dll",
            "Win32\\x64\\Release\\rocketmq-client-cpp.dll",
            "rocketmq-client-cpp-2.2.0-source-release\\Win32\\x64\\Debug\\rocketmq-client-cpp.dll",
            "rocketmq-client-cpp-2.2.0-source-release\\Win32\\x64\\Release\\rocketmq-client-cpp.dll"
        };

        DWORD lastErr = 0;
        std::string tried;
        for (const auto& path : candidates) {
            bool exists = fs::exists(fs::u8path(path));
            module_ = path.find('\\') == std::string::npos
                          ? LoadLibraryA(path.c_str())
                          : LoadLibraryExA(path.c_str(), nullptr, LOAD_WITH_ALTERED_SEARCH_PATH);
            if (module_) {
                loadedFrom_ = path;
                break;
            }
            lastErr = GetLastError();
            if (!tried.empty()) tried += ", ";
            tried += path;
            tried += exists ? "(load error=" + std::to_string(lastErr) + ")" : "(not found)";
        }

        if (!module_) {
            lastError_ = "No loadable RocketMQ DLL. Tried: " + tried
                       + ". LastError=" + std::to_string(lastErr)
                       + ". Error 126 means the DLL itself or one of its dependent DLLs is missing. "
                         "This pq_tool.exe is 32-bit, so RocketMQ must be built as x86.";
            return false;
            lastError_ = "未找到可加载的 RocketMQ DLL，已尝试: " + tried
                       + "，GetLastError=" + std::to_string(lastErr)
                       + "。当前 exe 是 32 位，DLL 也需要是 x86。";
            return false;
        }

        std::vector<std::string> missing;
#define LOAD_MQ_FN(name) \
        do { \
            name = reinterpret_cast<decltype(name)>(GetProcAddress(module_, #name)); \
            if (!name) missing.push_back(#name); \
        } while (0)
        LOAD_MQ_FN(CreateProducer);
        LOAD_MQ_FN(DestroyProducer);
        LOAD_MQ_FN(StartProducer);
        LOAD_MQ_FN(ShutdownProducer);
        LOAD_MQ_FN(SetProducerNameServerAddress);
        LOAD_MQ_FN(SetProducerSessionCredentials);
        LOAD_MQ_FN(SetProducerLogPath);
        LOAD_MQ_FN(CreateMessage);
        LOAD_MQ_FN(DestroyMessage);
        LOAD_MQ_FN(SetMessageTags);
        LOAD_MQ_FN(SetMessageKeys);
        LOAD_MQ_FN(SetMessageBody);
        LOAD_MQ_FN(SendMessageSync);
        LOAD_MQ_FN(CreatePushConsumer);
        LOAD_MQ_FN(DestroyPushConsumer);
        LOAD_MQ_FN(StartPushConsumer);
        LOAD_MQ_FN(ShutdownPushConsumer);
        LOAD_MQ_FN(SetPushConsumerNameServerAddress);
        LOAD_MQ_FN(SetPushConsumerSessionCredentials);
        LOAD_MQ_FN(SetPushConsumerLogPath);
        LOAD_MQ_FN(SetPushConsumerThreadCount);
        LOAD_MQ_FN(Subscribe);
        LOAD_MQ_FN(RegisterMessageCallback);
        LOAD_MQ_FN(GetMessageTopic);
        LOAD_MQ_FN(GetMessageTags);
        LOAD_MQ_FN(GetMessageKeys);
        LOAD_MQ_FN(GetMessageBody);
        LOAD_MQ_FN(GetMessageStoreTimestamp);
#undef LOAD_MQ_FN

        if (!missing.empty()) {
            std::ostringstream oss;
            for (size_t i = 0; i < missing.size(); ++i) {
                if (i) oss << ", ";
                oss << missing[i];
            }
            lastError_ = "RocketMQ DLL 缺少 C 接口导出: " + oss.str();
            FreeLibrary(module_);
            module_ = nullptr;
            return false;
        }

        return true;
    }
    /**
     * Function: detachLocked
     *
     * @brief 实现外部通信层的 “detachLocked” 功能，服务于 HTTP、RocketMQ 或运行时文件管理。
     * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
     * @return 构造、解析或查找到的结果；所有权与生命周期遵循返回类型。
     */
    DetachedHandles detachLocked() {
        DetachedHandles h;
        h.producer = producer_;
        h.consumer = consumer_;
        h.module = module_;
        h.ShutdownProducer = ShutdownProducer;
        h.DestroyProducer = DestroyProducer;
        h.ShutdownPushConsumer = ShutdownPushConsumer;
        h.DestroyPushConsumer = DestroyPushConsumer;
        active_ = false;
        producer_ = nullptr;
        consumer_ = nullptr;
        module_ = nullptr;
        loadedFrom_.clear();
        return h;
    }
    /**
     * Function: shutdownDetached
     *
     * @brief 实现外部通信层的 “shutdownDetached” 功能，服务于 HTTP、RocketMQ 或运行时文件管理。
     * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
     * @param h 调用方传入的业务参数；其用途与约束由类型、名称及函数体共同定义。
     */
    void shutdownDetached(const DetachedHandles& h) {
        if (h.consumer) {
            if (h.ShutdownPushConsumer) h.ShutdownPushConsumer(h.consumer);
            if (h.DestroyPushConsumer) h.DestroyPushConsumer(h.consumer);
        }
        if (h.producer) {
            if (h.ShutdownProducer) h.ShutdownProducer(h.producer);
            if (h.DestroyProducer) h.DestroyProducer(h.producer);
        }
        if (h.module) FreeLibrary(h.module);
    }
    /**
     * Function: shutdownDetachedAsync
     *
     * @brief 实现外部通信层的 “shutdownDetachedAsync” 功能，服务于 HTTP、RocketMQ 或运行时文件管理。
     * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
     * @param h 调用方传入的业务参数；其用途与约束由类型、名称及函数体共同定义。
     */
    void shutdownDetachedAsync(DetachedHandles h) {
        std::thread([h]() {
            try {
                if (h.consumer) {
                    if (h.ShutdownPushConsumer) h.ShutdownPushConsumer(h.consumer);
                    if (h.DestroyPushConsumer) h.DestroyPushConsumer(h.consumer);
                }
                if (h.producer) {
                    if (h.ShutdownProducer) h.ShutdownProducer(h.producer);
                    if (h.DestroyProducer) h.DestroyProducer(h.producer);
                }
                if (h.module) FreeLibrary(h.module);
            } catch (...) {
            }
        }).detach();
    }
    /**
     * Function: shutdownLocked
     *
     * @brief 实现外部通信层的 “shutdownLocked” 功能，服务于 HTTP、RocketMQ 或运行时文件管理。
     * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
     */
    void shutdownLocked() {
        active_ = false;
        if (consumer_) {
            ShutdownPushConsumer(consumer_);
            DestroyPushConsumer(consumer_);
            consumer_ = nullptr;
        }
        if (producer_) {
            ShutdownProducer(producer_);
            DestroyProducer(producer_);
            producer_ = nullptr;
        }
        if (module_) {
            FreeLibrary(module_);
            module_ = nullptr;
        }
    }

    std::mutex mu_;
    PqToolConfig cfg_;
    PqMqEventFn event_;
    HMODULE module_ = nullptr;
    std::string loadedFrom_;
    std::string lastError_;
    std::string currentTraceDir_;
    std::string currentXmlPath_;
    std::string currentMonitorId_;
    std::string currentMonitorName_;
    std::unordered_map<std::string, TraceSession> sessions_;
    std::unordered_map<std::string, std::string> sessionsByGuid_;
    bool active_ = false;
    long long dataTimestamp_ = 0;
    int originCounter_ = 0;
    CProducer* producer_ = nullptr;
    CPushConsumer* consumer_ = nullptr;

    CProducer* (*CreateProducer)(const char*) = nullptr;
    int (*DestroyProducer)(CProducer*) = nullptr;
    int (*StartProducer)(CProducer*) = nullptr;
    int (*ShutdownProducer)(CProducer*) = nullptr;
    int (*SetProducerNameServerAddress)(CProducer*, const char*) = nullptr;
    int (*SetProducerSessionCredentials)(CProducer*, const char*, const char*, const char*) = nullptr;
    int (*SetProducerLogPath)(CProducer*, const char*) = nullptr;
    CMessage* (*CreateMessage)(const char*) = nullptr;
    int (*DestroyMessage)(CMessage*) = nullptr;
    int (*SetMessageTags)(CMessage*, const char*) = nullptr;
    int (*SetMessageKeys)(CMessage*, const char*) = nullptr;
    int (*SetMessageBody)(CMessage*, const char*) = nullptr;
    int (*SendMessageSync)(CProducer*, CMessage*, CSendResult*) = nullptr;
    CPushConsumer* (*CreatePushConsumer)(const char*) = nullptr;
    int (*DestroyPushConsumer)(CPushConsumer*) = nullptr;
    int (*StartPushConsumer)(CPushConsumer*) = nullptr;
    int (*ShutdownPushConsumer)(CPushConsumer*) = nullptr;
    int (*SetPushConsumerNameServerAddress)(CPushConsumer*, const char*) = nullptr;
    int (*SetPushConsumerSessionCredentials)(CPushConsumer*, const char*, const char*, const char*) = nullptr;
    int (*SetPushConsumerLogPath)(CPushConsumer*, const char*) = nullptr;
    int (*SetPushConsumerThreadCount)(CPushConsumer*, int) = nullptr;
    int (*Subscribe)(CPushConsumer*, const char*, const char*) = nullptr;
    int (*RegisterMessageCallback)(CPushConsumer*, MessageCallback) = nullptr;
    const char* (*GetMessageTopic)(CMessageExt*) = nullptr;
    const char* (*GetMessageTags)(CMessageExt*) = nullptr;
    const char* (*GetMessageKeys)(CMessageExt*) = nullptr;
    const char* (*GetMessageBody)(CMessageExt*) = nullptr;
    long long (*GetMessageStoreTimestamp)(CMessageExt*) = nullptr;
};
/**
 * Function: rocketRuntime
 *
 * @brief 实现外部通信层的 “rocketRuntime” 功能，服务于 HTTP、RocketMQ 或运行时文件管理。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @return 构造、解析或查找到的结果；所有权与生命周期遵循返回类型。
 */
RocketMqRuntime& rocketRuntime() {
    return RocketMqRuntime::instance();
}
} // namespace
/**
 * Function: pqDefaultConfig
 *
 * @brief 实现外部通信层的 “pqDefaultConfig” 功能，服务于 HTTP、RocketMQ 或运行时文件管理。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @return 构造、解析或查找到的结果；所有权与生命周期遵循返回类型。
 */
PqToolConfig pqDefaultConfig() {
    PqToolConfig cfg;
    // 这里只覆盖需要根据 exe 位置动态计算的路径，以及 MQ 常用默认参数。
    cfg.producer = "TRACE_PRODUCER";
    cfg.consumer = "TRACE_CONSUMER";
    cfg.dataStatTopic = "MQ_INST_1697676490574298_Bm3DLamM%LN_Topic";
    cfg.dataRealtimeTopic = "MQ_INST_1697676490574298_Bm3DLamM%Real_Time_Data_Topic";
    cfg.dataStatTag = "6c2adb324940e19e208d717f4970f2ee";
    cfg.dataStatKey = "Test_Keys";
    cfg.dataRealtimeTag = "6c2adb324940e19e208d717f4970f2ee";
    cfg.dataRealtimeKey = "Test_Keys";
    cfg.dataTopic = cfg.dataStatTopic;
    cfg.dataTag = cfg.dataStatTag;
    cfg.dataKey = cfg.dataStatKey;
    cfg.xmlSaveDir = (runtimeDir() / "downloaded_xml").u8string();
    cfg.traceRootDir = (runtimeDir() / "trace_data").u8string();
    cfg.jsonDataPath = runtimeJsonPath("jsondata.txt").u8string();
    cfg.outputPath = (runtimeDir() / "final_sorted.xls").u8string();
    return cfg;
}
/**
 * Function: pqTraceDirectoryFor
 *
 * @brief 实现外部通信层的 “pqTraceDirectoryFor” 功能，服务于 HTTP、RocketMQ 或运行时文件管理。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @param cfg 运行配置；提供接口地址、MQ 参数、路径和当前数据模式。
 * @param device 调用方传入的业务参数；其用途与约束由类型、名称及函数体共同定义。
 * @param monitor 调用方传入的业务参数；其用途与约束由类型、名称及函数体共同定义。
 * @return 构造、解析或查找到的结果；所有权与生命周期遵循返回类型。
 */
std::string pqTraceDirectoryFor(const PqToolConfig& cfg,
                                const PqLedgerDevice& device,
                                const PqLedgerMonitor& monitor) {
    fs::path root = fs::u8path(cfg.traceRootDir.empty() ? "trace_data" : cfg.traceRootDir);
    // 路径中同时放名称和 id，既方便人读，也保证同名测点不互相覆盖。
    std::string proc = safePathPart(firstNonEmpty(device.processNo, "1"), "1");
    std::string dev = safePathPart(firstNonEmpty(device.name, device.id), "device");
    std::string mon = safePathPart(firstNonEmpty(monitor.name, monitor.id), "monitor");
    if (!device.id.empty()) dev += "_" + safePathPart(device.id, "id");
    if (!monitor.id.empty()) mon += "_" + safePathPart(monitor.id, "id");
    fs::path dir = root / fs::u8path("process_" + proc) /
                   fs::u8path("device_" + dev) /
                   fs::u8path("monitor_" + mon);
    return dir.u8string();
}
/**
 * Function: pqClearTraceRecords
 *
 * @brief 实现外部通信层的 “pqClearTraceRecords” 功能，服务于 HTTP、RocketMQ 或运行时文件管理。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @param cfg 运行配置；提供接口地址、MQ 参数、路径和当前数据模式。
 * @param device 调用方传入的业务参数；其用途与约束由类型、名称及函数体共同定义。
 * @param monitor 调用方传入的业务参数；其用途与约束由类型、名称及函数体共同定义。
 */
void pqClearTraceRecords(const PqToolConfig& cfg,
                         const PqLedgerDevice& device,
                         const PqLedgerMonitor& monitor) {
    fs::path dir = fs::u8path(pqTraceDirectoryFor(cfg, device, monitor));
    fs::create_directories(dir);
    clearTraceTextRecords(dir);
}
/**
 * Function: pqFindDeviceXmlPath
 *
 * @brief 实现外部通信层的 “pqFindDeviceXmlPath” 功能，服务于 HTTP、RocketMQ 或运行时文件管理。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @param cfg 运行配置；提供接口地址、MQ 参数、路径和当前数据模式。
 * @param device 调用方传入的业务参数；其用途与约束由类型、名称及函数体共同定义。
 * @return 构造、解析或查找到的结果；所有权与生命周期遵循返回类型。
 */
std::string pqFindDeviceXmlPath(const PqToolConfig& cfg, const PqLedgerDevice& device) {
    if (!device.xmlPath.empty() && fs::exists(fs::u8path(device.xmlPath))) return device.xmlPath;

    // 下载 ICD 后会把接口响应缓存到 runtime\json\icd_response.json，这里用 devType 反查本地 XML。
    fs::path saveDir = fs::u8path(cfg.xmlSaveDir.empty() ? "downloaded_xml" : cfg.xmlSaveDir);
    auto existing = [&](const std::string& name) -> std::string {
        if (name.empty()) return "";
        fs::path p = saveDir / fs::u8path(name);
        if (fs::exists(p)) return p.u8string();
        return "";
    };

    fs::path icdCache = runtimeJsonPath("icd_response.json");
    if (fs::exists(icdCache)) {
        try {
            std::string raw;
            {
                std::ifstream ifs(icdCache, std::ios::binary);
                std::ostringstream ss;
                ss << ifs.rdbuf();
                raw = ss.str();
            }
            json arr = parseResponseDataArray(raw, "ICD");
            for (const auto& model : arr) {
                if (jsonString(model, "devType") != device.devType) continue;
                std::string byId = existing(jsonString(model, "id") + ".xml");
                if (!byId.empty()) return byId;
                std::string fileName = jsonString(model, "fileName");
                std::string byFile = existing(fileName + ".xml");
                if (!byFile.empty()) return byFile;
            }
        } catch (...) {
        }
    }

    std::vector<std::string> guesses = {
        device.devType + ".xml",
        device.devKey + ".xml",
        device.id + ".xml"
    };
    for (const auto& g : guesses) {
        std::string p = existing(g);
        if (!p.empty()) return p;
    }
    if (!cfg.xmlPath.empty() && fs::exists(fs::u8path(cfg.xmlPath))) return cfg.xmlPath;
    return "";
}
/**
 * Function: pqFetchLedgerAndIcd
 *
 * @brief 实现外部通信层的 “pqFetchLedgerAndIcd” 功能，服务于 HTTP、RocketMQ 或运行时文件管理。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @param cfg 运行配置；提供接口地址、MQ 参数、路径和当前数据模式。
 * @param devices 调用方传入的业务参数；其用途与约束由类型、名称及函数体共同定义。
 * @param firstXmlPath 文件或目录路径；函数按 UTF-8/Windows 路径规则读取或写入。
 * @param log 回调函数；用于向界面或日志系统报告进度和结果。
 * @return 条件成立或操作成功时为 true，否则为 false。
 */
bool pqFetchLedgerAndIcd(const PqToolConfig& cfg,
                         std::vector<PqLedgerDevice>& devices,
                         std::string& firstXmlPath,
                         PqLogFn log) {
    devices.clear();
    firstXmlPath.clear();

    // 1. 请求台账：frontIp + terminalStatus/runFlag。
    json ledgerReq;
    ledgerReq["ip"] = cfg.frontIp;
    try {
        ledgerReq["runFlag"] = json::parse(cfg.terminalStatus);
    } catch (...) {
        ledgerReq["runFlag"] = cfg.terminalStatus;
    }

    if (log) log("获取台账: " + ledgerReq.dump());
    std::string ledgerRaw = httpRequest(cfg.webDevice, "", ledgerReq.dump(), log);
    writeTextFile(runtimeJsonPath("ledger_response.json"), ledgerRaw);

    std::unordered_set<std::string> terminalAllowed = parseStatusSet(cfg.terminalStatus);
    std::set<std::string> devTypes;

    // 2. 解析装置和 monitorData 测点，并按 terminalStatus 过滤装置。
    json data = parseResponseDataArray(ledgerRaw, "台账");
    for (const auto& item : data) {
        PqLedgerDevice dev;
        dev.id = jsonString(item, "id");
        dev.name = jsonString(item, "name");
        dev.ip = jsonString(item, "ip");
        dev.status = jsonString(item, "status");
        dev.devType = jsonString(item, "devType");
        dev.devKey = jsonString(item, "devKey");
        dev.orgName = jsonString(item, "org_name");
        dev.stationName = jsonString(item, "stationName");
        dev.subName = jsonString(item, "subName");
        dev.manufacturer = jsonString(item, "manufacturer");
        dev.port = jsonString(item, "port");
        dev.series = jsonString(item, "series");
        dev.updateTime = jsonString(item, "updateTime");
        dev.processNo = jsonString(item, "processNo");
        dev.maxProcessNum = jsonString(item, "maxProcessNum");
        dev.fields = objectFields(item, {"monitorData"});
        if (!setContains(terminalAllowed, dev.status)) continue;
        if (!dev.devType.empty()) devTypes.insert(dev.devType);

        auto monitors = item.find("monitorData");
        if (monitors != item.end() && monitors->is_array()) {
            for (const auto& m : *monitors) {
                PqLedgerMonitor mon;
                mon.id = jsonString(m, "id");
                mon.name = jsonString(m, "name");
                mon.lineNo = jsonString(m, "lineNo");
                mon.voltageLevel = jsonString(m, "voltageLevel");
                mon.connectType = jsonString(m, "ptType");
                mon.status = jsonString(m, "status");
                mon.fields = objectFields(m);
                dev.monitors.push_back(mon);
            }
        }
        devices.push_back(dev);
    }

    if (log) log("台账解析完成: 装置 " + std::to_string(devices.size()) + " 个");

    // 3. 请求 ICD 列表。icdFlag=1 时只请求本次台账中出现的 devType。
    json icdReq = json::array();
    if (cfg.icdFlag == "1") {
        for (const auto& t : devTypes) icdReq.push_back(t);
    }
    if (log) log("获取 ICD 列表: " + icdReq.dump());
    std::string icdRaw = httpRequest(cfg.webIcd, "", icdReq.dump(), log);
    writeTextFile(runtimeJsonPath("icd_response.json"), icdRaw);

    fs::path saveDir = fs::u8path(cfg.xmlSaveDir.empty() ? "downloaded_xml" : cfg.xmlSaveDir);
    fs::create_directories(saveDir);
    json icdData = parseResponseDataArray(icdRaw, "ICD");
    int downloaded = 0;
    std::unordered_map<std::string, std::string> devTypeToXml;
    // 4. 按 filePath 下载 XML，保存到 xmlSaveDir，并建立 devType -> XML 路径映射。
    for (const auto& model : icdData) {
        std::string modelId = jsonString(model, "id");
        std::string devType = jsonString(model, "devType");
        std::string fileName = jsonString(model, "fileName");
        std::string filePath = jsonString(model, "filePath");
        if (filePath.empty()) continue;

        std::string localName = !modelId.empty() ? (sanitizeFilePart(modelId) + ".xml")
                              : !fileName.empty() ? sanitizeFilePart(fileName)
                              : (sanitizeFilePart(devType) + ".xml");
        fs::path localPath = saveDir / fs::u8path(localName);
        std::string xml = httpRequest(cfg.webFiledownload, "filePath=" + urlEncode(filePath), "", log);
        if (!looksLikeXml(xml)) {
            if (log) log("跳过非 XML 下载结果: " + filePath);
            continue;
        }
        writeTextFile(localPath, xml);
        if (!devType.empty() && !devTypeToXml.count(devType)) devTypeToXml[devType] = localPath.u8string();
        if (firstXmlPath.empty()) firstXmlPath = localPath.u8string();
        ++downloaded;
        if (log) log("已下载 XML: " + localPath.u8string());
    }

    if (log) log("ICD/XML 下载完成: " + std::to_string(downloaded) + " 个文件");
    for (auto& dev : devices) {
        auto it = devTypeToXml.find(dev.devType);
        if (it != devTypeToXml.end()) dev.xmlPath = it->second;
    }
    return true;
}
/**
 * Function: pqInitializeMq
 *
 * @brief 实现外部通信层的 “pqInitializeMq” 功能，服务于 HTTP、RocketMQ 或运行时文件管理。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @param cfg 运行配置；提供接口地址、MQ 参数、路径和当前数据模式。
 * @param log 回调函数；用于向界面或日志系统报告进度和结果。
 * @param event 回调函数；用于向界面或日志系统报告进度和结果。
 * @return 条件成立或操作成功时为 true，否则为 false。
 */
bool pqInitializeMq(const PqToolConfig& cfg, PqLogFn log, PqMqEventFn event) {
    // 初始化时先把界面配置落盘，便于用户核对，也便于离线排查 MQ 参数。
    json saved = {
        {"producer", cfg.producer},
        {"ipport", cfg.ipport},
        {"producerAccessKey", cfg.producerAccessKey},
        {"producerSecretKey", cfg.producerSecretKey},
        {"topicLOG", cfg.topicLOG},
        {"tagLOG", cfg.tagLOG},
        {"keyLOG", cfg.keyLOG},
        {"topicAsk", cfg.topicAsk},
        {"tagAsk", cfg.tagAsk},
        {"keyAsk", cfg.keyAsk},
        {"consumer", cfg.consumer},
        {"consumerIpport", cfg.consumerIpport},
        {"consumerAccessKey", cfg.consumerAccessKey},
        {"consumerSecretKey", cfg.consumerSecretKey},
        {"dataStatTopic", cfg.dataStatTopic},
        {"dataStatTag", cfg.dataStatTag},
        {"dataStatKey", cfg.dataStatKey},
        {"dataRealtimeTopic", cfg.dataRealtimeTopic},
        {"dataRealtimeTag", cfg.dataRealtimeTag},
        {"dataRealtimeKey", cfg.dataRealtimeKey},
        {"activeDataTopic", cfg.dataTopic},
        {"dataTag", cfg.dataTag},
        {"dataKey", cfg.dataKey},
        {"traceTopic", cfg.traceTopic},
        {"traceTag", cfg.traceTag},
        {"traceKey", cfg.traceKey}
    };
    writeTextFile(runtimeJsonPath("mq_runtime_config.json"), saved.dump(2));
    if (log) log("MQ config saved to " + runtimeJsonPath("mq_runtime_config.json").u8string());
    if (!rocketRuntime().initialize(cfg, log, event) && log) {
        // 这里仍返回 true，让 GUI 可以继续构造追踪请求；发送失败时会进入 outbox 兜底。
        log("RocketMQ runtime is not active; trace messages will be written to mq_outbox_*.jsonl.");
        log("RocketMQ source is present, but a 32-bit rocketmq-client-cpp.dll or rocketmq.dll is still required.");
    }
    return true;
    if (log) {
        log("MQ 配置已初始化并写入 mq_runtime_config.json");
        log("当前 Windows 编译未链接 RocketMQ SDK，发送消息会先落地到 mq_outbox_*.jsonl。");
        log("接入 RocketMQ DLL 后只需要替换 mq.cpp 中 pqInitializeMq/pqSendTraceRequest 的适配实现。");
    }
    return true;
}
/**
 * Function: pqBuildTracePayload
 *
 * @brief 实现外部通信层的 “pqBuildTracePayload” 功能，服务于 HTTP、RocketMQ 或运行时文件管理。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @param cfg 运行配置；提供接口地址、MQ 参数、路径和当前数据模式。
 * @param device 调用方传入的业务参数；其用途与约束由类型、名称及函数体共同定义。
 * @param monitor 调用方传入的业务参数；其用途与约束由类型、名称及函数体共同定义。
 * @param frontType 调用方传入的业务参数；其用途与约束由类型、名称及函数体共同定义。
 * @return 构造、解析或查找到的结果；所有权与生命周期遵循返回类型。
 */
std::string pqBuildTracePayload(const PqToolConfig& cfg,
                                const PqLedgerDevice& device,
                                const PqLedgerMonitor& monitor,
                                const std::string& frontType) {
    std::string guid = makeGuid();
    int processNo = 0;
    if (frontType == "cfg_3s_data") {
        // 实时数据进程约定为 0；统计数据使用台账中的 processNo。
        processNo = 0;
    } else {
        try {
            processNo = device.processNo.empty() ? 0 : std::stoi(device.processNo);
        } catch (...) {
            processNo = 0;
        }
    }

    // body 是前置识别的 set_log 指令；outer 是 RocketMQ 业务包装。
    json body = {
        {"code", "set_log"},
        {"nodeId", cfg.frontInst},
        {"guid", guid},
        {"processNo", processNo},
        {"id", monitor.id},
        {"level", "measurepoint"},
        {"grade", "TRACE"},
        {"logtype", 1},
        {"frontType", frontType}
    };

    json outer = {
        {"key", guid},
        {"source", "WEB"},
        {"sendTime", nowIsoMillis()},
        {"retryTimes", 0},
        {"messageBody", body.dump()}
    };
    return outer.dump();
}
/**
 * Function: buildRealtimeAskPayload
 *
 * @brief 构造 “buildRealtimeAskPayload” 对应的对象、消息、窗口控件或输出结构。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @param device 调用方传入的业务参数；其用途与约束由类型、名称及函数体共同定义。
 * @param monitor 调用方传入的业务参数；其用途与约束由类型、名称及函数体共同定义。
 * @return 构造、解析或查找到的结果；所有权与生命周期遵循返回类型。
 */
std::string buildRealtimeAskPayload(const PqLedgerDevice& device,
                                    const PqLedgerMonitor& monitor) {
    const std::string guid = makeGuid();
    json body = {
        {"devSeries", device.id},
        {"limit", 20},
        {"line", monitor.id},
        {"realData", true},
        {"soeData", true}
    };
    json outer = {
        {"key", guid},
        {"source", "WEB"},
        {"sendTime", nowIsoMillis()},
        {"retryTimes", 0},
        {"messageBody", body.dump()}
    };
    return outer.dump();
}
/**
 * Function: pqSendTraceRequest
 *
 * @brief 实现外部通信层的 “pqSendTraceRequest” 功能，服务于 HTTP、RocketMQ 或运行时文件管理。
 * @details 本文件组合使用 WinHTTP、Windows 动态库 API、RocketMQ C 接口、nlohmann::json 与 std::filesystem；异常会转换为日志、错误字符串或离线 outbox。共享 MQ 会话状态由互斥锁保护。
 * @param cfg 运行配置；提供接口地址、MQ 参数、路径和当前数据模式。
 * @param device 调用方传入的业务参数；其用途与约束由类型、名称及函数体共同定义。
 * @param monitor 调用方传入的业务参数；其用途与约束由类型、名称及函数体共同定义。
 * @param frontType 调用方传入的业务参数；其用途与约束由类型、名称及函数体共同定义。
 * @param log 回调函数；用于向界面或日志系统报告进度和结果。
 * @param event 回调函数；用于向界面或日志系统报告进度和结果。
 * @return 条件成立或操作成功时为 true，否则为 false。
 */
bool pqSendTraceRequest(const PqToolConfig& cfg,
                        const PqLedgerDevice& device,
                        const PqLedgerMonitor& monitor,
                        const std::string& frontType,
                        PqLogFn log,
                        PqMqEventFn event) {
    std::string payload = pqBuildTracePayload(cfg, device, monitor, frontType);
    std::string guid = guidFromJsonText(payload);
    // 先设置追踪目标，确保追踪请求发出后立即到达的 MQ 回调也能正确路由。
    rocketRuntime().setTraceTarget(cfg, device, monitor, frontType, guid);
    fs::path traceDir = fs::u8path(pqTraceDirectoryFor(cfg, device, monitor));
    fs::create_directories(traceDir);
    fs::path sentPath = traceDir / "sent_trace_request.json";
    writeTextFile(sentPath, prettyJsonOrRaw(payload));
    writeTextFile(runtimeJsonPath("last_trace_request.json"), json::parse(payload).dump(2));
    if (event) event({"sent", eventTitle("Send", sentPath.u8string()), payload, sentPath.u8string(), monitor.id, traceDir.u8string()});
    auto sendOrSave = [&](const std::string& topic,
                          const std::string& tag,
                          const std::string& messageKey,
                          const std::string& body,
                          const std::string& label) {
        std::string mqError;
        if (rocketRuntime().sendPayload(topic, tag, messageKey, body, label, mqError, log)) return;

        // RocketMQ 未初始化或发送失败时不丢请求，按目标 Topic 追加到对应 outbox。
        std::string topicPart = sanitizeFilePart(topic.empty() ? label : topic);
        fs::path outbox = runtimeJsonPath("mq_outbox_" + topicPart + ".jsonl");
        std::ofstream ofs(outbox, std::ios::binary | std::ios::app);
        if (!ofs) throw std::runtime_error("无法写入 MQ outbox: " + outbox.string());
        ofs << body << "\n";
        if (log) log(label + "消息发送失败(" + mqError + ")，已写入 " + outbox.string());
    };

    // 第一条始终先发到 TopicLOG，请求前置建立本次追踪日志。
    sendOrSave(cfg.topicLOG, cfg.tagLOG, cfg.keyLOG, payload, "set_log");

    // cfg_3s_data 紧接着向 TopicAsk 请求该装置、测点的实时/SOE 数据。
    if (frontType == "cfg_3s_data") {
        std::string askPayload = buildRealtimeAskPayload(device, monitor);
        fs::path askPath = traceDir / "sent_realtime_request.json";
        writeTextFile(askPath, prettyJsonOrRaw(askPayload));
        writeTextFile(runtimeJsonPath("last_realtime_request.json"), json::parse(askPayload).dump(2));
        if (event) event({"sent", eventTitle("Send Ask", askPath.u8string()), askPayload, askPath.u8string(), monitor.id, traceDir.u8string()});
        sendOrSave(cfg.topicAsk, cfg.tagAsk, cfg.keyAsk, askPayload, "realtime ask");
    }

    if (log) log("Trace target monitor=" + monitor.name + "(" + monitor.id + "), frontType=" + frontType);
    return true;
}