#pragma once

#include <algorithm>

#include <cstdint>

#include <deque>

#include <memory>

#include <string>

#include "envoy/buffer/buffer.h"

#include "envoy/http/stream_reset_handler.h"

#include "source/common/common/assert.h"

#include "source/common/common/non_copyable.h"

#include "source/common/common/utility.h"

#include "source/common/event/libevent.h"

#include "absl/functional/any_invocable.h"

namespace Envoy {

namespace Buffer {

/**

 * A Slice manages a contiguous block of bytes.

 * The block is arranged like this:

 *                   |<- dataSize() ->|<- reservableSize() ->|

 * +-----------------+----------------+----------------------+

 * | Drained         | Data           | Reservable           |

 * | Unused space    | Usable content | New content can be   |

 * | that formerly   |                | added here with      |

 * | was in the Data |                | reserve()/commit()   |

 * | section         |                | or append()          |

 * +-----------------+----------------+----------------------+

 * ^                 ^                ^                      ^

 * |                 |                |                      |

 * base_             base_ + data_    base_ + reservable_    base_ + capacity_

 */

class Slice {

public:

  using Reservation = RawSlice;

  using StoragePtr = std::unique_ptr<uint8_t[]>;

  struct SizedStorage {

    StoragePtr mem_{};

    size_t len_{};

  };

  /**

   * Create an empty Slice with 0 capacity.

   */

  /**

   * Create an empty mutable Slice that owns its storage, which it charges to the provided account,

   * if any.

   * @param min_capacity number of bytes of space the slice should have. Actual capacity is rounded

   * up to the next multiple of 4kb.

   * @param account the account to charge.

   */

  Slice(uint64_t min_capacity, const BufferMemoryAccountSharedPtr& account)

  /**

   * Create an empty mutable Slice that owns its storage, which it charges to the provided account,

   * if any.

   * @param storage backend storage for the slice.

   * @param used_size the size already used in storage.

   * @param account the account to charge.

   */

  Slice(SizedStorage storage, uint64_t used_size, const BufferMemoryAccountSharedPtr& account)

  /**

   * Create an immutable Slice that refers to an external buffer fragment.

   * @param fragment provides externally owned immutable data.

   */

  Slice(BufferFragment& fragment)

  /**

   * @return true if the data in the slice is mutable

   */

  /**

   * @return true if content in this Slice can be coalesced into another Slice.

   */

  /**

   * @return a pointer to the start of the usable content.

   */

  /**

   * @return a pointer to the start of the usable content.

   */

  /**

   * @return the size in bytes of the usable content.

   */

  /**

   * Remove the first `size` bytes of usable content. Runs in O(1) time.

   * @param size number of bytes to remove. If greater than data_size(), the result is undefined.

   */

      // All the data in the slice has been drained. Reset the offsets so all

      // the data can be reused.

  /**

   * @return the number of bytes available to be reserved.

   * @note Read-only implementations of Slice should return zero from this method.

   */

  /**

   * Reserve `size` bytes that the caller can populate with content. The caller SHOULD then

   * call commit() to add the newly populated content from the Reserved section to the Data

   * section.

   * @note If there is already an outstanding reservation (i.e., a reservation obtained

   *       from reserve() that has not been released by calling commit()), this method will

   *       return a new reservation that replaces it.

   * @param size the number of bytes to reserve. The Slice implementation MAY reserve

   *        fewer bytes than requested (for example, if it doesn't have enough room in the

   *        Reservable section to fulfill the whole request).

   * @return a tuple containing the address of the start of resulting reservation and the

   *         reservation size in bytes. If the address is null, the reservation failed.

   * @note Read-only implementations of Slice should return {nullptr, 0} from this method.

   */

    // Verify the semantics that drain() enforces: if the slice is empty, either because

    // no data has been added or because all the added data has been drained, the data

    // section is at the very start of the slice.

