#include "lua_module_manage_internal.h"
#include "lua_binding_function.h"
#include "lua_binding_cfunc.h"

#include "stellar/mq.h"
#include "stellar/module_manager.h"
#include "stellar/packet.h"
#include "stellar/packet_manager.h"
#include "stellar/session.h"
#include "stellar/session_manager.h"
#include "stellar/utils.h"

#include <stdlib.h>
#include <string.h>
#include <utlist.h>
#include <lua.h>
#include <lualib.h>
#include <lauxlib.h>

#define LUA_BINDING_FUNCTION_GET_VALUE_POINTER(func_name, param_type, value_type, value_func) \
    int func_name(struct lua_state *state)                                                    \
    {                                                                                         \
        lua_State *L = (lua_State *)lua_state_get_lua_State(state);                           \
        if (lua_gettop(L) != 1)                                                               \
        {                                                                                     \
            lua_settop(L, 0);                                                                 \
            return 0;                                                                         \
        }                                                                                     \
        param_type *p = (param_type *)lua_topointer(L, -1);                                   \
        lua_settop(L, 0);                                                                     \
        value_type *l = (value_type *)value_func(p);                                          \
        lua_pushlightuserdata(L, (void *)l);                                                  \
        return 1;                                                                             \
    }

#define LUA_BINDING_FUNCTION_GET_VALUE_INT(func_name, param_type, value_func) \
    int func_name(struct lua_state *state)                                    \
    {                                                                         \
        lua_State *L = (lua_State *)lua_state_get_lua_State(state);           \
        if (lua_gettop(L) != 1)                                               \
        {                                                                     \
            lua_settop(L, 0);                                                 \
            return 0;                                                         \
        }                                                                     \
        param_type *p = (param_type *)lua_topointer(L, -1);                   \
        lua_settop(L, 0);                                                     \
        int v = (int)value_func(p);                                           \
        lua_pushinteger(L, v);                                                \
        return 1;                                                             \
    }

int lua_mq_schema_get_topic_id(struct lua_state *state)
{
    lua_State *L = (lua_State *)lua_state_get_lua_State(state);
    if (lua_gettop(L) != 2 || lua_type(L, -1) != LUA_TSTRING || lua_type(L, -2) != LUA_TLIGHTUSERDATA)
    {
        lua_settop(L, 0);
        return 0;
    }

    char *topic_name = strdup(lua_tostring(L, -1));
    lua_pop(L, 1);
    struct mq_schema *s = (struct mq_schema *)lua_topointer(L, -1);
    lua_settop(L, 0);

    int topic_id = mq_schema_get_topic_id(s, (const char *)topic_name);

    if (topic_name)
        FREE(topic_name);
    lua_pushinteger(L, topic_id);
    return 1;
}

