package pimport

import (
	"regexp"
	"testing"

	"github.com/pingcap/errors"
	"github.com/stretchr/testify/require"
	commonpb "github.com/tidbcloud/dataflow-service/idl/pbgen/proto/common"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	"github.com/tidbcloud/dataflow-service/internal/db/entity"
	errhandler "github.com/tidbcloud/dataflow-service/pkg/error-handler"
)

func TestFindFileNameOffset(t *testing.T) {
	errMsg := []string{
		"[Lightning:Restore:ErrEncodeKV]encode kv error in file db.test.000000002.sql:0 at offset 41: syntax error: premature EOF at offset 41",
		"[Lightning:Restore:ErrEncodeKV]encode kv error in file 2022-12-12 02:22:44.713051854 +0000 UTC m=+516.355801199-csv.threads.csv:0 at offset 4961: syntax error: cannot have consecutive fields without separator",
		"[Lightning:Restore:ErrEncodeKV]encode kv error in file 003_assets.csv:0 at offset 108818577: syntax error: cannot have consecutive fields without separator",
		"[Lightning:Restore:ErrEncodeKV]encode kv error in file matrix.label_console_staging._0_3_0.csv:0 at offset 58891848: syntax error: cannot have consecutive fields without separator",
	}
	prettyMsg := []string{}
	for _, msg := range errMsg {
		prettyMsg = append(prettyMsg, errhandler.PrettyMessage(msg, uint32(entity.ImportDataFormatCsv)))
	}
	expectedResult := []struct {
		fileName string
		offset   int64
	}{
		{
			fileName: "db.test.000000002.sql",
			offset:   41,
		},
		{
			fileName: "2022-12-12 02:22:44.713051854 +0000 UTC m=+516.355801199-csv.threads.csv",
			offset:   4961,
		},
		{
			fileName: "003_assets.csv",
			offset:   108818577,
		},
		{
			fileName: "matrix.label_console_staging._0_3_0.csv",
			offset:   58891848,
		},
	}
	for i, msg := range prettyMsg {
		r := regexp.MustCompile("SyntaxError: ")
		var idxs []int
		if idxs = r.FindIndex([]byte(msg)); idxs == nil {
			// not syntax error
			return
		}
		begIdx := idxs[1]
		fileName, offset, err := findFileNameOffset(msg, begIdx)
		require.NoError(t, err)
		require.Equal(t, expectedResult[i].fileName, fileName)
		require.Equal(t, expectedResult[i].offset, offset)
	}
}

func TestExtractCSVSyntaxErr(t *testing.T) {
	testCases := []struct {
		content       string
		start         int64
		expectContent string
		separator     byte
	}{
		{
			content: `a,b,c,d
1,2,3,4
2,10,4,5
3,4,5,""6,`,
			start:         34,
			expectContent: `3,4,5,""6,`,
			separator:     '\n',
		},
		{
			content:       "a,b,c,d\n1,2,3,4\n2,10,4,5\n1111,\"\"\",cc\n3,4,5,6",
			start:         45,
			expectContent: "",
			separator:     '\n',
		},
		{
			content:       "(a,b,c,d),(1,2,3,4),(2,,10,4,5)",
			start:         23,
			expectContent: ",(2,,10,4,5)",
			separator:     ')',
		},
	}
	for _, testCase := range testCases {
		content := extractFileSyntaxErr(testCase.start, testCase.content, '"', testCase.separator)
		require.Equal(t, testCase.expectContent, content)
	}
}

func TestImportEntity2CreateReq_NoLocalImport(t *testing.T) {
	imp := entity.Import{
		DataFormat:      entity.ImportDataFormatCsv,
		CustomCSVFormat: "{\"delimiter\": \",\"}", // Sample custom CSV format
		ClusterID:       123,
		BucketRegion:    "us-west-1",
		DataSourceType:  entity.ImportDataSourceS3,
		AwsRoleArn:      "your-role",
		OptionalArgs: &entity.OptionalArgs{
			SkipChecksum: true,
		},
		TargetTableInfos: entity.TargetTableInfos{
			Infos: []entity.TargetTableInfo{
				{
					Schema:     "public",
					Table:      "users",
					CustomFile: "users.csv",
				},
			},
		},
	}

	expectedResp := &dataflow.ImportCreateReq{
		ClusterId:      123,
		Region:         "us-west-1",
		DataFormat:     uint32(dataflow.ImportDataFormat_CSV),       // Assuming the conversion functions are correct
		DataSourceType: uint32(dataflow.ImportDataSource_S3_SOURCE), // Assuming the conversion functions are correct
		CsvFormat: &dataflow.CustomCSVFormat{
			Delimiter: ",",
		},
		AwsRoleArn:   "your-role",
		SkipChecksum: true,
		TargetTableInfos: []*dataflow.ImportTargetTableInfo{
			{
				TargetTable: &commonpb.Table{
					Schema: "public",
					Table:  "users",
				},
				CustomFile: "users.csv",
			},
		},
	}

	resp, err := ImportEntity2CreateReq(imp)
	require.NoError(t, err)
	require.Equal(t, expectedResp, resp)
}

func TestImportEntity2CreateReq_LocalImport(t *testing.T) {
	imp := entity.Import{
		DataFormat:      entity.ImportDataFormatCsv,
		CustomCSVFormat: "{\"delimiter\": \",\"}", // Sample custom CSV format
		ClusterID:       123,
		BucketRegion:    "us-west-1",
		DataSourceType:  entity.ImportDataSourceS3,
		AwsRoleArn:      "your-role",
		LocalImportCfg: &entity.LocalImportCfg{
			TargetDb:         "yourdb",
			TargetTable:      "yourtable",
			OriginalFileName: "yourfile.csv",
			TotalSize:        100,
		},
	}

	expectedResp := &dataflow.ImportCreateReq{
		ClusterId:      123,
		Region:         "us-west-1",
		DataFormat:     uint32(dataflow.ImportDataFormat_CSV),          // Assuming the conversion functions are correct
		DataSourceType: uint32(dataflow.ImportDataSource_LOCAL_SOURCE), // Assuming the conversion functions are correct
		CsvFormat: &dataflow.CustomCSVFormat{
			Delimiter: ",",
		},
		AwsRoleArn: "your-role",
		LocalImportCfg: &dataflow.LocalImportCfg{
			TargetTable: &commonpb.Table{
				Schema: "yourdb",
				Table:  "yourtable",
			},
			OriginalFileName: "yourfile.csv",
			TotalSize:        100,
		},
	}

	resp, err := ImportEntity2CreateReq(imp)
	require.NoError(t, err)
	require.Equal(t, expectedResp, resp)
}

func TestImportEntity2CreateReq_CustomCSVFormatError(t *testing.T) {
	imp := entity.Import{
		DataFormat:       entity.ImportDataFormatCsv,
		CustomCSVFormat:  "{\"invalidjson\":}", // Invalid JSON format
		ClusterID:        123,
		BucketRegion:     "us-west-1",
		DataSourceType:   1,
		AwsRoleArn:       "your-role",
		TargetTableInfos: entity.TargetTableInfos{},
		LocalImportCfg:   nil,
	}

	expectedError := errors.New("parse custom CSV format failed: invalid character '}' looking for beginning of value")
	resp, err := ImportEntity2CreateReq(imp)
	require.EqualError(t, err, expectedError.Error())
	require.Nil(t, resp)
}