#include <stdio.h>
#include <errno.h>
#include <assert.h>
#include <unistd.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <MESA/MESA_prof_load.h>

#include "log.h"
#include "utils.h"
#include "sf_metrics.h"

#define SCE_SF_METRICS "SCE-SF-METRICS,rule_id=%d,sff_profile_id=%d,sf_profile_id=%d,type=service_chaining_metrics sent_pkts=%lu,sent_bytes=%lu,recv_pkts=%lu,recv_bytes=%lu"

struct key_tuple
{
    int rule_id;
    int sff_profile_id;
    int sf_profile_id;
};

struct node
{
    struct key_tuple key;
    uint64_t sent_pkts;
    uint64_t sent_bytes;
    uint64_t recv_pkts;
    uint64_t recv_bytes;

    UT_hash_handle hh;
};

struct sf_metrics_config
{
    int enable;
    int interval_s;
    int telegraf_listen_port;
    char telegraf_bind_address[2048];
};

struct sf_metrics
{
    struct sf_metrics_config config;
    struct sockaddr_in sock_addr;
    int sockfd;

    struct node *htable;
    uint64_t htable_elem_count;
};

static void sf_metrics_parse_config(const char *profile, struct sf_metrics_config *config)
{
    MESA_load_profile_int_def(profile, "METRICS", "enable", &(config->enable), 1);
    MESA_load_profile_int_def(profile, "METRICS", "interval_s", &(config->interval_s), 1);
    MESA_load_profile_int_def(profile, "METRICS", "telegraf_listen_port", &(config->telegraf_listen_port), 8300);
    MESA_load_profile_string_def(profile, "METRICS", "telegraf_bind_address", config->telegraf_bind_address, sizeof(config->telegraf_bind_address), "127.0.0.1");

    LOG_DEBUG("%s: METRICS->enable                 : %d", LOG_TAG_SF_METRICS, config->enable);
    LOG_DEBUG("%s: METRICS->interval_s             : %d", LOG_TAG_SF_METRICS, config->interval_s);
    LOG_DEBUG("%s: METRICS->telegraf_listen_port   : %d", LOG_TAG_SF_METRICS, config->telegraf_listen_port);
    LOG_DEBUG("%s: METRICS->telegraf_bind_address  : %s", LOG_TAG_SF_METRICS, config->telegraf_bind_address);
}

struct sf_metrics *sf_metrics_create(const char *profile)
{
    struct sf_metrics *handle = (struct sf_metrics *)calloc(1, sizeof(struct sf_metrics));
    assert(handle);
    sf_metrics_parse_config(profile, &(handle->config));

    if (handle->config.enable == 0)
    {
        return handle;
    }

    handle->sockfd = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
    handle->sock_addr.sin_family = AF_INET;
    handle->sock_addr.sin_port = htons(handle->config.telegraf_listen_port);
    handle->sock_addr.sin_addr.s_addr = inet_addr(handle->config.telegraf_bind_address);
    handle->htable_elem_count = 0;
    if (handle->sockfd == -1)
    {
        LOG_ERROR("%s: failed to create udp sockfd %s:%d, errno: %d, %s", LOG_TAG_SF_METRICS, handle->config.telegraf_bind_address, handle->config.telegraf_listen_port, errno, strerror(errno));
        sf_metrics_destory(handle);
        return NULL;
    }

    return handle;
}

void sf_metrics_destory(struct sf_metrics *handle)
{
    if (handle)
    {
        if (handle->sockfd)
        {
            close(handle->sockfd);
            handle->sockfd = -1;
        }

        struct node *temp = NULL;
        struct node *node = NULL;
        HASH_ITER(hh, handle->htable, node, temp)
        {
            HASH_DELETE(hh, handle->htable, node);

            free(node);
            node = NULL;
        }

        handle->htable_elem_count = 0;
        free(handle);
        handle = NULL;
    }
}

void sf_metrics_reset(struct sf_metrics *handle)
{
    if (handle == NULL)
    {
        return;
    }

    if (handle->config.enable == 0)
    {
        return;
    }

    struct node *temp = NULL;
    struct node *node = NULL;
    HASH_ITER(hh, handle->htable, node, temp)
    {
        HASH_DELETE(hh, handle->htable, node);

        free(node);
        node = NULL;
        handle->htable_elem_count--;
    }
}

void sf_metrics_inc(struct sf_metrics *handle, int rule_id, int sff_profile_id, int sf_profile_id, uint64_t rx_pkts, uint64_t rx_bytes, uint64_t tx_pkts, uint64_t tx_bytes)
{
    if (handle->config.enable == 0)
    {
        return;
    }

    struct key_tuple key;
    memset(&key, 0, sizeof(struct key_tuple));
    key.rule_id = rule_id;
    key.sff_profile_id = sff_profile_id;
    key.sf_profile_id = sf_profile_id;

    struct node *temp = NULL;
    HASH_FIND(hh, handle->htable, &key, sizeof(struct key_tuple), temp);
    if (temp)
    {
        temp->recv_pkts += rx_pkts;
        temp->recv_bytes += rx_bytes;
        temp->sent_pkts += tx_pkts;
        temp->sent_bytes += tx_bytes;
    }
    else
    {
        temp = (struct node *)calloc(1, sizeof(struct node));
        temp->key.rule_id = rule_id;
        temp->key.sff_profile_id = sff_profile_id;
        temp->key.sf_profile_id = sf_profile_id;
        temp->recv_pkts = rx_pkts;
        temp->recv_bytes = rx_bytes;
        temp->sent_pkts = tx_pkts;
        temp->sent_bytes = tx_bytes;

        HASH_ADD(hh, handle->htable, key, sizeof(struct key_tuple), temp);
    }
}

void sf_metrics_send(struct sf_metrics *handle)
{
    char buff[2048];
    int nsend = 0;
    int size = sizeof(buff);

    struct node *temp = NULL;
    struct node *node = NULL;

    if (handle->config.enable == 0)
    {
        return;
    }

    HASH_ITER(hh, handle->htable, node, temp)
    {
        if (node->sent_pkts == 0 && node->recv_pkts == 0)
        {
            continue;
        }

        memset(buff, 0, size);
        nsend = snprintf(buff, size, SCE_SF_METRICS,
                         node->key.rule_id,
                         node->key.sff_profile_id,
                         node->key.sf_profile_id,
                         node->sent_pkts,
                         node->sent_bytes,
                         node->recv_pkts,
                         node->recv_bytes);
        sendto(handle->sockfd, buff, nsend, 0, (struct sockaddr *)&handle->sock_addr, sizeof(handle->sock_addr));
    }
}

int sf_metrics_get_interval(struct sf_metrics *handle)
{
    return handle->config.interval_s;
}