/*
 *  Functions to handle incoming data
 */

#include "rpsdist.h"
/*
 *  Parse a line at a time.
 *  haven't decided  how to put together an entire 'meta-object', but ideally:
 *  RETURNS:
 *      -1    on error
 *       0    on no key fields found
 *       1    pipe rest of text
 *    Should be built in a hierarchy.  Key fields checked here, then jump to
 *    proper function to handle that object.
 *    Eventually this function will be replaced by adding the recognition directly 
 *    into the lexer/parser, for just these attributes.
 */

int parse_transaction(int fd){
  char curline[4096], size[100], method[100], size2[100];
  char cumul_buffer[4096 * 4];
  int cumul_length = 0;
  int length = 0;
  long temp_int1, temp_int2;
  regmatch_t args[5];
  struct initial_transaction tran;
  peer_t * temp_peer;
  size[0] = '\0';
  method[0] = '\0';
  size2[0] = '\0';
  
  /********************** check for transaction-begin: *********************************/
  length = read_socket_line(default_trace, fd, curline, 4096);
  if(!regexec(&trans_begin, curline,  5, args, 0) ){
    memcpy(size, args[1].rm_sp, (args[1].rm_eo - args[1].rm_so));
    size[(args[1].rm_eo - args[1].rm_so)] = '\0';
    printf("match, trans begin: %s\n", size);
    
    length = read_socket_line(default_trace, fd, curline, 4096);
    if(!regexec(&trans_method, curline,  5, args, 0) ){ /* check for method */
      memcpy(method, args[1].rm_sp, (args[1].rm_eo - args[1].rm_so));
      method[(args[1].rm_eo - args[1].rm_so)] = '\0';
      printf("match, trans_method: %s\n", method);
      
      length = read_socket_line(default_trace, fd, curline, 4096);
      if(!regexec(&blank_line, curline, 0, 0, 0)) /* check for blank line */
	decode_transaction(size, method, fd);
      else
	printf("No blank line between trans-begin object and rest\n");
    }
    else{  /* no method */
      sprintf(method, "plain");
      if(!regexec(&blank_line, curline, 0, 0, 0)) /* check if it was blank */
	decode_transaction(size, method, fd);
    }
    return 1;
  } 
  
  /***************************  check if transaction-submit-begin: ***********************/
  else if(!regexec(&trans_sub_begin, curline, 5, args, 0) ){ 
    memcpy(tran.dbname, args[1].rm_sp, (args[1].rm_eo - args[1].rm_so));
    tran.dbname[(args[1].rm_eo - args[1].rm_so)] = '\0';
    memcpy(size, args[3].rm_sp, (args[3].rm_eo - args[3].rm_so));
    size[(args[3].rm_eo - args[3].rm_so)] = '\0';
    tran.ident = atoi(size);
    printf("\nmatch, trans_sub_begin: %s, size: %d\n", tran.dbname, tran.ident);
    
    length = read_socket_line(default_trace, fd, curline, 4096);
    if(!regexec(&resp_auth_type, curline, 5, args, 0) ){ /* check for a response type */
      printf("Do what with resp type?\n");
      
      length = read_socket_line(default_trace, fd, curline, 4096);
      if(!regexec(&trans_conf_type, curline, 5, args, 0) ){ /* check confirm type */
	memcpy(method, args[1].rm_sp, (args[1].rm_eo - args[1].rm_so));
	method[(args[1].rm_eo - args[1].rm_so)] = '\0';
	if(!strcasecmp(method, "none"))
	  tran.confirm_type = NONE;
	else if(!strcasecmp(method, "commit"))
	  tran.confirm_type = COMMIT;
	else if(!strcasecmp(method, "normal"))
	  tran.confirm_type = NORMAL;
	else if(!strcasecmp(method, "legacy"))
	  tran.confirm_type = LEGACY;
	
	length = read_socket_line(default_trace, fd, curline, 4096);
	if(!regexec(&blank_line, curline, 0, 0, 0) ) /* check for blank line */
	  handle_new_submit(fd, &tran);
	else
	  printf("missing a blank line\n");
      }
      else if(!regexec(&blank_line, curline, 0, 0, 0) ) /* check for blank line */
	  handle_new_submit(fd, &tran);
      else
	printf("Missing a blank line\n");
    }
    else{ /* if not, check if it was a confirm type */
      tran.response_type = 0;
      if(!regexec(&trans_conf_type, curline, 5, args, 0) ){ /* check confirm type */
	memcpy(method, args[1].rm_sp, (args[1].rm_eo - args[1].rm_so));
	method[(args[1].rm_eo - args[1].rm_so)] = '\0';
	if(!strcasecmp(method, "none"))
	  tran.confirm_type = NONE;
	else if(!strcasecmp(method, "commit"))
	  tran.confirm_type = COMMIT;
	else if(!strcasecmp(method, "normal"))
	  tran.confirm_type = NORMAL;
	else if(!strcasecmp(method, "legacy"))
	  tran.confirm_type = LEGACY;
      }
      else{ /* if neither, should be a blank line */
	if(!regexec(&blank_line, curline, 0, 0, 0) ) /* check for blank line */
	  handle_new_submit(fd, &tran);
	else
	  printf("Missing a blank line\n");
      }
    }
    return 1;
  }
  /*********************** check if transaction-request: ******************************/
  else if(!regexec(&trans_request, curline, 5, args, 0) ){
    memcpy(method, args[1].rm_sp, (args[1].rm_eo - args[1].rm_so));
    method[(args[1].rm_eo - args[1].rm_so)] = '\0';
    /* compare method to db_name? */
    
    length = read_socket_line(default_trace, fd, curline, 4096);
    if( !regexec(&seq_begin, curline, 5, args, 0) ){
      memcpy(size, args[1].rm_sp, (args[1].rm_eo - args[1].rm_so));
      size[(args[1].rm_eo - args[1].rm_so)] = '\0';
    }
    else if( !regexec(&seq_end, curline, 5, args, 0) ){
      memcpy(size2, args[1].rm_sp, (args[1].rm_eo - args[1].rm_so));
      size2[(args[1].rm_eo - args[1].rm_so)] = '\0';
    }
    else if( !regexec(&blank_line, curline, 0, 0, 0) ){ /* check for blank line */

      /* do action and return */

    }
    
    length = read_socket_line(default_trace, fd, curline, 4096);
    if( !regexec(&seq_begin, curline, 5, args, 0) ){
      if(strlen(size) == 0){
	memcpy(size, args[1].rm_sp, (args[1].rm_eo - args[1].rm_so));
	size[(args[1].rm_eo - args[1].rm_so)] = '\0';
      }
      else{
	printf("Cannot have two sequence-begin lines\n");
	return 0;
      }
    }
    else if( !regexec(&seq_end, curline, 5, args, 0) ){
      if(strlen(size2) == 0){
	memcpy(size2, args[1].rm_sp, (args[1].rm_eo - args[1].rm_so));
	size2[(args[1].rm_eo - args[1].rm_so)] = '\0';
      }
      else{
	printf("Cannot have two sequence-end lines\n");
	return 0;
      }
    }
    else if( !regexec(&blank_line, curline, 0, 0, 0) ){ 
      
      /* do action and return */
      
    }
    
    length = read_socket_line(default_trace, fd, curline, 4096);
    if( !regexec(&blank_line, curline, 0, 0, 0) ){ 
      
      /* do action and return */

    }
    return 1;
  }

  /**********************  check heartbeat: ************************/
  else if( !regexec(&heartbeat, curline, 5, args, 0) ){ 
    memcpy(method, args[1].rm_sp, (args[1].rm_eo - args[1].rm_so));
    method[(args[1].rm_eo - args[1].rm_so)] = '\0';
    memcpy(cumul_buffer+cumul_length, curline, length);
    cumul_length += length;
    
    length = read_socket_line(default_trace, fd, curline, 4096);
    if( !regexec(&sequence, curline, 5, args, 0) ){ 
       memcpy(size, args[1].rm_sp, (args[1].rm_eo - args[1].rm_so));
       size[(args[1].rm_eo - args[1].rm_so)] = '\0';
       memcpy(cumul_buffer+cumul_length, curline, length);
       cumul_length += length;
    }
    else{
      printf("Invalid heartbeat object, expected sequence\n");
      return 0;
    }
    
    length = read_socket_line(default_trace, fd, curline, 4096);
    if( !regexec(&timestamp, curline, 5, args, 0) ){
       memcpy(size2, args[1].rm_sp, (args[1].rm_eo - args[1].rm_so));
       size2[(args[1].rm_eo - args[1].rm_so)] = '\0';
       if(check_timestamp(size2) < 0){
	 printf("ignoring heartbeat\n");
	 return 0;
       }
       memcpy(cumul_buffer+cumul_length, curline, length);
       cumul_length += length;
    }
    else{
      printf("Invalid timestamp\n");
      return 0;
    }
    
    printf("Got heartbeat: %s, %s, %s\n", method, size, size2);
    /* process: method, size, size2 */
    
    if( strcasecmp(method, DB_NAME) )  /* avoid forwarding stuff we originated..BAD */
      send_to_peers(cumul_buffer, cumul_length, -1);
    
    return 1;
  }
  
  /******************** check snapshot-request *********************/
  else if( !regexec(&snap_req, curline, 5, args, 0) ){
    memcpy(method, args[1].rm_sp, (args[1].rm_eo - args[1].rm_so));
    method[(args[1].rm_eo - args[1].rm_so)] = '\0';
    
    /* do it */
    return 1;
  }
   

  /********************** temp: check for add-peer ***********/
  else if(!regexec(&add_peer, curline, 5, args, 0) ){
    memcpy(method, args[1].rm_sp, (args[1].rm_eo - args[1].rm_so));
    method[(args[1].rm_eo - args[1].rm_so)] = '\0';
    memcpy(size, args[2].rm_sp, (args[2].rm_eo - args[2].rm_so));
    size[(args[2].rm_eo - args[2].rm_so)] = '\0';
    
    if(find_peer_obj(&peer_list, method, atoi(size))){
      printf("Peer already exists\n");
    }
    else
      /* create a thread, open connection, handle connection() on it */
      /*add_peer_obj(&peer_list, 
	create_peer_obj(open_connection(default_trace,method, atoi(size)), method, atoi(size)));*/
      add_peer_obj(&peer_list, (peer_t*)spawn_new(open_connection(default_trace,method, atoi(size)), method, atoi(size)));
  }
  /********************* temp: rm-peer ********************/
  else if(!regexec(&rm_peer, curline, 5, args, 0) ){
    memcpy(method, args[1].rm_sp, (args[1].rm_eo - args[1].rm_so));
    method[(args[1].rm_eo - args[1].rm_so)] = '\0';
    memcpy(size, args[2].rm_sp, (args[2].rm_eo - args[2].rm_so));
    size[(args[2].rm_eo - args[2].rm_so)] = '\0';
    if(find_peer_obj(&peer_list, method, atoi(size))){
      close(find_peer_obj(&peer_list, method, atoi(size))->fd);
      delete_peer_obj(&peer_list, find_peer_obj(&peer_list, method, atoi(size)));
    }
    else
      write(fd, "** Doesn't exist **\n", 14);
  }
  else if(!regexec(&sh_peer, curline, 5, args, 0) ){
    temp_peer = peer_list.first;
    while(temp_peer){
      printf("Host: %s, Port: %d, FD: %d\n", temp_peer->host, temp_peer->port, temp_peer->fd);
      temp_peer = temp_peer->next;
    }
  }
  if (length < 0)
    return -1;
}
  
