package premium_dm

import (
	"context"
	"encoding/base64"
	"encoding/json"
	"errors"
	"net/http"
	"net/http/httptest"
	"testing"

	dmconfig "github.com/pingcap/tiflow/dm/config"
	"github.com/pingcap/tiflow/dm/pkg/checker"
	"github.com/stretchr/testify/require"
	"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"
	"github.com/tidbcloud/dataflow-service-ng/pkg/agent"
	bizerr_wrapper "github.com/tidbcloud/dataflow-service-ng/pkg/bizerr-wrapper"
	encryptormock "github.com/tidbcloud/dataflow-service-ng/pkg/encryptor-wrapper/mock"
	agentpb "github.com/tidbcloud/dataflow-service/dataflow-agent-ng/idl/pbgen/proto/ng_agent"
	"github.com/tidbcloud/pkgv2/bizerr"
	tidbmgmtservicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.uber.org/zap"
	"google.golang.org/protobuf/encoding/protojson"
	"google.golang.org/protobuf/proto"
)

type fakePremiumDMPrecheckDAO struct {
	createdPrechecks []*entity.PremiumDMPrecheck
	createdItems     []*entity.PremiumDMPrecheckItem
	updatedPrechecks []map[string]any
	updatedItems     []map[string]any
	listPrechecks    []*entity.PremiumDMPrecheck
	listFilter       *entity.PremiumDMPrecheckFilter
}

func (f *fakePremiumDMPrecheckDAO) CreatePrecheck(_ context.Context, precheck *entity.PremiumDMPrecheck) bizerr.BizError {
	f.createdPrechecks = append(f.createdPrechecks, precheck)
	return nil
}

func (f *fakePremiumDMPrecheckDAO) ListPrechecks(_ context.Context, filter *entity.PremiumDMPrecheckFilter) ([]*entity.PremiumDMPrecheck, bizerr.BizError) {
	f.listFilter = filter
	return f.listPrechecks, nil
}

func (f *fakePremiumDMPrecheckDAO) GetPrecheck(_ context.Context, clusterID, id string) (*entity.PremiumDMPrecheck, bizerr.BizError) {
	for _, precheck := range f.listPrechecks {
		if precheck != nil && precheck.TiDBID == clusterID && precheck.ID == id {
			return precheck, nil
		}
	}
	return nil, bizerr_wrapper.NewInternalErr("precheck not found")
}

func (f *fakePremiumDMPrecheckDAO) UpdatePrecheck(_ context.Context, id string, updates map[string]any) bizerr.BizError {
	f.updatedPrechecks = append(f.updatedPrechecks, updates)
	for _, precheck := range f.listPrechecks {
		if precheck == nil || precheck.ID != id {
			continue
		}
		if stage, ok := updates["stage"].(entity.PremiumDMPrecheckStage); ok {
			precheck.Stage = string(stage)
		}
		if stage, ok := updates["stage"].(string); ok {
			precheck.Stage = stage
		}
		if errMsg, ok := updates["err_msg"].(string); ok {
			precheck.ErrMsg = errMsg
		}
	}
	return nil
}

func (f *fakePremiumDMPrecheckDAO) CreatePrecheckItem(_ context.Context, item *entity.PremiumDMPrecheckItem) bizerr.BizError {
	if item != nil {
		copied := *item
		f.createdItems = append(f.createdItems, &copied)
		for _, precheck := range f.listPrechecks {
			if precheck != nil && precheck.ID == item.DMPrecheckID {
				precheck.DMPrecheckItems = append(precheck.DMPrecheckItems, copied)
			}
		}
	}
	return nil
}

func (f *fakePremiumDMPrecheckDAO) UpdatePrecheckItem(_ context.Context, id string, updates map[string]any) bizerr.BizError {
	f.updatedItems = append(f.updatedItems, updates)
	for _, precheck := range f.listPrechecks {
		if precheck == nil {
			continue
		}
		for i := range precheck.DMPrecheckItems {
			item := &precheck.DMPrecheckItems[i]
			if item.ID != id {
				continue
			}
			if stage, ok := updates["stage"].(string); ok {
				item.Stage = stage
			}
			if errMsg, ok := updates["err_msg"].(string); ok {
				item.ErrMsg = errMsg
			}
			if results, ok := updates["results"].(entity.PremiumResults); ok {
				item.Results = results
			}
		}
	}
	return nil
}

func TestPremiumDMPrecheckWorkerCreatePrecheckDefaultsPendingWithoutExecuting(t *testing.T) {
	dao := &fakePremiumDMPrecheckDAO{}
	executed := false
	worker := &PremiumDMPrecheckWorker{
		dao: dao,
		runPrecheck: func(ctx context.Context, precheck *entity.PremiumDMPrecheck) error {
			executed = true
			return nil
		},
	}

	precheck := &entity.PremiumDMPrecheck{
		TiDBID:   "cluster-1",
		TenantID: "tenant-1",
	}

	err := worker.CreatePrecheck(context.Background(), precheck)
	require.NoError(t, err)
	require.Len(t, dao.createdPrechecks, 1)
	require.NotEmpty(t, precheck.ID)
	require.Equal(t, string(entity.PremiumDMPrecheckStagePending), precheck.Stage)
	require.False(t, executed)
}

func TestPremiumDMPrecheckWorkerRunProcessesAllPendingPrechecksSequentially(t *testing.T) {
	dao := &fakePremiumDMPrecheckDAO{
		listPrechecks: []*entity.PremiumDMPrecheck{
			{ID: "pck-1", TiDBID: "cluster-1", Stage: string(entity.PremiumDMPrecheckStagePending)},
			{ID: "pck-2", TiDBID: "cluster-2", Stage: string(entity.PremiumDMPrecheckStagePending)},
		},
	}
	executed := make([]string, 0, 2)
	worker := &PremiumDMPrecheckWorker{
		dao: dao,
		runPrecheck: func(ctx context.Context, precheck *entity.PremiumDMPrecheck) error {
			executed = append(executed, precheck.ID)
			return nil
		},
	}

	err := worker.Run(context.Background())
	require.NoError(t, err)
	require.NotNil(t, dao.listFilter)
	require.Equal(t, []string{string(entity.PremiumDMPrecheckStagePending)}, dao.listFilter.Stages)
	require.Equal(t, []string{"pck-1", "pck-2"}, executed)
}

