#include <sys/select.h>
#include <sys/time.h>
#include <sys/types.h>
#include <unistd.h>
#include <sys/syscall.h>
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <errno.h>

#include "thread_comm.h"
#include "thread_lora.h"
#define LORA_POLL_DATA_MAX 200
#define DOWNLINK_LIMIT 100

gw_port_e ports[] = {LORA_1, LORA_DTU_RS485};

typedef struct {
    char sn_str[SN_MAX_LEN];
    char proc_name[PROGRAM_NAME_LEN];
    ipc_session_t* session; //内部broker的MQTT client，用于进程间通信
    nodes_cfg_table_t *nodes_cfg_table; //节点信息
    template_table_t *template_table; //模板信息
    period_service_t *period;
    dev_status_table_t *ds; //下挂节点的状态信息
    kv_array_t identifier_backup;
    time_t nodes_status_sync_date;
    time_t poll_preiod_msg;

    lora_cfg_t lora_cfg;
//此处根据模块需要添加
} thread_comm_t;

typedef struct
{
    char sn[SN_MAX_LEN];
    char src_identifier[64];
    time_t donwlink_time;
    int mi;
} my_regulate_cmd_backup_t;

bool exit_sig = false;
static bool service_flag = true;
static thread_comm_t var = {0};



static int dylora_data_downlink(ipc_msg_t *mqtt_msg) {
    pp_regulate_signal_t regulate_data;
    char *sn = NULL;

    int ret = get_rglt_data_from_json(&regulate_data, mqtt_msg->payload);
    if (ret < 0)
    {
        dy_syslog(LOG_ERR, "parse real data structure failed");
        return -1;
    }

    if(regulate_data.len < 3)
    {
        dy_syslog(LOG_ERR,"lora data format error!");
        free(regulate_data.data);
        return -1;
    }
    exp_lora_msg_t pram;
    char *p = (char *)regulate_data.data;
    p+=2;  //skip addr
    memcpy(&pram.code,p,1);p+=1;
    pram.len = regulate_data.len - 3;
    memcpy(&pram.data,p,pram.len);p+=pram.len;
    pram.retry = 0;
    pram.times = VALID_TIME_DEFAULT;
    if(!strcmp(regulate_data.sn,"*"))//broadcast
    {
        pram.addr.b = 0xff;
    }
    else
    {
        if (regulate_data.port == LORA_DTU_RS485)
        {
            sn = regulate_data.dtu_sn;
        }
        else
        {
            sn = regulate_data.sn;
        }
        node_cfg_t* cfg = dylora_cfg_search_by_sn(sn);
        if(!cfg)
        {
            free(regulate_data.data);
            return -1;
        }
        pram.addr.b = cfg->term_addr[1];
    }

    if(regulate_data.period > 0) {
        dy_syslog(LOG_INFO,"recv interval ctrl cmd, node addr [%02X,%02X],periad %d",pram.addr.a,pram.addr.b,regulate_data.period);
//        dylora_poll_msg_add(&pram,regulate_data.period);
        period_msg_t once;
        snprintf(once.key, 128, "%s_%s", regulate_data.sn, regulate_data.src_identifier);
        once.interval = regulate_data.period;
        once.last_poll = 0;
        once.rs = calloc(1, sizeof(pp_regulate_signal_t));
        memcpy(once.rs, &regulate_data, sizeof(pp_regulate_signal_t));
        once.started = 1;
        once.rs->data = calloc(1, regulate_data.len);
        memcpy(once.rs->data, regulate_data.data, regulate_data.len);

        period_msg_add(var.period, &once);
    }
    else {
        dy_syslog(LOG_INFO,"recv ctrl cmd to lora dev");
        int class = dylora_class_get(&pram.addr);
        if (class == 1) {
            exp_lora_msg_t tmp;
            int ret = msg_get_compare(DELAY_MSG_LIST,	pram.addr, &tmp);
            if(ret)
            {
                dy_syslog(LOG_WARNING,"discard msg of same device");
                return -1;
            }
            pram.retry = 1;
            pram.times = 24*3600L;
            msg_add_tail(DELAY_MSG_LIST, &pram);   
        } else if (class == 2) {
            int i = list_size(GENERAL_DOWNLINK_LIST);
            if(i >= DOWNLINK_LIMIT/2)
            {
                dy_syslog(LOG_WARNING,"queue is full");
                return -1;
            }
            pram.retry = 2;
            pram.times = VALID_TIME_DEFAULT;
            msg_add_tail(GENERAL_DOWNLINK_LIST, &pram);

            char key_str[64] = {0};
            my_regulate_cmd_backup_t fields;
            fields.donwlink_time  = time(NULL);
            strcpy(fields.src_identifier, regulate_data.src_identifier);
            strcpy(fields.sn, regulate_data.sn);
            fields.mi = regulate_data.mi;
            sprintf(key_str, "dy_lora_%s_%X",sn,(pram.code & 0x7f));
            var.identifier_backup.insert(&var.identifier_backup, key_str, &fields, sizeof(fields));
            flush_device_status_by_sn(var.ds, ACTION_SND, fields.sn);
        }
    }
    free(regulate_data.data);
    return 0;
}

