package service

import (
	"context"
	"encoding/json"
	stderrors "errors"
	"fmt"
	"net/url"
	"path/filepath"
	"regexp"
	"strconv"
	"strings"
	"time"
	"unicode/utf8"

	mapset "github.com/deckarep/golang-set/v2"
	"github.com/lithammer/shortuuid/v3"
	"github.com/pingcap/errors"
	"github.com/tidbcloud/eventbus-datastruct/event"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	config2 "github.com/tidbcloud/infra-provider/util/config"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/db"
	"github.com/tidbcloud/pkgv2/log"
	"github.com/tidbcloud/serverless-service/idl/pbgen/proto/v1beta"
	"github.com/tidbcloud/serverless-service/internal/config"
	"github.com/tidbcloud/serverless-service/internal/domain/constant"
	"github.com/tidbcloud/serverless-service/internal/domain/entity"
	"github.com/tidbcloud/serverless-service/internal/domain/facade"
	"github.com/tidbcloud/serverless-service/internal/domain/repo"
	"github.com/tidbcloud/serverless-service/internal/domain/vo"
	"github.com/tidbcloud/serverless-service/internal/errorcode"
	"github.com/tidbcloud/serverless-service/internal/infra/eventbus"
	"github.com/tidbcloud/serverless-service/internal/utils"
	serverlesscs "github.com/tidbcloud/serverless-service/pkg/cloud_storage"
	event_constant "github.com/tidbcloud/serverless-service/pkg/event"
	globalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/serverless/v1"
	tidbmgmtservicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"github.com/tidbcloud/tiinf/transport/tieventbus"
	"go.uber.org/zap"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)

const (
	FilePatternParquet = "parquet"
	FilePatternCSV     = "csv"
	FilePatternSQL     = "sql"

	empiricalCompressRatio = 8
	ImportStorageRequest   = "100Gi"
)

const lightningConfigTemplate = `
[checkpoint]
driver = "file"
dsn = "/etc/endpoint/cp.pb"
enable = true

[cron]
log-progress = "5m"
switch-mode = "5m"

[lightning]
check-requirements = true
file = ""
index-concurrency = 1
level = "info"
pprof-port = 8289
region-concurrency = 2
table-concurrency = 1

[mydumper]
strict-format = false

[post-restore]
analyze = "optional"
checksum = true
checksum-via-sql = false

[tidb]
build-stats-concurrency = 1
checksum-table-concurrency = 1
distsql-scan-concurrency = 1
index-serial-scan-concurrency = 1
log-level = "error"
port = 4000
status-port = 10080
tls = "false"
sql-mode = "ONLY_FULL_GROUP_BY,STRICT_TRANS_TABLES,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION"

[security]
redact-info-log = true

[tikv-importer]
on-duplicate = "ignore"
range-concurrency = 1
sorted-kv-dir = "/etc/endpoint"
`

type ImportService interface {
	CreateImport(ctx context.Context, imp *entity.ImportTask) (id string, err bizerr.BizError)
	GetImport(ctx context.Context, clusterID, importID string) (*entity.Import, bizerr.BizError)
	ListImports(ctx context.Context, opts *vo.ListImportOpt) (int64, []*entity.Import, bizerr.BizError)
	CancelImport(ctx context.Context, clusterID, importID string) bizerr.BizError
	GetDataIngestionProfile(
		ctx context.Context,
		cluster *tidbmgmtservicev1.Cluster,
	) (*entity.DataIngestionProfile, bizerr.BizError)
	GetImportByIDCompat(ctx context.Context, clusterID string, importIDCompat uint64) (*entity.Import, bizerr.BizError)
	CancelImportByIDCompat(ctx context.Context, clusterID string, importIDCompat uint64) bizerr.BizError
	NewInfraAPIImportObject(
		ctx context.Context,
		imp *entity.ImportTask,
		cloudAdminPassword string,
		isSuperVip bool,
	) (*cloudv1.Import, bizerr.BizError)
	PublishImportEvent(ctx context.Context, imp *entity.ImportTask, status event.EventStatus, eventType string)
}

type importService struct {
	encryptorFacade facade.EncryptorFacade
	eventbusService *eventbus.EventbusService
	infraFacade     facade.InfraFacade
	importRepo      repo.ImportRepo
	transProvider   db.TransProvider
	lockRepo        repo.LockRepo
	mgmtSvcFacade   facade.MgmtSvcFacade
}

