package premium_dm

import (
	"context"
	"testing"

	"github.com/pingcap/tiflow/dm/pkg/checker"
	"github.com/stretchr/testify/require"
	modelpremiumdm "github.com/tidbcloud/dataflow-service-ng/app/models/premium_dm"
	"github.com/tidbcloud/dataflow-service-ng/app/repo/entity"
	"github.com/tidbcloud/dataflow-service-ng/idl/pbgen/proto/premium_dm/v1beta2"
	"github.com/tidbcloud/dataflow-service-ng/internal/config"
	tidbmgmtservicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"google.golang.org/protobuf/proto"
)

func TestDMTaskStageToV1beta1AsyncTransitions(t *testing.T) {
	require.Equal(t, v1beta2.PremiumMigration_PAUSING,
		DMTaskStageToV1beta1(entity.PremiumDMTaskStagePausing))
	require.Equal(t, v1beta2.PremiumMigration_RESUMING,
		DMTaskStageToV1beta1(entity.PremiumDMTaskStageResuming))
}

func TestGetPremiumPrecheckItemList(t *testing.T) {
	contains := func(items []v1beta2.PrecheckItemType, target v1beta2.PrecheckItemType) bool {
		for _, item := range items {
			if item == target {
				return true
			}
		}
		return false
	}

	t.Run("incremental without checkpoint drops meta position", func(t *testing.T) {
		items := GetPremiumPrecheckItemList(
			entity.PremiumDMTaskModeIncremental, false, entity.PremiumDMTaskImportModeLogical,
			false,
			config.PrecheckSkipConfigs{}, 0, 0, 0,
		)
		require.False(t, contains(items, v1beta2.PrecheckItemType_META_POSITION_CHECKING))
	})

	t.Run("incremental with checkpoint keeps meta position", func(t *testing.T) {
		items := GetPremiumPrecheckItemList(
			entity.PremiumDMTaskModeIncremental, true, entity.PremiumDMTaskImportModeLogical,
			false,
			config.PrecheckSkipConfigs{}, 0, 0, 0,
		)
		require.True(t, contains(items, v1beta2.PrecheckItemType_META_POSITION_CHECKING))
	})

	t.Run("all mode logical import drops lightning item", func(t *testing.T) {
		items := GetPremiumPrecheckItemList(
			entity.PremiumDMTaskModeAll, false, entity.PremiumDMTaskImportModeLogical,
			false,
			config.PrecheckSkipConfigs{}, 0, 0, 0,
		)
		require.False(t, contains(items, v1beta2.PrecheckItemType_LIGHTNING_TABLE_EMPTY_CHECKING))
	})

	t.Run("all mode physical drops lightning item", func(t *testing.T) {
		items := GetPremiumPrecheckItemList(
			entity.PremiumDMTaskModeAll, false, entity.PremiumDMTaskImportModePhysical,
			false,
			config.PrecheckSkipConfigs{}, 0, 0, 0,
		)
		require.False(t, contains(items, v1beta2.PrecheckItemType_LIGHTNING_TABLE_EMPTY_CHECKING))
	})

	t.Run("alicloud rds prepends primary key item", func(t *testing.T) {
		items := GetPremiumPrecheckItemList(
			entity.PremiumDMTaskModeAll, false, entity.PremiumDMTaskImportModeLogical,
			true,
			config.PrecheckSkipConfigs{}, 0, 0, 0,
		)
		require.NotEmpty(t, items)
		require.Equal(t, v1beta2.PrecheckItemType_PRIMARY_KEY_CHECKING, items[0])
	})

	t.Run("skip config removes primary key item", func(t *testing.T) {
		skipConfigs := config.PrecheckSkipConfigs{{
			Matcher:   config.AllowListMatcher{ClusterID: []string{"42"}},
			SkipItems: []string{modelpremiumdm.PrimaryKeyChecking},
		}}
		items := GetPremiumPrecheckItemList(
			entity.PremiumDMTaskModeAll, false, entity.PremiumDMTaskImportModeLogical,
			true,
			skipConfigs, 0, 0, 42,
		)
		require.False(t, contains(items, v1beta2.PrecheckItemType_PRIMARY_KEY_CHECKING))
	})
}

