// MagdaFileCatalog.cxx

#include "dataset_magda/MagdaFileCatalog.h"
#include <iostream>
#include <vector>
#include <set>
#include <map>
#include <sstream>
#include <unistd.h>
#include "dataset_util/getcwd.h"
#include "dataset_util/ssystem.h"
#include "dataset_util/copy_file.h"
#include "dataset_util/FileName.h"
#include "dataset_util/FileStatus.h"
#include "dataset_util/Environment.h"
#include "dataset_util/Text.h"
#include "dataset_util/XmlElement.h"
#include "dataset_util/pthread_sleep.h"
#include "mysql++.h"

using std::string;
using std::cerr;
using std::endl;
using std::istringstream;
using std::set;
using std::vector;
using std::multimap;
using std::make_pair;
using std::random_shuffle;
using mysqlpp::Connection;
using mysqlpp::BadQuery;
using mysqlpp::BadQuery;
using mysqlpp::use_exceptions;
using mysqlpp::Query;
using mysqlpp::Result;
using mysqlpp::Row;
using dset::FileCatalog;
using dset::MagdaFileCatalog;
using dset::Url;

typedef MagdaFileCatalog::Name Name;
typedef Text::WordList NameList;
typedef FileCatalog::UrlList UrlList;

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

namespace {

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

// Locator.
FileCatalog* mlocate(Name intype, Name inname) {
  string name = inname;
  string type = intype;
  if ( type == MagdaFileCatalog::static_type() ) {
  } else if ( type == MagdaFileCatalog::static_url_scheme() ) {
    if ( name != "atlas" ) {
      if ( name.substr(0, 8) != "//atlas/" ) return 0;
      name = "atlas";
    }
    type = MagdaFileCatalog::static_type();
  } else {
    return 0;
  }
  if ( type != MagdaFileCatalog::static_type() ) return 0;
  MagdaFileCatalog& fc = MagdaFileCatalog::instance();
  // Allow the old name "Atlas" for backward compatibility"
  if ( name != ""  && name != fc.name() && name != "Atlas" ) return 0;
  return &fc;
}

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

// Register locator.
int LOC = FileCatalog::
  register_locator(MagdaFileCatalog::static_type(), mlocate);
int LOCU = FileCatalog::
  register_locator(MagdaFileCatalog::static_url_scheme(), mlocate);

//**********************************************************************
}  // end unnamed namespace
//**********************************************************************
// Static member functions.
//**********************************************************************

// Single instance of this class.

MagdaFileCatalog& MagdaFileCatalog::instance() {
  static MagdaFileCatalog catalog;
  return catalog;
}

//**********************************************************************
// Member functions.
//**********************************************************************

// Constructor.

MagdaFileCatalog::MagdaFileCatalog() : m_connection(0) {
  string prefix = "MagdaFileCatalog::ctor: ";
  // Search envirnment for password.
  Environment env;
  env.sysfill();
  string kaimen = env.value("MAGDA_KAIMEN");
  // If that fails, try password file ~/.dial/pwmagda.
  if ( ! kaimen.size() ) {
    string home = env.value("HOME");
    if ( home.size() ) {
      Text pwfile(home + "/.dial/pwmagda");
        if ( pwfile.size() == 1 ) {
        string pwval =  pwfile.line(0);
        if ( pwval.size() ) {
          kaimen = pwval;
        }
      }
    }
  }
  // Connect to DB in RRW mode if password is found or in readonly
  // if not.
  string server = env.value("MAGDA_DBSERVER");
  if ( ! server.size() ) {
    server = "magda.usatlas.bnl.gov";
  }
  m_connection = new Connection(use_exceptions);
  string user = "reader";
  string pw = "";
  if ( kaimen.size() ) {
    user = "dbya";
    pw = kaimen;
  }
  // Set connection timeout if zero.
  int maxtry = 5;
  int ntry = 0;
  bool connected = false;
  int ctimeout = m_connection->get_options().connect_timeout;
  int count = 0;
  while ( ctimeout == 0 ) {
    int settime = 10;
    const char* csettime = (const char*) &settime;
    m_connection->read_options(MYSQL_OPT_CONNECT_TIMEOUT, csettime);
    ctimeout = m_connection->get_options().connect_timeout;
    if ( ctimeout == settime ) break;
    cerr << prefix << "Unable to set connection timeout "
         << "set=" << settime << " get=" << ctimeout << endl;
    pthread_sleep(1);
    if ( ++count > 10 ) {
      cerr << "  Giving up..." << endl;
      break;
    }
  }
  //assert( ctimeout == settime );
  string serr = "timeout";
  while ( ! connected && ntry++ < maxtry ) {
    try {
      connected = m_connection->
        connect("dbya", server.c_str(), user.c_str(), pw.c_str());
    } catch(BadQuery er) {
      serr = er.error;
    } catch (...) {
      serr = "unknown exception";
    }
  }
  if ( ! connected ) {
    message("MagdaFileCatalog::ctor: Connection error: " + serr);
    delete m_connection;
    m_connection = 0;
  } 
  if ( m_connection != 0 ) {
    m_host = env.value("MAGDA_HOST");
    m_cache = env.value("MAGDA_CACHE");
    if ( m_host.size() == 0 ) {
      message( "Warning: MAGDA_HOST has not been defined." );
    } else {
      Query query = m_connection->query();
      string qstr = "select * from host where host='"+m_host+"'";  
      query << qstr;
      try {
        Result res = query.store(); 
        int n = res.size();
        if( n < 1) {
          message( "Warning: MAGDA_HOST has not been set properly." );
        }
      } catch( BadQuery er ) {
        message( "Error: "+er.error );
      }
    }
  }
}

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

