package vo

import (
	"context"
	"encoding/base64"
	"encoding/json"
	"fmt"
	"path/filepath"
	"regexp"
	"strconv"
	"strings"
	"time"

	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/juju/errors"
	"github.com/pingcap/tidb/pkg/lightning/config"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/agent"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/import/cloud_storage"
	"github.com/tidbcloud/pkgv2/log"
	"github.com/tidbcloud/serverless-service/idl/pbgen/proto/console"
	"github.com/tidbcloud/serverless-service/internal/domain/constant"
	"github.com/tidbcloud/serverless-service/internal/domain/entity"
	"github.com/tidbcloud/serverless-service/internal/errorcode"
	stashing "github.com/tidbcloud/serverless-service/internal/infra/stashing_area"
	"github.com/tidbcloud/serverless-service/internal/utils"
	tidbmgmtservicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.uber.org/zap"
	"google.golang.org/protobuf/types/known/timestamppb"
)

var tableNamePattern = regexp.MustCompile("^`([^`]+)`\\.`([^`]+)`$")

func ListImport2Console(
	ctx context.Context,
	imports []*entity.Import,
	total int64,
	token string,
) (*console.ListImportsResp, error) {
	v := new(console.ListImportsResp)
	v.TotalSize = uint32(total)
	v.NextPageToken = token

	for _, i := range imports {
		df2oa, err := GetImport2Console(ctx, i)
		if err != nil {
			return nil, err
		}
		v.Imports = append(v.Imports, df2oa)
	}
	return v, nil
}

func GetImport2Console(ctx context.Context, i *entity.Import) (*console.Import, error) {
	if i == nil {
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "import is nil")
	}

	creationDetails, err := ImportEntity2ConsoleCreationDetails(i)
	if err != nil {
		return nil, err
	}

	resp := &console.Import{
		Id:                         i.ID,
		DisplayName:                defaultDisplayName(i),
		ClusterId:                  i.ClusterID,
		CreateTime:                 timestamppb.New(i.CreatedAt),
		State:                      entityImportStatus2ConsoleState(i.Status),
		CompletePercent:            i.CompletedPercent,
		PostImportCompletedPercent: i.PostImportCompletedPercent,
		Message:                    PrettyMessage(i.Message, i.DataFormat),
		CreatedBy:                  i.Creator,
		CreationDetails:            creationDetails,
		TotalTablesCount:           i.TotalTablesCount,
		CompatibleId:               i.IDCompat,
	}

	if i.TotalSize != nil {
		resp.TotalSize = *i.TotalSize
	}
	if i.CompletedAt != nil {
		resp.CompleteTime = timestamppb.New(*i.CompletedAt)
	}
	if i.StartedAt != nil {
		resp.StartTime = timestamppb.New(*i.StartedAt)
	}

	resp.AllCompletedTables = buildTableInfos(ctx, i)

	return resp, nil
}

func defaultDisplayName(i *entity.Import) string {
	if i.Name != "" {
		return i.Name
	}
	return fmt.Sprintf("import_%s", i.CreatedAt.Format(time.RFC3339))
}

func getDefaultFilePatternSuffix(format constant.DataFormat) string {
	switch format {
	case constant.ImportDataFormatCsv:
		return ".csv*"
	case constant.ImportDataFormatSql:
		return ".sql*"
	case constant.ImportDataFormatParquet, constant.ImportDataFormatAuroraSnapshot:
		return ".parquet"
	default:
		return ""
	}
}

func buildTableInfos(ctx context.Context, i *entity.Import) []*console.ImportTableCompletionInfo {
	customFileMap := make(map[string]string)
	for _, info := range i.TargetTableInfos.Infos {
		customFileMap[UniqueTable(info.Schema, info.Table)] = info.CustomFile
	}

	suffix := getDefaultFilePatternSuffix(i.DataFormat)

	tableInfos := make([]*console.ImportTableCompletionInfo, 0, len(i.TableResults.Tables))
	for _, table := range i.TableResults.Tables {
		info := &console.ImportTableCompletionInfo{
			TableName: table.TableName,
			Result:    tableResult2Console(table.Result),
			Message:   PrettyMessage(table.Message, i.DataFormat),
		}

		if customFile, ok := customFileMap[strings.ToLower(table.TableName)]; ok {
			info.DataFilePattern = customFile
		} else {
			db, t, err := ParseDBAndTable(table.TableName)
			if err == nil {
				info.DataFilePattern = fmt.Sprintf("*%s.%s*%s", db, t, suffix)
			} else {
				log.WithContextL(ctx).Warn("parse table failed", zap.Error(err))
			}
		}

		tableInfos = append(tableInfos, info)
	}

	return tableInfos
}