/*
 * dump the rest to a file, and process
 */
int decode_transaction(char * size, char * method, int fd){
  char buffer[atoi(size)];
  char * orig_f_name = strdup(tmpnam(NULL));
  char * pln_f_name = strdup(tmpnam(NULL));
  int orig_file_d;
  int pln_file_d;
  int cum_size = 0;
  int cur_rd_size = 0;
  int expect_size = atoi(size);
  fd_set read_fds;
  struct timeval tv;
  peer_t * curr_obj = NULL;

  tv.tv_sec = 5; 
  tv.tv_usec = 0;
  FD_ZERO(&read_fds);
  FD_SET(fd, &read_fds);
  
  if(expect_size == 0){
    printf("size error\n");
    return 0;
  }
  
  orig_file_d = open(orig_f_name, O_RDWR|O_CREAT, S_IRUSR|S_IWUSR);
 
  /* dump to file */
  printf("-------ORIGINAL SUBMISSION--------\n");
  while(cum_size < expect_size){
    if(select (fd+1, &read_fds, NULL, NULL, &tv)){
      if((cur_rd_size = read(fd, buffer, expect_size)) > 0){
	if(write(orig_file_d, buffer, cur_rd_size) <= 0){
	  printf("rps-dist: write error to file\n");
	  exit(1);
	}
      }
    }
    else
      break;
    cum_size += cur_rd_size;
    printf("Cumulative size: %d\n", cum_size);
  }
  if(expect_size != cum_size){
    /* check for a heartbeat */
    write(STDOUT_FILENO, buffer, cur_rd_size);
    printf("Unexpected read error, got(%d), should be(%d)\n", cum_size, expect_size);
    return -1;
  }
  
  /* if it's gzip, decode to another file...pln_file_d */
 if(!strcasecmp(method, "gzip")){
   lseek(orig_file_d, 0, SEEK_SET);
   pln_file_d = open(pln_f_name, O_RDWR|O_CREAT, S_IRUSR|S_IWUSR);
   unzip_stream(orig_file_d, pln_file_d);
   close(orig_file_d);
 }
 else
   pln_file_d = orig_file_d;
 
 lseek(pln_file_d, 0, SEEK_SET);
 if(decapsulate_transaction(pln_file_d) > 0){
   write(STDOUT_FILENO, "about to forward trans\n", 23);
   lseek(pln_file_d, 0, SEEK_SET);
   encode_transaction(pln_file_d, ENCODING);
 }

 close(pln_file_d);

 if(remove(orig_f_name) < 0)
   printf("decode: Unable to remove original %s\n", orig_f_name);
  free(orig_f_name);

  if(!strcasecmp(method, "gzip")){
    if(remove(pln_f_name) < 0)
      printf("decode: Unable to remove plain %s\n", pln_f_name);
  }

 free(pln_f_name);
 
 printf("----------------------------------\n");
 
 return 1;
}

