/*******************************************************************************
    OpenAirInterface
    Copyright(c) 1999 - 2014 Eurecom

    OpenAirInterface is free software: you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
    the Free Software Foundation, either version 3 of the License, or
    (at your option) any later version.


    OpenAirInterface is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU General Public License for more details.

    You should have received a copy of the GNU General Public License
    along with OpenAirInterface.The full GNU General Public License is
   included in this distribution in the file called "COPYING". If not,
   see <http://www.gnu.org/licenses/>.

  Contact Information
  OpenAirInterface Admin: openair_admin@eurecom.fr
  OpenAirInterface Tech : openair_tech@eurecom.fr
  OpenAirInterface Dev  : openair4g-devel@eurecom.fr

  Address      : Eurecom, Compus SophiaTech 450, route des chappes, 06451 Biot, France.

 *******************************************************************************/

/*! \file sctp_primitives_server.c
 *  \brief Main server primitives
 *  \author Sebastien ROUX
 *  \date 2013
 *  \version 1.0
 *  @ingroup _sctp
 */

#include <pthread.h>

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <errno.h>

#include <sys/types.h>
#include <sys/socket.h>

#include <netinet/in.h>
#include <netinet/sctp.h>

#include <arpa/inet.h>

#include "assertions.h"

#include "intertask_interface.h"
#include "sctp_primitives_server.h"
#include "mme_config.h"
#include "conversions.h"

#include "sctp_common.h"
#include "sctp_itti_messaging.h"

#define SCTP_DUMP_LIST

#define SCTP_RC_ERROR       -1
#define SCTP_RC_NORMAL_READ  0
#define SCTP_RC_DISCONNECT   1

struct sctp_association_s {
  struct sctp_association_s *next_assoc;       ///< Next association in the list
  struct sctp_association_s *previous_assoc;   ///< Previous association in the list
  int      sd;            ///< Socket descriptor
  uint32_t ppid;          ///< Payload protocol Identifier
  uint16_t instreams;     ///< Number of input streams negociated for this connection
  uint16_t outstreams;    ///< Number of output strams negotiated for this connection
  int32_t  assoc_id;      ///< SCTP association id for the connection
  uint32_t messages_recv; ///< Number of messages received on this connection
  uint32_t messages_sent; ///< Number of messages sent on this connection

  struct   sockaddr *peer_addresses;   ///< A list of peer addresses
  int      nb_peer_addresses;
};

struct sctp_descriptor_s {
  // List of connected peers
  struct sctp_association_s *available_connections_head;
  struct sctp_association_s *available_connections_tail;

  uint32_t number_of_connections;
  uint16_t nb_instreams;
  uint16_t nb_outstreams;
};

struct sctp_arg_s {
  int      sd;
  uint32_t ppid;
};

static struct sctp_descriptor_s sctp_desc;

// Thread used to handle sctp messages
static pthread_t assoc_thread;

// LOCAL FUNCTIONS prototypes
void *sctp_receiver_thread(void *args_p);
static int sctp_send_msg(int32_t sctp_assoc_id, uint16_t stream,
                         const uint8_t *buffer, const uint32_t length);

// Association list related local functions prototypes
static struct sctp_association_s *sctp_is_assoc_in_list(int32_t assoc_id);
static struct sctp_association_s *sctp_add_new_peer(void);
static int sctp_handle_com_down(uint32_t assoc_id);
static void sctp_dump_list(void);

static struct sctp_association_s *sctp_add_new_peer(void)
{
  struct sctp_association_s *new_sctp_descriptor = NULL;

  new_sctp_descriptor = calloc(1, sizeof(struct sctp_association_s));

  if (new_sctp_descriptor == NULL) {
    SCTP_ERROR("Failed to allocate memory for new peer (%s:%d)\n",
               __FILE__, __LINE__);
    return NULL;
  }