func NewImportService(
	infraFacade facade.InfraFacade,
	encryptorFacade facade.EncryptorFacade,
	eventbusService *eventbus.EventbusService,
	transProvider db.TransProvider,
	lockRepo repo.LockRepo,
	importRepo repo.ImportRepo,
	mgmtSvcFacade facade.MgmtSvcFacade,
) ImportService {
	return &importService{
		encryptorFacade: encryptorFacade,
		eventbusService: eventbusService,
		infraFacade:     infraFacade,
		importRepo:      importRepo,
		transProvider:   transProvider,
		lockRepo:        lockRepo,
		mgmtSvcFacade:   mgmtSvcFacade,
	}
}

// CancelImport cancel the import
// we update the status to be canceling and wait it reach canceled if waitCanceled is true.
func (s *importService) CancelImport(ctx context.Context, clusterID, importID string) bizerr.BizError {
	imp, bizErr := s.importRepo.GetImportByID(ctx, clusterID, importID)
	if bizErr != nil {
		return bizErr
	}
	switch imp.Status {
	case constant.ImportStatusPreparing,
		constant.ImportStatusImporting:
	case constant.ImportStatusFailed,
		constant.ImportStatusCompleted:
		return bizerr.New(
			bizerr.ErrCodeParams,
			"The import cannot be canceled because it is in the %s state",
			imp.Status,
		)
	case constant.ImportStatusCanceling,
		constant.ImportStatusCanceled:
		return nil
	default:
		return bizerr.New(bizerr.ErrCodeParams, "invalid cluster status: %s", imp.Status)
	}

	imp.Status = constant.ImportStatusCanceling
	bizErr = s.importRepo.Update(ctx, imp, "status")
	if bizErr != nil {
		return bizErr
	}

	// publish cancel event
	clusterIDUint64, err := strconv.ParseUint(clusterID, 10, 64)
	if err != nil {
		return bizerr.New(bizerr.ErrCodeUnknown, "invalid cluster ID: %s", clusterID)
	}
	cluster, bizErr := s.mgmtSvcFacade.GetCluster(
		ctx,
		clusterIDUint64,
		tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_LOCAL,
	)
	if bizErr != nil {
		return bizErr
	}
	impTask := &entity.ImportTask{
		Import:  imp,
		Cluster: cluster,
	}
	go s.PublishImportEvent(ctx, impTask, event.EventStatusSucceeded, event_constant.EventTypeCancelImport)

	return nil
}

func (s *importService) CancelImportByIDCompat(
	ctx context.Context,
	clusterID string,
	importIDCompat uint64,
) bizerr.BizError {
	imp, bizErr := s.importRepo.GetImportByIDCompat(ctx, clusterID, importIDCompat)
	if bizErr != nil {
		return bizErr
	}
	switch imp.Status {
	case constant.ImportStatusPreparing,
		constant.ImportStatusImporting:
	case constant.ImportStatusFailed,
		constant.ImportStatusCompleted:
		return bizerr.New(
			bizerr.ErrCodeParams,
			"The import cannot be canceled because it is in the %s state",
			imp.Status,
		)
	case constant.ImportStatusCanceling,
		constant.ImportStatusCanceled:
		return nil
	default:
		return bizerr.New(bizerr.ErrCodeParams, "invalid cluster status: %s", imp.Status)
	}

	imp.Status = constant.ImportStatusCanceling
	bizErr = s.importRepo.Update(ctx, imp, "status")
	if bizErr != nil {
		return bizErr
	}

	// publish cancel event
	clusterIDUint64, err := strconv.ParseUint(clusterID, 10, 64)
	if err != nil {
		return bizerr.New(bizerr.ErrCodeUnknown, "invalid cluster ID: %s", clusterID)
	}
	cluster, bizErr := s.mgmtSvcFacade.GetCluster(
		ctx,
		clusterIDUint64,
		tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_LOCAL,
	)
	if bizErr != nil {
		return bizErr
	}
	impTask := &entity.ImportTask{
		Import:  imp,
		Cluster: cluster,
	}
	go s.PublishImportEvent(ctx, impTask, event.EventStatusSucceeded, event_constant.EventTypeCancelImport)

	return nil
}

func (s *importService) GetImport(ctx context.Context, clusterID, importID string) (*entity.Import, bizerr.BizError) {
	imp, bizErr := s.importRepo.GetImportByID(ctx, clusterID, importID)
	if bizErr != nil {
		return nil, bizErr
	}
	return imp, nil
}

func (s *importService) GetImportByIDCompat(
	ctx context.Context,
	clusterID string,
	importIDCompat uint64,
) (*entity.Import, bizerr.BizError) {
	imp, bizErr := s.importRepo.GetImportByIDCompat(ctx, clusterID, importIDCompat)
	if bizErr != nil {
		return nil, bizErr
	}
	return imp, nil
}

