package pimport

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

	"github.com/google/uuid"
	"github.com/pingcap/errcode"
	"github.com/pingcap/errors"
	plog "github.com/pingcap/log"
	"github.com/pingcap/tidb/pkg/errno"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	"github.com/tidbcloud/dataflow-service/app/container"
	"github.com/tidbcloud/dataflow-service/app/models/config_manager"
	entity2 "github.com/tidbcloud/dataflow-service/app/repo/entity"
	agentpb "github.com/tidbcloud/dataflow-service/idl/pbgen/proto/agent"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	"github.com/tidbcloud/dataflow-service/internal/bizutils"
	"github.com/tidbcloud/dataflow-service/internal/cluster_job"
	"github.com/tidbcloud/dataflow-service/internal/common"
	"github.com/tidbcloud/dataflow-service/internal/config"
	"github.com/tidbcloud/dataflow-service/internal/constants"
	"github.com/tidbcloud/dataflow-service/internal/db"
	"github.com/tidbcloud/dataflow-service/internal/db/dbutil"
	"github.com/tidbcloud/dataflow-service/internal/db/entity"
	eventpublisher "github.com/tidbcloud/dataflow-service/internal/event-bus/publisher"
	"github.com/tidbcloud/dataflow-service/internal/grpc/grpcli"
	"github.com/tidbcloud/dataflow-service/internal/metrics"
	"github.com/tidbcloud/dataflow-service/internal/model/feature"
	"github.com/tidbcloud/dataflow-service/internal/model/util"
	"github.com/tidbcloud/dataflow-service/internal/utils"
	"github.com/tidbcloud/dataflow-service/pkg/account/accountdb"
	"github.com/tidbcloud/dataflow-service/pkg/azblob"
	"github.com/tidbcloud/dataflow-service/pkg/event"
	"github.com/tidbcloud/dataflow-service/pkg/gcs"
	pkggrpcli "github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli"
	importcfg "github.com/tidbcloud/dataflow-service/pkg/import-config"
	infraapi "github.com/tidbcloud/dataflow-service/pkg/infra-api"
	"github.com/tidbcloud/dataflow-service/pkg/s3"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	infra_config "github.com/tidbcloud/infra-api/pkg/pingcap/util/config"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/log"
	"github.com/tidbcloud/pkgv2/logid"
	"github.com/tidbcloud/pkgv2/secret/encrypt"
	"github.com/tidbcloud/pkgv2/tracing"
	"go.opentelemetry.io/otel"
	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/codes"
	"go.uber.org/multierr"
	"go.uber.org/zap"
	"google.golang.org/genproto/googleapis/rpc/errdetails"
	"gorm.io/gorm"
)

const (
	// error messages
	ConnectorErr        = "Import can not be supported, because this Cluster has enabled the following connector(s): %s"
	ReadonlyForbidenErr = "Can not import because your cluster is read-only."
	InternalErr         = "Internal error, please feel free to contact support."

	importImageAnnotation = "import.tidbcloud.com/image"
	importLocalLabel      = "import.tidbcloud.com/local"

	maxSortBufferSize     = 20971520 // 20M
	defaultSortBufferSize = 1048576  // 1M

	jobDeletionAnnotation = "import.tidbcloud.com/scheduled-deletion-time"
)

var Tracer = otel.Tracer("central/model/import")

// Model is created with NewModel
type Model struct {
	log       *zap.Logger
	encryptor encrypt.Encryptor
	publisher eventpublisher.Publisher
	// aliveLocalImports is a map of import ID to import status
	// it is used to keep track of local imports that are still running
	// so that we can detect the worker's failover when load data is running
	aliveLocalImports *sync.Map
	configCenter      config_manager.ConfigCenterManager
}

// NewModel returns a Model
func NewModel(log *zap.Logger, encryptor encrypt.Encryptor,
	publisher eventpublisher.Publisher, configCenter config_manager.ConfigCenterManager,
) Model {
	return Model{
		log:               log,
		encryptor:         encryptor,
		publisher:         publisher,
		aliveLocalImports: &sync.Map{},
		configCenter:      configCenter,
	}
}

func (c Model) DecryptBase64(password string) (string, error) {
	if len(password) == 0 {
		return "", nil
	}
	secret, err := encrypt.DecryptBase64(c.encryptor, password)
	return string(secret), err
}

func (c Model) EncryptBase64(password string) (string, error) {
	if len(password) == 0 {
		return "", nil
	}
	return encrypt.EncryptBase64(c.encryptor, encrypt.Secret(password))
}

// UserContext attaches a User to a Context.
// This helps audit the User
type UserContext struct {
	Model
	tenantAndUser *accountdb.User
	log           *zap.Logger
}

// User returns a UserContext
func (m Model) User(ctx context.Context, user *accountdb.User) UserContext {
	return UserContext{m, user, m.log.With(
		zap.Uint64("tenantID", user.TenantID),
		zap.Uint64("userID", user.ID),
		zap.String("module", "import"),
	).With(log.FieldsFromContext(ctx)...)}
}

func (m Model) TenantLog(tenantID uint64) *zap.Logger {
	return m.log.With(zap.Uint64("tenantID", tenantID), zap.String("model", "import"))
}

func (c UserContext) UserID() uint64 {
	return c.tenantAndUser.ID
}

func (c UserContext) UserEmail() string {
	return c.tenantAndUser.Email
}

func (c UserContext) CheckOrgIDAndProjectID(orgID, projectID entity.ID) error {
	user := c.tenantAndUser
	if user.IsImpersonated {
		return nil
	}

	if orgID.Uint64() != user.TenantID {
		return errcode.NewForbiddenErr(errors.New("org_id parameter is inconsistent with the user's org ID"))
	}

	return dbutil.CheckProjectID(user.Projects, entity2.ID(projectID))
}

// CheckCluster checks whether user has privileges to this cluster and the cluster is in usable status
func (c UserContext) CheckIdAndCluster(ctx context.Context, orgID, projectID, clusterID uint64) error {
	logger := log.WithContextS(ctx)
	if err := c.CheckOrgIDAndProjectID(entity.ID(orgID), entity.ID(projectID)); err != nil {
		logger.Info("check org ID and project ID failed", zap.Error(err))
		if err := bizutils.AppendErrorDetails(ctx, &errdetails.ErrorInfo{
			Reason: err.Error(),
		}); err != nil {
			logger.Warn("append error detail error", zap.Error(err))
		}
		return err
	}

	return bizutils.CheckCluster(ctx, orgID, projectID, clusterID)
}

// CheckClusterOwner checks whether the current user has privileges to this cluster
func (c UserContext) CheckClusterOwner(ctx context.Context, orgID, projectID, clusterID entity.ID) error {
	if err := c.CheckOrgIDAndProjectID(orgID, projectID); err != nil {
		return err
	}
	return bizutils.CheckClusterOwner(ctx, uint64(orgID), uint64(projectID), uint64(clusterID))
}

func (c UserContext) Log() *zap.Logger {
	return c.log
}

// GetImportDefaultStorageRequestConfig gets the default storage request config string for an import Pod.
func GetImportDefaultStorageRequestConfig() string {
	return config.Global().ImportConfig.StorageRequest
}

type CancelImportInvalidStatusError struct {
	Status entity.ImportStatus
}

func (e *CancelImportInvalidStatusError) Error() string {
	return fmt.Sprintf("The cluster cannot be canceled because it is in the %s state", e.Status)
}

// CancelImport cancel the import
// we update the status to be canceling and wait it reach canceled if waitCanceled is true.
func (c UserContext) CancelImport(ctx context.Context, projectID entity.ID, clusterID entity.ID, importID entity.ID, waitCanceled bool) (err error) {
	orgID := entity.ID(c.tenantAndUser.TenantID)
	if err := c.CheckClusterOwner(ctx, orgID, projectID, clusterID); err != nil {
		return err
	}
	imp, err := GetImportTask(ctx, clusterID, importID, false)
	if err != nil {
		return err
	}
	oldStatus := imp.Status
	switch imp.Status {
	case entity.ImportStatusPreparing,
		entity.ImportStatusImporting,
		entity.ImportStatusCanceling:
	case entity.ImportStatusFailed,
		entity.ImportStatusCompleted:
		return &CancelImportInvalidStatusError{
			Status: imp.Status,
		}
	case entity.ImportStatusCanceled:
		return nil
	default:
		return errors.Errorf("unknown status: %s", imp.Status)
	}

	imp.Status = entity.ImportStatusCanceling
	err = imp.Update(db.GetDBWithContext(ctx).Where("status = ?", oldStatus), entity.ImportDBSchema.Status)
	if err != nil {
		return errors.Annotate(err, "update cluster status to canceling failed")
	}

	if !waitCanceled {
		return nil
	}

	for {
		imp, err := GetImportTask(ctx, clusterID, importID, false)
		if err != nil {
			return err
		}
		if isEndStatus(imp.Status) {
			return nil
		}

		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-time.After(time.Second):
		}
	}
}