static int dylora_data_uplink(exp_lora_msg_t *lora) {

    pp_real_time_data_t real_data = {0};
    real_data.port = LORA_1;
    real_data.len = lora->len + 3;
    real_data.instruction_code = lora->code;
    real_data.data = malloc(real_data.len);

    my_regulate_cmd_backup_t *fields = NULL;
    char key_str[64] = {0};
    data_info_t *data_info = NULL;
    node_cfg_t* cfg = dylora_cfg_search_by_addr(&lora->addr);
    if(!cfg)
    {
        free(real_data.data);
        return -1;
    }
    strcpy(real_data.sn, cfg->sn);

    sprintf(key_str, "dy_lora_%s_%X",real_data.sn,(lora->code & 0x7f));
    data_info = var.identifier_backup.get(&var.identifier_backup, key_str);
    if (data_info)
    {
        fields = (my_regulate_cmd_backup_t *)data_info->data;
    }
    if (fields != NULL && time(NULL) < fields->donwlink_time + 15)
    {
        real_data.mi = fields->mi;
        strncpy(real_data.src_identifier, fields->src_identifier, sizeof(real_data.src_identifier));
    }

    char *p = (char *)real_data.data;
    lora->addr.a = 0;//compatible term_addr 2 bytes,now modify device config term_add is 1 bytes
    memcpy(p,&lora->addr,2);p+=2;
    memcpy(p,&lora->code,1);p+=1;
    memcpy(p,lora->data,lora->len);p+=lora->len;
    send_to_proto_parser(var.session, &real_data);
    flush_device_status_by_sn(var.ds, ACTION_RCV, real_data.sn);
    free(real_data.data);

    return 0;
}

static int dylora_data_downlink_response(exp_lora_msg_t *pram,int status) {
    char topic[TOPIC_MAX_LEN];
    int ret = 0;

    char *msg = NULL;
    node_cfg_t* cfg = dylora_cfg_search_by_addr(&pram->addr);
    if(!cfg && cfg->sn)
        msg = downlink_response("*",status);
    else
        msg = downlink_response(cfg->sn,status);
    if(!msg)return -1;

    free(msg);
    return 0;
}

void dylora_node_register(void *info,int len,char *sn)
{
    flush_device_status_soft_info_by_sn(var.ds, info,len,sn);
}



node_status_t *dylora_status_serach_by_addr(lora_addr_t *addr)
{
    int i;
    node_status_t * tmp = NULL;
    if(!var.ds)return tmp;
    for(i = 0; i< var.ds->node_cnt ;i++)
    {
        if(var.ds->status[i].addr[1] == addr->b)
        {
            tmp = &var.ds->status[i];
            break;
        }
    }
    return tmp;
}


