package import_svc

import (
	"context"
	"fmt"
	"net/url"
	"strconv"
	"strings"
	"time"

	"github.com/pingcap/errors"
	"github.com/tidbcloud/eventbus-datastruct/event"
	v1 "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/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/service"
	"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"
	"go.opentelemetry.io/otel"
	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/codes"
	"go.uber.org/multierr"
	"go.uber.org/zap"
	errors2 "k8s.io/apimachinery/pkg/api/errors"
)

const (
	InternalErrMsg           = "Encountered a problem when creating Import, please contact support."
	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."

	ossInternalEndpointSuffix = "-internal.aliyuncs.com"
	ossPublicEndpointSuffix   = ".aliyuncs.com"
)

var Tracer = otel.Tracer("import.syncer")

type ImportSyncerService struct {
	importRepo                 repo.ImportRepo
	accountFacade              facade.AccountFacade
	transProvider              db.TransProvider
	encryptorFacade            facade.EncryptorFacade
	infraFacade                facade.InfraFacade
	clusterProxyProviderFacade facade.ClusterProxyProviderFacade
	mgmtFacade                 facade.MgmtSvcFacade
	importHelperService        *ImportHelperService
	stashingAreaProvider       *stashing.StashingAreaProvider
	uploadRepo                 repo.UploadRepo
	featureGateFacade          facade.FeatureGate
	importSvc                  service.ImportService
	eventbusService            *eventbus.EventbusService
}

func NewImportSyncerService(
	importRepo repo.ImportRepo,
	accountFacade facade.AccountFacade,
	transProvider db.TransProvider,
	encryptorFacade facade.EncryptorFacade,
	infraFacade facade.InfraFacade,
	clusterProxyProviderFacade facade.ClusterProxyProviderFacade,
	mgmtFacade facade.MgmtSvcFacade,
	importHelperService *ImportHelperService,
	stashingAreaProvider *stashing.StashingAreaProvider,
	uploadRepo repo.UploadRepo,
	featureGateFacade facade.FeatureGate,
	importSvc service.ImportService,
	eventbusService *eventbus.EventbusService,
) *ImportSyncerService {
	return &ImportSyncerService{
		importRepo:                 importRepo,
		accountFacade:              accountFacade,
		transProvider:              transProvider,
		encryptorFacade:            encryptorFacade,
		infraFacade:                infraFacade,
		clusterProxyProviderFacade: clusterProxyProviderFacade,
		mgmtFacade:                 mgmtFacade,
		importHelperService:        importHelperService,
		stashingAreaProvider:       stashingAreaProvider,
		uploadRepo:                 uploadRepo,
		featureGateFacade:          featureGateFacade,
		importSvc:                  importSvc,
		eventbusService:            eventbusService,
	}
}

func (s *ImportSyncerService) Run(ctx context.Context) error {
	logger := utils.GetWorkerLoggerFromContext(ctx)
	logger.Info("start to sync import")
	imports, err := s.importHelperService.GetImportsToSync(ctx, nil)
	if err != nil {
		logger.Error("failed to fetch imports to sync", zap.Error(err))
	}
	if len(imports) == 0 {
		logger.Info("no import to sync, skip")
		return nil
	}

	startTime := time.Now()
	logger.Info("[global]sync import with infra, start.", zap.Int("importCount", len(imports)))
	var errs error
	for _, imp := range imports {
		err = s.Sync(ctx, imp)
		if err != nil {
			errs = multierr.Append(errs, err)
		}
	}
	logger.Info(
		"[global]sync import with infra, done.",
		zap.Any("costSeconds", time.Since(startTime).Seconds()),
		zap.Error(errs),
	)
	return errs
}

