package resourcemanager

import (
	"context"
	"fmt"
	"math/rand"
	"sync"
	"testing"
	"time"

	"github.com/tidbcloud/observability/pkg/common/constants"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"

	"github.com/tidbcloud/observability/pkg/common/utils"

	"github.com/stretchr/testify/assert"
)

func Test_configMapSegment_GetSegment_Error(t *testing.T) {
	t.Run("not found in non-observability namespace", func(t *testing.T) {
		// Test configMap not found in a regular namespace - should succeed after creating
		testMgr := configMapSegmentMgr{
			name:       "non-existent-config",
			keyStartId: constants.DataAccountKeyStartID,
			retries:    minRetries,
			kubeClient: fakeClient,
			namespace:  "test-namespace", // Use non-observability namespace
		}
		// This should succeed because it will create the configmap in non-observability namespace
		seg, err := testMgr.GetSegment(context.TODO(), 10)
		assert.NoError(t, err)
		assert.NotNil(t, seg)
		assert.Equal(t, int32(1), seg.StartId()) // Should start from 1
	})

	t.Run("not found in observability namespace", func(t *testing.T) {
		// Test configMap not found in observability namespace (should fail to create)
		testMgr := configMapSegmentMgr{
			name:       "non-existent-config-obs",
			keyStartId: constants.DataAccountKeyStartID,
			retries:    minRetries,
			kubeClient: fakeClient,
			namespace:  constants.ObservabilityNamespace,
		}
		_, err := testMgr.GetSegment(context.TODO(), 10)
		assert.Error(t, err)
		assert.Contains(t, err.Error(), "is reserved")
	})

	t.Run("parse failed", func(t *testing.T) {
		newConfig := map[string]string{
			constants.DataAccountKeyStartID: "abcd",
		}

		_, err := fakeClient.CoreV1().
			ConfigMaps(constants.ObservabilityNamespace).
			Create(context.TODO(), &corev1.ConfigMap{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "testParseFailed",
					Namespace: constants.ObservabilityNamespace,
				},
				Data: newConfig,
			}, metav1.CreateOptions{})

		assert.NoError(t, err)

		testMgr := configMapSegmentMgr{
			name:       "testParseFailed",
			keyStartId: constants.DataAccountKeyStartID,
			retries:    minRetries,
			kubeClient: fakeClient,
			namespace:  constants.ObservabilityNamespace,
		}
		_, err = testMgr.GetSegment(context.TODO(), 10)
		assert.Error(t, err)
		assert.Contains(t, err.Error(), "parse segment failed")
	})

	t.Run("empty data", func(t *testing.T) {
		newConfig := map[string]string{
			constants.DataAccountKeyStartID: "",
		}

		_, err := fakeClient.CoreV1().
			ConfigMaps(constants.ObservabilityNamespace).
			Create(context.TODO(), &corev1.ConfigMap{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "testEmpty",
					Namespace: constants.ObservabilityNamespace,
				},
				Data: newConfig,
			}, metav1.CreateOptions{})
		assert.NoError(t, err)

		testMgr := configMapSegmentMgr{
			name:       "testEmpty",
			keyStartId: constants.DataAccountKeyStartID,
			retries:    minRetries,
			kubeClient: fakeClient,
			namespace:  constants.ObservabilityNamespace,
		}
		seg, err := testMgr.GetSegment(context.TODO(), 10)
		assert.NoError(t, err)
		assert.Equal(t, int32(1), seg.StartId())
	})

	t.Run("missing data key", func(t *testing.T) {
		// Test when the config map exists but doesn't have the required key
		_, err := fakeClient.CoreV1().
			ConfigMaps(constants.ObservabilityNamespace).
			Create(context.TODO(), &corev1.ConfigMap{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "testMissingKey",
					Namespace: constants.ObservabilityNamespace,
				},
				Data: map[string]string{
					"other-key": "value",
				},
			}, metav1.CreateOptions{})
		assert.NoError(t, err)

		testMgr := configMapSegmentMgr{
			name:       "testMissingKey",
			keyStartId: constants.DataAccountKeyStartID,
			retries:    minRetries,
			kubeClient: fakeClient,
			namespace:  constants.ObservabilityNamespace,
		}
		seg, err := testMgr.GetSegment(context.TODO(), 10)
		assert.NoError(t, err)
		assert.Equal(t, int32(1), seg.StartId()) // Should default to 1
	})

	t.Run("conflicted retries", func(t *testing.T) {
		method := isConflicted
		count := 0
		isConflicted = func(err error) bool {
			count++
			return count < 3 // First 2 attempts return conflict
		}
		defer func() {
			isConflicted = method
		}()

		testMgr := configMapSegmentMgr{
			name:       constants.DataAccountSegmentConfigMap,
			keyStartId: constants.DataAccountKeyStartID,
			retries:    minRetries,
			kubeClient: fakeClient,
			namespace:  constants.ObservabilityNamespace,
		}
		segment, err := testMgr.GetSegment(context.TODO(), 10)
		assert.NoError(t, err)
		assert.NotEmpty(t, segment.StartId())
		assert.Equal(t, 3, count) // Should have tried 3 times
	})

	t.Run("timeout due to max retries", func(t *testing.T) {
		method := isConflicted
		isConflicted = func(err error) bool {
			return true // Always return conflict to trigger timeout
		}
		defer func() {
			isConflicted = method
		}()

		testMgr := configMapSegmentMgr{
			name:       constants.DataAccountSegmentConfigMap,
			keyStartId: constants.DataAccountKeyStartID,
			retries:    2, // Set low retry count
			kubeClient: fakeClient,
			namespace:  constants.ObservabilityNamespace,
		}
		_, err := testMgr.GetSegment(context.TODO(), 10)
		assert.Error(t, err)
		assert.Contains(t, err.Error(), "get segment timeout")
	})
}

