package importcfg

import (
	"regexp"
	"testing"

	. "github.com/onsi/gomega"
	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
)

type testCase struct {
	CaseName            string
	FilePath            string
	ExpectedDBName      string
	ExpectedTableName   string
	ExpectedCompression string
}

var sqlTestCases = []testCase{
	{
		CaseName:          "sql data file normal",
		FilePath:          "my_db.my_table.sql",
		ExpectedDBName:    "my_db",
		ExpectedTableName: "my_table",
	},
	{
		CaseName:          "sql data file normal subfolder",
		FilePath:          "some_folder/my_db.my_table.sql",
		ExpectedDBName:    "my_db",
		ExpectedTableName: "my_table",
	},
	{
		CaseName:          "sql data file with file index",
		FilePath:          "my_db.my_table.0001.sql",
		ExpectedDBName:    "my_db",
		ExpectedTableName: "my_table",
	},
	{
		CaseName:          "sql data file with file index subfolder",
		FilePath:          "some_folder/my_db.my_table.0001.sql",
		ExpectedDBName:    "my_db",
		ExpectedTableName: "my_table",
	},
	{
		CaseName:            "sql data file compression",
		FilePath:            "my_db.my_table.sql.gz",
		ExpectedDBName:      "my_db",
		ExpectedTableName:   "my_table",
		ExpectedCompression: "gz",
	},
	{
		CaseName:            "sql data file normal subfolder compression",
		FilePath:            "some_folder/my_db.my_table.sql.gz",
		ExpectedDBName:      "my_db",
		ExpectedTableName:   "my_table",
		ExpectedCompression: "gz",
	},
	{
		CaseName:            "sql data file with file index compression",
		FilePath:            "my_db.my_table.0001.sql.gz",
		ExpectedDBName:      "my_db",
		ExpectedTableName:   "my_table",
		ExpectedCompression: "gz",
	},
	{
		CaseName:            "sql data file with file index subfolder compression",
		FilePath:            "some_folder/my_db.my_table.0001.sql.gz",
		ExpectedDBName:      "my_db",
		ExpectedTableName:   "my_table",
		ExpectedCompression: "gz",
	},
	{
		CaseName: "sql_snowflake_style_01",
		// For snowflake file name example, please refer to:
		// https://docs.snowflake.com/en/user-guide/data-unload-overview#bulk-unloading-into-single-or-multiple-files
		// https://docs.snowflake.com/en/sql-reference/sql/copy-into-location
		FilePath:          "my_db.my_table._000_0_1.sql",
		ExpectedDBName:    "my_db",
		ExpectedTableName: "my_table",
	},
	{
		CaseName: "sql_snowflake_style_02",
		// For snowflake file name example, please refer to:
		// https://docs.snowflake.com/en/user-guide/data-unload-overview#bulk-unloading-into-single-or-multiple-files
		// https://docs.snowflake.com/en/sql-reference/sql/copy-into-location
		FilePath:          "my_db.my_table_006_0_1.sql",
		ExpectedDBName:    "my_db",
		ExpectedTableName: "my_table",
	},
	{
		CaseName:          "sql_snowflake_style_subfolder_01",
		FilePath:          "some_folder/sub_folder/my_db.my_table._000_0_1.sql",
		ExpectedDBName:    "my_db",
		ExpectedTableName: "my_table",
	},
	{
		CaseName:          "sql_snowflake_style_subfolder_02",
		FilePath:          "some_folder/sub_folder/my_db.my_table_006_0_1.sql",
		ExpectedDBName:    "my_db",
		ExpectedTableName: "my_table",
	},
}