  new_sctp_descriptor->next_assoc = NULL;
  new_sctp_descriptor->previous_assoc = NULL;

  if (sctp_desc.available_connections_tail == NULL) {
    sctp_desc.available_connections_head = new_sctp_descriptor;
    sctp_desc.available_connections_tail = sctp_desc.available_connections_head;
  } else {
    new_sctp_descriptor->previous_assoc = sctp_desc.available_connections_tail;
    sctp_desc.available_connections_tail->next_assoc = new_sctp_descriptor;
    sctp_desc.available_connections_tail = new_sctp_descriptor;
  }

  sctp_desc.number_of_connections++;

  sctp_dump_list();

  return new_sctp_descriptor;
}

static struct sctp_association_s *sctp_is_assoc_in_list(int32_t assoc_id)
{
  struct sctp_association_s *assoc_desc;

  if (assoc_id < 0) {
    return NULL;
  }

  for (assoc_desc = sctp_desc.available_connections_head;
       assoc_desc;
       assoc_desc = assoc_desc->next_assoc) {
    if (assoc_desc->assoc_id == assoc_id) {
      break;
    }
  }

  return assoc_desc;
}

static int sctp_remove_assoc_from_list(int32_t assoc_id)
{
  struct sctp_association_s *assoc_desc;

  /* Association not in the list */
  if ((assoc_desc = sctp_is_assoc_in_list(assoc_id)) == NULL) {
    return -1;
  }

  if (assoc_desc->next_assoc == NULL) {
    if (assoc_desc->previous_assoc == NULL) {
      /* Head and tail */
      sctp_desc.available_connections_head = sctp_desc.available_connections_tail = NULL;
    } else {
      /* Not head but tail */
      sctp_desc.available_connections_tail = assoc_desc->previous_assoc;
      assoc_desc->previous_assoc->next_assoc = NULL;
    }
  } else {
    if (assoc_desc->previous_assoc == NULL) {
      /* Head but not tail */
      sctp_desc.available_connections_head = assoc_desc->next_assoc;
      assoc_desc->next_assoc->previous_assoc = NULL;
    } else {
      /* Not head and not tail */
      assoc_desc->previous_assoc->next_assoc = assoc_desc->next_assoc;
      assoc_desc->next_assoc->previous_assoc = assoc_desc->previous_assoc;
    }
  }

  free(assoc_desc);
  assoc_desc = NULL;
  sctp_desc.number_of_connections --;
  return 0;
}

static void sctp_dump_assoc(struct sctp_association_s *sctp_assoc_p)
{
#if defined(SCTP_DUMP_LIST)
  int i;

  if (sctp_assoc_p == NULL) {
    return;
  }

  SCTP_DEBUG("sd           : %d\n", sctp_assoc_p->sd);
  SCTP_DEBUG("input streams: %d\n", sctp_assoc_p->instreams);
  SCTP_DEBUG("out streams  : %d\n", sctp_assoc_p->outstreams);
  SCTP_DEBUG("assoc_id     : %d\n", sctp_assoc_p->assoc_id);
  SCTP_DEBUG("peer address :\n");

  for (i = 0; i < sctp_assoc_p->nb_peer_addresses; i++) {
    char address[40];

    memset(address, 0, sizeof(address));

    if (inet_ntop(sctp_assoc_p->peer_addresses[i].sa_family,
                  sctp_assoc_p->peer_addresses[i].sa_data, address, sizeof(address)) != NULL) {
      SCTP_DEBUG("    - [%s]\n", address);
    }
  }

#else
  sctp_assoc_p = sctp_assoc_p;
#endif
}

static void sctp_dump_list(void)
{
#if defined(SCTP_DUMP_LIST)
  struct sctp_association_s *sctp_assoc_p;

  sctp_assoc_p = sctp_desc.available_connections_head;

  SCTP_DEBUG("SCTP list contains %d associations\n", sctp_desc.number_of_connections);

  while (sctp_assoc_p != NULL) {
    sctp_dump_assoc(sctp_assoc_p);
    sctp_assoc_p = sctp_assoc_p->next_assoc;
  }

#else
  sctp_dump_assoc(NULL);
#endif
}

