// mqtt_client_wrapper.cpp
#include "mqtt_client_wrapper.hpp"
#include <thread>
#include <chrono>
#include <nlohmann/json.hpp>

std::shared_ptr<MQTTClient> mqtt_client;
std::atomic<bool> mqtt_restart_required{false};
extern std::atomic<bool> g_running;
std::atomic<bool> g_streaming{false};
std::string g_dispatch_id;
std::mutex g_dispatch_id_mutex;

static void send_heartbeat()
{
    if (!mqtt_client || !mqtt_client->isConnected())
        return;

    nlohmann::json hb_data;
    hb_data["time"] = Logger::get_current_time_utc8();
    hb_data["status"] = g_streaming ? 0 : 2;

    {
        std::lock_guard<std::mutex> lock(g_dispatch_id_mutex);
        hb_data["dispatchId"] = g_streaming ? g_dispatch_id : "";
    }

    nlohmann::json msg;
    msg["data"] = hb_data;
    msg["isEnc"] = 0;
    msg["type"] = 0;

    mqtt_client->publish(g_app_config.mqtt.topics.heartbeat_up, msg.dump());
}

// MQTT 回调定义
static void on_mqtt_connected()
{
    LOG_INFO("[MQTT] Connected to broker: " + g_app_config.mqtt.server_ip);

    const auto &topics = g_app_config.mqtt.topics;
    mqtt_client->subscribe(topics.video_down);
    mqtt_client->subscribe(topics.substream_down);
    mqtt_client->subscribe(topics.reset_down);
}

static void on_mqtt_disconnected()
{
    LOG_WARN("[MQTT] Disconnected from broker: " + g_app_config.mqtt.server_ip);
}

