package server

import (
	"context"
	"testing"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/require"
	apiv1 "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	bractionmocks "github.com/tidbcloud/cluster-service-ng/domain/workflow/action/br/mocks"
	dataplanemocks "github.com/tidbcloud/cluster-service-ng/pkg/dataplane/mocks"
	"github.com/tidbcloud/cluster-service-ng/pkg/metricsprovider"
	executormocks "github.com/tidbcloud/cluster-service-ng/pkg/runtime/executor/mocks"
	schemav1 "github.com/tidbcloud/cluster-service-ng/pkg/storage/etcd/schema/v1"
	repomocks "github.com/tidbcloud/cluster-service-ng/pkg/storage/repo/mocks"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	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/runtime"
	"k8s.io/apimachinery/pkg/types"
	"sigs.k8s.io/controller-runtime/pkg/client/fake"
)

func TestOpsForThresholdsMemoryObjectMetric(t *testing.T) {
	spec := &autoscalingv2.HorizontalPodAutoscalerSpec{
		Metrics: []autoscalingv2.MetricSpec{
			{
				Type: autoscalingv2.ResourceMetricSourceType,
				Resource: &autoscalingv2.ResourceMetricSource{
					Name: corev1.ResourceCPU,
				},
			},
			{
				Type: autoscalingv2.ObjectMetricSourceType,
				Object: &autoscalingv2.ObjectMetricSource{
					Metric: autoscalingv2.MetricIdentifier{
						Name: metricsprovider.MetricNameExtAvgMemoryUtilization,
					},
					Target: autoscalingv2.MetricTarget{
						Type:  autoscalingv2.ValueMetricType,
						Value: resource.NewQuantity(70, resource.DecimalSI),
					},
				},
			},
		},
	}

	ops, err := opsForThresholds(spec, &apiv1.ScaleOutThresholds{
		MemoryUtilizationPercentage: 80,
	})

	require.NoError(t, err)
	require.Len(t, ops, 1)
	assert.Equal(t, "replace", ops[0].Operation)
	assert.Equal(t, "spec/metrics/1/object/target/value", ops[0].Path)
	assert.Equal(t, uint64(80), ops[0].Value)
}

func TestOpsForThresholdsMemoryResourceMetric(t *testing.T) {
	spec := &autoscalingv2.HorizontalPodAutoscalerSpec{
		Metrics: []autoscalingv2.MetricSpec{
			{
				Type: autoscalingv2.ResourceMetricSourceType,
				Resource: &autoscalingv2.ResourceMetricSource{
					Name: corev1.ResourceMemory,
				},
			},
		},
	}

	ops, err := opsForThresholds(spec, &apiv1.ScaleOutThresholds{
		MemoryUtilizationPercentage: 90,
	})

	require.NoError(t, err)
	require.Len(t, ops, 1)
	assert.Equal(t, "replace", ops[0].Operation)
	assert.Equal(t, "spec/metrics/0/resource/target/averageUtilization", ops[0].Path)
	assert.Equal(t, uint64(90), ops[0].Value)
}

func TestApplyTiDBScaleInThreshold(t *testing.T) {
	hpa := &autoscalingv2.HorizontalPodAutoscaler{}

	changed, err := applyTiDBScaleInThreshold(hpa, apiv1.ComponentType_TIDB, &apiv1.ScaleOutThresholds{
		CpuScaleInUtilizationPercentage: 35,
	})

	require.NoError(t, err)
	require.True(t, changed)
	require.NotNil(t, hpa.Annotations)
	assert.Equal(t, "35", hpa.Annotations[metricsprovider.AnnotationTiDBCPUScaleInThreshold])
	require.NotEmpty(t, hpa.Spec.Metrics)
	require.NotNil(t, hpa.Spec.Metrics[0].Object)
	assert.Equal(t, metricsprovider.MetricNameExtTiDBCPUScaleInGate, hpa.Spec.Metrics[0].Object.Metric.Name)
}

func TestApplyTiDBScaleInThresholdRejectsNonTiDB(t *testing.T) {
	hpa := &autoscalingv2.HorizontalPodAutoscaler{}

	changed, err := applyTiDBScaleInThreshold(hpa, apiv1.ComponentType_TIDB_WORKER, &apiv1.ScaleOutThresholds{
		CpuScaleInUtilizationPercentage: 35,
	})

	require.Error(t, err)
	assert.False(t, changed)
}

