package metricsprovider

import (
	"context"
	"fmt"
	"testing"
	"time"

	"github.com/pingcap/tidb-operator/api/v2/core/v1alpha1"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/cluster-service-ng/pkg/k8s"
	autoscalingv1 "k8s.io/api/autoscaling/v1"
	autoscalingv2 "k8s.io/api/autoscaling/v2"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/client-go/kubernetes/scheme"
	scalefake "k8s.io/client-go/scale/fake"
	core "k8s.io/client-go/testing"
	metricsapi "k8s.io/metrics/pkg/apis/metrics/v1beta1"
	metricsfake "k8s.io/metrics/pkg/client/clientset/versioned/fake"
	"k8s.io/utils/ptr"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/client/fake"
	"sigs.k8s.io/custom-metrics-apiserver/pkg/provider"
)

const (
	testNamespace      = "test-namespace"
	testHPAName        = "test-hpa"
	testDeploymentName = "test-deployment"
	testPodNamePrefix  = "test-pod"
)

type extCPUTestCase struct {
	targetCPUUtilization       int32
	specReplicas               int32
	statusReplicas             int32
	reportedCPUMillis          []int64
	cpuRequestsMillis          []int64
	podReadiness               []corev1.ConditionStatus
	podStartTime               []metav1.Time
	podReadyLastTransitionTime []metav1.Time
	podGracefulShutdownBegin   []bool
	podPhase                   []corev1.PodPhase
	podDeletionTimestamp       []bool
	expectedMetricValue        int64
	expectedError              string
}

func (tc *extCPUTestCase) buildTestScheme(t *testing.T) *runtime.Scheme {
	t.Helper()
	s := runtime.NewScheme()
	require.NoError(t, scheme.AddToScheme(s))
	require.NoError(t, autoscalingv2.AddToScheme(s))
	require.NoError(t, v1alpha1.Install(s))
	return s
}

func (tc *extCPUTestCase) buildHPA() *autoscalingv2.HorizontalPodAutoscaler {
	return &autoscalingv2.HorizontalPodAutoscaler{
		ObjectMeta: metav1.ObjectMeta{
			Name:      testHPAName,
			Namespace: testNamespace,
		},
		Spec: autoscalingv2.HorizontalPodAutoscalerSpec{
			ScaleTargetRef: autoscalingv2.CrossVersionObjectReference{
				APIVersion: "apps/v1",
				Kind:       "Deployment",
				Name:       testDeploymentName,
			},
			MinReplicas: ptr.To(int32(1)),
			MaxReplicas: 10,
			Metrics: []autoscalingv2.MetricSpec{
				{
					Type: autoscalingv2.ResourceMetricSourceType,
					Resource: &autoscalingv2.ResourceMetricSource{
						Name: corev1.ResourceCPU,
						Target: autoscalingv2.MetricTarget{
							Type:               autoscalingv2.UtilizationMetricType,
							AverageUtilization: ptr.To(tc.targetCPUUtilization),
						},
					},
				},
			},
		},
	}
}