func TestCreateV1beta2TaskReqToPremiumDMPrecheck(t *testing.T) {
	req := &v1beta2.CreatePremiumMigrationReq{
		TidbId:      "123456",
		DisplayName: "test-precheck",
		Mode:        v1beta2.PremiumDMTaskMode_INCREMENTAL,
		ImportMode:  v1beta2.PremiumDMTaskImportMode_IMPORTMODE_LOGICAL,
		Target: &v1beta2.PremiumDMTarget{
			User:     "root",
			Password: "pwd",
		},
		Sources: []*v1beta2.PremiumDMSource{
			{
				BinlogName: proto.String("mysql-bin.000001"),
				SourceType: v1beta2.PremiumDMSourceType_MYSQL,
			},
		},
	}
	cluster := &tidbmgmtservicev1.Cluster{
		OrganizationId: 1001,
		ProjectId:      2002,
	}

	precheck, err := CreateV1beta2TaskReqToPremiumDMPrecheck(context.Background(), req, cluster, config.PrecheckSkipConfigs{})
	require.NoError(t, err)
	require.NotNil(t, precheck)
	require.NotEmpty(t, precheck.ID)
	require.Equal(t, "123456", precheck.TiDBID)
	require.Equal(t, "1001", precheck.TenantID)
	require.Equal(t, "2002", precheck.ProjectID)
	require.Equal(t, string(entity.PremiumDMPrecheckStagePending), precheck.Stage)
	require.NotEmpty(t, precheck.Request)
	require.Equal(
		t,
		int32(len(GetPremiumPrecheckItemList(
			entity.PremiumDMTaskModeIncremental, true, entity.PremiumDMTaskImportModeLogical,
			false,
			config.PrecheckSkipConfigs{},
			cluster.OrganizationId, cluster.ProjectId, cluster.ClusterId,
		))),
		precheck.TotalItems,
	)
}

func TestCreateV1beta2TaskReqToPremiumDMPrecheckUsesAnySourceCheckpointForTotalItems(t *testing.T) {
	req := &v1beta2.CreatePremiumMigrationReq{
		TidbId:      "123456",
		DisplayName: "test-precheck-any-source-checkpoint",
		Mode:        v1beta2.PremiumDMTaskMode_INCREMENTAL,
		ImportMode:  v1beta2.PremiumDMTaskImportMode_IMPORTMODE_LOGICAL,
		Target: &v1beta2.PremiumDMTarget{
			User:     "root",
			Password: "pwd",
		},
		Sources: []*v1beta2.PremiumDMSource{
			{
				SourceType: v1beta2.PremiumDMSourceType_MYSQL,
			},
			{
				BinlogGtid: proto.String("uuid:1-100"),
				SourceType: v1beta2.PremiumDMSourceType_MYSQL,
			},
		},
	}
	cluster := &tidbmgmtservicev1.Cluster{
		OrganizationId: 1001,
		ProjectId:      2002,
	}

	precheck, err := CreateV1beta2TaskReqToPremiumDMPrecheck(context.Background(), req, cluster, config.PrecheckSkipConfigs{})
	require.NoError(t, err)
	require.NotNil(t, precheck)
	require.Equal(
		t,
		int32(len(GetPremiumPrecheckItemList(
			entity.PremiumDMTaskModeIncremental, true, entity.PremiumDMTaskImportModeLogical,
			false,
			config.PrecheckSkipConfigs{},
			cluster.OrganizationId, cluster.ProjectId, cluster.ClusterId,
		))),
		precheck.TotalItems,
	)
}

func TestCreateV1beta2TaskReqToPremiumDMPrecheckAddsPrimaryKeyOnceForAnyAliCloudRDSSource(t *testing.T) {
	req := &v1beta2.CreatePremiumMigrationReq{
		TidbId:     "123456",
		Mode:       v1beta2.PremiumDMTaskMode_ALL,
		ImportMode: v1beta2.PremiumDMTaskImportMode_IMPORTMODE_LOGICAL,
		Target:     &v1beta2.PremiumDMTarget{User: "root", Password: "pwd"},
		Sources: []*v1beta2.PremiumDMSource{
			{SourceType: v1beta2.PremiumDMSourceType_MYSQL},
			{SourceType: v1beta2.PremiumDMSourceType_ALICLOUD_RDS_MYSQL},
			{SourceType: v1beta2.PremiumDMSourceType_ALICLOUD_RDS_MYSQL},
		},
	}
	cluster := &tidbmgmtservicev1.Cluster{OrganizationId: 1001, ProjectId: 2002, ClusterId: 3003}

	precheck, err := CreateV1beta2TaskReqToPremiumDMPrecheck(context.Background(), req, cluster, config.PrecheckSkipConfigs{})
	require.NoError(t, err)
	require.Equal(t, int32(len(GetPremiumPrecheckItemList(
		entity.PremiumDMTaskModeAll, false, entity.PremiumDMTaskImportModeLogical,
		true, config.PrecheckSkipConfigs{},
		cluster.OrganizationId, cluster.ProjectId, cluster.ClusterId,
	))), precheck.TotalItems)
}