var csvTestCases = []testCase{
	{
		CaseName:          "CSV_normal",
		FilePath:          "my_db.my_table.csv",
		ExpectedDBName:    "my_db",
		ExpectedTableName: "my_table",
	},
	{
		CaseName:          "CSV_with_file_index",
		FilePath:          "my_db.my_table.0000003.csv",
		ExpectedDBName:    "my_db",
		ExpectedTableName: "my_table",
	},
	{
		CaseName: "CSV_snowflake_style_01",
		// For snowflake file name example, please refer to:
		// https://docs.snowflake.com/en/user-guide/data-unload-overview#bulk-unloading-into-single-or-multiple-files
		// https://docs.snowflake.com/en/sql-reference/sql/copy-into-location
		FilePath:          "my_db.my_table._000_0_1.csv",
		ExpectedDBName:    "my_db",
		ExpectedTableName: "my_table",
	},
	{
		CaseName: "CSV_snowflake_style_02",
		// For snowflake file name example, please refer to:
		// https://docs.snowflake.com/en/user-guide/data-unload-overview#bulk-unloading-into-single-or-multiple-files
		// https://docs.snowflake.com/en/sql-reference/sql/copy-into-location
		FilePath:          "my_db.my_table_006_0_1.csv",
		ExpectedDBName:    "my_db",
		ExpectedTableName: "my_table",
	},
	{
		CaseName:          "CSV_subfolder",
		FilePath:          "some_folder/sub_folder/my_db.my_table.csv",
		ExpectedDBName:    "my_db",
		ExpectedTableName: "my_table",
	},
	{
		CaseName:          "CSV_with_file_index_subfolder",
		FilePath:          "some_folder/sub_folder/my_db.my_table.0000003.csv",
		ExpectedDBName:    "my_db",
		ExpectedTableName: "my_table",
	},
	{
		CaseName:          "CSV_snowflake_style_subfolder_01",
		FilePath:          "some_folder/sub_folder/my_db.my_table._000_0_1.csv",
		ExpectedDBName:    "my_db",
		ExpectedTableName: "my_table",
	},
	{
		CaseName:          "CSV_snowflake_style_subfolder_02",
		FilePath:          "some_folder/sub_folder/my_db.my_table_006_0_1.csv",
		ExpectedDBName:    "my_db",
		ExpectedTableName: "my_table",
	},
	{
		CaseName:          "CSV_file_with_number_suffix 01",
		FilePath:          "my_db.my_table00601.csv",
		ExpectedDBName:    "my_db",
		ExpectedTableName: "my_table00601",
	},
	{
		CaseName:          "CSV_file_with_number_suffix 02",
		FilePath:          "my_db.my_table_00601.csv",
		ExpectedDBName:    "my_db",
		ExpectedTableName: "my_table_00601",
	},
	{
		CaseName:            "CSV_compression",
		FilePath:            "my_db.my_table.csv.gz",
		ExpectedDBName:      "my_db",
		ExpectedTableName:   "my_table",
		ExpectedCompression: "gz",
	},
	{
		CaseName:            "CSV_with_file_index_compression",
		FilePath:            "my_db.my_table.0000003.csv.gz",
		ExpectedDBName:      "my_db",
		ExpectedTableName:   "my_table",
		ExpectedCompression: "gz",
	},
	{
		CaseName:            "CSV_subfolder_compression",
		FilePath:            "some_folder/sub_folder/my_db.my_table.csv.gz",
		ExpectedDBName:      "my_db",
		ExpectedTableName:   "my_table",
		ExpectedCompression: "gz",
	},
	{
		CaseName:            "CSV_with_file_index_subfolder_compression",
		FilePath:            "some_folder/sub_folder/my_db.my_table.0000003.csv.gz",
		ExpectedDBName:      "my_db",
		ExpectedTableName:   "my_table",
		ExpectedCompression: "gz",
	},
}

