#include <softbus_transport/TcpUplinkClient.h>

#include <softbus_core/logging/Logger.h>

#include <array>

namespace softbus::transport {

TcpUplinkClient::TcpUplinkClient(std::string host, uint16_t port)
    : host_(std::move(host))
    , port_(port)
    , workGuard_(std::make_unique<asio::executor_work_guard<asio::io_context::executor_type>>(
          asio::make_work_guard(io_)))
{
}

TcpUplinkClient::~TcpUplinkClient()
{
    disconnect();
}

void TcpUplinkClient::setCommandHandler(CommandHandler handler)
{
    commandHandler_ = std::move(handler);
}

bool TcpUplinkClient::connect()
{
    if (connected_) {
        return true;
    }
    socket_ = std::make_unique<asio::ip::tcp::socket>(io_);
    std::error_code ec;
    socket_->connect(asio::ip::tcp::endpoint(asio::ip::make_address(host_), port_), ec);
    if (ec) {
        SB_LOG_ERROR("TcpUplinkClient", "Connect failed: ", ec.message());
        socket_.reset();
        return false;
    }
    connected_ = true;
    ioThread_ = std::thread([this]() { io_.run(); });
    doRead();
    SB_LOG_INFO("TcpUplinkClient", "Connected to ", host_, ":", port_);
    return true;
}

void TcpUplinkClient::disconnect()
{
    connected_ = false;
    if (socket_) {
        std::error_code ec;
        socket_->close(ec);
        socket_.reset();
    }
    workGuard_.reset();
    io_.stop();
    if (ioThread_.joinable()) {
        ioThread_.join();
    }
    io_.restart();
    workGuard_ = std::make_unique<asio::executor_work_guard<asio::io_context::executor_type>>(
        asio::make_work_guard(io_));
}

bool TcpUplinkClient::sendCommand(const softbus::api::ExecuteCommand& command)
{
    if (!socket_ || !socket_->is_open()) {
        return false;
    }
    const auto line = encodeWireMessage(command);
    std::lock_guard<std::mutex> lock(sendMutex_);
    std::error_code ec;
    asio::write(*socket_, asio::buffer(line), ec);
    return !ec;
}

bool TcpUplinkClient::sendAck(const softbus::core::SoftbusAck& ack)
{
    if (!socket_ || !socket_->is_open()) {
        return false;
    }
    const auto line = encodeWireMessage(ack);
    std::lock_guard<std::mutex> lock(sendMutex_);
    std::error_code ec;
    asio::write(*socket_, asio::buffer(line), ec);
    return !ec;
}

bool TcpUplinkClient::sendHeartbeat(const std::string& deviceId)
{
    if (!socket_ || !socket_->is_open()) {
        return false;
    }
    const auto line = encodeWireMessage(deviceId);
    std::lock_guard<std::mutex> lock(sendMutex_);
    std::error_code ec;
    asio::write(*socket_, asio::buffer(line), ec);
    return !ec;
}

bool TcpUplinkClient::sendRawPacket(const softbus::core::RawPacket& packet)
{
    if (!socket_ || !socket_->is_open()) {
        return false;
    }
    const auto line = encodeWireMessage(packet);
    std::lock_guard<std::mutex> lock(sendMutex_);
    std::error_code ec;
    asio::write(*socket_, asio::buffer(line), ec);
    return !ec;
}

void TcpUplinkClient::poll()
{
    io_.poll();
}

void TcpUplinkClient::doRead()
{
    if (!socket_ || !socket_->is_open()) {
        return;
    }
    auto buffer = std::make_shared<std::array<char, 4096>>();
    socket_->async_read_some(asio::buffer(*buffer),
        [this, buffer](const std::error_code& ec, std::size_t bytes) {
            if (ec) {
                connected_ = false;
                SB_LOG_WARN("TcpUplinkClient", "Disconnected from daemon");
                return;
            }
            readBuffer_.append(buffer->data(), bytes);
            for (;;) {
                const auto pos = readBuffer_.find('\n');
                if (pos == std::string::npos) {
                    break;
                }
                const std::string line = readBuffer_.substr(0, pos);
                readBuffer_.erase(0, pos + 1);
                handleLine(line);
            }
            doRead();
        });
}

void TcpUplinkClient::handleLine(const std::string& line)
{
    const auto msg = decodeWireMessage(line);
    if (!msg) {
        return;
    }
    if (msg->type == WireMessageType::ExecuteCommand && commandHandler_) {
        commandHandler_(msg->command);
    }
}

} // namespace softbus::transport
