/*

1.1.1 Class Implementation

----
This file is part of SECONDO.

Copyright (C) 2017,
Faculty of Mathematics and Computer Science,
Database Systems for New Applications.

SECONDO 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 2 of the License, or
(at your option) any later version.

SECONDO 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 SECONDO; if not, write to the Free Software
Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
----

*/
#include <cstdlib>

#include <boost/make_shared.hpp>

#include "SecondoException.h"

#include "Algebras/DBService2/CommunicationClient.hpp"
#include "Algebras/DBService2/DBServiceClient.hpp"
#include "Algebras/DBService2/DebugOutput.hpp"
#include "Algebras/DBService2/ReplicationClient.hpp"
#include "Algebras/DBService2/ReplicationServer.hpp"
#include "Algebras/DBService2/ReplicationUtils.hpp"
#include "Algebras/DBService2/Replicator.hpp"
#include "Algebras/DBService2/SecondoUtilsLocal.hpp"
#include "Algebras/DBService2/ServerRunnable.hpp"

#include <loguru.hpp>

using namespace std;

namespace DBService {

DBServiceClient::DBServiceClient()
{
    printFunction("DBServiceClient::DBServiceClient", std::cout);
    LOG_SCOPE_FUNCTION(INFO);

    if(!SecondoUtilsLocal::lookupDBServiceLocation(
            dbServiceHost,
            dbServicePort))
    {
        dbServiceHost = "";
        dbServicePort = "";
        throw new SecondoException("Unable to connect to DBService");
    }
    print("dbServiceHost", dbServiceHost, std::cout);
    print("dbServicePort", dbServicePort, std::cout);
    LOG_F(INFO, "DBService (Host: %s, Port: %s)",
          dbServiceHost.c_str(), dbServicePort.c_str());

    startReplicationServer();
}

void DBServiceClient::startReplicationServer()
{
    printFunction("DBServiceClient::startReplicationServer", std::cout);
    LOG_SCOPE_FUNCTION(INFO);

    string fileTransferPort;
    SecondoUtilsLocal::readFromConfigFile(fileTransferPort,
            "DBService",
            "FileTransferPort",
            "");
    print(fileTransferPort, std::cout);
    LOG_F(INFO, "FileTransferPort: %s", fileTransferPort.c_str());

    ServerRunnable replicationServer(atoi(fileTransferPort.c_str()));
    replicationServer.run<ReplicationServer>();
}

DBServiceClient* DBServiceClient::getInstance()
{
    printFunction("DBServiceClient::getInstance", std::cout);
    LOG_SCOPE_FUNCTION(INFO);

    if (!_instance)
    {
      try{
          _instance = new DBServiceClient();
      } catch(...){
          _instance = 0;
      }
    }
    return _instance;
}

bool DBServiceClient::triggerReplication(const std::string& databaseName,
                                            const std::string& relationName,
                                            const ListExpr relType,
                                            const bool async)
{
    //LOG_SCOPE_FUNCTION(INFO);
    printFunction("DBServiceClient::triggerReplication", std::cout);
    print("databaseName", databaseName, std::cout);
    print("relationName", relationName, std::cout);
    print(relType, std::cout);

    LOG_F(INFO, "Database: %s, Relation: %s",
          databaseName.c_str(), relationName.c_str());

    CommunicationClient dbServiceMasterClient(dbServiceHost,
                                              atoi(dbServicePort.c_str()),
                                              0);

    if(dbServiceMasterClient.triggerReplication(
            databaseName,
            relationName))
    {
        Replicator replicator(
                *(const_cast<string*>(&databaseName)),
                *(const_cast<string*>(&relationName)),
                relType);
        replicator.run(async);
    }
    return true;
}


bool DBServiceClient::triggerDerivation(const std::string& databaseName,
                                        const std::string& targetName,
                                        const std::string& relationName,
                                        const std::string& fundef)
{
    printFunction("DBServiceClient::triggerDerivation", std::cout);
    LOG_SCOPE_FUNCTION(INFO);

    print("databaseName", databaseName, std::cout);
    print("targetName", targetName, std::cout);
    print("relationName", relationName, std::cout);
    print("fundef", fundef, std::cout);

    LOG_F(INFO, "Database: %s, Target: %s, Relation: %s, Fundef: %s",
          databaseName.c_str(), targetName.c_str(), relationName.c_str(),
          fundef.c_str());

    CommunicationClient dbServiceMasterClient(dbServiceHost,
                                              atoi(dbServicePort.c_str()),
                                              0);

    dbServiceMasterClient.triggerDerivation(
            databaseName,
            targetName,
            relationName,
            fundef);
    return true;
}




bool DBServiceClient::getReplicaLocation(
        const string& databaseName,
        const string& relationName,
        const std::vector<std::string>& otherObjects,
        string& host,
        string& transferPort,
        string& commPort)
{
    LOG_SCOPE_FUNCTION(INFO);
    LOG_F(INFO, "databaseName: %s", databaseName.c_str());
    LOG_F(INFO, "relationName: %s", relationName.c_str());

    printFunction("DBServiceClient::getReplicaLocation", std::cout);
    print("databaseName", databaseName, std::cout);
    print("relationName", relationName, std::cout);

    CommunicationClient dbServiceMasterClient(dbServiceHost,
                                              atoi(dbServicePort.c_str()),
                                              0);

    return dbServiceMasterClient.getReplicaLocation(databaseName,
                                                    relationName,
                                                    otherObjects,
                                                    host,
                                                    transferPort,
                                                    commPort);
}

fs::path DBServiceClient::retrieveReplicaAndGetFileName(
                                const string& databaseName,
                                const string& relationName,
                                const vector<string>& otherObjects,
                                const string& functionAsNestedListString)
{
    LOG_SCOPE_FUNCTION(INFO);
    LOG_F(INFO, "databaseName: %s", databaseName.c_str());
    LOG_F(INFO, "relationName: %s", relationName.c_str());    
    LOG_F(INFO, "function: %s", functionAsNestedListString.c_str());
    
    printFunction("DBServiceClient::retrieveReplicaAndGetFileName", std::cout);
    print("databaseName", databaseName, std::cout);
    print("relationName", relationName, std::cout);
    print("other objects", otherObjects, std::cout);
    print("function", functionAsNestedListString, std::cout);

    string host;
    string transferPort;
    string dummyCommPort;
    
    if(!getReplicaLocation(
            databaseName, relationName, otherObjects, host, transferPort,
            dummyCommPort))
    {
        print("No replica available", std::cout);
        LOG_F(WARNING, "No replica available.");

        return string("");
    }

    print("host", host, std::cout);
    print("transferPort", transferPort, std::cout);
    LOG_F(INFO, "Host: %s, TransferPort: %s",
          host.c_str(), transferPort.c_str());

    fs::path localPathOnClient = ReplicationUtils::expandFilenameToAbsPath(
        ReplicationUtils::getFileName(databaseName,relationName));

    string filenameOnDBSWorker = 
        ReplicationUtils::getFileNameOnDBServiceWorker(databaseName,
        relationName);

    /*
     Use case: Transfer from DBS-W to a O-Node.
     Here the DBServiceClient is used to get the fileName of the file
     to be retrieved by the requesting node. (e.g. an O-W)
    */
    ReplicationClient clientToDBServiceWorker(
            host,
            atoi(transferPort.c_str()),
            localPathOnClient, /*local*/
            filenameOnDBSWorker, /*remote*/
            *(const_cast<string*>(&databaseName)),
            *(const_cast<string*>(&relationName)));

    // requestReplica sets the fileName passed into it.
    // requestReplica acts upon the remoteFile (2nd argument of the constructor)
    fs::path fileName;
    clientToDBServiceWorker.requestReplica(
            functionAsNestedListString,
            fileName,
            otherObjects);

    LOG_F(INFO, "retrieveReplicaAndGetFileName: %s", fileName.string().c_str());

    return fileName;
}

bool DBServiceClient::deleteReplicas(const string& databaseName,
                                     const string& relationName,
                                     const string& derivateName)
{
    printFunction("DBServiceClient::deleteReplicas", std::cout);
    LOG_SCOPE_FUNCTION(INFO);

    try{
       CommunicationClient dbServiceMasterClient(dbServiceHost,
                                              atoi(dbServicePort.c_str()),
                                              0);
       return dbServiceMasterClient.requestReplicaDeletion(
            databaseName,
            relationName,
            derivateName);
    } catch(...) {
       return false;
    }
}

bool DBServiceClient::pingDBService()
{
    printFunction("DBServiceClient::pingDBService", std::cout);
    LOG_SCOPE_FUNCTION(INFO);

    CommunicationClient dbServiceMasterClient(dbServiceHost,
                                              atoi(dbServicePort.c_str()),
                                              0);
    return dbServiceMasterClient.pingDBService();
}

bool DBServiceClient::getStreamType(
        const string& databaseName,
        const string& relationName,
        string& nestedListAsString)
{
    printFunction("DBServiceClient::getRelType", std::cout);
    LOG_SCOPE_FUNCTION(INFO);

    CommunicationClient dbServiceMasterClient(dbServiceHost,
                                              atoi(dbServicePort.c_str()),
                                              0);

    string dbServiceWorkerHost;
    string dummyTransferPort;
    string commPort;
    vector<string> dummy; // for extracting the type, we can ignore 
                          // other objects
    if(!dbServiceMasterClient.getReplicaLocation(
            databaseName,
            relationName,
            dummy,
            dbServiceWorkerHost,
            dummyTransferPort,
            commPort))
    {
        print("The relation does not exist in DBService", std::cout);
        LOG_F(WARNING, "The relation (%s, %s) does not exist in DBService.",
              databaseName.c_str(), relationName.c_str());

        return false;
    }

    CommunicationClient dbServiceWorkerClient(dbServiceWorkerHost,
                                              atoi(commPort.c_str()),
                                              0);

    return dbServiceWorkerClient.getRelType(
            RelationInfo::getIdentifier(databaseName, relationName),
            nestedListAsString);
}


bool DBServiceClient::getDerivedType(
        const string& databaseName,
        const string& relName,
        const string& derivedName,
        string& nestedListAsString)
{
    printFunction("DBServiceClient::getDerivedType", std::cout);
    LOG_SCOPE_FUNCTION(INFO);


    CommunicationClient dbServiceMasterClient(dbServiceHost,
                                              atoi(dbServicePort.c_str()),
                                              0);

    string dbServiceWorkerHost;
    string dummyTransferPort;
    string commPort;
    vector<string> derived; 
    derived.push_back(derivedName);
    if(!dbServiceMasterClient.getReplicaLocation(
            databaseName,
            relName,
            derived,
            dbServiceWorkerHost,
            dummyTransferPort,
            commPort))
    {
        print("Relation and/or Derived Object does not exist in DBService",
               std::cout);

        LOG_F(WARNING, "Relation and/or Derived Object (%s, %s) does "
                       "not exist in DBService",
                       databaseName.c_str(), relName.c_str());

        return false;
    }

    CommunicationClient dbServiceWorkerClient(dbServiceWorkerHost,
                                              atoi(commPort.c_str()),
                                              0);

    return dbServiceWorkerClient.getDerivedType(
            RelationInfo::getIdentifier(databaseName, relName),
            derivedName,
            nestedListAsString);
}



bool DBServiceClient::relationExists(
        const std::string& databaseName,
        const std::string& relationName)
{
    printFunction("DBServiceClient::relationExists", std::cout);
    LOG_SCOPE_FUNCTION(INFO);

    CommunicationClient dbServiceMasterClient(dbServiceHost,
                                              atoi(dbServicePort.c_str()),
                                              0);

    string dbServiceWorkerHost;
    string dummyTransferPort;
    string commPort;
    vector<string> dummy;
    if(!dbServiceMasterClient.getReplicaLocation(
            databaseName,
            relationName,
            dummy,
            dbServiceWorkerHost,
            dummyTransferPort,
            commPort))
    {
        print("The Relation does not exist in DBService", std::cout);
        LOG_F(WARNING, "The Relation (%s, %s) does not exist in DBService",
              databaseName.c_str(), relationName.c_str());

        return false;
    }
    return true;
}

bool DBServiceClient::allExists(
         const std::string& databaseName,
         const std::string& relationName,
         const std::vector<std::string>& derivates){

    printFunction(__PRETTY_FUNCTION__, std::cout);
    LOG_SCOPE_FUNCTION(INFO);

    CommunicationClient dbServiceMasterClient(dbServiceHost,
                                              atoi(dbServicePort.c_str()),
                                              0);

    string dbServiceWorkerHost;
    string dummyTransferPort;
    string commPort;
    if(!dbServiceMasterClient.getReplicaLocation(
            databaseName,
            relationName,
            derivates,
            dbServiceWorkerHost,
            dummyTransferPort,
            commPort))
    {
        print("Relation or a derivate does not exist in "
              "DBService at the same location", std::cout);

        LOG_F(WARNING, "Relation or a derivate (%s, %s) does not exist in "
                       "DBService at the same location",
                       databaseName.c_str(), relationName.c_str());
        return false;
    }
    return true;
}

bool DBServiceClient::addNode(
    const std::string& nodeHost,
    const int& nodePort,
    const std::string& pathToNodeConfig) {

    LOG_SCOPE_FUNCTION(INFO);
    LOG_F(INFO, "nodeHost: %s", nodeHost.c_str());
    LOG_F(INFO, "nodePort: %d", nodePort);
    LOG_F(INFO, "pathToNodeConfig: %s", pathToNodeConfig.c_str());

    // Connect to the DBService Master to submit the worker nodes'
    // details.
    CommunicationClient dbServiceMasterClient(dbServiceHost,
        atoi(dbServicePort.c_str()),
        0);    
    
    bool success = dbServiceMasterClient.addNode(
        nodeHost, nodePort, pathToNodeConfig);

    if (!success) {
        LOG_F(ERROR, "%s", "Couldn't addNode.");
        return false;
    }

    return true;
}



DBServiceClient* DBServiceClient::_instance = nullptr;

} /* namespace DBService */