#include <netinet/in.h>
#include <stdatomic.h>
#include <stdio.h>
#include <string.h>
#include <sys/epoll.h>
// #include "../common.h"
#include <bits/types/struct_timeval.h>
#include <pthread.h>
#include <sys/epoll.h>
#include <sys/select.h>
#include <sys/socket.h>
#include <sys/syslog.h>
#include <unistd.h>
#include "frpc_proxy.h"
#include "tcp_server.h"
#include "../proto_forward.h"
#include "collector-api.h"
#include "define.h"

extern void channel_lock(channel_t* channel);
extern void channel_unlock(channel_t* channel);
extern int  read_data_timeout(int fd, unsigned char* buff, int buffsize, int timeout);

static void dev_set_channel_lock(device_t* dev, int lock)
{
    char nod[DEV_NAME_LEN + TAG_NAME_LEN] = {0};
    snprintf(nod, DEV_NAME_LEN + TAG_NAME_LEN, "%s.%s", dev->no, STATE_CHANNELLOCK);
    proto_forward_t *var = get_proto_forward_var();

    tag_t *tag = NULL;
    if (hash_intptr_findptr(var->tag_map_hash, nod, strlen(nod), (void **)&tag) < 0)
    {
        proto_syslog(LOG_INFO, "error:not find read tag :%s \n",nod);
        return;
    }
    if (tag != NULL)
    {
        tag->last_read_cache.to_int = tag->read_cache.to_int;
        tag->last_read_time = tag->read_cache_time;
        tag->read_cache.to_int = lock;
        tag->read_cache_time = clock_get_ms();
        reflush_data_up(tag);
    }
}


void change_channel_lock(channel_t* channel)
{
    for (int i = 0; i < channel->devs_size; i++)
    {
        dev_set_channel_lock(channel->devs[i], 1);
    }
     channel_lock(channel);
}

void change_channel_unlock(channel_t* channel)
{
    for (int i = 0; i < channel->devs_size; i++)
    {
        dev_set_channel_lock(channel->devs[i], 0);
    }
    channel_unlock(channel);
}

static int accept_with_timeout(int listen_fd, struct timeval* timeout)
{
    fd_set         fdset;
    int            result;
    struct timeval tv;

    FD_ZERO(&fdset);
    FD_SET(listen_fd, &fdset);

    // 复制超时时间结构体
    tv = *timeout;

    result = select(listen_fd + 1, &fdset, NULL, NULL, &tv);
    if (result < 0)
    {
        ems_syslog(LOG_NOTICE, "select error: %s", strerror(errno));
        return -1; // select 错误
    }
    else if (result == 0)
    {
        return -2; // 超时
    }
    else
    {
        if (FD_ISSET(listen_fd, &fdset))
        {
            return accept(listen_fd, NULL, NULL); // 正常接受连接
        }
    }
    return -3; // 其他错误情况
}

int write_bin_data(char* buffer, int len, FILE* log, bool tx)
{
    if (log == NULL) return 0;

    char* tmp = malloc(len * 3 + 256);
    int   l   = 0;
    if (tmp != NULL)
    {
        l = sprintf(tmp, "%s %d bytes:\n", tx ? "TX: " : "RX: ", len);
        for (int i = 0; i < len; i++)
        {
            l += sprintf(tmp + l, "%02X ", buffer[i]);
        }
        l += sprintf(tmp + l, "\n");
        fwrite(tmp, l, 1, log);
        free(tmp);
    }
    return l;
}