int lua_mq_schema_create_topic(struct lua_state *state)
{
    lua_State *L = (lua_State *)lua_state_get_lua_State(state);
    if (lua_gettop(L) != 6 || lua_type(L, -5) != LUA_TSTRING || lua_type(L, -6) != LUA_TLIGHTUSERDATA)
    {
        lua_settop(L, 0);
        return 0;
    }

    int free_arg_ref_id = luaL_ref(L, LUA_REGISTRYINDEX);
    int free_cb_fn_ref_id = -1;
    if (lua_type(L, -1) == LUA_TFUNCTION)
        free_cb_fn_ref_id = luaL_ref(L, LUA_REGISTRYINDEX);
    else
        lua_pop(L, 1);
    int on_dispatch_arg_ref_id = luaL_ref(L, LUA_REGISTRYINDEX);
    int on_dispatch_cb_fn_ref_id = -1;
    if (lua_type(L, -1) == LUA_TFUNCTION)
        on_dispatch_cb_fn_ref_id = luaL_ref(L, LUA_REGISTRYINDEX);
    else
        lua_pop(L, 1);
    char *topic_name = strdup((char *)lua_tostring(L, -1));
    lua_pop(L, 1);
    struct mq_schema *s = (struct mq_schema *)lua_topointer(L, -1);
    lua_settop(L, 0);

    struct lua_module_manager *lua_mod_mgr = lua_state_get_lua_module_manager(state);
    struct lua_fn_arg_pair *new_mq_dispatch_arg = lua_fn_arg_pair_new(lua_mod_mgr, on_dispatch_cb_fn_ref_id, on_dispatch_arg_ref_id);
    struct lua_fn_arg_pair *new_mq_msg_free_arg = lua_fn_arg_pair_new(lua_mod_mgr, free_cb_fn_ref_id, free_arg_ref_id);
    int topic_id = mq_schema_create_topic(s,
                                          (const char *)topic_name,
                                          lua_cfunc_mq_on_msg_dispatch_cb_func,
                                          new_mq_dispatch_arg,
                                          lua_cfunc_mq_msg_free_cb_func,
                                          new_mq_msg_free_arg);
    if (topic_id >= 0)
    {
        lua_fn_arg_pair_insert(new_mq_dispatch_arg);
        lua_fn_arg_pair_insert(new_mq_msg_free_arg);
    }
    else
    {
        if (free_arg_ref_id + 1)
            luaL_unref(L, LUA_REGISTRYINDEX, free_arg_ref_id);
        if (free_cb_fn_ref_id + 1)
            luaL_unref(L, LUA_REGISTRYINDEX, free_cb_fn_ref_id);
        if (on_dispatch_arg_ref_id + 1)
            luaL_unref(L, LUA_REGISTRYINDEX, on_dispatch_arg_ref_id);
        if (on_dispatch_cb_fn_ref_id + 1)
            luaL_unref(L, LUA_REGISTRYINDEX, on_dispatch_cb_fn_ref_id);
        lua_fn_arg_pair_free(new_mq_dispatch_arg);
        lua_fn_arg_pair_free(new_mq_msg_free_arg);
    }

    lua_pushinteger(L, topic_id);
    return 1;
}

int lua_mq_schema_update_topic(struct lua_state *state)
{
    lua_State *L = (lua_State *)state;
    if (lua_gettop(L) != 6 || lua_type(L, -5) != LUA_TNUMBER || lua_type(L, -6) != LUA_TLIGHTUSERDATA)
    {
        lua_settop(L, 0);
        return 0;
    }

    int free_arg_ref_id = luaL_ref(L, LUA_REGISTRYINDEX);
    int free_cb_fn_ref_id = -1;
    if (lua_type(L, -1) == LUA_TFUNCTION)
        free_cb_fn_ref_id = luaL_ref(L, LUA_REGISTRYINDEX);
    else
        lua_pop(L, 1);
    int on_dispatch_arg_ref_id = luaL_ref(L, LUA_REGISTRYINDEX);
    int on_dispatch_cb_fn_ref_id = -1;
    if (lua_type(L, -1) == LUA_TFUNCTION)
        on_dispatch_cb_fn_ref_id = luaL_ref(L, LUA_REGISTRYINDEX);
    else
        lua_pop(L, 1);
    int topic_id = lua_tointeger(L, -1);
    lua_pop(L, 1);
    struct mq_schema *s = (struct mq_schema *)lua_topointer(L, -1);
    lua_settop(L, 0);

    struct lua_module_manager *lua_mod_mgr = lua_state_get_lua_module_manager(state);
    struct lua_fn_arg_pair *new_mq_dispatch_arg = lua_fn_arg_pair_new(lua_mod_mgr, on_dispatch_cb_fn_ref_id, on_dispatch_arg_ref_id);
    struct lua_fn_arg_pair *new_mq_msg_free_arg = lua_fn_arg_pair_new(lua_mod_mgr, free_cb_fn_ref_id, free_arg_ref_id);
    int update_ret = mq_schema_update_topic(s,
                                            topic_id,
                                            lua_cfunc_mq_on_msg_dispatch_cb_func,
                                            new_mq_dispatch_arg,
                                            lua_cfunc_mq_msg_free_cb_func,
                                            new_mq_msg_free_arg);
    if (topic_id)
    {
        if (free_arg_ref_id + 1)
            luaL_unref(L, LUA_REGISTRYINDEX, free_arg_ref_id);
        if (free_cb_fn_ref_id + 1)
            luaL_unref(L, LUA_REGISTRYINDEX, free_cb_fn_ref_id);
        if (on_dispatch_arg_ref_id + 1)
            luaL_unref(L, LUA_REGISTRYINDEX, on_dispatch_arg_ref_id);
        if (on_dispatch_cb_fn_ref_id + 1)
            luaL_unref(L, LUA_REGISTRYINDEX, on_dispatch_cb_fn_ref_id);
        lua_fn_arg_pair_free(new_mq_dispatch_arg);
        lua_fn_arg_pair_free(new_mq_msg_free_arg);
    }
    else
    {
        lua_fn_arg_pair_insert(new_mq_dispatch_arg);
        lua_fn_arg_pair_insert(new_mq_msg_free_arg);
    }

    lua_pushinteger(L, update_ret);
    return 1;
}

