/**
 * @file    web_server.cpp
 * @brief   HTTP/WebSocket 服务器核心 — Mongoose 集成 + 连接管理
 * @details 在 app_web_server 线程的 EV_TIMER1 中调用 mg_mgr_poll，
 *          单线程复用 X-Macro 框架，消除原工程的双线程设计。
 *          EV_TIMER2: ws_push_task() 增量推送
 *          EV_TIMER3: run_cnt++
 *          EV_STOP: 优雅退出
 */
#include "web_server.h"
#include "myLog.h"
#include "myMongoose.h"
#include "cJSON.h"
#include "myFunc.h"

#include <string>
#include <vector>
#include <map>
#include <pthread.h>
#include <csignal>

/* ---- 硬编码配置 ---- */
#define WS_LISTEN_ON   "http://0.0.0.0:8080"
#define WS_UPGRADE_URI "/ws"

/* ---- 全局状态 ---- */
static struct mg_mgr           s_ws_mgr;
static std::vector<mg_connection *> g_ws_conns;
static pthread_mutex_t         g_ws_conns_mutex = PTHREAD_MUTEX_INITIALIZER;
static std::string             g_web_root;        /**< 静态文件根目录 */

/* ---- 前向声明 ---- */
void ws_recv(struct mg_connection *c, const char *p_rx, uint16_t rx_len);
void ws_session_destroy(struct mg_connection *c);
void ws_push_task(void);

/* ---- 广播发送 ---- */
void ws_send_all(const char *p_tx, size_t tx_len)
{
    if (!p_tx || tx_len == 0)
    {
        return;
    }

    pthread_mutex_lock(&g_ws_conns_mutex);

    for (size_t i = 0; i < g_ws_conns.size(); i++)
    {
        mg_connection *c = g_ws_conns[i];

        if (c->is_websocket && !c->is_draining)
        {
            mg_ws_send(c, p_tx, tx_len, WEBSOCKET_OP_TEXT);
        }
    }

    pthread_mutex_unlock(&g_ws_conns_mutex);
}

/* ---- 单连接发送 ---- */
bool ws_send_one(unsigned long conn_id, const char *p_tx, size_t tx_len)
{
    if (!p_tx || tx_len == 0)
    {
        return false;
    }

    bool ok = false;

    pthread_mutex_lock(&g_ws_conns_mutex);

    for (size_t i = 0; i < g_ws_conns.size(); i++)
    {
        mg_connection *c = g_ws_conns[i];

        if (c->id == conn_id && c->is_websocket && !c->is_draining)
        {
            size_t sent = mg_ws_send(c, p_tx, tx_len, WEBSOCKET_OP_TEXT);
            ok = (sent >= tx_len);
            break;
        }
    }

    pthread_mutex_unlock(&g_ws_conns_mutex);

    return ok;
}

/* ---- 二进制发送 ---- */
bool ws_send_binary(unsigned long conn_id, const void *data, size_t len)
{
    if (!data || len == 0)
    {
        return false;
    }

    bool ok = false;

    pthread_mutex_lock(&g_ws_conns_mutex);

    for (size_t i = 0; i < g_ws_conns.size(); i++)
    {
        mg_connection *c = g_ws_conns[i];

        if (c->id == conn_id && c->is_websocket && !c->is_draining)
        {
            size_t sent = mg_ws_send(c, data, len, WEBSOCKET_OP_BINARY);
            ok = (sent >= len);
            break;
        }
    }

    pthread_mutex_unlock(&g_ws_conns_mutex);

    return ok;
}

/* ---- 连接列表管理 ---- */
static void ws_conn_remove(mg_connection *c)
{
    pthread_mutex_lock(&g_ws_conns_mutex);

    for (size_t i = 0; i < g_ws_conns.size(); i++)
    {
        if (g_ws_conns[i] == c)
        {
            g_ws_conns.erase(g_ws_conns.begin() + i);
            break;
        }
    }

    pthread_mutex_unlock(&g_ws_conns_mutex);
}