/*
 * ack
 */
int handle_new_submit(int fd, struct initial_transaction * tran){
  char buffer[4096];
  int length = 0;
  char * pln_f_name = strdup(tmpnam(NULL));
  int pln_file_d = open(pln_f_name,O_CREAT|O_RDWR, S_IRUSR|S_IWUSR);
  int last_was_blank = 0;

  /* dump to a plaintext file */
  while( (length = read_socket_line(default_trace, fd, buffer, 4096)) > 0){
    if(!regexec(&trans_sub_end, buffer, 0, 0, 0)){ /* if we get a end, close */
      if(last_was_blank)
	break;
      else{
	printf("Error: missing blank line before end\n");
	return -1;
      }
    }
    if(!regexec(&blank_line, buffer, 0, 0, 0)) /* if a blank line */
      last_was_blank = 1;
    else
      last_was_blank = 0;
    if(write(pln_file_d, buffer, length) <= 0)
      printf("encode: error writing plain file\n");
  }
  
  lseek(pln_file_d, 0, SEEK_SET);
  
  /*if(auth_check(pln_file_d)) { */
  
  encapsulate_transaction(pln_file_d);

  confirm_transaction(fd, tran);
 
  if(remove(pln_f_name) < 0)
    printf("encode: Unable to remove plain file %s\n", pln_f_name);
  free(pln_f_name);
}
  