int lua_mq_shcema_destory_topic(struct lua_state *state)
{
    lua_State *L = (lua_State *)state;
    if (lua_gettop(L) != 2 || lua_type(L, -1) != LUA_TNUMBER || lua_type(L, -2) != LUA_TLIGHTUSERDATA)
    {
        lua_settop(L, 0);
        return 0;
    }

    int topic_id = lua_tointeger(L, -1);
    lua_pop(L, 1);
    struct mq_schema *s = (struct mq_schema *)lua_topointer(L, -1);
    lua_settop(L, 0);

    int destory_ret = mq_schema_destroy_topic(s, topic_id);

    lua_pushinteger(L, destory_ret);
    return 1;
}

int lua_mq_schema_subscribe(struct lua_state *state)
{
    lua_State *L = (lua_State *)state;
    if (lua_gettop(L) != 4 || lua_type(L, -2) != LUA_TFUNCTION || lua_type(L, -4) != LUA_TLIGHTUSERDATA)
    {
        lua_settop(L, 0);
        return 0;
    }

    int on_msg_arg_ref_id = luaL_ref(L, LUA_REGISTRYINDEX);
    int on_msg_fn_ref_id = luaL_ref(L, LUA_REGISTRYINDEX);
    int topic_id = lua_tointeger(L, -1);
    lua_pop(L, 1);
    struct mq_schema *s = (struct mq_schema *)lua_topointer(L, -1);
    lua_settop(L, 0);

    struct lua_module_manager *lua_mod_mgr = lua_state_get_lua_module_manager(state);
    struct lua_fn_arg_pair *new_mq_on_msg_arg = lua_fn_arg_pair_new(lua_mod_mgr, on_msg_fn_ref_id, on_msg_arg_ref_id);
    int subscribe_ret = mq_schema_subscribe(s, topic_id, lua_cfunc_mq_on_msg_cb_func, new_mq_on_msg_arg);
    if (subscribe_ret)
    {
        if (on_msg_arg_ref_id + 1)
            luaL_unref(L, LUA_REGISTRYINDEX, on_msg_arg_ref_id);
        if (on_msg_fn_ref_id + 1)
            luaL_unref(L, LUA_REGISTRYINDEX, on_msg_fn_ref_id);
        lua_fn_arg_pair_free(new_mq_on_msg_arg);
    }
    else
    {
        lua_fn_arg_pair_insert(new_mq_on_msg_arg);
    }

    lua_pushinteger(L, subscribe_ret);
    return 1;
}

int lua_mq_runtime_publish_message(struct lua_state *state)
{
    lua_State *L = (lua_State *)state;
    if (lua_gettop(L) != 3 || lua_type(L, -2) != LUA_TNUMBER || lua_type(L, -3) != LUA_TLIGHTUSERDATA)
    {
        lua_settop(L, 0);
        return 0;
    }

    int msg_ref_id = luaL_ref(L, LUA_REGISTRYINDEX);
    int topic_id = lua_tointeger(L, -1);
    lua_pop(L, 1);
    struct mq_runtime *rt = (struct mq_runtime *)lua_topointer(L, -1);
    lua_settop(L, 0);

    struct lua_context *new_context = lua_context_new_with_ref_id(state, msg_ref_id);

    int publish_ret = mq_runtime_publish_message(rt, topic_id, new_context);
    if (publish_ret)
    {
        lua_context_free(new_context);
    }
    lua_pushinteger(L, publish_ret);
    return 1;
}

LUA_BINDING_FUNCTION_GET_VALUE_POINTER(lua_stellar_get_stellar_module_manager,
                                       struct lua_module_manager,
                                       struct stellar_module_manager,
                                       lua_module_manager_get_stellar_module_manager)

