/*
 * 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 "CacheTransactionManagerImpl.hpp"

#include <geode/ExceptionTypes.hpp>
#include <geode/TransactionId.hpp>

#include "CacheImpl.hpp"
#include "CacheRegionHelper.hpp"
#include "SuspendedTxExpiryTask.hpp"
#include "TSSTXStateWrapper.hpp"
#include "TXCleaner.hpp"
#include "TcrMessage.hpp"
#include "ThinClientBaseDM.hpp"
#include "ThinClientPoolDM.hpp"
#include "TssConnectionWrapper.hpp"
#include "util/exception.hpp"

namespace apache {
namespace geode {
namespace client {

CacheTransactionManagerImpl::CacheTransactionManagerImpl(CacheImpl* cache)
    : m_cache(cache) {}

CacheTransactionManagerImpl::~CacheTransactionManagerImpl() {}

void CacheTransactionManagerImpl::begin() {
  if (TSSTXStateWrapper::get().getTXState() != nullptr) {
    GfErrTypeThrowException("Transaction already in progress",
                            GF_CACHE_ILLEGAL_STATE_EXCEPTION);
  }
  auto txState = new TXState(m_cache);
  TSSTXStateWrapper::get().setTXState(txState);
  addTx(txState->getTransactionId().getId());
}

void CacheTransactionManagerImpl::commit() {
  TXCleaner txCleaner(this);
  auto txState = txCleaner.getTXState();

  if (txState == nullptr) {
    GfErrTypeThrowException(
        "Transaction is null, cannot commit a null transaction",
        GF_CACHE_ILLEGAL_STATE_EXCEPTION);
  }

  TcrMessageCommit request(new DataOutput(m_cache->createDataOutput()));
  TcrMessageReply reply(true, nullptr);

  auto tcr_dm = getDM();
  // This is for the case when no cache operation/s is performed between
  // tx->begin() and tx->commit()/rollback(),
  // simply return without sending COMMIT message to server. tcr_dm is nullptr
  // implies no cache operation is performed.
  // Theres no need to call txCleaner.clean(); here, because TXCleaner
  // destructor is called which cleans ThreadLocal.
  if (tcr_dm == nullptr) {
    return;
  }

  GfErrType err = tcr_dm->sendSyncRequest(request, reply);

  if (err != GF_NOERR) {
    // err = rollback(txState, false);
    //		noteCommitFailure(txState, nullptr);
    GfErrTypeThrowException("Error while committing", err);
  } else {
    switch (reply.getMessageType()) {
      case TcrMessage::RESPONSE: {
        break;
      }
      case TcrMessage::EXCEPTION: {
        //			noteCommitFailure(txState, nullptr);
        const auto& exceptionMsg = reply.getException();
        err = ThinClientRegion::handleServerException(
            "CacheTransactionManager::commit", exceptionMsg);
        GfErrTypeThrowException("Commit Failed", err);
        break;
      }
      case TcrMessage::COMMIT_ERROR: {
        //			noteCommitFailure(txState, nullptr);
        GfErrTypeThrowException("Commit Failed", GF_COMMIT_CONFLICT_EXCEPTION);
        break;
      }
      default: {
        //			noteCommitFailure(txState, nullptr);
        LOGERROR("Unknown message type in commit reply %d",
                 reply.getMessageType());
        GfErrTypeThrowException("Commit Failed", GF_MSG);
        break;
      }
    }
  }

  auto commit = std::dynamic_pointer_cast<TXCommitMessage>(reply.getValue());
  txCleaner.clean();
  commit->apply(m_cache->getCache());
}

void CacheTransactionManagerImpl::rollback() {
  TXCleaner txCleaner(this);
  TXState* txState = txCleaner.getTXState();

  if (txState == nullptr) {
    GfErrTypeThrowException("Thread does not have an active transaction",
                            GF_CACHE_ILLEGAL_STATE_EXCEPTION);
  }

  try {
    GfErrType err = rollback(txState, true);
    if (err != GF_NOERR) {
      throwExceptionIfError("Error while committing", err);
    }
  } catch (const Exception& ex) {
    // TODO: put a log message
    throw ex;
  } catch (...) {
    // TODO: put a log message
    throw;
  }
}

GfErrType CacheTransactionManagerImpl::rollback(TXState*, bool) {
  TcrMessageRollback request(new DataOutput(m_cache->createDataOutput()));
  TcrMessageReply reply(true, nullptr);
  GfErrType err = GF_NOERR;
  ThinClientPoolDM* tcr_dm = getDM();
  // This is for the case when no cache operation/s is performed between
  // tx->begin() and tx->commit()/rollback(),
  // simply return without sending COMMIT message to server. tcr_dm is nullptr
  // implies no cache operation is performed.
  // Theres no need to call txCleaner.clean(); here, because TXCleaner
  // destructor is called which cleans ThreadLocal.
  if (tcr_dm == nullptr) {
    return err;
  }
  err = tcr_dm->sendSyncRequest(request, reply);

  if (err == GF_NOERR) {
    switch (reply.getMessageType()) {
      case TcrMessage::REPLY: {
        break;
      }
      case TcrMessage::EXCEPTION: {
        break;
      }
      default: {
        break;
      }
    }
  }

  /*	if(err == GF_NOERR && callListener)
          {
  //	auto commit =
  std::static_pointer_cast<TXCommitMessage>(reply.getValue());
                  noteRollbackSuccess(txState, nullptr);
          }
  */
  return err;
}

