package server

import (
	"context"
	"fmt"
	"testing"

	"github.com/pingcap/kvproto/pkg/keyspacepb"
	"github.com/pingcap/tidb-operator/api/v2/core/v1alpha1"
	"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"
	"github.com/tidbcloud/cluster-service-ng/domain/entity"
	"github.com/tidbcloud/cluster-service-ng/domain/workflow"
	bractionmocks "github.com/tidbcloud/cluster-service-ng/domain/workflow/action/br/mocks"
	"github.com/tidbcloud/cluster-service-ng/domain/workflow/dag"
	"github.com/tidbcloud/cluster-service-ng/pkg/clustergateway"
	clustergatewaymocks "github.com/tidbcloud/cluster-service-ng/pkg/clustergateway/mocks"
	dataplanemocks "github.com/tidbcloud/cluster-service-ng/pkg/dataplane/mocks"
	"github.com/tidbcloud/cluster-service-ng/pkg/runtime/actions"
	"github.com/tidbcloud/cluster-service-ng/pkg/runtime/executor"
	executormocks "github.com/tidbcloud/cluster-service-ng/pkg/runtime/executor/mocks"
	repomocks "github.com/tidbcloud/cluster-service-ng/pkg/storage/repo/mocks"
	pdapiv2 "github.com/tikv/pd/server/apiv2/handlers"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/encoding/protojson"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/types"
	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
	"sigs.k8s.io/controller-runtime/pkg/client/fake"
)

func TestCreatePlacementPolicyResolvesKeyspaceAndWritesOperatorCR(t *testing.T) {
	ctx := context.Background()
	sharedPool := createTestSharedPool()
	scheme := buildPlacementPolicyServerTestScheme(t)
	kubeClient := fake.NewClientBuilder().WithScheme(scheme).Build()
	srv, deps := newPlacementPolicyWriteTestServer(t, ctx, sharedPool)

	deps.clusterGateway.EXPECT().GetKeySpace(ctx, sharedPool, "1001").Return(&pdapiv2.KeyspaceMeta{
		KeyspaceMeta: &keyspacepb.KeyspaceMeta{
			Id:   42,
			Name: "1001",
		},
	}, nil)
	deps.dataplaneManager.EXPECT().GetClient(sharedPool.GetDataplaneName()).Return(kubeClient, nil)

	got, err := srv.CreatePlacementPolicy(ctx, sharedPool.GetId(), "1001", "default")
	require.NoError(t, err)
	require.NotNil(t, got)
	assert.Equal(t, "1001", got.GetPlacementPolicyId())
	assert.Equal(t, "1001", got.GetClusterId())
	assert.Equal(t, "default", got.GetTikvGroupId())

	cr := &v1alpha1.PlacementPolicy{}
	err = kubeClient.Get(ctx, types.NamespacedName{
		Namespace: "tidb" + sharedPool.GetId(),
		Name:      "1001",
	}, cr)
	require.NoError(t, err)
	require.Len(t, cr.Spec.GroupRefs, 1)
	assert.Equal(t, "db-default", cr.Spec.GroupRefs[0].Name)
	require.Len(t, cr.Spec.Rules, 1)
	assert.Equal(t, v1alpha1.PlacementPolicyRoleVoter, cr.Spec.Rules[0].Role)
	assert.Equal(t, int32(3), cr.Spec.Rules[0].Count)
	assert.Equal(t, []string{"42"}, cr.Spec.Rules[0].Selector.Keyspace.IDs)
}

func TestCreatePlacementPolicyRejectsExistingPolicy(t *testing.T) {
	ctx := context.Background()
	sharedPool := createTestSharedPool()
	scheme := buildPlacementPolicyServerTestScheme(t)
	kubeClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(&v1alpha1.PlacementPolicy{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: "tidb" + sharedPool.GetId(),
			Name:      "1001",
		},
	}).Build()
	srv, deps := newPlacementPolicyWriteTestServer(t, ctx, sharedPool)

	deps.clusterGateway.EXPECT().GetKeySpace(ctx, sharedPool, "1001").Return(&pdapiv2.KeyspaceMeta{
		KeyspaceMeta: &keyspacepb.KeyspaceMeta{
			Id:   42,
			Name: "1001",
		},
	}, nil)
	deps.dataplaneManager.EXPECT().GetClient(sharedPool.GetDataplaneName()).Return(kubeClient, nil)

	_, err := srv.CreatePlacementPolicy(ctx, sharedPool.GetId(), "1001", "default")
	require.Error(t, err)
	assert.Equal(t, codes.AlreadyExists, status.Code(err))
}

