package service

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"net/url"
	"slices"
	"strconv"
	"strings"
	"time"

	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/feature/s3/manager"
	"github.com/aws/aws-sdk-go-v2/service/s3"
	mapset "github.com/deckarep/golang-set/v2"
	"github.com/lithammer/shortuuid/v3"
	filter "github.com/pingcap/tidb/pkg/util/table-filter"
	"github.com/tidbcloud/eventbus-datastruct/event"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/clusterproxy"
	"github.com/tidbcloud/pkgv2/db"
	"github.com/tidbcloud/pkgv2/import/cloud_storage"
	"github.com/tidbcloud/pkgv2/log"
	"github.com/tidbcloud/pkgv2/tracing"
	"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"
	stashing "github.com/tidbcloud/serverless-service/internal/infra/stashing_area"
	"github.com/tidbcloud/serverless-service/internal/metric"
	"github.com/tidbcloud/serverless-service/internal/utils"
	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.opentelemetry.io/otel"
	"go.uber.org/zap"
	api_errors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)

const (
	exportCleanTime = 73 * time.Hour            // 3 days + 1 hour
	MaxExportSize   = 1024 * 1024 * 1024 * 1024 // 1T
)

var Tracer = otel.Tracer("domain.export")

const InvalidExportCreatePayloadMessage = "invalid export create payload"

type ExportService interface {
	CreateExport(ctx context.Context, exportVO *vo.Export) (*vo.Export, bizerr.BizError)
	NewInfraAPIExportObject(
		ctx context.Context,
		export *entity.Export,
		cluster *tidbmgmtservicev1.Cluster,
	) (*cloudv1.Export, bizerr.BizError)
	ListExports(ctx context.Context, opt *vo.ListExportOpt) ([]*vo.Export, uint32, bizerr.BizError)
	GetExport(ctx context.Context, clusterID, exportID string) (*vo.Export, bizerr.BizError)
	DeleteExport(ctx context.Context, clusterID, exportID string) (*vo.Export, bizerr.BizError)
	CancelExport(ctx context.Context, clusterID, exportID string) (*vo.Export, bizerr.BizError)
	ParseTableFilter(ctx context.Context, clusterID string, patterns []string) ([]string, bizerr.BizError)
}

type exportService struct {
	exportRepo                 repo.ExportRepo
	infraFacade                facade.InfraFacade
	accountFacade              facade.AccountFacade
	encryptorFacade            facade.EncryptorFacade
	eventbusService            *eventbus.EventbusService
	stashingAreaProvider       *stashing.StashingAreaProvider
	clusterProxyProviderFacade facade.ClusterProxyProviderFacade
	transProvider              db.TransProvider
	mgmtSvc                    facade.MgmtSvcFacade
	featureGateFacade          facade.FeatureGate
}

func NewExportService(exportRepo repo.ExportRepo, infraFacade facade.InfraFacade,
	accountFacade facade.AccountFacade, mgmtSvc facade.MgmtSvcFacade,
	encryptorFacade facade.EncryptorFacade, eventbusService *eventbus.EventbusService,
	stashingAreaProvider *stashing.StashingAreaProvider, clusterProxyProviderFacade facade.ClusterProxyProviderFacade,
	transProvider db.TransProvider, featureGateFacade facade.FeatureGate,
) ExportService {
	return &exportService{
		exportRepo:                 exportRepo,
		infraFacade:                infraFacade,
		accountFacade:              accountFacade,
		encryptorFacade:            encryptorFacade,
		eventbusService:            eventbusService,
		stashingAreaProvider:       stashingAreaProvider,
		clusterProxyProviderFacade: clusterProxyProviderFacade,
		transProvider:              transProvider,
		mgmtSvc:                    mgmtSvc,
		featureGateFacade:          featureGateFacade,
	}
}

func (e *exportService) exceedConcurrencyLimit(ctx context.Context, tenantID string) bizerr.BizError {
	// This is intentionally a soft limit: concurrent requests can observe the same running set
	// and temporarily exceed the configured limit. It avoids serializing export creation per organization.
	limit := config.Global().BizConfig.MaxExportConcurrencyPerOrg
	exports, err := e.exportRepo.ListByFilter(ctx, entity.ExportFilter{
		TenantID: tenantID,
		States: []constant.ExportState{
			constant.ExportStatePreparing,
			constant.ExportStateRunning,
		},
	})
	if err != nil {
		return err
	}
	if int64(len(exports)) >= limit {
		return bizerr.New(
			errorcode.ErrCodeExportConcurrencyLimitExceeded,
			"This organization can run up to %d concurrent exports.",
			limit,
		)
	}
	return nil
}

