#include "mqtt.h"

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <strings.h>
#include <time.h>
#include <unistd.h>
#include <cJSON.h>
#include "esp_heap_caps.h"
#include "esp_log.h"
#include "esp_timer.h"
#include "esp_wifi.h"
#include "freertos/FreeRTOS.h"
#include "freertos/task.h"
#include "mqtt_client.h"
#include "system.h"
#include "app_task.h"

#define BOARD_MQTT_KEEPALIVE_SEC 120
#define BOARD_MQTT_STATE_REPORT_INTERVAL_MS 5000
#define BOARD_MQTT_RECONNECT_DELAY_MS 30000
#define BOARD_MQTT_NETWORK_TIMEOUT_MS 15000
#define BOARD_MQTT_FINAL_REPORT_WAIT_MS 10000
#define BOARD_MQTT_VOICE_YIELD_WAIT_MS 120000
#define BOARD_MQTT_TASK_PRIORITY 5
#define BOARD_MQTT_RECONNECT_TASK_PRIORITY 4
#define BOARD_MQTT_TOPIC_MAX_LEN 128
#define BOARD_MQTT_PAYLOAD_MAX_LEN 8192
#define BOARD_MQTT_DOWNLOAD_FIELD_MAX_LEN 256
#define BOARD_MQTT_DOWNLOAD_INFO_MAX_LEN 128
#define BOARD_MQTT_PLAYLIST_POLL_MS 200
#define BOARD_MQTT_PLAYLIST_TASK_STACK_SIZE 6144
#define BOARD_MQTT_PLAYLIST_TASK_PRIORITY 5
#define BOARD_MQTT_PLAYLIST_PATH MOUNT_POINT "/.mqtt_play_media_playlist.json"
#define BOARD_MQTT_QOS 0

#define BOARD_MQTT_ERR_INVALID_PARAM  4001
#define BOARD_MQTT_ERR_UNSUPPORTED    4002
#define BOARD_MQTT_ERR_EXEC_FAILED    4003
#define BOARD_MQTT_ERR_TOPIC_MISMATCH 4004
#define BOARD_MQTT_ERR_NOT_FOUND      4005
#define BOARD_MQTT_ERR_BUSY           4006

static const char *TAG = "board_mqtt";

static esp_mqtt_client_handle_t s_mqtt_client = NULL;
static bool s_mqtt_connected = false;
static char s_device_id[BOARD_MQTT_TOPIC_MAX_LEN];
static char s_cmd_ctl_topic[BOARD_MQTT_TOPIC_MAX_LEN];
static char s_cmd_task_topic[BOARD_MQTT_TOPIC_MAX_LEN];
static char s_rpt_log_topic[BOARD_MQTT_TOPIC_MAX_LEN];
static char s_rpt_ack_topic[BOARD_MQTT_TOPIC_MAX_LEN];
static char s_rpt_progress_topic[BOARD_MQTT_TOPIC_MAX_LEN];
static char s_rpt_state_topic[BOARD_MQTT_TOPIC_MAX_LEN];
static char s_rpt_event_topic[BOARD_MQTT_TOPIC_MAX_LEN];
typedef enum {
    BOARD_MQTT_TASK_NONE = 0,
    BOARD_MQTT_TASK_DOWNLOAD,
    BOARD_MQTT_TASK_MANIFEST,
    BOARD_MQTT_TASK_OTA,
    BOARD_MQTT_TASK_PLAYLIST,
} board_mqtt_task_type_t;

static bool s_task_busy = false;
static bool s_manifest_pending = false;
static board_mqtt_task_type_t s_task_type = BOARD_MQTT_TASK_NONE;
static volatile bool s_download_voice_yield_requested = false;
static volatile bool s_download_cancel_requested = false;
static portMUX_TYPE s_task_lock = portMUX_INITIALIZER_UNLOCKED;
static TaskHandle_t s_reconnect_task_handle = NULL;
static TaskHandle_t s_state_task_handle = NULL;
static TaskHandle_t s_playlist_task_handle = NULL;
static volatile bool s_playlist_stop_requested = false;
static uint32_t s_state_sequence = 0;

static void BoardMqtt_HandleCommand(const char *topic, int topic_len, const char *data, int data_len);
static bool BoardMqtt_CopyString(char *dest, size_t dest_size, const char *src);
static bool BoardMqtt_IsMd5Hex(const char *value);
static void BoardMqtt_WaitConnectedForFinalReport(void);
static bool BoardMqtt_BeginTask(board_mqtt_task_type_t type);
static void BoardMqtt_FinishTask(board_mqtt_task_type_t type);
static bool BoardMqtt_PlaybackSpeedIsSupported(float speed);
static void BoardMqtt_StopPlaylistTask(void);
static esp_err_t BoardMqtt_StartPlaylistTask(void);

typedef struct {
    char topic[BOARD_MQTT_TOPIC_MAX_LEN];
    int topic_len;
    char *payload;
    int total_len;
    int received_len;
} board_mqtt_payload_assembly_t;

static board_mqtt_payload_assembly_t s_payload_assembly = {0};

typedef enum {
    BOARD_MQTT_TOPIC_UNKNOWN = 0,
    BOARD_MQTT_TOPIC_CTL,
    BOARD_MQTT_TOPIC_TASK,
} board_mqtt_topic_type_t;

typedef struct {
    bool success;
    int error_code;
    const char *info;
} board_mqtt_action_result_t;

static board_mqtt_action_result_t BoardMqtt_HandleUnsetAlarm(cJSON *payload);

typedef struct {
    char req_id[BOARD_MQTT_DOWNLOAD_FIELD_MAX_LEN];
    char url[BOARD_MQTT_DOWNLOAD_FIELD_MAX_LEN];
    char target_path[SDCARD_PATH_MAX_LEN];
    char md5[33];
} board_mqtt_download_job_t;

typedef struct {
    char url[BOARD_MQTT_DOWNLOAD_FIELD_MAX_LEN];
    char target_path[SDCARD_PATH_MAX_LEN];
    char md5[33];
} board_mqtt_manifest_asset_t;

typedef struct {
    char req_id[BOARD_MQTT_DOWNLOAD_FIELD_MAX_LEN];
    char course_id[ALARM_ID_MAX_LEN];
    char *manifest_json;
    size_t asset_count;
    bool replace_existing_download;
    board_mqtt_manifest_asset_t assets[];
} board_mqtt_manifest_job_t;

typedef struct {
    char req_id[BOARD_MQTT_DOWNLOAD_FIELD_MAX_LEN];
    char url[BOARD_MQTT_DOWNLOAD_FIELD_MAX_LEN];
    char version[64];
    int64_t last_publish_us;
    int last_progress;
} board_mqtt_ota_job_t;

typedef struct {
    char task_id[BOARD_MQTT_DOWNLOAD_FIELD_MAX_LEN];
    char path[SDCARD_PATH_MAX_LEN];
    char sd_path[SDCARD_PATH_MAX_LEN];
    int volume;
    bool has_volume;
    float speed;
    int index;
    int count;
} board_mqtt_playlist_item_t;

typedef struct {
    const char *req_id;
    int64_t last_publish_us;
    int last_progress;
} board_mqtt_download_progress_ctx_t;

typedef struct {
    const char *req_id;
    size_t asset_index;
    size_t asset_count;
    int64_t last_publish_us;
    int last_progress;
} board_mqtt_manifest_progress_ctx_t;

static bool BoardMqtt_IsDownloadTask(board_mqtt_task_type_t type)
{
    return type == BOARD_MQTT_TASK_DOWNLOAD || type == BOARD_MQTT_TASK_MANIFEST;
}

static board_mqtt_task_type_t BoardMqtt_GetTaskType(void)
{
    board_mqtt_task_type_t type;
    portENTER_CRITICAL(&s_task_lock);
    type = s_task_type;
    portEXIT_CRITICAL(&s_task_lock);
    return type;
}

static bool BoardMqtt_BeginTask(board_mqtt_task_type_t type)
{
    bool started = false;
    portENTER_CRITICAL(&s_task_lock);
    if (!s_task_busy) {
        s_task_busy = true;
        s_task_type = type;
        s_download_cancel_requested = false;
        started = true;
    }
    portEXIT_CRITICAL(&s_task_lock);
    return started;
}

static bool BoardMqtt_BeginManifestReplacement(void)
{
    bool started = false;
    portENTER_CRITICAL(&s_task_lock);
    if (s_task_busy && BoardMqtt_IsDownloadTask(s_task_type) && !s_manifest_pending) {
        s_manifest_pending = true;
        s_download_cancel_requested = true;
        started = true;
    }
    portEXIT_CRITICAL(&s_task_lock);
    return started;
}

static void BoardMqtt_FinishTask(board_mqtt_task_type_t type)
{
    portENTER_CRITICAL(&s_task_lock);
    if (s_task_type == type) {
        s_task_busy = false;
        s_task_type = BOARD_MQTT_TASK_NONE;
        s_download_cancel_requested = false;
    }
    portEXIT_CRITICAL(&s_task_lock);
}

static bool BoardMqtt_HasPendingManifest(void)
{
    bool pending;
    portENTER_CRITICAL(&s_task_lock);
    pending = s_manifest_pending;
    portEXIT_CRITICAL(&s_task_lock);
    return pending;
}

static bool BoardMqtt_ActivatePendingManifest(void)
{
    bool activated = false;
    portENTER_CRITICAL(&s_task_lock);
    if (!s_task_busy && s_manifest_pending) {
        s_task_busy = true;
        s_task_type = BOARD_MQTT_TASK_MANIFEST;
        s_manifest_pending = false;
        s_download_cancel_requested = false;
        activated = true;
    }
    portEXIT_CRITICAL(&s_task_lock);
    return activated;
}

static void BoardMqtt_ClearPendingManifest(void)
{
    portENTER_CRITICAL(&s_task_lock);
    s_manifest_pending = false;
    portEXIT_CRITICAL(&s_task_lock);
}

static void BoardMqtt_RequestDownloadCancel(void)
{
    portENTER_CRITICAL(&s_task_lock);
    if (s_task_busy && BoardMqtt_IsDownloadTask(s_task_type)) {
        s_download_cancel_requested = true;
    }
    portEXIT_CRITICAL(&s_task_lock);
}

static bool BoardMqtt_WaitDownloadCanceled(int timeout_ms)
{
    int waited_ms = 0;
    while (true) {
        board_mqtt_task_type_t type = BoardMqtt_GetTaskType();
        if (!BoardMqtt_IsDownloadTask(type)) {
            return true;
        }
        if (waited_ms >= timeout_ms) {
            return false;
        }
        vTaskDelay(pdMS_TO_TICKS(100));
        waited_ms += 100;
    }
}

bool BoardMqtt_CancelDownload(void)
{
    board_mqtt_task_type_t type = BoardMqtt_GetTaskType();
    if (!BoardMqtt_IsDownloadTask(type)) {
        return false;
    }
    BoardMqtt_RequestDownloadCancel();
    return true;
}

static esp_err_t BoardMqtt_FormatTopic(char *buffer, size_t buffer_size, const char *format, const char *device_id)
{
    int topic_len = snprintf(buffer, buffer_size, format, device_id);
    if (topic_len < 0 || topic_len >= (int)buffer_size) {
        ESP_LOGE(TAG, "MQTT topic is too long: %s", format);
        return ESP_ERR_INVALID_SIZE;
    }
    return ESP_OK;
}

static long BoardMqtt_GetTimestamp(void)
{
    time_t now = time(NULL);
    if (now < 0) {
        return 0;
    }
    return (long)now;
}

static void BoardMqtt_PublishJson(const char *topic, cJSON *json, int qos)
{
    if (!s_mqtt_client || !s_mqtt_connected || !topic || !json) {
        return;
    }

    char *payload = cJSON_PrintUnformatted(json);
    if (!payload) {
        ESP_LOGE(TAG, "Failed to format MQTT JSON payload");
        return;
    }

    int msg_id = esp_mqtt_client_publish(s_mqtt_client, topic, payload, 0, qos, 0);
    if (msg_id >= 0) {
        ESP_LOGI(TAG, "Published MQTT message, topic=%s, msg_id=%d", topic, msg_id);
    } else {
        ESP_LOGE(TAG, "Failed to publish MQTT message to %s", topic);
    }
    cJSON_free(payload);
}

static void BoardMqtt_WaitConnectedForFinalReport(void)
{
    int waited_ms = 0;
    while (s_mqtt_client && !s_mqtt_connected && waited_ms < BOARD_MQTT_FINAL_REPORT_WAIT_MS) {
        ESP_LOGW(TAG, "Waiting MQTT reconnect before final report, waited=%dms", waited_ms);
        vTaskDelay(pdMS_TO_TICKS(500));
        waited_ms += 500;
    }
}

