package ratecontroller

import (
	"context"
	"fmt"
	"time"

	"github.com/go-redis/redis/v8"
	"github.com/pingcap/dbaas/pkg/ratecontroller/types"
	"github.com/pingcap/errors"
)

type redisStore struct {
	cli redis.Client
}

func NewRedisStore(capacity int, ttl time.Duration, cli redis.Client) *redisStore {
	return &redisStore{
		cli: cli,
	}
}

func (r *redisStore) AddEntry(ctx context.Context, scene *types.Scene, accessor, target string, ts time.Time) error {
	currentMillisecond := ts.UnixNano() / int64(time.Millisecond)

	key := r.makeKey(scene.SceneKey, accessor)
	pl := r.cli.Pipeline()
	// add access log, current millisecond as score of member
	pl.ZAdd(ctx, key, &redis.Z{
		Score:  float64(currentMillisecond),
		Member: target,
	})

	pl.Expire(ctx, key, scene.TTL)
	if _, err := pl.Exec(ctx); err != nil {
		return errors.Annotate(err, "pipeline failed")
	}
	return nil
}

func (r *redisStore) PurgeExpired(ctx context.Context, scene *types.Scene, accessor string, expiredFrom time.Time) error {
	// do nothing
	return nil
}

func (r *redisStore) PurgeOverflowed(ctx context.Context, scene *types.Scene, accessor string) error {
	key := r.makeKey(scene.SceneKey, accessor)
	// purge old access log
	if err := r.cli.ZRemRangeByRank(ctx, key, 0, int64(-scene.Capacity-1)).Err(); err != nil {
		return errors.Annotate(err, "ZRemRangeByRank failed")
	}
	return nil
}

func (r *redisStore) Count(ctx context.Context, scene *types.Scene, accessor string, startTime, endTime time.Time) (int64, error) {
	key := r.makeKey(scene.SceneKey, accessor)
	startScore := startTime.UnixNano() / int64(time.Millisecond)
	endScore := endTime.UnixNano() / int64(time.Millisecond)

	result := r.cli.ZCount(ctx, key, fmt.Sprint(startScore), fmt.Sprint(endScore))
	if err := result.Err(); err != nil {
		return 0, errors.Annotate(err, "ZCount failed")
	}
	return result.Val(), nil
}

func (r *redisStore) makeKey(sceneKey, accessor string) string {
	return fmt.Sprintf("ratecontroller:redisStore:%s:%s", sceneKey, accessor)
}