// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
// vim: ts=8 sw=2 smarttab
/*
 * Ceph - scalable distributed file system
 *
 * Copyright (C) 2018 Red Hat <contact@redhat.com>
 * Author: Adam C. Emerson
 *
 * This is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License version 2.1, as published by the Free Software
 * Foundation.  See file COPYING.
 *
 */

#ifndef NEORADOS_RADOS_HPP
#define NEORADOS_RADOS_HPP

#include <cstddef>
#include <memory>
#include <tuple>
#include <string>
#include <string_view>
#include <type_traits>
#include <variant>

#include <boost/asio.hpp>

#include <boost/container/flat_map.hpp>
#include <boost/container/flat_set.hpp>
#include <boost/uuid/uuid.hpp>

#include <boost/system/error_code.hpp>

// Will be in C++20!

#include "include/expected.hpp"

// Had better be in C++20. Why is this not in Boost?

#include "include/function2.hpp"

// Things broken out so we can decode them in Objecter.

#include "include/neorados/RADOS_Decodable.hpp"

// Needed for type erasure and template support. We can't really avoid
// it.

#include "common/async/completion.h"

// These are needed for RGW, but in general as a 'shiny new interface'
// we should try to use forward declarations and provide standard alternatives.

#include "include/common_fwd.h"

#include "include/buffer.h"
#include "include/rados/librados_fwd.hpp"

#include "common/ceph_time.h"

namespace neorados {
class Object;
class IOContext;
}
namespace std {
template<>
struct hash<neorados::Object>;
template<>
struct hash<neorados::IOContext>;
}

namespace neorados {
namespace detail {
class Client;
}

class RADOS;

// Exists mostly so that repeated operations on the same object don't
// have to pay for the string copy to construct an object_t.

class Object final {
  friend RADOS;
  friend std::hash<Object>;

public:
  Object();
  Object(const char* s);
  Object(std::string_view s);
  Object(std::string&& s);
  Object(const std::string& s);
  ~Object();

  Object(const Object& o);
  Object& operator =(const Object& o);

  Object(Object&& o);
  Object& operator =(Object&& o);

  operator std::string_view() const;

  friend std::ostream& operator <<(std::ostream& m, const Object& o);
  friend bool operator <(const Object& lhs, const Object& rhs);
  friend bool operator <=(const Object& lhs, const Object& rhs);
  friend bool operator >=(const Object& lhs, const Object& rhs);
  friend bool operator >(const Object& lhs, const Object& rhs);

  friend bool operator ==(const Object& lhs, const Object& rhs);
  friend bool operator !=(const Object& lhs, const Object& rhs);

private:

  static constexpr std::size_t impl_size = 4 * 8;
  std::aligned_storage_t<impl_size> impl;
};

// Not the same as the librados::IoCtx, but it does gather together
// some of the same metadata. Since we're likely to do multiple
// operations in the same pool or namespace, it doesn't make sense to
// redo a bunch of lookups and string copies.

class IOContext final {
  friend RADOS;
  friend std::hash<IOContext>;

public:

  IOContext();
  explicit IOContext(std::int64_t pool);
  IOContext(std::int64_t _pool, std::string_view _ns);
  IOContext(std::int64_t _pool, std::string&& _ns);
  ~IOContext();

  IOContext(const IOContext& rhs);
  IOContext& operator =(const IOContext& rhs);

  IOContext(IOContext&& rhs);
  IOContext& operator =(IOContext&& rhs);

  std::int64_t pool() const;
  void pool(std::int64_t _pool);

  std::string_view ns() const;
  void ns(std::string_view _ns);
  void ns(std::string&& _ns);

  std::optional<std::string_view> key() const;
  void key(std::string_view _key);
  void key(std::string&& _key);
  void clear_key();

  std::optional<std::int64_t> hash() const;
  void hash(std::int64_t _hash);
  void clear_hash();

  std::optional<std::uint64_t> read_snap() const;
  void read_snap(std::optional<std::uint64_t> _snapid);

  // I can't actually move-construct here since snapid_t is its own
  // separate class type, not an alias.
  std::optional<
    std::pair<std::uint64_t,
	      std::vector<std::uint64_t>>> write_snap_context() const;
  void write_snap_context(std::optional<
			  std::pair<std::uint64_t,
			              std::vector<std::uint64_t>>> snapc);

  bool full_try() const;
  void full_try(bool _full_try);

  friend std::ostream& operator <<(std::ostream& m, const IOContext& o);
  friend bool operator <(const IOContext& lhs, const IOContext& rhs);
  friend bool operator <=(const IOContext& lhs, const IOContext& rhs);
  friend bool operator >=(const IOContext& lhs, const IOContext& rhs);
  friend bool operator >(const IOContext& lhs, const IOContext& rhs);

  friend bool operator ==(const IOContext& lhs, const IOContext& rhs);
  friend bool operator !=(const IOContext& lhs, const IOContext& rhs);

private:

  static constexpr std::size_t impl_size = 16 * 8;
  std::aligned_storage_t<impl_size> impl;
};

inline constexpr std::string_view all_nspaces("\001");

enum class cmpxattr_op : std::uint8_t {
  eq  = 1,
  ne  = 2,
  gt  = 3,
  gte = 4,
  lt  = 5,
  lte = 6
};

namespace alloc_hint {
enum alloc_hint_t {
  sequential_write = 1,
  random_write = 2,
  sequential_read = 4,
  random_read = 8,
  append_only = 16,
  immutable = 32,
  shortlived = 64,
  longlived = 128,
  compressible = 256,
  incompressible = 512
};
}

class Op {
  friend RADOS;

public:

  Op(const Op&) = delete;
  Op& operator =(const Op&) = delete;
  Op(Op&&);
  Op& operator =(Op&&);
  ~Op();

  void set_excl();
  void set_failok();
  void set_fadvise_random();
  void set_fadvise_sequential();
  void set_fadvise_willneed();
  void set_fadvise_dontneed();
  void set_fadvise_nocache();

  void cmpext(uint64_t off, ceph::buffer::list&& cmp_bl, std::size_t* s);
  void cmpxattr(std::string_view name, cmpxattr_op op,
		const ceph::buffer::list& val);
  void cmpxattr(std::string_view name, cmpxattr_op op, std::uint64_t val);
  void assert_version(uint64_t ver);
  void assert_exists();
  void cmp_omap(const boost::container::flat_map<
		  std::string,
		  std::pair<ceph::buffer::list, int>>& assertions);

  void exec(std::string_view cls, std::string_view method,
	    const ceph::buffer::list& inbl,
	    ceph::buffer::list* out,
	    boost::system::error_code* ec = nullptr);
  void exec(std::string_view cls, std::string_view method,
	    const ceph::buffer::list& inbl,
	    fu2::unique_function<void(boost::system::error_code,
				      const ceph::buffer::list&) &&> f);
  void exec(std::string_view cls, std::string_view method,
	    const ceph::buffer::list& inbl,
	    fu2::unique_function<void(boost::system::error_code, int,
				      const ceph::buffer::list&) &&> f);
  void exec(std::string_view cls, std::string_view method,
	    const ceph::buffer::list& inbl,
	    boost::system::error_code* ec = nullptr);


  // Flags that apply to all ops in the operation vector
  void balance_reads();
  void localize_reads();
  void order_reads_writes();
  void ignore_cache();
  void skiprwlocks();
  void ignore_overlay();
  void full_try();
  void full_force();
  void ignore_redirect();
  void ordersnap();
  void returnvec();

  std::size_t size() const;
  using Signature = void(boost::system::error_code);
  using Completion = ceph::async::Completion<Signature>;

  friend std::ostream& operator <<(std::ostream& m, const Op& o);
protected:
  Op();
  static constexpr std::size_t impl_size = 85 * 8;
  std::aligned_storage_t<impl_size> impl;
};

// This class is /not/ thread-safe. If you want you can wrap it in
// something that locks it.

class ReadOp final : public Op {
  friend RADOS;

public:

  ReadOp() = default;
  ReadOp(const ReadOp&) = delete;
  ReadOp(ReadOp&&) = default;

  ReadOp& operator =(const ReadOp&) = delete;
  ReadOp& operator =(ReadOp&&) = default;

  void read(size_t off, uint64_t len, ceph::buffer::list* out,
	    boost::system::error_code* ec = nullptr);
  void get_xattr(std::string_view name, ceph::buffer::list* out,
		 boost::system::error_code* ec = nullptr);
  void get_omap_header(ceph::buffer::list*,
		       boost::system::error_code* ec = nullptr);

  void sparse_read(uint64_t off, uint64_t len,
		   ceph::buffer::list* out,
		   std::vector<std::pair<std::uint64_t, std::uint64_t>>* extents,
		   boost::system::error_code* ec = nullptr);

  void stat(std::uint64_t* size, ceph::real_time* mtime,
	    boost::system::error_code* ec = nullptr);

  void get_omap_keys(std::optional<std::string_view> start_after,
		     std::uint64_t max_return,
		     boost::container::flat_set<std::string>* keys,
		     bool* truncated,
		     boost::system::error_code* ec = nullptr);


  void get_xattrs(boost::container::flat_map<std::string,
		                             ceph::buffer::list>* kv,
		     boost::system::error_code* ec = nullptr);

  void get_omap_vals(std::optional<std::string_view> start_after,
		     std::optional<std::string_view> filter_prefix,
		     uint64_t max_return,
		     boost::container::flat_map<std::string,
		                                ceph::buffer::list>* kv,
		     bool* truncated,
		     boost::system::error_code* ec = nullptr);


  void get_omap_vals_by_keys(const boost::container::flat_set<std::string>& keys,
			     boost::container::flat_map<std::string,
			                                ceph::buffer::list>* kv,
			     boost::system::error_code* ec = nullptr);

  void list_watchers(std::vector<struct ObjWatcher>* watchers,
		     boost::system::error_code* ec = nullptr);

  void list_snaps(struct SnapSet* snaps,
		  boost::system::error_code* ec = nullptr);
};

class WriteOp final : public Op {
  friend RADOS;
public:

  WriteOp() = default;
  WriteOp(const WriteOp&) = delete;
  WriteOp(WriteOp&&) = default;

  WriteOp& operator =(const WriteOp&) = delete;
  WriteOp& operator =(WriteOp&&) = default;

  void set_mtime(ceph::real_time t);
  void create(bool exclusive);
  void write(uint64_t off, ceph::buffer::list&& bl);
  void write_full(ceph::buffer::list&& bl);
  void writesame(std::uint64_t off, std::uint64_t write_len,
		 ceph::buffer::list&& bl);
  void append(ceph::buffer::list&& bl);
  void remove();
  void truncate(uint64_t off);
  void zero(uint64_t off, uint64_t len);
  void rmxattr(std::string_view name);
  void setxattr(std::string_view name,
		ceph::buffer::list&& bl);
  void rollback(uint64_t snapid);
  void set_omap(const boost::container::flat_map<std::string,
		                                 ceph::buffer::list>& map);
  void set_omap_header(ceph::buffer::list&& bl);
  void clear_omap();
  void rm_omap_keys(const boost::container::flat_set<std::string>& to_rm);
  void set_alloc_hint(uint64_t expected_object_size,
		      uint64_t expected_write_size,
		      alloc_hint::alloc_hint_t flags);
};


struct FSStats {
  uint64_t kb;
  uint64_t kb_used;
  uint64_t kb_avail;
  uint64_t num_objects;
};

// From librados.h, maybe move into a common file. But I want to see
// if we need/want to amend/add/remove anything first.
struct PoolStats {
  /// space used in bytes
  uint64_t num_bytes;
  /// space used in KB
  uint64_t num_kb;
  /// number of objects in the pool
  uint64_t num_objects;
  /// number of clones of objects
  uint64_t num_object_clones;
  /// num_objects * num_replicas
  uint64_t num_object_copies;
  /// number of objects missing on primary
  uint64_t num_objects_missing_on_primary;
  /// number of objects found on no OSDs
  uint64_t num_objects_unfound;
  /// number of objects replicated fewer times than they should be
  /// (but found on at least one OSD)
  uint64_t num_objects_degraded;
  /// number of objects read
  uint64_t num_rd;
  /// objects read in KB
  uint64_t num_rd_kb;
  /// number of objects written
  uint64_t num_wr;
  /// objects written in KB
  uint64_t num_wr_kb;
  /// bytes originally provided by user
  uint64_t num_user_bytes;
  /// bytes passed compression
  uint64_t compressed_bytes_orig;
  /// bytes resulted after compression
  uint64_t compressed_bytes;
  /// bytes allocated at storage
  uint64_t compressed_bytes_alloc;
};

// Placement group, for PG commands
struct PG {
  uint64_t pool;
  uint32_t seed;
};

class Cursor final {
public:
  static Cursor begin();
  static Cursor end();

  Cursor();
  Cursor(const Cursor&);
  Cursor& operator =(const Cursor&);
  Cursor(Cursor&&);
  Cursor& operator =(Cursor&&);
  ~Cursor();

  friend bool operator ==(const Cursor& lhs,
			  const Cursor& rhs);
  friend bool operator !=(const Cursor& lhs,
			  const Cursor& rhs);
  friend bool operator <(const Cursor& lhs,
			 const Cursor& rhs);
  friend bool operator <=(const Cursor& lhs,
			  const Cursor& rhs);
  friend bool operator >=(const Cursor& lhs,
			  const Cursor& rhs);
  friend bool operator >(const Cursor& lhs,
			 const Cursor& rhs);

  std::string to_str() const;
  static std::optional<Cursor> from_str(const std::string& s);

private:
  struct end_magic_t {};
  Cursor(end_magic_t);
  Cursor(void*);
  friend RADOS;
  static constexpr std::size_t impl_size = 16 * 8;
  std::aligned_storage_t<impl_size> impl;
};

class RADOS final
{
public:
  static constexpr std::tuple<uint32_t, uint32_t, uint32_t> version() {
    return {0, 0, 1};
  }

  using BuildSig = void(boost::system::error_code, RADOS);
  using BuildComp = ceph::async::Completion<BuildSig>;
  class Builder {
    std::optional<std::string> conf_files;
    std::optional<std::string> cluster;
    std::optional<std::string> name;
    std::vector<std::pair<std::string, std::string>> configs;
    bool no_default_conf = false;
    bool no_mon_conf = false;

  public:
    Builder() = default;
    Builder& add_conf_file(std::string_view v);
    Builder& set_cluster(std::string_view c) {
      cluster = std::string(c);
      return *this;
    }
    Builder& set_name(std::string_view n) {
      name = std::string(n);
      return *this;
    }
    Builder& set_no_default_conf() {
      no_default_conf = true;
      return *this;
    }
    Builder& set_no_mon_conf() {
      no_mon_conf = true;
      return *this;
    }
    Builder& set_conf_option(std::string_view opt, std::string_view val) {
      configs.emplace_back(std::string(opt), std::string(val));
      return *this;
    }