/* ---- Mongoose 事件处理器 ---- */
static void ws_event_handler(mg_connection *c, int ev, void *ev_data)
{
    if (ev == MG_EV_HTTP_MSG)
    {
        mg_http_message *hm = (mg_http_message *)ev_data;

        if (mg_match(hm->uri, mg_str(WS_UPGRADE_URI), NULL))
        {
            mg_ws_upgrade(c, hm, NULL);
            c->data[0] = 'W';
        }
        else
        {
            /* HTTP 静态文件服务 — 从 web/ 目录提供 */
            struct mg_http_serve_opts opts = {};
            opts.root_dir = g_web_root.c_str();
            mg_http_serve_dir(c, hm, &opts);
        }
    }
    else if (ev == MG_EV_WS_MSG)
    {
        mg_ws_message *wm = (mg_ws_message *)ev_data;

        bool found = false;
        pthread_mutex_lock(&g_ws_conns_mutex);
        for (size_t i = 0; i < g_ws_conns.size(); i++)
        {
            if (g_ws_conns[i] == c)
            {
                found = true;
                break;
            }
        }
        if (!found)
        {
            g_ws_conns.push_back(c);
        }
        pthread_mutex_unlock(&g_ws_conns_mutex);

        ws_recv(c, wm->data.buf, (uint16_t)wm->data.len);
    }
    else if (ev == MG_EV_WS_CTL)
    {
        mg_ws_message *wm = (mg_ws_message *)ev_data;
        uint8_t op = wm->flags & 15;

        if (op == WEBSOCKET_OP_CLOSE)
        {
            ws_session_destroy(c);
            ws_conn_remove(c);
        }
    }
    else if (ev == MG_EV_CLOSE)
    {
        if (c->is_websocket)
        {
            ws_session_destroy(c);
            ws_conn_remove(c);
        }
    }
}

/* ---- 命令行执行（捕获 stdout → 字符串） ---- */
#include "myCmd.h"
#include <unistd.h>
#include <cstring>

std::string ws_cmd_exec(const char *cmd_line)
{
    if (!cmd_line)
    {
        return "";
    }

    char line[512];
    strncpy(line, cmd_line, sizeof(line) - 1);
    line[sizeof(line) - 1] = '\0';

    /* strtok 解析参数 */
    char *argv[16];
    int argc = 0;
    char *p = strtok(line, " \t");
    while (p && argc < 16)
    {
        argv[argc++] = p;
        p = strtok(NULL, " \t");
    }

    if (argc == 0)
    {
        return "";
    }

    stru_cmd *p_cmd = cmd_find(argv[0]);
    if (!p_cmd)
    {
        return std::string("[unknown] ") + argv[0];
    }

    /* pipe 重定向 stdout */
    fflush(stdout);
    int pipefd[2];

    if (pipe(pipefd) != 0)
    {
        return "pipe failed";
    }

    std::string result;
    auto *ctx = new std::pair<int, std::string *>(pipefd[0], &result);

    pthread_t reader;
    pthread_create(&reader, NULL,
        [](void *arg) -> void *
        {
            auto *cx = (std::pair<int, std::string *> *)arg;
            char buf[4096];
            ssize_t n;
            while ((n = read(cx->first, buf, sizeof(buf))) > 0)
            {
                cx->second->append(buf, n);
            }
            close(cx->first);
            delete cx;
            return NULL;
        },
        ctx);

    int saved = dup(STDOUT_FILENO);
    dup2(pipefd[1], STDOUT_FILENO);
    close(pipefd[1]);

    p_cmd->func(argc, argv);
    fflush(stdout);

    dup2(saved, STDOUT_FILENO);
    close(saved);

    pthread_join(reader, NULL);

    if (!result.empty())
    {
        return result;
    }

    return "(no output)";
}

/* ---- 异步命令通道实现 ---- */
static std::map<std::string, unsigned long> g_async_cmd_conn_map;
static pthread_mutex_t g_async_cmd_conn_mutex = PTHREAD_MUTEX_INITIALIZER;

