#include "common.h"
#include "mqtt_verify.h"
#include "mqtt_init.h"
#include "device_identity.h"
#include "mqtt_infor_handle.h"
#include "http_config_mqtt.h"

mqttclient g_clients_t[MAX_SERVERS];

static char s_uuid_mqtt_topic_id[MAX_SERVERS][56];
static bool s_mosquitto_lib_inited = false;
static pthread_mutex_t s_mqtt_publish_mutex = PTHREAD_MUTEX_INITIALIZER;
// struct mosquitto *mosq;

// 新增：用于向其他MQTT服务器发送通知的函数
void send_server_failure_notification(const char *failed_host, int failed_port, const char *failed_client_id);

int add_mqtt_create(const char *host, int port, int clients_count);
static int mqtt_init_internal(bool reset_all_slots, bool retry_failed_only);

//mqtt初始化
int mqtt_init() {
    return mqtt_init_internal(true, false);
}

int mqtt_retry_failed_clients(void)
{
    return mqtt_init_internal(false, true);
}

bool mqtt_has_connected(void)
{
    int total_count = g_mqtt_cam_config_t ? g_mqtt_cam_config_t->mqtt_count : 0;
    for (int i = 0; i < total_count; i++) {
        if (g_clients_t[i].mosq && !g_clients_t[i].permanently_failed) {
            return true;
        }
    }
    return false;
}

int mqtt_publish_to_all(const char *topic, const char *payload, int qos)
{
    int total_count = g_mqtt_cam_config_t ? g_mqtt_cam_config_t->mqtt_count : 0;
    int sent_count = 0;

    if (!topic || !payload) {
        return 0;
    }

    pthread_mutex_lock(&s_mqtt_publish_mutex);
    for (int i = 0; i < total_count; i++) {
        if (!g_clients_t[i].mosq || g_clients_t[i].permanently_failed) {
            continue;
        }
        if (mosquitto_publish(g_clients_t[i].mosq, NULL, topic, strlen(payload), payload, qos, false) == MOSQ_ERR_SUCCESS) {
            sent_count++;
        }
    }
    pthread_mutex_unlock(&s_mqtt_publish_mutex);
    return sent_count;
}

int mqtt_publish_to_topics(const char *topic1, const char *topic2, const char *payload, int qos)
{
    int sent_count = 0;
    if (topic1) {
        sent_count += mqtt_publish_to_all(topic1, payload, qos);
    }
    if (topic2) {
        sent_count += mqtt_publish_to_all(topic2, payload, qos);
    }
    return sent_count;
}

static int mqtt_init_internal(bool reset_all_slots, bool retry_failed_only)
{
    int success_count = 0;
    int failure_count = 0;

    if (!s_mosquitto_lib_inited) {
        mosquitto_lib_init();
        s_mosquitto_lib_inited = true;
    }

    if (!g_mqtt_cam_config_t) {
        my_zlog_fatal("mqtt_init: g_mqtt_cam_config_t is NULL. Cannot initialize.");
        return -1;
    }

    if (reset_all_slots) {
        for (int i = 0; i < MAX_SERVERS; i++) {
            if (g_clients_t[i].mosq) {
                mosquitto_disconnect(g_clients_t[i].mosq);
                mosquitto_destroy(g_clients_t[i].mosq);
            }
            g_clients_t[i].mosq = NULL;
            g_clients_t[i].reconnect_attempts = 0;
            g_clients_t[i].permanently_failed = false;
            memset(g_clients_t[i].host, 0, sizeof(g_clients_t[i].host));
            g_clients_t[i].port = 0;
            memset(g_clients_t[i].client_id, 0, sizeof(g_clients_t[i].client_id));
            memset(s_uuid_mqtt_topic_id[i], 0, sizeof(s_uuid_mqtt_topic_id[i]));
        }
    }

    for (int i = 0; i < g_mqtt_cam_config_t->mqtt_count; i++) {
        if (retry_failed_only && (g_clients_t[i].mosq || g_clients_t[i].permanently_failed)) {
            continue;
        }

        int rc = add_mqtt_create(g_mqtt_cam_config_t->mqtt_servers[i], BROKER_PORT, i);
        if (rc == 0) {
            success_count++;
        } else {
            failure_count++;
            my_zlog_error("未能为 %s:%d 添加 MQTT 客户端", g_mqtt_cam_config_t->mqtt_servers[i], BROKER_PORT);
        }
    }

    my_zlog_info("MQTT初始化结果: success=%d failure=%d total=%d",
                 success_count, failure_count, g_mqtt_cam_config_t->mqtt_count);

    if (mqtt_has_connected()) {
        return 0;
    }

    return -1;
}