func TestPremiumDMPrecheckToV1beta2PbPrecheck(t *testing.T) {
	precheck := &entity.PremiumDMPrecheck{
		ID:         "pck-1",
		Stage:      string(entity.PremiumDMPrecheckStageRunning),
		TotalItems: 2,
		DMPrecheckItems: []entity.PremiumDMPrecheckItem{
			{
				ID:          "item-failed",
				Description: "failed item",
				Type:        1,
				Stage:       string(entity.PremiumDMPrecheckStageFailed),
				ErrMsg:      "failed to precheck dm job",
			},
			{
				ID:          "item-running",
				Description: "running item",
				Type:        2,
				Stage:       string(entity.PremiumDMPrecheckStageRunning),
			},
		},
	}

	resp := PremiumDMPrecheckToV1beta2PbPrecheck(precheck)
	require.NotNil(t, resp)
	require.Equal(t, "pck-1", resp.PrecheckId)
	require.Equal(t, int32(2), resp.Total)
	require.Equal(t, int32(1), resp.FailedCnt)
	require.Equal(t, int32(0), resp.WarnCnt)
	require.Equal(t, int32(0), resp.SuccessCnt)
	require.Equal(t, v1beta2.PremiumMigrationPrecheck_FAILED, resp.Status)
	require.Len(t, resp.Items, 1)
	require.Equal(t, v1beta2.PremiumDMPrecheckItem_FAILED, resp.Items[0].Status)
	require.Equal(t, "internal error: precheck item failed unexpectedly", resp.Items[0].Reason)
	require.Equal(t, v1beta2.PrecheckItemType_DUMP_PRIVILEGE_CHECKING, resp.Items[0].Type)
	require.Equal(t, premiumPrecheckItemHelpURL(v1beta2.PrecheckItemType_DUMP_PRIVILEGE_CHECKING), resp.Items[0].SolutionDocUrl)
}

func TestPremiumDMPrecheckToV1beta2PbPrecheckFallsBackToGenericReason(t *testing.T) {
	precheck := &entity.PremiumDMPrecheck{
		ID:         "pck-3",
		Stage:      string(entity.PremiumDMPrecheckStageFailed),
		TotalItems: 1,
		DMPrecheckItems: []entity.PremiumDMPrecheckItem{{
			ID:          "item-failed",
			Description: "failed item",
			Type:        1,
			Stage:       string(entity.PremiumDMPrecheckStageFailed),
		}},
	}

	resp := PremiumDMPrecheckToV1beta2PbPrecheck(precheck)
	require.NotNil(t, resp)
	require.Len(t, resp.Items, 1)
	require.Equal(t, "internal error: precheck item failed unexpectedly", resp.Items[0].Reason)
}

func TestPremiumSourceTypeRoundTripAzure(t *testing.T) {
	require.Equal(t, entity.PremiumDBTypeAzureFlexibleMySQL, v1beta1SourceTypeToEntity(v1beta2.PremiumDMSourceType_AZURE_FLEXIBLE_MYSQL))
	require.Equal(t, v1beta2.PremiumDMSourceType_AZURE_FLEXIBLE_MYSQL, entitySourceType2V1beta1(entity.PremiumDBTypeAzureFlexibleMySQL))
}

func TestPremiumSourceTypeRoundTripAliCloudRDS(t *testing.T) {
	require.Equal(t, entity.PremiumDBTypeAliCloudRDSMySQL, v1beta1SourceTypeToEntity(v1beta2.PremiumDMSourceType_ALICLOUD_RDS_MYSQL))
	require.Equal(t, v1beta2.PremiumDMSourceType_ALICLOUD_RDS_MYSQL, entitySourceType2V1beta1(entity.PremiumDBTypeAliCloudRDSMySQL))
}

func TestPremiumDMPrecheckToV1beta2PbPrecheckHidesInternalFailureReason(t *testing.T) {
	precheck := &entity.PremiumDMPrecheck{
		ID:         "pck-4",
		Stage:      string(entity.PremiumDMPrecheckStageFailed),
		TotalItems: 1,
		DMPrecheckItems: []entity.PremiumDMPrecheckItem{{
			ID:          "item-failed",
			Description: "failed item",
			Type:        1,
			Stage:       string(entity.PremiumDMPrecheckStageFailed),
			ErrMsg:      "failed to precheck dm job: rpc timeout",
		}},
	}

	resp := PremiumDMPrecheckToV1beta2PbPrecheck(precheck)
	require.NotNil(t, resp)
	require.Len(t, resp.Items, 1)
	require.Equal(t, "internal error: precheck item failed unexpectedly", resp.Items[0].Reason)
}