int write_and_read_dev_no(tcp_trans_t* cfg, char* buffer, int len, FILE* log);
// 流模式 只支持一个连接,报文完全透传
static void* tcp_server_stream_task(void* arg)
{
    tcp_trans_t* tcp = (tcp_trans_t*)arg;
    if (tcp == NULL || tcp->sfd < 0)
    {
        ems_syslog(LOG_ERR, "proxy_server cfg error");
        goto END;
    }

    struct timeval tv = {
        .tv_sec  = tcp->timeout / 1000,
        .tv_usec = tcp->timeout % 1000 * 1000};

    char buffer[4096] = {0};

    tcp->epfd = epoll_create1(0);
    if (tcp->epfd == -1)
    {
        ems_syslog(LOG_ERR, "epoll_create1: %s", strerror(errno));
        goto END;
    }

    struct epoll_event ev, events[10];
    ev.events  = EPOLLIN;
    ev.data.fd = tcp->chan_pr->s;
    if (epoll_ctl(tcp->epfd, EPOLL_CTL_ADD, tcp->chan_pr->s, &ev) == -1)
    {
        ems_syslog(LOG_ERR, "epoll_ctl: %s", strerror(errno));
        close(tcp->epfd);
        goto END;
    }

    int cfd = -1;
    while (atomic_load(&tcp->state) < FRPC_SERVER_STOPING)
    {
        cfd = accept_with_timeout(tcp->sfd, &tv);
        if (cfd < 0)
        {
            continue;
        }
        ev.events  = EPOLLIN | EPOLLET;
        ev.data.fd = cfd;
        if (epoll_ctl(tcp->epfd, EPOLL_CTL_ADD, cfd, &ev) == -1)
        {
            ems_syslog(LOG_ERR, "epoll_ctl: %s", strerror(errno));
            close(tcp->epfd);
            cfd = -1;
            continue;
        }

        change_channel_lock(tcp->chan_pr);
        while (atomic_load(&tcp->state) < FRPC_SERVER_STOPING && cfd > 0)
        {
            int nfds = epoll_wait(tcp->epfd, events, 10, 1000);
            if (nfds == -1)
            {
                ems_syslog(LOG_NOTICE, "epoll_wait: %s", strerror(errno));
                continue;
            }
            for (int n = 0; n < nfds; ++n)
            {
                int fd  = events[n].data.fd;
                int len = read(fd, buffer, sizeof(buffer));
                if (len < 0)
                {
                    ems_syslog(LOG_NOTICE, "read error: %s", strerror(errno));
                    close(fd);
                }
                else if (len == 0)
                {
                    ems_syslog(LOG_INFO, "client close!");
                    close(fd);
                    if (fd == cfd)
                    {
                        cfd = -1;
                    }
                }
                else
                {
                    int l = write(fd == tcp->chan_pr->s ? cfd : tcp->chan_pr->s, buffer, len);
                    if (l > 0)
                    {
                        write_bin_data(buffer, len, tcp->log_fp, fd == tcp->chan_pr->s);
                    }
                }
            }
        }

        change_channel_unlock(tcp->chan_pr);
        if (cfd > 0)
        {
            close(cfd);
            cfd = -1;
        }
    }
END:
    if (tcp->sfd > 0)
    {
        close(tcp->sfd);
        tcp->sfd = -1;
    }
    if (tcp->epfd > 0)
    {
        close(tcp->epfd);
        tcp->epfd = -1;
    }
    atomic_store(&tcp->state, FRPC_SERVER_STOPED);
    return NULL;
}

static int tcp_client_setup_local(tcp_trans_t* cfg, int timeout_sec)
{
    if (cfg->local_ip == NULL || cfg->local_ip[0] == '\0')
    {
        ems_syslog(LOG_ERR, "Local IP is empty\n");
        return -1;
    }
    int cfd = -1;
    int flags = -1;

    struct sockaddr_in addr;
    addr.sin_family = AF_INET;
    addr.sin_port   = htons(cfg->local_port);
    if (inet_pton(AF_INET, cfg->local_ip, &addr.sin_addr) <= 0)
    {
        ems_syslog(LOG_ERR, "Invalid local IP address: %s\n", cfg->local_ip);
        goto ERROR;
    }

    cfd = socket(AF_INET, SOCK_STREAM, 0);
    if (cfd < 0)
    {
        ems_syslog(LOG_ERR, "socket creation failed:%s", strerror(errno));
        goto ERROR;
    }
    // 设置为非阻塞模式
    flags = fcntl(cfd, F_GETFL, 0);
    if (flags < 0 || fcntl(cfd, F_SETFL, flags | O_NONBLOCK) < 0)
    {
        ems_syslog(LOG_ERR, "fcntl failed to set non-blocking mode: %s", strerror(errno));
        goto ERROR;
    }

    // 尝试连接
    if (connect(cfd, (struct sockaddr*)&addr, sizeof(addr)) < 0)
    {
        if (errno != EINPROGRESS)
        {
            ems_syslog(LOG_ERR, "connect failed: %s", strerror(errno));
            goto ERROR;
        }

        // 使用 select 等待连接完成或超时
        fd_set writefds;
        FD_ZERO(&writefds);
        FD_SET(cfd, &writefds);

        struct timeval timeout;
        timeout.tv_sec = timeout_sec;
        timeout.tv_usec = 0;

        int ret = select(cfd + 1, NULL, &writefds, NULL, &timeout);
        if (ret < 0)
        {
            ems_syslog(LOG_ERR, "select failed: %s", strerror(errno));
            goto ERROR;
        }
        else if (ret == 0)
        {
            ems_syslog(LOG_ERR, "connect timed out: %s", strerror(errno));
            goto ERROR;
        }

        // 检查连接是否成功
        int error;
        socklen_t len = sizeof(error);
        if (getsockopt(cfd, SOL_SOCKET, SO_ERROR, &error, &len) < 0 || error != 0)
        {
            ems_syslog(LOG_ERR,  "connect failed: %s\n", strerror(error));
            goto ERROR;
        }
    }

    return cfd;

ERROR :
    if (cfd > 0) close(cfd);
    return -1;
}