static void BoardMqtt_PublishState(void)
{
    cJSON *root = cJSON_CreateObject();
    cJSON *header = cJSON_CreateObject();
    cJSON *payload = cJSON_CreateObject();
    if (!root || !header || !payload) {
        cJSON_Delete(root);
        cJSON_Delete(header);
        cJSON_Delete(payload);
        return;
    }

    wifi_ap_record_t ap_info = {0};
    bool wifi_connected = esp_wifi_sta_get_ap_info(&ap_info) == ESP_OK;
    size_t free_heap = heap_caps_get_free_size(MALLOC_CAP_8BIT);
    size_t total_heap = heap_caps_get_total_size(MALLOC_CAP_8BIT);
    int mem_percent = total_heap > 0 ? (int)(((total_heap - free_heap) * 100) / total_heap) : 0;

    cJSON_AddStringToObject(header, "device_id", s_device_id);
    cJSON_AddNumberToObject(header, "timestamp", BoardMqtt_GetTimestamp());
    cJSON_AddItemToObject(root, "header", header);
    cJSON_AddStringToObject(root, "msg_type", "state");

    ++s_state_sequence;
    board_mqtt_task_type_t task_type = BoardMqtt_GetTaskType();
    cJSON_AddNumberToObject(payload, "cpu", Board_Get_CPU_Usage());
    cJSON_AddNumberToObject(payload, "mem", mem_percent);
    cJSON_AddNumberToObject(payload, "battery", Battery_Get_Value());
    cJSON_AddStringToObject(payload, "net", wifi_connected ? "wifi" : "offline");
    cJSON_AddStringToObject(payload, "mode", task_type != BOARD_MQTT_TASK_NONE ? "task" : (Board_Is_Voice_Session_Active() ? "voice" : "idle"));
    cJSON_AddItemToObject(root, "payload", payload);

    BoardMqtt_PublishJson(s_rpt_state_topic, root, BOARD_MQTT_QOS);
    cJSON_Delete(root);
}

static void BoardMqtt_PublishLog(const char *message)
{
    cJSON *root = cJSON_CreateObject();
    cJSON *header = cJSON_CreateObject();
    cJSON *payload = cJSON_CreateObject();
    if (!root || !header || !payload) {
        cJSON_Delete(root);
        cJSON_Delete(header);
        cJSON_Delete(payload);
        return;
    }

    cJSON_AddStringToObject(header, "device_id", s_device_id);
    cJSON_AddNumberToObject(header, "timestamp", BoardMqtt_GetTimestamp());
    cJSON_AddItemToObject(root, "header", header);
    cJSON_AddStringToObject(root, "msg_type", "log");
    cJSON_AddStringToObject(payload, "module", "mqtt");
    cJSON_AddStringToObject(payload, "message", message);
    cJSON_AddItemToObject(root, "payload", payload);

    BoardMqtt_PublishJson(s_rpt_log_topic, root, BOARD_MQTT_QOS);
    cJSON_Delete(root);
}

void BoardMqtt_PublishSttEvent(const char *text)
{
    if (!text || text[0] == '\0') {
        return;
    }

    cJSON *root = cJSON_CreateObject();
    cJSON *header = cJSON_CreateObject();
    cJSON *payload = cJSON_CreateObject();
    if (!root || !header || !payload) {
        cJSON_Delete(root);
        cJSON_Delete(header);
        cJSON_Delete(payload);
        return;
    }

    cJSON_AddStringToObject(header, "device_id", s_device_id);
    cJSON_AddNumberToObject(header, "timestamp", BoardMqtt_GetTimestamp());
    cJSON_AddItemToObject(root, "header", header);
    cJSON_AddStringToObject(root, "msg_type", "event");
    cJSON_AddStringToObject(payload, "event", "asr_result");
    cJSON_AddStringToObject(payload, "text", text);
    cJSON_AddItemToObject(root, "payload", payload);

    BoardMqtt_PublishJson(s_rpt_event_topic, root, BOARD_MQTT_QOS);
    cJSON_Delete(root);
}

static void BoardMqtt_PublishAck(const char *req_id, bool success, int error_code, const char *info)
{
    if (!req_id || req_id[0] == '\0') {
        return;
    }

    cJSON *root = cJSON_CreateObject();
    cJSON *header = cJSON_CreateObject();
    cJSON *payload = cJSON_CreateObject();
    if (!root || !header || !payload) {
        cJSON_Delete(root);
        cJSON_Delete(header);
        cJSON_Delete(payload);
        return;
    }

    cJSON_AddStringToObject(header, "device_id", s_device_id);
    cJSON_AddNumberToObject(header, "timestamp", BoardMqtt_GetTimestamp());
    cJSON_AddItemToObject(root, "header", header);
    cJSON_AddStringToObject(root, "msg_type", "ack");
    cJSON_AddStringToObject(payload, "req_id", req_id);
    cJSON_AddStringToObject(payload, "status", success ? "success" : "failed");
    cJSON_AddNumberToObject(payload, "error_code", success ? 0 : error_code);
    cJSON_AddStringToObject(payload, "info", info ? info : "");
    cJSON_AddItemToObject(root, "payload", payload);

    BoardMqtt_PublishJson(s_rpt_ack_topic, root, BOARD_MQTT_QOS);
    cJSON_Delete(root);
}

static void BoardMqtt_PublishProgress(const char *task_id, int progress, const char *info)
{
    if (!task_id || task_id[0] == '\0') {
        return;
    }
    if (progress < 0) {
        progress = 0;
    } else if (progress > 100) {
        progress = 100;
    }

    cJSON *root = cJSON_CreateObject();
    cJSON *header = cJSON_CreateObject();
    cJSON *payload = cJSON_CreateObject();
    if (!root || !header || !payload) {
        cJSON_Delete(root);
        cJSON_Delete(header);
        cJSON_Delete(payload);
        return;
    }

    cJSON_AddStringToObject(header, "device_id", s_device_id);
    cJSON_AddNumberToObject(header, "timestamp", BoardMqtt_GetTimestamp());
    cJSON_AddItemToObject(root, "header", header);
    cJSON_AddStringToObject(root, "msg_type", "progress");
    cJSON_AddStringToObject(payload, "task_id", task_id);
    cJSON_AddStringToObject(payload, "status", "processing");
    cJSON_AddNumberToObject(payload, "progress", progress);
    cJSON_AddStringToObject(payload, "info", info ? info : "");
    cJSON_AddItemToObject(root, "payload", payload);

    ESP_LOGI(TAG, "MQTT progress: task_id=%s progress=%d info=%s", task_id, progress, info ? info : "");
    BoardMqtt_PublishJson(s_rpt_progress_topic, root, BOARD_MQTT_QOS);
    cJSON_Delete(root);
}

static char *BoardMqtt_CopySegment(const char *data, int data_len)
{
    if (!data || data_len <= 0) {
        return NULL;
    }

    char *copy = (char *)calloc(1, data_len + 1);
    if (!copy) {
        ESP_LOGE(TAG, "Failed to allocate MQTT segment buffer");
        return NULL;
    }
    memcpy(copy, data, data_len);
    copy[data_len] = '\0';
    return copy;
}

static bool BoardMqtt_NumberToInt(cJSON *item, int *value)
{
    if (!cJSON_IsNumber(item) || !value) {
        return false;
    }
    double number = item->valuedouble;
    int int_value = item->valueint;
    if ((double)int_value != number) {
        return false;
    }
    *value = int_value;
    return true;
}

static void BoardMqtt_LogReceivedJson(const char *topic, cJSON *payload)
{
    cJSON *root = cJSON_CreateObject();
    if (!root) {
        ESP_LOGW(TAG, "Failed to format received MQTT JSON");
        return;
    }

    cJSON_AddStringToObject(root, "event", "mqtt_command_received");
    cJSON_AddStringToObject(root, "topic", topic ? topic : "");
#if BOARD_MQTT_EXECUTE_COMMANDS
    cJSON_AddBoolToObject(root, "execute_commands", true);
#else
    cJSON_AddBoolToObject(root, "execute_commands", false);
#endif
    cJSON_AddItemReferenceToObject(root, "payload", payload);

    char *formatted = cJSON_Print(root);
    if (!formatted) {
        ESP_LOGW(TAG, "Failed to format received MQTT JSON");
        cJSON_Delete(root);
        return;
    }
    ESP_LOGI(TAG, "%s", formatted);
    cJSON_free(formatted);
    cJSON_Delete(root);
}

static void BoardMqtt_LogReceivedText(const char *topic, const char *payload)
{
    cJSON *root = cJSON_CreateObject();
    if (!root) {
        ESP_LOGW(TAG, "Failed to format received MQTT text");
        return;
    }

    cJSON_AddStringToObject(root, "event", "mqtt_command_invalid_json");
    cJSON_AddStringToObject(root, "topic", topic ? topic : "");
#if BOARD_MQTT_EXECUTE_COMMANDS
    cJSON_AddBoolToObject(root, "execute_commands", true);
#else
    cJSON_AddBoolToObject(root, "execute_commands", false);
#endif
    cJSON_AddStringToObject(root, "payload", payload ? payload : "");

    char *formatted = cJSON_Print(root);
    if (!formatted) {
        ESP_LOGW(TAG, "Failed to format received MQTT text");
        cJSON_Delete(root);
        return;
    }
    ESP_LOGW(TAG, "%s", formatted);
    cJSON_free(formatted);
    cJSON_Delete(root);
}

static bool BoardMqtt_CopyBusinessPath(char *dest, size_t dest_size, const char *path)
{
    if (!dest || dest_size == 0 || !path || path[0] == '\0') {
        return false;
    }
    if (path[0] == '/') {
        return BoardMqtt_CopyString(dest, dest_size, path);
    }
    int written = snprintf(dest, dest_size, "/%s", path);
    return written > 0 && (size_t)written < dest_size;
}

static bool BoardMqtt_IsSupportedResourcePath(const char *sd_path)
{
    const char *dot = sd_path ? strrchr(sd_path, '.') : NULL;
    return dot && (strcasecmp(dot, ".mp3") == 0 || strcasecmp(dot, ".jpg") == 0);
}

static bool BoardMqtt_IsPlaceholderMd5(const char *md5)
{
    return !md5 || md5[0] == '\0' || strcmp(md5, "...") == 0;
}

static bool BoardMqtt_ValidateOptionalMd5(cJSON *md5)
{
    if (!md5) {
        return true;
    }
    return cJSON_IsString(md5) && (BoardMqtt_IsPlaceholderMd5(md5->valuestring) || BoardMqtt_IsMd5Hex(md5->valuestring));
}

static bool BoardMqtt_CopyOptionalMd5(char *dest, size_t dest_size, cJSON *md5)
{
    if (!cJSON_IsString(md5) || BoardMqtt_IsPlaceholderMd5(md5->valuestring)) {
        return true;
    }
    return BoardMqtt_CopyString(dest, dest_size, md5->valuestring);
}

static bool BoardMqtt_ActionIsCtl(const char *action)
{
    return strcmp(action, "SET_VOLUME") == 0 || strcmp(action, "SET_POWER") == 0 ||
        strcmp(action, "SET_CLOCK") == 0 || strcmp(action, "PLAY_MEDIA") == 0 ||
        strcmp(action, "STOP_MEDIA") == 0 || strcmp(action, "DELETE_FILE") == 0 ||
        strcmp(action, "SET_ALARM") == 0 || strcmp(action, "UNSET_ALARM") == 0 ||
        strcmp(action, "REBOOT") == 0;
}

static bool BoardMqtt_ActionIsTask(const char *action)
{
    return strcmp(action, "DOWNLOAD") == 0 || strcmp(action, "UPDATE_MANIFEST") == 0 ||
        strcmp(action, "SYNC_COURSE_TASK") == 0 || strcmp(action, "OTA_UPDATE") == 0;
}

static board_mqtt_action_result_t BoardMqtt_ResultOk(const char *info)
{
    return (board_mqtt_action_result_t){ .success = true, .error_code = 0, .info = info };
}

static board_mqtt_action_result_t BoardMqtt_ResultFail(int error_code, const char *info)
{
    return (board_mqtt_action_result_t){ .success = false, .error_code = error_code, .info = info };
}

static board_mqtt_action_result_t BoardMqtt_HandleSetVolume(cJSON *payload)
{
    int level = 0;
    if (!BoardMqtt_NumberToInt(cJSON_GetObjectItem(payload, "level"), &level) || level < 0 || level > 100) {
        return BoardMqtt_ResultFail(BOARD_MQTT_ERR_INVALID_PARAM, "invalid volume level");
    }
    esp_err_t err = Board_Set_Output_Volume(level);
    return err == ESP_OK ? BoardMqtt_ResultOk("volume updated") :
        BoardMqtt_ResultFail(BOARD_MQTT_ERR_EXEC_FAILED, esp_err_to_name(err));
}