func (e *exportService) CreateExport(ctx context.Context, exportVO *vo.Export) (_ *vo.Export, bizErr bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "exportService.CreateExport")
	defer func() {
		tracing.EndSpan(span, bizErr)
	}()
	logger := log.WithContextL(ctx)
	logger.Info("create export",
		zap.String("cluster_id", exportVO.ClusterID),
		zap.String("target_type", string(exportVO.TargetType)),
		zap.String("display_name", exportVO.DisplayName),
	)

	exportVO.ApplyDefaultValue()
	span.AddEvent("check export options")
	if err := exportVO.CheckOptions(ctx); err != nil {
		logger.Error("check export options failed", zap.Error(err))
		return nil, err
	}

	span.AddEvent("check cluster")
	cluster, err := e.checkCluster(ctx, exportVO.ClusterID)
	if err != nil {
		logger.Error("check cluster failed", zap.Error(err))
		return nil, err
	}

	// check if the cluster is throttled
	if cluster.GetServerlessCluster().GetCluster().GetAnnotations()[constant.AnnoRuLimitMode] == constant.RuLimitModeThrottled {
		return nil, bizerr.New(errorcode.ErrCodeExportFromThrottled, "")
	}

	span.AddEvent("check export target")
	if err := e.checkTarget(ctx, exportVO, cluster); err != nil {
		logger.Error("check export target failed", zap.Error(err))
		return nil, err
	}

	span.AddEvent("convert to entity")
	export, err := convertToEntity(ctx, exportVO)
	if err != nil {
		logger.Error("convert to entity failed", zap.Error(err))
		return nil, err
	}
	export.TenantID = strconv.FormatUint(cluster.OrganizationId, 10)
	export.ProjectID = strconv.FormatUint(cluster.GetOriginalProjectId(), 10)
	export.Region = cluster.Region
	export.CloudProvider = cluster.CloudProvider
	export.Kubernetes = cluster.GetServerlessCluster().GetCluster().GetKubernetes()
	export.ProxyEndpoint = cluster.GetServerlessCluster().GetCluster().GetProxyEndpoint()

	encryptedCreatePayload, err1 := e.encryptExportCreatePayload(exportVO)
	if err1 != nil {
		logger.Error("encrypt export create payload failed", zap.Error(err1))
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "failed to create export")
	}
	export.CreatePayload = encryptedCreatePayload

	span.AddEvent("check export concurrency soft limit")
	if err := e.exceedConcurrencyLimit(ctx, export.TenantID); err != nil {
		logger.Error("export concurrency soft limit check failed", zap.Error(err))
		return nil, err
	}

	if !slices.Contains(
		config.Global().BizConfig.SkipExportSizeCheckTenants,
		cluster.GetOrganizationId(),
	) {
		span.AddEvent("estimate export size")
		exportCR, err := e.buildExportCR(ctx, cluster, export, exportVO)
		if err != nil {
			logger.Error("build export CR for size check failed", zap.Error(err))
			return nil, err
		}
		if err := e.estimateExportSize(ctx, cluster, exportCR); err != nil {
			logger.Error("check export size failed", zap.Error(err))
			return nil, err
		}
	}

	uint64ClusterId, err1 := strconv.ParseUint(exportVO.ClusterID, 10, 64)
	if err1 != nil {
		logger.Error("failed to parse cluster id", zap.Error(err1))
		return nil, bizerr.New(bizerr.ErrCodeParams, "invalid cluster id")
	}

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

	cleanupClusterStatus := func() {
		if export.ClusterStatusID == 0 {
			return
		}
		rctx := context.WithoutCancel(ctx)
		if delErr := e.mgmtSvc.DeleteClusterCustomStatus(rctx, uint64ClusterId, export.ClusterStatusID); delErr != nil {
			logger.Error("failed to delete cluster custom status after create export failed", zap.Error(delErr))
		} else {
			logger.Info("deleted cluster custom status after create export failed")
		}
	}

	if err := e.transProvider.Transaction(ctx, func(ctx context.Context) error {
		span.AddEvent("create export in db")
		if err := e.exportRepo.Create(ctx, export); err != nil {
			logger.Error("create export failed", zap.Error(err))
			return err
		}
		return nil
	}); err != nil {
		logger.Error("create export transaction failed", zap.Error(err))
		cleanupClusterStatus()
		var transactionErr bizerr.BizError
		if errors.As(err, &transactionErr) {
			return nil, transactionErr
		}
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "")
	}

	span.AddEvent("create export accepted")
	logger.Info("create export accepted", zap.String("exportId", export.ID))

	// publish create event
	go func(ctx context.Context, export *entity.Export) {
		logger := log.WithContextL(ctx)
		exportEvent, err := BuildExportEvent(
			export,
			cluster,
			event.EventStatusStarted,
			event_constant.EventTypeCreateExport,
		)
		if err != nil {
			logger.Error("failed to build export event", zap.Error(err))
			return
		}
		if err := e.eventbusService.Publish(&tieventbus.Event{
			ID:   shortuuid.New(),
			Type: event.DisplayEventTopic,
			Data: exportEvent,
		}); err != nil {
			logger.Error("failed to publish events center", zap.Error(err))
			return
		}
	}(ctx, export)

	return convertToVO(export)
}

func (e *exportService) NewInfraAPIExportObject(
	ctx context.Context,
	export *entity.Export,
	cluster *tidbmgmtservicev1.Cluster,
) (*cloudv1.Export, bizerr.BizError) {
	exportVO, err := e.decryptExportCreatePayload(export.CreatePayload)
	if err != nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, InvalidExportCreatePayloadMessage)
	}
	return e.buildExportCR(ctx, cluster, export, exportVO)
}

func (e *exportService) ListExports(
	ctx context.Context,
	opt *vo.ListExportOpt,
) ([]*vo.Export, uint32, bizerr.BizError) {
	if opt == nil {
		return nil, 0, bizerr.New(bizerr.ErrCodeUnknown, "opt is required")
	}
	// check opt
	if opt.ClusterID == "" {
		return nil, 0, bizerr.New(errorcode.ErrCodeEmptyClusterID, "")
	}

	logger := log.WithContextL(ctx).With(zap.String("cluster_id", opt.ClusterID))

	// list exports
	exportFilter := entity.ExportFilter{
		ClusterID: opt.ClusterID,
		Offset:    &opt.Offset,
		PageSize:  &opt.PageSize,
		OrderBy:   opt.OrderBy,
		NameLike:  opt.NameLike,
	}
	exports, err := e.exportRepo.ListByFilter(ctx, exportFilter)
	if err != nil {
		logger.Error("list exports failed", zap.Error(err))
		return nil, 0, err
	}

	// count exports
	count, err := e.exportRepo.CountByFilter(ctx, exportFilter)
	if err != nil {
		logger.Error("count exports failed", zap.Error(err))
		return nil, 0, err
	}

	exportVos := make([]*vo.Export, 0, len(exports))
	for _, export := range exports {
		exportVo, err := convertToVO(export)
		if err != nil {
			return nil, 0, err
		}
		exportVos = append(exportVos, exportVo)
	}
	return exportVos, uint32(count), nil
}

// estimateExportSize estimates the size of the export data using cloud admin.
func (e *exportService) estimateExportSize(
	ctx context.Context,
	cluster *tidbmgmtservicev1.Cluster,
	exportCR *cloudv1.Export,
) bizerr.BizError {
	logger := log.WithContextL(ctx).With(zap.Uint64("cluster_id", cluster.ClusterId))

	tableFilters := exportCR.Spec.Options.TableFilter
	where := exportCR.Spec.Options.Where
	sql := exportCR.Spec.Options.SQL
	partitions := exportCR.Spec.Options.Partition
	if sql != "" || where != "" || len(partitions) > 0 {
		logger.Info("skip estimate size because sql, where or partitions is not empty")
		return nil
	}
	if len(tableFilters) == 0 {
		tableFilters = []string{"*.*"}
	}

	logger.Info("start to estimate export size", zap.Any("tableFilters", tableFilters))

	// build cluster proxy
	operator := &clusterproxy.ClusterOperator{
		Username:   exportCR.Spec.Secret.User,
		AuthMethod: clusterproxy.AuthMethodPassword,
		Credential: exportCR.Spec.Secret.Password,
	}
	proxy, err := e.clusterProxyProviderFacade.GetClusterProxy(ctx, cluster, operator)
	if err != nil {
		logger.Error("get cluster proxy failed", zap.Error(err))
		return bizerr.New(bizerr.ErrCodeUnknown, "")
	}

	// get all tables
	dbTables, err := queryDbTables(ctx, proxy, 0)
	if err != nil {
		return err
	}
	// ignore error because it should has been checked before
	filter, _ := filter.Parse(tableFilters)
	tables := filterDbTables(dbTables, filter)

	// query tables size
	size, err := queryTablesSize(ctx, proxy, tables)
	if err != nil {
		return err
	}

	// check size
	// 1. when exceed the limit, super vip can bypass the limit and use more resources in the data plane
	// 2. if not exceed the limit, all exports use the same resource as it should be in the data plane
	if size > MaxExportSize {
		if e.featureGateFacade.FeatureEnabled(constant.FeatureExportSuperVIP, cluster.OrganizationId, cluster.Region) {
			exportCR.Annotations[constant.AnnoExportSuperVip] = "true"
		} else {
			metric.ExportSizeGuage.WithLabelValues(exportCR.Name, "true").Set(float64(size) / (1024 * 1024))
			return bizerr.New(errorcode.ErrCodeExportStorageLimitExceed, "")
		}
	}
	metric.ExportSizeGuage.WithLabelValues(exportCR.Name, "false").Set(float64(size) / (1024 * 1024))
	return nil
}