func TestPremiumDMPrecheckToV1beta2PbPrecheckHidesInternalConfigFailureReason(t *testing.T) {
	precheck := &entity.PremiumDMPrecheck{
		ID:         "pck-5",
		Stage:      string(entity.PremiumDMPrecheckStageFailed),
		TotalItems: 1,
		DMPrecheckItems: []entity.PremiumDMPrecheckItem{{
			ID:          "item-failed",
			Description: "failed item",
			Type:        1,
			Stage:       string(entity.PremiumDMPrecheckStageFailed),
			ErrMsg:      "no valid subtask config for precheck",
		}},
	}

	resp := PremiumDMPrecheckToV1beta2PbPrecheck(precheck)
	require.NotNil(t, resp)
	require.Len(t, resp.Items, 1)
	require.Equal(t, "internal error: precheck item failed unexpectedly", resp.Items[0].Reason)
}

func TestPremiumDMPrecheckToV1beta2PbPrecheckAggregatesFinishedItemResults(t *testing.T) {
	precheck := &entity.PremiumDMPrecheck{
		ID:         "pck-2",
		Stage:      string(entity.PremiumDMPrecheckStageFinished),
		TotalItems: 1,
		DMPrecheckItems: []entity.PremiumDMPrecheckItem{
			{
				ID:          "item-finished",
				Description: "finished item",
				Type:        1,
				Stage:       string(entity.PremiumDMPrecheckStageFinished),
				Results: entity.PremiumResults{
					Results: []*checker.Result{
						{
							Desc:        "check desc",
							State:       checker.StateSuccess,
							Instruction: "first step",
						},
						{
							State: checker.StateFailure,
							Errors: []*checker.Error{{
								ShortErr: "failed later",
							}},
						},
					},
				},
			},
		},
	}

	resp := PremiumDMPrecheckToV1beta2PbPrecheck(precheck)
	require.NotNil(t, resp)
	require.Equal(t, v1beta2.PremiumMigrationPrecheck_FAILED, resp.Status)
	require.Equal(t, int32(1), resp.FailedCnt)
	require.Len(t, resp.Items, 1)
	require.Equal(t, v1beta2.PremiumDMPrecheckItem_FAILED, resp.Items[0].Status)
	require.Equal(t, "failed later", resp.Items[0].Reason)
	require.Equal(t, "first step", resp.Items[0].Solution)
	require.Equal(t, premiumPrecheckItemHelpURL(v1beta2.PrecheckItemType_DUMP_PRIVILEGE_CHECKING), resp.Items[0].SolutionDocUrl)
}

func TestPremiumDMPrecheckToV1beta2PbPrecheckHidesInitializationMarkerDescription(t *testing.T) {
	precheck := &entity.PremiumDMPrecheck{
		ID:         "pck-init",
		Stage:      string(entity.PremiumDMPrecheckStageFinished),
		TotalItems: 1,
		DMPrecheckItems: []entity.PremiumDMPrecheckItem{
			{
				ID:          "item-init",
				Description: "dump_privilege",
				Type:        1,
				Stage:       string(entity.PremiumDMPrecheckStageFinished),
				Results: entity.PremiumResults{
					Results: []*checker.Result{
						{
							Desc:  modelpremiumdm.PremiumDMPrecheckInitializationDescForAPI(),
							State: checker.StateFailure,
							Errors: []*checker.Error{{
								ShortErr: "fail to initialize checker: invalid route rules",
							}},
						},
					},
				},
			},
		},
	}

	resp := PremiumDMPrecheckToV1beta2PbPrecheck(precheck)
	require.NotNil(t, resp)
	require.Len(t, resp.Items, 1)
	require.Equal(t, "dump_privilege", resp.Items[0].Description)
}

func TestPremiumCheckerStateToV1beta2(t *testing.T) {
	require.Equal(t, v1beta2.PremiumDMPrecheckItem_FAILED, premiumCheckerStateToV1beta2(checker.StateFailure))
	require.Equal(t, v1beta2.PremiumDMPrecheckItem_WARNING, premiumCheckerStateToV1beta2(checker.StateWarning))
	require.Equal(t, v1beta2.PremiumDMPrecheckItem_SUCCESS, premiumCheckerStateToV1beta2(checker.StateSuccess))
	require.Equal(t, v1beta2.PremiumDMPrecheckItem_STATUS_UNSPECIFIED, premiumCheckerStateToV1beta2(checker.State("unknown")))
}