  /**

   * Commit a Reservation that was previously obtained from a call to reserve().

   * The Reservation's size is added to the Data section.

   * @param reservation a reservation obtained from a previous call to reserve().

   *        If the reservation is not from this Slice, commit() will return false.

   *        If the caller is committing fewer bytes than provided by reserve(), it

   *        should change the len_ field of the reservation before calling commit().

   *        For example, if a caller reserve()s 4KB to do a nonblocking socket read,

   *        and the read only returns two bytes, the caller should set

   *        reservation.len_ = 2 and then call `commit(reservation)`.

   * @return whether the Reservation was successfully committed to the Slice.

   * @note template parameter `SafeCommit` can be used to disable memory range check.

   */

        // The reservation is not from this Slice.

  /**

   * Copy as much of the supplied data as possible to the end of the slice.

   * @param data start of the data to copy.

   * @param size number of bytes to copy.

   * @return number of bytes copied (may be a smaller than size, may even be zero).

   */

    // NOLINTNEXTLINE(clang-analyzer-core.NullDereference)

  /**

   * Copy as much of the supplied data as possible to the front of the slice.

   * If only part of the data will fit in the slice, the bytes from the _end_ are

   * copied.

   * @param data start of the data to copy.

   * @param size number of bytes to copy.

   * @return number of bytes copied (may be a smaller than size, may even be zero).

   */

  uint64_t prepend(const void* data, uint64_t size);

  /**

   * Describe the in-memory representation of the slice. For use

   * in tests that want to make assertions about the specific arrangement of

   * bytes in a slice.

   */

  struct SliceRepresentation {

    uint64_t data;

    uint64_t reservable;

    uint64_t capacity;

  };

  /**

   * Move all drain trackers and charges from the current slice to the destination slice.

   */

    // The releasor needn't to be transferred, and actually if there is releasor, this

    // slice can't coalesce. Then there won't be a chance to calling this method.

  /**

   * Add a drain tracker to the slice.

   */

  /**

   * Call all drain trackers associated with the slice, then clear

   * the drain tracker list.

   */

  /**

   * Charges the provided account for the resources if these conditions hold:

   * - we're not already charging for this slice

   * - the given account is non-null

   * - the slice owns backing memory

   */

  static constexpr uint32_t default_slice_size_ = 16384;

public:

  /**

   * Compute a slice size big enough to hold a specified amount of data.

   * @param data_size the minimum amount of data the slice must be able to store, in bytes.

   * @return a recommended slice size, in bytes.

   */

  /**

   * Create new backend storage with min capacity. This method will create a recommended capacity

   * which will bigger or equal to the min capacity and create new backend storage based on the

   * recommended capacity.

   * @param min_capacity the min capacity of new created backend storage.

   * @return a backend storage for slice.

   */

protected:

  /** Length of the byte array that base_ points to. This is also the offset in bytes from the start

   * of the slice to the end of the Reservable section. */

  uint64_t capacity_ = 0;

  /** Backing storage for mutable slices which own their own storage. This storage should never be

   * accessed directly; access base_ instead. */

  StoragePtr storage_;

  /** Start of the slice. Points to storage_ iff the slice owns its own storage. */

  uint8_t* base_{nullptr};

  /** Offset in bytes from the start of the slice to the start of the Data section. */

  uint64_t data_ = 0;

  /** Offset in bytes from the start of the slice to the start of the Reservable section which is

   * also the end of the Data section. */

  uint64_t reservable_ = 0;

  /** Hooks to execute when the slice is destroyed. */

  std::list<std::function<void()>> drain_trackers_;

  /** Account associated with this slice. This may be null. When

   * coalescing with another slice, we do not transfer over their account. */

  BufferMemoryAccountSharedPtr account_;

  /** The releasor for the BufferFragment */

  std::function<void()> releasor_;

};

class OwnedImpl;

class SliceDataImpl : public SliceData {

public:

  // SliceData

private:

  friend OwnedImpl;

  Slice slice_;

};

/**

 * Queue of Slice that supports efficient read and write access to both

 * the front and the back of the queue.

 * @note This class has similar properties to std::deque<T>. The reason for using

 *       a custom deque implementation is that benchmark testing during development

 *       revealed that std::deque was too slow to reach performance parity with the

 *       prior evbuffer-based buffer implementation.

 */

class SliceDeque {

public:

    // This custom move constructor is needed so that ring_ will be updated properly.

    // This custom assignment move operator is needed so that ring_ will be updated properly.

