package pimport

import (
	"context"
	"encoding/json"
	"fmt"
	"strings"
	"time"
	"unicode/utf8"

	"github.com/dustin/go-humanize"
	"github.com/pingcap/errcode"
	"github.com/pingcap/errors"
	"github.com/pingcap/tidb/pkg/errno"
	"github.com/pingcap/tidb/pkg/parser"
	"github.com/pingcap/tidb/pkg/parser/ast"
	"github.com/pingcap/tidb/pkg/parser/format"
	"github.com/pingcap/tidb/pkg/parser/mysql"
	agentpb "github.com/tidbcloud/dataflow-service/idl/pbgen/proto/agent"
	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/bizutils"
	"github.com/tidbcloud/dataflow-service/internal/common"
	"github.com/tidbcloud/dataflow-service/internal/db/entity"
	"github.com/tidbcloud/dataflow-service/internal/grpc/grpcli"
	"github.com/tidbcloud/dataflow-service/internal/utils"
	"github.com/tidbcloud/dataflow-service/pkg/azblob"
	pkggrpcli "github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli"
	infraapi "github.com/tidbcloud/dataflow-service/pkg/infra-api"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/log"
	"go.uber.org/zap"
	"google.golang.org/genproto/googleapis/rpc/errdetails"
)

// This error will not temporarily appear in the prod environment,
// because only the dev environment has enabled resource restriction.
var ErrExceededResourceGroup = "Exceeded resource group quota limitation. Please enlarge the quota and try again later. If you have any questions, please feel free to contact us."

// TableName defines the structued table name,
// separated into database and table.
type FullTableName struct {
	DatabaseName string
	TableName    string
}

// CheckWarningsByTable checks whether there are warnings for a table import.
func (c UserContext) CheckWarningsByTable(ctx context.Context, imp *ImportTask, tableName string) (string, error) {
	logger := c.log.With(log.FieldsFromContext(ctx)...)
	if err := LoadImportClusterInfo(ctx, imp, false); err != nil {
		return "", errors.Annotate(err, "load cluster information error")
	}
	ctx = common.ContextWithImport(ctx)
	_, agentClient, err := common.GetNetworkAgentByCluster(ctx, imp.Cluster)
	if err != nil {
		return "", errors.Annotate(err, "get agent client error")
	}
	tcNameGetter := infraapi.ClusterNameGetter{
		InfraProvider:        imp.Cluster.InfraProviderType,
		InfraAPIObjNamespace: infraapi.GetObjectNamespace(imp.Cluster.GetProjectId()),
		ClusterName:          infraapi.ClusterName(uint64(imp.ClusterID)),
	}

	password, err := c.GenCloudAdminPassword(ctx, logger, imp.Cluster.GetTenantId(), imp.Cluster.GetProjectId(), imp.ClusterID.Uint64())
	if err != nil {
		return "", errors.Annotatef(err, "failed to generate cloud admin password")
	}

	agentReq := &agentpb.ExecSQLReq{
		Cluster: &agentpb.Cluster{
			ClusterId:    imp.ClusterID.Uint64(),
			Namespace:    tcNameGetter.ClusterNamespace(),
			User:         entity.CloudAdminUser,
			Password:     password,
			ProviderType: imp.Cluster.GetInfraProviderType(),
		},
		// the table name and its column are from https://github.com/pingcap/tidb/blob/cdcbfc838ab3d4768d5761181ca9f3ee361811ce/br/pkg/lightning/errormanager/errormanager.go#L71-L84
		Sql:       fmt.Sprintf("SELECT 1 FROM `lightning_task_info`.`conflict_error_v1` WHERE task_id=%d AND table_name='%s' LIMIT 1;", imp.TaskID, tableName),
		ColumnNum: 1,
	}

	execCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
	defer cancel()
	agentResp, err := agentClient.ExecSQLOnCluster(execCtx, agentReq)
	if err != nil {
		// ignore the table doesn't exist error
		if strings.Contains(err.Error(), "doesn't exist") {
			return "", nil
		}
		return "", errors.Annotatef(err, "failed to execute sql on cluster")
	}
	if len(agentResp.ErrorMsg) != 0 {
		return "", errors.New(agentResp.ErrorMsg)
	}

	var resp string
	if len(agentResp.Rows) > 0 {
		resp = fmt.Sprintf("table %s exists some duplicate keys", tableName)
	}
	return resp, nil
}