func (s *importService) ListImports(
	ctx context.Context,
	opts *vo.ListImportOpt,
) (int64, []*entity.Import, bizerr.BizError) {
	if opts == nil {
		return 0, nil, bizerr.New(bizerr.ErrCodeUnknown, "opts is required")
	}
	imports, err := s.importRepo.ListImportsByFilter(ctx, &entity.ImportFilter{
		ClusterID: opts.ClusterID,
		PageSize:  &opts.PageSize,
		Offset:    &opts.Offset,
		Statuses:  opts.Statuses,
		OrderBy:   opts.OrderBy,
	})
	if err != nil {
		return 0, nil, err
	}
	total, bizErr := s.importRepo.CountImports(ctx, &entity.ImportFilter{
		ClusterID: opts.ClusterID,
		Statuses:  opts.Statuses,
	})
	if bizErr != nil {
		return 0, nil, bizErr
	}

	return total, imports, nil
}

func (s *importService) CreateImport(ctx context.Context, imp *entity.ImportTask) (id string, bizErr bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.String("import_id", imp.ID))

	if imp.Cluster.GetServerlessCluster().UserStatus != globalserverlessv1.UserClusterStatus_USER_CLUSTER_STATUS_NORMAL {
		return "", bizerr.New(errorcode.ErrCodeClusterNotRunning, "Please import data when cluster is normal.")
	}

	var err error
	if imp.S3SecretAccessKey != "" {
		imp.S3SecretAccessKey, err = s.encryptorFacade.EncryptBase64(imp.S3SecretAccessKey)
		if err != nil {
			logger.Error("failed to encrypt s3 secret access key", zap.Error(err))
			return "", bizerr.New(bizerr.ErrCodeUnknown, "failed to create import")
		}
	} else if imp.GCSCredentials != "" {
		imp.GCSCredentials, err = s.encryptorFacade.EncryptBase64(imp.GCSCredentials)
		if err != nil {
			logger.Error("failed to encrypt gcs credentials", zap.Error(err))
			return "", bizerr.New(bizerr.ErrCodeUnknown, "failed to create import")
		}
	} else if imp.AzureSASToken != "" {
		imp.AzureSASToken, err = s.encryptorFacade.EncryptBase64(imp.AzureSASToken)
		if err != nil {
			logger.Error("failed to encrypt azure blob sas token", zap.Error(err))
			return "", bizerr.New(bizerr.ErrCodeUnknown, "failed to create import")
		}
	}

	err = s.transProvider.Transaction(ctx, func(ctx context.Context) (gerr error) {
		defer func() {
			if gerr != nil {
				// commit/rollback might have other errors that will replace the inner error. So log the error here
				logger.Error("create import record transaction has errors", zap.Error(gerr))
			} else {
				logger.Info("Tx logic has all finished. Ready to commit")
			}
		}()
		lock := &entity.Lock{
			Key:         fmt.Sprintf("create-import-cluster-%s", imp.ClusterID),
			TimeoutSecs: 10,
		}
		err := s.lockRepo.Lock(ctx, lock)
		if err != nil {
			logger.Error("failed to get lock", zap.Error(err))
			return bizerr.New(bizerr.ErrCodeDB, "failed to create import")
		}
		logger.Info("get lock successfully")

		defer func() {
			rctx := context.WithoutCancel(ctx)
			err := s.lockRepo.Release(rctx, lock)
			if err != nil {
				logger.Error("failed to release lock", zap.Error(err))
			} else {
				logger.Info("release lock successfully")
			}
		}()

		count, bizErr := s.importRepo.CountImports(ctx, &entity.ImportFilter{
			ClusterID: imp.ClusterID,
			Statuses: []constant.ImportStatus{
				constant.ImportStatusPreparing,
				constant.ImportStatusImporting,
				constant.ImportStatusCanceling,
			},
		})
		if bizErr != nil {
			logger.Error("failed to count imports", zap.Error(bizErr))
			return bizErr
		}
		if count > 0 {
			return bizerr.New(
				bizerr.ErrCodePermission,
				"There is already a running import task, please start a new task after the previous task is completed",
			)
		}

		uint64ClusterId, err := strconv.ParseUint(imp.ClusterID, 10, 64)
		if err != nil {
			logger.Error("failed to parse cluster id", zap.Error(err))
			return err
		}

		orgID := imp.Cluster.GetOrganizationId()
		set := mapset.NewSet(config.Global().BizConfig.SkipSetClusterStatusTenants...)
		if set.Contains(orgID) {
			logger.Info("org is whitelisted, skip setting cluster import status",
				zap.Uint64("org_id", orgID),
				zap.Uint64("cluster_id", uint64ClusterId),
			)
		} else {
			statusID, bizErr := s.mgmtSvcFacade.SetClusterImportStatus(ctx, uint64ClusterId)
			if bizErr != nil {
				if bizErr.ErrCode() == CustomStatusMutexErrorCode {
					return bizErr
				}
				logger.Error("failed to set cluster import status", zap.Error(bizErr))
				return bizErr
			}
			logger.Info("set cluster import status successfully", zap.Uint64("status_id", statusID))
			imp.ClusterStatusID = statusID
		}

		defer func() {
			if gerr != nil {
				rctx := context.WithoutCancel(ctx)
				if imp.ClusterStatusID != 0 {
					err = s.mgmtSvcFacade.DeleteClusterCustomStatus(rctx, uint64ClusterId, imp.ClusterStatusID)
					if err != nil {
						logger.Error(
							"failed to delete cluster custom status after create import failed",
							zap.Error(err),
						)
						return
					}
					logger.Info("deleted cluster custom status after create import failed")
				}
			}
		}()

		bizErr = s.importRepo.Create(ctx, imp.Import)
		if bizErr != nil {
			logger.Error("failed to create import record", zap.Error(bizErr))
			return bizErr
		}

		return nil
	})
	if err != nil {
		var e bizerr.BizError
		if stderrors.As(err, &e) {
			return "", e
		}

		logger.Error("failed to create import record", zap.Error(err))
		return "", bizerr.New(bizerr.ErrCodeDB, "failed to create import")
	}

	// publish create event
	go s.PublishImportEvent(ctx, imp, event.EventStatusStarted, event_constant.EventTypeCreateImport)

	return imp.ID, nil
}