func queryTablesSize(
	ctx context.Context,
	proxy facade.ClusterProxyFacade,
	dbTables map[string][]string,
) (int64, bizerr.BizError) {
	logger := log.WithContextL(ctx)

	if len(dbTables) == 0 {
		return 0, bizerr.New(errorcode.ErrCodeEmptyExport, "")
	}

	sql := "SELECT SUM(DATA_LENGTH) as total FROM information_schema.TABLES WHERE"
	index := 0
	for db, tables := range dbTables {
		for _, table := range tables {
			if index == 0 {
				index++
				sql += fmt.Sprintf(" (TABLE_SCHEMA = '%s' AND TABLE_NAME = '%s')", db, table)
			} else {
				sql += fmt.Sprintf(" OR (TABLE_SCHEMA = '%s' AND TABLE_NAME = '%s')", db, table)
			}
		}
	}
	logger.Info("query sql", zap.String("sql", sql))

	// query the size from information_schema.
	res, err := proxy.QuerySQL(ctx, sql)
	if err != nil {
		logger.Error("failed to query sql", zap.Error(err))
		return 0, err
	}

	// parse the result
	type Result struct {
		Total [][]byte `json:"total"`
	}
	result := Result{}
	err1 := json.Unmarshal(res, &result)
	if err1 != nil {
		logger.Error("failed to unmarshal json", zap.Error(err1))
		return 0, bizerr.New(bizerr.ErrCodeUnknown, "failed to unmarshal json")
	}
	if len(result.Total) == 0 || len(result.Total[0]) == 0 {
		logger.Info("estimate export size is 0, regard as database or table not find")
		return 0, bizerr.New(errorcode.ErrCodeEmptyExport, "")
	}
	// go database/sql scan will stringifies the numeric result. So we only need to parse the string to int here.
	logger.Info("estimate export size", zap.String("total size", string(result.Total[0])))
	total, err1 := strconv.ParseInt(string(result.Total[0]), 10, 64)
	if err1 != nil {
		logger.Error("failed to parse int from result", zap.String("result", string(res)), zap.Error(err1))
		return 0, bizerr.New(bizerr.ErrCodeUnknown, "")
	}
	return total, nil
}

func (e *exportService) GetExport(ctx context.Context, clusterID, exportID string) (*vo.Export, bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.String("clusterID", clusterID), zap.String("exportID", exportID))

	// get export
	export, err := e.exportRepo.GetById(ctx, clusterID, exportID)
	if err != nil {
		logger.Error("get export failed", zap.Error(err))
		return nil, err
	}

	return convertToVO(export)
}

func (e *exportService) DeleteExport(ctx context.Context, clusterID, exportID string) (*vo.Export, bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.String("clusterID", clusterID), zap.String("exportID", exportID))

	// get export
	export, err := e.exportRepo.GetById(ctx, clusterID, exportID)
	if err != nil {
		logger.Error("get export failed", zap.Error(err))
		return nil, err
	}

	preState := export.State

	if export.State.IsActiveState() {
		logger.Error("export is running, cannot be deleted")
		return nil, bizerr.New(errorcode.ErrCodeDeleteRunningExport, "")
	}

	// delete export CR
	projectIDUint64, err1 := strconv.ParseUint(export.ProjectID, 10, 64)
	if err1 != nil {
		logger.Error("parse projectID failed", zap.Error(err1))
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "parse projectID failed")
	}
	proxyEndpoint, bizErr := e.getExportProxyEndpoint(ctx, export)
	if bizErr != nil {
		logger.Error("get proxy endpoint failed", zap.Error(bizErr))
		return nil, bizErr
	}
	err1 = e.infraFacade.DeleteExport(ctx, proxyEndpoint, projectIDUint64, exportID)
	if err1 != nil {
		logger.Error("delete export cr failed", zap.Error(err1))
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "delete export cr")
	}

	// delete export
	export, err = e.exportRepo.SoftDelete(ctx, export)
	if err != nil {
		logger.Error("delete export failed", zap.Error(err))
		return nil, err
	}

	logger.Info("delete export success")

	// publish delete event
	go func(ctx context.Context, export *entity.Export) {
		logger := log.WithContextL(ctx)
		clusterIDUint64, err := strconv.ParseUint(clusterID, 10, 64)
		if err != nil {
			return
		}
		cluster, bizErr := e.mgmtSvc.GetCluster(
			ctx,
			clusterIDUint64,
			tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_LOCAL,
		)
		if bizErr != nil {
			logger.Error("failed to get cluster", zap.String("clusterID", clusterID), zap.Error(bizErr))
			return
		}
		exportEvent, err := BuildExportEvent(
			export,
			cluster,
			event.EventStatusSucceeded,
			event_constant.EventTypeDeleteExport,
		)
		if err != nil {
			logger.Error("failed to build export event", zap.Error(err))
			return
		}
		if err := e.eventbusService.Publish(&tieventbus.Event{
			ID:   shortuuid.New(),
			Type: event.DisplayEventTopic,
			Data: exportEvent,
		}); err != nil {
			logger.Error("failed to publish events center", zap.Error(err))
			return
		}
	}(ctx, export)

	// clean stashing area
	if export.TargetType == constant.ExportTargetTypeLocal && preState == constant.ExportStateSucceed {
		go e.CleanExport(context.WithoutCancel(ctx), export)
	}

	return convertToVO(export)
}