static board_mqtt_action_result_t BoardMqtt_HandleSetPower(cJSON *payload)
{
    cJSON *state = cJSON_GetObjectItem(payload, "state");
    if (!cJSON_IsString(state)) {
        return BoardMqtt_ResultFail(BOARD_MQTT_ERR_INVALID_PARAM, "missing power state");
    }

    esp_err_t err = ESP_OK;
    if (strcmp(state->valuestring, "sleep") == 0) {
        err = Board_Enter_Standby();
    } else if (strcmp(state->valuestring, "wake") == 0) {
        err = Board_Request_Wake(30000);
    } else {
        return BoardMqtt_ResultFail(BOARD_MQTT_ERR_INVALID_PARAM, "invalid power state");
    }

    return err == ESP_OK ? BoardMqtt_ResultOk("power state updated") :
        BoardMqtt_ResultFail(BOARD_MQTT_ERR_EXEC_FAILED, esp_err_to_name(err));
}

static board_mqtt_action_result_t BoardMqtt_HandleSetClock(cJSON *payload)
{
    cJSON *ts = cJSON_GetObjectItem(payload, "ts");
    if (!cJSON_IsNumber(ts) || ts->valuedouble <= 0) {
        return BoardMqtt_ResultFail(BOARD_MQTT_ERR_INVALID_PARAM, "invalid timestamp");
    }
    esp_err_t err = DS2312_Set_Unix_Time_Saved((time_t)ts->valuedouble);
    return err == ESP_OK ? BoardMqtt_ResultOk("clock updated") :
        BoardMqtt_ResultFail(BOARD_MQTT_ERR_EXEC_FAILED, esp_err_to_name(err));
}

static bool BoardMqtt_PlaybackSpeedIsSupported(float speed)
{
    const float supported_speeds[] = {0.5f, 0.75f, 1.0f, 1.25f, 1.5f, 2.0f};
    for (size_t i = 0; i < sizeof(supported_speeds) / sizeof(supported_speeds[0]); ++i) {
        float diff = speed - supported_speeds[i];
        if (diff < 0) {
            diff = -diff;
        }
        if (diff < 0.001f) {
            return true;
        }
    }
    return false;
}

static char *BoardMqtt_ReadPlaylistFile(void)
{
    FILE *fp = fopen(BOARD_MQTT_PLAYLIST_PATH, "rb");
    if (!fp) {
        return NULL;
    }

    if (fseek(fp, 0, SEEK_END) != 0) {
        fclose(fp);
        return NULL;
    }
    long file_size = ftell(fp);
    if (file_size <= 0 || file_size > BOARD_MQTT_PAYLOAD_MAX_LEN || fseek(fp, 0, SEEK_SET) != 0) {
        fclose(fp);
        return NULL;
    }

    char *buffer = (char *)calloc(1, (size_t)file_size + 1);
    if (!buffer) {
        fclose(fp);
        return NULL;
    }

    size_t read_size = fread(buffer, 1, (size_t)file_size, fp);
    fclose(fp);
    if (read_size != (size_t)file_size) {
        free(buffer);
        return NULL;
    }
    return buffer;
}

static esp_err_t BoardMqtt_WritePlaylistRoot(cJSON *root)
{
    char *json = cJSON_PrintUnformatted(root);
    if (!json) {
        return ESP_ERR_NO_MEM;
    }

    FILE *fp = fopen(BOARD_MQTT_PLAYLIST_PATH, "wb");
    if (!fp) {
        cJSON_free(json);
        return ESP_FAIL;
    }

    size_t len = strlen(json);
    size_t written = fwrite(json, 1, len, fp);
    int close_ret = fclose(fp);
    cJSON_free(json);
    return written == len && close_ret == 0 ? ESP_OK : ESP_FAIL;
}

static cJSON *BoardMqtt_LoadPlaylistRoot(void)
{
    char *json = BoardMqtt_ReadPlaylistFile();
    if (!json) {
        return NULL;
    }
    cJSON *root = cJSON_Parse(json);
    free(json);
    return root;
}

static cJSON *BoardMqtt_GetPlaylistItems(cJSON *root)
{
    cJSON *items = cJSON_IsObject(root) ? cJSON_GetObjectItem(root, "items") : NULL;
    return cJSON_IsArray(items) ? items : NULL;
}

static bool BoardMqtt_PlaylistHasUnfinished(cJSON *root)
{
    cJSON *items = BoardMqtt_GetPlaylistItems(root);
    if (!items) {
        return false;
    }

    cJSON *item = NULL;
    cJSON_ArrayForEach(item, items) {
        cJSON *completed = cJSON_IsObject(item) ? cJSON_GetObjectItem(item, "completed") : NULL;
        if (!cJSON_IsTrue(completed)) {
            return true;
        }
    }
    return false;
}

static esp_err_t BoardMqtt_SavePlaylist(const char *req_id, cJSON *items)
{
    cJSON *root = cJSON_CreateObject();
    cJSON *playlist_items = cJSON_CreateArray();
    if (!root || !playlist_items) {
        cJSON_Delete(root);
        cJSON_Delete(playlist_items);
        return ESP_ERR_NO_MEM;
    }

    cJSON_AddNumberToObject(root, "version", 1);
    cJSON_AddStringToObject(root, "task_id", req_id ? req_id : "PLAY_MEDIA");
    cJSON_AddItemToObject(root, "items", playlist_items);

    if (cJSON_GetArraySize(items) <= 0) {
        cJSON_Delete(root);
        return ESP_ERR_INVALID_ARG;
    }

    cJSON *item = NULL;
    cJSON_ArrayForEach(item, items) {
        if (!cJSON_IsObject(item)) {
            cJSON_Delete(root);
            return ESP_ERR_INVALID_ARG;
        }

        cJSON *path = cJSON_GetObjectItem(item, "path");
        if (!cJSON_IsString(path)) {
            cJSON_Delete(root);
            return ESP_ERR_INVALID_ARG;
        }

        char business_path[SDCARD_PATH_MAX_LEN];
        if (!BoardMqtt_CopyBusinessPath(business_path, sizeof(business_path), path->valuestring)) {
            cJSON_Delete(root);
            return ESP_ERR_INVALID_ARG;
        }

        char sd_path[SDCARD_PATH_MAX_LEN];
        esp_err_t err = SDCard_Map_Business_Path(business_path, sd_path, sizeof(sd_path));
        if (err != ESP_OK) {
            cJSON_Delete(root);
            return err;
        }
        const char *dot = strrchr(sd_path, '.');
        if (!dot || strcasecmp(dot, ".mp3") != 0) {
            cJSON_Delete(root);
            return ESP_ERR_NOT_SUPPORTED;
        }
        if (access(sd_path, F_OK) != 0) {
            cJSON_Delete(root);
            return ESP_ERR_NOT_FOUND;
        }

        int volume_value = 0;
        bool has_volume = false;
        cJSON *volume = cJSON_GetObjectItem(item, "volume");
        if (volume) {
            if (!BoardMqtt_NumberToInt(volume, &volume_value) || volume_value < 0 || volume_value > 100) {
                cJSON_Delete(root);
                return ESP_ERR_INVALID_ARG;
            }
            has_volume = true;
        }

        float speed = 1.0f;
        cJSON *speed_json = cJSON_GetObjectItem(item, "speed");
        if (speed_json) {
            if (!cJSON_IsNumber(speed_json)) {
                cJSON_Delete(root);
                return ESP_ERR_INVALID_ARG;
            }
            speed = (float)speed_json->valuedouble;
        }
        if (!BoardMqtt_PlaybackSpeedIsSupported(speed)) {
            cJSON_Delete(root);
            return ESP_ERR_INVALID_ARG;
        }

        cJSON *stored_item = cJSON_CreateObject();
        if (!stored_item) {
            cJSON_Delete(root);
            return ESP_ERR_NO_MEM;
        }
        cJSON_AddStringToObject(stored_item, "path", business_path);
        cJSON_AddStringToObject(stored_item, "sd_path", sd_path);
        if (has_volume) {
            cJSON_AddNumberToObject(stored_item, "volume", volume_value);
        }
        cJSON_AddNumberToObject(stored_item, "speed", speed);
        cJSON_AddBoolToObject(stored_item, "completed", false);
        cJSON_AddItemToArray(playlist_items, stored_item);
    }

    esp_err_t err = BoardMqtt_WritePlaylistRoot(root);
    cJSON_Delete(root);
    return err;
}

static bool BoardMqtt_LoadNextPlaylistItem(board_mqtt_playlist_item_t *next_item)
{
    if (!next_item) {
        return false;
    }

    cJSON *root = BoardMqtt_LoadPlaylistRoot();
    cJSON *items = BoardMqtt_GetPlaylistItems(root);
    if (!items) {
        cJSON_Delete(root);
        return false;
    }

    cJSON *task_id = cJSON_GetObjectItem(root, "task_id");
    int index = 0;
    int count = cJSON_GetArraySize(items);
    cJSON *item = NULL;
    cJSON_ArrayForEach(item, items) {
        cJSON *completed = cJSON_IsObject(item) ? cJSON_GetObjectItem(item, "completed") : NULL;
        if (!cJSON_IsTrue(completed)) {
            cJSON *path = cJSON_GetObjectItem(item, "path");
            cJSON *sd_path = cJSON_GetObjectItem(item, "sd_path");
            cJSON *volume = cJSON_GetObjectItem(item, "volume");
            cJSON *speed_json = cJSON_GetObjectItem(item, "speed");
            if (!cJSON_IsString(path) || !cJSON_IsString(sd_path)) {
                cJSON_Delete(root);
                return false;
            }

            memset(next_item, 0, sizeof(*next_item));
            if (cJSON_IsString(task_id)) {
                BoardMqtt_CopyString(next_item->task_id, sizeof(next_item->task_id), task_id->valuestring);
            }
            if (next_item->task_id[0] == '\0') {
                BoardMqtt_CopyString(next_item->task_id, sizeof(next_item->task_id), "PLAY_MEDIA");
            }
            if (!BoardMqtt_CopyString(next_item->path, sizeof(next_item->path), path->valuestring) ||
                !BoardMqtt_CopyString(next_item->sd_path, sizeof(next_item->sd_path), sd_path->valuestring)) {
                cJSON_Delete(root);
                return false;
            }
            if (volume) {
                if (!BoardMqtt_NumberToInt(volume, &next_item->volume) || next_item->volume < 0 || next_item->volume > 100) {
                    cJSON_Delete(root);
                    return false;
                }
                next_item->has_volume = true;
            }
            next_item->speed = cJSON_IsNumber(speed_json) ? (float)speed_json->valuedouble : 1.0f;
            if (!BoardMqtt_PlaybackSpeedIsSupported(next_item->speed)) {
                cJSON_Delete(root);
                return false;
            }
            next_item->index = index;
            next_item->count = count;
            cJSON_Delete(root);
            return true;
        }
        ++index;
    }

    cJSON_Delete(root);
    return false;
}

static void BoardMqtt_ShowPlaylistAlert(const char *message, const char *emotion)
{
    Board_Show_Alert("播放清单", message, emotion ? emotion : "music");
}

static esp_err_t BoardMqtt_MarkPlaylistItemCompleted(int target_index)
{
    cJSON *root = BoardMqtt_LoadPlaylistRoot();
    cJSON *items = BoardMqtt_GetPlaylistItems(root);
    if (!items) {
        cJSON_Delete(root);
        return ESP_ERR_NOT_FOUND;
    }

    cJSON *item = cJSON_GetArrayItem(items, target_index);
    if (!cJSON_IsObject(item)) {
        cJSON_Delete(root);
        return ESP_ERR_NOT_FOUND;
    }

    cJSON_DeleteItemFromObject(item, "completed");
    cJSON_AddBoolToObject(item, "completed", true);
    esp_err_t err = BoardMqtt_WritePlaylistRoot(root);
    cJSON_Delete(root);
    return err;
}

static void BoardMqtt_PlaylistTask(void *arg)
{
    (void)arg;
    Board_Acquire_Sleep_Inhibit();

    while (!s_playlist_stop_requested) {
        board_mqtt_playlist_item_t item;
        if (!BoardMqtt_LoadNextPlaylistItem(&item)) {
            if (!s_playlist_stop_requested) {
                BoardMqtt_ShowPlaylistAlert("没有未完成的播放清单", "circle_info");
            }
            break;
        }

        if (item.has_volume) {
            Board_Set_Output_Volume(item.volume);
        }

        char alert_message[96];
        snprintf(alert_message, sizeof(alert_message), "正在播放 %d/%d", item.index + 1, item.count);
        BoardMqtt_ShowPlaylistAlert(alert_message, "music");

        SDCard_Show_Matching_JPG_For_MP3_Path(item.sd_path);
        ESP_LOGI(TAG, "playlist play item %d/%d: %s speed=%.2fx", item.index + 1, item.count, item.sd_path, item.speed);
        esp_err_t err = SDCard_Play_MP3_File_With_Speed(item.sd_path, item.speed);
        if (err != ESP_OK) {
            ESP_LOGE(TAG, "playlist item playback failed: %s", esp_err_to_name(err));
            snprintf(alert_message, sizeof(alert_message), "播放失败 %d/%d: %s", item.index + 1, item.count, esp_err_to_name(err));
            BoardMqtt_ShowPlaylistAlert(alert_message, "circle_xmark");
            break;
        }

        while (!s_playlist_stop_requested && SDCard_Is_Playing()) {
            vTaskDelay(pdMS_TO_TICKS(BOARD_MQTT_PLAYLIST_POLL_MS));
        }
        if (s_playlist_stop_requested) {
            break;
        }

        err = BoardMqtt_MarkPlaylistItemCompleted(item.index);
        if (err != ESP_OK) {
            ESP_LOGE(TAG, "mark playlist item completed failed: %s", esp_err_to_name(err));
            snprintf(alert_message, sizeof(alert_message), "保存播放进度失败: %s", esp_err_to_name(err));
            BoardMqtt_ShowPlaylistAlert(alert_message, "circle_xmark");
            break;
        }
        int progress = item.count > 0 ? ((item.index + 1) * 100) / item.count : 100;
        BoardMqtt_PublishProgress(item.task_id, progress, "playlist item completed");
        if (progress >= 100) {
            BoardMqtt_ShowPlaylistAlert("播放清单已完成", "check");
        }
    }

    BoardMqtt_FinishTask(BOARD_MQTT_TASK_PLAYLIST);
    Board_Release_Sleep_Inhibit();
    s_playlist_stop_requested = false;
    s_playlist_task_handle = NULL;
    vTaskDelete(NULL);
}

