package pimport

import (
	"context"
	"encoding/json"
	"fmt"
	"time"

	"github.com/pingcap/dbaas/central/config"
	"github.com/pingcap/dbaas/central/db"
	"github.com/pingcap/dbaas/central/db/dbutil"
	"github.com/pingcap/dbaas/central/db/entity"
	"github.com/pingcap/dbaas/central/db/tenant"
	infraapi "github.com/pingcap/dbaas/central/infra-api"
	"github.com/pingcap/dbaas/central/metrics"
	cdcModel "github.com/pingcap/dbaas/central/model/cdc"
	modelutil "github.com/pingcap/dbaas/central/model/util"
	"github.com/pingcap/errcode"
	"github.com/pingcap/errors"
	"github.com/tidbcloud/idl/pbgen/proto/central"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/pkgv2/log"
	"github.com/tidbcloud/pkgv2/secret/encrypt"
	"github.com/tidbcloud/pkgv2/secret/vault"
	"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"
	"gorm.io/gorm"
)

const (
	DatasourceS3  = 1
	DataSourceGCS = 2

	// 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."
)

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

// Model is created with NewModel
type Model struct {
	log       *zap.Logger
	encryptor encrypt.Encryptor
	api       *infraapi.API
}

// NewModel returns a Model
func NewModel(log *zap.Logger, infraAPI *infraapi.API, encryptor encrypt.Encryptor) Model {
	return Model{
		log:       log,
		encryptor: encryptor,
		api:       infraAPI,
	}
}

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

