#include "../ndlb_alloc.h"
#include "../ndlb_mt_log.h"
#include "../ndlb_thread_handle.h"
#include "../ndlb_utiles.h"
#include "nd_proxy_server.h"
#include "../ndGlobalMacro.h"
#include "rudp.h"
#include <stdint.h>

rudpList_session rudpList;
int count = 0;
extern struct udp_thread_storage* udpStorage;
extern void MovetoUDPWriteList(rudp_packet_wrapper* curr_node);
static event_data_t* ee = NULL;

struct sockets* sockets_list_head = NULL;
static event_data_t* ee_timers = NULL;

#define MAX_EVENTS 1
event_data_t event_pool[MAX_EVENTS];
size_t event_pool_index = 0; 


int event_fd(apr_socket_t* fd, int (*fn)(apr_socket_t*, apr_sockaddr_t*, void*),
    void* arg, apr_sockaddr_t* sa)
{
 
    if (event_pool_index >= MAX_EVENTS)
    {
        NDTL0(trace_log_key, NULL, TL_INFO, "Method Called")
        return -1; // Pool is full
    }

    event_data_t* e = &event_pool[event_pool_index++];

    e->e_fd = fd;
    e->e_fn = fn;
    e->e_arg = arg;
    e->e_type = EVENT_FD;
    e->e_next = ee;
    e->sa = sa;
    ee = e;

    return 0;
}



int eventloop(apr_pool_t* pool) {

    apr_pollset_t* pollset;
    apr_pollfd_t pollfd;
    apr_interval_time_t timeout = -1;
    int num;
    apr_status_t rv;

    apr_pollset_create(&pollset, 1024, pool, 0);


    while (ee) {

        for (event_data_t* e = ee; e; e = e->e_next) {
            if (e->e_type == EVENT_FD) {
                pollfd.desc_type = APR_POLL_SOCKET;
                pollfd.reqevents = APR_POLLIN;
                pollfd.desc.s = e->e_fd;
                pollfd.client_data = e;

                apr_pollset_add(pollset, &pollfd);
            }
        }

        rv = apr_pollset_poll(pollset, timeout, &num, NULL);

        if (rv != APR_SUCCESS) {
            if (APR_STATUS_IS_EOF(rv) || APR_STATUS_IS_EINTR(rv))
                continue;
            else
                break;
        }
        if (NDProxy.shutdownProxy) break;
        const apr_pollfd_t* signalled = NULL;
        for (int i = 0; i < num; i++) {
            apr_pollset_poll(pollset, 0, &num, &signalled);
            if (!signalled) continue;
            event_data_t* e = signalled->client_data;
            if ((*e->e_fn)(e->e_fd,e->sa,e->e_arg) < 0) {
                apr_pollset_remove(pollset, signalled);
                return -1;
            }
        }
    }

    apr_pollset_destroy(pollset);
    apr_pool_destroy(pool);
    apr_terminate();

    return 0;
}




apr_socket_t* rudp_socket(apr_sockaddr_t* sa, int port, apr_pool_t* pool) {

    apr_socket_t* sockfd = NULL;
    apr_status_t status;

    status = apr_sockaddr_info_get(&sa, "0.0.0.0", APR_INET, port, 0, pool);
    if (status != APR_SUCCESS) {
        return NULL;
    }

    status = apr_socket_create(&sockfd, sa->family, SOCK_DGRAM, APR_PROTO_UDP, pool);

    if (status != APR_SUCCESS) {
        return NULL;
    }

    status = apr_socket_bind(sockfd, sa);
    if (status != APR_SUCCESS) {
        apr_socket_close(sockfd);
        return NULL;
    }

    apr_socket_t* socket = (apr_socket_t*)sockfd;
    struct sockets * newSocket = malloc(sizeof(struct sockets));
    newSocket->rsock = sockfd;
    newSocket->sessions_list_head = NULL;
    newSocket->next = NULL;

    if (sockets_list_head == NULL) {
        sockets_list_head = newSocket;
    }
    else {
        struct sockets* temp = sockets_list_head;
        while (temp->next != NULL) {
            temp = temp->next;
        }
        temp->next = newSocket;
    
    }

    event_fd(socket, receiveCallback, sockfd, sa);
    if (session_buff_list_init()) {
        NDTL0(trace_log_key, NULL, TL_ERROR, "Error in creating udp buffer...");
        return NULL;
    }
    return socket;
}