func (tc *extCPUTestCase) buildPods() []client.Object {
	numPods := len(tc.cpuRequestsMillis)
	if numPods == 0 {
		numPods = int(tc.statusReplicas)
	}

	pods := make([]client.Object, 0, numPods)
	for i := 0; i < numPods; i++ {
		podReadiness := corev1.ConditionTrue
		if tc.podReadiness != nil && i < len(tc.podReadiness) {
			podReadiness = tc.podReadiness[i]
		}

		var podStartTime metav1.Time
		if tc.podStartTime != nil && i < len(tc.podStartTime) {
			podStartTime = tc.podStartTime[i]
		} else {
			podStartTime = metav1.Time{Time: time.Now().Add(-10 * time.Minute)}
		}

		podPhase := corev1.PodRunning
		if tc.podPhase != nil && i < len(tc.podPhase) {
			podPhase = tc.podPhase[i]
		}

		cpuRequest := int64(1000)
		if tc.cpuRequestsMillis != nil && i < len(tc.cpuRequestsMillis) {
			cpuRequest = tc.cpuRequestsMillis[i]
		}

		readyTransition := podStartTime
		if tc.podReadyLastTransitionTime != nil && i < len(tc.podReadyLastTransitionTime) {
			readyTransition = tc.podReadyLastTransitionTime[i]
		}

		pod := &corev1.Pod{
			ObjectMeta: metav1.ObjectMeta{
				Name:      fmt.Sprintf("%s-%d", testPodNamePrefix, i),
				Namespace: testNamespace,
				Labels: map[string]string{
					"app": testDeploymentName,
				},
			},
			Spec: corev1.PodSpec{
				Containers: []corev1.Container{
					{
						Name: "main",
						Resources: corev1.ResourceRequirements{
							Requests: corev1.ResourceList{
								corev1.ResourceCPU: *resource.NewMilliQuantity(cpuRequest, resource.DecimalSI),
							},
						},
					},
				},
			},
			Status: corev1.PodStatus{
				Phase:     podPhase,
				StartTime: &podStartTime,
				Conditions: []corev1.PodCondition{
					{
						Type:               corev1.PodReady,
						Status:             podReadiness,
						LastTransitionTime: readyTransition,
					},
				},
			},
		}

		if tc.podDeletionTimestamp != nil && i < len(tc.podDeletionTimestamp) && tc.podDeletionTimestamp[i] {
			now := metav1.Now()
			pod.DeletionTimestamp = &now
			pod.Finalizers = []string{"test-finalizer"}
		}
		if tc.podGracefulShutdownBegin != nil && i < len(tc.podGracefulShutdownBegin) && tc.podGracefulShutdownBegin[i] {
			if pod.Annotations == nil {
				pod.Annotations = make(map[string]string)
			}
			pod.Annotations[v1alpha1.AnnoKeyTiProxyGracefulShutdownBeginTime] = podStartTime.Format(time.RFC3339)
		}

		pods = append(pods, pod)
	}
	return pods
}

func (tc *extCPUTestCase) buildPodMetrics() *metricsapi.PodMetricsList {
	metrics := &metricsapi.PodMetricsList{}
	for i, cpuMillis := range tc.reportedCPUMillis {
		podMetric := metricsapi.PodMetrics{
			ObjectMeta: metav1.ObjectMeta{
				Name:      fmt.Sprintf("%s-%d", testPodNamePrefix, i),
				Namespace: testNamespace,
				Labels: map[string]string{
					"app": testDeploymentName,
				},
			},
			Timestamp: metav1.Now(),
			Window:    metav1.Duration{Duration: time.Minute},
			Containers: []metricsapi.ContainerMetrics{
				{
					Name: "main",
					Usage: corev1.ResourceList{
						corev1.ResourceCPU: *resource.NewMilliQuantity(cpuMillis, resource.DecimalSI),
					},
				},
			},
		}
		metrics.Items = append(metrics.Items, podMetric)
	}
	return metrics
}

func (tc *extCPUTestCase) buildScale() *autoscalingv1.Scale {
	return &autoscalingv1.Scale{
		ObjectMeta: metav1.ObjectMeta{
			Name:      testDeploymentName,
			Namespace: testNamespace,
		},
		Spec: autoscalingv1.ScaleSpec{
			Replicas: tc.specReplicas,
		},
		Status: autoscalingv1.ScaleStatus{
			Replicas: tc.statusReplicas,
			Selector: "app=" + testDeploymentName,
		},
	}
}

func (tc *extCPUTestCase) setupProvider(t *testing.T) provider.MetricsProvider {
	t.Helper()

	s := tc.buildTestScheme(t)
	objects := tc.buildPods()
	objects = append(objects, tc.buildHPA())

	kubeClient := fake.NewClientBuilder().
		WithScheme(s).
		WithObjects(objects...).
		Build()

	fakeMetricsClientset := &metricsfake.Clientset{}
	fakeMetricsClientset.AddReactor("list", "pods", func(action core.Action) (bool, runtime.Object, error) {
		return true, tc.buildPodMetrics(), nil
	})
	metricsClient := fakeMetricsClientset.MetricsV1beta1()

	fakeScaleClient := &scalefake.FakeScaleClient{}
	fakeScaleClient.AddReactor("get", "deployments", func(action core.Action) (bool, runtime.Object, error) {
		return true, tc.buildScale(), nil
	})

	mapper := k8s.RESTMapper()

	return NewCMProvider(kubeClient, mapper, 0, metricsClient, fakeScaleClient)
}

