#include "abstract_read_write_operator.hpp"

#include <memory>
#include <ostream>

#include "concurrency/transaction_context.hpp"
#include "operators/abstract_operator.hpp"
#include "types.hpp"
#include "utils/assert.hpp"

namespace hyrise {

AbstractReadWriteOperator::AbstractReadWriteOperator(const OperatorType type,
                                                     const std::shared_ptr<const AbstractOperator>& left,
                                                     const std::shared_ptr<const AbstractOperator>& right)
    : AbstractOperator(type, left, right), _rw_state(ReadWriteOperatorState::Pending) {}

void AbstractReadWriteOperator::execute() {
  Assert(static_cast<bool>(transaction_context()),
         "AbstractReadWriteOperator::execute() should never be called without having set the transaction context.");
  Assert(transaction_context()->phase() == TransactionPhase::Active, "Transaction is not active anymore.");
  Assert(_rw_state == ReadWriteOperatorState::Pending, "Operator needs to have state Pending in order to be executed.");

  transaction_context()->register_read_write_operator(
      std::static_pointer_cast<AbstractReadWriteOperator>(shared_from_this()));

  try {
    AbstractOperator::execute();
  } catch (...) {
    // No matter what goes wrong, we need to mark the operators as failed. Otherwise, when the transaction context
    // gets destroyed, it will cause another exception that hides the one that caused the actual error. We are NOT
    // trying to handle the exception here - just making sure that we are not misled when we debug things.
    _mark_as_failed();
    throw;
  }

  if (_rw_state == ReadWriteOperatorState::Conflicted) {
    return;
  }

  _rw_state = ReadWriteOperatorState::Executed;
}

void AbstractReadWriteOperator::commit_records(const CommitID commit_id) {
  Assert(_rw_state == ReadWriteOperatorState::Executed,
         "Operator needs to have state Executed in order to be committed.");

  _on_commit_records(commit_id);
  _rw_state = ReadWriteOperatorState::Committed;
}

void AbstractReadWriteOperator::rollback_records() {
  Assert(_rw_state == ReadWriteOperatorState::Conflicted || _rw_state == ReadWriteOperatorState::Executed,
         "Operator needs to have state Failed or Executed in order to be rolled back.");

  _on_rollback_records();

  _rw_state = ReadWriteOperatorState::RolledBack;
}

bool AbstractReadWriteOperator::execute_failed() const {
  return _rw_state == ReadWriteOperatorState::Conflicted || _rw_state == ReadWriteOperatorState::RolledBack;
}

ReadWriteOperatorState AbstractReadWriteOperator::state() const {
  return _rw_state;
}

void AbstractReadWriteOperator::_mark_as_failed() {
  Assert(_rw_state == ReadWriteOperatorState::Pending, "Operator can only be marked as failed if pending.");

  _rw_state = ReadWriteOperatorState::Conflicted;
}

std::ostream& operator<<(std::ostream& stream, const ReadWriteOperatorState& phase) {
  switch (phase) {
    case ReadWriteOperatorState::Pending:
      stream << "Pending";
      break;
    case ReadWriteOperatorState::Executed:
      stream << "Executed";
      break;
    case ReadWriteOperatorState::Conflicted:
      stream << "Failed";
      break;
    case ReadWriteOperatorState::RolledBack:
      stream << "RolledBack";
      break;
    case ReadWriteOperatorState::Committed:
      stream << "Committed";
      break;
  }
  return stream;
}

}  // namespace hyrise