func (s *ImportSyncerService) Sync(ctx context.Context, imp *entity.ImportTask) (err error) {
	ctx, span := Tracer.Start(ctx, "ImportSyncerService.sync")
	defer func() { tracing.EndSpan(span, err) }()
	span.SetAttributes(attribute.String("cluster_id", imp.ClusterID),
		attribute.String("tenant_id", fmt.Sprintf("%v", imp.Cluster.OrganizationId)),
		attribute.String("import_id", imp.ID),
		attribute.String("status", string(imp.Status)),
	)
	// if the cluster is deleted, just failed the import and let it clean up resource(the machine for lighting).
	if imp.Cluster.GetServerlessCluster().
		GetUserStatus() ==
		globalserverlessv1.UserClusterStatus_USER_CLUSTER_STATUS_DELETED {
		if imp.Status != constant.ImportStatusFailed {
			return s.transToStatus(ctx, imp, constant.ImportStatusFailed, "cluster is deleted")
		}
	}

	logger := utils.GetWorkerLoggerFromContext(ctx)
	logger = logger.With(zap.String("import_id", imp.ID))
	logger.Info("try to sync import", zap.String("status", string(imp.Status)))

	switch imp.Status {
	case constant.ImportStatusPreparing:
		err = s.OnPreparing(ctx, imp)
	case constant.ImportStatusImporting:
		err = s.OnImporting(ctx, imp)
	case constant.ImportStatusCanceling:
		err = s.OnCanceling(ctx, imp)
	case constant.ImportStatusCanceled, constant.ImportStatusFailed, constant.ImportStatusCompleted:
		err = s.cleanUp(ctx, imp)
	default:
		return errors.Errorf("unknown status: %v", imp.Status)
	}

	if err != nil {
		logger.Error("failed to sync the import", zap.Error(err), zap.Any("import", imp))
		return err
	}

	return err
}

func (s *ImportSyncerService) OnPreparing(ctx context.Context, imp *entity.ImportTask) (err error) {
	ctx, span := Tracer.Start(ctx, "ImportSyncerService.OnPreparing")
	logger := utils.GetWorkerLoggerFromContext(ctx)
	logger = logger.With(zap.String("import_id", imp.ID))
	defer func() { tracing.EndSpan(span, err) }()
	span.SetAttributes(attribute.String("import_id", imp.ID))

	// NeedDropTables is only used in tutorial(https://jira.tidbcloud.com/browse/DM-9789)
	if imp.NeedDropTables && len(imp.TargetTableInfos.Infos) > 0 {
		if err := s.dropTables(ctx, imp); err != nil {
			logger.Error("truncate tables error, skip this error", zap.Error(err))
			return err
		}

		imp.TargetTableInfos.Infos = nil
		if err = s.importRepo.Update(ctx, imp.Import, "target_table_infos"); err != nil {
			logger.Error("failed to update table fields", zap.Error(err))
			return err
		}
	}

	var sourceUrl *url.URL
	var bizErr bizerr.BizError
	switch imp.DataSourceType {
	case constant.ImportDataSourceAzBlob:
		sourceUrl, bizErr = s.generateAzureURL(imp, logger)
		if bizErr != nil {
			return s.transToStatus(ctx, imp, constant.ImportStatusFailed, importFailureMessage(ctx, bizErr))
		}
	case constant.ImportDataSourceS3, constant.ImportDataSourceLocal, constant.ImportDataSourceS3Compatible:
		sourceUrl, bizErr = s.generateS3URL(ctx, imp, logger)
		if bizErr != nil {
			return s.transToStatus(ctx, imp, constant.ImportStatusFailed, importFailureMessage(ctx, bizErr))
		}
	case constant.ImportDataSourceOSS:
		imp.SourceURL = strings.Replace(imp.SourceURL, "oss://", "s3://", 1)
		sourceUrl, bizErr = s.generateS3URL(ctx, imp, logger)
		if bizErr != nil {
			return s.transToStatus(ctx, imp, constant.ImportStatusFailed, importFailureMessage(ctx, bizErr))
		}
	case constant.ImportDataSourceGCS:
		sourceUrl, bizErr = s.generateGCSURL(imp, logger)
		if bizErr != nil {
			return s.transToStatus(ctx, imp, constant.ImportStatusFailed, importFailureMessage(ctx, bizErr))
		}
	}
	if sourceUrl == nil {
		return s.transToStatus(ctx, imp, constant.ImportStatusFailed, "invalid source url")
	}
	imp.SourceURL = sourceUrl.String()

	clusterID, err := strconv.ParseUint(imp.ClusterID, 10, 64)
	if err != nil {
		return err
	}
	cluster, err := s.mgmtFacade.GetCluster(ctx, clusterID, tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC)
	if err != nil {
		logger.Error("failed to get cluster by ID", zap.Error(err))
		return err
	}
	imp.Cluster = cluster
	passwd, bizErr := s.accountFacade.GetCloudAdminPasswd(ctx, clusterID)
	if bizErr != nil {
		return bizErr
	}
	password, err := s.encryptorFacade.DecryptBase64(passwd)
	if err != nil {
		return err
	}

	isSuperVip := s.featureGateFacade.FeatureEnabled(
		constant.FeatureImportSuperVIP,
		imp.Cluster.OrganizationId,
		imp.Cluster.Region,
	)
	impCR, err := s.importSvc.NewInfraAPIImportObject(ctx, imp, password, isSuperVip)
	if err != nil {
		logger.Error("Failed to generate infra api import object", zap.Error(err))
		return err
	}

	impCR.Spec.StorageRequest = service.ImportStorageRequest
	proxyEndpoint := imp.Cluster.GetServerlessCluster().GetCluster().GetProxyEndpoint()
	if err := s.infraFacade.CreateImportIgnoreExists(ctx, proxyEndpoint, impCR); err != nil {
		logger.Error("Failed to create infra api import object", zap.Error(err))
		return err
	}
	return s.updateInfo(ctx, imp)
}

