/*
 * librdkafka - Apache Kafka C library
 *
 * Copyright (c) 2014-2022, Magnus Edenhill
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 *
 * 1. Redistributions of source code must retain the above copyright notice,
 *    this list of conditions and the following disclaimer.
 * 2. Redistributions in binary form must reproduce the above copyright notice,
 *    this list of conditions and the following disclaimer in the documentation
 *    and/or other materials provided with the distribution.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
 * POSSIBILITY OF SUCH DAMAGE.
 */

#ifndef _RDKAFKA_CONF_H_
#define _RDKAFKA_CONF_H_

#include "rdlist.h"
#include "rdkafka_cert.h"

#if WITH_SSL && OPENSSL_VERSION_NUMBER >= 0x10100000 &&                        \
    !defined(OPENSSL_IS_BORINGSSL)
#define WITH_SSL_ENGINE 1
/* Deprecated in OpenSSL 3 */
#include <openssl/engine.h>
#endif /* WITH_SSL && OPENSSL_VERSION_NUMBER >= 0x10100000 */

/**
 * Forward declarations
 */
struct rd_kafka_transport_s;


/**
 * MessageSet compression codecs
 */
typedef enum {
        RD_KAFKA_COMPRESSION_NONE,
        RD_KAFKA_COMPRESSION_GZIP   = RD_KAFKA_MSG_ATTR_GZIP,
        RD_KAFKA_COMPRESSION_SNAPPY = RD_KAFKA_MSG_ATTR_SNAPPY,
        RD_KAFKA_COMPRESSION_LZ4    = RD_KAFKA_MSG_ATTR_LZ4,
        RD_KAFKA_COMPRESSION_ZSTD   = RD_KAFKA_MSG_ATTR_ZSTD,
        RD_KAFKA_COMPRESSION_INHERIT, /* Inherit setting from global conf */
        RD_KAFKA_COMPRESSION_NUM
} rd_kafka_compression_t;

static RD_INLINE RD_UNUSED const char *
rd_kafka_compression2str(rd_kafka_compression_t compr) {
        static const char *names[RD_KAFKA_COMPRESSION_NUM] = {
            [RD_KAFKA_COMPRESSION_NONE]    = "none",
            [RD_KAFKA_COMPRESSION_GZIP]    = "gzip",
            [RD_KAFKA_COMPRESSION_SNAPPY]  = "snappy",
            [RD_KAFKA_COMPRESSION_LZ4]     = "lz4",
            [RD_KAFKA_COMPRESSION_ZSTD]    = "zstd",
            [RD_KAFKA_COMPRESSION_INHERIT] = "inherit"};
        static RD_TLS char ret[32];

        if ((int)compr < 0 || compr >= RD_KAFKA_COMPRESSION_NUM) {
                rd_snprintf(ret, sizeof(ret), "codec0x%x?", (int)compr);
                return ret;
        }

        return names[compr];
}

/**
 * MessageSet compression levels
 */
typedef enum {
        RD_KAFKA_COMPLEVEL_DEFAULT    = -1,
        RD_KAFKA_COMPLEVEL_MIN        = -1,
        RD_KAFKA_COMPLEVEL_GZIP_MAX   = 9,
        RD_KAFKA_COMPLEVEL_LZ4_MAX    = 12,
        RD_KAFKA_COMPLEVEL_SNAPPY_MAX = 0,
        RD_KAFKA_COMPLEVEL_ZSTD_MAX   = 22,
        RD_KAFKA_COMPLEVEL_MAX        = 12
} rd_kafka_complevel_t;

typedef enum {
        RD_KAFKA_PROTO_PLAINTEXT,
        RD_KAFKA_PROTO_SSL,
        RD_KAFKA_PROTO_SASL_PLAINTEXT,
        RD_KAFKA_PROTO_SASL_SSL,
        RD_KAFKA_PROTO_NUM,
} rd_kafka_secproto_t;


typedef enum {
        RD_KAFKA_CONFIGURED,
        RD_KAFKA_LEARNED,
        RD_KAFKA_INTERNAL,
        RD_KAFKA_LOGICAL
} rd_kafka_confsource_t;

static RD_INLINE RD_UNUSED const char *
rd_kafka_confsource2str(rd_kafka_confsource_t source) {
        static const char *names[] = {"configured", "learned", "internal",
                                      "logical"};

        return names[source];
}


