package dm

import (
	"testing"
	"time"

	"github.com/pingcap/errors"
	dmconfig "github.com/pingcap/tiflow/dm/config"
	"github.com/pingcap/tiflow/dm/pkg/checker"
	engineerrors "github.com/pingcap/tiflow/pkg/errors"
	"github.com/stretchr/testify/require"
	dmpb "github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dm"
	commondm "github.com/tidbcloud/dataflow-service/internal/common/dm"
	"github.com/tidbcloud/dataflow-service/internal/config"
	"github.com/tidbcloud/dataflow-service/internal/db/entity"
	pimport "github.com/tidbcloud/dataflow-service/internal/model/import"
	"k8s.io/utils/pointer"
)

func TestModelIsJobNotFoundErr(t *testing.T) {
	require.True(t, (&Model{}).isJobNotFoundErr(errors.Errorf("code: %s, msg: %s",
		engineerrors.ErrJobNotFound.ID(), engineerrors.ErrJobNotFound.GetMsg())))
	require.False(t, (&Model{}).isJobNotFoundErr(errors.New("invalid msg")))
}

func TestPrecheckHelpUrls(t *testing.T) {
	require.Len(t, precheckItemHelpUrls, len(allModeCheckList)+1) // meta position only exists in incr mode
	for _, item := range allModeCheckList {
		require.NotZero(t, len(precheckItemHelpUrls[item]))
	}
	for _, item := range fullModeCheckList {
		require.NotZero(t, len(precheckItemHelpUrls[item]))
	}
	for _, item := range incrModeCheckList {
		require.NotZero(t, len(precheckItemHelpUrls[item]))
	}
}

func TestOverwritePrecheckResult(t *testing.T) {
	cases := []struct {
		checkItem   string
		result      *entity.DMPrecheckResult
		expected    *entity.DMPrecheckResult
		errContains string
	}{
		{
			checkItem: dmconfig.TargetDBPrivilegeChecking,
			result: &entity.DMPrecheckResult{
				Results: []*checker.Result{},
			},
			expected:    nil,
			errContains: "unexpected check result",
		},
		{
			checkItem: dmconfig.TargetDBPrivilegeChecking,
			result: &entity.DMPrecheckResult{
				Results: []*checker.Result{
					nil,
				},
			},
			expected:    nil,
			errContains: "unexpected check result:[<nil>]",
		},
		{
			checkItem: dmconfig.TargetDBPrivilegeChecking,
			result: &entity.DMPrecheckResult{
				Results: []*checker.Result{
					{
						ID:    1,
						State: "warn",
					},
				},
			},
			expected: &entity.DMPrecheckResult{
				Results: []*checker.Result{
					{
						ID:    1,
						State: "fail",
					},
				},
			},
			errContains: "",
		},
		{
			checkItem: dmconfig.ConnNumberChecking,
			result: &entity.DMPrecheckResult{
				Results: []*checker.Result{
					{
						ID:    1,
						State: "warn",
					},
				},
			},
			expected: &entity.DMPrecheckResult{
				Results: []*checker.Result{
					{
						ID:    1,
						State: "warn",
					},
				},
			},
			errContains: "",
		},
	}

	for _, cs := range cases {
		err := overwritePrecheckResult(cs.checkItem, cs.result)
		if cs.errContains != "" {
			require.Contains(t, err.Error(), cs.errContains)
		} else {
			require.Equal(t, cs.expected, cs.result)
		}
	}
}

func TestDmpbSource2DMSource(t *testing.T) {
	filterRules := &entity.FilterRules{
		IgnoreDBs: commondm.DefaultIgnoreDBs(),
	}
	testCases := []struct {
		mode     string
		source   *dmpb.Source
		expected entity.DMJobSource
	}{
		{
			mode: dmconfig.ModeAll,
			source: &dmpb.Source{
				BinlogName: pointer.String("mysql_bin.000023"),
				BinlogPos:  pointer.Uint32(1668),
				BinlogGtid: pointer.String("84540389-cd7a-11ed-ad05-06e2c5168bf7:1-108"),
			},
			expected: entity.DMJobSource{
				Gtid:       false,
				BinlogName: "mysql_bin.000023",
				BinlogPos:  1668,
				BinlogGtid: "84540389-cd7a-11ed-ad05-06e2c5168bf7:1-108",
				BaRules:    filterRules,
			},
		},
		{
			mode: dmconfig.ModeFull,
			source: &dmpb.Source{
				BinlogName: pointer.String("mysql_bin.000023"),
				BinlogPos:  pointer.Uint32(1668),
				BinlogGtid: pointer.String("84540389-cd7a-11ed-ad05-06e2c5168bf7:1-108"),
			},
			expected: entity.DMJobSource{
				Gtid:       false,
				BinlogName: "mysql_bin.000023",
				BinlogPos:  1668,
				BinlogGtid: "84540389-cd7a-11ed-ad05-06e2c5168bf7:1-108",
				BaRules:    filterRules,
			},
		},
		{
			mode: dmconfig.ModeIncrement,
			source: &dmpb.Source{
				BinlogName: pointer.String("mysql_bin.000023"),
				BinlogPos:  pointer.Uint32(1668),
				BinlogGtid: pointer.String("84540389-cd7a-11ed-ad05-06e2c5168bf7:1-108"),
			},
			expected: entity.DMJobSource{
				Gtid:       true,
				BinlogName: "mysql_bin.000023",
				BinlogPos:  1668,
				BinlogGtid: "84540389-cd7a-11ed-ad05-06e2c5168bf7:1-108",
				BaRules:    filterRules,
			},
		},
		{
			mode: dmconfig.ModeIncrement,
			source: &dmpb.Source{
				BinlogName: pointer.String("mysql_bin.000023"),
				BinlogPos:  pointer.Uint32(1668),
				BinlogGtid: pointer.String(""),
			},
			expected: entity.DMJobSource{
				Gtid:       false,
				BinlogName: "mysql_bin.000023",
				BinlogPos:  1668,
				BinlogGtid: "",
				BaRules:    filterRules,
			},
		},
		{
			mode: dmconfig.ModeIncrement,
			source: &dmpb.Source{
				BinlogName: nil,
				BinlogPos:  nil,
				BinlogGtid: nil,
			},
			expected: entity.DMJobSource{
				Gtid:       false,
				BinlogName: "",
				BinlogPos:  0,
				BinlogGtid: "",
				BaRules:    filterRules,
			},
		},
	}
	for _, testCase := range testCases {
		actual := dmpbSource2DMSource(testCase.mode, testCase.source)
		require.Equal(t, testCase.expected, actual)
	}
}