func (tc *extCPUTestCase) runTest(t *testing.T) {
	t.Helper()

	ctx := context.Background()
	p := tc.setupProvider(t)

	info := provider.CustomMetricInfo{
		GroupResource: schema.GroupResource{
			Group:    "autoscaling",
			Resource: "horizontalpodautoscalers",
		},
		Metric:     MetricNameExtCPUScaledRatio,
		Namespaced: true,
	}

	result, err := p.GetMetricByName(
		ctx,
		types.NamespacedName{Namespace: testNamespace, Name: testHPAName},
		info,
		labels.Everything(),
	)

	if tc.expectedError != "" {
		require.Error(t, err)
		assert.Contains(t, err.Error(), tc.expectedError)
		return
	}

	require.NoError(t, err)
	require.NotNil(t, result)
	assert.Equal(t, tc.expectedMetricValue, result.Value.Value(),
		"metric value should be %d but got %d", tc.expectedMetricValue, result.Value.Value())
}

func hotCPUCreationTime() metav1.Time {
	return metav1.Time{Time: time.Now()}
}

func coolCPUCreationTime() metav1.Time {
	return metav1.Time{Time: time.Now().Add(-10 * time.Minute)}
}

func TestExtCPUScaledRatio_ScaleUp_HighUtilization(t *testing.T) {
	tc := extCPUTestCase{
		targetCPUUtilization: 70,
		specReplicas:         3,
		statusReplicas:       3,
		reportedCPUMillis:    []int64{900, 900, 900},
		cpuRequestsMillis:    []int64{1000, 1000, 1000},
		expectedMetricValue:  100,
	}
	tc.runTest(t)
}

func TestExtCPUScaledRatio_ScaleUp_VeryHighUtilization(t *testing.T) {
	tc := extCPUTestCase{
		targetCPUUtilization: 50,
		specReplicas:         2,
		statusReplicas:       2,
		reportedCPUMillis:    []int64{1000, 1000},
		cpuRequestsMillis:    []int64{1000, 1000},
		expectedMetricValue:  100,
	}
	tc.runTest(t)
}

func TestExtCPUScaledRatio_ScaleUp_MixedUtilization(t *testing.T) {
	tc := extCPUTestCase{
		targetCPUUtilization: 50,
		specReplicas:         3,
		statusReplicas:       3,
		reportedCPUMillis:    []int64{300, 500, 700},
		cpuRequestsMillis:    []int64{1000, 1000, 1000},
		expectedMetricValue:  1,
	}
	tc.runTest(t)
}

func TestExtCPUScaledRatio_ScaleDown_LowUtilization(t *testing.T) {
	tc := extCPUTestCase{
		targetCPUUtilization: 70,
		specReplicas:         5,
		statusReplicas:       5,
		reportedCPUMillis:    []int64{300, 300, 300, 300, 300},
		cpuRequestsMillis:    []int64{1000, 1000, 1000, 1000, 1000},
		expectedMetricValue:  0,
	}
	tc.runTest(t)
}

func TestExtCPUScaledRatio_ScaleDown_VeryLowUtilization(t *testing.T) {
	tc := extCPUTestCase{
		targetCPUUtilization: 70,
		specReplicas:         3,
		statusReplicas:       3,
		reportedCPUMillis:    []int64{100, 100, 100},
		cpuRequestsMillis:    []int64{1000, 1000, 1000},
		expectedMetricValue:  0,
	}
	tc.runTest(t)
}