func (e *exportService) CancelExport(ctx context.Context, clusterID, exportID string) (*vo.Export, bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.String("clusterID", clusterID), zap.String("exportID", exportID))

	oldExport, bizErr := e.exportRepo.GetById(ctx, clusterID, exportID)
	if bizErr != nil {
		logger.Error("get export failed", zap.Error(bizErr))
		return nil, bizErr
	}
	if oldExport.State == constant.ExportStateCanceled {
		logger.Error("export already be canceled")
		return nil, bizerr.New(errorcode.ErrCodeAlreadyCancelExport, "")
	}
	if oldExport.State == constant.ExportStatePreparing {
		oldExport.State = constant.ExportStateCanceled
		transitioned, err := e.exportRepo.TransitionPreparingState(ctx, oldExport)
		if err != nil {
			logger.Error("cancel preparing export failed", zap.Error(err))
			return nil, err
		}
		if transitioned {
			e.cleanupPreparingClusterStatus(ctx, oldExport)
			e.publishCancelExportEvent(ctx, oldExport)
			return convertToVO(oldExport)
		}

		// The create worker may have moved the export to running concurrently.
		oldExport, bizErr = e.exportRepo.GetById(ctx, clusterID, exportID)
		if bizErr != nil {
			logger.Error("reload export after cancel race failed", zap.Error(bizErr))
			return nil, bizErr
		}
	}
	if oldExport.State != constant.ExportStateRunning {
		logger.Error("cancel not running export", zap.String("state", string(oldExport.State)))
		return nil, bizerr.New(errorcode.ErrCodeCancelNotRunningExport, "")
	}

	// cancel export in transaction
	oldExport.State = constant.ExportStateCanceled
	var export *entity.Export
	err := e.transProvider.Transaction(ctx, func(ctx context.Context) error {
		// update export db
		export, bizErr = e.exportRepo.UpdateRunningState(ctx, oldExport)
		if bizErr != nil {
			logger.Error("cancel export failed", zap.Error(bizErr))
			return bizErr
		}
		proxyEndpoint, bizErr := e.getExportProxyEndpoint(ctx, export)
		if bizErr != nil {
			logger.Error("failed to get proxy endpoint", zap.Error(bizErr))
			return bizErr
		}
		// update export cr
		projectId, err := strconv.ParseUint(export.ProjectID, 10, 64)
		if err != nil {
			logger.Error("failed to parse projectID", zap.String("projectId", export.ProjectID), zap.Error(err))
			return bizerr.New(bizerr.ErrCodeUnknown, "failed to parse projectID")
		}
		bizErr = e.updateExportCRWithRetry(
			ctx,
			proxyEndpoint,
			projectId,
			exportID,
			func(exportCR *cloudv1.Export) *cloudv1.Export {
				exportCR.Annotations[constant.ExportStatusAnnotation] = string(cloudv1.ExportPhaseCancel)
				return exportCR
			},
		)
		if bizErr != nil {
			logger.Error("failed to update export cr", zap.Error(bizErr))
			return bizErr
		}
		return nil
	})
	if err != nil {
		logger.Error("cancel export failed", zap.Error(err))
		var e bizerr.BizError
		if errors.As(err, &e) {
			return nil, e
		}
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "")
	}

	e.publishCancelExportEvent(ctx, export)

	return convertToVO(export)
}

func (e *exportService) cleanupPreparingClusterStatus(ctx context.Context, export *entity.Export) {
	logger := log.WithContextL(ctx).With(zap.String("export_id", export.ID))
	if export.ClusterStatusID == 0 {
		return
	}
	clusterID, err := strconv.ParseUint(export.ClusterID, 10, 64)
	if err != nil {
		logger.Error("failed to parse cluster id while cleaning canceled preparing export", zap.Error(err))
		return
	}
	rctx := context.WithoutCancel(ctx)
	deleteErr := e.mgmtSvc.DeleteClusterCustomStatus(rctx, clusterID, export.ClusterStatusID)
	if deleteErr != nil && deleteErr.ErrCode() != bizerr.ErrCodeRecordNotFound {
		logger.Error("failed to delete cluster status for canceled preparing export", zap.Error(deleteErr))
		return
	}
	if clearErr := e.exportRepo.ClearClusterStatusID(rctx, export.ID, export.ClusterID, export.ClusterStatusID); clearErr != nil {
		logger.Error("failed to clear cluster status id for canceled preparing export", zap.Error(clearErr))
		return
	}
	export.ClusterStatusID = 0
}

func (e *exportService) publishCancelExportEvent(ctx context.Context, export *entity.Export) {
	go func(ctx context.Context, export *entity.Export) {
		logger := log.WithContextL(ctx)
		clusterIDUint64, err := strconv.ParseUint(export.ClusterID, 10, 64)
		if err != nil {
			return
		}
		cluster, bizErr := e.mgmtSvc.GetCluster(
			ctx,
			clusterIDUint64,
			tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_LOCAL,
		)
		if bizErr != nil {
			logger.Error("failed to get cluster", zap.String("clusterID", export.ClusterID), zap.Error(bizErr))
			return
		}
		exportEvent, err := BuildExportEvent(
			export,
			cluster,
			event.EventStatusSucceeded,
			event_constant.EventTypeCancelExport,
		)
		if err != nil {
			logger.Error("failed to build export event", zap.Error(err))
			return
		}
		if err := e.eventbusService.Publish(&tieventbus.Event{
			ID:   shortuuid.New(),
			Type: event.DisplayEventTopic,
			Data: exportEvent,
		}); err != nil {
			logger.Error("failed to publish events center", zap.Error(err))
			return
		}
	}(ctx, export)
}

// updateExportCRWithRetry update export CR with retry, the retry will read without cache
func (e *exportService) updateExportCRWithRetry(
	ctx context.Context,
	proxyEndpoint string,
	projectId uint64,
	exportId string,
	ef func(export *cloudv1.Export) *cloudv1.Export,
) bizerr.BizError {
	logger := log.WithContextL(ctx).With(zap.String("export_id", exportId))
	err := func() error {
		exportCR, err := e.infraFacade.GetExport(ctx, proxyEndpoint, projectId, exportId)
		if err != nil {
			return err
		}
		exportCR = ef(exportCR)
		err = e.infraFacade.UpdateExport(ctx, proxyEndpoint, exportCR)
		if err != nil {
			return err
		}
		return nil
	}()

	if err == nil {
		return nil
	}
	if !api_errors.IsConflict(err) && !api_errors.IsNotFound(err) {
		logger.Error("failed to update export CR", zap.Error(err))
		return bizerr.New(bizerr.ErrCodeSvcInfraAPI, "")
	}
	// retry with direct read
	logger.Info("failed to update export CR, retry with direct read", zap.Error(err))
	exportCR, err := e.infraFacade.GetExportDirectly(ctx, proxyEndpoint, projectId, exportId)
	if err != nil {
		logger.Error("failed to get export CR with direct read", zap.Error(err))
		return bizerr.New(bizerr.ErrCodeSvcInfraAPI, "")
	}
	exportCR = ef(exportCR)
	err = e.infraFacade.UpdateExport(ctx, proxyEndpoint, exportCR)
	if err != nil {
		logger.Error("failed to update export CR with direct read", zap.Error(err))
		return bizerr.New(bizerr.ErrCodeSvcInfraAPI, "")
	}
	return nil
}

func (e *exportService) getClusterProxyEndpoint(ctx context.Context, clusterID string) (string, bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.String("clusterID", clusterID))
	clusterIDUint64, err := strconv.ParseUint(clusterID, 10, 64)
	if err != nil {
		logger.Error("parse clusterID failed", zap.Error(err))
		return "", bizerr.New(bizerr.ErrCodeUnknown, "parse clusterID failed")
	}
	cluster, bizErr := e.mgmtSvc.GetCluster(
		ctx,
		clusterIDUint64,
		tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC,
	)
	if bizErr != nil {
		logger.Error("failed to get cluster", zap.Error(bizErr))
		return "", bizErr
	}

	serverlessCluster := cluster.GetServerlessCluster()
	if serverlessCluster == nil || serverlessCluster.GetCluster() == nil {
		logger.Error("invalid cluster runtime when resolving proxy endpoint")
		return "", bizerr.New(errorcode.ErrCodeInvalidClusterID, "")
	}
	proxyEndpoint := serverlessCluster.GetCluster().GetProxyEndpoint()
	if proxyEndpoint == "" {
		logger.Error("proxy endpoint is empty")
		return "", bizerr.New(errorcode.ErrCodeInvalidClusterID, "")
	}

	return proxyEndpoint, nil
}