func importFailureMessage(ctx context.Context, err error) string {
	return errorcode.PrepareErrorMessage(ctx, err, nil).ErrMsg()
}

func (s *ImportSyncerService) generateAzureURL(imp *entity.ImportTask, logger *zap.Logger) (*url.URL, bizerr.BizError) {
	account, sourceUrl, bizErr := utils.ParseAzureURI(imp.SourceURL)
	if bizErr != nil {
		return nil, bizErr
	}
	query := sourceUrl.Query()
	query.Set("account-name", account)

	if imp.AzureSASToken != "" {
		azureSASToken, err := s.encryptorFacade.DecryptBase64(imp.AzureSASToken)
		if err != nil {
			logger.Error("failed to decrypt AzureSASToken", zap.Error(err), zap.String("AzureSASToken", azureSASToken))
			return nil, bizerr.New(errorcode.ErrCodeInternalError, errorcode.InternalErrorMessage)
		}
		query.Set("sas-token", azureSASToken)
	}

	sourceUrl.RawQuery = query.Encode()
	return sourceUrl, nil
}

func (s *ImportSyncerService) generateS3URL(
	ctx context.Context,
	imp *entity.ImportTask,
	logger *zap.Logger,
) (*url.URL, bizerr.BizError) {
	sourceUrl, err := url.Parse(cloud_storage.EscapePath(imp.SourceURL))
	if err != nil {
		logger.Error("failed to parse source url", zap.Error(err), zap.String("source url", imp.SourceURL))
		return nil, bizerr.New(errorcode.ErrCodeInternalError, InternalErrMsg)
	}
	q := sourceUrl.Query()

	if imp.S3SecretAccessKey != "" && imp.S3AccessKeyID != "" {
		s3SecretAccessKey, err := s.encryptorFacade.DecryptBase64(imp.S3SecretAccessKey)
		if err != nil {
			logger.Error(
				"failed to decrypt AwsSecretAccessKey",
				zap.Error(err),
				zap.String("AwsSecretAccessKey", s3SecretAccessKey),
			)
			return nil, bizerr.New(errorcode.ErrCodeInternalError, InternalErrMsg)
		}
		q.Set("secret-access-key", s3SecretAccessKey)
		q.Set("access-key", imp.S3AccessKeyID)
	} else if imp.S3RoleArn != "" {
		q.Set("role-arn", imp.S3RoleArn)
		profile, bizErr := s.importSvc.GetDataIngestionProfile(ctx, imp.Cluster)
		if bizErr != nil {
			logger.Error("failed to get data ingestion profile", zap.Error(bizErr))
			return nil, bizErr
		}
		q.Set("external-id", profile.ExternalId)
	}

	if imp.BucketRegion != "" {
		q.Set("region", imp.BucketRegion)
	}

	if imp.S3Endpoint != "" {
		q.Set("endpoint", imp.S3Endpoint)
		// for s3 compatible storage, set provider to skip get region from bucket, see https://github.com/tidbcloud/tidb-cse/blob/release-7.5-keyspace/br/pkg/storage/s3.go#L398
		q.Set("provider", "aws-compatible")
		// for aliyun oss and netease NOS, set provider to set ForcePathStyle to false, see https://github.com/tidbcloud/tidb-cse/blob/release-7.5-keyspace/br/pkg/storage/s3.go#L172
		if strings.HasSuffix(imp.S3Endpoint, "aliyuncs.com") {
			// https://help.aliyun.com/zh/oss/user-guide/regions-and-endpoints
			q.Set("provider", "alibaba")
			if imp.BucketRegion == imp.Cluster.Region && imp.Cluster.CloudProvider == config.ProviderAliCloud &&
				!strings.HasSuffix(imp.S3Endpoint, ossInternalEndpointSuffix) {
				// use internal endpoint to reduce cost
				q.Set(
					"endpoint",
					strings.Replace(imp.S3Endpoint, ossPublicEndpointSuffix, ossInternalEndpointSuffix, 1),
				)
			}
		}
		if strings.HasSuffix(imp.S3Endpoint, "126.net") {
			// https://sf.163.com/help/documents/67078583131230208
			q.Set("provider", "netease")
		}
	}

	sourceUrl.RawQuery = q.Encode()
	return sourceUrl, nil
}