func TestPremiumDMPrecheckWorkerRunContinuesAfterPrecheckFailure(t *testing.T) {
	dao := &fakePremiumDMPrecheckDAO{
		listPrechecks: []*entity.PremiumDMPrecheck{
			{ID: "pck-1", TiDBID: "cluster-1", Stage: string(entity.PremiumDMPrecheckStagePending)},
			{ID: "pck-2", TiDBID: "cluster-2", Stage: string(entity.PremiumDMPrecheckStagePending)},
		},
	}
	executed := make([]string, 0, 2)
	worker := &PremiumDMPrecheckWorker{
		dao: dao,
		runPrecheck: func(ctx context.Context, precheck *entity.PremiumDMPrecheck) error {
			executed = append(executed, precheck.ID)
			if precheck.ID == "pck-1" {
				return bizerr_wrapper.NewInternalErr("boom")
			}
			return nil
		},
	}

	err := worker.Run(context.Background())
	require.Error(t, err)
	require.Equal(t, []string{"pck-1", "pck-2"}, executed)
}

func TestGetPremiumPrecheckExecutionItems(t *testing.T) {
	containsDescription := func(items []*entity.PremiumDMPrecheckItem, target string) bool {
		for _, item := range items {
			if item != nil && item.Description == target {
				return true
			}
		}
		return false
	}

	t.Run("incremental without checkpoint drops meta position", func(t *testing.T) {
		items := GetPremiumPrecheckExecutionItems(
			entity.PremiumDMTaskModeIncremental,
			false,
			entity.PremiumDMTaskImportModeLogical,
			false,
			"pck-1",
			config.PrecheckSkipConfigs{},
			0, 0, 0,
		)
		require.False(t, containsDescription(items, dmconfig.MetaPositionChecking))
	})

	t.Run("alicloud rds prepends primary key item", func(t *testing.T) {
		items := GetPremiumPrecheckExecutionItems(
			entity.PremiumDMTaskModeAll,
			false,
			entity.PremiumDMTaskImportModeLogical,
			true,
			"pck-1",
			config.PrecheckSkipConfigs{},
			0, 0, 0,
		)
		require.NotEmpty(t, items)
		require.Equal(t, PrimaryKeyChecking, items[0].Description)
	})
}

func TestSourcesForPremiumPrecheckItem(t *testing.T) {
	sources := []entity.PremiumDMSource{
		{ID: "mysql", SourceType: entity.PremiumDBTypeMySQL},
		{ID: "alicloud-rds", SourceType: entity.PremiumDBTypeAliCloudRDSMySQL},
		{ID: "azure", SourceType: entity.PremiumDBTypeAzureFlexibleMySQL},
	}
	passwords := []string{"mysql-password", "alicloud-rds-password", "azure-password"}

	t.Run("primary key only receives alicloud rds source and matching password", func(t *testing.T) {
		filteredSources, filteredPasswords, err := sourcesForPremiumPrecheckItem(
			sources,
			passwords,
			PrimaryKeyChecking,
		)
		require.NoError(t, err)
		require.Equal(t, []entity.PremiumDMSource{sources[1]}, filteredSources)
		require.Equal(t, []string{"alicloud-rds-password"}, filteredPasswords)
	})

	t.Run("regular checking keeps all sources", func(t *testing.T) {
		filteredSources, filteredPasswords, err := sourcesForPremiumPrecheckItem(
			sources,
			passwords,
			dmconfig.VersionChecking,
		)
		require.NoError(t, err)
		require.Equal(t, sources, filteredSources)
		require.Equal(t, passwords, filteredPasswords)
	})

	t.Run("mismatched source and password counts are rejected", func(t *testing.T) {
		filteredSources, filteredPasswords, err := sourcesForPremiumPrecheckItem(
			sources,
			passwords[:1],
			PrimaryKeyChecking,
		)
		require.Error(t, err)
		require.Nil(t, filteredSources)
		require.Nil(t, filteredPasswords)
	})
}

func TestPremiumPrecheckItemTypesForCluster_SkipConfigs(t *testing.T) {
	skipConfigs := config.PrecheckSkipConfigs{
		{
			Matcher:   config.AllowListMatcher{ClusterID: []string{"42"}},
			SkipItems: []string{dmconfig.VersionChecking, dmconfig.ServerIDChecking},
		},
	}

	types := PremiumPrecheckItemTypesForCluster(
		entity.PremiumDMTaskModeIncremental,
		false,
		entity.PremiumDMTaskImportModeLogical,
		false,
		skipConfigs,
		0, 0, 42,
	)
	containsType := func(target entity.PremiumPrecheckItemType) bool {
		for _, typ := range types {
			if typ == target {
				return true
			}
		}
		return false
	}
	require.False(t, containsType(entity.PremiumPrecheckItemTypeVersion))
	require.False(t, containsType(entity.PremiumPrecheckItemTypeServerID))
	require.True(t, containsType(entity.PremiumPrecheckItemTypeBinlogEnable))
}

