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/k8s"
	"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"
	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 TestSetHPAScaleBehavior_TiDBScaleDownPersistsGateMetric(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",
			},
		},
		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,
				},
			},
		},
	}

	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,
	}

	enable := true
	err := s.SetHPAScaleBehavior(context.Background(), "testsharedpoolid", "default", apiv1.ComponentType_TIDB, nil, &enable)
	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)

	require.NotNil(t, updated.Annotations)
	assert.Equal(t, "30", updated.Annotations[metricsprovider.AnnotationTiDBCPUScaleInThreshold])

	require.NotEmpty(t, updated.Spec.Metrics)
	require.NotNil(t, updated.Spec.Metrics[0].Object)
	assert.Equal(t, metricsprovider.MetricNameExtTiDBCPUScaleInGate, updated.Spec.Metrics[0].Object.Metric.Name)
	require.NotNil(t, updated.Spec.Metrics[0].Object.Target.Value)
	assert.Equal(t, int64(1), updated.Spec.Metrics[0].Object.Target.Value.Value())
}

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

	selectPolicy := autoscalingv2.DisabledPolicySelect
	hpa := &autoscalingv2.HorizontalPodAutoscaler{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "tiproxygroup-db-cluster-1",
			Namespace: "tidbtestsharedpoolid",
			Labels: map[string]string{
				schemav1.LabelPingcapGroupKey: "db-cluster-1",
				schemav1.LabelComponentKey:    "tiproxy",
			},
		},
		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,
				},
			},
		},
	}

	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,
	}

	enable := true
	err := s.SetHPAScaleBehavior(context.Background(), "testsharedpoolid", "cluster-1", apiv1.ComponentType_TIPROXY, nil, &enable)
	require.NoError(t, err)

	updated := &autoscalingv2.HorizontalPodAutoscaler{}
	err = kubeClient.Get(context.Background(), types.NamespacedName{
		Namespace: "tidbtestsharedpoolid",
		Name:      "tiproxygroup-db-cluster-1",
	}, 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)

	require.NotNil(t, updated.Annotations)
	assert.Equal(t, "30", updated.Annotations[metricsprovider.AnnotationTiProxyCPUScaleInThreshold])

	require.Len(t, updated.Spec.Metrics, 2)
	require.NotNil(t, updated.Spec.Metrics[0].Object)
	assert.Equal(t, metricsprovider.MetricNameExtTiProxyCPUScaleInGate, updated.Spec.Metrics[0].Object.Metric.Name)
	require.NotNil(t, updated.Spec.Metrics[0].Object.Target.Value)
	assert.Equal(t, int64(1), updated.Spec.Metrics[0].Object.Target.Value.Value())
	assert.Equal(t, int32(70), *updated.Spec.Metrics[1].Resource.Target.AverageUtilization)
}

func TestOpsForScaleBehavior_NilSelectPolicyUsesAdd(t *testing.T) {
	selectPolicy := autoscalingv2.DisabledPolicySelect
	spec := &autoscalingv2.HorizontalPodAutoscalerSpec{
		Behavior: &autoscalingv2.HorizontalPodAutoscalerBehavior{
			ScaleUp: &autoscalingv2.HPAScalingRules{
				SelectPolicy: nil,
				Policies: []autoscalingv2.HPAScalingPolicy{
					{Type: autoscalingv2.PodsScalingPolicy, Value: 1, PeriodSeconds: 60},
				},
			},
			ScaleDown: &autoscalingv2.HPAScalingRules{
				SelectPolicy: &selectPolicy,
			},
		},
	}
	enableUp := true

	ops := opsForScaleBehavior(spec, apiv1.ComponentType_TIDB_WORKER, &enableUp, nil)
	require.Len(t, ops, 1)

	assert.Equal(t, "add", ops[0].Operation)
	assert.Equal(t, "/spec/behavior/scaleUp/selectPolicy", ops[0].Path)
	assert.Equal(t, autoscalingv2.MaxChangePolicySelect, ops[0].Value)

	patch, err := k8s.ConstructPatch(ops...)
	require.NoError(t, err)
	require.NotNil(t, patch)
	data, err := patch.Data(nil)
	require.NoError(t, err)
	assert.Contains(t, string(data), `"op":"add"`)
	assert.Contains(t, string(data), `/spec/behavior/scaleUp/selectPolicy`)
}