// GetLightningConfigTemplate retrieves the lightning configuration template for a given cluster ID.
// It first fetches the cluster information using the provided cluster ID, and then retrieves the
// corresponding lightning configuration template from the configuration center or gitops.
//
// Parameters:
//
//	ctx - The context for the request.
//	clusterID - The ID of the cluster for which the lightning configuration template is to be retrieved.
//
// Returns:
//
//	*cloudv1.Configuration - The lightning configuration template for the specified cluster.
//	error - An error if the operation fails, otherwise nil.
func (c UserContext) GetLightningConfigTemplate(ctx context.Context, clusterID uint64) (*cloudv1.Configuration, error) {
	cluster, err := grpcli.GetClusterByID(ctx, clusterID)
	if err != nil {
		return nil, errors.Annotate(err, "get cluster by ID failed")
	}
	return importcfg.GetLightningConfigTemplateByCluster(ctx, c.configCenter, cluster)
}

func (c UserContext) GetImport(ctx context.Context, orgID, projectID, clusterID, importID entity.ID) (*ImportTask, error) {
	if err := c.CheckClusterOwner(ctx, orgID, projectID, clusterID); err != nil {
		return nil, err
	}
	imp, err := GetImportTask(ctx, clusterID, importID, true)
	if err != nil {
		return nil, err
	}
	return imp, nil
}

func (c UserContext) ListImport(ctx context.Context, orgID, projectID, clusterID entity.ID, pager utils.Pager, statuses []entity.ImportStatus) (int64, []*ImportTask, error) {
	if err := c.CheckClusterOwner(ctx, orgID, projectID, clusterID); err != nil {
		return 0, nil, err
	}
	return ListImport(ctx, clusterID, pager, statuses, false)
}

// GetImportTask retrieves single import task from metaDB and load the cluster information
func GetImportTask(ctx context.Context, clusterID, importID entity.ID, needLoadExtraInfo bool) (*ImportTask, error) {
	imp, err := dbutil.GetImportByID(db.GetDBWithContext(ctx), clusterID, importID)
	if err != nil {
		return nil, err
	}
	result := &ImportTask{
		Import: *imp,
	}
	if needLoadExtraInfo {
		if err := LoadImportClusterInfo(ctx, result, false); err != nil {
			return nil, errors.Annotate(err, "load import cluster info failed")
		}
	}
	return result, nil
}

func ListImport(ctx context.Context, clusterID entity.ID, pager utils.Pager, statuses []entity.ImportStatus, createdByDM bool) (int64, []*ImportTask, error) {
	var imports []entity.Import
	var totalCount int64
	logger := log.WithContextL(ctx)
	var gerr error = db.Transaction(func(tx *gorm.DB) error {
		sortBufferSize := config.Global().ImportConfig.SortBufferSize
		if sortBufferSize <= 0 || sortBufferSize > maxSortBufferSize {
			logger.Warn("invalid import sortBufferSize ", zap.Any("sortBufferSize", sortBufferSize))
			sortBufferSize = defaultSortBufferSize
		}
		if err := tx.Exec(fmt.Sprintf("SET SESSION sort_buffer_size = %d;", sortBufferSize)).Error; err != nil {
			logger.Warn("failed to set session sort_buffer_size", zap.Any("sortBufferSize", sortBufferSize), zap.Error(err))
		}
		var err error
		totalCount, err = dbutil.CountImport(tx, clusterID)
		if err != nil {
			return err
		}
		if createdByDM {
			imports, err = dbutil.ListImportCreatedByDM(tx, clusterID, pager, statuses)
		} else if len(statuses) == 0 {
			imports, err = dbutil.ListImport(tx, clusterID, pager)
		} else {
			imports, err = dbutil.ListImportByStatuses(tx, clusterID, pager, statuses)
		}
		if err != nil {
			return err
		}
		return nil
	})
	importTasks := make([]*ImportTask, len(imports))
	for i, imp := range imports {
		importTasks[i] = &ImportTask{
			Import: imp,
		}
	}
	return totalCount, importTasks, gerr
}

type adjustImportOpts struct {
	IsForCreate bool
}

type AdjustImportOption func(aios *adjustImportOpts)

func ForCreate() AdjustImportOption {
	return func(aios *adjustImportOpts) {
		aios.IsForCreate = true
	}
}

func (c Model) AdjustImport(
	ctx context.Context,
	orgID entity.ID, projectID entity.ID,
	imp *ImportTask,
	opts ...AdjustImportOption,
) error {
	var aios adjustImportOpts
	for _, o := range opts {
		o(&aios)
	}
	plainSourceURL, accessConfig, err := imp.ParseSourceURL()
	if err != nil {
		return errors.Annotate(err, "parse source URL error")
	}

	if imp.Cluster == nil {
		imp.Cluster, err = grpcli.GetClusterWithDataplaneByID(ctx, uint64(imp.ClusterID))
		if err != nil {
			c.log.Error("failed to get cluster by ID", zap.Error(err))
			return errors.Annotate(err, "call cluster rpc failed")
		}
	}
	imp.TargetUsername = entity.CloudAdminUser
	confidentialInfo, err := grpcli.GetClusterConfidentialInfoByID(ctx, uint64(imp.ClusterID))
	if err != nil {
		return errors.Annotate(err, "get cluster's confidential information by ID error")
	}
	imp.TargetPassword = confidentialInfo.GetEncryptedCloudAdminPasswd()

	if len(imp.Name) == 0 {
		imp.Name = fmt.Sprintf("import_%s", time.Now().Format(time.RFC3339))
	}

	ctx = common.ContextWithImport(ctx)
	cluster, network, _, err := common.GetClusterNetworkAgentById(ctx, uint64(imp.ClusterID))
	if err != nil {
		return errors.Annotatef(err, "get network info for the cluster ('%d') error", uint64(imp.ClusterID))
	}
	// add external ID / credential files into source URL
	switch imp.DataSourceType {
	case entity.ImportDataSourceS3:
		if len(accessConfig.AWSSecretAccessKey) > 0 && aios.IsForCreate {
			encryptedSK, err := c.EncryptBase64(accessConfig.AWSSecretAccessKey)
			if err != nil {
				return errors.Annotate(err, "failed to encrypt aws secret access key")
			}
			accessConfig.AWSSecretAccessKey = encryptedSK
		}
		if aios.IsForCreate {
			accessConfig.AWSRoleARN = ""
			accessConfig.AWSExternalID = ""
		} else if network != nil && network.AWS != nil {
			accessConfig.AWSExternalID = network.AWS.ExternalID
		}
	case entity.ImportDataSourceGcs:
		if !aios.IsForCreate {
			if cluster.InfraProviderType == infraapi.ProviderTypeGCP {
				accessConfig.GCPCredentialFile = gcs.CredentialsFilePath
			}
		}
	case entity.ImportDataSourceAzblob:
		if len(accessConfig.AzblobSASToken) > 0 && aios.IsForCreate {
			encryptedToken, err := c.EncryptBase64(accessConfig.AzblobSASToken)
			if err != nil {
				return errors.Annotate(err, "failed to encrypt azure sas token")
			}
			accessConfig.AzblobSASToken = encryptedToken
		}
	}

	accessConfig.Region = imp.BucketRegion

	imp.SourceURL = plainSourceURL // cleanup and re-generate the params
	if err := imp.SetAccessConfigInSourceURL(accessConfig); err != nil {
		return errors.Annotate(err, "update the access config error")
	}

	return nil
}

func (c UserContext) AdjustImport(
	ctx context.Context,
	orgID entity.ID, projectID entity.ID,
	imp *ImportTask,
	opts ...AdjustImportOption,
) error {
	if err := c.Model.AdjustImport(ctx, orgID, projectID, imp, opts...); err != nil {
		return err
	}
	imp.Creator = c.UserEmail()
	return nil
}

func (c UserContext) AdjustLocalImport(
	ctx context.Context,
	imp *ImportTask,
	req *dataflow.ImportCreateReq,
) error {
	if req.LocalImportCfg == nil {
		return nil
	}

	if err := LoadImportClusterInfo(ctx, imp, false); err != nil {
		return errors.Annotate(err, "load import cluster info error")
	}
	switch imp.Cluster.GetCloudProviderName() {
	case "aws":
		imp.DataSourceType = entity.ImportDataSourceS3
	case "gcp":
		imp.DataSourceType = entity.ImportDataSourceGcs
	default:
		return errors.New("unknown provider type for the cluster")
	}

	imp.SourceURL = NewLocalImportBucketNaming(req.Region, req.LocalImportCfg.FileName, c.UserID()).GenCompletedUrl()
	imp.AwsRoleArn = ""
	return nil
}