func TestApplyPremiumSourcePrecheckConfig(t *testing.T) {
	cfg := &dmconfig.SubTaskConfig{
		LoaderConfig: dmconfig.DefaultLoaderConfig(),
	}
	source := &entity.PremiumDMSource{
		ConnProfile: entity.PremiumConnProfile{
			Security: &entity.PremiumSecurity{
				CertAllowedCn:  []string{"cn-a", "cn-b"},
				SslCaContent:   []byte("ca"),
				SslCertContent: []byte("cert"),
				SslKeyContent:  []byte("key"),
			},
		},
		MigrationRules: entity.PremiumMigrationRules{
			{
				Source: entity.PremiumMigrationRuleSource{
					SchemaPattern: "db1",
					TablePattern:  "tbl1",
				},
				Target: entity.PremiumMigrationRuleTarget{
					Schema: "db2",
					Table:  "tbl2",
				},
			},
		},
		BinlogName: "mysql-bin.000001",
		BinlogPos:  123,
		BinlogGTID: "gtid-set",
	}

	got := ApplyPremiumSourcePrecheckConfig(
		cfg,
		source,
		entity.PremiumDMTaskModeAll,
		entity.PremiumDMTaskImportModePhysical,
	)

	require.Same(t, cfg, got)
	require.Len(t, got.RouteRules, 2)
	require.Equal(t, "db1", got.RouteRules[0].SchemaPattern)
	require.Equal(t, "tbl1", got.RouteRules[0].TablePattern)
	require.Equal(t, "db2", got.RouteRules[0].TargetSchema)
	require.Equal(t, "tbl2", got.RouteRules[0].TargetTable)
	require.Equal(t, "db1", got.RouteRules[1].SchemaPattern)
	require.Equal(t, "", got.RouteRules[1].TablePattern)
	require.Equal(t, "db2", got.RouteRules[1].TargetSchema)
	require.Equal(t, "", got.RouteRules[1].TargetTable)
	require.NotNil(t, got.BAList)
	require.ElementsMatch(t, []string{"db1"}, got.BAList.DoDBs)
	require.Len(t, got.BAList.DoTables, 1)
	require.Equal(t, "db1", got.BAList.DoTables[0].Schema)
	require.Equal(t, "tbl1", got.BAList.DoTables[0].Name)
	require.NotNil(t, got.From.Security)
	require.Equal(t, base64.StdEncoding.EncodeToString([]byte("ca")), got.From.Security.SSLCABase64)
	require.Equal(t, base64.StdEncoding.EncodeToString([]byte("cert")), got.From.Security.SSLCertBase64)
	require.Equal(t, base64.StdEncoding.EncodeToString([]byte("key")), got.From.Security.SSLKeyBase64)
	require.Equal(t, []string{"cn-a", "cn-b"}, []string(got.From.Security.CertAllowedCN))
	require.NotNil(t, got.Meta)
	require.Equal(t, "mysql-bin.000001", got.Meta.BinLogName)
	require.Equal(t, uint32(123), got.Meta.BinLogPos)
	require.Equal(t, "gtid-set", got.Meta.BinLogGTID)
	require.True(t, got.EnableGTID)
	require.Equal(t, dmconfig.LoadModePhysical, got.LoaderConfig.ImportMode)
}

func TestBuildPremiumRouteRulesFromBlockAllowRules(t *testing.T) {
	source := &entity.PremiumDMSource{
		BlockAllowRules: entity.PremiumBlockAllowRules{
			DoDBs: []string{"db1"},
			DoTables: []entity.PremiumTable{
				{Schema: "db2", Table: "*"},
				{Schema: "db3", Table: "~.*"},
				{Schema: "db4", Table: "t4"},
			},
		},
	}

	rules := buildPremiumRouteRules(source)
	require.Len(t, rules, 7)

	require.Equal(t, "db1", rules[0].SchemaPattern)
	require.Equal(t, "*", rules[0].TablePattern)
	require.Equal(t, "db1", rules[0].TargetSchema)
	require.Equal(t, "", rules[0].TargetTable)

	require.Equal(t, "db2", rules[1].SchemaPattern)
	require.Equal(t, "*", rules[1].TablePattern)
	require.Equal(t, "db2", rules[1].TargetSchema)
	require.Equal(t, "", rules[1].TargetTable)
	require.Equal(t, "db2", rules[2].SchemaPattern)
	require.Equal(t, "", rules[2].TablePattern)
	require.Equal(t, "db2", rules[2].TargetSchema)
	require.Equal(t, "", rules[2].TargetTable)

	require.Equal(t, "db3", rules[3].SchemaPattern)
	require.Equal(t, "~.*", rules[3].TablePattern)
	require.Equal(t, "db3", rules[3].TargetSchema)
	require.Equal(t, "", rules[3].TargetTable)
	require.Equal(t, "db3", rules[4].SchemaPattern)
	require.Equal(t, "", rules[4].TablePattern)
	require.Equal(t, "db3", rules[4].TargetSchema)
	require.Equal(t, "", rules[4].TargetTable)

	require.Equal(t, "db4", rules[5].SchemaPattern)
	require.Equal(t, "t4", rules[5].TablePattern)
	require.Equal(t, "db4", rules[5].TargetSchema)
	require.Equal(t, "t4", rules[5].TargetTable)
	require.Equal(t, "db4", rules[6].SchemaPattern)
	require.Equal(t, "", rules[6].TablePattern)
	require.Equal(t, "db4", rules[6].TargetSchema)
	require.Equal(t, "", rules[6].TargetTable)
}

func TestResolveSourcePasswords(t *testing.T) {
	sources := []entity.PremiumDMSource{
		{ConnProfile: entity.PremiumConnProfile{Password: "encrypted-source-pwd-1"}},
		{ConnProfile: entity.PremiumConnProfile{Password: "encrypted-source-pwd-2"}},
	}
	enc := encryptormock.NewEncryptor(t)
	enc.On("DecryptBase64", "encrypted-source-pwd-1").Return("source-pwd-1", nil)
	enc.On("DecryptBase64", "encrypted-source-pwd-2").Return("source-pwd-2", nil)

	worker := &PremiumDMPrecheckWorker{
		taskWorker: &PremiumDMTaskWorker{Encryptor: enc},
	}
	passwords, err := worker.resolveSourcePasswords(sources)
	require.NoError(t, err)
	require.Equal(t, []string{"source-pwd-1", "source-pwd-2"}, passwords)
}

func TestResolveSourcePasswordsRejectsEmptySources(t *testing.T) {
	worker := &PremiumDMPrecheckWorker{
		taskWorker: &PremiumDMTaskWorker{},
	}
	passwords, err := worker.resolveSourcePasswords(nil)
	require.Error(t, err)
	require.Nil(t, passwords)
}

func TestUnmarshalPrecheckRequest(t *testing.T) {
	req := &v1beta2.CreatePremiumMigrationReq{
		TidbId:      "321",
		DisplayName: "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"),
				BinlogGtid: proto.String("gtid"),
				SourceType: v1beta2.PremiumDMSourceType_MYSQL,
			},
		},
	}
	data, err := protojson.Marshal(req)
	require.NoError(t, err)

	got, err := UnmarshalPrecheckRequest(data)
	require.NoError(t, err)
	require.True(t, proto.Equal(req, got))

	_, err = UnmarshalPrecheckRequest(nil)
	require.Error(t, err)
}

