#include <stdlib.h>
#include <string.h>
#include <stdio.h>
#include <stdint.h>
#include <stddef.h>
#include <assert.h>
#include "stellar/session.h"
#include "stellar/monitor.h"
#include "session_manager/session_manager_rte.h"
#include "monitor/monitor_utils.h"
#include "monitor/monitor_rpc.h"
#include "sds/sds.h"
#include "session_manager/session_internal.h"

// temp add
extern struct session_manager_rte *session_manager_get_runtime(struct session_manager *sess_mgr, uint16_t thread_id);

#define SHOW_SESSION_BRIEF_LIMIT_DEFAULT 10
#define SHOW_SESSION_BRIEF_LIMIT_MAX 1000
#define SHOW_SESSION_BRIEF_SCAN_COUNT_DEFAULT 100

/* show session brief  */
struct stm_show_session_brief_opt
{
#define SESSION_SCAN_SNET (1 << 25)
#define SESSION_SCAN_DNET (1 << 26)
#define SESSION_SCAN_ID (1 << 27)
#define SESSION_SCAN_CURSOR (1 << 28)
#define SESSION_SCAN_COUNT (1 << 29)
#define SESSION_SCAN_LIMIT (1 << 30)
    struct session_scan_opts scan_opt;
    uint32_t limit;
};

/* show session detail <id> */
struct stm_show_session_detail_opt
{
    uint64_t sess_id;
    int thread_idx; // todo, not used now
};

struct show_session_brief_result
{
    uint64_t sid;
    int thread_index;
    enum session_state state;
    enum session_type protocol;
    time_t create_time_in_sec;
    time_t last_pkt_time_in_sec;
    uint8_t flow_dir;
    struct tuple6 addr;
} __attribute__((packed));

struct show_session_brief_array
{
    size_t array_num;
    struct show_session_brief_result array[0]; /* Continuous memory, array_num * sizeof(struct show_session_brief_result) */
} __attribute__((packed));

struct show_session_detail_result
{
    struct show_session_brief_result sess_brief;
    unsigned long long sess_stat[MAX_FLOW_TYPE][MAX_STAT];
    enum session_direction direction; // in or out
    unsigned char application;        // L7 appid
    // todo, other info
} __attribute__((packed));

static void stm_session_brief_cli_args_set_default(struct stm_show_session_brief_opt *show_opt)
{
    memset(show_opt, 0, sizeof(struct stm_show_session_brief_opt));
    show_opt->limit = SHOW_SESSION_BRIEF_LIMIT_DEFAULT;
    show_opt->scan_opt.cursor = 0;
    show_opt->scan_opt.count = SHOW_SESSION_BRIEF_SCAN_COUNT_DEFAULT;
}

static uint32_t stm_session_brief_cli_args_get_flags(const char *para_name)
{
    if (NULL == para_name)
    {
        return 0;
    }

    if (strncasecmp(para_name, "id", 2) == 0)
    {
        return SESSION_SCAN_ID;
    }
    else if (strncasecmp(para_name, "cursor", 6) == 0)
    {
        return SESSION_SCAN_CURSOR;
    }
    else if (strncasecmp(para_name, "count", 5) == 0)
    {
        return SESSION_SCAN_COUNT;
    }
    else if (strncasecmp(para_name, "state", 5) == 0)
    {
        return SESSION_SCAN_STATE;
    }
    else if (strncasecmp(para_name, "sip", 3) == 0)
    {
        return SESSION_SCAN_SIP;
    }
    else if (strncasecmp(para_name, "dip", 3) == 0)
    {
        return SESSION_SCAN_DIP;
    }
    else if (strncasecmp(para_name, "snet", 4) == 0)
    {
        return SESSION_SCAN_SNET;
    }
    else if (strncasecmp(para_name, "dnet", 4) == 0)
    {
        return SESSION_SCAN_DNET;
    }
    else if (strncasecmp(para_name, "sport", 5) == 0)
    {
        return SESSION_SCAN_SPORT;
    }
    else if (strncasecmp(para_name, "dport", 5) == 0)
    {
        return SESSION_SCAN_DPORT;
    }
    else if (strncasecmp(para_name, "protocol", 8) == 0)
    {
        return SESSION_SCAN_TYPE;
    }
    else if (strncasecmp(para_name, "limit", 5) == 0)
    {
        return SESSION_SCAN_LIMIT;
    }
    else if (strncasecmp(para_name, "create-time-in", strlen("create-time-in")) == 0)
    {
        return SESSION_SCAN_CREATE_TIME;
    }
    else if (strncasecmp(para_name, "last-pkt-time-in", strlen("last-pkt-time-in")) == 0)
    {
        return SESSION_SCAN_LASPKT_TIME;
    }
    return 0;
}