static void BoardMqtt_StopPlaylistTask(void)
{
    if (!s_playlist_task_handle) {
        s_playlist_stop_requested = false;
        return;
    }

    s_playlist_stop_requested = true;
    if (SDCard_Is_Playing()) {
        SDCard_Stop_Play();
    }

    int waited_ms = 0;
    while (s_playlist_task_handle && waited_ms < BOARD_MQTT_NETWORK_TIMEOUT_MS) {
        vTaskDelay(pdMS_TO_TICKS(50));
        waited_ms += 50;
    }
}

void BoardMqtt_StopPlaylistPlayback(void)
{
    BoardMqtt_StopPlaylistTask();
    Board_Show_Black_Screen();
}

static esp_err_t BoardMqtt_StartPlaylistTask(void)
{
    BoardMqtt_StopPlaylistTask();
    if (!BoardMqtt_BeginTask(BOARD_MQTT_TASK_PLAYLIST)) {
        return ESP_ERR_INVALID_STATE;
    }

    s_playlist_stop_requested = false;
    BaseType_t ret = xTaskCreate(BoardMqtt_PlaylistTask, "mqtt_playlist", BOARD_MQTT_PLAYLIST_TASK_STACK_SIZE,
                                 NULL, BOARD_MQTT_PLAYLIST_TASK_PRIORITY, &s_playlist_task_handle);
    if (ret != pdPASS) {
        s_playlist_task_handle = NULL;
        BoardMqtt_FinishTask(BOARD_MQTT_TASK_PLAYLIST);
        return ESP_FAIL;
    }
    return ESP_OK;
}

void BoardMqtt_ResumePlaylistIfUnfinished(void)
{
    ESP_LOGI(TAG, "checking unfinished MQTT playlist for offline resume");

    char *json = BoardMqtt_ReadPlaylistFile();
    if (!json) {
        ESP_LOGW(TAG, "no local MQTT playlist file for offline resume: %s", BOARD_MQTT_PLAYLIST_PATH);
        BoardMqtt_ShowPlaylistAlert("没有本地播放清单", "circle_info");
        return;
    }

    cJSON *root = cJSON_Parse(json);
    free(json);
    if (!root) {
        ESP_LOGW(TAG, "local MQTT playlist file is not valid JSON");
        BoardMqtt_ShowPlaylistAlert("本地播放清单格式错误", "circle_xmark");
        return;
    }

    bool has_unfinished = BoardMqtt_PlaylistHasUnfinished(root);
    cJSON_Delete(root);
    if (!has_unfinished) {
        ESP_LOGW(TAG, "local MQTT playlist has no unfinished items");
        BoardMqtt_ShowPlaylistAlert("播放清单已完成", "check");
        return;
    }

    BoardMqtt_ShowPlaylistAlert("Wi-Fi 连接失败，继续播放本地清单", "music");
    esp_err_t err = BoardMqtt_StartPlaylistTask();
    if (err == ESP_OK) {
        ESP_LOGI(TAG, "resumed unfinished MQTT playlist");
    } else {
        ESP_LOGW(TAG, "failed to resume MQTT playlist: %s", esp_err_to_name(err));
        BoardMqtt_ShowPlaylistAlert("本地清单续播失败", "circle_xmark");
    }
}

static board_mqtt_action_result_t BoardMqtt_HandlePlayMedia(const char *req_id, cJSON *payload)
{
    cJSON *items = cJSON_GetObjectItem(payload, "items");
    if (!cJSON_IsArray(items) || cJSON_GetArraySize(items) <= 0) {
        return BoardMqtt_ResultFail(BOARD_MQTT_ERR_INVALID_PARAM, "missing media playlist items");
    }

    BoardMqtt_StopPlaylistTask();

    esp_err_t err = BoardMqtt_SavePlaylist(req_id, items);
    if (err == ESP_ERR_INVALID_ARG) {
        BoardMqtt_ShowPlaylistAlert("播放清单参数无效", "circle_xmark");
        return BoardMqtt_ResultFail(BOARD_MQTT_ERR_INVALID_PARAM, "invalid media playlist item");
    }
    if (err == ESP_ERR_NOT_SUPPORTED) {
        BoardMqtt_ShowPlaylistAlert("播放清单只支持 MP3", "circle_xmark");
        return BoardMqtt_ResultFail(BOARD_MQTT_ERR_UNSUPPORTED, "only mp3 playback is supported");
    }
    if (err == ESP_ERR_NOT_FOUND) {
        BoardMqtt_ShowPlaylistAlert("播放清单文件不存在", "circle_xmark");
        return BoardMqtt_ResultFail(BOARD_MQTT_ERR_NOT_FOUND, "media file not found");
    }
    if (err != ESP_OK) {
        BoardMqtt_ShowPlaylistAlert("保存播放清单失败", "circle_xmark");
        return BoardMqtt_ResultFail(BOARD_MQTT_ERR_EXEC_FAILED, esp_err_to_name(err));
    }

    err = BoardMqtt_StartPlaylistTask();
    if (err != ESP_OK) {
        return BoardMqtt_ResultFail(BOARD_MQTT_ERR_EXEC_FAILED, esp_err_to_name(err));
    }
    BoardMqtt_ShowPlaylistAlert("已接收播放清单", "music");
    BoardMqtt_PublishProgress(req_id, 0, "media playlist accepted");
    return BoardMqtt_ResultOk("media playlist started");
}

static board_mqtt_action_result_t BoardMqtt_HandleStopMedia(cJSON *payload)
{
    (void)payload;
    BoardMqtt_StopPlaylistPlayback();
    return BoardMqtt_ResultOk("media playback stopped");
}

static board_mqtt_action_result_t BoardMqtt_HandleDeleteFile(cJSON *payload)
{
    cJSON *path = cJSON_GetObjectItem(payload, "path");
    if (!cJSON_IsString(path)) {
        return BoardMqtt_ResultFail(BOARD_MQTT_ERR_INVALID_PARAM, "missing file path");
    }
    esp_err_t err = SDCard_Delete_File_By_Business_Path(path->valuestring);
    if (err == ESP_OK) {
        return BoardMqtt_ResultOk("file deleted");
    }
    return BoardMqtt_ResultFail(err == ESP_ERR_NOT_FOUND ? BOARD_MQTT_ERR_NOT_FOUND : BOARD_MQTT_ERR_EXEC_FAILED,
        esp_err_to_name(err));
}

static bool BoardMqtt_ParseTime(const char *value, int *hour, int *minute, int *second)
{
    char tail = 0;
    if (!value || !hour || !minute || !second) {
        return false;
    }
    if (sscanf(value, "%2d:%2d:%2d%c", hour, minute, second, &tail) == 3) {
        return *hour >= 0 && *hour <= 23 && *minute >= 0 && *minute <= 59 && *second >= 0 && *second <= 59;
    }
    *second = 0;
    return sscanf(value, "%2d:%2d%c", hour, minute, &tail) == 2 &&
        *hour >= 0 && *hour <= 23 && *minute >= 0 && *minute <= 59;
}

static board_mqtt_action_result_t BoardMqtt_HandleSetAlarm(cJSON *payload)
{
    cJSON *id = cJSON_GetObjectItem(payload, "id");
    if (!cJSON_IsString(id)) {
        return BoardMqtt_ResultFail(BOARD_MQTT_ERR_INVALID_PARAM, "missing alarm id");
    }

    cJSON *command = cJSON_GetObjectItem(payload, "action");
    if (cJSON_IsString(command) && strcmp(command->valuestring, "delete") == 0) {
        return BoardMqtt_HandleUnsetAlarm(payload);
    }

    cJSON *time_value = cJSON_GetObjectItem(payload, "time");
    cJSON *repeat = cJSON_GetObjectItem(payload, "repeat");
    if (!cJSON_IsString(time_value) || !cJSON_IsArray(repeat) || cJSON_GetArraySize(repeat) == 0) {
        return BoardMqtt_ResultFail(BOARD_MQTT_ERR_INVALID_PARAM, "invalid alarm payload");
    }

    int hour = 0, minute = 0, second = 0;
    if (!BoardMqtt_ParseTime(time_value->valuestring, &hour, &minute, &second)) {
        return BoardMqtt_ResultFail(BOARD_MQTT_ERR_INVALID_PARAM, "invalid alarm time");
    }

    uint8_t repeat_mask = 0;
    cJSON *day = NULL;
    cJSON_ArrayForEach(day, repeat) {
        int value = 0;
        if (!BoardMqtt_NumberToInt(day, &value) || value < 1 || value > 7) {
            return BoardMqtt_ResultFail(BOARD_MQTT_ERR_INVALID_PARAM, "invalid alarm repeat");
        }
        switch (value) {
        case 1: repeat_mask |= ALARM_REPEAT_MON; break;
        case 2: repeat_mask |= ALARM_REPEAT_TUE; break;
        case 3: repeat_mask |= ALARM_REPEAT_WED; break;
        case 4: repeat_mask |= ALARM_REPEAT_THU; break;
        case 5: repeat_mask |= ALARM_REPEAT_FRI; break;
        case 6: repeat_mask |= ALARM_REPEAT_SAT; break;
        case 7: repeat_mask |= ALARM_REPEAT_SUN; break;
        default: break;
        }
    }

    esp_err_t err = DS2312_Set_Alarm_Id_Saved(id->valuestring, hour, minute, second, repeat_mask);
    if (err != ESP_OK) {
        return BoardMqtt_ResultFail(BOARD_MQTT_ERR_EXEC_FAILED, esp_err_to_name(err));
    }

    cJSON *app_id = cJSON_GetObjectItem(payload, "app_id");
    cJSON *task_id = cJSON_GetObjectItem(payload, "task_id");
    if (cJSON_IsString(app_id) && app_id->valuestring && app_id->valuestring[0]) {
        const char *tid = (cJSON_IsString(task_id) && task_id->valuestring) ? task_id->valuestring : "";
        cJSON *params = cJSON_GetObjectItem(payload, "params");
        char *params_json = params ? cJSON_PrintUnformatted(params) : NULL;
        Board_SavePendingTask(app_id->valuestring, tid, params_json, "schedule", id->valuestring);
        free(params_json);
    }
    return BoardMqtt_ResultOk("alarm updated");
}

static board_mqtt_action_result_t BoardMqtt_HandleUnsetAlarm(cJSON *payload)
{
    cJSON *id = cJSON_GetObjectItem(payload, "id");
    if (!cJSON_IsString(id)) {
        return BoardMqtt_ResultFail(BOARD_MQTT_ERR_INVALID_PARAM, "missing alarm id");
    }
    esp_err_t err = DS2312_Delete_Alarm_Id_Saved(id->valuestring);
    if (err == ESP_OK) {
        return BoardMqtt_ResultOk("alarm deleted");
    }
    return BoardMqtt_ResultFail(err == ESP_ERR_NOT_FOUND ? BOARD_MQTT_ERR_NOT_FOUND : BOARD_MQTT_ERR_EXEC_FAILED,
        esp_err_to_name(err));
}

static board_mqtt_action_result_t BoardMqtt_HandleReboot(cJSON *payload)
{
    int delay = 0;
    cJSON *delay_item = cJSON_GetObjectItem(payload, "delay");
    if (delay_item && (!BoardMqtt_NumberToInt(delay_item, &delay) || delay < 0 || delay > 3600)) {
        return BoardMqtt_ResultFail(BOARD_MQTT_ERR_INVALID_PARAM, "invalid reboot delay");
    }
    esp_err_t err = Board_Reboot_Delayed(delay);
    return err == ESP_OK ? BoardMqtt_ResultOk("reboot scheduled") :
        BoardMqtt_ResultFail(BOARD_MQTT_ERR_EXEC_FAILED, esp_err_to_name(err));
}

static bool BoardMqtt_HasPrefix(const char *value, const char *prefix)
{
    return value && prefix && strncmp(value, prefix, strlen(prefix)) == 0;
}