static void on_mqtt_message_received(const std::string &topic, const std::string &message)
{
    LOG_INFO("[MQTT] Received message on topic [" + topic + "], len = " + std::to_string(message.size()));
    LOG_INFO("[MQTT] Message content: " + message); // 打印实际内容

    try
    {
        if (topic == g_app_config.mqtt.topics.video_down)
        {
            // 处理 video_down
            auto j = nlohmann::json::parse(message);

            if (!j.contains("data") || !j["data"].contains("status") || !j["data"].contains("seqNo"))
            {
                LOG_WARN("[MQTT] video_down JSON missing required fields");
                return;
            }

            // 写 dispatchId 并设置 streaming 状态
            {
                std::lock_guard<std::mutex> lock(g_dispatch_id_mutex);
                if (j["data"].contains("dispatchId"))
                    g_dispatch_id = j["data"]["dispatchId"].get<std::string>();
            }

            int status = j["data"]["status"].get<int>();

            std::string seqNo = j["data"]["seqNo"].get<std::string>();

            bool success = true; // 标记是否操作成功

            if (status == 0)
            {
                g_streaming = true;
                // 启动推流：挂载本地配置中 enabled 的摄像头
                for (const auto &cam : g_app_config.cameras)
                {
                    if (!cam.enabled)
                        continue;

                    if (!RTSPManager::is_streaming(cam.name))
                    {
                        RTSPManager::mount_camera(cam);
                        LOG_INFO("[MQTT] Started streaming: " + cam.name);
                    }
                }
            }
            else if (status == 1)
            {
                g_streaming = false;
                std::lock_guard<std::mutex> lock(g_dispatch_id_mutex);
                g_dispatch_id.clear(); // 停止拉流就清空 dispatchId
                // 停止推流：卸载本地配置中 enabled 的摄像头
                for (const auto &cam : g_app_config.cameras)
                {
                    if (!cam.enabled)
                        continue;

                    if (RTSPManager::is_streaming(cam.name))
                    {
                        RTSPManager::unmount_camera(cam);
                        LOG_INFO("[MQTT] Stopped streaming: " + cam.name);
                    }
                }
            }
            else
            {
                LOG_WARN("[MQTT] video_down: unknown status value " + std::to_string(status));
                success = false;
            }

            // 获取当前时间 yyyyMMddHHmmssSSS (UTC+8)
            std::string time_str = Logger::get_current_time_utc8();

            nlohmann::json reply_data;
            reply_data["time"] = time_str;
            reply_data["result"] = success ? 0 : 1;
            reply_data["seqNo"] = seqNo;

            // 封装外层
            nlohmann::json reply;
            reply["data"] = reply_data;
            reply["isEnc"] = 0;
            reply["type"] = 0;

            // 发送应答
            if (mqtt_client)
            {
                mqtt_client->publish(g_app_config.mqtt.topics.video_down_ack, reply.dump());
                LOG_INFO("[MQTT] Replied to video_down: " + reply.dump());
            }
        }
        else if (topic == g_app_config.mqtt.topics.substream_down)
        {
            // 处理 substream_down
            LOG_INFO("[MQTT] substream_down message received (not implemented yet).");
        }
        else if (topic == g_app_config.mqtt.topics.reset_down)
        {
            auto j = nlohmann::json::parse(message);

            // reset/down 的 payload 在 data 里
            auto data = j.contains("data") ? j["data"] : nlohmann::json::object();

            std::string seqNo = data.value("seqNo", "");
            std::string errCode = data.value("errorCode", "");
            std::string des = data.value("des", "");

            LOG_WARN("[MQTT] Reset command received, errorCode=" + errCode + ", des=" + des);

            // 停止所有流，相当于复位
            for (const auto &cam : g_app_config.cameras)
            {
                if (RTSPManager::is_streaming(cam.name))
                {
                    RTSPManager::unmount_camera(cam);
                    LOG_INFO("[RTSP] Camera " + cam.name + " reset/unmounted");
                }
            }

            // 组装应答 data
            nlohmann::json reply_data;
            reply_data["time"] = Logger::get_current_time_utc8();
            reply_data["result"] = 0; // 0=成功
            reply_data["seqNo"] = seqNo;

            // 外层封装
            nlohmann::json reply;
            reply["data"] = reply_data;
            reply["isEnc"] = 0;
            reply["type"] = 0;

            if (mqtt_client)
            {
                mqtt_client->publish(g_app_config.mqtt.topics.reset_down_ack, reply.dump());
                LOG_INFO("[MQTT] Replied to reset_down: " + reply.dump());
            }
        }
        else
        {
            LOG_WARN("[MQTT] Unknown topic: " + topic);
        }
    }
    catch (const std::exception &e)
    {
        LOG_ERROR(std::string("[MQTT] Failed to process incoming JSON: ") + e.what());
    }
}

void mqtt_client_thread_func()
{
    const auto &cfg = g_app_config.mqtt;
    auto heartbeat_interval = std::chrono::milliseconds(static_cast<int>(cfg.keep_alive * 0.9));

    while (g_running)
    {
        mqtt_client = std::make_unique<MQTTClient>(cfg);
        mqtt_client->setConnectCallback(on_mqtt_connected);
        mqtt_client->setDisconnectCallback(on_mqtt_disconnected);
        mqtt_client->setMessageCallback(on_mqtt_message_received);

        while (g_running && !mqtt_client->isConnected())
        {
            try
            {
                mqtt_client->connect();
            }
            catch (...)
            {
            }
            if (!mqtt_client->isConnected())
                std::this_thread::sleep_for(std::chrono::seconds(2));
        }

        while (g_running && mqtt_client->isConnected())
        {
            try
            {
                send_heartbeat();
            }
            catch (const std::exception &e)
            {
                LOG_ERROR("[MQTT] Heartbeat error: " + std::string(e.what()));
            }
            std::this_thread::sleep_for(heartbeat_interval);
        }

        // 需要重启或退出
        mqtt_client->disconnect();
        mqtt_client.reset();
        mqtt_restart_required = false;
    }
    LOG_INFO("[MQTT] Client thread exiting.");
}