func (m Model) CheckDataSourceAccess(
	ctx context.Context,
	clusterID uint64,
	awsRoleArn, sourceUrl string,
	awsAccessKeyId, awsSecretAccessKey string,
	azblobSASToken string,
) (string, entity.DataSource, error) {
	ctx = common.ContextWithImport(ctx)
	cluster, _, agentClient, err := common.GetClusterNetworkAgentById(ctx, clusterID)
	if err != nil {
		return "", entity.ImportDataSourceUnknown, errors.Annotate(err, "get agent client error")
	}
	switch cluster.InfraProviderType {
	case infraapi.ProviderTypeAWS, infraapi.ProviderTypeEKS:
		network, err := common.GetNetworkByCluster(ctx, cluster)
		if err != nil {
			return "", entity.ImportDataSourceUnknown, err
		}
		req := &agentpb.CheckS3AccessReq{
			RoleArn:            awsRoleArn,
			S3Url:              sourceUrl,
			ExternalId:         network.AWSExternalID(),
			AwsAccessKeyId:     awsAccessKeyId,
			AwsSecretAccessKey: awsSecretAccessKey,
		}
		resp, err := agentClient.CheckS3Access(ctx, req)
		if err != nil {
			return "", entity.ImportDataSourceUnknown, errors.Annotate(err, "check S3 access from agent error")
		}
		return resp.ErrorMsg, entity.ImportDataSourceS3, nil
	case infraapi.ProviderTypeGCP, infraapi.ProviderTypeGKE:
		var nodeRole string
		if cluster.GetAttributes().GetUseNewDedicatedArch() {
			var bizErr bizerr.BizError
			nodeRole, bizErr = pkggrpcli.InfraClient().GetClusterNodeRole(ctx, cluster.Id)
			if bizErr != nil {
				return "", entity.ImportDataSourceUnknown, errors.Annotate(bizErr, "get node role error")
			}
		}

		resp, err := agentClient.CheckGcsAccess(ctx, &agentpb.CheckGcsAccessReq{
			GcsUrl: sourceUrl,
			GcsOptions: &agentpb.GcsOptions{
				ImpersonateServiceAccount: nodeRole,
			},
		})
		if err != nil {
			return "", entity.ImportDataSourceUnknown, errors.Annotate(err, "check GCS access from agent error")
		}
		return resp.ErrorMsg, entity.ImportDataSourceGcs, nil
	case infraapi.ProviderTypeAKS:
		req := &agentpb.CheckAzblobAccessReq{
			AzblobUrl: sourceUrl,
			SasToken:  azblobSASToken,
		}
		resp, err := agentClient.CheckAzblobAccess(ctx, req)
		if err != nil {
			return "", entity.ImportDataSourceUnknown, errors.Annotate(err, "check Azblob access from agent error")
		}

		return resp.ErrorMsg, entity.ImportDataSourceAzblob, nil
	}
	return "", entity.ImportDataSourceUnknown, errors.New("unknown provider for the cluster")
}