static bool BoardMqtt_IsMd5Hex(const char *value)
{
    if (!value || strlen(value) != 32) {
        return false;
    }
    for (const char *p = value; *p; ++p) {
        if (!((*p >= '0' && *p <= '9') || (*p >= 'a' && *p <= 'f') || (*p >= 'A' && *p <= 'F'))) {
            return false;
        }
    }
    return true;
}

static bool BoardMqtt_IsSafeCourseId(const char *course_id)
{
    if (!course_id || course_id[0] == '\0' || strlen(course_id) >= ALARM_ID_MAX_LEN) {
        return false;
    }
    for (const char *p = course_id; *p; ++p) {
        if (!((*p >= 'a' && *p <= 'z') || (*p >= 'A' && *p <= 'Z') || (*p >= '0' && *p <= '9') ||
              *p == '_' || *p == '-')) {
            return false;
        }
    }
    return true;
}

static bool BoardMqtt_CopyString(char *dest, size_t dest_size, const char *src)
{
    if (!dest || dest_size == 0 || !src) {
        return false;
    }
    int written = snprintf(dest, dest_size, "%s", src);
    return written >= 0 && (size_t)written < dest_size;
}

static bool BoardMqtt_DownloadProgressCallback(int progress, const char *info, void *ctx)
{
    board_mqtt_download_progress_ctx_t *progress_ctx = (board_mqtt_download_progress_ctx_t *)ctx;
    if (!progress_ctx || !progress_ctx->req_id) {
        return true;
    }

    if (s_download_cancel_requested) {
        ESP_LOGI(TAG, "Cancel download by request");
        return false;
    }
    if (Board_Is_Voice_Session_Active()) {
        s_download_voice_yield_requested = true;
        ESP_LOGI(TAG, "Cancel download for voice session");
        return false;
    }

    int64_t now = esp_timer_get_time();
    bool force = progress >= 100 || progress_ctx->last_progress < 0;
    if (!force && progress == progress_ctx->last_progress) {
        return true;
    }
    if (!force && progress_ctx->last_publish_us > 0 && now - progress_ctx->last_publish_us < 2000000) {
        return true;
    }

    progress_ctx->last_publish_us = now;
    progress_ctx->last_progress = progress;
    BoardMqtt_PublishProgress(progress_ctx->req_id, progress, info);
    return true;
}

static bool BoardMqtt_ManifestProgressCallback(int progress, const char *info, void *ctx)
{
    board_mqtt_manifest_progress_ctx_t *progress_ctx = (board_mqtt_manifest_progress_ctx_t *)ctx;
    if (!progress_ctx || !progress_ctx->req_id || progress_ctx->asset_count == 0) {
        return true;
    }

    if (s_download_cancel_requested) {
        ESP_LOGI(TAG, "Cancel manifest download by request");
        return false;
    }
    if (Board_Is_Voice_Session_Active()) {
        s_download_voice_yield_requested = true;
        ESP_LOGI(TAG, "Cancel manifest download for voice session");
        return false;
    }

    if (progress < 0) {
        progress = 0;
    } else if (progress > 100) {
        progress = 100;
    }
    int aggregate_progress = (int)(((progress_ctx->asset_index * 100) + (size_t)progress) / progress_ctx->asset_count);

    int64_t now = esp_timer_get_time();
    bool force = progress <= 0 || progress >= 100 || aggregate_progress >= 100;
    if (!force && aggregate_progress == progress_ctx->last_progress) {
        return true;
    }
    if (!force && progress_ctx->last_publish_us > 0 && now - progress_ctx->last_publish_us < 2000000) {
        return true;
    }

    char progress_info[BOARD_MQTT_DOWNLOAD_INFO_MAX_LEN];
    snprintf(progress_info, sizeof(progress_info), "asset %u/%u %s",
             (unsigned)(progress_ctx->asset_index + 1), (unsigned)progress_ctx->asset_count, info ? info : "downloading");
    progress_ctx->last_publish_us = now;
    progress_ctx->last_progress = aggregate_progress;
    BoardMqtt_PublishProgress(progress_ctx->req_id, aggregate_progress, progress_info);
    return true;
}

static void BoardMqtt_OtaProgressCallback(int progress, size_t speed, void *ctx)
{
    board_mqtt_ota_job_t *job = (board_mqtt_ota_job_t *)ctx;
    if (!job) {
        return;
    }

    int64_t now = esp_timer_get_time();
    bool force = progress >= 100 || job->last_progress < 0;
    if (!force && progress == job->last_progress) {
        return;
    }
    if (!force && job->last_publish_us > 0 && now - job->last_publish_us < 2000000) {
        return;
    }

    job->last_publish_us = now;
    job->last_progress = progress;
    char info[BOARD_MQTT_DOWNLOAD_INFO_MAX_LEN];
    snprintf(info, sizeof(info), "upgrading firmware %uKB/s", (unsigned)(speed / 1024));
    BoardMqtt_PublishProgress(job->req_id, progress, info);
}

static void BoardMqtt_OtaCompleteCallback(bool success, void *ctx)
{
    board_mqtt_ota_job_t *job = (board_mqtt_ota_job_t *)ctx;
    if (!job) {
        BoardMqtt_FinishTask(BOARD_MQTT_TASK_OTA);
        return;
    }

    BoardMqtt_WaitConnectedForFinalReport();
    if (success) {
        BoardMqtt_PublishProgress(job->req_id, 100, "ota completed");
        BoardMqtt_PublishAck(job->req_id, true, 0, "ota completed, rebooting");
    } else {
        BoardMqtt_PublishAck(job->req_id, false, BOARD_MQTT_ERR_EXEC_FAILED, "ota failed");
    }
    BoardMqtt_FinishTask(BOARD_MQTT_TASK_OTA);
    free(job);
}

static void BoardMqtt_OtaTask(void *arg)
{
    board_mqtt_ota_job_t *job = (board_mqtt_ota_job_t *)arg;
    if (!job) {
        vTaskDelete(NULL);
        return;
    }

    BoardMqtt_PublishProgress(job->req_id, 0, "starting ota");
    esp_err_t err = Board_Start_Firmware_Upgrade(job->url, job->version,
                                                 BoardMqtt_OtaProgressCallback,
                                                 BoardMqtt_OtaCompleteCallback,
                                                 job);
    if (err != ESP_OK) {
        BoardMqtt_PublishAck(job->req_id, false, BOARD_MQTT_ERR_EXEC_FAILED, esp_err_to_name(err));
        BoardMqtt_FinishTask(BOARD_MQTT_TASK_OTA);
        free(job);
    }
    vTaskDelete(NULL);
}

static void BoardMqtt_ManifestTask(void *arg)
{
    board_mqtt_manifest_job_t *job = (board_mqtt_manifest_job_t *)arg;
    if (!job) {
        vTaskDelete(NULL);
        return;
    }

    if (job->replace_existing_download) {
        if (!BoardMqtt_WaitDownloadCanceled(30000) || !BoardMqtt_ActivatePendingManifest()) {
            BoardMqtt_ClearPendingManifest();
            BoardMqtt_PublishAck(job->req_id, false, BOARD_MQTT_ERR_BUSY, "previous download stopping");
            cJSON_free(job->manifest_json);
            free(job);
            vTaskDelete(NULL);
            return;
        }
    }

    Board_Acquire_Sleep_Inhibit();
    s_download_voice_yield_requested = false;

    char final_path[SDCARD_PATH_MAX_LEN] = {0};
    char part_path[SDCARD_PATH_MAX_LEN] = {0};
    char error_info[BOARD_MQTT_DOWNLOAD_INFO_MAX_LEN] = {0};
    esp_err_t err = ESP_OK;

    vTaskDelay(pdMS_TO_TICKS(100));
    BoardMqtt_PublishProgress(job->req_id, 0, "saving manifest");
    err = SDCard_Save_Manifest(job->course_id, job->manifest_json);
    if (err != ESP_OK) {
        snprintf(error_info, sizeof(error_info), "%s", esp_err_to_name(err));
    }

    for (size_t i = 0; err == ESP_OK && i < job->asset_count; ++i) {
        board_mqtt_manifest_asset_t *asset = &job->assets[i];
        memset(final_path, 0, sizeof(final_path));
        memset(part_path, 0, sizeof(part_path));
        memset(error_info, 0, sizeof(error_info));

        err = SDCard_Prepare_Business_File(asset->target_path, final_path, sizeof(final_path), part_path, sizeof(part_path));
        if (err != ESP_OK) {
            snprintf(error_info, sizeof(error_info), "%s", esp_err_to_name(err));
            break;
        }

        board_mqtt_manifest_progress_ctx_t progress_ctx = {
            .req_id = job->req_id,
            .asset_index = i,
            .asset_count = job->asset_count,
            .last_publish_us = 0,
            .last_progress = -1,
        };
        err = Board_Download_File_To_Path(asset->url, part_path, asset->md5, BoardMqtt_ManifestProgressCallback,
                                          &progress_ctx, error_info, sizeof(error_info));
        if (err != ESP_OK) {
            SDCard_Remove_File_If_Exists(part_path);
            if (s_download_voice_yield_requested || s_download_cancel_requested) {
                int canceled_progress = progress_ctx.last_progress >= 0 ? progress_ctx.last_progress : 0;
                const char *cancel_info = s_download_voice_yield_requested ?
                    "download canceled for voice session" : "download canceled";
                s_download_voice_yield_requested = false;
                BoardMqtt_PublishProgress(job->req_id, canceled_progress, cancel_info);
                snprintf(error_info, sizeof(error_info), "%s", cancel_info);
            }
            break;
        }

        err = SDCard_Commit_Part_File(part_path, final_path);
        if (err != ESP_OK) {
            snprintf(error_info, sizeof(error_info), "%s", esp_err_to_name(err));
            SDCard_Remove_File_If_Exists(part_path);
            break;
        }

        int completed_progress = (int)(((i + 1) * 100) / job->asset_count);
        char progress_info[BOARD_MQTT_DOWNLOAD_INFO_MAX_LEN];
        snprintf(progress_info, sizeof(progress_info), "asset %u/%u completed",
                 (unsigned)(i + 1), (unsigned)job->asset_count);
        BoardMqtt_PublishProgress(job->req_id, completed_progress, progress_info);
    }

    bool replaced_by_new_manifest = err != ESP_OK && s_download_cancel_requested && BoardMqtt_HasPendingManifest();
    if (!replaced_by_new_manifest) {
        BoardMqtt_WaitConnectedForFinalReport();
        if (err == ESP_OK) {
            BoardMqtt_PublishProgress(job->req_id, 100, "manifest assets downloaded");
            BoardMqtt_PublishAck(job->req_id, true, 0, "manifest saved and assets downloaded");
        } else {
            BoardMqtt_PublishAck(job->req_id, false, BOARD_MQTT_ERR_EXEC_FAILED,
                                 error_info[0] ? error_info : esp_err_to_name(err));
        }
    }

    s_download_voice_yield_requested = false;
    Board_Release_Sleep_Inhibit();
    BoardMqtt_FinishTask(BOARD_MQTT_TASK_MANIFEST);
    cJSON_free(job->manifest_json);
    free(job);
    vTaskDelete(NULL);
}

static void BoardMqtt_DownloadTask(void *arg)
{
    board_mqtt_download_job_t *job = (board_mqtt_download_job_t *)arg;
    if (!job) {
        vTaskDelete(NULL);
        return;
    }

    Board_Acquire_Sleep_Inhibit();
    s_download_voice_yield_requested = false;

    char final_path[SDCARD_PATH_MAX_LEN] = {0};
    char part_path[SDCARD_PATH_MAX_LEN] = {0};
    char error_info[BOARD_MQTT_DOWNLOAD_INFO_MAX_LEN] = {0};
    board_mqtt_download_progress_ctx_t progress_ctx = {
        .req_id = job->req_id,
        .last_publish_us = 0,
        .last_progress = -1,
    };

    BoardMqtt_PublishProgress(job->req_id, 0, "starting download");
    esp_err_t err = SDCard_Prepare_Business_File(job->target_path, final_path, sizeof(final_path), part_path, sizeof(part_path));
    if (err == ESP_OK) {
        err = Board_Download_File_To_Path(job->url, part_path, job->md5, BoardMqtt_DownloadProgressCallback,
                                          &progress_ctx, error_info, sizeof(error_info));
    } else {
        snprintf(error_info, sizeof(error_info), "%s", esp_err_to_name(err));
    }

    if (err != ESP_OK) {
        SDCard_Remove_File_If_Exists(part_path);
        if (s_download_voice_yield_requested || s_download_cancel_requested) {
            int canceled_progress = progress_ctx.last_progress >= 0 ? progress_ctx.last_progress : 0;
            const char *cancel_info = s_download_voice_yield_requested ?
                "download canceled for voice session" : "download canceled";
            s_download_voice_yield_requested = false;
            BoardMqtt_PublishProgress(job->req_id, canceled_progress, cancel_info);
            snprintf(error_info, sizeof(error_info), "%s", cancel_info);
        }
    } else {
        err = SDCard_Commit_Part_File(part_path, final_path);
        if (err != ESP_OK) {
            snprintf(error_info, sizeof(error_info), "%s", esp_err_to_name(err));
            SDCard_Remove_File_If_Exists(part_path);
        }
    }

    bool replaced_by_new_manifest = err != ESP_OK && s_download_cancel_requested && BoardMqtt_HasPendingManifest();
    if (!replaced_by_new_manifest) {
        BoardMqtt_WaitConnectedForFinalReport();
        if (err == ESP_OK) {
            BoardMqtt_PublishProgress(job->req_id, 100, "download completed");
            BoardMqtt_PublishAck(job->req_id, true, 0, "download completed");
        } else {
            BoardMqtt_PublishAck(job->req_id, false, BOARD_MQTT_ERR_EXEC_FAILED,
                                 error_info[0] ? error_info : esp_err_to_name(err));
        }
    }

    s_download_voice_yield_requested = false;
    BoardMqtt_FinishTask(BOARD_MQTT_TASK_DOWNLOAD);
    Board_Release_Sleep_Inhibit();
    free(job);
    vTaskDelete(NULL);
}