// checkRunningChangefeeds queries the biz_changefeeds table and returns a forbidden error
// listing the active changefeed names if any non-deleted changefeeds exist on the cluster.
func (c UserContext) checkRunningChangefeeds(ctx context.Context, clusterID entity.ID) error {
	changefeeds, bizErr := container.Global().BizChangefeedRepo.List(ctx,
		func(qs entity2.BizChangefeedQuerySet) entity2.BizChangefeedQuerySet {
			return qs.ClusterIDEq(entity2.ID(clusterID)).
				OperationStatusNotIn(
					entity2.BizChangefeedOperationStatusDeleting,
					entity2.BizChangefeedOperationStatusDeleted,
				)
		},
	)
	if bizErr != nil {
		return errors.Annotate(bizErr, "list biz changefeeds failed")
	}
	if len(changefeeds) == 0 {
		return nil
	}
	names := make([]string, len(changefeeds))
	for i, cf := range changefeeds {
		names[i] = cf.Name
	}
	return errcode.NewForbiddenErr(errors.Errorf(ConnectorErr, strings.Join(names, ", ")))
}

func (c UserContext) CreateImport(
	ctx context.Context,
	orgID, projectID entity.ID,
	imp *ImportTask,
) (id uint64, err error) {
	logger := log.WithContextL(ctx)
	_, span := Tracer.Start(ctx, "UserContext.CreateImport")
	defer func() { tracing.EndSpan(span, err) }()

	if err := LoadImportClusterInfo(ctx, imp, false); err != nil {
		return 0, errors.Annotate(err, "load import cluster info error")
	}
	if imp.Cluster.GetReadOnly() {
		return 0, errcode.NewForbiddenErr(errors.New(ReadonlyForbidenErr))
	}

	// todo: move AdjustImport outside CreateImport
	if err := c.AdjustImport(ctx, orgID, projectID, imp, ForCreate()); err != nil {
		return 0, errors.AddStack(err)
	}

	if !feature.IsFeatureEnabled(feature.SkipImportVildation,
		orgID.Uint64(), projectID.Uint64(), imp.ClusterID.Uint64(), "", "") {
		if err := c.checkRunningChangefeeds(ctx, imp.ClusterID); err != nil {
			return 0, err
		}
	}
	imp.ServerTier = "dedicated"
	imp.Provider = imp.Cluster.GetInfraProviderType()
	imp.Region = imp.Cluster.GetRegionName()
	imp.ClusterVersion = imp.Cluster.GetVersion()

	// Azure dedicated tier does not support backup settings yet, ignore
	if imp.Cluster.InfraProviderType != infraapi.ProviderTypeAKS {
		clusterId := imp.ClusterID.Uint64()
		backupSetting, err := container.Global().BackupSettingModel.QueryBackupSettingForAPI(ctx, &clusterId)
		if err != nil {
			return 0, errors.AddStack(err)
		}
		if backupSetting.PiTRSetting != nil && backupSetting.PiTRSetting.Enable &&
			!feature.IsFeatureEnabled(feature.SkipCdcPitrCheckWhenImport,
				orgID.Uint64(), projectID.Uint64(), imp.ClusterID.Uint64(), "", "") {
			plog.Warn("pitr limited", zap.Uint64("org id", orgID.Uint64()), zap.Uint64("project id", projectID.Uint64()))
			return 0, errcode.NewForbiddenErr(errors.New("Unable to create an import task when PITR is enabled in the cluster."))
		}
	}

	if imp.Cluster.Status != core.ClusterStatus_CLUSTER_STATUS_NORMAL {
		return 0, errcode.NewForbiddenErr(errors.New("Please import data when cluster is normal."))
	}

	var count int64
	if err = db.TransactionWithContext(ctx, func(tx *gorm.DB) (gerr error) {
		// TODO: check whether there's an import job on cluster? yes, will do in another PR
		count, err = dbutil.CountRunningImports(tx, imp.ClusterID)
		if err != nil {
			return err
		}
		return nil
	}); err != nil {
		return 0, errors.AddStack(err)
	}
	if count > 0 && !feature.IsFeatureEnabled(feature.AllowMultiImportJobs, orgID.Uint64(), projectID.Uint64(), imp.ClusterID.Uint64(), "", "") {
		return 0, errcode.NewForbiddenErr(errors.New("There is already a running import task, please click and submit a new task after the previous task is completed"))
	}

	hasClusterJobRegistered := false

	err = db.TransactionWithContext(ctx, func(tx *gorm.DB) (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")
			}
		}()

		if err := imp.Create(tx); err != nil {
			return errors.AddStack(err)
		}

		logger.Info("import record inserted (not committed)")
		registerCtx, registerCancel := context.WithTimeout(ctx, 10*time.Second)
		defer registerCancel()
		if err := cluster_job.RegisterImportClusterJob(registerCtx, uint64(imp.ClusterID), uint64(imp.ID)); err != nil {
			logger.Error("register import cluster job has errors",
				zap.Error(err),
				zap.Uint64("import ID", imp.ID.Uint64()),
				zap.Uint64("cluster ID", imp.ClusterID.Uint64()),
			)
			return errors.Annotate(err, "register import cluster job failed")
		}
		hasClusterJobRegistered = true
		logger.Info("import cluster job registered",
			zap.Uint64("import ID", imp.ID.Uint64()),
			zap.Uint64("cluster ID", imp.ClusterID.Uint64()),
		)
		return nil
	})
	if err != nil {
		// NOTE: here we don't put the roll-back logic inside the transaction,
		// because in some situations, the gRPC call succeeded, but the commit / rollback failed.
		// In this situation, the data modifications in the previous Tx are all rolled-back.
		// So we need also roll back the creation of the cluster job.
		if hasClusterJobRegistered {
			func() {
				registerCtx, registerCancel := context.WithTimeout(ctx, 10*time.Second)
				defer registerCancel()
				if err := cluster_job.DeregisterImportClusterJob(registerCtx, uint64(imp.ClusterID), uint64(imp.ID)); err != nil {
					logger.Error("deregister import cluster job has errors",
						zap.Error(err),
						zap.Uint64("import ID", imp.ID.Uint64()),
						zap.Uint64("cluster ID", imp.ClusterID.Uint64()),
					)
				} else {
					logger.Info("import cluster job deregistered",
						zap.Uint64("import ID", imp.ID.Uint64()),
						zap.Uint64("cluster ID", imp.ClusterID.Uint64()),
					)
				}
			}()
		}
		return 0, err
	}

	if err := publishEvent(c.publisher, imp); err != nil {
		c.log.Error("publish import event fail", zap.Error(err), zap.Int("import ID", int(imp.ID)))
	}

	logger.Info("import created", zap.Uint64("import ID", imp.ID.Uint64()), zap.Any("import", imp))

	return uint64(imp.ID), nil
}

type ValidateImportSingleCheckItemError struct {
	Item dataflow.ImportValidationItem
	Err  error
}

func (e *ValidateImportSingleCheckItemError) Error() string {
	return fmt.Sprintf("validate on single item %s failed: %v", e.Item, e.Err)
}

type ValidateImportPrepareError struct {
	Err error
}

func (e *ValidateImportPrepareError) Error() string {
	return fmt.Sprintf("prepare validation failed: %v", e.Err)
}

func (c Model) ValidateImport(
	ctx context.Context,
	tenantID entity.ID,
	projectID entity.ID,
	imp *ImportTask,
	checkItems []dataflow.ImportValidationItem,
) error {
	logger := c.log.With(log.FieldsFromContext(ctx)...).
		With(zap.Uint64("cluster_id", imp.ClusterID.Uint64()))

	logger.Info("import entity before adjust", zap.Any("import", imp))

	if err := c.AdjustImport(ctx, tenantID, projectID, imp); err != nil {
		logger.Error("Failed to adjust import entity.", zap.Error(err))
		return &ValidateImportPrepareError{
			Err: err,
		}
	}
	logger.Info("import entity after adjust", zap.Any("import", imp))
	checkBuilder, err := c.newValidatorBuilder(imp)
	if err != nil {
		logger.Error("generate validator builder error", zap.Error(err))
		return &ValidateImportPrepareError{
			Err: err,
		}
	}
	for _, item := range checkItems {
		if (item == dataflow.ImportValidationItem_IMPORT_PRECHECK || item == dataflow.ImportValidationItem_IMPORT_DATA_ONLY_CHECK) &&
			feature.IsFeatureEnabled(feature.SkipImportVildation, tenantID.Uint64(), projectID.Uint64(), imp.ClusterID.Uint64(), "", "") {
			// skip IMPORT_PRECHECK/IMPORT_DATA_ONLY_CHECK validation if the organization need not to validate
			// Especially when backend of dedicated tier is set to tidb, we need to skip the table empty validation
			// TODO: Remove this comment after support backend switch
			logger.Info("skip IMPORT_PRECHECK import validation", zap.Uint64("tenant_id", tenantID.Uint64()),
				zap.Uint64("project_id", projectID.Uint64()), zap.Uint64("cluster_id", imp.ClusterID.Uint64()))
			continue
		}
		validator, err := checkBuilder.Build(item)
		if err != nil {
			return errors.Annotatef(err, "use builder to generate validator on item %s error", item.String())
		}
		if err := validator.Validate(ctx); err != nil {
			return &ValidateImportSingleCheckItemError{
				Item: item,
				Err:  err,
			}
		}
	}
	return nil
}

