// MySqlTable.cxx

#include "dataset_sql/SelectQuery.h"
#include <iostream>
#include <sstream> 
#include <stdlib.h> 
#include <iomanip> 
#include "mysql++.h"  
#include "dataset_util/FileStatus.h"
#include "dataset_mysqlpp/MySqlTable.h"
#include "dataset_sql/SqlResult.h"

using std::vector;
using std::string;  
using std::map;
using std::cerr;
using std::endl;
using std::flush;
using std::ostringstream;  
using mysqlpp::Connection;
using mysqlpp::use_exceptions;
using mysqlpp::BadQuery;
using mysqlpp::Query;
using mysqlpp::Row;
using mysqlpp::Result;
using mysqlpp::BadConversion;
using dset::SqlRow;
using dset::SqlTable;
using dset::MySqlTable;
using dset::SqlResult;
using dset::SelectQuery;

typedef unsigned long Index;
typedef map<string,string> InsRow;

//**********************************************************************
// Local definitions.
//**********************************************************************

namespace {

void message(string method, string msg) {
  cerr << "MySqlTable::" << method << ":" << msg << endl;    
}

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

// Creator.

SqlTable* create(string tech, string name) {
  assert( tech == "MYSQL" );
  if ( tech != "MYSQL" ) return 0;
  string::size_type ipos, jpos;
  ipos = name.find(":");
  string db = name.substr(0, ipos);
  jpos = name.find(":", ipos+1); 
  string host = name.substr(ipos+1, jpos-(ipos+1));
  ipos = name.find(":", jpos+1);
  string table = name.substr(jpos+1, ipos-(jpos+1));
  jpos = name.find(":", ipos+1);
  string user = name.substr(ipos+1, jpos-(ipos+1));
  string pass = name.substr(jpos+1);
  MySqlTable *ptable = new MySqlTable(
    db.c_str(), host.c_str(), table, user.c_str(), pass.c_str()
  );
  return ptable;  
}

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

// Register creator.  

int STAT_MySqlTable = SqlTable::
  register_creator("MYSQL", create);

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

// Replace with "'" with "\'" in a string.
// Also replace "\" with "\\".

string fixquotes(string in) {
  string out;
  for ( string::size_type ipos=0; ipos!=in.size(); ++ipos ) {
    char ch = in[ipos];
    if ( ch == '\'' || ch == '\\' ) {
      out += '\\';
    }
    out += ch;
  }
  assert( out.size() >= in.size() );
  return out;
}

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

// Debug?

bool debug() {
  FileStatus fstat("debug_MySqlTable");
  return fstat.exists();
}

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

// Display query.

void showquery(string qstr) {
  if ( qstr.size() > 60 ) {
    cerr << qstr.substr(0,60) << "...  size: " << qstr.size();
  } else {
    cerr << qstr;
  }
  cerr << flush;
}

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

} // end unnamed namespace


//**********************************************************************
// Implementation class.
//**********************************************************************

class MySqlTable::Imp {
public:
  Connection* pcon;
  std::string table;
  mutable int error;
  bool valid;
  bool have_schema;
  SqlResult schema;
};

//**********************************************************************
// Public members.
//**********************************************************************

// Constructor.

MySqlTable::MySqlTable(string db_name, string host_name, string table, 
                       string user, string password)
: m_pimp(new Imp) {
  m_pimp->pcon = 0;
  m_pimp->table = table;
  m_pimp->error = 0;
  m_pimp->valid = true;
  m_pimp->pcon = new Connection(use_exceptions);
  m_pimp->have_schema = false;
  // Set connection timeout if zero.
  int maxtry = 5;
  int ntry = 0;
  bool connected = false;
  int ctimeout = m_pimp->pcon->get_options().connect_timeout;
  if ( ctimeout == 0 ) {
    int settime = 10;
    const char* csettime = (const char*) &settime;
    m_pimp->pcon->read_options(MYSQL_OPT_CONNECT_TIMEOUT, csettime);
    ctimeout = m_pimp->pcon->get_options().connect_timeout;
    assert( ctimeout == settime );
  }
  string serr = "timeout";
  bool dbg = debug();
  while ( ! connected && ntry++ < maxtry ) {
    if ( dbg ) {
      if ( ntry == 1 ) {
        cerr << "MySqlTable::ctor: Connecting to " << db_name << " at "
             <<  host_name;
      } else {
        cerr << "\n  connecting again";
      }
    }
    try {
      connected = m_pimp->pcon->connect(db_name.c_str(), host_name.c_str(),
                                user.c_str(), password.c_str());
    } catch (...) {
      serr = "exception";
    }
  }
  if ( dbg ) cerr << endl;
  if ( ! connected ) {
    message("ctor", "Connection error: " + serr);
    cerr << "        DB: " << db_name << endl;
    cerr << "      host: " << host_name << endl;
    cerr << "      user: " << user << endl;
    if ( password == "<pwlocal>" ) {
      cerr << "  password: " << password << endl;
    } else {
      cerr << "  password: ";
      for ( int ich=0; ich<password.size(); ++ich ) cerr << '*';
      cerr << endl;
    }
    m_pimp->valid = false;
    m_pimp->error = 2;
    delete m_pimp->pcon;
    m_pimp->pcon = 0;
  }
}

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