//回调函数
void on_connect(struct mosquitto *mosq, void *obj, int rc) {

    mqttclient *client_t = (mqttclient*)obj;

    if (!client_t) { // 安全检查
        my_zlog_error("on_connect: client_t 为 NULL");
        return;
    }

    if (rc == 0) {
        my_zlog_info("MQTT已连接 %s:%d (%s)",
               client_t->host, client_t->port, client_t->client_id);
        client_t->reconnect_attempts = 0;
        client_t->permanently_failed = false;

        mosquitto_subscribe(mosq, NULL, mqtt_topic_app2dev_number(), 0);
        mosquitto_subscribe(mosq, NULL, mqtt_topic_ser2dev_number(), 0);
    } else {
        client_t->reconnect_attempts++;
        my_zlog_warn("客户端 %s:%d 连接失败。重试次数: %d/%d",
               client_t->host, client_t->port, client_t->reconnect_attempts, MAX_RECONNECT_ATTEMPTS);

        if (client_t->reconnect_attempts >= MAX_RECONNECT_ATTEMPTS) {
            my_zlog_error("客户端 %s:%d 达到最大重试次数，不再尝试连接。标记为永久失败。",
                   client_t->host, client_t->port);
            client_t->permanently_failed = true;
            // 停止 mosquitto 循环并销毁客户端
            // 注意：在回调中直接销毁当前正在操作的 mosq 可能会导致问题，
            // 更好的做法是设置一个标志，让外部循环（如果存在）或专门的清理线程来处理。
            // 但因为 mosquitto_loop_start 是独立的线程，这里尝试停止并销毁它。
            
            // mosquitto_loop_stop 可能需要等待线程结束，这里是同步调用
            mosquitto_loop_stop(mosq, false); // false 表示不强制，等待线程自行结束
            mosquitto_disconnect(mosq);
            mosquitto_destroy(mosq);
            client_t->mosq = NULL; // 清除指针

            // 向其他活跃的 MQTT 服务器发送通知
            send_server_failure_notification(client_t->host, client_t->port, client_t->client_id);
        }

    }
}


//消息回环函数  
void on_message(struct mosquitto *mosq, void *obj, const struct mosquitto_message *message) {

     mqttclient *client_t = (mqttclient*)obj;

    if (!client_t || !message || !message->topic) { // 安全检查
        my_zlog_error("on_message: 无效参数 (client_t, message 或 topic 为 NULL)");
        return;
     }

     my_zlog_debug("[Message from %s:%d] topic=%s payload=%s",
           client_t->host, client_t->port,
           message->topic, (char*)message->payload);

    // 确保消息有效
    if (message->payload && message->payloadlen) { // 将消息内容转换为字符串
        char *payload_str = (char *)malloc(message->payloadlen + 1);
        if (!payload_str) {
            return;
        } 

        memcpy(payload_str, message->payload, message->payloadlen);
        payload_str[message->payloadlen] = '\0';

        // 解析 JSON
        cJSON *json = cJSON_Parse(payload_str);
        if(json != NULL){
            if(device_message_receive(json)==0) my_zlog_debug("进入设备接收mqtt");
        }                     
        
        // 释放 payload 字符串
        free(payload_str); 
        cJSON_Delete(json); 
    }
}

