////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2014-2016 ArangoDB GmbH, Cologne, Germany
/// Copyright 2004-2014 triAGENS 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 Michael Hackstein
////////////////////////////////////////////////////////////////////////////////

#include "RestEdgesHandler.h"
#include "Aql/AstNode.h"
#include "Aql/Graphs.h"
#include "Aql/Variable.h"
#include "Cluster/ClusterMethods.h"
#include "Cluster/ServerState.h"
#include "Transaction/Methods.h"
#include "Transaction/StandaloneContext.h"
#include "Utils/CollectionNameResolver.h"
#include "Utils/OperationCursor.h"

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

using namespace arangodb;
using namespace arangodb::rest;

RestEdgesHandler::RestEdgesHandler(GeneralRequest* request, GeneralResponse* response)
    : RestVocbaseBaseHandler(request, response) {}

RestStatus RestEdgesHandler::execute() {
  // extract the sub-request type
  auto const type = _request->requestType();

  // execute one of the CRUD methods
  switch (type) {
    case rest::RequestType::GET:
      readEdges();
      break;
    case rest::RequestType::POST:
      readEdgesForMultipleVertices();
      break;
    default:
      generateNotImplemented("ILLEGAL " + EDGES_PATH);
      break;
  }

  // this handler is done
  return RestStatus::DONE;
}

void RestEdgesHandler::readCursor(aql::AstNode* condition, aql::Variable const* var,
                                  std::string const& collectionName,
                                  transaction::Methods& trx,
                                  std::function<void(LocalDocumentId const&)> const& cb) {
  transaction::Methods::IndexHandle indexId;
  bool foundIdx =
      trx.getBestIndexHandleForFilterCondition(collectionName, condition, var,
                                               1000, aql::IndexHint(), indexId);
  if (!foundIdx) {
    // Right now we enforce an edge index that can exactly! work on this
    // condition. So it is impossible to not find an index.
    TRI_ASSERT(false);
    THROW_ARANGO_EXCEPTION_MESSAGE(
        TRI_ERROR_ARANGO_NO_INDEX,
        "Unable to find an edge index to identify matching edges.");
  }

  IndexIteratorOptions opts;
  OperationCursor cursor(trx.indexScanForCondition(indexId, condition, var, opts));
  cursor.all(cb);
}

bool RestEdgesHandler::getEdgesForVertex(
    std::string const& id, TRI_voc_cid_t cid, std::string const& collectionName,
    TRI_edge_direction_e direction, transaction::Methods& trx,
    std::function<void(LocalDocumentId const&)> const& cb) {
  trx.pinData(cid);  // will throw when it fails

  // Create a conditionBuilder that manages the AstNodes for querying
  aql::EdgeConditionBuilderContainer condBuilder;
  condBuilder.setVertexId(id);

  aql::Variable const* var = condBuilder.getVariable();

  switch (direction) {
    case TRI_EDGE_IN:
      readCursor(condBuilder.getInboundCondition(), var, collectionName, trx, cb);
      break;
    case TRI_EDGE_OUT:
      readCursor(condBuilder.getOutboundCondition(), var, collectionName, trx, cb);
      break;
    case TRI_EDGE_ANY:
      // We have to call both directions
      readCursor(condBuilder.getInboundCondition(), var, collectionName, trx, cb);
      readCursor(condBuilder.getOutboundCondition(), var, collectionName, trx, cb);
      break;
  }
  return true;
}

bool RestEdgesHandler::parseDirection(TRI_edge_direction_e& direction) {
  bool found;
  std::string dir = _request->value("direction", found);

  if (!found || dir.empty()) {
    dir = "any";
  }

  std::string dirString(dir);

  if (dirString == "any") {
    direction = TRI_EDGE_ANY;
  } else if (dirString == "out" || dirString == "outbound") {
    direction = TRI_EDGE_OUT;
  } else if (dirString == "in" || dirString == "inbound") {
    direction = TRI_EDGE_IN;
  } else {
    generateError(rest::ResponseCode::BAD, TRI_ERROR_HTTP_BAD_PARAMETER,
                  "<direction> must by any, in, or out, not: " + dirString);
    return false;
  }

  return true;
}

bool RestEdgesHandler::validateCollection(std::string const& name) {
  CollectionNameResolver resolver(_vocbase);
  auto collection = resolver.getCollection(name);
  auto colType = collection ? collection->type() : TRI_COL_TYPE_UNKNOWN;

  if (colType == TRI_COL_TYPE_UNKNOWN) {
    generateError(rest::ResponseCode::NOT_FOUND, TRI_ERROR_ARANGO_DATA_SOURCE_NOT_FOUND);
    return false;
  }

  if (colType != TRI_COL_TYPE_EDGE) {
    generateError(rest::ResponseCode::BAD, TRI_ERROR_ARANGO_COLLECTION_TYPE_INVALID);
    return false;
  }

  return true;
}

