#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 "opc_ua_common.h"

static int opc_ua_data_report(opc_ua_var_t *var)
{
    TIMER_CONFIRM(var->report_timer);
    time_t now = time(NULL);
    connect_config_t *conn = NULL;
    connect_config_t *tmp = NULL;
    dev_node_t *node = NULL;
    dev_node_t *tmp2 = NULL;
    int i;
    char buf[4096] = {0};
    char topic[TOPIC_MAX_LEN] = {0};

    list_for_each_entry_safe(conn, tmp, &var->connect_list, list)
    {
        pthread_rwlock_rdlock(&conn->data_lock);
        list_for_each_entry_safe(node, tmp2, &conn->node_list, list2)
        {
            cJSON *data = cJSON_CreateObject();
            for (i = 0; i < node->tag_cnt; i++)
            {
                opc_ua_tag_t *tag = &node->tag[i];
                if (tag->updated)
                {
                    tag->updated = false;
                    if (UA_Variant_isScalar(&tag->val))
                    {
                        if (tag->val.type == &UA_TYPES[UA_TYPES_DOUBLE])
                        {
                            UA_Double val = *(UA_Double *)tag->val.data;
                            cJSON_AddNumberToObject(data, tag->perperty->identifier, val);
                        }
                        if (tag->val.type == &UA_TYPES[UA_TYPES_INT32])
                        {
                            UA_Int32 val = *(UA_Int32 *)tag->val.data;
                            cJSON_AddNumberToObject(data, tag->perperty->identifier, val);
                        }
                        else if (tag->val.type == &UA_TYPES[UA_TYPES_UINT32])
                        {
                            UA_UInt32 val = *(UA_UInt32 *)tag->val.data;
                            cJSON_AddNumberToObject(data, tag->perperty->identifier, val);
                        }
                        else if (tag->val.type == &UA_TYPES[UA_TYPES_STRING])
                        {
                            UA_String *val = (UA_String *)tag->val.data;
                            cJSON_AddStringToObject(data, tag->perperty->identifier, val->data);
                        }
                        else if (tag->val.type == &UA_TYPES[UA_TYPES_BOOLEAN])
                        {
                            UA_Boolean val = *(UA_Boolean *)tag->val.data;
                            cJSON_AddBoolToObject(data, tag->perperty->identifier, val);
                        }
                        else if (tag->val.type == &UA_TYPES[UA_TYPES_INT16])
                        {
                            UA_Int16 val = *(UA_Int16 *)tag->val.data;
                            cJSON_AddNumberToObject(data, tag->perperty->identifier, val);
                        }
                        else if (tag->val.type == &UA_TYPES[UA_TYPES_UINT16])
                        {
                            UA_UInt16 val = *(UA_UInt16 *)tag->val.data;
                            cJSON_AddNumberToObject(data, tag->perperty->identifier, val);
                        }
                        else if (tag->val.type == &UA_TYPES[UA_TYPES_FLOAT])
                        {
                            UA_Float val = *(UA_Float *)tag->val.data;
                            cJSON_AddNumberToObject(data, tag->perperty->identifier, val);
                        }
                        else if (tag->val.type == &UA_TYPES[UA_TYPES_INT64])
                        {
                            UA_Int64 val = *(UA_Int64 *)tag->val.data;
                            cJSON_AddNumberToObject(data, tag->perperty->identifier, val);
                        }
                        else if (tag->val.type == &UA_TYPES[UA_TYPES_UINT64])
                        {
                            UA_UInt64 val = *(UA_UInt64 *)tag->val.data;
                            cJSON_AddNumberToObject(data, tag->perperty->identifier, val);
                        }
                    }
                }
            }

            char *tmp = cJSON_Print(data);
            if (strlen(tmp) > 5)
            {
                cJSON *rt_data = cJSON_CreateObject();

                b64_encode(tmp, strlen(tmp), buf, sizeof(buf));
                cJSON_AddStringToObject(rt_data, "port", "OPC_UA");
                cJSON_AddStringToObject(rt_data, "sn", node->sn);
                cJSON_AddNumberToObject(rt_data, "ts", now);
                cJSON_AddNumberToObject(rt_data, "len", strlen(tmp));
                cJSON_AddStringToObject(rt_data, "data_b64", buf);
                char *data_tmp = cJSON_Print(rt_data);
                snprintf(topic, TOPIC_MAX_LEN, "ipc/%s/%s/device/%s/data/%s", var->sn_str, "OPC_UA",
                         node->sn, "raw_data");
                dy_syslog(LOG_DEBUG, "public topic:%s", topic);
                ipc_session_publish(var->session, topic, data_tmp, strlen(data_tmp));
                free(data_tmp);

                cJSON_Delete(rt_data);
            }
            free(tmp);
            cJSON_Delete(data);
        }
        pthread_rwlock_unlock(&conn->data_lock);
    }
}