func (s *ImportSyncerService) generateGCSURL(imp *entity.ImportTask, logger *zap.Logger) (*url.URL, bizerr.BizError) {
	sourceUrl, err := url.Parse(cloud_storage.EscapePath(imp.SourceURL))
	if err != nil {
		logger.Error("failed to parse source url", zap.Error(err), zap.String("source url", imp.SourceURL))
		return nil, bizerr.New(errorcode.ErrCodeInternalError, InternalErrMsg)
	}
	q := sourceUrl.Query()

	if imp.GCSCredentials != "" {
		gcsCredentials, err := s.encryptorFacade.DecryptBase64(imp.GCSCredentials)
		if err != nil {
			logger.Error(
				"failed to decrypt GCSCredentials",
				zap.Error(err),
				zap.String("GCSCredentials", gcsCredentials),
			)
			return nil, bizerr.New(errorcode.ErrCodeInternalError, InternalErrMsg)
		}
		q.Set("credentials", gcsCredentials)
	}

	sourceUrl.RawQuery = q.Encode()
	return sourceUrl, nil
}

func (s *ImportSyncerService) OnImporting(ctx context.Context, imp *entity.ImportTask) error {
	return s.updateInfo(ctx, imp)
}

func (s *ImportSyncerService) OnCanceling(ctx context.Context, imp *entity.ImportTask) (err error) {
	defer func() {
		// need transfer to `canceled` whatever happens
		err = s.transToStatus(ctx, imp, constant.ImportStatusCanceled, "")
	}()
	ctx, span := Tracer.Start(ctx, "ImportSyncerService.OnCanceling")
	logger := utils.GetWorkerLoggerFromContext(ctx)
	logger = logger.With(zap.String("import_id", imp.ID))
	defer func() { tracing.EndSpan(span, err) }()

	proxyEndpoint := imp.Cluster.GetServerlessCluster().GetCluster().GetProxyEndpoint()
	impcr, err := s.infraFacade.GetImport(ctx, proxyEndpoint, imp.Cluster.GetOriginalProjectId(), imp.ID)
	if err != nil {
		if errors2.IsNotFound(err) {
			err = nil
			return err
		}
		logger.Error("failed to get import object", zap.Error(err), zap.Reflect("import", imp))
		return err
	}

	if err = s.infraFacade.DeleteImport(ctx, proxyEndpoint, imp.Cluster.GetOriginalProjectId(), imp.ID); err != nil {
		logger.Error("failed to delete import object", zap.Error(err), zap.Reflect("import", imp))
		return err
	}

	if err = s.transProvider.Transaction(ctx, func(ctx context.Context) error {
		return s.saveCompletedTables(ctx, imp, &(impcr.Status))
	}); err != nil {
		logger.Error("failed to save completed tables", zap.Error(err), zap.Reflect("import", imp))
		return err
	}

	return err
}

