// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
// vim: ts=8 sw=2 smarttab

#pragma once

#include <memory>
#include <type_traits>
#include <boost/intrusive_ptr.hpp>
#include <boost/smart_ptr/intrusive_ref_counter.hpp>
#include <boost/smart_ptr/local_shared_ptr.hpp>
#include <seastar/core/chunked_fifo.hh>
#include <seastar/core/future.hh>
#include <seastar/core/shared_future.hh>

#include "common/dout.h"
#include "crimson/net/Fwd.h"
#include "os/Transaction.h"
#include "osd/osd_types.h"
#include "crimson/osd/object_context.h"

#include "crimson/common/errorator.h"
#include "crimson/common/type_helpers.h"
#include "crimson/osd/osd_operations/client_request.h"
#include "crimson/osd/osd_operations/peering_event.h"
#include "crimson/osd/shard_services.h"
#include "crimson/osd/osdmap_gate.h"

#include "crimson/osd/pg_backend.h"
#include "crimson/osd/exceptions.h"

#include "messages/MOSDOp.h"

class PG;
class PGLSFilter;
class OSDOp;

namespace crimson::osd {

// PgOpsExecuter -- a class for executing ops targeting a certain object.
class OpsExecuter {
  using call_errorator = crimson::errorator<
    crimson::stateful_ec,
    crimson::ct_error::enoent,
    crimson::ct_error::invarg,
    crimson::ct_error::permission_denied,
    crimson::ct_error::operation_not_supported,
    crimson::ct_error::input_output_error,
    crimson::ct_error::value_too_large>;
  using read_errorator = PGBackend::read_errorator;
  using write_ertr = PGBackend::write_ertr;
  using get_attr_errorator = PGBackend::get_attr_errorator;
  using watch_errorator = crimson::errorator<
    crimson::ct_error::enoent,
    crimson::ct_error::invarg,
    crimson::ct_error::not_connected,
    crimson::ct_error::timed_out>;

public:
  // because OpsExecuter is pretty heavy-weight object we want to ensure
  // it's not copied nor even moved by accident. Performance is the sole
  // reason for prohibiting that.
  OpsExecuter(OpsExecuter&&) = delete;
  OpsExecuter(const OpsExecuter&) = delete;

  using osd_op_errorator = crimson::compound_errorator_t<
    call_errorator,
    read_errorator,
    write_ertr,
    get_attr_errorator,
    watch_errorator,
    PGBackend::stat_errorator>;

private:
  // an operation can be divided into two stages: main and effect-exposing
  // one. The former is performed immediately on call to `do_osd_op()` while
  // the later on `submit_changes()` – after successfully processing main
  // stages of all involved operations. When any stage fails, none of all
  // scheduled effect-exposing stages will be executed.
  // when operation requires this division, some variant of `with_effect()`
  // should be used.
  struct effect_t {
    virtual osd_op_errorator::future<> execute() = 0;
    virtual ~effect_t() = default;
  };

  ObjectContextRef obc;
  const OpInfo& op_info;
  const pg_pool_t& pool_info;  // for the sake of the ObjClass API
  PGBackend& backend;
  const MOSDOp& msg;
  std::optional<osd_op_params_t> osd_op_params;
  bool user_modify = false;
  ceph::os::Transaction txn;

  size_t num_read = 0;    ///< count read ops
  size_t num_write = 0;   ///< count update ops

  // this gizmo could be wrapped in std::optional for the sake of lazy
  // initialization. we don't need it for ops that doesn't have effect
  // TODO: verify the init overhead of chunked_fifo
  seastar::chunked_fifo<std::unique_ptr<effect_t>> op_effects;

  template <class Context, class MainFunc, class EffectFunc>
  auto with_effect_on_obc(
    Context&& ctx,
    MainFunc&& main_func,
    EffectFunc&& effect_func);

  call_errorator::future<> do_op_call(class OSDOp& osd_op);
  watch_errorator::future<> do_op_watch(
    class OSDOp& osd_op,
    class ObjectState& os,
    ceph::os::Transaction& txn);
  watch_errorator::future<> do_op_watch_subop_watch(
    class OSDOp& osd_op,
    class ObjectState& os,
    ceph::os::Transaction& txn);
  watch_errorator::future<> do_op_watch_subop_reconnect(
    class OSDOp& osd_op,
    class ObjectState& os,
    ceph::os::Transaction& txn);
  watch_errorator::future<> do_op_watch_subop_unwatch(
    class OSDOp& osd_op,
    class ObjectState& os,
    ceph::os::Transaction& txn);
  watch_errorator::future<> do_op_watch_subop_ping(
    class OSDOp& osd_op,
    class ObjectState& os,
    ceph::os::Transaction& txn);
  watch_errorator::future<> do_op_notify(
    class OSDOp& osd_op,
    const class ObjectState& os);
  watch_errorator::future<> do_op_notify_ack(
    class OSDOp& osd_op,
    const class ObjectState& os);

