#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <time.h>
#include <sys/socket.h>
#include <sys/un.h>
#include <poll.h>
#include <errno.h>

#include "dataserver_message.h"
#include "dataserver_client.h"

#define ds_inf(_fmt_,...) printf("I %s:%d "_fmt_"\n", __func__, __LINE__, ##__VA_ARGS__)
#define ds_err(_fmt_,...) printf("E %s:%d "_fmt_"\n", __func__, __LINE__, ##__VA_ARGS__)

#define ds_conn_inf(_fmt_,...) printf("I %s:%d conn/fd[%p/%d] "_fmt_"\n", __func__, __LINE__, conn, conn? conn->fd:-1, ##__VA_ARGS__)
#define ds_conn_err(_fmt_,...) printf("E %s:%d conn/fd[%p/%d] "_fmt_"\n", __func__, __LINE__, conn, conn? conn->fd:-1, ##__VA_ARGS__)

typedef struct {
    int magic;
    #define DS_CONN_MAGIC ((int)('d'<<24|'a'<<16)|'s'<<8|'r')

    int fd;
    int evt_fd;
} ds_connection_t;

static char *ds_server_path_default = "/run/ccdataserver.sock";

/*ret:
 * -1:bad param,
 * -2:timeout,
 * -3:need re-connect
 */
static int dataserver_read(ds_connection_t *conn,
        uint8_t *buf, size_t size, int timeout_ms)
{
    int left = size;
    int fd = conn->fd;
    int evt_fd = conn->evt_fd;

    for (;;) {
        struct pollfd fds[2];

        memset(fds, 0, sizeof(fds));
        fds[0].fd = fd;
        fds[0].events = POLLIN | POLLERR;
        fds[1].fd = evt_fd;
        fds[1].events = POLLIN | POLLERR;

        int ret = poll(fds, 2, timeout_ms);

        if (ret == 0)
            return -2;
        else if (ret < 0)
            return -3;

        ret = 0;

        if (!(fds[0].revents & POLLIN))
            return -3;

        {
            ret = read(fd, buf + (size - left), left);
            if (ret <= 0) {
                if (errno == EAGAIN)
                    continue;
                return -3;
            } else if (ret < left) {
                left -= ret;
            } else if (ret == left) {
                return size;
            }
        }
    }
    return size;
}

/*ret:
 * -1:bad param,
 * -2:timeout,
 * -3:need re-connect
 */
static int dataserver_write(ds_connection_t *conn, uint8_t *buf, size_t size)
{
    struct pollfd fds[1];
    fds[0].fd = conn->fd;
    fds[0].events = POLLOUT | POLLERR;

    int ret = poll(fds, 1, 1000);

    if (ret == 0)
        return -2;
    else if (ret < 0)
        return -3;

    ret = 0;

    if (fds[0].revents & POLLOUT) {
        int sent = 0;
        while (sent < size) {
            ret = write(conn->fd, buf + sent, size - sent);
            if (ret <= 0) {
                if (errno == EAGAIN)
                    continue;
                return -3;
            } else {
                sent += ret;
            }
        }
    }

    return size;
}


static uint32_t ds_time_now(void)
{
    struct timespec ts;
    clock_gettime(CLOCK_MONOTONIC, &ts);
    return ts.tv_sec * 1000 + ts.tv_nsec / 1000000;
}

ds_conn_t* dataserver_connect(void)
{
    ds_inf("connecting...");

    char *server_path = getenv("CCDATASERVER_SOCKET");
    if (!server_path) {
        server_path = ds_server_path_default;
        ds_inf("env CCDATASERVER_SOCKET not found, use default:\"%s\"", server_path);
    }

    int sockfd = -1;
    if ((sockfd = socket(AF_UNIX, SOCK_STREAM, 0)) < 0) {
        ds_err("open socket fail");
        return NULL;
    }

    struct sockaddr_un s_un;
    int len;

    memset(&s_un, 0, sizeof(s_un));
    s_un.sun_family = AF_UNIX;
    strncpy(s_un.sun_path, server_path, sizeof(s_un.sun_path) - 1);
    len = offsetof(struct sockaddr_un, sun_path) + strlen(s_un.sun_path);

    uint32_t start = ds_time_now();

    while (1) {
        if (connect(sockfd, (struct sockaddr *)&s_un, len) == 0)
            break;

        ds_err("connect fail, retry...");

        uint32_t now = ds_time_now();

        if (now - start > 1000) {
            close(sockfd);
            return NULL;
        }

        usleep(100 * 1000);
    }

    ds_connection_t *conn = calloc(1, sizeof(ds_connection_t));
    if (!conn) {
        ds_err("mem fail");
        close(sockfd);
        return NULL;
    }

    conn->magic = DS_CONN_MAGIC;
    conn->fd = sockfd;
    conn->evt_fd = eventfd(0, 0);

    ds_conn_inf("connected.");

    return (ds_conn_t *)conn;
}

/*ret:
 * -1:bad param,
 * -2:timeout,
 * -3:need re-connect
 */