    template<typename CompletionToken>
    auto build(boost::asio::io_context& ioctx, CompletionToken&& token) {
      boost::asio::async_completion<CompletionToken, BuildSig> init(token);
      build(ioctx,
	    BuildComp::create(ioctx.get_executor(),
			      std::move(init.completion_handler)));
      return init.result.get();
    }

  private:
    void build(boost::asio::io_context& ioctx,
	       std::unique_ptr<BuildComp> c);
  };


  template<typename CompletionToken>
  static auto make_with_cct(CephContext* cct,
			    boost::asio::io_context& ioctx,
			    CompletionToken&& token) {
    boost::asio::async_completion<CompletionToken, BuildSig> init(token);
    make_with_cct(cct, ioctx,
		  BuildComp::create(ioctx.get_executor(),
				    std::move(init.completion_handler)));
    return init.result.get();
  }

  static RADOS make_with_librados(librados::Rados& rados);

  RADOS(const RADOS&) = delete;
  RADOS& operator =(const RADOS&) = delete;

  RADOS(RADOS&&);
  RADOS& operator =(RADOS&&);

  ~RADOS();

  CephContext* cct();

  using executor_type = boost::asio::io_context::executor_type;
  executor_type get_executor() const;
  boost::asio::io_context& get_io_context();

  template<typename CompletionToken>
  auto execute(const Object& o, const IOContext& ioc, ReadOp&& op,
	       ceph::buffer::list* bl,
	       CompletionToken&& token, uint64_t* objver = nullptr,
	       const blkin_trace_info* trace_info = nullptr) {
    boost::asio::async_completion<CompletionToken, Op::Signature> init(token);
    execute(o, ioc, std::move(op), bl,
	    ReadOp::Completion::create(get_executor(),
				       std::move(init.completion_handler)),
	    objver, trace_info);
    return init.result.get();
  }

  template<typename CompletionToken>
  auto execute(const Object& o, const IOContext& ioc, WriteOp&& op,
	       CompletionToken&& token, uint64_t* objver = nullptr,
	       const blkin_trace_info* trace_info = nullptr) {
    boost::asio::async_completion<CompletionToken, Op::Signature> init(token);
    execute(o, ioc, std::move(op),
	    Op::Completion::create(get_executor(),
				   std::move(init.completion_handler)),
	    objver, trace_info);
    return init.result.get();
  }

  template<typename CompletionToken>
  auto execute(const Object& o, std::int64_t pool,
	       ReadOp&& op,
	       ceph::buffer::list* bl,
	       CompletionToken&& token,
	       std::optional<std::string_view> ns = {},
	       std::optional<std::string_view> key = {},
	       uint64_t* objver = nullptr) {
    boost::asio::async_completion<CompletionToken, Op::Signature> init(token);
    execute(o, pool, std::move(op), bl,
	    ReadOp::Completion::create(get_executor(),
				       std::move(init.completion_handler)),
	    ns, key, objver);
    return init.result.get();
  }

  template<typename CompletionToken>
  auto execute(const Object& o, std::int64_t pool, WriteOp&& op,
	       CompletionToken&& token,
	       std::optional<std::string_view> ns = {},
	       std::optional<std::string_view> key = {},
	       uint64_t* objver = nullptr) {
    boost::asio::async_completion<CompletionToken, Op::Signature> init(token);
    execute(o, pool, std::move(op),
	    Op::Completion::create(get_executor(),
				   std::move(init.completion_handler)),
	    ns, key, objver);
    return init.result.get();
  }

  boost::uuids::uuid get_fsid() const noexcept;

  using LookupPoolSig = void(boost::system::error_code,
			     std::int64_t);
  using LookupPoolComp = ceph::async::Completion<LookupPoolSig>;
  template<typename CompletionToken>
  auto lookup_pool(std::string_view name,
		   CompletionToken&& token) {
    boost::asio::async_completion<CompletionToken, LookupPoolSig> init(token);
    lookup_pool(name,
		LookupPoolComp::create(get_executor(),
				       std::move(init.completion_handler)));
    return init.result.get();
  }

  std::optional<uint64_t> get_pool_alignment(int64_t pool_id);

  using LSPoolsSig = void(std::vector<std::pair<std::int64_t, std::string>>);
  using LSPoolsComp = ceph::async::Completion<LSPoolsSig>;
  template<typename CompletionToken>
  auto list_pools(CompletionToken&& token) {
    boost::asio::async_completion<CompletionToken, LSPoolsSig> init(token);
    list_pools(LSPoolsComp::create(get_executor(),
				   std::move(init.completion_handler)));
    return init.result.get();
  }



  using SimpleOpSig = void(boost::system::error_code);
  using SimpleOpComp = ceph::async::Completion<SimpleOpSig>;
  template<typename CompletionToken>
  auto create_pool_snap(int64_t pool, std::string_view snapName,
			CompletionToken&& token) {
    boost::asio::async_completion<CompletionToken, SimpleOpSig> init(token);
    create_pool_snap(pool, snapName,
		     SimpleOpComp::create(get_executor(),
					  std::move(init.completion_handler)));
    return init.result.get();
  }

