#ifndef TABLEAU_HYPER_IMPL_CONNECTION_IMPL_HPP
#define TABLEAU_HYPER_IMPL_CONNECTION_IMPL_HPP

#include <hyperapi/Catalog.hpp>
#include <hyperapi/Connection.hpp>
#include <hyperapi/Endpoint.hpp>
#include <hyperapi/HyperException.hpp>
#include <hyperapi/impl/Parameters.hpp>

#include <sstream>
#include <string>

namespace hyperapi {

inline Connection::~Connection() noexcept {
   close();
}

inline Connection::Connection(Connection&& other) noexcept
   : handle_(internal::exchange(other.handle_, nullptr)), catalog_(new Catalog(*this)) {
}

inline Connection::Connection(const Endpoint& endpoint, const std::unordered_map<std::string, std::string>& parameters)
   : Connection(endpoint, std::string{}, CreateMode::None, parameters) {
}

inline Connection::Connection(
   const Endpoint& endpoint, const std::string& databasePath, CreateMode createMode, const std::unordered_map<std::string, std::string>& parameters)
   : catalog_(new Catalog(*this)) {
   internal::Parameters p = internal::createConnectionParameters(nullptr);

   p.set("endpoint", endpoint.getConnectionDescriptor().c_str());

   if (!databasePath.empty()) {
      p.set("dbname", databasePath.c_str());
   }

   if (!endpoint.getUserAgent().empty()) {
      p.set("user_agent", endpoint.getUserAgent().c_str());
   }
   p.set("api_language", "C++");

   for (const auto& parameter : parameters) {
      p.set(parameter.first.c_str(), parameter.second.c_str());
   }

   if (hyper_error_t* error = hyper_connect(p.handle_, &handle_, static_cast<hyper_create_mode_t>(createMode))) {
      throw internal::makeHyperException(error);
   }
}

inline Connection& Connection::operator=(Connection&& other) noexcept {
   if (this != &other) {
      close();

      handle_ = internal::exchange(other.handle_, nullptr);
      catalog_ = std::unique_ptr<Catalog>(new Catalog(*this));
   }

   return *this;
}

inline void Connection::close() noexcept {
   if (handle_) { // Lock so a concurrent cancel works
      std::lock_guard<std::mutex> lock(mutex_);
      hyper_disconnect(handle_);
      handle_ = nullptr;
   }
}

inline void Connection::cancel() noexcept {
   // Lock so no-one closes while we are cancelling.
   std::lock_guard<std::mutex> lock(mutex_);
   if (handle_) {
      if (hyper_error_t* error = hyper_cancel(handle_)) {
         // TODO 923423: Log swallowed errors?
         // Swallow any errors, as the user is not able to react anyway
         HyperException he = internal::makeHyperException(error);
      }
   }
}

inline Catalog& Connection::getCatalog() noexcept {
   HYPER_PRECONDITION_MSG(isOpen(), "The connection is closed.");
   return *catalog_;
}

inline Result Connection::executeQuery(const std::string& sql) {
   HYPER_PRECONDITION_MSG(isOpen(), "The connection is closed.");
   hyper_rowset_t* rowset = nullptr;

   if (hyper_error_t* error = hyper_execute_query(handle_, sql.c_str(), &rowset)) {
      throw internal::makeHyperException(error);
   }

   return Result{rowset, *this};
}

inline int64_t Connection::executeCommand(const std::string& sql) {
   HYPER_PRECONDITION_MSG(isOpen(), "The connection is closed.");
   int affectedRowCount = -1;
   if (hyper_error_t* error = hyper_execute_command(handle_, sql.c_str(), &affectedRowCount)) {
      throw internal::makeHyperException(error);
   }

   return affectedRowCount;
}

template <typename T>
T Connection::executeScalarQuery(const std::string& sql) {
   HYPER_PRECONDITION_MSG(isOpen(), "The connection is closed.");
   // NOLINTBEGIN(modernize-type-traits)
   static_assert(!std::is_same<T, string_view>::value, "executeScalarQuery() closes the result and cannot return non-owning types. Try std::string instead.");
   static_assert(!std::is_same<T, optional<string_view>>::value, "executeScalarQuery() closes the result and cannot return non-owning types. Try std::string instead.");
   static_assert(!std::is_same<T, ByteSpan>::value, "executeScalarQuery() closes the result and cannot return non-owning types. Try std::vector<uint8_t> instead.");
   static_assert(!std::is_same<T, optional<ByteSpan>>::value, "executeScalarQuery() closes the result and cannot return non-owning types. Try std::vector<uint8_t> instead.");
   static_assert(!std::is_same<T, Value>::value, "executeScalarQuery() closes the result and cannot return non-owning types. Try an explicit type instead.");
   // NOLINTEND(modernize-type-traits)
   Result result = executeQuery(sql);
   if (result.getSchema().getColumnCount() != 1) {
      std::stringstream ss;
      ss << "Scalar queries must return exactly one column, but " << result.getSchema().getColumnCount() << " were returned.";
      throw internal::makeHyperException(ss.str(), "", ContextId(0x93b5fde0u));
   }
   auto rowIterator = begin(result);
   if (rowIterator == end(result)) {
      throw internal::makeHyperException("Scalar query didn't return a row.", "", ContextId(0xe8a0a244u));
   }
   T val = rowIterator->get<T>(0);
   if ((++rowIterator) != end(result)) {
      throw internal::makeHyperException("Scalar query returned more than one row.", "", ContextId(0xa3b5270eu));
   }
   return val;
}

inline bool Connection::isReady() {
   HYPER_PRECONDITION_MSG(isOpen(), "The connection is closed.");
   return hyper_connection_is_ready(handle_);
}

namespace internal {
/// Enable debug output on the connection
inline void enableDebugOutput(Connection& c, FILE* debugDestination) noexcept {
   hyper_connection_trace(getHandle(c), debugDestination);
}

/// Expose the handle of the connection
inline hyper_connection_t* getHandle(Connection& c) noexcept {
   return c.handle_;
}

/// Get a parameter status from the connection
inline std::string getParameterStatus(Connection& c, const std::string& parameter) noexcept {
   const char* paramStatus = hyper_connection_parameter_status(getHandle(c), parameter.c_str());
   return paramStatus ? std::string(paramStatus) : std::string();
}

/// Get the session identifier from the connection
inline std::string getSessionIdentifier(Connection& c) noexcept {
   return getParameterStatus(c, "session_identifier");
}

/**
 * Prepares a query with the given SQL query text and registers it under the given statement name.
 *
 * Refer to the Hyper SQL documentation for the syntax and semantics of the SQL query text. //TODO(W-13456468) link?
 *
 * \param connection  The connection.
 * \param query  The SQL query text.
 * \param statement_name  The name under which the prepared statement will be registered.
 * \pre connection.is_valid()
 * \throw hyperapi::error in case of error
 */
inline void prepareQuery(Connection& connection, const std::string& statement_name, const std::string& query) {
   if (hyper_error_t* error = hyper_prepare(getHandle(connection), statement_name.c_str(), query.c_str())) {
      throw internal::makeHyperException(error);
   }
}

/**
 * Execute a previously prepared statement.
 *
 * \param connection  The connection.
 * \param statement_name  The name under which the prepared statement was registered.
 * \param result_format  The result format code (0 = text, 1 = pg_binary, 2 = hyper_binary).
 * \throw hyperapi::error in case of error
 * \pre `connection.is_valid()`
 * result consumption.
 */
inline Result executePreparedQuery(Connection& connection, const std::string& statement_name, hyper_rowset_result_format_t result_format) {
   hyper_rowset_t* rowset = nullptr;

   if (hyper_error_t* error = hyper_execute_prepared(getHandle(connection), statement_name.c_str(), result_format, &rowset)) {
      throw internal::makeHyperException(error);
   }

   return Result(rowset, connection);
}

/**
 * Execute parameterized query.
 *
 * \param connection  The connection.
 * \param result_format  The result format code (0 = text, 1 = pg_binary, 2 =  hyper_binary).
 * \throw hyperapi::error in case of error
 * \pre `connection.is_valid()`
 * result consumption.
 */
inline Result executeQueryParams(Connection& connection, const std::string& query, hyper_rowset_result_format_t result_format) {
   hyper_rowset_t* rowset = nullptr;

   if (hyper_error_t* error = hyper_execute_query_params(getHandle(connection), query.c_str(), result_format, &rowset)) {
      throw internal::makeHyperException(error);
   }

   return Result(rowset, connection);
}

/** Internal helper function that converts the c-style notice receiver signature into a c++ equivalent. */
inline void callNoticeReceiver(void* noticeReceiver, hyper_error_t* notice) noexcept {
   (*static_cast<const NoticeReceiver*>(noticeReceiver))(makeHyperException(notice));
}

/**
 * Set the notice receiver.
 *
 * \param connection  The connection.
* \param noticeReceiver  The receiver that will be called asynchronously for notices.
 * \pre `connection.is_valid()`
 */
inline void setNoticeReceiver(Connection& connection, NoticeReceiver noticeReceiver) noexcept {
   NoticeReceiver& connectionNoticeReceiver = *connection.noticeReceiver_;
   connectionNoticeReceiver = std::move(noticeReceiver);
   hyper_set_notice_receiver(getHandle(connection), &callNoticeReceiver, &connectionNoticeReceiver);
}

/** Internal helper function that converts the c-style async result callback signature into a c++ equivalent. */
inline void callAsyncResultCallback(void* asyncResultCallback, bool isLastEmptyChunk) noexcept {
   (*static_cast<const AsyncResultCallback*>(asyncResultCallback))(isLastEmptyChunk);
}

/**
 * Set the async result callback.
 *
 * \param connection  The connection.
 * \param asyncResultCallback  The callback that will be called asynchronously for new results.
 * \pre `connection.is_valid()`
 */
inline void setAsyncResultCallback(Connection& connection, AsyncResultCallback asyncResultCallback) noexcept {
   AsyncResultCallback& connectionasyncResultCallback = *connection.asyncResultCallback_;
   connectionasyncResultCallback = std::move(asyncResultCallback);
   hyper_set_rowset_async_callback(getHandle(connection), &callAsyncResultCallback, &connectionasyncResultCallback);
}

/**
 * Get the async result callback.
 *
 * \param connection  The connection.
 * \pre `connection.is_valid()`
 */
inline const AsyncResultCallback& getAsyncResultCallback(Connection& connection) noexcept {
   return *connection.asyncResultCallback_;
}

/** Set the prefetch threshold. Used for testing. */
inline void setPrefetchThreshold(Connection& c, size_t threshold) noexcept {
   return hyper_set_prefetch_threshold(getHandle(c), threshold);
}
}
}
#endif