func (s *importService) GetDataIngestionProfile(
	ctx context.Context,
	cluster *tidbmgmtservicev1.Cluster,
) (*entity.DataIngestionProfile, bizerr.BizError) {
	stage := config.Global().ServerConfig.Stage
	// Use cluster ID as external ID for alibaba cloud serverless cluster.
	switch cluster.CloudProvider {
	case config.ProviderAliCloud:
		aliProfile := serverlesscs.GetAliCloudProfile(stage)
		return &entity.DataIngestionProfile{
			AccountId:  aliProfile.AccountId,
			ExternalId: cluster.GetServerlessCluster().ClusterId,
		}, nil
	case config.ProviderAWS:
		profile := serverlesscs.GetAWSProfile(stage, cluster.Region)
		if profile == nil {
			return nil, bizerr.New(
				bizerr.ErrCodeUnknown,
				"%s",
				"Failed to get profile, invalid region: "+cluster.Region,
			)
		}
		return &entity.DataIngestionProfile{
			AccountId:  profile.AccountId,
			ExternalId: profile.ExternalId,
		}, nil
	case config.ProviderGCP:
		// GCS Role/HMAC profile is not wired yet; reuse cluster ID as
		// external ID the same way AliCloud does so callers (e.g. DM
		// GetVpcInfo) do not fail as "unsupported provider".
		return &entity.DataIngestionProfile{
			ExternalId: cluster.GetServerlessCluster().ClusterId,
		}, nil
	}
	return nil, bizerr.New(
		bizerr.ErrCodeUnknown,
		"Failed to get profile, provider %s, region %s is not supported",
		cluster.CloudProvider,
		cluster.Region,
	)
}

func (s *importService) NewInfraAPIImportObject(
	ctx context.Context,
	imp *entity.ImportTask,
	cloudAdminPassword string,
	isSuperVip bool,
) (*cloudv1.Import, bizerr.BizError) {
	if imp.Cluster == nil {
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "cluster is nil")
	}
	importSpec := genImportObjectSpec(imp)

	importObj, bizErr := s.newImportAPIObject(imp.ID, importSpec, imp.Cluster)
	if bizErr != nil {
		return nil, bizErr
	}
	if isSuperVip {
		log.Info(ctx, "Enable super VIP for import", zap.String("import_id", imp.ID))
		importObj.Annotations[constant.ImportSuperVIPAnonKey] = "true"
	}

	impCfgBytes, err := generateImportConfig(ctx, imp.Import, cloudAdminPassword, isSuperVip)
	if err != nil {
		errMsg := "Failed to generate import config"
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "%s:%s", errMsg, err.Error())
	}

	importObj.Spec.LightningConfig = string(impCfgBytes)
	return importObj, nil
}

