catalog/src/query/query-adapter.hpp

/** NDN-Atmos: Cataloging Service for distributed data originally developed
 *  for atmospheric science data
 *  Copyright (C) 2015 Colorado State University
 *
 *  NDN-Atmos 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 3 of the License, or
 *  (at your option) any later version.
 *
 *  NDN-Atmos 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 NDN-Atmos.  If not, see <http://www.gnu.org/licenses/>.
**/

#ifndef ATMOS_QUERY_QUERY_ADAPTER_HPP
#define ATMOS_QUERY_QUERY_ADAPTER_HPP

#include "util/catalog-adapter.hpp"
#include "util/mysql-util.hpp"
#include "util/config-file.hpp"

#include <thread>

#include <json/reader.h>
#include <json/value.h>
#include <json/writer.h>

#include <ndn-cxx/data.hpp>
#include <ndn-cxx/face.hpp>
#include <ndn-cxx/interest.hpp>
#include <ndn-cxx/interest-filter.hpp>
#include <ndn-cxx/name.hpp>
#include <ndn-cxx/security/key-chain.hpp>
#include <ndn-cxx/util/time.hpp>
#include <ndn-cxx/encoding/encoding-buffer.hpp>
#include <ndn-cxx/util/in-memory-storage-lru.hpp>
#include <ndn-cxx/util/string-helper.hpp>
#include <ChronoSync/socket.hpp>

#include "mysql/mysql.h"

#include <map>
#include <unordered_map>
#include <memory>
#include <mutex>
#include <sstream>
#include <string>
#include <array>
#include <utility>

#include "util/logger.hpp"


namespace atmos {
namespace query {
#ifdef HAVE_LOG4CXX
  INIT_LOGGER("QueryAdapter");
#endif

// todo: calculate payload limit by get the size of a signed empty Data packet
static const size_t PAYLOAD_LIMIT = 7000;

/**
 * QueryAdapter handles the Query usecases for the catalog
 */
template <typename DatabaseHandler>
class QueryAdapter : public atmos::util::CatalogAdapter {
public:
  /**
   * Constructor
   *
   * @param face:       Face that will be used for NDN communications
   * @param keyChain:   KeyChain that will be used for data signing
   * @param syncSocket: ChronoSync socket
   */
  QueryAdapter(const std::shared_ptr<ndn::Face>& face,
               const std::shared_ptr<ndn::KeyChain>& keyChain,
               const std::shared_ptr<chronosync::Socket>& syncSocket);

  virtual
  ~QueryAdapter();

  /**
   * Helper function to specify section handler
   */
  void
  setConfigFile(util::ConfigFile& config,
                const ndn::Name& prefix,
                const std::vector<std::string>& nameFields,
                const std::string& databaseTable);

protected:
  /**
   * Helper function for configuration parsing
   */
  void
  onConfig(const util::ConfigSection& section,
           bool isDryDun,
           const std::string& fileName,
           const ndn::Name& prefix);

  /**
   * Handles incoming query requests by stripping the filter off the Interest to get the
   * actual request out. This removes the need for a 2-step Interest-Data retrieval.
   *
   * @param filter:   InterestFilter that caused this Interest to be routed
   * @param interest: Interest that needs to be handled
   */
  virtual void
  onIncomingQueryInterest(const ndn::InterestFilter& filter, const ndn::Interest& interest);

  /**
   * Handles requests for responses to an filter initialization request
   *
   * @param interest: Interest that needs to be handled
   */
  virtual void
  onFiltersInitializationInterest(std::shared_ptr<const ndn::Interest> interest);

  /**
   * Helper function that generates query results from a Json query carried in the Interest
   *
   * @param interest:  Interest that needs to be handled
   */
  void
  populateFiltersMenu(std::shared_ptr<const ndn::Interest> interest);

  void
  getFiltersMenu(Json::Value& value);

  /**
   * Helper function that makes query-results data
   *
   * @param segmentPrefix:  Name that identifies the Prefix for the Data
   * @param value:          Json::Value to be sent in the Data
   * @param segmentNo:      uint64_t the segment for this Data
   * @param isFinalBlock:   bool to indicate whether this needs to be flagged in the Data as the
   *                         last entry
   * @param isAutocomplete: bool to indicate whether this is an autocomplete message
   * @param resultCount:    the number of records in the query results
   * @param viewStart:      the start index of the record in the query results payload
   * @param viewEnd:        the end index of the record in the query results payload
   * @param lastComponent:  flag to indicate the content contains the last component for
                            autocompletion query
   */
  std::shared_ptr<ndn::Data>
  makeReplyData(const ndn::Name& segmentPrefix,
                const Json::Value& value,
                uint64_t segmentNo,
                bool isFinalBlock,
                bool isAutocomplete,
                uint64_t resultCount,
                uint64_t viewStart,
                uint64_t viewEnd,
                bool lastComponent);

  /**
   * Helper function that generates query results from a Json query carried in the Interest
   *
   * @param interest:  Interest that needs to be handled
   */
  void
  runJsonQuery(std::shared_ptr<const ndn::Interest> interest);

  /**
   * Helper function that makes ACK data
   *
   * @param interest: Intersts that needs to be handled
   * @param version:  Version that needs to be in the data name
   */
  std::shared_ptr<ndn::Data>
  makeAckData(std::shared_ptr<const ndn::Interest> interest,
              const ndn::Name::Component& version);

  /**
   * Helper function that sends NACK
   *
   * @param dataPrefix: prefix for the data packet
   */
  void
  sendNack(const ndn::Name& dataPrefix);