int session_buff_list_init()
{
    NDTL4(trace_log_key, NULL, TL_INFO, "Method called");
    int i;
    int maxSessionBuffers = 10;

    session* node = NULL;
    session* curr_node = NULL;
    rudpList.session_writeList = NULL;
    rudpList.session_bufferWriteLock = NULL;

    apr_thread_mutex_create(&rudpList.session_bufferWriteLock, APR_THREAD_MUTEX_NESTED, NDProxy.proxy_pool);
    if (!rudpList.session_bufferWriteLock) {
        NDTL0(trace_log_key, NULL, TL_INFO, "Failed to create lock for session");
        return 1;
    }

    char msg[512];
    for (i = 0; i < maxSessionBuffers; i++)
    {
        snprintf(msg, 512, "session buffer node : %d", i);
        NDLB_MALLOC_AND_MEMSET(node, sizeof(session), msg, -1);
        if (!node)
            return 1;

        NDLB_MALLOC_AND_MEMSET(node->sender, sizeof(struct sender_session), msg, -1);
        if (!node->sender)
            return 1;
        NDLB_MALLOC_AND_MEMSET(node->receiver, sizeof(struct receiver_session), msg, -1);
        if (!node->receiver)
            return 1;

        if (!rudpList.session_writeList)
        {
            rudpList.session_writeList = node;
            curr_node = node;
        }
        else
        {
            curr_node->next = node;
            curr_node = node;
        }
        rudpList.session_writeListLast = node;
    }

    NDTL4(trace_log_key, NULL, TL_INFO, "Method exit");
    return 0;
}
void MovetoSessionWriteList(session* curr_node) {
    apr_thread_mutex_lock(rudpList.session_bufferWriteLock);
    if (!rudpList.session_writeListLast) {
        rudpList.session_writeListLast = rudpList.session_writeList = curr_node;
    }
    else {
        rudpList.session_writeListLast->next = curr_node;
        rudpList.session_writeListLast = curr_node;
    }
    apr_thread_mutex_unlock(rudpList.session_bufferWriteLock);
}