LUA_BINDING_FUNCTION_GET_VALUE_INT(lua_stellar_get_thread_id,
                                   struct stellar_module_manager,
                                   stellar_module_manager_get_thread_id)

LUA_BINDING_FUNCTION_GET_VALUE_INT(lua_stellar_get_max_thread_num,
                                   struct stellar_module_manager,
                                   stellar_module_manager_get_max_thread_num)

LUA_BINDING_FUNCTION_GET_VALUE_POINTER(lua_stellar_get_mq_runtime,
                                       struct stellar_module_manager,
                                       struct mq_runtime,
                                       stellar_module_manager_get_mq_runtime)

LUA_BINDING_FUNCTION_GET_VALUE_POINTER(lua_stellar_get_mq_schema,
                                       struct stellar_module_manager,
                                       struct mq_schema,
                                       stellar_module_manager_get_mq_schema)

LUA_BINDING_FUNCTION_GET_VALUE_POINTER(lua_stellar_get_logger,
                                       struct stellar_module_manager,
                                       struct logger,
                                       stellar_module_manager_get_logger)

LUA_BINDING_FUNCTION_GET_VALUE_INT(lua_packet_get_direction,
                                   const struct packet,
                                   packet_get_direction)

LUA_BINDING_FUNCTION_GET_VALUE_POINTER(lua_packet_get_payload,
                                       const struct packet,
                                       const char,
                                       packet_get_payload)

LUA_BINDING_FUNCTION_GET_VALUE_INT(lua_packet_get_payload_len,
                                   const struct packet,
                                   packet_get_payload_len)

int lua_packet_manager_subscribe(struct lua_state *state)
{
    lua_State *L = (lua_State *)state;
    if (lua_gettop(L) != 4 || lua_type(L, -2) != LUA_TFUNCTION || lua_type(L, -3) != LUA_TNUMBER || lua_type(L, -4) != LUA_TLIGHTUSERDATA)
    {
        lua_settop(L, 0);
        return 0;
    }

    int on_packet_stage_arg_ref_id = luaL_ref(L, LUA_REGISTRYINDEX);
    int on_packet_stage_fn_ref_id = luaL_ref(L, LUA_REGISTRYINDEX);
    int stage = lua_tointeger(L, -1);
    lua_pop(L, 1);
    struct packet_manager *pkt_mgr = (struct packet_manager *)lua_topointer(L, -1);
    lua_settop(L, 0);

    struct lua_module_manager *lua_mod_mgr = lua_state_get_lua_module_manager(state);
    struct lua_fn_arg_pair *new_on_packet_arg = lua_fn_arg_pair_new(lua_mod_mgr, on_packet_stage_fn_ref_id, on_packet_stage_arg_ref_id);
    int subscribe_ret = packet_manager_subscribe(pkt_mgr, (enum packet_stage)stage, lua_cfunc_packet_on_stage_callback, new_on_packet_arg);
    if (subscribe_ret)
    {
        if (on_packet_stage_arg_ref_id + 1)
            luaL_unref(L, LUA_REGISTRYINDEX, on_packet_stage_arg_ref_id);
        if (on_packet_stage_fn_ref_id + 1)
            luaL_unref(L, LUA_REGISTRYINDEX, on_packet_stage_fn_ref_id);
        lua_fn_arg_pair_free(new_on_packet_arg);
    }
    else
    {
        lua_fn_arg_pair_insert(new_on_packet_arg);
    }

    lua_pushinteger(L, subscribe_ret);
    return 1;
}

LUA_BINDING_FUNCTION_GET_VALUE_POINTER(lua_session_get0_current_packet,
                                       const struct session,
                                       const struct packet,
                                       session_get0_current_packet)

enum sess_mgr_sub_type
{
    sub_type_begin = 0,
    sub_type_tcp,
    sub_type_udp,
    sub_type_control_pkt,
    sub_type_end
};

