#pragma once

#include <array>

#include <atomic>

#include <chrono>

#include <cstdint>

#include <functional>

#include <list>

#include <memory>

#include <string>

#include <tuple>

#include <utility>

#include <vector>

#include "envoy/api/api.h"

#include "envoy/common/random_generator.h"

#include "envoy/config/cluster/v3/cluster.pb.h"

#include "envoy/config/endpoint/v3/endpoint.pb.h"

#include "envoy/config/endpoint/v3/endpoint_components.pb.h"

#include "envoy/config/typed_metadata.h"

#include "envoy/event/dispatcher.h"

#include "envoy/event/timer.h"

#include "envoy/extensions/clusters/redis/v3/redis_cluster.pb.h"

#include "envoy/extensions/clusters/redis/v3/redis_cluster.pb.validate.h"

#include "envoy/extensions/filters/network/redis_proxy/v3/redis_proxy.pb.h"

#include "envoy/extensions/filters/network/redis_proxy/v3/redis_proxy.pb.validate.h"

#include "envoy/http/codec.h"

#include "envoy/local_info/local_info.h"

#include "envoy/network/dns.h"

#include "envoy/runtime/runtime.h"

#include "envoy/secret/secret_manager.h"

#include "envoy/server/transport_socket_config.h"

#include "envoy/singleton/manager.h"

#include "envoy/ssl/context_manager.h"

#include "envoy/stats/scope.h"

#include "envoy/thread_local/thread_local.h"

#include "envoy/upstream/cluster_manager.h"

#include "envoy/upstream/health_checker.h"

#include "envoy/upstream/load_balancer.h"

#include "envoy/upstream/locality.h"

#include "envoy/upstream/upstream.h"

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

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

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

#include "source/common/config/datasource.h"

#include "source/common/config/metadata.h"

#include "source/common/config/well_known_names.h"

#include "source/common/network/address_impl.h"

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

#include "source/common/stats/isolated_store_impl.h"

#include "source/common/upstream/cluster_factory_impl.h"

#include "source/common/upstream/load_balancer_context_base.h"

#include "source/common/upstream/outlier_detection_impl.h"

#include "source/common/upstream/resource_manager_impl.h"

#include "source/common/upstream/upstream_impl.h"

#include "source/extensions/clusters/redis/redis_cluster_lb.h"

#include "source/extensions/common/redis/cluster_refresh_manager_impl.h"

#include "source/extensions/filters/network/common/redis/client.h"

#include "source/extensions/filters/network/common/redis/client_impl.h"

#include "source/extensions/filters/network/common/redis/codec.h"

#include "source/extensions/filters/network/common/redis/utility.h"

#include "source/extensions/filters/network/redis_proxy/config.h"

#include "source/server/transport_socket_config_impl.h"

namespace Envoy {

namespace Extensions {

namespace Clusters {

namespace Redis {

/*

 * This class implements support for the topology part of `Redis Cluster

 * <https://redis.io/topics/cluster-spec>`_. Specifically, it allows Envoy to maintain an internal

 * representation of the topology of a Redis Cluster, and how often the topology should be

 * refreshed.

 *

 * The target Redis Cluster is obtained from the yaml config file as usual, and we choose a random

 * discovery address from DNS if there are no existing hosts (our startup condition). Otherwise, we

 * choose a random host from our known set of hosts. Then, against this host we make a topology

 * request.

 *

 * Topology requests are handled by RedisDiscoverySession, which handles the initialization of

 * the `CLUSTER SLOTS command <https://redis.io/commands/cluster-slots>`_, and the responses and

 * failure cases.

 *

 * Once the topology is fetched from Redis, the cluster will update the

 * RedisClusterLoadBalancerFactory, which will be used by the redis proxy filter for load balancing

 * purpose.

 */

class RedisCluster : public Upstream::BaseDynamicClusterImpl {

public:

  ~RedisCluster();

  static absl::StatusOr<std::unique_ptr<RedisCluster>>

