package redistool

import (
	"context"
	"encoding/base64"
	"errors"
	"log/slog"
	"strconv"
	"time"
	"unsafe"

	"github.com/redis/rueidis"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/tool/logz"
	"k8s.io/utils/clock"
)

type RPCAPI interface {
	Log() *slog.Logger
	HandleProcessingError(msg string, err error)
	RequestKey() []byte
}

// TokenLimiter is a redis-based rate limiter implementing the algorithm in https://redislabs.com/redis-best-practices/basic-rate-limiting/
type TokenLimiter struct {
	redisClient    rueidis.Client
	clock          clock.PassiveClock
	keyPrefix      string
	limitPerMinute uint64
	getAPI         func(context.Context) RPCAPI
}

// NewTokenLimiter returns a new TokenLimiter
func NewTokenLimiter(redisClient rueidis.Client, keyPrefix string,
	limitPerMinute uint64, getAPI func(context.Context) RPCAPI) *TokenLimiter {
	return &TokenLimiter{
		redisClient:    redisClient,
		clock:          clock.RealClock{},
		keyPrefix:      keyPrefix,
		limitPerMinute: limitPerMinute,
		getAPI:         getAPI,
	}
}

// Allow consumes one limitable event from the token in the context
func (l *TokenLimiter) Allow(ctx context.Context) bool {
	api := l.getAPI(ctx)
	key := buildTokenLimiterKey(l.keyPrefix, api.RequestKey(), byte(l.clock.Now().UTC().Minute()))
	getCmd := l.redisClient.B().Get().Key(key).Build()

	// Use the caller context to propagate OTEL span, etc (if used).
	// Don't care about caller's context cancellation or timeout - we don't want to see errors from that
	// so that we are able to tell if Redis is ok or not.
	// Only want DeadlineExceed from the timeout we set below.
	// That way we can clearly tell Redis taking too long vs caller aborting the call.
	redisCtx, redisCancel := context.WithTimeout(context.WithoutCancel(ctx), 3*time.Second)
	defer redisCancel()
	count, err := l.redisClient.Do(redisCtx, getCmd).AsUint64()
	if err != nil {
		if err != rueidis.Nil { //nolint:errorlint
			api.HandleProcessingError("redistool.TokenLimiter: error retrieving minute bucket count", err)
			return false
		}
		count = 0
	}
	if count >= l.limitPerMinute {
		api.Log().Info("redistool.TokenLimiter: rate limit exceeded",
			logz.RedisKey([]byte(key)), logz.U64Count(count), logz.TokenLimit(l.limitPerMinute))
		return false
	}

	resp := l.redisClient.DoMulti(redisCtx,
		l.redisClient.B().Multi().Build(),
		l.redisClient.B().Incr().Key(key).Build(),
		l.redisClient.B().Expire().Key(key).Seconds(59).Build(),
		l.redisClient.B().Exec().Build(),
	)
	err = errors.Join(MultiErrors(resp)...)
	if err != nil {
		api.HandleProcessingError("redistool.TokenLimiter: error while incrementing token key count", err)
		return false
	}

	return true
}

func buildTokenLimiterKey(keyPrefix string, requestKey []byte, currentMinute byte) string {
	result := make([]byte, 0, len(keyPrefix)+1+base64.StdEncoding.EncodedLen(len(requestKey))+1+2)
	result = append(result, keyPrefix...)
	result = append(result, ':')
	result = base64.StdEncoding.AppendEncode(result, requestKey)
	result = append(result, ':')
	result = strconv.AppendInt(result, int64(currentMinute), 10)

	return unsafe.String(unsafe.SliceData(result), len(result)) //nolint: gosec
}