func genImportObjectSpec(imp *ImportTask) *cloudv1.ImportSpec {
	result := &cloudv1.ImportSpec{
		ClusterName: strconv.FormatUint(imp.ClusterID.Uint64(), 10),
		Source: cloudv1.Source{
			// https://github.com/tidbcloud/aws-shared-provider/blob/fca18db6fd1ffb6ae8801498d858e5929adc5b69/pkg/controllers/imports/env.go#L181
			// 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: "cloud_admin",
			// 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),
		},
	}

	if imp.DataSourceType == entity.ImportDataSourceS3 {
		result.AccessConfig = cloudv1.AccessConfig{
			RoleArn: imp.AwsRoleArn,
		}
	}
	return result
}

func (c Model) newInfraAPIImportObject(ctx context.Context, logger *zap.Logger, imp *ImportTask) (*cloudv1.Import, error) {

	logger.Info("begin to build import CR")

	if imp.Cluster == nil {
		return nil, errors.New("cluster is nil")
	}
	importSpec := genImportObjectSpec(imp)

	if imp.Cluster.GetAttributes().GetUseNewDedicatedArch() && imp.Cluster.GetInfraProviderType() != infraapi.ProviderTypeAKS {
		// for cluster in new arch, every tidb cluster has dedicated node role
		// we use cluster node role instead of admin node role
		nodeRole, bizErr := pkggrpcli.InfraClient().GetClusterNodeRole(ctx, imp.Cluster.Id)
		if bizErr != nil {
			return nil, fmt.Errorf("get cluster %d node role error: %w", imp.Cluster.Id, bizErr)
		}
		importSpec.NodeRole = &nodeRole
	}
	importObj, err := newImportAPIObject(uint64(imp.ID), importSpec, imp.Cluster)
	if err != nil {
		return nil, err
	}

	imp.TargetPassword, err = c.DecryptBase64(imp.TargetPassword)
	if err != nil {
		return nil, errors.Annotatef(err, "failed to decrypt, import id: %d", imp.ID)
	}

	lightningCfg, impCfgBytes, err := importcfg.GenerateImportConfig(ctx, c.configCenter, imp.Import, imp.LightningSourceURL, imp.Cluster)
	if err != nil {
		errMsg := "Failed to generate import config"
		return nil, errors.Annotate(err, errMsg)
	}

	importObj.Spec.LightningConfig = string(impCfgBytes)
	image := fmt.Sprintf("%s:%s", *lightningCfg.BaseImage, *lightningCfg.Tag)
	importObj.Annotations[importImageAnnotation] = image
	hackImportLocalMode(imp, importObj)

	logger.Info("lightning raw configuration", zap.Any("rawConfig", lightningCfg))
	logger.Info("lightning applied configuration", zap.String("appliedConfig", string(impCfgBytes)))
	logger.Info("lightning image info", zap.String("image", image), zap.String("clusterVersion", imp.Cluster.GetVersion()))
	return importObj, nil
}

func hackImportLocalMode(imp *ImportTask, importCR *cloudv1.Import) {

	// only enabled in nightly env
	if config.Global().Env != "nightly" {
		return
	}

	// only enabled with cluster which name contains keyworkd 'FOR_LOCAL_TEST'
	if !strings.Contains(imp.Cluster.GetName(), "FOR_LOCAL_TEST") {
		return
	}

	importCR.Labels[importLocalLabel] = "true"
}

func (c Model) GetImportConfig(ctx context.Context, imp *ImportTask, id *dataflow.ClusterIdReq) (*infra_config.GenericConfig, error) {
	var err error
	if imp.Cluster == nil {
		cluster, err := grpcli.GetAnyClusterByID(ctx, id.ClusterId)
		if err != nil {
			return nil, errors.Annotate(err, "failed to get cluster")
		}
		imp.Cluster = cluster
	}

	oldPassword := imp.TargetPassword
	imp.TargetPassword, err = c.DecryptBase64(imp.TargetPassword)
	if err != nil {
		return nil, errors.Annotatef(err, "failed to decrypt, import id: %d", imp.ID)
	}

	_, impCfgBytes, err := importcfg.GenerateImportConfig(ctx, c.configCenter, imp.Import, imp.LightningSourceURL, imp.Cluster)
	if err != nil {
		errMsg := "Failed to generate import config"
		c.log.Error(errMsg, zap.Error(err))
		return nil, errors.Annotate(err, errMsg)
	}
	imp.TargetPassword = oldPassword

	impConfigTOML := new(cloudv1.TomlConfig)
	if err := impConfigTOML.UnmarshalTOML(impCfgBytes); err != nil {
		return nil, errors.Annotate(err, "parse config TOML data error")
	}
	impConfig := infra_config.New(impConfigTOML.MP)

	return impConfig, nil
}

// BgContext is meant for system worker/controller
type BgContext struct {
	UserContext
	// aliveLocalImports is a map of import ID to import status
	// it is used to keep track of local imports that are still running
	// so that we can detect the worker's failover when load data is running
	aliveLocalImports *sync.Map
	api               ResourceAPI
}

// BgContext returns a BgContext
func (m Model) BgContext(ctx context.Context, importTask *ImportTask) (BgContext, error) {
	tenantID := importTask.TenantID.Uint64()
	projectID := importTask.ProjectID.Uint64()
	clusterID := importTask.ClusterID.Uint64()
	providerType := importTask.Cluster.GetInfraProviderType()
	region := importTask.Cluster.GetRegionName()

	if tenantID == 0 && importTask.Cluster != nil {
		tenantID = importTask.Cluster.GetTenantId()
	}
	systemUser, err := accountdb.GetOrCreateSystemUser(tenantID)
	if err != nil {
		return BgContext{}, err
	}

	ctx = common.ContextWithImport(ctx)
	newModel := m
	resourceAPI, err := NewRegionalResourceAPI(ctx, providerType, region, tenantID)
	if err != nil {
		return BgContext{}, err
	}
	return BgContext{
		UserContext: UserContext{
			Model:         newModel,
			tenantAndUser: &accountdb.User{TenantID: tenantID},
			log: newModel.log.With(log.FieldsFromContext(ctx)...).With(
				zap.String("runID", logid.GenLogid()),
				zap.Uint64("userID", systemUser.ID),
				zap.Uint64("tenantID", tenantID),
				zap.Uint64("projectID", projectID),
				zap.Uint64("clusterID", clusterID),
				zap.String("module", "import"),
				zap.Uint64("importID", importTask.ID.Uint64()),
				zap.String("dataSource", entity.ConvertDataSource2Str(importTask.DataSourceType)),
				zap.String("dataFormat", entity.ConvertDataFormat2Str(importTask.DataFormat)),
				zap.String("importStatus", string(importTask.Status)),
			),
		},
		// must delivered by the Model(root context)
		aliveLocalImports: newModel.aliveLocalImports,
		api:               resourceAPI,
	}, nil
}

func (c BgContext) transToStatus(ctx context.Context, imp *ImportTask, status entity.ImportStatus, msg string) (err error) {
	_, span := Tracer.Start(ctx, "BgContext.transToStatus")
	defer func() { tracing.EndSpan(span, err) }()
	span.SetAttributes(
		attribute.String("import_name", infraapi.ImportName(imp.ID.Uint64())),
		attribute.String("cluster_id", fmt.Sprintf("%v", imp.ClusterID)),
		attribute.String("status", status.String()))
	logger := c.log.With(zap.Any("fromStatus", imp.Status), zap.Any("toStatus", status))

	logger.Info("begin to trans import status", zap.String("message", msg))
	if imp.Status == status {
		logger.Info("trans import status nothing todo because the status are the same")
		return nil
	}

	if imp.LocalImportCfg != nil {
		c.aliveLocalImports.Store(imp.ID, status)
	}

	imp.Status = status
	var fields []entity.ImportDBSchemaField
	fields = append(fields, entity.ImportDBSchema.Status)

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

	if isEndStatus(status) && imp.FinishedAt == nil {
		t := time.Now()
		imp.FinishedAt = &t
		fields = append(fields, entity.ImportDBSchema.FinishedAt)
	}

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

	err = db.Transaction(func(tx *gorm.DB) error {
		return imp.Update(tx, fields...)
	})
	if err != nil {
		logger.Info("failed to trans import status", zap.Error(err))
		return errors.AddStack(err)
	}
	logger.Info("successfully update import status")

	// local import manage the clean up by itself
	if isEndStatus(status) && imp.LocalImportCfg == nil {
		return c.cleanUp(ctx, imp)
	}

	return nil
}