func (s *ImportSyncerService) updateInfo(ctx context.Context, imp *entity.ImportTask) error {
	var errs error
	_, span := Tracer.Start(ctx, "ImportSyncerService.updateFromLightningV1")
	err := s.updateFromLightningV1(ctx, imp)
	if err != nil {
		errs = multierr.Append(errs, err)
		span.SetStatus(codes.Error, errs.Error())
		span.RecordError(errs)
	}
	span.End()

	return errs
}

func (s *ImportSyncerService) updateFromLightningV1(ctx context.Context, imp *entity.ImportTask) error {
	logger := utils.GetWorkerLoggerFromContext(ctx)
	logger = logger.With(zap.String("import_id", imp.ID))
	proxyEndpoint := imp.Cluster.GetServerlessCluster().GetCluster().GetProxyEndpoint()
	impcr, err := s.infraFacade.GetImport(ctx, proxyEndpoint, imp.Cluster.GetOriginalProjectId(), imp.ID)
	if err != nil {
		return err
	}
	imp.TotalSize = &impcr.Status.TotalSize
	imp.TotalSizeWithIndex = &impcr.Status.TotalSizeWithIndex
	imp.ProcessedDataSize = &impcr.Status.ProcessedDataSize
	imp.TotalFiles = impcr.Status.TotalFiles
	imp.CurrentTables.Tables = nil
	for i := range impcr.Status.CurrentTables {
		v := impcr.Status.CurrentTables[i]
		imp.CurrentTables.Tables = append(imp.CurrentTables.Tables, &entity.CurrentTable{
			Name:             v.Name,
			Size:             v.Size,
			CompletedPercent: v.CompletedPercent,
		})
	}
	imp.CompletedTables = impcr.Status.CompletedTables
	imp.PendingTables = impcr.Status.PendingTables
	imp.TotalTablesCount = impcr.Status.TotalTables
	imp.CompletedPercent = impcr.Status.CompletedPercent
	imp.PostImportCompletedPercent = impcr.Status.PostImportCompletedPercent
	imp.TaskID = impcr.Status.TaskId
	fields := []string{
		"total_size",
		"processed_data_size",
		"total_files",
		"current_tables",
		"total_tables_count",
		"completed_tables",
		"pending_tables",
		"completed_percent",
		"post_import_completed_percent",
		"task_id",
		"total_size_with_index",
	}

	err = s.importRepo.Update(ctx, imp.Import, fields...)
	if err != nil {
		logger.Error("failed to update table fields", zap.Error(err))
		return err
	}

	err = s.transProvider.Transaction(ctx, func(ctx context.Context) error {
		if err := s.saveCompletedTables(ctx, imp, &(impcr.Status)); err != nil {
			return errors.Annotate(err, "save completed tables error")
		}
		return nil
	})
	if err != nil {
		logger.Error("failed to save table information", zap.Error(err))
	}

	if imp.Status != constant.ImportStatusCanceled && imp.Status != constant.ImportStatusCanceling {
		if err = s.transToStatus(ctx, imp, convertStatus(string(impcr.Status.Phase)),
			impcr.Status.Message); err != nil {
			logger.Error("failed to transfer status", zap.Error(err))
		}
	}

	if imp.Status == constant.ImportStatusCompleted {
		usedTime := time.Since(imp.CreatedAt).Seconds()
		totalSize := float64(*imp.TotalSize)
		metric.ImportThroughputBytes.WithLabelValues(entity.ImportProviderType(imp.Cluster)).
			Observe(totalSize / usedTime)
	}
	return nil
}