session* AcquireSessionFromList() {

    apr_thread_mutex_lock(rudpList.session_bufferWriteLock);
    if (rudpList.session_writeList == NULL) {
        rudpList.session_writeListLast = NULL;
     

        NDTL0(trace_log_key, NULL, TL_ERROR, "rudpList.session_writeList == NULL");
        apr_thread_mutex_unlock(rudpList.session_bufferWriteLock);
        apr_sleep(1000000);
        return 0;
    }

    session* curr_node = rudpList.session_writeList;
    rudpList.session_writeList = rudpList.session_writeList->next;
    curr_node->next = NULL;
    if (!rudpList.session_writeList) rudpList.session_writeListLast = NULL;
    apr_thread_mutex_unlock(rudpList.session_bufferWriteLock);
    
    return curr_node;


}
int receiveCallback(apr_socket_t* file, apr_sockaddr_t* sender2,void * arg) {

    int* rvstatus = NULL;

    apr_size_t buf_len = sizeof(struct rudp_packet);

 
        apr_thread_mutex_lock(NDProxy.udp_bufferWriteLock);
        if (NDProxy.udp_writeList == NULL) {
            NDProxy.udp_writeListLast = NULL;
            apr_thread_mutex_unlock(NDProxy.udp_bufferWriteLock);
            apr_sleep(1000000);
            return 0;
        }
         
        rudp_packet_wrapper* curr_node = NDProxy.udp_writeList;
        NDProxy.udp_writeList = NDProxy.udp_writeList->next;
        curr_node->next = NULL;
        if (!NDProxy.udp_writeList) NDProxy.udp_writeListLast = NULL;
        apr_thread_mutex_unlock(NDProxy.udp_bufferWriteLock);

        apr_status_t status = apr_socket_recvfrom(sender2, file, 0, (char*)&curr_node->received_packet,
            &buf_len);


        NDTL0(trace_log_key, NULL, TL_ERROR, "sender payload %d count1: %d  %d %d\n",
            curr_node->received_packet.payload_length, ++count, curr_node->received_packet.header.seqno, curr_node->received_packet.header.duplicate_flag);
 

        if (curr_node->received_packet.payload_length == NDLB_PARTIAL_LINE ||
            curr_node->received_packet.payload_length == NDLB_CONNECTION_CLOSE) {
      
            MovetoUDPWriteList(curr_node);
            return 0;
        }

        int indxThread = 0;

#ifdef __unix__
        indxThread = (sender2->sa.sin.sin_addr.s_addr ^ sender2->sa.sin.sin_port) %
            NDProxy.num_udpThreads;

#else

        struct in_addr* ad = (struct in_addr*)sender2->ipaddr_ptr;
        indxThread = (ad->s_addr ^ sender2->port) % NDProxy.num_udpThreads;

#endif

        if (indxThread < 0 || indxThread > NDProxy.num_udpThreads - 1) {
            NDTL0(trace_log_key, NULL, TL_ERROR, "NDProxy.num_udpThreads");

            return 0;
        }

        curr_node->file = file;
        curr_node->sender = sender2;

        rudp_packet_wrapper* th_node = NULL;
        apr_thread_mutex_lock(udpStorage[indxThread].thread_bufferReadLock);
        udpStorage[indxThread].saddr = sender2;
        th_node = udpStorage[indxThread].node;

        if (th_node == NULL) {

            udpStorage[indxThread].node = curr_node;
        }
        else {

            while (th_node->next != NULL) th_node = th_node->next;
            th_node->next = curr_node;

        }
        apr_thread_cond_signal(udpStorage[indxThread].thread_cond_var);
        apr_thread_mutex_unlock(udpStorage[indxThread].thread_bufferReadLock);

   

   

    return 0;
}
int sendingAck(apr_socket_t* file, apr_sockaddr_t* sender1,struct rudp_packet *received_packet) {

    struct rudp_hdr rudpheader = received_packet->header;
    if (sockets_list_head == NULL) {
      //  NDTL0(trace_log_key, NULL, TL_ERROR, "sockets_list_head == NULL");

        return -1;
    }
    else 
    {
        struct sockets* temp = sockets_list_head;
        while (temp != NULL) {
            if (temp->rsock == file) {
              //  NDTL0(trace_log_key, NULL, TL_ERROR, "temp->rsock == file");

                break;
            }
            temp = temp->next;
        }
        if (temp->rsock == file) {

            if (temp->sessions_list_head == NULL) {
                if (rudpheader.type == RUDP_SYN) {


                    session* curr_node = AcquireSessionFromList();
                   
                    curr_node->address = *sender1;
                    curr_node->next = NULL;
                    curr_node->sender = NULL;
                    curr_node->receiver->status = OPENING;
                    curr_node->receiver->expected_seqNo = rudpheader.seqno + 1;
                  
                    temp->sessions_list_head = curr_node;

                    struct rudp_hdr ack;
                    ack.type = RUDP_ACK;
                    ack.version = RUDP_VERSION;
                    ack.seqno = curr_node->receiver->expected_seqNo;


                    struct rudp_packet p2;
                    p2.header.seqno = ack.seqno;
                    p2.header.version = ack.version;
                    p2.header.type = ack.type;
                    p2.payload_length = 0;
                    send_packet(file, &p2, sender1);

                  

                }
            }

            else {

                int sessionFound = 0;

                struct session* temp2 = temp->sessions_list_head;

  

                struct session* last_session = NULL;
                while (temp2 != NULL) {
                    if (temp2->next == NULL) {
                        last_session = temp2;
                    }

    #ifdef __unix__
                    if (sender1->sa.sin.sin_addr.s_addr == temp2->address.sa.sin.sin_addr.s_addr && temp2->address.port == sender1->port) {
                        sessionFound = 1;
                        break;
                    }
    #else
                    struct in_addr* ad1 = (struct in_addr*)sender1->ipaddr_ptr;
                    struct in_addr* ad2 = (struct in_addr*)temp2->address.ipaddr_ptr;
                    if (temp2->address.port == sender1->port && ad1->s_addr == ad2->s_addr) {
           //      NDTL0(trace_log_key, NULL, TL_ERROR, "temp2->address.port %d sender1->port %d ad1->s_addr %d ad2->s_addr %d", temp2->address.port, sender1->port, ad1->s_addr, ad2->s_addr);

                        sessionFound = 1;
                        break;
                    }
    #endif


                    temp2 = temp2->next;
                }

                if (sessionFound == 0) {
                    if (rudpheader.type == RUDP_SYN) {

                        session* curr_node1 = AcquireSessionFromList();

                        curr_node1->address = *sender1;
                        curr_node1->next = NULL;
                        curr_node1->sender = NULL;
                        curr_node1->receiver->status = OPENING;
                        curr_node1->receiver->expected_seqNo = rudpheader.seqno + 1;
                        last_session->next = curr_node1;
                        struct rudp_hdr ack;
                        ack.type = RUDP_ACK;
                        ack.version = RUDP_VERSION;
                        ack.seqno = curr_node1->receiver->expected_seqNo;

                        struct rudp_packet p2;
                        p2.header.seqno = ack.seqno;
                        p2.header.version = ack.version;
                        p2.header.type = ack.type;
                        p2.payload_length = 0;

                        send_packet( file, &p2, sender1);
                      
                    }
                }
                else {

                 //   NDTL0(trace_log_key, NULL, TL_ERROR, "sessionFound == 1");

                    if (rudpheader.type == RUDP_DATA) {

                        if (temp2->receiver->status!=NULL && temp2->receiver->status == OPENING) {
                            if (rudpheader.seqno == temp2->receiver->expected_seqNo) {
                                temp2->receiver->status = OPEN;
                            }
                        }
                     //   NDTL0(trace_log_key, NULL, TL_ERROR, "rudpheader.seqno %d temp2->receiver->expected_seqNo %d", rudpheader.seqno, temp2->receiver->expected_seqNo);

                        if (rudpheader.seqno == temp2->receiver->expected_seqNo) {

                            struct rudp_hdr ack;
                            ack.type = RUDP_ACK;
                            ack.version = RUDP_VERSION;
                            ack.seqno = (rudpheader.seqno + (uint32_t)1);

                            struct rudp_packet p;
                            p.payload_length = 0;
                            p.header.seqno = ack.seqno;
                            p.header.version = ack.version;
                            p.header.type = ack.type;

                            send_packet(file, &p, sender1);
                            
                            temp2->receiver->expected_seqNo = ack.seqno;

                  
                          
                           

                        }
                        else
                            
                            
                          /*  if (SEQ_GEQ(rudpheader.seqno,
                                          (temp2->receiver->expected_seqNo -
                                            (uint32_t)RUDP_WINDOW)) &&
                                SEQ_LT(rudpheader.seqno,
                                         temp2->receiver->expected_seqNo))*/


                        {
      
                            struct rudp_hdr ack;
                            ack.type = RUDP_ACK;
                            ack.version = RUDP_VERSION;
                            ack.seqno = (rudpheader.seqno + (uint32_t)1);
                          
                            struct rudp_packet p;
                            p.payload_length = 0;
                            p.header.seqno = ack.seqno;
                            p.header.version = ack.version;
                            p.header.type = ack.type;

                            send_packet(file, &p, sender1);
                        
                     
                    }
                  }
                }
            }
        }    
    }
    return 0;
}