func Test_configMapSegment_GetSegment(t *testing.T) {

	startIDs := make([]int32, 0)

	for i := 0; i < 100; i++ {
		segment, err := segmentServer.GetSegment(context.TODO(), 10)
		assert.NoError(t, err)
		assert.NotEmpty(t, segment.StartId())
		assert.NotEmpty(t, segment.CurrentId())
		assert.Equal(t, int32(10), segment.Step())
		startIDs = append(startIDs, segment.StartId())
	}

	assert.Equal(t, 100, len(startIDs))
	assert.Equal(t, startIDs[10]+10, startIDs[11])
	fmt.Printf("start ids %v \n", startIDs)
}

func Test_dataAccountReaderWriter_UniqueInt32ForCluster(t *testing.T) {
	for i := 1; i < 5; i++ {
		newRW := dataAccountReaderWriter{
			generator:  utils.NewIdGenerator(utils.NewConfig(int32(i)), segmentServer),
			kubeClient: fakeClient,
		}
		queue := i
		go func() {
			continued := 0
			last := int32(0)
			for j := int32(1); j < 30; j++ {
				value, err := newRW.UniqueInt32ForCluster(context.TODO())
				if err != nil {
					panic(err)
				}
				assert.NoError(t, err)
				assert.NotEmpty(t, value)

				if last == 0 {
					assert.Equal(t, continued, 0)
					continued = 1
				} else if value == last+1 {
					continued++
				} else {
					assert.Equal(t, 0, continued%queue)
					continued = 1
				}
				last = value
				time.Sleep(time.Millisecond * time.Duration(rand.Intn(10)))
			}
		}()
	}

	time.Sleep(time.Second * 2)
}

// Test dataAccountReaderWriterFactory
func Test_dataAccountReaderWriterFactory(t *testing.T) {
	config := utils.NewConfig(100)
	factory := NewdataAccountReaderWriterFactory(fakeClient, config)

	t.Run("get reader writer for same namespace", func(t *testing.T) {
		rw1 := factory.GetReaderWriter("test-ns")
		rw2 := factory.GetReaderWriter("test-ns")

		// Should return the same instance for the same namespace
		assert.Equal(t, rw1, rw2)
	})

	t.Run("get reader writer for different namespaces", func(t *testing.T) {
		rw1 := factory.GetReaderWriter("test-ns-1")
		rw2 := factory.GetReaderWriter("test-ns-2")

		// Should return different instances for different namespaces
		assert.NotEqual(t, rw1, rw2)
	})

	t.Run("concurrent access to factory", func(t *testing.T) {
		var wg sync.WaitGroup
		readers := make([]DataAccountReaderWriter, 10)

		for i := 0; i < 10; i++ {
			wg.Add(1)
			go func(index int) {
				defer wg.Done()
				readers[index] = factory.GetReaderWriter("concurrent-test")
			}(i)
		}

		wg.Wait()

		// All should return the same instance
		for i := 1; i < 10; i++ {
			assert.Equal(t, readers[0], readers[i])
		}
	})
}