static enum session_state show_session_state_pton(const char *state_str)
{
    if (strncasecmp(state_str, "opening", strlen("opening")) == 0)
    {
        return SESSION_STATE_OPENING;
    }
    else if (strncasecmp(state_str, "active", strlen("active")) == 0)
    {
        return SESSION_STATE_ACTIVE;
    }
    else if (strncasecmp(state_str, "closing", strlen("closing")) == 0)
    {
        return SESSION_STATE_CLOSING;
    }

    return MAX_STATE;
}

static uint32_t show_session_ipaddr_ntop(const char *ip_string, struct session_scan_opts *scan_opt, uint32_t flag)
{
    uint32_t addr_family;
    if (SESSION_SCAN_SIP == flag)
    {
        addr_family = stm_inet_pton(ip_string, &scan_opt->src_addr[0].v4, &scan_opt->src_addr[0].v6);
    }
    else
    {
        addr_family = stm_inet_pton(ip_string, &scan_opt->dst_addr[0].v4, &scan_opt->dst_addr[0].v6);
    }

    if (addr_family == 0)
    {
        return 0;
    }
    if (AF_INET == addr_family)
    {
        if (SESSION_SCAN_SIP == flag)
        {
            scan_opt->src_addr[1].v4 = scan_opt->src_addr[0].v4;
        }
        else
        {
            scan_opt->dst_addr[1].v4 = scan_opt->dst_addr[0].v4;
        }
    }
    else
    {
        if (SESSION_SCAN_SIP == flag)
        {
            scan_opt->src_addr[1].v6 = scan_opt->src_addr[0].v6;
        }
        else
        {
            scan_opt->dst_addr[1].v6 = scan_opt->dst_addr[0].v6;
        }
    }
    return addr_family;
}

static sds show_session_cli_args_sanity_check(const struct stm_show_session_brief_opt *brief_opt)
{
    uint32_t flags = brief_opt->scan_opt.flags;
    sds ss = sdsempty();

    if ((flags & SESSION_SCAN_SIP) && (flags & SESSION_SCAN_SNET))
    {
        ss = sdscatprintf(ss, "error: the 'sip' and 'snet' options conflict!");
        return ss;
    }
    if ((flags & SESSION_SCAN_DIP) && (flags & SESSION_SCAN_DNET))
    {
        ss = sdscatprintf(ss, "error: the 'dip' and 'dnet' options conflict!");
        return ss;
    }
    return ss;
}

static int show_session_is_help(int argc, char *argv[])
{
    for (int i = 0; i < argc; i++)
    {
        if (strncasecmp(argv[i], "help", 4) == 0)
        {
            return 1;
        }
        if (strncasecmp(argv[i], "--help", 6) == 0)
        {
            return 1;
        }
        if (strncasecmp(argv[i], "-h", 2) == 0)
        {
            return 1;
        }
    }
    return 0;
}

static struct monitor_reply *show_session_brief_usage(void)
{
    sds ss = sdsempty();
    ss = sdscatprintf(ss, "Usage: show session brief [options]\n");
    ss = sdscatprintf(ss, "Options:\n");
    ss = sdscatprintf(ss, "    -h, --help help\tdisplay usage information and exit\n");
    ss = sdscatprintf(ss, "    cursor <cursor>\n");
    ss = sdscatprintf(ss, "    count <count>\n");
    ss = sdscatprintf(ss, "    state <opening|active|closing>\n");
    ss = sdscatprintf(ss, "    protocol <tcp|udp>\n");
    ss = sdscatprintf(ss, "    sip <source ip>\n");
    ss = sdscatprintf(ss, "    dip <destination ip>\n");
    ss = sdscatprintf(ss, "    sport <source port>\n");
    ss = sdscatprintf(ss, "    dport <destination port>\n");
    ss = sdscatprintf(ss, "    snet <source network/netmask>\texample 192.168.1.0/24\n");
    ss = sdscatprintf(ss, "    dnet <destination network/netmask>\texample 1234::abcd/48\n");
    ss = sdscatprintf(ss, "    create-time-in <last-N-[seconds|minutes|hours|days]>\texample last-1-hours\n");
    ss = sdscatprintf(ss, "    last-pkt-time-in <last-N-[seconds|minutes|hours|days]>\texample last-7-days\n");
    ss = sdscatprintf(ss, "    limit <limit>\n");