func (c Model) doLightningPrecheck(
	ctx context.Context,
	imp *ImportTask,
	infraProviderType string,
) ([]*agentpb.CheckResult, error) {
	logger := c.log.With(log.FieldsFromContext(ctx)...)
	ctx = common.ContextWithImport(ctx)
	cluster, _, agentClient, err := common.GetClusterNetworkAgentById(ctx, uint64(imp.ClusterID))
	if err != nil {
		return nil, errors.Annotate(err, "get agent client error")
	}
	if err := adjustImportSourceURL(imp); err != nil {
		return nil, err
	}

	lnConfig, err := c.GetImportConfig(
		ctx,
		imp,
		&dataflow.ClusterIdReq{
			OrgId:     cluster.TenantId,
			ProjectId: cluster.ProjectId,
			ClusterId: cluster.Id,
		},
	)
	if err != nil {
		return nil, errors.Annotate(err, "failed to get lightning config")
	}
	lnConfigTOMLData, err := lnConfig.MarshalTOML()
	if err != nil {
		return nil, errors.Annotate(err, "encode Lightning config into TOML error")
	}

	tcNameGetter := infraapi.ClusterNameGetter{
		InfraProvider:        infraProviderType,
		InfraAPIObjNamespace: infraapi.GetObjectNamespace(cluster.ProjectId),
		ClusterName:          infraapi.ClusterName(uint64(imp.ClusterID)),
	}

	checkItems := []agentpb.CheckItem{
		agentpb.CheckItem_CHECK_ITEM_SOURCE_SCHEMA,
		agentpb.CheckItem_CHECK_ITEM_CSV_HEADER,
		agentpb.CheckItem_CHECK_ITEM_TARGET_TABLE_EMPTY,
		agentpb.CheckItem_CHECK_ITEM_TARGET_CLUSTER_SIZE,
	}

	clusterNamespace := tcNameGetter.ClusterNamespace()
	clusterNamePrefix := tcNameGetter.ClusterServicePrefix()

	agentClient.SetHttpTimeout(50 * time.Second)
	precheckResp, err := agentClient.LightningPrecheck(ctx, &agentpb.LightningPrecheckReq{
		LightningConfig:   string(lnConfigTOMLData),
		CheckItems:        checkItems,
		ClusterNamespace:  clusterNamespace,
		ClusterNamePrefix: clusterNamePrefix,
		GenGcpCredential:  infraProviderType == infraapi.ProviderTypeGKE,
		ProviderType:      cluster.GetInfraProviderType(),
	})
	if err != nil {
		// if timeout, return empty check results, all check items are success
		if strings.Contains(err.Error(), context.DeadlineExceeded.Error()) {
			return []*agentpb.CheckResult{}, nil
		}
		return nil, errors.Annotate(err, "lightning precheck from agent has errors")
	}
	respErrMsg := precheckResp.GetBaseResp().GetErrMsg()
	if len(respErrMsg) > 0 {
		logger.Warn("precheck response has some error messages returned", zap.String("error_message", respErrMsg))
	}
	return precheckResp.CheckResults, nil
}

func adjustImportSourceURL(imp *ImportTask) error {
	if imp.DataSourceType == entity.ImportDataSourceAzblob {
		azurl, err := azblob.ParseAzblobURL(imp.Import.SourceURL)
		if err != nil {
			return errors.Annotatef(err, "failed to parse source url: %s", imp.Import.SourceURL)
		}
		imp.LightningSourceURL = azurl.ToInnerURL(true)
	} else {
		imp.LightningSourceURL = imp.SourceURL
	}
	return nil
}

func (c Model) PrecheckImportFile(ctx context.Context, imp *ImportTask) error {
	logger := c.log.With(log.FieldsFromContext(ctx)...)
	ctx = common.ContextWithImport(ctx)
	cluster, network, agentClient, err := common.GetClusterNetworkAgentById(ctx, uint64(imp.ClusterID))
	if err != nil {
		return errors.Annotate(err, "get agent client error")
	}

	password, err := c.GenCloudAdminPassword(ctx, logger, cluster.TenantId, cluster.ProjectId, cluster.Id)
	if err != nil {
		return errors.Annotate(err, "generate cloud admin password error")
	}

	tcNameGetter := infraapi.ClusterNameGetter{
		InfraProvider:        cluster.InfraProviderType,
		InfraAPIObjNamespace: infraapi.GetObjectNamespace(imp.ProjectID.Uint64()),
		ClusterName:          infraapi.ClusterName(uint64(imp.ClusterID)),
	}

	_, accessConfig, err := imp.ParseSourceURL()
	if err != nil {
		logger.Info("cannot get access key info from entity's source URL", zap.Error(err))
		accessConfig = new(entity.ImportSourceAccessConfig)
	}

	csvCfg := dataflow.CustomCSVFormat{}
	if imp.CustomCSVFormat != "" {
		if err := json.Unmarshal([]byte(imp.CustomCSVFormat), &csvCfg); err != nil {
			logger.Error("unmarshal custom csv format failed", zap.Error(err), zap.String("csv format", imp.CustomCSVFormat))
			return err
		}
	}

	req := &agentpb.PrecheckImportFileReq{
		Cluster: &agentpb.Cluster{
			ClusterId:    cluster.Id,
			Namespace:    tcNameGetter.ClusterNamespace(),
			User:         entity.CloudAdminUser,
			Password:     password,
			ProviderType: cluster.GetInfraProviderType(),
		},

		CsvConfig: &agentpb.CsvConfig{
			Separator:         csvCfg.Separator,
			Delimiter:         csvCfg.Delimiter,
			Header:            csvCfg.Header,
			BackslashEscape:   csvCfg.BackslashEscape,
			TrimLastSeparator: csvCfg.GetTrimLastSeparator(),
		},

		DataFormat: uint64(imp.DataFormat),
		DataSource: uint64(imp.DataSourceType),

		AwsRoleArn:         imp.AwsRoleArn,
		AwsExternalId:      network.AWSExternalID(),
		AwsAccessKeyId:     accessConfig.AWSAccessKeyID,
		AwsSecretAccessKey: accessConfig.AWSSecretAccessKey,

		SasToken:  accessConfig.AzblobSASToken,
		SourceUrl: imp.SourceURL,
	}

	for _, info := range imp.TargetTableInfos.Infos {
		req.Sources = append(req.Sources, &agentpb.PrecheckImportFileReq_Source{
			Table: &commonpb.Table{
				Schema: info.Schema,
				Table:  info.Table,
			},
			File: info.MatchedFile,
		})
	}

	_, err = agentClient.PrecheckImportFile(ctx, req)

	logger.Info("PrecheckImportFile", zap.Any("req", req), zap.Error(err))

	return err
}