  using SMSnapSig = void(boost::system::error_code, std::uint64_t);
  using SMSnapComp = ceph::async::Completion<SMSnapSig>;
  template<typename CompletionToken>
  auto allocate_selfmanaged_snap(int64_t pool,
				 CompletionToken&& token) {
    boost::asio::async_completion<CompletionToken, SMSnapSig> init(token);
    allocate_selfmanaged_snap(pool,
			      SMSnapComp::create(
				get_executor(),
				std::move(init.completion_handler)));
    return init.result.get();
  }

  template<typename CompletionToken>
  auto delete_pool_snap(int64_t pool, std::string_view snapName,
			CompletionToken&& token) {
    boost::asio::async_completion<CompletionToken, SimpleOpSig> init(token);
    delete_pool_snap(pool, snapName,
		     SimpleOpComp::create(get_executor(),
					  std::move(init.completion_handler)));
    return init.result.get();
  }

  template<typename CompletionToken>
  auto delete_selfmanaged_snap(int64_t pool, std::string_view snapName,
			       CompletionToken&& token) {
    boost::asio::async_completion<CompletionToken, SimpleOpSig> init(token);
    delete_selfmanaged_snap(pool, snapName,
			    SimpleOpComp::create(
			      get_executor(),
			      std::move(init.completion_handler)));
    return init.result.get();
  }

  template<typename CompletionToken>
  auto create_pool(std::string_view name, std::optional<int> crush_rule,
		   CompletionToken&& token) {
    boost::asio::async_completion<CompletionToken, SimpleOpSig> init(token);
    create_pool(name, crush_rule,
		SimpleOpComp::create(get_executor(),
				     std::move(init.completion_handler)));
    return init.result.get();
  }

  template<typename CompletionToken>
  auto delete_pool(std::string_view name,
		   CompletionToken&& token) {
    boost::asio::async_completion<CompletionToken, SimpleOpSig> init(token);
    delete_pool(name,
		SimpleOpComp::create(get_executor(),
				     std::move(init.completion_handler)));
    return init.result.get();
  }

  template<typename CompletionToken>
  auto delete_pool(int64_t pool,
		   CompletionToken&& token) {
    boost::asio::async_completion<CompletionToken, SimpleOpSig> init(token);
    delete_pool(pool,
		SimpleOpComp::create(get_executor(),
				     std::move(init.completion_handler)));
    return init.result.get();
  }

  using PoolStatSig = void(boost::system::error_code,
			   boost::container::flat_map<std::string,
			                              PoolStats>, bool);
  using PoolStatComp = ceph::async::Completion<PoolStatSig>;
  template<typename CompletionToken>
  auto stat_pools(const std::vector<std::string>& pools,
		  CompletionToken&& token) {
    boost::asio::async_completion<CompletionToken, PoolStatSig> init(token);
    stat_pools(pools,
	       PoolStatComp::create(get_executor(),
				    std::move(init.completion_handler)));
    return init.result.get();
  }

  using StatFSSig = void(boost::system::error_code,
			 FSStats);
  using StatFSComp = ceph::async::Completion<StatFSSig>;
  template<typename CompletionToken>
  auto statfs(std::optional<int64_t> pool,
	      CompletionToken&& token) {
    boost::asio::async_completion<CompletionToken, StatFSSig> init(token);
    ceph_statfs(pool, StatFSComp::create(get_executor(),
					 std::move(init.completion_handler)));
    return init.result.get();
  }

  using WatchCB = fu2::unique_function<void(boost::system::error_code,
					    uint64_t notify_id,
					    uint64_t cookie,
					    uint64_t notifier_id,
					    ceph::buffer::list&& bl)>;

  using WatchSig = void(boost::system::error_code ec,
			uint64_t cookie);
  using WatchComp = ceph::async::Completion<WatchSig>;
  template<typename CompletionToken>
  auto watch(const Object& o, const IOContext& ioc,
	     std::optional<std::chrono::seconds> timeout,
	     WatchCB&& cb, CompletionToken&& token) {
    boost::asio::async_completion<CompletionToken, WatchSig> init(token);
    watch(o, ioc, timeout, std::move(cb),
	  WatchComp::create(get_executor(),
			    std::move(init.completion_handler)));
    return init.result.get();
  }

  template<typename CompletionToken>
  auto watch(const Object& o, std::int64_t pool,
	     std::optional<std::chrono::seconds> timeout,
	     WatchCB&& cb, CompletionToken&& token,
	     std::optional<std::string_view> ns = {},
	     std::optional<std::string_view> key = {}) {
    boost::asio::async_completion<CompletionToken, WatchSig> init(token);
    watch(o, pool, timeout, std::move(cb),
	  WatchComp::create(get_executor(),
			    std::move(init.completion_handler)),
	  ns, key);
    return init.result.get();
  }

