package redistool

import (
	"context"
	"crypto/rand"
	"fmt"
	"net/url"
	"os"
	"strconv"
	"testing"
	"time"

	"github.com/redis/rueidis"
	rmock "github.com/redis/rueidis/mock"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/api"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/tool/testing/mock_otel"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/tool/tlstool"
	metricnoop "go.opentelemetry.io/otel/metric/noop"
	"go.uber.org/mock/gomock"
	"google.golang.org/protobuf/proto"
)

const (
	redisURLEnvName = "REDIS_URL"
	ttl             = 2 * time.Second
)

var (
	_ ExpiringHash[int, int]    = (*RedisExpiringHash[int, int])(nil)
	_ ExpiringHashAPI[int, int] = (*RedisExpiringHashAPI[int, int])(nil)
)

func TestExpiringHash_Expires(t *testing.T) {
	client, hash, key, value, _ := setupHash(t)

	require.NoError(t, hash.SetEX(context.Background(), key, 123, value, time.Now()))
	time.Sleep(ttl + 100*time.Millisecond)

	require.Empty(t, getHash(t, client, key))
}

func TestExpiringHash_GC_for_SetEX(t *testing.T) {
	client, hash, key, value, gcHistMock := setupHash(t)

	require.NoError(t, hash.SetEX(context.Background(), key, 123, value, time.Now().Add(ttl)))
	newExpireIn := 3 * ttl
	cmd := client.B().Pexpire().Key(key).Milliseconds(newExpireIn.Milliseconds()).Build()
	err := client.Do(context.Background(), cmd).Error()
	require.NoError(t, err)
	time.Sleep(ttl + time.Second)
	require.NoError(t, hash.SetEX(context.Background(), key, 321, value, time.Now().Add(ttl)))

	expectedKeysDeleted := 1

	gcHistMock.EXPECT().Add(gomock.Any(), int64(expectedKeysDeleted), gomock.Any())

	keysDeleted, err := hash.GC(context.Background())
	require.NoError(t, err)
	assert.Equal(t, expectedKeysDeleted, keysDeleted)

	equalHash(t, client, key, 321, value)
}