  hobject_t &get_target() const {
    return obc->obs.oi.soid;
  }

  template <class Func>
  auto do_const_op(Func&& f) {
    // TODO: pass backend as read-only
    return std::forward<Func>(f)(backend, std::as_const(obc->obs));
  }

  template <class Func>
  auto do_read_op(Func&& f) {
    ++num_read;
    // TODO: pass backend as read-only
    return do_const_op(std::forward<Func>(f));
  }

  template <class Func>
  auto do_write_op(Func&& f, bool um) {
    ++num_write;
    if (!osd_op_params) {
      osd_op_params.emplace();
    }
    user_modify = um;
    return std::forward<Func>(f)(backend, obc->obs, txn);
  }

  decltype(auto) dont_do_legacy_op() {
    return crimson::ct_error::operation_not_supported::make();
  }

public:
  OpsExecuter(ObjectContextRef obc,
              const OpInfo& op_info,
              const pg_pool_t& pool_info,
              PGBackend& backend,
              const MOSDOp& msg)
    : obc(std::move(obc)),
      op_info(op_info),
      pool_info(pool_info),
      backend(backend),
      msg(msg) {
  }

  osd_op_errorator::future<> execute_op(class OSDOp& osd_op);

  template <typename Func, typename MutFunc>
  osd_op_errorator::future<> flush_changes(Func&& func, MutFunc&& mut_func) &&;

  const auto& get_message() const {
    return msg;
  }

  size_t get_processed_rw_ops_num() const {
    return num_read + num_write;
  }

  uint32_t get_pool_stripe_width() const {
    return pool_info.get_stripe_width();
  }

  bool has_seen_write() const {
    return num_write > 0;
  }
};

template <class Context, class MainFunc, class EffectFunc>
auto OpsExecuter::with_effect_on_obc(
  Context&& ctx,
  MainFunc&& main_func,
  EffectFunc&& effect_func)
{
  using context_t = std::decay_t<Context>;
  // the language offers implicit conversion to pointer-to-function for
  // lambda only when it's closureless. We enforce this restriction due
  // the fact that `flush_changes()` std::moves many executer's parts.
  using allowed_effect_func_t =
    seastar::future<> (*)(context_t&&, ObjectContextRef);
  static_assert(std::is_convertible_v<EffectFunc, allowed_effect_func_t>,
                "with_effect function is not allowed to capture");
  struct task_t final : effect_t {
    context_t ctx;
    EffectFunc effect_func;
    ObjectContextRef obc;

    task_t(Context&& ctx, EffectFunc&& effect_func, ObjectContextRef obc)
       : ctx(std::move(ctx)),
         effect_func(std::move(effect_func)),
         obc(std::move(obc)) {
    }
    osd_op_errorator::future<> execute() final {
      return std::move(effect_func)(std::move(ctx), std::move(obc));
    }
  };
  auto task =
    std::make_unique<task_t>(std::move(ctx), std::move(effect_func), obc);
  auto& ctx_ref = task->ctx;
  op_effects.emplace_back(std::move(task));
  return std::forward<MainFunc>(main_func)(ctx_ref);
}

template <typename Func,
          typename MutFunc>
OpsExecuter::osd_op_errorator::future<> OpsExecuter::flush_changes(
  Func&& func,
  MutFunc&& mut_func) &&
{
  const bool want_mutate = !txn.empty();
  // osd_op_params are instantiated by every wr-like operation.
  assert(osd_op_params || !want_mutate);
  assert(obc);
  if (__builtin_expect(op_effects.empty(), true)) {
    return want_mutate ? std::forward<MutFunc>(mut_func)(std::move(txn),
                                                         std::move(obc),
                                                         std::move(*osd_op_params),
                                                         user_modify)
                       : std::forward<Func>(func)(std::move(obc));
  } else {
    return (want_mutate ? std::forward<MutFunc>(mut_func)(std::move(txn),
                                                          std::move(obc),
                                                          std::move(*osd_op_params),
                                                          user_modify)
                        : std::forward<Func>(func)(std::move(obc))
    ).safe_then([this] {
      // let's do the cleaning of `op_effects` in destructor
      return crimson::do_for_each(op_effects, [] (auto& op_effect) {
        return op_effect->execute();
      });
    });
  }
}

// PgOpsExecuter -- a class for executing ops targeting a certain PG.
class PgOpsExecuter {
public:
  PgOpsExecuter(const PG& pg, const MOSDOp& msg)
    : pg(pg), nspace(msg.get_hobj().nspace) {
  }

  seastar::future<> execute_op(class OSDOp& osd_op);

private:
  const PG& pg;
  const std::string& nspace;
};

} // namespace crimson::osd