package cluster

import (
	"testing"

	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/fastflow/pkg/entity"
	"go.uber.org/mock/gomock"

	globaldedicatedv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/dedicated/v1"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/service/workflow/common"
	"github.com/tidbcloud/tidb-management-service/utilities/bizerr"
	"github.com/tidbcloud/tidb-management-service/utilities/workflow/mock_workflow"
)

func TestCheckNoPendingHeterogeneousDeployAllowsWhenNoPending(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	bizId := gkeHeterogeneousBizId(12345, globaldedicatedv1.ComponentType_TIKV, 0)

	mockWf := mock_workflow.NewMockWorkflow(ctrl)
	mockWf.EXPECT().ListRunningDagInstancesByNamePrefix(
		gomock.Any(),
		gkeHeterogeneousNamePrefix(applyGKEHeterogeneousDagNamePrefix, bizId),
		bizId,
	).Return(nil, nil)
	mockWf.EXPECT().ListRunningDagInstancesByNamePrefix(
		gomock.Any(),
		gkeHeterogeneousNamePrefix(deleteGKEHeterogeneousDagNamePrefix, bizId),
		bizId,
	).Return(nil, nil)

	svc := &serviceImpl{workflow: mockWf}
	err := svc.checkNoPendingHeterogeneousDeploy(t.Context(), 12345, globaldedicatedv1.ComponentType_TIKV, 0)
	require.NoError(t, err)
}

func TestCheckNoPendingHeterogeneousDeployWithTidbGroup(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	bizId := gkeHeterogeneousBizId(12345, globaldedicatedv1.ComponentType_TIDB, 100)

	mockWf := mock_workflow.NewMockWorkflow(ctrl)
	mockWf.EXPECT().ListRunningDagInstancesByNamePrefix(
		gomock.Any(),
		gkeHeterogeneousNamePrefix(applyGKEHeterogeneousDagNamePrefix, bizId),
		bizId,
	).Return(nil, nil)
	mockWf.EXPECT().ListRunningDagInstancesByNamePrefix(
		gomock.Any(),
		gkeHeterogeneousNamePrefix(deleteGKEHeterogeneousDagNamePrefix, bizId),
		bizId,
	).Return(nil, nil)

	svc := &serviceImpl{workflow: mockWf}
	err := svc.checkNoPendingHeterogeneousDeploy(t.Context(), 12345, globaldedicatedv1.ComponentType_TIDB, 100)
	require.NoError(t, err)
}

func TestCheckNoPendingHeterogeneousDeployRejectsWhenApplyPending(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	bizId := gkeHeterogeneousBizId(12345, globaldedicatedv1.ComponentType_TIDB, 100)

	mockWf := mock_workflow.NewMockWorkflow(ctrl)
	mockWf.EXPECT().ListRunningDagInstancesByNamePrefix(
		gomock.Any(),
		gkeHeterogeneousNamePrefix(applyGKEHeterogeneousDagNamePrefix, bizId),
		bizId,
	).Return([]*entity.DagInstance{
		{
			Status: entity.DagInstanceStatusRunning,
			Vars: entity.DagInstanceVars{
				common.KeyGKEHeterogeneousConcurrencyKey: entity.DagInstanceVar{
					Value: gkeHeterogeneousConcurrencyKey(12345, globaldedicatedv1.ComponentType_TIDB, 100, "n2d-custom-8-16384"),
				},
			},
		},
	}, nil)
	mockWf.EXPECT().ListRunningDagInstancesByNamePrefix(
		gomock.Any(),
		gkeHeterogeneousNamePrefix(deleteGKEHeterogeneousDagNamePrefix, bizId),
		bizId,
	).Return(nil, nil)

	svc := &serviceImpl{workflow: mockWf}
	err := svc.checkNoPendingHeterogeneousDeploy(t.Context(), 12345, globaldedicatedv1.ComponentType_TIDB, 100)
	require.Error(t, err)
	require.Equal(t, bizerr.ErrCodeParams, err.ErrCode())
	require.Contains(t, err.Error(), "already in progress")
}