  create(const envoy::config::cluster::v3::Cluster& cluster,

         const envoy::extensions::clusters::redis::v3::RedisClusterConfig& redis_cluster,

         Upstream::ClusterFactoryContext& context,

         NetworkFilters::Common::Redis::Client::ClientFactory& client_factory,

         Network::DnsResolverSharedPtr dns_resolver, ClusterSlotUpdateCallBackSharedPtr factory);

  struct ClusterSlotsRequest : public Extensions::NetworkFilters::Common::Redis::RespValue {

  public:

    static ClusterSlotsRequest instance_;

  };

  /// TimeSource& timeSource() const { return time_source_; }

protected:

  RedisCluster(const envoy::config::cluster::v3::Cluster& cluster,

               const envoy::extensions::clusters::redis::v3::RedisClusterConfig& redis_cluster,

               Upstream::ClusterFactoryContext& context,

               NetworkFilters::Common::Redis::Client::ClientFactory& client_factory,

               Network::DnsResolverSharedPtr dns_resolver,

               ClusterSlotUpdateCallBackSharedPtr factory, absl::Status& creation_status);

private:

  friend class RedisClusterFactory;

  friend class RedisClusterTest;

  void startPreInit() override;

  void updateAllHosts(const Upstream::HostVector& hosts_added,

                      const Upstream::HostVector& hosts_removed, uint32_t priority);

  void onClusterSlotUpdate(ClusterSlotsSharedPtr&&);

  void reloadHealthyHostsHelper(const Upstream::HostSharedPtr& host) override;

    // Always use the first endpoint.

    // Always use the first endpoint.

  // A redis node in the Redis cluster.

  class RedisHost : public Upstream::HostImpl {

  public:

    static absl::StatusOr<std::unique_ptr<RedisHost>>

    create(Upstream::ClusterInfoConstSharedPtr cluster, const std::string& hostname,

           Network::Address::InstanceConstSharedPtr address, RedisCluster& parent, bool primary);

  protected:

    RedisHost(Upstream::ClusterInfoConstSharedPtr cluster, const std::string& hostname,

              Network::Address::InstanceConstSharedPtr address, RedisCluster& parent, bool primary,

              absl::Status& creation_status)

              // TODO(zyfjeff): Created through metadata shared pool

              // TODO(adisuissa): Convert to use a shared pool of localities.

  private:

    const bool primary_;

  };

  // Resolves the discovery endpoint.

  struct DnsDiscoveryResolveTarget {

    DnsDiscoveryResolveTarget(RedisCluster& parent, const std::string& dns_address,

                              const uint32_t port);

    ~DnsDiscoveryResolveTarget();

    void startResolveDns();

    RedisCluster& parent_;

    Network::ActiveDnsQuery* active_query_{};

    const std::string dns_address_;

    const uint32_t port_;

    Event::TimerPtr resolve_timer_;

  };

  using DnsDiscoveryResolveTargetPtr = std::unique_ptr<DnsDiscoveryResolveTarget>;

  struct RedisDiscoverySession;

  struct RedisDiscoveryClient : public Network::ConnectionCallbacks {

    // Network::ConnectionCallbacks

    void onEvent(Network::ConnectionEvent event) override;

    RedisDiscoverySession& parent_;

    std::string host_;

    Extensions::NetworkFilters::Common::Redis::Client::ClientPtr client_;

  };

  using RedisDiscoveryClientPtr = std::unique_ptr<RedisDiscoveryClient>;

  struct RedisDiscoverySession

      : public Extensions::NetworkFilters::Common::Redis::Client::Config,

        public Extensions::NetworkFilters::Common::Redis::Client::ClientCallbacks,