    struct monitor_reply *reply = monitor_reply_new_string("%s", ss);
    sdsfree(ss);
    return reply;
}

static struct monitor_reply *show_session_detail_usage(void)
{
    sds ss = sdsempty();
    ss = sdscatprintf(ss, "Usage: show session detial [options]\n");
    ss = sdscatprintf(ss, "Options:\n");
    ss = sdscatprintf(ss, "    -h, --help help\tdisplay usage information and exit\n");
    ss = sdscatprintf(ss, "    id <session id>\n");
    struct monitor_reply *reply = monitor_reply_new_string("%s", ss);
    sdsfree(ss);
    return reply;
}

/*
   "show session ..." command args parser
   return:
         NULL: success, brief_opt is filled
     not NULL: error message
 */
static struct monitor_reply *show_session_brief_cli_args_parse(int argc, char *argv[], struct stm_show_session_brief_opt *brief_opt)
{
    uint32_t ipaddr_family = 0, history_ipaddr_family = 0;
    sds ss = NULL;
    uint32_t flag;
    struct monitor_reply *error_reply = NULL;
    const char *opt_name, *opt_value;

    stm_session_brief_cli_args_set_default(brief_opt);
    int i = 3; // skip "show session brief"
    while (i < argc)
    {
        ipaddr_family = 0;
        opt_name = argv[i];
        flag = stm_session_brief_cli_args_get_flags(opt_name);
        if (i + 1 >= argc)
        {
            error_reply = monitor_reply_new_error("option %s requires an argument\n", opt_name);
            return error_reply;
        }
        opt_value = argv[++i];

        switch (flag)
        {
        case SESSION_SCAN_TYPE:
        {
            if (strncasecmp(opt_value, "tcp", 3) == 0)
            {
                brief_opt->scan_opt.type = SESSION_TYPE_TCP;
            }
            else if (strncasecmp(opt_value, "udp", 3) == 0)
            {
                brief_opt->scan_opt.type = SESSION_TYPE_UDP;
            }
            else
            {
                error_reply = monitor_reply_new_error("unsupported protocol type: %s. should be <tcp|udp>\n", opt_value);
                goto error_exit;
            }
        }
        break;
        case SESSION_SCAN_STATE:
        {
            enum session_state tmp_state = show_session_state_pton(opt_value);
            if (tmp_state == MAX_STATE)
            {
                error_reply = monitor_reply_new_error("unrecognized session state: %s. should be <opening|active|closing>\n", opt_value);
                goto error_exit;
            }
            brief_opt->scan_opt.state = tmp_state;
        }
        break;
        // case SESSION_SCAN_ID:
        //     if (stm_string_isdigit(opt_value) == 0)
        //     {
        //         *error_reply = monitor_reply_new_error("invalid session id: %s. should be integer in range: <1 - UINT64_MAX>\n", opt_value);
        //         goto error_exit;
        //     }
        //     show_opt->detail_opt.sess_id = strtoull(opt_value, NULL, 10);
        //     break;
        case SESSION_SCAN_SIP:
            ipaddr_family = show_session_ipaddr_ntop(opt_value, &brief_opt->scan_opt, SESSION_SCAN_SIP);
            if (ipaddr_family == 0)
            {
                error_reply = monitor_reply_new_error("invalid sip address: %s\n", opt_value);
                goto error_exit;
            }
            brief_opt->scan_opt.addr_family = ipaddr_family;
            break;
        case SESSION_SCAN_DIP:
            ipaddr_family = show_session_ipaddr_ntop(opt_value, &brief_opt->scan_opt, SESSION_SCAN_DIP);
            if (ipaddr_family == 0)
            {
                error_reply = monitor_reply_new_error("invalid dip address: %s\n", opt_value);
                goto error_exit;
            }
            brief_opt->scan_opt.addr_family = ipaddr_family;
            break;

        case SESSION_SCAN_SNET:
        {
            uint32_t ipv4addr, ipv4mask;
            struct in6_addr ipv6addr, ipv6mask;
            ipaddr_family = stm_ip_cidr_pton(opt_value, &ipv4addr, &ipv4mask, &ipv6addr, &ipv6mask);
            if (ipaddr_family == 0)
            {
                error_reply = monitor_reply_new_error("invalid snet CIDR address: %s\n", opt_value);
                goto error_exit;
            }
            if (AF_INET == ipaddr_family)
            {
                uint32_t ipv4_range[2];
                stm_ipv4_cidr_to_range(ipv4addr, ipv4mask, ipv4_range);
                brief_opt->scan_opt.src_addr[0].v4.s_addr = ipv4_range[0];
                brief_opt->scan_opt.src_addr[1].v4.s_addr = ipv4_range[1];
            }
            else
            {
                struct in6_addr ipv6_range[2];
                stm_ipv6_cidr_to_range(&ipv6addr, &ipv6mask, ipv6_range);
                brief_opt->scan_opt.src_addr[0].v6 = ipv6_range[0];
                brief_opt->scan_opt.src_addr[1].v6 = ipv6_range[1];
            }
            brief_opt->scan_opt.addr_family = ipaddr_family;
            flag = SESSION_SCAN_SIP;
        }
        break;
        case SESSION_SCAN_DNET:
        {
            uint32_t ipv4addr, ipv4mask;
            struct in6_addr ipv6addr, ipv6mask;
            ipaddr_family = stm_ip_cidr_pton(opt_value, &ipv4addr, &ipv4mask, &ipv6addr, &ipv6mask);
            if (ipaddr_family == 0)
            {
                error_reply = monitor_reply_new_error("invalid dnet CIDR address: %s\n", opt_value);
                goto error_exit;
            }
            if (AF_INET == ipaddr_family)
            {
                uint32_t ipv4_range[2];
                stm_ipv4_cidr_to_range(ipv4addr, ipv4mask, ipv4_range);
                brief_opt->scan_opt.dst_addr[0].v4.s_addr = ipv4_range[0];
                brief_opt->scan_opt.dst_addr[1].v4.s_addr = ipv4_range[1];
            }
            else
            {
                struct in6_addr ipv6_range[2];
                stm_ipv6_cidr_to_range(&ipv6addr, &ipv6mask, ipv6_range);
                brief_opt->scan_opt.dst_addr[0].v6 = ipv6_range[0];
                brief_opt->scan_opt.dst_addr[1].v6 = ipv6_range[1];
            }
            brief_opt->scan_opt.addr_family = ipaddr_family;
            flag = SESSION_SCAN_DIP;
        }
        break;
        case SESSION_SCAN_SPORT:
        {
            if (stm_string_isdigit(opt_value) == 0)
            {
                error_reply = monitor_reply_new_error("illegal sport: %s. should be integer\n", opt_value);
                goto error_exit;
            }
            int tmp_val = atoi(opt_value);
            if (tmp_val <= 0 || tmp_val > 65535)
            {
                error_reply = monitor_reply_new_error("illegal sport: %s. should be <1-65535>\n", opt_value);
                goto error_exit;
            }
            brief_opt->scan_opt.src_port = htons((unsigned short)tmp_val);
        }
        break;
        case SESSION_SCAN_DPORT:
        {
            if (stm_string_isdigit(opt_value) == 0)
            {
                error_reply = monitor_reply_new_error("illegal sport: %s. should be integer\n", opt_value);
                goto error_exit;
            }
            int tmp_val = atoi(opt_value);
            if (tmp_val <= 0 || tmp_val > 65535)
            {
                error_reply = monitor_reply_new_error("illegal sport: %s. should be <1-65535>\n", opt_value);
                goto error_exit;
            }
            brief_opt->scan_opt.dst_port = htons((unsigned short)tmp_val);
        }
        break;

        case SESSION_SCAN_CURSOR:
            brief_opt->scan_opt.cursor = strtoul(opt_value, NULL, 10);
            break;
        case SESSION_SCAN_COUNT:
            brief_opt->scan_opt.count = strtoul(opt_value, NULL, 10);
            if (brief_opt->scan_opt.count == 0)
            {
                error_reply = monitor_reply_new_error("illegal count: %s. should be <1 - UINT32_MAX>\n", opt_value);
                goto error_exit;
            }
            break;
        case SESSION_SCAN_LIMIT:
            brief_opt->limit = strtoul(opt_value, NULL, 10);
            if (brief_opt->limit == 0 || brief_opt->limit > SHOW_SESSION_BRIEF_LIMIT_MAX)
            {
                error_reply = monitor_reply_new_error("illegal limit: %s. should be <1 - %u>\n", opt_value, SHOW_SESSION_BRIEF_LIMIT_MAX);
                goto error_exit;
            }
            break;
        case SESSION_SCAN_CREATE_TIME:
        {
            time_t tmp_range[2] = {0, 0};
            if (stm_time_range_pton(opt_value, time(NULL), tmp_range) < 0)
            {
                error_reply = monitor_reply_new_error("invalid create-time-in: %s. \r\nsyntax: <last-N-[seconds|minutes|hours|days]>\n", opt_value);
                goto error_exit;
            }
            brief_opt->scan_opt.create_time_ms[0] = tmp_range[0] * 1000; // second to ms
            brief_opt->scan_opt.create_time_ms[1] = tmp_range[1] * 1000; // second to ms
        }
        break;
        case SESSION_SCAN_LASPKT_TIME:
        {
            time_t tmp_range[2] = {0, 0};
            if (stm_time_range_pton(opt_value, time(NULL), tmp_range) < 0)
            {
                error_reply = monitor_reply_new_error("invalid last-pkt-time-in: %s. \r\nsyntax: <last-N-[seconds|minutes|hours|days]>\n", opt_value);
                goto error_exit;
            }
            brief_opt->scan_opt.laspkt_time_ms[0] = tmp_range[0] * 1000; // second to ms
            brief_opt->scan_opt.laspkt_time_ms[1] = tmp_range[1] * 1000; // second to ms
        }
        break;
        default:
            error_reply = monitor_reply_new_error("unrecognized params: %s \n", opt_name);
            return error_reply;
        }

        if ((history_ipaddr_family != 0) && (ipaddr_family != 0) && (history_ipaddr_family != ipaddr_family))
        {
            error_reply = monitor_reply_new_error("contradictory ip version, expression rejects all sessions!\n");
            goto error_exit;
        }
        history_ipaddr_family = ipaddr_family;
        i++; // to next option
        brief_opt->scan_opt.flags |= flag;
    }
    ss = show_session_cli_args_sanity_check(brief_opt);
    if (ss && sdslen(ss) > 0)
    {
        error_reply = monitor_reply_new_error("%s\n", ss);
        sdsfree(ss);
        goto error_exit;
    }
    sdsfree(ss);
    error_reply = NULL;
    return NULL;

error_exit:
    return error_reply;
}