static int sctp_send_msg(int32_t sctp_assoc_id, uint16_t stream,
                         const uint8_t *buffer, const uint32_t length)
{
  struct sctp_association_s *assoc_desc = NULL;

  DevAssert(buffer != NULL);

  if ((assoc_desc = sctp_is_assoc_in_list(sctp_assoc_id)) == NULL) {
    SCTP_DEBUG("This assoc id has not been fount in list (%d)\n",
               sctp_assoc_id);
    return -1;
  }

  if (assoc_desc->sd == -1) {
    /* The socket is invalid may be closed.
     */
    return -1;
  }

  SCTP_DEBUG("[%d][%d] Sending buffer %p of %d bytes on stream %d with ppid %d\n",
             assoc_desc->sd, sctp_assoc_id, buffer,
             length, stream, assoc_desc->ppid);

  /* Send message_p on specified stream of the sd association */
  if (sctp_sendmsg(assoc_desc->sd,
                   (const void *)buffer,
                   length,
                   NULL,
                   0,
                   assoc_desc->ppid, 0, stream, 0, 0) < 0) {
    SCTP_ERROR("send: %s:%d", strerror(errno), errno);
    return -1;
  }

  assoc_desc->messages_sent++;

  SCTP_DEBUG("Successfully sent %d bytes on stream %d\n", length, stream);

  return 0;
}

static int sctp_create_new_listener(SctpInit *init_p)
{
  struct sctp_event_subscribe event;
  struct sockaddr *addr;

  struct sctp_arg_s *sctp_arg_p;

  uint16_t i = 0, j;
  int sd;
  int used_addresses = 0;

  DevAssert(init_p != NULL);

  if (init_p->ipv4 == 0 && init_p->ipv6 == 0) {
    SCTP_ERROR("Illegal IP configuration upper layer should request at"
               "least ipv4 and/or ipv6 config\n");
    return -1;
  }

  if ((used_addresses = init_p->nb_ipv4_addr + init_p->nb_ipv6_addr) == 0) {
    SCTP_WARN("No address provided...\n");
    return -1;
  }

  addr = calloc(used_addresses, sizeof(struct sockaddr));

  SCTP_DEBUG("Creating new listen socket on port %u with\n", init_p->port);

  if (init_p->ipv4 == 1) {
    struct sockaddr_in *ip4_addr;

    SCTP_DEBUG("ipv4 addresses:\n");

    for (i = 0; i < init_p->nb_ipv4_addr; i++) {
      SCTP_DEBUG("\t- "IPV4_ADDR"\n",
                 IPV4_ADDR_FORMAT(init_p->ipv4_address[i]));
      ip4_addr = (struct sockaddr_in *)&addr[i];
      ip4_addr->sin_family = AF_INET;
      ip4_addr->sin_port   = htons(init_p->port);
      ip4_addr->sin_addr.s_addr = init_p->ipv4_address[i];
    }
  }

  if (init_p->ipv6 == 1) {
    struct sockaddr_in6 *ip6_addr;

    SCTP_DEBUG("ipv6 addresses:\n");

    for (j = 0; j < init_p->nb_ipv6_addr; j++) {
      SCTP_DEBUG("\t- %s\n", init_p->ipv6_address[j]);
      ip6_addr = (struct sockaddr_in6 *)&addr[i + j];
      ip6_addr->sin6_family = AF_INET6;
      ip6_addr->sin6_port  = htons(init_p->port);

      if (inet_pton(AF_INET6, init_p->ipv6_address[j],
                    ip6_addr->sin6_addr.s6_addr) <= 0) {
        SCTP_WARN("Provided ipv6 address %s is not valid\n",
                  init_p->ipv6_address[j]);
      }
    }
  }

  if ((sd = socket(AF_INET6, SOCK_STREAM, IPPROTO_SCTP)) < 0) {
    SCTP_ERROR("socket: %s:%d\n", strerror(errno), errno);
    return -1;
  }

  memset((void *)&event, 1, sizeof(struct sctp_event_subscribe));

  if (setsockopt(sd, IPPROTO_SCTP, SCTP_EVENTS, &event,
                 sizeof(struct sctp_event_subscribe)) < 0) {
    SCTP_ERROR("setsockopt: %s:%d\n", strerror(errno), errno);
    return -1;
  }

  /* Some pre-bind socket configuration */
  if (sctp_set_init_opt(sd, sctp_desc.nb_instreams, sctp_desc.nb_outstreams,
                        0, 0) < 0) {
    goto err;
  }

  if (sctp_bindx(sd, addr, used_addresses, SCTP_BINDX_ADD_ADDR) != 0) {
    SCTP_ERROR("sctp_bindx: %s:%d\n", strerror(errno), errno);
    return -1;
  }

  if (listen(sd, 5) < 0) {
    SCTP_ERROR("listen: %s:%d\n", strerror(errno), errno);
    return -1;
  }

  if ((sctp_arg_p = malloc(sizeof(struct sctp_arg_s))) == NULL) {
    return -1;
  }

  sctp_arg_p->sd = sd;
  sctp_arg_p->ppid = init_p->ppid;

  if (pthread_create(&assoc_thread, NULL, &sctp_receiver_thread,
                     (void *)sctp_arg_p) < 0) {
    SCTP_ERROR("pthread_create: %s:%d\n", strerror(errno), errno);
    return -1;
  }

  return sd;
err:

  if (sd != -1) {
    close(sd);
    sd = -1;
  }

  return -1;
}