func TestApplyPremiumSourcePrecheckConfigNilInput(t *testing.T) {
	require.Nil(t, ApplyPremiumSourcePrecheckConfig(nil, nil, entity.PremiumDMTaskModeAll, entity.PremiumDMTaskImportModeLogical))
	require.NotNil(t, ApplyPremiumSourcePrecheckConfig(
		&dmconfig.SubTaskConfig{},
		nil,
		entity.PremiumDMTaskModeAll,
		entity.PremiumDMTaskImportModeLogical,
	))
}

func TestFilterAllExceptPremium(t *testing.T) {
	items := filterAllExceptPremium(dmconfig.VersionChecking)
	require.NotContains(t, items, dmconfig.VersionChecking)
	require.NotContains(t, items, dmconfig.AllChecking)
	require.NotEmpty(t, items)
}

func TestBuildPremiumPrecheckAgentReqExtraCheckingItems(t *testing.T) {
	baseReq := &agentpb.PrecheckDMJobReq{ExtraCheckingItems: []string{"stale"}}
	configs := []*dmconfig.SubTaskConfig{{
		LoaderConfig:   dmconfig.DefaultLoaderConfig(),
		SyncerConfig:   dmconfig.DefaultSyncerConfig(),
		MydumperConfig: dmconfig.DefaultMydumperConfig(),
	}}

	primaryReq, resultErr := buildPremiumPrecheckAgentReq(baseReq, configs, PrimaryKeyChecking)
	require.Nil(t, resultErr)
	require.Equal(t, []string{PrimaryKeyChecking}, primaryReq.ExtraCheckingItems)
	require.Equal(t, []string{"stale"}, baseReq.ExtraCheckingItems)

	regularReq, resultErr := buildPremiumPrecheckAgentReq(baseReq, configs, dmconfig.VersionChecking)
	require.Nil(t, resultErr)
	require.Empty(t, regularReq.ExtraCheckingItems)
}

func TestGeneratePremiumPrecheckItemIDUsesConsistentPrefix(t *testing.T) {
	id := generatePremiumPrecheckItemID()
	require.NotEmpty(t, id)
	require.Contains(t, id, "dmpci")
}

func newPremiumPrecheckAgentReq(precheckID string) *agentpb.PrecheckDMJobReq {
	return &agentpb.PrecheckDMJobReq{
		ReqId:               precheckID,
		FullInstanceMigrate: true,
		ClusterPassword:     "target-password",
		SourcePasswords:     []string{"source-password"},
		ClusterId:           "1",
	}
}

func TestExtractPremiumDMPrecheckUserError(t *testing.T) {
	tests := map[string]struct {
		err        error
		wantOK     bool
		wantReason string
	}{
		"ng agent checker init error response": {
			err: &agent.ErrorResponse{
				Code:    13,
				Message: "fail to initialize checker: same table name in case-insensitive [same target table `db`.`t1` vs `db`.`T1`]",
			},
			wantOK:     true,
			wantReason: "fail to initialize checker: same table name in case-insensitive [same target table `db`.`t1` vs `db`.`T1`]",
		},
		"ng agent grpc gateway error response": {
			err: &agent.ErrorResponse{
				Code:    3,
				Message: "rpc error: code = InvalidArgument desc = fail to initialize checker: invalid route rules",
			},
			wantOK:     true,
			wantReason: "fail to initialize checker: invalid route rules",
		},
		"ng agent wrapped checker error response": {
			err: &agent.ErrorResponse{
				Code:    13,
				Message: "[/app/precheck.go:1][agent.PrecheckDMJob]code=159900104,msg=precheck execution failed: invalid route rules",
			},
			wantOK:     true,
			wantReason: "precheck execution failed: invalid route rules",
		},
		"clean checker execution error": {
			err:        errors.New("precheck execution failed: invalid route rules"),
			wantOK:     true,
			wantReason: "precheck execution failed: invalid route rules",
		},
		"wrapped checker execution error stays internal": {
			err:    bizerr_wrapper.NewInternalErr("precheck execution failed: invalid route rules"),
			wantOK: false,
		},
		"transport error stays internal": {
			err:    bizerr_wrapper.NewInternalErr("context deadline exceeded"),
			wantOK: false,
		},
	}

	for name, tc := range tests {
		t.Run(name, func(t *testing.T) {
			reason, ok := extractPremiumDMPrecheckUserError(tc.err)
			require.Equal(t, tc.wantOK, ok)
			require.Equal(t, tc.wantReason, reason)
		})
	}
}

func TestExtractPremiumDMPrecheckUserErrorFromResults(t *testing.T) {
	results := &entity.PremiumResults{
		Results: []*checker.Result{
			{
				Desc:  premiumDMPrecheckInitializationDesc,
				State: checker.StateFailure,
				Errors: []*checker.Error{
					{
						ShortErr: "[code=26001:class=task-check:scope=internal:level=medium], Message: fail to initialize checker: same table name in case-insensitive [same target table `test`.`T` vs `test`.`t`], Workaround: Please check `target-table` config in task configuration file.",
					},
				},
			},
		},
	}

	reason, ok := extractPremiumDMPrecheckUserErrorFromResults(results)
	require.True(t, ok)
	require.Equal(t, "fail to initialize checker: same table name in case-insensitive [same target table `test`.`T` vs `test`.`t`]", reason)
}

func TestExtractPremiumDMPrecheckUserMessageFromResultsIgnoresStructuredNonInitFailure(t *testing.T) {
	results := &entity.PremiumResults{
		Results: []*checker.Result{
			{
				Desc:  PrimaryKeyChecking,
				State: checker.StateFailure,
				Errors: []*checker.Error{
					{
						ShortErr: "[code=26001:class=task-check:scope=internal:level=medium], Message: table `test`.`t` does not have a primary key",
					},
				},
			},
		},
	}

	message, ok := extractPremiumDMPrecheckUserMessageFromResults(results)

	require.False(t, ok)
	require.Nil(t, message)
}