// User returns a UserContext
func (m Model) User(ctx context.Context, user *tenant.User) UserContext {
	return UserContext{m, user, m.log.With(
		zap.Uint64("tenantID", uint64(user.TenantID)),
		zap.Uint64("userID", uint64(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) CheckOrgIDAndProjectID(orgID, projectID entity.ID) error {
	user := c.tenantAndUser
	if user.IsImpersonated {
		return nil
	}

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

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

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

func GetImportStorageRequestConfig(isDevTier bool) string {
	if isDevTier {
		return config.DevTierImportStorageRequest
	}
	return config.Global().ImportConfig.StorageRequest
}

func (c UserContext) ExitImport(req *central.ClusterImportId) (err error) {
	return db.Transaction(func(tx *gorm.DB) error {
		importID := entity.ID(req.Id)
		orgID := entity.ID(req.OrgId)
		projectID := entity.ID(req.ProjectId)
		clusterID := entity.ID(req.ClusterId)

		imp, err := dbutil.GetImportByID(tx, orgID, projectID, clusterID, importID)
		if err != nil {
			return err
		}

		if !isEndStatus(imp.Status) {
			return errors.Errorf("can't exit at status: %s", imp.Status)
		}
		imp.Cluster.ImportID = nil
		err = imp.Cluster.Update(tx, entity.ClusterDBSchema.ImportID)
		_ = dbutil.AddAccessLogNow(tx, c.tenantAndUser.User.TenantID, c.tenantAndUser.Email, fmt.Sprintf("exit import for cluster:  %d, import id: %d", clusterID, imp.ID))
		return errors.AddStack(err)
	})
}

// 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, req *central.ClusterImportId, waitCanceled bool) (err error) {
	importID := entity.ID(req.Id)
	orgID := entity.ID(req.OrgId)
	projectID := entity.ID(req.ProjectId)
	clusterID := entity.ID(req.ClusterId)

	var imp = &entity.Import{}
	err = db.Transaction(func(tx *gorm.DB) error {
		imp, err = dbutil.GetImportByID(tx, orgID, projectID, clusterID, importID)
		if err != nil {
			return err
		}
		oldStatus := imp.Status
		switch imp.Status {
		case entity.ImportStatusPreparing:
		case entity.ImportStatusImporting:
		case entity.ImportStatusCanceling:
		case entity.ImportStatusFailed:
			return errors.New("import already failed")
		case entity.ImportStatusCompleted:
			return errors.New("import already completed")
		case entity.ImportStatusCanceled:
			return nil
		default:
			return errors.Errorf("unknown status: %s", imp.Status)
		}

		// Handle cancel here.
		imp.Status = entity.ImportStatusCanceling
		err = imp.Update(tx.Where("status = ?", oldStatus), entity.ImportDBSchema.Status)
		if err != nil {
			return errors.AddStack(err)
		}

		return dbutil.AddAccessLogNow(tx, c.tenantAndUser.User.TenantID, c.tenantAndUser.Email, fmt.Sprintf("cancel import for cluster:  %d, import id: %d", clusterID, imp.ID))
	})

	if err != nil {
		return errors.AddStack(err)
	}

	if !waitCanceled {
		return nil
	}

	for {
		err := db.Transaction(func(tx *gorm.DB) error {
			return entity.NewImportQuerySet(tx).IDEq(importID).One(imp)
		})
		if err != nil {
			return errors.AddStack(err)
		}

		if isEndStatus(imp.Status) {
			return nil
		}

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

func (c UserContext) GetImport(req *central.ClusterImportId) (*entity.Import, error) {
	var imp = &entity.Import{}
	var err error
	orgID := entity.ID(req.OrgId)
	projectID := entity.ID(req.ProjectId)
	clusterID := entity.ID(req.ClusterId)
	importID := entity.ID(req.Id)
	err = db.Transaction(func(tx *gorm.DB) error {
		imp, err = dbutil.GetImportByID(tx, orgID, projectID, clusterID, importID)
		if err != nil {
			return err
		}
		return nil
	})
	return imp, err
}

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

	var cluster entity.Cluster
	var tasks []entity.CDCTask
	var connectTypes string
	nodeRole := &entity.IamRole{}
	err = db.Transaction(func(tx *gorm.DB) error {
		cluster, err = dbutil.GetClusterByID(tx, orgID, projectID, imp.ClusterID)
		if err != nil {
			return err
		}
		if cluster.CrossAccountDeploymentID != 0 {
			if err := tx.Model(&entity.IamRole{}).Where("id=?", cluster.CrossAccountDeployment.NodeRoleID).Find(nodeRole).Error; err != nil {
				return err
			}
		}
		return nil
	})
	if err != nil {
		return 0, errors.AddStack(err)
	}
	if cluster.ReadOnly&entity.ReadOnlyStatus != 0 {
		return 0, errcode.NewInvalidInputErr(errors.New(ReadonlyForbidenErr))
	}
	if cluster.EncryptedCloudAdminPassword != "" {
		cloudCluster, err := c.api.GetCluster(ctx, uint64(projectID), uint64(imp.ClusterID))
		if err != nil {
			return 0, errors.AddStack(err)
		}
		useSem, err := modelutil.UseSemForCluster(cloudCluster.Status)
		if err != nil {
			return 0, errors.AddStack(err)
		}
		if useSem {
			imp.TargetUsername = entity.CloudAdminUser
			imp.TargetPassword = cluster.EncryptedCloudAdminPassword
		}
	}

	if imp.TargetPassword != "" && imp.TargetUsername != entity.CloudAdminUser {
		encryptedPassword, err := encrypt.EncryptBase64(c.encryptor, vault.Secret(imp.TargetPassword))
		if err != nil {
			return 0, err
		}

		imp.TargetPassword = encryptedPassword
		span.AddEvent("Encrypt password")
	}

	if cluster.ReplicationID != nil {
		return 0, errors.Errorf("cluster %d is in replication", cluster.ID.Uint64())
	}

	// use nodeRole as import role if the cluster is cross-account deployment
	if cluster.CrossAccountDeploymentID != 0 {
		imp.AwsRoleArn = nodeRole.Arn
	}

	if !config.Global().ImportConfig.SkipBanMyBuckets && cluster.Provider.Name == entity.Aws {
		bucket := parseBucketName(imp.SourceURL)
		isMy, err := isMyBuckets(imp.BucketRegion, bucket)
		if err != nil {
			return 0, err
		}
		if isMy {
			return 0, errcode.NewInvalidInputErr(fmt.Errorf("bucket %s belong to forbidden account", bucket))
		}
	}

	err = db.Transaction(func(tx *gorm.DB) error {
		cluster, err := dbutil.GetClusterByID(tx, orgID, projectID, imp.ClusterID)
		if err != nil {
			return errors.AddStack(err)
		}

		err = entity.NewCDCTaskQuerySet(tx).
			ClusterIDEq(cluster.ID).
			StatusNotIn(entity.CDCStatusDeleting, entity.CDCStatusDeleted).
			All(&tasks)
		if err != nil {
			return errors.AddStack(err)
		}
		for _, t := range tasks {
			connectTypes = connectTypes + cdcModel.DisplayNameTransfer[t.DownStreamType] + " "
		}
		if len(tasks) != 0 {
			return errors.Errorf(ConnectorErr, connectTypes)
		}

		if cluster.ImportID != nil {
			return errcode.NewInvalidInputErr(errors.New("already importing"))
		}

		if cluster.Status != entity.ClusterStatusNormal {
			return errors.NotValidf("cannot import data to cluster for now")
		}

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

		cluster.ImportID = new(uint64)
		*cluster.ImportID = uint64(imp.ID)
		if err := cluster.Update(tx, entity.ClusterDBSchema.ImportID); err != nil {
			return errors.AddStack(err)
		}

		return dbutil.AddAccessLogNow(tx, c.tenantAndUser.User.TenantID, c.tenantAndUser.Email, fmt.Sprintf("create import for cluster:  %d, import id: %d", cluster.ID, imp.ID))
	})

	if err != nil {
		return 0, err
	}

	return uint64(imp.ID), nil
}

// BgContext is meant for system worker/controller
type BgContext UserContext

// BgContext returns a BgContext
func (m Model) BgContext(tenantID entity.ID) (BgContext, error) {
	var systemUser entity.User
	err := db.Transaction(func(tx *gorm.DB) error {
		var err error
		systemUser, err = dbutil.SystemUser(tx, tenantID)
		return err
	})
	if err != nil {
		return BgContext{}, err
	}
	return BgContext{m, &tenant.User{TenantID: tenantID}, m.log.With(zap.Uint64("tenantID", tenantID.Uint64()), zap.Uint64("userID", uint64(systemUser.ID)), zap.String("module", "import"))}, nil
}

func (c BgContext) transToStatus(ctx context.Context, imp *entity.Import, 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()))
	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 {
		return errors.AddStack(err)
	}

	if isEndStatus(status) {
		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 *entity.Import) (err error) {
	ctx, span := Tracer.Start(ctx, "BgContext.cleanUp")
	defer func() { tracing.EndSpan(span, err) }()
	err = c.api.DeleteImport(ctx, uint64(imp.Cluster.ProjectID), uint64(imp.ID))
	if err != nil {
		return err
	}

	imp.CleanUp = true
	err = db.Transaction(func(tx *gorm.DB) error {
		return imp.Update(tx, entity.ImportDBSchema.CleanUp)
	})
	if err != nil {
		return errors.AddStack(err)
	}

	// currently we only fetch the imports not clean up to sync
	// but we cleanup also when in canceling status so make sure
	// update db status only when it's at the end status that we no longer need to sync the import.
	if isEndStatus(imp.Status) {
		imp.CleanUp = true
		err := db.Transaction(func(tx *gorm.DB) error {
			return imp.Update(tx, entity.ImportDBSchema.CleanUp)
		})
		if err != nil {
			return errors.AddStack(err)
		}
	}
	return nil
}

func (c BgContext) OnPreparing(ctx context.Context, imp *entity.Import) (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())))
	var password string
	storageRequest := GetImportStorageRequestConfig(imp.Cluster.Free)
	if imp.TargetPassword != "" {
		secret, err := encrypt.DecryptBase64(c.encryptor, imp.TargetPassword)
		if err != nil {
			return errors.Annotatef(err, "failed to decrypt, import id: %d", imp.ID)
		}
		password = string(secret)
	}
	csvConfig := central.CustomCSVFormat{}
	var csvFormat *cloudv1.CsvFormat
	if imp.DataFormat == entity.DataFormatCustomCSV {
		if err := json.Unmarshal([]byte(imp.CustomCSVFormat), &csvConfig); err != nil {
			log.Error(ctx, "failed to unmarshal csv config", zap.Error(err))
			return err
		}

		csvFormat = &cloudv1.CsvFormat{}
		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
	}

	if imp.DataSourceType == DatasourceS3 {
		err := c.api.CreateAwsImportIgnoreExists(
			context.TODO(),
			uint64(imp.Cluster.TenantID),
			uint64(imp.Cluster.ProjectID),
			uint64(imp.ClusterID),
			uint64(imp.ID),
			imp.TargetUsername,
			password,
			imp.SourceURL,
			imp.BucketRegion,
			imp.Filter,
			storageRequest,
			cloudv1.DataFormat(imp.DataFormat),
			csvFormat,
			imp.AwsRoleArn,
			imp.OptionalArgs.ObjectNamePattern,
			imp.OptionalArgs.TargetTablePattern,
		)
		if err != nil {
			return err
		}
	}
	if imp.DataSourceType == DataSourceGCS {
		err := c.api.CreateGcpImportIgnoreExists(
			context.TODO(),
			uint64(imp.Cluster.TenantID),
			uint64(imp.Cluster.ProjectID),
			uint64(imp.ClusterID),
			uint64(imp.ID),
			imp.TargetUsername,
			password,
			imp.SourceURL,
			imp.BucketRegion,
			imp.Filter,
			storageRequest,
			cloudv1.DataFormat(imp.DataFormat),
			csvFormat,
			imp.OptionalArgs.ObjectNamePattern,
			imp.OptionalArgs.TargetTablePattern,
		)
		if err != nil {
			return err
		}
	}

	return c.updateInfo(ctx, imp)
}

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

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

func (c BgContext) updateFromLightningV1(imp *entity.Import) error {
	oldStatus := imp.Status
	impcr, err := c.api.GetImport(context.TODO(), uint64(imp.Cluster.ProjectID), uint64(imp.ID))
	if err != nil {
		return err
	}
	imp.TotalSize = &impcr.Status.TotalSize
	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,
			CompletedPercent: v.CompletedPercent,
		})
	}
	imp.CompletedTables = impcr.Status.CompletedTables
	imp.PendingTables = impcr.Status.PendingTables
	imp.CompletedPercent = impcr.Status.CompletedPercent
	if !(imp.Status == entity.ImportStatusCanceled || imp.Status == entity.ImportStatusCanceling) {
		imp.Status = entity.ImportStatus(convertStatus(string(impcr.Status.Phase)))
	}

	// TODO: add metric
	if imp.Status == entity.ImportStatusCompleted {
		usedTime := time.Since(imp.CreatedAt).Seconds()
		//_, _, tikvList, _ := dbutil.AssembleComponentList(*imp.Cluster)
		totalSize := float64(*imp.TotalSize)
		metrics.ImportThroughputBytes.WithLabelValues(metrics.GetProviderType(imp.Cluster.Free)).Observe(totalSize / float64(dbutil.GetTikvReplicas(imp.Cluster)) / usedTime)
	}
	imp.Message = impcr.Status.Message
	if len(imp.Message) > 255 {
		imp.Message = imp.Message[:255]
	}
	fields := []entity.ImportDBSchemaField{
		entity.ImportDBSchema.TotalSize,
		entity.ImportDBSchema.TotalFiles,
		entity.ImportDBSchema.CurrentTables,
		entity.ImportDBSchema.CompletedTables,
		entity.ImportDBSchema.PendingTables,
		entity.ImportDBSchema.CompletedPercent,
		entity.ImportDBSchema.Status,
		entity.ImportDBSchema.Message,
	}
	if imp.Status == entity.ImportStatusImporting && imp.StartedAt == nil {
		t := time.Now()
		imp.StartedAt = &t
		fields = append(fields, entity.ImportDBSchema.StartedAt)
	}

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

	c.log.Debug("updating info from import object succeeded")

	err = db.Transaction(func(tx *gorm.DB) error {
		return imp.Update(tx.Where("status = ?", oldStatus),
			fields...)
	})
	if err != nil {
		return errors.Annotate(err, "failed to update info to db")
	}
	return nil
}