func tableResult2Console(result constant.ImportTableResult) console.ImportTableCompletionInfo_Result {
	switch result {
	case constant.ImportTableResultSuccess:
		return console.ImportTableCompletionInfo_SUCCESS
	case constant.ImportTableResultWarning:
		return console.ImportTableCompletionInfo_WARNING
	case constant.ImportTableResultError:
		return console.ImportTableCompletionInfo_ERROR
	}
	return console.ImportTableCompletionInfo_UNSPECIFIED
}

func BuildImportWithS3UriLocal(
	ctx context.Context,
	req *console.CreateImportReq,
	cluster *tidbmgmtservicev1.Cluster,
	upload *entity.Upload,
	area *stashing.StashingArea,
) (*entity.ImportTask, bizerr.BizError) {
	imp := &entity.ImportTask{
		Import: &entity.Import{
			ID:             entity.GenerateIDWithPrefix(constant.ImportIDPrefix),
			ProjectID:      strconv.FormatUint(cluster.GetOriginalProjectId(), 10),
			TenantID:       req.OrgId,
			ClusterID:      req.ClusterId,
			SourceURL:      fmt.Sprintf("s3://%s/%s", area.GetBucketName(), upload.BucketKey),
			BucketRegion:   area.Region,
			DataSourceType: ConsoleDataSource2Entity(req.GetSource().GetType()),
			DataFormat:     ConsoleFileType2Entity(req.GetImportOptions().GetFileType()),
			NeedDropTables: req.GetImportOptions().GetNeedDropTable(),
			TargetTableInfos: entity.TargetTableInfos{
				Infos: []entity.TargetTableInfo{
					{
						Schema: req.GetSource().GetLocal().GetTargetDatabase(),
						Table:  req.GetSource().GetLocal().GetTargetTable(),
					},
				},
			},
		},
		Cluster: cluster,
	}
	if req.GetImportOptions().GetDisplayName() != "" {
		imp.Name = req.GetImportOptions().GetDisplayName()
	}

	if req.GetImportOptions().GetFileType() == console.ImportFileType_CSV {
		csvBytes, err := json.Marshal(req.GetImportOptions().GetCsvFormat())
		if err != nil {
			log.Error(ctx, "failed to marshal CSV format", zap.Error(err))
			return nil, bizerr.New(errorcode.ErrCodeInternalError, errorcode.InternalErrorMessage)
		}
		imp.CSVFormat = string(csvBytes)
	}

	if req.GetSource().GetLocal() != nil {
		imp.LocalImportCfg = &entity.LocalImportCfg{
			FileName:    filepath.Base(upload.BucketKey),
			TargetDB:    req.GetSource().GetLocal().GetTargetDatabase(),
			TargetTable: req.GetSource().GetLocal().GetTargetTable(),
			UploadID:    upload.ID,
		}
	}

	if area.Endpoint != "" {
		imp.S3Endpoint = area.Endpoint
		imp.S3AccessKeyID = area.AccessKeyID
		imp.S3SecretAccessKey = area.SecretAccessKey
	}

	if req.GetCreator() != "" {
		imp.Creator = req.GetCreator()
	} else {
		imp.Creator = utils.BuildCreatorID(ctx)
	}

	return imp, nil
}