func TestUpdatePlacementPolicyResolvesKeyspaceAndWritesOperatorCR(t *testing.T) {
	ctx := context.Background()
	sharedPool := createTestSharedPool()
	scheme := buildPlacementPolicyServerTestScheme(t)
	kubeClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(&v1alpha1.PlacementPolicy{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: "tidb" + sharedPool.GetId(),
			Name:      "1001",
		},
	}).Build()
	srv, deps := newPlacementPolicyWriteTestServer(t, ctx, sharedPool)

	deps.clusterGateway.EXPECT().GetKeySpace(ctx, sharedPool, "1001").Return(&pdapiv2.KeyspaceMeta{
		KeyspaceMeta: &keyspacepb.KeyspaceMeta{
			Id:   42,
			Name: "1001",
		},
	}, nil)
	deps.dataplaneManager.EXPECT().GetClient(sharedPool.GetDataplaneName()).Return(kubeClient, nil)

	got, err := srv.UpdatePlacementPolicy(ctx, sharedPool.GetId(), "1001", "default")
	require.NoError(t, err)
	require.NotNil(t, got)
	assert.Equal(t, "1001", got.GetPlacementPolicyId())
	assert.Equal(t, "1001", got.GetClusterId())
	assert.Equal(t, "default", got.GetTikvGroupId())

	cr := &v1alpha1.PlacementPolicy{}
	err = kubeClient.Get(ctx, types.NamespacedName{
		Namespace: "tidb" + sharedPool.GetId(),
		Name:      "1001",
	}, cr)
	require.NoError(t, err)
	require.Len(t, cr.Spec.GroupRefs, 1)
	assert.Equal(t, "db-default", cr.Spec.GroupRefs[0].Name)
	require.Len(t, cr.Spec.Rules, 1)
	assert.Equal(t, v1alpha1.PlacementPolicyRoleVoter, cr.Spec.Rules[0].Role)
	assert.Equal(t, int32(3), cr.Spec.Rules[0].Count)
	assert.Equal(t, []string{"42"}, cr.Spec.Rules[0].Selector.Keyspace.IDs)
}

func TestUpdatePlacementPolicyReturnsNotFoundForMissingPolicy(t *testing.T) {
	ctx := context.Background()
	sharedPool := createTestSharedPool()
	scheme := buildPlacementPolicyServerTestScheme(t)
	kubeClient := fake.NewClientBuilder().WithScheme(scheme).Build()
	srv, deps := newPlacementPolicyWriteTestServer(t, ctx, sharedPool)

	deps.clusterGateway.EXPECT().GetKeySpace(ctx, sharedPool, "1001").Return(&pdapiv2.KeyspaceMeta{
		KeyspaceMeta: &keyspacepb.KeyspaceMeta{
			Id:   42,
			Name: "1001",
		},
	}, nil)
	deps.dataplaneManager.EXPECT().GetClient(sharedPool.GetDataplaneName()).Return(kubeClient, nil)

	_, err := srv.UpdatePlacementPolicy(ctx, sharedPool.GetId(), "1001", "default")
	require.Error(t, err)
	assert.Equal(t, codes.NotFound, status.Code(err))
}

func TestUpdatePlacementPolicyRejectsMissingTiKVGroup(t *testing.T) {
	ctx := context.Background()
	sharedPool := createTestSharedPool()
	srv, _ := newPlacementPolicyWriteTestServer(t, ctx, sharedPool)

	_, err := srv.UpdatePlacementPolicy(ctx, sharedPool.GetId(), "1001", "missing")
	require.Error(t, err)
	assert.Equal(t, codes.FailedPrecondition, status.Code(err))
}

func TestCreatePlacementPolicyReturnsFailedPreconditionForMissingCluster(t *testing.T) {
	ctx := context.Background()
	sharedPool := createTestSharedPool()
	srv, deps := newPlacementPolicyWriteTestServer(t, ctx, sharedPool)

	deps.clusterGateway.EXPECT().GetKeySpace(ctx, sharedPool, "1001").Return(nil,
		fmt.Errorf("get keyspace 1001: %w", clustergateway.ErrKeyspaceNotFound),
	)

	_, err := srv.CreatePlacementPolicy(ctx, sharedPool.GetId(), "1001", "default")
	require.Error(t, err)
	assert.Equal(t, codes.FailedPrecondition, status.Code(err))
}

func TestCreatePlacementPolicyRejectsDeletingTiKVGroup(t *testing.T) {
	ctx := context.Background()
	sharedPool := createTestSharedPool()
	tikvComp := findComponentByTypeAndId(sharedPool, apiv1.ComponentType_TIKV, "default")
	require.NotNil(t, tikvComp)
	tikvComp.Phase = apiv1.Component_DELETING
	srv, _ := newPlacementPolicyWriteTestServer(t, ctx, sharedPool)

	_, err := srv.CreatePlacementPolicy(ctx, sharedPool.GetId(), "1001", "default")
	require.Error(t, err)
	assert.Equal(t, codes.FailedPrecondition, status.Code(err))
	assert.Contains(t, err.Error(), "tikv group default is not available")
}

func TestUpdatePlacementPolicyRejectsDeletingTiKVGroup(t *testing.T) {
	ctx := context.Background()
	sharedPool := createTestSharedPool()
	tikvComp := findComponentByTypeAndId(sharedPool, apiv1.ComponentType_TIKV, "default")
	require.NotNil(t, tikvComp)
	tikvComp.Phase = apiv1.Component_DELETING
	srv, _ := newPlacementPolicyWriteTestServer(t, ctx, sharedPool)

	_, err := srv.UpdatePlacementPolicy(ctx, sharedPool.GetId(), "1001", "default")
	require.Error(t, err)
	assert.Equal(t, codes.FailedPrecondition, status.Code(err))
	assert.Contains(t, err.Error(), "tikv group default is not available")
}