func TestCheckNoPendingHeterogeneousDeployRejectsWhenDeletePending(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	bizId := gkeHeterogeneousBizId(12345, globaldedicatedv1.ComponentType_TIKV, 0)

	mockWf := mock_workflow.NewMockWorkflow(ctrl)
	mockWf.EXPECT().ListRunningDagInstancesByNamePrefix(
		gomock.Any(),
		gkeHeterogeneousNamePrefix(applyGKEHeterogeneousDagNamePrefix, bizId),
		bizId,
	).Return(nil, nil)
	mockWf.EXPECT().ListRunningDagInstancesByNamePrefix(
		gomock.Any(),
		gkeHeterogeneousNamePrefix(deleteGKEHeterogeneousDagNamePrefix, bizId),
		bizId,
	).Return([]*entity.DagInstance{
		{
			Status: entity.DagInstanceStatusRunning,
			Vars: entity.DagInstanceVars{
				common.KeyGKEHeterogeneousConcurrencyKey: entity.DagInstanceVar{
					Value: gkeHeterogeneousConcurrencyKey(12345, globaldedicatedv1.ComponentType_TIKV, 0, "n2-standard-8"),
				},
			},
		},
	}, nil)

	svc := &serviceImpl{workflow: mockWf}
	err := svc.checkNoPendingHeterogeneousDeploy(t.Context(), 12345, globaldedicatedv1.ComponentType_TIKV, 0)
	require.Error(t, err)
	require.Equal(t, bizerr.ErrCodeParams, err.ErrCode())
	require.Contains(t, err.Error(), "already in progress")
}

func TestCheckNoPendingHeterogeneousDeployPrefixBoundaryGroupNotConfused(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	bizId100 := gkeHeterogeneousBizId(12345, globaldedicatedv1.ComponentType_TIDB, 100)

	mockWf := mock_workflow.NewMockWorkflow(ctrl)
	mockWf.EXPECT().ListRunningDagInstancesByNamePrefix(
		gomock.Any(),
		gkeHeterogeneousNamePrefix(applyGKEHeterogeneousDagNamePrefix, bizId100),
		bizId100,
	).Return(nil, nil)
	mockWf.EXPECT().ListRunningDagInstancesByNamePrefix(
		gomock.Any(),
		gkeHeterogeneousNamePrefix(deleteGKEHeterogeneousDagNamePrefix, bizId100),
		bizId100,
	).Return(nil, nil)

	svc := &serviceImpl{workflow: mockWf}
	err := svc.checkNoPendingHeterogeneousDeploy(t.Context(), 12345, globaldedicatedv1.ComponentType_TIDB, 100)
	require.NoError(t, err)
}

func TestCheckNoPendingHeterogeneousDeployRejectsWhenFailedDag(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	bizId := gkeHeterogeneousBizId(12345, globaldedicatedv1.ComponentType_TIKV, 0)

	mockWf := mock_workflow.NewMockWorkflow(ctrl)
	mockWf.EXPECT().ListRunningDagInstancesByNamePrefix(
		gomock.Any(),
		gkeHeterogeneousNamePrefix(applyGKEHeterogeneousDagNamePrefix, bizId),
		bizId,
	).Return([]*entity.DagInstance{
		{
			Status: entity.DagInstanceStatusFailed,
			Vars: entity.DagInstanceVars{
				common.KeyGKEHeterogeneousConcurrencyKey: entity.DagInstanceVar{
					Value: gkeHeterogeneousConcurrencyKey(12345, globaldedicatedv1.ComponentType_TIKV, 0, "n2-standard-8"),
				},
			},
		},
	}, nil)
	mockWf.EXPECT().ListRunningDagInstancesByNamePrefix(
		gomock.Any(),
		gkeHeterogeneousNamePrefix(deleteGKEHeterogeneousDagNamePrefix, bizId),
		bizId,
	).Return(nil, nil)

	svc := &serviceImpl{workflow: mockWf}
	err := svc.checkNoPendingHeterogeneousDeploy(t.Context(), 12345, globaldedicatedv1.ComponentType_TIKV, 0)
	require.Error(t, err)
	require.Equal(t, bizerr.ErrCodeParams, err.ErrCode())
	require.Contains(t, err.Error(), "already in progress")
}

func TestCheckNoPendingHeterogeneousDeploySkipsSuccessDag(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	bizId := gkeHeterogeneousBizId(12345, globaldedicatedv1.ComponentType_TIKV, 0)

	mockWf := mock_workflow.NewMockWorkflow(ctrl)
	mockWf.EXPECT().ListRunningDagInstancesByNamePrefix(
		gomock.Any(),
		gkeHeterogeneousNamePrefix(applyGKEHeterogeneousDagNamePrefix, bizId),
		bizId,
	).Return([]*entity.DagInstance{
		{
			Status: entity.DagInstanceStatusSuccess,
			Vars: entity.DagInstanceVars{
				common.KeyGKEHeterogeneousConcurrencyKey: entity.DagInstanceVar{
					Value: gkeHeterogeneousConcurrencyKey(12345, globaldedicatedv1.ComponentType_TIKV, 0, "n2-standard-8"),
				},
			},
		},
	}, nil)
	mockWf.EXPECT().ListRunningDagInstancesByNamePrefix(
		gomock.Any(),
		gkeHeterogeneousNamePrefix(deleteGKEHeterogeneousDagNamePrefix, bizId),
		bizId,
	).Return(nil, nil)

	svc := &serviceImpl{workflow: mockWf}
	err := svc.checkNoPendingHeterogeneousDeploy(t.Context(), 12345, globaldedicatedv1.ComponentType_TIKV, 0)
	require.NoError(t, err)
}