static void get_single_session_brief(struct session *sess, int thread_id, struct show_session_brief_result *brief)
{
    brief->thread_index = thread_id;
    brief->state = session_get_current_state(sess);
    brief->protocol = session_get_type(sess);
    session_is_symmetric(sess, &brief->flow_dir);
    brief->create_time_in_sec = session_get_timestamp(sess, SESSION_TIMESTAMP_START) / 1000;  // ms to sec
    brief->last_pkt_time_in_sec = session_get_timestamp(sess, SESSION_TIMESTAMP_LAST) / 1000; // ms to sec
    const struct tuple6 *tp6 = session_get_tuple6(sess);
    memcpy(&brief->addr, tp6, sizeof(struct tuple6));
}

struct iovec show_session_brief_on_worker_thread_rpc_cb(int thread_idx, struct iovec request, void *args)
{
    struct iovec response = {};
    assert(request.iov_len == sizeof(struct stm_show_session_brief_opt));
    struct stm_show_session_brief_opt *show_brief_opt = (struct stm_show_session_brief_opt *)request.iov_base;

    uint64_t session_id_array[SHOW_SESSION_BRIEF_LIMIT_MAX];
    uint64_t session_id_array_count = SHOW_SESSION_BRIEF_LIMIT_MAX;
    struct session_manager *sess_mgr = stellar_module_get_session_manager((struct stellar_module_manager *)args);
    struct session_manager_rte *sess_mgr_rt = session_manager_get_runtime(sess_mgr, thread_idx);
    session_id_array_count = session_manager_rte_scan_session(sess_mgr_rt, &show_brief_opt->scan_opt, session_id_array, show_brief_opt->limit);
    if (session_id_array_count == 0)
    {
        // no session match the filter params, but no error! still need to build a empty reply!
        // go on !!!
        response.iov_base = NULL;
        response.iov_len = 0;
        return response;
    }