  /**

   * Forward const iterator for SliceDeque.

   * @note this implementation currently supports the minimum functionality needed to support

   *       the `for (const auto& slice : slice_deque)` idiom.

   */

  class ConstIterator {

  public:

    friend class SliceDeque;

  private:

    const SliceDeque& deque_;

    size_t index_;

  };

private:

  constexpr static size_t InlineRingCapacity = 8;

  Slice inline_ring_[InlineRingCapacity];

  std::unique_ptr<Slice[]> external_ring_;

  Slice* ring_; // points to start of either inline or external ring.

  size_t start_{0};

  size_t size_{0};

  size_t capacity_;

};

/**

 * An implementation of BufferFragment where a releasor callback is called when the data is

 * no longer needed.

 */

class BufferFragmentImpl : NonCopyable, public BufferFragment {

public:

  /**

   * Creates a new wrapper around the externally owned <data> of size <size>.

   * The caller must ensure <data> is valid until releasor() is called, or for the lifetime of the

   * fragment. releasor() is called with <data>, <size> and <this> to allow caller to delete

   * the fragment object.

   * @param data external data to reference

   * @param size size of data

   * @param releasor a callback function to be called when data is no longer needed.

   */

  BufferFragmentImpl(

      const void* data, size_t size,

      absl::AnyInvocable<void(const void*, size_t, const BufferFragmentImpl*)> releasor)

  // Buffer::BufferFragment

private:

  const void* const data_;

  const size_t size_;

  absl::AnyInvocable<void(const void*, size_t, const BufferFragmentImpl*)> releasor_;

};

class LibEventInstance : public Instance {

public:

  // Called after accessing the memory in buffer() directly to allow any post-processing.

  virtual void postProcess() PURE;

};

/**

 * Wrapper for uint64_t that asserts upon integer overflow and underflow.

 */

class OverflowDetectingUInt64 {

public:

private:

  uint64_t value_{0};

};

/**

 * Wraps an allocated and owned buffer.

 *

 * Note that due to the internals of move(), OwnedImpl is not

 * compatible with non-OwnedImpl buffers.

 */

class OwnedImpl : public LibEventInstance {

public:

  OwnedImpl();

  OwnedImpl(absl::string_view data);

  OwnedImpl(const Instance& data);

  OwnedImpl(const void* data, uint64_t size);

  OwnedImpl(BufferMemoryAccountSharedPtr account);

  // Buffer::Instance

  void addDrainTracker(std::function<void()> drain_tracker) override;

  void bindAccount(BufferMemoryAccountSharedPtr account) override;

  void add(const void* data, uint64_t size) override;

  void addBufferFragment(BufferFragment& fragment) override;

  void add(absl::string_view data) override;

  void add(const Instance& data) override;

  void prepend(absl::string_view data) override;

  void prepend(Instance& data) override;

  void copyOut(size_t start, uint64_t size, void* data) const override;

  uint64_t copyOutToSlices(uint64_t size, Buffer::RawSlice* slices,

                           uint64_t num_slice) const override;

  void drain(uint64_t size) override;

  RawSliceVector getRawSlices(absl::optional<uint64_t> max_slices = absl::nullopt) const override;

  RawSlice frontSlice() const override;

  SliceDataPtr extractMutableFrontSlice() override;

  uint64_t length() const override;

  void* linearize(uint32_t size) override;

  void move(Instance& rhs) override;

  void move(Instance& rhs, uint64_t length) override;

  void move(Instance& rhs, uint64_t length, bool reset_drain_trackers_and_accounting) override;

  Reservation reserveForRead() override;

  ReservationSingleSlice reserveSingleSlice(uint64_t length, bool separate_slice = false) override;

  ssize_t search(const void* data, uint64_t size, size_t start, size_t length) const override;

  bool startsWith(absl::string_view data) const override;

  std::string toString() const override;

  // LibEventInstance

  void postProcess() override;

  /**

   * Create a new slice at the end of the buffer, and copy the supplied content into it.

   * @param data start of the content to copy.

   *

   */

  virtual void appendSliceForTest(const void* data, uint64_t size);

  /**

   * Create a new slice at the end of the buffer, and copy the supplied string into it.

   * @param data the string to append to the buffer.

   */

  virtual void appendSliceForTest(absl::string_view data);