// read in- or outbound edges
bool RestEdgesHandler::readEdges() {
  std::vector<std::string> const& suffixes = _request->decodedSuffixes();

  if (suffixes.size() != 1) {
    generateError(rest::ResponseCode::BAD, TRI_ERROR_HTTP_BAD_PARAMETER,
                  "expected GET " + EDGES_PATH +
                      "/<collection-identifier>?vertex=<vertex-handle>&"
                      "direction=<direction>");
    return false;
  }

  std::string collectionName = suffixes[0];
  if (!validateCollection(collectionName)) {
    return false;
  }

  TRI_edge_direction_e direction;
  if (!parseDirection(direction)) {
    return false;
  }

  bool found;
  std::string const& startVertex = _request->value("vertex", found);

  if (!found || startVertex.empty()) {
    generateError(rest::ResponseCode::BAD, TRI_ERROR_ARANGO_DOCUMENT_HANDLE_BAD,
                  "illegal document handle");
    return false;
  }

  // find and load collection given by name or identifier
  auto trx = createTransaction(collectionName, AccessMode::Type::READ);

  if (ServerState::instance()->isCoordinator()) {
    std::string vertexString(startVertex);
    rest::ResponseCode responseCode;
    VPackBuffer<uint8_t> buffer;
    VPackBuilder resultDocument(buffer);
    resultDocument.openObject();
    int res = getFilteredEdgesOnCoordinator(*trx, collectionName, vertexString, direction,
                                            responseCode, resultDocument);

    if (res != TRI_ERROR_NO_ERROR) {
      generateError(responseCode, res);
      return false;
    }

    resultDocument.add(StaticStrings::Error, VPackValue(false));
    resultDocument.add("code", VPackValue(200));
    resultDocument.close();

    generateResult(rest::ResponseCode::OK, std::move(buffer));

    return true;
  }

  // .............................................................................
  // inside read transaction
  // .............................................................................

  Result res = trx->begin();

  if (!res.ok()) {
    generateTransactionError(collectionName, OperationResult(std::move(res)), "");
    return false;
  }

  size_t filtered = 0;
  size_t scannedIndex = 0;

  VPackBuffer<uint8_t> buffer;
  VPackBuilder resultBuilder(buffer);
  resultBuilder.openObject();
  // build edges
  resultBuilder.add(VPackValue("edges"));  // only key
  resultBuilder.openArray();

  TRI_voc_cid_t cid = trx->addCollectionAtRuntime(collectionName, AccessMode::Type::READ);
  auto collection = trx->documentCollection(cid);
  std::unordered_set<LocalDocumentId> foundTokens;
  auto cb = [&](LocalDocumentId const& token) {
    if (foundTokens.find(token) == foundTokens.end()) {
      collection->readDocumentWithCallback(trx.get(), token,
                                           [&resultBuilder](LocalDocumentId const&, VPackSlice doc) {
                                             resultBuilder.add(doc);
                                           });
      scannedIndex++;
      // Mark edges we find
      foundTokens.emplace(token);
    }
  };

  // NOTE: collectionName is the shard-name in DBServer case
  bool ok = getEdgesForVertex(startVertex, cid, collectionName, direction, *(trx.get()), cb);
  resultBuilder.close();

  res = trx->finish(res);
  if (!ok) {
    // Error has been built internally
    return false;
  }

  if (!res.ok()) {
    if (ServerState::instance()->isDBServer()) {
      // If we are a DBserver, we want to use the cluster-wide collection
      // name for error reporting:
      collectionName = trx->resolver()->getCollectionNameCluster(cid);
    }
    generateTransactionError(collectionName, OperationResult(std::move(res)), "");
    return false;
  }

  resultBuilder.add(StaticStrings::Error, VPackValue(false));
  resultBuilder.add("code", VPackValue(200));
  resultBuilder.add("stats", VPackValue(VPackValueType::Object));
  resultBuilder.add("scannedIndex", VPackValue(scannedIndex));
  resultBuilder.add("filtered", VPackValue(filtered));
  resultBuilder.close();  // inner object
  resultBuilder.close();

  // and generate a response
  generateResult(rest::ResponseCode::OK, std::move(buffer), trx->transactionContext());

  return true;
}