func TestDoPremiumPrecheckDMJobPersistsUserFacingAgentError(t *testing.T) {
	const reason = "fail to initialize checker: same table name in case-insensitive [same target table `db`.`t1` vs `db`.`T1`]"

	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		require.Equal(t, "/api/v1/dm/jobprecheck", r.URL.Path)
		w.WriteHeader(http.StatusBadRequest)
		_, _ = w.Write([]byte(`{"code":13,"message":"` + reason + `"}`))
	}))
	defer server.Close()

	dao := &fakePremiumDMPrecheckDAO{
		listPrechecks: []*entity.PremiumDMPrecheck{
			{ID: "pck-1", TiDBID: "1", Stage: string(entity.PremiumDMPrecheckStageRunning)},
		},
	}
	worker := &PremiumDMPrecheckWorker{dao: dao}
	agentClient := &agent.NGRegionalClient{AgentUrl: server.URL}
	item := &entity.PremiumDMPrecheckItem{
		ID:          "item-1",
		Description: dmconfig.DumpPrivilegeChecking,
		Type:        entity.PremiumPrecheckItemTypeDumpPrivilege,
	}
	cfg := &dmconfig.SubTaskConfig{}
	agentReq := newPremiumPrecheckAgentReq("pck-1")

	initMessage, err := worker.doPremiumPrecheckDMJob(
		context.Background(),
		agentClient,
		"pck-1",
		zap.NewNop(),
		[]*dmconfig.SubTaskConfig{cfg},
		item.Description,
		item,
		agentReq,
	)

	require.NoError(t, err)
	require.NotNil(t, initMessage)
	require.Equal(t, reason, initMessage.Reason)
	require.Empty(t, initMessage.Solution)
	require.Len(t, dao.updatedItems, 1)
	require.Equal(t, string(entity.PremiumDMPrecheckStageFinished), dao.updatedItems[0]["stage"])
	results := dao.updatedItems[0]["results"].(entity.PremiumResults)
	require.Len(t, results.Results, 1)
	require.Equal(t, checker.StateFailure, results.Results[0].State)
	require.Equal(t, reason, results.Results[0].Errors[0].ShortErr)
}

func TestDoPremiumPrecheckDMJobPersistsUserFacingCheckResults(t *testing.T) {
	const reason = "fail to initialize checker: invalid route rules"

	resultBytes, err := json.Marshal(buildPremiumDMPrecheckFailureResults(
		premiumDMPrecheckInitializationDesc,
		reason,
		"",
	))
	require.NoError(t, err)
	respBytes, err := protojson.Marshal(&agentpb.PrecheckDMJobResp{CheckResults: resultBytes})
	require.NoError(t, err)

	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		require.Equal(t, "/api/v1/dm/jobprecheck", r.URL.Path)
		w.WriteHeader(http.StatusOK)
		_, _ = w.Write(respBytes)
	}))
	defer server.Close()

	dao := &fakePremiumDMPrecheckDAO{
		listPrechecks: []*entity.PremiumDMPrecheck{
			{ID: "pck-1", TiDBID: "1", Stage: string(entity.PremiumDMPrecheckStageRunning)},
		},
	}
	worker := &PremiumDMPrecheckWorker{dao: dao}
	agentClient := &agent.NGRegionalClient{AgentUrl: server.URL}
	item := &entity.PremiumDMPrecheckItem{
		ID:          "item-1",
		Description: dmconfig.DumpPrivilegeChecking,
		Type:        entity.PremiumPrecheckItemTypeDumpPrivilege,
	}
	cfg := &dmconfig.SubTaskConfig{}
	agentReq := newPremiumPrecheckAgentReq("pck-1")

	initMessage, bizErr := worker.doPremiumPrecheckDMJob(
		context.Background(),
		agentClient,
		"pck-1",
		zap.NewNop(),
		[]*dmconfig.SubTaskConfig{cfg},
		item.Description,
		item,
		agentReq,
	)

	require.NoError(t, bizErr)
	require.NotNil(t, initMessage)
	require.Equal(t, reason, initMessage.Reason)
	require.Empty(t, initMessage.Solution)
	require.Len(t, dao.updatedItems, 1)
	require.Equal(t, string(entity.PremiumDMPrecheckStageFinished), dao.updatedItems[0]["stage"])
	results := dao.updatedItems[0]["results"].(entity.PremiumResults)
	require.Len(t, results.Results, 1)
	require.Equal(t, dmconfig.DumpPrivilegeChecking, results.Results[0].Desc)
	require.Equal(t, checker.StateFailure, results.Results[0].State)
	require.Equal(t, reason, results.Results[0].Errors[0].ShortErr)
}

func TestDoPremiumPrecheckDMJobNormalizesUserFacingCheckResultReason(t *testing.T) {
	const rawReason = "[code=11043:class=functional:scope=internal:level=high], Message: fail to initialize checker: generate table router, RawCause: table test.T matches more than one rule, Workaround: Please check `routes` config in task configuration file."
	const normalizedReason = "fail to initialize checker: generate table router, RawCause: table test.T matches more than one rule"
	const normalizedSolution = "Please check `routes` config in task configuration file."

	resultBytes, err := json.Marshal(buildPremiumDMPrecheckFailureResults(
		premiumDMPrecheckInitializationDesc,
		rawReason,
		"",
	))
	require.NoError(t, err)
	respBytes, err := protojson.Marshal(&agentpb.PrecheckDMJobResp{CheckResults: resultBytes})
	require.NoError(t, err)

	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		require.Equal(t, "/api/v1/dm/jobprecheck", r.URL.Path)
		w.WriteHeader(http.StatusOK)
		_, _ = w.Write(respBytes)
	}))
	defer server.Close()

	dao := &fakePremiumDMPrecheckDAO{
		listPrechecks: []*entity.PremiumDMPrecheck{
			{ID: "pck-1", TiDBID: "1", Stage: string(entity.PremiumDMPrecheckStageRunning)},
		},
	}
	worker := &PremiumDMPrecheckWorker{dao: dao}
	agentClient := &agent.NGRegionalClient{AgentUrl: server.URL}
	item := &entity.PremiumDMPrecheckItem{
		ID:          "item-1",
		Description: dmconfig.DumpPrivilegeChecking,
		Type:        entity.PremiumPrecheckItemTypeDumpPrivilege,
	}

	initMessage, bizErr := worker.doPremiumPrecheckDMJob(
		context.Background(),
		agentClient,
		"pck-1",
		zap.NewNop(),
		[]*dmconfig.SubTaskConfig{{}},
		item.Description,
		item,
		newPremiumPrecheckAgentReq("pck-1"),
	)

	require.NoError(t, bizErr)
	require.NotNil(t, initMessage)
	require.Equal(t, normalizedReason, initMessage.Reason)
	require.Equal(t, normalizedSolution, initMessage.Solution)
	results := dao.updatedItems[0]["results"].(entity.PremiumResults)
	require.Equal(t, normalizedReason, results.Results[0].Errors[0].ShortErr)
	require.Equal(t, normalizedSolution, results.Results[0].Instruction)
}