var parquetTestCases = []testCase{
	{
		CaseName:          "raw parquet file normal",
		FilePath:          "my_db.my_table.gz.parquet",
		ExpectedDBName:    "my_db",
		ExpectedTableName: "my_table",
	},
	{
		CaseName:          "raw parquet file sub-folder",
		FilePath:          "some_folder/my_db.my_table.gz.parquet",
		ExpectedDBName:    "my_db",
		ExpectedTableName: "my_table",
	},
	{
		CaseName:          "raw parquet file no compression",
		FilePath:          "my_db.my_table.parquet",
		ExpectedDBName:    "my_db",
		ExpectedTableName: "my_table",
	},
	{
		CaseName:          "raw parquet file no compression sub-folder",
		FilePath:          "some_folder/my_db.my_table.parquet",
		ExpectedDBName:    "my_db",
		ExpectedTableName: "my_table",
	},
	{
		CaseName:          "raw parquet file with file index",
		FilePath:          "my_db.my_table.00001.gz.parquet",
		ExpectedDBName:    "my_db",
		ExpectedTableName: "my_table",
	},
	{
		CaseName:          "raw parquet file with file index sub-folder",
		FilePath:          "some_folder/my_db.my_table.00001.gz.parquet",
		ExpectedDBName:    "my_db",
		ExpectedTableName: "my_table",
	},
	{
		CaseName: "parquet_snowflake_style_01",
		// For snowflake file name example, please refer to:
		// https://docs.snowflake.com/en/user-guide/data-unload-overview#bulk-unloading-into-single-or-multiple-files
		// https://docs.snowflake.com/en/sql-reference/sql/copy-into-location
		FilePath:          "my_db.my_table._000_0_1.parquet",
		ExpectedDBName:    "my_db",
		ExpectedTableName: "my_table",
	},
	{
		CaseName: "parquet_snowflake_style_02",
		// For snowflake file name example, please refer to:
		// https://docs.snowflake.com/en/user-guide/data-unload-overview#bulk-unloading-into-single-or-multiple-files
		// https://docs.snowflake.com/en/sql-reference/sql/copy-into-location
		FilePath:          "my_db.my_table_006_0_1.parquet",
		ExpectedDBName:    "my_db",
		ExpectedTableName: "my_table",
	},
	{
		CaseName:          "parquet_snowflake_style_subfolder_01",
		FilePath:          "some_folder/sub_folder/my_db.my_table._000_0_1.parquet",
		ExpectedDBName:    "my_db",
		ExpectedTableName: "my_table",
	},
	{
		CaseName:          "parquet_snowflake_style_subfolder_02",
		FilePath:          "some_folder/sub_folder/my_db.my_table_006_0_1.parquet",
		ExpectedDBName:    "my_db",
		ExpectedTableName: "my_table",
	},
}

var auroraTestCases = []testCase{
	{
		CaseName:          "aurora snapshot file normal",
		FilePath:          "my_db.my_table/1/part-00140-b5bd86f4-d410-45df-80f4-64d373aa0a24-c000.gz.parquet",
		ExpectedDBName:    "my_db",
		ExpectedTableName: "my_table",
	},
	{
		CaseName:          "aurora snapshot file subfolder",
		FilePath:          "some_folder/my_db/my_db.my_table/1/part-00140-b5bd86f4-d410-45df-80f4-64d373aa0a24-c000.gz.parquet",
		ExpectedDBName:    "my_db",
		ExpectedTableName: "my_table",
	},
}

func TestFilePattern(t *testing.T) {
	testFilePattern(t, DefaultSQLFilePattern, sqlTestCases)

	testFilePattern(t, DefaultCSVFilePattern, csvTestCases)

	testFilePattern(t, DefaultRawParquetFilePattern, parquetTestCases)

	testFilePattern(t, DefaultAuroraSnapshotFilePattern, auroraTestCases)
}

func testFilePattern(t *testing.T, filePattern string, testCases []testCase) {
	var (
		DBNameGroupIndex      = 1
		TableNameGroupIndex   = 2
		CompressionGroupIndex = 3
	)
	for _, tc := range testCases {
		re, err := regexp.Compile(filePattern)
		require.NoErrorf(t, err, "case name: %s", tc.CaseName)
		groups := re.FindStringSubmatch(tc.FilePath)
		require.NotNil(t, groups, "case name: %s", tc.CaseName)

		require.Greaterf(t, len(groups), DBNameGroupIndex, "case name: %s", tc.CaseName)
		dbName := groups[DBNameGroupIndex]
		require.Equalf(t, tc.ExpectedDBName, dbName, "case name: %s", tc.CaseName)

		require.Greaterf(t, len(groups), TableNameGroupIndex, "case name: %s", tc.CaseName)
		tableName := groups[TableNameGroupIndex]
		require.Equalf(t, tc.ExpectedTableName, tableName, "case name: %s", tc.CaseName)

		if tc.ExpectedCompression != "" {
			require.Greaterf(t, len(groups), CompressionGroupIndex, "case name: %s", tc.CaseName)
			compression := groups[CompressionGroupIndex]
			require.Equalf(t, tc.ExpectedCompression, compression, "case name: %s", tc.CaseName)
		}
	}
}