// Test configMapSegmentMgr createConfigMapIfNotExists
func Test_configMapSegmentMgr_createConfigMapIfNotExists(t *testing.T) {
	t.Run("create in regular namespace", func(t *testing.T) {
		mgr := configMapSegmentMgr{
			name:       "test-create-config",
			keyStartId: constants.DataAccountKeyStartID,
			retries:    minRetries,
			kubeClient: fakeClient,
			namespace:  "regular-namespace",
		}

		err := mgr.createConfigMapIfNotExists(context.TODO())
		assert.NoError(t, err)

		// Verify the config map was created
		config, err := fakeClient.CoreV1().
			ConfigMaps("regular-namespace").
			Get(context.TODO(), "test-create-config", metav1.GetOptions{})
		assert.NoError(t, err)
		assert.Equal(t, minSegmentStart, config.Data[constants.DataAccountKeyStartID])
	})

	t.Run("create in observability namespace should fail", func(t *testing.T) {
		mgr := configMapSegmentMgr{
			name:       "test-create-config-obs",
			keyStartId: constants.DataAccountKeyStartID,
			retries:    minRetries,
			kubeClient: fakeClient,
			namespace:  constants.ObservabilityNamespace,
		}

		err := mgr.createConfigMapIfNotExists(context.TODO())
		assert.Error(t, err)
		assert.Contains(t, err.Error(), "is reserved")
	})

	t.Run("already exists should not error", func(t *testing.T) {
		// First create the config map
		_, err := fakeClient.CoreV1().
			ConfigMaps("test-exists-ns").
			Create(context.TODO(), &corev1.ConfigMap{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "test-already-exists",
					Namespace: "test-exists-ns",
				},
				Data: map[string]string{
					constants.DataAccountKeyStartID: "100",
				},
			}, metav1.CreateOptions{})
		assert.NoError(t, err)

		mgr := configMapSegmentMgr{
			name:       "test-already-exists",
			keyStartId: constants.DataAccountKeyStartID,
			retries:    minRetries,
			kubeClient: fakeClient,
			namespace:  "test-exists-ns",
		}

		// Should not return error when config map already exists
		err = mgr.createConfigMapIfNotExists(context.TODO())
		assert.NoError(t, err)
	})
}

// Test updateConfigMapIgnoreVersionConflict
func Test_configMapSegmentMgr_updateConfigMapIgnoreVersionConflict(t *testing.T) {
	t.Run("successful update", func(t *testing.T) {
		// Create a config map first
		configMap := &corev1.ConfigMap{
			ObjectMeta: metav1.ObjectMeta{
				Name:      "test-update",
				Namespace: constants.ObservabilityNamespace,
			},
			Data: map[string]string{
				constants.DataAccountKeyStartID: "50",
			},
		}

		_, err := fakeClient.CoreV1().
			ConfigMaps(constants.ObservabilityNamespace).
			Create(context.TODO(), configMap, metav1.CreateOptions{})
		assert.NoError(t, err)

		mgr := configMapSegmentMgr{
			name:       "test-update",
			keyStartId: constants.DataAccountKeyStartID,
			retries:    minRetries,
			kubeClient: fakeClient,
			namespace:  constants.ObservabilityNamespace,
		}

		// Update the config map
		configMap.Data[constants.DataAccountKeyStartID] = "60"
		conflict, err := mgr.updateConfigMapIgnoreVersionConflict(context.TODO(), configMap)
		assert.NoError(t, err)
		assert.False(t, conflict)
	})

	t.Run("version conflict", func(t *testing.T) {
		mgr := configMapSegmentMgr{
			name:       "test-conflict",
			keyStartId: constants.DataAccountKeyStartID,
			retries:    minRetries,
			kubeClient: fakeClient,
			namespace:  constants.ObservabilityNamespace,
		}

		// Mock isConflicted to return true
		originalIsConflicted := isConflicted
		isConflicted = func(err error) bool {
			return true
		}
		defer func() {
			isConflicted = originalIsConflicted
		}()

		configMap := &corev1.ConfigMap{
			ObjectMeta: metav1.ObjectMeta{
				Name:      "test-conflict",
				Namespace: constants.ObservabilityNamespace,
			},
			Data: map[string]string{
				constants.DataAccountKeyStartID: "70",
			},
		}

		conflict, err := mgr.updateConfigMapIgnoreVersionConflict(context.TODO(), configMap)
		assert.NoError(t, err)
		assert.True(t, conflict)
	})
}