static board_mqtt_action_result_t BoardMqtt_HandleDownload(const char *req_id, cJSON *payload)
{
    cJSON *url = cJSON_GetObjectItem(payload, "url");
    cJSON *target_path = cJSON_GetObjectItem(payload, "target_path");
    cJSON *md5 = cJSON_GetObjectItem(payload, "md5");
    if (!cJSON_IsString(url) || !cJSON_IsString(target_path)) {
        return BoardMqtt_ResultFail(BOARD_MQTT_ERR_INVALID_PARAM, "invalid download payload");
    }
    if (!BoardMqtt_HasPrefix(url->valuestring, "http://") && !BoardMqtt_HasPrefix(url->valuestring, "https://")) {
        return BoardMqtt_ResultFail(BOARD_MQTT_ERR_INVALID_PARAM, "invalid download url");
    }
    if (!BoardMqtt_ValidateOptionalMd5(md5)) {
        return BoardMqtt_ResultFail(BOARD_MQTT_ERR_INVALID_PARAM, "invalid md5");
    }

    char business_path[SDCARD_PATH_MAX_LEN];
    if (!BoardMqtt_CopyBusinessPath(business_path, sizeof(business_path), target_path->valuestring)) {
        return BoardMqtt_ResultFail(BOARD_MQTT_ERR_INVALID_PARAM, "invalid target path");
    }

    char sd_path[SDCARD_PATH_MAX_LEN];
    esp_err_t err = SDCard_Map_Business_Path(business_path, sd_path, sizeof(sd_path));
    if (err != ESP_OK) {
        return BoardMqtt_ResultFail(BOARD_MQTT_ERR_INVALID_PARAM, "invalid target path");
    }
    if (!BoardMqtt_IsSupportedResourcePath(sd_path)) {
        return BoardMqtt_ResultFail(BOARD_MQTT_ERR_UNSUPPORTED, "only mp3 and jpg download is supported");
    }
    BoardMqtt_StopPlaylistTask();
    if (!BoardMqtt_BeginTask(BOARD_MQTT_TASK_DOWNLOAD)) {
        return BoardMqtt_ResultFail(BOARD_MQTT_ERR_BUSY, "download busy");
    }

    board_mqtt_download_job_t *job = (board_mqtt_download_job_t *)calloc(1, sizeof(board_mqtt_download_job_t));
    if (!job) {
        BoardMqtt_FinishTask(BOARD_MQTT_TASK_DOWNLOAD);
        return BoardMqtt_ResultFail(BOARD_MQTT_ERR_EXEC_FAILED, "no memory for download job");
    }
    if (!BoardMqtt_CopyString(job->req_id, sizeof(job->req_id), req_id) ||
        !BoardMqtt_CopyString(job->url, sizeof(job->url), url->valuestring) ||
        !BoardMqtt_CopyString(job->target_path, sizeof(job->target_path), business_path) ||
        !BoardMqtt_CopyOptionalMd5(job->md5, sizeof(job->md5), md5)) {
        BoardMqtt_FinishTask(BOARD_MQTT_TASK_DOWNLOAD);
        free(job);
        return BoardMqtt_ResultFail(BOARD_MQTT_ERR_INVALID_PARAM, "download field too long");
    }

    if (SDCard_Is_Playing()) {
        ESP_LOGI(TAG, "Stop MP3 playback before MQTT download");
        SDCard_Stop_Play();
    }

    BaseType_t ret = xTaskCreate(BoardMqtt_DownloadTask, "mqtt_download", 6144, job, 5, NULL);
    if (ret != pdPASS) {
        BoardMqtt_FinishTask(BOARD_MQTT_TASK_DOWNLOAD);
        free(job);
        return BoardMqtt_ResultFail(BOARD_MQTT_ERR_EXEC_FAILED, "failed to create download task");
    }
    return BoardMqtt_ResultOk("download accepted");
}

static board_mqtt_action_result_t BoardMqtt_HandleOtaUpdate(const char *req_id, cJSON *payload)
{
    cJSON *url = cJSON_GetObjectItem(payload, "url");
    cJSON *version = cJSON_GetObjectItem(payload, "version");
    if (!cJSON_IsString(url)) {
        return BoardMqtt_ResultFail(BOARD_MQTT_ERR_INVALID_PARAM, "missing ota url");
    }
    if (!BoardMqtt_HasPrefix(url->valuestring, "http://") && !BoardMqtt_HasPrefix(url->valuestring, "https://")) {
        return BoardMqtt_ResultFail(BOARD_MQTT_ERR_INVALID_PARAM, "invalid ota url");
    }
    if (version && !cJSON_IsString(version)) {
        return BoardMqtt_ResultFail(BOARD_MQTT_ERR_INVALID_PARAM, "invalid ota version");
    }
    if (!BoardMqtt_BeginTask(BOARD_MQTT_TASK_OTA)) {
        return BoardMqtt_ResultFail(BOARD_MQTT_ERR_BUSY, "task busy");
    }

    board_mqtt_ota_job_t *job = (board_mqtt_ota_job_t *)calloc(1, sizeof(board_mqtt_ota_job_t));
    if (!job) {
        BoardMqtt_FinishTask(BOARD_MQTT_TASK_OTA);
        return BoardMqtt_ResultFail(BOARD_MQTT_ERR_EXEC_FAILED, "no memory for ota job");
    }
    job->last_progress = -1;
    if (!BoardMqtt_CopyString(job->req_id, sizeof(job->req_id), req_id) ||
        !BoardMqtt_CopyString(job->url, sizeof(job->url), url->valuestring)) {
        BoardMqtt_FinishTask(BOARD_MQTT_TASK_OTA);
        free(job);
        return BoardMqtt_ResultFail(BOARD_MQTT_ERR_INVALID_PARAM, "ota field too long");
    }
    if (cJSON_IsString(version) && version->valuestring[0] != '\0' &&
        !BoardMqtt_CopyString(job->version, sizeof(job->version), version->valuestring)) {
        BoardMqtt_FinishTask(BOARD_MQTT_TASK_OTA);
        free(job);
        return BoardMqtt_ResultFail(BOARD_MQTT_ERR_INVALID_PARAM, "ota version too long");
    }

    BaseType_t ret = xTaskCreate(BoardMqtt_OtaTask, "mqtt_ota", 4096, job, 5, NULL);
    if (ret != pdPASS) {
        BoardMqtt_FinishTask(BOARD_MQTT_TASK_OTA);
        free(job);
        return BoardMqtt_ResultFail(BOARD_MQTT_ERR_EXEC_FAILED, "failed to create ota task");
    }
    return BoardMqtt_ResultOk("ota accepted");
}

static board_mqtt_action_result_t BoardMqtt_HandleUpdateManifest(const char *req_id, cJSON *payload)
{
    cJSON *course_id = cJSON_GetObjectItem(payload, "course_id");
    cJSON *assets = cJSON_GetObjectItem(payload, "assets");
    if (!cJSON_IsString(course_id) || !BoardMqtt_IsSafeCourseId(course_id->valuestring) ||
        !cJSON_IsArray(assets) || cJSON_GetArraySize(assets) <= 0) {
        return BoardMqtt_ResultFail(BOARD_MQTT_ERR_INVALID_PARAM, "invalid manifest payload");
    }
    BoardMqtt_StopPlaylistTask();

    bool replace_existing_download = false;
    board_mqtt_task_type_t current_type = BoardMqtt_GetTaskType();
    if (BoardMqtt_IsDownloadTask(current_type)) {
        ESP_LOGI(TAG, "Cancel current download before starting new manifest task");
        if (!BoardMqtt_BeginManifestReplacement()) {
            return BoardMqtt_ResultFail(BOARD_MQTT_ERR_BUSY, "task busy");
        }
        replace_existing_download = true;
    } else if (!BoardMqtt_BeginTask(BOARD_MQTT_TASK_MANIFEST)) {
        return BoardMqtt_ResultFail(BOARD_MQTT_ERR_BUSY, "task busy");
    }

#define BOARD_MQTT_RETURN_MANIFEST_FAIL(code, message) do { \
        if (replace_existing_download) { \
            BoardMqtt_ClearPendingManifest(); \
        } else { \
            BoardMqtt_FinishTask(BOARD_MQTT_TASK_MANIFEST); \
        } \
        return BoardMqtt_ResultFail((code), (message)); \
    } while (0)

    int asset_count_int = cJSON_GetArraySize(assets);
    size_t source_asset_count = (size_t)asset_count_int;
    if (source_asset_count > SIZE_MAX / 2) {
        BOARD_MQTT_RETURN_MANIFEST_FAIL(BOARD_MQTT_ERR_EXEC_FAILED, "manifest asset count too large");
    }
    size_t max_asset_count = source_asset_count * 2;
    if (max_asset_count > (SIZE_MAX - sizeof(board_mqtt_manifest_job_t)) / sizeof(board_mqtt_manifest_asset_t)) {
        BOARD_MQTT_RETURN_MANIFEST_FAIL(BOARD_MQTT_ERR_EXEC_FAILED, "manifest asset count too large");
    }

    char *manifest_json = cJSON_PrintUnformatted((cJSON *)payload);
    if (!manifest_json) {
        BOARD_MQTT_RETURN_MANIFEST_FAIL(BOARD_MQTT_ERR_EXEC_FAILED, "failed to format manifest");
    }

    size_t job_size = sizeof(board_mqtt_manifest_job_t) + max_asset_count * sizeof(board_mqtt_manifest_asset_t);
    board_mqtt_manifest_job_t *job = (board_mqtt_manifest_job_t *)calloc(1, job_size);
    if (!job) {
        cJSON_free(manifest_json);
        BOARD_MQTT_RETURN_MANIFEST_FAIL(BOARD_MQTT_ERR_EXEC_FAILED, "no memory for manifest job");
    }
    job->manifest_json = manifest_json;
    job->replace_existing_download = replace_existing_download;