  /**
   * Helper function that signs the data
   */
  void
  signData(ndn::Data& data);

  /**
   * Helper function that publishes query-results data segments
   */
  virtual void
  prepareSegmentsBySqlString(const ndn::Name& segmentPrefix,
                             const std::string& sqlString,
                             bool lastComponent,
                             const std::string& nameField);

  virtual void
  prepareSegmentsByParams(std::vector<std::pair<std::string, std::string>>& queryParams,
                          const ndn::Name& segmentPrefix);

  void
  generateSegments(ResultSet_T& res,
                   const ndn::Name& segmentPrefix,
                   int resultCount,
                   bool autocomplete,
                   bool lastComponent);

  /**
   * Helper function to set the DatabaseHandler
   */
  void
  setDatabaseHandler(const util::ConnectionDetails&  databaseId);

  void
  closeDatabaseHandler();

  /**
   * Helper function that set filters to make the adapter work
   */
  void
  setFilters();

  void
  setCatalogId();

  /**
   * Helper function that generates the sqlQuery string for autocomplete query
   * @param sqlQuery:      stringstream to save the sqlQuery string
   * @param jsonValue:     Json value that contains the query information
   * @param lastComponent: Flag to mark the last component query
   * @param nameField:     stringstream to save the nameField string
   */
  bool
  json2AutocompletionSql(std::stringstream& sqlQuery,
                         Json::Value& jsonValue,
                         bool& lastComponent,
                         std::stringstream& nameField);

  bool
  doPrefixBasedSearch(Json::Value& jsonValue,
                      std::vector<std::pair<std::string, std::string>>& typedComponents);

  bool
  doFilterBasedSearch(Json::Value& jsonValue,
                      std::vector<std::pair<std::string, std::string>>& typedComponents);

  ndn::Name
  getQueryResultsName(std::shared_ptr<const ndn::Interest> interest,
                      const ndn::Name::Component& version);

  std::string
  getChronoSyncDigest();

protected:
  typedef std::unordered_map<ndn::Name, const ndn::RegisteredPrefixId*> RegisteredPrefixList;
  // Handle to the Catalog's database
  std::shared_ptr<DatabaseHandler> m_dbConnPool;
  const std::shared_ptr<chronosync::Socket>& m_socket;

