package server

import (
	"context"
	"errors"
	"testing"

	"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"
	dataplanemocks "github.com/tidbcloud/cluster-service-ng/pkg/dataplane/mocks"
	"github.com/tidbcloud/cluster-service-ng/pkg/runtime/actions"
	executormocks "github.com/tidbcloud/cluster-service-ng/pkg/runtime/executor/mocks"
	repomocks "github.com/tidbcloud/cluster-service-ng/pkg/storage/repo/mocks"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/client/fake"
)

func TestDeleteTiKVGroupMarksDeletingAndSubmitsDAG(t *testing.T) {
	ctx := context.Background()
	sharedPool := deleteTiKVGroupTestSharedPool()
	namespace := sharedPool.GetDataPlaneNamespace()
	kubeClient := fake.NewClientBuilder().
		WithScheme(buildPlacementPolicyServerTestScheme(t)).
		WithObjects(deleteTiKVGroupTestCR(namespace)).
		Build()
	sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
	lockRepo := repomocks.NewLockRepo(t)
	dataplaneManager := dataplanemocks.NewManager(t)
	dagExecutor := executormocks.NewRemoteExecutor(t)
	scaleManager := bractionmocks.NewScaleManager(t)

	lockRepo.On("Lock", mock.Anything, mock.MatchedBy(func(lock *entity.Lock) bool {
		return lock.Key == sharedPool.GetId()
	})).Return(func() error { return nil }, nil)
	sharedPoolRepo.EXPECT().Load(ctx, sharedPool.GetId(), false).Return(sharedPool, nil)
	setupSharedPoolDAGMocks(dagExecutor)
	scaleManager.On("AnyFullfillingDecision", mock.Anything, sharedPool.GetId()).Return(false, nil).Maybe()
	dataplaneManager.EXPECT().GetClient(sharedPool.GetDataplaneName()).Return(kubeClient, nil)
	sharedPoolRepo.EXPECT().Save(ctx, mock.MatchedBy(func(sp *entity.SharedPool) bool {
		blue := findComponentByTypeAndId(sp, apiv1.ComponentType_TIKV, "blue")
		return sp.GetPhase() == apiv1.SharedPool_MODIFYING &&
			blue != nil &&
			blue.GetPhase() == apiv1.Component_DELETING &&
			blue.GetNodeQuota().GetNodeCount() == 0 &&
			findComponentByTypeAndId(sp, apiv1.ComponentType_TIKV, "default") != nil
	})).Return(nil)
	dagExecutor.On("Submit", mock.Anything, dag.PlanDeleteTiKVGroup, mock.MatchedBy(func(args actions.ActionArgs) bool {
		sharedPoolID, err := args.GetString(workflow.SharedPoolIdKey)
		if err != nil {
			return false
		}
		componentID, err := args.GetString(workflow.ComponentIdsKey)
		return err == nil && sharedPoolID == sharedPool.GetId() && componentID == "blue"
	}), mock.Anything).Return(nil)

	srv := &serverImpl{
		sharedPoolRepo:   sharedPoolRepo,
		lockRepo:         lockRepo,
		dataplaneManager: dataplaneManager,
		dagExecutor:      dagExecutor,
		scaleManager:     scaleManager,
		transactioner:    &mockCreateTiKVGroupTransactioner{},
	}

	err := srv.DeleteTiKVGroup(ctx, sharedPool.GetId(), "blue")
	require.NoError(t, err)

	stored := &v1alpha1.TiKVGroup{}
	err = kubeClient.Get(ctx, types.NamespacedName{Namespace: namespace, Name: "db-blue"}, stored)
	require.NoError(t, err)
}