func (c *Model) CheckIsTableEmpty(ctx context.Context, clusterID uint64, table FullTableName) (bool, error) {
	logger := c.log.With(log.FieldsFromContext(ctx)...)
	ctx = common.ContextWithImport(ctx)
	cluster, _, agentClient, err := common.GetClusterNetworkAgentById(ctx, clusterID)
	if err != nil {
		return false, errors.Annotate(err, "get client and agent client error")
	}
	password, err := c.GenCloudAdminPassword(ctx, logger, cluster.TenantId, cluster.ProjectId, cluster.Id)
	if err != nil {
		return false, errors.Annotate(err, "generate cloud admin password error")
	}

	tcNameGetter := &infraapi.ClusterNameGetter{
		InfraProvider:        cluster.InfraProviderType,
		InfraAPIObjNamespace: infraapi.GetObjectNamespace(cluster.ProjectId),
		ClusterName:          infraapi.ClusterName(cluster.Id),
	}
	agentReq := &agentpb.ExecSQLReq{
		Cluster: &agentpb.Cluster{
			ClusterId:    cluster.Id,
			Namespace:    tcNameGetter.ClusterNamespace(),
			User:         entity.CloudAdminUser,
			Password:     password,
			ProviderType: cluster.GetInfraProviderType(),
		},
		Sql:       fmt.Sprintf("SELECT 1 FROM `%s`.`%s` USE INDEX() LIMIT 1;", table.DatabaseName, table.TableName),
		ColumnNum: 1,
	}

	execCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
	defer cancel()
	agentResp, err := agentClient.ExecSQLOnCluster(execCtx, agentReq)
	if err != nil {
		if strings.Contains(err.Error(), "doesn't exist") {
			logger.Warn("Ignore the NOT FOUNT error", zap.Error(err))
			return true, nil
		}
		return false, errors.Annotate(err, "execute SQL on target DB from agent error")
	}

	// if return a result, target table is not empty
	if len(agentResp.Rows) == 1 && len(agentResp.Rows[0].ColumnValues) == 1 {
		return false, nil
	}

	return true, nil
}