  template<typename CompletionToken>
  auto notify_ack(const Object& o,
		  const IOContext& ioc,
		  uint64_t notify_id,
		  uint64_t cookie,
		  ceph::buffer::list&& bl,
		  CompletionToken&& token) {
    boost::asio::async_completion<CompletionToken, SimpleOpSig> init(token);
    notify_ack(o, ioc, notify_id, cookie, std::move(bl),
	       SimpleOpComp::create(get_executor(),
				    std::move(init.completion_handler)));
    return init.result.get();
  }

  template<typename CompletionToken>
  auto notify_ack(const Object& o,
		  std::int64_t pool,
		  uint64_t notify_id,
		  uint64_t cookie,
		  ceph::buffer::list&& bl,
		  CompletionToken&& token,
		  std::optional<std::string_view> ns = {},
		  std::optional<std::string_view> key = {}) {
    boost::asio::async_completion<CompletionToken, WatchSig> init(token);
    notify_ack(o, pool, notify_id, cookie, std::move(bl),
	       SimpleOpComp::create(get_executor(),
				    std::move(init.completion_handler)),
	       ns, key);
    return init.result.get();
  }

  template<typename CompletionToken>
  auto unwatch(uint64_t cookie, const IOContext& ioc,
	       CompletionToken&& token) {
    boost::asio::async_completion<CompletionToken, SimpleOpSig> init(token);
    unwatch(cookie, ioc,
	    SimpleOpComp::create(get_executor(),
				 std::move(init.completion_handler)));
    return init.result.get();
  }

  template<typename CompletionToken>
  auto unwatch(uint64_t cookie, std::int64_t pool,
	       CompletionToken&& token,
	       std::optional<std::string_view> ns = {},
	       std::optional<std::string_view> key = {}) {
    boost::asio::async_completion<CompletionToken, SimpleOpSig> init(token);
    unwatch(cookie, pool,
	    SimpleOpComp::create(get_executor(),
				 std::move(init.completion_handler)),
	    ns, key);
    return init.result.get();
  }

  // This is one of those places where having to force everything into
  // a .cc file is really infuriating. If we had modules, that would
  // let us separate out the implementation details without
  // sacrificing all the benefits of templates.
  using VoidOpSig = void();
  using VoidOpComp = ceph::async::Completion<VoidOpSig>;
  template<typename CompletionToken>
  auto flush_watch(CompletionToken&& token) {
    boost::asio::async_completion<CompletionToken, VoidOpSig> init(token);
    flush_watch(VoidOpComp::create(get_executor(),
				   std::move(init.completion_handler)));
    return init.result.get();
  }

  using NotifySig = void(boost::system::error_code, ceph::buffer::list);
  using NotifyComp = ceph::async::Completion<NotifySig>;
  template<typename CompletionToken>
  auto notify(const Object& oid, const IOContext& ioc, ceph::buffer::list&& bl,
	      std::optional<std::chrono::milliseconds> timeout,
	      CompletionToken&& token) {
    boost::asio::async_completion<CompletionToken, NotifySig> init(token);
    notify(oid, ioc, std::move(bl), timeout,
	   NotifyComp::create(get_executor(),
			      std::move(init.completion_handler)));

    return init.result.get();
  }

  template<typename CompletionToken>
  auto notify(const Object& oid, std::int64_t pool, ceph::buffer::list&& bl,
	      std::optional<std::chrono::milliseconds> timeout,
	      CompletionToken&& token,
	      std::optional<std::string_view> ns = {},
	      std::optional<std::string_view> key = {}) {
    boost::asio::async_completion<CompletionToken, NotifySig> init(token);
    notify(oid, pool, bl, timeout,
	   NotifyComp::create(get_executor(),
			      std::move(init.completion_handler)),
	   ns, key);

    return init.result.get();
  }

  // The versions with pointers are fine for coroutines, but
  // extraordinarily unappealing for callback-oriented programming.
  using EnumerateSig = void(boost::system::error_code,
			    std::vector<Entry>,
			    Cursor);
  using EnumerateComp = ceph::async::Completion<EnumerateSig>;
  template<typename CompletionToken>
  auto enumerate_objects(const IOContext& ioc, const Cursor& begin,
			 const Cursor& end, const std::uint32_t max,
			 const ceph::buffer::list& filter,
			 CompletionToken&& token) {
    boost::asio::async_completion<CompletionToken, EnumerateSig> init(token);
    enumerate_objects(ioc, begin, end, max, filter,
		      EnumerateComp::create(get_executor(),
					    std::move(init.completion_handler)));
    return init.result.get();
  }

  template<typename CompletionToken>
  auto enumerate_objects(std::int64_t pool, const Cursor& begin,
			 const Cursor& end, const std::uint32_t max,
			 const ceph::buffer::list& filter,
			 CompletionToken&& token,
			 std::optional<std::string_view> ns = {},
			 std::optional<std::string_view> key = {}) {
    boost::asio::async_completion<CompletionToken, EnumerateSig> init(token);
    enumerate_objects(pool, begin, end, max, filter,
		      EnumerateComp::create(get_executor(),
					    std::move(init.completion_handler)),
		      ns, key);
    return init.result.get();
  }