typedef enum {
        _RK_GLOBAL       = 0x1,
        _RK_PRODUCER     = 0x2,
        _RK_CONSUMER     = 0x4,
        _RK_TOPIC        = 0x8,
        _RK_CGRP         = 0x10,
        _RK_DEPRECATED   = 0x20,
        _RK_HIDDEN       = 0x40,
        _RK_HIGH         = 0x80,  /* High Importance */
        _RK_MED          = 0x100, /* Medium Importance */
        _RK_EXPERIMENTAL = 0x200, /* Experimental (unsupported) property */
        _RK_SENSITIVE    = 0x400  /* The configuration property's value
                                   * might contain sensitive information. */
} rd_kafka_conf_scope_t;

/**< While the client groups is a generic concept, it is currently
 *   only implemented for consumers in librdkafka. */
#define _RK_CGRP _RK_CONSUMER

typedef enum {
        _RK_CONF_PROP_SET_REPLACE, /* Replace current value (default) */
        _RK_CONF_PROP_SET_ADD,     /* Add value (S2F) */
        _RK_CONF_PROP_SET_DEL      /* Remove value (S2F) */
} rd_kafka_conf_set_mode_t;



typedef enum {
        RD_KAFKA_OFFSET_METHOD_NONE,
        RD_KAFKA_OFFSET_METHOD_FILE,
        RD_KAFKA_OFFSET_METHOD_BROKER
} rd_kafka_offset_method_t;

typedef enum {
        RD_KAFKA_SASL_OAUTHBEARER_METHOD_DEFAULT,
        RD_KAFKA_SASL_OAUTHBEARER_METHOD_OIDC
} rd_kafka_oauthbearer_method_t;

typedef enum {
        RD_KAFKA_SSL_ENDPOINT_ID_NONE,
        RD_KAFKA_SSL_ENDPOINT_ID_HTTPS, /**< RFC2818 */
} rd_kafka_ssl_endpoint_id_t;

typedef enum {
        RD_KAFKA_USE_ALL_DNS_IPS,
        RD_KAFKA_RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY,
} rd_kafka_client_dns_lookup_t;

typedef enum {
        RD_KAFKA_GROUP_PROTOCOL_GENERIC,
        RD_KAFKA_GROUP_PROTOCOL_CONSUMER,
} rd_kafka_group_protocol_t;

/* Increase in steps of 64 as needed.
 * This must be larger than sizeof(rd_kafka_[topic_]conf_t) */
#define RD_KAFKA_CONF_PROPS_IDX_MAX (64 * 33)

/**
 * @struct rd_kafka_anyconf_t
 * @brief The anyconf header must be the first field in the
 *        rd_kafka_conf_t and rd_kafka_topic_conf_t structs.
 *        It provides a way to track which property has been modified.
 */
struct rd_kafka_anyconf_hdr {
        uint64_t modified[RD_KAFKA_CONF_PROPS_IDX_MAX / 64];
};


/**
 * Optional configuration struct passed to rd_kafka_new*().
 *
 * The struct is populated ted through string properties
 * by calling rd_kafka_conf_set().
 *
 */
struct rd_kafka_conf_s {
        struct rd_kafka_anyconf_hdr hdr; /**< Must be first field */

        /*
         * Generic configuration
         */
        int enabled_events;
        int max_msg_size;
        int msg_copy_max_size;
        int recv_max_msg_size;
        int max_inflight;
        int metadata_request_timeout_ms;
        int metadata_refresh_interval_ms;
        int metadata_refresh_fast_cnt;
        int metadata_refresh_fast_interval_ms;
        int metadata_refresh_sparse;
        int metadata_max_age_ms;
        int metadata_propagation_max_ms;
        int debug;
        int broker_addr_ttl;
        int broker_addr_family;
        int socket_timeout_ms;
        int socket_blocking_max_ms;
        int socket_sndbuf_size;
        int socket_rcvbuf_size;
        int socket_keepalive;
        int socket_nagle_disable;
        int socket_max_fails;
        char *client_id_str;
        char *brokerlist;
        int stats_interval_ms;
        int term_sig;
        int reconnect_backoff_ms;
        int reconnect_backoff_max_ms;
        int reconnect_jitter_ms;
        int socket_connection_setup_timeout_ms;
        int connections_max_idle_ms;
        int sparse_connections;
        int sparse_connect_intvl;
        int api_version_request;
        int api_version_request_timeout_ms;
        int api_version_fallback_ms;
        char *broker_version_fallback;
        rd_kafka_secproto_t security_protocol;
        rd_kafka_client_dns_lookup_t client_dns_lookup;