// Destructor.

MagdaFileCatalog::~MagdaFileCatalog() { 
  if( m_connection != 0 ) { 
    m_connection->close();
    delete m_connection;
    m_connection = 0;
  }
}

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

void MagdaFileCatalog::message(string msg) const {
  cerr << "MagdaFileCatalog: " << msg << endl;
}

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

// Catalog type.

Name MagdaFileCatalog::type() const {
  return static_type();
}

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

// Catalog name.

Name MagdaFileCatalog::name() const {
  return "atlas";
}

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

// Existence.

bool MagdaFileCatalog::has(Url lurl) {
  if ( m_connection == 0 ) return false;
  if ( ! lurl.is_valid() ) return false;
  Name prot = lurl.prefix();
  if ( prot != "lfn" ) return false;
  // Allow both lfn://somesite/name and lfn:name.
  Name name = lurl.basename();
  if ( name == "" ) {
    name = lurl.suffix();
  }
  m_connection->ping();
  Query query = m_connection->query();
  string qstr = "select path,name from prime where logical='"+name+"'";
  query << qstr;
  bool exist = false;
  try {
    Result res = query.store(); 
    int n = res.size();
    if ( n > 0 ) exist = true;
  } catch( BadQuery er ) {
    message( "Error: "+er.error );
    return false;
  }
  return exist;
}

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

// GUID.

Url MagdaFileCatalog::guid(Url lurl) {
  return Url();
}

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

// LFN.

Url MagdaFileCatalog::lfn(Url lurl) {
  if ( lurl.prefix() != "lfn" ) return Url();
  if ( ! has(lurl) ) return Url();
  return lurl;
}

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

// Get replicas.

UrlList MagdaFileCatalog::replicas(Url lurl) {
  string prefix = "MagdaFileCatalog::replicas: ";
  UrlList urls;
  if ( ! lurl.is_valid() ) return urls;
  Name prot = lurl.prefix();
  if ( prot != "lfn" ) return urls;
  // Allow both lfn://somesite/name and lfn:name.
  Name name = lurl.basename();
  if ( name == "" ) {
    name = lurl.suffix();
  }
  m_connection->ping();
  Query query = m_connection->query();
  string qstr = "select path,name,site from prime where logical='"+name+"'";
  query << qstr;
  try {
    Result res = query.store(); 
    for ( Result::iterator irow=res.begin();
          irow!=res.end(); ++irow) {
      // Fetch directory, basename and site from magda.
      Row row = *irow;
      string directory = row[0].get_string();
      string basename = row[1].get_string();
      string site = row[2].get_string();
      string fullpath = directory + "/" + basename;
      // Assume if the file exists here, we have the correct file.
      // Could be wrong.
      if ( site == "usatlas" && FileStatus(fullpath).exists() ) {
        urls.push_back(Url("file:" + fullpath));
      }
      // For known sites, create a gsiftp URL.
      string xfprefix;
      if ( site == "usatlas" ) {
        xfprefix = "gsiftp://aftpexp02.bnl.gov";
      // Don't wait for castor to stage files.
      //} else if ( site == "cerncastor" ) {
        //xfprefix = "gsiftp://castorgrid.cern.ch";
      }
      if ( xfprefix.size() ) {
        urls.push_back(Url(xfprefix + fullpath));
      }
    }
  } catch( BadQuery er ) {
    message( "Error: "+er.error );
    return urls;
  }
  return urls;
}

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

// Insert.

Url MagdaFileCatalog::insert(Url, Url, Time) {
  if ( m_connection == 0 ) return Url();
  return Url();
}

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

// Remove file.

int MagdaFileCatalog::remove(Url, Url, bool) {
  return 99;
}

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