#define BOARD_MQTT_FREE_MANIFEST_JOB_AND_FAIL(code, message) do { \
        cJSON_free(job->manifest_json); \
        free(job); \
        BOARD_MQTT_RETURN_MANIFEST_FAIL((code), (message)); \
    } while (0)

    if (!BoardMqtt_CopyString(job->req_id, sizeof(job->req_id), req_id) ||
        !BoardMqtt_CopyString(job->course_id, sizeof(job->course_id), course_id->valuestring)) {
        BOARD_MQTT_FREE_MANIFEST_JOB_AND_FAIL(BOARD_MQTT_ERR_INVALID_PARAM, "manifest field too long");
    }

    size_t index = 0;
    cJSON *asset = NULL;
    cJSON_ArrayForEach(asset, assets) {
        cJSON *url = cJSON_IsObject(asset) ? cJSON_GetObjectItem(asset, "url") : NULL;
        cJSON *target_path = cJSON_IsObject(asset) ? cJSON_GetObjectItem(asset, "target_path") : NULL;
        cJSON *md5 = cJSON_IsObject(asset) ? cJSON_GetObjectItem(asset, "md5") : NULL;
        if (!cJSON_IsString(url) || !cJSON_IsString(target_path)) {
            BOARD_MQTT_FREE_MANIFEST_JOB_AND_FAIL(BOARD_MQTT_ERR_INVALID_PARAM, "invalid manifest asset");
        }
        if (!BoardMqtt_HasPrefix(url->valuestring, "http://") && !BoardMqtt_HasPrefix(url->valuestring, "https://")) {
            BOARD_MQTT_FREE_MANIFEST_JOB_AND_FAIL(BOARD_MQTT_ERR_INVALID_PARAM, "invalid manifest asset url");
        }
        if (!BoardMqtt_ValidateOptionalMd5(md5)) {
            BOARD_MQTT_FREE_MANIFEST_JOB_AND_FAIL(BOARD_MQTT_ERR_INVALID_PARAM, "invalid manifest asset md5");
        }

        char business_path[SDCARD_PATH_MAX_LEN];
        if (!BoardMqtt_CopyBusinessPath(business_path, sizeof(business_path), target_path->valuestring)) {
            BOARD_MQTT_FREE_MANIFEST_JOB_AND_FAIL(BOARD_MQTT_ERR_INVALID_PARAM, "invalid manifest target path");
        }

        char sd_path[SDCARD_PATH_MAX_LEN];
        if (SDCard_Map_Business_Path(business_path, sd_path, sizeof(sd_path)) != ESP_OK) {
            BOARD_MQTT_FREE_MANIFEST_JOB_AND_FAIL(BOARD_MQTT_ERR_INVALID_PARAM, "invalid manifest target path");
        }
        if (!BoardMqtt_IsSupportedResourcePath(sd_path)) {
            BOARD_MQTT_FREE_MANIFEST_JOB_AND_FAIL(BOARD_MQTT_ERR_UNSUPPORTED, "only mp3 and jpg manifest assets are supported");
        }

        board_mqtt_manifest_asset_t *job_asset = &job->assets[index];
        if (!BoardMqtt_CopyString(job_asset->url, sizeof(job_asset->url), url->valuestring) ||
            !BoardMqtt_CopyString(job_asset->target_path, sizeof(job_asset->target_path), business_path) ||
            !BoardMqtt_CopyOptionalMd5(job_asset->md5, sizeof(job_asset->md5), md5)) {
            BOARD_MQTT_FREE_MANIFEST_JOB_AND_FAIL(BOARD_MQTT_ERR_INVALID_PARAM, "manifest asset field too long");
        }
        ++index;

        cJSON *cover_url = cJSON_IsObject(asset) ? cJSON_GetObjectItem(asset, "cover_url") : NULL;
        cJSON *cover_target_path = cJSON_IsObject(asset) ? cJSON_GetObjectItem(asset, "cover_target_path") : NULL;
        if (!cJSON_IsString(cover_url) || !cJSON_IsString(cover_target_path) ||
            cover_url->valuestring[0] == '\0' || cover_target_path->valuestring[0] == '\0') {
            continue;
        }
        if (!BoardMqtt_HasPrefix(cover_url->valuestring, "http://") && !BoardMqtt_HasPrefix(cover_url->valuestring, "https://")) {
            BOARD_MQTT_FREE_MANIFEST_JOB_AND_FAIL(BOARD_MQTT_ERR_INVALID_PARAM, "invalid manifest cover url");
        }
        if (!BoardMqtt_CopyBusinessPath(business_path, sizeof(business_path), cover_target_path->valuestring)) {
            BOARD_MQTT_FREE_MANIFEST_JOB_AND_FAIL(BOARD_MQTT_ERR_INVALID_PARAM, "invalid manifest cover path");
        }
        if (SDCard_Map_Business_Path(business_path, sd_path, sizeof(sd_path)) != ESP_OK) {
            BOARD_MQTT_FREE_MANIFEST_JOB_AND_FAIL(BOARD_MQTT_ERR_INVALID_PARAM, "invalid manifest cover path");
        }
        const char *cover_dot = strrchr(sd_path, '.');
        if (!cover_dot || strcasecmp(cover_dot, ".jpg") != 0) {
            BOARD_MQTT_FREE_MANIFEST_JOB_AND_FAIL(BOARD_MQTT_ERR_UNSUPPORTED, "only jpg cover assets are supported");
        }

        job_asset = &job->assets[index];
        if (!BoardMqtt_CopyString(job_asset->url, sizeof(job_asset->url), cover_url->valuestring) ||
            !BoardMqtt_CopyString(job_asset->target_path, sizeof(job_asset->target_path), business_path)) {
            BOARD_MQTT_FREE_MANIFEST_JOB_AND_FAIL(BOARD_MQTT_ERR_INVALID_PARAM, "manifest cover field too long");
        }
        ++index;
    }
    job->asset_count = index;

    if (SDCard_Is_Playing()) {
        ESP_LOGI(TAG, "Stop MP3 playback before MQTT manifest download");
        SDCard_Stop_Play();
    }

    BaseType_t ret = xTaskCreate(BoardMqtt_ManifestTask, "mqtt_manifest", 6144, job, 5, NULL);
    if (ret != pdPASS) {
        cJSON_free(job->manifest_json);
        free(job);
        BOARD_MQTT_RETURN_MANIFEST_FAIL(BOARD_MQTT_ERR_EXEC_FAILED, "failed to create manifest task");
    }
#undef BOARD_MQTT_FREE_MANIFEST_JOB_AND_FAIL
#undef BOARD_MQTT_RETURN_MANIFEST_FAIL
    return BoardMqtt_ResultOk("manifest accepted");
}

static board_mqtt_action_result_t BoardMqtt_HandleStartAppTask(cJSON *payload)
{
    if (!payload) {
        return BoardMqtt_ResultFail(BOARD_MQTT_ERR_INVALID_PARAM, "missing payload");
    }
    cJSON *app_id = cJSON_GetObjectItem(payload, "app_id");
    cJSON *task_id = cJSON_GetObjectItem(payload, "task_id");
    cJSON *source = cJSON_GetObjectItem(payload, "source");
    cJSON *params = cJSON_GetObjectItem(payload, "params");
    if (!cJSON_IsString(app_id) || !app_id->valuestring || app_id->valuestring[0] == '\0') {
        return BoardMqtt_ResultFail(BOARD_MQTT_ERR_INVALID_PARAM, "missing app_id");
    }
    const char *tid = (cJSON_IsString(task_id) && task_id->valuestring) ? task_id->valuestring : "";
    const char *src = (cJSON_IsString(source) && source->valuestring) ? source->valuestring : "mqtt_immediate";
    char *params_json = params ? cJSON_PrintUnformatted(params) : NULL;
    esp_err_t err = Board_StartAppTask(app_id->valuestring, tid, params_json, src);
    free(params_json);
    if (err == ESP_ERR_NOT_FOUND) {
        return BoardMqtt_ResultFail(BOARD_MQTT_ERR_NOT_FOUND, "unknown app or task");
    }
    if (err == ESP_ERR_INVALID_STATE) {
        return BoardMqtt_ResultFail(BOARD_MQTT_ERR_BUSY, "preempted by higher priority task");
    }
    if (err != ESP_OK) {
        return BoardMqtt_ResultFail(BOARD_MQTT_ERR_EXEC_FAILED, esp_err_to_name(err));
    }
    return BoardMqtt_ResultOk("app task started");
}

static board_mqtt_action_result_t BoardMqtt_DispatchAction(const char *req_id, const char *action, cJSON *payload)
{
    if (strcmp(action, "SET_VOLUME") == 0) return BoardMqtt_HandleSetVolume(payload);
    if (strcmp(action, "SET_POWER") == 0) return BoardMqtt_HandleSetPower(payload);
    if (strcmp(action, "SET_CLOCK") == 0) return BoardMqtt_HandleSetClock(payload);
    if (strcmp(action, "PLAY_MEDIA") == 0) return BoardMqtt_HandlePlayMedia(req_id, payload);
    if (strcmp(action, "STOP_MEDIA") == 0) return BoardMqtt_HandleStopMedia(payload);
    if (strcmp(action, "DELETE_FILE") == 0) return BoardMqtt_HandleDeleteFile(payload);
    if (strcmp(action, "SET_ALARM") == 0) return BoardMqtt_HandleSetAlarm(payload);
    if (strcmp(action, "UNSET_ALARM") == 0) return BoardMqtt_HandleUnsetAlarm(payload);
    if (strcmp(action, "REBOOT") == 0) return BoardMqtt_HandleReboot(payload);
    if (strcmp(action, "UPDATE_MANIFEST") == 0 || strcmp(action, "SYNC_COURSE_TASK") == 0) return BoardMqtt_HandleUpdateManifest(req_id, payload);
    if (strcmp(action, "DOWNLOAD") == 0) return BoardMqtt_HandleDownload(req_id, payload);
    if (strcmp(action, "OTA_UPDATE") == 0) return BoardMqtt_HandleOtaUpdate(req_id, payload);
    if (strcmp(action, "START_APP_TASK") == 0) return BoardMqtt_HandleStartAppTask(payload);
    return BoardMqtt_ResultFail(BOARD_MQTT_ERR_UNSUPPORTED, "unsupported action");
}

static board_mqtt_topic_type_t BoardMqtt_GetTopicType(const char *topic)
{
    if (strcmp(topic, s_cmd_ctl_topic) == 0) {
        return BOARD_MQTT_TOPIC_CTL;
    }
    if (strcmp(topic, s_cmd_task_topic) == 0) {
        return BOARD_MQTT_TOPIC_TASK;
    }
    return BOARD_MQTT_TOPIC_UNKNOWN;
}

static void BoardMqtt_ResetPayloadAssembly(void)
{
    free(s_payload_assembly.payload);
    memset(&s_payload_assembly, 0, sizeof(s_payload_assembly));
}

static void BoardMqtt_HandleDataEvent(esp_mqtt_event_t *event)
{
    if (!event || event->data_len <= 0 || !event->data || event->total_data_len <= 0 || event->current_data_offset < 0 ||
        event->current_data_offset > event->total_data_len || event->data_len > event->total_data_len - event->current_data_offset) {
        ESP_LOGW(TAG, "Invalid MQTT data event");
        BoardMqtt_ResetPayloadAssembly();
        return;
    }

    if (event->total_data_len > BOARD_MQTT_PAYLOAD_MAX_LEN) {
        ESP_LOGW(TAG, "MQTT payload too large: %d", event->total_data_len);
        BoardMqtt_ResetPayloadAssembly();
        return;
    }

    if (event->current_data_offset == 0) {
        BoardMqtt_ResetPayloadAssembly();
        if (!event->topic || event->topic_len <= 0 || event->topic_len >= BOARD_MQTT_TOPIC_MAX_LEN) {
            ESP_LOGW(TAG, "Invalid MQTT command topic");
            return;
        }

        if (event->data_len == event->total_data_len) {
            BoardMqtt_HandleCommand(event->topic, event->topic_len, event->data, event->data_len);
            return;
        }

        s_payload_assembly.payload = (char *)calloc(1, event->total_data_len + 1);
        if (!s_payload_assembly.payload) {
            ESP_LOGE(TAG, "Failed to allocate MQTT payload assembly buffer");
            BoardMqtt_ResetPayloadAssembly();
            return;
        }
        memcpy(s_payload_assembly.topic, event->topic, event->topic_len);
        s_payload_assembly.topic[event->topic_len] = '\0';
        s_payload_assembly.topic_len = event->topic_len;
        s_payload_assembly.total_len = event->total_data_len;
        s_payload_assembly.received_len = 0;
    } else if (!s_payload_assembly.payload || s_payload_assembly.total_len != event->total_data_len ||
               event->current_data_offset != s_payload_assembly.received_len) {
        ESP_LOGW(TAG, "Invalid MQTT payload fragment sequence");
        BoardMqtt_ResetPayloadAssembly();
        return;
    }

    memcpy(s_payload_assembly.payload + event->current_data_offset, event->data, event->data_len);
    s_payload_assembly.received_len += event->data_len;

    if (s_payload_assembly.received_len == s_payload_assembly.total_len) {
        BoardMqtt_HandleCommand(s_payload_assembly.topic, s_payload_assembly.topic_len,
                                s_payload_assembly.payload, s_payload_assembly.total_len);
        BoardMqtt_ResetPayloadAssembly();
    } else if (s_payload_assembly.received_len > s_payload_assembly.total_len) {
        ESP_LOGW(TAG, "Invalid MQTT payload assembly length");
        BoardMqtt_ResetPayloadAssembly();
    }
}