static int lua_session_manager_subscribe(struct lua_state *state, enum sess_mgr_sub_type type)
{
    lua_State *L = (lua_State *)state;
    if (lua_gettop(L) != 3 || lua_type(L, -2) != LUA_TFUNCTION || lua_type(L, -3) != LUA_TLIGHTUSERDATA)
    {
        lua_settop(L, 0);
        return 0;
    }

    int on_session_arg_ref_id = luaL_ref(L, LUA_REGISTRYINDEX);
    int on_session_fn_ref_id = luaL_ref(L, LUA_REGISTRYINDEX);
    struct session_manager *sess_mgr = (struct session_manager *)lua_topointer(L, -1);
    lua_settop(L, 0);

    struct lua_module_manager *lua_mod_mgr = lua_state_get_lua_module_manager(state);
    struct lua_fn_arg_pair *new_on_sess_arg = lua_fn_arg_pair_new(lua_mod_mgr, on_session_fn_ref_id, on_session_arg_ref_id);
    int subscribe_ret = -1;
    switch (type)
    {
    case sub_type_tcp:
        subscribe_ret = session_manager_subscribe_tcp(sess_mgr, lua_cfunc_session_callback, new_on_sess_arg);
        break;
    case sub_type_udp:
        subscribe_ret = session_manager_subscribe_udp(sess_mgr, lua_cfunc_session_callback, new_on_sess_arg);
        break;
    case sub_type_control_pkt:
        subscribe_ret = session_manager_subscribe_control_packet(sess_mgr, lua_cfunc_session_callback, new_on_sess_arg);
        break;
    default:
        break;
    }
    if (subscribe_ret)
    {
        if (on_session_arg_ref_id + 1)
            luaL_unref(L, LUA_REGISTRYINDEX, on_session_arg_ref_id);
        if (on_session_fn_ref_id + 1)
            luaL_unref(L, LUA_REGISTRYINDEX, on_session_fn_ref_id);
        lua_fn_arg_pair_free(new_on_sess_arg);
    }
    else
    {
        lua_fn_arg_pair_insert(new_on_sess_arg);
    }

    lua_pushinteger(L, subscribe_ret);
    return 1;
}

int lua_session_manager_subscribe_tcp(struct lua_state *state)
{
    return lua_session_manager_subscribe(state, sub_type_tcp);
}

int lua_session_manager_subscribe_udp(struct lua_state *state)
{
    return lua_session_manager_subscribe(state, sub_type_udp);
}

int lua_session_manager_subscribe_control_packet(struct lua_state *state)
{
    return lua_session_manager_subscribe(state, sub_type_control_pkt);
}

int lua_session_manager_subscribe_tcp_stream(struct lua_state *state)
{
    lua_State *L = (lua_State *)state;
    if (lua_gettop(L) != 3 || lua_type(L, -2) != LUA_TFUNCTION || lua_type(L, -4) != LUA_TLIGHTUSERDATA)
    {
        lua_settop(L, 0);
        return 0;
    }

    int on_tcp_stream_arg_ref_id = luaL_ref(L, LUA_REGISTRYINDEX);
    int on_tcp_stream_fn_ref_id = luaL_ref(L, LUA_REGISTRYINDEX);
    struct session_manager *sess_mgr = (struct session_manager *)lua_topointer(L, -1);
    lua_settop(L, 0);

    struct lua_module_manager *lua_mod_mgr = lua_state_get_lua_module_manager(state);
    struct lua_fn_arg_pair *new_on_tcp_stream_arg = lua_fn_arg_pair_new(lua_mod_mgr, on_tcp_stream_fn_ref_id, on_tcp_stream_arg_ref_id);
    int subscribe_ret = session_manager_subscribe_tcp_stream(sess_mgr, lua_cfunc_tcp_stream_callback, new_on_tcp_stream_arg);
    if (subscribe_ret)
    {
        if (on_tcp_stream_arg_ref_id + 1)
            luaL_unref(L, LUA_REGISTRYINDEX, on_tcp_stream_arg_ref_id);
        if (on_tcp_stream_fn_ref_id + 1)
            luaL_unref(L, LUA_REGISTRYINDEX, on_tcp_stream_fn_ref_id);
        lua_fn_arg_pair_free(new_on_tcp_stream_arg);
    }
    else
    {
        lua_fn_arg_pair_insert(new_on_tcp_stream_arg);
    }

    lua_pushinteger(L, subscribe_ret);
    return 1;
}