    struct show_session_brief_result *brief_result_array = (struct show_session_brief_result *)calloc(session_id_array_count, sizeof(struct show_session_brief_result));

    for (uint32_t i = 0; i < session_id_array_count; i++)
    {
        struct session *sess = session_manager_rte_lookup_session_by_id(sess_mgr_rt, session_id_array[i]);
        assert(sess != NULL);
        get_single_session_brief(sess, thread_idx, &brief_result_array[i]);
        brief_result_array[i].sid = session_id_array[i];
    }
    response.iov_base = brief_result_array;
    response.iov_len = session_id_array_count;
    return response;
}

static sds session_brief_to_readable(const struct show_session_brief_result *sess_brief)
{
    char time_buf[64];
    char addr_buf[256];
    sds ss = sdsempty();
    strftime(time_buf, sizeof(time_buf), "%Y-%m-%d %H:%M:%S", localtime(&sess_brief->create_time_in_sec));
    ss = sdscatprintf(ss, "%-6d %-21lu %-5s %-8s %-20s %s", sess_brief->thread_index, sess_brief->sid,
                      (sess_brief->protocol == SESSION_TYPE_TCP) ? "tcp" : "udp",
                      stm_session_state_ntop(sess_brief->state), time_buf,
                      stm_get0_readable_session_addr(&sess_brief->addr, addr_buf, sizeof(addr_buf)));
    return ss;
}