func TestExtCPUScaledRatio_NoScale_WithinTolerance(t *testing.T) {
	tc := extCPUTestCase{
		targetCPUUtilization: 70,
		specReplicas:         3,
		statusReplicas:       3,
		reportedCPUMillis:    []int64{720, 720, 720},
		cpuRequestsMillis:    []int64{1000, 1000, 1000},
		expectedMetricValue:  1,
	}
	tc.runTest(t)
}

func TestExtCPUScaledRatio_NoScale_ExactlyAtTarget(t *testing.T) {
	tc := extCPUTestCase{
		targetCPUUtilization: 70,
		specReplicas:         3,
		statusReplicas:       3,
		reportedCPUMillis:    []int64{700, 700, 700},
		cpuRequestsMillis:    []int64{1000, 1000, 1000},
		expectedMetricValue:  1,
	}
	tc.runTest(t)
}

func TestExtCPUScaledRatio_NoScale_SlightlyBelowTarget(t *testing.T) {
	tc := extCPUTestCase{
		targetCPUUtilization: 70,
		specReplicas:         3,
		statusReplicas:       3,
		reportedCPUMillis:    []int64{650, 650, 650},
		cpuRequestsMillis:    []int64{1000, 1000, 1000},
		expectedMetricValue:  1,
	}
	tc.runTest(t)
}

// TestExtCPUScaledRatio_UnreadyPodsIgnored tests that unready pods are handled correctly.
// When there are unready pods AND usageRatio > 1.0, the implementation adds unready pods
// back with 0% utilization (rebalanceIgnored path), which can change the scale decision.
// In this test:
// - 3 pods, pod-0 is unready with 100m CPU, pods 1,2 are ready with 800m each
// - After removing unready pod: usageRatio = (1600/2000) / 0.5 = 1.6 > 1.0
// - rebalanceIgnored = true, so pod-0 is added back with 0 value
// - New usageRatio = (1600/3000) / 0.5 = 1.067, which is within 10% tolerance
// - Result: returns 1 (no scale)
func TestExtCPUScaledRatio_UnreadyPodsIgnored(t *testing.T) {
	tc := extCPUTestCase{
		targetCPUUtilization: 50,
		specReplicas:         3,
		statusReplicas:       3,
		reportedCPUMillis:    []int64{100, 800, 800},
		cpuRequestsMillis:    []int64{1000, 1000, 1000},
		podReadiness:         []corev1.ConditionStatus{corev1.ConditionFalse, corev1.ConditionTrue, corev1.ConditionTrue},
		podStartTime:         []metav1.Time{coolCPUCreationTime(), coolCPUCreationTime(), coolCPUCreationTime()},
		expectedMetricValue:  1, // Within tolerance after rebalance
	}
	tc.runTest(t)
}

func TestExtCPUScaledRatio_HotCPUPodsIgnored(t *testing.T) {
	tc := extCPUTestCase{
		targetCPUUtilization: 70,
		specReplicas:         3,
		statusReplicas:       3,
		reportedCPUMillis:    []int64{900, 700, 700},
		cpuRequestsMillis:    []int64{1000, 1000, 1000},
		podStartTime:         []metav1.Time{hotCPUCreationTime(), coolCPUCreationTime(), coolCPUCreationTime()},
		expectedMetricValue:  1,
	}
	tc.runTest(t)
}

func TestExtCPUScaledRatio_FailedPodsIgnored(t *testing.T) {
	tc := extCPUTestCase{
		targetCPUUtilization: 70,
		specReplicas:         3,
		statusReplicas:       3,
		reportedCPUMillis:    []int64{900, 900},
		cpuRequestsMillis:    []int64{1000, 1000, 1000},
		podPhase:             []corev1.PodPhase{corev1.PodRunning, corev1.PodRunning, corev1.PodFailed},
		expectedMetricValue:  100,
	}
	tc.runTest(t)
}

func TestExtCPUScaledRatio_DeletingPodsIgnored(t *testing.T) {
	tc := extCPUTestCase{
		targetCPUUtilization: 70,
		specReplicas:         3,
		statusReplicas:       3,
		reportedCPUMillis:    []int64{900, 900},
		cpuRequestsMillis:    []int64{1000, 1000, 1000},
		podDeletionTimestamp: []bool{false, false, true},
		expectedMetricValue:  100,
	}
	tc.runTest(t)
}