  using CommandSig = void(boost::system::error_code,
			  std::string, ceph::buffer::list);
  using CommandComp = ceph::async::Completion<CommandSig>;
  template<typename CompletionToken>
  auto osd_command(int osd, std::vector<std::string>&& cmd,
		   ceph::buffer::list&& in, CompletionToken&& token) {
    boost::asio::async_completion<CompletionToken, CommandSig> init(token);
    osd_command(osd, std::move(cmd), std::move(in),
		CommandComp::create(get_executor(),
				      std::move(init.completion_handler)));
    return init.result.get();
  }
  template<typename CompletionToken>
  auto pg_command(PG pg, std::vector<std::string>&& cmd,
		  ceph::buffer::list&& in, CompletionToken&& token) {
    boost::asio::async_completion<CompletionToken, CommandSig> init(token);
    pg_command(pg, std::move(cmd), std::move(in),
	       CommandComp::create(get_executor(),
				      std::move(init.completion_handler)));
    return init.result.get();
  }

  template<typename CompletionToken>
  auto mon_command(std::vector<std::string> command,
		   const ceph::buffer::list& bl,
		   std::string* outs, ceph::buffer::list* outbl,
		   CompletionToken&& token) {
    boost::asio::async_completion<CompletionToken, SimpleOpSig> init(token);
    mon_command(command, bl, outs, outbl,
		SimpleOpComp::create(get_executor(),
				     std::move(init.completion_handler)));
    return init.result.get();
  }

  template<typename CompletionToken>
  auto enable_application(std::string_view pool, std::string_view app_name,
			  bool force, CompletionToken&& token) {
    boost::asio::async_completion<CompletionToken, SimpleOpSig> init(token);
    enable_application(pool, app_name, force,
		       SimpleOpComp::create(get_executor(),
					    std::move(init.completion_handler)));
    return init.result.get();
  }

  template<typename CompletionToken>
  auto blocklist_add(std::string_view client_address,
                     std::optional<std::chrono::seconds> expire,
                     CompletionToken&& token) {
    boost::asio::async_completion<CompletionToken, SimpleOpSig> init(token);
    blocklist_add(client_address, expire,
                  SimpleOpComp::create(get_executor(),
                                       std::move(init.completion_handler)));
    return init.result.get();
  }

  template<typename CompletionToken>
  auto wait_for_latest_osd_map(CompletionToken&& token) {
    boost::asio::async_completion<CompletionToken, SimpleOpSig> init(token);
    wait_for_latest_osd_map(
      SimpleOpComp::create(get_executor(), std::move(init.completion_handler)));
    return init.result.get();
  }

  uint64_t instance_id() const;

private:

  RADOS();

  friend Builder;

  RADOS(std::unique_ptr<detail::Client> impl);
  static void make_with_cct(CephContext* cct,
			    boost::asio::io_context& ioctx,
		    std::unique_ptr<BuildComp> c);

  void execute(const Object& o, const IOContext& ioc, ReadOp&& op,
	       ceph::buffer::list* bl, std::unique_ptr<Op::Completion> c,
	       uint64_t* objver, const blkin_trace_info* trace_info);

  void execute(const Object& o, const IOContext& ioc, WriteOp&& op,
	       std::unique_ptr<Op::Completion> c, uint64_t* objver,
	       const blkin_trace_info* trace_info);

  void execute(const Object& o, std::int64_t pool, ReadOp&& op,
	       ceph::buffer::list* bl, std::unique_ptr<Op::Completion> c,
	       std::optional<std::string_view> ns,
	       std::optional<std::string_view> key,
	       uint64_t* objver);

  void execute(const Object& o, std::int64_t pool, WriteOp&& op,
	       std::unique_ptr<Op::Completion> c,
	       std::optional<std::string_view> ns,
	       std::optional<std::string_view> key,
	       uint64_t* objver);

  void lookup_pool(std::string_view name, std::unique_ptr<LookupPoolComp> c);
  void list_pools(std::unique_ptr<LSPoolsComp> c);
  void create_pool_snap(int64_t pool, std::string_view snapName,
			std::unique_ptr<SimpleOpComp> c);
  void allocate_selfmanaged_snap(int64_t pool, std::unique_ptr<SMSnapComp> c);
  void delete_pool_snap(int64_t pool, std::string_view snapName,
			std::unique_ptr<SimpleOpComp> c);
  void delete_selfmanaged_snap(int64_t pool, std::uint64_t snap,
			       std::unique_ptr<SimpleOpComp> c);
  void create_pool(std::string_view name, std::optional<int> crush_rule,
		   std::unique_ptr<SimpleOpComp> c);
  void delete_pool(std::string_view name,
		   std::unique_ptr<SimpleOpComp> c);
  void delete_pool(int64_t pool,
		   std::unique_ptr<SimpleOpComp> c);
  void stat_pools(const std::vector<std::string>& pools,
		  std::unique_ptr<PoolStatComp> c);
  void stat_fs(std::optional<std::int64_t> pool,
	       std::unique_ptr<StatFSComp> c);