func TestCreatePlacementPolicyRejectsTiKVGroupDeletingInRunningDAG(t *testing.T) {
	ctx := context.Background()
	sharedPool := createTestSharedPool()
	deletingSharedPool := createTestSharedPool()
	tikvComp := findComponentByTypeAndId(deletingSharedPool, apiv1.ComponentType_TIKV, "default")
	require.NotNil(t, tikvComp)
	tikvComp.Phase = apiv1.Component_DELETING
	srv, _ := newPlacementPolicyWriteTestServerWithDAGStatus(t, ctx, sharedPool, map[string]executor.ExecTaskIdStatus{
		dag.PlanDeleteTiKVGroup: placementPolicyRunningSharedPoolStatus(t, deletingSharedPool),
	})

	_, err := srv.CreatePlacementPolicy(ctx, sharedPool.GetId(), "1001", "default")
	require.Error(t, err)
	assert.Equal(t, codes.FailedPrecondition, status.Code(err))
	assert.Contains(t, err.Error(), "tikv group default is not available")
}

type placementPolicyWriteTestDeps struct {
	sharedPoolRepo   *repomocks.SharedPoolRepo
	lockRepo         *repomocks.LockRepo
	dataplaneManager *dataplanemocks.Manager
	clusterGateway   *clustergatewaymocks.Client
	dagExecutor      *executormocks.RemoteExecutor
	scaleManager     *bractionmocks.ScaleManager
}

func newPlacementPolicyWriteTestServer(t *testing.T, ctx context.Context, sharedPool *entity.SharedPool) (*serverImpl, placementPolicyWriteTestDeps) {
	t.Helper()

	return newPlacementPolicyWriteTestServerWithDAGStatus(t, ctx, sharedPool, nil)
}

func newPlacementPolicyWriteTestServerWithDAGStatus(t *testing.T, ctx context.Context, sharedPool *entity.SharedPool, runningDags map[string]executor.ExecTaskIdStatus) (*serverImpl, placementPolicyWriteTestDeps) {
	t.Helper()

	deps := placementPolicyWriteTestDeps{
		sharedPoolRepo:   repomocks.NewSharedPoolRepo(t),
		lockRepo:         repomocks.NewLockRepo(t),
		dataplaneManager: dataplanemocks.NewManager(t),
		clusterGateway:   clustergatewaymocks.NewClient(t),
		dagExecutor:      executormocks.NewRemoteExecutor(t),
		scaleManager:     bractionmocks.NewScaleManager(t),
	}

	deps.lockRepo.On("Lock", mock.Anything, mock.MatchedBy(func(lock *entity.Lock) bool {
		return lock.Key == sharedPool.GetId()
	})).Return(func() error { return nil }, nil).Once()
	deps.sharedPoolRepo.EXPECT().Load(ctx, sharedPool.GetId(), false).Return(sharedPool, nil)
	setupPlacementPolicyDAGMocks(deps.dagExecutor, runningDags)
	deps.scaleManager.On("AnyFullfillingDecision", mock.Anything, sharedPool.GetId()).Return(false, nil).Maybe()

	return &serverImpl{
		sharedPoolRepo:    deps.sharedPoolRepo,
		lockRepo:          deps.lockRepo,
		dagExecutor:       deps.dagExecutor,
		scaleManager:      deps.scaleManager,
		dataplaneManager:  deps.dataplaneManager,
		clustergatewayCli: deps.clusterGateway,
	}, deps
}

func setupPlacementPolicyDAGMocks(dagExecutor *executormocks.RemoteExecutor, runningDags map[string]executor.ExecTaskIdStatus) {
	dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(
		func(_ context.Context, planName string, _ executor.MatchFilter) executor.ExecTaskIdStatus {
			if runningDags != nil {
				if dagStatus, ok := runningDags[planName]; ok {
					return dagStatus
				}
			}
			return executor.NewExecTaskIdStatus()
		},
		nil,
	)
}

func placementPolicyRunningSharedPoolStatus(t *testing.T, sharedPool *entity.SharedPool) executor.ExecTaskIdStatus {
	t.Helper()

	sharedPoolBytes, err := protojson.Marshal(sharedPool)
	require.NoError(t, err)
	return executor.ExecTaskIdStatus{
		"running-task": &executor.ExecTaskStatusEx{
			ExecTaskStatus: executor.ExecTaskStatus{
				ExecTask: &executor.ExecutorTask{
					Args: actions.ActionArgs{
						workflow.SharedPoolKey: actions.ActionArgString(string(sharedPoolBytes)),
					},
				},
			},
		},
	}
}

func buildPlacementPolicyServerTestScheme(t *testing.T) *runtime.Scheme {
	t.Helper()

	scheme := runtime.NewScheme()
	require.NoError(t, clientgoscheme.AddToScheme(scheme))
	require.NoError(t, v1alpha1.AddToScheme(scheme))
	return scheme
}