package redefault

import (
	"context"
	"fmt"
	"net"
	"time"

	"code.justin.tv/chat/ratecache"
	"github.com/go-redis/redis/v7"
	"golang.org/x/time/rate"
)

// ClusterOpts specifies options for the cluster client
type ClusterOpts struct {
	// ReadOnly will read from replica slaves. Leave this as false for strong read-after-write consistency. Set this to true if you want
	// to spread load across read replicas.
	ReadOnly bool
	// Dialer specifies how new network connections will be created. Defaults to a standard net.Dialer with a 1s timeout and a 30s keep alive
	Dialer func(ctx context.Context, network, addr string) (net.Conn, error)
	// PoolSize determines the size of the fixed pool. Defaults to 30.
	PoolSize int
	// MaxRedirects specifies how many redirects to follow on MOVED/ASKED. Defaults to 1.
	MaxRedirects int
	// ReadTimeout defaults to 1s. This is how long the client will wait to read a response from a redis node.
	ReadTimeout time.Duration
	// WriteTimeout defaults to 500ms. This is how long the client will wait to write a request to a redis node.
	WriteTimeout time.Duration
}

// fillDefaults with settings load tested in https://git-aws.internal.justin.tv/chat/redismembench
// The load test generates 80k RPS (broken down into 65K GET RPS and 15K SetNX RPS) using 30 processes; each process generates ~2.6k rps.
func (c *ClusterOpts) fillDefaults() {
	if c.PoolSize <= 0 {
		c.PoolSize = 30
	}

	if c.MaxRedirects <= 0 {
		// The default is 8. This reduces how many times the cluster client follows redirects or retries. Reducing this number
		// prevents the client from generating too much load on retries. Error rates were low when testing node failover with
		// this setting.
		c.MaxRedirects = 1
	}

	// These Read and Write timeouts were tuned to reduce the number of client timeout errors during memory pressure.
	if c.ReadTimeout <= 0 {
		c.ReadTimeout = 1 * time.Second
	}

	if c.WriteTimeout <= 0 {
		c.WriteTimeout = 500 * time.Millisecond
	}
}

// NewClusterClient creates a new cluster client with a rate-limiting dialer, a fixed pool size, and reasonable defaults.
// opts is optional and can be left as nil. The rate-limiting will allow 1 connection per second, and a burst for first-time
// initialization of the cluster client. This function is intended to handle boiler plate, but can be copy-pasted if full
// customization is desired.
func NewClusterClient(addr string, opts *ClusterOpts) *redis.ClusterClient {
	if opts == nil {
		opts = &ClusterOpts{}
	}

	opts.fillDefaults()

	var dialer func(ctx context.Context, network, addr string) (net.Conn, error)

	if opts.Dialer != nil {
		dialer = opts.Dialer
	} else {
		d := net.Dialer{
			Timeout:   1 * time.Second,
			KeepAlive: 30 * time.Second,
		}

		dialer = d.DialContext
	}

	poolSize := opts.PoolSize

	// Maximum number of conns per minute to a node is: (Number of instances) * (Burst+60). The goal of the
	// rate limiting is to prevent tens of thousands of connections per minute, which increases CPU load. Also,
	// too many new connections can exhaust ephemeral ports on a node, which causes internal health checks to
	// fail and trigger a node replacement.
	rateCache := ratecache.Store{
		Limit: rate.Every(time.Second),
		Burst: poolSize + (poolSize / 2), // Add burst headroom
	}

	client := redis.NewClusterClient(&redis.ClusterOptions{
		Addrs:        []string{addr},
		MaxRedirects: opts.MaxRedirects,
		ReadOnly:     opts.ReadOnly,
		Dialer: func(ctx context.Context, network string, address string) (net.Conn, error) {
			limiter := rateCache.CreateOrGet(address)
			if !limiter.Allow() {
				return nil, fmt.Errorf("dialing connection rate limited: address=%s", address)
			}

			return dialer(ctx, network, address)
		},
		// The Read/Write timeouts work well when the cluster is under memory pressure to minimize errors and new connections.
		ReadTimeout:        opts.ReadTimeout,
		WriteTimeout:       opts.WriteTimeout,
		MinIdleConns:       0,
		PoolSize:           poolSize,
		IdleTimeout:        -1,
		IdleCheckFrequency: -1,
	})

	return client
}