func TestFileMatch(t *testing.T) {
	mustFileMatch(t, DefaultSQLFilePattern, dataflow.ImportDataFormat_SQL, sqlTestCases)

	mustFileMatch(t, DefaultCSVFilePattern, dataflow.ImportDataFormat_CSV, csvTestCases)

	mustFileMatch(t, DefaultRawParquetFilePattern, dataflow.ImportDataFormat_PARQUET, parquetTestCases)

	mustFileMatch(t, DefaultAuroraSnapshotFilePattern, dataflow.ImportDataFormat_AURORA_SNAPSHOT, auroraTestCases)
}

func mustFileMatch(t *testing.T, filePattern string, dataFormat dataflow.ImportDataFormat, testCases []testCase) {
	for _, tc := range testCases {
		re, err := regexp.Compile(filePattern)
		require.NoErrorf(t, err, "case name: %s", tc.CaseName)
		groups := re.FindStringSubmatch(tc.FilePath)
		require.NotNil(t, groups, "case name: %s", tc.CaseName)

		require.True(t, MatchDefaultNamingRule(
			tc.FilePath,
			tc.ExpectedDBName,
			tc.ExpectedTableName,
			uint32(dataFormat),
		), "case name: %s", tc.CaseName)
	}
}

func TestCustomFileMatch(t *testing.T) {
	testCases := []struct {
		caseName   string
		filePath   string
		customFile string
		dataFormat dataflow.ImportDataFormat
		expected   bool
	}{
		{
			caseName:   "custom file match",
			customFile: "s3://bucket/prefix/*",
			filePath:   "s3://bucket/prefix/my_db.my_table/1/part-00140-b5bd86f4-d410-45df-80f4-64d373aa0a24-c000.gz.parquet",
			dataFormat: dataflow.ImportDataFormat_PARQUET,
			expected:   true,
		},
		{
			caseName:   "custom file match, dataformat mismatch",
			customFile: "s3://bucket/prefix/*",
			filePath:   "s3://bucket/prefix/my_db.my_table/1/part-00140-b5bd86f4-d410-45df-80f4-64d373aa0a24-c000.gz.parquet",
			dataFormat: dataflow.ImportDataFormat_CSV,
			expected:   false,
		},
		{
			caseName:   "custom file match ?",
			customFile: "s3://bucket/prefix/my_db.my_table/?/part-00140-b5bd86f4-d410-45df-80f4-64d373aa0a24-c000.gz.parquet",
			filePath:   "s3://bucket/prefix/my_db.my_table/1/part-00140-b5bd86f4-d410-45df-80f4-64d373aa0a24-c000.gz.parquet",
			dataFormat: dataflow.ImportDataFormat_PARQUET,
			expected:   true,
		},
		{
			caseName:   "custom file match absolute path(.csv)",
			customFile: "s3://import-manual-test/check_compatible/data.csv",
			filePath:   "s3://import-manual-test/check_compatible/data.csv",
			dataFormat: dataflow.ImportDataFormat_CSV,
			expected:   true,
		},
		{
			caseName:   "custom file match absolute path(.csv.gz)",
			customFile: "s3://import-manual-test/check_compatible/data.csv.gz",
			filePath:   "s3://import-manual-test/check_compatible/data.csv.gz",
			dataFormat: dataflow.ImportDataFormat_CSV,
			expected:   true,
		},
		{
			caseName:   "custom file match absolute path(.csv.xx)",
			customFile: "s3://import-manual-test/check_compatible/data.csv.xx",
			filePath:   "s3://import-manual-test/check_compatible/data.csv.xx",
			dataFormat: dataflow.ImportDataFormat_CSV,
			expected:   false,
		},
	}

	for _, tc := range testCases {
		match, err := MatchCustomNamingRule(tc.filePath, tc.customFile, uint32(tc.dataFormat))
		require.NoError(t, err, tc.caseName)
		require.Equal(t, tc.expected, match, tc.caseName)
	}
}