func (c BgContext) OnImporting(ctx context.Context, imp *entity.Import) error {
	return c.updateInfo(ctx, imp)
}

func (c BgContext) OnCanceling(ctx context.Context, imp *entity.Import) (err error) {
	ctx, span := Tracer.Start(ctx, "BgContext.OnCanceling")
	defer func() { tracing.EndSpan(span, err) }()
	err = c.api.DeleteImport(ctx, uint64(imp.Cluster.ProjectID), uint64(imp.ID))
	if err != nil {
		return err
	}
	return c.transToStatus(ctx, imp, entity.ImportStatusCanceled, "")
}

func (c BgContext) sync(ctx context.Context, imp *entity.Import) (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.TenantID.Uint64())),
		attribute.String("import_name", infraapi.ImportName(imp.ID.Uint64())),
		attribute.String("status", imp.Status.String()),
	)
	// if the cluster is deleted, just failed the import and let it clean up resource(the machine for lighting).
	if imp.Cluster.Status.DeletedOrCleared() && imp.Status != entity.ImportStatusFailed {
		return c.transToStatus(ctx, imp, entity.ImportStatusFailed, "cluster is deleted")
	}

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

func (c BgContext) Run(ctx context.Context) error {
	imports, err := ImportsToSync(&c.tenantAndUser.TenantID)
	if err != nil {
		return err
	}
	if len(imports) == 0 {
		return nil
	}

	c.log.Debug("imports to sync", zap.Int("number", len(imports)))

	var errs error
	for _, imp := range imports {
		imp := imp
		err := c.sync(ctx, &imp)
		if err != nil {
			errs = multierr.Append(errs, err)
		}
	}

	return errs
}