        struct {
#if WITH_SSL
                SSL_CTX *ctx;
#endif
                char *cipher_suites;
                char *curves_list;
                char *sigalgs_list;
                char *key_location;
                char *key_pem;
                rd_kafka_cert_t *key;
                char *key_password;
                char *cert_location;
                char *cert_pem;
                rd_kafka_cert_t *cert;
                char *ca_location;
                char *ca_pem;
                rd_kafka_cert_t *ca;
                /** CSV list of Windows certificate stores */
                char *ca_cert_stores;
                char *crl_location;
#if WITH_SSL && OPENSSL_VERSION_NUMBER >= 0x10100000
                ENGINE *engine;
#endif
                char *engine_location;
                char *engine_id;
                void *engine_callback_data;
                char *providers;
                rd_list_t loaded_providers; /**< (SSL_PROVIDER*) */
                char *keystore_location;
                char *keystore_password;
                int endpoint_identification;
                int enable_verify;
                int (*cert_verify_cb)(rd_kafka_t *rk,
                                      const char *broker_name,
                                      int32_t broker_id,
                                      int *x509_error,
                                      int depth,
                                      const char *buf,
                                      size_t size,
                                      char *errstr,
                                      size_t errstr_size,
                                      void *opaque);
        } ssl;

        struct {
                const struct rd_kafka_sasl_provider *provider;
                char *principal;
                char *mechanisms;
                char *service_name;
                char *kinit_cmd;
                char *keytab;
                int relogin_min_time;
                /** Protects .username and .password access after client
                 *  instance has been created (see sasl_set_credentials()). */
                mtx_t lock;
                char *username;
                char *password;
#if WITH_SASL_SCRAM
                /* SCRAM EVP-wrapped hash function
                 * (return value from EVP_shaX()) */
                const void /*EVP_MD*/ *scram_evp;
                /* SCRAM direct hash function (e.g., SHA256()) */
                unsigned char *(*scram_H)(const unsigned char *d,
                                          size_t n,
                                          unsigned char *md);
                /* Hash size */
                size_t scram_H_size;
#endif
                char *oauthbearer_config;
                int enable_oauthbearer_unsecure_jwt;
                int enable_callback_queue;
                struct {
                        rd_kafka_oauthbearer_method_t method;
                        char *token_endpoint_url;
                        char *client_id;
                        char *client_secret;
                        char *scope;
                        char *extensions_str;
                        /* SASL/OAUTHBEARER token refresh event callback */
                        void (*token_refresh_cb)(rd_kafka_t *rk,
                                                 const char *oauthbearer_config,
                                                 void *opaque);
                } oauthbearer;
        } sasl;

        char *plugin_paths;
#if WITH_PLUGINS
        rd_list_t plugins;
#endif

        /* Interceptors */
        struct {
                /* rd_kafka_interceptor_method_t lists */
                rd_list_t on_conf_set;            /* on_conf_set interceptors
                                                   * (not copied on conf_dup()) */
                rd_list_t on_conf_dup;            /* .. (not copied) */
                rd_list_t on_conf_destroy;        /* .. (not copied) */
                rd_list_t on_new;                 /* .. (copied) */
                rd_list_t on_destroy;             /* .. (copied) */
                rd_list_t on_send;                /* .. (copied) */
                rd_list_t on_acknowledgement;     /* .. (copied) */
                rd_list_t on_consume;             /* .. (copied) */
                rd_list_t on_commit;              /* .. (copied) */
                rd_list_t on_request_sent;        /* .. (copied) */
                rd_list_t on_response_received;   /* .. (copied) */
                rd_list_t on_thread_start;        /* .. (copied) */
                rd_list_t on_thread_exit;         /* .. (copied) */
                rd_list_t on_broker_state_change; /* .. (copied) */

                /* rd_strtup_t list */
                rd_list_t config; /* Configuration name=val's
                                   * handled by interceptors. */
        } interceptors;

        /* Client group configuration */
        int coord_query_intvl_ms;
        int max_poll_interval_ms;

        int builtin_features;
        /*
         * Consumer configuration
         */
        int check_crcs;
        int queued_min_msgs;
        int queued_max_msg_kbytes;
        int64_t queued_max_msg_bytes;
        int fetch_wait_max_ms;
        int fetch_msg_max_bytes;
        int fetch_max_bytes;
        int fetch_min_bytes;
        int fetch_queue_backoff_ms;
        int fetch_error_backoff_ms;
        rd_kafka_group_protocol_t group_protocol;
        char *group_id_str;
        char *group_instance_id;
        char *group_remote_assignor;
        int allow_auto_create_topics;

