package dm

import (
	"testing"

	"github.com/pingcap/log"
	"github.com/pingcap/tiflow/dm/pkg/terror"
	engpb "github.com/pingcap/tiflow/engine/enginepb"
	"github.com/stretchr/testify/require"
	dmpb "github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dm"
	"github.com/tidbcloud/dataflow-service/internal/db/entity"
)

func TestRemovePercent(t *testing.T) {
	testCases := []struct {
		input    string
		expected float64
	}{
		{"10.24 %", 10.24},
		{"99. %", 99},
		{"0", 0},
		{"", 0},
	}
	for _, tc := range testCases {
		actual, _ := removePercent(tc.input)
		require.Equal(t, tc.expected, actual)
	}
}

func TestNormalizeErrMsg(t *testing.T) {
	in := "get binlog event error: ERROR 1236 (HY000): Could not find first log file name in binary log index file"
	require.Equal(t, "The required binary log for migration no longer exists on the source database. Please make sure binary log files are kept for long enough time for migration to succeed.",
		normalizeErrMsg(in, ""))

	in = "fail to restart streamer controller: start syncing binlog from remote streamer. raw_cause dial tcp 54.244.153.168:3306: i/o timeout"
	require.Equal(t, "Failed to connect to the source database using given parameters. Please make sure the source database is up and can be connected using the given parameters.",
		normalizeErrMsg(in, ""))

	in = "[Lightning:Restore:ErrRestoreTable]restore table `test_full_1k`.`tbl_5` failed: dial tcp: lookup db-tidb.tidb65343.svc on 172.20.0.10:53: no such host"
	require.Equal(t, "Failed to connect to the TiDB cluster using the given user and password. Please make sure TiDB Cluster is up and can be connected to using the given user and password.",
		normalizeErrMsg(in, ""))

	in = "raw_cause dial tcp 54.244.153.168:3306: i/o timeout"
	require.Equal(t, "Failed to connect to the source database using given parameters. Please make sure the source database is up and can be connected using the given parameters.",
		normalizeErrMsg(in, terror.ScopeUpstream.String()))

	in = "dial tcp: lookup db-tidb.tidb65343.svc on 172.20.0.10:53: no such host"
	require.Equal(t, "Failed to connect to the TiDB cluster using the given user and password. Please make sure TiDB Cluster is up and can be connected to using the given user and password.",
		normalizeErrMsg(in, terror.ScopeDownstream.String()))

	in = "[Lightning:Restore:ErrRestoreTable]restore table `test_full`.`large_tbl3` failed: [`test_full`.`large_tbl3`] batch write rows reach max retry 3 and still failed: Error 1105: tikv disk full"
	require.Equal(t, "TiDB cluster storage is not enough. Please increase the node storage of TiKV.",
		normalizeErrMsg(in, ""))

	in = "Error 9501: Timed-out waiting to acquire database connection"
	require.Equal(t, "Failed to connect to the source database. Please check whether the database is available or the maximum connections have been reached.",
		normalizeErrMsg(in, ""))

	in = "xxx"
	require.Equal(t, in, normalizeErrMsg(in, ""))
}

func TestEngineJob2Dataflow(t *testing.T) {
	jobID := 1
	sourceID := 1
	dmJobSources := []entity.DMJobSource{{ConnProfileID: entity.ID(sourceID)}}

	// engineJobInfo.Detail is nil, but there's error
	engineJobInfo := &engpb.Job{
		State: engpb.Job_Created,
		Error: &engpb.Job_Error{Message: "Error 1044: Access denied for user 'xxx'@'%' to database 'tidbcloud_dm_meta'"},
	}
	outputJobInfo := &dmpb.Job{
		Id: uint64(jobID),
		SubTasks: map[uint64]*dmpb.SubTask{
			uint64(sourceID): {},
		},
	}
	err := EngineJob2Dataflow(log.L(), engineJobInfo, outputJobInfo, dmJobSources)
	require.NoError(t, err)
	subtask := outputJobInfo.SubTasks[uint64(sourceID)]
	require.Equal(t, dmpb.SubTask_Failed, subtask.Stage)
	require.NotNil(t, subtask.ProcessError)
	require.Contains(t, subtask.ProcessError.Message, "Access denied")
	require.Nil(t, subtask.DumpDetail)
	require.Nil(t, subtask.LoadDetail)
	require.Nil(t, subtask.SyncDetail)

	// engineJobInfo.Detail is nil and there's no error
	for _, engineErr := range []*engpb.Job_Error{{}, nil} {
		engineJobInfo = &engpb.Job{
			State: engpb.Job_Created,
			Error: engineErr,
		}
		outputJobInfo = &dmpb.Job{
			Id: uint64(jobID),
			SubTasks: map[uint64]*dmpb.SubTask{
				uint64(sourceID): {},
			},
		}
		err = EngineJob2Dataflow(log.L(), engineJobInfo, outputJobInfo, dmJobSources)
		require.NoError(t, err)
		subtask = outputJobInfo.SubTasks[uint64(sourceID)]
		require.Equal(t, dmpb.SubTask_Running, subtask.Stage)
		require.Nil(t, subtask.ProcessError)
		require.Nil(t, subtask.DumpDetail)
		require.Nil(t, subtask.LoadDetail)
		require.Nil(t, subtask.SyncDetail)
	}
}