func isEndStatus(status entity.ImportStatus) bool {
	switch status {
	case entity.ImportStatusCompleted,
		entity.ImportStatusFailed,
		entity.ImportStatusCanceled:
		return true
	default:
		return false
	}
}

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

	logger := c.log.With(zap.String("status", string(imp.Status)))
	logger.Info("begin to clean up import task")
	defer func() {
		if err != nil {
			logger.Error("failed to clean up import task", zap.Error(err))
		}
	}()

	// Currently we only fetch the un-cleaned imports to sync.
	// And we just cleaned up the completed imports.
	if !isEndStatus(imp.Status) {
		logger.Info("skip clean up import task for status is not completed")
		return errors.Errorf("can't clean up at status: %s", imp.Status)
	}

	if imp.LocalImportCfg == nil {
		importCR, err := c.api.GetImport(ctx, imp.ProjectID.Uint64(), imp.ID.Uint64())
		if err != nil {
			if c.api.IsNotFound(err) {
				// continue
				logger.Info("import CR successfully executed deletion")
			} else if importCR != nil && !importCR.DeletionTimestamp.IsZero() {
				logger.Info("import CR is executing deletion, waiting for next retry",
					zap.Time("deletionTime", importCR.DeletionTimestamp.Time),
					zap.String("cleanUpTime", importCR.Annotations[jobDeletionAnnotation]),
				)
				if err := c.DeregisterImportClusterJobIgnoreExist(ctx, logger, imp.ID.Uint64(), imp.ClusterID.Uint64()); err != nil {
					logger.Error("failed to deregister import cluster job", zap.Error(err))
					return err
				}
				return nil
			} else {
				logger.Error("failed to check import CR when clean up import task", zap.Error(err))
				return errors.AddStack(err)
			}
		} else {
			if err = c.api.DeleteImport(ctx, imp.Cluster.GetProjectId(), uint64(imp.ID)); err != nil {
				logger.Error("failed to delete import CR when clean up import task", zap.Error(err))
				return errors.AddStack(err)
			}
			logger.Info("import CR marked for deletion, waiting for next retry")
			if err := c.DeregisterImportClusterJobIgnoreExist(ctx, logger, imp.ID.Uint64(), imp.ClusterID.Uint64()); err != nil {
				logger.Error("failed to deregister import cluster job", zap.Error(err))
				return err
			}
			return nil
		}
	} else {
		if _, ok := c.aliveLocalImports.Load(imp.ID); ok {
			c.aliveLocalImports.Delete(imp.ID)
		}
	}

	imp.CleanUp = true
	if err := imp.CleanSourceUrlQuery(); err != nil {
		logger.Error("failed to clean the query of source url", zap.Error(err), zap.String("sourceUrl", imp.SourceURL))
	}
	fields := []entity.ImportDBSchemaField{
		entity.ImportDBSchema.CleanUp,
		entity.ImportDBSchema.SourceURL,
	}

	// NOTE: Put this logic outside the transaction to avoid deadlock when calling the 'delete cluster job' API,
	// which will also update the clusters table.
	if err := db.TransactionWithContext(ctx, func(tx *gorm.DB) error {
		if err := imp.Update(tx, fields...); err != nil {
			return errors.Annotate(err, "update import entity error")
		}

		// to ensure the proper cleaning of cluster jobs
		if err := c.DeregisterImportClusterJob(ctx, logger, imp.ID.Uint64(), imp.ClusterID.Uint64()); err != nil {
			return err
		}

		return nil
	}); err != nil {
		logger.Error("failed to update import clean_up flag when clean up import task", zap.Error(err))
		return errors.AddStack(err)
	}

	logger.Info("successfully clean up import task")
	if err2 := publishEvent(c.publisher, imp); err2 != nil {
		c.log.Error("publish import event fail", zap.Error(err), zap.Int("importID", int(imp.ID)))
	}
	return nil
}

func (c BgContext) DeregisterImportClusterJob(ctx context.Context, logger *zap.Logger, importID, clusterID uint64) error {

	registerCtx, registerCancel := context.WithTimeout(ctx, 10*time.Second)
	defer registerCancel()
	if err := cluster_job.DeregisterImportClusterJob(registerCtx, clusterID, importID); err != nil {
		logger.Error("failed to deregister import cluster job",
			zap.Uint64("importID", importID),
			zap.Uint64("clusterID", clusterID),
		)
		return errors.Annotate(err, "deregister import cluster job failed")
	}

	logger.Info("deregister import cluster job success",
		zap.Uint64("importID", importID),
		zap.Uint64("clusterID", clusterID),
	)

	return nil
}

func (c BgContext) DeregisterImportClusterJobIgnoreExist(ctx context.Context, logger *zap.Logger, importID, clusterID uint64) error {
	logger.Info("begin deregister import cluster job")
	registerCtx, registerCancel := context.WithTimeout(ctx, 10*time.Second)
	defer registerCancel()
	if clusterJob, err := cluster_job.GetImportClusterJob(registerCtx, clusterID, importID); err != nil {
		logger.Error("failed to get import cluster job",
			zap.Uint64("importID", importID),
			zap.Uint64("clusterID", clusterID),
		)
		return errors.Annotate(err, "get import cluster job failed")
	} else if clusterJob == nil {
		logger.Info("import cluster job not found",
			zap.Uint64("importID", importID),
			zap.Uint64("clusterID", clusterID))
		return nil
	}

	return c.DeregisterImportClusterJob(ctx, logger, importID, clusterID)
}

func (c BgContext) OnPreparing(ctx context.Context, imp *ImportTask) (err error) {
	ctx, span := Tracer.Start(ctx, "BgContext.OnPreparing")
	defer func() { tracing.EndSpan(span, err) }()
	span.SetAttributes(attribute.String("import_name", infraapi.ImportName(imp.ID.Uint64())))

	logger := c.log
	logger.Info("begin to prepare import task")

	if _, err := c.api.GetImport(ctx, imp.ProjectID.Uint64(), imp.ID.Uint64()); err != nil {
		if !c.api.IsNotFound(err) {
			logger.Error("failed to get import CR on apiserver, try again latter", zap.Error(err))
			return err
		}
		if err := c.createImportCR(ctx, imp); err != nil {
			logger.Error("failed to create import CR", zap.Error(err))
			return err
		}
	}

	return c.updateInfo(ctx, imp)
}

func (c BgContext) createImportCR(ctx context.Context, imp *ImportTask) (err error) {
	logger := c.log
	logger.Info("begin to create import CR")

	cluster, berr := grpcli.GetClusterWithDataplaneByID(ctx, uint64(imp.ClusterID))
	if berr != nil {
		// we should try again latter if failed to get cluster by ID
		if berr.ErrCode() != bizerr.ErrCodeRecordNotFound {
			logger.Error("failed to get cluster by ID, try again latter", zap.Error(berr))
			return errors.AddStack(err)
		}

		logger.Error("cluster not found, may be has been deleted", zap.Any("clusterID", imp.ClusterID), zap.Error(berr))
		return c.transToStatus(ctx, imp, entity.ImportStatusFailed, "cluster not found, may be has been deleted")
	}
	imp.Cluster = cluster

	if err := c.adjustSourceURL(ctx, imp); err != nil {
		logger.Error("failed to adjust source url, maybe sourceURL is invalid", zap.String("sourceUrl", imp.SourceURL), zap.Error(err))
		return c.transToStatus(ctx, imp, entity.ImportStatusFailed, "Encountered a problem when creating Import, please contact support.")
	}

	impCR, err := c.newInfraAPIImportObject(ctx, logger, imp)
	if err != nil {
		logger.Error("failed to generate import CR", zap.Error(err))
		return c.transToStatus(ctx, imp, entity.ImportStatusFailed, "Encountered a problem when creating Import, please contact support.")
	}
	impCR.Spec.StorageRequest = GetImportDefaultStorageRequestConfig()

	if err := c.api.CreateImportIgnoreExists(ctx, impCR); err != nil {
		logger.Error("failed to create import CR on apiserver", zap.Error(err))
		return c.transToStatus(ctx, imp, entity.ImportStatusFailed, "Encountered a problem when creating Import, please contact support.")
	}

	logger.Info("successfully on import prepar stage, begin update import meta info")
	return nil
}

func (c BgContext) adjustSourceURL(_ context.Context, imp *ImportTask) error {

	switch imp.DataSourceType {
	case entity.ImportDataSourceS3:
		// decrypted AwsSecretAccessKey will be used in getEstimatedStorageRequest
		sourceUrl, err := url.Parse(imp.SourceURL)
		if err != nil {
			return errors.Annotatef(err, "failed to parse source url: %s", imp.SourceURL)
		}

		q := sourceUrl.Query()
		awsSecretAccessKey := q.Get("secret-access-key")
		if awsSecretAccessKey != "" {
			awsSecretAccessKey, err = c.DecryptBase64(awsSecretAccessKey)
			if err != nil {
				return errors.Annotatef(err, "failed to decrtpt AwsSecretAccessKey: %s", awsSecretAccessKey)
			}
			q.Set("secret-access-key", awsSecretAccessKey)
		}
		q.Set("region", imp.BucketRegion)
		sourceUrl.RawQuery = q.Encode()
		imp.LightningSourceURL = sourceUrl.String()

	case entity.ImportDataSourceAzblob:
		azurl, err := azblob.ParseAzblobURL(imp.SourceURL)
		if err != nil {
			return errors.Annotatef(err, "failed to parse source url: %s", imp.SourceURL)
		}
		if sasToken := azurl.SASToken(); sasToken != "" {
			decryptedToken, err := c.DecryptBase64(sasToken)
			if err != nil {
				return errors.Annotatef(err, "failed to decrtpt AzureSASToken: %s", sasToken)
			}
			azurl.SetSASToken(decryptedToken)
		}
		imp.LightningSourceURL = azurl.ToInnerURL(true)
	default:
		imp.LightningSourceURL = imp.SourceURL
	}

	return nil
}