static int udp_client_setup_local(tcp_trans_t* cfg)
{
    if (cfg->local_ip == NULL || cfg->local_ip[0] == '\0')
    {
        ems_syslog(LOG_ERR, "Local IP is empty\n");
        return -1;
    }

    struct sockaddr_in addr;
    addr.sin_family = AF_INET;
    addr.sin_port = htons(cfg->local_port);
    if (inet_pton(AF_INET, cfg->local_ip, &addr.sin_addr) <= 0)
    {
        ems_syslog(LOG_ERR, "Invalid local IP address: %s\n", cfg->local_ip);
        return -1;
    }

    int cfd = socket(AF_INET, SOCK_DGRAM, 0);
    if (cfd < 0)
    {
        ems_syslog(LOG_ERR,"socket creation failed: %s", strerror(errno));
        return -1;
    }

    // 设置为非阻塞模式
    int flags = fcntl(cfd, F_GETFL, 0);
    if (flags < 0 || fcntl(cfd, F_SETFL, flags | O_NONBLOCK) < 0)
    {
        ems_syslog(LOG_ERR,"fcntl failed to set non-blocking mode: %s", strerror(errno));
        close(cfd);
        return -1;
    }

    return cfd;
}

static void* tcp_server_thirdport_task(void* arg)
{
    tcp_trans_t* tcp = (tcp_trans_t*)arg;
    int local_cfd = -1;

    if (tcp == NULL || tcp->sfd < 0)
    {
        ems_syslog(LOG_ERR, "proxy_server cfg error");
        goto END;
    }

    struct timeval tv = {
        .tv_sec  = tcp->timeout,
        .tv_usec = 0};

    char buffer[4096] = {0};

    tcp->epfd = epoll_create1(0);
    if (tcp->epfd == -1)
    {
        ems_syslog(LOG_ERR, "epoll_create1: %s", strerror(errno));
        goto END;
    }

    if(tcp->local_type == FRPC_TYPE_TCP)
    {
        local_cfd = tcp_client_setup_local(tcp, 10); // 10s
    }
    else if(tcp->local_type == FRPC_TYPE_UDP)
    {
        local_cfd = udp_client_setup_local(tcp);
    }

    struct epoll_event ev, events[10];
    ev.events  = EPOLLIN;
    ev.data.fd = local_cfd;
    if (epoll_ctl(tcp->epfd, EPOLL_CTL_ADD, local_cfd, &ev) == -1)
    {
        ems_syslog(LOG_ERR, "epoll_ctl: %s", strerror(errno));
        close(tcp->epfd);
        goto END;
    }

    int cfd = -1;
    while (atomic_load(&tcp->state) < FRPC_SERVER_STOPING)
    {
        cfd = accept_with_timeout(tcp->sfd, &tv);
        if (cfd < 0)
        {
            continue;
        }
        ev.events  = EPOLLIN | EPOLLET;
        ev.data.fd = cfd;
        if (epoll_ctl(tcp->epfd, EPOLL_CTL_ADD, cfd, &ev) == -1)
        {
            ems_syslog(LOG_ERR, "epoll_ctl: %s", strerror(errno));
            close(tcp->epfd);
            cfd = -1;
            continue;
        }

        if(tcp->chan_pr != NULL)
            change_channel_lock(tcp->chan_pr);
        while (atomic_load(&tcp->state) < FRPC_SERVER_STOPING && cfd > 0)
        {
            int nfds = epoll_wait(tcp->epfd, events, 10, 1000);
            if (nfds == -1)
            {
                ems_syslog(LOG_NOTICE, "epoll_wait: %s", strerror(errno));
                continue;
            }
            for (int n = 0; n < nfds; ++n)
            {
                int fd  = events[n].data.fd;
                int len = read(fd, buffer, sizeof(buffer));
                if (len < 0)
                {
                    ems_syslog(LOG_NOTICE, "read error: %s", strerror(errno));
                    close(fd);
                }
                else if (len == 0)
                {
                    ems_syslog(LOG_INFO, "client close!");
                    close(fd);
                    if (fd == cfd)
                    {
                        cfd = -1;
                    }
                }
                else
                {
                    int l = write(fd == local_cfd ? cfd : local_cfd, buffer, len);
                    if (l > 0)
                    {
                        write_bin_data(buffer, len, tcp->log_fp, fd == local_cfd);
                    }
                }
            }
        }
        if (tcp->chan_pr != NULL)
            change_channel_unlock(tcp->chan_pr);
        if (cfd > 0)
        {
            close(cfd);
            cfd = -1;
        }
    }
END:
    if (local_cfd > 0) close(local_cfd);

    if (tcp->sfd > 0)
    {
        close(tcp->sfd);
        tcp->sfd = -1;
    }
    if (tcp->epfd > 0)
    {
        close(tcp->epfd);
        tcp->epfd = -1;
    }
    atomic_store(&tcp->state, FRPC_SERVER_STOPED);
    return NULL;
}
// 帧模式 支持多个连接,报文需要解析
static void* tcp_server_frame_task(void* arg)
{
    tcp_trans_t* tcp = (tcp_trans_t*)arg;
    if (tcp == NULL || tcp->sfd < 0)
    {
        ems_syslog(LOG_ERR, "proxy_server cfg error");
        goto END;
    }

    tcp->epfd = epoll_create1(0);
    if (tcp->epfd < 0)
    {
        ems_syslog(LOG_ERR, "epoll_create1 error: %s", strerror(errno));
        goto END;
    }

    struct epoll_event ev, events[10];
    ev.events  = EPOLLIN;
    ev.data.fd = tcp->sfd;
    if (epoll_ctl(tcp->epfd, EPOLL_CTL_ADD, tcp->sfd, &ev) < 0)
    {
        ems_syslog(LOG_ERR, "epoll_ctl error: %s", strerror(errno));
        goto END;
    }

    char buffer[4096] = {0};
    atomic_store(&tcp->state, FRPC_SERVER_RUNNING);

    while (atomic_load(&tcp->state) < FRPC_SERVER_STOPING)
    {
        int nfds = epoll_wait(tcp->epfd, events, 10, 1000);
        if (nfds < 0)
        {
            ems_syslog(LOG_NOTICE, "epoll_wait error: %s", strerror(errno));
            continue;
        }
        for (int i = 0; i < nfds; i++)
        {
            if (events[i].data.fd == tcp->sfd)
            {
                int cfd = accept(tcp->sfd, NULL, NULL);
                if (cfd < 0)
                {
                    ems_syslog(LOG_NOTICE, "accept error: %s", strerror(errno));
                    continue;
                }
                ev.events  = EPOLLIN;
                ev.data.fd = cfd;
                if (epoll_ctl(tcp->epfd, EPOLL_CTL_ADD, cfd, &ev) < 0)
                {
                    ems_syslog(LOG_NOTICE, "epoll_ctl error: %s", strerror(errno));
                    close(cfd);
                    continue;
                }
            }
            else
            {
                int len = read(events[i].data.fd, buffer, sizeof(buffer));
                if (len <= 0)
                {
                    ems_syslog(LOG_INFO, "client close!");
                    close(events[i].data.fd);
                    continue;
                }
                else
                {
                    int l = write_and_read_dev_no(tcp, buffer, len, tcp->log_fp);
                    if (l > 0)
                    {
                        write(events[i].data.fd, buffer, l);
                    }
                    else if (l < 0)
                    {
                        ems_syslog(LOG_NOTICE, "write_and_read_dev_no error: %d", l);
                    }
                }
            }
        }
    }

END:
    if (tcp->sfd > 0)
    {
        close(tcp->sfd);
        tcp->sfd = -1;
    }
    if (tcp->epfd > 0)
    {
        close(tcp->epfd);
        tcp->epfd = -1;
    }
    atomic_store(&tcp->state, FRPC_SERVER_STOPED);
    return NULL;
}