int send_packet( apr_socket_t* rsocket, struct rudp_packet *p,apr_sockaddr_t* recipient) {

    apr_size_t data_len = sizeof(struct rudp_packet);
    apr_status_t rv;

    rv = apr_socket_sendto(rsocket, recipient, 0, (const char*)p, &data_len);

    NDTL0(trace_log_key, NULL, TL_ERROR, "Sending  packet  seq number=%u on socket=%d\n", p->header.seqno, rsocket);
  //  NDTL0(trace_log_key, NULL, TL_ERROR, "Sending  packet");
    char error_buf[ERROR_BUFFER_SIZE];
    if (rv != APR_SUCCESS && !APR_STATUS_IS_EAGAIN(status)) {
        fprintf(stderr, "Error sending data via UDP\n");
        return 1;
    }
    if (rv == APR_TIMEUP) {
        apr_strerror(rv, error_buf, ERROR_BUFFER_SIZE);
        NDTL0(trace_log_key, NULL, TL_INFO, "Send Failed '%d', %s", rv, error_buf);
        return NDLB_CONNECTION_TIMEOUT;
    }
    else if (rv == APR_EOF || rv != APR_SUCCESS) {
        apr_strerror(rv, error_buf, ERROR_BUFFER_SIZE);
        NDTL0(trace_log_key, NULL, TL_INFO,
            "Send Failed while sending buffer'%d', %s", rv, error_buf);
        return NDLB_CONNECTION_CLOSE;
    }

  


    return 0;
}