func (c *Model) GetTableColumns(ctx context.Context, req *dataflow.ImportsGetTableColumnsReq) (*agentpb.Table, error) {
	logger := c.log.With(log.FieldsFromContext(ctx)...)
	password, err := c.GenCloudAdminPassword(ctx, logger, req.OrgId, req.ProjectId, req.ClusterId)
	if err != nil {
		return nil, errors.Annotate(err, "generate cloud admin password error.")
	}

	ctx = common.ContextWithImport(ctx)
	cluster, _, agentClient, err := common.GetClusterNetworkAgentById(ctx, req.ClusterId)
	if err != nil {
		return nil, errors.Annotate(err, "get cluster, network, agent client failed")
	}

	tcNameGetter := &infraapi.ClusterNameGetter{
		InfraProvider:        cluster.InfraProviderType,
		InfraAPIObjNamespace: infraapi.GetObjectNamespace(req.ProjectId),
		ClusterName:          infraapi.ClusterName(cluster.Id),
	}
	agentReq := &agentpb.ExecSQLReq{
		Cluster: &agentpb.Cluster{
			ClusterId:    req.ClusterId,
			Namespace:    tcNameGetter.ClusterNamespace(),
			User:         entity.CloudAdminUser,
			Password:     password,
			ProviderType: cluster.GetInfraProviderType(),
		},
		Sql:       fmt.Sprintf("SHOW CREATE TABLE `%s`.`%s`;", req.Db, req.Table),
		ColumnNum: 2,
	}

	// shorten the timeout to get table structure
	agentClient.SetHttpTimeout(30 * time.Second)
	agentResp, err := agentClient.ExecSQLOnCluster(ctx, agentReq)
	if err != nil {
		if strings.Contains(err.Error(), "doesn't exist") {
			logger.Warn("Ignore the NOT FOUNT error", zap.Error(err))
			return &agentpb.Table{}, nil
		}
		return nil, errors.Annotate(err, "execute the SQL on target DB by agent has errors.")
	}
	if len(agentResp.Rows) == 0 {
		logger.Warn("sql result is empty", zap.String("db", req.Db), zap.String("table", req.Table))
		return &agentpb.Table{}, nil
	}
	if len(agentResp.Rows[0].ColumnValues) != 2 {
		return nil, errors.Annotatef(err, "sql result is unexpected.  Column values: %v", agentResp.Rows[0].ColumnValues)
	}

	createSql := agentResp.Rows[0].ColumnValues[1]
	stmt, err := parser.New().ParseOneStmt(createSql, "", "")
	if err != nil {
		return nil, errors.Annotatef(err, "failed to parse create table sql: %s", createSql)
	}
	createTableStmt, ok := stmt.(*ast.CreateTableStmt)
	if !ok {
		return nil, errors.Annotatef(err, "the statement is not create table: %s", createSql)
	}
	result := &agentpb.Table{
		SchemaName: req.Db,
		TableName:  req.Table,
	}
	for _, col := range createTableStmt.Cols {
		result.RowHeaders = append(result.RowHeaders, col.Name.Name.O)
		result.RowDataTypes = append(result.RowDataTypes, strings.ToUpper(col.Tp.CompactStr()))
		result.ColumnIsPk = append(result.ColumnIsPk, columnIsPkuk(col, createTableStmt.Constraints))
		result.ColumnAutoGen = append(result.ColumnAutoGen, columnIsAutoGen(col))
	}

	return result, nil
}