int write_and_read_dev_no(tcp_trans_t* cfg, char* buffer, int len, FILE* log)
{
    int        ret     = -1;
    channel_t* chan_pr = cfg->chan_pr;

    channel_lock(chan_pr);
    if (chan_pr->s > 0)
    {
        if (write(chan_pr->s, buffer, len) < 0)
        {
            proto_syslog(LOG_ERR, "dev_no[%s] send data error!", cfg->chan_pr->channel);
            ret = -2;
        }
        else
        {
            write_bin_data(buffer, len, log, true);
            int l = read_data_timeout(chan_pr->s, (unsigned char*)buffer, 4096, cfg->timeout);
            if (l < 0)
                ret = -1;
            else
            {
                ret = l;
                write_bin_data(buffer, l, log, true);
            }
        }
    }
    else
        ret = -2;
    channel_unlock(chan_pr);
    return ret;
}

int tcp_server_listen(tcp_trans_t* cfg)
{
    int ret  = -1;
    cfg->sfd = socket(AF_INET, SOCK_STREAM, 0);
    if (cfg->sfd < 0)
    {
        ems_syslog(LOG_ERR, "socket error: %s", strerror(errno));
        goto END;
    }

    int opt = 1;
    setsockopt(cfg->sfd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));

    struct sockaddr_in addr = {
        .sin_family      = AF_INET,
        .sin_port        = htons(cfg->port),
        .sin_addr.s_addr = INADDR_ANY};

    if (bind(cfg->sfd, (struct sockaddr*)&addr, sizeof(addr)) < 0)
    {
        ems_syslog(LOG_ERR, "bind error: %s", strerror(errno));
        goto END;
    }

    socklen_t len = sizeof(addr);
    if (getsockname(cfg->sfd, (struct sockaddr*)&addr, &len) == -1)
    {
        ems_syslog(LOG_ERR, "getsockname failed");
        goto END;
    }
    cfg->port = ntohs(addr.sin_port);

    if (listen(cfg->sfd, 1) < 0)
    {
        ems_syslog(LOG_ERR, "listen error: %s", strerror(errno));
        goto END;
    }
    // 获取通道

    ret = 0;