int dylora_class_get(lora_addr_t *addr)
{
    if(addr->b == 0xff)return 2; //classC
    node_status_t * tmp = NULL;
    tmp = dylora_status_serach_by_addr(addr);
    if(tmp)
        return tmp->class;
    else return -1;
}

node_cfg_t* dylora_cfg_serach_by_reginfo(register_info_t *info)
{
    int i;
    node_cfg_t *cfg = NULL;
    for (i = 0;i < var.nodes_cfg_table->node_cnt;i++) {
        uint32_t sn = strtoull(var.nodes_cfg_table->node[i].sn,NULL,16);
        if(sn == info->sn)
        {
            cfg = &var.nodes_cfg_table->node[i];
            return cfg;
        }
    }
    return cfg;
}

node_cfg_t* dylora_cfg_search_by_addr(lora_addr_t *addr)
{
    int i;
    node_cfg_t *cfg = NULL;
    for (i = 0;i < var.nodes_cfg_table->node_cnt;i++) {
        if(addr->b == var.nodes_cfg_table->node[i].term_addr[1])
        {
            cfg = &var.nodes_cfg_table->node[i];
            return cfg;
        }
    }
    return cfg;
}

node_cfg_t* dylora_cfg_search_by_sn(char *sn)
{
    int i;
    node_cfg_t *cfg = NULL;
    for (i = 0;i < var.nodes_cfg_table->node_cnt;i++) {
        if(!strncmp(sn,var.nodes_cfg_table->node[i].sn,sizeof(var.nodes_cfg_table->node[i].sn)))
        {
            cfg = &var.nodes_cfg_table->node[i];
            return cfg;
        }
    }
    return cfg;
}

static int dylora_private_cfg_paser(const char *str)
{
    int ret;
    if(!str)return -1;
    char buff[255];
    cJSON* root=cJSON_Parse(str);
    if(!root)           return -1;
    cJSON* mode=cJSON_GetObjectItem(root,"mode");
    if(mode)
    {
        cJSON* radio1=cJSON_GetObjectItem(root,"radio1");
        if(!strcmp(mode->valuestring,"disable"))//解析radio配置参数由区别
        {
            return -1;//关闭lora射频
        }
        else if(!strcmp(mode->valuestring,"lora_dy"))//解析radio配置参数由区别
        {
            if(!radio1)goto out;
            GET_JSON_VALUE_INT(radio1,"frq_band",var.lora_cfg.frq_band);
            GET_JSON_VALUE_INT(radio1,"frq_code",var.lora_cfg.frq_code);
            GET_JSON_VALUE_INT(radio1,"gw_lora_addr",var.lora_cfg.gw_lora_addr);
            return 0;
        }
    }
    out:
    cJSON_Delete(root);
    return -1;
}

int dylora_config()
{
    char *json_str=NULL;
    json_str = read_file_data(LORAWAN_MODULE_CFG);
    int ret = dylora_private_cfg_paser(json_str);
    if(ret <0)
    {
        dy_syslog(LOG_ERR, "lora config load error");
        dy_syslog(LOG_ERR, "lora service will be exit");
        return -1;
    }
    if(json_str)free(json_str);
    return 0;
}


static void dylora_subscribe_all()
{
    ipc_session_t *ipc_session = var.session;
    char topic[TOPIC_MAX_LEN] = {0};
    int i = 0;

    for (i = 0; i < ARRAY_SIZE(ports); i++)
    {
        snprintf(topic, TOPIC_MAX_LEN, "ipc/+/%s/device/+/data/%s", port_enum2char(ports[i]), TOPIC_SEND_RGLT_SIGNAL_RAW_DATA);
        ipc_session_subscribe(ipc_session, topic);
    }
}