func BuildImportWithS3UriConsole(
	ctx context.Context,
	req *console.CreateImportReq,
	cluster *tidbmgmtservicev1.Cluster,
	region string,
) (*entity.ImportTask, bizerr.BizError) {
	imp := &entity.ImportTask{
		Import: &entity.Import{
			ID:             entity.GenerateIDWithPrefix(constant.ImportIDPrefix),
			ProjectID:      strconv.FormatUint(cluster.GetOriginalProjectId(), 10),
			TenantID:       req.OrgId,
			ClusterID:      req.ClusterId,
			BucketRegion:   region,
			DataSourceType: ConsoleDataSource2Entity(req.GetSource().GetType()),
			DataFormat:     ConsoleFileType2Entity(req.GetImportOptions().GetFileType()),
			NeedDropTables: req.GetImportOptions().GetNeedDropTable(),
		},
		Cluster: cluster,
	}
	if req.GetImportOptions().GetDisplayName() != "" {
		imp.Name = req.GetImportOptions().GetDisplayName()
	}

	if req.GetImportOptions().GetFileType() == console.ImportFileType_CSV {
		csvBytes, err := json.Marshal(req.GetImportOptions().GetCsvFormat())
		if err != nil {
			log.Error(ctx, "failed to marshal CSV format", zap.Error(err))
			return nil, bizerr.New(errorcode.ErrCodeInternalError, errorcode.InternalErrorMessage)
		}
		imp.CSVFormat = string(csvBytes)
	}

	if req.GetSource().GetS3() != nil {
		switch req.GetSource().GetS3().GetAuthType() {
		case console.ImportS3AuthType_ROLE_ARN:
			imp.S3RoleArn = req.GetSource().GetS3().GetRoleArn()
		case console.ImportS3AuthType_ACCESS_KEY:
			imp.S3AccessKeyID = req.GetSource().GetS3().GetAccessKey().Id
			imp.S3SecretAccessKey = req.GetSource().GetS3().GetAccessKey().Secret
		}
		imp.SourceURL = req.GetSource().GetS3().GetUri()
		imp.TargetTableInfos = console2TableInfos(
			req.GetSource().GetS3().GetUri(),
			req.GetSource().GetS3().GetTargetTableInfos(),
		)
	}

	if req.GetSource().GetS3Compatible() != nil {
		imp.S3Endpoint = req.GetSource().GetS3Compatible().GetEndpoint()
		switch req.GetSource().GetS3Compatible().GetAuthType() {
		case console.ImportS3CompatibleAuthType_ACCESS_KEY:
			imp.S3AccessKeyID = req.GetSource().GetS3Compatible().GetAccessKey().Id
			imp.S3SecretAccessKey = req.GetSource().GetS3Compatible().GetAccessKey().Secret
		}
		imp.SourceURL = req.GetSource().GetS3Compatible().GetUri()
		imp.TargetTableInfos = console2TableInfos(
			req.GetSource().GetS3Compatible().GetUri(),
			req.GetSource().GetS3Compatible().GetTargetTableInfos(),
		)
	}

	if req.GetSource().GetOss() != nil {
		imp.S3Endpoint = cloud_storage.GenerateOssEndpoint(region)
		switch req.GetSource().GetOss().GetAuthType() {
		case console.ImportOSSAuthType_ACCESS_KEY:
			imp.S3AccessKeyID = req.GetSource().GetOss().GetAccessKey().Id
			imp.S3SecretAccessKey = req.GetSource().GetOss().GetAccessKey().Secret
		case console.ImportOSSAuthType_ROLE_ARN:
			imp.S3RoleArn = req.GetSource().GetOss().GetRoleArn()
		}
		imp.SourceURL = req.GetSource().GetOss().GetUri()
		imp.TargetTableInfos = console2TableInfos(
			req.GetSource().GetOss().GetUri(),
			req.GetSource().GetOss().GetTargetTableInfos(),
		)
	}

	if req.GetCreator() != "" {
		imp.Creator = req.GetCreator()
	} else {
		imp.Creator = utils.BuildCreatorID(ctx)
	}

	return imp, nil
}

func ConsoleFileType2Entity(fileType console.ImportFileType_Enum) constant.DataFormat {
	switch fileType {
	case console.ImportFileType_CSV:
		return constant.ImportDataFormatCsv
	case console.ImportFileType_SQL:
		return constant.ImportDataFormatSql
	case console.ImportFileType_AURORA_SNAPSHOT:
		return constant.ImportDataFormatAuroraSnapshot
	case console.ImportFileType_PARQUET:
		return constant.ImportDataFormatParquet
	}
	return constant.ImportDataFormatUnknown
}