func TestDeleteTiKVGroupRejectsDefaultGroup(t *testing.T) {
	ctx := context.Background()
	sharedPool := createTestSharedPool()
	sharedPool.ComponentGroups = append(sharedPool.ComponentGroups, &apiv1.Component{
		Id:            "default",
		ComponentType: apiv1.ComponentType_PD,
		Phase:         apiv1.Component_AVAILABLE,
	})
	sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
	lockRepo := repomocks.NewLockRepo(t)
	dagExecutor := executormocks.NewRemoteExecutor(t)
	scaleManager := bractionmocks.NewScaleManager(t)

	lockRepo.On("Lock", mock.Anything, mock.Anything).Return(func() error { return nil }, nil)
	sharedPoolRepo.EXPECT().Load(ctx, sharedPool.GetId(), false).Return(sharedPool, nil)
	setupSharedPoolDAGMocks(dagExecutor)
	scaleManager.On("AnyFullfillingDecision", mock.Anything, sharedPool.GetId()).Return(false, nil).Maybe()

	srv := &serverImpl{
		sharedPoolRepo: sharedPoolRepo,
		lockRepo:       lockRepo,
		dagExecutor:    dagExecutor,
		scaleManager:   scaleManager,
		transactioner:  &mockCreateTiKVGroupTransactioner{},
	}

	err := srv.DeleteTiKVGroup(ctx, sharedPool.GetId(), "default")
	require.Error(t, err)
	assert.Equal(t, codes.FailedPrecondition, status.Code(err))
	assert.Contains(t, err.Error(), "cannot delete default tikv group")
}

func TestDeleteTiKVGroupRejectsPausingPool(t *testing.T) {
	ctx := context.Background()
	sharedPool := deleteTiKVGroupTestSharedPool()
	sharedPool.PauseResumeState = entity.PauseResumeStatePausing
	sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
	lockRepo := repomocks.NewLockRepo(t)
	dagExecutor := executormocks.NewRemoteExecutor(t)

	lockRepo.On("Lock", mock.Anything, mock.Anything).Return(func() error { return nil }, nil)
	sharedPoolRepo.EXPECT().Load(ctx, sharedPool.GetId(), false).Return(sharedPool, nil)

	srv := &serverImpl{
		sharedPoolRepo: sharedPoolRepo,
		lockRepo:       lockRepo,
		dagExecutor:    dagExecutor,
		transactioner:  &mockCreateTiKVGroupTransactioner{},
	}

	err := srv.DeleteTiKVGroup(ctx, sharedPool.GetId(), "blue")
	require.Error(t, err)
	assert.Equal(t, codes.FailedPrecondition, status.Code(err))
	dagExecutor.AssertNotCalled(t, "Submit", mock.Anything, dag.PlanDeleteTiKVGroup, mock.Anything, mock.Anything)
}

func TestDeleteTiKVGroupRejectsPlacementPolicyReference(t *testing.T) {
	ctx := context.Background()
	sharedPool := deleteTiKVGroupTestSharedPool()
	namespace := sharedPool.GetDataPlaneNamespace()
	kubeClient := fake.NewClientBuilder().
		WithScheme(buildPlacementPolicyServerTestScheme(t)).
		WithObjects(
			deleteTiKVGroupTestCR(namespace),
			deleteTiKVGroupPlacementPolicyRef(namespace, "1001", "blue"),
		).
		Build()
	sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
	lockRepo := repomocks.NewLockRepo(t)
	dataplaneManager := dataplanemocks.NewManager(t)
	dagExecutor := executormocks.NewRemoteExecutor(t)
	scaleManager := bractionmocks.NewScaleManager(t)

	lockRepo.On("Lock", mock.Anything, mock.Anything).Return(func() error { return nil }, nil)
	sharedPoolRepo.EXPECT().Load(ctx, sharedPool.GetId(), false).Return(sharedPool, nil)
	setupSharedPoolDAGMocks(dagExecutor)
	scaleManager.On("AnyFullfillingDecision", mock.Anything, sharedPool.GetId()).Return(false, nil).Maybe()
	dataplaneManager.EXPECT().GetClient(sharedPool.GetDataplaneName()).Return(kubeClient, nil)

	srv := &serverImpl{
		sharedPoolRepo:   sharedPoolRepo,
		lockRepo:         lockRepo,
		dataplaneManager: dataplaneManager,
		dagExecutor:      dagExecutor,
		scaleManager:     scaleManager,
		transactioner:    &mockCreateTiKVGroupTransactioner{},
	}

	err := srv.DeleteTiKVGroup(ctx, sharedPool.GetId(), "blue")
	require.Error(t, err)
	assert.Equal(t, codes.FailedPrecondition, status.Code(err))
	assert.Contains(t, err.Error(), "placement policies reference it: 1001")

	stored := &v1alpha1.TiKVGroup{}
	err = kubeClient.Get(ctx, types.NamespacedName{Namespace: namespace, Name: "db-blue"}, stored)
	require.NoError(t, err)
}