// TestOpsForScaleBehavior_NilSelectPolicyBothUsesAdd pins the legacy
// write path for SCALE_IN ENABLED on a Disabled-baseline component
// (TIPROXY): the Disabled→MinChange flip from design doc §HPA Behavior
// Assembly Rule must apply here too — see also
// resolve.SelectPolicyFromComponentState (single source of truth shared
// with the reconcile path).
func TestOpsForScaleBehavior_NilSelectPolicyBothUsesAdd(t *testing.T) {
	spec := &autoscalingv2.HorizontalPodAutoscalerSpec{
		Behavior: &autoscalingv2.HorizontalPodAutoscalerBehavior{
			ScaleUp: &autoscalingv2.HPAScalingRules{
				SelectPolicy: nil,
			},
			ScaleDown: &autoscalingv2.HPAScalingRules{
				SelectPolicy: nil,
			},
		},
	}
	enableUp := true
	enableDown := true

	ops := opsForScaleBehavior(spec, apiv1.ComponentType_TIPROXY, &enableUp, &enableDown)
	require.Len(t, ops, 2)

	assert.Equal(t, "add", ops[0].Operation)
	assert.Equal(t, "/spec/behavior/scaleUp/selectPolicy", ops[0].Path)
	assert.Equal(t, autoscalingv2.MaxChangePolicySelect, ops[0].Value)

	assert.Equal(t, "add", ops[1].Operation)
	assert.Equal(t, "/spec/behavior/scaleDown/selectPolicy", ops[1].Path)
	assert.Equal(t, autoscalingv2.MinChangePolicySelect, ops[1].Value,
		"TIPROXY SCALE_IN ENABLED must flip Disabled-baseline to MinChangePolicySelect "+
			"(design doc §HPA Behavior Assembly Rule — Disabled→MinChange flip)")

	patch, err := k8s.ConstructPatch(ops...)
	require.NoError(t, err)
	data, err := patch.Data(nil)
	require.NoError(t, err)
	assert.NotEmpty(t, string(data))
}

func TestOpsForScaleBehavior_NilBehavior(t *testing.T) {
	spec := &autoscalingv2.HorizontalPodAutoscalerSpec{}
	enable := true

	ops := opsForScaleBehavior(spec, apiv1.ComponentType_TIDB_WORKER, nil, &enable)
	require.Len(t, ops, 1)

	assert.Equal(t, "add", ops[0].Operation)
	assert.Equal(t, "/spec/behavior", ops[0].Path)
	behavior, ok := ops[0].Value.(*autoscalingv2.HorizontalPodAutoscalerBehavior)
	require.True(t, ok)
	require.Nil(t, behavior.ScaleUp)
	require.NotNil(t, behavior.ScaleDown)
	require.NotNil(t, behavior.ScaleDown.SelectPolicy)
	assert.Equal(t, autoscalingv2.MaxChangePolicySelect, *behavior.ScaleDown.SelectPolicy)
}

// TestOpsForScaleBehavior_NilScaleUpScaleDown pins the legacy write
// path end-to-end for a brand-new Behavior block on a Disabled-baseline
// component (TIPROXY): both directions go in as `add` ops and SCALE_IN
// ENABLED must flip to MinChangePolicySelect per design doc §HPA
// Behavior Assembly Rule. See resolve.SelectPolicyFromComponentState
// for the single source of truth.
func TestOpsForScaleBehavior_NilScaleUpScaleDown(t *testing.T) {
	spec := &autoscalingv2.HorizontalPodAutoscalerSpec{
		Behavior: &autoscalingv2.HorizontalPodAutoscalerBehavior{},
	}
	enableUp := true
	enableDown := true

	ops := opsForScaleBehavior(spec, apiv1.ComponentType_TIPROXY, &enableUp, &enableDown)
	require.Len(t, ops, 2)

	assert.Equal(t, "add", ops[0].Operation)
	assert.Equal(t, "/spec/behavior/scaleUp", ops[0].Path)
	scaleUp, ok := ops[0].Value.(*autoscalingv2.HPAScalingRules)
	require.True(t, ok)
	require.NotNil(t, scaleUp.SelectPolicy)
	assert.Equal(t, autoscalingv2.MaxChangePolicySelect, *scaleUp.SelectPolicy)

	assert.Equal(t, "add", ops[1].Operation)
	assert.Equal(t, "/spec/behavior/scaleDown", ops[1].Path)
	scaleDown, ok := ops[1].Value.(*autoscalingv2.HPAScalingRules)
	require.True(t, ok)
	require.NotNil(t, scaleDown.SelectPolicy)
	assert.Equal(t, autoscalingv2.MinChangePolicySelect, *scaleDown.SelectPolicy,
		"TIPROXY SCALE_IN ENABLED must flip Disabled-baseline to MinChangePolicySelect "+
			"(design doc §HPA Behavior Assembly Rule — Disabled→MinChange flip)")
}

