#include <softbus_command/CommandService.h>

#include <softbus_core/identity/DeviceIdentity.h>
#include <softbus_core/logging/Logger.h>

namespace softbus::command {

CommandService::CommandService(softbus::registry::PointRegistry* registry,
                               std::chrono::milliseconds timeout)
    : registry_(registry)
    , timeout_(timeout)
{
}

void CommandService::attachBus(softbus::bus::Bus* bus)
{
    bus_ = bus;
    if (!bus_) {
        return;
    }
    ackSub_ = bus_->subscribe("control/ack/#", [this](const softbus::bus::BusMessage& msg) {
        handleAck(msg);
    });
}

CommandContext CommandService::submit(const softbus::api::ExecuteCommand& command, bool waitForAck)
{
    CommandContext ctx;
    ctx.command = command;
    ctx.requestId = command.requestId.empty()
        ? generateRequestId()
        : command.requestId;
    ctx.command.requestId = ctx.requestId;

    std::string error;
    if (!ctx.command.validate(&error)) {
        ctx.state = CommandState::Rejected;
        ctx.error = error;
        return ctx;
    }

    if (!registry_) {
        ctx.state = CommandState::Rejected;
        ctx.error = "registry not available";
        return ctx;
    }

    const auto entry = registry_->findByObjectRef(ctx.command.objectRef);
    if (!entry) {
        ctx.state = CommandState::Rejected;
        ctx.error = "unknown objectRef";
        return ctx;
    }

    if (ctx.command.command == "write") {
        const double value = ctx.command.params.value("value", 0.0);
        if (!registry_->validateWriteValue(*entry, value, &error)) {
            ctx.state = CommandState::Rejected;
            ctx.error = error;
            return ctx;
        }
    } else if (!entry->writable) {
        ctx.state = CommandState::Rejected;
        ctx.error = "point is not writable";
        return ctx;
    }

    ctx.state = CommandState::Validated;
    {
        std::lock_guard<std::mutex> lock(mutex_);
        contexts_[ctx.requestId] = ctx;
    }

    if (!bus_) {
        ctx.state = CommandState::Failed;
        ctx.error = "bus not attached";
        return ctx;
    }

    const std::string topic = entry->cmdTopic.empty()
        ? registry_->topicRules().formatCommand(entry->deviceId, entry->pointId)
        : entry->cmdTopic;

    softbus::bus::BusMessage msg;
    msg.type = softbus::bus::BusPayloadType::ExecuteCommand;
    msg.command = ctx.command;
    bus_->publish(topic, msg);

    {
        std::lock_guard<std::mutex> lock(mutex_);
        contexts_[ctx.requestId].state = CommandState::Dispatched;
    }
    ctx.state = CommandState::Dispatched;

    SB_LOG_INFO_TRACE("CommandService", ctx.command.traceId,
        "command dispatched requestId=", ctx.requestId, " topic=", topic);

    if (!waitForAck) {
        return ctx;
    }

    std::unique_lock<std::mutex> lock(mutex_);
    const bool done = ackCv_.wait_for(lock, timeout_, [&]() {
        const auto it = contexts_.find(ctx.requestId);
        if (it == contexts_.end()) {
            return true;
        }
        return it->second.state == CommandState::Succeeded
            || it->second.state == CommandState::Failed
            || it->second.state == CommandState::Timeout;
    });

    const auto it = contexts_.find(ctx.requestId);
    if (it == contexts_.end()) {
        ctx.state = CommandState::Failed;
        ctx.error = "context lost";
        return ctx;
    }
    if (!done && it->second.state == CommandState::Dispatched) {
        it->second.state = CommandState::Timeout;
        it->second.error = "ack timeout";
    }
    return it->second;
}

std::optional<CommandContext> CommandService::findByRequestId(const std::string& requestId) const
{
    std::lock_guard<std::mutex> lock(mutex_);
    const auto it = contexts_.find(requestId);
    if (it == contexts_.end()) {
        return std::nullopt;
    }
    return it->second;
}

void CommandService::handleAck(const softbus::bus::BusMessage& message)
{
    const auto& ack = message.ack;
    if (ack.requestId.empty()) {
        return;
    }

    std::lock_guard<std::mutex> lock(mutex_);
    const auto it = contexts_.find(ack.requestId);
    if (it == contexts_.end()) {
        return;
    }
    it->second.state = ack.success ? CommandState::Succeeded : CommandState::Failed;
    it->second.error = ack.error;
    it->second.executedValue = ack.executedValue;
    ackCv_.notify_all();

    SB_LOG_INFO_TRACE("CommandService", ack.traceId,
        "ack matched requestId=", ack.requestId,
        " state=", commandStateToString(it->second.state));
}

std::string CommandService::generateRequestId() const
{
    return "req-" + softbus::core::DeviceIdentity::generateTraceId();
}

} // namespace softbus::command
