package rediczar

import (
	"context"
	"errors"
	"fmt"
	"time"

	"github.com/go-redis/redis/v7"
)

// Pipeline can send batches of commands and get a reply in a single step. See https://redis.io/topics/pipelining
// Exec() on the Pipeliner executes the batch. Close() the pipeliner after using it.
func (c *PrefixedCommands) Pipeline(ctx context.Context) redis.Pipeliner {
	ctxCmdable := cmdableWithContext(ctx, c.Redis)

	// Prefix all pipeline commands
	pfxPipeliner := prefixerPipeliner{
		Pipeliner: ctxCmdable.Pipeline(),
		keyPrefix: c.KeyPrefix,
	}

	return &pfxPipeliner
}

// Pipelined is like Pipeline. The commands are automatically executed at the end of fn
func (c *PrefixedCommands) Pipelined(ctx context.Context, fn func(redis.Pipeliner) error) ([]redis.Cmder, error) {
	ctxCmdable := cmdableWithContext(ctx, c.Redis)

	// Prefix all pipeline commands
	pfxPipeliner := prefixerPipeliner{
		Pipeliner: ctxCmdable.Pipeline(),
		keyPrefix: c.KeyPrefix,
	}

	return pfxPipeliner.Pipelined(func(_ redis.Pipeliner) error {
		// Call with prefixed pipeliner
		return fn(&pfxPipeliner)
	})
}

// TxPipeline is like Pipeline, but executes in a transaction using MULTI and EXEC. See https://redis.io/topics/transactions
// Exec() on the Pipeliner executes the transaction. Close() the pipeliner after using it.
func (c *PrefixedCommands) TxPipeline(ctx context.Context) redis.Pipeliner {
	ctxCmdable := cmdableWithContext(ctx, c.Redis)

	// Prefix all pipeline commands
	pfxPipeliner := prefixerPipeliner{
		Pipeliner: ctxCmdable.TxPipeline(),
		keyPrefix: c.KeyPrefix,
	}

	return &pfxPipeliner
}

// TxPipelined is like TxPipeline. The commands are automatically executed at the end of fn
func (c *PrefixedCommands) TxPipelined(ctx context.Context, fn func(redis.Pipeliner) error) ([]redis.Cmder, error) {
	ctxCmdable := cmdableWithContext(ctx, c.Redis)

	// Prefix all pipeline commands
	pfxPipeliner := prefixerPipeliner{
		Pipeliner: ctxCmdable.TxPipeline(),
		keyPrefix: c.KeyPrefix,
	}

	return pfxPipeliner.TxPipelined(func(_ redis.Pipeliner) error {
		// Call with prefixed pipeliner
		return fn(&pfxPipeliner)
	})
}

// MSetNXWithTTL leverages redis pipelining to issue many SET commands with NX and TTL specified.
// Pipelining is necessary since MSET alone does not support NX and TTL arguments. Booleans
// are returned to indicate whether each key-pair was set.
func (c *Client) MSetNXWithTTL(ctx context.Context, ttl time.Duration, pairs ...interface{}) ([]bool, error) {
	if len(pairs) == 0 {
		return nil, errors.New("MSetNXWithTTL is missing arguments")
	}

	if len(pairs)%2 != 0 {
		return nil, errors.New("MSetNXWithTTL requires an even number of key-value pairs")
	}

	cmds := make([]*redis.BoolCmd, len(pairs)/2)

	_, err := c.Pipelined(ctx, func(p redis.Pipeliner) error {
		for i := 0; i < len(pairs); i += 2 {
			key, ok := pairs[i].(string)
			if !ok {
				return fmt.Errorf("MSetNXWithTTL: received non-string key index=%d got=%v", i, pairs[i])
			}
			cmds[i/2] = p.SetNX(key, pairs[i+1], ttl)
		}

		return nil
	})

	if err != nil {
		return nil, err
	}

	res := make([]bool, len(pairs)/2)

	for i, cmd := range cmds {
		res[i] = cmd.Val()
	}

	return res, nil
}

// MSetWithTTL leverages redis pipelining to issue many SET commands with TTL specified.
// Pipelining is necessary since MSet alone does not support TTL arguments.
func (c *Client) MSetWithTTL(ctx context.Context, ttl time.Duration, pairs ...interface{}) error {
	if len(pairs) == 0 {
		return errors.New("MSetWithTTL is missing arguments")
	}

	if len(pairs)%2 != 0 {
		return errors.New("MSetWithTTL requires an even number of key-value pairs")
	}

	_, err := c.Pipelined(ctx, func(p redis.Pipeliner) error {
		for i := 0; i < len(pairs); i += 2 {
			key, ok := pairs[i].(string)
			if !ok {
				return fmt.Errorf("MSetWithTTL: received non-string key index=%d got=%v", i, pairs[i])
			}
			p.Set(key, pairs[i+1], ttl)
		}

		return nil
	})

	if err != nil {
		return err
	}

	return nil
}

// PipelinedGet leverages redis pipelining to issue multiple GET commands and returns the values of specified keys
// This can be used to perform a MGet across a cluster without client side hashing
// If the key is not found for index i, then the value is nil, otherwise the value is a string
func (c *Client) PipelinedGet(ctx context.Context, keys ...string) ([]interface{}, error) {
	cmds := make([]*redis.StringCmd, len(keys))

	_, err := c.Pipelined(ctx, func(p redis.Pipeliner) error {
		for i, key := range keys {
			cmds[i] = p.Get(key)
		}

		return nil
	})

	if err != nil && err != redis.Nil {
		return nil, err
	}

	resp := make([]interface{}, len(keys))

	for i, cmd := range cmds {
		val, err := cmd.Result()
		if err != nil {
			if err == redis.Nil {
				resp[i] = nil
			} else {
				return nil, err
			}
		} else {
			resp[i] = val
		}
	}

	return resp, nil
}

// PipelinedDel leverages redis pipelining to issue multiple DEL commands and returns the
// number of keys that were deleted
// This can be used to perform DELs on multiple keys across a cluster without client side hashing
func (c *Client) PipelinedDel(ctx context.Context, keys ...string) (int64, error) {
	cmds := make([]*redis.IntCmd, len(keys))

	_, err := c.Pipelined(ctx, func(p redis.Pipeliner) error {
		for i, key := range keys {
			cmds[i] = p.Del(key)
		}
		return nil
	})

	if err != nil {
		return 0, err
	}

	var deleted int64

	for _, cmd := range cmds {
		deleted += cmd.Val()
	}

	return deleted, nil
}