func (c *Model) CreateTargetTable(ctx context.Context,
	orgId, projectId, clusterId uint64,
	localImportCfg *dataflow.LocalImportCfg,
) (err error) {
	logger := c.log.With(log.FieldsFromContext(ctx)...)
	if utf8.RuneCountInString(localImportCfg.TargetTable.Schema) > mysql.MaxDatabaseNameLength {
		return errcode.NewInvalidInputErr(errors.Errorf("The schema name(%s)'s length should be less than or equal to %d", localImportCfg.TargetTable.Schema, mysql.MaxDatabaseNameLength))
	}
	if utf8.RuneCountInString(localImportCfg.TargetTable.Table) > mysql.MaxTableNameLength {
		return errcode.NewInvalidInputErr(errors.Errorf("The table name(%s)'s length should be less than or equal to %d", localImportCfg.TargetTable.Table, mysql.MaxTableNameLength))
	}
	cols := make(map[string]struct{})
	for _, col := range localImportCfg.Columns {
		if utf8.RuneCountInString(col.ColumnName) > mysql.MaxColumnNameLength {
			return errcode.NewInvalidInputErr(errors.Errorf("The column name(%s)'s length should be less than or equal to %d", col.ColumnName, mysql.MaxColumnNameLength))
		}
		if _, ok := cols[col.ColumnName]; ok {
			return errcode.NewInvalidInputErr(errors.Errorf("There are two same column names(%s).", col.ColumnName))
		} else {
			cols[col.ColumnName] = struct{}{}
		}
	}

	password, err := c.GenCloudAdminPassword(ctx, logger, orgId, projectId, clusterId)
	if err != nil {
		logger.Error("failed to generate cloud admin password", zap.Error(err))
		return errors.Annotate(err, "generate cloud admin password error.")
	}

	ctx = common.ContextWithImport(ctx)
	cluster, _, agentClient, err := common.GetClusterNetworkAgentById(ctx, clusterId)
	if err != nil {
		logger.Error("get cluster, network, agent client failed", zap.Error(err))
		return errors.Annotate(err, "get cluster, network, agent client failed")
	}

	tcNameGetter := &infraapi.ClusterNameGetter{
		InfraProvider:        cluster.InfraProviderType,
		InfraAPIObjNamespace: infraapi.GetObjectNamespace(projectId),
		ClusterName:          infraapi.ClusterName(cluster.Id),
	}

	agentClient.SetHttpTimeout(30 * time.Second)

	createDBSQL := generateCreateDatabaseSQL(localImportCfg.TargetTable.Schema)
	createTableSQL, err := generateCreateTableSQL(
		logger,
		localImportCfg.TargetTable.Schema,
		localImportCfg.TargetTable.Table,
		localImportCfg.GetColumns(),
		localImportCfg.GetPkColumns(),
	)
	if err != nil && len(localImportCfg.GetColumns()) != 0 {
		logger.Error("generate create table SQL error", zap.Error(err), zap.Reflect("columns", localImportCfg.GetColumns()))
		return err
	}

	if _, err := agentClient.ExecSQLOnCluster(ctx,
		&agentpb.ExecSQLReq{
			Cluster: &agentpb.Cluster{
				ClusterId:    clusterId,
				Namespace:    tcNameGetter.ClusterNamespace(),
				User:         entity.CloudAdminUser,
				Password:     password,
				ProviderType: cluster.GetInfraProviderType(),
			},
			Sql: createDBSQL,
		},
	); err != nil {
		if strings.Contains(strings.ToLower(err.Error()), strings.ToLower(errno.MySQLErrName[errno.ErrResourceGroupThrottled].Raw)) {
			return errcode.NewForbiddenErr(errors.New(ErrExceededResourceGroup))
		}
		return errors.Annotate(err, "execute SQL to create DB error")
	}

	if len(localImportCfg.GetColumns()) == 0 {
		return nil
	}
	if _, err := agentClient.ExecSQLOnCluster(ctx,
		&agentpb.ExecSQLReq{
			Cluster: &agentpb.Cluster{
				ClusterId:    clusterId,
				Namespace:    tcNameGetter.ClusterNamespace(),
				User:         entity.CloudAdminUser,
				Password:     password,
				ProviderType: cluster.GetInfraProviderType(),
			},
			Sql: createTableSQL,
		},
	); err != nil {
		if strings.Contains(err.Error(), "Error 1071") {
			return errcode.NewInvalidInputErr(errors.Errorf("%s. Please specify a valid primary key.", err.Error()))
		}
		if strings.Contains(strings.ToLower(err.Error()), strings.ToLower(errno.MySQLErrName[errno.ErrResourceGroupThrottled].Raw)) {
			return errcode.NewForbiddenErr(errors.New(ErrExceededResourceGroup))
		}
		logger.Error("execute SQL to create table error", zap.Error(err))
		return errors.Annotate(err, "execute SQL to create table error")
	}
	return nil
}

func generateCreateDatabaseSQL(dbName string) string {
	return fmt.Sprintf("CREATE DATABASE IF NOT EXISTS `%s`;",
		dbName,
	)
}