func ConsoleDataSource2Entity(dataSource console.ImportSourceType_Enum) constant.DataSource {
	switch dataSource {
	case console.ImportSourceType_LOCAL:
		return constant.ImportDataSourceLocal
	case console.ImportSourceType_S3:
		return constant.ImportDataSourceS3
	case console.ImportSourceType_GCS:
		return constant.ImportDataSourceGCS
	case console.ImportSourceType_AZURE_BLOB:
		return constant.ImportDataSourceAzBlob
	case console.ImportSourceType_S3_COMPATIBLE:
		return constant.ImportDataSourceS3Compatible
	case console.ImportSourceType_OSS:
		return constant.ImportDataSourceOSS
	}
	return constant.ImportDataSourceUnknown
}

func entityImportStatus2ConsoleState(entityResult constant.ImportStatus) console.ImportState_Enum {
	switch entityResult {
	case constant.ImportStatusUnknown:
		return console.ImportState_UNSPECIFIED
	case constant.ImportStatusPreparing:
		return console.ImportState_PREPARING
	case constant.ImportStatusImporting:
		return console.ImportState_IMPORTING
	case constant.ImportStatusCompleted:
		return console.ImportState_COMPLETED
	case constant.ImportStatusFailed:
		return console.ImportState_FAILED
	case constant.ImportStatusCanceling:
		return console.ImportState_CANCELING
	case constant.ImportStatusCanceled:
		return console.ImportState_CANCELED
	default:
		return console.ImportState_UNSPECIFIED
	}
}

func ImportEntity2ConsoleCreationDetails(imp *entity.Import) (*console.CreationDetails, error) {
	var csvFormat console.CSVFormat
	if imp.DataFormat == constant.ImportDataFormatCsv && imp.CSVFormat != "" {
		if err := json.Unmarshal([]byte(imp.CSVFormat), &csvFormat); err != nil {
			return nil, errors.Annotate(err, "parse custom CSV format failed")
		}
	}
	resp := &console.CreationDetails{
		ImportOptions: &console.ImportOptions{
			FileType: dataFormat2ConsoleFileType(imp.DataFormat),
			Format: &console.ImportOptions_CsvFormat{
				CsvFormat: &csvFormat,
			},
			NeedDropTable: imp.NeedDropTables,
		},
	}
	if imp.Name != "" {
		resp.ImportOptions.DisplayName = imp.Name
	}
	if imp.LocalImportCfg != nil {
		resp.Source = &console.ImportSource{
			Type: console.ImportSourceType_LOCAL,
			Source: &console.ImportSource_Local{
				Local: &console.LocalSource{
					UploadId:       imp.LocalImportCfg.UploadID,
					TargetTable:    imp.LocalImportCfg.TargetTable,
					TargetDatabase: imp.LocalImportCfg.TargetDB,
					FileName:       imp.LocalImportCfg.FileName,
				},
			},
		}
	}

	infos := make([]*console.ImportTargetTableInfo, 0, len(imp.TargetTableInfos.Infos))
	for _, info := range imp.TargetTableInfos.Infos {
		infos = append(infos, &console.ImportTargetTableInfo{
			TargetTable: &console.Table{
				Database: info.Schema,
				Table:    info.Table,
			},
			CustomFile: info.CustomFile,
		})
	}
	switch imp.DataSourceType {
	case constant.ImportDataSourceS3:
		resp.Source = &console.ImportSource{
			Type: console.ImportSourceType_S3,
			Source: &console.ImportSource_S3{
				S3: &console.S3Source{
					Uri:              imp.SourceURL,
					TargetTableInfos: infos,
				},
			},
		}
		if imp.S3RoleArn != "" {
			resp.GetSource().GetS3().AuthType = console.ImportS3AuthType_ROLE_ARN
			resp.Source.GetS3().Auth = &console.S3Source_RoleArn{
				RoleArn: imp.S3RoleArn,
			}
		} else if imp.S3AccessKeyID != "" {
			resp.GetSource().GetS3().AuthType = console.ImportS3AuthType_ACCESS_KEY
			resp.Source.GetS3().Auth = &console.S3Source_AccessKey_{
				AccessKey: &console.S3Source_AccessKey{
					Id: imp.S3AccessKeyID,
				},
			}
		}
	case constant.ImportDataSourceGCS:
		resp.Source = &console.ImportSource{
			Type: console.ImportSourceType_GCS,
			Source: &console.ImportSource_Gcs{
				Gcs: &console.GCSSource{
					Uri:              imp.SourceURL,
					AuthType:         console.ImportGcsAuthType_SERVICE_ACCOUNT_KEY,
					TargetTableInfos: infos,
				},
			},
		}
	case constant.ImportDataSourceAzBlob:
		resp.Source = &console.ImportSource{
			Type: console.ImportSourceType_AZURE_BLOB,
			Source: &console.ImportSource_AzureBlob{
				AzureBlob: &console.AzureBlobSource{
					Uri:              imp.SourceURL,
					AuthType:         console.ImportAzureBlobAuthType_SAS_TOKEN,
					TargetTableInfos: infos,
				},
			},
		}
	case constant.ImportDataSourceS3Compatible:
		resp.Source = &console.ImportSource{
			Type: console.ImportSourceType_S3_COMPATIBLE,
			Source: &console.ImportSource_S3Compatible{
				S3Compatible: &console.S3CompatibleSource{
					Uri:              imp.SourceURL,
					TargetTableInfos: infos,
				},
			},
		}
		if imp.S3Endpoint != "" {
			resp.GetSource().GetS3Compatible().Endpoint = aws.String(imp.S3Endpoint)
		}
		if imp.S3AccessKeyID != "" {
			resp.GetSource().GetS3Compatible().AuthType = console.ImportS3CompatibleAuthType_ACCESS_KEY
			resp.Source.GetS3Compatible().Auth = &console.S3CompatibleSource_AccessKey_{
				AccessKey: &console.S3CompatibleSource_AccessKey{
					Id: imp.S3AccessKeyID,
				},
			}
		}
	case constant.ImportDataSourceOSS:
		resp.Source = &console.ImportSource{
			Type: console.ImportSourceType_OSS,
			Source: &console.ImportSource_Oss{
				Oss: &console.OSSSource{
					Uri:              imp.SourceURL,
					TargetTableInfos: infos,
				},
			},
		}
		if imp.S3AccessKeyID != "" {
			resp.GetSource().GetOss().AuthType = console.ImportOSSAuthType_ACCESS_KEY
			resp.Source.GetOss().Auth = &console.OSSSource_AccessKey_{
				AccessKey: &console.OSSSource_AccessKey{
					Id: imp.S3AccessKeyID,
				},
			}
		}
	}

	return resp, nil
}