func generateImportConfig(
	ctx context.Context,
	imp *entity.Import,
	cloudAdminPassword string,
	isSuperVip bool,
) ([]byte, error) {
	lightningCfg := new(cloudv1.TomlConfig)
	if err := lightningCfg.UnmarshalTOML([]byte(lightningConfigTemplate)); err != nil {
		return nil, errors.Trace(err)
	}

	cfg := config2.New(lightningCfg.MP)
	if isSuperVip {
		ratio, exist := config.Global().BizConfig.ImportTenantCompressRatio[imp.TenantID]
		if !exist {
			ratio = empiricalCompressRatio
		}
		cfg.Set("mydumper.empirical-compress-ratio", ratio)
		cfg.Set("tikv-importer.chunk-cache-in-memory", true)
	}
	cfg.Set("tidb.password", cloudAdminPassword)
	cfg.Set("mydumper.data-source-dir", imp.SourceURL)
	cfg, err := setCsvCfg(cfg, imp)
	if err != nil {
		return nil, err
	}
	cfg, err = setFilePattern(cfg, imp)
	if err != nil {
		return nil, err
	}
	cfg = setFilter(cfg, imp)
	log.Debug(ctx, "lightning config", zap.Any("config", cfg))

	return cfg.MarshalTOML()
}

func (s *importService) newImportAPIObject(
	importID string,
	spec *cloudv1.ImportSpec,
	cluster *tidbmgmtservicev1.Cluster,
) (*cloudv1.Import, bizerr.BizError) {
	if cluster == nil {
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "cluster is nil")
	}
	objectName := importID
	labels := map[string]string{
		cloudv1.TenantTagKey:        strconv.FormatUint(cluster.OrganizationId, 10),
		cloudv1.ProjectTagKey:       strconv.FormatUint(cluster.GetOriginalProjectId(), 10),
		cloudv1.ClusterTagKey:       strconv.FormatUint(cluster.ClusterId, 10),
		cloudv1.ProviderTypeLabel:   cluster.GetServerlessCluster().GetCluster().ProviderType,
		cloudv1.RegionLabel:         cluster.Region,
		constant.LabelKeyKubernetes: cluster.GetServerlessCluster().GetCluster().Kubernetes,
	}
	annotations := map[string]string{
		// TODO: remove the hard code after infra-api code has been merged
		"import.tidbcloud.com/network-name": fmt.Sprintf(
			"%s-free-tier-%s",
			cluster.CloudProvider,
			cluster.Region,
		),
		"import.tidbcloud.com/cluster-namespace":   s.infraFacade.Namespace(cluster.GetOriginalProjectId()),
		"tags.tidbcloud.com/migrated-from-central": "true",
	}
	return &cloudv1.Import{
		ObjectMeta: metav1.ObjectMeta{
			Namespace:   s.infraFacade.Namespace(cluster.GetOriginalProjectId()),
			Name:        objectName,
			Labels:      labels,
			Annotations: annotations,
		},
		Spec: *spec,
	}, nil
}

func setCsvCfg(cfg *config2.GenericConfig, imp *entity.Import) (*config2.GenericConfig, error) {
	if imp.DataFormat != constant.ImportDataFormatCsv || imp.CSVFormat == "" {
		return cfg, nil
	}
	cloneCfg := cfg.DeepCopy()

	var csvCfg v1beta.CustomCSVFormat
	if err := json.Unmarshal([]byte(imp.CSVFormat), &csvCfg); err != nil {
		return nil, err
	}
	cloneCfg.Set("mydumper.csv.separator", csvCfg.Separator)
	cloneCfg.Set("mydumper.csv.delimiter", csvCfg.Delimiter)
	cloneCfg.Set("mydumper.csv.header", csvCfg.Header)
	cloneCfg.Set("mydumper.csv.not-null", csvCfg.NotNull)
	cloneCfg.Set("mydumper.csv.null", csvCfg.Null)
	cloneCfg.Set("mydumper.csv.backslash-escape", csvCfg.BackslashEscape)
	cloneCfg.Set("mydumper.csv.trim-last-separator", csvCfg.TrimLastSeparator)
	if imp.DataSourceType == constant.ImportDataSourceLocal && csvCfg.Header {
		// For local imports, we need to set header-schema-match false to bypass header schema matching. This is necessary
		// because the table schema may have different column names than those in the data file,
		// as the column names in the data file might not adhere to standard naming conventions.
		cloneCfg.Set("mydumper.csv.header-schema-match", false)
	}

	return cloneCfg, nil
}