void ws_set_async_cmd_conn(const char *module, unsigned long conn_id)
{
    if (!module)
    {
        return;
    }

    pthread_mutex_lock(&g_async_cmd_conn_mutex);
    g_async_cmd_conn_map[module] = conn_id;
    pthread_mutex_unlock(&g_async_cmd_conn_mutex);
}

unsigned long ws_get_async_cmd_conn(const char *module)
{
    if (!module)
    {
        return 0;
    }

    pthread_mutex_lock(&g_async_cmd_conn_mutex);
    auto it = g_async_cmd_conn_map.find(module);
    unsigned long id = (it != g_async_cmd_conn_map.end()) ? it->second : 0;
    pthread_mutex_unlock(&g_async_cmd_conn_mutex);

    return id;
}

/* ---- 异步命令输出（给其他模块调用） ---- */
void async_cmd_printf(const char *module, const char *fmt, ...)
{
    if (!module || !fmt)
    {
        return;
    }

    unsigned long conn_id = ws_get_async_cmd_conn(module);

    if (conn_id == 0)
    {
        return;
    }

    va_list ap;
    char buf[2048];

    va_start(ap, fmt);
    int n = vsnprintf(buf, sizeof(buf), fmt, ap);
    va_end(ap);

    if (n > 0)
    {
        ws_send_binary(conn_id, buf, (size_t)n);
    }
}

void async_cmd_complete(const char *module, bool success, const char *msg)
{
    if (!module || !msg)
    {
        return;
    }

    unsigned long conn_id = ws_get_async_cmd_conn(module);

    if (conn_id == 0)
    {
        return;
    }

    char buf[1024];
    int n = snprintf(buf, sizeof(buf), "[%s] %s %s\n",
                     module, success ? "OK" : "FAIL", msg);

    ws_send_binary(conn_id, buf, (size_t)n);
}

/* ================================================================
 * X-Macro 框架入口
 * ================================================================ */

int app_web_server_init1(void *arg)
{
    (void)arg;

    /* 静态文件根目录: <exe_dir>/web/ */
    char proc_dir[512] = {0};
    if (0 == func_proc_self_dir(proc_dir, sizeof(proc_dir)))
    {
        g_web_root = std::string(proc_dir) + "/config/web";
    }
    else
    {
        g_web_root = "web";  /* fallback: 相对路径 */
    }

    mg_log_set(MG_LL_ERROR);
    mg_mgr_init(&s_ws_mgr);
    mg_http_listen(&s_ws_mgr, WS_LISTEN_ON, ws_event_handler, NULL);

    LOG_I("web_server listening on %s, ws=%s, root=%s",
             WS_LISTEN_ON, WS_UPGRADE_URI, g_web_root.c_str());

    return 0;
}

int app_web_server_init2(void *arg)
{
    (void)arg;
    return 0;
}

void *app_web_server(void *arg)
{
    if (!arg)
    {
        LOG_E("app_web_server arg null");
        return NULL;
    }

    stru_app *p_app = (stru_app *)arg;

    uint32_t event;
    uint32_t cnt_1s = 0;

    while (1)
    {
        task_event_recv(p_app->p_event,
            EV_TIMER1 | EV_TIMER2 | EV_TIMER3,
            TASK_EVENT_FLAG_OR | TASK_EVENT_FLAG_CLEAR,
            TASK_EVENT_WAIT_FOREVER,
            &event);

        if (event & EV_STOP)
        {
            LOG_I("web_server stopping");

            break;
        }

        if (event & EV_TIMER1)
        {
            /* IO 轮询（非阻塞） */
            mg_mgr_poll(&s_ws_mgr, 0);
        }

        if (event & EV_TIMER2)
        {
            /* 增量推送 */
            ws_push_task();
        }

        if (event & EV_TIMER3)
        {
            cnt_1s++;
            p_app->run_cnt++;
        }
    }

    mg_mgr_free(&s_ws_mgr);

    LOG_I("web_server stopped, total 1s cycles=%u", cnt_1s);

    return NULL;
}