func dataFormat2ConsoleFileType(format constant.DataFormat) console.ImportFileType_Enum {
	switch format {
	case constant.ImportDataFormatCsv:
		return console.ImportFileType_CSV
	case constant.ImportDataFormatSql:
		return console.ImportFileType_SQL
	case constant.ImportDataFormatAuroraSnapshot:
		return console.ImportFileType_AURORA_SNAPSHOT
	case constant.ImportDataFormatParquet:
		return console.ImportFileType_PARQUET
	default:
		return console.ImportFileType_UNSPECIFIED
	}
}

func BuildConsoleAzureImport(
	ctx context.Context,
	req *console.CreateImportReq,
	cluster *tidbmgmtservicev1.Cluster,
) (*entity.ImportTask, bizerr.BizError) {
	_, _, bizErr := utils.ParseAzureURI(req.GetSource().GetAzureBlob().GetUri())
	if bizErr != nil {
		return nil, bizErr
	}

	imp := &entity.ImportTask{
		Import: &entity.Import{
			ID:             entity.GenerateIDWithPrefix(constant.ImportIDPrefix),
			ProjectID:      strconv.FormatUint(cluster.GetOriginalProjectId(), 10),
			TenantID:       strconv.FormatUint(cluster.OrganizationId, 10),
			ClusterID:      req.ClusterId,
			SourceURL:      req.GetSource().GetAzureBlob().GetUri(),
			DataSourceType: ConsoleDataSource2Entity(req.GetSource().GetType()),
			DataFormat:     ConsoleFileType2Entity(req.GetImportOptions().GetFileType()),
			NeedDropTables: req.GetImportOptions().GetNeedDropTable(),
			AzureSASToken:  req.GetSource().GetAzureBlob().GetSasToken(),
			TargetTableInfos: console2TableInfos(
				req.GetSource().GetAzureBlob().GetUri(),
				req.GetSource().GetAzureBlob().GetTargetTableInfos(),
			),
		},
		Cluster: cluster,
	}
	if req.GetImportOptions().GetDisplayName() != "" {
		imp.Name = req.GetImportOptions().GetDisplayName()
	}

	if req.GetImportOptions().GetFileType() == console.ImportFileType_CSV {
		csvBytes, err := json.Marshal(req.GetImportOptions().GetCsvFormat())
		if err != nil {
			log.Error(ctx, "failed to marshal CSV format", zap.Error(err))
			return nil, bizerr.New(errorcode.ErrCodeInternalError, errorcode.InternalErrorMessage)
		}
		imp.CSVFormat = string(csvBytes)
	}

	imp.Creator = utils.BuildCreatorID(ctx)

	return imp, nil
}