        rd_kafka_pattern_list_t *topic_blacklist;
        struct rd_kafka_topic_conf_s *topic_conf; /* Default topic config
                                                   * for automatically
                                                   * subscribed topics. */
        int enable_auto_commit;
        int enable_auto_offset_store;
        int auto_commit_interval_ms;
        int group_session_timeout_ms;
        int group_heartbeat_intvl_ms;
        rd_kafkap_str_t *group_protocol_type;
        char *partition_assignment_strategy;
        rd_list_t partition_assignors;
        int enabled_assignor_cnt;

        void (*rebalance_cb)(rd_kafka_t *rk,
                             rd_kafka_resp_err_t err,
                             rd_kafka_topic_partition_list_t *partitions,
                             void *opaque);

        void (*offset_commit_cb)(rd_kafka_t *rk,
                                 rd_kafka_resp_err_t err,
                                 rd_kafka_topic_partition_list_t *offsets,
                                 void *opaque);

        rd_kafka_offset_method_t offset_store_method;

        rd_kafka_isolation_level_t isolation_level;

        int enable_partition_eof;

        rd_kafkap_str_t *client_rack;

        /*
         * Producer configuration
         */
        struct {
                /*
                 * Idempotence
                 */
                int idempotence;   /**< Enable Idempotent Producer */
                rd_bool_t gapless; /**< Raise fatal error if
                                    *   gapless guarantee can't be
                                    *   satisfied. */
                /*
                 * Transactions
                 */
                char *transactional_id;     /**< Transactional Id */
                int transaction_timeout_ms; /**< Transaction timeout */
        } eos;
        int queue_buffering_max_msgs;
        int queue_buffering_max_kbytes;
        double buffering_max_ms_dbl; /**< This is the configured value */
        rd_ts_t buffering_max_us;    /**< This is the value used in the code */
        int queue_backpressure_thres;
        int max_retries;
        int retry_backoff_ms;
        int retry_backoff_max_ms;
        int batch_num_messages;
        int batch_size;
        rd_kafka_compression_t compression_codec;
        int dr_err_only;
        int sticky_partition_linger_ms;

        /* Message delivery report callback.
         * Called once for each produced message, either on
         * successful and acknowledged delivery to the broker in which
         * case 'err' is 0, or if the message could not be delivered
         * in which case 'err' is non-zero (use rd_kafka_err2str()
         * to obtain a human-readable error reason).
         *
         * If the message was produced with neither RD_KAFKA_MSG_F_FREE
         * or RD_KAFKA_MSG_F_COPY set then 'payload' is the original
         * pointer provided to rd_kafka_produce().
         * rdkafka will not perform any further actions on 'payload'
         * at this point and the application may rd_free the payload data
         * at this point.
         *
         * 'opaque' is 'conf.opaque', while 'msg_opaque' is
         * the opaque pointer provided in the rd_kafka_produce() call.
         */
        void (*dr_cb)(rd_kafka_t *rk,
                      void *payload,
                      size_t len,
                      rd_kafka_resp_err_t err,
                      void *opaque,
                      void *msg_opaque);

        void (*dr_msg_cb)(rd_kafka_t *rk,
                          const rd_kafka_message_t *rkmessage,
                          void *opaque);

        /* Consume callback */
        void (*consume_cb)(rd_kafka_message_t *rkmessage, void *opaque);

        /* Log callback */
        void (*log_cb)(const rd_kafka_t *rk,
                       int level,
                       const char *fac,
                       const char *buf);
        int log_level;
        int log_queue;
        int log_thread_name;
        int log_connection_close;

        /* PRNG seeding */
        int enable_random_seed;

        /* Error callback */
        void (*error_cb)(rd_kafka_t *rk,
                         int err,
                         const char *reason,
                         void *opaque);

        /* Throttle callback */
        void (*throttle_cb)(rd_kafka_t *rk,
                            const char *broker_name,
                            int32_t broker_id,
                            int throttle_time_ms,
                            void *opaque);

        /* Stats callback */
        int (*stats_cb)(rd_kafka_t *rk,
                        char *json,
                        size_t json_len,
                        void *opaque);

        /* Socket creation callback */
        int (*socket_cb)(int domain, int type, int protocol, void *opaque);

        /* Connect callback */
        int (*connect_cb)(int sockfd,
                          const struct sockaddr *addr,
                          int addrlen,
                          const char *id,
                          void *opaque);