// pass nil for all tenant or with a specified tenant.
func FailImportsInRecent(tenantID *entity.ID, recentDur time.Duration) ([]entity.Import, error) {
	var imports []entity.Import
	err := db.Transaction(func(tx *gorm.DB) error {
		return entity.NewImportQuerySet(tx).PreloadCluster().
			StatusEq(entity.ImportStatusFailed).
			FinishedAtGt(time.Now().Add(-recentDur)).
			All(&imports)
	})
	if err != nil {
		return nil, errors.Annotate(err, "failed to fetch imports")
	}

	if tenantID != nil {
		var n int
		for _, imp := range imports {
			if imp.Cluster.TenantID == *tenantID {
				imports[n] = imp
				n++
			}
		}
		imports = imports[:n]
	}

	return imports, nil
}

// pass nil for all tenant or with a specified tenant.
func ImportsToSync(tenantID *entity.ID) ([]entity.Import, error) {
	var imports []entity.Import
	err := db.Transaction(func(tx *gorm.DB) error {
		return entity.NewImportQuerySet(tx.Preload("Cluster.Provider")).CleanUpEq(false).PreloadCluster().All(&imports)
	})
	if err != nil {
		return nil, errors.Annotate(err, "failed to fetch imports")
	}

	if tenantID != nil {
		var n int
		for _, imp := range imports {
			if imp.Cluster.TenantID == *tenantID {
				imports[n] = imp
				n++
			}
		}
		imports = imports[:n]
	}

	return imports, nil
}