/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include "ThinClientLocatorHelper.hpp"

#include <algorithm>
#include <set>

#include <geode/DataInput.hpp>
#include <geode/DataOutput.hpp>
#include <geode/SystemProperties.hpp>

#include "ClientConnectionRequest.hpp"
#include "ClientConnectionResponse.hpp"
#include "ClientReplacementRequest.hpp"
#include "LocatorListRequest.hpp"
#include "LocatorListResponse.hpp"
#include "QueueConnectionRequest.hpp"
#include "QueueConnectionResponse.hpp"
#include "TcpSslConn.hpp"
#include "ThinClientPoolDM.hpp"
namespace apache {
namespace geode {
namespace client {
const int BUFF_SIZE = 3000;

class ConnectionWrapper {
 private:
  Connector*& m_conn;

 public:
  explicit ConnectionWrapper(Connector*& conn) : m_conn(conn) {}
  ~ConnectionWrapper() {
    LOGDEBUG("closing the connection locator1");
    if (m_conn != nullptr) {
      LOGDEBUG("closing the connection locator");
      m_conn->close();
      delete m_conn;
    }
  }
};

ThinClientLocatorHelper::ThinClientLocatorHelper(
    const std::vector<std::string>& locatorAddresses,
    const ThinClientPoolDM* poolDM)
    : m_poolDM(poolDM) {
  for (auto&& locatorAddress : locatorAddresses) {
    m_locHostPort.emplace_back(locatorAddress);
  }
}

Connector* ThinClientLocatorHelper::createConnection(
    Connector*& conn, const char* hostname, int32_t port,
    std::chrono::microseconds waitSeconds, int32_t maxBuffSizePool) {
  Connector* socket = nullptr;
  auto& systemProperties = m_poolDM->getConnectionManager()
                               .getCacheImpl()
                               ->getDistributedSystem()
                               .getSystemProperties();
  if (systemProperties.sslEnabled()) {
    socket = new TcpSslConn(hostname, port, waitSeconds, maxBuffSizePool,
                            systemProperties.sslTrustStore().c_str(),
                            systemProperties.sslKeyStore().c_str(),
                            systemProperties.sslKeystorePassword().c_str());
  } else {
    socket = new TcpConn(hostname, port, waitSeconds, maxBuffSizePool);
  }
  conn = socket;
  socket->init();
  return socket;
}

GfErrType ThinClientLocatorHelper::getAllServers(
    std::vector<std::shared_ptr<ServerLocation> >& servers,
    const std::string& serverGrp) {
  ACE_Guard<ACE_Thread_Mutex> guard(m_locatorLock);

  auto& sysProps = m_poolDM->getConnectionManager()
                       .getCacheImpl()
                       ->getDistributedSystem()
                       .getSystemProperties();
  for (size_t i = 0; i < m_locHostPort.size(); i++) {
    auto& loc = m_locHostPort[i];
    try {
      LOGDEBUG("getAllServers getting servers from server = %s ",
               loc.getServerName().c_str());
      auto buffSize = m_poolDM->getSocketBufferSize();
      Connector* conn = nullptr;
      ConnectionWrapper cw(conn);
      createConnection(conn, loc.getServerName().c_str(), loc.getPort(),
                       sysProps.connectTimeout(), buffSize);
      auto request = std::make_shared<GetAllServersRequest>(serverGrp);
      auto data =
          m_poolDM->getConnectionManager().getCacheImpl()->createDataOutput();
      data.writeInt(static_cast<int32_t>(1001));  // GOSSIPVERSION
      data.writeObject(request);
      auto sentLength = conn->send(
          reinterpret_cast<char*>(const_cast<uint8_t*>(data.getBuffer())),
          data.getBufferLength(), m_poolDM->getReadTimeout());
      if (sentLength <= 0) {
        continue;
      }
      char buff[BUFF_SIZE];
      auto receivedLength =
          conn->receive(buff, BUFF_SIZE, m_poolDM->getReadTimeout());
      if (receivedLength <= 0) {
        continue;
      }

      auto di =
          m_poolDM->getConnectionManager().getCacheImpl()->createDataInput(
              reinterpret_cast<uint8_t*>(buff), receivedLength);

      if (di.read() == REPLY_SSL_ENABLED && !sysProps.sslEnabled()) {
        LOGERROR("SSL is enabled on locator, enable SSL in client as well");
        throw AuthenticationRequiredException(
            "SSL is enabled on locator, enable SSL in client as well");
      }
      di.rewindCursor(1);

      auto response =
          std::dynamic_pointer_cast<GetAllServersResponse>(di.readObject());
      servers = response->getServers();
      return GF_NOERR;
    } catch (const AuthenticationRequiredException&) {
      continue;
    } catch (const Exception& excp) {
      LOGFINE("Exception while querying locator: %s: %s",
              excp.getName().c_str(), excp.what());
      continue;
    }
  }
  return GF_NOERR;
}

GfErrType ThinClientLocatorHelper::getEndpointForNewCallBackConn(
    ClientProxyMembershipID& memId, std::list<ServerLocation>& outEndpoint,
    std::string&, int redundancy, const std::set<ServerLocation>& exclEndPts,
    const std::string& serverGrp) {
  ACE_Guard<ACE_Thread_Mutex> guard(m_locatorLock);
  auto& sysProps = m_poolDM->getConnectionManager()
                       .getCacheImpl()
                       ->getDistributedSystem()
                       .getSystemProperties();
  int locatorsRetry = 3;
  if (m_poolDM) {
    int poolRetry = m_poolDM->getRetryAttempts();
    locatorsRetry = poolRetry <= 0 ? locatorsRetry : poolRetry;
  }
  LOGFINER(
      "ThinClientLocatorHelper::getEndpointForNewCallBackConn locatorsRetry = "
      "%d ",
      locatorsRetry);
  for (unsigned attempts = 0;
       attempts <
       (m_locHostPort.size() == 1 ? locatorsRetry : m_locHostPort.size());
       attempts++) {
    ServerLocation loc;
    if (m_locHostPort.size() == 1) {
      loc = m_locHostPort[0];
    } else {
      loc = m_locHostPort[attempts];
    }

    try {
      LOGFINER("Querying locator at [%s:%d] for queue server from group [%s]",
               loc.getServerName().c_str(), loc.getPort(), serverGrp.c_str());
      auto buffSize = m_poolDM->getSocketBufferSize();
      Connector* conn = nullptr;
      ConnectionWrapper cw(conn);
      createConnection(conn, loc.getServerName().c_str(), loc.getPort(),
                       sysProps.connectTimeout(), buffSize);
      auto request = std::make_shared<QueueConnectionRequest>(
          memId, exclEndPts, redundancy, false, serverGrp);
      auto data =
          m_poolDM->getConnectionManager().getCacheImpl()->createDataOutput();
      data.writeInt(static_cast<int32_t>(1001));  // GOSSIPVERSION
      data.writeObject(request);
      auto sentLength = conn->send(
          reinterpret_cast<char*>(const_cast<uint8_t*>(data.getBuffer())),
          data.getBufferLength(), m_poolDM->getReadTimeout());
      if (sentLength <= 0) {
        continue;
      }
      char buff[BUFF_SIZE];
      auto receivedLength =
          conn->receive(buff, BUFF_SIZE, m_poolDM->getReadTimeout());
      if (receivedLength <= 0) {
        continue;
      }
      auto di =
          m_poolDM->getConnectionManager().getCacheImpl()->createDataInput(
              reinterpret_cast<uint8_t*>(buff), receivedLength);

      const auto acceptanceCode = di.read();
      if (acceptanceCode == REPLY_SSL_ENABLED && !sysProps.sslEnabled()) {
        LOGERROR("SSL is enabled on locator, enable SSL in client as well");
        throw AuthenticationRequiredException(
            "SSL is enabled on locator, enable SSL in client as well");
      }
      di.rewindCursor(1);
      auto response =
          std::dynamic_pointer_cast<QueueConnectionResponse>(di.readObject());
      outEndpoint = response->getServers();
      return GF_NOERR;
    } catch (const AuthenticationRequiredException& excp) {
      throw excp;
    } catch (const Exception& excp) {
      LOGFINE("Exception while querying locator: %s: %s",
              excp.getName().c_str(), excp.what());
      continue;
    }
  }
  throw NoAvailableLocatorsException("Unable to query any locators");
}

GfErrType ThinClientLocatorHelper::getEndpointForNewFwdConn(
    ServerLocation& outEndpoint, std::string&,
    const std::set<ServerLocation>& exclEndPts, const std::string& serverGrp,
    const TcrConnection* currentServer) {
  bool locatorFound = false;
  ACE_Guard<ACE_Thread_Mutex> guard(m_locatorLock);
  auto& sysProps = m_poolDM->getConnectionManager()
                       .getCacheImpl()
                       ->getDistributedSystem()
                       .getSystemProperties();

  auto poolRetry = m_poolDM->getRetryAttempts();
  auto locatorsRetry = poolRetry <= 0 ? 3 : poolRetry;
  LOGFINER(
      "ThinClientLocatorHelper::getEndpointForNewFwdConn locatorsRetry = %d ",
      locatorsRetry);
  for (unsigned attempts = 0;
       attempts <
       (m_locHostPort.size() == 1 ? locatorsRetry : m_locHostPort.size());
       attempts++) {
    ServerLocation serLoc;
    if (m_locHostPort.size() == 1) {
      serLoc = m_locHostPort[0];
    } else {
      serLoc = m_locHostPort[attempts];
    }
    try {
      LOGFINE("Querying locator at [%s:%d] for server from group [%s]",
              serLoc.getServerName().c_str(), serLoc.getPort(),
              serverGrp.c_str());
      auto buffSize = m_poolDM->getSocketBufferSize();
      Connector* conn = nullptr;
      ConnectionWrapper cw(conn);
      createConnection(conn, serLoc.getServerName().c_str(), serLoc.getPort(),
                       sysProps.connectTimeout(), buffSize);
      auto data =
          m_poolDM->getConnectionManager().getCacheImpl()->createDataOutput();
      data.writeInt(1001);  // GOSSIPVERSION
      if (currentServer == nullptr) {
        LOGDEBUG("Creating ClientConnectionRequest");
        std::shared_ptr<ClientConnectionRequest> request =
            std::make_shared<ClientConnectionRequest>(exclEndPts, serverGrp);
        data.writeObject(request);
      } else {
        LOGDEBUG("Creating ClientReplacementRequest for connection: ",
                 currentServer->getEndpointObject()->name().c_str());
        std::shared_ptr<ClientReplacementRequest> request =
            std::make_shared<ClientReplacementRequest>(
                currentServer->getEndpointObject()->name(), exclEndPts,
                serverGrp);
        data.writeObject(request);
      }
      auto sentLength = conn->send(
          reinterpret_cast<char*>(const_cast<uint8_t*>(data.getBuffer())),
          data.getBufferLength(), m_poolDM->getReadTimeout());
      if (sentLength <= 0) {
        continue;
      }
      char buff[BUFF_SIZE];
      auto receivedLength =
          conn->receive(buff, BUFF_SIZE, m_poolDM->getReadTimeout());
      if (receivedLength <= 0) {
        continue;  // return GF_EUNDEF;
      }
      auto di =
          m_poolDM->getConnectionManager().getCacheImpl()->createDataInput(
              reinterpret_cast<uint8_t*>(buff), receivedLength);

      const auto acceptanceCode = di.read();
      if (acceptanceCode == REPLY_SSL_ENABLED && !sysProps.sslEnabled()) {
        LOGERROR("SSL is enabled on locator, enable SSL in client as well");
        throw AuthenticationRequiredException(
            "SSL is enabled on locator, enable SSL in client as well");
      }
      di.rewindCursor(1);

      auto response =
          std::dynamic_pointer_cast<ClientConnectionResponse>(di.readObject());
      response->printInfo();
      if (!response->serverFound()) {
        LOGFINE("Server not found");
        locatorFound = true;
        continue;
      }
      outEndpoint = response->getServerLocation();
      LOGFINE("Server found at [%s:%d]", outEndpoint.getServerName().c_str(),
              outEndpoint.getPort());
      return GF_NOERR;
    } catch (const AuthenticationRequiredException& excp) {
      throw excp;
    } catch (const Exception& excp) {
      LOGFINE("Exception while querying locator: %s: %s",
              excp.getName().c_str(), excp.what());
      continue;
    }
  }

  if (locatorFound) {
    throw NotConnectedException("No servers found");
  } else {
    throw NoAvailableLocatorsException("Unable to query any locators");
  }
}

GfErrType ThinClientLocatorHelper::updateLocators(
    const std::string& serverGrp) {
  ACE_Guard<ACE_Thread_Mutex> guard(m_locatorLock);
  auto& sysProps = m_poolDM->getConnectionManager()
                       .getCacheImpl()
                       ->getDistributedSystem()
                       .getSystemProperties();

  for (size_t attempts = 0; attempts < m_locHostPort.size(); attempts++) {
    auto&& serLoc = m_locHostPort[attempts];
    Connector* conn = nullptr;
    try {
      auto buffSize = m_poolDM->getSocketBufferSize();
      LOGFINER("Querying locator list at: [%s:%d] for update from group [%s]",
               serLoc.getServerName().c_str(), serLoc.getPort(),
               serverGrp.c_str());
      ConnectionWrapper cw(conn);
      createConnection(conn, serLoc.getServerName().c_str(), serLoc.getPort(),
                       sysProps.connectTimeout(), buffSize);
      auto request = std::make_shared<LocatorListRequest>(serverGrp);
      auto data =
          m_poolDM->getConnectionManager().getCacheImpl()->createDataOutput();
      data.writeInt(static_cast<int32_t>(1001));  // GOSSIPVERSION
      data.writeObject(request);
      auto sentLength = conn->send(
          reinterpret_cast<char*>(const_cast<uint8_t*>(data.getBuffer())),
          data.getBufferLength(), m_poolDM->getReadTimeout());
      if (sentLength <= 0) {
        conn = nullptr;
        continue;
      }
      char buff[BUFF_SIZE];
      auto receivedLength =
          conn->receive(buff, BUFF_SIZE, m_poolDM->getReadTimeout());
      if (receivedLength <= 0) {
        continue;
      }
      auto di =
          m_poolDM->getConnectionManager().getCacheImpl()->createDataInput(
              reinterpret_cast<uint8_t*>(buff), receivedLength);

      const auto acceptanceCode = di.read();
      if (acceptanceCode == REPLY_SSL_ENABLED && !sysProps.sslEnabled()) {
        LOGERROR("SSL is enabled on locator, enable SSL in client as well");
        throw AuthenticationRequiredException(
            "SSL is enabled on locator, enable SSL in client as well");
      }
      di.rewindCursor(1);

      auto response =
          std::dynamic_pointer_cast<LocatorListResponse>(di.readObject());
      auto locators = response->getLocators();
      if (locators.size() > 0) {
        RandGen randGen;
        std::random_shuffle(locators.begin(), locators.end(), randGen);
      }
      std::vector<ServerLocation> temp(m_locHostPort.begin(),
                                       m_locHostPort.end());
      m_locHostPort.clear();
      m_locHostPort.insert(m_locHostPort.end(), locators.begin(),
                           locators.end());
      for (std::vector<ServerLocation>::iterator it = temp.begin();
           it != temp.end(); it++) {
        std::vector<ServerLocation>::iterator it1 =
            std::find(m_locHostPort.begin(), m_locHostPort.end(), *it);
        if (it1 == m_locHostPort.end()) {
          m_locHostPort.push_back(*it);
        }
      }
      return GF_NOERR;
    } catch (const AuthenticationRequiredException& excp) {
      throw excp;
    } catch (const Exception& excp) {
      LOGFINE("Exception while querying locator: %s: %s",
              excp.getName().c_str(), excp.what());
      continue;
    }
  }
  return GF_NOTCON;
}

}  // namespace client
}  // namespace geode
}  // namespace apache