        /* Close socket callback */
        int (*closesocket_cb)(int sockfd, void *opaque);

        /* File open callback */
        int (*open_cb)(const char *pathname,
                       int flags,
                       mode_t mode,
                       void *opaque);

        /* Address resolution callback */
        int (*resolve_cb)(const char *node,
                          const char *service,
                          const struct addrinfo *hints,
                          struct addrinfo **res,
                          void *opaque);

        /* Background queue event callback */
        void (*background_event_cb)(rd_kafka_t *rk,
                                    rd_kafka_event_t *rkev,
                                    void *opaque);


        /* Opaque passed to callbacks. */
        void *opaque;

        /* For use with value-less properties. */
        int dummy;


        /* Admin client defaults */
        struct {
                int request_timeout_ms; /* AdminOptions.request_timeout */
        } admin;


        /*
         * Test mocks
         */
        struct {
                int broker_cnt; /**< Number of mock brokers */
                int broker_rtt; /**< Broker RTT */
        } mock;

        /*
         * Unit test pluggable interfaces
         */
        struct {
                /**< Inject errors in ProduceResponse handler */
                rd_kafka_resp_err_t (*handle_ProduceResponse)(
                    rd_kafka_t *rk,
                    int32_t brokerid,
                    uint64_t msgid,
                    rd_kafka_resp_err_t err);
        } ut;

        char *sw_name;    /**< Software/client name */
        char *sw_version; /**< Software/client version */

        struct {
                /** Properties on (implicit pass-thru) default_topic_conf were
                 *  overwritten by passing an explicit default_topic_conf. */
                rd_bool_t default_topic_conf_overwritten;
        } warn;
};

int rd_kafka_socket_cb_linux(int domain, int type, int protocol, void *opaque);
int rd_kafka_socket_cb_generic(int domain,
                               int type,
                               int protocol,
                               void *opaque);
#ifndef _WIN32
int rd_kafka_open_cb_linux(const char *pathname,
                           int flags,
                           mode_t mode,
                           void *opaque);
#endif
int rd_kafka_open_cb_generic(const char *pathname,
                             int flags,
                             mode_t mode,
                             void *opaque);



struct rd_kafka_topic_conf_s {
        struct rd_kafka_anyconf_hdr hdr; /**< Must be first field */

        int required_acks;
        int32_t request_timeout_ms;
        int message_timeout_ms;

        int32_t (*partitioner)(const rd_kafka_topic_t *rkt,
                               const void *keydata,
                               size_t keylen,
                               int32_t partition_cnt,
                               void *rkt_opaque,
                               void *msg_opaque);
        char *partitioner_str;

        rd_bool_t random_partitioner; /**< rd_true - random
                                       *  rd_false - sticky */

        int queuing_strategy; /* RD_KAFKA_QUEUE_FIFO|LIFO */
        int (*msg_order_cmp)(const void *a, const void *b);

        rd_kafka_compression_t compression_codec;
        rd_kafka_complevel_t compression_level;
        int produce_offset_report;

        int consume_callback_max_msgs;
        int auto_commit;
        int auto_commit_interval_ms;
        int auto_offset_reset;
        char *offset_store_path;
        int offset_store_sync_interval_ms;

        rd_kafka_offset_method_t offset_store_method;

        /* Application provided opaque pointer (this is rkt_opaque) */
        void *opaque;
};


char **rd_kafka_conf_kv_split(const char **input, size_t incnt, size_t *cntp);

void rd_kafka_anyconf_destroy(int scope, void *conf);

rd_bool_t rd_kafka_conf_is_modified(const rd_kafka_conf_t *conf,
                                    const char *name);

void rd_kafka_desensitize_str(char *str);

void rd_kafka_conf_desensitize(rd_kafka_conf_t *conf);
void rd_kafka_topic_conf_desensitize(rd_kafka_topic_conf_t *tconf);

const char *rd_kafka_conf_finalize(rd_kafka_type_t cltype,
                                   rd_kafka_conf_t *conf);
const char *rd_kafka_topic_conf_finalize(rd_kafka_type_t cltype,
                                         const rd_kafka_conf_t *conf,
                                         rd_kafka_topic_conf_t *tconf);


int rd_kafka_conf_warn(rd_kafka_t *rk);

void rd_kafka_anyconf_dump_dbg(rd_kafka_t *rk,
                               int scope,
                               const void *conf,
                               const char *description);

#include "rdkafka_confval.h"

int unittest_conf(void);

#endif /* _RDKAFKA_CONF_H_ */