// Internal function to receive all edges for a list of vertices
// Not publicly documented on purpose.
// NOTE: It ONLY except _id strings. Nothing else
bool RestEdgesHandler::readEdgesForMultipleVertices() {
  std::vector<std::string> const& suffixes = _request->decodedSuffixes();

  if (suffixes.size() != 1) {
    generateError(rest::ResponseCode::BAD, TRI_ERROR_HTTP_BAD_PARAMETER,
                  "expected POST " + EDGES_PATH +
                      "/<collection-identifier>?direction=<direction>");
    return false;
  }

  bool parseSuccess = false;
  VPackSlice body = this->parseVPackBody(parseSuccess);

  if (!parseSuccess) {
    generateError(rest::ResponseCode::BAD, TRI_ERROR_HTTP_BAD_PARAMETER,
                  "expected POST " + EDGES_PATH +
                      "/<collection-identifier>?direction=<direction>");
    // A body is required
    return false;
  }

  if (!body.isArray()) {
    generateError(rest::ResponseCode::BAD, TRI_ERROR_HTTP_BAD_PARAMETER,
                  "Expected an array of vertex _id's in body parameter");
    return false;
  }

  std::string collectionName = suffixes[0];
  if (!validateCollection(collectionName)) {
    return false;
  }

  TRI_edge_direction_e direction;
  if (!parseDirection(direction)) {
    return false;
  }

  // find and load collection given by name or identifier
  auto trx = createTransaction(collectionName, AccessMode::Type::READ);

  if (ServerState::instance()->isCoordinator()) {
    rest::ResponseCode responseCode;
    VPackBuffer<uint8_t> buffer;
    VPackBuilder resultDocument(buffer);

    resultDocument.openObject();

    for (auto const& it : VPackArrayIterator(body)) {
      if (it.isString()) {
        std::string vertexString(it.copyString());
        int res = getFilteredEdgesOnCoordinator(*trx, collectionName, vertexString, direction,
                                                responseCode, resultDocument);

        if (res != TRI_ERROR_NO_ERROR) {
          generateError(responseCode, res);
          return false;
        }
      }
    }

    resultDocument.add(StaticStrings::Error, VPackValue(false));
    resultDocument.add("code", VPackValue(200));
    resultDocument.close();

    generateResult(rest::ResponseCode::OK, std::move(buffer));
    return true;
  }

  // .............................................................................
  // inside read transaction
  // .............................................................................

  Result res = trx->begin();
  if (!res.ok()) {
    generateTransactionError(collectionName, OperationResult(std::move(res)), "");
    return false;
  }

  size_t filtered = 0;
  size_t scannedIndex = 0;

  VPackBuffer<uint8_t> buffer;
  VPackBuilder resultBuilder(buffer);
  resultBuilder.openObject();
  // build edges
  resultBuilder.add(VPackValue("edges"));  // only key
  resultBuilder.openArray();

  TRI_voc_cid_t cid = trx->addCollectionAtRuntime(collectionName, AccessMode::Type::READ);
  auto collection = trx->documentCollection(cid);
  std::unordered_set<LocalDocumentId> foundTokens;
  auto cb = [&](LocalDocumentId const& token) {
    if (foundTokens.find(token) == foundTokens.end()) {
      collection->readDocumentWithCallback(trx.get(), token,
                                           [&resultBuilder](LocalDocumentId const&, VPackSlice doc) {
                                             resultBuilder.add(doc);
                                           });
      scannedIndex++;
      // Mark edges we find
      foundTokens.emplace(token);
    }
  };

  for (auto const& it : VPackArrayIterator(body)) {
    if (it.isString()) {
      std::string startVertex = it.copyString();

      // We ignore if this fails
      getEdgesForVertex(startVertex, cid, collectionName, direction, *(trx.get()), cb);
    }
  }
  resultBuilder.close();

  res = trx->finish(res);
  if (!res.ok()) {
    if (ServerState::instance()->isDBServer()) {
      // If we are a DBserver, we want to use the cluster-wide collection
      // name for error reporting:
      collectionName = trx->resolver()->getCollectionNameCluster(cid);
    }
    generateTransactionError(collectionName, OperationResult(std::move(res)), "");
    return false;
  }

  resultBuilder.add(StaticStrings::Error, VPackValue(false));
  resultBuilder.add("code", VPackValue(200));
  resultBuilder.add("stats", VPackValue(VPackValueType::Object));
  resultBuilder.add("scannedIndex", VPackValue(scannedIndex));
  resultBuilder.add("filtered", VPackValue(filtered));
  resultBuilder.close();  // inner object
  resultBuilder.close();

  // and generate a response
  generateResult(rest::ResponseCode::OK, std::move(buffer), trx->transactionContext());

  return true;
}