func BuildConsoleGCSImport(
	ctx context.Context,
	req *console.CreateImportReq,
	cluster *tidbmgmtservicev1.Cluster,
) (*entity.ImportTask, bizerr.BizError) {
	accountKey := req.Source.GetGcs().GetServiceAccountKey()
	decodeString, err := DecodeBase64(accountKey)
	if err != nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "invalid service account key")
	}
	imp := &entity.ImportTask{
		Import: &entity.Import{
			ID:             entity.GenerateIDWithPrefix(constant.ImportIDPrefix),
			ProjectID:      strconv.FormatUint(cluster.GetOriginalProjectId(), 10),
			TenantID:       req.OrgId,
			ClusterID:      req.ClusterId,
			SourceURL:      req.Source.GetGcs().GetUri(),
			DataSourceType: ConsoleDataSource2Entity(req.GetSource().GetType()),
			DataFormat:     ConsoleFileType2Entity(req.GetImportOptions().GetFileType()),
			NeedDropTables: req.GetImportOptions().GetNeedDropTable(),
			GCSCredentials: base64.URLEncoding.EncodeToString(decodeString),
			TargetTableInfos: console2TableInfos(
				req.Source.GetGcs().GetUri(),
				req.GetSource().GetGcs().GetTargetTableInfos(),
			),
		},
		Cluster: cluster,
	}

	if req.GetImportOptions().GetDisplayName() != "" {
		imp.Name = req.GetImportOptions().GetDisplayName()
	}

	if req.GetImportOptions().GetFileType() == console.ImportFileType_CSV {
		csvBytes, err := json.Marshal(req.GetImportOptions().GetCsvFormat())
		if err != nil {
			log.Error(ctx, "failed to marshal CSV format", zap.Error(err))
			return nil, bizerr.New(errorcode.ErrCodeInternalError, errorcode.InternalErrorMessage)
		}
		imp.CSVFormat = string(csvBytes)
	}

	imp.Creator = utils.BuildCreatorID(ctx)

	return imp, nil
}