static
inline int sctp_read_from_socket(int sd, int ppid)
{
  int flags = 0, n;
  socklen_t from_len;
  struct sctp_sndrcvinfo sinfo;

  struct sockaddr_in addr;
  uint8_t buffer[SCTP_RECV_BUFFER_SIZE];

  if (sd < 0) {
    return -1;
  }

  memset((void *)&addr, 0, sizeof(struct sockaddr_in));
  from_len = (socklen_t)sizeof(struct sockaddr_in);
  memset((void *)&sinfo, 0, sizeof(struct sctp_sndrcvinfo));
  n = sctp_recvmsg(sd, (void *)buffer, SCTP_RECV_BUFFER_SIZE,
                   (struct sockaddr *)&addr, &from_len,
                   &sinfo, &flags);

  if (n < 0) {
    SCTP_DEBUG("An error occured during read\n");
    SCTP_ERROR("sctp_recvmsg: %s:%d\n", strerror(errno), errno);
    return SCTP_RC_ERROR;
  }

  if (flags & MSG_NOTIFICATION) {
    union sctp_notification *snp;
    snp = (union sctp_notification *)buffer;

    /* Client deconnection */
    if (SCTP_SHUTDOWN_EVENT == snp->sn_header.sn_type) {
      SCTP_DEBUG("SCTP_SHUTDOWN_EVENT received\n");
      return sctp_handle_com_down(snp->sn_shutdown_event.sse_assoc_id);
    }
    /* Association has changed. */
    else if (SCTP_ASSOC_CHANGE == snp->sn_header.sn_type) {
      struct sctp_assoc_change *sctp_assoc_changed;
      sctp_assoc_changed = &snp->sn_assoc_change;

      SCTP_DEBUG("Client association changed: %d\n", sctp_assoc_changed->sac_state);

      /* New physical association requested by a peer */
      switch (sctp_assoc_changed->sac_state) {
      case SCTP_COMM_UP: {
        struct sctp_association_s *new_association;

        sctp_get_sockinfo(sd, NULL, NULL, NULL);

        SCTP_DEBUG("New connection\n");

        if ((new_association = sctp_add_new_peer()) == NULL) {
          // TODO: handle this case
          DevMessage("Unexpected error...\n");
          return SCTP_RC_ERROR;
        } else {
          new_association->sd         = sd;
          new_association->ppid       = ppid;
          new_association->instreams  = sctp_assoc_changed->sac_inbound_streams;
          new_association->outstreams = sctp_assoc_changed->sac_outbound_streams;
          new_association->assoc_id   = sctp_assoc_changed->sac_assoc_id;

          sctp_get_localaddresses(sd, NULL, NULL);
          sctp_get_peeraddresses(sd, &new_association->peer_addresses,
                                 &new_association->nb_peer_addresses);

          if (sctp_itti_send_new_association(
                new_association->assoc_id, new_association->instreams,
                new_association->outstreams) < 0) {
            SCTP_ERROR("Failed to send message to S1AP\n");
            return SCTP_RC_ERROR;
          }
        }
      }
      break;

      default:
        break;
      }
    }
  } else {
    /* Data payload received */
    struct sctp_association_s *association;

    if ((association = sctp_is_assoc_in_list(sinfo.sinfo_assoc_id)) == NULL) {
      // TODO: handle this case
      return SCTP_RC_ERROR;
    }

    association->messages_recv++;

    if (ntohl(sinfo.sinfo_ppid) != association->ppid) {
      /* Mismatch in Payload Protocol Identifier,
       * may be we received unsollicited traffic from stack other than S1AP.
       */
      SCTP_ERROR("Received data from peer with unsollicited PPID %d, expecting %d\n",
                 ntohl(sinfo.sinfo_ppid), association->ppid);
      return SCTP_RC_ERROR;
    }

    SCTP_DEBUG("[%d][%d] Msg of length %d received from port %u, on stream %d, PPID %d\n",
               sinfo.sinfo_assoc_id, sd, n, ntohs(addr.sin_port),
               sinfo.sinfo_stream, ntohl(sinfo.sinfo_ppid));

    sctp_itti_send_new_message_ind(n, buffer, sinfo.sinfo_assoc_id, sinfo.sinfo_stream,
                                   association->instreams, association->outstreams);
  }

  return SCTP_RC_NORMAL_READ;
}