        public std::enable_shared_from_this<RedisDiscoverySession> {

    RedisDiscoverySession(RedisCluster& parent,

                          NetworkFilters::Common::Redis::Client::ClientFactory& client_factory);

    ~RedisDiscoverySession() override;

    void registerDiscoveryAddress(std::list<Network::DnsResponse>&& response, const uint32_t port);

    // Start discovery against a random host from existing hosts

    void startResolveRedis();

    // Extensions::NetworkFilters::Common::Redis::Client::Config

      // Allow the main Health Check infra to control timeout.

    // For any readPolicy other than Primary, the RedisClientFactory will send a READONLY command

    // when establishing a new connection. Since we're only using this for making the "cluster

    // slots" commands, the READONLY command is not relevant in this context. We're setting it to

    // Primary to avoid the additional READONLY command.

    // Extensions::NetworkFilters::Common::Redis::Client::ClientCallbacks

    void onResponse(NetworkFilters::Common::Redis::RespValuePtr&& value) override;

    void onFailure() override;

    // Note: Below callback isn't used in topology updates

    void onRedirection(NetworkFilters::Common::Redis::RespValuePtr&&, const std::string&,

    void onUnexpectedResponse(const NetworkFilters::Common::Redis::RespValuePtr&);

    Network::Address::InstanceConstSharedPtr

    ipAddressFromClusterEntry(const std::vector<NetworkFilters::Common::Redis::RespValue>& array);

    bool validateCluster(const NetworkFilters::Common::Redis::RespValue& value);

    void resolveClusterHostnames(ClusterSlotsSharedPtr&& slots,

                                 std::shared_ptr<std::uint64_t> hostname_resolution_required_cnt);

    void resolveReplicas(ClusterSlotsSharedPtr slots, std::size_t index,

                         std::shared_ptr<std::uint64_t> hostname_resolution_required_cnt);

    void finishClusterHostnameResolution(ClusterSlotsSharedPtr slots);

    void updateDnsStats(Network::DnsResolver::ResolutionStatus status, bool empty_response);

    RedisCluster& parent_;

    Event::Dispatcher& dispatcher_;

    std::string current_host_address_;

    Extensions::NetworkFilters::Common::Redis::Client::PoolRequest* current_request_{};

    absl::node_hash_map<std::string, RedisDiscoveryClientPtr> client_map_;

    std::list<Network::Address::InstanceConstSharedPtr> discovery_address_list_;

    Event::TimerPtr resolve_timer_;

    NetworkFilters::Common::Redis::Client::ClientFactory& client_factory_;

    const std::chrono::milliseconds buffer_timeout_;

    NetworkFilters::Common::Redis::RedisCommandStatsSharedPtr redis_command_stats_;

  };

  Upstream::ClusterManager& cluster_manager_;

  const std::chrono::milliseconds cluster_refresh_rate_;

  const std::chrono::milliseconds cluster_refresh_timeout_;

  const std::chrono::milliseconds redirect_refresh_interval_;

  const uint32_t redirect_refresh_threshold_;

  const uint32_t failure_refresh_threshold_;

  const uint32_t host_degraded_refresh_threshold_;

  std::list<DnsDiscoveryResolveTargetPtr> dns_discovery_resolve_targets_;

  Event::Dispatcher& dispatcher_;

  Network::DnsResolverSharedPtr dns_resolver_;

  Network::DnsLookupFamily dns_lookup_family_;

  const envoy::config::endpoint::v3::ClusterLoadAssignment load_assignment_;

  const LocalInfo::LocalInfo& local_info_;

  Random::RandomGenerator& random_;

  std::shared_ptr<RedisDiscoverySession> redis_discovery_session_;

  const ClusterSlotUpdateCallBackSharedPtr lb_factory_;

  Upstream::HostVector hosts_;

  const std::string auth_username_;

  const std::string auth_password_;

  const std::string cluster_name_;

  const Common::Redis::ClusterRefreshManagerSharedPtr refresh_manager_;

  Common::Redis::ClusterRefreshManager::HandlePtr registration_handle_;

  // Flag to prevent callbacks during destruction

  std::atomic<bool> is_destroying_{false};

};

class RedisClusterFactory : public Upstream::ConfigurableClusterFactoryBase<

                                envoy::extensions::clusters::redis::v3::RedisClusterConfig> {

public:

private:

  friend class RedisClusterTest;

  absl::StatusOr<

      std::pair<Upstream::ClusterImplBaseSharedPtr, Upstream::ThreadAwareLoadBalancerPtr>>

  createClusterWithConfig(

      const envoy::config::cluster::v3::Cluster& cluster,

      const envoy::extensions::clusters::redis::v3::RedisClusterConfig& proto_config,

      Upstream::ClusterFactoryContext& context) override;

};

} // namespace Redis

} // namespace Clusters

} // namespace Extensions

} // namespace Envoy