// 添加 MQTT 客户端并尝试连接
// clients_idx 是 g_clients_t 和 s_uuid_mqtt_topic_id 数组的索引
int add_mqtt_create(const char *host, int port, int clients_idx) {
    // 移除了 'static int uuid_index = 0;'
    // 不再需要独立的 uuid_index，直接使用传入的 clients_idx 作为索引

    if (clients_idx >= MAX_SERVERS) {
        my_zlog_error("达到最大服务器限制，索引 %d。", clients_idx);
        return -1;
    }

    // 检查主机名是否为空或无效
    if (!host || strlen(host) == 0) {
        my_zlog_error("主机名为空，索引 %d。", clients_idx);
        return -1;
    }

    mqttclient *client_t = &g_clients_t[clients_idx];
    if (client_t->mosq) {
        if (strcmp(client_t->host, host) == 0 && client_t->port == port) {
            return 0;
        }
        mosquitto_disconnect(client_t->mosq);
        mosquitto_destroy(client_t->mosq);
        client_t->mosq = NULL;
    }

    // 复制主机名，确保字符串末尾空终止
    strncpy(client_t->host, host, sizeof(client_t->host) - 1);
    client_t->host[sizeof(client_t->host) - 1] = '\0';
    client_t->port = port;
    client_t->reconnect_attempts = 0;
    client_t->permanently_failed = false;

    // 生成 UUID 作为客户端 ID
    uuid_t uuid;
    uuid_generate_time(uuid); // 获取 UUIDv1

    // 组合 UUID 的一部分作为短 ID，并存储在对应 clients_idx 的位置
    // 这将作为 mosquitto 客户端的 ID
    unsigned short time_part = (uuid[0] << 8) | uuid[1];
    unsigned short mac_part  = (uuid[10] << 8) | uuid[11];
    snprintf(s_uuid_mqtt_topic_id[clients_idx], sizeof(s_uuid_mqtt_topic_id[clients_idx]), "%04x%04x", time_part, mac_part);
    s_uuid_mqtt_topic_id[clients_idx][sizeof(s_uuid_mqtt_topic_id[clients_idx])-1] = '\0'; // 确保空终止

    // 将生成的 Client ID 也存储到 client_t 结构体中
    strncpy(client_t->client_id, s_uuid_mqtt_topic_id[clients_idx], sizeof(client_t->client_id) -1);
    client_t->client_id[sizeof(client_t->client_id) -1] = '\0'; // 确保空终止

    // 创建 mosquitto 客户端实例
    client_t->mosq = mosquitto_new(s_uuid_mqtt_topic_id[clients_idx], true, client_t);
    
    if (!client_t->mosq) {
        my_zlog_error("mosquitto_new 失败，客户端索引 %d", clients_idx);
        return -1;
    }

    // 设置 MQTT 协议版本
    mosquitto_int_option(client_t->mosq, MOSQ_OPT_PROTOCOL_VERSION, MQTT_PROTOCOL_V5);

    // 设置重连延迟策略
    mosquitto_reconnect_delay_set(client_t->mosq, 2, 10, true);
    // 设置回调函数
    mosquitto_connect_callback_set(client_t->mosq, on_connect);
    mosquitto_message_callback_set(client_t->mosq, on_message);

    // 设置用户名和密码
    mosquitto_username_pw_set(client_t->mosq, USERNAME, PASSWORD);
   
    // 尝试异步连接到 MQTT Broker
    int rc = mosquitto_connect_async(client_t->mosq, host, port, 60);
    if (rc != MOSQ_ERR_SUCCESS) {
        my_zlog_error("mosquitto_connect_async 失败，主机 %s:%d (rc=%d): %s", host, port, rc, mosquitto_strerror(rc));
        mosquitto_destroy(client_t->mosq); // 如果连接失败，立即销毁 mosquitto 客户端实例
        client_t->mosq = NULL; // 将指针置空，以防止在清理时操作无效指针
        return -1;
    }

    // 启动 mosquitto 循环线程，处理网络 I/O 和回调
    int loop_rc = mosquitto_loop_start(client_t->mosq);
    if (loop_rc != MOSQ_ERR_SUCCESS) {
        my_zlog_error("未能为客户端 %s 启动 mosquitto_loop_start: %s", client_t->client_id, mosquitto_strerror(loop_rc));
        // 如果 loop_start 失败，也应断开连接并销毁
        mosquitto_disconnect(client_t->mosq);
        mosquitto_destroy(client_t->mosq);
        client_t->mosq = NULL; // 确保指针置空
        return -1;
    }
  
    my_zlog_debug("成功为主机 %s 添加 MQTT 客户端", host);

    return 0; // 成功
}