ThinClientPoolDM* CacheTransactionManagerImpl::getDM() {
  if (auto conn = TssConnectionWrapper::get().getConnection()) {
    if (auto dm = conn->getEndpointObject()->getPoolHADM()) {
      return dm;
    }
  }
  return nullptr;
}

Cache* CacheTransactionManagerImpl::getCache() { return m_cache->getCache(); }
TransactionId& CacheTransactionManagerImpl::suspend() {
  // get the current state of the thread
  auto txState = TSSTXStateWrapper::get().getTXState();
  if (txState == nullptr) {
    LOGFINE("Transaction not in progress. Returning nullptr transaction Id.");
    throw TransactionException("Transaction not in progress.");
  }

  // get the current connection that this transaction is using
  auto conn = TssConnectionWrapper::get().getConnection();
  if (!conn) {
    LOGFINE(
        "Thread local connection is null. Returning nullptr transaction Id.");
    throw TransactionException("Thread local connection is null.");
  }

  // get the endpoint info from the connection
  TcrEndpoint* ep = conn->getEndpointObject();

  // store the endpoint info and the pool DM in the transaction state
  // this function setEPStr and setPoolDM is used only while suspending
  // the transaction. The info stored here is used during resume.
  txState->setEPStr(ep->name());
  txState->setPoolDM(ep->getPoolHADM());

  LOGFINE(
      "suspended Release the sticky connection associated with the "
      "transaction");
  txState->releaseStickyConnection();

  // set the expiry handler for the suspended transaction
  auto timeout = m_cache->getDistributedSystem()
                     .getSystemProperties()
                     .suspendedTxTimeout();
  auto& manager = m_cache->getExpiryTaskManager();
  auto task = std::make_shared<SuspendedTxExpiryTask>(
      manager, *this, txState->getTransactionId());
  auto id = manager.schedule(std::move(task), timeout);
  txState->setSuspendedExpiryTaskId(id);

  // add the transaction state to the list of suspended transactions
  addSuspendedTx(txState->getTransactionId().getId(), txState);

  // set the current transaction state as null
  TSSTXStateWrapper::get().setTXState(nullptr);

  // return the transaction ID
  return static_cast<TransactionId&>(txState->getTransactionId());
}

void CacheTransactionManagerImpl::resume(TransactionId& transactionId) {
  // get the current state of the thread
  if (TSSTXStateWrapper::get().getTXState() != nullptr) {
    GfErrTypeThrowException("A transaction is already in progress",
                            GF_CACHE_ILLEGAL_STATE_EXCEPTION);
  }

  // get the transaction state of the suspended transaction
  TXState* txState =
      removeSuspendedTx((static_cast<TXId&>(transactionId)).getId());
  if (txState == nullptr) {
    GfErrTypeThrowException(
        "Could not get transaction state for the transaction id.",
        GF_CACHE_ILLEGAL_STATE_EXCEPTION);
  }

  resumeTxUsingTxState(txState);
}

bool CacheTransactionManagerImpl::isSuspended(TransactionId& transactionId) {
  return isSuspendedTx((static_cast<TXId&>(transactionId)).getId());
}
bool CacheTransactionManagerImpl::tryResume(TransactionId& transactionId) {
  return tryResume(transactionId, true);
}
bool CacheTransactionManagerImpl::tryResume(TransactionId& transactionId,
                                            bool cancelExpiryTask) {
  // get the current state of the thread
  if (TSSTXStateWrapper::get().getTXState() != nullptr) {
    LOGFINE("A transaction is already in progress. Cannot resume transaction.");
    return false;
  }

  // get the transaction state of the suspended transaction
  TXState* txState =
      removeSuspendedTx((static_cast<TXId&>(transactionId)).getId());
  if (txState == nullptr) return false;

  resumeTxUsingTxState(txState, cancelExpiryTask);
  return true;
}

bool CacheTransactionManagerImpl::tryResume(
    TransactionId& transactionId, std::chrono::milliseconds waitTime) {
  // get the current state of the thread
  if (TSSTXStateWrapper::get().getTXState() != nullptr) {
    LOGFINE("A transaction is already in progress. Cannot resume transaction.");
    return false;
  }

  if (!exists(transactionId)) return false;

  // get the transaction state of the suspended transaction
  TXState* txState =
      removeSuspendedTx((static_cast<TXId&>(transactionId)).getId(), waitTime);
  if (txState == nullptr) return false;

  resumeTxUsingTxState(txState);
  return true;
}