  // mutex to control critical sections
  std::mutex m_mutex;
  // @{ needs m_mutex protection
  // The Queries we are currently writing to
  ndn::util::InMemoryStorageLru m_activeQueryToFirstResponse;
  ndn::util::InMemoryStorageLru m_cache;
  std::string m_chronosyncDigest;
  // @}
  RegisteredPrefixList m_registeredPrefixList;
  ndn::Name m_catalogId; // should be replaced with the PK digest
  std::vector<std::string> m_filterCategoryNames;
};

template <typename DatabaseHandler>
QueryAdapter<DatabaseHandler>::QueryAdapter(const std::shared_ptr<ndn::Face>& face,
                                            const std::shared_ptr<ndn::KeyChain>& keyChain,
                                            const std::shared_ptr<chronosync::Socket>& syncSocket)
  : util::CatalogAdapter(face, keyChain)
  , m_socket(syncSocket)
  , m_activeQueryToFirstResponse(100000)
  , m_cache(250000)
  , m_chronosyncDigest("0")
  , m_catalogId("catalogIdPlaceHolder") // initialize for unitests
{
}

template <typename DatabaseHandler>
void
QueryAdapter<DatabaseHandler>::setFilters()
{
  m_registeredPrefixList[m_prefix] = m_face->setInterestFilter(ndn::InterestFilter(m_prefix),
                            bind(&query::QueryAdapter<DatabaseHandler>::onIncomingQueryInterest,
                                 this, _1, _2),
                            bind(&query::QueryAdapter<DatabaseHandler>::onRegisterSuccess,
                                 this, _1),
                            bind(&query::QueryAdapter<DatabaseHandler>::onRegisterFailure,
                                 this, _1, _2));
}

template <typename DatabaseHandler>
void
QueryAdapter<DatabaseHandler>::setConfigFile(util::ConfigFile& config,
                                             const ndn::Name& prefix,
                                             const std::vector<std::string>& nameFields,
                                             const std::string& databaseTable)
{
  m_nameFields = nameFields;
  m_databaseTable = databaseTable;
  config.addSectionHandler("queryAdapter", bind(&QueryAdapter<DatabaseHandler>::onConfig, this,
                                                _1, _2, _3, prefix));
}

template <typename DatabaseHandler>
void
QueryAdapter<DatabaseHandler>::onConfig(const util::ConfigSection& section,
                                        bool isDryRun,
                                        const std::string& filename,
                                        const ndn::Name& prefix)
{
  using namespace util;
  if (isDryRun) {
    return;
  }
  std::string signingId, dbServer, dbName, dbUser, dbPasswd;
  for (auto item = section.begin();
       item != section.end();
       ++item)
  {
    if (item->first == "signingId") {
      signingId = item->second.get_value<std::string>();
      if (signingId.empty()) {
        throw Error("Empty value for \"signingId\""
                    " in \"query\" section");
      }
    }
    if (item->first == "filterCategoryNames") {
      std::istringstream ss(item->second.get_value<std::string>());
      std::string token;
      while(std::getline(ss, token, ',')) {
        m_filterCategoryNames.push_back(token);
      }
    }
    if (item->first == "database") {
      const util::ConfigSection& dataSection = item->second;
      for (auto subItem = dataSection.begin();
           subItem != dataSection.end();
           ++subItem)
      {
        if (subItem->first == "dbServer") {
          dbServer = subItem->second.get_value<std::string>();
        }
        if (subItem->first == "dbName") {
          dbName = subItem->second.get_value<std::string>();
        }
        if (subItem->first == "dbUser") {
          dbUser = subItem->second.get_value<std::string>();
        }
        if (subItem->first == "dbPasswd") {
          dbPasswd = subItem->second.get_value<std::string>();
        }
      }

      if (dbServer.empty()){
        throw Error("Invalid value for \"dbServer\""
                    " in \"query\" section");
      }
      if (dbName.empty()){
        throw Error("Invalid value for \"dbName\""
                    " in \"query\" section");
      }
      if (dbUser.empty()){
        throw Error("Invalid value for \"dbUser\""
                    " in \"query\" section");
      }
      if (dbPasswd.empty()){
        throw Error("Invalid value for \"dbPasswd\""
                    " in \"query\" section");
      }
    }
  }

  if (m_filterCategoryNames.empty()) {
    throw Error("Empty value for \"filterCategoryNames\" in \"query\" section");
  }

  m_prefix = prefix;

  m_signingId = ndn::Name(signingId);
  setCatalogId();

  util::ConnectionDetails mysqlId(dbServer, dbUser, dbPasswd, dbName);
  setDatabaseHandler(mysqlId);
  setFilters();
}

template <typename DatabaseHandler>
void
QueryAdapter<DatabaseHandler>::setCatalogId()
{
  //empty
}

template <>
void
QueryAdapter<ConnectionPool_T>::setCatalogId()
{
  // use public key digest as the catalog ID
  ndn::Name keyId;
  if (m_signingId.empty()) {
    keyId = m_keyChain->getDefaultKeyNameForIdentity(m_keyChain->getDefaultIdentity());
  } else {
    keyId = m_keyChain->getDefaultKeyNameForIdentity(m_signingId);
  }

  std::shared_ptr<ndn::PublicKey> pKey = m_keyChain->getPib().getPublicKey(keyId);
  ndn::Block keyDigest = pKey->computeDigest();
  m_catalogId.clear();
  m_catalogId.append(ndn::toHex(*keyDigest.getBuffer()));
}

template <typename DatabaseHandler>
void
QueryAdapter<DatabaseHandler>::setDatabaseHandler(const util::ConnectionDetails& databaseId)
{
  //empty
}

template <>
void
QueryAdapter<ConnectionPool_T>::setDatabaseHandler(const util::ConnectionDetails& databaseId)
{
  m_dbConnPool = zdbConnectionSetup(databaseId);
}

template <typename DatabaseHandler>
void
QueryAdapter<DatabaseHandler>::closeDatabaseHandler()
{
}

template <>
void
QueryAdapter<ConnectionPool_T>::closeDatabaseHandler()
{
  ConnectionPool_stop(*m_dbConnPool);
}


template <typename DatabaseHandler>
QueryAdapter<DatabaseHandler>::~QueryAdapter()
{
  for (const auto& itr : m_registeredPrefixList) {
    if (static_cast<bool>(itr.second))
      m_face->unsetInterestFilter(itr.second);
  }

  closeDatabaseHandler();
}

template <typename DatabaseHandler>
void
QueryAdapter<DatabaseHandler>::onIncomingQueryInterest(const ndn::InterestFilter& filter,
                                                       const ndn::Interest& interest)
{
  _LOG_DEBUG(">> QueryAdapter::onIncomingQueryInterest");

  // Interest must carry component "initialization" or "query"
  if (interest.getName().size() < filter.getPrefix().size()) {
    // must NACK incorrect interest
    sendNack(interest.getName());
    return;
  }

  _LOG_DEBUG("Interest : " << interest.getName());
  std::shared_ptr<const ndn::Interest> interestPtr = interest.shared_from_this();

  if (interest.getName()[filter.getPrefix().size()] == ndn::Name::Component("filters-initialization")) {
    std::thread queryThread(&QueryAdapter<DatabaseHandler>::onFiltersInitializationInterest,
                            this,
                            interestPtr);
    queryThread.detach();
  }
  else if (interest.getName()[filter.getPrefix().size()] == ndn::Name::Component("query")) {

    auto data = m_cache.find(interest);
    if (data) {
      m_face->put(*data);
      return;
    }

    // catalog must strip sequence number in an Interest for further process
    if (interest.getName().size() > (filter.getPrefix().size() + 2)) {
      // Interest carries sequence number, only grip the main part
      // e.g., /hep/query/<query-params>/<version>/#seq
      ndn::Interest queryInterest(interest.getName().getPrefix(filter.getPrefix().size() + 2));

      auto data = m_cache.find(queryInterest);
      if (data) {
        // catalog has generated some data, but still working on it
        return;
      }
      interestPtr = std::make_shared<ndn::Interest>(queryInterest);
    }

    std::thread queryThread(&QueryAdapter<DatabaseHandler>::runJsonQuery,
                            this,
                            interestPtr);
    queryThread.detach();
  }

  // ignore other Interests
}

template <typename DatabaseHandler>
void
QueryAdapter<DatabaseHandler>::onFiltersInitializationInterest(std::shared_ptr<const ndn::Interest> interest)
{
  _LOG_DEBUG(">> QueryAdapter::onFiltersInitializationInterest");

  if(m_socket != nullptr) {
    const ndn::ConstBufferPtr digestPtr = m_socket->getRootDigest();
    std::string digestStr = ndn::toHex(digestPtr->buf(), digestPtr->size());
    _LOG_DEBUG("Original digest :" << m_chronosyncDigest);
    _LOG_DEBUG("New digest : " << digestStr);
    // if the m_chronosyncDigest and the rootdigest are not equal
    if (digestStr != m_chronosyncDigest) {
      // (1) update chronosyncDigest
      // (2) clear all staled ACK data
      m_mutex.lock();
      m_chronosyncDigest = digestStr;
      m_activeQueryToFirstResponse.erase(ndn::Name("/"));
      m_mutex.unlock();
      _LOG_DEBUG("Change digest to " << m_chronosyncDigest);
    }
  }

  auto data = m_activeQueryToFirstResponse.find(*interest);
  if (data) {
    m_face->put(*data);
  }
  else {
    populateFiltersMenu(interest);
  }

  _LOG_DEBUG("<< QueryAdapter::onFiltersInitializationInterest");
}

template <typename DatabaseHandler>
void
QueryAdapter<DatabaseHandler>::populateFiltersMenu(std::shared_ptr<const ndn::Interest> interest)
{
  _LOG_DEBUG(">> QueryAdapter::populateFiltersMenu");
  Json::Value filters;
  Json::FastWriter fastWriter;
  getFiltersMenu(filters);

  const std::string filterValue = fastWriter.write(filters);

  if (!filters.empty()) {
    // use /<prefix>/filters-initialization/<seg> as data name
    ndn::Name filterDataName(interest->getName().getPrefix(-1));

    const char* payload = filterValue.c_str();
    size_t payloadLength = filterValue.size();
    size_t startIndex = 0, seqNo = 0;

    if (filterValue.length() > PAYLOAD_LIMIT) {
      payloadLength = PAYLOAD_LIMIT;
      ndn::Name segmentName = ndn::Name(filterDataName).appendSegment(seqNo);
      std::shared_ptr<ndn::Data> filterData = std::make_shared<ndn::Data>(segmentName);
      // freshnessPeriod 0 means permanent?
      filterData->setFreshnessPeriod(ndn::time::milliseconds(10));
      filterData->setContent(reinterpret_cast<const uint8_t*>(payload + startIndex), payloadLength);

      signData(*filterData);

      _LOG_DEBUG("Populate Filter Data :" << segmentName);

      m_mutex.lock();
      // save the filter results in the activeQueryToFirstResponse structure
      // when version changes, the activeQueryToFirstResponse should be cleaned
      m_activeQueryToFirstResponse.insert(*filterData);
      try {
        m_face->put(*filterData);
      }
      catch (std::exception& e) {
        _LOG_ERROR(e.what());
      }
      m_mutex.unlock();

      seqNo++;
      startIndex = payloadLength * seqNo + 1;
    }
    payloadLength = filterValue.size() - PAYLOAD_LIMIT * seqNo;

    ndn::Name lastSegment = ndn::Name(filterDataName).appendSegment(seqNo);
    std::shared_ptr<ndn::Data> filterData = std::make_shared<ndn::Data>(lastSegment);
    filterData->setFreshnessPeriod(ndn::time::milliseconds(10));
    filterData->setContent(reinterpret_cast<const uint8_t*>(payload + startIndex), payloadLength);
    filterData->setFinalBlockId(ndn::Name::Component::fromSegment(seqNo));

    signData(*filterData);
    m_mutex.lock();
    m_activeQueryToFirstResponse.insert(*filterData);
    m_face->put(*filterData);
    m_mutex.unlock();
  }
  _LOG_DEBUG("<< QueryAdapter::populateFiltersMenu");
}

template <typename DatabaseHandler>
void
QueryAdapter<DatabaseHandler>::getFiltersMenu(Json::Value& value)
{
  // empty
}

// get distinct value of each column
template <>
void
QueryAdapter<ConnectionPool_T>::getFiltersMenu(Json::Value& value)
{
  _LOG_DEBUG(">> QueryAdapter::getFiltersMenu");
  Json::Value tmp;

  Connection_T conn = ConnectionPool_getConnection(*m_dbConnPool);
  if (!conn) {
    _LOG_DEBUG("No available database connections");
    return;
  }

  for (size_t i = 0; i < m_filterCategoryNames.size(); i++) {
    std::string columnName = m_filterCategoryNames[i];
    std::string getFilterSql("SELECT DISTINCT " + columnName +
                             " FROM " + m_databaseTable + ";");

    ResultSet_T res4ColumnName;
    TRY {
      res4ColumnName = Connection_executeQuery(conn, reinterpret_cast<const char*>(getFilterSql.c_str()), getFilterSql.size());
    }
    CATCH(SQLException) {
      _LOG_ERROR(Connection_getLastError(conn));
    }
    END_TRY;

    while (ResultSet_next(res4ColumnName)) {
      tmp[columnName].append(ResultSet_getString(res4ColumnName, 1));
    }

    value.append(tmp);
    tmp.clear();
  }

  _LOG_DEBUG("<< QueryAdapter::getFiltersMenu");
}

template <typename DatabaseHandler>
void
QueryAdapter<DatabaseHandler>::signData(ndn::Data& data)
{
  if (m_signingId.empty())
    m_keyChain->sign(data);
  else {
    ndn::Name keyName = m_keyChain->getDefaultKeyNameForIdentity(m_signingId);
    ndn::Name certName = m_keyChain->getDefaultCertificateNameForKey(keyName);
    m_keyChain->sign(data, certName);
  }
}

template <typename DatabaseHandler>
ndn::Name
QueryAdapter<DatabaseHandler>::getQueryResultsName(std::shared_ptr<const ndn::Interest> interest,
                                                   const ndn::Name::Component& version)
{
  // use generic name, instead of specific one
  ndn::Name queryResultName = interest->getName();
  queryResultName.append(version);
  return queryResultName;
}

template <typename DatabaseHandler>
std::shared_ptr<ndn::Data>
QueryAdapter<DatabaseHandler>::makeAckData(std::shared_ptr<const ndn::Interest> interest,
                                           const ndn::Name::Component& version)
{
  std::string queryResultNameStr(getQueryResultsName(interest, version).toUri());

  std::shared_ptr<ndn::Data> ack = std::make_shared<ndn::Data>(interest->getName());
  ack->setContent(reinterpret_cast<const uint8_t*>(queryResultNameStr.c_str()),
                  queryResultNameStr.length());
  ack->setFreshnessPeriod(ndn::time::milliseconds(10000));

  signData(*ack);

  _LOG_DEBUG("Make ACK : " << queryResultNameStr);

  return ack;
}

template <typename DatabaseHandler>
void
QueryAdapter<DatabaseHandler>::sendNack(const ndn::Name& dataPrefix)
{
  uint64_t segmentNo = 0;

  std::shared_ptr<ndn::Data> nack =
    std::make_shared<ndn::Data>(ndn::Name(dataPrefix).appendSegment(segmentNo));
  nack->setFreshnessPeriod(ndn::time::milliseconds(10000));
  nack->setFinalBlockId(ndn::Name::Component::fromSegment(segmentNo));

  signData(*nack);

  _LOG_DEBUG("Send Nack: " << ndn::Name(dataPrefix).appendSegment(segmentNo));

  m_mutex.lock();
  m_cache.insert(*nack);
  m_face->put(*nack);
  m_mutex.unlock();
}

template <typename DatabaseHandler>
bool
QueryAdapter<DatabaseHandler>::json2AutocompletionSql(std::stringstream& sqlQuery,
                                                      Json::Value& jsonValue,
                                                      bool& lastComponent,
                                                      std::stringstream& fieldName)
{
  _LOG_DEBUG(">> QueryAdapter::json2AutocompletionSql");

  _LOG_DEBUG(jsonValue.toStyledString());

  if (jsonValue.type() != Json::objectValue) {
    return false;
  }

  std::string typedString;
  // get the string in the jsonValue
  for (Json::Value::iterator iter = jsonValue.begin(); iter != jsonValue.end(); ++iter)
  {
    Json::Value key = iter.key();
    Json::Value value = (*iter);

    if (key == Json::nullValue || value == Json::nullValue) {
      _LOG_ERROR("Null key or value in JsonValue");
      return false;
    }

    // cannot convert to string
    if (!key.isConvertibleTo(Json::stringValue) || !value.isConvertibleTo(Json::stringValue)) {
      _LOG_ERROR("Malformed JsonQuery string");
      return false;
    }

    if (key.asString().compare("?") == 0) {
      typedString = value.asString();
      // since the front end triggers the autocompletion when users typed '/',
      // there must be a '/' at the end, and the first char must be '/'
      if (typedString.at(typedString.length() - 1) != '/' || typedString.find("/") != 0)
        return false;
      break;
    }
  }

  // 1. get the expected column number by parsing the typedString, so we can get the filed name
  size_t pos = 0;
  size_t start = 1; // start from the 1st char which is not '/'
  size_t count = 0; // also the name to query for
  std::string token;
  std::string delimiter = "/";
  std::map<std::string, std::string> typedComponents;
  while ((pos = typedString.find(delimiter, start)) != std::string::npos) {
    token = typedString.substr(start, pos - start);
    if (count >= m_nameFields.size() - 1) {
      return false;
    }

    // add column name and value (token) into map
    typedComponents.insert(std::pair<std::string, std::string>(m_nameFields[count], token));
    count++;
    start = pos + 1;
  }

  // 2. generate the sql string (append what appears in the typed string, like activity='xxx'),
  // return true
  if (count == m_nameFields.size() - 1)
    lastComponent = true; // indicate this query is to query the last component

  bool more = false;

  fieldName << m_nameFields[count];
  for (std::map<std::string, std::string>::iterator it = typedComponents.begin();
       it != typedComponents.end(); ++it) {
    if (more)
      sqlQuery << " AND";
    else
      sqlQuery << " WHERE";

    sqlQuery << " " << it->first << "='" << it->second << "'";

    more = true;
  }
  sqlQuery << ";";
  return true;
}

template <typename databasehandler>
bool
QueryAdapter<databasehandler>::doPrefixBasedSearch(Json::Value& jsonValue,
                                                   std::vector<std::pair<std::string, std::string>>& typedComponents)
{
  _LOG_DEBUG(">> QueryAdapter::doPrefixBasedSearch");

  if (jsonValue.type() != Json::objectValue) {
    return false;
  }

  std::string typedString;
  // get the string in the jsonValue
  for (Json::Value::iterator iter = jsonValue.begin(); iter != jsonValue.end(); ++iter)
  {
    Json::Value key = iter.key();
    Json::Value value = (*iter);

    if (key == Json::nullValue || value == Json::nullValue) {
      _LOG_ERROR("null key or value in jsonValue");
      return false;
    }

    // cannot convert to string
    if (!key.isConvertibleTo(Json::stringValue) || !value.isConvertibleTo(Json::stringValue)) {
      _LOG_ERROR("malformed jsonquery string");
      return false;
    }

    if (key.asString().compare("??") == 0) {
      typedString = value.asString();
      if (typedString.empty() || typedString.find("/") != 0)
        return false;
      break;
    }
  }

  // 1. get the expected column number by parsing the typedString, so we can get the filed name
  size_t pos = 0;
  size_t start = 1; // start from the 1st char which is not '/'
  size_t count = 0; // also the name to query for
  size_t typedStringlen = typedString.length();
  std::string token;
  std::string delimiter = "/";

  while ((pos = typedString.find(delimiter, start)) != std::string::npos) {
    token = typedString.substr(start, pos - start);
    if (count >= m_nameFields.size()) {
      return false;
    }

    // add column name and value (token) into map
    typedComponents.push_back(std::make_pair(m_nameFields[count], token));

    count++;
    start = pos + 1;
  }

  // we may have a component after the last "/"
  if (start < typedStringlen) {
    typedComponents.push_back(std::make_pair(m_nameFields[count],
                                             typedString.substr(start, typedStringlen - start)));
  }

  return true;
}


template <typename databasehandler>
bool
QueryAdapter<databasehandler>::doFilterBasedSearch(Json::Value& jsonValue,
                                                   std::vector<std::pair<std::string, std::string>>& typedComponents)
{
  _LOG_DEBUG(">> QueryAdapter::doFilterBasedSearch");

  if (jsonValue.type() != Json::objectValue) {
    return false;
  }

  for (Json::Value::iterator iter = jsonValue.begin(); iter != jsonValue.end(); ++iter)
  {
    Json::Value key = iter.key();
    Json::Value value = (*iter);

    if (key == Json::nullValue || value == Json::nullValue) {
      _LOG_ERROR("null key or value in jsonValue");
      return false;
    }

    // cannot convert to string
    if (!key.isConvertibleTo(Json::stringValue) || !value.isConvertibleTo(Json::stringValue)) {
      _LOG_ERROR("malformed jsonQuery string");
      return false;
    }

    if (key.asString().compare("?") == 0 || key.asString().compare("??") == 0) {
      continue;
    }

    _LOG_DEBUG(key.asString() << " " << value.asString());
    typedComponents.push_back(std::make_pair(key.asString(), value.asString()));
  }

  return true;
}


template <typename DatabaseHandler>
void
QueryAdapter<DatabaseHandler>::runJsonQuery(std::shared_ptr<const ndn::Interest> interest)
{
  _LOG_DEBUG(">> QueryAdapter::runJsonQuery");

  // 1) Strip the prefix off the ndn::Interest's ndn::Name
  // +1 to grab JSON component after "query" component

  ndn::Name::Component jsonStr = interest->getName()[m_prefix.size()+1];
  // This one cannot parse the JsonQuery correctly, and should be moved to runJsonQuery
  const std::string jsonQuery(reinterpret_cast<const char*>(jsonStr.value()), jsonStr.value_size());

  if (jsonQuery.length() <= 0) {
    // no JSON query, send Nack
    sendNack(interest->getName());
    return;
  }

  // the version should be replaced with ChronoSync state digest
  ndn::name::Component version;

  if(m_socket != nullptr) {
    const ndn::ConstBufferPtr digestPtr = m_socket->getRootDigest();
    std::string digestStr = ndn::toHex(digestPtr->buf(), digestPtr->size());
    _LOG_DEBUG("Original digest " << m_chronosyncDigest);
    _LOG_DEBUG("New digest : " << digestStr);
    // if the m_chronosyncDigest and the rootdigest are not equal
    if (digestStr != m_chronosyncDigest) {
      // (1) update chronosyncDigest
      // (2) clear all staled ACK data
      m_mutex.lock();
      m_chronosyncDigest = digestStr;
      m_activeQueryToFirstResponse.erase(ndn::Name("/"));
      m_mutex.unlock();
      _LOG_DEBUG("Change digest to " << m_chronosyncDigest);
    }
    version = ndn::name::Component::fromEscapedString(digestStr);
  }
  else {
    version = ndn::name::Component::fromEscapedString(m_chronosyncDigest);
  }

  // 2) From the remainder of the ndn::Interest's ndn::Name, get the JSON out
  Json::Value parsedFromString;
  Json::Reader reader;
  if (!reader.parse(jsonQuery, parsedFromString)) {
    // json object is broken
    sendNack(interest->getName());
    _LOG_ERROR("Cannot parse the JsonQuery");
    return;
  }

  // 3) Convert the JSON Query into a MySQL one
  ndn::Name segmentPrefix(getQueryResultsName(interest, version));
  _LOG_DEBUG("segmentPrefix :" << segmentPrefix);

  Json::Value tmp;
  std::vector<std::pair<std::string, std::string>> typedComponents;

  // expect the autocomplete and the component-based query are separate
  // if Json::Value contains ? as key, is autocompletion
  if (parsedFromString.get("?", tmp) != tmp) {
    bool lastComponent = false;
    std::stringstream sqlQuery, fieldName;

    // must generate the sql string for autocomple, the selected column is changing
    if (!json2AutocompletionSql(sqlQuery, parsedFromString, lastComponent, fieldName)) {
      sendNack(segmentPrefix);
      return;
    }
    prepareSegmentsBySqlString(segmentPrefix, sqlQuery.str(), lastComponent, fieldName.str());
  }
  else if (parsedFromString.get("??", tmp) != tmp) {
    if (!doPrefixBasedSearch(parsedFromString, typedComponents)) {
      sendNack(segmentPrefix);
      return;
    }
    prepareSegmentsByParams(typedComponents, segmentPrefix);
  }
  else {
    if (!doFilterBasedSearch(parsedFromString, typedComponents)) {
      sendNack(segmentPrefix);
      return;
    }
    prepareSegmentsByParams(typedComponents, segmentPrefix);
  }

}

template <typename databasehandler>
void
QueryAdapter<databasehandler>::
prepareSegmentsByParams(std::vector<std::pair<std::string, std::string>>& queryParams,
                        const ndn::Name& segmentprefix)
{
}

template <>
void
QueryAdapter<ConnectionPool_T>::
prepareSegmentsByParams(std::vector<std::pair<std::string, std::string>>& queryParams,
                        const ndn::Name& segmentPrefix)
{
  _LOG_DEBUG(">> QueryAdapter::prepareSegmentsByParams");

  // the prepared_statement cannot improve the performance, but can simplify the code
  Connection_T conn = ConnectionPool_getConnection(*m_dbConnPool);
  if (!conn) {
    // do not answer for this request due to lack of connections, request will come back later
    _LOG_DEBUG("No available database connections");
    return;
  }
  std::string getRecordNumSqlStr("SELECT count(name) FROM ");
  getRecordNumSqlStr += m_databaseTable;
  getRecordNumSqlStr += " WHERE ";
  for (size_t i = 0; i < m_nameFields.size(); i++) {
    getRecordNumSqlStr += m_nameFields[i];
    getRecordNumSqlStr += " LIKE ?";
    if (i != m_nameFields.size() - 1) {
      getRecordNumSqlStr += " AND ";
    }
  }

  PreparedStatement_T ps4RecordNum =
    Connection_prepareStatement(conn, reinterpret_cast<const char*>(getRecordNumSqlStr.c_str()), getRecordNumSqlStr.size());

  // before query, initialize all params for statement
  for (size_t i = 0; i < m_nameFields.size(); i++) {
    PreparedStatement_setString(ps4RecordNum, i + 1, "%");
  }

  // reset params based on the query
  for (std::vector<std::pair<std::string, std::string>>::iterator it = queryParams.begin();
       it != queryParams.end(); ++it) {
    // dictionary is faster
    for (size_t i = 0; i < m_nameFields.size(); i++) {
      if (it->first == m_nameFields[i]) {
        PreparedStatement_setString(ps4RecordNum, i + 1, it->second.c_str());
      }
    }
  }

  ResultSet_T res4RecordNum;
  TRY {
    res4RecordNum = PreparedStatement_executeQuery(ps4RecordNum);
  }
  CATCH(SQLException) {
    _LOG_ERROR(Connection_getLastError(conn));
  }
  END_TRY;

  uint64_t resultCount = 0; // use count sql to get

  // result for record number
  while (ResultSet_next(res4RecordNum)) {
    resultCount = ResultSet_getInt(res4RecordNum, 1);
  }

  // get name list statement
  std::string getNameListSqlStr("SELECT name, has_metadata FROM ");
  getNameListSqlStr += m_databaseTable;
  getNameListSqlStr += " WHERE ";
  for (size_t i = 0; i < m_nameFields.size(); i++) {
    getNameListSqlStr += m_nameFields[i];
    getNameListSqlStr += " LIKE ?";
    if (i != m_nameFields.size() - 1) {
      getNameListSqlStr += " AND ";
    }
  }

  PreparedStatement_T ps4Name =
    Connection_prepareStatement(conn, reinterpret_cast<const char*>(getNameListSqlStr.c_str()), getNameListSqlStr.size());

  // before query, initialize all params for statement
  for (size_t i = 0; i < m_nameFields.size(); i++) {
    PreparedStatement_setString(ps4Name, i + 1, "%");
  }

  // reset params based on the query
  for (std::vector<std::pair<std::string, std::string>>::iterator it = queryParams.begin();
       it != queryParams.end(); ++it) {
    // dictionary is faster
    for (size_t i = 0; i < m_nameFields.size(); i++) {
      if (it->first == m_nameFields[i]) {
        PreparedStatement_setString(ps4Name, i + 1, it->second.c_str());
      }
    }
  }

  ResultSet_T res4Name;
  TRY {
    res4Name = PreparedStatement_executeQuery(ps4Name);
  }
  CATCH(SQLException) {
    _LOG_ERROR(Connection_getLastError(conn));
  }
  END_TRY;

  generateSegments(res4Name, segmentPrefix, resultCount, false, false);

  Connection_close(conn);
}

template <typename DatabaseHandler>
void
QueryAdapter<DatabaseHandler>::generateSegments(ResultSet_T& res,
                                                const ndn::Name& segmentPrefix,
                                                int resultCount,
                                                bool autocomplete,
                                                bool lastComponent)
{
  uint64_t segmentno = 0;
  Json::Value tmp, buf, resultjson;
  Json::FastWriter fastWriter;

  bool twoColumns = false;
  if (ResultSet_getColumnCount(res) > 1) {
    twoColumns = true;
  }

  uint64_t viewstart = 0, viewend = 0;
  while (ResultSet_next(res)) {
    tmp["name"] = ResultSet_getString(res, 1);
    if (twoColumns) {
      tmp["has_metadata"] = ResultSet_getInt(res, 2);
    } else {
      tmp["has_metadata"] = 0;
    }
    buf.append(tmp);
    const std::string tmpString = fastWriter.write(buf);
    if (tmpString.length() > PAYLOAD_LIMIT) {
      std::shared_ptr<ndn::Data> data
        = makeReplyData(segmentPrefix, resultjson, segmentno, false,
                        autocomplete, resultCount, viewstart, viewend, lastComponent);
      m_mutex.lock();
      m_cache.insert(*data);
      m_face->put(*data);
      m_mutex.unlock();

      buf.clear();
      resultjson.clear();
      segmentno++;
      viewstart = viewend + 1;
    }
    resultjson.append(tmp);
    buf = resultjson;
    tmp.clear();
    viewend++;
  }
  std::shared_ptr<ndn::Data> data
    = makeReplyData(segmentPrefix, resultjson, segmentno, true,
                    autocomplete, resultCount, viewstart, viewend, lastComponent);
  m_mutex.lock();
  m_cache.insert(*data);
  m_face->put(*data);
  m_mutex.unlock();
}

template <typename DatabaseHandler>
void
QueryAdapter<DatabaseHandler>::prepareSegmentsBySqlString(const ndn::Name& segmentPrefix,
                                                          const std::string& sqlString,
                                                          bool lastComponent,
                                                          const std::string& nameField)
{
  // empty
}


template <>
void
QueryAdapter<ConnectionPool_T>::prepareSegmentsBySqlString(const ndn::Name& segmentPrefix,
                                                          const std::string& sqlString,
                                                          bool lastComponent,
                                                          const std::string& nameField)
{
  _LOG_DEBUG(">> QueryAdapter::prepareSegmentsBySqlString");

  _LOG_DEBUG(sqlString);

  Connection_T conn = ConnectionPool_getConnection(*m_dbConnPool);
  if (!conn) {
    _LOG_DEBUG("No available database connections");
    return;
  }

  //// just for get the rwo count ...
  std::string getRecordNumSqlStr("SELECT COUNT( DISTINCT ");
  getRecordNumSqlStr += nameField;
  getRecordNumSqlStr += ") FROM ";
  getRecordNumSqlStr += m_databaseTable;
  getRecordNumSqlStr += sqlString;

  ResultSet_T res4RecordNum;
  TRY {
    res4RecordNum = Connection_executeQuery(conn, reinterpret_cast<const char*>(getRecordNumSqlStr.c_str()), getRecordNumSqlStr.size());
  }
  CATCH(SQLException) {
    _LOG_ERROR(Connection_getLastError(conn));
  }
  END_TRY;

  uint64_t resultCount = 0;
  while (ResultSet_next(res4RecordNum)) {
    resultCount = ResultSet_getInt(res4RecordNum, 1);
  }
  ////

  std::string getNextFieldsSqlStr("SELECT DISTINCT ");
  getNextFieldsSqlStr += nameField;
  getNextFieldsSqlStr += " FROM ";
  getNextFieldsSqlStr += m_databaseTable;
  getNextFieldsSqlStr += sqlString;

  ResultSet_T res4NextFields;
  TRY {
    res4NextFields = Connection_executeQuery(conn, reinterpret_cast<const char*>(getNextFieldsSqlStr.c_str()), getNextFieldsSqlStr.size());
  }
  CATCH(SQLException) {
    _LOG_ERROR(Connection_getLastError(conn));
  }
  END_TRY;

  generateSegments(res4NextFields, segmentPrefix, resultCount, true, lastComponent);

  Connection_close(conn);
}

template <typename DatabaseHandler>
std::shared_ptr<ndn::Data>
QueryAdapter<DatabaseHandler>::makeReplyData(const ndn::Name& segmentPrefix,
                                             const Json::Value& value,
                                             uint64_t segmentNo,
                                             bool isFinalBlock,
                                             bool isAutocomplete,
                                             uint64_t resultCount,
                                             uint64_t viewStart,
                                             uint64_t viewEnd,
                                             bool lastComponent)
{
  Json::Value entry;
  Json::FastWriter fastWriter;

  entry["resultCount"] = Json::UInt64(resultCount);;
  entry["viewStart"] = Json::UInt64(viewStart);
  entry["viewEnd"] = Json::UInt64(viewEnd);

  if (lastComponent)
    entry["lastComponent"] = Json::Value(true);

  _LOG_DEBUG("resultCount " << resultCount << "; "
             << "viewStart " << viewStart << "; "
             << "viewEnd " << viewEnd);

  if (isAutocomplete) {
    entry["next"] = value;
  } else {
    entry["results"] = value;
  }
  const std::string jsonMessage = fastWriter.write(entry);
  const char* payload = jsonMessage.c_str();
  size_t payloadLength = jsonMessage.size() + 1;
  ndn::Name segmentName(segmentPrefix);
  segmentName.appendSegment(segmentNo);

  std::shared_ptr<ndn::Data> data = std::make_shared<ndn::Data>(segmentName);
  data->setContent(reinterpret_cast<const uint8_t*>(payload), payloadLength);
  data->setFreshnessPeriod(ndn::time::milliseconds(10000));

  if (isFinalBlock) {
    data->setFinalBlockId(ndn::Name::Component::fromSegment(segmentNo));
  }

  _LOG_DEBUG(segmentName);

  signData(*data);
  return data;
}


} // namespace query
} // namespace atmos
#endif //ATMOS_QUERY_QUERY_ADAPTER_HPP