package pimport

import (
	"fmt"
	"strings"
	"testing"

	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	"go.uber.org/zap"
)

func TestGenerateCreateTableSQL(t *testing.T) {
	logger := zap.NewExample()
	type testCase struct {
		DBName         string
		TableName      string
		Columns        []*dataflow.ColumnInfo
		pkCols         []string
		ExpectedSQL    string
		ExpectErrorMsg string
	}
	tcs := []testCase{
		{
			DBName:    "db01",
			TableName: "tbl_basic",
			Columns: []*dataflow.ColumnInfo{
				{
					ColumnName: "col01",
					ColumnType: "int",
				},
				{
					ColumnName: "col02",
					ColumnType: "varchar(255)",
				},
			},
			ExpectedSQL: "CREATE TABLE IF NOT EXISTS `db01`.`tbl_basic` (`col01` INT,`col02` VARCHAR(255))",
		},
		{
			DBName:    "db01",
			TableName: "tbl_err",
			Columns: []*dataflow.ColumnInfo{
				{
					ColumnName: "col01",
					ColumnType: "int",
				},
				{
					ColumnName: "col02",
					ColumnType: "abcde",
				},
			},
			ExpectErrorMsg: "The table cannot be created due to invalid table name, column names or column types. Please check these settings.",
		},
		{
			DBName:    "db01",
			TableName: "TBL_Capital",
			Columns: []*dataflow.ColumnInfo{
				{
					ColumnName: "Col01",
					ColumnType: "integer",
				},
				{
					ColumnName: "col02",
					ColumnType: "varchar(64)",
				},
			},
			ExpectedSQL: "CREATE TABLE IF NOT EXISTS `db01`.`TBL_Capital` (`Col01` INT,`col02` VARCHAR(64))",
		},
		{
			DBName:    "db01",
			TableName: "tbl_empty_col",
			Columns: []*dataflow.ColumnInfo{
				{
					ColumnName: "",
					ColumnType: "int",
				},
				{
					ColumnName: "col02",
					ColumnType: "varchar(64)",
				},
			},
			ExpectErrorMsg: "The 1st column name is empty.",
		},
		// happy path: create table with pk
		{
			DBName:    "db01",
			TableName: "TBL_PK",
			Columns: []*dataflow.ColumnInfo{
				{
					ColumnName: "Col01",
					ColumnType: "integer",
				},
				{
					ColumnName: "col02",
					ColumnType: "varchar(64)",
				},
			},
			pkCols:      []string{"Col01", "col02"},
			ExpectedSQL: "CREATE TABLE IF NOT EXISTS `db01`.`TBL_PK` (`Col01` INT,`col02` VARCHAR(64),PRIMARY KEY(`Col01`, `col02`) CLUSTERED)",
		},
		// happy path: create table with different case pk
		{
			DBName:    "db01",
			TableName: "TBL_PK",
			Columns: []*dataflow.ColumnInfo{
				{
					ColumnName: "Col01",
					ColumnType: "integer",
				},
				{
					ColumnName: "col02",
					ColumnType: "varchar(64)",
				},
			},
			pkCols:      []string{"Col01", "Col02"},
			ExpectedSQL: "CREATE TABLE IF NOT EXISTS `db01`.`TBL_PK` (`Col01` INT,`col02` VARCHAR(64),PRIMARY KEY(`Col01`, `Col02`) CLUSTERED)",
		},
		// bad path: create table with pk
		{
			DBName:    "db01",
			TableName: "TBL_PK",
			Columns: []*dataflow.ColumnInfo{
				{
					ColumnName: "Col01",
					ColumnType: "integer",
				},
				{
					ColumnName: "col02",
					ColumnType: "varchar(64)",
				},
			},
			pkCols:         []string{"Col03", "col02"},
			ExpectErrorMsg: "Primary key column 'Col03' cannot be found in the table's column list.",
		},
		// happy path: create table with keyword pk
		{
			DBName:    "db01",
			TableName: "TBL_KEYWORD_PK",
			Columns: []*dataflow.ColumnInfo{
				{
					ColumnName: "rank",
					ColumnType: "integer",
				},
				{
					ColumnName: "col02",
					ColumnType: "varchar(64)",
				},
			},
			pkCols:      []string{"rank"},
			ExpectedSQL: "CREATE TABLE IF NOT EXISTS `db01`.`TBL_KEYWORD_PK` (`rank` INT,`col02` VARCHAR(64),PRIMARY KEY(`rank`) CLUSTERED)",
		},
		// happy path: create table with backslash
		{
			DBName:    "db`01",
			TableName: "TBL_`PK",
			Columns: []*dataflow.ColumnInfo{
				{
					ColumnName: "Col`01",
					ColumnType: "integer",
				},
				{
					ColumnName: "col`02",
					ColumnType: "varchar(64)",
				},
			},
			pkCols:      []string{"Col`01", "Col`02"},
			ExpectedSQL: "CREATE TABLE IF NOT EXISTS `db``01`.`TBL_``PK` (`Col``01` INT,`col``02` VARCHAR(64),PRIMARY KEY(`Col``01`, `Col``02`) CLUSTERED)",
		},
	}
	allSupportedTypes := GetDefaultDataType()
	expectedColDefStrs := make([]string, 0, len(allSupportedTypes))
	colInfos := make([]*dataflow.ColumnInfo, 0, len(allSupportedTypes))
	for i, colType := range allSupportedTypes {
		colName := fmt.Sprintf("col%02d", i+1)
		expectedColDefStrs = append(expectedColDefStrs, fmt.Sprintf("`%s` %s", colName, colType))
		colInfos = append(colInfos, &dataflow.ColumnInfo{
			ColumnName: colName,
			ColumnType: colType,
		})
	}
	dbName := "db01"
	tableName := "tbl_all_types"
	expectedSQL := fmt.Sprintf("CREATE TABLE IF NOT EXISTS `%s`.`%s` (%s)", dbName, tableName, strings.Join(expectedColDefStrs, ","))
	tcs = append(tcs, testCase{
		DBName:      dbName,
		TableName:   tableName,
		Columns:     colInfos,
		ExpectedSQL: expectedSQL,
	})
	for _, tc := range tcs {
		createSQL, err := generateCreateTableSQL(logger, tc.DBName, tc.TableName, tc.Columns, tc.pkCols)
		if len(tc.ExpectErrorMsg) > 0 {
			require.Error(t, err)
			require.Equal(t, tc.ExpectErrorMsg, err.Error())
		} else {
			require.NoError(t, err)
			require.Equal(t, tc.ExpectedSQL, createSQL)
		}
	}
}