#include "hosts.h"
#include "msg.h"
#include "online.h"
#include "parse.h"
#include <poll.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unistd.h>

extern char *f_host; // parsing data for hosts.

extern struct host *conns[1024]; // connection data for each peer.

extern int num_conns; // number of connected peers.

extern int f; // maximum number of potentially faulty nodes.

extern char *glob_name; // my name (not necessary here).

extern uint32_t glob_replica_index; // my replica index.

extern struct msg_log *mlog;

struct msg_log *msg_reqs =
    NULL; // message requests received from a pre-prepare.

extern struct msg_log *outstanding;   // outstanding requests.
extern struct msg_log *out_responses; // outstanding responses.

uint32_t state_machine_state = 0; // initialize state machine state to 0.

uint32_t view_next_seq_num = 1;

uint32_t low_watermark = 0;

uint32_t high_watermark = 300;

uint32_t curr_view = 0;

uint32_t accepted_req_num = 0;

// Initialize host connections.
void initialize() {
  init_hosts(f_host);

  printf("\n\n\n --- CONNECTIONS --- \n\n");
  for (int i = 0; i < num_conns; i++) {
    printf("Connected with %s:%d at %d\n", conns[i]->name, conns[i]->c_port,
           conns[i]->h_port);
  }
}

// Start up peer.
int main(int argc, char **argv) {
  srand(time(NULL));
  parse(argc, argv);

  printf("Got host file: %s\n", f_host);

  initialize();

  uint32_t request_num = 1;

  // Take requests from stdin: we want to poll on stdin and incoming messages.

  struct pollfd *pollfds = malloc(sizeof(struct pollfd) * ((num_conns) + 1));

  int infd = fileno(stdin);

  pollfds[0].fd = infd;
  pollfds[0].events = POLLIN | POLLPRI;

  for (int i = 0; i < num_conns; i++) {
    pollfds[i + 1].fd = conns[i]->sockfd;
    pollfds[i + 1].events = POLLIN | POLLPRI;
  }

  printf("%d > ", request_num);
  fflush(stdout);

  struct timespec start_time;
  clock_gettime(CLOCK_REALTIME, &start_time);

  int timeout = 30; // this should be fixed for a view change.

  while (1) {

    struct timespec now;
    clock_gettime(CLOCK_MONOTONIC, &now);
    if (poll(pollfds, num_conns + 1, timeout) != 0) {
      if (pollfds[0].revents > 0) {
        // update timeouts.

        // handle stdin request for state change.
        char *str = NULL;
        size_t len;
        if (getline(&str, &len, stdin) != -1) {
          bool proper_request = false;
          struct msg *req = init_msg();
          req->msg_type = 0; // request
          if (strncmp("ADD", str, strlen("ADD")) == 0) {
            char *argument = str + 4;
            uint32_t x = atoi(argument);
            req->req_message = 0;
            req->option_value = x;
            proper_request = true;
          }
          if (strncmp("SUB", str, strlen("SUB")) == 0) {
            char *argument = str + 4;
            uint32_t x = atoi(argument);
            req->req_message = 1;
            req->option_value = x;
            proper_request = true;
          }
          if (strncmp("MUL", str, strlen("MUL")) == 0) {
            char *argument = str + 4;
            uint32_t x = atoi(argument);
            req->req_message = 2;
            req->option_value = x;
            proper_request = true;
          }
          if (strncmp("GET", str, strlen("GET")) == 0) {
            req->req_message = 3;
            proper_request = true;
          }

          if (proper_request) {
            struct timespec creq_timestamp;
            clock_gettime(CLOCK_MONOTONIC, &creq_timestamp);
            req->timestamp = creq_timestamp.tv_sec;
            req->client_index = glob_replica_index;
            packet *p = parse_msg_as_packet(req);
            if (curr_view % (num_conns + 1) == glob_replica_index) {
              // case: we are the primary.
              handle_primary_recv_request(req, todo_key());
              struct msg_log *out_msg = malloc(sizeof(struct msg_log));
              out_msg->m = req;
              out_msg->m->seq = view_next_seq_num;
              out_msg->next = msg_reqs;
              msg_reqs = out_msg;
              view_next_seq_num += 1;
            } else {
              // case: we are a client sending to primary.
              send_to_replica(conns, num_conns, curr_view % (num_conns + 1), p);
            }
            clean_packet(p);
            struct msg_log *ml = malloc(sizeof(struct msg_log));
            struct msg *standing = malloc(sizeof(struct msg));
            memcpy(standing, req, sizeof(struct msg));
            standing->client_index = request_num;
            ml->m = standing;
            ml->m->client_index = request_num;
            ml->next = outstanding;
            outstanding = ml;
            request_num += 1;
          }
        }
      }

      for (int i = 0; i < num_conns; i++) {
        if (pollfds[i + 1].revents > 0) {
          // handle message from peer at this point.
          uint8_t msglen;
          socklen_t addr_len = sizeof(conns[i]->conn);
          recvfrom(conns[i]->sockfd, &msglen, sizeof(msglen), MSG_PEEK,
                   (struct sockaddr *)&(conns[i]->conn), &addr_len);
          uint8_t *msg_buf = malloc(msglen + sizeof(msglen));
          recvfrom(conns[i]->sockfd, msg_buf, msglen + sizeof(msglen), 0,
                   (struct sockaddr *)&(conns[i]->conn), &addr_len);

          uint8_t *new_data = malloc(msglen);
          memcpy(new_data, msg_buf + 1, msglen);
          free(msg_buf);
          packet *p = malloc(sizeof(packet));
          p->is_timed_out = false;
          p->len = msglen;
          p->data = new_data;
          struct msg *message = parse_packet_as_msg(p);
          clean_packet(p);

          if (message->msg_type == 0) {
            // printf("\nRequest\n");
            // request for message coming from a client
            // check that we are the primary
            if (curr_view % (num_conns + 1) == glob_replica_index) {
              handle_primary_recv_request(message, todo_key());
              struct msg_log *out_msg = malloc(sizeof(struct msg_log));
              out_msg->m = message;
              out_msg->m->seq = view_next_seq_num;
              out_msg->next = msg_reqs;
              msg_reqs = out_msg;
              view_next_seq_num += 1;
            }
          } else if (message->msg_type == 1) {
            // printf("\nPreprepare\n");
            // pre-prepare message sent out
            //
            // 1. parse the pre-prepare message.
            //
            // 2. receive the piggybacked message.
            //
            // validate and accept pre-prepare message.

            // receive the piggybacked message.
            uint8_t msglen;
            socklen_t addr_len = sizeof(conns[i]->conn);
            recvfrom(conns[i]->sockfd, &msglen, sizeof(msglen), MSG_PEEK,
                     (struct sockaddr *)&(conns[i]->conn), &addr_len);
            uint8_t *msg_buf = malloc(msglen + sizeof(msglen));
            recvfrom(conns[i]->sockfd, msg_buf, msglen + sizeof(msglen), 0,
                     (struct sockaddr *)&(conns[i]->conn), &addr_len);

            uint8_t *new_data = malloc(msglen);
            memcpy(new_data, msg_buf + 1, msglen);
            free(msg_buf);
            packet *p = malloc(sizeof(packet));
            p->is_timed_out = false;
            p->len = msglen;
            p->data = new_data;
            struct msg *piggyback = parse_packet_as_msg(p);

            uint32_t d = small_digest(piggyback);

            bool check1 = d == message->digest;
            bool check2 = validate_signature(message, todo_key());
            bool check3 = message->view == curr_view;
            bool check4 = true;
            struct msg_log *m;
            for (m = mlog; m != NULL; m = m->next) {
              if (m->m->msg_type == 1 && m->m->view == message->view &&
                  m->m->seq == message->seq &&
                  m->m->digest != message->digest) {
                check4 = false;
              }
            }
            bool check5 = (low_watermark <= message->seq &&
                           message->seq <= high_watermark);
            if (check1 && check2 && check3 && check4 && check5) {
              struct msg_log *ml = malloc(sizeof(struct msg_log));
              ml->m = message;
              ml->next = mlog;
              struct msg_log *ml2 = malloc(sizeof(struct msg_log));
              ml2->m = piggyback;
              ml2->next = ml;
              mlog = ml2;

              struct msg_log *out_msg = malloc(sizeof(struct msg_log));
              struct msg *cloned_msg = malloc(sizeof(struct msg));
              memcpy(cloned_msg, piggyback, sizeof(struct msg));
              cloned_msg->seq = message->seq;
              out_msg->m = cloned_msg;
              out_msg->next = msg_reqs;
              msg_reqs = out_msg;

              // now that the message log has been updated, we should broadcast
              // prepare messages.
              handle_send_prepare(message, todo_key());

              struct msg_log *out;

              for (out = msg_reqs; out != NULL; out = out->next) {
                if (prepared(mlog, f, out->m, message->view, message->seq)) {
                  handle_send_commit(out->m, message, todo_key());
                }
              }
            }
          } else if (message->msg_type == 2) {
            // printf("\nPrepare\n");
            // receive a prepare message.
            // 1. validate the prepare message.
            // 2. check if this prepare message completes the `prepared` for any
            // currently waiting message.
            // if it does, send out a commit!
            if (message->view == curr_view && low_watermark <= message->seq &&
                message->seq <= high_watermark &&
                validate_signature(message, todo_key())) {

              struct msg_log *out_msg = malloc(sizeof(struct msg_log));
              out_msg->m = message;
              out_msg->next = mlog;
              mlog = out_msg;

              struct msg_log *out;

              for (out = msg_reqs; out != NULL; out = out->next) {
                if (prepared(mlog, f, out->m, message->view, message->seq)) {
                  handle_send_commit(out->m, message, todo_key());
                }
              }
            }
          } else if (message->msg_type == 3) {
            // printf("\nCommit\n");
            // someone committed.
            // validate committed message.

            // struct msg_log *loop;
            // for (loop = mlog; loop != NULL; loop = loop->next) {
            //   display_msg(loop->m);
            //   fflush(stdout);
            // }

            if (message->view == curr_view && low_watermark <= message->seq &&
                message->seq <= high_watermark &&
                validate_signature(message, todo_key())) {
              struct msg_log *out_msg = malloc(sizeof(struct msg_log));
              out_msg->m = message;
              out_msg->next = mlog;
              mlog = out_msg;

              struct msg_log *out;

              uint32_t lowest_seq = UINT32_MAX;

              struct msg_log *prev_to_curr;
              struct msg_log *curr;
              struct msg *curr_handle;

              bool valid = false;

              for (out = msg_reqs; out != NULL; out = out->next) {
                if (out->next != NULL &&
                    committed_local(mlog, f, out->next->m, message->view,
                                    message->seq)) {

                  if (out->next->m->seq < lowest_seq) {
                    lowest_seq = out->next->m->seq;
                    curr_handle = out->next->m;
                    prev_to_curr = out;
                    curr = out->next;
                    valid = true;
                  }
                  // [HERE] check if all reqs w lower sequence numbers have been
                  // established.
                }
              }

              if (msg_reqs != NULL &&
                  committed_local(mlog, f, msg_reqs->m, message->view,
                                  message->seq)) {
                if (msg_reqs->m->seq < lowest_seq) {
                  lowest_seq = msg_reqs->m->seq;
                  curr_handle = msg_reqs->m;
                  prev_to_curr = NULL;
                  curr = msg_reqs;
                  valid = true;
                }
              }

              if (valid == true && prev_to_curr != NULL && curr != NULL) {
                printf("\nTest1\n");
                prev_to_curr->next = curr->next;
                printf("\nTest2\n");
              } else {
                if (valid == true && msg_reqs != NULL) {
                  msg_reqs = msg_reqs->next;
                }
              }

              if (valid) {
                handle_operation(curr_handle, todo_key());
              }
            }
          } else if (message->msg_type == 5) {
            // printf("\nResponse\n");

            if (validate_signature(message, todo_key())) {

              struct msg_log *out_msg = malloc(sizeof(struct msg_log));
              out_msg->m = message;
              out_msg->next = out_responses;
              out_responses = out_msg;
            }
            struct msg_log *out;

            uint32_t resp = message->option_value;

            uint8_t count = 1;

            for (out = out_responses; out != NULL; out = out->next) {
              if (out->m->option_value == resp &&
                  out->m->timestamp == message->timestamp) {
                count += 1;
              }
            }

            uint32_t req_num = 0;

            if (count > f) {
              struct msg_log *out;
              struct msg_log *curr;

              for (out = outstanding; out != NULL; out = out->next) {
                curr = out->next;

                if (curr != NULL && curr->m->timestamp == message->timestamp) {
                  out->next = curr->next;
                  req_num = curr->m->client_index;
                  clean_msg(curr->m);
                  free(curr);
                }
              }

              if (outstanding != NULL &&
                  outstanding->m->timestamp == message->timestamp) {
                req_num = outstanding->m->client_index;
                outstanding = outstanding->next;
              }

              if (req_num > accepted_req_num) {

                struct msg_log *loop;
                // for (loop = mlog; loop != NULL; loop = loop->next) {
                //   display_msg(loop->m);
                //   fflush(stdout);
                // }

                printf("\nRESP [%d] > %d \n", req_num, resp);
                fflush(stdout);
                printf("%d > ", request_num);
                fflush(stdout);
                accepted_req_num = req_num;
              }
            }
          }
        }
      }

    } else {
      // handle view change.
    }
  }

  // stdin msg -> forward to primary (check if it is us), start online phase
  // other messages: keep message log, and validate as we proceed
}