func setFilePattern(cfg *config2.GenericConfig, imp *entity.Import) (*config2.GenericConfig, error) {
	cloneCfg := cfg.DeepCopy()

	filePatterns := make([]map[string]string, 0)
	// imp.SourceURL should be escaped here(in syncer, sourceURL.String())
	if isOneFile(imp.SourceURL) {
		// remove the file name from the source url for single file import
		u, _ := url.Parse(imp.SourceURL)
		u.Path = filepath.Dir(u.Path) + "/"
		cloneCfg.Set("mydumper.data-source-dir", u.String())

		filePatterns = append(filePatterns, getFilePatternsInConfig(cfg)...)
		var filePatternLocalImport map[string]string
		if imp.LocalImportCfg != nil {
			// local import
			filePatternLocalImport = setFilePatternOneFile(
				imp.LocalImportCfg.FileName,
				imp.LocalImportCfg.TargetDB,
				imp.LocalImportCfg.TargetTable,
				constant.ConvertDataFormat2Str(imp.DataFormat),
			)
		} else {
			// import one file
			if len(imp.TargetTableInfos.Infos) == 0 {
				return nil, errors.New("no target table info")
			}
			info := imp.TargetTableInfos.Infos[0]
			filePatternLocalImport = setFilePatternOneFile(filepath.Base(info.CustomFile), info.Schema, info.Table,
				constant.ConvertDataFormat2Str(imp.DataFormat))
		}
		filePatterns = append(filePatterns, filePatternLocalImport)
	} else if len(imp.TargetTableInfos.Infos) == 0 {
		// no target table info
		filePatterns = append(filePatterns, getFilePatternsInConfig(cfg)...)
		filePatterns = append(filePatterns, setFilePatternWithNoTargetTable(imp)...)
	} else {
		// target table info exists
		filePatternsWithTargetTable := setFilePatternWithTargetTable(imp)
		filePatterns = append(filePatterns, filePatternsWithTargetTable...)
		filePatterns = append(filePatterns, getFilePatternsInConfig(cfg)...)
	}

	cloneCfg.Set("mydumper.files", filePatterns)
	return cloneCfg, nil
}

func setFilePatternOneFile(fileName, targetDatabase, targetTable, fileType string) map[string]string {
	compression := getFileCompression(fileName)

	filePattern := map[string]string{
		"pattern": regexp.QuoteMeta(fileName),
		"schema":  targetDatabase,
		"table":   targetTable,
		"type":    fileType,
	}
	if compression != "" {
		filePattern["compression"] = compression
	}

	return filePattern
}

func getFilePatternsInConfig(cfg *config2.GenericConfig) []map[string]string {
	val := cfg.Get("mydumper.files")
	if val == nil {
		return nil
	}
	jsonBytes, err := json.Marshal(val.Interface())
	if err != nil {
		return nil
	}
	result := []map[string]string{}
	if err := json.Unmarshal(jsonBytes, &result); err != nil {
		return nil
	}
	return result
}

func setFilePatternWithTargetTable(imp *entity.Import) []map[string]string {
	filePatterns := make([]map[string]string, 0)

	// custom file pattern should have a higher priority
	for _, info := range imp.TargetTableInfos.Infos {
		if info.CustomFile == "" {
			filePattern := map[string]string{
				"schema": info.Schema,
				"table":  info.Table,
			}
			escapedSchema := regexp.QuoteMeta(info.Schema)
			escapedTable := regexp.QuoteMeta(info.Table)
			switch imp.DataFormat {
			case constant.ImportDataFormatSql:
				filePattern["pattern"] = buildSQLFilePatternByTargetTable(escapedSchema, escapedTable)
				filePattern["type"] = FilePatternSQL
				filePattern["compression"] = "$1"
			case constant.ImportDataFormatCsv:
				filePattern["pattern"] = buildCSVFilePatternByTargetTable(escapedSchema, escapedTable)
				filePattern["type"] = FilePatternCSV
				filePattern["compression"] = "$1"
			case constant.ImportDataFormatAuroraSnapshot:
				filePattern["pattern"] = buildAuroraSnapshotFilePatternByTargetTable(escapedSchema, escapedTable)
				filePattern["type"] = FilePatternParquet
			case constant.ImportDataFormatParquet:
				filePattern["pattern"] = buildRawParquetFilePatternByTargetTable(escapedSchema, escapedTable)
				filePattern["type"] = FilePatternParquet
			}
			filePatterns = append(filePatterns, filePattern)

			continue
		}

		// extract the file pattern without the folder part
		pattern := genCustomNamingRuleRegexp(strings.TrimPrefix(info.CustomFile, imp.SourceURL))
		filePattern := map[string]string{
			"pattern": pattern,
			"schema":  utils.EscapeFieldExtractor(info.Schema),
			"table":   utils.EscapeFieldExtractor(info.Table),
			"type":    constant.ConvertDataFormat2Str(imp.DataFormat),
		}
		compression := getFileCompression(info.CustomFile)
		if compression != "" {
			filePattern["compression"] = compression
		}
		filePatterns = append(filePatterns, filePattern)
	}

	return filePatterns
}