//mqtt清理
void mqtt_clean(){
      // 清理
    for(int i=0;i<g_mqtt_cam_config_t->mqtt_count;i++){
        if (!g_clients_t[i].mosq) {
            continue;
        }
        mosquitto_disconnect(g_clients_t[i].mosq); 
        mosquitto_destroy(g_clients_t[i].mosq);
        g_clients_t[i].mosq = NULL; // 将指针置空，防止悬挂指针
    }
    
    if (s_mosquitto_lib_inited) {
        mosquitto_lib_cleanup();
        s_mosquitto_lib_inited = false;
    }
    my_zlog_debug("MQTT 清理完成");
}

int mqtt_cycle() {//非阻塞型

    my_zlog_debug("MQTT 异步监控线程启动");
    
    int check_interval = 30; // 30秒检查一次，减少日志频率
    
    while (1) {
                // 简单的存活检查，记录连接状态
        int active_count = 0;
        int total_count = g_mqtt_cam_config_t ? g_mqtt_cam_config_t->mqtt_count : 0;
        
        for (int i = 0; i < total_count; i++) {
            if (g_clients_t[i].mosq && !g_clients_t[i].permanently_failed) {
                active_count++;
            }
        }
        
        // 只在状态变化或定期记录
        static int last_active_count = -1;
        if (active_count != last_active_count) {
            my_zlog_info("MQTT 连接状态: %d/%d 活跃", active_count, total_count);
            last_active_count = active_count;
        }
        delay_s(check_interval);
    }

    return 0;
}

// 向其他活跃的 MQTT 服务器发送通知
void send_server_failure_notification(const char *failed_host, int failed_port, const char *failed_client_id) {
    my_zlog_info("准备发送服务器故障通知： %s:%d (%s)", failed_host, failed_port, failed_client_id);

    cJSON *root = cJSON_CreateObject();
    if (!root) {
        my_zlog_error("创建 JSON 对象失败。");
        return;
    }
    cJSON_AddStringToObject(root, "event_type", "server_failure");
    cJSON_AddStringToObject(root, "failed_host", failed_host);
    cJSON_AddNumberToObject(root, "failed_port", failed_port);
    cJSON_AddStringToObject(root, "failed_client_id", failed_client_id);
    cJSON_AddStringToObject(root, "message", "MQTT server permanently disconnected after multiple retries.");
    cJSON_AddNumberToObject(root, "message_type", 9999);

    char *json_string = cJSON_PrintUnformatted(root);
    if (!json_string) {
        my_zlog_error("将 JSON 打印到字符串失败。");
        cJSON_Delete(root);
        return;
    }

    mqtt_publish_to_topics(mqtt_topic_dev2app_number(), mqtt_topic_pure_number(), json_string, 0);

    free(json_string);
    cJSON_Delete(root);
}

void send_fail_mqtt_conect(){
    cJSON *root = cJSON_CreateObject();
    if (!root) {
        my_zlog_error("创建 JSON 对象失败。");
        return;
    }

    cJSON_AddStringToObject(root, "event_type", "server_connect_failure");
    cJSON_AddNumberToObject(root, "message_type", 9999);

    char *json_string = cJSON_PrintUnformatted(root);
    if (!json_string) {
        my_zlog_error("将 JSON 打印到字符串失败。");
        cJSON_Delete(root);
        return;
    }

    mqtt_publish_to_topics(mqtt_topic_dev2app_number(), mqtt_topic_pure_number(), json_string, 0);

    free(json_string);
    cJSON_Delete(root);
}