static int sctp_handle_com_down(uint32_t assoc_id)
{
  SCTP_DEBUG("Sending close connection for assoc_id %u\n", assoc_id);

  if (sctp_itti_send_com_down_ind(assoc_id) < 0) {
    SCTP_ERROR("Failed to send message to TASK_S1AP\n");
  }

  if (sctp_remove_assoc_from_list(assoc_id) < 0) {
    SCTP_ERROR("Failed to find client in list\n");
  }

  return SCTP_RC_DISCONNECT;
}

void *sctp_receiver_thread(void *args_p)
{
  struct sctp_arg_s *sctp_arg_p;

  /* maximum file descriptor number */
  int fdmax, clientsock, i;
  /* master file descriptor list */
  fd_set master;
  /* temp file descriptor list for select() */
  fd_set read_fds;

  if ((sctp_arg_p = (struct sctp_arg_s *)args_p) == NULL) {
    pthread_exit(NULL);
  }

  /* clear the master and temp sets */
  FD_ZERO(&master);
  FD_ZERO(&read_fds);

  FD_SET(sctp_arg_p->sd, &master);
  fdmax = sctp_arg_p->sd; /* so far, it's this one*/

  while(1) {

    memcpy(&read_fds, &master, sizeof(master));

    if (select(fdmax+1, &read_fds, NULL, NULL, NULL) == -1) {
      SCTP_ERROR("[%d] Select() error: %s",
                 sctp_arg_p->sd, strerror(errno));
      free(args_p);
      args_p = NULL;
      pthread_exit(NULL);
    }

    for (i = 0; i <= fdmax; i++) {
      if (FD_ISSET(i, &read_fds)) {
        if (i == sctp_arg_p->sd) {
          /* There is data to read on listener socket. This means we have to accept
           * the connection.
           */
          if ((clientsock = accept(sctp_arg_p->sd, NULL, NULL)) < 0) {
            SCTP_ERROR("[%d] accept: %s:%d\n", sctp_arg_p->sd, strerror(errno), errno);
            free(args_p);
            args_p = NULL;
            pthread_exit(NULL);
          } else {
            FD_SET(clientsock, &master); /* add to master set */

            if(clientsock > fdmax) {
              /* keep track of the maximum */
              fdmax = clientsock;
            }
          }
        } else {
          int ret;

          /* Read from socket */
          ret = sctp_read_from_socket(i, sctp_arg_p->ppid);

          /* When the socket is disconnected we have to update
           * the fd_set.
           */
          if (ret == SCTP_RC_DISCONNECT) {
            /* Remove the socket from the FD set and update the max sd */
            FD_CLR(i, &master);

            if (i == fdmax) {
              while (FD_ISSET(fdmax, &master) == FALSE)
                fdmax -= 1;
            }
          }
        }
      }
    }
  }

  free(args_p);
  args_p = NULL;
  return NULL;
}