func (c BgContext) updateInfo(ctx context.Context, imp *ImportTask) error {
	_, span := Tracer.Start(ctx, "BgContext.updateFromLightningV1")
	err := c.updateFromLightningV1(ctx, imp)
	if err != nil {
		span.SetStatus(codes.Error, err.Error())
		span.RecordError(err)
	}
	span.End()
	return err
}

func convertStatus(status string) entity.ImportStatus {
	switch status {
	case string(cloudv1.ImportCompleted):
		return entity.ImportStatusCompleted
	case string(cloudv1.ImportFailed):
		return entity.ImportStatusFailed
	case string(cloudv1.ImportPrepare), "":
		return entity.ImportStatusPreparing
	case string(cloudv1.ImportRunning):
		return entity.ImportStatusImporting
	default:
		return entity.ImportStatusUnknown
	}
}

func (c BgContext) saveCompletedTables(ctx context.Context, tx *gorm.DB, imp *ImportTask, importStatus *cloudv1.ImportStatus) error {
	logger := c.log.With(log.FieldsFromContext(ctx)...)
	importID := imp.ID
	imp.TableResults = entity.ImportTables{Tables: make([]entity.ImportTable, 0, len(importStatus.AllTableDetails))}
	for _, tblDetail := range importStatus.AllTableDetails {
		importTableResult := convertToTableResult(tblDetail.Phase)
		if len(importTableResult) <= 0 {
			continue
		}
		importStartTime := time.Unix(tblDetail.TimeRecord.ImportStartTime, 0)
		importFinishTime := time.Unix(tblDetail.TimeRecord.ImportFinishTime, 0)
		postImportStartTime := time.Unix(tblDetail.TimeRecord.PostProcessStartTime, 0)
		postImportFinishTime := time.Unix(tblDetail.TimeRecord.PostProcessFinishTime, 0)
		importTblEntity := entity.ImportTable{
			ImportID:             importID,
			TableName:            tblDetail.Name,
			Result:               importTableResult,
			Message:              tblDetail.Message,
			ImportStartedAt:      &importStartTime,
			ImportFinishedAt:     &importFinishTime,
			PostImportStartedAt:  &postImportStartTime,
			PostImportFinishedAt: &postImportFinishTime,
		}
		tableWarningMsg, err := c.CheckWarningsByTable(ctx, imp, tblDetail.Name)
		if err != nil {
			logger.Warn("check warnings failed for a table", zap.String("table_name", tblDetail.Name), zap.Error(err))
			tableWarningMsg = ""
		}
		if len(tableWarningMsg) > 0 {
			importTblEntity.Message = tableWarningMsg
			importTblEntity.Result = entity.ImportTableResultWarning
		}
		imp.TableResults.Tables = append(imp.TableResults.Tables, importTblEntity)
	}
	// update import tables
	if err := imp.Update(tx, entity.ImportDBSchema.TableResults); err != nil {
		return err
	}
	return nil
}