  /**

   * @return the BufferMemoryAccount bound to this buffer, if any.

   */

  BufferMemoryAccountSharedPtr getAccountForTest();

  // Does not implement watermarking.

  // TODO(antoniovicente) Implement watermarks by merging the OwnedImpl and WatermarkBuffer

  // implementations. Also, make high-watermark config a constructor argument.

  /**

   * Describe the in-memory representation of the slices in the buffer. For use

   * in tests that want to make assertions about the specific arrangement of

   * bytes in the buffer.

   */

  std::vector<Slice::SliceRepresentation> describeSlicesForTest() const;

  /**

   * Create a reservation for reading with a non-default length. Used in benchmark tests.

   */

  size_t addFragments(absl::Span<const absl::string_view> fragments) override;

protected:

  static constexpr uint64_t default_read_reservation_size_ =

      Reservation::MAX_SLICES_ * Slice::default_slice_size_;

  /**

   * Create a reservation with a maximum length.

   */

  Reservation reserveWithMaxLength(uint64_t max_length);

  void commit(uint64_t length, absl::Span<RawSlice> slices,

              ReservationSlicesOwnerPtr slices_owner) override;

private:

  /**

   * @param rhs another buffer

   * @return whether the rhs buffer is also an instance of OwnedImpl (or a subclass) that

   *         uses the same internal implementation as this buffer.

   */

  bool isSameBufferImpl(const Instance& rhs) const;

  void addImpl(const void* data, uint64_t size);

  void drainImpl(uint64_t size);

  /**

   * Moves contents of the `other_slice` by either taking its ownership or coalescing it

   * into an existing slice.

   * NOTE: the caller is responsible for draining the buffer that contains the `other_slice`.

   */

  void coalesceOrAddSlice(Slice&& other_slice);

  /** Ring buffer of slices. */

  SliceDeque slices_;

  /** Sum of the dataSize of all slices. */

  OverflowDetectingUInt64 length_;

  BufferMemoryAccountSharedPtr account_;

  struct OwnedImplReservationSlicesOwner : public ReservationSlicesOwner {

    virtual absl::Span<Slice::SizedStorage> ownedStorages() PURE;

  };

  struct OwnedImplReservationSlicesOwnerMultiple : public OwnedImplReservationSlicesOwner {

  public:

    static constexpr uint32_t free_list_max_ = Buffer::Reservation::MAX_SLICES_;

    absl::InlinedVector<Slice::SizedStorage, Buffer::Reservation::MAX_SLICES_> owned_storages_;

  private:

    // Thread local resolving introduces additional overhead. Initialize this reference once when

    // constructing the owner to reduce thread local resolving to improve performance.

    absl::InlinedVector<Slice::StoragePtr, free_list_max_>& free_list_ref_;

    // Simple thread local cache to reduce unnecessary memory allocation and release. This cache

    // is currently only used for multiple slices reservation because of the additional overhead

    // that thread local resolving would introduce.

    static thread_local absl::InlinedVector<Slice::StoragePtr, free_list_max_> free_list_;

  };

  struct OwnedImplReservationSlicesOwnerSingle : public OwnedImplReservationSlicesOwner {

    Slice::SizedStorage owned_storage_;

  };

};

using BufferFragmentPtr = std::unique_ptr<BufferFragment>;

/**

 * An implementation of BufferFragment where a releasor callback is called when the data is

 * no longer needed. Copies data into internal buffer.

 */

class OwnedBufferFragmentImpl final : public BufferFragment, public InlineStorage {

public:

  using Releasor = std::function<void(const OwnedBufferFragmentImpl*)>;

  /**

   * Copies the data into internal buffer. The releasor is called when the data has been

   * fully drained or the buffer that contains this fragment is destroyed.

   * @param data external data to reference

   * @param releasor a callback function to be called when data is no longer needed.

   */

  // Buffer::BufferFragment

private:

  OwnedBufferFragmentImpl(absl::string_view data, const Releasor& releasor)

  const Releasor releasor_;

  const size_t size_;

  uint8_t data_[];

};

using OwnedBufferFragmentImplPtr = std::unique_ptr<OwnedBufferFragmentImpl>;

} // namespace Buffer

} // namespace Envoy