static void *sctp_intertask_interface(void *args_p)
{
  itti_mark_task_ready(TASK_SCTP);

  while(1) {
    MessageDef *received_message_p;
    itti_receive_msg(TASK_SCTP, &received_message_p);

    switch (ITTI_MSG_ID(received_message_p)) {
    case SCTP_INIT_MSG: {
      SCTP_DEBUG("Received SCTP_INIT_MSG\n");

      /* We received a new connection request */
      if (sctp_create_new_listener(&received_message_p->ittiMsg.sctpInit) < 0) {
        /* SCTP socket creation or bind failed... */
        SCTP_ERROR("Failed to create new SCTP listener\n");
      }
    }
    break;

    case SCTP_CLOSE_ASSOCIATION: {

    } break;

    case SCTP_DATA_REQ: {
      if (sctp_send_msg(SCTP_DATA_REQ(received_message_p).assocId,
                        SCTP_DATA_REQ(received_message_p).stream,
                        SCTP_DATA_REQ(received_message_p).buffer,
                        SCTP_DATA_REQ(received_message_p).bufLen) < 0) {
        SCTP_ERROR("Failed to send message over SCTP\n");
      }
    }
    break;

    case MESSAGE_TEST: {
      //                 int i = 10000;
      //                 while(i--);
    } break;

    case TERMINATE_MESSAGE: {
      itti_exit_task();
    }
    break;

    default: {
      SCTP_DEBUG("Unkwnon message ID %d:%s\n",
                 ITTI_MSG_ID(received_message_p),
                 ITTI_MSG_NAME(received_message_p));
    }
    break;
    }

    itti_free(ITTI_MSG_ORIGIN_ID(received_message_p), received_message_p);
    received_message_p = NULL;
  }

  return NULL;
}

int sctp_init(const mme_config_t *mme_config_p)
{
  SCTP_DEBUG("Initializing SCTP task interface\n");

  memset(&sctp_desc, 0, sizeof(struct sctp_descriptor_s));

  /* Number of streams from configuration */
  sctp_desc.nb_instreams  = mme_config_p->sctp_config.in_streams;
  sctp_desc.nb_outstreams = mme_config_p->sctp_config.out_streams;

  if (itti_create_task(TASK_SCTP, &sctp_intertask_interface,
                       NULL) < 0) {
    SCTP_ERROR("create task failed");
    SCTP_DEBUG("Initializing SCTP task interface: FAILED\n");
    return -1;
  }

  SCTP_DEBUG("Initializing SCTP task interface: DONE\n");
  return 0;
}