func Console2CSVConfig(format *console.CSVFormat) *config.CSVConfig {
	if format == nil {
		return nil
	}

	c := &config.CSVConfig{
		Separator:         format.GetSeparator(),
		Delimiter:         format.GetDelimiter(),
		Header:            format.GetHeader(),
		TrimLastSep:       format.GetTrimLastSeparator(),
		NotNull:           format.GetNotNull(),
		Null:              config.StringOrStringSlice{format.GetNull()},
		HeaderSchemaMatch: true,
	}
	if format.GetBackslashEscape() {
		c.EscapedBy = `\`
	}

	return c
}

func Table2Console(table *agent.Table) *console.FullTable {
	rows := make([]*console.Row, 0, len(table.RowValues))
	for i := range table.RowValues {
		row := &console.Row{
			ColumnValues: table.RowValues[i].ColumnValues,
		}
		rows = append(rows, row)
	}
	return &console.FullTable{
		DatabaseName:  table.SchemaName,
		TableName:     table.TableName,
		RowHeaders:    table.RowHeaders,
		RowValues:     rows,
		RowDataTypes:  table.RowDataTypes,
		ColumnIsPk:    table.ColumnIsPk,
		ColumnAutoGen: table.ColumnAutoGen,
	}
}

func Console2Columns(columns []*console.ColumnInfo) []*entity.ColumnInfo {
	res := make([]*entity.ColumnInfo, 0, len(columns))
	for i := range columns {
		res = append(res, &entity.ColumnInfo{
			ColumnName: columns[i].ColumnName,
			ColumnType: columns[i].ColumnType,
		})
	}

	return res
}

func console2TableInfos(uri string, infos []*console.ImportTargetTableInfo) entity.TargetTableInfos {
	tables := make([]entity.TargetTableInfo, 0, len(infos))
	for i := range infos {
		info := entity.TargetTableInfo{
			Schema:     infos[i].GetTargetTable().GetDatabase(),
			Table:      infos[i].GetTargetTable().GetTable(),
			CustomFile: infos[i].CustomFile,
		}
		// one file import, set custom file to uri
		if !strings.HasSuffix(uri, "/") {
			info.CustomFile = uri
		}
		tables = append(tables, info)
	}

	return entity.TargetTableInfos{
		Infos: tables,
	}
}

// UniqueTable returns a unique table name.
func UniqueTable(schema, table string) string {
	var builder strings.Builder
	WriteMySQLIdentifier(&builder, strings.ToLower(schema))
	builder.WriteByte('.')
	WriteMySQLIdentifier(&builder, strings.ToLower(table))
	return builder.String()
}

// WriteMySQLIdentifier writes a MySQL identifier into the string builder.
// Writes a MySQL identifier into the string builder.
// The identifier is always escaped into the form "`foo`".
func WriteMySQLIdentifier(builder *strings.Builder, identifier string) {
	builder.Grow(len(identifier) + 2)
	builder.WriteByte('`')

	// use a C-style loop instead of range loop to avoid UTF-8 decoding
	for i := 0; i < len(identifier); i++ {
		b := identifier[i]
		if b == '`' {
			builder.WriteString("``")
		} else {
			builder.WriteByte(b)
		}
	}

	builder.WriteByte('`')
}

func ParseDBAndTable(input string) (dbname, tableName string, err error) {
	matches := tableNamePattern.FindStringSubmatch(input)

	if len(matches) != 3 {
		return "", "", fmt.Errorf("invalid format: %s", input)
	}

	dbname = matches[1]
	tableName = matches[2]

	return dbname, tableName, nil
}

func GetSourceUri(ctx context.Context, source *console.ImportSource) (string, bizerr.BizError) {
	if source.GetType() == console.ImportSourceType_S3 {
		return source.GetS3().GetUri(), nil
	} else if source.GetType() == console.ImportSourceType_GCS {
		return source.GetGcs().GetUri(), nil
	} else if source.GetType() == console.ImportSourceType_AZURE_BLOB {
		return source.GetAzureBlob().GetUri(), nil
	} else if source.GetType() == console.ImportSourceType_S3_COMPATIBLE {
		return source.GetS3Compatible().GetUri(), nil
	} else if source.GetType() == console.ImportSourceType_OSS {
		return source.GetOss().GetUri(), nil
	}

	return "", bizerr.New(bizerr.ErrCodeParams, "unsupported source type for cloud storage")
}

func GetTargetTableInfos(
	ctx context.Context,
	source *console.ImportSource,
) ([]*console.ImportTargetTableInfo, bizerr.BizError) {
	if source.GetType() == console.ImportSourceType_S3 {
		return source.GetS3().GetTargetTableInfos(), nil
	} else if source.GetType() == console.ImportSourceType_GCS {
		return source.GetGcs().GetTargetTableInfos(), nil
	} else if source.GetType() == console.ImportSourceType_AZURE_BLOB {
		return source.GetAzureBlob().GetTargetTableInfos(), nil
	} else if source.GetType() == console.ImportSourceType_S3_COMPATIBLE {
		return source.GetS3Compatible().GetTargetTableInfos(), nil
	} else if source.GetType() == console.ImportSourceType_OSS {
		return source.GetOss().GetTargetTableInfos(), nil
	}
	return nil, bizerr.New(bizerr.ErrCodeParams, "unsupported source type for cloud storage")
}