/* 
 * ACK!!
 */
int decapsulate_transaction(int fd){
  char buffer[4096];
  regmatch_t args[5];
  char arg1[100];
  char arg2[100];
  long temp1;
  int length = 0;

  length = read_socket_line(default_trace, fd, buffer, 4096);
  /************** Check for transaction-label *******************/
  if(!regexec(&trans_label, buffer, 5, args, 0) ){
    memcpy(arg1, args[1].rm_sp, (args[1].rm_eo - args[1].rm_so));
    arg1[(args[1].rm_eo - args[1].rm_so)] = '\0';
    
    printf("Compare: %s and %s\n", arg1, DB_NAME);
    if( !strcasecmp(arg1, DB_NAME)){
      printf("Not forwarding my own transaction\n");
      return -1;
    }
    length = read_socket_line(default_trace, fd, buffer, 4096);
    /* Check for sequence */
    if(!regexec(&sequence, buffer, 5, args, 0) ){
      memcpy(arg1, args[1].rm_sp, (args[1].rm_eo - args[1].rm_so));
      arg1[(args[1].rm_eo - args[1].rm_so)] = '\0';
      temp1 = atoi(arg1);
    }      
    
    length = read_socket_line(default_trace, fd, buffer, 4096);
    if(!regexec(&timestamp, buffer, 5, args, 0) ){
      memcpy(arg1, args[1].rm_sp, (args[1].rm_eo - args[1].rm_so));
      arg1[(args[1].rm_eo - args[1].rm_so)] = '\0';
      if(check_timestamp(arg1) < 0){
	printf("Invalid timestamp on transaction\n");
	return -1;
      }
    }
    
    length = read_socket_line(default_trace, fd, buffer, 4096);
    if(!regexec(&integrity, buffer, 5, args, 0) ){
      memcpy(arg1, args[1].rm_sp, (args[1].rm_eo - args[1].rm_so));
      arg1[(args[1].rm_eo - args[1].rm_so)] = '\0';
      
      length = read_socket_line(default_trace, fd, buffer, 4096);
      if(!regexec(&blank_line, buffer, 0, 0, 0) );
      else{
	printf("Missing a blank line\n");
      return -1;
      } 
    }
    else if(!regexec(&blank_line, buffer, 0, 0, 0) );
    else{
      printf("Missing a blank line\n");
      return -1;
    }
      
  }
  else{
    write(STDOUT_FILENO, buffer, length);
    printf("Invalid object\n");
    return -1;
  }
  
  return 1;
}  
      
  