func TestDecodePremiumPrecheckResultsSkipsPrimaryKeyInstructionForInitFailure(t *testing.T) {
	const reason = "fail to initialize checker: invalid route rules"

	resultBytes, err := json.Marshal(buildPremiumDMPrecheckFailureResults(
		premiumDMPrecheckInitializationDesc,
		reason,
		"",
	))
	require.NoError(t, err)

	results, resultErr := decodePremiumPrecheckResults(resultBytes, PrimaryKeyChecking)

	require.Nil(t, resultErr)
	require.Len(t, results.Results, 1)
	require.Equal(t, PrimaryKeyChecking, results.Results[0].Desc)
	require.Equal(t, checker.StateFailure, results.Results[0].State)
	require.Equal(t, reason, results.Results[0].Errors[0].ShortErr)
	require.Empty(t, results.Results[0].Instruction)
}

func TestRunPrecheckSkipsRemainingItemsAfterUserFacingInitFailure(t *testing.T) {
	const reason = "fail to initialize checker: same table name in case-insensitive [same target table `db`.`t1` vs `db`.`T1`]"

	dao := &fakePremiumDMPrecheckDAO{
		listPrechecks: []*entity.PremiumDMPrecheck{
			{ID: "pck-1", TiDBID: "cluster-1", Stage: string(entity.PremiumDMPrecheckStagePending), TotalItems: 2},
		},
	}
	worker := &PremiumDMPrecheckWorker{dao: dao}
	items := []*entity.PremiumDMPrecheckItem{
		{ID: "item-1", Description: dmconfig.DumpPrivilegeChecking, Type: entity.PremiumPrecheckItemTypeDumpPrivilege},
		{ID: "item-2", Description: dmconfig.VersionChecking, Type: entity.PremiumPrecheckItemTypeVersion},
	}
	var runnerCalls int

	worker.RunPrecheck(context.Background(), "cluster-1", "pck-1", items,
		func(
			ctx context.Context,
			precheckID string,
			item *entity.PremiumDMPrecheckItem,
		) (*premiumDMPrecheckUserMessage, bizerr.BizError) {
			runnerCalls++
			require.Equal(t, "item-1", item.ID)
			item.DMPrecheckID = precheckID
			item.Results = buildPremiumDMPrecheckFailureResults(item.Description, reason, "")
			item.Stage = string(entity.PremiumDMPrecheckStageFinished)
			require.NoError(t, dao.CreatePrecheckItem(ctx, item))
			require.NoError(t, dao.UpdatePrecheckItem(ctx, item.ID, map[string]any{
				"stage":   item.Stage,
				"err_msg": item.ErrMsg,
				"results": item.Results,
			}))
			return &premiumDMPrecheckUserMessage{Reason: reason}, nil
		})

	require.Equal(t, 1, runnerCalls)
	require.Len(t, dao.createdItems, 2)
	require.Len(t, dao.updatedItems, 2)
	finalPrecheck := dao.listPrechecks[0]
	require.Equal(t, string(entity.PremiumDMPrecheckStageFailed), finalPrecheck.Stage)
	require.Len(t, finalPrecheck.DMPrecheckItems, 2)
	for i := range finalPrecheck.DMPrecheckItems {
		item := &finalPrecheck.DMPrecheckItems[i]
		require.Equal(t, string(entity.PremiumDMPrecheckStageFinished), item.Stage)
		summary, ok := item.SummarizeResults()
		require.True(t, ok)
		require.Equal(t, checker.StateFailure, summary.State)
		if item.ID == "item-1" {
			require.Contains(t, summary.Reasons, reason)
			continue
		}
		require.Contains(t, summary.Reasons,
			"This check was not executed because precheck initialization failed: "+reason)
	}
}

func TestResolveFinalPrecheckStatus(t *testing.T) {
	t.Run("internal runner error wins", func(t *testing.T) {
		stage, errMsg := resolveFinalPrecheckStatus(&entity.PremiumDMPrecheck{}, 1, "runner failed")
		require.Equal(t, entity.PremiumDMPrecheckStageFailed, stage)
		require.Equal(t, "runner failed", errMsg)
	})

	t.Run("failed item stage marks precheck failed", func(t *testing.T) {
		stage, errMsg := resolveFinalPrecheckStatus(&entity.PremiumDMPrecheck{
			DMPrecheckItems: []entity.PremiumDMPrecheckItem{
				{Stage: string(entity.PremiumDMPrecheckStageFailed), ErrMsg: "item failed"},
			},
		}, 0, "")
		require.Equal(t, entity.PremiumDMPrecheckStageFailed, stage)
		require.Equal(t, "item failed", errMsg)
	})

	t.Run("failed checker result marks precheck failed", func(t *testing.T) {
		stage, errMsg := resolveFinalPrecheckStatus(&entity.PremiumDMPrecheck{
			DMPrecheckItems: []entity.PremiumDMPrecheckItem{
				{
					Stage: string(entity.PremiumDMPrecheckStageFinished),
					Results: entity.PremiumResults{
						Results: []*checker.Result{
							{State: "success"},
							{State: checker.StateFailure},
						},
					},
				},
			},
		}, 0, "")
		require.Equal(t, entity.PremiumDMPrecheckStageFailed, stage)
		require.Empty(t, errMsg)
	})

	t.Run("warnings without failures stay finished", func(t *testing.T) {
		stage, errMsg := resolveFinalPrecheckStatus(&entity.PremiumDMPrecheck{
			DMPrecheckItems: []entity.PremiumDMPrecheckItem{
				{
					Stage: string(entity.PremiumDMPrecheckStageFinished),
					Results: entity.PremiumResults{
						Results: []*checker.Result{
							{State: "success"},
							{State: "warning"},
						},
					},
				},
			},
		}, 0, "")
		require.Equal(t, entity.PremiumDMPrecheckStageFinished, stage)
		require.Empty(t, errMsg)
	})
}