static int dylora_mqtt_handle_recv_msg(void *obj, ipc_msg_t *mqtt_msg)
{
    dy_syslog(LOG_DEBUG, " MQTT client received MQTT topic:%s payload length:%d",
              mqtt_msg->topic, mqtt_msg->payloadLen);
    if (strstr(mqtt_msg->topic, TOPIC_SEND_RGLT_SIGNAL_RAW_DATA))
    {
        //发送下行数据
        dylora_data_downlink(mqtt_msg);
    }
    else if (strstr(mqtt_msg->topic, TOPIC_NOTIFY_UPGRADE))
    {
        if (strcmp(mqtt_msg->payload, START_UPGRADE) == 0)
        {
            dy_syslog(LOG_ERR, "receive start upgrade, stop period poll");
            service_lora_enable(false);
            service_common_enable(false);
        }
        else
        {
            dy_syslog(LOG_ERR, "receive stop upgrade, start period poll");
            service_lora_enable(true);
            service_common_enable(true);
        }
    }

}

// 建立与内部broker之间的MQTT连接
int dylora_mqtt_client_init()
{
    char clientId[MAX_CLIENT_ID_LEN] = {0};

    snprintf(clientId, MAX_CLIENT_ID_LEN, "INT_dylora_%s", var.sn_str);
    var.session = ipc_session_new(clientId, (void*)&var, IPC_DEFAULT);
    if (var.session == NULL) return -1;

    ipc_session_set_callbacks(var.session, dylora_mqtt_handle_recv_msg, NULL);
    dylora_subscribe_all();
    ipc_session_start(var.session);
}

static void *thread_launcher(void *opts)
{
    static int all_started = 0;
    exp_lora_msg_t tmp={0};
    while (!exit_sig) {
        time_t now = time(NULL);
        if(list_size(GENERAL_UPLINK_LIST) > 0)
        {
            int ret = msg_get_first(GENERAL_UPLINK_LIST, &tmp);
            if (ret)
            {
                exp_lora_msg_t delay_msg={0};
                ret = msg_get_compare(DELAY_MSG_LIST, tmp.addr,&delay_msg);
                if(ret)
                {
                    msg_add_tail(GENERAL_DOWNLINK_LIST, &delay_msg);
                }
                dylora_data_uplink(&tmp);
            }
            continue;
        }

        if(list_size(GENERAL_DOWNLINK_RESP_LIST) > 0)
        {
            int ret = msg_get_first(GENERAL_DOWNLINK_RESP_LIST, &tmp);
            if (ret)
            {
                dylora_data_downlink_response(&tmp,1);
            }
            continue;
        }

        if(now >=  var.poll_preiod_msg + 2)
        {
            char *sn = NULL;
            var.poll_preiod_msg = now;
            period_msg_t *msg = period_msg_next(var.period);
            if(msg && msg->rs)
            {
                int i = list_size(GENERAL_DOWNLINK_LIST);
                if(i >= DOWNLINK_LIMIT/2)
                {
                    dy_syslog(LOG_WARNING,"queue is full");
                    continue;
                }
                exp_lora_msg_t pram;
                char *p = (char *)msg->rs->data;
                p+=2;  //skip addr
                memcpy(&pram.code,p,1);p+=1;
                pram.len = msg->rs->len - 3;
                memcpy(&pram.data,p,pram.len);p+=pram.len;
                pram.retry = 1;
                pram.times = VALID_TIME_DEFAULT;
                if (msg->rs->port == LORA_DTU_RS485)
                    sn = msg->rs->dtu_sn;
                else
                    sn = msg->rs->sn;
                node_cfg_t* cfg = dylora_cfg_search_by_sn(sn);
                if(cfg)
                {
                    pram.addr.b = cfg->term_addr[1];
                }
                node_status_t * status = dylora_status_serach_by_addr(&pram.addr);
                if (status && status->online)
                {
                    char key_str[64] = {0};
                    my_regulate_cmd_backup_t fields;
                    fields.donwlink_time  = time(NULL);
                    strcpy(fields.src_identifier, msg->rs->src_identifier);
                    strcpy(fields.sn, sn);
                    fields.mi = msg->rs->mi;
                    sprintf(key_str, "dy_lora_%s_%X",sn,(pram.code & 0x7f));
                    var.identifier_backup.insert(&var.identifier_backup, key_str, &fields, sizeof(fields));
                    msg_add_tail(GENERAL_DOWNLINK_LIST, &pram);
                }
                else
                {
                    dy_syslog(LOG_INFO, "device [%s] was offline,discard this preiod data %s",sn,msg->key);
                }
            }
            if (all_started == 0)
            {
                all_started = period_services_retrigger(var.session, var.period, var.nodes_cfg_table, var.template_table, ports, ARRAY_SIZE(ports));
            }
        }

        if (service_flag &&  now >= var.nodes_status_sync_date + 30)
        {
            static time_t lasttime = 0;
            var.nodes_status_sync_date = now;
            char *node_status = ds_print(var.ds);

            dy_syslog(LOG_INFO, "online_status_changed:%d strlen(node_status):%d, (now-lasttime):%d",
                      var.ds->online_status_changed, strlen(node_status), now - lasttime);
            if (node_status && strlen(node_status) > 20 &&
                    (var.ds->online_status_changed || now > lasttime + 3600)) //在线状态变化上报，1小时也会上报
            {
                char topic[TOPIC_MAX_LEN];
                int ret = 0;

                snprintf(topic, TOPIC_MAX_LEN, "ipc/%s/%s/%s", var.sn_str, var.proc_name, TOPIC_EVT_NODESSTATUS);

                ret = ipc_session_publish(var.session, topic, node_status, strlen(node_status));
                if (ret == 0)
                {
                    lasttime = now;
                    var.ds->online_status_changed = false;
                }
                node_sta_backup(var.ds, LORA_NODE_STATUS_BAK_FILE);
            }
            if (node_status)
            {
                write_file_data(NODES_CACHE "/dy_lora_status.json", node_status, strlen(node_status));
                free(node_status);
            }
        }
        usleep(10000);
    }


    exit(EXIT_SUCCESS);
}