static int ds_flush_timer(opc_ua_var_t *var)
{
    time_t now = time(NULL);

    TIMER_CONFIRM(var->node_status_timer);    

    if (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, OPC_UA_NODE_STATUS_BAK_FILE);
        }
        if (node_status)
        {
            write_file_data(NODES_CACHE "/opc_ua_status.json", node_status, strlen(node_status));
            free(node_status);
        }
    }

    return 0;
}

static void opc_ua_loop(opc_ua_var_t *var)
{
    int ret = -1, maxfd, i;
    fd_set rset;
    struct timeval timeout;

    while (1)
    {
        SELECT_INIT();
        SELECT_ADD_FD(var->report_timer);
        SELECT_ADD_FD(var->node_status_timer);

        timeout.tv_usec = 0;
        timeout.tv_sec = 5;

        ret = select(maxfd + 1, &rset, 0, 0, &timeout);
        if (ret < 0)
        {
            int error = errno;
            dy_syslog(LOG_INFO, "errno %d\n", error);
            if (error == EINTR)
            {
                continue;
            }
            else
            {
                break;
            }
        }
        else if (ret > 0)
        {
            if (var->report_timer > 0 && FD_ISSET(var->report_timer, &rset))
            {
                FD_CLR(var->report_timer, &rset);
                opc_ua_data_report(var);
            }
            if (var->node_status_timer > 0 && FD_ISSET(var->node_status_timer, &rset))
            {
                FD_CLR(var->node_status_timer, &rset);
                ds_flush_timer(var);
            }
        }
    }
}

static void opc_ua_subscribe_all(opc_ua_var_t *var)
{
    ipc_session_t *ipc_session = var->session;
    char topic[TOPIC_MAX_LEN] = {0};
    int i = 0;

    snprintf(topic, TOPIC_MAX_LEN, "ipc/+/%s/device/+/data/%s", OPC_UA_APP_KEY, TOPIC_SEND_RGLT_SIGNAL_RAW_DATA);
    ipc_session_subscribe(ipc_session, topic);
    snprintf(topic, TOPIC_MAX_LEN, "ipc/+/%s/%s", OPC_UA_APP_KEY, TOPIC_NOTIFY_UPGRADE);
    ipc_session_subscribe(ipc_session, topic);
}

static int opc_ua_mqtt_handle_recv_msg(void *obj, ipc_msg_t *mqtt_msg)
{
    opc_ua_var_t *var = (opc_ua_var_t *)obj;
    dy_syslog(LOG_DEBUG, "received MQTT topic:%s payload length:%d", mqtt_msg->topic, mqtt_msg->payloadLen);
    if (strstr(mqtt_msg->topic, TOPIC_SEND_RGLT_SIGNAL_RAW_DATA))
    {
        //发送下行数据
        // opc_ua_msg_tag_data_ctrl(var, mqtt_msg);
    }
}

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

    snprintf(clientId, MAX_CLIENT_ID_LEN, "INT_opc_ua_%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, opc_ua_mqtt_handle_recv_msg, NULL);
    opc_ua_subscribe_all(var);
    ipc_session_start(var->session);
}