func (e *exportService) getExportProxyEndpoint(ctx context.Context, export *entity.Export) (string, bizerr.BizError) {
	if export.ProxyEndpoint != "" {
		return export.ProxyEndpoint, nil
	}

	proxyEndpoint, bizErr := e.getClusterProxyEndpoint(ctx, export.ClusterID)
	if bizErr != nil {
		return "", bizErr
	}
	if bizErr := e.exportRepo.UpdateProxyEndpoint(ctx, export.ID, export.ClusterID, proxyEndpoint); bizErr != nil {
		return "", bizErr
	}
	export.ProxyEndpoint = proxyEndpoint
	return proxyEndpoint, nil
}

func (e *exportService) checkCluster(
	ctx context.Context,
	clusterID string,
) (*tidbmgmtservicev1.Cluster, bizerr.BizError) {
	logger := log.WithContextL(ctx)
	if clusterID == "" {
		return nil, bizerr.New(errorcode.ErrCodeEmptyClusterID, "")
	}
	cid, err := strconv.ParseUint(clusterID, 10, 64)
	if err != nil {
		logger.Error("parse clusterID failed", zap.String("clusterID", clusterID), zap.Error(err))
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "parse clusterID failed")
	}
	cluster, bizErr := e.mgmtSvc.GetClusterIncludeDedicated(
		ctx,
		cid,
		tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC,
	)
	if bizErr != nil {
		logger.Error("failed to get cluster from mgmt svc", zap.Error(bizErr))
		return nil, bizErr
	}
	if cluster.DeployType != tidbmgmtservicev1.DeployType_DEPLOY_TYPE_SERVERLESS {
		logger.Error("cluster not serverless")
		return nil, bizerr.New(errorcode.ErrCodeUnsupportedClusterType, "")
	}
	if cluster.GetServerlessCluster().
		GetUserStatus() !=
		globalserverlessv1.UserClusterStatus_USER_CLUSTER_STATUS_NORMAL {
		logger.Error("cluster not running")
		return nil, bizerr.New(errorcode.ErrCodeClusterNotRunning, "")
	}
	return cluster, nil
}

func convertToEntity(ctx context.Context, e *vo.Export) (*entity.Export, bizerr.BizError) {
	export := &entity.Export{
		ID:             e.ID,
		State:          e.State,
		FileType:       e.FileType,
		TargetType:     e.TargetType,
		S3BucketURI:    e.S3URI,
		S3AccessKeyID:  e.S3AccessKeyID,
		LocalURL:       e.LocalURL,
		CreatedAt:      e.CreatedAt,
		UpdatedAt:      e.UpdatedAt,
		ClusterID:      e.ClusterID,
		CreatedBy:      e.CreatedBy,
		Compression:    e.Compression,
		SnapshotTime:   e.SnapshotTime,
		SQL:            e.SQL,
		Name:           e.DisplayName,
		S3RoleArn:      e.S3RoleArn,
		GCSURI:         e.GCSURI,
		AzBlobURI:      e.AzBlobURI,
		OSSURI:         e.OSSURI,
		OSSAccessKeyID: e.OSSAccessKeyID,
		S3Endpoint:     e.S3Endpoint,
	}
	export.ID = entity.GenerateIDWithPrefix(constant.ExportIDPrefix)
	export.State = constant.ExportStatePreparing
	export.CreatedBy = utils.BuildCreatorID(ctx)

	if e.TableFilter != nil {
		export.Where = e.TableFilter.Where
		export.Partitions = e.TableFilter.Partitions
		if len(e.TableFilter.Patterns) > 0 {
			tableFilters, err := json.Marshal(e.TableFilter.Patterns)
			if err != nil {
				return nil, bizerr.New(errorcode.ErrCodeInternalError, errorcode.InternalErrorMessage)
			}
			export.TableFilters = tableFilters
		}
	}
	if len(e.ColumnSelectors) > 0 {
		export.ColumnSelectors = e.ColumnSelectors
	}
	// filter backward compatibility
	if e.Database != "" && e.Table != "" {
		export.Filter = fmt.Sprintf("%s.%s", e.Database, e.Table)
	}

	if e.CSVFormat != nil {
		csvFormat, err := json.Marshal(e.CSVFormat)
		if err != nil {
			return nil, bizerr.New(errorcode.ErrCodeInternalError, errorcode.InternalErrorMessage)
		}
		export.CSVFormat = csvFormat
	}

	if e.ParquetFormat != nil {
		parquetFormat, err := json.Marshal(e.ParquetFormat)
		if err != nil {
			return nil, bizerr.New(errorcode.ErrCodeInternalError, errorcode.InternalErrorMessage)
		}
		export.ParquetFormat = parquetFormat
	}
	return export, nil
}

func convertToVO(e *entity.Export) (*vo.Export, bizerr.BizError) {
	exp := &vo.Export{
		ID:             e.ID,
		State:          e.State,
		FileType:       e.FileType,
		TargetType:     e.TargetType,
		S3URI:          e.S3BucketURI,
		S3AccessKeyID:  e.S3AccessKeyID,
		LocalURL:       e.LocalURL,
		CreatedAt:      e.CreatedAt,
		UpdatedAt:      e.UpdatedAt,
		ClusterID:      e.ClusterID,
		CreatedBy:      e.CreatedBy,
		Reason:         e.Reason,
		CloudProvider:  e.GetCloudProvider(),
		Region:         e.Region,
		TenantID:       e.TenantID,
		CompletedAt:    e.CompletedAt,
		Compression:    e.Compression,
		SnapshotTime:   e.SnapshotTime,
		ExpiredAt:      e.ExpiredAt,
		SQL:            e.SQL,
		S3RoleArn:      e.S3RoleArn,
		DisplayName:    e.Name,
		GCSURI:         e.GCSURI,
		AzBlobURI:      e.AzBlobURI,
		OSSURI:         e.OSSURI,
		OSSAccessKeyID: e.OSSAccessKeyID,
		S3Endpoint:     e.S3Endpoint,
	}

	var tableFilters []string
	if len(e.TableFilters) > 0 {
		tableFilters = make([]string, 0)
		err := json.Unmarshal(e.TableFilters, &tableFilters)
		if err != nil {
			return nil, bizerr.New(errorcode.ErrCodeInternalError, errorcode.InternalErrorMessage)
		}
	}
	if e.Where != "" || len(tableFilters) > 0 || len(e.Partitions) > 0 {
		exp.TableFilter = &vo.TableFilter{
			Where:      e.Where,
			Patterns:   tableFilters,
			Partitions: e.Partitions,
		}
	}
	if len(e.ColumnSelectors) > 0 {
		exp.ColumnSelectors = e.ColumnSelectors
	}

	// filter backward compatibility
	if e.Filter != "" {
		parts := strings.Split(e.Filter, ".")
		exp.Database = parts[0]
		exp.Table = parts[1]
	}
	// CSVFormat
	if e.CSVFormat != nil {
		err := json.Unmarshal(e.CSVFormat, &exp.CSVFormat)
		if err != nil {
			return nil, bizerr.New(errorcode.ErrCodeInternalError, errorcode.InternalErrorMessage)
		}
	}
	// ParquetOptions
	if e.ParquetFormat != nil {
		err := json.Unmarshal(e.ParquetFormat, &exp.ParquetFormat)
		if err != nil {
			return nil, bizerr.New(errorcode.ErrCodeInternalError, errorcode.InternalErrorMessage)
		}
	}

	exp.GenerateExportName()
	return exp, nil
}