// Destructor.

MySqlTable::~MySqlTable() {
  if ( m_pimp->pcon != 0 ) {
    m_pimp->pcon->close();
    delete m_pimp->pcon;
    m_pimp->pcon = 0;  
  }
  delete m_pimp;
}

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

// Validity.

bool MySqlTable::is_valid() const {
  // Ping server twice.
  // Once to make sure there is a connection and once to check that we
  // do have a connection.
  bool dbg = debug();
  if ( dbg ) cerr << "MySqlTable::is_valid: Ping 1" << flush;
  if ( m_pimp->pcon == 0 ) {
    if ( dbg ) cerr << endl;
    return 0;
  }
  if ( dbg ) cerr << "\nMySqlTable::is_valid: Ping 2" << flush;
  m_pimp->pcon->ping();
  if ( m_pimp->pcon->ping() ) {
    message("is_valid", "Ping failed");
    if ( dbg ) cerr << " failed" << endl;
    return false;
  }
  if ( dbg ) cerr << " ok" << endl;
  return m_pimp->valid;
}

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

// Error code.

int MySqlTable::error() const {
  return m_pimp->error;
}

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

// Return a string describing the error.

string MySqlTable::error_string(int ecode) const {
  if ( ecode == 0 ) return "No error";
  if ( ecode == 1 ) return "Invalid query";
  if ( ecode == 2 ) return "Cound not connect to the host";
  if ( ecode == 3 ) return "No such element in repository";
  if ( ecode == 4 ) return "Empty query";
  if ( ecode == 5 ) return "No attribute id in query";
  return "Unknown error";
}

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

//  Displays the database schema so that the user can specify constraints
    
SqlResult MySqlTable::get_schema() const {

  // Return schema if we obtained them in an earlier call.
  // NO: these can change.
  //if ( m_pimp->have_schema ) {
  //  return m_pimp->schema;
  //}

  // Exit if this table is invalid.
  if ( ! is_valid() ) return SqlResult();

  // Initialize error code.
  m_pimp->error = 0;

  Query query = m_pimp->pcon->query();
  ostringstream oss; 
  oss << "describe ";
  oss << m_pimp->table; 
  string qstr = oss.str();
  query << qstr;
  SqlResult::NameList column_names;
  SqlRow row2insert;
  bool dbg = debug();
  try {
    if ( dbg ) showquery(qstr);
    Result res = query.store();
    for ( Result::size_type irow=0; irow<res.size(); ++irow ) {
      Row row = res[irow];
      column_names.push_back(string(row[0]));
      row2insert[string(row[0])] = row[1];
    }
  } catch( BadQuery er ) {
    message("get_schema", "Query error " + er.error);
    m_pimp->error = 1;
    return SqlResult();
  } catch ( BadConversion er ) {
    message("get_schema", "Conversion error");
    m_pimp->error = 1;
    return SqlResult();
  }
  if ( dbg ) cerr << endl;

  m_pimp->schema = SqlResult(column_names);
  m_pimp->schema.insert_row(row2insert);
  m_pimp->have_schema = true;
                                
  return m_pimp->schema;
}

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

// Return the modification time column.

string MySqlTable::modify_time_column() const {
  SqlResult::NameList cols = get_schema().get_col_names();
  if ( find(cols.begin(), cols.end(), "modtime") == cols.end() ) {
    return "";
  }
  return "unix_timestamp(modtime)";
}

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