static void BoardMqtt_HandleCommand(const char *topic, int topic_len, const char *data, int data_len)
{
    char *topic_text = BoardMqtt_CopySegment(topic, topic_len);
    char *payload_text = BoardMqtt_CopySegment(data, data_len);
    if (!topic_text || !payload_text) {
        free(topic_text);
        free(payload_text);
        return;
    }

    cJSON *root = cJSON_Parse(payload_text);
    if (!root) {
        BoardMqtt_LogReceivedText(topic_text, payload_text);
        free(topic_text);
        free(payload_text);
        return;
    }
    BoardMqtt_LogReceivedJson(topic_text, root);

#if !BOARD_MQTT_EXECUTE_COMMANDS
    cJSON_Delete(root);
    free(topic_text);
    free(payload_text);
    return;
#endif

    const char *req_id_text = NULL;
    cJSON *header = cJSON_GetObjectItem(root, "header");
    cJSON *req_id = cJSON_IsObject(header) ? cJSON_GetObjectItem(header, "req_id") : NULL;
    if (cJSON_IsString(req_id) && req_id->valuestring && req_id->valuestring[0] != '\0') {
        req_id_text = req_id->valuestring;
    } else {
        ESP_LOGW(TAG, "MQTT command has no header.req_id, skip ACK");
        cJSON_Delete(root);
        free(topic_text);
        free(payload_text);
        return;
    }

    board_mqtt_topic_type_t topic_type = BoardMqtt_GetTopicType(topic_text);
    cJSON *version = cJSON_IsObject(header) ? cJSON_GetObjectItem(header, "version") : NULL;
    cJSON *device_id = cJSON_IsObject(header) ? cJSON_GetObjectItem(header, "device_id") : NULL;
    cJSON *timestamp = cJSON_IsObject(header) ? cJSON_GetObjectItem(header, "timestamp") : NULL;
    cJSON *action = cJSON_GetObjectItem(root, "action");
    cJSON *payload = cJSON_GetObjectItem(root, "payload");

    board_mqtt_action_result_t result = BoardMqtt_ResultOk("ok");
    if (topic_type == BOARD_MQTT_TOPIC_UNKNOWN) {
        result = BoardMqtt_ResultFail(BOARD_MQTT_ERR_TOPIC_MISMATCH, "invalid command topic");
    } else if (!cJSON_IsString(version) || strcmp(version->valuestring, "1.0") != 0) {
        result = BoardMqtt_ResultFail(BOARD_MQTT_ERR_INVALID_PARAM, "invalid header.version");
    } else if (!cJSON_IsString(device_id) || strcmp(device_id->valuestring, s_device_id) != 0) {
        result = BoardMqtt_ResultFail(BOARD_MQTT_ERR_INVALID_PARAM, "invalid header.device_id");
    } else if (!cJSON_IsNumber(timestamp)) {
        result = BoardMqtt_ResultFail(BOARD_MQTT_ERR_INVALID_PARAM, "invalid header.timestamp");
    } else if (!cJSON_IsString(action) || action->valuestring[0] == '\0') {
        result = BoardMqtt_ResultFail(BOARD_MQTT_ERR_INVALID_PARAM, "invalid action");
    } else if (!cJSON_IsObject(payload)) {
        result = BoardMqtt_ResultFail(BOARD_MQTT_ERR_INVALID_PARAM, "invalid payload");
    } else if ((topic_type == BOARD_MQTT_TOPIC_CTL && !BoardMqtt_ActionIsCtl(action->valuestring)) ||
               (topic_type == BOARD_MQTT_TOPIC_TASK && !BoardMqtt_ActionIsTask(action->valuestring))) {
        result = BoardMqtt_ResultFail(BOARD_MQTT_ERR_TOPIC_MISMATCH, "action not allowed on this topic");
    } else {
        result = BoardMqtt_DispatchAction(req_id_text, action->valuestring, payload);
    }

    BoardMqtt_PublishAck(req_id_text, result.success, result.error_code, result.info);
    cJSON_Delete(root);
    free(topic_text);
    free(payload_text);
}

static void BoardMqtt_ReconnectTask(void *arg)
{
    (void)arg;
    while (s_mqtt_client && !s_mqtt_connected) {
        if (Board_Is_Voice_Session_Active()) {
            ESP_LOGI(TAG, "Delay local MQTT reconnect while voice session is active");
            vTaskDelay(pdMS_TO_TICKS(BOARD_MQTT_RECONNECT_DELAY_MS));
            continue;
        }
        ESP_LOGI(TAG, "Reconnecting local MQTT client");
        esp_err_t err = esp_mqtt_client_reconnect(s_mqtt_client);
        if (err != ESP_OK) {
            ESP_LOGW(TAG, "Local MQTT reconnect failed: %s", esp_err_to_name(err));
            vTaskDelay(pdMS_TO_TICKS(BOARD_MQTT_RECONNECT_DELAY_MS));
            continue;
        }
        break;
    }
    s_reconnect_task_handle = NULL;
    vTaskDelete(NULL);
}

static void BoardMqtt_ScheduleReconnect(void)
{
    if (!s_mqtt_client || s_reconnect_task_handle) {
        return;
    }
    BaseType_t ret = xTaskCreate(BoardMqtt_ReconnectTask, "mqtt_reconnect", 4096, NULL, BOARD_MQTT_RECONNECT_TASK_PRIORITY, &s_reconnect_task_handle);
    if (ret != pdPASS) {
        ESP_LOGE(TAG, "Failed to create local MQTT reconnect task");
        s_reconnect_task_handle = NULL;
    }
}

static void BoardMqtt_StateTask(void *arg)
{
    (void)arg;
    while (s_mqtt_client) {
        vTaskDelay(pdMS_TO_TICKS(BOARD_MQTT_STATE_REPORT_INTERVAL_MS));
        BoardMqtt_PublishState();
    }
    s_state_task_handle = NULL;
    vTaskDelete(NULL);
}

static void BoardMqtt_StartStateTask(void)
{
    if (s_state_task_handle) {
        return;
    }
    BaseType_t ret = xTaskCreate(BoardMqtt_StateTask, "mqtt_state", 4096, NULL, 2, &s_state_task_handle);
    if (ret != pdPASS) {
        ESP_LOGE(TAG, "Failed to create MQTT state report task");
        s_state_task_handle = NULL;
    }
}

static void BoardMqtt_Subscribe(const char *topic)
{
    int msg_id = esp_mqtt_client_subscribe_single(s_mqtt_client, topic, BOARD_MQTT_QOS);
    if (msg_id >= 0) {
        ESP_LOGI(TAG, "Subscribing to %s, qos=%d, msg_id=%d", topic, BOARD_MQTT_QOS, msg_id);
    } else {
        ESP_LOGE(TAG, "Failed to subscribe to %s", topic);
    }
}

static void BoardMqtt_EventHandler(void *handler_args, esp_event_base_t base, int32_t event_id, void *event_data)
{
    (void)handler_args;
    (void)base;
    esp_mqtt_event_t *event = (esp_mqtt_event_t *)event_data;

    switch (event_id) {
    case MQTT_EVENT_CONNECTED:
        s_mqtt_connected = true;
        ESP_LOGI(TAG, "MQTT business server connected, host=%s, port=%d, username=%s", BOARD_MQTT_BROKER_HOST, BOARD_MQTT_BROKER_PORT, BOARD_MQTT_USERNAME);
        ESP_LOGI(TAG, "MQTT command topics ready, ctl=%s, task=%s", s_cmd_ctl_topic, s_cmd_task_topic);
        BoardMqtt_Subscribe(s_cmd_ctl_topic);
        BoardMqtt_Subscribe(s_cmd_task_topic);
        BoardMqtt_PublishLog("MQTT connected");
        BoardMqtt_PublishState();
        break;
    case MQTT_EVENT_DISCONNECTED:
        s_mqtt_connected = false;
        ESP_LOGW(TAG, "Disconnected from broker%s", BoardMqtt_GetTaskType() != BOARD_MQTT_TASK_NONE ? ", current task will continue locally" : "");
        BoardMqtt_ScheduleReconnect();
        break;
    case MQTT_EVENT_SUBSCRIBED:
        ESP_LOGI(TAG, "Subscribed, msg_id=%d", event->msg_id);
        break;
    case MQTT_EVENT_DATA:
        BoardMqtt_HandleDataEvent(event);
        break;
    case MQTT_EVENT_ERROR:
        s_mqtt_connected = false;
        if (event->error_handle) {
            ESP_LOGE(TAG, "MQTT error, esp_tls=%s, transport_sock_errno=%d",
                     esp_err_to_name(event->error_handle->esp_tls_last_esp_err),
                     event->error_handle->esp_transport_sock_errno);
        } else {
            ESP_LOGE(TAG, "MQTT error");
        }
        BoardMqtt_ScheduleReconnect();
        break;
    default:
        break;
    }
}

esp_err_t BoardMqtt_Start(const char *client_id)
{
    if (s_mqtt_client) {
        ESP_LOGI(TAG, "MQTT client already started");
        return ESP_OK;
    }

    if (!client_id || strlen(client_id) == 0) {
        ESP_LOGE(TAG, "Invalid MQTT client id");
        return ESP_ERR_INVALID_ARG;
    }

    int id_len = snprintf(s_device_id, sizeof(s_device_id), "%s", client_id);
    if (id_len < 0 || id_len >= (int)sizeof(s_device_id)) {
        ESP_LOGE(TAG, "MQTT client id is too long");
        return ESP_ERR_INVALID_SIZE;
    }

    esp_err_t ret = BoardMqtt_FormatTopic(s_cmd_ctl_topic, sizeof(s_cmd_ctl_topic), BOARD_MQTT_CMD_CTL_TOPIC_FORMAT, client_id);
    if (ret != ESP_OK) {
        return ret;
    }
    ret = BoardMqtt_FormatTopic(s_cmd_task_topic, sizeof(s_cmd_task_topic), BOARD_MQTT_CMD_TASK_TOPIC_FORMAT, client_id);
    if (ret != ESP_OK) {
        return ret;
    }
    ret = BoardMqtt_FormatTopic(s_rpt_log_topic, sizeof(s_rpt_log_topic), BOARD_MQTT_RPT_LOG_TOPIC_FORMAT, client_id);
    if (ret != ESP_OK) {
        return ret;
    }
    ret = BoardMqtt_FormatTopic(s_rpt_ack_topic, sizeof(s_rpt_ack_topic), BOARD_MQTT_RPT_ACK_TOPIC_FORMAT, client_id);
    if (ret != ESP_OK) {
        return ret;
    }
    ret = BoardMqtt_FormatTopic(s_rpt_progress_topic, sizeof(s_rpt_progress_topic), BOARD_MQTT_RPT_PROGRESS_TOPIC_FORMAT, client_id);
    if (ret != ESP_OK) {
        return ret;
    }
    ret = BoardMqtt_FormatTopic(s_rpt_state_topic, sizeof(s_rpt_state_topic), BOARD_MQTT_RPT_STATE_TOPIC_FORMAT, client_id);
    if (ret != ESP_OK) {
        return ret;
    }
    ret = BoardMqtt_FormatTopic(s_rpt_event_topic, sizeof(s_rpt_event_topic), BOARD_MQTT_RPT_EVENT_TOPIC_FORMAT, client_id);
    if (ret != ESP_OK) {
        return ret;
    }

    esp_mqtt_client_config_t mqtt_config = {0};
    mqtt_config.task.priority = BOARD_MQTT_TASK_PRIORITY;
    mqtt_config.task.stack_size = 6144;
    mqtt_config.buffer.size = 8192;
    mqtt_config.buffer.out_size = 8192;
    mqtt_config.broker.address.hostname = BOARD_MQTT_BROKER_HOST;
    mqtt_config.broker.address.port = BOARD_MQTT_BROKER_PORT;
    mqtt_config.broker.address.transport = MQTT_TRANSPORT_OVER_TCP;
    mqtt_config.credentials.client_id = client_id;
    if (strlen(BOARD_MQTT_USERNAME) > 0) {
        mqtt_config.credentials.username = BOARD_MQTT_USERNAME;
        mqtt_config.credentials.authentication.password = BOARD_MQTT_PASSWORD;
    }
    mqtt_config.session.keepalive = BOARD_MQTT_KEEPALIVE_SEC;
    mqtt_config.network.disable_auto_reconnect = true;
    mqtt_config.network.timeout_ms = BOARD_MQTT_NETWORK_TIMEOUT_MS;

    s_mqtt_client = esp_mqtt_client_init(&mqtt_config);
    if (!s_mqtt_client) {
        ESP_LOGE(TAG, "Failed to initialize MQTT client");
        return ESP_FAIL;
    }

    ret = esp_mqtt_client_register_event(s_mqtt_client, MQTT_EVENT_ANY, BoardMqtt_EventHandler, NULL);
    if (ret != ESP_OK) {
        ESP_LOGE(TAG, "Failed to register MQTT event handler: %s", esp_err_to_name(ret));
        esp_mqtt_client_destroy(s_mqtt_client);
        s_mqtt_client = NULL;
        return ret;
    }

    ret = esp_mqtt_client_start(s_mqtt_client);
    if (ret != ESP_OK) {
        ESP_LOGE(TAG, "Failed to start MQTT client: %s", esp_err_to_name(ret));
        esp_mqtt_client_destroy(s_mqtt_client);
        s_mqtt_client = NULL;
        return ret;
    }

    BoardMqtt_StartStateTask();
    ESP_LOGI(TAG, "Starting MQTT client, host=%s, port=%d, ctl_topic=%s, task_topic=%s, state_topic=%s",
             BOARD_MQTT_BROKER_HOST, BOARD_MQTT_BROKER_PORT, s_cmd_ctl_topic, s_cmd_task_topic, s_rpt_state_topic);
    return ESP_OK;
}

void BoardMqtt_Stop(void)
{
    if (!s_mqtt_client) {
        s_mqtt_connected = false;
        return;
    }

    ESP_LOGI(TAG, "Stopping MQTT client");
    esp_mqtt_client_stop(s_mqtt_client);
    if (s_state_task_handle) {
        TaskHandle_t state_task = s_state_task_handle;
        s_state_task_handle = NULL;
        vTaskDelete(state_task);
    }
    esp_mqtt_client_destroy(s_mqtt_client);
    s_mqtt_client = NULL;
    s_mqtt_connected = false;
}

bool BoardMqtt_IsConnected(void)
{
    return s_mqtt_connected;
}