func TestOpsForScaleBehavior_ExistingSelectPolicy(t *testing.T) {
	originalPolicy := autoscalingv2.DisabledPolicySelect
	spec := &autoscalingv2.HorizontalPodAutoscalerSpec{
		Behavior: &autoscalingv2.HorizontalPodAutoscalerBehavior{
			ScaleDown: &autoscalingv2.HPAScalingRules{
				SelectPolicy: &originalPolicy,
				Policies: []autoscalingv2.HPAScalingPolicy{
					{Type: autoscalingv2.PodsScalingPolicy, Value: 1, PeriodSeconds: 60},
				},
			},
		},
	}
	enable := true

	ops := opsForScaleBehavior(spec, apiv1.ComponentType_TIDB_WORKER, nil, &enable)
	require.Len(t, ops, 1)

	assert.Equal(t, "replace", ops[0].Operation)
	assert.Equal(t, "/spec/behavior/scaleDown/selectPolicy", ops[0].Path)
	assert.Equal(t, autoscalingv2.MaxChangePolicySelect, ops[0].Value)
}

func TestOpsForScaleBehavior_NilBehaviorWithBothScaleUpAndDown(t *testing.T) {
	spec := &autoscalingv2.HorizontalPodAutoscalerSpec{}
	enableUp := true
	enableDown := false

	ops := opsForScaleBehavior(spec, apiv1.ComponentType_TIDB, &enableUp, &enableDown)
	require.Len(t, ops, 1)

	assert.Equal(t, "add", ops[0].Operation)
	assert.Equal(t, "/spec/behavior", ops[0].Path)
	behavior, ok := ops[0].Value.(*autoscalingv2.HorizontalPodAutoscalerBehavior)
	require.True(t, ok)
	require.NotNil(t, behavior.ScaleUp)
	require.NotNil(t, behavior.ScaleUp.SelectPolicy)
	assert.Equal(t, autoscalingv2.MaxChangePolicySelect, *behavior.ScaleUp.SelectPolicy)
	require.NotNil(t, behavior.ScaleDown)
	require.NotNil(t, behavior.ScaleDown.SelectPolicy)
	assert.Equal(t, autoscalingv2.DisabledPolicySelect, *behavior.ScaleDown.SelectPolicy)
}

func TestSetHPAScaleBehavior_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()

			scaleUpEnabled := true
			err := (&serverImpl{sharedPoolRepo: sharedPoolRepo, lockRepo: lockRepo, dagExecutor: dagExecutor, scaleManager: scaleManager}).SetHPAScaleBehavior(context.Background(), testTKRWSharedPoolID, "", apiv1.ComponentType_TIKV, &scaleUpEnabled, nil)
			require.Error(t, err)
			assert.Equal(t, codes.FailedPrecondition, status.Code(err))
		})
	}
}

func TestSetHPAScaleBehavior_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),
	}

	scaleUpEnabled := true
	err := server.SetHPAScaleBehavior(context.Background(), testTKRWSharedPoolID, "", apiv1.ComponentType_TIKV, &scaleUpEnabled, nil)
	require.Error(t, err)
	assert.Equal(t, codes.FailedPrecondition, status.Code(err))
}