func (c BgContext) updateFromLightningV1(ctx context.Context, imp *ImportTask) (err error) {

	logger := c.log
	logger.Info("begin to update import info")
	defer func() {
		if err != nil {
			logger.Error("failed to update import info", zap.Error(err))
		} else {
			logger.Info("successfully update import info")
		}
	}()

	impcr, err := c.api.GetImport(context.TODO(), imp.Cluster.GetProjectId(), uint64(imp.ID))
	if err != nil {
		return errors.Annotate(err, "failed to get import CR")
	}
	imp.TotalSize = &impcr.Status.TotalSize
	imp.ProcessedDataSize = &impcr.Status.ProcessedDataSize
	imp.TotalFiles = impcr.Status.TotalFiles
	imp.CurrentTables.Tables = nil
	for _, v := range impcr.Status.CurrentTables {
		imp.CurrentTables.Tables = append(imp.CurrentTables.Tables, &entity.CurrentTable{
			Name:             v.Name,
			Size:             v.Size,
			Phase:            string(v.Phase),
			ProcessedSize:    v.ProcessedSize,
			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
	if !(imp.Status == entity.ImportStatusCanceled || imp.Status == entity.ImportStatusCanceling) {
		if err = c.transToStatus(ctx, imp, convertStatus(string(impcr.Status.Phase)),
			impcr.Status.Message); err != nil {
			logger.Error("failed to transfer status", zap.Error(err))
		}
	}

	// TODO: add metric
	if imp.Status == entity.ImportStatusCompleted {
		usedTime := time.Since(imp.CreatedAt).Seconds()
		totalSize := float64(*imp.TotalSize)
		replicas := func() float64 {
			components := imp.Cluster.GetComponents()
			for _, c := range components {
				if c.GetComponentType() == core.ComponentType_COMPONENT_TYPE_TIKV {
					replicas := c.GetReplicas()
					if replicas > 0 {
						return float64(replicas)
					}
				}
			}
			return 1.0
		}()
		metrics.ImportThroughputBytes.WithLabelValues(metrics.Dedicate).
			Observe(totalSize / replicas / usedTime)
	}
	ctx = common.ContextWithImport(ctx)
	imp.TaskID = impcr.Status.TaskId
	imp.ArchType = entity.ThreeLayerArch
	fields := []entity.ImportDBSchemaField{
		entity.ImportDBSchema.TotalSize,
		entity.ImportDBSchema.ProcessedDataSize,
		entity.ImportDBSchema.TotalFiles,
		entity.ImportDBSchema.CurrentTables,
		entity.ImportDBSchema.TotalTablesCount,
		entity.ImportDBSchema.CompletedTables,
		entity.ImportDBSchema.PendingTables,
		entity.ImportDBSchema.CompletedPercent,
		entity.ImportDBSchema.PostImportCompletedPercent,
		entity.ImportDBSchema.TaskID,
		entity.ImportDBSchema.ArchType,
	}
	if impcr.Status.TaskPhase != "" {
		imp.TaskPhase = string(impcr.Status.TaskPhase)
		fields = append(fields, entity.ImportDBSchema.TaskPhase)
	}

	if cloudv1.ImportTaskPhaseType(imp.TaskPhase) == cloudv1.ImportTaskPhaseImporting &&
		imp.TotalSize != nil && imp.ProcessedDataSize != nil &&
		*imp.ProcessedDataSize >= *imp.TotalSize && imp.TaskImportingAt == nil {
		t := time.Now()
		imp.TaskImportingAt = &t
		fields = append(fields, entity.ImportDBSchema.TaskImportingAt)
	}
	if cloudv1.ImportTaskPhaseType(imp.TaskPhase) == cloudv1.ImportTaskPhasePostProcessing &&
		imp.TaskPostproceAt == nil {
		t := time.Now()
		imp.TaskPostproceAt = &t
		fields = append(fields, entity.ImportDBSchema.TaskPostproceAt)
	}
	if err := imp.CleanSourceUrlQuery(); err != nil {
		logger.Error("failed to clean the query of source url", zap.String("sourceUrl", imp.SourceURL), zap.Error(err))
	}
	fields = append(fields, entity.ImportDBSchema.SourceURL)

	if image := impcr.Annotations[importImageAnnotation]; image != "" {
		imp.ComponentVersion = util.ParseImage(image)
		fields = append(fields, entity.ImportDBSchema.ComponentVersion)
	}

	err = db.Transaction(func(tx *gorm.DB) error {
		if err := imp.Update(tx, fields...); err != nil {
			return errors.Annotate(err, "update table fields failed")
		}
		return nil
	})
	if err != nil {
		logger.Error("failed to update table fields failed", zap.Error(err))
		return c.transToStatus(ctx, imp, entity.ImportStatusFailed, InternalErr)
	}
	err = db.Transaction(func(tx *gorm.DB) error {
		switch imp.Status {
		case entity.ImportStatusCompleted, entity.ImportStatusFailed:
			if err := c.saveCompletedTables(ctx, tx, 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))
	}

	return nil
}

func convertToTableResult(tblPhase cloudv1.TableImportPhaseType) entity.ImportTableResult {
	switch tblPhase {
	case cloudv1.TableImportPhaseComplete:
		return entity.ImportTableResultSuccess
	case cloudv1.TableImportPhaseFailed:
		return entity.ImportTableResultError
	default:
		return entity.ImportTableResult("")
	}
}

func (c BgContext) OnImporting(ctx context.Context, imp *ImportTask) error {
	// TODO: will remove this after adding import progress here
	importCr, err := c.api.GetImport(ctx, imp.Cluster.GetProjectId(), uint64(imp.ID))
	if err == nil && importCr != nil && importCr.Status.IsTaskCreated {
		lightningClient, err := common.GetLightningClient(ctx, imp.ClusterID, imp.ID)
		if err != nil {
			c.log.Error("failed to get lightning client", zap.Error(err))
		} else {
			progress, err := lightningClient.GetTaskProgress(ctx)
			if err != nil {
				c.log.Error("failed to get task progress", zap.Error(err))
			} else {
				c.log.Info("task progress",
					zap.Any("progress status", progress.Status),
					zap.String("progress message", progress.Message),
					zap.Int("tables count", len(progress.Tables)),
				)
			}
		}
	}

	return c.updateInfo(ctx, imp)
}

func (c BgContext) OnCanceling(ctx context.Context, imp *ImportTask) (err error) {

	logger := c.Log()
	logger.Info("begin to cancel import task")

	defer func() {
		// need transfer to `canceled` whatever happens
		if err != nil {
			logger.Error("some error happened when canceling import task, "+
				"but we still try to trans import task to canceled", zap.Error(err))
		}
		err = c.transToStatus(ctx, imp, entity.ImportStatusCanceled, "")
		if err != nil {
			logger.Error("failed to trans import task to canceled", zap.Error(err))
		} else {
			logger.Info("successfully trans import task to canceled")
		}
	}()
	ctx, span := Tracer.Start(ctx, "BgContext.OnCanceling")
	defer func() { tracing.EndSpan(span, err) }()

	impcr, err := c.api.GetImport(ctx, imp.Cluster.GetProjectId(), uint64(imp.ID))
	if err != nil {
		if c.api.IsNotFound(err) {
			err = nil
			logger.Info("nothing to do, import CR not exists")
			return err
		}
		logger.Error("failed to get import CR", zap.Error(err), zap.Reflect("import", imp))
		return err
	}

	if err = c.api.DeleteImport(ctx, imp.Cluster.GetProjectId(), uint64(imp.ID)); err != nil {
		logger.Error("failed to delete import CR", zap.Error(err), zap.Reflect("import", imp))
		return err
	}

	if err := db.Transaction(func(tx *gorm.DB) error {
		if err := c.saveCompletedTables(ctx, tx, imp, &(impcr.Status)); err != nil {
			return err
		}
		return nil
	}); err != nil {
		logger.Error("failed to save completed tables", zap.Error(err), zap.Reflect("import", imp))
		return err
	}

	return nil
}

func (c BgContext) syncLocalImport(ctx context.Context, imp *ImportTask) error {
	logger := c.log.With(log.FieldsFromContext(ctx)...).Sugar()

	// need to get agent client for each syncing local import?
	// maybe not, because every local import is in the same cluster(serverless)
	ctx = common.ContextWithImport(ctx)
	cluster, _, agentClient, err := common.GetClusterNetworkAgentById(ctx, uint64(imp.ClusterID))
	if err != nil {
		logger.Error("failed to get cluster network agent", zap.Error(err))
		return c.transToStatus(ctx, imp, entity.ImportStatusFailed, InternalErr)
	}
	password, err := c.GenCloudAdminPassword(ctx, logger.Desugar(), cluster.TenantId, cluster.ProjectId, cluster.Id)
	if err != nil {
		logger.Error("failed to generate cloud admin password", zap.Error(err))
		return c.transToStatus(ctx, imp, entity.ImportStatusFailed, InternalErr)
	}

	tcNameGetter := &infraapi.ClusterNameGetter{
		InfraProvider:        cluster.InfraProviderType,
		InfraAPIObjNamespace: c.api.GetObjectNamespace(cluster.ProjectId),
		ClusterName:          infraapi.ClusterName(cluster.Id),
	}
	agentCluster := &agentpb.Cluster{
		ClusterId:    cluster.Id,
		Namespace:    tcNameGetter.ClusterNamespace(),
		User:         entity.CloudAdminUser,
		Password:     password,
		ProviderType: cluster.GetInfraProviderType(),
		// Timeout is 2 minutes, because the load data sql may take a long time
		// MaxUploadSize is 50 * units.MiB, it will take 40 seconds
		ReadTimeout:  120,
		WriteTimeout: 120,
	}

	switch imp.Status {
	case entity.ImportStatusPreparing:
		c.aliveLocalImports.Store(imp.ID, imp.Status)

		if err := c.transToStatus(ctx, imp, entity.ImportStatusImporting, ""); err != nil {
			return err
		}
		go func() {
			var errMessage string
			defer func(errMessage *string) {
				if errMessage != nil && len(*errMessage) > 0 {
					tranErr := c.transToStatus(ctx, imp, entity.ImportStatusFailed, *errMessage)
					if tranErr != nil {
						logger.Error("failed to transfer to status", zap.Error(tranErr))
					}
				}
			}(&errMessage)

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

			body, err := s3.GetFileBody(ctx, imp.SourceURL, "", "")
			if err != nil {
				logger.Error("get file body error", zap.Error(err))
				errMessage = InternalErr
				return
			}
			// nolint:staticcheck, errcheck
			defer body.Close()

			if csvCfg.Delimiter == "'" {
				csvCfg.Delimiter = "\\'"
			}

			terminator, err := getLineTerminator(body)
			if err != nil {
				logger.Error("get line terminator failed", zap.Error(err))
				errMessage = InternalErr
				return
			}
			req := &agentpb.LoadDataReq{
				Cluster:         agentCluster,
				TargetDb:        imp.LocalImportCfg.TargetDb,
				TargetTable:     imp.LocalImportCfg.TargetTable,
				Terminator:      terminator,
				SourceUri:       imp.SourceURL,
				Separator:       csvCfg.Separator,
				Delimiter:       csvCfg.Delimiter,
				BackslashEscape: csvCfg.BackslashEscape,
				IgnoreOneLine:   csvCfg.Header,
			}
			agentClient.SetHttpTimeout(2 * time.Minute)
			resp, err := agentClient.LoadDataToCluster(ctx, req)
			if err != nil {
				req.Cluster = nil
				logger.Error("load data failed", zap.Error(err), zap.Stringer("request", req))
				// `err` is the error except the error returned by running load data sql
				errMessage = InternalErr
				return
			}
			if resp != nil && len(resp.ErrMessage) != 0 {
				logger.Error("exec load data sql failed", zap.Stringer("error", resp), zap.Stringer("request", req))

				if strings.Contains(strings.ToLower(resp.ErrMessage), strings.ToLower(errno.MySQLErrName[errno.ErrResourceGroupThrottled].Raw)) {
					errMessage = ErrExceededResourceGroup
				} else {
					// `resp.ErrMessage` is the error of running load data sql
					errMessage = resp.ErrMessage
				}
				return
			}

			status, ok := c.aliveLocalImports.Load(imp.ID)
			if !ok {
				logger.Error("import status not found", zap.Uint64("import id", imp.ID.Uint64()))
				// if the worker is stopped, the status will be deleted
				// in this case, we should change the import status to failed
				errMessage = InternalErr
				return
			}
			var tranErr error
			switch status {
			case entity.ImportStatusImporting:
				// `resp.WarningMessage`` is the warning of running load data sql
				// now don't return warning message to user
				tranErr = c.transToStatus(ctx, imp, entity.ImportStatusCompleted, "")
			case entity.ImportStatusCanceling:
				// if the user cancels the import, the status will be changed to canceling
				tranErr = c.transToStatus(ctx, imp, entity.ImportStatusCanceled, "")
			}
			if tranErr != nil {
				logger.Error("failed to transfer to status", zap.Error(tranErr))
			}
		}()
	case entity.ImportStatusImporting:
		if _, ok := c.aliveLocalImports.Load(imp.ID); !ok {
			logger.Warn("local import not found", zap.Uint64("import id", imp.ID.Uint64()))
			return c.transToStatus(ctx, imp, entity.ImportStatusFailed, InternalErr)
		}

		newPercent := estimateProgress(imp.CreatedAt, imp.LocalImportCfg.TotalSize)
		// only update the progress when it's larger than the previous one.
		if newPercent >= imp.CompletedPercent {
			if err = db.Transaction(func(tx *gorm.DB) error {
				imp.CompletedPercent = newPercent
				return imp.Update(tx, entity.ImportDBSchema.CompletedPercent)
			}); err != nil {
				logger.Error("failed to update import progress", zap.Error(err))
				return c.transToStatus(ctx, imp, entity.ImportStatusFailed, InternalErr)
			}
		}
	case entity.ImportStatusCanceling:
		if _, ok := c.aliveLocalImports.Load(imp.ID); !ok {
			logger.Warn("local import not found", zap.Uint64("import id", imp.ID.Uint64()))
			return c.transToStatus(ctx, imp, entity.ImportStatusCanceled, "")
		}
		c.aliveLocalImports.Store(imp.ID, imp.Status)
	case entity.ImportStatusCanceled, entity.ImportStatusFailed, entity.ImportStatusCompleted:
		return c.cleanUp(ctx, imp)
	default:
		return errors.Errorf("unknown status: %v", imp.Status)
	}

	return nil
}

func (c BgContext) Sync(ctx context.Context, imp *ImportTask) (err error) {
	ctx, span := Tracer.Start(ctx, "BgContext.sync")
	defer func() { tracing.EndSpan(span, err) }()
	span.SetAttributes(attribute.String("cluster_id", fmt.Sprintf("%v", imp.ClusterID)),
		attribute.String("tenant_id", fmt.Sprintf("%v", imp.Cluster.GetTenantId())),
		attribute.String("import_name", infraapi.ImportName(imp.ID.Uint64())),
		attribute.String("status", imp.Status.String()),
	)

	logger := c.log
	logger.Info("begin to sync import task")
	defer func() {
		logger.Info("completed sync import task", zap.Any("status", imp.Status), zap.Error(err))
	}()

	// if the cluster is deleted, just failed the import and let it clean up resource(the machine for lighting).
	switch imp.Cluster.GetStatus() {
	case core.ClusterStatus_CLUSTER_STATUS_DELETED,
		core.ClusterStatus_CLUSTER_STATUS_DELETECOMPLETE,
		core.ClusterStatus_CLUSTER_STATUS_CLEARED:
		if imp.Status != entity.ImportStatusFailed {
			logger.Info("cluster is deleted, begion trans import status to failed")
			return c.transToStatus(ctx, imp, entity.ImportStatusFailed, "cluster is deleted")
		}
	}

	if imp.LocalImportCfg != nil {
		return c.syncLocalImport(ctx, imp)
	}

	switch imp.Status {
	case entity.ImportStatusPreparing:
		err = c.OnPreparing(ctx, imp)
	case entity.ImportStatusImporting:
		err = c.OnImporting(ctx, imp)
	case entity.ImportStatusCanceling:
		err = c.OnCanceling(ctx, imp)
	case entity.ImportStatusCanceled, entity.ImportStatusFailed, entity.ImportStatusCompleted:
		err = c.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 (m Model) Run(ctx context.Context) error {
	ctx = common.ContextWithImport(ctx)
	logger := m.log.With(log.FieldsFromContext(ctx)...).Sugar()
	imports, err := ImportsToSync(nil)
	if err != nil {
		logger.Errorf("failed to fetch imports to sync, err: %s", err.Error())
	}
	if len(imports) == 0 {
		return nil
	}

	startTime := time.Now()
	logger.Infof("[global]sync import with infra, start. cnt=%d", len(imports))
	var errs error
	for _, imp := range imports {
		c, err := m.BgContext(ctx, imp)
		if err != nil {
			errs = multierr.Append(errs, err)
			continue
		}

		err = c.Sync(ctx, imp)
		if err != nil {
			errs = multierr.Append(errs, err)
		}
	}
	logger.Infof("[global]sync import with infra, done. cost=%s,err=%s", time.Since(startTime), errs)
	return errs
}

// pass nil for all tenant or with a specified tenant.
func FailImportsInRecent(tenantID *entity.ID, recentDur time.Duration) ([]*ImportTask, error) {
	imports, err := dbutil.ListFailedImportsFinishedAfter(db.GetDB(), time.Now().Add(-recentDur))
	if err != nil {
		return nil, errors.Annotate(err, "failed to fetch imports")
	}
	results := make([]*ImportTask, len(imports))
	for i, impEntity := range imports {
		results[i] = &ImportTask{
			Import: impEntity,
		}
	}
	if results, err = BatchLoadImportTenantInfo(context.Background(), results); err != nil {
		plog.Error("batch load tenant info error", zap.Error(err))
	}
	if len(results) <= 0 {
		return nil, nil
	}
	if tenantID == nil {
		return results, nil
	}

	filteredResults := []*ImportTask{}
	for _, imp := range results {
		if imp.Cluster.GetTenantId() == uint64(*tenantID) {
			filteredResults = append(filteredResults, imp)
		}
	}
	return filteredResults, nil
}

// pass nil for all tenant or with a specified tenant.
func ImportsToSync(tenantID *entity.ID) ([]*ImportTask, error) {
	var imports []entity.Import
	imports, err := dbutil.ListActiveImports(db.GetDB())
	if err != nil {
		return nil, errors.Annotate(err, "failed to fetch imports")
	}
	results := make([]*ImportTask, len(imports))
	for i, imp := range imports {
		results[i] = &ImportTask{
			Import: imp,
		}
	}
	if results, err = BatchLoadImportTenantInfo(context.Background(), results); err != nil {
		plog.Error("batch load cluster info error", zap.Error(err))
	}
	if len(results) <= 0 {
		return nil, nil
	}

	if tenantID == nil {
		return results, nil
	}

	filteredResults := []*ImportTask{}
	for _, imp := range results {
		if imp.Cluster.GetTenantId() == uint64(*tenantID) {
			filteredResults = append(filteredResults, imp)
		}
	}
	return filteredResults, nil
}

func convertCsvStringToCr(ctx context.Context, csvString string) (*cloudv1.CsvFormat, error) {
	csvFormat := &cloudv1.CsvFormat{}
	if csvString == "" {
		return csvFormat, nil
	}
	csvConfig := dataflow.CustomCSVFormat{}

	if err := json.Unmarshal([]byte(csvString), &csvConfig); err != nil {
		log.Error(ctx, "failed to unmarshal csv config", zap.Error(err))
		return nil, err
	}

	csvFormat.Separator = &csvConfig.Separator
	csvFormat.Delimiter = &csvConfig.Delimiter
	csvFormat.Header = &csvConfig.Header
	csvFormat.NotNull = csvConfig.NotNull
	csvFormat.Null = csvConfig.Null
	csvFormat.BackslashEscape = &csvConfig.BackslashEscape
	csvFormat.TrimLastSeparator = csvConfig.TrimLastSeparator
	csvFormat.HeaderSchemaMatch = csvConfig.HeaderSchemaMatch

	return csvFormat, nil
}

func (m *Model) CleanBucket(ctx context.Context) error {
	// only support serverless tier now
	deployType := core.DeployType_DEPLOY_TYPE_SHARED
	regions, err := grpcli.GetRegionsByProvider(ctx, constants.CloudProviderAWS, &deployType)
	if err != nil {
		return errors.Annotate(err, "get cloud regions")
	}
	for _, region := range regions {
		ctx2, cancel := context.WithTimeout(ctx, 5*time.Minute)
		if err := s3.DeleteExpiredObject(
			ctx2,
			region.Name,
			NewLocalImportBucketNaming(region.Name, "", 0).GenBucketName(),
			10*time.Hour,
		); err != nil {
			// MalformedXML will not effect the delete operation
			m.log.Error("failed to delete expired object", zap.Error(err))
		}
		cancel()
	}
	return nil
}

// publishEvent is used to trigger eventbus to change the status of cluster(importing)
// maybe we can use event center to replace it in the future
func publishEvent(publisher eventpublisher.Publisher, newImport *ImportTask) error {
	// Note: default publish timeout for eventbus is 1s
	return publisher.Publish(context.Background(),
		event.ImportTopic,
		&event.Meta{EventID: uuid.New().String()},
		convertEntityToEvent(newImport))
}

func convertEntityToEvent(newImport *ImportTask) *event.ImportEvent {
	return &event.ImportEvent{
		EventType:  convertStatusToEventType(newImport.Status),
		ID:         uint64(newImport.ID),
		Name:       newImport.Name,
		ClusterID:  uint64(newImport.ClusterID),
		Status:     string(newImport.Status),
		Creator:    newImport.Creator,
		CreatedAt:  &newImport.CreatedAt,
		StartedAt:  newImport.StartedAt,
		FinishedAt: newImport.FinishedAt,
		ExtraInfo: &event.ImportEventExtraInfo{
			Message: newImport.Message,
		},
	}
}

func convertStatusToEventType(status entity.ImportStatus) event.ImportEventType {
	et := event.ImportEventTypeUnknown
	switch status {
	case entity.ImportStatusPreparing:
		et = event.ImportEventTypePreparing
	case entity.ImportStatusImporting:
		et = event.ImportEventTypeImporting
	case entity.ImportStatusCompleted:
		et = event.ImportEventTypeCompleted
	case entity.ImportStatusFailed:
		et = event.ImportEventTypeFailed
	case entity.ImportStatusCanceled:
		et = event.ImportEventTypeCanceled
	case entity.ImportStatusCanceling:
		et = event.ImportEventTypeCanceling
	default:
	}

	return et
}