func generateCreateTableSQL(logger *zap.Logger, dbName string, tableName string, columns []*dataflow.ColumnInfo, pkCols []string) (string, error) {
	colDefStrs := make([]string, 0, len(columns))
	colMap := make(map[string]struct{})
	for i, colInfo := range columns {
		if len(colInfo.GetColumnName()) <= 0 {
			return "", errcode.NewInvalidInputErr(errors.Errorf("The %s column name is empty.", humanize.Ordinal(i+1)))
		}
		colDefStrs = append(colDefStrs, fmt.Sprintf("`%s` %s", EscapeName(colInfo.GetColumnName()), colInfo.GetColumnType()))
		colMap[strings.ToLower(colInfo.ColumnName)] = struct{}{}
	}

	// check whether pkCol is in the columns
	for i, pkCol := range pkCols {
		// the column name is case-insensitive
		if _, ok := colMap[strings.ToLower(pkCol)]; !ok {
			return "", errcode.NewInvalidInputErr(errors.Errorf("Primary key column '%s' cannot be found in the table's column list.", pkCol))
		}

		pkCols[i] = fmt.Sprintf("`%s`", EscapeName(pkCol))
	}
	var pkStr string
	if len(pkCols) > 0 {
		pkStr = fmt.Sprintf(", PRIMARY KEY( %s ) CLUSTERED", strings.Join(pkCols, ", "))
	}
	rawCreateTableStmtStr := fmt.Sprintf("CREATE TABLE IF NOT EXISTS `%s`.`%s` ( %s %s );",
		EscapeName(dbName),
		EscapeName(tableName),
		strings.Join(colDefStrs, ", "),
		pkStr,
	)
	createStmtAST, err := parser.New().ParseOneStmt(rawCreateTableStmtStr, "", "")
	if err != nil {
		logger.Error("parse create statement error", zap.Error(err))
		return "", errcode.NewInvalidInputErr(errors.New("The table cannot be created due to invalid table name, column names or column types. Please check these settings."))
	}
	var sb strings.Builder
	if err := createStmtAST.Restore(format.NewRestoreCtx(format.DefaultRestoreFlags, &sb)); err != nil {
		return "", errors.Annotate(err, "restore AST into SQL string error")
	}
	return sb.String(), nil
}

func (c *Model) PreviewCsvFile(ctx context.Context, req *dataflow.ImportsPreviewCsvFileReq) (*dataflow.ImportsPreviewCsvFileResp, error) {
	ctx = common.ContextWithImport(ctx)
	_, network, agentClient, err := common.GetClusterNetworkAgentById(ctx, req.ClusterId)
	if err != nil {
		return nil, errors.Annotate(err, "get cluster, network, agent client failed")
	}

	preq := &agentpb.PreviewCsvFileReq{
		SourceUri: req.SourceUri,
		CsvConfig: &agentpb.CsvConfig{
			Separator:         req.CsvFormat.Separator,
			Delimiter:         req.CsvFormat.Delimiter,
			Header:            req.CsvFormat.Header,
			BackslashEscape:   req.CsvFormat.BackslashEscape,
			TrimLastSeparator: req.CsvFormat.GetTrimLastSeparator(),
		},
		FirstN:             req.FirstN,
		AwsRoleArn:         req.AwsRoleArn,
		AwsExternalId:      network.AWSExternalID(),
		AwsAccessKeyId:     req.AwsAccessKeyId,
		AwsSecretAccessKey: req.AwsSecretAccessKey,
	}

	agentResp, err := agentClient.PreviewCsvFile(ctx, preq)
	if err != nil {
		if strings.Contains(err.Error(), "syntax error") {
			return nil, errcode.NewInvalidInputErr(errors.New("There is a syntax error. Please edit the 'CSV Configuration' to enable the program to parse and preview the source file content correctly."))
		}
		return nil, err
	}

	// targetTable contains the schema info.
	targetTable, err := c.GetTableColumns(ctx, &dataflow.ImportsGetTableColumnsReq{
		OrgId:     req.OrgId,
		ProjectId: req.ProjectId,
		ClusterId: req.ClusterId,
		Db:        req.TargetTable.Schema,
		Table:     req.TargetTable.Table,
	})
	if err != nil {
		return nil, err
	}

	// check header

	// due to the order of the columns in the target table may not be the same as the order of the columns in the csv file
	// so we need to sort the columns according to the order of the columns in the target table
	targetTable.RowValues = reorderDataColumns(agentResp.Table.RowValues, agentResp.Table.RowHeaders, targetTable.RowHeaders)

	resp := &dataflow.ImportsPreviewCsvFileResp{
		Table: targetTable,
	}
	return resp, nil
}