void CacheTransactionManagerImpl::resumeTxUsingTxState(TXState* txState,
                                                       bool cancelExpiryTask) {
  if (txState == nullptr) return;

  TcrConnection* conn;

  LOGDEBUG("Resuming transaction for tid: %d",
           txState->getTransactionId().getId());

  if (cancelExpiryTask) {
    // cancel the expiry task for the transaction
    m_cache->getExpiryTaskManager().cancel(txState->getSuspendedExpiryTaskId());
  }

  // set the current state as the state of the suspended transaction
  TSSTXStateWrapper::get().setTXState(txState);

  LOGFINE("Get connection for transaction id %d",
          txState->getTransactionId().getId());
  // get connection to the endpoint specified in the transaction state
  GfErrType error = txState->getPoolDM()->getConnectionToAnEndPoint(
      txState->getEPStr(), conn);
  if (conn == nullptr || error != GF_NOERR) {
    // throw an exception and set the current state as nullptr because
    // the transaction cannot be resumed
    TSSTXStateWrapper::get().setTXState(nullptr);
    GfErrTypeThrowException(
        "Could not get a connection for the transaction id.",
        GF_CACHE_ILLEGAL_STATE_EXCEPTION);
  } else {
    txState->getPoolDM()->setThreadLocalConnection(conn);
    LOGFINE("Set the thread local connection for transaction id %d",
            txState->getTransactionId().getId());
  }
}

bool CacheTransactionManagerImpl::exists(TransactionId& transactionId) {
  return findTx((static_cast<TXId&>(transactionId)).getId());
}

bool CacheTransactionManagerImpl::exists() {
  return TSSTXStateWrapper::get().getTXState() != nullptr;
}

void CacheTransactionManagerImpl::addTx(int32_t txId) {
  std::unique_lock<decltype(m_txLock)> _guard(m_txLock);
  m_TXs.push_back(txId);
}

void CacheTransactionManagerImpl::removeTx(int32_t txId) {
  std::unique_lock<decltype(m_txLock)> _guard(m_txLock);
  m_TXs.erase(std::remove(m_TXs.begin(), m_TXs.end(), txId), m_TXs.end());
}
bool CacheTransactionManagerImpl::findTx(int32_t txId) {
  std::unique_lock<decltype(m_txLock)> _guard(m_txLock);
  return std::find(m_TXs.begin(), m_TXs.end(), txId) != m_TXs.end();
}

void CacheTransactionManagerImpl::addSuspendedTx(int32_t txId,
                                                 TXState* txState) {
  std::unique_lock<decltype(m_suspendedTxLock)> _guard(m_suspendedTxLock);
  m_suspendedTXs[txId] = txState;
  m_txCond.notify_all();
}

TXState* CacheTransactionManagerImpl::getSuspendedTx(int32_t txId) {
  std::unique_lock<decltype(m_suspendedTxLock)> _guard(m_suspendedTxLock);
  auto&& it = m_suspendedTXs.find(txId);
  if (it == m_suspendedTXs.end()) return nullptr;
  auto rettxState = (*it).second;
  return rettxState;
}

TXState* CacheTransactionManagerImpl::removeSuspendedTx(int32_t txId) {
  std::unique_lock<decltype(m_suspendedTxLock)> _guard(m_suspendedTxLock);
  auto&& it = m_suspendedTXs.find(txId);
  if (it == m_suspendedTXs.end()) return nullptr;
  auto rettxState = (*it).second;
  m_suspendedTXs.erase(it);
  return rettxState;
}

TXState* CacheTransactionManagerImpl::removeSuspendedTx(
    int32_t txId, std::chrono::milliseconds waitTime) {
  std::unique_lock<decltype(m_suspendedTxLock)> _guard(m_suspendedTxLock);

  auto txState = removeSuspendedTx(txId);
  if (txState == nullptr) {
    LOGFINE("Wait for the connection to get suspended, Tid: %d", txId);
    m_txCond.wait_for(_guard, waitTime, [this, txId, &txState] {
      return nullptr != (txState = removeSuspendedTx(txId));
    });
  }

  return txState;
}

bool CacheTransactionManagerImpl::isSuspendedTx(int32_t txId) {
  std::unique_lock<decltype(m_suspendedTxLock)> _guard(m_suspendedTxLock);
  auto&& it = m_suspendedTXs.find(txId);
  if (it == m_suspendedTXs.end()) {
    return false;
  } else {
    return true;
  }
}

TransactionId& CacheTransactionManagerImpl::getTransactionId() {
  auto txState = TSSTXStateWrapper::get().getTXState();
  return txState->getTransactionId();
}

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