func TestExpiringHash_GCContinuesOnConflict(t *testing.T) {
	ctrl := gomock.NewController(t)
	client := rmock.NewClient(ctrl)
	meter := mock_otel.NewMockMeter(ctrl)
	gcCounterMock := mock_otel.NewMockInt64Counter(ctrl)
	gcConflictMock := mock_otel.NewMockInt64Counter(ctrl)
	meter.EXPECT().
		Int64Counter(gcDeletedKeysMetricName, gomock.Any()).
		Return(gcCounterMock, nil)
	meter.EXPECT().
		Int64Counter(gcConflictMetricName, gomock.Any()).
		Return(gcConflictMock, nil)

	gcCounterMock.EXPECT().Add(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes()
	gcConflictMock.EXPECT().Add(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes()

	ded := rmock.NewDedicatedClient(ctrl)
	val := &ExpiringValue{
		ExpiresAt: time.Now().Unix() - 1000, // long expired
		Value:     nil,
	}
	valBytes, err := proto.Marshal(val)
	require.NoError(t, err)
	watch := func() *gomock.Call {
		return ded.EXPECT(). // Transaction
					Do(
				gomock.Any(),
				matchCmd("WATCH"),
			)
	}
	scan := func() *gomock.Call {
		return ded.EXPECT(). // scan()
					Do(
				gomock.Any(),
				matchCmd("HSCAN"),
			).
			Return(rmock.Result(rmock.RedisArray(
				rmock.RedisInt64(0), // 0 means no more elements to scan
				rmock.RedisArray(rmock.RedisString("key"), rmock.RedisString(string(valBytes))),
			)))
	}
	failDelete := func() *gomock.Call {
		return ded.EXPECT().
			DoMulti(
				gomock.Any(),
				rmock.Match("MULTI"),
				matchCmd("HDEL"),
				rmock.Match("EXEC"),
			).
			Return([]rueidis.RedisResult{
				rmock.Result(rmock.RedisString("OK")),
				rmock.Result(rmock.RedisString("QUEUED")),
				rmock.Result(rmock.RedisNil()), // conflict!
			})
	}
	set := func(key string) *gomock.Call {
		return client.EXPECT().
			DoMulti(
				gomock.Any(),
				rmock.Match("MULTI"),
				rmock.MatchFn(func(cmd []string) bool {
					return len(cmd) == 4 && cmd[0] == "HSET" && cmd[1] == key && cmd[2] == "key"
				}),
				rmock.Match("PEXPIRE", key, "2000"),
				rmock.Match("EXEC"),
			)
	}
	gomock.InOrder(
		set("1"), // SET 1,
		set("2"), // SET 2,
		client.EXPECT(). // GC() start
					Dedicate().
					Return(ded, func() {}),
		// GC first element - attempt 1
		watch(),
		scan(),
		failDelete(),
		// GC first element - attempt 2
		watch(),
		scan(),
		failDelete(),
		// GC second element
		watch(),
		scan(),
		ded.EXPECT().
			DoMulti(
				gomock.Any(),
				rmock.Match("MULTI"),
				matchCmd("HDEL"),
				rmock.Match("EXEC"),
			).
			Return([]rueidis.RedisResult{
				rmock.Result(rmock.RedisString("OK")),
				rmock.Result(rmock.RedisString("QUEUED")),
				rmock.Result(rmock.RedisArray(rmock.RedisInt64(1))), // removed 1
			}),
	)

	hash, err := NewRedisExpiringHash("test", client, s2s, s2s, ttl, meter, true)
	require.NoError(t, err)
	err = hash.SetEX(context.Background(), "1", "key", nil, time.Now().Add(1*time.Hour))
	require.NoError(t, err)
	err = hash.SetEX(context.Background(), "2", "key", nil, time.Now().Add(1*time.Hour))
	require.NoError(t, err)

	deleted, err := hash.GC(context.Background())
	require.NoError(t, err)
	assert.Equal(t, 1, deleted)
}

func TestExpiringHash_GCBatch_Transaction(t *testing.T) {
	ctrl := gomock.NewController(t)
	client := rmock.NewClient(ctrl)
	meter := mock_otel.NewMockMeter(ctrl)
	gcCounterMock := mock_otel.NewMockInt64Counter(ctrl)
	gcConflictMock := mock_otel.NewMockInt64Counter(ctrl)
	meter.EXPECT().
		Int64Counter(gcDeletedKeysMetricName, gomock.Any()).
		Return(gcCounterMock, nil)
	meter.EXPECT().
		Int64Counter(gcConflictMetricName, gomock.Any()).
		Return(gcConflictMock, nil)

	gcCounterMock.EXPECT().Add(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes()
	gcConflictMock.EXPECT().Add(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes()

	ded := rmock.NewDedicatedClient(ctrl)
	val := &ExpiringValue{
		ExpiresAt: time.Now().Unix() - 1000, // long expired
		Value:     nil,
	}
	valBytes, err := proto.Marshal(val)
	require.NoError(t, err)
	watch := func() *gomock.Call {
		return ded.EXPECT(). // Transaction
					Do(
				gomock.Any(),
				matchCmd("WATCH"),
			)
	}
	scan := func(n int, cursor int64) *gomock.Call {
		res := make([]rueidis.RedisMessage, 0, n*2)
		for i := range n {
			res = append(res, rmock.RedisString(fmt.Sprintf("key_%d", i)), rmock.RedisString(rueidis.BinaryString(valBytes)))
		}
		return ded.EXPECT(). // scan()
					Do(
				gomock.Any(),
				matchCmd("HSCAN"),
			).
			Return(rmock.Result(
				rmock.RedisArray(
					rmock.RedisInt64(cursor),
					rmock.RedisArray(res...),
				),
			))
	}
	set := func(key string) *gomock.Call {
		return client.EXPECT().
			DoMulti(
				gomock.Any(),
				rmock.Match("MULTI"),
				rmock.MatchFn(func(cmd []string) bool {
					return len(cmd) == 4 && cmd[0] == "HSET" && cmd[1] == key && cmd[2] == "key"
				}),
				rmock.Match("PEXPIRE", key, "2000"),
				rmock.Match("EXEC"),
			)
	}
	del := func(n int64) *gomock.Call {
		return ded.EXPECT().
			DoMulti(
				gomock.Any(),
				rmock.Match("MULTI"),
				matchCmd("HDEL"),
				rmock.Match("EXEC"),
			).
			Return([]rueidis.RedisResult{
				rmock.Result(rmock.RedisString("OK")),
				rmock.Result(rmock.RedisString("QUEUED")),
				rmock.Result(rmock.RedisArray(rmock.RedisInt64(n))), // removed n elements
			})
	}
	gomock.InOrder(
		set("1"), // SET 1,
		client.EXPECT(). // GC() start
					Dedicate().
					Return(ded, func() {}),
		// GC first batch
		watch(),
		scan(maxDeleteBatchSize, 5),
		del(maxDeleteBatchSize),
		//// GC second batch
		watch(),
		scan(1, 0),
		del(1),
	)

	hash, err := NewRedisExpiringHash("test", client, s2s, s2s, ttl, meter, true)
	require.NoError(t, err)
	err = hash.SetEX(context.Background(), "1", "key", nil, time.Now().Add(1*time.Hour))
	require.NoError(t, err)

	deleted, err := hash.GC(context.Background())
	require.NoError(t, err)
	assert.Equal(t, maxDeleteBatchSize+1, deleted)
}

func matchCmd(cmd string) gomock.Matcher {
	return rmock.MatchFn(func(cmdAndArgs []string) bool {
		return cmdAndArgs[0] == cmd
	})
}

func s2s(key string) string {
	return key
}

func TestExpiringHash_ScanEmpty(t *testing.T) {
	_, hash, key, _, _ := setupHash(t)

	for _, err := range hash.Scan(context.Background(), key) {
		require.NoError(t, err)
		assert.FailNow(t, "unexpected callback invocation")
	}
}

func TestExpiringHash_Scan(t *testing.T) {
	_, hash, key, value, _ := setupHash(t)
	cbCalled := false

	require.NoError(t, hash.SetEX(context.Background(), key, 123, value, time.Now().Add(1*time.Hour)))
	for se, err := range hash.Scan(context.Background(), key) {
		cbCalled = true
		require.NoError(t, err)
		assert.Equal(t, value, se.Value)
		assert.Equal(t, "123", se.RawHashKey)
	}
	assert.True(t, cbCalled)
}

func TestExpiringHash_Len(t *testing.T) {
	_, hash, key, value, _ := setupHash(t)
	require.NoError(t, hash.SetEX(context.Background(), key, 123, value, time.Now().Add(1*time.Hour)))
	size, err := hash.Len(context.Background(), key)
	require.NoError(t, err)
	assert.EqualValues(t, 1, size)
}

func TestTransactionConflict_Sibling(t *testing.T) {
	client, _, key, _, _ := setupHash(t)
	c, cancel := client.Dedicate()
	defer cancel()
	iteration := 0
	tm := metricnoop.NewMeterProvider().Meter("test")
	gcC, err := tm.Int64Counter(gcConflictMetricName)
	require.NoError(t, err)
	err = transaction(context.Background(), 10, c, gcC, nil, func(ctx context.Context) ([]rueidis.Completed, error) {
		switch iteration {
		case 0:
			// Mutate a sibling mapping on purpose to trigger a conflict situation.
			doErr := client.Do(context.Background(), client.B().Hset().Key(key).FieldValue().FieldValue("1", "123").Build()).Error() //nolint: contextcheck
			if doErr != nil {
				return nil, doErr
			}
		case 1:
			// the expected retry
		default:
			require.FailNow(t, "unexpected invocation")
		}
		iteration++
		return []rueidis.Completed{
			client.B().Hset().Key(key).FieldValue().FieldValue("2", "234").Build(),
		}, nil
	}, key)
	require.NoError(t, err)
	v1, err := client.Do(context.Background(), client.B().Hget().Key(key).Field("1").Build()).ToString()
	require.NoError(t, err)
	assert.Equal(t, "123", v1)
	v2, err := client.Do(context.Background(), client.B().Hget().Key(key).Field("2").Build()).ToString()
	require.NoError(t, err)
	assert.Equal(t, "234", v2)
}

func TestTransactionConflict_SameKey(t *testing.T) {
	client, _, key, _, _ := setupHash(t)
	c, cancel := client.Dedicate()
	defer cancel()
	iteration := 0
	tm := metricnoop.NewMeterProvider().Meter("test")
	gcC, err := tm.Int64Counter(gcConflictMetricName)
	require.NoError(t, err)
	err = transaction(context.Background(), 10, c, gcC, nil, func(ctx context.Context) ([]rueidis.Completed, error) {
		switch iteration {
		case 0:
			// Mutate a sibling mapping on purpose to trigger a conflict situation.
			doErr := client.Do(context.Background(), client.B().Hset().Key(key).FieldValue().FieldValue("1", "xxxxxx").Build()).Error() //nolint: contextcheck
			if doErr != nil {
				return nil, doErr
			}
		case 1:
			// the expected retry
		default:
			require.FailNow(t, "unexpected invocation")
		}
		iteration++
		return []rueidis.Completed{
			client.B().Hset().Key(key).FieldValue().FieldValue("1", "123").Build(),
		}, nil
	}, key)
	require.NoError(t, err)
	v1, err := client.Do(context.Background(), client.B().Hget().Key(key).Field("1").Build()).ToString()
	require.NoError(t, err)
	assert.Equal(t, "123", v1)
}

func TestTransactionConflict_AttemptsExceeded(t *testing.T) {
	client, _, key, _, _ := setupHash(t)
	c, cancel := client.Dedicate()
	defer cancel()
	iteration := 0
	tm := metricnoop.NewMeterProvider().Meter("test")
	gcC, err := tm.Int64Counter(gcConflictMetricName)
	require.NoError(t, err)
	err = transaction(context.Background(), 1, c, gcC, nil, func(ctx context.Context) ([]rueidis.Completed, error) {
		switch iteration {
		case 0:
			// Mutate a sibling mapping on purpose to trigger a conflict situation.
			doErr := client.Do(context.Background(), client.B().Hset().Key(key).FieldValue().FieldValue("1", "xxxxxx").Build()).Error() //nolint: contextcheck
			if doErr != nil {
				return nil, doErr
			}
		default:
			require.FailNow(t, "unexpected invocation")
		}
		iteration++
		return []rueidis.Completed{
			client.B().Hset().Key(key).FieldValue().FieldValue("1", "123").Build(),
		}, nil
	}, key)
	require.Same(t, errAttemptsExceeded, err)
	v1, err := client.Do(context.Background(), client.B().Hget().Key(key).Field("1").Build()).ToString()
	require.NoError(t, err)
	assert.Equal(t, "xxxxxx", v1)
}

func BenchmarkExpiringValue_Unmarshal(b *testing.B) {
	d, err := proto.Marshal(&ExpiringValue{
		ExpiresAt: 123123123,
		Value:     []byte("1231231231232313"),
	})
	require.NoError(b, err)
	b.Run("ExpiringValue", func(b *testing.B) {
		b.ReportAllocs()
		var val ExpiringValue
		for b.Loop() {
			err = proto.Unmarshal(d, &val)
		}
	})
	b.Run("ExpiringValueTimestamp", func(b *testing.B) {
		b.ReportAllocs()
		var val ExpiringValueTimestamp
		for b.Loop() {
			err = proto.Unmarshal(d, &val)
		}
	})
	b.Run("ExpiringValueTimestamp DiscardUnknown", func(b *testing.B) {
		b.ReportAllocs()
		var val ExpiringValueTimestamp
		for b.Loop() {
			err = proto.UnmarshalOptions{
				DiscardUnknown: true,
			}.Unmarshal(d, &val)
		}
	})
}

func BenchmarkPrefixedInt64Key(b *testing.B) {
	b.ReportAllocs()
	const prefix = "pref"
	var sink string
	for i := range b.N {
		sink = PrefixedInt64Key(prefix, int64(i))
	}
	_ = sink
}

func BenchmarkPrefixedAgentKey(b *testing.B) {
	b.ReportAllocs()
	const prefix = "pref"
	var sink string
	for i := range b.N {
		sink = PrefixedAgentKey(prefix, api.AgentKey{ID: int64(i)})
	}
	_ = sink
}

func TestPrefixedInt64Key(t *testing.T) {
	const prefix = "pref"
	key := PrefixedInt64Key(prefix, 0x1122334455667788)

	assert.Equal(t, prefix+"128hj8hamcts8", key)
}

func TestPrefixedAgentKey(t *testing.T) {
	const prefix = "pref"
	key := PrefixedAgentKey(prefix, api.AgentKey{ID: 0x1122334455667788})

	assert.Equal(t, prefix+"agentk:128hj8hamcts8", key)
}

func setupHash(t *testing.T) (rueidis.Client, *RedisExpiringHash[string, int64], string, []byte, *mock_otel.MockInt64Counter) {
	t.Parallel()
	client := redisClient(t)
	ctrl := gomock.NewController(t)
	meter := mock_otel.NewMockMeter(ctrl)
	gcCounterMock := mock_otel.NewMockInt64Counter(ctrl)
	gcConflictMock := mock_otel.NewMockInt64Counter(ctrl)
	meter.EXPECT().
		Int64Counter(gcDeletedKeysMetricName, gomock.Any()).
		Return(gcCounterMock, nil)
	meter.EXPECT().
		Int64Counter(gcConflictMetricName, gomock.Any()).
		Return(gcConflictMock, nil)

	t.Cleanup(client.Close)
	prefix := make([]byte, 32)
	_, err := rand.Read(prefix)
	require.NoError(t, err)
	key := string(prefix)
	strToStr := func(key string) string {
		return key
	}
	hash, err := NewRedisExpiringHash("test", client, strToStr, int64ToStr, ttl, meter, true)
	require.NoError(t, err)
	return client, hash, key, []byte{1, 2, 3}, gcCounterMock
}

func redisClient(t *testing.T) rueidis.Client {
	redisURL := os.Getenv(redisURLEnvName)
	if redisURL == "" {
		t.Skipf("%s environment variable not set, skipping test", redisURLEnvName)
	}

	u, err := url.Parse(redisURL)
	require.NoError(t, err)
	opts := rueidis.ClientOption{
		ClientName:   "gitlab-agent-test",
		DisableCache: true,
	}
	switch u.Scheme {
	case "unix":
		opts.DialCtxFn = UnixDialer
		opts.InitAddress = []string{u.Path}
	case "redis":
		opts.InitAddress = []string{u.Host}
	case "rediss":
		opts.InitAddress = []string{u.Host}
		opts.TLSConfig = tlstool.ClientConfig()
	default:
		opts.InitAddress = []string{redisURL}
	}
	client, err := rueidis.NewClient(opts)
	require.NoError(t, err)
	return client
}

func getHash(t *testing.T, client rueidis.Client, key string) map[string]string {
	cmd := client.B().Hgetall().Key(key).Build()
	reply, err := client.Do(context.Background(), cmd).AsStrMap()
	require.NoError(t, err)
	return reply
}

func equalHash(t *testing.T, client rueidis.Client, key string, hashKey int64, value []byte) {
	hash := getHash(t, client, key)
	require.Len(t, hash, 1)
	connectionIDStr := strconv.FormatInt(hashKey, 10)
	require.Contains(t, hash, connectionIDStr)
	val := hash[connectionIDStr]
	var msg ExpiringValue
	err := proto.Unmarshal([]byte(val), &msg)
	require.NoError(t, err)
	assert.Equal(t, value, msg.Value)
}

func int64ToStr(key int64) string {
	return strconv.FormatInt(key, 10)
}