func (c *Model) SearchImportFiles(ctx context.Context, req *dataflow.ImportsSearchImportFilesReq) (*dataflow.ImportsSearchImportFilesResp, error) {
	logger := c.log.With(log.FieldsFromContext(ctx)...)
	ctx = common.ContextWithImport(ctx)

	cluster, bizErr := grpcli.GetClusterByID(ctx, req.ClusterId)
	if bizErr != nil {
		errMsg := "Failed to get cluster information. Please check whether the cluster is cleared or deleted."
		logger.Warn(errMsg, zap.Error(bizErr))
		if bizErr := bizutils.AppendErrorDetails(ctx, &errdetails.ErrorInfo{
			Reason: bizErr.Error(),
		}); bizErr != nil {
			logger.Warn("append error detail error", zap.Error(bizErr))
		}
		return nil, utils.WrapAsErrorCodeErr(bizErr, errMsg)
	}

	network, err := common.GetNetworkByCluster(ctx, cluster)
	if err != nil {
		return nil, err
	}

	// if cluster.InfraProviderType != infraapi.ProviderTypeAKS && network != nil && len(network.DataflowAgentUrl) <= 0 {
	// 	return nil, errors.Errorf("cannot find the agent URL for this cluster: %d", cluster.Id)
	// }
	agentClient, err := common.GetAgentClientByNetwork(ctx, cluster.Id, network)
	if err != nil {
		logger := log.WithContextS(ctx)
		errMsg := "Network error. If it still fails after several attempts, please contact our support team."
		logger.Error(errMsg, zap.Error(err))
		return nil, utils.WrapAsErrorCodeErr(err, errMsg)
	}

	agentReq := &agentpb.SearchImportFilesReq{
		TargetSource:       make([]*agentpb.SearchImportFilesReq_TargetSource, 0, len(req.TargetTableInfos)),
		DataFormat:         req.DataFormat,
		DataSourceType:     req.DataSourceType,
		SourceUri:          req.SourceUri,
		AwsRoleArn:         req.AwsRoleArn,
		AwsExternalId:      network.AWSExternalID(),
		AwsAccessKeyId:     req.AwsAccessKeyId,
		AwsSecretAccessKey: req.AwsSecretAccessKey,
		GcsOptions: &agentpb.GcsOptions{
			ImpersonateServiceAccount: network.GCPServiceAccountEmail(),
		},
		SasToken: req.AzblobSasToken,
	}

	for _, tableMapping := range req.TargetTableInfos {
		agentReq.TargetSource = append(agentReq.TargetSource, &agentpb.SearchImportFilesReq_TargetSource{
			TargetTable: tableMapping.GetTargetTable(),
			CustomFile:  tableMapping.CustomFile,
		})
	}

	agentClient.SetHttpTimeout(50 * time.Second)
	agentResp, err := agentClient.SearchImportFiles(ctx, agentReq)
	logger.Info("validate target source match", zap.Any("agentResp", agentResp), zap.Error(err))
	if err != nil {
		return nil, errors.Annotate(err, "execute the SQL on target DB by agent has errors.")
	}

	resp := &dataflow.ImportsSearchImportFilesResp{
		TargetTables: make([]*dataflow.ImportsSearchImportFilesResp_TableMatch, 0, len(agentResp.TableMatches)),
	}

	warningTables := make([]string, 0)
	for _, match := range agentResp.TableMatches {
		var warning string
		if len(match.SourceUri) == 0 {
			// this message will not be used
			warning = "source uri can't find"
			warningTables = append(warningTables, GenTableID(match.TargetTable.Schema, match.TargetTable.Table))
		}
		resp.TargetTables = append(resp.TargetTables, &dataflow.ImportsSearchImportFilesResp_TableMatch{
			TargetTable:    match.TargetTable,
			SourceUri:      match.SourceUri,
			WarningMessage: warning,
		})
	}

	if len(warningTables) > 0 {
		logger.Warn("some tables have no source uri", zap.Strings("warning tables", warningTables))
		warningTableMsg := strings.Join(warningTables, ", ")
		if len(warningTables) > 3 {
			warningTableMsg = strings.Join(warningTables[:3], ", ") + " etc"
		}
		resp.WarningMessage = fmt.Sprintf("Tables include \"%s\" that with warning icons indicate no matching source files. Resolve this by providing the correct source file, renaming the existing one according to <a href=\"https://docs.pingcap.com/tidbcloud/naming-conventions-for-data-import\" target=\"_blank\">ducumentation</a>, or using Advanced settings mapping. Ignoring the warning will result in no data being imported to those tables.", warningTableMsg)
	}

	return resp, nil
}