/*
 * Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved.
 * ubs-engine is licensed under Mulan PSL v2.
 * You can use this software according to the terms and conditions of the Mulan PSL v2.
 * You may obtain a copy of Mulan PSL v2 at:
 *          http://license.coscl.org.cn/MulanPSL2
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PSL v2 for more details.
 */

#include "ubse_urma_controller_rpc.h"
#include <cstdint>
#include "ubse_com_module.h"
#include "ubse_common_def.h"
#include "ubse_context.h"
#include "ubse_election.h"
#include "ubse_ipc_message.h"
#include "ubse_logger.h"
#include "ubse_serial_util.h"
#include "ubse_smbios.h"
#include "ubse_urma_controller.h"
#include "ubse_urma_controller_manager.h"
#include "ubse_urma_controller_util.h"
#include "adapter_plugins/urma/ubse_urma_uvs.h"

namespace ubse::urmaController {
using namespace ubse::com;
using namespace ubse::utils;
using namespace ubse::serial;
using namespace ubse::election;
using namespace ubse::context;
using namespace ubse::nodeController;
using namespace ubse::urma;
using namespace ubse::urmaController;
using namespace ubse::common::def;

UBSE_DEFINE_THIS_MODULE("ubse");

static UbseResult DeserializeFilteredDeviceNames(UbseDeSerialization& in, std::vector<std::string>& deviceNames);

UbseResult UrmaDevQueryReqSimpo::Serialize()
{
    UbseSerialization out;
    out << req.nodeId << req.deviceNames;
    if (!out.Check()) {
        UBSE_LOG_ERROR << "Failed to serialize URMA device query request, nodeId=" << req.nodeId
                       << ", filterCount=" << req.deviceNames.size();
        return UBSE_ERROR;
    }
    mOutputRawDataSize = out.GetLength();
    mOutputRawData = std::unique_ptr<uint8_t[]>(out.GetBuffer(true));
    return UBSE_OK;
}

UbseResult UrmaDevQueryReqSimpo::Deserialize()
{
    req = {};
    if (mInputRawData == nullptr) {
        UBSE_LOG_ERROR << "InputRawData is null.";
        return UBSE_ERROR;
    }
    UbseDeSerialization in(mInputRawData.get(), mInputRawDataSize);
    in >> req.nodeId;
    if (!in.Check()) {
        UBSE_LOG_ERROR << "Failed to deserialize URMA device query node ID, inputSize=" << mInputRawDataSize;
        return UBSE_ERROR;
    }
    // 兼容旧节点仅携带 nodeId 的请求；新节点在同一消息尾部追加可选名称过滤条件。
    UbseSerialization legacyRequest;
    legacyRequest << req.nodeId;
    if (!legacyRequest.Check()) {
        UBSE_LOG_ERROR << "Failed to calculate legacy URMA device query request length, nodeId=" << req.nodeId;
        return UBSE_ERROR;
    }
    if (mInputRawDataSize == legacyRequest.GetLength()) {
        return UBSE_OK;
    }
    const auto ret = DeserializeFilteredDeviceNames(in, req.deviceNames);
    if (ret != UBSE_OK) {
        UBSE_LOG_ERROR << "Failed to deserialize URMA device query filters, nodeId=" << req.nodeId
                       << ", inputSize=" << mInputRawDataSize << ", ret=" << ret;
    }
    return ret;
}

static UbseResult DeserializeFilteredDeviceNames(UbseDeSerialization& in, std::vector<std::string>& deviceNames)
{
    common_len deviceCount = 0;
    // 先读取并限制数组长度，再分配 vector，避免不可信报文触发超大内存申请。
    in >> array_len_capture(deviceCount);
    if (!in.Check() || deviceCount > static_cast<common_len>(NO_1024)) {
        UBSE_LOG_ERROR << "Invalid filtered URMA device count=" << deviceCount;
        return UBSE_ERROR_DESERIALIZE_FAILED;
    }

    std::vector<std::string> candidate;
    candidate.reserve(static_cast<size_t>(deviceCount));
    for (common_len i = 0; i < deviceCount; ++i) {
        std::string deviceName;
        in >> deviceName;
        if (!in.Check()) {
            UBSE_LOG_ERROR << "Failed to deserialize filtered URMA device name at index=" << i;
            return UBSE_ERROR_DESERIALIZE_FAILED;
        }
        candidate.push_back(std::move(deviceName));
    }
    deviceNames = std::move(candidate);
    return UBSE_OK;
}

UbseResult UrmaDevQueryRspSimpo::Serialize()
{
    UbseSerialization out;
    out << rsp;
    if (!out.Check() || out.GetLength() > UBSE_MESSAGE_SIZE) {
        UBSE_LOG_ERROR << "Failed to serialize URMA device query response, rowCount=" << rsp.urmaInfos.size()
                       << ", serializedSize=" << out.GetLength() << ", maxMessageSize=" << UBSE_MESSAGE_SIZE;
        return UBSE_ERROR;
    }
    mOutputRawDataSize = out.GetLength();
    mOutputRawData = std::unique_ptr<uint8_t[]>(out.GetBuffer(true));
    return UBSE_OK;
}

UbseResult UrmaDevQueryRspSimpo::Deserialize()
{
    if (mInputRawData == nullptr) {
        UBSE_LOG_ERROR << "InputRawData is null.";
        return UBSE_ERROR;
    }
    UbseDeSerialization in(mInputRawData.get(), mInputRawDataSize);
    in >> rsp;
    if (!in.Check()) {
        UBSE_LOG_ERROR << "Failed to deserialize URMA device query response, inputSize=" << mInputRawDataSize;
        return UBSE_ERROR;
    }
    return UBSE_OK;
}

static void SetLocalUrmaDeviceQueryResponse(const UrmaDevQueryRpcReq& request, const UbseUrmaDevQueryRspPtr& response)
{
    UrmaDevQueryRpcRsp rpcRsp{};
    rpcRsp.result = UbseUrmaController::GetInstance().GetLocalUrmaDevs(request.deviceNames, rpcRsp.urmaInfos);
    if (rpcRsp.result != UBSE_OK) {
        UBSE_LOG_ERROR << "Failed to build local URMA device query response, nodeId=" << request.nodeId
                       << ", filterCount=" << request.deviceNames.size() << ", ret=" << rpcRsp.result;
        rpcRsp.urmaInfos.clear();
    }
    response->SetUbseUrmaDevQueryRsp(rpcRsp);
}

static UbseResult GetUrmaQueryRoleInfo(UbseRoleInfo& masterInfo, UbseRoleInfo& currentNodeInfo)
{
    auto ret = UbseGetMasterInfo(masterInfo);
    if (ret != UBSE_OK) {
        UBSE_LOG_ERROR << "Failed to get master node while handling URMA device query, ret=" << ret;
        return ret;
    }
    ret = UbseGetCurrentNodeInfo(currentNodeInfo);
    if (ret != UBSE_OK) {
        UBSE_LOG_ERROR << "Failed to get current node while handling URMA device query, ret=" << ret;
    }
    return ret;
}

UbseResult UbseUrmaDevQueryMessageHandler::Handle(const UbseBaseMessagePtr& req, const UbseBaseMessagePtr& rsp,
                                                  UbseComBaseMessageHandlerCtxPtr ctx)
{
    AsyncHandlerGuard cntGuard;
    if (g_globalStop) {
        UBSE_LOG_INFO << "Stop urma controller, ignore message";
        return UBSE_OK;
    }
    auto request = UbseBaseMessage::DeConvert<UrmaDevQueryReqSimpo>(req);
    auto response = UbseBaseMessage::DeConvert<UrmaDevQueryRspSimpo>(rsp);
    if (request == nullptr || response == nullptr) {
        UBSE_LOG_ERROR << "Failed to convert message";
        return UBSE_ERROR;
    }
    auto urmaReq = request->GetUbseUrmaDevReq();
    UbseRoleInfo currentNodeInfo{};
    UbseRoleInfo masterInfo{};
    if (GetUrmaQueryRoleInfo(masterInfo, currentNodeInfo) != UBSE_OK) {
        return UBSE_ERROR;
    }
    /* 如果是本节点的消息就查询，如果是主节点就转发，其他情况丢弃 */
    if (std::to_string(urmaReq.nodeId) == currentNodeInfo.nodeId) {
        SetLocalUrmaDeviceQueryResponse(urmaReq, response);
        return UBSE_OK;
    } else if (masterInfo.nodeId == currentNodeInfo.nodeId) {
        UrmaDevQueryRpcRsp rpcRsp{};
        SendParam sendParam{std::to_string(urmaReq.nodeId), static_cast<uint16_t>(UbseModuleCode::UBSE_URMA),
                            static_cast<uint16_t>(UbseUrmaRpcOpCode::URMA_RPC_DEV_QUERY)};
        auto comModule = ubse::context::UbseContext::GetInstance().GetModule<ubse::com::UbseComModule>();
        if (comModule == nullptr) {
            UBSE_LOG_ERROR << "Failed to get communication module while forwarding URMA device query, targetNodeId="
                           << urmaReq.nodeId;
            rpcRsp.result = UBSE_ERROR_NULLPTR;
            response->SetUbseUrmaDevQueryRsp(rpcRsp);
            return UBSE_ERROR_NULLPTR;
        }
        auto ret = comModule->RpcSend(sendParam, request, response);
        if (ret != UBSE_OK) {
            UBSE_LOG_ERROR << "Failed to forward URMA device query, targetNodeId=" << urmaReq.nodeId
                           << ", filterCount=" << urmaReq.deviceNames.size() << ", ret=" << ret;
            rpcRsp.result = ret;
            response->SetUbseUrmaDevQueryRsp(rpcRsp);
            return ret;
        }
    }
    UBSE_LOG_WARN << "URMA device query cannot be handled by current node, targetNodeId=" << urmaReq.nodeId
                  << ", currentNodeId=" << currentNodeInfo.nodeId << ", masterNodeId=" << masterInfo.nodeId;
    return UBSE_ERROR_INVAL;
}

uint16_t UbseUrmaDevQueryMessageHandler::GetModuleCode()
{
    return static_cast<uint16_t>(UbseModuleCode::UBSE_URMA);
}

uint16_t UbseUrmaDevQueryMessageHandler::GetOpCode()
{
    return static_cast<uint16_t>(UbseUrmaRpcOpCode::URMA_RPC_DEV_QUERY);
}

UbseResult UbseUrmaBrocastReqSimpo::Serialize()
{
    UbseSerialization out;
    out << req;
    if (!out.Check()) {
        UBSE_LOG_ERROR << "Failed to serialize brocast req";
        return UBSE_ERROR;
    }
    mOutputRawDataSize = out.GetLength();
    mOutputRawData = std::unique_ptr<uint8_t[]>(out.GetBuffer(true));
    return UBSE_OK;
}

UbseResult UbseUrmaBrocastReqSimpo::Deserialize()
{
    if (mInputRawData == nullptr) {
        UBSE_LOG_ERROR << "InputRawData is null.";
        return UBSE_ERROR;
    }
    UbseDeSerialization in(mInputRawData.get(), mInputRawDataSize);
    in >> req;
    if (!in.Check()) {
        UBSE_LOG_ERROR << "Failed to deserialize brocast req";
        return UBSE_ERROR;
    }
    return UBSE_OK;
}

UbseResult UbseUrmaBrocastRspSimpo::Serialize()
{
    UbseSerialization out;
    out << mErrCode;
    if (!out.Check()) {
        UBSE_LOG_ERROR << "Failed to serialize broadcast rsp";
        return UBSE_ERROR;
    }
    mOutputRawDataSize = out.GetLength();
    mOutputRawData = std::unique_ptr<uint8_t[]>(out.GetBuffer(true));
    return UBSE_OK;
}

UbseResult UbseUrmaBrocastRspSimpo::Deserialize()
{
    if (mInputRawData == nullptr) {
        UBSE_LOG_ERROR << "InputRawData is null.";
        return UBSE_ERROR;
    }
    UbseDeSerialization in(mInputRawData.get(), mInputRawDataSize);
    in >> mErrCode;
    if (!in.Check()) {
        UBSE_LOG_ERROR << "Failed to deserialize broadcast rsp";
        return UBSE_ERROR;
    }
    return UBSE_OK;
}

UbseResult QueryUrmaInfoFromMaster(const UbseRoleInfo& roleInfo, std::vector<std::string>& updateNodeIds)
{
    SendParam sendParam{roleInfo.nodeId, static_cast<uint16_t>(UbseModuleCode::UBSE_URMA),
                        static_cast<uint16_t>(UbseUrmaRpcOpCode::URMA_RPC_URMA_INFO_QUERY)};
    auto comModule = ubse::context::UbseContext::GetInstance().GetModule<ubse::com::UbseComModule>();
    if (comModule == nullptr) {
        UBSE_LOG_ERROR << "Getting ComModule failed.";
        return UBSE_ERROR_NULLPTR;
    }
    UbseUrmaQueryReqSimpoPtr request = new (std::nothrow) UbseUrmaQueryReqSimpo();
    UbseUrmaQueryRspSimpoPtr response = new (std::nothrow) UbseUrmaQueryRspSimpo();
    if (request == nullptr || response == nullptr) {
        return UBSE_ERROR;
    }
    QueryUrmaInfoReq req{updateNodeIds};
    request->SetUrmaQueryReq(req);
    auto ret = comModule->RpcSend(sendParam, request, response);
    if (ret != UBSE_OK || response->GetErrCode() != UBSE_OK) {
        UBSE_LOG_ERROR << "Do rpc query failed, ret=" << ret << ", response code=" << response->GetErrCode();
        return ret;
    }
    auto rsp = response->GetUbseUrmaQueryRsp();
    for (auto& nodeInfo : rsp.queryNodeInfos) {
        UbseUrmaControllerManager::GetInstance().InsertNewNodeInfo(nodeInfo.nodeId, nodeInfo);
    }
    return UBSE_OK;
}

void ActivateHostBonding()
{
    // 1pfe + 5vfe场景下，如果host bonding未被ubse占用，需要起定时器，创建设备预留给主机
    bool isClosType = adapter_plugins::smbios::UbseSmbios::GetInstance().IsClosType();
    FeTopoType feTopoType = UbseUrmaControllerManager::GetInstance().GetFeTopoType();
    bool isHostUrmaDevOccupied = UbseNodeController::GetInstance().IsHostBondingRegistered();
    if (!isClosType || feTopoType != FeTopoType::PFE_VFE_HYBRID || isHostUrmaDevOccupied) {
        UBSE_LOG_INFO << "Skip activating host bonding, isClosType=" << static_cast<int>(isClosType)
                      << "fe topo=" << static_cast<int>(UbseUrmaControllerManager::GetInstance().GetFeTopoType())
                      << "isHostUrmaDevOccupied=" << static_cast<int>(isHostUrmaDevOccupied);
        return;
    }
    std::string taskExecutor = "UrmaExecutor";
    std::string taskName = "UrmaActivateHostBondingRetryTimer";
    const uint32_t retryInterval = 10;
    auto task = []() {
        return UbseUrmaController::GetInstance().ActivateSpecifyUrmaDev(UBSE_HOST_URMA_DEV_NAME);
    };
    HandleTaskWithRetry(taskExecutor, taskName, retryInterval, task);
}

UbseResult DoUpdateUrmaInfos(std::vector<std::string> updateNodeIds)
{
    AsyncHandlerGuard cntGuard;
    if (g_globalStop) {
        UBSE_LOG_INFO << "Urma controller is stopped, ignore msg";
        return UBSE_OK;
    }
    UBSE_LOG_INFO << "Start to query urma info";
    UbseRoleInfo masterRoleInfo;
    if (auto ret = UbseGetMasterInfo(masterRoleInfo); ret != UBSE_OK) {
        UBSE_LOG_ERROR << "UbseGetMasterInfo failed, ret=" << ret;
        return ret;
    }
    // 如果节点状态不是最新，更新节点信息
    if (auto ret = QueryUrmaInfoFromMaster(masterRoleInfo, updateNodeIds); ret != UBSE_OK) {
        UBSE_LOG_WARN << "QueryUrmaInfoFromMaster failed.";
        return ret;
    }
    // 下发拓扑
    auto curNode = UbseNodeController::GetInstance().GetCurNode();
    if (auto ret = PushNodesTopoToUvs(curNode.nodeId); ret != UBSE_OK) {
        UBSE_LOG_WARN << "Failed to push topology to uvs, ret=" << ret;
        return ret;
    }
    // 从UVS恢复本节点bonding设备
    std::vector<UbseUrmaUvsNodeInfo> uvsInfos;
    UbseUrmaControllerManager::GetInstance().BuildUvsTopoNodeInfo(false, 0, 0, uvsInfos);
    UbseUrmaController::GetInstance().FillUrmaDevsByUvsInfo(curNode.nodeId, uvsInfos);
    bool isAllPortDown = false;
    if (auto ret = QueryAllPortsDown(isAllPortDown); ret != UBSE_OK) {
        UBSE_LOG_WARN << "Failed to query all ports status, ret=" << ret;
        return ret;
    }
    if (isAllPortDown) {
        // 将该节点的所有urmaInfo状态改成Inactive
        UBSE_LOG_INFO << "All ports are down for nodeId=" << curNode.nodeId << ", set all URMA info to PORT_DOWN";
        UbseUrmaControllerManager::GetInstance().SetAllUrmaDevStateForNode(UrmaDevState::PORT_DOWN);
    }
    UBSE_LOG_INFO << "End to update urma info";
    return UBSE_OK;
}

UbseResult PostUpdateUrmaInfosTask(const std::map<std::string, uint64_t>& urmaInfoTimestamps)
{
    UBSE_LOG_INFO << "Start to update urma info";
    std::vector<std::string> updateNodeIds; // 存储urma info有更新的节点
    static std::atomic<uint64_t> globalTimeStampUpdateId{1};
    uint64_t timeStampUpdateId{0};
    for (auto& kv : urmaInfoTimestamps) {
        auto nodeId = kv.first;
        auto brocastTimeStamp = kv.second;
        if (g_globalStop) {
            UBSE_LOG_INFO << "Urma controller is stopped, ignore msg";
            return UBSE_OK;
        }
        if (brocastTimeStamp > UbseUrmaControllerManager::GetInstance().GetUrmaUpdateTimeStamp(nodeId)) {
            updateNodeIds.emplace_back(nodeId);
        }
        timeStampUpdateId = globalTimeStampUpdateId.fetch_add(1);
    }

    static std::mutex postUpdateUrmaInfosTaskMtx;
    std::lock_guard<std::mutex> lock(postUpdateUrmaInfosTaskMtx);
    if (timeStampUpdateId < globalTimeStampUpdateId - 1) {
        UBSE_LOG_INFO << "Urma info has been updated, ignore this task";
        return UBSE_OK;
    }
    auto taskExecutor = ubse::context::UbseContext::GetInstance().GetModule<task_executor::UbseTaskExecutorModule>();
    if (taskExecutor == nullptr) {
        UBSE_LOG_ERROR << "Get task executor failed";
        return UBSE_ERROR_NULLPTR;
    }
    std::string executorName = "UrmaExecutor";
    auto urmaExecutor = taskExecutor->Get(executorName);
    if (urmaExecutor == nullptr) {
        UBSE_LOG_ERROR << "Get task executor for urma failed";
        return UBSE_ERROR_NULLPTR;
    }
    urmaExecutor->Execute([updateNodeIds, executorName]() {
        auto task = [updateNodeIds]() {
            return DoUpdateUrmaInfos(updateNodeIds);
        };
        std::string taskName = "UrmaUpdateUrmaInfoRetryTimer";
        HandleTaskWithRetry(executorName, taskName, NO_10, task);
    });
    return UBSE_OK;
}

UbseResult UbseUrmaNotifyMessageHandler::Handle(const UbseBaseMessagePtr& req, const UbseBaseMessagePtr& rsp,
                                                UbseComBaseMessageHandlerCtxPtr ctx)
{
    UBSE_LOG_INFO << "Receive urma notification, start to handle";
    AsyncHandlerGuard cntGuard;
    if (g_globalStop) {
        UBSE_LOG_INFO << "Urma controller is stopped, ignore msg";
        return UBSE_OK;
    }
    auto request = UbseBaseMessage::DeConvert<UbseUrmaBrocastReqSimpo>(req);
    auto response = UbseBaseMessage::DeConvert<UbseUrmaBrocastRspSimpo>(rsp);
    if (request == nullptr || response == nullptr) {
        UBSE_LOG_ERROR << "Failed to convert rpc message";
        return UBSE_ERROR;
    }
    UrmaBrocastReq newReq = request->GetUrmaNotifyReq();
    response->SetErrCode(UBSE_OK);
    if (auto ret = PostUpdateUrmaInfosTask(newReq.urmaInfoTimestamps); ret != UBSE_OK) {
        UBSE_LOG_WARN << "Failed to update urma info, ret=" << ret;
        response->SetErrCode(ret);
        return ret;
    }
    return UBSE_OK;
}

uint16_t UbseUrmaNotifyMessageHandler::GetModuleCode()
{
    return static_cast<uint16_t>(UbseModuleCode::UBSE_URMA);
}

uint16_t UbseUrmaNotifyMessageHandler::GetOpCode()
{
    return static_cast<uint16_t>(UbseUrmaRpcOpCode::URMA_RPC_URMA_INFO_BROADCAST);
}
// query

UbseResult UbseUrmaQueryReqSimpo::Serialize()
{
    UbseSerialization out;
    out << req.updateNodeIds;
    if (!out.Check()) {
        UBSE_LOG_ERROR << "Failed to serialize urma query req";
        return UBSE_ERROR;
    }
    mOutputRawDataSize = out.GetLength();
    mOutputRawData = std::unique_ptr<uint8_t[]>(out.GetBuffer(true));
    return UBSE_OK;
}

UbseResult UbseUrmaQueryReqSimpo::Deserialize()
{
    if (mInputRawData == nullptr) {
        UBSE_LOG_ERROR << "InputRawData is null.";
        return UBSE_ERROR;
    }
    UbseDeSerialization in(mInputRawData.get(), mInputRawDataSize);
    in >> req.updateNodeIds;
    if (!in.Check()) {
        UBSE_LOG_ERROR << "Failed to deserialize urma query req";
        return UBSE_ERROR;
    }
    return UBSE_OK;
}

UbseResult UbseUrmaQueryRspSimpo::Serialize()
{
    UbseSerialization out;
    out << rsp.queryNodeInfos;
    if (!out.Check()) {
        UBSE_LOG_ERROR << "Failed to serialize urma node infos";
        return UBSE_ERROR;
    }
    mOutputRawDataSize = out.GetLength();
    mOutputRawData = std::unique_ptr<uint8_t[]>(out.GetBuffer(true));
    return UBSE_OK;
}

UbseResult UbseUrmaQueryRspSimpo::Deserialize()
{
    if (mInputRawData == nullptr) {
        UBSE_LOG_ERROR << "InputRawData is null.";
        return UBSE_ERROR;
    }
    UbseDeSerialization in(mInputRawData.get(), mInputRawDataSize);
    in >> rsp.queryNodeInfos;
    if (!in.Check()) {
        UBSE_LOG_ERROR << "Failed to deserialize urma fe infos";
        return UBSE_ERROR;
    }
    return UBSE_OK;
}

UbseResult UbseUrmaQueryMessageHandler::Handle(const UbseBaseMessagePtr& req, const UbseBaseMessagePtr& rsp,
                                               UbseComBaseMessageHandlerCtxPtr ctx)
{
    auto request = UbseBaseMessage::DeConvert<UbseUrmaQueryReqSimpo>(req);
    auto response = UbseBaseMessage::DeConvert<UbseUrmaQueryRspSimpo>(rsp);
    if (request == nullptr || response == nullptr) {
        UBSE_LOG_ERROR << "Failed to convert rpc message";
        return UBSE_ERROR;
    }
    auto queryNodeIds = request->GetUrmaQueryReq().updateNodeIds;
    std::vector<UbseUrmaNodeInfo> updateUrmaInfos;
    response->SetErrCode(UBSE_OK);
    for (const auto& nodeId : queryNodeIds) {
        if (g_globalStop) {
            UBSE_LOG_INFO << "Urma controller is stopped, ignore msg";
            return UBSE_OK;
        }
        auto nodeInfo = UbseUrmaControllerManager::GetInstance().GetUrmaNodeInfo(nodeId);
        updateUrmaInfos.push_back(nodeInfo);
    }
    QueryUrmaInfoRsp newRsp{.queryNodeInfos = std::move(updateUrmaInfos)};
    response->SetUbseQueryRsp(newRsp);
    return UBSE_OK;
}

uint16_t UbseUrmaQueryMessageHandler::GetModuleCode()
{
    return static_cast<uint16_t>(UbseModuleCode::UBSE_URMA);
}

uint16_t UbseUrmaQueryMessageHandler::GetOpCode()
{
    return static_cast<uint16_t>(UbseUrmaRpcOpCode::URMA_RPC_URMA_INFO_QUERY);
}

UbseResult UbseUrmaReportUrmaNodeInfoReqSimpo::Serialize()
{
    UbseSerialization out;
    out << urmaNodeInfo;
    if (!out.Check()) {
        UBSE_LOG_ERROR << "Failed to serialize urma infos";
        return UBSE_ERROR;
    }
    mOutputRawDataSize = out.GetLength();
    mOutputRawData = std::unique_ptr<uint8_t[]>(out.GetBuffer(true));
    return UBSE_OK;
}

UbseResult UbseUrmaReportUrmaNodeInfoReqSimpo::Deserialize()
{
    if (mInputRawData == nullptr) {
        UBSE_LOG_ERROR << "InputRawData is null.";
        return UBSE_ERROR;
    }
    UbseDeSerialization in(mInputRawData.get(), mInputRawDataSize);
    in >> urmaNodeInfo;
    if (!in.Check()) {
        UBSE_LOG_ERROR << "Failed to deserialize urma infos";
        return UBSE_ERROR;
    }
    return UBSE_OK;
}

UbseResult UbseUrmaReportUrmaNodeInfoRspSimpo::Serialize()
{
    UbseSerialization out;
    out << mErrCode;
    if (!out.Check()) {
        UBSE_LOG_ERROR << "Failed to serialize urma infos";
        return UBSE_ERROR;
    }
    mOutputRawDataSize = out.GetLength();
    mOutputRawData = std::unique_ptr<uint8_t[]>(out.GetBuffer(true));
    return UBSE_OK;
}

UbseResult UbseUrmaReportUrmaNodeInfoRspSimpo::Deserialize()
{
    if (mInputRawData == nullptr) {
        UBSE_LOG_ERROR << "InputRawData is null.";
        return UBSE_ERROR;
    }
    UbseDeSerialization in(mInputRawData.get(), mInputRawDataSize);
    in >> mErrCode;
    if (!in.Check()) {
        UBSE_LOG_ERROR << "Failed to deserialize urma infos";
        return UBSE_ERROR;
    }
    return UBSE_OK;
}

UbseResult UbseUrmaAsyncNotifyOneNodeUrmaInfoChange(const std::string& notifyNodeId)
{
    AsyncHandlerGuard cntGuard;
    if (g_globalStop) {
        UBSE_LOG_INFO << "Urma controller is stopped, ignore msg";
        return UBSE_OK;
    }
    UBSE_LOG_INFO << "Brocast urma info timestamp to nodeId=" << notifyNodeId;
    UbseUrmaBrocastReqPtr req = new (std::nothrow) UbseUrmaBrocastReqSimpo;
    UbseUrmaBrocastRspPtr rsp = new (std::nothrow) UbseUrmaBrocastRspSimpo;
    if (req == nullptr || rsp == nullptr) {
        UBSE_LOG_ERROR << "Failed to create rpc message";
        return UBSE_ERROR;
    }
    auto nodes = UbseNodeController::GetInstance().GetAllNodes();
    // 确认待通知的节点是否在集群中，避免定时给不存在的节点发信息
    if (nodes.find(notifyNodeId) == nodes.end()) {
        UBSE_LOG_WARN << "nodeId=" << notifyNodeId << " is not in cluster, will stop brocast urma info";
        return UBSE_OK; // 返回UBSE_OK，结束定时任务
    }
    std::map<std::string, uint64_t> urmaInfoTimestamps;
    for (auto& node : nodes) {
        if (g_globalStop) {
            UBSE_LOG_INFO << "Urma controller is stopped, ignore msg";
            return UBSE_OK;
        }
        auto timeStamp = UbseUrmaControllerManager::GetInstance().GetUrmaUpdateTimeStamp(node.second.nodeId);
        urmaInfoTimestamps[node.second.nodeId] = timeStamp;
    }

    UrmaBrocastReq brocastReq{.urmaInfoTimestamps = std::move(urmaInfoTimestamps)};
    req->SetUrmaNotifyReq(brocastReq);
    SendParam sendParam{notifyNodeId, static_cast<uint16_t>(UbseModuleCode::UBSE_URMA),
                        static_cast<uint16_t>(UbseUrmaRpcOpCode::URMA_RPC_URMA_INFO_BROADCAST)};
    auto comModule = ubse::context::UbseContext::GetInstance().GetModule<ubse::com::UbseComModule>();
    if (comModule == nullptr) {
        UBSE_LOG_ERROR << "Failed to get com module";
        return UBSE_ERROR;
    }
    if (auto ret = comModule->RpcSend(sendParam, req, rsp); ret != UBSE_OK || rsp->GetErrCode() != UBSE_OK) {
        UBSE_LOG_ERROR << "Failed to send rpc message, ret=" << ret << ", " << FormatRetCode(rsp->GetErrCode());
        if (ret == UBSE_OK) {
            return rsp->GetErrCode();
        }
        return ret;
    }
    return UBSE_OK;
}

UbseResult BrocastUrmaInfoTask(const std::string& nodeId)
{
    UBSE_LOG_INFO << "Brocast urma info timestamp to nodeId=" << nodeId;
    if (auto ret = UbseUrmaAsyncNotifyOneNodeUrmaInfoChange(nodeId); ret != UBSE_OK) {
        UBSE_LOG_ERROR << "Failed to brocast urma info timestamp to nodeId=" << nodeId;
        return ret;
    }
    return UBSE_OK;
}

UbseResult UbseUrmaAsyncBrocastUrmaInfo()
{
    std::string executorName = "UrmaExecutor";
    auto taskExecutor = ubse::context::UbseContext::GetInstance().GetModule<task_executor::UbseTaskExecutorModule>();
    if (taskExecutor == nullptr) {
        UBSE_LOG_ERROR << "Get task executor failed";
        return UBSE_ERROR_NULLPTR;
    }
    auto urmaExecutor = taskExecutor->Get(executorName);
    if (urmaExecutor == nullptr) {
        UBSE_LOG_ERROR << "Get task executor for urma failed";
        return UBSE_ERROR_NULLPTR;
    }

    auto nodes = UbseNodeController::GetInstance().GetAllNodes();
    for (auto& node : nodes) {
        if (g_globalStop) {
            return UBSE_OK;
        }
        urmaExecutor->Execute([executorName, nodeId = node.second.nodeId]() {
            UBSE_LOG_INFO << "Brocast urma info timestamp to nodeId=" << nodeId;
            std::string taskName = "UrmaMasterBrocastRetryTimer_" + nodeId;
            auto task = [nodeId]() {
                return BrocastUrmaInfoTask(nodeId);
            };
            // 广播事件设置的定时器时间为10s
            HandleTaskWithRetry(executorName, taskName, NO_10, task);
        });
    }

    return UBSE_OK;
}

UbseResult UbseUrmaReportUrmaNodeInfoMessageHandler::Handle(const UbseBaseMessagePtr& req,
                                                            const UbseBaseMessagePtr& rsp,
                                                            UbseComBaseMessageHandlerCtxPtr ctx)
{
    AsyncHandlerGuard cntGuard;
    if (g_globalStop) {
        UBSE_LOG_INFO << "Urma controller is stopped, ignore msg";
        return UBSE_OK;
    }
    UBSE_LOG_INFO << "Handling URMA report node info message";
    auto request = UbseBaseMessage::DeConvert<UbseUrmaReportUrmaNodeInfoReqSimpo>(req);
    if (request == nullptr || rsp == nullptr) {
        UBSE_LOG_ERROR << "Failed to convert rpc message";
        return UBSE_ERROR;
    }
    auto nodeInfoReq = request->GetUbseUrmaNodeInfo();
    auto changeNodeId = nodeInfoReq.nodeId;
    auto& nodeInfo = nodeInfoReq.urmaNodeInfo;
    if (changeNodeId.empty() || nodeInfo.nodeId.empty()) {
        UBSE_LOG_ERROR << "node id is empty, changeNodeId=" << changeNodeId << ", nodeId=" << nodeInfo.nodeId;
        rsp->SetErrCode(UBSE_ERROR);
        return UBSE_ERROR;
    }
    // 保存到全量列表中，待其它节点获取
    UbseUrmaControllerManager::GetInstance().InsertNewNodeInfo(changeNodeId, nodeInfo);
    // 异步通知各节点nodeInfo变化
    if (auto ret = UbseUrmaAsyncBrocastUrmaInfo(); ret != UBSE_OK) {
        UBSE_LOG_ERROR << "Failed to brocast urma info timestamps";
        rsp->SetErrCode(UBSE_ERROR);
        return ret;
    }
    rsp->SetErrCode(UBSE_OK);
    return UBSE_OK;
}

uint16_t UbseUrmaReportUrmaNodeInfoMessageHandler::GetOpCode()
{
    return static_cast<uint16_t>(UbseUrmaRpcOpCode::URMA_RPC_URMA_INFO_REPORT);
}

uint16_t UbseUrmaReportUrmaNodeInfoMessageHandler::GetModuleCode()
{
    return static_cast<uint16_t>(UbseModuleCode::UBSE_URMA);
}

UbseResult ReportUrmaNodeInfoToMaster(const std::string& nodeId)
{
    UBSE_LOG_INFO << "Report urma node info to master, nodeId=" << nodeId;
    // 向master节点上报本节点urma信息
    UbseUrmaReportUrmaNodeInfoReqSimpoPtr req = new (std::nothrow) UbseUrmaReportUrmaNodeInfoReqSimpo();
    UbseUrmaReportUrmaNodeInfoRspSimpoPtr rsp = new (std::nothrow) UbseUrmaReportUrmaNodeInfoRspSimpo();
    if (req == nullptr || rsp == nullptr) {
        UBSE_LOG_ERROR << "Failed to allocate memory for rpc req or rsp";
        return UBSE_ERROR;
    }
    auto urmaNodeInfo = UbseUrmaControllerManager::GetInstance().GetUrmaNodeInfo(nodeId);
    ReportUrmaNodeInfoReq nodeInfoReq{.nodeId = nodeId, .urmaNodeInfo = std::move(urmaNodeInfo)};
    req->SetUbseUrmaNodeInfo(std::move(nodeInfoReq));
    UbseRoleInfo masterInfo{};
    if (auto ret = UbseGetMasterInfo(masterInfo); ret != UBSE_OK) {
        UBSE_LOG_ERROR << "Failed to get master info, " << FormatRetCode(ret);
        return ret;
    }
    SendParam sendParam{masterInfo.nodeId, static_cast<uint16_t>(UbseModuleCode::UBSE_URMA),
                        static_cast<uint16_t>(UbseUrmaRpcOpCode::URMA_RPC_URMA_INFO_REPORT)};
    auto comModule = ubse::context::UbseContext::GetInstance().GetModule<ubse::com::UbseComModule>();
    if (comModule == nullptr) {
        UBSE_LOG_ERROR << "Getting ComModule failed";
        return UBSE_ERROR_NULLPTR;
    }
    if (auto ret = comModule->RpcSend(sendParam, req, rsp); ret != UBSE_OK || rsp->GetErrCode() != UBSE_OK) {
        UBSE_LOG_ERROR << "Failed to do rpc send, ret=" << ret << ", " << FormatRetCode(rsp->GetErrCode());
        return ret;
    }
    return UBSE_OK;
}

UbseResult GetCurNodeIdAndMasterNodeId(std::string& curNodeId, std::string& masterNodeId)
{
    UbseRoleInfo currentNodeInfo{};
    if (UbseGetCurrentNodeInfo(currentNodeInfo) != UBSE_OK) {
        UBSE_LOG_WARN << "Failed to get current node info";
        return UBSE_ERROR_AGAIN;
    }
    curNodeId = currentNodeInfo.nodeId;
    UbseRoleInfo masterInfo{};
    if (UbseGetMasterInfo(masterInfo) != UBSE_OK) {
        UBSE_LOG_ERROR << "Failed to get master info";
        return UBSE_ERROR_AGAIN;
    }
    masterNodeId = masterInfo.nodeId;
    return UBSE_OK;
}
} // namespace ubse::urmaController