func generateExportLabelsFromCluster(cluster *tidbmgmtservicev1.Cluster) map[string]string {
	labels := make(map[string]string)
	labels[constant.LabelKeyProviderType] = cluster.GetServerlessCluster().GetCluster().GetProviderType()
	labels[constant.LabelKeyKubernetes] = cluster.GetServerlessCluster().GetCluster().GetKubernetes()
	labels[constant.LabelKeyRegion] = cluster.Region
	return labels
}

func generateExportAnnotations(
	ctx context.Context,
	cluster *tidbmgmtservicev1.Cluster,
	export *entity.Export,
) map[string]string {
	logger := log.WithContextL(ctx)
	anno := make(map[string]string)
	exportMeta := &vo.ExportMeta{
		ID:            export.ID,
		ClusterID:     export.ClusterID,
		ProjectID:     export.ProjectID,
		TenantID:      export.TenantID,
		CreatedBy:     export.CreatedBy,
		KeyspaceToken: cluster.GetServerlessCluster().GetCluster().GetUserPrefix(),
		KeyspaceID:    cluster.GetServerlessCluster().GetCluster().GetKeyspaceId(),
	}
	if cluster.GetServerlessCluster().GetCluster().GetAnnotations() != nil {
		exportMeta.TiDBPoolSelector = cluster.GetServerlessCluster().GetCluster().GetAnnotations()[constant.AnnoTidbPoolSelector]
	}
	meta, err := json.Marshal(exportMeta)
	if err != nil {
		logger.Warn("marshal export meta failed", zap.Error(err))
		return anno
	}

	anno[constant.ExportMetaAnnotation] = string(meta)
	return anno
}

func (e *exportService) buildExportAdminPassword(ctx context.Context, clusterID uint64) (string, bizerr.BizError) {
	logger := log.WithContextL(ctx)
	// get cloud admin user
	passwd, err := e.accountFacade.GetCloudAdminPasswd(ctx, clusterID)
	if err != nil {
		logger.Error("get cluster confidential info failed", zap.Error(err))
		return "", bizerr.New(bizerr.ErrCodeUnknown, "get cluster confidential info failed")
	}
	if passwd == "" {
		logger.Error("admin password is empty")
		return "", bizerr.New(bizerr.ErrCodeUnknown, "admin password is empty")
	}

	cloudAdminPwd, err1 := e.encryptorFacade.DecryptBase64(passwd)
	if err1 != nil {
		logger.Error("decrypt admin password failed", zap.Error(err1))
		return "", bizerr.New(bizerr.ErrCodeUnknown, "decrypt admin password failed")
	}
	return cloudAdminPwd, nil
}

func BuildExportEvent(export *entity.Export,
	cluster *tidbmgmtservicev1.Cluster, status event.EventStatus, eventType string,
) (*event.DisplayEvent, bizerr.BizError) {
	tenantID, err := strconv.ParseUint(export.TenantID, 10, 64)
	if err != nil {
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "parse tenantID failed")
	}

	clusterID, err := strconv.ParseUint(export.ClusterID, 10, 64)
	if err != nil {
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "parse clusterID failed")
	}

	displayEvent := &event.DisplayEvent{
		HappenedAtMs:     uint64(time.Now().UnixMilli()),
		Severity:         event.EventSeverityInfo,
		TenantID:         tenantID,
		ProjectID:        cluster.GetProjectId(),
		ClusterID:        clusterID,
		Author:           export.CreatedBy,
		DisplayStatus:    status,
		DisplayEventType: eventType,
	}

	var summary string
	switch export.State {
	case constant.ExportStatePreparing, constant.ExportStateRunning:
		summary = fmt.Sprintf("Export job %s started to run.", export.ID)
	case constant.ExportStateFailed:
		summary = fmt.Sprintf("Export job %s failed.", export.ID)
	case constant.ExportStateSucceed:
		summary = fmt.Sprintf("Export job %s succeeded", export.ID)
	case constant.ExportStateCanceled:
		summary = fmt.Sprintf("Export job %s was canceled.", export.ID)
	case constant.ExportStateDeleted:
		summary = fmt.Sprintf("Export job %s was deleted.", export.ID)
	}
	displayEvent.Summary = summary
	exportDetail := event_constant.ExportDetail{
		ID:            export.ID,
		Filter:        export.Filter,
		TargetType:    string(export.TargetType),
		FileType:      string(export.FileType),
		S3BucketURI:   export.S3BucketURI,
		S3AccessKeyID: export.S3AccessKeyID,
	}
	detail, err := json.Marshal(exportDetail)
	if err != nil {
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "marshal detail failed")
	}
	displayEvent.Detail = string(detail)
	return displayEvent, nil
}

func (e *exportService) CleanExport(ctx context.Context, export *entity.Export) {
	logger := log.WithContextL(ctx).With(zap.String("exportID", export.ID))

	logger.Info("start to clean export")

	// Need not clean export before 3 days as stashing area will clean them automatically
	if export.CompletedAt != nil && time.Now().Add(-exportCleanTime).After(*export.CompletedAt) {
		return
	}
	stashingArea, err := e.stashingAreaProvider.GetStashingArea(export.GetCloudProvider(), export.Region)
	if err != nil {
		logger.Error("get stashing area failed", zap.Error(err))
		return
	}

	folderName := stashingArea.GetObjectPath(export.TenantID, stashing.Export, export.ID)
	err = stashingArea.DeleteFolder(ctx, folderName)
	if err != nil {
		logger.Error("delete folder failed", zap.String("folder", folderName), zap.Error(err))
		return
	}
	logger.Info("clean export success")
}

func (e *exportService) checkTarget(
	ctx context.Context,
	exportVo *vo.Export,
	cluster *tidbmgmtservicev1.Cluster,
) bizerr.BizError {
	cloudStorage := exportVo.ToCloudStorage(cluster.Region)
	if cloudStorage == nil {
		return bizerr.New(bizerr.ErrCodeUnknown, "cloud storage is nil")
	}

	// validate cloud storage
	if err := cloudStorage.Validate(cluster.CloudProvider); err != nil {
		return err
	}

	// check cloud storage permission
	switch exportVo.TargetType {
	case constant.ExportTargetTypeLocal:
		if slices.Contains(
			config.Global().BizConfig.SkipExportLocalStorageCheckTenants,
			cluster.GetOrganizationId(),
		) {
			return nil
		}
		return e.checkStashingArea(ctx, cluster)
	default:
		if err := e.mgmtSvc.CheckCloudStoragePermission(ctx, &globalserverlessv1.CheckCloudStoragePermissionRequest{
			ClusterId:    exportVo.ClusterID,
			CloudStorage: cloudStorage.ToRegionalCloudStorage(),
			FeatureName:  globalserverlessv1.CheckCloudStoragePermissionRequest_FEATURE_NAME_EXPORT,
		}); err != nil {
			return err
		}
	}

	// fill in necessary fields
	if exportVo.TargetType == constant.ExportTargetTypeOSS {
		options := []cloud_storage.OSSOption{cloud_storage.WithOSSAccessKey(
			exportVo.OSSAccessKeyID, exportVo.OSSAccessKeySecret)}
		region, err := cloud_storage.GetOssBucketRegion(ctx, exportVo.OSSURI, options...)
		if err != nil {
			return err
		}
		// add s3 endpoint because we use s3 sdk underlying. remove it when we support oss sdk in the future.
		exportVo.S3Endpoint = fmt.Sprintf("https://oss-%s.aliyuncs.com", region)
		// when alicloud cluster export to oss in the same region, use internal endpoint to reduce cost
		if cluster.CloudProvider == config.ProviderAliCloud && region == cluster.Region {
			exportVo.S3Endpoint = fmt.Sprintf("https://oss-%s-internal.aliyuncs.com", region)
		}
	}
	return nil
}