// Test edge cases for segment step values
func Test_configMapSegment_GetSegment_EdgeCases(t *testing.T) {
	t.Run("zero step", func(t *testing.T) {
		testMgr := configMapSegmentMgr{
			name:       constants.DataAccountSegmentConfigMap,
			keyStartId: constants.DataAccountKeyStartID,
			retries:    minRetries,
			kubeClient: fakeClient,
			namespace:  constants.ObservabilityNamespace,
		}

		seg, err := testMgr.GetSegment(context.TODO(), 0)
		assert.NoError(t, err)
		assert.Equal(t, int32(0), seg.Step())
	})

	t.Run("negative step", func(t *testing.T) {
		testMgr := configMapSegmentMgr{
			name:       constants.DataAccountSegmentConfigMap,
			keyStartId: constants.DataAccountKeyStartID,
			retries:    minRetries,
			kubeClient: fakeClient,
			namespace:  constants.ObservabilityNamespace,
		}

		seg, err := testMgr.GetSegment(context.TODO(), -10)
		assert.NoError(t, err)
		assert.Equal(t, int32(-10), seg.Step())
	})

	t.Run("large step", func(t *testing.T) {
		testMgr := configMapSegmentMgr{
			name:       constants.DataAccountSegmentConfigMap,
			keyStartId: constants.DataAccountKeyStartID,
			retries:    minRetries,
			kubeClient: fakeClient,
			namespace:  constants.ObservabilityNamespace,
		}

		seg, err := testMgr.GetSegment(context.TODO(), 1000000)
		assert.NoError(t, err)
		assert.Equal(t, int32(1000000), seg.Step())
	})
}

// Test interface implementation
func Test_dataAccountReaderWriter_Interface(t *testing.T) {
	rw := dataAccountReaderWriter{
		generator:  utils.NewIdGenerator(utils.NewConfig(10), segmentServer),
		kubeClient: fakeClient,
	}

	// Test that it implements the interface
	var _ DataAccountReaderWriter = rw

	// Test normal operation
	val, err := rw.UniqueInt32ForCluster(context.TODO())
	assert.NoError(t, err)
	assert.Greater(t, val, int32(0))
}

func Test_dataAccountReaderWriterFactory_DoubleCheckedLocking(t *testing.T) {
	config := utils.NewConfig(100)
	factory := NewdataAccountReaderWriterFactory(fakeClient, config)

	// Test the double-checked locking pattern by simulating concurrent access
	// where the first check might pass but the second check during creation might find existing
	var wg sync.WaitGroup
	const numGoroutines = 20
	results := make([]DataAccountReaderWriter, numGoroutines)

	for i := 0; i < numGoroutines; i++ {
		wg.Add(1)
		go func(index int) {
			defer wg.Done()
			results[index] = factory.GetReaderWriter("double-check-test")
		}(i)
	}

	wg.Wait()

	// Verify all goroutines got the same instance
	for i := 1; i < numGoroutines; i++ {
		assert.Equal(t, results[0], results[i], "All instances should be the same")
	}

	// Verify the factory has only one instance for this namespace
	assert.Equal(t, 1, len(factory.generators))
}