func TestConvertRequestSourcesToEntities(t *testing.T) {
	t.Run("PUBLIC source with nil endpoint map", func(t *testing.T) {
		sources := ConvertRequestSourcesToEntities(
			"3",
			"tenant-1",
			[]*v1beta2.PremiumDMSource{
				{
					ConnProfile: &v1beta2.PremiumDMConnProfile{
						ConnType: v1beta2.PremiumDMConnType_PUBLIC,
						Endpoint: &v1beta2.PremiumDMConnProfile_Host{Host: "127.0.0.1"},
						Port:     3306,
						User:     "root",
						Password: "pwd",
					},
					PremiumMigrationRules: []*v1beta2.PremiumMigrationRule{
						{
							SourceTable: &v1beta2.PremiumMigrationRule_Source{
								SchemaPattern: "db1",
								TablePattern:  "tbl1",
							},
							TargetTable: &v1beta2.PremiumMigrationRule_Table{
								Schema: "db2",
								Table:  "tbl2",
							},
						},
					},
					BinlogName: proto.String("mysql-bin.000001"),
					BinlogGtid: proto.String("gtid"),
					SourceType: v1beta2.PremiumDMSourceType_MYSQL,
				},
			},
			nil, // no PRIVATELINK endpoints
		)

		require.Len(t, sources, 1)
		require.Equal(t, "3", sources[0].TiDBID)
		require.Equal(t, "tenant-1", sources[0].TenantID)
		require.Equal(t, entity.PremiumDBTypeMySQL, sources[0].SourceType)
		require.Equal(t, "127.0.0.1", sources[0].ConnProfile.Target)
		require.Equal(t, "root", sources[0].ConnProfile.User)
		require.Equal(t, "pwd", sources[0].ConnProfile.Password)
		require.Equal(t, "mysql-bin.000001", sources[0].BinlogName)
		require.Equal(t, "gtid", sources[0].BinlogGTID)
		require.Len(t, sources[0].MigrationRules, 1)
	})

	t.Run("PRIVATELINK source resolves endpoint address from map", func(t *testing.T) {
		endpointMap := map[uint64]*entity.BizSinkEndpoint{
			100: {
				EndpointAddress: "vpce-abc.rds.amazonaws.com",
				Status:          entity.BizPLEndpointStatusAvailable,
			},
		}
		endpointMap[100].ID = 100

		sources := ConvertRequestSourcesToEntities(
			"5",
			"tenant-2",
			[]*v1beta2.PremiumDMSource{
				{
					ConnProfile: &v1beta2.PremiumDMConnProfile{
						ConnType: v1beta2.PremiumDMConnType_PRIVATELINK,
						Endpoint: &v1beta2.PremiumDMConnProfile_SinkEndpointId{SinkEndpointId: 100},
						Port:     3306,
						User:     "admin",
						Password: "secret",
					},
					SourceType: v1beta2.PremiumDMSourceType_MYSQL,
				},
			},
			endpointMap,
		)

		require.Len(t, sources, 1)
		require.Equal(t, entity.PremiumConnTypePrivateLink, sources[0].ConnProfile.Type)
		require.Equal(t, "vpce-abc.rds.amazonaws.com", sources[0].ConnProfile.Target)
		require.Equal(t, uint64(100), sources[0].ConnProfile.SinkEndpointID)
		require.Equal(t, "admin", sources[0].ConnProfile.User)
	})

	t.Run("PRIVATELINK source with missing endpoint in map", func(t *testing.T) {
		endpointMap := map[uint64]*entity.BizSinkEndpoint{
			200: {
				EndpointAddress: "vpce-other.rds.amazonaws.com",
			},
		}
		endpointMap[200].ID = 200

		sources := ConvertRequestSourcesToEntities(
			"5",
			"tenant-2",
			[]*v1beta2.PremiumDMSource{
				{
					ConnProfile: &v1beta2.PremiumDMConnProfile{
						ConnType: v1beta2.PremiumDMConnType_PRIVATELINK,
						Endpoint: &v1beta2.PremiumDMConnProfile_SinkEndpointId{SinkEndpointId: 999},
						Port:     3306,
						User:     "admin",
						Password: "secret",
					},
					SourceType: v1beta2.PremiumDMSourceType_MYSQL,
				},
			},
			endpointMap,
		)

		require.Len(t, sources, 1)
		require.Equal(t, entity.PremiumConnTypePrivateLink, sources[0].ConnProfile.Type)
		require.Equal(t, "", sources[0].ConnProfile.Target, "missing endpoint should result in empty target")
		require.Equal(t, uint64(999), sources[0].ConnProfile.SinkEndpointID)
	})
}

func TestConvertRequestSourcesToEntitiesAzureSourceType(t *testing.T) {
	sources := ConvertRequestSourcesToEntities(
		"3",
		"tenant-1",
		[]*v1beta2.PremiumDMSource{{
			ConnProfile: &v1beta2.PremiumDMConnProfile{
				ConnType: v1beta2.PremiumDMConnType_PUBLIC,
				Endpoint: &v1beta2.PremiumDMConnProfile_Host{Host: "127.0.0.1"},
				Port:     3306,
				User:     "root",
				Password: "pwd",
			},
			SourceType: v1beta2.PremiumDMSourceType_AZURE_FLEXIBLE_MYSQL,
		}},
		nil,
	)

	require.Len(t, sources, 1)
	require.Equal(t, entity.PremiumDBTypeAzureFlexibleMySQL, sources[0].SourceType)
}

func TestConvertRequestSourcesToEntitiesAliCloudRDSSourceType(t *testing.T) {
	sources := ConvertRequestSourcesToEntities(
		"3",
		"tenant-1",
		[]*v1beta2.PremiumDMSource{{
			ConnProfile: &v1beta2.PremiumDMConnProfile{
				ConnType: v1beta2.PremiumDMConnType_PUBLIC,
				Endpoint: &v1beta2.PremiumDMConnProfile_Host{Host: "127.0.0.1"},
				Port:     3306,
				User:     "root",
				Password: "pwd",
			},
			SourceType: v1beta2.PremiumDMSourceType_ALICLOUD_RDS_MYSQL,
		}},
		nil,
	)

	require.Len(t, sources, 1)
	require.Equal(t, entity.PremiumDBTypeAliCloudRDSMySQL, sources[0].SourceType)
	require.True(t, hasAliCloudRDSSource(sources))
}