func (e *exportService) checkStashingArea(
	ctx context.Context,
	cluster *tidbmgmtservicev1.Cluster,
) (bizErr bizerr.BizError) {
	stashingArea, err := e.stashingAreaProvider.GetStashingArea(cluster.CloudProvider, cluster.Region)
	if err != nil {
		return bizerr.New(bizerr.ErrCodeUnknown, "get stashing area failed")
	}
	exceeded, err := stashingArea.CheckExceedSize(ctx, strconv.FormatUint(cluster.OrganizationId, 10))
	if err != nil {
		return bizerr.New(bizerr.ErrCodeUnknown, "failed to check storage limit")
	}
	if exceeded {
		return bizerr.New(errorcode.ErrCodeStorageLimitExceed, "")
	}
	return nil
}

func (e *exportService) buildExportCR(ctx context.Context, cluster *tidbmgmtservicev1.Cluster,
	export *entity.Export, exportVO *vo.Export,
) (_ *cloudv1.Export, bizErr bizerr.BizError) {
	user := cluster.GetServerlessCluster().GetCluster().GetUserPrefix() + ".cloud_admin"
	pswd, err := e.buildExportAdminPassword(ctx, cluster.ClusterId)
	if err != nil {
		return nil, err
	}

	exportCR := &cloudv1.Export{
		ObjectMeta: metav1.ObjectMeta{
			Labels:      generateExportLabelsFromCluster(cluster),
			Annotations: generateExportAnnotations(ctx, cluster, export),
			Namespace:   e.infraFacade.Namespace(cluster.GetOriginalProjectId()),
			Name:        export.ID,
		},
		Spec: cloudv1.ExportSpec{
			ClusterName: export.ClusterID,
			Target: cloudv1.ExportTarget{
				ExportType: cloudv1.ExportType(export.TargetType),
			},
			Options: cloudv1.ExportOptions{
				FileType: cloudv1.FileType(export.FileType),
			},
			Secret: cloudv1.ExportSecret{
				User:     user,
				Password: pswd,
			},
		},
	}
	// filter
	if exportVO.SQL != "" {
		exportCR.Spec.Options.SQL = export.SQL
	}
	if exportVO.TableFilter != nil {
		exportCR.Spec.Options.Where = exportVO.TableFilter.Where
		exportCR.Spec.Options.TableFilter = exportVO.TableFilter.Patterns
		if len(exportVO.TableFilter.Partitions) > 0 {
			exportCR.Spec.Options.Partition = exportVO.TableFilter.Partitions
		}
	}
	if len(exportVO.ColumnSelectors) > 0 {
		exportCR.Spec.Options.ColumnSelectors = convertColumnSelectorsToInfraAPI(exportVO.ColumnSelectors)
	}
	// target
	switch export.TargetType {
	case constant.ExportTargetTypeLocal:
		switch cluster.CloudProvider {
		case config.ProviderAliCloud:
			// reuse the S3 auth, will use RSSA in the future
			exportCR.Spec.Target.S3 = cloudv1.ExportS3{
				AccessKey:    config.Global().OssConfig.AccessKeyID,
				SecretAccess: config.Global().OssConfig.SecretAccessKey,
			}
		}
	case constant.ExportTargetTypeS3:
		exportCR.Spec.Target.S3 = cloudv1.ExportS3{
			BucketUri:    export.S3BucketURI,
			AccessKey:    export.S3AccessKeyID,
			SecretAccess: exportVO.S3SecretAccessKey,
			RoleArn:      export.S3RoleArn,
		}
	case constant.ExportTargetTypeGCS:
		exportCR.Spec.Target.GCS = cloudv1.ExportGCS{
			BucketUri:         export.GCSURI,
			ServiceAccountKey: exportVO.GCSServiceAccountKey,
		}
	case constant.ExportTargetTypeAzureBlob:
		accountName, containerURI, err := utils.ParseAzureURI(export.AzBlobURI)
		if err != nil {
			return nil, err
		}
		exportCR.Spec.Target.Azure = cloudv1.ExportAzure{
			ContainerName: containerURI.String(),
			SasToken:      exportVO.AzBlobSasToken,
			AccountName:   accountName,
		}
	case constant.ExportTargetTypeOSS:
		// use S3 target underlying
		exportCR.Spec.Target.ExportType = cloudv1.ExportTypeS3
		exportCR.Spec.Target.S3 = cloudv1.ExportS3{
			BucketUri:    strings.Replace(export.OSSURI, "oss://", "s3://", 1),
			AccessKey:    export.OSSAccessKeyID,
			SecretAccess: exportVO.OSSAccessKeySecret,
			Endpoint:     export.S3Endpoint,
			// https://help.aliyun.com/zh/oss/user-guide/regions-and-endpoints
			Provider: "alibaba",
		}
	}

	region, err := e.getTargetRegion(ctx, export, exportVO, cluster)
	if err != nil {
		return nil, err
	}
	exportCR.Spec.Target.Region = region
	// compression
	if export.Compression != constant.CompressionNone {
		exportCR.Spec.Options.Compression = cloudv1.CompressionType(export.Compression)
	}
	// snapshot
	if export.SnapshotTime != nil {
		// get PD TSO from UTC time
		pdTSO := export.SnapshotTime.UnixMilli() << 18
		exportCR.Spec.Options.Snapshot = strconv.FormatInt(pdTSO, 10)
	}
	// CSV
	if exportVO.CSVFormat != nil {
		exportCR.Spec.Options.CSVOptions = cloudv1.CSVOptions{
			Separator: exportVO.CSVFormat.Separator,
			Null:      exportVO.CSVFormat.NullValue,
			Delimiter: exportVO.CSVFormat.Delimiter,
			NoHeader:  exportVO.CSVFormat.SkipHeader,
			Dialect:   exportVO.CSVFormat.Dialect,
		}
	}
	// escape backslash (tri-state: nil keeps the dumpling default)
	exportCR.Spec.Options.EscapeBackslash = exportVO.EscapeBackslash
	// Parquet
	if exportVO.ParquetFormat != nil {
		exportCR.Spec.Options.ParquetOptions = cloudv1.ParquetOptions{
			Compression: cloudv1.ParquetCompressionType(exportVO.ParquetFormat.Compression),
		}
	}
	return exportCR, nil
}