void *lora_cfg_get(void)
{
    return &var.lora_cfg;
}

//--------------------------------------------
int thread_common_start(void)
{
    pthread_t thread;
    const char *board_name = NULL;
    gw_port_e ports[] = {LORA_1};

    check_make_dir(NODES_CACHE);
    check_make_dir(NODES_CFG);
    get_board_sn(var.sn_str);
    get_process_name(var.proc_name);
    board_name = get_board_name();

    dy_syslog(LOG_INFO, "board name:%s", board_name);
    dy_syslog(LOG_INFO, "board SN:%s", var.sn_str);

    if (load_nodes_cfg(&var.nodes_cfg_table, NODES_CFG_PATH) == -1)
    {
        dy_syslog(LOG_ERR, "load nodes cfg fail");
    }
    if (load_templates_cfg(&var.template_table, TEMPLATES_CFG_PATH) == -1)
    {
        dy_syslog(LOG_ERR, "load template cfg fail");
    }
    dev_status_init(&var.ds, var.nodes_cfg_table, var.template_table, ports, ARRAY_SIZE(ports));
    node_sta_recovery(var.ds, LORA_NODE_STATUS_BAK_FILE);

    kv_array_init(&var.identifier_backup, 32);

    if(dylora_config() != 0)
        while(1) sleep(1);

    dylora_mqtt_client_init();
    period_msg_init(&var.period, 128);
    period_services_trigger(var.session, var.period, var.nodes_cfg_table, var.template_table, ports, ARRAY_SIZE(ports));
    thread_begin(thread_launcher, NULL, &thread);
    return 0;
}

//--------------------------------------------
void thread_common_stop(void)
{

}

//--------------------------------------------
void service_common_enable(bool enable)
{
    service_flag = enable;
}