static int opc_ua_load_cfg(opc_ua_var_t *var)
{
    int i = 0;
    connect_config_t *connect_cfg = NULL;
    connect_config_t *tmp = NULL;
    dev_node_t *node = NULL;

    for (i = 0; i < var->nodes_cfg_table->node_cnt; i++)
    {
        if (strcmp(var->nodes_cfg_table->node[i].app_key, OPC_UA_APP_KEY) == 0)
        {
            int find = 0;
            char url[128] = {0};
            node_cfg_t *node_cfg = &var->nodes_cfg_table->node[i];
            char *username = NULL;
            char *password = NULL;
            time_t now = time(NULL);

            cJSON *ext = cJSON_Parse(node_cfg->ext_data);
            if (ext != NULL)
            {
                GET_JSON_VALUE_STRING(ext, "url", url);
                GET_JSON_VALUE_DY_STRING(ext, "username", username);
                GET_JSON_VALUE_DY_STRING(ext, "password", password);
                cJSON_Delete(ext);
            }
            if (strlen(url) == 0 &&
                strlen(node_cfg->tcp_ip_addr) > 5 && node_cfg->tcp_port > 0)
            {
                // use tcp_ip_addr and port if no url in ext data
                sprintf(url, "opc.tcp://%s:%d", node_cfg->tcp_ip_addr, node_cfg->tcp_port);
            }
            if (strlen(url) == 0)
            {
                dy_syslog(LOG_ERR, "url is empty");
                continue;
            }

            list_for_each_entry_safe(connect_cfg, tmp, &var->connect_list, list)
            {
                if (strcmp(url, connect_cfg->url) == 0)
                {
                    find = 1;
                    break;
                }
            }

            if (find == 0)
            {
                UA_StatusCode retval;

                connect_cfg = calloc(1, sizeof(connect_config_t));

                INIT_LIST_HEAD(&connect_cfg->node_list);
                INIT_LIST_HEAD(&connect_cfg->used_node_list);
                connect_cfg->var = var;
                connect_cfg->url = strdup(url);
                connect_cfg->username = username;
                connect_cfg->password = password;
                connect_cfg->browsered = false;
                list_add_tail(&connect_cfg->list, &var->connect_list);
                pthread_rwlock_init(&connect_cfg->data_lock, NULL);
                var->conn_cnt++;

                connect_cfg->client = UA_Client_new();
                UA_ClientConfig *cc = UA_Client_getConfig(connect_cfg->client);

                cc->inactivityCallback = inactivityCallback; /* Set stateCallback */
                cc->connectivityCheckInterval = 2000;        /* Perform a connectivity check every 2 seconds */
                UA_ClientConfig_setDefault(cc);
            }

            node = calloc(1, sizeof(dev_node_t));
            node->var = var;
            strncpy(node->sn, node_cfg->sn, sizeof(node->sn));
            node->connect_config = connect_cfg;
            node->node_cfg = node_cfg;
            list_add_tail(&node->list, &var->node_list);
            list_add_tail(&node->list2, &connect_cfg->node_list);
            var->node_cnt++;

            object_cfg_t *pobject = find_object_by_id(var->object_table, node_cfg->product_key);
            if (pobject)
            {
                node->property_tab = pobject->property_tab;
                int j = 0;
                node->tag_cnt = node->property_tab->propertyCnt;
                if (node->tag_cnt > 0)
                {
                    node->tag = calloc(node->tag_cnt, sizeof(opc_ua_tag_t));

                    for (j = 0; j < node->property_tab->propertyCnt; j++)
                    {
                        node->tag[j].type = NODE_ID_NEED_BROWSER;

                        if (node->property_tab->property[j].ext_data != NULL)
                        {
                            cJSON *ext = cJSON_Parse(node->property_tab->property[j].ext_data);
                            if (ext != NULL)
                            {
                                /*
                                    {"type":"string","namespace":2,"identifier":"a.b.c.d"}
                                */
                                char type[16];

                                GET_JSON_VALUE_STRING(ext, "type", type);
                                if (strcmp(type, "string") == 0)
                                {
                                    int namespace_id;
                                    char *identifier = NULL;

                                    GET_JSON_VALUE_INT(ext, "namespace", namespace_id);
                                    GET_JSON_VALUE_DY_STRING(ext, "identifier", identifier);
                                    node->tag[j].nodeId = UA_NODEID_STRING(namespace_id, identifier);
                                    node->tag[j].type = NODE_ID_USER_DEFINE;
                                }
                                else if (strcmp(type, "int") == 0)
                                {
                                    int namespace_id;
                                    int numberic;

                                    GET_JSON_VALUE_INT(ext, "namespace", namespace_id);
                                    GET_JSON_VALUE_INT(ext, "identifier", numberic);
                                    node->tag[j].nodeId = UA_NODEID_NUMERIC(namespace_id, numberic);
                                    node->tag[j].type = NODE_ID_USER_DEFINE;
                                }
                                else if (strcmp(type, "search") == 0)
                                {
                                    // 通过browser搜索
                                    char *display_name = NULL;
                                    GET_JSON_VALUE_DY_STRING(ext, "display_name", display_name);
                                    node->tag[j].display_name = display_name;
                                    node->tag[j].type = NODE_ID_NEED_BROWSER;
                                }
                                else if (strcmp(type, "none") == 0)
                                {
                                    node->tag[j].type = NODE_ID_NOT_UA;
                                }
                                cJSON_Delete(ext);
                            }
                            else
                            {
                                dy_syslog(LOG_DEBUG, "not valid json");
                            }
                        }
                        else
                        {
                            node->tag[j].display_name = node->property_tab->property[j].identifier;
                            node->tag[j].type = NODE_ID_NEED_BROWSER;
                        }
                        if (node->tag[j].type == NODE_ID_NEED_BROWSER)
                        {
                            if (node->tag[j].display_name == NULL)
                            {
                                node->tag[j].display_name = node->property_tab->property[j].identifier;
                            }
                            connect_cfg->browsered = false;
                        }
                        node->tag[j].perperty = &node->property_tab->property[j];
                    }
                }
            }
            else
            {
                dy_syslog(LOG_WARNING, "not find object %s\n", node_cfg->product_key);
            }
        }
    }
}