static struct monitor_reply *show_session_brief_cmd_cb(struct stellar_monitor *monitor, int argc, char *argv[], void *arg)
{
    struct stm_show_session_brief_opt brief_opt = {};
    if (show_session_is_help(argc, argv))
    {
        return show_session_brief_usage();
    }
    struct monitor_reply *error_reply = show_session_brief_cli_args_parse(argc, argv, &brief_opt);
    if (error_reply != NULL)
    {
        return error_reply;
    }
    struct monitor_reply *cmd_reply;
    struct stellar_module_manager *mod_mgr = (struct stellar_module_manager *)arg;
    int thread_num = stellar_module_manager_get_max_thread_num(mod_mgr);
    struct iovec request;
    request.iov_base = (void *)&brief_opt;
    request.iov_len = sizeof(struct stm_show_session_brief_opt);

    struct iovec response[thread_num];
    memset(response, 0, sizeof(response));
    size_t tot_sess_id_num = 0;
    sds ss = sdsempty();
    ss = sdscatprintf(ss, "%-6s %-21s %-5s %-8s %-20s %s\r\n", "thread", "session-id", "proto", "state", "create-time", "tuple4(sip:sport-dip:dport)");
    ss = sdscatprintf(ss, "--------------------------------------------------------------------------------------------\r\n");

    for (int i = 0; i < thread_num; i++)
    {
        response[i] = monitor_worker_thread_rpc(monitor, i, request, show_session_brief_on_worker_thread_rpc_cb, mod_mgr);
        if (response[i].iov_base == NULL || response[i].iov_len == 0) // empty result
        {
            continue;
        }
        struct show_session_brief_result *brief_res_array = (struct show_session_brief_result *)(response[i].iov_base);
        tot_sess_id_num += response[i].iov_len; // session_id_array_count
        for (size_t j = 0; j < response[i].iov_len; j++)
        {
            ss = sdscatprintf(ss, "%s\n", session_brief_to_readable(&brief_res_array[j]));
        }
        if (tot_sess_id_num >= brief_opt.limit)
        {
            break;
        }
    }

    if (tot_sess_id_num == 0)
    {
        cmd_reply = monitor_reply_new_string("No session found");
        goto empty_result;
    }
    cmd_reply = monitor_reply_new_string("%s", ss);

empty_result:
    sdsfree(ss);
    for (int i = 0; i < thread_num; i++)
    {
        if (response[i].iov_base)
        {
            free(response[i].iov_base);
        }
    }
    return cmd_reply;
}

/*
   todo: add thread id,
         fast patch, avoid traversing all worker threads
 */
static struct monitor_reply *show_session_detail_cli_args_parse(int argc, char *argv[], struct stm_show_session_detail_opt *detail_opt)
{
    if (argc < 4)
    {
        return monitor_reply_new_error("missing session id\n");
    }
    if (argc > 5)
    {
        return monitor_reply_new_error("too many arguments\n");
    }
    if (strncasecmp(argv[3], "id", 2) != 0)
    {
        return monitor_reply_new_error("missing session id\n");
    }