func TestNeedsEnableTiDBScaleInScaleDown(t *testing.T) {
	thresholds := &apiv1.ScaleOutThresholds{CpuScaleInUtilizationPercentage: 30}

	t.Run("returns true when scaleDown is disabled", func(t *testing.T) {
		selectPolicy := autoscalingv2.DisabledPolicySelect
		hpa := &autoscalingv2.HorizontalPodAutoscaler{
			Spec: autoscalingv2.HorizontalPodAutoscalerSpec{
				Behavior: &autoscalingv2.HorizontalPodAutoscalerBehavior{
					ScaleDown: &autoscalingv2.HPAScalingRules{SelectPolicy: &selectPolicy},
				},
			},
		}
		assert.True(t, needsEnableTiDBScaleInScaleDown(hpa, apiv1.ComponentType_TIDB, thresholds))
	})

	t.Run("returns false when scaleDown is already enabled", func(t *testing.T) {
		selectPolicy := autoscalingv2.MinChangePolicySelect
		hpa := &autoscalingv2.HorizontalPodAutoscaler{
			Spec: autoscalingv2.HorizontalPodAutoscalerSpec{
				Behavior: &autoscalingv2.HorizontalPodAutoscalerBehavior{
					ScaleDown: &autoscalingv2.HPAScalingRules{SelectPolicy: &selectPolicy},
				},
			},
		}
		assert.False(t, needsEnableTiDBScaleInScaleDown(hpa, apiv1.ComponentType_TIDB, thresholds))
	})

	t.Run("returns false when threshold is not requested", func(t *testing.T) {
		selectPolicy := autoscalingv2.DisabledPolicySelect
		hpa := &autoscalingv2.HorizontalPodAutoscaler{
			Spec: autoscalingv2.HorizontalPodAutoscalerSpec{
				Behavior: &autoscalingv2.HorizontalPodAutoscalerBehavior{
					ScaleDown: &autoscalingv2.HPAScalingRules{SelectPolicy: &selectPolicy},
				},
			},
		}
		assert.False(t, needsEnableTiDBScaleInScaleDown(hpa, apiv1.ComponentType_TIDB, &apiv1.ScaleOutThresholds{}))
	})
}

func TestSetScaleOutConfig_TiDBSameThresholdReEnablesScaleDown(t *testing.T) {
	scheme := runtime.NewScheme()
	require.NoError(t, autoscalingv2.AddToScheme(scheme))

	selectPolicy := autoscalingv2.DisabledPolicySelect
	hpa := &autoscalingv2.HorizontalPodAutoscaler{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "tidbgroup-db-default",
			Namespace: "tidbtestsharedpoolid",
			Labels: map[string]string{
				schemav1.LabelPingcapGroupKey: "db-default",
				schemav1.LabelComponentKey:    "tidb",
			},
			Annotations: map[string]string{
				metricsprovider.AnnotationTiDBCPUScaleInThreshold: "30",
			},
		},
		Spec: autoscalingv2.HorizontalPodAutoscalerSpec{
			MinReplicas: int32Ptr(1),
			MaxReplicas: 3,
			Metrics: []autoscalingv2.MetricSpec{
				{
					Type: autoscalingv2.ResourceMetricSourceType,
					Resource: &autoscalingv2.ResourceMetricSource{
						Name: corev1.ResourceCPU,
						Target: autoscalingv2.MetricTarget{
							Type:               autoscalingv2.UtilizationMetricType,
							AverageUtilization: int32Ptr(70),
						},
					},
				},
			},
			Behavior: &autoscalingv2.HorizontalPodAutoscalerBehavior{
				ScaleDown: &autoscalingv2.HPAScalingRules{
					SelectPolicy: &selectPolicy,
				},
			},
		},
	}
	require.True(t, ensureCPUScaleInGateForHPA(hpa, metricsprovider.TiDBCPUScaleInGateConfig))

	kubeClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(hpa).Build()
	sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
	dataplaneManager := dataplanemocks.NewManager(t)

	sharedPoolRepo.EXPECT().Load(context.Background(), "testsharedpoolid", false).Return(createTestSharedPool(), nil)
	dataplaneManager.EXPECT().GetClient("test-dataplane").Return(kubeClient, nil)

	s := &serverImpl{
		sharedPoolRepo:   sharedPoolRepo,
		dataplaneManager: dataplaneManager,
	}

	err := s.SetScaleOutConfig(context.Background(), "testsharedpoolid", "default", apiv1.ComponentType_TIDB, &apiv1.ScaleOutConfig{
		MinCount: 1,
		MaxCount: 3,
		Thresholds: &apiv1.ScaleOutThresholds{
			CpuScaleInUtilizationPercentage: 30,
		},
	})
	require.NoError(t, err)

	updated := &autoscalingv2.HorizontalPodAutoscaler{}
	err = kubeClient.Get(context.Background(), types.NamespacedName{
		Namespace: "tidbtestsharedpoolid",
		Name:      "tidbgroup-db-default",
	}, updated)
	require.NoError(t, err)

	require.NotNil(t, updated.Spec.Behavior)
	require.NotNil(t, updated.Spec.Behavior.ScaleDown)
	require.NotNil(t, updated.Spec.Behavior.ScaleDown.SelectPolicy)
	assert.Equal(t, autoscalingv2.MinChangePolicySelect, *updated.Spec.Behavior.ScaleDown.SelectPolicy)
	assert.Equal(t, "30", updated.Annotations[metricsprovider.AnnotationTiDBCPUScaleInThreshold])

	foundGate := false
	for _, metric := range updated.Spec.Metrics {
		if metric.Type == autoscalingv2.ObjectMetricSourceType && metric.Object != nil &&
			metric.Object.Metric.Name == metricsprovider.MetricNameExtTiDBCPUScaleInGate {
			foundGate = true
			break
		}
	}
	assert.True(t, foundGate)
}