static int opc_ua_init(opc_ua_var_t *var)
{
    int i, j, k;

    check_make_dir(NODES_CACHE);
    check_make_dir(NODES_CFG);
    get_board_sn(var->sn_str);
    get_process_name(var->proc_name);

    INIT_LIST_HEAD(&var->connect_list);
    INIT_LIST_HEAD(&var->node_list);

    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");
    }
    if (load_objects_cfg(&var->object_table, OBJECTS_CFG_PATH) == -1)
    {
        dy_syslog(LOG_ERR, "load object cfg fail");
    }

    dev_status_init_by_app_key(&var->ds, var->nodes_cfg_table, var->template_table, OPC_UA_APP_KEY);
    node_sta_recovery(var->ds, OPC_UA_NODE_STATUS_BAK_FILE);

    opc_ua_load_cfg(var);

    var->report_timer = my_timer_create();
    if (var->report_timer > 0)
    {
        my_timer_set(var->report_timer, 1, 1000);
    }
    var->node_status_timer = my_timer_create();
    if (var->node_status_timer > 0)
    {
        my_timer_set(var->node_status_timer, 1, 5000);
    }

    opc_ua_mqtt_client_init(var);

    connect_config_t *connect_cfg = NULL;
    connect_config_t *tmp = NULL;
    list_for_each_entry_safe(connect_cfg, tmp, &var->connect_list, list)
    {
        opc_ua_start_thread(var, connect_cfg);
    }

    return 0;
}

int main(int argc, char *argv[])
{
    opc_ua_var_t var = {0};

    openlog("OPC-UA", LOG_PID, LOG_DAEMON);

    memset(&var, 0, sizeof(var));

    opc_ua_init(&var);
    opc_ua_loop(&var);

    return 0;
}