    if (stm_string_isdigit(argv[4]) == 0)
    {
        return monitor_reply_new_error("invalid session id: %s. should be integer in range: <1 - UINT64_MAX>\n", argv[4]);
    }
    detail_opt->sess_id = strtoull(argv[4], NULL, 10);
    return NULL;
}

struct iovec show_session_detail_on_worker_thread_rpc_cb(int thread_idx, struct iovec request, void *args)
{
    struct iovec response = {};
    assert(request.iov_len == sizeof(struct stm_show_session_detail_opt));
    struct stm_show_session_detail_opt *detail_opt = (struct stm_show_session_detail_opt *)request.iov_base;

    struct session_manager *sess_mgr = stellar_module_get_session_manager((struct stellar_module_manager *)args);
    struct session_manager_rte *sess_mgr_rt = session_manager_get_runtime(sess_mgr, thread_idx);

    struct session *sess = session_manager_rte_lookup_session_by_id(sess_mgr_rt, detail_opt->sess_id);
    if (NULL == sess)
    {
        return response;
    }

    struct show_session_detail_result *detail_res = (struct show_session_detail_result *)calloc(1, sizeof(struct show_session_detail_result));
    get_single_session_brief(sess, thread_idx, &detail_res->sess_brief);
    detail_res->sess_brief.sid = detail_opt->sess_id;
    detail_res->direction = session_get_direction(sess);
    // todo, get some exact stat, not all
    memcpy(detail_res->sess_stat, sess->stats, sizeof(detail_res->sess_stat));

    // todo, get application info

    response.iov_base = detail_res;
    response.iov_len = sizeof(struct show_session_detail_result);
    return response;
}

static sds session_detail_to_readable(const struct show_session_detail_result *sess_detail)
{
    char addr_buf[256];
    sds ss = sdsempty();
    char time_buf[64];
#define SHOW_SESSION_DETAIL_NAME_FORMAT "%-30s"
#define SHOW_SESSION_DETAIL_VALUE_FORMAT "%llu"
    const char *flow_str[MAX_FLOW_TYPE] = {"C2S flow", "S2C flow"};
    ss = sdscatprintf(ss, "%-15s: %lu\r\n", "session-id", sess_detail->sess_brief.sid);
    ss = sdscatprintf(ss, "%-15s: %d\r\n", "thread", sess_detail->sess_brief.thread_index);
    ss = sdscatprintf(ss, "%-15s: %s\r\n", "state", stm_session_state_ntop(sess_detail->sess_brief.state));
    ss = sdscatprintf(ss, "%-15s: %s\r\n", "protocol", (sess_detail->sess_brief.protocol == SESSION_TYPE_TCP) ? "tcp" : "udp");
    strftime(time_buf, sizeof(time_buf), "%Y-%m-%d %H:%M:%S", localtime(&sess_detail->sess_brief.create_time_in_sec));
    ss = sdscatprintf(ss, "%-15s: %s\r\n", "create-time", time_buf);
    strftime(time_buf, sizeof(time_buf), "%Y-%m-%d %H:%M:%S", localtime(&sess_detail->sess_brief.last_pkt_time_in_sec));
    ss = sdscatprintf(ss, "%-15s: %s\r\n", "last-pkt-time", time_buf);
    ss = sdscatprintf(ss, "%-15s: %s\r\n", "tuple4", stm_get0_readable_session_addr(&sess_detail->sess_brief.addr, addr_buf, sizeof(addr_buf)));
    ss = sdscatprintf(ss, "%-15s: %s\r\n", "symmetric", stm_session_flow_dir_ntop(sess_detail->sess_brief.flow_dir));
    ss = sdscatprintf(ss, "%-15s: %s\r\n", "direction", sess_detail->direction == SESSION_DIRECTION_INBOUND ? "INBOUND" : "OUTBOUND");

    ss = sdscatprintf(ss, "statistics:\r\n");
    char printf_format[256];
    snprintf(printf_format, sizeof(printf_format), "\t%s : %s\r\n", SHOW_SESSION_DETAIL_NAME_FORMAT, SHOW_SESSION_DETAIL_VALUE_FORMAT);
    for (int flow = 0; flow < MAX_FLOW_TYPE; flow++)
    {
        ss = sdscatprintf(ss, "  %s:\r\n", flow_str[flow]);
        ss = sdscatprintf(ss, printf_format, "raw_packets_received", sess_detail->sess_stat[flow][STAT_RAW_PACKETS_RECEIVED]);
        ss = sdscatprintf(ss, printf_format, "raw_bytes_received", sess_detail->sess_stat[flow][STAT_RAW_BYTES_RECEIVED]);
        ss = sdscatprintf(ss, printf_format, "duplicate_packets_bypass", sess_detail->sess_stat[flow][STAT_DUPLICATE_PACKETS_BYPASS]);
        ss = sdscatprintf(ss, printf_format, "injected_packets", sess_detail->sess_stat[flow][STAT_INJECTED_PACKETS_SUCCESS]);
        ss = sdscatprintf(ss, printf_format, "injected_bytes", sess_detail->sess_stat[flow][STAT_INJECTED_BYTES_SUCCESS]);
        if (SESSION_TYPE_TCP == sess_detail->sess_brief.protocol)
        {
            ss = sdscatprintf(ss, printf_format, "tcp_segments_retransmit", sess_detail->sess_stat[flow][STAT_TCP_SEGMENTS_RETRANSMIT]);
            ss = sdscatprintf(ss, printf_format, "tcp_payloads_retransmit", sess_detail->sess_stat[flow][STAT_TCP_PAYLOADS_RETRANSMIT]);
            ss = sdscatprintf(ss, printf_format, "tcp_segments_reordered", sess_detail->sess_stat[flow][STAT_TCP_SEGMENTS_REORDERED]);
            ss = sdscatprintf(ss, printf_format, "tcp_payloads_reordered", sess_detail->sess_stat[flow][STAT_TCP_PAYLOADS_REORDERED]);
        }
    }

    // todo:
    ss = sdscatprintf(ss, "\r\n  \033[31mtodo: add security policy rule id, HTTP URL, Server FQDN, etc... \033[0m");
    return ss;
}

