////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2018 ArangoDB GmbH, Cologne, Germany
///
/// Licensed 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.
///
/// Copyright holder is ArangoDB GmbH, Cologne, Germany
///
/// @author Simon Grätzer
////////////////////////////////////////////////////////////////////////////////

#include "ClusterIndexFactory.h"
#include "Basics/StaticStrings.h"
#include "Basics/StringUtils.h"
#include "Basics/VelocyPackHelper.h"
#include "Cluster/ServerState.h"
#include "ClusterEngine/ClusterEngine.h"
#include "ClusterEngine/ClusterIndex.h"
#include "Indexes/Index.h"
#include "StorageEngine/EngineSelectorFeature.h"
#include "VocBase/LogicalCollection.h"
#include "VocBase/ticks.h"
#include "VocBase/voc-types.h"

#include <velocypack/Builder.h>
#include <velocypack/Iterator.h>
#include <velocypack/Slice.h>
#include <velocypack/velocypack-aliases.h>

namespace arangodb {

ClusterIndexFactory::ClusterIndexFactory() {
  emplaceFactory(
    "edge",
    [](
        LogicalCollection* collection,
        velocypack::Slice const& definition,
        TRI_idx_iid_t id,
        bool isClusterConstructor
    )->std::shared_ptr<Index> {
      if (!isClusterConstructor) {
        // this indexes cannot be created directly
        THROW_ARANGO_EXCEPTION_MESSAGE(
          TRI_ERROR_INTERNAL, "cannot create edge index"
        );
      }

      auto* ce = static_cast<ClusterEngine*>(EngineSelectorFeature::ENGINE);
      auto ct = ce->engineType();

      return std::make_shared<ClusterIndex>(
        id, collection, ct, Index::TRI_IDX_TYPE_EDGE_INDEX, definition
      );
    }
  );

  emplaceFactory(
    "primary",
    [](
        LogicalCollection* collection,
        velocypack::Slice const& definition,
        TRI_idx_iid_t id,
        bool isClusterConstructor
    )->std::shared_ptr<Index> {
      if (!isClusterConstructor) {
        // this indexes cannot be created directly
        THROW_ARANGO_EXCEPTION_MESSAGE(
          TRI_ERROR_INTERNAL, "cannot create primary index"
        );
      }

      auto* ce = static_cast<ClusterEngine*>(EngineSelectorFeature::ENGINE);
      auto ct = ce->engineType();

      return std::make_shared<ClusterIndex>(
        0, collection, ct, Index::TRI_IDX_TYPE_PRIMARY_INDEX, definition
      );
    }
  );

  // both engines support all types right now
  static const std::vector<std::string> supported = {
    "fulltext",
    "geo",
    "geo1",
    "geo2",
    "hash",
    "persistent",
    "skiplist"
  };

  for (auto& typeStr: supported) {
    auto type = Index::type(typeStr);

    emplaceFactory(
      typeStr,
      [type](
          LogicalCollection* collection,
          velocypack::Slice const& definition,
          TRI_idx_iid_t id,
          bool isClusterConstructor
      )->std::shared_ptr<Index> {
        auto* ce = static_cast<ClusterEngine*>(EngineSelectorFeature::ENGINE);
        auto ct = ce->engineType();

        return std::make_shared<ClusterIndex>(
          id, collection, ct, type, definition
        );
      }
    );
  }

  // both engines support all types right now
  static const std::vector<std::string> supported_norm = {
    "edge",
    "fulltext",
    "geo",
    "geo1",
    "geo2",
    "hash",
    "persistent",
    "primary",
    "skiplist"
  };

  // delegate normalization to the 'actualEngine'
  // FIXME TODO is it actually correct to tie the definition in the Agency to the DBServer engine?
  for (auto& typeStr: supported_norm) {
    emplaceNormalizer(
      typeStr,
      [](
         velocypack::Builder& normalized,
         velocypack::Slice definition,
         bool isCreation
      ) -> Result {
        auto* ce = static_cast<ClusterEngine*>(EngineSelectorFeature::ENGINE);

        if (!ce) {
          return TRI_ERROR_INTERNAL;
        }

        auto* ae = ce->actualEngine();

        if (!ae) {
          return TRI_ERROR_INTERNAL;
        }

        normalized.clear(); // enhanceIndexDefinition(...) expects an empty open object

        return ae->indexFactory().enhanceIndexDefinition(
          definition, normalized, isCreation, true
        );
      }
    );
  }
}

void ClusterIndexFactory::fillSystemIndexes(
    arangodb::LogicalCollection* col,
    std::vector<std::shared_ptr<arangodb::Index>>& systemIndexes) const {
  // create primary index
  VPackBuilder input;
  input.openObject();
  input.add(StaticStrings::IndexType, VPackValue("primary"));
  input.add(StaticStrings::IndexId, VPackValue("0"));
  input.add(StaticStrings::IndexFields, VPackValue(VPackValueType::Array));
  input.add(VPackValue(StaticStrings::KeyString));
  input.close();
  input.add(StaticStrings::IndexUnique, VPackValue(true));
  input.add(StaticStrings::IndexSparse, VPackValue(false));
  input.close();

  // get the storage engine type
  ClusterEngine* ce =
      static_cast<ClusterEngine*>(EngineSelectorFeature::ENGINE);
  ClusterEngineType ct = ce->engineType();

  systemIndexes.emplace_back(std::make_shared<arangodb::ClusterIndex>(
      0, col, ct, Index::TRI_IDX_TYPE_PRIMARY_INDEX, input.slice()));
  // create edges indexes
  if (col->type() == TRI_COL_TYPE_EDGE) {
    // first edge index
    input.clear();
    input.openObject();
    input.add(StaticStrings::IndexType,
              VPackValue(Index::oldtypeName(Index::TRI_IDX_TYPE_EDGE_INDEX)));
    input.add(StaticStrings::IndexId, VPackValue("1"));
    input.add(StaticStrings::IndexFields, VPackValue(VPackValueType::Array));
    input.add(VPackValue(StaticStrings::FromString));
    if (ct == ClusterEngineType::MMFilesEngine) {
      input.add(VPackValue(StaticStrings::ToString));
    }
    input.close();
    input.add(StaticStrings::IndexUnique, VPackValue(false));
    input.add(StaticStrings::IndexSparse, VPackValue(false));
    input.close();
    systemIndexes.emplace_back(std::make_shared<arangodb::ClusterIndex>(
        1, col, ct, Index::TRI_IDX_TYPE_EDGE_INDEX, input.slice()));

    // second edge index
    if (ct == ClusterEngineType::RocksDBEngine) {
      input.clear();
      input.openObject();
      input.add(StaticStrings::IndexType,
                VPackValue(Index::oldtypeName(Index::TRI_IDX_TYPE_EDGE_INDEX)));
      input.add(StaticStrings::IndexId, VPackValue("2"));
      input.add(StaticStrings::IndexFields, VPackValue(VPackValueType::Array));
      input.add(VPackValue(StaticStrings::ToString));
      input.close();
      input.add(StaticStrings::IndexUnique, VPackValue(false));
      input.add(StaticStrings::IndexSparse, VPackValue(false));
      input.close();
      systemIndexes.emplace_back(std::make_shared<arangodb::ClusterIndex>(
          2, col, ct, Index::TRI_IDX_TYPE_EDGE_INDEX, input.slice()));
    }
  }
}

void ClusterIndexFactory::prepareIndexes(
    LogicalCollection* col, VPackSlice const& indexesSlice,
    std::vector<std::shared_ptr<arangodb::Index>>& indexes) const {
  TRI_ASSERT(indexesSlice.isArray());

  for (auto const& v : VPackArrayIterator(indexesSlice)) {
    if (basics::VelocyPackHelper::getBooleanValue(v, "error", false)) {
      // We have an error here.
      // Do not add index.
      continue;
    }

    auto idx = prepareIndexFromSlice(v, false, col, true);

    if (!idx) {
      LOG_TOPIC(ERR, arangodb::Logger::ENGINES)
        << "error creating index from definition '" << v.toString() << "'";

      continue;
    }

    indexes.emplace_back(std::move(idx));
  }
}

} // arangodb

// -----------------------------------------------------------------------------
// --SECTION--                                                       END-OF-FILE
// -----------------------------------------------------------------------------