func TestDeleteTiKVGroupDoesNotSubmitDAGWhenTransactionCommitFails(t *testing.T) {
	ctx := context.Background()
	sharedPool := deleteTiKVGroupTestSharedPool()
	namespace := sharedPool.GetDataPlaneNamespace()
	kubeClient := fake.NewClientBuilder().
		WithScheme(buildPlacementPolicyServerTestScheme(t)).
		WithObjects(deleteTiKVGroupTestCR(namespace)).
		Build()
	sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
	lockRepo := repomocks.NewLockRepo(t)
	dataplaneManager := dataplanemocks.NewManager(t)
	dagExecutor := executormocks.NewRemoteExecutor(t)
	scaleManager := bractionmocks.NewScaleManager(t)

	lockRepo.On("Lock", mock.Anything, mock.Anything).Return(func() error { return nil }, nil)
	sharedPoolRepo.EXPECT().Load(ctx, sharedPool.GetId(), false).Return(sharedPool, nil)
	setupSharedPoolDAGMocks(dagExecutor)
	scaleManager.On("AnyFullfillingDecision", mock.Anything, sharedPool.GetId()).Return(false, nil).Maybe()
	dataplaneManager.EXPECT().GetClient(sharedPool.GetDataplaneName()).Return(kubeClient, nil)
	sharedPoolRepo.EXPECT().Save(ctx, mock.Anything).Return(nil)

	srv := &serverImpl{
		sharedPoolRepo:   sharedPoolRepo,
		lockRepo:         lockRepo,
		dataplaneManager: dataplaneManager,
		dagExecutor:      dagExecutor,
		scaleManager:     scaleManager,
		transactioner:    &recordingDeleteTiKVGroupTransactioner{commitErr: errors.New("commit failed")},
	}

	err := srv.DeleteTiKVGroup(ctx, sharedPool.GetId(), "blue")
	require.Error(t, err)
	assert.Contains(t, err.Error(), "commit failed")
	dagExecutor.AssertNotCalled(t, "Submit", mock.Anything, dag.PlanDeleteTiKVGroup, mock.Anything, mock.Anything)
}

func TestDeleteTiKVGroupKeepsLockThroughCommitAndSubmit(t *testing.T) {
	ctx := context.Background()
	sharedPool := deleteTiKVGroupTestSharedPool()
	namespace := sharedPool.GetDataPlaneNamespace()
	kubeClient := fake.NewClientBuilder().
		WithScheme(buildPlacementPolicyServerTestScheme(t)).
		WithObjects(deleteTiKVGroupTestCR(namespace)).
		Build()
	sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
	lockRepo := repomocks.NewLockRepo(t)
	dataplaneManager := dataplanemocks.NewManager(t)
	dagExecutor := executormocks.NewRemoteExecutor(t)
	scaleManager := bractionmocks.NewScaleManager(t)
	events := []string{}

	lockRepo.On("Lock", mock.Anything, mock.Anything).Run(func(args mock.Arguments) {
		events = append(events, "lock")
	}).Return(func() error {
		events = append(events, "unlock")
		return nil
	}, nil)
	sharedPoolRepo.EXPECT().Load(ctx, sharedPool.GetId(), false).Return(sharedPool, nil)
	setupSharedPoolDAGMocks(dagExecutor)
	scaleManager.On("AnyFullfillingDecision", mock.Anything, sharedPool.GetId()).Return(false, nil).Maybe()
	dataplaneManager.EXPECT().GetClient(sharedPool.GetDataplaneName()).Return(kubeClient, nil)
	sharedPoolRepo.EXPECT().Save(ctx, mock.Anything).Run(func(_ context.Context, _ *entity.SharedPool) {
		events = append(events, "save")
	}).Return(nil)
	dagExecutor.On("Submit", mock.Anything, dag.PlanDeleteTiKVGroup, mock.Anything, mock.Anything).Run(func(args mock.Arguments) {
		events = append(events, "submit")
	}).Return(nil)

	srv := &serverImpl{
		sharedPoolRepo:   sharedPoolRepo,
		lockRepo:         lockRepo,
		dataplaneManager: dataplaneManager,
		dagExecutor:      dagExecutor,
		scaleManager:     scaleManager,
		transactioner:    &recordingDeleteTiKVGroupTransactioner{events: &events},
	}

	err := srv.DeleteTiKVGroup(ctx, sharedPool.GetId(), "blue")
	require.NoError(t, err)
	assert.Equal(t, []string{"lock", "save", "commit", "submit", "unlock"}, events)
}