func TestImpTask2LoadStatus(t *testing.T) {
	currentTime := time.Now()
	currentTimeMinus10s := currentTime.Add(-10 * time.Second)
	testCases := []struct {
		expected   *entity.DMLoadStatus
		importTask *pimport.ImportTask
		jobSource  entity.DMJobSource
	}{
		{
			&entity.DMLoadStatus{
				FinishedBytes:  10,
				TotalBytes:     20,
				Progress:       "25.00",
				MetaBinlog:     "(mysql_bin.000023, 1668)",
				MetaBinlogGTID: "84540389-cd7a-11ed-ad05-06e2c5168bf7:1-108",
				Bps:            0,
				Duration:       0,
			},
			&pimport.ImportTask{
				Import: entity.Import{
					ProcessedDataSize:          pointer.Uint64(10),
					TotalSize:                  pointer.Uint64(20),
					CompletedPercent:           50,
					PostImportCompletedPercent: 0,
				},
			},
			entity.DMJobSource{
				BinlogName: "mysql_bin.000023",
				BinlogPos:  1668,
				BinlogGtid: "84540389-cd7a-11ed-ad05-06e2c5168bf7:1-108",
			},
		},
		{
			&entity.DMLoadStatus{
				FinishedBytes:  1000000,
				TotalBytes:     1000000,
				Progress:       "90.00",
				MetaBinlog:     "(mysql_bin.000023, 1668)",
				MetaBinlogGTID: "84540389-cd7a-11ed-ad05-06e2c5168bf7:1-108",
				Bps:            100000,
				Duration:       10,
			},
			&pimport.ImportTask{
				Import: entity.Import{
					ProcessedDataSize:          pointer.Uint64(1000000),
					TotalSize:                  pointer.Uint64(1000000),
					CompletedPercent:           100,
					PostImportCompletedPercent: 80,
					StartedAt:                  &currentTimeMinus10s,
					FinishedAt:                 &currentTime,
				},
			},
			entity.DMJobSource{
				BinlogName: "mysql_bin.000023",
				BinlogPos:  1668,
				BinlogGtid: "84540389-cd7a-11ed-ad05-06e2c5168bf7:1-108",
			},
		},
	}
	for _, testCase := range testCases {
		actual := impTask2LoadStatus(testCase.importTask, testCase.jobSource)
		require.Equal(t, testCase.expected, actual)
	}
}

func Test_parseLoadBinlog(t *testing.T) {
	type args struct {
		binlog string
	}
	tests := []struct {
		name           string
		args           args
		wantBinlogName string
		wantBinlogPos  int
		wantErr        bool
	}{
		{
			name: "OK",
			args: args{
				binlog: "(mysql-bin.9999999, 1398)",
			},
			wantBinlogName: "mysql-bin.9999999",
			wantBinlogPos:  1398,
			wantErr:        false,
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			gotBinlogName, gotBinlogPos, err := parseLoadBinlog(tt.args.binlog)
			if (err != nil) != tt.wantErr {
				t.Errorf("parseLoadBinlog() error = %v, wantErr %v", err, tt.wantErr)
				return
			}
			if gotBinlogName != tt.wantBinlogName {
				t.Errorf("parseLoadBinlog() gotBinlogName = %v, want %v", gotBinlogName, tt.wantBinlogName)
			}
			if gotBinlogPos != tt.wantBinlogPos {
				t.Errorf("parseLoadBinlog() gotBinlogPos = %v, want %v", gotBinlogPos, tt.wantBinlogPos)
			}
		})
	}
}

