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>

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

namespace {

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();
}

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); });
    if (name == L"bin" && dir.has_parent_path()) return dir.parent_path();
    return dir;
}

fs::path runtimeDir() {
    return appRootDir() / "runtime";
}

fs::path runtimeJsonDir() {
    return runtimeDir() / "json";
}

fs::path runtimeLogsDir() {
    return runtimeDir() / "logs";
}

fs::path runtimeJsonPath(const std::string& name) {
    return runtimeJsonDir() / fs::u8path(name);
}

fs::path runtimeLogPath(const std::string& name) {
    return runtimeLogsDir() / fs::u8path(name);
}

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;
}

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;
}

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();
}

bool setContains(const std::unordered_set<std::string>& allowed, const std::string& value) {
    return allowed.empty() || allowed.count(value) > 0;
}

std::unordered_set<std::string> parseStatusSet(const std::string& text) {
    std::unordered_set<std::string> out;
    try {
        json j = json::parse(text);
        if (j.is_array()) {
            for (const auto& v : j) out.insert(jsonString(json{{"v", v}}, "v"));
        }
    } catch (...) {
        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;
}

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;
}

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;
}

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;
        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;
}

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;
}

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;
}

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;
}

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;
}

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();
}

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;
}

const char* nonEmptyCStr(const std::string& value, const char* fallback) {
    return value.empty() ? fallback : value.c_str();
}

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;
}

bool isWildcardTag(const std::string& tag) {
    std::string t = trimCopy(tag);
    return t.empty() || t == "*";
}

std::string subscribeTagExpression(const std::string& tag) {
    return isWildcardTag(tag) ? "*" : trimCopy(tag);
}

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;
}

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));
        if (part == "*" || part == actual) return true;
        if (pos == std::string::npos) break;
        start = pos + 2;
    }
    return false;
}

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;
}

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()) {
        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;
}

long long timestampFromJsonText(const std::string& text) {
    try {
        return findTimestampValue(json::parse(text));
    } catch (...) {
        return 0;
    }
}

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();
}

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;
}

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;
}

std::string dataTypeFromJsonText(const std::string& text) {
    try {
        std::string dataType;
        if (findDataTypeValue(json::parse(text), dataType)) return dataType;
    } catch (...) {
    }
    return "unknown";
}

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;
}

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);

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";
}

bool looksLikeJsonText(const std::string& text) {
    std::string s = trimCopy(text);
    return !s.empty() && (s.front() == '{' || s.front() == '[');
}

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()) {
                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;
}

std::string monitorIdFromJsonText(const std::string& text) {
    try {
        std::string monitorId;
        if (findMonitorIdValue(json::parse(text), monitorId)) return monitorId;
    } catch (...) {
    }
    return "";
}

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;
}

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 "";
}

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);
}

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;
}

std::string firstNonEmpty(const std::string& a, const std::string& b) {
    return !a.empty() ? a : b;
}

std::string prettyJsonOrRaw(const std::string& text) {
    try {
        return json::parse(text).dump(2);
    } catch (...) {
        return text;
    }
}

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));
}

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;
}

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;
}

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;
}

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;
};

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;
        std::unordered_set<std::string> receivedDataTypes;
        long long dataTimestamp = 0;
        int originCounter = 0;
    };

public:
    ~RocketMqRuntime() {
        DetachedHandles old;
        {
            std::lock_guard<std::mutex> lock(mu_);
            old = detachLocked();
        }
        shutdownDetached(old);
    }

    bool initialize(const PqToolConfig& cfg, PqLogFn log, PqMqEventFn event) {
        DetachedHandles old;
        {
            std::lock_guard<std::mutex> lock(mu_);
            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_) {
                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);

        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;
    }

    void setTraceTarget(const PqToolConfig& cfg,
                        const PqLedgerDevice& device,
                        const PqLedgerMonitor& monitor,
                        const std::string& frontType,
                        const std::string& guid) {
        std::lock_guard<std::mutex> lock(mu_);
        cfg_ = cfg;
        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);
        auto old = sessions_.find(monitorKey);
        if (old != sessions_.end() && !old->second.guid.empty()) {
            sessionsByGuid_.erase(normalizeMonitorId(old->second.guid));
        }
        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.receivedDataTypes.clear();
        session.dataTimestamp = 0;
        session.originCounter = 0;
        if (!guid.empty()) sessionsByGuid_[normalizeMonitorId(guid)] = monitorKey;

        json context = {
            {"deviceId", device.id},
            {"deviceName", device.name},
            {"devType", device.devType},
            {"processNo", device.processNo},
            {"monitorId", monitor.id},
            {"monitorName", monitor.name},
            {"guid", guid},
            {"frontType", frontType},
            {"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));
    }

    bool sendTrace(const PqToolConfig& cfg,
                   const std::string& payload,
                   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(cfg.topicLOG, ""));
        if (!msg) {
            err = "CreateMessage 返回空指针";
            return false;
        }

        SetMessageTags(msg, nonEmptyCStr(cfg.tagLOG, ""));
        SetMessageKeys(msg, nonEmptyCStr(cfg.keyLOG, ""));
        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 数据追踪消息已发送，TopicLOG=" + cfg.topicLOG
                + "，MsgId=" + std::string(result.msgId));
        }
        return true;
    }