func getFileCompression(fileName string) string {
	var compression string
	for _, c := range []string{"gzip", "gz", "snappy", "zstd", "zst"} {
		if strings.HasSuffix(fileName, "."+c) {
			compression = c
		}
	}
	return compression
}

func genCustomNamingRuleRegexp(wildcards string) string {
	// regexp syntax: https://en.wikipedia.org/wiki/Regular_expression
	var sb strings.Builder

	sb.WriteRune('^')

	cursor := 0
	maxlen := len(wildcards)

	for cursor < maxlen {
		r, sz := utf8.DecodeRuneInString(wildcards[cursor:])

		switch r {
		// defined in https://en.wikipedia.org/wiki/Glob_(programming)#Syntax
		case '{', '}', '(', ')', '^', '$', '.', '|', '+', '\\':
			sb.WriteString(`\`)
			sb.WriteRune(r)
		case '?':
			sb.WriteRune('.')
		case '*':
			sb.WriteString(".*")
		default:
			sb.WriteRune(r)
		}

		cursor += sz
	}

	sb.WriteRune('$')
	return sb.String()
}

func genImportObjectSpec(imp *entity.ImportTask) *cloudv1.ImportSpec {
	result := &cloudv1.ImportSpec{
		ClusterName: imp.ClusterID,
		Source: cloudv1.Source{
			// https://github.com/tidbcloud/aws-shared-provider/blob/bca66143cb6c97940b6c4f97a723ad3214534ac1/pkg/controllers/imports/env.go#L305
			// This field is used in aws-shared-provider.
			// To be compatible with aws-shared-provider, we need to set this field to "cloud_admin".
			// TODO: We will remove this field after aws-shared-provider is removed.
			UserName: constant.CloudAdminUsername,
			// This field is used in infra-provider
			// https://github.com/tidbcloud/infra-provider/blob/6511ba9aa15b2d8f82eb4b36a97840dddf93e259/infra-api/controllers/imports/import_controller.go#L682
			Format: cloudv1.DataFormat(imp.DataFormat),
		},
	}

	return result
}

func buildSQLFilePatternByTargetTable(schema, table string) string {
	return fmt.Sprintf(utils.DefaultSQLFilePatternWithTargetTable, schema, table)
}

func buildCSVFilePatternByTargetTable(schema, table string) string {
	return fmt.Sprintf(utils.DefaultCSVFilePatternWithTargetTable, schema, table)
}

func buildRawParquetFilePatternByTargetTable(schema, table string) string {
	return fmt.Sprintf(utils.DefaultRawParquetFilePatternWithTargetTable, schema, table)
}

func buildAuroraSnapshotFilePatternByTargetTable(schema, table string) string {
	return fmt.Sprintf(utils.DefaultAuroraSnapshotFilePatternWithTargetTable, schema, schema, table)
}

func setFilePatternWithNoTargetTable(imp *entity.Import) []map[string]string {
	filePatterns := make([]map[string]string, 0)

	// append the default file pattern
	fileSchema := map[string]string{
		"pattern":     utils.DefaultDBSchemaFilePattern,
		"schema":      "$1",
		"compression": "$2",
		"type":        "schema-schema",
	}
	fileTableSchema := map[string]string{
		"pattern":     utils.DefaultTableSchemaFilePattern,
		"schema":      "$1",
		"table":       "$2",
		"compression": "$3",
		"type":        "table-schema",
	}
	fileViewSchema := map[string]string{
		"pattern":     utils.DefaultViewSchemaFilePattern,
		"schema":      "$1",
		"table":       "$2",
		"compression": "$3",
		"type":        "view-schema",
	}
	filePatterns = append(filePatterns, fileSchema, fileTableSchema, fileViewSchema)

	switch imp.DataFormat {
	case constant.ImportDataFormatSql:
		filePatterns = append(filePatterns, map[string]string{
			"pattern":     utils.DefaultSQLFilePattern,
			"schema":      "$1",
			"table":       "$2",
			"type":        FilePatternSQL,
			"compression": "$3",
		})
	case constant.ImportDataFormatCsv:
		filePatterns = append(filePatterns, map[string]string{
			"pattern":     utils.DefaultCSVFilePattern,
			"schema":      "$1",
			"table":       "$2",
			"type":        FilePatternCSV,
			"compression": "$3",
		})
	case constant.ImportDataFormatAuroraSnapshot:
		filePatterns = append(filePatterns, map[string]string{
			"pattern": utils.DefaultAuroraSnapshotFilePattern,
			"schema":  "$1",
			"table":   "$2",
			"type":    FilePatternParquet,
		})
	case constant.ImportDataFormatParquet:
		filePatterns = append(filePatterns, map[string]string{
			"pattern": utils.DefaultRawParquetFilePattern,
			"schema":  "$1",
			"table":   "$2",
			"type":    FilePatternParquet,
		})
	}

	return filePatterns
}

func setFilter(cfg *config2.GenericConfig, imp *entity.Import) *config2.GenericConfig {
	cloneCfg := cfg.DeepCopy()

	var filters []string
	if imp.DataSourceType == constant.ImportDataSourceLocal {
		filters = make([]string, 0, 1)
		filters = append(filters, GenTableID(imp.LocalImportCfg.TargetDB, imp.LocalImportCfg.TargetTable))
	} else {
		if len(imp.TargetTableInfos.Infos) == 0 {
			return cfg
		}
		filters = make([]string, 0, len(imp.TargetTableInfos.Infos))
		for _, info := range imp.TargetTableInfos.Infos {
			filters = append(filters, GenTableID(info.Schema, info.Table))
		}
	}

	cloneCfg.Set("mydumper.filter", filters)

	return cloneCfg
}

func GenTableID(schema, table string) string {
	return fmt.Sprintf("`%s`.`%s`", utils.EscapeBackticks(schema), utils.EscapeBackticks(table))
}

func isOneFile(sourceUrl string) bool {
	u, _ := url.Parse(sourceUrl)
	return !strings.HasSuffix(u.Path, "/")
}

func BuildImportEvent(imp *entity.ImportTask,
	status event.EventStatus, eventType string,
) (*event.DisplayEvent, bizerr.BizError) {
	displayEvent := &event.DisplayEvent{
		HappenedAtMs: uint64(time.Now().UnixMilli()),
		Severity:     event.EventSeverityInfo,
		TenantID:     imp.Cluster.GetOrganizationId(),
		ProjectID:    imp.Cluster.GetProjectId(),
		ClusterID:    imp.Cluster.GetClusterId(),
		Author:       imp.Creator,

		DisplayStatus:    status,
		DisplayEventType: eventType,
	}

	var summary string
	switch imp.Status {
	case constant.ImportStatusPreparing, constant.ImportStatusImporting:
		summary = fmt.Sprintf("Import job %s started to run.", imp.ID)
	case constant.ImportStatusFailed:
		summary = fmt.Sprintf("Import job %s failed.", imp.ID)
	case constant.ImportStatusCompleted:
		summary = fmt.Sprintf("Import job %s succeeded.", imp.ID)
	case constant.ImportStatusCanceling, constant.ImportStatusCanceled:
		summary = fmt.Sprintf("Import job %s was canceled.", imp.ID)
	}
	displayEvent.Summary = summary
	importDetail := event_constant.ImportDetail{
		ID:             imp.ID,
		SourceURL:      imp.SourceURL,
		DataFormat:     strings.ToUpper(constant.ConvertDataFormat2Str(imp.DataFormat)),
		DataSourceType: strings.ToUpper(constant.ConvertDataSource2Str(imp.DataSourceType)),
	}
	switch imp.DataSourceType {
	case constant.ImportDataSourceS3:
		importDetail.S3AccessKeyID = imp.S3AccessKeyID
		importDetail.S3RoleArn = imp.S3RoleArn
	case constant.ImportDataSourceOSS:
		importDetail.OSSAccessKeyID = imp.S3AccessKeyID
	}
	detail, err := json.Marshal(importDetail)
	if err != nil {
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "marshal detail failed")
	}
	displayEvent.Detail = string(detail)
	return displayEvent, nil
}

func (s *importService) PublishImportEvent(
	ctx context.Context,
	imp *entity.ImportTask,
	status event.EventStatus,
	eventType string,
) {
	logger := log.WithContextL(ctx).With(zap.String("import_id", imp.ID))
	importEvent, err := BuildImportEvent(imp, status, eventType)
	if err != nil {
		logger.Error("failed to build import event", zap.Error(err))
		return
	}
	if err := s.eventbusService.Publish(&tieventbus.Event{
		ID:   shortuuid.New(),
		Type: event.DisplayEventTopic,
		Data: importEvent,
	}); err != nil {
		logger.Error("failed to publish events center", zap.Error(err))
		return
	}
}