func convertColumnSelectorsToInfraAPI(selectors []entity.ColumnSelector) []cloudv1.ColumnSelector {
	if len(selectors) == 0 {
		return nil
	}
	result := make([]cloudv1.ColumnSelector, 0, len(selectors))
	for _, selector := range selectors {
		result = append(result, cloudv1.ColumnSelector{
			Matcher: selector.Matcher,
			Columns: selector.Columns,
		})
	}
	return result
}

func (e *exportService) ParseTableFilter(
	ctx context.Context,
	clusterID string,
	patterns []string,
) ([]string, bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.String("clusterID", clusterID), zap.Any("patterns", patterns))

	// build table filter
	filter, err1 := filter.Parse(patterns)
	if err1 != nil {
		logger.Error("parse table filter failed", zap.Error(err1))
		trimErr := strings.TrimPrefix(err1.Error(), "at <cmdline>:1: ")
		errorMsg := fmt.Sprintf("%s Error: %s",
			errorcode.ExportErrorMessages[errorcode.ErrCodeInvalidTableFilter], trimErr)
		return nil, bizerr.New(errorcode.ErrCodeInvalidTableFilter, "%s", errorMsg)
	}

	// check cluster
	cluster, err := e.checkCluster(ctx, clusterID)
	if err != nil {
		return nil, err
	}
	// build cluster proxy
	user := cluster.GetServerlessCluster().GetCluster().GetUserPrefix() + ".cloud_admin"
	password, err := e.buildExportAdminPassword(ctx, cluster.ClusterId)
	if err != nil {
		logger.Error("build admin password failed", zap.Error(err))
		return nil, err
	}
	operator := &clusterproxy.ClusterOperator{
		Username:   user,
		AuthMethod: clusterproxy.AuthMethodPassword,
		Credential: password,
	}
	proxy, err := e.clusterProxyProviderFacade.GetClusterProxy(ctx, cluster, operator)
	if err != nil {
		logger.Error("get cluster proxy failed", zap.Error(err))
		return nil, bizerr.New(errorcode.ErrCodeSvcClusterProxy, "")
	}

	// query databases and tables
	dbTables, err := queryDbTables(ctx, proxy, 10000)
	if err != nil {
		return nil, err
	}
	// get tables
	results := filterDbTables(dbTables, filter)
	tables := make([]string, 0)
	for db, tables := range results {
		for _, table := range tables {
			tables = append(tables, fmt.Sprintf("%s.%s", db, table))
		}
	}
	logger.Info("parse table filter success", zap.Int("dbTables number", len(tables)))
	return tables, nil
}

type DatabaseResult struct {
	Databases [][]byte `json:"Database"`
	Tables    [][]byte `json:"Table"`
}

func queryDbTables(ctx context.Context, proxy facade.ClusterProxyFacade, limit int) (*DatabaseResult, bizerr.BizError) {
	logger := log.WithContextL(ctx)
	excludeDatabases := "('" + strings.Join(constant.ExcludeDatabases, "', '") + "')"
	sql := fmt.Sprintf(
		"select TABLE_SCHEMA as 'Database', TABLE_NAME as 'Table' from information_schema.tables where TABLE_SCHEMA not in %s",
		excludeDatabases,
	)
	if limit > 0 {
		sql += fmt.Sprintf(" limit %d", limit)
	}
	res, err := proxy.QuerySQL(ctx, sql)
	if err != nil {
		logger.Error("failed to query sql", zap.Error(err))
		return nil, err
	}
	result := DatabaseResult{}
	err1 := json.Unmarshal(res, &result)
	if err1 != nil {
		logger.Error("failed to unmarshal result", zap.Error(err1))
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "")
	}
	return &result, nil
}

func filterDbTables(dbTables *DatabaseResult, filter filter.Filter) map[string][]string {
	results := make(map[string][]string)
	dbMatch := make(map[string]bool)
	for i, table := range dbTables.Tables {
		db := string(dbTables.Databases[i])
		if _, ok := dbMatch[db]; !ok {
			dbMatch[db] = filter.MatchSchema(db)
		}
		if !dbMatch[db] {
			continue
		}
		if !filter.MatchTable(db, string(table)) {
			continue
		}
		results[db] = append(results[db], string(table))
	}
	return results
}

func (e *exportService) getTargetRegion(
	ctx context.Context,
	export *entity.Export,
	exportVO *vo.Export,
	cluster *tidbmgmtservicev1.Cluster,
) (string, bizerr.BizError) {
	switch export.TargetType {
	case constant.ExportTargetTypeS3:
		return e.getS3TargetRegion(ctx, exportVO)
	case constant.ExportTargetTypeGCS:
		return e.getGCSTargetRegion(ctx, exportVO)
	case constant.ExportTargetTypeAzureBlob:
		return e.getAzureTargetRegion(ctx, exportVO)
	case constant.ExportTargetTypeLocal:
		// For local storage, use cluster region
		return cluster.Region, nil
	case constant.ExportTargetTypeOSS:
		return e.getOSSTargetRegion(ctx, exportVO)
	}
	return "", nil
}

func (e *exportService) getS3TargetRegion(ctx context.Context, exportVO *vo.Export) (string, bizerr.BizError) {
	s3Uri := exportVO.S3URI
	p, err := url.Parse(cloud_storage.EscapePath(s3Uri))
	if err != nil {
		return "", bizerr.New(bizerr.ErrCodeParams, "Invalid S3 URL")
	}
	if p.Host == "" {
		return "", bizerr.New(bizerr.ErrCodeParams, "Invalid S3 URL: Bucket name is missing or improperly specified.")
	}
	bucketName := p.Host

	awsCfg := aws.Config{
		Region: "us-west-2",
	}
	s3Client := s3.NewFromConfig(awsCfg)
	region, err := manager.GetBucketRegion(ctx, s3Client, bucketName)
	if err != nil {
		var bnf manager.BucketNotFound
		if errors.As(err, &bnf) {
			return "", bizerr.New(errorcode.ErrCodeS3BucketNotFound, errorcode.ErrMsgBucketNotFound)
		}
		return "", bizerr.WrapError(err, "GetBucketRegion failed")
	}
	return region, nil
}

func (e *exportService) getOSSTargetRegion(ctx context.Context, exportVO *vo.Export) (string, bizerr.BizError) {
	options := []cloud_storage.OSSOption{cloud_storage.WithOSSAccessKey(
		exportVO.OSSAccessKeyID, exportVO.OSSAccessKeySecret)}
	region, err := cloud_storage.GetOssBucketRegion(ctx, exportVO.OSSURI, options...)
	if err != nil {
		return "", err
	}
	return region, nil
}

func (e *exportService) getGCSTargetRegion(ctx context.Context, exportVO *vo.Export) (string, bizerr.BizError) {
	return "", nil
}

func (e *exportService) getAzureTargetRegion(ctx context.Context, exportVO *vo.Export) (string, bizerr.BizError) {
	return "", nil
}