// Return the creation time column.

string MySqlTable::create_time_column() const {
  SqlResult::NameList cols = get_schema().get_col_names();
  if ( find(cols.begin(), cols.end(), "cretime") == cols.end() ) {
    return "";
  }
  return "unix_timestamp(cretime)";
}

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

// Fetch the number of rows

int MySqlTable::get_row_count(SelectQuery sql) const{

  // Exit if this table is invalid.
  if ( ! is_valid() ) return 0;

  // Initialize error code.
  m_pimp->error = 0;

  Query query = m_pimp->pcon->query();
  string constraint;
  string scon = sql.get_constraint();
  if ( scon == "") {
    constraint = "";
  } else {
    string::size_type ipos = scon.find("limit");
    if ( ipos != string::npos) {
      if ( ipos == 0 ) {
        constraint = " " + scon;
      } else {
        constraint = " where " + scon;
      }
    } else {
      constraint = " where " + scon;
    }
  }

  ostringstream oss;	
  oss << "select count(*) from " << m_pimp->table << constraint;
  string qstr = oss.str();   
  query << qstr;
  Result res;    
  bool dbg = debug();
  try {
    if ( dbg ) showquery(qstr);
    res = query.store();
  } catch ( BadQuery er ) {
    message("get_row_count", "Query error " + er.error);
    return 0;
  } catch ( BadConversion er ) {
    message("get_row_count", "Conversion error");
    return 0;
  }
  if ( dbg ) cerr << endl;

  Row row = *res.begin();
  assert( row.size() );
  return row[0];
}

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

// Fetch the number of columns

int MySqlTable::get_col_count(SelectQuery sql) const{

  // Exit if this table is invalid.
  if ( ! is_valid() ) return 0;

  // Initialize error code.
  m_pimp->error = 0;

  string con = sql.get_constraint();
  if ( con.size() == 0 ) {
    SqlResult res = get_schema();
    return res.num_cols();
  }

  SqlResult res = execute_select_query(sql);
  if ( ! res.is_valid() ) {
    m_pimp->error = 1;
    return 0;
  }
  return res.num_cols();
}

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

// Execute an SQL SELECT query

SqlResult MySqlTable::execute_select_query(SelectQuery sql) const{

  // Exit if this table is invalid.
  if ( ! is_valid() ) return SqlResult();

  // Initialize error code.
  m_pimp->error = 0;

  Query query = m_pimp->pcon->query();
  SqlResult::NameList column_names;

  // Define "select" clause.
  string querystring1;
  if ( sql.get_columns() == "" ) {
    querystring1 = "select * from " + m_pimp->table; 
  } else {
    querystring1 = "select " + sql.get_columns() +
                   " from " + m_pimp->table; 
  }

  // Define "where" clause.
  string querystring2;
  string scon = sql.get_constraint();
  if ( scon == "" ) {
    querystring2 = "";
  } else {
    string::size_type ipos = scon.find("limit");
    if ( ipos != string::npos ) {
      if ( ipos == 0 ) {
        querystring2 = " " + scon;
      } else {
        querystring2 = " where " + scon;
      }
    } else {
      querystring2 = " where " + scon;
    }
  }

  // Construct the SQL query from "select" and "where" clauses.
  ostringstream oss;	
  oss << querystring1 << querystring2;
  string qstr = oss.str(); 
  query << qstr;

  // Contact the MySQL server and issue the query.
  Result res;    
  bool dbg = debug();
  try {
    if ( dbg ) showquery(qstr);
    res = query.store();
  } catch( BadQuery er ) {
    message("execute_select_query", "Query error " + er.error);
    m_pimp->error = 1;
    return SqlResult();
  } catch ( BadConversion er ) {
    message("execute_select_query", "Conversion error");
    m_pimp->error = 1;
    return SqlResult();
  }
  if ( dbg ) cerr << endl;

  if ( res.size() == 0 ){
    m_pimp->error = 3; 		
    return SqlResult();
  }
		
  Row row;
  Result::iterator i;
  i = res.begin();
  row = *i;

  for ( unsigned int i = 0; i<row.size(); i++ ) 
    column_names.push_back(res.names(i).c_str());
 	
  SqlResult result(column_names);	

  for ( i=res.begin(); i!=res.end(); i++ ) {
    row = *i;
    SqlRow row2insert;	
    for ( int j = 0; j < (int)row.size() ; j++ ) 
      row2insert[res.names(j).c_str()]=row[j];
    assert( ! result.insert_row(row2insert) );
  }

  return result;
}  

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