// TestExtCPUScaledRatio_PendingPodsIgnored tests pending pod handling.
// Pending pods are treated as unready. When rebalanceIgnored path is triggered
// (unready pods exist AND usageRatio > 1.0), adding pending pods with 0 value
// causes scale direction change, resulting in no-scale decision.
func TestExtCPUScaledRatio_PendingPodsIgnored(t *testing.T) {
	tc := extCPUTestCase{
		targetCPUUtilization: 70,
		specReplicas:         3,
		statusReplicas:       3,
		reportedCPUMillis:    []int64{900, 900},
		cpuRequestsMillis:    []int64{1000, 1000, 1000},
		podPhase:             []corev1.PodPhase{corev1.PodRunning, corev1.PodRunning, corev1.PodPending},
		expectedMetricValue:  1, // Scale direction change after rebalance
	}
	tc.runTest(t)
}

// TestExtCPUScaledRatio_Error_AllPodsUnready tests the edge case where all pods are unready.
// When all pods are unready, no metrics remain after filtering, which should return an error
// at provider.go:424-426 ("did not receive metrics for any ready pods").
func TestExtCPUScaledRatio_Error_AllPodsUnready(t *testing.T) {
	tc := extCPUTestCase{
		targetCPUUtilization: 70,
		specReplicas:         3,
		statusReplicas:       3,
		reportedCPUMillis:    []int64{700, 700, 700},
		cpuRequestsMillis:    []int64{1000, 1000, 1000},
		podReadiness:         []corev1.ConditionStatus{corev1.ConditionFalse, corev1.ConditionFalse, corev1.ConditionFalse},
		podStartTime:         []metav1.Time{coolCPUCreationTime(), coolCPUCreationTime(), coolCPUCreationTime()},
		expectedError:        "did not receive metrics for any ready pods",
	}
	tc.runTest(t)
}

func TestExtCPUScaledRatio_Error_HPANotFound(t *testing.T) {
	ctx := context.Background()

	s := runtime.NewScheme()
	require.NoError(t, scheme.AddToScheme(s))
	require.NoError(t, autoscalingv2.AddToScheme(s))

	kubeClient := fake.NewClientBuilder().WithScheme(s).Build()
	fakeMetricsClientset := &metricsfake.Clientset{}
	fakeScaleClient := &scalefake.FakeScaleClient{}
	mapper := k8s.RESTMapper()

	p := NewCMProvider(kubeClient, mapper, 0, fakeMetricsClientset.MetricsV1beta1(), fakeScaleClient)

	info := provider.CustomMetricInfo{
		GroupResource: schema.GroupResource{Group: "autoscaling", Resource: "horizontalpodautoscalers"},
		Metric:        MetricNameExtCPUScaledRatio,
		Namespaced:    true,
	}

	_, err := p.GetMetricByName(
		ctx,
		types.NamespacedName{Namespace: testNamespace, Name: "nonexistent-hpa"},
		info,
		labels.Everything(),
	)

	require.Error(t, err)
	assert.Contains(t, err.Error(), "not found")
}

func TestExtCPUScaledRatio_Error_InvalidResourceType(t *testing.T) {
	tc := extCPUTestCase{
		targetCPUUtilization: 70,
		specReplicas:         3,
		statusReplicas:       3,
		reportedCPUMillis:    []int64{700, 700, 700},
		cpuRequestsMillis:    []int64{1000, 1000, 1000},
	}

	ctx := context.Background()
	p := tc.setupProvider(t)

	info := provider.CustomMetricInfo{
		GroupResource: schema.GroupResource{Group: "core.pingcap.com", Resource: "tidbgroups"},
		Metric:        MetricNameExtCPUScaledRatio,
		Namespaced:    true,
	}

	_, err := p.GetMetricByName(
		ctx,
		types.NamespacedName{Namespace: testNamespace, Name: testHPAName},
		info,
		labels.Everything(),
	)

	require.Error(t, err)
	assert.Contains(t, err.Error(), "not supported")
}

