#include "tcp_client.hpp"
#include "logger.hpp"
#include "broadcast_ack_manager.hpp"

#include <memory>

static std::unique_ptr<TcpClient> broadcast_client;

static std::atomic<bool> heartbeat_running{false};
static std::atomic<bool> heartbeat_ack_received{false}; // 是否收到应答
static std::thread heartbeat_thread;
static std::thread ack_timeout_thread;

static void heartbeat_task()
{
    while (heartbeat_running.load())
    {
        if (broadcast_client && broadcast_client->is_connected())
        {
            heartbeat_ack_received = false;
            // 构造并发送心跳包
            FullPacket heartPacket = ProtocolCodec::create_heartbeat_packet(g_mqtt_config.vin);
            std::vector<uint8_t> raw = ProtocolCodec::encode_full_packet(heartPacket);
            std::string payload(reinterpret_cast<const char *>(raw.data()), raw.size());
            broadcast_client->send_data(payload);
            LOG_INFO("[broadcast] Sent heartbeat.");

            // 等待10秒期间，判断是否收到回应
            for (int i = 0; i < 10; ++i)
            {
                std::this_thread::sleep_for(std::chrono::seconds(1));
                if (!heartbeat_running.load())
                    return;
            }

            if (!heartbeat_ack_received.load())
            {
                LOG_WARN("[broadcast] Heartbeat ACK not received within 10 seconds.");
            }
        }
        else
        {
            LOG_WARN("[broadcast] Not connected, stop sending heartbeat.");
            std::this_thread::sleep_for(std::chrono::seconds(10));
        }
    }
    LOG_INFO("[broadcast] Heartbeat thread exited.");
}

// 额外线程定时检测超时的流水号
static void ack_timeout_check_task()
{
    while (heartbeat_running.load())
    {
        std::this_thread::sleep_for(std::chrono::seconds(5));
        auto timedOutSeqs = broadcastAckMgr.checkTimeouts(std::chrono::seconds(10));
        for (auto seq : timedOutSeqs)
        {
            LOG_WARN("[AckManager] Timeout for seq = " + std::to_string(seq) + ", consider retransmitting or alerting.");
            // 这里可以触发重发逻辑或告警
        }
    }
}

// 发送广播消息，自动登记流水号
void send_broadcast_message_with_seq(const std::string &msg, uint16_t seq)
{
    if (broadcast_client && broadcast_client->is_connected())
    {
        broadcast_client->send_data(msg);
        LOG_INFO("[broadcast] Sent message, length = " + std::to_string(msg.size()) + ", seq = " + std::to_string(seq));
        broadcastAckMgr.addPending(seq);
    }
    else
    {
        LOG_WARN("[broadcast] Cannot send message: client not connected.");
    }
}

// 接收处理逻辑
static void handle_broadcast_message(const std::string &data)
{
    LOG_INFO("[broadcast] Received data, length = " + std::to_string(data.size()) + " bytes");

    std::vector<uint8_t> bytes(data.begin(), data.end());
    auto pkt_opt = ProtocolCodec::decode_full_packet(bytes);

    if (!pkt_opt.has_value())
    {
        LOG_WARN("[broadcast] Invalid packet received.");
        return;
    }

    const FullPacket &pkt = pkt_opt.value();

    if (pkt.command_id == 0x07 && pkt.response_flag == 0x01 && pkt.data_length == 0)
    {
        heartbeat_ack_received = true;
        LOG_INFO("[broadcast] Received heartbeat ACK.");
    }
    else if (pkt.command_id == 0xD5 && pkt.data_length == 8) // 应答包
    {
        uint16_t seq = 0;
        if (pkt.data_unit.size() == 8)
        {
            seq = (pkt.data_unit[6] << 8) | pkt.data_unit[7];
        }
        if (pkt.response_flag == 0x01)
        {
            LOG_INFO("[broadcast] Send broadcast packet success, seq = " + std::to_string(seq));
            broadcastAckMgr.receiveAck(seq);
        }
        else if (pkt.response_flag == 0x02)
        {
            LOG_WARN("[broadcast] Send broadcast packet failed, seq = " + std::to_string(seq));
            broadcastAckMgr.receiveAck(seq);
            // 这里可以加失败重发逻辑
        }
    }
}

// 状态变化处理逻辑
static void handle_broadcast_status(bool connected)
{
    LOG_INFO("[broadcast] status: " + std::string(connected ? "connected" : "disconnected"));

    if (connected)
    {
        if (!heartbeat_running.load())
        {
            heartbeat_running = true;
            heartbeat_thread = std::thread(heartbeat_task);
            ack_timeout_thread = std::thread(ack_timeout_check_task); //  启动 ACK 超时检测线程
            LOG_INFO("[broadcast] Heartbeat thread started.");
        }
    }
    else
    {
        if (heartbeat_running.load())
        {
            heartbeat_running = false;
            if (heartbeat_thread.joinable())
                heartbeat_thread.join();

            if (ack_timeout_thread.joinable()) // 结束 ACK 超时检测线程
                ack_timeout_thread.join();

            LOG_INFO("[broadcast] Heartbeat + ACK Timeout threads stopped.");
        }
    }
}

void start_broadcast_client(const std::string &ip, int port)
{
    broadcast_client = std::make_unique<TcpClient>("broadcast", ip, port);

    broadcast_client->set_receive_callback(handle_broadcast_message);

    broadcast_client->set_status_callback(handle_broadcast_status);

    broadcast_client->start();
}