func TestSetScaleOutConfig_GuardBlocksOnRunningTiKVRemoteWorkerDAG(t *testing.T) {
	for _, runningPlan := range tikvRemoteWorkersDAGPlans() {
		t.Run(runningPlan, func(t *testing.T) {
			sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
			lockRepo := repomocks.NewLockRepo(t)
			dagExecutor := executormocks.NewRemoteExecutor(t)
			scaleManager := bractionmocks.NewScaleManager(t)
			sharedPoolRepo.On("Load", mock.Anything, testTKRWSharedPoolID, false).Return(newTKRWSharedPool(tkrwEnabledBoth(t), 3), nil)
			lockRepo.On("Lock", mock.Anything, mock.Anything).Return(func() error { return nil }, nil)
			setupAllDAGMocksWithConflict(dagExecutor, runningPlan, runningDagStatus(tkrwKey(t, tkrwEnabledBoth(t))))
			scaleManager.On("AnyFullfillingDecision", mock.Anything, mock.Anything).Return(false, nil).Maybe()

			err := (&serverImpl{sharedPoolRepo: sharedPoolRepo, lockRepo: lockRepo, dagExecutor: dagExecutor, scaleManager: scaleManager}).SetScaleOutConfig(context.Background(), testTKRWSharedPoolID, "default", apiv1.ComponentType_TIKV_WORKER, &apiv1.ScaleOutConfig{MinCount: 1, MaxCount: 3})
			require.Error(t, err)
			assert.Equal(t, codes.FailedPrecondition, status.Code(err))
		})
	}
}

func TestSetScaleOutConfig_GuardBlocksOnUnconvergedTiKVRemoteWorkerTopology(t *testing.T) {
	sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
	lockRepo := repomocks.NewLockRepo(t)
	dagExecutor := executormocks.NewRemoteExecutor(t)
	scaleManager := bractionmocks.NewScaleManager(t)

	sharedPool := newTKRWSharedPool(tkrwEnabledBoth(t), 3)
	sharedPoolRepo.On("Load", mock.Anything, testTKRWSharedPoolID, false).Return(sharedPool, nil)
	lockRepo.On("Lock", mock.Anything, mock.Anything).Return(func() error { return nil }, nil)
	setupAllDAGMocks(dagExecutor)
	scaleManager.On("AnyFullfillingDecision", mock.Anything, mock.Anything).Return(false, nil).Maybe()

	server := &serverImpl{
		sharedPoolRepo:   sharedPoolRepo,
		lockRepo:         lockRepo,
		dagExecutor:      dagExecutor,
		scaleManager:     scaleManager,
		dataplaneManager: newTKRWDataplaneManager(t),
	}

	err := server.SetScaleOutConfig(context.Background(), testTKRWSharedPoolID, "default", apiv1.ComponentType_TIKV_WORKER, &apiv1.ScaleOutConfig{
		MinCount: 1,
		MaxCount: 3,
	})
	require.Error(t, err)
	assert.Equal(t, codes.FailedPrecondition, status.Code(err))
}