// Set an SQL INSERT query
// Returns 0 on success

int MySqlTable::execute_insert_query(InsertQuery sql) {
  	
  // Exit if this table is invalid.
  if ( ! is_valid() ) return 1;

  // Initialize error code.
  m_pimp->error = 0;

  Query query = m_pimp->pcon->query();
  SqlResult::NameList column_names;
  ostringstream fields;
  ostringstream values;
	
  fields << "insert into " ;
  fields <<  m_pimp->table ;
  fields <<  "(" ;
  values << "values(";

  InsRow row = sql.get_row();
  // Empty query
  if ( row.size() == 0) {
    m_pimp->error = 4;
    return 1;
  }
	             	
  // For each column name.
  for ( InsRow::const_iterator icol = row.begin(); 
        icol != row.end(); ++icol ){
    fields << icol->first;
    fields << ",";
    values << "'"; 
    values << fixquotes(icol->second); 
    values << "',";  
  }

  string first = fields.str(); 
  first.erase((first.size())-1);
  string second = values.str();
  second.erase((second.size())-1);
  string qstr = first + ") " + second + ")";   
  ostringstream oss;	
  oss <<  qstr;
  string q = oss.str();   
  query << q;
  Result res;    
  bool dbg = debug();
  try {
    if ( dbg ) showquery(qstr);
    res = query.store();
  } catch ( BadQuery er ) {
    message("execute_insert_query", "Query error " + er.error);
    message("execute_insert_query", "Query was: " + q);
    m_pimp->error = 1;
    return 1;
  } catch( BadConversion er ) {
    message("execute_insert_query", "Conversion error");
    m_pimp->error = 1;
    return 1;
  }
  if ( dbg ) cerr << endl;

  return 0;
}  

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

// Set an SQL UPDATE query
// Returns 0 on success

int MySqlTable::execute_update_query(InsertQuery sql, 
                                     SelectQuery selquery) {

  // Exit if this table is invalid.
  if ( ! is_valid() ) return 1;

  // Initialize error code.
  m_pimp->error = 0;

  InsRow temprow = sql.get_row();
  Query query = m_pimp->pcon->query();
  SqlResult::NameList column_names;
  ostringstream fields;
  ostringstream values;
	
  fields << "update " ;
  fields <<  m_pimp->table ;
  fields <<  " set " ;
  string quote = "'";
  string cond = selquery.get_constraint();
  string first = fields.str(); 
  string second = values.str();

  // For each column name.
  bool dbg = debug();
  for( InsRow::const_iterator miter = temprow.begin(); 
       miter != temprow.end(); ++miter ){
    string qstr = first + miter->first + " = " + "'" + miter->second + 
                  "'" + " where " + cond;   
    ostringstream oss;	
    oss <<  qstr;
    string q = oss.str();   
    query << q;
    Result res;    
    try {
      if ( dbg ) showquery(qstr);
      res = query.store();
    } catch( BadQuery er ) {
      message("execute_update_query", "Query error " + er.error);
      m_pimp->error = 1;
      return 1;
    } catch( BadConversion er ) {
      message("execute_update_query", "Conversion error");
      m_pimp->error = 1;
      return 1;
    }
    if ( dbg ) cerr << endl;
  }
  return 0;
}  

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

// Set an SQL DELETE query
// Returns 0 on success

int MySqlTable::execute_delete_query(DeleteQuery sql) {

  // Exit if this table is invalid.
  if ( ! is_valid() ) return 1;

  // Initialize error code.
  m_pimp->error = 0;

  Query query = m_pimp->pcon->query();
  SqlResult::NameList column_names;
  string querystring;
  querystring = "delete from " + m_pimp->table + " where " + 
                sql.get_constraint();
  ostringstream oss;	
  oss <<  querystring;
  string qstr = oss.str();   
  query << qstr;
  Result res;    
  bool dbg = debug();
  try {
    if ( dbg ) showquery(qstr);
   res = query.store();
  } catch ( BadQuery er ) {
    message("execute_delete_query", "Query error " + er.error);
    m_pimp->error = 1;
    return 1;
  }
  if ( dbg ) cerr << endl;
  return 0;
}  

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