  void watch(const Object& o, const IOContext& ioc,
	     std::optional<std::chrono::seconds> timeout,
	     WatchCB&& cb, std::unique_ptr<WatchComp> c);
  void watch(const Object& o, std::int64_t pool,
	     std::optional<std::chrono::seconds> timeout,
	     WatchCB&& cb, std::unique_ptr<WatchComp> c,
	     std::optional<std::string_view> ns,
	     std::optional<std::string_view> key);
  tl::expected<ceph::timespan, boost::system::error_code>
  watch_check(uint64_t cookie);
  void notify_ack(const Object& o,
		  const IOContext& _ioc,
		  uint64_t notify_id,
		  uint64_t cookie,
		  ceph::buffer::list&& bl,
		  std::unique_ptr<SimpleOpComp>);
  void notify_ack(const Object& o,
		  std::int64_t pool,
		  uint64_t notify_id,
		  uint64_t cookie,
		  ceph::buffer::list&& bl,
		  std::unique_ptr<SimpleOpComp>,
		  std::optional<std::string_view> ns,
		  std::optional<std::string_view> key);
  void unwatch(uint64_t cookie, const IOContext& ioc,
	       std::unique_ptr<SimpleOpComp>);
  void unwatch(uint64_t cookie, std::int64_t pool,
	       std::unique_ptr<SimpleOpComp>,
	       std::optional<std::string_view> ns,
	       std::optional<std::string_view> key);
  void notify(const Object& oid, const IOContext& ioctx,
	      ceph::buffer::list&& bl,
	      std::optional<std::chrono::milliseconds> timeout,
	      std::unique_ptr<NotifyComp> c);
  void notify(const Object& oid, std::int64_t pool,
	      ceph::buffer::list&& bl,
	      std::optional<std::chrono::milliseconds> timeout,
	      std::unique_ptr<NotifyComp> c,
	      std::optional<std::string_view> ns,
	      std::optional<std::string_view> key);
  void flush_watch(std::unique_ptr<VoidOpComp>);

  void enumerate_objects(const IOContext& ioc, const Cursor& begin,
			 const Cursor& end, const std::uint32_t max,
			 const ceph::buffer::list& filter,
			 std::vector<Entry>* ls,
			 Cursor* cursor,
			 std::unique_ptr<SimpleOpComp> c);
  void enumerate_objects(std::int64_t pool, const Cursor& begin,
			 const Cursor& end, const std::uint32_t max,
			 const ceph::buffer::list& filter,
			 std::vector<Entry>* ls,
			 Cursor* cursor,
			 std::unique_ptr<SimpleOpComp> c,
			 std::optional<std::string_view> ns,
			 std::optional<std::string_view> key);
  void enumerate_objects(const IOContext& ioc, const Cursor& begin,
			 const Cursor& end, const std::uint32_t max,
			 const ceph::buffer::list& filter,
			 std::unique_ptr<EnumerateComp> c);
  void enumerate_objects(std::int64_t pool, const Cursor& begin,
			 const Cursor& end, const std::uint32_t max,
			 const ceph::buffer::list& filter,
			 std::unique_ptr<EnumerateComp> c,
			 std::optional<std::string_view> ns,
			 std::optional<std::string_view> key);
  void osd_command(int osd, std::vector<std::string>&& cmd,
		   ceph::buffer::list&& in, std::unique_ptr<CommandComp> c);
  void pg_command(PG pg, std::vector<std::string>&& cmd,
		  ceph::buffer::list&& in, std::unique_ptr<CommandComp> c);

  void mon_command(std::vector<std::string> command,
		   const ceph::buffer::list& bl,
		   std::string* outs, ceph::buffer::list* outbl,
		   std::unique_ptr<SimpleOpComp> c);

  void enable_application(std::string_view pool, std::string_view app_name,
			  bool force, std::unique_ptr<SimpleOpComp> c);

  void blocklist_add(std::string_view client_address,
                     std::optional<std::chrono::seconds> expire,
                     std::unique_ptr<SimpleOpComp> c);

  void wait_for_latest_osd_map(std::unique_ptr<SimpleOpComp> c);

  // Proxy object to provide access to low-level RADOS messaging clients
  std::unique_ptr<detail::Client> impl;
};

enum class errc {
  pool_dne = 1,
  invalid_snapcontext
};

const boost::system::error_category& error_category() noexcept;
}

namespace boost::system {
template<>
struct is_error_code_enum<::neorados::errc> {
  static const bool value = true;
};

template<>
struct is_error_condition_enum<::neorados::errc> {
  static const bool value = false;
};
}

namespace neorados {
//  explicit conversion:
inline boost::system::error_code make_error_code(errc e) noexcept {
  return { static_cast<int>(e), error_category() };
}

// implicit conversion:
inline boost::system::error_condition make_error_condition(errc e) noexcept {
  return { static_cast<int>(e), error_category() };
}
}

namespace std {
template<>
struct hash<neorados::Object> {
  size_t operator ()(const neorados::Object& r) const;
};
template<>
struct hash<neorados::IOContext> {
  size_t operator ()(const neorados::IOContext& r) const;
};
} // namespace std

#endif // NEORADOS_RADOS_HPP