private:
    using MessageCallback = int (*)(CPushConsumer*, CMessageExt*);

    static int messageCallback(CPushConsumer*, CMessageExt* msg) {
        return instance().onMessage(msg);
    }

    TraceSession* findSessionForMessageLocked(const std::string& body,
                                              const std::string& key,
                                              std::string& actualMonitor,
                                              std::string& reason) {
        actualMonitor.clear();
        reason.clear();

        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;
    }

    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");
            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;
                }
            }
            if (isData) {
                std::string dataType = dataTypeFromJsonText(body);
                std::string normalizedType = normalizeDataType(dataType);
                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);
                if (session->dataTimestamp > 0 && bodyTs > 0 && bodyTs < session->dataTimestamp) {
                    appendTextFile(runtimeLogPath("mq_messages.log"), nowIsoMillis() + " TraceTopic tag=" + tag + " skipped by timestamp\n");
                } else {
                    fs::path out = nextOriginOutputPathLocked(*session, reportNameFromJsonText(body));
                    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;
    }

    static RocketMqRuntime& instance() {
        static RocketMqRuntime runtime;
        return runtime;
    }

    friend RocketMqRuntime& rocketRuntime();

    fs::path originOutputPath() const {
        if (!cfg_.originPaths.empty() && !cfg_.originPaths.front().empty()) {
            return fs::u8path(cfg_.originPaths.front());
        }
        return fs::u8path("origin.txt");
    }

    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 "";
    }

    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);
    }

    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;
    }

    bool failLocked(const std::string& message, PqLogFn log) {
        lastError_ = message;
        shutdownLocked();
        if (log) log("RocketMQ 初始化失败: " + message);
        return 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);
    }

    bool loadLibraryLocked() {
        if (module_) return true;

        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;
    }

    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;
    }

    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);
    }

    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();
    }

    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;
};

RocketMqRuntime& rocketRuntime() {
    return RocketMqRuntime::instance();
}

} // namespace

PqToolConfig pqDefaultConfig() {
    PqToolConfig cfg;
    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;
}

std::string pqTraceDirectoryFor(const PqToolConfig& cfg,
                                const PqLedgerDevice& device,
                                const PqLedgerMonitor& monitor) {
    fs::path root = fs::u8path(cfg.traceRootDir.empty() ? "trace_data" : cfg.traceRootDir);
    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();
}

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);
}

std::string pqFindDeviceXmlPath(const PqToolConfig& cfg, const PqLedgerDevice& device) {
    if (!device.xmlPath.empty() && fs::exists(fs::u8path(device.xmlPath))) return device.xmlPath;

    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 "";
}

bool pqFetchLedgerAndIcd(const PqToolConfig& cfg,
                         std::vector<PqLedgerDevice>& devices,
                         std::string& firstXmlPath,
                         PqLogFn log) {
    devices.clear();
    firstXmlPath.clear();

    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;

    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()) + " 个");

    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;
    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;
}

bool pqInitializeMq(const PqToolConfig& cfg, PqLogFn log, PqMqEventFn event) {
    json saved = {
        {"producer", cfg.producer},
        {"ipport", cfg.ipport},
        {"producerAccessKey", cfg.producerAccessKey},
        {"producerSecretKey", cfg.producerSecretKey},
        {"topicLOG", cfg.topicLOG},
        {"tagLOG", cfg.tagLOG},
        {"keyLOG", cfg.keyLOG},
        {"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) {
        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;
}

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") {
        processNo = 0;
    } else {
        try {
            processNo = device.processNo.empty() ? 0 : std::stoi(device.processNo);
        } catch (...) {
            processNo = 0;
        }
    }

    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();
}

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);
    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()});
    std::string mqError;
    if (rocketRuntime().sendTrace(cfg, payload, mqError, log)) {
        if (log) log("Trace target monitor=" + monitor.name + "(" + monitor.id + "), frontType=" + frontType);
        return true;
    }

    std::string topicPart = sanitizeFilePart(cfg.topicLOG.empty() ? "TopicLOG" : cfg.topicLOG);
    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 << payload << "\n";
    writeTextFile(runtimeJsonPath("last_trace_request.json"), json::parse(payload).dump(2));

    if (log) {
        log("已构造数据追踪消息，TopicLOG=" + cfg.topicLOG + ", TagLOG=" + cfg.tagLOG + ", KeyLOG=" + cfg.keyLOG);
        log("测点=" + monitor.name + "(" + monitor.id + "), frontType=" + frontType);
        log("当前未直连 RocketMQ，消息已写入 " + outbox.string());
    }
    return true;
}