static struct monitor_reply *show_session_detail_cmd_cb(struct stellar_monitor *stm, int argc, char *argv[], void *arg)
{
    struct stm_show_session_detail_opt detail_opt = {};
    detail_opt.thread_idx = -1;

    if (show_session_is_help(argc, argv))
    {
        return show_session_detail_usage();
    }
    struct monitor_reply *error_reply = show_session_detail_cli_args_parse(argc, argv, &detail_opt);
    if (error_reply != NULL)
    {
        return error_reply;
    }

    struct monitor_reply *cmd_reply;
    struct stellar_module_manager *mod_mgr = (struct stellar_module_manager *)arg;
    int thread_num = stellar_module_manager_get_max_thread_num(mod_mgr);
    struct iovec request;
    request.iov_base = (void *)&detail_opt;
    request.iov_len = sizeof(struct stm_show_session_detail_opt);

    struct iovec response[thread_num];
    memset(response, 0, sizeof(response));
    size_t tot_sess_id_num = 0;
    sds ss = sdsempty();

    for (int i = 0; i < thread_num; i++)
    {
        if (detail_opt.thread_idx != -1 && detail_opt.thread_idx != i)
        {
            continue;
        }
        response[i] = monitor_worker_thread_rpc(stm, i, request, show_session_detail_on_worker_thread_rpc_cb, mod_mgr);
        if (response[i].iov_base == NULL || response[i].iov_len == 0) // empty result
        {
            continue;
        }
        struct show_session_detail_result *detail_res = (struct show_session_detail_result *)(response[i].iov_base);
        ss = sdscatprintf(ss, "%s\n", session_detail_to_readable(detail_res));
        tot_sess_id_num++;
        break;
    }
    if (tot_sess_id_num == 0)
    {
        cmd_reply = monitor_reply_new_string("No session found by id %lu", detail_opt.sess_id);
        goto empty_result;
    }
    cmd_reply = monitor_reply_new_string("%s", ss);

empty_result:
    sdsfree(ss);
    for (int i = 0; i < thread_num; i++)
    {
        if (response[i].iov_base)
        {
            free(response[i].iov_base);
        }
    }
    return cmd_reply;
}

int show_session_enforcer_init(struct stellar_module_manager *mod_mgr, struct stellar_monitor *stm)
{
    monitor_register_cmd(stm, "show session brief", show_session_brief_cmd_cb, "readonly",
                         "[sip | dip | sport | dport | protocol | state | cursor | count | limit | create-time-in | last-pkt-time-in ]",
                         "Show session brief information", (void *)mod_mgr);
    monitor_register_cmd(stm, "show session detail", show_session_detail_cmd_cb, "readonly",
                         "id <id>",
                         "Show session verbose information", (void *)mod_mgr);
    return 0;
}