END:
    if (ret == 0) { return ret; }

    if (cfg->sfd > 0)
    {
        close(cfg->sfd);
        cfg->sfd = -1;
    }
    return ret;
}
void* tcp_server_tcp2rtu_task(void* arg);

int tcp_server_task_start(tcp_trans_t* trans)
{
    char name[128] = {0};
    int  ret       = -1;
    if (tcp_server_listen(trans) < 0)
    {
        return ret;
    }

    if (trans->mode == PROXY_MODE_STREAM)
    {
        if (pthread_create(&trans->server_tid, NULL, tcp_server_stream_task, trans) != 0)
        {
            trans->server_tid = -1;
        }
        else
        {
            atomic_store(&trans->state, FRPC_SERVER_RUNNING);
            snprintf(name, sizeof(name), "tcp_server_stream_task_%s", trans->chan_pr->channel);
            pthread_setname_np(trans->server_tid, name);
            ret = 0;
        }
    }
    else if (trans->mode == PROXY_MODE_FRAME)
    {
        if (pthread_create(&trans->server_tid, NULL, tcp_server_frame_task, trans) != 0)
        {
            trans->server_tid = -1;
        }
        else
        {
            atomic_store(&trans->state, FRPC_SERVER_RUNNING);
            snprintf(name, sizeof(name), "tcp_server_frame_task_%s", trans->chan_pr->channel);
            pthread_setname_np(trans->server_tid, name);
            ret = 0;
        }
    }
    else if (trans->mode == PROXY_MODE_OTHERPORT)
    {
        if (pthread_create(&trans->server_tid, NULL, tcp_server_thirdport_task, trans) != 0)
        {
            trans->server_tid = -1;
        }
        else
        {
            atomic_store(&trans->state, FRPC_SERVER_RUNNING);
            snprintf(name, sizeof(name), "tcp_server_otherport_task_%s", trans->chan_pr->channel);
            pthread_setname_np(trans->server_tid, name);
            ret = 0;
        }
    }
    else if (trans->mode == PROXY_MODE_TCP2RTU)
    {
        if (pthread_create(&trans->server_tid, NULL, tcp_server_tcp2rtu_task, trans) != 0)
        {
            trans->server_tid = -1;
        }
        else
        {
            atomic_store(&trans->state, FRPC_SERVER_RUNNING);
            snprintf(name, sizeof(name), "tcp_server_tcp2rtu_task_%s", trans->chan_pr->channel);
            pthread_setname_np(trans->server_tid, name);
            ret = 0;
        }
    }
    else
    {
        trans->server_tid = -1;
    }
    return ret;
}

int tcp_server_task_stop(tcp_trans_t* cfg)
{
    if (atomic_load(&cfg->state) != FRPC_SERVER_STOPING)
    {
        atomic_store(&cfg->state, FRPC_SERVER_STOPING);
    }
    if (cfg->server_tid > 0)
    {
        pthread_join(cfg->server_tid, NULL);
        cfg->server_tid = -1;
    }
    return 0;
}

int tcp_server_task_destroy(tcp_trans_t* cfg)
{
    if(cfg->chan_pr)
    {
        cfg->chan_pr->proxy_state = PROXY_MODE_NONE;
    }
    if (cfg->log_fp)
    {
        fclose(cfg->log_fp);
        cfg->log_fp = NULL;
    }
    return 0;
}