func TestCheckNoApplyHeterogeneousDeployPendingRejectsWhenRunning(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	bizId := gkeHeterogeneousBizId(12345, globaldedicatedv1.ComponentType_TIDB, 100)

	mockWf := mock_workflow.NewMockWorkflow(ctrl)
	mockWf.EXPECT().ListRunningDagInstancesByNamePrefix(
		gomock.Any(),
		gkeHeterogeneousNamePrefix(applyGKEHeterogeneousDagNamePrefix, bizId),
		bizId,
	).Return([]*entity.DagInstance{
		{Status: entity.DagInstanceStatusRunning},
	}, nil)

	svc := &serviceImpl{workflow: mockWf}
	err := svc.checkNoApplyHeterogeneousDeployPending(t.Context(), 12345, globaldedicatedv1.ComponentType_TIDB, 100)
	require.Error(t, err)
	require.Equal(t, bizerr.ErrCodeParams, err.ErrCode())
	require.Contains(t, err.Error(), "in progress")
}

func TestCheckNoApplyHeterogeneousDeployPendingRejectsWhenFailed(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	bizId := gkeHeterogeneousBizId(12345, globaldedicatedv1.ComponentType_TIKV, 0)

	mockWf := mock_workflow.NewMockWorkflow(ctrl)
	mockWf.EXPECT().ListRunningDagInstancesByNamePrefix(
		gomock.Any(),
		gkeHeterogeneousNamePrefix(applyGKEHeterogeneousDagNamePrefix, bizId),
		bizId,
	).Return([]*entity.DagInstance{
		{Status: entity.DagInstanceStatusFailed},
	}, nil)

	svc := &serviceImpl{workflow: mockWf}
	err := svc.checkNoApplyHeterogeneousDeployPending(t.Context(), 12345, globaldedicatedv1.ComponentType_TIKV, 0)
	require.Error(t, err)
	require.Equal(t, bizerr.ErrCodeParams, err.ErrCode())
	require.Contains(t, err.Error(), "in progress")
}

func TestCheckNoApplyHeterogeneousDeployPendingAllowsWhenBlocked(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	bizId := gkeHeterogeneousBizId(12345, globaldedicatedv1.ComponentType_TIDB, 100)

	mockWf := mock_workflow.NewMockWorkflow(ctrl)
	mockWf.EXPECT().ListRunningDagInstancesByNamePrefix(
		gomock.Any(),
		gkeHeterogeneousNamePrefix(applyGKEHeterogeneousDagNamePrefix, bizId),
		bizId,
	).Return(nil, nil)

	svc := &serviceImpl{workflow: mockWf}
	err := svc.checkNoApplyHeterogeneousDeployPending(t.Context(), 12345, globaldedicatedv1.ComponentType_TIDB, 100)
	require.NoError(t, err)
}

func TestCheckNoApplyHeterogeneousDeployPendingAllowsWhenSuccess(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	bizId := gkeHeterogeneousBizId(12345, globaldedicatedv1.ComponentType_TIKV, 0)

	mockWf := mock_workflow.NewMockWorkflow(ctrl)
	mockWf.EXPECT().ListRunningDagInstancesByNamePrefix(
		gomock.Any(),
		gkeHeterogeneousNamePrefix(applyGKEHeterogeneousDagNamePrefix, bizId),
		bizId,
	).Return(nil, nil)

	svc := &serviceImpl{workflow: mockWf}
	err := svc.checkNoApplyHeterogeneousDeployPending(t.Context(), 12345, globaldedicatedv1.ComponentType_TIKV, 0)
	require.NoError(t, err)
}

func TestCheckNoApplyHeterogeneousDeployPendingAllowsWhenNoDags(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	bizId := gkeHeterogeneousBizId(12345, globaldedicatedv1.ComponentType_TIDB, 0)

	mockWf := mock_workflow.NewMockWorkflow(ctrl)
	mockWf.EXPECT().ListRunningDagInstancesByNamePrefix(
		gomock.Any(),
		gkeHeterogeneousNamePrefix(applyGKEHeterogeneousDagNamePrefix, bizId),
		bizId,
	).Return(nil, nil)

	svc := &serviceImpl{workflow: mockWf}
	err := svc.checkNoApplyHeterogeneousDeployPending(t.Context(), 12345, globaldedicatedv1.ComponentType_TIDB, 0)
	require.NoError(t, err)
}