func (s *ImportSyncerService) saveCompletedTables(
	ctx context.Context,
	imp *entity.ImportTask,
	importStatus *v1.ImportStatus,
) error {
	logger := utils.GetWorkerLoggerFromContext(ctx)
	logger = logger.With(zap.String("import_id", imp.ID))
	tableNames := make([]string, 0, len(importStatus.AllTableDetails))
	for i := range importStatus.AllTableDetails {
		tableNames = append(tableNames, importStatus.AllTableDetails[i].Name)
	}
	warnings, err := s.importHelperService.CheckWarningsByTables(ctx, imp, tableNames)
	if err != nil {
		logger.Error("batch check warnings failed", zap.Error(err))
		return err
	}
	imp.TableResults = entity.ImportTables{Tables: make([]entity.ImportTable, 0, len(importStatus.AllTableDetails))}
	for i := range importStatus.AllTableDetails {
		tblDetail := importStatus.AllTableDetails[i]
		importTableResult := convertToTableResult(tblDetail.Phase)
		if len(importTableResult) == 0 {
			continue
		}
		importTblEntity := entity.ImportTable{
			TableName: tblDetail.Name,
			Result:    importTableResult,
			Message:   tblDetail.Message,
		}
		tableWarningMsg := warnings[tblDetail.Name]
		if tableWarningMsg != "" {
			importTblEntity.Message = tableWarningMsg
			importTblEntity.Result = constant.ImportTableResultWarning
		}
		imp.TableResults.Tables = append(imp.TableResults.Tables, importTblEntity)
	}
	// update import tables
	if err := s.importRepo.Update(ctx, imp.Import, "table_results"); err != nil {
		logger.Error("failed to update import tables", zap.Error(err))
		return err
	}
	return nil
}

func (s *ImportSyncerService) dropTables(ctx context.Context, imp *entity.ImportTask) error {
	clusterID, err := strconv.ParseUint(imp.ClusterID, 10, 64)
	if err != nil {
		return err
	}
	passwd, bizErr := s.accountFacade.GetCloudAdminPasswd(ctx, clusterID)
	if bizErr != nil {
		return bizErr
	}
	password, err := s.encryptorFacade.DecryptBase64(passwd)
	if err != nil {
		return err
	}

	operator := &clusterproxy.ClusterOperator{
		Username:   constant.CloudAdminUsername,
		AuthMethod: clusterproxy.AuthMethodPassword,
		Credential: password,
	}
	proxy, err := s.clusterProxyProviderFacade.GetClusterProxy(ctx, imp.Cluster, operator)
	if err != nil {
		return bizerr.New(errorcode.ErrCodeSvcClusterProxy, "failed to get cluster proxy")
	}

	var tables []string
	for _, t := range imp.TargetTableInfos.Infos {
		tables = append(tables, service.GenTableID(t.Schema, t.Table))
	}

	err = proxy.ExecuteSQL(ctx, "DROP TABLE IF EXISTS "+strings.Join(tables, ", "))
	if err != nil {
		return err
	}

	return err
}

func (s *ImportSyncerService) transToStatus(
	ctx context.Context,
	imp *entity.ImportTask,
	status constant.ImportStatus,
	msg string,
) (err error) {
	_, span := Tracer.Start(ctx, "ImportSyncerService.transToStatus")
	defer func() { tracing.EndSpan(span, err) }()
	span.SetAttributes(
		attribute.String("import_id", imp.ID),
		attribute.String("cluster_id", imp.ClusterID),
		attribute.String("status", string(status)))

	if imp.Status == status {
		return nil
	}

	imp.Status = status
	var fields []string
	fields = append(fields, "status")

	if status == constant.ImportStatusImporting && imp.StartedAt == nil {
		t := time.Now()
		imp.StartedAt = &t
		fields = append(fields, "started_at")
	}

	if entity.IsEndStatus(status) && imp.CompletedAt == nil {
		t := time.Now()
		imp.CompletedAt = &t
		fields = append(fields, "completed_at")
	}

	if msg != "" {
		imp.Message = msg
		fields = append(fields, "message")
	}

	bizErr := s.importRepo.Update(ctx, imp.Import, fields...)
	if bizErr != nil {
		return errors.AddStack(bizErr)
	}

	if entity.IsEndStatus(status) {
		if status != constant.ImportStatusCanceled {
			// publish create event
			eventStatus := event.EventStatusSucceeded
			if imp.Status == constant.ImportStatusFailed {
				eventStatus = event.EventStatusFailed
			}
			go s.importSvc.PublishImportEvent(ctx, imp, eventStatus, event_constant.EventTypeCreateImport)
		}
		return s.cleanUp(ctx, imp)
	}

	return nil
}