func TestDeleteTiKVGroupRejectsNonAvailableComponent(t *testing.T) {
	ctx := context.Background()
	sharedPool := deleteTiKVGroupTestSharedPool()
	findComponentByTypeAndId(sharedPool, apiv1.ComponentType_TIKV, "blue").Phase = apiv1.Component_CREATING
	sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
	lockRepo := repomocks.NewLockRepo(t)
	dagExecutor := executormocks.NewRemoteExecutor(t)
	scaleManager := bractionmocks.NewScaleManager(t)

	lockRepo.On("Lock", mock.Anything, mock.Anything).Return(func() error { return nil }, nil)
	sharedPoolRepo.EXPECT().Load(ctx, sharedPool.GetId(), false).Return(sharedPool, nil)
	setupSharedPoolDAGMocks(dagExecutor)
	scaleManager.On("AnyFullfillingDecision", mock.Anything, sharedPool.GetId()).Return(false, nil).Maybe()

	srv := &serverImpl{
		sharedPoolRepo: sharedPoolRepo,
		lockRepo:       lockRepo,
		dagExecutor:    dagExecutor,
		scaleManager:   scaleManager,
		transactioner:  &mockCreateTiKVGroupTransactioner{},
	}

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

func deleteTiKVGroupTestSharedPool() *entity.SharedPool {
	sharedPool := createTestSharedPool()
	sharedPool.ComponentGroups = append(sharedPool.ComponentGroups, &apiv1.Component{
		Id:            "blue",
		ComponentType: apiv1.ComponentType_TIKV,
		Phase:         apiv1.Component_AVAILABLE,
		NodeQuota:     &apiv1.NodeQuota{NodeCount: 3},
	})
	return sharedPool
}

func deleteTiKVGroupTestCR(namespace string) *v1alpha1.TiKVGroup {
	return &v1alpha1.TiKVGroup{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: namespace,
			Name:      "db-blue",
		},
	}
}

type recordingDeleteTiKVGroupTransactioner struct {
	events    *[]string
	commitErr error
}

func (t *recordingDeleteTiKVGroupTransactioner) Transaction(ctx context.Context, fn func(context.Context) error) error {
	if err := fn(ctx); err != nil {
		return err
	}
	if t.events != nil {
		*t.events = append(*t.events, "commit")
	}
	return t.commitErr
}

func deleteTiKVGroupPlacementPolicyRef(namespace, placementPolicyID, tikvGroupID string) client.Object {
	return &v1alpha1.PlacementPolicy{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: namespace,
			Name:      placementPolicyID,
			Labels: map[string]string{
				"tags.tidbcloud.com/tikvgroup": "db-" + tikvGroupID,
			},
		},
		Spec: v1alpha1.PlacementPolicySpec{
			GroupRefs: []v1alpha1.PlacementPolicyGroupRef{
				{
					Group: "core.pingcap.com",
					Kind:  "TiKVGroup",
					Name:  "db-" + tikvGroupID,
				},
			},
			Rules: []v1alpha1.PlacementPolicyRule{
				{},
			},
		},
	}
}