func TestGetCheckItems(t *testing.T) {
	// helper to build a minimal CreateJobReq
	makeReq := func(mode, importMode string, orgID, projectID, clusterID uint64, binlogGtid, binlogName string) *dmpb.CreateJobReq {
		src := &dmpb.Source{}
		if binlogGtid != "" {
			src.BinlogGtid = pointer.String(binlogGtid)
		}
		if binlogName != "" {
			src.BinlogName = pointer.String(binlogName)
		}
		return &dmpb.CreateJobReq{
			Mode:       mode,
			ImportMode: importMode,
			OrgId:      orgID,
			ProjectId:  projectID,
			ClusterId:  clusterID,
			Sources:    []*dmpb.Source{src},
		}
	}

	tests := map[string]struct {
		skipConfigs             config.PrecheckSkipConfigs
		enableLightningPrecheck bool
		req                     *dmpb.CreateJobReq
		expectedContains        []string
		expectedNotContains     []string
	}{
		"full mode: standard items returned": {
			req:                 makeReq(dmconfig.ModeFull, "", 1, 2, 3, "", ""),
			expectedContains:    []string{dmconfig.DumpPrivilegeChecking, dmconfig.VersionChecking},
			expectedNotContains: []string{dmconfig.ReplicationPrivilegeChecking},
		},
		"incr mode with gtid: meta position included": {
			req:              makeReq(dmconfig.ModeIncrement, "", 1, 2, 3, "abc:1-10", ""),
			expectedContains: []string{dmconfig.MetaPositionChecking},
		},
		"incr mode without gtid or binlog name: meta position excluded": {
			req:                 makeReq(dmconfig.ModeIncrement, "", 1, 2, 3, "", ""),
			expectedNotContains: []string{dmconfig.MetaPositionChecking},
		},
		"all mode physical with lightning precheck: lightning table empty included": {
			enableLightningPrecheck: true,
			req:                     makeReq(dmconfig.ModeAll, string(entity.DMImportModePhysical), 1, 2, 3, "", ""),
			expectedContains:        []string{dmconfig.LightningTableEmptyChecking},
		},
		"all mode logical: lightning table empty excluded": {
			enableLightningPrecheck: true,
			req:                     makeReq(dmconfig.ModeAll, string(entity.DMImportModeLogical), 1, 2, 3, "", ""),
			expectedNotContains:     []string{dmconfig.LightningTableEmptyChecking},
		},
		"blacklist by tenantID skips item": {
			skipConfigs: config.PrecheckSkipConfigs{
				{
					Matcher:   config.AllowListMatcher{TenantID: []string{"10"}},
					SkipItems: []string{dmconfig.VersionChecking},
				},
			},
			req:                 makeReq(dmconfig.ModeFull, "", 10, 2, 3, "", ""),
			expectedNotContains: []string{dmconfig.VersionChecking},
			expectedContains:    []string{dmconfig.DumpPrivilegeChecking},
		},
		"blacklist by projectID skips item": {
			skipConfigs: config.PrecheckSkipConfigs{
				{
					Matcher:   config.AllowListMatcher{ProjectID: []string{"20"}},
					SkipItems: []string{dmconfig.ConnNumberChecking},
				},
			},
			req:                 makeReq(dmconfig.ModeFull, "", 1, 20, 3, "", ""),
			expectedNotContains: []string{dmconfig.ConnNumberChecking},
		},
		"blacklist by clusterID skips item": {
			skipConfigs: config.PrecheckSkipConfigs{
				{
					Matcher:   config.AllowListMatcher{ClusterID: []string{"30"}},
					SkipItems: []string{dmconfig.TableSchemaChecking},
				},
			},
			req:                 makeReq(dmconfig.ModeFull, "", 1, 2, 30, "", ""),
			expectedNotContains: []string{dmconfig.TableSchemaChecking},
		},
		"blacklist does not affect non-matching cluster": {
			skipConfigs: config.PrecheckSkipConfigs{
				{
					Matcher:   config.AllowListMatcher{ClusterID: []string{"99"}},
					SkipItems: []string{dmconfig.VersionChecking},
				},
			},
			req:              makeReq(dmconfig.ModeFull, "", 1, 2, 3, "", ""),
			expectedContains: []string{dmconfig.VersionChecking},
		},
	}

	for name, tc := range tests {
		t.Run(name, func(t *testing.T) {
			// Save and restore DMConfig after each subtest.
			orig := config.Global().DMConfig
			defer func() { config.Global().DMConfig = orig }()

			config.Global().DMConfig.PrecheckSkipConfigs = tc.skipConfigs
			config.Global().DMConfig.EnableLightningPrecheck = tc.enableLightningPrecheck

			got := getCheckItems(tc.req)
			for _, item := range tc.expectedContains {
				require.Contains(t, got, item, "expected %q in result", item)
			}
			for _, item := range tc.expectedNotContains {
				require.NotContains(t, got, item, "expected %q NOT in result", item)
			}
		})
	}
}