func TestGenerateObjectNamePatternRegexp(t *testing.T) {
	wildcardsWorks := map[string][]string{
		`db1.tbl1`:                    {"db1.tbl1"},
		`db[0-9].tbl[0-9a-f][0-9a-f]`: {"db0.tblaa", "db0.tbla0"},
		`data.*`:                      {"data.blabla", "data.bla"},
		`*.backup_*`:                  {"hello.backup_2021"},
		`employees.*`:                 {"employees.1", "employees.1ajskldla"},
		`*.WorkOrder`:                 {"hello.WorkOrder"},
		`!*.Password`:                 {"!hello.Password"},
		`!employees.salaries`:         {"!employees.salaries"},
		`db.with.dots.*`:              {"db.with.dots.hello"},
		`"db.with.dots"."tbl\1"`:      {`"db.with.dots"."tbl\1"`},
		"`db.with.dots`.`tbl\\2`":     {"`db.with.dots`.`tbl\\2`"},
		`"foo""bar".` + "foo``bar``":  {`"foo""bar".` + "foo``bar``"},
		`foo\"bar.` + "foo\\`bar":     {`foo\"bar.` + "foo\\`bar"},
		`[0-9]`:                       {"0"},
		`[a-z]`:                       {"a"},
		`[a-z][0-9][a-z]`:             {"a0b"},
		`^%$`:                         {"^%$"},
		`^.$*`:                        {`^.$*`},
		`hello`:                       {`hello`},
		`hello.`:                      {`hello.`},
		`hello*`:                      {`hello.*`, `helloabcdksdk`},
		`hello?`:                      {`hello.`, `helloa`, `hello?`},
		`hello=world\\`:               {`hello=world\\`},
		`*hello=123?`:                 {`blabalhello=123a`},
		`*hello=123/*/abc?`:           {`*hello=123/ablablabl/abcd`},
		`*station=402260*`:            {`s3://wjktest123/mydata3/data/station=402260/part-r-00000-ddaee723-f3f6-4f25-a34b-3312172aa6d7.snappy.parquet`},
		`*station=402260*.parquet`:    {`s3://wjktest123/mydata3/data/station=402260/part-r-00000-ddaee723-f3f6-4f25-a34b-3312172aa6d7.snappy.parquet`},
		`*.parquet`:                   {`s3://wjktest123/mydata3/data/station=402265/part-r-00283-ddaee723-f3f6-4f25-a34b-3312172aa6d7.snappy.parquet`, `s3://wjktest123/mydata3/data/station=402260/part-r-00001-ddaee723-f3f6-4f25-a34b-3312172aa6d7.snappy.parquet`},
	}

	g := NewGomegaWithT(t)
	for k, vs := range wildcardsWorks {
		for _, v := range vs {
			generatedRegexp := genCustomNamingRuleRegexp(k)
			r, err := regexp.Compile(generatedRegexp)
			g.Expect(err).To(BeNil())
			g.Expect(r.MatchString(v)).To(BeTrue())
		}
	}

	wildcardsNotWorks := map[string][]string{
		"hello":      {"helloa", "hellob", "chello"},
		"hello[0-9]": {"helloa", "bhello"},
		"*hello":     {"ahellob", "hello*"},
		"hello?":     {"helloa.", "bhello"},
		"^%$":        {"^%$ww"},
		"^.$*":       {"?^*$*"},
	}
	for k, vs := range wildcardsNotWorks {
		for _, v := range vs {
			generatedRegexp := genCustomNamingRuleRegexp(k)
			r, err := regexp.Compile(generatedRegexp)
			g.Expect(err).To(BeNil())
			g.Expect(r.MatchString(v)).To(BeFalse())
		}
	}
}