func TestBuildSharedPoolDataDir(t *testing.T) {
	tests := map[string]struct {
		task     *entity.PremiumDMTask
		storage  *agent.SharedPool_ObjectStorage
		provider *agent.Provider
		expected string
		wantErr  string
	}{
		"aws keeps s3 scheme and prefix": {
			task:     &entity.PremiumDMTask{TiDBID: "1", ID: "task1", CloudProvider: string(entity.CloudProviderTypeAWS)},
			storage:  &agent.SharedPool_ObjectStorage{Bucket: "bucket", Prefix: "internal/path"},
			expected: "s3://bucket/internal/path/1/task1",
		},
		"gcp uses gcs scheme": {
			task:     &entity.PremiumDMTask{TiDBID: "1", ID: "task1", CloudProvider: string(entity.CloudProviderTypeGCP)},
			storage:  &agent.SharedPool_ObjectStorage{Bucket: "bucket", Prefix: "internal/path"},
			expected: "gcs://bucket/internal/path/1/task1",
		},
		"alicloud uses s3-compatible scheme before OSS query decoration": {
			task:     &entity.PremiumDMTask{TiDBID: "1", ID: "task1", CloudProvider: string(entity.CloudProviderTypeAliCloud)},
			storage:  &agent.SharedPool_ObjectStorage{Bucket: "bucket", Prefix: "internal/path"},
			expected: "s3://bucket/internal/path/1/task1",
		},
		"azure decodes account and container from bucket and uses node client id": {
			task:     &entity.PremiumDMTask{TiDBID: "1", ID: "task1", CloudProvider: string(entity.CloudProviderTypeAzure)},
			storage:  &agent.SharedPool_ObjectStorage{Bucket: "acct/internal-container", Prefix: "internal/path"},
			provider: &agent.Provider{NodeClientID: "node-client-id"},
			expected: "azure://internal-container/internal/path/1/task1?account-name=acct&client-id=node-client-id",
		},
		"azure keeps explicit account key when present": {
			task:     &entity.PremiumDMTask{TiDBID: "1", ID: "task1", CloudProvider: string(entity.CloudProviderTypeAzure)},
			storage:  &agent.SharedPool_ObjectStorage{Bucket: "bucket", Prefix: "internal/path", AccountName: "acct", AccountKey: "key-1"},
			expected: "azure://bucket/internal/path/1/task1?account-key=key-1&account-name=acct",
		},
		"azure missing identity and credentials fails": {
			task:    &entity.PremiumDMTask{TiDBID: "1", ID: "task1", CloudProvider: string(entity.CloudProviderTypeAzure)},
			storage: &agent.SharedPool_ObjectStorage{Bucket: "acct/internal-container", Prefix: "internal/path"},
			wantErr: "azure internal storage credentials are empty",
		},
		"azure malformed bucket fails": {
			task:    &entity.PremiumDMTask{TiDBID: "1", ID: "task1", CloudProvider: string(entity.CloudProviderTypeAzure)},
			storage: &agent.SharedPool_ObjectStorage{Bucket: "internal-container-only", Prefix: "internal/path"},
			wantErr: "azure internal storage account name is empty",
		},
		"empty provider falls back to s3": {
			task:     &entity.PremiumDMTask{TiDBID: "1", ID: "task1", CloudProvider: ""},
			storage:  &agent.SharedPool_ObjectStorage{Bucket: "bucket", Prefix: "path"},
			expected: "s3://bucket/path/1/task1",
		},
		"unsupported provider returns error": {
			task:    &entity.PremiumDMTask{TiDBID: "1", ID: "task1", CloudProvider: "unsupported"},
			storage: &agent.SharedPool_ObjectStorage{Bucket: "bucket", Prefix: "path"},
			wantErr: "unsupported cloud provider unsupported",
		},
	}

	for name, tc := range tests {
		t.Run(name, func(t *testing.T) {
			got, err := buildSharedPoolDataDir(tc.task, tc.storage, tc.provider)
			if tc.wantErr != "" {
				require.EqualError(t, err, tc.wantErr)
				return
			}
			require.NoError(t, err)
			require.Equal(t, tc.expected, got)
		})
	}
}

func TestUnmarshalPrecheckRequestInvalidJSON(t *testing.T) {
	_, err := UnmarshalPrecheckRequest([]byte(`{"tidb_id":`))
	require.Error(t, err)
}

func TestHasAnySourceBinlogCheckpoint(t *testing.T) {
	require.False(t, hasAnySourceBinlogCheckpoint(nil))
	require.False(t, hasAnySourceBinlogCheckpoint([]entity.PremiumDMSource{{ID: "s1"}}))
	require.True(t, hasAnySourceBinlogCheckpoint([]entity.PremiumDMSource{{ID: "s1", BinlogName: "mysql-bin.000001"}}))
	require.True(t, hasAnySourceBinlogCheckpoint([]entity.PremiumDMSource{{ID: "s1", BinlogGTID: "gtid-set"}}))
}

func TestBuildPremiumPrecheckBaseConfigLeavesTargetHostEmpty(t *testing.T) {
	cluster := &tidbmgmtservicev1.Cluster{
		ClusterId:     42,
		ProjectId:     100,
		CloudProvider: "aws",
	}
	req := &v1beta2.CreatePremiumMigrationReq{
		Target: &v1beta2.PremiumDMTarget{
			User: "target-user",
		},
	}
	source := &entity.PremiumDMSource{
		ConnProfile: entity.PremiumConnProfile{
			Target: "source-host",
			Port:   3306,
			User:   "source-user",
		},
	}

	cfg, err := BuildPremiumPrecheckBaseConfig(cluster, req, source)
	require.NoError(t, err)
	require.NotNil(t, cfg)
	require.Equal(t, "", cfg.To.Host)
	require.Equal(t, 4000, cfg.To.Port)
	require.Equal(t, "target-user", cfg.To.User)
}