// delete worker created and resources created at k8s.
func (s *ImportSyncerService) cleanUp(ctx context.Context, imp *entity.ImportTask) (err error) {
	ctx, span := Tracer.Start(ctx, "ImportSyncerService.cleanUp")
	defer func() { tracing.EndSpan(span, err) }()

	logger := utils.GetWorkerLoggerFromContext(ctx)
	logger = logger.With(zap.String("import_id", imp.ID))

	// Currently we only fetch the un-cleaned imports to sync.
	// And we just cleaned up the completed imports.
	if !entity.IsEndStatus(imp.Status) {
		return errors.Errorf("can't clean up at status: %s", imp.Status)
	}

	uint64ClusterId, err := strconv.ParseUint(imp.ClusterID, 10, 64)
	if err != nil {
		logger.Error("failed to parse cluster id", zap.Error(err))
		return bizerr.New(bizerr.ErrCodeParams, "invalid cluster id")
	}
	if imp.ClusterStatusID != 0 {
		if err = s.mgmtFacade.DeleteClusterCustomStatus(ctx, uint64ClusterId, imp.ClusterStatusID); err != nil {
			logger.Error("failed to delete import cluster custom status",
				zap.Uint64("cluster_status_id", imp.ClusterStatusID),
				zap.Error(err))
			return err
		}
	}

	proxyEndpoint := imp.Cluster.GetServerlessCluster().GetCluster().GetProxyEndpoint()
	if err = s.infraFacade.DeleteImport(ctx, proxyEndpoint, imp.Cluster.GetOriginalProjectId(), imp.ID); err != nil {
		return errors.AddStack(err)
	}

	imp.CleanUp = true
	fields := []string{
		"clean_up",
	}
	if imp.S3SecretAccessKey != "" {
		imp.S3SecretAccessKey = ""
		fields = append(fields, "s3_secret_access_key")
	} else if imp.GCSCredentials != "" {
		imp.GCSCredentials = ""
		fields = append(fields, "gcs_credentials")
	} else if imp.AzureSASToken != "" {
		imp.AzureSASToken = ""
		fields = append(fields, "azure_sas_token")
	}
	if bizErr := s.importRepo.Update(ctx, imp.Import, fields...); bizErr != nil {
		return bizErr
	}

	if imp.DataSourceType == constant.ImportDataSourceLocal {
		upload, bizErr := s.uploadRepo.GetUploadByID(ctx, imp.LocalImportCfg.UploadID)
		if bizErr != nil {
			return bizErr
		}
		stashingArea, err := s.stashingAreaProvider.GetStashingArea(imp.Cluster.CloudProvider, imp.BucketRegion)
		if err != nil {
			return err
		}
		if upload == nil {
			logger.Warn("upload not found", zap.String("upload_id", imp.LocalImportCfg.UploadID))
			return nil
		}
		// Temporary postpone the deletion of the object in stashing area for failed import for debugging.
		if imp.Status != constant.ImportStatusFailed {
			_, err := stashingArea.DeleteObject(ctx, upload.BucketKey)
			if err != nil {
				// The object will expire in 3 days, so we can ignore the error here.
				logger.Warn("failed to delete object from stashing area", zap.Error(err))
			}
			bizErr := s.uploadRepo.UpdateStatus(ctx, upload.ID, constant.UploadStatusCleaned)
			if bizErr != nil {
				logger.Warn("failed to update upload status", zap.Error(bizErr))
			}
		}
	}

	return nil
}