#pragma once

#include <cstdint>
#include <stdexcept>
#include <string_view>

USERVER_NAMESPACE_BEGIN

namespace kafka {

/// @brief Base exception thrown by Producer::Send and Producer::SendAsync
/// on send or delivery errors.
class SendException : public std::runtime_error {
public:
    using std::runtime_error::runtime_error;

    /// @brief Returns whether it makes sense to retry failed send.
    ///
    /// @see
    /// https://docs.confluent.io/platform/current/clients/librdkafka/html/md_INTRODUCTION.html#autotoc_md8
    bool IsRetryable() const noexcept;

protected:
    SendException(const char* what, bool is_retryable);

private:
    const bool is_retryable_{false};
};

class DeliveryTimeoutException final : public SendException {
    static constexpr const char* kWhat{
        "Message is not delivered after `delivery_timeout` milliseconds. Hint: "
        "Adjust `delivery_timeout` and `queue_buffering_*` options or manually "
        "retry the send request."};

public:
    DeliveryTimeoutException();
};

class QueueFullException final : public SendException {
    static constexpr const char* kWhat{
        "The sending queue is full - send request cannot be scheduled. Hint: "
        "Manually retry the error or increase `queue_buffering_max_messages` "
        "and/or `queue_buffering_max_kbytes` config option."};

public:
    QueueFullException();
};

class MessageTooLargeException final : public SendException {
    static constexpr const char* kWhat{
        "Message size exceeds configured limit. Hint: increase "
        "`message_max_bytes` config option."};

public:
    MessageTooLargeException();
};

class UnknownTopicException final : public SendException {
    static constexpr const char* kWhat{"Given topic does not exist in cluster."};

public:
    UnknownTopicException();
};

class UnknownPartitionException final : public SendException {
    static constexpr const char* kWhat = "Topic does not have given partition.";

public:
    UnknownPartitionException();
};

/// @brief Exception thrown when there is an error retrieving the offset range.
class OffsetRangeException : public std::runtime_error {
public:
    using std::runtime_error::runtime_error;

public:
    OffsetRangeException(std::string_view what, std::string_view topic, std::uint32_t partition);
};

class OffsetRangeTimeoutException final : public OffsetRangeException {
    static constexpr const char* kWhat = "Timeout while fetching offsets.";

public:
    OffsetRangeTimeoutException(std::string_view topic, std::uint32_t partition);
};

class TopicNotFoundException final : public std::runtime_error {
public:
    using std::runtime_error::runtime_error;
};

/// @brief Exception thrown when fetching metadata.
class GetMetadataException : public std::runtime_error {
public:
    using std::runtime_error::runtime_error;

public:
    GetMetadataException(std::string_view what, std::string_view topic);
};

class GetMetadataTimeoutException final : public GetMetadataException {
    static constexpr const char* kWhat = "Timeout while getting metadata.";

public:
    GetMetadataTimeoutException(std::string_view topic);
};

}  // namespace kafka

USERVER_NAMESPACE_END
