package cluster

import (
	"context"
	"strconv"
	"testing"

	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/pkgv2/db"
	"go.uber.org/mock/gomock"
	"gorm.io/gorm"

	globalcommonv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/common/v1"
	globalngv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/nextgen/v1"
	openapicommonv1beta1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_cloud_open_api/common/v1beta1"
	tidbmgmtservicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	mockaccount "github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/adapter/account/mock_account"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/adapter/nextgenglobal/mock_nextgenglobal"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/constant"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/entity"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/repository"
	grpcmetadata "github.com/tidbcloud/tidb-management-service/utilities/grpc/metadata"
	mockworkflow "github.com/tidbcloud/tidb-management-service/utilities/workflow/mock_workflow"
)

func TestCleanClustersSubmitsPremiumAndEssentialV2Deletion(t *testing.T) {
	ctrl := gomock.NewController(t)
	clusters := []*entity.Cluster{
		{ClusterId: 1, OrganizationId: 8, DeployType: entity.DeployTypeNextGen, ServicePlan: openapicommonv1beta1.ServicePlan_Premium},
		{ClusterId: 2, OrganizationId: 8, DeployType: entity.DeployTypeNextGen, ServicePlan: openapicommonv1beta1.ServicePlan_Essential_V2},
	}
	repo := &cleanClusterRepository{clusters: clusters}
	client := mock_nextgenglobal.NewMockNextGenGlobal(ctrl)
	accountClient := mockaccount.NewMockClient(ctrl)
	flow := mockworkflow.NewMockWorkflow(ctrl)
	for _, cluster := range clusters {
		shouldDeleteResourcePool := cluster.ServicePlan == openapicommonv1beta1.ServicePlan_Premium
		client.EXPECT().DeleteCluster(gomock.Any(), gomock.Any()).DoAndReturn(
			func(_ context.Context, request *globalngv1.DeleteClusterRequest, _ ...any) (*globalngv1.Cluster, error) {
				require.Equal(t, cluster.ClusterId, request.GetClusterId())
				require.True(t, request.GetValidateOnly())
				return &globalngv1.Cluster{
					ClusterId:      cluster.ClusterId,
					OrganizationId: cluster.OrganizationId,
					SharedPoolId:   strconv.FormatUint(cluster.ClusterId, 10),
					ServicePlan:    globalcommonv1.ServicePlan(cluster.ServicePlan),
				}, nil
			},
		)
		flow.EXPECT().DeleteDagInstances(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil)
		flow.EXPECT().RunDagWithBizId(gomock.Any(), constant.DagNextGenDeleteCluster, "8", "0", gomock.Any(), gomock.Any()).DoAndReturn(
			func(_ context.Context, _, _, _, _ string, vars map[string]string) error {
				require.Equal(t, "true", vars[constant.KeyFinalBackup])
				if shouldDeleteResourcePool {
					require.Equal(t, "true", vars[constant.KeyDeleteResourcePoolIfEmpty])
					require.Equal(t, strconv.FormatUint(cluster.ClusterId, 10), vars[constant.KeyResourcePoolID])
				} else {
					require.NotContains(t, vars, constant.KeyDeleteResourcePoolIfEmpty)
					require.NotContains(t, vars, constant.KeyResourcePoolID)
				}
				return nil
			},
		)
		accountClient.EXPECT().DeleteClusterRelations(gomock.Any(), cluster.ClusterId).Return(nil)
	}

	service := &service{
		clusterRepo:      repo,
		nextGenCli:       client,
		customStatusRepo: cleanCustomStatusRepository{},
		accountCli:       accountClient,
		transaction:      cleanTransactioner{},
		workflow:         flow,
	}
	ctx := grpcmetadata.AppendToOutgoingMeta(t.Context(), &grpcmetadata.RequestMeta{OrgID: 8})
	require.NoError(t, service.CleanClusters(ctx, 8, true))
	require.Equal(t, []int{0, 0}, repo.offsets)
	require.Equal(t, []string{
		openapicommonv1beta1.ServicePlan_Premium.String(),
		openapicommonv1beta1.ServicePlan_Essential_V2.String(),
	}, repo.filter.ServicePlans)
	require.Equal(t, 2, repo.deleted)
}