int dataserver_read_message(ds_conn_t *connection, void *buf, size_t size, int timeout_ms)
{
    ds_connection_t *conn = (ds_connection_t *)connection;

    if (!VALID_DS_CONN(conn) || conn->magic != DS_CONN_MAGIC || (!buf && size) ) {
        ds_err("bad parameter");
        return -1;
    }

    if (!size)
        return 0;

    int count = 0;

    ds_msg_head_t msg;

    int ret = dataserver_read(conn, (uint8_t *)&msg, sizeof(ds_msg_head_t), timeout_ms);

    if (ret <= -3) {
        ds_conn_err("read msg head fail (%d)", ret);
        return ret;
    } else if (ret == -2)
        /*time out*/
        return ret;
    else if (ret <= 0)
        return ret;

    if (ret == sizeof(ds_msg_head_t)) {
        /*fill the header*/
        memcpy(buf, &msg, ret);
        count += ret;

        if (msg.size) {
            if ((msg.size + sizeof(ds_msg_head_t)) > size) {
                ds_conn_err("insufficient buffer, required[%zd] now[%zd]", msg.size + sizeof(ds_msg_head_t), size);
                return -3;
            }

            /*fill the body*/
            ret = dataserver_read(conn, (uint8_t *)buf + sizeof(ds_msg_head_t), msg.size, timeout_ms);
            if (ret == msg.size)
                count += ret;
            else {
                ds_conn_err("read msg body fail (size:%d), err(%d)", msg.size, ret);
            }
        }
    }

    return (ret < 0) ? ret : count;
}

/*ret:
 * -1:bad param,
 * -2:timeout,
 * -3:need re-connect
 */
int dataserver_write_message(ds_conn_t *connection, void *buf, size_t size)
{
    ds_connection_t *conn = (ds_connection_t *)connection;

    if (!VALID_DS_CONN(conn) || conn->magic != DS_CONN_MAGIC || (!buf && size != 0)) {
        ds_err("bad parameter");
        return -1;
    }

    if (size == 0)
        return 0;

    int ret = dataserver_write(conn, (uint8_t *)buf, size);

    return ret;
}

int dataserver_disconnect(ds_conn_t *connection)
{
    ds_connection_t *conn = (ds_connection_t *)connection;

    ds_conn_inf("disconnecting...");

    if (!VALID_DS_CONN(conn) || conn->magic != DS_CONN_MAGIC) {
        ds_err("bad parameter");
        return -1;
    }

    if (conn->fd >= 0) {
        int64_t pad = 1;
        write(conn->evt_fd, &pad, sizeof(pad));

        close(conn->fd);
        conn->fd = -1;

        if (conn->evt_fd >= 0) {
            close(conn->evt_fd);
            conn->evt_fd = -1;
        }

        conn->magic = 0;
        free(conn);
        conn = NULL;
    }

    ds_conn_inf("disconnected.");

    return 0;
}

int dataserver_destroy(ds_conn_t *connection)
{
    ds_connection_t *conn = (ds_connection_t *)connection;

    ds_conn_inf("destroying...");
/*
    if (!VALID_DS_CONN(conn) || conn->magic != DS_CONN_MAGIC) {
        ds_err("bad parameter");
        return -1;
    }
*/

    ds_inf("destroyed.");

    return 0;
}

#include <arpa/inet.h>

ds_conn_t* dataserver_connect_ip(uint32_t addr, uint32_t port)
{
    ds_inf("connecting...");

    int sockfd = -1;
    if ((sockfd = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
        ds_err("open socket fail");
        return NULL;
    }

    struct sockaddr_in s_in;

    memset(&s_in, 0, sizeof(s_in));
    s_in.sin_family = AF_INET;
    s_in.sin_port = htons(port);
    s_in.sin_addr.s_addr = addr;

    uint32_t start = ds_time_now();

    while (1) {
        if (connect(sockfd, (struct sockaddr *)&s_in, sizeof(s_in)) == 0)
            break;

        ds_err("connect fail, retry...");

        uint32_t now = ds_time_now();

        if (now - start > 1000) {
            close(sockfd);
            return NULL;
        }

        usleep(100 * 1000);
    }

#ifdef BUILD_FOR_ANDROID
    {
        int val = 1;
        errno = 0;
        if (setsockopt(sockfd, SOL_SOCKET, SO_KEEPALIVE, &val, sizeof(val))) {
            ds_err("setsockopt SO_KEEPALIVE failed. err(%d/%s)", errno, strerror(errno));
        }
            ds_err("setsockopt SO_KEEPALIVE failed. err(%d/%s)", errno, strerror(errno));
        int keepidle = 20;
        int keepinterval = 3;
        setsockopt(sockfd, SOL_TCP, TCP_KEEPIDLE, &keepidle, sizeof(keepidle));
        setsockopt(sockfd, SOL_TCP, TCP_KEEPINTVL, &keepinterval, sizeof(keepinterval));
    }
#endif

    ds_connection_t *conn = calloc(1, sizeof(ds_connection_t));
    if (!conn) {
        ds_err("mem fail");
        close(sockfd);
        return NULL;
    }

    conn->magic = DS_CONN_MAGIC;
    conn->fd = sockfd;
    conn->evt_fd = eventfd(0, 0);

    ds_conn_inf("connected.");

    return (ds_conn_t *)conn;
}