func TestExtCPUScaledRatio_Error_NoCPUMetricInHPA(t *testing.T) {
	ctx := context.Background()

	hpa := &autoscalingv2.HorizontalPodAutoscaler{
		ObjectMeta: metav1.ObjectMeta{
			Name:      testHPAName,
			Namespace: testNamespace,
		},
		Spec: autoscalingv2.HorizontalPodAutoscalerSpec{
			ScaleTargetRef: autoscalingv2.CrossVersionObjectReference{
				APIVersion: "apps/v1",
				Kind:       "Deployment",
				Name:       testDeploymentName,
			},
			MinReplicas: ptr.To(int32(1)),
			MaxReplicas: 10,
			Metrics: []autoscalingv2.MetricSpec{
				{
					Type: autoscalingv2.ResourceMetricSourceType,
					Resource: &autoscalingv2.ResourceMetricSource{
						Name: corev1.ResourceMemory,
						Target: autoscalingv2.MetricTarget{
							Type:               autoscalingv2.UtilizationMetricType,
							AverageUtilization: ptr.To(int32(70)),
						},
					},
				},
			},
		},
	}

	s := runtime.NewScheme()
	require.NoError(t, scheme.AddToScheme(s))
	require.NoError(t, autoscalingv2.AddToScheme(s))

	kubeClient := fake.NewClientBuilder().WithScheme(s).WithObjects(hpa).Build()
	fakeMetricsClientset := &metricsfake.Clientset{}
	fakeScaleClient := &scalefake.FakeScaleClient{}
	mapper := k8s.RESTMapper()

	p := NewCMProvider(kubeClient, mapper, 0, fakeMetricsClientset.MetricsV1beta1(), fakeScaleClient)

	info := provider.CustomMetricInfo{
		GroupResource: schema.GroupResource{Group: "autoscaling", Resource: "horizontalpodautoscalers"},
		Metric:        MetricNameExtCPUScaledRatio,
		Namespaced:    true,
	}

	_, err := p.GetMetricByName(
		ctx,
		types.NamespacedName{Namespace: testNamespace, Name: testHPAName},
		info,
		labels.Everything(),
	)

	require.Error(t, err)
	assert.Contains(t, err.Error(), "cpu")
}

func TestExtCPUScaledRatio_Error_NoMetricsReturned(t *testing.T) {
	ctx := context.Background()

	tc := extCPUTestCase{
		targetCPUUtilization: 70,
		specReplicas:         3,
		statusReplicas:       3,
		cpuRequestsMillis:    []int64{1000, 1000, 1000},
		reportedCPUMillis:    []int64{},
	}

	s := tc.buildTestScheme(t)
	objects := tc.buildPods()
	objects = append(objects, tc.buildHPA())

	kubeClient := fake.NewClientBuilder().
		WithScheme(s).
		WithObjects(objects...).
		Build()

	fakeMetricsClientset := &metricsfake.Clientset{}
	fakeMetricsClientset.AddReactor("list", "pods", func(action core.Action) (bool, runtime.Object, error) {
		return true, &metricsapi.PodMetricsList{}, nil
	})

	fakeScaleClient := &scalefake.FakeScaleClient{}
	fakeScaleClient.AddReactor("get", "deployments", func(action core.Action) (bool, runtime.Object, error) {
		return true, tc.buildScale(), nil
	})

	mapper := k8s.RESTMapper()
	p := NewCMProvider(kubeClient, mapper, 0, fakeMetricsClientset.MetricsV1beta1(), fakeScaleClient)

	info := provider.CustomMetricInfo{
		GroupResource: schema.GroupResource{Group: "autoscaling", Resource: "horizontalpodautoscalers"},
		Metric:        MetricNameExtCPUScaledRatio,
		Namespaced:    true,
	}

	_, err := p.GetMetricByName(
		ctx,
		types.NamespacedName{Namespace: testNamespace, Name: testHPAName},
		info,
		labels.Everything(),
	)

	require.Error(t, err)
	assert.Contains(t, err.Error(), "no metrics")
}