func TestCleanClustersRejectsUnexpectedServicePlan(t *testing.T) {
	repo := &cleanClusterRepository{clusters: []*entity.Cluster{{
		ClusterId:      1,
		OrganizationId: 8,
		DeployType:     entity.DeployTypeNextGen,
		ServicePlan:    openapicommonv1beta1.ServicePlan_BYOC,
	}}}
	service := &service{clusterRepo: repo}
	err := service.CleanClusters(t.Context(), 8, true)
	require.ErrorContains(t, err, "is not cleanable")
	require.Zero(t, repo.deleted)
}

func TestDeleteClusterDoesNotRequireResourcePoolCleanupMetadata(t *testing.T) {
	ctrl := gomock.NewController(t)
	cluster := &entity.Cluster{
		ClusterId:      1,
		OrganizationId: 8,
		DeployType:     entity.DeployTypeNextGen,
		ServicePlan:    openapicommonv1beta1.ServicePlan_Premium,
	}
	repo := &cleanClusterRepository{clusters: []*entity.Cluster{cluster}}
	client := mock_nextgenglobal.NewMockNextGenGlobal(ctrl)
	accountClient := mockaccount.NewMockClient(ctrl)
	flow := mockworkflow.NewMockWorkflow(ctrl)
	client.EXPECT().DeleteCluster(gomock.Any(), gomock.Any()).Return(&globalngv1.Cluster{
		ClusterId:      cluster.ClusterId,
		OrganizationId: cluster.OrganizationId,
		SharedPoolId:   "not-a-resource-pool-id",
		ServicePlan:    globalcommonv1.ServicePlan_Premium,
	}, nil)
	flow.EXPECT().DeleteDagInstances(gomock.Any(), "1", gomock.Any()).Return(nil)
	flow.EXPECT().RunDagWithBizId(gomock.Any(), constant.DagNextGenDeleteCluster, "8", "0", "1", gomock.Any()).DoAndReturn(
		func(_ context.Context, _, _, _, _ string, vars map[string]string) error {
			require.NotContains(t, vars, constant.KeyResourcePoolID)
			require.NotContains(t, vars, constant.KeyDeleteResourcePoolIfEmpty)
			return nil
		},
	)
	accountClient.EXPECT().DeleteClusterRelations(gomock.Any(), cluster.ClusterId).Return(nil)

	service := &service{
		clusterRepo:      repo,
		nextGenCli:       client,
		customStatusRepo: cleanCustomStatusRepository{},
		accountCli:       accountClient,
		transaction:      cleanTransactioner{},
		workflow:         flow,
	}
	ctx := grpcmetadata.AppendToOutgoingMeta(t.Context(), &grpcmetadata.RequestMeta{OrgID: 8})
	_, _, err := service.DeleteCluster(ctx, cluster.ClusterId, false)
	require.NoError(t, err)
}

type cleanClusterRepository struct {
	repository.ClusterRepository

	clusters []*entity.Cluster
	filter   *repository.ClusterFilter
	offsets  []int
	queries  int
	deleted  int
}

func (r *cleanClusterRepository) QueryClusterByFilterAndPage(
	_ context.Context,
	filter *repository.ClusterFilter,
	offset, _ int,
) ([]*entity.Cluster, error) {
	r.filter = filter
	r.offsets = append(r.offsets, offset)
	r.queries++
	if r.queries > 1 {
		return nil, nil
	}
	return r.clusters, nil
}

func (r *cleanClusterRepository) GetCluster(_ context.Context, clusterID uint64, _ bool) (*entity.Cluster, error) {
	for _, cluster := range r.clusters {
		if cluster.ClusterId == clusterID {
			return cluster, nil
		}
	}
	return nil, gorm.ErrRecordNotFound
}

func (r *cleanClusterRepository) DeleteCluster(context.Context, uint64) error {
	r.deleted++
	return nil
}

type cleanCustomStatusRepository struct {
	repository.CustomStatusRepository
}

func (cleanCustomStatusRepository) IsMutexClusterAction(
	context.Context,
	uint64,
	tidbmgmtservicev1.ClusterAction,
) (bool, *entity.ClusterCustomStatus, error) {
	return false, nil, gorm.ErrRecordNotFound
}

func (cleanCustomStatusRepository) DeleteClusterCustomStatusByCluster(context.Context, uint64) error {
	return nil
}

type cleanTransactioner struct {
	db.TransProvider
}

func (cleanTransactioner) Transaction(ctx context.Context, callback func(context.Context) error) error {
	return callback(ctx)
}